fix(validation): harden policy scalar trust boundary

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.

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]:

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)