fix(assessment): reject ambiguous trust keyrings

docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md

@@ -356,9 +356,11 @@ Use `--catalog CATALOG --keyring PUBLISHED_OR_CANDIDATE_KEYRING
 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.
+windows. Only `active` and `next` entries are trusted; unknown statuses,
+duplicate key IDs and malformed structured keyrings fail closed, while revoked,
+disabled and retired entries are ignored. 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

tests/test_capability_fit_review.py

@@ -272,7 +272,7 @@ class CapabilityFitReviewTests(unittest.TestCase):
         )
         self.assertTrue(report["proof_scope"]["published_keyring_hash"]["valid"])
 
-    def test_locally_pinned_rotation_keys_follow_core_status_semantics(self) -> None:
+    def test_locally_pinned_rotation_keys_fail_closed(self) -> None:
         keys = trusted_keys_from_keyring(
             {
                 "keys": [
@@ -288,6 +288,38 @@ class CapabilityFitReviewTests(unittest.TestCase):
             {"legacy": "base64-key"},
             trusted_keys_from_keyring({"legacy": "base64-key"}),
         )
+        self.assertEqual(
+            {},
+            trusted_keys_from_keyring(
+                {
+                    "keys": [
+                        {
+                            "key_id": "typo",
+                            "status": "retierd",
+                            "public_key": "unsafe",
+                        }
+                    ]
+                }
+            ),
+        )
+        self.assertEqual(
+            {},
+            trusted_keys_from_keyring(
+                {
+                    "keys": [
+                        {"key_id": "duplicate", "public_key": "a"},
+                        {"key_id": "duplicate", "public_key": "b"},
+                    ]
+                }
+            ),
+        )
+        self.assertEqual({}, trusted_keys_from_keyring({"keys": "invalid"}))
+        self.assertEqual(
+            {},
+            trusted_keys_from_keyring(
+                {"keys": [{"key_id": " ", "public_key": "unsafe"}]}
+            ),
+        )
 
     def test_schema_error_blocks_before_comparison(self) -> None:
         assessment = deepcopy(self.assessment)

tools/assessments/govoplan_assessment/capability_fit.py

@@ -415,40 +415,57 @@ def validate_catalog(
 def trusted_keys_from_keyring(payload: dict[str, Any]) -> dict[str, str]:
     now = datetime.now(tz=UTC)
     result: dict[str, str] = {}
+    if "keys" not in payload:
+        if not all(
+            isinstance(key, str)
+            and key.strip()
+            and key == key.strip()
+            and isinstance(value, str)
+            and value.strip()
+            and value == value.strip()
+            for key, value in payload.items()
+        ):
+            return {}
+        return {key: value for key, value in payload.items()}
     keys = payload.get("keys")
     if not isinstance(keys, list):
-        return {
-            str(key): str(value)
-            for key, value in payload.items()
-            if str(key).strip() and str(value).strip()
-        }
+        return {}
+    seen_key_ids: set[str] = set()
     for item in keys:
         if not isinstance(item, dict):
-            continue
-        if str(item.get("status") or "active").strip().lower() in {
-            "revoked",
-            "disabled",
-            "retired",
-        }:
+            return {}
+        raw_key_id = _text(item.get("key_id"))
+        if (
+            raw_key_id is None
+            or raw_key_id != raw_key_id.strip()
+            or raw_key_id in seen_key_ids
+        ):
+            return {}
+        key_id = raw_key_id
+        seen_key_ids.add(key_id)
+        status = str(item.get("status") or "active").strip().lower()
+        if status not in {"active", "next", "revoked", "disabled", "retired"}:
+            return {}
+        if status in {"revoked", "disabled", "retired"}:
             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
+            return {}
         if raw_not_after is not None and not_after is None:
-            continue
+            return {}
         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
+        if public_key is None:
+            return {}
+        result[key_id] = public_key
     return result