"""Tenant-Bundle-Loader: holt die Tenant-Daten zur Laufzeit von der Admin-Plattform (TENANT_SOURCE=admin). git (Standard, wie bisher) Tenant-Ordner aus dem Image: tenants// admin Bundle von {ADMIN_URL}/api/tenants//bundle (Bearer ADMIN_BUNDLE_TOKEN), geprüft und lokal gecacht Ausfallsicherheit bei admin: Admin nicht erreichbar → letzter gecachter Stand (/data/bundle) → eingebackener Tenant (nur solange BUNDLE_FALLBACK_BAKED nicht „false“ ist) → sonst Fehler beim Start. Die Admin-Plattform ist damit nur Kontrollebene, nicht Teil des Laufzeitpfads. Sicherheit: Hashes aller Dateien werden gegen MANIFEST.json geprüft, nur normale Dateien, keine Pfad-Tricks, Größenlimits. Format siehe docs/CONTRACT.md. Nur Standardbibliothek – wird beim Import von core.config genutzt. """ from __future__ import annotations import hashlib import io import json import os import shutil import tarfile import time import urllib.error import urllib.request from pathlib import Path, PurePosixPath SUPPORTED_BUNDLE_FORMAT = 1 MAX_FILES = 200 MAX_TOTAL_BYTES = 64 * 1024 * 1024 KEEP_VERSIONS = 2 class BundleError(RuntimeError): pass def _truthy(v: str | None, default: bool) -> bool: return default if v is None else v.strip().lower() in ("1", "true", "yes", "on") def _http_fetch(url: str, token: str, etag: str | None, timeout: float = 20.0): """→ (status, headers, body). 304 bei unverändertem ETag. Wirft OSError bei Netzwerkfehlern.""" req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}", **({"If-None-Match": etag} if etag else {})}) try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, dict(r.headers), r.read(MAX_TOTAL_BYTES + 1) except urllib.error.HTTPError as exc: return exc.code, dict(exc.headers or {}), b"" def parse_bundle(data: bytes, tenant_id: str) -> tuple[dict, dict[str, bytes]]: """Prüft ein Bundle und gibt (manifest, {pfad: inhalt}) zurück. Wirft BundleError bei jeder Auffälligkeit.""" if len(data) > MAX_TOTAL_BYTES: raise BundleError("Bundle zu groß") files: dict[str, bytes] = {} total = 0 try: with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tar: members = tar.getmembers() if len(members) > MAX_FILES: raise BundleError(f"Bundle enthält mehr als {MAX_FILES} Dateien") for m in members: if not m.isreg(): raise BundleError(f"Nicht erlaubter Eintrag im Bundle: {m.name!r}") p = PurePosixPath(m.name) if p.is_absolute() or ".." in p.parts or not p.parts or "\\" in m.name: raise BundleError(f"Unzulässiger Pfad im Bundle: {m.name!r}") total += m.size if total > MAX_TOTAL_BYTES: raise BundleError("Bundle entpackt zu groß") files[str(p)] = tar.extractfile(m).read() except (tarfile.TarError, EOFError, OSError) as exc: raise BundleError(f"Bundle nicht lesbar: {exc}") from exc if "MANIFEST.json" not in files or "tenant.json" not in files: raise BundleError("MANIFEST.json oder tenant.json fehlt im Bundle") try: manifest = json.loads(files.pop("MANIFEST.json")) cfg = json.loads(files["tenant.json"]) except ValueError as exc: raise BundleError(f"MANIFEST.json/tenant.json ungültig: {exc}") from exc if int(manifest.get("bundle_format", 1)) > SUPPORTED_BUNDLE_FORMAT: raise BundleError(f"Bundle-Format {manifest.get('bundle_format')} wird von diesem Core nicht unterstützt") if manifest.get("tenant") != tenant_id or cfg.get("id") != tenant_id: raise BundleError(f"Bundle gehört nicht zu Tenant {tenant_id!r}") expected = manifest.get("files") or {} if set(expected) != set(files): raise BundleError("Dateiliste weicht vom MANIFEST ab") for path, sha in expected.items(): if hashlib.sha256(files[path]).hexdigest() != sha: raise BundleError(f"Hash-Abweichung: {path}") return manifest, files def _install(cache: Path, version: int, manifest: dict, files: dict[str, bytes], etag: str | None) -> Path: """Schreibt das Bundle in ein neues Versionsverzeichnis und schaltet `current` atomar um.""" cache.mkdir(parents=True, exist_ok=True) target = cache / f"v{version}-{hashlib.sha256(json.dumps(manifest.get('files', {}), sort_keys=True).encode()).hexdigest()[:8]}" if not target.exists(): tmp = cache / f".tmp-{os.getpid()}-{int(time.time() * 1000)}" for path, content in files.items(): dest = tmp / path dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(content) (tmp / ".bundle.json").write_text(json.dumps({"version": version, "etag": etag, "fetched_at": time.time()})) os.replace(tmp, target) link = cache / "current" new_link = cache / f".current-{os.getpid()}" if new_link.is_symlink() or new_link.exists(): new_link.unlink() os.symlink(target.name, new_link) os.replace(new_link, link) keep = {target.name} for d in sorted((p for p in cache.glob("v*") if p.is_dir()), key=lambda p: p.stat().st_mtime, reverse=True)[:KEEP_VERSIONS]: keep.add(d.name) for d in cache.glob("v*"): if d.is_dir() and d.name not in keep: shutil.rmtree(d, ignore_errors=True) return target def _cached(cache: Path) -> tuple[Path, dict] | None: link = cache / "current" if link.is_symlink() and (link / "tenant.json").exists(): try: meta = json.loads((link / ".bundle.json").read_text()) except (OSError, ValueError): meta = {} return link, meta return None def resolve(tenant_id: str, tenants_dir: Path, env: dict | None = None, fetch=_http_fetch) -> tuple[Path, dict]: """Liefert (Tenant-Verzeichnis, Info). Info: {source: git|admin|cache|baked, version, stale, error, fetched_at}.""" env = os.environ if env is None else env source = (env.get("TENANT_SOURCE") or "git").strip().lower() baked = tenants_dir / tenant_id if source == "git": return baked, {"source": "git", "version": None, "stale": False, "error": None, "fetched_at": None} if source != "admin": raise BundleError(f"Unbekannte TENANT_SOURCE {source!r} (erlaubt: git, admin)") base = (env.get("ADMIN_URL") or "").rstrip("/") token = env.get("ADMIN_BUNDLE_TOKEN") or "" cache = Path(env.get("BUNDLE_CACHE_DIR") or Path(env.get("DATA_DIR") or "/data") / "bundle") / tenant_id have = _cached(cache) error: str | None = None if not base or not token: error = "ADMIN_URL oder ADMIN_BUNDLE_TOKEN nicht gesetzt" else: for attempt in range(3): # kurze Wiederholung: Container und Admin starten oft gleichzeitig try: status, headers, body = fetch(f"{base}/api/tenants/{tenant_id}/bundle", token, (have[1].get("etag") if have else None)) except OSError as exc: error, status = f"Admin nicht erreichbar: {exc}", None else: if status == 200: try: manifest, files = parse_bundle(body, tenant_id) version = int(manifest.get("version") or headers.get("X-Bundle-Version") or 0) etag = {k.lower(): v for k, v in headers.items()}.get("etag") d = _install(cache, version, manifest, files, etag) return d, {"source": "admin", "version": version, "stale": False, "error": None, "fetched_at": time.time()} except BundleError as exc: error = f"Ungültiges Bundle: {exc}" break # kein Retry: Inhalt ist kaputt, nicht das Netz elif status == 304 and have: return have[0], {"source": "admin", "version": have[1].get("version"), "stale": False, "error": None, "fetched_at": have[1].get("fetched_at")} elif status in (401, 403, 404): error = f"Admin verweigert das Bundle (HTTP {status})" break else: error = f"Admin-Antwort HTTP {status}" if attempt < 2: time.sleep(1.5 * (attempt + 1)) if have: return have[0], {"source": "cache", "version": have[1].get("version"), "stale": True, "error": error, "fetched_at": have[1].get("fetched_at")} if _truthy(env.get("BUNDLE_FALLBACK_BAKED"), True) and (baked / "tenant.json").exists(): return baked, {"source": "baked", "version": None, "stale": True, "error": error, "fetched_at": None} raise BundleError(f"Tenant-Bundle für {tenant_id!r} nicht verfügbar und kein Cache/Rückfall vorhanden: {error}")