fix(security): strict signature decoding, documented refusals, ClusterFuzzLite

.clusterfuzzlite/Dockerfile

@@ -0,0 +1,5 @@
+FROM gcr.io/oss-fuzz-base/base-builder-python
+
+COPY . $SRC/trace-spec
+COPY .clusterfuzzlite/build.sh $SRC/build.sh
+WORKDIR $SRC/trace-spec

.clusterfuzzlite/build.sh

@@ -0,0 +1,32 @@
+#!/bin/bash -eu
+# Build the fuzz targets for ClusterFuzzLite.
+#
+# Third-party dependencies come from the same hash-pinned lock CI uses, and the
+# package then goes in with --no-deps, so the targets run against the versions
+# the test suite runs against rather than whatever resolves on the day.
+
+cd "$SRC/trace-spec"
+pip3 install --no-cache-dir --require-hashes -r requirements/dev.txt
+pip3 install --no-cache-dir --no-deps .
+
+# compile_python_fuzzer bundles each target with PyInstaller, which follows
+# static imports only. The cryptography and pydantic stacks reach email.mime
+# lazily, so without this the bundled target dies at runtime with
+# "ModuleNotFoundError: No module named 'email.mime'" and libFuzzer reports it
+# as a crash in the target. The package schemas are data files PyInstaller does
+# not see through importlib.resources, and neither are the metaschemas jsonschema
+# loads the same way, so both are collected explicitly.
+PYI_ARGS=(
+  --collect-submodules=email
+  --collect-data=agentrust_trace
+  --collect-data=jsonschema_specifications
+)
+
+for target in "$SRC"/trace-spec/.clusterfuzzlite/fuzz_*.py; do
+  compile_python_fuzzer "$target" "${PYI_ARGS[@]}"
+done
+
+# Seed corpora: the committed example records and bundles, plus one valid input
+# per target built with the library itself, so the fuzzer starts past the
+# schema instead of spending its budget discovering JSON.
+python3 .clusterfuzzlite/make_seeds.py "$OUT"

.clusterfuzzlite/fuzz_canonicalize.py

@@ -0,0 +1,87 @@
+#!/usr/bin/python3
+"""Fuzz the two canonicalizations: JCS for signatures, anchor form for digests.
+
+Input: JSON text. ``sign._canonical_bytes`` is the signature pre-image of spec
+section 3.2.2 and ``sign.anchor_bytes`` is registry-anchor-v1 section 1.
+
+Properties, stronger than "does not crash":
+
+- Round trip. Parsing either output must reproduce the input, up to JSON's
+  single number type: a float whose shortest form has no fraction re-parses as
+  an int, which is correct RFC 8785 output. A canonicalization that silently
+  drops or merges a member still produces bytes somebody signs.
+- Determinism. The same value canonicalizes to the same bytes twice.
+- Declared refusals only. JCS refuses with ``rfc8785.CanonicalizationError``
+  (a ``ValueError``) and the anchor form with ``UnanchorableValue``.
+"""
+
+import json
+import math
+import sys
+
+import atheris
+
+with atheris.instrument_imports():
+    import rfc8785
+
+    from agentrust_trace.sign import UnanchorableValue, _canonical_bytes, anchor_bytes
+
+
+def _json_equal(a, b) -> bool:
+    if isinstance(a, bool) or isinstance(b, bool):
+        return a is b
+    if isinstance(a, (int, float)) and isinstance(b, (int, float)):
+        return float(a) == float(b)
+    if isinstance(a, dict) and isinstance(b, dict):
+        return a.keys() == b.keys() and all(_json_equal(a[k], b[k]) for k in a)
+    if isinstance(a, list) and isinstance(b, list):
+        return len(a) == len(b) and all(_json_equal(x, y) for x, y in zip(a, b, strict=True))
+    return type(a) is type(b) and a == b
+
+
+def _has_nonfinite(value) -> bool:
+    if isinstance(value, float):
+        return not math.isfinite(value)
+    if isinstance(value, dict):
+        return any(_has_nonfinite(v) for v in value.values())
+    if isinstance(value, list):
+        return any(_has_nonfinite(v) for v in value)
+    return False
+
+
+def TestOneInput(data: bytes) -> None:
+    try:
+        value = json.loads(data)
+    except (ValueError, RecursionError):
+        return
+    try:
+        if not isinstance(value, dict) or _has_nonfinite(value):
+            # The signature pre-image is always an object, and Python's json
+            # accepts NaN and Infinity, which have no JSON form to return to.
+            return
+    except RecursionError:
+        return
+    try:
+        jcs = _canonical_bytes(value)
+    except (rfc8785.CanonicalizationError, RecursionError):
+        # RecursionError: nesting deeper than the interpreter stack. The
+        # verifiers refuse such documents before they reach this function.
+        jcs = None
+    if jcs is not None:
+        assert _json_equal(json.loads(jcs), value), f"JCS output did not round-trip: {jcs!r}"
+        assert _canonical_bytes(value) == jcs, "JCS is not deterministic"
+    try:
+        anchored = anchor_bytes(value)
+    except (UnanchorableValue, RecursionError):
+        return
+    assert _json_equal(json.loads(anchored), value), f"anchor form did not round-trip: {anchored!r}"
+    assert anchor_bytes(value) == anchored, "anchor form is not deterministic"
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

.clusterfuzzlite/fuzz_content_marking.py

@@ -0,0 +1,73 @@
+#!/usr/bin/python3
+"""Fuzz the C2PA content-marking assertion checker.
+
+Input: one mode byte, then a JSON object ``{"assertion": ..., "record": ...}``
+where ``record`` is the text served at the assertion's URL. With mode bit 0
+set, the assertion is rebuilt over those exact bytes with ``build_assertion``,
+so the fuzzer reaches the record parsing and subject and profile comparison
+that only run once the hash matches.
+
+Property: ``verify_assertion`` and ``build_assertion`` document
+``ContentMarkingError`` (``RecordMismatch`` is a subclass). Both the assertion
+and the served record come from whoever controls the asset and the URL, so any
+other exception is that party crashing the checker.
+"""
+
+import json
+import sys
+
+import atheris
+
+with atheris.instrument_imports():
+    from agentrust_trace.content_marking import (
+        ContentMarkingError,
+        build_assertion,
+        verify_assertion,
+    )
+
+
+def TestOneInput(data: bytes) -> None:
+    if not data:
+        return
+    mode, body = data[0], data[1:]
+    try:
+        case = json.loads(body)
+    except (ValueError, RecursionError):
+        return
+    if not isinstance(case, dict):
+        return
+    record_text = case.get("record")
+    if not isinstance(record_text, str):
+        return
+    try:
+        record_bytes = record_text.encode("utf-8")
+    except UnicodeEncodeError:
+        return
+    assertion = case.get("assertion")
+    if mode & 1 and isinstance(assertion, dict):
+        data_obj = assertion.get("data")
+        ref = data_obj.get("record") if isinstance(data_obj, dict) else None
+        url = ref.get("url") if isinstance(ref, dict) else None
+        try:
+            assertion = build_assertion(
+                record_bytes,
+                url=url if isinstance(url, str) else "https://records.example/r.json",
+                alg="sha384" if mode & 2 else "sha256",
+                anchor=assertion.get("anchor") if mode & 4 else None,
+            )
+        except ContentMarkingError:
+            return
+    try:
+        record = verify_assertion(assertion, record_bytes)
+    except ContentMarkingError:
+        return
+    assert isinstance(record, dict), type(record)
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

.clusterfuzzlite/fuzz_intent_bridge.py

@@ -0,0 +1,99 @@
+#!/usr/bin/python3
+"""Fuzz ``intent_bridge.verify_bridge`` and the successor-observation evaluator.
+
+Input: one mode byte, then a JSON object carrying the verifier's inputs:
+``bridge``, ``declaration``, ``pic_intent_digest``, ``pic_args_digest``,
+``tool_call`` and ``transcript``. With mode bit 0 set, ``bridge.authorization``
+is signed with the pinned authorizer key, so the fuzzer reaches the scope,
+digest and transcript checks behind the signature.
+
+Property: ``verify_bridge`` documents ``IntentBridgeError`` and its two
+subclasses; ``evaluate_successor_observation`` documents
+``SuccessorObservationError``. Every argument here is execution evidence from
+outside the verifier, so any other exception is that evidence crashing it. An
+accepted bridge must carry the decision ``allow``.
+"""
+
+import json
+import sys
+
+import atheris
+
+with atheris.instrument_imports():
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+    from agentrust_trace.intent_bridge import IntentBridgeError, sign_bridge, verify_bridge
+    from agentrust_trace.sign import key_to_jwk
+    from agentrust_trace.successor_observation import (
+        SuccessorObservationError,
+        evaluate_successor_observation,
+    )
+
+KEY = Ed25519PrivateKey.from_private_bytes(bytes(range(96, 128)))
+TRUSTED_JWK = {**key_to_jwk(KEY), "kid": "authorizer-key-1"}
+NOW = 150
+
+
+def _predicate(observation):
+    status = observation.get("status")
+    if not isinstance(status, str):
+        return None
+    return {"accepted": True, "rejected": False}.get(status)
+
+
+def TestOneInput(data: bytes) -> None:
+    if not data:
+        return
+    mode, body = data[0], data[1:]
+    try:
+        case = json.loads(body)
+    except (ValueError, RecursionError):
+        return
+    if not isinstance(case, dict):
+        return
+    bridge = case.get("bridge")
+    if mode & 1 and isinstance(bridge, dict):
+        try:
+            bridge = sign_bridge(bridge.get("authorization"), KEY)
+        except IntentBridgeError:
+            return
+    try:
+        authorization = verify_bridge(
+            bridge,
+            TRUSTED_JWK,
+            declaration=case.get("declaration"),
+            pic_intent_digest=case.get("pic_intent_digest"),
+            pic_args_digest=case.get("pic_args_digest"),
+            tool_call=case.get("tool_call"),
+            transcript=case.get("transcript"),
+            now=NOW,
+        )
+    except IntentBridgeError:
+        pass
+    else:
+        assert authorization["decision"] == "allow", authorization
+
+    transcript = case.get("transcript")
+    after = transcript.get("after") if isinstance(transcript, dict) else None
+    try:
+        evaluate_successor_observation(
+            after,
+            expected_successor_digest=case.get("successor_digest", "sha256:" + "0" * 64),
+            trusted_observers=["observer-1"],
+            executor_id="executor-1" if mode & 2 else None,
+            independence_required=bool(mode & 4),
+            now=NOW,
+            max_age_seconds=3600,
+            predicate=_predicate,
+        )
+    except SuccessorObservationError:
+        pass
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

.clusterfuzzlite/fuzz_provenance.py

@@ -0,0 +1,82 @@
+#!/usr/bin/python3
+"""Fuzz the MCP server provenance verifier.
+
+Input: one mode byte, then JSON text. A JSON object with a ``record`` member is
+read as ``{"record": ..., "tools": ...}``; anything else is the record, with an
+empty tool list. With mode bit 0 set the record is signed with the pinned key
+first, so the fuzzer gets past the signature to the tool-catalog comparison.
+
+Property: ``verify_record``, ``check_tool_catalog`` and ``tool_catalog_hash``
+document ``ProvenanceError`` (``ToolCatalogMismatch`` is a subclass). The
+tools are exactly what a live, untrusted server returned, so any other
+exception is that server crashing its verifier.
+"""
+
+import json
+import sys
+
+import atheris
+
+with atheris.instrument_imports():
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+    from agentrust_trace.provenance import (
+        FORMAT,
+        FORMAT_V2,
+        ProvenanceError,
+        check_tool_catalog,
+        sign_record,
+        tool_catalog_hash,
+        verify_record,
+    )
+    from agentrust_trace.sign import key_to_jwk
+
+KEY = Ed25519PrivateKey.from_private_bytes(bytes(range(64, 96)))
+TRUSTED_JWK = key_to_jwk(KEY)
+NOW = 1_785_000_000
+
+
+def TestOneInput(data: bytes) -> None:
+    if not data:
+        return
+    mode, body = data[0], data[1:]
+    try:
+        value = json.loads(body)
+    except (ValueError, RecursionError):
+        return
+    if isinstance(value, dict) and "record" in value:
+        record, tools = value.get("record"), value.get("tools", [])
+    else:
+        record, tools = value, []
+    if mode & 1 and isinstance(record, dict):
+        try:
+            record = sign_record(record, KEY)
+        except ProvenanceError:
+            return
+    try:
+        verify_record(
+            record,
+            TRUSTED_JWK,
+            now=NOW,
+            max_age_seconds=None if mode & 2 else 86400,
+            required_format=FORMAT_V2 if mode & 4 else None,
+        )
+    except ProvenanceError:
+        pass
+    try:
+        check_tool_catalog(record, tools)
+    except ProvenanceError:
+        pass
+    try:
+        tool_catalog_hash(tools, format=FORMAT_V2 if mode & 8 else FORMAT)
+    except ProvenanceError:
+        pass
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()