diff options
| -rw-r--r-- | cloudflare_api.py | 143 | ||||
| -rw-r--r-- | namecheap_api.py | 165 | ||||
| -rw-r--r-- | pyproject.toml | 4 | ||||
| -rw-r--r-- | regflow/__init__.py | 3 | ||||
| -rw-r--r-- | regflow/config.py (renamed from config.py) | 7 | ||||
| -rwxr-xr-x | regflow/domain_manager.py (renamed from domain_manager.py) | 251 | ||||
| -rw-r--r-- | regflow/providers/__init__.py | 1 | ||||
| -rw-r--r-- | regflow/providers/cloudflare_api.py | 165 | ||||
| -rw-r--r-- | regflow/providers/namecheap_api.py | 163 | ||||
| -rw-r--r-- | regflow/tests/__init__.py | 1 | ||||
| -rw-r--r-- | regflow/tests/test_integration.py (renamed from tests/test_integration.py) | 6 | ||||
| -rw-r--r-- | tests/__init__.py | 1 |
12 files changed, 485 insertions, 425 deletions
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/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/config.py b/regflow/config.py index 488c52d..164d20d 100644 --- a/config.py +++ b/regflow/config.py @@ -4,13 +4,14 @@ 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( @@ -18,5 +19,5 @@ class Config(BaseModel): 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 + cloudflare_api_token=os.getenv("CLOUDFLARE_API_TOKEN", ""), + ) diff --git a/domain_manager.py b/regflow/domain_manager.py index 707ce4f..2f1d15a 100755 --- a/domain_manager.py +++ b/regflow/domain_manager.py @@ -2,21 +2,25 @@ import sys from typing import Dict, Any, Optional -from config import Config -from namecheap_api import NamecheapAPI -from cloudflare_api import CloudflareAPI +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]: + + 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 @@ -26,216 +30,247 @@ class DomainManager: 5. Update nameservers in Namecheap 6. Set up basic DNS records """ - result = { - 'domain': domain, - 'steps_completed': [], - 'errors': [] - } - + 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") + result["errors"].append(f"Domain {domain} is not available") return result - - result['steps_completed'].append('availability_check') + + 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)}") + 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)}") + result["errors"].append(f"Failed to get account balance: {str(e)}") return result - - if balance < pricing['register']: - result['errors'].append( + + 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') + + 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"\n" + "=" * 50) print(f"DOMAIN REGISTRATION CONFIRMATION") - print(f"="*50) + 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) - + 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") + 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") + + 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') + 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}") + 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') - + 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') - + 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 + 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)}") + 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 + + result["nameservers"] = nameservers print(f"✓ Cloudflare nameservers: {', '.join(nameservers)}") except Exception as e: - result['errors'].append(f"Failed to get Cloudflare nameservers: {str(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") + result["errors"].append( + "Failed to update nameservers in Namecheap" + ) return result - - result['steps_completed'].append('nameserver_update') + + 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)}") + 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') - + 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') + 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)}") + 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') + 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)}") + result["errors"].append( + f"Failed to set up worker subdomain: {str(e)}" + ) return result - - result['success'] = True + + 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)}") + 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} - + 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)} + return {"error": str(e)} + def main(): if len(sys.argv) < 2: print("Usage: python domain_manager.py <domain> [--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 - + 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 - ]): + 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'): + 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', []): + for error in result.get("errors", []): print(f" - {error}") sys.exit(1) + if __name__ == "__main__": - main()
\ No newline at end of file + 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/tests/test_integration.py b/regflow/tests/test_integration.py index 4748172..7cb52b9 100644 --- a/tests/test_integration.py +++ b/regflow/tests/test_integration.py @@ -1,8 +1,8 @@ import pytest import os -from config import Config -from namecheap_api import NamecheapAPI -from cloudflare_api import CloudflareAPI +from ..config import Config +from ..providers.namecheap_api import NamecheapAPI +from ..providers.cloudflare_api import CloudflareAPI @pytest.fixture 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 |