fix(rubric): harden text schema callback boundary

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.

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
 
 

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",

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,
-        }
+        }

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",

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