1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
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)
response.raise_for_status()
result = response.json()
if not result.get('success', False):
errors = result.get('errors', [])
error_msg = ', '.join([error.get('message', 'Unknown error') for error in errors])
raise Exception(f"Cloudflare API Error: {error_msg}")
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', [])
|
