Quick Cloudflare Dynamic DNS Script Using UPnP

I have an OpenWRT router which is constantly running as a switch, (it's an old Netgear DGN3500), I recently installed a USB drive and Python3 onto this which makes it a really powerful little thing.

I also have a home server - but any smaller jobs I can migrate from it, the better. I created the below script to constantly ping Cloudflare using UPnP to update my external IP Address onto an A record.

I like this approach as it requires no external server to tell me my address - just the routers UPnP service, it also self-recovers and logs any occurrences of an internet outage, as well as previous IP addresses.

I use this for things like a home VPN and a self hosted Gitlab.

To run this on windows, you can use the below batch file coupled with a hidden scheduled task.

@echo off
python cf.py
exit /b 0

Replace the following variables:

VALUE_TO_UPDATE: The name of the record - e.g. vpn.example.com
CF_EMAIL: Cloudflare email
CF_API_KEY: Cloudflare API key
CF_RECORD_URL: The URL of the resource to update e.g. https://cloudflare_api_url/zones/3b...c5/dns_records/30...22

import upnpclient
import pickle
from datetime import datetime
import json
import requests
import os

try:
    if not os.path.exists("igd.p"):
        pickle.dump({}, open("igd.p", "wb"))

    def log_format(msg):
        time = datetime.now().strftime("%m/%d/%Y, %H:%M:%S")
        return f"[{time}] - {msg}"

    def log_msg(msg, file='idg.log'):
        with open(file, "a") as f:
            f.write(log_format(msg))

    def dump_idg(ip, status, last_error, uptime):
        igd = {
            "ip": ip,
            "status": status,
            "last_error": last_error,
            "uptime": uptime
        }

        pickle.dump(igd, open("igd.p", "wb"))

    device = upnpclient.Device("http://192.168.4.1:1900/igd.xml")
    # devices = upnpclient.discover() - for auto discovery

    igd = pickle.load(open("igd.p", "rb"))

    ip = device.WANIPConnection \
        .GetExternalIPAddress().get('NewExternalIPAddress')

    stats = device.WANIPConnection.GetStatusInfo()

    status = stats.get('NewConnectionStatus')
    last_error = stats.get('NewLastConnectionError')
    uptime = stats.get('NewUptime')

    if (
        ip != igd.get("ip") or
        status != igd.get("status") or
        last_error != igd.get("last_error") or
        int(uptime) < int(igd.get("uptime")
                          )):

        if (status != "Connected"):
            log_msg(status)
            dump_idg(ip, status, last_error, uptime)
            exit()

        log_msg(json.dumps(igd))

        data = {
            "type": "A",
            "name": "VALUE_TO_UPDATE",
            "content": ip,
            "ttl": 600
        }

        headers = {
            "X-Auth-Email": "CF_EMAIL",
            "X-Auth-Key": "CF_API_KEY",
            "Content-Type": "application/json"
        }

        resp = requests.put("CF_RECORD_URL",
                            data=json.dumps(data), headers=headers)

    dump_idg(ip, status, last_error, uptime)
except Exception as e:
    log_msg(str(e), 'exceptions.log')