python-cloudflare

ddns.py

import time
import socket

import CloudFlare
import CloudFlare.exceptions
import requests

EMAIL = '<email>'
TOKEN = '<cf token>'

ZONE_ID = '<domain id>'
RECORD_NAME = 'example.com'
RECORD_ID = '<record id>'


def get_real_ipv4():
    return requests.get('http://ip.sb',headers={'user-agent':'curl'}).text.strip()


def get_real_ipv6():
    s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
    try:
        s.connect(('2001:da8::666', 1))
        ip = s.getsockname()[0]
    finally:
        s.close()
    return ip


def main():
    cf = CloudFlare.CloudFlare(email=EMAIL, token=TOKEN)

    dns_records = cf.zones.dns_records.get(ZONE_ID, params={
        'name': RECORD_NAME,
        'match': 'all'
    })

    remote_ipv4_record = next(x for x in dns_records if x['type'] == 'A')
    real_ipv4 = get_real_ipv4()
    if real_ipv4 != remote_ipv4_record['content']:
        print(time.strftime('%Y-%m-%d %H:%M:%S'),
              remote_ipv4_record['content'], '->', real_ipv4)
        cf.zones.dns_records.put(ZONE_ID, remote_ipv4_record['id'], data={
            'name': RECORD_NAME,
            'type': 'A',
            'content': real_ipv4
        })
    else:
        print('IPv4 did not change')

    # remote_ipv6_record = next(x for x in dns_records if x['type'] == 'AAAA')
    # real_ipv6 = get_real_ipv6()
    # if real_ipv6 != remote_ipv6_record['content']:
    #     print(time.strftime('%Y-%m-%d %H:%M:%S'),
    #           remote_ipv6_record['content'], '->', real_ipv6)
    #     cf.zones.dns_records.put(ZONE_ID, remote_ipv6_record['id'], data={
    #         'name': RECORD_NAME,
    #         'type': 'AAAA',
    #         'content': real_ipv6
    #     })
    # else:
    #     print('IPv6 did not change')


if __name__ == "__main__":
    try:
        main()
    except CloudFlare.exceptions.CloudFlareAPIError as e:
        print('Failed', e)

get_zones.py

import CloudFlare


def main():
    cf = CloudFlare.CloudFlare()
    zones = cf.zones.get()
    for zone in zones:
        zone_id = zone["id"]
        zone_name = zone["name"]
        print(zone_id, zone_name)


if __name__ == "__main__":
    main()

update_dynamic_dns.py

import requests

import CloudFlare


def my_ip_address():
    # This list is adjustable - plus some v6 enabled services are needed
    # url = 'http://myip.dnsomatic.com'
    # url = 'http://www.trackip.net/ip'
    # url = 'http://myexternalip.com/raw'
    url = "https://api.ipify.org"
    try:
        ip_address = requests.get(url).text
    except:
        exit("%s: failed" % (url))
    if ip_address == "":
        exit("%s: failed" % (url))

    if ":" in ip_address:
        ip_address_type = "AAAA"
    else:
        ip_address_type = "A"

    return ip_address, ip_address_type


def do_dns_update(cf, zone_name, zone_id, dns_name, ip_address, ip_address_type):
    try:
        params = {"name": dns_name, "match": "all", "type": ip_address_type}
        dns_records = cf.zones.dns_records.get(zone_id, params=params)
    except CloudFlare.exceptions.CloudFlareAPIError as e:
        exit("/zones/dns_records %s - %d %s - api call failed" % (dns_name, e, e))

    updated = False

    # update the record - unless it's already correct
    for dns_record in dns_records:
        old_ip_address = dns_record["content"]
        old_ip_address_type = dns_record["type"]

        if ip_address_type not in ["A", "AAAA"]:
            # we only deal with A / AAAA records
            continue

        if ip_address_type != old_ip_address_type:
            # only update the correct address type (A or AAAA)
            # we don't see this becuase of the search params above
            print("IGNORED: %s %s ; wrong address family" % (dns_name, old_ip_address))
            continue

        if ip_address == old_ip_address:
            print("UNCHANGED: %s %s" % (dns_name, ip_address))
            updated = True
            continue

        proxied_state = dns_record["proxied"]

        # Yes, we need to update this record - we know it's the same address type

        dns_record_id = dns_record["id"]
        dns_record = {
            "name": dns_name,
            "type": ip_address_type,
            "content": ip_address,
            "proxied": proxied_state,
        }
        try:
            dns_record = cf.zones.dns_records.put(
                zone_id, dns_record_id, data=dns_record
            )
        except CloudFlare.exceptions.CloudFlareAPIError as e:
            exit(
                "/zones.dns_records.put %s - %d %s - api call failed" % (dns_name, e, e)
            )
        print("UPDATED: %s %s -> %s" % (dns_name, old_ip_address, ip_address))
        updated = True

    if updated:
        return

    # no exsiting dns record to update - so create dns record
    dns_record = {"name": dns_name, "type": ip_address_type, "content": ip_address}
    try:
        dns_record = cf.zones.dns_records.post(zone_id, data=dns_record)
    except CloudFlare.exceptions.CloudFlareAPIError as e:
        exit("/zones.dns_records.post %s - %d %s - api call failed" % (dns_name, e, e))
    print("CREATED: %s %s" % (dns_name, ip_address))


def main():
    dns_name = "test.artin.li"
    host_name, zone_name = ".".join(dns_name.split(".")[:2]), ".".join(
        dns_name.split(".")[-2:]
    )

    ip_address, ip_address_type = my_ip_address()

    print("MY IP: %s %s" % (dns_name, ip_address))

    cf = CloudFlare.CloudFlare()

    # grab the zone identifier
    try:
        params = {"name": zone_name}
        zones = cf.zones.get(params=params)
    except CloudFlare.exceptions.CloudFlareAPIError as e:
        exit("/zones %d %s - api call failed" % (e, e))
    except Exception as e:
        exit("/zones.get - %s - api call failed" % (e))

    if len(zones) == 0:
        exit("/zones.get - %s - zone not found" % (zone_name))

    if len(zones) != 1:
        exit("/zones.get - %s - api call returned %d items" % (zone_name, len(zones)))

    zone = zones[0]

    zone_name = zone["name"]
    zone_id = zone["id"]

    do_dns_update(cf, zone_name, zone_id, dns_name, ip_address, ip_address_type)
    exit(0)


if __name__ == "__main__":
    main()