fix(interface): add timeout to git clone during scan setup (#1105)

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

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