feat(scanner): upgrade runtime to 2.1.0

Signed-off-by: dongmucat <1127093059@qq.com>

scanner/Dockerfile

@@ -1,27 +1,20 @@
-FROM python:3.11-alpine
-
-ARG SKILL_SCANNER_VERSION=1.0.2
+FROM python:3.11-slim-bookworm
 
 WORKDIR /app
 
-COPY backports/apply_1_0_2_llm_base_url_backport.py /tmp/apply_1_0_2_llm_base_url_backport.py
-COPY skillhub_scanner_app.py /app/skillhub_scanner_app.py
+RUN pip install --no-cache-dir --only-binary=:all: \
+        "cisco-ai-skill-scanner==2.1.0" && \
+    groupadd --system app && \
+    useradd --system --gid app --home-dir /nonexistent --no-create-home app && \
+    install -d -o app -g app /tmp/skillhub-scans
 
-RUN pip install --no-cache-dir \
-        "cisco-ai-skill-scanner==${SKILL_SCANNER_VERSION}" \
-        "litellm==1.90.2" && \
-    python /tmp/apply_1_0_2_llm_base_url_backport.py /usr/local/lib/python3.11/site-packages && \
-    rm /tmp/apply_1_0_2_llm_base_url_backport.py && \
-    addgroup -S app && \
-    adduser -S app -G app && \
-    mkdir -p /tmp/skillhub-scans && \
-    chown app:app /tmp/skillhub-scans
+COPY skillhub_scanner_app.py /app/skillhub_scanner_app.py
 
 USER app
 
 EXPOSE 8000
 
 HEALTHCHECK --interval=10s --timeout=3s \
-  CMD wget -qO- http://127.0.0.1:8000/health || exit 1
+  CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=2).read()"]
 
 CMD ["uvicorn", "skillhub_scanner_app:app", "--host", "0.0.0.0", "--port", "8000"]

scanner/backports/apply_1_0_2_llm_base_url_backport.py

@@ -1,62 +0,0 @@
-#!/usr/bin/env python3
-"""Backport SKILL_SCANNER_LLM_BASE_URL support into cisco-ai-skill-scanner 1.0.2."""
-
-from __future__ import annotations
-
-import re
-import sys
-from pathlib import Path
-
-EXPECTED_DIST_INFO = "cisco_ai_skill_scanner-1.0.2.dist-info"
-ROUTER_RELATIVE_PATH = Path("skill_scanner/api/router.py")
-
-def replace_exact(content: str, old: str, new: str, expected_count: int, label: str) -> str:
-    actual_count = content.count(old)
-    if actual_count != expected_count:
-        raise SystemExit(f"Expected {expected_count} occurrences of {label}, found {actual_count}.")
-    return content.replace(old, new, expected_count)
-
-
-def replace_regex(content: str, pattern: str, replacement: str, expected_count: int, label: str) -> str:
-    updated, actual_count = re.subn(pattern, replacement, content, count=expected_count, flags=re.MULTILINE)
-    if actual_count != expected_count:
-        raise SystemExit(f"Expected {expected_count} regex replacements for {label}, found {actual_count}.")
-    return updated
-
-
-def main() -> int:
-    site_packages = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("/usr/local/lib/python3.11/site-packages")
-    dist_info = site_packages / EXPECTED_DIST_INFO
-    if not dist_info.exists():
-        raise SystemExit(f"Expected {EXPECTED_DIST_INFO} under {site_packages}, but it was not found.")
-
-    router_path = site_packages / ROUTER_RELATIVE_PATH
-    content = router_path.read_text(encoding="utf-8")
-    content = replace_regex(
-        content,
-        r'^(?P<indent>\s*)llm_model = os.getenv\("SKILL_SCANNER_LLM_MODEL"\)$',
-        r'\g<0>\n\g<indent>llm_base_url = os.getenv("SKILL_SCANNER_LLM_BASE_URL")',
-        2,
-        "llm_model environment lookup",
-    )
-    content = replace_exact(
-        content,
-        "LLMAnalyzer(model=llm_model)",
-        "LLMAnalyzer(model=llm_model, base_url=llm_base_url)",
-        2,
-        "LLMAnalyzer model constructor",
-    )
-    content = replace_exact(
-        content,
-        "LLMAnalyzer(provider=provider_str)",
-        "LLMAnalyzer(provider=provider_str, base_url=llm_base_url)",
-        2,
-        "LLMAnalyzer provider constructor",
-    )
-
-    router_path.write_text(content, encoding="utf-8")
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

scanner/skillhub_scanner_app.py

@@ -3,8 +3,11 @@
 import asyncio
 import logging
 import os
+import re
 import shutil
 import tempfile
+from functools import wraps
+from importlib import import_module
 from pathlib import Path
 from typing import NoReturn
 
@@ -13,19 +16,56 @@ from fastapi.responses import JSONResponse
 from skill_scanner.api.api import app
 
 
+_upstream_router = import_module("skill_scanner.api.router")
 _MAX_CONCURRENT_SCANS = max(1, int(os.getenv("SKILLHUB_SCANNER_MAX_CONCURRENT_SCANS", "1")))
 _HARD_TIMEOUT_SECONDS = max(1, int(os.getenv("SKILLHUB_SCANNER_HARD_TIMEOUT_SECONDS", "930")))
+_upstream_router.MAX_UPLOAD_SIZE_BYTES = max(
+    1, int(os.getenv("SKILLHUB_SCANNER_MAX_UPLOAD_SIZE_BYTES", "110100480"))
+)
 _active_scans = 0
 _active_scans_guard = asyncio.Lock()
 _SCAN_PATHS = {"/scan", "/scan-upload"}
+_SUPPORTED_TOKEN_PATTERN = re.compile(
+    r"\b(?:gh[pousr]_[A-Za-z0-9]{20,255}|sk-(?:proj-)?[A-Za-z0-9_-]{20,255})\b"
+)
 _log = logging.getLogger(__name__)
 
 
+def _redact_supported_tokens(value):
+    if isinstance(value, str):
+        return _SUPPORTED_TOKEN_PATTERN.sub("<redacted>", value)
+    if isinstance(value, list):
+        return [_redact_supported_tokens(item) for item in value]
+    if isinstance(value, dict):
+        return {key: _redact_supported_tokens(item) for key, item in value.items()}
+    return value
+
+
+def _install_scan_response_redaction() -> None:
+    """Redact supported token forms before FastAPI serializes scan findings."""
+    for route in _upstream_router.router.routes:
+        if getattr(route, "path", None) not in _SCAN_PATHS or "POST" not in getattr(route, "methods", set()):
+            continue
+        endpoint = route.endpoint
+
+        @wraps(endpoint)
+        async def redacting_endpoint(*args, __endpoint=endpoint, **kwargs):
+            response = await __endpoint(*args, **kwargs)
+            findings = getattr(response, "findings", None)
+            if isinstance(findings, list):
+                response.findings = _redact_supported_tokens(findings)
+            return response
+
+        route.endpoint = redacting_endpoint
+        route.dependant.call = redacting_endpoint
+
+
 def _cleanup_stale_scan_directories(temp_root: Path | None = None) -> None:
     """Remove incomplete upstream extraction directories left by a process restart."""
     root = temp_root or Path(tempfile.gettempdir())
+    current_upload_root = Path(_upstream_router._API_UPLOAD_ROOT).resolve()
     for candidate in root.glob("skill_scanner_*"):
-        if not candidate.is_dir():
+        if not candidate.is_dir() or candidate.resolve() == current_upload_root:
             continue
         try:
             shutil.rmtree(candidate)
@@ -54,6 +94,7 @@ async def _await_scan_until(scan_task: asyncio.Task, deadline: float, request_pa
         _restart_after_hard_timeout(request_path)
 
 
+_install_scan_response_redaction()
 app.router.add_event_handler("startup", _cleanup_stale_scan_directories)
 
 

scanner/tests/test_skillhub_scanner_app.py

@@ -1,5 +1,6 @@
 import asyncio
 import importlib.util
+import os
 import sys
 import tempfile
 import types
@@ -19,6 +20,26 @@ class _FakeRouter:
 class _FakeApp:
     def __init__(self):
         self.router = _FakeRouter()
+        self.routes = [object()]
+        github_canary = "ghp_" + "A1b2C3d4E5f6G7h8I9j0K1l2M3n4O5p6Q7r8"
+        openai_canary = "sk-proj-" + "Z9y8X7w6V5u4T3s2R1q0" * 3
+        self.upstream_routes = [
+            _FakeRoute(
+                "/scan",
+                _FakeScanResponse(
+                    [
+                        {
+                            "description": f"YARA match: {github_canary}",
+                            "metadata": {"evidence": openai_canary},
+                        }
+                    ]
+                ),
+            ),
+            _FakeRoute(
+                "/scan-upload",
+                _FakeScanResponse([{"description": "No credentials", "metadata": {"safe": True}}]),
+            ),
+        ]
 
     def middleware(self, _kind):
         return lambda function: function
@@ -31,30 +52,55 @@ class _FakeResponse:
         self.headers = headers
 
 
+class _FakeScanResponse:
+    def __init__(self, findings):
+        self.findings = findings
+        self.status_code = 200
+        self.headers = {"X-Contract": "preserved"}
+
+
+class _FakeRoute:
+    def __init__(self, path, response):
+        self.path = path
+        self.methods = {"POST"}
+
+        async def endpoint():
+            return response
+
+        self.endpoint = endpoint
+        self.dependant = types.SimpleNamespace(call=endpoint)
+
+
 class _Request:
     method = "POST"
     url = types.SimpleNamespace(path="/scan-upload")
 
 
-def _load_module():
+def _load_module(environment=None):
     fastapi = types.ModuleType("fastapi")
     fastapi.Request = object
     responses = types.ModuleType("fastapi.responses")
     responses.JSONResponse = _FakeResponse
     api = types.ModuleType("skill_scanner.api.api")
     api.app = _FakeApp()
+    router = types.ModuleType("skill_scanner.api.router")
+    router.MAX_UPLOAD_SIZE_BYTES = -1
+    router._API_UPLOAD_ROOT = Path(tempfile.gettempdir()) / "skill_scanner_current"
+    router.router = types.SimpleNamespace(routes=api.app.upstream_routes)
     stubs = {
         "fastapi": fastapi,
         "fastapi.responses": responses,
         "skill_scanner": types.ModuleType("skill_scanner"),
         "skill_scanner.api": types.ModuleType("skill_scanner.api"),
         "skill_scanner.api.api": api,
+        "skill_scanner.api.router": router,
     }
-    with patch.dict(sys.modules, stubs):
+    with patch.dict(os.environ, environment or {}, clear=True), patch.dict(sys.modules, stubs):
         module_path = Path(__file__).parents[1] / "skillhub_scanner_app.py"
         spec = importlib.util.spec_from_file_location("skillhub_scanner_app_under_test", module_path)
         module = importlib.util.module_from_spec(spec)
         spec.loader.exec_module(module)
+        module._router_stub = router
         return module
 
 
@@ -120,6 +166,51 @@ class SkillHubScannerAppTest(unittest.IsolatedAsyncioTestCase):
             self.assertFalse(stale.exists())
             self.assertTrue(unrelated.exists())
 
+    async def test_startup_cleanup_keeps_current_upstream_upload_directory(self):
+        with tempfile.TemporaryDirectory() as temp_root:
+            root = Path(temp_root)
+            current = root / "skill_scanner_current"
+            stale = root / "skill_scanner_stale"
+            current.mkdir()
+            stale.mkdir()
+            self.module._router_stub._API_UPLOAD_ROOT = current
+
+            self.module._cleanup_stale_scan_directories(root)
+
+            self.assertTrue(current.exists())
+            self.assertFalse(stale.exists())
+
+    async def test_default_upload_limit_matches_skillhub_package_limit(self):
+        self.assertEqual(110100480, self.module._router_stub.MAX_UPLOAD_SIZE_BYTES)
+
+    async def test_upload_limit_can_be_overridden_by_environment(self):
+        module = _load_module({"SKILLHUB_SCANNER_MAX_UPLOAD_SIZE_BYTES": "123456"})
+
+        self.assertEqual(123456, module._router_stub.MAX_UPLOAD_SIZE_BYTES)
+
+    async def test_upload_limit_is_at_least_one_byte(self):
+        module = _load_module({"SKILLHUB_SCANNER_MAX_UPLOAD_SIZE_BYTES": "0"})
+
+        self.assertEqual(1, module._router_stub.MAX_UPLOAD_SIZE_BYTES)
+
+    async def test_scan_findings_redact_supported_tokens_without_changing_response_contract(self):
+        route = next(route for route in self.module._router_stub.router.routes if route.path == "/scan")
+
+        response = await route.endpoint()
+
+        self.assertNotIn("ghp_", str(response.findings))
+        self.assertNotIn("sk-proj-", str(response.findings))
+        self.assertEqual(200, response.status_code)
+        self.assertEqual({"X-Contract": "preserved"}, response.headers)
+
+    async def test_safe_scan_findings_are_unchanged(self):
+        route = next(route for route in self.module._router_stub.router.routes if route.path == "/scan-upload")
+        expected = [{"description": "No credentials", "metadata": {"safe": True}}]
+
+        response = await route.dependant.call()
+
+        self.assertEqual(expected, response.findings)
+
     async def test_startup_cleanup_is_registered_on_the_upstream_router(self):
         self.assertEqual(
             [("startup", self.module._cleanup_stale_scan_directories)],

scripts/tests/scanner-2-1-contract-test.sh

@@ -0,0 +1,288 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+TMP_DIR="$(mktemp -d)"
+CONTAINERS=()
+
+cleanup() {
+  local status=$?
+  trap - EXIT
+  if ((${#CONTAINERS[@]})); then
+    docker rm -f "${CONTAINERS[@]}" >/dev/null 2>&1 || true
+  fi
+  rm -rf "$TMP_DIR"
+  exit "$status"
+}
+trap cleanup EXIT
+
+if [[ -n "${SCANNER_IMAGE:-}" ]]; then
+  IMAGE="$SCANNER_IMAGE"
+else
+  IMAGE="skillhub-scanner-contract-test:$(date +%s)"
+  docker build -t "$IMAGE" "$REPO_ROOT/scanner"
+fi
+
+python3 - "$TMP_DIR" <<'PY'
+import json
+import sys
+import zipfile
+from pathlib import Path
+
+root = Path(sys.argv[1])
+github_canary = "ghp_" + "A1b2C3d4E5f6G7h8I9j0K1l2M3n4O5p6Q7r8"
+openai_canary = "sk-proj-" + "Z9y8X7w6V5u4T3s2R1q0" * 3
+
+
+def write_zip(name, files):
+    with zipfile.ZipFile(root / name, "w", compression=zipfile.ZIP_STORED) as archive:
+        for path, content in files.items():
+            archive.writestr(path, content)
+
+
+manifest = """---
+name: contract-skill
+description: Scanner 2.1 runtime contract fixture.
+---
+
+Harmless contract fixture.
+"""
+write_zip("safe.zip", {"contract-skill/SKILL.md": manifest})
+write_zip(
+    "javascript-secret.zip",
+    {
+        "contract-skill/SKILL.md": manifest,
+        "contract-skill/index.js": f'const githubToken = "{github_canary}";\n',
+    },
+)
+write_zip(
+    "dotenv-secret.zip",
+    {
+        "contract-skill/SKILL.md": manifest,
+        "contract-skill/.env": f"OPENAI_API_KEY={openai_canary}\n",
+    },
+)
+
+large_files = {"large-skill/SKILL.md": manifest.replace("contract-skill", "large-skill")}
+for index in range(5):
+    large_files[f"large-skill/payload-{index}.txt"] = b"x" * (10 * 1024 * 1024)
+large_files["large-skill/payload-5.txt"] = b"x" * (1024 * 1024)
+write_zip("large-51mib.zip", large_files)
+
+(root / "canaries.json").write_text(
+    json.dumps({"github": github_canary, "openai": openai_canary}), encoding="utf-8"
+)
+
+local_skill = root / "local-scan" / "contract-skill"
+local_skill.mkdir(parents=True)
+(local_skill / "SKILL.md").write_text(manifest, encoding="utf-8")
+PY
+
+start_scanner() {
+  local name=$1
+  shift
+  docker run -d --name "$name" -p 127.0.0.1::8000 "$@" "$IMAGE" >/dev/null
+  CONTAINERS+=("$name")
+  SCANNER_PORT="$(docker port "$name" 8000/tcp | awk -F: 'END {print $NF}')"
+}
+
+wait_for_health() {
+  local url=$1
+  python3 - "$url" <<'PY'
+import json
+import sys
+import time
+import urllib.error
+import urllib.request
+
+url = sys.argv[1]
+last_error = None
+for _ in range(120):
+    try:
+        with urllib.request.urlopen(url + "/health", timeout=2) as response:
+            payload = json.load(response)
+            if response.status == 200:
+                break
+    except (OSError, urllib.error.URLError, json.JSONDecodeError) as error:
+        last_error = error
+    time.sleep(1)
+else:
+    raise SystemExit(f"scanner did not become healthy: {last_error}")
+
+if payload.get("version") != "2.1.0":
+    raise SystemExit(f"expected scanner 2.1.0, got {payload!r}")
+expected = {"static_analyzer", "bytecode_analyzer", "pipeline_analyzer"}
+available = set(payload.get("analyzers_available", []))
+if not expected.issubset(available):
+    raise SystemExit(f"missing deterministic analyzers: {sorted(expected - available)}")
+PY
+}
+
+post_zip() {
+  local url=$1
+  local archive=$2
+  local output=$3
+  python3 - "$url" "$archive" "$output" <<'PY'
+import json
+import sys
+import urllib.error
+import urllib.request
+import uuid
+from pathlib import Path
+
+url, archive_path, output_path = sys.argv[1:]
+boundary = "----skillhub-" + uuid.uuid4().hex
+archive = Path(archive_path).read_bytes()
+prefix = (
+    f"--{boundary}\r\n"
+    'Content-Disposition: form-data; name="policy"\r\n\r\n'
+    "balanced\r\n"
+    f"--{boundary}\r\n"
+    f'Content-Disposition: form-data; name="file"; filename="{Path(archive_path).name}"\r\n'
+    "Content-Type: application/zip\r\n\r\n"
+).encode()
+body = prefix + archive + f"\r\n--{boundary}--\r\n".encode()
+request = urllib.request.Request(
+    url + "/scan-upload",
+    data=body,
+    headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
+    method="POST",
+)
+try:
+    with urllib.request.urlopen(request, timeout=900) as response:
+        raw = response.read()
+        status = response.status
+except urllib.error.HTTPError as error:
+    raw = error.read()
+    status = error.code
+Path(output_path).write_bytes(raw)
+if status != 200:
+    raise SystemExit(f"scan upload returned HTTP {status}: {raw[:1000]!r}")
+try:
+    payload = json.loads(raw)
+except json.JSONDecodeError as error:
+    raise SystemExit(f"scan upload did not return JSON: {error}") from error
+required = {"scan_id", "skill_name", "findings", "scan_metadata"}
+if not required.issubset(payload) or not isinstance(payload["findings"], list):
+    raise SystemExit(f"invalid scan response contract: {payload!r}")
+PY
+}
+
+assert_hardcoded_secret() {
+  local response=$1
+  python3 - "$response" <<'PY'
+import json
+import sys
+
+payload = json.load(open(sys.argv[1], encoding="utf-8"))
+if not any(finding.get("category") == "hardcoded_secrets" for finding in payload["findings"]):
+    raise SystemExit("expected a hardcoded_secrets finding")
+PY
+}
+
+MAIN_CONTAINER="skillhub-scanner-contract-main-$$"
+start_scanner "$MAIN_CONTAINER"
+MAIN_PORT="$SCANNER_PORT"
+MAIN_URL="http://127.0.0.1:$MAIN_PORT"
+wait_for_health "$MAIN_URL"
+
+post_zip "$MAIN_URL" "$TMP_DIR/safe.zip" "$TMP_DIR/safe.json"
+post_zip "$MAIN_URL" "$TMP_DIR/javascript-secret.zip" "$TMP_DIR/javascript-secret.json"
+post_zip "$MAIN_URL" "$TMP_DIR/dotenv-secret.zip" "$TMP_DIR/dotenv-secret.json"
+post_zip "$MAIN_URL" "$TMP_DIR/large-51mib.zip" "$TMP_DIR/large-51mib.json"
+
+assert_hardcoded_secret "$TMP_DIR/javascript-secret.json"
+assert_hardcoded_secret "$TMP_DIR/dotenv-secret.json"
+
+python3 - "$TMP_DIR/safe.json" <<'PY'
+import json
+import sys
+
+payload = json.load(open(sys.argv[1], encoding="utf-8"))
+cel = payload.get("scan_metadata", {}).get("cel", {})
+if cel.get("runtime") != "cel-go":
+    raise SystemExit(f"expected CEL runtime cel-go, got {cel!r}")
+if not cel.get("runtime_version"):
+    raise SystemExit(f"expected a non-empty CEL runtime version, got {cel!r}")
+if cel.get("fallbacks") != 0 or cel.get("errors") != []:
+    raise SystemExit(f"unexpected CEL fallback/error telemetry: {cel!r}")
+PY
+
+docker logs "$MAIN_CONTAINER" >"$TMP_DIR/main-container.log" 2>&1
+python3 - "$TMP_DIR/canaries.json" "$TMP_DIR" <<'PY'
+import json
+import sys
+from pathlib import Path
+
+canaries = json.load(open(sys.argv[1], encoding="utf-8"))
+root = Path(sys.argv[2])
+for path in [*root.glob("*.json"), root / "main-container.log"]:
+    if path.name == "canaries.json":
+        continue
+    content = path.read_text(encoding="utf-8", errors="replace")
+    for label, canary in canaries.items():
+        if canary in content:
+            def find_canary(value, location="$"):
+                if isinstance(value, dict):
+                    for key, child in value.items():
+                        found = find_canary(child, f"{location}.{key}")
+                        if found:
+                            return found
+                elif isinstance(value, list):
+                    for index, child in enumerate(value):
+                        found = find_canary(child, f"{location}[{index}]")
+                        if found:
+                            return found
+                elif isinstance(value, str) and canary in value:
+                    return location
+                return None
+
+            location = None
+            if path.suffix == ".json":
+                location = find_canary(json.loads(content))
+            raise SystemExit(
+                f"full {label} canary leaked through {path.name}"
+                + (f" at {location}" if location else "")
+            )
+PY
+
+LOCAL_CONTAINER="skillhub-scanner-contract-local-$$"
+start_scanner "$LOCAL_CONTAINER" \
+  -e SKILL_SCANNER_ALLOWED_ROOTS=/tmp/skillhub-scans \
+  -v "$TMP_DIR/local-scan:/tmp/skillhub-scans:ro"
+LOCAL_PORT="$SCANNER_PORT"
+LOCAL_URL="http://127.0.0.1:$LOCAL_PORT"
+wait_for_health "$LOCAL_URL"
+
+python3 - "$LOCAL_URL" <<'PY'
+import json
+import sys
+import urllib.error
+import urllib.request
+
+url = sys.argv[1]
+
+
+def scan(path):
+    request = urllib.request.Request(
+        url + "/scan",
+        data=json.dumps({"skill_directory": path, "policy": "balanced"}).encode(),
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=900) as response:
+            return response.status, json.load(response)
+    except urllib.error.HTTPError as error:
+        return error.code, json.loads(error.read())
+
+
+inside_status, inside = scan("/tmp/skillhub-scans/contract-skill")
+if inside_status != 200 or not isinstance(inside, dict):
+    raise SystemExit(f"allowed local scan failed: HTTP {inside_status}: {inside!r}")
+outside_status, outside = scan("/etc")
+if outside_status not in (403, 404) or not isinstance(outside, dict):
+    raise SystemExit(f"outside-root scan should be denied, got HTTP {outside_status}: {outside!r}")
+PY
+
+echo "scanner-2-1-contract-test passed"

scripts/tests/scanner-llm-base-url-test.sh

@@ -1,232 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
-SCANNER_DIR="$REPO_ROOT/scanner"
-TMP_DIRS=()
-
-cleanup() {
-  local status=$?
-  local d
-  for d in "${TMP_DIRS[@]+"${TMP_DIRS[@]}"}"; do
-    rm -rf "$d"
-  done
-  exit "$status"
-}
-trap cleanup EXIT
-
-new_tmp() {
-  local d
-  d="$(mktemp -d)"
-  TMP_DIRS+=("$d")
-  echo "$d"
-}
-
-fail() {
-  echo "FAIL: $*" >&2
-  exit 1
-}
-
-tmp="$(new_tmp)"
-skill_dir="$tmp/skill"
-mkdir -p "$skill_dir/demo-skill"
-
-cat >"$skill_dir/demo-skill/SKILL.md" <<'EOF'
----
-name: demo-skill
-description: Minimal valid skill used for scanner integration coverage.
-license: Apache-2.0
----
-
-This is a harmless demo skill used for scanner integration testing.
-EOF
-
-cat >"$skill_dir/demo-skill/run.sh" <<'EOF'
-#!/usr/bin/env sh
-echo "demo"
-EOF
-chmod +x "$skill_dir/demo-skill/run.sh"
-
-IMAGE_TAG="skillhub-scanner-llm-base-url-test:$(date +%s)"
-docker build --no-cache -t "$IMAGE_TAG" "$SCANNER_DIR" >/dev/null
-
-docker run --rm -i \
-  -v "$skill_dir:/work/skill:ro" \
-  --entrypoint python \
-  "$IMAGE_TAG" - <<'PY'
-import asyncio
-from datetime import datetime, timezone
-import http.server
-import io
-import inspect
-import json
-import os
-from pathlib import Path
-import threading
-import urllib.request
-import zipfile
-
-from fastapi.params import Query
-from skill_scanner.core.models import ScanResult
-import skill_scanner.api.router as router
-
-signature = inspect.signature(router.scan_uploaded_skill)
-if not isinstance(signature.parameters["use_llm"].default, Query):
-    raise SystemExit("scan-upload use_llm should remain a Query parameter")
-if not isinstance(signature.parameters["llm_provider"].default, Query):
-    raise SystemExit("scan-upload llm_provider should remain a Query parameter")
-
-state = {"base_urls": [], "paths": []}
-
-
-class Handler(http.server.BaseHTTPRequestHandler):
-    def log_message(self, format, *args):  # noqa: A003
-        return
-
-    def do_POST(self):  # noqa: N802
-        length = int(self.headers.get("content-length", "0"))
-        self.rfile.read(length)
-        state["paths"].append(self.path)
-
-        payload = json.dumps(
-            {
-                "id": "chatcmpl-test",
-                "object": "chat.completion",
-                "created": int(datetime.now(timezone.utc).timestamp()),
-                "model": "local-model",
-                "choices": [
-                    {
-                        "index": 0,
-                        "message": {"role": "assistant", "content": "No findings."},
-                        "finish_reason": "stop",
-                    }
-                ],
-                "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
-            }
-        ).encode("utf-8")
-        self.send_response(200)
-        self.send_header("Content-Type", "application/json")
-        self.send_header("Content-Length", str(len(payload)))
-        self.end_headers()
-        self.wfile.write(payload)
-
-
-server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
-thread = threading.Thread(target=server.serve_forever, daemon=True)
-thread.start()
-
-target_base_url = f"http://127.0.0.1:{server.server_port}/v1"
-os.environ["SKILL_SCANNER_LLM_BASE_URL"] = target_base_url
-os.environ["SKILL_SCANNER_LLM_MODEL"] = "test-model"
-
-
-class FakeStaticAnalyzer:
-    pass
-
-
-class FakeLLMAnalyzer:
-    def __init__(self, model=None, provider=None, base_url=None):
-        self.model = model
-        self.provider = provider
-        self.base_url = base_url
-        state["base_urls"].append(base_url)
-
-    def analyze(self, skill_path):
-        request = urllib.request.Request(
-            self.base_url + "/chat/completions",
-            data=b"{}",
-            headers={"Content-Type": "application/json"},
-            method="POST",
-        )
-        with urllib.request.urlopen(request, timeout=5) as response:
-            response.read()
-
-
-class FakeSkillScanner:
-    def __init__(self, analyzers):
-        self.analyzers = analyzers
-
-    def scan_skill(self, skill_path):
-        for analyzer in self.analyzers:
-            analyze = getattr(analyzer, "analyze", None)
-            if callable(analyze):
-                analyze(skill_path)
-
-        return ScanResult(
-            skill_name="demo-skill",
-            skill_directory=str(skill_path),
-            findings=[],
-            scan_duration_seconds=0.05,
-            analyzers_used=["fake-llm"],
-            timestamp=datetime.now(timezone.utc),
-        )
-
-
-router.StaticAnalyzer = FakeStaticAnalyzer
-router.LLMAnalyzer = FakeLLMAnalyzer
-router.SkillScanner = FakeSkillScanner
-router.LLM_AVAILABLE = True
-
-request = router.ScanRequest(
-    skill_directory="/work/skill/demo-skill",
-    use_llm=True,
-    llm_provider="openai",
-    use_behavioral=False,
-    use_aidefense=False,
-    aidefense_api_key=None,
-)
-
-def build_skill_archive_bytes(skill_root: str) -> bytes:
-    skill_path = Path(skill_root)
-    buffer = io.BytesIO()
-    with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as archive:
-        for path in skill_path.rglob("*"):
-            if path.is_file():
-                archive.writestr(str(path.relative_to(skill_path.parent)), path.read_bytes())
-    return buffer.getvalue()
-
-class FakeUploadFile:
-    def __init__(self, filename: str, payload: bytes):
-        self.filename = filename
-        self._payload = payload
-
-    async def read(self) -> bytes:
-        return self._payload
-
-try:
-    direct_response = asyncio.run(router.scan_skill(request))
-
-    upload_response = asyncio.run(
-        router.scan_uploaded_skill(
-            file=FakeUploadFile("demo-skill.zip", build_skill_archive_bytes("/work/skill/demo-skill")),
-            use_llm=True,
-            llm_provider="openai",
-            use_behavioral=False,
-            use_aidefense=False,
-            aidefense_api_key=None,
-        )
-    )
-finally:
-    server.shutdown()
-    thread.join(timeout=5)
-
-if not getattr(direct_response, "scan_id", None):
-    raise SystemExit("scan_skill should still return a scan response")
-if not getattr(upload_response, "scan_id", None):
-    raise SystemExit("scan_uploaded_skill should still return a scan response")
-if len(state["base_urls"]) != 2:
-    raise SystemExit(f"expected two LLM analyzer constructions, got {len(state['base_urls'])}")
-if any(base_url != target_base_url for base_url in state["base_urls"]):
-    raise SystemExit(f"expected every base_url to be {target_base_url}, got {state['base_urls']}")
-if len(state["paths"]) != 2:
-    raise SystemExit(f"expected two LLM requests, got {state['paths']}")
-if not all(path.startswith("/v1/") for path in state["paths"]):
-    raise SystemExit(f"expected every request path to start with /v1/, got {state['paths']}")
-PY
-
-grep -Fq "name: SKILL_SCANNER_LLM_BASE_URL" "$REPO_ROOT/deploy/k8s/base/scanner-deployment.yaml" \
-  || fail "Kubernetes scanner deployment must expose SKILL_SCANNER_LLM_BASE_URL"
-grep -Fq "skill-scanner-llm-base-url" "$REPO_ROOT/deploy/k8s/base/secret.yaml.example" \
-  || fail "Kubernetes secret example must document skill-scanner-llm-base-url"
-
-echo "scanner-llm-base-url-test passed"