201 lines
8.2 KiB
Python
201 lines
8.2 KiB
Python
#!/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:18765").rstrip("/")
|
|
SITES = ("linux", "suse")
|
|
META_PATHS = {"linux": Path("LINUX-BUILD-INFO.txt"), "suse": Path("SUSE-BUILD-INFO.txt")}
|
|
DEB_FILENAME = re.compile(r"^(?P<name>.+?)_(?P<version>[^_]+)_(?P<arch>[^_]+)\.deb$")
|
|
RPM_FILENAME = re.compile(r"^(?P<name>.+)-(?P<version>\d[^-]*)-(?P<release>[^/]+)\.(?P<arch>x86_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 = [
|
|
("linux", "DEB", "deb", current / "apt" / "pool", "*.deb"),
|
|
("linux", "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", "/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", "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()
|