""" 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).