diff --git a/.github/workflows/knowledge-model-policy.yml b/.github/workflows/knowledge-model-policy.yml index ad65f358..288fbe26 100644 --- a/.github/workflows/knowledge-model-policy.yml +++ b/.github/workflows/knowledge-model-policy.yml @@ -17,9 +17,23 @@ permissions: jobs: declarations: - if: github.event_name != 'issue_comment' || github.event.issue.pull_request + # Keep eligibility identical here and in the group: skipped comments must not + # replace a pending writer, regardless of GitHub's evaluation order. + if: >- + ${{ github.event_name != 'issue_comment' || !github.event.issue || + !github.event.comment || !github.event.action || + (github.event.issue.pull_request && (github.event.action != 'created' || + !github.event.comment.body || + contains(github.event.comment.body, 'knowledge-policy-approval'))) }} concurrency: - group: knowledge-policy-status-writer + group: >- + ${{ (github.event_name != 'issue_comment' || !github.event.issue || + !github.event.comment || !github.event.action || + (github.event.issue.pull_request && (github.event.action != 'created' || + !github.event.comment.body || + contains(github.event.comment.body, 'knowledge-policy-approval')))) && + 'knowledge-policy-status-writer' || + format('knowledge-policy-skipped-{0}-{1}', github.run_id, github.run_attempt) }} cancel-in-progress: false runs-on: ubuntu-latest timeout-minutes: 10 @@ -35,3 +49,4 @@ jobs: env: GITHUB_TOKEN: ${{ github.token }} run: bash tools/run_knowledge_model_policy.sh + diff --git a/docs/knowledge-model-policy.md b/docs/knowledge-model-policy.md index 2c7bdc64..c4361e44 100644 --- a/docs/knowledge-model-policy.md +++ b/docs/knowledge-model-policy.md @@ -310,7 +310,7 @@ rejection neither greens nor fails another PR's status. The workflow handles PR open/reopen/close/commits/body-or-base edits on any target branch (a PR against another base can share a main PR's head), approval-comment creation/edit/deletion, every main push and manual dispatch. Main pushes refresh -all open main-target PRs. All triggers share one serialized status writer and each +all open main-target PRs. All eligible triggers share one serialized status writer and each run refreshes every open main-target PR, so replacing a pending job cannot lose another PR's refresh or let an older run overwrite a newer result. It checks out `main`, executes only trusted code, installs no PR @@ -321,8 +321,14 @@ approvals are reread before publication; changes fail closed and need a fresh ru PR edits and status publication are not atomic: integrators still check the current head and latest run. -The shared concurrency group belongs to the declaration job. Ordinary Issue -comments skip that job and do not displace queued PR/base refreshes. +The shared concurrency group belongs to the declaration job. Ordinary non-PR Issue +comments skip that job. A newly created, nonempty PR comment without the +`knowledge-policy-approval` marker also skips the job; every PR comment edit or +deletion remains eligible, including removal of an approval marker. Missing event +fields or an empty body are handled conservatively. The same eligibility expression +controls both the job condition and its concurrency group: skipped events use a +unique no-op group, never the status writer group. They therefore cannot replace +a pending legitimate refresh, regardless of GitHub's condition/queue evaluation order. File pagination must exactly match `changed_files`; more than the GitHub files API's 3,000-file limit fails visibly. Exception/policy approvals require complete @@ -331,6 +337,56 @@ never become compliant/non-applicable results. Current rules have no merge queue `merge_group` is explicitly unsupported. Validate a trusted integration before enabling a queue with this required check. +HTTP diagnostics retain only the response status, a conservative category, +numeric rate-limit timing/remaining headers, and a bounded GitHub request ID. +Raw response messages, error arrays, request URLs and credentials are not logged. +A 403 alone is not proof of rate limiting; a 422 is a validation error, not proof +that the status-history cap was reached. Read requests may retry once when an +explicit rate-limit delay fits inside their remaining retry budget (at most 60 +seconds). The wait is deducted before passing the remaining timeout to the retry. +The urllib timeout applies to socket operations, not a strict total wall-clock +deadline for reading the response. +Status POSTs are never retried by the transport. The existing outer sweep retry +plan remains unchanged; this diagnostic improvement does not repair permissions, +restore exhausted status capacity, or guarantee that historical red checks clear. + + +### Trigger-PR conclusion versus global sweep health + +Every eligible workflow event still refreshes **all open policy-branch PRs** through the +single serialized writer. This is necessary because concurrency may replace a +pending event. The runner still makes up to three attempts when global work is +unfinished; a green triggering PR never stops those retries early. The triggering +PR is evaluated first without dropping any peer from the sweep. + +Only after the final attempt does a PR or PR-comment job use that triggering PR's +own verified result for its job conclusion. A `pull_request_target` result must +match the immutable event head; rerunning an old event cannot turn its historical +head green using a newer head's compliance. Missing/malformed event heads and +advanced heads fail closed with an explicit stale-event reason. PR comments lack +a head snapshot and retain current-head semantics. A peer PR's history/publication error +therefore does not turn a verified compliant trigger PR red. A noncompliant +trigger PR, unavailable evidence, status-publication failure, capacity failure, +advanced head, or missing target still fails its own job. A freshly checked +closed/non-policy-branch target is explicitly **not applicable**, not an admission +or acceptance PASS. It is added to the sweep even when absent from the open list. +Push-to-main and manual workflow-dispatch jobs continue to report global execution +health. Explicit CLI `--pull-request` and dry-run behavior is unchanged. + +The runner deletes the report before each attempt. The checker atomically writes +a report bound to repository, event digest/type, policy SHA and attempt identifier; +the final consumer rejects missing, malformed, duplicate, stale or mismatched +reports. It never falls back to an earlier attempt's success. Failed collection +reads additionally produce synthetic target `0` in reporting sweeps (included in +result counts), which blocks projection even if an individual result is green. + +All target results and global health remain in structured logs. The final scope, +target kind, policy SHA and global summary are also written to the job summary; +an incomplete/failed sweep emits a warning even when the trigger PR job passes. +This isolates unrelated job conclusions without clearing any peer's failing +commit status or hiding repository-wide failures. No admission rules, publishing +permissions, status-history limits, or acceptance requirements are weakened. + The bootstrap PR cannot exercise a target-branch workflow that has not landed: independently review and accept it before merging, then validate report-only on real PRs. Local/PR CI success does not close #50 or authorize required enforcement. @@ -345,3 +401,4 @@ python tools/check_knowledge_model_policy.py --repository --pull-re publishing statuses. It does not show the main workflow is operational. Production uses runtime-injected `GITHUB_TOKEN`; never pass tokens as CLI arguments. The existing ready lane runs unit tests on Python 3.10 and 3.11 in its full suite. + diff --git a/tests/test_knowledge_model_policy.py b/tests/test_knowledge_model_policy.py index ca6014d4..98ac9fd8 100644 --- a/tests/test_knowledge_model_policy.py +++ b/tests/test_knowledge_model_policy.py @@ -329,9 +329,22 @@ def test_workflow_never_executes_pr_code_and_refreshes_metadata_and_base(): assert value["permissions"] == {"contents": "read", "pull-requests": "read", "issues": "read", "statuses": "write"} assert "concurrency" not in value - assert value["jobs"]["declarations"]["concurrency"] == { - "group": "knowledge-policy-status-writer", "cancel-in-progress": False, - } + job = value["jobs"]["declarations"] + eligibility = ( + "github.event_name != 'issue_comment' || !github.event.issue || " + "!github.event.comment || !github.event.action || " + "(github.event.issue.pull_request && (github.event.action != 'created' || " + "!github.event.comment.body || " + "contains(github.event.comment.body, 'knowledge-policy-approval')))" + ) + assert " ".join(job["if"].split()) == "${{ " + eligibility + " }}" + group = ( + "${{ (" + eligibility + ") && 'knowledge-policy-status-writer' || " + "format('knowledge-policy-skipped-{0}-{1}', github.run_id, github.run_attempt) }}" + ) + assert " ".join(job["concurrency"]["group"].split()) == group + assert job["concurrency"]["cancel-in-progress"] is False + assert set(job["concurrency"]) == {"group", "cancel-in-progress"} steps = value["jobs"]["declarations"]["steps"] assert steps[0]["with"] == {"ref": "main", "persist-credentials": False} assert all("pip install" not in step.get("run", "") for step in steps) @@ -1535,3 +1548,4 @@ def test_api_other_contexts_do_not_consume_the_budget(api_fixture, policy): seed_history(state, 400, context="other-ci") assert check_pr(api, 123, policy, BASE, False)["state"] == "success" assert latest_status(state)["state"] == "success" + diff --git a/tests/test_knowledge_policy_comment_routing.py b/tests/test_knowledge_policy_comment_routing.py new file mode 100644 index 00000000..42a042e7 --- /dev/null +++ b/tests/test_knowledge_policy_comment_routing.py @@ -0,0 +1,110 @@ +"""Offline routing contracts, not an execution of GitHub's expression engine. + +Model the documented truthiness, case-insensitive contains, and short-circuit +operators used by this exact expression. CI additionally parses the workflow YAML. +https://docs.github.com/en/actions/reference/workflows-and-actions/expressions +""" + +import re +import unittest +from pathlib import Path + +from tools import check_knowledge_model_policy as policy + +ROOT = Path(__file__).resolve().parents[1] +ELIGIBILITY = ( + "github.event_name != 'issue_comment' || !github.event.issue || " + "!github.event.comment || !github.event.action || " + "(github.event.issue.pull_request && (github.event.action != 'created' || " + "!github.event.comment.body || " + "contains(github.event.comment.body, 'knowledge-policy-approval')))" +) +WRITER = "knowledge-policy-status-writer" + + +def truthy(value): + # GitHub objects/arrays are truthy, including empty objects/arrays. + if isinstance(value, (dict, list)): + return True + return bool(value) + + +def eligible(event_name, event): + """Explicit model of ELIGIBILITY, with absent properties evaluating to null.""" + issue = event.get("issue") + comment = event.get("comment") + action = event.get("action") + if (event_name.casefold() != "issue_comment" or not truthy(issue) + or not truthy(comment) or not truthy(action)): + return True + body = comment.get("body") + return truthy(issue.get("pull_request")) and ( + action.casefold() != "created" or not truthy(body) + or "knowledge-policy-approval" in body.casefold() + ) + + +def group(event_name, event, run_id=1, run_attempt=1): + return WRITER if eligible(event_name, event) else ( + f"knowledge-policy-skipped-{run_id}-{run_attempt}" + ) + + +class CommentRoutingTests(unittest.TestCase): + def test_identical_eligibility_controls_if_and_group(self): + self.assertIn(f"contains(github.event.comment.body, '{policy.APPROVAL}')", ELIGIBILITY) + text = (ROOT / ".github/workflows/knowledge-model-policy.yml").read_text() + expressions = [" ".join(value.split()) for value in re.findall( + r"\$\{\{ (.*?) \}\}", text, re.S + )] + self.assertEqual(expressions[0], ELIGIBILITY) + self.assertEqual(expressions[1], ( + f"({ELIGIBILITY}) && '{WRITER}' || " + "format('knowledge-policy-skipped-{0}-{1}', github.run_id, github.run_attempt)" + )) + self.assertIn("cancel-in-progress: false", text) + self.assertIn("ref: main", text) + self.assertIn("persist-credentials: false", text) + + def test_truth_matrix(self): + for is_pr in (False, True): + issue = {"pull_request": {}} if is_pr else {"number": 1} + for action in ("created", "edited", "deleted"): + for body in (None, "", "LGTM", "```knowledge-policy-approval\n{}\n```", + "KNOWLEDGE-POLICY-APPROVAL"): + event = {"issue": issue, "comment": {"body": body}, "action": action} + expected = is_pr and ( + action != "created" or body in (None, "") + or "knowledge-policy-approval" in body.casefold() + ) + with self.subTest(is_pr=is_pr, action=action, body=body): + self.assertEqual(eligible("issue_comment", event), expected) + self.assertEqual(group("issue_comment", event) == WRITER, expected) + + def test_missing_payload_is_conservative(self): + complete = {"issue": {"pull_request": {}}, "comment": {"body": "LGTM"}, + "action": "created"} + for key in complete: + for missing in (None, ""): + event = {**complete, key: missing} + self.assertTrue(eligible("issue_comment", event)) + event = {key_: value for key_, value in complete.items() if key_ != key} + self.assertTrue(eligible("issue_comment", event)) + self.assertTrue(eligible("issue_comment", {})) + self.assertTrue(eligible("issue_comment", {**complete, "comment": {}})) + + def test_other_triggers_keep_shared_writer(self): + for event_name in ("pull_request_target", "push", "workflow_dispatch"): + self.assertEqual(group(event_name, {}), WRITER) + + def test_irrelevant_comment_never_enters_writer_group(self): + event = {"issue": {"pull_request": {}}, "comment": {"body": "Thanks!"}, + "action": "created"} + groups = {group("issue_comment", event, run, attempt) + for run in (1, 2) for attempt in (1, 2)} + self.assertEqual(len(groups), 4) + self.assertNotIn(WRITER, groups) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_knowledge_policy_http.py b/tests/test_knowledge_policy_http.py new file mode 100644 index 00000000..b9c6bd74 --- /dev/null +++ b/tests/test_knowledge_policy_http.py @@ -0,0 +1,141 @@ +"""Offline HTTP diagnostics; all transport calls are replaced by fixtures.""" + +import io +import json +import unittest +from unittest.mock import patch +from urllib.error import HTTPError + +from tools import check_knowledge_model_policy as policy + + +def failure(code, message="", headers=None): + return HTTPError("https://example.invalid/private?token=FAKE_SECRET", code, + "FAKE_SECRET", headers or {}, + io.BytesIO(json.dumps({"message": message, + "errors": ["FAKE_SECRET"]}).encode())) + + +class DiagnosticsTests(unittest.TestCase): + def test_permission_is_not_rate_limit(self): + error = policy.GitHubHTTPError(failure(403, "Resource not accessible by integration")) + self.assertEqual(error.category, "permission") + self.assertIsNone(error.delay) + + def test_unknown_403_and_validation_do_not_infer_cause(self): + self.assertEqual(policy.GitHubHTTPError(failure(403)).category, "unknown") + error = policy.GitHubHTTPError(failure(422, "Validation Failed")) + self.assertEqual(error.category, "validation") + self.assertNotIn("capacity", str(error)) + + def test_allowlist_does_not_echo_secrets(self): + error = policy.GitHubHTTPError(failure(403, "FAKE_SECRET", { + "Authorization": "FAKE_SECRET", "Retry-After": "FAKE_SECRET", + "X-GitHub-Request-Id": "FAKE_SECRET", "X-RateLimit-Remaining": "0"})) + self.assertNotIn("FAKE_SECRET", policy.evaluation_failure(error)["errors"][0]) + self.assertNotIn("example.invalid", str(error)) + self.assertIn('"x-ratelimit-remaining": 0', str(error)) + + def test_explicit_status_capacity_reason_is_recognized(self): + body = {"message": "Validation Failed", "errors": [{"resource": "Status", + "code": "custom", "message": + "This SHA and context has reached the maximum number of statuses."}]} + original = HTTPError("https://example.invalid", 422, "", {}, + io.BytesIO(json.dumps(body).encode())) + error = policy.GitHubHTTPError(original) + self.assertEqual(error.category, "validation") + self.assertIn('"reason": "status-capacity"', str(error)) + self.assertNotIn("This SHA", str(error)) + + def test_unknown_validation_is_redacted_and_not_inferred_as_capacity(self): + capacity = "This SHA and context has reached the maximum number of statuses." + for item in ({"resource": "Status", "code": "custom", "message": "FAKE_SECRET"}, + {"resource": "Status", "code": "custom", + "message": capacity + " FAKE_SECRET"}, + {"resource": "Other", "code": "custom", "message": capacity}, + "FAKE_SECRET", None): + with self.subTest(item=item): + body = {"message": "FAKE_SECRET", "errors": [item]} + original = HTTPError("https://example.invalid", 422, "", {}, + io.BytesIO(json.dumps(body).encode())) + error = policy.GitHubHTTPError(original) + self.assertIn('"reason": "unknown"', str(error)) + self.assertNotIn("status-capacity", str(error)) + self.assertNotIn("FAKE_SECRET", str(error)) + + def test_body_read_is_bounded_and_invalid_json_is_safe(self): + body = io.BytesIO(b"x" * 20000) + error = HTTPError("https://example.invalid", 403, "", {}, body) + with patch.object(body, "read", wraps=body.read) as read: + self.assertEqual(policy.GitHubHTTPError(error).category, "unknown") + read.assert_called_once_with(8192) + + def test_one_header_evidenced_get_retry(self): + response = io.BytesIO(b"{}") + response.headers = {} + api = policy.GitHub("example/repository", None) + with patch.object(policy, "urlopen", side_effect=[ + failure(403, headers={"Retry-After": "1"}), response]) as transport, \ + patch.object(policy.time, "sleep") as sleep: + self.assertEqual(api.get("pulls"), {}) + self.assertEqual(transport.call_count, 2) + sleep.assert_called_once_with(1) + + def test_retry_timeout_shrinks_after_wait_and_request_time(self): + response = io.BytesIO(b"{}") + response.headers = {} + with patch.object(policy, "urlopen", side_effect=[ + failure(429, headers={"Retry-After": "2"}), response]) as transport, \ + patch.object(policy.time, "monotonic", side_effect=[100, 100, 103, 105]), \ + patch.object(policy.time, "sleep") as sleep: + self.assertEqual(policy.GitHub("example/repository", None).get("pulls"), {}) + self.assertEqual([call.kwargs["timeout"] for call in transport.call_args_list], + [30, 25]) + sleep.assert_called_once_with(2) + + def test_reset_header_fallback_request_id_and_response_closed(self): + original = failure(403, headers={"X-RateLimit-Remaining": "0", + "X-RateLimit-Reset": "105", + "X-GitHub-Request-Id": "ABCD:1234:EF56"}) + with patch.object(policy.time, "time", return_value=100), \ + patch.object(original, "close", wraps=original.close) as close: + error = policy.GitHubHTTPError(original) + self.assertEqual(error.category, "rate-limit") + self.assertEqual(error.delay, 6) + self.assertIn('"request_id": "ABCD:1234:EF56"', str(error)) + close.assert_called_once_with() + + def test_no_guessed_or_unbounded_wait(self): + for code, headers, timeout in [(403, {}, 30), (422, {}, 30), + (429, {"Retry-After": "120"}, 30), + (403, {"Retry-After": "5"}, 3)]: + with self.subTest(code=code, headers=headers), \ + patch.object(policy, "urlopen", side_effect=failure(code, headers=headers)) \ + as transport, patch.object(policy.time, "sleep") as sleep: + with self.assertRaises(policy.GitHubHTTPError): + policy.GitHub("example/repository", None).request("pulls", timeout=timeout) + self.assertEqual(transport.call_count, 1) + sleep.assert_not_called() + + def test_never_replay_status_post(self): + with patch.object(policy, "urlopen", side_effect=failure( + 429, headers={"Retry-After": "1"})) as transport, \ + patch.object(policy.time, "sleep") as sleep: + with self.assertRaises(policy.GitHubHTTPError): + policy.GitHub("example/repository", None).status("a" * 40, "failure", "test") + self.assertEqual(transport.call_count, 1) + sleep.assert_not_called() + + def test_retry_is_at_most_once(self): + with patch.object(policy, "urlopen", side_effect=[ + failure(429, headers={"Retry-After": "1"}), + failure(429, headers={"Retry-After": "1"})]) as transport, \ + patch.object(policy.time, "sleep"): + with self.assertRaises(policy.GitHubHTTPError): + policy.GitHub("example/repository", None).get("pulls") + self.assertEqual(transport.call_count, 2) + + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_knowledge_policy_reporting.py b/tests/test_knowledge_policy_reporting.py new file mode 100644 index 00000000..9d5716da --- /dev/null +++ b/tests/test_knowledge_policy_reporting.py @@ -0,0 +1,269 @@ +"""Offline final-attempt report and scope checks (stdlib only).""" +import contextlib +import io +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import Mock, patch + +from tools import check_knowledge_model_policy as policy + +SHA = "a" * 40 + + +def result(number=1, kind="compliant"): + return {"pull_request": number, "head_sha": "b" * 40, "policy_sha": SHA, + "dry_run": False, "rerun": kind in policy.RERUN_KINDS, + "state": "success" if kind in {"compliant", "closed"} else "failure", + "kind": kind, "description": "fixture", "errors": []} + + +class ReportingTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.path = Path(self.temp.name) / "report.json" + self.event = Path(self.temp.name) / "event.json" + self.event.write_text(json.dumps({"number": 1, "pull_request": {"number": 1, + "head": {"sha": "b" * 40}}})) + self.binding = policy.report_binding("owner/repo", "pull_request_target", + self.event, SHA, "attempt-3") + + def project(self, results, binding=None): + binding = binding or self.binding + policy.write_report(self.path, binding, results) + with contextlib.redirect_stdout(io.StringIO()): + return policy.report_exit(self.path, binding, policy.exit_code(results)) + + def test_own_success_unrelated_failure(self): + self.assertEqual(self.project([result(), result(2, "history-unavailable")]), 0) + + def test_each_own_failure_is_red(self): + for kind in policy.KINDS: + if kind not in {"compliant", "closed"}: + with self.subTest(kind=kind): + self.assertEqual(self.project([result(1, kind), result(2)]), 1) + + def test_closed_is_explicit_na(self): + self.assertEqual(self.project([result(1, "closed")]), 0) + + def test_missing_own_and_global_listing_failure_fail(self): + self.assertEqual(self.project([result(2)]), 1) + self.assertEqual(self.project([result(), result(0, "unverified")]), 1) + + def test_push_and_dispatch_keep_global_health(self): + for event in ("push", "workflow_dispatch"): + binding = dict(self.binding, event_name=event, trigger=None) + self.assertEqual(self.project([result(), result(2, "history-unavailable")], binding), 1) + self.assertEqual(self.project([result()], binding), 0) + + def test_unknown_dry_run_and_bad_success_fail_closed(self): + for key, value in (("dry_run", True), ("kind", "new-kind"), ("head_sha", "bad")): + row = result() + row[key] = value + with self.subTest(key=key), self.assertRaises(ValueError): + self.project([row]) + row = result() + row["errors"] = ["not verified"] + self.assertEqual(self.project([row]), 1) + row = result() + row["state"] = "failure" + self.assertEqual(self.project([row]), 1) + + def test_duplicate_targets_fail(self): + with self.assertRaises(ValueError): + self.project([result(), result()]) + + def test_stale_attempt_policy_event_or_repo_fail(self): + policy.write_report(self.path, self.binding, [result()]) + for key in ("report_id", "policy_sha", "event_digest", "repository", "trigger"): + binding = dict(self.binding, **{key: "different"}) + with self.subTest(key=key), self.assertRaises(ValueError): + policy.report_exit(self.path, binding, 0) + + def test_missing_or_invalid_report_fails(self): + with self.assertRaises(OSError): + policy.report_exit(self.path, self.binding, 0) + self.path.write_text("not json") + with self.assertRaises(ValueError): + policy.report_exit(self.path, self.binding, 0) + + def test_summary_and_exit_tampering_fail(self): + policy.write_report(self.path, self.binding, [result()]) + with self.assertRaises(ValueError): + policy.report_exit(self.path, self.binding, 2) + report = json.loads(self.path.read_text()) + report["summary"]["results"] = 9 + self.path.write_text(json.dumps(report)) + with self.assertRaises(ValueError): + policy.report_exit(self.path, self.binding, 0) + + def test_last_attempt_replaces_earlier_success(self): + self.assertEqual(self.project([result()]), 0) + self.assertEqual(self.project([result(1, "unverified")]), 1) + self.assertFalse(self.path.with_name("report.json.tmp").exists()) + + def test_issue_comment_scope_and_event_digest(self): + self.event.write_text(json.dumps({"issue": {"number": 7, "pull_request": {}}})) + binding = policy.report_binding("owner/repo", "issue_comment", self.event, SHA, "3") + self.assertEqual(binding["trigger"], 7) + self.assertNotEqual(binding["event_digest"], self.binding["event_digest"]) + + def test_inconsistent_or_boolean_pr_number_rejected(self): + for event in ({"number": 1, "pull_request": {"number": 2}}, + {"number": True, "pull_request": {"number": True}}): + self.event.write_text(json.dumps(event)) + with self.assertRaises(ValueError): + policy.report_binding("owner/repo", "pull_request_target", self.event, SHA, "3") + + def test_closed_trigger_is_added_to_global_sweep(self): + api = Mock() + api.pages.return_value = [] + api.get.return_value = {"number": 1, "head": {"sha": "b" * 40}, + "base": {"ref": "main"}, "state": "closed"} + results = policy.check_prs(api, {"base_branch": "main"}, SHA, False, + trigger=1, report_scope=True) + self.assertEqual([(r["pull_request"], r["kind"]) for r in results], [(1, "closed")]) + self.assertEqual(api.get.call_count, 2) + api.status.assert_not_called() + self.assertEqual(self.project(results), 0) + + def test_final_collection_failure_blocks_even_closed_target(self): + api = Mock() + api.pages.side_effect = [[], OSError("fixture collection failure")] + api.get.return_value = {"number": 1, "head": {"sha": "b" * 40}, + "base": {"ref": "main"}, "state": "closed"} + results = policy.check_prs(api, {"base_branch": "main"}, SHA, False, + trigger=1, report_scope=True) + self.assertIn(0, [r["pull_request"] for r in results]) + self.assertEqual(self.project(results), 1) + api.status.assert_not_called() + + def test_trigger_is_first_without_dropping_other_targets(self): + api = Mock() + prs = {number: {"number": number, "head": {"sha": str(number) * 40}, + "base": {"ref": "main"}, "state": "open"} for number in (1, 2)} + api.pages.return_value = list(prs.values()) + api.get.side_effect = lambda path: prs[int(path.split("/")[-1])] + failure = {"state": "failure", "kind": "budget-saturated", + "description": "fixture", "errors": []} + with patch.object(policy, "invalidated", return_value=(failure, False)): + results = policy.check_prs(api, {"base_branch": "main"}, SHA, False, + trigger=2, report_scope=True) + self.assertEqual(api.get.call_args_list[0].args, ("pulls/2",)) + self.assertEqual({r["pull_request"] for r in results}, {1, 2}) + + def test_boolean_schema_or_exit_is_not_valid_integer(self): + for field, value in (("schema_version", True), ("sweep_exit", False)): + policy.write_report(self.path, self.binding, [result()]) + report = json.loads(self.path.read_text()) + report[field] = value + self.path.write_text(json.dumps(report)) + with self.subTest(field=field), self.assertRaises(ValueError): + policy.report_exit(self.path, self.binding, 0) + + def test_pull_request_event_head_must_match_evaluated_head(self): + self.assertEqual(self.project([result()]), 0) + advanced = result() + advanced["head_sha"] = "c" * 40 + policy.write_report(self.path, self.binding, [advanced]) + output = io.StringIO() + with contextlib.redirect_stdout(output): + self.assertEqual(policy.report_exit(self.path, self.binding, 0), 1) + summary = json.loads(output.getvalue()) + self.assertEqual(summary["projection_reason"], "stale-event-head") + self.assertEqual(summary["expected_head"], "b" * 40) + self.assertEqual(summary["evaluated_head"], "c" * 40) + + def test_missing_or_malformed_event_head_fails_closed(self): + for head in (None, {}, {"sha": "bad"}, {"sha": True}, {"sha": "b" * 39}): + event = {"number": 1, "pull_request": {"number": 1, "head": head}} + self.event.write_text(json.dumps(event)) + with self.subTest(head=head), self.assertRaises(ValueError): + policy.report_binding("owner/repo", "pull_request_target", self.event, SHA, "3") + + def test_issue_comment_still_uses_current_head(self): + self.event.write_text(json.dumps({"issue": {"number": 1, "pull_request": {}}})) + binding = policy.report_binding("owner/repo", "issue_comment", self.event, SHA, "3") + advanced = result() + advanced["head_sha"] = "c" * 40 + self.assertEqual(self.project([advanced], binding), 0) + + +class RunnerTests(unittest.TestCase): + def run_runner(self, mode, event_name="pull_request_target"): + root = Path(__file__).resolve().parents[1] + with tempfile.TemporaryDirectory() as directory: + temp = Path(directory) + event = temp / "event.json" + event.write_text(json.dumps({"number": 1, "pull_request": {"number": 1, + "head": {"sha": "b" * 40}}})) + git = temp / "git" + git.write_text("#!/bin/sh\nif [ \"$1\" = rev-parse ]; then printf '%040d\\n' 0; fi\n") + python = temp / "python" + python.write_text("#!" + sys.executable + "\n" + r'''import json, os, runpy, sys +from pathlib import Path +from unittest.mock import Mock, patch +sys.path.insert(0, os.environ["FIXTURE_ROOT"]) +from tools import check_knowledge_model_policy as policy +args = sys.argv[2:] +def arg(name): return args[args.index(name) + 1] +if "--report-exit" in args: + sys.argv = [sys.argv[1]] + args + runpy.run_path(str(Path(os.environ["FIXTURE_ROOT"]) / "tools/check_knowledge_model_policy.py"), + run_name="__main__") +counter = Path(os.environ["RUNNER_TEMP"]) / "count" +count = int(counter.read_text()) + 1 if counter.exists() else 1 +counter.write_text(str(count)) +mode = os.environ["FIXTURE_MODE"] +if mode == "missing-final" and count == 3: + sys.exit(2) +kind = "history-unavailable" if mode == "own-final-failure" and count == 3 else "compliant" +def row(number, kind): + return {"pull_request": number, "head_sha": "b" * 40, "policy_sha": arg("--policy-sha"), + "dry_run": False, "rerun": kind in policy.RERUN_KINDS, + "state": "success" if kind == "compliant" else "failure", "kind": kind, + "description": "fixture", "errors": []} +binding = policy.report_binding(os.environ["GITHUB_REPOSITORY"], os.environ["GITHUB_EVENT_NAME"], + os.environ["GITHUB_EVENT_PATH"], arg("--policy-sha"), + arg("--report-id")) +policy.write_report(Path(arg("--report")), binding, [row(1, kind), row(2, "history-unavailable")]) +sys.exit(2) +''') + git.chmod(0o700) + python.chmod(0o700) + env = {"PATH": str(temp) + ":/usr/bin:/bin", "HOME": directory, + "RUNNER_TEMP": directory, "GITHUB_RUN_ID": "fixture", + "GITHUB_EVENT_NAME": event_name, "GITHUB_EVENT_PATH": str(event), + "GITHUB_REPOSITORY": "owner/repo", "FIXTURE_ROOT": str(root), + "FIXTURE_MODE": mode, "PYTHONDONTWRITEBYTECODE": "1"} + completed = subprocess.run( + ["/bin/bash", str(root / "tools/run_knowledge_model_policy.sh")], + cwd=root, env=env, capture_output=True, text=True, timeout=10) + self.assertEqual((temp / "count").read_text(), "3", completed.stderr) + self.assertFalse(list(temp.glob("knowledge-model-policy-*.json"))) + return completed + + def test_no_early_exit_and_other_pr_failure_only_warns(self): + outcome = self.run_runner("own-success") + self.assertEqual(outcome.returncode, 0, outcome.stderr) + self.assertIn("::warning::", outcome.stdout) + + def test_final_own_failure_overrides_previous_green(self): + outcome = self.run_runner("own-final-failure") + self.assertEqual(outcome.returncode, 1, outcome.stderr) + + def test_missing_final_report_cannot_reuse_previous_green(self): + outcome = self.run_runner("missing-final") + self.assertEqual(outcome.returncode, 1, outcome.stderr) + + def test_main_push_stays_global_failure(self): + outcome = self.run_runner("own-success", "push") + self.assertEqual(outcome.returncode, 1, outcome.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/check_knowledge_model_policy.py b/tools/check_knowledge_model_policy.py index f3470ec4..6b82b57a 100644 --- a/tools/check_knowledge_model_policy.py +++ b/tools/check_knowledge_model_policy.py @@ -1,4 +1,5 @@ import argparse +import hashlib import json import os import re @@ -285,6 +286,64 @@ def assess(pr, paths, policy, comments=(), contributors=()): } +class GitHubHTTPError(OSError): + """An allowlisted diagnostic, never a raw response body or credential-bearing URL.""" + + def __init__(self, error): + self.code = error.code + headers = error.headers or {} + self.delay = None + retry_after = headers.get("Retry-After", "") + remaining = headers.get("X-RateLimit-Remaining", "") + reset = headers.get("X-RateLimit-Reset", "") + if re.fullmatch(r"[0-9]{1,10}", retry_after): + self.delay = max(1, int(retry_after)) + elif remaining == "0" and re.fullmatch(r"[0-9]{1,10}", reset): + self.delay = max(1, int(reset) - int(time.time()) + 1) + try: + body = json.loads(error.read(8192)) + except (ValueError, OSError, TypeError, AttributeError): + body = {} + error.close() + message = body.get("message", "") if isinstance(body, dict) else "" + message = message.lower() if isinstance(message, str) else "" + rate = self.code == 429 or (self.code == 403 and ( + remaining == "0" or self.delay is not None or "rate limit" in message)) + if rate: + category = "rate-limit" + elif self.code == 401 or (self.code == 403 and any(text in message for text in ( + "resource not accessible", "insufficient permission", "requires authentication"))): + category = "permission" + elif self.code == 422: + category = "validation" + else: + category = "unknown" + self.category = category + detail = {"http_status": self.code, "category": category} + if self.code == 422: + # Recognize only the explicit status-capacity response, never infer it from + # HTTP 422, an arbitrary validation message, or the locally counted history. + capacity_message = "this sha and context has reached the maximum number of statuses." + errors = body.get("errors", []) if isinstance(body, dict) else [] + capacity = isinstance(errors, list) and any( + isinstance(item, dict) and item.get("resource") == "Status" + and item.get("code") == "custom" + and isinstance(item.get("message"), str) + and item["message"].strip().lower() == capacity_message + for item in errors) + detail["reason"] = "status-capacity" if capacity else "unknown" + # Only numeric rate metadata and a bounded request ID are safe to retain. Response + # messages/errors may echo user content, tokens or URLs, so they are never logged. + for name in ("X-RateLimit-Remaining", "X-RateLimit-Reset", "Retry-After"): + value = headers.get(name, "") + if re.fullmatch(r"[0-9]{1,10}", value): + detail[name.lower()] = int(value) + request_id = headers.get("X-GitHub-Request-Id", "") + if re.fullmatch(r"[A-Fa-f0-9:]{1,80}", request_id): + detail["request_id"] = request_id + super().__init__("GitHub API " + json.dumps(detail, sort_keys=True)) + + class GitHub: def __init__(self, repository, token): if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repository): @@ -301,8 +360,21 @@ def request(self, route, payload=None, timeout=30): if data is not None: headers["Content-Type"] = "application/json" request = Request(self.root + route, data=data, headers=headers) - with urlopen(request, timeout=timeout) as response: - return json.load(response), response.headers.get("Link", "") + deadline = time.monotonic() + timeout + for attempt in range(2): + try: + with urlopen(request, timeout=max(0.001, deadline - time.monotonic())) as response: + return json.load(response), response.headers.get("Link", "") + except HTTPError as error: + failure = GitHubHTTPError(error) + # Retry reads only, once, with explicit rate timing and within the existing + # request budget. Never replay a status POST or guess a delay from HTTP 403. + if (attempt == 0 and payload is None and failure.category == "rate-limit" + and failure.delay is not None and failure.delay <= 60 + and failure.delay + 1 < deadline - time.monotonic()): + time.sleep(failure.delay) + continue + raise failure from None def get(self, route): return self.request(route)[0] @@ -327,7 +399,8 @@ def status(self, head, state, description): payload = {"state": state, "context": CONTEXT, "description": description} run_id = os.environ.get("GITHUB_RUN_ID") if run_id: - repository_url = self.root.replace("https://api.github.com/repos/", "https://github.com/") + repository_url = self.root.replace( + "https://api.github.com/repos/", "https://github.com/") payload["target_url"] = f"{repository_url}actions/runs/{run_id}" self.request(f"statuses/{head}", payload) @@ -372,7 +445,8 @@ def deferred(reason): def evaluation_failure(error): - message = (f"GitHub API HTTP {error.code}" if isinstance(error, HTTPError) + message = (str(error) if isinstance(error, GitHubHTTPError) + else f"GitHub API HTTP {error.code}" if isinstance(error, HTTPError) else "GitHub API unavailable" if isinstance(error, OSError) else str(error) if isinstance(error, ValueError) else "Invalid GitHub API response") @@ -645,7 +719,8 @@ def still_current(api, number, entry, policy, open_prs, scan_error): return True -def check_prs(api, policy, policy_sha, dry_run, numbers=None, plan=None): +def check_prs(api, policy, policy_sha, dry_run, numbers=None, plan=None, trigger=None, + report_scope=False): # The sweep deadline bounds the whole evaluation pass and keeps publication time free; each # target's read is bounded separately by its share of what remains. deadline = time.monotonic() + SWEEP_SECONDS @@ -662,6 +737,9 @@ def check_prs(api, policy, policy_sha, dry_run, numbers=None, plan=None): "kind": "unverified", "description": "Lifecycle metadata unavailable", "errors": [evaluation_failure(scan_error)["errors"][0]]}] numbers = [pr["number"] for pr in open_prs if addressable(pr, policy)] + # A close/base-change event must receive a fresh explicit N/A result as well. + if trigger is not None: + numbers = [trigger, *(number for number in numbers if number != trigger)] # Work another attempt already settled is not repeated: re-running it would spend the same # bounded time and, in a later attempt's exhausted tail, replace a published verdict with a # failure. A carried target is trusted only after its own premise is re-confirmed here. @@ -850,6 +928,11 @@ def check_prs(api, policy, policy_sha, dry_run, numbers=None, plan=None): "rerun": item["rerun"], "state": item["state"], "kind": item["kind"], "description": item["description"], "errors": item["errors"]}) + if report_scope and scan_error is not None: + results.append({"pull_request": 0, "head_sha": "", "policy_sha": policy_sha, + "dry_run": dry_run, "rerun": True, "state": "failure", + "kind": "unverified", "description": "Lifecycle metadata unavailable", + "errors": [evaluation_failure(scan_error)["errors"][0]]}) return results @@ -865,6 +948,108 @@ def refresh_event(event, event_name): raise ValueError("Unsupported policy event; merge queue requires an explicit integration") + +def report_binding(repository, event_name, event_path, policy_sha, report_id): + """Bind a report to trusted runner inputs, not declaration text or an older attempt.""" + if not report_id or not repository: + raise ValueError("Missing report binding") + raw = Path(event_path).read_bytes() + event = json.loads(raw, object_pairs_hook=unique_object) + refresh_event(event, event_name) + trigger = None + expected_head = None + if event_name == "pull_request_target": + trigger = event["number"] + if event["pull_request"]["number"] != trigger: + raise ValueError("Inconsistent trigger") + elif event_name == "issue_comment" and "pull_request" in event["issue"]: + trigger = event["issue"]["number"] + if trigger is not None and (type(trigger) is not int or trigger <= 0): + raise ValueError("Invalid trigger") + if event_name == "pull_request_target": + head = event["pull_request"].get("head") + expected_head = head.get("sha") if isinstance(head, dict) else None + if not isinstance(expected_head, str) or not SHA.fullmatch(expected_head): + raise ValueError("Missing or malformed trigger head") + return {"repository": repository, "event_name": event_name, + "event_digest": hashlib.sha256(raw).hexdigest(), "trigger": trigger, + "policy_sha": policy_sha, "report_id": report_id, "expected_head": expected_head} + + +def write_report(path, binding, results): + if path is None: + return + report = {"schema_version": 1, "binding": binding, "results": results, + "sweep_exit": exit_code(results), "summary": summarize(results)} + temporary = path.with_name(path.name + ".tmp") + temporary.write_text(json.dumps(report, ensure_ascii=True, sort_keys=True), encoding="utf-8") + temporary.replace(path) + + +def report_exit(path, binding, sweep_exit): + """Project ONLY the final completed attempt; global health remains visible.""" + report = json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=unique_object) + if (type(report.get("schema_version")) is not int or report["schema_version"] != 1 + or report.get("binding") != binding): + raise ValueError("Stale or mismatched admission report") + results = report["results"] + if not isinstance(results, list): + raise ValueError("Invalid admission results") + seen = set() + for result in results: + number = result["pull_request"] + if type(number) is not int or number < 0 or number in seen: + raise ValueError("Invalid or duplicate admission target") + seen.add(number) + if (result["policy_sha"] != binding["policy_sha"] + or result["dry_run"] is not False + or type(result["rerun"]) is not bool + or result["kind"] not in KINDS + or result["state"] not in {"success", "failure"} + or not isinstance(result["errors"], list) + or not isinstance(result["description"], str) + or not isinstance(result["head_sha"], str) + or (result["head_sha"] and not SHA.fullmatch(result["head_sha"]))): + raise ValueError("Invalid admission result") + if (type(sweep_exit) is not int or sweep_exit not in {0, 1, 2} + or type(report["sweep_exit"]) is not int + or report["sweep_exit"] != sweep_exit or exit_code(results) != sweep_exit + or report["summary"] != summarize(results)): + raise ValueError("Inconsistent admission report") + trigger = binding["trigger"] + own = [result for result in results if result["pull_request"] == trigger] + stale_event = bool(own and binding["event_name"] == "pull_request_target" + and own[0]["head_sha"] != binding["expected_head"]) + if trigger is None: + conclusion = int(sweep_exit != 0) + else: + conclusion = 1 + if 0 not in seen and len(own) == 1 and not stale_event: + result = own[0] + if (result["kind"] in {"compliant", "closed"} + and result["state"] == "success" and not result["rerun"] + and not result["errors"] and SHA.fullmatch(result["head_sha"])): + conclusion = 0 + summary = {"scope": "trigger-pr" if trigger is not None else "global-sweep", + "trigger": trigger, "job_exit": conclusion, + "target_kind": own[0]["kind"] if own else None, + "projection_reason": "stale-event-head" if stale_event else "final-target-result", + "expected_head": binding["expected_head"], + "evaluated_head": own[0]["head_sha"] if own else None, + "target_applicability": ("not-applicable" if own and own[0]["kind"] == "closed" + else "see-target-result"), + "global_summary": report["summary"], "policy_sha": binding["policy_sha"]} + print(json.dumps(summary, ensure_ascii=True)) + if sweep_exit: + print("::warning::Global admission sweep remains incomplete or failed; " + "see structured results. Trigger-PR conclusion does not clear other PRs.") + summary_path = os.environ.get("GITHUB_STEP_SUMMARY") + if summary_path: + with Path(summary_path).open("a", encoding="utf-8") as stream: + stream.write("### Admission result (final attempt)\n\n```json\n" + + json.dumps(summary, indent=2, ensure_ascii=True) + "\n```\n") + return conclusion + def main(argv=None): parser = argparse.ArgumentParser(description="Check declared knowledge authorship") parser.add_argument("--repository", default=os.environ.get("GITHUB_REPOSITORY"), required=False) @@ -874,9 +1059,22 @@ def main(argv=None): parser.add_argument("--pull-request", type=int) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--plan", type=Path) + parser.add_argument("--report", type=Path) + parser.add_argument("--report-id") + parser.add_argument("--report-exit", type=int) args = parser.parse_args(argv) if not SHA.fullmatch(args.policy_sha): raise ValueError("Expected the trusted policy checkout SHA") + binding = None + if args.report is not None: + binding = report_binding(args.repository, os.environ["GITHUB_EVENT_NAME"], + args.event_path, args.policy_sha, args.report_id) + if args.pull_request is not None or args.dry_run: + raise ValueError("Scoped reporting requires the full publishing sweep") + if args.report_exit is not None: + if binding is None: + raise ValueError("Missing final report") + return report_exit(args.report, binding, args.report_exit) token = os.environ.get("GITHUB_TOKEN") if not args.dry_run and not token: raise ValueError("GITHUB_TOKEN is required to publish policy statuses") @@ -893,11 +1091,14 @@ def main(argv=None): # Every trigger shares one writer, so a superseded queued event still covers all open PRs. numbers = None if refresh_event(event, os.environ["GITHUB_EVENT_NAME"]) else [] plan = plan_file(args.plan, args.policy_sha) - results = check_prs(api, policy, args.policy_sha, args.dry_run, numbers, plan) + results = check_prs(api, policy, args.policy_sha, args.dry_run, numbers, plan, + trigger=binding["trigger"] if binding else None, + report_scope=binding is not None) write_plan(args.plan, plan) for result in results: print(json.dumps(result, ensure_ascii=True)) print(json.dumps(summarize(results), ensure_ascii=True)) + write_report(args.report, binding, results) return exit_code(results) @@ -910,3 +1111,4 @@ def main(argv=None): else "Invalid policy or event input") print(message, file=sys.stderr) sys.exit(1) + diff --git a/tools/run_knowledge_model_policy.sh b/tools/run_knowledge_model_policy.sh index 9f7f97fe..feca6640 100644 --- a/tools/run_knowledge_model_policy.sh +++ b/tools/run_knowledge_model_policy.sh @@ -8,20 +8,47 @@ set -uo pipefail attempts=3 +scoped=false +case "${GITHUB_EVENT_NAME:-}" in + pull_request_target|issue_comment|push|workflow_dispatch) scoped=true ;; +esac +# Keep explicit local/dry-run operation separate from workflow reporting. +for argument in "$@"; do + case "$argument" in + --pull-request|--pull-request=*|--dry-run) scoped=false ;; + esac +done plan="${RUNNER_TEMP:-${TMPDIR:-/tmp}}/knowledge-model-policy-plan-${GITHUB_RUN_ID:-local}.json" -rm -f "$plan" +report="${plan%.json}-report.json" +rm -f "$plan" "$report" +trap 'rm -f "$plan" "$report" "$report.tmp"' EXIT for attempt in $(seq 1 "$attempts"); do git fetch --quiet origin main || exit 1 git checkout --quiet origin/main || exit 1 + policy_sha="$(git rev-parse HEAD)" + report_id="${GITHUB_RUN_ID:-local}:${GITHUB_RUN_ATTEMPT:-1}:$$:${attempt}" + rm -f "$report" "$report.tmp" + report_args=() + if "$scoped"; then + report_args=(--report "$report" --report-id "$report_id") + fi status=0 - python tools/check_knowledge_model_policy.py --policy-sha "$(git rev-parse HEAD)" \ - --plan "$plan" "$@" || status=$? + python tools/check_knowledge_model_policy.py --policy-sha "$policy_sha" \ + --plan "$plan" "${report_args[@]}" "$@" || status=$? if [ "$status" -ne 2 ]; then - rm -f "$plan" - exit "$status" + break fi echo "Admission work incomplete; retrying against current main (${attempt}/${attempts})" >&2 done -rm -f "$plan" -echo "Admission work still incomplete after ${attempts} attempts" >&2 -exit 1 +if [ "$status" -eq 2 ]; then + echo "Admission work still incomplete after ${attempts} attempts" >&2 +fi +if ! "$scoped"; then + if [ "$status" -eq 2 ]; then exit 1; fi + exit "$status" +fi +# Only after the global sweep has finished/retried may this job project its own verdict. +# The report must match this exact final attempt and checkout; absent/invalid evidence fails. +python tools/check_knowledge_model_policy.py --policy-sha "$policy_sha" \ + --report "$report" --report-id "$report_id" --report-exit "$status" "$@" +