diff --git a/src/agentrust_trace/content_marking.py b/src/agentrust_trace/content_marking.py index 4ea1eee..e9ef689 100644 --- a/src/agentrust_trace/content_marking.py +++ b/src/agentrust_trace/content_marking.py @@ -21,6 +21,8 @@ import re from typing import Any +from rfc3986_validator import validate_rfc3986 # type: ignore[import-untyped] + __all__ = [ "ASSERTION_LABEL", "ASSERTION_VERSION", @@ -37,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): @@ -58,6 +61,48 @@ 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.""" + 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) + 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) + return value + + def build_assertion( record_bytes: bytes, *, @@ -79,6 +124,7 @@ def build_assertion( ) if not url: raise ContentMarkingError("url is required: an assertion with no reference binds nothing") + url = _record_url(url) import json @@ -143,8 +189,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 +203,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 +214,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 +223,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." ) 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..b378132 --- /dev/null +++ b/tests/test_content_marking_record_url_boundary.py @@ -0,0 +1,64 @@ +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) + + +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", 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)