diff --git a/.github/workflows/reference-implementation.yml b/.github/workflows/reference-implementation.yml index 6222485..f11f8f8 100644 --- a/.github/workflows/reference-implementation.yml +++ b/.github/workflows/reference-implementation.yml @@ -22,6 +22,9 @@ on: # the Trace mappings at runtime, so a spec-only change can break them. - "specification/**" - ".github/workflows/reference-implementation.yml" + # The negative controls start this tree's Guardian, so they run here too. + - "tests/negative_controls_runner.py" + - "tests/test_negative_controls.py" push: branches: ["main", "integration"] paths: @@ -30,6 +33,9 @@ on: # the Trace mappings at runtime, so a spec-only change can break them. - "specification/**" - ".github/workflows/reference-implementation.yml" + # The negative controls start this tree's Guardian, so they run here too. + - "tests/negative_controls_runner.py" + - "tests/test_negative_controls.py" workflow_dispatch: permissions: {} @@ -83,3 +89,19 @@ jobs: # schemas in this repository rather than a copy of them. - name: Run the tests run: bun test + + - name: Install uv + uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0 + with: + version: "0.9.9" + + # The test job in deploy-pages.yml has no Bun, so the reference Guardian group in + # this file skips there. Here the variable turns a missing Guardian into a failure, + # so the group cannot skip in the one job that can run it. + - name: Run the negative controls against this Guardian + working-directory: . + env: + ACS_REQUIRE_REFERENCE_GUARDIAN: "1" + run: | + uv sync --locked + uv run pytest -v tests/test_negative_controls.py diff --git a/tests/negative_controls_runner.py b/tests/negative_controls_runner.py new file mode 100644 index 0000000..8ae2528 --- /dev/null +++ b/tests/negative_controls_runner.py @@ -0,0 +1,253 @@ +"""Post ACS-Core envelope vectors to a Guardian over HTTP and score them by error code. + + python3 -I tests/negative_controls_runner.py --guardian http://127.0.0.1:8787/acs + +This is the runner that `tests/test_negative_controls.py` holds to the requirements in +issue #53. It is standard library only and runs as its own process, one child per run, so +nothing it measures can reach its exit status (#53 requirement 4). + +Output is one line per vector, `PASS `, `FAIL ...` or `NOT-RUN ...`, then a +closing `RESULT pass= fail= not_run=` line. Exit 0 means every vector was scored +and passed, 1 means at least one failed, and 2 means nothing could be measured: the suite +is structurally invalid, or no vector was scored (#53 requirement 9). + +A vector names the section it tests and the spec version it was written against, and its +expectation is either an ALLOW decision or a JSON-RPC error code from section 17.1. Prose +in a message is never compared (#53 requirement 7). + +The signature is HMAC-SHA256, the ACS-Core baseline in section 10, over the RFC 8785 form +of the JSON-RPC request with `params.signature` removed. The per-session key is HKDF-SHA256 +with no salt, the session_id as info and a 32-byte output, applied to a pre-shared secret: +the parameters #272 writes into section 10. A Guardian under test holds the same secret, +passed with `--psk`. The canonical form here is +`json.dumps` with sorted keys and no whitespace, which equals RFC 8785 for the ASCII +strings and integers these vectors carry. +""" + +from __future__ import annotations + +import argparse +import base64 +import datetime +import hashlib +import hmac +import json +import sys +import urllib.error +import urllib.request +import uuid + +SPEC_VERSION = "0.1.4" +SPEC = "docs/spec/instrument/specification.md" +DEFAULT_PSK = "acs-negative-controls-test-key" + +# Each vector is data: the runner has no per-vector code. A step's fields say how its +# envelope differs from a fresh, signed, current request in the vector's session. +VECTORS = [ + { + "id": "signature-valid", + "group": "signature", + "kind": "positive", + "requirement": {"spec_version": SPEC_VERSION, "anchor": SPEC + "#10-cryptographic-signatures", + "quote": "A signature over the \u00a710 canonical input is REQUIRED in ACS-Core."}, + "steps": [{"signature": "valid"}], + "expect": {"decision": "allow"}, + }, + { + "id": "signature-invalid", + "group": "signature", + "kind": "negative", + "requirement": {"spec_version": SPEC_VERSION, "anchor": SPEC + "#10-cryptographic-signatures", + "quote": "A verifier MUST recompute this canonical form and MUST reject a " + "signature that does not cover it (`SIGNATURE_INVALID`, \u00a717.1)."}, + "steps": [{"signature": "invalid"}], + "expect": {"error": -32004}, + }, + { + "id": "signature-missing", + "group": "signature", + "kind": "negative", + "requirement": {"spec_version": SPEC_VERSION, "anchor": SPEC + "#171-acs-error-code-registry", + "quote": "A required signature is missing, malformed, or fails verification (\u00a710)."}, + "steps": [{"signature": "absent"}], + "expect": {"error": -32004}, + }, + { + "id": "request-id-fresh", + "group": "replay", + "kind": "positive", + "requirement": {"spec_version": SPEC_VERSION, "anchor": SPEC + "#103-replay-protection", + "quote": "MUST reject duplicate `request_id` values within the session"}, + "steps": [{}, {}], + "expect": {"decision": "allow"}, + }, + { + "id": "request-id-repeated", + "group": "replay", + "kind": "negative", + "requirement": {"spec_version": SPEC_VERSION, "anchor": SPEC + "#103-replay-protection", + "quote": "MUST reject duplicate `request_id` values within the session with " + "`REPLAY_DETECTED` (`-32005`, \u00a717.1)"}, + "steps": [{}, {"request_id": "repeat"}], + "expect": {"error": -32005}, + }, + { + "id": "timestamp-inside-window", + "group": "window", + "kind": "positive", + "requirement": {"spec_version": SPEC_VERSION, "anchor": SPEC + "#103-replay-protection", + "quote": "Guardians MUST reject requests whose `timestamp` is more than the " + "negotiated skew window"}, + "steps": [{"timestamp_offset_s": -60}], + "expect": {"decision": "allow"}, + }, + { + "id": "timestamp-outside-window", + "group": "window", + "kind": "negative", + "requirement": {"spec_version": SPEC_VERSION, "anchor": SPEC + "#103-replay-protection", + "quote": "returning `TIMESTAMP_OUT_OF_WINDOW` (`-32006`, \u00a717.1)"}, + "steps": [{"timestamp_offset_s": -86400}], + "expect": {"error": -32006}, + }, +] + +# Envelope properties a vector can vary, with the values that count as a change from a +# conformant request. A suite that never changes one of them cannot tell a Guardian that +# checks it from one that ignores it (#53 requirement 5). +VARIED = { + "signature": ("invalid", "absent"), + "request_id": ("repeat",), +} + + +def canonical(obj: object) -> bytes: + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + + +def session_key(psk: str, session_id: str) -> bytes: + """HKDF-SHA256 (RFC 5869), one 32-byte block, empty salt, session_id as info.""" + prk = hmac.new(b"\x00" * 32, psk.encode(), hashlib.sha256).digest() + return hmac.new(prk, session_id.encode() + b"\x01", hashlib.sha256).digest() + + +def sign(body: dict, psk: str) -> dict: + """Return the signature object for a request whose params carry no signature yet.""" + key = session_key(psk, body["params"]["metadata"]["session_id"]) + mac = hmac.new(key, canonical(body), hashlib.sha256).digest() + return {"algorithm": "HMAC-SHA256", "value": base64.b64encode(mac).decode(), "key_id": "test"} + + +def now() -> datetime.datetime: + return datetime.datetime.now(datetime.timezone.utc) + + +def envelope(step: dict, session_id: str, last_request_id: str | None, psk: str) -> dict: + request_id = last_request_id if step.get("request_id") == "repeat" else str(uuid.uuid4()) + when = now() + datetime.timedelta(seconds=step.get("timestamp_offset_s", 0)) + params = { + "acs_version": SPEC_VERSION, + "request_id": request_id, + "timestamp": when.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z", + "metadata": {"agent_id": "negative-controls", "session_id": session_id}, + "payload": {"tool": {"name": "Bash"}, "arguments": {"command": {"value": "ls"}}}, + } + body = {"jsonrpc": "2.0", "id": request_id, "method": "steps/toolCallRequest", "params": params} + mode = step.get("signature", "valid") + if mode == "valid": + params["signature"] = sign(body, psk) + elif mode == "invalid": + good = sign(body, psk) + mac = base64.b64decode(good["value"]) + flipped = bytes([mac[0] ^ 1]) + mac[1:] + params["signature"] = dict(good, value=base64.b64encode(flipped).decode()) + return body + + +def post(url: str, body: dict) -> dict: + request = urllib.request.Request(url, data=canonical(body), headers={"content-type": "application/json"}) + with urllib.request.urlopen(request, timeout=20) as response: + return json.loads(response.read().decode()) + + +def observe(response: dict) -> dict: + """Reduce a response to what is scored: an error code, or a decision.""" + if "error" in response: + return {"error": response["error"].get("code")} + return {"decision": (response.get("result") or {}).get("decision")} + + +def structural_errors(vectors: list) -> list: + """Rules the suite must meet before any vector is posted.""" + errors = [] + groups: dict = {} + for vector in vectors: + req = vector.get("requirement") or {} + if not req.get("spec_version") or not req.get("anchor"): + errors.append("%s names no requirement and spec version" % vector.get("id")) + groups.setdefault(vector.get("group"), set()).add(vector.get("kind")) + for group, kinds in sorted(groups.items(), key=lambda item: str(item[0])): + # A positive control with no negative beside it certifies an allow-everything + # Guardian, and a negative with no positive certifies a deny-everything one. + if kinds != {"positive", "negative"}: + errors.append("group %s needs a positive and a negative control, has %s" % (group, sorted(kinds))) + steps = [step for vector in vectors for step in vector.get("steps", [])] + for field, changed in VARIED.items(): + if steps and not any(step.get(field) in changed for step in steps): + errors.append("no vector changes %s, so a Guardian that ignores it passes" % field) + return errors + + +def score(vector: dict, observations: list) -> tuple[str, str]: + expect, last = vector["expect"], observations[-1] + if "decision" in expect: + ok = all(o.get("decision") == expect["decision"] for o in observations) + return ("PASS" if ok else "FAIL"), "expected %s observed %s" % (expect, observations) + ok = last.get("error") == expect["error"] + return ("PASS" if ok else "FAIL"), "expected %s observed %s" % (expect, last) + + +def run(url: str, vectors: list, psk: str) -> int: + errors = structural_errors(vectors) + if errors: + for error in errors: + print("SUITE-INVALID " + error) + print("RESULT pass=0 fail=0 not_run=%d" % len(vectors)) + return 2 + counts = {"PASS": 0, "FAIL": 0, "NOT-RUN": 0} + for vector in vectors: + session_id, last, observations = str(uuid.uuid4()), None, [] + try: + for step in vector["steps"]: + body = envelope(step, session_id, last, psk) + last = body["params"]["request_id"] + observations.append(observe(post(url, body))) + except (OSError, ValueError, urllib.error.URLError) as exc: + # A transport failure measured nothing, so it is never scored as a pass. + counts["NOT-RUN"] += 1 + print("NOT-RUN %s %s" % (vector["id"], type(exc).__name__)) + continue + verdict, detail = score(vector, observations) + counts[verdict] += 1 + print("FAIL %s %s" % (vector["id"], detail) if verdict == "FAIL" else "PASS " + vector["id"]) + print("RESULT pass=%d fail=%d not_run=%d" % (counts["PASS"], counts["FAIL"], counts["NOT-RUN"])) + if counts["PASS"] + counts["FAIL"] == 0: + return 2 + return 1 if counts["FAIL"] or counts["NOT-RUN"] else 0 + + +def main(argv: list | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--guardian", required=True, help="the Guardian's JSON-RPC URL") + parser.add_argument("--vectors", help="a JSON file holding a vector list; default: VECTORS") + parser.add_argument("--psk", default=DEFAULT_PSK, help="pre-shared secret for the HMAC key") + args = parser.parse_args(argv) + vectors = VECTORS + if args.vectors: + with open(args.vectors) as handle: + vectors = json.load(handle) + return run(args.guardian, vectors, args.psk) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_negative_controls.py b/tests/test_negative_controls.py new file mode 100644 index 0000000..f3af0d6 --- /dev/null +++ b/tests/test_negative_controls.py @@ -0,0 +1,436 @@ +"""Negative controls for the conformance runner, one group per #53 requirement it can test. + +Issue #53 lists what a v0.2 suite has to get right. Seven of those requirements are about +the gate itself rather than any one Guardian, so each is tested here by handing the runner +in `tests/negative_controls_runner.py` a subject or a corpus it must refuse: + +- 2, the gate has negative controls of its own: every runner mutant below is caught. +- 3, structural rules have a mirror: an allow-everything Guardian fails, and a suite with + only positive controls is refused. +- 4, the runner owns its process: every run is a child process, and a child that exits 0 + without a report is not read as a pass. +- 5, a constant cannot be measured: a Guardian that skips signature checks fails exactly + the signature vectors, and a suite whose signatures never change is refused. +- 7, codes, never prose: the right message with the wrong code fails, and the right code + with other wording passes. +- 9, a run that scores nothing does not exit 0. +- 11, CI reads the report: `read_report` fails a run whose exit code disagrees with it. + +The last group runs the same vectors against the reference Guardian in +`reference-implementations/agt`. Its four known gaps are strict expected failures, so the +day the Guardian closes one, this file fails until the marker is removed. +""" + +from __future__ import annotations + +import datetime +import importlib.util +import json +import os +import shutil +import signal +import socket +import subprocess +import sys +import threading +import time +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parent.parent +RUNNER = ROOT / "tests" / "negative_controls_runner.py" +AGT = ROOT / "reference-implementations" / "agt" + +_spec = importlib.util.spec_from_file_location("negative_controls_runner", RUNNER) +runner = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(runner) + +VECTOR_IDS = [vector["id"] for vector in runner.VECTORS] +NEGATIVE_IDS = {v["id"] for v in runner.VECTORS if v["kind"] == "negative"} +POSITIVE_IDS = set(VECTOR_IDS) - NEGATIVE_IDS +SIGNATURE_IDS = {"signature-invalid", "signature-missing"} + + +# --- reading a run -------------------------------------------------------------------- + + +@dataclass +class Report: + returncode: int + verdicts: dict + counts: dict + stdout: str + + +def read_report(proc: subprocess.CompletedProcess) -> Report: + """Score a run from its report, and fail when the exit code disagrees (requirement 11).""" + lines = proc.stdout.splitlines() + results = [line for line in lines if line.startswith("RESULT ")] + assert len(results) == 1 and lines[-1] == results[0], ( + "no closing RESULT line; exit %d, stdout %r" % (proc.returncode, proc.stdout[-500:])) + counts = {k: int(v) for k, v in (item.split("=") for item in results[0].split()[1:])} + verdicts = {} + for line in lines: + word, _, rest = line.partition(" ") + if word in ("PASS", "FAIL", "NOT-RUN"): + verdicts[rest.split(" ")[0]] = word + tally = {"pass": "PASS", "fail": "FAIL", "not_run": "NOT-RUN"} + invalid = any(line.startswith("SUITE-INVALID ") for line in lines) + if not invalid: + for key, word in tally.items(): + assert counts[key] == list(verdicts.values()).count(word), ( + "RESULT says %s=%d but the report lists %d" % (key, counts[key], list(verdicts.values()).count(word))) + if invalid or counts["pass"] + counts["fail"] == 0: + expected = 2 + elif counts["fail"] or counts["not_run"]: + expected = 1 + else: + expected = 0 + assert proc.returncode == expected, ( + "exit code %d disagrees with the report, which implies %d" % (proc.returncode, expected)) + return Report(proc.returncode, verdicts, counts, proc.stdout) + + +def run_runner(url: str, tmp_path: Path, vectors: list | None = None, script: Path = RUNNER) -> Report: + """One child process per run, isolated from this interpreter (requirement 4).""" + args = [sys.executable, "-I", str(script), "--guardian", url] + if vectors is not None: + path = tmp_path / ("vectors-%d.json" % time.monotonic_ns()) + path.write_text(json.dumps(vectors)) + args += ["--vectors", str(path)] + proc = subprocess.run(args, capture_output=True, text=True, timeout=120) + return read_report(proc) + + +# --- stub Guardians ------------------------------------------------------------------- + + +def _error(body: dict, code: int, message: str) -> dict: + return {"jsonrpc": "2.0", "id": body.get("id"), "error": {"code": code, "message": message}} + + +def _allow(body: dict) -> dict: + return {"jsonrpc": "2.0", "id": body.get("id"), "result": {"decision": "allow"}} + + +class Conformant: + """Checks the three properties the vectors vary, the way sections 10 and 10.3 say.""" + + def __init__(self, check_signature: bool = True, codes: dict | None = None, wording: str = "plain"): + self.check_signature = check_signature + self.codes = codes or {} + self.wording = wording + self.seen: set = set() + self.lock = threading.Lock() + + def refuse(self, body: dict, code: int, name: str) -> dict: + message = {"plain": name, "other": "request refused, see code"}[self.wording] + return _error(body, self.codes.get(code, code), message) + + def __call__(self, body: dict) -> dict: + params = body.get("params", {}) + sent = params.get("signature") + if self.check_signature: + unsigned = json.loads(json.dumps(body)) + unsigned["params"].pop("signature", None) + if not sent or sent != runner.sign(unsigned, runner.DEFAULT_PSK): + return self.refuse(body, -32004, "SIGNATURE_INVALID: signature invalid") + stamp = datetime.datetime.fromisoformat(params["timestamp"].replace("Z", "+00:00")) + if abs((runner.now() - stamp).total_seconds()) > 300: + return self.refuse(body, -32006, "TIMESTAMP_OUT_OF_WINDOW: timestamp outside window") + key = (params["metadata"]["session_id"], params["request_id"]) + with self.lock: + if key in self.seen: + return self.refuse(body, -32005, "REPLAY_DETECTED: duplicate request_id") + self.seen.add(key) + return _allow(body) + + +def allow_all(body: dict) -> dict: + return _allow(body) + + +def deny_all(body: dict) -> dict: + return _error(body, -32004, "SIGNATURE_INVALID") + + +def drop(body: dict) -> None: + return None + + +class Stub: + """Serve a policy on a loopback port; a policy returning None closes the connection unanswered.""" + + def __init__(self, policy): + policy_ref = policy + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["content-length"]))) + answer = policy_ref(body) + if answer is None: + self.close_connection = True + return + data = json.dumps(answer).encode() + self.send_response(200) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *args): + pass + + self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.url = "http://127.0.0.1:%d/acs" % self.server.server_address[1] + + def __enter__(self): + threading.Thread(target=self.server.serve_forever, daemon=True).start() + return self.url + + def __exit__(self, *exc): + self.server.shutdown() + self.server.server_close() + + +def vectors_where(predicate) -> list: + return [json.loads(json.dumps(v)) for v in runner.VECTORS if predicate(v)] + + +# --- controls, as functions of a runner script so the mutants can reuse them ---------- + + +def control_conformant_passes(script: Path, tmp_path: Path) -> None: + with Stub(Conformant()) as url: + report = run_runner(url, tmp_path, script=script) + assert report.returncode == 0 and set(report.verdicts.values()) == {"PASS"}, report.stdout + + +def control_allow_all_fails(script: Path, tmp_path: Path) -> None: + with Stub(allow_all) as url: + report = run_runner(url, tmp_path, script=script) + assert report.returncode == 1, report.stdout + assert {i for i, v in report.verdicts.items() if v == "FAIL"} == NEGATIVE_IDS, report.stdout + + +def control_deny_all_fails(script: Path, tmp_path: Path) -> None: + with Stub(deny_all) as url: + report = run_runner(url, tmp_path, script=script) + assert report.returncode == 1, report.stdout + assert POSITIVE_IDS <= {i for i, v in report.verdicts.items() if v == "FAIL"}, report.stdout + + +def control_positives_only_refused(script: Path, tmp_path: Path) -> None: + with Stub(allow_all) as url: + report = run_runner(url, tmp_path, vectors_where(lambda v: v["kind"] == "positive"), script=script) + assert report.returncode == 2 and "needs a positive and a negative control" in report.stdout, report.stdout + + +def control_constant_signature_refused(script: Path, tmp_path: Path) -> None: + vectors = vectors_where(lambda v: True) + for vector in vectors: + for step in vector["steps"]: + step.pop("signature", None) + with Stub(allow_all) as url: + report = run_runner(url, tmp_path, vectors, script=script) + assert report.returncode == 2 and "no vector changes signature" in report.stdout, report.stdout + + +def control_unchecked_signature_fails(script: Path, tmp_path: Path) -> None: + with Stub(Conformant(check_signature=False)) as url: + report = run_runner(url, tmp_path, script=script) + assert {i for i, v in report.verdicts.items() if v == "FAIL"} == SIGNATURE_IDS, report.stdout + + +def control_wrong_code_fails(script: Path, tmp_path: Path) -> None: + wrong = {code: -32603 for code in (-32004, -32005, -32006)} + with Stub(Conformant(codes=wrong)) as url: + report = run_runner(url, tmp_path, script=script) + assert {i for i, v in report.verdicts.items() if v == "FAIL"} == NEGATIVE_IDS, report.stdout + + +def control_empty_suite_exits_2(script: Path, tmp_path: Path) -> None: + with Stub(allow_all) as url: + report = run_runner(url, tmp_path, [], script=script) + assert report.returncode == 2 and report.counts == {"pass": 0, "fail": 0, "not_run": 0}, report.stdout + + +def control_unanswered_scores_nothing(script: Path, tmp_path: Path) -> None: + with Stub(drop) as url: + report = run_runner(url, tmp_path, script=script) + assert report.returncode == 2 and set(report.verdicts.values()) == {"NOT-RUN"}, report.stdout + + +# Requirement 2: each mutant weakens one guard in the runner, and at least one control +# above must fail against it. A mutant whose source line is not found fails the test, +# so a rewrite of the runner cannot quietly turn a mutant into a copy of the original. +MUTANTS = [ + ("the error code is not compared", + 'ok = last.get("error") == expect["error"]', 'ok = "error" in last', + control_wrong_code_fails), + ("an allow expectation is not checked", + 'ok = all(o.get("decision") == expect["decision"] for o in observations)', "ok = True", + control_deny_all_fails), + ("the positive and negative mirror rule is dropped", + 'if kinds != {"positive", "negative"}:', "if not kinds:", + control_positives_only_refused), + ("the rule that a suite varies each property is dropped", + "if steps and not any(", "if False and not any(", + control_constant_signature_refused), + ("an empty run exits 0", + 'if counts["PASS"] + counts["FAIL"] == 0:\n return 2', 'if False:\n return 2', + control_empty_suite_exits_2), + ("a transport failure is scored as a pass", + 'counts["NOT-RUN"] += 1\n print("NOT-RUN %s %s" % (vector["id"], type(exc).__name__))', + 'counts["PASS"] += 1\n print("PASS " + vector["id"])', + control_unanswered_scores_nothing), + ("every run exits 0", + 'return 1 if counts["FAIL"] or counts["NOT-RUN"] else 0', "return 0", + control_allow_all_fails), +] + + +# --- requirement 3 ---------------------------------------------------------------------- + + +def test_a_conformant_guardian_passes(tmp_path): + """The positive control for the gate itself: a strict runner that fails everything is no gate.""" + control_conformant_passes(RUNNER, tmp_path) + + +def test_an_allow_everything_guardian_fails_every_negative(tmp_path): + control_allow_all_fails(RUNNER, tmp_path) + + +def test_a_deny_everything_guardian_fails_every_positive(tmp_path): + control_deny_all_fails(RUNNER, tmp_path) + + +def test_a_suite_of_positive_controls_alone_is_refused(tmp_path): + control_positives_only_refused(RUNNER, tmp_path) + + +# --- requirement 5 ---------------------------------------------------------------------- + + +def test_a_guardian_that_skips_signatures_fails_the_signature_vectors(tmp_path): + control_unchecked_signature_fails(RUNNER, tmp_path) + + +def test_a_suite_whose_signatures_never_change_is_refused(tmp_path): + control_constant_signature_refused(RUNNER, tmp_path) + + +# --- requirement 7 ---------------------------------------------------------------------- + + +def test_the_right_message_with_the_wrong_code_fails(tmp_path): + control_wrong_code_fails(RUNNER, tmp_path) + + +def test_the_right_code_with_other_wording_passes(tmp_path): + with Stub(Conformant(wording="other")) as url: + report = run_runner(url, tmp_path) + assert report.returncode == 0, report.stdout + + +# --- requirement 9 ---------------------------------------------------------------------- + + +def test_an_empty_suite_does_not_exit_0(tmp_path): + control_empty_suite_exits_2(RUNNER, tmp_path) + + +def test_a_guardian_that_never_answers_scores_nothing(tmp_path): + control_unanswered_scores_nothing(RUNNER, tmp_path) + + +def test_a_closed_port_scores_nothing(tmp_path): + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + report = run_runner("http://127.0.0.1:%d/acs" % port, tmp_path) + assert report.returncode == 2, report.stdout + + +# --- requirements 4 and 11 -------------------------------------------------------------- + + +def test_a_child_that_exits_0_without_a_report_is_not_a_pass(tmp_path): + silent = tmp_path / "silent.py" + silent.write_text("import sys\nsys.exit(0)\n") + with pytest.raises(AssertionError, match="no closing RESULT line"): + run_runner("http://127.0.0.1:9/acs", tmp_path, script=silent) + + +def test_an_exit_code_that_disagrees_with_the_report_fails(tmp_path): + proc = subprocess.CompletedProcess([], 0, stdout="FAIL x expected\nRESULT pass=0 fail=1 not_run=0\n", stderr="") + with pytest.raises(AssertionError, match="disagrees with the report"): + read_report(proc) + + +# --- requirement 2 ---------------------------------------------------------------------- + + +@pytest.mark.parametrize("name, old, new, control", MUTANTS, ids=[m[0] for m in MUTANTS]) +def test_every_runner_mutant_is_caught(tmp_path, name, old, new, control): + source = RUNNER.read_text() + assert source.count(old) == 1, "mutant %r no longer matches the runner" % name + mutant = tmp_path / "mutant_runner.py" + mutant.write_text(source.replace(old, new)) + with pytest.raises(AssertionError): + control(mutant, tmp_path) + + +# --- the reference Guardian --------------------------------------------------------------- + +# Each is a MUST the reference Guardian fails, not an optional case and not a defect in the +# vector. The strict marker turns a fix into a failing run, so the list cannot go stale. +REFERENCE_GAPS = { + "signature-invalid": "a section 10 MUST the reference Guardian fails today: it does not verify signatures (#184)", + "signature-missing": "a section 10 MUST the reference Guardian fails today: it accepts an unsigned request (#184)", + "request-id-repeated": "a section 10.3 MUST the reference Guardian fails today: it does not track request_id (#184)", + "timestamp-outside-window": "a section 10.3 MUST the reference Guardian fails today: it does not check the skew window (#184)", +} + + +@pytest.fixture(scope="module") +def reference_report(tmp_path_factory): + bun = shutil.which("bun") + if bun is None or not (AGT / "node_modules").is_dir(): + reason = ("the reference Guardian needs Bun and `bun install` in %s; the Reference " + "implementation workflow runs this group with both" % AGT.relative_to(ROOT)) + if os.environ.get("ACS_REQUIRE_REFERENCE_GUARDIAN"): + pytest.fail(reason) + pytest.skip(reason) + work = tmp_path_factory.mktemp("reference-guardian") + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + env = dict(os.environ, ACS_GUARDIAN_PORT=str(port), ACS_GUARDIAN_HOST="127.0.0.1", + ACS_ENVELOPE_LOG=str(work / "envelopes.jsonl"), + ACS_SESSION_CONTEXT_LOG=str(work / "session-context.jsonl")) + log = work / "guardian.log" + with open(log, "w") as out: + guardian = subprocess.Popen([bun, "run", "guardian"], cwd=AGT, env=env, stdout=out, + stderr=subprocess.STDOUT, start_new_session=True) + try: + deadline = time.monotonic() + 60 + while "Guardian listening" not in log.read_text(): + assert guardian.poll() is None and time.monotonic() < deadline, log.read_text() + time.sleep(0.2) + yield run_runner("http://127.0.0.1:%d/acs" % port, work) + finally: + os.killpg(guardian.pid, signal.SIGTERM) + guardian.wait(timeout=30) + + +@pytest.mark.parametrize("vector_id", [ + pytest.param(i, marks=pytest.mark.xfail(strict=True, reason=REFERENCE_GAPS[i])) if i in REFERENCE_GAPS else i + for i in VECTOR_IDS +]) +def test_the_reference_guardian(reference_report, vector_id): + assert reference_report.verdicts.get(vector_id) == "PASS", reference_report.stdout