fix: apply ECMA regex boundaries in Trust Record validation

src/agentrust_trace/validate.py

@@ -4,11 +4,46 @@
 import importlib.resources
 import json
 import re
+from collections.abc import Iterator
 from functools import lru_cache
 from ipaddress import IPv6Address
 from typing import Any, cast
 
 import jsonschema
+from jsonschema.protocols import Validator
+
+
+# Only the four pattern forms shipped in TRACE v0.2 are adapted here. This is
+# not a general ECMA-262 translator: a new schema pattern needs explicit review.
+# None of these forms has a literal/escaped dot or dollar, or a dot in a class.
+_ECMA_PATTERNS = (
+    '^(spiffe://[^/]+/.+|did:[a-z0-9]+:.+)$',
+    '^sha(256:[0-9a-f]{64}|384:[0-9a-f]{96})$',
+    (r"^P(\d+W|(\d+Y(\d+M)?(\d+D)?|\d+M(\d+D)?|\d+D)"
+     r"(T(\d+H(\d+M)?(\d+S)?|\d+M(\d+S)?|\d+S))?"
+     r"|T(\d+H(\d+M)?(\d+S)?|\d+M(\d+S)?|\d+S))$"),
+    '^[A-Za-z0-9_-]+$',
+)
+_PYTHON_PATTERNS = {
+    pattern: re.compile(
+        pattern.replace("$", r"\Z")
+        .replace(".", r"[^\n\r\u2028\u2029]")
+        .replace(r"\d", "[0-9]")
+    )
+    for pattern in _ECMA_PATTERNS
+}
+
+
+def _ecma_pattern(
+    validator: Validator, pattern: str, instance: Any, schema: Any
+) -> Iterator[jsonschema.ValidationError]:
+    if validator.is_type(instance, "string") and not _PYTHON_PATTERNS[pattern].search(instance):
+        yield jsonschema.ValidationError(f"{instance!r} does not match {pattern!r}")
+
+
+_TraceValidator = jsonschema.validators.extend(  # type: ignore[no-untyped-call]
+    jsonschema.Draft202012Validator, {"pattern": _ecma_pattern}
+)
 
 
 @lru_cache(maxsize=1)
@@ -18,14 +53,17 @@ def _schema() -> dict[str, Any]:
 
 
 @lru_cache(maxsize=1)
-def _validator() -> jsonschema.Draft202012Validator:
-    # Keep jsonschema's URI grammar, then close its permissive embedded-IPv4 check.
-    # Use instance-local registration so other consumers' format checkers are untouched.
+def _validator() -> Validator:
     base_uri_checker = jsonschema.FormatChecker(formats=["uri"])
     checker = jsonschema.FormatChecker()
 
+    # Convert IPv6Address failures into schema validation errors.
     @checker.checks("uri", raises=ValueError)
     def is_uri(value: Any) -> bool:
+        # URI grammar excludes literal line breaks; Python regex end anchors can
+        # otherwise accept a final LF. Percent-encoded characters remain valid.
+        if isinstance(value, str) and any(c in value for c in "\n\r\u2028\u2029"):
+            return False
         if not base_uri_checker.conforms(value, "uri"):
             return False
         if isinstance(value, str):
@@ -39,7 +77,7 @@ def is_uri(value: Any) -> bool:
                         IPv6Address(literal)
         return True
 
-    return jsonschema.Draft202012Validator(_schema(), format_checker=checker)
+    return cast(Validator, _TraceValidator(_schema(), format_checker=checker))
 
 
 @lru_cache(maxsize=1)

tests/test_schema_regex_semantics.py

@@ -0,0 +1,136 @@
+"""ECMA-262 boundaries for the patterns in the packaged TRACE schema."""
+import copy
+import json
+import shutil
+import subprocess
+from pathlib import Path
+
+import jsonschema
+import pytest
+
+from agentrust_trace import generate_key, key_to_jwk, sign_record, validate_json, verify_record
+from agentrust_trace.validate import SCHEMA, iter_errors
+
+BASE = json.loads((Path(__file__).parents[1] /
+    "examples/canonicalization-boundary/01-non-ascii-values.json").read_text())["record"]
+TERMINATORS = ["\n", "\r", "\u2028", "\u2029"]
+
+
+@pytest.mark.parametrize("subject", ["spiffe://example.org/agent", "did:web:example.org"])
+@pytest.mark.parametrize("tail", TERMINATORS)
+@pytest.mark.parametrize("position", ["end", "middle"])
+def test_subject_line_terminators_are_rejected(subject, tail, position):
+    record = copy.deepcopy(BASE)
+    record["subject"] = subject + tail + ("more" if position == "middle" else "")
+    with pytest.raises(jsonschema.ValidationError):
+        validate_json(record)
+    errors = iter_errors(record)
+    assert any(list(e.path) == ["subject"] and e.validator == "pattern" for e in errors)
+
+
+@pytest.mark.parametrize("tail", TERMINATORS)
+def test_signed_subject_line_terminators_are_rejected(tail):
+    body = copy.deepcopy(BASE)
+    body["subject"] = "spiffe://example.org/agent" + tail
+    key = generate_key()
+    with pytest.raises(ValueError, match="does not conform.*subject"):
+        verify_record(sign_record(body, key), key_to_jwk(key), now=body["iat"])
+
+
+@pytest.mark.parametrize("tail", TERMINATORS)
+def test_signed_uri_line_terminators_are_rejected(tail):
+    body = copy.deepcopy(BASE)
+    body["appraisal"]["verifier"] = "https://verifier.example/v1" + tail
+    key = generate_key()
+    with pytest.raises(ValueError, match="does not conform.*appraisal[.]verifier"):
+        verify_record(sign_record(body, key), key_to_jwk(key), now=body["iat"])
+
+
+@pytest.mark.parametrize("subject", [
+    "spiffe://example.org/agent", "did:web:example.org", "did:web:caf\u00e9", "did:web:\U0001f600",
+    # The schema's negated authority class permits line breaks. Do not add a new rule.
+    "spiffe://exam\nple.org/agent",
+])
+def test_valid_subjects_keep_their_schema_meaning(subject):
+    body = copy.deepcopy(BASE)
+    body["subject"] = subject
+    key = generate_key()
+    verify_record(sign_record(body, key), key_to_jwk(key), now=body["iat"])
+
+
+@pytest.mark.parametrize("field", ["signature", "measurement"])
+def test_other_anchored_patterns_reject_final_newline(field):
+    record = copy.deepcopy(BASE)
+    target = record if field == "signature" else record["runtime"]
+    target[field] += "\n"
+    with pytest.raises(jsonschema.ValidationError):
+        validate_json(record)
+
+
+def _patterns(node):
+    if isinstance(node, dict):
+        if "pattern" in node:
+            yield node["pattern"]
+        for value in node.values():
+            yield from _patterns(value)
+    elif isinstance(node, list):
+        for value in node:
+            yield from _patterns(value)
+
+
+def test_every_shipped_pattern_has_an_explicit_adapter():
+    from agentrust_trace.validate import _ECMA_PATTERNS
+
+    assert set(_patterns(SCHEMA)) == set(_ECMA_PATTERNS)
+
+
+@pytest.mark.parametrize("value,valid", [("P1D", True), ("P\u0661D", False), ("P1D\n", False)])
+def test_duration_digits_and_end_anchor(value, valid):
+    from agentrust_trace.validate import _TraceValidator
+
+    pattern = next(p for p in _patterns(SCHEMA) if p.startswith("^P"))
+    assert _TraceValidator({"pattern": pattern}).is_valid(value) is valid
+
+
+def test_non_strings_are_left_to_the_type_keyword():
+    from agentrust_trace.validate import _TraceValidator
+
+    assert _TraceValidator({"pattern": SCHEMA["properties"]["subject"]["pattern"]}).is_valid(42)
+
+
+def test_pattern_and_format_registration_do_not_mutate_jsonschema():
+    from agentrust_trace.validate import _validator
+
+    original = jsonschema.Draft202012Validator.VALIDATORS["pattern"]
+    formats = jsonschema.FormatChecker.checkers.copy()
+    schema = copy.deepcopy(SCHEMA)
+    _validator.cache_clear()
+    validate_json(BASE)
+    assert jsonschema.Draft202012Validator.VALIDATORS["pattern"] is original
+    assert jsonschema.FormatChecker.checkers == formats
+    assert SCHEMA == schema
+
+
+def test_shipped_pattern_results_match_javascript():
+    """An independent runtime checks the actual published patterns, not our rewrite."""
+    from agentrust_trace.validate import _TraceValidator
+
+    node = shutil.which("node")
+    if node is None:
+        pytest.skip("Node is needed for the independent ECMA-262 comparison")
+    samples = [
+        "", "spiffe://example.org/agent", "did:web:example.org", "did:web:\U0001f600",
+        "spiffe://exam\nple.org/agent", "sha256:" + "a" * 64, "sha384:" + "f" * 96,
+        "abc_DEF-123", "P1D", "PT2H", "P\u0661D", "P\uff11D", "P1W",
+    ]
+    cases = [[pattern, value + tail + suffix]
+             for pattern in sorted(set(_patterns(SCHEMA)))
+             for value in samples for tail in ["", *TERMINATORS] for suffix in ["", "x"]]
+    result = subprocess.run(
+        [node, "-e", "const fs=require('fs'); const cases=JSON.parse(fs.readFileSync(0,'utf8')); "
+         "console.log(JSON.stringify(cases.map(([p,v])=>new RegExp(p).test(v))));"],
+        input=json.dumps(cases), text=True, capture_output=True, check=True,
+    )
+    expected = json.loads(result.stdout)
+    for (pattern, value), valid in zip(cases, expected, strict=True):
+        assert _TraceValidator({"pattern": pattern}).is_valid(value) == valid, (pattern, value)

tests/test_uri_formats_are_enforced.py

@@ -129,6 +129,18 @@ def test_the_base_record_is_valid() -> None:
     validate_json(BASE)
 
 
+@pytest.mark.parametrize("path", URI_FIELDS, ids=lambda p: ".".join(p))
+@pytest.mark.parametrize("tail", ["\n", "\r", "\u2028", "\u2029"])
+def test_uri_fields_reject_literal_line_terminators(path, tail):
+    with pytest.raises(jsonschema.ValidationError):
+        validate_json(_with(path, "https://example.org/a" + tail))
+
+
+@pytest.mark.parametrize("path", URI_FIELDS, ids=lambda p: ".".join(p))
+def test_uri_fields_accept_percent_encoded_newline(path):
+    validate_json(_with(path, "https://example.org/a%0A"))
+
+
 @pytest.mark.parametrize("path", URI_FIELDS, ids=lambda p: ".".join(p))
 @pytest.mark.parametrize("address", ["01.2.3.4", "1.02.3.4", "1.2.03.4", "1.2.3.04"])
 def test_ipv6_embedded_ipv4_octets_cannot_have_leading_zeros(