fix(cli,transport): reject duplicate JSON keys, non-finite constants, and non-string coercion in CLI and HTTP transport (#272)

src/ca2a_runtime/cli.py

@@ -78,13 +78,48 @@ def _cmd_verify_chain(args: argparse.Namespace) -> int:
     return 0
 
 
+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_constant(value: str) -> Any:
+    raise ValueError(f"non-finite JSON constant is not allowed: {value}")
+
+
+def _require_non_empty_str(field: str, value: Any) -> str:
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"{field} must be a non-empty string")
+    return value
+
+
+def _require_optional_str(field: str, value: Any) -> str | None:
+    if value is None:
+        return None
+    return _require_non_empty_str(field, value)
+
+
+def _require_str_frozenset(field: str, value: Any) -> frozenset[str]:
+    if not isinstance(value, list) or not all(isinstance(s, str) and s for s in value):
+        raise ValueError(f"{field} must be a list of non-empty strings")
+    return frozenset(value)
+
+
 def _load_chain(path: str) -> list[DelegationCredential]:
     p = Path(path)
     if not p.is_file():
         raise InvalidCredential(f"chain file not found: {p}")
     try:
-        data: Any = json.loads(p.read_text(encoding="utf-8"))
-    except json.JSONDecodeError as exc:
+        data: Any = json.loads(
+            p.read_text(encoding="utf-8"),
+            object_pairs_hook=_no_duplicate_keys,
+            parse_constant=_no_nonfinite_constant,
+        )
+    except (json.JSONDecodeError, ValueError) as exc:
         raise InvalidCredential(f"invalid JSON in {p}", detail=str(exc)) from exc
     if isinstance(data, dict) and "chain" in data:
         data = data["chain"]
@@ -98,45 +133,52 @@ def _load_records(path: str) -> list[DelegationRecord]:
     if not p.is_file():
         raise ProvenanceLinkBroken(f"dag file not found: {p}")
     try:
-        data: Any = json.loads(p.read_text(encoding="utf-8"))
-    except json.JSONDecodeError as exc:
+        data: Any = json.loads(
+            p.read_text(encoding="utf-8"),
+            object_pairs_hook=_no_duplicate_keys,
+            parse_constant=_no_nonfinite_constant,
+        )
+    except (json.JSONDecodeError, ValueError) as exc:
         raise ProvenanceLinkBroken(f"invalid JSON in {p}", detail=str(exc)) from exc
     if isinstance(data, dict) and "records" in data:
         data = data["records"]
     if not isinstance(data, list):
         raise ProvenanceLinkBroken('dag document must be a list or {"records": [...]}')
     records: list[DelegationRecord] = []
     for item in data:
+        if not isinstance(item, dict):
+            raise ProvenanceLinkBroken(
+                "malformed provenance record", detail="record must be an object"
+            )
         try:
             records.append(
                 DelegationRecord(
-                    record_id=str(item["record_id"]),
-                    credential_id=str(item["credential_id"]),
-                    subject=str(item["subject"]),
-                    scope=frozenset(str(s) for s in item["scope"]),
-                    parent_record_hash=(
-                        None
-                        if item.get("parent_record_hash") is None
-                        else str(item["parent_record_hash"])
+                    record_id=_require_non_empty_str("record_id", item["record_id"]),
+                    credential_id=_require_non_empty_str("credential_id", item["credential_id"]),
+                    subject=_require_non_empty_str("subject", item["subject"]),
+                    scope=_require_str_frozenset("scope", item["scope"]),
+                    parent_record_hash=_require_optional_str(
+                        "parent_record_hash", item.get("parent_record_hash")
                     ),
-                    decision=str(item.get("decision", "allow")),
-                    requested_capability=(
-                        None
-                        if item.get("requested_capability") is None
-                        else str(item["requested_capability"])
+                    decision=_require_non_empty_str("decision", item.get("decision", "allow")),
+                    requested_capability=_require_optional_str(
+                        "requested_capability", item.get("requested_capability")
                     ),
                     effective_scope=(
                         None
                         if item.get("effective_scope") is None
-                        else frozenset(str(s) for s in item["effective_scope"])
+                        else _require_str_frozenset("effective_scope", item.get("effective_scope"))
                     ),
-                    denial_reason=(
-                        None if item.get("denial_reason") is None else str(item["denial_reason"])
+                    denial_reason=_require_optional_str(
+                        "denial_reason", item.get("denial_reason")
                     ),
                     # A record written before this field existed hashes as
                     # "not_offered", which is what its emitter could honestly have
                     # claimed: it never appraised a caller.
-                    caller_attestation=str(item.get("caller_attestation", CALLER_NOT_OFFERED)),
+                    caller_attestation=_require_non_empty_str(
+                        "caller_attestation",
+                        item.get("caller_attestation", CALLER_NOT_OFFERED),
+                    ),
                 )
             )
         except (KeyError, TypeError, ValueError) as exc:
@@ -203,7 +245,11 @@ def _cmd_verify_lineage(args: argparse.Namespace) -> int:
             revocations=_load_revocations(args),
             max_revocation_staleness=args.max_revocation_staleness,
         )
-        records = json.loads(Path(args.dag).read_text(encoding="utf-8"))
+        records = json.loads(
+            Path(args.dag).read_text(encoding="utf-8"),
+            object_pairs_hook=_no_duplicate_keys,
+            parse_constant=_no_nonfinite_constant,
+        )
         if isinstance(records, dict):
             records = records.get("records")
         if not isinstance(records, list) or not all(isinstance(r, dict) for r in records):

src/ca2a_runtime/provenance.py

@@ -70,6 +70,8 @@ class DelegationRecord:
     caller_attestation: str = CALLER_NOT_OFFERED
 
     def __post_init__(self) -> None:
+        if self.decision not in {"allow", "deny"}:
+            raise ValueError(f"decision must be 'allow' or 'deny', got {self.decision!r}")
         if self.caller_attestation not in CALLER_ATTESTATION_VALUES:
             raise ValueError(
                 f"caller_attestation must be one of {sorted(CALLER_ATTESTATION_VALUES)}, "

src/ca2a_runtime/transport/client.py

@@ -82,10 +82,38 @@ def _read_bounded(resp: Any) -> bytes:
     return body
 
 
+def _parse_json_object(raw: bytes) -> dict[str, Any]:
+    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_constant(value: str) -> Any:
+        raise ValueError(f"non-finite JSON constant is not allowed: {value}")
+
+    try:
+        parsed = json.loads(
+            raw.decode("utf-8"),
+            object_pairs_hook=_no_duplicate_keys,
+            parse_constant=_no_nonfinite_constant,
+        )
+    except (ValueError, UnicodeDecodeError, RecursionError) as exc:
+        raise TransportError("peer response is not valid JSON", detail=str(exc)) from exc
+    if not isinstance(parsed, dict):
+        raise TransportError(
+            "peer response must be a JSON object",
+            detail=f"got {type(parsed).__name__}",
+        )
+    return parsed
+
+
 def _get_json(url: str) -> dict[str, Any]:
     _require_http_url(url)
     with urllib.request.urlopen(url, timeout=_TIMEOUT) as resp:  # noqa: S310  # nosec B310
-        return json.loads(_read_bounded(resp))
+        return _parse_json_object(_read_bounded(resp))
 
 
 def _post_json(url: str, body: dict[str, Any]) -> tuple[int, dict[str, Any]]:
@@ -96,9 +124,9 @@ def _post_json(url: str, body: dict[str, Any]) -> tuple[int, dict[str, Any]]:
     )
     try:
         with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:  # noqa: S310  # nosec B310
-            return resp.status, json.loads(_read_bounded(resp))
+            return resp.status, _parse_json_object(_read_bounded(resp))
     except urllib.error.HTTPError as exc:
-        return exc.code, json.loads(_read_bounded(exc))
+        return exc.code, _parse_json_object(_read_bounded(exc))
 
 
 class _NoRedirect(urllib.request.HTTPRedirectHandler):

src/ca2a_runtime/transport/server.py

@@ -34,6 +34,19 @@
 _READ_TIMEOUT_SECONDS = 10.0
 
 
+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_constant(value: str) -> Any:
+    raise ValueError(f"non-finite JSON constant is not allowed: {value}")
+
+
 class PeerHTTPServer(ThreadingHTTPServer):
     """A threading HTTP server carrying the PeerNode its handler serves."""
 
@@ -122,7 +135,15 @@ def do_POST(self) -> None:
             return
         try:
             raw = self.rfile.read(length)
-            message = decode(raw) if path == AUTHENTICATED_TASK_PATH else json.loads(raw)
+            message = (
+                decode(raw)
+                if path == AUTHENTICATED_TASK_PATH
+                else json.loads(
+                    raw.decode("utf-8"),
+                    object_pairs_hook=_no_duplicate_keys,
+                    parse_constant=_no_nonfinite_constant,
+                )
+            )
         # ValueError covers JSONDecodeError and UnicodeDecodeError, and also the
         # integer-digit limit json.loads enforces on a long number literal;
         # RecursionError is deeply nested arrays or objects. Both are reachable

tests/unit/test_lineage_cli.py

@@ -147,3 +147,57 @@ def test_untrusted_root_is_rejected(evidence, capsys):
 def test_malformed_signed_path_fails_closed(evidence, capsys, bad):
     rc, out = run(evidence, capsys, bad)
     assert rc == 1 and out["verified"] is False
+
+
+def test_cli_rejects_duplicate_json_keys_and_non_string_record_fields(evidence, capsys):
+    chain, _, _, native, chain_path, dag_path = evidence
+    # Duplicate key in DAG JSON
+    raw_rec = json.dumps(native[0].body())
+    dup_dag = "[" + raw_rec[:-1] + ', "decision": "allow", "decision": "deny"}]'
+    dag_path.write_text(dup_dag, encoding="utf-8")
+    rc = main(["verify-dag", "--dag", str(dag_path), "--structural-only"])
+    out = json.loads(capsys.readouterr().out)
+    assert rc == 1 and out["code"] == "PROVENANCE_LINK_BROKEN"
+
+    # String scope instead of list (previously split into characters)
+    bad_scope = native[0].body()
+    bad_scope["scope"] = "read"
+    dag_path.write_text(json.dumps([bad_scope]), encoding="utf-8")
+    rc = main(["verify-dag", "--dag", str(dag_path), "--structural-only"])
+    out = json.loads(capsys.readouterr().out)
+    assert rc == 1 and out["code"] == "PROVENANCE_LINK_BROKEN"
+
+    # Non-string record_id (previously coerced via str(...))
+    bad_id = native[0].body()
+    bad_id["record_id"] = 123
+    dag_path.write_text(json.dumps([bad_id]), encoding="utf-8")
+    rc = main(["verify-dag", "--dag", str(dag_path), "--structural-only"])
+    out = json.loads(capsys.readouterr().out)
+    assert rc == 1 and out["code"] == "PROVENANCE_LINK_BROKEN"
+
+    # Invalid decision value
+    bad_dec = native[0].body()
+    bad_dec["decision"] = "maybe"
+    dag_path.write_text(json.dumps([bad_dec]), encoding="utf-8")
+    rc = main(["verify-dag", "--dag", str(dag_path), "--structural-only"])
+    out = json.loads(capsys.readouterr().out)
+    assert rc == 1 and out["code"] == "PROVENANCE_LINK_BROKEN"
+
+    # Duplicate key in chain JSON
+    raw_cred = json.dumps(chain[0].body() | {"signature": chain[0].signature})
+    dup_chain = "[" + raw_cred[:-1] + ', "depth": 0}]'
+    chain_path.write_text(dup_chain, encoding="utf-8")
+    rc = main(
+        [
+            "verify-lineage",
+            "--dag",
+            str(dag_path),
+            "--chain",
+            str(chain_path),
+            "--trusted-root-issuer",
+            chain[0].issuer,
+        ]
+    )
+    out = json.loads(capsys.readouterr().out)
+    assert rc == 1 and out["code"] == "INVALID_CREDENTIAL"
+