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.
This commit is contained in:
parent
7d8195bdd2
commit
9d7d03d6db
5 changed files with 639 additions and 55 deletions
|
|
@ -40,6 +40,7 @@ from __future__ import annotations
|
|||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
|
|
@ -54,6 +55,52 @@ __all__ = [
|
|||
|
||||
_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
|
||||
|
|
@ -138,6 +185,28 @@ def fetch_ci_result(
|
|||
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:
|
||||
|
|
@ -181,15 +250,43 @@ def fetch_ci_result(
|
|||
_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 run_id. The gate independently re-checks this
|
||||
# against expected_run_id, so this is provenance, not the trust decision.
|
||||
# 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")
|
||||
result_run_id = str(fetched_id) if fetched_id is not None else run_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": conclusion,
|
||||
"conclusion": normalized_conclusion,
|
||||
"diff_hash": echoed_hash,
|
||||
}
|
||||
|
||||
|
|
@ -199,20 +296,26 @@ def build_ci_result_fetcher(
|
|||
owner: str,
|
||||
repo: str,
|
||||
client: Any = None,
|
||||
api_root: str = GITHUB_API_ROOT,
|
||||
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`` / ``api_root``
|
||||
/ ``timeout``). Compose it with
|
||||
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:
|
||||
|
|
@ -221,7 +324,9 @@ def build_ci_result_fetcher(
|
|||
owner=owner,
|
||||
repo=repo,
|
||||
client=client,
|
||||
api_root=api_root,
|
||||
# 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,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
run-id mismatch, missing/ambiguous authenticated conclusion). This is an
|
||||
ALARM-worthy refuse-to-proceed, never a silent pass.
|
||||
|
||||
TRUST SOURCE (§3.3.2, QUESTION-1). ``state["run_id"]`` (the run id the fetcher
|
||||
reads and the value compared against ``expected_run_id`` here) and
|
||||
``state["diff_hash"]`` MUST be written ONLY by the trusted dispatcher / ledger at
|
||||
dispatch time — NEVER by an LLM / builder / verifier node or by anything a
|
||||
candidate diff can influence. The dispatcher records the run id it dispatched the
|
||||
apply/verify workflow under (and the ledger-recorded diff hash); the LLM nodes
|
||||
only read them. The gate's run-id EQUALITY check below (``actual_run_id`` vs the
|
||||
dispatcher-supplied ``expected_run_id``) plus the fetcher's run_id format
|
||||
validation are the defense if that assumption is ever broken: a tampered
|
||||
``state["run_id"]`` would still have to equal the dispatcher's expected run id to
|
||||
pass, and a malformed value fails closed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -85,12 +97,19 @@ class GateDecision(Enum):
|
|||
# touches any of these is escalated to mandatory human + GPT cross-review, never
|
||||
# auto-built — they are the mandatory-cross-review surface regardless. Globs are
|
||||
# matched against POSIX-canonicalized repo-relative paths.
|
||||
# INJ-03: this denylist is the UNION SUPERSET shared verbatim across all three
|
||||
# trust-control copies — this tuple, the guard-job inline DENY_GLOBS, and the
|
||||
# post-build inline DENY_GLOBS in ci/agent-team-apply-verify.yml. The three had
|
||||
# drifted in BOTH directions (each carried entries the others lacked); they are
|
||||
# now identical, and tests/test_apply_verify_workflow_hardening.py asserts the
|
||||
# identity so any future drift fails CI. When editing one, edit all three.
|
||||
DENYLIST_GLOBS: tuple[str, ...] = (
|
||||
# CI workflow definitions — the "pwn request" surface.
|
||||
".github/workflows/**",
|
||||
".github/actions/**",
|
||||
# Branch protection / ownership / dependency automation config.
|
||||
".github/CODEOWNERS",
|
||||
"**/CODEOWNERS",
|
||||
"CODEOWNERS",
|
||||
".github/dependabot.yml",
|
||||
".github/dependabot.yaml",
|
||||
|
|
@ -101,9 +120,15 @@ DENYLIST_GLOBS: tuple[str, ...] = (
|
|||
"**/template.yaml",
|
||||
"**/samconfig.toml",
|
||||
"**/*.tf",
|
||||
"**/*-stack.ts",
|
||||
"**/*_stack.py",
|
||||
"**/iam/**",
|
||||
"**/policies/**",
|
||||
"**/*iam*",
|
||||
"**/policy*.json",
|
||||
"**/*policy*.json",
|
||||
"**/*.pem",
|
||||
"**/*.key",
|
||||
)
|
||||
|
||||
# Authenticated GitHub run conclusions that count as a recognised failure (the
|
||||
|
|
|
|||
|
|
@ -162,23 +162,32 @@ jobs:
|
|||
# an auto-reject; such a diff is escalated to mandatory human + GPT
|
||||
# cross-review, never auto-built (these are the mandatory-cross-review
|
||||
# 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, ...] = (
|
||||
".github/workflows/**",
|
||||
".github/actions/**",
|
||||
".github/CODEOWNERS",
|
||||
"**/CODEOWNERS",
|
||||
"CODEOWNERS",
|
||||
".github/dependabot.yml",
|
||||
".github/dependabot.yaml",
|
||||
".github/settings.yml",
|
||||
# IAM / policy / permission IaC (CDK / SAM / Terraform).
|
||||
"**/template.yaml",
|
||||
"**/template.yml",
|
||||
"**/*.tf",
|
||||
"**/cdk.json",
|
||||
"**/template.yml",
|
||||
"**/template.yaml",
|
||||
"**/samconfig.toml",
|
||||
"**/*.tf",
|
||||
"**/*-stack.ts",
|
||||
"**/*_stack.py",
|
||||
"**/policy*.json",
|
||||
"**/iam/**",
|
||||
"**/policies/**",
|
||||
"**/*iam*",
|
||||
"**/policy*.json",
|
||||
"**/*policy*.json",
|
||||
"**/*.pem",
|
||||
"**/*.key",
|
||||
)
|
||||
|
|
@ -566,15 +575,26 @@ jobs:
|
|||
# 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. Collect it
|
||||
# via `git diff --name-only` (tracked changes the hook made on top of
|
||||
# the patch) + `git status --porcelain` (new untracked files the hook
|
||||
# dropped) and FAIL the job if any of it lands on a denied path or
|
||||
# outside scope. Reuses the SAME denylist + scope logic as the guard.
|
||||
{
|
||||
git diff --name-only HEAD
|
||||
git status --porcelain --untracked-files=all | sed -E 's/^...//'
|
||||
} | sort -u > ./_build_written_paths.txt
|
||||
# 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
|
||||
|
||||
|
|
@ -583,30 +603,44 @@ jobs:
|
|||
import re
|
||||
import sys
|
||||
|
||||
# SAME denylist as the guard job (boundary 2). Kept in sync by review.
|
||||
# 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",
|
||||
"**/template.yaml",
|
||||
"**/template.yml",
|
||||
"**/*.tf",
|
||||
"**/cdk.json",
|
||||
"**/template.yml",
|
||||
"**/template.yaml",
|
||||
"**/samconfig.toml",
|
||||
"**/*.tf",
|
||||
"**/*-stack.ts",
|
||||
"**/*_stack.py",
|
||||
"**/policy*.json",
|
||||
"**/iam/**",
|
||||
"**/policies/**",
|
||||
"**/*iam*",
|
||||
"**/policy*.json",
|
||||
"**/*policy*.json",
|
||||
"**/*.pem",
|
||||
"**/*.key",
|
||||
)
|
||||
_GLOB_META = set("*?[]")
|
||||
|
||||
def canonical(path: str) -> str:
|
||||
p = path.strip().strip('"').replace("\\", "/")
|
||||
# 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}")
|
||||
|
|
@ -662,10 +696,89 @@ jobs:
|
|||
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:
|
||||
with open("./_build_written_paths.txt", encoding="utf-8") as fh:
|
||||
raw_paths = [ln for ln in (x.strip() for x in fh) if ln]
|
||||
if not raw_paths:
|
||||
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
|
||||
|
||||
|
|
@ -673,7 +786,6 @@ jobs:
|
|||
[s for s in os.environ.get("DECLARED_SCOPE", "").splitlines() if s.strip()]
|
||||
)
|
||||
|
||||
violations: list[str] = []
|
||||
for raw in raw_paths:
|
||||
try:
|
||||
path = canonical(raw)
|
||||
|
|
@ -762,28 +874,40 @@ jobs:
|
|||
# ───────────────────────────────────────────────────────────────────────────
|
||||
gate-and-pr:
|
||||
needs: [guard, build-test]
|
||||
# `always()` so the gate runs even when build-test failed, to record the
|
||||
# authoritative conclusion. The gate itself decides pass/fail from results.
|
||||
if: always()
|
||||
# Defense-in-depth: the privileged job NEVER runs on a failed/skipped/
|
||||
# cancelled guard or build-test. We still want it to run on a build-test
|
||||
# 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
|
||||
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. Until the
|
||||
# environment exists, dispatching this workflow is itself blocked, which is
|
||||
# why the live flip is deferred to provisioning.
|
||||
environment:
|
||||
name: agent-apply
|
||||
# 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:
|
||||
contents: read
|
||||
# The ONLY privileged grant: open a draft PR. Enabled here (no longer
|
||||
# commented) because the auth model is locked — there is no cloud
|
||||
# credential to federate. The draft-PR STEP stays `if: ${{ false }}` until
|
||||
# provisioning, so nothing privileged actually runs yet even with the
|
||||
# grant declared.
|
||||
pull-requests: write
|
||||
# uncomment at provisioning AFTER the agent-apply environment is created with a required reviewer
|
||||
# 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:
|
||||
|
|
@ -934,7 +1058,11 @@ jobs:
|
|||
# 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.
|
||||
title="agent-apply: ${TASK_ID} (diff ${DIFF_HASH})"
|
||||
# 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 \
|
||||
|
|
|
|||
|
|
@ -9,9 +9,13 @@ later edit cannot silently regress it:
|
|||
* 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 declares ``pull-requests: write`` and the ``agent-apply``
|
||||
environment, and download-artifact steps are run-id pinned.
|
||||
* 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
|
||||
|
|
@ -137,14 +141,47 @@ def test_draft_pr_step_is_not_flipped_live() -> None:
|
|||
)
|
||||
|
||||
|
||||
def test_gate_job_declares_pr_write_and_environment() -> None:
|
||||
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 {}
|
||||
assert perms.get("pull-requests") == "write"
|
||||
env = job.get("environment")
|
||||
# environment may be a mapping {name: agent-apply} or a bare string.
|
||||
name = env.get("name") if isinstance(env, dict) else env
|
||||
assert name == "agent-apply"
|
||||
# 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:
|
||||
|
|
@ -238,3 +275,137 @@ def test_workflow_gate_blocks_on_empty_guard_hash(tmp_path: Path) -> None:
|
|||
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
|
||||
|
|
|
|||
|
|
@ -200,3 +200,158 @@ 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")
|
||||
|
|
|
|||
Reference in a new issue