fix(content-marking): validate record.url external references

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<authority>[^/?#]+)(?:[/?#].*)?$", 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."
         )

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)