From 07da5e540ffce64628f417e72306ce91cdf6ae18 Mon Sep 17 00:00:00 2001 From: AOS Architecture Agent Date: Sat, 29 Aug 2026 15:06:10 -0400 Subject: [PATCH] Add frozen D-003 evidence runner --- decisions/d003_evidence_run.py | 447 ++++++++++++++++++++++++++++ decisions/test_d003_evidence_run.py | 374 +++++++++++++++++++++++ 2 files changed, 821 insertions(+) create mode 100644 decisions/d003_evidence_run.py create mode 100644 decisions/test_d003_evidence_run.py diff --git a/decisions/d003_evidence_run.py b/decisions/d003_evidence_run.py new file mode 100644 index 0000000000..8167c5f17e --- /dev/null +++ b/decisions/d003_evidence_run.py @@ -0,0 +1,447 @@ +#!/usr/bin/env python3 +"""Bounded retained-evidence orchestration for frozen D-003 v2.""" +from __future__ import annotations + +import argparse +import base64 +import ctypes +import errno +import hashlib +import json +import os +import re +import shutil +import subprocess +import sys +import tempfile +import unicodedata +from pathlib import Path +from typing import Any, Mapping + +ROOT = Path(__file__).resolve().parent.parent +FREEZE_COMMIT = "ef8b20fde411d029e6feac64a11a9dda96122a82" +FREEZE_TREE = "f1d4148b0184e75fe9cb6396fed4d40333e9e54b" +IMPLEMENTATION_COMMIT = "0af0171d7f1c1012d69f98c83a9dcaec4dbc1c6d" +IMPLEMENTATION_TREE = "f556c4a187566fca72b842c735a243991dca8369" +PREREGISTRATION_COMMIT = "c69e49abaf5d5b018931a54b4993ed9f500bafb7" +PREREGISTRATION_TREE = "ed394b98080618d7e9cbd92e6a05c5bd1450dc1a" +FREEZE_PATH = "decisions/d003-capability-manifest-v2.freeze.json" +MANIFEST_PATH = "decisions/d003-capability-manifest-v2.json" +PACKET_PATH = "decisions/D-003-PACKET-v2.md" +FREEZE_SHA256 = "47193803c59e6b1017b10b7101865c43b4955f771b9e77b4a7e4994fd16440e0" +FREEZE_LENGTH = 2426 +FREEZE_BLOB = "58034991fe1f61e4d2f0d43f52c2b25e21a19292" +MANIFEST_SHA256 = "552ee918457b04f3bfe2bf9bf272709fc0052ba1e704ed8abaf608f75a40193c" +MANIFEST_LENGTH = 629981 +OUTPUT_SHA256 = "e0446c952d65872ddb81333dc48e1a0407e96a96c100cd816f45c01413b82b21" +OUTPUT_LENGTH = 31348 +TIMEOUT = 90 +AT_FDCWD = -100 +RENAME_NOREPLACE = 1 +ARTIFACT_NAMES = ("O1.json", "O2.json", "I1.json", "I2.json", "V3.json", "E.json", "summary.json") + +BOUND = { + MANIFEST_PATH: (PREREGISTRATION_COMMIT, MANIFEST_LENGTH, MANIFEST_SHA256, "bd77b945c3f8e3925d041d0a7e6bd9909c43c5f0"), + PACKET_PATH: (PREREGISTRATION_COMMIT, 27037, "2a30e5b5cfba78af6def1927e93f57cd695fd821203285ff92e81714764a2dc9", "2976696b707b44be23b80eac0f9334bef69d3c10"), + "decisions/d003_r1_v2.py": (IMPLEMENTATION_COMMIT, 28214, "ce726d40be09b08587fa166f04f58103e6b59858a8485892f3eeecfafae7efcc", "320a0c7b9b61307e8e188df579588d23aff4e99a"), + "decisions/test_d003_r1_v2.py": (IMPLEMENTATION_COMMIT, 14715, "dda5b9fc321919f66a5eafe94da9d5b8fa83499e45a2f3ff0499a8033ce36e5a", "18a667609e9818a89a9ee84c409bfb2eaf8de99b"), + "decisions/d003_r2_v2.py": (IMPLEMENTATION_COMMIT, 31731, "c26ec0e3e2025925d44f72ec682489b30f3d08b6bc73594472702bf3b3b5997e", "9262019421987538cd7d5fe1d80202449e6b50fd"), + "decisions/test_d003_r2_v2.py": (IMPLEMENTATION_COMMIT, 14236, "f5cdb5245e73786ad42648f8a0543bc04e3571663506c2334127af29397f0581", "35757719f484ee11581d6fe4e7ca2bcee3f5416c"), + "decisions/d003_v3_v2.py": (IMPLEMENTATION_COMMIT, 63021, "d209a5d8007a13d702216442bd42d25327e508cd074bc689c24c532da755a4a5", "b6c8fef7feafbd166a584e4b4b2cd7de00421ffc"), + "decisions/test_d003_v3_v2.py": (IMPLEMENTATION_COMMIT, 27071, "82e6a7c0ba92e30ea43fda2ee61c6116efc56fb2dbd57dbf17bfa865d9bb34d6", "5855c11faa87a4aa1bff4a95385e1024dcb6c95c"), + "decisions/d003_manifest_v2.py": (PREREGISTRATION_COMMIT, 76921, "d2660a6cf49bf3127d259fa75df15d2056691c532ba20a3d9be5876bc08f3332", "64e2d7833d7e6c06bce436b78e38f6eabf1ede0c"), + "decisions/test_d003_manifest_v2.py": (PREREGISTRATION_COMMIT, 35473, "07c2eb9945ccdf113b8374c03b450a3f04eaa7f1c45941d03259309593f53640", "f051bf96282cb0cc6bfb77144d03b76ec3b46ea6"), +} + +FREEZE_EXPECTED = { + "assertion_count": 270, "decision_id": "D-003", "embedded_decision_status": "OPEN", + "external_freeze_status": "FROZEN", + "freeze_purpose": "Freeze the exact reviewed D-003 v2 preregistration, two independent executors, verifier, and tests before retained execution; this record adopts no algebra or representation, selects no winner, and D-003 remains OPEN.", + "freeze_schema_version": "2.0.0", "implementation_commit": IMPLEMENTATION_COMMIT, + "implementation_tree": IMPLEMENTATION_TREE, "manifest_byte_length": MANIFEST_LENGTH, + "manifest_git_blob": BOUND[MANIFEST_PATH][3], "manifest_id": "d003-capability-manifest-v2", + "manifest_path": MANIFEST_PATH, "manifest_sha256": MANIFEST_SHA256, + "output_expected_byte_length": OUTPUT_LENGTH, "output_expected_sha256": OUTPUT_SHA256, + "packet_git_blob": BOUND[PACKET_PATH][3], "packet_sha256": BOUND[PACKET_PATH][2], + "preregistration_commit": PREREGISTRATION_COMMIT, "preregistration_tree": PREREGISTRATION_TREE, + "r1_git_blob": BOUND["decisions/d003_r1_v2.py"][3], "r1_sha256": BOUND["decisions/d003_r1_v2.py"][2], + "r1_test_git_blob": BOUND["decisions/test_d003_r1_v2.py"][3], "r1_test_sha256": BOUND["decisions/test_d003_r1_v2.py"][2], + "r2_git_blob": BOUND["decisions/d003_r2_v2.py"][3], "r2_sha256": BOUND["decisions/d003_r2_v2.py"][2], + "r2_test_git_blob": BOUND["decisions/test_d003_r2_v2.py"][3], "r2_test_sha256": BOUND["decisions/test_d003_r2_v2.py"][2], + "required_output_count": 2, "selection_claim": None, "trace_count": 10, + "v3_git_blob": BOUND["decisions/d003_v3_v2.py"][3], "v3_sha256": BOUND["decisions/d003_v3_v2.py"][2], + "v3_test_git_blob": BOUND["decisions/test_d003_v3_v2.py"][3], "v3_test_sha256": BOUND["decisions/test_d003_v3_v2.py"][2], + "validator_git_blob": BOUND["decisions/d003_manifest_v2.py"][3], "validator_sha256": BOUND["decisions/d003_manifest_v2.py"][2], + "validator_test_git_blob": BOUND["decisions/test_d003_manifest_v2.py"][3], "validator_test_sha256": BOUND["decisions/test_d003_manifest_v2.py"][2], +} + +class ValidationError(ValueError): + pass + +def sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + +def git_oid(kind: str, data: bytes) -> str: + return hashlib.sha1(kind.encode() + b" " + str(len(data)).encode() + b"\0" + data).hexdigest() + +def canonical(value: Any) -> bytes: + def walk(v: Any) -> None: + if isinstance(v, bool) or v is None or isinstance(v, int) or isinstance(v, str): + if isinstance(v, str) and unicodedata.normalize("NFC", v) != v: + raise ValidationError("non-NFC text") + return + if isinstance(v, list): + for x in v: walk(x) + return + if isinstance(v, dict): + for k, x in v.items(): + if not isinstance(k, str) or unicodedata.normalize("NFC", k) != k: raise ValidationError("non-NFC key") + walk(x) + return + raise ValidationError("non-canonical JSON type") + walk(value) + return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode() + +def parse_canonical(raw: bytes, label: str) -> Any: + def pairs(items: list[tuple[str, Any]]) -> dict[str, Any]: + out = {} + for k, v in items: + if k in out or unicodedata.normalize("NFC", k) != k: raise ValidationError(label + ": duplicate/non-NFC key") + out[k] = v + return out + def bad(_: str) -> Any: raise ValidationError(label + ": float/constant forbidden") + try: + value = json.loads(raw.decode("utf-8"), object_pairs_hook=pairs, parse_float=bad, parse_constant=bad) + except (UnicodeError, json.JSONDecodeError) as exc: + raise ValidationError(label + ": invalid JSON") from exc + if canonical(value) != raw: raise ValidationError(label + ": noncanonical bytes") + return value + +def _git(*args: str, binary: bool = False) -> bytes | str: + try: + cp = subprocess.run(["git", "-C", str(ROOT), *args], stdout=subprocess.PIPE, stderr=subprocess.PIPE, + timeout=TIMEOUT, check=False) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ValidationError("Git unavailable or timed out") from exc + if cp.returncode or cp.stderr: raise ValidationError("Git command failed: " + " ".join(args)) + return cp.stdout if binary else cp.stdout.decode().strip() + +def _require_freeze_ancestor() -> None: + try: + cp = subprocess.run(["git", "-C", str(ROOT), "merge-base", "--is-ancestor", FREEZE_COMMIT, "HEAD"], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=TIMEOUT, check=False) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ValidationError("Git ancestry check unavailable or timed out") from exc + if cp.stdout or cp.stderr or cp.returncode not in (0, 1): + raise ValidationError("Git ancestry check failed") + if cp.returncode == 1: + raise ValidationError("freeze commit is not an ancestor of HEAD") + +def _show(commit: str, path: str) -> bytes: + return _git("show", commit + ":" + path, binary=True) # type: ignore[return-value] + +def _object_type(oid: str) -> str: + return _git("cat-file", "-t", oid) # type: ignore[return-value] + +def _git_tree(files: Mapping[str, bytes]) -> str: + root: dict[str, Any] = {} + for path, data in files.items(): + cur = root + parts = path.split("/") + for part in parts[:-1]: cur = cur.setdefault(part, {}) + if parts[-1] in cur: raise ValidationError("tree path collision") + cur[parts[-1]] = data + def build(node: Mapping[str, Any]) -> str: + rows = [] + for name, value in sorted(node.items(), key=lambda x: x[0].encode()): + if isinstance(value, dict): mode, oid = b"40000", build(value) + else: mode, oid = b"100644", git_oid("blob", value) + rows.append(mode + b" " + name.encode() + b"\0" + bytes.fromhex(oid)) + body = b"".join(rows) + return git_oid("tree", body) + return build(root) + +def make_capture(source: Mapping[str, bytes], locks: Mapping[str, bytes], producer: str) -> dict[str, Any]: + tracked = dict(source) + for path, data in locks.items(): + if path in tracked: raise ValidationError("capture inventory overlap") + tracked[path] = data + paths = sorted(tracked) + tree = _git_tree(tracked) + ident = b"D003 Evidence 0 +0000" + payload = (b"tree " + tree.encode() + b"\nauthor " + ident + b"\ncommitter " + ident + + b"\n\nD-003 v2 synthetic " + producer.encode() + b" capture\n") + return {"source": dict(source), "dependencies": {}, "locks": dict(locks), "tracked": tracked, + "tracked_paths": paths, "commit_payload": payload, "commit": git_oid("commit", payload), "clean_tree": True} + +def _freeze_bytes() -> tuple[dict[str, Any], dict[str, bytes]]: + _require_freeze_ancestor() + if _git("rev-parse", FREEZE_COMMIT + "^{tree}") != FREEZE_TREE: raise ValidationError("freeze tree mismatch") + if _git("rev-parse", IMPLEMENTATION_COMMIT + "^{tree}") != IMPLEMENTATION_TREE: raise ValidationError("implementation tree mismatch") + if _git("rev-parse", PREREGISTRATION_COMMIT + "^{tree}") != PREREGISTRATION_TREE: raise ValidationError("preregistration tree mismatch") + for oid, kind in ((FREEZE_COMMIT, "commit"), (IMPLEMENTATION_COMMIT, "commit"), (PREREGISTRATION_COMMIT, "commit"), + (FREEZE_TREE, "tree"), (IMPLEMENTATION_TREE, "tree"), (PREREGISTRATION_TREE, "tree")): + if _object_type(oid) != kind: raise ValidationError("Git object type mismatch") + freeze_raw = _show(FREEZE_COMMIT, FREEZE_PATH) + if len(freeze_raw) != FREEZE_LENGTH or sha256(freeze_raw) != FREEZE_SHA256 or git_oid("blob", freeze_raw) != FREEZE_BLOB: + raise ValidationError("freeze byte identity mismatch") + if _git("rev-parse", FREEZE_COMMIT + ":" + FREEZE_PATH) != FREEZE_BLOB: raise ValidationError("freeze committed blob mismatch") + freeze = parse_canonical(freeze_raw, "freeze") + if freeze != FREEZE_EXPECTED or set(freeze) != set(FREEZE_EXPECTED): raise ValidationError("freeze schema/value mismatch") + blobs: dict[str, bytes] = {} + for path, (commit, length, digest, blob) in BOUND.items(): + raw = _show(FREEZE_COMMIT, path) + bound_raw = _show(commit, path) + if raw != bound_raw or len(raw) != length or sha256(raw) != digest or git_oid("blob", raw) != blob: + raise ValidationError("frozen path byte mismatch: " + path) + if _git("rev-parse", commit + ":" + path) != blob or _git("rev-parse", FREEZE_COMMIT + ":" + path) != blob: + raise ValidationError("frozen path Git blob mismatch: " + path) + if _object_type(blob) != "blob": raise ValidationError("frozen path object type mismatch: " + path) + blobs[path] = raw + manifest = parse_canonical(blobs[MANIFEST_PATH], "manifest") + if manifest.get("schema") != "D003-MANIFEST-V2": raise ValidationError("manifest schema mismatch") + frozen_paths = [FREEZE_PATH, *BOUND] + if _git("diff", "--name-only", "--", *frozen_paths) or _git("diff", "--cached", "--name-only", "--", *frozen_paths): + raise ValidationError("frozen path is dirty") + return freeze, blobs + +def validate_boundary() -> dict[str, Any]: + freeze, blobs = _freeze_bytes() + _run_validator(blobs, {"action": "manifest"}) + return {"freeze_commit": FREEZE_COMMIT, "freeze_sha256": FREEZE_SHA256, + "manifest_sha256": sha256(blobs[MANIFEST_PATH]), "bound_path_count": len(BOUND) + 1, + "decision_status": freeze["embedded_decision_status"]} + +def _encode(value: Any) -> Any: + if isinstance(value, bytes): return {"$bytes": base64.b64encode(value).decode("ascii")} + if isinstance(value, dict): return {k: _encode(v) for k, v in value.items()} + if isinstance(value, list): return [_encode(v) for v in value] + return value + +def _decode(value: Any) -> Any: + if isinstance(value, dict) and set(value) == {"$bytes"}: + try: return base64.b64decode(value["$bytes"], validate=True) + except Exception as exc: raise ValidationError("invalid child base64") from exc + if isinstance(value, dict): return {k: _decode(v) for k, v in value.items()} + if isinstance(value, list): return [_decode(v) for v in value] + return value + +CHILD_COMMON = r''' +import base64,json,sys,unicodedata + +def pairs(xs): + d={} + for k,v in xs: + if k in d or unicodedata.normalize("NFC",k)!=k:raise ValueError("bad key") + d[k]=v + return d +def dec(v): + if isinstance(v,dict) and set(v)=={"$bytes"}:return base64.b64decode(v["$bytes"],validate=True) + if isinstance(v,dict):return {k:dec(x) for k,x in v.items()} + if isinstance(v,list):return [dec(x) for x in v] + return v +def enc(v): + if isinstance(v,bytes):return {"$bytes":base64.b64encode(v).decode()} + if isinstance(v,dict):return {k:enc(x) for k,x in v.items()} + if isinstance(v,(list,tuple)):return [enc(x) for x in v] + return v +req=dec(json.loads(sys.stdin.buffer.read().decode(),object_pairs_hook=pairs,parse_float=lambda x:(_ for _ in()).throw(ValueError("float")),parse_constant=lambda x:(_ for _ in()).throw(ValueError("constant")))) +''' + +R_CHILD = CHILD_COMMON + r''' +import importlib.util +spec=importlib.util.spec_from_file_location("producer",sys.argv[1]);mod=importlib.util.module_from_spec(spec);spec.loader.exec_module(mod) +cap=req["capture"] +if req["runner"]=="R1": + _,o=mod.run_manifest_bytes(req["manifest"]);_,i=mod.build_identity("R1",o,cap["source"],cap["dependencies"],cap["locks"],cap["commit"],True) +else: + _,o=mod.run(req["manifest"]);repo={"clean_tree":True,"commit":cap["commit"],"commit_payload":cap["commit_payload"],"tracked_files":cap["tracked"],"tracked_paths":cap["tracked_paths"]};_,i=mod.make_identity(o,"R2",cap["source"],cap["dependencies"],cap["locks"],repo) +sys.stdout.write(json.dumps(enc({"O":o,"I":i}),sort_keys=True,separators=(",",":"))+"\n") +''' + +V_CHILD = CHILD_COMMON + r''' +import importlib.util +spec=importlib.util.spec_from_file_location("verifier",sys.argv[1]);v=importlib.util.module_from_spec(spec);spec.loader.exec_module(v) +def cap(c):return v.ProducerCapture(c["source"],c["dependencies"],c["locks"],c["tracked"],c["tracked_paths"],c["commit_payload"],c["clean_tree"]) +r1,r2,v3=cap(req["r1"]),cap(req["r2"]),cap(req["v3"]) +report=v.verify_v3(req["manifest"],req["O1"],req["O2"],req["I1"],req["I2"],r1=r1,r2=r2,v3_capture=v3);V=v.canonical_bytes(report) +evidence=v.verify_evidence(req["manifest"],req["O1"],req["O2"],req["I1"],req["I2"],V,r1=r1,r2=r2,v3_capture=v3);E=v.canonical_bytes(evidence) +v.validate_v3_value(report,V,[req["manifest"],req["O1"],req["O2"],req["I1"],req["I2"]],v.load_manifest(req["manifest"]),["R1","R2"]) +sys.stdout.write(json.dumps(enc({"V3":V,"E":E}),sort_keys=True,separators=(",",":"))+"\n") +''' + +VALIDATOR_CHILD = CHILD_COMMON + r''' +import importlib.util +spec=importlib.util.spec_from_file_location("validator",sys.argv[1]);d=importlib.util.module_from_spec(spec);spec.loader.exec_module(d) +m=d.parse_bytes(req["manifest"]);d.validate_bytes(req["manifest"]) +if req["action"]=="artifacts": + p=req["provenance"];d.validate_o_bytes(req["O1"],m,req["manifest"]);d.validate_o_bytes(req["O2"],m,req["manifest"]) + d.validate_i_value(d.parse_bytes(req["I1"]),req["O1"],m["schemas"],p["I1"]);d.validate_i_value(d.parse_bytes(req["I2"]),req["O2"],m["schemas"],p["I2"]) + d.validate_v3_bytes(req["V3"],m,req["manifest"],req["O1"],req["O2"],req["I1"],req["I2"]) + result=d.validate_e_bytes(req["E"],m,req["manifest"],req["O1"],req["O2"],req["I1"],req["I2"],req["V3"],p) +else:result="VALID" +sys.stdout.write(json.dumps({"result":result},sort_keys=True,separators=(",",":"))+"\n") +''' + +def _child(source_name: str, source: bytes, script: str, request: Mapping[str, Any]) -> dict[str, Any]: + with tempfile.TemporaryDirectory(prefix="d003-child-") as td: + cwd = Path(td) + src = cwd / source_name + driver = cwd / "driver.py" + src.write_bytes(source); driver.write_text(script, encoding="utf-8") + env = {"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "PYTHONHASHSEED": "0", "PYTHONIOENCODING": "utf-8", + "PYTHONNOUSERSITE": "1", "PYTHONDONTWRITEBYTECODE": "1", "PYTHONPATH": ""} + try: + cp = subprocess.run([sys.executable, "-I", "-B", str(driver), str(src)], input=canonical(_encode(dict(request))), + stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd, env=env, timeout=TIMEOUT, check=False) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ValidationError("child failed or timed out") from exc + if cp.returncode != 0 or cp.stderr: + detail = cp.stderr.decode("utf-8", "replace")[-500:].replace("\n", " ") + raise ValidationError(source_name + ": child nonzero/stderr: " + detail) + value = parse_canonical(cp.stdout, "child stdout") + if not isinstance(value, dict): raise ValidationError("child stdout schema") + return _decode(value) + +def _run_validator(blobs: Mapping[str, bytes], request: Mapping[str, Any]) -> str: + req = {"manifest": blobs[MANIFEST_PATH], **request} + result = _child("d003_manifest_v2.py", blobs["decisions/d003_manifest_v2.py"], VALIDATOR_CHILD, req) + if set(result) != {"result"} or result["result"] not in ("VALID", "ADOPT_ALGEBRA"): + raise ValidationError("validator child result") + return result["result"] + +def _repository(cap: Mapping[str, Any]) -> dict[str, Any]: + return {"tracked_paths": cap["tracked_paths"], "blob_ids": {p: git_oid("blob", b) for p, b in cap["tracked"].items()}, + "tree_id": _git_tree(cap["tracked"]), "commit_bytes": cap["commit_payload"], "commit_id": cap["commit"], "clean_tree": True} + +def _provenance(cap: Mapping[str, Any], root: str) -> dict[str, Any]: + return {"source": cap["source"], "dependency": cap["dependencies"], "lock": cap["locks"], + "repository": _repository(cap), "root": root} + +def _check_artifact(raw: bytes, schema: str, label: str) -> dict[str, Any]: + value = parse_canonical(raw, label) + if not isinstance(value, dict) or value.get("schema") != schema: raise ValidationError(label + ": wrong schema") + return value + +def _fsync_directory(path: Path) -> None: + fd = os.open(path, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + +def _atomic_publish_noreplace(stage: Path, output: Path) -> None: + if not sys.platform.startswith("linux"): + cause = OSError(errno.ENOSYS, "renameat2 is required on Linux") + raise ValidationError("atomic no-replace directory publication is unavailable") from cause + try: + libc = ctypes.CDLL(None, use_errno=True) + renameat2 = libc.renameat2 + except (OSError, AttributeError) as exc: + cause = OSError(errno.ENOSYS, "libc renameat2 symbol is unavailable") + raise ValidationError("atomic no-replace directory publication is unavailable") from cause + renameat2.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint] + renameat2.restype = ctypes.c_int + ctypes.set_errno(0) + result = renameat2(AT_FDCWD, os.fsencode(stage), AT_FDCWD, os.fsencode(output), RENAME_NOREPLACE) + if result == 0: + return + err = ctypes.get_errno() + cause = OSError(err or errno.EIO, os.strerror(err) if err else "renameat2 failed without errno") + if err in (errno.EEXIST, errno.ENOTEMPTY): + raise ValidationError("output path appeared before publication") from cause + raise ValidationError("atomic no-replace directory publication failed") from cause + +def _retain(output: Path, artifacts: Mapping[str, bytes]) -> None: + if output.exists(): raise ValidationError("output path must not exist") + parent = output.parent + if not parent.is_dir(): raise ValidationError("output parent directory must exist") + stage = Path(tempfile.mkdtemp(prefix=".d003-stage-", dir=parent)) + published = False + try: + for name in ARTIFACT_NAMES: + path = stage / name + with path.open("xb") as f: + f.write(artifacts[name]); f.flush(); os.fsync(f.fileno()) + _fsync_directory(stage) + _atomic_publish_noreplace(stage, output) + published = True + try: + _fsync_directory(parent) + except OSError as exc: + raise ValidationError("evidence published but parent directory fsync failed; durability uncertain") from exc + finally: + if not published: + shutil.rmtree(stage, ignore_errors=True) + +def run(output: str | os.PathLike[str]) -> dict[str, Any]: + out = Path(output) + if out.exists(): raise ValidationError("output path must not exist") + if not out.parent.is_dir(): raise ValidationError("output parent directory must exist") + freeze, blobs = _freeze_bytes() + _run_validator(blobs, {"action": "manifest"}) + locks = {PACKET_PATH: blobs[PACKET_PATH], MANIFEST_PATH: blobs[MANIFEST_PATH]} + r1 = make_capture({"decisions/d003_r1_v2.py": blobs["decisions/d003_r1_v2.py"]}, locks, "R1") + r2 = make_capture({"decisions/d003_r2_v2.py": blobs["decisions/d003_r2_v2.py"]}, locks, "R2") + v3 = make_capture({"decisions/d003_v3_v2.py": blobs["decisions/d003_v3_v2.py"]}, locks, "V3") + first = _child("d003_r1_v2.py", blobs["decisions/d003_r1_v2.py"], R_CHILD, + {"runner": "R1", "manifest": blobs[MANIFEST_PATH], "capture": r1}) + second = _child("d003_r2_v2.py", blobs["decisions/d003_r2_v2.py"], R_CHILD, + {"runner": "R2", "manifest": blobs[MANIFEST_PATH], "capture": r2}) + if set(first) != {"O", "I"} or set(second) != {"O", "I"}: raise ValidationError("producer child schema") + O1, I1, O2, I2 = first["O"], first["I"], second["O"], second["I"] + for label, raw in (("O1", O1), ("O2", O2), ("I1", I1), ("I2", I2)): + if not isinstance(raw, bytes): raise ValidationError(label + ": child did not return bytes") + _check_artifact(O1, "D003-O-2", "O1"); _check_artifact(O2, "D003-O-2", "O2") + _check_artifact(I1, "D003-I-2", "I1"); _check_artifact(I2, "D003-I-2", "I2") + # Expected output is a post-execution oracle, never a generated fixture. + for label, raw in (("O1", O1), ("O2", O2)): + if len(raw) != freeze["output_expected_byte_length"] or sha256(raw) != freeze["output_expected_sha256"]: + raise ValidationError(label + ": frozen expected output mismatch") + if O1 != O2: raise ValidationError("O raw bytes differ") + third = _child("d003_v3_v2.py", blobs["decisions/d003_v3_v2.py"], V_CHILD, + {"manifest": blobs[MANIFEST_PATH], "O1": O1, "O2": O2, "I1": I1, "I2": I2, + "r1": r1, "r2": r2, "v3": v3}) + if set(third) != {"V3", "E"} or not all(isinstance(third[k], bytes) for k in third): raise ValidationError("V3 child schema") + V3, E = third["V3"], third["E"] + vv = _check_artifact(V3, "D003-V3-2", "V3"); ee = _check_artifact(E, "D003-E-2", "E") + provenance = {"I1": _provenance(r1, "decisions/d003_r1_v2.py"), + "I2": _provenance(r2, "decisions/d003_r2_v2.py"), + "V3": _provenance(v3, "decisions/d003_v3_v2.py")} + result = _run_validator(blobs, {"action": "artifacts", "O1": O1, "O2": O2, "I1": I1, "I2": I2, + "V3": V3, "E": E, "provenance": provenance}) + if result != "ADOPT_ALGEBRA" or vv.get("accepted") is not True or ee.get("accepted") is not True: + result = "OPEN_IMPLEMENTATION_DEFECT" + raw_artifacts = {"O1.json": O1, "O2.json": O2, "I1.json": I1, "I2.json": I2, "V3.json": V3, "E.json": E} + summary = { + "schema": "D003-RETAINED-EVIDENCE-SUMMARY-1", "decision_id": "D-003", + "freeze": {"commit": FREEZE_COMMIT, "sha256": FREEZE_SHA256}, + "manifest": {"sha256": MANIFEST_SHA256, "byte_length": MANIFEST_LENGTH}, + "artifacts": {name[:-5]: {"sha256": sha256(raw), "byte_length": len(raw)} for name, raw in raw_artifacts.items()}, + "counts": {"traces": freeze["trace_count"], "assertions": freeze["assertion_count"], + "required_outputs": freeze["required_output_count"], "v3_checks": len(vv["checks"]), "evidence_checks": len(ee["checks"])}, + "raw_outputs_equal": O1 == O2, "raw_outputs_match_frozen_expected": True, + "v3_accepted": vv["accepted"], "evidence_accepted": ee["accepted"], "result": result, + "selects_representation": False, "decision_status_before": "OPEN", "decision_status_after": "OPEN", + "register_or_adr_update_performed": False, + "statement": "No decision register or ADR update was performed; D-003 remains OPEN." + } + summary_raw = canonical(summary); parse_canonical(summary_raw, "summary") + all_artifacts = {**raw_artifacts, "summary.json": summary_raw} + _retain(out, all_artifacts) + return summary + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="command", required=True) + sub.add_parser("validate", help="validate the frozen boundary without executing producers") + rp = sub.add_parser("run", help="execute and atomically retain evidence"); rp.add_argument("output_directory") + args = parser.parse_args(argv) + try: + value = validate_boundary() if args.command == "validate" else run(args.output_directory) + except (ValidationError, OSError) as exc: + parser.error(str(exc)) + sys.stdout.buffer.write(canonical(value)) + return 0 + +if __name__ == "__main__": raise SystemExit(main()) diff --git a/decisions/test_d003_evidence_run.py b/decisions/test_d003_evidence_run.py new file mode 100644 index 0000000000..58b81ca79b --- /dev/null +++ b/decisions/test_d003_evidence_run.py @@ -0,0 +1,374 @@ +"""Focused adversarial tests for the bounded D-003 evidence orchestrator.""" +import ast +import ctypes +import errno +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from decisions import d003_evidence_run as r + + +def artifact(schema, **extra): + return r.canonical({"schema": schema, **extra}) + + +def sys_platform_renameat2_available(): + if not sys.platform.startswith("linux"): return False + try: return bool(ctypes.CDLL(None, use_errno=True).renameat2) + except (OSError, AttributeError): return False + + +class CanonicalAndCapture(unittest.TestCase): + def test_canonical_rejects_float_and_duplicate_stdout(self): + with self.assertRaises(r.ValidationError): r.canonical({"x": 1.0}) + with self.assertRaises(r.ValidationError): r.parse_canonical(b'{"x":1,"x":1}\n', "x") + with self.assertRaises(r.ValidationError): r.parse_canonical(b'{"x":1}\nextra\n', "x") + + def test_capture_is_exact_deterministic_and_producer_specific(self): + locks = {r.PACKET_PATH: b"p", r.MANIFEST_PATH: b"m"} + a = r.make_capture({"decisions/d003_r1_v2.py": b"r1"}, locks, "R1") + b = r.make_capture({"decisions/d003_r1_v2.py": b"r1"}, locks, "R1") + self.assertEqual(a, b) + self.assertEqual(a["dependencies"], {}) + self.assertEqual(a["tracked_paths"], sorted(a["tracked"])) + self.assertEqual(set(a["tracked"]), {"decisions/d003_r1_v2.py", r.PACKET_PATH, r.MANIFEST_PATH}) + self.assertEqual(a["commit"], r.git_oid("commit", a["commit_payload"])) + self.assertIn(b"synthetic R1 capture", a["commit_payload"]) + + def test_capture_rejects_inventory_overlap(self): + with self.assertRaises(r.ValidationError): r.make_capture({r.PACKET_PATH: b"x"}, {r.PACKET_PATH: b"x"}, "R1") + + +class Boundary(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.result = r.validate_boundary() + + def test_exact_frozen_boundary_validates(self): + self.assertEqual(self.result["freeze_commit"], r.FREEZE_COMMIT) + self.assertEqual(self.result["manifest_sha256"], r.MANIFEST_SHA256) + self.assertEqual(self.result["decision_status"], "OPEN") + + def test_descendant_head_is_accepted_by_ancestry_helper(self): + descendant = subprocess.CompletedProcess([], 0, b"", b"") + with mock.patch.object(subprocess, "run", return_value=descendant) as run: + r._require_freeze_ancestor() + self.assertEqual(run.call_args.args[0][-4:], ["merge-base", "--is-ancestor", r.FREEZE_COMMIT, "HEAD"]) + + def test_ancestry_abnormal_results_fail_closed(self): + cases = [ + ("status2", subprocess.CompletedProcess([], 2, b"", b"")), + ("signal", subprocess.CompletedProcess([], -9, b"", b"")), + ("stdout", subprocess.CompletedProcess([], 0, b"unexpected", b"")), + ("stderr", subprocess.CompletedProcess([], 0, b"", b"unexpected")), + ("oserror", OSError("git missing")), + ("timeout", subprocess.TimeoutExpired(["git"], 1)), + ] + for name, outcome in cases: + patch = mock.patch.object(subprocess, "run", side_effect=outcome) if isinstance(outcome, BaseException) \ + else mock.patch.object(subprocess, "run", return_value=outcome) + with self.subTest(name=name), patch: + with self.assertRaises(r.ValidationError) as caught: r._require_freeze_ancestor() + if name in ("oserror", "timeout"): self.assertIs(caught.exception.__cause__, outcome) + + def test_nonancestor_fails_before_child(self): + unrelated = subprocess.CompletedProcess([], 1, b"", b"") + with mock.patch.object(subprocess, "run", return_value=unrelated), mock.patch.object(r, "_child") as child: + with self.assertRaisesRegex(r.ValidationError, "not an ancestor"): + r.validate_boundary() + child.assert_not_called() + + def test_dirty_frozen_path_fails_before_child(self): + real = r._git + def dirty(*args, **kwargs): + if args[:3] == ("diff", "--name-only", "--"): return r.MANIFEST_PATH + return real(*args, **kwargs) + with mock.patch.object(r, "_git", side_effect=dirty), mock.patch.object(r, "_child") as child: + with self.assertRaisesRegex(r.ValidationError, "dirty"): + r.validate_boundary() + child.assert_not_called() + + def test_freeze_record_tamper_fails_closed(self): + real = r._show + def tamper(commit, path): + data = real(commit, path) + return data[:-2] + b"x\n" if path == r.FREEZE_PATH else data + with mock.patch.object(r, "_show", side_effect=tamper), mock.patch.object(r, "_child") as child: + with self.assertRaisesRegex(r.ValidationError, "freeze byte"): + r.validate_boundary() + child.assert_not_called() + + +class ChildSecurity(unittest.TestCase): + def test_child_rejects_stderr_nonzero_and_extra_output(self): + completed = subprocess.CompletedProcess([], 0, b'{}\n', b'noise') + with mock.patch.object(subprocess, "run", return_value=completed): + with self.assertRaisesRegex(r.ValidationError, "stderr"): + r._child("x.py", b"", "", {}) + completed = subprocess.CompletedProcess([], 0, b'{}\n{}\n', b'') + with mock.patch.object(subprocess, "run", return_value=completed): + with self.assertRaises(r.ValidationError): r._child("x.py", b"", "", {}) + + def test_child_uses_no_shell_isolated_flags_env_and_cwd(self): + seen = {} + def fake(argv, **kwargs): + seen.update(argv=argv, kwargs=kwargs) + return subprocess.CompletedProcess(argv, 0, b'{}\n', b'') + with mock.patch.object(subprocess, "run", side_effect=fake): r._child("x.py", b"x=1\n", "", {}) + self.assertNotIn("shell", seen["kwargs"]) + self.assertIn("-I", seen["argv"]); self.assertIn("-B", seen["argv"]) + self.assertEqual(seen["kwargs"]["env"]["PYTHONPATH"], "") + self.assertNotEqual(Path(seen["kwargs"]["cwd"]), r.ROOT) + self.assertEqual(seen["kwargs"]["timeout"], r.TIMEOUT) + + +class AtomicPublishPrimitive(unittest.TestCase): + class Function: + def __init__(self, result, err=0): self.result, self.err, self.calls = result, err, [] + def __call__(self, *args): + self.calls.append(args); ctypes.set_errno(self.err); return self.result + + def test_libc_signature_flags_dirfds_and_fsencoding(self): + function = self.Function(0); libc = type("Libc", (), {"renameat2": function})() + with mock.patch.object(r.ctypes, "CDLL", return_value=libc): + r._atomic_publish_noreplace(Path("stage-é"), Path("output-é")) + self.assertEqual(function.restype, ctypes.c_int) + self.assertEqual(function.argtypes, [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]) + self.assertEqual(function.calls, [(r.AT_FDCWD, os.fsencode("stage-é"), r.AT_FDCWD, os.fsencode("output-é"), r.RENAME_NOREPLACE)]) + + def test_conflict_errno_mapping_has_oserror_cause(self): + for err in (errno.EEXIST, errno.ENOTEMPTY): + function = self.Function(-1, err); libc = type("Libc", (), {"renameat2": function})() + with self.subTest(errno=err), mock.patch.object(r.ctypes, "CDLL", return_value=libc): + with self.assertRaisesRegex(r.ValidationError, "output path appeared") as caught: + r._atomic_publish_noreplace(Path("stage"), Path("output")) + self.assertIsInstance(caught.exception.__cause__, OSError) + self.assertEqual(caught.exception.__cause__.errno, err) + + def test_unavailable_and_other_errno_fail_closed_with_oserror_cause(self): + cases = [("symbol", type("Libc", (), {})()), + ("errno", type("Libc", (), {"renameat2": self.Function(-1, errno.EXDEV)})())] + for name, libc in cases: + with self.subTest(name=name), mock.patch.object(r.ctypes, "CDLL", return_value=libc): + with self.assertRaises(r.ValidationError) as caught: + r._atomic_publish_noreplace(Path("stage"), Path("output")) + self.assertIsInstance(caught.exception.__cause__, OSError) + + +class Orchestration(unittest.TestCase): + def fake_boundary(self): + blobs = {p: b"source\n" for p in r.BOUND} + blobs[r.MANIFEST_PATH] = b"manifest\n"; blobs[r.PACKET_PATH] = b"packet\n" + return ({"output_expected_byte_length": len(self.o), "output_expected_sha256": r.sha256(self.o), + "trace_count": 10, "assertion_count": 270, "required_output_count": 2}, blobs) + + def setUp(self): + self.o = artifact("D003-O-2") + self.i1 = artifact("D003-I-2", runner_id="R1") + self.i2 = artifact("D003-I-2", runner_id="R2") + self.v3 = artifact("D003-V3-2", accepted=True, checks=["v"]) + self.e = artifact("D003-E-2", accepted=True, checks=["e"]) + + def execute(self, out): + calls = [] + def child(name, source, script, request): + calls.append((name, request)) + if name == "d003_r1_v2.py": return {"O": self.o, "I": self.i1} + if name == "d003_r2_v2.py": return {"O": self.o, "I": self.i2} + if name == "d003_v3_v2.py": return {"V3": self.v3, "E": self.e} + return {"result": "ADOPT_ALGEBRA"} + with mock.patch.object(r, "_freeze_bytes", side_effect=self.fake_boundary), \ + mock.patch.object(r, "_run_validator", return_value="ADOPT_ALGEBRA"), \ + mock.patch.object(r, "_child", side_effect=child): + summary = r.run(out) + return summary, calls + + def test_success_is_atomic_canonical_and_two_independent_r_invocations(self): + with tempfile.TemporaryDirectory() as td: + out = Path(td) / "one"; summary, calls = self.execute(out) + self.assertEqual(sorted(p.name for p in out.iterdir()), sorted(r.ARTIFACT_NAMES)) + for p in out.iterdir(): r.parse_canonical(p.read_bytes(), p.name) + self.assertEqual([x[0] for x in calls[:2]], ["d003_r1_v2.py", "d003_r2_v2.py"]) + self.assertIsNot(calls[0][1]["capture"], calls[1][1]["capture"]) + self.assertEqual(summary["result"], "ADOPT_ALGEBRA") + self.assertFalse(summary["selects_representation"]) + self.assertEqual(summary["decision_status_before"], "OPEN") + self.assertEqual(summary["decision_status_after"], "OPEN") + self.assertFalse(summary["register_or_adr_update_performed"]) + + def test_deterministic_rerun_separate_directories(self): + with tempfile.TemporaryDirectory() as td: + a, b = Path(td) / "a", Path(td) / "b" + self.execute(a); self.execute(b) + self.assertEqual({p.name: p.read_bytes() for p in a.iterdir()}, {p.name: p.read_bytes() for p in b.iterdir()}) + + def test_any_preexisting_output_rejected_before_execution(self): + with tempfile.TemporaryDirectory() as td: + for name, populate in (("empty", False), ("nonempty", True)): + out = Path(td) / name; out.mkdir() + if populate: (out / "occupied").write_text("x") + with self.subTest(name=name), mock.patch.object(r, "_freeze_bytes") as boundary: + with self.assertRaisesRegex(r.ValidationError, "must not exist"): r.run(out) + boundary.assert_not_called() + + def test_success_invokes_atomic_primitive_once_and_publishes_complete_set(self): + real_publish = r._atomic_publish_noreplace + with tempfile.TemporaryDirectory() as td, mock.patch.object(r, "_atomic_publish_noreplace", wraps=real_publish) as publish: + out = Path(td) / "one"; self.execute(out) + publish.assert_called_once() + stage, final = map(Path, publish.call_args.args) + self.assertEqual(final, out) + self.assertTrue(stage.name.startswith(".d003-stage-")) + self.assertFalse(stage.exists()) + self.assertEqual(sorted(p.name for p in out.iterdir()), sorted(r.ARTIFACT_NAMES)) + + def test_real_atomic_primitive_does_not_replace_competing_destinations(self): + if not sys_platform_renameat2_available(): self.skipTest("libc renameat2 truly unavailable") + for populated in (False, True): + with self.subTest(populated=populated), tempfile.TemporaryDirectory() as td: + parent = Path(td); stage = parent / "stage"; output = parent / "output" + stage.mkdir(); (stage / "new").write_bytes(b"new") + output.mkdir() + if populated: (output / "old").write_bytes(b"old") + with self.assertRaisesRegex(r.ValidationError, "output path appeared"): + r._atomic_publish_noreplace(stage, output) + self.assertTrue(stage.is_dir()); self.assertEqual((stage / "new").read_bytes(), b"new") + self.assertTrue(output.is_dir()) + self.assertEqual(sorted(p.name for p in output.iterdir()), ["old"] if populated else []) + + def test_competing_destination_at_retain_primitive_call_is_untouched_and_stage_cleans(self): + artifacts = {name: b"new" for name in r.ARTIFACT_NAMES} + if not sys_platform_renameat2_available(): self.skipTest("libc renameat2 truly unavailable") + for populated in (False, True): + with self.subTest(populated=populated), tempfile.TemporaryDirectory() as td: + out = Path(td) / "out"; real_publish = r._atomic_publish_noreplace + def compete(stage, output): + output.mkdir() + if populated: (output / "old").write_bytes(b"old") + return real_publish(stage, output) + with mock.patch.object(r, "_atomic_publish_noreplace", side_effect=compete): + with self.assertRaisesRegex(r.ValidationError, "output path appeared"): r._retain(out, artifacts) + self.assertEqual(sorted(p.name for p in out.iterdir()), ["old"] if populated else []) + self.assertFalse(any(p.name.startswith(".d003-stage-") for p in Path(td).iterdir())) + + def test_prepublication_stage_fsync_failure_leaves_final_absent(self): + artifacts = {name: b"x" for name in r.ARTIFACT_NAMES} + with tempfile.TemporaryDirectory() as td: + out = Path(td) / "out" + with mock.patch.object(r, "_fsync_directory", side_effect=OSError("injected")), \ + mock.patch.object(r, "_atomic_publish_noreplace") as publish: + with self.assertRaisesRegex(OSError, "injected"): r._retain(out, artifacts) + publish.assert_not_called() + self.assertFalse(out.exists()) + self.assertFalse(any(p.name.startswith(".d003-stage-") for p in Path(td).iterdir())) + + def test_file_fsync_failure_leaves_no_final_or_stage(self): + artifacts = {name: b"x" for name in r.ARTIFACT_NAMES} + real_fsync = os.fsync; calls = 0 + def fail_file(fd): + nonlocal calls + calls += 1 + if calls == 1: raise OSError("file fsync injected") + return real_fsync(fd) + with tempfile.TemporaryDirectory() as td: + out = Path(td) / "out" + with mock.patch.object(r.os, "fsync", side_effect=fail_file), \ + mock.patch.object(r, "_atomic_publish_noreplace") as publish: + with self.assertRaisesRegex(OSError, "file fsync injected"): r._retain(out, artifacts) + publish.assert_not_called() + self.assertFalse(out.exists()) + self.assertFalse(any(p.name.startswith(".d003-stage-") for p in Path(td).iterdir())) + + def test_parent_fsync_failure_is_reported_after_publication(self): + artifacts = {name: b"x" for name in r.ARTIFACT_NAMES} + calls = 0 + def fsync_dir(path): + nonlocal calls + calls += 1 + if calls == 2: raise OSError("parent fsync") + with tempfile.TemporaryDirectory() as td: + out = Path(td) / "out" + with mock.patch.object(r, "_fsync_directory", side_effect=fsync_dir): + with self.assertRaisesRegex(r.ValidationError, "published.*durability uncertain"): r._retain(out, artifacts) + self.assertTrue(out.is_dir()) + self.assertEqual(sorted(p.name for p in out.iterdir()), sorted(r.ARTIFACT_NAMES)) + + def test_postexecution_expected_hash_check_and_no_retention(self): + bad = artifact("D003-O-2", wrong=True); calls = [] + def child(name, source, script, request): + calls.append(name) + return {"O": bad, "I": self.i1 if "r1" in name else self.i2} + with tempfile.TemporaryDirectory() as td, \ + mock.patch.object(r, "_freeze_bytes", side_effect=self.fake_boundary), \ + mock.patch.object(r, "_run_validator", return_value="VALID"), \ + mock.patch.object(r, "_child", side_effect=child): + out = Path(td) / "out" + with self.assertRaisesRegex(r.ValidationError, "expected output"): r.run(out) + self.assertEqual(calls, ["d003_r1_v2.py", "d003_r2_v2.py"]) + self.assertFalse(out.exists()) + + def test_v3_tamper_leaves_no_artifacts(self): + def child(name, source, script, request): + if name == "d003_r1_v2.py": return {"O": self.o, "I": self.i1} + if name == "d003_r2_v2.py": return {"O": self.o, "I": self.i2} + return {"V3": b'{"schema":"BAD"}\n', "E": self.e} + with tempfile.TemporaryDirectory() as td, \ + mock.patch.object(r, "_freeze_bytes", side_effect=self.fake_boundary), \ + mock.patch.object(r, "_run_validator", return_value="VALID"), \ + mock.patch.object(r, "_child", side_effect=child): + out = Path(td) / "out" + with self.assertRaisesRegex(r.ValidationError, "wrong schema"): r.run(out) + self.assertFalse(out.exists()) + + +class FrozenDefectAndSourceGuards(unittest.TestCase): + def test_actual_frozen_run_fails_closed_on_undeclared_stdlib_and_retains_none(self): + with tempfile.TemporaryDirectory() as td: + out = Path(td) / "evidence" + with self.assertRaisesRegex(r.ValidationError, "undeclared import binascii"): + r.run(out) + self.assertFalse(out.exists()) + + def test_no_semantic_imports_or_decision_writes(self): + source = Path(r.__file__).read_text() + tree = ast.parse(source) + imported = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): imported.update(x.name for x in node.names) + elif isinstance(node, ast.ImportFrom): imported.add(node.module or "") + self.assertFalse(any("d003_r1_v2" in x or "d003_r2_v2" in x or "d003_v3_v2" in x for x in imported)) + self.assertNotIn("DECISIONS.md", source) + self.assertNotIn("decisions/adr", source.lower()) + self.assertNotIn("shell=True", source) + + def test_frozen_producers_do_not_import_each_other(self): + for path in ("decisions/d003_r1_v2.py", "decisions/d003_r2_v2.py", "decisions/d003_v3_v2.py"): + source = r._show(r.FREEZE_COMMIT, path).decode() + self.assertNotRegex(source, r"(?:import|from)\s+.*d003_(?:r1|r2|v3)_v2") + + def test_expected_output_is_only_postexecution_oracle(self): + source = Path(r.__file__).read_text() + run_body = source[source.index("def run(output"):source.index("def main(")] + self.assertLess(run_body.index("first = _child"), run_body.index("output_expected_sha256")) + self.assertNotIn("bytes.fromhex(freeze[\"output_expected", run_body) + + def test_publication_source_requires_renameat2_noreplace_and_has_no_fallback(self): + source = Path(r.__file__).read_text() + primitive = source[source.index("def _atomic_publish_noreplace("):source.index("def _retain(")] + retain = source[source.index("def _retain("):source.index("def run(output")] + self.assertIn("RENAME_NOREPLACE", source) + self.assertIn("renameat2", primitive) + self.assertIn("os.fsencode(stage)", primitive); self.assertIn("os.fsencode(output)", primitive) + self.assertEqual(retain.count("_atomic_publish_noreplace(stage, output)"), 1) + for forbidden in ("os.replace(", "os.rename(", "os.link(", "shutil.copy", "subprocess", "shell=True"): + self.assertNotIn(forbidden, primitive + retain) + self.assertNotIn("stage / name, output / name", retain) + + +if __name__ == "__main__": unittest.main() -- 2.54.0