521 lines
22 KiB
Python
521 lines
22 KiB
Python
"""FlowDeck — Permission manager: workspace roles + granular ACL (v6.0.0).
|
|
|
|
Two layers:
|
|
|
|
1. **Workspace roles** (v4.10.0, agent guard): every user has a single role in
|
|
each workspace (owner > owner-membership > editor > commenter > viewer).
|
|
The FlowDeck Agent always acts with *at most* the permissions of the
|
|
invoking user (Notion Agent principle).
|
|
|
|
2. **Granular permissions** (v6.0.0): explicit page / collection / property
|
|
grants plus reusable user groups. Resolution follows the least-privilege
|
|
rule — an explicit grant on a resource overrides the inherited chain
|
|
(page → collection → workspace), while ``restricted`` / ``private``
|
|
resources deny access unless a grant (or the workspace owner / admin)
|
|
applies.
|
|
|
|
Resolution results are cached for 60 s to keep the hot paths (sidebar, view
|
|
rendering, route guards) < 10 ms per check; ``PermissionManager.invalidate()``
|
|
drops the cache after any grant/revoke/type change.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
|
|
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"}
|
|
|
|
# Granular resource roles (ranked, least → most privileged).
|
|
_GRANULAR_ROLES = ("viewer", "commenter", "editor", "owner")
|
|
_ROLE_RANK = {role: i for i, role in enumerate(_GRANULAR_ROLES)}
|
|
_PROPERTY_ROLES = ("viewer", "editor")
|
|
_PROPERTY_RANK = {"viewer": 0, "editor": 1}
|
|
|
|
# Tools that mutate state and therefore require at least an editor role.
|
|
WRITE_TOOLS = {
|
|
"create_collection", "create_view", "create_page", "update_page",
|
|
"write_blocks", "create_document", "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_document",
|
|
"delete_property", "delete_view",
|
|
}
|
|
|
|
|
|
class PermissionManager:
|
|
"""Resolves workspace role and gates agent + granular ACL checks."""
|
|
|
|
def __init__(self, user_id: int, is_admin: bool = False):
|
|
self.user_id = user_id
|
|
self._is_admin_override = bool(is_admin)
|
|
self._cache: dict[str, tuple[float, object]] = {}
|
|
|
|
# ── Cache helpers ──
|
|
|
|
def _cached(self, key: str, ttl: float, fn):
|
|
now = time.monotonic()
|
|
hit = self._cache.get(key)
|
|
if hit and now - hit[0] < ttl:
|
|
return hit[1]
|
|
val = fn()
|
|
self._cache[key] = (now, val)
|
|
return val
|
|
|
|
def invalidate(self) -> None:
|
|
"""Drop the resolution cache after a grant/revoke/type change."""
|
|
self._cache.clear()
|
|
|
|
# ── 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"
|
|
|
|
# ── SSO (v6.7.0, design §7.2) ─────────────────────────────────────────
|
|
|
|
def is_sso_only_workspace(self, workspace_id: int | None = None) -> bool:
|
|
"""True when that workspace can only be reached through SSO.
|
|
|
|
FlowDeck keeps a single instance-wide SSO-only switch (design §7.1 /
|
|
§4.2): when it is on, local login is refused for every non-admin, so
|
|
every workspace on the instance is effectively SSO-only.
|
|
``workspace_id`` is accepted to mirror the design's per-workspace API.
|
|
"""
|
|
from app.services.sso_provisioning import is_sso_only
|
|
|
|
return is_sso_only()
|
|
|
|
def _user_auth_method(self) -> str:
|
|
with get_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT auth_method FROM users WHERE id=?", (self.user_id,)
|
|
).fetchone()
|
|
return (row["auth_method"] or "local") if row else "local"
|
|
|
|
def get_sso_roles(
|
|
self, user_id: int | None = None, workspace_id: int | None = None
|
|
) -> list[str]:
|
|
"""Roles granted to that user through SSO group mapping (design §7.2).
|
|
|
|
SSO grants land in the regular ``workspace_members`` row (the mapping
|
|
is re-applied at every SSO login), so the answer is the explicit
|
|
membership role of a non-local account — local accounts and users
|
|
without an explicit grant (the implicit *viewer* fallback is not an
|
|
SSO grant) get ``[]``.
|
|
"""
|
|
if workspace_id is None:
|
|
return []
|
|
pm = PermissionManager(int(user_id)) if (user_id and int(user_id) != self.user_id) else self
|
|
if pm._user_auth_method() == "local":
|
|
return []
|
|
with get_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT role FROM workspace_members WHERE workspace_id=? AND user_id=?",
|
|
(int(workspace_id), pm.user_id),
|
|
).fetchone()
|
|
return [row["role"]] if row else []
|
|
|
|
def sync_sso_permissions(
|
|
self,
|
|
user_id: int | None,
|
|
sso_groups: list[str],
|
|
workspace_id: int | None = None,
|
|
) -> list[int]:
|
|
"""Re-apply the group → workspace role mapping (design §7.2).
|
|
|
|
Delegates to ``sso_provisioning.sync_sso_groups`` (the single source
|
|
of truth used at login and by ``POST /api/v2/sso/sync``). Returns the
|
|
touched workspace ids, narrowed to ``workspace_id`` when given.
|
|
"""
|
|
from app.services.sso_provisioning import get_sso_config, sync_sso_groups
|
|
|
|
cfg = get_sso_config()
|
|
if not cfg:
|
|
return []
|
|
touched = sync_sso_groups(int(user_id or self.user_id), list(sso_groups or []), cfg)
|
|
if workspace_id is not None:
|
|
touched = [w for w in touched if int(w) == int(workspace_id)]
|
|
if touched:
|
|
self.invalidate()
|
|
return touched
|
|
|
|
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")
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Granular permissions (v6.0.0)
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
def _is_admin(self, conn) -> bool:
|
|
if self._is_admin_override:
|
|
return True
|
|
row = conn.execute(
|
|
"SELECT is_admin FROM users WHERE id=?", (self.user_id,)
|
|
).fetchone()
|
|
return bool(row and row["is_admin"])
|
|
|
|
def _owns_workspace(self, conn, workspace_id: int | None) -> bool:
|
|
if workspace_id is None:
|
|
# No workspace → single-user semantics: the actor is the owner.
|
|
return True
|
|
row = conn.execute(
|
|
"SELECT id FROM workspaces WHERE id=? AND owner_id=?",
|
|
(workspace_id, self.user_id),
|
|
).fetchone()
|
|
return bool(row)
|
|
|
|
def user_group_ids(self, conn) -> list[int]:
|
|
return [
|
|
r["group_id"]
|
|
for r in conn.execute(
|
|
"SELECT group_id FROM group_members WHERE user_id=?", (self.user_id,)
|
|
).fetchall()
|
|
]
|
|
|
|
def _explicit_grant_role(self, conn, table: str, fk: str, resource_id: int,
|
|
role_rank: dict[str, int] | None = None) -> str | None:
|
|
"""Most-privileged explicit role on ``table`` for the user / groups."""
|
|
rank = role_rank or _ROLE_RANK
|
|
groups = self.user_group_ids(conn)
|
|
if groups:
|
|
placeholders = ", ".join("?" * len(groups))
|
|
rows = conn.execute(
|
|
f"SELECT role FROM {table} WHERE {fk}=? "
|
|
f"AND (user_id=? OR group_id IN ({placeholders}))",
|
|
(resource_id, self.user_id, *groups),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
f"SELECT role FROM {table} WHERE {fk}=? AND user_id=?",
|
|
(resource_id, self.user_id),
|
|
).fetchall()
|
|
best = max((rank.get(r["role"], -1) for r in rows), default=-1)
|
|
if best < 0:
|
|
return None
|
|
rev = {rank[k]: k for k in rank}
|
|
return rev[best]
|
|
|
|
# ── Page-level ──
|
|
|
|
def get_page_permission(self, page_id: int) -> str | None:
|
|
"""Effective page role for ``self.user_id`` (least privilege).
|
|
|
|
Chain: explicit page grant > explicit collection grant > workspace
|
|
role. ``restricted`` / ``private`` pages ignore the inherited chain.
|
|
Returns ``None`` when the user must not see the page at all.
|
|
"""
|
|
|
|
def _resolve() -> str | None:
|
|
with get_conn() as conn:
|
|
page = conn.execute(
|
|
"SELECT permission_type, workspace_id, collection_id FROM pages WHERE id=?",
|
|
(page_id,),
|
|
).fetchone()
|
|
if not page:
|
|
return None
|
|
if self._is_admin(conn) or self._owns_workspace(conn, page["workspace_id"]):
|
|
return "owner"
|
|
explicit = self._explicit_grant_role(
|
|
conn, "page_permissions", "page_id", page_id
|
|
)
|
|
if explicit:
|
|
return explicit
|
|
ptype = page["permission_type"] or "inherit"
|
|
if ptype in ("restricted", "private"):
|
|
return None
|
|
if page["collection_id"]:
|
|
coll_role = self._collection_role(conn, page["collection_id"])
|
|
if coll_role:
|
|
return coll_role
|
|
return self.role_in_workspace(page["workspace_id"])
|
|
return self._cached(f"page:{page_id}", 60, _resolve)
|
|
|
|
def can_view_page(self, page_id: int) -> bool:
|
|
return self.get_page_permission(page_id) is not None
|
|
|
|
def can_edit_page(self, page_id: int) -> bool:
|
|
role = self.get_page_permission(page_id)
|
|
return bool(role and _ROLE_RANK[role] >= _ROLE_RANK["editor"])
|
|
|
|
def can_comment_page(self, page_id: int) -> bool:
|
|
role = self.get_page_permission(page_id)
|
|
return bool(role and _ROLE_RANK[role] >= _ROLE_RANK["commenter"])
|
|
|
|
def can_manage_page_permissions(self, page_id: int) -> bool:
|
|
role = self.get_page_permission(page_id)
|
|
return bool(role and _ROLE_RANK[role] >= _ROLE_RANK["owner"])
|
|
|
|
# ── Collection-level ──
|
|
|
|
def _collection_role(self, conn, collection_id: int) -> str | None:
|
|
coll = conn.execute(
|
|
"SELECT permission_type, workspace_id FROM collections WHERE id=?",
|
|
(collection_id,),
|
|
).fetchone()
|
|
if not coll:
|
|
return None
|
|
if self._is_admin(conn) or self._owns_workspace(conn, coll["workspace_id"]):
|
|
return "owner"
|
|
explicit = self._explicit_grant_role(
|
|
conn, "collection_permissions", "collection_id", collection_id
|
|
)
|
|
if explicit:
|
|
return explicit
|
|
ptype = coll["permission_type"] or "inherit"
|
|
if ptype in ("restricted", "private"):
|
|
return None
|
|
return self.role_in_workspace(coll["workspace_id"])
|
|
|
|
def get_collection_permission(self, collection_id: int) -> str | None:
|
|
def _resolve() -> str | None:
|
|
with get_conn() as conn:
|
|
return self._collection_role(conn, collection_id)
|
|
return self._cached(f"collection:{collection_id}", 60, _resolve)
|
|
|
|
def can_view_collection(self, collection_id: int) -> bool:
|
|
return self.get_collection_permission(collection_id) is not None
|
|
|
|
def can_edit_collection(self, collection_id: int) -> bool:
|
|
role = self.get_collection_permission(collection_id)
|
|
return bool(role and _ROLE_RANK[role] >= _ROLE_RANK["editor"])
|
|
|
|
def can_manage_collection_permissions(self, collection_id: int) -> bool:
|
|
role = self.get_collection_permission(collection_id)
|
|
return bool(role and _ROLE_RANK[role] >= _ROLE_RANK["owner"])
|
|
|
|
# ── Property-level ──
|
|
|
|
def _property_grants_exist(self, conn, property_id: int) -> bool:
|
|
row = conn.execute(
|
|
"SELECT 1 FROM property_permissions WHERE property_id=? LIMIT 1",
|
|
(property_id,),
|
|
).fetchone()
|
|
return row is not None
|
|
|
|
def _has_property_grant(self, conn, property_id: int, min_rank: int) -> bool:
|
|
groups = self.user_group_ids(conn)
|
|
if groups:
|
|
placeholders = ", ".join("?" * len(groups))
|
|
rows = conn.execute(
|
|
f"SELECT role FROM property_permissions WHERE property_id=? "
|
|
f"AND (user_id=? OR group_id IN ({placeholders}))",
|
|
(property_id, self.user_id, *groups),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"SELECT role FROM property_permissions WHERE property_id=? AND user_id=?",
|
|
(property_id, self.user_id),
|
|
).fetchall()
|
|
return any(_PROPERTY_RANK.get(r["role"], -1) >= min_rank for r in rows)
|
|
|
|
def can_view_property(self, collection_id: int, property_id: int) -> bool:
|
|
"""A property is visible unless it carries explicit grants excluding
|
|
the user; without any grant it inherits from the collection. Collection
|
|
owners/admins always see every property."""
|
|
if not self.can_view_collection(collection_id):
|
|
return False
|
|
return self._cached(
|
|
f"prop:{property_id}", 60, lambda: self._property_visible(collection_id, property_id)
|
|
)
|
|
|
|
def _collection_workspace_id(self, conn, collection_id: int) -> int | None:
|
|
row = conn.execute(
|
|
"SELECT workspace_id FROM collections WHERE id=?", (collection_id,)
|
|
).fetchone()
|
|
return row["workspace_id"] if row else None
|
|
|
|
def _property_visible(self, collection_id: int, property_id: int) -> bool:
|
|
with get_conn() as conn:
|
|
workspace_id = self._collection_workspace_id(conn, collection_id)
|
|
if self._is_admin(conn) or self._owns_workspace(conn, workspace_id):
|
|
return True
|
|
if self.can_manage_collection_permissions(collection_id):
|
|
return True
|
|
if not self._property_grants_exist(conn, property_id):
|
|
return True
|
|
return self._has_property_grant(conn, property_id, _PROPERTY_RANK["viewer"])
|
|
|
|
def can_edit_property(self, collection_id: int, property_id: int) -> bool:
|
|
if not self.can_edit_collection(collection_id):
|
|
return False
|
|
with get_conn() as conn:
|
|
workspace_id = self._collection_workspace_id(conn, collection_id)
|
|
if self._is_admin(conn) or self._owns_workspace(conn, workspace_id):
|
|
return True
|
|
if self.can_manage_collection_permissions(collection_id):
|
|
return True
|
|
if not self._property_grants_exist(conn, property_id):
|
|
return True
|
|
return self._has_property_grant(conn, property_id, _PROPERTY_RANK["editor"])
|
|
|
|
def get_visible_properties(self, collection_id: int) -> list[int]:
|
|
def _resolve() -> list[int]:
|
|
with get_conn() as conn:
|
|
props = conn.execute(
|
|
"SELECT id FROM collection_properties WHERE collection_id=?",
|
|
(collection_id,),
|
|
).fetchall()
|
|
return [p["id"] for p in props if self.can_view_property(collection_id, p["id"])]
|
|
return self._cached(f"visible_props:{collection_id}", 60, _resolve)
|
|
|
|
# ── Groups ──
|
|
|
|
def is_workspace_admin(self, workspace_id: int | None) -> bool:
|
|
with get_conn() as conn:
|
|
return self._is_admin(conn) or self._owns_workspace(conn, workspace_id)
|
|
|
|
def create_group(self, workspace_id: int | None, name: str,
|
|
description: str = "", created_by: int | None = None) -> int:
|
|
if not self.is_workspace_admin(workspace_id):
|
|
raise HTTPException(403, "Only a workspace owner or admin can create groups")
|
|
if not name.strip():
|
|
raise HTTPException(400, "name is required")
|
|
with get_conn() as conn:
|
|
dupe = conn.execute(
|
|
"SELECT id FROM user_groups WHERE workspace_id IS ? AND name=?",
|
|
(workspace_id, name.strip()),
|
|
).fetchone()
|
|
if dupe:
|
|
raise HTTPException(400, "A group with this name already exists")
|
|
cur = conn.execute(
|
|
"INSERT INTO user_groups (workspace_id, name, description, created_by) "
|
|
"VALUES (?, ?, ?, ?)",
|
|
(workspace_id, name.strip(), description or "", created_by),
|
|
)
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
def add_user_to_group(self, group_id: int, user_id: int) -> None:
|
|
with get_conn() as conn:
|
|
group = conn.execute(
|
|
"SELECT workspace_id FROM user_groups WHERE id=?", (group_id,)
|
|
).fetchone()
|
|
if not group:
|
|
raise HTTPException(404, "Group not found")
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO group_members (group_id, user_id) VALUES (?, ?)",
|
|
(group_id, user_id),
|
|
)
|
|
conn.commit()
|
|
|
|
def remove_user_from_group(self, group_id: int, user_id: int) -> None:
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"DELETE FROM group_members WHERE group_id=? AND user_id=?",
|
|
(group_id, user_id),
|
|
)
|
|
conn.commit()
|
|
|
|
def delete_group(self, group_id: int) -> None:
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM user_groups WHERE id=?", (group_id,))
|
|
conn.commit()
|
|
|
|
def get_groups_for_workspace(self, workspace_id: int | None) -> list[dict]:
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"""SELECT g.id, g.name, g.description, g.created_by, g.created_at,
|
|
(SELECT COUNT(*) FROM group_members m WHERE m.group_id=g.id) AS member_count
|
|
FROM user_groups g WHERE g.workspace_id IS ? ORDER BY g.name""",
|
|
(workspace_id,),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def get_group_members(self, group_id: int) -> list[dict]:
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"""SELECT u.id, u.login, u.full_name, u.email, m.joined_at
|
|
FROM group_members m JOIN users u ON u.id=m.user_id
|
|
WHERE m.group_id=? ORDER BY u.login""",
|
|
(group_id,),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
# ── Audit log ──
|
|
|
|
def log_permission_change(self, resource_type: str, resource_id: int, action: str,
|
|
target_user_id: int | None = None,
|
|
target_group_id: int | None = None,
|
|
old_role: str | None = None,
|
|
new_role: str | None = None,
|
|
ip_address: str = "") -> None:
|
|
"""Write one immutable audit row for a permission change."""
|
|
try:
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO permission_audit_log
|
|
(resource_type, resource_id, action, target_user_id, target_group_id,
|
|
old_role, new_role, performed_by, ip_address)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(resource_type, resource_id, action, target_user_id, target_group_id,
|
|
old_role, new_role, self.user_id, ip_address),
|
|
)
|
|
conn.commit()
|
|
except Exception as exc: # audit must never break the caller
|
|
logger.warning("permission audit log failed: %s", exc)
|