diff --git a/.github/workflows/agent-completion-enforcement.yml b/.github/workflows/agent-completion-enforcement.yml index 1c4fd1b5d..61f18d8d4 100644 --- a/.github/workflows/agent-completion-enforcement.yml +++ b/.github/workflows/agent-completion-enforcement.yml @@ -44,7 +44,7 @@ jobs: if test -s trusted-report.json; then python3 scripts/ci/agent_completion_enforcement.py trusted-report.json .github/agent-lock/trusted-publishers.json "$head" "$PR" > enforcement-verdict.json else - printf '%s\n' '{"conclusion":"failure","reason":"missing_trusted_publication","details":{}}' > enforcement-verdict.json + python3 scripts/ci/agent_completion_enforcement.py --missing-publication .github/agent-lock/trusted-publishers.json > enforcement-verdict.json fi - name: Publish the required head-bound Check run if: always() @@ -71,8 +71,15 @@ jobs: } catch (error) { core.warning(error.message); } - const conclusion = verdict.conclusion === 'success' - ? 'success' + // Verifier currently emits success/failure/neutral. The remaining + // values are reserved for future use, but accepted here to avoid + // hard-failing if verifier capabilities are extended. + const allowedConclusions = new Set([ + 'success', 'failure', 'neutral', 'cancelled', 'timed_out', + 'action_required' + ]); + const conclusion = allowedConclusions.has(verdict.conclusion) + ? verdict.conclusion : 'failure'; const summary = JSON.stringify(verdict); await github.rest.checks.create({ @@ -89,6 +96,6 @@ jobs: summary: summary.slice(0, 60000) } }); - if (conclusion !== 'success') { + if (conclusion === 'failure') { core.setFailed(verdict.reason || 'trusted evidence blocked'); } diff --git a/scripts/ci/agent_completion_enforcement.py b/scripts/ci/agent_completion_enforcement.py index f746c6f20..d1b739fa5 100644 --- a/scripts/ci/agent_completion_enforcement.py +++ b/scripts/ci/agent_completion_enforcement.py @@ -21,6 +21,37 @@ def verdict(reason: str, **details: Any) -> dict[str, Any]: return {"conclusion": "failure", "reason": reason, "details": details} +def _is_non_empty_string_list(value: Any) -> bool: + return isinstance(value, list) and all( + isinstance(item, str) and len(item) > 0 for item in value + ) + + +def _extract_trust_policy_lists(policy: Any) -> tuple[list[str], list[str], list[str]] | None: + if not isinstance(policy, dict): + return None + apps = policy.get("trusted_check_app_slugs") + labels = policy.get("trusted_label_actors") + exemptions = policy.get("trusted_human_exemption_actors") + if not all(_is_non_empty_string_list(value) for value in (apps, labels, exemptions)): + return None + return apps, labels, exemptions + + +def missing_publication(policy: Any) -> dict[str, Any]: + trusted = _extract_trust_policy_lists(policy) + if trusted is None: + return verdict("invalid_trust_policy") + apps, labels, exemptions = trusted + if not apps or not labels or not exemptions: + return { + "conclusion": "neutral", + "reason": "missing_publication_policy_unprovisioned", + "details": {}, + } + return verdict("missing_trusted_publication") + + def verify(payload: Any, policy: Any, head_sha: str, pull_number: int) -> dict[str, Any]: if not isinstance(payload, dict) or not isinstance(policy, dict): return verdict("invalid_payload") @@ -28,12 +59,10 @@ def verify(payload: Any, policy: Any, head_sha: str, pull_number: int) -> dict[s return verdict("invalid_invocation") if policy.get("custom_role_policy") != "fail_closed": return verdict("invalid_custom_role_policy") - apps = policy.get("trusted_check_app_slugs") - labels = policy.get("trusted_label_actors") - exemptions = policy.get("trusted_human_exemption_actors") - if not all(isinstance(value, list) and all(isinstance(item, str) and item for item in value) - for value in (apps, labels, exemptions)): + trusted = _extract_trust_policy_lists(policy) + if trusted is None: return verdict("invalid_trust_policy") + apps, labels, exemptions = trusted if not apps or not labels or not exemptions: return verdict("trust_policy_unprovisioned") required = {"schema_version", "pull_number", "head_sha", "publisher", "applicability", "label_authorization", "focused_tests", "agent_events"} @@ -78,10 +107,23 @@ def verify(payload: Any, policy: Any, head_sha: str, pull_number: int) -> dict[s def main() -> int: + if len(sys.argv) == 3 and sys.argv[1] == "--missing-publication": + policy = json.loads(Path(sys.argv[2]).read_text()) + result = missing_publication(policy) + print(json.dumps(result, sort_keys=True)) + return 0 if result["conclusion"] != "failure" else 1 if len(sys.argv) != 5: - raise SystemExit("usage: verifier REPORT POLICY HEAD_SHA PULL_NUMBER") + raise SystemExit( + "usage: verifier REPORT POLICY HEAD_SHA PULL_NUMBER " + "or verifier --missing-publication POLICY" + ) report, policy, head, pull = sys.argv[1:] - result = verify(json.loads(Path(report).read_text()), json.loads(Path(policy).read_text()), head, int(pull)) + result = verify( + json.loads(Path(report).read_text()), + json.loads(Path(policy).read_text()), + head, + int(pull), + ) print(json.dumps(result, sort_keys=True)) return 0 if result["conclusion"] == "success" else 1 diff --git a/tests/unit/test_agent_completion_enforcement.py b/tests/unit/test_agent_completion_enforcement.py index 722a5cdd8..48f6bdb1d 100644 --- a/tests/unit/test_agent_completion_enforcement.py +++ b/tests/unit/test_agent_completion_enforcement.py @@ -1,6 +1,6 @@ import unittest -from scripts.ci.agent_completion_enforcement import verify +from scripts.ci.agent_completion_enforcement import missing_publication, verify HEAD = "a" * 40 @@ -30,3 +30,25 @@ def test_error_cannot_be_erased(self): def test_untrusted_label_blocks(self): body = report(); body["label_authorization"]["applied_by"] = "agent" self.assertEqual(verify(body, POLICY, HEAD, 9)["reason"], "untrusted_label_authorization") + + def test_missing_publication_is_neutral_when_policy_unprovisioned(self): + policy = dict( + POLICY, + trusted_check_app_slugs=[], + trusted_label_actors=[], + trusted_human_exemption_actors=[], + ) + result = missing_publication(policy) + self.assertEqual(result["conclusion"], "neutral") + self.assertEqual(result["reason"], "missing_publication_policy_unprovisioned") + + def test_missing_publication_fails_when_policy_is_provisioned(self): + policy = dict( + POLICY, + trusted_check_app_slugs=["agent-lock-trusted"], + trusted_label_actors=["maintainer"], + trusted_human_exemption_actors=["maintainer"], + ) + result = missing_publication(policy) + self.assertEqual(result["conclusion"], "failure") + self.assertEqual(result["reason"], "missing_trusted_publication")