From 770275095aa42215bb12fc4b9399c2df1f400a94 Mon Sep 17 00:00:00 2001 From: Manideep Malyala Date: Mon, 24 Aug 2026 00:35:27 +0530 Subject: [PATCH] style: fix pre-commit trailing whitespace and line lengths --- strix/core/inputs.py | 34 ++++++++++++++------- strix/report/state.py | 7 +++-- strix/runtime/backends.py | 11 +++++-- strix/runtime/session_manager.py | 52 ++++++++++++++++++-------------- tests/test_mount_free.py | 26 ++++++++++------ 5 files changed, 82 insertions(+), 48 deletions(-) diff --git a/strix/core/inputs.py b/strix/core/inputs.py index 341c5cdb..98d967d2 100644 --- a/strix/core/inputs.py +++ b/strix/core/inputs.py @@ -8,6 +8,7 @@ from typing import TYPE_CHECKING, Any from agents.model_settings import ModelSettings from openai.types.shared import Reasoning +from strix.config.loader import load_settings from strix.config.models import ( DEFAULT_MODEL_RETRY, OPENROUTER_ATTRIBUTION_HEADERS, @@ -19,9 +20,8 @@ from strix.config.models import ( model_supports_reasoning, request_timeout_extra_args, ) -from strix.config.loader import load_settings -from strix.runtime.backends import backend_supports_bind_mounts from strix.core.sessions import scrub_images_from_items +from strix.runtime.backends import backend_supports_bind_mounts if TYPE_CHECKING: @@ -134,13 +134,19 @@ def build_root_task(scan_config: dict[str, Any]) -> str: elif ttype == "local_code": path = details.get("target_path", "unknown") backend_name = load_settings().runtime.backend - use_bind_mounts = backend_supports_bind_mounts(backend_name) and not load_settings().runtime.require_mount_free - + use_bind_mounts = ( + backend_supports_bind_mounts(backend_name) + and not load_settings().runtime.require_mount_free + ) + if use_bind_mounts: desc = "this is the user's real directory, mounted live and writable" else: - desc = "this is a bounded snapshot of the user's directory, isolated from the live host system" - + desc = ( + "this is a bounded snapshot of the user's directory, " + "isolated from the live host system" + ) + sections["Local Codebases"].append( f"- {path} (available at: {workspace_path}; " f"{desc} — .git/.agents/.codex are read-only)" @@ -163,15 +169,21 @@ def build_root_task(scan_config: dict[str, Any]) -> str: if workspace_mount := scan_config.get("workspace_mount") or "": subdir = scan_config.get("workspace_subdir") or "" workspace_path = f"/workspace/{subdir}" if subdir else "/workspace" - + backend_name = load_settings().runtime.backend - use_bind_mounts = backend_supports_bind_mounts(backend_name) and not load_settings().runtime.require_mount_free - + use_bind_mounts = ( + backend_supports_bind_mounts(backend_name) + and not load_settings().runtime.require_mount_free + ) + if use_bind_mounts: desc = "this is the user's real directory, mounted live and writable" else: - desc = "this is a bounded snapshot of the user's directory, isolated from the live host system" - + desc = ( + "this is a bounded snapshot of the user's directory, " + "isolated from the live host system" + ) + parts.append("\n\nWorking Directory:") parts.append( f"- {workspace_mount} (available at: {workspace_path}; " diff --git a/strix/report/state.py b/strix/report/state.py index 529594e7..46b9f4ea 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -13,7 +13,6 @@ from agents.usage import Usage from strix.config import codex from strix.config.loader import load_settings -from strix.runtime.backends import backend_supports_bind_mounts from strix.core.paths import run_dir_for from strix.report.pricing import resolve_litellm_model from strix.report.sarif import write_sarif @@ -24,6 +23,7 @@ from strix.report.writer import ( write_run_record, write_vulnerabilities, ) +from strix.runtime.backends import backend_supports_bind_mounts from strix.telemetry import posthog, scarf @@ -402,7 +402,10 @@ class ReportState: transport = None else: backend_name = load_settings().runtime.backend - use_bind_mounts = backend_supports_bind_mounts(backend_name) and not load_settings().runtime.require_mount_free + use_bind_mounts = ( + backend_supports_bind_mounts(backend_name) + and not load_settings().runtime.require_mount_free + ) transport = "bind-mount" if use_bind_mounts else "mount-free" self.run_record.update( diff --git a/strix/runtime/backends.py b/strix/runtime/backends.py index d77bb8f8..a5a4b8c4 100644 --- a/strix/runtime/backends.py +++ b/strix/runtime/backends.py @@ -97,13 +97,18 @@ def register_backend( _BIND_MOUNT_BACKENDS.add(name) else: _BIND_MOUNT_BACKENDS.discard(name) - + if supports_mount_free: _MOUNT_FREE_BACKENDS.add(name) else: _MOUNT_FREE_BACKENDS.discard(name) - - logger.info("Registered sandbox backend: %s (bind mounts: %s, mount-free: %s)", name, supports_bind_mounts, supports_mount_free) + + logger.info( + "Registered sandbox backend: %s (bind mounts: %s, mount-free: %s)", + name, + supports_bind_mounts, + supports_mount_free, + ) def backend_supports_bind_mounts(name: str) -> bool: diff --git a/strix/runtime/session_manager.py b/strix/runtime/session_manager.py index 3ea88859..2f83fd1d 100644 --- a/strix/runtime/session_manager.py +++ b/strix/runtime/session_manager.py @@ -79,6 +79,34 @@ def _symlink_safe_dir_entry( Out-of-tree symlinks, dangling symlinks, and directory symlink loops are skipped with a warning to preserve sandbox containment. """ + + def _process_symlink( + item: Path, _source_root: Path, children: dict[str | Path, BaseEntry] + ) -> None: + try: + target = item.resolve() + except RuntimeError: + logger.warning("mount-free: skipping symlink loop at %s", item) + return + + if not target.exists(): + logger.warning("mount-free: skipping dangling symlink %s", item) + return + + try: + target.relative_to(_source_root) + except ValueError: + logger.warning("mount-free: skipping out-of-tree symlink %s -> %s", item, target) + return + + if target.is_dir(): + 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) + if _source_root is None: _source_root = root.resolve() if _visited_dirs is None: @@ -93,29 +121,7 @@ def _symlink_safe_dir_entry( children: dict[str | Path, BaseEntry] = {} for item in sorted(root.iterdir()): if item.is_symlink(): - 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(): - 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) + _process_symlink(item, _source_root, children) elif item.is_dir(): children[item.name] = _symlink_safe_dir_entry( item, _source_root=_source_root, _visited_dirs=_visited_dirs.copy() diff --git a/tests/test_mount_free.py b/tests/test_mount_free.py index 583d4af5..4b7eebac 100644 --- a/tests/test_mount_free.py +++ b/tests/test_mount_free.py @@ -1,17 +1,21 @@ """Tests for mount-free transport fail-closed semantics and symlink containment.""" -import os from pathlib import Path from typing import Any import pytest import strix.config.loader -from strix.runtime.backends import _BACKENDS, _BIND_MOUNT_BACKENDS, _MOUNT_FREE_BACKENDS, register_backend +from strix.runtime.backends import ( + _BACKENDS, + _BIND_MOUNT_BACKENDS, + _MOUNT_FREE_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]: +async def _dummy_backend(*_args: Any, **_kwargs: Any) -> tuple[Any, Any]: return None, None @@ -44,7 +48,9 @@ async def test_mount_free_transport_refusal_raises_error(monkeypatch: pytest.Mon async def test_mount_free_success_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: captured: dict[str, Any] = {} - async def _stub_backend(image: str, manifest: Any, exposed_ports: Any, bind_mounts: Any) -> tuple[Any, Any]: + async def _stub_backend( + image: str, manifest: Any, _exposed_ports: Any, bind_mounts: Any + ) -> tuple[Any, Any]: captured["image"] = image captured["manifest"] = manifest captured["bind_mounts"] = bind_mounts @@ -55,12 +61,12 @@ async def test_mount_free_success_path(monkeypatch: pytest.MonkeyPatch, tmp_path tls = False class FakeSession: - async def resolve_exposed_port(self, port: int) -> FakeEndpoint: + async def resolve_exposed_port(self, _port: int) -> FakeEndpoint: return FakeEndpoint() return None, FakeSession() - async def _fake_bootstrap_caido(*args: Any, **kwargs: Any) -> Any: + async def _fake_bootstrap_caido(*_args: Any, **_kwargs: Any) -> Any: return None try: @@ -81,7 +87,9 @@ async def test_mount_free_success_path(monkeypatch: pytest.MonkeyPatch, tmp_path (source_dir / "app.py").write_text("print('test')") local_sources = [{"workspace_subdir": "repo", "source_path": str(source_dir)}] - bundle = await create_or_reuse("scan_success_123", image="dummy_image", local_sources=local_sources) + bundle = await create_or_reuse( + "scan_success_123", image="dummy_image", local_sources=local_sources + ) assert bundle["session"] is not None assert captured["bind_mounts"] == [] @@ -109,7 +117,7 @@ def test_symlink_safe_dir_entry_skips_out_of_tree_symlinks(tmp_path: Path) -> No # Create symlink pointing outside the source root tree leak_symlink = repo_dir / "leak_key" - os.symlink(secret_file, leak_symlink) + leak_symlink.symlink_to(secret_file) dir_entry = _symlink_safe_dir_entry(repo_dir) @@ -127,7 +135,7 @@ def test_symlink_safe_dir_entry_prevents_directory_symlink_loops(tmp_path: Path) sub_dir.mkdir() # Create directory symlink loop (sub/up -> repo) - os.symlink(repo_dir, sub_dir / "up", target_is_directory=True) + (sub_dir / "up").symlink_to(repo_dir, target_is_directory=True) # Must complete without RecursionError or infinite loop dir_entry = _symlink_safe_dir_entry(repo_dir)