fix(examples): reject duplicate JSON keys, non-finite numbers, and boolean sequences in verify_receipts and verify_purchase (#369)

examples/agentic-commerce-accountability/verify_purchase.py

@@ -9,13 +9,40 @@
 from typing import Any
 
 
+_MAX_SAFE_INTEGER = 9_007_199_254_740_991
+
+
+def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    out: dict[str, Any] = {}
+    for k, v in pairs:
+        if k in out:
+            raise ValueError(f"duplicate JSON object key: {k!r}")
+        out[k] = v
+    return out
+
+
+def _no_nonfinite_constant(token: str) -> Any:
+    raise ValueError(f"non-finite JSON constant is not allowed: {token}")
+
+
 def digest(value: Any) -> str:
     encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
     return "sha256:" + hashlib.sha256(encoded).hexdigest()
 
 
 def verify(bundle: dict[str, Any]) -> list[str]:
+    if not isinstance(bundle, dict):
+        return ["bundle must be a JSON object"]
     errors: list[str] = []
+    for section in (
+        "authority_grant",
+        "purchase_request",
+        "policy_decision",
+        "runtime_evidence",
+        "purchase_receipt",
+    ):
+        if not isinstance(bundle.get(section), dict):
+            return [f"{section} must be a JSON object"]
     grant = bundle["authority_grant"]
     request = bundle["purchase_request"]
     decision = bundle["policy_decision"]
@@ -37,9 +64,9 @@ def verify(bundle: dict[str, Any]) -> list[str]:
     ceiling = grant["max_amount_minor"]
     # bool is an int subclass; a negative or zero amount would pass the ceiling,
     # and every amount compares false against a NaN ceiling.
-    if type(amount) is not int or amount <= 0:
+    if type(amount) is not int or amount <= 0 or amount > _MAX_SAFE_INTEGER:
         errors.append("amount is not a positive integer in minor units")
-    elif type(ceiling) is not int:
+    elif type(ceiling) is not int or ceiling <= 0 or ceiling > _MAX_SAFE_INTEGER:
         errors.append("authority grant ceiling is not an integer")
     elif amount > ceiling:
         errors.append("amount exceeds delegated authority")
@@ -64,7 +91,15 @@ def main() -> int:
     if len(sys.argv) != 2:
         print("usage: python verify_purchase.py <bundle.json>", file=sys.stderr)
         return 2
-    bundle = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
+    try:
+        bundle = json.loads(
+            Path(sys.argv[1]).read_text(encoding="utf-8"),
+            object_pairs_hook=_unique_object,
+            parse_constant=_no_nonfinite_constant,
+        )
+    except (OSError, ValueError) as exc:
+        print(f"REJECTED\n- {exc}")
+        return 1
     errors = verify(bundle)
     if errors:
         print("REJECTED")

examples/embodied-action-receipts/tests/test_verify_receipts.py

@@ -113,6 +113,21 @@ def truncate(fixture: dict) -> None:
             {"result": "invalid", "receipt_state": "signature_format"},
         )
 
+    def test_boolean_sequence_and_duplicate_json_keys_rejected(self) -> None:
+        from verify_receipts import verify_text
+
+        self.assertEqual(
+            self.verify_mutated(lambda f: f["receipts"][0].__setitem__("sequence", True)),
+            {"result": "invalid", "receipt_state": "malformed"},
+        )
+        raw = (ROOT / "fixtures" / "valid-chain.json").read_text(encoding="utf-8")
+        dup = raw[:-1] + ', "extra": 1, "extra": 2}'
+        self.assertEqual(
+            verify_text(dup, load_trusted_keys()),
+            {"result": "invalid", "receipt_state": "malformed"},
+        )
+
+
 
 if __name__ == "__main__":
     unittest.main()

examples/embodied-action-receipts/verify_receipts.py

@@ -25,8 +25,31 @@ class TrustedSigner:
     public_key: Ed25519PublicKey
 
 
+def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    out: dict[str, Any] = {}
+    for k, v in pairs:
+        if k in out:
+            raise ValueError(f"duplicate JSON object key: {k!r}")
+        out[k] = v
+    return out
+
+
+def _no_nonfinite_constant(token: str) -> Any:
+    raise ValueError(f"non-finite JSON constant is not allowed: {token}")
+
+
+def _loads_strict(text: str) -> Any:
+    return json.loads(
+        text,
+        object_pairs_hook=_unique_object,
+        parse_constant=_no_nonfinite_constant,
+    )
+
+
 def canonical_bytes(value: Any) -> bytes:
-    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
+    return json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False
+    ).encode()
 
 
 def b64url_decode(value: str) -> bytes:
@@ -61,7 +84,7 @@ def receipt_hash(receipt: dict[str, Any]) -> str:
 
 
 def load_trusted_keys(path: Path = ROOT / "trusted-keys.json") -> dict[str, TrustedSigner]:
-    data = json.loads(path.read_text())
+    data = _loads_strict(path.read_text(encoding="utf-8"))
     keys = {}
     for key_id, key in data["controller_signers"].items():
         keys[key_id] = TrustedSigner(
@@ -78,13 +101,13 @@ def verify_fixture(
 ) -> dict[str, Any]:
     if trusted_keys is None:
         trusted_keys = load_trusted_keys()
-    return verify_text(path.read_text(), trusted_keys)
+    return verify_text(path.read_text(encoding="utf-8"), trusted_keys)
 
 
 def verify_text(text: str, trusted_keys: dict[str, TrustedSigner]) -> dict[str, Any]:
     """Verify fixture JSON. Always returns a verdict; never raises on bad input."""
     try:
-        return verify_document(json.loads(text), trusted_keys)
+        return verify_document(_loads_strict(text), trusted_keys)
     except (KeyError, TypeError, AttributeError, ValueError, RecursionError):
         # The fixture is untrusted input: a wrong shape is a verdict, not a crash.
         return dict(MALFORMED)
@@ -109,6 +132,10 @@ def verify_document(
     final_state = "absent"
     final_verdict = None
 
+    for receipt in receipts:
+        if not isinstance(receipt, dict) or type(receipt["sequence"]) is not int:
+            raise TypeError("receipt sequence must be an integer")
+
     for expected_sequence, receipt in enumerate(
         sorted(receipts, key=lambda r: r["sequence"]), start=1
     ):