291 lines
10 KiB
Python
291 lines
10 KiB
Python
"""
|
||
SkillSpector Web – Wrapper um die skillspector-CLI, geeignet für öffentlichen Betrieb.
|
||
|
||
Sicherheitsprinzipien:
|
||
- Führt NIEMALS den gescannten Skill aus (das macht skillspector selbst auch nicht).
|
||
- Nur http/https-URLs werden akzeptiert; Ziele, die auf private/interne Adressen
|
||
auflösen (SSRF-Schutz: localhost, 127.x, 10.x, 192.168.x, 172.16-31.x, link-local,
|
||
IPv6-Loopback/ULA), werden abgelehnt.
|
||
- Läuft standardmäßig mit --no-llm; LLM-Analyse ist nur nutzbar, wenn der Admin sie
|
||
explizit über ENABLE_LLM_OPTION freischaltet UND ein API-Key konfiguriert ist.
|
||
- Zugriff optional per Basic-Auth (REQUIRE_AUTH); bei öffentlichem Betrieb ohne Auth
|
||
greifen automatisch Rate-Limit und Concurrency-Limit.
|
||
- Jeder Scan läuft in einem eigenen, danach gelöschten Temp-Verzeichnis.
|
||
- Harte Zeit- und Upload-Größenbegrenzung.
|
||
"""
|
||
|
||
import ipaddress
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import socket
|
||
import subprocess
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from functools import wraps
|
||
from pathlib import Path
|
||
from urllib.parse import urlparse
|
||
|
||
from flask import Flask, Response, render_template_string, request
|
||
|
||
app = Flask(__name__)
|
||
app.config["MAX_CONTENT_LENGTH"] = 50 * 1024 * 1024 # 50 MB Upload-Limit
|
||
|
||
SCAN_TMP_DIR = os.environ.get("SCAN_TMP_DIR", "/tmp/skillspector")
|
||
Path(SCAN_TMP_DIR).mkdir(parents=True, exist_ok=True)
|
||
|
||
REQUIRE_AUTH = os.environ.get("REQUIRE_AUTH", "true").lower() == "true"
|
||
WEBUI_USER = os.environ.get("WEBUI_USER", "admin")
|
||
WEBUI_PASSWORD = os.environ.get("WEBUI_PASSWORD")
|
||
|
||
SCAN_TIMEOUT = int(os.environ.get("SCAN_TIMEOUT_SECONDS", "300"))
|
||
MAX_CONCURRENT_SCANS = int(os.environ.get("MAX_CONCURRENT_SCANS", "2"))
|
||
MAX_SCANS_PER_HOUR_PER_IP = int(os.environ.get("MAX_SCANS_PER_HOUR_PER_IP", "10"))
|
||
|
||
LLM_KEY_PRESENT = bool(os.environ.get("OPENAI_API_KEY") or os.environ.get("ANTHROPIC_API_KEY"))
|
||
ENABLE_LLM_OPTION = os.environ.get("ENABLE_LLM_OPTION", "false").lower() == "true"
|
||
ALLOW_LLM = LLM_KEY_PRESENT and ENABLE_LLM_OPTION
|
||
|
||
GIT_URL_RE = re.compile(r"^https?://[^\s]+$")
|
||
|
||
_scan_semaphore = threading.Semaphore(MAX_CONCURRENT_SCANS)
|
||
_rate_lock = threading.Lock()
|
||
_rate_log: dict[str, list[float]] = {}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Auth
|
||
# --------------------------------------------------------------------------- #
|
||
def check_auth(username: str, password: str) -> bool:
|
||
return bool(WEBUI_PASSWORD) and username == WEBUI_USER and password == WEBUI_PASSWORD
|
||
|
||
|
||
def authenticate():
|
||
return Response(
|
||
"Zugriff verweigert.",
|
||
401,
|
||
{"WWW-Authenticate": 'Basic realm="SkillSpector"'},
|
||
)
|
||
|
||
|
||
def requires_auth(f):
|
||
@wraps(f)
|
||
def decorated(*args, **kwargs):
|
||
if not REQUIRE_AUTH:
|
||
return f(*args, **kwargs)
|
||
if not WEBUI_PASSWORD:
|
||
return Response(
|
||
"REQUIRE_AUTH=true, aber WEBUI_PASSWORD ist nicht gesetzt – "
|
||
"Zugriff aus Sicherheitsgründen gesperrt.",
|
||
500,
|
||
)
|
||
auth = request.authorization
|
||
if not auth or not check_auth(auth.username, auth.password):
|
||
return authenticate()
|
||
return f(*args, **kwargs)
|
||
|
||
return decorated
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Rate-Limiting (pro Client-IP, in-memory – reicht für einen einzelnen Prozess)
|
||
# --------------------------------------------------------------------------- #
|
||
def client_ip() -> str:
|
||
# Wenn hinter einem Reverse Proxy: X-Forwarded-For des Proxys vertrauen.
|
||
fwd = request.headers.get("X-Forwarded-For")
|
||
return fwd.split(",")[0].strip() if fwd else request.remote_addr or "unknown"
|
||
|
||
|
||
def rate_limited(ip: str) -> bool:
|
||
now = time.time()
|
||
with _rate_lock:
|
||
entries = [t for t in _rate_log.get(ip, []) if now - t < 3600]
|
||
if len(entries) >= MAX_SCANS_PER_HOUR_PER_IP:
|
||
_rate_log[ip] = entries
|
||
return True
|
||
entries.append(now)
|
||
_rate_log[ip] = entries
|
||
return False
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# SSRF-Schutz: nur öffentliche http(s)-Ziele erlauben
|
||
# --------------------------------------------------------------------------- #
|
||
def is_public_target(url: str) -> bool:
|
||
if not GIT_URL_RE.match(url):
|
||
return False
|
||
host = urlparse(url).hostname
|
||
if not host:
|
||
return False
|
||
if host.lower() == "localhost":
|
||
return False
|
||
try:
|
||
infos = socket.getaddrinfo(host, None)
|
||
except socket.gaierror:
|
||
return False
|
||
for info in infos:
|
||
ip = ipaddress.ip_address(info[4][0])
|
||
if (
|
||
ip.is_private
|
||
or ip.is_loopback
|
||
or ip.is_link_local
|
||
or ip.is_reserved
|
||
or ip.is_multicast
|
||
or ip.is_unspecified
|
||
):
|
||
return False
|
||
return True
|
||
|
||
|
||
PAGE = """
|
||
<!doctype html>
|
||
<html lang="de">
|
||
<head>
|
||
<meta charset="utf-8">
|
||
<title>SkillSpector</title>
|
||
<style>
|
||
body{font-family:sans-serif;max-width:820px;margin:40px auto;padding:0 16px;background:#111;color:#eee}
|
||
input,button{width:100%;padding:9px;margin:6px 0;box-sizing:border-box;font-size:1em}
|
||
input[type=checkbox]{width:auto;display:inline}
|
||
button{background:#2ecc71;color:#000;border:none;cursor:pointer;font-weight:bold;padding:10px}
|
||
pre{background:#000;padding:12px;overflow-x:auto;white-space:pre-wrap;border-radius:6px}
|
||
.score{font-size:1.8em;font-weight:bold;margin:12px 0}
|
||
.SAFE{color:#4caf50}.CAUTION{color:#ffb300}.DO_NOT_INSTALL{color:#f44336}
|
||
label{display:block;margin-top:14px;font-size:.9em;color:#bbb}
|
||
.err{color:#f44336;font-weight:bold}
|
||
.notice{font-size:.85em;color:#999;margin-top:24px;border-top:1px solid #333;padding-top:12px}
|
||
hr{border-color:#333}
|
||
a{color:#6cf}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<h1>🔍 SkillSpector</h1>
|
||
<p>Scannt einen Skill (Git-Repo oder ZIP) auf Sicherheitsrisiken, <strong>bevor</strong> er installiert wird.
|
||
Basiert auf <a href="https://github.com/NVIDIA/SkillSpector" target="_blank">NVIDIA SkillSpector</a> (Apache-2.0).</p>
|
||
<form method="post" enctype="multipart/form-data">
|
||
<label>Git-Repository-URL (nur öffentlich erreichbares http/https)</label>
|
||
<input type="text" name="git_url" placeholder="https://git.obraasch.de/olli1986/mein-skill">
|
||
<label>...oder ZIP-Datei hochladen (max. 50 MB)</label>
|
||
<input type="file" name="zip_file" accept=".zip">
|
||
{% if allow_llm %}
|
||
<label><input type="checkbox" name="use_llm"> LLM-Analyse aktivieren (sendet Dateiinhalte des Skills an den konfigurierten KI-Anbieter – genauer, aber nicht mehr rein lokal)</label>
|
||
{% endif %}
|
||
<button type="submit">Scan starten</button>
|
||
</form>
|
||
<hr>
|
||
{% if result %}
|
||
<h2>Ergebnis</h2>
|
||
<p class="score {{ result.risk_assessment.recommendation }}">
|
||
{{ result.risk_assessment.score }}/100 — {{ result.risk_assessment.severity }} — {{ result.risk_assessment.recommendation }}
|
||
</p>
|
||
<pre>{{ result_json }}</pre>
|
||
{% elif error %}
|
||
<p class="err">{{ error }}</p>
|
||
{% endif %}
|
||
<p class="notice">
|
||
Statische Analyse läuft immer lokal. {% if allow_llm %}Bei aktivierter LLM-Analyse werden
|
||
Dateiinhalte an einen externen KI-Anbieter übertragen.{% endif %}
|
||
Der gescannte Skill wird zu keinem Zeitpunkt ausgeführt. Kein Ergebnis wird dauerhaft
|
||
gespeichert – es erscheint nur auf dieser Seite.
|
||
</p>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
|
||
@app.route("/", methods=["GET", "POST"])
|
||
@requires_auth
|
||
def index():
|
||
if request.method == "GET":
|
||
return render_template_string(PAGE, allow_llm=ALLOW_LLM, result=None, error=None)
|
||
|
||
ip = client_ip()
|
||
if rate_limited(ip):
|
||
return render_template_string(
|
||
PAGE,
|
||
allow_llm=ALLOW_LLM,
|
||
result=None,
|
||
error=f"Limit erreicht: maximal {MAX_SCANS_PER_HOUR_PER_IP} Scans pro Stunde und IP. Bitte später erneut versuchen.",
|
||
)
|
||
|
||
use_llm = ALLOW_LLM and request.form.get("use_llm") == "on"
|
||
git_url = (request.form.get("git_url") or "").strip()
|
||
zip_file = request.files.get("zip_file")
|
||
|
||
if not _scan_semaphore.acquire(blocking=False):
|
||
return render_template_string(
|
||
PAGE,
|
||
allow_llm=ALLOW_LLM,
|
||
result=None,
|
||
error="Gerade laufen bereits zu viele Scans gleichzeitig. Bitte kurz warten und erneut versuchen.",
|
||
)
|
||
|
||
work_dir = tempfile.mkdtemp(dir=SCAN_TMP_DIR)
|
||
try:
|
||
if zip_file and zip_file.filename:
|
||
zip_path = Path(work_dir) / "upload.zip"
|
||
zip_file.save(zip_path)
|
||
target = str(zip_path)
|
||
elif git_url:
|
||
if not is_public_target(git_url):
|
||
return render_template_string(
|
||
PAGE,
|
||
allow_llm=ALLOW_LLM,
|
||
result=None,
|
||
error="Diese URL ist nicht erlaubt (nur öffentlich erreichbare http/https-Ziele).",
|
||
)
|
||
target = git_url
|
||
else:
|
||
return render_template_string(
|
||
PAGE,
|
||
allow_llm=ALLOW_LLM,
|
||
result=None,
|
||
error="Bitte eine Git-URL angeben oder eine ZIP-Datei hochladen.",
|
||
)
|
||
|
||
report_path = Path(work_dir) / "report.json"
|
||
cmd = ["skillspector", "scan", target, "--format", "json", "-o", str(report_path)]
|
||
if not use_llm:
|
||
cmd.append("--no-llm")
|
||
|
||
proc = subprocess.run(
|
||
cmd,
|
||
cwd=work_dir,
|
||
timeout=SCAN_TIMEOUT,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
|
||
if report_path.exists():
|
||
result = json.loads(report_path.read_text())
|
||
return render_template_string(
|
||
PAGE,
|
||
allow_llm=ALLOW_LLM,
|
||
result=result,
|
||
result_json=json.dumps(result, indent=2, ensure_ascii=False),
|
||
error=None,
|
||
)
|
||
return render_template_string(
|
||
PAGE,
|
||
allow_llm=ALLOW_LLM,
|
||
result=None,
|
||
error=f"Scan fehlgeschlagen (Exit {proc.returncode}):\n{proc.stderr[-2000:]}",
|
||
)
|
||
except subprocess.TimeoutExpired:
|
||
return render_template_string(
|
||
PAGE,
|
||
allow_llm=ALLOW_LLM,
|
||
result=None,
|
||
error=f"Scan abgebrochen: Zeitlimit von {SCAN_TIMEOUT}s überschritten.",
|
||
)
|
||
finally:
|
||
shutil.rmtree(work_dir, ignore_errors=True)
|
||
_scan_semaphore.release()
|
||
|
||
|
||
@app.route("/healthz")
|
||
def healthz():
|
||
return {"status": "ok"}
|