#!/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"