From af3f4c1b55e33b0a3b0123e9f020d84af0a944e6 Mon Sep 17 00:00:00 2001 From: "Altru.dev" Date: Fri, 4 Sep 2026 06:20:54 -0700 Subject: [PATCH 1/7] fix(content-marking): validate record URL boundary --- src/agentrust_trace/content_marking.py | 28 +++++++++++++++++++------- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py index 4ea1eee..c29fc58 100644 --- a/src/agentrust_trace/content_marking.py +++ b/src/agentrust_trace/content_marking.py @@ -20,6 +20,7 @@ import hashlib import re from typing import Any +from urllib.parse import urlsplit __all__ = [ "ASSERTION_LABEL", @@ -58,6 +59,19 @@ def _digest(data: bytes, alg: str) -> str: return f"{alg}:{_ALGS[alg](data).hexdigest()}" +def _record_url(value: Any) -> str: + """Return a C2PA external-reference URL or refuse the malformed value.""" + if not isinstance(value, str) or not value or any(ch.isspace() for ch in value): + raise ContentMarkingError("record.url must be an absolute http(s) URI") + try: + parsed = urlsplit(value) + except ValueError as exc: + raise ContentMarkingError("record.url must be an absolute http(s) URI") from exc + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ContentMarkingError("record.url must be an absolute http(s) URI") + return value + + def build_assertion( record_bytes: bytes, *, @@ -77,8 +91,7 @@ def build_assertion( "record_bytes must be the serialized record as it will be served. A hash " "over a re-serialized object is a hash of bytes nobody will fetch." ) - if not url: - raise ContentMarkingError("url is required: an assertion with no reference binds nothing") + url = _record_url(url) import json @@ -143,8 +156,9 @@ def verify_assertion(assertion: dict[str, Any], record_bytes: bytes) -> dict[str ) ref = data.get("record") - if not isinstance(ref, dict) or not ref.get("url"): + if not isinstance(ref, dict): raise ContentMarkingError("assertion carries no record reference") + url = _record_url(ref.get("url")) alg = ref.get("alg") if not isinstance(alg, str): raise ContentMarkingError( @@ -156,7 +170,7 @@ def verify_assertion(assertion: dict[str, Any], record_bytes: bytes) -> dict[str if not isinstance(record_bytes, bytes | bytearray) or not record_bytes: raise ContentMarkingError( - f"record_bytes must be the bytes retrieved from {ref['url']}, got " + f"record_bytes must be the bytes retrieved from {url}, got " f"{type(record_bytes).__name__}. `build_assertion` already refuses this and " "the reason it matters more here is that `bytes(5)` is five zero bytes: an " "int would be hashed, would not match, and the caller would be told the " @@ -167,7 +181,7 @@ def verify_assertion(assertion: dict[str, Any], record_bytes: bytes) -> dict[str actual = _digest(bytes(record_bytes), alg) if actual != expected: raise RecordMismatch( - f"the record at {ref['url']} does not match the assertion: computed {actual}, " + f"the record at {url} does not match the assertion: computed {actual}, " f"assertion says {expected}. The record changed after the asset was signed, or " "the URL is serving a different one." ) @@ -176,10 +190,10 @@ def verify_assertion(assertion: dict[str, Any], record_bytes: bytes) -> dict[str try: record = json.loads(record_bytes) except ValueError as exc: - raise ContentMarkingError(f"record at {ref['url']} is not JSON: {exc}") from exc + raise ContentMarkingError(f"record at {url} is not JSON: {exc}") from exc if not isinstance(record, dict): raise ContentMarkingError( - f"the record at {ref['url']} must decode to a JSON object, got " + f"the record at {url} must decode to a JSON object, got " 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." ) From ef377a3da6ddc9b9f3e3add18f5733a184ca7136 Mon Sep 17 00:00:00 2001 From: "Altru.dev" Date: Fri, 4 Sep 2026 06:21:05 -0700 Subject: [PATCH 2/7] test(content-marking): hold record URL boundary --- ...est_content_marking_record_url_boundary.py | 68 +++++++++++++++++++ 1 file changed, 68 insertions(+) create mode 100644 tests/test_content_marking_record_url_boundary.py diff --git a/tests/test_content_marking_record_url_boundary.py b/tests/test_content_marking_record_url_boundary.py new file mode 100644 index 0000000..5cc7dff --- /dev/null +++ b/tests/test_content_marking_record_url_boundary.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +import json + +import pytest + +from agentrust_trace.content_marking import ContentMarkingError, build_assertion, verify_assertion + +HTTPS_URL = "https://registry.example/records/abc123.json" +HTTP_URL = "http://registry.example/records/abc123.json" + + +def _record_bytes() -> bytes: + return json.dumps( + { + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", + "iat": 1760000000, + "subject": "spiffe://example.org/agent/image-bot", + "data_class": "public", + } + ).encode() + + +@pytest.mark.parametrize("url", [HTTPS_URL, HTTP_URL]) +def test_http_and_https_record_urls_are_accepted(url: str) -> None: + raw = _record_bytes() + assertion = build_assertion(raw, url=url) + assert assertion["data"]["record"]["url"] == url + assert verify_assertion(assertion, raw) + + +@pytest.mark.parametrize( + "bad_url", + [ + True, + 1, + [1], + {"x": 1}, + " ", + "not a url", + "ftp://registry.example/records/abc123.json", + "https:///records/abc123.json", + ], +) +def test_build_assertion_refuses_values_outside_the_record_url_boundary(bad_url) -> None: + with pytest.raises(ContentMarkingError, match=r"record\.url must be an absolute http\(s\) URI"): + build_assertion(_record_bytes(), url=bad_url) + + +@pytest.mark.parametrize( + "bad_url", + [ + True, + 1, + [1], + {"x": 1}, + " ", + "not a url", + "ftp://registry.example/records/abc123.json", + "https:///records/abc123.json", + ], +) +def test_verify_assertion_refuses_values_outside_the_record_url_boundary(bad_url) -> None: + raw = _record_bytes() + assertion = build_assertion(raw, url=HTTPS_URL) + assertion["data"]["record"]["url"] = bad_url + with pytest.raises(ContentMarkingError, match=r"record\.url must be an absolute http\(s\) URI"): + verify_assertion(assertion, raw) From 80c14d7a0a8322fea9a6867915a9c151f3d9bcfd Mon Sep 17 00:00:00 2001 From: "Altru.dev" Date: Fri, 4 Sep 2026 06:23:31 -0700 Subject: [PATCH 3/7] fix(content-marking): harden record URL authority validation --- src/agentrust_trace/content_marking.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py index c29fc58..0d4aa3e 100644 --- a/src/agentrust_trace/content_marking.py +++ b/src/agentrust_trace/content_marking.py @@ -61,14 +61,18 @@ def _digest(data: bytes, alg: str) -> str: def _record_url(value: Any) -> str: """Return a C2PA external-reference URL or refuse the malformed value.""" + message = "record.url must be an absolute http(s) URI" if not isinstance(value, str) or not value or any(ch.isspace() for ch in value): - raise ContentMarkingError("record.url must be an absolute http(s) URI") + raise ContentMarkingError(message) try: parsed = urlsplit(value) + port = parsed.port except ValueError as exc: - raise ContentMarkingError("record.url must be an absolute http(s) URI") from exc - if parsed.scheme not in {"http", "https"} or not parsed.netloc: - raise ContentMarkingError("record.url must be an absolute http(s) URI") + raise ContentMarkingError(message) from exc + if parsed.scheme not in {"http", "https"} or parsed.hostname is None: + raise ContentMarkingError(message) + # Reading parsed.port above is intentional: urllib rejects malformed/non-numeric ports there. + _ = port return value From ace43b54a8f8dcfb5f7ab6f338d49fb46f446f88 Mon Sep 17 00:00:00 2001 From: "Altru.dev" Date: Fri, 4 Sep 2026 06:23:45 -0700 Subject: [PATCH 4/7] test(content-marking): cover malformed record URL authorities --- ...est_content_marking_record_url_boundary.py | 48 +++++++++---------- 1 file changed, 22 insertions(+), 26 deletions(-) diff --git a/tests/test_content_marking_record_url_boundary.py b/tests/test_content_marking_record_url_boundary.py index 5cc7dff..b378132 100644 --- a/tests/test_content_marking_record_url_boundary.py +++ b/tests/test_content_marking_record_url_boundary.py @@ -29,40 +29,36 @@ def test_http_and_https_record_urls_are_accepted(url: str) -> None: assert verify_assertion(assertion, raw) -@pytest.mark.parametrize( - "bad_url", - [ - True, - 1, - [1], - {"x": 1}, - " ", - "not a url", - "ftp://registry.example/records/abc123.json", - "https:///records/abc123.json", - ], -) +BAD_URLS = [ + True, + 1, + [1], + {"x": 1}, + " ", + "not a url", + "ftp://registry.example/records/abc123.json", + "https:///records/abc123.json", + "https://@/records/abc123.json", + "https://registry.example:bad/records/abc123.json", +] + + +@pytest.mark.parametrize("bad_url", BAD_URLS) def test_build_assertion_refuses_values_outside_the_record_url_boundary(bad_url) -> None: with pytest.raises(ContentMarkingError, match=r"record\.url must be an absolute http\(s\) URI"): build_assertion(_record_bytes(), url=bad_url) -@pytest.mark.parametrize( - "bad_url", - [ - True, - 1, - [1], - {"x": 1}, - " ", - "not a url", - "ftp://registry.example/records/abc123.json", - "https:///records/abc123.json", - ], -) +@pytest.mark.parametrize("bad_url", BAD_URLS) def test_verify_assertion_refuses_values_outside_the_record_url_boundary(bad_url) -> None: raw = _record_bytes() assertion = build_assertion(raw, url=HTTPS_URL) assertion["data"]["record"]["url"] = bad_url with pytest.raises(ContentMarkingError, match=r"record\.url must be an absolute http\(s\) URI"): verify_assertion(assertion, raw) + + +def test_truthiness_only_mutation_would_reopen_the_boundary() -> None: + """Truthy malformed values are the regression class, not just empty URLs.""" + for bad_url in [True, 1, [1], {"x": 1}, "not a url"]: + assert bool(bad_url) From 0a75dd6070277423410afd890c6f1c9b19a29606 Mon Sep 17 00:00:00 2001 From: "Altru.dev" Date: Fri, 4 Sep 2026 06:25:05 -0700 Subject: [PATCH 5/7] fix(content-marking): preserve empty-url diagnostic --- src/agentrust_trace/content_marking.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py index 0d4aa3e..851db7a 100644 --- a/src/agentrust_trace/content_marking.py +++ b/src/agentrust_trace/content_marking.py @@ -95,6 +95,8 @@ def build_assertion( "record_bytes must be the serialized record as it will be served. A hash " "over a re-serialized object is a hash of bytes nobody will fetch." ) + if not url: + raise ContentMarkingError("url is required: an assertion with no reference binds nothing") url = _record_url(url) import json From 8ab1c8c6e2b1af869824d3440d4ae0e43a3543a3 Mon Sep 17 00:00:00 2001 From: "Altru.dev" Date: Fri, 4 Sep 2026 06:55:21 -0700 Subject: [PATCH 6/7] fix(content-marking): keep URL validation offline-safe --- src/agentrust_trace/content_marking.py | 45 ++++++++++++++++++++------ 1 file changed, 36 insertions(+), 9 deletions(-) diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py index 851db7a..2c79bab 100644 --- a/src/agentrust_trace/content_marking.py +++ b/src/agentrust_trace/content_marking.py @@ -20,7 +20,8 @@ import hashlib import re from typing import Any -from urllib.parse import urlsplit + +from rfc3986_validator import validate_rfc3986 __all__ = [ "ASSERTION_LABEL", @@ -38,6 +39,7 @@ _ALGS = {"sha256": hashlib.sha256, "sha384": hashlib.sha384} _SUBJECT_RE = re.compile(r"^(spiffe://[^/]+/.+|did:[a-z0-9]+:.+)$") _DIGEST_RE = re.compile(r"^sha(256:[0-9a-f]{64}|384:[0-9a-f]{96})$") +_HTTP_URL_RE = re.compile(r"^https?://(?P[^/?#]+)(?:[/?#].*)?$", re.IGNORECASE) class ContentMarkingError(ValueError): @@ -64,15 +66,40 @@ def _record_url(value: Any) -> str: message = "record.url must be an absolute http(s) URI" if not isinstance(value, str) or not value or any(ch.isspace() for ch in value): raise ContentMarkingError(message) - try: - parsed = urlsplit(value) - port = parsed.port - except ValueError as exc: - raise ContentMarkingError(message) from exc - if parsed.scheme not in {"http", "https"} or parsed.hostname is None: + if validate_rfc3986(value) is None: + raise ContentMarkingError(message) + + match = _HTTP_URL_RE.fullmatch(value) + if match is None: + raise ContentMarkingError(message) + + authority = match.group("authority") + hostport = authority.rsplit("@", 1)[-1] + if not hostport: + raise ContentMarkingError(message) + + port: str | None = None + if hostport.startswith("["): + close = hostport.find("]") + if close <= 1: + raise ContentMarkingError(message) + tail = hostport[close + 1 :] + if tail: + if not tail.startswith(":") or not tail[1:].isdigit(): + raise ContentMarkingError(message) + port = tail[1:] + else: + if hostport.count(":") > 1: + raise ContentMarkingError(message) + if ":" in hostport: + host, port = hostport.rsplit(":", 1) + if not host or not port.isdigit(): + raise ContentMarkingError(message) + elif not hostport: + raise ContentMarkingError(message) + + if port is not None and int(port) > 65535: raise ContentMarkingError(message) - # Reading parsed.port above is intentional: urllib rejects malformed/non-numeric ports there. - _ = port return value From e15651d9f6a23c04f2d07e111563b9e76e36bd6e Mon Sep 17 00:00:00 2001 From: "Altru.dev" Date: Fri, 4 Sep 2026 06:57:31 -0700 Subject: [PATCH 7/7] fix(content-marking): declare untyped URI validator import --- src/agentrust_trace/content_marking.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py index 2c79bab..e9ef689 100644 --- a/src/agentrust_trace/content_marking.py +++ b/src/agentrust_trace/content_marking.py @@ -21,7 +21,7 @@ import re from typing import Any -from rfc3986_validator import validate_rfc3986 +from rfc3986_validator import validate_rfc3986 # type: ignore[import-untyped] __all__ = [ "ASSERTION_LABEL",