"""Fetch a public GitHub file or repo tree by URL, server-side. Master prompt 6. Only github.com and raw.githubusercontent.com. A tool that fetched any URL the model named would be a server-side request forgery waiting for one prompt injection — "fetch http://169.254.169.254/..." — so the host list is closed and the URL is rebuilt from parsed parts rather than passed through. """ import re from dataclasses import dataclass from urllib.parse import urlparse import httpx from .attachments import MAX_FILE_BYTES, is_text_file API = "https://api.github.com" RAW = "https://raw.githubusercontent.com" MAX_TREE_ENTRIES = 1500 _SAFE_PART = re.compile(r"^[A-Za-z0-9._-]+$") class GithubError(ValueError): pass @dataclass(frozen=True) class GithubTarget: owner: str repo: str ref: str | None path: str #: "file" (blob or raw link) or "tree" (repo root or a directory). kind: str def parse(url: str) -> GithubTarget: parsed = urlparse(url.strip()) if parsed.scheme != "https": raise GithubError("only https GitHub links can be fetched") host = parsed.netloc.lower() parts = [part for part in parsed.path.split("/") if part] if host == "raw.githubusercontent.com": if len(parts) < 4: raise GithubError("a raw link needs owner/repo/ref/path") owner, repo, ref, *rest = parts return _checked(GithubTarget(owner, repo, ref, "/".join(rest), "file")) if host not in {"github.com", "www.github.com"}: raise GithubError("only github.com links can be fetched") if len(parts) < 2: raise GithubError("that is not a repository link") owner, repo = parts[0], parts[1].removesuffix(".git") if len(parts) == 2: return _checked(GithubTarget(owner, repo, None, "", "tree")) if len(parts) >= 4 and parts[2] in {"blob", "tree"}: kind = "file" if parts[2] == "blob" else "tree" return _checked(GithubTarget(owner, repo, parts[3], "/".join(parts[4:]), kind)) raise GithubError("link a repository, a file (blob) or a folder (tree)") def _checked(target: GithubTarget) -> GithubTarget: for part in (target.owner, target.repo, *(target.ref or "").split("/")): if part and not _SAFE_PART.match(part): raise GithubError("unexpected characters in the link") if ".." in target.path.split("/"): raise GithubError("unexpected characters in the link") return target async def fetch(http: httpx.AsyncClient, url: str) -> dict: """Returns ``{"kind": "file", "name", "content"}`` or ``{"kind": "tree", "paths": [...]}``.""" target = parse(url) headers = {"Accept": "application/vnd.github+json", "User-Agent": "prova-brain"} if target.kind == "file": if not is_text_file(target.path): raise GithubError("only text and code files can be fetched") ref = target.ref or "HEAD" response = await http.get( f"{RAW}/{target.owner}/{target.repo}/{ref}/{target.path}", headers=headers ) if response.status_code == 404: raise GithubError("file not found, or the repository is private") response.raise_for_status() if len(response.content) > MAX_FILE_BYTES: raise GithubError("that file is larger than 1 MB") return {"kind": "file", "name": target.path, "content": response.text} ref = target.ref or "HEAD" response = await http.get( f"{API}/repos/{target.owner}/{target.repo}/git/trees/{ref}", params={"recursive": "1"}, headers=headers, ) if response.status_code == 404: raise GithubError("repository not found, or it is private") response.raise_for_status() prefix = f"{target.path.rstrip('/')}/" if target.path else "" paths = [ entry["path"] for entry in response.json().get("tree", []) if entry.get("type") == "blob" and entry["path"].startswith(prefix) ] return { "kind": "tree", "repo": f"{target.owner}/{target.repo}", "paths": paths[:MAX_TREE_ENTRIES], "truncated": len(paths) > MAX_TREE_ENTRIES, }