import re from datetime import datetime, timedelta, timezone from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi.responses import RedirectResponse, StreamingResponse from anode_client import anode_request, anode_stream from auth import get_current_user from db import get_database_connection from encryption import decrypt_psk, encrypt_psk from templating import templates router = APIRouter() def _dateikomponente(text: str) -> str: """ Bereinigt einen Textbaustein fuer den Einsatz in einem Download- Dateinamen (Leerzeichen/Umlaute/Sonderzeichen -> Bindestrich) - live gefunden: der Download hiess immer nur 'tuxflotte-installationsmedium.iso', bei mehreren Bau-Durchlaeufen/Workspaces kaum auseinanderzuhalten. """ bereinigt = re.sub(r"[^A-Za-z0-9]+", "-", text).strip("-") return bereinigt or "unbekannt" def fetch_konfiguration(organization_id: str) -> dict | None: with get_database_connection() as conn: with conn.cursor() as cur: cur.execute( """ SELECT wifi_ssid, wifi_psk_encrypted, bereitstellungsvorlage_id, max_devices, validity_days, enrollment_session_id, iso_build_id FROM installationsmedium_konfiguration WHERE organization_id = %s """, (organization_id,), ) row = cur.fetchone() if row is None: return None (wifi_ssid, wifi_psk_encrypted, bereitstellungsvorlage_id, max_devices, validity_days, enrollment_session_id, iso_build_id) = row # psycopg liefert TEXT-Spalten hier teils als bytes zurueck, nicht str - # dasselbe bekannte, nicht deterministische Verhalten wie in # auth.py:_row_to_user() (live gefunden: b'...' erschien roh im # WLAN-SSID-Feld auf /installationsmedium). if isinstance(wifi_ssid, bytes): wifi_ssid = wifi_ssid.decode("utf-8") return { "wifi_ssid": wifi_ssid, "hat_wifi_psk": wifi_psk_encrypted is not None, "bereitstellungsvorlage_id": str(bereitstellungsvorlage_id) if bereitstellungsvorlage_id else None, "max_devices": max_devices, "validity_days": validity_days, "enrollment_session_id": str(enrollment_session_id) if enrollment_session_id else None, "iso_build_id": str(iso_build_id) if iso_build_id else None, } def speichere_konfiguration_basis( organization_id: str, wifi_ssid: str, wifi_psk: str, bereitstellungsvorlage_id: str, max_devices: int, validity_days: int, ) -> None: # Leeres PSK-Feld laesst ein zuvor gespeichertes PSK unangetastet (siehe # installationsmedium.html - das Feld zeigt nie den Klartext an, nur # explizites Neueintippen ueberschreibt es). psk_encrypted = encrypt_psk(wifi_psk) if wifi_psk else None with get_database_connection() as conn: with conn.cursor() as cur: if wifi_psk: cur.execute( """ INSERT INTO installationsmedium_konfiguration (organization_id, wifi_ssid, wifi_psk_encrypted, bereitstellungsvorlage_id, max_devices, validity_days, updated_at) VALUES (%s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP) ON CONFLICT (organization_id) DO UPDATE SET wifi_ssid = EXCLUDED.wifi_ssid, wifi_psk_encrypted = EXCLUDED.wifi_psk_encrypted, bereitstellungsvorlage_id = EXCLUDED.bereitstellungsvorlage_id, max_devices = EXCLUDED.max_devices, validity_days = EXCLUDED.validity_days, updated_at = CURRENT_TIMESTAMP """, (organization_id, wifi_ssid, psk_encrypted, bereitstellungsvorlage_id, max_devices, validity_days), ) else: cur.execute( """ INSERT INTO installationsmedium_konfiguration (organization_id, wifi_ssid, bereitstellungsvorlage_id, max_devices, validity_days, updated_at) VALUES (%s, %s, %s, %s, %s, CURRENT_TIMESTAMP) ON CONFLICT (organization_id) DO UPDATE SET wifi_ssid = EXCLUDED.wifi_ssid, bereitstellungsvorlage_id = EXCLUDED.bereitstellungsvorlage_id, max_devices = EXCLUDED.max_devices, validity_days = EXCLUDED.validity_days, updated_at = CURRENT_TIMESTAMP """, (organization_id, wifi_ssid, bereitstellungsvorlage_id, max_devices, validity_days), ) def setze_aktuellen_build(organization_id: str, enrollment_session_id: str, iso_build_id: str) -> None: with get_database_connection() as conn: with conn.cursor() as cur: cur.execute( """ UPDATE installationsmedium_konfiguration SET enrollment_session_id = %s, iso_build_id = %s WHERE organization_id = %s """, (enrollment_session_id, iso_build_id, organization_id), ) def lade_gespeichertes_psk(organization_id: str) -> str | None: with get_database_connection() as conn: with conn.cursor() as cur: cur.execute( "SELECT wifi_psk_encrypted FROM installationsmedium_konfiguration WHERE organization_id = %s", (organization_id,), ) row = cur.fetchone() return decrypt_psk(row[0]) if row else None def finde_enrollment_session(organization_id: str, session_id: str) -> dict | None: result = anode_request("GET", f"/api/v1/organizations/{organization_id}/enrollment-sessions") sessions = result.get("enrollment_sessions", []) if result.get("success") else [] return next((s for s in sessions if s["id"] == session_id), None) def _iso_dateiname(vorlage: dict | None, iso_build: dict | None) -> str: """ Sprechender Dateiname statt immer 'tuxflotte-installationsmedium.iso' - von installationsmedium_ansicht() (fuer die dd-Anleitung) UND installationsmedium_download() (fuer den tatsaechlichen Content- Disposition-Header) gemeinsam genutzt, damit beide garantiert densel ben Namen zeigen. """ if vorlage: backend_teil = _dateikomponente(f"{vorlage['backend']['name']}-{vorlage['backend']['version']}") workspace_teil = _dateikomponente(vorlage["workspace"]["name"]) else: backend_teil = "unbekannt" workspace_teil = "unbekannt" datum_teil = ((iso_build or {}).get("finished_at") or datetime.now(timezone.utc).isoformat())[:10] return f"tuxflotte-{backend_teil}-{workspace_teil}-{datum_teil}.iso" @router.get("/installationsmedium") def installationsmedium_ansicht(request: Request, user=Depends(get_current_user)): organization_id = str(user.organization_id) konfiguration = fetch_konfiguration(organization_id) vorlagen_result = anode_request("GET", f"/api/v1/organizations/{organization_id}/bereitstellungsvorlagen") vorlagen = vorlagen_result.get("bereitstellungsvorlagen", []) if vorlagen_result.get("success") else [] enrollment_session = None iso_build = None iso_dateiname = None if konfiguration and konfiguration["enrollment_session_id"]: enrollment_session = finde_enrollment_session(organization_id, konfiguration["enrollment_session_id"]) if konfiguration and konfiguration["iso_build_id"]: build_result = anode_request("GET", f"/api/v1/iso-builds/{konfiguration['iso_build_id']}") iso_build = build_result.get("iso_build") if build_result.get("success") else None if konfiguration and iso_build and iso_build["status"] == "completed": vorlage = next((v for v in vorlagen if v["id"] == konfiguration["bereitstellungsvorlage_id"]), None) iso_dateiname = _iso_dateiname(vorlage, iso_build) return templates.TemplateResponse( request, "installationsmedium.html", { "user": user, "konfiguration": konfiguration, "vorlagen": vorlagen, "enrollment_session": enrollment_session, "iso_build": iso_build, "iso_dateiname": iso_dateiname, }, ) @router.post("/installationsmedium") def installationsmedium_erstellen( wifi_ssid: str = Form(default=""), wifi_psk: str = Form(default=""), bereitstellungsvorlage_id: str = Form(...), max_devices: int = Form(...), validity_days: int = Form(...), user=Depends(get_current_user), ): organization_id = str(user.organization_id) speichere_konfiguration_basis( organization_id, wifi_ssid, wifi_psk, bereitstellungsvorlage_id, max_devices, validity_days ) # Nur fuer den ISO-Bau gebraucht, nie ueber diesen Aufruf hinaus # gespeichert (siehe Phase-4-Design von provisioning-server) - bei # leerem Feld wird das zuvor gespeicherte PSK weiterverwendet. psk_fuer_build = wifi_psk or (lade_gespeichertes_psk(organization_id) or "") expires_at = (datetime.now(timezone.utc) + timedelta(days=validity_days)).isoformat() session_result = anode_request( "POST", f"/api/v1/organizations/{organization_id}/enrollment-sessions", json={ "bereitstellungsvorlage_id": bereitstellungsvorlage_id, "max_devices": max_devices, "expires_at": expires_at, }, ) if not session_result.get("success"): raise HTTPException(status_code=400, detail=session_result.get("message", "Enrollment Session konnte nicht erstellt werden.")) session_id = session_result["enrollment_session"]["id"] build_result = anode_request( "POST", f"/api/v1/organizations/{organization_id}/iso-builds", json={"activation_code": session_id, "wifi_ssid": wifi_ssid, "wifi_psk": psk_fuer_build}, ) if not build_result.get("success"): raise HTTPException(status_code=400, detail=build_result.get("message", "ISO-Bau konnte nicht gestartet werden.")) setze_aktuellen_build(organization_id, session_id, build_result["iso_build"]["id"]) return RedirectResponse("/installationsmedium", status_code=303) @router.post("/installationsmedium/aufladen") def installationsmedium_aufladen( additional_devices: int = Form(default=0), additional_days: int = Form(default=0), user=Depends(get_current_user), ): organization_id = str(user.organization_id) konfiguration = fetch_konfiguration(organization_id) if not konfiguration or not konfiguration["enrollment_session_id"]: raise HTTPException(status_code=404, detail="Es wurde noch kein Installationsmedium erstellt.") session_id = konfiguration["enrollment_session_id"] neues_expires_at = None if additional_days > 0: aktuelle_session = finde_enrollment_session(organization_id, session_id) if aktuelle_session is None: raise HTTPException(status_code=404, detail="Enrollment Session wurde nicht gefunden.") aktuelles_ablaufdatum = datetime.fromisoformat(aktuelle_session["expires_at"]) neues_expires_at = (aktuelles_ablaufdatum + timedelta(days=additional_days)).isoformat() result = anode_request( "PATCH", f"/api/v1/enrollment-sessions/{session_id}", json={"additional_devices": additional_devices, "expires_at": neues_expires_at}, ) if not result.get("success"): raise HTTPException(status_code=400, detail=result.get("message", "Aufladen fehlgeschlagen.")) return RedirectResponse("/installationsmedium", status_code=303) @router.get("/installationsmedium/download") def installationsmedium_download(user=Depends(get_current_user)): organization_id = str(user.organization_id) konfiguration = fetch_konfiguration(organization_id) if not konfiguration or not konfiguration["iso_build_id"]: raise HTTPException(status_code=404, detail="Es wurde noch kein Installationsmedium erstellt.") build_id = konfiguration["iso_build_id"] status_result = anode_request("GET", f"/api/v1/iso-builds/{build_id}") iso_build = status_result.get("iso_build") if status_result.get("success") else None # Besitz-Validierung wie bei Geraeten (ADR-0011): iso_build_id kommt # zwar aus der eigenen Konfigurationszeile, aber sicherheitshalber noch # einmal explizit gegenpruefen statt der gespeicherten Kennung blind zu # vertrauen. if iso_build is None or iso_build["organization_id"] != organization_id: raise HTTPException(status_code=404, detail="Installationsmedium wurde nicht gefunden.") if iso_build["status"] != "completed": raise HTTPException(status_code=409, detail="Das Installationsmedium ist noch nicht fertig gebaut.") vorlagen_result = anode_request("GET", f"/api/v1/organizations/{organization_id}/bereitstellungsvorlagen") vorlagen = vorlagen_result.get("bereitstellungsvorlagen", []) if vorlagen_result.get("success") else [] vorlage = next((v for v in vorlagen if v["id"] == konfiguration["bereitstellungsvorlage_id"]), None) dateiname = _iso_dateiname(vorlage, iso_build) def stream_datei(): with anode_stream("GET", f"/api/v1/iso-builds/{build_id}/download") as response: response.raise_for_status() for chunk in response.iter_bytes(chunk_size=1024 * 1024): yield chunk return StreamingResponse( stream_datei(), media_type="application/octet-stream", headers={"Content-Disposition": f'attachment; filename="{dateiname}"'}, )