mirror of
https://github.com/usestrix/strix.git
synced 2026-08-20 02:23:35 +02:00
Support large target repos with with bind-mount option. (#577)
* fix: resolve pre-commit check failures - Change RuntimeError to TypeError for type validation in report/writer.py - Update pyupgrade to v3.21.2 for Python 3.14 compatibility * chore: add pytest test infrastructure Mirror the layout introduced on feature/438-token_budget: pytest + pytest-asyncio dev deps, asyncio_mode auto, a tests.* mypy override, and pytest in the mypy pre-commit hook deps so the tests/ package type-checks. * feat: add --mount and large-target pre-flight for local repos (#492) Large local targets were copied into the sandbox file-by-file via the SDK LocalDir entry, which stalls on big repos and could leave /workspace empty. - --mount <path> bind-mounts a host directory read-only at /workspace/<subdir> instead of copying it, bypassing the per-file stream. - A size pre-flight (STRIX_MAX_LOCAL_COPY_MB, default 1024) fails fast with a clear message suggesting --mount when a non-mounted local target is too big. * fix: reject empty --mount paths An empty or whitespace-only --mount value resolves to the current working directory and would silently bind-mount it into the sandbox. Reject it. * fix: dedupe local targets so a dir is never both copied and mounted If the same directory is passed via --target and --mount (or as duplicate values), it previously produced two targets — copied AND bind-mounted, and the copied one could trip the size pre-flight. Dedupe by resolved path, preferring the bind mount. * fix: treat non-positive STRIX_MAX_LOCAL_COPY_MB as disabled Previously a value of 0 (or negative) made every local target count as oversized, aborting all local scans. Now <= 0 disables the pre-flight. * fix: log unreadable subtrees during size pre-flight os.walk silently swallowed directory-listing errors, so a permission-denied subtree could make a large repo under-count and slip past the pre-flight. Surface such omissions via an onerror warning. * docs: document --mount and STRIX_MAX_LOCAL_COPY_MB Add CLI reference + example for --mount, document the size pre-flight env var, note the read-only-is-not-a-hard-boundary caveat and that remote repos are not size-checked, and clarify the backends docstring on when bind mounts apply. * Update strix/interface/main.py * Update strix/runtime/docker_client.py ---------
This commit is contained in:
+121
-2
@@ -1,5 +1,6 @@
|
||||
import ipaddress
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
@@ -23,6 +24,9 @@ from rich.text import Text
|
||||
from strix.config import load_settings
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def get_severity_color(severity: str) -> str:
|
||||
severity_colors = {
|
||||
"critical": "#dc2626",
|
||||
@@ -1185,8 +1189,8 @@ def is_whitebox_scan(targets_info: list[dict[str, Any]]) -> bool:
|
||||
return any(t.get("type") == "local_code" for t in targets_info or [])
|
||||
|
||||
|
||||
def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str, str]]:
|
||||
local_sources: list[dict[str, str]] = []
|
||||
def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
local_sources: list[dict[str, Any]] = []
|
||||
|
||||
for target_info in targets_info:
|
||||
details = target_info["details"]
|
||||
@@ -1197,6 +1201,7 @@ def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str,
|
||||
{
|
||||
"source_path": details["target_path"],
|
||||
"workspace_subdir": workspace_subdir,
|
||||
"mount": bool(details.get("mount", False)),
|
||||
}
|
||||
)
|
||||
|
||||
@@ -1205,12 +1210,126 @@ def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str,
|
||||
{
|
||||
"source_path": details["cloned_repo_path"],
|
||||
"workspace_subdir": workspace_subdir,
|
||||
"mount": False,
|
||||
}
|
||||
)
|
||||
|
||||
return local_sources
|
||||
|
||||
|
||||
def directory_size_bytes(path: Path) -> int:
|
||||
"""Total size in bytes of regular files under ``path`` (symlinks not followed).
|
||||
|
||||
Best-effort: files that disappear or can't be stat'd mid-walk are skipped.
|
||||
Used as a cheap (stat-only) pre-flight to estimate the cost of streaming a
|
||||
local target into the sandbox before we actually try to copy it.
|
||||
|
||||
Directories that can't be listed (e.g. permission denied) are logged and
|
||||
skipped rather than silently dropped — so an under-count is at least
|
||||
visible — but the returned total then excludes their contents.
|
||||
"""
|
||||
|
||||
def _on_walk_error(error: OSError) -> None:
|
||||
logger.warning("Could not read %s while measuring size: %s", error.filename, error)
|
||||
|
||||
total = 0
|
||||
for root, _dirs, files in os.walk(path, followlinks=False, onerror=_on_walk_error):
|
||||
for name in files:
|
||||
file_path = os.path.join(root, name) # noqa: PTH118
|
||||
try:
|
||||
if os.path.islink(file_path): # noqa: PTH114
|
||||
continue
|
||||
total += os.path.getsize(file_path) # noqa: PTH202
|
||||
except OSError:
|
||||
continue
|
||||
return total
|
||||
|
||||
|
||||
def find_oversized_local_targets(
|
||||
targets_info: list[dict[str, Any]], max_bytes: int
|
||||
) -> list[tuple[str, int]]:
|
||||
"""Return ``(path, size_bytes)`` for non-mounted local targets over ``max_bytes``.
|
||||
|
||||
Mounted targets are bind-mounted rather than copied, so their size is
|
||||
irrelevant and they are excluded. A ``max_bytes`` of zero or less disables
|
||||
the check entirely (returns no targets).
|
||||
"""
|
||||
if max_bytes <= 0:
|
||||
return []
|
||||
oversized: list[tuple[str, int]] = []
|
||||
for target in targets_info:
|
||||
if target.get("type") != "local_code":
|
||||
continue
|
||||
details = target.get("details") or {}
|
||||
if details.get("mount"):
|
||||
continue
|
||||
target_path = details.get("target_path")
|
||||
if not target_path:
|
||||
continue
|
||||
size = directory_size_bytes(Path(target_path))
|
||||
if size > max_bytes:
|
||||
oversized.append((target_path, size))
|
||||
return oversized
|
||||
|
||||
|
||||
def build_mount_targets_info(mount_paths: list[str]) -> list[dict[str, Any]]:
|
||||
"""Build ``targets_info`` entries for ``--mount`` directories.
|
||||
|
||||
Each path must be an existing local directory; it is bind-mounted into the
|
||||
sandbox (read-only) instead of being copied file-by-file. Raises
|
||||
``ValueError`` for an empty path, or one that does not exist or is not a
|
||||
directory.
|
||||
"""
|
||||
targets_info: list[dict[str, Any]] = []
|
||||
for raw in mount_paths:
|
||||
if not raw or not raw.strip():
|
||||
raise ValueError("--mount path must not be empty.")
|
||||
path = Path(raw).expanduser()
|
||||
try:
|
||||
resolved = path.resolve()
|
||||
is_dir = resolved.is_dir()
|
||||
except (OSError, RuntimeError) as e:
|
||||
raise ValueError(f"Invalid mount path '{raw}': {e!s}") from e
|
||||
if not is_dir:
|
||||
raise ValueError(
|
||||
f"Mount path '{raw}' is not an existing directory. "
|
||||
"--mount requires a path to a local directory."
|
||||
)
|
||||
targets_info.append(
|
||||
{
|
||||
"type": "local_code",
|
||||
"details": {"target_path": str(resolved), "mount": True},
|
||||
"original": str(resolved),
|
||||
}
|
||||
)
|
||||
return targets_info
|
||||
|
||||
|
||||
def dedupe_local_targets(targets_info: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""Collapse local_code targets that resolve to the same path.
|
||||
|
||||
When a directory is supplied both as a copied ``--target`` and via
|
||||
``--mount`` (or as duplicate values of either), keep one entry and prefer
|
||||
the bind-mounted one — so the same tree is never both streamed in and
|
||||
mounted. Order is preserved; non-local targets pass through untouched.
|
||||
"""
|
||||
result: list[dict[str, Any]] = []
|
||||
index_by_path: dict[str, int] = {}
|
||||
for target in targets_info:
|
||||
details = target.get("details") or {}
|
||||
path = details.get("target_path")
|
||||
if target.get("type") != "local_code" or not path:
|
||||
result.append(target)
|
||||
continue
|
||||
existing = index_by_path.get(path)
|
||||
if existing is None:
|
||||
index_by_path[path] = len(result)
|
||||
result.append(target)
|
||||
elif details.get("mount") and not (result[existing].get("details") or {}).get("mount"):
|
||||
result[existing] = target # bind mount supersedes the copied entry
|
||||
return result
|
||||
|
||||
|
||||
def _is_localhost_host(host: str) -> bool:
|
||||
host_lower = host.lower().strip("[]")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user