916 lines
38 KiB
Python
916 lines
38 KiB
Python
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-<username>.
|
|
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}")
|