edffd8b3c0
Cache VM112/VM122 para lista e detalhe domínio, modal purge com loading animado, cards Escopo OPS clicáveis (camada + Spec + navegação), blocklist visível na UI, e documentação nas specs e anais de referência 20260625. Co-authored-by: Cursor <cursoragent@cursor.com>
485 lines
16 KiB
Python
485 lines
16 KiB
Python
"""Provisionamento local via zmprov (Carbonio)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
from app.config import settings
|
||
|
||
NGINX_WEB_HTTPS = Path("/opt/zextras/conf/nginx/includes/nginx.conf.web.https")
|
||
ZMPROXYCONFGEN = "/opt/zextras/libexec/zmproxyconfgen"
|
||
from app.services import activity_log, carbonio_cache
|
||
|
||
EMAIL_RE = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
|
||
_NO_SUCH_DOMAIN = "NO_SUCH_DOMAIN"
|
||
_NO_SUCH_ACCOUNT = "NO_SUCH_ACCOUNT"
|
||
|
||
|
||
class CarbonioError(Exception):
|
||
pass
|
||
|
||
|
||
def _zmprov_run(*args: str, log_cmd: bool = True) -> tuple[int, str, str]:
|
||
cmd = ["sudo", "-u", settings.zextras_user, settings.zmprov, *args]
|
||
if log_cmd:
|
||
activity_log.cmd(" ".join(cmd), source="vm112")
|
||
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
|
||
return proc.returncode, (proc.stdout or "").strip(), (proc.stderr or "").strip()
|
||
|
||
|
||
def _run_zmprov(*args: str) -> str:
|
||
"""Comando zmprov que deve ter sucesso (criar domínio/conta, etc.)."""
|
||
code, out, err = _zmprov_run(*args, log_cmd=True)
|
||
if code != 0:
|
||
msg = err or out or "zmprov failed"
|
||
activity_log.error(f"zmprov falhou: {msg}", source="vm112")
|
||
raise CarbonioError(msg)
|
||
if out:
|
||
activity_log.ok(out[:500], source="vm112")
|
||
return out
|
||
|
||
|
||
def _is_missing_domain(stderr: str, stdout: str) -> bool:
|
||
blob = f"{stderr}\n{stdout}"
|
||
return _NO_SUCH_DOMAIN in blob
|
||
|
||
|
||
def _is_missing_account(stderr: str, stdout: str) -> bool:
|
||
blob = f"{stderr}\n{stdout}"
|
||
return _NO_SUCH_ACCOUNT in blob or _NO_SUCH_DOMAIN in blob
|
||
|
||
|
||
def _is_account_exists(stderr: str, stdout: str) -> bool:
|
||
blob = f"{stderr}\n{stdout}"
|
||
return "ACCOUNT_EXISTS" in blob
|
||
|
||
|
||
def _carbonio_unreachable(err: str, out: str) -> bool:
|
||
blob = f"{err}\n{out}".lower()
|
||
return "connection refused" in blob or "io_error" in blob
|
||
|
||
|
||
def domain_exists(domain: str, *, use_cache: bool = True) -> bool:
|
||
"""Verificação silenciosa — domínio inexistente NÃO é erro de onboarding."""
|
||
domain = domain.lower().strip()
|
||
cache_key = f"domain_exists:{domain}"
|
||
if use_cache:
|
||
cached = carbonio_cache.get(cache_key)
|
||
if cached is not None:
|
||
return bool(cached)
|
||
code, out, err = _zmprov_run("gd", domain, log_cmd=False)
|
||
if code == 0:
|
||
exists = f"# name {domain}" in out or f"name {domain}" in out
|
||
elif _is_missing_domain(err, out):
|
||
exists = False
|
||
elif _carbonio_unreachable(err, out):
|
||
exists = False
|
||
else:
|
||
activity_log.warn(f"zmprov gd {domain}: {err or out}", source="vm112")
|
||
exists = False
|
||
if use_cache:
|
||
carbonio_cache.set(cache_key, exists, carbonio_cache.TTL_DOMAIN_EXISTS)
|
||
return exists
|
||
|
||
|
||
def list_all_domains(*, use_cache: bool = True) -> list[str]:
|
||
"""zmprov gad (~5s) — cache TTL curto para Serviços IaaS / Desk."""
|
||
cache_key = carbonio_cache.CACHE_KEY_ALL_DOMAINS
|
||
if use_cache:
|
||
cached = carbonio_cache.get(cache_key)
|
||
if cached is not None:
|
||
return list(cached)
|
||
code, out, _err = _zmprov_run("gad", log_cmd=False)
|
||
if code != 0:
|
||
return []
|
||
domains = [ln.strip().lower() for ln in out.splitlines() if ln.strip()]
|
||
if use_cache:
|
||
carbonio_cache.set(cache_key, domains, carbonio_cache.TTL_ALL_DOMAINS)
|
||
return domains
|
||
|
||
|
||
def invalidate_domain_list_cache() -> None:
|
||
carbonio_cache.invalidate_all_domains()
|
||
|
||
|
||
def set_domain_public_hostname(domain: str) -> str:
|
||
"""Webmail mail.{domínio} — evita redirect para mail.ligbox.com.br."""
|
||
domain = domain.lower().strip()
|
||
mail_host = f"mail.{domain}"
|
||
logo = f"https://{mail_host}/public/logos/ligbox-logo.png"
|
||
admin = f"admin@{domain}"
|
||
_run_zmprov(
|
||
"md",
|
||
domain,
|
||
"zimbraPublicServiceHostname",
|
||
mail_host,
|
||
"zimbraVirtualHostname",
|
||
mail_host,
|
||
"zimbraPublicServiceProtocol",
|
||
"https",
|
||
"zimbraPublicServicePort",
|
||
"443",
|
||
"zimbraSkinLogoURL",
|
||
logo,
|
||
"carbonioLogoUrl",
|
||
logo,
|
||
"carbonioNotificationFrom",
|
||
admin,
|
||
"carbonioNotificationRecipients",
|
||
admin,
|
||
)
|
||
return mail_host
|
||
|
||
|
||
def add_mail_alias_hostnames(domain: str, hostnames: list[str]) -> None:
|
||
"""Adiciona hostnames virtuais (alias webmail) ao domínio Carbonio."""
|
||
domain = domain.lower().strip()
|
||
primary_mail = f"mail.{domain}"
|
||
for host in hostnames:
|
||
h = host.lower().strip().rstrip(".")
|
||
if not h or h == primary_mail:
|
||
continue
|
||
try:
|
||
_run_zmprov("md", domain, "zimbraVirtualHostname", h)
|
||
activity_log.ok(f"Alias hostname Carbonio: {h}", source="vm112")
|
||
except CarbonioError as e:
|
||
activity_log.warn(f"Alias {h}: {e}", source="vm112")
|
||
|
||
|
||
def create_domain(domain: str) -> str:
|
||
domain = domain.lower().strip()
|
||
if domain_exists(domain):
|
||
activity_log.info(f"Domínio {domain} já existe no Carbonio", source="vm112")
|
||
set_domain_public_hostname(domain)
|
||
return f"domain {domain} already exists"
|
||
activity_log.info(f"Criar domínio no Carbonio: {domain}", source="vm112")
|
||
out = _run_zmprov("cd", domain)
|
||
set_domain_public_hostname(domain)
|
||
invalidate_domain_list_cache()
|
||
activity_log.ok(f"Webmail: https://mail.{domain}/", source="vm112")
|
||
return out
|
||
|
||
|
||
def create_account(email: str, password: str, display_name: str | None = None) -> str:
|
||
return create_account_full(
|
||
email,
|
||
password,
|
||
display_name=display_name,
|
||
quota_mb=0,
|
||
must_change_password=False,
|
||
)
|
||
|
||
|
||
def ensure_onboarding_account(
|
||
email: str,
|
||
password: str,
|
||
display_name: str | None = None,
|
||
) -> tuple[str, bool]:
|
||
"""
|
||
Cria conta admin no onboarding ou reutiliza se já existir (retry do wizard).
|
||
Returns (message, reused).
|
||
"""
|
||
email = email.lower().strip()
|
||
domain = email.split("@", 1)[1]
|
||
if account_exists(email):
|
||
activity_log.info(
|
||
f"Conta {email} já existe no Carbonio — actualizar senha e continuar",
|
||
source="vm112",
|
||
)
|
||
if display_name:
|
||
try:
|
||
_run_zmprov("ma", email, "displayName", display_name)
|
||
except CarbonioError:
|
||
pass
|
||
set_password(email, password)
|
||
set_password_must_change(email, False)
|
||
invalidate_domain_cache(domain)
|
||
return f"account {email} already exists (password updated)", True
|
||
msg = create_account(email, password, display_name)
|
||
return msg, False
|
||
|
||
|
||
def create_account_full(
|
||
email: str,
|
||
password: str,
|
||
*,
|
||
display_name: str | None = None,
|
||
quota_mb: int = 0,
|
||
must_change_password: bool = True,
|
||
) -> str:
|
||
"""Uma única chamada zmprov ca (evita 3–4 round-trips de ~5s cada)."""
|
||
if not EMAIL_RE.match(email):
|
||
raise CarbonioError("invalid email")
|
||
domain = email.split("@", 1)[1]
|
||
if not domain_exists(domain):
|
||
create_domain(domain)
|
||
activity_log.info(f"Criar conta: {email}", source="vm112")
|
||
args = ["ca", email, password]
|
||
if display_name:
|
||
args.extend(["displayName", display_name])
|
||
if quota_mb > 0:
|
||
args.extend(["zimbraMailQuota", str(int(quota_mb) * 1024 * 1024)])
|
||
if must_change_password:
|
||
args.extend(["zimbraPasswordMustChange", "TRUE"])
|
||
out = _run_zmprov(*args)
|
||
invalidate_domain_cache(domain)
|
||
return out
|
||
|
||
|
||
def account_summary_fast(
|
||
email: str,
|
||
*,
|
||
display_name: str | None = None,
|
||
quota_mb: int = 0,
|
||
must_change_password: bool = True,
|
||
) -> dict:
|
||
"""Resposta imediata sem zmprov ga (~5s)."""
|
||
quota_bytes = int(quota_mb) * 1024 * 1024 if quota_mb > 0 else 0
|
||
return {
|
||
"email": email.lower(),
|
||
"display_name": display_name or "",
|
||
"status": "active",
|
||
"active": True,
|
||
"suspended": False,
|
||
"quota_bytes": quota_bytes,
|
||
"quota_label": _quota_label(str(quota_bytes)),
|
||
"two_factor_enabled": False,
|
||
"password_must_change": must_change_password,
|
||
"last_logon": None,
|
||
}
|
||
|
||
|
||
def list_accounts(domain: str, *, use_cache: bool = True) -> list[str]:
|
||
"""Lista contas do domínio (zmprov -l gaa ~3–5s — cache TTL curto)."""
|
||
domain = domain.lower().strip()
|
||
cache_key = f"accounts_list:{domain}"
|
||
if use_cache:
|
||
cached = carbonio_cache.get(cache_key)
|
||
if cached is not None:
|
||
return list(cached)
|
||
code, out, err = _zmprov_run("-l", "gaa", domain, log_cmd=False)
|
||
if code != 0:
|
||
msg = err or out or "zmprov gaa failed"
|
||
raise CarbonioError(msg)
|
||
result = sorted(line.strip() for line in out.splitlines() if line.strip())
|
||
if use_cache:
|
||
carbonio_cache.set(cache_key, result, carbonio_cache.TTL_ACCOUNTS_LIST)
|
||
return result
|
||
|
||
|
||
def set_password(email: str, password: str) -> str:
|
||
return _run_zmprov("sp", email, password)
|
||
|
||
|
||
def account_exists(email: str) -> bool:
|
||
code, out, err = _zmprov_run("ga", email, log_cmd=False)
|
||
if code == 0:
|
||
return email.lower() in out.lower()
|
||
if _is_missing_account(err, out):
|
||
return False
|
||
activity_log.warn(f"zmprov ga {email}: {err or out}", source="vm112")
|
||
return False
|
||
|
||
|
||
def _parse_ga_attributes(out: str) -> dict[str, str]:
|
||
attrs: dict[str, str] = {}
|
||
for line in out.splitlines():
|
||
if ":" not in line:
|
||
continue
|
||
key, _, val = line.partition(":")
|
||
attrs[key.strip()] = val.strip()
|
||
return attrs
|
||
|
||
|
||
def get_account_attributes(email: str) -> dict[str, str]:
|
||
code, out, err = _zmprov_run("ga", email, log_cmd=False)
|
||
if code != 0:
|
||
raise CarbonioError(err or out or f"Conta {email} não encontrada")
|
||
return _parse_ga_attributes(out)
|
||
|
||
|
||
def get_domain_attributes(domain: str) -> dict[str, str]:
|
||
code, out, err = _zmprov_run("gd", domain, log_cmd=False)
|
||
if code != 0:
|
||
raise CarbonioError(err or out or f"Domínio {domain} não encontrado")
|
||
return _parse_ga_attributes(out)
|
||
|
||
|
||
def _quota_label(bytes_raw: str) -> str:
|
||
try:
|
||
n = int(bytes_raw)
|
||
except (TypeError, ValueError):
|
||
return bytes_raw or "—"
|
||
if n <= 0:
|
||
return "Ilimitada (COS)"
|
||
if n >= 1024**3:
|
||
return f"{n / 1024**3:.1f} GB"
|
||
if n >= 1024**2:
|
||
return f"{n / 1024**2:.0f} MB"
|
||
return f"{n} B"
|
||
|
||
|
||
def account_summary(email: str, *, use_cache: bool = True) -> dict:
|
||
email = email.lower().strip()
|
||
cache_key = f"account_summary:{email}"
|
||
if use_cache:
|
||
cached = carbonio_cache.get(cache_key)
|
||
if cached is not None:
|
||
return cached
|
||
attrs = get_account_attributes(email)
|
||
status = attrs.get("zimbraAccountStatus", "unknown")
|
||
quota_b = attrs.get("zimbraMailQuota", "0")
|
||
tfa = attrs.get("zimbraTwoFactorAuthEnabled", "").upper() == "TRUE"
|
||
must_change = attrs.get("zimbraPasswordMustChange", "").upper() == "TRUE"
|
||
last_logon = attrs.get("zimbraLastLogonTimestamp", "")
|
||
if last_logon and len(last_logon) >= 8:
|
||
last_logon = f"{last_logon[0:4]}-{last_logon[4:6]}-{last_logon[6:8]}"
|
||
summary = {
|
||
"email": email.lower(),
|
||
"display_name": attrs.get("displayName") or attrs.get("cn") or "",
|
||
"status": status,
|
||
"active": status == "active",
|
||
"suspended": status in ("locked", "closed", "maintenance"),
|
||
"quota_bytes": int(quota_b) if quota_b.isdigit() else 0,
|
||
"quota_label": _quota_label(quota_b),
|
||
"two_factor_enabled": tfa,
|
||
"password_must_change": must_change,
|
||
"last_logon": last_logon or None,
|
||
"_summary": True,
|
||
}
|
||
if use_cache:
|
||
carbonio_cache.set(cache_key, summary, carbonio_cache.TTL_ACCOUNT_SUMMARY)
|
||
return summary
|
||
|
||
|
||
def list_domain_accounts_fast(domain: str, *, use_cache: bool = True) -> list[dict]:
|
||
"""Uma chamada zmprov -l gaa (~5s) em vez de ga por conta (~5s × N)."""
|
||
domain = domain.lower().strip()
|
||
cache_key = f"accounts_fast:{domain}"
|
||
if use_cache:
|
||
cached = carbonio_cache.get(cache_key)
|
||
if cached is not None:
|
||
return cached
|
||
emails = list_accounts(domain)
|
||
result = [
|
||
{
|
||
"email": e,
|
||
"display_name": "",
|
||
"status": "—",
|
||
"active": True,
|
||
"suspended": False,
|
||
"quota_label": "—",
|
||
"two_factor_enabled": False,
|
||
"password_must_change": False,
|
||
"last_logon": None,
|
||
"_summary": False,
|
||
}
|
||
for e in emails
|
||
]
|
||
if use_cache:
|
||
carbonio_cache.set(cache_key, result, carbonio_cache.TTL_ACCOUNTS_LIST)
|
||
return result
|
||
|
||
|
||
def list_domain_accounts(domain: str, *, full: bool = False, use_cache: bool = True) -> list[dict]:
|
||
if not full:
|
||
return list_domain_accounts_fast(domain, use_cache=use_cache)
|
||
domain = domain.lower().strip()
|
||
emails = list_accounts(domain)
|
||
result = []
|
||
for email in emails:
|
||
try:
|
||
result.append(account_summary(email, use_cache=use_cache))
|
||
except CarbonioError:
|
||
result.append(
|
||
{
|
||
"email": email,
|
||
"status": "unknown",
|
||
"active": False,
|
||
"suspended": False,
|
||
"_summary": False,
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def invalidate_domain_cache(domain: str) -> None:
|
||
carbonio_cache.invalidate_domain(domain.lower().strip())
|
||
|
||
|
||
def set_account_status(email: str, active: bool) -> str:
|
||
status = "active" if active else "locked"
|
||
activity_log.info(f"Conta {email} → {status}", source="vm112")
|
||
out = _run_zmprov("ma", email, "zimbraAccountStatus", status)
|
||
carbonio_cache.invalidate_account(email)
|
||
return out
|
||
|
||
|
||
def set_mail_quota(email: str, quota_mb: int) -> str:
|
||
if quota_mb <= 0:
|
||
bytes_q = "0"
|
||
else:
|
||
bytes_q = str(int(quota_mb) * 1024 * 1024)
|
||
activity_log.info(f"Quota {email} → {quota_mb} MB", source="vm112")
|
||
out = _run_zmprov("ma", email, "zimbraMailQuota", bytes_q)
|
||
carbonio_cache.invalidate_account(email)
|
||
return out
|
||
|
||
|
||
def set_password_must_change(email: str, must_change: bool = True) -> str:
|
||
val = "TRUE" if must_change else "FALSE"
|
||
out = _run_zmprov("ma", email, "zimbraPasswordMustChange", val)
|
||
carbonio_cache.invalidate_account(email)
|
||
return out
|
||
|
||
|
||
def set_password_with_policy(
|
||
email: str,
|
||
password: str,
|
||
*,
|
||
must_change: bool = True,
|
||
) -> str:
|
||
out = set_password(email, password)
|
||
if must_change:
|
||
set_password_must_change(email, True)
|
||
else:
|
||
carbonio_cache.invalidate_account(email)
|
||
return out
|
||
|
||
|
||
def domain_two_factor_capabilities(domain: str) -> dict:
|
||
attrs = get_domain_attributes(domain)
|
||
avail = attrs.get("zimbraFeatureTwoFactorAuthAvailable", "FALSE").upper() == "TRUE"
|
||
required = attrs.get("zimbraFeatureTwoFactorAuthRequired", "FALSE").upper() == "TRUE"
|
||
return {
|
||
"available": avail,
|
||
"required": required,
|
||
"automated": avail,
|
||
"message": (
|
||
"2FA disponível neste domínio (política via painel activa)."
|
||
if avail
|
||
else (
|
||
"Política 2FA via painel inactiva no COS/domínio. "
|
||
"Utilizadores podem activar OTP em webmail: Settings → Auth → OTP Authentication → NEW OTP."
|
||
)
|
||
),
|
||
}
|
||
|
||
|
||
def set_user_two_factor(email: str, enabled: bool) -> str:
|
||
caps = domain_two_factor_capabilities(email.split("@", 1)[1])
|
||
if not caps["available"]:
|
||
raise CarbonioError(caps["message"])
|
||
val = "TRUE" if enabled else "FALSE"
|
||
activity_log.info(f"2FA {email} → {val}", source="vm112")
|
||
return _run_zmprov("ma", email, "zimbraTwoFactorAuthEnabled", val)
|
||
|
||
|
||
def set_domain_two_factor_required(domain: str, required: bool) -> str:
|
||
caps = domain_two_factor_capabilities(domain)
|
||
if not caps["available"]:
|
||
raise CarbonioError(caps["message"])
|
||
val = "TRUE" if required else "FALSE"
|
||
return _run_zmprov("md", domain, "zimbraFeatureTwoFactorAuthRequired", val)
|