"""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"