feat(config): expose configurable git clone timeout via CLI and settings

This commit is contained in:
Aneesh Sharma 2026-08-22 14:45:43 +05:30
parent 8e16f28596
commit 297661ffa1
5 changed files with 139 additions and 9 deletions

View file

@ -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):

View file

@ -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,

View file

@ -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)

View file

@ -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)

View file

@ -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"