100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
"""FlowDeck — Agent permission guard (v4.10.0).
|
|
|
|
The agent always acts with *at most* the permissions of the invoking user
|
|
(Notion Agent principle). This manager resolves the user's role in the active
|
|
workspace and gates tool execution before any write reaches the database.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from app.db import get_conn
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Workspace roles, from least to most privileged.
|
|
READ_ROLES = {"viewer", "commenter", "editor", "admin", "owner"}
|
|
WRITE_ROLES = {"editor", "admin", "owner"}
|
|
DESTRUCTIVE_ROLES = {"admin", "owner"}
|
|
|
|
# Tools that mutate state and therefore require at least an editor role.
|
|
WRITE_TOOLS = {
|
|
"create_collection", "create_view", "create_page", "update_page",
|
|
"write_blocks", "add_property", "add_relation", "create_sub_item",
|
|
"add_dependency", "sync_gitea", "create_gitea_issue", "apply_template",
|
|
}
|
|
|
|
# Tools that delete / are destructive → admin/owner (or confirm mode).
|
|
DESTRUCTIVE_TOOLS = {
|
|
"delete_page", "delete_collection", "delete_property", "delete_view",
|
|
}
|
|
|
|
|
|
class PermissionManager:
|
|
"""Resolves workspace role and gates agent tool calls."""
|
|
|
|
def __init__(self, user_id: int):
|
|
self.user_id = user_id
|
|
|
|
# ── Role resolution ──
|
|
|
|
def role_in_workspace(self, workspace_id: int | None) -> str:
|
|
"""Return the user's role for a workspace (owner > member role)."""
|
|
if workspace_id is None:
|
|
# No workspace → fall back to the most permissive own-content model.
|
|
return "owner"
|
|
with get_conn() as conn:
|
|
member = conn.execute(
|
|
"SELECT role FROM workspace_members WHERE workspace_id=? AND user_id=?",
|
|
(workspace_id, self.user_id),
|
|
).fetchone()
|
|
if member:
|
|
return member["role"] or "editor"
|
|
owner = conn.execute(
|
|
"SELECT id FROM workspaces WHERE id=? AND owner_id=?",
|
|
(workspace_id, self.user_id),
|
|
).fetchone()
|
|
return "owner" if owner else "viewer"
|
|
|
|
def can_read(self, workspace_id: int | None) -> bool:
|
|
return self.role_in_workspace(workspace_id) in READ_ROLES
|
|
|
|
def can_write(self, workspace_id: int | None) -> bool:
|
|
return self.role_in_workspace(workspace_id) in WRITE_ROLES
|
|
|
|
def can_destructive(self, workspace_id: int | None) -> bool:
|
|
return self.role_in_workspace(workspace_id) in DESTRUCTIVE_ROLES
|
|
|
|
# ── Gate for the engine ──
|
|
|
|
def assert_can(self, tool: str, args: dict, workspace_id: int | None,
|
|
approval_mode: str = "auto") -> None:
|
|
"""Raise HTTPException if the tool call exceeds the user's permissions.
|
|
|
|
- read tools: any authenticated user in the workspace (viewer+).
|
|
- write tools: editor+.
|
|
- destructive tools: admin/owner, or requires confirm approval mode.
|
|
"""
|
|
role = self.role_in_workspace(workspace_id)
|
|
if tool in WRITE_TOOLS and role not in WRITE_ROLES:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=f"Agent tool '{tool}' requires editor+ role (user is '{role}')",
|
|
)
|
|
if tool in DESTRUCTIVE_TOOLS:
|
|
if role not in DESTRUCTIVE_ROLES:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=f"Agent tool '{tool}' is destructive and requires admin/owner "
|
|
f"(user is '{role}')",
|
|
)
|
|
if approval_mode != "confirm":
|
|
raise HTTPException(
|
|
status_code=428, # Precondition Required
|
|
detail=f"Destructive tool '{tool}' requires approval (confirm mode)",
|
|
)
|
|
# A viewer can always read; editor can read+write.
|
|
if role not in READ_ROLES:
|
|
raise HTTPException(status_code=403, detail="User has no access to this workspace") |