fix(mcp): harden auth — PermissionError propagation, passthrough client_id guard, fail-closed on missing token

- _tool_allowed_for_current_user (server.py): catch PermissionError
  alongside ValueError so invalid API keys return False instead of
  propagating through the tool-search permission filter
- _setup_user_context (auth.py): catch PermissionError alongside
  ValueError so g.user is cleared and the error is logged consistently
  regardless of which failure type get_user_from_request() raises
- _resolve_user_from_api_key (auth.py): require client_id=="api_key"
  (set by CompositeTokenVerifier) in addition to API_KEY_PASSTHROUGH_CLAIM
  to prevent an external IdP JWT that happens to include the claim name
  from being misclassified as an API-key pass-through (DoS vector)
- _resolve_user_from_jwt_context (auth.py): same client_id guard so
  a rogue-claim JWT continues through JWT resolution instead of deferring
  to the API-key path (which would raise PermissionError for the user)
- _resolve_user_from_api_key (auth.py): raise PermissionError (not
  return None) when the pass-through claim is present but the raw token
  is absent — fail closed rather than falling through to weaker auth
- Tests: set client_id="api_key" on _passthrough_access_token helper;
  update test_jwt_context_with_api_key_passthrough_returns_none docstring;
  add test for namespaced claim on non-API-key client_id being ignored

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

superset/mcp_service/auth.py

@@ -223,10 +223,21 @@ def _resolve_user_from_jwt_context(app: Any) -> User | None:
     # 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):
-        logger.debug("API key pass-through token detected, deferring to API key auth")
-        return None
+        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(
+            "Ignoring %s claim on non-API-key token (client_id=%r); processing as JWT",
+            API_KEY_PASSTHROUGH_CLAIM,
+            getattr(access_token, "client_id", None),
+        )
 
     # Use configurable resolver or default
     from superset.mcp_service.mcp_config import default_user_resolver
@@ -295,10 +306,18 @@ def _resolve_user_from_api_key(app: Any) -> User | None:
     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:
-        return None
+        # 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"):
@@ -510,7 +529,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.

superset/mcp_service/server.py

@@ -429,13 +429,13 @@ def _tool_allowed_for_current_user(tool: Any) -> bool:
         if not getattr(g, "user", None):
             try:
                 g.user = get_user_from_request()
-            except ValueError:
+            except (ValueError, PermissionError):
                 return False
 
         method_permission_name = getattr(tool_func, METHOD_PERMISSION_ATTR, "read")
         permission_name = f"{PERMISSION_PREFIX}{method_permission_name}"
         return security_manager.can_access(permission_name, class_permission_name)
-    except (AttributeError, RuntimeError, ValueError):
+    except (AttributeError, RuntimeError, ValueError, PermissionError):
         logger.debug("Could not evaluate tool search permission", exc_info=True)
         return False
 

tests/unit_tests/mcp_service/test_auth_api_key.py

@@ -49,6 +49,7 @@ def _passthrough_access_token(token: str) -> MagicMock:
     """Build an AccessToken matching what CompositeTokenVerifier emits."""
     access_token = MagicMock()
     access_token.token = token
+    access_token.client_id = "api_key"
     access_token.claims = {API_KEY_PASSTHROUGH_CLAIM: True}
     return access_token
 
@@ -265,9 +266,10 @@ def test_jwt_access_token_skips_api_key_auth(app: SupersetApp) -> None:
 def test_jwt_context_with_api_key_passthrough_returns_none(app: SupersetApp) -> None:
     """When CompositeTokenVerifier passes through an API key token,
     _resolve_user_from_jwt_context should detect the namespaced
-    pass-through claim and return None so get_user_from_request falls
-    through to _resolve_user_from_api_key."""
+    pass-through claim AND client_id=="api_key" and return None so
+    get_user_from_request falls through to _resolve_user_from_api_key."""
     mock_access_token = MagicMock()
+    mock_access_token.client_id = "api_key"
     mock_access_token.claims = {API_KEY_PASSTHROUGH_CLAIM: True}
 
     with patch(
@@ -279,6 +281,30 @@ def test_jwt_context_with_api_key_passthrough_returns_none(app: SupersetApp) ->
     assert result is None
 
 
+def test_namespaced_claim_without_api_key_client_id_is_ignored(
+    app: SupersetApp,
+) -> None:
+    """An external IdP JWT that includes the namespaced API_KEY_PASSTHROUGH_CLAIM
+    but does NOT have client_id=='api_key' must NOT divert into the API-key path.
+    The client_id guard prevents misclassification / DoS for affected JWT users."""
+    mock_sm = MagicMock()
+
+    rogue_token = MagicMock()
+    rogue_token.token = "eyJhbGciOiJSUzI1NiJ9.idp_jwt_with_rogue_claim"  # noqa: S105
+    rogue_token.client_id = "some-idp-client"
+    rogue_token.claims = {API_KEY_PASSTHROUGH_CLAIM: True, "sub": "alice"}
+
+    with _mock_sm_ctx(app, mock_sm):
+        with _patch_access_token(rogue_token):
+            # JWT path tries to resolve user "alice" from DB and raises
+            # ValueError in this isolated unit-test setup.
+            # validate_api_key must NOT be called — the rogue claim was ignored.
+            with pytest.raises(ValueError, match="not found"):
+                get_user_from_request()
+
+    mock_sm.validate_api_key.assert_not_called()
+
+
 # -- Plain JWT with a colliding non-namespaced claim is NOT mistaken for API key --