ligbox-ops-platform/projects/ops-desk/api/app/governance_routes.py

"""Governance API — user wizard, audit. Spec 040 · DS-API-002."""

from __future__ import annotations

import json
import secrets
import string
from datetime import datetime, timezone

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field

from app import auth, mail_notify
from app.desk_governance_store import (
    ACCESS_LEVELS,
    GOVERNANCE_MODULES,
    ROLE_GROUPS,
    default_module_permissions,
    ensure_user_meta,
    get_user_meta,
    group_for_role,
    init_governance_schema,
    list_audit,
    log_audit,
    upsert_user_meta,
)
from app.permissions import ROLES, can_manage_users

router = APIRouter(prefix="/api/v1/governance", tags=["governance"])


class ModulePermissions(BaseModel):
    desk: str = "none"
    openpanel: str = "none"
    billing: str = "none"
    api: str = "none"
    security: str = "none"
    ai_agents: str = "none"


class UserWizardRequest(BaseModel):
    display_name: str = Field(min_length=1, max_length=120)
    email: str = Field(min_length=3, max_length=320)
    phone: str | None = Field(default=None, max_length=32)
    password: str = Field(min_length=6)
    account_status: str = Field(default="active", pattern="^(active|frozen|pending|invited)$")
    force_password_change: bool = False
    mfa_required: bool = False
    notifications_enabled: bool = True
    api_access: bool = False
    role: str
    main_group: str | None = None
    secondary_groups: list[str] = Field(default_factory=list)
    module_permissions: ModulePermissions | None = None
    notes: str | None = Field(default=None, max_length=200)
    send_invite_email: bool = True
    activate_account: bool = True


def _db():
    conn = auth.db()
    try:
        init_governance_schema(conn)
        yield conn
    finally:
        conn.close()


def _validate_levels(perms: dict[str, str]) -> None:
    for mod, lv in perms.items():
        if mod not in GOVERNANCE_MODULES:
            raise HTTPException(400, f"módulo inválido: {mod}")
        if lv not in ACCESS_LEVELS:
            raise HTTPException(400, f"nível inválido: {lv}")


@router.get("/modules")
def governance_modules(user: auth.DeskUser = Depends(auth.get_current_user)):
    if not can_manage_users(user.role):
        raise HTTPException(403, "insufficient permissions")
    return {
        "modules": list(GOVERNANCE_MODULES),
        "levels": list(ACCESS_LEVELS),
        "groups": sorted(set(ROLE_GROUPS.values())),
    }


@router.get("/audit")
def governance_audit(
    target_type: str | None = Query(default=None),
    target_id: str | None = Query(default=None),
    limit: int = Query(default=50, ge=1, le=200),
    user: auth.DeskUser = Depends(auth.get_current_user),
    conn=Depends(_db),
):
    if not can_manage_users(user.role):
        raise HTTPException(403, "insufficient permissions")
    return {"events": list_audit(conn, target_type=target_type, target_id=target_id, limit=limit)}


@router.get("/users/{username}/meta")
def user_meta(
    username: str,
    user: auth.DeskUser = Depends(auth.get_current_user),
    conn=Depends(_db),
):
    if not can_manage_users(user.role):
        raise HTTPException(403, "insufficient permissions")
    meta = get_user_meta(conn, username)
    if not meta:
        row = auth._user_row(username)
        if not row:
            raise HTTPException(404, "user not found")
        meta = ensure_user_meta(conn, username, row["role"])
        conn.commit()
    return {"meta": meta}


@router.get("/users/stats")
def users_stats(
    user: auth.DeskUser = Depends(auth.get_current_user),
    conn=Depends(_db),
):
    if not can_manage_users(user.role):
        raise HTTPException(403, "insufficient permissions")
    rows = conn.execute(
        "SELECT username, role, active FROM desk_users"
    ).fetchall()
    total = len(rows)
    active = sum(1 for r in rows if r["active"])
    frozen = total - active
    super_admin = sum(1 for r in rows if r["role"] == "super_admin")
    return {
        "total": total,
        "active": active,
        "frozen": frozen,
        "super_admin": super_admin,
    }


@router.post("/users/wizard")
def create_user_wizard(
    body: UserWizardRequest,
    user: auth.DeskUser = Depends(auth.get_current_user),
    conn=Depends(_db),
):
    if not can_manage_users(user.role):
        raise HTTPException(403, "insufficient permissions")
    email = body.email.strip().lower()
    if "@" not in email:
        raise HTTPException(400, "email inválido")
    if body.role not in ROLES:
        raise HTTPException(400, "invalid role")
    if body.role == "super_admin" and user.role != "super_admin":
        raise HTTPException(403, "apenas Super Admin pode criar Super Admin")

    perms = (
        body.module_permissions.model_dump()
        if body.module_permissions
        else default_module_permissions(body.role)
    )
    _validate_levels(perms)

    main_group = body.main_group or group_for_role(body.role)
    now = datetime.now(timezone.utc).isoformat()
    active = body.activate_account and body.account_status == "active"
    invite_token = secrets.token_urlsafe(24)

    exists = conn.execute(
        "SELECT 1 FROM desk_users WHERE username = ? OR email = ?",
        (email, email),
    ).fetchone()
    if exists:
        raise HTTPException(409, "utilizador já existe")

    conn.execute(
        """
        INSERT INTO desk_users
        (username, password_hash, role, display_name, email, phone,
         mfa_enabled, totp_secret, totp_enabled, active, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, NULL, 0, ?, ?, ?)
        """,
        (
            email,
            auth.hash_password(body.password),
            body.role,
            body.display_name.strip(),
            email,
            body.phone,
            1 if body.mfa_required else 0,
            1 if active else 0,
            now,
            now,
        ),
    )

    internal_id = f"LB-{secrets.token_hex(4).upper()}"
    conn.execute(
        """
        INSERT INTO desk_user_meta
        (username, internal_id, main_group, secondary_groups_json, account_status,
         force_password_change, api_access, notifications_enabled, module_permissions_json,
         invite_token, notes, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            email,
            internal_id,
            main_group,
            json.dumps(body.secondary_groups),
            body.account_status,
            1 if body.force_password_change else 0,
            1 if body.api_access else 0,
            1 if body.notifications_enabled else 0,
            json.dumps(perms),
            invite_token,
            body.notes,
            now,
            now,
        ),
    )

    actor_label = user.display_name or user.username
    audit = log_audit(
        conn,
        actor_username=user.username,
        actor_role=user.role,
        action="user.created",
        target_type="user",
        target_id=email,
        summary=f"Created by {actor_label}",
        payload={"role": body.role, "internal_id": internal_id},
    )
    conn.commit()

    invite_link = f"https://desk.ligbox.com.br/register.html?invite={invite_token}"
    invite_email_sent = False
    if body.send_invite_email:
        try:
            invite_email_sent = mail_notify.send_email(
                email,
                "Convite Ligbox Ops Desk",
                f"Olá {body.display_name},\n\nFoi criada a sua conta.\nDefina a sua senha: {invite_link}\n",
            )
        except Exception:
            invite_email_sent = False

    row = conn.execute(
        """
        SELECT u.username, u.role, u.display_name, u.active, u.last_login_at,
               u.created_at, u.updated_at, u.email, u.phone, u.mfa_enabled, u.totp_enabled
        FROM desk_users u WHERE u.username = ?
        """,
        (email,),
    ).fetchone()
    public = auth.user_public_dict(row)
    meta = get_user_meta(conn, email)

    return {
        "user": public,
        "meta": meta,
        "audit": audit,
        "internal_id": internal_id,
        "invite_link": invite_link,
        "invite_email_sent": invite_email_sent if body.send_invite_email else None,
        "message": "Utilizador criado",
    }


@router.post("/users/{username}/freeze")
def freeze_user(
    username: str,
    user: auth.DeskUser = Depends(auth.get_current_user),
    conn=Depends(_db),
):
    if not can_manage_users(user.role):
        raise HTTPException(403, "insufficient permissions")
    target = username.strip().lower() if username.lower() != "root" else "root"
    if target == "root":
        raise HTTPException(400, "não é possível congelar root")
    row = auth._user_row(target)
    if not row:
        raise HTTPException(404, "user not found")
    new_active = not bool(row["active"])
    now = datetime.now(timezone.utc).isoformat()
    conn.execute(
        "UPDATE desk_users SET active = ?, updated_at = ? WHERE username = ?",
        (1 if new_active else 0, now, target),
    )
    meta = ensure_user_meta(conn, target, row["role"])
    upsert_user_meta(conn, target, account_status="active" if new_active else "frozen")
    verb = "Activated" if new_active else "Frozen"
    actor_label = user.display_name or user.username
    audit = log_audit(
        conn,
        actor_username=user.username,
        actor_role=user.role,
        action="user.frozen" if not new_active else "user.activated",
        target_type="user",
        target_id=target,
        summary=f"{verb} by {actor_label}",
    )
    conn.commit()
    return {"user": auth.user_public_dict(auth._user_row(target)), "audit": audit}


@router.post("/users/{username}/reset-password")
def admin_reset_password(
    username: str,
    user: auth.DeskUser = Depends(auth.get_current_user),
    conn=Depends(_db),
):
    if not can_manage_users(user.role):
        raise HTTPException(403, "insufficient permissions")
    target = username.strip().lower() if username.lower() != "root" else "root"
    row = auth._user_row(target)
    if not row:
        raise HTTPException(404, "user not found")
    pwd = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(12))
    now = datetime.now(timezone.utc).isoformat()
    conn.execute(
        "UPDATE desk_users SET password_hash = ?, updated_at = ? WHERE username = ?",
        (auth.hash_password(pwd), now, target),
    )
    actor_label = user.display_name or user.username
    audit = log_audit(
        conn,
        actor_username=user.username,
        actor_role=user.role,
        action="user.password.reset",
        target_type="user",
        target_id=target,
        summary=f"Password reset by {actor_label}",
    )
    conn.commit()
    return {"ok": True, "generated_password": pwd, "audit": audit}