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

"""Spec 039 — Catálogo de acções + overrides persistidos (Matriz editável)."""

from __future__ import annotations

import os
import sqlite3
from datetime import datetime, timezone
from functools import lru_cache
from pathlib import Path
from typing import Any

import yaml

VALID_LEVELS = frozenset({"full", "read", "link", "api", "approve", "system", "none"})

EXECUTIVE_ROWS: list[dict[str, Any]] = [
    {
        "id": "desk-users-admin",
        "area": "Criar / editar / freeze utilizadores Desk",
        "who": ["SU"],
        "why": "Credenciais internas Ligbox",
        "groups": ["desk-auth"],
        "action_ids": ["desk.auth.user.edit", "desk.auth.user.freeze", "desk.auth.user.password.reset"],
    },
    {
        "id": "desk-registration",
        "area": "Aprovar cadastros",
        "who": ["SU", "CO"],
        "why": "CO autónomo — Roger 2026-06-29",
        "groups": ["desk-auth"],
        "action_ids": ["desk.auth.user.approve_registration", "desk.auth.user.reject_registration"],
    },
    {
        "id": "desk-freeze-policy",
        "area": "Congelar users (SSU nunca)",
        "who": ["SU"],
        "why": "Segregação comercial",
        "groups": ["desk-auth"],
        "action_ids": ["desk.auth.user.freeze"],
    },
    {
        "id": "vm112-purge",
        "area": "Purge domínio / dados cliente",
        "who": ["SU", "CO"],
        "why": "Irreversível — Spec 032",
        "groups": ["vm112"],
        "action_ids": ["vm112.domain.purge", "vm112.purge.job.recover"],
    },
    {
        "id": "billing-validate",
        "area": "Validar billing / faturação",
        "who": ["SU", "CO", "FIN", "SAD"],
        "why": "Segregação comercial vs financeira",
        "groups": ["desk-ops"],
        "action_ids": ["desk.billing.state.validate"],
    },
    {
        "id": "foss-orders",
        "area": "FOSS pedidos e clientes",
        "who": ["SAD", "SSU", "PTR"],
        "why": "Linha de frente comercial",
        "groups": ["vm123-foss"],
        "action_ids": ["vm123_foss.client.create", "vm123_foss.order.create"],
    },
    {
        "id": "foss-void",
        "area": "FOSS faturas / void",
        "who": ["FIN", "SU"],
        "why": "Risco fiscal",
        "groups": ["vm123-foss"],
        "action_ids": ["vm123_foss.invoice.void"],
    },
    {
        "id": "openpanel-delete",
        "area": "OpenPanel delete instância",
        "who": ["SU", "CO"],
        "why": "Downtime cliente — Roger 2026-06-29",
        "groups": ["vm123-openpanel"],
        "action_ids": ["vm123_openpanel.site.delete"],
    },
    {
        "id": "openpanel-content",
        "area": "OpenPanel conteúdo sites",
        "who": ["CMS", "SEO", "MKT"],
        "why": "Operação editorial",
        "groups": ["vm123-openpanel"],
        "action_ids": ["vm123_openpanel.site.create", "vm123_openpanel.ssl.manage"],
    },
    {
        "id": "tickets-assist",
        "area": "Tickets / assist",
        "who": ["TEC", "CO", "SU"],
        "why": "Menor privilégio",
        "groups": ["desk-ops"],
        "action_ids": ["desk.ticket.patch", "desk.assist.takeover"],
    },
    {
        "id": "infra-deploy",
        "area": "Infra / deploy",
        "who": ["DVO", "DEV", "SU"],
        "why": "Separação código vs infra",
        "groups": ["desk-ops"],
        "action_ids": ["desk.infra.deploy"],
    },
    {
        "id": "agents-a7",
        "area": "Agentes A7 remediação",
        "who": ["AIO", "CO", "SU"],
        "why": "Human-in-the-loop",
        "groups": ["agents"],
        "action_ids": ["desk.agent.runbook.approve"],
    },
    {
        "id": "modules-toggle",
        "area": "Módulos Desk ON/OFF",
        "who": ["SU"],
        "why": "Feature flags globais",
        "groups": ["desk-auth"],
        "action_ids": ["desk.auth.modules.toggle"],
    },
    {
        "id": "console-ops",
        "area": "Console (mesma matriz Desk)",
        "who": ["SU", "CO", "TEC", "SOC"],
        "why": "Handoff Desk → Console",
        "groups": ["console"],
        "action_ids": ["console.runbook.execute", "console.case.assign"],
    },
    {
        "id": "rbac-custom",
        "area": "RBAC custom (templates)",
        "who": ["SU"],
        "why": "Herda CO ou TEC only",
        "groups": ["desk-auth"],
        "action_ids": ["desk.auth.role.create"],
    },
]

ROLE_CODES: dict[str, str] = {
    "super_admin": "SU",
    "ops_lead": "CO",
    "technician": "TEC",
    "noc": "NOC",
    "sales_admin": "SAD",
    "sales_support": "SSU",
    "finance": "FIN",
    "marketing": "MKT",
    "seo": "SEO",
    "developer": "DEV",
    "devops": "DVO",
    "security_analyst": "SOC",
    "content_editor": "CMS",
    "agentic_operator": "AIO",
    "partner": "PTR",
    "api_service": "SVC",
    "agent_system": "AGT",
}


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _catalog_paths() -> list[Path]:
    env = os.getenv("ACTION_CATALOG_PATH", "").strip()
    paths: list[Path] = []
    if env:
        paths.append(Path(env))
    paths.extend(
        [
            Path("/opt/ligbox-ops-platform/specs/039-ligbox-ops-authorization-catalog/contracts/action-catalog.yaml"),
            Path(__file__).resolve().parent / "data" / "action-catalog.yaml",
        ]
    )
    return paths


@lru_cache(maxsize=1)
def load_base_catalog() -> dict[str, Any]:
    for path in _catalog_paths():
        if path.is_file():
            raw = yaml.safe_load(path.read_text(encoding="utf-8"))
            if isinstance(raw, dict) and raw.get("actions"):
                return raw
    raise FileNotFoundError("action-catalog.yaml not found")


def init_schema(conn: sqlite3.Connection) -> None:
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS rbac_action_overrides (
            action_id TEXT NOT NULL,
            role_id TEXT NOT NULL,
            level TEXT NOT NULL,
            updated_at TEXT NOT NULL,
            updated_by TEXT,
            PRIMARY KEY (action_id, role_id)
        );
        CREATE TABLE IF NOT EXISTS rbac_action_audit (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            action_id TEXT NOT NULL,
            role_id TEXT NOT NULL,
            old_level TEXT,
            new_level TEXT NOT NULL,
            username TEXT,
            created_at TEXT NOT NULL
        );
        """
    )
    conn.commit()


def _load_overrides(conn: sqlite3.Connection) -> dict[tuple[str, str], str]:
    rows = conn.execute("SELECT action_id, role_id, level FROM rbac_action_overrides").fetchall()
    return {(r["action_id"], r["role_id"]): r["level"] for r in rows}


def effective_level(
    action: dict[str, Any],
    role_id: str,
    overrides: dict[tuple[str, str], str],
) -> str:
    key = (action["id"], role_id)
    if key in overrides:
        return overrides[key]
    return (action.get("roles") or {}).get(role_id, "none")


def export_catalog(conn: sqlite3.Connection, *, edit_enabled: bool) -> dict[str, Any]:
    base = load_base_catalog()
    overrides = _load_overrides(conn)
    roles_meta = base.get("roles") or {}
    for rid, code in ROLE_CODES.items():
        if rid not in roles_meta:
            roles_meta[rid] = {"code": code, "label": rid.replace("_", " ").title()}

    actions_out: list[dict[str, Any]] = []
    for action in base.get("actions") or []:
        aid = action.get("id")
        if not aid:
            continue
        defaults = action.get("roles") or {}
        effective = {role: effective_level(action, role, overrides) for role in roles_meta}
        overridden = {
            role
            for role in roles_meta
            if (aid, role) in overrides and overrides[(aid, role)] != defaults.get(role, "none")
        }
        actions_out.append(
            {
                "id": aid,
                "group": action.get("group"),
                "label": action.get("label"),
                "api": action.get("api"),
                "why": action.get("why"),
                "gap": bool(action.get("gap")),
                "defaults": defaults,
                "effective": effective,
                "overridden_roles": sorted(overridden),
            }
        )

    groups = base.get("groups") or []
    group_by_id = {g["id"]: g for g in groups if g.get("id")}

    executive = []
    for row in EXECUTIVE_ROWS:
        executive.append({**row, "editable": edit_enabled})

    recent_audit = conn.execute(
        """
        SELECT action_id, role_id, old_level, new_level, username, created_at
        FROM rbac_action_audit ORDER BY id DESC LIMIT 30
        """
    ).fetchall()

    return {
        "spec": "039",
        "version": base.get("version", "1.0"),
        "editable": edit_enabled,
        "roles": roles_meta,
        "role_codes": ROLE_CODES,
        "access_levels": base.get("access_levels") or {},
        "groups": groups,
        "group_by_id": group_by_id,
        "executive_map": executive,
        "actions": actions_out,
        "stats": {"action_count": len(actions_out), "override_count": len(overrides)},
        "recent_audit": [dict(r) for r in recent_audit],
    }


def set_action_override(
    conn: sqlite3.Connection,
    *,
    action_id: str,
    role_id: str,
    level: str,
    username: str,
    reset: bool = False,
) -> dict[str, Any]:
    if level not in VALID_LEVELS:
        raise ValueError(f"invalid level: {level}")

    base = load_base_catalog()
    roles_meta = base.get("roles") or {}
    if role_id not in roles_meta and role_id not in ROLE_CODES:
        raise ValueError(f"unknown role: {role_id}")

    action = next((a for a in base.get("actions") or [] if a.get("id") == action_id), None)
    if not action:
        raise ValueError(f"unknown action: {action_id}")

    defaults = action.get("roles") or {}
    default_level = defaults.get(role_id, "none")
    overrides = _load_overrides(conn)
    old = overrides.get((action_id, role_id), default_level)

    if reset or level == default_level:
        conn.execute(
            "DELETE FROM rbac_action_overrides WHERE action_id = ? AND role_id = ?",
            (action_id, role_id),
        )
        new_level = default_level
    else:
        conn.execute(
            """
            INSERT INTO rbac_action_overrides (action_id, role_id, level, updated_at, updated_by)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(action_id, role_id) DO UPDATE SET
                level = excluded.level,
                updated_at = excluded.updated_at,
                updated_by = excluded.updated_by
            """,
            (action_id, role_id, level, _now(), username),
        )
        new_level = level

    conn.execute(
        """
        INSERT INTO rbac_action_audit (action_id, role_id, old_level, new_level, username, created_at)
        VALUES (?, ?, ?, ?, ?, ?)
        """,
        (action_id, role_id, old, new_level, username, _now()),
    )
    conn.commit()

    return {
        "action_id": action_id,
        "role_id": role_id,
        "level": new_level,
        "default_level": default_level,
        "is_override": new_level != default_level,
    }


def can_action(conn: sqlite3.Connection, role_id: str, action_id: str) -> bool:
    """True se nível efectivo != none."""
    base = load_base_catalog()
    action = next((a for a in base.get("actions") or [] if a.get("id") == action_id), None)
    if not action:
        return False
    overrides = _load_overrides(conn)
    return effective_level(action, role_id, overrides) != "none"