refactor(rbac): pass scopes via role ctx

Author: jordan-umusu

frontend/src/client/schemas.gen.ts

@@ -13171,6 +13171,15 @@ export const $Role = {
       title: "Is Platform Superuser",
       default: false,
     },
+    scopes: {
+      items: {
+        type: "string",
+      },
+      type: "array",
+      uniqueItems: true,
+      title: "Scopes",
+      default: [],
+    },
   },
   type: "object",
   required: ["type", "service_id"],

frontend/src/client/types.gen.ts

@@ -4210,6 +4210,7 @@ export type Role = {
     | "tracecat-service"
     | "tracecat-ui"
   is_platform_superuser?: boolean
+  scopes?: Array<string>
 }
 
 export type type3 = "user" | "service"

tests/unit/test_rbac_scopes.py

@@ -4,6 +4,7 @@
 
 import pytest
 
+from tracecat.auth.types import Role
 from tracecat.authz.controls import (
     get_missing_scopes,
     has_all_scopes,
@@ -24,10 +25,16 @@
     PRESET_ROLE_SCOPES,
     VIEWER_SCOPES,
 )
-from tracecat.contexts import ctx_scopes
+from tracecat.contexts import ctx_role
 from tracecat.exceptions import ScopeDeniedError
 
 
+def _set_role_with_scopes(scopes: frozenset[str]) -> None:
+    """Helper to set ctx_role with the given scopes."""
+    role = Role(type="user", service_id="tracecat-api", scopes=scopes)
+    ctx_role.set(role)
+
+
 class TestValidateScopeString:
     """Tests for scope string validation."""
 
@@ -228,7 +235,7 @@ class TestRequireScopeDecorator:
     """Tests for the @require_scope decorator."""
 
     def test_require_scope_passes_with_exact_scope(self):
-        ctx_scopes.set(frozenset({"workflow:read"}))
+        _set_role_with_scopes(frozenset({"workflow:read"}))
 
         @require_scope("workflow:read")
         def protected_func():
@@ -237,7 +244,7 @@ def protected_func():
         assert protected_func() == "success"
 
     def test_require_scope_passes_with_wildcard(self):
-        ctx_scopes.set(frozenset({"workflow:*"}))
+        _set_role_with_scopes(frozenset({"workflow:*"}))
 
         @require_scope("workflow:read")
         def protected_func():
@@ -246,7 +253,7 @@ def protected_func():
         assert protected_func() == "success"
 
     def test_require_scope_passes_with_superuser(self):
-        ctx_scopes.set(frozenset({"*"}))
+        _set_role_with_scopes(frozenset({"*"}))
 
         @require_scope("org:delete")
         def protected_func():
@@ -255,7 +262,7 @@ def protected_func():
         assert protected_func() == "success"
 
     def test_require_scope_fails_without_scope(self):
-        ctx_scopes.set(frozenset({"case:read"}))
+        _set_role_with_scopes(frozenset({"case:read"}))
 
         @require_scope("workflow:read")
         def protected_func():
@@ -268,7 +275,7 @@ def protected_func():
         assert "workflow:read" in exc_info.value.missing_scopes
 
     def test_require_scope_multiple_all_required(self):
-        ctx_scopes.set(frozenset({"workflow:read", "workflow:execute"}))
+        _set_role_with_scopes(frozenset({"workflow:read", "workflow:execute"}))
 
         @require_scope("workflow:read", "workflow:execute", require_all=True)
         def protected_func():
@@ -277,7 +284,7 @@ def protected_func():
         assert protected_func() == "success"
 
     def test_require_scope_multiple_missing_one(self):
-        ctx_scopes.set(frozenset({"workflow:read"}))
+        _set_role_with_scopes(frozenset({"workflow:read"}))
 
         @require_scope("workflow:read", "workflow:execute", require_all=True)
         def protected_func():
@@ -289,7 +296,7 @@ def protected_func():
         assert "workflow:execute" in exc_info.value.missing_scopes
 
     def test_require_scope_any_one_sufficient(self):
-        ctx_scopes.set(frozenset({"workflow:read"}))
+        _set_role_with_scopes(frozenset({"workflow:read"}))
 
         @require_scope("workflow:read", "workflow:execute", require_all=False)
         def protected_func():
@@ -298,7 +305,7 @@ def protected_func():
         assert protected_func() == "success"
 
     def test_require_scope_any_none_present(self):
-        ctx_scopes.set(frozenset({"case:read"}))
+        _set_role_with_scopes(frozenset({"case:read"}))
 
         @require_scope("workflow:read", "workflow:execute", require_all=False)
         def protected_func():
@@ -307,9 +314,22 @@ def protected_func():
         with pytest.raises(ScopeDeniedError):
             protected_func()
 
+    def test_require_scope_fails_without_role(self):
+        """Test that require_scope fails when ctx_role is None."""
+        ctx_role.set(None)
+
+        @require_scope("workflow:read")
+        def protected_func():
+            return "success"
+
+        with pytest.raises(ScopeDeniedError) as exc_info:
+            protected_func()
+
+        assert "workflow:read" in exc_info.value.required_scopes
+
     @pytest.mark.anyio
     async def test_require_scope_async_function(self):
-        ctx_scopes.set(frozenset({"workflow:read"}))
+        _set_role_with_scopes(frozenset({"workflow:read"}))
 
         @require_scope("workflow:read")
         async def async_protected_func():
@@ -320,7 +340,7 @@ async def async_protected_func():
 
     @pytest.mark.anyio
     async def test_require_scope_async_function_denied(self):
-        ctx_scopes.set(frozenset({"case:read"}))
+        _set_role_with_scopes(frozenset({"case:read"}))
 
         @require_scope("workflow:read")
         async def async_protected_func():

tracecat/auth/credentials.py

@@ -34,7 +34,7 @@
 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, ctx_scopes
+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
@@ -201,6 +201,10 @@ async def _authenticate_service(
         if (org_role_str := request.headers.get("x-tracecat-role-org-role")) is not None
         else None
     )
+    # Parse scopes from header if present (for inter-service calls)
+    scopes: frozenset[str] = frozenset()
+    if scopes_header := request.headers.get("x-tracecat-role-scopes"):
+        scopes = frozenset(s.strip() for s in scopes_header.split(",") if s.strip())
     service_id: InternalServiceID = service_role_id  # type: ignore[assignment]
     return Role(
         type="service",
@@ -210,6 +214,7 @@ async def _authenticate_service(
         workspace_id=workspace_id,
         workspace_role=workspace_role,
         org_role=org_role,
+        scopes=scopes,
     )
 
 
@@ -616,15 +621,14 @@ def _validate_role(
                 status_code=status.HTTP_403_FORBIDDEN, detail="Forbidden"
             )
 
-    # Compute and set effective scopes
+    # Compute effective scopes and create new role with scopes included
     scopes = compute_effective_scopes(role)
-    ctx_scopes.set(scopes)
     logger.debug(
         "Computed effective scopes",
         scope_count=len(scopes),
     )
 
-    return role
+    return role.model_copy(update={"scopes": scopes})
 
 
 # --- Main Auth Orchestrator ---
@@ -919,7 +923,7 @@ async def _authenticated_user_only(
         # organization_id intentionally None - user may not belong to any org
     )
     scopes = compute_effective_scopes(role)
-    ctx_scopes.set(scopes)
+    role = role.model_copy(update={"scopes": scopes})
     ctx_role.set(role)
     return role
 

tracecat/auth/types.py

@@ -58,6 +58,8 @@ class Role(BaseModel):
     service_id: InternalServiceID = Field(frozen=True)
     is_platform_superuser: bool = Field(default=False, frozen=True)
     """Whether this role belongs to a platform superuser (User.is_superuser=True)."""
+    scopes: frozenset[str] = Field(default=frozenset(), frozen=True)
+    """Effective scopes for this role. Computed during authentication."""
 
     @property
     def is_superuser(self) -> bool:
@@ -95,6 +97,8 @@ def to_headers(self) -> dict[str, str]:
             headers["x-tracecat-role-workspace-role"] = self.workspace_role.value
         if self.org_role is not None:
             headers["x-tracecat-role-org-role"] = self.org_role.value
+        if self.scopes:
+            headers["x-tracecat-role-scopes"] = ",".join(sorted(self.scopes))
         return headers
 
 
@@ -125,5 +129,5 @@ def system_role() -> Role:
         type="service",
         service_id="tracecat-api",
         access_level=AccessLevel.ADMIN,
-        is_platform_superuser=True,
+        scopes=frozenset({"*"}),
     )

tracecat/authz/controls.py

@@ -2,11 +2,12 @@
 import functools
 import re
 from collections.abc import Callable, Coroutine
+from fnmatch import fnmatch
 from typing import Any, Protocol, TypeVar, cast, runtime_checkable
 
-from tracecat.auth.types import AccessLevel, Role
-from tracecat.authz.enums import WorkspaceRole
-from tracecat.contexts import ctx_scopes
+from tracecat.auth.types import Role
+from tracecat.authz.enums import OrgRole, WorkspaceRole
+from tracecat.contexts import ctx_role
 from tracecat.exceptions import ScopeDeniedError, TracecatAuthorizationError
 from tracecat.logger import logger
 
@@ -56,10 +57,8 @@ def scope_matches(granted_scope: str, required_scope: str) -> bool:
         # No wildcard - exact match required
         return granted_scope == required_scope
 
-    # Convert fnmatch-style pattern to regex
-    # Escape all regex special chars except *, then convert * to .*
-    pattern = re.escape(granted_scope).replace(r"\*", ".*")
-    return bool(re.fullmatch(pattern, required_scope))
+    # Use fnmatch for wildcard matching (avoids regex backtracking issues)
+    return fnmatch(required_scope, granted_scope)
 
 
 def has_scope(user_scopes: frozenset[str], required_scope: str) -> bool:
@@ -242,8 +241,8 @@ def wrapper(self, *args, **kwargs):
 def require_scope(*scopes: str, require_all: bool = True) -> Callable[[T], T]:
     """Decorator that requires specific scopes to access an endpoint or method.
 
-    This decorator checks the current request's scopes (from ctx_scopes) against
-    the required scopes. Platform superusers with the "*" scope bypass all checks.
+    This decorator checks the current request's scopes (from ctx_role.get().scopes)
+    against the required scopes. Platform superusers with the "*" scope bypass all checks.
 
     Args:
         *scopes: The scope(s) required for access (e.g., "workflow:read", "org:member:invite")
@@ -276,7 +275,13 @@ def check_scopes():
         if not required:
             return
 
-        user_scopes = ctx_scopes.get()
+        role = ctx_role.get()
+        if role is None:
+            raise ScopeDeniedError(
+                required_scopes=list(required), missing_scopes=list(required)
+            )
+
+        user_scopes = role.scopes
 
         # Platform superuser has "*" scope - bypass all checks
         if "*" in user_scopes:

tracecat/contexts.py

@@ -15,7 +15,6 @@
 __all__ = [
     "ctx_run",
     "ctx_role",
-    "ctx_scopes",
     "ctx_logger",
     "ctx_interaction",
     "ctx_stream_id",
@@ -27,8 +26,6 @@
 
 ctx_run: ContextVar[RunContext | None] = ContextVar("run", default=None)
 ctx_role: ContextVar[Role | None] = ContextVar("role", default=None)
-ctx_scopes: ContextVar[frozenset[str]] = ContextVar("scopes", default=frozenset())
-"""Effective scopes for the current request. Computed during authentication."""
 ctx_logger: ContextVar[loguru.Logger | None] = ContextVar("logger", default=None)
 ctx_interaction: ContextVar[InteractionContext | None] = ContextVar(
     "interaction", default=None