nikrozman/cloudflare-dyndns
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve reliability

Browse Source
This commit is contained in:
Nik Rozman
2026-03-27 19:23:58 +01:00
parent 8ab85c002e
commit 5408add86b
1 changed files with 91 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
cloudflare_dyndns.py
View File

@@ -1,7 +1,7 @@
"""
Simple Cloudflare dynamic DNS updater.
- Pulls current IPv4 from ipinfo.io (or a fallback service).
- Pulls current IPv4 from multiple sources and cross-validates.
- Every hour, checks for changes and updates any A records that had the old IP.
Config (config.ini by default):
@@ -15,14 +15,12 @@ Config (config.ini by default):
Environment overrides (take precedence over config file):
CLOUDFLARE_API_TOKEN, CLOUDFLARE_ZONE_ID, IPINFO_URL, POLL_SECONDS, CONFIG_PATH, LAST_IP_OVERRIDE
Install dependency:
pip install requests
"""
from __future__ import annotations
import configparser
import os
import re
import sys
import time
from typing import List, Optional
@@ -30,8 +28,23 @@ from typing import List, Optional
import requests
IPINFO_DEFAULT = "https://ipinfo.io/ip"
FALLBACK_URL = "https://api.ipify.org"
IP_SOURCES = [
"https://ipinfo.io/ip",
"https://api.ipify.org",
"https://checkip.amazonaws.com",
"https://icanhazip.com",
]
DEFAULT_POLL_SECONDS = 3600
# Minimum agreeing sources needed to trust an IP result.
MIN_AGREEING_SOURCES = 2
_IPV4_RE = re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
def is_valid_ipv4(ip: str) -> bool:
if not _IPV4_RE.match(ip):
return False
return all(0 <= int(part) <= 255 for part in ip.split("."))
def load_config(path: str) -> dict:
@@ -68,7 +81,40 @@ def load_config(path: str) -> dict:
def get_public_ip(session: requests.Session, url: str) -> str:
resp = session.get(url, timeout=10)
resp.raise_for_status()
return resp.text.strip()
ip = resp.text.strip()
if not is_valid_ipv4(ip):
raise ValueError(f"Invalid IPv4 from {url!r}: {ip!r}")
return ip
def get_reliable_ip(session: requests.Session, primary_url: str) -> Optional[str]:
"""
Query multiple IP sources and return the IP that at least MIN_AGREEING_SOURCES
agree on. Returns None if no consensus can be reached.
"""
sources = [primary_url] + [s for s in IP_SOURCES if s != primary_url]
votes: dict[str, int] = {}
for url in sources:
try:
ip = get_public_ip(session, url)
votes[ip] = votes.get(ip, 0) + 1
if votes[ip] >= MIN_AGREEING_SOURCES:
return ip
except Exception as exc:
print(f" IP source {url!r} failed: {exc}", file=sys.stderr)
if votes:
# Fall back to majority if we couldn't reach the threshold.
best_ip = max(votes, key=lambda k: votes[k])
print(
f" Warning: could not get {MIN_AGREEING_SOURCES} agreeing sources "
f"(got {votes}); using {best_ip!r} with {votes[best_ip]} vote(s).",
file=sys.stderr,
)
return best_ip
return None
def fetch_records_with_ip(
@@ -77,7 +123,6 @@ def fetch_records_with_ip(
zone_id: str,
ip: str,
) -> List[dict]:
# Fetch all A records matching the old IP so we can update them.
records: List[dict] = []
page = 1
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
@@ -134,44 +179,59 @@ def main() -> None:
sys.exit(1)
ipinfo_url = os.getenv("IPINFO_URL") or config.get("ipinfo_url") or IPINFO_DEFAULT
poll_seconds = os.getenv("POLL_SECONDS")
poll_seconds = int(poll_seconds) if poll_seconds else config.get("poll_seconds", DEFAULT_POLL_SECONDS)
poll_seconds_env = os.getenv("POLL_SECONDS")
if poll_seconds_env:
poll_seconds = int(poll_seconds_env)
else:
poll_seconds = config.get("poll_seconds") or DEFAULT_POLL_SECONDS
last_ip_override = os.getenv("LAST_IP_OVERRIDE") or config.get("last_ip_override")
session = requests.Session()
last_ip: Optional[str] = last_ip_override
print(f"Starting Cloudflare DDNS watcher (interval={poll_seconds}s)")
print(f"Using IP endpoint: {ipinfo_url}")
print(f"Primary IP endpoint: {ipinfo_url} (with {len(IP_SOURCES) - 1} fallbacks)")
if last_ip_override:
print(f"Starting with overridden last_ip: {last_ip_override}")
try:
while True:
try:
current_ip = get_public_ip(session, ipinfo_url)
except Exception:
# Fallback if primary service fails.
current_ip = get_public_ip(session, FALLBACK_URL)
current_ip = get_reliable_ip(session, ipinfo_url)
if current_ip is None:
print("Could not determine public IP from any source; skipping this cycle.", file=sys.stderr)
time.sleep(poll_seconds)
continue
if last_ip is None:
last_ip = current_ip
print(f"Initial IP: {current_ip}")
elif current_ip != last_ip:
print(f"IP changed: {last_ip} -> {current_ip}")
try:
records = fetch_records_with_ip(session, token, zone_id, last_ip)
if not records:
print("No records matched old IP; nothing to update.")
else:
for rec in records:
update_record_ip(session, token, zone_id, rec, current_ip)
print(f"Updated {rec['name']} to {current_ip}")
if last_ip is None:
last_ip = current_ip
print(f"Initial IP: {current_ip}")
elif current_ip != last_ip:
print(f"IP changed: {last_ip} -> {current_ip}")
try:
records = fetch_records_with_ip(session, token, zone_id, last_ip)
if not records:
print(
f"No A records matched old IP {last_ip!r}; "
"they may have already been updated externally."
)
else:
for rec in records:
update_record_ip(session, token, zone_id, rec, current_ip)
print(f"Updated {rec['name']} to {current_ip}")
# Advance last_ip regardless — old IP is gone, no point
# searching for it again next cycle.
last_ip = current_ip
except Exception as api_err:
print(f"Cloudflare update failed: {api_err}", file=sys.stderr)
else:
print(f"No change. Current IP: {current_ip}")
except Exception as api_err:
print(f"Cloudflare update failed: {api_err}", file=sys.stderr)
# Do NOT advance last_ip; retry on next cycle.
else:
print(f"No change. Current IP: {current_ip}")
except Exception as exc:
# Catch-all so a transient error never kills the process.
print(f"Unexpected error in poll cycle: {exc}", file=sys.stderr)
time.sleep(poll_seconds)
except KeyboardInterrupt: