diff --git a/.aiqt/orchestration.json b/.aiqt/orchestration.json index 6a0e677..88c1e50 100644 --- a/.aiqt/orchestration.json +++ b/.aiqt/orchestration.json @@ -1,4 +1,15 @@ { "version": 1, - "companion_stores": ["/opt/cleanlanguage/private"] + "companion_stores": ["/opt/cleanlanguage/private"], + "record": { + "findings": "/opt/cleanlanguage/private/open-findings.md", + "pending_decisions": "/opt/cleanlanguage/private/pending-decisions.md", + "handoff": "/opt/cleanlanguage/private/session-handoff.md" + }, + "lease": { + "path": "/opt/cleanlanguage/private/session-state.md", + "max_age_hours": 24 + }, + "state_dir": "/opt/cleanlanguage/orch-state", + "dispatch_tools": [] } diff --git a/.github/workflows/orchestration-registry.yml b/.github/workflows/orchestration-registry.yml index 7bb4854..6b8a658 100644 --- a/.github/workflows/orchestration-registry.yml +++ b/.github/workflows/orchestration-registry.yml @@ -34,6 +34,17 @@ jobs: shell: bash run: python3 tools/check-orchestration-registry.py - - name: Run the orchestration-registry tests + - name: Run the orchestration-registry gate tests shell: bash run: python3 tools/test-orchestration-registry.py + + - name: Run the stage-1 hook behaviour tests + shell: bash + run: python3 tools/test-orch-stage1-hooks.py + + # Re-run the gate against the committed registry by its explicit path, so this step is RED both on + # a malformed edit and (via the gate tests above, which assert the arming keys are present) on a + # tree that lacks the stage-1 arming change. + - name: Validate the committed orchestration registry explicitly + shell: bash + run: python3 tools/check-orchestration-registry.py .aiqt/orchestration.json diff --git a/tools/check-orchestration-registry.py b/tools/check-orchestration-registry.py index 7facf9f..532786e 100644 --- a/tools/check-orchestration-registry.py +++ b/tools/check-orchestration-registry.py @@ -17,7 +17,8 @@ is inert by design), a present-but-unreadable one fails (matching the hook's `lstat` then read, which classify a permission fault or a dangling symlink as `bad`), and each `companion_stores` entry must be a non-empty absolute path (OS-agnostic, like the hook's -`_is_absolute`) with no control character (below 0x20 or 0x7f). Whether an entry resolves to +`_is_absolute`) with no control character (below 0x20 or 0x7f) or lone surrogate code point +(U+D800..U+DFFF). Whether an entry resolves to a live git top level is a runtime property the hook checks fail-safe; the path need not exist on the machine running this gate. @@ -29,11 +30,51 @@ import json import os import pathlib +import re +import stat import sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DEFAULT_REGISTRY = os.path.join(ROOT, ".aiqt", "orchestration.json") +# Stage-1 top-level allowlist. Any other top-level key is rejected: a new arming key +# (an enumerator, a yield-tool roster, a mode record) lands only together with the gate +# extension that validates it, so an unrecognized key is an ahead-of-gate or malformed +# registry, never silently accepted. This also mechanically keeps a strict-lint omission +# such as `lease.holder_is_session_id` out of the committed registry. +ALLOWED_KEYS = {"version", "companion_stores", "record", "lease", "state_dir", "dispatch_tools"} +# The record surfaces the hook's resume audit and barrier read (aiqt_hooks.py:9355-9391, 9820-9832). +RECORD_KEYS = {"findings", "pending_decisions", "handoff"} +# The lease shape _orch_scope_live and the resume audit read (aiqt_hooks.py:7786-7817, 9392-9409). +LEASE_KEYS = {"path", "max_age_hours"} +# A gate-side sanity bound on `lease.max_age_hours`, STRICTER than the hook. The hook's lease-freshness +# read (aiqt_hooks.py:7793-7798) checks only `max_age_hours > 0` and applies NO upper bound to it (the +# freshness window is `max_age * 3600`). _ORCH_MAX_HORIZON_HOURS = 8760 (aiqt_hooks.py:8265) bounds the +# STALENESS schema's task_hours/external_hours (aiqt_hooks.py:8298-8299), NOT the lease. Capping the +# committed lease at one year here is stricter than the hook, which is fail-safe for a committed artefact. +MAX_AGE_HOURS_MAX = 8760 +# POSIX filesystem length limits, applied as a pure commit-time string check to every declared path +# (codex QA HIGH-2). A component longer than NAME_MAX, or a whole path longer than PATH_MAX, is what the +# kernel rejects with ENAMETOOLONG at runtime. An over-long declared state_dir would, through exactly that +# ENAMETOOLONG, drive write_scope_guard to a persistent covered-write denial (see _check_declared_path), so +# it is caught here at author time rather than left to fail closed at every write. +NAME_MAX = 255 # bytes, a single path component +PATH_MAX = 4096 # bytes, a whole path +# Reserve room UNDER state_dir for the machine-state files the hooks WRITE there (codex QA HIGH-1, round 5). +# _check_declared_path already bounds the state_dir STRING at PATH_MAX, but the hooks then join a basename +# onto it, so a DERIVED path `/` can exceed PATH_MAX even when state_dir itself does +# not. At runtime that overflow raises ENAMETOOLONG in write_scope_guard._load_write_scope's +# `os.lstat(/write-scope.json)` (aiqt_hooks.py:9948-9953 via 10014-10015), the write-scope reader +# classifies it 'bad', and an armed session then denies every covered write persistently +# (aiqt_hooks.py:10388-10392). The longest basename the hooks derive under state_dir is +# "attestations-validated.json" (27 bytes; the longest of every filename joined onto +# _orch_state_dir_for_root in aiqt_hooks.py -- e.g. write-scope.json, resume-barrier.json, +# dispatch-ledger.jsonl, forced-exit-surfaced.json, backlog-checkpoint.json, ESCAPE-ALLOW-YIELD). We reserve +# that PLUS generous headroom, rounded up to 64, so a future recorder filename needs no re-count and one +# path separator is covered. The state_dir-specific check below rejects a state_dir that leaves less than +# this reserve (plus a separator) below PATH_MAX. +STATE_DERIVED_RESERVE = 64 # bytes; >= len("attestations-validated.json") (27) + margin + def _fail(msg): print("check-orchestration-registry: FAIL: {}".format(msg), file=sys.stderr) @@ -44,9 +85,95 @@ def _is_absolute(path): # OS-agnostic, matching the hook's _is_absolute: a POSIX absolute path OR a Windows path # carrying both a drive and a root. This keeps the gate from falsely rejecting an entry the # hook would accept when the gate runs on a different OS than the committer. + # + # Disclosed residual (codex QA LOW, not fixed): because this is OS-agnostic, a PureWindowsPath value + # such as `C:\x` reads as absolute here, matching the installed hook and the pack's lab_infra + # validator. On this POSIX host the POSIX hook would instead resolve such a value RELATIVE to the + # repo. This is left as a disclosed residual rather than fixed, to preserve cross-OS parity with the + # pack; the committed store paths are POSIX-absolute, and both human review and the committed-registry + # change-carries-check guard the committed registry against a stray Windows-shaped path. return pathlib.PurePosixPath(path).is_absolute() or pathlib.PureWindowsPath(path).is_absolute() +def _has_unsafe_char(text): + # A control character (below 0x20 or 0x7f DEL) OR a lone surrogate code point (U+D800..U+DFFF) in a + # declared path is load-bearing to reject, not cosmetic. Both crash the same PreToolUse path. + # Control character: a NUL in a record path reaches os.path.realpath inside orch_resume_barrier + # (aiqt_hooks.py:9828), which raises ValueError. Surrogate: a lone surrogate CANNOT be UTF-8 encoded + # (it is unpaired), so when the hook resolves the declared path (os.path.realpath / open in + # orch_resume_barrier, aiqt_hooks.py:9828-9830, a PreToolUse event) the encode to the filesystem + # encoding raises UnicodeEncodeError. Either crash makes the PreToolUse dispatcher fail closed exit 2 + # (aiqt_hooks.py:10705-10716) -- the one route by which a stage-1 key could reach a blocking outcome. + # Rejecting surrogates here closes exactly the route the control-char rejection claims to close + # (closes the codex QA HIGH). The gate closes it for the committed file (matching _wrtscp/companion_stores). + return any(ord(ch) < 0x20 or ord(ch) == 0x7f or 0xD800 <= ord(ch) <= 0xDFFF for ch in text) + + +def _norm_pathspell(p): + # Canonicalize a declared path for a containment comparison: collapse `.`, `..`, and redundant + # separators with normpath, THEN collapse any leading run of 2+ slashes to a single slash. POSIX + # normpath PRESERVES a leading exactly-"//" (its value is implementation-defined per POSIX) and already + # collapses 3+, so the only residual after normpath is a leading "//"; collapsing it makes "/a", "//a", + # and "///a" all compare equal. Without this, an alias spelling defeats the commonpath containment test + # below (norm("//opt/x") != norm("/opt/x")), which is codex QA HIGH-1's bypass. + return re.sub(r"^/{2,}", "/", os.path.normpath(p)) + + +def _contains_or_equal(child, parent): + # True when absolute `child` equals `parent` or lies inside it, decided component-wise by commonpath + # (so a genuine sibling like /opt/x/orch-state-extra is NOT read as inside /opt/x/orch-state, whose + # commonpath is /opt/x). Both arguments are absolute -- either canonicalized spellings (_norm_pathspell) + # or realpath'd values. commonpath raises ValueError only on mixing absolute and relative paths, or (on + # Windows) paths on different drives; neither applies to two validated absolute POSIX-shaped paths, but + # if it somehow does, treat it as a collision and fail defensively rather than accept an unproven-safe + # path. Shared by the lexical and the resolved-path containment checks (codex QA HIGH-1). + if child == parent: + return True + try: + return os.path.commonpath([child, parent]) == parent + except ValueError: + return True + + +def _check_declared_path(label, value): + # A registry-declared path (a record surface, the lease, the state dir, a companion store) must be a + # non-empty absolute string with no control or surrogate character, the same shape the hook's _orch_path + # plus its realpath walk need. + if not isinstance(value, str) or not value: + _fail("`{}` must be a non-empty string, got {!r}".format(label, value)) + if not _is_absolute(value): + _fail("`{}` must be an absolute path, got {!r}".format(label, value)) + if _has_unsafe_char(value): + _fail("`{}` contains a control or surrogate character".format(label)) + # Reject an over-long path: any single component longer than NAME_MAX bytes, or a whole path longer than + # PATH_MAX bytes. This is a PURE STRING check (fires in CI, no filesystem access). It closes codex QA + # HIGH-2's ENAMETOOLONG case: an over-long declared state_dir drives write_scope_guard._load_write_scope + # to a persistent covered-write DENIAL -- lstat(/write-scope.json) raises ENAMETOOLONG, the + # artifact reader classifies it 'bad', and an armed session then denies every covered write + # (aiqt_hooks.py:9940-9946, 10015, 10388-10392). Measured in BYTES, not code points, because the kernel + # limits are on the UTF-8 encoded length; surrogatepass keeps this from crashing on a surrogate that + # _has_unsafe_char has already rejected above. + # PATH_MAX (4096) INCLUDES the terminating NUL, so the longest USABLE pathname is PATH_MAX - 1 (4095) + # bytes; a 4096-byte path passes a `> PATH_MAX` test but the kernel rejects it with ENAMETOOLONG at + # runtime (os.stat -> errno 36), which for a lease/state path drives the hook's stat to an OSError and, + # with no mode declared, an inactive scope (aiqt_hooks.py:7791,7813) -- so reject `>= PATH_MAX` (codex QA + # round 8). NAME_MAX (255) EXCLUDES the NUL, so the component check below correctly stays `> NAME_MAX`. + if len(value.encode("utf-8", "surrogatepass")) >= PATH_MAX: + _fail("`{}` is {} bytes; the longest usable path is {} bytes (PATH_MAX {} counts the NUL)".format( + label, len(value.encode("utf-8", "surrogatepass")), PATH_MAX - 1, PATH_MAX)) + # Measure each component against NAME_MAX splitting on "/" ONLY -- POSIX component semantics. On POSIX a + # backslash is a VALID filename character, so an intrinsically over-long single component like + # ("a"*150 + "\\" + "a"*150) (301 bytes, ONE POSIX component) must NOT be broken at the backslash into + # two <=255-byte sub-components and wrongly accepted, only to hit ENAMETOOLONG at runtime (codex QA + # HIGH-2, round 6; this corrects the `[/\\]` split introduced in round 5, which under-counted such a + # component). A Windows-shaped path's backslash-separated segments are therefore measured as one run, + # which OVER-counts rather than under-counts: it can only make an over-long Windows path reject, never + # wrongly accept, so it stays fail-safe. The committed store paths are POSIX. + for comp in value.split("/"): + if len(comp.encode("utf-8", "surrogatepass")) > NAME_MAX: + _fail("`{}` has a path component longer than {} bytes (NAME_MAX)".format(label, NAME_MAX)) + + def validate(path): # Presence, matching the hook: a genuine FileNotFoundError on lstat is absence (pass); any other # stat fault (a permission error, a symlink loop) or a read failure (a dangling symlink follows to a @@ -59,6 +186,16 @@ def validate(path): return except OSError as exc: _fail("cannot stat {}: {}".format(path, exc)) + # Disclosed residual (codex/claude QA LOW, not fixed here): the presence check is an lstat followed by + # a plain blocking open() below, so a FIFO/named-pipe left at the registry path with no writer would + # block this open indefinitely. The same special-file hazard applies to a declared record/lease/state + # path: were one to point at a FIFO, it would block the HOOK's own read at read time, not this gate. + # That is out of the gate's scope: the gate validates the registry STRING (non-empty, absolute, + # control/surrogate-free), not the file type of the target, which may not exist at gate time. This is + # CI-SAFE: git cannot check out a FIFO, so the only exposure is an exotic hand-crafted local run, never + # a checked-out tree; the declared paths are store paths. lab_infra's validator opens + # O_RDONLY|O_NONBLOCK and fstat-checks S_ISREG, and is the consolidation point at the stage-2 lift; + # this gate is left as a disclosed residual rather than duplicating that hardening now. try: with open(path, "r", encoding="utf-8") as fh: raw = fh.read() @@ -70,7 +207,10 @@ def validate(path): _fail("{} is not valid UTF-8: {}".format(path, exc)) try: obj = json.loads(raw) - except ValueError as exc: + except (ValueError, RecursionError) as exc: + # ValueError covers ordinary malformed JSON; RecursionError (NOT a ValueError) covers deeply + # nested JSON (e.g. many thousands of open brackets) that overflows json's recursive scanner. + # Catching it keeps the gate a clean FAIL rather than an uncaught traceback (still fail-closed). _fail("{} is not valid JSON: {}".format(path, exc)) if not isinstance(obj, dict): _fail("{} top level must be a JSON object, got {}".format(path, type(obj).__name__)) @@ -87,15 +227,219 @@ def validate(path): if not isinstance(stores, list): _fail("`companion_stores` must be a list, got {}".format(type(stores).__name__)) for i, entry in enumerate(stores): + # Route through the shared declared-path helper: same non-empty/absolute/control-and-surrogate + # checks as before, now also covered by the NAME_MAX/PATH_MAX over-long check (codex QA HIGH-2). + _check_declared_path("companion_stores[{}]".format(i), entry) + # Reject any top-level key outside the stage-1 allowlist. A new arming key lands only together with + # its gate extension, so an unrecognized key is an ahead-of-gate or malformed registry. + for key in obj: + if key not in ALLOWED_KEYS: + _fail("unknown top-level key {!r}; a new arming key lands only together with its gate " + "extension".format(key)) + # `record`, if present: a non-empty object whose keys are a subset of RECORD_KEYS, each value a + # declared (non-empty absolute control-free) path. + if "record" in obj: + record = obj["record"] + if not isinstance(record, dict): + _fail("`record` must be a JSON object, got {}".format(type(record).__name__)) + if not record: + _fail("`record` must not be empty") + extra = set(record) - RECORD_KEYS + if extra: + _fail("`record` has unknown key(s) {}".format(sorted(extra))) + for key in sorted(record): + _check_declared_path("record.{}".format(key), record[key]) + # `lease`, if present: an object whose keys are a subset of LEASE_KEYS; `path` is REQUIRED and a + # declared path; `max_age_hours` is optional but, when present, a non-bool int or float in (0, 8760]. + if "lease" in obj: + lease = obj["lease"] + if not isinstance(lease, dict): + _fail("`lease` must be a JSON object, got {}".format(type(lease).__name__)) + extra = set(lease) - LEASE_KEYS + if extra: + _fail("`lease` has unknown key(s) {}".format(sorted(extra))) + if "path" not in lease: + _fail("`lease.path` is required") + _check_declared_path("lease.path", lease["path"]) + if "max_age_hours" in lease: + v = lease["max_age_hours"] + # bool is a subclass of int; reject it. NaN/Infinity fail the range comparison and are + # rejected too. The (0, 8760] window is a gate-side sanity cap, STRICTER than the hook, which + # applies no upper lease bound (see MAX_AGE_HOURS_MAX above); stricter is fail-safe here. + if isinstance(v, bool) or not isinstance(v, (int, float)) or not (0 < v <= MAX_AGE_HOURS_MAX): + _fail("`lease.max_age_hours` must be a number in (0, {}], got {!r}".format( + MAX_AGE_HOURS_MAX, v)) + # `state_dir`, if present: a declared path. + if "state_dir" in obj: + _check_declared_path("state_dir", obj["state_dir"]) + # Reserve room for the DERIVED machine-state paths (codex QA HIGH-1, round 5). _check_declared_path + # bounded the state_dir STRING at PATH_MAX, but the hooks join a basename onto it, so a near-limit + # state_dir makes `/` overflow PATH_MAX even though state_dir itself did not. + # At runtime that overflow raises ENAMETOOLONG in lstat(/write-scope.json), the + # write-scope reader classifies it 'bad', and an armed session denies every covered write + # persistently. Require len(state_dir) + one separator + STATE_DERIVED_RESERVE <= PATH_MAX, measured + # in BYTES (surrogatepass matches _check_declared_path, though surrogates are already rejected above). + _sd_bytes = len(obj["state_dir"].encode("utf-8", "surrogatepass")) + # `>= PATH_MAX`, not `> PATH_MAX`: the derived path / must itself be <= PATH_MAX-1 + # usable bytes (codex QA round 8, same off-by-one class as the whole-path check above). + if _sd_bytes + 1 + STATE_DERIVED_RESERVE >= PATH_MAX: + _fail("`state_dir` ({!r}) is {} bytes and leaves no room under PATH_MAX ({}) for the " + "machine-state files the hooks derive under it: {} bytes + 1 separator + {} reserved " + "for the longest derived filename exceeds PATH_MAX, so a path like " + "/write-scope.json would hit ENAMETOOLONG and drive write_scope_guard to a " + "persistent covered-write denial".format( + obj["state_dir"], _sd_bytes, PATH_MAX, _sd_bytes, STATE_DERIVED_RESERVE)) + # BEST-EFFORT gate-time type check (codex QA HIGH-2): classify the declared state_dir by what its + # lstat actually reports. write_scope_guard._load_write_scope reads /write-scope.json; + # when state_dir is a regular file (e.g. "/etc/passwd") or a path THROUGH a regular file (e.g. + # "/etc/passwd/" or "/etc/passwd/state") that read's lstat raises ENOTDIR, the artifact reader + # classifies it 'bad', and an armed session then DENIES every covered write persistently + # (aiqt_hooks.py:9948-9953, 10015, 10388-10392). Three outcomes, distinguished (round-5 fix; round-4 + # collapsed every OSError to absence and ACCEPTED a known-unusable path): + # - FileNotFoundError: GENUINE ABSENCE -> do NOT fail (the not-yet-created residual stays + # disclosed; the gate cannot verify a not-yet-created directory). + # - a successful lstat that is NOT a directory -> _fail (an existing non-dir, e.g. a regular file + # or a dangling symlink whose lstat succeeds). + # - ANY OTHER OSError (ENOTDIR, ENAMETOOLONG, ELOOP, EACCES, ...): the path is KNOWN-UNUSABLE at + # gate time -> _fail naming it, since the hook hits the same fault at runtime. Do NOT swallow it + # as absence. + # This is BEST-EFFORT, NOT a soundness guarantee: a state_dir absent at gate time but created as a + # FILE, or unusable at runtime (a read-only or different filesystem), is NOT caught here. os.lstat + # (not os.stat/os.path.exists) so a symlink is judged by the link itself. + try: + _sd_st = os.lstat(obj["state_dir"]) + except FileNotFoundError: + _sd_st = None # genuine absence: do NOT fail + except OSError as exc: + _fail("`state_dir` ({!r}) cannot be used as a directory at gate time ({}); the hooks write " + "machine-state files inside it, so a known-unusable state_dir drives write_scope_guard " + "to a persistent covered-write denial".format(obj["state_dir"], exc)) + if _sd_st is not None and not stat.S_ISDIR(_sd_st.st_mode): + _fail("`state_dir` ({!r}) exists at gate time but is not a directory; the hooks write " + "machine-state files inside it, and a non-directory state_dir drives write_scope_guard " + "to a persistent covered-write denial".format(obj["state_dir"])) + # Cross-field REQUIREMENT (claude QA HIGH, round 7): a declared `record` or `lease` must be accompanied + # by an explicit `state_dir`. When `state_dir` is OMITTED the hooks do NOT stop using a state dir -- + # _state_dir_from_registry (aiqt_hooks.py:7638-7651) falls back to + # ${XDG_STATE_HOME:-~/.local/state}/aiqt-guardrails/orch//, and orch_resume_audit still opens + # `/resume-barrier.json` with mode "w" there (aiqt_hooks.py:9783-9785). A record/lease path + # landing inside that DEFAULT dir is clobbered exactly as the round-3 declared-state_dir collision, but + # the collision check below runs only `if "state_dir" in obj`, so the default-dir sibling slips through + # (claude QA round 7, reproduced gate-exit-0 -> hook clobber of the declared record). Requiring + # `state_dir` alongside record/lease closes the whole class at gate time: the location is then validated + # and collision-checked, and a gate-passing armed registry never falls back to the XDG default. + # `dispatch_tools` carries tool NAMES, not a path, so it cannot collide and does not trigger this. + if ("record" in obj or "lease" in obj) and "state_dir" not in obj: + _fail("`record`/`lease` is declared without `state_dir`: the hooks then fall back to an XDG-default " + "state directory ($XDG_STATE_HOME/aiqt-guardrails/orch/, aiqt_hooks.py:7638-7651) " + "that this gate never sees, and orch_resume_audit's unconditional `/" + "resume-barrier.json` write clobbers any record/lease path landing inside it; declare " + "`state_dir` explicitly so the machine-state location is validated and collision-checked") + # Cross-field collision (codex QA HIGH): a declared record/lease path that lands ON or INSIDE + # `state_dir` collides with the machine-state files the hooks write there. orch_resume_audit opens + # `/resume-barrier.json` with mode "w" UNCONDITIONALLY (aiqt_hooks.py:9775, 9783-9785), so + # a record/lease path pointing into state_dir would be clobbered by that hook write (destructive, no + # lease needed). The per-key checks above validate each path in isolation and cannot see this; this + # runs ONLY when `state_dir` is declared and valid (execution reaches here only past its + # _check_declared_path). `companion_stores` is intentionally NOT checked against state_dir: a state_dir + # legitimately MAY sit under a companion store; the hazard is a record/lease FILE landing on a + # machine-state file, not the state directory living under an exempt store. + if "state_dir" in obj: + # Robust against path-spelling ALIASES (codex QA HIGH-1). The round-3 normpath+startswith test was + # bypassed by aliases the two spellings do not share: state_dir "/" + record "/x.json" (norm_state + # "/" + os.sep = "//", which "/x.json" does not start with), and any leading double-slash spelling + # (POSIX normpath preserves a leading "//"). _norm_pathspell canonicalizes BOTH sides and collapses + # that leading "//", and os.path.commonpath then decides containment component-wise, so a genuine + # SIBLING (e.g. /opt/x/orch-state-extra vs state_dir /opt/x/orch-state -> commonpath /opt/x) is + # still accepted while ON-or-INSIDE is rejected. + norm_state_dir = _norm_pathspell(obj["state_dir"]) + # ADDITIONALLY compare RESOLVED paths (codex QA HIGH-1, round 6): an EXISTING symlink alias makes two + # different spellings resolve to the same file, which the lexical check cannot see. os.path.realpath + # resolves existing symlink components and leaves a non-existent tail intact, so a record/lease path + # that resolves onto or into the resolved state_dir is caught even when the spellings differ (codex + # reproduced: state_dir "/dev/shm" + record "/run/shm/resume-barrier.json" where /run/shm -> /dev/shm + # resolve to the same barrier file). BOTH the lexical and the resolved check run -- defence in depth. + # realpath is computed on the same state_dir as on each declared path, so a symlink anywhere in a + # shared prefix resolves identically on both sides and cannot manufacture a false collision for a + # genuine sibling. + real_state_dir = os.path.realpath(obj["state_dir"]) + declared = [] + if isinstance(obj.get("record"), dict): + for key in sorted(obj["record"]): + declared.append(("record.{}".format(key), obj["record"][key])) + if isinstance(obj.get("lease"), dict) and "path" in obj["lease"]: + declared.append(("lease.path", obj["lease"]["path"])) + for label, value in declared: + if _contains_or_equal(_norm_pathspell(value), norm_state_dir): + _fail("`{}` ({!r}) is on or inside `state_dir` ({!r}); a record/lease path on or under " + "state_dir collides with the machine-state files the hooks write there (e.g. " + "resume-barrier.json), so a hook write would clobber the declared record".format( + label, value, obj["state_dir"])) + real_value = os.path.realpath(value) + if _contains_or_equal(real_value, real_state_dir): + _fail("`{}` ({!r}) RESOLVES onto or inside `state_dir` ({!r}) through an existing symlink " + "(record/lease resolves to {!r}, state_dir to {!r}); a hook write to a machine-state " + "file there (e.g. resume-barrier.json) would clobber the declared record".format( + label, value, obj["state_dir"], real_value, real_state_dir)) + # --- THE BOUND: accepted-input soundness is bounded to commit-time validation --------------------- + # This gate performs COMMIT-TIME validation only. What it establishes is: structural shape + concrete + # hazards (surrogate/control character, NAME_MAX/PATH_MAX length, derived-path reserve, record/lease- + # vs-state_dir collision) + BEST-EFFORT resolved-path containment (above) + BEST-EFFORT state_dir + # gate-time directory-type. Its accepted-input soundness is BOUNDED to what commit-time validation can + # establish; the maintainer directed this bound after round 6. The residuals below are IRREDUCIBLE + # (no commit-time string or stat check can close them). They do NOT all resolve to a fail-safe denial + # -- claude QA round 10 corrected an earlier overstatement that each was "a denial or a caught + # collision at runtime, never a bypass". They split by runtime outcome: + # FAIL-SAFE DENIALS -- a denial or a caught collision at runtime, never a bypass: + # (b) Runtime directory-usability of a NOT-YET-CREATED state_dir: an absent state_dir later created + # as a regular file, or living on a read-only or different filesystem. The best-effort lstat + # above classifies only what exists at gate time; genuine absence is accepted (it cannot be + # verified). At runtime an unusable state_dir drives write_scope_guard._load_write_scope to a + # persistent covered-write DENIAL (fail-safe), operator-fixable. + # (c) A FIFO or other special file at a declared registry/record/lease/state path: it would block + # the HOOK's own read at read time (see the FIFO disclosure at the presence check above, in + # validate()). git cannot check out a FIFO, so the only exposure is an exotic hand-crafted local + # run, never a checked-out tree. + # POST-GATE FILESYSTEM ALIASING -- a SILENT CLOBBER, not a denial; reachable ONLY with out-of-band + # write access to the store or state_dir that ALREADY lets the actor clobber the record directly + # (the recorder/audit hooks are DEFENCE IN DEPTH, not a security boundary), so it adds no exposure: + # (a) FUTURE symlink swap: a symlink created or retargeted AFTER this gate runs. The resolved-path + # check above resolves symlinks at COMMIT time only; a later swap is not observable here, and a + # record/lease path later aliased onto a machine-state file is clobbered by the hook write. + # (d) HARD LINK (or a symlink planted at a derived state path, e.g. the barrier name) between a + # record/lease file and a machine-state file the hooks write. os.path.realpath resolves symlinks + # but NOT hard links, the JSON cannot express a link, and the colliding derived file (e.g. + # /resume-barrier.json) does not exist at gate time, so no commit-time comparison can + # see it. orch_resume_audit's barrier write is a direct truncating open(barrier, "w") with NO + # S_ISREG guard (aiqt_hooks.py:9783, unlike the shared _wrtscp_read_json_artifact reader), so an + # aliased record file is SILENTLY CLOBBERED, not denied. Precondition: write+search access to + # state_dir (and, under fs.protected_hardlinks, to the record file) -- already store-defeating. + # Also disclosed and left as a parity residual, not closed here: the OS-agnostic _is_absolute Windows + # spelling (see its LOW disclosure in _is_absolute above). A consequence of accepting that spelling: + # on a POSIX host the record/lease-vs-state_dir collision check does NOT catch a BACKSLASH-spelled + # Windows collision (e.g. state_dir "C:\\s" + record "C:\\s\\resume-barrier.json"), because + # os.path.commonpath treats each backslash string as one POSIX component and finds them disjoint; the + # FORWARD-slash Windows spelling IS caught, and a mixed POSIX/Windows pair fails closed. Unreachable + # for the committed POSIX artefact and guarded by the committed-registry test plus human review. + # write_scope_guard and the recorder/audit hooks are DEFENCE IN DEPTH, not a security boundary, so a + # fail-safe denial there is degraded-but-safe, not exploitable. The COMMITTED registry is verified + # safe by this gate plus its tests, and human review via change-carries-check is the backstop for + # future edits. + # `dispatch_tools`, if present: a list of non-empty control-free strings. + if "dispatch_tools" in obj: + tools = obj["dispatch_tools"] + if not isinstance(tools, list): + _fail("`dispatch_tools` must be a list, got {}".format(type(tools).__name__)) + for i, entry in enumerate(tools): if not isinstance(entry, str) or not entry: - _fail("`companion_stores[{}]` must be a non-empty string, got {!r}".format(i, entry)) - if not _is_absolute(entry): - _fail("`companion_stores[{}]` must be an absolute path, got {!r}".format(i, entry)) - if any(ord(ch) < 0x20 or ord(ch) == 0x7f for ch in entry): - _fail("`companion_stores[{}]` contains a control character".format(i)) + _fail("`dispatch_tools[{}]` must be a non-empty string, got {!r}".format(i, entry)) + if _has_unsafe_char(entry): + _fail("`dispatch_tools[{}]` contains a control or surrogate character".format(i)) summary = "version 1" if "companion_stores" in obj: summary += ", {} companion store(s)".format(len(obj["companion_stores"])) + arming = [k for k in ("record", "lease", "state_dir", "dispatch_tools") if k in obj] + summary += ", arming keys: {}".format(", ".join(arming)) if arming else ", no arming keys" print("check-orchestration-registry: OK ({}: {})".format(path, summary)) diff --git a/tools/test-orch-stage1-hooks.py b/tools/test-orch-stage1-hooks.py new file mode 100644 index 0000000..4a5610b --- /dev/null +++ b/tools/test-orch-stage1-hooks.py @@ -0,0 +1,684 @@ +#!/usr/bin/env python3 +"""Behavioural change-carries-check for the item 28.2 stage-1 arming of the recorder/audit hooks. + +This suite proves that the stage-1 `.aiqt/orchestration.json` arming keys `record`, `lease`, and +`state_dir` actually change the installed hook's behaviour, and that the change adds no unbounded or +fail-closed path beyond the pre-existing bad-registry class. The fourth stage-1 key, `dispatch_tools`, +is NOT behaviourally load-bearing at its committed value `[]`: the recorder reads it as +`reg.get("dispatch_tools") if isinstance(..., list) else []` (aiqt_hooks.py:9277-9280), so declaring +`[]` is identical to omitting the key, and a background Bash or a TaskOutput is recorded regardless of +it. `dispatch_tools` only EXTENDS the dispatch set with extra tool names; the committed `[]` is declared +for honesty and explicitness, not because it flips any assertion here. It does so by +driving the REAL installed hook script directly: per case it builds a throwaway git repo and a +temporary stand-in store, writes a fixture registry into the repo, and invokes + + python3 -I /hooks/scripts/aiqt_hooks.py + +with a JSON stdin payload, then asserts the exit code, any stdout verdict JSON, and the state files +the hook wrote under the temporary store. It NEVER touches the real /opt/cleanlanguage/private store, +so it runs in CI. + +Every armed-behaviour proof runs against BOTH the stage-1 fixture (behaviour PRESENT) and today's +two-key registry (behaviour ABSENT), so each behaviourally load-bearing arming key (`record`, `lease`, +`state_dir`) is shown load-bearing: a recorder that wrote unconditionally, or an audit that read an +undeclared surface, would flip the two-key assertion. That dual run is the change-carries-check. +(`dispatch_tools` is exempt from this contrast, per the note above: its committed `[]` is +behaviourally identical to omitting it, so a two-key contrast on it would prove nothing.) + +Grounding: every hook-behaviour assertion below is read from +plugin/aiqt-guardrails-hooks/hooks/scripts/aiqt_hooks.py at source; the cited line numbers are in the +per-test comments. Dispatch-action names are the HANDLERS keys (aiqt_hooks.py:10594-10619), passed as +the single argv (main(sys.argv[1:]), aiqt_hooks.py:10724). The repo root is the git top level of +data["cwd"] (_orch_root -> _recovery_toplevel, aiqt_hooks.py:7590-7595, 3859), and scope goes live only +when the declared lease file is present, non-empty and fresh (_orch_scope_live, aiqt_hooks.py:7779-7817). +""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import stat +import subprocess +import sys +import tempfile +import time +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +HERE = Path(__file__).resolve().parent +REPO_ROOT = HERE.parent +HOOK = REPO_ROOT / "plugin" / "aiqt-guardrails-hooks" / "hooks" / "scripts" / "aiqt_hooks.py" + + +def _sha256_hex(text: str) -> str: + return hashlib.sha256(text.encode("utf-8", "replace")).hexdigest() + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _git(repo, *args, extra_env=None): + env = dict(os.environ) + if extra_env: + env.update(extra_env) + return subprocess.run(["git", "-C", str(repo), *args], + capture_output=True, text=True, env=env, check=True) + + +class HookHarness(unittest.TestCase): + """A throwaway git repo plus a temporary stand-in store, torn down after each test.""" + + def setUp(self): + if not HOOK.exists(): + self.skipTest("installed hook not found: {}".format(HOOK)) + self._tmp = Path(tempfile.mkdtemp(prefix="orch-stage1-")) + self.repo = self._tmp / "repo" + self.store = self._tmp / "store" + self.repo.mkdir() + self.store.mkdir() + # A hermetic git repo with exactly one commit, so `git rev-parse HEAD` resolves for the resume + # audit's gate-marker probe and the default branch is a known value ("main"). + subprocess.run(["git", "init", "-b", "main", str(self.repo)], + capture_output=True, text=True, check=True) + _git(self.repo, "-c", "user.email=t@example.invalid", "-c", "user.name=t", + "commit", "--allow-empty", "-m", "init") + + def tearDown(self): + # Restore any 0o555 dir written by the unwritable-state-dir case so rmtree can clean up. + for dirpath, _dirs, _files in os.walk(self._tmp): + try: + os.chmod(dirpath, 0o755) + except OSError: + pass + shutil.rmtree(self._tmp, ignore_errors=True) + + # --- registry fixtures --------------------------------------------------------------------------- + def _p(self, *parts): + return str(self.store.joinpath(*parts)) + + def state_dir(self): + return self.store / "orch-state" + + def stage1(self, **overrides): + reg = { + "version": 1, + "companion_stores": [str(self.store)], + "record": { + "findings": self._p("open-findings.md"), + "pending_decisions": self._p("pending-decisions.md"), + "handoff": self._p("session-handoff.md"), + }, + "lease": {"path": self._p("session-state.md"), "max_age_hours": 24}, + "state_dir": str(self.state_dir()), + "dispatch_tools": [], + } + reg.update(overrides) + return reg + + def two_key(self): + return {"version": 1, "companion_stores": [str(self.store)]} + + # --- fixture builders ---------------------------------------------------------------------------- + def write_registry(self, obj): + aiqt = self.repo / ".aiqt" + aiqt.mkdir(exist_ok=True) + path = aiqt / "orchestration.json" + if isinstance(obj, str): + path.write_text(obj, encoding="utf-8") + else: + path.write_text(json.dumps(obj), encoding="utf-8") + + def arm_lease(self, age_seconds=0): + """Create a non-empty lease file; age_seconds>0 back-dates its mtime (a stale lease).""" + lease = Path(self._p("session-state.md")) + lease.write_text("held by the orchestrator\n", encoding="utf-8") + if age_seconds: + t = time.time() - age_seconds + os.utime(lease, (t, t)) + return lease + + def write_record_files(self, handoff="", findings="", pending=""): + Path(self._p("session-handoff.md")).write_text(handoff, encoding="utf-8") + Path(self._p("open-findings.md")).write_text(findings, encoding="utf-8") + Path(self._p("pending-decisions.md")).write_text(pending, encoding="utf-8") + + def seed_state(self, name, obj): + sd = self.state_dir() + sd.mkdir(parents=True, exist_ok=True) + (sd / name).write_text(json.dumps(obj), encoding="utf-8") + + # --- invocation ---------------------------------------------------------------------------------- + def run_hook(self, action, payload): + """Invoke `python3 -I aiqt_hooks.py ` with the JSON payload on stdin. + + HOME and XDG_STATE_HOME are pinned inside the throwaway tree, so an UNARMED (two-key) run whose + state_dir falls back to the XDG default writes into the sandbox, never the developer's home; a + stage-1 run's explicit state_dir (registry wins over XDG) still lands in /orch-state. + """ + payload = dict(payload) + payload.setdefault("cwd", str(self.repo)) + env = dict(os.environ) + env["HOME"] = str(self.store / "home") + env["XDG_STATE_HOME"] = str(self.store / "xdg") + result = subprocess.run( + [sys.executable, "-I", str(HOOK), action], + input=json.dumps(payload), capture_output=True, text=True, env=env, + ) + return result + + def stdout_json(self, result): + out = result.stdout.strip() + if not out: + return None + return json.loads(out.splitlines()[0]) + + def read_jsonl(self, name): + path = self.state_dir() / name + if not path.exists(): + return None + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] + + def read_json(self, name): + path = self.state_dir() / name + if not path.exists(): + return None + return json.loads(path.read_text(encoding="utf-8")) + + def find_anywhere(self, name): + return [str(p) for p in self.store.rglob(name)] + + def read_json_anywhere(self, name): + # A bad or two-key registry declares no state_dir, so the hook falls back to the XDG default + # (pinned inside the sandbox by run_hook); the artefact lands there, not in /orch-state. + matches = list(self.store.rglob(name)) + if not matches: + return None + return json.loads(matches[0].read_text(encoding="utf-8")) + + +# ===================================================================================================== +# 1. orch_dispatch_ledger (PostToolUse recorder; aiqt_hooks.py:9248-9304) +# ===================================================================================================== +class DispatchLedger(HookHarness): + def test_background_bash_synthesized_id_wake_false(self): + # A background Bash with no observed task id gets a synthesized "disp-" id and wake=False, so it + # can never be counted as a live task (aiqt_hooks.py:9279-9296). + self.write_registry(self.stage1()) + self.arm_lease() + r = self.run_hook("orch_dispatch_ledger", { + "tool_name": "Bash", + "tool_input": {"run_in_background": True, "command": "sleep 1"}, + "tool_response": {}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + rows = self.read_jsonl("dispatch-ledger.jsonl") + self.assertIsNotNone(rows, "stage-1 must record a launch row") + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["event"], "launch") + self.assertFalse(rows[0]["wake"]) + self.assertTrue(rows[0]["task_id"].startswith("disp-")) + + def test_background_bash_observed_id_wake_true(self): + # A REAL observed task id (tool_response.task_id) records wake=True (aiqt_hooks.py:9284-9296). + self.write_registry(self.stage1()) + self.arm_lease() + r = self.run_hook("orch_dispatch_ledger", { + "tool_name": "Bash", + "tool_input": {"run_in_background": True, "command": "sleep 1"}, + "tool_response": {"task_id": "task-abc"}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + rows = self.read_jsonl("dispatch-ledger.jsonl") + self.assertEqual(rows[0], {**rows[0], "event": "launch", "task_id": "task-abc", "wake": True}) + + def test_taskoutput_with_id_completes(self): + # TaskOutput carrying a task id appends a complete row, wake=True (aiqt_hooks.py:9264-9267). + self.write_registry(self.stage1()) + self.arm_lease() + r = self.run_hook("orch_dispatch_ledger", { + "tool_name": "TaskOutput", "tool_input": {"task_id": "task-abc"}, "tool_response": {}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + rows = self.read_jsonl("dispatch-ledger.jsonl") + self.assertEqual(rows[0]["event"], "complete") + self.assertEqual(rows[0]["task_id"], "task-abc") + self.assertTrue(rows[0]["wake"]) + + def test_two_key_registry_writes_nothing(self): + # LOAD-BEARING contrast: with the two-key registry scope is never live (no lease, no mode), so + # the recorder returns _allow() and writes NO ledger anywhere (aiqt_hooks.py:9258-9259). + self.write_registry(self.two_key()) + self.arm_lease() # a lease file exists but the registry declares none, so scope stays inert + r = self.run_hook("orch_dispatch_ledger", { + "tool_name": "Bash", + "tool_input": {"run_in_background": True, "command": "sleep 1"}, + "tool_response": {}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(self.find_anywhere("dispatch-ledger.jsonl"), [], + "the two-key registry must record nothing") + + +# ===================================================================================================== +# 2. TaskOutput with no id -> unbound systemMessage, no row (aiqt_hooks.py:9268-9275) +# ===================================================================================================== +class TaskOutputNoId(HookHarness): + def test_unbound_taskoutput(self): + self.write_registry(self.stage1()) + self.arm_lease() + r = self.run_hook("orch_dispatch_ledger", { + "tool_name": "TaskOutput", "tool_input": {}, "tool_response": {}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertIn("UNBOUND", out["systemMessage"]) + self.assertIsNone(self.read_jsonl("dispatch-ledger.jsonl"), "an unbound read records no row") + + +# ===================================================================================================== +# 3. orch_prompt_stamp (UserPromptSubmit recorder; aiqt_hooks.py:9307-9348) +# ===================================================================================================== +class PromptStamp(HookHarness): + def test_human_prompt_stamps_and_zeroes_counters(self): + # A genuine human prompt stamps last_human_input_utc and zeroes both denial counters + # (aiqt_hooks.py:9327-9333). LOAD-BEARING: the turn-state is SEEDED with NONZERO stop/schedule + # denial counters first, so a broken implementation that dropped the reset (preserving the seeded + # values) would fail these assertions instead of passing on a fresh (absent-key => 0) turn-state. + # The prompt is not a seeded wake digest, so it is genuine human input, not timer-originated. + self.write_registry(self.stage1()) + self.arm_lease() + self.seed_state("turn-state.json", {"stop_denials": 5, "schedule_denials": 7}) + r = self.run_hook("orch_prompt_stamp", {"prompt": "please continue the work"}) + self.assertEqual(r.returncode, 0, r.stderr) + ts = self.read_json("turn-state.json") + self.assertIsNotNone(ts, "stage-1 must write turn-state.json") + self.assertIn("last_human_input_utc", ts) + self.assertEqual(ts["stop_denials"], 0, "a human prompt must zero a seeded nonzero stop counter") + self.assertEqual(ts["schedule_denials"], 0, + "a human prompt must zero a seeded nonzero schedule counter") + + def test_timer_originated_prompt(self): + # A prompt whose digest is a seeded wake digest is classified TIMER-ORIGINATED, injects the + # measured gap, and consumes exactly one digest without resetting the counters + # (aiqt_hooks.py:9325, 9334-9348). + self.write_registry(self.stage1()) + self.arm_lease() + prompt = "scheduled wake: recheck ci-7" + earlier = (_now() - timedelta(minutes=12)).isoformat() + self.seed_state("turn-state.json", { + "wake_digests": [_sha256_hex(prompt)], + "last_human_input_utc": earlier, + "stop_denials": 1, "schedule_denials": 2, + }) + r = self.run_hook("orch_prompt_stamp", {"prompt": prompt}) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + ctx = out["hookSpecificOutput"]["additionalContext"] + self.assertIn("TIMER-ORIGINATED", ctx) + self.assertIn("minutes", ctx) + ts = self.read_json("turn-state.json") + self.assertEqual(ts["wake_digests"], [], "the matched wake digest is consumed") + self.assertEqual(ts["last_human_input_utc"], earlier, "a timer wake never restamps human input") + self.assertEqual(ts["stop_denials"], 1, "a timer wake never resets the loop counters") + + def test_two_key_registry_writes_nothing(self): + # LOAD-BEARING contrast: two-key scope is inert, so no turn-state is written anywhere + # (aiqt_hooks.py:9317-9318). + self.write_registry(self.two_key()) + self.arm_lease() + r = self.run_hook("orch_prompt_stamp", {"prompt": "please continue"}) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(self.find_anywhere("turn-state.json"), [], + "the two-key registry must stamp nothing") + + +# ===================================================================================================== +# 4. orch_resume_audit (SessionStart, warn-only; aiqt_hooks.py:9765-9794) +# ===================================================================================================== +class ResumeAudit(HookHarness): + def test_wrong_branch_handoff_arms_barrier(self): + # A handoff naming a branch other than HEAD is a divergence finding; the audit warns and arms + # the resume barrier active:true (aiqt_hooks.py:9365-9370, 9783-9793). + self.write_registry(self.stage1()) + self.arm_lease() + self.write_record_files(handoff="Branch: some-other-branch\n") + r = self.run_hook("orch_resume_audit", {}) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertIn("resume audit", out["systemMessage"]) + self.assertIn("some-other-branch", out["systemMessage"]) + barrier = self.read_json("resume-barrier.json") + self.assertTrue(barrier["active"]) + + def test_stale_gate_marker_arms_barrier(self): + # A handoff gate marker citing a commit that is not HEAD is a divergence finding + # (aiqt_hooks.py:9371-9382). + self.write_registry(self.stage1()) + self.arm_lease() + self.write_record_files(handoff="Gate: green @ deadbeef1234\n") + r = self.run_hook("orch_resume_audit", {}) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertIn("gate marker", out["systemMessage"]) + self.assertTrue(self.read_json("resume-barrier.json")["active"]) + + def test_clean_audit_clears_barrier(self): + # With readable record surfaces, a fresh lease, and no divergence markers, the audit is clean: + # a silent allow and resume-barrier.json active:false (aiqt_hooks.py:9784, 9794). + # LOAD-BEARING: an ACTIVE barrier is SEEDED first (shape grounded at aiqt_hooks.py:9783-9785), + # so this proves the clean audit CLEARS an existing barrier (overwrites active:true -> false), + # not merely that a fresh barrier initializes inactive. orch_resume_audit writes the barrier with + # "active": bool(findings) unconditionally (mode "w", aiqt_hooks.py:9783-9785), so a no-findings + # audit overwrites the seed; removing that clear-write breaks this test. + self.write_registry(self.stage1()) + self.arm_lease() + self.write_record_files(handoff="Session handoff, no markers here.\n") + self.seed_state("resume-barrier.json", { + "active": True, "findings": ["seeded"], "ts": _now().isoformat(), "warned": False, + }) + self.assertTrue(self.read_json("resume-barrier.json")["active"], + "precondition: an active barrier is seeded before the clean audit") + r = self.run_hook("orch_resume_audit", {}) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(r.stdout.strip(), "", "a clean audit is a silent allow") + self.assertFalse(self.read_json("resume-barrier.json")["active"], + "a clean audit clears the seeded active barrier") + + def test_two_key_registry_finds_no_divergence(self): + # LOAD-BEARING contrast: the same wrong-branch handoff file is on disk, but the two-key registry + # declares no record surface, so the audit reads nothing and stays clean (aiqt_hooks.py:9355, + # 9420) -- proving record.handoff is what makes the finding possible. + self.write_registry(self.two_key()) + self.arm_lease() + self.write_record_files(handoff="Branch: some-other-branch\n") + r = self.run_hook("orch_resume_audit", {}) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(r.stdout.strip(), "") + # The two-key registry declares no state_dir, so the barrier lands at the XDG default. + self.assertFalse(self.read_json_anywhere("resume-barrier.json")["active"]) + + +# ===================================================================================================== +# 5. orch_resume_barrier (PreToolUse, warn-first BAKE posture; aiqt_hooks.py:9797-9845) +# ===================================================================================================== +class ResumeBarrier(HookHarness): + def _arm_barrier(self, warned=False): + self.seed_state("resume-barrier.json", { + "active": True, "warned": warned, "findings": ["a divergence"], "ts": _now().isoformat(), + }) + + def test_first_out_of_allowlist_write_warns_once(self): + # An active barrier surfaces the first mutation outside the allowlist exactly once, flipping + # warned:true (aiqt_hooks.py:9833-9845). + self.write_registry(self.stage1()) + self._arm_barrier(warned=False) + r = self.run_hook("orch_resume_barrier", { + "tool_name": "Write", "tool_input": {"file_path": self._p("some", "unrelated.py")}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertIn("resume barrier", out["systemMessage"]) + self.assertTrue(self.read_json("resume-barrier.json")["warned"]) + + def test_second_write_is_silent(self): + # Once warned, further out-of-allowlist writes are silent: surface once per arming, never a nag + # wall (aiqt_hooks.py:9833-9834). + self.write_registry(self.stage1()) + self._arm_barrier(warned=True) + r = self.run_hook("orch_resume_barrier", { + "tool_name": "Write", "tool_input": {"file_path": self._p("some", "unrelated.py")}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(r.stdout.strip(), "") + + def test_write_to_record_surface_is_silent(self): + # The record surfaces stay writable, so the only exit (correcting the record) is never obstructed + # (aiqt_hooks.py:9820-9832). + self.write_registry(self.stage1()) + self._arm_barrier(warned=False) + self.write_record_files() + r = self.run_hook("orch_resume_barrier", { + "tool_name": "Write", "tool_input": {"file_path": self._p("open-findings.md")}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(r.stdout.strip(), "", "a write to a record surface passes silently") + + def test_write_to_state_dir_is_silent(self): + self.write_registry(self.stage1()) + self._arm_barrier(warned=False) + r = self.run_hook("orch_resume_barrier", { + "tool_name": "Write", "tool_input": {"file_path": str(self.state_dir() / "scratch.json")}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(r.stdout.strip(), "") + + def test_write_to_registry_file_is_silent(self): + self.write_registry(self.stage1()) + self._arm_barrier(warned=False) + r = self.run_hook("orch_resume_barrier", { + "tool_name": "Write", + "tool_input": {"file_path": str(self.repo / ".aiqt" / "orchestration.json")}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(r.stdout.strip(), "") + + def test_record_key_is_load_bearing_for_allowlist(self): + # LOAD-BEARING contrast: drop only the `record` key (keeping state_dir so the barrier is found). + # The same write to the former findings surface is now OUTSIDE the allowlist and warns, proving + # the record key is what keeps the record surfaces writable under the barrier. + reg = self.stage1() + findings_path = reg["record"]["findings"] + del reg["record"] + self.write_registry(reg) + self._arm_barrier(warned=False) + r = self.run_hook("orch_resume_barrier", { + "tool_name": "Write", "tool_input": {"file_path": findings_path}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertIsNotNone(out, "without the record key the findings surface is no longer exempt") + self.assertIn("resume barrier", out["systemMessage"]) + + +# ===================================================================================================== +# 6. Adversarial: a corrupted (truncated-JSON) registry -> the bad-registry class (unchanged by stage 1) +# ===================================================================================================== +class CorruptedRegistry(HookHarness): + TRUNCATED = '{ "version": 1, "record": { "findings":' + + def test_dispatch_ledger_silent_allow(self): + # status != "ok" -> _allow(); no row (aiqt_hooks.py:9256-9257). + self.write_registry(self.TRUNCATED) + self.arm_lease() + r = self.run_hook("orch_dispatch_ledger", { + "tool_name": "Bash", "tool_input": {"run_in_background": True}, "tool_response": {}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(r.stdout.strip(), "") + self.assertEqual(self.find_anywhere("dispatch-ledger.jsonl"), []) + + def test_prompt_stamp_silent_allow(self): + # status != "ok" -> _allow() (aiqt_hooks.py:9314-9315). + self.write_registry(self.TRUNCATED) + self.arm_lease() + r = self.run_hook("orch_prompt_stamp", {"prompt": "hello"}) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(r.stdout.strip(), "") + self.assertEqual(self.find_anywhere("turn-state.json"), []) + + def test_resume_audit_warns_naming_the_registry(self): + # status == "bad" -> a warn naming the unreadable registry and barrier active:true + # (aiqt_hooks.py:9776-9793). + self.write_registry(self.TRUNCATED) + r = self.run_hook("orch_resume_audit", {}) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertIn("registry could not be read", out["systemMessage"]) + # A bad registry declares no state_dir, so the barrier lands at the XDG default. + self.assertTrue(self.read_json_anywhere("resume-barrier.json")["active"]) + + def test_resume_barrier_allows(self): + # status != "ok" -> _allow(), even with an ACTIVE barrier present (aiqt_hooks.py:9805-9807). + # LOAD-BEARING (codex QA round 5, LOW): on a BAD registry _orch_state_dir_for_root falls back to + # the XDG default (aiqt_hooks.py:7658-7659 -> _state_dir_from_registry(root, None)), NOT the + # registry-declared state_dir, so orch_resume_barrier reads its barrier from the XDG default. We + # ARM the barrier THERE by running orch_resume_audit first, which on a bad registry writes an + # active barrier to that XDG default (proven by test_resume_audit_warns_naming_the_registry). The + # real code takes the `status != "ok"` early-return and allows WITHOUT reading the barrier; a + # mutant that DELETES that early-return would fall through, read this active barrier, and reach the + # record-allowlist build where `reg` is the bad-registry detail STRING (aiqt_hooks.py:9821), + # changing the observed result (a fail-closed exit 2, not this clean exit-0 allow). The previous + # version seeded into the DECLARED state_dir, where the fall-through never looks, so the barrier + # was absent on both paths and the mutant survived. + self.write_registry(self.TRUNCATED) + self.run_hook("orch_resume_audit", {}) # arms an active barrier at the XDG default (bad registry) + self.assertTrue(self.read_json_anywhere("resume-barrier.json")["active"], + "precondition: an active barrier is armed at the XDG default the handler reads") + r = self.run_hook("orch_resume_barrier", { + "tool_name": "Write", "tool_input": {"file_path": self._p("x.py")}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(r.stdout.strip(), "") + + def test_stop_guard_fails_open_warn(self): + # DURABLE PIN of the fail-OPEN posture at aiqt_hooks.py:8579-8583, refuting the gemini seed's + # fail-closed claim: a bad registry warns (exit 0) and blocks NOTHING at the Stop boundary. + self.write_registry(self.TRUNCATED) + r = self.run_hook("orch_stop_guard", {}) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertIn("registry could not be read", out["systemMessage"]) + self.assertIn("fails open", out["systemMessage"]) + + def test_yield_tool_fails_closed_deny(self): + # PIN of the PRE-EXISTING fail-closed class (unchanged by stage 1): a bad registry denies a + # scheduling call. A PreToolUse deny is exit 0 with permissionDecision "deny" + # (aiqt_hooks.py:8654-8661, _deny at :195-201). + self.write_registry(self.TRUNCATED) + r = self.run_hook("orch_yield_tool", { + "tool_name": "ScheduleWakeup", "tool_input": {}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertEqual(out["hookSpecificOutput"]["permissionDecision"], "deny") + + +# ===================================================================================================== +# 7. Missing / unwritable state_dir (aiqt_hooks.py:7666, 9301-9303) +# ===================================================================================================== +class StateDirIO(HookHarness): + def test_missing_state_dir_is_created_on_demand(self): + # The recorders makedirs the state dir on demand and the row lands (aiqt_hooks.py:7666). + self.write_registry(self.stage1()) + self.arm_lease() + self.assertFalse(self.state_dir().exists(), "precondition: state_dir does not yet exist") + r = self.run_hook("orch_dispatch_ledger", { + "tool_name": "Bash", "tool_input": {"run_in_background": True}, "tool_response": {}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIsNotNone(self.read_jsonl("dispatch-ledger.jsonl")) + + @unittest.skipIf(os.geteuid() == 0, "root bypasses directory write permissions") + def test_unwritable_state_dir_surfaces_failed_write(self): + # An unwritable state dir makes the ledger append fail; it surfaces a non-blocking systemMessage + # and never denies (aiqt_hooks.py:9301-9303). The prompt stamp likewise exits 0. + self.write_registry(self.stage1()) + self.arm_lease() + sd = self.state_dir() + sd.mkdir(parents=True) + os.chmod(sd, 0o555) + try: + r = self.run_hook("orch_dispatch_ledger", { + "tool_name": "Bash", "tool_input": {"run_in_background": True}, "tool_response": {}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertIn("write failed", out["systemMessage"]) + r2 = self.run_hook("orch_prompt_stamp", {"prompt": "hello"}) + self.assertEqual(r2.returncode, 0, r2.stderr) + finally: + os.chmod(sd, 0o755) + + +# ===================================================================================================== +# 8. Stale lease (aiqt_hooks.py:7793-7798, 9400-9401) +# ===================================================================================================== +class StaleLease(HookHarness): + def test_recorders_noop_under_stale_lease(self): + # A lease older than max_age_hours is not fresh, so scope is not live and the recorders write + # nothing (aiqt_hooks.py:7798-7799). + self.write_registry(self.stage1()) + self.arm_lease(age_seconds=2 * 24 * 3600) # two days old, past the 24h horizon + r = self.run_hook("orch_dispatch_ledger", { + "tool_name": "Bash", "tool_input": {"run_in_background": True}, "tool_response": {}, + }) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(self.find_anywhere("dispatch-ledger.jsonl"), []) + + def test_audit_reports_stale_lease(self): + # The resume audit surfaces the staleness as a warn-only finding (aiqt_hooks.py:9400-9401). + self.write_registry(self.stage1()) + self.arm_lease(age_seconds=2 * 24 * 3600) + self.write_record_files(handoff="no markers\n") + r = self.run_hook("orch_resume_audit", {}) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertIn("stale", out["systemMessage"]) + self.assertTrue(self.read_json("resume-barrier.json")["active"]) + + +# ===================================================================================================== +# 9. Stop-family surface: stage-1 fixture, live lease, NO operator sentinel (aiqt_hooks.py:8196-8213, +# 8190-8214, 8542-8566) +# ===================================================================================================== +class StopFamilySurface(HookHarness): + def test_stop_denies_with_no_enumerator(self): + # With scope live and no enumerator declared, a turn-end Stop is DENIED once (exit 2) with the + # NO_ENUMERATOR reason (aiqt_hooks.py:7966-7968, 8196+8208-8213, 8590-8599). + self.write_registry(self.stage1()) + self.arm_lease() + r = self.run_hook("orch_stop_guard", {}) + self.assertEqual(r.returncode, 2, "a live-lease no-enumerator Stop denies (exit 2)") + self.assertIn("NO_ENUMERATOR", r.stderr) + + def test_followup_stop_releases_with_forced_exit_row(self): + # The follow-up Stop carrying stop_hook_active:true releases as ALLOW_WITH_FINDINGS (exit 0) and + # appends a forced-exit row (aiqt_hooks.py:8192-8195, 8601-8607, 8542-8561). + self.write_registry(self.stage1()) + self.arm_lease() + r = self.run_hook("orch_stop_guard", {"stop_hook_active": True}) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.stdout_json(r) + self.assertIn("loop bound", out["systemMessage"]) + rows = self.read_jsonl("forced-exit.jsonl") + self.assertIsNotNone(rows, "a forced release past an unenumerable backlog is recorded") + self.assertEqual(rows[0]["event"], "Stop") + + def test_self_owned_sentinel_is_ignored_and_spoof_recorded(self): + # A sentinel owned by the hook's OWN euid is not the operator's; it is ignored (the Stop still + # denies) and recorded as a spoof (aiqt_hooks.py:7897-7898, 8511-8521). + # + # NOTE: the sentinel HONOUR leg (a sentinel owned by a DIFFERENT uid opening the clean-ALLOW + # channel, aiqt_hooks.py:8190-8191) needs a second uid to create a foreign-owned file and cannot + # run in single-uid CI; it is verified at arming on the host with the root-owned sentinel. + self.write_registry(self.stage1()) + self.arm_lease() + sd = self.state_dir() + sd.mkdir(parents=True, exist_ok=True) + sentinel = sd / "ESCAPE-ALLOW-YIELD" + sentinel.write_text("", encoding="utf-8") + os.chmod(sentinel, 0o644) + self.assertEqual(os.stat(sentinel).st_uid, os.geteuid(), "precondition: self-owned sentinel") + r = self.run_hook("orch_stop_guard", {}) + self.assertEqual(r.returncode, 2, "a self-owned sentinel does not open the escape channel") + spoof = self.read_json("escape-spoof.json") + self.assertIsNotNone(spoof, "an ignored sentinel is spoof-recorded") + self.assertIn("uid", (spoof.get("detail") or "")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/test-orchestration-registry.py b/tools/test-orchestration-registry.py index 9bbd34c..bd19de5 100644 --- a/tools/test-orchestration-registry.py +++ b/tools/test-orchestration-registry.py @@ -24,6 +24,22 @@ REPO_ROOT = HERE.parent REAL_REGISTRY = REPO_ROOT / ".aiqt" / "orchestration.json" +# A full stage-1 registry with every optional arming key present, used as the accept fixture. The +# paths are illustrative absolutes (the gate validates STRUCTURE, not existence), so this stays +# independent of any one machine's store layout. +STAGE1 = { + "version": 1, + "companion_stores": ["/opt/x/private"], + "record": { + "findings": "/opt/x/private/open-findings.md", + "pending_decisions": "/opt/x/private/pending-decisions.md", + "handoff": "/opt/x/private/session-handoff.md", + }, + "lease": {"path": "/opt/x/private/session-state.md", "max_age_hours": 24}, + "state_dir": "/opt/x/private/orch-state", + "dispatch_tools": [], +} + def _run(path) -> int: """Run the validator against `path`; return its exit code.""" @@ -50,6 +66,20 @@ def test_committed_registry_is_valid(self): self.skipTest("no committed .aiqt/orchestration.json") self.assertEqual(_run(REAL_REGISTRY), 0) + def test_committed_registry_is_stage1_armed(self): + # The change-carries-check on the REAL file: the committed stage-1 registry must carry the + # recorder/audit arming keys AND validate. This assertion is RED on a tree that still ships the + # two-key registry (record/lease/state_dir absent) and green only once the arming change lands, + # so the gate's own test suite fails without the flip it guards. + if not REAL_REGISTRY.exists(): + self.fail("committed .aiqt/orchestration.json is missing; the stage-1 arming registry " + "must be present and armed") + obj = json.loads(REAL_REGISTRY.read_text(encoding="utf-8")) + for key in ("record", "lease", "state_dir", "dispatch_tools"): + self.assertIn(key, obj, "committed registry is missing the stage-1 arming key {!r}".format(key)) + self.assertIn("path", obj["lease"]) + self.assertEqual(_run(REAL_REGISTRY), 0) + class Accepts(unittest.TestCase): def test_absent_is_ok(self): @@ -74,9 +104,92 @@ def test_windows_absolute_path_accepted(self): with tempfile.TemporaryDirectory() as d: self.assertEqual(_run(_write(d, {"version": 1, "companion_stores": ["C:\\repo"]})), 0) + def test_full_stage1_fixture(self): + # A full stage-1 registry with every optional arming key present must validate. + with tempfile.TemporaryDirectory() as d: + self.assertEqual(_run(_write(d, STAGE1)), 0) + + def test_lease_without_max_age(self): + # max_age_hours is optional: a lease with only a path is valid. state_dir is declared because a + # declared lease now requires it (claude QA round 7 co-requirement); this test isolates max_age + # optionality, not the state_dir requirement. + with tempfile.TemporaryDirectory() as d: + self.assertEqual(_run(_write(d, {"version": 1, "lease": {"path": "/opt/x/lease"}, + "state_dir": "/opt/x/orch-state"})), 0) + + def test_dispatch_tools_with_strings(self): + with tempfile.TemporaryDirectory() as d: + self.assertEqual(_run(_write(d, {"version": 1, "dispatch_tools": ["Workflow", "Task"]})), 0) + + def test_record_sibling_of_state_dir_accepted(self): + # A record path that is a SIBLING of state_dir (shares a parent but is NOT inside it) is not a + # collision and must still validate. The containment check compares component-wise (commonpath), + # so "/opt/x/orch-state-extra/..." must not be read as inside "/opt/x/orch-state" (their commonpath + # is /opt/x, not the state_dir). (codex QA HIGH-1) + with tempfile.TemporaryDirectory() as d: + self.assertEqual(_run(_write(d, { + "version": 1, "state_dir": "/opt/x/orch-state", + "record": {"handoff": "/opt/x/orch-state-extra/session-handoff.md"}, + })), 0) + + def test_absent_state_dir_path_accepted(self): + # The best-effort non-directory check (codex QA HIGH-2) must NOT fire when the path does not exist: + # the gate cannot verify a not-yet-created directory, so an absent state_dir path stays valid. + with tempfile.TemporaryDirectory() as d: + self.assertEqual(_run(_write(d, { + "version": 1, "state_dir": str(Path(d) / "not-created-yet")})), 0) + + def test_existing_directory_state_dir_accepted(self): + # A state_dir that EXISTS as a real directory passes the best-effort gate-time type check. + with tempfile.TemporaryDirectory() as d: + sd = Path(d) / "orch-state" + sd.mkdir() + self.assertEqual(_run(_write(d, {"version": 1, "state_dir": str(sd)})), 0) + + def test_symlink_record_resolving_outside_state_dir_accepted(self): + # HIGH-1 (codex QA round 6) over-rejection guard: a record path through a symlink that resolves + # OUTSIDE state_dir must still be ACCEPTED. real dir D is state_dir; a SEPARATE real dir E; sibling + # symlink L -> E; record.handoff = "/session-handoff.md" resolves to E/session-handoff.md, which + # is not on or inside D. Both the lexical and the resolved-path checks must accept (exit 0), so the + # new resolved-path containment does not over-reject a legitimate symlinked record location. + with tempfile.TemporaryDirectory() as d: + state_dir = Path(d) / "state" + state_dir.mkdir() + elsewhere = Path(d) / "elsewhere" + elsewhere.mkdir() + alias = Path(d) / "alias" + os.symlink(str(elsewhere), str(alias)) + reg = _write(d, { + "version": 1, "state_dir": str(state_dir), + "record": {"handoff": str(alias / "session-handoff.md")}, + }) + self.assertEqual(_run(reg), 0) + + def test_absent_state_dir_deep_path_accepted(self): + # HIGH-2 (codex QA round 5) over-rejection guard: a GENUINELY-ABSENT state_dir (missing leaf AND + # missing parent) raises FileNotFoundError on lstat, which is absence -> accept (the not-yet-created + # residual stays disclosed). Only a NON-FileNotFound OSError (ENOTDIR/ENAMETOOLONG/...) is the + # known-unusable case the round-5 classification rejects, so this genuinely-absent path must stay + # GREEN after that change. RED would signal the classification over-rejects real absence. + with tempfile.TemporaryDirectory() as d: + self.assertEqual(_run(_write(d, { + "version": 1, "state_dir": str(Path(d) / "absent-parent" / "state")})), 0) + class Rejects(unittest.TestCase): - def _reject(self, payload): + # claude QA round 8 (MEDIUM, test-isolation): the round-7 gate rule "a declared record/lease requires + # state_dir" rejects ANY record/lease fixture that omits state_dir, regardless of whether the fixture's + # OWN targeted check fires -- so a regression of that targeted check would go undetected (proven by + # mutation). For a reject fixture that declares record/lease but no state_dir, inject a DISJOINT (no + # collision with the fixture's paths) and ABSENT (the gate's lstat type check stays inert) state_dir, so + # the round-7 rule is inert here and ONLY the fixture's targeted check can reject it. The two + # test_*_without_state_dir_rejected tests, whose target IS the rule, opt out with add_state_dir=False. + _ISO_STATE_DIR = "/opt/orch-isolate-sd" + + def _reject(self, payload, add_state_dir=True): + if (add_state_dir and isinstance(payload, dict) + and ("record" in payload or "lease" in payload) and "state_dir" not in payload): + payload = {**payload, "state_dir": self._ISO_STATE_DIR} with tempfile.TemporaryDirectory() as d: self.assertEqual(_run(_write(d, payload)), 1) @@ -141,6 +254,288 @@ def test_invalid_utf8(self): self.assertIn("FAIL", r.stderr) self.assertNotIn("Traceback", r.stderr) + def test_deeply_nested_json_is_clean_fail(self): + # Deeply-nested JSON overflows json's recursive scanner with RecursionError, which is NOT a + # ValueError. Without RecursionError in the parse except it would escape as an uncaught traceback + # (still exit 1, but an ugly diagnostic); with it the gate emits a clean FAIL. Assert exit 1, a + # FAIL diagnostic, and NO traceback. The payload is written as raw bytes, not via the dict-based + # _write helper (which would round-trip through json.dumps). + with tempfile.TemporaryDirectory() as d: + p = Path(d) / "orchestration.json" + p.write_bytes(b"[" * 100000 + b"]" * 100000) + r = subprocess.run( + [sys.executable, str(VALIDATOR), str(p)], capture_output=True, text=True, + ) + self.assertEqual(r.returncode, 1) + self.assertIn("FAIL", r.stderr) + self.assertNotIn("Traceback", r.stderr) + + # --- stage-1 arming keys ------------------------------------------------------------------------ + def test_unknown_top_level_key(self): + # `enumerator` is a real stage-2 key deliberately used here: it documents the stage-2 seam. + # A key outside the stage-1 allowlist lands only with its own gate extension, so it is rejected + # now rather than shipping an ahead-of-gate registry an installed reader would act on. + self._reject({"version": 1, "enumerator": {"argv": ["true"]}}) + + def test_record_not_a_dict(self): + self._reject({"version": 1, "record": "/opt/x/findings.md"}) + + def test_record_empty(self): + self._reject({"version": 1, "record": {}}) + + def test_record_unknown_subkey(self): + self._reject({"version": 1, "record": {"bogus": "/opt/x/findings.md"}}) + + def test_record_empty_path(self): + self._reject({"version": 1, "record": {"findings": ""}}) + + def test_record_relative_path(self): + self._reject({"version": 1, "record": {"findings": "relative/findings.md"}}) + + def test_record_control_char_path(self): + self._reject({"version": 1, "record": {"findings": "/opt/x/find\nings.md"}}) + + def test_record_nul_in_path(self): + # A NUL in a declared record path is the load-bearing case: it would raise inside the hook's + # orch_resume_barrier realpath walk (aiqt_hooks.py:9828) on a PreToolUse event, and that + # dispatcher fails closed exit 2 on a handler crash. The gate rejects it here. + self._reject({"version": 1, "record": {"findings": "/opt/x/find" + chr(0) + "ings.md"}}) + + def test_lease_missing_path(self): + self._reject({"version": 1, "lease": {"max_age_hours": 24}}) + + def test_lease_max_age_bool(self): + # bool is a subclass of int; True/False are not durations. + self._reject({"version": 1, "lease": {"path": "/opt/x/lease", "max_age_hours": True}}) + + def test_lease_max_age_zero(self): + self._reject({"version": 1, "lease": {"path": "/opt/x/lease", "max_age_hours": 0}}) + + def test_lease_max_age_negative(self): + self._reject({"version": 1, "lease": {"path": "/opt/x/lease", "max_age_hours": -1}}) + + def test_lease_max_age_string(self): + self._reject({"version": 1, "lease": {"path": "/opt/x/lease", "max_age_hours": "24"}}) + + def test_lease_max_age_over_horizon(self): + self._reject({"version": 1, "lease": {"path": "/opt/x/lease", "max_age_hours": 9999}}) + + def test_lease_unknown_subkey(self): + # holder_is_session_id is the settled omission: an unknown lease key is rejected, mechanically + # keeping it out of the committed registry. + self._reject({"version": 1, "lease": {"path": "/opt/x/lease", "holder_is_session_id": True}}) + + def test_state_dir_control_char(self): + self._reject({"version": 1, "state_dir": "/opt/x/orch\tstate"}) + + def test_state_dir_relative(self): + self._reject({"version": 1, "state_dir": "orch-state"}) + + def test_dispatch_tools_not_a_list(self): + self._reject({"version": 1, "dispatch_tools": "Workflow"}) + + def test_dispatch_tools_non_string_entry(self): + self._reject({"version": 1, "dispatch_tools": ["Workflow", 123]}) + + # --- record/lease path colliding with state_dir (codex QA HIGH) --------------------------------- + # A declared record/lease path ON or INSIDE state_dir collides with the machine-state files the hooks + # write there: orch_resume_audit opens /resume-barrier.json with mode "w" unconditionally + # (aiqt_hooks.py:9775, 9783-9785), so such a path would be clobbered by that hook write. The gate + # rejects the dangerous config. These are RED without the cross-field collision check (the per-key + # validation accepts each path in isolation). + def test_record_path_inside_state_dir_rejected(self): + self._reject({"version": 1, "state_dir": "/opt/x/orch-state", + "record": {"handoff": "/opt/x/orch-state/resume-barrier.json"}}) + + def test_lease_path_inside_state_dir_rejected(self): + self._reject({"version": 1, "state_dir": "/opt/x/orch-state", + "lease": {"path": "/opt/x/orch-state/session-state.md"}}) + + def test_record_path_equal_to_state_dir_rejected(self): + self._reject({"version": 1, "state_dir": "/opt/x/orch-state", + "record": {"handoff": "/opt/x/orch-state"}}) + + def test_record_without_state_dir_rejected(self): + # claude QA round 7 HIGH: a declared `record` with NO `state_dir` must be rejected. Without state_dir + # the hooks fall back to the XDG-default orch state dir (aiqt_hooks.py:7638-7651) that this gate never + # sees, and orch_resume_audit clobbers a record path landing inside it. RED before the co-requirement + # (the gate exited 0 accepting record-without-state_dir). + self._reject({"version": 1, "record": { + "handoff": "/home/x/.local/state/aiqt-guardrails/orch/deadbeef/resume-barrier.json"}}, + add_state_dir=False) + + def test_lease_without_state_dir_rejected(self): + # claude QA round 7 HIGH (sibling): a declared `lease` with NO `state_dir` must be rejected for the + # same reason -- lease.path can land in the same unseen XDG-default state dir. RED before the fix. + self._reject({"version": 1, "lease": {"path": "/opt/x/lease.md", "max_age_hours": 24}}, + add_state_dir=False) + + # --- codex QA round 4: path-spelling ALIASES, over-long paths, non-dir state_dir ---------------- + # These are RED against the round-3 normpath+startswith collision check (which accepted the aliases) + # and against the round-3 path validation (which had no length or state_dir-type check). + def test_state_dir_root_alias_collision_rejected(self): + # HIGH-1: state_dir "/" + a record path directly under it. The round-3 check accepted this because + # norm_state_dir "/" + os.sep = "//", which "/resume-barrier.json" does not start with. The + # commonpath check rejects it (commonpath(["/resume-barrier.json","/"]) == "/" == state_dir). + self._reject({"version": 1, "state_dir": "/", + "record": {"handoff": "/resume-barrier.json"}}) + + def test_double_slash_collision_in_record_rejected(self): + # HIGH-1: a leading-double-slash spelling of an inside-state_dir record path. POSIX normpath + # PRESERVES a leading "//", so the round-3 startswith missed it; _norm_pathspell collapses it. + self._reject({"version": 1, "state_dir": "/opt/x/orch-state", + "record": {"handoff": "//opt/x/orch-state/resume-barrier.json"}}) + + def test_double_slash_collision_in_state_dir_rejected(self): + # HIGH-1: the double-slash on the state_dir side instead; both spellings must reject. + self._reject({"version": 1, "state_dir": "//opt/x/orch-state", + "lease": {"path": "/opt/x/orch-state/session-state.md"}}) + + def test_lexical_collision_branch_symlink_out_of_state_dir(self): + # change-carries-check for the LEXICAL containment branch (claude QA round 10, mutation survivor). + # Disabling the lexical branch while keeping the realpath branch leaves the other collision fixtures + # green, because os.path.realpath already normalizes their "//"/".." aliases, so the resolved branch + # subsumes them. THIS fixture rejects ONLY via the lexical branch: a record path lexically INSIDE + # state_dir but whose symlink component RESOLVES OUTSIDE it. The lexical check (no symlink resolution) + # sees it inside and rejects; the realpath check resolves the symlink out and would accept. So the + # gate exits 1 here only while the lexical branch is present -- reverting that branch flips this RED. + with tempfile.TemporaryDirectory() as sd, \ + tempfile.TemporaryDirectory() as outside, \ + tempfile.TemporaryDirectory() as reg: + os.symlink(outside, os.path.join(sd, "link-out")) + payload = {"version": 1, "state_dir": sd, + "record": {"handoff": os.path.join(sd, "link-out", "f")}} + self.assertEqual(_run(_write(reg, payload)), 1) + + def test_overlong_state_dir_component_rejected(self): + # HIGH-2: a path component over NAME_MAX (255) bytes. At runtime, lstat(/write-scope.json) + # would raise ENAMETOOLONG, the write-scope reader would classify it 'bad', and an armed session + # would deny every covered write persistently; the gate rejects the over-long string at author time. + self._reject({"version": 1, "state_dir": "/opt/" + "a" * 256}) + + def test_overlong_record_component_rejected(self): + # HIGH-2: the same over-long component in a record path. + self._reject({"version": 1, "record": {"findings": "/opt/" + "a" * 256}}) + + def test_overlong_component_with_backslash_rejected(self): + # HIGH-2 (codex QA round 6): on POSIX a backslash is a VALID filename character, so + # "a"*150 + "\\" + "a"*150 is ONE 301-byte POSIX component, over NAME_MAX (255). Round 5's + # `re.split(r"[/\\]", ...)` split it AT the backslash into two <=255-byte fragments and wrongly + # ACCEPTED it, only to hit ENAMETOOLONG at runtime. Splitting on "/" ONLY measures the true 301-byte + # component and rejects. RED against the round-5 `[/\\]` split (which returns exit 0 here). + comp = "a" * 150 + "\\" + "a" * 150 # 301 bytes, ONE POSIX component + self.assertEqual(len(comp.encode()), 301, "fixture must be a 301-byte single POSIX component") + self._reject({"version": 1, "record": {"findings": "/opt/x/" + comp}}) + + def test_lease_path_at_path_max_rejected(self): + # codex QA round 8: PATH_MAX (4096) INCLUDES the NUL, so a path of exactly 4096 bytes is UNUSABLE + # (os.stat -> ENAMETOOLONG at runtime). The whole-path check must reject >= PATH_MAX, not just + # > PATH_MAX. Every component is <= NAME_MAX and state_dir is explicit+disjoint, so ONLY the + # whole-path length check can reject. RED against the `> PATH_MAX` check (exit 0). + base = "/" + "".join("a" * 255 + "/" for _ in range(15)) # 3841 bytes, ends with "/" + p4096 = base + "a" * (4096 - len(base.encode())) + self.assertEqual(len(p4096.encode()), 4096, "fixture must be exactly PATH_MAX bytes") + self._reject({"version": 1, "state_dir": "/opt/orch-isolate-sd", + "lease": {"path": p4096}}) + + def test_lease_path_one_below_path_max_accepted(self): + # The true maximum usable path is PATH_MAX - 1 (4095) bytes; it must still be ACCEPTED (not + # over-rejected by the >= change). Same shape, one byte shorter. + base = "/" + "".join("a" * 255 + "/" for _ in range(15)) + p4095 = base + "a" * (4095 - len(base.encode())) + self.assertEqual(len(p4095.encode()), 4095) + with tempfile.TemporaryDirectory() as d: + self.assertEqual(_run(_write(d, {"version": 1, "state_dir": "/opt/orch-isolate-sd", + "lease": {"path": p4095}})), 0) + + def test_state_dir_existing_non_directory_rejected(self): + # HIGH-2 best-effort: a state_dir that EXISTS at gate time as a regular file (the "/etc/passwd" + # shape) is rejected. A real temp regular file is created and state_dir points at it. At runtime + # write_scope_guard._load_write_scope would read /write-scope.json -> ENOTDIR -> 'bad' -> a + # persistent covered-write denial; the gate catches the gate-time type. RED without the fix. + with tempfile.TemporaryDirectory() as d: + real_file = Path(d) / "state-is-a-file" + real_file.write_text("not a directory\n", encoding="utf-8") + self.assertEqual(_run(_write(d, {"version": 1, "state_dir": str(real_file)})), 1) + + # --- codex QA round 5: derived-path reserve (HIGH-1) and known-unusable state_dir (HIGH-2) ------- + def test_state_dir_derived_path_reserve_rejected(self): + # HIGH-1: a state_dir WITHIN PATH_MAX whose length leaves no room for the machine-state files the + # hooks WRITE under it. Every component is <= NAME_MAX and the whole path <= PATH_MAX, so the + # per-component and whole-path checks BOTH pass; only the state_dir-specific reserve check rejects. + # At runtime lstat(/write-scope.json) would raise ENAMETOOLONG, classify the write-scope + # declaration 'bad', and drive a persistent covered-write denial. RED without STATE_DERIVED_RESERVE. + seg = "a" * 255 # NAME_MAX + sd = "/" + "/".join([seg] * 15) + "/" + "a" * 244 # 4085 bytes + self.assertLessEqual(len(sd.encode()), 4096, "fixture must stay within PATH_MAX") + self.assertGreater(len(sd.encode()) + 1 + 64, 4096, "fixture must overflow the derived reserve") + self._reject({"version": 1, "state_dir": sd}) + + def test_state_dir_path_through_regular_file_rejected(self): + # HIGH-2: a state_dir whose path runs THROUGH a regular file (the "/etc/passwd/state" shape). lstat + # raises ENOTDIR (a NON-FileNotFound OSError), which round-4 SWALLOWED as absence and accepted. The + # round-5 classification treats any non-FileNotFound stat fault as known-unusable and rejects, since + # the hook would hit the same ENOTDIR at runtime reading /write-scope.json -> 'bad' -> a + # persistent covered-write denial. RED without the fix. Hermetic: a real temp regular file stands in + # for /etc/passwd (verified at source: os.lstat("/etc/passwd/state") raises ENOTDIR). + with tempfile.TemporaryDirectory() as d: + real_file = Path(d) / "passwd" + real_file.write_text("root:x:0:0:root:/root:/bin/sh\n", encoding="utf-8") + self.assertEqual(_run(_write(d, { + "version": 1, "state_dir": str(real_file / "state")})), 1) + + def test_state_dir_trailing_slash_on_regular_file_rejected(self): + # HIGH-2: the "/etc/passwd/" shape -- a trailing separator on a path that IS a regular file also + # makes lstat raise ENOTDIR, not FileNotFoundError, so round-4 swallowed it as absence too. Round-5 + # rejects it. RED without the fix. (verified at source: os.lstat("/etc/passwd/") raises ENOTDIR.) + with tempfile.TemporaryDirectory() as d: + real_file = Path(d) / "passwd" + real_file.write_text("root:x:0:0:root:/root:/bin/sh\n", encoding="utf-8") + self.assertEqual(_run(_write(d, { + "version": 1, "state_dir": str(real_file) + "/"})), 1) + + # --- codex QA round 6: existing symlink alias defeats the LEXICAL containment check -------------- + def test_symlink_alias_record_into_state_dir_rejected(self): + # HIGH-1 (codex QA round 6): an EXISTING symlink makes two DIFFERENT spellings resolve to the same + # file, so a record path that LEXICALLY looks like a sibling of state_dir actually resolves ONTO a + # machine-state file inside it. Hermetic reproduction: real dir D is state_dir; sibling symlink + # L -> D; record.handoff = "/resume-barrier.json". Lexically norm(L/...) and norm(D) share only + # their parent tmp dir (commonpath != state_dir), so the lexical check ACCEPTS. The resolved-path + # check realpath's both -- realpath(L/resume-barrier.json) == realpath(D)/resume-barrier.json, + # inside realpath(D) -- and REJECTS. RED without the resolved-path check (gate would exit 0). + with tempfile.TemporaryDirectory() as d: + real_dir = Path(d) / "state" + real_dir.mkdir() + alias = Path(d) / "alias" + os.symlink(str(real_dir), str(alias)) + reg = _write(d, { + "version": 1, "state_dir": str(real_dir), + "record": {"handoff": str(alias / "resume-barrier.json")}, + }) + self.assertEqual(_run(reg), 1) + + # --- lone surrogate code points (unencodable to UTF-8) ------------------------------------------ + # A lone surrogate (U+D800..U+DFFF) passes an ord()<0x20-or-0x7f control-char test but CANNOT be + # UTF-8 encoded, so when the hook resolves such a declared path (os.path.realpath in + # orch_resume_barrier, aiqt_hooks.py:9828-9830, a PreToolUse event) the encode raises + # UnicodeEncodeError and the PreToolUse dispatcher fails closed exit 2 (aiqt_hooks.py:10705-10716) -- + # the very route the control-char rejection claims to close. The gate must reject surrogates in every + # declared-path surface. (codex QA HIGH) + def test_record_surrogate_path(self): + self._reject({"version": 1, "record": {"findings": "/opt/x/find" + chr(0xD800) + "ings.md"}}) + + def test_companion_surrogate(self): + self._reject({"version": 1, "companion_stores": ["/opt/x/" + chr(0xD800) + "store"]}) + + def test_state_dir_surrogate(self): + self._reject({"version": 1, "state_dir": "/opt/x/" + chr(0xD800) + "state"}) + + def test_lease_path_surrogate(self): + self._reject({"version": 1, "lease": {"path": "/opt/x/" + chr(0xD800) + "lease"}}) + + def test_dispatch_tools_surrogate(self): + self._reject({"version": 1, "dispatch_tools": ["Workflow", "Ta" + chr(0xD800) + "sk"]}) + if __name__ == "__main__": unittest.main()