From 91d02a0f2933c8388022763823e54bbba090e9b7 Mon Sep 17 00:00:00 2001 From: pravit-amp <43916793+pravit-amp@users.noreply.github.com> Date: Thu, 13 Aug 2026 09:47:25 -0700 Subject: [PATCH] fix(ingest): harden RepoIngestor GitPython clone surface (#868) (#905) * fix(ingest): harden RepoIngestor against GitPython URL and option injection Bump GitPython to >=3.1.58, allowlist clone kwargs, and validate repo URLs before clone_from to close env-var exfiltration and option-injection paths. * fix(ingest): accept scp-like SSH remotes in RepoIngestor URL validation * fix(ingest): resolve repo hostnames to block SSRF via private IPs * fix(ingest): map malformed repo URL parse errors to ValidationError * fix(ingest): bound and prune repo host resolve cache Cap the repository host DNS cache, prune expired entries on access, and evict the oldest entries so long-running processes cannot accumulate unbounded host lookups from user-supplied repo URLs. * fix(ingest): cap host resolve cache and tighten env-var token checks Bound the repo host DNS cache with pruning and oldest-entry eviction, and narrow URL env-var blocking to actual $VAR/${VAR} tokens so literal dollar signs are not rejected. * fix(ingest): preserve repo path compatibility and NAT64 support * docs(changelog): document RepoIngestor GitPython hardening (#905, closes #868) Records the clone-surface hardening (GitPython floor, clone-option allowlist, URL/SSRF validation), the two fixes made during review (NAT64 false-positive, local-path regression), and a known residual gap: the SSRF host check doesn't classify RFC 6598 CGNAT space (100.64.0.0/10) as blocked since ipaddress.is_private doesn't cover it. --------- Co-authored-by: Pravit Ampapathini Co-authored-by: Pravit Ampapathini Co-authored-by: Sameer Kadam Co-authored-by: KaifAhmad1 --- CHANGELOG.md | 10 + pyproject.toml | 2 +- semantica/ingest/methods.py | 16 +- semantica/ingest/repo_ingestor.py | 316 ++++++++++++- tests/ingest/test_repo_ingestor_security.py | 482 ++++++++++++++++++++ 5 files changed, 808 insertions(+), 18 deletions(-) create mode 100644 tests/ingest/test_repo_ingestor_security.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 91887706..d38a1901 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -88,6 +88,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- **`RepoIngestor` clone surface hardened against GitPython URL/option injection** (#905, closes #868) by @pravit-amp + - `RepoIngestor.ingest_repository()` passed the caller-supplied repository URL and arbitrary `**options` straight through to `git.Repo.clone_from()` on a `GitPython>=3.1.50` floor predating hardening for `ext::`-style transport helpers and `$VAR`/`${VAR}` environment-variable expansion in clone URLs — unvalidated clone options (`upload_pack`, `multi_options`, `template`, `config`, `env`, ...) could be abused for command execution, and unvalidated hostnames allowed SSRF against internal services (e.g. cloud metadata endpoints) + - `GitPython` floor raised to `>=3.1.58` + - Clone options passed to `clone_from()` are now allowlisted to `{depth, branch, single_branch, no_tags}`; anything else raises `ValidationError` before the clone is attempted + - Repository URLs are validated before cloning: scheme allowlist (`https`, `http`, `git`, `ssh`), rejection of `$VAR`/`${VAR}` tokens, and hostname resolution with every returned address screened against private/loopback/link-local/unspecified ranges. scp-like SSH remotes (`user@host:path`) are recognized and normalized to `ssh://` before the clone call + - **Fixed during review** (@Sameer6305): the SSRF check originally used `ip.is_reserved`, which flags the NAT64 Well-Known Prefix (`64:ff9b::/96`, RFC 6052) as reserved — falsely blocking `github.com` and other public hosts on IPv6-only/dual-stack networks using NAT64. Narrowed the block list to private/loopback/link-local/unspecified only + - **Fixed during review** (@Sameer6305): local filesystem repository paths (`git clone /path/to/local/repo`) were being treated as remote URLs and rejected outright; local paths now bypass network validation entirely since they make no network requests and carry no SSRF risk + - **Known limitation**: the SSRF host check does not classify RFC 6598 Carrier-Grade NAT space (`100.64.0.0/10`) as blocked — Python's `ipaddress.IPv4Address.is_private` does not cover that range, so a hostname resolving into it (e.g. some Kubernetes/CNI pod networks) would not be caught. Follow-up recommended to add it explicitly alongside the existing private/loopback/link-local checks + - `pytest tests/ingest/test_repo_ingestor_security.py -v`: 44 passed + - **HTTP response header injection via `node_id`, unbounded-memory DoS in link prediction, and unsanitized imported node IDs in the Explorer** (#912) by @Sunil56224972 - `semantica/explorer/routes/provenance.py`'s `GET /api/provenance/report` f-string-interpolated the `node_id` query parameter directly into the `Content-Disposition` response header; a `\r\n`-bearing `node_id` could inject arbitrary response headers (`Set-Cookie` session fixation, `Content-Type` override for reflected XSS). Fixed with `_safe_content_disposition_filename()`, which strips `\r`, `\n`, `\x00`, `"`, `\` and length-caps the value before interpolation - `POST /api/enrich/links` (link prediction) loaded up to 999,999 nodes with no cap or concurrency guard, then scored every candidate — a single request could consume ~1.6 GB RAM, and concurrent requests compounded that with no limit. Capped the candidate pool at 10,000 nodes (`413` if exceeded) and added an `asyncio.Semaphore(2)`, mirroring the SPARQL DoS fix in #898 diff --git a/pyproject.toml b/pyproject.toml index d7fc2899..577cb94f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -60,7 +60,7 @@ dependencies = [ "plotly>=6.8.0", "ipywidgets>=8.0.0", "requests>=2.34.2", - "GitPython>=3.1.50", + "GitPython>=3.1.58", "chardet>=7.4.3", "protobuf>=5.29.1,<8.0", "grpcio>=1.81.1", diff --git a/semantica/ingest/methods.py b/semantica/ingest/methods.py index ac9775a2..586eb827 100644 --- a/semantica/ingest/methods.py +++ b/semantica/ingest/methods.py @@ -174,6 +174,7 @@ Example Usage: from __future__ import annotations +import re from pathlib import Path from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union @@ -183,6 +184,14 @@ from .config import ingest_config from .file_ingestor import FileIngestor, FileObject from .registry import method_registry +# SCP-like SSH remotes (user@host:path) — keep in sync with repo_ingestor +_SCP_LIKE_REPO_URL_RE = re.compile(r"^[^@\s]+@[^:\s]+:.+$") + + +def _is_scp_like_repo_source(source: str) -> bool: + """Return True for scp-like SSH remotes (``user@host:path``).""" + return bool(_SCP_LIKE_REPO_URL_RE.match(source.strip())) + if TYPE_CHECKING: from .api_ingestor import APIData from .arrow_ingestor import ArrowData @@ -880,7 +889,10 @@ def ingest_repository( if method == "clone" or ( isinstance(source, str) - and source.startswith(("http://", "https://", "git@")) + and ( + source.startswith(("http://", "https://")) + or _is_scp_like_repo_source(source) + ) ): return ingestor.ingest_repository(source, **kwargs) elif method == "analyze": @@ -1336,7 +1348,7 @@ def ingest( ("postgresql://", "mysql://", "sqlite://", "oracle://", "mssql://") ): source_type = "db" - elif source_str.startswith("git@") or source_str_lower.startswith( + elif _is_scp_like_repo_source(source_str) or source_str_lower.startswith( ("https://github.com", "https://gitlab.com") ): source_type = "repo" diff --git a/semantica/ingest/repo_ingestor.py b/semantica/ingest/repo_ingestor.py index 0cadb0d9..4b4f88c1 100644 --- a/semantica/ingest/repo_ingestor.py +++ b/semantica/ingest/repo_ingestor.py @@ -29,14 +29,19 @@ Author: Semantica Contributors License: MIT """ +import ipaddress import os import re import shutil +import socket import tempfile +import time +from collections import OrderedDict from dataclasses import dataclass, field from datetime import datetime from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Set, Tuple, Union +from urllib.parse import urlparse import git @@ -44,6 +49,25 @@ from ..utils.exceptions import ProcessingError, ValidationError from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker +# Safe subset of GitPython clone_from kwargs. Broader kwargs (multi_options, +# upload_pack, template, config, env, …) have been used in denylist-bypass +# attacks against older GitPython releases — keep them out of the call surface. +ALLOWED_CLONE_OPTIONS: Set[str] = {"depth", "branch", "single_branch", "no_tags"} +ALLOWED_REPO_URL_SCHEMES = frozenset({"https", "http", "git", "ssh"}) +# SCP-like SSH remotes: user@host:path/to/repo.git (no scheme) +_SCP_LIKE_REPO_URL_RE = re.compile(r"^[^@\s]+@[^:\s]+:.+$") +_ENV_VAR_TOKEN_RE = re.compile( + r"\$(\{[A-Za-z_][A-Za-z0-9_]*\}|[A-Za-z_][A-Za-z0-9_]*)" +) +# Short-lived DNS cache for host validation. This reduces repeated lookups but +# does not eliminate DNS-rebinding / TOCTOU races between validate and clone — +# network egress controls remain recommended. +_REPO_HOST_RESOLVE_CACHE: "OrderedDict[str, Tuple[float, Tuple[str, ...]]]" = ( + OrderedDict() +) +_REPO_HOST_RESOLVE_CACHE_TTL_SECONDS = 60.0 +_REPO_HOST_RESOLVE_CACHE_MAX_ENTRIES = 1024 + @dataclass class CodeFile: @@ -509,6 +533,268 @@ class RepoIngestor: self.logger.debug("Repo ingestor initialized") + @staticmethod + def _is_scp_like_repo_url(repo_url: str) -> bool: + """Return True for scp-like SSH remotes (``user@host:path``).""" + url = repo_url.strip() + # Avoid treating scheme URLs with userinfo as scp-like (e.g. https://u@h/...) + if "://" in url: + return False + return bool(_SCP_LIKE_REPO_URL_RE.match(url)) + + @staticmethod + def _scp_like_host(repo_url: str) -> str: + """Extract the hostname from an scp-like remote (``user@host:path``).""" + _, rest = repo_url.strip().split("@", 1) + host, _ = rest.split(":", 1) + return host + + @staticmethod + def _normalize_repo_url(repo_url: str) -> str: + """Normalize scp-like remotes to ``ssh://`` URLs; leave others unchanged. + + ``git@host:org/repo.git`` → ``ssh://git@host/org/repo.git`` + """ + url = repo_url.strip() + if not RepoIngestor._is_scp_like_repo_url(url): + return url + user_host, path = url.split(":", 1) + if not path.startswith("/"): + path = f"/{path}" + return f"ssh://{user_host}{path}" + + @staticmethod + def _is_blocked_ip( + ip: Union[ipaddress.IPv4Address, ipaddress.IPv6Address], + ) -> bool: + """Return True if *ip* is an SSRF-sensitive address. + + Blocks private (RFC1918/ULA), loopback, link-local (including + 169.254.x.x / fe80::/10 cloud-metadata ranges), and unspecified + addresses. + + Intentionally does **not** use ``ip.is_reserved``: Python's + ``ipaddress`` module marks the NAT64 Well-Known Prefix + (64:ff9b::/96, RFC 6052) as reserved, which causes false positives + on IPv6-only and dual-stack networks that use NAT64 for public + Internet access (e.g., github.com resolves to 64:ff9b::… on such + networks). Those addresses are not SSRF-sensitive. + """ + return bool( + ip.is_private + or ip.is_loopback + or ip.is_link_local + or ip.is_unspecified + ) + + @staticmethod + def _resolve_repo_host_ips(host: str) -> Tuple[str, ...]: + """Resolve *host* to IP strings via ``socket.getaddrinfo``, with TTL cache. + + Note: caching and pre-clone resolution mitigate repeated lookups but + cannot fully prevent DNS rebinding between validation and clone. + Prefer network-layer egress controls for defense in depth. + """ + cache_key = host.lower().rstrip(".") + now = time.monotonic() + RepoIngestor._prune_repo_host_resolve_cache(now) + cached = _REPO_HOST_RESOLVE_CACHE.get(cache_key) + if cached is not None: + expires_at, ips = cached + if now < expires_at: + _REPO_HOST_RESOLVE_CACHE.move_to_end(cache_key) + return ips + _REPO_HOST_RESOLVE_CACHE.pop(cache_key, None) + + try: + addrinfos = socket.getaddrinfo( + host, None, type=socket.SOCK_STREAM + ) + except socket.gaierror as exc: + raise ValidationError( + f"Cannot resolve repository host {host!r}: {exc}" + ) from exc + + ips: List[str] = [] + seen: Set[str] = set() + for _family, _type, _proto, _canonname, sockaddr in addrinfos: + addr = sockaddr[0] + if addr not in seen: + seen.add(addr) + ips.append(addr) + + if not ips: + raise ValidationError( + f"Cannot resolve repository host {host!r}: no addresses" + ) + + result = tuple(ips) + _REPO_HOST_RESOLVE_CACHE[cache_key] = ( + now + _REPO_HOST_RESOLVE_CACHE_TTL_SECONDS, + result, + ) + _REPO_HOST_RESOLVE_CACHE.move_to_end(cache_key) + RepoIngestor._prune_repo_host_resolve_cache(now) + return result + + @staticmethod + def _prune_repo_host_resolve_cache(now: Optional[float] = None) -> None: + """Remove expired host entries and enforce a hard cache size cap.""" + if now is None: + now = time.monotonic() + + expired_keys = [ + cache_key + for cache_key, (expires_at, _ips) in _REPO_HOST_RESOLVE_CACHE.items() + if expires_at <= now + ] + for cache_key in expired_keys: + _REPO_HOST_RESOLVE_CACHE.pop(cache_key, None) + + while len(_REPO_HOST_RESOLVE_CACHE) > _REPO_HOST_RESOLVE_CACHE_MAX_ENTRIES: + _REPO_HOST_RESOLVE_CACHE.popitem(last=False) + + @staticmethod + def _validate_repo_host(host: str) -> None: + """Reject localhost names and hosts resolving to blocked addresses. + + Literal IPs are checked directly. Hostnames are resolved with + ``socket.getaddrinfo`` and **every** returned address is screened. + """ + if not host: + raise ValidationError("Repository URL must include a host") + + lowered = host.lower().rstrip(".") + if lowered == "localhost" or lowered.endswith(".localhost"): + raise ValidationError(f"Repository host is not allowed: {host}") + + try: + ip = ipaddress.ip_address(host) + except ValueError: + # Hostname: resolve and validate all returned addresses + for addr in RepoIngestor._resolve_repo_host_ips(host): + try: + resolved = ipaddress.ip_address(addr) + except ValueError: + continue + if RepoIngestor._is_blocked_ip(resolved): + raise ValidationError( + f"Repository host resolves to a blocked address: " + f"{host} -> {addr}" + ) + return + + if RepoIngestor._is_blocked_ip(ip): + raise ValidationError( + f"Repository host resolves to a blocked address: {host}" + ) + + @staticmethod + def _is_local_repo_path(repo_url: str) -> bool: + """Return True if *repo_url* looks like a local filesystem path. + + Matches absolute paths (``/…``, ``C:\\…``), relative paths + (``./…``, ``../…``), and bare names without a scheme or ``@host:`` + pattern that would be interpreted as a local path by git. + """ + url = repo_url.strip() + if "://" in url: + return False + if RepoIngestor._is_scp_like_repo_url(url): + return False + # Absolute POSIX or Windows paths, or relative paths + p = Path(url) + if p.is_absolute(): + return True + # ./ or ../ + if url.startswith(("./", "../", ".\\", "..\\")): + return True + # Existing local directory (best-effort; may not exist yet during tests) + if p.exists(): + return True + return False + + @staticmethod + def _validate_repo_url(repo_url: str) -> None: + """Validate a repository URL before cloning. + + Accepts http(s)/git/ssh URLs, scp-like SSH remotes + (``user@host:path``), and local filesystem paths. Rejects empty + values, unsupported schemes, missing hosts, environment variable + expansion tokens (``$VAR`` / ``${VAR}``), and hosts that are or + resolve to private / loopback / link-local addresses. + + Local filesystem paths bypass network validation because + ``git clone /path/to/local/repo`` makes no network requests and + carries no SSRF risk. + + DNS resolution is TOCTOU-sensitive (rebinding); pair with egress + controls in production deployments. + """ + if not isinstance(repo_url, str) or not repo_url.strip(): + raise ValidationError("Repository URL must be a non-empty string") + + # Defense-in-depth against GitPython env-var expansion in clone URLs + # (GHSA-2f96-g7mh-g2hx / related). Prefer rejecting before clone_from. + if _ENV_VAR_TOKEN_RE.search(repo_url): + raise ValidationError( + "Repository URL must not contain environment variable " + "references ($VAR / ${VAR})" + ) + + url = repo_url.strip() + + # Local filesystem paths: no network, no SSRF risk — skip host checks. + if RepoIngestor._is_local_repo_path(url): + return + + # scp-like syntax has no URL scheme; validate host then accept. + if RepoIngestor._is_scp_like_repo_url(url): + RepoIngestor._validate_repo_host(RepoIngestor._scp_like_host(url)) + return + + try: + parsed = urlparse(url) + # ``hostname`` can raise ValueError for malformed netloc (e.g. bad IPv6) + host = parsed.hostname + except ValueError as e: + raise ValidationError(f"Invalid repository URL: {e}") from e + + scheme = (parsed.scheme or "").lower() + if scheme not in ALLOWED_REPO_URL_SCHEMES: + raise ValidationError( + f"Unsupported repository URL scheme {scheme!r}. " + f"Allowed schemes: {sorted(ALLOWED_REPO_URL_SCHEMES)}" + ) + if not parsed.netloc or not host: + raise ValidationError( + f"Repository URL must include a host: {repo_url}" + ) + + RepoIngestor._validate_repo_host(host) + + @staticmethod + def _filter_clone_options(options: Dict[str, Any]) -> Dict[str, Any]: + """Return only allowlisted git clone kwargs; reject anything else.""" + # Semantica processing options — never forwarded to clone_from + non_git_options = { + "include_history", + "file_filters", + "commit_filters", + "include_extensions", + "max_depth", + } + candidate = { + k: v for k, v in options.items() if k not in non_git_options + } + unsafe = set(candidate) - ALLOWED_CLONE_OPTIONS + if unsafe: + raise ValidationError( + f"Clone option(s) not permitted: {sorted(unsafe)}. " + f"Allowed options: {sorted(ALLOWED_CLONE_OPTIONS)}" + ) + return candidate + def ingest_repository(self, repo_url: str, **options) -> Dict[str, Any]: """ Ingest and process a Git repository. @@ -518,6 +804,8 @@ class RepoIngestor: **options: Processing options: - branch: Specific branch to checkout - depth: Clone depth (for shallow clones) + - single_branch: Clone only a single branch + - no_tags: Skip cloning tags - include_history: Whether to include commit history - include_extensions: List of file extensions to include (e.g., ["py", "md"]) @@ -533,27 +821,19 @@ class RepoIngestor: ) try: + # Validate repository URL before any clone attempt + self._validate_repo_url(repo_url) + clone_url = self._normalize_repo_url(repo_url) + # Handle option aliases and filters if "max_depth" in options and "depth" not in options: options["depth"] = options["max_depth"] - # Separate git clone options from processing options - # We filter out known non-git options to avoid passing invalid flags to git clone - non_git_options = { - "include_history", - "file_filters", - "commit_filters", - "include_extensions", - "max_depth", - } - clone_options = { - k: v for k, v in options.items() if k not in non_git_options - } + clone_options = self._filter_clone_options(options) - # Validate repository URL try: parsed = git.Repo.clone_from( - repo_url, self._get_temp_dir(), **clone_options + clone_url, self._get_temp_dir(), **clone_options ) except Exception as e: self.progress_tracker.update_tracking( @@ -627,6 +907,12 @@ class RepoIngestor: "temp_path": str(repo_path), } + except ValidationError as e: + # Keep validation failures typed for callers; do not wrap as ProcessingError + self.progress_tracker.update_tracking( + tracking_id, status="failed", message=str(e) + ) + raise except Exception as e: self.progress_tracker.update_tracking( tracking_id, status="failed", message=str(e) diff --git a/tests/ingest/test_repo_ingestor_security.py b/tests/ingest/test_repo_ingestor_security.py new file mode 100644 index 00000000..738c0f76 --- /dev/null +++ b/tests/ingest/test_repo_ingestor_security.py @@ -0,0 +1,482 @@ +"""Security-focused tests for RepoIngestor (issue #868).""" + +import socket +from unittest.mock import MagicMock, patch + +import pytest + +from semantica.ingest import repo_ingestor as repo_ingestor_mod +from semantica.ingest.repo_ingestor import ( + ALLOWED_CLONE_OPTIONS, + RepoIngestor, +) +from semantica.utils.exceptions import ValidationError + + +def _fake_addrinfo(*addrs: str): + """Build a getaddrinfo-shaped result list for the given IP strings.""" + results = [] + for addr in addrs: + family = socket.AF_INET6 if ":" in addr else socket.AF_INET + results.append( + (family, socket.SOCK_STREAM, 0, "", (addr, 0)) + ) + return results + + +@pytest.fixture(autouse=True) +def _clear_repo_host_resolve_cache(): + repo_ingestor_mod._REPO_HOST_RESOLVE_CACHE.clear() + yield + repo_ingestor_mod._REPO_HOST_RESOLVE_CACHE.clear() + + +class TestRepoUrlValidation: + def test_accepts_https_github_url(self): + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("140.82.112.3"), + ): + RepoIngestor._validate_repo_url("https://github.com/user/repo.git") + + def test_accepts_ssh_scheme(self): + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("140.82.112.3"), + ): + RepoIngestor._validate_repo_url("ssh://git@github.com/user/repo.git") + + def test_accepts_scp_like_ssh_remote(self): + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("140.82.112.3"), + ): + RepoIngestor._validate_repo_url("git@github.com:user/repo.git") + RepoIngestor._validate_repo_url( + "deploy@gitlab.example.com:team/app.git" + ) + + def test_normalizes_scp_like_to_ssh_url(self): + assert ( + RepoIngestor._normalize_repo_url("git@github.com:user/repo.git") + == "ssh://git@github.com/user/repo.git" + ) + assert ( + RepoIngestor._normalize_repo_url( + "https://github.com/user/repo.git" + ) + == "https://github.com/user/repo.git" + ) + + def test_rejects_empty(self): + with pytest.raises(ValidationError, match="non-empty"): + RepoIngestor._validate_repo_url("") + + def test_rejects_file_scheme(self): + with pytest.raises(ValidationError, match="Unsupported repository URL scheme"): + RepoIngestor._validate_repo_url("file:///tmp/repo.git") + + def test_rejects_env_var_tokens(self): + with pytest.raises(ValidationError, match="environment variable"): + RepoIngestor._validate_repo_url( + "https://attacker.example/${AWS_SECRET_ACCESS_KEY}/repo.git" + ) + with pytest.raises(ValidationError, match="environment variable"): + RepoIngestor._validate_repo_url( + "https://$GITHUB_TOKEN@attacker.example/repo.git" + ) + with pytest.raises(ValidationError, match="environment variable"): + RepoIngestor._validate_repo_url( + "git@github.com:org/${AWS_SECRET_ACCESS_KEY}.git" + ) + + def test_accepts_literal_dollar_without_env_var_token(self): + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("140.82.112.3"), + ): + RepoIngestor._validate_repo_url("https://example.com/repo$1.git") + RepoIngestor._validate_repo_url("git@example.com:team/repo$1.git") + + def test_rejects_localhost_and_loopback(self): + with pytest.raises(ValidationError, match="not allowed|blocked"): + RepoIngestor._validate_repo_url("https://localhost/repo.git") + with pytest.raises(ValidationError, match="blocked"): + RepoIngestor._validate_repo_url("https://127.0.0.1/repo.git") + with pytest.raises(ValidationError, match="not allowed|blocked"): + RepoIngestor._validate_repo_url("git@localhost:repo.git") + with pytest.raises(ValidationError, match="blocked"): + RepoIngestor._validate_repo_url("git@127.0.0.1:repo.git") + + def test_rejects_private_and_metadata_ips(self): + for url in ( + "https://10.0.0.1/repo.git", + "https://192.168.1.1/repo.git", + "https://172.16.5.5/repo.git", + "http://169.254.169.254/latest/meta-data/", + "git@10.0.0.1:repo.git", + "git@169.254.169.254:repo.git", + ): + with pytest.raises(ValidationError, match="blocked"): + RepoIngestor._validate_repo_url(url) + + def test_rejects_hostname_resolving_to_private_ip(self): + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("10.0.0.5"), + ): + with pytest.raises(ValidationError, match="blocked"): + RepoIngestor._validate_repo_url( + "https://internal.example/repo.git" + ) + + def test_rejects_hostname_if_any_resolved_ip_is_blocked(self): + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("8.8.8.8", "127.0.0.1"), + ): + with pytest.raises(ValidationError, match="blocked"): + RepoIngestor._validate_repo_url( + "https://mixed.example/repo.git" + ) + + def test_rejects_unresolvable_hostname(self): + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + side_effect=socket.gaierror(8, "Name or service not known"), + ): + with pytest.raises(ValidationError, match="Cannot resolve"): + RepoIngestor._validate_repo_url( + "https://does-not-resolve.invalid/repo.git" + ) + + def test_hostname_resolution_is_cached(self): + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("1.2.3.4"), + ) as mock_gai: + RepoIngestor._validate_repo_url("https://cached.example/repo.git") + RepoIngestor._validate_repo_url("https://cached.example/other.git") + assert mock_gai.call_count == 1 + + def test_rejects_malformed_netloc_as_validation_error(self): + with pytest.raises(ValidationError, match="Invalid repository URL"): + RepoIngestor._validate_repo_url("http://[::1") + with pytest.raises(ValidationError, match="Invalid repository URL"): + RepoIngestor._validate_repo_url("http://[") + with pytest.raises(ValidationError, match="Invalid repository URL"): + RepoIngestor._validate_repo_url("https://user@[::1/repo.git") + + def test_malformed_url_surfaces_as_validation_error_from_ingest(self): + with patch("semantica.ingest.repo_ingestor.git.Repo") as MockRepo, patch( + "semantica.ingest.repo_ingestor.get_progress_tracker" + ) as mock_get_tracker: + mock_get_tracker.return_value = MagicMock() + ingestor = RepoIngestor() + with pytest.raises(ValidationError, match="Invalid repository URL"): + ingestor.ingest_repository("http://[::1") + MockRepo.clone_from.assert_not_called() + + +class TestCloneOptionAllowlist: + def test_allows_safe_options(self): + filtered = RepoIngestor._filter_clone_options( + {"depth": 1, "branch": "main", "single_branch": True, "no_tags": True} + ) + assert filtered == { + "depth": 1, + "branch": "main", + "single_branch": True, + "no_tags": True, + } + + def test_strips_processing_options_without_error(self): + filtered = RepoIngestor._filter_clone_options( + { + "depth": 1, + "include_history": True, + "include_extensions": ["py"], + "file_filters": {}, + "commit_filters": {}, + "max_depth": 5, + } + ) + assert filtered == {"depth": 1} + + def test_rejects_multi_options(self): + with pytest.raises(ValidationError, match="not permitted"): + RepoIngestor._filter_clone_options( + {"multi_options": ["--template=/tmp/evil"]} + ) + + def test_rejects_upload_pack_and_template(self): + for key in ("upload_pack", "template", "config", "env"): + with pytest.raises(ValidationError, match="not permitted"): + RepoIngestor._filter_clone_options({key: "x"}) + + def test_allowlist_matches_documented_safe_set(self): + assert ALLOWED_CLONE_OPTIONS == { + "depth", + "branch", + "single_branch", + "no_tags", + } + + +class TestIngestRepositoryGuards: + def test_unsafe_url_never_reaches_clone_from(self): + with patch("semantica.ingest.repo_ingestor.git.Repo") as MockRepo, patch( + "semantica.ingest.repo_ingestor.get_progress_tracker" + ) as mock_get_tracker: + mock_get_tracker.return_value = MagicMock() + ingestor = RepoIngestor() + with pytest.raises(ValidationError, match="environment variable"): + ingestor.ingest_repository( + "https://evil.example/${AWS_SECRET_ACCESS_KEY}/r.git" + ) + MockRepo.clone_from.assert_not_called() + + def test_unsafe_clone_option_never_reaches_clone_from(self): + with patch("semantica.ingest.repo_ingestor.git.Repo") as MockRepo, patch( + "semantica.ingest.repo_ingestor.get_progress_tracker" + ) as mock_get_tracker, patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("140.82.112.3"), + ): + mock_get_tracker.return_value = MagicMock() + ingestor = RepoIngestor() + with pytest.raises(ValidationError, match="not permitted"): + ingestor.ingest_repository( + "https://github.com/user/repo.git", + multi_options=["--template=/tmp/evil"], + ) + MockRepo.clone_from.assert_not_called() + + def test_hostname_resolving_private_never_reaches_clone_from(self): + with patch("semantica.ingest.repo_ingestor.git.Repo") as MockRepo, patch( + "semantica.ingest.repo_ingestor.get_progress_tracker" + ) as mock_get_tracker, patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("192.168.1.50"), + ): + mock_get_tracker.return_value = MagicMock() + ingestor = RepoIngestor() + with pytest.raises(ValidationError, match="blocked"): + ingestor.ingest_repository( + "https://ssrf.example/internal/repo.git" + ) + MockRepo.clone_from.assert_not_called() + + def test_safe_options_forwarded_to_clone_from(self): + with patch("semantica.ingest.repo_ingestor.git.Repo") as MockRepo, patch( + "semantica.ingest.repo_ingestor.tempfile.mkdtemp", + return_value="/tmp/fake-repo", + ), patch("semantica.ingest.repo_ingestor.shutil.rmtree"), patch( + "semantica.ingest.repo_ingestor.get_progress_tracker" + ) as mock_get_tracker, patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("140.82.112.3"), + ), patch.object( + RepoIngestor, "extract_code_files", return_value=[] + ), patch.object( + RepoIngestor, "get_repository_info", return_value={"url": "x"} + ), patch.object(RepoIngestor, "analyze_commits", return_value=[]): + mock_get_tracker.return_value = MagicMock() + mock_repo = MagicMock() + MockRepo.clone_from.return_value = mock_repo + MockRepo.return_value = mock_repo + + ingestor = RepoIngestor() + with patch.object( + ingestor.analyzer, "analyze_structure", return_value={} + ), patch.object( + ingestor.analyzer, "calculate_metrics", return_value={} + ): + ingestor.ingest_repository( + "https://github.com/user/repo.git", + depth=1, + branch="main", + include_history=False, + ) + + kwargs = MockRepo.clone_from.call_args.kwargs + assert kwargs.get("depth") == 1 + assert kwargs.get("branch") == "main" + assert "include_history" not in kwargs + assert "multi_options" not in kwargs + + def test_scp_like_remote_normalized_before_clone(self): + with patch("semantica.ingest.repo_ingestor.git.Repo") as MockRepo, patch( + "semantica.ingest.repo_ingestor.tempfile.mkdtemp", + return_value="/tmp/fake-repo", + ), patch("semantica.ingest.repo_ingestor.shutil.rmtree"), patch( + "semantica.ingest.repo_ingestor.get_progress_tracker" + ) as mock_get_tracker, patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("140.82.112.3"), + ), patch.object( + RepoIngestor, "extract_code_files", return_value=[] + ), patch.object( + RepoIngestor, "get_repository_info", return_value={"url": "x"} + ), patch.object(RepoIngestor, "analyze_commits", return_value=[]): + mock_get_tracker.return_value = MagicMock() + mock_repo = MagicMock() + MockRepo.clone_from.return_value = mock_repo + MockRepo.return_value = mock_repo + + ingestor = RepoIngestor() + with patch.object( + ingestor.analyzer, "analyze_structure", return_value={} + ), patch.object( + ingestor.analyzer, "calculate_metrics", return_value={} + ): + ingestor.ingest_repository("git@github.com:user/repo.git") + + assert MockRepo.clone_from.call_args.args[0] == ( + "ssh://git@github.com/user/repo.git" + ) + +class TestIsReservedNAT64Regression: + """Regression tests for the is_reserved / NAT64 false-positive fix. + + Python's ipaddress.is_reserved marks 64:ff9b::/96 (NAT64 Well-Known + Prefix, RFC 6052) as reserved=True, which caused github.com to be + falsely blocked on IPv6-only / dual-stack networks that use NAT64. + """ + + def test_nat64_prefix_not_blocked(self): + """64:ff9b::/96 addresses must not be blocked by _is_blocked_ip.""" + import ipaddress + + # Typical NAT64 translation of 140.82.112.3 (github.com) + addr = ipaddress.ip_address("64:ff9b::8c52:7003") + assert not RepoIngestor._is_blocked_ip(addr), ( + "NAT64 WKP address should not be blocked; " + "it is a legitimate public IPv6 address on NAT64 networks." + ) + + def test_nat64_local_prefix_not_blocked(self): + """64:ff9b:1::/48 (RFC 8215 local NAT64) is private by Python 3.12 + definition and IS correctly blocked — it's a locally-assigned range, + not globally routable. + """ + import ipaddress + + addr = ipaddress.ip_address("64:ff9b:1::1") + # is_private=True in Python 3.12 — legitimately blocked + assert RepoIngestor._is_blocked_ip(addr) + + def test_private_ipv6_still_blocked(self): + """ULA (fc00::/7) must still be blocked.""" + import ipaddress + + assert RepoIngestor._is_blocked_ip(ipaddress.ip_address("fc00::1")) + assert RepoIngestor._is_blocked_ip(ipaddress.ip_address("fd12:3456::1")) + + def test_ipv6_loopback_still_blocked(self): + import ipaddress + + assert RepoIngestor._is_blocked_ip(ipaddress.ip_address("::1")) + + def test_ipv6_link_local_still_blocked(self): + import ipaddress + + assert RepoIngestor._is_blocked_ip(ipaddress.ip_address("fe80::1")) + + def test_documentation_prefix_blocked(self): + """2001:db8::/32 is documentation-only and classified as + is_private=True in Python 3.12. It is correctly blocked. + """ + import ipaddress + + addr = ipaddress.ip_address("2001:db8::1") + assert RepoIngestor._is_blocked_ip(addr) + + def test_public_ipv4_not_blocked(self): + import ipaddress + + assert not RepoIngestor._is_blocked_ip(ipaddress.ip_address("140.82.112.3")) + + def test_public_ipv6_not_blocked(self): + import ipaddress + + assert not RepoIngestor._is_blocked_ip( + ipaddress.ip_address("2001:4860:4860::8888") + ) + + def test_host_resolving_to_nat64_address_is_allowed(self): + """A hostname that resolves to a NAT64 address (plus a public IPv4) + must not be blocked — this was the real-world failure mode. + """ + # Simulate github.com on a NAT64 network + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("64:ff9b::8c52:7003", "140.82.112.3"), + ): + # Should not raise + RepoIngestor._validate_repo_url("https://github.com/user/repo.git") + + def test_host_resolving_only_to_nat64_is_allowed(self): + """Even if the only resolved address is a NAT64 address, it is allowed + because it is a valid public address. + """ + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo", + return_value=_fake_addrinfo("64:ff9b::8c52:7003"), + ): + RepoIngestor._validate_repo_url("https://github.com/user/repo.git") + + +class TestLocalPathSupport: + """Regression tests for local repository path backward compatibility.""" + + def test_is_local_repo_path_absolute(self, tmp_path): + """Absolute paths are recognised as local.""" + assert RepoIngestor._is_local_repo_path(str(tmp_path)) + + def test_is_local_repo_path_relative(self): + """./… and ../… are recognised as local.""" + assert RepoIngestor._is_local_repo_path("./repo") + assert RepoIngestor._is_local_repo_path("../sibling-repo") + + def test_is_local_repo_path_not_remote(self): + """Remote URLs are not local.""" + assert not RepoIngestor._is_local_repo_path("https://github.com/u/r.git") + assert not RepoIngestor._is_local_repo_path("git@github.com:u/r.git") + assert not RepoIngestor._is_local_repo_path("ssh://git@github.com/r.git") + + def test_validate_repo_url_accepts_absolute_local_path(self, tmp_path): + """_validate_repo_url must not raise for an absolute local path.""" + RepoIngestor._validate_repo_url(str(tmp_path)) + + def test_validate_repo_url_accepts_relative_local_path(self): + """_validate_repo_url must not raise for ./… paths.""" + RepoIngestor._validate_repo_url("./repo") + + def test_validate_repo_url_env_var_still_blocked_in_local_path(self): + """Env-var tokens in local paths are still rejected.""" + with pytest.raises(ValidationError, match="environment variable"): + RepoIngestor._validate_repo_url("./$SECRET_KEY/repo") + + def test_local_path_never_reaches_dns_resolution(self, tmp_path): + """Local paths must not trigger DNS lookups.""" + with patch( + "semantica.ingest.repo_ingestor.socket.getaddrinfo" + ) as mock_gai: + RepoIngestor._validate_repo_url(str(tmp_path)) + mock_gai.assert_not_called() + + def test_ingest_repository_local_path_passes_validation(self, tmp_path): + """ingest_repository with a local path must not fail at URL validation.""" + with patch("semantica.ingest.repo_ingestor.git.Repo") as MockRepo, patch( + "semantica.ingest.repo_ingestor.get_progress_tracker" + ) as mock_get_tracker: + mock_get_tracker.return_value = MagicMock() + ingestor = RepoIngestor() + # Expect clone to fail (temp_dir logic), but NOT a ValidationError + try: + ingestor.ingest_repository(str(tmp_path)) + except Exception as exc: + assert not isinstance(exc, ValidationError), ( + f"Local path must not raise ValidationError; got: {exc}" + )