Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
25 kB · 616 lines
Python
at sv-fe
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617#!/usr/bin/env python3import jsonimport osimport reimport structimport subprocessimport sysimport tempfileimport urllib.parseimport zlibfrom datetime import datetime, timedelta, timezonefrom http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
PORT = int(os.environ.get("PORT", "8000"))CLIENT_ID = os.environ.get("GITHUB_CLIENT_ID", "ghstub-client-id")CLIENT_SECRET = os.environ.get("GITHUB_CLIENT_SECRET", "ghstub-client-secret")APP_SLUG = os.environ.get("GITHUB_APP_SLUG", "tangled-dev-stub")OBJECT_FORMAT = os.environ.get("GITHUB_STUB_OBJECT_FORMAT", "sha1")REPOS_DIR = os.environ.get("REPOS_DIR", os.path.join(tempfile.gettempdir(), "github-stub-repos"))
PUBLIC_ORIGIN = os.environ.get("GITHUB_PUBLIC_ORIGIN", "https://github.com").rstrip("/")API_ORIGIN = os.environ.get("GITHUB_API_ORIGIN", "https://api.github.com").rstrip("/")AVATARS_ORIGIN = os.environ.get("GITHUB_AVATARS_ORIGIN", "https://avatars.githubusercontent.com").rstrip("/")
OWNER = { "login": "octostub", "id": 10001, "avatar_url": f"{AVATARS_ORIGIN}/u/10001", "bio": "stub bio for the import flow", "blog": "https://octostub.example.com",}
# sha256 exercises knot2 object-format negotiation locallySEED = [ ( "hello", "Hello world stub repository", { "README.md": "# Hello\n\nHello world stub repository for Tangled import testing.\n", "hello.txt": "Hello from octostub/hello!\n", }, "sha1", ), ( "world", "World stub repository for bulk import", {"README.md": "# World\n\nWorld stub repository for bulk import testing.\n"}, "sha1", ), ( "tangled-demo", "Tangled demo repository", {"README.md": "# Tangled Demo\n\nDemo repository for tangled import.\n"}, "sha1", ), ( "fresh", "Fresh stub repository, never imported before", { "README.md": "# Fresh\n\nA repository nobody has imported yet.\n", "src/lib.rs": "pub fn fresh() -> &'static str {\n \"fresh\"\n}\n", }, "sha256", ), ( "second", "Second fresh stub repository, for bulk import", { "README.md": "# Second\n\nThe second repository in a bulk import.\n", "notes.md": "two\n", }, "sha1", ), ( "toolkit", "Small tools, for the bulk import", { "README.md": "# Toolkit\n\nTools nobody has imported yet.\n", "tools.md": "hammer\n", }, "sha1", ), ( "notes", "Scratch notes, also unimported", {"README.md": "# Notes\n\nNothing to see here yet.\n", "today.md": "one\n"}, "sha1", ),]
REPO_METADATA = { "hello": {"language": "TypeScript", "stars": 12, "issues": 3, "forks": 2, "size_kb": 21504, "days_ago": 2}, "world": {"language": "Rust", "stars": 340, "issues": 7, "forks": 41, "size_kb": 512, "days_ago": 5}, "tangled-demo": {"language": "Go", "stars": 88, "issues": 1, "forks": 9, "size_kb": 330, "hours_ago": 3}, "fresh": {"language": "Rust", "stars": 0, "issues": 0, "forks": 1, "size_kb": 64, "days_ago": 1}, "second": {"language": "Python", "stars": 2048, "issues": 22, "forks": 190, "size_kb": 20480, "days_ago": 200}, "toolkit": {"language": "Shell", "stars": 5, "issues": 0, "forks": 0, "size_kb": 88, "days_ago": 3}, "notes": {"language": None, "stars": 1, "issues": 0, "forks": 0, "size_kb": 16, "days_ago": 45},}
def _iso_age(**kwargs) -> str: return (datetime.now(timezone.utc) - timedelta(**kwargs)).strftime("%Y-%m-%dT%H:%M:%SZ")
def repository(name: str, description: str, files: dict, object_format: str = "sha1") -> dict: meta = REPO_METADATA[name] pushed_at = _iso_age(hours=meta.get("hours_ago", 0), days=meta.get("days_ago", 0)) return { "id": 101 + [seed[0] for seed in SEED].index(name), "name": name, "full_name": f"octostub/{name}", "owner": OWNER, "html_url": f"{PUBLIC_ORIGIN}/octostub/{name}", "description": description, "clone_url": f"{PUBLIC_ORIGIN}/octostub/{name}.git", "default_branch": "main", "visibility": "public", "private": False, "language": meta["language"], "stargazers_count": meta["stars"], "open_issues_count": meta["issues"], "forks_count": meta["forks"], "size": meta["size_kb"], "pushed_at": pushed_at, "updated_at": pushed_at, "files": files, "object_format": object_format, }
REPOSITORIES = [repository(*seed) for seed in SEED]
STUB_TOKEN_PREFIX = "ghstub-token-"
def email_for(token: str) -> str: """The caller's address. Deliberi refuses one address for two accounts, so every authorization gets its own: the token is derived from the code minted per connect.""" match = re.fullmatch(STUB_TOKEN_PREFIX + r"([0-9a-f]{16})", token) suffix = f"+{match.group(1)[:8]}" if match else "" return f"octostub{suffix}@octostub.dev"
GITHUB_EMAILS = [ {"email": "octostub@octostub.dev", "primary": True, "verified": True, "visibility": "public"}, { "email": "octostub-secondary@octostub.dev", "primary": False, "verified": False, "visibility": "public", }, { "email": "octostub-alt@octostub.dev", "primary": False, "verified": True, "visibility": "public", },]
def _clean_repo(r: dict) -> dict: return {k: v for k, v in r.items() if k not in ("files", "object_format")}
def make_png(width=16, height=16, color=(120, 80, 200)) -> bytes: png = b"\x89PNG\r\n\x1a\n" ihdr_data = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) ihdr_crc = struct.pack(">I", zlib.crc32(b"IHDR" + ihdr_data) & 0xffffffff) png += struct.pack(">I", len(ihdr_data)) + b"IHDR" + ihdr_data + ihdr_crc
raw = bytearray() for _ in range(height): raw.append(0) for _ in range(width): raw.extend(color)
compressed = zlib.compress(bytes(raw)) idat_crc = struct.pack(">I", zlib.crc32(b"IDAT" + compressed) & 0xffffffff) png += struct.pack(">I", len(compressed)) + b"IDAT" + compressed + idat_crc
iend_crc = struct.pack(">I", zlib.crc32(b"IEND") & 0xffffffff) png += struct.pack(">I", 0) + b"IEND" + iend_crc return png
AVATAR_PNG = make_png()
def seed_repositories(): os.makedirs(REPOS_DIR, exist_ok=True) for repo_info in REPOSITORIES: full_name = repo_info["full_name"] bare_path = os.path.join(REPOS_DIR, f"{full_name}.git") if os.path.exists(bare_path): continue os.makedirs(os.path.dirname(bare_path), exist_ok=True) fmt = repo_info.get("object_format") or OBJECT_FORMAT with tempfile.TemporaryDirectory() as tmp_dir: subprocess.run( ["git", "init", f"--object-format={fmt}", tmp_dir], check=True, capture_output=True, ) subprocess.run(["git", "-C", tmp_dir, "checkout", "-B", "main"], check=True, capture_output=True) for fname, content in repo_info["files"].items(): fpath = os.path.join(tmp_dir, fname) os.makedirs(os.path.dirname(fpath), exist_ok=True) with open(fpath, "w") as f: f.write(content) subprocess.run(["git", "-C", tmp_dir, "config", "user.name", "Octo Stub"], check=True, capture_output=True) subprocess.run(["git", "-C", tmp_dir, "config", "user.email", "octostub@octostub.dev"], check=True, capture_output=True) subprocess.run(["git", "-C", tmp_dir, "add", "."], check=True, capture_output=True) subprocess.run(["git", "-C", tmp_dir, "commit", "-m", "Initial commit"], check=True, capture_output=True) subprocess.run(["git", "clone", "--bare", tmp_dir, bare_path], check=True, capture_output=True) subprocess.run(["git", "-C", bare_path, "symbolic-ref", "HEAD", "refs/heads/main"], check=True, capture_output=True) subprocess.run(["git", "-C", bare_path, "config", "http.receivepack", "false"], check=True, capture_output=True) desc_path = os.path.join(bare_path, "description") with open(desc_path, "w") as f: f.write(repo_info["description"] + "\n") print(f"Seeded bare repository ({fmt}): {bare_path}", flush=True)
class GitHubStubHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1"
def log_message(self, format, *args): sys.stderr.write(f"[{self.log_date_time_string()}] {self.address_string()} {format % args}\n") sys.stderr.flush()
def read_body(self) -> bytes: if self.headers.get("Transfer-Encoding", "").lower() == "chunked": chunks = [] while True: line = self.rfile.readline().strip() if not line: break chunk_len = int(line.split(b";")[0], 16) if chunk_len == 0: while self.rfile.readline().strip(): pass break chunk = self.rfile.read(chunk_len) chunks.append(chunk) self.rfile.readline() return b"".join(chunks) length = int(self.headers.get("Content-Length", 0)) return self.rfile.read(length) if length > 0 else b""
def send_json(self, status: int, data: any, link: str = ""): body = json.dumps(data, indent=2).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") if link: self.send_header("Link", link) self.end_headers() self.wfile.write(body)
def page_link(self, path: str, query: dict, page: int, per_page: int, total: int) -> str: if page * per_page >= total: return "" params = {k: v[0] for k, v in query.items()} params.update({"page": str(page + 1), "per_page": str(per_page)}) nxt = urllib.parse.urlencode(params) return f'<{API_ORIGIN}{path}?{nxt}>; rel="next"'
def send_avatar(self): self.send_response(200) self.send_header("Content-Type", "image/png") self.send_header("Content-Length", str(len(AVATAR_PNG))) self.send_header("Cache-Control", "public, max-age=86400") self.end_headers() self.wfile.write(AVATAR_PNG)
def do_GET(self): parsed = urllib.parse.urlparse(self.path) path = parsed.path.rstrip("/") if not path: path = "/" query = urllib.parse.parse_qs(parsed.query)
if path.startswith("/u/") or path.startswith("/avatar") or path.startswith("/avatars/") or path == "/user/avatar": self.send_avatar() return
git_refs_match = re.match(r"^/([^/]+)/([^/]+?)(?:\.git)?/info/refs$", parsed.path) if git_refs_match: owner, repo_name = git_refs_match.group(1), git_refs_match.group(2) service = query.get("service", [""])[0] if service != "git-upload-pack": self.send_error(400, "Unsupported git service") return repo_path = os.path.join(REPOS_DIR, owner, f"{repo_name}.git") if not os.path.exists(repo_path): self.send_error(404, "Repository not found") return env = os.environ.copy() proto = self.headers.get("Git-Protocol") if proto: env["GIT_PROTOCOL"] = proto env["GIT_CONFIG_NOSYSTEM"] = "1" proc = subprocess.run( ["git", "upload-pack", "--stateless-rpc", "--advertise-refs", repo_path], capture_output=True, env=env ) if proc.returncode != 0: self.send_error(500, f"git upload-pack failed: {proc.stderr.decode('utf-8', errors='replace')}") return body = b"001e# service=git-upload-pack\n0000" + proc.stdout self.send_response(200) self.send_header("Content-Type", "application/x-git-upload-pack-advertisement") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-cache, no-store, must-revalidate") self.send_header("Pragma", "no-cache") self.end_headers() self.wfile.write(body) return
if path == "/login/oauth/authorize": redirect_uri = query.get("redirect_uri", [""])[0] state = query.get("state", [""])[0] code = "ghstub-" + os.urandom(8).hex() if not redirect_uri: self.send_error(400, "Missing redirect_uri") return sep = "&" if "?" in redirect_uri else "?" dest = f"{redirect_uri}{sep}code={code}&state={urllib.parse.quote(state)}" self.send_response(302) self.send_header("Location", dest) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", "0") self.end_headers() return
if re.match(r"^/apps/[^/]+/installations/new$", path): setup = os.environ.get( "GITHUB_APP_SETUP_URL", "http://127.0.0.1:5174/_internal/github/install/callback", ) sep = "&" if "?" in setup else "?" state = query.get("state", [""])[0] dest = f"{setup}{sep}installation_id=1&setup_action=install" if state: dest += f"&state={urllib.parse.quote(state)}" self.send_response(302) self.send_header("Location", dest) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", "0") self.end_headers() return
if path == "/stub/installed": html = ( b"<!DOCTYPE html><html><head><title>GitHub App Installed</title></head>" b"<body><h1>GitHub App Installed</h1>" b"<p>The Tangled GitHub App has been installed on your stub account.</p>" b"</body></html>" ) self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(html))) self.end_headers() self.wfile.write(html) return
if path == "/user": token = self.headers.get("Authorization", "").removeprefix("Bearer ").strip() user_data = { "login": "octostub", "id": 10001, "node_id": "MDQ6VXNlcjEwMDAx", "avatar_url": f"{AVATARS_ORIGIN}/u/10001", "gravatar_id": "", "url": f"{API_ORIGIN}/users/octostub", "html_url": f"{PUBLIC_ORIGIN}/octostub", "type": "User", "site_admin": False, "name": "Octo Stub", "company": None, "blog": "https://octostub.example.com", "location": None, "email": email_for(token), "hireable": None, "bio": "Stub GitHub user for local development", "public_repos": len(REPOSITORIES), "public_gists": 0, "followers": 0, "following": 0, "created_at": "2024-01-01T00:00:00Z", "updated_at": "2024-01-01T00:00:00Z" } self.send_json(200, user_data) return
if path == "/user/emails": token = self.headers.get("Authorization", "").removeprefix("Bearer ").strip() address = email_for(token) local_suffix = address[len("octostub") : address.index("@")] self.send_json( 200, [ {**row, "email": row["email"].replace("@octostub.dev", "").replace("octostub", "octostub" + local_suffix) + "@octostub.dev"} for row in GITHUB_EMAILS ], ) return
if path == "/user/keys": keys = [ { "id": 1, "key": "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOr0s3gZf7W/8vV/EXAMPLEEXAMPLEEXAMPLEEXAMPLE octostub@octostub.dev", "title": "Stub Key" } ] self.send_json(200, keys) return
if path == "/user/installations": installations_data = { "total_count": 1, "installations": [ { "id": 1, "account": { "login": "octostub", "id": 10001, "type": "User", "avatar_url": f"{AVATARS_ORIGIN}/u/10001" }, "app_id": 1, "app_slug": APP_SLUG, "target_id": 10001, "target_type": "User", "permissions": { "metadata": "read", "contents": "read" }, "events": [], "repository_selection": "all" } ] } self.send_json(200, installations_data) return
inst_repos_match = re.match(r"^/user/installations/(\d+)/repositories$", path) if inst_repos_match: clean_repos = [_clean_repo(r) for r in REPOSITORIES] page = max(1, int(query.get("page", ["1"])[0])) per_page = max(1, min(100, int(query.get("per_page", ["30"])[0]))) self.send_json(200, { "total_count": len(clean_repos), "repositories": clean_repos[(page - 1) * per_page : page * per_page] }, self.page_link(path, query, page, per_page, len(clean_repos))) return
if path == "/user/repos": visibility = query.get("visibility", [""])[0] affiliation = query.get("affiliation", ["owner,collaborator,organization_member"])[0] page = max(1, int(query.get("page", ["1"])[0])) per_page = max(1, min(100, int(query.get("per_page", ["30"])[0]))) owned = "owner" in {a for a in affiliation.split(",") if a} rows = [] if not owned else [ _clean_repo(r) for r in REPOSITORIES if not visibility or r.get("visibility") == visibility ] self.send_json(200, rows[(page - 1) * per_page : page * per_page], self.page_link(path, query, page, per_page, len(rows))) return
users_repos_match = re.match(r"^/users/([^/]+)/repos$", path) if users_repos_match: login = users_repos_match.group(1) page = max(1, int(query.get("page", ["1"])[0])) per_page = max(1, min(100, int(query.get("per_page", ["30"])[0]))) rows = [ _clean_repo(r) for r in REPOSITORIES if r.get("visibility") == "public" and r.get("full_name", "").startswith(f"{login}/") ] self.send_json(200, rows[(page - 1) * per_page : page * per_page], self.page_link(path, query, page, per_page, len(rows))) return
repo_match = re.match(r"^/repos/([^/]+)/([^/]+)$", path) if repo_match: owner, name = repo_match.group(1), repo_match.group(2) full_name = f"{owner}/{name}" for r in REPOSITORIES: if r["full_name"] == full_name: self.send_json(200, _clean_repo(r)) return self.send_json(404, {"message": "Not Found"}) return
if path == "/": self.send_json(200, { "status": "ok", "service": "github-stub", "client_id": CLIENT_ID, "app_slug": APP_SLUG, "repositories": [r["full_name"] for r in REPOSITORIES] }) return
self.send_json(404, {"message": f"Not Found: {path}"})
def do_POST(self): parsed = urllib.parse.urlparse(self.path) path = parsed.path.rstrip("/") if not path: path = "/"
git_pack_match = re.match(r"^/([^/]+)/([^/]+?)(?:\.git)?/git-upload-pack$", parsed.path) if git_pack_match: owner, repo_name = git_pack_match.group(1), git_pack_match.group(2) repo_path = os.path.join(REPOS_DIR, owner, f"{repo_name}.git") if not os.path.exists(repo_path): self.send_error(404, "Repository not found") return
body = self.read_body() content_encoding = self.headers.get("Content-Encoding", "").lower() if content_encoding == "gzip": body = zlib.decompress(body, 16 + zlib.MAX_WBITS) elif content_encoding == "deflate": body = zlib.decompress(body)
env = os.environ.copy() proto = self.headers.get("Git-Protocol") if proto: env["GIT_PROTOCOL"] = proto env["GIT_CONFIG_NOSYSTEM"] = "1"
proc = subprocess.Popen( ["git", "upload-pack", "--stateless-rpc", repo_path], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env ) stdout, stderr = proc.communicate(input=body) if proc.returncode != 0: self.send_error(500, f"git upload-pack failed: {stderr.decode('utf-8', errors='replace')}") return
self.send_response(200) self.send_header("Content-Type", "application/x-git-upload-pack-result") self.send_header("Content-Length", str(len(stdout))) self.send_header("Cache-Control", "no-cache, no-store, must-revalidate") self.send_header("Pragma", "no-cache") self.end_headers() self.wfile.write(stdout) return
if path == "/login/oauth/access_token": raw_body = self.read_body() content_type = self.headers.get("Content-Type", "").lower()
params = {} if "application/json" in content_type: try: params = json.loads(raw_body.decode("utf-8")) except Exception: pass else: try: parsed_qs = urllib.parse.parse_qs(raw_body.decode("utf-8")) params = {k: v[0] for k, v in parsed_qs.items()} except Exception: pass
token_data = { "access_token": STUB_TOKEN_PREFIX + (str(params.get("code", "")).removeprefix("ghstub-") or "0123456789abcdef")[:16], "token_type": "bearer", "scope": "repo,read:user,user:email" }
accept = self.headers.get("Accept", "").lower() if "application/json" in accept or "application/json" in content_type: self.send_json(200, token_data) else: body = urllib.parse.urlencode(token_data).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/x-www-form-urlencoded") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) return
self.send_json(404, {"message": f"Not Found: {path}"})
def run(): print(f"Starting GitHub Stub on port {PORT}...", flush=True) seed_repositories() server = ThreadingHTTPServer(("0.0.0.0", PORT), GitHubStubHandler) print(f"GitHub Stub listening on 0.0.0.0:{PORT}", flush=True) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close()
if __name__ == "__main__": run()