From e77f498da979b41fb1855cf1661f3a953ac6ad5d Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Mon, 21 Sep 2026 09:11:27 -0700 Subject: [PATCH] fix: apply ECMA regex boundaries in TRACE validation Signed-off-by: Imran Siddique --- src/agentrust_trace/validate.py | 51 +++++++++- tests/test_schema_regex_semantics.py | 136 +++++++++++++++++++++++++ tests/test_uri_formats_are_enforced.py | 12 +++ 3 files changed, 197 insertions(+), 2 deletions(-) create mode 100644 tests/test_schema_regex_semantics.py diff --git a/src/agentrust_trace/validate.py b/src/agentrust_trace/validate.py index 18879eb2..1e8ff1cb 100644 --- a/src/agentrust_trace/validate.py +++ b/src/agentrust_trace/validate.py @@ -3,10 +3,46 @@ import copy import importlib.resources import json +import re +from collections.abc import Iterator from functools import lru_cache 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) @@ -16,8 +52,19 @@ def _schema() -> dict[str, Any]: @lru_cache(maxsize=1) -def _validator() -> jsonschema.Draft202012Validator: - return jsonschema.Draft202012Validator(_schema(), format_checker=jsonschema.FormatChecker()) +def _validator() -> Validator: + base_uri_checker = jsonschema.FormatChecker(formats=["uri"]) + checker = jsonschema.FormatChecker() + + @checker.checks("uri") + 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 + return base_uri_checker.conforms(value, "uri") + + return cast(Validator, _TraceValidator(_schema(), format_checker=checker)) @lru_cache(maxsize=1) diff --git a/tests/test_schema_regex_semantics.py b/tests/test_schema_regex_semantics.py new file mode 100644 index 00000000..e48add08 --- /dev/null +++ b/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) diff --git a/tests/test_uri_formats_are_enforced.py b/tests/test_uri_formats_are_enforced.py index d2af2aea..184176f2 100644 --- a/tests/test_uri_formats_are_enforced.py +++ b/tests/test_uri_formats_are_enforced.py @@ -126,3 +126,15 @@ def test_the_uris_a_record_legitimately_carries_are_accepted( 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"))