ligbox-ops-platform/projects/ops-desk/api/app/dns_viewer.py

"""Unified DNS viewer (read-only) — Spec 037-DNS-VIEWER."""

from __future__ import annotations

import os
import re
from typing import Any

import httpx

from app.cloudflare_dns import fetch_domain_dns
from app.collectors.dns import _dig, collect
from app.openpanel_dns import (
    fetch_openpanel_bind_records,
    ns_points_to_openpanel,
)

VM112_API = os.getenv("VM112_API_URL", "http://10.10.10.112:8090")

CF_NS_SUFFIX = ".ns.cloudflare.com"

MODE_LABELS: dict[str, str] = {
    "ligbox_cf_ligit": "Cloudflare Ligbox (ligit)",
    "ligbox_cf_itecnologys": "Cloudflare Ligbox (itecnologys)",
    "ligbox_cf_ibytera": "Cloudflare Ligbox (ibytera)",
    "ligbox_cf": "Cloudflare Ligbox",
    "ligbox_cf_provision_pending": "DNS Ligbox (aguarda NS)",
    "byo_cf": "Cloudflare cliente (BYO)",
    "external": "DNS externo / registrador",
    "registrar": "DNS externo / registrador",
    "openpanel_bind": "OpenPanel BIND",
    "unknown": "A determinar",
}


def _ns_list(domain: str) -> list[str]:
    lines = _dig(domain, "NS")
    return [re.sub(r"\.$", "", ln.lower()) for ln in lines if ln.strip()]


def _ligbox_cf_ns() -> list[str]:
    return []


def _ns_match_cloudflare(ns: list[str]) -> bool:
    return any(n.endswith(CF_NS_SUFFIX) for n in ns)


def _public_checks(domain: str) -> dict[str, Any]:
    raw = collect(domain)
    out: dict[str, Any] = {}
    for key, check_id in (
        ("mx", "dns_mx"),
        ("spf", "dns_spf"),
        ("dkim", "dns_dkim"),
        ("dmarc", "dns_dmarc"),
    ):
        item = raw.get(check_id, {})
        status = item.get("status", "fail")
        out[key] = {
            "ok": status == "pass",
            "warn": status == "warn",
            "values": (item.get("evidence") or {}).get("records") or [],
            "hint": item.get("message") if status != "pass" else None,
        }
    return out


def _records_from_public(domain: str, checks: dict[str, Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    mx_vals = checks.get("mx", {}).get("values") or []
    for line in mx_vals[:5]:
        m = re.match(r"(\d+)\s+(.+)", line.strip())
        if m:
            rows.append(
                {
                    "source": "public_resolver",
                    "status": "actual",
                    "type": "MX",
                    "name": domain,
                    "content": m.group(2).rstrip("."),
                    "priority": int(m.group(1)),
                    "ttl": None,
                    "purpose": "mx",
                    "email_related": True,
                }
            )
    for txt in (checks.get("spf", {}).get("values") or [])[:2]:
        rows.append(
            {
                "source": "public_resolver",
                "status": "actual",
                "type": "TXT",
                "name": domain,
                "content": txt,
                "priority": None,
                "ttl": None,
                "purpose": "spf",
                "email_related": True,
            }
        )
    for txt in (checks.get("dkim", {}).get("values") or [])[:1]:
        rows.append(
            {
                "source": "public_resolver",
                "status": "actual",
                "type": "TXT",
                "name": f"default._domainkey.{domain}",
                "content": txt[:200],
                "priority": None,
                "ttl": None,
                "purpose": "dkim",
                "email_related": True,
            }
        )
    for txt in (checks.get("dmarc", {}).get("values") or [])[:1]:
        rows.append(
            {
                "source": "public_resolver",
                "status": "actual",
                "type": "TXT",
                "name": f"_dmarc.{domain}",
                "content": txt[:200],
                "priority": None,
                "ttl": None,
                "purpose": "dmarc",
                "email_related": True,
            }
        )
    a_mail = _dig(f"mail.{domain}", "A")
    for ip in a_mail[:2]:
        rows.append(
            {
                "source": "public_resolver",
                "status": "actual",
                "type": "A",
                "name": f"mail.{domain}",
                "content": ip,
                "priority": None,
                "ttl": None,
                "purpose": "mail-host",
                "email_related": True,
            }
        )
    return rows


async def _wizard_resolve(domain: str) -> dict[str, Any] | None:
    try:
        async with httpx.AsyncClient(timeout=6.0) as client:
            res = await client.get(f"{VM112_API}/api/onboarding/dns/resolve/{domain}")
            if res.status_code == 200:
                data = res.json()
                if isinstance(data, dict):
                    return data
    except Exception:
        pass
    return None


def _infer_mode(
    resolve: dict[str, Any] | None,
    cf_zone: dict | None,
    public_ns: list[str],
    *,
    openpanel_zone: bool = False,
    openpanel_ns: bool = False,
) -> tuple[str, str, str]:
    if openpanel_zone or openpanel_ns:
        return "openpanel_bind", "applied", "openpanel_bind"
    if resolve:
        mode = str(resolve.get("dns_mode") or "unknown")
        if mode.startswith("ligbox_cf"):
            if cf_zone and public_ns and not _ns_match_cloudflare(public_ns):
                return "ligbox_cf_provision_pending", "planned", "cf_ligbox"
            return mode if mode != "unknown" else "ligbox_cf", "applied", "cf_ligbox"
        if mode in ("byo_cf", "external", "registrar", "openpanel_bind"):
            display = "actual" if mode != "byo_cf" else "applied"
            src = "cf_byo" if mode == "byo_cf" else "public_resolver"
            return mode, display, src
    if cf_zone:
        if public_ns and not _ns_match_cloudflare(public_ns):
            return "ligbox_cf_provision_pending", "planned", "cf_ligbox"
        return "ligbox_cf", "applied", "cf_ligbox"
    if public_ns:
        return "external", "actual", "public_resolver"
    return "unknown", "actual", "public_resolver"


def _edit_links(
    domain: str,
    dns_mode: str,
    zone: dict | None,
    role: str,
    authoritative: str = "",
) -> list[dict[str, Any]]:
    from app.permissions import can_open_cloudflare_dns_link

    links: list[dict[str, Any]] = []
    if can_open_cloudflare_dns_link(role) and dns_mode.startswith("ligbox_cf"):
        zone_name = (zone or {}).get("name") or domain
        links.append(
            {
                "label": "Editar na Cloudflare (Ligbox)",
                "provider": "cf_ligbox",
                "url": f"https://dash.cloudflare.com/?to=/:account/{zone_name}/dns",
                "roles": ["super_admin", "ops_lead", "devops", "seo"],
            }
        )
    if dns_mode in ("external", "registrar", "unknown"):
        links.append(
            {
                "label": "Registro.br / registrador",
                "provider": "registrar",
                "url": "https://registro.br",
                "roles": ["super_admin", "ops_lead", "technician", "seo"],
            }
        )
    if dns_mode == "openpanel_bind" or authoritative == "openpanel_bind":
        from app.permissions import can_open_openpanel_link

        if can_open_openpanel_link(role):
            from app.openpanel_dns import OPENPANEL_URL

            links.append(
                {
                    "label": "Editar no OpenPanel",
                    "provider": "openpanel_bind",
                    "url": f"{OPENPANEL_URL}/domains/{domain}/dns",
                    "roles": ["super_admin", "ops_lead", "sales_admin", "seo"],
                }
            )
    return links


def _mode_message(display_mode: str, dns_mode: str) -> str:
    if dns_mode.startswith("ligbox_cf") and display_mode == "planned":
        return (
            "Estes apontamentos serão configurados na Cloudflare Ligbox "
            "após delegar os nameservers."
        )
    if dns_mode.startswith("ligbox_cf"):
        return "Apontamentos activos ou geridos na Cloudflare Ligbox."
    if dns_mode in ("external", "registrar"):
        return "O domínio usa DNS fora da Ligbox. Abaixo: resolução pública actual."
    if dns_mode == "openpanel_bind":
        return "Zona servida pelo DNS Ligbox (OpenPanel BIND)."
    return "Origem DNS em análise — verifique nameservers e zona."


async def fetch_dns_viewer(
    domain: str,
    *,
    role: str,
    email_service: bool | None = None,
    include_public: bool = True,
) -> dict[str, Any]:
    domain = domain.lower().strip().rstrip(".")
    errors: list[str] = []

    resolve = await _wizard_resolve(domain)
    cf_payload = await fetch_domain_dns(domain, email_service=email_service)
    cf_zone = cf_payload.get("zone")
    public_ns = _ns_list(domain)
    op_bind = fetch_openpanel_bind_records(domain)
    op_records = op_bind.get("records") or []
    op_zone = bool(op_bind.get("zone_responds") or op_records)
    op_ns = bool(op_bind.get("ns_on_openpanel") or ns_points_to_openpanel(domain))

    ligbox_ns = []
    if resolve and resolve.get("matched"):
        ligbox_ns = resolve.get("nameservers") or []

    dns_mode, display_mode, authoritative = _infer_mode(
        resolve, cf_zone, public_ns, openpanel_zone=op_zone, openpanel_ns=op_ns
    )
    public_checks = _public_checks(domain) if include_public else {}

    records: list[dict[str, Any]] = []
    planned_records: list[dict[str, Any]] = []

    if authoritative == "openpanel_bind" and op_records:
        records = list(op_records)
        if op_bind.get("error") and op_bind["error"] not in errors:
            errors.append(op_bind["error"])
    elif display_mode == "actual" and not cf_zone:
        records = _records_from_public(domain, public_checks)
        authoritative = "public_resolver"
    elif cf_payload.get("records"):
        for r in cf_payload["records"]:
            records.append(
                {
                    **r,
                    "source": "cf_ligbox",
                    "status": "applied" if display_mode == "applied" else "planned",
                }
            )
    elif display_mode == "actual":
        records = _records_from_public(domain, public_checks)

    if not records and op_records and not cf_zone:
        records = list(op_records)
        dns_mode = "openpanel_bind"
        authoritative = "openpanel_bind"
        display_mode = "applied"

    if display_mode == "planned" and cf_payload.get("records"):
        planned_records = [{**r, "source": "planned_ligbox", "status": "planned"} for r in cf_payload["records"]]

    cf_ns_expected = ligbox_ns or ([n for n in public_ns if n.endswith(CF_NS_SUFFIX)] if _ns_match_cloudflare(public_ns) else [])

    return {
        "domain": domain,
        "dns_mode": dns_mode,
        "mode_label": MODE_LABELS.get(dns_mode, dns_mode),
        "display_mode": display_mode,
        "authoritative_source": authoritative,
        "mode_message": _mode_message(display_mode, dns_mode),
        "nameservers": {
            "current_public": public_ns,
            "ligbox_cloudflare": cf_ns_expected,
            "match_ligbox": _ns_match_cloudflare(public_ns) if public_ns else False,
        },
        "records": records,
        "planned_records": planned_records,
        "public_checks": public_checks,
        "edit_links": _edit_links(domain, dns_mode, cf_zone, role, authoritative),
        "instructions": None,
        "resolve": resolve,
        "openpanel": {
            "bind_host": op_bind.get("bind_host"),
            "zone_responds": op_bind.get("zone_responds"),
            "ns_on_openpanel": op_bind.get("ns_on_openpanel"),
            "edit_url": op_bind.get("edit_url"),
        },
        "zone": cf_zone,
        "email_service": cf_payload.get("email_service"),
        "summary": {
            "total": len(records),
            "planned": len(planned_records),
            "email_related": sum(1 for r in records if r.get("email_related")),
        },
        "errors": errors + ([cf_payload["error"]] if cf_payload.get("error") else []),
        "legacy_cf": cf_payload,
    }