"""Independent safety/regression coverage for delegation evidence acceptance.""" import copy import hashlib import json import os import shutil import subprocess import sys import tempfile import unittest from pathlib import Path from harness.lib import delegation_evidence as evidence ROOT = Path(__file__).resolve().parents[3] HELPER = ROOT / "harness" / "lib" / "delegation_evidence.py" CONTEXT_SCHEMA = "gitea-agent.delegation-evidence-context/v1" RESULT_SCHEMA = "gitea-agent.delegation-evidence-result/v1" REVISION = "a" * 40 class DelegationEvidenceSafetyTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch") self.base = Path(self.tmp.name) self.bundle = self.base / "bundle" (self.bundle / "tests").mkdir(parents=True) (self.bundle / "src").mkdir() (self.bundle / "evidence").mkdir() (self.bundle / "tests/example.py").write_bytes( b"def test_fn():\n assert True\n" ) (self.bundle / "src/value.py").write_bytes(b"VALUE = 1\n") (self.bundle / "evidence/run.txt").write_bytes(b"recorded output\n") self.files = { path: hashlib.sha256((self.bundle / path).read_bytes()).hexdigest() for path in ("src/value.py", "tests/example.py") } self.bundle_sha = self.digest(self.files) self.evidence_sha = hashlib.sha256( (self.bundle / "evidence/run.txt").read_bytes() ).hexdigest() self.context = self.make_context() self.result = self.make_result() def tearDown(self): self.tmp.cleanup() @staticmethod def digest(value): return hashlib.sha256( json.dumps(value, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() def make_context(self, *, actions=None, obligations=None, executions=None): return { "schema": CONTEXT_SCHEMA, "source": {"kind": "captured_worktree", "base_revision": REVISION}, "stage": "implementation", "bundle_sha256": self.bundle_sha, "files": copy.deepcopy(self.files), "allowed_actions": ["static", "execute"] if actions is None else actions, "obligations": [ {"id": "static-src", "kind": "static"}, {"id": "run-test", "kind": "execution"}, ] if obligations is None else obligations, "executions": [{ "id": "exec-1", "origin": "direct", "test_id": "", "bundle_sha256": self.bundle_sha, "test_file_sha256": "0" * 64, "argv": ["python3", "-B", "-c", "pass"], "exit_code": 0, "failure_kind": "none", "evidence_path": "evidence/run.txt", "evidence_sha256": self.evidence_sha, }] if executions is None else executions, } def make_result(self, *, source=None, stage=None, bundle_sha=None, claims=None): return { "schema": RESULT_SCHEMA, "source": copy.deepcopy(self.context["source"] if source is None else source), "stage": self.context["stage"] if stage is None else stage, "bundle_sha256": self.bundle_sha if bundle_sha is None else bundle_sha, "claims": [ {"id": "static-src", "status": "supported", "anchors": [{ "kind": "lines", "path": "src/value.py", "start_line": 1, "end_line": 1, "text_sha256": self.line_hash("src/value.py", 1, 1), }], "execution_ids": []}, {"id": "run-test", "status": "supported", "anchors": [], "execution_ids": ["exec-1"]}, ] if claims is None else claims, } def line_hash(self, path, start, end): lines = (self.bundle / path).read_bytes().splitlines(keepends=True) return hashlib.sha256(b"".join(lines[start - 1:end])).hexdigest() def api(self, context=None, result=None, root=None): return evidence.verify_evidence( str(self.bundle if root is None else root), self.context if context is None else context, self.result if result is None else result, ) def cli(self, context_text=None, result=None, *, root=None, context_path=None, result_path=None, extra=None, timeout=3): root = self.bundle if root is None else Path(root) context_path = self.base / "context.json" if context_path is None else Path(context_path) result_path = self.base / "result.json" if result_path is None else Path(result_path) if context_text is None: context_path.write_text(json.dumps(self.context), encoding="utf-8") elif isinstance(context_text, bytes): context_path.write_bytes(context_text) else: context_path.write_text(context_text, encoding="utf-8") if result is not None: result_path.write_text(json.dumps(result), encoding="utf-8") else: result_path.write_text(json.dumps(self.result), encoding="utf-8") command = [sys.executable, "-B", str(HELPER), "--bundle-root", str(root), "--context", str(context_path), "--result", str(result_path)] if extra: command.extend(extra) return subprocess.run(command, capture_output=True, text=True, timeout=timeout) def assert_rejected(self, value, *, field=None): bad = copy.deepcopy(self.context) if field is None: bad = value else: target = bad for part in field[:-1]: target = target[part] target[field[-1]] = value with self.assertRaises(evidence.EvidenceError): self.api(context=bad) def test_real_paths_reject_root_file_ancestor_and_canonical_aliases(self): alias = self.base / "bundle-alias" alias.symlink_to(self.bundle, target_is_directory=True) with self.assertRaises(evidence.EvidenceError): self.api(root=alias) with self.assertRaises(evidence.EvidenceError): self.api(root=self.base / "missing" / ".." / "bundle") outside = self.base / "outside.txt" outside.write_bytes(b"outside") (self.bundle / "src" / "alias.py").symlink_to(outside) context = copy.deepcopy(self.context) context["files"]["src/alias.py"] = hashlib.sha256(outside.read_bytes()).hexdigest() context["bundle_sha256"] = self.digest(context["files"]) with self.assertRaises(evidence.EvidenceError): self.api(context=context) def test_json_input_symlink_fifo_and_source_fifo_are_rejected_without_hang(self): real_context = self.base / "real-context.json" real_context.write_text(json.dumps(self.context), encoding="utf-8") alias = self.base / "context.json" alias.symlink_to(real_context) result_path = self.base / "result.json" result_path.write_text(json.dumps(self.result), encoding="utf-8") with self.assertRaises(evidence.EvidenceError): evidence._json_file(str(alias)) fifo = self.base / "context-fifo" os.mkfifo(fifo) proc = subprocess.Popen( [sys.executable, "-B", str(HELPER), "--bundle-root", str(self.bundle), "--context", str(fifo), "--result", str(result_path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) stdout, stderr = proc.communicate(timeout=3) self.assertEqual(proc.returncode, 2, (stdout, stderr)) self.assertEqual(json.loads(stdout), {"status": "block", "reason_codes": ["rejected_input"], "counts": {}}) source_fifo = self.bundle / "src" / "fifo.py" os.mkfifo(source_fifo) context = copy.deepcopy(self.context) context["files"]["src/fifo.py"] = "0" * 64 context["bundle_sha256"] = self.digest(context["files"]) with self.assertRaises(evidence.EvidenceError): self.api(context=context) def test_oversized_source_and_json_are_bounded(self): huge = self.bundle / "src" / "huge.py" huge.write_bytes(b"x" * (1024 * 1024 + 1)) context = copy.deepcopy(self.context) context["files"]["src/huge.py"] = hashlib.sha256(huge.read_bytes()).hexdigest() context["bundle_sha256"] = self.digest(context["files"]) with self.assertRaises(evidence.EvidenceError): self.api(context=context) proc = self.cli(context_text=b"{" + b"\"x\":\"" + b"a" * (1024 * 1024) + b"\"}") self.assertEqual(proc.returncode, 2) self.assertEqual(json.loads(proc.stdout)["reason_codes"], ["rejected_input"]) def test_duplicate_unknown_and_wrong_typed_json_fields_are_rejected(self): cases = [ b'{"schema":"x","schema":"y"}', json.dumps({**self.context, "unknown": 1}).encode(), ] for payload in cases: with self.subTest(case=payload[:20]): proc = self.cli(context_text=payload) self.assertEqual(proc.returncode, 2) for path, value in [ (("stage",), 1), (("source", "revision"), True), (("files",), []), (("allowed_actions",), "static"), (("obligations",), {}), (("executions",), "none"), ]: context = copy.deepcopy(self.context) target = context for part in path[:-1]: target = target[part] if path == ("source", "revision"): context["source"] = {"kind": "git_tree", "revision": value} else: target[path[-1]] = value with self.subTest(path=path): with self.assertRaises(evidence.EvidenceError): self.api(context=context) def test_malformed_digests_revisions_and_boolean_integer_fields_are_rejected(self): cases = [ (("bundle_sha256",), "not-a-digest"), (("source", "base_revision"), "b" * 39), (("files", "src/value.py"), "A" * 64), ] for path, value in cases: context = copy.deepcopy(self.context) if path[0] == "files": context[path[0]][path[1]] = value elif len(path) == 2: context[path[0]][path[1]] = value else: context[path[0]] = value with self.subTest(path=path): with self.assertRaises(evidence.EvidenceError): self.api(context=context) context = copy.deepcopy(self.context) context["executions"][0]["exit_code"] = True with self.assertRaises(evidence.EvidenceError): self.api(context=context) def test_duplicate_missing_and_extra_claim_or_obligation_ids_block_input(self): cases = [] duplicate_claim = copy.deepcopy(self.result) duplicate_claim["claims"][1]["id"] = duplicate_claim["claims"][0]["id"] cases.append(duplicate_claim) missing_claim = copy.deepcopy(self.result) missing_claim["claims"] = missing_claim["claims"][:1] cases.append(missing_claim) extra_claim = copy.deepcopy(self.result) extra_claim["claims"].append({"id": "extra", "status": "supported", "anchors": [], "execution_ids": []}) cases.append(extra_claim) for result in cases: with self.subTest(ids=[claim["id"] for claim in result["claims"]]): with self.assertRaises(evidence.EvidenceError): self.api(result=result) context = copy.deepcopy(self.context) context["obligations"].append({"id": "static-src", "kind": "static"}) with self.assertRaises(evidence.EvidenceError): self.api(context=context) def test_stale_source_stage_and_bundle_claims_block(self): stale_context = copy.deepcopy(self.context) stale_context["stage"] = "review" self.assertIn("result_stage_mismatch", self.api(context=stale_context)["reason_codes"]) for field, value in [("stage", "review"), ("bundle_sha256", "f" * 64), ("source", {"kind": "captured_worktree", "base_revision": "b" * 40})]: stale_result = copy.deepcopy(self.result) stale_result[field] = value with self.subTest(field=field): out = self.api(result=stale_result) self.assertEqual(out["status"], "block") self.assertIn("result_" + field + "_mismatch", out["reason_codes"]) def test_line_anchor_rejects_zero_negative_reversed_out_of_range_and_wrong_hash(self): for start, end in [(0, 1), (-1, 1), (2, 1), (1, 99)]: result = copy.deepcopy(self.result) result["claims"][0]["anchors"][0].update(start_line=start, end_line=end) with self.subTest(start=start, end=end): if end == 99: out = self.api(result=result) self.assertIn("anchor_range_invalid", out["reason_codes"]) else: with self.assertRaises(evidence.EvidenceError): self.api(result=result) result = copy.deepcopy(self.result) result["claims"][0]["anchors"][0]["text_sha256"] = "0" * 64 out = self.api(result=result) self.assertEqual(out["status"], "block") self.assertIn("anchor_hash_mismatch", out["reason_codes"]) def test_ast_anchor_rejects_nested_local_dynamic_and_syntax_error_sources(self): variants = { "nested": b"def outer():\n def test_fn():\n pass\n", "dynamic": b"globals()['test_fn'] = lambda: None\n", "syntax": b"def test_fn(:\n", } for name, source in variants.items(): path = self.bundle / "tests/example.py" path.write_bytes(source) context = copy.deepcopy(self.context) context["files"]["tests/example.py"] = hashlib.sha256(source).hexdigest() context["bundle_sha256"] = self.digest(context["files"]) result = copy.deepcopy(self.result) result["bundle_sha256"] = context["bundle_sha256"] result["claims"][0]["anchors"] = [{ "kind": "test", "path": "tests/example.py", "symbol": "test_fn", }] with self.subTest(source=name): out = self.api(context=context, result=result) self.assertEqual(out["status"], "block") self.assertIn("test_anchor_missing", out["reason_codes"]) def test_unknown_execution_reference_and_authorization_origin_are_blocked(self): result = copy.deepcopy(self.result) result["claims"][1]["execution_ids"] = ["missing"] out = self.api(result=result) self.assertEqual(out["status"], "block") self.assertIn("unknown_execution_id", out["reason_codes"]) context = copy.deepcopy(self.context) context["allowed_actions"] = ["static"] out = self.api(context=context) self.assertEqual(out["status"], "block") self.assertIn("unauthorized_execute", out["reason_codes"]) context["executions"][0]["origin"] = "inherited" self.assertEqual(self.api(context=context)["status"], "pass") def test_red_green_rejects_import_environment_timeout_and_mismatch_records(self): test_sha = self.files["tests/example.py"] obligations = [{"id": "tdd", "kind": "red_green"}] def execution(eid, exit_code, failure_kind, bundle, test_file=test_sha): return { "id": eid, "origin": "inherited", "test_id": "tests/example.py::test_fn", "bundle_sha256": bundle, "test_file_sha256": test_file, "argv": ["python3", "-B", "-m", "unittest"], "exit_code": exit_code, "failure_kind": failure_kind, "evidence_path": "evidence/run.txt", "evidence_sha256": self.evidence_sha, } for failure_kind in ("import", "environment", "timeout"): context = self.make_context(actions=["static"], obligations=obligations, executions=[execution("red", 1, failure_kind, "b" * 64), execution("green", 0, "none", self.bundle_sha)]) result = self.make_result(claims=[{ "id": "tdd", "status": "supported", "anchors": [{ "kind": "test", "path": "tests/example.py", "symbol": "test_fn", }], "execution_ids": ["red", "green"], }]) with self.subTest(failure_kind=failure_kind): out = self.api(context=context, result=result) self.assertEqual(out["status"], "block") self.assertIn("red_green_missing_red", out["reason_codes"]) context = self.make_context(actions=["static"], obligations=obligations, executions=[execution("red", 1, "assertion", "b" * 64, "c" * 64), execution("green", 0, "none", self.bundle_sha)]) result = self.make_result(claims=[{ "id": "tdd", "status": "supported", "anchors": [{ "kind": "test", "path": "tests/example.py", "symbol": "test_fn", }], "execution_ids": ["red", "green"], }]) out = self.api(context=context, result=result) self.assertIn("red_green_test_scope_mismatch", out["reason_codes"]) def test_missing_green_and_evidence_hash_mismatch_block(self): obligations = [{"id": "tdd", "kind": "red_green"}] test_sha = self.files["tests/example.py"] executions = [{ "id": "red", "origin": "inherited", "test_id": "tests/example.py::test_fn", "bundle_sha256": "b" * 64, "test_file_sha256": test_sha, "argv": ["python3", "-B", "-m", "unittest"], "exit_code": 1, "failure_kind": "assertion", "evidence_path": "evidence/run.txt", "evidence_sha256": "0" * 64, }] context = self.make_context(actions=["static"], obligations=obligations, executions=executions) result = self.make_result(claims=[{ "id": "tdd", "status": "supported", "anchors": [{ "kind": "test", "path": "tests/example.py", "symbol": "test_fn", }], "execution_ids": ["red"], }]) out = self.api(context=context, result=result) self.assertIn("red_green_missing_green", out["reason_codes"]) self.assertIn("evidence_hash_mismatch", out["reason_codes"]) if __name__ == "__main__": unittest.main()