WAN DNS Loadbalancing via Cloudflare – automatic IP address updating

flaredns.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-


# / *************************************
# (c) Rau Systemberatung 2025, alfonsrv
# Kurzbeschreibung des Skripts in einem Satz
# \ *************************************

__author__      = 'github/alfonsrv'
__copyright__   = '(c) Rau Systemberatung GmbH, 2025'
__version__      = '0.2'
__python__      = '3.11'


import os
import sys
import traceback
import json
from email.mime.text import MIMEText
import smtplib
import logging
import logging.handlers

import requests
from dotenv import load_dotenv


load_dotenv()
try:
    CF_KEY = os.environ['CF_KEY']
    CF_ZONE_ID = os.environ['CF_ZONE_ID']
    CF_RECORD_ID = os.environ['CF_RECORD_ID']
except KeyError as e:
    logger.error(f'Cannot load environment variable - please configure first! "{e}"')
    exit(2)


os.chdir(os.path.realpath(os.path.join(os.getcwd(), os.path.dirname(__file__))))
TRACKING_FILE = os.path.join(os.getcwd(), 'previous-ip.txt')

LOG_LEVEL = logging.INFO  # logging.DEBUG // .ERROR...
LOG_FILE = 'FlareDNS.log'
LOG_PATH = os.path.join(os.getcwd(), LOG_FILE)
logging.basicConfig(
    format='%(asctime)s - [%(levelname)s] %(message)s',
    level=LOG_LEVEL,
    handlers=[
        logging.StreamHandler(),
        logging.handlers.RotatingFileHandler(
            LOG_PATH,
            maxBytes=10000000,   # 10 MB
            backupCount=5
        )
    ]
)
logger = logging.getLogger(__name__)


def send_mail(subject, body='', recipients: list = None):
    pass


def global_exceptionz(exc_type, exc_value, exc_traceback, log=True):
    """
    Intended to be assigned to sys.exception as a hook.
    Gives programmer opportunity to do something useful with info from uncaught exceptions.
    """
    import sys
    if issubclass(exc_type, KeyboardInterrupt):
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return

    traceback_details = ''.join(traceback.extract_tb(exc_traceback).format())
    if log:
        logger.exception(f'An Unexpected Exception occurred:\n {traceback_details} \n{exc_type} {exc_value}')

    error_msg = 'An exception has been raised outside of a try/except. – Error Details:\n\n' \
                f'Traceback (most recent call last): \n{traceback_details}\n' \
                f'{exc_type} {exc_value}\n'

    send_mail(subject=f'[ERROR] Cloudflare DNS', body=error_msg)


def ip_address():
    return requests.get('https://api.ipify.org?format=json').json()['ip']


def previous_ip_address(*, ip_address: str) -> bool:
    last_ip = ''
    if os.path.exists(TRACKING_FILE):
        with open(TRACKING_FILE, 'r') as f:
            try:
                last_ip = json.load(f).get('last_ip')
            except Exception:
                last_ip = ''
    
    if ip_address != last_ip:
        with open(TRACKING_FILE, 'w') as f:
            json.dump({'last_ip': ip_address}, f)
    return last_ip or False


def update_cloudflare(*, ip_address: str):
    payload = {
        'content': ip_address,
        'ttl': 60,
        'type': 'A'
    }
    headers = {
        'Authorization': f'Bearer {CF_KEY}'
    }

    r = requests.patch(
        f'https://api.cloudflare.com/client/v4/zones/{CF_ZONE_ID}/dns_records/{CF_RECORD_ID}',
        json=payload,
        headers=headers
    )
    logger.info(f'Updated Cloudflare: {r.text}')
    assert r.status_code == 200


def current_cloudflare() -> str:
    headers = {
        'Authorization': f'Bearer {CF_KEY}'
    }

    r = requests.get(
        f'https://api.cloudflare.com/client/v4/zones/{CF_ZONE_ID}/dns_records/{CF_RECORD_ID}',
        headers=headers
    )
    logger.info(f'Current Cloudflare: {r.text}')
    assert r.status_code == 200
    return json.dumps(r.json(), indent=4)
    #return r.json()['result']['content']


if __name__ == '__main__':
    ip_address = ip_address()
    logger.info(f'Current {ip_address=}')
    previous_ip = previous_ip_address(ip_address=ip_address)
    if previous_ip and previous_ip == ip_address:
        logger.debug(f'{previous_ip=} == {ip_address=}')
        exit()

    logger.warning(f'New IP address! {previous_ip=} ==> {ip_address=}')
    update_cloudflare(ip_address=ip_address)
    send_mail(subject=f'Main WAN is now {ip_address}', body=current_cloudflare())