""" Spec 035/043 — Provisionamento mail bundle pós-activação Desk/FOSS. Garante domínio Carbonio, conta gerente (admin@domínio), quotas e metadata no registry. Deploy: /opt/ligbox-wizard/backend/app/services/mail_bundle.py """ from __future__ import annotations import json import secrets import string from datetime import datetime, timezone from typing import Any from app.services import activity_log, carbonio, domain_registry from app.services.carbonio import CarbonioError _BUNDLE_VAULT = __import__("pathlib").Path("/var/lib/ligbox-wizard/mail_bundles") def _now() -> str: return datetime.now(timezone.utc).isoformat() def _generate_password(length: int = 16) -> str: alphabet = string.ascii_letters + string.digits + "!@#$%&*" while True: pwd = "".join(secrets.choice(alphabet) for _ in range(length)) if any(c.islower() for c in pwd) and any(c.isupper() for c in pwd) and any(c.isdigit() for c in pwd): return pwd def _bundle_path(domain: str) -> __import__("pathlib").Path: safe = domain.lower().strip().replace("/", "_") return _BUNDLE_VAULT / f"{safe}.json" def load_bundle(domain: str) -> dict[str, Any] | None: p = _bundle_path(domain) if not p.is_file(): return None try: return json.loads(p.read_text(encoding="utf-8")) except json.JSONDecodeError: return None def save_bundle(domain: str, data: dict[str, Any]) -> None: _BUNDLE_VAULT.mkdir(parents=True, exist_ok=True) _bundle_path(domain).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def provision_mail_bundle( *, domain: str, admin_email: str, admin_name: str | None = None, seats: int = 25, mail_gb_per_seat: int = 30, files_gb_per_seat: int = 200, foss_order_id: str | int | None = None, admin_password: str | None = None, ) -> dict[str, Any]: """ Idempotente — reutiliza domínio/conta existentes; actualiza quotas e metadata. """ domain = domain.lower().strip().rstrip(".") email = admin_email.lower().strip() if not email.endswith(f"@{domain}"): raise ValueError(f"admin_email deve ser @{domain}") existing = load_bundle(domain) if existing and existing.get("provisioned"): return { **existing, "already_provisioned": True, "mail_host": existing.get("mail_host") or f"mail.{domain}", } quota_mb = max(1, int(mail_gb_per_seat)) * 1024 password = admin_password or _generate_password() display = admin_name or email.split("@")[0] activity_log.info(f"mail-bundle: domínio {domain} seats={seats}", source="internal") try: if not carbonio.domain_exists(domain): try: carbonio.create_domain(domain) except CarbonioError as exc: if "DOMAIN_EXISTS" not in str(exc): raise carbonio.set_domain_public_hostname(domain) else: carbonio.set_domain_public_hostname(domain) _msg, reused_account = carbonio.ensure_onboarding_account( email, password, display_name=display ) if hasattr(carbonio, "set_mail_quota"): carbonio.set_mail_quota(email, quota_mb) except CarbonioError as exc: activity_log.error(f"mail-bundle Carbonio: {exc}", source="internal") raise mail_host = f"mail.{domain}" bundle = { "domain": domain, "admin_email": email, "admin_name": display, "seats": int(seats), "mail_gb_per_seat": int(mail_gb_per_seat), "files_gb_per_seat": int(files_gb_per_seat), "foss_order_id": foss_order_id, "mail_host": mail_host, "webmail_url": f"https://{mail_host}/", "files_url": f"https://files.{domain}/", "provisioned": True, "provisioned_at": _now(), "account_reused": reused_account, } save_bundle(domain, bundle) rec = domain_registry.get_domain_record(domain) or {} rec.update( { "domain": domain, "mail_bundle": { "seats": bundle["seats"], "mail_gb_per_seat": bundle["mail_gb_per_seat"], "files_gb_per_seat": bundle["files_gb_per_seat"], "foss_order_id": foss_order_id, "provisioned_at": bundle["provisioned_at"], }, "manager_email": email, "manager_name": display, } ) domain_registry.save_domain_record(domain, rec) activity_log.ok(f"mail-bundle OK {domain} → {mail_host}", source="internal") return { **bundle, "already_provisioned": False, "password_generated": not bool(admin_password), }