TP44 Teil B: CF-Estate-Sync als Nautobot-Job (ORM, aus Git-Datenquelle)
jobs/cf_estate_sync.py — kanonische Runtime des CF→Nautobot-Sync als geplanter Nautobot-Job im vorhandenen nautobot-worker (kein neuer Container, keine Nautobot-Exposure). Portierung der Reconcile-Logik aus nautobot/tp44-cf-sync.py von REST auf ORM: - ID-Matching (tunnel_id/app-id/script/service_id), extra_config-Merge - Prefix-Assignments live aus teamnet/routes (subsumiert Seeder [7]) - Verwaisung = stale-Markierung statt Loeschen, Scope-sicher uebersprungen - CF-Token aus Nautobot-Secret 'cf-api-token', dryrun-BooleanVar default True Import im Worker verifiziert (IMPORT_OK), CF-Egress vom Worker bestaetigt. Governance: periodische externe→Nautobot-Syncs = Jobs hier, kein per-Task-Container. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
256
jobs/cf_estate_sync.py
Normal file
256
jobs/cf_estate_sync.py
Normal file
@@ -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)
|
||||
Reference in New Issue
Block a user