Merge 297661ffa1 into 717ffc8f4c

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

strix/interface/cli_args.py

@@ -51,6 +51,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",
@@ -269,6 +279,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,

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)

strix/interface/utils.py

@@ -1550,7 +1550,32 @@ 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 = 300.0
+
+
+def clone_repository(
+    repo_url: str,
+    run_name: str,
+    dest_name: str | None = None,
+    timeout: float | None = None,
+) -> 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.
+            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.
+
+    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")
@@ -1570,6 +1595,18 @@ def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None)
     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
@@ -1582,10 +1619,20 @@ def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None)
                 capture_output=True,
                 text=True,
                 check=True,
+                timeout=effective_timeout,
             )
 
         return str(clone_path.absolute())
 
+    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 {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)
         raise ValueError(f"Could not clone repository {repo_url}: {detail}") from e

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/"):

tests/test_clone_repository.py

@@ -0,0 +1,197 @@
+"""Tests for clone_repository in strix.interface.utils."""
+
+from __future__ import annotations
+
+import argparse
+import subprocess
+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,
+)
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+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)
+
+    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_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)
+
+    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.*--git-clone-timeout",
+        ),
+    ):
+        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")
+
+
+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"