provisioning-server/app.py

from pathlib import Path
from datetime import datetime, timezone
import json

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


CONFIG_PATH = Path("config.json")
LOG_PATH = Path("activation.log")

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")

DEVICES_DIR = Path("data/devices")

def load_or_create_device(device_fingerprint: str) -> tuple[dict, bool]:
    """
    Lädt ein bekanntes Gerät oder legt es neu an.

    Rückgabe:
        (device, created)
    """

    DEVICES_DIR.mkdir(parents=True, exist_ok=True)

    device_file = DEVICES_DIR / f"{device_fingerprint}.json"

    if device_file.exists():
        with device_file.open("r", encoding="utf-8") as f:
            device = json.load(f)

        device["last_seen"] = datetime.now(timezone.utc).isoformat()
        device["activation_count"] = device.get("activation_count", 0) + 1

        with device_file.open("w", encoding="utf-8") as f:
            json.dump(device, f, indent=2)

        return device, False

    device = {
        "device_fingerprint": device_fingerprint,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "last_seen": datetime.now(timezone.utc).isoformat(),
        "activation_count": 1
    }

    with device_file.open("w", encoding="utf-8") as f:
        json.dump(device, f, indent=2)

    return device, True

@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:
        device, device_created = load_or_create_device(
            payload.device_fingerprint
        )

    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": {
        "fingerprint": device["device_fingerprint"],
        },

        "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"
    )