Christoph Emsenhuber
43c592e5d4
Dry-Run-Befund: CloudService 'terpene-db' (VPC) wurde faelschlich als stale markiert und fremde extra_config in 'App Launcher' gemischt. Ursache: VPC- Directory-Services tragen ihren Identifier in "service_id", der Job las "id" (=None). Ueber .get(mk)==None matchte der None-Wert das erste Objekt ohne den Key. Fix: VPC nutzt v["service_id"]; _upsert ueberspringt mv=None generell. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
264 lines
12 KiB
Python
264 lines
12 KiB
Python
"""
|
|
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"
|
|
|
|
# ---------------------------------------------------------------- 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).
|