"""Teahub pages: serve each org's public `site` repo by hostname. SITES maps hostnames to `org/repo`, or `org/repo#branch/subdir`, on GITEA (public repos only; cloned over https, no token). A clone lives under /data/____; `POST /_hook//` (a Gitea webhook) or the periodic loop pulls it. WILDCARDS maps a zone to a site whose top-level directories are its subdomains: `{"yolopreneur.lol": "teahubxyz/time#gh-pages/ventures"}` serves `.yolopreneur.lol` from `ventures//`, and `yolopreneur.lol//` redirects there. Static files only: a path maps to the file, a directory to its index.html, anything missing to 404.html or a plain 404. A `_headers` file at a site's root adds response headers, Cloudflare Pages style (a path line, exact or ending in `*`, then indented `Name: value` lines; a later rule wins for the same header). No listing, no execution, nothing writes except git. stdlib only. """ import http.server import json import os import re import subprocess import threading import time from pathlib import Path from urllib.parse import unquote, urlsplit GITEA = os.environ.get("GITEA_URL", "https://teahub.xyz").rstrip("/") DATA = Path(os.environ.get("PAGES_DATA", "/data")) SITES = json.loads(os.environ.get("PAGES_SITES") or "{}") # {"persimmons.app": "persimmons/site", ...} PULL_EVERY = int(os.environ.get("PAGES_PULL_SECONDS") or 300) WILDCARDS = json.loads(os.environ.get("PAGES_WILDCARDS") or "{}") # {"zone": "org/repo#branch/dir"}: .zone -> dir/ TYPES = {".html": "text/html; charset=utf-8", ".css": "text/css; charset=utf-8", ".js": "text/javascript; charset=utf-8", ".mjs": "text/javascript; charset=utf-8", ".json": "application/json", ".svg": "image/svg+xml", ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".gif": "image/gif", ".webp": "image/webp", ".ico": "image/x-icon", ".txt": "text/plain; charset=utf-8", ".md": "text/markdown; charset=utf-8", ".xml": "application/xml", ".woff2": "font/woff2", ".woff": "font/woff", ".pdf": "application/pdf", ".webmanifest": "application/manifest+json", ".mp4": "video/mp4", ".webm": "video/webm", ".mp3": "audio/mpeg", ".m4a": "audio/mp4", ".wav": "audio/wav", ".zip": "application/zip"} LOCKS: dict[str, threading.Lock] = {} def parse(site: str) -> tuple[str, str, str]: """'org/repo#branch/sub/dir' -> (repo, branch, subdir); branch '' means the default branch.""" repo, _, rest = site.partition("#") branch, _, sub = rest.partition("/") return repo, branch, sub.strip("/") def clone_dir(site: str) -> Path: repo, branch, _ = parse(site) return DATA / (repo.replace("/", "__") + (f"__{branch}" if branch else "")) def site_root(site: str) -> Path: return clone_dir(site) / parse(site)[2] def sync(site: str) -> str: """Clone or fast-forward one site repo. Returns the short HEAD or an error line.""" repo, branch, _ = parse(site) d = clone_dir(site) lock = LOCKS.setdefault(site, threading.Lock()) with lock: try: if (d / ".git").exists(): subprocess.run(["git", "-C", str(d), "fetch", "-q", "--depth", "1", "origin"], check=True, capture_output=True, timeout=120) subprocess.run(["git", "-C", str(d), "reset", "-q", "--hard", f"origin/{branch}" if branch else "origin/HEAD"], check=True, capture_output=True, timeout=60) else: subprocess.run(["git", "clone", "-q", "--depth", "1", *(["--branch", branch] if branch else []), f"{GITEA}/{repo}.git", str(d)], check=True, capture_output=True, timeout=180) return subprocess.run(["git", "-C", str(d), "rev-parse", "--short", "HEAD"], capture_output=True, text=True).stdout.strip() except subprocess.CalledProcessError as e: return f"error: {e.stderr.decode(errors='replace')[-200:]}" except (OSError, subprocess.TimeoutExpired) as e: return f"error: {e}" def loop() -> None: while True: for repo in set(SITES.values()) | set(WILDCARDS.values()): print(repo, sync(repo), flush=True) time.sleep(PULL_EVERY) def extra_headers(root: Path, path: str) -> dict[str, str]: """Headers from `/_headers` whose rule matches `path`.""" try: lines = (root / "_headers").read_text().splitlines() except OSError: return {} rules: list[tuple[str, list[tuple[str, str]]]] = [] for line in lines: if not line.strip() or line.lstrip().startswith("#"): continue if not line[0].isspace(): rules.append((line.strip(), [])) elif rules and ":" in line: name, _, value = line.strip().partition(":") rules[-1][1].append((name.strip(), value.strip())) out: dict[str, tuple[str, str]] = {} for pattern, headers in rules: if pattern == path or (pattern.endswith("*") and path.startswith(pattern[:-1])): for name, value in headers: out[name.lower()] = (name, value) return dict(out.values()) class Handler(http.server.BaseHTTPRequestHandler): server_version = "teahub-pages" def log_message(self, fmt, *args): # one line per request, no client address print(f"{self.headers.get('host', '-')} {self.command} {self.path.split('?')[0]} {args[1] if len(args) > 1 else ''}", flush=True) def host(self) -> str: return (self.headers.get("host") or "").split(":")[0].lower() def site(self) -> tuple[str, str] | None: """(site, subdir) for this request's host: an exact site, its www, or a wildcard subdomain.""" host = self.host() s = SITES.get(host) or SITES.get(host[4:] if host.startswith("www.") else "") if s: return s, "" sub, _, zone = host.partition(".") w = WILDCARDS.get(zone) if w and sub and sub != "www" and "/" not in sub and ".." not in sub: return w, sub return None def do_POST(self): parts = urlsplit(self.path).path.strip("/").split("/") hit = [s for s in [*SITES.values(), *WILDCARDS.values()] if len(parts) == 3 and parts[0] == "_hook" and parse(s)[0] == f"{parts[1]}/{parts[2]}"] if hit: self.rfile.read(int(self.headers.get("content-length") or 0)) outs = [sync(s) for s in dict.fromkeys(hit)] # every site on that repo (any branch) body = "\n".join(outs).encode() self.send_response(200 if not any(o.startswith("error") for o in outs) else 502) self.send_header("content-type", "text/plain"); self.send_header("content-length", str(len(body))); self.end_headers() self.wfile.write(body) return self.send_error(404) def do_HEAD(self): self.do_GET(head=True) def do_GET(self, head: bool = False): path = urlsplit(self.path).path if path == "/healthz": body = json.dumps({"ok": True, "sites": {h: site_root(r).is_dir() for h, r in SITES.items()}}).encode() self.send_response(200); self.send_header("content-type", "application/json"); self.send_header("content-length", str(len(body))); self.end_headers() if not head: self.wfile.write(body) return found = self.site() if not found: self.send_error(404, "no site for this host") return repo, sub = found root = site_root(repo).resolve() rel = unquote(path).lstrip("/") # on a wildcard zone's apex, //... lives at .zone: send people there first = rel.split("/", 1)[0] # only folders that are a venture page (have an index.html); assets folders, .well-known etc. stay on the apex if not sub and self.host() in WILDCARDS and first and (root / first / "index.html").is_file() and ".." not in rel: self.send_response(301); self.send_header("location", f"https://{first}.{self.host()}/" + rel[len(first):].lstrip("/")); self.end_headers() return if sub: root = (root / sub).resolve() if not root.is_dir(): self.send_error(404, "no such venture") return target = (root / rel).resolve() if rel else root if root not in target.parents and target != root: self.send_error(404) return if target.is_dir(): if rel and not path.endswith("/"): self.send_response(301); self.send_header("location", path + "/"); self.end_headers() return target = target / "index.html" status = 200 if not target.is_file() or ".git" in target.parts or target.name == "_headers": status, target = 404, root / "404.html" if not target.is_file(): self.send_error(404) return data = target.read_bytes() ctype = TYPES.get(target.suffix.lower(), "application/octet-stream") lead = data[:64].lstrip().lower() # not `head`: that name is the HEAD-request flag if not target.suffix and (lead.startswith(b"= size or lo > hi: self.send_response(416); self.send_header("content-range", f"bytes */{size}"); self.send_header("content-length", "0"); self.end_headers() return hi = min(hi, size - 1) status, span, data = 206, f"bytes {lo}-{hi}/{size}", data[lo:hi + 1] self.send_response(status) self.send_header("content-type", ctype) self.send_header("content-length", str(len(data))) self.send_header("accept-ranges", "bytes") if span: self.send_header("content-range", span) if not any(n.lower() == "cache-control" for n in extra): self.send_header("cache-control", "public, max-age=60") self.send_header("x-content-type-options", "nosniff") for name, value in extra.items(): if name.lower() not in ("content-type", "content-length", "x-content-type-options"): self.send_header(name, value) self.end_headers() if not head: self.wfile.write(data) if __name__ == "__main__": DATA.mkdir(parents=True, exist_ok=True) threading.Thread(target=loop, daemon=True).start() bind = os.environ.get("PAGES_HOST") or "0.0.0.0" # 127.0.0.1 for a private copy (hb-5) http.server.ThreadingHTTPServer((bind, int(os.environ.get("PORT") or 8080)), Handler).serve_forever()