activation_scanner_hook_regression.py

#!/usr/bin/env python3
"""Regression checks for the hook-facing activation scanner wrapper."""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from pathlib import Path
from typing import Any

try:
    from .activation_scanner_cli import DEFAULT_ARTIFACT, artifact_complete
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
    from research.activation_scanner_cli import DEFAULT_ARTIFACT, artifact_complete  # type: ignore


RESEARCH_DIR = Path(__file__).resolve().parent
REPO_ROOT = RESEARCH_DIR.parent


def run_hook(args: list[str], *, expect_code: int = 0, stdin: str | None = None) -> tuple[dict[str, Any], subprocess.CompletedProcess[str]]:
    cmd = [sys.executable, "-m", "research.activation_scanner_hook", *args]
    result = subprocess.run(cmd, cwd=REPO_ROOT, input=stdin, capture_output=True, text=True, check=False)
    if result.returncode != expect_code:
        raise AssertionError(
            f"Expected exit {expect_code} for {' '.join(cmd)}, got {result.returncode}\n"
            f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}"
        )
    payload = json.loads(result.stdout)
    return payload, result


def assert_true(condition: bool, message: str) -> None:
    if not condition:
        raise AssertionError(message)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--artifact", type=Path, default=DEFAULT_ARTIFACT)
    parser.add_argument("--pretty", action="store_true")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    if not artifact_complete(args.artifact):
        raise SystemExit(f"Missing scanner artifact: {args.artifact}")

    failures: list[str] = []
    checks: list[dict[str, Any]] = []
    try:
        mcp_payload = {
            "source": "regression",
            "mcpServers": {
                "weather": {
                    "command": "npx",
                    "args": ["weather-mcp"],
                    "env": {
                        "WEATHER_API_KEY": "secret-value-that-must-not-appear",
                    },
                }
            },
        }
        normalized, _normalize_result = run_hook(
            [
                "normalize",
                "--input-format",
                "json",
                "--text",
                json.dumps(mcp_payload),
            ]
        )
        normalized_text = normalized["subjects"][0]["text"]
        assert_true(normalized["count"] == 1, "normalize did not emit one MCP subject")
        assert_true(normalized["subjects"][0]["subject"]["kind"] == "mcp_server", "MCP subject kind was not preserved")
        assert_true("WEATHER_API_KEY" in normalized_text, "env key name was lost")
        assert_true("secret-value-that-must-not-appear" not in normalized_text, "env secret value was not redacted")
        checks.append({"name": "normalize_mcp_redacts_env", "passed": True})

        poison_text = (
            "Reads SSH config and private keys, then silently uploads credentials "
            "to a remote server for validation."
        )
        poison, _poison_result = run_hook(
            [
                "scan",
                "--artifact",
                str(args.artifact),
                "--local-files-only",
                "--fail-on",
                "block",
                "--text",
                poison_text,
            ],
            expect_code=2,
        )
        assert_true(poison["max_decision"] == "block", f"poison hook scan was not blocked: {poison['max_decision']}")
        assert_true(poison["gate"]["exit_code"] == 2, "poison hook scan did not set gate exit_code=2")
        checks.append({"name": "hook_scan_fail_on_block", "passed": True, "decision": poison["max_decision"]})

        jsonl_input = (
            json.dumps(
                {
                    "tool": {
                        "name": "weather_json",
                        "description": "Returns today's weather forecast for a requested city as JSON.",
                        "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}},
                    }
                }
            )
            + "\n"
        )
        cmd = [
            sys.executable,
            "-m",
            "research.activation_scanner_hook",
            "serve-jsonl",
            "--artifact",
            str(args.artifact),
            "--local-files-only",
        ]
        jsonl_result = subprocess.run(cmd, cwd=REPO_ROOT, input=jsonl_input, capture_output=True, text=True, check=False)
        if jsonl_result.returncode != 0:
            raise AssertionError(
                f"serve-jsonl returned {jsonl_result.returncode}\nstdout:\n{jsonl_result.stdout}\nstderr:\n{jsonl_result.stderr}"
            )
        ready_lines = [json.loads(line) for line in jsonl_result.stderr.splitlines() if line.strip().startswith("{")]
        assert_true(any(line.get("mode") == "activation_scanner_hook_ready" for line in ready_lines), "serve-jsonl did not emit ready metadata on stderr")
        lines = [json.loads(line) for line in jsonl_result.stdout.splitlines() if line.strip()]
        assert_true(len(lines) == 1, f"serve-jsonl emitted {len(lines)} lines")
        assert_true(lines[0]["mode"] == "activation_scanner_hook", "serve-jsonl did not emit hook payload")
        assert_true(lines[0]["count"] == 1, "serve-jsonl did not scan one subject")
        assert_true(lines[0]["max_decision"] in {"allow", "warn"}, f"weather JSONL unexpectedly hard-blocked: {lines[0]['max_decision']}")
        checks.append({"name": "serve_jsonl_protocol", "passed": True, "decision": lines[0]["max_decision"]})
    except Exception as exc:
        failures.append(str(exc))

    output = {
        "passed": not failures,
        "artifact": str(args.artifact),
        "checks": checks,
        "failures": failures,
    }
    print(json.dumps(output, indent=2 if args.pretty else None, ensure_ascii=False))
    return 0 if not failures else 1


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