Adam Moussa
e77ec6514f
First controls of the P3-live-flip Phase-1 CI hardening (workflow stays INERT; this only tightens the trust boundary). Whole Phase-1 surface is gated by /sh-security-review + GPT-4.1 cross-review before any flip. §4.2 — expand the trust-control denylist with direct code-execution / supply-chain vectors, kept byte-identical across all three copies (ci_gate.DENYLIST_GLOBS + the guard + post-build inline DENY_GLOBS), drift-guarded: .gitmodules, .husky/**, .githooks/**, .gitattributes, .npmrc, and generated/build artifacts (__generated__, *.generated.*, dist/**, build/**, *.min.js). Deliberate: lockfiles are NOT wholesale denied — lockfile-postinstall RCE is already contained by the credential-less egress-blocked build sandbox, and the Tier-3 dep-CVE fixer rewrites lockfiles to produce its draft PRs; a blanket deny would make it un-shippable. Flagged in-code for the security gate. Direct code-execution config (hooks/filters/npmrc/submodules) is the actual §4.2 RCE surface. §4.1 — runner-trust: assert no job (esp. the privileged gate-and-pr) can run on a self-hosted/user-provided runner; all must be GitHub-hosted. 998 tests pass, ruff clean.
413 lines
13 KiB
Python
413 lines
13 KiB
Python
"""Unit tests for agent_team.ci_gate — the pure-code pass/fail gate (§3.3.2)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from agent_team.ci_gate import (
|
|
DENYLIST_GLOBS,
|
|
CiGateError,
|
|
GateDecision,
|
|
denylist_violations,
|
|
diff_touched_paths,
|
|
evaluate_ci_gate,
|
|
verify_diff_hash,
|
|
)
|
|
from agent_team.state_store import compute_content_hash
|
|
|
|
|
|
def _diff_for(*paths: str) -> str:
|
|
"""Build a minimal unified diff touching ``paths`` (no rename)."""
|
|
chunks = []
|
|
for p in paths:
|
|
chunks.append(
|
|
f"diff --git a/{p} b/{p}\n--- a/{p}\n+++ b/{p}\n@@ -1 +1 @@\n-old\n+new\n"
|
|
)
|
|
return "".join(chunks)
|
|
|
|
|
|
def _ledger_hash(diff: str) -> str:
|
|
return compute_content_hash(diff.encode("utf-8"))
|
|
|
|
|
|
def _good_ci(run_id: str, diff: str, conclusion: str = "success") -> dict:
|
|
return {
|
|
"run_id": run_id,
|
|
"conclusion": conclusion,
|
|
"diff_hash": _ledger_hash(diff),
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# diff_touched_paths
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_touched_paths_basic() -> None:
|
|
diff = _diff_for("src/foo.py", "tests/test_foo.py")
|
|
assert diff_touched_paths(diff) == ["src/foo.py", "tests/test_foo.py"]
|
|
|
|
|
|
def test_touched_paths_strips_git_prefix_and_dedups() -> None:
|
|
diff = "diff --git a/pkg/mod.py b/pkg/mod.py\n@@ @@\n+x\n"
|
|
assert diff_touched_paths(diff) == ["pkg/mod.py"]
|
|
|
|
|
|
def test_touched_paths_includes_rename_lines() -> None:
|
|
diff = (
|
|
"diff --git a/safe.txt b/.github/workflows/evil.yml\n"
|
|
"similarity index 100%\n"
|
|
"rename from safe.txt\n"
|
|
"rename to .github/workflows/evil.yml\n"
|
|
)
|
|
paths = diff_touched_paths(diff)
|
|
assert ".github/workflows/evil.yml" in paths
|
|
assert "safe.txt" in paths
|
|
|
|
|
|
def test_touched_paths_non_string_raises() -> None:
|
|
with pytest.raises(CiGateError):
|
|
diff_touched_paths(None) # type: ignore[arg-type]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# denylist_violations
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_clean_diff_has_no_violations() -> None:
|
|
diff = _diff_for("src/foo.py", "README.md")
|
|
assert denylist_violations(diff) == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
".github/workflows/ci.yml",
|
|
".github/actions/deploy/action.yml",
|
|
".github/CODEOWNERS",
|
|
"CODEOWNERS",
|
|
".github/dependabot.yml",
|
|
"infra/cdk.json",
|
|
"service/template.yaml",
|
|
"deploy/iam/role.json",
|
|
"stacks/policies/admin.json",
|
|
"modules/main.tf",
|
|
# P3-flip §4.2 — direct code-execution / supply-chain vectors.
|
|
".gitmodules",
|
|
"vendor/.gitmodules",
|
|
".husky/pre-commit",
|
|
"frontend/.husky/commit-msg",
|
|
".githooks/pre-push",
|
|
".gitattributes",
|
|
"pkg/.gitattributes",
|
|
".npmrc",
|
|
"web/.npmrc",
|
|
"src/__generated__/schema.ts",
|
|
"api/types.generated.ts",
|
|
"web/dist/bundle.js",
|
|
"service/build/output.o",
|
|
"assets/app.min.js",
|
|
],
|
|
)
|
|
def test_denylisted_paths_flagged(path: str) -> None:
|
|
violations = denylist_violations(_diff_for(path))
|
|
assert violations, f"expected {path!r} to be denylisted"
|
|
|
|
|
|
def test_lockfiles_are_NOT_denylisted_so_the_tier3_fixer_can_ship() -> None:
|
|
"""Deliberate (§4.2 note): lockfile RCE is contained by the credential-less
|
|
egress-blocked build sandbox, and the dep-CVE fixer rewrites lockfiles to
|
|
produce draft PRs — so lockfiles are intentionally NOT on the denylist."""
|
|
for lock in ("package-lock.json", "web/yarn.lock", "poetry.lock", "Cargo.lock"):
|
|
assert denylist_violations(_diff_for(lock)) == [], (
|
|
f"{lock!r} must not be denylisted (would break the Tier-3 fixer)"
|
|
)
|
|
|
|
|
|
def test_rename_into_workflow_is_flagged() -> None:
|
|
diff = (
|
|
"diff --git a/safe.txt b/.github/workflows/evil.yml\n"
|
|
"rename from safe.txt\n"
|
|
"rename to .github/workflows/evil.yml\n"
|
|
)
|
|
violations = denylist_violations(diff)
|
|
assert any(".github/workflows/evil.yml" in v for v in violations)
|
|
|
|
|
|
def test_path_escape_via_dotdot_flagged() -> None:
|
|
diff = "diff --git a/x b/../../etc/passwd\n@@ @@\n+x\n"
|
|
violations = denylist_violations(diff)
|
|
assert any("escapes repo root" in v for v in violations)
|
|
|
|
|
|
def test_allowed_scope_blocks_out_of_scope_path() -> None:
|
|
diff = _diff_for("src/in_scope.py", "other/out_of_scope.py")
|
|
violations = denylist_violations(diff, allowed_scope=["src/"])
|
|
assert any("declared scope" in v for v in violations)
|
|
# In-scope path alone is clean.
|
|
assert (
|
|
denylist_violations(_diff_for("src/in_scope.py"), allowed_scope=["src/"]) == []
|
|
)
|
|
|
|
|
|
def test_allowed_scope_exact_file_prefix() -> None:
|
|
diff = _diff_for("pkg/exact.py")
|
|
assert denylist_violations(diff, allowed_scope=["pkg/exact.py"]) == []
|
|
|
|
|
|
def test_denylist_globs_exported_nonempty() -> None:
|
|
assert isinstance(DENYLIST_GLOBS, tuple)
|
|
assert ".github/workflows/**" in DENYLIST_GLOBS
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# verify_diff_hash
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_hash_matches_ledger() -> None:
|
|
diff = _diff_for("a.py")
|
|
assert verify_diff_hash(diff, ledger_hash=_ledger_hash(diff)) is True
|
|
|
|
|
|
def test_hash_mismatch_ledger() -> None:
|
|
diff = _diff_for("a.py")
|
|
assert verify_diff_hash(diff, ledger_hash="deadbeef") is False
|
|
|
|
|
|
def test_hash_none_ledger_fails() -> None:
|
|
diff = _diff_for("a.py")
|
|
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:
|
|
diff = _diff_for("a.py")
|
|
h = _ledger_hash(diff)
|
|
assert verify_diff_hash(diff, ledger_hash=h, ci_verified_hash=h) is True
|
|
assert verify_diff_hash(diff, ledger_hash=h, ci_verified_hash="nope") is False
|
|
|
|
|
|
def test_hash_non_string_raises() -> None:
|
|
with pytest.raises(CiGateError):
|
|
verify_diff_hash(123, ledger_hash="x") # type: ignore[arg-type]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# evaluate_ci_gate — the block decision
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_gate_pass_on_authenticated_success() -> None:
|
|
diff = _diff_for("src/foo.py")
|
|
result = evaluate_ci_gate(
|
|
candidate_diff=diff,
|
|
ledger_hash=_ledger_hash(diff),
|
|
ci_result=_good_ci("run-1", diff, "success"),
|
|
expected_run_id="run-1",
|
|
)
|
|
assert result.decision is GateDecision.PASS
|
|
assert result.passed is True
|
|
assert result.ci_conclusion == "success"
|
|
|
|
|
|
def test_gate_fail_on_authenticated_failure() -> None:
|
|
diff = _diff_for("src/foo.py")
|
|
result = evaluate_ci_gate(
|
|
candidate_diff=diff,
|
|
ledger_hash=_ledger_hash(diff),
|
|
ci_result=_good_ci("run-1", diff, "failure"),
|
|
expected_run_id="run-1",
|
|
)
|
|
assert result.decision is GateDecision.FAIL
|
|
assert result.ci_conclusion == "failure"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"conclusion", ["timed_out", "cancelled", "startup_failure", "action_required"]
|
|
)
|
|
def test_gate_recognised_failures(conclusion: str) -> None:
|
|
diff = _diff_for("src/foo.py")
|
|
result = evaluate_ci_gate(
|
|
candidate_diff=diff,
|
|
ledger_hash=_ledger_hash(diff),
|
|
ci_result=_good_ci("run-1", diff, conclusion),
|
|
expected_run_id="run-1",
|
|
)
|
|
assert result.decision is GateDecision.FAIL
|
|
|
|
|
|
def test_gate_blocks_denylisted_diff_even_if_ci_success() -> None:
|
|
# A diff that touches the trust-control surface BLOCKs regardless of CI.
|
|
diff = _diff_for(".github/workflows/ci.yml")
|
|
result = evaluate_ci_gate(
|
|
candidate_diff=diff,
|
|
ledger_hash=_ledger_hash(diff),
|
|
ci_result=_good_ci("run-1", diff, "success"),
|
|
expected_run_id="run-1",
|
|
)
|
|
assert result.decision is GateDecision.BLOCK
|
|
assert result.blocked is True
|
|
assert any("denylisted" in r for r in result.reasons)
|
|
|
|
|
|
def test_gate_blocks_on_hash_mismatch() -> None:
|
|
diff = _diff_for("src/foo.py")
|
|
result = evaluate_ci_gate(
|
|
candidate_diff=diff,
|
|
ledger_hash="deadbeef", # does not match recomputed 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_run_id_mismatch() -> None:
|
|
diff = _diff_for("src/foo.py")
|
|
result = evaluate_ci_gate(
|
|
candidate_diff=diff,
|
|
ledger_hash=_ledger_hash(diff),
|
|
ci_result=_good_ci("run-OTHER", diff, "success"),
|
|
expected_run_id="run-1",
|
|
)
|
|
assert result.decision is GateDecision.BLOCK
|
|
assert any("run-id" in r for r in result.reasons)
|
|
|
|
|
|
def test_gate_blocks_on_missing_ci_result() -> None:
|
|
diff = _diff_for("src/foo.py")
|
|
result = evaluate_ci_gate(
|
|
candidate_diff=diff,
|
|
ledger_hash=_ledger_hash(diff),
|
|
ci_result=None,
|
|
expected_run_id="run-1",
|
|
)
|
|
assert result.decision is GateDecision.BLOCK
|
|
|
|
|
|
@pytest.mark.parametrize("conclusion", [None, "neutral", "skipped", "in_progress"])
|
|
def test_gate_blocks_on_ambiguous_conclusion(conclusion) -> None:
|
|
# Anything not an explicit success or recognised failure must NOT silently
|
|
# pass — it BLOCKs (refuse-to-proceed).
|
|
diff = _diff_for("src/foo.py")
|
|
ci = _good_ci("run-1", diff, "success")
|
|
ci["conclusion"] = conclusion
|
|
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_ignores_patch_written_success_field() -> None:
|
|
# The gate reads only the authenticated conclusion. A patch-controlled
|
|
# "passed" flag must not flip a failing run to pass.
|
|
diff = _diff_for("src/foo.py")
|
|
ci = _good_ci("run-1", diff, "failure")
|
|
ci["passed"] = True # attacker-controlled artifact field
|
|
ci["success"] = "true"
|
|
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.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:
|
|
# CI verified a different diff than the ledger recorded -> BLOCK.
|
|
diff = _diff_for("src/foo.py")
|
|
ci = _good_ci("run-1", diff, "success")
|
|
ci["diff_hash"] = "tampered"
|
|
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_scope_violation_blocks() -> None:
|
|
diff = _diff_for("src/foo.py", "unrelated/bar.py")
|
|
result = evaluate_ci_gate(
|
|
candidate_diff=diff,
|
|
ledger_hash=_ledger_hash(diff),
|
|
ci_result=_good_ci("run-1", diff, "success"),
|
|
expected_run_id="run-1",
|
|
allowed_scope=["src/"],
|
|
)
|
|
assert result.decision is GateDecision.BLOCK
|
|
assert any("scope" in r for r in result.reasons)
|
|
|
|
|
|
def test_gate_empty_run_id_raises() -> None:
|
|
diff = _diff_for("src/foo.py")
|
|
with pytest.raises(CiGateError):
|
|
evaluate_ci_gate(
|
|
candidate_diff=diff,
|
|
ledger_hash=_ledger_hash(diff),
|
|
ci_result=_good_ci("run-1", diff),
|
|
expected_run_id="",
|
|
)
|
|
|
|
|
|
def test_gate_result_carries_provenance() -> None:
|
|
diff = _diff_for("src/foo.py")
|
|
h = _ledger_hash(diff)
|
|
result = evaluate_ci_gate(
|
|
candidate_diff=diff,
|
|
ledger_hash=h,
|
|
ci_result=_good_ci("run-7", diff, "success"),
|
|
expected_run_id="run-7",
|
|
)
|
|
assert result.run_id == "run-7"
|
|
assert result.diff_hash == h
|
|
assert result.reasons # always records why
|