provisioning-agent/agent.py

import json
import os
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


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 checkin(device_id, agent_secret):
    return api_request(
        "/api/v1/agent/checkin", agent_secret, {"device_id": device_id}
    )


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

        time.sleep(interval)


if __name__ == "__main__":
    main()