conformance(inclusion): enforce registry-anchor-v1 surrogate/key/cycle rules and validate leaf_index/entry fields (#625)

conformance/src/trace_tests/inclusion.py

@@ -44,26 +44,50 @@ class InclusionError(ValueError):
 def _outside_profile(value: Any) -> str | None:
     """Name the first value registry-anchor-v1 section 1 excludes, or None.
 
-    Non-integer numbers (NaN and the infinities among them) and integers outside
-    -(2**53 - 1) to 2**53 - 1 are outside the anchor-leaf profile: two
-    implementations of the four serialization rules write different bytes for
-    them, so a proof over them only proves what Python's ``json.dumps`` wrote.
-    Iterative, because the claim is attacker-controlled and may be deep.
+    Non-integer numbers (NaN and the infinities among them), integers outside
+    -(2**53 - 1) to 2**53 - 1, and unpaired UTF-16 surrogates (U+D800..U+DFFF)
+    are outside the anchor-leaf profile: two implementations of the serialization
+    rules write different bytes for them, so a proof over them only proves what
+    Python's ``json.dumps`` wrote. Iterative, because the claim is
+    attacker-controlled and may be deep.
     """
-    stack = [value]
+    active: set[int] = set()
+    stack: list[tuple[Any, bool]] = [(value, False)]
     while stack:
-        v = stack.pop()
-        if isinstance(v, bool) or v is None or isinstance(v, str):
+        v, exiting = stack.pop()
+        if exiting:
+            active.discard(id(v))
+            continue
+        if isinstance(v, bool) or v is None:
+            continue
+        if isinstance(v, str):
+            if any(0xD800 <= ord(ch) <= 0xDFFF for ch in v):
+                return "unpaired UTF-16 surrogate"
             continue
         if isinstance(v, int):
             if not -_SAFE_INT <= v <= _SAFE_INT:
                 return f"integer {v} is outside the safe-integer range"
         elif isinstance(v, float):
             return f"non-integer number {v!r}"
         elif isinstance(v, dict):
-            stack.extend(v.values())
+            obj_id = id(v)
+            if obj_id in active:
+                return "circular reference"
+            for k in v:
+                if not isinstance(k, str):
+                    return f"non-string object key of type {type(k).__name__}"
+                if any(0xD800 <= ord(ch) <= 0xDFFF for ch in k):
+                    return "unpaired UTF-16 surrogate in object key"
+            active.add(obj_id)
+            stack.append((v, True))
+            stack.extend((item, False) for item in v.values())
         elif isinstance(v, (list, tuple)):
-            stack.extend(v)
+            obj_id = id(v)
+            if obj_id in active:
+                return "circular reference"
+            active.add(obj_id)
+            stack.append((v, True))
+            stack.extend((item, False) for item in v)
     return None
 
 
@@ -157,14 +181,26 @@ def parse_proof(receipt: object) -> tuple[int, list[bytes], int | None, bytes |
     if not isinstance(raw_path, list):
         raise InclusionError(f"audit_path must be an array, got {type(raw_path).__name__}")
 
+    leaf_index = receipt["leaf_index"]
+    if type(leaf_index) is not int or not 0 <= leaf_index <= _SAFE_INT:
+        raise InclusionError(
+            f"receipt leaf_index must be a non-negative integer, got {leaf_index!r}"
+        )
+
     leaf_count = receipt.get("leaf_count")
-    if "leaf_count" in receipt and (type(leaf_count) is not int):
+    if "leaf_count" in receipt and (
+        type(leaf_count) is not int or not 1 <= leaf_count <= _SAFE_INT
+    ):
+        raise InclusionError(
+            f"receipt leaf_count must be an integer >= 1, got {leaf_count!r}"
+        )
+    if leaf_count is not None and leaf_index >= leaf_count:
         raise InclusionError(
-            f"receipt leaf_count must be an integer, got {type(leaf_count).__name__}"
+            f"receipt leaf_index {leaf_index} must be less than leaf_count {leaf_count}"
         )
 
     return (
-        receipt["leaf_index"],
+        leaf_index,
         [decode_hash(node) for node in raw_path],
         leaf_count,
         decode_hash(receipt["merkle_root"]) if "merkle_root" in receipt else None,
@@ -226,14 +262,22 @@ def parse_entry(entry: object) -> tuple[bytes, int, str, str]:
         if not isinstance(entry[key], str) or not entry[key]:
             raise InclusionError(f"registry entry {key} must be a non-empty string")
 
+    for key in ("producer", "batch_id"):
+        val = entry[key]
+        if not val.strip() or any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in val):
+            raise InclusionError(
+                f"registry entry {key} must be a non-blank string without "
+                "ASCII control characters"
+            )
+
     if not _is_utc_timestamp(entry["ts"]):
         raise InclusionError(
             "registry entry ts must be an ISO-8601 UTC timestamp ending in Z, "
             f"got {entry['ts'][:40]!r}"
         )
 
     leaf_count = entry["leaf_count"]
-    if not isinstance(leaf_count, int) or isinstance(leaf_count, bool) or leaf_count < 1:
+    if type(leaf_count) is not int or not 1 <= leaf_count <= _SAFE_INT:
         raise InclusionError(
             f"registry entry leaf_count must be an integer >= 1, got {leaf_count!r}"
         )

conformance/tests/unit/test_tr_anc.py

@@ -151,3 +151,76 @@ def test_tuple_outside_anchor_profile_raises_inclusion_error_and_fails_tr_anc_00
     assert by_code["TR-ANC-002"].status is Status.FAIL
     assert expected in by_code["TR-ANC-002"].message
 
+
+@pytest.mark.parametrize(
+    ("claim", "expected"),
+    [
+        ({"s": "\ud800"}, "surrogate"),
+        ({"nested": ("\udfff",)}, "surrogate"),
+        ({"\ud800key": 1}, "surrogate"),
+        ({1: "val"}, "non-string object key"),  # type: ignore[dict-item]
+    ],
+    ids=["surrogate-value", "surrogate-tuple", "surrogate-key", "int-key"],
+)
+def test_surrogates_and_non_string_keys_raise_inclusion_error(
+    claim: dict[str, Any], expected: str
+):
+    with pytest.raises(InclusionError, match=expected):
+        canonical_claim_bytes(claim)
+
+
+def test_cyclic_claim_raises_inclusion_error_while_shared_acyclic_succeeds():
+    shared = {"k": 1}
+    assert canonical_claim_bytes({"a": shared, "b": shared}) == b'{"a":{"k":1},"b":{"k":1}}'
+
+    cyc_dict: dict[str, Any] = {}
+    cyc_dict["self"] = cyc_dict
+    with pytest.raises(InclusionError, match="circular reference"):
+        canonical_claim_bytes(cyc_dict)
+
+    cyc_list: list[Any] = []
+    cyc_list.append(cyc_list)
+    with pytest.raises(InclusionError, match="circular reference"):
+        canonical_claim_bytes({"lst": cyc_list})
+
+
+@pytest.mark.parametrize(
+    "bad_receipt",
+    [
+        {**_RECEIPT, "leaf_index": True},
+        {**_RECEIPT, "leaf_index": "0"},
+        {**_RECEIPT, "leaf_index": -1},
+        {**_RECEIPT, "leaf_index": 2**53},
+        {**_RECEIPT, "leaf_count": 0},
+        {**_RECEIPT, "leaf_count": 2**53},
+        {**_RECEIPT, "leaf_index": 1, "leaf_count": 1},
+    ],
+)
+def test_parse_proof_and_receipt_reject_invalid_leaf_index_or_count(
+    bad_receipt: dict[str, Any]
+):
+    from trace_tests.inclusion import parse_proof, parse_receipt
+
+    with pytest.raises(InclusionError):
+        parse_proof(bad_receipt)
+    with pytest.raises(InclusionError):
+        parse_receipt(bad_receipt)
+
+
+@pytest.mark.parametrize(
+    "bad_entry",
+    [
+        {**_ENTRY, "producer": "   "},
+        {**_ENTRY, "producer": "prod\n1.0"},
+        {**_ENTRY, "batch_id": "batch\x001"},
+        {**_ENTRY, "leaf_count": 2**53},
+    ],
+)
+def test_parse_entry_rejects_whitespace_or_control_chars_and_unsafe_leaf_count(
+    bad_entry: dict[str, Any]
+):
+    from trace_tests.inclusion import parse_entry
+
+    assert parse_entry({**_ENTRY, "producer": "Example Gateway 1.0"})[2] == "Example Gateway 1.0"
+    with pytest.raises(InclusionError):
+        parse_entry(bad_entry)

spec/registry-anchor-v1.md

@@ -56,7 +56,7 @@ In Python this is exactly:
 json.dumps(claim, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("ascii")
 ```
 
-Claims MUST be JSON objects. TRACE Trust Records contain only strings, integers, booleans, nulls, arrays, and objects. Claims containing non-integer numbers are outside this profile, because cross-language float serialization is not canonical, and claims containing an integer outside -9007199254740991 to 9007199254740991 are outside it for the same reason: an implementation of the four rules above in a language whose only number type is the IEEE 754 double writes one value for two distinct integers. TRACE v0.2 §3.2.2 already holds a Trust Record to that range, so a schema-valid record cannot reach this; a claim that is not a Trust Record can, and this is where it is excluded.
+Claims MUST be JSON objects. TRACE Trust Records contain only strings, integers, booleans, nulls, arrays, and objects. Claims containing non-integer numbers are outside this profile, because cross-language float serialization is not canonical, and claims containing an integer outside -9007199254740991 to 9007199254740991 are outside it for the same reason: an implementation of the four rules above in a language whose only number type is the IEEE 754 double writes one value for two distinct integers. Claims containing an unpaired UTF-16 surrogate (`U+D800` to `U+DFFF`) in a string or object key are also outside this profile, because languages whose string type is UTF-8 cannot represent an unpaired surrogate. TRACE v0.2 §3.2.2 already holds a Trust Record to that range, so a schema-valid record cannot reach this; a claim that is not a Trust Record can, and this is where it is excluded.
 
 ## 2. Leaf hash
 
@@ -89,9 +89,9 @@ An anchor is recorded as one JSON object carrying these required fields:
 |---|---|---|
 | `ts` | string | Anchoring time, ISO-8601 UTC with `Z` suffix |
 | `merkle_root` | string | `sha256:<hex>` root of the batch tree (§3) |
-| `leaf_count` | integer | Number of leaves in the batch, `>= 1` |
-| `producer` | string | Identifier of the party that produced and submitted the batch |
-| `batch_id` | string | Producer-scoped unique identifier for the batch |
+| `leaf_count` | integer | Number of leaves in the batch, `1 <= leaf_count <= 9007199254740991` (`2^53 - 1`) |
+| `producer` | string | Non-blank identifier of the party that produced and submitted the batch (no ASCII control characters `U+0000` to `U+001F`, `U+007F`) |
+| `batch_id` | string | Non-blank producer-scoped unique identifier for the batch (no ASCII control characters `U+0000` to `U+001F`, `U+007F`) |
 
 ```json
 {"ts":"2026-06-12T18:30:00Z","merkle_root":"sha256:9c4f...","leaf_count":1,"producer":"cmcp-gateway/0.1.0","batch_id":"2026-06-12-001"}
@@ -111,7 +111,7 @@ A producer gives each claim holder an inclusion proof:
 {"leaf_index": 0, "audit_path": ["sha256:<hex>", "sha256:<hex>"]}
 ```
 
-- `leaf_index`: 0-based position of the claim's leaf in the batch.
+- `leaf_index`: 0-based position of the claim's leaf in the batch (`0 <= leaf_index <= 9007199254740991`).
 - `audit_path`: the RFC 6962 audit path, ordered leaf-to-root: at each tree level, the sibling hash needed to recompute the parent. A level where the node was promoted, and therefore had no sibling, contributes no path element. A single-leaf batch has an empty `audit_path`.
 
 ### 5.1 Verification algorithm

tests/test_conformance_inclusion_profile_and_receipt_boundaries.py

@@ -0,0 +1,133 @@
+"""Behavioral boundary checks for conformance inclusion profile and receipt parsing (#625)."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+_CONFORMANCE_SRC = str(Path(__file__).resolve().parents[1] / "conformance" / "src")
+if sys.path[:1] != [_CONFORMANCE_SRC]:
+    sys.path.insert(0, _CONFORMANCE_SRC)
+
+from trace_tests.inclusion import (  # type: ignore[import-untyped] # noqa: E402
+    InclusionError,
+    canonical_claim_bytes,
+    parse_entry,
+    parse_proof,
+    parse_receipt,
+)
+
+_RECEIPT: dict[str, Any] = {
+    "leaf_index": 0,
+    "audit_path": [],
+    "leaf_count": 1,
+    "merkle_root": "sha256:" + "0" * 64,
+}
+_ENTRY: dict[str, Any] = {
+    "ts": "2026-10-15T00:00:00Z",
+    "merkle_root": "sha256:" + "0" * 64,
+    "leaf_count": 1,
+    "producer": "unit-test/1.0.0",
+    "batch_id": "unit-1",
+}
+
+
+@pytest.mark.parametrize(
+    ("claim", "expected"),
+    [
+        ({"s": "\ud800"}, "surrogate"),
+        ({"nested": ("\udfff",)}, "surrogate"),
+        ({"\ud800key": 1}, "surrogate"),
+        ({1: "val"}, "non-string object key"),  # type: ignore[dict-item]
+    ],
+    ids=["surrogate-value", "surrogate-tuple", "surrogate-key", "int-key"],
+)
+def test_canonical_claim_bytes_rejects_surrogates_and_non_string_keys(
+    claim: dict[str, Any], expected: str
+) -> None:
+    with pytest.raises(InclusionError, match=expected):
+        canonical_claim_bytes(claim)
+
+
+def test_canonical_claim_bytes_rejects_cycles_and_allows_shared_acyclic() -> None:
+    import threading
+
+    shared = {"k": 1}
+    assert canonical_claim_bytes({"a": shared, "b": shared}) == b'{"a":{"k":1},"b":{"k":1}}'
+
+    cyc_dict: dict[str, Any] = {}
+    cyc_dict["self"] = cyc_dict
+    caught: list[BaseException] = []
+
+    def _run_cyclic() -> None:
+        try:
+            canonical_claim_bytes(cyc_dict)
+        except BaseException as exc:
+            caught.append(exc)
+
+    worker = threading.Thread(target=_run_cyclic, daemon=True)
+    worker.start()
+    worker.join(timeout=0.25)
+    assert not worker.is_alive(), "canonical_claim_bytes hung on a circular reference"
+    assert len(caught) == 1 and isinstance(caught[0], InclusionError)
+    assert "circular reference" in str(caught[0])
+
+
+@pytest.mark.parametrize(
+    ("bad_receipt", "expected_match"),
+    [
+        ({**_RECEIPT, "leaf_index": True}, "receipt leaf_index must be a non-negative integer"),
+        ({**_RECEIPT, "leaf_index": "0"}, "receipt leaf_index must be a non-negative integer"),
+        ({**_RECEIPT, "leaf_index": -1}, "receipt leaf_index must be a non-negative integer"),
+        ({**_RECEIPT, "leaf_index": 2**53}, "receipt leaf_index must be a non-negative integer"),
+        ({**_RECEIPT, "leaf_count": 0}, "receipt leaf_count must be an integer >= 1"),
+        ({**_RECEIPT, "leaf_count": -1}, "receipt leaf_count must be an integer >= 1"),
+        ({**_RECEIPT, "leaf_count": 2**53}, "receipt leaf_count must be an integer >= 1"),
+        ({**_RECEIPT, "leaf_index": 1, "leaf_count": 1}, "must be less than leaf_count"),
+    ],
+)
+def test_parse_proof_and_receipt_reject_invalid_leaf_index_or_count(
+    bad_receipt: dict[str, Any], expected_match: str
+) -> None:
+    assert parse_receipt(_RECEIPT)[0] == 0
+    assert parse_proof({"leaf_index": 2**53 - 1, "audit_path": []})[0] == 2**53 - 1
+    assert (
+        parse_receipt({**_RECEIPT, "leaf_index": 2**53 - 2, "leaf_count": 2**53 - 1})[2]
+        == 2**53 - 1
+    )
+    with pytest.raises(InclusionError, match=expected_match):
+        parse_proof(bad_receipt)
+    with pytest.raises(InclusionError, match=expected_match):
+        parse_receipt(bad_receipt)
+
+
+def test_parse_proof_without_leaf_count_rejects_leaf_index_above_safe_int() -> None:
+    assert parse_proof({"leaf_index": 2**53 - 1, "audit_path": []})[0] == 2**53 - 1
+    with pytest.raises(
+        InclusionError, match="receipt leaf_index must be a non-negative integer"
+    ):
+        parse_proof({"leaf_index": 2**53, "audit_path": []})
+
+
+@pytest.mark.parametrize(
+    "bad_entry",
+    [
+        {**_ENTRY, "producer": "   "},
+        {**_ENTRY, "producer": "prod\n1.0"},
+        {**_ENTRY, "producer": "prod\x7f1.0"},
+        {**_ENTRY, "batch_id": "batch\x001"},
+        {**_ENTRY, "batch_id": "batch\x7f1"},
+        {**_ENTRY, "leaf_count": 2**53},
+    ],
+)
+def test_parse_entry_rejects_whitespace_control_chars_and_unsafe_leaf_count(
+    bad_entry: dict[str, Any],
+) -> None:
+    assert parse_entry(_ENTRY)[1] == 1
+    assert parse_entry({**_ENTRY, "producer": "Example Gateway 1.0"})[2] == "Example Gateway 1.0"
+    assert parse_entry({**_ENTRY, "leaf_count": 2**53 - 1})[1] == 2**53 - 1
+    with pytest.raises(InclusionError):
+        parse_entry(bad_entry)