From 6dddbf3a50d16795bfdfc56d6a19fa40316385be Mon Sep 17 00:00:00 2001 From: Aneesh Sharma Date: Sat, 22 Aug 2026 14:19:31 +0530 Subject: [PATCH 1/3] fix(pricing): add noqa PLC0415 for deferred litellm import --- strix/report/pricing.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/strix/report/pricing.py b/strix/report/pricing.py index 57c89959..40fd01a4 100644 --- a/strix/report/pricing.py +++ b/strix/report/pricing.py @@ -10,7 +10,7 @@ from typing import Any, cast def resolve_litellm_model(model: str) -> str | None: """Return a provider-qualified model name that LiteLLM can price.""" try: - import litellm + import litellm # noqa: PLC0415 normalized = model.strip() for prefix in ("litellm/", "any-llm/", "openai/"): From 8e16f28596f0070043e1a3ff9cccdd265dac2c25 Mon Sep 17 00:00:00 2001 From: Aneesh Sharma Date: Sat, 22 Aug 2026 14:19:34 +0530 Subject: [PATCH 2/3] fix(interface): add timeout to git clone during scan setup (#1105) --- strix/interface/utils.py | 33 +++++++++- tests/test_clone_repository.py | 112 +++++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 1 deletion(-) create mode 100644 tests/test_clone_repository.py diff --git a/strix/interface/utils.py b/strix/interface/utils.py index 6789abe0..9cf08df7 100644 --- a/strix/interface/utils.py +++ b/strix/interface/utils.py @@ -1552,7 +1552,30 @@ def stage_api_specs(targets_info: list[dict[str, Any]], run_name: str) -> list[d ] -def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None) -> str: +DEFAULT_GIT_CLONE_TIMEOUT_SECONDS: float = 120.0 + + +def clone_repository( + repo_url: str, + run_name: str, + dest_name: str | None = None, + timeout: float = DEFAULT_GIT_CLONE_TIMEOUT_SECONDS, +) -> str: + """Clone a git repository to a temporary workspace for scanning. + + Args: + repo_url: The URL or path of the git repository to clone. + run_name: The current run identifier used for namespacing temporary files. + dest_name: Optional custom subdirectory/destination name for the clone. + timeout: Maximum time in seconds to wait for the clone operation before timing out. + + Returns: + The absolute path to the cloned repository directory. + + Raises: + ValueError: If git fails to clone, times out, or git is not installed. + FileNotFoundError: If git executable cannot be found in PATH. + """ console = Console() git_executable = shutil.which("git") @@ -1584,10 +1607,18 @@ def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None) capture_output=True, text=True, check=True, + timeout=timeout, ) return str(clone_path.absolute()) + except subprocess.TimeoutExpired as e: + if clone_path.exists(): + shutil.rmtree(clone_path, ignore_errors=True) + raise ValueError( + f"Cloning repository {repo_url} timed out after {int(timeout)}s. " + "Please check network connectivity or clone the repository locally first." + ) from e except subprocess.CalledProcessError as e: detail = e.stderr if hasattr(e, "stderr") and e.stderr else str(e) raise ValueError(f"Could not clone repository {repo_url}: {detail}") from e diff --git a/tests/test_clone_repository.py b/tests/test_clone_repository.py new file mode 100644 index 00000000..8fabd98f --- /dev/null +++ b/tests/test_clone_repository.py @@ -0,0 +1,112 @@ +"""Tests for clone_repository in strix.interface.utils.""" + +from __future__ import annotations + +import subprocess +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import pytest + +from strix.interface.utils import ( + DEFAULT_GIT_CLONE_TIMEOUT_SECONDS, + clone_repository, +) + + +if TYPE_CHECKING: + from pathlib import Path + + +def test_clone_repository_success(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("tempfile.gettempdir", lambda: str(tmp_path)) + monkeypatch.setattr("shutil.which", lambda cmd: "/usr/bin/git" if cmd == "git" else None) + + mock_run = MagicMock() + with patch("subprocess.run", mock_run): + res = clone_repository("https://github.com/example/test-repo.git", "run_123") + + expected_path = tmp_path / "strix_repos" / "run_123" / "test-repo" + assert res == str(expected_path.resolve()) + mock_run.assert_called_once_with( + ["/usr/bin/git", "clone", "https://github.com/example/test-repo.git", str(expected_path)], + capture_output=True, + text=True, + check=True, + timeout=DEFAULT_GIT_CLONE_TIMEOUT_SECONDS, + ) + + +def test_clone_repository_custom_timeout(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("tempfile.gettempdir", lambda: str(tmp_path)) + monkeypatch.setattr("shutil.which", lambda cmd: "/usr/bin/git" if cmd == "git" else None) + + mock_run = MagicMock() + with patch("subprocess.run", mock_run): + res = clone_repository( + "https://github.com/example/test-repo.git", + "run_123", + dest_name="custom_dest", + timeout=45.0, + ) + + expected_path = tmp_path / "strix_repos" / "run_123" / "custom_dest" + assert res == str(expected_path.resolve()) + mock_run.assert_called_once_with( + ["/usr/bin/git", "clone", "https://github.com/example/test-repo.git", str(expected_path)], + capture_output=True, + text=True, + check=True, + timeout=45.0, + ) + + +def test_clone_repository_timeout_expired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("tempfile.gettempdir", lambda: str(tmp_path)) + monkeypatch.setattr("shutil.which", lambda cmd: "/usr/bin/git" if cmd == "git" else None) + + def _mock_timeout(*_args: object, **_kwargs: object) -> None: + clone_dir = tmp_path / "strix_repos" / "run_123" / "slow-repo" + clone_dir.mkdir(parents=True, exist_ok=True) + (clone_dir / "partial_file.txt").write_text("partial", encoding="utf-8") + raise subprocess.TimeoutExpired(cmd="git clone", timeout=30.0) + + with ( + patch("subprocess.run", side_effect=_mock_timeout), + pytest.raises(ValueError, match=r"Cloning repository .* timed out after 30s"), + ): + clone_repository( + "https://github.com/example/slow-repo.git", + "run_123", + timeout=30.0, + ) + + # Check partial clone dir is cleaned up on timeout + clone_dir = tmp_path / "strix_repos" / "run_123" / "slow-repo" + assert not clone_dir.exists() + + +def test_clone_repository_called_process_error( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr("tempfile.gettempdir", lambda: str(tmp_path)) + monkeypatch.setattr("shutil.which", lambda cmd: "/usr/bin/git" if cmd == "git" else None) + + with ( + patch( + "subprocess.run", + side_effect=subprocess.CalledProcessError( + returncode=128, cmd="git clone", stderr="fatal: repository not found" + ), + ), + pytest.raises(ValueError, match=r"fatal: repository not found"), + ): + clone_repository("https://github.com/example/missing.git", "run_123") + + +def test_clone_repository_git_not_found(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("tempfile.gettempdir", lambda: str(tmp_path)) + monkeypatch.setattr("shutil.which", lambda _cmd: None) + + with pytest.raises(FileNotFoundError, match="Git executable not found"): + clone_repository("https://github.com/example/repo.git", "run_123") From 297661ffa110d4cb465dbaff61097ce0c615ada6 Mon Sep 17 00:00:00 2001 From: Aneesh Sharma Date: Sat, 22 Aug 2026 14:45:43 +0530 Subject: [PATCH 3/3] feat(config): expose configurable git clone timeout via CLI and settings --- strix/config/settings.py | 6 +++ strix/interface/cli_args.py | 22 ++++++++ strix/interface/scan_setup.py | 3 +- strix/interface/utils.py | 26 ++++++++-- tests/test_clone_repository.py | 91 ++++++++++++++++++++++++++++++++-- 5 files changed, 139 insertions(+), 9 deletions(-) diff --git a/strix/config/settings.py b/strix/config/settings.py index 42a2c97e..be143516 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -112,6 +112,12 @@ class RuntimeSettings(BaseSettings): backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND") # Max screenshot/image tool outputs kept live per agent context (0 = none). max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES") + # Timeout in seconds for cloning remote git repositories (0 = no timeout). + git_clone_timeout: int = Field( + default=300, + ge=0, + alias="STRIX_GIT_CLONE_TIMEOUT", + ) class TelemetrySettings(BaseSettings): diff --git a/strix/interface/cli_args.py b/strix/interface/cli_args.py index d354106a..a4a6261a 100644 --- a/strix/interface/cli_args.py +++ b/strix/interface/cli_args.py @@ -50,6 +50,16 @@ def _positive_int(value: str) -> int: return parsed +def _non_negative_int(value: str) -> int: + try: + parsed = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"invalid int value: {value!r}") from exc + if parsed < 0: + raise argparse.ArgumentTypeError("must be an integer greater than or equal to 0") + return parsed + + def parse_arguments() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Strix Multi-Agent Cybersecurity Penetration Testing Tool", @@ -244,6 +254,18 @@ Examples: ), ) + parser.add_argument( + "--git-clone-timeout", + dest="git_clone_timeout", + metavar="SECONDS", + type=_non_negative_int, + default=None, + help=( + "Maximum time in seconds to wait when cloning a remote git repository " + "(default: from STRIX_GIT_CLONE_TIMEOUT or 300s, 0 disables timeout)." + ), + ) + parser.add_argument( "--resume", type=str, diff --git a/strix/interface/scan_setup.py b/strix/interface/scan_setup.py index ae7caf2f..aac50465 100644 --- a/strix/interface/scan_setup.py +++ b/strix/interface/scan_setup.py @@ -177,7 +177,8 @@ def prepare_run(args: argparse.Namespace) -> None: if target_info["type"] == "repository": repo_url = target_info["details"]["target_repo"] dest_name = target_info["details"].get("workspace_subdir") - cloned_path = clone_repository(repo_url, args.run_name, dest_name) + timeout = getattr(args, "git_clone_timeout", None) + cloned_path = clone_repository(repo_url, args.run_name, dest_name, timeout=timeout) target_info["details"]["cloned_repo_path"] = cloned_path args.local_sources = collect_local_sources(args.targets_info) diff --git a/strix/interface/utils.py b/strix/interface/utils.py index 9cf08df7..1a4c739a 100644 --- a/strix/interface/utils.py +++ b/strix/interface/utils.py @@ -1552,14 +1552,14 @@ def stage_api_specs(targets_info: list[dict[str, Any]], run_name: str) -> list[d ] -DEFAULT_GIT_CLONE_TIMEOUT_SECONDS: float = 120.0 +DEFAULT_GIT_CLONE_TIMEOUT_SECONDS: float = 300.0 def clone_repository( repo_url: str, run_name: str, dest_name: str | None = None, - timeout: float = DEFAULT_GIT_CLONE_TIMEOUT_SECONDS, + timeout: float | None = None, ) -> str: """Clone a git repository to a temporary workspace for scanning. @@ -1568,6 +1568,8 @@ def clone_repository( run_name: The current run identifier used for namespacing temporary files. dest_name: Optional custom subdirectory/destination name for the clone. timeout: Maximum time in seconds to wait for the clone operation before timing out. + If None, the timeout is loaded from settings (STRIX_GIT_CLONE_TIMEOUT, defaulting + to 300s). Set to 0 to disable the timeout. Returns: The absolute path to the cloned repository directory. @@ -1595,6 +1597,18 @@ def clone_repository( if clone_path.exists(): shutil.rmtree(clone_path) + effective_timeout: float | None + if timeout is None: + try: + cfg_timeout = load_settings().runtime.git_clone_timeout + effective_timeout = float(cfg_timeout) if cfg_timeout > 0 else None + except Exception: + effective_timeout = DEFAULT_GIT_CLONE_TIMEOUT_SECONDS + elif timeout <= 0: + effective_timeout = None + else: + effective_timeout = float(timeout) + try: with console.status(f"[bold cyan]Cloning repository {repo_url}...", spinner="dots"): subprocess.run( # noqa: S603 @@ -1607,7 +1621,7 @@ def clone_repository( capture_output=True, text=True, check=True, - timeout=timeout, + timeout=effective_timeout, ) return str(clone_path.absolute()) @@ -1615,9 +1629,11 @@ def clone_repository( except subprocess.TimeoutExpired as e: if clone_path.exists(): shutil.rmtree(clone_path, ignore_errors=True) + timeout_str = f"{int(effective_timeout)}s" if effective_timeout else "configured limit" raise ValueError( - f"Cloning repository {repo_url} timed out after {int(timeout)}s. " - "Please check network connectivity or clone the repository locally first." + f"Cloning repository {repo_url} timed out after {timeout_str}. " + "You can increase or disable the limit with --git-clone-timeout or " + "STRIX_GIT_CLONE_TIMEOUT, or clone the repository locally first." ) from e except subprocess.CalledProcessError as e: detail = e.stderr if hasattr(e, "stderr") and e.stderr else str(e) diff --git a/tests/test_clone_repository.py b/tests/test_clone_repository.py index 8fabd98f..e58d3ea0 100644 --- a/tests/test_clone_repository.py +++ b/tests/test_clone_repository.py @@ -2,12 +2,15 @@ from __future__ import annotations +import argparse import subprocess -from typing import TYPE_CHECKING +from types import SimpleNamespace +from typing import TYPE_CHECKING, Any from unittest.mock import MagicMock, patch import pytest +from strix.interface.scan_setup import prepare_run from strix.interface.utils import ( DEFAULT_GIT_CLONE_TIMEOUT_SECONDS, clone_repository, @@ -18,7 +21,9 @@ if TYPE_CHECKING: from pathlib import Path -def test_clone_repository_success(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: +def test_clone_repository_default_settings_timeout( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: monkeypatch.setattr("tempfile.gettempdir", lambda: str(tmp_path)) monkeypatch.setattr("shutil.which", lambda cmd: "/usr/bin/git" if cmd == "git" else None) @@ -37,6 +42,48 @@ def test_clone_repository_success(tmp_path: Path, monkeypatch: pytest.MonkeyPatc ) +def test_clone_repository_env_var_timeout(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("tempfile.gettempdir", lambda: str(tmp_path)) + monkeypatch.setattr("shutil.which", lambda cmd: "/usr/bin/git" if cmd == "git" else None) + monkeypatch.setattr( + "strix.interface.utils.load_settings", + lambda: SimpleNamespace(runtime=SimpleNamespace(git_clone_timeout=600)), + ) + + mock_run = MagicMock() + with patch("subprocess.run", mock_run): + res = clone_repository("https://github.com/example/test-repo.git", "run_123") + + expected_path = tmp_path / "strix_repos" / "run_123" / "test-repo" + assert res == str(expected_path.resolve()) + mock_run.assert_called_once_with( + ["/usr/bin/git", "clone", "https://github.com/example/test-repo.git", str(expected_path)], + capture_output=True, + text=True, + check=True, + timeout=600.0, + ) + + +def test_clone_repository_disabled_timeout(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("tempfile.gettempdir", lambda: str(tmp_path)) + monkeypatch.setattr("shutil.which", lambda cmd: "/usr/bin/git" if cmd == "git" else None) + + mock_run = MagicMock() + with patch("subprocess.run", mock_run): + res = clone_repository("https://github.com/example/test-repo.git", "run_123", timeout=0) + + expected_path = tmp_path / "strix_repos" / "run_123" / "test-repo" + assert res == str(expected_path.resolve()) + mock_run.assert_called_once_with( + ["/usr/bin/git", "clone", "https://github.com/example/test-repo.git", str(expected_path)], + capture_output=True, + text=True, + check=True, + timeout=None, + ) + + def test_clone_repository_custom_timeout(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("tempfile.gettempdir", lambda: str(tmp_path)) monkeypatch.setattr("shutil.which", lambda cmd: "/usr/bin/git" if cmd == "git" else None) @@ -73,7 +120,10 @@ def test_clone_repository_timeout_expired(tmp_path: Path, monkeypatch: pytest.Mo with ( patch("subprocess.run", side_effect=_mock_timeout), - pytest.raises(ValueError, match=r"Cloning repository .* timed out after 30s"), + pytest.raises( + ValueError, + match=r"Cloning repository .* timed out after 30s.*--git-clone-timeout", + ), ): clone_repository( "https://github.com/example/slow-repo.git", @@ -110,3 +160,38 @@ def test_clone_repository_git_not_found(tmp_path: Path, monkeypatch: pytest.Monk with pytest.raises(FileNotFoundError, match="Git executable not found"): clone_repository("https://github.com/example/repo.git", "run_123") + + +def test_prepare_run_passes_git_clone_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + target_info: dict[str, Any] = { + "type": "repository", + "details": {"target_repo": "https://github.com/example/repo.git"}, + } + args = argparse.Namespace( + resume=None, + targets_info=[target_info], + run_name=None, + git_clone_timeout=450, + scope_mode="auto", + diff_base=None, + non_interactive=True, + instruction=None, + ) + + mock_clone = MagicMock(return_value="/cloned/path") + monkeypatch.setattr("strix.interface.scan_setup.clone_repository", mock_clone) + monkeypatch.setattr("strix.interface.scan_setup.collect_local_sources", lambda _t: []) + monkeypatch.setattr("strix.interface.scan_setup.stage_api_specs", lambda _t, _r: []) + monkeypatch.setattr( + "strix.interface.scan_setup.resolve_diff_scope_context", + lambda **_kwargs: SimpleNamespace(metadata={"active": False}, instruction_block=None), + ) + monkeypatch.setattr("strix.interface.scan_setup.attach_workspace_mount", lambda _a: None) + monkeypatch.setattr("strix.interface.scan_setup._persist_run_record", lambda _a: None) + + prepare_run(args) + + mock_clone.assert_called_once_with( + "https://github.com/example/repo.git", args.run_name, None, timeout=450 + ) + assert target_info["details"]["cloned_repo_path"] == "/cloned/path"