"""Mittwald-API-Client (Port der Logik aus transkript-platform/infra/deploy.py). Sicherheitsnetze wie dort: differenzielles PATCH (nie PUT), Namensraum `app-`, Post-Write-Guard. Zusätzlich: `live=False` (Standard) blockiert JEDEN schreibenden Aufruf – es wird nur geloggt, was passieren würde. Lesende Aufrufe laufen nur, wenn ein Token gesetzt ist. """ from __future__ import annotations import json from typing import Callable import httpx class DeployError(RuntimeError): pass class DryRunSkip(Exception): pass def mask_env(env: dict) -> dict: import re return {k: ("••••" if re.search(r"(SECRET|PASSWORD|KEY|TOKEN)$", k) else v) for k, v in env.items()} class Mittwald: def __init__(self, token: str, base: str, live: bool, log: Callable[[str], None], http: Callable | None = None): self.token, self.base, self.live, self.log = token, base.rstrip("/"), live, log self._http = http or self._real_http self._project_uuid: str | None = None @staticmethod def _real_http(method, url, headers, body): r = httpx.request(method, url, headers=headers, json=body, timeout=60) try: data = r.json() if r.content else None except ValueError: data = None return r.status_code, data, r.text[:500] def api(self, method: str, path: str, body: dict | None = None, *, write: bool, fatal: bool = True): if write and not self.live: self.log(f"[DRY-RUN] {method} {path}") if body is not None: shown = json.loads(json.dumps(body)) for svc in (shown.get("services") or {}).values(): if isinstance(svc, dict) and "environment" in svc: svc["environment"] = mask_env(svc["environment"]) self.log(" body: " + json.dumps(shown, ensure_ascii=False)[:1500]) raise DryRunSkip() if not self.token: if not self.live: self.log(f"(übersprungen: lesender Aufruf {method} {path} – kein MITTWALD_API_TOKEN)") raise DryRunSkip() raise DeployError("MITTWALD_API_TOKEN nicht gesetzt.") status, data, text = self._http(method, f"{self.base}{path}", {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}, body) if status >= 400: msg = f"{method} {path} → HTTP {status}: {text}" if fatal: raise DeployError(msg) self.log(f"WARNUNG (nicht fatal): {msg}") return None return data def try_api(self, *a, **kw): try: return self.api(*a, **kw) except DryRunSkip: return None # ── Stack ── def resolve_stack_id(self, project_id: str, stack_id: str) -> str: if stack_id: return stack_id if not project_id: if not self.live: self.log("(Hinweis: Mittwald PROJECT_ID/STACK_ID sind noch nicht in den Einstellungen gesetzt – Platzhalter im Dry-Run)") return "" raise DeployError("Mittwald PROJECT_ID (oder STACK_ID) fehlt in den Einstellungen.") try: stacks = self.api("GET", f"/projects/{project_id}/stacks/", write=False) except DryRunSkip: return "" items = stacks if isinstance(stacks, list) else (stacks or {}).get("items", []) for s in items: if str(s.get("description") or s.get("name") or "").lower() == "default": return s["id"] if items: return items[0]["id"] raise DeployError("Kein Stack im Projekt gefunden.") def get_stack(self, stack_id: str) -> dict | None: return self.try_api("GET", f"/stacks/{stack_id}/", write=False) @staticmethod def services_of(stack: dict | None) -> dict[str, dict]: if not stack: return {} svcs = stack.get("services") if isinstance(svcs, dict): return svcs if isinstance(svcs, list): return {(s.get("serviceName") or s.get("name") or s.get("id")): s for s in svcs if (s.get("serviceName") or s.get("name") or s.get("id"))} return {} def project_uuid(self, project_id: str) -> str: if self._project_uuid: return self._project_uuid proj = self.try_api("GET", f"/projects/{project_id}/", write=False, fatal=False) self._project_uuid = (proj or {}).get("id") or project_id return self._project_uuid # ── Operationen ── def deploy_service(self, *, stack_id: str, service: str, volume: str, definition: dict, container_port: str, project_id: str, hostname: str | None) -> None: stack = self.get_stack(stack_id) existing = set(self.services_of(stack)) protected = existing - {service} if protected: self.log(f"geschützte bestehende Services (bleiben unangetastet): {sorted(protected)}") self.try_api("PATCH", f"/stacks/{stack_id}/", {"volumes": {volume: {"name": volume}}}, write=True) self.try_api("PATCH", f"/stacks/{stack_id}/", {"services": {service: definition}}, write=True) svc_id = "" if self.live: after = self.services_of(self.get_stack(stack_id)) missing = protected - set(after) if missing: raise DeployError(f"ABBRUCH: bestehende Services verschwunden nach PATCH: {sorted(missing)}") svc_id = after.get(service, {}).get("id", service) self.try_api("POST", f"/stacks/{stack_id}/services/{svc_id}/actions/recreate/", write=True) self.log("Post-Write-Guard: alle vorherigen Services noch vorhanden ✔") if hostname: self.ensure_ingress(project_id, hostname, svc_id, container_port) def ensure_ingress(self, project_id: str, hostname: str, container_id: str, port: str) -> None: listing = self.try_api("GET", f"/projects/{project_id}/ingresses/", write=False, fatal=False) items = listing if isinstance(listing, list) else (listing or {}).get("items", []) if any(i.get("hostname") == hostname for i in items): self.log(f"Ingress {hostname} existiert bereits → ok") return body = {"hostname": hostname, "projectId": self.project_uuid(project_id), "paths": [{"path": "/", "target": {"container": {"id": container_id, "portProtocol": f"{port}/tcp"}}}]} res = self.try_api("POST", "/ingresses", body, write=True, fatal=False) if self.live: self.log(f"Ingress {hostname} angelegt (Auto-TLS)" if res is not None else f"⚠ Ingress {hostname} NICHT angelegt – Token braucht Domain/Ingress-Recht (Service läuft trotzdem).") def offboard_service(self, *, stack_id: str, service: str, volume: str, project_id: str, hostname: str | None) -> None: self.try_api("PATCH", f"/stacks/{stack_id}/", {"services": {service: {}}}, write=True) self.try_api("PATCH", f"/stacks/{stack_id}/", {"volumes": {volume: {}}}, write=True) if hostname: listing = self.try_api("GET", f"/projects/{project_id}/ingresses/", write=False, fatal=False) items = listing if isinstance(listing, list) else (listing or {}).get("items", []) for i in items: if i.get("hostname") == hostname: self.try_api("DELETE", f"/ingresses/{i['id']}", write=True, fatal=False) self.log(f"Ingress {hostname} entfernt")