fix(mcp): API key authentication for MCP — transport, validation, and RBAC

superset/mcp_service/auth.py

@@ -49,7 +49,11 @@
 from typing import Any, Callable, TYPE_CHECKING, TypeVar
 
 from flask import current_app, g, has_app_context, has_request_context
-from flask_appbuilder.security.sqla.models import Group, User
+from flask_appbuilder.security.sqla.models import User
+
+from superset import security_manager
+from superset.mcp_service.composite_token_verifier import API_KEY_PASSTHROUGH_CLAIM
+from superset.mcp_service.mcp_config import default_user_resolver
 
 if TYPE_CHECKING:
     from superset.connectors.sqla.models import SqlaTable
@@ -66,8 +70,13 @@
 METHOD_PERMISSION_ATTR = "_method_permission_name"
 
 
-class MCPPermissionDeniedError(Exception):
-    """Raised when user lacks required RBAC permission for an MCP tool."""
+class MCPPermissionDeniedError(PermissionError):
+    """Raised when user lacks required RBAC permission for an MCP tool.
+
+    Inherits from ``PermissionError`` so the middleware classifies denials as
+    user errors (HTTP 403 / WARNING log / "Access denied" sanitized message)
+    rather than unexpected server errors.
+    """
 
     def __init__(
         self,
@@ -107,13 +116,9 @@ def check_tool_permission(func: Callable[..., Any], *, log_denial: bool = True)
         True if user has permission or no permission is required.
     """
     try:
-        from flask import current_app
-
         if not current_app.config.get("MCP_RBAC_ENABLED", True):
             return True
 
-        from superset import security_manager
-
         if not hasattr(g, "user") or not g.user:
             if log_denial:
                 logger.warning(
@@ -140,16 +145,16 @@ def check_tool_permission(func: Callable[..., Any], *, log_denial: bool = True)
         if not has_permission:
             if log_denial:
                 logger.warning(
-                    "Permission denied for user %s: %s on %s (tool: %s)",
-                    g.user.username,
+                    "Permission denied for user id=%s: %s on %s (tool: %s)",
+                    getattr(g.user, "id", "?"),
                     permission_str,
                     class_permission_name,
                     func.__name__,
                 )
             else:
                 logger.debug(
-                    "Tool hidden for user %s: %s on %s (tool: %s)",
-                    g.user.username,
+                    "Tool hidden for user id=%s: %s on %s (tool: %s)",
+                    getattr(g.user, "id", "?"),
                     permission_str,
                     class_permission_name,
                     func.__name__,
@@ -206,54 +211,29 @@ def is_tool_visible_to_current_user(tool: Any) -> bool:
         return check_tool_permission(tool_func, log_denial=False)
 
     except (AttributeError, RuntimeError, ValueError):
-        logger.debug(
-            "Could not evaluate tool visibility for current user", exc_info=True
-        )
+        logger.debug("Could not evaluate tool visibility for current user")
         return False
 
 
 def load_user_with_relationships(
     username: str | None = None, email: str | None = None
 ) -> User | None:
-    """
-    Load a user with all relationships needed for permission checks.
+    """Load a user with roles and group roles eagerly loaded.
 
-    This function eagerly loads User.roles, User.groups, and Group.roles
-    to prevent detached instance errors when the session is closed/rolled back.
-
-    IMPORTANT: Always use this function instead of security_manager.find_user()
-    when loading users for MCP tool execution. The find_user() method doesn't
-    eagerly load Group.roles, causing "detached instance" errors when permission
-    checks access group.roles after the session is rolled back.
-
-    Args:
-        username: The username to look up (optional if email provided)
-        email: The email to look up (optional if username provided)
-
-    Returns:
-        User object with relationships loaded, or None if not found
+    Delegates to :meth:`SupersetSecurityManager.find_user_with_relationships`,
+    which mirrors FAB's ``find_user`` (including ``auth_username_ci`` and
+    ``MultipleResultsFound`` handling) while adding eager loading of
+    ``User.roles`` and ``User.groups.roles`` to prevent detached-instance
+    errors when the SQLAlchemy session is closed or rolled back after the
+    lookup — as happens in MCP tool-execution contexts.
 
     Raises:
         ValueError: If neither username nor email is provided
     """
     if not username and not email:
         raise ValueError("Either username or email must be provided")
 
-    from sqlalchemy.orm import joinedload
-
-    from superset.extensions import db
-
-    query = db.session.query(User).options(
-        joinedload(User.roles),
-        joinedload(User.groups).joinedload(Group.roles),
-    )
-
-    if username:
-        query = query.filter(User.username == username)
-    else:
-        query = query.filter(User.email == email)
-
-    return query.first()
+    return security_manager.find_user_with_relationships(username=username, email=email)
 
 
 def _resolve_user_from_jwt_context(app: Any) -> User | None:
@@ -285,9 +265,25 @@ def _resolve_user_from_jwt_context(app: Any) -> User | None:
     if access_token is None:
         return None
 
-    # Use configurable resolver or default
-    from superset.mcp_service.mcp_config import default_user_resolver
+    # API key pass-through: CompositeTokenVerifier accepted this token
+    # at the transport layer but defers actual validation to
+    # _resolve_user_from_api_key() (priority 2 in get_user_from_request).
+    # Require client_id=="api_key" (set by CompositeTokenVerifier) in addition
+    # to the claim so that an external IdP JWT that happens to include the
+    # claim name is not misclassified as an API-key pass-through.
+    claims = getattr(access_token, "claims", None)
+    if isinstance(claims, dict) and claims.get(API_KEY_PASSTHROUGH_CLAIM):
+        if getattr(access_token, "client_id", None) == "api_key":
+            logger.debug(
+                "API key pass-through token detected, deferring to API key auth"
+            )
+            return None
+        logger.debug(
+            "API key passthrough claim present but client_id is not 'api_key';"
+            " processing as JWT"
+        )
 
+    # Use configurable resolver or default
     resolver = app.config.get("MCP_USER_RESOLVER", default_user_resolver)
     username = resolver(app, access_token)
 
@@ -305,47 +301,70 @@ def _resolve_user_from_jwt_context(app: Any) -> User | None:
     if not user:
         # Fail closed: JWT says this user should exist but they don't.
         # Do NOT fall through to MCP_DEV_USERNAME or stale g.user.
+        # Avoid echoing the JWT-extracted username in the exception message
+        # (CodeQL py/clear-text-logging-sensitive-data).
+        logger.debug("JWT-authenticated user not found in database (identity from JWT)")
         raise ValueError(
-            f"JWT authenticated user '{username}' not found in Superset database. "
-            f"Ensure the user exists before granting MCP access."
+            "JWT authenticated user not found in Superset database. "
+            "Ensure the user exists before granting MCP access."
         )
 
     return user
 
 
 def _resolve_user_from_api_key(app: Any) -> User | None:
     """
-    Resolve the current user from an API key in the Authorization header.
+    Resolve the current user from an API key passed via Bearer token.
 
-    Uses FAB SecurityManager's API key validation. Only attempts when
-    FAB_API_KEY_ENABLED is True and a request context is active.
+    Reads the token from FastMCP's per-request ``AccessToken`` (set by
+    ``CompositeTokenVerifier`` when a Bearer token matches an API key
+    prefix). The streamable-http transport does not push a Flask request
+    context, so we cannot rely on ``flask.request`` headers — the verifier
+    already saw the token and stashed it on the ``AccessToken``.
 
     Returns:
-        User object with relationships loaded, or None if no API key present
-        or API key auth is not enabled/available.
+        User object with relationships loaded, or None if no API key
+        pass-through token is present or API key auth is not enabled.
 
     Raises:
-        PermissionError: If an API key is present but invalid/expired,
-            or if validation is not available in this FAB version.
+        PermissionError: If an API key pass-through token is present but
+            invalid/expired (fail closed — do NOT fall through to weaker
+            auth sources like ``MCP_DEV_USERNAME``), or if validation is
+            not available in this FAB version.
     """
-    if not app.config.get("FAB_API_KEY_ENABLED", False) or not has_request_context():
+    if not app.config.get("FAB_API_KEY_ENABLED", False):
         return None
 
-    sm = app.appbuilder.sm
-    # extract_api_key_from_request is FAB's method for reading
-    # the Bearer token from the Authorization header and matching prefixes.
-    # Not all FAB versions include this method, so guard with hasattr.
-    if not hasattr(sm, "extract_api_key_from_request"):
-        logger.debug(
-            "FAB SecurityManager does not have extract_api_key_from_request; "
-            "API key authentication is not available in this FAB version"
-        )
+    try:
+        from fastmcp.server.dependencies import get_access_token
+    except ImportError:
+        logger.debug("fastmcp.server.dependencies not available, skipping API key auth")
+        return None
+
+    access_token = get_access_token()
+    if access_token is None:
         return None
 
-    api_key_string = sm.extract_api_key_from_request()
-    if api_key_string is None:
+    # Only validate tokens that the CompositeTokenVerifier flagged as
+    # API key pass-throughs. Plain JWTs were already validated by the JWT
+    # verifier and resolved in _resolve_user_from_jwt_context.
+    claims = getattr(access_token, "claims", None)
+    if not (isinstance(claims, dict) and claims.get(API_KEY_PASSTHROUGH_CLAIM)):
+        return None
+    # Defense-in-depth: require client_id=="api_key" (set by CompositeTokenVerifier)
+    # to guard against rogue external IdP JWTs that include the passthrough claim.
+    if getattr(access_token, "client_id", None) != "api_key":
         return None
 
+    api_key_string = getattr(access_token, "token", None)
+    if not api_key_string:
+        # Passthrough claim is set but the raw token is absent — fail closed
+        # rather than silently falling through to weaker auth sources.
+        raise PermissionError(
+            "API key pass-through token is missing the raw token value."
+        )
+
+    sm = app.appbuilder.sm
     if not hasattr(sm, "validate_api_key"):
         logger.warning(
             "FAB SecurityManager does not have validate_api_key; "
@@ -367,10 +386,10 @@ def _resolve_user_from_api_key(app: Any) -> User | None:
     user_with_rels = load_user_with_relationships(username=user.username)
     if user_with_rels is None:
         logger.warning(
-            "Failed to reload API key user %s with relationships; "
+            "Failed to reload API key user id=%s with relationships; "
             "using original user object which may have lazy-loaded "
             "relationships",
-            user.username,
+            getattr(user, "id", "?"),
         )
         return user
     return user_with_rels
@@ -397,8 +416,6 @@ def get_user_from_request() -> User:
     Raises:
         ValueError: If user cannot be authenticated or found
     """
-    from flask import current_app
-
     # Priority 1: JWT context (per-request safe via ContextVar)
     if (jwt_user := _resolve_user_from_jwt_context(current_app)) is not None:
         return jwt_user
@@ -437,7 +454,12 @@ def get_user_from_request() -> User:
         "g.user was not set by external middleware",
     ]
     configured_prefixes = current_app.config.get("FAB_API_KEY_PREFIXES", ["sst_"])
-    prefix_example = configured_prefixes[0] if configured_prefixes else "sst_"
+    if isinstance(configured_prefixes, str):
+        prefix_example = configured_prefixes
+    elif configured_prefixes:
+        prefix_example = configured_prefixes[0]
+    else:
+        prefix_example = "sst_"
     raise ValueError(
         "No authenticated user found. Tried:\n"
         + "\n".join(f"  - {d}" for d in details)
@@ -464,8 +486,6 @@ def has_dataset_access(dataset: "SqlaTable") -> bool:
         Returns False on any error to fail securely.
     """
     try:
-        from superset import security_manager
-
         # Check if user has read access to the dataset
         if hasattr(g, "user") and g.user:
             # Use Superset's security manager to check dataset access
@@ -497,14 +517,14 @@ def check_chart_data_access(chart: Any) -> "DatasetValidationResult":
     return validate_chart_dataset(chart, check_access=True)
 
 
-def _log_user_resolution_failure(exc: ValueError) -> None:
-    """Log a user-resolution ValueError at the appropriate level.
+def _log_user_resolution_failure(exc: ValueError | PermissionError) -> None:
+    """Log a user-resolution failure at the appropriate level.
 
     "No authenticated user found" is expected in unauthenticated/dev
     deployments (no JWT, no API key, no MCP_DEV_USERNAME configured) and
     during tools/list scanning — log at DEBUG to avoid ERROR noise.
-    All other ValueErrors (e.g. dev username not in DB) are genuine
-    credential failures and are logged at ERROR.
+    All other failures (e.g. dev username not in DB, permission denied) are
+    genuine credential failures and are logged at ERROR.
     """
     if "No authenticated user found" in str(exc):
         logger.debug("MCP: no auth source configured, unauthenticated request")
@@ -527,7 +547,6 @@ def _setup_user_context() -> User | None:
     # tool calls when no per-request middleware refreshes it.
     # Only clear in app-context-only mode; preserve g.user when
     # a request context is active (external middleware set it).
-    from flask import has_request_context
 
     if not has_request_context():
         g.pop("user", None)
@@ -571,7 +590,7 @@ def _setup_user_context() -> User | None:
             logger.error("DB connection failed on retry during user setup: %s", e)
             _cleanup_session_on_error()
             raise
-        except ValueError as e:
+        except (ValueError, PermissionError) as e:
             # User resolution failed — fail closed. Do not fall back to
             # g.user from middleware, as that could allow a request to
             # proceed as a different user in multi-tenant deployments.