diff --git a/src/skillspector/input_handler.py b/src/skillspector/input_handler.py index c42eaf1ce..9334a8068 100644 --- a/src/skillspector/input_handler.py +++ b/src/skillspector/input_handler.py @@ -50,7 +50,7 @@ from stat import S_IFMT, S_ISDIR, S_ISLNK, S_ISREG from time import monotonic from typing import BinaryIO, NoReturn, cast -from urllib.parse import urljoin, urlparse +from urllib.parse import unquote, urljoin, urlparse import httpx @@ -739,6 +739,22 @@ def resolve(self, input_path: str) -> tuple[Path, str]: """ input_path = input_path.strip() + git_target = self._github_tree_target(input_path) + if git_target is not None: + repository_url, branch, subdirectory = git_target + clone_dir = self._clone_git(repository_url, branch=branch) + try: + clone_root = clone_dir.resolve() + target = (clone_root / subdirectory).resolve() + target.relative_to(clone_root) + if not target.is_dir() or target.is_symlink(): + raise ValueError("Git URL subdirectory does not exist or is not a directory") + return target, "git" + except (OSError, ValueError): + # No caller receives the resolver after a failed selection, so it + # cannot clean an owned clone on our behalf. + self.cleanup() + raise if self._is_git_url(input_path): return self._clone_git(input_path), "git" if self._is_file_url(input_path): @@ -1009,6 +1025,78 @@ def _is_git_url(self, path: str) -> bool: return True return False + def _github_tree_target(self, path: str) -> tuple[str, str, PurePosixPath] | None: + """Return a canonical clone target for a GitHub ``/tree//`` URL. + + The ref itself may contain ``/`` (for example ``feature/foo``), so the + split between ref and subdirectory is resolved against the remote's + advertised refs: the longest ``refs/heads/`` or ``refs/tags/`` name + that prefixes the ``/tree/`` segments wins. Without this, a URL for + branch ``feature/foo`` would clone branch ``feature`` and treat + ``foo`` as part of the subdirectory. + """ + parsed = urlparse(path) + if parsed.scheme != "https" or parsed.hostname != "github.com": + return None + parts = [unquote(part) for part in parsed.path.split("/") if part] + if len(parts) < 4 or parts[2] != "tree": + return None + owner, repository = parts[0], parts[1] + segments = parts[3:] + if any(part in {"", ".", ".."} or "/" in part or "\\" in part for part in segments): + raise ValueError("Git URL subdirectory must stay within the repository") + repository_url = f"https://github.com/{owner}/{repository}.git" + ref, subdirectory = self._resolve_tree_ref(repository_url, segments) + return (repository_url, ref, PurePosixPath(*subdirectory)) + + def _resolve_tree_ref(self, repository_url: str, segments: list[str]) -> tuple[str, list[str]]: + """Split ``/tree/`` *segments* into ``(ref, subdirectory)``. + + Uses the longest remote branch/tag name that prefixes the segments, so + refs containing ``/`` resolve to the intended tree. Raises ValueError + when no advertised ref matches the URL. + """ + remote_refs = self._list_remote_refs(repository_url) + for end in range(len(segments), 0, -1): + candidate = "/".join(segments[:end]) + if candidate in remote_refs: + return candidate, segments[end:] + raise ValueError( + "GitHub tree URL does not name a known branch or tag: " + f"{repository_url} ({'/'.join(segments)})" + ) + + def _list_remote_refs(self, repository_url: str) -> set[str]: + """Return the branch/tag names advertised by the remote repository. + + Bounded by the ingest deadline; the host allowlist and private-IP + checks from URL validation apply. + """ + self._validate_url_host(repository_url, ALLOWED_GIT_HOSTS) + deadline = self._deadline() + self._check_deadline(deadline, "git") + timeout = max(1.0, deadline - monotonic()) + try: + process = subprocess.run( + ["git", "ls-remote", repository_url], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + timeout=timeout, + check=False, + ) + except subprocess.TimeoutExpired as exc: + raise IngestLimitExceededError("Git ref listing exceeded its time limit") from exc + if process.returncode != 0: + raise ValueError(f"Could not list refs for GitHub tree URL: {repository_url}") + refs: set[str] = set() + for line in process.stdout.decode("utf-8", errors="replace").splitlines(): + _, _, ref = line.partition("\t") + for prefix in ("refs/heads/", "refs/tags/"): + if ref.startswith(prefix): + refs.add(ref[len(prefix) :]) + break + return refs + def _is_file_url(self, path: str) -> bool: """Check if path is a direct file URL.""" if not path.startswith("https://"): @@ -1044,7 +1132,7 @@ def _validate_url_host(self, url: str, allowed_hosts: frozenset[str]) -> str: ) return host - def _clone_git(self, url: str) -> Path: + def _clone_git(self, url: str, *, branch: str | None = None) -> Path: """Clone a Git repository to a temporary directory, bounded by ``INGEST_MAX_BYTES``.""" remaining_seconds = self._remaining_seconds() remaining_bytes = self._remaining_bytes() @@ -1070,6 +1158,8 @@ def _clone_git(self, url: str) -> Path: url, str(clone_dir), ] + if branch is not None: + clone_command[6:6] = ["--branch", branch] if remaining_bytes is not None: clone_command.insert(6, f"--filter=blob:limit={remaining_bytes}") process: subprocess.Popen[bytes] | None = None diff --git a/tests/unit/test_input_handler.py b/tests/unit/test_input_handler.py index e3243a5a6..663274b12 100644 --- a/tests/unit/test_input_handler.py +++ b/tests/unit/test_input_handler.py @@ -19,7 +19,7 @@ import os import sys from errno import ENOENT -from pathlib import Path +from pathlib import Path, PurePosixPath from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -455,6 +455,100 @@ def test_scp_url_is_git_url() -> None: assert InputHandler()._is_git_url("git@github.com:org/repo.git") is True +def test_github_tree_url_resolves_a_checked_out_subdirectory(tmp_path: Path) -> None: + handler = InputHandler() + clone = tmp_path / "repo" + (clone / "skills" / "biome-gritql").mkdir(parents=True) + with ( + patch.object(handler, "_clone_git", return_value=clone) as clone_git, + patch.object(handler, "_list_remote_refs", return_value={"main"}), + ): + resolved, source_type = handler.resolve( + "https://github.com/somtougeh/somto-dev-toolkit/tree/main/skills/biome-gritql" + ) + assert resolved == clone / "skills" / "biome-gritql" + assert source_type == "git" + clone_git.assert_called_once_with( + "https://github.com/somtougeh/somto-dev-toolkit.git", branch="main" + ) + + +def test_github_tree_url_resolves_slash_containing_ref(tmp_path: Path) -> None: + """A branch name containing / must not be split into ref + subdirectory.""" + handler = InputHandler() + clone = tmp_path / "repo" + (clone / "skills" / "demo").mkdir(parents=True) + with ( + patch.object(handler, "_clone_git", return_value=clone) as clone_git, + patch.object(handler, "_list_remote_refs", return_value={"main", "feature", "feature/foo"}), + ): + resolved, source_type = handler.resolve( + "https://github.com/example/repo/tree/feature/foo/skills/demo" + ) + assert resolved == clone / "skills" / "demo" + assert source_type == "git" + clone_git.assert_called_once_with("https://github.com/example/repo.git", branch="feature/foo") + + +def test_github_tree_url_prefers_shorter_ref_when_longest_absent() -> None: + """The longest *advertised* ref wins, not the longest URL prefix.""" + handler = InputHandler() + with patch.object(handler, "_list_remote_refs", return_value={"feature"}): + repository_url, ref, subdirectory = handler._github_tree_target( + "https://github.com/example/repo/tree/feature/sub" + ) + assert repository_url == "https://github.com/example/repo.git" + assert ref == "feature" + assert subdirectory == PurePosixPath("sub") + + +def test_github_tree_url_rejects_unknown_ref() -> None: + handler = InputHandler() + with ( + patch.object(handler, "_list_remote_refs", return_value={"main"}), + pytest.raises(ValueError, match="does not name a known branch or tag"), + ): + handler._github_tree_target("https://github.com/example/repo/tree/nope/sub") + + +def test_github_tree_url_supports_ref_without_subdirectory() -> None: + handler = InputHandler() + with patch.object(handler, "_list_remote_refs", return_value={"main"}): + repository_url, ref, subdirectory = handler._github_tree_target( + "https://github.com/example/repo/tree/main" + ) + assert repository_url == "https://github.com/example/repo.git" + assert ref == "main" + assert subdirectory == PurePosixPath(".") + + +@pytest.mark.parametrize("segment", ["%2Fetc", "%2E%2E%2Frepo", "%5Coutside"]) +def test_github_tree_url_rejects_encoded_path_escapes(segment: str) -> None: + with pytest.raises(ValueError, match="stay within the repository"): + InputHandler()._github_tree_target( + f"https://github.com/example/repo/tree/main/skills/{segment}" + ) + + +@pytest.mark.parametrize("target", ["missing", "SKILL.md"]) +def test_github_tree_url_selection_failure_cleans_owned_clone(tmp_path: Path, target: str) -> None: + """A post-clone tree selection error must not strand the owned checkout.""" + handler = InputHandler() + clone = tmp_path / "repo" + clone.mkdir() + if target == "SKILL.md": + (clone / target).write_text("# skill\n") + handler._temp_dir = tmp_path + with ( + patch.object(handler, "_clone_git", return_value=clone), + patch.object(handler, "_list_remote_refs", return_value={"main"}), + ): + with pytest.raises(ValueError): + handler.resolve(f"https://github.com/example/repo/tree/main/{target}") + assert not tmp_path.exists() + assert handler.temp_dir_for_cleanup() is None + + def test_http_urls_are_not_accepted_as_remote_inputs() -> None: """Network inputs require HTTPS unless they use SSH's scp-style syntax.""" handler = InputHandler()