Eduardo David Paredes Vara
281 lines
10 KiB
Python
Executable File
281 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Manage direct Cloudflare A records for this server.
|
|
|
|
Credentials and zone settings are read from cloudflared/.env. This utility does
|
|
not create, configure or inspect Cloudflare Tunnels.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
API_BASE = "https://api.cloudflare.com/client/v4"
|
|
ENV_FILE = Path(__file__).resolve().with_name(".env")
|
|
DNS_NAME_RE = re.compile(
|
|
r"^(?:\*\.)?(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*"
|
|
r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$"
|
|
)
|
|
|
|
|
|
class CloudflareError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def load_env(path: Path) -> dict[str, str]:
|
|
if not path.is_file():
|
|
raise CloudflareError(f"No existe el archivo de configuración: {path}")
|
|
|
|
values: dict[str, str] = {}
|
|
for number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if line.startswith("export "):
|
|
line = line[7:].lstrip()
|
|
if "=" not in line:
|
|
raise CloudflareError(f"Línea inválida en {path}:{number}")
|
|
key, value = line.split("=", 1)
|
|
key = key.strip()
|
|
value = value.strip()
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
|
|
value = value[1:-1]
|
|
values[key] = value
|
|
return values
|
|
|
|
|
|
def required(env: dict[str, str], key: str) -> str:
|
|
value = env.get(key, "").strip()
|
|
if not value or value.lower().startswith(("change-me", "changeme")):
|
|
raise CloudflareError(f"{key} no está configurado en {ENV_FILE}")
|
|
return value
|
|
|
|
|
|
def optional(env: dict[str, str], key: str) -> str | None:
|
|
value = env.get(key, "").strip()
|
|
if not value or value.lower().startswith(("change-me", "changeme")):
|
|
return None
|
|
return value
|
|
|
|
|
|
def normalize_name(value: str, domain: str) -> str:
|
|
value = value.strip().lower().rstrip(".")
|
|
if value == "@":
|
|
return domain
|
|
if "." not in value:
|
|
value = f"{value}.{domain}"
|
|
if value != domain and not value.endswith(f".{domain}"):
|
|
raise CloudflareError(f"{value!r} está fuera de la zona {domain}")
|
|
if len(value) > 253 or not DNS_NAME_RE.fullmatch(value):
|
|
raise CloudflareError(f"Nombre DNS inválido: {value!r}")
|
|
return value
|
|
|
|
|
|
class CloudflareDNS:
|
|
def __init__(self, token: str, zone_id: str) -> None:
|
|
self.token = token
|
|
self.zone_id = zone_id
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
payload: dict[str, Any] | None = None,
|
|
query: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
url = f"{API_BASE}{path}"
|
|
if query:
|
|
url = f"{url}?{urlencode(query)}"
|
|
body = json.dumps(payload).encode() if payload is not None else None
|
|
request = Request(
|
|
url,
|
|
data=body,
|
|
method=method,
|
|
headers={
|
|
"Authorization": f"Bearer {self.token}",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "portainer-repo-cloudflare-dns/1.0",
|
|
},
|
|
)
|
|
try:
|
|
with urlopen(request, timeout=30) as response:
|
|
result = json.load(response)
|
|
except HTTPError as exc:
|
|
try:
|
|
detail = json.load(exc)
|
|
errors = detail.get("errors") or []
|
|
except (ValueError, AttributeError):
|
|
errors = []
|
|
message = "; ".join(str(item.get("message", item)) for item in errors)
|
|
raise CloudflareError(
|
|
f"Cloudflare respondió HTTP {exc.code}" + (f": {message}" if message else "")
|
|
) from exc
|
|
except (URLError, TimeoutError) as exc:
|
|
raise CloudflareError(f"No se pudo conectar con Cloudflare: {exc}") from exc
|
|
|
|
if not result.get("success"):
|
|
errors = result.get("errors") or []
|
|
message = "; ".join(str(item.get("message", item)) for item in errors)
|
|
raise CloudflareError(message or "Cloudflare rechazó la operación")
|
|
return result
|
|
|
|
def list_records(self) -> list[dict[str, Any]]:
|
|
records: list[dict[str, Any]] = []
|
|
page = 1
|
|
while True:
|
|
response = self.request(
|
|
"GET",
|
|
f"/zones/{self.zone_id}/dns_records",
|
|
query={"type": "A", "page": page, "per_page": 100},
|
|
)
|
|
records.extend(response.get("result") or [])
|
|
info = response.get("result_info") or {}
|
|
if page >= int(info.get("total_pages", 1)):
|
|
return records
|
|
page += 1
|
|
|
|
def records_for_name(self, name: str) -> list[dict[str, Any]]:
|
|
response = self.request(
|
|
"GET",
|
|
f"/zones/{self.zone_id}/dns_records",
|
|
query={"name": name, "per_page": 100},
|
|
)
|
|
return response.get("result") or []
|
|
|
|
def create(self, name: str, server_ip: str) -> None:
|
|
existing = self.records_for_name(name)
|
|
if existing:
|
|
types = ", ".join(sorted({str(record.get("type")) for record in existing}))
|
|
raise CloudflareError(f"{name} ya existe ({types}); usa update")
|
|
self.request(
|
|
"POST",
|
|
f"/zones/{self.zone_id}/dns_records",
|
|
payload=record_payload(name, server_ip),
|
|
)
|
|
|
|
def update(self, name: str, server_ip: str) -> None:
|
|
records = self.records_for_name(name)
|
|
a_records = [record for record in records if record.get("type") == "A"]
|
|
if not a_records:
|
|
raise CloudflareError(f"No existe un registro A para {name}; usa create")
|
|
if len(a_records) != 1:
|
|
raise CloudflareError(f"Hay {len(a_records)} registros A para {name}; corrígelos manualmente")
|
|
self.request(
|
|
"PUT",
|
|
f"/zones/{self.zone_id}/dns_records/{a_records[0]['id']}",
|
|
payload=record_payload(name, server_ip),
|
|
)
|
|
|
|
def delete(self, name: str) -> None:
|
|
records = self.records_for_name(name)
|
|
a_records = [record for record in records if record.get("type") == "A"]
|
|
if not a_records:
|
|
raise CloudflareError(f"No existe un registro A para {name}")
|
|
if len(a_records) != 1:
|
|
raise CloudflareError(f"Hay {len(a_records)} registros A para {name}; corrígelos manualmente")
|
|
self.request("DELETE", f"/zones/{self.zone_id}/dns_records/{a_records[0]['id']}")
|
|
|
|
|
|
def record_payload(name: str, server_ip: str) -> dict[str, Any]:
|
|
return {
|
|
"type": "A",
|
|
"name": name,
|
|
"content": server_ip,
|
|
"ttl": 1,
|
|
"proxied": False,
|
|
"comment": "Direct to homelab server",
|
|
}
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
description="Gestiona registros A directos de Cloudflare para este servidor."
|
|
)
|
|
parser.add_argument("--dry-run", action="store_true", help="muestra cambios sin enviarlos")
|
|
parser.add_argument(
|
|
"--server-ip",
|
|
metavar="IP",
|
|
help="IPv4 de destino; reemplaza CLOUDFLARE_SERVER_IP del .env",
|
|
)
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
subparsers.add_parser("list", help="lista los registros A de la zona")
|
|
for command in ("create", "update"):
|
|
child = subparsers.add_parser(command, help=f"{command} registros A")
|
|
child.add_argument("names", nargs="+", metavar="NAME")
|
|
delete = subparsers.add_parser("delete", help="elimina registros A")
|
|
delete.add_argument("names", nargs="+", metavar="NAME")
|
|
delete.add_argument("--yes", action="store_true", help="no solicita confirmación")
|
|
return parser
|
|
|
|
|
|
def main() -> int:
|
|
args = build_parser().parse_args()
|
|
env = load_env(ENV_FILE)
|
|
token = required(env, "CLOUDFLARE_API_TOKEN")
|
|
zone_id = required(env, "CLOUDFLARE_ZONE_ID")
|
|
domain = required(env, "CLOUDFLARE_DOMAIN").lower().rstrip(".")
|
|
server_ip = args.server_ip or optional(env, "CLOUDFLARE_SERVER_IP")
|
|
if server_ip:
|
|
try:
|
|
ipaddress.IPv4Address(server_ip)
|
|
except ipaddress.AddressValueError as exc:
|
|
source = "--server-ip" if args.server_ip else "CLOUDFLARE_SERVER_IP"
|
|
raise CloudflareError(f"{source} debe ser una dirección IPv4 válida") from exc
|
|
elif args.command in ("create", "update"):
|
|
raise CloudflareError(
|
|
"Falta la IP de destino: configura CLOUDFLARE_SERVER_IP en .env "
|
|
"o usa --server-ip"
|
|
)
|
|
|
|
client = CloudflareDNS(token, zone_id)
|
|
if args.command == "list":
|
|
for record in sorted(client.list_records(), key=lambda item: item.get("name", "")):
|
|
if server_ip:
|
|
marker = "this-server" if record.get("content") == server_ip else "other-target"
|
|
print(f"{record.get('name')}\t{record.get('content')}\t{marker}")
|
|
else:
|
|
print(f"{record.get('name')}\t{record.get('content')}")
|
|
return 0
|
|
|
|
names = [normalize_name(value, domain) for value in args.names]
|
|
if args.command == "delete" and not args.yes and not args.dry_run:
|
|
print("Se eliminarán estos registros A:")
|
|
for name in names:
|
|
print(f" - {name}")
|
|
if input("Escribe DELETE para continuar: ").strip() != "DELETE":
|
|
print("Cancelado")
|
|
return 1
|
|
|
|
failed = False
|
|
for name in names:
|
|
try:
|
|
if args.dry_run:
|
|
target = "" if args.command == "delete" else f" -> {server_ip}"
|
|
print(f"DRY-RUN {args.command} {name}{target}")
|
|
else:
|
|
getattr(client, args.command)(name, server_ip) if args.command != "delete" else client.delete(name)
|
|
target = "" if args.command == "delete" else f" -> {server_ip}"
|
|
print(f"OK {args.command} {name}{target}")
|
|
except CloudflareError as exc:
|
|
failed = True
|
|
print(f"ERROR {name}: {exc}", file=sys.stderr)
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except (CloudflareError, KeyboardInterrupt) as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
raise SystemExit(1)
|