From e6eb7ef34a06ece633c5dc961572694d49a46548 Mon Sep 17 00:00:00 2001 From: Yuval Adam <_@yuv.al> Date: Fri, 18 Jul 2025 20:17:48 +0200 Subject: restructure --- cloudflare_api.py | 143 ------------------- config.py | 22 --- domain_manager.py | 241 ------------------------------- namecheap_api.py | 165 --------------------- pyproject.toml | 4 +- regflow/__init__.py | 3 + regflow/config.py | 23 +++ regflow/domain_manager.py | 276 ++++++++++++++++++++++++++++++++++++ regflow/providers/__init__.py | 1 + regflow/providers/cloudflare_api.py | 165 +++++++++++++++++++++ regflow/providers/namecheap_api.py | 163 +++++++++++++++++++++ regflow/tests/__init__.py | 1 + regflow/tests/test_integration.py | 126 ++++++++++++++++ tests/__init__.py | 1 - tests/test_integration.py | 126 ---------------- 15 files changed, 760 insertions(+), 700 deletions(-) delete mode 100644 cloudflare_api.py delete mode 100644 config.py delete mode 100755 domain_manager.py delete mode 100644 namecheap_api.py create mode 100644 regflow/__init__.py create mode 100644 regflow/config.py create mode 100755 regflow/domain_manager.py create mode 100644 regflow/providers/__init__.py create mode 100644 regflow/providers/cloudflare_api.py create mode 100644 regflow/providers/namecheap_api.py create mode 100644 regflow/tests/__init__.py create mode 100644 regflow/tests/test_integration.py delete mode 100644 tests/__init__.py delete mode 100644 tests/test_integration.py diff --git a/cloudflare_api.py b/cloudflare_api.py deleted file mode 100644 index 16fd926..0000000 --- a/cloudflare_api.py +++ /dev/null @@ -1,143 +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 as e: - # 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(f"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', []) \ No newline at end of file diff --git a/config.py b/config.py deleted file mode 100644 index 488c52d..0000000 --- a/config.py +++ /dev/null @@ -1,22 +0,0 @@ -import os -from dotenv import load_dotenv -from pydantic import BaseModel - -load_dotenv() - -class Config(BaseModel): - namecheap_api_user: str - namecheap_api_key: str - namecheap_username: str - namecheap_client_ip: str - cloudflare_api_token: str - - @classmethod - def from_env(cls): - return cls( - namecheap_api_user=os.getenv("NAMECHEAP_API_USER", ""), - namecheap_api_key=os.getenv("NAMECHEAP_API_KEY", ""), - namecheap_username=os.getenv("NAMECHEAP_USERNAME", ""), - namecheap_client_ip=os.getenv("NAMECHEAP_CLIENT_IP", ""), - cloudflare_api_token=os.getenv("CLOUDFLARE_API_TOKEN", "") - ) \ No newline at end of file diff --git a/domain_manager.py b/domain_manager.py deleted file mode 100755 index 707ce4f..0000000 --- a/domain_manager.py +++ /dev/null @@ -1,241 +0,0 @@ -#!/usr/bin/env python3 - -import sys -from typing import Dict, Any, Optional -from config import Config -from namecheap_api import NamecheapAPI -from cloudflare_api import CloudflareAPI - -class DomainManager: - def __init__(self, config: Config): - self.config = config - self.namecheap = NamecheapAPI(config) - self.cloudflare = CloudflareAPI(config) - - def register_and_setup_domain(self, domain: str, - registrant_info: Optional[Dict[str, str]] = None, - setup_workers: bool = True, - dry_run: bool = False, - skip_registration: bool = False) -> Dict[str, Any]: - """ - Complete domain registration and setup workflow: - 1. Check domain availability - 2. Verify pricing and account balance - 3. Register domain (with user confirmation) - 4. Add domain to Cloudflare - 5. Update nameservers in Namecheap - 6. Set up basic DNS records - """ - result = { - 'domain': domain, - 'steps_completed': [], - 'errors': [] - } - - try: - if not skip_registration: - # Step 1: Check domain availability - print(f"Checking availability for {domain}...") - if not self.namecheap.check_domain_availability(domain): - result['errors'].append(f"Domain {domain} is not available") - return result - - result['steps_completed'].append('availability_check') - print(f"✓ Domain {domain} is available") - - # Step 2: Check pricing and balance - print("Checking pricing and account balance...") - - try: - pricing = self.namecheap.get_domain_pricing(domain) - except Exception as e: - result['errors'].append(f"Failed to get domain pricing: {str(e)}") - return result - - try: - balance = self.namecheap.get_account_balance() - except Exception as e: - result['errors'].append(f"Failed to get account 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 - - result['steps_completed'].append('pricing_check') - print(f"✓ Pricing: ${pricing['register']:.2f}, Balance: ${balance:.2f}") - - # Step 3: User confirmation with multiple safeguards - print(f"\n" + "="*50) - print(f"DOMAIN REGISTRATION CONFIRMATION") - print(f"="*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(f"="*50) - - # Multiple confirmation steps to prevent accidents - print(f"\nWARNING: This will charge ${pricing['register']:.2f} to your account!") - 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 - - # Step 4: Register domain - if dry_run: - print(f"DRY RUN: Would register domain {domain} (skipping actual registration)") - result['steps_completed'].append('domain_registration_dry_run') - else: - 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"✓ Skipping domain registration for {domain} (assuming already registered)") - result['steps_completed'].append('domain_registration_skipped') - - # Step 5: Add domain to Cloudflare - print(f"Adding {domain} to Cloudflare...") - try: - 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"✓ Domain added to Cloudflare (Zone ID: {zone_id})") - except Exception as e: - result['errors'].append(f"Failed to add domain to Cloudflare: {str(e)}") - return result - - # Step 6: 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 7: Update nameservers in Namecheap - if not dry_run: - print(f"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(f"✓ 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') - - # Step 8: Set up basic DNS records (skip for now, empty DNS list) - print("Setting up basic DNS records...") - try: - # For now, just create an empty DNS list as requested - dns_records = [] - result['dns_records'] = dns_records - result['steps_completed'].append('basic_dns_setup') - print(f"✓ 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 9: Set up worker subdomain if requested - if setup_workers: - print("Setting up worker subdomain...") - try: - 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") - 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!") - print(f"Zone ID: {zone_id}") - print(f"Nameservers: {', '.join(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: python domain_manager.py [--dry-run] [--setup-only]") - sys.exit(1) - - domain = sys.argv[1] - dry_run = '--dry-run' in sys.argv - setup_only = '--setup-only' in sys.argv - - # 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) - - # Register and setup domain - if setup_only: - print("Running in SETUP-ONLY mode - skipping domain registration") - if dry_run: - print("Running in DRY RUN mode - no actual registration will occur") - result = manager.register_and_setup_domain(domain, dry_run=dry_run, skip_registration=setup_only) - - if result.get('success'): - print(f"\nSuccess! Domain {domain} is ready for use.") - else: - print(f"\nErrors occurred:") - for error in result.get('errors', []): - print(f" - {error}") - sys.exit(1) - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/namecheap_api.py b/namecheap_api.py deleted file mode 100644 index c85da02..0000000 --- a/namecheap_api.py +++ /dev/null @@ -1,165 +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' \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 5c6435d..ec332f1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,7 +20,7 @@ dev = [ ] [project.scripts] -regflow = "domain_manager:main" +regflow = "regflow.domain_manager:main" [tool.hatch.build.targets.wheel] -packages = ["."] \ No newline at end of file +packages = ["regflow"] \ No newline at end of file diff --git a/regflow/__init__.py b/regflow/__init__.py new file mode 100644 index 0000000..1d12396 --- /dev/null +++ b/regflow/__init__.py @@ -0,0 +1,3 @@ +"""RegFlow - Domain registration and DNS management automation tool.""" + +__version__ = "0.1.0" \ No newline at end of file diff --git a/regflow/config.py b/regflow/config.py new file mode 100644 index 0000000..164d20d --- /dev/null +++ b/regflow/config.py @@ -0,0 +1,23 @@ +import os +from dotenv import load_dotenv +from pydantic import BaseModel + +load_dotenv() + + +class Config(BaseModel): + namecheap_api_user: str + namecheap_api_key: str + namecheap_username: str + namecheap_client_ip: str + cloudflare_api_token: str + + @classmethod + def from_env(cls): + return cls( + namecheap_api_user=os.getenv("NAMECHEAP_API_USER", ""), + namecheap_api_key=os.getenv("NAMECHEAP_API_KEY", ""), + namecheap_username=os.getenv("NAMECHEAP_USERNAME", ""), + namecheap_client_ip=os.getenv("NAMECHEAP_CLIENT_IP", ""), + cloudflare_api_token=os.getenv("CLOUDFLARE_API_TOKEN", ""), + ) diff --git a/regflow/domain_manager.py b/regflow/domain_manager.py new file mode 100755 index 0000000..2f1d15a --- /dev/null +++ b/regflow/domain_manager.py @@ -0,0 +1,276 @@ +#!/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 register_and_setup_domain( + self, + domain: str, + registrant_info: Optional[Dict[str, str]] = None, + setup_workers: bool = True, + dry_run: bool = False, + skip_registration: bool = False, + ) -> Dict[str, Any]: + """ + Complete domain registration and setup workflow: + 1. Check domain availability + 2. Verify pricing and account balance + 3. Register domain (with user confirmation) + 4. Add domain to Cloudflare + 5. Update nameservers in Namecheap + 6. Set up basic DNS records + """ + result = {"domain": domain, "steps_completed": [], "errors": []} + + try: + if not skip_registration: + # Step 1: Check domain availability + print(f"Checking availability for {domain}...") + if not self.namecheap.check_domain_availability(domain): + result["errors"].append(f"Domain {domain} is not available") + return result + + result["steps_completed"].append("availability_check") + print(f"✓ Domain {domain} is available") + + # Step 2: Check pricing and balance + print("Checking pricing and account balance...") + + try: + pricing = self.namecheap.get_domain_pricing(domain) + except Exception as e: + result["errors"].append(f"Failed to get domain pricing: {str(e)}") + return result + + try: + balance = self.namecheap.get_account_balance() + except Exception as e: + result["errors"].append(f"Failed to get account 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 + + result["steps_completed"].append("pricing_check") + print(f"✓ Pricing: ${pricing['register']:.2f}, Balance: ${balance:.2f}") + + # Step 3: User confirmation with multiple safeguards + print(f"\n" + "=" * 50) + print(f"DOMAIN REGISTRATION CONFIRMATION") + print(f"=" * 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(f"=" * 50) + + # Multiple confirmation steps to prevent accidents + print( + f"\nWARNING: This will charge ${pricing['register']:.2f} to your account!" + ) + 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 + + # Step 4: Register domain + if dry_run: + print( + f"DRY RUN: Would register domain {domain} (skipping actual registration)" + ) + result["steps_completed"].append("domain_registration_dry_run") + else: + 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"✓ Skipping domain registration for {domain} (assuming already registered)" + ) + result["steps_completed"].append("domain_registration_skipped") + + # Step 5: Add domain to Cloudflare + print(f"Adding {domain} to Cloudflare...") + try: + 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"✓ Domain added to Cloudflare (Zone ID: {zone_id})") + except Exception as e: + result["errors"].append(f"Failed to add domain to Cloudflare: {str(e)}") + return result + + # Step 6: 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 7: Update nameservers in Namecheap + if not dry_run: + print(f"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(f"✓ 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") + + # Step 8: Set up basic DNS records (skip for now, empty DNS list) + print("Setting up basic DNS records...") + try: + # For now, just create an empty DNS list as requested + dns_records = [] + result["dns_records"] = dns_records + result["steps_completed"].append("basic_dns_setup") + print(f"✓ 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 9: Set up worker subdomain if requested + if setup_workers: + print("Setting up worker subdomain...") + try: + 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") + 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!") + print(f"Zone ID: {zone_id}") + print(f"Nameservers: {', '.join(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: python domain_manager.py [--dry-run] [--setup-only]") + sys.exit(1) + + domain = sys.argv[1] + dry_run = "--dry-run" in sys.argv + setup_only = "--setup-only" in sys.argv + + # 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) + + # Register and setup domain + if setup_only: + print("Running in SETUP-ONLY mode - skipping domain registration") + if dry_run: + print("Running in DRY RUN mode - no actual registration will occur") + result = manager.register_and_setup_domain( + domain, dry_run=dry_run, skip_registration=setup_only + ) + + if result.get("success"): + print(f"\nSuccess! Domain {domain} is ready for use.") + else: + print(f"\nErrors occurred:") + for error in result.get("errors", []): + print(f" - {error}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/regflow/providers/__init__.py b/regflow/providers/__init__.py new file mode 100644 index 0000000..1babe54 --- /dev/null +++ b/regflow/providers/__init__.py @@ -0,0 +1 @@ +"""DNS and domain registration providers.""" \ No newline at end of file diff --git a/regflow/providers/cloudflare_api.py b/regflow/providers/cloudflare_api.py new file mode 100644 index 0000000..e1154f1 --- /dev/null +++ b/regflow/providers/cloudflare_api.py @@ -0,0 +1,165 @@ +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 as e: + # 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( + f"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", []) diff --git a/regflow/providers/namecheap_api.py b/regflow/providers/namecheap_api.py new file mode 100644 index 0000000..6449ecd --- /dev/null +++ b/regflow/providers/namecheap_api.py @@ -0,0 +1,163 @@ +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" diff --git a/regflow/tests/__init__.py b/regflow/tests/__init__.py new file mode 100644 index 0000000..a9a6965 --- /dev/null +++ b/regflow/tests/__init__.py @@ -0,0 +1 @@ +"""Test suite for RegFlow.""" \ No newline at end of file diff --git a/regflow/tests/test_integration.py b/regflow/tests/test_integration.py new file mode 100644 index 0000000..7cb52b9 --- /dev/null +++ b/regflow/tests/test_integration.py @@ -0,0 +1,126 @@ +import pytest +import os +from ..config import Config +from ..providers.namecheap_api import NamecheapAPI +from ..providers.cloudflare_api import CloudflareAPI + + +@pytest.fixture +def config(): + """Load configuration for tests""" + return Config.from_env() + + +@pytest.fixture +def namecheap_api(config): + """Create Namecheap API instance""" + return NamecheapAPI(config) + + +@pytest.fixture +def cloudflare_api(config): + """Create Cloudflare API instance""" + return CloudflareAPI(config) + + +def test_namecheap_credentials_loaded(config): + """Test that Namecheap credentials are loaded""" + assert config.namecheap_api_key, "Namecheap API key not found" + assert config.namecheap_api_user, "Namecheap API user not found" + assert config.namecheap_username, "Namecheap username not found" + assert config.namecheap_client_ip, "Namecheap client IP not found" + + +def test_cloudflare_credentials_loaded(config): + """Test that Cloudflare credentials are loaded""" + assert config.cloudflare_api_token, "Cloudflare API token not found" + assert config.cloudflare_api_token != "your_cloudflare_api_token", "Cloudflare API token is placeholder" + assert len(config.cloudflare_api_token) == 40, f"Cloudflare API token should be 40 chars, got {len(config.cloudflare_api_token)}" + + +def test_namecheap_domain_availability(namecheap_api): + """Test domain availability check""" + test_domain = "test-domain-12345.com" + is_available = namecheap_api.check_domain_availability(test_domain) + assert isinstance(is_available, bool), "Domain availability should return boolean" + + +def test_namecheap_account_balance(namecheap_api): + """Test account balance retrieval""" + balance = namecheap_api.get_account_balance() + assert isinstance(balance, (int, float)), "Balance should be numeric" + assert balance >= 0, "Balance should be non-negative" + + +def test_namecheap_domain_pricing(namecheap_api): + """Test domain pricing retrieval""" + pricing = namecheap_api.get_domain_pricing("example.com") + assert isinstance(pricing, dict), "Pricing should be a dictionary" + assert "register" in pricing, "Pricing should contain 'register' key" + assert "renew" in pricing, "Pricing should contain 'renew' key" + assert pricing["register"] > 0, "Registration price should be positive" + assert pricing["renew"] > 0, "Renewal price should be positive" + + +def test_namecheap_different_tld_pricing(namecheap_api): + """Test pricing for different TLD""" + pricing = namecheap_api.get_domain_pricing("example.xyz") + assert isinstance(pricing, dict), "Pricing should be a dictionary" + assert pricing["register"] > 0, ".xyz registration price should be positive" + + +def test_cloudflare_list_zones(cloudflare_api): + """Test listing Cloudflare zones""" + zones = cloudflare_api.list_zones() + assert isinstance(zones, list), "Zones should be a list" + + if zones: + zone = zones[0] + assert "id" in zone, "Zone should have 'id' field" + assert "name" in zone, "Zone should have 'name' field" + + +def test_cloudflare_zone_info(cloudflare_api): + """Test getting zone information""" + zones = cloudflare_api.list_zones() + + if zones: + zone_name = zones[0]["name"] + zone_info = cloudflare_api.get_zone_info(zone_name) + + assert zone_info is not None, "Zone info should not be None" + assert zone_info["name"] == zone_name, "Zone name should match" + assert "id" in zone_info, "Zone info should have 'id' field" + + +def test_cloudflare_nameservers(cloudflare_api): + """Test getting zone nameservers""" + zones = cloudflare_api.list_zones() + + if zones: + zone_id = zones[0]["id"] + nameservers = cloudflare_api.get_zone_nameservers(zone_id) + + assert isinstance(nameservers, list), "Nameservers should be a list" + assert len(nameservers) > 0, "Should have at least one nameserver" + + for ns in nameservers: + assert isinstance(ns, str), "Nameserver should be a string" + assert "cloudflare.com" in ns, "Nameserver should be from Cloudflare" + + +def test_cloudflare_dns_records(cloudflare_api): + """Test getting DNS records""" + zones = cloudflare_api.list_zones() + + if zones: + zone_id = zones[0]["id"] + records = cloudflare_api.get_zone_dns_records(zone_id) + + assert isinstance(records, list), "DNS records should be a list" + + if records: + record = records[0] + 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" \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py deleted file mode 100644 index 9b534f0..0000000 --- a/tests/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# Integration tests for regflow \ No newline at end of file diff --git a/tests/test_integration.py b/tests/test_integration.py deleted file mode 100644 index 4748172..0000000 --- a/tests/test_integration.py +++ /dev/null @@ -1,126 +0,0 @@ -import pytest -import os -from config import Config -from namecheap_api import NamecheapAPI -from cloudflare_api import CloudflareAPI - - -@pytest.fixture -def config(): - """Load configuration for tests""" - return Config.from_env() - - -@pytest.fixture -def namecheap_api(config): - """Create Namecheap API instance""" - return NamecheapAPI(config) - - -@pytest.fixture -def cloudflare_api(config): - """Create Cloudflare API instance""" - return CloudflareAPI(config) - - -def test_namecheap_credentials_loaded(config): - """Test that Namecheap credentials are loaded""" - assert config.namecheap_api_key, "Namecheap API key not found" - assert config.namecheap_api_user, "Namecheap API user not found" - assert config.namecheap_username, "Namecheap username not found" - assert config.namecheap_client_ip, "Namecheap client IP not found" - - -def test_cloudflare_credentials_loaded(config): - """Test that Cloudflare credentials are loaded""" - assert config.cloudflare_api_token, "Cloudflare API token not found" - assert config.cloudflare_api_token != "your_cloudflare_api_token", "Cloudflare API token is placeholder" - assert len(config.cloudflare_api_token) == 40, f"Cloudflare API token should be 40 chars, got {len(config.cloudflare_api_token)}" - - -def test_namecheap_domain_availability(namecheap_api): - """Test domain availability check""" - test_domain = "test-domain-12345.com" - is_available = namecheap_api.check_domain_availability(test_domain) - assert isinstance(is_available, bool), "Domain availability should return boolean" - - -def test_namecheap_account_balance(namecheap_api): - """Test account balance retrieval""" - balance = namecheap_api.get_account_balance() - assert isinstance(balance, (int, float)), "Balance should be numeric" - assert balance >= 0, "Balance should be non-negative" - - -def test_namecheap_domain_pricing(namecheap_api): - """Test domain pricing retrieval""" - pricing = namecheap_api.get_domain_pricing("example.com") - assert isinstance(pricing, dict), "Pricing should be a dictionary" - assert "register" in pricing, "Pricing should contain 'register' key" - assert "renew" in pricing, "Pricing should contain 'renew' key" - assert pricing["register"] > 0, "Registration price should be positive" - assert pricing["renew"] > 0, "Renewal price should be positive" - - -def test_namecheap_different_tld_pricing(namecheap_api): - """Test pricing for different TLD""" - pricing = namecheap_api.get_domain_pricing("example.xyz") - assert isinstance(pricing, dict), "Pricing should be a dictionary" - assert pricing["register"] > 0, ".xyz registration price should be positive" - - -def test_cloudflare_list_zones(cloudflare_api): - """Test listing Cloudflare zones""" - zones = cloudflare_api.list_zones() - assert isinstance(zones, list), "Zones should be a list" - - if zones: - zone = zones[0] - assert "id" in zone, "Zone should have 'id' field" - assert "name" in zone, "Zone should have 'name' field" - - -def test_cloudflare_zone_info(cloudflare_api): - """Test getting zone information""" - zones = cloudflare_api.list_zones() - - if zones: - zone_name = zones[0]["name"] - zone_info = cloudflare_api.get_zone_info(zone_name) - - assert zone_info is not None, "Zone info should not be None" - assert zone_info["name"] == zone_name, "Zone name should match" - assert "id" in zone_info, "Zone info should have 'id' field" - - -def test_cloudflare_nameservers(cloudflare_api): - """Test getting zone nameservers""" - zones = cloudflare_api.list_zones() - - if zones: - zone_id = zones[0]["id"] - nameservers = cloudflare_api.get_zone_nameservers(zone_id) - - assert isinstance(nameservers, list), "Nameservers should be a list" - assert len(nameservers) > 0, "Should have at least one nameserver" - - for ns in nameservers: - assert isinstance(ns, str), "Nameserver should be a string" - assert "cloudflare.com" in ns, "Nameserver should be from Cloudflare" - - -def test_cloudflare_dns_records(cloudflare_api): - """Test getting DNS records""" - zones = cloudflare_api.list_zones() - - if zones: - zone_id = zones[0]["id"] - records = cloudflare_api.get_zone_dns_records(zone_id) - - assert isinstance(records, list), "DNS records should be a list" - - if records: - record = records[0] - 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" \ No newline at end of file -- cgit v1.3.1