Cloudflare Dynamic DNS

cloudflare_ddns.py

import json
import os
import sys
import tempfile
from typing import cast

import requests
from requests.models import Response

CLOUDFLARE_API_TOKEN = "YOUR_API_TOKEN"
ZONE_NAME = "example.com"  # Your root domain
RECORD_NAME = "dynamic.example.com"  # The subdomain to update

REQ_HEADERS = {
    "Authorization": f"Bearer {CLOUDFLARE_API_TOKEN}",
    "Content-Type": "application/json",
}

STATE_FILE = os.path.join(tempfile.gettempdir(), "cloudflare_ddns.json")


def load_state() -> dict:
    if not os.path.exists(STATE_FILE):
        return {}
    try:
        with open(STATE_FILE, mode="r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {}


def save_state(state: dict):
    try:
        with open(STATE_FILE, "w", encoding="utf-8") as f:
            json.dump(state, f)
    except Exception as e:
        print(f"Warning, failed to write state file: {e}")


def handle_response(
    resp: Response, http_err_msg: str, api_err_msg: str = None
) -> dict | str:
    if resp.headers.get("Content-Type", "").startswith("application/json"):
        data = cast(dict, resp.json())
        err_text = json.dumps(data, indent=2)
    else:
        data = resp.text
        err_text = data

    if not resp.ok:
        print(f"{http_err_msg} ({resp.status_code})\n{err_text}")
        sys.exit(1)

    if isinstance(data, str):
        return data

    if not data.get("success") or not data.get("result"):
        print(f"{api_err_msg}\n{err_text}")
        sys.exit(1)
    return data


def get_external_ip() -> str:
    resp = requests.get("https://api.ipify.org", timeout=10)
    text = handle_response(resp, "Failed to fetch external IP")
    return text.strip()


def get_zone_id() -> str:
    url = f"https://api.cloudflare.com/client/v4/zones?name={ZONE_NAME}"
    resp = requests.get(url, headers=REQ_HEADERS, timeout=10)
    data = handle_response(
        resp, "Failed to fetch zone info", f"Could not find zone: {ZONE_NAME}"
    )
    return data["result"][0]["id"]


def get_record_id(zone_id: str) -> tuple[str, str]:
    url = f"https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records?type=A&name={RECORD_NAME}"
    resp = requests.get(url, headers=REQ_HEADERS, timeout=10)
    data = handle_response(
        resp, "Failed to fetch DNS record", f"Could not find record: {RECORD_NAME}"
    )
    rec = data["result"][0]
    return rec["id"], rec["content"]


def update_dns_record(zone_id: str, record_id: str, new_ip: str):
    resp = requests.patch(
        f"https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records/{record_id}",
        json={
            "type": "A",
            "content": new_ip,
            "ttl": 1,  # 1 means "auto" (usually 5 min)
            "proxied": False,  # Dynamic DNS should not be proxied
        },
        headers=REQ_HEADERS,
        timeout=10,
    )
    handle_response(
        resp,
        "Failed to update DNS record",
        "Failed to update DNS record",
    )


def main():
    state = load_state()

    current_ip = get_external_ip()
    print(f"Current external IP: {current_ip}")

    if state.get("last_external_ip") == current_ip:
        print(f"No update needed, external IP is unchanged: {current_ip}")
        sys.exit(0)
    state["last_external_ip"] = current_ip

    zone_id = state.get("zone_id") or get_zone_id()
    state["zone_id"] = zone_id

    record_id, old_ip = get_record_id(zone_id)

    if old_ip == current_ip:
        print(f"No update needed, DNS record is already set to {current_ip}")
    else:
        update_dns_record(zone_id, record_id, current_ip)
        print(f"DNS record updated from {old_ip} to {current_ip}")

    save_state(state)


if __name__ == "__main__":
    main()