Add A / AAAA record of the current host to Cloudflare DNS

cf-ddns.py

#!/usr/bin/env python3

# Add A / AAAA record of the current host to Cloudflare DNS
# Use with cron
# */5 * * * * ~/cf-ddns.py

# Configuration
API_TOKEN = "dc-aacf8c54-74fd6-6f-94ec750f-16629c8"

ZONE_ID = "586672b1f20119e08a7daf544e026ae3"
DOMAIN = "iserver.oooiioooo.cc"
IP = "All"  # "IPv4", "IPv6", "All"

# Code section

ENDPOINT = "https://api.cloudflare.com/client/v4"
HEADERS = {
    "Authorization": "Bearer " + API_TOKEN,
    "Content-Type": "application/json"
}

# Install missing dependencies

import subprocess
import sys

import pkg_resources

required = {'requests'}
installed = {pkg.key for pkg in pkg_resources.working_set}
missing = required - installed

if missing:
    print("Installing missing dependencies...")
    subprocess.check_call([sys.executable, '-m', 'pip', 'install', *missing], stdout=subprocess.DEVNULL)

# Start

import socket
import ipaddress
import requests


# Get ip address


def get_ip_address(ver):
    if ver == 4:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("1.1.1.1", 53))
        return s.getsockname()[0]
    if ver == 6:
        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
        s.connect(("2606:4700:4700::1111", 53, 0, 0))
        return s.getsockname()[0]


ipv4_addr = get_ip_address(4)
ipv4_addr_private = ipaddress.ip_address(ipv4_addr).is_private
ipv6_addr = get_ip_address(6)
ipv6_addr_private = ipaddress.ip_address(ipv6_addr).is_private

print(f"IPv4 address: [{ipv4_addr_private and 'Private' or 'Public'}]\t{ipv4_addr}\n"
      f"IPv6 address: [{ipv6_addr_private and 'Private' or 'Public'}]\t{ipv6_addr}\n")

print("Reading DNS...")
dns_records = None
try:
    res = requests.get(f"{ENDPOINT}/zones/{ZONE_ID}/dns_records", headers=HEADERS)
    if res.status_code == 200 and res.json()['success']:
        dns_records = res.json()['result']
    else:
        print(f"Failed to read DNS\n{res.json()['errors'][0]}")
        exit(1)
except Exception as e:
    print(f"Failed to read DNS\n{e}")
    exit(1)


def set_record(dns_type, content):
    record = next((item for item in dns_records if item["type"] == dns_type and item["name"] == DOMAIN), None)
    if record:
        if record["content"] == content:
            print("IP not changed, skipped")
        else:
            try:
                print(f"Updating {DOMAIN} ({dns_type}): {content} ...")
                res = requests.put(f"{ENDPOINT}/zones/{ZONE_ID}/dns_records/{record['id']}", headers=HEADERS,
                                   json={
                                       "type": dns_type,
                                       "name": DOMAIN,
                                       "content": content,
                                       "ttl": 1,
                                       "proxied": False
                                   })
                if res.status_code == 200 and res.json()['success']:
                    print("Done")
                else:
                    print(f"Failed to update {DOMAIN} ({dns_type})\n{res.json()['errors'][0]}")
            except Exception as e:
                print(f"Failed to update {DOMAIN} ({dns_type})\n{e}")
    else:
        try:
            print(f"Creating {DOMAIN} ({dns_type}): {content} ...")
            res = requests.post(f"{ENDPOINT}/zones/{ZONE_ID}/dns_records", headers=HEADERS, json={
                "type": dns_type,
                "name": DOMAIN,
                "content": content,
                "ttl": 1,
                "proxied": False
            })
            if res.status_code == 200 and res.json()['success']:
                print("Done")
            else:
                print(f"Failed to update {DOMAIN} ({dns_type})\n{res.json()['errors'][0]}")
        except Exception as e:
            print(f"Failed to create {DOMAIN} ({dns_type})\n{e}")


if IP == "All" or IP == "IPv4":
    print("Configuring DNS for IPv4...")
    if ipv4_addr_private:
        print("Private IPv4 address, skipped")
    else:
        set_record("A", ipv4_addr)

if IP == "All" or IP == "IPv6":
    print("Configuring DNS for IPv6...")
    if ipv6_addr_private:
        print("Private IPv6 address, skipped")
    else:
        set_record("AAAA", ipv6_addr)

print("Job completed")