""" Spec 037-C — Identidade projeto Ligbox: proj_{id}@ligbox.com.br O agente Ligbox cria caixa no Carbonio (ligbox.com.br), entrega senha ao cliente e referencia este e-mail em CF, DNS, registry e webhooks. Deploy: /opt/ligbox-wizard/backend/app/services/project_identity.py """ from __future__ import annotations import json import secrets import string from datetime import datetime, timezone from pathlib import Path from app.config import settings from app.services import carbonio, domain_registry LIGBOX_PROJECT_DOMAIN = "ligbox.com.br" _PROJECT_COUNTER = Path("/var/lib/ligbox-wizard/project_id_counter.txt") _PROJECT_VAULT = Path("/var/lib/ligbox-wizard/project_passwords") def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _project_email(project_id: str) -> str: pid = str(project_id).strip().zfill(3) return f"proj_{pid}@{LIGBOX_PROJECT_DOMAIN}" def allocate_project_id() -> str: """ID sequencial (001, 002, …).""" _PROJECT_COUNTER.parent.mkdir(parents=True, exist_ok=True) current = 0 if _PROJECT_COUNTER.is_file(): try: current = int(_PROJECT_COUNTER.read_text(encoding="utf-8").strip() or "0") except ValueError: current = 0 nxt = current + 1 _PROJECT_COUNTER.write_text(str(nxt), encoding="utf-8") return str(nxt).zfill(3) def generate_project_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 _vault_path(domain: str) -> Path: safe = domain.lower().strip().replace("/", "_") return _PROJECT_VAULT / f"{safe}.json" def save_password_vault(domain: str, project_email: str, password: str) -> None: _PROJECT_VAULT.mkdir(parents=True, exist_ok=True) _vault_path(domain).write_text( json.dumps( { "domain": domain.lower().strip(), "ligbox_project_email": project_email, "password": password, "created_at": _now_iso(), } ), encoding="utf-8", ) def load_password_vault(domain: str) -> dict | None: p = _vault_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 provision_project_email( client_domain: str, display_name: str | None = None, project_id: str | None = None, ) -> dict: """ Cria proj_{id}@ligbox.com.br no Carbonio e regista no domain_registry. Devolve email + password (mostrar uma vez ao cliente). """ client_domain = client_domain.lower().strip().rstrip(".") pid = project_id or allocate_project_id() email = _project_email(pid) label = display_name or f"Projeto {client_domain}" existing = domain_registry.get_domain_record(client_domain) or {} if existing.get("ligbox_project_email"): vault = load_password_vault(client_domain) return { "project_id": existing.get("project_id"), "ligbox_project_email": existing["ligbox_project_email"], "already_exists": True, "password": vault.get("password") if vault else None, "webmail": f"https://mail.{LIGBOX_PROJECT_DOMAIN}/", } password = generate_project_password() carbonio.create_account_full( email, password, display_name=label, must_change_password=False, ) save_password_vault(client_domain, email, password) record = { **existing, "domain": client_domain, "project_id": pid, "ligbox_project_email": email, "ligbox_project_display_name": label, "ligbox_project_created_at": _now_iso(), "ligbox_project_password_delivered": False, } domain_registry.save_domain_record(client_domain, record) return { "project_id": pid, "ligbox_project_email": email, "password": password, "already_exists": False, "webmail": f"https://mail.{LIGBOX_PROJECT_DOMAIN}/", "message": ( f"Caixa {email} criada. Use para Cloudflare e acompanhamento do setup. " "Guarde a senha — não será mostrada novamente." ), } def get_project_identity(client_domain: str) -> dict | None: rec = domain_registry.get_domain_record(client_domain.lower().strip()) if not rec or not rec.get("ligbox_project_email"): return None return { "domain": rec["domain"], "project_id": rec.get("project_id"), "ligbox_project_email": rec["ligbox_project_email"], "ligbox_project_display_name": rec.get("ligbox_project_display_name"), "ligbox_project_created_at": rec.get("ligbox_project_created_at"), "ligbox_project_password_delivered": rec.get("ligbox_project_password_delivered", False), "webmail": f"https://mail.{LIGBOX_PROJECT_DOMAIN}/", } def mark_password_delivered(client_domain: str) -> None: rec = domain_registry.get_domain_record(client_domain.lower().strip()) or {} rec["ligbox_project_password_delivered"] = True rec["ligbox_project_password_delivered_at"] = _now_iso() domain_registry.save_domain_record(client_domain.lower().strip(), rec)