ipupdater.py

ipupdater.py

#!/usr/bin/env python3

'''
This script gets the public IPv4 Address from your Fritz!Box (!!!)
and the public IPv6 from your Interface.

When they change, the specified cloudflare dny record will get updated.

Adjust your settings in the config section (this is nothing serious, hence
no proper external config).

The dependencies can be installed with:
$ pip3 install cloudflare fritzconnection sh
'''


from CloudFlare import CloudFlare
from fritzconnection import FritzConnection
from sh import ip
from time import sleep


# ------------------------------
# Config
FRITZBOX_IP='IP_OF_YOUR_ROUTER'
INTERFACE_WITH_IPV6='eth0'
IP_UPDATE_CHECK_INTERVAL=60*5                                     # check every x seconds for changed ipv4 and ipv6
IPV4_DOMAIN_NAMES = [ 'some.domain.tld' ]                         # Requires an existing A dns entry per domain
IPV6_DOMAIN_NAMES = [ 'some.domain.tld' ]                         # Requires an existing AAAA dns entry per domain
CLOUDFLARE_API_TOKEN = '0123456789012345678901234567890123456789' # Must be able to edit the zones used in IPVX_DOMAIN_NAMES
# ------------------------------

ZONE_ID_CACHE = {}

fritz = FritzConnection(address=FRITZBOX_IP)
cloudflare = CloudFlare(token=CLOUDFLARE_API_TOKEN)


#
# Get own public IPs
#

def getPublicIPv4():
    return fritz.call_action('WANIPConn1', 'GetExternalIPAddress')['NewExternalIPAddress']

#def getIPv6Prefix():
#    ''' Not used atm, maybe find the interface device id to get
#        a proper ipv6 address that way rather than with the
#        current implementation of getPublicIpv6() '''
#    res = fritz.call_action('WANIPConn1', 'X_AVM_DE_GetIPv6Prefix')
#    return res['NewIPv6Prefix'], res['NewPrefixLength']

def getPublicIPv6():
    '''Get IPv6 address from `ip show` command'''
    lines = str(ip('address', 'show', INTERFACE_WITH_IPV6)).split('\n')
    ipv6line = next(filter(lambda l: 'inet6' in l and 'global' in l and 'deprecated' not in l, lines))
    return ipv6line.split()[1].split('/')[0]


#
# Interact with Cloudflares DNS entries
#

def toZoneName(domain):
    return '.'.join(domain.split('.')[-2:])

def getZoneId(domain):
    name = toZoneName(domain)
    if name in ZONE_ID_CACHE:
        return ZONE_ID_CACHE[name]
    else:
        # The api call seems a bit hacky. Is there a official way
        # to do this without the 'com.cloudflare.api.account.zone.list'
        # perm and hacking '?name=' into the path?
        zone_id = cloudflare.zones(params={'name': name})[0]['id']
        ZONE_ID_CACHE[name] = zone_id
        return zone_id

def getCloudflareDnsRecord(domain, dns_type):
    records = cloudflare.zones.dns_records(getZoneId(domain), params={'name': domain, 'type': dns_type})
    return records[0] if len(records) == 1 else None

def getCloudflareIPv4(domain):
    record = getCloudflareDnsRecord(domain, 'A')
    return record['content'] if record is not None else None

def getCloudflareIPv6(domain):
    record = getCloudflareDnsRecord(domain, 'AAAA')
    return record['content'] if record is not None else None

def updateCloudflareDnsRecordContent(domain, dns_type, new_content, ttl=1) -> bool:
    record = getCloudflareDnsRecord(domain, dns_type)
    if not record:
        return False
    records = cloudflare.zones.dns_records.put(record['zone_id'], record['id'],
        data={'name': domain, 'type': dns_type, 'ttl': ttl, 'content': new_content}) # ttl 1 == 'automatic'
    return True

def updateCloudflareIPv4(domain, new_ip):
    if updateCloudflareDnsRecordContent(domain, 'A', new_ip):
        print('Updated entry %s to %s' % (domain, new_ip))
        return True
    else:
        print('Failed to update entry %s to %s' % (domain, new_ip))
        return False

def updateCloudflareIPv6(domain, new_ip):
    if updateCloudflareDnsRecordContent(domain, 'AAAA', new_ip):
        print('Updated entry %s to %s' % (domain, new_ip))
        return True
    else:
        print('Failed to update entry %s to %s' % (domain, new_ip))
        return False


#
# Main / Update logic
#

def main():
    last_ipv4 = getPublicIPv4()
    last_ipv6 = getPublicIPv6()

    print('IPv4: %s | IPv6: %s' % (last_ipv4, last_ipv6))

    print('Ensuring Cloudflare domains are up-to-date')
    for domain in IPV4_DOMAIN_NAMES:
        if last_ipv4 != getCloudflareIPv4(domain):
            updateCloudflareIPv4(domain, last_ipv4)
    for domain in IPV6_DOMAIN_NAMES:
        if last_ipv6 != getCloudflareIPv6(domain):
            updateCloudflareIPv6(domain, last_ipv6)

    while True:
        sleep(IP_UPDATE_CHECK_INTERVAL)

        ipv4 = getPublicIPv4()
        ipv6 = getPublicIPv6()

        if ipv4 != last_ipv4:
            print('The public IPv4 changed from %s to %s' % (last_ipv4, ipv4))
            for domain in IPV4_DOMAIN_NAMES:
                if ipv4 != getCloudflareIPv4(domain):
                    updateCloudflareIPv4(domain, ipv4)
            last_ipv4 = ipv4


        if ipv6 != last_ipv6:
            print('The public IPv6 changed from %s to %s' % (last_ipv6, ipv6))
            for domain in IPV6_DOMAIN_NAMES:
                if ipv6 != getCloudflareIPv6(domain):
                    updateCloudflareIPv6(domain, ipv6)
            last_ipv6 = ipv6

if __name__ == '__main__':
    main()

ipupdater.service

[Unit]
Description=Keep cloudflare dns entries up-to-date

[Service]
ExecStart=/path/to/ipupdater.py
User=your_user

[Install]
WantedBy=multi-user.target