rust-package-builder/dashboard/app.py

#!/usr/bin/env python3
"""Read-only, live RustProxy package registry UI.

The existing Nginx continues to serve APT, DNF and pkg artifacts unchanged.
This service only generates the dashboard and a public JSON catalog.
"""

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("rustproxy.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")))
MAX_PACKAGES = 5000

SITES = ("linux", "suse", "opnsense", "freebsd")
META_PATHS = {
    "linux": Path("LINUX-BUILD-INFO.txt"),
    "suse": Path("SUSE-BUILD-INFO.txt"),
    "opnsense": Path("opnsense/BUILD-INFO.txt"),
    "freebsd": Path("freebsd/BUILD-INFO.txt"),
}
REPO_URLS = {
    "linux": os.environ.get("REPOSITORY_URL", "http://localhost:18765").rstrip("/"),
    "opnsense": os.environ.get("OPNSENSE_REPOSITORY_URL", "http://localhost:18766").rstrip("/"),
}
REPO_URLS["freebsd"] = REPO_URLS["opnsense"] + "/freebsd"
REPO_URLS["suse"] = REPO_URLS["linux"] + "/suse"

DEB_FILENAME = re.compile(r"^(?P<name>.+?)_(?P<version>[^_]+)_(?P<arch>[^_]+)\.deb$")
RPM_FILENAME = re.compile(
    r"^(?P<name>.+)-(?P<version>\d[^-]*)-(?P<release>[^/]+)\."
    r"(?P<arch>x86_64|amd64|aarch64|noarch|arm64)\.rpm$"
)
PKG_FILENAME = re.compile(r"^(?P<name>.+)-(?P<version>\d[^/]*)\.pkg$")


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:
        # Metadata is emitted by our builder, not submitted through the dashboard.
        text = path.read_text(encoding="utf-8")
    except (FileNotFoundError, PermissionError, OSError):
        return {}
    info: dict[str, str] = {}
    for line in text.splitlines():
        key, separator, value = line.partition("=")
        if separator and re.fullmatch(r"[a-z][a-z0-9_]*", key):
            info[key] = value.strip()[:1024]
    return info


def parse_filename(filename: str, kind: str) -> tuple[str, str, str] | None:
    if kind == "deb":
        m = DEB_FILENAME.fullmatch(filename)
        return (m["name"], m["version"], m["arch"]) if m else None
    if kind == "rpm":
        m = RPM_FILENAME.fullmatch(filename)
        if m:
            return (m["name"], f"{m['version']}-{m['release']}", m["arch"])
        return None
    if filename in ("packagesite.pkg", "data.pkg") or filename.endswith("-latest.pkg"):
        return None
    m = PKG_FILENAME.fullmatch(filename)
    return (m["name"], m["version"], "amd64") if m else None


def publication_info(current: Path, channel: str) -> dict[str, object]:
    location = current / META_PATHS[channel]
    raw = parse_metadata(location)
    info: dict[str, object] = {
        "channel": channel,
        "published": bool(raw),
        "app_version": raw.get("app_version", ""),
        "package_version": raw.get("package_version", ""),
        "package_release": raw.get("package_release", ""),
        "source_commit": raw.get("source_commit", ""),
        "source_ref": raw.get("source_ref", ""),
        "source_url": raw.get("source_url", ""),
        "build_reason": raw.get("build_reason", ""),
        "packaging_revision": raw.get("packaging_revision", ""),
        "opnsense_series": raw.get("opnsense_series", ""),
        "freebsd_abi": raw.get("freebsd_abi", ""),
        "built_at": None,
        "date_source": "unknown",
    }
    if channel in ("linux", "suse") and info["app_version"]:
        info["package_version"] = f"{info['app_version']}-{info['package_release']}" if info["package_release"] else info["app_version"]

    stamp = raw.get("built_at", "")
    if stamp:
        try:
            candidate = datetime.fromisoformat(stamp.replace("Z", "+00:00"))
            if candidate.tzinfo is not None:
                info["built_at"] = candidate.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
                info["date_source"] = "build"
        except ValueError:
            pass
    if info["built_at"] is None and raw:
        # Backward compatibility with builds made before 'built_at' was saved.
        try:
            info["built_at"] = iso_time(location.stat().st_mtime)
            info["date_source"] = "file"
        except OSError:
            pass
    return info


def get_current(repository: Path) -> Path | None:
    try:
        # Never follow a manipulated symlink outside the repository tree.
        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 package_entry(path: Path, current: Path, site: str, kind: str, channel: str,
                  builds: dict[str, dict[str, object]]) -> dict[str, object] | None:
    parsed = parse_filename(path.name, kind)
    if not parsed or not path.is_file():
        return None
    name, version, arch = parsed
    # Discovery agent is packaged for Linux DEB/RPM only; never advertise it as FreeBSD pkg.
    if channel in ('opnsense', 'freebsd') and name == 'rustproxy-docker-agent':
        return None
    try:
        stat = path.stat()
    except OSError:
        return None
    relative = path.relative_to(current)
    if site == "linux":
        public_path = relative
    else:
        public_path = relative.relative_to(site)
    # Each URL segment is encoded independently; separators remain separators.
    encoded_path = "/".join(quote(p, safe="") for p in public_path.parts)
    base = REPO_URLS[site]
    built_at = iso_time(stat.st_mtime)
    source = "file"
    build = builds[channel]
    if build.get("date_source") == "build" and version == build.get("package_version"):
        built_at = str(build["built_at"])
        source = "build"
    return {
        "name": name,
        "version": version,
        "architecture": arch,
        "format": kind.upper() if kind != "pkg" else "PKG",
        "channel": channel,
        "size_bytes": stat.st_size,
        "built_at": built_at,
        "date_source": source,
        "download_url": f"{base}/{encoded_path}",
        "relative_path": relative.as_posix(),
    }


def generate_catalog(repository: Path = REPOSITORY) -> dict[str, object]:
    current = get_current(repository)
    builds: dict[str, dict[str, object]] = {}
    packages: list[dict[str, object]] = []
    for channel in SITES:
        builds[channel] = publication_info(current, channel) if current else {
            "channel": channel, "published": False, "app_version": "", "package_version": "",
            "package_release": "", "source_commit": "", "source_ref": "",
            "source_url": "", "build_reason": "", "packaging_revision": "",
            "opnsense_series": "", "freebsd_abi": "", "built_at": None, "date_source": "unknown",
        }

    if current is not None:
        folders = (
            ("linux", "deb", "linux", current / "apt" / "pool", "*.deb"),
            ("linux", "rpm", "linux", current / "rpm" / "x86_64", "*.rpm"),
            ("suse", "rpm", "suse", current / "suse" / "x86_64", "*.rpm"),
            ("opnsense", "pkg", "opnsense", current / "opnsense", "*.pkg"),
            ("freebsd", "pkg", "freebsd", current / "freebsd", "*.pkg"),
        )
        for site, kind, channel, directory, pattern in folders:
            if not directory.is_dir():
                continue
            try:
                for file in directory.rglob(pattern):
                    result = package_entry(file, current, site, kind, channel, builds)
                    if result:
                        packages.append(result)
            except OSError as e:
                LOG.warning("Cannot scan %s: %s", directory, e)

    packages.sort(key=lambda item: (str(item["built_at"]), str(item["version"])), reverse=True)
    total_packages = len(packages)
    packages = packages[:MAX_PACKAGES]
    channel_counts = {site: sum(p["channel"] == site for p in packages) for site in SITES}
    build_times = [str(b["built_at"]) for b in builds.values() if b["built_at"]]
    return {
        "generated_at": iso_time(time.time()),
        "ready": any(b["published"] for b in builds.values()) or bool(packages),
        "repository_urls": REPO_URLS,
        "apt_distribution": os.environ.get("REPO_DISTRIBUTION", "stable"),
        "apt_component": os.environ.get("REPO_COMPONENT", "main"),
        "builds": builds,
        "packages": packages,
        "summary": {
            "packages_count": total_packages,
            "packages_visible": len(packages),
            "channel_counts": channel_counts,
            "channels_published": sum(bool(b["published"]) for b in builds.values()),
            "latest_build_at": max(build_times, default=None),
        },
    }


class CatalogCache:
    def __init__(self, repository: Path = REPOSITORY, ttl: int = CACHE_SECONDS):
        self.repository = repository
        self.ttl = ttl
        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.repository)
                self._expires = time.monotonic() + self.ttl
            return self._snapshot


CACHE = CatalogCache()


class DashboardHandler(BaseHTTPRequestHandler):
    server_version = "RustProxyRepository/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", cache="no-store")
        elif path == "/api/v1/packages":
            try:
                payload = json.dumps(CACHE.get(), ensure_ascii=False, separators=(",", ":"))
            except Exception:
                LOG.exception("Failed to build package catalog")
                self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR)
                return
            self.send_bytes(payload.encode("utf-8"), "application/json; charset=utf-8", cache="no-store")
        elif path in ("/", "/index.html", "/freebsd/", "/freebsd/index.html"):
            site = self.headers.get("X-Repo-Site", "linux")
            if path.startswith("/freebsd/"):
                site = "freebsd"
            if site not in SITES:
                site = "linux"
            page = (WEB / "index.html").read_text(encoding="utf-8")
            page = page.replace("__REPOSITORY_SITE__", site)
            self.send_bytes(page.encode("utf-8"), "text/html; charset=utf-8", cache="no-store", html=True)
        elif path in ("/ui/style.css", "/ui/app.js", "/ui/favicon.svg"):
            name = path.rsplit("/", 1)[-1]
            mime = {"style.css": "text/css", "app.js": "application/javascript", "favicon.svg": "image/svg+xml"}[name]
            self.send_bytes((WEB / name).read_bytes(), f"{mime}; charset=utf-8", cache="public, max-age=600")
        else:
            self.send_error(HTTPStatus.NOT_FOUND, "Unknown dashboard route")

    def do_HEAD(self) -> None:
        # Nginx allows HEAD and GET; suppress response content for HEAD.
        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")
    httpd = ThreadingHTTPServer(("0.0.0.0", PORT), DashboardHandler)
    httpd.daemon_threads = True
    LOG.info("Dashboard ready on port %s (repository: %s)", PORT, REPOSITORY)
    try:
        httpd.serve_forever(poll_interval=0.5)
    except KeyboardInterrupt:
        pass
    finally:
        httpd.server_close()


if __name__ == "__main__":
    main()