ligbox-ops-platform/deploy/vm112-wizard/onboarding-dns037.patch.py

#!/usr/bin/env python3
"""Aplica patch Spec 037 em onboarding.py na VM112. Idempotente."""
from pathlib import Path

ROUTER = Path("/opt/ligbox-wizard/backend/app/routers/onboarding.py")
text = ROUTER.read_text(encoding="utf-8")

IMPORT_OLD = "from app.services.cloudflare import CloudflareDNS, CloudflareError, is_cloudflare_sandbox, mail_dns_records, wizard_nameservers"
IMPORT_NEW = """from app.services.cloudflare import CloudflareDNS, CloudflareError, is_cloudflare_sandbox, mail_dns_records, wizard_nameservers
from app.services.dns_account_registry import resolve_ligbox_zone, resolve_payload
from app.services import dns_byo_vault"""

if "dns_account_registry" not in text:
    text = text.replace(IMPORT_OLD, IMPORT_NEW)

MODEL_ANCHOR = "class CloudflareApplyRequest(BaseModel):"
MODEL_INSERT = """class CloudflareConnectCustomRequest(BaseModel):
    domain: str
    api_token: str = Field(..., min_length=20, max_length=512)


"""
if "CloudflareConnectCustomRequest" not in text:
    text = text.replace(MODEL_ANCHOR, MODEL_INSERT + MODEL_ANCHOR)

HELPER = '''

def _cf_client_for_domain(domain: str, session_id: str | None = None):
    """Resolve cliente CF: conta Ligbox, BYO sessão, ou token legacy."""
    match = resolve_ligbox_zone(domain)
    if match:
        return match.account.client(), match.zone_id, match.dns_mode, match.account.id
    if session_id:
        byo = dns_byo_vault.load_byo_token(session_id, domain)
        if byo:
            return CloudflareDNS(token=byo.api_token), byo.zone_id, "byo_cloudflare", None
    if settings.cloudflare_api_token:
        cf = CloudflareDNS()
        zone = cf.get_zone_by_name(domain)
        if zone:
            return cf, zone["id"], "ligbox_cf_legacy", None
    return None, None, None, None


'''

if "_cf_client_for_domain" not in text:
    anchor = "def _cloudflare_zone_status(domain: str) -> dict:"
    text = text.replace(anchor, HELPER + anchor)

# Replace _cloudflare_zone_status body start
OLD_STATUS = '''def _cloudflare_zone_status(domain: str) -> dict:
    domain = domain.lower().strip()
    if not settings.cloudflare_api_token:
        return {
            "domain": domain,
            "cloudflare_configured": False,
            "zone_in_account": False,
            "can_apply_mail_records": False,
            "zone_status": None,
            "nameservers": [],
            "managed_zones": [],
            "portal_onboarding_required": True,
            "message": "API Cloudflare não configurada no servidor.",
        }
    cf = CloudflareDNS()
    cf.verify_token()
    managed = cf.list_zone_names()
    zone = cf.get_zone_by_name(domain)
    in_account = zone is not None
    zone_status = zone.get("status") if zone else None
    nameservers = wizard_nameservers(zone) if zone else []

    return {
        "domain": domain,
        "cloudflare_configured": True,
        "zone_in_account": in_account,
        "can_apply_mail_records": in_account,
        "zone_id": zone["id"] if zone else None,
        "zone_status": zone_status,
        "nameservers": nameservers,
        "managed_zones": managed,
        "portal_onboarding_required": not in_account,
        "message": (
            f"Zona {domain} pronta na Cloudflare Ligbox ({zone_status})."
            if in_account
            else f"Adicione {domain} na Cloudflare Ligbox e altere os NS no registrador."
        ),
    }'''

NEW_STATUS = '''def _cloudflare_zone_status(domain: str) -> dict:
    domain = domain.lower().strip()
    resolved = resolve_payload(domain)
    match = resolve_ligbox_zone(domain)
    if match:
        cf = match.account.client()
        try:
            cf.verify_token()
        except CloudflareError:
            pass
        zone = match.zone
        zone_status = zone.get("status")
        nameservers = wizard_nameservers(zone)
        return {
            "domain": domain,
            "cloudflare_configured": True,
            "zone_in_account": True,
            "can_apply_mail_records": True,
            "zone_id": match.zone_id,
            "zone_status": zone_status,
            "nameservers": nameservers,
            "ligbox_account_id": match.account.id,
            "ligbox_account_label": f"{match.account.label} ({match.account.admin_email})",
            "dns_mode": match.dns_mode,
            "resolve": resolved,
            "managed_zones": [],
            "portal_onboarding_required": zone_status != "active",
            "message": (
                f"Zona {domain} na conta {match.account.label} ({zone_status})."
            ),
        }
    legacy_ok = bool(settings.cloudflare_api_token)
    return {
        "domain": domain,
        "cloudflare_configured": legacy_ok,
        "zone_in_account": False,
        "can_apply_mail_records": False,
        "zone_id": None,
        "zone_status": None,
        "nameservers": [],
        "ligbox_account_id": None,
        "dns_mode": None,
        "resolve": resolved,
        "managed_zones": [],
        "portal_onboarding_required": True,
        "paths_available": resolved.get("paths_available") or ["byo", "external"],
        "message": resolved.get("message")
        or "Domínio fora das contas Ligbox — use BYO Cloudflare ou registrador.",
    }'''

if "ligbox_account_id" not in text:
    text = text.replace(OLD_STATUS, NEW_STATUS)

RESOLVE_ENDPOINT = '''
@router.get("/dns/resolve/{domain}")
def dns_resolve_account(domain: str):
    """Spec 037 — em qual conta Ligbox (ligit/itecnologys/ibytera) está a zona."""
    domain = domain.lower().strip().rstrip(".")
    try:
        activity_log.info(f"Resolver conta CF Ligbox: {domain}", source="cloudflare")
        return resolve_payload(domain)
    except CloudflareError as e:
        activity_log.error(str(e), source="cloudflare")
        raise HTTPException(502, f"Erro Cloudflare: {e}") from e


@router.post("/dns/cloudflare/connect-custom")
def connect_custom_cloudflare(body: CloudflareConnectCustomRequest, request: Request):
    """BYO — token API Cloudflare do cliente (zona já na conta dele)."""
    sid = get_session_from_request(request)
    domain = normalize_domain(body.domain)
    try:
        validate_primary_domain(domain)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=invalid_domain_http_detail()) from e
  cf = CloudflareDNS(token=body.api_token.strip())
    try:
        cf.verify_token()
        zone = cf.get_zone_by_name(domain)
        if not zone:
            raise HTTPException(
                422,
                detail={
                    "code": "zone_not_in_customer_account",
                    "domain": domain,
                    "message": f"Zona {domain} não encontrada com este API Token.",
                },
            )
        if sid:
            dns_byo_vault.save_byo_token(sid, domain, body.api_token.strip(), str(zone["id"]))
        activity_log.ok(f"BYO Cloudflare ligado: {domain}", source="cloudflare")
        return {
            "domain": domain,
            "matched": True,
            "dns_mode": "byo_cloudflare",
            "zone_id": zone.get("id"),
            "zone_status": zone.get("status"),
            "nameservers": wizard_nameservers(zone),
            "message": "Conta Cloudflare do cliente validada. Pode aplicar apontamentos.",
        }
    except CloudflareError as e:
        activity_log.error(f"BYO Cloudflare falhou: {e}", source="cloudflare")
        raise HTTPException(401, f"Token inválido: {e}") from e


'''

if "/dns/resolve/" not in text:
    text = text.replace(
        '@router.get("/dns/cloudflare/status/{domain}")',
        RESOLVE_ENDPOINT + '@router.get("/dns/cloudflare/status/{domain}")',
    )

# Fix indentation typo in patch if any - the RESOLVE has wrong indent on cf line
text = text.replace("  cf = CloudflareDNS(token=body.api_token.strip())", "    cf = CloudflareDNS(token=body.api_token.strip())")

# provision-zone guard
OLD_PROVISION_START = '''    cf = CloudflareDNS()
    try:
        activity_log.info(f"Criar zona Cloudflare: {domain}", source="cloudflare")
        cf.verify_token()
        result = cf.ensure_zone(domain)'''

NEW_PROVISION_START = '''    match = resolve_ligbox_zone(domain)
    if match:
        activity_log.info(
            f"Zona {domain} já na conta {match.account.id} — sem create",
            source="cloudflare",
        )
        payload = _portal_onboarding_payload(domain, zone_created=False)
        payload["zone_id"] = match.zone_id
        payload["ligbox_account_id"] = match.account.id
        payload["dns_mode"] = match.dns_mode
        payload["message"] = (
            f"Domínio {domain} já está na Cloudflare Ligbox ({match.account.label})."
        )
        if nameservers := wizard_nameservers(match.zone):
            payload["nameservers"] = nameservers
            payload["status"]["nameservers"] = nameservers
        return payload

    raise HTTPException(
        status_code=422,
        detail={
            "code": "zone_not_in_ligbox_accounts",
            "domain": domain,
            "message": (
                "Domínio não está nas contas Ligbox (ligit, itecnologys, ibytera). "
                "Use API Token da sua Cloudflare ou apontamentos no registrador."
            ),
            "paths_available": ["byo", "external"],
        },
    )

    cf = CloudflareDNS()
    try:
        activity_log.info(f"Criar zona Cloudflare: {domain}", source="cloudflare")
        cf.verify_token()
        result = cf.ensure_zone(domain)'''

if "zone_not_in_ligbox_accounts" not in text:
    text = text.replace(OLD_PROVISION_START, NEW_PROVISION_START)

OLD_APPLY = '''    if not settings.cloudflare_api_token:
        raise HTTPException(400, _cloudflare_token_missing_error())

    cf = CloudflareDNS()
    try:
        activity_log.info(f"Aplicar registos mail na Cloudflare: {domain}", source="cloudflare")
        cf.verify_token()
    except CloudflareError as e:
        activity_log.error(str(e), source="cloudflare")
        raise HTTPException(401, f"Token Cloudflare inválido: {e}") from e

    zone_id = body.zone_id
    if not zone_id:
        zone = cf.get_zone_by_name(domain)
        if not zone:
            activity_log.error(f"Zona {domain} não está na conta Ligbox", source="cloudflare")
            raise HTTPException(
                status_code=422,
                detail={
                    "code": "zone_not_in_account",
                    "domain": domain,
                    "message": f"Zona {domain} ainda não está na Cloudflare Ligbox.",
                    "use_portal_onboarding": True,
                },
            )
        zone_id = zone["id"]'''

NEW_APPLY = '''    sid = get_session_from_request(request)
    cf, zone_id, dns_mode, ligbox_acct = _cf_client_for_domain(domain, sid)
    if not cf or not zone_id:
        activity_log.error(f"Sem conta CF para {domain}", source="cloudflare")
        raise HTTPException(
            status_code=422,
            detail={
                "code": "zone_not_in_account",
                "domain": domain,
                "message": (
                    f"Zona {domain} não está nas contas Ligbox nem BYO. "
                    "Ligue BYO ou use registrador externo."
                ),
                "paths_available": ["byo", "external"],
            },
        )
    try:
        activity_log.info(
            f"Aplicar registos mail ({dns_mode or 'cf'}): {domain}",
            source="cloudflare",
        )
        cf.verify_token()
    except CloudflareError as e:
        activity_log.error(str(e), source="cloudflare")
        raise HTTPException(401, f"Token Cloudflare inválido: {e}") from e

    if body.zone_id:
        zone_id = body.zone_id'''

if "ligbox_acct" not in text:
    text = text.replace(OLD_APPLY, NEW_APPLY)

# apply return add dns_mode
OLD_APPLY_RET = '    return {"domain": domain, "zone_id": zone_id, "applied": applied, "verification": verification}'
NEW_APPLY_RET = '''    return {
        "domain": domain,
        "zone_id": zone_id,
        "dns_mode": dns_mode,
        "ligbox_account_id": ligbox_acct,
        "applied": applied,
        "verification": verification,
    }'''
if '"dns_mode": dns_mode' not in text:
    text = text.replace(OLD_APPLY_RET, NEW_APPLY_RET)

# dns instructions paths
OLD_PATHS = '''        "dns_paths": {
            "portal": "Trazer DNS para o portal (Cloudflare conta Ligbox)",
            "external": "Manter DNS no provedor atual (apontamentos manuais)",
        },'''
NEW_PATHS = '''        "dns_paths": {
            "ligbox": "Cloudflare Ligbox (ligit / itecnologys / ibytera) — apontamentos automáticos",
            "byo": "Sua conta Cloudflare (API Token)",
            "external": "Registrador / DNS externo (apontamentos manuais)",
        },
        "resolve": resolve_payload(domain),'''
if '"byo"' not in text.split("dns_paths")[1][:200]:
    text = text.replace(OLD_PATHS, NEW_PATHS)

ROUTER.write_text(text, encoding="utf-8")
print("onboarding.py patched OK")