#!/usr/bin/env python3 """ ddns.py -- Update DDNS provider(s) with current public IP. Reads ddns.json, fetches the current public IP, and updates each enabled provider block only if the IP has changed since the last successful update for that provider. Designed to be run on a systemd timer. IP check services are rotated each run using .ddns-last-service so no single provider is spammed. If the selected service fails, the script falls back through the remaining services in order. Per-provider cache files are named .ddns-last-ip-. Logs to ddns.log in the same directory as this script. Log is cleared when it exceeds general.log_max_kb from config. Usage: sudo python3 ddns.py --start Run update and install systemd timer sudo python3 ddns.py --disable Stop updates and remove systemd timer python3 ddns.py --apply Run update once (used by timer) python3 ddns.py --force Force update regardless of cached IP python3 ddns.py --status Show timer/service status python3 ddns.py --getip Print current public IP and exit """ import json import os import subprocess import re import urllib.request import urllib.error import sys import logging from pathlib import Path SCRIPT_DIR = Path(__file__).parent CONFIG_FILE = SCRIPT_DIR / "ddns.json" CACHE_SERVICE_FILE = SCRIPT_DIR / ".ddns-last-service" LOG_FILE = SCRIPT_DIR / "ddns.log" TIMER_NAME = "ddns-update" SERVICE_FILE = Path(f"/etc/systemd/system/{TIMER_NAME}.service") TIMER_FILE = Path(f"/etc/systemd/system/{TIMER_NAME}.timer") # log is assigned in setup_logging() after config is loaded log = None # =================================================================== # Load config # =================================================================== def load_config(): if not CONFIG_FILE.exists(): print(f"ERROR: Config file not found: {CONFIG_FILE}", file=sys.stderr) sys.exit(1) with open(CONFIG_FILE) as f: data = json.load(f) # Validate general block required_general = {"log_max_kb", "log_errors_only", "ip_check_services"} missing = required_general - set(data.get("general", {}).keys()) if missing: print(f"ERROR: Missing keys in general block: {missing}", file=sys.stderr) sys.exit(1) if not data["general"]["ip_check_services"]: print("ERROR: ip_check_services list is empty.", file=sys.stderr) sys.exit(1) # Validate providers block if not data.get("providers"): print("ERROR: No providers defined in config.", file=sys.stderr) sys.exit(1) for p in data["providers"]: base_required = {"description", "provider", "enabled"} missing = base_required - set(p.keys()) if missing: print(f"ERROR: Provider '{p.get('description', '?')}' missing keys: {missing}", file=sys.stderr) sys.exit(1) ptype = p.get("provider", "").lower() if ptype == "noip": extra = {"username", "password", "hostnames"} elif ptype == "duckdns": extra = {"api_token", "hostnames"} elif ptype == "cloudflare": extra = {"api_token", "hostnames"} else: print(f"ERROR: Provider '{p.get('description', '?')}' has unknown provider type: '{ptype}'", file=sys.stderr) sys.exit(1) missing = extra - set(p.keys()) if missing: print(f"ERROR: Provider '{p.get('description', '?')}' missing keys for {ptype}: {missing}", file=sys.stderr) sys.exit(1) return data # =================================================================== # Helpers # =================================================================== def chown_to_script_dir_owner(path): """Chown a file to the owner of the script directory. This works correctly whether invoked via sudo, directly as root (e.g. systemd timer), or as a normal user - the script directory owner is always the right target. """ try: stat = SCRIPT_DIR.stat() os.chown(path, stat.st_uid, stat.st_gid) except OSError: pass # non-fatal # =================================================================== # Logging # =================================================================== def setup_logging(max_kb, errors_only): """Clear log if oversized, then initialise logger. Must be called before log is used.""" global log max_bytes = int(max_kb * 1024) try: if LOG_FILE.exists() and LOG_FILE.stat().st_size > max_bytes: LOG_FILE.write_text("") if not LOG_FILE.exists(): LOG_FILE.touch() chown_to_script_dir_owner(LOG_FILE) file_handler = logging.FileHandler(LOG_FILE) except PermissionError: print(f"WARNING: Cannot write to {LOG_FILE} (permission denied). " f"Run with sudo or fix ownership: sudo chown $USER {LOG_FILE}") file_handler = None level = logging.ERROR if errors_only else logging.INFO handlers = [logging.StreamHandler(sys.stdout)] if file_handler: handlers.insert(0, file_handler) logging.basicConfig( level=level, format="%(asctime)s %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S", handlers=handlers, ) log = logging.getLogger("ddns") # =================================================================== # Per-provider IP cache # =================================================================== def cache_file_for(description): """Return the cache file path for a given provider description.""" safe_name = description.replace(" ", "-") return SCRIPT_DIR / f".ddns-last-ip-{safe_name}" def get_cached_ip(description): f = cache_file_for(description) if f.exists(): return f.read_text().strip() return None def save_cached_ip(description, ip): f = cache_file_for(description) f.write_text(ip) chown_to_script_dir_owner(f) # =================================================================== # Service rotation # =================================================================== def get_next_service_index(total): """Read last used index, increment, wrap around, return next index.""" if CACHE_SERVICE_FILE.exists(): try: last = int(CACHE_SERVICE_FILE.read_text().strip()) except ValueError: last = -1 else: last = -1 return (last + 1) % total def save_service_index(index): CACHE_SERVICE_FILE.write_text(str(index)) chown_to_script_dir_owner(CACHE_SERVICE_FILE) # =================================================================== # Public IP detection # =================================================================== def extract_ip(body): """ Extract an IP address from a service response. Handles plain text, key=value format (e.g. Cloudflare /cdn-cgi/trace where the ip= line is the caller's IP while h= is the server's IP), and HTML. """ # Check for key=value format first (e.g. /cdn-cgi/trace) for line in body.splitlines(): if line.startswith("ip="): candidate = line[3:].strip() if re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', candidate): return candidate # Try plain text (strip and validate) plain = body.strip() if re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', plain): return plain # Fall back to extracting from HTML match = re.search(r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})', body) if match: return match.group(1) return None def _get_ip_via_cf_dns(spec): """Query Cloudflare's myip.cloudflare via DNS TXT (chaos class) for the caller's IP. spec format: 'cf-dns:' e.g. 'cf-dns:myip.cloudflare' Requires the 'dig' utility to be installed. """ hostname = spec[len("cf-dns:"):] cmd = ["dig", "+short", "@1.1.1.1", "chaos", "txt", hostname] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if result.returncode != 0: return None # TXT records come back quoted: "203.0.113.50" ip = result.stdout.strip().strip('"').split()[0] if re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', ip): return ip except FileNotFoundError: log.warning("'dig' command not found; cannot use cf-dns IP check service.") except Exception: pass return None # =================================================================== def get_public_ip(services): """ Start at the next service in rotation. If it fails, fall through the remaining services in order. Saves the index of the service that succeeded so the next run starts with the following one. """ total = len(services) start = get_next_service_index(total) ordered = [services[(start + i) % total] for i in range(total)] for i, service in enumerate(ordered): try: if service.startswith("cf-dns:"): ip = _get_ip_via_cf_dns(service) else: req = urllib.request.Request(service, headers={"User-Agent": "ddns-update/1.0"}) with urllib.request.urlopen(req, timeout=10) as r: body = r.read().decode().strip() ip = extract_ip(body) if ip: used_index = (start + i) % total save_service_index(used_index) log.info(f"Public IP retrieved from {service}: {ip}") return ip except Exception as e: log.warning(f"IP check failed for {service}: {e}") continue log.error("Could not determine public IP from any configured service.") sys.exit(1) # =================================================================== # No-IP update # =================================================================== def update_noip(provider, ip): """ No-IP HTTP update API. Docs: https://www.noip.com/integrate/request Uses HTTP Basic Auth. Supports comma-separated list of hostnames. """ username = provider["username"] password = provider["password"] hostnames = ",".join(provider["hostnames"]) url = f"https://dynupdate.no-ip.com/nic/update?hostname={hostnames}&myip={ip}" password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm() password_mgr.add_password(None, url, username, password) handler = urllib.request.HTTPBasicAuthHandler(password_mgr) opener = urllib.request.build_opener(handler) req = urllib.request.Request(url, headers={"User-Agent": "ddns-update/1.0"}) try: with opener.open(req, timeout=10) as r: return r.read().decode().strip() except urllib.error.URLError as e: log.error(f"Network error contacting No-IP: {e}") return None def interpret_noip_response(response, hostnames, ip): """ No-IP response codes: good -- update successful nochg -- IP already set to this value (no change needed) nohost -- hostname not found in account badauth -- invalid credentials badagent -- client blocked !donator -- feature requires paid account abuse -- account blocked for abuse 911 -- server-side error, retry later """ if response is None: return False if response.startswith("good"): log.info(f"No-IP updated successfully: {hostnames} -> {ip}") return True elif response.startswith("nochg"): log.info(f"No-IP: no change needed ({hostnames} already set to {ip})") return True elif response == "nohost": log.error(f"No-IP: hostname '{hostnames}' not found in account.") elif response == "badauth": log.error(f"No-IP: authentication failed for '{hostnames}'. Check username and password.") elif response == "badagent": log.error("No-IP: client blocked by No-IP.") elif response == "!donator": log.error("No-IP: this feature requires a paid account.") elif response == "abuse": log.error("No-IP: account blocked for abuse.") elif response == "911": log.error("No-IP: server error. Will retry on next run.") else: log.error(f"No-IP: unexpected response: {response}") return False # =================================================================== # DuckDNS update # =================================================================== def update_duckdns(provider, ip): """ DuckDNS HTTP update API. Docs: https://www.duckdns.org/spec.jsp Token-based, no username/password. Subdomains are the short name only (e.g. "myhome", not "myhome.duckdns.org"). Supports multiple subdomains as a comma-separated list. Returns True on success, False on failure. """ token = provider["api_token"] subdomains = ",".join(h.replace(".duckdns.org", "") for h in provider["hostnames"]) description = provider["description"] url = f"https://www.duckdns.org/update?domains={subdomains}&token={token}&ip={ip}" try: req = urllib.request.Request(url, headers={"User-Agent": "ddns-update/1.0"}) with urllib.request.urlopen(req, timeout=10) as r: response = r.read().decode().strip() if response == "OK": log.info(f"DuckDNS updated successfully: {subdomains} -> {ip}") return True else: log.error(f"DuckDNS update failed for '{description}': response was '{response}'") return False except urllib.error.URLError as e: log.error(f"Network error contacting DuckDNS: {e}") return False # =================================================================== # Cloudflare DNS update # =================================================================== def _cf_api_get(url, headers): req = urllib.request.Request(url, headers=headers) try: with urllib.request.urlopen(req, timeout=10) as r: return json.loads(r.read().decode()) except Exception as e: log.error(f"Cloudflare API GET error ({url}): {e}") return None def _cf_get_zone_id(zone_name, headers): data = _cf_api_get( f"https://api.cloudflare.com/client/v4/zones?name={zone_name}", headers ) if data and data.get("success") and data["result"]: return data["result"][0]["id"] return None def _cf_get_record_id(zone_id, hostname, headers): data = _cf_api_get( f"https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records?name={hostname}&type=A", headers, ) if data and data.get("success") and data["result"]: return data["result"][0]["id"] return None def update_cloudflare(provider, ip): """ Cloudflare DNS update API. Docs: https://developers.cloudflare.com/api/resources/dns/subresources/records/methods/edit/ Bearer-token auth. Looks up zone and record IDs dynamically, then PATCHes each A record. """ token = provider["api_token"] headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "User-Agent": "ddns-update/1.0", } success = True for hostname in provider["hostnames"]: zone_name = ".".join(hostname.split(".")[-2:]) zone_id = _cf_get_zone_id(zone_name, headers) if not zone_id: log.error(f"Cloudflare: zone '{zone_name}' not found in account.") success = False continue record_id = _cf_get_record_id(zone_id, hostname, headers) if not record_id: log.error(f"Cloudflare: A record for '{hostname}' not found in zone '{zone_name}'.") success = False continue url = f"https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records/{record_id}" payload = json.dumps({"content": ip}).encode() req = urllib.request.Request(url, data=payload, headers=headers, method="PATCH") try: with urllib.request.urlopen(req, timeout=10) as r: data = json.loads(r.read().decode()) if data.get("success"): log.info(f"Cloudflare updated successfully: {hostname} -> {ip}") else: log.error(f"Cloudflare update failed for '{hostname}': {data.get('errors')}") success = False except Exception as e: log.error(f"Cloudflare API PATCH error for '{hostname}': {e}") success = False return success # =================================================================== # Process a single provider block # =================================================================== def process_provider(provider, current_ip, force=False): description = provider["description"] if not provider.get("enabled") is True: log.info(f"Provider '{description}' is disabled, skipping.") return cached_ip = get_cached_ip(description) if not force and current_ip == cached_ip: log.info(f"[{description}] IP unchanged ({current_ip}), skipping update.") return if force: log.info(f"[{description}] Force update requested. Updating with {current_ip}...") elif cached_ip: log.info(f"[{description}] IP changed: {cached_ip} -> {current_ip}. Updating...") else: log.info(f"[{description}] No cached IP found. Updating with {current_ip}...") ptype = provider["provider"].lower() if ptype == "noip": hostnames = ",".join(provider["hostnames"]) response = update_noip(provider, current_ip) success = interpret_noip_response(response, hostnames, current_ip) elif ptype == "duckdns": success = update_duckdns(provider, current_ip) elif ptype == "cloudflare": success = update_cloudflare(provider, current_ip) else: log.error(f"[{description}] Unknown provider type: '{ptype}'") return if success: save_cached_ip(description, current_ip) # =================================================================== # Timer management # =================================================================== def parse_interval(interval_str): """ Convert interval string (e.g. 5m, 2h, 1d) to systemd OnUnitActiveSec value. Supported units: m (minutes), h (hours), d (days). """ interval_str = interval_str.strip() if interval_str.endswith("m"): return f"{interval_str[:-1]}min" elif interval_str.endswith("h"): return f"{interval_str[:-1]}h" elif interval_str.endswith("d"): return f"{interval_str[:-1]}day" else: print(f"ERROR: Invalid timer_interval format: '{interval_str}'. Use e.g. 5m, 2h, 1d.") sys.exit(1) def get_current_timer_interval(): """Read the current OnUnitActiveSec value from the timer file, or None if not present.""" if not TIMER_FILE.exists(): return None for line in TIMER_FILE.read_text().splitlines(): if line.strip().startswith("OnUnitActiveSec="): return line.strip().split("=", 1)[1] return None def install_timer(cfg): interval = cfg["general"].get("timer_interval", "5m") systemd_unit = parse_interval(interval) script_path = Path(__file__).resolve() current_interval = get_current_timer_interval() if current_interval == systemd_unit: log.info(f"Timer already set to {interval}, no update needed.") return service_content = f"""[Unit] Description=DDNS update service After=network-online.target Wants=network-online.target [Service] Type=oneshot ExecStart=/usr/bin/python3 {script_path} --apply """ timer_content = f"""[Unit] Description=DDNS update timer [Timer] OnUnitActiveSec={systemd_unit} OnBootSec=1min [Install] WantedBy=timers.target """ SERVICE_FILE.write_text(service_content) TIMER_FILE.write_text(timer_content) subprocess.run(["systemctl", "daemon-reload"], check=True) subprocess.run(["systemctl", "enable", "--now", f"{TIMER_NAME}.timer"], check=True) if current_interval is None: log.info(f"Timer installed: runs every {interval}.") else: log.info(f"Timer updated: was {current_interval}, now runs every {interval}.") def remove_timer(): if TIMER_FILE.exists() or SERVICE_FILE.exists(): subprocess.run( ["systemctl", "disable", "--now", f"{TIMER_NAME}.timer"], capture_output=True ) for f in (TIMER_FILE, SERVICE_FILE): if f.exists(): f.unlink() print(f"Removed: {f}") subprocess.run(["systemctl", "daemon-reload"], capture_output=True) print("Timer removed.") else: print("No timer found, nothing to remove.") # =================================================================== # Main # =================================================================== def run_update(cfg, force=False, getip_only=False): """Perform a single DDNS update pass. Called by both timer and --start. If force=True, bypasses the cached IP check and always updates. If getip_only=True, prints the detected public IP and returns without updating providers.""" general = cfg["general"] current_ip = get_public_ip(general["ip_check_services"]) if getip_only: print(current_ip) return enabled = [p for p in cfg["providers"] if p.get("enabled") is True] if not enabled: log.error("No enabled providers found in config.") sys.exit(1) for provider in enabled: process_provider(provider, current_ip, force=force) def show_status(): """Show status of managed timer.""" result = subprocess.run( ["systemctl", "status", f"{TIMER_NAME}.timer", "--no-pager"], capture_output=True, text=True ) print(result.stdout) def main(): import argparse parser = argparse.ArgumentParser( description="Update DDNS provider(s) with current public IP", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=( "examples:\n" " sudo python3 ddns.py --start Run update and install systemd timer\n" " sudo python3 ddns.py --disable Stop updates and remove systemd timer\n" " python3 ddns.py --apply Run update once (used by timer)\n" " python3 ddns.py --force Force update regardless of cached IP\n" " python3 ddns.py --status Show timer/service status\n" " python3 ddns.py --getip Print current public IP and exit\n" ) ) parser.add_argument("--start", action="store_true", help="Run update and install systemd timer") parser.add_argument("--disable", action="store_true", help="Stop updates and remove systemd timer") parser.add_argument("--apply", action="store_true", help="Run update once (used by timer)") parser.add_argument("--force", action="store_true", help="Force update regardless of cached IP") parser.add_argument("--status", action="store_true", help="Show timer/service status") parser.add_argument("--getip", action="store_true", help="Print current public IP and exit") args = parser.parse_args() if not any([args.start, args.disable, args.apply, args.force, args.status, args.getip]): parser.print_help() return if args.status: show_status() return if args.getip: global log log = logging.getLogger("ddns_quiet") log.addHandler(logging.NullHandler()) log.propagate = False cfg = load_config() run_update(cfg, getip_only=True) return cfg = load_config() general = cfg["general"] setup_logging(general["log_max_kb"], general["log_errors_only"]) if args.disable: remove_timer() return if args.start: run_update(cfg) install_timer(cfg) return if args.apply or args.force: run_update(cfg, force=args.force) if __name__ == "__main__": main()