Skip to content
Merged
Show file tree
Hide file tree
Changes from 7 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions docs/changelog.d/991-rubric-text-trust-boundary.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
# 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.
10 changes: 6 additions & 4 deletions python/fast_mlsirm/rubric/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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


Expand Down Expand Up @@ -558,4 +560,4 @@ def to_dict(self) -> dict[str, Any]:
**self._fingerprint_payload(),
"blueprint_id": self.blueprint_id,
"blueprint_fingerprint": self.blueprint_fingerprint,
}
}
110 changes: 110 additions & 0 deletions tests/test_rubric_text_callback_safety.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
"""Regression tests for callback-free rubric text admission."""

from __future__ import annotations

import pytest

from fast_mlsirm.rubric import 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):
calls = 0

def strip(self, *args: object, **kwargs: object) -> str:
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",)
40 changes: 30 additions & 10 deletions tests/test_scoring_contract_public_callback_safety.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand All @@ -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


Expand Down Expand Up @@ -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:
Expand All @@ -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",
Expand Down
Loading