diff --git a/app/scanner/clone.py b/app/scanner/clone.py new file mode 100644 index 0000000..cb50dbe --- /dev/null +++ b/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) diff --git a/app/scanner/exceptions.py b/app/scanner/exceptions.py new file mode 100644 index 0000000..b126515 --- /dev/null +++ b/app/scanner/exceptions.py @@ -0,0 +1,6 @@ +class RepoCloneError(Exception): + pass + + +class RepoTooLargeError(Exception): + pass diff --git a/app/scanner/files.py b/app/scanner/files.py new file mode 100644 index 0000000..70759da --- /dev/null +++ b/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 diff --git a/tests/test_clone.py b/tests/test_clone.py new file mode 100644 index 0000000..3bd6ada --- /dev/null +++ b/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), + ("git@github.com:owner/repo.git", True), + ("git@gitlab.com:owner/repo", True), + ("file:///etc/passwd", False), + ("local/path/to/repo", False), + ("../relative/path", False), + ("https://badurl.com/owner/repo", False), + ("git@evil.com:owner/repo", False), + ("git@github.com", False), + ("", False), + ("https://github.com.badurl.com", False), + ("HTTPS://GITHUB.COM/owner/repo.git", True), + ("git@github.com:", 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 diff --git a/tests/test_files.py b/tests/test_files.py new file mode 100644 index 0000000..522af91 --- /dev/null +++ b/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)) == []