"""Read-only GitHub repository explorer over pinned, verified Git trees. No OAuth tokens or repository file content are persisted here. All supplied GitHub API requests are read-only, bounded, path-validated and evidence-scoped. """ from __future__ import annotations import re from urllib.parse import quote SHA = re.compile(r"^[a-f0-9]{40}$") REF = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._/-]{0,120}$") MAX_ENTRIES = 20000 MAX_HISTORY_PAGES = 5 class ExplorerError(ValueError): pass def safe_path(path, *, allow_root=True): value = str(path or "").strip() if value in ("", "/") and allow_root: return "" if (not value or len(value) > 220 or value.startswith("/") or any(part in ("", ".", "..") for part in value.split("/")) or re.search(r"[\x00-\x1f\\?#%*{}\[\]<>|]", value)): raise ExplorerError("Invalid relative GitHub path") return value def safe_branch(ref): value = str(ref or "").strip() if not REF.fullmatch(value) or any(part in ("", ".", "..") for part in value.split("/")): raise ExplorerError("Invalid GitHub branch or ref") return value def pinned_tree(api, token, repo, branch): branch = safe_branch(branch) revision = api(token, "/repos/" + repo + "/commits/" + quote(branch, safe="")) sha = revision.get("sha") if isinstance(revision, dict) else None if not isinstance(sha, str) or not SHA.fullmatch(sha): raise ExplorerError("Source revision cannot be verified") body = api(token, "/repos/" + repo + "/git/trees/" + sha + "?recursive=1") if not isinstance(body, dict) or not isinstance(body.get("tree"), list): raise ExplorerError("Repository tree could not be verified") rows = body["tree"] complete = not body.get("truncated") and len(rows) <= MAX_ENTRIES if not complete: raise ExplorerError("Repository tree is incomplete; cannot report exact totals") items = [] for item in rows: if not isinstance(item, dict): continue path = item.get("path") kind = item.get("type") if not isinstance(path, str) or len(path) > 220 or kind not in ("blob", "tree"): continue try: safe_path(path, allow_root=False) except ExplorerError: continue size = item.get("size") items.append({"path": path, "type": "file" if kind == "blob" else "dir", "sizeBytes": max(0, int(size or 0)) if kind == "blob" and isinstance(size, (int, float)) else None, "sha": str(item.get("sha") or "") if SHA.fullmatch(str(item.get("sha") or "")) else ""}) return {"repository": repo, "branch": branch, "sourceSha": sha, "complete": True, "items": items} def scope_stats(snapshot, path=""): root = safe_path(path) entries = snapshot["items"] if root and not any(x["path"] == root and x["type"] == "dir" for x in entries): raise ExplorerError("Directory not present in verified source tree") prefix = root + "/" if root else "" contained = [x for x in entries if x["path"].startswith(prefix)] files = [x for x in contained if x["type"] == "file"] dirs = [x for x in contained if x["type"] == "dir"] measured = [x["sizeBytes"] for x in files if isinstance(x["sizeBytes"], int)] known = len(measured) == len(files) return {"path": root, "recursiveFiles": len(files), "recursiveDirectories": len(dirs), "gitBlobBytes": sum(measured) if known else None, "sizeComplete": known} def directory(snapshot, path=""): root = safe_path(path) stats = scope_stats(snapshot, root) prefix = root + "/" if root else "" candidates = [x for x in snapshot["items"] if x["path"].startswith(prefix)] children = [x for x in candidates if "/" not in x["path"][len(prefix):]] enriched = [] for child in children: if child["type"] == "dir": sub = scope_stats(snapshot, child["path"]) enriched.append({**child, "name": child["path"].split("/")[-1], "sizeBytes": sub["gitBlobBytes"], "sizeComplete": sub["sizeComplete"], "fileCount": sub["recursiveFiles"], "directoryCount": sub["recursiveDirectories"]}) else: enriched.append({**child, "name": child["path"].split("/")[-1], "sizeComplete": isinstance(child["sizeBytes"], int), "fileCount": None, "directoryCount": None}) enriched.sort(key=lambda x: (x["type"] != "dir", x["name"].casefold(), x["path"])) return {key: snapshot[key] for key in ("repository", "branch", "sourceSha", "complete")} | { "path": root, "stats": stats, "entries": enriched, "count": len(enriched), "measuredAtSource": True, "sizeLabel": "Git blob bytes, excluding Git history, Git LFS payloads and checkout overhead"} def find_files(snapshot, term, path="", limit=50): scope = safe_path(path) term = str(term or "").strip().casefold() if len(term) < 2 or len(term) > 100 or re.search(r"[\x00-\x1f\\?#%*{}\[\]<>|]", term): raise ExplorerError("Search requires a valid filename term (2–100 characters)") prefix = scope + "/" if scope else "" found = [x for x in snapshot["items"] if x["type"] == "file" and x["path"].startswith(prefix) and (term in x["path"].split("/")[-1].casefold() or term in x["path"].casefold())] found.sort(key=lambda x: (term not in x["path"].split("/")[-1].casefold(), x["path"].casefold())) return {"repository": snapshot["repository"], "branch": snapshot["branch"], "sourceSha": snapshot["sourceSha"], "query": term, "path": scope, "total": len(found), "truncatedResults": len(found) > limit, "matches": found[:limit]} def history(api, token, repo, branch, path): path = safe_path(path, allow_root=False) branch = safe_branch(branch) records = [] fully_scanned = False for page in range(1, MAX_HISTORY_PAGES + 1): route = ("/repos/" + repo + "/commits?sha=" + quote(branch, safe="") + "&path=" + quote(path, safe="") + "&per_page=100&page=" + str(page)) chunk = api(token, route) if not isinstance(chunk, list): raise ExplorerError("GitHub did not return a commit history list") for item in chunk: commit = item.get("commit") if isinstance(item, dict) else None if not isinstance(commit, dict): continue person = commit.get("committer") or commit.get("author") or {} date = person.get("date") if isinstance(person, dict) else None ident = str(item.get("sha") or "") if SHA.fullmatch(ident): records.append({"sha": ident, "date": str(date or "")[:40], "summary": str(commit.get("message") or "").split("\n")[0][:240]}) if len(chunk) < 100: fully_scanned = True break return {"repository": repo, "branch": branch, "path": path, "lastCommit": records[0] if records else None, "firstSeenPathCommit": records[-1] if records and fully_scanned else None, "firstSeenComplete": fully_scanned, "commitsScanned": len(records), "caution": "Git path history is not a filesystem creation timestamp; renames may hide earlier history."} def comparison(api, token, repo, base, head): base = safe_branch(base) head = safe_branch(head) record = api(token, "/repos/" + repo + "/compare/" + quote(base, safe="") + "..." + quote(head, safe="") + "?per_page=30") if not isinstance(record, dict) or not isinstance(record.get("files"), list): raise ExplorerError("GitHub did not return a verifiable comparison") files = [] for entry in record["files"][:30]: if not isinstance(entry, dict): continue name = str(entry.get("filename") or "")[:220] try: safe_path(name, allow_root=False) except ExplorerError: continue files.append({"path": name, "status": str(entry.get("status") or "")[:40], "additions": int(entry.get("additions") or 0), "deletions": int(entry.get("deletions") or 0), "patch": str(entry.get("patch") or "")[:10000]}) return {"repository": repo, "base": base, "head": head, "baseSha": str((record.get("base_commit") or {}).get("sha") or ""), "headSha": str((record.get("head_commit") or {}).get("sha") or ""), "totalCommits": int(record.get("total_commits") or 0), "files": files, "partial": len(record["files"]) >= 30, "caution": "Up to 30 files and 10,000 patch characters each; not a complete merge review."}