RFC 8785 integer domain fix and unsafe-integer write policy

src/agent_passport/agora.py

@@ -13,7 +13,7 @@
 from typing import Any, Optional
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 
 def _msg_id() -> str:
@@ -73,7 +73,7 @@ def create_agora_message(
     if reply_to:
         message_content["replyTo"] = reply_to
 
-    canonical = canonicalize(message_content)
+    canonical = canonicalize_for_write(message_content)
     signature = sign(canonical, private_key)
 
     return {**message_content, "signature": signature}

src/agent_passport/canonical.py

@@ -12,6 +12,12 @@
 import math
 import re
 
+from .write_policy import (
+    MAX_SAFE_INTEGER,
+    UnsafeIntegerError,
+    assert_write_safe_numbers,
+)
+
 
 class JCSCanonicalizationError(ValueError):
     """A value cannot be canonicalized under RFC 8785.
@@ -163,6 +169,68 @@ def canonicalize(obj) -> str:
     return json.dumps(obj, ensure_ascii=False)
 
 
+def canonicalize_for_write(obj, path: str = "$") -> str:
+    """Legacy canonicalization for a NEW WRITE, with the APS unsafe-integer rule applied.
+
+    Byte-identical to :func:`canonicalize` for every value it accepts: same key order,
+    same null stripping, same number and string formatting. The only difference is that
+    an integer-valued number outside the interoperable IEEE 754 range is refused rather
+    than emitted. See :mod:`agent_passport.write_policy` for the rule.
+
+    Use at signing and new-write boundaries ONLY. Verification, recompute, and any path
+    rebuilding the preimage of an existing artifact must keep calling
+    :func:`canonicalize`, which stays unrestricted so historical bytes keep verifying.
+
+    READS EACH KEY EXACTLY ONCE. A separate validating pre-pass followed by
+    :func:`canonicalize` would walk the mapping twice, and a Mapping subclass or an
+    object overriding ``__getitem__`` can answer differently on the second read, so the
+    value checked would not be the value signed. Here the value is captured once into
+    ``val``, checked, and emitted from that same capture.
+
+    Note this deliberately does NOT fix the legacy int-verbatim divergence from the other
+    SDKs: refusing the unsafe range makes that divergence unreachable on write without
+    altering a single historical byte.
+    """
+    if obj is None:
+        return "null"
+    if isinstance(obj, bool):
+        return "true" if obj else "false"
+    if isinstance(obj, int):
+        if abs(obj) > MAX_SAFE_INTEGER:
+            raise UnsafeIntegerError(f"{path}: integer exceeds the interoperable IEEE 754 range")
+        return json.dumps(obj)
+    if isinstance(obj, float):
+        if math.isnan(obj) or math.isinf(obj):
+            raise ValueError(f"Cannot canonicalize {obj}: NaN and Infinity are not valid JSON per RFC 8259")
+        if obj.is_integer() and abs(obj) > MAX_SAFE_INTEGER:
+            raise UnsafeIntegerError(f"{path}: integer exceeds the interoperable IEEE 754 range")
+        return _es_number(obj)
+    if isinstance(obj, str):
+        return json.dumps(obj, ensure_ascii=False)
+    if isinstance(obj, list):
+        return "[" + ",".join(
+            canonicalize_for_write(item, f"{path}[{index}]") for index, item in enumerate(obj)
+        ) + "]"
+    if isinstance(obj, dict):
+        pairs = []
+        for key in _canonical_keys(obj.keys()):
+            val = obj[key]  # the single read; everything below uses this capture
+            if val is None:
+                continue
+            pairs.append(
+                json.dumps(key, ensure_ascii=False)
+                + ":"
+                + canonicalize_for_write(val, f"{path}.{key}")
+            )
+        return "{" + ",".join(pairs) + "}"
+    # Anything that reaches here (a tuple, in practice) is emitted by the same json.dumps
+    # fallback the READ twin uses, so the bytes stay identical. Validate it first, or an
+    # out-of-range integer inside a tuple would be signed unchecked. A second read is safe
+    # here precisely because the values that reach this branch are immutable.
+    assert_write_safe_numbers(obj, path)
+    return json.dumps(obj, ensure_ascii=False)
+
+
 def canonicalize_jcs(obj) -> str:
     """RFC 8785 JSON Canonicalization Scheme (strict).
 
@@ -179,11 +247,31 @@ def canonicalize_jcs(obj) -> str:
     if obj is None:
         return "null"
     if isinstance(obj, bool):
+        # Must stay ahead of the int branch: bool is a subclass of int in Python,
+        # and reordering would serialize True as 1.
         return "true" if obj else "false"
     if isinstance(obj, int):
-        return json.dumps(obj)
+        # RFC 8785 section 3.2.2.3 defines the JCS number domain as IEEE-754
+        # binary64 serialized under ECMAScript Number::toString. Python's int is
+        # arbitrary precision, so emitting it verbatim preserves a decimal
+        # spelling the double does not have: 2**60 would serialize as
+        # 1152921504606846976 where the double is 1152921504606847000. Widen to
+        # binary64 first, then take exactly the same path a float takes.
+        try:
+            as_double = float(obj)
+        except OverflowError:
+            as_double = math.inf
+        if math.isinf(as_double):
+            raise JCSCanonicalizationError(
+                f"canonicalize_jcs: integer {obj} exceeds the IEEE-754 binary64 "
+                "range and has no representation in the RFC 8785 number domain",
+                reason="number_out_of_double_range",
+            )
+        return _es_number(as_double)
     if isinstance(obj, float):
-        import math
+        # math is imported at module scope. A local "import math" here would make
+        # the name function-local for the whole of canonicalize_jcs and shadow it
+        # from the int branch above.
         if math.isnan(obj) or math.isinf(obj):
             raise ValueError(f"Cannot canonicalize {obj}")
         return _es_number(obj)
@@ -206,3 +294,57 @@ def canonicalize_jcs(obj) -> str:
             )
         return "{" + ",".join(pairs) + "}"
     return json.dumps(obj, ensure_ascii=False)
+
+
+def canonicalize_jcs_for_write(obj, path: str = "$") -> str:
+    """RFC 8785 canonicalization for a NEW WRITE, with the unsafe-integer rule applied.
+
+    Byte-identical to :func:`canonicalize_jcs` for every value it accepts. The only
+    difference is that an integer-valued number outside the interoperable IEEE 754 range
+    is refused rather than emitted.
+
+    Use at signing and new-write boundaries ONLY. Verification and recompute keep calling
+    :func:`canonicalize_jcs`, which stays unrestricted so historical bytes keep verifying.
+
+    READS EACH KEY EXACTLY ONCE, so a Mapping subclass or an object overriding
+    ``__getitem__`` cannot answer safe on a validating pre-pass and unsafe on the
+    emitting pass. The value is captured once, checked, and emitted from that capture.
+    """
+    if obj is None:
+        return "null"
+    if isinstance(obj, bool):
+        return "true" if obj else "false"
+    if isinstance(obj, int):
+        if abs(obj) > MAX_SAFE_INTEGER:
+            raise UnsafeIntegerError(f"{path}: integer exceeds the interoperable IEEE 754 range")
+        return _es_number(float(obj))
+    if isinstance(obj, float):
+        if math.isnan(obj) or math.isinf(obj):
+            raise ValueError(f"Cannot canonicalize {obj}")
+        if obj.is_integer() and abs(obj) > MAX_SAFE_INTEGER:
+            raise UnsafeIntegerError(f"{path}: integer exceeds the interoperable IEEE 754 range")
+        return _es_number(obj)
+    if isinstance(obj, str):
+        _assert_no_lone_surrogate(obj)
+        return json.dumps(obj, ensure_ascii=False)
+    if isinstance(obj, list):
+        return "[" + ",".join(
+            canonicalize_jcs_for_write(item, f"{path}[{index}]") for index, item in enumerate(obj)
+        ) + "]"
+    if isinstance(obj, dict):
+        for key in obj.keys():
+            _assert_no_lone_surrogate(key)
+        pairs = []
+        for key in _canonical_keys(obj.keys()):
+            val = obj[key]  # the single read; everything below uses this capture
+            pairs.append(
+                json.dumps(key, ensure_ascii=False)
+                + ":"
+                + canonicalize_jcs_for_write(val, f"{path}.{key}")
+            )
+        return "{" + ",".join(pairs) + "}"
+    # Same reasoning as canonicalize_for_write above: emit through the shared fallback so
+    # the bytes match the read twin, but validate first so a tuple cannot smuggle an
+    # out-of-range integer past the rule.
+    assert_write_safe_numbers(obj, path)
+    return json.dumps(obj, ensure_ascii=False)

src/agent_passport/commerce.py

@@ -14,7 +14,7 @@
 from typing import Any, Optional
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 from .passport import verify_passport
 
 
@@ -269,7 +269,7 @@ def sign_commerce_receipt(
         "delegationChain": delegation_chain,
         "beneficiary": beneficiary,
     }
-    payload = canonicalize(receipt)
+    payload = canonicalize_for_write(receipt)
     sig = sign(payload, private_key)
     return {**receipt, "signature": sig}
 

src/agent_passport/coordination.py

@@ -14,7 +14,7 @@
 from typing import Any, Optional
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 
 def _rand_hex(n: int = 4) -> str:
@@ -60,7 +60,7 @@ def create_task_brief(
         "status": "draft",
     }
 
-    signature = sign(canonicalize(brief), operator_private_key)
+    signature = sign(canonicalize_for_write(brief), operator_private_key)
     return {**brief, "signature": signature}
 
 
@@ -127,7 +127,7 @@ def assign_task(
         "assignedAt": datetime.now(timezone.utc).isoformat(),
     }
 
-    op_sig = sign(canonicalize(assignment_content), operator_private_key)
+    op_sig = sign(canonicalize_for_write(assignment_content), operator_private_key)
     assignment = {**assignment_content, "operatorSignature": op_sig}
 
     updated_roles = [
@@ -138,7 +138,7 @@ def assign_task(
     brief_content = {k: v for k, v in brief.items() if k != "signature"}
     brief_content["roles"] = updated_roles
     brief_content["status"] = "assigned" if all_assigned else "draft"
-    new_sig = sign(canonicalize(brief_content), operator_private_key)
+    new_sig = sign(canonicalize_for_write(brief_content), operator_private_key)
 
     return {"assignment": assignment, "updatedBrief": {**brief_content, "signature": new_sig}}
 
@@ -147,7 +147,7 @@ def accept_task(assignment: dict, agent_private_key: str) -> dict:
     """Agent accepts a task assignment."""
     accepted_at = datetime.now(timezone.utc).isoformat()
     to_sign = {"assignmentId": assignment["assignmentId"], "taskId": assignment["taskId"], "acceptedAt": accepted_at}
-    agent_sig = sign(canonicalize(to_sign), agent_private_key)
+    agent_sig = sign(canonicalize_for_write(to_sign), agent_private_key)
     return {**assignment, "acceptedAt": accepted_at, "agentSignature": agent_sig}
 
 
@@ -188,7 +188,7 @@ def submit_evidence(
         },
     }
 
-    signature = sign(canonicalize(content), submitter_private_key)
+    signature = sign(canonicalize_for_write(content), submitter_private_key)
     return {**content, "signature": signature}
 
 
@@ -245,7 +245,7 @@ def review_evidence(
         "rationale": rationale,
         "issues": issues,
     }
-    signature = sign(canonicalize(content), reviewer_private_key)
+    signature = sign(canonicalize_for_write(content), reviewer_private_key)
     return {**content, "signature": signature}
 
 
@@ -298,7 +298,7 @@ def handoff_evidence(
         "toAgent": to_agent_public_key,
         "handoffAt": datetime.now(timezone.utc).isoformat(),
     }
-    op_sig = sign(canonicalize(content), operator_private_key)
+    op_sig = sign(canonicalize_for_write(content), operator_private_key)
     return {**content, "operatorSignature": op_sig}
 
 
@@ -343,7 +343,7 @@ def submit_deliverable(
         "citationCount": citation_count,
         "gapsFlagged": gaps_flagged,
     }
-    signature = sign(canonicalize(deliv), submitter_private_key)
+    signature = sign(canonicalize_for_write(deliv), submitter_private_key)
     return {**deliv, "signature": signature}
 
 
@@ -409,7 +409,7 @@ def complete_task(
         "metrics": metrics,
         "retrospective": retrospective,
     }
-    signature = sign(canonicalize(completion), operator_private_key)
+    signature = sign(canonicalize_for_write(completion), operator_private_key)
     return {**completion, "signature": signature}
 
 

src/agent_passport/data_settlement.py

@@ -17,7 +17,7 @@
 from typing import Any
 
 from .crypto import sign, verify
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 
 
 def _merkle_root(items: list[str]) -> str:
@@ -74,7 +74,7 @@ def generate_settlement(
         "receiptCount": len(all_receipt_ids),
         "merkleRoot": _merkle_root(all_receipt_ids),
     }
-    record["signature"] = sign(canonicalize(record), generator_private_key)
+    record["signature"] = sign(canonicalize_for_write(record), generator_private_key)
     return record
 
 

src/agent_passport/data_source.py

@@ -16,7 +16,7 @@
 from typing import Any
 
 from .crypto import sign, verify, generate_key_pair, public_key_from_private
-from .canonical import canonicalize
+from .canonical import canonicalize, canonicalize_for_write
 from .attribution import build_merkle_root, get_merkle_proof, verify_merkle_proof
 
 
@@ -57,7 +57,7 @@ def register_self_attested_source(
         "ownerPublicKey": owner_public_key,
         "registeredAt": _now_iso(),
     }
-    receipt["signature"] = sign(canonicalize(receipt), owner_private_key)
+    receipt["signature"] = sign(canonicalize_for_write(receipt), owner_private_key)
     return receipt
 
 
@@ -76,7 +76,7 @@ def register_custodian_attested_source(
         "custodianPublicKey": custodian_public_key,
         "registeredAt": _now_iso(),
     }
-    receipt["signature"] = sign(canonicalize(receipt), custodian_private_key)
+    receipt["signature"] = sign(canonicalize_for_write(receipt), custodian_private_key)
     return receipt
 
 
@@ -95,7 +95,7 @@ def register_gateway_observed_source(
         "gatewayPublicKey": gateway_public_key,
         "registeredAt": _now_iso(),
     }
-    receipt["signature"] = sign(canonicalize(receipt), gateway_private_key)
+    receipt["signature"] = sign(canonicalize_for_write(receipt), gateway_private_key)
     return receipt
 
 
@@ -166,7 +166,7 @@ def record_data_access(
         "accessedAt": _now_iso(),
         "termsAtAccessTime": terms_snapshot,
     }
-    receipt["signature"] = sign(canonicalize(receipt), gateway_private_key)
+    receipt["signature"] = sign(canonicalize_for_write(receipt), gateway_private_key)
     return receipt