refactor(rbac): query role and group tables to compute scopes

Author: jordan-umusu

tests/unit/test_auth_middleware.py

@@ -557,7 +557,11 @@ async def test_organization_id_populated_when_require_workspace_no(mocker):
     org_role_result = MagicMock()
     org_role_result.scalar_one_or_none.return_value = None
 
-    mock_session.execute.side_effect = [org_result, org_role_result]
+    # Third call: compute_effective_scopes query returns empty scopes
+    scopes_result = MagicMock()
+    scopes_result.scalars.return_value.all.return_value = []
+
+    mock_session.execute.side_effect = [org_result, org_role_result, scopes_result]
 
     # Mock is_unprivileged to return False for admin users
     mocker.patch("tracecat.auth.credentials.is_unprivileged", return_value=False)

tests/unit/test_rbac_scopes.py

@@ -14,7 +14,6 @@
     scope_matches,
     validate_scope_string,
 )
-from tracecat.authz.enums import OrgRole, WorkspaceRole
 from tracecat.authz.scopes import (
     ADMIN_SCOPES,
     EDITOR_SCOPES,
@@ -202,9 +201,9 @@ def test_admin_includes_editor(self):
         assert EDITOR_SCOPES.issubset(ADMIN_SCOPES)
 
     def test_system_role_mapping(self):
-        assert PRESET_ROLE_SCOPES[WorkspaceRole.VIEWER] == VIEWER_SCOPES
-        assert PRESET_ROLE_SCOPES[WorkspaceRole.EDITOR] == EDITOR_SCOPES
-        assert PRESET_ROLE_SCOPES[WorkspaceRole.ADMIN] == ADMIN_SCOPES
+        assert PRESET_ROLE_SCOPES["workspace-viewer"] == VIEWER_SCOPES
+        assert PRESET_ROLE_SCOPES["workspace-editor"] == EDITOR_SCOPES
+        assert PRESET_ROLE_SCOPES["workspace-admin"] == ADMIN_SCOPES
 
 
 class TestOrgRoleScopes:
@@ -226,9 +225,9 @@ def test_member_has_minimal_scopes(self):
         assert ORG_MEMBER_SCOPES == frozenset({"org:read", "org:member:read"})
 
     def test_org_role_mapping(self):
-        assert ORG_ROLE_SCOPES[OrgRole.OWNER] == ORG_OWNER_SCOPES
-        assert ORG_ROLE_SCOPES[OrgRole.ADMIN] == ORG_ADMIN_SCOPES
-        assert ORG_ROLE_SCOPES[OrgRole.MEMBER] == ORG_MEMBER_SCOPES
+        assert ORG_ROLE_SCOPES["organization-owner"] == ORG_OWNER_SCOPES
+        assert ORG_ROLE_SCOPES["organization-admin"] == ORG_ADMIN_SCOPES
+        assert ORG_ROLE_SCOPES["organization-member"] == ORG_MEMBER_SCOPES
 
 
 class TestRequireScopeDecorator:

tracecat/auth/credentials.py

@@ -32,12 +32,24 @@
     optional_current_active_user,
 )
 from tracecat.authz.enums import OrgRole, WorkspaceRole
-from tracecat.authz.scopes import ORG_ROLE_SCOPES, PRESET_ROLE_SCOPES
 from tracecat.authz.service import MembershipService, MembershipWithOrg
 from tracecat.contexts import ctx_role
 from tracecat.db.dependencies import AsyncDBSession
 from tracecat.db.engine import get_async_session_context_manager
-from tracecat.db.models import Organization, OrganizationMembership, User, Workspace
+from tracecat.db.models import (
+    GroupMember,
+    GroupRoleAssignment,
+    Organization,
+    OrganizationMembership,
+    RoleScope,
+    Scope,
+    User,
+    UserRoleAssignment,
+    Workspace,
+)
+from tracecat.db.models import (
+    Role as RoleModel,
+)
 from tracecat.identifiers import InternalServiceID
 from tracecat.logger import logger
 from tracecat.organization.management import get_default_organization_id
@@ -85,42 +97,66 @@ async def _get_workspace_org_id(workspace_id: uuid.UUID) -> uuid.UUID | None:
 ORG_OVERRIDE_COOKIE = "tracecat-org-id"
 
 
-def compute_effective_scopes(role: Role) -> frozenset[str]:
-    """Compute the effective scopes for a role.
+async def compute_effective_scopes(
+    session: AsyncDBSession, role: Role
+) -> frozenset[str]:
+    """Compute the effective scopes for a role from the database.
+
+    Queries UserRoleAssignment and GroupRoleAssignment tables to resolve
+    scopes through Role → RoleScope → Scope.
 
     Scope computation follows this hierarchy:
     1. Platform superusers get "*" (all scopes)
-    2. Org OWNER/ADMIN get their org-level scopes (includes full workspace access)
-    3. Org MEMBER gets base org scopes + workspace membership scopes (if in workspace)
-    4. Service roles inherit scopes based on the user they're acting on behalf of
-
-    For workspace-scoped requests:
-    - Org OWNER/ADMIN: org-level scopes (they can access all workspaces)
-    - Workspace members: workspace role scopes from PRESET_ROLE_SCOPES
-
-    Note: Group-based scopes will be added in PR 4 (RBAC Service & APIs).
+    2. Direct user role assignments (org-wide and workspace-specific)
+    3. Group role assignments (org-wide and workspace-specific)
     """
     if role.is_platform_superuser:
         return frozenset({"*"})
 
-    scope_set: set[str] = set()
-
-    # Add org-level scopes based on org role
-    if role.org_role is not None:
-        scope_set |= ORG_ROLE_SCOPES.get(role.org_role, set())
-
-    # For workspace-scoped requests, add workspace role scopes
-    # (only if not an org admin/owner, who already have full access via org scopes)
-    if role.workspace_id and role.workspace_role:
-        # Org admins/owners already have workspace scopes via their org role
-        # Regular members need their workspace role scopes
-        if not role.is_org_admin:
-            scope_set |= PRESET_ROLE_SCOPES.get(role.workspace_role, set())
+    if role.user_id is None or role.organization_id is None:
+        return frozenset()
+
+    # Direct user role assignments → Role → RoleScope → Scope
+    user_scopes = (
+        select(Scope.name)
+        .join(RoleScope, RoleScope.scope_id == Scope.id)
+        .join(RoleModel, RoleModel.id == RoleScope.role_id)
+        .join(UserRoleAssignment, UserRoleAssignment.role_id == RoleModel.id)
+        .where(
+            UserRoleAssignment.user_id == role.user_id,
+            UserRoleAssignment.organization_id == role.organization_id,
+            (
+                UserRoleAssignment.workspace_id.is_(None)
+                | (UserRoleAssignment.workspace_id == role.workspace_id)
+            )
+            if role.workspace_id
+            else UserRoleAssignment.workspace_id.is_(None),
+        )
+    )
 
-    # Note: Group-based scopes (from group_assignment table) will be added in PR 4
-    # via RBACService.get_group_scopes()
+    # Group role assignments → GroupMember → GroupRoleAssignment → Role → RoleScope → Scope
+    group_scopes = (
+        select(Scope.name)
+        .join(RoleScope, RoleScope.scope_id == Scope.id)
+        .join(RoleModel, RoleModel.id == RoleScope.role_id)
+        .join(GroupRoleAssignment, GroupRoleAssignment.role_id == RoleModel.id)
+        .join(GroupMember, GroupMember.group_id == GroupRoleAssignment.group_id)
+        .where(
+            GroupMember.user_id == role.user_id,
+            GroupRoleAssignment.organization_id == role.organization_id,
+            (
+                GroupRoleAssignment.workspace_id.is_(None)
+                | (GroupRoleAssignment.workspace_id == role.workspace_id)
+            )
+            if role.workspace_id
+            else GroupRoleAssignment.workspace_id.is_(None),
+        )
+    )
 
-    return frozenset(scope_set)
+    # Single atomic query: union both assignment paths
+    combined = user_scopes.union(group_scopes)
+    result = await session.execute(combined)
+    return frozenset(result.scalars().all())
 
 
 def get_role_from_user(
@@ -599,7 +635,8 @@ async def _authenticate_executor(
     return role
 
 
-def _validate_role(
+async def _validate_role(
+    session: AsyncSession,
     role: Role,
     *,
     require_workspace: Literal["yes", "no", "optional"],
@@ -631,7 +668,7 @@ def _validate_role(
             )
 
     # Compute effective scopes and create new role with scopes included
-    scopes = compute_effective_scopes(role)
+    scopes = await compute_effective_scopes(session, role)
     logger.debug(
         "Computed effective scopes",
         scope_count=len(scopes),
@@ -701,7 +738,8 @@ async def _role_dependency(
             detail="Unauthorized",
         )
     # Validate structural requirements and set context
-    role = _validate_role(
+    role = await _validate_role(
+        session,
         role,
         require_workspace=require_workspace,
         min_access_level=min_access_level,
@@ -911,6 +949,7 @@ async def _require_superuser(
 
 async def _authenticated_user_only(
     user: Annotated[User, Depends(current_active_user)],
+    session: AsyncDBSession,
 ) -> Role:
     """Dependency for endpoints requiring only an authenticated user.
 
@@ -931,7 +970,7 @@ async def _authenticated_user_only(
         is_platform_superuser=user.is_superuser,
         # organization_id intentionally None - user may not belong to any org
     )
-    scopes = compute_effective_scopes(role)
+    scopes = await compute_effective_scopes(session, role)
     role = role.model_copy(update={"scopes": scopes})
     ctx_role.set(role)
     return role

tracecat/authz/scopes.py

@@ -16,8 +16,6 @@
 
 from __future__ import annotations
 
-from tracecat.authz.enums import OrgRole, WorkspaceRole
-
 # =============================================================================
 # Viewer Role Scopes (read-only)
 # =============================================================================
@@ -86,10 +84,10 @@
 # Preset Role -> Scope Set Mapping
 # =============================================================================
 
-PRESET_ROLE_SCOPES: dict[WorkspaceRole, frozenset[str]] = {
-    WorkspaceRole.VIEWER: VIEWER_SCOPES,
-    WorkspaceRole.EDITOR: EDITOR_SCOPES,
-    WorkspaceRole.ADMIN: ADMIN_SCOPES,
+PRESET_ROLE_SCOPES: dict[str, frozenset[str]] = {
+    "workspace-viewer": VIEWER_SCOPES,
+    "workspace-editor": EDITOR_SCOPES,
+    "workspace-admin": ADMIN_SCOPES,
 }
 
 # =============================================================================
@@ -225,8 +223,8 @@
 # Organization Role -> Scope Set Mapping
 # =============================================================================
 
-ORG_ROLE_SCOPES: dict[OrgRole, frozenset[str]] = {
-    OrgRole.OWNER: ORG_OWNER_SCOPES,
-    OrgRole.ADMIN: ORG_ADMIN_SCOPES,
-    OrgRole.MEMBER: ORG_MEMBER_SCOPES,
+ORG_ROLE_SCOPES: dict[str, frozenset[str]] = {
+    "organization-owner": ORG_OWNER_SCOPES,
+    "organization-admin": ORG_ADMIN_SCOPES,
+    "organization-member": ORG_MEMBER_SCOPES,
 }