pantheon/cloudflare/sync.py

#!/usr/bin/env python3

import json
import os
from datetime import datetime
from pathlib import Path

import requests
import yaml
from dotenv import load_dotenv

load_dotenv(Path(__file__).with_name(".env"))

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

if not TOKEN:
    raise RuntimeError("CF_API_TOKEN missing")

if not ZONE:
    raise RuntimeError("CF_ZONE_ID missing")

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

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

# -----------------------------------------------------
# Backup current DNS
# -----------------------------------------------------

print("Backing up current Cloudflare zone...")

records = requests.get(API, headers=HEADERS).json()["result"]

backup_dir = Path("backups")
backup_dir.mkdir(exist_ok=True)

backup_file = backup_dir / f"{datetime.now():%Y-%m-%d_%H-%M-%S}.yaml"

yaml.safe_dump(records, open(backup_file, "w"), sort_keys=False)

print(f"Backup saved to {backup_file}")

# -----------------------------------------------------
# Load execution plan
# -----------------------------------------------------

plan = json.load(open("plan.json"))

def print_section(title, symbol, items):

    print()
    print("=" * 70)
    print(title)
    print("=" * 70)

    if not items:
        print("  None")
        return

    for item in items:

        if "name" in item:
            print(f" {symbol} {item['name']}")

        elif "old" in item:
            print(f" {symbol} {item['old']}  ->  {item['new']}")


print("\nExecution Plan")
print("=" * 70)

print_section(
    f"CREATE ({len(plan['create'])})",
    "+",
    sorted(plan["create"], key=lambda x: x["name"])
)

print_section(
    f"RENAME ({len(plan['rename'])})",
    "~",
    sorted(plan["rename"], key=lambda x: x["old"])
)

print_section(
    f"UPDATE ({len(plan['update'])})",
    "*",
    sorted(plan["update"], key=lambda x: x["name"])
)

print_section(
    f"DELETE ({len(plan['delete'])})",
    "-",
    sorted(plan["delete"], key=lambda x: x["name"])
)

print()
answer = input("Apply these changes to Cloudflare? [y/N]: ").strip().lower()

if answer not in ("y", "yes"):
    print("\nAborted.")
    raise SystemExit

if answer != "y":
    print("Cancelled.")
    raise SystemExit

# -----------------------------------------------------
# CREATE
# -----------------------------------------------------

print("\nCreating records...")

for item in plan["create"]:

    body = {
        "type": item["desired"]["type"],
        "name": item["name"],
        "content": item["desired"]["content"],
        "ttl": item["desired"]["ttl"],
        "proxied": item["desired"]["proxied"],
    }

    r = requests.post(API, headers=HEADERS, json=body)
    r.raise_for_status()

    print(f" + {item['name']}")

# -----------------------------------------------------
# RENAME
# -----------------------------------------------------

print("\nRenaming records...")

for item in plan["rename"]:

    body = {
        "type": item["desired"]["type"],
        "name": item["new"],
        "content": item["desired"]["content"],
        "ttl": item["desired"]["ttl"],
        "proxied": item["desired"]["proxied"],
    }

    r = requests.patch(
        f"{API}/{item['id']}",
        headers=HEADERS,
        json=body,
    )

    r.raise_for_status()

    print(f" ~ {item['old']} -> {item['new']}")

# -----------------------------------------------------
# UPDATE
# -----------------------------------------------------

print("\nUpdating records...")

for item in plan["update"]:

    body = {
        "type": item["desired"]["type"],
        "name": item["name"],
        "content": item["desired"]["content"],
        "ttl": item["desired"]["ttl"],
        "proxied": item["desired"]["proxied"],
    }

    r = requests.patch(
        f"{API}/{item['id']}",
        headers=HEADERS,
        json=body,
    )

    r = requests.patch(
    f"{API}/{item['id']}",
    headers=HEADERS,
    json=body,
)

    if not r.ok:
        print(f"\nERROR updating {item['name']}")
        print(r.status_code)
        print(r.text)
        raise SystemExit

    print(f" * {item['name']}")


# -----------------------------------------------------
# DELETE
# -----------------------------------------------------

print("\nDeleting records...")

for item in plan["delete"]:

    r = requests.delete(
        f"{API}/{item['id']}",
        headers=HEADERS,
    )

    r.raise_for_status()

    print(f" - {item['name']}")

print("\nDone.")