Proof: a dispatch admits the run's origin, not only the artifact's name

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
         )