038fb8f7ce
Entrega read-only de apontamentos DNS (Cloudflare, OpenPanel BIND, público) no Desk e Console /admin/dominio, com spec, scripts de rollback e patches VM112 para painel lateral no passo DNS do onboarding (deploy wizard pendente). Co-authored-by: Cursor <cursoragent@cursor.com>
138 lines
4.5 KiB
Python
138 lines
4.5 KiB
Python
"""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("/")
|
|
|
|
# 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 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": f"{OPENPANEL_URL}/domains/{domain}/dns",
|
|
"error": None if rows or bind_zone_responds(domain) else "Zona não encontrada no BIND OpenPanel",
|
|
}
|