dynamic-infra-tooling/infra_cli/cloudflare.py

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)