Skip to content

Instantly share code, notes, and snippets.

@HBIDamian
Created February 13, 2025 12:14
Show Gist options
  • Select an option

  • Save HBIDamian/abb4603ba655fbf40875e11ba66cdfd5 to your computer and use it in GitHub Desktop.

Select an option

Save HBIDamian/abb4603ba655fbf40875e11ba66cdfd5 to your computer and use it in GitHub Desktop.
import re
import requests
import logging
# Configuration
auth_email = "" # The email used to login 'https://dash.cloudflare.com'
auth_method = "global" # Set to "global" for Global API Key or "token" for Scoped API Token
auth_key = "" # Your API Token or Global API Key
zone_identifier = "" # Can be found in the "Overview" tab of your domain
record_name = "" # Which record you want to be synced
ttl = 3600 # Set the DNS TTL (seconds)
proxy = True # Set the proxy to true or false
sitename = "" # Title of site "Example Site"
slackchannel = "" # Slack Channel #example
slackuri = "" # URI for Slack WebHook "https://hooks.slack.com/services/xxxxx"
discorduri = "" # URI for Discord WebHook "https://discordapp.com/api/webhooks/xxxxx"
debugMode = False # Set to True to enable console logging
# Setup logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG if debugMode else logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# File handler
# To same directory as script directory
file_handler = logging.FileHandler(filename='cloudflareDNS.log', mode='a', encoding='utf-8')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# Console handler for debug mode
if debugMode:
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
# IPv4 regex pattern
ipv4_regex = re.compile(r'([01]?[0-9]?[0-9]|2[0-4][0-9]|25[0-5])\.([01]?[0-9]?[0-9]|2[0-4][0-9]|25[0-5])\.([01]?[0-9]?[0-9]|2[0-4][0-9]|25[0-5])\.([01]?[0-9]?[0-9]|2[0-4][0-9]|25[0-5])')
# Function to get public IP
def get_public_ip():
try:
response = requests.get('https://cloudflare.com/cdn-cgi/trace')
response.raise_for_status()
ip = re.search(r'^ip=([\d.]+)$', response.text, re.MULTILINE)
if ip:
return ip.group(1)
except requests.RequestException:
pass
try:
response = requests.get('https://api.ipify.org')
response.raise_for_status()
return response.text
except requests.RequestException:
pass
try:
response = requests.get('https://ipv4.icanhazip.com')
response.raise_for_status()
return response.text.strip()
except requests.RequestException:
pass
return None
# Get the public IP
ip = get_public_ip()
if not ip or not ipv4_regex.match(ip):
logger.error("DDNS Updater: Failed to find a valid IP.")
exit(2)
# Set the proper auth header
auth_header = "X-Auth-Key" if auth_method == "global" else "Authorization: Bearer"
# Seek for the A record
logger.info("DDNS Updater: Check Initiated")
record_url = f"https://api.cloudflare.com/client/v4/zones/{zone_identifier}/dns_records?type=A&name={record_name}"
record_headers = {
"X-Auth-Email": auth_email,
auth_header: auth_key,
"Content-Type": "application/json"
}
record_response = requests.get(record_url, headers=record_headers)
record = record_response.json()
# Check if the domain has an A record
if record.get("result_info", {}).get("count") == 0:
logger.error(f"DDNS Updater: Record does not exist, perhaps create one first? ({ip} for {record_name})")
exit(1)
# Get existing IP
old_ip = next((r["content"] for r in record["result"] if r["type"] == "A"), None)
if ip == old_ip:
logger.info(f"DDNS Updater: IP ({ip}) for {record_name} has not changed.")
exit(0)
# Set the record identifier from result
record_identifier = next((r["id"] for r in record["result"] if r["type"] == "A"), None)
# Change the IP@Cloudflare using the API
update_url = f"https://api.cloudflare.com/client/v4/zones/{zone_identifier}/dns_records/{record_identifier}"
update_data = {
"type": "A",
"name": record_name,
"content": ip,
"ttl": ttl,
"proxied": proxy
}
update_response = requests.patch(update_url, headers=record_headers, json=update_data)
update = update_response.json()
# Report the status
if not update.get("success", False):
error_message = f"DDNS Updater: {ip} {record_name} DDNS failed for {record_identifier} ({ip}). DUMPING RESULTS:\n{update}"
logger.error(error_message)
if slackuri:
requests.post(slackuri, json={
"channel": slackchannel,
"text": f"{sitename} DDNS Update Failed: {record_name}: {record_identifier} ({ip})."
})
if discorduri:
requests.post(discorduri, headers={"Content-Type": "application/json"}, json={
"content": f"{sitename} DDNS Update Failed: {record_name}: {record_identifier} ({ip})."
})
exit(1)
else:
success_message = f"DDNS Updater: {ip} {record_name} DDNS updated."
logger.info(success_message)
if slackuri:
requests.post(slackuri, json={
"channel": slackchannel,
"text": f"{sitename} Updated: {record_name}'s new IP Address is {ip}"
})
if discorduri:
requests.post(discorduri, headers={"Content-Type": "application/json"}, json={
"content": f"{sitename} Updated: {record_name}'s new IP Address is {ip}"
})
exit(0)
s
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment