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>
316 lines
9.6 KiB
Python
316 lines
9.6 KiB
Python
"""Sandbox Área Gerente — produção create-only, nunca apaga existentes (Spec 035)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import secrets
|
|
import sqlite3
|
|
import uuid
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app import domain_console_sandbox_store, vm112_domains
|
|
|
|
VM112_API = os.getenv("VM112_API_URL", "http://10.10.10.112:8090")
|
|
SANDBOX_DOMAIN_PREFIX = os.getenv("DOMAIN_CONSOLE_SANDBOX_PREFIX", "cenario-")
|
|
SANDBOX_ALLOWED_SUFFIX = os.getenv(
|
|
"DOMAIN_CONSOLE_SANDBOX_SUFFIX",
|
|
".ops.ligbox.com.br",
|
|
)
|
|
SANDBOX_ENABLED = os.getenv("DOMAIN_CONSOLE_SANDBOX", "1") == "1"
|
|
|
|
PROTECTED_DOMAINS = vm112_domains.PURGE_BLOCKLIST | frozenset(
|
|
{"ligbox.com.br", "itecnologys.com", "ligbox.com", "myvexx.com"}
|
|
)
|
|
|
|
_SCENARIO_DOMAIN_RE = re.compile(
|
|
rf"^{re.escape(SANDBOX_DOMAIN_PREFIX)}[a-z0-9]([a-z0-9-]{{0,61}}[a-z0-9])?{re.escape(SANDBOX_ALLOWED_SUFFIX)}$",
|
|
re.I,
|
|
)
|
|
|
|
|
|
class SandboxError(ValueError):
|
|
pass
|
|
|
|
|
|
class SandboxForbidden(PermissionError):
|
|
pass
|
|
|
|
|
|
def assert_enabled() -> None:
|
|
if not SANDBOX_ENABLED:
|
|
raise SandboxForbidden("DOMAIN_CONSOLE_SANDBOX desactivado")
|
|
|
|
|
|
def assert_no_delete(operation: str) -> None:
|
|
raise SandboxForbidden(
|
|
f"Sandbox create-only: operação «{operation}» proibida — nada existente pode ser apagado"
|
|
)
|
|
|
|
|
|
def is_scenario_domain(domain: str) -> bool:
|
|
return bool(_SCENARIO_DOMAIN_RE.match(domain.lower().strip()))
|
|
|
|
|
|
def suggest_scenario_domain(slug: str | None = None) -> str:
|
|
slug = (slug or uuid.uuid4().hex[:8]).lower()
|
|
slug = re.sub(r"[^a-z0-9-]", "-", slug).strip("-") or uuid.uuid4().hex[:8]
|
|
return f"{SANDBOX_DOMAIN_PREFIX}{slug}{SANDBOX_ALLOWED_SUFFIX}"
|
|
|
|
|
|
def _vm112_onboarding_headers(session_id: str) -> dict[str, str]:
|
|
return {"X-Onboarding-Session": session_id, "Content-Type": "application/json"}
|
|
|
|
|
|
def _extract_account_emails(domain_data: dict[str, Any]) -> list[str]:
|
|
emails: list[str] = []
|
|
for key in ("accounts", "accounts_preview", "emails"):
|
|
raw = domain_data.get(key)
|
|
if isinstance(raw, list):
|
|
for item in raw:
|
|
if isinstance(item, str) and "@" in item:
|
|
emails.append(item.lower())
|
|
elif isinstance(item, dict):
|
|
e = item.get("email") or item.get("name")
|
|
if e and "@" in str(e):
|
|
emails.append(str(e).lower())
|
|
admin = domain_data.get("admin_email") or domain_data.get("portal_admin")
|
|
if admin and "@" in str(admin):
|
|
emails.append(str(admin).lower())
|
|
return sorted(set(emails))
|
|
|
|
|
|
def fetch_domain_readonly(domain: str) -> dict[str, Any]:
|
|
domain = domain.lower().strip()
|
|
try:
|
|
data = vm112_domains.get_domain(domain)
|
|
return {"ok": True, "source": "vm112", "domain": domain, "data": data}
|
|
except httpx.HTTPError as exc:
|
|
return {
|
|
"ok": False,
|
|
"source": "vm112",
|
|
"domain": domain,
|
|
"error": str(exc),
|
|
"data": None,
|
|
}
|
|
|
|
|
|
def list_domains_readonly(q: str = "") -> dict[str, Any]:
|
|
try:
|
|
data = vm112_domains.list_domains(q)
|
|
return {"ok": True, "source": "vm112", "data": data}
|
|
except httpx.HTTPError as exc:
|
|
return {"ok": False, "source": "vm112", "error": str(exc), "data": None}
|
|
|
|
|
|
def _create_account_vm112(
|
|
*,
|
|
domain: str,
|
|
local_part: str,
|
|
password: str,
|
|
display_name: str,
|
|
) -> dict[str, Any]:
|
|
session_id = f"sandbox-{uuid.uuid4().hex}"
|
|
body = {
|
|
"domain": domain,
|
|
"local_part": local_part,
|
|
"password": password,
|
|
"display_name": display_name,
|
|
"use_server_password": False,
|
|
"send_welcome": False,
|
|
"dns_mode": "sandbox",
|
|
}
|
|
with httpx.Client(timeout=180.0) as client:
|
|
r = client.post(
|
|
f"{VM112_API}/api/onboarding/account/create",
|
|
json=body,
|
|
headers=_vm112_onboarding_headers(session_id),
|
|
)
|
|
if r.status_code >= 400:
|
|
try:
|
|
detail = r.json()
|
|
except Exception:
|
|
detail = r.text
|
|
raise SandboxError(f"VM112 account/create HTTP {r.status_code}: {detail}")
|
|
return r.json()
|
|
|
|
|
|
def create_scenario(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
label: str,
|
|
domain: str | None,
|
|
slug: str | None,
|
|
created_by: str,
|
|
plan_mock: str = "Business",
|
|
admin_password: str | None = None,
|
|
) -> dict[str, Any]:
|
|
assert_enabled()
|
|
domain = (domain or suggest_scenario_domain(slug)).lower().strip()
|
|
|
|
if domain in PROTECTED_DOMAINS:
|
|
raise SandboxForbidden(f"Domínio protegido: {domain}")
|
|
|
|
if not is_scenario_domain(domain):
|
|
raise SandboxForbidden(
|
|
f"Sandbox só cria domínios novos com prefixo «{SANDBOX_DOMAIN_PREFIX}» "
|
|
f"e sufixo «{SANDBOX_ALLOWED_SUFFIX}». Sugestão: {suggest_scenario_domain(slug)}"
|
|
)
|
|
|
|
existing = fetch_domain_readonly(domain)
|
|
baseline: list[str] = []
|
|
if existing.get("ok") and existing.get("data"):
|
|
baseline = _extract_account_emails(existing["data"])
|
|
if baseline:
|
|
raise SandboxForbidden(
|
|
f"Domínio {domain} já tem contas em produção — sandbox não altera existentes. "
|
|
"Use outro nome de cenário."
|
|
)
|
|
|
|
password = admin_password or secrets.token_urlsafe(12)
|
|
if len(password) < 8:
|
|
raise SandboxError("Senha admin mínimo 8 caracteres")
|
|
|
|
vm112_result = _create_account_vm112(
|
|
domain=domain,
|
|
local_part="admin",
|
|
password=password,
|
|
display_name=f"Gerente {label or domain}",
|
|
)
|
|
|
|
scenario = domain_console_sandbox_store.create_scenario(
|
|
conn,
|
|
domain=domain,
|
|
label=label or domain,
|
|
baseline_accounts=baseline,
|
|
created_by=created_by,
|
|
plan_mock=plan_mock,
|
|
)
|
|
|
|
admin_email = f"admin@{domain}"
|
|
return {
|
|
"scenario": scenario,
|
|
"vm112": vm112_result,
|
|
"credentials": {
|
|
"admin_email": admin_email,
|
|
"password": password,
|
|
"webmail": f"https://mail.{domain}/",
|
|
"console": "https://onboard.ligbox.com.br/admin",
|
|
},
|
|
"guards": {
|
|
"delete_blocked": True,
|
|
"baseline_protected": baseline,
|
|
"mode": "create-only",
|
|
},
|
|
}
|
|
|
|
|
|
def add_account_to_scenario(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
scenario_id: str,
|
|
local_part: str,
|
|
display_name: str,
|
|
password: str | None = None,
|
|
) -> dict[str, Any]:
|
|
assert_enabled()
|
|
scenario = domain_console_sandbox_store.get_scenario(conn, scenario_id)
|
|
if not scenario:
|
|
raise SandboxError("Cenário não encontrado")
|
|
|
|
domain = scenario["domain"]
|
|
local_part = re.sub(r"[^a-z0-9._+-]", "", local_part.lower().strip())
|
|
if not local_part:
|
|
raise SandboxError("local_part inválido")
|
|
|
|
email = f"{local_part}@{domain}"
|
|
baseline = set(scenario["baseline_accounts"])
|
|
if email in baseline:
|
|
raise SandboxForbidden(
|
|
f"{email} já existia antes do cenário — sandbox não sobrescreve contas existentes"
|
|
)
|
|
|
|
if email in scenario["created_accounts"]:
|
|
raise SandboxError(f"{email} já foi criada neste cenário")
|
|
|
|
pwd = password or secrets.token_urlsafe(10)
|
|
if len(pwd) < 8:
|
|
raise SandboxError("Senha mínimo 8 caracteres")
|
|
|
|
vm112_result = _create_account_vm112(
|
|
domain=domain,
|
|
local_part=local_part,
|
|
password=pwd,
|
|
display_name=display_name or local_part,
|
|
)
|
|
|
|
updated = domain_console_sandbox_store.append_created_account(conn, scenario_id, email)
|
|
return {
|
|
"scenario": updated,
|
|
"account": {
|
|
"email": email,
|
|
"password": pwd,
|
|
"display_name": display_name or local_part,
|
|
},
|
|
"vm112": vm112_result,
|
|
"guards": {"delete_blocked": True, "mode": "create-only"},
|
|
}
|
|
|
|
|
|
def build_console_state(conn: sqlite3.Connection, scenario_id: str) -> dict[str, Any]:
|
|
"""Estado UI — merge VM112 read + cenário local."""
|
|
scenario = domain_console_sandbox_store.get_scenario(conn, scenario_id)
|
|
if not scenario:
|
|
raise SandboxError("Cenário não encontrado")
|
|
|
|
domain = scenario["domain"]
|
|
live = fetch_domain_readonly(domain)
|
|
live_emails = _extract_account_emails(live["data"]) if live.get("data") else []
|
|
|
|
accounts = []
|
|
seen = set()
|
|
for email in live_emails or scenario["created_accounts"]:
|
|
if email in seen:
|
|
continue
|
|
seen.add(email)
|
|
is_baseline = email in scenario["baseline_accounts"]
|
|
is_created = email in scenario["created_accounts"]
|
|
accounts.append(
|
|
{
|
|
"email": email,
|
|
"name": email.split("@")[0].title(),
|
|
"mail_gb": 30,
|
|
"files_gb": 200,
|
|
"nc": True,
|
|
"active": True,
|
|
"protected": is_baseline,
|
|
"sandbox_created": is_created,
|
|
"can_delete": False,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"mode": "live-sandbox",
|
|
"sandbox": True,
|
|
"delete_blocked": True,
|
|
"scenario": scenario,
|
|
"domain": domain,
|
|
"plan": {
|
|
"name": scenario["plan_mock"],
|
|
"price": 549,
|
|
"maxSeats": 25,
|
|
"mailGbDefault": 30,
|
|
"filesGbDefault": 200,
|
|
},
|
|
"billing": {"status": "Sandbox", "nextRenewal": "—"},
|
|
"dmarc": {"spf": True, "dkim": True, "dmarc": True, "score": 95},
|
|
"dns": {
|
|
"mx": live.get("ok", False),
|
|
"aMail": live.get("ok", False),
|
|
"subdomain": f"intranet.{domain}",
|
|
},
|
|
"accounts": accounts,
|
|
"vm112_live": live,
|
|
}
|