From 35c347507964ac00218e6b02c008035460f15b75 Mon Sep 17 00:00:00 2001 From: Sankalp Gilda <23521054+astrogilda@users.noreply.github.com> Date: Sat, 17 Oct 2026 10:23:44 -0400 Subject: [PATCH 1/4] test: add negative controls for the conformance runner Issue 53 lists what a conformance suite has to get right. Requirements 2, 3, 4, 5, 7, 9 and 11 are about the gate rather than any one Guardian, so each gets a control here: a stub Guardian or a corpus the runner must refuse, and seven runner mutants that at least one control must catch. The runner is standard library only and runs as a child process. It posts signed envelopes over HTTP and scores by section 17.1 code, never by message text. The same vectors run against the reference Guardian, where the four known gaps (signature, missing signature, repeated request_id and the skew window) are strict expected failures that cite issue 184. Signed-off-by: Sankalp Gilda <23521054+astrogilda@users.noreply.github.com> --- tests/negative_controls_runner.py | 253 +++++++++++++++++ tests/test_negative_controls.py | 436 ++++++++++++++++++++++++++++++ 2 files changed, 689 insertions(+) create mode 100644 tests/negative_controls_runner.py create mode 100644 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..cbd243f --- /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 the pre-shared secret as input keying material, an empty salt, and the session_id as +info. Section 10 leaves the HKDF salt and info to the deployment, so this runner fixes them +and a Guardian under test is configured the same way. 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..e8d4f54 --- /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 finding against the reference Guardian, not a defect in the vector. The strict +# marker turns a fix into a failing run, so the list cannot go stale silently. +REFERENCE_GAPS = { + "signature-invalid": "the reference Guardian does not verify signatures (#184; #229 proposes a fix)", + "signature-missing": "the reference Guardian accepts an unsigned request (#184; #229 proposes a fix)", + "request-id-repeated": "the reference Guardian does not track request_id (#184; #229 proposes a fix)", + "timestamp-outside-window": "the reference Guardian does not check the skew window (#184; #229 proposes a fix)", +} + + +@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 From 4fef8137a5b0073a1b0c863f83aa655ac19c93cd Mon Sep 17 00:00:00 2001 From: Sankalp Gilda <23521054+astrogilda@users.noreply.github.com> Date: Sat, 17 Oct 2026 10:23:44 -0400 Subject: [PATCH 2/4] ci: run the negative controls against the reference Guardian The test job in deploy-pages.yml has no Bun, so the reference Guardian group in tests/test_negative_controls.py skips there. The agt job already installs Bun and the tree's dependencies, so it now runs that file too, with ACS_REQUIRE_REFERENCE_GUARDIAN set so the group fails rather than skips when the Guardian cannot start. Signed-off-by: Sankalp Gilda <23521054+astrogilda@users.noreply.github.com> --- .../workflows/reference-implementation.yml | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) 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 From 07b0ae42bd2fdbf58a8f1c7431c7c5ff2e8b7d89 Mon Sep 17 00:00:00 2001 From: Sankalp Gilda <23521054+astrogilda@users.noreply.github.com> Date: Sat, 17 Oct 2026 10:26:06 -0400 Subject: [PATCH 3/4] test: name the HKDF parameters the runner shares with section 10 The runner's key derivation matches the parameters pull request 272 adds to section 10, so the docstring now says so instead of describing them as this runner's own choice. Signed-off-by: Sankalp Gilda <23521054+astrogilda@users.noreply.github.com> --- tests/negative_controls_runner.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/negative_controls_runner.py b/tests/negative_controls_runner.py index cbd243f..8ae2528 100644 --- a/tests/negative_controls_runner.py +++ b/tests/negative_controls_runner.py @@ -17,9 +17,9 @@ 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 the pre-shared secret as input keying material, an empty salt, and the session_id as -info. Section 10 leaves the HKDF salt and info to the deployment, so this runner fixes them -and a Guardian under test is configured the same way. The canonical form here is +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 9ecc0bf6fb8143dd8f942c8de5db307d38f354f6 Mon Sep 17 00:00:00 2001 From: Sankalp Gilda <23521054+astrogilda@users.noreply.github.com> Date: Sat, 17 Oct 2026 10:37:50 -0400 Subject: [PATCH 4/4] test: label the reference Guardian gaps as MUSTs it fails Each strict xfail reason now names the section whose MUST the reference Guardian fails, so the marker reads as a known failure rather than as an optional case. Signed-off-by: Sankalp Gilda <23521054+astrogilda@users.noreply.github.com> --- tests/test_negative_controls.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/test_negative_controls.py b/tests/test_negative_controls.py index e8d4f54..f3af0d6 100644 --- a/tests/test_negative_controls.py +++ b/tests/test_negative_controls.py @@ -387,13 +387,13 @@ def test_every_runner_mutant_is_caught(tmp_path, name, old, new, control): # --- the reference Guardian --------------------------------------------------------------- -# Each is a finding against the reference Guardian, not a defect in the vector. The strict -# marker turns a fix into a failing run, so the list cannot go stale silently. +# 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": "the reference Guardian does not verify signatures (#184; #229 proposes a fix)", - "signature-missing": "the reference Guardian accepts an unsigned request (#184; #229 proposes a fix)", - "request-id-repeated": "the reference Guardian does not track request_id (#184; #229 proposes a fix)", - "timestamp-outside-window": "the reference Guardian does not check the skew window (#184; #229 proposes a fix)", + "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)", }