teahub/pages/server.py

"""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/<org>__<repo>__<branch>; `POST /_hook/<org>/<repo>` (a Gitea webhook) or the
periodic loop pulls it. Static files only:
a path maps to the file, a directory to its index.html, anything missing to 404.html or a plain 404. No listing, no
execution, nothing writes except git. stdlib only.
"""
import http.server
import json
import os
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)
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"}
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()):
            print(repo, sync(repo), flush=True)
        time.sleep(PULL_EVERY)


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 site(self) -> str | None:
        host = (self.headers.get("host") or "").split(":")[0].lower()
        return SITES.get(host) or SITES.get(host[4:] if host.startswith("www.") else "")

    def do_POST(self):
        parts = urlsplit(self.path).path.strip("/").split("/")
        hit = [s for s in SITES.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
        repo = self.site()
        if not repo:
            self.send_error(404, "no site for this host")
            return
        root = site_root(repo).resolve()
        rel = unquote(path).lstrip("/")
        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:
            status, target = 404, root / "404.html"
            if not target.is_file():
                self.send_error(404)
                return
        data = target.read_bytes()
        self.send_response(status)
        self.send_header("content-type", TYPES.get(target.suffix.lower(), "application/octet-stream"))
        self.send_header("content-length", str(len(data)))
        self.send_header("cache-control", "public, max-age=60")
        self.send_header("x-content-type-options", "nosniff")
        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()
    http.server.ThreadingHTTPServer(("0.0.0.0", int(os.environ.get("PORT") or 8080)), Handler).serve_forever()