from datetime import datetime import dns.resolver import logging import os import re import requests import time from typing import Dict, List, Optional from urllib.parse import quote 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: """ Claim a hostname below a user's namespace. DNS is provisioned once per user with a namespace CNAME and wildcard CNAME in the parent zone. Individual hostnames are application records. Returns: dict with 'success', 'message', and optionally 'domain_data' """ try: # Validate inputs if not username or subdomain_name is None: 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() if not self._is_valid_username_label(username): return {'success': False, 'message': "Username cannot be represented as a DNS name", 'domain_data': None} # Hostnames are single labels below u-. if subdomain_name and 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}" if subdomain_name else user_domain namespace = self.ensure_user_namespace(username) if not namespace['success']: return {'success': False, 'message': namespace['message'], 'domain_data': None} domain_info = { 'domain': full_subdomain, 'subdomain': subdomain_name, 'username': username, 'record_type': 'CNAME', 'ip_address': self.base_domain, 'ttl': 3600, 'created_at': datetime.now().isoformat() } return { 'success': True, 'message': f"Domain {full_subdomain} is ready", 'domain_data': domain_info } 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 ensure_user_namespace(self, username: str) -> Dict: """Ensure the base and wildcard CNAME records exist for a user.""" username = (username or '').lower().strip() if not self._is_valid_username_label(username): return {'success': False, 'message': "Username cannot be represented as a DNS name"} user_domain = f"u-{username}.{self.base_domain}" for owner in (f"u-{username}", f"*.u-{username}"): if not self._ensure_namespace_cname(owner): return { 'success': False, 'message': f"Could not safely provision DNS for {user_domain}; check for existing DNS records or a legacy delegation", } return {'success': True, 'message': f"DNS namespace is ready for {user_domain}"} def _ensure_namespace_cname(self, owner: str) -> bool: """Create an idempotent CNAME in the configured parent zone.""" try: path = f"{self.base_url}/domains/{self.base_domain}/rrsets/{quote(owner, safe='.-*')}/CNAME/" response = requests.get(path, headers=self.headers, timeout=30) if response.status_code == 200: records = response.json().get('records', []) return records == [f"{self.base_domain}."] or records == [self.base_domain] if response.status_code != 404: logger.error("Could not inspect DNS record %s (%s)", owner, response.status_code) return False # CNAME cannot coexist with other record types at the same owner. for record_type in ('A', 'AAAA', 'TXT', 'MX', 'NS', 'DS', 'CAA', 'SRV', 'TLSA', 'HTTPS', 'SVCB'): conflict = requests.get( f"{self.base_url}/domains/{self.base_domain}/rrsets/{quote(owner, safe='.-*')}/{record_type}/", headers=self.headers, timeout=30, ) if conflict.status_code == 200: logger.error("Refusing to replace existing %s record at %s", record_type, owner) return False if conflict.status_code != 404: logger.error("Could not inspect %s record at %s (%s)", record_type, owner, conflict.status_code) return False payload = { "subname": owner, "type": "CNAME", "ttl": 3600, "records": [f"{self.base_domain}."] } created = requests.post( f"{self.base_url}/domains/{self.base_domain}/rrsets/", headers=self.headers, json=payload, timeout=30, ) if created.status_code in (200, 201): return True logger.error("Could not create CNAME %s (%s)", owner, created.status_code) return False except requests.exceptions.RequestException as exc: logger.error("DNS API request failed for %s: %s", owner, exc) return False except (ValueError, TypeError) as exc: logger.error("Invalid DNS API response for %s: %s", owner, exc) return False 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 subdomain_name is None: return { 'success': False, 'message': "Username and subdomain name are required" } username = username.lower().strip() subdomain_name = subdomain_name.lower().strip() if not self._is_valid_username_label(username): return {'success': False, 'message': "Invalid username or subdomain"} user_domain = f"u-{username}.{self.base_domain}" # Releasing the optional root claim must never remove the shared namespace DNS. if not subdomain_name: return {'success': True, 'message': f"Domain claim {user_domain} deleted"} if not self._is_valid_subdomain(subdomain_name): return {'success': False, 'message': "Invalid username or subdomain"} full_subdomain = f"{subdomain_name}.{user_domain}" # New namespaces use a shared wildcard in the parent zone. Do not # remove shared DNS or stale child-zone data when releasing a claim. parent_cname = requests.get( f"{self.base_url}/domains/{self.base_domain}/rrsets/{quote(f'u-{username}', safe='.-')}/CNAME/", headers=self.headers, timeout=30, ) if parent_cname.status_code == 200: records = parent_cname.json().get('records', []) if records in ([f"{self.base_domain}."], [self.base_domain]): for owner in (f"{subdomain_name}.u-{username}", f"www.{subdomain_name}.u-{username}"): rrsets = requests.get( f"{self.base_url}/domains/{self.base_domain}/rrsets/", headers=self.headers, params={'subname': owner}, timeout=30, ) if rrsets.status_code != 200: return {'success': False, 'message': f"Could not inspect DNS records for {owner}"} for rrset in rrsets.json(): delete_url = ( f"{self.base_url}/domains/{self.base_domain}/rrsets/" f"{quote(owner, safe='.-*')}/{rrset['type']}/" ) deleted = requests.delete(delete_url, headers=self.headers, timeout=30) if deleted.status_code not in (200, 204, 404): return {'success': False, 'message': f"Could not remove DNS record {owner} {rrset['type']}"} return {'success': True, 'message': f"Domain claim {full_subdomain} deleted"} elif parent_cname.status_code != 404: return {'success': False, 'message': f"Could not inspect parent DNS for {user_domain}"} # Existing delegated domains retain their old record cleanup until # their namespace has been migrated to the parent-zone CNAME model. zone_exists = self._check_domain_exists(user_domain) if zone_exists is False: return { 'success': True, 'message': f"Domain claim {full_subdomain} deleted" } if zone_exists is None: return {'success': False, 'message': f"Could not inspect legacy zone {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 _is_valid_username_label(self, username: str) -> bool: """Validate the username's DNS path while preserving its dot separators.""" labels = f"u-{username}".split('.') label_pattern = r"[a-z0-9](?:[a-z0-9-]*[a-z0-9])?" return ( all(len(label) <= 63 and re.fullmatch(label_pattern, label) for label in labels) and len('.'.join(labels) + f".{self.base_domain}") <= 253 ) 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() # First validate the subdomain name if subdomain_name and not self._is_valid_subdomain(subdomain_name): return { 'available': False, 'message': f"Invalid subdomain name: {subdomain_name}" } if not self._is_valid_username_label(username): return {'available': False, 'message': "Username cannot be represented as a DNS name"} # Individual hostnames are claims stored in user_domains. Wildcard # DNS means DNS existence cannot determine whether a name is free. return { 'available': True, 'message': f"Subdomain {subdomain_name} is available" } 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}")