"""Pure-code CI pass/fail gate — the deterministic block decision (design §3.3.2). This module is the §3.3.2 boundary #4: *"Pass/fail is a pure-code gate over authenticated CI results, not the LLM verifier."* Mirroring secrev's "one pure-code script owns the block decision", the gate reads the CI run **conclusion** (already fetched, authenticated as the box read-only PAT via the GitHub Checks/Actions API) keyed to a specific ``run_id`` + ``diff_hash`` and makes a deterministic ``pass | fail | block`` decision. It consumes **only** that authenticated, patch-independent conclusion; it never trusts a success/failure file or artifact the patch could have written. The gate also enforces the two box-side/CI defences that must hold before a diff is even allowed to build (defence in depth — the CI side enforces the same checks as a hard fail): * **boundary #2** — the **trust-control-surface denylist**: a candidate diff may not touch ``.github/workflows/**``, IAM/policy IaC, branch-protection / ``CODEOWNERS`` / Dependabot config, or paths outside the task's declared scope. A match is canonicalized (symlink/``..`` resolution) and rejects renames into denied paths, so a path match cannot be bypassed by indirection. * **boundary #3** — **diff integrity**: the gate recomputes the candidate diff hash and compares it to the ledger-recorded hash and to the hash CI verified, so a tampered or substituted diff fails closed. This module is pure stdlib and imports the committed foundation contracts verbatim (it redefines none of them). It performs **no** network I/O: the authenticated CI conclusion is passed in as data (the caller fetches it via the read-only PAT). Keeping the gate I/O-free is what makes the block decision deterministic and unit-testable. Decision semantics (a diff "ships" only on an unambiguous authenticated pass): * :data:`GateDecision.PASS` — the run concluded ``success`` for the exact ``run_id``/``diff_hash`` and every guard held; the verifier may open a draft PR. * :data:`GateDecision.FAIL` — the run concluded a recognised failure (``failure``/``timed_out``/``cancelled``/...); the verifier loops back to the builders. * :data:`GateDecision.BLOCK` — a trust violation (denylist hit, hash mismatch, run-id mismatch, missing/ambiguous authenticated conclusion). This is an ALARM-worthy refuse-to-proceed, never a silent pass. TRUST SOURCE (§3.3.2, QUESTION-1). ``state["run_id"]`` (the run id the fetcher reads and the value compared against ``expected_run_id`` here) and ``state["diff_hash"]`` MUST be written ONLY by the trusted dispatcher / ledger at dispatch time — NEVER by an LLM / builder / verifier node or by anything a candidate diff can influence. The dispatcher records the run id it dispatched the apply/verify workflow under (and the ledger-recorded diff hash); the LLM nodes only read them. The gate's run-id EQUALITY check below (``actual_run_id`` vs the dispatcher-supplied ``expected_run_id``) plus the fetcher's run_id format validation are the defense if that assumption is ever broken: a tampered ``state["run_id"]`` would still have to equal the dispatcher's expected run id to pass, and a malformed value fails closed. """ from __future__ import annotations import re from dataclasses import dataclass, field from enum import Enum from pathlib import PurePosixPath from typing import Any, Mapping, Sequence from agent_team.state_store import compute_content_hash __all__ = [ "DENYLIST_GLOBS", "CiGateError", "GateDecision", "GateResult", "denylist_violations", "diff_touched_paths", "evaluate_ci_gate", "gate_weakening_violations", "verify_diff_hash", ] class CiGateError(Exception): """Raised when the gate is called with structurally invalid inputs. Distinct from a :data:`GateDecision.BLOCK`: a ``BLOCK`` is a *valid* gate run that found a trust violation, whereas this exception means the caller handed the gate malformed data (e.g. a non-string diff). Fails loud rather than guessing. """ class GateDecision(Enum): """The deterministic gate outcome (§3.3.2 boundary #4).""" PASS = "pass" FAIL = "fail" BLOCK = "block" # Trust-control-surface denylist (§3.3.2 boundary #2). A candidate diff that # touches any of these is escalated to mandatory human + GPT cross-review, never # auto-built — they are the mandatory-cross-review surface regardless. Globs are # matched against POSIX-canonicalized repo-relative paths. # INJ-03: this denylist is the UNION SUPERSET shared verbatim across all three # trust-control copies — this tuple, the guard-job inline DENY_GLOBS, and the # post-build inline DENY_GLOBS in ci/agent-team-apply-verify.yml. The three had # drifted in BOTH directions (each carried entries the others lacked); they are # now identical, and tests/test_apply_verify_workflow_hardening.py asserts the # identity so any future drift fails CI. When editing one, edit all three. DENYLIST_GLOBS: tuple[str, ...] = ( # CI workflow definitions — the "pwn request" surface. ".github/workflows/**", ".github/actions/**", # Branch protection / ownership / dependency automation config. ".github/CODEOWNERS", "**/CODEOWNERS", "CODEOWNERS", ".github/dependabot.yml", ".github/dependabot.yaml", ".github/settings.yml", # IAM / policy IaC (CDK / SAM / Terraform / CloudFormation). "**/cdk.json", "**/template.yml", "**/template.yaml", "**/samconfig.toml", "**/*.tf", "**/*-stack.ts", "**/*_stack.py", "**/iam/**", "**/policies/**", "**/*iam*", "**/policy*.json", "**/*policy*.json", "**/*.pem", "**/*.key", # --- P3-flip §4.2: direct code-execution / supply-chain vectors --- # Submodule pointers / config — a changed submodule pulls in arbitrary # external code at the pinned commit. ".gitmodules", "**/.gitmodules", # Git hook directories / hook-path redirection — code that runs on git ops. # (`.git/hooks` itself is never in a checkout; `.husky` / `.githooks` are the # in-tree hook dirs a repo points `core.hooksPath` at.) ".husky/**", "**/.husky/**", ".githooks/**", "**/.githooks/**", # .gitattributes — clean/smudge filters execute arbitrary processes on checkout. ".gitattributes", "**/.gitattributes", # npm/registry config — can inject install scripts, a hostile registry, or auth. ".npmrc", "**/.npmrc", # Generated / build artifacts — codegen output is not reviewable as source, so # a diff that writes into it is escalated to a human rather than auto-built. "**/__generated__/**", "**/*.generated.*", "**/dist/**", "**/build/**", "**/*.min.js", # NOTE (deliberate, for the security gate): lockfiles are NOT wholesale denied # here. Lockfile-postinstall RCE is already contained by the credential-less, # egress-blocked build-test sandbox, and the Tier-3 dep-CVE fixer legitimately # rewrites lockfiles to produce its draft PRs — a wholesale lockfile deny would # make the fixer un-shippable. The direct code-execution config above (hooks, # filters, .npmrc, submodules) is the actual §4.2 RCE surface. Revisit if the # fixer's lockfile writes ever need a scoped allow vs a general deny. ) # Authenticated GitHub run conclusions that count as a recognised failure (the # verifier loops back). Anything not in PASS/this set is ambiguous -> BLOCK. _FAILURE_CONCLUSIONS: frozenset[str] = frozenset( {"failure", "timed_out", "cancelled", "action_required", "stale", "startup_failure"} ) # The single authenticated conclusion that means "ship-able". _SUCCESS_CONCLUSION = "success" # ``git diff`` file-header line, e.g. ``diff --git a/foo.py b/foo.py``. We read # the post-image (``b/``) path as the touched path and also surface the pre-image # (``a/``) so a *rename into* a denied path is caught (boundary #2). _DIFF_GIT_RE = re.compile(r"^diff --git a/(?P.+?) b/(?P.+?)\s*$") # ``rename from``/``rename to`` lines carry the rename source/target explicitly. _RENAME_FROM_RE = re.compile(r"^rename from (?P.+?)\s*$") _RENAME_TO_RE = re.compile(r"^rename to (?P.+?)\s*$") @dataclass class GateResult: """The gate's deterministic verdict plus the evidence behind it. ``decision`` is the block decision the verifier node acts on. ``reasons`` enumerates every concrete trigger (denylist hits, hash mismatch, the CI conclusion consumed) so the decision is auditable and an ALARM can quote it. ``run_id``/``diff_hash`` echo the keys the gate was bound to (provenance, §3.3.2). ``ci_conclusion`` is the authenticated conclusion actually consumed. """ decision: GateDecision reasons: list[str] = field(default_factory=list) run_id: str | None = None diff_hash: str | None = None ci_conclusion: str | None = None @property def passed(self) -> bool: """``True`` only on an unambiguous authenticated pass.""" return self.decision is GateDecision.PASS @property def blocked(self) -> bool: """``True`` on a trust violation (ALARM-worthy refuse-to-proceed).""" return self.decision is GateDecision.BLOCK def _canonical_repo_path(raw: str) -> str: """Canonicalize a repo-relative path for denylist matching (§3.3.2). Strips a leading ``a/``/``b/`` git prefix, normalizes separators, resolves ``.``/``..`` segments without touching the filesystem (the diff describes paths that may not exist locally), and drops a leading ``/`` so the result is always repo-relative. A path that escapes the repo root via ``..`` is returned with a sentinel ``..`` prefix preserved so it cannot silently match *nothing* — the caller treats an escaping path as a denylist hit. """ text = raw.strip().strip('"') # Drop a single git a//b/ prefix if present. if text.startswith(("a/", "b/")): text = text[2:] # PurePosixPath normalizes separators; resolve . and .. logically. parts: list[str] = [] for segment in PurePosixPath(text).parts: if segment in ("", "."): continue if segment == "..": # Escaping the repo root — keep the marker so it never matches a # benign glob and is treated as suspicious by the caller. parts.append("..") continue parts.append(segment) return "/".join(parts) def diff_touched_paths(unified_diff: str) -> list[str]: """Extract the canonicalized repo-relative paths a unified diff touches. Reads ``diff --git`` headers (both the ``a/`` pre-image and ``b/`` post-image) plus explicit ``rename from``/``rename to`` lines, so a rename *into* a denied path is surfaced (§3.3.2 boundary #2 — "rejects renames into denied paths"). Returns a de-duplicated, sorted list of canonical paths. Raises :class:`CiGateError` if ``unified_diff`` is not a string. """ if not isinstance(unified_diff, str): raise CiGateError(f"diff must be str, got {type(unified_diff).__name__}") touched: set[str] = set() for line in unified_diff.splitlines(): m = _DIFF_GIT_RE.match(line) if m is not None: touched.add(_canonical_repo_path(m.group("a"))) touched.add(_canonical_repo_path(m.group("b"))) continue m = _RENAME_FROM_RE.match(line) if m is not None: touched.add(_canonical_repo_path(m.group("path"))) continue m = _RENAME_TO_RE.match(line) if m is not None: touched.add(_canonical_repo_path(m.group("path"))) touched.discard("") return sorted(touched) def _glob_to_regex(glob: str) -> re.Pattern[str]: """Compile a denylist glob to an anchored regex. Supports ``**`` (any number of path segments, including zero), ``*`` (within a single segment), and ``?``. Everything else is matched literally. Matching is done on canonical POSIX repo-relative paths. """ out: list[str] = ["^"] i = 0 n = len(glob) while i < n: ch = glob[i] if ch == "*": if i + 1 < n and glob[i + 1] == "*": # ``**`` — any chars incl. ``/``. Swallow an optional trailing # ``/`` so ``dir/**`` also matches ``dir`` itself's children # without requiring a separator artifact. out.append(".*") i += 2 if i < n and glob[i] == "/": i += 1 continue # single ``*`` — anything but a path separator. out.append("[^/]*") i += 1 continue if ch == "?": out.append("[^/]") i += 1 continue out.append(re.escape(ch)) i += 1 out.append("$") return re.compile("".join(out)) _DENYLIST_RES: tuple[tuple[str, re.Pattern[str]], ...] = tuple( (g, _glob_to_regex(g)) for g in DENYLIST_GLOBS ) def denylist_violations( unified_diff: str, *, allowed_scope: Sequence[str] | None = None, ) -> list[str]: """Return the trust-control-surface violations in ``unified_diff`` (§3.3.2). A violation is any touched path that (a) matches a :data:`DENYLIST_GLOBS` entry, (b) escapes the repo root via ``..`` (path-indirection attempt), or (c) — when ``allowed_scope`` is given — falls outside the task's declared scope. ``allowed_scope`` is a sequence of canonical path prefixes (directories or exact files) the task is allowed to modify; a touched path outside every prefix is a violation ("files outside the task's declared scope"). Returns a sorted list of human-readable reason strings; an empty list means the diff is clean for the denylist boundary. """ violations: list[str] = [] normalized_scope = ( [_canonical_repo_path(p) for p in allowed_scope] if allowed_scope is not None else None ) for path in diff_touched_paths(unified_diff): if ".." in PurePosixPath(path).parts: violations.append(f"path escapes repo root via '..': {path!r}") continue for glob, pattern in _DENYLIST_RES: if pattern.match(path): violations.append(f"denylisted path {path!r} matches glob {glob!r}") break else: if normalized_scope is not None and not _within_scope( path, normalized_scope ): violations.append(f"path {path!r} is outside the task's declared scope") return sorted(violations) def _within_scope(path: str, scope: Sequence[str]) -> bool: """Return ``True`` if ``path`` is within one declared-scope prefix.""" candidate = PurePosixPath(path) for prefix in scope: if not prefix: continue if path == prefix: return True prefix_path = PurePosixPath(prefix) try: candidate.relative_to(prefix_path) return True except ValueError: continue return False # §4.5 gate-weakening: substrings whose appearance on an ADDED diff line means the # change is suppressing a quality/security gate (a self-passing move). Matched # case-insensitively against added-line content. _GATE_WEAKENING_MARKERS: tuple[tuple[str, str], ...] = ( ("noqa", "ruff/flake8 lint suppression (noqa)"), ("type: ignore", "type-check suppression (type: ignore)"), ("type:ignore", "type-check suppression (type:ignore)"), ("pragma: no cover", "coverage suppression (pragma: no cover)"), ("pragma: no-cover", "coverage suppression (pragma: no-cover)"), ("nosec", "bandit security suppression (nosec)"), ("nosemgrep", "semgrep suppression (nosemgrep)"), ("--no-verify", "git/pre-commit hook bypass (--no-verify)"), ) # Test skip/xfail decorators and calls (carry args, so regex not substring). _TEST_SKIP_RE: re.Pattern[str] = re.compile( r"@(?:pytest\.mark\.(?:skip|xfail)|unittest\.skip\w*)\b" r"|\bpytest\.(?:skip|xfail)\s*\(" r"|\.skipTest\s*\(" ) def _added_diff_lines(unified_diff: str) -> list[str]: """Return the content of lines ADDED by a unified diff (excluding ``+++`` headers).""" if not isinstance(unified_diff, str): raise CiGateError( f"unified_diff must be str, got {type(unified_diff).__name__}" ) return [ line[1:] for line in unified_diff.splitlines() if line.startswith("+") and not line.startswith("+++") ] def gate_weakening_violations(unified_diff: str) -> list[str]: """Return reasons the diff WEAKENS a quality/security gate (empty if none). A diff that ADDS a lint/type/coverage/security suppression, a test skip/xfail, or a hook bypass could make CI pass falsely — so the pure-code gate flags it even when the authenticated CI conclusion is ``success`` (§4.5). Such diffs are escalated to a human, never auto-built. Only ADDED lines are inspected (removing a suppression is fine); matching is case-insensitive on the marker text. """ violations: list[str] = [] for content in _added_diff_lines(unified_diff): low = content.lower() snippet = content.strip()[:120] for marker, desc in _GATE_WEAKENING_MARKERS: if marker in low: violations.append(f"gate-weakening: added {desc}: {snippet!r}") if _TEST_SKIP_RE.search(content): violations.append(f"gate-weakening: added a test skip/xfail: {snippet!r}") return violations def verify_diff_hash( candidate_diff: str, *, ledger_hash: str | None, ci_verified_hash: str | None = None, ) -> bool: """Verify the candidate diff hash matches the ledger (and CI) (§3.3.2 #3). Recomputes the content hash of ``candidate_diff`` (sha256, via the committed :func:`agent_team.state_store.compute_content_hash`) and compares it to the ``ledger_hash`` recorded by the builder and, when supplied, to the ``ci_verified_hash`` CI checked before applying. Comparison is constant-time. Returns ``True`` only when all supplied hashes agree; a ``None`` ledger hash is treated as a failure (there is nothing to bind to). Raises :class:`CiGateError` if ``candidate_diff`` is not a string. """ if not isinstance(candidate_diff, str): raise CiGateError( f"candidate_diff must be str, got {type(candidate_diff).__name__}" ) if not ledger_hash: return False actual = compute_content_hash(candidate_diff.encode("utf-8")) if not _consteq(actual, ledger_hash): return False if ci_verified_hash is not None and not _consteq(actual, ci_verified_hash): return False return True def _consteq(a: str, b: str) -> bool: """Constant-time string compare (hashes are not secret, but be tidy).""" if len(a) != len(b): return False result = 0 for x, y in zip(a, b): result |= ord(x) ^ ord(y) return result == 0 def evaluate_ci_gate( *, candidate_diff: str, ledger_hash: str | None, ci_result: Mapping[str, Any] | None, expected_run_id: str | None, allowed_scope: Sequence[str] | None = None, ) -> GateResult: """Make the deterministic pass/fail/block decision (§3.3.2 boundary #4). Inputs (all authenticated/patch-independent — the gate does no I/O): * ``candidate_diff`` — the builder's diff text (re-hashed here). * ``ledger_hash`` — the hash the builder recorded in the task ledger. * ``ci_result`` — the authenticated CI conclusion the caller fetched via the read-only PAT (GitHub Checks/Actions API). The gate reads only ``run_id``, ``conclusion``, and (optionally) ``diff_hash`` from it; it **never** reads a patch-written success file/artifact. * ``expected_run_id`` — the run id THIS task dispatched the apply/verify workflow under (per-task, sourced from ``state["run_id"]`` at node-run time, NOT a static wiring-time constant). The conclusion must be keyed to it (a stale/substituted run id is a BLOCK). A ``None``/empty value means the dispatcher captured no run id to bind to (e.g. the run-id poll fell through) — there is nothing to anchor the verdict against, so the gate BLOCKs (refuse-to-proceed, never a vacuous pass). * ``allowed_scope`` — optional declared-scope prefixes for the task. Decision order (a trust violation always wins over a CI verdict): 1. **Missing binding** — a ``None``/empty ``expected_run_id`` -> :data:`GateDecision.BLOCK` (no per-task run to bind the verdict to). 2. **Denylist / scope** — any violation -> :data:`GateDecision.BLOCK`. 3. **Diff-hash integrity** — hash mismatch (ledger or CI-verified) -> ``BLOCK``. 4. **Authenticated conclusion** — missing result, a ``run_id`` that does not match ``expected_run_id``, or an unrecognised/ambiguous conclusion -> ``BLOCK``; a recognised failure -> :data:`GateDecision.FAIL`; ``success`` -> :data:`GateDecision.PASS`. Returns a :class:`GateResult` with the decision and the reasons behind it. Raises :class:`CiGateError` on structurally invalid inputs (a non-string ``expected_run_id`` is structurally invalid; ``None``/empty is a BLOCK, not an exception, because it is the legitimate "no run captured" runtime state). """ if expected_run_id is not None and not isinstance(expected_run_id, str): raise CiGateError("expected_run_id must be a string or None") # (0) Missing per-task binding: the dispatcher captured no run id for this # task (None/empty). There is nothing to anchor the verdict to, so refuse to # proceed rather than gating against an empty string (which a substituted # ``ci_result`` with no/empty run_id could otherwise vacuously satisfy). if not expected_run_id: return GateResult( decision=GateDecision.BLOCK, reasons=["no per-task run_id to bind the verdict to (dispatch unresolved)"], run_id=None, diff_hash=ledger_hash, ci_conclusion=None, ) reasons: list[str] = [] # (1) Trust-control-surface denylist + declared scope. Highest priority: # these files are the mandatory-cross-review surface, never auto-built. violations = denylist_violations(candidate_diff, allowed_scope=allowed_scope) if violations: reasons.extend(violations) return GateResult( decision=GateDecision.BLOCK, reasons=reasons, run_id=expected_run_id, diff_hash=ledger_hash, ci_conclusion=None, ) # (1b) Gate-weakening (§4.5): a diff that suppresses a lint/type/coverage/ # security check or skips tests could make CI pass falsely -> BLOCK regardless # of the authenticated conclusion. A build cannot pass itself by disabling the # checks; these escalate to a human. weakening = gate_weakening_violations(candidate_diff) if weakening: reasons.extend(weakening) return GateResult( decision=GateDecision.BLOCK, reasons=reasons, run_id=expected_run_id, diff_hash=ledger_hash, ci_conclusion=None, ) # (2) Diff-hash integrity: bind the decision to the exact bytes the ledger # recorded (and, if present, what CI verified before applying). ci_verified_hash = ( str(ci_result.get("diff_hash")) if ci_result is not None and ci_result.get("diff_hash") is not None else None ) if not verify_diff_hash( candidate_diff, ledger_hash=ledger_hash, ci_verified_hash=ci_verified_hash, ): reasons.append( "diff-hash mismatch: candidate diff does not match the ledger" + (" / CI-verified" if ci_verified_hash is not None else "") + " hash" ) return GateResult( decision=GateDecision.BLOCK, reasons=reasons, run_id=expected_run_id, diff_hash=ledger_hash, ci_conclusion=None, ) # (3) Authenticated, patch-independent CI conclusion. if ci_result is None: reasons.append("no authenticated CI result supplied") return GateResult( decision=GateDecision.BLOCK, reasons=reasons, run_id=expected_run_id, diff_hash=ledger_hash, ci_conclusion=None, ) actual_run_id = ci_result.get("run_id") if str(actual_run_id) != expected_run_id: reasons.append( f"CI run-id mismatch: expected {expected_run_id!r}, " f"conclusion is keyed to {actual_run_id!r}" ) return GateResult( decision=GateDecision.BLOCK, reasons=reasons, run_id=expected_run_id, diff_hash=ledger_hash, ci_conclusion=None, ) conclusion = ci_result.get("conclusion") normalized = str(conclusion).strip().lower() if conclusion is not None else None if normalized == _SUCCESS_CONCLUSION: reasons.append("authenticated CI conclusion: success") return GateResult( decision=GateDecision.PASS, reasons=reasons, run_id=expected_run_id, diff_hash=ledger_hash, ci_conclusion=normalized, ) if normalized in _FAILURE_CONCLUSIONS: reasons.append(f"authenticated CI conclusion: {normalized}") return GateResult( decision=GateDecision.FAIL, reasons=reasons, run_id=expected_run_id, diff_hash=ledger_hash, ci_conclusion=normalized, ) # Unknown / null / still-running conclusion: refuse to proceed (never a # silent pass). e.g. ``None`` (in-progress), ``"neutral"``, ``"skipped"``. reasons.append( f"unrecognised/ambiguous CI conclusion {conclusion!r}; refusing to proceed" ) return GateResult( decision=GateDecision.BLOCK, reasons=reasons, run_id=expected_run_id, diff_hash=ledger_hash, ci_conclusion=normalized, )