fix(runtime): harden mount-free transport against symlink escapes and add path grants

docs/advanced/configuration.mdx

@@ -124,6 +124,10 @@ When remote vars are set, Strix dual-writes telemetry to both local JSONL and th
 
 ## Sandbox Configuration
 
+<ParamField path="STRIX_REQUIRE_MOUNT_FREE" default="0" type="string">
+  Enforces mount-free (bind-mount-free) transport for local-code Docker scans. Set to `1`, `true`, `yes`, or `on` to ensure no host directories are bind-mounted into the sandbox container. Local project files are transferred as an isolated snapshot payload.
+</ParamField>
+
 <ParamField path="STRIX_SANDBOX_EXECUTION_TIMEOUT" default="120" type="integer">
   Maximum execution time in seconds for sandbox operations.
 </ParamField>

strix/config/models.py

@@ -445,6 +445,137 @@ def _response_usage(usage: Usage | None) -> ResponseUsage | None:
     )
 
 
+class _TextTagDispatchModel(Model):
+    """Fallback dispatch mode for models that fail to emit structured tool_calls.
+    
+    Extracts [TOOL: name] ... [/TOOL] text tags from the response output and
+    synthesizes ResponseFunctionToolCall instances so the SDK can execute them.
+    """
+
+    def __init__(self, inner: Model) -> None:
+        self._inner = inner
+
+    async def close(self) -> None:
+        await self._inner.close()
+
+    def get_retry_advice(self, request: ModelRetryAdviceRequest) -> ModelRetryAdvice | None:
+        return self._inner.get_retry_advice(request)
+
+    async def get_response(
+        self,
+        system_instructions: str | None,
+        input: str | list[TResponseInputItem],  # noqa: A002
+        model_settings: ModelSettings,
+        tools: list[Tool],
+        output_schema: AgentOutputSchemaBase | None,
+        handoffs: list[Handoff],
+        tracing: ModelTracing,
+        *,
+        previous_response_id: str | None,
+        conversation_id: str | None,
+        prompt: ResponsePromptParam | None,
+    ) -> ModelResponse:
+        import re
+        import uuid
+        import json
+        from agents.items import ResponseFunctionToolCall
+
+        # We need the inner get_response first
+        response = await self._inner.get_response(
+            system_instructions,
+            input,
+            model_settings,
+            tools,
+            output_schema,
+            handoffs,
+            tracing,
+            previous_response_id=previous_response_id,
+            conversation_id=conversation_id,
+            prompt=prompt,
+        )
+
+        TEXT_TAG_PATTERN = re.compile(r"\[TOOL:\s*([^\]]+)\](.*?)\[/TOOL\]", re.DOTALL | re.IGNORECASE)
+        new_output = []
+
+        for item in response.output:
+            if getattr(item, "type", None) == "message":
+                raw_content = getattr(item, "content", "")
+                
+                if isinstance(raw_content, list):
+                    content = ""
+                    for part in raw_content:
+                        if isinstance(part, str):
+                            content += part
+                        elif isinstance(part, dict) and "text" in part:
+                            content += part["text"]
+                        elif hasattr(part, "text"):
+                            content += part.text
+                else:
+                    content = str(raw_content) if raw_content else ""
+                    
+                if content and "[TOOL:" in content:
+                    matches = list(TEXT_TAG_PATTERN.finditer(content))
+                    if matches:
+                        clean_content = TEXT_TAG_PATTERN.sub("", content).strip()
+                        if clean_content:
+                            try:
+                                item.content = clean_content
+                            except AttributeError:
+                                if hasattr(item, "raw_item") and hasattr(item.raw_item, "content"):
+                                    item.raw_item.content = clean_content
+                            new_output.append(item)
+                        for match in matches:
+                            tool_name = match.group(1).strip()
+                            tool_args = match.group(2).strip()
+                            # Check if the args are valid JSON, otherwise it will fail gracefully later
+                            try:
+                                json.loads(tool_args)
+                            except ValueError:
+                                pass
+
+                            tool_call = ResponseFunctionToolCall(
+                                id=uuid.uuid4().hex[:8],
+                                name=tool_name,
+                                arguments=tool_args,
+                                caller="agent",
+                            )
+                            new_output.append(tool_call)
+                        continue
+            new_output.append(item)
+
+        response.output = new_output
+        return response
+
+    async def stream_response(
+        self,
+        system_instructions: str | None,
+        input: str | list[TResponseInputItem],  # noqa: A002
+        model_settings: ModelSettings,
+        tools: list[Tool],
+        output_schema: AgentOutputSchemaBase | None,
+        handoffs: list[Handoff],
+        tracing: ModelTracing,
+        *,
+        previous_response_id: str | None,
+        conversation_id: str | None,
+        prompt: ResponsePromptParam | None,
+    ) -> AsyncIterator[TResponseStreamEvent]:
+        # Text-tag parsing over a stream is complex; delegate to get_response like _NonStreamingModel
+        response = await self.get_response(
+            system_instructions,
+            input,
+            model_settings,
+            tools,
+            output_schema,
+            handoffs,
+            tracing,
+            previous_response_id=previous_response_id,
+            conversation_id=conversation_id,
+            prompt=prompt,
+        )
+        yield _completed_stream_event(response, getattr(self._inner, "model", None))
+
+
 class StrixProvider(MultiProvider):
     """Route any non-OpenAI prefix through LiteLLM with the prefix preserved,
     so users type ``deepseek/deepseek-chat`` rather than
@@ -483,12 +614,9 @@ class StrixProvider(MultiProvider):
             )
         else:
             model = super().get_model(model_name)
-            if type(model).__name__ == "LitellmModel":
-                if llm.api_key:
-                    model.api_key = llm.api_key
-                if llm.api_base:
-                    model.base_url = llm.api_base
-            if llm.disable_streaming:
+            if getattr(llm, "tool_mode", "native") == "text-tags":
+                model = _TextTagDispatchModel(model)
+            if llm.disable_streaming or getattr(llm, "tool_mode", "native") == "text-tags":
                 model = _NonStreamingModel(model)
                 # The wrapper emits its single event only once the whole request
                 # is done, so an idle gap is meaningless here; the request
@@ -567,9 +695,11 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
     _configure_openrouter_attribution(llm.model)
     if llm.api_key:
         set_default_openai_key(llm.api_key, use_for_tracing=False)
+        _configure_litellm_default("api_key", llm.api_key)
         _mirror_api_key_to_provider_env(llm.model, llm.api_key)
     if llm.api_base:
         os.environ["OPENAI_BASE_URL"] = llm.api_base
+        _configure_litellm_default("api_base", llm.api_base)
         set_default_openai_api("chat_completions")
     else:
         set_default_openai_api("responses")
@@ -752,6 +882,18 @@ def uses_chat_completions_tool_schema(model_name: str, settings: Settings) -> bo
     return not model_supports_reasoning(model_name)
 
 
+def supports_strict_tool_schemas(model_name: str) -> bool:
+    """Return whether the route accepts strict tool schemas for Strix's toolset.
+
+    Claude caps a request at 20 strict tools and 16 union-typed parameters
+    across all strict schemas. Strix ships ~30 tools and the strict dialect
+    turns every optional parameter into a nullable union, so both caps are
+    exceeded and the request is rejected outright.
+    """
+    name = model_name.strip().lower()
+    return not any(marker in name for marker in _ANTHROPIC_MODEL_MARKERS)
+
+
 def model_supports_reasoning(model_name: str) -> bool:
     import litellm
 
@@ -848,6 +990,9 @@ def is_known_openai_bare_model(model_name: str) -> bool:
     return bool(entry and entry.get("litellm_provider") == "openai")
 
 
+_ANTHROPIC_MODEL_MARKERS = ("anthropic", "claude", "sonnet", "opus", "haiku")
+
+
 def is_claude_model(model_name: str) -> bool:
     return "claude" in (model_name or "").strip().lower()
 

strix/runtime/session_manager.py

@@ -11,7 +11,7 @@ from typing import TYPE_CHECKING, Any
 import stat
 
 from agents.sandbox.entries import BaseEntry, Dir, File, LocalDir
-from agents.sandbox.manifest import Environment, Manifest
+from agents.sandbox.manifest import Environment, Manifest, SandboxPathGrant
 
 from strix.config import load_settings
 from strix.core.paths import run_dir_for, runtime_state_dir
@@ -69,29 +69,59 @@ def build_bind_mounts(local_sources: list[dict[str, Any]]) -> list[dict[str, Any
     return bind_mounts
 
 
-def _symlink_safe_dir_entry(root: Path) -> Dir:
+def _symlink_safe_dir_entry(
+    root: Path,
+    *,
+    _source_root: Path | None = None,
+    _visited_dirs: set[Path] | None = None,
+) -> Dir:
     """Walk *root* recursively and build a ``Dir`` entry tree.
 
-    Symlinks are resolved to their real targets and copied as regular ``File``
-    entries so that the SDK ``LocalDir`` symlink rejection is bypassed entirely.
-    Dangling symlinks are silently skipped with a warning.
+    Symlinks are resolved only when their real target stays inside *_source_root*.
+    Out-of-tree symlinks, dangling symlinks, and directory symlink loops are
+    skipped with a warning to preserve sandbox containment.
     """
+    if _source_root is None:
+        _source_root = root.resolve()
+    if _visited_dirs is None:
+        _visited_dirs = set()
+
+    real_root = root.resolve()
+    if real_root in _visited_dirs:
+        logger.warning("mount-free: skipping directory loop at %s", root)
+        return Dir(children={})
+    _visited_dirs.add(real_root)
+
     children: dict[str | Path, BaseEntry] = {}
     for item in sorted(root.iterdir()):
         if item.is_symlink():
-            target = item.resolve()
+            try:
+                target = item.resolve()
+            except RuntimeError:
+                logger.warning("mount-free: skipping symlink loop at %s", item)
+                continue
+
             if not target.exists():
                 logger.warning("mount-free: skipping dangling symlink %s", item)
                 continue
+
+            try:
+                target.relative_to(_source_root)
+            except ValueError:
+                logger.warning("mount-free: skipping out-of-tree symlink %s -> %s", item, target)
+                continue
+
             if target.is_dir():
-                children[item.name] = _symlink_safe_dir_entry(target)
+                logger.warning("mount-free: skipping directory symlink %s -> %s", item, target)
             elif target.is_file():
                 try:
                     children[item.name] = File(content=target.read_bytes())
                 except OSError:
                     logger.warning("mount-free: could not read symlink target %s -> %s", item, target)
         elif item.is_dir():
-            children[item.name] = _symlink_safe_dir_entry(item)
+            children[item.name] = _symlink_safe_dir_entry(
+                item, _source_root=_source_root, _visited_dirs=_visited_dirs.copy()
+            )
         elif item.is_file():
             try:
                 children[item.name] = File(content=item.read_bytes())
@@ -120,6 +150,18 @@ def build_manifest_entries(local_sources: list[dict[str, Any]]) -> dict[str | Pa
     return entries
 
 
+def build_manifest_grants(local_sources: list[dict[str, Any]]) -> list[SandboxPathGrant]:
+    grants: list[SandboxPathGrant] = []
+    for src in local_sources:
+        ws_subdir = src.get("workspace_subdir") or ""
+        host_path = src.get("source_path") or ""
+        if not ws_subdir or not host_path:
+            continue
+        resolved = Path(host_path).expanduser().resolve()
+        grants.append(SandboxPathGrant(path=resolved))
+    return grants
+
+
 def _extra_file_rel_path(workspace_path: str) -> str | None:
     """Validate an extra-file target path and return it relative to /workspace.
 
@@ -337,6 +379,7 @@ async def create_or_reuse(
     if use_bind_mounts:
         bind_mounts = build_bind_mounts(local_sources)
         entries: dict[str | Path, BaseEntry] = {}
+        grants: list[SandboxPathGrant] = []
         if extra_files:
             staging_dir = runtime_state_dir(run_dir_for(scan_id)) / "extra_files"
             bind_mounts.extend(
@@ -345,6 +388,7 @@ async def create_or_reuse(
     else:
         bind_mounts = []
         entries = build_manifest_entries(local_sources)
+        grants = build_manifest_grants(local_sources)
         if extra_files:
             entries.update(build_extra_file_entries(extra_files, local_sources))
 
@@ -356,6 +400,7 @@ async def create_or_reuse(
     container_caido_url = f"http://127.0.0.1:{_CONTAINER_CAIDO_PORT}"
     manifest = Manifest(
         entries=entries,
+        extra_path_grants=grants,
         environment=Environment(
             value={
                 "PYTHONUNBUFFERED": "1",

tests/test_mount_free.py

@@ -1,33 +1,78 @@
-"""Tests for mount-free transport fail-closed semantics."""
+"""Tests for mount-free transport fail-closed semantics and symlink containment."""
 
+import os
+from pathlib import Path
 from typing import Any
 
 import pytest
 
-from strix.runtime.backends import register_backend
-from strix.runtime.session_manager import create_or_reuse
+from strix.runtime.backends import _BACKENDS, register_backend
+from strix.runtime.session_manager import _symlink_safe_dir_entry, create_or_reuse
 
 
 async def _dummy_backend(*args: Any, **kwargs: Any) -> tuple[Any, Any]:
     return None, None
 
 
-def test_mount_free_transport_refusal_raises_error(monkeypatch: pytest.MonkeyPatch) -> None:
-    # Register a backend that ONLY supports bind mounts and does NOT support mount-free
-    register_backend(
-        "stub_legacy",
-        _dummy_backend,
-        supports_bind_mounts=True,
-        supports_mount_free=False,
-    )
-    
-    monkeypatch.setenv("STRIX_RUNTIME_BACKEND", "stub_legacy")
-    monkeypatch.setenv("STRIX_REQUIRE_MOUNT_FREE", "1")
+@pytest.mark.asyncio
+async def test_mount_free_transport_refusal_raises_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    try:
+        register_backend(
+            "stub_legacy",
+            _dummy_backend,
+            supports_bind_mounts=True,
+            supports_mount_free=False,
+        )
 
-    # Clear the settings cache so env vars are picked up
-    import strix.config.loader
-    strix.config.loader._cached = None
+        monkeypatch.setenv("STRIX_RUNTIME_BACKEND", "stub_legacy")
+        monkeypatch.setenv("STRIX_REQUIRE_MOUNT_FREE", "1")
 
-    with pytest.raises(RuntimeError, match="Sandbox backend 'stub_legacy' does not support mount-free transport"):
-        import asyncio
-        asyncio.run(create_or_reuse("scan_123", image="dummy_image", local_sources=[]))
+        import strix.config.loader
+        monkeypatch.setattr(strix.config.loader, "_cached", None)
+
+        with pytest.raises(
+            RuntimeError,
+            match="Sandbox backend 'stub_legacy' does not support mount-free transport",
+        ):
+            await create_or_reuse("scan_123", image="dummy_image", local_sources=[])
+    finally:
+        _BACKENDS.pop("stub_legacy", None)
+
+
+def test_symlink_safe_dir_entry_skips_out_of_tree_symlinks(tmp_path: Path) -> None:
+    outside_dir = tmp_path / "outside"
+    outside_dir.mkdir()
+    secret_file = outside_dir / "id_rsa"
+    secret_file.write_bytes(b"SECRET_KEY_DATA")
+
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    (repo_dir / "app.py").write_text("print('hello')")
+
+    # Create symlink pointing outside the source root tree
+    leak_symlink = repo_dir / "leak_key"
+    os.symlink(secret_file, leak_symlink)
+
+    dir_entry = _symlink_safe_dir_entry(repo_dir)
+
+    # Valid files are included, out-of-tree symlinks are skipped!
+    assert "app.py" in dir_entry.children
+    assert "leak_key" not in dir_entry.children
+
+
+def test_symlink_safe_dir_entry_prevents_directory_symlink_loops(tmp_path: Path) -> None:
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    (repo_dir / "main.py").write_text("import sys")
+
+    sub_dir = repo_dir / "sub"
+    sub_dir.mkdir()
+
+    # Create directory symlink loop (sub/up -> repo)
+    os.symlink(repo_dir, sub_dir / "up", target_is_directory=True)
+
+    # Must complete without RecursionError or infinite loop
+    dir_entry = _symlink_safe_dir_entry(repo_dir)
+
+    assert "main.py" in dir_entry.children
+    assert "sub" in dir_entry.children