diff --git a/jobs/cf_estate_sync.py b/jobs/cf_estate_sync.py new file mode 100644 index 0000000..907677d --- /dev/null +++ b/jobs/cf_estate_sync.py @@ -0,0 +1,256 @@ +""" +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, register_jobs +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" + + # ---------------------------------------------------------------- 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 []): + 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"))) + + 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): + 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") + + +register_jobs(CFEstateSync)