From 25fbbaf5ad80c8de6dc4142953866cc98386d3a8 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 22 Jul 2026 16:31:28 +0200 Subject: [PATCH] feat(assessment): gate signed release drift --- docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md | 22 + requirements-dev.txt | 1 + tests/test_capability_fit_review.py | 256 +++++++ tools/assessments/capability-fit.py | 129 ++++ .../govoplan_assessment/__init__.py | 5 + .../govoplan_assessment/capability_fit.py | 694 ++++++++++++++++++ 6 files changed, 1107 insertions(+) create mode 100644 tests/test_capability_fit_review.py create mode 100755 tools/assessments/capability-fit.py create mode 100644 tools/assessments/govoplan_assessment/__init__.py create mode 100644 tools/assessments/govoplan_assessment/capability_fit.py diff --git a/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md b/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md index 0279454..7cc9368 100644 --- a/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md +++ b/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md @@ -330,6 +330,28 @@ assumptions, gaps, risks, recommendation and proof check. Reassessment must pin new composition and review any row whose code, evidence, configuration or target requirement changed. +### Repeatable release rerun + +The release-aware rerun tool validates the machine-readable assessment against +its JSON Schema, verifies the catalog's Ed25519 signature and pinned keyring +hash, compares catalog sequence and module versions, and—when local checkouts +are available—checks that each annotated release tag still peels to the assessed +commit. Release drift produces a machine-readable list of affected capability +and infrastructure conclusions instead of silently carrying their prior status +forward: + +```bash +./.venv/bin/python tools/assessments/capability-fit.py --public +``` + +Use `--catalog PATH --keyring PATH` for an offline or release-candidate rerun and +`--json` or `--output PATH` for automation. Exit status `0` means the assessed +release metadata is current, `2` means explicit review is required, and `1` +means the schema or trust checks are blocked. A successful rerun proves only the +assessment schema, signed catalog metadata, and optional local tag provenance; +it does not prove installed artifacts, target providers, target infrastructure, +or production fitness. Those remain separate proof checks above. + ## Evidence used in this slice - [Production-like profile](../dev/production-like/README.md) and diff --git a/requirements-dev.txt b/requirements-dev.txt index 6915338..a4aa7ba 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -29,6 +29,7 @@ httpx==0.28.1 httpx2>=2.5,<3 filelock>=3.20.3 idna>=3.15 +jsonschema>=4,<5 pip>=26.1.2 pip-audit>=2.9,<3 python-multipart>=0.0.31 diff --git a/tests/test_capability_fit_review.py b/tests/test_capability_fit_review.py new file mode 100644 index 0000000..72089cf --- /dev/null +++ b/tests/test_capability_fit_review.py @@ -0,0 +1,256 @@ +from __future__ import annotations + +import base64 +from copy import deepcopy +from datetime import UTC, datetime, timedelta +import hashlib +import importlib.util +import json +from pathlib import Path +import sys +from unittest import mock +import unittest + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + +META_ROOT = Path(__file__).resolve().parents[1] +ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments" +if str(ASSESSMENT_TOOLS_ROOT) not in sys.path: + sys.path.insert(0, str(ASSESSMENT_TOOLS_ROOT)) + +from govoplan_assessment.capability_fit import ( # noqa: E402 + local_tag_provenance, + render_review, + review_capability_fit, +) + + +class CapabilityFitReviewTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.assessment = json.loads( + (META_ROOT / "docs" / "capability-fit-current.json").read_text( + encoding="utf-8" + ) + ) + cls.schema = json.loads( + (META_ROOT / "docs" / "capability-fit.schema.json").read_text( + encoding="utf-8" + ) + ) + + def test_matching_signed_catalog_is_current(self) -> None: + catalog, keyring = signed_catalog(self.assessment) + + report = review_capability_fit( + assessment=deepcopy(self.assessment), + schema=self.schema, + catalog=catalog, + keyring=keyring, + ) + + self.assertEqual("current", report["status"]) + self.assertEqual([], report["changes"]) + self.assertFalse(report["proof_scope"]["installed_artifacts"]["checked"]) + self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) + self.assertFalse(report["proof_scope"]["local_tag_provenance"]["checked"]) + repeated = review_capability_fit( + assessment=deepcopy(self.assessment), + schema=self.schema, + catalog=catalog, + keyring=keyring, + ) + self.assertEqual( + json.dumps(report, sort_keys=True), json.dumps(repeated, sort_keys=True) + ) + + def test_release_drift_identifies_affected_conclusions(self) -> None: + catalog, keyring = signed_catalog( + self.assessment, versions={"campaigns": "0.1.11"}, sequence=202607230001 + ) + + report = review_capability_fit( + assessment=deepcopy(self.assessment), + schema=self.schema, + catalog=catalog, + keyring=keyring, + ) + + self.assertEqual("review_required", report["status"]) + target_ids = {item["id"] for item in report["review_targets"]} + self.assertIn("campaign.journey", target_ids) + self.assertIn("composition.campaigns", target_ids) + self.assertIn("assessment.release", target_ids) + self.assertIn( + "composition_version_changed", {item["code"] for item in report["findings"]} + ) + + def test_untrusted_or_substituted_keyring_blocks_rerun(self) -> None: + catalog, _ = signed_catalog(self.assessment) + _, other_keyring = signed_catalog(self.assessment) + + report = review_capability_fit( + assessment=deepcopy(self.assessment), + schema=self.schema, + catalog=catalog, + keyring=other_keyring, + ) + + self.assertEqual("blocked", report["status"]) + codes = {item["code"] for item in report["findings"]} + self.assertIn("catalog_trust", codes) + self.assertIn("catalog_keyring_hash_mismatch", codes) + self.assertTrue(report["proof_scope"]["assessment_schema"]["valid"]) + self.assertFalse( + report["proof_scope"]["catalog_signature_and_keyring"]["valid"] + ) + + def test_schema_error_blocks_before_comparison(self) -> None: + assessment = deepcopy(self.assessment) + assessment.pop("scope") + catalog, keyring = signed_catalog(self.assessment) + + report = review_capability_fit( + assessment=assessment, + schema=self.schema, + catalog=catalog, + keyring=keyring, + ) + + self.assertEqual("blocked", report["status"]) + self.assertEqual( + {"assessment_schema"}, {item["code"] for item in report["findings"]} + ) + self.assertFalse(report["proof_scope"]["assessment_schema"]["valid"]) + self.assertFalse( + report["proof_scope"]["catalog_signature_and_keyring"]["checked"] + ) + self.assertIsNone( + report["proof_scope"]["catalog_signature_and_keyring"]["valid"] + ) + + def test_human_report_states_proof_limit(self) -> None: + catalog, keyring = signed_catalog(self.assessment) + report = review_capability_fit( + assessment=deepcopy(self.assessment), + schema=self.schema, + catalog=catalog, + keyring=keyring, + ) + + rendered = render_review(report) + + self.assertIn("Capability fit rerun: current", rendered) + self.assertIn( + "No installed-artifact, target-provider, or production proof", rendered + ) + + def test_public_fetch_failure_is_generic_and_blocking(self) -> None: + script = META_ROOT / "tools" / "assessments" / "capability-fit.py" + spec = importlib.util.spec_from_file_location("capability_fit_cli", script) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + with mock.patch.object( + module, + "fetch_json", + return_value={ + "ok": False, + "error": "Authorization: secret at https://internal.invalid", + }, + ): + with self.assertRaisesRegex( + SystemExit, r"^Could not fetch fixed public catalog endpoint[.]$" + ): + module.main(["--public", "--skip-tag-provenance"]) + + def test_repository_path_traversal_is_not_resolved(self) -> None: + with mock.patch("govoplan_assessment.capability_fit.subprocess.run") as run: + result = local_tag_provenance( + workspace_root=META_ROOT.parent, + repository="../outside", + version="0.1.0", + assessed_commit="0123456", + ) + + self.assertEqual("tag_provenance_unavailable", result["kind"]) + self.assertIn("invalid repository identifier", result["message"]) + run.assert_not_called() + + +def signed_catalog( + assessment: dict[str, object], + *, + versions: dict[str, str] | None = None, + sequence: int = 202607220843, +) -> tuple[dict[str, object], dict[str, object]]: + versions = versions or {} + private_key = Ed25519PrivateKey.generate() + public_key = base64.b64encode( + private_key.public_key().public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw, + ) + ).decode("ascii") + keyring: dict[str, object] = { + "keyring_version": "1", + "purpose": "test", + "generated_at": "2026-01-01T00:00:00Z", + "keys": [{"key_id": "test-key", "status": "active", "public_key": public_key}], + } + now = datetime.now(tz=UTC) + modules: list[dict[str, object]] = [] + core_release: dict[str, object] | None = None + for component in assessment["composition"]: + module_id = str(component["module_id"]) + repository = str(component["repository"]) + version = versions.get(module_id, str(component["manifest_version"])) + entry: dict[str, object] = { + "module_id": module_id, + "name": module_id, + "version": version, + "python_package": repository, + "python_ref": f"{repository} @ git+ssh://git@example.invalid/example/{repository}.git@v{version}", + } + if module_id == "core": + core_release = entry + else: + modules.append(entry) + assert core_release is not None + catalog: dict[str, object] = { + "catalog_version": "1", + "channel": "stable", + "sequence": sequence, + "generated_at": now.isoformat().replace("+00:00", "Z"), + "expires_at": (now + timedelta(days=30)).isoformat().replace("+00:00", "Z"), + "core_release": core_release, + "modules": modules, + "release": {"keyring_sha256": canonical_hash(keyring)}, + } + signature_payload = json.dumps( + catalog, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + catalog["signatures"] = [ + { + "algorithm": "ed25519", + "key_id": "test-key", + "value": base64.b64encode(private_key.sign(signature_payload)).decode( + "ascii" + ), + } + ] + return catalog, keyring + + +def canonical_hash(payload: object) -> str: + encoded = json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/assessments/capability-fit.py b/tools/assessments/capability-fit.py new file mode 100755 index 0000000..09c482f --- /dev/null +++ b/tools/assessments/capability-fit.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 +"""Re-run a capability-fit assessment against a trusted release catalog.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import sys + + +META_ROOT = Path(__file__).resolve().parents[2] +ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments" +RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release" +for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT): + if str(tools_root) not in sys.path: + sys.path.insert(0, str(tools_root)) + +from govoplan_assessment import render_review, review_capability_fit # noqa: E402 +from govoplan_release.catalog import DEFAULT_PUBLIC_BASE_URL, fetch_json # noqa: E402 + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Validate a fit assessment and identify conclusions needing review after release drift." + ) + parser.add_argument( + "--assessment", + type=Path, + default=META_ROOT / "docs" / "capability-fit-current.json", + ) + parser.add_argument( + "--schema", + type=Path, + default=META_ROOT / "docs" / "capability-fit.schema.json", + ) + source = parser.add_mutually_exclusive_group(required=True) + source.add_argument( + "--public", + action="store_true", + help="Use the fixed public GovOPlaN stable catalog and keyring URLs.", + ) + source.add_argument( + "--catalog", type=Path, help="Read a catalog from a local file." + ) + parser.add_argument( + "--keyring", type=Path, help="Trusted local keyring; required with --catalog." + ) + parser.add_argument( + "--workspace-root", + type=Path, + default=META_ROOT.parent, + help="Workspace containing repository checkouts for local tag provenance.", + ) + parser.add_argument( + "--skip-tag-provenance", + action="store_true", + help="Compare signed metadata without checking local annotated tags.", + ) + parser.add_argument( + "--json", action="store_true", help="Print the machine-readable review report." + ) + parser.add_argument( + "--output", + type=Path, + help="Also write the machine-readable review report to this path.", + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + if args.catalog is not None and args.keyring is None: + raise SystemExit("--keyring is required with --catalog") + if args.public and args.keyring is not None: + raise SystemExit("--keyring cannot be combined with --public") + + assessment = read_object(args.assessment, label="assessment") + schema = read_object(args.schema, label="schema") + if args.public: + catalog = fetch_public_json( + f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/channels/stable.json", + label="catalog", + ) + keyring = fetch_public_json( + f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/keyring.json", label="keyring" + ) + else: + catalog = read_object(args.catalog, label="catalog") + keyring = read_object(args.keyring, label="keyring") + + report = review_capability_fit( + assessment=assessment, + schema=schema, + catalog=catalog, + keyring=keyring, + workspace_root=None + if args.skip_tag_provenance + else args.workspace_root.resolve(), + ) + encoded = json.dumps(report, indent=2, sort_keys=True) + "\n" + if args.output is not None: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(encoded, encoding="utf-8") + sys.stdout.write(encoded if args.json else render_review(report)) + return {"current": 0, "blocked": 1, "review_required": 2}[report["status"]] + + +def read_object(path: Path | None, *, label: str) -> dict[str, object]: + if path is None: + raise SystemExit(f"Missing {label} path") + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise SystemExit(f"Could not read {label} {path}: {exc}") from exc + if not isinstance(payload, dict): + raise SystemExit(f"{label.capitalize()} must be a JSON object: {path}") + return payload + + +def fetch_public_json(url: str, *, label: str) -> dict[str, object]: + result = fetch_json(url) + if result.get("ok") is not True or not isinstance(result.get("payload"), dict): + raise SystemExit(f"Could not fetch fixed public {label} endpoint.") + return result["payload"] + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/assessments/govoplan_assessment/__init__.py b/tools/assessments/govoplan_assessment/__init__.py new file mode 100644 index 0000000..05fdc9e --- /dev/null +++ b/tools/assessments/govoplan_assessment/__init__.py @@ -0,0 +1,5 @@ +"""Repeatable GovOPlaN product-assessment tooling.""" + +from .capability_fit import review_capability_fit, render_review + +__all__ = ("render_review", "review_capability_fit") diff --git a/tools/assessments/govoplan_assessment/capability_fit.py b/tools/assessments/govoplan_assessment/capability_fit.py new file mode 100644 index 0000000..711a5f0 --- /dev/null +++ b/tools/assessments/govoplan_assessment/capability_fit.py @@ -0,0 +1,694 @@ +"""Compare a capability-fit assessment with a signed release catalog. + +The result is deliberately limited to release metadata and local tag +provenance. It never implies installed-artifact, target-provider, or production +proof. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +import hashlib +import json +from pathlib import Path +import re +import subprocess +import tempfile +from typing import Any, Iterable + +from jsonschema import Draft202012Validator, FormatChecker +from jsonschema.exceptions import SchemaError + +from govoplan_core.core.module_package_catalog import validate_module_package_catalog + + +RELEASE_REF_PATTERN = re.compile( + r"^(?P[a-z][a-z0-9_-]*)-catalog-(?P[0-9]+)$" +) +TAG_PATTERN = re.compile(r"(?:[.]git[@#])(?P[^\s]+)$") +REPOSITORY_ID_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]*$") + + +@dataclass(frozen=True, slots=True) +class Finding: + severity: str + code: str + message: str + assessment_ids: tuple[str, ...] = () + + +def review_capability_fit( + *, + assessment: dict[str, Any], + schema: dict[str, Any], + catalog: dict[str, Any], + keyring: dict[str, Any], + workspace_root: Path | None = None, +) -> dict[str, Any]: + """Return a secret-free release-drift report for one fit assessment.""" + + findings: list[Finding] = [] + schema_errors = validate_assessment(assessment=assessment, schema=schema) + findings.extend( + Finding("blocker", "assessment_schema", message) for message in schema_errors + ) + if schema_errors: + return build_report( + assessment=assessment, + catalog=catalog, + findings=findings, + changes=[], + review_targets=[], + catalog_checked=False, + local_tag_provenance_checked=False, + ) + + catalog_validation = validate_catalog(catalog=catalog, keyring=keyring) + if catalog_validation.get("valid") is not True: + findings.append( + Finding( + "blocker", + "catalog_trust", + str(catalog_validation.get("error") or "Catalog validation failed."), + ("assessment.release",), + ) + ) + expected_keyring_hash = _catalog_keyring_hash(catalog) + actual_keyring_hash = canonical_hash(keyring) + if expected_keyring_hash is None: + findings.append( + Finding( + "blocker", + "catalog_keyring_hash_missing", + "Catalog release metadata does not pin the trusted keyring hash.", + ("assessment.release",), + ) + ) + elif expected_keyring_hash != actual_keyring_hash: + findings.append( + Finding( + "blocker", + "catalog_keyring_hash_mismatch", + "The supplied keyring does not match the hash pinned by the catalog.", + ("assessment.release",), + ) + ) + + release = assessment["release"] + ref_match = RELEASE_REF_PATTERN.fullmatch(str(release["ref"])) + if ref_match is None: + findings.append( + Finding( + "blocker", + "assessment_release_ref", + "Assessment release.ref must have the form '-catalog-'.", + ("assessment.release",), + ) + ) + assessed_channel = None + assessed_sequence = None + else: + assessed_channel = ref_match.group("channel") + assessed_sequence = int(ref_match.group("sequence")) + + catalog_channel = _text(catalog.get("channel")) + catalog_sequence = _integer(catalog.get("sequence")) + if assessed_channel is not None and catalog_channel != assessed_channel: + findings.append( + Finding( + "blocker", + "catalog_channel_mismatch", + f"Assessment pins channel {assessed_channel!r}, but the supplied catalog is {catalog_channel!r}.", + ("assessment.release",), + ) + ) + if assessed_sequence is not None and catalog_sequence != assessed_sequence: + findings.append( + Finding( + "review", + "catalog_sequence_changed", + f"Assessment pins catalog sequence {assessed_sequence}; supplied catalog sequence is {catalog_sequence!r}.", + ("assessment.release",), + ) + ) + + catalog_entries, entry_findings = catalog_entries_by_id(catalog) + findings.extend(entry_findings) + changes: list[dict[str, Any]] = [] + changed_repositories: set[str] = set() + assessed_module_ids: set[str] = set() + + for module in assessment["composition"]: + module_id = str(module["module_id"]) + repository = str(module["repository"]) + assessed_module_ids.add(module_id) + entry = catalog_entries.get(module_id) + if entry is None: + changes.append( + { + "module_id": module_id, + "repository": repository, + "kind": "missing_from_catalog", + "assessed_version": module["manifest_version"], + "catalog_version": None, + } + ) + changed_repositories.add(repository) + findings.append( + Finding( + "review", + "composition_entry_missing", + f"Assessed module {module_id!r} is not present in the supplied catalog.", + (f"composition.{module_id}",), + ) + ) + continue + + catalog_repository = entry["repository"] + if catalog_repository != repository: + changes.append( + { + "module_id": module_id, + "repository": repository, + "kind": "repository_changed", + "assessed_repository": repository, + "catalog_repository": catalog_repository, + } + ) + changed_repositories.add(repository) + findings.append( + Finding( + "review", + "composition_repository_changed", + f"Module {module_id!r} maps to {catalog_repository!r} in the catalog, not {repository!r}.", + (f"composition.{module_id}",), + ) + ) + + assessed_version = str(module["manifest_version"]) + catalog_version = entry["version"] + if assessed_version != catalog_version: + changes.append( + { + "module_id": module_id, + "repository": repository, + "kind": "version_changed", + "assessed_version": assessed_version, + "catalog_version": catalog_version, + } + ) + changed_repositories.add(repository) + findings.append( + Finding( + "review", + "composition_version_changed", + f"Module {module_id!r} changed from {assessed_version!r} to {catalog_version!r}.", + (f"composition.{module_id}",), + ) + ) + + expected_tag = f"v{catalog_version}" if catalog_version else None + catalog_tags = entry["tags"] + if ( + expected_tag is None + or not catalog_tags + or any(tag != expected_tag for tag in catalog_tags) + ): + findings.append( + Finding( + "blocker", + "catalog_ref_version_mismatch", + f"Catalog refs for {module_id!r} do not consistently select expected tag {expected_tag!r}: {catalog_tags!r}.", + (f"composition.{module_id}",), + ) + ) + + if workspace_root is not None: + provenance = local_tag_provenance( + workspace_root=workspace_root, + repository=repository, + version=assessed_version, + assessed_commit=str(module["commit"]), + ) + if provenance is not None: + changes.append( + { + "module_id": module_id, + "repository": repository, + "kind": provenance["kind"], + "assessed_commit": module["commit"], + "tag_commit": provenance.get("tag_commit"), + } + ) + changed_repositories.add(repository) + findings.append( + Finding( + "review", + "tag_provenance_changed", + str(provenance["message"]), + (f"composition.{module_id}",), + ) + ) + + extras = sorted(set(catalog_entries) - assessed_module_ids) + if extras: + findings.append( + Finding( + "info", + "catalog_entries_outside_scope", + "Catalog entries outside this assessment composition: " + + ", ".join(extras) + + ".", + ) + ) + + review_targets = conclusions_needing_review( + assessment=assessment, + changed_repositories=changed_repositories, + release_changed=assessed_sequence != catalog_sequence, + ) + return build_report( + assessment=assessment, + catalog=catalog, + findings=findings, + changes=changes, + review_targets=review_targets, + catalog_checked=True, + local_tag_provenance_checked=workspace_root is not None, + ) + + +def validate_assessment( + *, assessment: dict[str, Any], schema: dict[str, Any] +) -> list[str]: + try: + Draft202012Validator.check_schema(schema) + except SchemaError as exc: + return [f"$schema: {exc.message}"] + validator = Draft202012Validator(schema, format_checker=FormatChecker()) + return [ + f"{_json_path(error.absolute_path)}: {error.message}" + for error in sorted( + validator.iter_errors(assessment), + key=lambda item: ( + tuple(str(part) for part in item.absolute_path), + str(item.validator), + item.message, + ), + ) + ] + + +def validate_catalog( + *, catalog: dict[str, Any], keyring: dict[str, Any] +) -> dict[str, object]: + trusted_keys = trusted_keys_from_keyring(keyring) + channel = _text(catalog.get("channel")) + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", encoding="utf-8" + ) as handle: + json.dump(catalog, handle) + handle.flush() + return validate_module_package_catalog( + Path(handle.name), + require_trusted=True, + approved_channels=(channel,) if channel else (), + trusted_keys=trusted_keys, + ) + + +def trusted_keys_from_keyring(payload: dict[str, Any]) -> dict[str, str]: + now = datetime.now(tz=UTC) + result: dict[str, str] = {} + keys = payload.get("keys") + if not isinstance(keys, list): + return result + for item in keys: + if not isinstance(item, dict): + continue + if str(item.get("status") or "active").strip().lower() != "active": + continue + raw_not_before = item.get("not_before") + raw_not_after = item.get("not_after") + not_before = _datetime(raw_not_before) + not_after = _datetime(raw_not_after) + if raw_not_before is not None and not_before is None: + continue + if raw_not_after is not None and not_after is None: + continue + if not_before is not None and now < not_before: + continue + if not_after is not None and now > not_after: + continue + key_id = _text(item.get("key_id")) + public_key = _text(item.get("public_key")) or _text( + item.get("public_key_base64") + ) + if key_id and public_key: + result[key_id] = public_key + return result + + +def catalog_entries_by_id( + catalog: dict[str, Any], +) -> tuple[dict[str, dict[str, Any]], list[Finding]]: + findings: list[Finding] = [] + entries: dict[str, dict[str, Any]] = {} + raw_entries: list[tuple[str, dict[str, Any]]] = [] + core = catalog.get("core_release") + if isinstance(core, dict): + raw_entries.append(("core", core)) + modules = catalog.get("modules") + if isinstance(modules, list): + raw_entries.extend( + (str(item.get("module_id") or ""), item) + for item in modules + if isinstance(item, dict) + ) + for module_id, item in raw_entries: + if not module_id: + findings.append( + Finding( + "blocker", + "catalog_module_id_missing", + "Catalog entry has no module_id.", + ) + ) + continue + if module_id in entries: + findings.append( + Finding( + "blocker", + "catalog_module_id_duplicate", + f"Catalog repeats module_id {module_id!r}.", + ) + ) + continue + refs = [ + value + for value in (item.get("python_ref"), item.get("webui_ref")) + if isinstance(value, str) and value + ] + entries[module_id] = { + "module_id": module_id, + "repository": catalog_repository(item), + "version": _text(item.get("version")), + "tags": tuple( + tag for tag in (catalog_ref_tag(value) for value in refs) if tag + ), + } + return entries, findings + + +def catalog_repository(entry: dict[str, Any]) -> str | None: + package = _text(entry.get("python_package")) + if package: + return package.split("[", 1)[0] + for field in ("python_ref", "webui_ref"): + value = _text(entry.get(field)) + if not value: + continue + match = re.search(r"/([^/@#]+)[.]git(?:[@#]|$)", value) + if match: + return match.group(1) + return None + + +def catalog_ref_tag(value: str) -> str | None: + match = TAG_PATTERN.search(value) + return match.group("tag") if match else None + + +def local_tag_provenance( + *, + workspace_root: Path, + repository: str, + version: str, + assessed_commit: str, +) -> dict[str, str] | None: + if REPOSITORY_ID_PATTERN.fullmatch(repository) is None: + return { + "kind": "tag_provenance_unavailable", + "message": "Cannot verify tag provenance for an invalid repository identifier.", + } + resolved_workspace = workspace_root.resolve() + repo = (resolved_workspace / repository).resolve() + if repo.parent != resolved_workspace: + return { + "kind": "tag_provenance_unavailable", + "message": f"Cannot verify {repository} tag provenance outside the configured workspace.", + } + if not (repo / ".git").exists(): + return { + "kind": "tag_provenance_unavailable", + "message": f"Cannot verify {repository} tag provenance because the local checkout is unavailable.", + } + try: + tag_type = subprocess.run( + ["git", "-C", str(repo), "cat-file", "-t", f"refs/tags/v{version}"], + check=False, + capture_output=True, + text=True, + ) + result = subprocess.run( + [ + "git", + "-C", + str(repo), + "rev-parse", + "--verify", + f"refs/tags/v{version}^{{commit}}", + ], + check=False, + capture_output=True, + text=True, + ) + except OSError: + return { + "kind": "tag_provenance_unavailable", + "message": f"Cannot run the local tag provenance check for {repository}.", + } + if tag_type.returncode != 0: + return { + "kind": "tag_provenance_unavailable", + "message": f"Cannot inspect {repository} tag v{version} in the local checkout.", + } + if tag_type.stdout.strip() != "tag": + return { + "kind": "tag_not_annotated", + "message": f"{repository} v{version} is not an annotated release tag.", + } + if result.returncode != 0: + return { + "kind": "tag_provenance_unavailable", + "message": f"Cannot peel {repository} tag v{version} in the local checkout.", + } + tag_commit = result.stdout.strip().lower() + if not tag_commit.startswith(assessed_commit.lower()): + return { + "kind": "tag_commit_changed", + "message": f"{repository} v{version} peels to {tag_commit[:12]}, not assessed commit {assessed_commit}.", + "tag_commit": tag_commit, + } + return None + + +def conclusions_needing_review( + *, + assessment: dict[str, Any], + changed_repositories: set[str], + release_changed: bool, +) -> list[dict[str, Any]]: + targets: dict[str, dict[str, Any]] = {} + if release_changed: + targets["assessment.release"] = { + "id": "assessment.release", + "section": "release", + "reason": "Catalog sequence changed; review release-scoped evidence and conclusions.", + } + for section in ("capabilities", "infrastructure"): + for item in assessment.get(section, []): + if not isinstance(item, dict): + continue + locators = "\n".join( + str(evidence.get("locator") or "") + for evidence in item.get("evidence", []) + if isinstance(evidence, dict) + ).lower() + matched = sorted( + repository + for repository in changed_repositories + if repository.lower() in locators + ) + if not matched: + continue + item_id = str(item["id"]) + targets[item_id] = { + "id": item_id, + "section": section, + "current_status": item.get("status"), + "repositories": matched, + "reason": "Evidence references a repository whose assessed release changed.", + } + for module in assessment.get("composition", []): + if ( + not isinstance(module, dict) + or module.get("repository") not in changed_repositories + ): + continue + module_id = str(module.get("module_id")) + target_id = f"composition.{module_id}" + targets[target_id] = { + "id": target_id, + "section": "composition", + "repositories": [module.get("repository")], + "reason": "Pinned module release or tag provenance changed.", + } + return [targets[key] for key in sorted(targets)] + + +def build_report( + *, + assessment: dict[str, Any], + catalog: dict[str, Any], + findings: Iterable[Finding], + changes: list[dict[str, Any]], + review_targets: list[dict[str, Any]], + catalog_checked: bool, + local_tag_provenance_checked: bool, +) -> dict[str, Any]: + finding_list = sorted( + findings, + key=lambda item: (_severity_rank(item.severity), item.code, item.message), + ) + finding_codes = {item.code for item in finding_list} + schema_valid = "assessment_schema" not in finding_codes + catalog_valid = catalog_checked and not finding_codes.intersection( + { + "catalog_trust", + "catalog_keyring_hash_missing", + "catalog_keyring_hash_mismatch", + } + ) + release_valid = catalog_valid and not any( + item.severity in {"blocker", "review"} for item in finding_list + ) + tag_provenance_valid = ( + local_tag_provenance_checked and "tag_provenance_changed" not in finding_codes + ) + if any(item.severity == "blocker" for item in finding_list): + status = "blocked" + elif any(item.severity == "review" for item in finding_list): + status = "review_required" + else: + status = "current" + return { + "report_version": "0.1.0", + "status": status, + "assessment_id": assessment.get("assessment_id"), + "assessment_release": assessment.get("release", {}).get("ref") + if isinstance(assessment.get("release"), dict) + else None, + "catalog": { + "channel": catalog.get("channel"), + "sequence": catalog.get("sequence"), + "generated_at": catalog.get("generated_at"), + }, + "proof_scope": { + "assessment_schema": {"checked": True, "valid": schema_valid}, + "catalog_signature_and_keyring": { + "checked": catalog_checked, + "valid": catalog_valid if catalog_checked else None, + }, + "release_metadata": { + "checked": catalog_checked, + "valid": release_valid if catalog_checked else None, + }, + "local_tag_provenance": { + "checked": local_tag_provenance_checked, + "valid": tag_provenance_valid if local_tag_provenance_checked else None, + }, + "installed_artifacts": {"checked": False, "valid": None}, + "target_environment": {"checked": False, "valid": None}, + "external_providers": {"checked": False, "valid": None}, + "production_approval": {"checked": False, "valid": None}, + }, + "findings": [asdict(item) for item in finding_list], + "changes": sorted( + changes, + key=lambda item: (str(item.get("module_id")), str(item.get("kind"))), + ), + "review_targets": review_targets, + } + + +def render_review(report: dict[str, Any]) -> str: + lines = [ + f"Capability fit rerun: {report['status']}", + f"Assessment: {report.get('assessment_id') or '-'} ({report.get('assessment_release') or '-'})", + f"Catalog: {report['catalog'].get('channel') or '-'} sequence {report['catalog'].get('sequence') or '-'}", + "Scope: schema, signed release metadata, and optional local tag provenance only.", + ] + findings = report.get("findings") or [] + if findings: + lines.append("Findings:") + lines.extend( + f"- [{item['severity']}] {item['code']}: {item['message']}" + for item in findings + ) + else: + lines.append("Findings: none") + targets = report.get("review_targets") or [] + if targets: + lines.append("Conclusions requiring review:") + lines.extend(f"- {item['id']}: {item['reason']}" for item in targets) + lines.append( + "No installed-artifact, target-provider, or production proof was performed." + ) + return "\n".join(lines) + "\n" + + +def canonical_hash(payload: object) -> str: + data = json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("utf-8") + return hashlib.sha256(data).hexdigest() + + +def _catalog_keyring_hash(catalog: dict[str, Any]) -> str | None: + release = catalog.get("release") + return _text(release.get("keyring_sha256")) if isinstance(release, dict) else None + + +def _datetime(value: object) -> datetime | None: + text = _text(value) + if text is None: + return None + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) + + +def _integer(value: object) -> int | None: + try: + return int(value) if value is not None else None + except (TypeError, ValueError): + return None + + +def _json_path(parts: Iterable[object]) -> str: + path = "$" + for part in parts: + path += f"[{part}]" if isinstance(part, int) else f".{part}" + return path + + +def _severity_rank(severity: str) -> int: + return {"blocker": 0, "review": 1, "info": 2}.get(severity, 3) + + +def _text(value: object) -> str | None: + return value if isinstance(value, str) and value else None