Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 7 additions & 28 deletions app/scanner/clone.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,24 +2,22 @@
import shutil
import subprocess
import tempfile
from urllib.parse import urlparse

from app.scanner.exceptions import RepoCloneError, RepoTooLargeError
from app.utils import is_allowed_url

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
total: int = 0

for dirpath, dirnames, filenames in os.walk(path, followlinks=False):
for name in filenames:
file_path = os.path.join(dirpath, name)
file_path: str = os.path.join(dirpath, name)
if os.path.islink(file_path):
continue
try:
Expand Down Expand Up @@ -47,27 +45,8 @@ def _run_git(args: list[str], cwd: str, timeout: int) -> None:
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):
if not is_allowed_url(url):
raise RepoCloneError("URL не разрешён (нужен http(s):// или git@host:...)")

ref: str = commit_sha or "HEAD"
Expand All @@ -83,10 +62,10 @@ def clone_repo(url: str, commit_sha: str | None = None) -> tuple[str, str]:
)
_run_git(["checkout", "FETCH_HEAD"], temp_folder_path, GIT_TIMEOUT_SECONDS)

resolved_sha = _get_head_sha(temp_folder_path)
resolved_sha: str = _get_head_sha(temp_folder_path)

size_bytes = _check_repo_size(temp_folder_path)
max_bytes = MAX_REPO_SIZE_MB * 1024 * 1024
size_bytes: int = _check_repo_size(temp_folder_path)
max_bytes: int = MAX_REPO_SIZE_MB * 1024 * 1024

if size_bytes > max_bytes:
raise RepoTooLargeError(
Expand Down
22 changes: 22 additions & 0 deletions app/utils.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,27 @@
from datetime import datetime, timezone
from urllib.parse import ParseResult, urlparse

ALLOWED_HOSTS = ("github.com", "gitlab.com")


def utcnow() -> datetime:
return datetime.now(timezone.utc)


def is_allowed_url(url: str) -> bool:
if not url:
return False

parsed: ParseResult = urlparse(url)
if parsed.scheme in ("http", "https"):
host: str | None = parsed.hostname

return host is not None and host.lower() in ALLOWED_HOSTS

if url.startswith("git@"):
rest: str = 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
5 changes: 3 additions & 2 deletions tests/test_clone.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,9 @@
import pytest

import app.scanner.clone as clone_module
from app.scanner.clone import _is_allowed_url, cleanup, clone_repo
from app.scanner.clone import cleanup, clone_repo
from app.scanner.exceptions import RepoCloneError, RepoTooLargeError
from app.utils import is_allowed_url


def test_clone_repo_real_github():
Expand Down Expand Up @@ -56,7 +57,7 @@ def test_clone_nonexist_repo_raises():
],
)
def test_is_allowed_url(url, status):
assert _is_allowed_url(url) is status
assert is_allowed_url(url) is status


def test_cleanup_removes_directory(tmp_path):
Expand Down