bf-infra/nautobot/tp44-cf-sync.py

#!/usr/bin/env python3
"""
TP44 Etappe 2 Teil B — CF-Estate LIVE aus der Cloudflare-API nach Nautobot syncen.

Ersetzt den Inline-Snapshot von tp44-cf-estate-seed.py durch einen periodischen
Live-Pull (get-or-create + Merge). Idempotent, Dry-Run default. Kein Push nach CF —
read-only Pull, Nautobot bleibt Dokumentations-SSoT, CF operativer Master.

Deckung (Live-Pull, je Endpoint 403-tolerant):
  CloudNetwork  cfd_tunnel + warp_connector           (4)
  CloudService  workers/scripts + access/apps         (11)  [Repo-Token OK]
  CloudService  hyperdrive/configs + VPC directory    (3)   [braucht Token-Scope]

Verwaisung: ein CF-Objekt, das in Nautobot steht, aber im Live-Pull fehlt, wird
NICHT geloescht, sondern extra_config.stale=true gesetzt (Doku-SSoT bewahrt Historie).
Ein Objekt, das wieder auftaucht, verliert die Markierung.

Tokens:
  CF     : CF_API_TOKEN + CF_ACCOUNT_ID aus ~/.secrets/credentials.env (nie im Output)
  Nautobot: NAUTOBOT_SUPERUSER_API_TOKEN aus docs/nas-vlan20/stacks/210-nautobot.yml (gitignored)

Aufruf:
  python3 nautobot/tp44-cf-sync.py            # Dry-Run (default) — zeigt Diff
  python3 nautobot/tp44-cf-sync.py --commit   # schreibt
"""
import json
import os
import re
import sys
import urllib.request
import urllib.parse
import urllib.error
from pathlib import Path

NB = os.environ.get("NAUTOBOT_URL", "http://10.40.20.30:8080")
REPO = Path(__file__).resolve().parent.parent
YML = REPO / "docs" / "nas-vlan20" / "stacks" / "210-nautobot.yml"
COMMIT = "--commit" in sys.argv

# ------------------------------------------------------------ Cloudflare-Seite
def cf_secrets():
    env = {}
    with open(os.path.expanduser("~/.secrets/credentials.env")) as fh:
        for line in fh:
            line = line.strip()
            if line.startswith("#") or "=" not in line:
                continue
            k, v = line.split("=", 1)
            env[k.strip()] = v.strip().strip("'\"")
    return env


_CF = cf_secrets()
CF_TOK, CF_ACCT = _CF.get("CF_API_TOKEN"), _CF.get("CF_ACCOUNT_ID")
if not CF_TOK or not CF_ACCT:
    sys.exit("CF_API_TOKEN/CF_ACCOUNT_ID fehlen in ~/.secrets/credentials.env")


def cf_get(path):
    """(ok, result) — faengt 403/404 ab, damit fehlende Token-Scopes den Sync nicht kippen."""
    req = urllib.request.Request("https://api.cloudflare.com/client/v4" + path,
                                 headers={"Authorization": "Bearer " + CF_TOK})
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return True, json.load(r).get("result")
    except urllib.error.HTTPError as e:
        if e.code in (403, 404):
            return False, None
        raise


def cf_ingress(tunnel_id):
    ok, res = cf_get(f"/accounts/{CF_ACCT}/cfd_tunnel/{tunnel_id}/configurations")
    rules = ((res or {}).get("config") or {}).get("ingress") or [] if ok else []
    return [r["hostname"] for r in rules if r.get("hostname")]


def load_cf_estate():
    """Baut (networks, services) aus dem Live-Pull. Jedes Objekt:
    (name, resource_type, description, extra_config, match_key, match_val).
    match_key/-val = stabile CF-ID → idempotentes Matching trotz kuratierter Namen.
    Live-Zaehl-/Statusfelder liegen unter eigenen Keys (connector_count, ingress_hostnames),
    damit der Merge kuratierte Strings (z.B. connectors='3x @ ap1/ap2/ap3') nicht ueberschreibt."""
    networks, services = [], []
    skipped = []

    # --- CloudNetworks: cloudflared-Tunnel (+ Connector-Zahl, Ingress-Hostnames)
    ok, tuns = cf_get(f"/accounts/{CF_ACCT}/cfd_tunnel?is_deleted=false&per_page=50")
    for t in (tuns or []):
        networks.append((t["name"], "Cloudflare cloudflared Tunnel",
                         f"cloudflared-Tunnel ({len(t.get('connections') or [])} Connectoren)",
                         {"tunnel_id": t["id"], "kind": "cloudflared",
                          "connector_count": len(t.get("connections") or []),
                          "ingress_hostnames": cf_ingress(t["id"])},
                         "tunnel_id", t["id"]))
    # --- CloudNetworks: WARP-Connector
    ok, warps = cf_get(f"/accounts/{CF_ACCT}/warp_connector?per_page=50")
    for w in (warps or []):
        networks.append((w["name"], "Cloudflare WARP Connector",
                         f"WARP-Connector ({len(w.get('connections') or [])} Connectoren)",
                         {"tunnel_id": w["id"], "kind": "warp_connector",
                          "connector_count": len(w.get("connections") or [])},
                         "tunnel_id", w["id"]))

    # --- CloudServices: Workers
    ok, scripts = cf_get(f"/accounts/{CF_ACCT}/workers/scripts")
    for s in (scripts or []):
        services.append((s["id"], "Cloudflare Worker", "Cloudflare Worker",
                         {"script": s["id"]}, "script", s["id"]))

    # --- CloudServices: Access-Apps
    ok, apps = cf_get(f"/accounts/{CF_ACCT}/access/apps?per_page=100")
    if not ok:
        skipped.append("access/apps")
    for a in (apps or []):
        services.append((a.get("name") or a.get("id"), "Cloudflare Access Application",
                         f"Access-App ({a.get('type')})",
                         {"id": a.get("id"), "type": a.get("type"), "domain": a.get("domain")},
                         "id", a.get("id")))

    # --- CloudServices: Hyperdrive (braucht Scope)
    ok, hyper = cf_get(f"/accounts/{CF_ACCT}/hyperdrive/configs")
    if not ok:
        skipped.append("hyperdrive/configs")
    for h in (hyper or []):
        origin = h.get("origin") or {}
        services.append((h.get("name") or h.get("id"), "Cloudflare Hyperdrive",
                         "Hyperdrive-Config",
                         {"id": h.get("id"), "database": origin.get("database"),
                          "user": origin.get("user")}, "id", h.get("id")))

    # --- CloudServices: VPC-Service (braucht Scope)
    ok, vpc = cf_get(f"/accounts/{CF_ACCT}/connectivity/directory/services")
    if not ok:
        skipped.append("connectivity/directory/services")
    for v in (vpc or []):
        services.append((v.get("name") or v.get("id"), "Cloudflare VPC Service",
                         "VPC-Service",
                         {"service_id": v.get("id"), "type": v.get("type")},
                         "service_id", v.get("id")))

    return networks, services, skipped


# ------------------------------------------------------------ Nautobot-Seite
def nb_token():
    for line in YML.read_text().splitlines():
        m = re.search(r"NAUTOBOT_SUPERUSER_API_TOKEN[=:]\s*([^\s\"']+)", line)
        if m:
            return m.group(1)
    sys.exit(f"NAUTOBOT_SUPERUSER_API_TOKEN nicht in {YML} gefunden")


NB_HDRS = {"Authorization": f"Token {nb_token()}", "Accept": "application/json",
           "Content-Type": "application/json"}


def nb(method, path, body=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(f"{NB}{path}", data=data, headers=NB_HDRS, method=method)
    try:
        with urllib.request.urlopen(r, timeout=20) as resp:
            return resp.status, json.loads(resp.read().decode() or "{}")
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read().decode() or "{}")


def nb_all(endpoint, params):
    """Alle Rows eines Endpoints (paginiert)."""
    out, url = [], f"/api/{endpoint}/?{urllib.parse.urlencode(dict(params, depth=0, limit=200))}"
    while url:
        st, d = nb("GET", url)
        out += d.get("results", [])
        nxt = d.get("next")
        url = nxt.replace(NB, "") if nxt else None
    return out


_INDEX = {}  # endpoint -> [rows] (einmal je Account geladen)


def index(endpoint, acct_id):
    if endpoint not in _INDEX:
        _INDEX[endpoint] = nb_all(endpoint, {"cloud_account": acct_id})
    return _INDEX[endpoint]


def upsert(endpoint, name, payload, label, seen, acct_id=None,
           match_key=None, match_val=None):
    """get-or-create + Merge von extra_config.
    Matching: bei (match_key,match_val) nach der stabilen CF-ID in extra_config
    (Rename-fest), sonst nach Name. Trackt die getroffene id in seen[endpoint]."""
    ex = None
    if match_val is not None:
        for row in index(endpoint, acct_id):
            if (row.get("extra_config") or {}).get(match_key) == match_val:
                ex = row
                break
    else:
        for row in (index(endpoint, acct_id) if acct_id else
                    nb_all(endpoint, {"name": name})):
            if row.get("name") == name:
                ex = row
                break
    if ex:
        seen.setdefault(endpoint, set()).add(ex["id"])
        merged = dict(ex.get("extra_config") or {})
        merged.update(payload.get("extra_config") or {})
        if merged == (ex.get("extra_config") or {}):
            print(f"  = {label}: '{name}' unveraendert")
            return ex
        if not COMMIT:
            print(f"  ~ {label}: '{name}' extra_config-Update  [DRY-RUN]")
            return ex
        st, d = nb("PATCH", f"/api/{endpoint}/{ex['id']}/", {"extra_config": merged})
        print(f"  ~ {label}: '{name}' aktualisiert" if st == 200
              else f"  ! {label}: '{name}' PATCH {st}: {json.dumps(d)[:200]}")
        return d if st == 200 else ex
    if not COMMIT:
        print(f"  + {label}: '{name}'  [DRY-RUN]")
        return {"id": f"<dry:{name}>"}
    st, d = nb("POST", f"/api/{endpoint}/", payload)
    if st in (200, 201):
        seen.setdefault(endpoint, set()).add(d["id"])
        print(f"  + {label}: '{name}' angelegt")
        return d
    print(f"  ! {label}: '{name}' FEHLER {st}: {json.dumps(d)[:200]}")
    return None


def mark_stale(endpoint, acct_id, seen_ids, label):
    """CF-Objekte in Nautobot (per id), die im Live-Pull fehlen → extra_config.stale=true.
    Ein wieder aufgetauchtes Objekt verliert die Markierung."""
    for row in index(endpoint, acct_id):
        extra = row.get("extra_config") or {}
        nm = row.get("name")
        if row["id"] in seen_ids:
            if extra.get("stale") and COMMIT:
                nb("PATCH", f"/api/{endpoint}/{row['id']}/",
                   {"extra_config": {k: v for k, v in extra.items() if k != "stale"}})
                print(f"  ↩ {label}: '{nm}' stale-Markierung entfernt (wieder da)")
            continue
        if extra.get("stale"):
            continue
        if not COMMIT:
            print(f"  ⚠ {label}: '{nm}' fehlt im Live-Pull → wuerde stale markiert  [DRY-RUN]")
            continue
        nb("PATCH", f"/api/{endpoint}/{row['id']}/", {"extra_config": dict(extra, stale=True)})
        print(f"  ⚠ {label}: '{nm}' als stale markiert (nicht mehr in CF)")


RESOURCE_TYPES = [
    ("Cloudflare cloudflared Tunnel", "cloud.cloudnetwork"),
    ("Cloudflare WARP Connector", "cloud.cloudnetwork"),
    ("Cloudflare VPC Service", "cloud.cloudservice"),
    ("Cloudflare Hyperdrive", "cloud.cloudservice"),
    ("Cloudflare Worker", "cloud.cloudservice"),
    ("Cloudflare Access Application", "cloud.cloudservice"),
]


def main():
    print(f"== TP44 CF-Sync  ({'COMMIT' if COMMIT else 'DRY-RUN'})  CF→{NB} ==")
    networks, services, skipped = load_cf_estate()
    print(f"Live-Pull: {len(networks)} CloudNetworks · {len(services)} CloudServices"
          + (f"  ·  SKIP (Token-Scope): {', '.join(skipped)}" if skipped else "  ·  alle Endpoints lesbar"))

    print("\n[1] Manufacturer + CloudAccount")
    mfr = upsert("dcim/manufacturers", "Cloudflare", {"name": "Cloudflare"}, "Manufacturer", {})
    mfr_id = mfr["id"]
    acct = upsert("cloud/cloud-accounts", "Cloudflare",
                  {"name": "Cloudflare", "account_number": CF_ACCT, "provider": mfr_id},
                  "CloudAccount", {})
    acct_id = acct["id"]

    print("\n[2] CloudResourceTypes")
    rt_ids = {}
    for name, ct in RESOURCE_TYPES:
        rt = upsert("cloud/cloud-resource-types", name,
                    {"name": name, "provider": mfr_id, "content_types": [ct]}, "ResourceType", {})
        rt_ids[name] = rt["id"]

    seen = {}

    print("\n[3] CloudNetworks")
    for name, rt_name, desc, extra, mk, mv in networks:
        upsert("cloud/cloud-networks", name,
               {"name": name, "cloud_resource_type": rt_ids.get(rt_name), "cloud_account": acct_id,
                "description": desc, "extra_config": extra}, "CloudNetwork", seen,
               acct_id=acct_id, match_key=mk, match_val=mv)

    print("\n[4] CloudServices")
    for name, rt_name, desc, extra, mk, mv in services:
        upsert("cloud/cloud-services", name,
               {"name": name, "cloud_resource_type": rt_ids.get(rt_name), "cloud_account": acct_id,
                "description": desc, "extra_config": extra}, "CloudService", seen,
               acct_id=acct_id, match_key=mk, match_val=mv)

    print("\n[5] Verwaisung (stale-Markierung; nur Endpoints, die gelesen wurden)")
    # Nur markieren, wenn der zugehoerige Endpoint erreichbar war — sonst wuerde ein
    # fehlender Token-Scope echte Objekte faelschlich als verwaist markieren.
    if "access/apps" not in skipped and "hyperdrive/configs" not in skipped \
            and "connectivity/directory/services" not in skipped:
        mark_stale("cloud/cloud-services", acct_id, seen.get("cloud/cloud-services", set()), "CloudService")
    else:
        print("  · CloudService-Verwaisung uebersprungen (Token-Scope unvollstaendig — kein sicherer Vergleich)")
    mark_stale("cloud/cloud-networks", acct_id, seen.get("cloud/cloud-networks", set()), "CloudNetwork")

    print(f"\n== fertig ({'COMMIT' if COMMIT else 'DRY-RUN — nichts geschrieben'}) ==")
    if skipped:
        print(f"Hinweis: {', '.join(skipped)} nicht lesbar → Token-Scope erweitern, dann syncen sie mit.")


if __name__ == "__main__":
    main()