diff --git a/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md b/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md index 7cc9368..fea5b1f 100644 --- a/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md +++ b/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md @@ -333,24 +333,42 @@ 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 +its JSON Schema, verifies the catalog's Ed25519 signature against an +independently pinned local trust keyring, and separately verifies the canonical +hash of the published or candidate keyring pinned by the signed catalog. The +downloaded keyring is therefore never its own sole trust root. The tool compares +catalog sequence and module versions and—when local checkouts are available—checks +annotated release tags against assessed commits. For repositories represented in +signed `release.selected_units`, both the peeled commit and annotated tag-object +ID must match exactly; re-annotating an unchanged commit requires review. +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 +./.venv/bin/python tools/assessments/capability-fit.py \ + --public \ + --trusted-keyring /srv/govoplan/trust/catalog-keyring.json ``` -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. +Use `--catalog CATALOG --keyring PUBLISHED_OR_CANDIDATE_KEYRING +--trusted-keyring PINNED_LOCAL_KEYRING` for an offline or release-candidate +rerun. `GOVOPLAN_MODULE_PACKAGE_CATALOG_TRUSTED_KEYS_FILE` may provide the +independently managed local trust-root path instead. During key rotation this +local file may contain both current and `next` keys within their validity +windows; revoked, disabled and retired keys are rejected. Do not establish this +trust file by downloading it in the same rerun. Public JSON responses are +bounded to 16 MiB. Use `--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. The machine report distinguishes independent signature trust, +published-keyring hash agreement, and local-tag proof. Local-tag proof is only +marked checked when every composition entry is available for comparison and was +attempted. A successful rerun proves only the assessment schema, signed catalog +metadata, published-keyring integrity 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 @@ -365,7 +383,8 @@ or production fitness. Those remain separate proof checks above. Files, Mail, Audit and Addresses tests - [Campaign delivery runbook](https://git.add-ideas.de/add-ideas/govoplan-campaign/src/branch/main/docs/CAMPAIGN_DELIVERY_RUNBOOK.md) - [Live signed stable catalog](https://govoplan.add-ideas.de/catalogs/v1/channels/stable.json) - and [trusted keyring](https://govoplan.add-ideas.de/catalogs/v1/keyring.json) + and [published keyring](https://govoplan.add-ideas.de/catalogs/v1/keyring.json), + verified against a separately provisioned local trust keyring - Annotated source tags `govoplan-core/v0.1.13` and `govoplan-campaign/v0.1.10`, including their catalogued Python and WebUI refs diff --git a/tests/test_capability_fit_review.py b/tests/test_capability_fit_review.py index 3d57c64..47c8930 100644 --- a/tests/test_capability_fit_review.py +++ b/tests/test_capability_fit_review.py @@ -8,6 +8,7 @@ import importlib.util import json import os from pathlib import Path +import subprocess import sys import tempfile from unittest import mock @@ -28,6 +29,7 @@ from govoplan_assessment.capability_fit import ( # noqa: E402 local_tag_provenance, render_review, review_capability_fit, + trusted_keys_from_keyring, ) @@ -52,7 +54,8 @@ class CapabilityFitReviewTests(unittest.TestCase): assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, - keyring=keyring, + published_keyring=keyring, + trusted_keyring=keyring, ) self.assertEqual("current", report["status"]) @@ -64,7 +67,8 @@ class CapabilityFitReviewTests(unittest.TestCase): assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, - keyring=keyring, + published_keyring=keyring, + trusted_keyring=keyring, ) self.assertEqual( json.dumps(report, sort_keys=True), json.dumps(repeated, sort_keys=True) @@ -79,7 +83,8 @@ class CapabilityFitReviewTests(unittest.TestCase): assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, - keyring=keyring, + published_keyring=keyring, + trusted_keyring=keyring, ) self.assertEqual("review_required", report["status"]) @@ -114,7 +119,8 @@ class CapabilityFitReviewTests(unittest.TestCase): assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, - keyring=keyring, + published_keyring=keyring, + trusted_keyring=keyring, ) self.assertEqual("blocked", report["status"]) @@ -147,7 +153,8 @@ class CapabilityFitReviewTests(unittest.TestCase): assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, - keyring=keyring, + published_keyring=keyring, + trusted_keyring=keyring, ) self.assertEqual("review_required", report["status"]) @@ -201,7 +208,8 @@ class CapabilityFitReviewTests(unittest.TestCase): assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, - keyring=keyring, + published_keyring=keyring, + trusted_keyring=keyring, ) self.assertEqual("blocked", report["status"]) @@ -231,7 +239,8 @@ class CapabilityFitReviewTests(unittest.TestCase): assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, - keyring=keyring, + published_keyring=keyring, + trusted_keyring=keyring, ) self.assertEqual( replay_environment, @@ -241,24 +250,43 @@ class CapabilityFitReviewTests(unittest.TestCase): self.assertEqual("current", report["status"]) self.assertTrue(report["proof_scope"]["catalog_signature_and_keyring"]["valid"]) - def test_untrusted_or_substituted_keyring_blocks_rerun(self) -> None: - catalog, _ = signed_catalog(self.assessment) - _, other_keyring = signed_catalog(self.assessment) + def test_self_consistent_substitution_fails_independent_trust_root(self) -> None: + _, trusted_keyring = signed_catalog(self.assessment) + catalog, published_keyring = signed_catalog(self.assessment) report = review_capability_fit( assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, - keyring=other_keyring, + published_keyring=published_keyring, + trusted_keyring=trusted_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.assertNotIn("catalog_keyring_hash_mismatch", codes) self.assertTrue(report["proof_scope"]["assessment_schema"]["valid"]) self.assertFalse( - report["proof_scope"]["catalog_signature_and_keyring"]["valid"] + report["proof_scope"]["catalog_signature_and_trusted_keyring"]["valid"] + ) + self.assertTrue(report["proof_scope"]["published_keyring_hash"]["valid"]) + + def test_locally_pinned_rotation_keys_follow_core_status_semantics(self) -> None: + keys = trusted_keys_from_keyring( + { + "keys": [ + {"key_id": "active", "status": "active", "public_key": "a"}, + {"key_id": "next", "status": "next", "public_key": "b"}, + {"key_id": "retired", "status": "retired", "public_key": "c"}, + ] + } + ) + + self.assertEqual({"active": "a", "next": "b"}, keys) + self.assertEqual( + {"legacy": "base64-key"}, + trusted_keys_from_keyring({"legacy": "base64-key"}), ) def test_schema_error_blocks_before_comparison(self) -> None: @@ -270,7 +298,8 @@ class CapabilityFitReviewTests(unittest.TestCase): assessment=assessment, schema=self.schema, catalog=catalog, - keyring=keyring, + published_keyring=keyring, + trusted_keyring=keyring, ) self.assertEqual("blocked", report["status"]) @@ -291,7 +320,8 @@ class CapabilityFitReviewTests(unittest.TestCase): assessment=deepcopy(self.assessment), schema=self.schema, catalog=catalog, - keyring=keyring, + published_keyring=keyring, + trusted_keyring=keyring, ) rendered = render_review(report) @@ -308,19 +338,190 @@ class CapabilityFitReviewTests(unittest.TestCase): 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[.]$" + with tempfile.TemporaryDirectory() as temp_dir: + trusted_path = Path(temp_dir) / "trusted-keyring.json" + trusted_path.write_text('{"keys": []}\n', encoding="utf-8") + 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", + "--trusted-keyring", + str(trusted_path), + "--skip-tag-provenance", + ] + ) + + def test_cli_requires_an_independent_local_trust_root(self) -> None: + script = META_ROOT / "tools" / "assessments" / "capability-fit.py" + spec = importlib.util.spec_from_file_location( + "capability_fit_cli_trust", 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.dict( + os.environ, + {module.TRUSTED_KEYRING_FILE_ENV: ""}, + clear=False, + ): + with self.assertRaisesRegex(SystemExit, r"--trusted-keyring .* required"): module.main(["--public", "--skip-tag-provenance"]) + def test_local_selected_tag_commit_and_object_must_match_exactly(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + workspace = Path(temp_dir) + commit_sha, tag_object_sha = create_tagged_repository( + workspace=workspace, + repository="govoplan-core", + version="0.1.0", + ) + assessment = deepcopy(self.assessment) + core = next( + deepcopy(item) + for item in assessment["composition"] + if item["module_id"] == "core" + ) + core["manifest_version"] = "0.1.0" + core["commit"] = commit_sha[:12] + assessment["composition"] = [core] + selected_unit = { + "repo": "govoplan-core", + "version": "0.1.0", + "tag": "v0.1.0", + "commit_sha": commit_sha, + "tag_object_sha": tag_object_sha, + } + catalog, keyring = signed_catalog( + assessment, + selected_units=[selected_unit], + ) + + baseline = review_capability_fit( + assessment=deepcopy(assessment), + schema=self.schema, + catalog=catalog, + published_keyring=keyring, + trusted_keyring=keyring, + workspace_root=workspace, + ) + + self.assertEqual("current", baseline["status"]) + self.assertTrue(baseline["proof_scope"]["local_tag_provenance"]["valid"]) + + forged_commit = commit_sha[:12] + ("0" * 28) + if forged_commit == commit_sha: + forged_commit = commit_sha[:12] + ("f" * 28) + forged_catalog, forged_keyring = signed_catalog( + assessment, + selected_units=[{**selected_unit, "commit_sha": forged_commit}], + ) + commit_mismatch = review_capability_fit( + assessment=deepcopy(assessment), + schema=self.schema, + catalog=forged_catalog, + published_keyring=forged_keyring, + trusted_keyring=forged_keyring, + workspace_root=workspace, + ) + + self.assertEqual("review_required", commit_mismatch["status"]) + self.assertIn( + "tag_signed_commit_changed", + {item["kind"] for item in commit_mismatch["changes"]}, + ) + + repo = workspace / "govoplan-core" + git_text(repo, "tag", "-d", "v0.1.0") + git_text(repo, "tag", "-a", "v0.1.0", "-m", "Re-annotated release") + reannotated = review_capability_fit( + assessment=deepcopy(assessment), + schema=self.schema, + catalog=catalog, + published_keyring=keyring, + trusted_keyring=keyring, + workspace_root=workspace, + ) + + self.assertEqual("review_required", reannotated["status"]) + self.assertIn( + "tag_object_changed", + {item["kind"] for item in reannotated["changes"]}, + ) + self.assertTrue(reannotated["proof_scope"]["local_tag_provenance"]["checked"]) + self.assertFalse(reannotated["proof_scope"]["local_tag_provenance"]["valid"]) + + def test_missing_composition_entry_does_not_claim_complete_tag_proof(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + workspace = Path(temp_dir) + core_commit, core_tag_object = create_tagged_repository( + workspace=workspace, + repository="govoplan-core", + version="0.1.0", + ) + tenancy_commit, _ = create_tagged_repository( + workspace=workspace, + repository="govoplan-tenancy", + version="0.1.0", + ) + reviewed_assessment = deepcopy(self.assessment) + core = next( + deepcopy(item) + for item in reviewed_assessment["composition"] + if item["module_id"] == "core" + ) + tenancy = next( + deepcopy(item) + for item in reviewed_assessment["composition"] + if item["module_id"] == "tenancy" + ) + core.update(manifest_version="0.1.0", commit=core_commit[:12]) + tenancy.update(manifest_version="0.1.0", commit=tenancy_commit[:12]) + catalog_assessment = deepcopy(reviewed_assessment) + catalog_assessment["composition"] = [core] + reviewed_assessment["composition"] = [core, tenancy] + catalog, keyring = signed_catalog( + catalog_assessment, + selected_units=[ + { + "repo": "govoplan-core", + "version": "0.1.0", + "tag": "v0.1.0", + "commit_sha": core_commit, + "tag_object_sha": core_tag_object, + } + ], + ) + + report = review_capability_fit( + assessment=reviewed_assessment, + schema=self.schema, + catalog=catalog, + published_keyring=keyring, + trusted_keyring=keyring, + workspace_root=workspace, + ) + + proof = report["proof_scope"]["local_tag_provenance"] + self.assertEqual("review_required", report["status"]) + self.assertFalse(proof["checked"]) + self.assertIsNone(proof["valid"]) + self.assertEqual(2, proof["attempted_count"]) + self.assertEqual(2, proof["expected_count"]) + self.assertNotIn( + "tag_provenance_changed", {item["code"] for item in report["findings"]} + ) + 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( @@ -335,6 +536,36 @@ class CapabilityFitReviewTests(unittest.TestCase): run.assert_not_called() +def create_tagged_repository( + *, workspace: Path, repository: str, version: str +) -> tuple[str, str]: + repo = workspace / repository + git_text(workspace, "init", "-b", "main", str(repo)) + git_text(repo, "config", "user.name", "Capability Fit Test") + git_text(repo, "config", "user.email", "capability-fit@example.invalid") + (repo / "README.md").write_text(f"{repository}\n", encoding="utf-8") + git_text(repo, "add", "README.md") + git_text(repo, "commit", "-m", "Initial release fixture") + git_text(repo, "tag", "-a", f"v{version}", "-m", f"Release v{version}") + return ( + git_text(repo, "rev-parse", "HEAD"), + git_text(repo, "rev-parse", f"refs/tags/v{version}"), + ) + + +def git_text(cwd: Path, *args: str) -> str: + result = subprocess.run( + ("git", *args), + cwd=cwd, + check=False, + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise AssertionError(result.stderr or result.stdout) + return result.stdout.strip() + + def signed_catalog( assessment: dict[str, object], *, diff --git a/tests/test_release_http_fetch.py b/tests/test_release_http_fetch.py new file mode 100644 index 0000000..8b96ba1 --- /dev/null +++ b/tests/test_release_http_fetch.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import sys +import unittest +from pathlib import Path +from unittest import mock + + +META_ROOT = Path(__file__).resolve().parents[1] +RELEASE_ROOT = META_ROOT / "tools" / "release" +if str(RELEASE_ROOT) not in sys.path: + sys.path.insert(0, str(RELEASE_ROOT)) + +from govoplan_release.http_fetch import ( # noqa: E402 + DEFAULT_MAX_RESPONSE_BYTES, + fetch_http, +) + + +class ReleaseHttpFetchTests(unittest.TestCase): + def test_content_length_over_default_limit_blocks_before_read(self) -> None: + response = FakeResponse( + body=b"ignored", + headers={"Content-Length": str(DEFAULT_MAX_RESPONSE_BYTES + 1)}, + ) + + with mock.patch( + "govoplan_release.http_fetch.urllib.request.urlopen", + return_value=response, + ): + with self.assertRaisesRegex(ValueError, "exceeds the configured size"): + fetch_http("https://example.invalid/catalog.json", timeout=1) + + self.assertEqual([], response.read_sizes) + + def test_streamed_body_is_capped_when_length_is_absent(self) -> None: + response = FakeResponse(body=b"x" * 9, headers={}) + + with mock.patch( + "govoplan_release.http_fetch.urllib.request.urlopen", + return_value=response, + ): + with self.assertRaisesRegex(ValueError, "exceeds the configured size"): + fetch_http( + "https://example.invalid/catalog.json", + timeout=1, + max_response_bytes=8, + ) + + self.assertEqual([9], response.read_sizes) + + +class FakeResponse: + def __init__(self, *, body: bytes, headers: dict[str, str]) -> None: + self.status = 200 + self.headers = headers + self.body = body + self.read_sizes: list[int] = [] + + def __enter__(self) -> FakeResponse: + return self + + def __exit__(self, *args: object) -> None: + return None + + def read(self, size: int) -> bytes: + self.read_sizes.append(size) + return self.body[:size] + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/assessments/capability-fit.py b/tools/assessments/capability-fit.py index 09c482f..645975a 100755 --- a/tools/assessments/capability-fit.py +++ b/tools/assessments/capability-fit.py @@ -5,6 +5,7 @@ from __future__ import annotations import argparse import json +import os from pathlib import Path import sys @@ -20,6 +21,9 @@ from govoplan_assessment import render_review, review_capability_fit # noqa: E4 from govoplan_release.catalog import DEFAULT_PUBLIC_BASE_URL, fetch_json # noqa: E402 +TRUSTED_KEYRING_FILE_ENV = "GOVOPLAN_MODULE_PACKAGE_CATALOG_TRUSTED_KEYS_FILE" + + 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." @@ -44,7 +48,17 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: "--catalog", type=Path, help="Read a catalog from a local file." ) parser.add_argument( - "--keyring", type=Path, help="Trusted local keyring; required with --catalog." + "--keyring", + type=Path, + help="Published or candidate keyring whose hash is pinned by --catalog.", + ) + parser.add_argument( + "--trusted-keyring", + type=Path, + help=( + "Independently pinned local keyring used as the catalog signature trust " + f"root; defaults to ${TRUSTED_KEYRING_FILE_ENV}." + ), ) parser.add_argument( "--workspace-root", @@ -74,7 +88,14 @@ def main(argv: list[str] | None = None) -> int: raise SystemExit("--keyring is required with --catalog") if args.public and args.keyring is not None: raise SystemExit("--keyring cannot be combined with --public") - + configured_trusted_keyring = os.getenv(TRUSTED_KEYRING_FILE_ENV, "").strip() + trusted_keyring_path = args.trusted_keyring or ( + Path(configured_trusted_keyring) if configured_trusted_keyring else None + ) + if trusted_keyring_path is None: + raise SystemExit( + f"--trusted-keyring or ${TRUSTED_KEYRING_FILE_ENV} is required" + ) assessment = read_object(args.assessment, label="assessment") schema = read_object(args.schema, label="schema") if args.public: @@ -82,18 +103,24 @@ def main(argv: list[str] | None = None) -> int: 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" + published_keyring = fetch_public_json( + f"{DEFAULT_PUBLIC_BASE_URL}/catalogs/v1/keyring.json", + label="published keyring", ) else: catalog = read_object(args.catalog, label="catalog") - keyring = read_object(args.keyring, label="keyring") + published_keyring = read_object(args.keyring, label="published keyring") + trusted_keyring = read_object( + trusted_keyring_path, + label="trusted keyring", + ) report = review_capability_fit( assessment=assessment, schema=schema, catalog=catalog, - keyring=keyring, + published_keyring=published_keyring, + trusted_keyring=trusted_keyring, workspace_root=None if args.skip_tag_provenance else args.workspace_root.resolve(), diff --git a/tools/assessments/govoplan_assessment/capability_fit.py b/tools/assessments/govoplan_assessment/capability_fit.py index 02c2eea..1a616cf 100644 --- a/tools/assessments/govoplan_assessment/capability_fit.py +++ b/tools/assessments/govoplan_assessment/capability_fit.py @@ -46,7 +46,8 @@ def review_capability_fit( assessment: dict[str, Any], schema: dict[str, Any], catalog: dict[str, Any], - keyring: dict[str, Any], + published_keyring: dict[str, Any], + trusted_keyring: dict[str, Any], workspace_root: Path | None = None, ) -> dict[str, Any]: """Return a secret-free release-drift report for one fit assessment.""" @@ -64,10 +65,15 @@ def review_capability_fit( changes=[], review_targets=[], catalog_checked=False, - local_tag_provenance_checked=False, + local_tag_provenance_attempted=0, + local_tag_provenance_expected=0, + local_tag_catalog_scope_complete=False, ) - catalog_validation = validate_catalog(catalog=catalog, keyring=keyring) + catalog_validation = validate_catalog( + catalog=catalog, + trusted_keyring=trusted_keyring, + ) if catalog_validation.get("valid") is not True: findings.append( Finding( @@ -100,13 +106,13 @@ def review_capability_fit( for issue in source_selection.issues ) expected_keyring_hash = _catalog_keyring_hash(catalog) - actual_keyring_hash = canonical_hash(keyring) + actual_keyring_hash = canonical_hash(published_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.", + "Catalog release metadata does not pin the published keyring hash.", ("assessment.release",), ) ) @@ -115,7 +121,7 @@ def review_capability_fit( Finding( "blocker", "catalog_keyring_hash_mismatch", - "The supplied keyring does not match the hash pinned by the catalog.", + "The published keyring does not match the hash pinned by the catalog.", ("assessment.release",), ) ) @@ -163,13 +169,58 @@ def review_capability_fit( changes: list[dict[str, Any]] = [] changed_repositories: set[str] = set() assessed_module_ids: set[str] = set() + composition = assessment["composition"] + local_tag_provenance_expected = len(composition) + local_tag_provenance_attempted = 0 + local_tag_catalog_scope_complete = True - for module in assessment["composition"]: + for module in composition: module_id = str(module["module_id"]) repository = str(module["repository"]) + assessed_version = str(module["manifest_version"]) + selected_version = source_selection.selected_versions.get(repository) + selected_commit = source_selection.selected_commits.get(repository) + selected_tag_object = source_selection.selected_tag_objects.get(repository) assessed_module_ids.add(module_id) + if workspace_root is not None: + local_tag_provenance_attempted += 1 + provenance = local_tag_provenance( + workspace_root=workspace_root, + repository=repository, + version=assessed_version, + assessed_commit=str(module["commit"]), + selected_commit=( + selected_commit if selected_version == assessed_version else None + ), + selected_tag_object=( + selected_tag_object + if selected_version == assessed_version + else None + ), + ) + 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"), + "tag_object": provenance.get("tag_object"), + } + ) + changed_repositories.add(repository) + findings.append( + Finding( + "review", + "tag_provenance_changed", + str(provenance["message"]), + (f"composition.{module_id}",), + ) + ) entry = catalog_entries.get(module_id) if entry is None: + local_tag_catalog_scope_complete = False changes.append( { "module_id": module_id, @@ -211,7 +262,6 @@ def review_capability_fit( ) ) - assessed_version = str(module["manifest_version"]) catalog_version = entry["version"] if assessed_version != catalog_version: changes.append( @@ -233,8 +283,6 @@ def review_capability_fit( ) ) - selected_version = source_selection.selected_versions.get(repository) - selected_commit = source_selection.selected_commits.get(repository) assessed_commit = str(module["commit"]).lower() if ( selected_version == assessed_version @@ -280,33 +328,6 @@ def review_capability_fit( ) ) - 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( @@ -331,7 +352,9 @@ def review_capability_fit( changes=changes, review_targets=review_targets, catalog_checked=True, - local_tag_provenance_checked=workspace_root is not None, + local_tag_provenance_attempted=local_tag_provenance_attempted, + local_tag_provenance_expected=local_tag_provenance_expected, + local_tag_catalog_scope_complete=local_tag_catalog_scope_complete, ) @@ -357,9 +380,9 @@ def validate_assessment( def validate_catalog( - *, catalog: dict[str, Any], keyring: dict[str, Any] + *, catalog: dict[str, Any], trusted_keyring: dict[str, Any] ) -> dict[str, object]: - trusted_keys = trusted_keys_from_keyring(keyring) + trusted_keys = trusted_keys_from_keyring(trusted_keyring) channel = _text(catalog.get("channel")) with tempfile.NamedTemporaryFile( mode="w", suffix=".json", encoding="utf-8" @@ -394,11 +417,19 @@ def trusted_keys_from_keyring(payload: dict[str, Any]) -> dict[str, str]: result: dict[str, str] = {} keys = payload.get("keys") if not isinstance(keys, list): - return result + return { + str(key): str(value) + for key, value in payload.items() + if str(key).strip() and str(value).strip() + } for item in keys: if not isinstance(item, dict): continue - if str(item.get("status") or "active").strip().lower() != "active": + if str(item.get("status") or "active").strip().lower() in { + "revoked", + "disabled", + "retired", + }: continue raw_not_before = item.get("not_before") raw_not_after = item.get("not_after") @@ -497,6 +528,8 @@ def local_tag_provenance( repository: str, version: str, assessed_commit: str, + selected_commit: str | None = None, + selected_tag_object: str | None = None, ) -> dict[str, str] | None: if REPOSITORY_ID_PATTERN.fullmatch(repository) is None: return { @@ -535,6 +568,19 @@ def local_tag_provenance( capture_output=True, text=True, ) + tag_object_result = subprocess.run( + [ + "git", + "-C", + str(repo), + "rev-parse", + "--verify", + f"refs/tags/v{version}", + ], + check=False, + capture_output=True, + text=True, + ) except OSError: return { "kind": "tag_provenance_unavailable", @@ -555,13 +601,40 @@ def local_tag_provenance( "kind": "tag_provenance_unavailable", "message": f"Cannot peel {repository} tag v{version} in the local checkout.", } + if tag_object_result.returncode != 0: + return { + "kind": "tag_provenance_unavailable", + "message": f"Cannot resolve {repository} tag object v{version} in the local checkout.", + } tag_commit = result.stdout.strip().lower() + tag_object = tag_object_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, } + if selected_commit is not None and tag_commit != selected_commit.lower(): + return { + "kind": "tag_signed_commit_changed", + "message": ( + f"{repository} v{version} peels to {tag_commit[:12]}, not signed " + f"selected-unit commit {selected_commit[:12]}." + ), + "tag_commit": tag_commit, + "tag_object": tag_object, + } + if selected_tag_object is not None and tag_object != selected_tag_object.lower(): + return { + "kind": "tag_object_changed", + "message": ( + f"{repository} v{version} resolves to annotated tag object " + f"{tag_object[:12]}, not signed selected-unit tag object " + f"{selected_tag_object[:12]}." + ), + "tag_commit": tag_commit, + "tag_object": tag_object, + } return None @@ -627,7 +700,9 @@ def build_report( changes: list[dict[str, Any]], review_targets: list[dict[str, Any]], catalog_checked: bool, - local_tag_provenance_checked: bool, + local_tag_provenance_attempted: int, + local_tag_provenance_expected: int, + local_tag_catalog_scope_complete: bool, ) -> dict[str, Any]: finding_list = sorted( findings, @@ -635,18 +710,24 @@ def build_report( ) 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( + trusted_signature_valid = catalog_checked and "catalog_trust" not in finding_codes + published_keyring_valid = catalog_checked and not finding_codes.intersection( { - "catalog_trust", "catalog_keyring_hash_missing", "catalog_keyring_hash_mismatch", } ) + catalog_valid = trusted_signature_valid and published_keyring_valid release_valid = catalog_valid and not any( item.severity in {"blocker", "review"} for item in finding_list ) + tag_provenance_checked = ( + local_tag_provenance_expected > 0 + and local_tag_provenance_attempted == local_tag_provenance_expected + and local_tag_catalog_scope_complete + ) tag_provenance_valid = ( - local_tag_provenance_checked and "tag_provenance_changed" not in finding_codes + tag_provenance_checked and "tag_provenance_changed" not in finding_codes ) if any(item.severity == "blocker" for item in finding_list): status = "blocked" @@ -672,13 +753,23 @@ def build_report( "checked": catalog_checked, "valid": catalog_valid if catalog_checked else None, }, + "catalog_signature_and_trusted_keyring": { + "checked": catalog_checked, + "valid": trusted_signature_valid if catalog_checked else None, + }, + "published_keyring_hash": { + "checked": catalog_checked, + "valid": published_keyring_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, + "checked": tag_provenance_checked, + "valid": tag_provenance_valid if tag_provenance_checked else None, + "attempted_count": local_tag_provenance_attempted, + "expected_count": local_tag_provenance_expected, }, "installed_artifacts": {"checked": False, "valid": None}, "target_environment": {"checked": False, "valid": None}, diff --git a/tools/release/govoplan_release/catalog.py b/tools/release/govoplan_release/catalog.py index 8e579a4..cb6d95e 100644 --- a/tools/release/govoplan_release/catalog.py +++ b/tools/release/govoplan_release/catalog.py @@ -7,11 +7,12 @@ import json from pathlib import Path from urllib.error import HTTPError, URLError -from .http_fetch import fetch_http +from .http_fetch import DEFAULT_MAX_RESPONSE_BYTES, fetch_http from .model import CatalogSnapshot from .workspace import website_root DEFAULT_PUBLIC_BASE_URL = "https://govoplan.add-ideas.de" +MAX_PUBLIC_JSON_RESPONSE_BYTES = DEFAULT_MAX_RESPONSE_BYTES def collect_catalog_snapshot( @@ -22,9 +23,13 @@ def collect_catalog_snapshot( check_public: bool = False, ) -> CatalogSnapshot: web_root = website_root(workspace_root) - catalog_path = web_root / "public" / "catalogs" / "v1" / "channels" / f"{channel}.json" + catalog_path = ( + web_root / "public" / "catalogs" / "v1" / "channels" / f"{channel}.json" + ) keyring_path = web_root / "public" / "catalogs" / "v1" / "keyring.json" - public_catalog_url = f"{public_base_url.rstrip('/')}/catalogs/v1/channels/{channel}.json" + public_catalog_url = ( + f"{public_base_url.rstrip('/')}/catalogs/v1/channels/{channel}.json" + ) public_keyring_url = f"{public_base_url.rstrip('/')}/catalogs/v1/keyring.json" catalog_version: str | None = None @@ -129,8 +134,12 @@ def collect_catalog_snapshot( public_keyring_hash=public_keyring_hash, public_catalog_error=public_catalog_error, public_keyring_error=public_keyring_error, - catalog_matches_public=compare_hashes(local_catalog_hash, public_catalog_hash) if check_public else None, - keyring_matches_public=compare_hashes(local_keyring_hash, public_keyring_hash) if check_public else None, + catalog_matches_public=compare_hashes(local_catalog_hash, public_catalog_hash) + if check_public + else None, + keyring_matches_public=compare_hashes(local_keyring_hash, public_keyring_hash) + if check_public + else None, ) @@ -149,7 +158,9 @@ def parse_catalog_payload(payload: object) -> dict[str, str | int | None]: signatures = payload.get("signatures") return { "catalog_version": string_or_none(payload.get("catalog_version")), - "core_release_version": string_or_none(core_release.get("version")) if isinstance(core_release, dict) else None, + "core_release_version": string_or_none(core_release.get("version")) + if isinstance(core_release, dict) + else None, "module_count": len(modules) if isinstance(modules, list) else 0, "signature_count": len(signatures) if isinstance(signatures, list) else 0, "generated_at": string_or_none(payload.get("generated_at")), @@ -164,6 +175,7 @@ def fetch_json(url: str) -> dict[str, object]: timeout=10, label="Public release catalog URL", headers={"User-Agent": "GovOPlaN release-console"}, + max_response_bytes=MAX_PUBLIC_JSON_RESPONSE_BYTES, ) return {"ok": True, "payload": json.loads(response.text())} except (HTTPError, URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc: @@ -171,7 +183,9 @@ def fetch_json(url: str) -> dict[str, object]: def canonical_hash(payload: object) -> str: - canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + canonical = json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("utf-8") return hashlib.sha256(canonical).hexdigest() diff --git a/tools/release/govoplan_release/http_fetch.py b/tools/release/govoplan_release/http_fetch.py index f3ea512..fbb54a6 100644 --- a/tools/release/govoplan_release/http_fetch.py +++ b/tools/release/govoplan_release/http_fetch.py @@ -8,6 +8,9 @@ from dataclasses import dataclass from typing import Mapping +DEFAULT_MAX_RESPONSE_BYTES = 16 * 1024 * 1024 + + @dataclass(frozen=True, slots=True) class HttpFetchResponse: status: int @@ -34,11 +37,29 @@ def fetch_http( label: str = "URL", method: str = "GET", headers: Mapping[str, str] | None = None, + max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES, ) -> HttpFetchResponse: - request = urllib.request.Request(validate_http_url(url, label=label), headers=dict(headers or {}), method=method) + if max_response_bytes <= 0: + raise ValueError("max_response_bytes must be positive.") + request = urllib.request.Request( + validate_http_url(url, label=label), + headers=dict(headers or {}), + method=method, + ) with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - URL is validated by validate_http_url. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + raw_length = response.headers.get("Content-Length") + if raw_length is not None: + try: + declared_length = int(raw_length) + except (TypeError, ValueError): + declared_length = None + if declared_length is not None and declared_length > max_response_bytes: + raise ValueError(f"{label} response exceeds the configured size limit.") + body = response.read(max_response_bytes + 1) + if len(body) > max_response_bytes: + raise ValueError(f"{label} response exceeds the configured size limit.") return HttpFetchResponse( status=int(getattr(response, "status", 0)), headers=dict(response.headers.items()), - body=response.read(), + body=body, )