provisioning-agent/agent.py

import json
import os
import random
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.request


ANODE_URL = os.environ.get("TUXFLOTTE_ANODE_URL", "https://anode.tuxflotte.de")
CREDENTIALS_PATH = os.environ.get(
    "TUXFLOTTE_AGENT_CREDENTIALS", "/etc/tuxflotte/agent.credentials"
)
FALLBACK_INTERVAL_SECONDS = 60
# Jitter (26.08.2026-Planung): nur auf "grosse" Intervalle (Standard 6h)
# anwenden, nicht auf den kurzen Debug-Modus (60s) oder den Fallback-Retry
# (FALLBACK_INTERVAL_SECONDS) - sonst wuerde Jitter genau den Zweck des
# schnellen Debug-Zyklus untergraben.
JITTER_THRESHOLD_SECONDS = 60 * 60
JITTER_MAX_SECONDS = 30 * 60


def log(message):
    print(f"[tuxflotte-agent] {message}", flush=True)


def load_credentials():
    with open(CREDENTIALS_PATH, "r", encoding="utf-8") as f:
        credentials = json.load(f)

    return credentials["device_id"], credentials["agent_secret"]


def save_credentials(device_id, agent_secret):
    with open(CREDENTIALS_PATH, "w", encoding="utf-8") as f:
        json.dump({"device_id": device_id, "agent_secret": agent_secret}, f)


def api_request(path, agent_secret, body):
    data = json.dumps(body).encode("utf-8")

    req = urllib.request.Request(
        ANODE_URL + path,
        data=data,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {agent_secret}",
        },
        method="POST",
    )

    with urllib.request.urlopen(req, timeout=15) as response:
        return json.loads(response.read().decode("utf-8"))


def bootstrap(device_id):
    """
    Holt ein frisches agent_secret über /api/v1/agent/bootstrap. Braucht
    keinen Authorization-Header (der Endpoint kannte von Anfang an keine
    andere Authentifizierung als die device_id, siehe provisioning-server
    app.py) - macht sich zunutze, dass so ein Re-Bootstrap nach einer
    Reprovisionierung im Kundenportal möglich ist, ohne physisch am Gerät
    etwas tun zu müssen.
    """

    data = json.dumps({"device_id": device_id}).encode("utf-8")

    req = urllib.request.Request(
        ANODE_URL + "/api/v1/agent/bootstrap",
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )

    with urllib.request.urlopen(req, timeout=15) as response:
        return json.loads(response.read().decode("utf-8"))


def collect_health():
    """
    Health-Metriken fuers Checkin (26.08.2026-Planung) - reines Stdlib,
    keine neue Abhaengigkeit. Jeder Einzelwert einzeln in try/except: ein
    fehlschlagender Wert (z.B. Load Average auf einem ungewoehnlichen
    System) darf die anderen nicht verhindern.
    """

    health = {}

    try:
        nutzung = shutil.disk_usage("/")
        health["disk_used_percent"] = round(nutzung.used / nutzung.total * 100, 1)
    except OSError as exc:
        log(f"Health: Diskbelegung nicht ermittelbar: {exc}")

    try:
        werte = {}
        with open("/proc/meminfo", "r", encoding="utf-8") as f:
            for zeile in f:
                schluessel, _, rest = zeile.partition(":")
                if schluessel in ("MemTotal", "MemAvailable"):
                    werte[schluessel] = int(rest.strip().split()[0])
        if werte.get("MemTotal"):
            belegt = werte["MemTotal"] - werte.get("MemAvailable", 0)
            health["ram_used_percent"] = round(belegt / werte["MemTotal"] * 100, 1)
    except (OSError, ValueError) as exc:
        log(f"Health: RAM-Auslastung nicht ermittelbar: {exc}")

    try:
        with open("/proc/uptime", "r", encoding="utf-8") as f:
            health["uptime_seconds"] = int(float(f.read().split()[0]))
    except (OSError, ValueError) as exc:
        log(f"Health: Uptime nicht ermittelbar: {exc}")

    try:
        load1, load5, load15 = os.getloadavg()
        health["load_1m"] = round(load1, 2)
        health["load_5m"] = round(load5, 2)
        health["load_15m"] = round(load15, 2)
    except OSError as exc:
        log(f"Health: Load Average nicht ermittelbar: {exc}")

    return health


def checkin(device_id, agent_secret):
    return api_request(
        "/api/v1/agent/checkin", agent_secret, {"device_id": device_id, "health": collect_health()}
    )


def report(device_id, agent_secret, results):
    return api_request(
        "/api/v1/agent/report",
        agent_secret,
        {"device_id": device_id, "results": results},
    )


def apply_roles(ansible_repo, blueprints, auftraege):
    """
    Führt sowohl die additiv über den Workspace zugewiesenen Blueprints als
    auch die Auftragskatalog-Einträge in einem gemeinsamen ansible-pull-Lauf
    aus. Auftragskatalog-Rollen müssen unabhängig von present/absent immer
    in --tags stehen, sonst laufen ihre Tasks (inklusive des absent-Zweigs,
    siehe ADR-0010) gar nicht erst.
    """

    roles = sorted(
        {b["ansible_role"] for b in blueprints}
        | {a["ansible_role"] for a in auftraege}
    )
    auftrag_states = {a["ansible_role"]: a["state"] for a in auftraege}
    auftrag_optionen = {a["ansible_role"]: a.get("optionen", {}) for a in auftraege}

    log(f"Applying roles via ansible-pull: {','.join(roles)}")

    result = subprocess.run(
        [
            "ansible-pull",
            "-U", ansible_repo,
            "--tags", ",".join(roles),
            "-i", "localhost,",
            "-e", json.dumps({
                "tuxflotte_auftrag_states": auftrag_states,
                "tuxflotte_auftrag_optionen": auftrag_optionen,
            }),
            "site.yml",
        ],
        check=False,
    )

    if result.returncode != 0:
        log(f"ansible-pull exited with status {result.returncode}")

    return result.returncode == 0


def build_report_results(auftraege, pull_succeeded):
    results = []

    for entry in auftraege:
        if entry["state"] == "present":
            event_type = "applied" if pull_succeeded else "apply_failed"
        else:
            event_type = "removed" if pull_succeeded else "remove_failed"

        results.append({"merkmal": entry["merkmal"], "event_type": event_type})

    return results


def run_once(device_id, agent_secret):
    """
    Rückgabe: (poll_interval_seconds, agent_secret) - der Aufrufer muss das
    zurückgegebene agent_secret für den nächsten Zyklus übernehmen, falls
    hier ein Re-Bootstrap stattgefunden hat (sonst würde main() dauerhaft mit
    dem alten, ungültigen Secret weiterlaufen und bei jedem Zyklus erneut
    bootstrappen).
    """

    result = checkin(device_id, agent_secret)

    if not result.get("success") and result.get("error") == "unauthorized":
        # Passiert nach einer Deprovisionierung im Kundenportal - das lokale
        # Secret wurde serverseitig ungültig gemacht. Einmalig neu
        # bootstrappen und den Check-in wiederholen, statt für immer 401 zu
        # loggen (siehe bootstrap()-Kommentar).
        log("Check-in unauthorized, versuche Re-Bootstrap.")

        try:
            bootstrap_result = bootstrap(device_id)
        except (urllib.error.URLError, OSError) as exc:
            log(f"Re-Bootstrap-Fehler: {exc}")
            return FALLBACK_INTERVAL_SECONDS, agent_secret

        if not bootstrap_result.get("success"):
            log(f"Re-Bootstrap failed: {bootstrap_result.get('message', bootstrap_result.get('error'))}")
            return FALLBACK_INTERVAL_SECONDS, agent_secret

        agent_secret = bootstrap_result["agent_secret"]
        save_credentials(device_id, agent_secret)
        result = checkin(device_id, agent_secret)

    if not result.get("success"):
        log(f"Check-in failed: {result.get('message', result.get('error'))}")
        return FALLBACK_INTERVAL_SECONDS, agent_secret

    blueprints = result.get("blueprints", [])
    auftraege = result.get("auftraege", [])

    if blueprints or auftraege:
        pull_succeeded = apply_roles(result["ansible_repo"], blueprints, auftraege)

        if auftraege:
            try:
                report(device_id, agent_secret, build_report_results(auftraege, pull_succeeded))
            except (urllib.error.URLError, OSError) as exc:
                log(f"Report-Fehler: {exc}")
    else:
        log("Check-in ok, nothing to do.")

    return result.get("poll_interval_seconds", FALLBACK_INTERVAL_SECONDS), agent_secret


def main():
    if shutil.which("ansible-pull") is None:
        log("ansible-pull nicht gefunden. Bitte ansible-core installieren.")
        sys.exit(1)

    device_id, agent_secret = load_credentials()

    while True:
        try:
            interval, agent_secret = run_once(device_id, agent_secret)
        except (urllib.error.URLError, OSError) as exc:
            log(f"Check-in-Fehler: {exc}")
            interval = FALLBACK_INTERVAL_SECONDS

        if interval >= JITTER_THRESHOLD_SECONDS:
            interval += random.uniform(0, JITTER_MAX_SECONDS)

        time.sleep(interval)


if __name__ == "__main__":
    main()