Created
November 6, 2024 16:08
-
-
Save TownLake/01f4075258fbdf9fc755c67fd83ed1c8 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests | |
import json | |
import os | |
from datetime import datetime | |
class CloudflareExporter: | |
def __init__(self, auth_token=None, api_key=None, email=None): | |
""" | |
Initialize with either an API Token or API Key + Email combination | |
:param auth_token: Cloudflare API Token | |
:param api_key: Cloudflare API Key | |
:param email: Cloudflare account email | |
""" | |
if auth_token: | |
self.headers = { | |
"Authorization": f"Bearer {auth_token}", | |
"Content-Type": "application/json" | |
} | |
elif api_key and email: | |
self.headers = { | |
"X-Auth-Key": api_key, | |
"X-Auth-Email": email, | |
"Content-Type": "application/json" | |
} | |
else: | |
raise ValueError("Either auth_token or both api_key and email must be provided") | |
self.base_url = "https://api.cloudflare.com/client/v4" | |
def _make_request(self, method, endpoint, params=None): | |
""" | |
Make an API request with error handling | |
""" | |
url = f"{self.base_url}{endpoint}" | |
try: | |
response = requests.request(method, url, headers=self.headers, params=params) | |
response.raise_for_status() | |
return response.json() | |
except requests.exceptions.HTTPError as e: | |
error_msg = f"HTTP Error: {e}" | |
try: | |
error_detail = response.json() | |
if 'errors' in error_detail: | |
error_msg += f"\nAPI Errors: {json.dumps(error_detail['errors'], indent=2)}" | |
except: | |
pass | |
raise Exception(error_msg) | |
except requests.exceptions.RequestException as e: | |
raise Exception(f"Request Error: {e}") | |
def get_zone_id(self, domain_name): | |
"""Get zone ID for a domain""" | |
try: | |
response = self._make_request('GET', '/zones', params={"name": domain_name}) | |
zones = response["result"] | |
if not zones: | |
raise ValueError(f"No zones found for domain: {domain_name}") | |
return zones[0]["id"] | |
except Exception as e: | |
raise Exception(f"Error getting zone ID: {str(e)}") | |
def export_dns_records(self, zone_id): | |
"""Export all DNS records""" | |
try: | |
return self._make_request('GET', f'/zones/{zone_id}/dns_records')["result"] | |
except Exception as e: | |
print(f"Warning: Could not export DNS records: {str(e)}") | |
return [] | |
def export_ssl_config(self, zone_id): | |
"""Export SSL configuration""" | |
try: | |
return self._make_request('GET', f'/zones/{zone_id}/ssl/verification')["result"] | |
except Exception as e: | |
print(f"Warning: Could not export SSL config: {str(e)}") | |
return {} | |
def export_page_rules(self, zone_id): | |
"""Export page rules""" | |
try: | |
return self._make_request('GET', f'/zones/{zone_id}/pagerules')["result"] | |
except Exception as e: | |
print(f"Warning: Could not export page rules: {str(e)}") | |
return [] | |
def export_firewall_rules(self, zone_id): | |
"""Export firewall rules""" | |
try: | |
return self._make_request('GET', f'/zones/{zone_id}/firewall/rules')["result"] | |
except Exception as e: | |
print(f"Warning: Could not export firewall rules: {str(e)}") | |
return [] | |
def export_zone_settings(self, zone_id): | |
"""Export zone settings""" | |
try: | |
return self._make_request('GET', f'/zones/{zone_id}/settings')["result"] | |
except Exception as e: | |
print(f"Warning: Could not export zone settings: {str(e)}") | |
return [] | |
def export_workers(self, zone_id): | |
"""Export Workers scripts""" | |
try: | |
return self._make_request('GET', f'/zones/{zone_id}/workers/routes')["result"] | |
except Exception as e: | |
print(f"Warning: Could not export workers: {str(e)}") | |
return [] | |
def export_rate_limits(self, zone_id): | |
"""Export rate limiting rules""" | |
try: | |
return self._make_request('GET', f'/zones/{zone_id}/rate_limits')["result"] | |
except Exception as e: | |
print(f"Warning: Could not export rate limits: {str(e)}") | |
return [] | |
def export_zone_details(self, zone_id): | |
"""Export basic zone details""" | |
try: | |
return self._make_request('GET', f'/zones/{zone_id}')["result"] | |
except Exception as e: | |
print(f"Warning: Could not export zone details: {str(e)}") | |
return {} | |
def export_all(self, domain_name): | |
"""Export all available configurations""" | |
print(f"Starting export for {domain_name}...") | |
try: | |
zone_id = self.get_zone_id(domain_name) | |
print(f"Found zone ID: {zone_id}") | |
except Exception as e: | |
raise Exception(f"Failed to get zone ID: {str(e)}") | |
export_data = { | |
"domain": domain_name, | |
"zone_id": zone_id, | |
"export_date": datetime.now().isoformat(), | |
"zone_details": self.export_zone_details(zone_id), | |
"dns_records": self.export_dns_records(zone_id), | |
"ssl_config": self.export_ssl_config(zone_id), | |
"page_rules": self.export_page_rules(zone_id), | |
"firewall_rules": self.export_firewall_rules(zone_id), | |
"zone_settings": self.export_zone_settings(zone_id), | |
"workers": self.export_workers(zone_id), | |
"rate_limits": self.export_rate_limits(zone_id) | |
} | |
# Create export directory if it doesn't exist | |
export_dir = "cloudflare_exports" | |
if not os.path.exists(export_dir): | |
os.makedirs(export_dir) | |
# Save to JSON file | |
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
filename = f"{export_dir}/{domain_name}_export_{timestamp}.json" | |
with open(filename, 'w') as f: | |
json.dump(export_data, f, indent=2) | |
print(f"Export completed successfully. File saved as: {filename}") | |
return filename | |
def main(): | |
# You can use either API Token | |
AUTH_TOKEN = "your_api_token_here" | |
# Or API Key + Email combination | |
API_KEY = "<>" | |
EMAIL = "<>" | |
DOMAIN = "<>" | |
try: | |
# Use either: | |
# exporter = CloudflareExporter(auth_token=AUTH_TOKEN) | |
# or: | |
exporter = CloudflareExporter(api_key=API_KEY, email=EMAIL) | |
export_file = exporter.export_all(DOMAIN) | |
print(f"Successfully exported configuration to: {export_file}") | |
except Exception as e: | |
print(f"Error during export: {str(e)}") | |
if __name__ == "__main__": | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment