#!/usr/bin/env python3 from __future__ import annotations import json import logging import os import re import threading import time from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import quote, urlsplit LOG = logging.getLogger("rustbox.repository") REPOSITORY = Path(os.environ.get("REPOSITORY_ROOT", "/srv/repository")) WEB = Path(__file__).resolve().parent / "web" PORT = int(os.environ.get("DASHBOARD_PORT", "8000")) CACHE_SECONDS = max(1, int(os.environ.get("DASHBOARD_CACHE_SECONDS", "15"))) REPOSITORY_URL = os.environ.get("REPOSITORY_URL", "http://localhost:18965").rstrip("/") SITES = ("debian", "rhel", "suse") META_PATHS = {"debian": Path("DEBIAN-BUILD-INFO.txt"), "rhel": Path("RHEL-BUILD-INFO.txt"), "suse": Path("SUSE-BUILD-INFO.txt")} DEB_FILENAME = re.compile(r"^(?P.+?)_(?P[^_]+)_(?P[^_]+)\.deb$") RPM_FILENAME = re.compile(r"^(?P.+)-(?P\d[^-]*)-(?P[^/]+)\.(?Px86_64|amd64|aarch64|noarch|arm64)\.rpm$") def iso_time(stamp: float) -> str: return datetime.fromtimestamp(stamp, tz=timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") def parse_metadata(path: Path) -> dict[str, str]: try: text = path.read_text(encoding="utf-8") except OSError: return {} result: dict[str, str] = {} for line in text.splitlines(): key, sep, value = line.partition("=") if sep and re.fullmatch(r"[a-z][a-z0-9_]*", key): result[key] = value.strip()[:1024] return result def get_current(repository: Path) -> Path | None: try: root = repository.resolve(strict=True) current = (root / "current").resolve(strict=True) if current.is_dir() and current.is_relative_to(root): return current except (OSError, RuntimeError): pass return None def publication_info(current: Path, channel: str) -> dict[str, object]: location = current / META_PATHS[channel] raw = parse_metadata(location) built_at = raw.get("built_at") or None return { "channel": channel, "published": bool(raw), "app_version": raw.get("app_version", ""), "package_release": raw.get("package_release", ""), "package_version": f"{raw.get('app_version','')}-{raw.get('package_release','')}" if raw.get("app_version") and raw.get("package_release") else raw.get("app_version", ""), "source_commit": raw.get("source_commit", ""), "source_ref": raw.get("source_ref", ""), "source_url": raw.get("source_url", ""), "packaging_revision": raw.get("packaging_revision", ""), "built_at": built_at, } def parse_package(path: Path, kind: str) -> tuple[str, str, str] | None: pattern = DEB_FILENAME if kind == "deb" else RPM_FILENAME m = pattern.fullmatch(path.name) if not m: return None if kind == "deb": return m["name"], m["version"], m["arch"] return m["name"], f"{m['version']}-{m['release']}", m["arch"] def generate_catalog(repository: Path = REPOSITORY) -> dict[str, object]: current = get_current(repository) empty = {"published": False, "app_version": "", "package_release": "", "package_version": "", "source_commit": "", "source_ref": "", "source_url": "", "packaging_revision": "", "built_at": None} builds = {site: publication_info(current, site) if current else {"channel": site, **empty} for site in SITES} packages: list[dict[str, object]] = [] if current: scans = [ ("debian", "DEB", "deb", current / "apt" / "pool", "*.deb"), ("rhel", "RPM", "rpm", current / "rpm" / "x86_64", "*.rpm"), ("suse", "RPM", "rpm", current / "suse" / "x86_64", "*.rpm"), ] for channel, label, kind, directory, pattern in scans: if not directory.is_dir(): continue for path in directory.rglob(pattern): parsed = parse_package(path, kind) if not parsed or not path.is_file(): continue name, version, arch = parsed stat = path.stat() rel = path.relative_to(current) encoded = "/".join(quote(part, safe="") for part in rel.parts) packages.append({ "name": name, "version": version, "architecture": arch, "format": label, "channel": channel, "size_bytes": stat.st_size, "built_at": iso_time(stat.st_mtime), "download_url": f"{REPOSITORY_URL}/{encoded}", "relative_path": rel.as_posix(), }) packages.sort(key=lambda item: (str(item["built_at"]), str(item["version"])), reverse=True) build_times = [str(item["built_at"]) for item in builds.values() if item.get("built_at")] return { "generated_at": iso_time(time.time()), "ready": bool(packages), "repository_url": REPOSITORY_URL, "apt_distribution": os.environ.get("REPO_DISTRIBUTION", "stable"), "apt_component": os.environ.get("REPO_COMPONENT", "main"), "builds": builds, "packages": packages, "summary": { "packages_count": len(packages), "channels_published": sum(bool(item.get("published")) for item in builds.values()), "latest_build_at": max(build_times, default=None), }, } class Cache: def __init__(self) -> None: self.lock = threading.Lock() self.expires = 0.0 self.snapshot: dict[str, object] | None = None def get(self) -> dict[str, object]: with self.lock: if self.snapshot is None or time.monotonic() >= self.expires: self.snapshot = generate_catalog() self.expires = time.monotonic() + CACHE_SECONDS return self.snapshot CACHE = Cache() class Handler(BaseHTTPRequestHandler): server_version = "RustBoxRepository/1.0" sys_version = "" def do_GET(self) -> None: path = urlsplit(self.path).path if path == "/healthz": self.send_bytes(b"ok\n", "text/plain; charset=utf-8", "no-store") elif path == "/api/v1/packages": self.send_bytes(json.dumps(CACHE.get(), ensure_ascii=False, separators=(",", ":")).encode(), "application/json; charset=utf-8", "no-store") elif path in ("/", "/index.html"): self.send_bytes((WEB / "index.html").read_bytes(), "text/html; charset=utf-8", "no-store", True) elif path in ("/ui/style.css", "/ui/app.js"): name = path.rsplit("/", 1)[-1] mime = {"style.css": "text/css", "app.js": "application/javascript"}[name] self.send_bytes((WEB / name).read_bytes(), f"{mime}; charset=utf-8", "public, max-age=600") else: self.send_error(HTTPStatus.NOT_FOUND) def do_HEAD(self) -> None: self.do_GET() def send_bytes(self, payload: bytes, mime: str, cache: str, html: bool = False) -> None: self.send_response(HTTPStatus.OK) self.send_header("Content-Type", mime) self.send_header("Content-Length", str(len(payload))) self.send_header("Cache-Control", cache) self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") if html: self.send_header("Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src 'self'; connect-src 'self'; img-src 'self' data:; object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'none'") self.end_headers() if self.command != "HEAD": self.wfile.write(payload) def log_message(self, fmt: str, *args: object) -> None: LOG.info("%s - %s", self.address_string(), fmt % args) def main() -> None: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler) server.daemon_threads = True LOG.info("RustBox repository dashboard ready on %s", PORT) server.serve_forever(poll_interval=0.5) if __name__ == "__main__": main()