fix(scoring): harden policy integer callback boundary

docs/changelog.d/868-scoring-policy-integer-callback-safety.md

@@ -0,0 +1,5 @@
+# Harden scoring-policy integer callback boundaries
+
+## Fixed
+
+- Reject caller-defined integer coercion at scoring-policy positive-integer boundaries before any `__index__` callback can run, while preserving exact built-in and genuine NumPy integer scalar compatibility and existing bounded `AssessmentSpecError` semantics.

python/fast_mlsirm/scoring/_contract_safety.py

@@ -6,13 +6,28 @@
 from enum import Enum
 import hashlib
 import json
-import operator
 from typing import Any, TypeVar
 
+import numpy as np
+
 from fast_mlsirm.rubric.models import _identifier, _semantic_version, _text
 
 from . import _validation as base
 
+_NUMPY_INTEGER_SCALAR_TYPES = (
+    np.int8,
+    np.int16,
+    np.int32,
+    np.int64,
+    np.intp,
+    np.longlong,
+    np.uint8,
+    np.uint16,
+    np.uint32,
+    np.uint64,
+    np.uintp,
+    np.ulonglong,
+)
 _SENSITIVE_METADATA_FIELDS = frozenset(
     {
         "answer_text",
@@ -121,37 +136,38 @@ def enum_value(
         ) from None
 
 
+def _has_exact_type(value: Any, trusted_types: tuple[type, ...]) -> bool:
+    """Return whether a control has one exact package-trusted scalar type."""
+    value_type = type(value)
+    return any(value_type is trusted_type for trusted_type in trusted_types)
+
+
 def bounded_positive_integer(
     value: Any,
     name: str,
     maximum: int,
     path: str | None = None,
 ) -> int:
-    """Normalize one positive integer without leaking numeric callbacks."""
+    """Normalize one positive integer without caller-controlled coercion."""
     resolved_path = path or f"$.{name}"
-    if isinstance(value, bool):
+    value_type = type(value)
+    if value_type is int:
+        normalized = value
+    elif _has_exact_type(value, _NUMPY_INTEGER_SCALAR_TYPES):
+        normalized = int(value)
+    else:
         raise base.assessment_error(
             f"invalid_{name}",
             resolved_path,
             f"{name} must be an integer between 1 and {maximum}",
         )
-    try:
-        normalized = operator.index(value)
-    except base.AssessmentSpecError:
-        raise
-    except Exception:
-        raise base.assessment_error(
-            f"invalid_{name}",
-            resolved_path,
-            f"{name} must be an integer between 1 and {maximum}",
-        ) from None
-    if isinstance(normalized, bool) or not 1 <= normalized <= maximum:
+    if not 1 <= normalized <= maximum:
         raise base.assessment_error(
             f"invalid_{name}",
             resolved_path,
             f"{name} must be between 1 and {maximum}",
         )
-    return int(normalized)
+    return normalized
 
 
 def bounded_values(

python/fast_mlsirm/scoring/_validation.py

@@ -7,11 +7,12 @@
 import hashlib
 import json
 import math
-import operator
 import re
 from types import MappingProxyType
 from typing import Any, TypeVar
 
+import numpy as np
+
 from fast_mlsirm.rubric.models import _identifier, _semantic_version, _text
 
 ASSESSMENT_SCHEMA_VERSION = "1.0"
@@ -31,6 +32,20 @@
 MAX_SIGNED_INTEGER = (1 << 63) - 1
 FINGERPRINT_PATTERN = re.compile(r"^[0-9a-f]{64}$")
 _ERROR_CODE_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:_[a-z0-9]+)+$")
+_NUMPY_INTEGER_SCALAR_TYPES = (
+    np.int8,
+    np.int16,
+    np.int32,
+    np.int64,
+    np.intp,
+    np.longlong,
+    np.uint8,
+    np.uint16,
+    np.uint32,
+    np.uint64,
+    np.uintp,
+    np.ulonglong,
+)
 _SENSITIVE_METADATA_FIELDS = frozenset(
     {
         "answer_text",
@@ -190,35 +205,38 @@ def strict_boolean(value: Any, name: str, path: str | None = None) -> bool:
     return value
 
 
+def _has_exact_type(value: Any, trusted_types: tuple[type, ...]) -> bool:
+    """Return whether a control has one exact package-trusted scalar type."""
+    value_type = type(value)
+    return any(value_type is trusted_type for trusted_type in trusted_types)
+
+
 def bounded_positive_integer(
     value: Any,
     name: str,
     maximum: int,
     path: str | None = None,
 ) -> int:
-    """Return a bounded positive integer with stable conversion failures."""
+    """Return a bounded positive integer without caller-controlled coercion."""
     resolved_path = path or f"$.{name}"
-    if isinstance(value, bool):
+    value_type = type(value)
+    if value_type is int:
+        normalized = value
+    elif _has_exact_type(value, _NUMPY_INTEGER_SCALAR_TYPES):
+        normalized = int(value)
+    else:
         raise assessment_error(
             f"invalid_{name}",
             resolved_path,
             f"{name} must be an integer between 1 and {maximum}",
         )
-    try:
-        normalized = operator.index(value)
-    except (TypeError, ValueError, OverflowError):
-        raise assessment_error(
-            f"invalid_{name}",
-            resolved_path,
-            f"{name} must be an integer between 1 and {maximum}",
-        ) from None
-    if isinstance(normalized, bool) or not 1 <= normalized <= maximum:
+    if not 1 <= normalized <= maximum:
         raise assessment_error(
             f"invalid_{name}",
             resolved_path,
             f"{name} must be between 1 and {maximum}",
         )
-    return int(normalized)
+    return normalized
 
 
 def bounded_values(

tests/test_scoring_contract_public_callback_safety.py

@@ -60,10 +60,13 @@ def strip(self, chars=None):
 
 
 class _DomainErrorInteger:
-    """Integer-like fixture that raises an existing package-owned domain error."""
+    """Integer-like fixture whose callback must never cross the trust boundary."""
+
+    calls = 0
 
     def __index__(self):
-        """Raise the shared sentinel domain error unchanged."""
+        """Record forbidden dispatch before raising the shared sentinel error."""
+        type(self).calls += 1
         raise _SENTINEL_ERROR
 
 
@@ -211,8 +214,8 @@ def test_response_type_equality_callback_failure_is_redacted() -> None:
     assert "private enum comparison payload" not in str(captured.value)
 
 
-def test_package_owned_callback_errors_are_preserved_unchanged() -> None:
-    """The public boundary re-raises an existing AssessmentSpecError object."""
+def test_package_owned_text_callback_errors_are_preserved_unchanged() -> None:
+    """String normalization re-raises an existing package-owned domain error."""
     with pytest.raises(AssessmentSpecError) as text_error:
         ConstructSpec(
             construct_id=_DomainErrorString("argument_quality"),
@@ -221,6 +224,10 @@ def test_package_owned_callback_errors_are_preserved_unchanged() -> None:
         )
     assert text_error.value is _SENTINEL_ERROR
 
+
+def test_integer_callback_domain_errors_are_rejected_before_dispatch() -> None:
+    """Untrusted integer callbacks cannot execute even to raise domain errors."""
+    _DomainErrorInteger.calls = 0
     with pytest.raises(AssessmentSpecError) as integer_error:
         EnginePolicy(
             policy_id="engine_policy",
@@ -229,7 +236,11 @@ def test_package_owned_callback_errors_are_preserved_unchanged() -> None:
             allow_automated_raters=False,
             minimum_raters_per_response=_DomainErrorInteger(),  # type: ignore[arg-type]
         )
-    assert integer_error.value is _SENTINEL_ERROR
+
+    assert integer_error.value.code == "invalid_minimum_raters_per_response"
+    assert integer_error.value.path == "$.minimum_raters_per_response"
+    assert integer_error.value is not _SENTINEL_ERROR
+    assert _DomainErrorInteger.calls == 0
 
 
 def test_base_exceptions_are_not_swallowed_by_callback_redaction() -> None:

tests/test_scoring_policy_integer_callback_safety.py

@@ -0,0 +1,90 @@
+"""Callback-safety regressions for scoring-policy integer controls."""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from fast_mlsirm.scoring import AssessmentSpecError, EnginePolicy
+from fast_mlsirm.scoring import _contract_safety, _validation
+
+
+class _HostileIndex:
+    """Index provider whose callback records any attempted coercion."""
+
+    calls = 0
+
+    def __index__(self) -> int:
+        """Record execution and return an otherwise valid policy value."""
+        type(self).calls += 1
+        return 2
+
+
+class _HostileInt(int):
+    """Integer subclass that must not cross the trusted control boundary."""
+
+
+def _assert_rejected_without_index_callback(callable_) -> None:
+    """Require one hostile index provider to fail before callback dispatch."""
+    _HostileIndex.calls = 0
+    with pytest.raises(AssessmentSpecError):
+        callable_(_HostileIndex())
+    assert _HostileIndex.calls == 0
+
+
+def test_engine_policy_rejects_index_provider_without_callback() -> None:
+    """The public policy boundary rejects arbitrary index providers inertly."""
+    _assert_rejected_without_index_callback(
+        lambda value: EnginePolicy(
+            policy_id="engine_policy",
+            engine_ids=(),
+            allow_human_raters=True,
+            allow_automated_raters=False,
+            minimum_raters_per_response=value,
+        )
+    )
+
+
+def test_integer_validators_reject_index_provider_without_callback() -> None:
+    """Both validator layers reject before caller-controlled coercion."""
+    for validator in (
+        _validation.bounded_positive_integer,
+        _contract_safety.bounded_positive_integer,
+    ):
+        _assert_rejected_without_index_callback(
+            lambda value, validator=validator: validator(
+                value,
+                "minimum_raters_per_response",
+                _validation.MAX_RATERS_PER_RESPONSE,
+            )
+        )
+
+
+def test_engine_policy_rejects_integer_subclass() -> None:
+    """A caller-defined integer subclass is not a trusted policy control."""
+    with pytest.raises(AssessmentSpecError) as captured:
+        EnginePolicy(
+            policy_id="engine_policy",
+            engine_ids=(),
+            allow_human_raters=True,
+            allow_automated_raters=False,
+            minimum_raters_per_response=_HostileInt(2),
+        )
+
+    assert captured.value.code == "invalid_minimum_raters_per_response"
+    assert captured.value.path == "$.minimum_raters_per_response"
+
+
+@pytest.mark.parametrize("value", [1, np.int32(2), np.int64(3), np.uint64(4)])
+def test_engine_policy_preserves_trusted_integer_scalars(value: object) -> None:
+    """Built-in and genuine NumPy integer scalars retain compatibility."""
+    policy = EnginePolicy(
+        policy_id="engine_policy",
+        engine_ids=(),
+        allow_human_raters=True,
+        allow_automated_raters=False,
+        minimum_raters_per_response=value,  # type: ignore[arg-type]
+    )
+
+    assert type(policy.minimum_raters_per_response) is int
+    assert policy.minimum_raters_per_response == int(value)