import hashlib from typing import Dict, List, Optional import CloudFlare class CloudflareService: """Cloudflare API işlemleri""" def __init__(self, api_token: str): self.cf = CloudFlare.CloudFlare(token=api_token) self.api_token = api_token def validate_token_and_get_zone(self, domain: str) -> Dict: """ API token doğrula ve zone bilgilerini al """ try: # Zone ara zones = self.cf.zones.get(params={"name": domain}) if not zones: return { "status": "error", "message": f"{domain} zone bulunamadı. Domain Cloudflare hesabınızda olduğundan emin olun." } zone = zones[0] zone_id = zone["id"] # Mevcut DNS kayıtlarını al dns_records = self.cf.zones.dns_records.get(zone_id) return { "status": "success", "zone_id": zone_id, "zone_name": zone["name"], "zone_status": zone["status"], "nameservers": zone.get("name_servers", []), "account_email": zone.get("account", {}).get("email", "N/A"), "current_dns_records": [ { "type": r["type"], "name": r["name"], "content": r["content"], "proxied": r.get("proxied", False), "ttl": r["ttl"], "id": r["id"] } for r in dns_records ] } except CloudFlare.exceptions.CloudFlareAPIError as e: return { "status": "error", "message": f"Cloudflare API hatası: {str(e)}" } except Exception as e: return { "status": "error", "message": f"Beklenmeyen hata: {str(e)}" } def generate_dns_preview(self, domain: str, zone_id: str, new_ip: str) -> Dict: """ DNS değişiklik önizlemesi oluştur """ try: # Mevcut A kayıtlarını al dns_records = self.cf.zones.dns_records.get( zone_id, params={"type": "A"} ) current_root = None current_www = None for record in dns_records: if record["name"] == domain: current_root = record elif record["name"] == f"www.{domain}": current_www = record # Önizleme oluştur preview = { "domain": domain, "new_ip": new_ip, "changes": [] } # Root domain (@) değişikliği if current_root: preview["changes"].append({ "record_type": "A", "name": "@", "current": { "value": current_root["content"], "proxied": current_root.get("proxied", False), "ttl": current_root["ttl"] }, "new": { "value": new_ip, "proxied": current_root.get("proxied", True), "ttl": "auto" }, "action": "update", "record_id": current_root["id"] }) else: preview["changes"].append({ "record_type": "A", "name": "@", "current": None, "new": { "value": new_ip, "proxied": True, "ttl": "auto" }, "action": "create" }) # www subdomain değişikliği if current_www: preview["changes"].append({ "record_type": "A", "name": "www", "current": { "value": current_www["content"], "proxied": current_www.get("proxied", False), "ttl": current_www["ttl"] }, "new": { "value": new_ip, "proxied": current_www.get("proxied", True), "ttl": "auto" }, "action": "update", "record_id": current_www["id"] }) else: preview["changes"].append({ "record_type": "A", "name": "www", "current": None, "new": { "value": new_ip, "proxied": True, "ttl": "auto" }, "action": "create" }) # Diğer kayıtlar (değişmeyecek) all_records = self.cf.zones.dns_records.get(zone_id) other_records = [ r for r in all_records if r["type"] != "A" or (r["name"] != domain and r["name"] != f"www.{domain}") ] preview["preserved_records"] = [ { "type": r["type"], "name": r["name"], "content": r["content"] } for r in other_records[:10] # İlk 10 kayıt ] preview["preserved_count"] = len(other_records) return preview except Exception as e: return { "status": "error", "message": f"Önizleme oluşturma hatası: {str(e)}" } def apply_dns_changes(self, zone_id: str, preview: Dict, proxy_enabled: bool = True) -> Dict: """ DNS değişikliklerini uygula """ results = { "domain": preview["domain"], "applied_changes": [], "errors": [] } for change in preview["changes"]: try: if change["action"] == "update": # Mevcut kaydı güncelle self.cf.zones.dns_records.patch( zone_id, change["record_id"], data={ "type": "A", "name": change["name"], "content": change["new"]["value"], "proxied": proxy_enabled, "ttl": 1 if proxy_enabled else 300 } ) results["applied_changes"].append({ "name": change["name"], "action": "updated", "new_value": change["new"]["value"] }) elif change["action"] == "create": # Yeni kayıt oluştur self.cf.zones.dns_records.post( zone_id, data={ "type": "A", "name": change["name"], "content": change["new"]["value"], "proxied": proxy_enabled, "ttl": 1 if proxy_enabled else 300 } ) results["applied_changes"].append({ "name": change["name"], "action": "created", "new_value": change["new"]["value"] }) except Exception as e: results["errors"].append({ "name": change["name"], "error": str(e) }) if results["errors"]: results["status"] = "partial" else: results["status"] = "success" return results def configure_ssl(self, zone_id: str) -> Dict: """ Cloudflare SSL ayarlarını yapılandır """ ssl_config = { "steps": [], "errors": [] } try: # 1. SSL/TLS Mode: Full (strict) self.cf.zones.settings.ssl.patch(zone_id, data={"value": "full"}) ssl_config["steps"].append({"name": "ssl_mode", "status": "success", "value": "full"}) except Exception as e: ssl_config["errors"].append({"step": "ssl_mode", "error": str(e)}) try: # 2. Always Use HTTPS self.cf.zones.settings.always_use_https.patch(zone_id, data={"value": "on"}) ssl_config["steps"].append({"name": "always_https", "status": "success"}) except Exception as e: ssl_config["errors"].append({"step": "always_https", "error": str(e)}) try: # 3. Automatic HTTPS Rewrites self.cf.zones.settings.automatic_https_rewrites.patch(zone_id, data={"value": "on"}) ssl_config["steps"].append({"name": "auto_https_rewrites", "status": "success"}) except Exception as e: ssl_config["errors"].append({"step": "auto_https_rewrites", "error": str(e)}) try: # 4. Minimum TLS Version self.cf.zones.settings.min_tls_version.patch(zone_id, data={"value": "1.2"}) ssl_config["steps"].append({"name": "min_tls", "status": "success", "value": "1.2"}) except Exception as e: ssl_config["errors"].append({"step": "min_tls", "error": str(e)}) try: # 5. TLS 1.3 self.cf.zones.settings.tls_1_3.patch(zone_id, data={"value": "on"}) ssl_config["steps"].append({"name": "tls_1_3", "status": "success"}) except Exception as e: ssl_config["errors"].append({"step": "tls_1_3", "error": str(e)}) return ssl_config