286 lines
10 KiB
Python
286 lines
10 KiB
Python
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
|