diff --git a/docs/changelog.d/validation-policy-callback-safety.md b/docs/changelog.d/validation-policy-callback-safety.md new file mode 100644 index 000000000..2fb275c6e --- /dev/null +++ b/docs/changelog.d/validation-policy-callback-safety.md @@ -0,0 +1,7 @@ +# Harden validation-policy scalar trust boundaries + +## Security + +- Reject caller-defined string and numeric subclasses at `ValidationPolicy` construction before `strip`, numeric conversion, or comparison callbacks can execute. +- Normalize only exact built-in and package-trusted NumPy real scalar identities for scoring-policy thresholds while preserving the existing closed `0..1` domains and Rust-owned pass/fail arithmetic. +- Require an exact built-in integer for `min_subgroup_n` before range comparison and preserve the existing `rust_kwargs()` payload contract. diff --git a/python/fast_mlsirm/validation.py b/python/fast_mlsirm/validation.py index 38bf67641..09e6ab1b9 100644 --- a/python/fast_mlsirm/validation.py +++ b/python/fast_mlsirm/validation.py @@ -21,6 +21,12 @@ np.dtype(code).type for code in ("b", "B", "h", "H", "i", "I", "l", "L", "q", "Q", "p", "P") ) +_TRUSTED_NUMPY_FLOAT_SCALAR_TYPES = ( + np.float16, + np.float32, + np.float64, + np.longdouble, +) def _is_exact_numpy_integer_scalar_type(value_type: type) -> bool: @@ -33,6 +39,25 @@ def _is_exact_numpy_integer_scalar_type(value_type: type) -> bool: return any(value_type is trusted_type for trusted_type in _TRUSTED_NUMPY_INTEGER_SCALAR_TYPES) +def _trusted_policy_real(value: object, name: str) -> float: + """Normalize one trusted policy threshold without caller callback dispatch.""" + value_type = type(value) + if not ( + value_type is int + or value_type is float + or _is_exact_numpy_integer_scalar_type(value_type) + or any(value_type is scalar_type for scalar_type in _TRUSTED_NUMPY_FLOAT_SCALAR_TYPES) + ): + raise ValueError(f"{name} must be a real number in 0..1") + try: + normalized = float(value) + except (OverflowError, ValueError) as exc: + raise ValueError(f"{name} must be a real number in 0..1") from exc + if not (0.0 <= normalized <= 1.0): + raise ValueError(f"{name} must be in 0..1") + return normalized + + def _trusted_judge_category_count(value: object) -> int: """Return a trusted built-in category count without caller coercion callbacks. @@ -105,9 +130,9 @@ class ValidationPolicy: def __post_init__(self) -> None: """Reject empty identities and thresholds outside closed unit intervals.""" - if not isinstance(self.policy_id, str) or not self.policy_id.strip(): + if type(self.policy_id) is not str or not self.policy_id.strip(): raise ValueError("policy_id must be a non-empty string") - if not isinstance(self.policy_version, str) or not self.policy_version.strip(): + if type(self.policy_version) is not str or not self.policy_version.strip(): raise ValueError("policy_version must be a non-empty string") for name in ( "qwk_min", @@ -116,15 +141,9 @@ def __post_init__(self) -> None: "overall_smd_max", "subgroup_smd_max", ): - value = getattr(self, name) - if not isinstance(value, (int, float)) or isinstance(value, bool): - raise ValueError(f"{name} must be a real number in 0..1") - f = float(value) - if not (0.0 <= f <= 1.0): - raise ValueError(f"{name} must be in 0..1") - object.__setattr__(self, name, f) + object.__setattr__(self, name, _trusted_policy_real(getattr(self, name), name)) n = self.min_subgroup_n - if not isinstance(n, int) or isinstance(n, bool) or n < 2: + if type(n) is not int or n < 2: raise ValueError("min_subgroup_n must be an integer >= 2") def rust_kwargs(self) -> dict[str, Any]: diff --git a/tests/test_validation_policy_callback_safety.py b/tests/test_validation_policy_callback_safety.py new file mode 100644 index 000000000..c0910d4e7 --- /dev/null +++ b/tests/test_validation_policy_callback_safety.py @@ -0,0 +1,156 @@ +"""Callback-safety regressions for automated-scoring validation policy controls.""" + +from __future__ import annotations + +import numpy as np +import pytest + +from fast_mlsirm.validation import ValidationPolicy + + +class _HostileText(str): + """String subclass that records any caller-dispatchable normalization.""" + + def __new__(cls, value: str, calls: list[str]) -> "_HostileText": + """Attach the callback log to one hostile text scalar.""" + instance = super().__new__(cls, value) + instance.calls = calls + return instance + + def strip(self, *args: object, **kwargs: object) -> str: + """Fail if validation dispatches caller-controlled text normalization.""" + self.calls.append("strip") + raise AssertionError("caller string callback executed") + + +class _HostileFloat(float): + """Float subclass that records caller-dispatchable numeric coercion.""" + + def __new__(cls, value: float, calls: list[str]) -> "_HostileFloat": + """Attach the callback log to one hostile floating scalar.""" + instance = super().__new__(cls, value) + instance.calls = calls + return instance + + def __float__(self) -> float: + """Fail if validation dispatches caller-controlled float conversion.""" + self.calls.append("float") + raise AssertionError("caller float callback executed") + + +class _HostileInt(int): + """Integer subclass that records numeric coercion and comparisons.""" + + def __new__(cls, value: int, calls: list[str]) -> "_HostileInt": + """Attach the callback log to one hostile integer scalar.""" + instance = super().__new__(cls, value) + instance.calls = calls + return instance + + def __float__(self) -> float: + """Fail if validation dispatches caller-controlled float conversion.""" + self.calls.append("float") + raise AssertionError("caller integer float callback executed") + + def __lt__(self, other: object) -> bool: + """Fail if validation dispatches a caller-controlled range comparison.""" + self.calls.append("lt") + raise AssertionError("caller integer comparison executed") + + +@pytest.mark.parametrize("field", ["policy_id", "policy_version"]) +def test_policy_identity_rejects_string_subclass_without_callbacks(field: str) -> None: + """Policy identities reject caller string subclasses before ``strip`` dispatch.""" + calls: list[str] = [] + + with pytest.raises(ValueError, match=rf"{field} must be a non-empty string"): + ValidationPolicy(**{field: _HostileText("trusted-looking", calls)}) + + assert calls == [] + + +@pytest.mark.parametrize( + "field", + [ + "qwk_min", + "pearson_r_min", + "degradation_max", + "overall_smd_max", + "subgroup_smd_max", + ], +) +def test_policy_threshold_rejects_float_subclass_without_callbacks(field: str) -> None: + """Every real threshold rejects caller float subclasses before coercion.""" + calls: list[str] = [] + + with pytest.raises(ValueError, match=rf"{field} must be a real number in 0\.\.1"): + ValidationPolicy(**{field: _HostileFloat(0.5, calls)}) + + assert calls == [] + + +def test_policy_threshold_rejects_integer_subclass_without_callbacks() -> None: + """Numeric admission must not treat an ``int`` subclass as a trusted scalar.""" + calls: list[str] = [] + + with pytest.raises(ValueError, match=r"qwk_min must be a real number in 0\.\.1"): + ValidationPolicy(qwk_min=_HostileInt(1, calls)) + + assert calls == [] + + +def test_min_subgroup_n_rejects_integer_subclass_without_callbacks() -> None: + """The subgroup-size control rejects integer subclasses before comparison.""" + calls: list[str] = [] + + with pytest.raises(ValueError, match=r"min_subgroup_n must be an integer >= 2"): + ValidationPolicy(min_subgroup_n=_HostileInt(2, calls)) + + assert calls == [] + + +def test_policy_builtin_controls_still_normalize_for_rust_marshalling() -> None: + """Trusted built-in policy controls preserve the established Rust payload.""" + policy = ValidationPolicy( + policy_id="research_diagnostic", + policy_version="2.0", + qwk_min=1, + pearson_r_min=0.8, + degradation_max=0, + overall_smd_max=0.2, + subgroup_smd_max=0.1, + min_subgroup_n=3, + ) + + assert policy.policy_id == "research_diagnostic" + assert policy.policy_version == "2.0" + assert type(policy.qwk_min) is float + assert type(policy.degradation_max) is float + assert policy.rust_kwargs() == { + "qwk_min": 1.0, + "pearson_r_min": 0.8, + "degradation_max": 0.0, + "overall_smd_max": 0.2, + "subgroup_smd_max": 0.1, + "min_subgroup_n": 3, + } + + +@pytest.mark.parametrize( + "threshold", + [ + np.float16(0.5), + np.float32(0.5), + np.float64(0.5), + np.longdouble(0.5), + np.int64(1), + ], +) +def test_policy_trusted_numpy_thresholds_marshal_as_builtin_floats(threshold: object) -> None: + """Supported concrete NumPy scalar identities normalize to Rust-ready floats.""" + policy = ValidationPolicy(qwk_min=threshold) + rust_kwargs = policy.rust_kwargs() + + assert type(policy.qwk_min) is float + assert type(rust_kwargs["qwk_min"]) is float + assert rust_kwargs["qwk_min"] == float(threshold)