""" 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 import urllib.request import zipfile from datetime import datetime 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 # Für die deutsche Klartext-Zusammenfassung wird direkt die OpenRouter-kompatible # Chat-API angesprochen (unabhängig von SkillSpectors eigener --llm-Option). OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY") OPENAI_BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://openrouter.ai/api/v1") SUMMARY_MODEL = os.environ.get("SKILLSPECTOR_MODEL") or "openai/gpt-4o-mini" 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 # --------------------------------------------------------------------------- # # Rohreport lesbar aufbereiten: nach Severity gruppieren, Feldnamen # vereinheitlichen (SkillSpector nennt Detail-Felder je Version leicht anders). # --------------------------------------------------------------------------- # SEVERITY_ORDER = ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"] SEVERITY_LABELS_DE = { "CRITICAL": "Kritisch", "HIGH": "Hoch", "MEDIUM": "Mittel", "LOW": "Niedrig", "INFO": "Info", } def normalize_issues(raw_issues: list) -> dict: grouped: dict[str, list] = {} for issue in raw_issues or []: severity = str(issue.get("severity", "INFO")).upper() location = issue.get("location") or {} confidence = issue.get("confidence") if isinstance(confidence, (int, float)) and confidence <= 1: confidence_display = f"{round(confidence * 100)}%" else: confidence_display = confidence grouped.setdefault(severity, []).append( { "id": issue.get("id") or issue.get("rule_id") or "?", "category": issue.get("category") or issue.get("type") or "", "message": ( issue.get("finding") or issue.get("message") or issue.get("description") or issue.get("title") or "" ), "explanation": issue.get("explanation") or issue.get("rationale") or "", "file": location.get("file") or location.get("path") or "", "line": location.get("start_line") or location.get("line") or "", "confidence": confidence_display, } ) ordered = {lvl: grouped[lvl] for lvl in SEVERITY_ORDER if lvl in grouped} for lvl, items in grouped.items(): if lvl not in ordered: ordered[lvl] = items return ordered def normalize_components(raw_components: list) -> list: normalized = [] for c in raw_components or []: normalized.append( { "file": c.get("path") or c.get("file") or c.get("name") or "-", "type": c.get("type") or c.get("kind") or "-", "lines": c.get("lines") or c.get("line_count") or "-", "executable": bool(c.get("executable")), } ) return normalized # --------------------------------------------------------------------------- # # Anfang der eigentlichen Skill-Datei (Name/Beschreibung/Einleitung) holen – # getrennt von SkillSpectors eigenem Scan, nur um der KI echten Kontext für # den "Wofür ist das?"-Absatz zu geben statt nur der Sicherheitsfunde. # --------------------------------------------------------------------------- # def extract_skill_context(target: str, is_zip: bool, work_dir: str) -> str: context_dir = Path(work_dir) / "context_extract" try: context_dir.mkdir(exist_ok=True) if is_zip: with zipfile.ZipFile(target) as zf: zf.extractall(context_dir) else: subprocess.run( ["git", "clone", "--depth", "1", target, str(context_dir / "repo")], timeout=60, capture_output=True, text=True, ) candidates = ( list(context_dir.rglob("SKILL.md")) or list(context_dir.rglob("skill.md")) or list(context_dir.rglob("README.md")) ) if not candidates: return "" return candidates[0].read_text(errors="ignore")[:3000] except Exception: return "" finally: shutil.rmtree(context_dir, ignore_errors=True) # --------------------------------------------------------------------------- # # Deutsche Klartext-Zusammenfassung des Reports per LLM (OpenRouter-kompatibel) # --------------------------------------------------------------------------- # def generate_report_summary(result: dict, skill_context: str = "") -> str | None: if not OPENAI_API_KEY: return None risk = result.get("risk_assessment") or {} issues = result.get("issues") or result.get("findings") or [] components = result.get("components") or [] other_fields = { k: v for k, v in result.items() if k not in ("issues", "findings", "components", "risk_assessment") } compact = { "score": risk.get("score"), "severity": risk.get("severity"), "recommendation": risk.get("recommendation"), "total_issues": len(issues), "scanned_files": [c.get("path") or c.get("file") for c in components][:15], "other_metadata": other_fields, "top_issues": [ { "severity": i.get("severity"), "category": i.get("category") or i.get("type"), "message": i.get("finding") or i.get("message") or i.get("description"), "file": (i.get("location") or {}).get("file") or (i.get("location") or {}).get("path"), } for i in issues[:20] ], } prompt = ( "Du bekommst das JSON-Ergebnis eines automatisierten Sicherheits-Scans für einen " "AI-Agent-Skill, sowie (falls vorhanden) den Anfang der eigentlichen Skill-Datei. " "Schreibe auf Deutsch einen kurzen, laienverständlichen Bericht für jemanden ohne " "Security-Hintergrund, als Fließtext in 4 kurzen Absätzen, getrennt durch eine " "Leerzeile (keine Überschriften, keine Code-Blöcke, keine Aufzählungszeichen):\n" "Absatz 1: Wofür dieser Skill gedacht ist – nutze dafür in erster Linie den Auszug " "aus der Skill-Datei (Name/Beschreibung/Einleitung), 1-2 Sätze. Nur falls kein " "Auszug vorliegt oder daraus kein Zweck erkennbar ist, stütze dich ersatzweise auf " "Dateinamen und Fundtexte aus dem Scan-Ergebnis; ist auch daraus nichts erkennbar, " "schreibe das ehrlich so.\n" "Absatz 2: Gesamteinschätzung des Risikos in 1-2 Sätzen.\n" "Absatz 3: Die wichtigsten Funde in einfachen Worten, ohne unerklärte Fachbegriffe.\n" "Absatz 4: Konkrete Handlungsempfehlung (bedenkenlos nutzbar / mit Vorsicht " "nutzbar / besser nicht installieren) und eine kurze Begründung warum.\n\n" f"Auszug aus der Skill-Datei:\n{skill_context or '(kein Auszug verfügbar)'}\n\n" f"Scan-Ergebnis:\n{json.dumps(compact, ensure_ascii=False)}" ) payload = json.dumps( { "model": SUMMARY_MODEL, "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, "max_tokens": 700, } ).encode("utf-8") req = urllib.request.Request( f"{OPENAI_BASE_URL.rstrip('/')}/chat/completions", data=payload, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {OPENAI_API_KEY}", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=60) as resp: data = json.loads(resp.read().decode("utf-8")) return data["choices"][0]["message"]["content"].strip() except Exception: return None # --------------------------------------------------------------------------- # # 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 result %}

Gescannt am {{ scanned_at }} · Ziel: {{ target_label }} ·

{% if summary_paragraphs %}

Verständliche Zusammenfassung

{% for p in summary_paragraphs %}

{{ p }}

{% endfor %}
{% elif summary_failed %}

Die deutsche Zusammenfassung konnte nicht erstellt werden (KI-Anbieter nicht erreichbar oder Fehler). Der technische Bericht unten ist trotzdem vollständig.

{% endif %}
{{ result.risk_assessment.score }}/100 {{ result.risk_assessment.recommendation }}
{% if quick_mode %} {% if issues_by_severity %}

{{ total_issues }} Fund(e) insgesamt — {% for sev, items in issues_by_severity.items() %}{{ severity_labels.get(sev, sev) }}: {{ items|length }}{% if not loop.last %} · {% endif %}{% endfor %}

{% else %}

Keine Funde.

{% endif %}

Für Details je Fund und die Komponenten-Übersicht "Bericht" oder "Ausführlicher Bericht" wählen.

{% elif not issues_by_severity %}

Keine Funde – das Ergebnis unten (Rohdaten) bestätigt das im Detail.

{% endif %}
{% if not quick_mode %} {% for severity, items in issues_by_severity.items() %}

{{ severity_labels.get(severity, severity) }} ({{ items|length }})

{% for issue in items %}
{{ issue.category }} {% if issue.confidence %}Konfidenz: {{ issue.confidence }}{% endif %}

{{ issue.message }}

{% if issue.file %}

{{ issue.file }}{% if issue.line %}:{{ issue.line }}{% endif %}

{% endif %} {% if issue.explanation %}

{{ issue.explanation }}

{% endif %}
{% endfor %}
{% endfor %} {% if components %}

Gescannte Komponenten ({{ components|length }})

{% for c in components %} {% endfor %}
DateiTypZeilenAusführbar
{{ c.file }} {{ c.type }} {{ c.lines }} {{ 'ja' if c.executable else 'nein' }}
{% endif %}
Rohdaten (vollständiges JSON) 
{{ result_json }}
{% endif %} {% 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.", ) report_level = request.form.get("report_level", "standard") if report_level not in ("quick", "standard", "full"): report_level = "standard" use_llm = ALLOW_LLM and report_level == "full" quick_mode = report_level == "quick" 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()) raw_issues = result.get("issues") or result.get("findings") or [] summary_text = None summary_failed = False if use_llm: skill_context = extract_skill_context( target, is_zip=bool(zip_file and zip_file.filename), work_dir=work_dir ) summary_text = generate_report_summary(result, skill_context) summary_failed = summary_text is None target_label = git_url if git_url else (zip_file.filename if zip_file else "-") return render_template_string( PAGE, allow_llm=ALLOW_LLM, result=result, issues_by_severity=normalize_issues(raw_issues), total_issues=len(raw_issues), quick_mode=quick_mode, components=normalize_components(result.get("components", [])), severity_labels=SEVERITY_LABELS_DE, summary_paragraphs=[p for p in (summary_text or "").split("\n\n") if p.strip()], summary_failed=summary_failed, result_json=json.dumps(result, indent=2, ensure_ascii=False), scanned_at=datetime.now().strftime("%d.%m.%Y %H:%M"), target_label=target_label, 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"}