diff --git a/docs/changelog.d/991-rubric-text-trust-boundary.md b/docs/changelog.d/991-rubric-text-trust-boundary.md new file mode 100644 index 000000000..de657289f --- /dev/null +++ b/docs/changelog.d/991-rubric-text-trust-boundary.md @@ -0,0 +1,6 @@ +# Harden rubric text schema callback safety + +## Fixed + +- Harden rubric, item-blueprint, and shared scoring text/identifier schema admission so caller-defined `str` subclasses fail closed before any overridable text callback executes, while preserving normalization for exact built-in strings. +- Apply the same exact-built-in-string admission to item-bank evidence enums so lifecycle evidence cannot dispatch caller-defined equality or hash callbacks during enum lookup. diff --git a/python/fast_mlsirm/rubric/audit.py b/python/fast_mlsirm/rubric/audit.py index ef4fbfeca..9c1be848a 100644 --- a/python/fast_mlsirm/rubric/audit.py +++ b/python/fast_mlsirm/rubric/audit.py @@ -98,12 +98,14 @@ def _fingerprint(value: Any, name: str) -> str: def _enum_value(value: Any, enum_type: type[Enum], name: str) -> Enum: """Normalize one exact enum member or its serialized string value.""" - if isinstance(value, enum_type): + choices = [member.value for member in enum_type] + if type(value) is enum_type: return value + if type(value) is not str: + raise ValueError(f"{name} must be one of {choices}") try: return enum_type(value) except (TypeError, ValueError) as exc: - choices = [member.value for member in enum_type] raise ValueError(f"{name} must be one of {choices}") from exc diff --git a/python/fast_mlsirm/rubric/item_bank.py b/python/fast_mlsirm/rubric/item_bank.py index 1744618d6..ec59fedc1 100644 --- a/python/fast_mlsirm/rubric/item_bank.py +++ b/python/fast_mlsirm/rubric/item_bank.py @@ -117,12 +117,14 @@ def _fingerprint(value: Any, name: str) -> str: def _enum_value(value: Any, enum_type: type[Enum], name: str) -> Enum: """Normalize an exact enum member or its serialized value.""" - if isinstance(value, enum_type): + choices = [member.value for member in enum_type] + if type(value) is enum_type: return value + if type(value) is not str: + raise ValueError(f"{name} must be one of {choices}") try: return enum_type(value) except (TypeError, ValueError) as exc: - choices = [member.value for member in enum_type] raise ValueError(f"{name} must be one of {choices}") from exc @@ -174,7 +176,7 @@ def _normalize_evidence_references( minimum=0, maximum=_MAX_EVIDENCE_REFERENCES, ) - except ValueError as exc: + except ValueError: if error_type is ItemBankLifecycleError: raise ItemBankLifecycleError( "invalid_evidence_references", @@ -604,7 +606,7 @@ def transition_item_bank_record( current = _verify_current_record(current_record) try: target = _enum_value(target_state, ItemBankLifecycleState, "target_state") - except ValueError as exc: + except ValueError: raise ItemBankLifecycleError( "invalid_target_state", "$.target_state", @@ -664,7 +666,7 @@ def transition_item_bank_record( ) ) ) - except ValueError as exc: + except ValueError: raise ItemBankLifecycleError( "invalid_approved_use", "$.approved_use_ids", @@ -693,7 +695,7 @@ def transition_item_bank_record( transition_reason_id, "transition_reason_id", ) - except ValueError as exc: + except ValueError: raise ItemBankLifecycleError( "invalid_transition_reason", "$.transition_reason_id", diff --git a/python/fast_mlsirm/rubric/models.py b/python/fast_mlsirm/rubric/models.py index cf818618c..f17a49737 100644 --- a/python/fast_mlsirm/rubric/models.py +++ b/python/fast_mlsirm/rubric/models.py @@ -64,7 +64,7 @@ class EvidenceMode(str, Enum): def _text(value: Any, name: str, *, maximum: int = MAX_TEXT_LENGTH) -> str: """Normalize bounded non-empty text or raise a field-specific error.""" - if not isinstance(value, str): + if type(value) is not str: raise ValueError(f"{name} must be a string") normalized = value.strip() if not normalized: @@ -152,13 +152,15 @@ def _identifier_tuple( def _enum_value(value: Any, enum_type: type[EnumValue], name: str) -> EnumValue: - """Normalize an enum instance or its exact string value.""" + """Normalize an enum instance or its exact built-in string value.""" if isinstance(value, enum_type): return value + choices = [member.value for member in enum_type] + if type(value) is not str: + raise ValueError(f"{name} must be one of {choices}") try: return enum_type(value) except (TypeError, ValueError) as exc: - choices = [member.value for member in enum_type] raise ValueError(f"{name} must be one of {choices}") from exc @@ -558,4 +560,4 @@ def to_dict(self) -> dict[str, Any]: **self._fingerprint_payload(), "blueprint_id": self.blueprint_id, "blueprint_fingerprint": self.blueprint_fingerprint, - } + } \ No newline at end of file diff --git a/python/fast_mlsirm/scoring/_contract_safety.py b/python/fast_mlsirm/scoring/_contract_safety.py index 6fff3e104..c86baaa0d 100644 --- a/python/fast_mlsirm/scoring/_contract_safety.py +++ b/python/fast_mlsirm/scoring/_contract_safety.py @@ -142,6 +142,20 @@ def _has_exact_type(value: Any, trusted_types: tuple[type, ...]) -> bool: return any(value_type is trusted_type for trusted_type in trusted_types) +def _normalize_metadata_scalar(value: Any) -> tuple[bool, Any]: + """Normalize JSON scalar subclasses through inert base-type descriptors.""" + value_type = type(value) + if value is None or value_type in (bool, int, float, str): + return True, value + if isinstance(value, str): + return True, str.__str__(value) + if isinstance(value, int) and not isinstance(value, bool): + return True, int.__int__(value) + if isinstance(value, float): + return True, float.__float__(value) + return False, value + + def bounded_positive_integer( value: Any, name: str, @@ -315,6 +329,10 @@ def _preflight_metadata( f"metadata exceeds the maximum node count of {base.MAX_METADATA_NODES}", ) + is_scalar, normalized_scalar = _normalize_metadata_scalar(value) + if is_scalar: + return normalized_scalar + if isinstance(value, Mapping): marker = id(value) if marker in active: @@ -354,7 +372,10 @@ def _preflight_metadata( "metadata mapping entries must contain one key and value", ) from None key_path = f"{path}.keys[{index}]" - key = base._metadata_key(raw_key, key_path) + key_is_scalar, key_input = _normalize_metadata_scalar(raw_key) + if not key_is_scalar or type(key_input) is not str: + key_input = raw_key + key = base._metadata_key(key_input, key_path) if key.casefold() in _SENSITIVE_METADATA_FIELDS: raise base.assessment_error( "sensitive_metadata_field", diff --git a/tests/test_rubric_audit_enum_callback_safety.py b/tests/test_rubric_audit_enum_callback_safety.py new file mode 100644 index 000000000..c60bc2229 --- /dev/null +++ b/tests/test_rubric_audit_enum_callback_safety.py @@ -0,0 +1,109 @@ +"""Callback-safety regressions for rubric audit enum-valued controls.""" + +from __future__ import annotations + +import pytest + +from fast_mlsirm.rubric import ( + AuditSeverity, + CandidateAuditFinding, + CandidateAuditReport, + CandidateLifecycleState, +) +from fast_mlsirm.rubric.audit import PilotCandidateRecord as _CorePilotCandidateRecord + + +class _HostileString(str): + """String subclass that records any callback dispatch during enum lookup.""" + + callbacks = 0 + + def __hash__(self) -> int: + type(self).callbacks += 1 + raise AssertionError("caller __hash__ must not execute") + + def __eq__(self, other: object) -> bool: + type(self).callbacks += 1 + raise AssertionError("caller __eq__ must not execute") + + +def _hostile(value: str) -> _HostileString: + """Return one fresh hostile serialized enum value.""" + _HostileString.callbacks = 0 + return _HostileString(value) + + +def _core_pilot_record(lifecycle_state: object) -> _CorePilotCandidateRecord: + """Build the internal pilot record with one caller-controlled state value.""" + return _CorePilotCandidateRecord( + pilot_study_id="pilot_study_alpha", + query_testlet_id="query_testlet_alpha", + generator_family_id="generator_family_alpha", + judge_policy_id="judge_policy_alpha", + occasion_id="occasion_window_alpha", + item_id="generated_item_alpha", + candidate_fingerprint="a" * 64, + audit_report_fingerprint="b" * 64, + audit_policy_id="generated_item_audit", + audit_policy_version="1.0.0", + blueprint_id="blueprint_alpha", + rubric_id="rubric_alpha", + rubric_version="1.0.0", + lifecycle_state=lifecycle_state, + ) + + +@pytest.mark.parametrize( + ("factory", "serialized"), + [ + ( + lambda value: CandidateAuditFinding( + finding_code="audit_finding_alpha", + severity=value, + path="$.stem", + message="review required", + ), + "blocking", + ), + ( + lambda value: CandidateAuditReport( + audit_policy_id="generated_item_audit", + audit_policy_version="1.0.0", + candidate_fingerprint="a" * 64, + findings=(), + lifecycle_state=value, + ), + "audited", + ), + (_core_pilot_record, "pilot"), + ], +) +def test_audit_enum_strings_reject_subclasses_without_callback_dispatch(factory, serialized): + """Caller string subclasses fail closed before Enum lookup can call them.""" + value = _hostile(serialized) + with pytest.raises(ValueError): + factory(value) + assert _HostileString.callbacks == 0 + + +def test_audit_enum_strings_preserve_exact_serialized_values_and_members(): + """Exact strings and exact enum members retain their established semantics.""" + finding = CandidateAuditFinding( + finding_code="audit_finding_alpha", + severity="blocking", + path="$.stem", + message="review required", + ) + assert finding.severity is AuditSeverity.BLOCKING + + report = CandidateAuditReport( + audit_policy_id="generated_item_audit", + audit_policy_version="1.0.0", + candidate_fingerprint="a" * 64, + findings=(), + lifecycle_state="audited", + ) + assert report.lifecycle_state is CandidateLifecycleState.AUDITED + + pilot = _core_pilot_record(CandidateLifecycleState.PILOT) + assert pilot.lifecycle_state is CandidateLifecycleState.PILOT diff --git a/tests/test_rubric_text_callback_safety.py b/tests/test_rubric_text_callback_safety.py new file mode 100644 index 000000000..d124f2961 --- /dev/null +++ b/tests/test_rubric_text_callback_safety.py @@ -0,0 +1,148 @@ +"""Regression tests for callback-free rubric text admission.""" + +from __future__ import annotations + +import pytest + +from fast_mlsirm.rubric import ( + ItemBankEvidenceKind, + ItemBankEvidenceReference, + ResponseFormat, + RubricLevel, + RubricSpecification, +) + + +def _hostile_text(value: str) -> tuple[str, type[str]]: + """Return a string subclass whose text callbacks must never execute.""" + + class HostileText(str): + """Record forbidden normalization callbacks without executing them.""" + + calls = 0 + + def strip(self, *args: object, **kwargs: object) -> str: + """Record an unexpected normalization callback.""" + type(self).calls += 1 + raise AssertionError("caller strip callback executed") + + def __hash__(self) -> int: + type(self).calls += 1 + raise AssertionError("caller hash callback executed") + + return HostileText(value), HostileText + + +def _levels() -> tuple[RubricLevel, RubricLevel]: + """Return a minimal valid ordinal rubric scale.""" + return ( + RubricLevel(0, "not_met", "Requirement is not met.", ("missing evidence",)), + RubricLevel(1, "fully_met", "Requirement is fully met.", ("complete evidence",)), + ) + + +def test_rubric_level_rejects_string_subclass_without_callback() -> None: + """Scalar level text rejects caller subclasses before invoking text methods.""" + hostile, hostile_type = _hostile_text("not_met") + + with pytest.raises(ValueError, match="label must be a string"): + RubricLevel(0, hostile, "Requirement is not met.", ("missing evidence",)) + + assert hostile_type.calls == 0 + + +def test_rubric_specification_rejects_identifier_subclass_without_callback() -> None: + """Identifier text rejects caller subclasses before normalization callbacks.""" + hostile, hostile_type = _hostile_text("evidence_rubric") + + with pytest.raises(ValueError, match="rubric_id must be a string"): + RubricSpecification( + rubric_id=hostile, + construct_id="evidence_quality", + construct_definition="Quality of evidence support.", + response_format=ResponseFormat.ORDINAL_RATING, + levels=_levels(), + task_families=("evidence_review",), + evidence_requirements=("Cite supporting evidence.",), + ) + + assert hostile_type.calls == 0 + + +def test_rubric_specification_rejects_enum_text_subclass_without_callback() -> None: + """Enum text rejects caller subclasses before value lookup callbacks.""" + hostile, hostile_type = _hostile_text("ordinal_rating") + + with pytest.raises(ValueError, match="response_format must be one of"): + RubricSpecification( + rubric_id="evidence_rubric", + construct_id="evidence_quality", + construct_definition="Quality of evidence support.", + response_format=hostile, + levels=_levels(), + task_families=("evidence_review",), + evidence_requirements=("Cite supporting evidence.",), + ) + + assert hostile_type.calls == 0 + + +def test_rubric_specification_rejects_collection_text_subclass_without_callback() -> None: + """Nested rubric text rejects caller subclasses before tuple normalization callbacks.""" + hostile, hostile_type = _hostile_text("Cite supporting evidence.") + + with pytest.raises(ValueError, match=r"evidence_requirements\[0\] must be a string"): + RubricSpecification( + rubric_id="evidence_rubric", + construct_id="evidence_quality", + construct_definition="Quality of evidence support.", + response_format=ResponseFormat.ORDINAL_RATING, + levels=_levels(), + task_families=("evidence_review",), + evidence_requirements=(hostile,), + ) + + assert hostile_type.calls == 0 + + +def test_builtin_rubric_text_still_normalizes() -> None: + """Exact built-in strings retain whitespace normalization semantics.""" + level = RubricLevel( + 0, + " not_met ", + " Requirement is not met. ", + (" missing evidence ",), + ) + + assert level.label == "not_met" + assert level.descriptor == "Requirement is not met." + assert level.observable_indicators == ("missing evidence",) + + +def test_item_bank_enum_rejects_string_subclass_before_enum_lookup() -> None: + """Item-bank evidence kind admission must not dispatch hostile string hooks.""" + + class HostileKind(str): + """Record forbidden equality and hash callbacks during enum lookup.""" + + calls = 0 + + def __eq__(self, other: object) -> bool: + """Record an equality callback that must never execute.""" + type(self).calls += 1 + raise AssertionError("caller equality callback executed") + + def __hash__(self) -> int: + """Record a hash callback that must never execute.""" + type(self).calls += 1 + raise AssertionError("caller hash callback executed") + + value = HostileKind(ItemBankEvidenceKind.CALIBRATION.value) + with pytest.raises(ValueError, match="evidence_kind must be one of"): + ItemBankEvidenceReference( + evidence_kind=value, + evidence_id="calibration_evidence_alpha", + evidence_fingerprint="a" * 64, + ) + + assert HostileKind.calls == 0 diff --git a/tests/test_scoring_contract_public_callback_safety.py b/tests/test_scoring_contract_public_callback_safety.py index 21856638f..07240635f 100644 --- a/tests/test_scoring_contract_public_callback_safety.py +++ b/tests/test_scoring_contract_public_callback_safety.py @@ -52,10 +52,13 @@ def __eq__(self, other): class _DomainErrorString(str): - """String fixture that raises an existing package-owned domain error.""" + """String fixture whose package-error callback must never execute.""" + + calls = 0 def strip(self, chars=None): - """Raise the shared sentinel domain error unchanged.""" + """Record forbidden text dispatch before raising the sentinel error.""" + type(self).calls += 1 raise _SENTINEL_ERROR @@ -71,10 +74,13 @@ def __index__(self): class _KeyboardString(str): - """String fixture proving BaseException is outside the redaction boundary.""" + """String fixture whose BaseException callback must never execute.""" + + calls = 0 def strip(self, chars=None): - """Raise KeyboardInterrupt rather than an ordinary Exception.""" + """Record forbidden text dispatch before raising KeyboardInterrupt.""" + type(self).calls += 1 raise KeyboardInterrupt @@ -214,15 +220,20 @@ def test_response_type_equality_callback_failure_is_redacted() -> None: assert "private enum comparison payload" not in str(captured.value) -def test_package_owned_text_callback_errors_are_preserved_unchanged() -> None: - """String normalization re-raises an existing package-owned domain error.""" +def test_package_owned_text_callback_errors_are_rejected_before_dispatch() -> None: + """String subclasses cannot execute even package-owned error callbacks.""" + _DomainErrorString.calls = 0 with pytest.raises(AssessmentSpecError) as text_error: ConstructSpec( construct_id=_DomainErrorString("argument_quality"), construct_definition="Definition.", rubric_fingerprints=("a" * 64,), ) - assert text_error.value is _SENTINEL_ERROR + + assert text_error.value.code == "invalid_construct_id" + assert text_error.value.path == "$.construct_id" + assert text_error.value is not _SENTINEL_ERROR + assert _DomainErrorString.calls == 0 def test_integer_callback_domain_errors_are_rejected_before_dispatch() -> None: @@ -243,14 +254,23 @@ def test_integer_callback_domain_errors_are_rejected_before_dispatch() -> None: assert _DomainErrorInteger.calls == 0 -def test_base_exceptions_are_not_swallowed_by_callback_redaction() -> None: - """KeyboardInterrupt propagates through text and collection boundaries.""" - with pytest.raises(KeyboardInterrupt): +def test_text_base_exceptions_are_rejected_before_dispatch() -> None: + """String subclasses cannot execute BaseException-raising text callbacks.""" + _KeyboardString.calls = 0 + with pytest.raises(AssessmentSpecError) as text_error: ConstructSpec( construct_id=_KeyboardString("argument_quality"), construct_definition="Definition.", rubric_fingerprints=("a" * 64,), ) + + assert text_error.value.code == "invalid_construct_id" + assert text_error.value.path == "$.construct_id" + assert _KeyboardString.calls == 0 + + +def test_collection_base_exceptions_are_not_swallowed() -> None: + """BaseException still propagates after a collection callback is admitted.""" with pytest.raises(KeyboardInterrupt): ConstructSpec( construct_id="argument_quality", diff --git a/tests/test_scoring_metadata_scalar_callback_safety.py b/tests/test_scoring_metadata_scalar_callback_safety.py new file mode 100644 index 000000000..f5b499e10 --- /dev/null +++ b/tests/test_scoring_metadata_scalar_callback_safety.py @@ -0,0 +1,85 @@ +"""Regression tests for callback-free scoring metadata scalar normalization.""" + +from __future__ import annotations + +import pytest + +from fast_mlsirm.scoring import EngineKind, build_engine_descriptor + + +class _HostileText(str): + """String subclass whose UTF-8 callback must stay inert.""" + + calls = 0 + + def encode(self, *args, **kwargs): + """Record forbidden callback dispatch during metadata validation.""" + type(self).calls += 1 + raise RuntimeError("private metadata text callback") + + +class _HostileInteger(int): + """Integer subclass whose comparison callbacks must stay inert.""" + + calls = 0 + + def __le__(self, other): + """Record forbidden callback dispatch during range validation.""" + type(self).calls += 1 + raise RuntimeError("private metadata integer callback") + + def __ge__(self, other): + """Record forbidden callback dispatch during range validation.""" + type(self).calls += 1 + raise RuntimeError("private metadata integer callback") + + +class _HostileFloat(float): + """Float subclass whose conversion callback must stay inert.""" + + calls = 0 + + def __float__(self): + """Record forbidden callback dispatch during finite-value validation.""" + type(self).calls += 1 + raise RuntimeError("private metadata float callback") + + +def _engine(metadata): + """Build one public automated-engine descriptor around caller metadata.""" + return build_engine_descriptor( + engine_id="metadata_engine", + engine_family_id="metadata_family", + provider_id="local_provider", + engine_version="1.0.0", + engine_kind=EngineKind.AUTOMATED, + model_id="metadata_model", + prompt_driven=False, + prompt_template_fingerprint=None, + metadata=metadata, + ) + + +@pytest.mark.parametrize( + ("control_type", "raw_value", "expected"), + [ + pytest.param(_HostileText, "pilot", "pilot", id="text"), + pytest.param(_HostileInteger, 7, 7, id="integer"), + pytest.param(_HostileFloat, 0.75, 0.75, id="float"), + ], +) +def test_engine_metadata_scalar_subclasses_normalize_without_callbacks( + control_type: type, + raw_value: object, + expected: object, +) -> None: + """JSON scalar subclasses cannot execute caller callbacks during freezing.""" + control_type.calls = 0 + value = control_type(raw_value) + + descriptor = _engine({"deployment_value": value}) + + normalized = descriptor.to_dict()["metadata"]["deployment_value"] + assert normalized == expected + assert type(normalized) is type(expected) + assert control_type.calls == 0