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