Cloudflare "DynamicDNS" - Automate DNS records to automatically update IPs from non-static ISP using Cloudflare API

cloudflare-dns-update.py

#!/usr/bin/env python3

"""Cloudflare "DynamicDNS"

This script automate DNS records to automatically update IPs from non-static ISP using Cloudflare API.

Recommendation:
 - Create a 'cloudflare' folder inside user home:
mkdir ~/cloudflare

 - Create the python file:
vim ~/cloudflare/cloudflare-dns-update.py

 - Cron Job (Every 10 minutes):
*/10 * * * * cd ~/cloudflare && python3 ./cloudflare-dns-update.py

Get the Global API Key:
[https://dash.cloudflare.com/profile/api-tokens]

DNS Record Update documentation:
[https://api.cloudflare.com/#dns-records-for-a-zone-update-dns-record]
"""

import os
import logging
import pickle
import json
import requests


# Config variables.
zone = 'yourdomain.com'
email = 'your@email.com'
api_key = 'yourGlobalApiKey'


# Configure log format.
log_file = 'cloudflare_dns.log'
logging.basicConfig(filename=log_file,
                    filemode='a',
                    format='%(asctime)s %(name)s %(levelname)s %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S',
                    level=logging.INFO)


# Handle IPs
def get_ips(file_name='ip_addresses.pickle'):
    # Get IPv4 and IPv6 from this host.
    # Warning: Whitelist jsonip.com if you use PiHole, AdGuard or any AD blocking service.
    ip_addresses_new = {
        'ipv4': requests.get('https://ipv4.jsonip.com').json()['ip'],
        'ipv6': requests.get('https://ipv6.jsonip.com').json()['ip'],
    }

    logging.info(ip_addresses_new)

    # Check if 'file_name' exists.
    if os.path.exists(file_name):
        # If the file exists, compare the ip values to see if any has changed.
        with open(file_name, 'rb') as file:
            ip_addresses = pickle.load(file)
        
        if ip_addresses_new['ipv4'] == ip_addresses['ipv4']:
            ip_addresses['ipv4'] = 'unchanged'
        else:
            ip_addresses['ipv4'] = ip_addresses_new['ipv4']
            with open(file_name, 'wb') as file:
                pickle.dump(ip_addresses_new, file)

        if ip_addresses_new['ipv6'] == ip_addresses['ipv6']:
            ip_addresses['ipv6'] = 'unchanged'
        else:
            ip_addresses['ipv6'] = ip_addresses_new['ipv6']
            with open(file_name, 'wb') as file:
                pickle.dump(ip_addresses_new, file)

    else:
        # Else, create a new file with the IPs values.
        with open(file_name, 'wb') as file:
            pickle.dump(ip_addresses_new, file)
        ip_addresses = ip_addresses_new
    
    return ip_addresses



# Handle Cloudflare API
def update_dns(zone, email, api_key, ip_addresses):

    # If none of the IPs have renewed, terminate the script
    if ip_addresses['ipv4'] == 'unchanged' and ip_addresses['ipv6'] == 'unchanged':
        logging.info('=== Both IPv4 and IPv6 are unchanged, skipping configuration. ===')
        exit()


    # Required API values
    headers = {'X-Auth-Key': api_key, 'X-Auth-Email': email, 'Content-type': 'application/json'}
    base_url = 'https://api.cloudflare.com/client/v4'


    # Get Zone ID
    url_path = '/zones'
    response = requests.get(base_url + url_path, headers=headers)
    try:
        for res in response.json()['result']:
            if res['name'] == zone:
                zone_id = res['id']
    except:
        logging.error('=!= Could not retrieve Zone ID, check if API Key is correct =!=')
        logging.info(response.json())
        exit()


    # Get list of Zone DNS records
    url_path = f"/zones/{zone_id}/dns_records"
    response = requests.get(base_url + url_path, headers=headers)
    records_ipv4 = {}
    records_ipv6 = {}
    for res in response.json()['result']:
        if res['type'] == 'A':
            records_ipv4[res['id']] = res['name']
        elif res['type'] == 'AAAA':
            records_ipv6[res['id']] = res['name']


    # Check if IPv4 has changed
    if ip_addresses['ipv4'] != 'unchanged':
        # Update Zone Records.
        logging.info(f"=== Updating IPv4: {ip_addresses['ipv4']} ===")
        for record_id, name in records_ipv4.items():
            request = requests.put(url = f"{base_url}/zones/{zone_id}/dns_records/{record_id}", headers=headers, data=json.dumps({'type': 'A', 'name': name, 'content': ip_addresses['ipv4'], 'proxied': True})).json()
            logging.info(f"{name} | Success: {request['success']} | Errors: {request['errors']}")   
    else:
        logging.info('=== Unchanged IPv4, skipping configuration. ===')
    

    # Check if IPv6 has changed
    if ip_addresses['ipv6'] != 'unchanged':
        # Update Zone Records.
        logging.info(f"=== Updating IPv6: {ip_addresses['ipv6']} ===")
        for record_id, name in records_ipv6.items():
            request = requests.put(url = f"{base_url}/zones/{zone_id}/dns_records/{record_id}", headers=headers, data=json.dumps({'type': 'AAAA', 'name': name, 'content': ip_addresses['ipv6'], 'proxied': True})).json()
            logging.info(f"{name} | Success: {request['success']} | Errors: {request['errors']}")   
    else:
        logging.info('=== Unchanged IPv6, skipping configuration. ===')


if __name__ == '__main__':
    update_dns(zone, email, api_key, ip_addresses=get_ips())