from datetime import datetime import dns.resolver import logging import os import re import requests import time from typing import Dict, List, Optional logger = logging.getLogger(__name__) class DomainManager: def __init__(self): # Token deSEC API - should be in environment variables self.desesec_api_token = os.environ.get('DESEC_API_TOKEN', '') if not self.desesec_api_token: logger.warning("deSEC API is disabled because DESEC_API_TOKEN is not configured") self.base_domain = os.environ.get('BASE_DOMAIN', 'DOMAIN') self.node_ip = os.environ.get('NODE_IP', 'IP') # Initialize the client self.base_url = "https://desec.io/api/v1" self.headers = { "Authorization": f"Token {self.desesec_api_token}", "Content-Type": "application/json" } logger.info(f"Domain Manager initialized for {self.base_domain}") if not self.test_connection(): logger.error("✗ Failed to connect to deSEC API!") # Don't raise error, just log - we'll handle gracefully def test_connection(self) -> bool: """Test connection to deSEC API""" if not self.desesec_api_token: return False try: url = f"{self.base_url}/domains/" response = requests.get(url, headers=self.headers, timeout=30) if response.status_code == 200: domains = response.json() logger.info(f"✓ deSEC API connection successful") logger.info(f"✓ Account has {len(domains)} domains") return True else: logger.error(f"✗ deSEC API connection failed: {response.status_code}") return False except Exception as e: logger.error(f"✗ deSEC API connection error: {e}") return False def _get_child_domain_ds_records(self, child_domain: str) -> List[str]: """Get DS records from child domain's DNSKEY""" try: url = f"{self.base_url}/domains/{child_domain}/" logger.info(f"Fetching DNSKEY info for: {child_domain}") response = requests.get(url, headers=self.headers, timeout=30) if response.status_code != 200: logger.error(f"Failed to get child domain info: {response.status_code}") return [] domain_info = response.json() ds_records = domain_info.get('keys', [{}])[0].get('ds', []) logger.info(f"Found {len(ds_records)} DS records for {child_domain}") for ds in ds_records: logger.info(f" DS: {ds}") return ds_records except Exception as e: logger.error(f"Error getting DS records: {e}") return [] def _check_existing_ds_record(self, subdomain_prefix: str) -> tuple[bool, List[str]]: """Check if DS record already exists in parent domain""" try: url = f"{self.base_url}/domains/{self.base_domain}/rrsets/?subname={subdomain_prefix}&type=DS" response = requests.get(url, headers=self.headers, timeout=30) if response.status_code != 200: logger.error(f"Failed to check existing DS records: {response.status_code}") return False, [] records = response.json() if records and len(records) > 0: existing_records = records[0].get('records', []) logger.info(f"Found existing DS records: {existing_records}") return True, existing_records return False, [] except Exception as e: logger.error(f"Error checking existing DS records: {e}") return False, [] def _delete_existing_ds_record(self, subdomain_prefix: str) -> bool: """Delete existing DS record from parent domain""" try: url = f"{self.base_url}/domains/{self.base_domain}/rrsets/{subdomain_prefix}/DS/" response = requests.delete(url, headers=self.headers, timeout=30) if response.status_code in [200, 204]: logger.info(f"✓ Deleted existing DS record for {subdomain_prefix}") return True elif response.status_code == 404: logger.info(f"No DS record to delete for {subdomain_prefix}") return True else: logger.error(f"Failed to delete DS record: {response.status_code}") return False except Exception as e: logger.error(f"Error deleting DS record: {e}") return False def _create_ds_record(self, subdomain_prefix: str, ds_records: List[str]) -> bool: """Create DS record in parent domain""" try: url = f"{self.base_url}/domains/{self.base_domain}/rrsets/" data = { "subname": subdomain_prefix, "type": "DS", "ttl": 3600, "records": ds_records } logger.info(f"Creating DS record for {subdomain_prefix}") logger.info(f"DS records: {ds_records}") response = requests.post(url, headers=self.headers, json=data, timeout=30) if response.status_code in [200, 201]: logger.info(f"✓ Created DS record for {subdomain_prefix}") return True elif response.status_code == 400: # Try PUT if POST fails (record might exist) logger.info("Trying PUT instead of POST...") return self._update_ds_record(subdomain_prefix, ds_records) else: logger.error(f"Failed to create DS record: {response.status_code} - {response.text}") return False except Exception as e: logger.error(f"Error creating DS record: {e}") return False def _update_ds_record(self, subdomain_prefix: str, ds_records: List[str]) -> bool: """Update existing DS record in parent domain""" try: url = f"{self.base_url}/domains/{self.base_domain}/rrsets/{subdomain_prefix}/DS/" data = { "subname": subdomain_prefix, "type": "DS", "ttl": 3600, "records": ds_records } response = requests.put(url, headers=self.headers, json=data, timeout=30) if response.status_code in [200, 201]: logger.info(f"✓ Updated DS record for {subdomain_prefix}") return True else: logger.error(f"Failed to update DS record: {response.status_code} - {response.text}") return False except Exception as e: logger.error(f"Error updating DS record: {e}") return False def _wait_for_dns_propagation(self, child_domain: str, timeout: int = 300) -> bool: """Wait for DS record to propagate to nameservers""" logger.info(f"Waiting for DNS propagation (max {timeout}s)...") start_time = time.time() check_interval = 10 while time.time() - start_time < timeout: try: # Check if DS record appears on authoritative nameserver resolver = dns.resolver.Resolver() resolver.nameservers = ['8.8.8.8'] # Use Google DNS to bypass cache try: answer = resolver.resolve(child_domain, 'DS') if len(answer) > 0: logger.info(f"✓ DS record propagated: {[str(r) for r in answer]}") return True except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN): pass logger.info("DS record not yet propagated, waiting...") time.sleep(check_interval) except Exception as e: logger.warning(f"Error checking propagation: {e}") time.sleep(check_interval) logger.warning(f"DS record propagation timeout after {timeout}s") return False def _verify_dnssec_chain(self, child_domain: str) -> bool: """Verify DNSSEC chain is valid""" try: logger.info(f"Verifying DNSSEC chain for {child_domain}...") # Check DNSKEY record resolver = dns.resolver.Resolver() resolver.nameservers = ['8.8.8.8'] try: dnskey_answer = resolver.resolve(child_domain, 'DNSKEY') logger.info(f"✓ DNSKEY record found with {len(dnskey_answer)} entries") # Check if we get authenticated data flag soa_answer = resolver.resolve(child_domain, 'SOA') # Note: Python's dnspython doesn't expose AD flag directly # We'll check for RRSIG records instead # Check for RRSIG on SOA soa_rrset = resolver.resolve(child_domain, 'SOA', raise_on_no_answer=False) if hasattr(soa_rrset.response, 'answer') and len(soa_rrset.response.answer) > 0: for rrset in soa_rrset.response.answer: if rrset.rdtype == dns.rdatatype.SOA: # Check for RRSIG in additional section if len(soa_rrset.response.additional) > 0: logger.info("✓ RRSIG records present (DNSSEC signing active)") return True logger.warning("⚠ RRSIG records not found, DNSSEC might not be fully active") return True # Still return True as the chain might be building except dns.resolver.NoAnswer: logger.error("✗ No DNSKEY answer received") return False except Exception as e: logger.error(f"✗ DNSSEC verification error: {e}") return False except Exception as e: logger.error(f"Error in DNSSEC verification: {e}") return False def _get_child_domain_ns_records(self, child_domain: str, retries: int = 3) -> List[str]: """Get NS records from child domain with retries""" for attempt in range(retries): try: url = f"{self.base_url}/domains/{child_domain}/" response = requests.get(url, headers=self.headers, timeout=30) if response.status_code == 200: domain_info = response.json() ns_records = domain_info.get('nameservers', []) if ns_records: logger.info(f"Found {len(ns_records)} NS records for {child_domain} (attempt {attempt + 1})") return ns_records logger.warning(f"Attempt {attempt + 1}: No NS records yet for {child_domain}") if attempt < retries - 1: time.sleep(2) except Exception as e: logger.error(f"Error getting NS records (attempt {attempt + 1}): {e}") if attempt < retries - 1: time.sleep(2) return [] def _get_child_domain_ds_records(self, child_domain: str, retries: int = 3) -> List[str]: """Get DS records from child domain's DNSKEY with retries""" for attempt in range(retries): try: url = f"{self.base_url}/domains/{child_domain}/" logger.info(f"Fetching DNSKEY info for: {child_domain} (attempt {attempt + 1})") response = requests.get(url, headers=self.headers, timeout=30) if response.status_code == 200: domain_info = response.json() ds_records = domain_info.get('keys', [{}])[0].get('ds', []) if ds_records: logger.info(f"Found {len(ds_records)} DS records for {child_domain}") return ds_records logger.warning(f"Attempt {attempt + 1}: No DS records yet for {child_domain}") if attempt < retries - 1: time.sleep(2) except Exception as e: logger.error(f"Error getting DS records: {e}") if attempt < retries - 1: time.sleep(2) return [] def _check_existing_ns_record(self, subdomain_prefix: str) -> tuple[bool, List[str]]: """Check if NS record already exists in parent domain""" try: url = f"{self.base_url}/domains/{self.base_domain}/rrsets/?subname={subdomain_prefix}&type=NS" response = requests.get(url, headers=self.headers, timeout=30) if response.status_code == 200: records = response.json() if records and len(records) > 0: existing_records = records[0].get('records', []) # Clean up trailing dots for comparison cleaned_records = [r.rstrip('.') for r in existing_records] logger.info(f"Found existing NS records: {cleaned_records}") return True, cleaned_records return False, [] except Exception as e: logger.error(f"Error checking existing NS records: {e}") return False, [] def _create_ns_record(self, subdomain_prefix: str, ns_records: List[str]) -> bool: """Create NS record in parent domain""" try: # Ensure nameservers end with a dot for deSEC formatted_ns = [ns if ns.endswith('.') else f"{ns}." for ns in ns_records] url = f"{self.base_url}/domains/{self.base_domain}/rrsets/" data = { "subname": subdomain_prefix, "type": "NS", "ttl": 3600, "records": formatted_ns } logger.info(f"Creating NS record for {subdomain_prefix}") response = requests.post(url, headers=self.headers, json=data, timeout=30) if response.status_code in [200, 201]: logger.info(f"✓ Created NS record for {subdomain_prefix}") return True elif response.status_code == 400: # Try PUT if record exists logger.info("NS record likely exists, trying PUT...") url_put = f"{self.base_url}/domains/{self.base_domain}/rrsets/{subdomain_prefix}/NS/" response = requests.put(url_put, headers=self.headers, json=data, timeout=30) return response.status_code in [200, 201] logger.error(f"Failed to create NS record: {response.status_code} - {response.text}") return False except Exception as e: logger.error(f"Error creating NS record: {e}") return False def setup_dnssec_delegation(self, username: str) -> Dict: """ Set up proper DNSSEC delegation for a user's domain Process: 1. Get NS and DS records from child domain (u-username.anchorhost.eu) 2. Create/update NS and DS record in parent domain (anchorhost.eu) 3. Wait for propagation 4. Verify DNSSEC chain """ try: username = username.lower().strip() child_domain = f"u-{username}.{self.base_domain}" subdomain_prefix = f"u-{username}" logger.info("="*60) logger.info(f"Setting up DNSSEC delegation for: {child_domain}") logger.info("="*60) # Step 1: Check if child domain exists if not self._check_domain_exists(child_domain): logger.error(f"Child domain doesn't exist: {child_domain}") return { 'success': False, 'message': f"Child domain {child_domain} doesn't exist" } # Step 2: Handle NS delegation logger.info("Step 1: Handling NS delegation...") ns_records = self._get_child_domain_ns_records(child_domain) if not ns_records: logger.error("No NS records found for child domain") return { 'success': False, 'message': "No NS records found for child domain" } exists_ns, existing_ns = self._check_existing_ns_record(subdomain_prefix) if exists_ns and set(existing_ns) == set([ns.rstrip('.') for ns in ns_records]): logger.info("✓ NS records already correctly configured") else: if not self._create_ns_record(subdomain_prefix, ns_records): logger.error("Failed to create NS record delegation") return { 'success': False, 'message': "Failed to create NS record delegation" } # Step 3: Handle DS records logger.info("Step 2: Handling DS records...") ds_records = self._get_child_domain_ds_records(child_domain) if not ds_records: logger.warning("No DS records found in child domain (DNSSEC might not be enabled yet)") else: exists_ds, existing_ds = self._check_existing_ds_record(subdomain_prefix) if exists_ds and set(existing_ds) == set(ds_records): logger.info("✓ DS record already exists and matches") else: if exists_ds: logger.info("DS record exists but doesn't match, deleting...") self._delete_existing_ds_record(subdomain_prefix) if not self._create_ds_record(subdomain_prefix, ds_records): logger.warning("Failed to create DS record") # Step 4: Wait for propagation logger.info("Step 3: Waiting for DNS propagation...") if not self._wait_for_dns_propagation(child_domain, 120): # 2 minutes timeout logger.warning("DS record propagation timeout, continuing anyway...") # Step 5: Verify DNSSEC chain logger.info("Step 4: Verifying DNSSEC chain...") if not self._verify_dnssec_chain(child_domain): logger.warning("DNSSEC chain verification had issues, but continuing...") logger.info("="*60) logger.info(f"✅ Delegation setup completed for {child_domain}") logger.info("="*60) return { 'success': True, 'message': f"Delegation configured for {child_domain}", 'ds_records': ds_records if 'ds_records' in locals() else [] } except Exception as e: logger.error(f"Error setting up DNSSEC delegation: {e}", exc_info=True) return { 'success': False, 'message': f"DNSSEC setup error: {str(e)}" } def create_subdomain(self, username: str, subdomain_name: str) -> Dict: """ Create a subdomain for user with proper DNSSEC setup Returns: dict with 'success', 'message', and optionally 'domain_data' """ try: # Validate inputs if not username or not subdomain_name: return { 'success': False, 'message': "Username and subdomain name are required", 'domain_data': None } # Clean and validate inputs username = username.lower().strip() subdomain_name = subdomain_name.lower().strip() # Validate subdomain format if not self._is_valid_subdomain(subdomain_name): return { 'success': False, 'message': "Invalid subdomain name. Use only letters, numbers, and hyphens", 'domain_data': None } # Construct domains user_domain = f"u-{username}.{self.base_domain}" full_subdomain = f"{subdomain_name}.{user_domain}" logger.info("="*60) logger.info(f"Creating subdomain: {full_subdomain}") logger.info(f"User domain: {user_domain}") logger.info(f"Subdomain: {subdomain_name}") logger.info(f"Target IP: {self.node_ip}") logger.info("="*60) # Step 1: Ensure user's domain exists logger.info(f"Step 1: Checking if user domain exists: {user_domain}") user_domain_exists = self._check_domain_exists(user_domain) logger.info(f"Domain exists: {user_domain_exists}") if user_domain_exists is None: return { 'success': False, 'message': f"Error checking domain {user_domain}", 'domain_data': None } if not user_domain_exists: logger.info(f"Step 2: Creating user domain: {user_domain}") domain_created = self._create_domain(user_domain) if not domain_created: return { 'success': False, 'message': f"Failed to create user domain {user_domain}", 'domain_data': None } logger.info(f"✓ User domain created: {user_domain}") # Step 3: Setup DNSSEC delegation for the new domain logger.info(f"Step 3: Setting up DNSSEC delegation...") dnssec_result = self.setup_dnssec_delegation(username) if not dnssec_result.get('success'): logger.warning(f"⚠ DNSSEC setup had issues: {dnssec_result.get('message')}") # Continue anyway, as DNSSEC is not critical for basic functionality else: logger.info(f"✓ User domain already exists: {user_domain}") # Check if DNSSEC is properly configured logger.info("Checking DNSSEC configuration...") dnssec_result = self.setup_dnssec_delegation(username) if not dnssec_result.get('success'): logger.warning(f"⚠ DNSSEC check had issues: {dnssec_result.get('message')}") # Step 4: Create the A record for the subdomain logger.info(f"Step 4: Creating A record for subdomain: {subdomain_name}") logger.info(f"Target: {subdomain_name}.{user_domain} → {self.node_ip}") url = f"{self.base_url}/domains/{user_domain}/rrsets/" data = { "subname": subdomain_name, "type": "A", "ttl": 3600, "records": [self.node_ip] } logger.info(f"POST URL: {url}") response = requests.post(url, headers=self.headers, json=data, timeout=30) logger.info(f"Response Status: {response.status_code}") if response.status_code in [200, 201]: domain_info = { 'domain': full_subdomain, 'subdomain': subdomain_name, 'username': username, 'record_type': 'A', 'ip_address': self.node_ip, 'ttl': 3600, 'created_at': datetime.now().isoformat() } logger.info("="*60) logger.info(f"✅ SUCCESS: Subdomain created!") logger.info(f"Domain: {full_subdomain}") logger.info(f"Points to: {self.node_ip}") logger.info("="*60) # Step 5: Create www CNAME record self._create_www_cname(username, subdomain_name) return { 'success': True, 'message': f"Subdomain {full_subdomain} created successfully", 'domain_data': domain_info } else: error_msg = f"API error: {response.status_code} - {response.text}" logger.error("="*60) logger.error(f"❌ FAILED to create subdomain") logger.error(f"Status: {response.status_code}") logger.error(f"Error: {response.text}") logger.error("="*60) return { 'success': False, 'message': error_msg, 'domain_data': None } except requests.exceptions.RequestException as e: logger.error(f"Network error creating subdomain: {e}") return { 'success': False, 'message': f"Network error: {str(e)}", 'domain_data': None } except Exception as e: logger.error(f"Error creating subdomain: {e}", exc_info=True) return { 'success': False, 'message': f"Internal error: {str(e)}", 'domain_data': None } def _create_www_cname(self, username: str, subdomain_name: str): """Create www CNAME record for the subdomain""" try: user_domain = f"u-{username}.{self.base_domain}" # Create www.subdomain CNAME pointing to the subdomain cname_data = { "subname": f"www.{subdomain_name}", "type": "CNAME", "ttl": 3600, "records": [f"{subdomain_name}.{user_domain}."] # Note the trailing dot } url = f"{self.base_url}/domains/{user_domain}/rrsets/" response = requests.post(url, headers=self.headers, json=cname_data, timeout=30) if response.status_code in [200, 201]: logger.info(f"✓ Created www CNAME for {subdomain_name}.{user_domain}") else: logger.warning(f"Failed to create www CNAME: {response.status_code}") except Exception as e: logger.error(f"Error creating www CNAME: {e}") def _check_domain_exists(self, domain_name: str) -> Optional[bool]: """Check if a domain exists in deSEC""" try: url = f"{self.base_url}/domains/{domain_name}/" response = requests.get(url, headers=self.headers, timeout=30) if response.status_code == 200: logger.info(f"✓ Domain exists: {domain_name}") return True elif response.status_code == 404: logger.info(f"✓ Domain does not exist: {domain_name}") return False else: logger.error(f"Unexpected response: {response.status_code} - {response.text}") return None except Exception as e: logger.error(f"Error checking domain: {e}") return None def _create_domain(self, domain_name: str) -> bool: """Create a new domain in deSEC""" try: url = f"{self.base_url}/domains/" data = {"name": domain_name} response = requests.post(url, headers=self.headers, json=data, timeout=30) if response.status_code in [200, 201]: logger.info(f"✓ Domain {domain_name} created successfully") return True else: logger.error(f"✗ Failed to create domain {domain_name}") logger.error(f"Status: {response.status_code}") logger.error(f"Error: {response.text}") return False except Exception as e: logger.error(f"Error creating domain: {e}", exc_info=True) return False def delete_subdomain(self, username: str, subdomain_name: str) -> Dict: """Delete a subdomain""" try: if not username or not subdomain_name: return { 'success': False, 'message': "Username and subdomain name are required" } username = username.lower().strip() subdomain_name = subdomain_name.lower().strip() user_domain = f"u-{username}.{self.base_domain}" full_subdomain = f"{subdomain_name}.{user_domain}" logger.info("="*60) logger.info(f"Deleting subdomain: {full_subdomain}") logger.info("="*60) # Check if user domain exists if not self._check_domain_exists(user_domain): return { 'success': False, 'message': f"User domain {user_domain} not found" } # Delete the A record url = f"{self.base_url}/domains/{user_domain}/rrsets/{subdomain_name}/A/" response = requests.delete(url, headers=self.headers, timeout=30) # Try to delete www CNAME if it exists try: www_url = f"{self.base_url}/domains/{user_domain}/rrsets/www.{subdomain_name}/CNAME/" www_response = requests.delete(www_url, headers=self.headers, timeout=10) if www_response.status_code in [200, 204]: logger.info(f"✓ Also deleted www CNAME record") except: pass if response.status_code in [200, 204]: logger.info(f"✓ Deleted subdomain: {full_subdomain}") return { 'success': True, 'message': f"Subdomain {full_subdomain} deleted successfully" } elif response.status_code == 404: return { 'success': False, 'message': f"Subdomain {full_subdomain} not found" } else: error_msg = f"API error: {response.status_code} - {response.text}" logger.error(f"✗ Failed to delete subdomain: {error_msg}") return { 'success': False, 'message': error_msg } except requests.exceptions.RequestException as e: logger.error(f"Network error deleting subdomain: {e}") return { 'success': False, 'message': f"Network error: {str(e)}" } except Exception as e: logger.error(f"Error deleting subdomain: {e}") return { 'success': False, 'message': f"Internal error: {str(e)}" } def list_user_domains(self, username: str) -> Dict: """List all domains/subdomains for a user""" try: if not username: return { 'success': False, 'message': "Username required", 'domains': [] } username = username.lower().strip() user_domain = f"u-{username}.{self.base_domain}" logger.info(f"Listing domains for user: {username}") # Check if user domain exists if not self._check_domain_exists(user_domain): logger.info(f"User domain doesn't exist: {user_domain}") return { 'success': True, 'message': "No domains found for user", 'domains': [] } # Get all records for the user domain url = f"{self.base_url}/domains/{user_domain}/rrsets/" response = requests.get(url, headers=self.headers, timeout=30) if response.status_code != 200: return { 'success': False, 'message': f"API error: {response.status_code}", 'domains': [] } records = response.json() user_domains = [] for record in records: # Filter for A records that are subdomains (not empty subname) if record.get('type') == 'A' and record.get('subname'): domain_info = { 'subdomain': record.get('subname', ''), 'full_domain': f"{record.get('subname', '')}.{user_domain}", 'record_type': record.get('type', ''), 'ip_address': record.get('records', [''])[0] if record.get('records') else '', 'ttl': record.get('ttl', 3600), 'created_at': record.get('created', datetime.now().isoformat()) } user_domains.append(domain_info) logger.info(f"Retrieved {len(user_domains)} domains for user {username}") return { 'success': True, 'message': f"Found {len(user_domains)} domains", 'domains': user_domains } except requests.exceptions.RequestException as e: logger.error(f"Network error listing domains: {e}") return { 'success': False, 'message': f"Network error: {str(e)}", 'domains': [] } except Exception as e: logger.error(f"Error listing domains: {e}") return { 'success': False, 'message': f"Internal error: {str(e)}", 'domains': [] } def _is_valid_subdomain(self, subdomain: str) -> bool: """Validate subdomain name format""" # Allow letters, numbers, and hyphens, 1-63 chars # Can't start or end with hyphen, can't have consecutive hyphens pattern = r'^[a-z0-9]([a-z0-9-]*[a-z0-9])?$' if not re.match(pattern, subdomain): return False if len(subdomain) > 63: return False if '--' in subdomain: return False return True def check_domain_availability(self, subdomain_name: str, username: str) -> Dict: """Check if a subdomain is available for the user""" try: username = username.lower().strip() subdomain_name = subdomain_name.lower().strip() user_domain = f"u-{username}.{self.base_domain}" # First validate the subdomain name if not self._is_valid_subdomain(subdomain_name): return { 'available': False, 'message': f"Invalid subdomain name: {subdomain_name}" } # Check if user domain exists first if not self._check_domain_exists(user_domain): return { 'available': True, 'message': f"Subdomain {subdomain_name} is available" } # Check if the specific A record exists url = f"{self.base_url}/domains/{user_domain}/rrsets/{subdomain_name}/A/" response = requests.get(url, headers=self.headers, timeout=30) if response.status_code == 404: return { 'available': True, 'message': f"Subdomain {subdomain_name} is available" } elif response.status_code == 200: return { 'available': False, 'message': f"Subdomain {subdomain_name} already exists" } else: return { 'available': False, 'message': f"API error: {response.status_code}" } except Exception as e: return { 'available': False, 'message': f"Error checking availability: {str(e)}" } def fix_all_dnssec_delegations(self): """Fix DNSSEC delegation for all existing user domains""" try: # Get all domains url = f"{self.base_url}/domains/" response = requests.get(url, headers=self.headers, timeout=30) if response.status_code != 200: logger.error(f"Failed to get domains: {response.status_code}") return domains = response.json() fixed_count = 0 error_count = 0 for domain in domains: domain_name = domain.get('name', '') # Check if it's a user domain (starts with u- and ends with .anchorhost.eu) if domain_name.startswith('u-') and domain_name.endswith(f'.{self.base_domain}'): # Extract username from domain username = domain_name.replace(f".{self.base_domain}", "").replace("u-", "") logger.info(f"Fixing DNSSEC delegation for: {domain_name} (user: {username})") result = self.setup_dnssec_delegation(username) if result.get('success'): fixed_count += 1 logger.info(f"✓ Fixed DNSSEC for {domain_name}") else: error_count += 1 logger.error(f"✗ Failed to fix DNSSEC for {domain_name}: {result.get('message')}") logger.info(f"DNSSEC fix completed: {fixed_count} fixed, {error_count} errors") except Exception as e: logger.error(f"Error fixing DNSSEC delegations: {e}")