fix(provenance): reject duplicate JSON keys, private JWKs, and non-dict records in check_server_provenance and validate required_kind in meets() (#813)

src/cmcp_runtime/provenance.py

@@ -59,6 +59,7 @@ class ProvenanceOutcome(StrEnum):
 
 #: Ordered weakest to strongest, for `required_kind` comparisons.
 _KIND_ORDER = ("publisher-asserted", "observer-attested", "tee-attested")
+_PRIVATE_JWK_PARAMS = frozenset({"d", "p", "q", "dp", "dq", "qi", "oth", "k"})
 
 
 @dataclass(frozen=True)
@@ -77,17 +78,23 @@ def meets(self, required_kind: str | None) -> bool:
         """Whether this satisfies a configured minimum assurance."""
         if required_kind is None:
             return True
-        if self.outcome is not ProvenanceOutcome.VERIFIED or self.kind is None:
+        if not isinstance(required_kind, str) or required_kind not in _KIND_ORDER:
+            raise ValueError(f"unknown provenance required_kind: {required_kind!r}")
+        if (
+            self.outcome is not ProvenanceOutcome.VERIFIED
+            or not isinstance(self.kind, str)
+            or self.kind not in _KIND_ORDER
+        ):
             return False
         return _KIND_ORDER.index(self.kind) >= _KIND_ORDER.index(required_kind)
 
     def to_audit(self) -> dict[str, str]:
         block = {"provenance": self.outcome.value}
-        if self.kind:
+        if isinstance(self.kind, str) and self.kind:
             block["provenance_kind"] = self.kind
-        if self.publisher:
+        if isinstance(self.publisher, str) and self.publisher:
             block["provenance_publisher"] = self.publisher[:256]
-        if self.detail:
+        if isinstance(self.detail, str) and self.detail:
             block["provenance_detail"] = self.detail[:512]
         return block
 
@@ -123,12 +130,31 @@ def check_server_provenance(
             detail="agentrust-trace is not installed; provenance was not checked",
         )
 
+    def _no_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+        out: dict[str, Any] = {}
+        for key, value in pairs:
+            if key in out:
+                raise ValueError(f"duplicate JSON object key: {key!r}")
+            out[key] = value
+        return out
+
+    def _no_nonfinite(token: str) -> Any:
+        raise ValueError(f"non-finite JSON constant: {token}")
+
     try:
-        record = json.loads(pathlib.Path(record_path).read_text(encoding="utf-8"))
-    except (OSError, ValueError) as exc:
+        record = json.loads(
+            pathlib.Path(record_path).read_text(encoding="utf-8"),
+            object_pairs_hook=_no_duplicate_keys,
+            parse_constant=_no_nonfinite,
+        )
+    except (OSError, ValueError, RecursionError) as exc:
         return ProvenanceResult(
             ProvenanceOutcome.INVALID, detail=f"could not read the record: {exc}"
         )
+    if not isinstance(record, dict):
+        return ProvenanceResult(
+            ProvenanceOutcome.INVALID, detail="provenance record must be a JSON object"
+        )
 
     if trusted_jwk is None:
         # The record embeds a key. Using it would verify the document against
@@ -140,9 +166,16 @@ def check_server_provenance(
                 "the record cannot be used to check the record"
             ),
         )
+    if not isinstance(trusted_jwk, dict) or _PRIVATE_JWK_PARAMS.intersection(trusted_jwk):
+        return ProvenanceResult(
+            ProvenanceOutcome.INVALID,
+            detail="trusted_jwk must be a public JWK object without private key parameters",
+        )
 
-    kind = record.get("kind")
-    publisher = record.get("publisher")
+    raw_kind = record.get("kind")
+    raw_publisher = record.get("publisher")
+    kind = raw_kind if isinstance(raw_kind, str) else None
+    publisher = raw_publisher if isinstance(raw_publisher, str) else None
 
     try:
         verify_record(record, trusted_jwk)

tests/unit/test_provenance_consumption.py

@@ -158,3 +158,38 @@ def test_audit_detail_is_bounded(signed_record) -> None:
     path, _ = signed_record
     block = check_server_provenance(path, None, TOOLS).to_audit()
     assert len(block.get("provenance_detail", "")) <= 512
+
+
+@pytest.mark.parametrize(
+    "raw_json",
+    [
+        "[1, 2]",
+        '"just-a-string"',
+        '{"kind": "publisher-asserted", "kind": "tee-attested"}',
+        '{"kind": "publisher-asserted", "x": NaN}',
+        '{"kind": 123, "publisher": 456}',
+    ],
+)
+def test_check_server_provenance_rejects_non_dict_duplicate_keys_and_non_str_fields(
+    tmp_path, signed_record, raw_json: str
+) -> None:
+    _, jwk = signed_record
+    p = tmp_path / "bad_prov.json"
+    p.write_text(raw_json, encoding="utf-8")
+    res = check_server_provenance(str(p), jwk, TOOLS)
+    assert res.outcome is ProvenanceOutcome.INVALID
+    assert res.to_audit()["provenance"] == "invalid"
+
+
+def test_check_server_provenance_rejects_private_jwk_and_meets_validates_required_kind(
+    signed_record,
+) -> None:
+    path, jwk = signed_record
+    for param in ("d", "k"):
+        res = check_server_provenance(path, {**jwk, param: "secret"}, TOOLS)
+        assert res.outcome is ProvenanceOutcome.INVALID
+    with pytest.raises(ValueError, match="unknown provenance required_kind"):
+        ProvenanceResult(ProvenanceOutcome.ABSENT).meets("unknown-kind")
+    assert ProvenanceResult(ProvenanceOutcome.VERIFIED, kind="unknown-kind").meets(
+        "publisher-asserted"
+    ) is False