Correct qualified recovery authority validation and consumer regressions

Author: durable-workflow-ops

scripts/ci/component-release-recovery.py

@@ -29,13 +29,9 @@
 from typing import Any
 
 import tomllib
-from recovery_workflow_authority import (
-    SOURCE_IDENTITY as RECOVERY_WORKFLOW_AUTHORITY_SOURCE,
-)
 from recovery_workflow_authority import (
     RecoveryWorkflowAuthorityError,
-    authority_url,
-    decode_authority,
+    load_qualified_authority,
     verify_workflow_source,
 )
 
@@ -732,14 +728,15 @@ def discover_plan(
     return tag, commit, plan, preparation
 
 
-def load_recovery_workflow_authority(client: PublicClient) -> dict[str, dict[str, str]]:
+def load_recovery_workflow_authority(
+    client: PublicClient,
+) -> tuple[dict[str, dict[str, str]], dict[str, Any]]:
     identities = {
         name: (component.repository, component.default_branch)
         for name, component in COMPONENTS.items()
     }
     try:
-        raw = client.bytes(authority_url(), accept="application/vnd.github.raw+json")
-        return decode_authority(raw, identities)
+        return load_qualified_authority(client, identities)
     except RecoveryWorkflowAuthorityError as error:
         raise RecoveryError(str(error), "default-branch-preflight") from error
 
@@ -816,7 +813,7 @@ def verify_plan_authority(
     foundation = read_record(client, FOUNDATION_TAG, FOUNDATION_COMMIT, "candidate.json")
     if foundation.get("candidate") != "beta-continuity-foundation":
         raise RecoveryError("immutable candidate foundation has an unexpected identity", "plan-preflight")
-    authority = load_recovery_workflow_authority(client)
+    authority, authority_source = load_recovery_workflow_authority(client)
     branches: dict[str, str] = {}
     recovery_workflows: dict[str, dict[str, Any]] = {}
     for name, component in COMPONENTS.items():
@@ -845,7 +842,7 @@ def verify_plan_authority(
         ).decode("utf-8")
         source_sha256 = verify_recovery_workflow_source(name, source, expected["sha256"])
         recovery_workflows[name] = {
-            "authority": RECOVERY_WORKFLOW_AUTHORITY_SOURCE,
+            "authority": authority_source,
             "default_branch": component.default_branch,
             "path": expected_path,
             "sha256": source_sha256,
@@ -1368,6 +1365,9 @@ def resolve_component(
             ),
         }
     )
+    authority_evidence = next(iter(recovery_workflows.values()), {}).get("authority")
+    if authority_evidence is not None:
+        state["recovery_workflow_authority"] = authority_evidence
     if prepared_identity is not None:
         state["release_preparation"] = {
             "record_commit": record_commit,

scripts/ci/recovery_workflow_authority.py

@@ -1,4 +1,4 @@
-"""Validate the protected component release-recovery workflow tuple."""
+"""Resolve and validate the qualified component recovery-workflow authority."""
 
 from __future__ import annotations
 
@@ -8,15 +8,22 @@
 from collections.abc import Mapping
 from typing import Any
 
-SCHEMA = "durable-workflow.component-release-recovery-authority/v1"
+SCHEMA = "durable-workflow.component-release-recovery-authority/v2"
 CONTROL_REPOSITORY = "durable-workflow/.github"
 AUTHORITY_REF = "main"
 AUTHORITY_PATH = "release-recovery/authority.json"
+QUALIFICATION_WORKFLOW = ".github/workflows/beta-candidate.yml"
+QUALIFICATION_EVENT = "push"
+QUALIFICATION_REF_PATH = f"{QUALIFICATION_WORKFLOW}@{AUTHORITY_REF}"
 WORKFLOW_PATH = ".github/workflows/release-plan-recovery.yml"
 SOURCE_IDENTITY = {
     "repository": CONTROL_REPOSITORY,
     "ref": f"refs/heads/{AUTHORITY_REF}",
     "path": AUTHORITY_PATH,
+    "qualification": {
+        "workflow": QUALIFICATION_WORKFLOW,
+        "event": QUALIFICATION_EVENT,
+    },
 }
 
 
@@ -28,13 +35,123 @@ def normalized_source_sha256(source: str) -> str:
     return hashlib.sha256(source.replace("\r\n", "\n").encode("utf-8")).hexdigest()
 
 
-def authority_url() -> str:
+def authority_ref_url() -> str:
+    return f"https://api.github.com/repos/{CONTROL_REPOSITORY}/commits/{AUTHORITY_REF}"
+
+
+def authority_url(commit: str) -> str:
     return (
         f"https://api.github.com/repos/{CONTROL_REPOSITORY}/contents/{AUTHORITY_PATH}"
-        f"?ref={AUTHORITY_REF}"
+        f"?ref={commit}"
+    )
+
+
+def qualification_runs_url(commit: str) -> str:
+    workflow = QUALIFICATION_WORKFLOW.rsplit("/", 1)[-1]
+    return (
+        f"https://api.github.com/repos/{CONTROL_REPOSITORY}/actions/workflows/{workflow}/runs"
+        f"?branch={AUTHORITY_REF}&event={QUALIFICATION_EVENT}&head_sha={commit}&per_page=100"
+    )
+
+
+def validate_authority_commit(value: Any) -> str:
+    commit = value.get("sha") if isinstance(value, dict) else None
+    if (
+        not isinstance(commit, str)
+        or len(commit) != 40
+        or any(character not in "0123456789abcdef" for character in commit)
+    ):
+        raise RecoveryWorkflowAuthorityError("recovery workflow authority ref has an invalid commit")
+    return commit
+
+
+def _qualification_evidence(run: dict[str, Any], commit: str) -> dict[str, Any]:
+    run_id = run.get("id")
+    run_attempt = run.get("run_attempt")
+    if (
+        not isinstance(run_id, int)
+        or isinstance(run_id, bool)
+        or run_id < 1
+        or not isinstance(run_attempt, int)
+        or isinstance(run_attempt, bool)
+        or run_attempt < 1
+    ):
+        raise RecoveryWorkflowAuthorityError(
+            "recovery workflow authority qualification has an invalid run identity"
+        )
+    return {
+        "workflow": QUALIFICATION_WORKFLOW,
+        "path": run["path"],
+        "event": QUALIFICATION_EVENT,
+        "head_branch": AUTHORITY_REF,
+        "head_sha": commit,
+        "run_id": run_id,
+        "run_attempt": run_attempt,
+        "status": "completed",
+        "conclusion": "success",
+        "url": f"https://github.com/{CONTROL_REPOSITORY}/actions/runs/{run_id}",
+    }
+
+
+def validate_authority_qualification(value: Any, commit: str) -> dict[str, Any]:
+    runs = value.get("workflow_runs") if isinstance(value, dict) else None
+    if not isinstance(runs, list):
+        raise RecoveryWorkflowAuthorityError(
+            "recovery workflow authority qualification response has an invalid shape"
+        )
+
+    candidates = [
+        run
+        for run in runs
+        if isinstance(run, dict)
+        and run.get("path") in (QUALIFICATION_WORKFLOW, QUALIFICATION_REF_PATH)
+        and run.get("event") == QUALIFICATION_EVENT
+        and run.get("head_branch") == AUTHORITY_REF
+    ]
+    if not candidates:
+        raise RecoveryWorkflowAuthorityError(
+            "recovery workflow authority qualification is absent for the resolved commit"
+        )
+    if any(run.get("head_sha") != commit for run in candidates):
+        raise RecoveryWorkflowAuthorityError(
+            "recovery workflow authority qualification is bound to another commit"
+        )
+
+    successful = [
+        run
+        for run in candidates
+        if run.get("status") == "completed" and run.get("conclusion") == "success"
+    ]
+    if successful:
+        return _qualification_evidence(successful[0], commit)
+    if any(run.get("status") != "completed" for run in candidates):
+        raise RecoveryWorkflowAuthorityError(
+            "recovery workflow authority qualification is pending for the resolved commit"
+        )
+    if any(run.get("conclusion") == "cancelled" for run in candidates):
+        raise RecoveryWorkflowAuthorityError(
+            "recovery workflow authority qualification was cancelled for the resolved commit"
+        )
+    raise RecoveryWorkflowAuthorityError(
+        "recovery workflow authority qualification failed for the resolved commit"
     )
 
 
+def qualified_source_identity(
+    raw: bytes,
+    commit: str,
+    qualification: dict[str, Any],
+) -> dict[str, Any]:
+    return {
+        "repository": CONTROL_REPOSITORY,
+        "ref": f"refs/heads/{AUTHORITY_REF}",
+        "commit": commit,
+        "path": AUTHORITY_PATH,
+        "sha256": hashlib.sha256(raw).hexdigest(),
+        "qualification": qualification,
+    }
+
+
 def validate_authority(
     value: Any,
     components: Mapping[str, tuple[str, str]],
@@ -83,6 +200,20 @@ def decode_authority(
     return validate_authority(value, components)
 
 
+def load_qualified_authority(
+    client: Any,
+    components: Mapping[str, tuple[str, str]],
+) -> tuple[dict[str, dict[str, str]], dict[str, Any]]:
+    commit = validate_authority_commit(client.json(authority_ref_url()))
+    qualification = validate_authority_qualification(
+        client.json(qualification_runs_url(commit)),
+        commit,
+    )
+    raw = client.bytes(authority_url(commit), accept="application/vnd.github.raw+json")
+    workflows = decode_authority(raw, components)
+    return workflows, qualified_source_identity(raw, commit, qualification)
+
+
 def verify_workflow_source(name: str, source: str, expected_sha256: str) -> str:
     actual_sha256 = normalized_source_sha256(source)
     if not hmac.compare_digest(actual_sha256, expected_sha256):

scripts/ci/test-component-release-recovery.py

@@ -6,6 +6,7 @@
 import hashlib
 import importlib.util
 import io
+import json
 import sys
 import unittest
 import urllib.error
@@ -16,6 +17,15 @@
     CliRecoveryWorkflowSourceTest,
     CliReleaseAuthorityTest,
 )
+from recovery_workflow_authority import (
+    SCHEMA as AUTHORITY_SCHEMA,
+)
+from recovery_workflow_authority import (
+    SOURCE_IDENTITY,
+    authority_ref_url,
+    authority_url,
+    qualification_runs_url,
+)
 
 RECOVERY_SCRIPT = Path(__file__).with_name("component-release-recovery.py")
 RUST_WORKFLOW_FIXTURE = Path(__file__).with_name("sdk-rust-release-plan-recovery.fixture.yml")
@@ -73,6 +83,129 @@ def load_recovery_for_retry_tests():
     return loader()
 
 
+AUTHORITY_COMMIT = "a" * 40
+
+
+def qualification_run(
+    status: str = "completed",
+    conclusion: str | None = "success",
+    *,
+    head_sha: str = AUTHORITY_COMMIT,
+    head_branch: str = "main",
+    path: str = ".github/workflows/beta-candidate.yml",
+) -> dict[str, object]:
+    return {
+        "id": 81,
+        "run_attempt": 2,
+        "name": "Beta candidate",
+        "workflow_id": 37,
+        "path": path,
+        "event": "push",
+        "head_branch": head_branch,
+        "head_sha": head_sha,
+        "status": status,
+        "conclusion": conclusion,
+        "url": "https://api.github.com/repos/durable-workflow/.github/actions/runs/81",
+        "html_url": "https://github.com/durable-workflow/.github/actions/runs/81",
+    }
+
+
+class QualifiedAuthorityConsumerTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.recovery = load_recovery_for_retry_tests()
+
+    def authority(self) -> dict[str, object]:
+        return {
+            "schema": AUTHORITY_SCHEMA,
+            "source": SOURCE_IDENTITY,
+            "workflows": {
+                name: {
+                    "repository": component.repository,
+                    "ref": f"refs/heads/{component.default_branch}",
+                    "path": ".github/workflows/release-plan-recovery.yml",
+                    "state": "active",
+                    "sha256": "b" * 64,
+                }
+                for name, component in self.recovery.COMPONENTS.items()
+            },
+        }
+
+    def client(self, runs: list[dict[str, object]]):
+        authority_raw = json.dumps(self.authority()).encode("utf-8")
+
+        class Client:
+            def __init__(self) -> None:
+                self.requests: list[tuple[str, str]] = []
+
+            def json(self, url: str) -> dict[str, object]:
+                self.requests.append(("json", url))
+                if url == authority_ref_url():
+                    return {"sha": AUTHORITY_COMMIT}
+                if url == qualification_runs_url(AUTHORITY_COMMIT):
+                    return {"total_count": len(runs), "workflow_runs": runs}
+                raise AssertionError(f"peer source was read before authority qualification: {url}")
+
+            def bytes(self, url: str, *, accept: str | None = None) -> bytes:
+                self.requests.append(("bytes", url))
+                if url != authority_url(AUTHORITY_COMMIT):
+                    raise AssertionError(f"peer source was read before authority qualification: {url}")
+                return authority_raw
+
+        return Client(), authority_raw
+
+    def test_green_qualification_binds_manifest_bytes_and_revision(self) -> None:
+        client, authority_raw = self.client([qualification_run()])
+        workflows, source = self.recovery.load_recovery_workflow_authority(client)
+
+        self.assertEqual(set(self.recovery.COMPONENTS), set(workflows))
+        self.assertEqual(AUTHORITY_COMMIT, source["commit"])
+        self.assertEqual(hashlib.sha256(authority_raw).hexdigest(), source["sha256"])
+        self.assertEqual(AUTHORITY_COMMIT, source["qualification"]["head_sha"])
+        self.assertEqual(".github/workflows/beta-candidate.yml", source["qualification"]["path"])
+        self.assertEqual("main", source["qualification"]["head_branch"])
+        self.assertEqual(
+            [
+                ("json", authority_ref_url()),
+                ("json", qualification_runs_url(AUTHORITY_COMMIT)),
+                ("bytes", authority_url(AUTHORITY_COMMIT)),
+            ],
+            client.requests,
+        )
+
+    def test_non_green_fails_before_authority_or_peer_source_reads(self) -> None:
+        cases = (
+            ("pending", [qualification_run("in_progress", None)], "pending"),
+            ("failed", [qualification_run("completed", "failure")], "failed"),
+            ("cancelled", [qualification_run("completed", "cancelled")], "cancelled"),
+            ("absent", [], "absent"),
+            ("revision-mismatch", [qualification_run(head_sha="c" * 40)], "another commit"),
+            (
+                "wrong-workflow",
+                [qualification_run(path=".github/workflows/source-qualification.yml")],
+                "absent",
+            ),
+            ("wrong-ref", [qualification_run(head_branch="v2")], "absent"),
+            (
+                "wrong-path-ref",
+                [qualification_run(path=".github/workflows/beta-candidate.yml@v2")],
+                "absent",
+            ),
+        )
+        for label, runs, message in cases:
+            with self.subTest(state=label):
+                client, _authority_raw = self.client(runs)
+                with self.assertRaisesRegex(self.recovery.RecoveryError, message):
+                    self.recovery.load_recovery_workflow_authority(client)
+                self.assertEqual(
+                    [
+                        ("json", authority_ref_url()),
+                        ("json", qualification_runs_url(AUTHORITY_COMMIT)),
+                    ],
+                    client.requests,
+                )
+
+
 class ContinuityGateTest(unittest.TestCase):
     @classmethod
     def setUpClass(cls) -> None: