fix: keep sentencepiece tokenizers out of xgboost routing

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Bug Fixes
 
+- keep strongly identified SentencePiece tokenizer `.model` artifacts, including custom-unknown models with disabled special tokens, out of XGBoost routing while preserving fail-closed `.model` ambiguity for malformed tails.
 - tolerate picklescan reports that omit private metadata
 - stream bounded Llamafile runtime coverage across preview gaps and report incomplete runtime reads or bounds
 - treat canonical storage persistent IDs as informational only after legacy PyTorch framing, storage tuples, and storage payloads validate completely

modelaudit/core.py

@@ -82,9 +82,11 @@ def shared_source_sensitive_caches() -> Iterator[None]:
     ONNX_ROUTING_INCONCLUSIVE_FORMAT,
     PICKLE_ROUTING_INCONCLUSIVE_FORMAT,
     PROTOBUF_MODEL_CANDIDATE_FORMAT,
+    SENTENCEPIECE_MODEL_PROTO_INCONCLUSIVE_FORMAT,
     TENSORFLOW_PROTOBUF_ROUTING_INCONCLUSIVE_FORMAT,
     XGBOOST_UBJSON_ROUTING_INCONCLUSIVE_FORMAT,
     XML_MODEL_INCONCLUSIVE_FORMAT,
+    _is_malformed_sentencepiece_model_proto_candidate_file,
     detect_file_format,
     detect_file_format_from_magic,
     detect_flax_msgpack_overlap_routes,
@@ -96,6 +98,7 @@ def shared_source_sensitive_caches() -> Iterator[None]:
     is_executorch_archive,
     is_keras_zip_archive,
     is_pytorch_zip_archive,
+    is_sentencepiece_model_proto_file,
     is_skops_archive,
     is_torchserve_mar_archive,
     should_defer_safetensors_header_limit_hash,
@@ -277,6 +280,7 @@ def is_covered(target: Path) -> bool:
 _FORMAT_DETECTION_READ_FAILED_REASON = "format_detection_read_failed"
 _XML_MODEL_ROUTING_INCOMPLETE_REASON = "xml_model_routing_incomplete"
 _PROTOBUF_MODEL_ROUTING_INCOMPLETE_REASON = "protobuf_model_routing_incomplete"
+_SENTENCEPIECE_MODEL_PROTO_ROUTING_INCOMPLETE_REASON = "sentencepiece_model_proto_routing_incomplete"
 _LLAMAFILE_ROUTING_INCOMPLETE_REASON = "llamafile_routing_incomplete"
 _MXNET_SYMBOL_ROUTING_INCOMPLETE_REASON = "mxnet_symbol_routing_incomplete"
 _PICKLE_ROUTING_INCOMPLETE_REASON = "pickle_routing_incomplete"
@@ -1504,6 +1508,26 @@ def _make_incomplete_protobuf_model_result(path: str) -> ScanResult:
     return result
 
 
+def _make_incomplete_sentencepiece_model_proto_result(path: str) -> ScanResult:
+    """Fail closed when a SentencePiece-like protobuf fails ownership validation."""
+    result = ScanResult(scanner_name="unknown")
+    result.add_check(
+        name="SentencePiece ModelProto Routing",
+        passed=False,
+        message=(
+            "SentencePiece ModelProto routing was inconclusive because the payload "
+            "looked like a tokenizer protobuf but failed ownership validation"
+        ),
+        severity=IssueSeverity.INFO,
+        location=path,
+        details={"format": SENTENCEPIECE_MODEL_PROTO_INCONCLUSIVE_FORMAT, "path": path},
+    )
+    _mark_inconclusive_scan_outcome(result, _SENTENCEPIECE_MODEL_PROTO_ROUTING_INCOMPLETE_REASON)
+    _mark_operational_scan_error(result, _SENTENCEPIECE_MODEL_PROTO_ROUTING_INCOMPLETE_REASON)
+    result.finish(success=False)
+    return result
+
+
 def _make_incomplete_llamafile_routing_result(path: str, config: dict[str, Any]) -> ScanResult:
     """Fail closed when an executable Llamafile marker probe cannot complete."""
     result = ScanResult(scanner_name="unknown")
@@ -3654,6 +3678,7 @@ def _scan_file_internal(path: str, config: dict[str, Any] | None = None) -> Scan
         format_probe_error is None
         and header_format in {"unknown", "pytorch_binary"}
         and pytorch_binary_supplemental_scanner_id is None
+        and not (ext == ".model" and _is_malformed_sentencepiece_model_proto_candidate_file(path))
         and scanner_selection.allows("zip")
         and ZipScanner.can_handle(path)
     ):
@@ -3727,6 +3752,13 @@ def _scan_file_internal(path: str, config: dict[str, Any] | None = None) -> Scan
             if nested_xgboost_route == "xgboost":
                 config[XGBOOST_CONTENT_ROUTED_UBJSON_CONFIG_KEY] = True
     is_xgboost_pickle_spoof = ext in _XGBOOST_BINARY_EXTENSIONS and header_format == "pickle"
+    sentencepiece_model_proto_owned = (
+        format_probe_error is None
+        and ext == ".model"
+        and header_format == "unknown"
+        and magic_format == "unknown"
+        and is_sentencepiece_model_proto_file(path)
+    )
     # Record telemetry for file type detection
     detected_format = header_format if header_format != "unknown" else ext_format
     record_file_type_detected(path, detected_format)
@@ -3772,6 +3804,14 @@ def _scan_file_internal(path: str, config: dict[str, Any] | None = None) -> Scan
         if sr.bytes_scanned == 0 and file_size > 0:
             sr.bytes_scanned = file_size
         return sr
+    if (
+        header_format == SENTENCEPIECE_MODEL_PROTO_INCONCLUSIVE_FORMAT
+        or magic_format == SENTENCEPIECE_MODEL_PROTO_INCONCLUSIVE_FORMAT
+    ):
+        sr = _make_incomplete_sentencepiece_model_proto_result(path)
+        if sr.bytes_scanned == 0 and file_size > 0:
+            sr.bytes_scanned = file_size
+        return sr
     if header_format == PICKLE_ROUTING_INCONCLUSIVE_FORMAT or magic_format == PICKLE_ROUTING_INCONCLUSIVE_FORMAT:
         sr = _make_incomplete_pickle_routing_result(path)
         merge_safetensors_overlap_analysis(
@@ -3990,7 +4030,7 @@ def _scan_file_internal(path: str, config: dict[str, Any] | None = None) -> Scan
                 and scanner_selection.allows(fallback_scanner_id)
             ):
                 scanner_class = _registry.load_scanner_by_id(fallback_scanner_id)
-        elif scanner_class is None:
+        elif scanner_class is None and not sentencepiece_model_proto_owned:
             scanner_class = _registry.get_scanner_for_path(
                 path,
                 scanner_selection=scanner_selection if scanner_selection.active else None,
@@ -4066,7 +4106,11 @@ def _scan_file_internal(path: str, config: dict[str, Any] | None = None) -> Scan
                     kind=SCANNER_SELECTION_PREFERRED_KIND,
                 )
         else:
-            if unavailable_preferred_scanner_id is None and scanner_selection.active:
+            if (
+                unavailable_preferred_scanner_id is None
+                and scanner_selection.active
+                and not sentencepiece_model_proto_owned
+            ):
                 candidate_scanner_id = skipped_preferred_scanner_id
                 if candidate_scanner_id is None:
                     candidate_scanner_class = _registry.get_scanner_for_path(path)
@@ -4126,6 +4170,8 @@ def _scan_file_internal(path: str, config: dict[str, Any] | None = None) -> Scan
                 sr = _make_incomplete_xgboost_ubjson_routing_result(path)
             elif magic_format == ONNX_ROUTING_INCONCLUSIVE_FORMAT:
                 sr = _make_incomplete_onnx_routing_result(path)
+            elif magic_format == SENTENCEPIECE_MODEL_PROTO_INCONCLUSIVE_FORMAT:
+                sr = _make_incomplete_sentencepiece_model_proto_result(path)
             elif magic_format == PICKLE_ROUTING_INCONCLUSIVE_FORMAT:
                 sr = _make_incomplete_pickle_routing_result(path)
             elif magic_format == TENSORFLOW_PROTOBUF_ROUTING_INCONCLUSIVE_FORMAT:

modelaudit/scanners/__init__.py

@@ -367,6 +367,11 @@ def get_scanner_for_path(
         # If stricter extension-specific scanners all decline, fall back to the
         # generic ZIP scanner so helper-level routing does not drop coverage.
         if is_zip_file and (scanner_selection is None or scanner_selection.allows("zip")):
+            if file_ext == ".model":
+                from modelaudit.utils.file.detection import _is_malformed_sentencepiece_model_proto_candidate_file
+
+                if _is_malformed_sentencepiece_model_proto_candidate_file(path):
+                    return None
             scanner_class = self._load_scanner("zip")
             if scanner_class and scanner_class.can_handle(path):
                 return scanner_class

modelaudit/scanners/archive_dispatch.py

@@ -38,9 +38,11 @@
     ONNX_ROUTING_INCONCLUSIVE_FORMAT,
     PICKLE_ROUTING_INCONCLUSIVE_FORMAT,
     PROTOBUF_MODEL_CANDIDATE_FORMAT,
+    SENTENCEPIECE_MODEL_PROTO_INCONCLUSIVE_FORMAT,
     TENSORFLOW_PROTOBUF_ROUTING_INCONCLUSIVE_FORMAT,
     XGBOOST_UBJSON_ROUTING_INCONCLUSIVE_FORMAT,
     XML_MODEL_INCONCLUSIVE_FORMAT,
+    _is_malformed_sentencepiece_model_proto_candidate_file,
     detect_file_format,
     detect_file_format_from_magic,
     detect_flax_msgpack_overlap_routes,
@@ -87,6 +89,7 @@ def _build_header_format_to_scanner_id() -> dict[str, str]:
 _RECOGNIZED_FORMAT_SCANNER_UNAVAILABLE_REASON = "recognized_format_scanner_unavailable"
 _XML_MODEL_ROUTING_INCOMPLETE_REASON = "xml_model_routing_incomplete"
 _PROTOBUF_MODEL_ROUTING_INCOMPLETE_REASON = "protobuf_model_routing_incomplete"
+_SENTENCEPIECE_MODEL_PROTO_ROUTING_INCOMPLETE_REASON = "sentencepiece_model_proto_routing_incomplete"
 _LLAMAFILE_ROUTING_INCOMPLETE_REASON = "llamafile_routing_incomplete"
 _MXNET_SYMBOL_ROUTING_INCOMPLETE_REASON = "mxnet_symbol_routing_incomplete"
 _PICKLE_ROUTING_INCOMPLETE_REASON = "pickle_routing_incomplete"
@@ -426,6 +429,26 @@ def _make_incomplete_protobuf_model_result(path: str) -> ScanResult:
     return result
 
 
+def _make_incomplete_sentencepiece_model_proto_result(path: str) -> ScanResult:
+    """Fail closed when a nested SentencePiece-like protobuf fails ownership validation."""
+    result = ScanResult(scanner_name="unknown")
+    result.add_check(
+        name="SentencePiece ModelProto Routing",
+        passed=False,
+        message=(
+            "SentencePiece ModelProto routing was inconclusive because the payload "
+            "looked like a tokenizer protobuf but failed ownership validation"
+        ),
+        severity=IssueSeverity.INFO,
+        location=path,
+        details={"format": SENTENCEPIECE_MODEL_PROTO_INCONCLUSIVE_FORMAT, "path": path},
+    )
+    mark_inconclusive_scan_result(result, _SENTENCEPIECE_MODEL_PROTO_ROUTING_INCOMPLETE_REASON)
+    mark_operational_scan_error(result, _SENTENCEPIECE_MODEL_PROTO_ROUTING_INCOMPLETE_REASON)
+    result.finish(success=False)
+    return result
+
+
 def _deduplicate_exact_merged_findings(result: ScanResult) -> None:
     """Remove identical output emitted by composed subtype and ZIP analyses."""
 
@@ -1199,8 +1222,12 @@ def with_safetensors_overlap(result: ScanResult) -> ScanResult:
     except (TypeError, ValueError):
         max_zip_entries = ZipScanner.DEFAULT_MAX_ENTRIES
     max_zip_directory_size = ZipScanner.central_directory_size_limit(raw_config)
+    malformed_sentencepiece_model_candidate = Path(
+        path
+    ).suffix.lower() == ".model" and _is_malformed_sentencepiece_model_proto_candidate_file(path)
     if (
-        (not is_safetensors_hdf5_overlap or hdf5_signature_offset in (None, 0))
+        not malformed_sentencepiece_model_candidate
+        and (not is_safetensors_hdf5_overlap or hdf5_signature_offset in (None, 0))
         and allows_zip_structure_analysis(scanner_selection, path)
         and ZipScanner.requires_preflight_result(
             path,
@@ -1395,6 +1422,11 @@ def with_safetensors_overlap(result: ScanResult) -> ScanResult:
 
         if trusted_content_format == XML_MODEL_INCONCLUSIVE_FORMAT:
             return with_safetensors_overlap(_make_incomplete_xml_model_result(path))
+        if (
+            routed_content_format == SENTENCEPIECE_MODEL_PROTO_INCONCLUSIVE_FORMAT
+            or trusted_content_format == SENTENCEPIECE_MODEL_PROTO_INCONCLUSIVE_FORMAT
+        ):
+            return with_safetensors_overlap(_make_incomplete_sentencepiece_model_proto_result(path))
         if routed_content_format == PROTOBUF_MODEL_CANDIDATE_FORMAT:
             return with_safetensors_overlap(_make_incomplete_protobuf_model_result(path))
         if trusted_content_format == PROTOBUF_MODEL_CANDIDATE_FORMAT and routed_content_format != "unknown":

modelaudit/scanners/compressed_scanner.py

@@ -15,6 +15,10 @@
 from .. import core
 from ..scanner_results import INCONCLUSIVE_SCAN_OUTCOME, mark_inconclusive_scan_result
 from ..utils.file._compression import is_zlib_header
+from ..utils.file.detection import (
+    _is_malformed_sentencepiece_model_proto_candidate_file,
+    is_sentencepiece_model_proto_file,
+)
 from ._archive_config import get_archive_depth
 from ._archive_locations import rewrite_extracted_member_location
 from ._archive_outcomes import member_scan_incomplete
@@ -181,7 +185,44 @@ def _derive_inner_suffix(path: str) -> str:
             wrapper_path.name[: -len(wrapper_path.suffix)] if wrapper_path.suffix else wrapper_path.name
         )
         inferred_suffix = Path(stem_without_wrapper).suffix
-        return inferred_suffix or ".bin"
+        if inferred_suffix:
+            return inferred_suffix
+        if stem_without_wrapper.lower() in {"tokenizer", "spiece", "sentencepiece"}:
+            return ""
+        return ".bin"
+
+    @staticmethod
+    def _uses_tokenizer_extensionless_inner(path: str) -> bool:
+        wrapper_path = Path(path)
+        if not wrapper_path.suffix or not CompressedScanner._has_declared_wrapper_extension(path):
+            return False
+        stem_without_wrapper = wrapper_path.name[: -len(wrapper_path.suffix)]
+        return not Path(stem_without_wrapper).suffix and stem_without_wrapper.lower() in {
+            "tokenizer",
+            "spiece",
+            "sentencepiece",
+        }
+
+    @staticmethod
+    def _is_sentencepiece_candidate(path: str) -> bool:
+        return is_sentencepiece_model_proto_file(path) or _is_malformed_sentencepiece_model_proto_candidate_file(path)
+
+    @staticmethod
+    def _replace_temp_suffix(path: str, suffix: str) -> str:
+        fd, replacement_path = tempfile.mkstemp(suffix=suffix)
+        os.close(fd)
+        os.unlink(replacement_path)
+        os.replace(path, replacement_path)
+        return replacement_path
+
+    @classmethod
+    def _route_tokenizer_extensionless_or_bin(cls, wrapper_path: str, temp_paths: list[str]) -> list[str]:
+        if not cls._uses_tokenizer_extensionless_inner(wrapper_path):
+            return temp_paths
+        return [
+            temp_path if cls._is_sentencepiece_candidate(temp_path) else cls._replace_temp_suffix(temp_path, ".bin")
+            for temp_path in temp_paths
+        ]
 
     @staticmethod
     def _derive_inner_display_name(path: str) -> str:
@@ -916,6 +957,9 @@ def scan(self, path: str) -> ScanResult:
         decompressed_bytes = 0
         try:
             temp_path, member_temp_paths, decompressed_bytes = self._decompress_to_tempfiles(path, expected_codec)
+            routed_temp_paths = self._route_tokenizer_extensionless_or_bin(path, [temp_path, *member_temp_paths])
+            temp_path = routed_temp_paths[0]
+            member_temp_paths = routed_temp_paths[1:]
             temp_paths = [temp_path, *member_temp_paths]
             result.metadata["decompressed_bytes"] = decompressed_bytes
             result.metadata["compressed_member_count"] = len(member_temp_paths)

modelaudit/scanners/xgboost_scanner.py

@@ -30,7 +30,7 @@
 from typing import Any, ClassVar, cast
 
 from ..scanner_selection import add_scanner_selection_skip_check, policy_from_config
-from ..utils.file.detection import has_jax_json_checkpoint_structure
+from ..utils.file.detection import has_jax_json_checkpoint_structure, is_sentencepiece_model_proto_file
 from .base import INCONCLUSIVE_SCAN_OUTCOME, BaseScanner, IssueSeverity, ScanResult
 
 logger = logging.getLogger(__name__)
@@ -513,9 +513,10 @@ def can_handle(cls, path: str) -> bool:
         if file_ext == ".json":
             return cls._is_xgboost_json(path) or cls._is_probable_xgboost_json_candidate(path)
 
-        # For .model files, accept (generic extension)
+        # For .model files, keep ambiguous inputs on the XGBoost route but do
+        # not claim strongly identified SentencePiece tokenizer ModelProto files.
         if file_ext == ".model":
-            return True
+            return not is_sentencepiece_model_proto_file(path)
 
         # Check for XGBoost files without extension
         if file_ext == "":