bf-infra/jobs/cf_estate_sync.py

"""
TP44 Teil B — CF-Estate → Nautobot, als geplanter Nautobot-Job (ORM).

Kanonische Runtime des CF→Nautobot-Sync. Laeuft im nautobot-worker (kein extra
Container), schreibt per ORM (keine Nautobot-Exposure), zieht die CF-API outbound.
Portierung der Reconcile-Logik aus nautobot/tp44-cf-sync.py (REST) auf das ORM.

Governance (User 2026-09-09): periodische externe->Nautobot-Syncs leben als
Nautobot-Jobs in bf-infra/jobs/ und werden in Nautobot geplant — KEIN per-Task-
Container/-Cron. Kuenftige Syncs = weitere Jobs hier.

Deckung: CloudNetwork (cfd_tunnel + warp_connector), CloudService (workers,
access-apps, hyperdrive, VPC), Prefix-Assignments (teamnet/routes), Verwaisung
(stale-Markierung statt Loeschen). ID-Matching + extra_config-Merge wie im Skript.

Token: Nautobot-Secret "cf-api-token" (Provider environment-variable -> CF_API_TOKEN
am Worker). CF-Account-ID: Secret "cf-account-id" ODER Konstante unten.

CF bleibt operativer Master; Nautobot ist Dokumentations-SSoT (read-only Pull).
"""
import json
import urllib.request
import urllib.parse
import urllib.error

from django.contrib.contenttypes.models import ContentType

from nautobot.apps.jobs import BooleanVar, Job
from nautobot.cloud.models import (
    CloudAccount,
    CloudNetwork,
    CloudNetworkPrefixAssignment,
    CloudResourceType,
    CloudService,
)
from nautobot.dcim.models import Manufacturer
from nautobot.extras.models import Secret
from nautobot.ipam.models import Prefix

CF_API = "https://api.cloudflare.com/client/v4"
CF_ACCOUNT_ID = "393c051f3accf51881d6c98fc5b66c36"

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


class CFEstateSync(Job):
    dryrun = BooleanVar(
        default=True,
        description="Nur berichten, nichts schreiben (empfohlen fuer den ersten Lauf).",
    )

    class Meta:
        name = "CF Estate Sync"
        description = "Cloudflare-Estate (Tunnel/Worker/Access/Hyperdrive/VPC) nach Nautobot syncen (Doku-SSoT)."
        grouping = "TP44 / Cloudflare"
        # Nur ein Boolean (dryrun) als Eingabe; das CF-Token kommt aus dem Nautobot-Secret,
        # nicht aus einer Job-Variable. Ohne diese Zeile (Default True) verweigert Nautobot
        # das Scheduling: "Job may have sensitive input variables".
        has_sensitive_variables = False

    # ---------------------------------------------------------------- CF-Pull
    def _cf_get(self, path):
        """(ok, result) — 403/404 werden abgefangen, damit fehlende Scopes den Sync nicht kippen."""
        req = urllib.request.Request(CF_API + path, headers={"Authorization": "Bearer " + self._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 _ingress(self, tunnel_id):
        ok, res = self._cf_get(f"/accounts/{CF_ACCOUNT_ID}/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(self):
        """networks/services: (name, rt_name, desc, extra, match_key, match_val).
        routes: {tunnel_id: cidr}. skipped: Liste nicht lesbarer Endpoints."""
        A = CF_ACCOUNT_ID
        networks, services, routes, skipped = [], [], {}, []

        _, tuns = self._cf_get(f"/accounts/{A}/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": self._ingress(t["id"])},
                             "tunnel_id", t["id"]))
        _, warps = self._cf_get(f"/accounts/{A}/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"]))

        _, scripts = self._cf_get(f"/accounts/{A}/workers/scripts")
        for s in (scripts or []):
            services.append((s["id"], "Cloudflare Worker", "Cloudflare Worker",
                             {"script": s["id"]}, "script", s["id"]))

        ok, apps = self._cf_get(f"/accounts/{A}/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")))

        ok, hyper = self._cf_get(f"/accounts/{A}/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")))

        ok, vpc = self._cf_get(f"/accounts/{A}/connectivity/directory/services")
        if not ok:
            skipped.append("connectivity/directory/services")
        for v in (vpc or []):
            # VPC-Directory-Services tragen ihren Identifier in "service_id" (NICHT "id").
            sid = v.get("service_id") or v.get("id")
            services.append((v.get("name") or sid, "Cloudflare VPC Service", "VPC-Service",
                             {"service_id": sid, "type": v.get("type")}, "service_id", sid))

        ok, rts = self._cf_get(f"/accounts/{A}/teamnet/routes?per_page=50")
        if not ok:
            skipped.append("teamnet/routes")
        for r in (rts or []):
            if r.get("network") and r.get("tunnel_id"):
                routes[r["tunnel_id"]] = r["network"]

        return networks, services, routes, skipped

    # ---------------------------------------------------------------- ORM-Upsert
    def _upsert(self, model, cache, name, rt, acct, desc, extra, mk, mv, label, seen, dryrun):
        # mv=None nie matchen: sonst greift .get(mk)==None auf das erste Objekt OHNE mk
        # und mischt fremde extra_config zusammen (z.B. VPC ohne id -> App-Launcher).
        if mv is None:
            self.logger.warning("~ %s '%s' ohne Match-Wert (%s) — uebersprungen", label, name, mk)
            return None
        ex = next((o for o in cache if (o.extra_config or {}).get(mk) == mv), None)
        if ex:
            seen.add(ex.id)
            merged = dict(ex.extra_config or {})
            merged.update(extra)
            if merged == (ex.extra_config or {}):
                return ex
            if dryrun:
                self.logger.info("~ %s '%s': extra_config-Update [DRY-RUN]", label, ex.name)
                return ex
            ex.extra_config = merged
            ex.validated_save()
            self.logger.info("~ %s '%s' aktualisiert", label, ex.name)
            return ex
        if dryrun:
            self.logger.info("+ %s '%s' [DRY-RUN]", label, name)
            return None
        obj = model(name=name, cloud_resource_type=rt, cloud_account=acct,
                    description=desc, extra_config=extra)
        obj.validated_save()
        cache.append(obj)
        seen.add(obj.id)
        self.logger.info("+ %s '%s' angelegt", label, name)
        return obj

    def _mark_stale(self, cache, seen, label, dryrun):
        for o in cache:
            extra = o.extra_config or {}
            if o.id in seen:
                if extra.get("stale") and not dryrun:
                    o.extra_config = {k: v for k, v in extra.items() if k != "stale"}
                    o.validated_save()
                    self.logger.info("↩ %s '%s': stale entfernt (wieder da)", label, o.name)
                continue
            if extra.get("stale"):
                continue
            if dryrun:
                self.logger.warning("⚠ %s '%s' fehlt im Live-Pull → wuerde stale markiert [DRY-RUN]", label, o.name)
                continue
            o.extra_config = dict(extra, stale=True)
            o.validated_save()
            self.logger.warning("⚠ %s '%s' als stale markiert (nicht mehr in CF)", label, o.name)

    # ---------------------------------------------------------------- run
    def run(self, dryrun=True):
        try:
            self._tok = Secret.objects.get(name="cf-api-token").get_value()
        except Secret.DoesNotExist:
            raise RuntimeError("Nautobot-Secret 'cf-api-token' fehlt — bitte anlegen (Provider env, CF_API_TOKEN).")

        networks, services, routes, skipped = self._load_cf()
        self.logger.info("Live-Pull: %d CloudNetworks · %d CloudServices%s",
                         len(networks), len(services),
                         f" · SKIP: {', '.join(skipped)}" if skipped else " · alle Endpoints lesbar")

        mfr, _ = Manufacturer.objects.get_or_create(name="Cloudflare")
        acct, created = CloudAccount.objects.get_or_create(
            name="Cloudflare", defaults={"account_number": CF_ACCOUNT_ID, "provider": mfr})
        if created and dryrun:
            pass

        rt_by_name = {}
        for name, app, model_name in RESOURCE_TYPES:
            ct = ContentType.objects.get(app_label=app, model=model_name)
            rt, _ = CloudResourceType.objects.get_or_create(name=name, defaults={"provider": mfr})
            rt.content_types.add(ct)
            rt_by_name[name] = rt

        net_cache = list(CloudNetwork.objects.filter(cloud_account=acct))
        svc_cache = list(CloudService.objects.filter(cloud_account=acct))
        seen_net, seen_svc = set(), set()

        self.logger.info("[CloudNetworks]")
        for name, rt_name, desc, extra, mk, mv in networks:
            self._upsert(CloudNetwork, net_cache, name, rt_by_name.get(rt_name), acct,
                         desc, extra, mk, mv, "CloudNetwork", seen_net, dryrun)

        self.logger.info("[CloudServices]")
        for name, rt_name, desc, extra, mk, mv in services:
            self._upsert(CloudService, svc_cache, name, rt_by_name.get(rt_name), acct,
                         desc, extra, mk, mv, "CloudService", seen_svc, dryrun)

        self.logger.info("[Prefix-Assignments]")
        for tunnel_id, cidr in routes.items():
            net = next((n for n in net_cache if (n.extra_config or {}).get("tunnel_id") == tunnel_id), None)
            if not net or getattr(net, "id", None) is None:
                continue
            pfx = Prefix.objects.filter(prefix=cidr).first()
            if not pfx:
                self.logger.info("  ~ Route %s: Prefix nicht in IPAM → uebersprungen", cidr)
                continue
            if CloudNetworkPrefixAssignment.objects.filter(cloud_network=net, prefix=pfx).exists():
                continue
            if dryrun:
                self.logger.info("  + %s ↔ %s [DRY-RUN]", net.name, cidr)
                continue
            CloudNetworkPrefixAssignment(cloud_network=net, prefix=pfx).validated_save()
            self.logger.info("  + %s ↔ %s angelegt", net.name, cidr)

        self.logger.info("[Verwaisung]")
        # CloudServices nur markieren, wenn ALLE Service-Endpoints lesbar waren — sonst
        # wuerde ein fehlender Scope echte Objekte faelschlich als verwaist markieren.
        svc_safe = not ({"access/apps", "hyperdrive/configs", "connectivity/directory/services"} & set(skipped))
        if svc_safe:
            self._mark_stale(svc_cache, seen_svc, "CloudService", dryrun)
        else:
            self.logger.info("  · CloudService-Verwaisung uebersprungen (Token-Scope unvollstaendig)")
        self._mark_stale(net_cache, seen_net, "CloudNetwork", dryrun)

        self.logger.info("Fertig (%s).", "DRY-RUN — nichts geschrieben" if dryrun else "COMMIT")


# Registrierung zentral in jobs/__init__.py (Nautobot importiert nur das Paket).