diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 339640ad..187e1c4f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1719,6 +1719,16 @@ jobs: - name: S2 step 8 — canonical patch bundle (patch + manifest + postimage) run: bash tests/patch_bundle_regressions.sh "$RUNNER_TEMP/s8" + # S2 step 9 — the structural self-gate over a step 8 bundle, ending in an INDEPENDENT + # git apply --check → apply in a hermetic throwaway repo that reproduces the postimage + # byte for byte. Asserts the ten gates, deterministic evidence, and — via forged + # fixtures rebound to reach the intended gate — the full APPLY_CHECK / APPLY_MISMATCH / + # HASH_MISMATCH / BUNDLE_LAYOUT / PRISTINE_SOURCE / PUBLICATION taxonomy, with the real + # checkout, index and config never touched. (The pure-function + byte-tampering cases + # are in tests/test_gate_patch.py, in the dotnet-free `tests` job.) + - name: S2 step 9 — structural self-gate (git apply verification) + run: bash tests/gate_regressions.sh "$RUNNER_TEMP/s9" + # A stale preimage SHA is the one refusal worth pinning in the workflow itself: it is # the invariant the whole hash-bound chain rests on. - name: S2 owen-rewrite — a stale preimage SHA is a hard refusal diff --git a/ownlang/__main__.py b/ownlang/__main__.py index cb7b9a37..92795028 100644 --- a/ownlang/__main__.py +++ b/ownlang/__main__.py @@ -597,6 +597,40 @@ def _cmd_apply(rest: list[str]) -> int: return 0 +def _cmd_gate(rest: list[str]) -> int: + """S2 step 9: `own-fix subscriptions gate` — the structural self-gate. Re-validates a + canonical step 8 bundle against its plan + candidates and proves the patch's semantics + with an INDEPENDENT host Git in a hermetic throwaway repo, then publishes a + byte-deterministic gate-result.json. No model, no o7, no analyzer, no target tests; the + real checkout / index / config are never touched. There is deliberately no `--git` + override — evidence of an INDEPENDENT apply cannot come from a caller-supplied stand-in.""" + from ownlang.fix_gate import GateError, run_gate + + flags = {"--bundle", "--plan", "--candidates", "--root", "--out"} + parsed = _own_fix_parse(rest, flags, set()) + if parsed is None: + return 2 + positional, opts = parsed + if positional or not all(opts.get(k) for k in + ("--bundle", "--plan", "--candidates", "--out")): + print("usage: own-fix subscriptions gate --bundle " + "--plan --candidates " + "--root --out ", file=sys.stderr) + return 2 + try: + published = run_gate(opts["--bundle"], opts["--plan"], opts["--candidates"], + opts.get("--root") or ".", opts["--out"]) + except GateError as exc: + print(f"own-fix: refuse: {exc.category}: {exc}", file=sys.stderr) + return 2 + except Exception as exc: # fail closed: any surprise is a refusal, not a traceback + print(f"own-fix: refuse: INFRASTRUCTURE: internal error " + f"({type(exc).__name__}: {exc})", file=sys.stderr) + return 2 + print(f"own-fix: wrote gate-result.json -> {published}") + return 0 + + def cmd_own_fix(rest: list[str]) -> int: """`own-fix subscriptions {candidates|render|validate-plan|apply} ...`.""" if len(rest) < 2 or rest[0] != "subscriptions": @@ -612,8 +646,10 @@ def cmd_own_fix(rest: list[str]) -> int: return _cmd_validate_plan(args) if verb == "apply": return _cmd_apply(args) + if verb == "gate": + return _cmd_gate(args) print(f"own-fix: unknown subcommand {verb!r} " - "(candidates | render | validate-plan | apply)", file=sys.stderr) + "(candidates | render | validate-plan | apply | gate)", file=sys.stderr) return 2 diff --git a/ownlang/fix_gate.py b/ownlang/fix_gate.py new file mode 100644 index 00000000..cfbfd1f5 --- /dev/null +++ b/ownlang/fix_gate.py @@ -0,0 +1,1089 @@ +"""S2 step 9 — the structural self-gate over a canonical step 8 bundle. + + python -m ownlang own-fix subscriptions gate \ + --bundle --plan \ + --candidates --root --out + +Step 9 proves not that the patch came from OUR generator, but that it is structurally +admissible and that its SEMANTICS hold when an INDEPENDENT Git applies it to a pristine +preimage in a hermetic throwaway repository. It never touches the real checkout, index or +config, never applies to a real tree, runs no model, no o7, no analyzer, no target tests. + +Trust: Step 9 trusts NOTHING it is handed. The manifest is a claim to be re-derived from +the plan (actions, in candidate order) and the hash-bound candidates (identity); the +patch, postimage and preimage are re-hashed over bytes read exactly once; `git` from the +host PATH is the independent applier. The six external byte inputs — plan, candidates, +manifest, patch, postimage, pristine source — are each read through ONE snapshot boundary +(reject symlink/reparse, open O_NOFOLLOW where available, fstat the handle, require a +regular file, read once) and every hash / parse / materialization runs over those memory +bytes. There is no second read, so there is no TOCTOU between checking and using a file. + +Concurrent mutation of the input directories by another privileged process is NOT in the +threat model; a pre-planted malicious filesystem entry or tampered bytes IS. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import stat +import subprocess +from typing import Any + +from ownlang.fix_candidates import ( + CollectError, + _require_canonical_relpath, + _require_finding_id, + _require_sha256, +) + +# --- failure taxonomy (the stable branch markers regressions assert on) --- +BUNDLE_LAYOUT = "BUNDLE_LAYOUT" +MANIFEST_SHAPE = "MANIFEST_SHAPE" +AUTHORITY_BINDING = "AUTHORITY_BINDING" +HASH_MISMATCH = "HASH_MISMATCH" +PRISTINE_SOURCE = "PRISTINE_SOURCE" +PATCH_STRUCTURE = "PATCH_STRUCTURE" +APPLY_CHECK = "APPLY_CHECK" +APPLY_MISMATCH = "APPLY_MISMATCH" +ISOLATION = "ISOLATION" +PUBLICATION = "PUBLICATION" +INFRASTRUCTURE = "INFRASTRUCTURE" + +_GATE_NAMES = ( + "bundle_layout", "manifest_shape", "authority_binding", "artifact_hashes", + "pristine_preimage", "patch_structure", "git_apply_check", "git_apply", + "postimage_equality", "isolated_tree", +) + +_ACTIONS = ("convert_acquire", "manual_review") +_CONTRACTS = ("inotify_property_changed", "name_only", "other", "unresolved") +# The frozen S0 span shape: absolute (start,length) + 1-based line/column, all ints. +_SPAN_KEYS = ("start", "length", "start_line", "start_column", "end_line", "end_column") +# The frozen S0 candidate shape — enforced exactly so no unknown field can ride a +# self-consistent, re-hashed bundle past the gate (amendment 2: known fields only). +_CANDIDATE_KEYS = ( + "finding_id", "diagnostic_code", "containing_type", "file", "enclosing_member", + "event", "event_identity", "event_contract", "source", "source_identity", + "source_identity_kind", "handler", "handler_identity", "handler_identity_kind", + "occurrence_ordinal", "acquire_span", "teardown", "allowed_actions", +) + + +class GateError(Exception): + """A controlled refusal, carrying the stable category for regression assertions.""" + + def __init__(self, category: str, message: str) -> None: + super().__init__(message) + self.category = category + + +# --- bytes / hashing / canonical JSON (independent of step 8) ---------------------- + + +def _sha_bytes(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def _canonical_json(obj: Any) -> bytes: + """Canonical JSON bytes — the shared serialization. The FILE artifacts (manifest, + evidence) add a trailing newline (see `_canonical_bytes`); the bundle HASH does not.""" + return json.dumps( + obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + + +def _canonical_bytes(obj: Any) -> bytes: + return _canonical_json(obj) + b"\n" + + +def _norm(path: str) -> str: + return os.path.normcase(os.path.normpath(path)) + + +def _same_path(a: str, b: str) -> bool: + """Platform-aware path equality — the SAME rule as _same_or_inside (normcase is + identity on POSIX, case-fold on Windows), so `C:\\R\\x` and `c:\\r\\x` compare equal on + Windows. A raw string `==` would give a casing-only false refusal.""" + return _norm(a) == _norm(b) + + +def _same_or_inside(parent: str, path: str) -> bool: + """Is `path` the directory `parent` itself, or under it? Both must be physical. + Case sensitivity is a PLATFORM property (normcase is identity on POSIX), so `C:\\Repo` + and `c:\\repo` are one directory on Windows and two elsewhere.""" + p = _norm(parent) + c = _norm(path) + if c == p: + return True + return c.startswith(p if p.endswith(os.sep) else p + os.sep) + + +# --- the one snapshot boundary (amendment 3) --------------------------------------- + +_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) +_O_BINARY = getattr(os, "O_BINARY", 0) +_REPARSE = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0) + + +def _is_link(st: os.stat_result) -> bool: + if stat.S_ISLNK(st.st_mode): + return True + # Windows junctions / reparse points are not S_ISLNK. + return bool(getattr(st, "st_file_attributes", 0) & _REPARSE) + + +def _snapshot(path: str, category: str, what: str) -> bytes: + """Read a regular file's bytes exactly once. Reject a symlink/reparse point, open + O_NOFOLLOW where the platform has it, fstat the OPEN handle, require a regular file, + then read. Every later operation runs over the returned bytes, never the file.""" + try: + lst = os.lstat(path) + except OSError as exc: + raise GateError(category, f"{what}: cannot stat ({exc.strerror or exc})") from exc + if _is_link(lst): + raise GateError(category, f"{what}: is a symlink / reparse point") + try: + fd = os.open(path, os.O_RDONLY | _O_NOFOLLOW | _O_BINARY) + except OSError as exc: + raise GateError(category, f"{what}: cannot open ({exc.strerror or exc})") from exc + try: + st = os.fstat(fd) + if not stat.S_ISREG(st.st_mode): + raise GateError(category, f"{what}: is not a regular file") + chunks = [] + while True: + chunk = os.read(fd, 1 << 20) + if not chunk: + break + chunks.append(chunk) + except OSError as exc: + raise GateError(category, f"{what}: cannot read ({exc.strerror or exc})") from exc + finally: + os.close(fd) + return b"".join(chunks) + + +def _load_json(data: bytes, category: str, what: str) -> Any: + try: + return json.loads(data) + except ValueError as exc: + raise GateError(category, f"{what}: not valid JSON ({exc})") from exc + + +# --- typed accessors, category-tagged ---------------------------------------------- + + +def _obj(v: Any, cat: str, where: str) -> dict[str, Any]: + if not isinstance(v, dict): + raise GateError(cat, f"{where}: must be an object") + return v + + +def _need(obj: dict[str, Any], name: str, cat: str, where: str) -> Any: + if name not in obj: + raise GateError(cat, f"{where}: missing '{name}'") + return obj[name] + + +def _s(obj: dict[str, Any], name: str, cat: str, where: str) -> str: + v = _need(obj, name, cat, where) + if not isinstance(v, str): + raise GateError(cat, f"{where}.{name}: must be a string") + return v + + +def _int(obj: dict[str, Any], name: str, cat: str, where: str) -> int: + v = _need(obj, name, cat, where) + if not isinstance(v, int) or isinstance(v, bool) or v < 0: + raise GateError(cat, f"{where}.{name}: must be a non-negative int") + return v + + +def _list(obj: dict[str, Any], name: str, cat: str, where: str) -> list[Any]: + v = _need(obj, name, cat, where) + if not isinstance(v, list): + raise GateError(cat, f"{where}.{name}: must be an array") + return v + + +def _exact(obj: dict[str, Any], cat: str, where: str, *keys: str) -> None: + have = set(obj) + want = set(keys) + if have != want: + raise GateError(cat, f"{where}: key set {sorted(have)} != {sorted(want)}") + + +def _sha(obj: dict[str, Any], name: str, cat: str, where: str) -> str: + v = _s(obj, name, cat, where) + try: + _require_sha256(v, f"{where}.{name}") + except CollectError as exc: + raise GateError(cat, str(exc)) from exc + return v + + +def _relpath(v: str, cat: str, where: str) -> None: + try: + _require_canonical_relpath(v, where) + except CollectError as exc: + raise GateError(cat, str(exc)) from exc + + +# --- the strict authority validator (amendments 1 & 2; no filesystem I/O) ----------- + + +class GateAuthority: + """The verified authority context, derived ONLY from plan + candidates (no I/O). rel + is the canonical source path; applied/manual are the finding ids in candidate order; + pre_sha256 / target_subscribe / input_bundle_sha256 come from the hash-bound bundle.""" + + __slots__ = ("applied", "input_bundle_sha256", "manual", "pre_sha256", "rel", + "selected_findings", "target_subscribe") + + def __init__(self, rel: str, target_subscribe: str, pre_sha256: str, + input_bundle_sha256: str, applied: list[str], manual: list[str], + selected_findings: list[str] | None) -> None: + self.rel = rel + self.target_subscribe = target_subscribe + self.pre_sha256 = pre_sha256 + self.input_bundle_sha256 = input_bundle_sha256 + self.applied = applied + self.manual = manual + self.selected_findings = selected_findings + + +def _bundle_sha256(candidates: dict[str, Any]) -> str: + # The step 8 input-bundle hash is over canonical JSON with NO trailing newline. + return _sha_bytes(_canonical_json(candidates)) + + +def _span(obj: dict[str, Any], name: str, cat: str, where: str) -> tuple[int, ...]: + """The frozen 6-key int span. Returns the values as a tuple for exact comparison.""" + sp = _obj(_need(obj, name, cat, where), cat, f"{where}.{name}") + _exact(sp, cat, f"{where}.{name}", *_SPAN_KEYS) + return tuple(_int(sp, k, cat, f"{where}.{name}") for k in _SPAN_KEYS) + + +def _validate_teardown(td: Any, cat: str, where: str) -> None: + """The frozen teardown block: status vocabulary + exact-key candidates with a span.""" + t = _obj(td, cat, f"{where}.teardown") + _exact(t, cat, f"{where}.teardown", "status", "candidates") + if _s(t, "status", cat, f"{where}.teardown") not in ("none", "exact", "ambiguous"): + raise GateError(cat, f"{where}.teardown: unknown status") + for i, tc_any in enumerate(_list(t, "candidates", cat, f"{where}.teardown")): + tctx = f"{where}.teardown.candidates[{i}]" + tc = _obj(tc_any, cat, tctx) + _exact(tc, cat, tctx, "source", "handler", "match", "span") + _s(tc, "source", cat, tctx) + _s(tc, "handler", cat, tctx) + _s(tc, "match", cat, tctx) + _span(tc, "span", cat, tctx) + + +def _check_constraints(cons: dict[str, Any], cat: str, where: str) -> None: + _exact(cons, cat, where, "max_types_changed", "max_files_changed", + "allow_helper_changes", "allow_config_changes", "allow_suppressions") + if (_int(cons, "max_types_changed", cat, where) != 1 + or _int(cons, "max_files_changed", cat, where) != 1): + raise GateError(cat, f"{where}: changes exactly one type in one file") + for k in ("allow_helper_changes", "allow_config_changes", "allow_suppressions"): + if cons.get(k) is not False: + raise GateError(cat, f"{where}.{k} must be false") + + +def _validate_candidates(bundle: dict[str, Any], cat: str) -> dict[str, Any]: + """The frozen candidates envelope, returning the derived facts the plan is checked + against. Exact key sets on every object; the permission tiering; one type / one file.""" + _exact(bundle, cat, "candidates", "version", "operation", "target_api", "selection", + "source_files", "candidates") + if bundle["version"] != 1 or isinstance(bundle["version"], bool): + raise GateError(cat, "candidates.version must be 1") + if bundle["operation"] != "fix-subscriptions": + raise GateError(cat, "candidates.operation must be 'fix-subscriptions'") + b_target = _obj(bundle["target_api"], cat, "candidates.target_api") + _exact(b_target, cat, "candidates.target_api", "subscribe") + target_subscribe = _s(b_target, "subscribe", cat, "candidates.target_api") + + b_sel = _obj(bundle["selection"], cat, "candidates.selection") + _exact(b_sel, cat, "candidates.selection", "allowed_types", "selected_findings", + "constraints") + b_types = _list(b_sel, "allowed_types", cat, "candidates.selection") + if len(b_types) != 1: + raise GateError(cat, "candidates.selection.allowed_types needs exactly one entry") + b_type = _obj(b_types[0], cat, "candidates.selection.allowed_types[0]") + _exact(b_type, cat, "candidates.selection.allowed_types[0]", "full_name", "file") + type_name = _s(b_type, "full_name", cat, "candidates.selection.allowed_types[0]") + type_file = _s(b_type, "file", cat, "candidates.selection.allowed_types[0]") + _relpath(type_file, cat, "candidates.selection.allowed_types[0].file") + _check_constraints(_obj(b_sel["constraints"], cat, "candidates.selection.constraints"), + cat, "candidates.selection.constraints") + + b_files = _list(bundle, "source_files", cat, "candidates") + if len(b_files) != 1: + raise GateError(cat, "candidates.source_files needs exactly one entry") + b_src = _obj(b_files[0], cat, "candidates.source_files[0]") + _exact(b_src, cat, "candidates.source_files[0]", "path", "sha256") + src_path = _s(b_src, "path", cat, "candidates.source_files[0]") + _relpath(src_path, cat, "candidates.source_files[0].path") + pre_sha256 = _sha(b_src, "sha256", cat, "candidates.source_files[0]") + if type_file != src_path: + raise GateError(cat, f"selected type file '{type_file}' != source '{src_path}'") + + cands = _list(bundle, "candidates", cat, "candidates") + if not cands: + raise GateError(cat, "candidates: the bundle is empty") + ids: list[str] = [] + id_set: set[str] = set() + allowed_by_id: dict[str, list[str]] = {} + span_by_id: dict[str, tuple[int, ...]] = {} + for index, c_any in enumerate(cands): + where = f"candidates[{index}]" + c = _obj(c_any, cat, where) + _exact(c, cat, where, *_CANDIDATE_KEYS) + fid = _s(c, "finding_id", cat, where) + try: + _require_finding_id(fid, where) + except CollectError as exc: + raise GateError(cat, str(exc)) from exc + if fid in id_set: + raise GateError(cat, f"{where}: duplicate finding_id {fid}") + id_set.add(fid) + ids.append(fid) + # EVERY string field the frozen S0 candidate carries — not just the identity + # subset — so a wrong-typed field on a re-hashed bundle cannot ride through. + for k in ("diagnostic_code", "enclosing_member", "event", "event_identity", + "source", "source_identity", "source_identity_kind", + "handler", "handler_identity", "handler_identity_kind"): + _s(c, k, cat, where) + _int(c, "occurrence_ordinal", cat, where) + _validate_teardown(c["teardown"], cat, where) + if _s(c, "containing_type", cat, where) != type_name: + raise GateError(cat, f"{where}: outside the selected type {type_name}") + if _s(c, "file", cat, where) != src_path: + raise GateError(cat, f"{where}: outside the selected file {src_path}") + contract = _s(c, "event_contract", cat, where) + if contract not in _CONTRACTS: + raise GateError(cat, f"{where}: unknown event_contract '{contract}'") + span_by_id[fid] = _span(c, "acquire_span", cat, where) + actions = _list(c, "allowed_actions", cat, where) + if not actions: + raise GateError(cat, f"{where}: allowed_actions must be non-empty") + seen: set[str] = set() + for a in actions: + if not isinstance(a, str) or a not in _ACTIONS: + raise GateError(cat, f"{where}: unknown action in allowed_actions") + if a in seen: + raise GateError(cat, f"{where}: duplicate action in allowed_actions") + seen.add(a) + if "convert_acquire" in seen and contract != "inotify_property_changed": + raise GateError(cat, f"{where}: convert_acquire not permitted for '{contract}'") + allowed_by_id[fid] = list(actions) + + selected = b_sel["selected_findings"] + if selected is not None: + if not isinstance(selected, list) or not all(isinstance(x, str) for x in selected): + raise GateError(cat, "selection.selected_findings must be null or a string array") + if len(set(selected)) != len(selected) or set(selected) != id_set: + raise GateError(cat, "selection.selected_findings does not name the candidates") + + return { + "target_subscribe": target_subscribe, "type_name": type_name, + "type_file": type_file, "src_path": src_path, "pre_sha256": pre_sha256, + "ids": ids, "allowed_by_id": allowed_by_id, "span_by_id": span_by_id, + "selected": selected, + } + + +def validate_gate_authority(validated_plan: Any, candidates: Any) -> GateAuthority: + """Restate the FROZEN envelope the rewriter holds — a hash proves only that the two + files agree with each other, never that they obey policy. Pure: no filesystem I/O.""" + cat = AUTHORITY_BINDING + plan = _obj(validated_plan, cat, "plan") + bundle = _obj(candidates, cat, "candidates") + facts = _validate_candidates(bundle, cat) + src_path = facts["src_path"] + ids = facts["ids"] + + _exact(plan, cat, "plan", "version", "operation", "input_bundle_sha256", "target_api", + "selection", "source_files", "decisions") + if plan["version"] != 1 or isinstance(plan["version"], bool): + raise GateError(cat, "plan.version must be 1") + if plan["operation"] != "fix-subscriptions": + raise GateError(cat, "plan.operation must be 'fix-subscriptions'") + input_bundle_sha256 = _sha(plan, "input_bundle_sha256", cat, "plan") + if input_bundle_sha256 != _bundle_sha256(bundle): + raise GateError(cat, "plan.input_bundle_sha256 does not bind these candidates") + + p_target = _obj(plan["target_api"], cat, "plan.target_api") + _exact(p_target, cat, "plan.target_api", "subscribe") + if _s(p_target, "subscribe", cat, "plan.target_api") != facts["target_subscribe"]: + raise GateError(cat, "plan.target_api.subscribe != the candidates bundle") + + p_sel = _obj(plan["selection"], cat, "plan.selection") + _exact(p_sel, cat, "plan.selection", "allowed_types", "selected_findings", "constraints") + p_types = _list(p_sel, "allowed_types", cat, "plan.selection") + if len(p_types) != 1: + raise GateError(cat, "plan.selection.allowed_types needs exactly one entry") + p_type = _obj(p_types[0], cat, "plan.selection.allowed_types[0]") + _exact(p_type, cat, "plan.selection.allowed_types[0]", "full_name", "file") + if (_s(p_type, "full_name", cat, "plan.selection.allowed_types[0]") != facts["type_name"] + or _s(p_type, "file", cat, "plan.selection.allowed_types[0]") != facts["type_file"]): + raise GateError(cat, "plan.selection.allowed_types != the candidates bundle") + _check_constraints(_obj(p_sel["constraints"], cat, "plan.selection.constraints"), + cat, "plan.selection.constraints") + if p_sel["selected_findings"] != facts["selected"]: + raise GateError(cat, "plan.selection.selected_findings != the candidates bundle") + + p_files = _list(plan, "source_files", cat, "plan") + if len(p_files) != 1: + raise GateError(cat, "plan.source_files needs exactly one entry") + p_src = _obj(p_files[0], cat, "plan.source_files[0]") + _exact(p_src, cat, "plan.source_files[0]", "path", "sha256") + if _s(p_src, "path", cat, "plan.source_files[0]") != src_path: + raise GateError(cat, "plan.source_files[0].path != the candidates bundle") + if _s(p_src, "sha256", cat, "plan.source_files[0]") != facts["pre_sha256"]: + raise GateError(cat, "plan.source_files[0].sha256 != the candidates bundle") + + decisions = _list(plan, "decisions", cat, "plan") + if len(decisions) != len(ids): + raise GateError(cat, f"plan.decisions covers {len(decisions)} of {len(ids)}") + applied: list[str] = [] + manual: list[str] = [] + for index, d_any in enumerate(decisions): + where = f"plan.decisions[{index}]" + d = _obj(d_any, cat, where) + _exact(d, cat, where, "finding_id", "action", "file", "acquire_span") + fid = _s(d, "finding_id", cat, where) + if fid != ids[index]: + raise GateError(cat, f"{where}: {fid} out of candidate order (want {ids[index]})") + action = _s(d, "action", cat, where) + if action not in _ACTIONS: + raise GateError(cat, f"{where}: out-of-scope action '{action}'") + if action not in facts["allowed_by_id"][fid]: + raise GateError(cat, f"{where}: action '{action}' not allowed for {fid}") + if _s(d, "file", cat, where) != src_path: + raise GateError(cat, f"{where}: file != the selected source file") + if _span(d, "acquire_span", cat, where) != facts["span_by_id"][fid]: + raise GateError(cat, f"{where}: acquire_span != the candidate") + (applied if action == "convert_acquire" else manual).append(fid) + + return GateAuthority(src_path, facts["target_subscribe"], facts["pre_sha256"], + input_bundle_sha256, applied, manual, facts["selected"]) + + +# --- manifest shape (its own gate + category) -------------------------------------- + + +def _finding_list(m: dict[str, Any], name: str, cat: str) -> list[str]: + values = _list(m, name, cat, "manifest") + seen: set[str] = set() + for v in values: + if not isinstance(v, str): + raise GateError(cat, f"manifest.{name}: must be a list of strings") + try: + _require_finding_id(v, f"manifest.{name}") + except CollectError as exc: + raise GateError(cat, str(exc)) from exc + if v in seen: + raise GateError(cat, f"manifest.{name}: duplicate finding_id {v}") + seen.add(v) + return values + + +def validate_manifest_shape(manifest: Any) -> tuple[str, str, str, str]: + """Exact top-level shape of apply-manifest.json, independent of the plan. Returns + (rel, pre_sha256, post_sha256, patch_sha256).""" + cat = MANIFEST_SHAPE + m = _obj(manifest, cat, "manifest") + _exact(m, cat, "manifest", "version", "operation", "input_bundle_sha256", + "validated_plan_sha256", "target_api", "source_files", "applied_findings", + "manual_review_findings", "patch_sha256") + if m["version"] != 1 or isinstance(m["version"], bool): + raise GateError(cat, "manifest.version must be 1") + if m["operation"] != "apply-subscription-fixes": + raise GateError(cat, "manifest.operation must be 'apply-subscription-fixes'") + _sha(m, "input_bundle_sha256", cat, "manifest") + _sha(m, "validated_plan_sha256", cat, "manifest") + patch_sha = _sha(m, "patch_sha256", cat, "manifest") + t = _obj(m["target_api"], cat, "manifest.target_api") + _exact(t, cat, "manifest.target_api", "subscribe") + _s(t, "subscribe", cat, "manifest.target_api") + files = _list(m, "source_files", cat, "manifest") + if len(files) != 1: + raise GateError(cat, "manifest.source_files needs exactly one entry") + src = _obj(files[0], cat, "manifest.source_files[0]") + _exact(src, cat, "manifest.source_files[0]", "path", "pre_sha256", "post_sha256") + rel = _s(src, "path", cat, "manifest.source_files[0]") + _relpath(rel, cat, "manifest.source_files[0].path") + pre_sha = _sha(src, "pre_sha256", cat, "manifest.source_files[0]") + post_sha = _sha(src, "post_sha256", cat, "manifest.source_files[0]") + applied = _finding_list(m, "applied_findings", cat) + manual = _finding_list(m, "manual_review_findings", cat) + if set(applied) & set(manual): + raise GateError(cat, "manifest: applied and manual_review findings overlap") + return rel, pre_sha, post_sha, patch_sha + + +# --- the strict step 8 patch language (amendment 7) -------------------------------- + + +def _records(data: bytes) -> list[bytes]: + """LF-terminated records; a CR is ordinary content. Every record MUST end in LF — + step 8's canonical_patch emits only LF-terminated records.""" + recs: list[bytes] = [] + i = 0 + n = len(data) + while i < n: + j = data.find(b"\n", i) + if j < 0: + raise GateError(PATCH_STRUCTURE, "patch has an unterminated final line") + recs.append(data[i:j]) + i = j + 1 + return recs + + +def _posint(text: bytes) -> int: + if not text or not text.isdigit(): + raise GateError(PATCH_STRUCTURE, "malformed hunk number") + return int(text) + + +def _range(text: bytes) -> tuple[int, int]: + parts = text.split(b",") + if len(parts) == 1: + return _posint(parts[0]), 1 + if len(parts) == 2: + return _posint(parts[0]), _posint(parts[1]) + raise GateError(PATCH_STRUCTURE, "malformed hunk range") + + +def parse_step8_patch(patch: bytes, rel: str, preimage: bytes) -> None: + """Refuse anything outside the frozen step 8 grammar: one file header whose three + paths are BYTE-EQUAL to `rel` (so a quoted/escaped or alternate path will not match), + hunks whose arithmetic is consistent and whose old ranges lie within the preimage, + context/`-`/`+` body lines and the no-newline marker — and NOTHING else.""" + if patch == b"": + return # the empty patch; the caller checks applied_findings == [] + rb = rel.encode("utf-8") + recs = _records(patch) + if len(recs) < 4: + raise GateError(PATCH_STRUCTURE, "patch is too short for a single-file diff") + if recs[0] != b"diff --git a/" + rb + b" b/" + rb: + raise GateError(PATCH_STRUCTURE, "patch: 'diff --git' header is not the allowed path") + if recs[1] != b"--- a/" + rb: + raise GateError(PATCH_STRUCTURE, "patch: '---' header is not the allowed path") + if recs[2] != b"+++ b/" + rb: + raise GateError(PATCH_STRUCTURE, "patch: '+++' header is not the allowed path") + + pre_lines = preimage.count(b"\n") + if preimage and not preimage.endswith(b"\n"): + pre_lines += 1 + i = 3 + prev_old_end = 0 + saw_hunk = False + saw_marker = False # FILE-level: a no-newline marker means EOF-without-newline, so no + while i < len(recs): # further hunk may follow it — the marked line is the file's last. + rec = recs[i] + if not (rec.startswith(b"@@ -") and rec.endswith(b" @@")): + raise GateError(PATCH_STRUCTURE, f"patch: expected a hunk header, got {rec[:40]!r}") + if saw_marker: + raise GateError(PATCH_STRUCTURE, "patch: a hunk after a no-newline-at-EOF marker") + body = rec[len(b"@@ -"):-len(b" @@")] + try: + old_part, new_part = body.split(b" +", 1) + except ValueError as exc: + raise GateError(PATCH_STRUCTURE, "patch: malformed hunk header") from exc + old_start, old_len = _range(old_part) + _new_start, new_len = _range(new_part) + # Range bounds, including the zero-length (pure-insertion) case: a 0-length old + # range names the line BEFORE it (0..pre_lines); a non-zero range is 1-based and + # must lie within the preimage. + if old_len > 0: + if old_start < 1 or old_start + old_len - 1 > pre_lines: + raise GateError(PATCH_STRUCTURE, "patch: a hunk range is outside the preimage") + elif old_start > pre_lines: + raise GateError(PATCH_STRUCTURE, "patch: an insertion range is past the preimage") + if old_start < prev_old_end: + raise GateError(PATCH_STRUCTURE, "patch: hunks not increasing / non-overlapping") + prev_old_end = old_start + old_len + saw_hunk = True + i += 1 + ctx = minus = plus = 0 + old_closed = new_closed = False # a no-newline marker closes a side; once each + last_head: bytes | None = None # the head of the last body line (None after a marker) + while i < len(recs) and not (recs[i].startswith(b"@@ -") + and recs[i].endswith(b" @@")): + line = recs[i] + if line == b"\\ No newline at end of file": + # Must sit immediately after an eligible body line, mark the LAST line of + # its side(s), and appear at most once per side. A context line is on both + # sides; a `-` line only old; a `+` line only new. + if last_head is None: + raise GateError(PATCH_STRUCTURE, "patch: no-newline marker not after a line") + marks_old = last_head in (b" ", b"-") + marks_new = last_head in (b" ", b"+") + if (marks_old and old_closed) or (marks_new and new_closed): + raise GateError(PATCH_STRUCTURE, "patch: duplicate no-newline marker") + old_closed = old_closed or marks_old + new_closed = new_closed or marks_new + saw_marker = True + last_head = None + i += 1 + continue + if not line or line[:1] not in (b" ", b"-", b"+"): + raise GateError(PATCH_STRUCTURE, f"patch: illegal hunk line {line[:40]!r}") + head = line[:1] + # A line of a side already closed by a marker means the marker did not mark the + # LAST line of that side. + if head in (b" ", b"-") and old_closed: + raise GateError(PATCH_STRUCTURE, "patch: old-side line after a no-newline marker") + if head in (b" ", b"+") and new_closed: + raise GateError(PATCH_STRUCTURE, "patch: new-side line after a no-newline marker") + if head == b" ": + ctx += 1 + elif head == b"-": + minus += 1 + else: + plus += 1 + last_head = head + i += 1 + if ctx + minus != old_len or ctx + plus != new_len: + raise GateError(PATCH_STRUCTURE, "patch: hunk line counts disagree with header") + if minus == 0 and plus == 0: + raise GateError(PATCH_STRUCTURE, "patch: a hunk changes nothing") + if not saw_hunk: + raise GateError(PATCH_STRUCTURE, "patch: the file header carries no hunk") + + +# --- the pristine source verifier (platform-aware physical) ------------------------- + + +def verify_pristine(root: str, rel: str, expected_pre_sha: str) -> bytes: + """Find `rel` under the physical `root`, confined platform-aware and symlink-aware, + read its bytes ONCE, and require sha == the manifest's pre_sha256. os.path.realpath + resolves every intermediate symlink physically, so an intermediate-symlink escape is + caught by confinement; a symlink AT the leaf is refused outright. The isolated tree is + later built from THESE bytes — no second read, so no TOCTOU.""" + cat = PRISTINE_SOURCE + try: + root_phys = os.path.realpath(root) + except OSError as exc: + raise GateError(cat, f"cannot resolve --root ({exc.strerror or exc})") from exc + if not os.path.isdir(root_phys): + raise GateError(cat, f"--root '{root}' is not a directory") + joined = os.path.join(root_phys, *rel.split("/")) + try: + leaf = os.lstat(joined) + except OSError as exc: + raise GateError(cat, f"source '{rel}' not found ({exc.strerror or exc})") from exc + if _is_link(leaf): + raise GateError(cat, f"source '{rel}' is a symlink / reparse point") + src_phys = os.path.realpath(joined) + if not _same_or_inside(root_phys, src_phys): + raise GateError(cat, f"source '{rel}' resolves outside the root") + data = _snapshot(src_phys, cat, f"source '{rel}'") + if _sha_bytes(data) != expected_pre_sha: + raise GateError(cat, f"STALE PREIMAGE / PRISTINE SOURCE MISMATCH for {rel}") + return data + + +# --- the hermetic throwaway repository (amendments 5 & 6) --------------------------- + + +def _git_env(home: str, xdg: str, empty_global: str) -> dict[str, str]: + """A minimal ALLOWLIST env — not the inherited environment with dangerous GIT_* keys + guessed at and stripped. Anything not listed here is simply absent from git's world.""" + env = { + "PATH": os.environ.get("PATH", os.defpath), + "LC_ALL": "C", + "LANG": "C", + "HOME": home, + "XDG_CONFIG_HOME": xdg, + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_CONFIG_GLOBAL": empty_global, + "GIT_ATTR_NOSYSTEM": "1", + "GIT_TERMINAL_PROMPT": "0", + } + if os.name == "nt": + for key in ("SystemRoot", "ComSpec", "PATHEXT", "TEMP", "TMP", + "SystemDrive", "windir"): + val = os.environ.get(key) + if val is not None: + env[key] = val + return env + + +def _git(args: list[str], cwd: str, env: dict[str, str], + stdin: bytes | None = None) -> tuple[int, bytes, bytes]: + cmd = ["git", "-c", "core.autocrlf=false", "-c", "core.eol=lf", + "-c", "core.safecrlf=false", *args] + try: + proc = subprocess.run(cmd, cwd=cwd, env=env, input=stdin, + capture_output=True, check=False) + except FileNotFoundError as exc: + raise GateError(INFRASTRUCTURE, "git is not available on PATH") from exc + except OSError as exc: + raise GateError(INFRASTRUCTURE, f"cannot run git ({exc.strerror or exc})") from exc + return proc.returncode, proc.stdout, proc.stderr + + +def _walk_worktree(repo: str) -> set[str]: + seen: set[str] = set() + for dirpath, dirnames, filenames in os.walk(repo): + if ".git" in dirnames: + dirnames.remove(".git") + for name in filenames: + full = os.path.join(dirpath, name) + seen.add(os.path.relpath(full, repo).replace("\\", "/")) + return seen + + +def apply_in_throwaway(workdir: str, rel: str, preimage: bytes, postimage: bytes, + patch: bytes, post_sha256: str) -> None: + """Build a throwaway repo containing ONLY the preimage (from bytes already read), give + it a baseline index via `git add`, apply the patch with a hermetic Git, and prove that + exactly `rel` changed to the exact postimage — index, config and the real checkout all + untouched. `git apply` runs without --index, so the index must stay byte-identical.""" + home = os.path.join(workdir, "home") + xdg = os.path.join(workdir, "xdg") + empty_global = os.path.join(workdir, "gitconfig-none") + template = os.path.join(workdir, "git-template") + repo = os.path.join(workdir, "pristine") + for d in (home, xdg, template, repo): + os.makedirs(d) + with open(empty_global, "wb"): + pass + env = _git_env(home, xdg, empty_global) + + rc, _o, err = _git(["init", "-q", f"--template={template}", "."], repo, env) + if rc != 0: + raise GateError(INFRASTRUCTURE, + "git init failed: " + err.decode("utf-8", "replace").strip()) + + target = os.path.join(repo, *rel.split("/")) + if not _same_or_inside(os.path.realpath(repo), + os.path.realpath(os.path.dirname(target))): + raise GateError(ISOLATION, "the preimage path escapes the throwaway repo") + os.makedirs(os.path.dirname(target), exist_ok=True) + with open(target, "wb") as fh: + fh.write(preimage) + + rc, _o, err = _git(["add", "--", rel], repo, env) + if rc != 0: + raise GateError(INFRASTRUCTURE, + "git add failed: " + err.decode("utf-8", "replace").strip()) + + index_path = os.path.join(repo, ".git", "index") + config_path = os.path.join(repo, ".git", "config") + with open(index_path, "rb") as fh: + index_before = fh.read() + with open(config_path, "rb") as fh: + config_before = fh.read() + + rc, _o, _e = _git(["apply", "--check", "--whitespace=nowarn", "-"], repo, env, stdin=patch) + if rc != 0: + raise GateError(APPLY_CHECK, "git apply --check refused the patch") + rc, _o, _e = _git(["apply", "--whitespace=nowarn", "-"], repo, env, stdin=patch) + if rc != 0: + raise GateError(APPLY_MISMATCH, "git apply failed after --check passed") + + rc, out, _e = _git(["diff-files", "--name-only", "-z"], repo, env) + if rc != 0 or out != rel.encode("utf-8") + b"\0": + raise GateError(ISOLATION, f"the tree does not show exactly '{rel}' modified") + rc, out, _e = _git(["ls-files", "--others", "-z"], repo, env) + if rc != 0 or out != b"": + raise GateError(ISOLATION, "an unexpected untracked file appeared") + + seen = _walk_worktree(repo) + if seen != {rel}: + raise GateError(ISOLATION, f"the worktree holds {sorted(seen)}, want ['{rel}']") + + with open(index_path, "rb") as fh: + if fh.read() != index_before: + raise GateError(ISOLATION, "git apply mutated the temporary index") + with open(config_path, "rb") as fh: + if fh.read() != config_before: + raise GateError(ISOLATION, "git apply mutated the temporary config") + + with open(target, "rb") as fh: + applied = fh.read() + if applied != postimage: + raise GateError(APPLY_MISMATCH, "the applied file != postimage/") + if _sha_bytes(applied) != post_sha256: + raise GateError(APPLY_MISMATCH, "applied file sha != manifest post_sha256") + + +# --- publication (the step 8 protocol, reused) ------------------------------------- + + +def _out_parent(out: str, root: str) -> tuple[str, str, str]: + """(out_phys, parent_phys, root_phys). The out-dir must be fresh and PHYSICALLY off the + source tree — its parent is resolved and confined here, and re-proven immediately + before the publishing rename.""" + out_abs = os.path.abspath(out) + name = os.path.basename(out_abs.rstrip(os.sep)) + if not name: + raise GateError(PUBLICATION, f"--out {out!r}: not a directory name") + parent = os.path.dirname(out_abs.rstrip(os.sep)) + if not os.path.isdir(parent): + raise GateError(PUBLICATION, f"--out {out!r}: parent directory does not exist") + parent_phys = os.path.realpath(parent) + root_phys = os.path.realpath(root) + if _same_or_inside(root_phys, parent_phys): + raise GateError(PUBLICATION, f"--out {out!r} resolves inside the source root") + out_phys = os.path.join(parent_phys, name) + if os.path.exists(out_phys) or os.path.islink(out_phys): + raise GateError(PUBLICATION, f"--out {out!r} already exists") + return out_phys, parent_phys, root_phys + + +def _claim_workdir(parent_phys: str) -> str: + """CLAIM an unpredictable working directory: create it here and now (owner-only on + POSIX, as part of the mkdir), then PROVE we own it — not a link/reparse, resolving to + itself under a PLATFORM-AWARE comparison, and empty. A name that is merely checked and + then written into is a window; this closes it. If any proof AFTER the mkdir fails, the + directory we just created is removed before raising — no leftover.""" + import shutil + + for _ in range(8): + path = os.path.join(parent_phys, f".owen-gate-{os.urandom(16).hex()}") + if os.path.exists(path) or os.path.islink(path): + continue + try: + os.mkdir(path, mode=0o700) + except FileExistsError: + continue + except OSError as exc: + raise GateError(PUBLICATION, + f"cannot claim a work directory ({exc.strerror or exc})") from exc + try: + lst = os.lstat(path) + if _is_link(lst): + raise GateError(PUBLICATION, "the claimed work directory is a link") + if not stat.S_ISDIR(lst.st_mode) or not _same_path(os.path.realpath(path), path) \ + or not _same_or_inside(parent_phys, os.path.realpath(path)): + raise GateError(PUBLICATION, + "the claimed work directory does not resolve to itself") + if any(os.scandir(path)): + raise GateError(PUBLICATION, "the claimed work directory is not empty") + except BaseException: + shutil.rmtree(path, ignore_errors=True) + raise + return path + raise GateError(PUBLICATION, "could not claim a work directory") + + +def _publish(staging: str, out_phys: str, evidence: bytes, root_phys: str) -> None: + os.makedirs(staging) + with open(os.path.join(staging, "gate-result.json"), "wb") as fh: + fh.write(evidence) + # Re-prove the destination against the filesystem AS IT IS NOW, right before the rename. + parent = os.path.dirname(out_phys) + if not os.path.isdir(parent): + raise GateError(PUBLICATION, "the out-dir parent vanished before publication") + if not _same_path(os.path.realpath(parent), os.path.dirname(out_phys)) \ + or _same_or_inside(root_phys, os.path.realpath(parent)): + raise GateError(PUBLICATION, "the out-dir parent changed to resolve inside the root") + if os.path.exists(out_phys) or os.path.islink(out_phys): + raise GateError(PUBLICATION, "the out-dir appeared before publication") + try: + os.rename(staging, out_phys) + except OSError as exc: + raise GateError(PUBLICATION, + f"cannot publish evidence ({exc.strerror or exc})") from exc + + +# --- bundle-layout helpers --------------------------------------------------------- + + +def _lentry(path: str, what: str) -> os.stat_result: + try: + return os.lstat(path) + except OSError as exc: + raise GateError(BUNDLE_LAYOUT, f"{what}: cannot stat ({exc.strerror or exc})") from exc + + +def _require_top_level(bundle: str) -> None: + try: + names = set(os.listdir(bundle)) + except OSError as exc: + raise GateError(BUNDLE_LAYOUT, f"cannot list --bundle ({exc.strerror or exc})") from exc + if names != {"change.patch", "apply-manifest.json", "postimage"}: + raise GateError(BUNDLE_LAYOUT, f"bundle holds {sorted(names)}, want " + "['apply-manifest.json', 'change.patch', 'postimage']") + for f in ("change.patch", "apply-manifest.json"): + st = _lentry(os.path.join(bundle, f), f) + if _is_link(st): + raise GateError(BUNDLE_LAYOUT, f"{f}: is a symlink / reparse point") + if not stat.S_ISREG(st.st_mode): + raise GateError(BUNDLE_LAYOUT, f"{f}: is not a regular file") + pst = _lentry(os.path.join(bundle, "postimage"), "postimage") + if _is_link(pst) or not stat.S_ISDIR(pst.st_mode): + raise GateError(BUNDLE_LAYOUT, "postimage: is not a real directory") + + +def _walk_tree(root: str, what: str) -> tuple[set[str], set[str]]: + """Every entry under `root` must be a real directory or a regular file — no symlinks, + reparse points, fifos, sockets or devices. Returns (dir paths, file paths), both + `/`-joined and relative to `root`, so the caller can require an EXACT layout (extra or + hidden empty directories are a violation, not just extra files).""" + dirs: set[str] = set() + files: set[str] = set() + stack = [root] + while stack: + current = stack.pop() + try: + entries = list(os.scandir(current)) + except OSError as exc: + raise GateError(BUNDLE_LAYOUT, f"{what}: cannot scan ({exc.strerror or exc})") from exc + for entry in entries: + st = _lentry(entry.path, entry.path) + rel = os.path.relpath(entry.path, root).replace("\\", "/") + if _is_link(st): + raise GateError(BUNDLE_LAYOUT, f"{what}: '{entry.name}' is a symlink/reparse") + if stat.S_ISDIR(st.st_mode): + dirs.add(rel) + stack.append(entry.path) + elif stat.S_ISREG(st.st_mode): + files.add(rel) + else: + raise GateError(BUNDLE_LAYOUT, + f"{what}: '{entry.name}' is not a regular file or dir") + return dirs, files + + +# --- evidence + orchestration ------------------------------------------------------ + + +def _build_evidence(auth: GateAuthority, rel: str, plan_bytes: bytes, + manifest_bytes: bytes, patch_bytes: bytes, pre_sha: str, + post_sha: str, git_gates: str) -> bytes: + gates = dict.fromkeys(_GATE_NAMES, "pass") + gates["git_apply_check"] = git_gates + gates["git_apply"] = git_gates + gates["isolated_tree"] = git_gates + evidence = { + "version": 1, + "operation": "gate-subscription-fix-bundle", + "input_bundle_sha256": auth.input_bundle_sha256, + "validated_plan_sha256": _sha_bytes(plan_bytes), + "apply_manifest_sha256": _sha_bytes(manifest_bytes), + "patch_sha256": _sha_bytes(patch_bytes), + "target_api": {"subscribe": auth.target_subscribe}, + "source_files": [{"path": rel, "pre_sha256": pre_sha, "post_sha256": post_sha}], + "applied_findings": auth.applied, + "manual_review_findings": auth.manual, + "gates": gates, + } + return _canonical_bytes(evidence) + + +def run_gate(bundle: str, plan_path: str, candidates_path: str, root: str, out: str) -> str: + """Gate a canonical step 8 bundle and publish gate-result.json. Returns the published + path; raises GateError (no output, source untouched) on any refusal.""" + import shutil + + # [1] bundle layout — the bundle root itself must not be a symlink/reparse (lstat + # BEFORE realpath), then exact entry types + the full postimage subtree. + blst = _lentry(bundle, "--bundle") + if _is_link(blst): + raise GateError(BUNDLE_LAYOUT, "--bundle is a symlink / reparse point") + if not stat.S_ISDIR(blst.st_mode): + raise GateError(BUNDLE_LAYOUT, "--bundle is not a directory") + bundle_phys = os.path.realpath(bundle) + _require_top_level(bundle_phys) + postimage_root = os.path.join(bundle_phys, "postimage") + post_dirs, post_leaves = _walk_tree(postimage_root, "postimage") + + # [2] snapshot the inputs that do NOT need rel (each read exactly once). + manifest_bytes = _snapshot(os.path.join(bundle_phys, "apply-manifest.json"), + BUNDLE_LAYOUT, "apply-manifest.json") + patch_bytes = _snapshot(os.path.join(bundle_phys, "change.patch"), + BUNDLE_LAYOUT, "change.patch") + plan_bytes = _snapshot(plan_path, AUTHORITY_BINDING, "--plan") + cand_bytes = _snapshot(candidates_path, AUTHORITY_BINDING, "--candidates") + + # [3] manifest shape → rel. + manifest = _load_json(manifest_bytes, MANIFEST_SHAPE, "apply-manifest.json") + rel, m_pre, m_post, m_patch = validate_manifest_shape(manifest) + + # [4] bundle layout, final: the postimage subtree is EXACTLY rel's ancestor dirs plus + # rel — no extra, hidden or empty directory rides through. + parts = rel.split("/") + expected_dirs = {"/".join(parts[:i]) for i in range(1, len(parts))} + if post_leaves != {rel} or post_dirs != expected_dirs: + raise GateError(BUNDLE_LAYOUT, f"postimage layout {sorted(post_dirs | post_leaves)} " + f"!= exactly {sorted(expected_dirs | {rel})}") + postimage_bytes = _snapshot(os.path.join(postimage_root, *rel.split("/")), + BUNDLE_LAYOUT, f"postimage/{rel}") + + # [5] authority: pure plan + candidates; the manifest is re-derived, never trusted. + plan = _load_json(plan_bytes, AUTHORITY_BINDING, "--plan") + candidates = _load_json(cand_bytes, AUTHORITY_BINDING, "--candidates") + auth = validate_gate_authority(plan, candidates) + if rel != auth.rel: + raise GateError(AUTHORITY_BINDING, "manifest source path != plan/candidates") + + # [6] pristine preimage — read once; the isolated tree is built from THESE bytes. + preimage_bytes = verify_pristine(root, rel, auth.pre_sha256) + + # [7] artifact hashes — recompute over the bytes read in [2]-[6]. + if _sha_bytes(patch_bytes) != m_patch: + raise GateError(HASH_MISMATCH, "recomputed patch_sha256 != the manifest") + if _sha_bytes(postimage_bytes) != m_post: + raise GateError(HASH_MISMATCH, "recomputed post_sha256 != the manifest") + if m_pre != auth.pre_sha256: + raise GateError(HASH_MISMATCH, "manifest pre_sha256 != plan/candidates") + + # [8] the manifest must BE the canonical projection of plan + candidates + real bytes. + expected_manifest = _canonical_bytes({ + "version": 1, + "operation": "apply-subscription-fixes", + "input_bundle_sha256": auth.input_bundle_sha256, + "validated_plan_sha256": _sha_bytes(plan_bytes), + "target_api": {"subscribe": auth.target_subscribe}, + "source_files": [{"path": rel, "pre_sha256": auth.pre_sha256, + "post_sha256": _sha_bytes(postimage_bytes)}], + "applied_findings": auth.applied, + "manual_review_findings": auth.manual, + "patch_sha256": _sha_bytes(patch_bytes), + }) + if manifest_bytes != expected_manifest: + raise GateError(MANIFEST_SHAPE, + "manifest is not the canonical projection of plan/candidates/bytes") + + # [9] patch structure (frozen step 8 language). + empty_patch = patch_bytes == b"" + if empty_patch and auth.applied: + raise GateError(PATCH_STRUCTURE, "empty patch valid only when no convert_acquire") + if not empty_patch and not auth.applied: + raise GateError(PATCH_STRUCTURE, "a non-empty patch needs a convert_acquire") + parse_step8_patch(patch_bytes, rel, preimage_bytes) + + # [10] apply semantics + publication. + out_phys, parent_phys, root_phys = _out_parent(out, root) + workdir = _claim_workdir(parent_phys) + staging = os.path.join(workdir, "staging") + try: + if empty_patch: + # manual-only: pre == post == the pristine bytes; Git is not run. + if postimage_bytes != preimage_bytes: + raise GateError(APPLY_MISMATCH, "empty patch needs postimage == preimage") + if m_pre != m_post: + raise GateError(APPLY_MISMATCH, "empty patch needs pre_sha256 == post_sha256") + git_gates = "not_applicable" + else: + apply_in_throwaway(workdir, rel, preimage_bytes, postimage_bytes, + patch_bytes, m_post) + git_gates = "pass" + + evidence = _build_evidence(auth, rel, plan_bytes, manifest_bytes, patch_bytes, + m_pre, m_post, git_gates) + _publish(staging, out_phys, evidence, root_phys) + except BaseException: + shutil.rmtree(workdir, ignore_errors=True) + raise + shutil.rmtree(workdir, ignore_errors=True) + return out_phys diff --git a/tests/_preflight.py b/tests/_preflight.py new file mode 100644 index 00000000..768d6767 --- /dev/null +++ b/tests/_preflight.py @@ -0,0 +1,93 @@ +"""Preflight for the aggregate test runner. + +run_tests.py discovers every `test_*.py` and imports it with importlib. A single +import-time `sys.exit(...)` / `raise SystemExit(...)` in ANY of them ends the whole process +before the aggregate return code is collected — and the offender can sort BEFORE whatever +module is meant to police it, so a policing test module cannot catch it. The check must +therefore run as a PREFLIGHT, before the first test import (run_tests.run() calls +check_test_files() first and aborts on any violation). + +The invariant is deliberately strict and purely LOCATION-based — no call-graph analysis, +which an immediately-invoked helper or lambda defeats: + + a test_*.py may use sys.exit / raise SystemExit ONLY inside the body of a standalone + top-level `if __name__ == "__main__":` guard — nowhere else (not module scope, not a + function/lambda/class body, not a decorator or default). +""" + +from __future__ import annotations + +import ast +import os + + +def _is_main_guard(test: ast.expr) -> bool: + """Structurally EXACTLY `__name__ == "__main__"`.""" + return (isinstance(test, ast.Compare) + and isinstance(test.left, ast.Name) and test.left.id == "__name__" + and len(test.ops) == 1 and isinstance(test.ops[0], ast.Eq) + and len(test.comparators) == 1 + and isinstance(test.comparators[0], ast.Constant) + and test.comparators[0].value == "__main__") + + +def _is_systemexit(exc: ast.expr | None) -> bool: + if isinstance(exc, ast.Call): + exc = exc.func + return isinstance(exc, ast.Name) and exc.id == "SystemExit" + + +def _is_sys_exit(func: ast.expr) -> bool: + return (isinstance(func, ast.Attribute) and func.attr == "exit" + and isinstance(func.value, ast.Name) and func.value.id == "sys") + + +def _guard_body_ids(module: ast.Module) -> set[int]: + """Every AST node lexically inside a standalone top-level `if __name__ == "__main__":` + BODY (its `else`/`elif` are NOT included — they run on import).""" + allowed: set[int] = set() + for stmt in module.body: + if isinstance(stmt, ast.If) and _is_main_guard(stmt.test): + for s in stmt.body: + for node in ast.walk(s): + allowed.add(id(node)) + return allowed + + +def exit_violations(tree: ast.Module) -> list[ast.AST]: + """Every sys.exit call / raise SystemExit that sits OUTSIDE a main-guard body — which, + for a module imported by the runner, is any that could run at import (directly or via a + helper called at module scope). No exemption for function/lambda/class bodies.""" + allowed = _guard_body_ids(tree) + bad: list[ast.AST] = [] + for node in ast.walk(tree): + if id(node) in allowed: + continue + if isinstance(node, ast.Raise) and _is_systemexit(node.exc): + bad.append(node) + elif isinstance(node, ast.Call) and _is_sys_exit(node.func): + bad.append(node) + return bad + + +def check_test_files(tests_dir: str) -> list[str]: + """Return a list of human-readable violations across every test_*.py in `tests_dir`. + Empty means every test module is safe for the runner to import.""" + problems: list[str] = [] + for fname in sorted(os.listdir(tests_dir)): + if not (fname.startswith("test_") and fname.endswith(".py")): + continue + path = os.path.join(tests_dir, fname) + try: + with open(path, encoding="utf-8") as fh: + tree = ast.parse(fh.read(), filename=fname) + except (OSError, SyntaxError) as exc: + problems.append(f"{fname}: cannot parse ({exc})") + continue + for node in exit_violations(tree): + problems.append( + f"{fname}:{node.lineno}: sys.exit / raise SystemExit outside the " + "`if __name__ == \"__main__\"` guard body — it would end the aggregate " + "runner at import time" + ) + return problems diff --git a/tests/gate_regressions.sh b/tests/gate_regressions.sh new file mode 100644 index 00000000..96fa388c --- /dev/null +++ b/tests/gate_regressions.sh @@ -0,0 +1,258 @@ +#!/usr/bin/env bash +# S2 step 9 — the structural self-gate, end to end. Run from the repo root with +# `dotnet` + `python3` + `git` on PATH: +# +# bash tests/gate_regressions.sh +# +# Builds a REAL step 8 bundle (extractor -> candidates -> validate_plan -> apply, which +# runs the accepted Owen.CSharp.Rewriter), then gates it. Covers the filesystem-real and +# git-real cases; the pure-function + byte-tampering cases are in tests/test_gate_patch.py. +# +# Every forged fixture rebinds the upstream bindings it would otherwise trip first, so the +# refusal comes from the branch under test — a test that refuses for the wrong reason is +# not a test. +set -uo pipefail + +T="${1:?usage: gate_regressions.sh }" +mkdir -p "$T"; T=$(cd "$T" && pwd) +REPO="$PWD" +EXT="$REPO/frontend/roslyn/OwnSharp.Extractor" +RW="$REPO/frontend/roslyn/Owen.CSharp.Rewriter" +FC=frontend/roslyn/samples/FixCandidatesSample.cs +fails=0 +ok() { echo " ok: $1"; } +bad() { echo " FAIL: $1"; fails=$((fails + 1)); } + +printf '[weak-subscription]\nsubscribe = ["WeakEvents.AddPropertyChanged"]\n' > "$T/own.toml" +dotnet build "$EXT" -v q --nologo > /dev/null || { echo "FAIL: extractor build"; exit 1; } +dotnet build "$RW" -v q --nologo > /dev/null || { echo "FAIL: rewriter build"; exit 1; } +REWRITER="dotnet run --project $RW --no-build --" + +cat > "$T/mkplan.py" <<'PY' +import json +import sys + +sys.path.insert(0, ".") +from ownlang.fix_plan import validate_plan + +c = json.load(open(sys.argv[1])) +conv = set(sys.argv[3:]) if len(sys.argv) > 3 else {x["finding_id"] for x in c["candidates"]} +d = [{"finding_id": x["finding_id"], + "action": "convert_acquire" if x["finding_id"] in conv else "manual_review"} + for x in c["candidates"]] +json.dump(validate_plan(c, {"version": 1, "decisions": d}), open(sys.argv[2], "w")) +PY + +# A helper to rebind a bundle's manifest after forging patch/postimage bytes, so a fixture +# reaches the git gates instead of dying at the hash gate. +cat > "$T/rebind.py" <<'PY' +import hashlib +import json +import sys + + +def sha(b): + return "sha256:" + hashlib.sha256(b).hexdigest() + + +bundle, rel = sys.argv[1], sys.argv[2] +m = json.load(open(f"{bundle}/apply-manifest.json")) +m["patch_sha256"] = sha(open(f"{bundle}/change.patch", "rb").read()) +m["source_files"][0]["post_sha256"] = sha(open(f"{bundle}/postimage/{rel}", "rb").read()) +blob = json.dumps(m, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + b"\n" +open(f"{bundle}/apply-manifest.json", "wb").write(blob) +PY + +gate() { # gate [plan] [candidates] [root] + python3 -m ownlang own-fix subscriptions gate --bundle "$2" \ + --plan "${3:-$T/plan.json}" --candidates "${4:-$T/candidates.json}" \ + --root "${5:-.}" --out "$1" > /dev/null 2>"$T/err.txt" +} +refuse() { # refuse [plan] [candidates] [root] + rm -rf "$2" + gate "$2" "$3" "${5:-$T/plan.json}" "${6:-$T/candidates.json}" "${7:-.}" + local rc=$? + [ "$rc" = 2 ] || { bad "$1: expected exit 2, got $rc"; return; } + [ ! -e "$2" ] || { bad "$1: a refused run left an out-dir"; return; } + grep -q "own-fix: refuse: $4:" "$T/err.txt" \ + || { bad "$1: wrong category: $(cat "$T/err.txt")"; return; } + ok "$1 -> $4" +} + +echo "== build the real step 8 bundle ==" +dotnet run --project "$EXT" --no-build -- "$FC" --fix-candidates -o "$T/fc.json" > /dev/null 2>&1 \ + || { echo "FAIL: extractor"; exit 1; } +python3 -m ownlang own-fix subscriptions candidates "$T/fc.json" --config "$T/own.toml" \ + --class Own.Samples.FixCandidates.TwoOnOneLine --output "$T/candidates.json" --root . > /dev/null \ + || { echo "FAIL: candidates"; exit 1; } +python3 "$T/mkplan.py" "$T/candidates.json" "$T/plan.json" || { echo "FAIL: plan"; exit 1; } +rm -rf "$T/bundle" +python3 -m ownlang own-fix subscriptions apply --plan "$T/plan.json" \ + --candidates "$T/candidates.json" --root . --out "$T/bundle" --rewriter "$REWRITER" > /dev/null \ + || { echo "FAIL: apply"; exit 1; } + +echo "== 0. happy path ==" +rm -rf "$T/g0" +gate "$T/g0" "$T/bundle" && ok "the gate passes a valid bundle" || bad "gate: $(cat "$T/err.txt")" +python3 - "$T/g0/gate-result.json" "$FC" <<'PY' || bad "evidence shape" +import json +import sys + +m = json.load(open(sys.argv[1])) +raw = open(sys.argv[1], "rb").read() +assert raw == json.dumps(m, sort_keys=True, separators=(",", ":"), + ensure_ascii=False).encode() + b"\n", "not canonical bytes" +assert m["operation"] == "gate-subscription-fix-bundle" +assert set(m["gates"].values()) == {"pass"}, m["gates"] +assert set(m["gates"]) == {"bundle_layout", "manifest_shape", "authority_binding", + "artifact_hashes", "pristine_preimage", "patch_structure", + "git_apply_check", "git_apply", "postimage_equality", + "isolated_tree"}, sorted(m["gates"]) +assert m["source_files"][0]["path"] == sys.argv[2] +for banned in ("1970", "/tmp", "/home", "runner", "owen-gate"): + assert banned not in raw.decode(), banned +print("evidence ok") +PY +ok "evidence: exact shape, canonical bytes, all gates pass, no host/temp/timestamp" +git diff --quiet -- "$FC" && ok "source tree untouched" || bad "source tree modified" + +echo "== 1. determinism ==" +rm -rf "$T/g0b" +gate "$T/g0b" "$T/bundle" +diff -r "$T/g0" "$T/g0b" > /dev/null && ok "two runs give byte-identical evidence" \ + || bad "evidence is not deterministic" + +echo "== 2. manual-only: empty patch, git not_applicable ==" +python3 "$T/mkplan.py" "$T/candidates.json" "$T/plan_m.json" __none__ +rm -rf "$T/bundle_m" +python3 -m ownlang own-fix subscriptions apply --plan "$T/plan_m.json" \ + --candidates "$T/candidates.json" --root . --out "$T/bundle_m" --rewriter "$REWRITER" > /dev/null \ + || bad "manual-only apply" +[ -f "$T/bundle_m/change.patch" ] && [ ! -s "$T/bundle_m/change.patch" ] \ + && ok "the step 8 patch is zero length" || bad "manual-only patch is not empty" +rm -rf "$T/gm" +gate "$T/gm" "$T/bundle_m" "$T/plan_m.json" && ok "the gate passes a manual-only bundle" \ + || bad "manual-only gate: $(cat "$T/err.txt")" +python3 - "$T/gm/gate-result.json" <<'PY' || bad "manual-only evidence" +import json +import sys + +m = json.load(open(sys.argv[1])) +for g in ("git_apply_check", "git_apply", "isolated_tree"): + assert m["gates"][g] == "not_applicable", (g, m["gates"][g]) +for g in ("bundle_layout", "manifest_shape", "authority_binding", "artifact_hashes", + "pristine_preimage", "patch_structure", "postimage_equality"): + assert m["gates"][g] == "pass", (g, m["gates"][g]) +assert m["applied_findings"] == [] +assert len(m["manual_review_findings"]) == 2 +print("manual-only ok") +PY +ok "manual-only: git gates not_applicable, applied empty, all else pass" + +echo "== 3. apply semantics (git is the independent applier) ==" +# git apply --check must fail: forge a context line the source will not match, rebind. +cp -r "$T/bundle" "$T/bundle_ctx" +sed -i 's/ public sealed class TwoOnOneLine/ public sealed class SomethingElse/' \ + "$T/bundle_ctx/change.patch" 2>/dev/null || true +# ...rebind so the hash gate passes and git apply --check is what fires. +python3 "$T/rebind.py" "$T/bundle_ctx" "$FC" +refuse "structurally valid but git apply --check fails" "$T/g_ctx" "$T/bundle_ctx" APPLY_CHECK + +# git apply succeeds but the result != postimage: forge the postimage, rebind post_sha. +cp -r "$T/bundle" "$T/bundle_pm" +printf '// forged\n' >> "$T/bundle_pm/postimage/$FC" +python3 "$T/rebind.py" "$T/bundle_pm" "$FC" +refuse "applied bytes != postimage" "$T/g_pm" "$T/bundle_pm" APPLY_MISMATCH + +# a modified postimage WITHOUT rebinding is caught earlier, at the hash gate. +cp -r "$T/bundle" "$T/bundle_h" +printf '// forged\n' >> "$T/bundle_h/postimage/$FC" +refuse "modified postimage (no rebind)" "$T/g_h" "$T/bundle_h" HASH_MISMATCH + +echo "== 4. bundle layout ==" +cp -r "$T/bundle" "$T/bundle_extra"; touch "$T/bundle_extra/surprise.txt" +refuse "an extra file in the bundle" "$T/g_extra" "$T/bundle_extra" BUNDLE_LAYOUT +cp -r "$T/bundle" "$T/bundle_sym"; rm "$T/bundle_sym/change.patch" +ln -s /etc/hostname "$T/bundle_sym/change.patch" +refuse "a symlink entry in the bundle" "$T/g_sym" "$T/bundle_sym" BUNDLE_LAYOUT +cp -r "$T/bundle" "$T/bundle_pi_extra" +touch "$T/bundle_pi_extra/postimage/frontend/roslyn/samples/Extra.cs" +refuse "an extra postimage file" "$T/g_pie" "$T/bundle_pi_extra" BUNDLE_LAYOUT +# An extra EMPTY directory (a file-set check alone would miss it). +cp -r "$T/bundle" "$T/bundle_ed"; mkdir -p "$T/bundle_ed/postimage/frontend/roslyn/samples/empty" +refuse "an extra empty directory" "$T/g_ed" "$T/bundle_ed" BUNDLE_LAYOUT +cp -r "$T/bundle" "$T/bundle_hd"; mkdir -p "$T/bundle_hd/postimage/.hidden" +refuse "a hidden empty directory" "$T/g_hd" "$T/bundle_hd" BUNDLE_LAYOUT +cp -r "$T/bundle" "$T/bundle_nd"; mkdir -p "$T/bundle_nd/postimage/a/b/c" +refuse "a nested empty directory" "$T/g_nd" "$T/bundle_nd" BUNDLE_LAYOUT +# A symlinked bundle ROOT (rejected by lstat before realpath). +ln -s "$T/bundle" "$T/bundle_link" +refuse "a symlinked bundle root" "$T/g_bl" "$T/bundle_link" BUNDLE_LAYOUT + +echo "== 5. pristine source ==" +zero=$(printf '0%.0s' $(seq 1 64)) +# stale preimage: the manifest/plan say a pre_sha the real source no longer has. Rebind the +# candidates + plan + manifest so ONLY the pristine compare can refuse. +python3 - "$T" "$FC" "$zero" <<'PY' +import hashlib +import json +import os +import sys + +sys.path.insert(0, ".") +from ownlang.fix_plan import bundle_sha256 + +t, fc, zero = sys.argv[1], sys.argv[2], sys.argv[3] +stale = "sha256:" + zero +cand = json.load(open(f"{t}/candidates.json")) +cand["source_files"][0]["sha256"] = stale +json.dump(cand, open(f"{t}/cand_stale.json", "w")) +plan = json.load(open(f"{t}/plan.json")) +plan["source_files"][0]["sha256"] = stale +plan["input_bundle_sha256"] = bundle_sha256(cand) +json.dump(plan, open(f"{t}/plan_stale.json", "w")) +os.makedirs(f"{t}/bundle_stale/postimage/{os.path.dirname(fc)}", exist_ok=True) +for f in ("change.patch",): + with open(f"{t}/bundle/{f}", "rb") as r, open(f"{t}/bundle_stale/{f}", "wb") as w: + w.write(r.read()) +with open(f"{t}/bundle/postimage/{fc}", "rb") as r, open(f"{t}/bundle_stale/postimage/{fc}", "wb") as w: + w.write(r.read()) +m = json.load(open(f"{t}/bundle/apply-manifest.json")) +m["input_bundle_sha256"] = bundle_sha256(cand) +plan_bytes = open(f"{t}/plan_stale.json", "rb").read() +m["validated_plan_sha256"] = "sha256:" + hashlib.sha256(plan_bytes).hexdigest() +m["source_files"][0]["pre_sha256"] = stale +blob = json.dumps(m, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + b"\n" +open(f"{t}/bundle_stale/apply-manifest.json", "wb").write(blob) +PY +refuse "a stale preimage" "$T/g_stale" "$T/bundle_stale" PRISTINE_SOURCE \ + "$T/plan_stale.json" "$T/cand_stale.json" . + +# a source reached through a symlinked directory that escapes the root. +rm -rf "$T/symroot"; mkdir -p "$T/symroot/outside" +cp -r "$REPO/frontend" "$T/symroot/outside/frontend" +ln -s outside/frontend "$T/symroot/frontend" +mkdir -p "$T/symroot/real" +refuse "source through a symlinked dir (escape)" "$T/g_symsrc" "$T/bundle" PRISTINE_SOURCE \ + "$T/plan.json" "$T/candidates.json" "$T/symroot/real" + +echo "== 6. publication ==" +rm -rf "$T/exists"; mkdir -p "$T/exists"; touch "$T/exists/stale" +gate "$T/exists" "$T/bundle" +{ [ $? = 2 ] && grep -q "PUBLICATION" "$T/err.txt" && [ -f "$T/exists/stale" ]; } \ + && ok "a pre-existing out is refused, untouched" || bad "pre-existing out: $(cat "$T/err.txt")" +gate "./gate_inside" "$T/bundle" +{ [ $? = 2 ] && grep -q "inside the source root" "$T/err.txt" && [ ! -e ./gate_inside ]; } \ + && ok "an out inside the source root is refused" || bad "out inside root: $(cat "$T/err.txt")" +rm -rf "$T/ro"; mkdir -p "$T/ro"; chmod 555 "$T/ro" +gate "$T/ro/out" "$T/bundle"; rc=$?; chmod 755 "$T/ro" +{ [ "$rc" = 2 ] && [ ! -e "$T/ro/out" ] && [ -z "$(ls -A "$T/ro")" ]; } \ + && ok "a failed publication leaves no out and no staging" \ + || bad "read-only parent: rc=$rc leftovers=[$(ls -A "$T/ro")]" + +git diff --quiet -- "$FC" && ok "source tree still untouched after every refusal" \ + || bad "source tree was modified" + +echo +[ "$fails" = 0 ] && echo "GATE REGRESSIONS: ALL PASS" || echo "GATE REGRESSIONS: $fails FAILURE(S)" +exit "$fails" diff --git a/tests/run_tests.py b/tests/run_tests.py index 6aba88a8..988989b7 100644 --- a/tests/run_tests.py +++ b/tests/run_tests.py @@ -994,6 +994,18 @@ def helper_and_report_smoke() -> list[str]: def run() -> int: + # PREFLIGHT — BEFORE importing any test module. A test_*.py that ends the process at + # import (sys.exit / raise SystemExit outside its `__main__` guard) would silently + # truncate this run, and it can sort before whatever module is meant to police it, so + # the check must happen here, first, over the source rather than by importing. + from _preflight import check_test_files + _here = os.path.dirname(os.path.abspath(__file__)) + preflight = check_test_files(_here) + for problem in preflight: + print(f"PREFLIGHT FAIL: {problem}") + if preflight: + return 1 + passed = 0 failed = 0 for name, body, expected in CASES: diff --git a/tests/test_gate_patch.py b/tests/test_gate_patch.py new file mode 100644 index 00000000..22859ab4 --- /dev/null +++ b/tests/test_gate_patch.py @@ -0,0 +1,460 @@ +"""S2 step 9 — the gate's pure functions and the tampering cases that need forged bytes. + +These call the validators / patch parser directly, so they run WITHOUT dotnet or git (the +`tests (pyX)` job). The filesystem-real cases — symlink entries, git-apply semantics, +publication — live in tests/gate_regressions.sh, which drives the whole chain. + +Every forged fixture that must reach a SPECIFIC gate rebinds the upstream bindings it would +otherwise trip first (bundle_sha256, the canonical plan projection), so the refusal comes +from the branch under test — a test that refuses for the wrong reason is not a test. +""" + +from __future__ import annotations + +import hashlib +import os +import shutil +import sys +import tempfile +from typing import Any + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from ownlang.fix_gate import ( + AUTHORITY_BINDING, + MANIFEST_SHAPE, + PATCH_STRUCTURE, + GateError, + _bundle_sha256, + _claim_workdir, + _same_or_inside, + _same_path, + parse_step8_patch, + validate_gate_authority, + validate_manifest_shape, +) + +checks = 0 +failures: list[str] = [] + + +def check(cond: bool, label: str) -> None: + global checks + checks += 1 + if not cond: + failures.append(label) + + +def refuses(fn: Any, category: str, label: str) -> None: + global checks + checks += 1 + try: + fn() + except GateError as exc: + if exc.category != category: + failures.append(f"{label}: category {exc.category} != {category} ({exc})") + return + except Exception as exc: + failures.append(f"{label}: raised {type(exc).__name__}, not GateError ({exc})") + return + failures.append(f"{label}: expected a GateError[{category}]") + + +def sha(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def cp(o: Any) -> Any: + import json + return json.loads(json.dumps(o)) + + +REL = "src/Sample.cs" +PRE = b"class A\n{\n void M()\n {\n p.PropertyChanged += OnX;\n }\n}\n" +POST = PRE.replace(b"p.PropertyChanged += OnX;", b"WeakEvents.AddPropertyChanged(p, OnX);") +FID_A = "OWN001:sha256:" + "a" * 64 +FID_B = "OWN050:sha256:" + "b" * 64 + + +def a_span(start: int) -> dict[str, int]: + return {"start": start, "length": 10, "start_line": 5, "start_column": 9, + "end_line": 5, "end_column": 19} + + +def a_candidate(fid: str, start: int, contract: str = "inotify_property_changed", + actions: list[str] | None = None) -> dict[str, Any]: + return { + "finding_id": fid, + "diagnostic_code": "OWN001", + "containing_type": "N.A", + "file": REL, + "enclosing_member": "N.A..ctor(N.IPub)", + "event": "PropertyChanged", + "event_identity": "System.ComponentModel.INotifyPropertyChanged.PropertyChanged", + "event_contract": contract, + "source": "p", + "source_identity": "p", + "source_identity_kind": "computed", + "handler": "OnX", + "handler_identity": "N.A.OnX(object, System.ComponentModel.PropertyChangedEventArgs)", + "handler_identity_kind": "stable_symbol", + "occurrence_ordinal": 0, + "acquire_span": a_span(start), + "teardown": {"status": "none", "candidates": []}, + "allowed_actions": actions or ["convert_acquire", "manual_review"], + } + + +def base_candidates() -> dict[str, Any]: + return { + "version": 1, + "operation": "fix-subscriptions", + "target_api": {"subscribe": "WeakEvents.AddPropertyChanged"}, + "selection": { + "allowed_types": [{"full_name": "N.A", "file": REL}], + "selected_findings": None, + "constraints": {"max_types_changed": 1, "max_files_changed": 1, + "allow_helper_changes": False, "allow_config_changes": False, + "allow_suppressions": False}, + }, + "source_files": [{"path": REL, "sha256": sha(PRE)}], + "candidates": [a_candidate(FID_A, 40), a_candidate(FID_B, 80)], + } + + +def plan_for(cands: dict[str, Any], actions: list[str]) -> dict[str, Any]: + return { + "version": 1, + "operation": "fix-subscriptions", + "input_bundle_sha256": _bundle_sha256(cands), + "target_api": {"subscribe": cands["target_api"]["subscribe"]}, + "selection": { + "allowed_types": [dict(cands["selection"]["allowed_types"][0])], + "selected_findings": cands["selection"]["selected_findings"], + "constraints": dict(cands["selection"]["constraints"]), + }, + "source_files": [dict(cands["source_files"][0])], + "decisions": [ + {"finding_id": c["finding_id"], "action": actions[i], "file": c["file"], + "acquire_span": c["acquire_span"]} + for i, c in enumerate(cands["candidates"]) + ], + } + + +# --- authority validator ----------------------------------------------------------- + +cands = base_candidates() +plan = plan_for(cands, ["convert_acquire", "manual_review"]) +auth = validate_gate_authority(plan, cands) +check(auth.rel == REL, "authority: rel") +check(auth.applied == [FID_A] and auth.manual == [FID_B], "authority: partition in candidate order") +check(auth.pre_sha256 == sha(PRE), "authority: pre_sha256") +check(auth.input_bundle_sha256 == _bundle_sha256(cands), "authority: input_bundle_sha256") + +# A forged permission tier: name_only claiming convert_acquire, freshly re-bound so only +# the tiering can refuse it. +forged = base_candidates() +forged["candidates"][0]["event_contract"] = "name_only" +refuses(lambda: validate_gate_authority(plan_for(forged, ["convert_acquire", "manual_review"]), + forged), AUTHORITY_BINDING, "authority: forged tier") + +# The hash must actually bind. +bad_hash = plan_for(cands, ["convert_acquire", "manual_review"]) +bad_hash["input_bundle_sha256"] = "sha256:" + "0" * 64 +refuses(lambda: validate_gate_authority(bad_hash, cands), AUTHORITY_BINDING, + "authority: hash does not bind") + +# Decision order must equal candidate order. +reordered = plan_for(cands, ["convert_acquire", "manual_review"]) +reordered["decisions"].reverse() +refuses(lambda: validate_gate_authority(reordered, cands), AUTHORITY_BINDING, + "authority: decisions out of candidate order") + +# An unknown key anywhere in the envelope. +extra = plan_for(cands, ["convert_acquire", "manual_review"]) +extra["autofix_everything"] = True +refuses(lambda: validate_gate_authority(extra, cands), AUTHORITY_BINDING, + "authority: unknown plan key") +extra_c = base_candidates() +extra_c["candidates"][0]["surprise"] = 1 +refuses(lambda: validate_gate_authority(plan_for(extra_c, ["convert_acquire", "manual_review"]), + extra_c), + AUTHORITY_BINDING, "authority: unknown candidate key") + +# The type file and the source file must be the same file. +drift = base_candidates() +drift["selection"]["allowed_types"][0]["file"] = "src/Other.cs" +refuses(lambda: validate_gate_authority(plan_for(drift, ["convert_acquire", "manual_review"]), + drift), + AUTHORITY_BINDING, "authority: type/source file drift") + +# An action outside the candidate's own allowed list (the candidate only permits +# manual_review, the plan tries to convert it). +def base_no_convert() -> dict[str, Any]: + c = base_candidates() + c["candidates"][1]["event_contract"] = "name_only" + c["candidates"][1]["allowed_actions"] = ["manual_review"] + return c + + +nc = base_no_convert() +refuses(lambda: validate_gate_authority(plan_for(nc, ["convert_acquire", "convert_acquire"]), nc), + AUTHORITY_BINDING, "authority: action not in candidate's allowed list") + +# A span that disagrees between decision and candidate. +span_drift = plan_for(cands, ["convert_acquire", "manual_review"]) +span_drift["decisions"][0]["acquire_span"] = a_span(999) +refuses(lambda: validate_gate_authority(span_drift, cands), AUTHORITY_BINDING, + "authority: decision span != candidate") + +# selected_findings must name exactly the candidates. +sel = base_candidates() +sel["selection"]["selected_findings"] = [FID_A] +refuses(lambda: validate_gate_authority(plan_for(sel, ["convert_acquire", "manual_review"]), sel), + AUTHORITY_BINDING, "authority: selected_findings incomplete") + + +# Full value / nested-shape validation of the frozen candidate (amendment 2): the exact key +# set alone is not enough — a wrong-typed value or a malformed teardown on a re-hashed bundle +# must still be refused, so each fixture is bound through plan_for. +def tamper(mut: Any, label: str) -> None: + c = base_candidates() + mut(c["candidates"][0]) + refuses(lambda: validate_gate_authority(plan_for(c, ["convert_acquire", "manual_review"]), c), + AUTHORITY_BINDING, label) + + +def _set(key: str, value: Any) -> Any: + return lambda cand: cand.__setitem__(key, value) + + +tamper(_set("event_identity", 123), "authority: event_identity wrong type") +tamper(_set("occurrence_ordinal", "banana"), "authority: occurrence_ordinal wrong type") +tamper(_set("occurrence_ordinal", -1), "authority: occurrence_ordinal negative") +tamper(_set("diagnostic_code", 7), "authority: diagnostic_code wrong type") +tamper(_set("enclosing_member", None), "authority: enclosing_member wrong type") +tamper(lambda c: c["teardown"].__setitem__("whatever", True), "authority: teardown extra key") +tamper(lambda c: c["teardown"].__setitem__("status", "maybe"), "authority: teardown unknown status") +tamper(_set("teardown", {"status": "exact", "candidates": [{"source": "a"}]}), + "authority: teardown candidate malformed") +tamper(_set("teardown", {"status": "exact", "candidates": [ + {"source": "a", "handler": "b", "match": "text", "span": {"start": 1, "length": 2}}]}), + "authority: teardown candidate span shape") + + +# --- manifest shape ---------------------------------------------------------------- + + +def a_manifest(**over: Any) -> dict[str, Any]: + m = { + "version": 1, + "operation": "apply-subscription-fixes", + "input_bundle_sha256": _bundle_sha256(cands), + "validated_plan_sha256": "sha256:" + "d" * 64, + "target_api": {"subscribe": "WeakEvents.AddPropertyChanged"}, + "source_files": [{"path": REL, "pre_sha256": sha(PRE), "post_sha256": sha(POST)}], + "applied_findings": [FID_A], + "manual_review_findings": [FID_B], + "patch_sha256": sha(b"x"), + } + m.update(over) + return m + + +rel_m, pre_m, post_m, patch_m = validate_manifest_shape(a_manifest()) +check((rel_m, pre_m, post_m) == (REL, sha(PRE), sha(POST)), "manifest: returns rel + shas") +refuses(lambda: validate_manifest_shape(a_manifest(surprise=1)), MANIFEST_SHAPE, + "manifest: extra key") +refuses(lambda: validate_manifest_shape(a_manifest(version=2)), MANIFEST_SHAPE, + "manifest: bad version") +refuses(lambda: validate_manifest_shape(a_manifest(operation="rm -rf")), MANIFEST_SHAPE, + "manifest: bad operation") +refuses(lambda: validate_manifest_shape(a_manifest(patch_sha256="nope")), MANIFEST_SHAPE, + "manifest: bad sha format") +refuses(lambda: validate_manifest_shape(a_manifest(applied_findings=[FID_A, FID_A])), + MANIFEST_SHAPE, "manifest: duplicate findings") +refuses(lambda: validate_manifest_shape(a_manifest(applied_findings=[FID_A], + manual_review_findings=[FID_A])), + MANIFEST_SHAPE, "manifest: overlapping partitions") +refuses(lambda: validate_manifest_shape(a_manifest( + source_files=[{"path": "/abs/x.cs", "pre_sha256": sha(PRE), "post_sha256": sha(POST)}])), + MANIFEST_SHAPE, "manifest: non-canonical path") +refuses(lambda: validate_manifest_shape(a_manifest(source_files=[])), MANIFEST_SHAPE, + "manifest: not exactly one source file") + + +# --- the strict step 8 patch language ---------------------------------------------- + +GOOD = (b"diff --git a/" + REL.encode() + b" b/" + REL.encode() + b"\n" + b"--- a/" + REL.encode() + b"\n" + b"+++ b/" + REL.encode() + b"\n" + b"@@ -5,1 +5,1 @@\n" + b"- p.PropertyChanged += OnX;\n" + b"+ WeakEvents.AddPropertyChanged(p, OnX);\n") + +parse_step8_patch(GOOD, REL, PRE) # must not raise +check(True, "patch: a canonical step 8 patch parses") +parse_step8_patch(b"", REL, PRE) # the empty patch is structurally fine here +check(True, "patch: the empty patch parses") + +# A space in the path is legal and must be accepted (git reads names literally). +SP = "src/with space.cs" +sp_patch = (b"diff --git a/" + SP.encode() + b" b/" + SP.encode() + b"\n" + b"--- a/" + SP.encode() + b"\n+++ b/" + SP.encode() + b"\n" + b"@@ -1,1 +1,1 @@\n-a\n+b\n") +parse_step8_patch(sp_patch, SP, b"a\n") +check(True, "patch: a space in the path is accepted") + + +def variant(replace_pairs: list[tuple[bytes, bytes]], insert_after: bytes = b"", + extra: bytes = b"") -> bytes: + p = GOOD + for a, b in replace_pairs: + p = p.replace(a, b) + if insert_after: + p = p.replace(insert_after, insert_after + extra) + return p + + +REB = REL.encode() +# A second file. +_SECOND = (b"diff --git a/other.cs b/other.cs\n--- a/other.cs\n" + b"+++ b/other.cs\n@@ -1 +1 @@\n-a\n+b\n") +refuses(lambda: parse_step8_patch(GOOD + _SECOND, REL, PRE), + PATCH_STRUCTURE, "patch: a second file") +# Wrong header path. +refuses(lambda: parse_step8_patch(GOOD.replace(b"a/" + REB, b"a/evil.cs", 1), REL, PRE), + PATCH_STRUCTURE, "patch: wrong '---' path") +# A rename / copy / mode / index / binary record grafted in. +for rec, name in ((b"rename from x\n", "rename"), (b"copy from x\n", "copy"), + (b"old mode 100644\n", "mode"), (b"index abc..def 100644\n", "index"), + (b"GIT binary patch\n", "binary"), + (b"new file mode 100644\n", "new-file"), + (b"deleted file mode 100644\n", "deleted-file")): + refuses(lambda r=rec: parse_step8_patch( + GOOD.replace(b"@@ -5,1", r + b"@@ -5,1", 1), REL, PRE), + PATCH_STRUCTURE, f"patch: a {name} record") +# An absolute or traversal path in the header. +refuses(lambda: parse_step8_patch( + (b"diff --git a//etc/passwd b//etc/passwd\n--- a//etc/passwd\n" + b"+++ b//etc/passwd\n@@ -1 +1 @@\n-a\n+b\n"), REL, PRE), + PATCH_STRUCTURE, "patch: an absolute path") +refuses(lambda: parse_step8_patch(GOOD.replace(REB, b"../../etc/passwd"), REL, PRE), + PATCH_STRUCTURE, "patch: a traversal path") +# A quoted (C-escaped) alternate filename never matches the exact expected header. +refuses(lambda: parse_step8_patch(GOOD.replace(b"a/" + REB, b"\"a/" + REB + b"\"", 1), + REL, PRE), PATCH_STRUCTURE, "patch: a quoted path") +# Hunk arithmetic that does not add up. +refuses(lambda: parse_step8_patch(GOOD.replace(b"@@ -5,1 +5,1 @@", b"@@ -5,2 +5,1 @@"), + REL, PRE), PATCH_STRUCTURE, "patch: wrong old count") +# A malformed hunk header. +refuses(lambda: parse_step8_patch(GOOD.replace(b"@@ -5,1 +5,1 @@", b"@@ nonsense @@"), + REL, PRE), PATCH_STRUCTURE, "patch: malformed hunk header") +# An unterminated final line. +refuses(lambda: parse_step8_patch(GOOD[:-1], REL, PRE), PATCH_STRUCTURE, + "patch: unterminated final line") +# A hunk range past the end of the preimage. +refuses(lambda: parse_step8_patch(GOOD.replace(b"@@ -5,1 +5,1 @@", b"@@ -99,1 +99,1 @@"), + REL, PRE), PATCH_STRUCTURE, "patch: range past the preimage") + +# --- the no-newline marker + range invariants (amendment / blocker 5) -------------- + +PRE3 = b"a\nb\nc\n" +X = "x.cs" + + +def small(hunk: bytes) -> bytes: + xb = X.encode() + return (b"diff --git a/" + xb + b" b/" + xb + b"\n--- a/" + xb + b"\n+++ b/" + xb + b"\n" + + hunk) + + +parse_step8_patch(small(b"@@ -1,1 +1,1 @@\n-a\n+z\n"), X, PRE3) +check(True, "patch: a minimal single-line change parses") +# A legitimate no-newline marker on the last line of each side. +parse_step8_patch(small(b"@@ -3,1 +3,1 @@\n-c\n\\ No newline at end of file\n+z\n" + b"\\ No newline at end of file\n"), X, b"a\nb\nc") +check(True, "patch: a well-placed no-newline marker parses") +# A marker before any eligible line. +refuses(lambda: parse_step8_patch( + small(b"@@ -1,1 +1,1 @@\n\\ No newline at end of file\n-a\n+z\n"), X, PRE3), + PATCH_STRUCTURE, "patch: marker before any line") +# Two markers in a row (a duplicate / stray marker). +refuses(lambda: parse_step8_patch( + small(b"@@ -1,1 +1,1 @@\n-a\n\\ No newline at end of file\n" + b"\\ No newline at end of file\n+z\n"), X, PRE3), + PATCH_STRUCTURE, "patch: a duplicate no-newline marker") +# A marker on a line that is NOT the last of its side (within the hunk). +refuses(lambda: parse_step8_patch( + small(b"@@ -1,2 +1,1 @@\n-a\n\\ No newline at end of file\n-b\n+z\n"), X, PRE3), + PATCH_STRUCTURE, "patch: marker with more of the same side after it") +# A marker in an EARLIER hunk: the no-newline is EOF, so no later hunk may follow it. This +# is the file-level invariant (a per-hunk reset would wrongly accept it). +refuses(lambda: parse_step8_patch( + small(b"@@ -1,1 +1,1 @@\n-a\n\\ No newline at end of file\n+x\n" + b"@@ -3,1 +3,1 @@\n-c\n+z\n"), X, PRE3), + PATCH_STRUCTURE, "patch: a second hunk after a no-newline marker") +# A zero-length (insertion) range whose start is past the preimage. +refuses(lambda: parse_step8_patch(small(b"@@ -99,0 +4,1 @@\n+z\n"), X, PRE3), + PATCH_STRUCTURE, "patch: an insertion range past the preimage") +# A non-zero range starting at line 0. +refuses(lambda: parse_step8_patch(small(b"@@ -0,1 +1,1 @@\n-a\n+z\n"), X, PRE3), + PATCH_STRUCTURE, "patch: a non-zero range starting at 0") +# A valid pure insertion at the top (old range -0,0). +parse_step8_patch(small(b"@@ -0,0 +1,1 @@\n+z\n"), X, PRE3) +check(True, "patch: a pure insertion at the top parses") + + +# --- claim cleanup + platform-aware path equality ----------------------------------- + +check(_same_path("/a/b", "/a/b"), "same_path: identical") +check(not _same_path("/a/b", "/a/c"), "same_path: different") +if os.name == "nt": + check(_same_path("C:\\A\\B", "c:\\a\\b"), "same_path: Windows case-insensitive") +else: + check(not _same_path("/a/B", "/a/b"), "same_path: POSIX case-sensitive") + +# A claim whose self-resolution proof fails (a symlinked parent, so realpath(path) != path) +# must REMOVE the directory it just created — no leftover. Symlink-capable platforms only. +_tmp = tempfile.mkdtemp() +_real = os.path.join(_tmp, "real") +_link = os.path.join(_tmp, "link") +os.mkdir(_real) +try: + os.symlink(_real, _link, target_is_directory=True) + _have_symlink = True +except (OSError, NotImplementedError): + _have_symlink = False +if _have_symlink: + _before = set(os.listdir(_real)) + refuses(lambda: _claim_workdir(_link), "PUBLICATION", + "claim: a non-self-resolving parent is refused") + check(set(os.listdir(_real)) == _before, + "claim: a failed proof leaves no directory behind") +shutil.rmtree(_tmp, ignore_errors=True) + + +# --- containment platform rule ----------------------------------------------------- + +check(_same_or_inside("/repo", "/repo"), "containment: root contains itself") +check(_same_or_inside("/repo", "/repo/sub/x"), "containment: a descendant") +check(not _same_or_inside("/repo", "/repository"), "containment: a name prefix is not inside") +check(not _same_or_inside("/repo", "/other"), "containment: unrelated is outside") +if os.name == "nt": + check(_same_or_inside("C:\\Repo", "c:\\repo\\x"), "containment: Windows case-insensitive") +else: + check(not _same_or_inside("/repo", "/REPO/x"), "containment: POSIX case-sensitive") + + +def run() -> int: + """The aggregate contract run_tests.py expects: report + an int rc, NEVER a + process-ending sys.exit at import time (which would silence every later module).""" + print(f"gate (S2 step 9): {checks - len(failures)}/{checks} checks pass") + for f in failures: + print(f" FAIL: {f}") + return 1 if failures else 0 + + +if __name__ == "__main__": + raise SystemExit(run()) diff --git a/tests/test_harness_contract.py b/tests/test_harness_contract.py new file mode 100644 index 00000000..ef1de828 --- /dev/null +++ b/tests/test_harness_contract.py @@ -0,0 +1,119 @@ +"""Guard the aggregate test-runner contract — the ENFORCEABLE version. + +The real enforcement is a PREFLIGHT in run_tests.py: before importing any test_*.py it +calls _preflight.check_test_files(), which refuses any sys.exit / raise SystemExit that +is not inside a standalone `if __name__ == "__main__":` guard body. That runs first, so an +offender cannot sort ahead of the check and end the process during import. + +This module proves the scanner behind that preflight: it exercises the location invariant +(no exemption for function/lambda/class bodies — an immediately-invoked helper defeats any +call-graph exemption), checks the live tests directory, and — the load-bearing regression — +builds a throwaway tests directory whose EARLIEST-sorting file exits at import and confirms +the preflight reports it (so the runner would return failure, not silent success). +""" + +from __future__ import annotations + +import ast +import os +import tempfile + +from _preflight import check_test_files, exit_violations + +failures: list[str] = [] +checks = 0 + +_HERE = os.path.dirname(os.path.abspath(__file__)) + +# MUST be reported as violations — including exits buried in a function / lambda / class +# body, because a module-scope call (or an IIFE) runs them at import. +_MUST_CATCH = ( + ("bare module-scope exit", "import sys\nsys.exit(0)\n"), + ("exit in `if True:`", "import sys\nif True:\n sys.exit(0)\n"), + ("exit in a top-level try", + "import sys\ntry:\n sys.exit(0)\nexcept Exception:\n pass\n"), + ("raise SystemExit at module scope", "raise SystemExit(1)\n"), + ("exit in a for loop", "import sys\nfor _ in range(1):\n sys.exit(0)\n"), + ("exit in a main-guard else", + "import sys\nif __name__ == '__main__':\n pass\nelse:\n sys.exit(0)\n"), + ("raise SystemExit in a main-guard else", + "if __name__ == '__main__':\n pass\nelse:\n raise SystemExit(1)\n"), + ("exit in a main-guard elif", + "import sys\nif __name__ == '__main__':\n pass\nelif True:\n sys.exit(0)\n"), + ("exit in a class body", "import sys\nclass C:\n sys.exit(7)\n"), + ("exit in a class base", "import sys\nclass C(sys.exit(7)):\n pass\n"), + ("exit in a class decorator", "import sys\n@sys.exit(7)\nclass C:\n pass\n"), + ("exit in a function default", "import sys\ndef f(value=sys.exit(7)):\n pass\n"), + ("exit in a function decorator", "import sys\n@sys.exit(7)\ndef f():\n pass\n"), + ("exit in a lambda default", "import sys\nf = lambda value=sys.exit(7): None\n"), + # The blocker-2 forms: a body that IS reached at import. + ("exit in a helper called at module scope", + "import sys\ndef abort():\n sys.exit(7)\nabort()\n"), + ("exit in an immediately-invoked lambda", "import sys\nv = (lambda: sys.exit(7))()\n"), + # And even an UNCALLED body — the strict location rule refuses it regardless. + ("exit in an uncalled function body", "import sys\ndef f():\n sys.exit(7)\n"), + ("exit in a class method body", + "import sys\nclass C:\n def m(self):\n sys.exit(7)\n"), +) +# MUST be accepted: only the `__main__` guard body, or no exit at all, or a string literal. +_MUST_PASS = ( + ("guarded entrypoint", "import sys\nif __name__ == '__main__':\n sys.exit(0)\n"), + ("guarded raise SystemExit(run())", + "def run():\n return 0\nif __name__ == '__main__':\n raise SystemExit(run())\n"), + ("guarded exit nested under an inner if", + "import sys\nif __name__ == '__main__':\n if '--x' in sys.argv:\n" + " raise SystemExit(0)\n"), + ("no exit at all", "def run():\n return 0\n"), + ("exit only as a string literal", "MSG = 'call sys.exit(0) to quit'\n"), +) + + +def _violates(src: str) -> bool: + return bool(exit_violations(ast.parse(src))) + + +def run() -> int: + global checks + for label, src in _MUST_CATCH: + checks += 1 + if not _violates(src): + failures.append(f"scanner self-test: failed to catch {label}") + for label, src in _MUST_PASS: + checks += 1 + if _violates(src): + failures.append(f"scanner self-test: wrongly flagged {label}") + + # The live tests directory must itself be clean (this is what the runner enforces). + checks += 1 + live = check_test_files(_HERE) + if live: + failures.append("live tests directory has import-time-exit violations: " + + "; ".join(live)) + + # The load-bearing regression: an offender that sorts BEFORE any policing module must be + # caught by the preflight, so the runner returns failure instead of a silent green. Also + # a helper invoked at module scope — the blocker-2 case call-graph pruning would miss. + checks += 1 + with tempfile.TemporaryDirectory() as tmp: + with open(os.path.join(tmp, "test_aaa_offender.py"), "w", encoding="utf-8") as fh: + fh.write("import sys\nsys.exit(0)\n") + with open(os.path.join(tmp, "test_zzz_helper.py"), "w", encoding="utf-8") as fh: + fh.write("import sys\n\n\ndef _boom():\n sys.exit(0)\n\n\n_boom()\n") + with open(os.path.join(tmp, "test_mmm_clean.py"), "w", encoding="utf-8") as fh: + fh.write("def run():\n return 0\n\n\nif __name__ == '__main__':\n" + " raise SystemExit(run())\n") + found = check_test_files(tmp) + offenders = {p.split(":", 1)[0] for p in found} + if offenders != {"test_aaa_offender.py", "test_zzz_helper.py"}: + failures.append(f"preflight regression: flagged {sorted(offenders)}, expected " + "the bare-exit and helper-invoked offenders only " + "(the guarded module must pass)") + + print(f"harness contract: {checks - len(failures)}/{checks} checks pass") + for f in failures: + print(f" FAIL: {f}") + return 1 if failures else 0 + + +if __name__ == "__main__": + raise SystemExit(run()) diff --git a/tests/test_patch_bundle.py b/tests/test_patch_bundle.py index 983478f4..941752f9 100644 --- a/tests/test_patch_bundle.py +++ b/tests/test_patch_bundle.py @@ -292,7 +292,14 @@ def validate(wd: str) -> bytes: pass # the CLI maps this to `own-fix: refuse:` + exit 2, never a traceback checks += 1 -print(f"patch bundle (S2 step 8): {checks - len(failures)}/{checks} checks pass") -for f in failures: - print(f" FAIL: {f}") -sys.exit(1 if failures else 0) +def run() -> int: + """The aggregate contract run_tests.py expects: report + an int rc, NEVER a + process-ending sys.exit at import time (which would silence every later module).""" + print(f"patch bundle (S2 step 8): {checks - len(failures)}/{checks} checks pass") + for f in failures: + print(f" FAIL: {f}") + return 1 if failures else 0 + + +if __name__ == "__main__": + raise SystemExit(run())