IPv6 Dynamic DNS (DDNS) for Cloudflare

ddns-ipv6-cloudflare.py

#!/bin/env python3

import requests
import json
import subprocess
import time
import sys

dns_cloudflare_email = 'example@gmail.com'
dns_cloudflare_api_key = 'YOUR_API_KEY_HERE'
domain = 'example.com'
# your NIC name like eth0, enp30s0. Usually it's fine to leave it empty.
nic_device = ''


def _set_cloudflare_auth_headers():
    s.headers['X-Auth-Email'] = dns_cloudflare_email
    s.headers['X-Auth-Key'] = dns_cloudflare_api_key
    return


# global requests session for a keep-alive connection.
s = requests.Session()
_set_cloudflare_auth_headers()


def get_ipv6_addr() -> str:
    addr = subprocess.check_output("ip -o -6 addr show %s scope global dynamic| awk '{print $4}' | cut -d/ -f1 | head -n 1" % nic_device, shell=True)
    return addr.strip().decode('ascii')


def _get_cloudflare_results(r:requests.Response) -> list:
    r.raise_for_status()
    return r.json().get('result', [])


def _get_cloudflare_result(r:requests.Response) -> dict:
    res = _get_cloudflare_results(r)
    assert len(res) == 1, 'there should be exactly one result'
    return res[0]

def fetch_cloudflare_account():
    r = s.get('https://api.cloudflare.com/client/v4/accounts')
    res = _get_cloudflare_result(r)
    account_id, account_name = res.get('id'), res.get('name')
    assert account_id is not None and account_name is not None
    return account_id, account_name


def fetch_cloundflare_zone_id(account_id:str, account_name:str, domain:str) -> str:
    root_domain = domain.rsplit('.', 2)[-2:]
    assert len(root_domain)==2
    params = {
        'name': '.'.join(root_domain),
        'status': 'active',
        'account.id': account_id,
        'account.name': account_name,
    }
    r = s.get('https://api.cloudflare.com/client/v4/zones', params=params)
    res = _get_cloudflare_result(r)
    zone_id = res.get('id')
    assert zone_id is not None
    return zone_id


def fetch_cloundflare_dns(zone_id:str, domain:str, record_type:str = 'AAAA'):
    url = 'https://api.cloudflare.com/client/v4/zones/{}/dns_records'.format(zone_id)
    params = {'name': domain}
    r = s.get(url, params=params)
    res = _get_cloudflare_results(r)
    dns_id, addr = '', ''
    print('Effected DNS records:')
    for rc in res:
        print('{}\t{}\t{}\t{}'.format(
            rc.get('name'), rc.get('type'), rc.get('content'), rc.get('modified_on'),
        ))
        if rc.get('type') == record_type:
            dns_id, addr = rc.get('id'), rc.get('content')
    return dns_id, addr


def refresh_cloudflare_ipv6_dns(domain:str, addr:str, record_type:str = 'AAAA'):
    account_id, account_name = fetch_cloudflare_account()
    zone_id = fetch_cloundflare_zone_id(account_id, account_name, domain)
    dns_id, effected_addr = fetch_cloundflare_dns(zone_id, domain, record_type)
    if effected_addr == addr:
        print('No need to update.')
    elif dns_id == '':
        print('There is no {} record for {} yet. Create one.'.format(record_type, domain))
        url = 'https://api.cloudflare.com/client/v4/zones/{}/dns_records'.format(zone_id)
        data = {'type': record_type, 'name': domain, 'content': addr, 'ttl': 1}
        r = s.post(url, json=data)
        r.raise_for_status()
    else:
        url = 'https://api.cloudflare.com/client/v4/zones/{}/dns_records/{}'.format(zone_id, dns_id)
        data = {'type': record_type, 'name': domain, 'content': addr}
        r = s.patch(url, json=data)
        r.raise_for_status()
    return


if __name__ == "__main__":
    last_addr = ''
    while True:
        local_addr = get_ipv6_addr()
        delay = 30
        if local_addr != last_addr:
            print('Detect local IP changed: from "{}" to "{}"'.format(last_addr, local_addr))
            print('Refresh remote DNS record.')
            try:
                refresh_cloudflare_ipv6_dns(domain, local_addr)
                last_addr = local_addr
                delay = 30
            except requests.HTTPError as e:
                print('{}\n{}'.format(e, e.response.json()))
                delay = 10
                print("Retry in {} seconds.".format(delay))
            except Exception as e:
                print(e)
                sys.exit(1)

        time.sleep(delay)

ddns-ipv6-cloudflare.service

# /etc/systemd/system/ddns-ipv6-cloudflare.service
[Unit]
Description=Dynamic DNS for CloudFlare IPv6
After=network.target
Wants=network.target

[Service]
Type=simple
ExecStart=/usr/local/bin/ddns-ipv6-cloudflare.py
Restart=on-failure
# RestartPreventExitStatus=23

[Install]
WantedBy=multi-user.target