#!/usr/bin/env python3 import hashlib import io import json import os import re import subprocess import tarfile import threading import time from collections import defaultdict from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path REPO = Path(os.getenv("REPO_DIR", "./repo")).resolve() HOST = os.getenv("HOST", "0.0.0.0") PORT = int(os.getenv("PORT", "8478")) POLL_INTERVAL = float(os.getenv("POLL_INTERVAL", "3600")) STABLE_SECONDS = float(os.getenv("STABLE_SECONDS", "60")) RETENTION = int(os.getenv("RETENTION", "10")) LATEST_NAMES = { x.strip() for x in os.getenv("LATEST_NAMES", "latest.pkg").split(",") if x.strip() } CATALOG_FILES = {"meta.conf", "data.pkg", "packagesite.pkg"} LOCK = threading.Lock() _seen = {} def log(msg): print(time.strftime("%Y-%m-%d %H:%M:%S"), msg, flush=True) def sha256(path): h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def read_manifest(pkg): for member in ("+COMPACT_MANIFEST", "+MANIFEST"): p = subprocess.run( ["bsdtar", "-xOf", str(pkg), member], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False, ) if p.returncode == 0 and p.stdout: return json.loads(p.stdout.decode("utf-8")) raise RuntimeError(f"{pkg.name}: brak +COMPACT_MANIFEST / +MANIFEST") def safe_filename(value): # Zachowuje typowe znaki wersji FreeBSD, usuwa tylko znaki niebezpieczne dla sciezki. value = str(value).strip().replace("/", "_").replace("\\", "_") value = re.sub(r"[\x00-\x1f]", "_", value) return value def is_payload_pkg(path): return ( path.is_file() and path.suffix == ".pkg" and path.name not in CATALOG_FILES and not path.name.endswith(".tmp.pkg") ) def stable(path): try: st = path.stat() except FileNotFoundError: return False key = str(path) state = (st.st_size, st.st_mtime_ns) now = time.monotonic() old = _seen.get(key) if old is None or old[:2] != state: _seen[key] = (*state, now) return False return now - old[2] >= STABLE_SECONDS def normalize_latest(): changed = False for path in sorted(REPO.rglob("*.pkg")): if path.name not in LATEST_NAMES or not stable(path): continue try: manifest = read_manifest(path) name = safe_filename(manifest["name"]) version = safe_filename(manifest["version"]) target = path.with_name(f"{name}-{version}.pkg") if target == path: continue if target.exists(): if sha256(target) == sha256(path): path.unlink() log(f"[ingest] {path.name}: wersja {name}-{version} juz istnieje, usunieto duplikat") else: os.replace(path, target) log(f"[ingest] {path.name} -> {target.name} (nadpisano ta sama wersje)") else: os.replace(path, target) log(f"[ingest] {path.name} -> {target.name}") _seen.pop(str(path), None) changed = True except Exception as e: log(f"[ingest] BLAD {path.name}: {e}") return changed def collect_packages(): packages = [] for path in sorted(REPO.rglob("*.pkg")): if not is_payload_pkg(path): continue if path.name in LATEST_NAMES: # latest.pkg nie trafia do indeksu przed normalizacja. continue if not stable(path): continue manifest = read_manifest(path) manifest["path"] = path.relative_to(REPO).as_posix() manifest["pkgsize"] = path.stat().st_size manifest["sum"] = sha256(path) packages.append((path, manifest)) return packages def enforce_retention(packages): if RETENTION <= 0: return False groups = defaultdict(list) for path, manifest in packages: groups[manifest.get("name", path.stem)].append((path, manifest)) removed = False for name, entries in groups.items(): entries.sort(key=lambda item: item[0].stat().st_mtime_ns, reverse=True) for path, manifest in entries[RETENTION:]: try: path.unlink() _seen.pop(str(path), None) log(f"[retention] usunieto {path.name} ({name}, limit={RETENTION})") removed = True except FileNotFoundError: pass return removed def make_tar(dst, member_name, payload): tmp = dst.with_name(dst.name + ".tmp") with tarfile.open(tmp, "w", format=tarfile.PAX_FORMAT) as tf: info = tarfile.TarInfo(member_name) info.size = len(payload) info.mode = 0o644 info.mtime = int(time.time()) tf.addfile(info, io.BytesIO(payload)) os.replace(tmp, dst) def rebuild(): with LOCK: packages = collect_packages() if enforce_retention(packages): packages = collect_packages() manifests = [m for _, m in packages] packagesite = b"".join( json.dumps(m, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + b"\n" for m in manifests ) # Repo v2. packagesite.pkg pozostaje dla zgodnosci ze starszymi klientami. data = json.dumps( {"groups": [], "expired_packages": [], "packages": manifests}, separators=(",", ":"), ensure_ascii=False, ).encode("utf-8") make_tar(REPO / "packagesite.pkg", "packagesite.yaml", packagesite) make_tar(REPO / "data.pkg", "data", data) meta = ( 'version: 1;\n' 'packing_format: "tar";\n' 'manifests: "packagesite.yaml";\n' 'manifests_archive: "packagesite";\n' 'data: "data";\n' 'data_archive: "data";\n' 'filesite: "files";\n' 'filesite_archive: "files";\n' ) tmp = REPO / "meta.conf.tmp" tmp.write_text(meta, encoding="utf-8") os.replace(tmp, REPO / "meta.conf") log(f"[repo] przebudowano katalog: {len(manifests)} pakiet(ow)") def payload_fingerprint(): items = [] for path in sorted(REPO.rglob("*.pkg")): if not is_payload_pkg(path): continue try: st = path.stat() items.append((path.relative_to(REPO).as_posix(), st.st_size, st.st_mtime_ns)) except FileNotFoundError: pass return tuple(items) def watcher(): previous = None while True: try: normalize_latest() current = payload_fingerprint() if current != previous: # collect_packages() wpusci tylko pliki stabilne przez STABLE_SECONDS. time.sleep(STABLE_SECONDS) normalize_latest() rebuild() previous = payload_fingerprint() except Exception as e: log(f"[repo] BLAD przebudowy: {e}") time.sleep(POLL_INTERVAL) class Handler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(REPO), **kwargs) def end_headers(self): self.send_header("Cache-Control", "no-cache, no-store, must-revalidate") super().end_headers() def log_message(self, fmt, *args): log("[http] " + (fmt % args)) def main(): REPO.mkdir(parents=True, exist_ok=True) rebuild() threading.Thread(target=watcher, daemon=True).start() server = ThreadingHTTPServer((HOST, PORT), Handler) log(f"[http] repo: http://{HOST}:{PORT}/ katalog={REPO}") server.serve_forever() if __name__ == "__main__": main()