Spaces:
Running
Running
| """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 | |
| 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, | |
| } | |