Thomas Stallinger eb73f2b6b7 feat: extend agent check-in with Auftragskatalog state, add report endpoint
Check-in now also returns "auftraege": the full present/absent state
of every catalog-eligible Merkmal (im_auftragskatalog = true) for the
device's backend, derived statelessly from device_merkmale rather than
just the workspace-resolved blueprints list, which stays untouched.
This is what lets a deselected Auftrag actually be reverted once the
corresponding Ansible role grows an absent branch (ADR-0010) - the
agent doesn't have that branch yet, so this is additive and inert
until it does.

Adds POST /api/v1/agent/report so the agent can write execution
results (applied/apply_failed/removed/remove_failed) back into
device_merkmal_events; selected/deselected is intentionally rejected
here since those are meant to come from wherever Auftrag selection
ends up being triggered, not from the agent itself.

Migration 0007 adds the backing tables (kategorien, device_merkmale,
device_merkmal_events) and the two new merkmale columns.

Verified against anode: checkin toggles present/absent correctly
after inserting a device_merkmale row, report endpoint accepts valid
entries and rejects unknown merkmale / non-agent event types / bad
secrets, and the written event round-trips through device_merkmal_events
correctly as jsonb.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-04 09:17:19 +02:00

716 lines
22 KiB
Python

from pathlib import Path
from datetime import datetime, timezone
import hashlib
import hmac
import json
import os
import secrets
from uuid import uuid4
import psycopg
from psycopg.types.json import Jsonb
from fastapi import FastAPI, Header, Request
from pydantic import BaseModel
from fastapi.responses import FileResponse
LOG_PATH = Path("activation.log")
DATABASE_URL = os.environ.get("TUXFLOTTE_DATABASE_URL")
AGENT_POLL_INTERVAL_SECONDS = int(os.environ.get("TUXFLOTTE_AGENT_POLL_INTERVAL", "300"))
ANSIBLE_CONTENT_REPO = os.environ.get("TUXFLOTTE_ANSIBLE_CONTENT_REPO", "")
AGENT_REPORTABLE_EVENT_TYPES = {"applied", "apply_failed", "removed", "remove_failed"}
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
hardware: dict
class ResolveRequest(BaseModel):
device_id: str
class AgentBootstrapRequest(BaseModel):
device_id: str
class AgentCheckinRequest(BaseModel):
device_id: str
class AgentReportEntry(BaseModel):
merkmal: str
event_type: str
detail: dict | None = None
class AgentReportRequest(BaseModel):
device_id: str
results: list[AgentReportEntry]
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,
organization_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[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
""",
(code,),
)
row = cur.fetchone()
if row is None:
return None
return {"organization_id": str(row[0]), "organization_name": row[1]}
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
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],
}
for row in rows
]
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))
def fetch_assigned_blueprints(device_id: str):
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
SELECT bv.workspace_id, bv.backend_id
FROM assignments a
JOIN bereitstellungsvorlagen bv ON bv.id = a.bereitstellungsvorlage_id
WHERE a.device_id = %s
""",
(device_id,),
)
assignment_row = cur.fetchone()
if assignment_row is None:
return []
workspace_id, backend_id = assignment_row
cur.execute(
"""
SELECT m.key, bp.ansible_role
FROM workspace_merkmale wm
JOIN merkmale m ON m.id = wm.merkmal_id
JOIN blueprints bp ON bp.merkmal_id = m.id AND bp.backend_id = %s
WHERE wm.workspace_id = %s
""",
(backend_id, workspace_id),
)
return [
{"merkmal": key, "ansible_role": ansible_role}
for key, ansible_role in cur.fetchall()
]
def fetch_auftragskatalog_state(device_id: str):
"""
Voller Soll-Zustand (present/absent) aller katalogfähigen Merkmale für
das Backend des Device, unabhängig von der Workspace-Zuweisung
(siehe ADR-0010). Zustandslos aus der aktuellen Auswahl abgeleitet,
keine Historie nötig.
"""
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
SELECT bv.backend_id
FROM assignments a
JOIN bereitstellungsvorlagen bv ON bv.id = a.bereitstellungsvorlage_id
WHERE a.device_id = %s
""",
(device_id,),
)
assignment_row = cur.fetchone()
if assignment_row is None:
return []
(backend_id,) = assignment_row
cur.execute(
"""
SELECT m.key, bp.ansible_role, COALESCE(dm.aktiv, FALSE)
FROM merkmale m
JOIN blueprints bp ON bp.merkmal_id = m.id AND bp.backend_id = %s
LEFT JOIN device_merkmale dm
ON dm.merkmal_id = m.id AND dm.device_id = %s
WHERE m.im_auftragskatalog
""",
(backend_id, device_id),
)
return [
{
"merkmal": key,
"ansible_role": ansible_role,
"state": "present" if aktiv else "absent",
}
for key, ansible_role, aktiv in cur.fetchall()
]
@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):
activation = find_activation_code(payload.activation_code)
success = activation is not None
device = None
device_created = False
hardware_snapshot_id = None
if success:
organization_id = activation["organization_id"]
device, device_created = load_or_create_device(
organization_id,
payload.device_fingerprint,
payload.hostname,
)
hardware_snapshot_id = store_hardware_snapshot(
device["id"],
payload.hardware,
)
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."
}
return {
"success": True,
"device": {
"id": device["id"],
"fingerprint": device["device_fingerprint"],
"hostname": device["hostname"],
"registration_status": (
"registered"
if device_created
else "existing"
),
"hardware_snapshot_id": hardware_snapshot_id,
},
"customer": {
"id": activation["organization_id"],
"organization_id": activation["organization_id"],
"name": activation["organization_name"],
},
"templates": fetch_templates(activation["organization_id"]),
}
@app.post("/api/v1/templates/{template_id}/resolve")
def resolve_template(template_id: str, payload: ResolveRequest):
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
SELECT bv.workspace_id, bv.backend_id, w.key, b.key,
bv.disk_encryption, bv.partitioning, bv.secure_boot_required
FROM bereitstellungsvorlagen bv
JOIN workspaces w ON w.id = bv.workspace_id
JOIN backends b ON b.id = bv.backend_id
WHERE bv.id = %s
""",
(template_id,),
)
template_row = cur.fetchone()
if template_row is None:
return {
"success": False,
"error": "template_not_found",
"message": "Die angeforderte Bereitstellungsvorlage wurde nicht gefunden."
}
(
workspace_id, backend_id, workspace_key, backend_key,
disk_encryption, partitioning, secure_boot_required,
) = template_row
cur.execute(
"""
SELECT m.key, bp.ansible_role
FROM workspace_merkmale wm
JOIN merkmale m ON m.id = wm.merkmal_id
JOIN blueprints bp ON bp.merkmal_id = m.id AND bp.backend_id = %s
WHERE wm.workspace_id = %s
""",
(backend_id, workspace_id),
)
blueprints = [
{"merkmal": key, "ansible_role": ansible_role}
for key, ansible_role in cur.fetchall()
]
cur.execute(
"""
INSERT INTO assignments (id, device_id, bereitstellungsvorlage_id)
VALUES (%s, %s, %s)
ON CONFLICT (device_id) DO UPDATE
SET bereitstellungsvorlage_id = EXCLUDED.bereitstellungsvorlage_id
""",
(uuid4(), payload.device_id, template_id),
)
return {
"success": True,
"runtime_blueprint": {
"workspace_id": workspace_key,
"backend_id": backend_key,
"blueprints": blueprints,
"installation_directives": {
"disk_encryption": disk_encryption,
"partitioning": partitioning,
"secure_boot_required": secure_boot_required,
},
},
}
@app.get("/installers/fedora-workstation/ks.cfg")
def get_fedora_kickstart():
return FileResponse(
"installers/fedora-workstation/ks.cfg",
media_type="text/plain"
)
@app.post("/api/v1/agent/bootstrap")
def agent_bootstrap(payload: AgentBootstrapRequest):
agent_secret = secrets.token_urlsafe(32)
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
UPDATE devices
SET agent_secret_hash = %s, agent_secret_issued_at = %s
WHERE id = %s
RETURNING id
""",
(
hash_agent_secret(agent_secret),
datetime.now(timezone.utc),
payload.device_id,
),
)
updated = cur.fetchone()
if updated is None:
return {
"success": False,
"error": "device_not_found",
"message": "Das angegebene Gerät wurde nicht gefunden.",
}
return {"success": True, "agent_secret": agent_secret}
@app.post("/api/v1/agent/checkin")
def agent_checkin(
payload: AgentCheckinRequest,
authorization: str | None = Header(default=None),
):
if authorization is None or not authorization.startswith("Bearer "):
return {
"success": False,
"error": "unauthorized",
"message": "Fehlendes oder ungültiges Authorization-Header.",
}
provided_secret = authorization.removeprefix("Bearer ")
if not verify_agent_secret(payload.device_id, provided_secret):
return {
"success": False,
"error": "unauthorized",
"message": "Ungültiges Agent-Secret.",
}
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE devices SET agent_last_checkin = %s WHERE id = %s",
(datetime.now(timezone.utc), payload.device_id),
)
return {
"success": True,
"blueprints": fetch_assigned_blueprints(payload.device_id),
"auftraege": fetch_auftragskatalog_state(payload.device_id),
"ansible_repo": ANSIBLE_CONTENT_REPO,
"poll_interval_seconds": AGENT_POLL_INTERVAL_SECONDS,
}
@app.post("/api/v1/agent/report")
def agent_report(
payload: AgentReportRequest,
authorization: str | None = Header(default=None),
):
if authorization is None or not authorization.startswith("Bearer "):
return {
"success": False,
"error": "unauthorized",
"message": "Fehlendes oder ungültiges Authorization-Header.",
}
provided_secret = authorization.removeprefix("Bearer ")
if not verify_agent_secret(payload.device_id, provided_secret):
return {
"success": False,
"error": "unauthorized",
"message": "Ungültiges Agent-Secret.",
}
for entry in payload.results:
if entry.event_type not in AGENT_REPORTABLE_EVENT_TYPES:
return {
"success": False,
"error": "invalid_event_type",
"message": f"Kein vom Agenten meldbarer event_type: {entry.event_type}",
}
with get_database_connection() as conn:
with conn.cursor() as cur:
merkmal_ids = {}
for entry in payload.results:
if entry.merkmal in merkmal_ids:
continue
cur.execute(
"SELECT id FROM merkmale WHERE key = %s",
(entry.merkmal,),
)
merkmal_row = cur.fetchone()
if merkmal_row is None:
return {
"success": False,
"error": "unknown_merkmal",
"message": f"Unbekanntes Merkmal: {entry.merkmal}",
}
merkmal_ids[entry.merkmal] = merkmal_row[0]
for entry in payload.results:
cur.execute(
"""
INSERT INTO device_merkmal_events (
id, device_id, merkmal_id, event_type, detail
)
VALUES (%s, %s, %s, %s, %s)
""",
(
uuid4(),
payload.device_id,
merkmal_ids[entry.merkmal],
entry.event_type,
Jsonb(entry.detail) if entry.detail is not None else None,
),
)
return {"success": True}