pantheon/cloudflare/diff.py

#!/usr/bin/env python3

import json
import os

import requests
import yaml
from dotenv import load_dotenv

load_dotenv()


TOKEN = os.getenv("CF_API_TOKEN")
ZONE = os.getenv("CF_ZONE_ID")

HEADERS = {
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json",
}

def dns_key(record_type: str, name: str) -> str:
    return f"{record_type}:{name}"

# ------------------------------------------------------------
# Load desired configuration
# ------------------------------------------------------------

with open("dns.yaml", "r") as f:
    config = yaml.safe_load(f)

zone = config["zone"]

defaults = config.get("defaults", {})
aliases = config.get("aliases", {})
ignore = set(config.get("ignore", []))

default_type = defaults.get("type", "A")
default_content = defaults.get("content")
default_ttl = defaults.get("ttl", 1)
default_proxied = defaults.get("proxied", False)

desired = {}

for name, record in config.get("records", {}).items():

    fqdn = zone if name == "@" else f"{name}.{zone}"

    rtype = record.get("type", default_type)
    # Only manage A records for now
    if rtype != "A":
        continue
    key = dns_key(rtype, fqdn)

    desired[key] = {
        "name": key,
        "type": rtype,
        "content": record.get("content", default_content),
        "ttl": record.get("ttl", default_ttl),
        "proxied": record.get("proxied", default_proxied),
    }

# ------------------------------------------------------------
# Download Cloudflare records
# ------------------------------------------------------------

url = f"https://api.cloudflare.com/client/v4/zones/{ZONE}/dns_records"

response = requests.get(url, headers=HEADERS)
response.raise_for_status()

current = {}

for r in response.json()["result"]:

    if r["type"] != "A":
        continue

    key = f"{r['type']}:{r['name']}"

    current[key] = {
        "id": r["id"],
        "type": r["type"],
        "content": r["content"],
        "ttl": r["ttl"],
        "proxied": r.get("proxied", False),
    }

# ------------------------------------------------------------
# Build execution plan
# ------------------------------------------------------------

plan = {
    "create": [],
    "update": [],
    "rename": [],
    "delete": [],
}

handled_current = set()

# ------------------------------------------------------------
# Creates / Updates / Renames
# ------------------------------------------------------------

for key, wanted in desired.items():

    fqdn = wanted["name"]
    short = fqdn.replace(f".{zone}", "")

    #
    # Existing record?
    #

    if key in current:

        handled_current.add(key)

        existing = current[key]

        changed = False

        for key in ("type", "content", "ttl", "proxied"):
            if existing[key] != wanted[key]:
                print(f"\nDifference: {fqdn}")
                print(f"  {key}")
                print(f"    Current : {existing[key]!r}")
                print(f"    Wanted  : {wanted[key]!r}")
                changed = True

        if changed:
            plan["update"].append({
                "name": key,
                "id": existing["id"],
                "desired": wanted,
            })

        continue

    #
    # Alias?
    #

    old = None

    for alias, target in aliases.items():

        if target != short:
            continue

        alias_key = dns_key(wanted["type"], f"{alias}.{zone}")

        if alias_key in current:

            handled_current.add(alias_key)

            plan["rename"].append({
                "old": alias_fqdn,
                "new": fqdn,
                "id": current[alias_key]["id"],
                "desired": wanted,
            })

            old = alias_key
            break

    if old:
        continue

    #
    # Create
    #

    plan["create"].append({
        "name": key,
        "desired": wanted,
    })

# ------------------------------------------------------------
# Deletes
# ------------------------------------------------------------

for key, existing in current.items():

    _, fqdn = key.split(":", 1)

    short = fqdn.replace(f".{zone}", "")

    if short in ignore:
        continue

    if key in handled_current:
        continue

    if key in desired:
        continue

    plan["delete"].append({
        "name": key,
        "id": existing["id"],
    })

# ------------------------------------------------------------
# Print
# ------------------------------------------------------------

def heading(title):
    print()
    print("=" * 70)
    print(title)
    print("=" * 70)

heading(f"CREATE ({len(plan['create'])})")

for item in sorted(plan["create"], key=lambda x: x["name"]):
    print(item["name"])

heading(f"RENAME ({len(plan['rename'])})")

for item in sorted(plan["rename"], key=lambda x: x["old"]):
    print(f"{item['old']}  ->  {item['new']}")

heading(f"UPDATE ({len(plan['update'])})")

for item in sorted(plan["update"], key=lambda x: x["name"]):
    print(item["name"])

heading(f"DELETE ({len(plan['delete'])})")

for item in sorted(plan["delete"], key=lambda x: x["name"]):
    print(item["name"])

# ------------------------------------------------------------
# Save plan
# ------------------------------------------------------------

with open("plan.json", "w") as f:
    json.dump(plan, f, indent=4)

print()
print("Execution plan written to plan.json")