provisioning-server/app.py

from pathlib import Path
from datetime import datetime, timezone
import json
import os
from uuid import uuid4

import psycopg

from fastapi import FastAPI, Request
from pydantic import BaseModel
from fastapi.responses import FileResponse


CONFIG_PATH = Path("config.json")
LOG_PATH = Path("activation.log")
DATABASE_URL = os.environ.get("TUXFLOTTE_DATABASE_URL")
app = FastAPI(title="Provisioning Activation Server", version="0.1.0")


class ActivationRequest(BaseModel):
    activation_code: str
    device_fingerprint: str
    hostname: str | None = None
    machine_id: str | None = None
    client_version: str | None = None


def load_config() -> dict:
    with CONFIG_PATH.open("r", encoding="utf-8") as f:
        return json.load(f)


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."
                )
                cur.execute(
                    """
                    UPDATE devices
                    SET
                        hostname = %s,
                        last_seen = CURRENT_TIMESTAMP
                    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(
                    """
                    INSERT INTO devices (
                        id,
                        organization_id,
                        device_fingerprint,
                        hostname
                    )
                    VALUES (%s, %s, %s, %s)
                    RETURNING
                        id,
                        device_fingerprint,
                        hostname,
                        created_at,
                        last_seen
                    """,
                    (
                        device_id,
                        organization_id,
                        device_fingerprint,
                        hostname,
                    ),
                )

                row = cur.fetchone()
                created = True

    device = {
        "id": str(row[0]),
        "organization_id": str(row[1]),
        "device_fingerprint": row[1],
        "hostname": row[2],
        "created_at": row[3].isoformat(),
        "last_seen": row[4].isoformat(),
    }

    return device, created

@app.get("/health")
def health():
    return {
        "status": "ok",
        "service": "provisioning-server",
        "version": "0.1.0"
    }


@app.post("/api/v1/activate")
def activate(payload: ActivationRequest, request: Request):
    config = load_config()
    activation_entry = config.get("activation_codes", {}).get(payload.activation_code)

    success = activation_entry is not None
    device = None
    device_created = False

    if success:
        organization_id = activation_entry["customer"]["organization_id"]

        device, device_created = load_or_create_device(
            organization_id,
            payload.device_fingerprint,
            payload.hostname,
        )

    write_log({
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "event": "activate",
        "success": success,
        "remote_ip": request.client.host if request.client else None,
        "activation_code": payload.activation_code,
        "hostname": payload.hostname,
        "machine_id": payload.machine_id,
        "client_version": payload.client_version
    })

    if not success:
        return {
            "success": False,
            "error": "invalid_activation_code",
            "message": "Der Aktivierungscode ist ungültig."
        }

    profiles = config.get("profiles", {})
    available_profiles = [
        profiles[profile_id]
        for profile_id in activation_entry.get("profile_ids", [])
        if profile_id in profiles
    ]

    return {
        "success": True,

        "device": {
            "id": device["id"],
            "fingerprint": device["device_fingerprint"],
            "hostname": device["hostname"],
            "created": device_created,
        },

        "customer": activation_entry["customer"],
        "profiles": available_profiles
    }


@app.get("/api/v1/profiles/{profile_id}")
def get_profile(profile_id: str):
    config = load_config()
    profile = config.get("profiles", {}).get(profile_id)

    if not profile:
        return {
            "success": False,
            "error": "profile_not_found",
            "message": "Das angeforderte Profil wurde nicht gefunden."
        }

    return {
        "success": True,
        "profile": profile
    }

@app.get("/api/v1/profiles/{profile_id}/bootstrap")
def get_profile_bootstrap(profile_id: str):
    config = load_config()
    profile = config.get("profiles", {}).get(profile_id)

    if not profile:
        return {
            "success": False,
            "error": "profile_not_found",
            "message": "Das angeforderte Profil wurde nicht gefunden."
        }

    return {
        "success": True,
        "profile": {
            "id": profile["id"],
            "label": profile["label"],
            "distribution": profile["distribution"],
            "version": profile["version"]
        },
        "installer": profile.get("installer"),
        "ansible": {
            "repo": profile.get("ansible_repo"),
            "mode": "pull"
        },
        "ssh": {
            "authorized_keys_url": "https://anode.tuxflotte.de/keys/default.pub"
        }
    }

@app.get("/installers/fedora-workstation/ks.cfg")
def get_fedora_kickstart():
    return FileResponse(
        "installers/fedora-workstation/ks.cfg",
        media_type="text/plain"
    )