diff --git a/proof/region/v1/corpus_dispatch.py b/proof/region/v1/corpus_dispatch.py index 4928baaf..ad7e2332 100644 --- a/proof/region/v1/corpus_dispatch.py +++ b/proof/region/v1/corpus_dispatch.py @@ -11,6 +11,11 @@ commands is the default and dispatching is the opt-in `--execute`, which has to name the campaign's size: the incident this coordinator exists to prevent was a forgotten flag turning one mistake into 133 runs. + +The verification lanes replay evidence produced by another run, so that run +is admitted before the first dispatch: which workflow produced it, what +triggered it, whether it succeeded, and which commit it stands on — the last +of these cannot be decided here and is reported to the operator instead. """ from __future__ import annotations @@ -19,6 +24,7 @@ import json import subprocess import sys +from dataclasses import dataclass from pathlib import Path PROOF = Path(__file__).resolve().parent @@ -38,6 +44,21 @@ "verification-evidence-arb", "verification-evidence-mpfi", ) +# An artifact name proves nothing about where the artifact came from. Only a +# successful operator-triggered run of the producer workflow may be replayed: +# the path pins which workflow built the bundle, and the trigger pins whose +# code it was — a fork's pull request can run the same workflow and publish an +# artifact of exactly the allowlisted name. +EVIDENCE_WORKFLOW_PATH_V1 = ".github/workflows/full-domain-run.yml" +EVIDENCE_RUN_EVENT_V1 = "workflow_dispatch" +EVIDENCE_RUN_STATUS_V1 = "completed" +EVIDENCE_RUN_CONCLUSION_V1 = "success" +# A projection, never the whole run object: the reply carries fields this +# module has no reason to read, and printing a reply of unknown shape is how +# this project has leaked before. +RUN_PROVENANCE_JQ_V1 = "[.path, .event, .status, .conclusion, .head_sha] | @tsv" +COMMIT_SHA_LENGTH_V1 = 40 +COMMIT_SHA_ALPHABET_V1 = frozenset("0123456789abcdef") DEFAULT_LANE_WIDTH = 1 << 16 DEFAULT_SHARD_WIDTH = corpus_lane.DEFAULT_SHARD_POINTS FULL_DOMAIN = protocol.OUTPUT_CARDINALITY_V1 @@ -253,6 +274,175 @@ def parse_artifact_listing_v1(stdout: str) -> tuple[tuple[str, bool], ...]: return tuple(observed) +@dataclass(frozen=True) +class RunProvenanceV1: + """The run fields the admission reads, and nothing else. + + Where the evidence came from is four coordinates — which workflow built + it, what triggered that workflow, whether it finished successfully, and + which commit it stands on. They travel as one record so no caller can + check two of them and forget the rest. + """ + + path: str + event: str + status: str + conclusion: str + head_sha: str + + +def gh_run_provenance_v1(run_id: int) -> RunProvenanceV1: + """The run's origin as GitHub reports it, projected to the named fields. + + The second impure boundary of this admission, and the same contract as + the first: it observes and reports, deciding nothing. The query names + the fields instead of fetching the run object, because everything not + named here is a field this module would carry without ever reading it. + """ + + completed = subprocess.run( + ( + "gh", + "api", + f"repos/{{owner}}/{{repo}}/actions/runs/{run_id}", + "--jq", + RUN_PROVENANCE_JQ_V1, + ), + capture_output=True, + text=True, + check=True, + # The first call of the whole campaign: a hung `gh` here is even + # earlier than the artifact listing, and just as indistinguishable + # from work in progress. + timeout=OBSERVATION_TIMEOUT_SECONDS_V1, + ) + return parse_run_provenance_v1(completed.stdout) + + +def parse_run_provenance_v1(stdout: str) -> RunProvenanceV1: + """Decode one run's projected wire form into the observed record. + + Shape only: which values are acceptable is `admit_run_provenance_v1`, + where a test can reach it. What is refused here is a reply that is not + exactly one five-column record — an empty or drifted reply read as a + default would admit precisely the runs this observation exists to catch. + """ + + records = [line for line in stdout.splitlines() if line.strip()] + if len(records) != 1: + raise ValueError( + f"run provenance is not one record: {len(records)} lines" + ) + fields = records[0].split("\t") + if len(fields) != 5: + raise ValueError( + f"run provenance record is not five columns: {len(fields)}" + ) + path, event, status, conclusion, head_sha = ( + field.strip() for field in fields + ) + return RunProvenanceV1(path, event, status, conclusion, head_sha) + + +def admit_run_provenance_v1( + provenance: object, +) -> corpus.ShardCorpusRejectedV1 | None: + """Which observed run may be replayed — the whole rule, and nothing impure. + + A run id and an artifact name say only that some run holds a file of the + right name. Three runs pass that and must not pass this: one produced by + a different workflow, one a fork's pull request produced, and one that + never finished successfully. Each of them sends 256 lanes to replay + something the operator did not intend. + """ + + if type(provenance) is not RunProvenanceV1: + return corpus._reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + "evidence run provenance is not an observed record", + ) + if provenance.path != EVIDENCE_WORKFLOW_PATH_V1: + # The path, not the file name: a workflow of the same basename in a + # foreign directory is a different producer. + return corpus._reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + f"evidence run was produced by {provenance.path!r}," + f" not {EVIDENCE_WORKFLOW_PATH_V1}", + ) + if provenance.event != EVIDENCE_RUN_EVENT_V1: + return corpus._reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + f"evidence run was triggered by {provenance.event!r}," + f" not {EVIDENCE_RUN_EVENT_V1}", + ) + if ( + provenance.status != EVIDENCE_RUN_STATUS_V1 + or provenance.conclusion != EVIDENCE_RUN_CONCLUSION_V1 + ): + # Both halves: a run still in flight can already have uploaded one + # engine's artifact, and a conclusion is only final once the status + # says the run is. + return corpus._reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + f"evidence run is {provenance.status!r}/{provenance.conclusion!r}," + f" not {EVIDENCE_RUN_STATUS_V1}/{EVIDENCE_RUN_CONCLUSION_V1}", + ) + if ( + len(provenance.head_sha) != COMMIT_SHA_LENGTH_V1 + or not COMMIT_SHA_ALPHABET_V1.issuperset(provenance.head_sha) + ): + # The commit is what the operator checks the campaign against, so a + # run that carries no readable one cannot be admitted for being green: + # an abbreviated or absent sha is not something to paste into `git`. + return corpus._reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + f"evidence run carries no commit sha: {provenance.head_sha!r}", + ) + return None + + +def admit_evidence_run_v1( + evidence_run_id: int, + observer: object | None = None, +) -> corpus.ShardCorpusRejectedV1 | None: + """Refuse a dispatch whose evidence run is not the one the campaign means. + + Observation and reporting only; the rule is `admit_run_provenance_v1`. + Any failure to observe is a refusal, never a crash and never a silent + proceed, on the same reasoning as the artifact admission. + + An admitted run's commit goes to the operator here. Nothing in this + process can tell last week's green producer run from this week's — the + rules above admit both — so the one coordinate that decides it is put in + front of the operator at the moment of admission, while 256 lanes have + still not started. + """ + + if observer is None: + observer = gh_run_provenance_v1 + try: + provenance = observer(evidence_run_id) # type: ignore[operator] + except Exception as error: + # Same hostile boundary, same reason to carry the cause: "no such + # run" and "no token" are indistinguishable without stderr, and + # `CalledProcessError.__repr__` drops it. + cause = getattr(error, "stderr", None) or repr(error) + return corpus._reject( + corpus.ShardCorpusReasonV1.FOREIGN_INPUT, + f"verification dispatch cannot observe run {evidence_run_id}:" + f" {str(cause).strip()}", + ) + refusal = admit_run_provenance_v1(provenance) + if refusal is not None: + return refusal + print( + f"evidence run {evidence_run_id} admitted:" + f" head_sha={provenance.head_sha}", + file=sys.stderr, + ) + return None + + def admit_evidence_artifact_v1( evidence_run_id: int, evidence_artifact: str, @@ -404,8 +594,16 @@ def main(argv: list[str] | None = None) -> int: return 64 if args.mode == "verification-dispatch": # Fail closed before the first dispatch, and only here: the printing - # path stays offline by contract, and this observation costs an - # authenticated call that a mistyped width should never spend. + # path stays offline by contract, and these observations cost + # authenticated calls that a mistyped width should never spend. + # Origin before contents: a run of the wrong workflow, the wrong + # trigger or the wrong outcome lists an artifact of exactly the right + # name, so asking about the artifact first would clear a run that + # should never have been considered. + refusal = admit_evidence_run_v1(args.evidence_run_id) + if refusal is not None: + print(f"verification dispatch refused: {refusal.detail}", file=sys.stderr) + return 64 refusal = admit_evidence_artifact_v1( args.evidence_run_id, args.evidence_artifact ) diff --git a/proof/region/v1/tests/test_corpus_dispatch.py b/proof/region/v1/tests/test_corpus_dispatch.py index 79e73403..506eede2 100644 --- a/proof/region/v1/tests/test_corpus_dispatch.py +++ b/proof/region/v1/tests/test_corpus_dispatch.py @@ -374,6 +374,9 @@ def record(command: tuple[str, ...], **kwargs: object) -> object: self.assertEqual(len(seen), 1) self.assertIs(kwargs_seen[0].get("check"), True) self.assertIs(kwargs_seen[0].get("capture_output"), True) + # Without text=True the reply arrives as bytes, the decode raises, + # and every run is refused — a gate that looks alive, admits nobody. + self.assertIs(kwargs_seen[0].get("text"), True) # A hung `gh` would otherwise stall the one call standing between an # operator and 256 dispatches, indistinguishable from work in # progress. @@ -391,16 +394,288 @@ def record(command: tuple[str, ...], **kwargs: object) -> object: ) -class DispatchSeamTests(unittest.TestCase): - """The gate has to be wired in, not merely defined. +ADMITTED_SHA_V1 = "3f0f0a1b2c3d4e5f60718293a4b5c6d7e8f90a1b" + + +def _admitted_provenance(**overrides: str) -> object: + """The run an operator is allowed to replay, with one field disturbed.""" + + fields = dict( + path=".github/workflows/full-domain-run.yml", + event="workflow_dispatch", + status="completed", + conclusion="success", + head_sha=ADMITTED_SHA_V1, + ) + fields.update(overrides) + return corpus_dispatch.RunProvenanceV1(**fields) + + +class RunProvenanceWireTests(unittest.TestCase): + """Observing a run means reading named fields, never a raw reply. - Four unit tests over the admission prove the decision; none of them - prove that `main` asks. Deleting the call would leave those green while - the live path dispatched blind again, so the seam gets its own contract: - a refusal must stop the campaign before the first invocation, and an - admission must let it through. + The artifact listing proves a name exists inside some run; it says + nothing about which workflow produced that run, what triggered it, + whether it finished, or which commit it stands on. Asking GitHub for the + whole run object would drag a large reply of unknown shape through this + process — the project's standing hazard — so the query projects exactly + the fields the rule reads, and the decode refuses anything that is not + that shape rather than inventing a field. """ + CAPTURED_V1 = ( + ".github/workflows/full-domain-run.yml\tworkflow_dispatch\tcompleted\t" + f"success\t{ADMITTED_SHA_V1}\n" + ) + + def test_the_captured_wire_form_decodes_to_the_named_fields(self) -> None: + self.assertEqual( + corpus_dispatch.parse_run_provenance_v1(self.CAPTURED_V1), + _admitted_provenance(), + ) + + def test_a_reply_that_is_not_one_five_column_record_refuses(self) -> None: + # An empty or drifted reply read as a default record would admit the + # exact runs this observation exists to refuse, so shape drift is a + # decode failure — which the admission turns into a typed refusal. + for hostile in ( + "", + "\n", + ".github/workflows/full-domain-run.yml\tworkflow_dispatch\tcompleted" + f"\t{ADMITTED_SHA_V1}\n", + self.CAPTURED_V1.replace("\n", "\textra\n"), + self.CAPTURED_V1 + self.CAPTURED_V1, + f"{{\"path\": \".github/workflows/full-domain-run.yml\"}}\n", + ): + with self.subTest(hostile=hostile): + with self.assertRaises(ValueError): + corpus_dispatch.parse_run_provenance_v1(hostile) + + def test_the_query_projects_the_fields_and_not_the_whole_reply(self) -> None: + # Behavioural, not a look at the source, and load-bearing twice: the + # run object carries far more than this module reads, and printing an + # unknown reply is the known way this project leaks. + seen: list[tuple[str, ...]] = [] + + class _Completed: + stdout = RunProvenanceWireTests.CAPTURED_V1 + returncode = 0 + + def record(command: tuple[str, ...], **kwargs: object) -> object: + seen.append(tuple(command)) + assert kwargs.get("check") is True + assert kwargs.get("capture_output") is True + # Without this the reply arrives as bytes, the decode raises, and + # every run is refused — a gate that looks alive and admits nobody. + assert kwargs.get("text") is True + return _Completed() + + with unittest.mock.patch.object(corpus_dispatch.subprocess, "run", record): + observed = corpus_dispatch.gh_run_provenance_v1(31116022208) + + self.assertEqual(observed, _admitted_provenance()) + self.assertEqual(len(seen), 1) + argv = seen[0] + self.assertEqual(argv[:2], ("gh", "api")) + self.assertIn("repos/{owner}/{repo}/actions/runs/31116022208", argv) + self.assertIn("--jq", argv) + query = argv[argv.index("--jq") + 1] + for field in (".path", ".event", ".status", ".conclusion", ".head_sha"): + self.assertIn(field, query) + self.assertIn("@tsv", query) + # Exactly five selectors: a bare `.` or a dropped projection would + # pull the whole run object through this process. + self.assertEqual(query.count("."), 5) + + +class RunProvenanceRuleTests(unittest.TestCase): + """Which observed run may be replayed is a rule, so it gets tests. + + A run id plus an artifact name is not provenance. A stale run of an old + commit, a run of a different workflow, and a run a fork's pull request + produced all list an artifact of exactly the right name, and all three + send 256 lanes to replay under something the operator did not intend. + """ + + def test_the_producer_run_an_operator_dispatched_is_admitted(self) -> None: + self.assertIsNone( + corpus_dispatch.admit_run_provenance_v1(_admitted_provenance()) + ) + + def test_a_run_of_another_workflow_is_refused(self) -> None: + # The path, not the file name: a workflow of the same basename in a + # foreign directory is a different producer. + for path in ( + ".github/workflows/verification-lanes.yml", + ".github/workflows/ci.yml", + "full-domain-run.yml", + "vendor/.github/workflows/full-domain-run.yml", + "", + ): + with self.subTest(path=path): + result = corpus_dispatch.admit_run_provenance_v1( + _admitted_provenance(path=path) + ) + self.assertIs(type(result), corpus.ShardCorpusRejectedV1) + self.assertEqual( + result.reason, corpus.ShardCorpusReasonV1.FOREIGN_INPUT + ) + + def test_a_run_a_pull_request_produced_is_refused(self) -> None: + # A fork's pull request can run the producer workflow and upload an + # artifact of exactly the allowlisted name carrying a comparator + # bundle nobody reviewed. Only an operator's own dispatch counts. + for event in ("pull_request", "pull_request_target", "push", "schedule", ""): + with self.subTest(event=event): + result = corpus_dispatch.admit_run_provenance_v1( + _admitted_provenance(event=event) + ) + self.assertIs(type(result), corpus.ShardCorpusRejectedV1) + + def test_a_run_that_did_not_finish_successfully_is_refused(self) -> None: + for status, conclusion in ( + ("completed", "failure"), + ("completed", "cancelled"), + ("completed", "timed_out"), + # A run still in flight reports a null conclusion, which the wire + # form renders as an empty column. + ("in_progress", ""), + ("queued", ""), + # Hostile rather than observed: a finished-looking conclusion + # under an unfinished status must not pass on the conclusion + # alone. + ("in_progress", "success"), + ): + with self.subTest(status=status, conclusion=conclusion): + result = corpus_dispatch.admit_run_provenance_v1( + _admitted_provenance(status=status, conclusion=conclusion) + ) + self.assertIs(type(result), corpus.ShardCorpusRejectedV1) + + def test_a_run_without_a_commit_sha_is_refused(self) -> None: + # The sha is what the operator checks the campaign against, so a run + # that does not carry one cannot be admitted merely for being green. + for head_sha in ( + "", + "3f0f0a1b", + ADMITTED_SHA_V1[:-1], + ADMITTED_SHA_V1 + "0", + ADMITTED_SHA_V1[:-1] + "g", + ADMITTED_SHA_V1.upper(), + ): + with self.subTest(head_sha=head_sha): + result = corpus_dispatch.admit_run_provenance_v1( + _admitted_provenance(head_sha=head_sha) + ) + self.assertIs(type(result), corpus.ShardCorpusRejectedV1) + + def test_a_record_that_was_never_observed_is_refused_not_crashed(self) -> None: + for foreign in ( + None, + ( + ".github/workflows/full-domain-run.yml", + "workflow_dispatch", + "completed", + "success", + ADMITTED_SHA_V1, + ), + {"path": ".github/workflows/full-domain-run.yml"}, + "completed", + object(), + ): + with self.subTest(foreign=foreign): + result = corpus_dispatch.admit_run_provenance_v1(foreign) + self.assertIs(type(result), corpus.ShardCorpusRejectedV1) + + +class EvidenceRunAdmissionTests(unittest.TestCase): + """The impure half: observe, refuse on any failure, report the sha. + + Nothing here decides which run is acceptable — that is the pure rule + above. What is proven here is that observation failures land as typed + refusals carrying their cause, that the named run is the one observed, + and that an admitted run's commit reaches the operator. + """ + + def test_an_unreachable_run_is_a_typed_refusal_carrying_the_cause(self) -> None: + def observer(_run_id: int) -> object: + raise subprocess.CalledProcessError( + 1, "gh", stderr="gh: Not Found (HTTP 404)" + ) + + result = corpus_dispatch.admit_evidence_run_v1(99999999, observer) + self.assertIs(type(result), corpus.ShardCorpusRejectedV1) + self.assertIn("Not Found", result.detail) + + def test_an_undecodable_reply_becomes_a_refusal_not_a_traceback(self) -> None: + def observer(_run_id: int) -> object: + return corpus_dispatch.parse_run_provenance_v1("garbage\n") + + result = corpus_dispatch.admit_evidence_run_v1(1, observer) + self.assertIs(type(result), corpus.ShardCorpusRejectedV1) + + def test_the_observer_is_asked_about_the_named_run(self) -> None: + # Anti-vacuity: an observer blind to its argument would let one run be + # bound while another was vouched for. + seen: list[int] = [] + + def observer(run_id: int) -> object: + seen.append(run_id) + return _admitted_provenance() + + errors = io.StringIO() + with contextlib.redirect_stderr(errors): + self.assertIsNone(corpus_dispatch.admit_evidence_run_v1(77, observer)) + self.assertEqual(seen, [77]) + + def test_the_admitted_commit_reaches_the_operator(self) -> None: + # A green producer run of last week's comparator is admissible by + # every rule here and still wrong. Nothing in this process can tell + # which commit the operator meant, so the one coordinate that decides + # it is put in front of them at the moment of admission. + errors = io.StringIO() + with contextlib.redirect_stderr(errors): + result = corpus_dispatch.admit_evidence_run_v1( + 424242, lambda run_id: _admitted_provenance() + ) + self.assertIsNone(result) + self.assertIn(ADMITTED_SHA_V1, errors.getvalue()) + + def test_a_refused_run_reports_no_commit(self) -> None: + # The report belongs to admission: a sha printed beside a refusal + # reads as a cleared campaign. + errors = io.StringIO() + with contextlib.redirect_stderr(errors): + result = corpus_dispatch.admit_evidence_run_v1( + 424242, lambda run_id: _admitted_provenance(event="pull_request") + ) + self.assertIs(type(result), corpus.ShardCorpusRejectedV1) + self.assertNotIn(ADMITTED_SHA_V1, errors.getvalue()) + + +class DispatchSeamTests(unittest.TestCase): + """The gates have to be wired in, not merely defined. + + The unit tests above prove what each admission decides; none of them + prove that `main` asks. Deleting either call would leave them green + while the live path dispatched blind again, so both seams get their own + contract: a refusal must stop the campaign before the first invocation, + an admission must let it through, and the admitted commit must reach the + operator before the lanes do. + """ + + def _provenance_patch(self, observer: object = None) -> object: + """Patch the run observation the live path performs before anything else.""" + + return unittest.mock.patch.object( + corpus_dispatch, + "gh_run_provenance_v1", + observer + if observer is not None + else (lambda run_id: _admitted_provenance()), + ) + + def _argv(self, out: Path, *, live: bool = True) -> list[str]: argv = [ "--mode", @@ -419,7 +694,7 @@ def _argv(self, out: Path, *, live: bool = True) -> list[str]: def test_a_refused_evidence_run_dispatches_nothing(self) -> None: launched: list[tuple[str, ...]] = [] with tempfile.TemporaryDirectory() as tmp: - with unittest.mock.patch.object( + with self._provenance_patch(), unittest.mock.patch.object( corpus_dispatch, "gh_run_artifacts_v1", lambda run_id: (("verification-evidence-arb", False),) @@ -441,7 +716,7 @@ class _Completed: returncode = 0 with tempfile.TemporaryDirectory() as tmp: - with unittest.mock.patch.object( + with self._provenance_patch(), unittest.mock.patch.object( corpus_dispatch, "gh_run_artifacts_v1", # Sensitive to the run id on purpose: an observer that ignores @@ -482,7 +757,7 @@ def test_the_admission_asks_about_the_artifact_the_operator_named(self) -> None: "--out", ] with tempfile.TemporaryDirectory() as tmp: - with unittest.mock.patch.object( + with self._provenance_patch(), unittest.mock.patch.object( corpus_dispatch, "gh_run_artifacts_v1", lambda run_id: (("verification-evidence-arb", False),), @@ -580,7 +855,7 @@ class _Completed: with tempfile.TemporaryDirectory() as tmp: argv = self._argv(Path(tmp) / "out.txt") argv[argv.index("424242")] = "31116022208" - with unittest.mock.patch.object( + with self._provenance_patch(), unittest.mock.patch.object( corpus_dispatch, "gh_run_artifacts_v1", lambda run_id: ( @@ -630,7 +905,9 @@ def test_a_wrong_scale_is_refused_before_the_network_is_asked(self) -> None: # "wrong scale" rather than a diagnosis about the evidence run. observed: list[int] = [] with tempfile.TemporaryDirectory() as tmp: - with unittest.mock.patch.object( + with self._provenance_patch( + lambda run_id: observed.append(run_id) or _admitted_provenance() + ), unittest.mock.patch.object( corpus_dispatch, "gh_run_artifacts_v1", lambda run_id: observed.append(run_id) or (), @@ -664,7 +941,7 @@ def flaky(command: tuple[str, ...], **_kwargs: object) -> object: return _Completed() with tempfile.TemporaryDirectory() as tmp: - with unittest.mock.patch.object( + with self._provenance_patch(), unittest.mock.patch.object( corpus_dispatch, "gh_run_artifacts_v1", lambda run_id: (("verification-evidence-arb", False),), @@ -683,7 +960,7 @@ def flaky(command: tuple[str, ...], **_kwargs: object) -> object: def test_a_missing_gh_is_a_typed_stop_not_a_traceback(self) -> None: with tempfile.TemporaryDirectory() as tmp: - with unittest.mock.patch.object( + with self._provenance_patch(), unittest.mock.patch.object( corpus_dispatch, "gh_run_artifacts_v1", lambda run_id: (("verification-evidence-arb", False),), @@ -697,12 +974,47 @@ def test_a_missing_gh_is_a_typed_stop_not_a_traceback(self) -> None: code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt")) self.assertEqual(code, 64) + def test_the_origin_is_asked_before_the_artifact(self) -> None: + # A run of the wrong workflow lists an artifact of exactly the right + # name, so asking about the artifact first clears a run that should + # never have been considered — and spends a call to do it. The order + # is the claim, so the order is what this pins. + asked: list[str] = [] + with tempfile.TemporaryDirectory() as tmp: + with unittest.mock.patch.object( + corpus_dispatch, + "gh_run_provenance_v1", + lambda run_id: asked.append("origin") + or corpus_dispatch.RunProvenanceV1( + corpus_dispatch.EVIDENCE_WORKFLOW_PATH_V1, + "pull_request", + "completed", + "success", + "0" * 40, + ), + ), unittest.mock.patch.object( + corpus_dispatch, + "gh_run_artifacts_v1", + lambda run_id: asked.append("artifact") + or (("verification-evidence-arb", False),), + ), unittest.mock.patch.object( + corpus_dispatch.subprocess, + "run", + lambda command, **kwargs: None, + ): + code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt")) + # The wrong trigger is refused, and the artifact was never asked about. + self.assertEqual(code, 64) + self.assertEqual(asked, ["origin"]) + def test_printing_stays_offline_and_asks_nobody(self) -> None: # Printing is documented as offline: it must not even observe. observed: list[int] = [] launched: list[tuple[str, ...]] = [] with tempfile.TemporaryDirectory() as tmp: - with unittest.mock.patch.object( + with self._provenance_patch( + lambda run_id: observed.append(run_id) or _admitted_provenance() + ), unittest.mock.patch.object( corpus_dispatch, "gh_run_artifacts_v1", lambda run_id: observed.append(run_id) or (), @@ -718,6 +1030,146 @@ def test_printing_stays_offline_and_asks_nobody(self) -> None: self.assertEqual(observed, []) self.assertEqual(launched, []) + def test_a_run_of_a_foreign_workflow_dispatches_nothing(self) -> None: + # The artifact name is right, the artifact is live, and the run is + # still not the producer's. Without the provenance seam this is 256 + # lanes replaying whatever a foreign workflow happened to upload. + launched: list[tuple[str, ...]] = [] + with tempfile.TemporaryDirectory() as tmp: + with self._provenance_patch( + lambda run_id: _admitted_provenance( + path=".github/workflows/ci.yml" + ) + ), unittest.mock.patch.object( + corpus_dispatch, + "gh_run_artifacts_v1", + lambda run_id: (("verification-evidence-arb", False),), + ), unittest.mock.patch.object( + corpus_dispatch.subprocess, + "run", + lambda command, **kwargs: launched.append(tuple(command)), + ): + errors = io.StringIO() + with contextlib.redirect_stderr(errors): + code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt")) + self.assertEqual(code, 64) + self.assertEqual(launched, []) + + def test_a_run_a_pull_request_produced_dispatches_nothing(self) -> None: + # A fork's pull request can run the producer workflow and publish an + # artifact of exactly the allowlisted name; every earlier rule admits + # it, and the lanes then replay a comparator bundle nobody reviewed. + launched: list[tuple[str, ...]] = [] + with tempfile.TemporaryDirectory() as tmp: + with self._provenance_patch( + lambda run_id: _admitted_provenance(event="pull_request") + ), unittest.mock.patch.object( + corpus_dispatch, + "gh_run_artifacts_v1", + lambda run_id: (("verification-evidence-arb", False),), + ), unittest.mock.patch.object( + corpus_dispatch.subprocess, + "run", + lambda command, **kwargs: launched.append(tuple(command)), + ): + errors = io.StringIO() + with contextlib.redirect_stderr(errors): + code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt")) + self.assertEqual(code, 64) + self.assertEqual(launched, []) + + def test_a_run_that_failed_dispatches_nothing(self) -> None: + # A failed producer run can still have uploaded one engine's artifact + # before the other job died; the artifact listing cannot tell. + launched: list[tuple[str, ...]] = [] + with tempfile.TemporaryDirectory() as tmp: + with self._provenance_patch( + lambda run_id: _admitted_provenance(conclusion="failure") + ), unittest.mock.patch.object( + corpus_dispatch, + "gh_run_artifacts_v1", + lambda run_id: (("verification-evidence-arb", False),), + ), unittest.mock.patch.object( + corpus_dispatch.subprocess, + "run", + lambda command, **kwargs: launched.append(tuple(command)), + ): + errors = io.StringIO() + with contextlib.redirect_stderr(errors): + code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt")) + self.assertEqual(code, 64) + self.assertEqual(launched, []) + + def test_the_admitted_commit_reaches_the_operator_before_the_campaign( + self, + ) -> None: + # Which commit the evidence stands on is the one thing this process + # cannot decide and the operator can. It has to be on their terminal + # before 256 lanes start, not discoverable afterwards in the run log. + launched: list[tuple[str, ...]] = [] + + class _Completed: + returncode = 0 + + with tempfile.TemporaryDirectory() as tmp: + with self._provenance_patch(), unittest.mock.patch.object( + corpus_dispatch, + "gh_run_artifacts_v1", + lambda run_id: (("verification-evidence-arb", False),), + ), unittest.mock.patch.object( + corpus_dispatch.subprocess, + "run", + lambda command, **kwargs: ( + launched.append(tuple(command)) or _Completed() + ), + ): + errors = io.StringIO() + with contextlib.redirect_stderr(errors): + code = corpus_dispatch.main(self._argv(Path(tmp) / "out.txt")) + self.assertEqual(code, 0) + self.assertEqual(len(launched), 256) + reported = errors.getvalue() + self.assertIn(ADMITTED_SHA_V1, reported) + # Before, not after: an operator reading it below "dispatching 256 + # lanes" learns the commit once the campaign is already gone. + self.assertLess( + reported.index(ADMITTED_SHA_V1), reported.index("dispatching 256") + ) + + def test_the_provenance_admission_asks_about_the_run_the_operator_named( + self, + ) -> None: + # Mirror of the artifact pin: a call site that hardcoded a run id + # would vouch for one run while the lanes replayed another. + observed: list[int] = [] + launched: list[tuple[str, ...]] = [] + + class _Completed: + returncode = 0 + + with tempfile.TemporaryDirectory() as tmp: + argv = self._argv(Path(tmp) / "out.txt") + argv[argv.index("424242")] = "31116022208" + with self._provenance_patch( + lambda run_id: observed.append(run_id) or _admitted_provenance() + ), unittest.mock.patch.object( + corpus_dispatch, + "gh_run_artifacts_v1", + lambda run_id: (("verification-evidence-arb", False),), + ), unittest.mock.patch.object( + corpus_dispatch.subprocess, + "run", + lambda command, **kwargs: ( + launched.append(tuple(command)) or _Completed() + ), + ): + errors = io.StringIO() + with contextlib.redirect_stderr(errors): + code = corpus_dispatch.main(argv) + self.assertEqual(code, 0) + self.assertEqual(observed, [31116022208]) + self.assertEqual(len(launched), 256) + def test_a_rejected_command_set_is_exit_64_not_a_type_error(self) -> None: # A foreign artifact name is refused by the pure builder; that # refusal must leave through the typed exit, never as a TypeError