diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx
index f1542b75..ef2d07eb 100644
--- a/docs/advanced/configuration.mdx
+++ b/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
+
+ 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.
+
+
Maximum execution time in seconds for sandbox operations.
diff --git a/strix/config/models.py b/strix/config/models.py
index eb2a2c5c..1c14d2b0 100644
--- a/strix/config/models.py
+++ b/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()
diff --git a/strix/runtime/session_manager.py b/strix/runtime/session_manager.py
index 2ef8f488..16b4d7ab 100644
--- a/strix/runtime/session_manager.py
+++ b/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",
diff --git a/tests/test_mount_free.py b/tests/test_mount_free.py
index 1f33935f..1bb87601 100644
--- a/tests/test_mount_free.py
+++ b/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