feat(rbac): add @require_scope decorators to all API endpoints

Author: jordan-umusu

tests/unit/api/conftest.py

@@ -32,11 +32,6 @@
     WorkspaceUser as TablesWorkspaceUser,
 )
 from tracecat.workspaces.router import (
-    OrgAdminUser as WorkspacesOrgAdminUser,
-)
-from tracecat.workspaces.router import (
-    OrgUser,
-    WorkspaceAdminUserInPath,
     WorkspaceUserInPath,
 )
 
@@ -63,12 +58,9 @@ def client() -> Generator[TestClient, None, None]:
         ExecutorWorkspaceRole,
         WorkspaceUser,
         WorkspaceUserInPath,
-        WorkspaceAdminUserInPath,
         SuperuserRole,
         OrganizationUserRole,
         OrganizationAdminUserRole,
-        OrgUser,
-        WorkspacesOrgAdminUser,
         SecretsWorkspaceUser,
         WorkspaceAdminUser,
         OrgAdminUser,

tests/unit/test_rbac_scopes.py

@@ -9,6 +9,7 @@
     has_all_scopes,
     has_any_scope,
     has_scope,
+    require_action_scope,
     require_scope,
     scope_matches,
     validate_scope_string,
@@ -328,3 +329,81 @@ async def async_protected_func():
 
         with pytest.raises(ScopeDeniedError):
             await async_protected_func()
+
+
+class TestRequireActionScope:
+    """Tests for the require_action_scope function."""
+
+    def test_require_action_scope_with_exact_scope(self):
+        """User with exact action scope can execute."""
+        ctx_scopes.set(frozenset({"action:core.http_request:execute"}))
+        # Should not raise
+        require_action_scope("core.http_request")
+
+    def test_require_action_scope_with_global_wildcard(self):
+        """Superuser with * scope can execute any action."""
+        ctx_scopes.set(frozenset({"*"}))
+        require_action_scope("core.http_request")
+        require_action_scope("tools.okta.list_users")
+
+    def test_require_action_scope_with_action_wildcard(self):
+        """User with action:*:execute can execute any action."""
+        ctx_scopes.set(frozenset({"action:*:execute"}))
+        require_action_scope("core.http_request")
+        require_action_scope("tools.okta.list_users")
+
+    def test_require_action_scope_with_prefix_wildcard(self):
+        """User with action:core.*:execute can execute core actions."""
+        ctx_scopes.set(frozenset({"action:core.*:execute"}))
+        require_action_scope("core.http_request")
+        require_action_scope("core.transform.forward")
+
+        # Should fail for non-core actions
+        with pytest.raises(ScopeDeniedError) as exc_info:
+            require_action_scope("tools.okta.list_users")
+        assert "action:tools.okta.list_users:execute" in exc_info.value.missing_scopes
+
+    def test_require_action_scope_with_integration_wildcard(self):
+        """User with action:tools.okta.*:execute can execute okta actions."""
+        ctx_scopes.set(frozenset({"action:tools.okta.*:execute"}))
+        require_action_scope("tools.okta.list_users")
+        require_action_scope("tools.okta.suspend_user")
+
+        # Should fail for other integrations
+        with pytest.raises(ScopeDeniedError):
+            require_action_scope("tools.slack.send_message")
+
+    def test_require_action_scope_denied(self):
+        """User without action scope gets denied."""
+        ctx_scopes.set(frozenset({"workflow:execute"}))
+
+        with pytest.raises(ScopeDeniedError) as exc_info:
+            require_action_scope("core.http_request")
+
+        assert exc_info.value.required_scopes == ["action:core.http_request:execute"]
+        assert exc_info.value.missing_scopes == ["action:core.http_request:execute"]
+
+    def test_require_action_scope_empty_scopes(self):
+        """User with no scopes gets denied."""
+        ctx_scopes.set(frozenset())
+
+        with pytest.raises(ScopeDeniedError):
+            require_action_scope("core.http_request")
+
+    def test_require_action_scope_multiple_scopes(self):
+        """User with multiple action scopes can execute matching actions."""
+        ctx_scopes.set(
+            frozenset(
+                {
+                    "action:core.*:execute",
+                    "action:tools.okta.*:execute",
+                    "workflow:execute",
+                }
+            )
+        )
+        require_action_scope("core.http_request")
+        require_action_scope("tools.okta.list_users")
+
+        # Should fail for non-matching actions
+        with pytest.raises(ScopeDeniedError):
+            require_action_scope("tools.slack.send_message")

tracecat/agent/router.py

@@ -11,6 +11,7 @@
 from tracecat.agent.service import AgentManagementService
 from tracecat.auth.credentials import RoleACL
 from tracecat.auth.types import AccessLevel, Role
+from tracecat.authz.controls import require_scope
 from tracecat.db.dependencies import AsyncDBSession
 from tracecat.exceptions import TracecatNotFoundError
 
@@ -46,6 +47,7 @@
 
 
 @router.get("/models")
+@require_scope("agent:read")
 async def list_models(
     *,
     role: OrganizationUserRole,
@@ -57,6 +59,7 @@ async def list_models(
 
 
 @router.get("/providers")
+@require_scope("agent:read")
 async def list_providers(
     *,
     role: OrganizationUserRole,
@@ -68,6 +71,7 @@ async def list_providers(
 
 
 @router.get("/providers/status")
+@require_scope("agent:read")
 async def get_providers_status(
     *,
     role: OrganizationUserRole,
@@ -79,6 +83,7 @@ async def get_providers_status(
 
 
 @router.get("/providers/configs")
+@require_scope("agent:read")
 async def list_provider_credential_configs(
     *,
     role: OrganizationAdminUserRole,
@@ -90,6 +95,7 @@ async def list_provider_credential_configs(
 
 
 @router.get("/providers/{provider}/config")
+@require_scope("agent:read")
 async def get_provider_credential_config(
     *,
     provider: str,
@@ -108,6 +114,7 @@ async def get_provider_credential_config(
 
 
 @router.post("/credentials", status_code=status.HTTP_201_CREATED)
+@require_scope("agent:execute")
 async def create_provider_credentials(
     *,
     params: ModelCredentialCreate,
@@ -127,6 +134,7 @@ async def create_provider_credentials(
 
 
 @router.put("/credentials/{provider}")
+@require_scope("agent:update")
 async def update_provider_credentials(
     *,
     provider: str,
@@ -152,6 +160,7 @@ async def update_provider_credentials(
 
 
 @router.delete("/credentials/{provider}")
+@require_scope("agent:delete")
 async def delete_provider_credentials(
     *,
     provider: str,
@@ -165,6 +174,7 @@ async def delete_provider_credentials(
 
 
 @router.get("/default-model")
+@require_scope("agent:read")
 async def get_default_model(
     *,
     role: OrganizationUserRole,
@@ -176,6 +186,7 @@ async def get_default_model(
 
 
 @router.put("/default-model")
+@require_scope("agent:update")
 async def set_default_model(
     *,
     model_name: str,
@@ -200,6 +211,7 @@ async def set_default_model(
 
 
 @router.get("/workspace/providers/status")
+@require_scope("agent:read")
 async def get_workspace_providers_status(
     *,
     role: WorkspaceUserRole,

tracecat/authz/controls.py

@@ -231,6 +231,50 @@ def wrapper(self, *args, **kwargs):
 # =============================================================================
 
 
+def require_action_scope(action_key: str) -> None:
+    """Check if the current user has permission to execute a specific action.
+
+    This function checks the context scopes against the required action scope.
+    The required scope is `action:{action_key}:execute`.
+
+    Scope matching supports wildcards:
+    - `action:*:execute` → any action (Admin)
+    - `action:core.*:execute` → core actions (Editor)
+    - `action:tools.okta.*:execute` → okta actions (custom role)
+    - `action:tools.okta.list_users:execute` → specific action
+
+    Args:
+        action_key: The action key (e.g., "core.http_request", "tools.okta.list_users")
+
+    Raises:
+        ScopeDeniedError: If the user doesn't have permission to execute the action
+    """
+    user_scopes = ctx_scopes.get()
+
+    # Platform superuser has "*" scope - bypass all checks
+    if "*" in user_scopes:
+        return
+
+    required_scope = f"action:{action_key}:execute"
+
+    if not has_scope(user_scopes, required_scope):
+        logger.warning(
+            "Action scope check failed",
+            action_key=action_key,
+            required_scope=required_scope,
+        )
+        raise ScopeDeniedError(
+            required_scopes=[required_scope],
+            missing_scopes=[required_scope],
+        )
+
+    logger.debug(
+        "Action scope check passed",
+        action_key=action_key,
+        required_scope=required_scope,
+    )
+
+
 def require_scope(*scopes: str, require_all: bool = True) -> Callable[[T], T]:
     """Decorator that requires specific scopes to access an endpoint or method.
 

tracecat/authz/scopes.py

@@ -30,6 +30,8 @@
         "schedule:read",
         "agent:read",
         "secret:read",
+        "tag:read",
+        "variable:read",
         "workspace:read",
         "workspace:member:read",
     }
@@ -53,6 +55,10 @@
         "agent:execute",
         "secret:create",
         "secret:update",
+        "tag:create",
+        "tag:update",
+        "variable:create",
+        "variable:update",
         # Core actions available to editors
         "action:core.*:execute",
     }
@@ -69,6 +75,8 @@
         "table:delete",
         "schedule:delete",
         "secret:delete",
+        "tag:delete",
+        "variable:delete",
         "agent:create",
         "agent:update",
         "agent:delete",
@@ -116,6 +124,9 @@
         # RBAC management
         "org:rbac:read",
         "org:rbac:manage",
+        # Org settings management
+        "org:settings:read",
+        "org:settings:manage",
         # Full workspace control across the org
         "workspace:read",
         "workspace:create",
@@ -139,6 +150,14 @@
         "table:create",
         "table:update",
         "table:delete",
+        "tag:read",
+        "tag:create",
+        "tag:update",
+        "tag:delete",
+        "variable:read",
+        "variable:create",
+        "variable:update",
+        "variable:delete",
         "schedule:read",
         "schedule:create",
         "schedule:update",
@@ -172,6 +191,9 @@
         # RBAC management
         "org:rbac:read",
         "org:rbac:manage",
+        # Org settings management
+        "org:settings:read",
+        "org:settings:manage",
         # Full workspace control across the org
         "workspace:read",
         "workspace:create",
@@ -195,6 +217,14 @@
         "table:create",
         "table:update",
         "table:delete",
+        "tag:read",
+        "tag:create",
+        "tag:update",
+        "tag:delete",
+        "variable:read",
+        "variable:create",
+        "variable:update",
+        "variable:delete",
         "schedule:read",
         "schedule:create",
         "schedule:update",

tracecat/authz/seeding.py

@@ -116,6 +116,19 @@
     ("secret:create", "secret", "create", "Create new secrets"),
     ("secret:update", "secret", "update", "Modify existing secrets"),
     ("secret:delete", "secret", "delete", "Delete secrets"),
+    # Tag scopes
+    ("tag:read", "tag", "read", "View tags"),
+    ("tag:create", "tag", "create", "Create new tags"),
+    ("tag:update", "tag", "update", "Modify existing tags"),
+    ("tag:delete", "tag", "delete", "Delete tags"),
+    # Variable scopes
+    ("variable:read", "variable", "read", "View variables"),
+    ("variable:create", "variable", "create", "Create new variables"),
+    ("variable:update", "variable", "update", "Modify existing variables"),
+    ("variable:delete", "variable", "delete", "Delete variables"),
+    # Organization settings scopes
+    ("org:settings:read", "org:settings", "read", "View organization settings"),
+    ("org:settings:manage", "org:settings", "manage", "Manage organization settings"),
     # Wildcard action scopes (for role assignments)
     ("action:*:execute", "action", "execute", "Execute any registry action"),
     (

tracecat/cases/attachments/router.py

@@ -1,7 +1,5 @@
 """Router for case attachments endpoints."""
 
-from __future__ import annotations
-
 import hashlib
 import uuid
 from typing import Annotated