Adam Moussa
7d8195bdd2
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.
388 lines
12 KiB
Python
388 lines
12 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",
|
|
],
|
|
)
|
|
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_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
|