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

"""OpenPanel BIND DNS records (read-only) — Spec 037 V2."""

from __future__ import annotations

import os
import re
from typing import Any

from app.cloudflare_dns import _classify_record, _record_belongs
from app.collectors.dns import _dig

OPENPANEL_BIND_HOST = os.getenv("OPENPANEL_BIND_HOST", "10.10.10.123")
OPENPANEL_PUBLIC_IP = os.getenv("OPENPANEL_PUBLIC_DNS_IP", "95.216.14.162")
OPENPANEL_URL = os.getenv("OPENPANEL_URL", "https://openpanel.ligbox.com.br").rstrip("/")


def _normalize_openadmin_public_url(url: str) -> str:
    """OpenAdmin via Traefik (443) — porta :2087 não é exposta publicamente."""
    url = (url or "").rstrip("/")
    if url.endswith(":2087"):
        return url[:-5]
    return url or "https://admin.openpanel.ligbox.com.br"


OPENADMIN_URL = _normalize_openadmin_public_url(
    os.getenv("OPENADMIN_URL", "https://admin.openpanel.ligbox.com.br")
)

# Ligbox authoritative BIND — common public NS hostnames
OPENPANEL_NS_HINTS = ("openpanel", "ligbox", "itecnologys")


def _dig_bind(name: str, rtype: str) -> list[str]:
    host = OPENPANEL_BIND_HOST.strip()
    if not host:
        return []
    try:
        import subprocess

        proc = subprocess.run(
            ["dig", f"@{host}", "+short", name, rtype],
            capture_output=True,
            text=True,
            timeout=8,
        )
        if proc.returncode == 0 and proc.stdout.strip():
            return [ln.strip().strip('"') for ln in proc.stdout.splitlines() if ln.strip()]
    except Exception:
        pass
    return []


def _parse_mx(line: str) -> tuple[int | None, str]:
    m = re.match(r"(\d+)\s+(.+)", line.strip())
    if m:
        return int(m.group(1)), m.group(2).rstrip(".")
    return None, line.strip().rstrip(".")


def _names_to_query(domain: str) -> list[str]:
    domain = domain.lower().strip().rstrip(".")
    names = {domain, f"mail.{domain}", f"www.{domain}", f"_dmarc.{domain}", f"default._domainkey.{domain}"}
    return sorted(names)


def ns_points_to_openpanel(domain: str) -> bool:
    for ns in _dig(domain, "NS"):
        ns_l = ns.lower().rstrip(".")
        if any(h in ns_l for h in OPENPANEL_NS_HINTS):
            return True
        for a in _dig(ns_l, "A"):
            if a == OPENPANEL_PUBLIC_IP or a == OPENPANEL_BIND_HOST:
                return True
    return False


def openpanel_admin_dns_edit_url(domain: str) -> str:
    """OpenAdmin DNS Zone Editor — vista staff (Spec 037)."""
    domain = domain.lower().strip().rstrip(".")
    return f"{OPENADMIN_URL}/domains/dns?domain={domain}"


def openpanel_client_dns_edit_url(domain: str) -> str:
    """Painel cliente OpenPanel — só gerente dono do domínio (nunca staff suporte)."""
    domain = domain.lower().strip().rstrip(".")
    return f"{OPENPANEL_URL}/domains/{domain}/dns"


def bind_zone_responds(domain: str) -> bool:
    host = OPENPANEL_BIND_HOST
    if not host:
        return False
    try:
        import subprocess

        proc = subprocess.run(
            ["dig", f"@{host}", "+norecurse", domain, "SOA", "+short"],
            capture_output=True,
            text=True,
            timeout=8,
        )
        return proc.returncode == 0 and bool(proc.stdout.strip())
    except Exception:
        return False


def fetch_openpanel_bind_records(domain: str) -> dict[str, Any]:
    """List records from OpenPanel BIND via dig @ bind host (read-only)."""
    domain = domain.lower().strip().rstrip(".")
    rows: list[dict[str, Any]] = []
    seen: set[tuple[str, str, str]] = set()

    def add(name: str, rtype: str, content: str, priority: int | None = None) -> None:
        name = name.rstrip(".").lower()
        content = content.strip().strip('"')
        key = (name, rtype, content)
        if not content or key in seen:
            return
        if not _record_belongs(name, domain):
            return
        seen.add(key)
        purpose = _classify_record(name, rtype, content)
        rows.append(
            {
                "source": "openpanel_bind",
                "status": "applied",
                "type": rtype,
                "name": name,
                "content": content,
                "priority": priority,
                "ttl": None,
                "purpose": purpose,
                "email_related": purpose
                in ("mx", "spf", "dkim", "dmarc", "mail-host", "autodiscover", "mail-alias"),
            }
        )

    for name in _names_to_query(domain):
        for rtype in ("A", "AAAA", "CNAME", "TXT"):
            for line in _dig_bind(name, rtype):
                if rtype == "CNAME" and line.endswith("."):
                    line = line.rstrip(".")
                add(name, rtype, line)

        for line in _dig_bind(name, "MX"):
            prio, target = _parse_mx(line)
            add(name, "MX", target, priority=prio)

    rows.sort(key=lambda r: (0 if r["email_related"] else 1, r["type"], r["name"]))

    return {
        "domain": domain,
        "bind_host": OPENPANEL_BIND_HOST,
        "zone_responds": bind_zone_responds(domain),
        "ns_on_openpanel": ns_points_to_openpanel(domain),
        "records": rows,
        "summary": {
            "total": len(rows),
            "email_related": sum(1 for r in rows if r.get("email_related")),
        },
        "edit_url": openpanel_admin_dns_edit_url(domain),
        "edit_url_client": openpanel_client_dns_edit_url(domain),
        "error": None if rows or bind_zone_responds(domain) else "Zona não encontrada no BIND OpenPanel",
    }