diff --git a/src/agentrust_trace/validate.py b/src/agentrust_trace/validate.py index de263e94..35c1f4ff 100644 --- a/src/agentrust_trace/validate.py +++ b/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) 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 b8497bb5..57ef72a0 100644 --- a/tests/test_uri_formats_are_enforced.py +++ b/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(