DDNS Update Scripts for Cloudflare

cloudflare_dns.py

#!/usr/bin/env python

import sys
import json
import requests


# Simple Cloudflare API
class Couldflare(object):

    def __init__(self, email, apiKey):
        self.email = email
        self.apiKey = apiKey

    def _request(self, method='GET', url='', **kwargs):
        h = {
            'Content-Type': 'application/json',
            'X-Auth-Email': self.email,
            'X-Auth-Key': self.apiKey
            }
        # default timeout: 5s
        s = requests.Session()
        s.trust_env = False
        r = s.request(method, url, headers=h, timeout=5, **kwargs)
        return r.json()

    def getZoneID(self, zoneName):
        return self._request(
            url='https://api.cloudflare.com/client/v4/zones?name=%s' % zoneName
            )['result'][0]['id']

    def getRecord(self, zoneID, recordName):
        return self._request(
            url='https://api.cloudflare.com/client/v4/zones/%s/dns_records?name=%s' % (zoneID, recordName)
            )

    def postRecord(self, zoneID, recordData):
        return self._request(
            method='POST',
            url='https://api.cloudflare.com/client/v4/zones/%s/dns_records' % zoneID,
            data=json.dumps(recordData)
            )

    def putRecode(self, zoneID, recordID, recordData):
        return self._request(
            method='PUT',
            url='https://api.cloudflare.com/client/v4/zones/%s/dns_records/%s' % (zoneID, recordID),
            data=json.dumps(recordData)
            )

    def updateRecordIP(self, recordName, ip):
        zoneName = '.'.join(recordName.split('.')[-2:])
        zoneID = self.getZoneID(zoneName)
        # Record Data
        recordData = {'type': 'A', 'name': recordName, 'content': ip, 'ttl': 120, 'proxied': False}
        # Records
        record = self.getRecord(zoneID, recordName)
        if not record['result']:
            # create record
            return self.postRecord(zoneID, recordData)

        firstResult = record['result'][0]
        if firstResult['content'] == ip:
            return record
        # update record
        recordID = record[0]['id']
        return self.putRecode(zoneID, recordID, recordData)


def getIP():
    def curl(url):
        s = requests.Session()
        s.trust_env = False
        r = s.get(
            url, timeout=5,
            headers={'User-Agent': 'curl', 'Accept': '*/*'}
            )
        r.raise_for_status()
        return r.text

    providers = (
        'http://ip.sb',
        'http://api.ipify.org',
        'http://ifconfig.io/ip'
        )
    for provider in providers:
        try:
            ip = curl(provider)
        except:
            continue
        else:
            return ip.strip()


def main():
    # params
    if len(sys.argv) == 4:
        ip = getIP()
        email, apiKey, recordName = sys.argv[1:]
    elif len(sys.argv) == 5:
        email, apiKey, recordName, ip = sys.argv[1:]
    else:
        sys.exit('Usage: python %s username password hostname [ip]' % __file__)

    cf = Couldflare(email, apiKey)
    try:
        result = cf.updateRecordIP(recordName, ip)
    except Exception as e:
        result = {'success': False, 'messages': e}

    if not result['success']:
        print('Record: update failed: %s' % result['messages'])
    else:
        print('Record: %s => %s updated!' % (recordName, ip))


if __name__ == '__main__':
    main()

ddns-start.sh

#!/bin/sh

DEBUG=0

EMAIL="example@mail.com"
API_KEY="e4808ed2d91e85c0393e737fa7649cd429f41"
ZONE_NAME="example.com"
RECORD_NAME="sub.example.com"

_curl() {
    curl -fs -m5 \
    -H Content-Type:application/json \
    -H X-Auth-Email:"$EMAIL" \
    -H X-Auth-Key:"$API_KEY" "$@"
}

_getip() {
    curl -fs -m5 --noproxy "*" "$@"
}

_log() {
    [ $DEBUG -ne 0 ] && echo $@
}

ZONE_ID=$(_curl "https://api.cloudflare.com/client/v4/zones?name=$ZONE_NAME" | jq -r '{"result"}[] | .[0] | .id')
_log "ZONE_ID=$ZONE_ID"

RECORD_ID=$(_curl "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/dns_records?name=$RECORD_NAME" | jq -r '{"result"}[] | .[0] | .id')
_log "RECORD_ID=$RECORD_ID"

IP=$(_getip ip.sb || _getip api.ipify.org || echo "$1")
_log "CURRENT_IP=$IP"

_curl -o /dev/null -X PUT "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/dns_records/$RECORD_ID" \
--data "{\"type\":\"A\",\"name\":\"$RECORD_NAME\",\"content\":\"$IP\",\"ttl\":120,\"proxied\":false}"

if [ $? -eq 0 ];then
    _log "RECORD: $RECORD_NAME => $IP"
    /sbin/ddns_custom_updated 1 && exit 0
else
    _log "RECORD UPDATE FAILED"
    /sbin/ddns_custom_updated 0 && exit 1
fi

ddns.py

#!/usr/bin/env python3

import re
import os
import sys
import logging
import requests
import ipaddress
import CloudFlare

# basic info
CF_DNS_TTL = 120
CF_DNS_NAME = "sub"
CF_ZONE_NAME = "example.com"
# token
CF_EMAIL = "example@mail.com"
CF_API_KEY = "e4808ed2d91e85c0393e737fa7649cd429f41"


def init():
    # logging config
    logging.basicConfig(
        #level=logging.INFO,
        level=logging.DEBUG,
        format="%(asctime)s:%(levelname)s:%(message)s",
        handlers=[
            logging.FileHandler(f"{os.path.splitext(__file__)[0]}.log"),
            logging.StreamHandler(sys.stdout)
        ])


init()


def curl(url: str, timeout: int = None, allow_redirects: bool = False) -> str:
    # process URL
    if not re.match("(http|https)://", url):
        url = "http://" + url
    # curl headers
    headers = {
        "User-Agent": "curl",
        "Accept": "*/*"
    }
    # requests session
    s = requests.Session()
    s.trust_env = False
    # requests
    try:
        r = s.get(url, headers=headers, timeout=timeout, allow_redirects=allow_redirects)
        r.raise_for_status()
        return r.text
    except:
        return ""


def get_ip():
    urls = ('ip.sb', 'ipv4.ip.sb', 'api.ipify.org', 'ip-api.com/line/?fields=query')
    for url in urls:
        ip = curl(url, 5).strip()
        if ip and ip.count('.') == 3:
            logging.debug(f"GET IP: {ip}")
            break
    return ip


def update_dns_record(ip):
    try:
        ipaddress.IPv4Address(ip)
    except ValueError as e:
        sys.exit(f'IP invalid! ({e})')

    cf = CloudFlare.CloudFlare(email=CF_EMAIL, token=CF_API_KEY)

    # query for the zone name and expect only one value back
    try:
        zones = cf.zones.get(params={'name': CF_ZONE_NAME, 'per_page': 1})
    except CloudFlare.exceptions.CloudFlareAPIError as e:
        sys.exit(f'/zones.get {int(e)} {e} - api call failed')
    except Exception as e:
        sys.exit(f'/zones.get - {e} - api call failed')

    if len(zones) == 0:
        sys.exit('No zones found')

    # extract the zone_id which is needed to process that zone
    zone = zones[0]
    zone_id = zone.get('id', None)
    if not zone_id:
        sys.exit("No zone id found")
    logging.debug(f"Zone ID: {zone_id}")

    # request the DNS records from that zone
    try:
        dns_records = cf.zones.dns_records.get(zone_id)
    except CloudFlare.exceptions.CloudFlareAPIError as e:
        sys.exit(f'/zones/dns_records.get {int(e)} {e} - api call failed')

    dns_record_id = None
    dns_record_content = None
    dns_name = '.'.join([CF_DNS_NAME, CF_ZONE_NAME]).lower()
    for dns_record in dns_records:
        r_name = dns_record['name']
        if r_name.lower() != dns_name:
            continue
        dns_record_id = dns_record['id']
        dns_record_content = dns_record['content']
        break

    # generate dns record data
    dns_record = {
        'name': CF_DNS_NAME,
        'type': 'A',
        'content': ip,
        'ttl': CF_DNS_TTL
    }

    # edit dns record
    if not dns_record_id:
        # create dns record
        logging.debug(f"No DNS record {dns_name} found")
        try:
            logging.debug("Creating DNS record")
            cf.zones.dns_records.post(zone_id, data=dns_record)
        except CloudFlare.exceptions.CloudFlareAPIError as e:
            sys.exit(f'/zones/dns_records.post {int(e)} {e} - api call failed')
        logging.debug(f"DNS record {dns_name} -> {ip} created")
    else:
        # update dns record
        logging.debug(f"Old DNS record {dns_name} -> {dns_record_content} found")
        # compare dns_record_content and ip
        if dns_record_content == ip:
            logging.debug("New IP equal Old IP, exit...")
            return

        try:
            logging.debug("Updating DNS record")
            cf.zones.dns_records.put(zone_id, dns_record_id, data=dns_record)
        except CloudFlare.exceptions.CloudFlareAPIError as e:
            sys.exit(f'/zones/dns_records.put {int(e)} {e} - api call failed')
        logging.info(f"DNS record {dns_name} -> {ip} updated")


def main():
    try:
        ip = get_ip()
        update_dns_record(ip)
    except SystemExit as e:
        logging.error(f"{e}")
    except KeyboardInterrupt:
        sys.exit(0)


if __name__ == '__main__':
    main()