From 16424507b8b4ecc9a5706da32e6da929e3effa19 Mon Sep 17 00:00:00 2001 From: Yuval Adam <_@yuv.al> Date: Fri, 18 Jul 2025 20:48:13 +0200 Subject: refactor --- pyproject.toml | 2 +- regflow/domain_manager.py | 439 ----------------------------------- regflow/domains.py | 443 ++++++++++++++++++++++++++++++++++++ regflow/providers/cloudflare.py | 170 ++++++++++++++ regflow/providers/cloudflare_api.py | 170 -------------- regflow/providers/namecheap.py | 241 ++++++++++++++++++++ regflow/providers/namecheap_api.py | 238 ------------------- regflow/tests/test_integration.py | 195 +++++++++++++++- 8 files changed, 1048 insertions(+), 850 deletions(-) delete mode 100755 regflow/domain_manager.py create mode 100755 regflow/domains.py create mode 100644 regflow/providers/cloudflare.py delete mode 100644 regflow/providers/cloudflare_api.py create mode 100644 regflow/providers/namecheap.py delete mode 100644 regflow/providers/namecheap_api.py diff --git a/pyproject.toml b/pyproject.toml index ec332f1..3fde396 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,7 +20,7 @@ dev = [ ] [project.scripts] -regflow = "regflow.domain_manager:main" +regflow = "regflow.domains:main" [tool.hatch.build.targets.wheel] packages = ["regflow"] \ No newline at end of file diff --git a/regflow/domain_manager.py b/regflow/domain_manager.py deleted file mode 100755 index 8905a0a..0000000 --- a/regflow/domain_manager.py +++ /dev/null @@ -1,439 +0,0 @@ -#!/usr/bin/env python3 - -import sys -from typing import Dict, Any, Optional -from .config import Config -from .providers.namecheap_api import NamecheapAPI -from .providers.cloudflare_api import CloudflareAPI - - -class DomainManager: - def __init__(self, config: Config): - self.config = config - self.namecheap = NamecheapAPI(config) - self.cloudflare = CloudflareAPI(config) - - def get_domain_status(self, domain: str) -> Dict[str, Any]: - """Get current status of domain across all services""" - status = { - "domain": domain, - "registered": False, - "cloudflare_zone": None, - "nameservers": {"namecheap": [], "cloudflare": []}, - "nameservers_match": False, - } - - # Check if domain is registered - try: - status["registered"] = self.namecheap.is_domain_registered(domain) - except Exception as e: - status["registration_error"] = str(e) - - # Get Cloudflare zone info - try: - zone_info = self.cloudflare.get_zone_info(domain) - if zone_info: - status["cloudflare_zone"] = { - "id": zone_info["id"], - "name": zone_info["name"], - "status": zone_info.get("status", "unknown"), - } - - # Get Cloudflare nameservers - cf_nameservers = self.cloudflare.get_zone_nameservers(zone_info["id"]) - status["nameservers"]["cloudflare"] = cf_nameservers - except Exception as e: - status["cloudflare_error"] = str(e) - - # Get Namecheap nameservers if domain is registered - if status["registered"]: - try: - nc_nameservers = self.namecheap.get_domain_nameservers(domain) - status["nameservers"]["namecheap"] = nc_nameservers - - # Check if nameservers match - cf_ns = set(status["nameservers"]["cloudflare"]) - nc_ns = set(nc_nameservers) - - # Only consider nameservers matching if we have both sets and they match - if len(cf_ns) > 0 and len(nc_ns) > 0: - status["nameservers_match"] = cf_ns == nc_ns - else: - # If we can't retrieve nameservers from either side, assume they don't match - status["nameservers_match"] = False - except Exception as e: - status["namecheap_ns_error"] = str(e) - - return status - - def print_domain_status(self, domain: str): - """Print formatted status of domain""" - status = self.get_domain_status(domain) - - print(f"\n=== Domain Status: {domain} ===") - - # Registration status - if status["registered"]: - print("✓ Domain is registered in Namecheap") - else: - print("✗ Domain is NOT registered in Namecheap") - if "registration_error" in status: - print(f" Error: {status['registration_error']}") - - # Cloudflare zone status - if status["cloudflare_zone"]: - zone = status["cloudflare_zone"] - print( - f"✓ Cloudflare zone exists (ID: {zone['id']}, Status: {zone['status']})" - ) - else: - print("✗ No Cloudflare zone found") - if "cloudflare_error" in status: - print(f" Error: {status['cloudflare_error']}") - - # Nameserver status - nc_ns = status["nameservers"]["namecheap"] - cf_ns = status["nameservers"]["cloudflare"] - - if nc_ns: - print(f"Namecheap nameservers: {', '.join(nc_ns)}") - else: - print("Namecheap nameservers: None") - - if cf_ns: - print(f"Cloudflare nameservers: {', '.join(cf_ns)}") - else: - print("Cloudflare nameservers: None") - - if len(nc_ns) == 0 and len(cf_ns) == 0: - print("⚠ Cannot retrieve nameservers from either service") - elif len(nc_ns) == 0: - print("⚠ Cannot retrieve Namecheap nameservers - unable to verify configuration") - elif len(cf_ns) == 0: - print("⚠ Cannot retrieve Cloudflare nameservers - unable to verify configuration") - elif status["nameservers_match"]: - print("✓ Nameservers are properly configured") - else: - print("✗ Nameservers do NOT match") - - print("=" * 50) - - def setup_domain( - self, - domain: str, - registrant_info: Optional[Dict[str, str]] = None, - setup_workers: bool = True, - dry_run: bool = False, - force_registration: bool = False, - ) -> Dict[str, Any]: - """ - Complete domain setup workflow (idempotent): - 1. Check domain registration status - 2. Register domain if needed (with user confirmation) - 3. Create Cloudflare zone if needed - 4. Update nameservers in Namecheap if needed - 5. Set up basic DNS records - """ - result = {"domain": domain, "steps_completed": [], "errors": []} - - try: - # Get current status - status = self.get_domain_status(domain) - - # Step 1: Handle domain registration - if not status["registered"]: - if not force_registration: - print( - f"Domain {domain} is not registered. Use --force-registration to register it." - ) - result["errors"].append( - "Domain not registered and force_registration not enabled" - ) - return result - - # Check availability - print(f"Checking availability for {domain}...") - if not self.namecheap.check_domain_availability(domain): - result["errors"].append( - f"Domain {domain} is not available for registration" - ) - return result - - # Get pricing and balance - try: - pricing = self.namecheap.get_domain_pricing(domain) - balance = self.namecheap.get_account_balance() - except Exception as e: - result["errors"].append(f"Failed to get pricing/balance: {str(e)}") - return result - - if balance < pricing["register"]: - result["errors"].append( - f"Insufficient balance. Required: ${pricing['register']:.2f}, " - f"Available: ${balance:.2f}" - ) - return result - - # User confirmation - print("\n" + "=" * 50) - print("DOMAIN REGISTRATION CONFIRMATION") - print("=" * 50) - print(f"Domain: {domain}") - print(f"Registration Price: ${pricing['register']:.2f}") - print(f"Account Balance: ${balance:.2f}") - print(f"Remaining Balance: ${balance - pricing['register']:.2f}") - print("=" * 50) - - print( - f"\nWARNING: This will charge ${pricing['register']:.2f} to your account!" - ) - - if not dry_run: - first_confirm = input( - f"Type 'REGISTER' to proceed with registration of {domain}: " - ).strip() - if first_confirm != "REGISTER": - result["errors"].append("Registration cancelled by user") - return result - - second_confirm = ( - input( - f"Are you absolutely sure you want to register {domain} for ${pricing['register']:.2f}? (yes/no): " - ) - .lower() - .strip() - ) - if second_confirm != "yes": - result["errors"].append("Registration cancelled by user") - return result - - # Register domain - print(f"Registering domain {domain}...") - if not self.namecheap.register_domain( - domain, registrant_info=registrant_info - ): - result["errors"].append(f"Failed to register domain {domain}") - return result - - result["steps_completed"].append("domain_registration") - print(f"✓ Domain {domain} registered successfully") - else: - print(f"DRY RUN: Would register domain {domain}") - result["steps_completed"].append("domain_registration_dry_run") - else: - print(f"✓ Domain {domain} is already registered") - result["steps_completed"].append("domain_already_registered") - - # Step 2: Handle Cloudflare zone creation - if not status["cloudflare_zone"]: - print(f"Creating Cloudflare zone for {domain}...") - try: - if not dry_run: - zone_info = self.cloudflare.add_zone(domain) - zone_id = zone_info["id"] - result["steps_completed"].append("cloudflare_zone_creation") - result["zone_id"] = zone_id - print(f"✓ Cloudflare zone created (ID: {zone_id})") - else: - print(f"DRY RUN: Would create Cloudflare zone for {domain}") - result["steps_completed"].append( - "cloudflare_zone_creation_dry_run" - ) - # For dry run, we can't continue with nameserver setup - return result - except Exception as e: - result["errors"].append( - f"Failed to create Cloudflare zone: {str(e)}" - ) - return result - else: - zone_id = status["cloudflare_zone"]["id"] - result["zone_id"] = zone_id - print(f"✓ Cloudflare zone already exists (ID: {zone_id})") - result["steps_completed"].append("cloudflare_zone_already_exists") - - # Step 3: Get Cloudflare nameservers - print("Getting Cloudflare nameservers...") - try: - nameservers = self.cloudflare.get_zone_nameservers(zone_id) - result["nameservers"] = nameservers - print(f"✓ Cloudflare nameservers: {', '.join(nameservers)}") - except Exception as e: - result["errors"].append( - f"Failed to get Cloudflare nameservers: {str(e)}" - ) - return result - - # Step 4: Update nameservers in Namecheap if needed - current_nc_nameservers = set(status["nameservers"]["namecheap"]) - cloudflare_nameservers = set(nameservers) - - if current_nc_nameservers != cloudflare_nameservers: - if not dry_run: - print("Updating nameservers in Namecheap...") - try: - if not self.namecheap.set_dns_servers(domain, nameservers): - result["errors"].append( - "Failed to update nameservers in Namecheap" - ) - return result - - result["steps_completed"].append("nameserver_update") - print("✓ Nameservers updated in Namecheap") - except Exception as e: - result["errors"].append( - f"Failed to update nameservers in Namecheap: {str(e)}" - ) - return result - else: - print( - f"DRY RUN: Would update nameservers in Namecheap to: {', '.join(nameservers)}" - ) - result["steps_completed"].append("nameserver_update_dry_run") - else: - print("✓ Nameservers already configured correctly") - result["steps_completed"].append("nameservers_already_configured") - - # Step 5: Set up basic DNS records (placeholder for now) - print("Setting up basic DNS records...") - try: - dns_records = [] - result["dns_records"] = dns_records - result["steps_completed"].append("basic_dns_setup") - print("✓ Basic DNS records ready (empty list as requested)") - except Exception as e: - result["errors"].append(f"Failed to set up DNS records: {str(e)}") - return result - - # Step 6: Set up worker subdomain if requested - if setup_workers: - print("Setting up worker subdomain...") - try: - if not dry_run: - worker_record = self.cloudflare.create_worker_subdomain( - zone_id, f"app.{domain}" - ) - result["worker_record"] = worker_record - result["steps_completed"].append("worker_subdomain_setup") - print(f"✓ Worker subdomain app.{domain} created") - else: - print(f"DRY RUN: Would create worker subdomain app.{domain}") - result["steps_completed"].append( - "worker_subdomain_setup_dry_run" - ) - except Exception as e: - result["errors"].append( - f"Failed to set up worker subdomain: {str(e)}" - ) - return result - - result["success"] = True - print(f"\n🎉 Domain {domain} setup completed successfully!") - if "zone_id" in result: - print(f"Zone ID: {result['zone_id']}") - if "nameservers" in result: - print(f"Nameservers: {', '.join(result['nameservers'])}") - - return result - - except Exception as e: - result["errors"].append(f"Unexpected error: {str(e)}") - return result - - def setup_google_analytics_dns(self, domain: str) -> Dict[str, Any]: - """Set up DNS records for Google Analytics""" - try: - zone_info = self.cloudflare.get_zone_info(domain) - if not zone_info: - return {"error": f"Domain {domain} not found in Cloudflare"} - - records = self.cloudflare.setup_google_analytics_dns( - zone_info["id"], domain - ) - return {"success": True, "records": records} - - except Exception as e: - return {"error": str(e)} - - -def main(): - if len(sys.argv) < 2: - print("Usage: regflow [options]") - print("") - print("Options:") - print(" --status Show current status of domain") - print(" --setup Set up domain (idempotent)") - print( - " --dry-run Show what would be done without making changes" - ) - print(" --force-registration Allow domain registration (costs money!)") - print(" --no-workers Skip worker subdomain setup") - print("") - print("Examples:") - print(" regflow example.com --status") - print(" regflow example.com --setup --dry-run") - print(" regflow example.com --setup --force-registration") - sys.exit(1) - - domain = sys.argv[1] - - # Parse command line arguments - show_status = "--status" in sys.argv - setup_domain = "--setup" in sys.argv - dry_run = "--dry-run" in sys.argv - force_registration = "--force-registration" in sys.argv - setup_workers = "--no-workers" not in sys.argv - - # Default to setup if no action specified - if not show_status and not setup_domain: - setup_domain = True - - # Load configuration - config = Config.from_env() - - # Validate required configuration - if not all( - [ - config.namecheap_api_key, - config.namecheap_api_user, - config.namecheap_username, - config.namecheap_client_ip, - config.cloudflare_api_token, - ] - ): - print("Error: Missing required API credentials. Please check your .env file.") - sys.exit(1) - - # Initialize domain manager - manager = DomainManager(config) - - # Handle status command - if show_status: - manager.print_domain_status(domain) - return - - # Handle setup command - if setup_domain: - if dry_run: - print("Running in DRY RUN mode - no actual changes will be made") - if force_registration: - print("FORCE REGISTRATION enabled - will register domain if needed") - - result = manager.setup_domain( - domain, - dry_run=dry_run, - force_registration=force_registration, - setup_workers=setup_workers, - ) - - if result.get("success"): - print(f"\nSuccess! Domain {domain} is ready for use.") - else: - print("\nErrors occurred:") - for error in result.get("errors", []): - print(f" - {error}") - sys.exit(1) - - -if __name__ == "__main__": - main() diff --git a/regflow/domains.py b/regflow/domains.py new file mode 100755 index 0000000..1e86780 --- /dev/null +++ b/regflow/domains.py @@ -0,0 +1,443 @@ +#!/usr/bin/env python3 + +import sys +from typing import Dict, Any, Optional +from .config import Config +from .providers.namecheap import NamecheapAPI +from .providers.cloudflare import CloudflareAPI + + +class DomainManager: + def __init__(self, config: Config): + self.config = config + self.namecheap = NamecheapAPI(config) + self.cloudflare = CloudflareAPI(config) + + def get_domain_status(self, domain: str) -> Dict[str, Any]: + """Get current status of domain across all services""" + status = { + "domain": domain, + "registered": False, + "cloudflare_zone": None, + "nameservers": {"namecheap": [], "cloudflare": []}, + "nameservers_match": False, + } + + # Check if domain is registered + try: + status["registered"] = self.namecheap.is_domain_registered(domain) + except Exception as e: + status["registration_error"] = str(e) + + # Get Cloudflare zone info + try: + zone_info = self.cloudflare.get_zone_info(domain) + if zone_info: + status["cloudflare_zone"] = { + "id": zone_info["id"], + "name": zone_info["name"], + "status": zone_info.get("status", "unknown"), + } + + # Get Cloudflare nameservers + cf_nameservers = self.cloudflare.get_zone_nameservers(zone_info["id"]) + status["nameservers"]["cloudflare"] = cf_nameservers + except Exception as e: + status["cloudflare_error"] = str(e) + + # Get Namecheap nameservers if domain is registered + if status["registered"]: + try: + nc_nameservers = self.namecheap.get_domain_nameservers(domain) + status["nameservers"]["namecheap"] = nc_nameservers + + # Check if nameservers match + cf_ns = set(status["nameservers"]["cloudflare"]) + nc_ns = set(nc_nameservers) + + # Only consider nameservers matching if we have both sets and they match + if len(cf_ns) > 0 and len(nc_ns) > 0: + status["nameservers_match"] = cf_ns == nc_ns + else: + # If we can't retrieve nameservers from either side, assume they don't match + status["nameservers_match"] = False + except Exception as e: + status["namecheap_ns_error"] = str(e) + + return status + + def print_domain_status(self, domain: str): + """Print formatted status of domain""" + status = self.get_domain_status(domain) + + print(f"\n=== Domain Status: {domain} ===") + + # Registration status + if status["registered"]: + print("✓ Domain is registered in Namecheap") + else: + print("✗ Domain is NOT registered in Namecheap") + if "registration_error" in status: + print(f" Error: {status['registration_error']}") + + # Cloudflare zone status + if status["cloudflare_zone"]: + zone = status["cloudflare_zone"] + print( + f"✓ Cloudflare zone exists (ID: {zone['id']}, Status: {zone['status']})" + ) + else: + print("✗ No Cloudflare zone found") + if "cloudflare_error" in status: + print(f" Error: {status['cloudflare_error']}") + + # Nameserver status + nc_ns = status["nameservers"]["namecheap"] + cf_ns = status["nameservers"]["cloudflare"] + + if nc_ns: + print(f"Namecheap nameservers: {', '.join(nc_ns)}") + else: + print("Namecheap nameservers: None") + + if cf_ns: + print(f"Cloudflare nameservers: {', '.join(cf_ns)}") + else: + print("Cloudflare nameservers: None") + + if len(nc_ns) == 0 and len(cf_ns) == 0: + print("⚠ Cannot retrieve nameservers from either service") + elif len(nc_ns) == 0: + print( + "⚠ Cannot retrieve Namecheap nameservers - unable to verify configuration" + ) + elif len(cf_ns) == 0: + print( + "⚠ Cannot retrieve Cloudflare nameservers - unable to verify configuration" + ) + elif status["nameservers_match"]: + print("✓ Nameservers are properly configured") + else: + print("✗ Nameservers do NOT match") + + print("=" * 50) + + def setup_domain( + self, + domain: str, + registrant_info: Optional[Dict[str, str]] = None, + setup_workers: bool = True, + dry_run: bool = False, + force_registration: bool = False, + ) -> Dict[str, Any]: + """ + Complete domain setup workflow (idempotent): + 1. Check domain registration status + 2. Register domain if needed (with user confirmation) + 3. Create Cloudflare zone if needed + 4. Update nameservers in Namecheap if needed + 5. Set up basic DNS records + """ + result = {"domain": domain, "steps_completed": [], "errors": []} + + try: + # Get current status + status = self.get_domain_status(domain) + + # Step 1: Handle domain registration + if not status["registered"]: + if not force_registration: + print( + f"Domain {domain} is not registered. Use --force-registration to register it." + ) + result["errors"].append( + "Domain not registered and force_registration not enabled" + ) + return result + + # Check availability + print(f"Checking availability for {domain}...") + if not self.namecheap.check_domain_availability(domain): + result["errors"].append( + f"Domain {domain} is not available for registration" + ) + return result + + # Get pricing and balance + try: + pricing = self.namecheap.get_domain_pricing(domain) + balance = self.namecheap.get_account_balance() + except Exception as e: + result["errors"].append(f"Failed to get pricing/balance: {str(e)}") + return result + + if balance < pricing["register"]: + result["errors"].append( + f"Insufficient balance. Required: ${pricing['register']:.2f}, " + f"Available: ${balance:.2f}" + ) + return result + + # User confirmation + print("\n" + "=" * 50) + print("DOMAIN REGISTRATION CONFIRMATION") + print("=" * 50) + print(f"Domain: {domain}") + print(f"Registration Price: ${pricing['register']:.2f}") + print(f"Account Balance: ${balance:.2f}") + print(f"Remaining Balance: ${balance - pricing['register']:.2f}") + print("=" * 50) + + print( + f"\nWARNING: This will charge ${pricing['register']:.2f} to your account!" + ) + + if not dry_run: + first_confirm = input( + f"Type 'REGISTER' to proceed with registration of {domain}: " + ).strip() + if first_confirm != "REGISTER": + result["errors"].append("Registration cancelled by user") + return result + + second_confirm = ( + input( + f"Are you absolutely sure you want to register {domain} for ${pricing['register']:.2f}? (yes/no): " + ) + .lower() + .strip() + ) + if second_confirm != "yes": + result["errors"].append("Registration cancelled by user") + return result + + # Register domain + print(f"Registering domain {domain}...") + if not self.namecheap.register_domain( + domain, registrant_info=registrant_info + ): + result["errors"].append(f"Failed to register domain {domain}") + return result + + result["steps_completed"].append("domain_registration") + print(f"✓ Domain {domain} registered successfully") + else: + print(f"DRY RUN: Would register domain {domain}") + result["steps_completed"].append("domain_registration_dry_run") + else: + print(f"✓ Domain {domain} is already registered") + result["steps_completed"].append("domain_already_registered") + + # Step 2: Handle Cloudflare zone creation + if not status["cloudflare_zone"]: + print(f"Creating Cloudflare zone for {domain}...") + try: + if not dry_run: + zone_info = self.cloudflare.add_zone(domain) + zone_id = zone_info["id"] + result["steps_completed"].append("cloudflare_zone_creation") + result["zone_id"] = zone_id + print(f"✓ Cloudflare zone created (ID: {zone_id})") + else: + print(f"DRY RUN: Would create Cloudflare zone for {domain}") + result["steps_completed"].append( + "cloudflare_zone_creation_dry_run" + ) + # For dry run, we can't continue with nameserver setup + return result + except Exception as e: + result["errors"].append( + f"Failed to create Cloudflare zone: {str(e)}" + ) + return result + else: + zone_id = status["cloudflare_zone"]["id"] + result["zone_id"] = zone_id + print(f"✓ Cloudflare zone already exists (ID: {zone_id})") + result["steps_completed"].append("cloudflare_zone_already_exists") + + # Step 3: Get Cloudflare nameservers + print("Getting Cloudflare nameservers...") + try: + nameservers = self.cloudflare.get_zone_nameservers(zone_id) + result["nameservers"] = nameservers + print(f"✓ Cloudflare nameservers: {', '.join(nameservers)}") + except Exception as e: + result["errors"].append( + f"Failed to get Cloudflare nameservers: {str(e)}" + ) + return result + + # Step 4: Update nameservers in Namecheap if needed + current_nc_nameservers = set(status["nameservers"]["namecheap"]) + cloudflare_nameservers = set(nameservers) + + if current_nc_nameservers != cloudflare_nameservers: + if not dry_run: + print("Updating nameservers in Namecheap...") + try: + if not self.namecheap.set_dns_servers(domain, nameservers): + result["errors"].append( + "Failed to update nameservers in Namecheap" + ) + return result + + result["steps_completed"].append("nameserver_update") + print("✓ Nameservers updated in Namecheap") + except Exception as e: + result["errors"].append( + f"Failed to update nameservers in Namecheap: {str(e)}" + ) + return result + else: + print( + f"DRY RUN: Would update nameservers in Namecheap to: {', '.join(nameservers)}" + ) + result["steps_completed"].append("nameserver_update_dry_run") + else: + print("✓ Nameservers already configured correctly") + result["steps_completed"].append("nameservers_already_configured") + + # Step 5: Set up basic DNS records (placeholder for now) + print("Setting up basic DNS records...") + try: + dns_records = [] + result["dns_records"] = dns_records + result["steps_completed"].append("basic_dns_setup") + print("✓ Basic DNS records ready (empty list as requested)") + except Exception as e: + result["errors"].append(f"Failed to set up DNS records: {str(e)}") + return result + + # Step 6: Set up worker subdomain if requested + if setup_workers: + print("Setting up worker subdomain...") + try: + if not dry_run: + worker_record = self.cloudflare.create_worker_subdomain( + zone_id, f"app.{domain}" + ) + result["worker_record"] = worker_record + result["steps_completed"].append("worker_subdomain_setup") + print(f"✓ Worker subdomain app.{domain} created") + else: + print(f"DRY RUN: Would create worker subdomain app.{domain}") + result["steps_completed"].append( + "worker_subdomain_setup_dry_run" + ) + except Exception as e: + result["errors"].append( + f"Failed to set up worker subdomain: {str(e)}" + ) + return result + + result["success"] = True + print(f"\n🎉 Domain {domain} setup completed successfully!") + if "zone_id" in result: + print(f"Zone ID: {result['zone_id']}") + if "nameservers" in result: + print(f"Nameservers: {', '.join(result['nameservers'])}") + + return result + + except Exception as e: + result["errors"].append(f"Unexpected error: {str(e)}") + return result + + def setup_google_analytics_dns(self, domain: str) -> Dict[str, Any]: + """Set up DNS records for Google Analytics""" + try: + zone_info = self.cloudflare.get_zone_info(domain) + if not zone_info: + return {"error": f"Domain {domain} not found in Cloudflare"} + + records = self.cloudflare.setup_google_analytics_dns( + zone_info["id"], domain + ) + return {"success": True, "records": records} + + except Exception as e: + return {"error": str(e)} + + +def main(): + if len(sys.argv) < 2: + print("Usage: regflow [options]") + print("") + print("Options:") + print(" --status Show current status of domain") + print(" --setup Set up domain (idempotent)") + print( + " --dry-run Show what would be done without making changes" + ) + print(" --force-registration Allow domain registration (costs money!)") + print(" --no-workers Skip worker subdomain setup") + print("") + print("Examples:") + print(" regflow example.com --status") + print(" regflow example.com --setup --dry-run") + print(" regflow example.com --setup --force-registration") + sys.exit(1) + + domain = sys.argv[1] + + # Parse command line arguments + show_status = "--status" in sys.argv + setup_domain = "--setup" in sys.argv + dry_run = "--dry-run" in sys.argv + force_registration = "--force-registration" in sys.argv + setup_workers = "--no-workers" not in sys.argv + + # Default to setup if no action specified + if not show_status and not setup_domain: + setup_domain = True + + # Load configuration + config = Config.from_env() + + # Validate required configuration + if not all( + [ + config.namecheap_api_key, + config.namecheap_api_user, + config.namecheap_username, + config.namecheap_client_ip, + config.cloudflare_api_token, + ] + ): + print("Error: Missing required API credentials. Please check your .env file.") + sys.exit(1) + + # Initialize domain manager + manager = DomainManager(config) + + # Handle status command + if show_status: + manager.print_domain_status(domain) + return + + # Handle setup command + if setup_domain: + if dry_run: + print("Running in DRY RUN mode - no actual changes will be made") + if force_registration: + print("FORCE REGISTRATION enabled - will register domain if needed") + + result = manager.setup_domain( + domain, + dry_run=dry_run, + force_registration=force_registration, + setup_workers=setup_workers, + ) + + if result.get("success"): + print(f"\nSuccess! Domain {domain} is ready for use.") + else: + print("\nErrors occurred:") + for error in result.get("errors", []): + print(f" - {error}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/regflow/providers/cloudflare.py b/regflow/providers/cloudflare.py new file mode 100644 index 0000000..edc17a7 --- /dev/null +++ b/regflow/providers/cloudflare.py @@ -0,0 +1,170 @@ +import requests +from typing import Dict, Any, List, Optional +from ..config import Config + + +class CloudflareAPI: + def __init__(self, config: Config): + self.config = config + self.base_url = "https://api.cloudflare.com/client/v4" + self.headers = { + "Authorization": f"Bearer {config.cloudflare_api_token}", + "Content-Type": "application/json", + } + + def _make_request( + self, method: str, endpoint: str, data: Optional[Dict] = None + ) -> Dict[str, Any]: + """Make a request to Cloudflare API""" + url = f"{self.base_url}{endpoint}" + + response = requests.request( + method, url, headers=self.headers, json=data, timeout=30 + ) + + try: + response.raise_for_status() + except requests.exceptions.HTTPError: + # Try to get the error details from the response + try: + error_details = response.json() + errors = error_details.get("errors", []) + if errors: + error_msg = ", ".join( + [error.get("message", "Unknown error") for error in errors] + ) + raise Exception(f"Cloudflare API Error: {error_msg}") + else: + raise Exception( + f"Cloudflare API Error: {response.status_code} - {response.text}" + ) + except ValueError: + raise Exception( + f"Cloudflare API Error: {response.status_code} - {response.text}" + ) + + result = response.json() + + if not result.get("success", False): + errors = result.get("errors", []) + if errors: + error_msg = ", ".join( + [error.get("message", "Unknown error") for error in errors] + ) + raise Exception(f"Cloudflare API Error: {error_msg}") + else: + raise Exception( + "Cloudflare API Error: Request failed but no error details provided" + ) + + return result + + def add_zone(self, domain: str) -> Dict[str, Any]: + """Add a new zone (domain) to Cloudflare""" + data = {"name": domain, "type": "full"} + + result = self._make_request("POST", "/zones", data) + return result["result"] + + def get_zone_info(self, domain: str) -> Optional[Dict[str, Any]]: + """Get zone information for a domain""" + result = self._make_request("GET", f"/zones?name={domain}") + + zones = result.get("result", []) + if zones: + return zones[0] + + return None + + def get_zone_nameservers(self, zone_id: str) -> List[str]: + """Get nameservers for a zone""" + result = self._make_request("GET", f"/zones/{zone_id}") + + zone = result.get("result", {}) + return zone.get("name_servers", []) + + def create_dns_record( + self, + zone_id: str, + record_type: str, + name: str, + content: str, + ttl: int = 300, + proxied: bool = False, + ) -> Dict[str, Any]: + """Create a DNS record""" + data = {"type": record_type, "name": name, "content": content, "ttl": ttl} + + if record_type in ["A", "AAAA", "CNAME"]: + data["proxied"] = proxied + + result = self._make_request("POST", f"/zones/{zone_id}/dns_records", data) + return result["result"] + + def create_worker_subdomain(self, zone_id: str, subdomain: str) -> Dict[str, Any]: + """Create a worker route for a subdomain""" + # Create A record pointing to dummy IP (will be overridden by worker) + return self.create_dns_record( + zone_id=zone_id, + record_type="A", + name=subdomain, + content="192.0.2.1", # Dummy IP + proxied=True, + ) + + def setup_google_analytics_dns( + self, zone_id: str, domain: str + ) -> List[Dict[str, Any]]: + """Set up DNS records for Google Analytics""" + records = [] + + # Google Analytics doesn't typically require specific DNS records + # But we can add common verification records if needed + # This is a placeholder for future GA4 requirements + + return records + + def setup_basic_dns_records( + self, zone_id: str, domain: str + ) -> List[Dict[str, Any]]: + """Set up basic DNS records for a domain""" + records = [] + + # Root domain A record (placeholder) + records.append( + self.create_dns_record( + zone_id=zone_id, + record_type="A", + name=domain, + content="192.0.2.1", + proxied=True, + ) + ) + + # WWW CNAME record + records.append( + self.create_dns_record( + zone_id=zone_id, + record_type="CNAME", + name=f"www.{domain}", + content=domain, + proxied=True, + ) + ) + + return records + + def list_zones(self) -> List[Dict[str, Any]]: + """List all zones in the account""" + result = self._make_request("GET", "/zones") + return result.get("result", []) + + def get_zone_dns_records(self, zone_id: str) -> List[Dict[str, Any]]: + """Get all DNS records for a zone""" + result = self._make_request("GET", f"/zones/{zone_id}/dns_records") + return result.get("result", []) + + def zone_exists(self, domain: str) -> bool: + """Check if domain exists as a zone in Cloudflare""" + zone_info = self.get_zone_info(domain) + return zone_info is not None diff --git a/regflow/providers/cloudflare_api.py b/regflow/providers/cloudflare_api.py deleted file mode 100644 index edc17a7..0000000 --- a/regflow/providers/cloudflare_api.py +++ /dev/null @@ -1,170 +0,0 @@ -import requests -from typing import Dict, Any, List, Optional -from ..config import Config - - -class CloudflareAPI: - def __init__(self, config: Config): - self.config = config - self.base_url = "https://api.cloudflare.com/client/v4" - self.headers = { - "Authorization": f"Bearer {config.cloudflare_api_token}", - "Content-Type": "application/json", - } - - def _make_request( - self, method: str, endpoint: str, data: Optional[Dict] = None - ) -> Dict[str, Any]: - """Make a request to Cloudflare API""" - url = f"{self.base_url}{endpoint}" - - response = requests.request( - method, url, headers=self.headers, json=data, timeout=30 - ) - - try: - response.raise_for_status() - except requests.exceptions.HTTPError: - # Try to get the error details from the response - try: - error_details = response.json() - errors = error_details.get("errors", []) - if errors: - error_msg = ", ".join( - [error.get("message", "Unknown error") for error in errors] - ) - raise Exception(f"Cloudflare API Error: {error_msg}") - else: - raise Exception( - f"Cloudflare API Error: {response.status_code} - {response.text}" - ) - except ValueError: - raise Exception( - f"Cloudflare API Error: {response.status_code} - {response.text}" - ) - - result = response.json() - - if not result.get("success", False): - errors = result.get("errors", []) - if errors: - error_msg = ", ".join( - [error.get("message", "Unknown error") for error in errors] - ) - raise Exception(f"Cloudflare API Error: {error_msg}") - else: - raise Exception( - "Cloudflare API Error: Request failed but no error details provided" - ) - - return result - - def add_zone(self, domain: str) -> Dict[str, Any]: - """Add a new zone (domain) to Cloudflare""" - data = {"name": domain, "type": "full"} - - result = self._make_request("POST", "/zones", data) - return result["result"] - - def get_zone_info(self, domain: str) -> Optional[Dict[str, Any]]: - """Get zone information for a domain""" - result = self._make_request("GET", f"/zones?name={domain}") - - zones = result.get("result", []) - if zones: - return zones[0] - - return None - - def get_zone_nameservers(self, zone_id: str) -> List[str]: - """Get nameservers for a zone""" - result = self._make_request("GET", f"/zones/{zone_id}") - - zone = result.get("result", {}) - return zone.get("name_servers", []) - - def create_dns_record( - self, - zone_id: str, - record_type: str, - name: str, - content: str, - ttl: int = 300, - proxied: bool = False, - ) -> Dict[str, Any]: - """Create a DNS record""" - data = {"type": record_type, "name": name, "content": content, "ttl": ttl} - - if record_type in ["A", "AAAA", "CNAME"]: - data["proxied"] = proxied - - result = self._make_request("POST", f"/zones/{zone_id}/dns_records", data) - return result["result"] - - def create_worker_subdomain(self, zone_id: str, subdomain: str) -> Dict[str, Any]: - """Create a worker route for a subdomain""" - # Create A record pointing to dummy IP (will be overridden by worker) - return self.create_dns_record( - zone_id=zone_id, - record_type="A", - name=subdomain, - content="192.0.2.1", # Dummy IP - proxied=True, - ) - - def setup_google_analytics_dns( - self, zone_id: str, domain: str - ) -> List[Dict[str, Any]]: - """Set up DNS records for Google Analytics""" - records = [] - - # Google Analytics doesn't typically require specific DNS records - # But we can add common verification records if needed - # This is a placeholder for future GA4 requirements - - return records - - def setup_basic_dns_records( - self, zone_id: str, domain: str - ) -> List[Dict[str, Any]]: - """Set up basic DNS records for a domain""" - records = [] - - # Root domain A record (placeholder) - records.append( - self.create_dns_record( - zone_id=zone_id, - record_type="A", - name=domain, - content="192.0.2.1", - proxied=True, - ) - ) - - # WWW CNAME record - records.append( - self.create_dns_record( - zone_id=zone_id, - record_type="CNAME", - name=f"www.{domain}", - content=domain, - proxied=True, - ) - ) - - return records - - def list_zones(self) -> List[Dict[str, Any]]: - """List all zones in the account""" - result = self._make_request("GET", "/zones") - return result.get("result", []) - - def get_zone_dns_records(self, zone_id: str) -> List[Dict[str, Any]]: - """Get all DNS records for a zone""" - result = self._make_request("GET", f"/zones/{zone_id}/dns_records") - return result.get("result", []) - - def zone_exists(self, domain: str) -> bool: - """Check if domain exists as a zone in Cloudflare""" - zone_info = self.get_zone_info(domain) - return zone_info is not None diff --git a/regflow/providers/namecheap.py b/regflow/providers/namecheap.py new file mode 100644 index 0000000..f91acc6 --- /dev/null +++ b/regflow/providers/namecheap.py @@ -0,0 +1,241 @@ +import requests +import xml.etree.ElementTree as ET +from typing import Dict, Any, Optional +from ..config import Config + + +class NamecheapAPI: + def __init__(self, config: Config): + self.config = config + self.base_url = "https://api.namecheap.com/xml.response" + + def _make_request(self, command: str, params: Dict[str, Any]) -> ET.Element: + """Make a request to Namecheap API""" + default_params = { + "ApiUser": self.config.namecheap_api_user, + "ApiKey": self.config.namecheap_api_key, + "UserName": self.config.namecheap_username, + "ClientIp": self.config.namecheap_client_ip, + "Command": command, + } + + all_params = {**default_params, **params} + + try: + response = requests.get(self.base_url, params=all_params, timeout=60) + response.raise_for_status() + except requests.exceptions.Timeout: + raise Exception(f"API request timed out for command: {command}") + except requests.exceptions.RequestException as e: + raise Exception(f"API request failed for command {command}: {str(e)}") + + root = ET.fromstring(response.text) + + # Check for API errors + if root.get("Status") == "ERROR": + ns = {"ns": "http://api.namecheap.com/xml.response"} + errors = root.find(".//ns:Errors", ns) + if errors is not None: + error_elem = errors.find(".//ns:Error", ns) + if error_elem is not None: + error_msg = error_elem.text + raise Exception(f"Namecheap API Error: {error_msg}") + raise Exception("Namecheap API returned an error") + + return root + + def check_domain_availability(self, domain: str) -> bool: + """Check if domain is available for registration""" + params = {"DomainList": domain} + + root = self._make_request("namecheap.domains.check", params) + + # Parse the response + ns = {"ns": "http://api.namecheap.com/xml.response"} + domain_check = root.find(".//ns:DomainCheckResult", ns) + + if domain_check is None: + raise Exception(f"Could not find domain check result for {domain}") + + return domain_check.get("Available") == "true" + + def get_domain_pricing(self, domain: str) -> Dict[str, float]: + """Get pricing information for a domain""" + tld = domain.split(".")[-1].upper() + + params = { + "ProductType": "DOMAIN", + "ProductCategory": "DOMAINS", + "ActionName": "REGISTER", + "ProductName": tld, + } + + root = self._make_request("namecheap.users.getPricing", params) + + # Find pricing for the TLD + ns = {"ns": "http://api.namecheap.com/xml.response"} + + # Look for the specific TLD in the response + for product in root.findall(".//ns:Product", ns): + product_name = product.get("Name") + if product_name and product_name.lower() == tld.lower(): + for price in product.findall(".//ns:Price", ns): + duration = price.get("Duration") + if duration == "1": + price_val = float(price.get("Price", 0)) + + # Get renewal price (might be in a separate call or same structure) + renew_val = ( + price_val # Use same price for renewal if not specified + ) + + return {"register": price_val, "renew": renew_val} + + raise Exception(f"No pricing information found for .{tld} domains") + + def get_account_balance(self) -> float: + """Get current account balance""" + root = self._make_request("namecheap.users.getBalances", {}) + + ns = {"ns": "http://api.namecheap.com/xml.response"} + balance_result = root.find(".//ns:UserGetBalancesResult", ns) + + if balance_result is not None: + available_balance = balance_result.get("AvailableBalance") + if available_balance: + return float(available_balance) + + raise Exception("Could not retrieve account balance") + + def register_domain( + self, + domain: str, + years: int = 1, + registrant_info: Optional[Dict[str, str]] = None, + ) -> bool: + """Register a domain""" + if not registrant_info: + registrant_info = { + "FirstName": "John", + "LastName": "Doe", + "Address1": "123 Main St", + "City": "Anytown", + "StateProvince": "NY", + "PostalCode": "12345", + "Country": "US", + "Phone": "+1.5551234567", + "EmailAddress": "john.doe@example.com", + } + + params = { + "DomainName": domain, + "Years": str(years), + **{f"Registrant{key}": value for key, value in registrant_info.items()}, + **{f"Tech{key}": value for key, value in registrant_info.items()}, + **{f"Admin{key}": value for key, value in registrant_info.items()}, + **{f"AuxBilling{key}": value for key, value in registrant_info.items()}, + } + + root = self._make_request("namecheap.domains.create", params) + + # Check if registration was successful + ns = {"ns": "http://api.namecheap.com/xml.response"} + domain_create = root.find(".//ns:DomainCreateResult", ns) + if domain_create is None: + raise Exception(f"Could not find domain creation result for {domain}") + + return domain_create.get("Registered") == "true" + + def set_dns_servers(self, domain: str, nameservers: list) -> bool: + """Set DNS servers for a domain""" + sld, tld = domain.split(".", 1) + + params = {"SLD": sld, "TLD": tld, "Nameservers": ",".join(nameservers)} + + root = self._make_request("namecheap.domains.dns.setCustom", params) + + # Check if DNS update was successful + ns = {"ns": "http://api.namecheap.com/xml.response"} + dns_result = root.find(".//ns:DomainDNSSetCustomResult", ns) + if dns_result is None: + raise Exception(f"Could not find DNS update result for {domain}") + + return dns_result.get("Updated") == "true" + + def is_domain_registered(self, domain: str) -> bool: + """Check if domain is registered in user's account""" + try: + # Get list of all domains in the account + root = self._make_request("namecheap.domains.getList", {}) + + ns = {"ns": "http://api.namecheap.com/xml.response"} + + # Look for the domain in the list + for domain_elem in root.findall(".//ns:Domain", ns): + domain_name = domain_elem.get("Name") + if domain_name and domain_name.lower() == domain.lower(): + return True + + return False + except Exception: + # If API call fails, assume not registered + return False + + def get_domain_nameservers(self, domain: str) -> list: + """Get current nameservers for a domain""" + try: + params = {"DomainName": domain} + + # Use getInfo to get domain details including nameservers + root = self._make_request("namecheap.domains.getInfo", params) + + ns = {"ns": "http://api.namecheap.com/xml.response"} + nameservers = [] + + # Look for nameservers in the domain info response + # The structure might be different, let's check multiple possible locations + + # Try to find nameservers in DnsDetails + dns_details = root.find(".//ns:DnsDetails", ns) + if dns_details is not None: + # Look for nameserver elements + for ns_elem in dns_details.findall(".//ns:Nameserver", ns): + if ns_elem.text: + nameservers.append(ns_elem.text) + + # If no nameservers found in DnsDetails, try other locations + if not nameservers: + # Try to find in different structure + for ns_elem in root.findall(".//ns:Nameserver", ns): + if ns_elem.text: + nameservers.append(ns_elem.text) + + # If still no nameservers, check for attributes in the domain result + if not nameservers: + domain_result = root.find(".//ns:DomainGetInfoResult", ns) + if domain_result is not None: + # Check for nameserver attributes (common in Namecheap responses) + for i in range(1, 5): # Check for ns1, ns2, ns3, ns4 + ns_attr = domain_result.get(f"Nameserver{i}") + if ns_attr: + nameservers.append(ns_attr) + + # Also check for DNS details in attributes + dns_type = domain_result.get("DnsProviderType") + if dns_type == "CUSTOM": + # For custom DNS, nameservers should be in the response + # Try looking for nameservers in different attribute names + for attr_name in domain_result.attrib: + if ( + "nameserver" in attr_name.lower() + or "ns" in attr_name.lower() + ): + ns_value = domain_result.get(attr_name) + if ns_value and ns_value not in nameservers: + nameservers.append(ns_value) + + return nameservers + except Exception: + # For debugging, you might want to print the exception + # print(f"Error getting nameservers for {domain}: {e}") + return [] diff --git a/regflow/providers/namecheap_api.py b/regflow/providers/namecheap_api.py deleted file mode 100644 index 20cdfa9..0000000 --- a/regflow/providers/namecheap_api.py +++ /dev/null @@ -1,238 +0,0 @@ -import requests -import xml.etree.ElementTree as ET -from typing import Dict, Any, Optional -from ..config import Config - - -class NamecheapAPI: - def __init__(self, config: Config): - self.config = config - self.base_url = "https://api.namecheap.com/xml.response" - - def _make_request(self, command: str, params: Dict[str, Any]) -> ET.Element: - """Make a request to Namecheap API""" - default_params = { - "ApiUser": self.config.namecheap_api_user, - "ApiKey": self.config.namecheap_api_key, - "UserName": self.config.namecheap_username, - "ClientIp": self.config.namecheap_client_ip, - "Command": command, - } - - all_params = {**default_params, **params} - - try: - response = requests.get(self.base_url, params=all_params, timeout=60) - response.raise_for_status() - except requests.exceptions.Timeout: - raise Exception(f"API request timed out for command: {command}") - except requests.exceptions.RequestException as e: - raise Exception(f"API request failed for command {command}: {str(e)}") - - root = ET.fromstring(response.text) - - # Check for API errors - if root.get("Status") == "ERROR": - ns = {"ns": "http://api.namecheap.com/xml.response"} - errors = root.find(".//ns:Errors", ns) - if errors is not None: - error_elem = errors.find(".//ns:Error", ns) - if error_elem is not None: - error_msg = error_elem.text - raise Exception(f"Namecheap API Error: {error_msg}") - raise Exception("Namecheap API returned an error") - - return root - - def check_domain_availability(self, domain: str) -> bool: - """Check if domain is available for registration""" - params = {"DomainList": domain} - - root = self._make_request("namecheap.domains.check", params) - - # Parse the response - ns = {"ns": "http://api.namecheap.com/xml.response"} - domain_check = root.find(".//ns:DomainCheckResult", ns) - - if domain_check is None: - raise Exception(f"Could not find domain check result for {domain}") - - return domain_check.get("Available") == "true" - - def get_domain_pricing(self, domain: str) -> Dict[str, float]: - """Get pricing information for a domain""" - tld = domain.split(".")[-1].upper() - - params = { - "ProductType": "DOMAIN", - "ProductCategory": "DOMAINS", - "ActionName": "REGISTER", - "ProductName": tld, - } - - root = self._make_request("namecheap.users.getPricing", params) - - # Find pricing for the TLD - ns = {"ns": "http://api.namecheap.com/xml.response"} - - # Look for the specific TLD in the response - for product in root.findall(".//ns:Product", ns): - product_name = product.get("Name") - if product_name and product_name.lower() == tld.lower(): - for price in product.findall(".//ns:Price", ns): - duration = price.get("Duration") - if duration == "1": - price_val = float(price.get("Price", 0)) - - # Get renewal price (might be in a separate call or same structure) - renew_val = ( - price_val # Use same price for renewal if not specified - ) - - return {"register": price_val, "renew": renew_val} - - raise Exception(f"No pricing information found for .{tld} domains") - - def get_account_balance(self) -> float: - """Get current account balance""" - root = self._make_request("namecheap.users.getBalances", {}) - - ns = {"ns": "http://api.namecheap.com/xml.response"} - balance_result = root.find(".//ns:UserGetBalancesResult", ns) - - if balance_result is not None: - available_balance = balance_result.get("AvailableBalance") - if available_balance: - return float(available_balance) - - raise Exception("Could not retrieve account balance") - - def register_domain( - self, - domain: str, - years: int = 1, - registrant_info: Optional[Dict[str, str]] = None, - ) -> bool: - """Register a domain""" - if not registrant_info: - registrant_info = { - "FirstName": "John", - "LastName": "Doe", - "Address1": "123 Main St", - "City": "Anytown", - "StateProvince": "NY", - "PostalCode": "12345", - "Country": "US", - "Phone": "+1.5551234567", - "EmailAddress": "john.doe@example.com", - } - - params = { - "DomainName": domain, - "Years": str(years), - **{f"Registrant{key}": value for key, value in registrant_info.items()}, - **{f"Tech{key}": value for key, value in registrant_info.items()}, - **{f"Admin{key}": value for key, value in registrant_info.items()}, - **{f"AuxBilling{key}": value for key, value in registrant_info.items()}, - } - - root = self._make_request("namecheap.domains.create", params) - - # Check if registration was successful - ns = {"ns": "http://api.namecheap.com/xml.response"} - domain_create = root.find(".//ns:DomainCreateResult", ns) - if domain_create is None: - raise Exception(f"Could not find domain creation result for {domain}") - - return domain_create.get("Registered") == "true" - - def set_dns_servers(self, domain: str, nameservers: list) -> bool: - """Set DNS servers for a domain""" - sld, tld = domain.split(".", 1) - - params = {"SLD": sld, "TLD": tld, "Nameservers": ",".join(nameservers)} - - root = self._make_request("namecheap.domains.dns.setCustom", params) - - # Check if DNS update was successful - ns = {"ns": "http://api.namecheap.com/xml.response"} - dns_result = root.find(".//ns:DomainDNSSetCustomResult", ns) - if dns_result is None: - raise Exception(f"Could not find DNS update result for {domain}") - - return dns_result.get("Updated") == "true" - - def is_domain_registered(self, domain: str) -> bool: - """Check if domain is registered in user's account""" - try: - # Get list of all domains in the account - root = self._make_request("namecheap.domains.getList", {}) - - ns = {"ns": "http://api.namecheap.com/xml.response"} - - # Look for the domain in the list - for domain_elem in root.findall(".//ns:Domain", ns): - domain_name = domain_elem.get("Name") - if domain_name and domain_name.lower() == domain.lower(): - return True - - return False - except Exception: - # If API call fails, assume not registered - return False - - def get_domain_nameservers(self, domain: str) -> list: - """Get current nameservers for a domain""" - try: - params = {"DomainName": domain} - - # Use getInfo to get domain details including nameservers - root = self._make_request("namecheap.domains.getInfo", params) - - ns = {"ns": "http://api.namecheap.com/xml.response"} - nameservers = [] - - # Look for nameservers in the domain info response - # The structure might be different, let's check multiple possible locations - - # Try to find nameservers in DnsDetails - dns_details = root.find(".//ns:DnsDetails", ns) - if dns_details is not None: - # Look for nameserver elements - for ns_elem in dns_details.findall(".//ns:Nameserver", ns): - if ns_elem.text: - nameservers.append(ns_elem.text) - - # If no nameservers found in DnsDetails, try other locations - if not nameservers: - # Try to find in different structure - for ns_elem in root.findall(".//ns:Nameserver", ns): - if ns_elem.text: - nameservers.append(ns_elem.text) - - # If still no nameservers, check for attributes in the domain result - if not nameservers: - domain_result = root.find(".//ns:DomainGetInfoResult", ns) - if domain_result is not None: - # Check for nameserver attributes (common in Namecheap responses) - for i in range(1, 5): # Check for ns1, ns2, ns3, ns4 - ns_attr = domain_result.get(f"Nameserver{i}") - if ns_attr: - nameservers.append(ns_attr) - - # Also check for DNS details in attributes - dns_type = domain_result.get("DnsProviderType") - if dns_type == "CUSTOM": - # For custom DNS, nameservers should be in the response - # Try looking for nameservers in different attribute names - for attr_name in domain_result.attrib: - if "nameserver" in attr_name.lower() or "ns" in attr_name.lower(): - ns_value = domain_result.get(attr_name) - if ns_value and ns_value not in nameservers: - nameservers.append(ns_value) - - return nameservers - except Exception as e: - # For debugging, you might want to print the exception - # print(f"Error getting nameservers for {domain}: {e}") - return [] diff --git a/regflow/tests/test_integration.py b/regflow/tests/test_integration.py index 4b44471..8c63445 100644 --- a/regflow/tests/test_integration.py +++ b/regflow/tests/test_integration.py @@ -1,7 +1,8 @@ import pytest from ..config import Config -from ..providers.namecheap_api import NamecheapAPI -from ..providers.cloudflare_api import CloudflareAPI +from ..providers.namecheap import NamecheapAPI +from ..providers.cloudflare import CloudflareAPI +from ..domains import DomainManager @pytest.fixture @@ -22,6 +23,12 @@ def cloudflare_api(config): return CloudflareAPI(config) +@pytest.fixture +def domain_manager(config): + """Create DomainManager instance""" + return DomainManager(config) + + def test_namecheap_credentials_loaded(config): """Test that Namecheap credentials are loaded""" assert config.namecheap_api_key, "Namecheap API key not found" @@ -127,3 +134,187 @@ def test_cloudflare_dns_records(cloudflare_api): assert "type" in record, "Record should have 'type' field" assert "name" in record, "Record should have 'name' field" assert "content" in record, "Record should have 'content' field" + + +# New tests for domain registration and nameserver functionality + + +def test_namecheap_domain_registration_check_existing(namecheap_api): + """Test domain registration check for existing domain""" + # Test with stackvital.com which we know exists + is_registered = namecheap_api.is_domain_registered("stackvital.com") + assert is_registered, "stackvital.com should be registered" + + +def test_namecheap_domain_registration_check_nonexistent(namecheap_api): + """Test domain registration check for non-existent domain""" + # Test with a domain that definitely doesn't exist in the account + is_registered = namecheap_api.is_domain_registered("nonexistent-domain-12345.com") + assert not is_registered, "Non-existent domain should not be registered" + + +def test_namecheap_get_nameservers_existing_domain(namecheap_api): + """Test nameserver retrieval for existing domain""" + nameservers = namecheap_api.get_domain_nameservers("stackvital.com") + + assert isinstance(nameservers, list), "Nameservers should be a list" + assert len(nameservers) > 0, "Should have at least one nameserver" + + # Check that all nameservers are strings + for ns in nameservers: + assert isinstance(ns, str), "Each nameserver should be a string" + assert "." in ns, "Nameserver should be a valid domain name" + + +def test_namecheap_get_nameservers_nonexistent_domain(namecheap_api): + """Test nameserver retrieval for non-existent domain""" + nameservers = namecheap_api.get_domain_nameservers("nonexistent-domain-12345.com") + + # Should return empty list for non-existent domain + assert isinstance(nameservers, list), ( + "Should return a list even for non-existent domain" + ) + assert len(nameservers) == 0, "Should return empty list for non-existent domain" + + +def test_cloudflare_zone_exists_check(cloudflare_api): + """Test Cloudflare zone existence check""" + # Test with stackvital.com which should exist in Cloudflare + zone_exists = cloudflare_api.zone_exists("stackvital.com") + assert zone_exists, "stackvital.com should exist in Cloudflare" + + # Test with a domain that doesn't exist + zone_exists = cloudflare_api.zone_exists("nonexistent-domain-12345.com") + assert not zone_exists, "Non-existent domain should not exist in Cloudflare" + + +def test_domain_manager_get_status_existing_domain(domain_manager): + """Test domain status retrieval for existing domain""" + status = domain_manager.get_domain_status("stackvital.com") + + # Verify status structure + assert "domain" in status, "Status should contain domain field" + assert "registered" in status, "Status should contain registered field" + assert "cloudflare_zone" in status, "Status should contain cloudflare_zone field" + assert "nameservers" in status, "Status should contain nameservers field" + assert "nameservers_match" in status, ( + "Status should contain nameservers_match field" + ) + + # Verify domain name + assert status["domain"] == "stackvital.com", "Domain name should match" + + # Verify registration status + assert status["registered"], "stackvital.com should be registered" + + # Verify Cloudflare zone exists + assert status["cloudflare_zone"] is not None, "Cloudflare zone should exist" + assert "id" in status["cloudflare_zone"], "Zone should have ID" + assert "name" in status["cloudflare_zone"], "Zone should have name" + assert "status" in status["cloudflare_zone"], "Zone should have status" + + # Verify nameservers structure + assert "namecheap" in status["nameservers"], "Should have namecheap nameservers" + assert "cloudflare" in status["nameservers"], "Should have cloudflare nameservers" + assert isinstance(status["nameservers"]["namecheap"], list), ( + "Namecheap nameservers should be a list" + ) + assert isinstance(status["nameservers"]["cloudflare"], list), ( + "Cloudflare nameservers should be a list" + ) + + # Verify nameservers are populated + assert len(status["nameservers"]["namecheap"]) > 0, ( + "Should have Namecheap nameservers" + ) + assert len(status["nameservers"]["cloudflare"]) > 0, ( + "Should have Cloudflare nameservers" + ) + + +def test_domain_manager_get_status_nonexistent_domain(domain_manager): + """Test domain status retrieval for non-existent domain""" + status = domain_manager.get_domain_status("nonexistent-domain-12345.com") + + # Verify status structure + assert "domain" in status, "Status should contain domain field" + assert "registered" in status, "Status should contain registered field" + assert "cloudflare_zone" in status, "Status should contain cloudflare_zone field" + assert "nameservers" in status, "Status should contain nameservers field" + assert "nameservers_match" in status, ( + "Status should contain nameservers_match field" + ) + + # Verify domain name + assert status["domain"] == "nonexistent-domain-12345.com", ( + "Domain name should match" + ) + + # Verify registration status + assert not status["registered"], "Non-existent domain should not be registered" + + # Verify Cloudflare zone doesn't exist + assert status["cloudflare_zone"] is None, "Cloudflare zone should not exist" + + # Verify nameservers are empty + assert len(status["nameservers"]["namecheap"]) == 0, ( + "Should have no Namecheap nameservers" + ) + assert len(status["nameservers"]["cloudflare"]) == 0, ( + "Should have no Cloudflare nameservers" + ) + + # Verify nameservers don't match + assert not status["nameservers_match"], ( + "Nameservers should not match for non-existent domain" + ) + + +def test_nameserver_matching_logic(domain_manager): + """Test nameserver matching logic with stackvital.com""" + status = domain_manager.get_domain_status("stackvital.com") + + # Get the nameservers + nc_nameservers = status["nameservers"]["namecheap"] + cf_nameservers = status["nameservers"]["cloudflare"] + + # Both should have nameservers + assert len(nc_nameservers) > 0, "Should have Namecheap nameservers" + assert len(cf_nameservers) > 0, "Should have Cloudflare nameservers" + + # Check if they match (they should for stackvital.com) + nc_set = set(nc_nameservers) + cf_set = set(cf_nameservers) + + # The nameservers should match + assert nc_set == cf_set, "Nameservers should match for properly configured domain" + assert status["nameservers_match"], "Status should indicate nameservers match" + + +def test_domain_manager_print_status_no_errors(domain_manager, capsys): + """Test that print_domain_status runs without errors""" + # This should not raise any exceptions + domain_manager.print_domain_status("stackvital.com") + + # Verify some output was produced + captured = capsys.readouterr() + assert "Domain Status: stackvital.com" in captured.out, ( + "Should print domain status header" + ) + assert "registered" in captured.out.lower(), "Should mention registration status" + assert "nameservers" in captured.out.lower(), "Should mention nameservers" + + +def test_setup_domain_dry_run_existing_domain(domain_manager): + """Test setup_domain with dry run for existing domain""" + result = domain_manager.setup_domain( + "stackvital.com", dry_run=True, force_registration=False + ) + + # Should succeed since domain is already registered + assert "success" in result, "Result should contain success field" + assert result.get("success"), "Should succeed for existing domain" + assert "errors" in result, "Result should contain errors field" + assert len(result["errors"]) == 0, "Should have no errors" + assert "steps_completed" in result, "Result should contain steps_completed field" + assert len(result["steps_completed"]) > 0, "Should have completed some steps" -- cgit v1.3.1