38 clone repo and list files

app/scanner/clone.py

@@ -0,0 +1,123 @@
+import os
+import shutil
+import subprocess
+import tempfile
+from urllib.parse import urlparse
+
+from app.scanner.exceptions import RepoCloneError, RepoTooLargeError
+
+SCAN_TEMP_DIR = ""
+
+MAX_REPO_SIZE_MB = 500
+GIT_TIMEOUT_SECONDS = 30
+
+ALLOWED_HOSTS = ("github.com", "gitlab.com")
+
+
+def _check_repo_size(path: str) -> int:
+    total = 0
+
+    for dirpath, dirnames, filenames in os.walk(path, followlinks=False):
+        for name in filenames:
+            file_path = os.path.join(dirpath, name)
+            if os.path.islink(file_path):
+                continue
+            try:
+                total += os.path.getsize(file_path)
+            except OSError:
+                continue
+
+    return total
+
+
+def _run_git(args: list[str], cwd: str, timeout: int) -> None:
+    try:
+        subprocess.run(
+            ["git", *args],
+            check=True,
+            cwd=cwd,
+            timeout=timeout,
+            capture_output=True,
+            text=True,
+        )
+    except subprocess.CalledProcessError as e:
+        stderr = (e.stderr or "").strip()
+        raise RepoCloneError(f"git {' '.join(args)} failed: {stderr or e}") from e
+    except subprocess.TimeoutExpired as e:
+        raise RepoCloneError(f"git {' '.join(args)} timed out after {timeout}s") from e
+
+
+def _is_allowed_url(url: str) -> bool:
+    if not url:
+        return False
+
+    parsed = urlparse(url)
+    if parsed.scheme in ("http", "https"):
+        host = parsed.hostname
+
+        return host is not None and host.lower() in ALLOWED_HOSTS
+
+    if url.startswith("git@"):
+        rest = url[len("git@") :]
+        host, sep, path = rest.partition(":")
+
+        return bool(sep) and bool(host) and bool(path) and host.lower() in ALLOWED_HOSTS
+
+    return False
+
+
+def clone_repo(url: str, commit_sha: str | None = None) -> tuple[str, str]:
+    if not _is_allowed_url(url):
+        raise RepoCloneError("URL не разрешён (нужен http(s):// или git@host:...)")
+
+    ref: str = commit_sha or "HEAD"
+    temp_folder_path: str = tempfile.mkdtemp(dir=SCAN_TEMP_DIR or None, prefix="scan_")
+
+    try:
+        _run_git(["init"], temp_folder_path, GIT_TIMEOUT_SECONDS)
+        _run_git(
+            ["remote", "add", "origin", url], temp_folder_path, GIT_TIMEOUT_SECONDS
+        )
+        _run_git(
+            ["fetch", "--depth=1", "origin", ref], temp_folder_path, GIT_TIMEOUT_SECONDS
+        )
+        _run_git(["checkout", "FETCH_HEAD"], temp_folder_path, GIT_TIMEOUT_SECONDS)
+
+        resolved_sha = _get_head_sha(temp_folder_path)
+
+        size_bytes = _check_repo_size(temp_folder_path)
+        max_bytes = MAX_REPO_SIZE_MB * 1024 * 1024
+
+        if size_bytes > max_bytes:
+            raise RepoTooLargeError(
+                f"Размер репозитория {size_bytes / (1024 * 1024):.1f}MB "
+                f"превышает лимит {MAX_REPO_SIZE_MB}MB"
+            )
+    except (RepoCloneError, RepoTooLargeError):
+        shutil.rmtree(temp_folder_path, ignore_errors=True)
+        raise
+    except Exception as e:
+        shutil.rmtree(temp_folder_path, ignore_errors=True)
+        raise RepoCloneError(f"Не удалось клонировать {url}: {e}") from e
+
+    return temp_folder_path, resolved_sha
+
+
+def _get_head_sha(repo_path: str) -> str:
+    try:
+        result = subprocess.run(
+            ["git", "rev-parse", "HEAD"],
+            check=True,
+            cwd=repo_path,
+            timeout=GIT_TIMEOUT_SECONDS,
+            capture_output=True,
+            text=True,
+        )
+    except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
+        raise RepoCloneError(f"Не удалось получить хэш коммита: {e}") from e
+
+    return result.stdout.strip()
+
+
+def cleanup(repo_path: str) -> None:
+    shutil.rmtree(repo_path, ignore_errors=True)

app/scanner/exceptions.py

@@ -0,0 +1,6 @@
+class RepoCloneError(Exception):
+    pass
+
+
+class RepoTooLargeError(Exception):
+    pass

app/scanner/files.py

@@ -0,0 +1,21 @@
+import os
+
+
+def list_files(repo_path: str) -> list[str]:
+    files: list[str] = []
+
+    for dirpath, dirnames, filenames in os.walk(repo_path, followlinks=False):
+        if ".git" in dirnames:
+            dirnames.remove(".git")
+
+        for name in filenames:
+            file_path: str = os.path.join(dirpath, name)
+
+            if os.path.islink(file_path):
+                continue
+
+            rel_path: str = os.path.relpath(file_path, repo_path)
+            files.append(rel_path.replace(os.sep, "/"))
+
+    files.sort()
+    return files

tests/test_clone.py

@@ -0,0 +1,86 @@
+import os
+
+import pytest
+
+import app.scanner.clone as clone_module
+from app.scanner.clone import _is_allowed_url, cleanup, clone_repo
+from app.scanner.exceptions import RepoCloneError, RepoTooLargeError
+
+
+def test_clone_repo_real_github():
+    path, sha = clone_repo(url="https://github.com/mshqq/mshqq.git")
+    try:
+        assert os.path.isdir(path)
+        assert sha
+        assert len(sha) == 40
+    finally:
+        cleanup(path)
+
+
+def test_clone_repo_specific_commit():
+    commit_sha = "bce38ad98aa578b9603d84ef74505c60ceb287ef"
+
+    path, sha = clone_repo(
+        url="https://github.com/mshqq/mshqq.git", commit_sha=commit_sha
+    )
+    try:
+        assert os.path.isdir(path)
+        assert sha == commit_sha
+    finally:
+        cleanup(path)
+
+
+def test_clone_nonexist_repo_raises():
+    with pytest.raises(RepoCloneError):
+        clone_repo(url="https://github.com/nonexist/nonexist-repo.git")
+
+
+@pytest.mark.parametrize(
+    "url, status",
+    [
+        ("https://github.com/owner/repo.git", True),
+        ("http://github.com/owner/repo.git", True),
+        ("https://gitlab.com/owner/repo.git", True),
+        ("[email protected]:owner/repo.git", True),
+        ("[email protected]:owner/repo", True),
+        ("file:///etc/passwd", False),
+        ("local/path/to/repo", False),
+        ("../relative/path", False),
+        ("https://badurl.com/owner/repo", False),
+        ("[email protected]:owner/repo", False),
+        ("[email protected]", False),
+        ("", False),
+        ("https://github.com.badurl.com", False),
+        ("HTTPS://GITHUB.COM/owner/repo.git", True),
+        ("[email protected]:", False),
+    ],
+)
+def test_is_allowed_url(url, status):
+    assert _is_allowed_url(url) is status
+
+
+def test_cleanup_removes_directory(tmp_path):
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    (repo_dir / "file.txt").write_text("data")
+
+    cleanup(str(repo_dir))
+
+    assert not os.path.isdir(repo_dir)
+
+
+def test_cleanup_nonexistent_path_does_not_raise(tmp_path):
+    missing_dir = tmp_path / "missing"
+
+    cleanup(str(missing_dir))
+
+
+def test_clone_repo_too_large():
+    original_max = clone_module.MAX_REPO_SIZE_MB
+    clone_module.MAX_REPO_SIZE_MB = 0
+
+    try:
+        with pytest.raises(RepoTooLargeError):
+            clone_repo(url="https://github.com/mshqq/mshqq.git")
+    finally:
+        clone_module.MAX_REPO_SIZE_MB = original_max

tests/test_files.py

@@ -0,0 +1,75 @@
+from app.scanner.files import list_files
+
+
+def test_files(tmp_path):
+    assert list_files(str(tmp_path)) == []
+
+
+def test_single_file_in_root(tmp_path):
+    (tmp_path / "a.txt").write_text("X")
+
+    assert list_files(str(tmp_path)) == ["a.txt"]
+
+
+def test_multiple_file_in_root(tmp_path):
+    (tmp_path / "a.txt").write_text("X")
+    (tmp_path / "b.txt").write_text("X")
+    (tmp_path / "c.txt").write_text("X")
+
+    assert list_files(str(tmp_path)) == ["a.txt", "b.txt", "c.txt"]
+
+
+def test_nested_dir(tmp_path):
+    (tmp_path / "app").mkdir()
+    (tmp_path / "app" / "models").mkdir()
+    (tmp_path / "app" / "models" / "user.py").write_text("X")
+    (tmp_path / "app" / "config.py").write_text("X")
+    (tmp_path / ".env").write_text("X")
+
+    result: list[str] = list_files(str(tmp_path))
+
+    assert result == sorted(
+        [
+            ".env",
+            "app/models/user.py",
+            "app/config.py",
+        ]
+    )
+
+
+def test_git_dir_excluded(tmp_path):
+    (tmp_path / ".git").mkdir()
+    (tmp_path / ".git" / "config").write_text("X")
+    (tmp_path / "file.txt").write_text("X")
+
+    result: list[str] = sorted(list_files(str(tmp_path)))
+
+    assert result == ["file.txt"]
+
+
+def test_symlink(tmp_path):
+    target_dir = tmp_path / "target"
+    target_dir.mkdir()
+    (target_dir / "inner.txt").write_text("x")
+
+    link_dir = tmp_path / "link_dir"
+    link_dir.symlink_to(target_dir, target_is_directory=True)
+
+    result: list[str] = list_files(str(tmp_path))
+
+    assert result == ["target/inner.txt"]
+
+
+def test_symlink_to_file_outside_repo(tmp_path, tmp_path_factory):
+    outside = tmp_path_factory.mktemp("outside") / "secret.txt"
+    outside.write_text("X")
+
+    (tmp_path / "link.txt").symlink_to(outside)
+    (tmp_path / "real.txt").write_text("X")
+
+    assert list_files(str(tmp_path)) == ["real.txt"]
+
+
+def test_nonexist_path(tmp_path):
+    missing = tmp_path / "does_not_exist"
+    assert list_files(str(missing)) == []