fix(content-marking): establish presence of the required binding fields (#335)

verify_assertion() compared the duplicated binding fields with
`record.get("subject") != data.get("subject")` and the same line for
eat_profile. A comparison establishes that two reads agree, not that either
exists, and two absences compare equal. A peer-produced assertion omitting
data.subject, paired with a hash-matching record that also omitted subject,
agreed by mutual absence and the function returned the parsed record as a
successful binding.

spec/content-marking-v1.md section 2 marks both fields required and section 6
says a conforming consumer checks both against the fetched record. This layer
performs only the binding check and returns before any Trust Record signature
or schema verification, and a caller is allowed to run it on its own, so it has
to establish its own required shape rather than relying on a later verifier.

Presence is now checked on both sides. An assertion missing a required field is
ContentMarkingError, because a malformed assertion is the caller's own input and
RecordMismatch would point the reader at whoever serves the URL, which is the
reasoning test_an_int_no_longer_reports_a_record_mismatch already pins. A record
missing one is RecordMismatch, because it matched the declared hash and that URL
really is serving something that is not a conformant record. Two present values
that disagree are unchanged.

Regression coverage carries all six cases from the reproduction, including the
two single-side controls that make the hole precisely mutual absence, plus a
complete-pair control. The pair helper recomputes the hash, without which every
case would fail at the digest check and pass for the wrong reason.

Reported by @altrudev in #326, reproduced independently by @lywinged with the
six-case matrix and the check against #325's head.


Claude-Session: https://claude.ai/code/session_01X27MBo6tGVw1UmmchuGFnW

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: imran-siddique

CHANGELOG.md

@@ -13,6 +13,8 @@ Format: [Semantic Versioning](https://semver.org/). Spec versions follow `MAJOR.
 
 ### Fixed
 
+- **`content_marking.verify_assertion()` established that the duplicated binding fields agreed, not that they existed.** `record.get("subject") != data.get("subject")` and the same line for `eat_profile` compare two reads, and two absences compare equal. A peer-produced assertion omitting `data.subject`, paired with a hash-matching record that also omitted `subject`, agreed by mutual absence and `verify_assertion` returned the parsed record as a successful binding. `spec/content-marking-v1.md` section 2 marks both fields required and section 6 says a conforming consumer checks both against the fetched record, so this layer has to establish its own required shape: the function performs only the binding check and returns before any Trust Record signature or schema verification, and a caller is allowed to run it on its own. Presence is now checked for each field on both sides. An assertion missing one is `ContentMarkingError`, because a malformed assertion is the caller's own input and `RecordMismatch` would point the reader at the server serving the URL, the same reasoning `test_an_int_no_longer_reports_a_record_mismatch` already pins. A record missing one is `RecordMismatch`, because it matched the declared hash and that URL really is serving something that is not a conformant record. Two present values that disagree are unchanged. The only behaviour change for input that was already refused is the class on an assertion-side omission, from `RecordMismatch` to its `ContentMarkingError` parent, which no caller catching the documented contract loses. Regression coverage carries all six cases from the reproduction, including the two single-side controls that make the hole precisely mutual absence rather than something wider, and a complete-pair control. Reported by @altrudev in #326, reproduced independently by @lywinged with the six-case matrix and the check against #325's head.
+
 - **`provenance.build_record()` coerced an explicitly supplied `issued_at` before the validator that exists to inspect it ever ran.** `stamped_at = int(issued_at if issued_at is not None else time.time())` handed the converted value to `_check_structure()`, whose guard carries the comment "bool is an int subclass, and True would otherwise pass as a timestamp". The order defeated that guard: `True` arrived as `1`, `False` as `0`, `1.9` as `1`, `"123"` as `123`, and `-0.5` as `0`, so every one of them satisfied the non-negative-integer test and was written into the record. The last case is the diagnostic one, since a negative non-integer became an accepted non-negative timestamp. The same line leaked two exception classes the module does not document: `issued_at=[1]` left `build_record` as a `TypeError` and `issued_at="abc"` as a `ValueError`, where every other public function in the module is held to `ProvenanceError`. An explicitly supplied value now reaches `_check_structure()` untouched and only an omitted one is stamped with `int(time.time())`, which puts `isinstance` in front of the conversion and closes both at once. This is distinct from #142 and #146: those moved the structural rules into the shared helper, and the helper was always strict. The caller path defeated it by normalizing first. Nothing that used to produce a valid record stops doing so; a valid integer is carried through unchanged and an omitted value is still stamped. `tests/test_public_functions_raise_what_they_document.py` listed `provenance.build_record` under `NO_ARGUMENT_TO_SWEEP` because it has no positional argument, which is why the junk matrix never reached it, so the function is now wired into that sweep with `issued_at` as the varied argument and a witness that pins the sweep actually arrives. Reported by @altrudev in #320, with the two undocumented exception classes and the reason the sweep never saw them found by @lywinged.
 
 - **`intent_bridge.verify_bridge()` compared the required transcript's call to the executed call with host-language equality, which is not the identity relation the bridge digests under.** Every other comparison in that function is over RFC 8785 canonical bytes, but the transcript binding used `before.get("tool_call") != tool_call`. Python holds `True == 1` and `False == 0`, nested objects included, so a transcript whose `before.tool_call` substituted a boolean for the corresponding integer (or the reverse) had different JCS bytes from the call the authorization digested and was still reported as bound to the execution. The signed `tool_call_digest` was never affected: it is checked against the actual `tool_call`, so the executed call could not differ from the authorized one. What could differ was the separately supplied transcript, in the one place whose purpose is to show that the two agree. The comparison is now `compare_digest` over `digest_jcs`, reusing the digest already computed for the `tool_call_digest` check. The isinstance guard stays in front of it, and a `transcript.before.tool_call` that JCS has no form for raises `AuthorizationMismatch` rather than `IntentBridgeError`, so the documented result class for a malformed transcript is unchanged. Regression coverage pins all four substitutions with an assertion that each one is Python-equal, so a test that stopped exercising the defect would fail rather than pass quietly, plus an unchanged-call control and a parametrized check that a non-object transcript call keeps its original exception class. No wire format, schema, scope, digest definition, or normative bridge semantics change. Reported by @altrudev in #317.

src/agentrust_trace/content_marking.py

@@ -189,6 +189,19 @@ def verify_assertion(assertion: dict[str, Any], record_bytes: bytes) -> dict[str
             "An unknown version is rejected rather than parsed best-effort."
         )
 
+    # Section 2 marks both of these required. The comparisons further down establish
+    # agreement, not presence, so without this two absences read as a match (#326).
+    # A malformed assertion is the caller's own input: it is ContentMarkingError rather
+    # than RecordMismatch, which would accuse the server at the URL of serving the wrong
+    # record.
+    for field in ("subject", "eat_profile"):
+        if field not in data:
+            raise ContentMarkingError(
+                f"assertion data has no {field}, which section 2 marks required. The "
+                "binding check compares it against the fetched record, and a comparison "
+                "establishes agreement rather than presence."
+            )
+
     ref = data.get("record")
     if not isinstance(ref, dict):
         raise ContentMarkingError("assertion carries no record reference")
@@ -231,6 +244,13 @@ def verify_assertion(assertion: dict[str, Any], record_bytes: bytes) -> dict[str
             f"{type(record).__name__}. It matched the declared hash, so this is what the "
             "record actually is at that URL, not a mismatch to report as RecordMismatch."
         )
+    for field in ("subject", "eat_profile"):
+        if field not in record:
+            raise RecordMismatch(
+                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."
+            )
     if record.get("subject") != data.get("subject"):
         raise RecordMismatch(
             f"assertion names subject {data.get('subject')!r} and the record says "

tests/test_content_marking.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 
 import pytest
@@ -246,3 +247,78 @@ def test_empty_bytes_are_refused_like_build_assertion_refuses_them() -> None:
     a = build_assertion(_bytes(_record()), url=URL)
     with pytest.raises(ContentMarkingError, match="record_bytes must be the bytes retrieved"):
         verify_assertion(a, b"")
+
+
+# --- presence, not just agreement (#326) -----------------------------------
+#
+# `verify_assertion` compared the duplicated binding fields with `.get()` equality,
+# which establishes that two values agree and not that either exists. A peer-produced
+# assertion omitting `data.subject`, paired with a hash-matching record that also
+# omitted `subject`, compared `None` against `None` and verified.
+#
+# `build_assertion` refuses to build from an incomplete record, so these pairs are
+# constructed by hand: the defect is in what a consumer accepts from a peer, not in
+# what this producer emits.
+
+
+def _pair(record: dict, drop_from_assertion: tuple[str, ...] = ()) -> tuple[dict, bytes]:
+    """An assertion whose hash genuinely matches *record*, so the pair is self-consistent.
+
+    Without recomputing the hash the pair fails at the digest check and the presence
+    check is never reached, which would make every test below pass for the wrong reason.
+    """
+    record_bytes = _bytes(record)
+    assertion = build_assertion(_bytes(_record()), url=URL)
+    assertion["data"]["record"]["hash"] = (
+        "sha256:" + hashlib.sha256(record_bytes).hexdigest()
+    )
+    for field in drop_from_assertion:
+        assert field in assertion["data"], "the field must be present for its removal to count"
+        del assertion["data"][field]
+    return assertion, record_bytes
+
+
+@pytest.mark.parametrize("field", ["subject", "eat_profile"])
+def test_a_field_absent_from_both_sides_is_not_a_binding(field: str) -> None:
+    record = _record()
+    del record[field]
+    assertion, record_bytes = _pair(record, drop_from_assertion=(field,))
+    with pytest.raises(ContentMarkingError, match="required"):
+        verify_assertion(assertion, record_bytes)
+
+
+def test_both_fields_absent_from_both_sides_is_not_a_binding() -> None:
+    record = _record()
+    del record["subject"], record["eat_profile"]
+    assertion, record_bytes = _pair(record, drop_from_assertion=("subject", "eat_profile"))
+    with pytest.raises(ContentMarkingError, match="required"):
+        verify_assertion(assertion, record_bytes)
+
+
+@pytest.mark.parametrize("field", ["subject", "eat_profile"])
+def test_an_assertion_missing_a_required_field_does_not_accuse_the_server(field: str) -> None:
+    """The caller's assertion is malformed, so the reader must not be pointed at the URL.
+
+    Same principle as `test_an_int_no_longer_reports_a_record_mismatch`.
+    """
+    assertion, record_bytes = _pair(_record(), drop_from_assertion=(field,))
+    with pytest.raises(ContentMarkingError) as excinfo:
+        verify_assertion(assertion, record_bytes)
+    assert not isinstance(excinfo.value, RecordMismatch)
+
+
+@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."""
+    record = _record()
+    del record[field]
+    assertion, record_bytes = _pair(record)
+    with pytest.raises(RecordMismatch, match="required"):
+        verify_assertion(assertion, record_bytes)
+
+
+def test_a_complete_pair_still_verifies() -> None:
+    """The control: presence checks must refuse nothing that used to bind."""
+    record = _record()
+    assertion = build_assertion(_bytes(record), url=URL)
+    assert verify_assertion(assertion, _bytes(record)) == record