From c495cb1bedbc67f61fc176227efc6d53c9fa98ae Mon Sep 17 00:00:00 2001
From: Raghuvar Durga <177662415+drk-ram@users.noreply.github.com>
Date: Sat, 17 Oct 2026 10:24:08 -0500
Subject: [PATCH 1/4] content_marking: enforce consumer field and strict UTF-8
 / RFC 8259 JSON boundaries

Closes #555

Signed-off-by: Raghuvar Durga <177662415+drk-ram@users.noreply.github.com>

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 src/agentrust_trace/content_marking.py        |  85 +++++-
 .../test_content_marking_consumer_boundary.py | 243 ++++++++++++++++++
 2 files changed, 319 insertions(+), 9 deletions(-)
 create mode 100644 tests/test_content_marking_consumer_boundary.py

diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py
index d0265914..b98855e6 100644
--- a/src/agentrust_trace/content_marking.py
+++ b/src/agentrust_trace/content_marking.py
@@ -74,16 +74,63 @@ def _no_duplicate_members(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
     return obj
 
 
+def _no_nonfinite_constant(value: str, what: str = "JSON") -> Any:
+    raise ContentMarkingError(f"{what} contains non-RFC-8259 numeric constant {value!r}")
+
+
+def _has_surrogate(value: Any) -> bool:
+    stack: list[Any] = [value]
+    seen: set[int] = set()
+    while stack:
+        item = stack.pop()
+        if isinstance(item, str):
+            if any(0xD800 <= ord(ch) <= 0xDFFF for ch in item):
+                return True
+        elif isinstance(item, dict):
+            oid = id(item)
+            if oid in seen:
+                continue
+            seen.add(oid)
+            stack.extend(item.keys())
+            stack.extend(item.values())
+        elif isinstance(item, (list, tuple)):
+            oid = id(item)
+            if oid in seen:
+                continue
+            seen.add(oid)
+            stack.extend(item)
+    return False
+
+
 def _loads(data: bytes | bytearray, what: str) -> Any:
+    if not isinstance(data, bytes | bytearray):
+        raise ContentMarkingError(f"{what} must be bytes, got {type(data).__name__}")
     try:
-        return json.loads(data, object_pairs_hook=_no_duplicate_members)
+        text = data.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise ContentMarkingError(f"{what} is not valid UTF-8: {exc}") from exc
+    if text.startswith("\ufeff"):
+        raise ContentMarkingError(
+            f"{what} must not start with a UTF-8 byte-order mark (RFC 8259 section 8.1)"
+        )
+    try:
+        parsed = json.loads(
+            text,
+            object_pairs_hook=_no_duplicate_members,
+            parse_constant=lambda value: _no_nonfinite_constant(value, what),
+        )
     except ContentMarkingError as exc:
+        if str(exc).startswith(what):
+            raise
         raise ContentMarkingError(f"{what} is not usable JSON: {exc}") from exc
     except (ValueError, RecursionError) as exc:
         # RecursionError: nesting deeper than the interpreter stack. The bytes can
         # come from whoever serves the URL, so a caller written against
         # ContentMarkingError has to be able to refuse them.
         raise ContentMarkingError(f"{what} is not JSON: {exc}") from exc
+    if _has_surrogate(parsed):
+        raise ContentMarkingError(f"{what} contains an unpaired UTF-16 surrogate")
+    return parsed
 
 
 def _digest(data: bytes, alg: str) -> str:
@@ -171,13 +218,16 @@ def build_assertion(
         raise ContentMarkingError(
             f"alg must be a digest algorithm name, got {type(alg).__name__}; use sha256 or sha384"
         )
-    if anchor is not None and (not isinstance(anchor, str) or not anchor):
-        # `if anchor:` below used to drop a non-string or empty anchor on the floor, so a
-        # caller who passed one got an assertion with no anchor and no error. Refuse it.
-        raise ContentMarkingError(
-            "anchor must be a non-empty registry entry URI string or None, got "
-            f"{anchor!r}"
-        )
+    if anchor is not None:
+        if not isinstance(anchor, str) or not anchor:
+            # `if anchor:` below used to drop a non-string or empty anchor on the floor, so a
+            # caller who passed one got an assertion with no anchor and no error. Refuse it.
+            raise ContentMarkingError(
+                "anchor must be a non-empty registry entry URI string or None, got "
+                f"{anchor!r}"
+            )
+        if _has_surrogate(anchor):
+            raise ContentMarkingError("anchor must not contain an unpaired UTF-16 surrogate")
 
     record = _loads(record_bytes, "record_bytes")
     if not isinstance(record, dict):
@@ -188,7 +238,7 @@ def build_assertion(
 
     subject = record.get("subject")
     profile = record.get("eat_profile")
-    if not _SUBJECT_RE.match(str(subject or "")):
+    if not isinstance(subject, str) or not _SUBJECT_RE.match(subject):
         raise ContentMarkingError("the record has no SPIFFE or DID subject to bind to")
     if not isinstance(profile, str) or not profile:
         raise ContentMarkingError("the record declares no non-empty string eat_profile")
@@ -248,6 +298,23 @@ def verify_assertion(assertion: dict[str, Any], record_bytes: bytes) -> dict[str
                 "establishes agreement rather than presence."
             )
 
+    subject = data.get("subject")
+    if not isinstance(subject, str) or not _SUBJECT_RE.match(subject):
+        raise ContentMarkingError("assertion subject must be a SPIFFE or DID URI")
+    profile = data.get("eat_profile")
+    if not isinstance(profile, str) or not profile:
+        raise ContentMarkingError("assertion eat_profile must be a non-empty string")
+    if "anchor" in data:
+        anchor = data.get("anchor")
+        if not isinstance(anchor, str) or not anchor:
+            raise ContentMarkingError("assertion anchor must be a non-empty string when present")
+        if _has_surrogate(anchor):
+            raise ContentMarkingError(
+                "assertion anchor must not contain an unpaired UTF-16 surrogate"
+            )
+    if _has_surrogate(assertion):
+        raise ContentMarkingError("assertion contains an unpaired UTF-16 surrogate")
+
     ref = data.get("record")
     if not isinstance(ref, dict):
         raise ContentMarkingError("assertion carries no record reference")
diff --git a/tests/test_content_marking_consumer_boundary.py b/tests/test_content_marking_consumer_boundary.py
new file mode 100644
index 00000000..20db02de
--- /dev/null
+++ b/tests/test_content_marking_consumer_boundary.py
@@ -0,0 +1,243 @@
+"""Consumer and RFC 8259 UTF-8 / JSON boundary tests for content_marking (#555)."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Any
+
+import pytest
+
+from agentrust_trace.content_marking import (
+    ASSERTION_LABEL,
+    ContentMarkingError,
+    RecordMismatch,
+    _loads,
+    build_assertion,
+    verify_assertion,
+)
+
+URL = "https://registry.example/records/abc123.json"
+ANCHOR = "https://registry.example/entries/2026-08-09-001"
+
+
+def _valid_record(
+    *,
+    subject: str = "spiffe://example.org/agent/image-bot",
+    eat_profile: str = "tag:agentrust-io.com,2026:trace-v0.2",
+) -> dict[str, Any]:
+    return {
+        "eat_profile": eat_profile,
+        "iat": 1760000000,
+        "subject": subject,
+        "data_class": "public",
+    }
+
+
+def _assert_positive_controls() -> None:
+    for subject in ("spiffe://example.org/agent/image-bot", "did:web:example.com"):
+        raw = json.dumps(_valid_record(subject=subject)).encode()
+        for alg in ("sha256", "sha384"):
+            for anchor in (None, ANCHOR):
+                built = build_assertion(raw, url=URL, alg=alg, anchor=anchor)
+                verified = verify_assertion(built, raw)
+                assert verified["subject"] == subject
+
+
+def _crafted_pair(
+    *,
+    subject: Any = "spiffe://example.org/agent/image-bot",
+    eat_profile: Any = "tag:agentrust-io.com,2026:trace-v0.2",
+    anchor: Any = ...,
+    extra_record: dict[str, Any] | None = None,
+) -> tuple[dict[str, Any], bytes]:
+    record_obj: dict[str, Any] = {
+        "eat_profile": eat_profile,
+        "iat": 1760000000,
+        "subject": subject,
+        "data_class": "public",
+    }
+    if extra_record:
+        record_obj.update(extra_record)
+    raw = json.dumps(record_obj).encode()
+    data: dict[str, Any] = {
+        "version": 1,
+        "record": {
+            "url": URL,
+            "alg": "sha256",
+            "hash": "sha256:" + hashlib.sha256(raw).hexdigest(),
+        },
+        "subject": subject,
+        "eat_profile": eat_profile,
+    }
+    if anchor is not ...:
+        data["anchor"] = anchor
+    return {"label": ASSERTION_LABEL, "data": data}, raw
+
+
+@pytest.mark.parametrize(
+    "bad_subject",
+    [123, True, False, 0, None, "", "not-a-uri", "spiffe://host-only", [], {}],
+)
+def test_verify_assertion_rejects_invalid_subject_even_when_record_agrees(
+    bad_subject: Any,
+) -> None:
+    _assert_positive_controls()
+
+    valid_a, _ = _crafted_pair(subject="spiffe://example.org/agent/image-bot")
+    _, other_raw = _crafted_pair(subject="did:web:example.com")
+    valid_a["data"]["record"]["hash"] = "sha256:" + hashlib.sha256(other_raw).hexdigest()
+    with pytest.raises(RecordMismatch, match="assertion names subject"):
+        verify_assertion(valid_a, other_raw)
+
+    assertion, raw = _crafted_pair(subject=bad_subject)
+    with pytest.raises(ContentMarkingError, match="SPIFFE or DID subject"):
+        build_assertion(raw, url=URL)
+    with pytest.raises(
+        ContentMarkingError, match="assertion subject must be a SPIFFE or DID URI"
+    ) as exc_info:
+        verify_assertion(assertion, raw)
+    assert not isinstance(exc_info.value, RecordMismatch)
+
+
+@pytest.mark.parametrize("bad_profile", ["", None, 0, False, 123, [], {}])
+def test_verify_assertion_rejects_invalid_eat_profile_even_when_record_agrees(
+    bad_profile: Any,
+) -> None:
+    _assert_positive_controls()
+
+    valid_a, _ = _crafted_pair(eat_profile="tag:agentrust-io.com,2026:trace-v0.2")
+    _, other_raw = _crafted_pair(eat_profile="tag:agentrust.io,2026:trace-v0.1")
+    valid_a["data"]["record"]["hash"] = "sha256:" + hashlib.sha256(other_raw).hexdigest()
+    with pytest.raises(RecordMismatch, match="eat_profile"):
+        verify_assertion(valid_a, other_raw)
+
+    assertion, raw = _crafted_pair(eat_profile=bad_profile)
+    with pytest.raises(
+        ContentMarkingError, match="assertion eat_profile must be a non-empty string"
+    ) as exc_info:
+        verify_assertion(assertion, raw)
+    assert not isinstance(exc_info.value, RecordMismatch)
+
+
+@pytest.mark.parametrize("bad_anchor", ["", None, 0, False, 123, [], {}, "\ud800"])
+def test_verify_assertion_rejects_invalid_anchor_when_present(bad_anchor: Any) -> None:
+    _assert_positive_controls()
+
+    assertion, raw = _crafted_pair(
+        anchor=bad_anchor,
+        extra_record={"anchor": bad_anchor if bad_anchor != "\ud800" else "ok"},
+    )
+    with pytest.raises(ContentMarkingError, match="assertion anchor must") as exc_info:
+        verify_assertion(assertion, raw)
+    assert not isinstance(exc_info.value, RecordMismatch)
+
+
+@pytest.mark.parametrize("surrogate", ["\ud800", "\udfff", "https://example.com/\ud800"])
+def test_build_assertion_rejects_unpaired_surrogate_in_anchor(surrogate: str) -> None:
+    _assert_positive_controls()
+    raw = json.dumps(_valid_record()).encode()
+    with pytest.raises(
+        ContentMarkingError, match="anchor must not contain an unpaired UTF-16 surrogate"
+    ):
+        build_assertion(raw, url=URL, anchor=surrogate)
+
+
+@pytest.mark.parametrize(
+    ("encoding", "bom"),
+    [
+        ("utf-16-le", b"\xff\xfe"),
+        ("utf-16-be", b"\xfe\xff"),
+        ("utf-32-le", b"\xff\xfe\x00\x00"),
+        ("utf-32-be", b"\x00\x00\xfe\xff"),
+    ],
+)
+def test_utf16_and_utf32_bom_encoded_record_bytes_are_rejected(
+    encoding: str, bom: bytes
+) -> None:
+    _assert_positive_controls()
+    encoded = bom + json.dumps(_valid_record()).encode(encoding)
+    with pytest.raises(ContentMarkingError, match="is not valid UTF-8"):
+        build_assertion(encoded, url=URL)
+
+    assertion, _ = _crafted_pair()
+    assertion["data"]["record"]["hash"] = "sha256:" + hashlib.sha256(encoded).hexdigest()
+    with pytest.raises(ContentMarkingError, match="is not valid UTF-8"):
+        verify_assertion(assertion, encoded)
+
+
+def test_utf8_bom_prefixed_record_bytes_are_rejected() -> None:
+    _assert_positive_controls()
+    bom_bytes = json.dumps(_valid_record()).encode("utf-8-sig")
+    with pytest.raises(
+        ContentMarkingError, match="must not start with a UTF-8 byte-order mark"
+    ):
+        build_assertion(bom_bytes, url=URL)
+
+    assertion, _ = _crafted_pair()
+    assertion["data"]["record"]["hash"] = "sha256:" + hashlib.sha256(bom_bytes).hexdigest()
+    with pytest.raises(
+        ContentMarkingError, match="must not start with a UTF-8 byte-order mark"
+    ):
+        verify_assertion(assertion, bom_bytes)
+
+
+@pytest.mark.parametrize("token", ["NaN", "Infinity", "-Infinity"])
+def test_non_rfc8259_numeric_constants_are_rejected(token: str) -> None:
+    _assert_positive_controls()
+    raw = (
+        '{"subject":"spiffe://example.org/agent/image-bot",'
+        f'"eat_profile":"tag:agentrust-io.com,2026:trace-v0.2","iat":{token}}}'
+    ).encode()
+    with pytest.raises(ContentMarkingError, match="contains non-RFC-8259 numeric constant"):
+        build_assertion(raw, url=URL)
+
+    assertion, _ = _crafted_pair()
+    assertion["data"]["record"]["hash"] = "sha256:" + hashlib.sha256(raw).hexdigest()
+    with pytest.raises(ContentMarkingError, match="contains non-RFC-8259 numeric constant"):
+        verify_assertion(assertion, raw)
+
+
+@pytest.mark.parametrize("escape", [r"\ud800", r"\udfff"])
+@pytest.mark.parametrize("location", ["value", "key"])
+def test_unpaired_surrogates_in_record_bytes_are_rejected(
+    escape: str, location: str
+) -> None:
+    _assert_positive_controls()
+    if location == "value":
+        raw = (
+            '{"subject":"spiffe://example.org/agent/image-bot",'
+            '"eat_profile":"tag:agentrust-io.com,2026:trace-v0.2",'
+            f'"note":"{escape}"}}'
+        ).encode()
+    else:
+        raw = (
+            '{"subject":"spiffe://example.org/agent/image-bot",'
+            '"eat_profile":"tag:agentrust-io.com,2026:trace-v0.2",'
+            f'"{escape}":1}}'
+        ).encode()
+
+    with pytest.raises(ContentMarkingError, match="contains an unpaired UTF-16 surrogate"):
+        build_assertion(raw, url=URL)
+
+    assertion, _ = _crafted_pair()
+    assertion["data"]["record"]["hash"] = "sha256:" + hashlib.sha256(raw).hexdigest()
+    with pytest.raises(ContentMarkingError, match="contains an unpaired UTF-16 surrogate"):
+        verify_assertion(assertion, raw)
+
+
+@pytest.mark.parametrize("surrogate", ["\ud800", "\udfff"])
+def test_unpaired_surrogates_anywhere_in_assertion_are_rejected(
+    surrogate: str,
+) -> None:
+    _assert_positive_controls()
+    assertion, raw = _crafted_pair()
+    assertion["data"]["extra"] = {"nested": [surrogate]}
+    with pytest.raises(ContentMarkingError, match="contains an unpaired UTF-16 surrogate"):
+        verify_assertion(assertion, raw)
+
+
+def test_loads_rejects_non_bytes_input() -> None:
+    _assert_positive_controls()
+    with pytest.raises(ContentMarkingError, match="record_bytes must be bytes, got str"):
+        _loads("{}", "record_bytes")  # type: ignore[arg-type]

From 4b4de13a49c98398c4e7e207ee98f922e94efb8f Mon Sep 17 00:00:00 2001
From: Srinivasa Consultants <78673153+dinakarjs@users.noreply.github.com>
Date: Sun, 18 Oct 2026 18:48:33 +0530
Subject: [PATCH 2/4] test: cover content marking byte and Unicode boundaries

Add 22 public-API paired controls offered for PR #565. Retain exact-byte hash semantics and distinguish malformed fetched JSON from stale hashes.

Signed-off-by: Srinivasa Consultants <78673153+dinakarjs@users.noreply.github.com>
---
 tests/test_content_marking_review_controls.py | 140 ++++++++++++++++++
 1 file changed, 140 insertions(+)
 create mode 100644 tests/test_content_marking_review_controls.py

diff --git a/tests/test_content_marking_review_controls.py b/tests/test_content_marking_review_controls.py
new file mode 100644
index 00000000..76c0197a
--- /dev/null
+++ b/tests/test_content_marking_review_controls.py
@@ -0,0 +1,140 @@
+"""Public-API paired regressions offered as a companion to PR #565.
+
+These preserve current behavior without deciding the author's pending error-message,
+profile-documentation or fetched-record validation choices.
+"""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+from typing import Any
+
+import pytest
+
+from agentrust_trace.content_marking import (
+    ContentMarkingError,
+    RecordMismatch,
+    build_assertion,
+    verify_assertion,
+)
+
+URL = "https://registry.example/records/review.json"
+RECORD = {
+    "subject": "spiffe://example.org/agent/review",
+    "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2",
+    "iat": 1760000000,
+}
+
+
+def _raw(**extra: object) -> bytes:
+    return json.dumps({**RECORD, **extra}, ensure_ascii=False).encode("utf-8")
+
+
+def _retarget(assertion: dict[str, Any], raw: bytes, alg: str) -> dict[str, Any]:
+    result = copy.deepcopy(assertion)
+    result["data"]["record"]["hash"] = f"{alg}:" + hashlib.new(alg, raw).hexdigest()
+    return result
+
+
+@pytest.mark.parametrize("alg", ["sha256", "sha384"])
+def test_bytearray_round_trip_matches_exact_bytes_and_hash(alg: str) -> None:
+    raw = _raw(note="supplementary scalar: \U0001f600")
+    mutable = bytearray(raw)
+    expected = build_assertion(raw, url=URL, alg=alg)
+    built = build_assertion(mutable, url=URL, alg=alg)  # type: ignore[arg-type]
+    assert built == expected
+    assert built["data"]["record"]["hash"] == f"{alg}:" + hashlib.new(alg, raw).hexdigest()
+    assert verify_assertion(built, mutable) == RECORD | {  # type: ignore[arg-type]
+        "note": "supplementary scalar: \U0001f600"
+    }
+    assert bytes(mutable) == raw
+
+
+@pytest.mark.parametrize("alg", ["sha256", "sha384"])
+@pytest.mark.parametrize(
+    "member",
+    [
+        b'"subject":"did:web:other.example","sub\\u006aect":"did:web:last.example"',
+        b'"extra":{"k":1,"\\u006b":2}',
+        b'"extra":[{"k":1,"\\u006b":2}]',
+    ],
+    ids=["escaped-top-level", "nested-object", "object-inside-array"],
+)
+def test_escaped_duplicate_members_refuse_build_and_hash_matching_verify(
+    alg: str, member: bytes
+) -> None:
+    good = _raw(extra=[{"k": 1, "other": 2}])
+    assertion = build_assertion(good, url=URL, alg=alg)
+    assert verify_assertion(assertion, good)["extra"] == [{"k": 1, "other": 2}]
+    # JSON escaping changes spelling, not member identity. Check both producers
+    # and consumers, without requiring a specific wrapping of the diagnostic.
+    bad = _raw()[:-1] + b"," + member + b"}"
+    with pytest.raises(ContentMarkingError, match="duplicate member") as error:
+        build_assertion(bad, url=URL, alg=alg)
+    assert not isinstance(error.value, RecordMismatch)
+    with pytest.raises(ContentMarkingError, match="duplicate member") as error:
+        verify_assertion(_retarget(assertion, bad, alg), bad)
+    assert not isinstance(error.value, RecordMismatch)
+
+
+@pytest.mark.parametrize("alg", ["sha256", "sha384"])
+def test_utf8_scalar_and_json_escape_pair_both_work_but_bind_distinct_bytes(alg: str) -> None:
+    scalar = _raw(note="\U0001f600")
+    escaped = json.dumps(RECORD | {"note": "\U0001f600"}, ensure_ascii=True).encode("utf-8")
+    assert scalar != escaped
+    a = build_assertion(scalar, url=URL, alg=alg)
+    b = build_assertion(escaped, url=URL, alg=alg)
+    assert verify_assertion(a, scalar) == verify_assertion(b, escaped)
+    assert a["data"]["record"]["hash"] != b["data"]["record"]["hash"]
+    with pytest.raises(RecordMismatch):
+        verify_assertion(a, escaped)
+
+
+@pytest.mark.parametrize("alg", ["sha256", "sha384"])
+def test_interior_bom_scalar_is_content_and_not_a_leading_byte_order_mark(alg: str) -> None:
+    raw = _raw(note="before\ufeffafter")
+    assertion = build_assertion(raw, url=URL, alg=alg)
+    assert verify_assertion(assertion, raw)["note"] == "before\ufeffafter"
+    prefixed = b"\xef\xbb\xbf" + raw
+    with pytest.raises(ContentMarkingError, match="byte-order mark") as error:
+        verify_assertion(_retarget(assertion, prefixed, alg), prefixed)
+    assert not isinstance(error.value, RecordMismatch)
+
+
+@pytest.mark.parametrize("alg", ["sha256", "sha384"])
+def test_direct_anchor_accepts_utf8_scalar_without_normalizing_its_spelling(alg: str) -> None:
+    raw = _raw()
+    anchor = "https://registry.example/entries/\U0001f600"
+    assertion = build_assertion(raw, url=URL, alg=alg, anchor=anchor)
+    assert assertion["data"]["anchor"] == anchor
+    assert verify_assertion(assertion, raw) == RECORD
+
+
+@pytest.mark.parametrize("alg", ["sha256", "sha384"])
+def test_trailing_newline_changes_binding_despite_identical_decoded_record(alg: str) -> None:
+    raw = _raw()
+    newline = raw + b"\n"
+    assertion = build_assertion(raw, url=URL, alg=alg)
+    assert verify_assertion(assertion, raw) == RECORD
+    with pytest.raises(RecordMismatch):
+        verify_assertion(assertion, newline)
+    rebound = build_assertion(newline, url=URL, alg=alg)
+    assert verify_assertion(rebound, newline) == RECORD
+    assert rebound["data"]["record"]["hash"] != assertion["data"]["record"]["hash"]
+
+
+@pytest.mark.parametrize("alg", ["sha256", "sha384"])
+@pytest.mark.parametrize("bad", [b"{not-json}", b"\xff", b'{"n":NaN}'])
+def test_stale_hash_and_matching_hash_distinguish_changed_bytes_from_bad_json(
+    alg: str, bad: bytes
+) -> None:
+    good = _raw()
+    assertion = build_assertion(good, url=URL, alg=alg)
+    assert verify_assertion(assertion, good) == RECORD
+    with pytest.raises(RecordMismatch):
+        verify_assertion(assertion, bad)
+    with pytest.raises(ContentMarkingError) as error:
+        verify_assertion(_retarget(assertion, bad, alg), bad)
+    assert not isinstance(error.value, RecordMismatch)

From 06ddb481bdf776adfd78c1a4d9567544df9730a8 Mon Sep 17 00:00:00 2001
From: Raghuvar Durga <177662415+drk-ram@users.noreply.github.com>
Date: Sun, 18 Oct 2026 11:23:16 -0500
Subject: [PATCH 3/4] content_marking: address review feedback on record
 validation, types, and spec

Signed-off-by: Raghuvar Durga <177662415+drk-ram@users.noreply.github.com>

Co-authored-by: Srinivasa Consultants <78673153+dinakarjs@users.noreply.github.com>

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 CHANGELOG.md                                  |  6 +++
 spec/content-marking-v1.md                    | 10 +++--
 src/agentrust_trace/content_marking.py        | 44 ++++++++++++++-----
 .../test_content_marking_consumer_boundary.py | 44 +++++++++++++++++--
 tests/test_content_marking_review_controls.py |  4 +-
 5 files changed, 88 insertions(+), 20 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b1190af5..3243cb94 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,12 @@ Format: [Semantic Versioning](https://semver.org/). Spec versions follow `MAJOR.
 
 ## [Unreleased]
 
+- **`content_marking`: consumer field and strict UTF-8 / RFC 8259 JSON boundaries.**
+  `verify_assertion` now validates `subject` (SPIFFE or DID URI), `eat_profile` (non-empty
+  string), and optional `anchor` (non-empty string) on both the assertion and the resolved
+  record before comparing them, and `_loads` decodes `bytes | bytearray` as strict UTF-8
+  JSON without a leading byte-order mark, non-finite numeric constants (`NaN`, `Infinity`,
+  `-Infinity`), or unpaired UTF-16 surrogates (#555, #565, by @drk-ram).
 - **Section 3.1.5: a layer is established by its event log or by a reference value
   (#279).** As merged, a layer was `established` only when its event log replayed to the
   quoted value, so a quote that equals signed reference values with no log could not be
diff --git a/spec/content-marking-v1.md b/spec/content-marking-v1.md
index 6de34228..67d47d5f 100644
--- a/spec/content-marking-v1.md
+++ b/spec/content-marking-v1.md
@@ -101,10 +101,14 @@ proves an execution happened, not that this asset came from it.
 1. Verify the C2PA manifest per C2PA 2.1. Failure here ends it.
 2. Read `com.agentrust-io.trace`. An unknown `version` is rejected, not best-effort parsed.
 3. Fetch `record.url` and check the bytes against `record.hash`. A mismatch means the record
-   was changed after the asset was signed, and is the finding.
+   was changed after the asset was signed, and is the finding. The record bytes are decoded
+   as strict UTF-8 JSON without a leading byte-order mark, non-finite numeric constants
+   (`NaN`, `Infinity`, `-Infinity`), or unpaired UTF-16 surrogates.
 4. Verify the Trust Record per TRACE v0.2 §3.3, against a key you already trust.
-5. Check `subject` and `eat_profile` in the assertion match the fetched record. A mismatch
-   means the assertion describes a different record than it points at.
+5. Check that both the assertion and the resolved record carry a valid SPIFFE or DID
+   `subject`, a non-empty string `eat_profile`, and (when present) a non-empty string
+   `anchor`, and that `subject` and `eat_profile` in the assertion match the fetched record.
+   A mismatch means the assertion describes a different record than it points at.
 6. If `anchor` is present, verify inclusion per Anchor Format v1 §5.1.
 7. **Weigh the result by the record's conformance level.** A Level 0 record binds this asset
    to an unattested software execution. That is worth having and it is not hardware-rooted
diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py
index b98855e6..4ec9e683 100644
--- a/src/agentrust_trace/content_marking.py
+++ b/src/agentrust_trace/content_marking.py
@@ -79,6 +79,14 @@ def _no_nonfinite_constant(value: str, what: str = "JSON") -> Any:
 
 
 def _has_surrogate(value: Any) -> bool:
+    """Return True if *value* holds any UTF-16 surrogate code point (U+D800..U+DFFF).
+
+    A JSON surrogate-pair escape such as ``\\ud83d\\ude00`` is decoded by ``json.loads``
+    into the supplementary Unicode scalar ``U+1F600`` before this runs, so valid JSON
+    surrogate pairs pass. Any code point still in ``U+D800..U+DFFF`` after decoding, or
+    passed directly in a Python ``str`` (including two adjacent surrogate code points),
+    is not valid UTF-8 and is refused.
+    """
     stack: list[Any] = [value]
     seen: set[int] = set()
     while stack:
@@ -104,7 +112,9 @@ def _has_surrogate(value: Any) -> bool:
 
 def _loads(data: bytes | bytearray, what: str) -> Any:
     if not isinstance(data, bytes | bytearray):
-        raise ContentMarkingError(f"{what} must be bytes, got {type(data).__name__}")
+        raise ContentMarkingError(
+            f"{what} must be bytes or bytearray, got {type(data).__name__}"
+        )
     try:
         text = data.decode("utf-8")
     except UnicodeDecodeError as exc:
@@ -119,10 +129,8 @@ def _loads(data: bytes | bytearray, what: str) -> Any:
             object_pairs_hook=_no_duplicate_members,
             parse_constant=lambda value: _no_nonfinite_constant(value, what),
         )
-    except ContentMarkingError as exc:
-        if str(exc).startswith(what):
-            raise
-        raise ContentMarkingError(f"{what} is not usable JSON: {exc}") from exc
+    except ContentMarkingError:
+        raise
     except (ValueError, RecursionError) as exc:
         # RecursionError: nesting deeper than the interpreter stack. The bytes can
         # come from whoever serves the URL, so a caller written against
@@ -133,10 +141,10 @@ def _loads(data: bytes | bytearray, what: str) -> Any:
     return parsed
 
 
-def _digest(data: bytes, alg: str) -> str:
+def _digest(data: bytes | bytearray, alg: str) -> str:
     if alg not in _ALGS:
         raise ContentMarkingError(f"unsupported digest algorithm {alg!r}; use sha256 or sha384")
-    return f"{alg}:{_ALGS[alg](data).hexdigest()}"
+    return f"{alg}:{_ALGS[alg](bytes(data)).hexdigest()}"
 
 
 def _record_url(value: Any) -> str:
@@ -188,7 +196,7 @@ def _record_url(value: Any) -> str:
 
 
 def build_assertion(
-    record_bytes: bytes,
+    record_bytes: bytes | bytearray,
     *,
     url: str,
     alg: str = "sha256",
@@ -254,7 +262,9 @@ def build_assertion(
     return {"label": ASSERTION_LABEL, "data": data}
 
 
-def verify_assertion(assertion: dict[str, Any], record_bytes: bytes) -> dict[str, Any]:
+def verify_assertion(
+    assertion: dict[str, Any], record_bytes: bytes | bytearray
+) -> dict[str, Any]:
     """Check an assertion against the record bytes actually retrieved from its URL.
 
     *record_bytes* is not optional and has no default. The hash is the binding,
@@ -360,13 +370,23 @@ def verify_assertion(assertion: dict[str, Any], record_bytes: bytes) -> dict[str
                 "the assertion has nothing to be bound to. It matched the declared hash, "
                 "so this is what that URL is serving."
             )
-    if record.get("subject") != data.get("subject"):
+    record_subject = record.get("subject")
+    if not isinstance(record_subject, str) or not _SUBJECT_RE.match(record_subject):
+        raise ContentMarkingError("the record has no SPIFFE or DID subject to bind to")
+    record_profile = record.get("eat_profile")
+    if not isinstance(record_profile, str) or not record_profile:
+        raise ContentMarkingError("the record declares no non-empty string eat_profile")
+    if "anchor" in record and (
+        not isinstance(record["anchor"], str) or not record["anchor"]
+    ):
+        raise ContentMarkingError("the record anchor must be a non-empty string")
+    if record_subject != data.get("subject"):
         raise RecordMismatch(
             f"assertion names subject {data.get('subject')!r} and the record says "
-            f"{record.get('subject')!r}: the assertion describes a different record than it "
+            f"{record_subject!r}: the assertion describes a different record than it "
             "points at"
         )
-    if record.get("eat_profile") != data.get("eat_profile"):
+    if record_profile != data.get("eat_profile"):
         raise RecordMismatch(
             "assertion and record disagree about eat_profile, so the assertion describes a "
             "different record than it points at"
diff --git a/tests/test_content_marking_consumer_boundary.py b/tests/test_content_marking_consumer_boundary.py
index 20db02de..78e0aacb 100644
--- a/tests/test_content_marking_consumer_boundary.py
+++ b/tests/test_content_marking_consumer_boundary.py
@@ -133,10 +133,22 @@ def test_verify_assertion_rejects_invalid_anchor_when_present(bad_anchor: Any) -
     assert not isinstance(exc_info.value, RecordMismatch)
 
 
-@pytest.mark.parametrize("surrogate", ["\ud800", "\udfff", "https://example.com/\ud800"])
+@pytest.mark.parametrize(
+    "surrogate",
+    ["\ud800", "\udfff", "\ud800\udc00", "\ud83d\ude00", "https://example.com/\ud800"],
+)
 def test_build_assertion_rejects_unpaired_surrogate_in_anchor(surrogate: str) -> None:
     _assert_positive_controls()
     raw = json.dumps(_valid_record()).encode()
+    escaped_pair_raw = (
+        b'{"subject":"spiffe://example.org/agent/image-bot",'
+        b'"eat_profile":"tag:agentrust-io.com,2026:trace-v0.2",'
+        b'"note":"\\ud83d\\ude00"}'
+    )
+    built_pair = build_assertion(
+        escaped_pair_raw, url=URL, anchor="https://example.com/\U0001f600"
+    )
+    assert verify_assertion(built_pair, escaped_pair_raw)["note"] == "\U0001f600"
     with pytest.raises(
         ContentMarkingError, match="anchor must not contain an unpaired UTF-16 surrogate"
     ):
@@ -226,7 +238,7 @@ def test_unpaired_surrogates_in_record_bytes_are_rejected(
         verify_assertion(assertion, raw)
 
 
-@pytest.mark.parametrize("surrogate", ["\ud800", "\udfff"])
+@pytest.mark.parametrize("surrogate", ["\ud800", "\udfff", "\ud800\udc00", "\ud83d\ude00"])
 def test_unpaired_surrogates_anywhere_in_assertion_are_rejected(
     surrogate: str,
 ) -> None:
@@ -237,7 +249,33 @@ def test_unpaired_surrogates_anywhere_in_assertion_are_rejected(
         verify_assertion(assertion, raw)
 
 
+@pytest.mark.parametrize(
+    ("bad_record_overrides", "expected_match"),
+    [
+        ({"subject": 123}, "the record has no SPIFFE or DID subject to bind to"),
+        ({"subject": "not-a-uri"}, "the record has no SPIFFE or DID subject to bind to"),
+        ({"eat_profile": ""}, "the record declares no non-empty string eat_profile"),
+        ({"eat_profile": 123}, "the record declares no non-empty string eat_profile"),
+        ({"anchor": ""}, "the record anchor must be a non-empty string"),
+        ({"anchor": 123}, "the record anchor must be a non-empty string"),
+    ],
+)
+def test_fetched_record_with_invalid_fields_raises_content_marking_error_not_mismatch(
+    bad_record_overrides: dict[str, Any], expected_match: str
+) -> None:
+    _assert_positive_controls()
+    assertion, _ = _crafted_pair(anchor=ANCHOR)
+    bad_record = {**_valid_record(), **bad_record_overrides}
+    bad_raw = json.dumps(bad_record).encode()
+    assertion["data"]["record"]["hash"] = "sha256:" + hashlib.sha256(bad_raw).hexdigest()
+    with pytest.raises(ContentMarkingError, match=expected_match) as exc_info:
+        verify_assertion(assertion, bad_raw)
+    assert not isinstance(exc_info.value, RecordMismatch)
+
+
 def test_loads_rejects_non_bytes_input() -> None:
     _assert_positive_controls()
-    with pytest.raises(ContentMarkingError, match="record_bytes must be bytes, got str"):
+    with pytest.raises(
+        ContentMarkingError, match="record_bytes must be bytes or bytearray, got str"
+    ):
         _loads("{}", "record_bytes")  # type: ignore[arg-type]
diff --git a/tests/test_content_marking_review_controls.py b/tests/test_content_marking_review_controls.py
index 76c0197a..991084ac 100644
--- a/tests/test_content_marking_review_controls.py
+++ b/tests/test_content_marking_review_controls.py
@@ -43,10 +43,10 @@ def test_bytearray_round_trip_matches_exact_bytes_and_hash(alg: str) -> None:
     raw = _raw(note="supplementary scalar: \U0001f600")
     mutable = bytearray(raw)
     expected = build_assertion(raw, url=URL, alg=alg)
-    built = build_assertion(mutable, url=URL, alg=alg)  # type: ignore[arg-type]
+    built = build_assertion(mutable, url=URL, alg=alg)
     assert built == expected
     assert built["data"]["record"]["hash"] == f"{alg}:" + hashlib.new(alg, raw).hexdigest()
-    assert verify_assertion(built, mutable) == RECORD | {  # type: ignore[arg-type]
+    assert verify_assertion(built, mutable) == RECORD | {
         "note": "supplementary scalar: \U0001f600"
     }
     assert bytes(mutable) == raw

From 7948259f2edbd5c875a9beaf4d92ab2e86ce9bf9 Mon Sep 17 00:00:00 2001
From: Srinivasa Consultants <78673153+dinakarjs@users.noreply.github.com>
Date: Mon, 19 Oct 2026 01:30:43 +0530
Subject: [PATCH 4/4] fix(content-marking): snapshot mutable record bytes
 consistently

Parse and hash one immutable bytes snapshot in build and verify. Classify hash-matching missing required fields as malformed, matching other invalid fields. Preserve RecordMismatch for hash or well-formed field disagreement.

Add deterministic seam-mutation controls for both digest algorithms and paired malformed-record controls. Remove issue numbers from test docstrings.

Signed-off-by: Srinivasa Consultants <78673153+dinakarjs@users.noreply.github.com>

Signed-off-by: Raghuvar Durga <177662415+drk-ram@users.noreply.github.com>

Co-authored-by: Srinivasa Consultants <78673153+dinakarjs@users.noreply.github.com>

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 CHANGELOG.md                                  |  4 +
 src/agentrust_trace/content_marking.py        | 13 ++-
 tests/test_content_marking.py                 |  7 +-
 .../test_content_marking_consumer_boundary.py |  2 +-
 tests/test_content_marking_review_controls.py |  2 +-
 tests/test_content_marking_snapshot.py        | 80 +++++++++++++++++++
 6 files changed, 100 insertions(+), 8 deletions(-)
 create mode 100644 tests/test_content_marking_snapshot.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3243cb94..9cbf8bd1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,6 +17,10 @@ Format: [Semantic Versioning](https://semver.org/). Spec versions follow `MAJOR.
   record before comparing them, and `_loads` decodes `bytes | bytearray` as strict UTF-8
   JSON without a leading byte-order mark, non-finite numeric constants (`NaN`, `Infinity`,
   `-Infinity`), or unpaired UTF-16 surrogates (#555, #565, by @drk-ram).
+  Both public functions parse and hash one immutable snapshot of mutable record input.
+  Hash-matching records missing a required field now raise `ContentMarkingError`,
+  consistent with other malformed fields, rather than its `RecordMismatch` subclass.
+  Hash or well-formed field disagreements still raise `RecordMismatch`.
 - **Section 3.1.5: a layer is established by its event log or by a reference value
   (#279).** As merged, a layer was `established` only when its event log replayed to the
   quoted value, so a quote that equals signed reference values with no log could not be
diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py
index 4ec9e683..edb3453e 100644
--- a/src/agentrust_trace/content_marking.py
+++ b/src/agentrust_trace/content_marking.py
@@ -204,6 +204,7 @@ def build_assertion(
 ) -> dict[str, Any]:
     """Build the assertion from the **exact bytes** that will be served at *url*.
 
+    Accepts bytes or bytearray and freezes one bytes snapshot for parsing and hashing.
     Takes bytes rather than a record object on purpose. A hash computed over a
     re-serialized dict is a hash of something nobody will ever fetch: key order,
     separators and escaping all change the bytes without changing the record, and
@@ -237,6 +238,8 @@ def build_assertion(
         if _has_surrogate(anchor):
             raise ContentMarkingError("anchor must not contain an unpaired UTF-16 surrogate")
 
+    # Hash and parse the same immutable snapshot of caller-owned mutable input.
+    record_bytes = bytes(record_bytes)
     record = _loads(record_bytes, "record_bytes")
     if not isinstance(record, dict):
         raise ContentMarkingError(
@@ -253,7 +256,7 @@ def build_assertion(
 
     data: dict[str, Any] = {
         "version": ASSERTION_VERSION,
-        "record": {"url": url, "hash": _digest(bytes(record_bytes), alg), "alg": alg},
+        "record": {"url": url, "hash": _digest(record_bytes, alg), "alg": alg},
         "subject": subject,
         "eat_profile": profile,
     }
@@ -270,6 +273,8 @@ def verify_assertion(
     *record_bytes* is not optional and has no default. The hash is the binding,
     and an assertion whose hash was never checked is a URL in a file.
 
+    Hashing and parsing use one immutable snapshot of bytes or bytearray input.
+
     Returns the parsed record on success. Raises :class:`ContentMarkingError` or
     :class:`RecordMismatch` otherwise.
 
@@ -348,7 +353,9 @@ def verify_assertion(
             "somebody else's server and would be false."
         )
 
-    actual = _digest(bytes(record_bytes), alg)
+    # A caller may mutate a bytearray between hashing and JSON decoding.
+    record_bytes = bytes(record_bytes)
+    actual = _digest(record_bytes, alg)
     if actual != expected:
         raise RecordMismatch(
             f"the record at {url} does not match the assertion: computed {actual}, "
@@ -365,7 +372,7 @@ def verify_assertion(
         )
     for field in ("subject", "eat_profile"):
         if field not in record:
-            raise RecordMismatch(
+            raise ContentMarkingError(
                 f"the record at {url} has no {field}, which section 2 marks required, so "
                 "the assertion has nothing to be bound to. It matched the declared hash, "
                 "so this is what that URL is serving."
diff --git a/tests/test_content_marking.py b/tests/test_content_marking.py
index 1e5efd0d..5384c3cb 100644
--- a/tests/test_content_marking.py
+++ b/tests/test_content_marking.py
@@ -308,13 +308,14 @@ def test_an_assertion_missing_a_required_field_does_not_accuse_the_server(field:
 
 
 @pytest.mark.parametrize("field", ["subject", "eat_profile"])
-def test_a_record_missing_a_required_field_is_a_record_mismatch(field: str) -> None:
-    """Here the URL really is serving something that is not a conformant record."""
+def test_a_hash_matching_record_missing_a_required_field_is_malformed(field: str) -> None:
+    """Matching bytes with missing fields are malformed, rather than substituted."""
     record = _record()
     del record[field]
     assertion, record_bytes = _pair(record)
-    with pytest.raises(RecordMismatch, match="required"):
+    with pytest.raises(ContentMarkingError, match="required") as excinfo:
         verify_assertion(assertion, record_bytes)
+    assert not isinstance(excinfo.value, RecordMismatch)
 
 
 def test_a_complete_pair_still_verifies() -> None:
diff --git a/tests/test_content_marking_consumer_boundary.py b/tests/test_content_marking_consumer_boundary.py
index 78e0aacb..a8d4d9a2 100644
--- a/tests/test_content_marking_consumer_boundary.py
+++ b/tests/test_content_marking_consumer_boundary.py
@@ -1,4 +1,4 @@
-"""Consumer and RFC 8259 UTF-8 / JSON boundary tests for content_marking (#555)."""
+"""Consumer and RFC 8259 UTF-8 / JSON boundary tests for content_marking."""
 
 from __future__ import annotations
 
diff --git a/tests/test_content_marking_review_controls.py b/tests/test_content_marking_review_controls.py
index 991084ac..d0341445 100644
--- a/tests/test_content_marking_review_controls.py
+++ b/tests/test_content_marking_review_controls.py
@@ -1,4 +1,4 @@
-"""Public-API paired regressions offered as a companion to PR #565.
+"""Public-API paired regressions for content-marking boundaries.
 
 These preserve current behavior without deciding the author's pending error-message,
 profile-documentation or fetched-record validation choices.
diff --git a/tests/test_content_marking_snapshot.py b/tests/test_content_marking_snapshot.py
new file mode 100644
index 00000000..16513563
--- /dev/null
+++ b/tests/test_content_marking_snapshot.py
@@ -0,0 +1,80 @@
+"""Immutable record snapshots and malformed-record error classification."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+
+import pytest
+
+from agentrust_trace import content_marking as marking
+
+URL = "https://registry.example/records/snapshot.json"
+RECORD = {"subject": "did:web:example.com", "eat_profile": "trace", "note": "original"}
+RAW = json.dumps(RECORD).encode()
+CHANGED = json.dumps(RECORD | {"note": "changed"}).encode()
+
+
+@pytest.mark.parametrize("alg", ["sha256", "sha384"])
+def test_build_parses_and_hashes_one_snapshot(alg: str, monkeypatch: pytest.MonkeyPatch) -> None:
+    expected = marking.build_assertion(RAW, url=URL, alg=alg)
+    mutable = bytearray(RAW)
+    original_loads = marking._loads
+
+    def mutate_after_parse(data: bytes | bytearray, what: str) -> object:
+        parsed = original_loads(data, what)
+        mutable[:] = CHANGED
+        return parsed
+
+    # A deterministic caller mutation at the parse/hash seam replaces a flaky
+    # thread race. The original buffer changes, but the built hash must not.
+    monkeypatch.setattr(marking, "_loads", mutate_after_parse)
+    actual = marking.build_assertion(mutable, url=URL, alg=alg)
+    assert bytes(mutable) == CHANGED
+    assert actual == expected
+    assert marking.verify_assertion(actual, RAW) == RECORD
+    with pytest.raises(marking.RecordMismatch):
+        marking.verify_assertion(actual, CHANGED)
+
+
+@pytest.mark.parametrize("alg", ["sha256", "sha384"])
+def test_verify_hashes_and_parses_one_snapshot(alg: str, monkeypatch: pytest.MonkeyPatch) -> None:
+    assertion = marking.build_assertion(RAW, url=URL, alg=alg)
+    assert marking.verify_assertion(assertion, bytearray(RAW)) == RECORD
+    mutable = bytearray(RAW)
+    original_digest = marking._digest
+
+    def mutate_after_hash(data: bytes | bytearray, algorithm: str) -> str:
+        digest = original_digest(data, algorithm)
+        mutable[:] = CHANGED
+        return digest
+
+    monkeypatch.setattr(marking, "_digest", mutate_after_hash)
+    actual = marking.verify_assertion(assertion, mutable)
+    assert bytes(mutable) == CHANGED
+    assert actual == RECORD
+    with pytest.raises(marking.RecordMismatch):
+        marking.verify_assertion(assertion, CHANGED)
+
+
+@pytest.mark.parametrize("alg", ["sha256", "sha384"])
+@pytest.mark.parametrize("field", ["subject", "eat_profile"])
+@pytest.mark.parametrize("missing", [False, True], ids=["null", "absent"])
+def test_hash_matching_malformed_record_is_not_a_mismatch(
+    alg: str, field: str, missing: bool
+) -> None:
+    assertion = marking.build_assertion(RAW, url=URL, alg=alg)
+    assert marking.verify_assertion(assertion, RAW) == RECORD
+    bad: dict[str, object] = dict(RECORD)
+    if missing:
+        del bad[field]
+    else:
+        bad[field] = None
+    raw = json.dumps(bad).encode()
+    # First retain the stale hash: substituted bytes remain a mismatch.
+    with pytest.raises(marking.RecordMismatch):
+        marking.verify_assertion(assertion, raw)
+    assertion["data"]["record"]["hash"] = f"{alg}:" + hashlib.new(alg, raw).hexdigest()
+    with pytest.raises(marking.ContentMarkingError) as error:
+        marking.verify_assertion(assertion, raw)
+    assert type(error.value) is marking.ContentMarkingError