void-installer/tools/xbps-proxy.py

#!/usr/bin/env python3
"""
XBPS caching HTTP reverse proxy.
Cache-first, web fallback. Caches .xbps packages forever; never caches
-repodata files (must stay fresh for xbps to see new package versions).

Usage:
    XBPS_CACHE_DIR=./cache/xbps-pkgs \\
    XBPS_UPSTREAM=https://repo-default.voidlinux.org \\
    XBPS_PROXY_PORT=3142 \\
    python3 tools/xbps-proxy.py
"""
import http.server
import socket
import time
import urllib.request
import urllib.error
import os
import sys
from pathlib import Path

# Force IPv4 only — many hosts have no working IPv6 route to voidlinux mirrors,
# which surfaces as `[Errno 101] Network is unreachable` mid-download. Pin
# socket.getaddrinfo to AF_INET so urllib never tries an AAAA record.
_orig_getaddrinfo = socket.getaddrinfo


def _ipv4_only(host, port, family=0, type=0, proto=0, flags=0):  # noqa: A002
    return _orig_getaddrinfo(host, port, socket.AF_INET, type, proto, flags)


socket.getaddrinfo = _ipv4_only

UPSTREAM  = os.environ.get("XBPS_UPSTREAM",   "https://repo-default.voidlinux.org")
CACHE_DIR = Path(os.environ.get("XBPS_CACHE_DIR", "./cache/xbps-pkgs")).resolve()
PORT      = int(os.environ.get("XBPS_PROXY_PORT", "3142"))
BIND      = os.environ.get("XBPS_PROXY_BIND",  "0.0.0.0")

# Suffixes that must always be fetched fresh (small metadata files).
ALWAYS_FRESH = ("-repodata",)

CACHE_DIR.mkdir(parents=True, exist_ok=True)


def should_cache(path: str) -> bool:
    for suffix in ALWAYS_FRESH:
        if path.endswith(suffix):
            return False
    return True


class Handler(http.server.BaseHTTPRequestHandler):

    def do_GET(self):
        self._handle(head_only=False)

    def do_HEAD(self):
        self._handle(head_only=True)

    def _handle(self, head_only: bool):
        path  = self.path.lstrip("/")
        if not path:
            self.send_error(404, "empty path")
            return
        fpath = CACHE_DIR / path
        hit   = should_cache(path) and fpath.is_file()

        if hit:
            self._serve_cached(fpath, head_only)
            print(f"[HIT ] {path}", flush=True)
            return

        upstream_url = f"{UPSTREAM}/{path}"
        print(f"[MISS] {path}  <-  {upstream_url}", flush=True)
        last_exc = None
        for attempt in range(3):
            try:
                self._fetch_upstream(path, fpath, upstream_url, head_only)
                return
            except urllib.error.HTTPError as exc:
                print(f"[ERR ] {path}  HTTP {exc.code}", flush=True)
                try:
                    self.send_error(exc.code, exc.reason)
                except Exception:  # noqa: BLE001
                    pass
                return
            except (urllib.error.URLError, OSError) as exc:
                last_exc = exc
                if attempt < 2:
                    print(f"[WARN] {path}  attempt {attempt + 1} failed: {exc}; retrying", flush=True)
                    time.sleep(1 + attempt)
        print(f"[ERR ] {path}  {last_exc}", flush=True)
        try:
            self.send_error(503, str(last_exc))
        except Exception:  # noqa: BLE001
            pass

    def _fetch_upstream(self, path, fpath, upstream_url, head_only):
        req = urllib.request.Request(upstream_url)
        req.get_method = lambda: "HEAD" if head_only else "GET"
        with urllib.request.urlopen(req, timeout=120) as resp:
            ctype = resp.headers.get("Content-Type", "application/octet-stream")
            clen  = resp.headers.get("Content-Length")
            self.send_response(200)
            self.send_header("Content-Type", ctype)
            if clen:
                self.send_header("Content-Length", clen)
            self.end_headers()
            if head_only:
                return

            cache_this = should_cache(path)
            tmp = None
            tmp_fh = None
            if cache_this:
                fpath.parent.mkdir(parents=True, exist_ok=True)
                tmp = fpath.with_suffix(fpath.suffix + ".tmp")
                tmp_fh = open(tmp, "wb")
            try:
                while True:
                    chunk = resp.read(64 * 1024)
                    if not chunk:
                        break
                    self.wfile.write(chunk)
                    if tmp_fh is not None:
                        tmp_fh.write(chunk)
            finally:
                if tmp_fh is not None:
                    tmp_fh.close()
            if cache_this and tmp is not None and tmp.exists() and not fpath.is_dir():
                tmp.rename(fpath)

    def _serve_cached(self, fpath: Path, head_only: bool):
        size = fpath.stat().st_size
        self.send_response(200)
        self.send_header("Content-Type", "application/octet-stream")
        self.send_header("Content-Length", str(size))
        self.end_headers()
        if head_only:
            return
        with open(fpath, "rb") as fh:
            while True:
                chunk = fh.read(64 * 1024)
                if not chunk:
                    break
                self.wfile.write(chunk)

    # Silence default access log — we do our own.
    def log_message(self, *_):
        pass


if __name__ == "__main__":
    server = http.server.ThreadingHTTPServer((BIND, PORT), Handler)
    print(f"XBPS proxy  :  {BIND}:{PORT}  ->  {UPSTREAM}", flush=True)
    print(f"Cache dir   :  {CACHE_DIR}", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nProxy stopped.", flush=True)