from pathlib import Path
from datetime import datetime, timedelta, timezone
import hashlib
import hmac
import json
import mimetypes
import os
import re
import secrets
import subprocess
import threading
import time
import unicodedata
import urllib.error
import urllib.request
from urllib.parse import urljoin, urlparse
from uuid import UUID, uuid4
from typing import Literal
import psycopg
from psycopg.types.json import Jsonb
from cryptography.fernet import Fernet, InvalidToken
from fastapi import FastAPI, Header, HTTPException, Request
from pydantic import BaseModel
from fastapi.responses import FileResponse
LOG_PATH = Path("activation.log")
DATABASE_URL = os.environ.get("TUXFLOTTE_DATABASE_URL")
# Standard 6h (26.08.2026-Planung) - Jitter kommt bewusst nicht von hier,
# sondern aus dem Agenten selbst (agent.py), der pro Geraet zufaellig
# streut, damit nicht alle Geraete exakt gleichzeitig einchecken.
AGENT_POLL_INTERVAL_SECONDS = int(os.environ.get("TUXFLOTTE_AGENT_POLL_INTERVAL", str(6 * 60 * 60)))
# Debug-Modus: kurzes Intervall zum Testen, faellt nach DEBUG_MODE_DURATION_HOURS
# automatisch auf AGENT_POLL_INTERVAL_SECONDS zurueck - rein durch Zeitvergleich
# bei jedem Checkin, kein Cron/Aufraeum-Job noetig (siehe agent_checkin()).
DEBUG_MODE_POLL_INTERVAL_SECONDS = 60
DEBUG_MODE_DURATION_HOURS = 6
ANSIBLE_CONTENT_REPO = os.environ.get("TUXFLOTTE_ANSIBLE_CONTENT_REPO", "")
ADMIN_TOKEN = os.environ.get("TUXFLOTTE_ADMIN_TOKEN", "")
KUNDENPLATTFORM_TOKEN = os.environ.get("TUXFLOTTE_KUNDENPLATTFORM_TOKEN", "")
AGENT_REPORTABLE_EVENT_TYPES = {"applied", "apply_failed", "removed", "remove_failed"}
# Name der pro Organisation automatisch angelegten Standard-OE (siehe
# Konfigurationsgruppen-Plan) - Landeplatz fuer neu aktivierte Geraete.
STANDARD_OE_NAME = "Neue Geräte"
# Phase 4 des Self-Service-ISO-Plans (siehe tuxflotte-installer/scripts/
# build_customer_iso.sh). TUXFLOTTE_INSTALLER_DIR ist ein separat auf anode
# geklontes Repo (nicht Teil von provisioning-server selbst), BOOT_MEDIUM_PATH
# das eigene, distributionsunabhaengige Boot-Medium (build_boot_medium.sh +
# publish_boot_medium.sh), aus dem JEDE Kunden-ISO gebaut wird - egal ob Mint,
# Debian oder Ubuntu. Hiess frueher MINT_ISO_PATH (aus der Zeit, als die
# Mint-Live-ISO gepatcht wurde); die alte Variable gilt als Rueckfallebene.
TUXFLOTTE_INSTALLER_DIR = os.environ.get("TUXFLOTTE_INSTALLER_DIR", "/root/tuxflotte-installer")
# Absolut aufloesen: run_iso_build() startet build_customer_iso.sh mit
# cwd=TUXFLOTTE_INSTALLER_DIR, ein relativer Pfad wuerde also dort statt
# relativ zum eigenen Arbeitsverzeichnis (WorkingDirectory des Service)
# gesucht - genau das hat den ersten Live-Test auf anode fehlschlagen lassen.
BOOT_MEDIUM_PATH = str(Path(
os.environ.get("TUXFLOTTE_BOOT_MEDIUM_PATH")
or os.environ.get("TUXFLOTTE_MINT_ISO_PATH")
or "data/upstream/tuxflotte-boot-medium.iso"
).resolve())
# Aus demselben Grund wie BOOT_MEDIUM_PATH absolut aufloesen: der als $2 an
# build_customer_iso.sh uebergebene Ausgabepfad wird von xorriso relativ
# zum Subprocess-cwd (TUXFLOTTE_INSTALLER_DIR) interpretiert, nicht relativ
# zum eigenen Arbeitsverzeichnis.
ISO_BUILD_OUTPUT_DIR = Path(os.environ.get("TUXFLOTTE_ISO_BUILD_OUTPUT_DIR", "data/iso-builds")).resolve()
# Golden Images (ADR-0024): bewusst unauthentifiziert, analog zu
# /installers/fedora-workstation/ks.cfg - werden von backend_launch()
# per curl geladen, bevor das Zielgeraet irgendein Token besitzt. Kein
# personalisierter/kundenspezifischer Inhalt (anders als ISO-Builds oben),
# nur ein generisches Basis-Root-Dateisystem pro Distribution.
GOLDEN_IMAGES_DIR = Path(os.environ.get("TUXFLOTTE_GOLDEN_IMAGES_DIR", "data/golden-images")).resolve()
# Webapp-Favicons (Phase 2): serverseitig beim Anlegen/Bearbeiten einer
# Webapp geholt und hier abgelegt (siehe fetch_favicon()), unauthentifiziert
# ausgeliefert wie die Golden Images - der Agent laedt sie vor jedem
# ansible-pull-Lauf, kennt aber (anders als Golden Images) keinen
# Aktivierungs-Kontext dafuer.
WEBAPP_ICONS_DIR = Path(os.environ.get("TUXFLOTTE_WEBAPP_ICONS_DIR", "data/webapp-icons")).resolve()
# Merkmal-Icons (Phase 6, 11.09.2026) - eigenes Verzeichnis statt
# WEBAPP_ICONS_DIR mitzunutzen, obwohl der Mechanismus fast identisch ist
# (fetch_merkmal_icon()/_write_merkmal_icon() unten): Merkmale sind
# Katalog-Stammdaten (per Migration angelegt), Webapps sind kundeneigene
# Laufzeitdaten - getrennte Verzeichnisse halten das auseinander.
MERKMAL_ICONS_DIR = Path(os.environ.get("TUXFLOTTE_MERKMAL_ICONS_DIR", "data/merkmal-icons")).resolve()
SIMPLE_ICONS_CDN_URL = "https://cdn.simpleicons.org"
# At-Rest-Verschluesselung fuer Merkmal-Optionen (10.09.2026, Vorbereitung
# AD/OIDC/Fileserver-Merkmale mit echten Zugangsdaten) - gleiches Muster
# wie kundenplattform/encryption.py fuer den WLAN-PSK, aber eigener
# Schluessel/eigenes Modul: dort ist es ein einmaliger Transfer beim
# ISO-Bau, hier muss der Server bei JEDEM Agent-Checkin entschluesseln
# koennen, andere Betriebsanforderung. Erzeugen wie dort:
# python3 -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
ENCRYPTION_KEY = os.environ.get("TUXFLOTTE_ENCRYPTION_KEY", "")
# Namenskonvention statt Allowlist pro Merkmal: jeder optionen-Key, der auf
# diese Endungen passt, wird transparent verschluesselt/entschluesselt -
# neue Merkmale mit einem Zugangsdaten-Feld muessen dafuer nur ihren
# optionen-Key entsprechend benennen (z.B. "ad_join_passwort"), keine
# Aenderung an dieser Stelle noetig.
SENSITIVE_OPTIONEN_SUFFIXES = ("_passwort", "_password", "_secret", "_token")
ISO_BUILD_TIMEOUT_SECONDS = 1800
# ISO-Ablauf nach 7 Tagen (Migration 0021): ergaenzt die bestehende
# "neuer Build ersetzt alten"-Aufraeumung um eine reine Zeitdimension.
ISO_BUILD_EXPIRY_DAYS = 7
ISO_BUILD_EXPIRY_SWEEP_INTERVAL_SECONDS = 60 * 60
app = FastAPI(title="Provisioning Activation Server", version="0.1.0")
@app.on_event("startup")
def start_iso_build_expiry_sweep() -> None:
threading.Thread(target=run_iso_build_expiry_sweep, daemon=True).start()
class ActivationRequest(BaseModel):
activation_code: str
device_fingerprint: str
hostname: str | None = None
machine_id: str | None = None
client_version: str | None = None
hardware: dict
class CreateActivationCodeRequest(BaseModel):
bezeichnung: str | None = None
gueltig_tage: int | None = None
class ActivationCodeStatusRequest(BaseModel):
active: bool
class ResolveRequest(BaseModel):
device_id: str
class InstallationEventRequest(BaseModel):
"""
Meilenstein/Fehlschlag-Meldung vom Boot-Medium waehrend der Installation
(siehe device_installation_events, Migration 0025) - das Geraet hat zu
diesem Zeitpunkt noch kein agent_secret, deshalb device_fingerprint als
Nachweis statt eines Service-Tokens (gleiches Vertrauensniveau wie
/api/v1/activate selbst).
"""
device_fingerprint: str
event_type: str
detail: str | None = None
class AgentBootstrapRequest(BaseModel):
device_id: str
class AgentHealthPayload(BaseModel):
"""
Health-Metriken (26.08.2026-Planung) - alle optional, damit ein
fehlgeschlagener Einzelwert (z.B. load average auf ungewoehnlichen
Systemen) nicht den ganzen Checkin blockiert. Werte kommen direkt aus
Python-Stdlib-Aufrufen im Agenten (kein neues Paket noetig).
"""
disk_used_percent: float | None = None
ram_used_percent: float | None = None
uptime_seconds: int | None = None
load_1m: float | None = None
load_5m: float | None = None
load_15m: float | None = None
class AgentCheckinRequest(BaseModel):
device_id: str
health: AgentHealthPayload | None = None
class AgentReportEntry(BaseModel):
merkmal: str
event_type: str
detail: dict | None = None
class AgentReportRequest(BaseModel):
device_id: str
results: list[AgentReportEntry]
class AuftragSelectRequest(BaseModel):
optionen: dict = {}
class CreateOrganizationRequest(BaseModel):
name: str
class UpdateOrganizationRequest(BaseModel):
name: str
kontakt_name: str | None = None
adresse: str | None = None
telefon: str | None = None
kontakt_email: str | None = None
notizen: str | None = None
class CreateMerkmalRequest(BaseModel):
key: str
name: str
description: str | None = None
class UpdateMerkmalRequest(BaseModel):
name: str
description: str | None = None
kategorie_id: str | None = None
im_auftragskatalog: bool = False
class CreateKategorieRequest(BaseModel):
key: str
name: str
description: str | None = None
sort_order: int = 0
art: Literal["app", "richtlinie"] = "app"
class WorkspaceAktualisierenRequest(BaseModel):
name: str
description: str | None = None
class WorkspaceVeroeffentlichenRequest(BaseModel):
notiz: str | None = None
class WorkspaceVorlageRequest(BaseModel):
ist_vorlage: bool
sichtbarkeit: Literal["global", "kundengruppe", "organisation"] = "global"
kundengruppe_ids: list[str] = []
organization_ids: list[str] = []
class WorkspaceKopieRequest(BaseModel):
name: str
backend_id: str
organization_id: str | None = None
class CreateWorkspaceAssistentRequest(BaseModel):
"""Workspace-Creator neu, Schritt D: alles, was der Assistent sammelt."""
name: str
description: str | None = None
organization_id: str | None = None
backend_id: str
merkmal_ids: list[str] = []
webapp_template_ids: list[str] = []
class CreateWorkspaceRequest(BaseModel):
key: str
name: str
description: str | None = None
organization_id: str | None = None
backend_id: str | None = None
class CloneWorkspaceRequest(BaseModel):
organization_id: str
class CreateWebappRequest(BaseModel):
name: str
url: str
sort_order: int = 0
kategorie_id: str | None = None
kiosk_modus: bool = False
class UpdateWebappRequest(BaseModel):
name: str
url: str
sort_order: int = 0
kategorie_id: str | None = None
kiosk_modus: bool = False
class CreateWebappTemplateRequest(BaseModel):
name: str
url: str
description: str | None = None
kategorie_id: str | None = None
class UpdateWebappTemplateRequest(BaseModel):
name: str
url: str
description: str | None = None
kategorie_id: str | None = None
class ExtraPartitionRequest(BaseModel):
mountpoint: str
percent: int
filesystem: str
class CreateBereitstellungsvorlageRequest(BaseModel):
workspace_id: str
# Seit ADR-0038 optional: ohne Angabe gilt die Distro des Workspace.
backend_id: str | None = None
label: str
is_default: bool = False
disk_encryption: bool = False
root_filesystem: str = "ext4"
secure_boot_required: bool = False
# Rang nach Plattengroesse (1 = groesste Platte) fuer Geraete mit
# mehreren internen Datentraegern - siehe backends/mint-image/backend.sh
# _mint_image_detect_target_disk() und platform-docs/roadmap/
# multi-distro-merkmal-workspace-creator.md, Phase 0 (08.09.2026).
target_disk_rank: int = 1
# scheme:"custom" + extra_partitions (siehe platform-docs/roadmap/
# e2e-testrunde-lokale-adressierung.md Phase 3, 19.09.2026) - loest die
# bisherige Einschraenkung ab, dass ein custom-Schema nur per direkter
# DB-Pflege moeglich war (siehe Kommentar in create_bereitstellungsvorlage()).
scheme: str = "single"
extra_partitions: list[ExtraPartitionRequest] = []
class CreateEnrollmentSessionRequest(BaseModel):
bereitstellungsvorlage_id: str
max_devices: int
expires_at: str
neuinstallation_ohne_rueckfrage: bool = False
ziel_oe_id: str | None = None
class SetOeWorkspaceRequest(BaseModel):
workspace_id: str | None = None
version_id: str | None = None
automatisch: bool = True
class RechargeEnrollmentSessionRequest(BaseModel):
additional_devices: int = 0
expires_at: str | None = None
class CreateIsoBuildRequest(BaseModel):
activation_code: str
wifi_ssid: str | None = None
wifi_psk: str | None = None
# Diagnosezugang (02.10.2026): nur der SHA-512-crypt-Hash, nie das
# Passwort selbst - das kennt allein die Kundenplattform (Root-Admin).
root_password_hash: str | None = None
# Offline-Medium (Roadmap V5): Golden Image auf die ISO legen.
offline_medium: bool = False
class CreateOrganisationseinheitRequest(BaseModel):
name: str
parent_id: str | None = None
class UpdateOrganisationseinheitRequest(BaseModel):
name: str
parent_id: str | None = None
class MoveDeviceOeRequest(BaseModel):
oe_id: str
class CreateGruppeRequest(BaseModel):
name: str
class UpdateGruppeRequest(BaseModel):
name: str
class SetHostnameKonventionRequest(BaseModel):
praefix: str | None = None
stellen: int = 4
nummer_modus: str = "sequenziell"
def write_log(entry: dict) -> None:
with LOG_PATH.open("a", encoding="utf-8") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
def get_database_connection():
if not DATABASE_URL:
raise RuntimeError("TUXFLOTTE_DATABASE_URL ist nicht gesetzt.")
return psycopg.connect(DATABASE_URL)
def load_or_create_device(
organization_id,
device_fingerprint: str,
hostname: str | None,
) -> tuple[dict, bool]:
"""
Lädt ein bekanntes Gerät oder legt es neu an.
Rückgabe:
(device, created)
"""
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
SELECT
id,
organization_id,
device_fingerprint,
hostname,
created_at,
last_seen
FROM devices
WHERE device_fingerprint = %s
""",
(device_fingerprint,),
)
row = cur.fetchone()
if row is not None:
if str(row[1]) != str(organization_id):
raise RuntimeError(
"Das Gerät ist bereits einer anderen Organization zugeordnet."
)
# archived_at/deprovisioned_at explizit mit zuruecksetzen (nicht
# nur hostname/last_seen): live gefunden (02.09.2026) - ein per
# "Gerät löschen" archiviertes Geraet (archive_device(), Soft
# Delete) blieb nach einer frischen Neuinstallation derselben
# physischen Hardware (gleicher device_fingerprint) dauerhaft aus
# fetch_devices_for_organization() verschwunden, weil load_or_
# create_device() den bestehenden Datensatz zwar fand und
# aktualisierte, archived_at dabei aber unangetastet liess - ein
# erfolgreicher /api/v1/activate-Aufruf ist der eindeutigste
# mögliche Beweis, dass das Geraet wieder aktiv genutzt wird.
cur.execute(
"""
UPDATE devices
SET
hostname = %s,
last_seen = CURRENT_TIMESTAMP,
archived_at = NULL,
deprovisioned_at = NULL
WHERE id = %s
RETURNING
id,
organization_id,
device_fingerprint,
hostname,
created_at,
last_seen
""",
(hostname, row[0]),
)
row = cur.fetchone()
created = False
else:
device_id = uuid4()
cur.execute(
"SELECT id FROM organisationseinheiten WHERE organization_id = %s AND ist_standard",
(organization_id,),
)
standard_oe_row = cur.fetchone()
standard_oe_id = standard_oe_row[0] if standard_oe_row is not None else None
cur.execute(
"""
INSERT INTO devices (
id,
organization_id,
device_fingerprint,
hostname,
oe_id
)
VALUES (%s, %s, %s, %s, %s)
RETURNING
id,
organization_id,
device_fingerprint,
hostname,
created_at,
last_seen
""",
(
device_id,
organization_id,
device_fingerprint,
hostname,
standard_oe_id,
),
)
row = cur.fetchone()
created = True
device = {
"id": str(row[0]),
"organization_id": str(row[1]),
"device_fingerprint": row[2],
"hostname": row[3],
"created_at": row[4].isoformat(),
"last_seen": row[5].isoformat(),
}
return device, created
def store_hardware_snapshot(
device_id: str,
hardware: dict,
) -> str:
snapshot_id = uuid4()
identity = hardware.get("identity", {})
system = hardware.get("system", {})
mainboard = hardware.get("mainboard", {})
firmware = hardware.get("firmware", {})
security = hardware.get("security", {})
cpu = system.get("cpu", {})
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO hardware_snapshots (
id,
device_id,
architecture,
manufacturer,
product_name,
product_version,
system_uuid,
system_serial,
board_vendor,
board_name,
board_serial,
bios_vendor,
bios_version,
boot_mode,
secure_boot,
tpm_version,
cpu_model,
cpu_logical_count,
memory_bytes
)
VALUES (
%s, %s, %s, %s, %s,
%s, %s, %s, %s, %s,
%s, %s, %s, %s, %s,
%s, %s, %s, %s
)
""",
(
snapshot_id,
device_id,
system["architecture"],
system.get("manufacturer"),
system.get("product_name"),
system.get("product_version"),
identity.get("system_uuid"),
identity.get("system_serial"),
mainboard.get("vendor"),
mainboard.get("name"),
identity.get("board_serial"),
firmware.get("bios_vendor"),
firmware.get("bios_version"),
firmware.get("boot_mode"),
firmware.get("secure_boot"),
security.get("tpm_version"),
cpu.get("model"),
cpu.get("logical_count"),
system.get("memory_bytes"),
),
)
for interface in hardware.get("network_interfaces", []):
cur.execute(
"""
INSERT INTO network_interfaces (
id,
hardware_snapshot_id,
name,
type,
mac_address
)
VALUES (%s, %s, %s, %s, %s)
""",
(
uuid4(),
snapshot_id,
interface["name"],
interface["type"],
interface["mac"],
),
)
for storage_device in hardware.get("storage_devices", []):
cur.execute(
"""
INSERT INTO storage_devices (
id,
hardware_snapshot_id,
name,
model,
serial,
size_bytes,
transport
)
VALUES (%s, %s, %s, %s, %s, %s, %s)
""",
(
uuid4(),
snapshot_id,
storage_device["name"],
storage_device.get("model"),
storage_device.get("serial"),
storage_device["size_bytes"],
storage_device.get("transport"),
),
)
return str(snapshot_id)
def find_activation_code(code: str):
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
SELECT ac.organization_id, o.name
FROM activation_codes ac
JOIN organizations o ON o.id = ac.organization_id
WHERE ac.code = %s AND ac.active
AND (ac.gueltig_bis IS NULL OR ac.gueltig_bis > now())
""",
(code,),
)
row = cur.fetchone()
if row is None:
return None
return {"organization_id": str(row[0]), "organization_name": row[1]}
def mark_activation_code_used(code: str):
"""Nutzung zaehlen (GUI-Runde 09.10.2026) - nur nach erfolgreicher
Aktivierung, damit die Verwaltungsseite zeigt, ob ein Code noch
gebraucht wird."""
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
UPDATE activation_codes
SET verwendet_anzahl = verwendet_anzahl + 1, zuletzt_verwendet = now()
WHERE code = %s
""",
(code,),
)
def find_enrollment_session(code: str):
"""
Enrollment Sessions haben kein eigenes code-Feld - ihre id dient direkt
als Bearer-Credential (unratbare UUID, siehe migrations/0012). Nur
gültige Sessions (nicht widerrufen, nicht abgelaufen, Kontingent nicht
erschöpft) werden zurückgegeben; jede andere Bedingung lässt /activate
denselben "invalid_activation_code"-Fehlschlag melden wie ein
ungültiger klassischer Aktivierungscode - das bricht den Installer
bereits beim Server-Handshake (Schritt 15) ab, lange vor jeder
destruktiven Aktion (siehe ADR-0008).
"""
try:
session_id = UUID(code)
except ValueError:
return None
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
SELECT es.id, es.organization_id, o.name, es.bereitstellungsvorlage_id,
es.max_devices, es.devices_used, es.expires_at, es.revoked_at,
es.neuinstallation_ohne_rueckfrage, es.ziel_oe_id
FROM enrollment_sessions es
JOIN organizations o ON o.id = es.organization_id
WHERE es.id = %s
""",
(session_id,),
)
row = cur.fetchone()
if row is None:
return None
(
session_id, organization_id, organization_name, bereitstellungsvorlage_id,
max_devices, devices_used, expires_at, revoked_at, neuinstallation_ohne_rueckfrage, ziel_oe_id,
) = row
if revoked_at is not None:
return None
if expires_at <= datetime.now(timezone.utc):
return None
if devices_used >= max_devices:
return None
return {
"session_id": str(session_id),
"organization_id": str(organization_id),
"organization_name": organization_name,
"bereitstellungsvorlage_id": str(bereitstellungsvorlage_id),
"neuinstallation_ohne_rueckfrage": bool(neuinstallation_ohne_rueckfrage),
"ziel_oe_id": str(ziel_oe_id) if ziel_oe_id else None,
}
def geraet_in_ziel_oe(device_id: str, ziel_oe_id: str) -> None:
"""Gerät aus einem Installationsmedium mit Ziel-OE dorthin setzen
(Schritt F) - auch bei einer Neuinstallation eines bekannten Geräts:
wer mit diesem Stick installiert, will es dort haben. Die Distro-
Verträglichkeit wurde beim Erstellen des Mediums geprüft."""
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
UPDATE devices SET oe_id = %s
WHERE id = %s AND organization_id = (SELECT organization_id FROM organisationseinheiten WHERE id = %s)
""",
(ziel_oe_id, device_id, ziel_oe_id),
)
def consume_enrollment_session(session_id: str) -> None:
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE enrollment_sessions SET devices_used = devices_used + 1 WHERE id = %s",
(session_id,),
)
def fetch_templates(organization_id: str) -> list[dict]:
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
SELECT
bv.id, bv.label, bv.is_default,
w.id, w.name,
b.id, b.name, b.version,
bv.partitioning, bv.disk_encryption, bv.secure_boot_required,
(SELECT count(*) FROM assignments a WHERE a.bereitstellungsvorlage_id = bv.id),
(SELECT count(*) FROM enrollment_sessions es WHERE es.bereitstellungsvorlage_id = bv.id)
FROM bereitstellungsvorlagen bv
JOIN workspaces w ON w.id = bv.workspace_id
JOIN backends b ON b.id = bv.backend_id
WHERE bv.organization_id = %s
ORDER BY bv.is_default DESC, bv.label
""",
(organization_id,),
)
rows = cur.fetchall()
return [
{
"id": str(row[0]),
"label": row[1],
"workspace": {"id": str(row[3]), "name": row[4]},
"backend": {"id": str(row[5]), "name": row[6], "version": row[7]},
"is_default": row[2],
"partitioning": row[8],
"disk_encryption": row[9],
"secure_boot_required": row[10],
# GUI-Runde 09.10.2026 (Punkt 5): Nutzung - solange Geräte oder
# Installationsmedien darauf verweisen, ist Löschen gesperrt.
"geraete_anzahl": row[11],
"medien_anzahl": row[12],
}
for row in rows
]
def create_bereitstellungsvorlage(
organization_id: str,
workspace_id: str,
backend_id: str,
label: str,
is_default: bool,
disk_encryption: bool,
root_filesystem: str,
secure_boot_required: bool,
target_disk_rank: int = 1,
scheme: str = "single",
extra_partitions: list[dict] | None = None,
) -> dict:
"""
Bislang gab es dafuer keinen API-/UI-Weg - Bereitstellungsvorlagen kamen
nur per Seed-Migration (0005/0011) fuer die "Default Lab"-Organisation
zustande (siehe Phase 5 des Self-Service-ISO-Plans, dort als offene
Luecke notiert). Seit 19.09.2026 (siehe platform-docs/roadmap/
e2e-testrunde-lokale-adressierung.md Phase 3) deckt dieses Formular auch
scheme:"custom" + extra_partitions ab - Validierung liegt im aufrufenden
Endpoint (create_bereitstellungsvorlage_endpoint()).
"""
vorlage_id = uuid4()
partitioning = {
"scheme": scheme,
"root_filesystem": root_filesystem,
"target_disk_rank": target_disk_rank,
"extra_partitions": extra_partitions or [],
}
with get_database_connection() as conn:
with conn.cursor() as cur:
if is_default:
# Der Unique-Index bereitstellungsvorlagen_one_default_per_org
# erlaubt nur eine is_default=TRUE-Zeile je Organisation -
# bestehenden Default zuerst explizit zuruecksetzen (gleiches
# Zwei-Schritte-Muster wie Migration 0011).
cur.execute(
"UPDATE bereitstellungsvorlagen SET is_default = FALSE WHERE organization_id = %s AND is_default",
(organization_id,),
)
cur.execute(
"""
INSERT INTO bereitstellungsvorlagen
(id, organization_id, workspace_id, backend_id, label, is_default,
disk_encryption, partitioning, secure_boot_required)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
(
vorlage_id, organization_id, workspace_id, backend_id, label, is_default,
disk_encryption, Jsonb(partitioning), secure_boot_required,
),
)
return {"id": str(vorlage_id), "label": label, "is_default": is_default}
def require_service_token(authorization: str | None) -> bool:
"""
Akzeptiert entweder den Betreiber-Admin-Token oder das separate
Kundenplattform-Service-Token (siehe ADR-0011) - beide dürfen die
Auftragskatalog-Endpoints aufrufen, Kompromittierung/Rotation des einen
betrifft den anderen nicht.
"""
if authorization is None or not authorization.startswith("Bearer "):
return False
provided_token = authorization.removeprefix("Bearer ")
if ADMIN_TOKEN and hmac.compare_digest(provided_token, ADMIN_TOKEN):
return True
if KUNDENPLATTFORM_TOKEN and hmac.compare_digest(provided_token, KUNDENPLATTFORM_TOKEN):
return True
return False
def hash_agent_secret(secret: str) -> str:
return hashlib.sha256(secret.encode("utf-8")).hexdigest()
def verify_agent_secret(device_id: str, provided_secret: str) -> bool:
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT agent_secret_hash FROM devices WHERE id = %s",
(device_id,),
)
row = cur.fetchone()
if row is None or row[0] is None:
return False
return hmac.compare_digest(row[0], hash_agent_secret(provided_secret))
FAVICON_FETCH_TIMEOUT_SECONDS = 5
# Grosszuegig genug fuer jedes realistische Favicon (ueblich: wenige KB bis
# niedrige zweistellige KB), verhindert aber, dass eine falsch verlinkte
# grosse Datei den Request lange blockiert oder viel Speicher belegt.
FAVICON_MAX_BYTES = 2 * 1024 * 1024
FAVICON_LINK_REGEX = re.compile(
r']+rel=["\'](?:shortcut icon|icon|apple-touch-icon)["\'][^>]*>',
re.IGNORECASE,
)
FAVICON_HREF_REGEX = re.compile(r'href=["\']([^"\']+)["\']', re.IGNORECASE)
# Live gefunden (11.09.2026, Phase 6 Icon-Backfill): learningequality.org
# liefert sein Favicon mit Content-Type "image/svg+xml", tatsaechlich sind
# die Bytes ein PNG-in-ICO-Container - ein Browser haette das als kaputtes
# Bild angezeigt (SVG-Parser auf Binaerdaten). Server-Header sind nicht
# vertrauenswuerdig genug, um sie ungeprueft in icon_content_type zu
# uebernehmen - _erkanntes_bildformat() sniffed stattdessen die echten
# Magic Bytes, der Header dient nur noch als Fallback, falls die Signatur
# nichts Bekanntes ergibt (z.B. ein Format ausserhalb dieser Liste).
_BILD_SIGNATUREN = [
(b"\x89PNG\r\n\x1a\n", "image/png"),
(b"\xff\xd8\xff", "image/jpeg"),
(b"GIF87a", "image/gif"),
(b"GIF89a", "image/gif"),
(b"\x00\x00\x01\x00", "image/x-icon"),
]
def _erkanntes_bildformat(data: bytes) -> str | None:
kopf = data[:16]
for signatur, mime in _BILD_SIGNATUREN:
if kopf.startswith(signatur):
return mime
if kopf.startswith(b"RIFF") and data[8:12] == b"WEBP":
return "image/webp"
if b"