From fbe88bfe98da895cd4f09228ecbe189216a9c7d0 Mon Sep 17 00:00:00 2001 From: PhysShell Date: Fri, 17 Jul 2026 08:39:21 +0500 Subject: [PATCH 1/6] =?UTF-8?q?feat(S2):=20step=209=20=E2=80=94=20the=20st?= =?UTF-8?q?ructural=20self-gate=20over=20a=20canonical=20step=208=20bundle?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 host Git applies it to a pristine preimage in a hermetic throwaway repository. It never touches the real checkout, index or config, runs no model, no o7, no analyzer, no target tests, and publishes ONE byte-deterministic gate-result.json. Authority model (locked amendment 1): plan + candidates are REQUIRED and the gate calls its OWN pure validator, NOT the step 8 validate_apply_inputs() — that function reads the source itself, which would make two reads and a TOCTOU against the single-snapshot contract. The manifest is a claim, re-derived from the plan (actions, in candidate order) and the hash-bound candidates (identity) and required to EQUAL the canonical projection byte for byte; the patch, postimage and preimage are re-hashed over bytes read exactly once. The strict authority validator (amendment 2) restates the FROZEN envelope with exact key sets on EVERY object — including each candidate — the permission tiering (convert_acquire only for inotify_property_changed), one type / one file, and decision order == candidate order, because a hash proves only that two files agree with each other, never that they obey policy. The one snapshot boundary (amendment 3): the six external byte inputs are each read once — reject symlink/reparse, open O_NOFOLLOW where available, fstat the handle, require a regular file — and every hash / parse / materialization runs over those memory bytes. The throwaway repo is built from the SAME preimage bytes that were hashed, so there is no second read and no TOCTOU between the pre-SHA check and the tree. The patch parser (amendment 7) accepts EXACTLY the step 8 language: one file header whose three paths are byte-equal to the manifest's canonical rel (so a quoted/escaped or alternate path never matches), LF-terminated records, hunks whose arithmetic is consistent and whose old ranges lie in the preimage, context/`-`/`+` lines and the no-newline marker — and nothing else. index/mode/rename/copy/binary/submodule/new/deleted/second-file records are refused by construction, not by a blocklist. The hermetic repo (amendments 5 & 6): git init → write the preimage → `git add` (a BASELINE index, so `git diff-files` can prove exactly rel changed) → `git apply --check` → `git apply`, all with a minimal ALLOWLIST env (no inherited GIT_*; GIT_CONFIG_NOSYSTEM, GIT_ATTR_NOSYSTEM, empty HOME/XDG/global config) and `-c core.autocrlf=false -c core.eol=lf -c core.safecrlf=false`, the patch fed via stdin from the memory snapshot. After apply: diff == rel only, no untracked file, a raw worktree walk == {rel}, the index and config byte-unchanged, and the applied bytes == postimage == manifest.post_sha256. There is deliberately NO public --git override (amendment 4) — evidence of an INDEPENDENT apply cannot come from a caller-supplied stand-in; a missing git is INFRASTRUCTURE / exit 2. An empty patch (manual_review-only) is valid, not a refusal: pre == post == the pristine bytes, Git is not run, and git_apply_check / git_apply / isolated_tree record "not_applicable". Publication reuses the step 8 protocol: claimed unpredictable staging under a physically-off-tree parent, one atomic rename; a refusal leaves no , no staging, and a byte-identical source tree. Failure taxonomy (all exit 2, stable category for assertions): BUNDLE_LAYOUT, MANIFEST_SHAPE, AUTHORITY_BINDING, HASH_MISMATCH, PRISTINE_SOURCE, PATCH_STRUCTURE, APPLY_CHECK, APPLY_MISMATCH, ISOLATION, PUBLICATION, INFRASTRUCTURE. Tests: tests/test_gate_patch.py (46 checks — the patch grammar, manifest shape, authority validator and every byte-tampering case; dotnet-free, runs in the `tests` job). tests/gate_regressions.sh + a CI step drive the real chain and the filesystem/git-real cases: happy path, determinism, manual-only, and — via forged fixtures REBOUND so the refusal reaches the branch under test — APPLY_CHECK vs APPLY_MISMATCH vs HASH_MISMATCH, a symlink/extra bundle entry, a stale/symlink-escaping pristine source, and the publication refusals, with the real checkout never touched. _resolve_source ruling: Step 9 does NOT reach S0's case-sensitive _resolve_source (it does not call validate_apply_inputs); it uses its own platform-aware physical verifier. That S0 finding stays a separate maintenance scope. Steps 10-12 NOT started: no analyzer delta, no own-check/extractor re-run, no OWN001/OWN050 assertions, no fake target gate, no gate.toml, no 007 evidence, no STS clone. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JxKjqdGEFzq4UzZupw379G --- .github/workflows/ci.yml | 10 + ownlang/__main__.py | 38 +- ownlang/fix_gate.py | 974 ++++++++++++++++++++++++++++++++++++++ tests/gate_regressions.sh | 248 ++++++++++ tests/test_gate_patch.py | 345 ++++++++++++++ 5 files changed, 1614 insertions(+), 1 deletion(-) create mode 100644 ownlang/fix_gate.py create mode 100644 tests/gate_regressions.sh create mode 100644 tests/test_gate_patch.py 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..d9cffd4f --- /dev/null +++ b/ownlang/fix_gate.py @@ -0,0 +1,974 @@ +"""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 _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 = os.path.normcase(os.path.normpath(parent)) + c = os.path.normcase(os.path.normpath(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 _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) + for k in ("event", "source", "handler", "source_identity", "source_identity_kind", + "handler_identity", "handler_identity_kind"): + _s(c, k, 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 = _obj(c.get("acquire_span"), cat, f"{where}.acquire_span") + _exact(span, cat, f"{where}.acquire_span", *_SPAN_KEYS) + span_by_id[fid] = tuple(_int(span, k, cat, f"{where}.acquire_span") + for k in _SPAN_KEYS) + 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") + d_span = _obj(d.get("acquire_span"), cat, f"{where}.acquire_span") + _exact(d_span, cat, f"{where}.acquire_span", *_SPAN_KEYS) + got = tuple(_int(d_span, k, cat, f"{where}.acquire_span") for k in _SPAN_KEYS) + if got != 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 + while i < len(recs): + 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}") + 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) + if old_start < prev_old_end: + raise GateError(PATCH_STRUCTURE, "patch: hunks not increasing / non-overlapping") + if old_len > 0 and old_start + old_len - 1 > pre_lines: + raise GateError(PATCH_STRUCTURE, "patch: a hunk range is outside the preimage") + prev_old_end = old_start + old_len + saw_hunk = True + i += 1 + ctx = minus = plus = 0 + body_lines = 0 + 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": + if body_lines == 0: + raise GateError(PATCH_STRUCTURE, "patch: no-newline marker with no line") + 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] + if head == b" ": + ctx += 1 + elif head == b"-": + minus += 1 + else: + plus += 1 + body_lines += 1 + 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 _prepare_out(out: str, root: str) -> tuple[str, str, str]: + """(out_phys, workdir, staging). The out-dir must be fresh and PHYSICALLY off the + source tree; the workdir (holding staging + the throwaway repo) is claimed under the + verified physical parent with an unpredictable name.""" + 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") + workdir = os.path.join(parent_phys, f".{name}.owen-gate-{os.urandom(16).hex()}") + if os.path.exists(workdir) or os.path.islink(workdir): + raise GateError(PUBLICATION, "the work directory already exists") + return out_phys, workdir, os.path.join(workdir, "staging") + + +def _publish(staging: str, out_phys: str, evidence: bytes) -> None: + os.makedirs(staging) + with open(os.path.join(staging, "gate-result.json"), "wb") as fh: + fh.write(evidence) + 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_regular(root: str, what: 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 `/`-joined file paths.""" + leaves: 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) + if _is_link(st): + raise GateError(BUNDLE_LAYOUT, f"{what}: '{entry.name}' is a symlink/reparse") + if stat.S_ISDIR(st.st_mode): + stack.append(entry.path) + elif stat.S_ISREG(st.st_mode): + leaves.add(os.path.relpath(entry.path, root).replace("\\", "/")) + else: + raise GateError(BUNDLE_LAYOUT, + f"{what}: '{entry.name}' is not a regular file or dir") + return leaves + + +# --- 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 — entry types + top-level set; collect the postimage leaves. + bundle_phys = os.path.realpath(bundle) + if not os.path.isdir(bundle_phys): + raise GateError(BUNDLE_LAYOUT, "--bundle is not a directory") + _require_top_level(bundle_phys) + postimage_root = os.path.join(bundle_phys, "postimage") + post_leaves = _walk_regular(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: exactly one postimage leaf, and it is rel. + if post_leaves != {rel}: + raise GateError(BUNDLE_LAYOUT, + f"postimage holds {sorted(post_leaves)}, want ['{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, workdir, staging = _prepare_out(out, root) + try: + os.makedirs(workdir) + 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) + except BaseException: + shutil.rmtree(workdir, ignore_errors=True) + raise + shutil.rmtree(workdir, ignore_errors=True) + return out_phys diff --git a/tests/gate_regressions.sh b/tests/gate_regressions.sh new file mode 100644 index 00000000..a8844693 --- /dev/null +++ b/tests/gate_regressions.sh @@ -0,0 +1,248 @@ +#!/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 + +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/test_gate_patch.py b/tests/test_gate_patch.py new file mode 100644 index 00000000..f74b8841 --- /dev/null +++ b/tests/test_gate_patch.py @@ -0,0 +1,345 @@ +"""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 sys +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, + _same_or_inside, + 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") + + +# --- 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") + + +# --- 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") + + +print(f"gate (S2 step 9): {checks - len(failures)}/{checks} checks pass") +for f in failures: + print(f" FAIL: {f}") +sys.exit(1 if failures else 0) From ea47d4538761ba573df9075b1982bbda4e36c584 Mon Sep 17 00:00:00 2001 From: PhysShell Date: Fri, 17 Jul 2026 09:12:08 +0500 Subject: [PATCH 2/6] =?UTF-8?q?fix(S2):=20step=209=20=E2=80=94=20the=20fiv?= =?UTF-8?q?e=20HOLD=20blockers=20(and=20a=20latent=20step-8=20test-harness?= =?UTF-8?q?=20defect)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. THE TEST MODULE KILLED THE AGGREGATE RUNNER. test_gate_patch.py ran its checks at import and ended with sys.exit(), so when run_tests.py imported it every module discovered after it — and the aggregate return code — was silently dropped: a green job that had "turned the lights off in the building". The SAME latent defect was already in the merged step-8 test_patch_bundle.py (it sorts AFTER test_gate_patch, so it had never actually run in the aggregate). Both now expose `def run() -> int` with no import-time sys.exit and an `if __name__ == "__main__"` guard. A new tests/test_harness_contract.py STATICALLY proves (via ast) that every sibling test_*.py exposes run() and never calls sys.exit / raise SystemExit at module scope — so this class of bug cannot recur. With the fix the suite now reaches its summary and both modules gate: step 8 = 79 checks, step 9 = 63, harness = 29, all under one run_tests return code (verified: an injected failure flips rc to 1). 2. FULL FROZEN-CANDIDATE VALIDATION. The exact key set was held but not every value. The gate now validates every string field (diagnostic_code, enclosing_member, event_identity, ...), occurrence_ordinal as a non-negative int, and the full teardown block — exact keys, status in {none,exact,ambiguous}, and each teardown candidate's exact keys with its 6-int span. A re-hashed bundle with `event_identity: 123` or a malformed teardown now refuses at AUTHORITY_BINDING (fixtures added for event_identity/occurrence_ordinal wrong type, teardown extra key / unknown status / malformed candidate / bad span shape). 3. EXACT DIRECTORY LAYOUT + BUNDLE-ROOT lstat. The layout walk returned only files, so an extra/hidden/nested EMPTY directory rode through, and --bundle was realpath'd first so a symlinked bundle root was accepted. The walk now returns (dirs, files) and the postimage subtree must equal EXACTLY rel's ancestor dirs plus rel; --bundle is lstat'd before realpath and a symlink/reparse root is refused. Fixtures: extra/hidden/nested empty dir, symlinked bundle root — all BUNDLE_LAYOUT. 4. A TRULY CLAIMED WORKDIR + RE-PROVEN PUBLICATION. _prepare_out only checked a name; the dir was created later with a plain makedirs. A new _claim_workdir creates the unpredictable dir immediately (mode 0700 on POSIX, in the mkdir itself) and PROVES it: not a link/reparse, realpath == itself under the platform-aware comparison, empty. And _publish re-resolves the out-dir parent, re-confirms it is off the source tree, and re-confirms no final out exists — immediately before the atomic rename, not only at parse time. 5. TIGHTER PATCH GRAMMAR. The no-newline marker was allowed after any seen body line and any number of times. It is now stateful: it must sit immediately after an eligible ( / - / + line, marks the LAST line of its side(s) (a later line of a closed side is refused), and appears at most once per side. Range bounds now cover the zero-length (insertion) case: old_len>0 requires 1 <= old_start and old_start+old_len-1 <= preimage lines (so a non-zero range starting at 0 is refused); old_len==0 requires 0 <= old_start <= preimage lines (so an insertion past EOF is refused rather than dying later as APPLY_CHECK). Fixtures: marker before any line, duplicate marker, marker with more of the same side after it, insertion past the preimage, non-zero range at 0, plus positive cases (a well-placed marker, a top insertion). The step-8 change is strictly the test-harness run() wrapper — no test logic, no Steps 4-8 semantics/schema touched. All other Step 9 architecture (authority model, six snapshots, manifest schema, git environment, baseline index, evidence schema, empty-patch policy, taxonomy) is unchanged. Steps 10-12 not started. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JxKjqdGEFzq4UzZupw379G --- ownlang/fix_gate.py | 178 +++++++++++++++++++++++++-------- tests/gate_regressions.sh | 10 ++ tests/test_gate_patch.py | 84 +++++++++++++++- tests/test_harness_contract.py | 73 ++++++++++++++ tests/test_patch_bundle.py | 15 ++- 5 files changed, 309 insertions(+), 51 deletions(-) create mode 100644 tests/test_harness_contract.py diff --git a/ownlang/fix_gate.py b/ownlang/fix_gate.py index d9cffd4f..2a9007fe 100644 --- a/ownlang/fix_gate.py +++ b/ownlang/fix_gate.py @@ -248,6 +248,29 @@ def _bundle_sha256(candidates: dict[str, Any]) -> str: 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") @@ -317,9 +340,14 @@ def _validate_candidates(bundle: dict[str, Any], cat: str) -> dict[str, Any]: raise GateError(cat, f"{where}: duplicate finding_id {fid}") id_set.add(fid) ids.append(fid) - for k in ("event", "source", "handler", "source_identity", "source_identity_kind", - "handler_identity", "handler_identity_kind"): + # 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: @@ -327,10 +355,7 @@ def _validate_candidates(bundle: dict[str, Any], cat: str) -> dict[str, Any]: contract = _s(c, "event_contract", cat, where) if contract not in _CONTRACTS: raise GateError(cat, f"{where}: unknown event_contract '{contract}'") - span = _obj(c.get("acquire_span"), cat, f"{where}.acquire_span") - _exact(span, cat, f"{where}.acquire_span", *_SPAN_KEYS) - span_by_id[fid] = tuple(_int(span, k, cat, f"{where}.acquire_span") - for k in _SPAN_KEYS) + 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") @@ -429,10 +454,7 @@ def validate_gate_authority(validated_plan: Any, candidates: Any) -> GateAuthori 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") - d_span = _obj(d.get("acquire_span"), cat, f"{where}.acquire_span") - _exact(d_span, cat, f"{where}.acquire_span", *_SPAN_KEYS) - got = tuple(_int(d_span, k, cat, f"{where}.acquire_span") for k in _SPAN_KEYS) - if got != facts["span_by_id"][fid]: + 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) @@ -561,33 +583,56 @@ def parse_step8_patch(patch: bytes, rel: str, preimage: bytes) -> None: 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") - if old_len > 0 and old_start + old_len - 1 > pre_lines: - raise GateError(PATCH_STRUCTURE, "patch: a hunk range is outside the preimage") prev_old_end = old_start + old_len saw_hunk = True i += 1 ctx = minus = plus = 0 - body_lines = 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": - if body_lines == 0: - raise GateError(PATCH_STRUCTURE, "patch: no-newline marker with no line") + # 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 + 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 - body_lines += 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") @@ -758,10 +803,10 @@ def apply_in_throwaway(workdir: str, rel: str, preimage: bytes, postimage: bytes # --- publication (the step 8 protocol, reused) ------------------------------------- -def _prepare_out(out: str, root: str) -> tuple[str, str, str]: - """(out_phys, workdir, staging). The out-dir must be fresh and PHYSICALLY off the - source tree; the workdir (holding staging + the throwaway repo) is claimed under the - verified physical parent with an unpredictable name.""" +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: @@ -776,16 +821,50 @@ def _prepare_out(out: str, root: str) -> tuple[str, str, str]: 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") - workdir = os.path.join(parent_phys, f".{name}.owen-gate-{os.urandom(16).hex()}") - if os.path.exists(workdir) or os.path.islink(workdir): - raise GateError(PUBLICATION, "the work directory already exists") - return out_phys, workdir, os.path.join(workdir, "staging") - - -def _publish(staging: str, out_phys: str, evidence: bytes) -> None: + 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.""" + 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 + 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 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") + 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 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: @@ -822,10 +901,13 @@ def _require_top_level(bundle: str) -> None: raise GateError(BUNDLE_LAYOUT, "postimage: is not a real directory") -def _walk_regular(root: str, what: str) -> set[str]: +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 `/`-joined file paths.""" - leaves: set[str] = set() + 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() @@ -835,16 +917,18 @@ def _walk_regular(root: str, what: str) -> set[str]: 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): - leaves.add(os.path.relpath(entry.path, root).replace("\\", "/")) + files.add(rel) else: raise GateError(BUNDLE_LAYOUT, f"{what}: '{entry.name}' is not a regular file or dir") - return leaves + return dirs, files # --- evidence + orchestration ------------------------------------------------------ @@ -878,13 +962,17 @@ def run_gate(bundle: str, plan_path: str, candidates_path: str, root: str, out: path; raises GateError (no output, source untouched) on any refusal.""" import shutil - # [1] bundle layout — entry types + top-level set; collect the postimage leaves. - bundle_phys = os.path.realpath(bundle) - if not os.path.isdir(bundle_phys): + # [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_leaves = _walk_regular(postimage_root, "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"), @@ -898,10 +986,13 @@ def run_gate(bundle: str, plan_path: str, candidates_path: str, root: str, out: 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: exactly one postimage leaf, and it is rel. - if post_leaves != {rel}: - raise GateError(BUNDLE_LAYOUT, - f"postimage holds {sorted(post_leaves)}, want ['{rel}']") + # [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}") @@ -949,9 +1040,10 @@ def run_gate(bundle: str, plan_path: str, candidates_path: str, root: str, out: parse_step8_patch(patch_bytes, rel, preimage_bytes) # [10] apply semantics + publication. - out_phys, workdir, staging = _prepare_out(out, root) + out_phys, parent_phys, root_phys = _out_parent(out, root) + workdir = _claim_workdir(parent_phys) + staging = os.path.join(workdir, "staging") try: - os.makedirs(workdir) if empty_patch: # manual-only: pre == post == the pristine bytes; Git is not run. if postimage_bytes != preimage_bytes: @@ -966,7 +1058,7 @@ def run_gate(bundle: str, plan_path: str, candidates_path: str, root: str, out: evidence = _build_evidence(auth, rel, plan_bytes, manifest_bytes, patch_bytes, m_pre, m_post, git_gates) - _publish(staging, out_phys, evidence) + _publish(staging, out_phys, evidence, root_phys) except BaseException: shutil.rmtree(workdir, ignore_errors=True) raise diff --git a/tests/gate_regressions.sh b/tests/gate_regressions.sh index a8844693..96fa388c 100644 --- a/tests/gate_regressions.sh +++ b/tests/gate_regressions.sh @@ -178,6 +178,16 @@ 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)) diff --git a/tests/test_gate_patch.py b/tests/test_gate_patch.py index f74b8841..3627f493 100644 --- a/tests/test_gate_patch.py +++ b/tests/test_gate_patch.py @@ -211,6 +211,34 @@ def base_no_convert() -> dict[str, Any]: 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 ---------------------------------------------------------------- @@ -326,6 +354,47 @@ def variant(replace_pairs: list[tuple[bytes, bytes]], insert_after: bytes = b"", 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. +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 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") + # --- containment platform rule ----------------------------------------------------- @@ -339,7 +408,14 @@ def variant(replace_pairs: list[tuple[bytes, bytes]], insert_after: bytes = b"", check(not _same_or_inside("/repo", "/REPO/x"), "containment: POSIX case-sensitive") -print(f"gate (S2 step 9): {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"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..a795cea9 --- /dev/null +++ b/tests/test_harness_contract.py @@ -0,0 +1,73 @@ +"""Guard the aggregate test-runner contract itself. + +The bug this pins down: a `test_*.py` that executes its checks at import time and ends with +`sys.exit(...)` ends the WHOLE process when run_tests.py imports it, silently dropping every +module discovered after it and the aggregate return code. This module statically proves that +CANNOT happen — every sibling `test_*.py` exposes `run()` and never calls sys.exit / +raise SystemExit at module scope (only under an `if __name__ == "__main__"` guard). +""" + +from __future__ import annotations + +import ast +import os + +failures: list[str] = [] +checks = 0 + +_HERE = os.path.dirname(os.path.abspath(__file__)) + + +def _module_scope_exit(tree: ast.AST) -> bool: + """True if a sys.exit(...) / raise SystemExit(...) sits at module scope (i.e. would + fire on import) rather than inside a function or an `if __name__ == '__main__'` guard.""" + for node in ast.iter_child_nodes(tree): + if isinstance(node, ast.If): + continue # the `if __name__ == "__main__"` entrypoint is fine + for sub in ast.walk(node): + if isinstance(sub, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + # do not descend into nested scopes for THIS module-scope check + break + if isinstance(sub, ast.Raise) and _is_systemexit(sub.exc): + return True + if isinstance(sub, ast.Call) and _is_sys_exit(sub.func): + return True + return False + + +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 run() -> int: + global checks + for fname in sorted(os.listdir(_HERE)): + if not (fname.startswith("test_") and fname.endswith(".py")): + continue + if fname == os.path.basename(__file__): + continue + checks += 1 + src = open(os.path.join(_HERE, fname), encoding="utf-8").read() + tree = ast.parse(src) + has_run = any(isinstance(n, ast.FunctionDef) and n.name == "run" + for n in ast.iter_child_nodes(tree)) + if not has_run: + failures.append(f"{fname}: has no module-level run()") + if _module_scope_exit(tree): + failures.append(f"{fname}: calls sys.exit/raise SystemExit at import scope " + "(would short-circuit the aggregate runner)") + print(f"harness contract: {checks - len(failures)}/{checks} test modules honour run()") + 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()) From 5b666c60448cd321be4ab17c5efab08684eb0e07 Mon Sep 17 00:00:00 2001 From: PhysShell Date: Fri, 17 Jul 2026 09:32:46 +0500 Subject: [PATCH 3/6] =?UTF-8?q?fix(S2):=20step=209=20=E2=80=94=20the=20thr?= =?UTF-8?q?ee=20narrow=20HOLD=20defects?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. THE HARNESS GUARD WAS TOO LOOSE. test_harness_contract.py skipped ANY top-level `if` (so `if True: sys.exit(0)` slid through) and broke traversal at the first nested FunctionDef (so a module-scope exit inside a top-level `try` was missed). The guard is rewritten as a proper recursive walk: it skips ONLY a structurally-recognised `__name__ == "__main__"` (a real ast.Compare of __name__ == "__main__"), does NOT descend into function / class / lambda scopes, and DOES descend through module-scope compound statements (if / try / with / for / while) — so an exit inside `if True:` or a top-level `try:` is caught. Added 8 guard self-tests: five that MUST be caught (bare exit, `if True:` exit, exit in a top-level try, `raise SystemExit`, exit in a for loop) and three that MUST pass (the guarded entrypoint, an exit inside a function, an exit inside a lambda). 2. NO-NEWLINE STATE WAS PER-HUNK. old_closed/new_closed reset each hunk, so a no-newline marker in the first hunk did not forbid a second hunk after it — the marker means EOF-without-newline, so nothing can follow on that side. State is now FILE-level (`saw_marker`): once any no-newline marker appears, a following hunk header is a PATCH_STRUCTURE refusal (previously it would have slipped to APPLY_CHECK — the right refusal via the wrong branch). Added a two-hunk fixture asserting PATCH_STRUCTURE. 3. CLAIM/PUBLICATION EQUALITY WAS NOT PLATFORM-AWARE, AND A POST-mkdir FAILURE LEAKED THE DIRECTORY. Both the workdir self-resolution proof and the publication parent re-proof used a raw string `realpath(x) != y`, which on case-insensitive Windows disagrees with the accepted _same_or_inside rule and could give a casing-only false refusal. A new platform-aware `_same_path` (normcase+normpath, the same rule) is now used in both. And `_claim_workdir` created the directory before the outer cleanup scope began, so a proof failure after the mkdir would strand it — the post-mkdir proofs are now wrapped so the just-created directory is removed on ANY failure before raising. Added a pure-function regression (symlink-capable platforms): a claim under a symlinked parent fails the self-resolution proof with PUBLICATION and leaves NO directory behind, plus _same_path platform cases. Nothing else in Step 9 reopened. Steps 10-12 not started. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JxKjqdGEFzq4UzZupw379G --- ownlang/fix_gate.py | 51 ++++++++++++++++++------- tests/test_gate_patch.py | 41 +++++++++++++++++++- tests/test_harness_contract.py | 69 +++++++++++++++++++++++++++------- 3 files changed, 132 insertions(+), 29 deletions(-) diff --git a/ownlang/fix_gate.py b/ownlang/fix_gate.py index 2a9007fe..cfbfd1f5 100644 --- a/ownlang/fix_gate.py +++ b/ownlang/fix_gate.py @@ -98,12 +98,23 @@ 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 = os.path.normcase(os.path.normpath(parent)) - c = os.path.normcase(os.path.normpath(path)) + p = _norm(parent) + c = _norm(path) if c == p: return True return c.startswith(p if p.endswith(os.sep) else p + os.sep) @@ -572,10 +583,13 @@ def parse_step8_patch(patch: bytes, rel: str, preimage: bytes) -> None: i = 3 prev_old_end = 0 saw_hunk = False - while i < len(recs): + 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) @@ -614,6 +628,7 @@ def parse_step8_patch(patch: bytes, rel: str, preimage: bytes) -> None: 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 @@ -827,8 +842,11 @@ def _out_parent(out: str, root: str) -> tuple[str, str, str]: 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.""" + 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): @@ -840,14 +858,19 @@ def _claim_workdir(parent_phys: str) -> str: except OSError as exc: raise GateError(PUBLICATION, f"cannot claim a work directory ({exc.strerror or exc})") from exc - 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 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") + 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") @@ -860,7 +883,7 @@ def _publish(staging: str, out_phys: str, evidence: bytes, root_phys: str) -> No parent = os.path.dirname(out_phys) if not os.path.isdir(parent): raise GateError(PUBLICATION, "the out-dir parent vanished before publication") - if os.path.realpath(parent) != os.path.dirname(out_phys) \ + 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): diff --git a/tests/test_gate_patch.py b/tests/test_gate_patch.py index 3627f493..22859ab4 100644 --- a/tests/test_gate_patch.py +++ b/tests/test_gate_patch.py @@ -13,7 +13,9 @@ 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__)))) @@ -24,7 +26,9 @@ PATCH_STRUCTURE, GateError, _bundle_sha256, + _claim_workdir, _same_or_inside, + _same_path, parse_step8_patch, validate_gate_authority, validate_manifest_shape, @@ -381,10 +385,16 @@ def small(hunk: bytes) -> bytes: 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. +# 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") @@ -396,6 +406,35 @@ def small(hunk: bytes) -> bytes: 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") diff --git a/tests/test_harness_contract.py b/tests/test_harness_contract.py index a795cea9..f75a16a2 100644 --- a/tests/test_harness_contract.py +++ b/tests/test_harness_contract.py @@ -18,20 +18,35 @@ _HERE = os.path.dirname(os.path.abspath(__file__)) -def _module_scope_exit(tree: ast.AST) -> bool: - """True if a sys.exit(...) / raise SystemExit(...) sits at module scope (i.e. would - fire on import) rather than inside a function or an `if __name__ == '__main__'` guard.""" - for node in ast.iter_child_nodes(tree): - if isinstance(node, ast.If): - continue # the `if __name__ == "__main__"` entrypoint is fine - for sub in ast.walk(node): - if isinstance(sub, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): - # do not descend into nested scopes for THIS module-scope check - break - if isinstance(sub, ast.Raise) and _is_systemexit(sub.exc): - return True - if isinstance(sub, ast.Call) and _is_sys_exit(sub.func): - return True +def _is_main_guard(test: ast.expr) -> bool: + """Structurally recognise EXACTLY `__name__ == "__main__"` — not any top-level `if`, + so `if True: sys.exit(0)` is NOT waved through.""" + 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 _module_scope_exit(node: ast.AST) -> bool: + """True if a sys.exit(...) / raise SystemExit(...) would fire at IMPORT time. Descends + through module-scope compound statements (if / try / with / for / while) but NOT into + a new scope (function / class / lambda) and NOT into the `if __name__ == "__main__"` + entrypoint — so an exit inside `if True:` or a top-level `try:` IS caught, while the + real guard and any function body are not.""" + for child in ast.iter_child_nodes(node): + if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, + ast.Lambda, ast.ClassDef)): + continue # a new scope: its body does not run on import + if isinstance(child, ast.If) and _is_main_guard(child.test): + continue # the sanctioned entrypoint + if isinstance(child, ast.Raise) and _is_systemexit(child.exc): + return True + if isinstance(child, ast.Call) and _is_sys_exit(child.func): + return True + if _module_scope_exit(child): + return True return False @@ -46,8 +61,34 @@ def _is_sys_exit(func: ast.expr) -> bool: and isinstance(func.value, ast.Name) and func.value.id == "sys") +_SELFTEST_MUST_CATCH = ( + ("bare sys.exit at module scope", "import sys\nsys.exit(0)\n"), + ("exit inside `if True:`", "import sys\nif True:\n sys.exit(0)\n"), + ("exit inside 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 inside a for loop", "import sys\nfor _ in range(1):\n sys.exit(0)\n"), +) +_SELFTEST_MUST_PASS = ( + ("guarded entrypoint", "import sys\nif __name__ == '__main__':\n sys.exit(0)\n"), + ("exit inside a function", "import sys\ndef run():\n sys.exit(0)\n"), + ("exit inside a lambda", "f = lambda: __import__('sys').exit(0)\n"), +) + + def run() -> int: global checks + # Self-test the guard first: it must catch the ways the original bug could recur, and + # must NOT flag the sanctioned entrypoint or an exit that only lives inside a scope. + for label, src in _SELFTEST_MUST_CATCH: + checks += 1 + if not _module_scope_exit(ast.parse(src)): + failures.append(f"guard self-test: failed to catch {label}") + for label, src in _SELFTEST_MUST_PASS: + checks += 1 + if _module_scope_exit(ast.parse(src)): + failures.append(f"guard self-test: wrongly flagged {label}") + for fname in sorted(os.listdir(_HERE)): if not (fname.startswith("test_") and fname.endswith(".py")): continue From 215f74a579fe33c579aad0ea329294d6823cfae4 Mon Sep 17 00:00:00 2001 From: PhysShell Date: Fri, 17 Jul 2026 09:47:01 +0500 Subject: [PATCH 4/6] fix(S2): the harness guard must walk a main-guard's else, not skip it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `if isinstance(child, ast.If) and _is_main_guard(child.test): continue` exempted the WHOLE guard, but its `else:` runs on import — so `if __name__ == '__main__': ... else: sys.exit(0)` slipped through. Only the guard's body is exempt now; its orelse is re-checked with the same per-child logic (via a synthetic Module, so a bare `raise SystemExit` there is caught as a statement). Added MUST_CATCH self-tests for sys.exit / raise SystemExit in a main-guard else and an elif. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JxKjqdGEFzq4UzZupw379G --- tests/test_harness_contract.py | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/tests/test_harness_contract.py b/tests/test_harness_contract.py index f75a16a2..870c6e8b 100644 --- a/tests/test_harness_contract.py +++ b/tests/test_harness_contract.py @@ -40,7 +40,15 @@ def _module_scope_exit(node: ast.AST) -> bool: ast.Lambda, ast.ClassDef)): continue # a new scope: its body does not run on import if isinstance(child, ast.If) and _is_main_guard(child.test): - continue # the sanctioned entrypoint + # ONLY the guard's body is exempt — its `else:` still runs on import, so a + # `if __name__ == "__main__": ... else: sys.exit(0)` must not slip through. + # Re-run the SAME per-child logic over the orelse statements (a synthetic + # Module so a bare `raise SystemExit` there is checked as a statement, not just + # its children). + if child.orelse and _module_scope_exit( + ast.Module(body=child.orelse, type_ignores=[])): + return True + continue if isinstance(child, ast.Raise) and _is_systemexit(child.exc): return True if isinstance(child, ast.Call) and _is_sys_exit(child.func): @@ -68,6 +76,12 @@ def _is_sys_exit(func: ast.expr) -> bool: "import sys\ntry:\n sys.exit(0)\nexcept Exception:\n pass\n"), ("raise SystemExit at module scope", "raise SystemExit(1)\n"), ("exit inside a for loop", "import sys\nfor _ in range(1):\n sys.exit(0)\n"), + ("sys.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"), ) _SELFTEST_MUST_PASS = ( ("guarded entrypoint", "import sys\nif __name__ == '__main__':\n sys.exit(0)\n"), From f1b26cdbaf9fd7ed36e41f205ed4d8112908e7dd Mon Sep 17 00:00:00 2001 From: PhysShell Date: Fri, 17 Jul 2026 10:03:08 +0500 Subject: [PATCH 5/6] fix(S2): the harness scanner must not skip whole class/function/lambda nodes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Skipping a FunctionDef/AsyncFunctionDef/Lambda/ClassDef subtree missed code that runs at IMPORT: a class body, class bases/keywords, decorators, argument defaults and lambda defaults all execute while the module is imported — any of them can sys.exit and abort run_tests' importlib.import_module before the aggregate rc is collected. The scanner now prunes ONLY the genuinely deferred subtrees (function/lambda BODY, the main-guard body), checks the node itself before its children (so a decorator/default that IS sys.exit is caught), scans class bodies normally, and models 'from __future__ import annotations' so eager annotations are checked but stringified ones are not. Added MUST_CATCH for class body / base / decorator, function default + decorator, keyword-only default, lambda default and an eager annotation; MUST_PASS for a class method body and a stringified annotation. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JxKjqdGEFzq4UzZupw379G --- tests/test_harness_contract.py | 89 ++++++++++++++++++++++++---------- 1 file changed, 64 insertions(+), 25 deletions(-) diff --git a/tests/test_harness_contract.py b/tests/test_harness_contract.py index 870c6e8b..2df43ea6 100644 --- a/tests/test_harness_contract.py +++ b/tests/test_harness_contract.py @@ -29,35 +29,59 @@ def _is_main_guard(test: ast.expr) -> bool: and test.comparators[0].value == "__main__") -def _module_scope_exit(node: ast.AST) -> bool: - """True if a sys.exit(...) / raise SystemExit(...) would fire at IMPORT time. Descends - through module-scope compound statements (if / try / with / for / while) but NOT into - a new scope (function / class / lambda) and NOT into the `if __name__ == "__main__"` - entrypoint — so an exit inside `if True:` or a top-level `try:` IS caught, while the - real guard and any function body are not.""" - for child in ast.iter_child_nodes(node): - if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, - ast.Lambda, ast.ClassDef)): - continue # a new scope: its body does not run on import - if isinstance(child, ast.If) and _is_main_guard(child.test): - # ONLY the guard's body is exempt — its `else:` still runs on import, so a - # `if __name__ == "__main__": ... else: sys.exit(0)` must not slip through. - # Re-run the SAME per-child logic over the orelse statements (a synthetic - # Module so a bare `raise SystemExit` there is checked as a statement, not just - # its children). - if child.orelse and _module_scope_exit( - ast.Module(body=child.orelse, type_ignores=[])): +def _future_annotations(module: ast.AST) -> bool: + body = getattr(module, "body", []) + for stmt in body: + if isinstance(stmt, ast.ImportFrom) and stmt.module == "__future__": + if any(alias.name == "annotations" for alias in stmt.names): return True - continue - if isinstance(child, ast.Raise) and _is_systemexit(child.exc): - return True - if isinstance(child, ast.Call) and _is_sys_exit(child.func): - return True - if _module_scope_exit(child): - return True return False +def _annotation_exprs(func: ast.FunctionDef | ast.AsyncFunctionDef) -> list[ast.expr]: + a = func.args + exprs = [arg.annotation for arg in + (*a.posonlyargs, *a.args, *a.kwonlyargs) if arg.annotation] + for extra in (a.vararg, a.kwarg): + if extra is not None and extra.annotation is not None: + exprs.append(extra.annotation) + if func.returns is not None: + exprs.append(func.returns) + return exprs + + +def _module_scope_exit(node: ast.AST, ann_eager: bool | None = None) -> bool: + """True if a sys.exit(...) / raise SystemExit(...) would fire at IMPORT time. Only the + genuinely DEFERRED subtrees are pruned: a function / async-function / lambda BODY, and + the `if __name__ == "__main__"` body. Everything else runs on import and is scanned — + class bodies + bases + keywords, decorators, argument defaults, lambda defaults, and + (unless `from __future__ import annotations` is in force) annotations. The node itself + is checked BEFORE its children, so a decorator / default that IS `sys.exit(...)` is + caught, not just its arguments.""" + if ann_eager is None: # decided once, at the module root, then threaded down + ann_eager = not _future_annotations(node) + + if isinstance(node, ast.Raise) and _is_systemexit(node.exc): + return True + if isinstance(node, ast.Call) and _is_sys_exit(node.func): + return True + if isinstance(node, ast.If) and _is_main_guard(node.test): + # ONLY the guard's body is exempt — its `else:` (and `elif`) still run on import. + return any(_module_scope_exit(s, ann_eager) for s in node.orelse) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + eager = [*node.decorator_list, *node.args.defaults, + *[d for d in node.args.kw_defaults if d is not None]] + if ann_eager: + eager += _annotation_exprs(node) + return any(_module_scope_exit(e, ann_eager) for e in eager) + if isinstance(node, ast.Lambda): + eager = [*node.args.defaults, *[d for d in node.args.kw_defaults if d is not None]] + return any(_module_scope_exit(e, ann_eager) for e in eager) + # Everything else — module body, class body/bases/keywords/decorators, module-scope + # control flow — executes on import; scan every child. + return any(_module_scope_exit(c, ann_eager) for c in ast.iter_child_nodes(node)) + + def _is_systemexit(exc: ast.expr | None) -> bool: if isinstance(exc, ast.Call): exc = exc.func @@ -82,11 +106,26 @@ def _is_sys_exit(func: ast.expr) -> bool: "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"), + # Declarations that run code AT IMPORT — the trapdoors that skipping a whole + # FunctionDef / ClassDef / Lambda node would miss. + ("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 keyword-only default", + "import sys\ndef f(*, value=sys.exit(7)):\n pass\n"), + ("exit in a lambda default", "import sys\nf = lambda value=sys.exit(7): None\n"), + ("exit in an eager annotation", "import sys\ndef f(x: sys.exit(7)):\n pass\n"), ) _SELFTEST_MUST_PASS = ( ("guarded entrypoint", "import sys\nif __name__ == '__main__':\n sys.exit(0)\n"), ("exit inside a function", "import sys\ndef run():\n sys.exit(0)\n"), ("exit inside a lambda", "f = lambda: __import__('sys').exit(0)\n"), + ("exit inside a class method body", + "import sys\nclass C:\n def m(self):\n sys.exit(0)\n"), + ("stringified annotation under future-annotations", + "from __future__ import annotations\nimport sys\ndef f(x: sys.exit(7)):\n pass\n"), ) From d1487c06cd211d79eef2193eeea2e401a19d0196 Mon Sep 17 00:00:00 2001 From: PhysShell Date: Fri, 17 Jul 2026 11:09:55 +0500 Subject: [PATCH 6/6] fix(S2): enforce the no-import-time-exit contract as a real preflight MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two unsoundnesses in the harness guard, both fixed: (1) The guard ran too late. test_harness_contract.py was itself an auto-discovered module, so an earlier-sorting test_*.py that sys.exit()s at import ended the process before the scan ran. The check is now a PREFLIGHT (tests/_preflight.py) that run_tests.run() calls FIRST, over the source, before importing any test module — so an offender cannot sort ahead of it. A violation prints PREFLIGHT FAIL and the runner returns 1. (2) Pruning function/lambda/class bodies was unsound: a body is reached at import when a module-scope call or an immediately-invoked lambda runs it. The invariant is now strict and purely LOCATION-based (no call-graph analysis): sys.exit / raise SystemExit is allowed ONLY inside a standalone guard body — nowhere else, function bodies included. Regressions: MUST_CATCH now includes a helper called at module scope, an IIFE, an uncalled function body and a class method body; the MUST_PASS cases that permitted function-body exits are removed. A load-bearing regression builds a throwaway tests dir whose earliest-sorting file exits at import and confirms the preflight reports it (plus a helper-invoked offender and a clean guarded module). Verified end to end: injecting an earlier-sorting offender into the real tests dir makes run_tests return 1, not a silent 0. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JxKjqdGEFzq4UzZupw379G --- tests/_preflight.py | 93 +++++++++++++++ tests/run_tests.py | 12 ++ tests/test_harness_contract.py | 204 +++++++++++++-------------------- 3 files changed, 183 insertions(+), 126 deletions(-) create mode 100644 tests/_preflight.py 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/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_harness_contract.py b/tests/test_harness_contract.py index 2df43ea6..ef1de828 100644 --- a/tests/test_harness_contract.py +++ b/tests/test_harness_contract.py @@ -1,163 +1,115 @@ -"""Guard the aggregate test-runner contract itself. - -The bug this pins down: a `test_*.py` that executes its checks at import time and ends with -`sys.exit(...)` ends the WHOLE process when run_tests.py imports it, silently dropping every -module discovered after it and the aggregate return code. This module statically proves that -CANNOT happen — every sibling `test_*.py` exposes `run()` and never calls sys.exit / -raise SystemExit at module scope (only under an `if __name__ == "__main__"` guard). +"""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__)) - -def _is_main_guard(test: ast.expr) -> bool: - """Structurally recognise EXACTLY `__name__ == "__main__"` — not any top-level `if`, - so `if True: sys.exit(0)` is NOT waved through.""" - 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 _future_annotations(module: ast.AST) -> bool: - body = getattr(module, "body", []) - for stmt in body: - if isinstance(stmt, ast.ImportFrom) and stmt.module == "__future__": - if any(alias.name == "annotations" for alias in stmt.names): - return True - return False - - -def _annotation_exprs(func: ast.FunctionDef | ast.AsyncFunctionDef) -> list[ast.expr]: - a = func.args - exprs = [arg.annotation for arg in - (*a.posonlyargs, *a.args, *a.kwonlyargs) if arg.annotation] - for extra in (a.vararg, a.kwarg): - if extra is not None and extra.annotation is not None: - exprs.append(extra.annotation) - if func.returns is not None: - exprs.append(func.returns) - return exprs - - -def _module_scope_exit(node: ast.AST, ann_eager: bool | None = None) -> bool: - """True if a sys.exit(...) / raise SystemExit(...) would fire at IMPORT time. Only the - genuinely DEFERRED subtrees are pruned: a function / async-function / lambda BODY, and - the `if __name__ == "__main__"` body. Everything else runs on import and is scanned — - class bodies + bases + keywords, decorators, argument defaults, lambda defaults, and - (unless `from __future__ import annotations` is in force) annotations. The node itself - is checked BEFORE its children, so a decorator / default that IS `sys.exit(...)` is - caught, not just its arguments.""" - if ann_eager is None: # decided once, at the module root, then threaded down - ann_eager = not _future_annotations(node) - - if isinstance(node, ast.Raise) and _is_systemexit(node.exc): - return True - if isinstance(node, ast.Call) and _is_sys_exit(node.func): - return True - if isinstance(node, ast.If) and _is_main_guard(node.test): - # ONLY the guard's body is exempt — its `else:` (and `elif`) still run on import. - return any(_module_scope_exit(s, ann_eager) for s in node.orelse) - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): - eager = [*node.decorator_list, *node.args.defaults, - *[d for d in node.args.kw_defaults if d is not None]] - if ann_eager: - eager += _annotation_exprs(node) - return any(_module_scope_exit(e, ann_eager) for e in eager) - if isinstance(node, ast.Lambda): - eager = [*node.args.defaults, *[d for d in node.args.kw_defaults if d is not None]] - return any(_module_scope_exit(e, ann_eager) for e in eager) - # Everything else — module body, class body/bases/keywords/decorators, module-scope - # control flow — executes on import; scan every child. - return any(_module_scope_exit(c, ann_eager) for c in ast.iter_child_nodes(node)) - - -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") - - -_SELFTEST_MUST_CATCH = ( - ("bare sys.exit at module scope", "import sys\nsys.exit(0)\n"), - ("exit inside `if True:`", "import sys\nif True:\n sys.exit(0)\n"), - ("exit inside a top-level try", +# 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 inside a for loop", "import sys\nfor _ in range(1):\n sys.exit(0)\n"), - ("sys.exit in a main-guard else", + ("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"), - # Declarations that run code AT IMPORT — the trapdoors that skipping a whole - # FunctionDef / ClassDef / Lambda node would miss. ("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 keyword-only default", - "import sys\ndef f(*, value=sys.exit(7)):\n pass\n"), ("exit in a lambda default", "import sys\nf = lambda value=sys.exit(7): None\n"), - ("exit in an eager annotation", "import sys\ndef f(x: sys.exit(7)):\n pass\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"), ) -_SELFTEST_MUST_PASS = ( +# 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"), - ("exit inside a function", "import sys\ndef run():\n sys.exit(0)\n"), - ("exit inside a lambda", "f = lambda: __import__('sys').exit(0)\n"), - ("exit inside a class method body", - "import sys\nclass C:\n def m(self):\n sys.exit(0)\n"), - ("stringified annotation under future-annotations", - "from __future__ import annotations\nimport sys\ndef f(x: sys.exit(7)):\n pass\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 - # Self-test the guard first: it must catch the ways the original bug could recur, and - # must NOT flag the sanctioned entrypoint or an exit that only lives inside a scope. - for label, src in _SELFTEST_MUST_CATCH: - checks += 1 - if not _module_scope_exit(ast.parse(src)): - failures.append(f"guard self-test: failed to catch {label}") - for label, src in _SELFTEST_MUST_PASS: + for label, src in _MUST_CATCH: checks += 1 - if _module_scope_exit(ast.parse(src)): - failures.append(f"guard self-test: wrongly flagged {label}") - - for fname in sorted(os.listdir(_HERE)): - if not (fname.startswith("test_") and fname.endswith(".py")): - continue - if fname == os.path.basename(__file__): - continue + if not _violates(src): + failures.append(f"scanner self-test: failed to catch {label}") + for label, src in _MUST_PASS: checks += 1 - src = open(os.path.join(_HERE, fname), encoding="utf-8").read() - tree = ast.parse(src) - has_run = any(isinstance(n, ast.FunctionDef) and n.name == "run" - for n in ast.iter_child_nodes(tree)) - if not has_run: - failures.append(f"{fname}: has no module-level run()") - if _module_scope_exit(tree): - failures.append(f"{fname}: calls sys.exit/raise SystemExit at import scope " - "(would short-circuit the aggregate runner)") - print(f"harness contract: {checks - len(failures)}/{checks} test modules honour run()") + 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