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>
56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Spec 037 V4 — GET /onboarding/dns/viewer/{domain} na VM112. Idempotente."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
ROUTER = Path("/opt/ligbox-wizard/backend/app/routers/onboarding.py")
|
|
text = ROUTER.read_text(encoding="utf-8")
|
|
|
|
VIEWER_ENDPOINT = '''
|
|
|
|
@router.get("/dns/viewer/{domain}")
|
|
async def dns_viewer_onboarding(domain: str, request: Request):
|
|
"""Painel DNS read-only — proxy Desk (Spec 037 V4). Sem edit_links para cliente."""
|
|
get_session_from_request(request)
|
|
domain = normalize_domain(domain)
|
|
try:
|
|
validate_primary_domain(domain)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=invalid_domain_http_detail()) from e
|
|
|
|
import httpx
|
|
|
|
desk = os.getenv("DESK_API_URL", "http://10.10.10.122:8080").rstrip("/")
|
|
token = os.getenv("OPS_INTERNAL_TOKEN", "")
|
|
headers = {"Accept": "application/json"}
|
|
if token:
|
|
headers["X-Ops-Internal-Token"] = token
|
|
url = f"{desk}/api/v1/dns/viewer/{domain}?email_service=true&include_public=true"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
res = await client.get(url, headers=headers)
|
|
if res.status_code >= 400:
|
|
raise HTTPException(res.status_code, res.text[:500])
|
|
payload = res.json()
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(502, f"Desk DNS viewer indisponível: {e}") from e
|
|
|
|
payload["edit_links"] = []
|
|
payload["viewer_context"] = "wizard_onboarding"
|
|
return payload
|
|
|
|
'''
|
|
|
|
if "/dns/viewer/" not in text:
|
|
anchor = '@router.get("/dns/resolve/{domain}")'
|
|
if anchor not in text:
|
|
anchor = '@router.get("/dns/instructions/{domain}")'
|
|
text = text.replace(anchor, VIEWER_ENDPOINT + anchor)
|
|
ROUTER.write_text(text, encoding="utf-8")
|
|
print("onboarding.py: dns/viewer endpoint added")
|
|
else:
|
|
print("onboarding.py: dns/viewer already present")
|