6c4b063f76
Adds RBAC matrix API/UI behind feature flags, agent×role toggles with audit, and extends Agentic Ops with streaming chat, Kimi LLM support, and UI updates.
358 lines
12 KiB
Python
358 lines
12 KiB
Python
"""Persistência e enforcement — atribuições Agentics × função (Spec 027 UI-C)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sqlite3
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
RELATIONS = ("ui", "focus", "approve")
|
||
CAP_IDS = ("use_ui", "trigger_runs", "approve_runbooks", "configure_models")
|
||
|
||
LOCKED_AGENT_BINDINGS: frozenset[tuple[str, str, str]] = frozenset({
|
||
("A7", "agentic_operator", "approve"),
|
||
("A7", "super_admin", "approve"),
|
||
})
|
||
|
||
AGENT_IDS = ("A0", "A1", "A2", "A3", "A4", "A5", "A6", "A7")
|
||
|
||
|
||
def _spec():
|
||
from app.rbac_matrix import AGENTIC_GOVERNANCE, AGENT_ROLE_MAP, ROLE_COLUMNS
|
||
|
||
return AGENTIC_GOVERNANCE, AGENT_ROLE_MAP, ROLE_COLUMNS
|
||
|
||
|
||
def _now() -> str:
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
def _default_agent_enabled(agent_id: str, role_id: str, relation: str) -> bool:
|
||
_, AGENT_ROLE_MAP, _ = _spec()
|
||
from app.rbac_matrix import AGENTIC_GOVERNANCE
|
||
|
||
mapping = AGENT_ROLE_MAP.get(agent_id, {})
|
||
if relation == "approve":
|
||
return role_id in mapping.get("approvers", [])
|
||
if relation == "focus":
|
||
return role_id in mapping.get("operators", [])
|
||
if relation == "ui":
|
||
return role_id in AGENTIC_GOVERNANCE["use_ui"]
|
||
return False
|
||
|
||
|
||
def _default_cap_enabled(cap_id: str, role_id: str) -> bool:
|
||
AGENTIC_GOVERNANCE, _, _ = _spec()
|
||
roles = AGENTIC_GOVERNANCE.get(cap_id)
|
||
if isinstance(roles, list):
|
||
return role_id in roles
|
||
return False
|
||
|
||
|
||
def _is_locked(agent_id: str, role_id: str, relation: str) -> bool:
|
||
return (agent_id, role_id, relation) in LOCKED_AGENT_BINDINGS
|
||
|
||
|
||
def init_schema(conn: sqlite3.Connection) -> None:
|
||
conn.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS agent_role_bindings (
|
||
agent_id TEXT NOT NULL,
|
||
role_id TEXT NOT NULL,
|
||
relation TEXT NOT NULL,
|
||
enabled INTEGER NOT NULL DEFAULT 1,
|
||
locked INTEGER NOT NULL DEFAULT 0,
|
||
updated_at TEXT NOT NULL,
|
||
updated_by TEXT,
|
||
PRIMARY KEY (agent_id, role_id, relation)
|
||
);
|
||
CREATE TABLE IF NOT EXISTS agent_governance_caps (
|
||
cap_id TEXT NOT NULL,
|
||
role_id TEXT NOT NULL,
|
||
enabled INTEGER NOT NULL DEFAULT 1,
|
||
locked INTEGER NOT NULL DEFAULT 0,
|
||
updated_at TEXT NOT NULL,
|
||
updated_by TEXT,
|
||
PRIMARY KEY (cap_id, role_id)
|
||
);
|
||
CREATE TABLE IF NOT EXISTS rbac_agent_audit (
|
||
id INTEGER PRIMARY KEY,
|
||
ts TEXT NOT NULL,
|
||
username TEXT NOT NULL,
|
||
action TEXT NOT NULL,
|
||
payload_json TEXT NOT NULL
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_rbac_agent_audit_ts ON rbac_agent_audit(ts DESC);
|
||
"""
|
||
)
|
||
_seed_defaults(conn)
|
||
|
||
|
||
def _seed_defaults(conn: sqlite3.Connection) -> None:
|
||
_, _, ROLE_COLUMNS = _spec()
|
||
now = _now()
|
||
if conn.execute("SELECT COUNT(*) c FROM agent_role_bindings").fetchone()["c"] == 0:
|
||
rows = []
|
||
for agent_id in AGENT_IDS:
|
||
for role_id in ROLE_COLUMNS:
|
||
for relation in RELATIONS:
|
||
enabled = 1 if _default_agent_enabled(agent_id, role_id, relation) else 0
|
||
locked = 1 if _is_locked(agent_id, role_id, relation) else 0
|
||
rows.append((agent_id, role_id, relation, enabled, locked, now, "system"))
|
||
conn.executemany(
|
||
"""INSERT INTO agent_role_bindings
|
||
(agent_id, role_id, relation, enabled, locked, updated_at, updated_by)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
||
rows,
|
||
)
|
||
if conn.execute("SELECT COUNT(*) c FROM agent_governance_caps").fetchone()["c"] == 0:
|
||
cap_rows = []
|
||
for cap_id in CAP_IDS:
|
||
for role_id in ROLE_COLUMNS:
|
||
enabled = 1 if _default_cap_enabled(cap_id, role_id) else 0
|
||
cap_rows.append((cap_id, role_id, enabled, 0, now, "system"))
|
||
conn.executemany(
|
||
"""INSERT INTO agent_governance_caps
|
||
(cap_id, role_id, enabled, locked, updated_at, updated_by)
|
||
VALUES (?, ?, ?, ?, ?, ?)""",
|
||
cap_rows,
|
||
)
|
||
|
||
|
||
def audit_log(
|
||
conn: sqlite3.Connection,
|
||
*,
|
||
username: str,
|
||
action: str,
|
||
payload: dict[str, Any],
|
||
) -> None:
|
||
conn.execute(
|
||
"INSERT INTO rbac_agent_audit (ts, username, action, payload_json) VALUES (?, ?, ?, ?)",
|
||
(_now(), username, action, json.dumps(payload)),
|
||
)
|
||
|
||
|
||
def list_audit(conn: sqlite3.Connection, *, limit: int = 50) -> list[dict[str, Any]]:
|
||
rows = conn.execute(
|
||
"SELECT id, ts, username, action, payload_json FROM rbac_agent_audit ORDER BY id DESC LIMIT ?",
|
||
(limit,),
|
||
).fetchall()
|
||
out = []
|
||
for r in rows:
|
||
item = dict(r)
|
||
try:
|
||
item["payload"] = json.loads(item.pop("payload_json") or "{}")
|
||
except json.JSONDecodeError:
|
||
item["payload"] = {}
|
||
out.append(item)
|
||
return out
|
||
|
||
|
||
def _row_enabled(row: sqlite3.Row | None, default: bool = False) -> bool:
|
||
if row is None:
|
||
return default
|
||
return bool(row["enabled"])
|
||
|
||
|
||
def agent_relations(conn: sqlite3.Connection, agent_id: str, role_id: str) -> list[str]:
|
||
"""Relações activas função ↔ agente."""
|
||
if role_id == "super_admin":
|
||
return list(RELATIONS)
|
||
rels: list[str] = []
|
||
for relation in RELATIONS:
|
||
row = conn.execute(
|
||
"""SELECT enabled FROM agent_role_bindings
|
||
WHERE agent_id=? AND role_id=? AND relation=?""",
|
||
(agent_id, role_id, relation),
|
||
).fetchone()
|
||
if _row_enabled(row, _default_agent_enabled(agent_id, role_id, relation)):
|
||
rels.append(relation)
|
||
return rels
|
||
|
||
|
||
def cap_enabled(conn: sqlite3.Connection, cap_id: str, role_id: str) -> bool:
|
||
if role_id == "super_admin":
|
||
return True
|
||
row = conn.execute(
|
||
"SELECT enabled FROM agent_governance_caps WHERE cap_id=? AND role_id=?",
|
||
(cap_id, role_id),
|
||
).fetchone()
|
||
return _row_enabled(row, _default_cap_enabled(cap_id, role_id))
|
||
|
||
|
||
def can_use_agentics_ui(conn: sqlite3.Connection, role_id: str) -> bool:
|
||
if role_id == "super_admin":
|
||
return True
|
||
return cap_enabled(conn, "use_ui", role_id)
|
||
|
||
|
||
def can_chat_agent(conn: sqlite3.Connection, role_id: str, agent_id: str) -> bool:
|
||
if role_id == "super_admin":
|
||
return True
|
||
rels = agent_relations(conn, agent_id, role_id)
|
||
return bool(set(rels) & {"ui", "focus", "approve"})
|
||
|
||
|
||
def can_approve_agent(conn: sqlite3.Connection, role_id: str, agent_id: str) -> bool:
|
||
if role_id == "super_admin":
|
||
return True
|
||
return "approve" in agent_relations(conn, agent_id, role_id)
|
||
|
||
|
||
def can_trigger_runs(conn: sqlite3.Connection, role_id: str) -> bool:
|
||
if role_id == "super_admin":
|
||
return True
|
||
return cap_enabled(conn, "trigger_runs", role_id)
|
||
|
||
|
||
def can_approve_runbooks(conn: sqlite3.Connection, role_id: str) -> bool:
|
||
if role_id == "super_admin":
|
||
return True
|
||
return cap_enabled(conn, "approve_runbooks", role_id)
|
||
|
||
|
||
def load_bindings_matrix(conn: sqlite3.Connection) -> dict[str, Any]:
|
||
"""Estado completo para UI / export."""
|
||
agent_rows = conn.execute(
|
||
"SELECT agent_id, role_id, relation, enabled, locked FROM agent_role_bindings"
|
||
).fetchall()
|
||
cap_rows = conn.execute(
|
||
"SELECT cap_id, role_id, enabled, locked FROM agent_governance_caps"
|
||
).fetchall()
|
||
|
||
agents: dict[str, dict[str, dict[str, dict[str, Any]]]] = {}
|
||
for r in agent_rows:
|
||
agents.setdefault(r["agent_id"], {}).setdefault(r["role_id"], {})[r["relation"]] = {
|
||
"enabled": bool(r["enabled"]),
|
||
"locked": bool(r["locked"]),
|
||
}
|
||
|
||
caps: dict[str, dict[str, dict[str, Any]]] = {}
|
||
for r in cap_rows:
|
||
caps.setdefault(r["cap_id"], {})[r["role_id"]] = {
|
||
"enabled": bool(r["enabled"]),
|
||
"locked": bool(r["locked"]),
|
||
}
|
||
|
||
return {"agents": agents, "caps": caps}
|
||
|
||
|
||
def role_agentic_caps(conn: sqlite3.Connection, role_id: str) -> list[dict[str, str]]:
|
||
from app.rbac_matrix import AGENTIC_GOVERNANCE
|
||
|
||
caps = []
|
||
for cap_id in CAP_IDS:
|
||
if cap_enabled(conn, cap_id, role_id):
|
||
caps.append({
|
||
"id": cap_id,
|
||
"label": AGENTIC_GOVERNANCE["labels"].get(cap_id, cap_id),
|
||
})
|
||
return caps
|
||
|
||
|
||
def set_agent_binding(
|
||
conn: sqlite3.Connection,
|
||
*,
|
||
agent_id: str,
|
||
role_id: str,
|
||
relation: str,
|
||
enabled: bool,
|
||
username: str,
|
||
) -> dict[str, Any]:
|
||
_, _, ROLE_COLUMNS = _spec()
|
||
if agent_id not in AGENT_IDS:
|
||
raise ValueError(f"agente inválido: {agent_id}")
|
||
if role_id not in ROLE_COLUMNS:
|
||
raise ValueError(f"função inválida: {role_id}")
|
||
if relation not in RELATIONS:
|
||
raise ValueError(f"relação inválida: {relation}")
|
||
|
||
row = conn.execute(
|
||
"""SELECT enabled, locked FROM agent_role_bindings
|
||
WHERE agent_id=? AND role_id=? AND relation=?""",
|
||
(agent_id, role_id, relation),
|
||
).fetchone()
|
||
locked = bool(row["locked"]) if row else _is_locked(agent_id, role_id, relation)
|
||
if locked and not enabled:
|
||
raise ValueError("binding bloqueado pela Spec 027 (obrigatório)")
|
||
|
||
now = _now()
|
||
if row:
|
||
conn.execute(
|
||
"""UPDATE agent_role_bindings SET enabled=?, updated_at=?, updated_by=?
|
||
WHERE agent_id=? AND role_id=? AND relation=?""",
|
||
(1 if enabled else 0, now, username, agent_id, role_id, relation),
|
||
)
|
||
else:
|
||
conn.execute(
|
||
"""INSERT INTO agent_role_bindings
|
||
(agent_id, role_id, relation, enabled, locked, updated_at, updated_by)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
||
(agent_id, role_id, relation, 1 if enabled else 0, 1 if locked else 0, now, username),
|
||
)
|
||
|
||
audit_log(
|
||
conn,
|
||
username=username,
|
||
action="agent_binding.update",
|
||
payload={
|
||
"agent_id": agent_id,
|
||
"role_id": role_id,
|
||
"relation": relation,
|
||
"enabled": enabled,
|
||
},
|
||
)
|
||
return {
|
||
"agent_id": agent_id,
|
||
"role_id": role_id,
|
||
"relation": relation,
|
||
"enabled": enabled,
|
||
"locked": locked,
|
||
}
|
||
|
||
|
||
def set_governance_cap(
|
||
conn: sqlite3.Connection,
|
||
*,
|
||
cap_id: str,
|
||
role_id: str,
|
||
enabled: bool,
|
||
username: str,
|
||
) -> dict[str, Any]:
|
||
_, _, ROLE_COLUMNS = _spec()
|
||
if cap_id not in CAP_IDS:
|
||
raise ValueError(f"capacidade inválida: {cap_id}")
|
||
if role_id not in ROLE_COLUMNS:
|
||
raise ValueError(f"função inválida: {role_id}")
|
||
|
||
row = conn.execute(
|
||
"SELECT enabled, locked FROM agent_governance_caps WHERE cap_id=? AND role_id=?",
|
||
(cap_id, role_id),
|
||
).fetchone()
|
||
locked = bool(row["locked"]) if row else False
|
||
if locked and not enabled:
|
||
raise ValueError("capacidade bloqueada")
|
||
|
||
now = _now()
|
||
if row:
|
||
conn.execute(
|
||
"""UPDATE agent_governance_caps SET enabled=?, updated_at=?, updated_by=?
|
||
WHERE cap_id=? AND role_id=?""",
|
||
(1 if enabled else 0, now, username, cap_id, role_id),
|
||
)
|
||
else:
|
||
conn.execute(
|
||
"""INSERT INTO agent_governance_caps
|
||
(cap_id, role_id, enabled, locked, updated_at, updated_by)
|
||
VALUES (?, ?, ?, 0, ?, ?)""",
|
||
(cap_id, role_id, 1 if enabled else 0, now, username),
|
||
)
|
||
|
||
audit_log(
|
||
conn,
|
||
username=username,
|
||
action="agent_cap.update",
|
||
payload={"cap_id": cap_id, "role_id": role_id, "enabled": enabled},
|
||
)
|
||
return {"cap_id": cap_id, "role_id": role_id, "enabled": enabled, "locked": locked}
|