diff --git a/scripts/ci/release_dependency_gate.py b/scripts/ci/release_dependency_gate.py index 8989bfea84..673955ff90 100644 --- a/scripts/ci/release_dependency_gate.py +++ b/scripts/ci/release_dependency_gate.py @@ -141,6 +141,7 @@ def resolve_evidence_binder(script_dir: Path | None = None) -> Path: STRIX_BINDING_UNBOUND = "STRIX_BINDING_UNBOUND" STRIX_TEXTUAL_PASS_REJECTED = "STRIX_TEXTUAL_PASS_REJECTED" STRIX_FINDINGS_OPEN = "STRIX_FINDINGS_OPEN" +STRIX_MATRIX_LIMIT = 256 SHA256_RE = re.compile(r"^[0-9a-f]{64}$") GIT_SHA_RE = re.compile(r"^[0-9a-f]{40}$") @@ -1844,6 +1845,81 @@ def gate(capture_root: Path, stage: str = FULL_STAGE) -> GateReport: return report +def strix_fanout_plan( + capture_root: Path, + license_report: Path, + control_sha: str, + run_id: int, + run_attempt: int, +) -> dict[str, Any]: + """Bind one bounded scan matrix to the passing licence stage's full set.""" + + capture = Path(capture_root) + report = load_json(license_report, LICENSE_MISSING) + release = load_json(capture / "release.json") + if not isinstance(report, Mapping) or not isinstance(release, Mapping): + raise GateError(CAPTURE_INCOMPLETE, "fanout needs release and licence objects") + repository = str(release.get("source_repository", "")) + source_sha = str(release.get("source_sha", "")) + validate_release_identity(repository, source_sha) + if (report.get("result") != "PASS" or report.get("stage") != LICENSE_STAGE + or report.get("source_repository") != repository + or report.get("source_sha") != source_sha): + raise GateError(LICENSE_MISSING, "fanout requires a passing matching licence report") + if (not GIT_SHA_RE.fullmatch(control_sha) or type(run_id) is not int or run_id <= 0 + or type(run_attempt) is not int or run_attempt <= 0): + raise GateError(CAPTURE_INCOMPLETE, "fanout execution identity is invalid") + rows = report.get("dependencies") + if not isinstance(rows, list) or not 1 <= len(rows) <= STRIX_MATRIX_LIMIT: + raise GateError(SCOPE_UNVERIFIABLE, "dependency matrix is empty or exceeds 256 jobs") + fixtures = capture / "strix" / "fixtures" + if fixtures.is_symlink() or not fixtures.is_dir(): + raise GateError(CAPTURE_INCOMPLETE, "fixture directory is unavailable") + planned: list[dict[str, str]] = [] + members: set[str] = set() + keys: set[str] = set() + for row in rows: + if not isinstance(row, Mapping): + raise GateError(CAPTURE_INCOMPLETE, "licence dependency row is malformed") + key = row.get("key") + expected_digest = row.get("fixture_sha256") + if (not isinstance(key, str) or not key or key in keys + or not isinstance(expected_digest, str) + or not SHA256_RE.fullmatch(expected_digest)): + raise GateError(CAPTURE_INCOMPLETE, "licence dependency key or fixture digest is invalid") + slug = _slug_for_key(key) + if (slug in {"", ".", ".."} or Path(slug).name != slug + or "/" in slug or "\\" in slug): + raise GateError(CAPTURE_INCOMPLETE, "dependency fixture slug is unsafe") + fixture_path = _require_regular_file(fixtures / f"{slug}.json", CAPTURE_INCOMPLETE) + digest_path = _require_regular_file(fixtures / f"{slug}.sha256", CAPTURE_INCOMPLETE) + fixture = load_json(fixture_path) + if (fixture_digest(fixture) != expected_digest + or digest_path.read_text(encoding="utf-8").strip() != expected_digest): + raise GateError(SOURCE_HASH_MISMATCH, f"{key}: fixture differs from the licence report") + artifact_name = f"release-strix-binding-a{run_attempt}-" + hashlib.sha256( + key.encode("utf-8") + ).hexdigest() + planned.append({"key": key, "slug": slug, "fixture_sha256": expected_digest, + "artifact_name": artifact_name}) + members.update({f"{slug}.json", f"{slug}.sha256"}) + keys.add(key) + if (len({item["slug"] for item in planned}) != len(planned) + or {entry.name for entry in fixtures.iterdir()} != members + or any(entry.is_symlink() or not entry.is_file() for entry in fixtures.iterdir())): + raise GateError(SCOPE_SET_MISMATCH, "fixture directory differs from the exact licence set") + return { + "schema": "cwl.release-strix-fanout-plan/1", + "source_repository": repository, + "source_sha": source_sha, + "control_sha": control_sha, + "run_id": run_id, + "run_attempt": run_attempt, + "license_report_sha256": _sha256_file(license_report), + "dependencies": planned, + } + + # --------------------------------------------------------------------------- # Sealed-evidence composition with exact-artifact-sbom-attestation.yml # --------------------------------------------------------------------------- @@ -2259,6 +2335,16 @@ def main(argv: Sequence[str] | None = None) -> int: screen.add_argument("--capture", required=True) screen.add_argument("--report", required=True) + fanout = sub.add_parser( + "fanout-plan", help="Emit a bounded exact dependency matrix after licence approval" + ) + fanout.add_argument("--capture", required=True) + fanout.add_argument("--license-report", required=True) + fanout.add_argument("--control-sha", required=True) + fanout.add_argument("--run-id", required=True, type=int) + fanout.add_argument("--run-attempt", required=True, type=int) + fanout.add_argument("--output", required=True) + sub.add_parser( "require-strix-credentials", help="Refuse the Strix stage when a credential is absent" ) @@ -2350,6 +2436,19 @@ def main(argv: Sequence[str] | None = None) -> int: for failure in failures: print(f"ERROR: {failure.code}: {failure.detail}", file=sys.stderr) return 2 if failures else 0 + if args.command == "fanout-plan": + plan = strix_fanout_plan( + Path(args.capture), Path(args.license_report), args.control_sha, + args.run_id, args.run_attempt, + ) + path = Path(args.output) + if path.exists() or path.is_symlink(): + raise GateError(CAPTURE_INCOMPLETE, "fanout plan output already exists") + path.write_text(json.dumps(plan, sort_keys=True) + "\n", encoding="utf-8") + matrix = {"include": plan["dependencies"]} + write_github_output({"matrix_json": json.dumps(matrix, separators=(",", ":"))}, destination) + print(json.dumps(matrix, sort_keys=True)) + return 0 if args.command in {"gate", "prescreen"}: stage = FULL_STAGE if args.command == "gate" else LICENSE_STAGE report = gate(Path(args.capture), stage=stage) diff --git a/tests/test_release_dependency_fanout_plan.py b/tests/test_release_dependency_fanout_plan.py new file mode 100644 index 0000000000..6f10923d32 --- /dev/null +++ b/tests/test_release_dependency_fanout_plan.py @@ -0,0 +1,75 @@ +"""The Strix matrix may only come from the passing full licence set.""" + +from __future__ import annotations + +import copy +import hashlib +import json +from pathlib import Path + +import pytest + +from scripts.ci import release_dependency_gate as gate +from tests.test_release_dependency_gate import build_capture + + +CONTROL = "d" * 40 + + +def _allowed(tmp_path: Path) -> tuple[Path, Path]: + capture = build_capture(tmp_path) + for fixture in (capture / "strix/fixtures").glob("*.json"): + digest = gate.fixture_digest(json.loads(fixture.read_text())) + fixture.with_suffix(".sha256").write_text(digest + "\n") + report = gate.gate(capture, stage=gate.LICENSE_STAGE) + assert report.passed + report_path = tmp_path / "license-report.json" + report_path.write_text(json.dumps(report.to_json()) + "\n") + return capture, report_path + + +def test_fanout_plan_matches_every_prescreened_fixture(tmp_path: Path) -> None: + capture, report_path = _allowed(tmp_path) + plan = gate.strix_fanout_plan(capture, report_path, CONTROL, 42, 2) + report = json.loads(report_path.read_text()) + assert plan["source_sha"] == report["source_sha"] + assert plan["license_report_sha256"] == hashlib.sha256(report_path.read_bytes()).hexdigest() + assert (plan["control_sha"], plan["run_id"], plan["run_attempt"]) == (CONTROL, 42, 2) + assert {row["key"] for row in plan["dependencies"]} == { + row["key"] for row in report["dependencies"] + } + assert len({row["artifact_name"] for row in plan["dependencies"]}) == len(plan["dependencies"]) + assert all(row["artifact_name"].startswith("release-strix-binding-a2-") for row in plan["dependencies"]) + + +def test_plan_refuses_denied_missing_extra_and_duplicate_scope(tmp_path: Path) -> None: + mutators = { + "denied": lambda capture, report: report.__setitem__("result", "FAIL"), + "duplicate": lambda capture, report: report["dependencies"].append(copy.deepcopy(report["dependencies"][0])), + "limit": lambda capture, report: report.__setitem__("dependencies", report["dependencies"] * 257), + "missing": lambda capture, report: next((capture / "strix/fixtures").glob("*.json")).unlink(), + "extra": lambda capture, report: (capture / "strix/fixtures/unlisted.json").write_text("{}"), + "wrong-source": lambda capture, report: report.__setitem__("source_sha", "e" * 40), + } + for name, mutate in mutators.items(): + capture, report_path = _allowed(tmp_path / name) + report = json.loads(report_path.read_text()) + mutate(capture, report) + report_path.write_text(json.dumps(report) + "\n") + with pytest.raises(gate.GateError): + gate.strix_fanout_plan(capture, report_path, CONTROL, 42, 2) + + +def test_fanout_cli_emits_one_bounded_matrix_output(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + capture, report_path = _allowed(tmp_path) + output = tmp_path / "matrix-output.txt" + monkeypatch.setenv("GITHUB_OUTPUT", str(output)) + plan_path = tmp_path / "plan.json" + assert gate.main([ + "fanout-plan", "--capture", str(capture), "--license-report", str(report_path), + "--control-sha", CONTROL, "--run-id", "42", "--run-attempt", "2", + "--output", str(plan_path), + ]) == 0 + matrix = json.loads(output.read_text().removeprefix("matrix_json=")) + assert matrix["include"] == json.loads(plan_path.read_text())["dependencies"] + assert len(matrix["include"]) <= gate.STRIX_MATRIX_LIMIT