#3808 role-aware Cedar policy plugin with tool-level gating and role …

mcpgateway/cache/auth_cache.py

@@ -623,6 +623,118 @@ async def set_user_role(self, email: str, team_id: str, role: Optional[str]) ->
                 expiry=time.time() + self._role_ttl,
             )
 
+    async def get_user_roles_list(self, email: str) -> Optional[List[str]]:
+        """Get the cached list of all role names for a user across all teams.
+
+        Returns:
+            - None: Cache miss (caller should query DB)
+            - list[str]: Cached role names (may be empty if user has no roles)
+
+        Args:
+            email: User email
+
+        Examples:
+            >>> import asyncio
+            >>> cache = AuthCache()
+            >>> result = asyncio.run(cache.get_user_roles_list("test@example.com"))
+            >>> result is None  # Cache miss
+            True
+        """
+        if not self._enabled:
+            return None
+
+        cache_key = f"roles_list:{email}"
+
+        # Check L1 in-memory cache
+        entry = self._role_cache.get(cache_key)
+        if entry and not entry.is_expired():
+            self._hit_count += 1
+            return entry.value if entry.value is not None else []
+
+        # Check L2 Redis cache
+        redis = await self._get_redis_client()
+        if redis:
+            try:
+                import json  # pylint: disable=import-outside-toplevel
+
+                redis_key = self._get_redis_key("roles_list", cache_key)
+                data = await redis.get(redis_key)
+                if data is not None:
+                    self._hit_count += 1
+                    self._redis_hit_count += 1
+                    roles = json.loads(data.decode() if isinstance(data, bytes) else data)
+                    with self._lock:
+                        self._role_cache[cache_key] = CacheEntry(value=roles, expiry=time.time() + self._role_ttl)
+                    return roles
+                self._redis_miss_count += 1
+            except Exception as e:
+                logger.warning(f"AuthCache Redis get_user_roles_list failed: {e}")
+
+        self._miss_count += 1
+        return None
+
+    async def set_user_roles_list(self, email: str, roles: List[str]) -> None:
+        """Store the list of all role names for a user.
+
+        Uses a long safety-net TTL; the primary invalidation mechanism is
+        explicit invalidation via invalidate_user_roles_list() on role changes.
+
+        Args:
+            email: User email
+            roles: List of role name strings
+
+        Examples:
+            >>> import asyncio
+            >>> cache = AuthCache()
+            >>> asyncio.run(cache.set_user_roles_list("test@example.com", ["developer"]))
+        """
+        if not self._enabled:
+            return
+
+        import json  # pylint: disable=import-outside-toplevel
+
+        cache_key = f"roles_list:{email}"
+        safety_net_ttl = 3600  # 1 hour fallback; write-invalidation is the primary mechanism
+
+        redis = await self._get_redis_client()
+        if redis:
+            try:
+                redis_key = self._get_redis_key("roles_list", cache_key)
+                await redis.setex(redis_key, safety_net_ttl, json.dumps(roles))
+            except Exception as e:
+                logger.warning(f"AuthCache Redis set_user_roles_list failed: {e}")
+
+        with self._lock:
+            self._role_cache[cache_key] = CacheEntry(value=roles, expiry=time.time() + safety_net_ttl)
+
+    async def invalidate_user_roles_list(self, email: str) -> None:
+        """Invalidate the cached roles list for a user.
+
+        Call this whenever a role is assigned to or revoked from the user.
+
+        Args:
+            email: User email
+
+        Examples:
+            >>> import asyncio
+            >>> cache = AuthCache()
+            >>> asyncio.run(cache.invalidate_user_roles_list("test@example.com"))
+        """
+        cache_key = f"roles_list:{email}"
+        logger.debug(f"AuthCache: Invalidating roles_list cache for {email}")
+
+        with self._lock:
+            self._role_cache.pop(cache_key, None)
+
+        redis = await self._get_redis_client()
+        if redis:
+            try:
+                redis_key = self._get_redis_key("roles_list", cache_key)
+                await redis.delete(redis_key)
+                await redis.publish("mcpgw:auth:invalidate", f"roles_list:{email}")
+            except Exception as e:
+                logger.warning(f"AuthCache Redis invalidate_user_roles_list failed: {e}")
+
     async def invalidate_user_role(self, email: str, team_id: str) -> None:
         """Invalidate cached role for a user in a team.
 

mcpgateway/main.py

@@ -95,7 +95,7 @@
 from mcpgateway.middleware.token_scoping import token_scoping_middleware
 from mcpgateway.middleware.validation_middleware import ValidationMiddleware
 from mcpgateway.observability import init_telemetry
-from mcpgateway.plugins.framework import HttpHookType, PluginError, PluginManager, PluginViolationError, PromptHookType, ResourceHookType
+from mcpgateway.plugins.framework import GlobalContext, HttpAuthCheckPermissionPayload, HttpHookType, PluginError, PluginManager, PluginViolationError, PromptHookType, ResourceHookType
 from mcpgateway.plugins.framework.constants import PLUGIN_VIOLATION_CODE_MAPPING, PluginViolationCode, VALID_HTTP_STATUS_CODES
 from mcpgateway.routers.server_well_known import router as server_well_known_router
 from mcpgateway.routers.well_known import router as well_known_router
@@ -1004,6 +1004,49 @@ async def _ensure_rpc_permission(user, db: Session, permission: str, method: str
     if permission == "admin.system_config" and _is_permission_admin_user(user):
         return
 
+    # Layer 2a: Plugin hook — HTTP_AUTH_CHECK_PERMISSION
+    # Mirrors the same hook invoked by @require_permission in rbac.py.
+    if plugin_manager and plugin_manager.has_hooks_for(HttpHookType.HTTP_AUTH_CHECK_PERMISSION):
+        user_email = get_user_email(user) if not isinstance(user, dict) else (user.get("email") or "")
+        is_admin = _is_permission_admin_user(user)
+        user_role_names: list[str] = []
+        try:
+            from mcpgateway.cache.auth_cache import auth_cache as _rpc_auth_cache  # pylint: disable=import-outside-toplevel
+
+            _cached_roles = await _rpc_auth_cache.get_user_roles_list(user_email)
+            if _cached_roles is not None:
+                user_role_names = _cached_roles
+            else:
+                _role_assignments = await PermissionService(db).get_user_roles(user_email)
+                user_role_names = [ur.role.name for ur in _role_assignments]
+                await _rpc_auth_cache.set_user_roles_list(user_email, user_role_names)
+        except Exception:  # nosec B110 - role fetch failure must never block the request
+            pass
+        _rpc_global_ctx = GlobalContext(
+            request_id=getattr(request, "state", None) and getattr(request.state, "request_id", None) or "",
+            user={"email": user_email, "is_admin": is_admin, "roles": user_role_names},
+        )
+        _rpc_result, _ = await plugin_manager.invoke_hook(
+            HttpHookType.HTTP_AUTH_CHECK_PERMISSION,
+            payload=HttpAuthCheckPermissionPayload(
+                user_email=user_email,
+                permission=permission,
+                resource_type="rpc",
+                is_admin=is_admin,
+                roles=user_role_names,
+                auth_method="rpc",
+            ),
+            global_context=_rpc_global_ctx,
+        )
+        if _rpc_result and _rpc_result.modified_payload and hasattr(_rpc_result.modified_payload, "granted"):
+            decision_reason = getattr(_rpc_result.modified_payload, "reason", None)
+            if _rpc_result.modified_payload.granted is False:
+                logger.warning(
+                    "RPC permission denied by plugin: method=%s permission=%s user=%s reason=%s",
+                    method, permission, user_email, decision_reason,
+                )
+                raise JSONRPCError(-32003, _ACCESS_DENIED_MSG, {"method": method})
+
     # Layer 2: RBAC check
     # Session tokens have no explicit team_id, so check across all team-scoped roles.
     # Mirrors the @require_permission decorator's check_any_team fallback (rbac.py:562-576).

mcpgateway/middleware/rbac.py

@@ -653,6 +653,29 @@ async def wrapper(*args, **kwargs):
                 plugin_context_table = user_context.get("plugin_context_table")
                 plugin_global_context = user_context.get("plugin_global_context")
 
+                # Fetch the user's role names so plugins (e.g. Cedar) can act on them.
+                # Try the roles list cache first; fall back to DB on miss.
+                # The cache is invalidated on assign/revoke, so it is always fresh on normal paths.
+                # Failure is non-fatal — plugins still receive email and is_admin.
+                user_role_names: List[str] = []
+                try:
+                    from mcpgateway.cache.auth_cache import auth_cache as _rbac_auth_cache  # pylint: disable=import-outside-toplevel
+
+                    _cached_roles = await _rbac_auth_cache.get_user_roles_list(user_context["email"])
+                    if _cached_roles is not None:
+                        user_role_names = _cached_roles
+                    else:
+                        _db_for_roles = kwargs.get("db") or user_context.get("db")
+                        if _db_for_roles:
+                            _role_assignments = await PermissionService(_db_for_roles).get_user_roles(user_context["email"], team_id=team_id)
+                        else:
+                            with fresh_db_session() as _db_roles:
+                                _role_assignments = await PermissionService(_db_roles).get_user_roles(user_context["email"], team_id=team_id)
+                        user_role_names = [ur.role.name for ur in _role_assignments]
+                        await _rbac_auth_cache.set_user_roles_list(user_context["email"], user_role_names)
+                except Exception:  # nosec B110 - role fetch failure must never block the request
+                    pass
+
                 # Reuse existing global context from middleware if available for consistency
                 # Otherwise create a new one (fallback for cases where middleware didn't run)
                 if plugin_global_context:
@@ -662,7 +685,13 @@ async def wrapper(*args, **kwargs):
                     global_context = GlobalContext(
                         request_id=request_id,
                         server_id=None,
-                        tenant_id=None,
+                        tenant_id=team_id,
+                        user={
+                            "email": user_context["email"],
+                            "is_admin": user_context.get("is_admin", False),
+                            "roles": user_role_names,
+                            "team_id": team_id,
+                        },
                     )
 
                 # Invoke permission check hook, passing plugin contexts from HTTP_PRE_REQUEST hook
@@ -674,6 +703,7 @@ async def wrapper(*args, **kwargs):
                         resource_type=resource_type,
                         team_id=team_id,
                         is_admin=user_context.get("is_admin", False),
+                        roles=user_role_names,
                         auth_method=user_context.get("auth_method"),
                         client_host=user_context.get("ip_address"),
                         user_agent=user_context.get("user_agent"),

mcpgateway/plugins/framework/hooks/http.py

@@ -155,6 +155,7 @@ class HttpAuthCheckPermissionPayload(PluginPayload):
         resource_type: Type of resource being accessed (e.g., "tool", "server", "prompt")
         team_id: Team context for the permission check (if applicable)
         is_admin: Whether the user has admin privileges
+        roles: List of role names assigned to the user (e.g., ["developer", "viewer"])
         auth_method: Authentication method used (e.g., "simple_token", "jwt", "oauth")
         client_host: Client IP address for IP-based permission checks
         user_agent: User agent string for device-based permission checks
@@ -165,6 +166,7 @@ class HttpAuthCheckPermissionPayload(PluginPayload):
     resource_type: str | None = None
     team_id: str | None = None
     is_admin: bool = False
+    roles: list[str] = []
     auth_method: str | None = None
     client_host: str | None = None
     user_agent: str | None = None

mcpgateway/services/role_service.py

@@ -628,6 +628,14 @@ async def assign_role_to_user(
         self.db.commit()
         self.db.refresh(user_role)
 
+        # Invalidate roles list cache so next request fetches fresh roles from DB
+        try:
+            from mcpgateway.cache.auth_cache import auth_cache  # pylint: disable=import-outside-toplevel
+
+            await auth_cache.invalidate_user_roles_list(user_email)
+        except Exception:  # nosec B110 - cache invalidation failure must never block the request
+            pass
+
         logger.info(f"Assigned role {role.name} to {user_email} (scope: {scope}, scope_id: {scope_id})")
         return user_role
 
@@ -671,6 +679,14 @@ async def revoke_role_from_user(self, user_email: str, role_id: str, scope: str,
         user_role.is_active = False
         self.db.commit()
 
+        # Invalidate roles list cache so next request fetches fresh roles from DB
+        try:
+            from mcpgateway.cache.auth_cache import auth_cache  # pylint: disable=import-outside-toplevel
+
+            await auth_cache.invalidate_user_roles_list(user_email)
+        except Exception:  # nosec B110 - cache invalidation failure must never block the request
+            pass
+
         logger.info(f"Revoked role {role_id} from {user_email} (scope: {scope}, scope_id: {scope_id})")
         return True
 

plugins/config.yaml

@@ -16,6 +16,55 @@ plugin_settings:
   include_user_info: false
 
 plugins:
+  # Cedar Policy Plugin - evaluates HTTP permission checks against Cedar policies
+  - name: "CedarPolicyPlugin"
+    kind: "cedarpolicyplugin.plugin.CedarPolicyPlugin"
+    description: "Evaluates user role-based permissions using Cedar policy language"
+    version: "0.1.0"
+    author: "ContextForge"
+    hooks: ["http_auth_check_permission", "tool_pre_invoke"]
+    tags: ["security", "cedar", "policy", "rbac", "auth"]
+    mode: "permissive"  # set to "enforce" to block on Deny
+    priority: 5
+    conditions: []
+    config:
+      policy_lang: "cedar"
+      # HTTP-level RBAC permits (used by http_auth_check_permission):
+      # Permit roles to perform gateway-level permission checks.
+      # Anything not permitted is implicitly denied.
+      #
+      # Tool-level permits (used by tool_pre_invoke):
+      # principal = Role::"<role>", action = Action::"<tool_name>", resource = Server::"<server_name>"
+      policy:
+        # HTTP-level: data_scientist can perform data:read permission checks
+        - principal: 'Role::"data_scientist"'
+          action: 'Action::"data:read"'
+          resource: 'Resource::"*"'
+        # HTTP-level: team_admin can perform multiple permission checks
+        - principal: 'Role::"team_admin"'
+          action:
+            - 'Action::"data:read"'
+            - 'Action::"security:read"'
+            - 'Action::"tools:read"'
+          resource: 'Resource::"*"'
+        # Tool-level: data_scientist can call ping and flaky_tool on any server
+        - principal: 'Role::"data_scientist"'
+          action:
+            - 'Action::"ping"'
+            - 'Action::"flaky_tool"'
+            - 'Action::"recovers_quickly"'
+          resource: 'Server::"*"'
+        # Tool-level: team_admin can call all dummy_mcp tools
+        - principal: 'Role::"team_admin"'
+          action:
+            - 'Action::"ping"'
+            - 'Action::"flaky_tool"'
+            - 'Action::"reset_counter"'
+            - 'Action::"status_code_tool"'
+            - 'Action::"recovers_quickly"'
+            - 'Action::"always_fails"'
+          resource: 'Server::"*"'
+
   # Argument Normalizer - stabilize inputs before anything else
   - name: "ArgumentNormalizer"
     kind: "plugins.argument_normalizer.argument_normalizer.ArgumentNormalizerPlugin"

plugins/external/cedar/cedarpolicyplugin/plugin-manifest.yaml

@@ -2,6 +2,7 @@ description: "A plugin that does policy decision and enforcement using cedar"
 author: "Shriti Priya"
 version: "0.1.0"
 available_hooks:
+  - "http_auth_check_permission"
   - "prompt_pre_hook"
   - "prompt_post_hook"
   - "tool_pre_hook"