fix(security): enforce explicit trust for local ST modules

packages/sie_server/bundles/default.yaml

@@ -62,7 +62,8 @@ deps:
   # colbert_rotary_flash, jina_flash_cross_encoder
   einops: '>=0.8'
   # sentence_transformer; native causal CrossEncoder module format requires 5.4.1
-  sentence-transformers: '>=5.4.1,<6'
+  # The local-code trust boundary is validated against this exact release.
+  sentence-transformers: '==5.7.0'
   # bge_m3_flag
   FlagEmbedding: '>=1.3'
   # grounding_dino. EXACT, not a range. `uv run --with-requirements` resolves the

packages/sie_server/bundles/transformers5.yaml

@@ -28,9 +28,9 @@ deps:
   # installs already resolved to the latest 5.x, so the floor is the
   # guarantee, not a version change in practice.
   transformers: '>=5.17,<6'
-  # Multimodal SparseEncoder (image documents) landed on the 5.6 line
-  # (st_sparse_vision adapter).
-  sentence-transformers: '>=5.6,<6'
+  # Includes multimodal SparseEncoder (image documents), with the local-code
+  # trust boundary validated against this exact release.
+  sentence-transformers: '==5.7.0'
   # Pin torch — flash-attn's unpinned torch dep can pull in 2.10+ which breaks torchvision
   torch: '>=2.9,<2.10'
   # Keep the compiled torchvision operators paired with torch 2.9.x. A broad

packages/sie_server/pyproject.toml

@@ -23,7 +23,7 @@ dependencies = [
     # the base venv can't silently jump to a transformers 6 major.
     "transformers>=4.57,<6",
     "tokenizers>=0.20,<1",
-    "sentence-transformers>=5.0,<6",
+    "sentence-transformers==5.7.0",
     "safetensors>=0.4,<1",
     # Artifact resolution (snapshot_download / hf_hub_download / scan_cache_dir)
     # is imported directly by the adapters and the disk cache, and `requests` by

packages/sie_server/src/sie_server/adapters/_sentence_transformer_trust.py

@@ -0,0 +1,172 @@
+"""Load-only compatibility boundary for Sentence Transformers 5.x custom code.
+
+The supported 5.x releases implicitly trust local model directories. Enforce an
+explicit opt-in at the actual class-resolution sites without changing library
+globals, native model classes, or their inference methods.
+"""
+
+from __future__ import annotations
+
+import inspect
+from collections.abc import Mapping
+from types import FunctionType, MethodType, ModuleType
+from typing import Any
+
+import sentence_transformers
+from sentence_transformers import CrossEncoder, SentenceTransformer, SparseEncoder
+
+_SUPPORTED_VERSIONS = frozenset({"5.4.1", "5.6.0", "5.7.0"})
+_ROOT_CLASSES = (SentenceTransformer, SparseEncoder, CrossEncoder)
+_RESOLVER = "_load_module_class_from_ref"
+_REFUSAL = "Custom Sentence Transformers code requires trust_remote_code=True."
+_UNSUPPORTED = "Unsupported Sentence Transformers loading implementation."
+_NESTED_IMPORTS = {
+    "5.4.1": ("import_from_string", "import_from_string"),
+    "5.6.0": ("import_module_class", "import_from_string"),
+    "5.7.0": ("import_module_class", "import_module_class"),
+}
+_KWARG_GROUPS = (
+    "model_kwargs",
+    "processor_kwargs",
+    "config_kwargs",
+    "tokenizer_kwargs",
+    "automodel_args",
+    "model_args",
+    "tokenizer_args",
+    "config_args",
+)
+
+
+def _native_references(exports: tuple[ModuleType, ...], base: type, aliases: dict[str, str]) -> dict[str, type]:
+    """Build a closed name map from installed, trusted library exports.
+
+    Model-supplied names never reach an import function, even for malformed
+    names resembling a native namespace. Legacy aliases map to the same class.
+    """
+    references: dict[str, type] = {}
+    for namespace in exports:
+        for name in namespace.__all__:
+            value = getattr(namespace, name)
+            if not inspect.isclass(value) or not issubclass(value, base) or inspect.isabstract(value):
+                continue
+            references[f"{namespace.__name__}.{name}"] = value
+            references[f"{value.__module__}.{value.__name__}"] = value
+    for old, new in aliases.items():
+        for reference, value in tuple(references.items()):
+            if reference.startswith(new + "."):
+                references[old + reference[len(new) :]] = value
+            # Historical import_from_string also accepts a module alias whose
+            # final component is the exported class name (models.Pooling).
+            if new == value.__module__ and old.rsplit(".", 1)[-1] == value.__name__:
+                references[old] = value
+    return references
+
+
+class _NativeLoadBoundary:
+    """Private state used only while one native root constructor executes."""
+
+    def __init__(self) -> None:
+        # This private layout is optional across the public ST5 range. Import
+        # it only after selecting a supported untrusted path, never for True.
+        from sentence_transformers.base import modules as base_modules
+        from sentence_transformers.cross_encoder import modules as cross_modules
+        from sentence_transformers.sentence_transformer import modules as sentence_modules
+        from sentence_transformers.sentence_transformer.modules import tokenizer as tokenizers
+        from sentence_transformers.sparse_encoder import modules as sparse_modules
+        from sentence_transformers.util.deprecated_import import DEPRECATED_MODULE_PATHS
+
+        self.router = base_modules.Router
+        self.word_embeddings = sentence_modules.WordEmbeddings
+        self.modules = _native_references(
+            (base_modules, sentence_modules, sparse_modules, cross_modules),
+            base_modules.Module,
+            DEPRECATED_MODULE_PATHS,
+        )
+        self.tokenizers = _native_references((tokenizers,), tokenizers.WordTokenizer, DEPRECATED_MODULE_PATHS)
+        self.facades: dict[type, type] = {}
+        self.import_names = _NESTED_IMPORTS[sentence_transformers.__version__]
+
+    def resolve_module(self, class_ref: object, *_args: object, **_kwargs: object) -> type:
+        if not isinstance(class_ref, str) or class_ref not in self.modules:
+            raise ValueError(_REFUSAL)
+        native = self.modules[class_ref]
+        if issubclass(native, self.router) or native is self.word_embeddings:
+            return self._facade(native)
+        return native
+
+    def resolve_tokenizer(self, class_ref: object, *_args: object, **_kwargs: object) -> type:
+        if not isinstance(class_ref, str) or class_ref not in self.tokenizers:
+            raise ValueError(_REFUSAL)
+        return self.tokenizers[class_ref]
+
+    def resolve_root(self, _model: object, class_ref: object, *_args: object, **_kwargs: object) -> type:
+        return self.resolve_module(class_ref)
+
+    def _facade(self, native: type) -> type:
+        if native in self.facades:
+            return self.facades[native]
+        method = getattr(native, "load", None)
+        if not isinstance(method, MethodType):
+            raise ValueError(_UNSUPPORTED)
+        loader = method.__func__
+        import_name = self.import_names[1 if native is self.word_embeddings else 0]
+        if not isinstance(loader, FunctionType) or import_name not in loader.__code__.co_names:
+            raise ValueError(_UNSUPPORTED)
+        # Copy globals for this function only. The original library function,
+        # class and globals remain unchanged, including during concurrent loads.
+        globals_copy = dict(loader.__globals__)
+        globals_copy[import_name] = self.resolve_tokenizer if native is self.word_embeddings else self.resolve_module
+        guarded = FunctionType(loader.__code__, globals_copy, loader.__name__, loader.__defaults__, loader.__closure__)
+        guarded.__kwdefaults__ = loader.__kwdefaults__
+        guarded.__annotations__ = loader.__annotations__
+        # Bind cls to the ORIGINAL native class. The facade is never a model
+        # instance; only its load entry point is presented to upstream code.
+        facade = type(native.__name__, (native,), {"load": staticmethod(MethodType(guarded, native))})
+        self.facades[native] = facade
+        return facade
+
+
+def load_sentence_transformer[T: (SentenceTransformer, SparseEncoder, CrossEncoder)](
+    model_class: type[T],
+    model_name_or_path: str,
+    *,
+    trust_remote_code: bool | None = False,
+    **kwargs: Any,
+) -> T:
+    """Construct the native model, enforcing explicit trust for custom classes.
+
+    This compatibility helper intentionally supports only the inspected 5.x
+    releases. An unknown version fails before an untrusted model is loaded.
+    Explicit True preserves the original constructor path exactly.
+    """
+    if trust_remote_code is True:
+        return model_class(model_name_or_path, trust_remote_code=True, **kwargs)
+    if trust_remote_code is not False and trust_remote_code is not None:
+        raise ValueError(_REFUSAL)
+    if sentence_transformers.__version__ not in _SUPPORTED_VERSIONS or model_class not in _ROOT_CLASSES:
+        raise ValueError(_UNSUPPORTED)
+    # The upstream decorators rename these aliases after this boundary. Check
+    # every supported spelling before that happens, including canonical/alias
+    # collisions. Do not mutate the caller's dictionaries.
+    for name in _KWARG_GROUPS:
+        nested = kwargs.get(name)
+        if nested is not None and not isinstance(nested, Mapping):
+            raise ValueError(_REFUSAL)
+        if isinstance(nested, Mapping) and "trust_remote_code" in nested:
+            nested_trust = nested["trust_remote_code"]
+            if nested_trust is not False and nested_trust is not None:
+                raise ValueError(_REFUSAL)
+            kwargs[name] = {**nested, "trust_remote_code": False}
+
+    boundary = _NativeLoadBoundary()
+    model = model_class.__new__(model_class)
+    # nn.Module is not initialized yet; use object assignment deliberately.
+    # Native nn.Module initialization preserves unrelated instance attributes.
+    object.__setattr__(model, _RESOLVER, MethodType(boundary.resolve_root, model))
+    try:
+        model_class.__init__(model, model_name_or_path, trust_remote_code=False, **kwargs)
+        return model
+    finally:
+        # Remove even if construction/nn.Module initialization failed. Never
+        # leave a bound method cycle or an inference-time compatibility wrapper.
+        object.__getattribute__(model, "__dict__").pop(_RESOLVER, None)

packages/sie_server/src/sie_server/adapters/cross_encoder/__init__.py

@@ -25,6 +25,7 @@
 from sentence_transformers import CrossEncoder
 
 from sie_server.adapters._base_adapter import BaseAdapter
+from sie_server.adapters._sentence_transformer_trust import load_sentence_transformer
 from sie_server.adapters._spec import AdapterSpec
 from sie_server.adapters._types import ERR_NOT_LOADED, ERR_REQUIRES_TEXT, AttnImplementation, ComputePrecision
 from sie_server.adapters.cross_encoder._cpu_alignment import normalize_bert_cpu_storage
@@ -127,7 +128,8 @@ def load(self, device: str) -> None:
                 model_kwargs["dtype"] = torch.bfloat16
             # float32 -> CrossEncoder default, no dtype kwarg needed
 
-        model = CrossEncoder(
+        model = load_sentence_transformer(
+            CrossEncoder,
             self._model_name_or_path,
             device=device,
             trust_remote_code=self._trust_remote_code,

packages/sie_server/src/sie_server/adapters/cross_encoder/_cpu_alignment.py

@@ -10,7 +10,7 @@
 from transformers import BertForSequenceClassification
 
 
-def normalize_bert_cpu_storage(model: torch.nn.Module) -> None:
+def normalize_bert_cpu_storage(model: torch.nn.Module | None) -> None:
     """Copy unsafe standard BERT float32 storage once, retaining tensor objects.
 
     The caller restricts this optional compatibility path to Darwin arm64 CPU.

packages/sie_server/src/sie_server/adapters/cross_encoder/_lora.py

@@ -113,12 +113,12 @@ def runtime_config(self) -> LoraRuntimeConfig:
 
     def bind(self, outer: CrossEncoder) -> None:
         model = outer.model
-        encoder_prefix = _ENCODER_PREFIXES.get(type(model))
-        if encoder_prefix is None or getattr(model, "is_quantized", False):
+        if model is None or type(model) not in _ENCODER_PREFIXES or getattr(model, "is_quantized", False):
             raise ValueError(
                 "CrossEncoder LoRA requires a standard, unquantized BertForSequenceClassification "
                 "or XLMRobertaForSequenceClassification"
             )
+        encoder_prefix = _ENCODER_PREFIXES[type(model)]
         if getattr(model, "peft_config", None) or getattr(model, "_hf_peft_config_loaded", False):
             raise ValueError("The cross-encoder must start from a base checkpoint without attached adapters")
         self._outer, self._model = outer, model

packages/sie_server/src/sie_server/adapters/native_causal_cross_encoder/adapter.py

@@ -125,10 +125,13 @@ def load(self, device: str) -> None:
         from sentence_transformers.cross_encoder.modules import LogitScore
         from sentence_transformers.util import batch_to_device
 
+        from sie_server.adapters._sentence_transformer_trust import load_sentence_transformer
+
         self._require_saved_modules()
         precision = self._compute_precision or ("bfloat16" if device.startswith("cuda") else "float32")
         dtype = {"float16": torch.float16, "bfloat16": torch.bfloat16, "float32": torch.float32}[precision]
-        model = CrossEncoder(
+        model = load_sentence_transformer(
+            CrossEncoder,
             self._model_name_or_path,
             revision=self._revision,
             device=device,

packages/sie_server/src/sie_server/adapters/sentence_transformer/__init__.py

@@ -15,6 +15,7 @@
 from sentence_transformers import SentenceTransformer, SparseEncoder
 
 from sie_server.adapters._base_adapter import BaseAdapter
+from sie_server.adapters._sentence_transformer_trust import load_sentence_transformer
 from sie_server.adapters._spec import AdapterSpec
 from sie_server.adapters._types import ERR_NOT_LOADED, ERR_REQUIRES_TEXT, ComputePrecision
 from sie_server.core.inference_output import EncodeOutput, SparseVector
@@ -88,7 +89,7 @@ def load(self, device: str) -> None:
         # Omit revision unless pinned so the library default (main) is unchanged.
         if self._revision is not None:
             load_kwargs["revision"] = self._revision
-        self._model = SentenceTransformer(self._model_name_or_path, **load_kwargs)
+        self._model = load_sentence_transformer(SentenceTransformer, self._model_name_or_path, **load_kwargs)
 
         if self._max_seq_length is not None:
             self._model.max_seq_length = self._max_seq_length
@@ -249,7 +250,7 @@ def load(self, device: str) -> None:
         # Omit revision unless pinned so the library default (main) is unchanged.
         if self._revision is not None:
             load_kwargs["revision"] = self._revision
-        self._model = SparseEncoder(self._model_name_or_path, **load_kwargs)
+        self._model = load_sentence_transformer(SparseEncoder, self._model_name_or_path, **load_kwargs)
 
         if self._max_seq_length is not None:
             self._model.max_seq_length = self._max_seq_length

packages/sie_server/src/sie_server/adapters/st_sparse_vision/adapter.py

@@ -4,9 +4,9 @@
 documents in addition to text (visual-SPLADE family, e.g.
 naver/v-splade-quality): text or page-image in, sparse vocab-sized vector out.
 
-Requires the ``transformers5`` bundle (sentence-transformers >=5.6 /
-transformers >=5.14); the multimodal ``SparseEncoder`` routing does not exist
-on the sentence-transformers 5.0 line the default bundle pins.
+Requires the ``transformers5`` bundle for the checkpoint's Transformers >=5.14
+modeling support. Both this bundle and the default bundle pin
+sentence-transformers 5.7.0, which includes multimodal ``SparseEncoder`` routing.
 """
 
 from __future__ import annotations
@@ -21,6 +21,7 @@
 from sentence_transformers import SparseEncoder
 
 from sie_server.adapters._base_adapter import BaseAdapter
+from sie_server.adapters._sentence_transformer_trust import load_sentence_transformer
 from sie_server.adapters._spec import AdapterSpec
 from sie_server.adapters._types import ERR_NOT_LOADED, ComputePrecision
 from sie_server.adapters._utils import validate_output_types
@@ -106,7 +107,7 @@ def load(self, device: str) -> None:
             load_kwargs["revision"] = self._revision
         if self._trust_remote_code:
             self._register_remote_code_dir()
-        self._model = SparseEncoder(self._model_name_or_path, **load_kwargs)
+        self._model = load_sentence_transformer(SparseEncoder, self._model_name_or_path, **load_kwargs)
 
         if self._max_seq_length is not None:
             self._model.max_seq_length = self._max_seq_length

packages/sie_server/tests/adapters/test_cross_encoder_cpu_alignment.py

@@ -225,7 +225,7 @@ def test_off_platform_load_does_not_invoke_alignment(monkeypatch, system, machin
     cache_calls = []
     monkeypatch.setattr(cross_encoder.sys, "platform", system)
     monkeypatch.setattr(cross_encoder.platform, "machine", lambda: machine)
-    monkeypatch.setattr(cross_encoder, "CrossEncoder", lambda *args, **kwargs: outer)
+    monkeypatch.setattr(cross_encoder, "load_sentence_transformer", lambda *args, **kwargs: outer)
     # Both the model and its accelerator are simulated, including cache cleanup.
     monkeypatch.setattr(torch.cuda, "empty_cache", lambda: cache_calls.append("cuda"))
     monkeypatch.setattr(torch.mps, "empty_cache", lambda: cache_calls.append("mps"))
@@ -250,7 +250,7 @@ def test_failed_normalization_never_publishes_outer(model, monkeypatch):
     outer = SimpleNamespace(model=model)
     monkeypatch.setattr(cross_encoder.sys, "platform", "darwin")
     monkeypatch.setattr(cross_encoder.platform, "machine", lambda: "arm64")
-    monkeypatch.setattr(cross_encoder, "CrossEncoder", lambda *args, **kwargs: outer)
+    monkeypatch.setattr(cross_encoder, "load_sentence_transformer", lambda *args, **kwargs: outer)
     adapter = cross_encoder.CrossEncoderAdapter("unused")
     with pytest.raises(ValueError, match="storage overlaps"):
         adapter.load("cpu")
@@ -263,7 +263,9 @@ def test_failed_copy_never_publishes_outer(model, monkeypatch):
     before = pointers(model)
     monkeypatch.setattr(cross_encoder.sys, "platform", "darwin")
     monkeypatch.setattr(cross_encoder.platform, "machine", lambda: "arm64")
-    monkeypatch.setattr(cross_encoder, "CrossEncoder", lambda *args, **kwargs: SimpleNamespace(model=model))
+    monkeypatch.setattr(
+        cross_encoder, "load_sentence_transformer", lambda *args, **kwargs: SimpleNamespace(model=model)
+    )
 
     def fail_copy(*args, **kwargs):
         raise MemoryError("controlled load copy failure")
@@ -305,7 +307,7 @@ def factory(*args, **kwargs):
 
     monkeypatch.setattr(cross_encoder.sys, "platform", "darwin")
     monkeypatch.setattr(cross_encoder.platform, "machine", lambda: "arm64")
-    monkeypatch.setattr(cross_encoder, "CrossEncoder", factory)
+    monkeypatch.setattr(cross_encoder, "load_sentence_transformer", factory)
     adapter = cross_encoder.CrossEncoderAdapter("unused")
 
     def attempt():

packages/sie_server/tests/adapters/test_native_causal_cross_encoder.py

@@ -13,7 +13,6 @@
 
 import numpy as np
 import pytest
-import sentence_transformers
 import torch
 from jinja2.sandbox import ImmutableSandboxedEnvironment
 from sentence_transformers.base import modules as native_modules
@@ -141,7 +140,10 @@ def construct(*args: Any, **kwargs: Any) -> ToyCrossEncoder:
         return model
 
     loader = Mock(side_effect=construct)
-    monkeypatch.setattr(sentence_transformers, "CrossEncoder", loader)
+    monkeypatch.setattr(
+        "sie_server.adapters._sentence_transformer_trust.load_sentence_transformer",
+        lambda model_class, *args, **kwargs: loader(*args, **kwargs),
+    )
     monkeypatch.setattr(native_modules, "Transformer", ToyTransformer)
     modules_file = tmp_path / "modules.json"
     modules_file.write_text(