Skip to content
On this page

DDNS for cloudflare

Code

py
import time
import socket
import requests

import logging
from logging.handlers import RotatingFileHandler

config = {
    "api_token": "xxx",
    "api_key": "xxx",
    "account_email": "[email protected]",
    "zone_id": "xxx",
    "lan_ipv4": "192.168.1.123", # for vpn networking
    "domain_name": [
        {
            # direct will be: i-d.example.net
            # vpn will be: i-v.example.net
            "name": "i",
        },{
            "name": "file",
        },{
            "name": "video",
        }],
}
logFile = '/home/yunyuyuan/ddns/ddns.log'

log_formatter = logging.Formatter('%(asctime)s %(levelname)s %(funcName)s(%(lineno)d) %(message)s')


my_handler = RotatingFileHandler(logFile, mode='a', maxBytes=1024*1024,
                                 backupCount=2, encoding=None, delay=False)
my_handler.setFormatter(log_formatter)
my_handler.setLevel(logging.INFO)

app_log = logging.getLogger('root')
app_log.setLevel(logging.INFO)

app_log.addHandler(my_handler)

def get_ipv6_address():
    try:
        # Create a socket object
        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
        
        # Connect to a remote server (doesn't actually send data)
        s.connect(("v2ray-sw-d.yunyuyuan.net", 80))
        
        # Get the global IPv6 address from the connected socket
        ipv6_address = s.getsockname()[0]
        
        return ipv6_address
    except Exception as e:
        app_log.error(f"Error: {e}")
        return None

def get_ip(ip_type):
    if ip_type == 'LOCAL':
        return config['lan_ipv4']
    elif ip_type == 'A':
        return requests.get('https://4.ipw.cn').text
    return get_ipv6_address()

def cf_api(endpoint, method, headers={}, data=False):
    api_token = config['api_token']
    if api_token != '' and api_token != 'api_token_here':
        headers = {
            "Authorization": "Bearer " + api_token, **headers
        }
    else:
        headers = {
            "X-Auth-Email": config['account_email'],
            "X-Auth-Key": config['api_key'],
        }
    try:
        if(data == False):
            response = requests.request(method, "https://api.cloudflare.com/client/v4/" + endpoint, headers=headers)
        else:
            response = requests.request(
                method, "https://api.cloudflare.com/client/v4/" + endpoint,
                headers=headers, json=data)

        if response.ok:
            return response.json()
        else:
            app_log.error("Error sending '" + method +
                  "' request to '" + response.url + "':")
            return None
    except Exception as e:
        app_log.error("An exception occurred while sending '" +
              method + "' request to '" + endpoint + "': " + str(e))
        return None

def call_api(domain_name, c_domain, domain_type):
    record = {
        "type": 'A' if domain_type == 'LOCAL' else domain_type,
        "name": domain_name,
        "content": get_ip(domain_type),
        "proxied": False,
        "ttl": 1
    }
    dns_records = cf_api("zones/" + config['zone_id'] + f"/dns_records?per_page=100&type={domain_type}&name={domain_name}", "GET")
    response = {}
    if dns_records['result']:
        # updating
        app_log.info(f'updating {domain_name}')
        dns_record = dns_records['result'][0]
        if dns_record['content'] == record['content']:
            app_log.info(f'local ip: {record["content"]} same to server, no need to update.')
        else:
            response = cf_api("zones/" + config['zone_id'] + "/dns_records/" + dns_record['id'], "PUT", {}, record)
    else:
        # creating
        app_log.info(f'creating {domain_name}')
        response = cf_api("zones/" + config['zone_id'] + "/dns_records", "POST", {}, record)
    if response and response.get('success', None):
        app_log.info(f'succeeded set ip to: {record["content"]}.')

def ddns():
    response = cf_api("zones/" + config['zone_id'], "GET")
    if response is None or response["result"]["name"] is None:
        app_log.error(f'get domain info error with zones api')
        return
    zone_result_name = response["result"]["name"]
    for c_domain in config["domain_name"]:
        call_api(c_domain["name"] + '-d.' + zone_result_name, c_domain, 'AAAA')
        if config['lan_ipv4']:
            call_api(c_domain["name"] + '-v.' + zone_result_name, c_domain, 'LOCAL')

if __name__ == '__main__':
    try:
        ddns()
    except Exception as e:
        app_log.error(f'error: {str(e)}')

Make a systemd service

sh
sudo vim /etc/systemd/system/pyddns.service
ini
[Unit]
Description=CloudFlare DDNS
After=network-online.target syslog.target

[Service]
User=yunyuyuan
ExecStart=/usr/bin/python3 /home/yunyuyuan/ddns/ddns-cf.py
RestartSec=300
Restart=always

[Install]
WantedBy=multi-user.target

sh
sudo systemctl enable --now pyddns.service