From 9409e5dd3158fb9b57a999645573b44e9db4a0f2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 23:40:00 +0900 Subject: [PATCH 1/7] test(cdm): specify native control trust boundary --- tests/test_cdm_native_control_boundary.py | 365 ++++++++++++++++++++++ 1 file changed, 365 insertions(+) create mode 100644 tests/test_cdm_native_control_boundary.py diff --git a/tests/test_cdm_native_control_boundary.py b/tests/test_cdm_native_control_boundary.py new file mode 100644 index 000000000..2f6af8766 --- /dev/null +++ b/tests/test_cdm_native_control_boundary.py @@ -0,0 +1,365 @@ +"""Trust-boundary regressions for cognitive-diagnosis native controls.""" + +from __future__ import annotations + +from collections.abc import Callable + +import numpy as np +import pytest + +import fast_mlsirm.fitstats as fitstats +from fast_mlsirm import ( + fit_cdm, + fit_gdina, + fit_ho_cdm, + fit_ho_gdina, + fit_seq_gdina, + fit_seq_gdina_qr, + gdina_wald_selection, + validate_q_matrix, +) + + +class _HostileInt(int): + """Integer subclass whose coercion/comparison hooks must remain unreachable.""" + + calls = 0 + + @classmethod + def reset(cls) -> None: + cls.calls = 0 + + def __int__(self): + type(self).calls += 1 + return int.__int__(self) + + def __index__(self): + type(self).calls += 1 + return int.__index__(self) + + def __lt__(self, other): + type(self).calls += 1 + return int.__lt__(self, other) + + def __le__(self, other): + type(self).calls += 1 + return int.__le__(self, other) + + +class _HostileFloat(float): + """Floating subclass whose conversion/comparison hooks must remain unreachable.""" + + calls = 0 + + @classmethod + def reset(cls) -> None: + cls.calls = 0 + + def __float__(self): + type(self).calls += 1 + return float.__float__(self) + + def __lt__(self, other): + type(self).calls += 1 + return float.__lt__(self, other) + + def __le__(self, other): + type(self).calls += 1 + return float.__le__(self, other) + + +class _HostileNumpyInt(np.int64): + """NumPy integer subclass whose hooks must remain unreachable.""" + + calls = 0 + + @classmethod + def reset(cls) -> None: + cls.calls = 0 + + def __int__(self): + type(self).calls += 1 + return super().__int__() + + def __index__(self): + type(self).calls += 1 + return super().__index__() + + +class _HostileNumpyFloat(np.float64): + """NumPy floating subclass whose hooks must remain unreachable.""" + + calls = 0 + + @classmethod + def reset(cls) -> None: + cls.calls = 0 + + def __float__(self): + type(self).calls += 1 + return super().__float__() + + +class _NumberProvider: + """Arbitrary numeric protocol provider that must never be normalized.""" + + calls = 0 + + def __init__(self, value): + self.value = value + + @classmethod + def reset(cls) -> None: + cls.calls = 0 + + def __int__(self): + type(self).calls += 1 + return int(self.value) + + def __index__(self): + type(self).calls += 1 + return int(self.value) + + def __float__(self): + type(self).calls += 1 + return float(self.value) + + def __lt__(self, other): + type(self).calls += 1 + return self.value < other + + def __le__(self, other): + type(self).calls += 1 + return self.value <= other + + +class _HostileStr(str): + """String subclass whose representation hook must remain unreachable.""" + + calls = 0 + + @classmethod + def reset(cls) -> None: + cls.calls = 0 + + def __str__(self): + type(self).calls += 1 + return str.__str__(self) + + +_INTEGER_HOSTILES = (_HostileInt, _HostileNumpyInt, _NumberProvider) +_FLOAT_HOSTILES = (_HostileFloat, _HostileNumpyFloat, _NumberProvider) +_NUMPY_INTEGER_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, +) +_NUMPY_FLOAT_TYPES = (np.float16, np.float32, np.float64, np.longdouble) + + +def _binary_inputs() -> tuple[np.ndarray, np.ndarray]: + """Return a small complete binary matrix and a matching Q-matrix.""" + responses = np.array( + [[1.0, 0.0, 1.0], [0.0, 1.0, 0.0], [1.0, 1.0, 0.0]], + dtype=np.float64, + ) + q_matrix = np.array([[1], [1], [1]], dtype=np.int64) + return responses, q_matrix + + +def _three_attribute_inputs() -> tuple[np.ndarray, np.ndarray]: + """Return valid inputs for higher-order CDM dispatch.""" + responses = np.array( + [[1.0, 0.0, 1.0], [0.0, 1.0, 0.0], [1.0, 1.0, 0.0]], + dtype=np.float64, + ) + q_matrix = np.eye(3, dtype=np.int64) + return responses, q_matrix + + +def _sequential_qr_inputs() -> tuple[np.ndarray, np.ndarray, np.ndarray]: + """Return valid one-step-per-item sequential restricted-Q inputs.""" + responses, q_matrix = _binary_inputs() + return responses, q_matrix, np.ones(responses.shape[1], dtype=np.int64) + + +def _entry_points() -> tuple[Callable[..., object], ...]: + """Return every public CDM adapter that consumes shared stopping controls.""" + responses, q_matrix = _binary_inputs() + ho_responses, ho_q = _three_attribute_inputs() + seq_responses, step_q, n_steps = _sequential_qr_inputs() + return ( + lambda **kw: fit_cdm(responses, q_matrix, **kw), + lambda **kw: fit_gdina(responses, q_matrix, **kw), + lambda **kw: validate_q_matrix(responses, q_matrix, **kw), + lambda **kw: gdina_wald_selection(responses, q_matrix, **kw), + lambda **kw: fit_ho_cdm(ho_responses, ho_q, **kw), + lambda **kw: fit_ho_gdina(ho_responses, ho_q, **kw), + lambda **kw: fit_seq_gdina(responses, q_matrix, **kw), + lambda **kw: fit_seq_gdina_qr(seq_responses, step_q, n_steps, **kw), + ) + + +def _unexpected_core_discovery(): + """Fail if rejected public input reaches native-core discovery.""" + raise AssertionError("compiled core must not be discovered for invalid CDM input") + + +@pytest.mark.parametrize("factory", _INTEGER_HOSTILES) +def test_max_iter_rejects_untrusted_integer_controls_without_callbacks( + monkeypatch, factory +): + """All CDM adapters reject hostile iteration controls before native discovery.""" + monkeypatch.setattr(fitstats, "_core_module", _unexpected_core_discovery) + for call in _entry_points(): + factory.reset() + with pytest.raises(ValueError, match="max_iter must be an integer between"): + call(max_iter=factory(5)) + assert factory.calls == 0 + + +@pytest.mark.parametrize("factory", _FLOAT_HOSTILES) +def test_tol_rejects_untrusted_numeric_controls_without_callbacks(monkeypatch, factory): + """All CDM adapters reject hostile tolerances before native discovery.""" + monkeypatch.setattr(fitstats, "_core_module", _unexpected_core_discovery) + for call in _entry_points(): + factory.reset() + with pytest.raises(ValueError, match="tol must be a finite number > 0"): + call(tol=factory(1e-6)) + assert factory.calls == 0 + + +@pytest.mark.parametrize( + ("call", "value"), + ( + (lambda y, q, value: fit_cdm(y, q, model=value), _HostileStr("dina")), + (lambda y, q, value: fit_ho_cdm(y, q, model=value), _HostileStr("dino")), + ), +) +def test_model_rejects_string_subclasses_without_callbacks(monkeypatch, call, value): + """DINA/DINO selectors accept exact built-in strings only.""" + monkeypatch.setattr(fitstats, "_core_module", _unexpected_core_discovery) + value.reset() + responses, q_matrix = _three_attribute_inputs() + + with pytest.raises(ValueError, match="model must be 'dina' or 'dino'"): + call(responses, q_matrix, value) + + assert value.calls == 0 + + +@pytest.mark.parametrize("model", ["other", "DINA", "dina ", ""]) +def test_model_rejects_unknown_exact_strings_before_native_discovery(monkeypatch, model): + """Python preserves the Rust selector vocabulary instead of coercing strings.""" + monkeypatch.setattr(fitstats, "_core_module", _unexpected_core_discovery) + responses, q_matrix = _binary_inputs() + + with pytest.raises(ValueError, match="model must be 'dina' or 'dino'"): + fit_cdm(responses, q_matrix, model=model) + + +@pytest.mark.parametrize("bad", [True, np.bool_(True)]) +def test_boolean_stopping_controls_fail_before_native_discovery(monkeypatch, bad): + """Boolean integer/numeric aliases remain outside the trusted control domain.""" + monkeypatch.setattr(fitstats, "_core_module", _unexpected_core_discovery) + responses, q_matrix = _binary_inputs() + + with pytest.raises(ValueError, match="max_iter must be an integer between"): + fit_gdina(responses, q_matrix, max_iter=bad) + with pytest.raises(ValueError, match="tol must be a finite number > 0"): + fit_gdina(responses, q_matrix, tol=bad) + + +@pytest.mark.parametrize("bad", [0, -1, 10**12]) +def test_iteration_bounds_fail_before_native_discovery(monkeypatch, bad): + """Iteration work remains positively bounded by the repository cap.""" + monkeypatch.setattr(fitstats, "_core_module", _unexpected_core_discovery) + responses, q_matrix = _binary_inputs() + + with pytest.raises(ValueError, match="max_iter must be an integer between"): + fit_gdina(responses, q_matrix, max_iter=bad) + + +@pytest.mark.parametrize("bad", [0.0, -1.0, np.nan, np.inf, -np.inf]) +def test_tolerance_domain_fails_before_native_discovery(monkeypatch, bad): + """Tolerance must be finite and strictly positive before native discovery.""" + monkeypatch.setattr(fitstats, "_core_module", _unexpected_core_discovery) + responses, q_matrix = _binary_inputs() + + with pytest.raises(ValueError, match="tol must be a finite number > 0"): + fit_gdina(responses, q_matrix, tol=bad) + + +def test_malformed_response_shape_fails_before_native_discovery(monkeypatch): + """Existing structural validation precedes native-loader access after hardening.""" + monkeypatch.setattr(fitstats, "_core_module", _unexpected_core_discovery) + q_matrix = np.ones((2, 1), dtype=np.int64) + + with pytest.raises(ValueError, match="responses must be a 2-D persons x items array"): + fit_gdina(np.zeros(2), q_matrix) + + +@pytest.mark.parametrize("numpy_type", _NUMPY_INTEGER_TYPES) +def test_exact_numpy_iteration_controls_reach_native_boundary(monkeypatch, numpy_type): + """Supported exact NumPy integer scalars retain compatibility.""" + calls = 0 + + def missing_core(): + nonlocal calls + calls += 1 + return None + + monkeypatch.setattr(fitstats, "_core_module", missing_core) + responses, q_matrix = _binary_inputs() + + with pytest.raises(RuntimeError, match="fit_gdina requires the compiled Rust core"): + fit_gdina(responses, q_matrix, max_iter=numpy_type(5)) + + assert calls == 1 + + +@pytest.mark.parametrize("numpy_type", (*_NUMPY_INTEGER_TYPES, *_NUMPY_FLOAT_TYPES)) +def test_exact_numpy_tolerance_controls_reach_native_boundary(monkeypatch, numpy_type): + """Supported exact NumPy real scalars retain tolerance compatibility.""" + calls = 0 + + def missing_core(): + nonlocal calls + calls += 1 + return None + + monkeypatch.setattr(fitstats, "_core_module", missing_core) + responses, q_matrix = _binary_inputs() + + with pytest.raises(RuntimeError, match="fit_gdina requires the compiled Rust core"): + fit_gdina(responses, q_matrix, tol=numpy_type(1e-4)) + + assert calls == 1 + + +@pytest.mark.parametrize("model", ["dina", "dino"]) +def test_exact_model_strings_reach_native_boundary(monkeypatch, model): + """Both existing DINA/DINO selector values remain accepted.""" + calls = 0 + + def missing_core(): + nonlocal calls + calls += 1 + return None + + monkeypatch.setattr(fitstats, "_core_module", missing_core) + responses, q_matrix = _three_attribute_inputs() + + with pytest.raises(RuntimeError, match="fit_ho_cdm requires the compiled Rust core"): + fit_ho_cdm(responses, q_matrix, model=model) + + assert calls == 1 From 66cb895f16efc10d3a1f6afadbe6934b665085df Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 23:43:20 +0900 Subject: [PATCH 2/7] fix(cdm): harden native dispatch controls --- python/fast_mlsirm/cdm.py | 195 ++++++++++++++++++++++---------------- 1 file changed, 111 insertions(+), 84 deletions(-) diff --git a/python/fast_mlsirm/cdm.py b/python/fast_mlsirm/cdm.py index aceb12735..72384b83c 100644 --- a/python/fast_mlsirm/cdm.py +++ b/python/fast_mlsirm/cdm.py @@ -12,6 +12,22 @@ _MAX_ATTRIBUTES = 15 +_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, +) +_NUMPY_FLOAT_SCALAR_TYPES = (np.float16, np.float32, np.float64, np.longdouble) +_CDM_MODELS = frozenset(("dina", "dino")) def _prepare_binary_responses(y: np.ndarray) -> tuple[np.ndarray, np.ndarray]: @@ -26,21 +42,37 @@ def _prepare_binary_responses(y: np.ndarray) -> tuple[np.ndarray, np.ndarray]: def _validate_stopping_controls(max_iter: int, tol: float) -> tuple[int, float]: - """Validate and return the EM iteration cap and convergence tolerance.""" - if ( - not isinstance(max_iter, (int, np.integer)) - or isinstance(max_iter, (bool, np.bool_)) - or not 1 <= int(max_iter) <= MAX_MAX_ITER - ): + """Validate exact scalar EM controls without caller-controlled coercion hooks.""" + max_iter_type = type(max_iter) + if max_iter_type is int: + iteration_cap = max_iter + elif any(max_iter_type is scalar_type for scalar_type in _NUMPY_INTEGER_SCALAR_TYPES): + iteration_cap = int(max_iter) + else: + raise ValueError(f"max_iter must be an integer between 1 and {MAX_MAX_ITER}") + if not 1 <= iteration_cap <= MAX_MAX_ITER: raise ValueError(f"max_iter must be an integer between 1 and {MAX_MAX_ITER}") - if not isinstance(tol, (int, float, np.integer, np.floating)) or isinstance( - tol, (bool, np.bool_) + + tol_type = type(tol) + if tol_type is int or tol_type is float: + tolerance = float(tol) + elif any( + tol_type is scalar_type + for scalar_type in (*_NUMPY_INTEGER_SCALAR_TYPES, *_NUMPY_FLOAT_SCALAR_TYPES) ): + tolerance = float(tol) + else: raise ValueError("tol must be a finite number > 0") - tolerance = float(tol) if not np.isfinite(tolerance) or tolerance <= 0: raise ValueError("tol must be a finite number > 0") - return int(max_iter), tolerance + return iteration_cap, tolerance + + +def _validate_model_selector(model: str) -> str: + """Accept only the exact built-in DINA/DINO selector vocabulary.""" + if type(model) is not str or model not in _CDM_MODELS: + raise ValueError("model must be 'dina' or 'dino'") + return model def _validate_q_matrix_input( @@ -138,20 +170,20 @@ def fit_cdm( disorders using cognitive diagnosis models. *Psychological Methods, 11*(3), 287–305. https://doi.org/10.1037/1082-989X.11.3.287 """ - from .fitstats import _core_module - - core = _core_module() - if core is None or not hasattr(core, "fit_cdm"): - raise RuntimeError("fit_cdm requires the compiled Rust core") - y = np.asarray(responses, dtype=np.float64) if y.ndim != 2: raise ValueError("responses must be a 2-D persons x items array") n_persons, n_items = y.shape q, n_attributes = _validate_q_matrix_input(q_matrix, "q_matrix", n_items) - + model = _validate_model_selector(model) max_iter, tol = _validate_stopping_controls(max_iter, tol) yy, observed = _prepare_binary_responses(y) + + from .fitstats import _core_module + + core = _core_module() + if core is None or not hasattr(core, "fit_cdm"): + raise RuntimeError("fit_cdm requires the compiled Rust core") res = core.fit_cdm( yy, observed, @@ -159,9 +191,9 @@ def fit_cdm( int(n_persons), int(n_items), int(n_attributes), - str(model), - int(max_iter), - float(tol), + model, + max_iter, + tol, ) return CdmFit( model=str(res["model"]), @@ -245,20 +277,19 @@ def fit_gdina( diagnosis modeling. *Journal of Statistical Software, 93*(14), 1-26. https://doi.org/10.18637/jss.v093.i14 """ - from .fitstats import _core_module - - core = _core_module() - if core is None or not hasattr(core, "fit_gdina"): - raise RuntimeError("fit_gdina requires the compiled Rust core") - y = np.asarray(responses, dtype=np.float64) if y.ndim != 2: raise ValueError("responses must be a 2-D persons x items array") n_persons, n_items = y.shape q, n_attributes = _validate_q_matrix_input(q_matrix, "q_matrix", n_items) - max_iter, tol = _validate_stopping_controls(max_iter, tol) yy, observed = _prepare_binary_responses(y) + + from .fitstats import _core_module + + core = _core_module() + if core is None or not hasattr(core, "fit_gdina"): + raise RuntimeError("fit_gdina requires the compiled Rust core") res = core.fit_gdina( yy, observed, @@ -266,8 +297,8 @@ def fit_gdina( int(n_persons), int(n_items), int(n_attributes), - int(max_iter), - float(tol), + max_iter, + tol, ) return GdinaFit( item_off=np.asarray(res["item_off"], dtype=np.int64), @@ -347,20 +378,19 @@ def validate_q_matrix( Measurement, 45*(4), 343-362. https://doi.org/10.1111/j.1745-3984.2008.00069.x """ - from .fitstats import _core_module - - core = _core_module() - if core is None or not hasattr(core, "validate_q_matrix"): - raise RuntimeError("validate_q_matrix requires the compiled Rust core") - y = np.asarray(responses, dtype=np.float64) if y.ndim != 2: raise ValueError("responses must be a 2-D persons x items array") n_persons, n_items = y.shape q, n_attributes = _validate_q_matrix_input(provisional_q, "provisional_q", n_items) - max_iter, tol = _validate_stopping_controls(max_iter, tol) yy, observed = _prepare_binary_responses(y) + + from .fitstats import _core_module + + core = _core_module() + if core is None or not hasattr(core, "validate_q_matrix"): + raise RuntimeError("validate_q_matrix requires the compiled Rust core") res = core.validate_q_matrix( yy, observed, @@ -369,8 +399,8 @@ def validate_q_matrix( int(n_items), int(n_attributes), float(epsilon), - int(max_iter), - float(tol), + max_iter, + tol, ) return QMatrixValidation( suggested_q=np.asarray(res["suggested_q"], dtype=np.int64).reshape(n_items, n_attributes), @@ -460,20 +490,19 @@ def gdina_wald_selection( selection, and attribute classification. *Applied Psychological Measurement, 40*(3), 200–217. https://doi.org/10.1177/0146621615621717 """ - from .fitstats import _core_module - - core = _core_module() - if core is None or not hasattr(core, "gdina_wald_selection"): - raise RuntimeError("gdina_wald_selection requires the compiled Rust core") - y = np.asarray(responses, dtype=np.float64) if y.ndim != 2: raise ValueError("responses must be a 2-D persons x items array") n_persons, n_items = y.shape q, n_attributes = _validate_q_matrix_input(q_matrix, "q_matrix", n_items) - max_iter, tol = _validate_stopping_controls(max_iter, tol) yy, observed = _prepare_binary_responses(y) + + from .fitstats import _core_module + + core = _core_module() + if core is None or not hasattr(core, "gdina_wald_selection"): + raise RuntimeError("gdina_wald_selection requires the compiled Rust core") res = core.gdina_wald_selection( yy, observed, @@ -482,8 +511,8 @@ def gdina_wald_selection( int(n_items), int(n_attributes), float(alpha), - int(max_iter), - float(tol), + max_iter, + tol, ) models = list(res["models"]) n_models = len(models) @@ -567,20 +596,20 @@ class distribution, so the higher-order parameters are identified only for cognitive diagnosis. *Psychometrika, 69*(3), 333-353. https://doi.org/10.1007/BF02295640 """ - from .fitstats import _core_module - - core = _core_module() - if core is None or not hasattr(core, "fit_ho_cdm"): - raise RuntimeError("fit_ho_cdm requires the compiled Rust core") - y = np.asarray(responses, dtype=np.float64) if y.ndim != 2: raise ValueError("responses must be a 2-D persons x items array") n_persons, n_items = y.shape q, n_attributes = _validate_q_matrix_input(q_matrix, "q_matrix", n_items) - + model = _validate_model_selector(model) max_iter, tol = _validate_stopping_controls(max_iter, tol) yy, observed = _prepare_binary_responses(y) + + from .fitstats import _core_module + + core = _core_module() + if core is None or not hasattr(core, "fit_ho_cdm"): + raise RuntimeError("fit_ho_cdm requires the compiled Rust core") res = core.fit_ho_cdm( yy, observed, @@ -588,9 +617,9 @@ class distribution, so the higher-order parameters are identified only for int(n_persons), int(n_items), int(n_attributes), - str(model), - int(max_iter), - float(tol), + model, + max_iter, + tol, ) return HoCdmFit( model=str(res["model"]), @@ -685,20 +714,19 @@ def fit_ho_gdina( de la Torre, J. (2011). The generalized DINA model framework. *Psychometrika, 76*(2), 179-199. https://doi.org/10.1007/s11336-011-9207-7 """ - from .fitstats import _core_module - - core = _core_module() - if core is None or not hasattr(core, "fit_ho_gdina"): - raise RuntimeError("fit_ho_gdina requires the compiled Rust core") - y = np.asarray(responses, dtype=np.float64) if y.ndim != 2: raise ValueError("responses must be a 2-D persons x items array") n_persons, n_items = y.shape q, n_attributes = _validate_q_matrix_input(q_matrix, "q_matrix", n_items) - max_iter, tol = _validate_stopping_controls(max_iter, tol) yy, observed = _prepare_binary_responses(y) + + from .fitstats import _core_module + + core = _core_module() + if core is None or not hasattr(core, "fit_ho_gdina"): + raise RuntimeError("fit_ho_gdina requires the compiled Rust core") res = core.fit_ho_gdina( yy, observed, @@ -706,8 +734,8 @@ def fit_ho_gdina( int(n_persons), int(n_items), int(n_attributes), - int(max_iter), - float(tol), + max_iter, + tol, ) return HoGdinaFit( item_off=np.asarray(res["item_off"], dtype=np.int64), @@ -825,12 +853,6 @@ def fit_seq_gdina( de la Torre, J. (2011). The generalized DINA model framework. *Psychometrika, 76*(2), 179-199. https://doi.org/10.1007/s11336-011-9207-7 """ - from .fitstats import _core_module - - core = _core_module() - if core is None or not hasattr(core, "fit_seq_gdina"): - raise RuntimeError("fit_seq_gdina requires the compiled Rust core") - y = np.asarray(responses, dtype=np.float64) if y.ndim != 2: raise ValueError("responses must be a 2-D persons x items array") @@ -838,10 +860,15 @@ def fit_seq_gdina( q, n_attributes = _validate_q_matrix_input(q_matrix, "q_matrix", n_items) if np.isinf(y).any(): raise ValueError("responses must be finite ordered categories or NaN (missing)") - max_iter, tol = _validate_stopping_controls(max_iter, tol) observed = ~np.isnan(y) yy = np.where(observed, y, 0.0).reshape(-1) + + from .fitstats import _core_module + + core = _core_module() + if core is None or not hasattr(core, "fit_seq_gdina"): + raise RuntimeError("fit_seq_gdina requires the compiled Rust core") res = core.fit_seq_gdina( yy, observed.reshape(-1), @@ -849,8 +876,8 @@ def fit_seq_gdina( int(n_persons), int(n_items), int(n_attributes), - int(max_iter), - float(tol), + max_iter, + tol, ) return SeqGdinaFit( s_off=np.asarray(res["s_off"], dtype=np.int64), @@ -872,6 +899,7 @@ def fit_seq_gdina( n_parameters=int(res["n_parameters"]), ) + @dataclass class SeqGdinaQrFit: """Fitted per-step-Q sequential G-DINA (Ma & de la Torre, 2016, restricted-Q). @@ -943,12 +971,6 @@ def fit_seq_gdina_qr( de la Torre, J. (2011). The generalized DINA model framework. *Psychometrika, 76*(2), 179-199. https://doi.org/10.1007/s11336-011-9207-7 """ - from .fitstats import _core_module - - core = _core_module() - if core is None or not hasattr(core, "fit_seq_gdina_qr"): - raise RuntimeError("fit_seq_gdina_qr requires the compiled Rust core") - y = np.asarray(responses, dtype=np.float64) if y.ndim != 2: raise ValueError("responses must be a 2-D persons x items array") @@ -972,10 +994,15 @@ def fit_seq_gdina_qr( sq, n_attributes = _validate_q_matrix_input(step_q, "step_q", n_step_rows) if np.isinf(y).any(): raise ValueError("responses must be finite ordered categories or NaN (missing)") - max_iter, tol = _validate_stopping_controls(max_iter, tol) observed = ~np.isnan(y) yy = np.where(observed, y, 0.0).reshape(-1) + + from .fitstats import _core_module + + core = _core_module() + if core is None or not hasattr(core, "fit_seq_gdina_qr"): + raise RuntimeError("fit_seq_gdina_qr requires the compiled Rust core") res = core.fit_seq_gdina_qr( yy, observed.reshape(-1), @@ -984,8 +1011,8 @@ def fit_seq_gdina_qr( int(n_persons), int(n_items), int(n_attributes), - int(max_iter), - float(tol), + max_iter, + tol, ) return SeqGdinaQrFit( step_off=np.asarray(res["step_off"], dtype=np.int64), From 00b84af2ac59e7a1321ede6b79fac5bf7be37eff Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 23:45:15 +0900 Subject: [PATCH 3/7] docs(changelog): record CDM native control hardening --- docs/changelog.d/902-cdm-native-control-boundary.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 docs/changelog.d/902-cdm-native-control-boundary.md diff --git a/docs/changelog.d/902-cdm-native-control-boundary.md b/docs/changelog.d/902-cdm-native-control-boundary.md new file mode 100644 index 000000000..b0ad7dcd0 --- /dev/null +++ b/docs/changelog.d/902-cdm-native-control-boundary.md @@ -0,0 +1,5 @@ +# Cognitive-diagnosis native control boundary + +## Security + +- Validate CDM-family `max_iter`, `tol`, and DINA/DINO model selectors before compiled-core discovery. Only exact built-in values and explicitly supported concrete NumPy scalar types are normalized; booleans, subclasses, protocol providers, non-finite/out-of-range values, and unknown model selectors fail locally without executing caller conversion callbacks. Rust-owned psychometric arithmetic and result schemas are unchanged. From 3e6e8d831ef591cfcb6db2c714365f88bf00d508 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 23:45:42 +0900 Subject: [PATCH 4/7] docs(doctoring): specify CDM control trust boundary --- docs/doctoring/cdm_native_control_boundary.md | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 docs/doctoring/cdm_native_control_boundary.md diff --git a/docs/doctoring/cdm_native_control_boundary.md b/docs/doctoring/cdm_native_control_boundary.md new file mode 100644 index 000000000..89320ba89 --- /dev/null +++ b/docs/doctoring/cdm_native_control_boundary.md @@ -0,0 +1,53 @@ +# CDM native-control validation trust boundary + +## Status + +Implemented on the issue #902 source branch until protected-main integration. This is a bounded public-input validation correction for the cognitive-diagnosis Python adapters; it does not change DINA, DINO, G-DINA, higher-order, sequential, PVAF, or Wald psychometric arithmetic in the Rust core. + +## Problem + +The CDM-family adapters previously discovered the compiled native core before validating caller-facing controls. Their shared stopping-control validator accepted broad Python/NumPy numeric classes and then normalized them through `int(...)` / `float(...)`; the DINA/DINO adapters similarly normalized `model` with `str(...)` at native dispatch. Python special-method dispatch means those conversions are executable behavior for caller-defined subclasses and protocol providers, rather than passive type checks. A rejected configuration could therefore execute caller-owned conversion or representation callbacks, or touch the native-loader boundary, before the package had established that the control value belonged to the documented input domain. + +This is treated as a specified-type validation boundary consistent with CWE-1287. OWASP ASVS 5.0.0 V2.2.1 likewise requires validation against expected structure and logical limits, while V2.2.2 places such enforcement in a trusted service layer. NIST SP 800-218 SSDF Version 1.1 is used as secure-development-process context; later draft revisions are not represented here as final standards. These references are engineering guidance and taxonomy, not certification or severity claims. + +## Decision + +For the affected public CDM adapters: + +- validate the response/Q-matrix structure and public controls before `_core_module()` discovery; +- admit `max_iter` only when its runtime type is exactly built-in `int` or one of the explicitly supported concrete NumPy integer scalar types, then enforce the repository iteration cap; +- admit `tol` only when its runtime type is exactly built-in `int`/`float` or one of the explicitly supported concrete NumPy integer/floating scalar types, then require a finite strictly positive value; +- reject booleans, Python/NumPy subclasses, and arbitrary conversion-protocol providers before calling their conversion or comparison hooks; +- admit DINA/DINO selectors only as exact built-in strings in the existing vocabulary `{"dina", "dino"}` and pass the trusted string directly to PyO3; and +- keep all result-affecting likelihood, EM, model-selection, classification, convergence, uncertainty, PVAF, Wald, higher-order, and sequential arithmetic in Rust unchanged. + +## Evidence contract + +The fail-first commit `9409e5dd3158fb9b57a999645573b44e9db4a0f2` exercises the public adapters with Python integer/float subclasses, NumPy scalar subclasses, arbitrary numeric protocol providers, and a string subclass. Each hostile value increments a counter from any conversion/comparison hook, while native-core discovery is replaced by a sentinel that must remain unreachable for invalid input. + +GREEN requires all of the following: + +- hostile `max_iter`, `tol`, and `model` controls raise stable package-owned `ValueError`s with callback counts remaining zero; +- malformed public inputs are rejected before native discovery; +- booleans, non-finite tolerances, out-of-range iteration caps, and unknown model strings remain rejected; +- genuine supported concrete NumPy scalar controls and exact `"dina"`/`"dino"` strings still reach the established native-dispatch boundary; +- valid inputs retain the existing compiled-core-required `RuntimeError` when the extension is unavailable; and +- package Python/Rust/PyO3, coverage, security, package, fuzz, provenance, and required-review gates remain authoritative integration evidence on the exact PR head. + +The implementation commit `66cb895f16efc10d3a1f6afadbe6934b665085df` changes validation and marshalling order only. Exact-head CI remains the integration authority; predecessor-head results do not transfer after a source update. + +## Scope limitations + +This correction does not redesign CDM formulas, estimators, convergence criteria, Q-matrix identifiability theory, missing-data assumptions, or structural model selection. It does not introduce a Python numerical fallback or transfer arithmetic ownership away from Rust. Other caller-control boundaries outside the affected CDM adapters require their own bounded regressions rather than being inferred safe from this change. + +## References + +MITRE. (2026). *CWE-1287: Improper validation of specified type of input (CWE version 4.20)*. Common Weakness Enumeration. https://cwe.mitre.org/data/definitions/1287.html + +National Institute of Standards and Technology. (2022). *Secure Software Development Framework (SSDF) Version 1.1: Recommendations for mitigating the risk of software vulnerabilities from software development practices (NIST SP 800-218)*. https://doi.org/10.6028/NIST.SP.800-218 + +OWASP Foundation. (2025). *OWASP Application Security Verification Standard 5.0.0*. https://owasp.org/www-project-application-security-verification-standard/ + +Python Software Foundation. (2026). *Data model*. Python 3.14 documentation. https://docs.python.org/3.14/reference/datamodel.html + +NumPy Developers. (2026). *Scalars*. NumPy reference. https://numpy.org/doc/stable/reference/arrays.scalars.html From 4a07ab8a72ba18e36f65e78f1cfea2b4f91ed390 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 01:59:16 +0900 Subject: [PATCH 5/7] test(cdm): use representable NumPy tolerance values --- tests/test_cdm_native_control_boundary.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/tests/test_cdm_native_control_boundary.py b/tests/test_cdm_native_control_boundary.py index 2f6af8766..9bf75f101 100644 --- a/tests/test_cdm_native_control_boundary.py +++ b/tests/test_cdm_native_control_boundary.py @@ -327,8 +327,14 @@ def missing_core(): assert calls == 1 -@pytest.mark.parametrize("numpy_type", (*_NUMPY_INTEGER_TYPES, *_NUMPY_FLOAT_TYPES)) -def test_exact_numpy_tolerance_controls_reach_native_boundary(monkeypatch, numpy_type): +@pytest.mark.parametrize( + ("numpy_type", "value"), + tuple((numpy_type, 1) for numpy_type in _NUMPY_INTEGER_TYPES) + + tuple((numpy_type, 1e-4) for numpy_type in _NUMPY_FLOAT_TYPES), +) +def test_exact_numpy_tolerance_controls_reach_native_boundary( + monkeypatch, numpy_type, value +): """Supported exact NumPy real scalars retain tolerance compatibility.""" calls = 0 @@ -341,7 +347,7 @@ def missing_core(): responses, q_matrix = _binary_inputs() with pytest.raises(RuntimeError, match="fit_gdina requires the compiled Rust core"): - fit_gdina(responses, q_matrix, tol=numpy_type(1e-4)) + fit_gdina(responses, q_matrix, tol=numpy_type(value)) assert calls == 1 From 5dd0a1c06c43fea1c2fcb319f67c7ec4cf7b319f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 02:47:41 +0900 Subject: [PATCH 6/7] test(cdm): cover oversized integer tolerance --- tests/test_cdm_native_control_boundary.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_cdm_native_control_boundary.py b/tests/test_cdm_native_control_boundary.py index 9bf75f101..501516014 100644 --- a/tests/test_cdm_native_control_boundary.py +++ b/tests/test_cdm_native_control_boundary.py @@ -289,7 +289,7 @@ def test_iteration_bounds_fail_before_native_discovery(monkeypatch, bad): fit_gdina(responses, q_matrix, max_iter=bad) -@pytest.mark.parametrize("bad", [0.0, -1.0, np.nan, np.inf, -np.inf]) +@pytest.mark.parametrize("bad", [0.0, -1.0, np.nan, np.inf, -np.inf, 10**1000]) def test_tolerance_domain_fails_before_native_discovery(monkeypatch, bad): """Tolerance must be finite and strictly positive before native discovery.""" monkeypatch.setattr(fitstats, "_core_module", _unexpected_core_discovery) From 7e11c1c6d495567b0612ff1bbc022d280fc17b8f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 16 Aug 2026 02:52:06 +0900 Subject: [PATCH 7/7] fix(cdm): normalize oversized tolerances safely --- python/fast_mlsirm/cdm.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/python/fast_mlsirm/cdm.py b/python/fast_mlsirm/cdm.py index 72384b83c..e8cd7bd62 100644 --- a/python/fast_mlsirm/cdm.py +++ b/python/fast_mlsirm/cdm.py @@ -54,15 +54,19 @@ def _validate_stopping_controls(max_iter: int, tol: float) -> tuple[int, float]: raise ValueError(f"max_iter must be an integer between 1 and {MAX_MAX_ITER}") tol_type = type(tol) - if tol_type is int or tol_type is float: - tolerance = float(tol) - elif any( - tol_type is scalar_type - for scalar_type in (*_NUMPY_INTEGER_SCALAR_TYPES, *_NUMPY_FLOAT_SCALAR_TYPES) + if not ( + tol_type is int + or tol_type is float + or any( + tol_type is scalar_type + for scalar_type in (*_NUMPY_INTEGER_SCALAR_TYPES, *_NUMPY_FLOAT_SCALAR_TYPES) + ) ): - tolerance = float(tol) - else: raise ValueError("tol must be a finite number > 0") + try: + tolerance = float(tol) + except OverflowError as exc: + raise ValueError("tol must be a finite number > 0") from exc if not np.isfinite(tolerance) or tolerance <= 0: raise ValueError("tol must be a finite number > 0") return iteration_cap, tolerance