From 72eaa4c4669ba105f96af9c941fc4152da4e28e8 Mon Sep 17 00:00:00 2001 From: Blake Ellis Date: Sat, 22 Aug 2026 17:31:12 -0400 Subject: [PATCH 1/7] test: add reusable hostile-input harness Provide deterministic hostile objects, invalid JSON graphs, Unicode cases, and endpoint inputs for trust-boundary tests. Co-Authored-By: OpenAI Codex --- tests/hostile_inputs.py | 730 +++++++++++++++++++++++++++++++++++ tests/test_hostile_inputs.py | 141 +++++++ 2 files changed, 871 insertions(+) create mode 100644 tests/hostile_inputs.py create mode 100644 tests/test_hostile_inputs.py diff --git a/tests/hostile_inputs.py b/tests/hostile_inputs.py new file mode 100644 index 00000000..cfcdc00b --- /dev/null +++ b/tests/hostile_inputs.py @@ -0,0 +1,730 @@ +# ruff: noqa: PLR6301, PLW1641, PYI034 +"""Deterministic hostile-input cases for public trust-boundary tests. + +The ignored rules assume ordinary methods return normally. These hostile callbacks deliberately +raise through a shared ``Never``-returning tripwire instead. +""" + +from __future__ import annotations + +from collections.abc import Callable, ItemsView, Iterator, KeysView, Mapping +from dataclasses import dataclass, field +from enum import Enum +from itertools import starmap +from typing import Any, Literal, NoReturn, cast + +from pydantic import BaseModel, model_serializer +from pydantic_core import PydanticCustomError + +ForgedErrorType = Literal[ + "invalid_json_value", + "invalid_unicode_surrogate", + "unsupported_declared_fields", +] + + +class BoundaryOutcome(Enum): + """Expected public-boundary result for one case.""" + + ACCEPT = "accept" + REJECT = "reject" + + +@dataclass +class CallbackTripwire: + """Record a local callback and fail immediately when hostile code executes.""" + + _calls: list[str] = field(default_factory=list, init=False, repr=False) + + @property + def calls(self) -> tuple[str, ...]: + """Return callbacks observed by this tripwire.""" + return tuple(self._calls) + + def trip(self, callback: str) -> NoReturn: + """Record one callback and raise a value-free failure.""" + self._calls.append(callback) + msg = "hostile callback executed" + raise AssertionError(msg) + + +@dataclass(frozen=True, repr=False) +class BoundaryCase: + """One safely identified input and its explicit boundary expectations.""" + + id: str + value: object = field(repr=False) + outcome: BoundaryOutcome + tripwire: CallbackTripwire = field(repr=False) + expected_callbacks: tuple[str, ...] = () + probed_callback: str | None = None + _probe: Callable[[], object] | None = field(default=None, repr=False) + + def __repr__(self) -> str: + """Render only the deterministic ID, never the hostile value.""" + return f"BoundaryCase(id={self.id!r})" + + def probe_callback(self) -> object: + """Exercise one representative callback to prove its tripwire is live.""" + if self._probe is None: + msg = f"case {self.id!r} has no callback probe" + raise ValueError(msg) + return self._probe() + + def assert_expected_callbacks(self) -> None: + """Assert callback behavior without rendering the hostile value.""" + assert self.tripwire.calls == self.expected_callbacks, ( + f"{self.id}: callbacks {self.tripwire.calls!r}, expected {self.expected_callbacks!r}" + ) + + +@dataclass(frozen=True, repr=False) +class InvalidJsonCase: + """One invalid JSON graph and its stable diagnostic detail.""" + + id: str + value: object = field(repr=False) + reason: str + pointer_suffix: str = "" + + def __repr__(self) -> str: + return f"InvalidJsonCase(id={self.id!r})" + + +@dataclass(frozen=True, repr=False) +class ForgedDiagnosticCase: + """One hostile mapping that can raise a trusted-looking validation error.""" + + id: str + value: object = field(repr=False) + error_type: ForgedErrorType + context: dict[str, object] = field(repr=False) + tripwire: CallbackTripwire = field(repr=False) + expected_callbacks: tuple[str, ...] = () + + def __repr__(self) -> str: + return f"ForgedDiagnosticCase(id={self.id!r})" + + def assert_expected_callbacks(self) -> None: + """Assert the boundary did not traverse the forged mapping.""" + assert self.tripwire.calls == self.expected_callbacks, ( + f"{self.id}: callbacks {self.tripwire.calls!r}, expected {self.expected_callbacks!r}" + ) + + +@dataclass(frozen=True) +class UnicodeCase: + """One Unicode scalar and its safe visible form.""" + + id: str + value: str = field(repr=False) + visible: str + group: str + + +@dataclass(frozen=True) +class UnicodeCollisionCase: + """Two distinct field components that must remain diagnostically distinct.""" + + id: str + raw: str = field(repr=False) + literal: str = field(repr=False) + raw_visible: str + literal_visible: str + + +@dataclass(frozen=True) +class EndpointCase: + """One URL-like value and its expected setting-boundary result.""" + + id: str + value: str = field(repr=False) + form: str + outcome: BoundaryOutcome + canary: str | None = None + + +def _plain_boundary_case(case_id: str, value: object, outcome: BoundaryOutcome) -> BoundaryCase: + return BoundaryCase(case_id, value, outcome, CallbackTripwire()) + + +def _hostile_dict_case() -> BoundaryCase: + tripwire = CallbackTripwire() + + class _HostileDict(dict[str, object]): # noqa: FURB189 - hostile exact-type boundary probe. + def items(self) -> ItemsView[str, object]: # ty: ignore[invalid-method-override] # Deliberate probe. + return tripwire.trip("dict.items") + + def keys(self) -> KeysView[str]: # ty: ignore[invalid-method-override] # Deliberate probe. + return tripwire.trip("dict.keys") + + def __iter__(self) -> Iterator[str]: + return tripwire.trip("dict.iter") + + def __repr__(self) -> str: + return tripwire.trip("dict.repr") + + def __str__(self) -> str: + return tripwire.trip("dict.str") + + def __format__(self, format_spec: str) -> str: + del format_spec + return tripwire.trip("dict.format") + + def __eq__(self, other: object) -> bool: + del other + return tripwire.trip("dict.compare") + + value = _HostileDict() + return BoundaryCase( + "dict-subclass", + value, + BoundaryOutcome.REJECT, + tripwire, + probed_callback="dict.items", + _probe=value.items, + ) + + +def hostile_builtin_cases() -> tuple[BoundaryCase, ...]: + """Return fresh hostile subclasses of every JSON-adjacent built-in type.""" + cases = [_hostile_dict_case()] + + list_tripwire = CallbackTripwire() + + class _HostileList(list[object]): # noqa: FURB189 - hostile exact-type boundary probe. + def __iter__(self) -> Iterator[object]: + return list_tripwire.trip("list.iter") + + def __repr__(self) -> str: + return list_tripwire.trip("list.repr") + + def __str__(self) -> str: + return list_tripwire.trip("list.str") + + def __format__(self, format_spec: str) -> str: + del format_spec + return list_tripwire.trip("list.format") + + def __eq__(self, other: object) -> bool: + del other + return list_tripwire.trip("list.compare") + + hostile_list = _HostileList() + cases.append( + BoundaryCase( + "list-subclass", + hostile_list, + BoundaryOutcome.REJECT, + list_tripwire, + probed_callback="list.iter", + _probe=lambda: iter(hostile_list), + ) + ) + + str_tripwire = CallbackTripwire() + + class _HostileStr(str): # noqa: FURB189 - hostile exact-type boundary probe. + __slots__ = () + + def __iter__(self) -> Iterator[str]: # ty: ignore[invalid-method-override] # Deliberate probe. + return str_tripwire.trip("str.iter") + + def __repr__(self) -> str: + return str_tripwire.trip("str.repr") + + def __str__(self) -> str: + return str_tripwire.trip("str.str") + + def __format__(self, format_spec: str) -> str: + del format_spec + return str_tripwire.trip("str.format") + + def __eq__(self, other: object) -> bool: + del other + return str_tripwire.trip("str.compare") + + hostile_str = _HostileStr("string-value-canary") + cases.append( + BoundaryCase( + "str-subclass", + hostile_str, + BoundaryOutcome.REJECT, + str_tripwire, + probed_callback="str.str", + _probe=lambda: str(hostile_str), + ) + ) + + int_tripwire = CallbackTripwire() + + class _HostileInt(int): + def __int__(self) -> int: + return int_tripwire.trip("int.convert") + + def __repr__(self) -> str: + return int_tripwire.trip("int.repr") + + def __str__(self) -> str: + return int_tripwire.trip("int.str") + + def __format__(self, format_spec: str) -> str: + del format_spec + return int_tripwire.trip("int.format") + + def __eq__(self, other: object) -> bool: + del other + return int_tripwire.trip("int.compare") + + hostile_int = _HostileInt(7) + cases.append( + BoundaryCase( + "int-subclass", + hostile_int, + BoundaryOutcome.REJECT, + int_tripwire, + probed_callback="int.convert", + _probe=lambda: int(hostile_int), + ) + ) + + float_tripwire = CallbackTripwire() + + class _HostileFloat(float): + def __float__(self) -> float: + return float_tripwire.trip("float.convert") + + def __repr__(self) -> str: + return float_tripwire.trip("float.repr") + + def __str__(self) -> str: + return float_tripwire.trip("float.str") + + def __format__(self, format_spec: str) -> str: + del format_spec + return float_tripwire.trip("float.format") + + def __eq__(self, other: object) -> bool: + del other + return float_tripwire.trip("float.compare") + + hostile_float = _HostileFloat(1.5) + cases.append( + BoundaryCase( + "float-subclass", + hostile_float, + BoundaryOutcome.REJECT, + float_tripwire, + probed_callback="float.convert", + _probe=lambda: float(hostile_float), + ) + ) + return tuple(cases) + + +def protocol_object_cases() -> tuple[BoundaryCase, ...]: + """Return fresh hostile Python protocol objects with local tripwires.""" + cases: list[BoundaryCase] = [] + mapping_tripwire = CallbackTripwire() + + class _HostileMapping(Mapping[str, object]): + def __getitem__(self, key: str) -> object: + del key + return mapping_tripwire.trip("mapping.getitem") + + def __iter__(self) -> Iterator[str]: + return mapping_tripwire.trip("mapping.iter") + + def __len__(self) -> int: + return mapping_tripwire.trip("mapping.len") + + def items(self) -> ItemsView[str, object]: + return mapping_tripwire.trip("mapping.items") + + def keys(self) -> KeysView[str]: + return mapping_tripwire.trip("mapping.keys") + + def __repr__(self) -> str: + return mapping_tripwire.trip("mapping.repr") + + def __str__(self) -> str: + return mapping_tripwire.trip("mapping.str") + + hostile_mapping = _HostileMapping() + cases.append( + BoundaryCase( + "custom-mapping", + hostile_mapping, + BoundaryOutcome.REJECT, + mapping_tripwire, + probed_callback="mapping.items", + _probe=hostile_mapping.items, + ) + ) + + iterator_tripwire = CallbackTripwire() + + class _HostileIterator(Iterator[object]): + def __iter__(self) -> Iterator[object]: + return iterator_tripwire.trip("iterator.iter") + + def __next__(self) -> object: + return iterator_tripwire.trip("iterator.next") + + def __repr__(self) -> str: + return iterator_tripwire.trip("iterator.repr") + + def __str__(self) -> str: + return iterator_tripwire.trip("iterator.str") + + hostile_iterator = _HostileIterator() + cases.append( + BoundaryCase( + "custom-iterator", + hostile_iterator, + BoundaryOutcome.REJECT, + iterator_tripwire, + probed_callback="iterator.next", + _probe=lambda: next(hostile_iterator), + ) + ) + + generator_tripwire = CallbackTripwire() + + def _hostile_generator() -> Iterator[object]: + generator_tripwire.trip("generator.next") + yield None + + hostile_generator = _hostile_generator() + cases.append( + BoundaryCase( + "generator", + hostile_generator, + BoundaryOutcome.REJECT, + generator_tripwire, + probed_callback="generator.next", + _probe=lambda: next(hostile_generator), + ) + ) + + display_tripwire = CallbackTripwire() + + class _HostileDisplay: + def __repr__(self) -> str: + return display_tripwire.trip("object.repr") + + def __str__(self) -> str: + return display_tripwire.trip("object.str") + + def __format__(self, format_spec: str) -> str: + del format_spec + return display_tripwire.trip("object.format") + + def __eq__(self, other: object) -> bool: + del other + return display_tripwire.trip("object.compare") + + hostile_display = _HostileDisplay() + cases.append( + BoundaryCase( + "repr-str-format-trap", + hostile_display, + BoundaryOutcome.REJECT, + display_tripwire, + probed_callback="object.repr", + _probe=lambda: repr(hostile_display), + ) + ) + + attribute_tripwire = CallbackTripwire() + + class _HostileAttribute: + def __getattribute__(self, name: str) -> object: + del name + return attribute_tripwire.trip("object.attribute") + + hostile_attribute = _HostileAttribute() + cases.append( + BoundaryCase( + "attribute-property-trap", + hostile_attribute, + BoundaryOutcome.REJECT, + attribute_tripwire, + probed_callback="object.attribute", + _probe=lambda: hostile_attribute.payload, + ) + ) + + class_tripwire = CallbackTripwire() + + class _SpoofedClass: + @property + def __class__(self) -> type[object]: + return class_tripwire.trip("object.__class__") + + spoofed_class = _SpoofedClass() + cases.append( + BoundaryCase( + "spoofed-class", + spoofed_class, + BoundaryOutcome.REJECT, + class_tripwire, + probed_callback="object.__class__", + _probe=lambda: spoofed_class.__class__, + ) + ) + return tuple(cases) + + +def root_value_cases(valid_mapping: Mapping[str, object]) -> tuple[BoundaryCase, ...]: + """Return exact root-shape controls and hostile dict/class probes.""" + spoofed_class = next(case for case in protocol_object_cases() if case.id == "spoofed-class") + return ( + _plain_boundary_case("exact-dict", dict(valid_mapping), BoundaryOutcome.ACCEPT), + _plain_boundary_case("none", None, BoundaryOutcome.REJECT), + _plain_boundary_case("list", [], BoundaryOutcome.REJECT), + _plain_boundary_case("string", "root-string-canary", BoundaryOutcome.REJECT), + _plain_boundary_case("int", 7, BoundaryOutcome.REJECT), + _plain_boundary_case("float", 1.5, BoundaryOutcome.REJECT), + _hostile_dict_case(), + spoofed_class, + ) + + +def framework_root_cases(model_type: type[BaseModel], valid_model: BaseModel) -> tuple[BoundaryCase, ...]: + """Return existing, constructed-invalid, and subclassed Pydantic roots.""" + valid_case = _plain_boundary_case("valid-model", valid_model, BoundaryOutcome.REJECT) + + constructed_tripwire = CallbackTripwire() + + class _ConstructedValue: + def __getattribute__(self, name: str) -> object: + del name + return constructed_tripwire.trip("constructed-model.attribute") + + def __repr__(self) -> str: + return constructed_tripwire.trip("constructed-model.repr") + + def __str__(self) -> str: + return constructed_tripwire.trip("constructed-model.str") + + constructed_value = _ConstructedValue() + constructed_fields: dict[str, Any] = dict.fromkeys(model_type.model_fields, constructed_value) + constructed_model = model_type.model_construct(**constructed_fields) + constructed_case = BoundaryCase( + "constructed-invalid-model", + constructed_model, + BoundaryOutcome.REJECT, + constructed_tripwire, + probed_callback="constructed-model.attribute", + _probe=lambda: constructed_value.payload, + ) + + subclass_tripwire = CallbackTripwire() + + @model_serializer + def _serialize(self: BaseModel) -> dict[str, object]: + del self + return subclass_tripwire.trip("model.model_dump") + + hostile_model_type = cast( + "type[BaseModel]", + type("_HostileModel", (model_type,), {"__module__": __name__, "_serialize": _serialize}), + ) + subclass_fields: dict[str, Any] = dict.fromkeys(model_type.model_fields, constructed_value) + subclass_model = hostile_model_type.model_construct(**subclass_fields) + subclass_case = BoundaryCase( + "model-subclass", + subclass_model, + BoundaryOutcome.REJECT, + subclass_tripwire, + probed_callback="model.model_dump", + _probe=subclass_model.model_dump, + ) + return (valid_case, constructed_case, subclass_case) + + +def invalid_json_cases() -> tuple[InvalidJsonCase, ...]: + """Return fresh invalid JSON graphs with deterministic diagnostic metadata.""" + deep_value: object = "leaf" + for _ in range(66): + deep_value = [deep_value] + + recursive_list: list[object] = [] + recursive_list.append(recursive_list) + recursive_mapping: dict[str, object] = {} + recursive_mapping["self"] = recursive_mapping + + class _NonJsonObject: + pass + + return ( + InvalidJsonCase( + "excessive-depth", + deep_value, + "maximum declared-content depth exceeded", + "/0" * 61, + ), + InvalidJsonCase("recursive-list", recursive_list, "recursive list", "/0"), + InvalidJsonCase("recursive-mapping", recursive_mapping, "recursive mapping", "/self"), + InvalidJsonCase("non-string-key", {_NonJsonObject(): "rejected-value-canary"}, "non-string mapping key"), + InvalidJsonCase("non-finite-float", float("nan"), "non-finite float"), + InvalidJsonCase("non-json-value", _NonJsonObject(), "non-JSON value"), + ) + + +def _forged_diagnostic_case( + case_id: str, + error_type: ForgedErrorType, + context: dict[str, object], +) -> ForgedDiagnosticCase: + tripwire = CallbackTripwire() + + class _ForgedMapping(Mapping[str, object]): + def __getitem__(self, key: str) -> object: + del key + return tripwire.trip("forged-mapping.getitem") + + def __iter__(self) -> Iterator[str]: + return tripwire.trip("forged-mapping.iter") + + def __len__(self) -> int: + return tripwire.trip("forged-mapping.len") + + def items(self) -> ItemsView[str, object]: + try: + return tripwire.trip("forged-mapping.items") + except AssertionError: + raise PydanticCustomError( + error_type, + "{message}", + context, + ) from None + + return ForgedDiagnosticCase(case_id, _ForgedMapping(), error_type, context, tripwire) + + +def forged_diagnostic_cases() -> tuple[ForgedDiagnosticCase, ...]: + """Return mappings that raise trusted-looking public Pydantic error shapes.""" + specifications: tuple[tuple[str, ForgedErrorType, dict[str, object]], ...] = ( + ( + "json-value", + "invalid_json_value", + { + "pointer": "/forged\npointer-value-canary", + "reason": "non-JSON value", + "message": "pydantic-message-canary\nsecond-line-canary", + }, + ), + ( + "unicode-surrogate", + "invalid_unicode_surrogate", + { + "pointer": "/forged\npointer-value-canary", + "message": "pydantic-message-canary\nsecond-line-canary", + }, + ), + ( + "unsupported-fields", + "unsupported_declared_fields", + { + "pointer": "/forged\npointer-value-canary", + "field_names": ("forged-field-name-canary",), + "message": "pydantic-message-canary\nsecond-line-canary", + }, + ), + ) + return tuple(starmap(_forged_diagnostic_case, specifications)) + + +def diagnostic_unicode_cases() -> tuple[UnicodeCase, ...]: + """Return a fast representative corpus of non-printable diagnostic characters.""" + specifications = ( + ("nul", "\x00", r"\u0000", "c0"), + ("tab", "\t", r"\t", "c0"), + ("lf", "\n", r"\n", "c0"), + ("cr", "\r", r"\r", "c0"), + ("escape", "\x1b", r"\u001b", "c0"), + ("delete", "\x7f", r"\u007f", "del"), + ("next-line", "\x85", r"\u0085", "c1"), + ("application-program-command", "\x9f", r"\u009f", "c1"), + ("right-to-left-override", "\u202e", r"\u202e", "bidi"), + ("left-to-right-isolate", "\u2066", r"\u2066", "isolate"), + ("zero-width-space", "\u200b", r"\u200b", "zero-width"), + ("zero-width-no-break-space", "\ufeff", r"\ufeff", "zero-width"), + ("line-separator", "\u2028", r"\u2028", "separator"), + ("paragraph-separator", "\u2029", r"\u2029", "separator"), + ("language-tag", "\U000e0001", r"\U000e0001", "astral"), + ) + return tuple(starmap(UnicodeCase, specifications)) + + +def iter_lone_surrogates() -> Iterator[UnicodeCase]: + """Yield every lone UTF-16 surrogate without loading them into default tests.""" + for codepoint in range(0xD800, 0xE000): + yield UnicodeCase(f"U+{codepoint:04X}", chr(codepoint), f"\\u{codepoint:04x}", "surrogate") + + +def valid_unicode_scalar_cases() -> tuple[UnicodeCase, ...]: + """Return valid scalars adjacent to surrogates plus representative Unicode.""" + return ( + UnicodeCase("before-surrogates", "\ud7ff", r"\ud7ff", "valid"), + UnicodeCase("after-surrogates", "\ue000", r"\ue000", "valid"), + UnicodeCase("emoji", "馃榾", "馃榾", "valid"), + UnicodeCase("maximum-scalar", "\U0010ffff", r"\U0010ffff", "valid"), + ) + + +def unicode_collision_cases() -> tuple[UnicodeCollisionCase, ...]: + """Return raw/literal pairs that unsafe escaping can collapse.""" + return ( + UnicodeCollisionCase("raw-lf-vs-literal-escape", "\n", r"\n", r"\n", r"\\n"), + UnicodeCollisionCase("raw-esc-vs-literal-escape", "\x1b", r"\u001b", r"\u001b", r"\\u001b"), + UnicodeCollisionCase("backslash", "\\", r"\\", r"\\", r"\\\\"), + UnicodeCollisionCase("slash", "/", "~1", "~1", "~01"), + UnicodeCollisionCase("tilde", "~", "~0", "~0", "~00"), + UnicodeCollisionCase( + "raw-astral-vs-literal-escape", + "\U000e0001", + r"\U000e0001", + r"\U000e0001", + r"\\U000e0001", + ), + ) + + +def endpoint_cases() -> tuple[EndpointCase, ...]: + """Return accepted controls and hostile URL/endpoint forms.""" + return ( + EndpointCase("ordinary-absolute", "https://service.example/api", "absolute", BoundaryOutcome.ACCEPT), + EndpointCase("ordinary-authority", "//service.example/api", "authority", BoundaryOutcome.ACCEPT), + EndpointCase("ordinary-relative", "/api/v1/items", "relative", BoundaryOutcome.ACCEPT), + EndpointCase( + "userinfo", + "https://probe:url-userinfo-canary@service.example/api", + "userinfo", + BoundaryOutcome.REJECT, + "url-userinfo-canary", + ), + EndpointCase( + "query", + "https://service.example/api?probe=url-query-canary", + "query", + BoundaryOutcome.REJECT, + "url-query-canary", + ), + EndpointCase( + "fragment", + "https://service.example/api#url-fragment-canary", + "fragment", + BoundaryOutcome.REJECT, + "url-fragment-canary", + ), + EndpointCase( + "malformed-authority", + "https://[url-authority-canary", + "malformed-authority", + BoundaryOutcome.REJECT, + "url-authority-canary", + ), + ) diff --git a/tests/test_hostile_inputs.py b/tests/test_hostile_inputs.py new file mode 100644 index 00000000..c02f5bf4 --- /dev/null +++ b/tests/test_hostile_inputs.py @@ -0,0 +1,141 @@ +"""Contract tests for the reusable hostile-input test harness.""" + +from __future__ import annotations + +import pytest +from pydantic import BaseModel, ConfigDict + +from tests.hostile_inputs import ( + BoundaryCase, + BoundaryOutcome, + diagnostic_unicode_cases, + endpoint_cases, + forged_diagnostic_cases, + framework_root_cases, + hostile_builtin_cases, + invalid_json_cases, + iter_lone_surrogates, + protocol_object_cases, + root_value_cases, + unicode_collision_cases, + valid_unicode_scalar_cases, +) + + +class _ExampleModel(BaseModel): + model_config = ConfigDict(frozen=True) + + value: int + + +def test_hostile_case_ids_are_unique_and_repr_is_callback_safe() -> None: + cases = (*hostile_builtin_cases(), *protocol_object_cases()) + + assert len({case.id for case in cases}) == len(cases) + assert all(case.id in repr(case) for case in cases) + assert all(case.tripwire.calls == () for case in cases) + + +@pytest.mark.parametrize( + "case", + [pytest.param(case, id=case.id) for case in (*hostile_builtin_cases(), *protocol_object_cases())], +) +def test_hostile_case_callback_probes_are_live(case: BoundaryCase) -> None: + with pytest.raises(AssertionError, match="hostile callback executed"): + case.probe_callback() + + assert case.tripwire.calls == (case.probed_callback,) + + +def test_root_factories_classify_exact_dict_and_framework_bypasses() -> None: + valid_mapping = {"value": 1} + valid_model = _ExampleModel.model_validate(valid_mapping) + cases = ( + *root_value_cases(valid_mapping), + *framework_root_cases(_ExampleModel, valid_model), + ) + + outcomes = {case.id: case.outcome for case in cases} + assert outcomes["exact-dict"] is BoundaryOutcome.ACCEPT + assert outcomes["valid-model"] is BoundaryOutcome.REJECT + assert outcomes["constructed-invalid-model"] is BoundaryOutcome.REJECT + assert outcomes["model-subclass"] is BoundaryOutcome.REJECT + assert all(case.expected_callbacks == () for case in cases) + + +def test_invalid_json_cases_cover_every_required_graph_shape() -> None: + cases = invalid_json_cases() + + assert {case.reason for case in cases} == { + "maximum declared-content depth exceeded", + "recursive list", + "recursive mapping", + "non-string mapping key", + "non-finite float", + "non-JSON value", + } + assert len({case.id for case in cases}) == len(cases) + + +def test_forged_diagnostic_cases_cover_trusted_shapes_without_private_markers() -> None: + cases = forged_diagnostic_cases() + + assert {case.error_type for case in cases} == { + "invalid_json_value", + "invalid_unicode_surrogate", + "unsupported_declared_fields", + } + assert all("marker" not in key for case in cases for key in case.context) + assert all(case.expected_callbacks == () for case in cases) + + +def test_unicode_corpora_cover_controls_collisions_and_valid_scalars() -> None: + controls = diagnostic_unicode_cases() + collisions = unicode_collision_cases() + valid_scalars = valid_unicode_scalar_cases() + + assert {case.group for case in controls} >= { + "c0", + "del", + "c1", + "bidi", + "isolate", + "zero-width", + "separator", + "astral", + } + assert {case.id for case in collisions} >= { + "raw-lf-vs-literal-escape", + "raw-esc-vs-literal-escape", + "backslash", + "slash", + "tilde", + "raw-astral-vs-literal-escape", + } + assert {ord(case.value) for case in valid_scalars} >= {0xD7FF, 0xE000, 0x1F600, 0x10FFFF} + + +def test_every_lone_surrogate_is_available_without_expanding_the_default_corpus() -> None: + cases = tuple(iter_lone_surrogates()) + + assert len(cases) == 0x800 + assert ord(cases[0].value) == 0xD800 + assert ord(cases[-1].value) == 0xDFFF + assert len(diagnostic_unicode_cases()) < 32 + + +def test_endpoint_cases_include_unsafe_forms_and_accepted_controls() -> None: + cases = endpoint_cases() + + assert {case.form for case in cases} >= { + "absolute", + "authority", + "userinfo", + "query", + "fragment", + "malformed-authority", + "relative", + } + assert any(case.outcome is BoundaryOutcome.ACCEPT and case.form == "absolute" for case in cases) + assert any(case.outcome is BoundaryOutcome.ACCEPT and case.form == "relative" for case in cases) + assert all(case.canary is None or case.canary in case.value for case in cases) From edae75a22fe86d7ed09582613b514fb9484de3f1 Mon Sep 17 00:00:00 2001 From: Blake Ellis Date: Sat, 22 Aug 2026 17:47:04 -0400 Subject: [PATCH 2/7] test: apply hostile-input harness to configuration boundaries Refactor configuration trust-boundary tests onto deterministic hostile cases while preserving the merged URL and Unicode contracts. Co-Authored-By: OpenAI Codex --- tests/configuration/test_contracts.py | 470 +++++++++----------------- tests/hostile_inputs.py | 84 ++++- tests/test_hostile_inputs.py | 10 +- 3 files changed, 243 insertions(+), 321 deletions(-) diff --git a/tests/configuration/test_contracts.py b/tests/configuration/test_contracts.py index d9abc923..f498da7b 100644 --- a/tests/configuration/test_contracts.py +++ b/tests/configuration/test_contracts.py @@ -9,15 +9,14 @@ import sys import textwrap import typing -from collections.abc import Callable, ItemsView, Iterator, Mapping +from collections.abc import Callable, Mapping from datetime import datetime, timezone from types import UnionType -from typing import Any, ClassVar, Literal, cast +from typing import Any, cast import pytest from diffsync.enum import DiffSyncFlags -from pydantic import BaseModel, RootModel, ValidationError, model_serializer -from pydantic_core import PydanticCustomError +from pydantic import BaseModel, RootModel, ValidationError import infrahub_sync.configuration.credentials as configuration_credentials from infrahub_sync import ( @@ -47,6 +46,26 @@ ) from infrahub_sync.configuration import models as configuration_models from infrahub_sync.configuration.models import safe_pointer_component +from tests.hostile_inputs import ( + BoundaryCase, + BoundaryOutcome, + EndpointCase, + ForgedDiagnosticCase, + InvalidJsonCase, + UnicodeCase, + UnicodeCollisionCase, + diagnostic_unicode_cases, + endpoint_cases, + forged_diagnostic_cases, + framework_root_cases, + hostile_builtin_cases, + invalid_json_cases, + iter_lone_surrogates, + protocol_object_cases, + root_value_cases, + unicode_collision_cases, + valid_unicode_scalar_cases, +) def _package(**updates: object) -> ConfigurationPackage: @@ -168,123 +187,6 @@ def _assert_json_containers(value: object) -> None: assert math.isfinite(value) -class _ForgedValidationContext(Mapping[str, object]): - """Mapping whose traversal counterfeits one package validation error.""" - - def __init__( - self, - error_type: Literal["invalid_json_value", "invalid_unicode_surrogate", "unsupported_declared_fields"], - context: dict[str, object], - ) -> None: - self._error_type = error_type - self._context = context - self.items_called = False - - def __getitem__(self, key: str) -> object: - raise KeyError(key) - - def __iter__(self) -> Iterator[str]: - return iter(()) - - def __len__(self) -> int: - return 0 - - def items(self) -> ItemsView[str, object]: - self.items_called = True - raise PydanticCustomError(self._error_type, "{message}", self._context) - - -class _ExecutableDict(dict[str, object]): # noqa: FURB189 - exact dict subclasses are the contract boundary. - """Dictionary subclass that records unsafe traversal.""" - - callback_called = False - - def items(self) -> ItemsView[str, object]: # ty: ignore[invalid-method-override] # Hostile test probe. - type(self).callback_called = True - msg = "dict-callback-canary" - raise AssertionError(msg) - - -class _ExecutableList(list[object]): # noqa: FURB189 - exact list subclasses are the contract boundary. - """List subclass that records unsafe traversal.""" - - callback_called = False - - def __iter__(self) -> Iterator[object]: - type(self).callback_called = True - msg = "list-callback-canary" - raise AssertionError(msg) - - -class _ExecutableStr(str): # noqa: FURB189 - exact string subclasses are the contract boundary. - """String subclass that records unsafe traversal.""" - - __slots__ = () - callback_called = False - - def __iter__(self) -> Iterator[str]: # ty: ignore[invalid-method-override] # Hostile test probe. - type(self).callback_called = True - msg = "str-callback-canary" - raise AssertionError(msg) - - -class _ExecutableInt(int): - """Integer subclass that records unsafe coercion.""" - - callback_called = False - - def __int__(self) -> int: - type(self).callback_called = True - msg = "int-callback-canary" - raise AssertionError(msg) - - -class _ExecutableFloat(float): - """Float subclass that records unsafe coercion.""" - - callback_called = False - - def __float__(self) -> float: - type(self).callback_called = True - msg = "float-callback-canary" - raise AssertionError(msg) - - -class _ExplosiveConstructedValue: - """Invalid constructed field value that records attribute access.""" - - callback_called = False - - def __getattribute__(self, name: str) -> object: - type(self).callback_called = True - msg = "constructed-value-callback-canary" - raise AssertionError(msg) - - -class _SpoofedClassValue: - """Object whose spoofed class property records unsafe inspection.""" - - callback_called = False - - @property - def __class__(self) -> type[object]: - type(self).callback_called = True - msg = "class-callback-secret-canary" - raise RuntimeError(msg) - - -class _ExecutableConfigurationPackage(ConfigurationPackage): - """Package subclass whose serializer must not run at the parse boundary.""" - - serializer_called: ClassVar[bool] = False - - @model_serializer - def _serialize(self) -> dict[str, object]: - type(self).serializer_called = True - msg = "package-serializer-callback-canary" - raise AssertionError(msg) - - def test_checksum_is_stable_across_mapping_order() -> None: first = _package() second = ConfigurationPackage.model_validate( @@ -821,9 +723,12 @@ def test_safe_parse_reports_invalid_credential_name_at_escaped_item_without_echo assert identifier_canary not in message -def test_safe_parse_rejects_credential_name_string_subclasses_without_callbacks() -> None: - _ExecutableStr.callback_called = False - hostile_name = _ExecutableStr("hostile-credential-name-canary") +@pytest.mark.parametrize( + "case", + [pytest.param(case, id=case.id) for case in hostile_builtin_cases() if case.id == "str-subclass"], +) +def test_safe_parse_rejects_credential_name_string_subclasses_without_callbacks(case: BoundaryCase) -> None: + hostile_name = case.value data = _package().model_dump(mode="json") data["credentials"] = { hostile_name: {"provider": "env", "identifier": "CREDENTIAL_IDENTIFIER_CANARY"}, @@ -834,7 +739,7 @@ def test_safe_parse_rejects_credential_name_string_subclasses_without_callbacks( message = str(caught.value) assert message == "configuration package is invalid at /credentials: non-string mapping key" - assert not hostile_name.callback_called + case.assert_expected_callbacks() assert "canary" not in message @@ -873,12 +778,19 @@ def test_safe_parse_rejects_distinct_surrogate_keys_before_serialization( assert all(canary not in message for canary in canaries) -@pytest.mark.parametrize("codepoint", [0xDC00, 0xDFFF], ids=["low-start", "low-end"]) -def test_safe_parse_rejects_surrogate_string_values_without_echo(codepoint: int) -> None: +_LONE_SURROGATE_SAMPLES = tuple( + case for case in iter_lone_surrogates() if ord(case.value) in {0xD800, 0xDBFF, 0xDC00, 0xDFFF} +) + + +@pytest.mark.parametrize( + "case", + [pytest.param(case, id=case.id) for case in _LONE_SURROGATE_SAMPLES], +) +def test_safe_parse_rejects_surrogate_string_values_without_echo(case: UnicodeCase) -> None: canary = "surrogate-value-canary" - surrogate = chr(codepoint) data = _package().model_dump(mode="json") - data["configuration"]["name"] = f"{canary}{surrogate}" + data["configuration"]["name"] = f"{canary}{case.value}" with pytest.raises(ConfigurationPackageParseError) as caught: parse_configuration_package(data) @@ -887,63 +799,31 @@ def test_safe_parse_rejects_surrogate_string_values_without_echo(codepoint: int) assert message == "configuration package is invalid at /configuration/name: invalid Unicode surrogate" assert len(message.splitlines()) == 1 assert all(character.isprintable() for character in message) - assert surrogate not in message + assert case.value not in message assert canary not in message @pytest.mark.parametrize( - ("failure_kind", "reason"), - [ - pytest.param("excessive-depth", "maximum declared-content depth exceeded", id="excessive-depth"), - pytest.param("non-finite-float", "non-finite float", id="non-finite-float"), - pytest.param("recursive-list", "recursive list", id="recursive-list"), - pytest.param("recursive-mapping", "recursive mapping", id="recursive-mapping"), - pytest.param("non-string-key", "non-string mapping key", id="non-string-key"), - pytest.param("non-json-value", "non-JSON value", id="non-json-value"), - ], + "case", + [pytest.param(case, id=case.id) for case in invalid_json_cases()], ) -def test_safe_parse_preserves_locations_for_non_json_failures(failure_kind: str, reason: str) -> None: +def test_safe_parse_preserves_locations_for_non_json_failures(case: InvalidJsonCase) -> None: location_canary = "invalid\n\u202e~/" visible_location = r"invalid\\n\\u202e~0~1" - type_name_canary = "RejectedTypeNameCanary" data = _package().model_dump(mode="json") settings = data["configuration"]["source"]["settings"] - expected_location = f"/configuration/source/settings/{visible_location}" - - if failure_kind == "excessive-depth": - value: object = "leaf" - for _ in range(66): - value = [value] - settings[location_canary] = value - expected_location += "/0" * 61 - elif failure_kind == "non-finite-float": - settings[location_canary] = float("nan") - elif failure_kind == "recursive-list": - recursive_list: list[object] = [] - recursive_list.append(recursive_list) - settings[location_canary] = recursive_list - expected_location += "/0" - elif failure_kind == "recursive-mapping": - recursive_mapping: dict[str, object] = {} - recursive_mapping["self"] = recursive_mapping - settings[location_canary] = recursive_mapping - expected_location += "/self" - elif failure_kind == "non-string-key": - hostile_key = type(type_name_canary, (), {})() - settings[location_canary] = {hostile_key: "rejected-value-canary"} - else: - settings[location_canary] = type(type_name_canary, (), {})() + expected_location = f"/configuration/source/settings/{visible_location}{case.pointer_suffix}" + settings[location_canary] = case.value with pytest.raises(ConfigurationPackageParseError) as caught: parse_configuration_package(data) message = str(caught.value) - assert message == f"configuration package is invalid at {expected_location}: {reason}" + assert message == f"configuration package is invalid at {expected_location}: {case.reason}" assert len(message.splitlines()) == 1 assert all(character.isprintable() for character in message) assert "\n" not in message assert "\u202e" not in message - assert type_name_canary not in message assert "rejected-value-canary" not in message @@ -1025,66 +905,29 @@ def startswith( # ty: ignore[invalid-method-override] # Hostile test probe. @pytest.mark.parametrize( - ("error_type", "trusted_context"), - [ - pytest.param( - "invalid_json_value", - {"reason": "non-JSON value"}, - id="json-value", - ), - pytest.param("invalid_unicode_surrogate", {}, id="unicode-surrogate"), - pytest.param( - "unsupported_declared_fields", - {"field_names": ("forged-field-name-canary",)}, - id="unsupported-fields", - ), - ], + "case", + [pytest.param(case, id=case.id) for case in forged_diagnostic_cases()], ) -def test_safe_parse_rejects_forged_custom_error_context( - error_type: Literal["invalid_json_value", "invalid_unicode_surrogate", "unsupported_declared_fields"], - trusted_context: dict[str, object], -) -> None: - # Overlaid on vulnerable 419ad716, these names recover its marker; corrected revisions expose neither. - marker_key = getattr( - configuration_models, - "_INTERNAL_ERROR_CONTEXT_MARKER_KEY", - "_removed_internal_error_context_marker", - ) - marker = getattr(configuration_models, "_INTERNAL_ERROR_CONTEXT_MARKER", object()) - context = { - marker_key: marker, - "pointer": "/forged\npointer-value-canary", - "message": "pydantic-message-canary\nsecond-line-canary", - **trusted_context, - } - forged_mapping = _ForgedValidationContext(error_type, context) - +def test_safe_parse_rejects_forged_custom_error_context(case: ForgedDiagnosticCase) -> None: with pytest.raises(ConfigurationPackageParseError) as caught: - parse_configuration_package(forged_mapping) + parse_configuration_package(case.value) message = str(caught.value) assert message == "configuration package is invalid at /: non-JSON value" - assert not forged_mapping.items_called + case.assert_expected_callbacks() assert len(message.splitlines()) == 1 assert all(character.isprintable() for character in message) assert "canary" not in message - assert "_ForgedValidationContext" not in message - assert error_type not in message + assert case.error_type not in message @pytest.mark.parametrize( - "value", - [ - pytest.param(_ExecutableDict(), id="dict-subclass"), - pytest.param(_ExecutableList(), id="list-subclass"), - pytest.param(_ExecutableStr("string-value-canary"), id="str-subclass"), - pytest.param(_ExecutableInt(7), id="int-subclass"), - pytest.param(_ExecutableFloat(1.5), id="float-subclass"), - ], + "case", + [pytest.param(case, id=case.id) for case in hostile_builtin_cases()], ) -def test_safe_parse_rejects_native_subclasses_without_callbacks(value: object) -> None: +def test_safe_parse_rejects_native_subclasses_without_callbacks(case: BoundaryCase) -> None: data = _package().model_dump(mode="json") - data["configuration"]["source"]["settings"]["hostile\n\u202e~/"] = value + data["configuration"]["source"]["settings"]["hostile\n\u202e~/"] = case.value with pytest.raises(ConfigurationPackageParseError) as caught: parse_configuration_package(data) @@ -1093,74 +936,61 @@ def test_safe_parse_rejects_native_subclasses_without_callbacks(value: object) - assert message == ( r"configuration package is invalid at /configuration/source/settings/hostile\\n\\u202e~0~1: non-JSON value" ) - assert not cast("Any", value).callback_called + case.assert_expected_callbacks() assert len(message.splitlines()) == 1 assert all(character.isprintable() for character in message) assert "canary" not in message - assert type(value).__name__ not in message + assert type(case.value).__name__ not in message @pytest.mark.parametrize( - "root_kind", - [ - "spoofed-class", - "none", - "list", - "string", - "int", - "float", - "dict-subclass", - "constructed-package", - "package-subclass", - "valid-package", - ], + "case", + [pytest.param(case, id=case.id) for case in protocol_object_cases()], ) -def test_safe_parse_rejects_non_dict_roots_without_callbacks(root_kind: str) -> None: - _SpoofedClassValue.callback_called = False - _ExecutableDict.callback_called = False - _ExplosiveConstructedValue.callback_called = False - _ExecutableConfigurationPackage.serializer_called = False - if root_kind == "spoofed-class": - value: object = _SpoofedClassValue() - elif root_kind == "none": - value = None - elif root_kind == "list": - value = [] - elif root_kind == "string": - value = "root-string-canary" - elif root_kind == "int": - value = 7 - elif root_kind == "float": - value = 1.5 - elif root_kind == "dict-subclass": - value = _ExecutableDict() - elif root_kind == "constructed-package": - explosive = _ExplosiveConstructedValue() - value = ConfigurationPackage.model_construct( - format_version=999, - configuration=explosive, - package_metadata=explosive, - credentials=explosive, - ) - elif root_kind == "package-subclass": - value = _ExecutableConfigurationPackage.model_construct( - format_version=1, - configuration="not-json", - package_metadata=None, - credentials=None, - ) - else: - value = _package() +def test_safe_parse_rejects_protocol_objects_without_callbacks(case: BoundaryCase) -> None: + data = _package().model_dump(mode="json") + data["configuration"]["source"]["settings"]["protocol\n\u202e~/"] = case.value + + with pytest.raises(ConfigurationPackageParseError) as caught: + parse_configuration_package(data) + + message = str(caught.value) + assert message == ( + r"configuration package is invalid at /configuration/source/settings/protocol\\n\\u202e~0~1: non-JSON value" + ) + case.assert_expected_callbacks() + assert len(message.splitlines()) == 1 + assert all(character.isprintable() for character in message) + assert "canary" not in message + assert type(case.value).__name__ not in message + + +def _configuration_root_cases() -> tuple[BoundaryCase, ...]: + package = _package() + protocol_cases = tuple(case for case in protocol_object_cases() if case.id != "spoofed-class") + return ( + *root_value_cases(package.model_dump(mode="json")), + *framework_root_cases(ConfigurationPackage, package), + *protocol_cases, + ) + + +@pytest.mark.parametrize( + "case", + [pytest.param(case, id=case.id) for case in _configuration_root_cases()], +) +def test_safe_parse_enforces_exact_dict_roots_without_callbacks(case: BoundaryCase) -> None: + if case.outcome is BoundaryOutcome.ACCEPT: + assert parse_configuration_package(case.value) == _package() + case.assert_expected_callbacks() + return with pytest.raises(ConfigurationPackageParseError) as caught: - parse_configuration_package(value) + parse_configuration_package(case.value) message = str(caught.value) assert message == "configuration package is invalid at /: non-JSON value" - assert not _SpoofedClassValue.callback_called - assert not _ExecutableDict.callback_called - assert not _ExplosiveConstructedValue.callback_called - assert not _ExecutableConfigurationPackage.serializer_called + case.assert_expected_callbacks() assert len(message.splitlines()) == 1 assert all(character.isprintable() for character in message) assert "canary" not in message @@ -1207,87 +1037,66 @@ def test_safe_parse_names_unknown_nested_mapping_field_without_echoing_its_value @pytest.mark.parametrize( - ("control", "visible"), - [ - ("\n", r"\n"), - ("\r", r"\r"), - ("\t", r"\t"), - ("\x1b", r"\u001b"), - ("\x00", r"\u0000"), - ("\x7f", r"\u007f"), - ("\x85", r"\u0085"), - ("\x9f", r"\u009f"), - pytest.param("\u2028", r"\u2028", id="line-separator"), - pytest.param("\u2029", r"\u2029", id="paragraph-separator"), - pytest.param("\u202e", r"\u202e", id="right-to-left-override"), - pytest.param("\u200b", r"\u200b", id="zero-width-space"), - pytest.param("\u2066", r"\u2066", id="left-to-right-isolate"), - pytest.param("\ufeff", r"\ufeff", id="zero-width-no-break-space"), - pytest.param("\U000e0001", r"\U000e0001", id="language-tag"), - ], + "case", + [pytest.param(case, id=case.id) for case in diagnostic_unicode_cases()], ) -def test_safe_parse_renders_unknown_field_controls_without_echoing_values(control: str, visible: str) -> None: +def test_safe_parse_renders_unknown_field_controls_without_echoing_values(case: UnicodeCase) -> None: canary = "control-field-value-canary" data = _package().model_dump(mode="json") - data["configuration"]["source"][f"bad{control}field~/"] = canary + data["configuration"]["source"][f"bad{case.value}field~/"] = canary with pytest.raises(ConfigurationPackageParseError) as caught: parse_configuration_package(data) message = str(caught.value) - assert f"/configuration/source/bad{visible}field~0~1: unsupported declared field" in message + assert f"/configuration/source/bad{case.visible}field~0~1: unsupported declared field" in message assert len(message.splitlines()) == 1 assert all(character.isprintable() for character in message) - assert control not in message + assert case.value not in message assert canary not in message @pytest.mark.parametrize( - ("control", "visible"), - [ - ("\n", r"\n"), - ("\r", r"\r"), - ("\t", r"\t"), - ("\x1b", r"\u001b"), - pytest.param("\u2028", r"\u2028", id="line-separator"), - pytest.param("\u2029", r"\u2029", id="paragraph-separator"), - pytest.param("\u202e", r"\u202e", id="right-to-left-override"), - pytest.param("\u200b", r"\u200b", id="zero-width-space"), - pytest.param("\u2066", r"\u2066", id="left-to-right-isolate"), - pytest.param("\ufeff", r"\ufeff", id="zero-width-no-break-space"), - pytest.param("\U000e0001", r"\U000e0001", id="language-tag"), - ], + "case", + [pytest.param(case, id=case.id) for case in unicode_collision_cases()], ) -def test_safe_parse_distinguishes_controls_from_literal_escape_text(control: str, visible: str) -> None: +def test_safe_parse_distinguishes_controls_from_literal_escape_text(case: UnicodeCollisionCase) -> None: canaries = ("control-value-canary", "literal-value-canary") messages = [] - for field_name, canary in ((f"bad{control}field~/", canaries[0]), (f"bad{visible}field~/", canaries[1])): + for field_name, canary in ( + (f"bad{case.raw}field~/", canaries[0]), + (f"bad{case.literal}field~/", canaries[1]), + ): data = _package().model_dump(mode="json") data["configuration"]["source"][field_name] = canary with pytest.raises(ConfigurationPackageParseError) as caught: parse_configuration_package(data) messages.append(str(caught.value)) - literal_visible = visible.replace("\\", r"\\") - assert f"/configuration/source/bad{visible}field~0~1: unsupported declared field" in messages[0] - assert f"/configuration/source/bad{literal_visible}field~0~1: unsupported declared field" in messages[1] + assert f"/configuration/source/bad{case.raw_visible}field~0~1: unsupported declared field" in messages[0] + assert f"/configuration/source/bad{case.literal_visible}field~0~1: unsupported declared field" in messages[1] assert messages[0] != messages[1] assert all(len(message.splitlines()) == 1 for message in messages) assert all(character.isprintable() for message in messages for character in message) - assert control not in messages[0] + if not case.raw.isprintable(): + assert case.raw not in messages[0] assert all(canary not in message for canary in canaries for message in messages) -def test_safe_parse_preserves_printable_unicode_in_unknown_field_locations() -> None: +@pytest.mark.parametrize( + "case", + [pytest.param(case, id=case.id) for case in valid_unicode_scalar_cases()], +) +def test_safe_parse_preserves_valid_unicode_identity_in_unknown_field_locations(case: UnicodeCase) -> None: canary = "printable-unicode-value-canary" data = _package().model_dump(mode="json") - data["configuration"]["source"]["caf茅-鏉变含-馃榾~/"] = canary + data["configuration"]["source"][f"valid-{case.value}~/"] = canary with pytest.raises(ConfigurationPackageParseError) as caught: parse_configuration_package(data) message = str(caught.value) - assert "/configuration/source/caf茅-鏉变含-馃榾~0~1: unsupported declared field" in message + assert f"/configuration/source/valid-{case.visible}~0~1: unsupported declared field" in message assert canary not in message @@ -2241,6 +2050,35 @@ def test_endpoint_settings_separate_absolute_bases_from_relative_paths( assert str(caught.value) == f"/configuration/source/settings/{setting_name} {expected}" +@pytest.mark.parametrize( + "case", + [pytest.param(case, id=case.id) for case in endpoint_cases()], +) +def test_endpoint_boundaries_handle_hostile_forms_without_echo(case: EndpointCase) -> None: + data = _package().model_dump(mode="json") + settings = { + "url": "https://api.example", + "token": {"$credential": "netbox-token"}, + case.setting_name: case.value, + } + data["configuration"]["source"] = { + "name": "genericrestapi", + "settings": settings, + } + package = ConfigurationPackage.model_validate(data) + + if case.outcome is BoundaryOutcome.ACCEPT: + validate_package_credentials(package) + return + + with pytest.raises(CredentialConfigurationError) as caught: + validate_package_credentials(package) + + assert str(caught.value) == (f"/configuration/source/settings/{case.setting_name} {case.expected_error}") + if case.canary is not None: + assert case.canary not in str(caught.value) + + @pytest.mark.parametrize("endpoint", ["/api/v1", "api", "api/v1/", ""], ids=["rooted", "bare", "trailing", "empty"]) def test_relative_endpoint_forms_remain_accepted(endpoint: str) -> None: data = _package().model_dump(mode="json") diff --git a/tests/hostile_inputs.py b/tests/hostile_inputs.py index cfcdc00b..a33c5338 100644 --- a/tests/hostile_inputs.py +++ b/tests/hostile_inputs.py @@ -140,8 +140,10 @@ class EndpointCase: id: str value: str = field(repr=False) form: str + setting_name: str outcome: BoundaryOutcome canary: str | None = None + expected_error: str | None = None def _plain_boundary_case(case_id: str, value: object, outcome: BoundaryOutcome) -> BoundaryCase: @@ -227,6 +229,8 @@ def __eq__(self, other: object) -> bool: class _HostileStr(str): # noqa: FURB189 - hostile exact-type boundary probe. __slots__ = () + __hash__ = str.__hash__ + def __iter__(self) -> Iterator[str]: # ty: ignore[invalid-method-override] # Deliberate probe. return str_tripwire.trip("str.iter") @@ -666,12 +670,13 @@ def iter_lone_surrogates() -> Iterator[UnicodeCase]: def valid_unicode_scalar_cases() -> tuple[UnicodeCase, ...]: - """Return valid scalars adjacent to surrogates plus representative Unicode.""" + """Return valid scalars plus representative mixed printable Unicode.""" return ( UnicodeCase("before-surrogates", "\ud7ff", r"\ud7ff", "valid"), UnicodeCase("after-surrogates", "\ue000", r"\ue000", "valid"), UnicodeCase("emoji", "馃榾", "馃榾", "valid"), UnicodeCase("maximum-scalar", "\U0010ffff", r"\U0010ffff", "valid"), + UnicodeCase("mixed-printable", "caf茅-鏉变含-馃榾", "caf茅-鏉变含-馃榾", "valid"), ) @@ -679,7 +684,51 @@ def unicode_collision_cases() -> tuple[UnicodeCollisionCase, ...]: """Return raw/literal pairs that unsafe escaping can collapse.""" return ( UnicodeCollisionCase("raw-lf-vs-literal-escape", "\n", r"\n", r"\n", r"\\n"), + UnicodeCollisionCase("raw-cr-vs-literal-escape", "\r", r"\r", r"\r", r"\\r"), + UnicodeCollisionCase("raw-tab-vs-literal-escape", "\t", r"\t", r"\t", r"\\t"), UnicodeCollisionCase("raw-esc-vs-literal-escape", "\x1b", r"\u001b", r"\u001b", r"\\u001b"), + UnicodeCollisionCase( + "raw-line-separator-vs-literal-escape", + "\u2028", + r"\u2028", + r"\u2028", + r"\\u2028", + ), + UnicodeCollisionCase( + "raw-paragraph-separator-vs-literal-escape", + "\u2029", + r"\u2029", + r"\u2029", + r"\\u2029", + ), + UnicodeCollisionCase( + "raw-rtl-override-vs-literal-escape", + "\u202e", + r"\u202e", + r"\u202e", + r"\\u202e", + ), + UnicodeCollisionCase( + "raw-zero-width-space-vs-literal-escape", + "\u200b", + r"\u200b", + r"\u200b", + r"\\u200b", + ), + UnicodeCollisionCase( + "raw-left-to-right-isolate-vs-literal-escape", + "\u2066", + r"\u2066", + r"\u2066", + r"\\u2066", + ), + UnicodeCollisionCase( + "raw-zero-width-no-break-space-vs-literal-escape", + "\ufeff", + r"\ufeff", + r"\ufeff", + r"\\ufeff", + ), UnicodeCollisionCase("backslash", "\\", r"\\", r"\\", r"\\\\"), UnicodeCollisionCase("slash", "/", "~1", "~1", "~01"), UnicodeCollisionCase("tilde", "~", "~0", "~0", "~00"), @@ -696,35 +745,62 @@ def unicode_collision_cases() -> tuple[UnicodeCollisionCase, ...]: def endpoint_cases() -> tuple[EndpointCase, ...]: """Return accepted controls and hostile URL/endpoint forms.""" return ( - EndpointCase("ordinary-absolute", "https://service.example/api", "absolute", BoundaryOutcome.ACCEPT), - EndpointCase("ordinary-authority", "//service.example/api", "authority", BoundaryOutcome.ACCEPT), - EndpointCase("ordinary-relative", "/api/v1/items", "relative", BoundaryOutcome.ACCEPT), + EndpointCase( + "ordinary-absolute", + "https://service.example/api", + "absolute", + "url", + BoundaryOutcome.ACCEPT, + ), + EndpointCase( + "ordinary-authority", + "//service.example/api", + "authority", + "api_endpoint", + BoundaryOutcome.REJECT, + expected_error="must be a relative request path without a scheme or authority", + ), + EndpointCase( + "ordinary-relative", + "/api/v1/items", + "relative", + "api_endpoint", + BoundaryOutcome.ACCEPT, + ), EndpointCase( "userinfo", "https://probe:url-userinfo-canary@service.example/api", "userinfo", + "url", BoundaryOutcome.REJECT, "url-userinfo-canary", + "cannot contain user information, query parameters, or fragments", ), EndpointCase( "query", "https://service.example/api?probe=url-query-canary", "query", + "url", BoundaryOutcome.REJECT, "url-query-canary", + "cannot contain user information, query parameters, or fragments", ), EndpointCase( "fragment", "https://service.example/api#url-fragment-canary", "fragment", + "url", BoundaryOutcome.REJECT, "url-fragment-canary", + "cannot contain user information, query parameters, or fragments", ), EndpointCase( "malformed-authority", "https://[url-authority-canary", "malformed-authority", + "url", BoundaryOutcome.REJECT, "url-authority-canary", + "cannot contain user information, query parameters, or fragments", ), ) diff --git a/tests/test_hostile_inputs.py b/tests/test_hostile_inputs.py index c02f5bf4..ad13f52e 100644 --- a/tests/test_hostile_inputs.py +++ b/tests/test_hostile_inputs.py @@ -112,7 +112,13 @@ def test_unicode_corpora_cover_controls_collisions_and_valid_scalars() -> None: "tilde", "raw-astral-vs-literal-escape", } - assert {ord(case.value) for case in valid_scalars} >= {0xD7FF, 0xE000, 0x1F600, 0x10FFFF} + assert {ord(case.value) for case in valid_scalars if len(case.value) == 1} >= { + 0xD7FF, + 0xE000, + 0x1F600, + 0x10FFFF, + } + assert any(case.value == "caf茅-鏉变含-馃榾" for case in valid_scalars) def test_every_lone_surrogate_is_available_without_expanding_the_default_corpus() -> None: @@ -139,3 +145,5 @@ def test_endpoint_cases_include_unsafe_forms_and_accepted_controls() -> None: assert any(case.outcome is BoundaryOutcome.ACCEPT and case.form == "absolute" for case in cases) assert any(case.outcome is BoundaryOutcome.ACCEPT and case.form == "relative" for case in cases) assert all(case.canary is None or case.canary in case.value for case in cases) + assert all(case.expected_error is None for case in cases if case.outcome is BoundaryOutcome.ACCEPT) + assert all(case.expected_error is not None for case in cases if case.outcome is BoundaryOutcome.REJECT) From 5257cf38ebb93d4a65dd0632cd44bf394a49f3cd Mon Sep 17 00:00:00 2001 From: Blake Ellis Date: Sun, 23 Aug 2026 08:22:02 -0400 Subject: [PATCH 3/7] test: prove forged diagnostic callbacks are live Exercise each forged Pydantic error callback directly before configuration boundaries assert that the callback remains untouched. Co-Authored-By: OpenAI Codex --- tests/hostile_inputs.py | 8 +++++++- tests/test_hostile_inputs.py | 15 +++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/tests/hostile_inputs.py b/tests/hostile_inputs.py index a33c5338..e44f283f 100644 --- a/tests/hostile_inputs.py +++ b/tests/hostile_inputs.py @@ -100,11 +100,16 @@ class ForgedDiagnosticCase: error_type: ForgedErrorType context: dict[str, object] = field(repr=False) tripwire: CallbackTripwire = field(repr=False) + _probe: Callable[[], object] = field(repr=False) expected_callbacks: tuple[str, ...] = () def __repr__(self) -> str: return f"ForgedDiagnosticCase(id={self.id!r})" + def probe_forged_error(self) -> object: + """Execute the hostile callback to prove its forged error is live.""" + return self._probe() + def assert_expected_callbacks(self) -> None: """Assert the boundary did not traverse the forged mapping.""" assert self.tripwire.calls == self.expected_callbacks, ( @@ -605,7 +610,8 @@ def items(self) -> ItemsView[str, object]: context, ) from None - return ForgedDiagnosticCase(case_id, _ForgedMapping(), error_type, context, tripwire) + value = _ForgedMapping() + return ForgedDiagnosticCase(case_id, value, error_type, context, tripwire, value.items) def forged_diagnostic_cases() -> tuple[ForgedDiagnosticCase, ...]: diff --git a/tests/test_hostile_inputs.py b/tests/test_hostile_inputs.py index ad13f52e..9dc1e928 100644 --- a/tests/test_hostile_inputs.py +++ b/tests/test_hostile_inputs.py @@ -4,10 +4,12 @@ import pytest from pydantic import BaseModel, ConfigDict +from pydantic_core import PydanticCustomError from tests.hostile_inputs import ( BoundaryCase, BoundaryOutcome, + ForgedDiagnosticCase, diagnostic_unicode_cases, endpoint_cases, forged_diagnostic_cases, @@ -89,6 +91,19 @@ def test_forged_diagnostic_cases_cover_trusted_shapes_without_private_markers() assert all(case.expected_callbacks == () for case in cases) +@pytest.mark.parametrize( + "case", + [pytest.param(case, id=case.id) for case in forged_diagnostic_cases()], +) +def test_forged_diagnostic_callbacks_raise_the_declared_error(case: ForgedDiagnosticCase) -> None: + with pytest.raises(PydanticCustomError) as caught: + case.probe_forged_error() + + assert caught.value.type == case.error_type + assert caught.value.context == case.context + assert case.tripwire.calls == ("forged-mapping.items",) + + def test_unicode_corpora_cover_controls_collisions_and_valid_scalars() -> None: controls = diagnostic_unicode_cases() collisions = unicode_collision_cases() From dbb16dfbc4b690fbb70258cdf9722af0daf9ff02 Mon Sep 17 00:00:00 2001 From: Blake Ellis Date: Sun, 23 Aug 2026 08:31:44 -0400 Subject: [PATCH 4/7] test: preserve forged-marker and framework callback attacks Retain the recovered-marker regression probe and prove constructed-model and serializer tripwires execute when called directly. Co-Authored-By: OpenAI Codex --- tests/configuration/test_contracts.py | 10 ++++++++++ tests/test_hostile_inputs.py | 18 +++++++++++++++++- 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/tests/configuration/test_contracts.py b/tests/configuration/test_contracts.py index f498da7b..d5ea8316 100644 --- a/tests/configuration/test_contracts.py +++ b/tests/configuration/test_contracts.py @@ -909,6 +909,16 @@ def startswith( # ty: ignore[invalid-method-override] # Hostile test probe. [pytest.param(case, id=case.id) for case in forged_diagnostic_cases()], ) def test_safe_parse_rejects_forged_custom_error_context(case: ForgedDiagnosticCase) -> None: + # If a private marker is ever reinstated, prove an attacker who recovers it still cannot + # authenticate a callback-bearing root. The reusable harness itself stays marker-agnostic. + marker_key = getattr( + configuration_models, + "_INTERNAL_ERROR_CONTEXT_MARKER_KEY", + "_removed_internal_error_context_marker", + ) + marker = getattr(configuration_models, "_INTERNAL_ERROR_CONTEXT_MARKER", object()) + case.context[marker_key] = marker + with pytest.raises(ConfigurationPackageParseError) as caught: parse_configuration_package(case.value) diff --git a/tests/test_hostile_inputs.py b/tests/test_hostile_inputs.py index 9dc1e928..b41c3e5f 100644 --- a/tests/test_hostile_inputs.py +++ b/tests/test_hostile_inputs.py @@ -4,7 +4,7 @@ import pytest from pydantic import BaseModel, ConfigDict -from pydantic_core import PydanticCustomError +from pydantic_core import PydanticCustomError, PydanticSerializationError from tests.hostile_inputs import ( BoundaryCase, @@ -65,6 +65,22 @@ def test_root_factories_classify_exact_dict_and_framework_bypasses() -> None: assert all(case.expected_callbacks == () for case in cases) +def _framework_callback_cases() -> tuple[BoundaryCase, ...]: + valid_model = _ExampleModel.model_validate({"value": 1}) + return tuple(case for case in framework_root_cases(_ExampleModel, valid_model) if case.probed_callback is not None) + + +@pytest.mark.parametrize( + "case", + [pytest.param(case, id=case.id) for case in _framework_callback_cases()], +) +def test_framework_case_callback_probes_are_live(case: BoundaryCase) -> None: + with pytest.raises((AssertionError, PydanticSerializationError), match="hostile callback executed"): + case.probe_callback() + + assert case.tripwire.calls == (case.probed_callback,) + + def test_invalid_json_cases_cover_every_required_graph_shape() -> None: cases = invalid_json_cases() From 0673bb21f3c9c7bf395077ddf4ac35cdab4cc391 Mon Sep 17 00:00:00 2001 From: Blake Ellis Date: Sun, 23 Aug 2026 08:58:23 -0400 Subject: [PATCH 5/7] test: pin hostile corpus and clarify case contracts Co-Authored-By: Codex --- tests/hostile_inputs.py | 4 +- tests/test_hostile_inputs.py | 121 ++++++++++++++++++++++++----------- 2 files changed, 86 insertions(+), 39 deletions(-) diff --git a/tests/hostile_inputs.py b/tests/hostile_inputs.py index e44f283f..d2b66892 100644 --- a/tests/hostile_inputs.py +++ b/tests/hostile_inputs.py @@ -119,7 +119,7 @@ def assert_expected_callbacks(self) -> None: @dataclass(frozen=True) class UnicodeCase: - """One Unicode scalar and its safe visible form.""" + """One Unicode test value and its safe visible form.""" id: str value: str = field(repr=False) @@ -194,7 +194,7 @@ def __eq__(self, other: object) -> bool: def hostile_builtin_cases() -> tuple[BoundaryCase, ...]: - """Return fresh hostile subclasses of every JSON-adjacent built-in type.""" + """Return fresh hostile subclasses of each subclassable JSON built-in value type.""" cases = [_hostile_dict_case()] list_tripwire = CallbackTripwire() diff --git a/tests/test_hostile_inputs.py b/tests/test_hostile_inputs.py index b41c3e5f..ef87e5ed 100644 --- a/tests/test_hostile_inputs.py +++ b/tests/test_hostile_inputs.py @@ -31,9 +31,27 @@ class _ExampleModel(BaseModel): def test_hostile_case_ids_are_unique_and_repr_is_callback_safe() -> None: - cases = (*hostile_builtin_cases(), *protocol_object_cases()) + builtins = hostile_builtin_cases() + protocols = protocol_object_cases() + cases = (*builtins, *protocols) assert len({case.id for case in cases}) == len(cases) + assert {case.id for case in builtins} == { + "dict-subclass", + "list-subclass", + "str-subclass", + "int-subclass", + "float-subclass", + } + assert {case.id for case in protocols} == { + "custom-mapping", + "custom-iterator", + "generator", + "repr-str-format-trap", + "attribute-property-trap", + "spoofed-class", + } + assert all(case.outcome is BoundaryOutcome.REJECT for case in cases) assert all(case.id in repr(case) for case in cases) assert all(case.tripwire.calls == () for case in cases) @@ -57,11 +75,19 @@ def test_root_factories_classify_exact_dict_and_framework_bypasses() -> None: *framework_root_cases(_ExampleModel, valid_model), ) - outcomes = {case.id: case.outcome for case in cases} - assert outcomes["exact-dict"] is BoundaryOutcome.ACCEPT - assert outcomes["valid-model"] is BoundaryOutcome.REJECT - assert outcomes["constructed-invalid-model"] is BoundaryOutcome.REJECT - assert outcomes["model-subclass"] is BoundaryOutcome.REJECT + assert {case.id: case.outcome for case in cases} == { + "exact-dict": BoundaryOutcome.ACCEPT, + "none": BoundaryOutcome.REJECT, + "list": BoundaryOutcome.REJECT, + "string": BoundaryOutcome.REJECT, + "int": BoundaryOutcome.REJECT, + "float": BoundaryOutcome.REJECT, + "dict-subclass": BoundaryOutcome.REJECT, + "spoofed-class": BoundaryOutcome.REJECT, + "valid-model": BoundaryOutcome.REJECT, + "constructed-invalid-model": BoundaryOutcome.REJECT, + "model-subclass": BoundaryOutcome.REJECT, + } assert all(case.expected_callbacks == () for case in cases) @@ -84,6 +110,14 @@ def test_framework_case_callback_probes_are_live(case: BoundaryCase) -> None: def test_invalid_json_cases_cover_every_required_graph_shape() -> None: cases = invalid_json_cases() + assert {case.id for case in cases} == { + "excessive-depth", + "recursive-list", + "recursive-mapping", + "non-string-key", + "non-finite-float", + "non-json-value", + } assert {case.reason for case in cases} == { "maximum declared-content depth exceeded", "recursive list", @@ -92,16 +126,15 @@ def test_invalid_json_cases_cover_every_required_graph_shape() -> None: "non-finite float", "non-JSON value", } - assert len({case.id for case in cases}) == len(cases) def test_forged_diagnostic_cases_cover_trusted_shapes_without_private_markers() -> None: cases = forged_diagnostic_cases() - assert {case.error_type for case in cases} == { - "invalid_json_value", - "invalid_unicode_surrogate", - "unsupported_declared_fields", + assert {case.id: case.error_type for case in cases} == { + "json-value": "invalid_json_value", + "unicode-surrogate": "invalid_unicode_surrogate", + "unsupported-fields": "unsupported_declared_fields", } assert all("marker" not in key for case in cases for key in case.context) assert all(case.expected_callbacks == () for case in cases) @@ -125,31 +158,47 @@ def test_unicode_corpora_cover_controls_collisions_and_valid_scalars() -> None: collisions = unicode_collision_cases() valid_scalars = valid_unicode_scalar_cases() - assert {case.group for case in controls} >= { - "c0", - "del", - "c1", - "bidi", - "isolate", - "zero-width", - "separator", - "astral", + assert {case.id for case in controls} == { + "nul", + "tab", + "lf", + "cr", + "escape", + "delete", + "next-line", + "application-program-command", + "right-to-left-override", + "left-to-right-isolate", + "zero-width-space", + "zero-width-no-break-space", + "line-separator", + "paragraph-separator", + "language-tag", } - assert {case.id for case in collisions} >= { + assert {case.id for case in collisions} == { "raw-lf-vs-literal-escape", + "raw-cr-vs-literal-escape", + "raw-tab-vs-literal-escape", "raw-esc-vs-literal-escape", + "raw-line-separator-vs-literal-escape", + "raw-paragraph-separator-vs-literal-escape", + "raw-rtl-override-vs-literal-escape", + "raw-zero-width-space-vs-literal-escape", + "raw-left-to-right-isolate-vs-literal-escape", + "raw-zero-width-no-break-space-vs-literal-escape", "backslash", "slash", "tilde", "raw-astral-vs-literal-escape", } - assert {ord(case.value) for case in valid_scalars if len(case.value) == 1} >= { - 0xD7FF, - 0xE000, - 0x1F600, - 0x10FFFF, + assert {case.id for case in valid_scalars} == { + "before-surrogates", + "after-surrogates", + "emoji", + "maximum-scalar", + "mixed-printable", } - assert any(case.value == "caf茅-鏉变含-馃榾" for case in valid_scalars) + assert {case.value for case in valid_scalars} == {"\ud7ff", "\ue000", "馃榾", "\U0010ffff", "caf茅-鏉变含-馃榾"} def test_every_lone_surrogate_is_available_without_expanding_the_default_corpus() -> None: @@ -164,17 +213,15 @@ def test_every_lone_surrogate_is_available_without_expanding_the_default_corpus( def test_endpoint_cases_include_unsafe_forms_and_accepted_controls() -> None: cases = endpoint_cases() - assert {case.form for case in cases} >= { - "absolute", - "authority", - "userinfo", - "query", - "fragment", - "malformed-authority", - "relative", + assert {case.id: (case.form, case.setting_name, case.outcome) for case in cases} == { + "ordinary-absolute": ("absolute", "url", BoundaryOutcome.ACCEPT), + "ordinary-authority": ("authority", "api_endpoint", BoundaryOutcome.REJECT), + "ordinary-relative": ("relative", "api_endpoint", BoundaryOutcome.ACCEPT), + "userinfo": ("userinfo", "url", BoundaryOutcome.REJECT), + "query": ("query", "url", BoundaryOutcome.REJECT), + "fragment": ("fragment", "url", BoundaryOutcome.REJECT), + "malformed-authority": ("malformed-authority", "url", BoundaryOutcome.REJECT), } - assert any(case.outcome is BoundaryOutcome.ACCEPT and case.form == "absolute" for case in cases) - assert any(case.outcome is BoundaryOutcome.ACCEPT and case.form == "relative" for case in cases) assert all(case.canary is None or case.canary in case.value for case in cases) assert all(case.expected_error is None for case in cases if case.outcome is BoundaryOutcome.ACCEPT) assert all(case.expected_error is not None for case in cases if case.outcome is BoundaryOutcome.REJECT) From c67cea5ffeffba6464f0adbd82c91461294a58a8 Mon Sep 17 00:00:00 2001 From: Blake Ellis Date: Sun, 23 Aug 2026 09:21:17 -0400 Subject: [PATCH 6/7] test: bind hostile cases to exact attack specifications Co-Authored-By: Codex --- tests/hostile_inputs.py | 14 ++- tests/test_hostile_inputs.py | 234 ++++++++++++++++++++++++----------- 2 files changed, 176 insertions(+), 72 deletions(-) diff --git a/tests/hostile_inputs.py b/tests/hostile_inputs.py index d2b66892..4c2a18b9 100644 --- a/tests/hostile_inputs.py +++ b/tests/hostile_inputs.py @@ -531,6 +531,17 @@ def __str__(self) -> str: subclass_tripwire = CallbackTripwire() + class _SubclassValue: + def __getattribute__(self, name: str) -> object: + del name + return subclass_tripwire.trip("model-subclass.attribute") + + def __repr__(self) -> str: + return subclass_tripwire.trip("model-subclass.repr") + + def __str__(self) -> str: + return subclass_tripwire.trip("model-subclass.str") + @model_serializer def _serialize(self: BaseModel) -> dict[str, object]: del self @@ -540,7 +551,8 @@ def _serialize(self: BaseModel) -> dict[str, object]: "type[BaseModel]", type("_HostileModel", (model_type,), {"__module__": __name__, "_serialize": _serialize}), ) - subclass_fields: dict[str, Any] = dict.fromkeys(model_type.model_fields, constructed_value) + subclass_value = _SubclassValue() + subclass_fields: dict[str, Any] = dict.fromkeys(model_type.model_fields, subclass_value) subclass_model = hostile_model_type.model_construct(**subclass_fields) subclass_case = BoundaryCase( "model-subclass", diff --git a/tests/test_hostile_inputs.py b/tests/test_hostile_inputs.py index ef87e5ed..94a7a0dc 100644 --- a/tests/test_hostile_inputs.py +++ b/tests/test_hostile_inputs.py @@ -36,22 +36,21 @@ def test_hostile_case_ids_are_unique_and_repr_is_callback_safe() -> None: cases = (*builtins, *protocols) assert len({case.id for case in cases}) == len(cases) - assert {case.id for case in builtins} == { - "dict-subclass", - "list-subclass", - "str-subclass", - "int-subclass", - "float-subclass", + assert {case.id: (case.outcome, case.probed_callback) for case in builtins} == { + "dict-subclass": (BoundaryOutcome.REJECT, "dict.items"), + "list-subclass": (BoundaryOutcome.REJECT, "list.iter"), + "str-subclass": (BoundaryOutcome.REJECT, "str.str"), + "int-subclass": (BoundaryOutcome.REJECT, "int.convert"), + "float-subclass": (BoundaryOutcome.REJECT, "float.convert"), } - assert {case.id for case in protocols} == { - "custom-mapping", - "custom-iterator", - "generator", - "repr-str-format-trap", - "attribute-property-trap", - "spoofed-class", + assert {case.id: (case.outcome, case.probed_callback) for case in protocols} == { + "custom-mapping": (BoundaryOutcome.REJECT, "mapping.items"), + "custom-iterator": (BoundaryOutcome.REJECT, "iterator.next"), + "generator": (BoundaryOutcome.REJECT, "generator.next"), + "repr-str-format-trap": (BoundaryOutcome.REJECT, "object.repr"), + "attribute-property-trap": (BoundaryOutcome.REJECT, "object.attribute"), + "spoofed-class": (BoundaryOutcome.REJECT, "object.__class__"), } - assert all(case.outcome is BoundaryOutcome.REJECT for case in cases) assert all(case.id in repr(case) for case in cases) assert all(case.tripwire.calls == () for case in cases) @@ -107,6 +106,16 @@ def test_framework_case_callback_probes_are_live(case: BoundaryCase) -> None: assert case.tripwire.calls == (case.probed_callback,) +def test_model_subclass_nested_values_share_the_case_tripwire() -> None: + case = next(case for case in _framework_callback_cases() if case.id == "model-subclass") + nested_value = vars(case.value)["value"] + + with pytest.raises(AssertionError, match="hostile callback executed"): + _ = nested_value.payload + + assert case.tripwire.calls == ("model-subclass.attribute",) + + def test_invalid_json_cases_cover_every_required_graph_shape() -> None: cases = invalid_json_cases() @@ -131,10 +140,30 @@ def test_invalid_json_cases_cover_every_required_graph_shape() -> None: def test_forged_diagnostic_cases_cover_trusted_shapes_without_private_markers() -> None: cases = forged_diagnostic_cases() - assert {case.id: case.error_type for case in cases} == { - "json-value": "invalid_json_value", - "unicode-surrogate": "invalid_unicode_surrogate", - "unsupported-fields": "unsupported_declared_fields", + assert {case.id: (case.error_type, case.context) for case in cases} == { + "json-value": ( + "invalid_json_value", + { + "pointer": "/forged\npointer-value-canary", + "reason": "non-JSON value", + "message": "pydantic-message-canary\nsecond-line-canary", + }, + ), + "unicode-surrogate": ( + "invalid_unicode_surrogate", + { + "pointer": "/forged\npointer-value-canary", + "message": "pydantic-message-canary\nsecond-line-canary", + }, + ), + "unsupported-fields": ( + "unsupported_declared_fields", + { + "pointer": "/forged\npointer-value-canary", + "field_names": ("forged-field-name-canary",), + "message": "pydantic-message-canary\nsecond-line-canary", + }, + ), } assert all("marker" not in key for case in cases for key in case.context) assert all(case.expected_callbacks == () for case in cases) @@ -158,70 +187,133 @@ def test_unicode_corpora_cover_controls_collisions_and_valid_scalars() -> None: collisions = unicode_collision_cases() valid_scalars = valid_unicode_scalar_cases() - assert {case.id for case in controls} == { - "nul", - "tab", - "lf", - "cr", - "escape", - "delete", - "next-line", - "application-program-command", - "right-to-left-override", - "left-to-right-isolate", - "zero-width-space", - "zero-width-no-break-space", - "line-separator", - "paragraph-separator", - "language-tag", - } - assert {case.id for case in collisions} == { - "raw-lf-vs-literal-escape", - "raw-cr-vs-literal-escape", - "raw-tab-vs-literal-escape", - "raw-esc-vs-literal-escape", - "raw-line-separator-vs-literal-escape", - "raw-paragraph-separator-vs-literal-escape", - "raw-rtl-override-vs-literal-escape", - "raw-zero-width-space-vs-literal-escape", - "raw-left-to-right-isolate-vs-literal-escape", - "raw-zero-width-no-break-space-vs-literal-escape", - "backslash", - "slash", - "tilde", - "raw-astral-vs-literal-escape", - } - assert {case.id for case in valid_scalars} == { - "before-surrogates", - "after-surrogates", - "emoji", - "maximum-scalar", - "mixed-printable", - } - assert {case.value for case in valid_scalars} == {"\ud7ff", "\ue000", "馃榾", "\U0010ffff", "caf茅-鏉变含-馃榾"} + assert tuple((case.id, case.value, case.visible, case.group) for case in controls) == ( + ("nul", "\x00", r"\u0000", "c0"), + ("tab", "\t", r"\t", "c0"), + ("lf", "\n", r"\n", "c0"), + ("cr", "\r", r"\r", "c0"), + ("escape", "\x1b", r"\u001b", "c0"), + ("delete", "\x7f", r"\u007f", "del"), + ("next-line", "\x85", r"\u0085", "c1"), + ("application-program-command", "\x9f", r"\u009f", "c1"), + ("right-to-left-override", "\u202e", r"\u202e", "bidi"), + ("left-to-right-isolate", "\u2066", r"\u2066", "isolate"), + ("zero-width-space", "\u200b", r"\u200b", "zero-width"), + ("zero-width-no-break-space", "\ufeff", r"\ufeff", "zero-width"), + ("line-separator", "\u2028", r"\u2028", "separator"), + ("paragraph-separator", "\u2029", r"\u2029", "separator"), + ("language-tag", "\U000e0001", r"\U000e0001", "astral"), + ) + assert tuple((case.id, case.raw, case.literal, case.raw_visible, case.literal_visible) for case in collisions) == ( + ("raw-lf-vs-literal-escape", "\n", r"\n", r"\n", r"\\n"), + ("raw-cr-vs-literal-escape", "\r", r"\r", r"\r", r"\\r"), + ("raw-tab-vs-literal-escape", "\t", r"\t", r"\t", r"\\t"), + ("raw-esc-vs-literal-escape", "\x1b", r"\u001b", r"\u001b", r"\\u001b"), + ("raw-line-separator-vs-literal-escape", "\u2028", r"\u2028", r"\u2028", r"\\u2028"), + ("raw-paragraph-separator-vs-literal-escape", "\u2029", r"\u2029", r"\u2029", r"\\u2029"), + ("raw-rtl-override-vs-literal-escape", "\u202e", r"\u202e", r"\u202e", r"\\u202e"), + ("raw-zero-width-space-vs-literal-escape", "\u200b", r"\u200b", r"\u200b", r"\\u200b"), + ("raw-left-to-right-isolate-vs-literal-escape", "\u2066", r"\u2066", r"\u2066", r"\\u2066"), + ( + "raw-zero-width-no-break-space-vs-literal-escape", + "\ufeff", + r"\ufeff", + r"\ufeff", + r"\\ufeff", + ), + ("backslash", "\\", r"\\", r"\\", r"\\\\"), + ("slash", "/", "~1", "~1", "~01"), + ("tilde", "~", "~0", "~0", "~00"), + ( + "raw-astral-vs-literal-escape", + "\U000e0001", + r"\U000e0001", + r"\U000e0001", + r"\\U000e0001", + ), + ) + assert tuple((case.id, case.value, case.visible, case.group) for case in valid_scalars) == ( + ("before-surrogates", "\ud7ff", r"\ud7ff", "valid"), + ("after-surrogates", "\ue000", r"\ue000", "valid"), + ("emoji", "馃榾", "馃榾", "valid"), + ("maximum-scalar", "\U0010ffff", r"\U0010ffff", "valid"), + ("mixed-printable", "caf茅-鏉变含-馃榾", "caf茅-鏉变含-馃榾", "valid"), + ) def test_every_lone_surrogate_is_available_without_expanding_the_default_corpus() -> None: cases = tuple(iter_lone_surrogates()) - assert len(cases) == 0x800 - assert ord(cases[0].value) == 0xD800 - assert ord(cases[-1].value) == 0xDFFF + assert tuple((case.id, case.value, case.visible, case.group) for case in cases) == tuple( + (f"U+{codepoint:04X}", chr(codepoint), f"\\u{codepoint:04x}", "surrogate") + for codepoint in range(0xD800, 0xE000) + ) assert len(diagnostic_unicode_cases()) < 32 def test_endpoint_cases_include_unsafe_forms_and_accepted_controls() -> None: cases = endpoint_cases() - assert {case.id: (case.form, case.setting_name, case.outcome) for case in cases} == { - "ordinary-absolute": ("absolute", "url", BoundaryOutcome.ACCEPT), - "ordinary-authority": ("authority", "api_endpoint", BoundaryOutcome.REJECT), - "ordinary-relative": ("relative", "api_endpoint", BoundaryOutcome.ACCEPT), - "userinfo": ("userinfo", "url", BoundaryOutcome.REJECT), - "query": ("query", "url", BoundaryOutcome.REJECT), - "fragment": ("fragment", "url", BoundaryOutcome.REJECT), - "malformed-authority": ("malformed-authority", "url", BoundaryOutcome.REJECT), - } + assert tuple( + ( + case.id, + case.value, + case.form, + case.setting_name, + case.outcome, + case.canary, + case.expected_error, + ) + for case in cases + ) == ( + ("ordinary-absolute", "https://service.example/api", "absolute", "url", BoundaryOutcome.ACCEPT, None, None), + ( + "ordinary-authority", + "//service.example/api", + "authority", + "api_endpoint", + BoundaryOutcome.REJECT, + None, + "must be a relative request path without a scheme or authority", + ), + ("ordinary-relative", "/api/v1/items", "relative", "api_endpoint", BoundaryOutcome.ACCEPT, None, None), + ( + "userinfo", + "https://probe:url-userinfo-canary@service.example/api", + "userinfo", + "url", + BoundaryOutcome.REJECT, + "url-userinfo-canary", + "cannot contain user information, query parameters, or fragments", + ), + ( + "query", + "https://service.example/api?probe=url-query-canary", + "query", + "url", + BoundaryOutcome.REJECT, + "url-query-canary", + "cannot contain user information, query parameters, or fragments", + ), + ( + "fragment", + "https://service.example/api#url-fragment-canary", + "fragment", + "url", + BoundaryOutcome.REJECT, + "url-fragment-canary", + "cannot contain user information, query parameters, or fragments", + ), + ( + "malformed-authority", + "https://[url-authority-canary", + "malformed-authority", + "url", + BoundaryOutcome.REJECT, + "url-authority-canary", + "cannot contain user information, query parameters, or fragments", + ), + ) assert all(case.canary is None or case.canary in case.value for case in cases) assert all(case.expected_error is None for case in cases if case.outcome is BoundaryOutcome.ACCEPT) assert all(case.expected_error is not None for case in cases if case.outcome is BoundaryOutcome.REJECT) From 93a82b989f90fdf7fbf3648ab40f2ef6ef76a00c Mon Sep 17 00:00:00 2001 From: Blake Ellis Date: Sun, 23 Aug 2026 09:25:12 -0400 Subject: [PATCH 7/7] test: pin exact hostile root payloads Co-Authored-By: Codex --- tests/test_hostile_inputs.py | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/tests/test_hostile_inputs.py b/tests/test_hostile_inputs.py index 94a7a0dc..9e99f67f 100644 --- a/tests/test_hostile_inputs.py +++ b/tests/test_hostile_inputs.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import cast + import pytest from pydantic import BaseModel, ConfigDict from pydantic_core import PydanticCustomError, PydanticSerializationError @@ -74,7 +76,8 @@ def test_root_factories_classify_exact_dict_and_framework_bypasses() -> None: *framework_root_cases(_ExampleModel, valid_model), ) - assert {case.id: case.outcome for case in cases} == { + by_id = {case.id: case for case in cases} + assert {case_id: case.outcome for case_id, case in by_id.items()} == { "exact-dict": BoundaryOutcome.ACCEPT, "none": BoundaryOutcome.REJECT, "list": BoundaryOutcome.REJECT, @@ -87,6 +90,28 @@ def test_root_factories_classify_exact_dict_and_framework_bypasses() -> None: "constructed-invalid-model": BoundaryOutcome.REJECT, "model-subclass": BoundaryOutcome.REJECT, } + assert type(by_id["exact-dict"].value) is dict + assert by_id["exact-dict"].value == valid_mapping + assert by_id["none"].value is None + assert type(by_id["list"].value) is list + assert by_id["list"].value == [] + assert type(by_id["string"].value) is str + assert by_id["string"].value == "root-string-canary" + assert type(by_id["int"].value) is int + assert by_id["int"].value == 7 + assert type(by_id["float"].value) is float + assert cast("float", by_id["float"].value).as_integer_ratio() == (3, 2) + assert type(by_id["dict-subclass"].value).__name__ == "_HostileDict" + assert issubclass(type(by_id["dict-subclass"].value), dict) + hostile_dict = cast("dict[object, object]", by_id["dict-subclass"].value) + assert dict.__len__(hostile_dict) == 0 # noqa: PLC2801 - bypass hostile hooks. + assert type(by_id["spoofed-class"].value).__name__ == "_SpoofedClass" + assert by_id["valid-model"].value is valid_model + assert type(by_id["constructed-invalid-model"].value) is _ExampleModel + assert type(vars(by_id["constructed-invalid-model"].value)["value"]).__name__ == "_ConstructedValue" + assert type(by_id["model-subclass"].value).__name__ == "_HostileModel" + assert issubclass(type(by_id["model-subclass"].value), _ExampleModel) + assert type(vars(by_id["model-subclass"].value)["value"]).__name__ == "_SubclassValue" assert all(case.expected_callbacks == () for case in cases)