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.
132 lines
3.9 KiB
Python
132 lines
3.9 KiB
Python
"""Rotas RBAC — Spec 027-UI / UI-C."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sqlite3
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from pydantic import BaseModel, Field
|
||
|
||
from app import agent_bindings, auth
|
||
from app.rbac_matrix import matrix_export
|
||
|
||
router = APIRouter(prefix="/api/v1", tags=["rbac"])
|
||
|
||
|
||
def _access_matrix_ui_enabled() -> bool:
|
||
return os.getenv("ACCESS_MATRIX_UI", "0").strip().lower() in ("1", "true", "yes", "on")
|
||
|
||
|
||
def _access_matrix_edit_enabled() -> bool:
|
||
return os.getenv("ACCESS_MATRIX_EDIT", "0").strip().lower() in ("1", "true", "yes", "on")
|
||
|
||
|
||
def _db():
|
||
conn = auth.db()
|
||
try:
|
||
yield conn
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _require_matrix_view(user: auth.DeskUser) -> None:
|
||
if not _access_matrix_ui_enabled():
|
||
raise HTTPException(404, "Matriz de acessos não activa neste ambiente")
|
||
if user.role != "super_admin":
|
||
raise HTTPException(403, "Matriz: apenas super_admin")
|
||
|
||
|
||
def _require_matrix_edit(user: auth.DeskUser) -> None:
|
||
_require_matrix_view(user)
|
||
if not _access_matrix_edit_enabled():
|
||
raise HTTPException(403, "Edição da matriz desactivada neste ambiente")
|
||
|
||
|
||
@router.get("/config/features")
|
||
def config_features(user: auth.DeskUser = Depends(auth.get_current_user)):
|
||
"""Feature flags expostas ao frontend (sem segredos)."""
|
||
enabled = _access_matrix_ui_enabled()
|
||
return {
|
||
"access_matrix_ui": enabled,
|
||
"access_matrix_preview": enabled and not _access_matrix_edit_enabled(),
|
||
"access_matrix_edit": enabled and _access_matrix_edit_enabled(),
|
||
}
|
||
|
||
|
||
@router.get("/rbac/matrix")
|
||
def rbac_matrix(
|
||
user: auth.DeskUser = Depends(auth.get_current_user),
|
||
conn: sqlite3.Connection = Depends(_db),
|
||
):
|
||
"""Matriz completa — Spec 027 (read-only ou editável)."""
|
||
_require_matrix_view(user)
|
||
return matrix_export(conn)
|
||
|
||
|
||
class AgentBindingPatch(BaseModel):
|
||
agent_id: str = Field(..., min_length=2, max_length=8)
|
||
role_id: str = Field(..., min_length=2, max_length=32)
|
||
relation: str = Field(..., pattern="^(ui|focus|approve)$")
|
||
enabled: bool
|
||
|
||
|
||
class GovernanceCapPatch(BaseModel):
|
||
cap_id: str = Field(..., min_length=3, max_length=32)
|
||
role_id: str = Field(..., min_length=2, max_length=32)
|
||
enabled: bool
|
||
|
||
|
||
@router.patch("/rbac/agent-bindings")
|
||
def patch_agent_binding(
|
||
body: AgentBindingPatch,
|
||
user: auth.DeskUser = Depends(auth.get_current_user),
|
||
conn: sqlite3.Connection = Depends(_db),
|
||
):
|
||
"""Ligar/desligar atribuição agente × função × relação."""
|
||
_require_matrix_edit(user)
|
||
try:
|
||
result = agent_bindings.set_agent_binding(
|
||
conn,
|
||
agent_id=body.agent_id,
|
||
role_id=body.role_id,
|
||
relation=body.relation,
|
||
enabled=body.enabled,
|
||
username=user.username,
|
||
)
|
||
conn.commit()
|
||
except ValueError as exc:
|
||
raise HTTPException(400, str(exc)) from exc
|
||
return result
|
||
|
||
|
||
@router.patch("/rbac/agent-governance")
|
||
def patch_agent_governance(
|
||
body: GovernanceCapPatch,
|
||
user: auth.DeskUser = Depends(auth.get_current_user),
|
||
conn: sqlite3.Connection = Depends(_db),
|
||
):
|
||
"""Ligar/desligar capacidade global Agentics por função."""
|
||
_require_matrix_edit(user)
|
||
try:
|
||
result = agent_bindings.set_governance_cap(
|
||
conn,
|
||
cap_id=body.cap_id,
|
||
role_id=body.role_id,
|
||
enabled=body.enabled,
|
||
username=user.username,
|
||
)
|
||
conn.commit()
|
||
except ValueError as exc:
|
||
raise HTTPException(400, str(exc)) from exc
|
||
return result
|
||
|
||
|
||
@router.get("/rbac/agent-bindings/audit")
|
||
def agent_bindings_audit(
|
||
user: auth.DeskUser = Depends(auth.get_current_user),
|
||
conn: sqlite3.Connection = Depends(_db),
|
||
limit: int = Query(50, ge=1, le=200),
|
||
):
|
||
_require_matrix_view(user)
|
||
return {"entries": agent_bindings.list_audit(conn, limit=limit)}
|