108 lines
No EOL
3.7 KiB
Python
108 lines
No EOL
3.7 KiB
Python
import requests
|
|
import json
|
|
import os
|
|
|
|
class CloudflareManager:
|
|
def __init__(self, config):
|
|
self.token = config.get('cloudflare.token')
|
|
# Path to store the dynamic list of domains
|
|
self.state_file = config.get('cloudflare.ddns_state_file', '/etc/haproxy/dynamic/ddns_domains.json')
|
|
self.api_url = "https://api.cloudflare.com/client/v4"
|
|
self.headers = {
|
|
"Authorization": f"Bearer {self.token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
def _load_domains(self):
|
|
if not os.path.exists(self.state_file):
|
|
return []
|
|
try:
|
|
with open(self.state_file, 'r') as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, IOError):
|
|
return []
|
|
|
|
def _save_domains(self, domains):
|
|
os.makedirs(os.path.dirname(self.state_file), exist_ok=True)
|
|
with open(self.state_file, 'w') as f:
|
|
json.dump(list(set(domains)), f, indent=4)
|
|
|
|
def add_domain(self, domain):
|
|
domains = self._load_domains()
|
|
if domain not in domains:
|
|
domains.append(domain)
|
|
self._save_domains(domains)
|
|
return True
|
|
return False
|
|
|
|
def remove_domain(self, domain):
|
|
domains = self._load_domains()
|
|
if domain in domains:
|
|
domains.remove(domain)
|
|
self._save_domains(domains)
|
|
return True
|
|
return False
|
|
|
|
def list_domains(self):
|
|
return self._load_domains()
|
|
|
|
def get_external_ip(self):
|
|
try:
|
|
return requests.get("https://checkip.amazonaws.com").text.strip()
|
|
except Exception as e:
|
|
print(f"Error fetching external IP: {e}")
|
|
return None
|
|
|
|
def get_zone_id(self, domain):
|
|
res = requests.get(f"{self.api_url}/zones?name={domain}", headers=self.headers)
|
|
data = res.json()
|
|
if data.get('success') and data['result']:
|
|
return data['result'][0]['id']
|
|
return None
|
|
|
|
def update_ddns(self, force=False):
|
|
current_ip = self.get_external_ip()
|
|
if not current_ip:
|
|
return "Failed to determine current external IP."
|
|
|
|
domains = self._load_domains()
|
|
if not domains:
|
|
return "No domains configured for DDNS."
|
|
|
|
results = []
|
|
for domain in domains:
|
|
zone_id = self.get_zone_id(domain)
|
|
if not zone_id:
|
|
results.append(f"[{domain}] Zone not found.")
|
|
continue
|
|
|
|
# Find A record for the root domain
|
|
res = requests.get(f"{self.api_url}/zones/{zone_id}/dns_records?type=A&name={domain}", headers=self.headers)
|
|
records = res.json().get('result', [])
|
|
|
|
if not records:
|
|
results.append(f"[{domain}] No A record found to update.")
|
|
continue
|
|
|
|
record = records[0]
|
|
if record['content'] == current_ip and not force:
|
|
results.append(f"[{domain}] Already up to date ({current_ip}).")
|
|
continue
|
|
|
|
# Update record
|
|
update_data = {
|
|
"type": "A",
|
|
"name": domain,
|
|
"content": current_ip,
|
|
"ttl": 1, # Auto
|
|
"proxied": record.get('proxied', True)
|
|
}
|
|
u_res = requests.put(f"{self.api_url}/zones/{zone_id}/dns_records/{record['id']}",
|
|
headers=self.headers, json=update_data)
|
|
|
|
if u_res.json().get('success'):
|
|
results.append(f"[{domain}] Updated to {current_ip}.")
|
|
else:
|
|
results.append(f"[{domain}] Update failed: {u_res.text}")
|
|
|
|
return "\n".join(results) |