""" 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: 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 # --------------------------------------------------------------------------- # # Design im Stil von software.obraasch.de: dunkler Header, orange Akzente, # helle Cards. # --------------------------------------------------------------------------- # PAGE = """ SkillSpector

Sicherheits-Scanner für AI-Agent-Skills – vor dem Installieren prüfen

{% if allow_llm %}
{% endif %}
{% if result %}
{{ result.risk_assessment.score }}/100 {{ result.risk_assessment.recommendation }} 
{{ result_json }}
{% elif error %}

{{ error }}

{% endif %}
""" @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"}