diff --git a/tests/configuration/test_contracts.py b/tests/configuration/test_contracts.py index d9abc923..d5ea8316 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,39 @@ 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. +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()) - context = { - marker_key: marker, - "pointer": "/forged\npointer-value-canary", - "message": "pydantic-message-canary\nsecond-line-canary", - **trusted_context, - } - forged_mapping = _ForgedValidationContext(error_type, context) + case.context[marker_key] = marker 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 +946,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(value) + 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(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 +1047,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 +2060,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 new file mode 100644 index 00000000..4c2a18b9 --- /dev/null +++ b/tests/hostile_inputs.py @@ -0,0 +1,824 @@ +# 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) + _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, ( + f"{self.id}: callbacks {self.tripwire.calls!r}, expected {self.expected_callbacks!r}" + ) + + +@dataclass(frozen=True) +class UnicodeCase: + """One Unicode test value 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 + 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: + 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 each subclassable JSON built-in value 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__ = () + + __hash__ = str.__hash__ + + 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() + + 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 + return subclass_tripwire.trip("model.model_dump") + + hostile_model_type = cast( + "type[BaseModel]", + type("_HostileModel", (model_type,), {"__module__": __name__, "_serialize": _serialize}), + ) + 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", + 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 + + value = _ForgedMapping() + return ForgedDiagnosticCase(case_id, value, error_type, context, tripwire, value.items) + + +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 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"), + ) + + +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"), + 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", + "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 new file mode 100644 index 00000000..9e99f67f --- /dev/null +++ b/tests/test_hostile_inputs.py @@ -0,0 +1,344 @@ +"""Contract tests for the reusable hostile-input test harness.""" + +from __future__ import annotations + +from typing import cast + +import pytest +from pydantic import BaseModel, ConfigDict +from pydantic_core import PydanticCustomError, PydanticSerializationError + +from tests.hostile_inputs import ( + BoundaryCase, + BoundaryOutcome, + ForgedDiagnosticCase, + 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: + builtins = hostile_builtin_cases() + protocols = protocol_object_cases() + cases = (*builtins, *protocols) + + assert len({case.id for case in cases}) == len(cases) + 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: (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.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), + ) + + 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, + "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 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) + + +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_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() + + 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", + "recursive mapping", + "non-string mapping key", + "non-finite float", + "non-JSON value", + } + + +def test_forged_diagnostic_cases_cover_trusted_shapes_without_private_markers() -> None: + cases = forged_diagnostic_cases() + + 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) + + +@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() + valid_scalars = valid_unicode_scalar_cases() + + 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 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 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)