scripts/audit_runner.py

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
AUDIT_DIR = ROOT / "audit"
BASE_URL = os.environ.get("KINK_AUDIT_BASE_URL", "http://127.0.0.1:8011")
HEADER_URL = BASE_URL if BASE_URL.endswith("/") else BASE_URL + "/"


def run_step(name: str, cmd: list[str], artifact_dir: Path, env: dict[str, str] | None = None, optional: bool = False) -> dict:
    started = time.time()
    merged_env = os.environ.copy()
    if env:
        merged_env.update(env)
    result = subprocess.run(cmd, cwd=ROOT, env=merged_env, capture_output=True, text=True)
    step = {
        "name": name,
        "command": cmd,
        "ok": result.returncode == 0,
        "optional": optional,
        "elapsed_ms": round((time.time() - started) * 1000, 1),
        "stdout_path": None,
        "stderr_path": None,
        "returncode": result.returncode,
    }
    stdout_path = artifact_dir / f"{name}.stdout.txt"
    stderr_path = artifact_dir / f"{name}.stderr.txt"
    stdout_path.write_text(result.stdout)
    stderr_path.write_text(result.stderr)
    step["stdout_path"] = str(stdout_path)
    step["stderr_path"] = str(stderr_path)
    step["status"] = "passed" if step["ok"] else ("skipped_or_failed_optional" if optional else "failed")
    return step


def which(path: str) -> str | None:
    return shutil.which(path)


def header_probe() -> dict:
    proc = subprocess.run(["curl", "--max-time", "10", "-sD", "-", HEADER_URL, "-o", "/dev/null"], cwd=ROOT, capture_output=True, text=True, check=False)
    headers = {}
    for line in proc.stdout.splitlines():
        if ":" in line:
            key, value = line.split(":", 1)
            headers[key.strip().lower()] = value.strip()
    issues = []
    if "cache-control" not in headers:
        issues.append("missing_cache_control")
    if "content-security-policy" not in headers:
        issues.append("missing_csp")
    if "x-frame-options" not in headers:
        issues.append("missing_x_frame_options")
    return {"headers": headers, "issues": issues, "ok": not issues}


def build_steps(tier: str, artifact_dir: Path) -> list[dict]:
    steps: list[dict] = []
    env = {"KINK_AUDIT_ARTIFACT_DIR": str(artifact_dir), "KINK_AUDIT_BASE_URL": BASE_URL}
    python = sys.executable

    core = [
        ("diagnose_ui_flow", [python, "scripts/diagnose_ui_flow.py"], False),
        ("ui_visual_audit", [python, "scripts/ui_visual_audit.py"], False),
        ("regression_checks", [python, "scripts/regression_checks.py"], False),
        ("audit_data_quality", [python, "scripts/audit_data_quality.py"], False),
    ]
    if tier == "deep":
        core.append(("record_product_flows", [python, "scripts/record_product_flows.py"], False))

    for name, cmd, optional in core:
        steps.append(run_step(name, cmd, artifact_dir, env=env, optional=optional))

    header_report = header_probe()
    steps.append({
        "name": "security_headers",
        "status": "passed" if header_report["ok"] else "failed",
        "ok": header_report["ok"],
        "optional": False,
        "report": header_report,
    })

    if which("npm"):
        for name, cmd, optional in [
            ("axe_core", ["npm", "--prefix", str(AUDIT_DIR), "run", "axe"], False),
            ("lhci", ["npm", "--prefix", str(AUDIT_DIR), "run", "lhci"], False),
        ]:
            steps.append(run_step(name, cmd, artifact_dir, env=env, optional=optional))

    if which("bandit"):
        steps.append(
            run_step(
                "bandit",
                [
                    "bandit",
                    "-c",
                    str(AUDIT_DIR / "config" / "bandit.yaml"),
                    "-r",
                    str(ROOT / "api.py"),
                    str(ROOT / "backend"),
                ],
                artifact_dir,
                optional=False,
            )
        )
    if which("pip-audit"):
        steps.append(run_step("pip_audit", ["pip-audit"], artifact_dir, optional=False))
    if which("semgrep"):
        steps.append(
            run_step(
                "semgrep",
                [
                    "semgrep",
                    "--config",
                    str(AUDIT_DIR / "config" / "semgrep.yml"),
                    "--exclude",
                    "node_modules",
                    "--exclude",
                    ".venv",
                    "--exclude",
                    ".cursor",
                    "--exclude",
                    "data",
                    "--exclude",
                    "frontend/dist",
                    str(ROOT),
                ],
                artifact_dir,
                optional=False,
            )
        )
    if which("schemathesis"):
        steps.append(run_step(
            "schemathesis",
            [
                "schemathesis", "run", f"{BASE_URL}/openapi.json",
                "--phases", "coverage",
                "--checks", "not_a_server_error,status_code_conformance,content_type_conformance,response_headers_conformance,response_schema_conformance,negative_data_rejection,unsupported_method",
                "--include-path-regex", r"^/$|^/health$|^/stats$|^/users$",
                "--max-failures=10",
                "--request-timeout", "5",
                "--max-examples", "3",
                "--generation-allow-x00=false",
                "--workers", "1",
            ],
            artifact_dir,
            optional=False,
        ))
    return steps


def main() -> int:
    parser = argparse.ArgumentParser(description="Run fast or deep audits for the kink CLI web app.")
    parser.add_argument("--tier", choices=["fast", "deep"], default="fast")
    args = parser.parse_args()

    artifact_dir = Path("/tmp/kink_audit") / f"{args.tier}_{int(time.time())}"
    artifact_dir.mkdir(parents=True, exist_ok=True)

    steps = build_steps(args.tier, artifact_dir)
    summary = {
        "tier": args.tier,
        "base_url": BASE_URL,
        "artifact_dir": str(artifact_dir),
        "generated_at": time.time(),
        "steps": steps,
    }
    manifest = artifact_dir / "manifest.json"
    manifest.write_text(json.dumps(summary, indent=2))
    print(json.dumps({"artifact_dir": str(artifact_dir), "manifest": str(manifest)}, indent=2))

    failures = [step for step in steps if not step.get("ok") and not step.get("optional")]
    return 1 if failures else 0


if __name__ == "__main__":
    raise SystemExit(main())