Merge ed92a726e5 into cb581bf93d

strix/interface/utils.py

@@ -1136,6 +1136,9 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]:  # noqa: PLR09
 
     target = target.strip()
 
+    if target.startswith("-"):
+        raise ValueError(f"Invalid target: {target}. Targets cannot start with a dash.")
+
     if target.startswith("git@"):
         return "repository", {"target_repo": target}
 
@@ -1254,6 +1257,7 @@ def read_target_list_file(path_str: str) -> list[str]:
 
 def sanitize_name(name: str) -> str:
     sanitized = re.sub(r"[^A-Za-z0-9._-]", "-", name.strip())
+    sanitized = sanitized.lstrip(".-")
     return sanitized or "target"
 
 
@@ -1282,7 +1286,7 @@ def derive_local_base_name(path_str: str) -> str:
 
 
 def assign_workspace_subdirs(targets_info: list[dict[str, Any]]) -> None:
-    name_counts: dict[str, int] = {}
+    assigned_names: set[str] = set()
 
     for target in targets_info:
         target_type = target["type"]
@@ -1297,12 +1301,14 @@ def assign_workspace_subdirs(targets_info: list[dict[str, Any]]) -> None:
         if base_name is None:
             continue
 
-        count = name_counts.get(base_name, 0) + 1
-        name_counts[base_name] = count
+        candidate = base_name
+        count = 2
+        while candidate in assigned_names:
+            candidate = f"{base_name}-{count}"
+            count += 1
 
-        workspace_subdir = base_name if count == 1 else f"{base_name}-{count}"
-
-        details["workspace_subdir"] = workspace_subdir
+        assigned_names.add(candidate)
+        details["workspace_subdir"] = candidate
 
 
 def is_whitebox_scan(targets_info: list[dict[str, Any]]) -> bool:
@@ -1555,6 +1561,11 @@ 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:
     console = Console()
 
+    if not repo_url or repo_url.strip().startswith("-"):
+        raise ValueError(
+            f"Invalid repository URL: {repo_url}. Repository URL cannot start with a dash."
+        )
+
     git_executable = shutil.which("git")
     if git_executable is None:
         raise FileNotFoundError("Git executable not found in PATH")
@@ -1567,8 +1578,23 @@ def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None)
     else:
         repo_name = Path(repo_url).stem if repo_url.endswith(".git") else Path(repo_url).name
 
+    if (
+        not repo_name
+        or repo_name in (".", "..")
+        or "/" in repo_name
+        or "\\" in repo_name
+        or repo_name.startswith("-")
+    ):
+        raise ValueError(f"Invalid repository destination name: {repo_name}")
+
     clone_path = temp_dir / repo_name
 
+    if (
+        not clone_path.resolve().is_relative_to(temp_dir.resolve())
+        or clone_path.resolve() == temp_dir.resolve()
+    ):
+        raise ValueError(f"Invalid clone path outside temporary directory: {clone_path}")
+
     if clone_path.exists():
         shutil.rmtree(clone_path)
 
@@ -1578,6 +1604,7 @@ def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None)
                 [
                     git_executable,
                     "clone",
+                    "--",
                     repo_url,
                     str(clone_path),
                 ],

tests/test_clone_repository.py

@@ -0,0 +1,151 @@
+"""Tests for clone_repository, infer_target_type, and sanitization in strix.interface.utils."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from strix.interface.utils import (
+    assign_workspace_subdirs,
+    clone_repository,
+    derive_repo_base_name,
+    infer_target_type,
+    sanitize_name,
+)
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def test_clone_repository_includes_double_dash_separator(
+    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,
+    )
+
+
+def test_clone_repository_rejects_leading_dash_url() -> None:
+    with pytest.raises(ValueError, match="Repository URL cannot start with a dash"):
+        clone_repository("-u/upload-pack.git", "run_123")
+
+    with pytest.raises(ValueError, match="Repository URL cannot start with a dash"):
+        clone_repository("--upload-pack=touch /tmp/pwn", "run_123")
+
+
+def test_clone_repository_rejects_invalid_dest_names(
+    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 pytest.raises(ValueError, match="Invalid repository destination name"):
+        clone_repository("https://example.com/repo.git", "run_123", dest_name="..")
+
+    with pytest.raises(ValueError, match="Invalid repository destination name"):
+        clone_repository("https://example.com/repo.git", "run_123", dest_name=".")
+
+    with pytest.raises(ValueError, match="Invalid repository destination name"):
+        clone_repository("https://example.com/repo.git", "run_123", dest_name="-dashed")
+
+    with pytest.raises(ValueError, match="Invalid repository destination name"):
+        clone_repository("https://example.com/repo.git", "run_123", dest_name="sub/dir")
+
+
+def test_infer_target_type_rejects_leading_dash() -> None:
+    with pytest.raises(ValueError, match="Targets cannot start with a dash"):
+        infer_target_type("-u/upload-pack.git")
+
+    with pytest.raises(ValueError, match="Targets cannot start with a dash"):
+        infer_target_type("--target-flag")
+
+
+def test_sanitize_name_strips_leading_dots_and_dashes() -> None:
+    assert sanitize_name("..") == "target"
+    assert sanitize_name(".") == "target"
+    assert sanitize_name("-dashed") == "dashed"
+    assert sanitize_name(".hidden") == "hidden"
+    assert sanitize_name("normal_repo-1") == "normal_repo-1"
+
+
+def test_derive_repo_base_name_handles_dot_git() -> None:
+    assert derive_repo_base_name("..git") == "target"
+    assert derive_repo_base_name("https://github.com/foo/bar.git") == "bar"
+    assert derive_repo_base_name("git@github.com:foo/bar.git") == "bar"
+
+
+def test_clone_repository_rejects_symlink_pointing_outside_temp_dir(
+    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)
+
+    outside_dir = tmp_path / "sensitive_outside_directory"
+    outside_dir.mkdir(parents=True)
+    canary_file = outside_dir / "canary.txt"
+    canary_file.write_text("critical data", encoding="utf-8")
+
+    run_temp_dir = tmp_path / "strix_repos" / "run_123"
+    run_temp_dir.mkdir(parents=True)
+    symlink_dest = run_temp_dir / "external_symlink"
+    symlink_dest.symlink_to(outside_dir)
+
+    mock_run = MagicMock()
+    with (
+        patch("subprocess.run", mock_run),
+        pytest.raises(ValueError, match="Invalid clone path outside temporary directory"),
+    ):
+        clone_repository(
+            "https://example.com/repo.git",
+            "run_123",
+            dest_name="external_symlink",
+        )
+
+    # Verify external directory and canary were not touched by shutil.rmtree
+    assert outside_dir.exists()
+    assert canary_file.exists()
+    assert canary_file.read_text(encoding="utf-8") == "critical data"
+    mock_run.assert_not_called()
+
+
+def test_assign_workspace_subdirs_avoids_collisions_with_generated_suffixes() -> None:
+    targets_info: list[dict[str, Any]] = [
+        {"type": "repository", "details": {"target_repo": "https://example.com/foo.git"}},
+        {"type": "repository", "details": {"target_repo": "https://example.com/.foo.git"}},
+        {"type": "repository", "details": {"target_repo": "https://example.com/foo-2.git"}},
+    ]
+    assign_workspace_subdirs(targets_info)
+    subdirs = [t["details"]["workspace_subdir"] for t in targets_info]
+    assert len(subdirs) == len(set(subdirs)), f"Collisions found: {subdirs}"
+    assert subdirs == ["foo", "foo-2", "foo-2-2"]
+
+    # Also verify repeated names increment cleanly
+    repeated_targets: list[dict[str, Any]] = [
+        {"type": "repository", "details": {"target_repo": "https://example.com/bar.git"}},
+        {"type": "repository", "details": {"target_repo": "https://example.com/bar.git"}},
+        {"type": "repository", "details": {"target_repo": "https://example.com/bar.git"}},
+    ]
+    assign_workspace_subdirs(repeated_targets)
+    repeated_subdirs = [t["details"]["workspace_subdir"] for t in repeated_targets]
+    assert repeated_subdirs == ["bar", "bar-2", "bar-3"]