fix(security): support API key copy fallback

Adds superset:<resource>:<action> scope support to the MCP service:

- CompositeTokenVerifier now propagates an API key's own ApiKey.scopes
  (via a new SupersetSecurityManager.get_api_key_scopes lookup) instead
  of always stamping the verifier-global required_scopes on the token.
- check_tool_permission/_token_scope_allows accept a per-resource scope
  (e.g. superset:dashboard:read, derived from the tool's
  class_permission_name) as an alternative grant path alongside the
  existing flat superset:read/superset:write scopes, which keep working
  for already-issued tokens.
- SupersetSecurityManager.create_api_key validates requested scopes
  against the issuing user's own RBAC before delegating to FAB
  (intersection rule: a key can never be scoped beyond the user's own
  permissions; flat scopes are Admin-only to self-issue; unknown scopes
  are rejected).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docs/admin_docs/security/security.mdx

@@ -400,7 +400,7 @@ Once enabled, each user manages their own keys from their profile page:
 1. Open the user menu (top-right) and click **Info** to navigate to the User Info page
 2. Expand the **API Keys** section
 3. Click **+ API Key**
-4. Enter a name and (optionally) an expiration date
+4. Enter a name and optionally select resource scopes
 5. Copy the generated token — it is shown only once
 
 Only users with the `can_read` and `can_write` permissions on `ApiKey` (granted by default to Admins) can manage API keys.
@@ -415,6 +415,18 @@ Authorization: Bearer <your-api-key>
 
 This works for all REST API endpoints and the MCP server. The request is executed with the permissions of the user who created the key.
 
+#### API Key Scopes
+
+The creation dialog can restrict an API key to MCP resource actions such as
+`superset:dashboard:read` or `superset:chart:write`. A scope is an additional
+restriction: it never grants a permission that the creating user does not
+already have through Superset RBAC. Write scopes also cover update and delete
+operations for that resource; `superset:sqllab:write` covers SQL execution.
+
+Keys created without scopes retain legacy RBAC-only behavior. The scoped-key
+restrictions described here are enforced by the MCP server; regular REST API
+routes continue to apply their existing Superset RBAC checks.
+
 #### Use Cases
 
 - **CI/CD pipelines** — automated chart/dashboard exports and imports

superset-frontend/scripts/bundle-size-summary.js

@@ -0,0 +1,86 @@
+#!/usr/bin/env node
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied.  See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+// Reduces a webpack `--json` stats file down to the handful of headline
+// numbers worth tracking over time, in the flat array format
+// benchmark-action/github-action-benchmark expects for its
+// "customSmallerIsBetter" tool. The full stats file also includes a
+// `modules`/`chunks` graph across ~15k modules, which is enormous and not
+// useful for this purpose, so we only ever read `entrypoints`.
+//
+// Usage: node scripts/bundle-size-summary.js <path-to-stats.json>
+
+const fs = require('fs');
+
+// Entrypoints worth tracking: the two user-facing app shells. `menu`,
+// `preamble`, `theme`, and `service-worker` are small, low-variance
+// infrastructure chunks, not where bundle bloat actually shows up.
+const TRACKED_ENTRYPOINTS = ['spa', 'embedded'];
+
+function entrypointSizeByExt(entrypoint, ext) {
+  return (entrypoint.assets || [])
+    .filter(asset => asset.name.endsWith(ext))
+    .reduce((total, asset) => total + asset.size, 0);
+}
+
+function main() {
+  const statsPath = process.argv[2];
+  if (!statsPath) {
+    console.error('Usage: bundle-size-summary.js <path-to-stats.json>');
+    process.exit(1);
+  }
+
+  const stats = JSON.parse(fs.readFileSync(statsPath, 'utf8'));
+  const { entrypoints } = stats;
+  if (!entrypoints) {
+    console.error(
+      'stats.json has no `entrypoints` key -- was it built with ' +
+        '`BUNDLE_SIZE_STATS=true` set? Without it, webpack.config.js uses ' +
+        '`stats: "minimal"`, which omits `entrypoints`.',
+    );
+    process.exit(1);
+  }
+
+  const results = [];
+  TRACKED_ENTRYPOINTS.forEach(name => {
+    const entrypoint = entrypoints[name];
+    if (!entrypoint) {
+      console.error(`stats.json is missing the "${name}" entrypoint`);
+      process.exit(1);
+    }
+    results.push({
+      name: `${name} entrypoint (JS)`,
+      unit: 'bytes',
+      value: entrypointSizeByExt(entrypoint, '.js'),
+    });
+    results.push({
+      name: `${name} entrypoint (CSS)`,
+      unit: 'bytes',
+      value: entrypointSizeByExt(entrypoint, '.css'),
+    });
+  });
+
+  console.log(JSON.stringify(results, null, 2));
+}
+
+if (require.main === module) {
+  main();
+}
+
+module.exports = { entrypointSizeByExt, main, TRACKED_ENTRYPOINTS };

superset-frontend/src/features/apiKeys/ApiKeyCreateModal.tsx

@@ -27,8 +27,15 @@ import {
   Input,
   Button,
   Modal,
+  Select,
 } from '@superset-ui/core/components';
 import { useToasts } from 'src/components/MessageToasts/withToasts';
+import copyTextToClipboard from 'src/utils/copy';
+import {
+  API_KEY_SCOPE_OPTIONS,
+  getApiKeyScopesHelpText,
+  serializeApiKeyScopes,
+} from './apiKeyScopes';
 
 interface ApiKeyCreateModalProps {
   show: boolean;
@@ -38,6 +45,7 @@ interface ApiKeyCreateModalProps {
 
 interface FormValues {
   name: string;
+  scopes?: string[];
 }
 
 export function ApiKeyCreateModal({
@@ -62,9 +70,13 @@ export function ApiKeyCreateModal({
 
   const handleFormSubmit = async (values: FormValues) => {
     try {
+      const scopes = serializeApiKeyScopes(values.scopes);
       const response = await SupersetClient.post({
         endpoint: '/api/v1/security/api_keys/',
-        jsonPayload: values,
+        jsonPayload: {
+          name: values.name,
+          ...(scopes && { scopes }),
+        },
       });
       const key = response.json?.result?.key;
       if (!key) {
@@ -83,7 +95,7 @@ export function ApiKeyCreateModal({
       return;
     }
     try {
-      await navigator.clipboard.writeText(createdKey);
+      await copyTextToClipboard(() => Promise.resolve(createdKey));
       setCopied(true);
       if (copyTimerRef.current) {
         clearTimeout(copyTimerRef.current);
@@ -170,6 +182,24 @@ export function ApiKeyCreateModal({
           placeholder={t('e.g., CI/CD Pipeline, Analytics Script')}
         />
       </FormItem>
+      <FormItem
+        name="scopes"
+        label={t('MCP scopes')}
+        help={getApiKeyScopesHelpText()}
+      >
+        <Select
+          name="scopes"
+          mode="multiple"
+          allowClear
+          showSearch
+          options={API_KEY_SCOPE_OPTIONS}
+          placeholder={t('Select MCP resource scopes (optional)')}
+          data-test="api-key-scopes-select"
+          getPopupContainer={(trigger: HTMLElement) =>
+            trigger.closest<HTMLElement>('.ant-modal-container')
+          }
+        />
+      </FormItem>
     </FormModal>
   );
 }

superset-frontend/src/features/apiKeys/ApiKeyList.tsx

@@ -162,6 +162,19 @@ export function ApiKeyList() {
       key: 'status',
       render: (_: unknown, record: ApiKey) => getStatusBadge(record),
     },
+    {
+      title: t('MCP scopes'),
+      dataIndex: 'scopes',
+      key: 'scopes',
+      render: (scopes: string | null) =>
+        scopes ? (
+          <Tooltip title={scopes}>
+            <Tag>{t('%s MCP scopes', scopes.split(',').length)}</Tag>
+          </Tooltip>
+        ) : (
+          <Tag>{t('RBAC only')}</Tag>
+        ),
+    },
     {
       title: t('Actions'),
       key: 'actions',

superset-frontend/src/features/apiKeys/apiKeyScopes.test.ts

@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import {
+  API_KEY_SCOPE_OPTIONS,
+  getApiKeyScopesHelpText,
+  serializeApiKeyScopes,
+} from './apiKeyScopes';
+
+test('offers read and write scopes for every supported resource', () => {
+  expect(API_KEY_SCOPE_OPTIONS).toHaveLength(32);
+  expect(API_KEY_SCOPE_OPTIONS).toContainEqual({
+    label: 'superset:dashboard:read',
+    value: 'superset:dashboard:read',
+  });
+  expect(API_KEY_SCOPE_OPTIONS).toContainEqual({
+    label: 'superset:sqllab:write',
+    value: 'superset:sqllab:write',
+  });
+});
+
+test('serializes selected scopes for the FAB API', () => {
+  expect(
+    serializeApiKeyScopes(['superset:dashboard:read', 'superset:chart:write']),
+  ).toBe('superset:dashboard:read,superset:chart:write');
+  expect(serializeApiKeyScopes([])).toBeUndefined();
+  expect(serializeApiKeyScopes()).toBeUndefined();
+});
+
+test('explains that scopes apply to MCP rather than REST APIs', () => {
+  expect(getApiKeyScopesHelpText()).toContain('MCP resources');
+  expect(getApiKeyScopesHelpText()).toContain(
+    'do not restrict REST API requests',
+  );
+});

superset-frontend/src/features/apiKeys/apiKeyScopes.ts

@@ -0,0 +1,55 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { t } from '@apache-superset/core/translation';
+
+const API_KEY_SCOPE_RESOURCES = [
+  'annotation',
+  'chart',
+  'dashboard',
+  'database',
+  'dataset',
+  'explore',
+  'query',
+  'report',
+  'role',
+  'rls',
+  'savedquery',
+  'sqllab',
+  'tag',
+  'task',
+  'theme',
+  'user',
+] as const;
+
+const API_KEY_SCOPE_ACTIONS = ['read', 'write'] as const;
+
+export const API_KEY_SCOPE_OPTIONS = API_KEY_SCOPE_RESOURCES.flatMap(resource =>
+  API_KEY_SCOPE_ACTIONS.map(action => {
+    const value = `superset:${resource}:${action}`;
+    return { label: value, value };
+  }),
+);
+
+export const serializeApiKeyScopes = (scopes?: string[]) =>
+  scopes?.length ? scopes.join(',') : undefined;
+
+export const getApiKeyScopesHelpText = () =>
+  t(
+    'Limit which MCP resources and actions this key can access. These scopes do not restrict REST API requests and never grant permissions the user does not already have. Leave empty for legacy RBAC-only behavior.',
+  );

superset/mcp_service/auth.py

@@ -66,6 +66,11 @@ from superset.mcp_service.session_scope import _mcp_session_token
 from superset.mcp_service.utils.error_sanitization import (
     sanitize_for_log as _sanitize_for_log,
 )
+from superset.security.api_key_scopes import (
+    get_resource_scope,
+    METHOD_PERMISSION_SCOPE_ACTION,
+    RESOURCE_SCOPE_NAME as RESOURCE_SCOPE_NAME,
+)
 from superset.security.guest_token import GuestUser
 
 if TYPE_CHECKING:
@@ -115,19 +120,24 @@ class MCPNoAuthSourceError(ValueError):
 # is a privileged, write-class operation and therefore requires the write
 # scope. When introducing a new method permission, add it here.
 _METHOD_TO_REQUIRED_SCOPE = {
-    "read": "superset:read",
-    # "get" is the read-class permission FAB registers on its security API
-    # views (User/Role) — those views have no can_read, so tools targeting
-    # them declare method_permission_name="get".
-    "get": "superset:read",
-    "write": "superset:write",
-    "delete": "superset:write",
-    # SQL execution (execute_sql, get_chart_sql) runs arbitrary queries and is
-    # treated as a write-class privileged operation for scope purposes.
-    "execute_sql_query": "superset:write",
+    method: f"superset:{action}"
+    for method, action in METHOD_PERMISSION_SCOPE_ACTION.items()
 }
 
 
+def _required_resource_scope(
+    class_permission_name: str, method_permission_name: str
+) -> str | None:
+    """Compute the ``superset:<resource>:<action>`` scope string for a tool.
+
+    Returns None if either the resource or the action isn't mapped — callers
+    must treat that as "no per-resource scope available," not as a grant;
+    the flat ``_METHOD_TO_REQUIRED_SCOPE`` fallback still applies in that case
+    (see ``_token_scope_allows``).
+    """
+    return get_resource_scope(class_permission_name, method_permission_name)
+
+
 def _get_token_scopes() -> set[str] | None:
     """Return the set of scopes on the current JWT access token, or None.
 
@@ -143,8 +153,13 @@ def _get_token_scopes() -> set[str] | None:
 
     try:
         access_token = get_access_token()
-    except Exception:  # noqa: BLE001 - no JWT context for this request
-        return None
+    except Exception:  # noqa: BLE001 - fail closed on token-context errors
+        logger.exception("Unable to resolve MCP access-token scopes")
+        # ``None`` means that no scoped credential was presented and enables
+        # legacy RBAC-only behavior. An empty set instead makes every scope
+        # check fail, so an unexpected context error cannot erase restrictions
+        # carried by a credential.
+        return set()
 
     if access_token is None:
         return None
@@ -156,12 +171,21 @@ def _get_token_scopes() -> set[str] | None:
     return {str(s) for s in scopes}
 
 
-def _token_scope_allows(method_permission_name: str) -> bool:
+def _token_scope_allows(
+    method_permission_name: str, class_permission_name: str | None = None
+) -> bool:
     """Return whether the current token's scopes permit the given method.
 
     Back-compat: returns True (allow) when the token carries no scopes or there
     is no JWT context, so deployments not using scopes keep RBAC-only behavior.
     Only when the token advertises scopes is the mapped required scope enforced.
+
+    The per-resource scope (``superset:<resource>:<action>``, derived via
+    ``_required_resource_scope``) is an ALTERNATIVE grant path alongside the
+    flat method scope: a token carrying either the flat scope
+    (e.g. ``superset:read``) or the matching per-resource scope
+    (e.g. ``superset:dashboard:read``) is allowed, so already-issued
+    flat-scoped tokens keep working unchanged.
     """
     token_scopes = _get_token_scopes()
     if token_scopes is None:
@@ -179,7 +203,15 @@ def _token_scope_allows(method_permission_name: str) -> bool:
             method_permission_name,
         )
         return False
-    return required_scope in token_scopes
+    if required_scope in token_scopes:
+        return True
+    if class_permission_name is not None:
+        resource_scope = _required_resource_scope(
+            class_permission_name, method_permission_name
+        )
+        if resource_scope is not None and resource_scope in token_scopes:
+            return True
+    return False
 
 
 class MCPPermissionDeniedError(PermissionError):
@@ -223,12 +255,20 @@ def _log_scope_denial(
     cyclomatic complexity in check.
     """
     required_scope = _METHOD_TO_REQUIRED_SCOPE.get(method_permission_name)
+    resource_scope = _required_resource_scope(
+        class_permission_name, method_permission_name
+    )
+    scope_desc = (
+        resource_scope
+        or required_scope
+        or f"unmapped method permission '{method_permission_name}'"
+    )
     if log_denial:
         logger.warning(
             "Scope denied for user %s: token lacks required scope "
             "'%s' for %s on %s (tool: %s)",
             _sanitize_for_log(g.user.username),
-            required_scope,
+            scope_desc,
             permission_str,
             class_permission_name,
             func.__name__,
@@ -237,7 +277,7 @@ def _log_scope_denial(
         logger.debug(
             "Tool hidden for user %s: token lacks required scope '%s' (tool: %s)",
             _sanitize_for_log(g.user.username),
-            required_scope,
+            scope_desc,
             func.__name__,
         )
 
@@ -355,8 +395,13 @@ def check_tool_permission(  # noqa: C901
                 )
             return False
 
+        method_permission_name = getattr(func, METHOD_PERMISSION_ATTR, "read")
+        class_permission_name = getattr(func, CLASS_PERMISSION_ATTR, None)
+
+        # Token capabilities and user RBAC are independent restrictions.
+        # Disabling RBAC must not discard scopes explicitly carried by a key.
         if not current_app.config.get("MCP_RBAC_ENABLED", True):
-            return True
+            return _token_scope_allows(method_permission_name, class_permission_name)
 
         if not hasattr(g, "user") or not g.user:
             if log_denial:
@@ -369,7 +414,6 @@ def check_tool_permission(  # noqa: C901
                 )
             return False
 
-        class_permission_name = getattr(func, CLASS_PERMISSION_ATTR, None)
         if not class_permission_name:
             # No RBAC configured for this tool; allow by default. This is a
             # supported configuration (a protected tool may intentionally
@@ -383,9 +427,17 @@ def check_tool_permission(  # noqa: C901
                     "class_permission_name; allowing access without an RBAC check",
                     func.__name__,
                 )
+            if not _token_scope_allows(method_permission_name):
+                if log_denial:
+                    logger.warning(
+                        "Scope denied for permission-less tool %s: token lacks "
+                        "flat scope for method %s",
+                        func.__name__,
+                        method_permission_name,
+                    )
+                return False
             return True
 
-        method_permission_name = getattr(func, METHOD_PERMISSION_ATTR, "read")
         permission_str = f"{PERMISSION_PREFIX}{method_permission_name}"
 
         has_permission = security_manager.can_access(
@@ -400,7 +452,9 @@ def check_tool_permission(  # noqa: C901
         # advertises scopes. Tokens/deployments that don't use scopes (API keys,
         # scope-less JWTs, dev-mode) fall through to RBAC-only behavior — see
         # ``_token_scope_allows``.
-        if has_permission and not _token_scope_allows(method_permission_name):
+        if has_permission and not _token_scope_allows(
+            method_permission_name, class_permission_name
+        ):
             _log_scope_denial(
                 func,
                 method_permission_name,
@@ -463,7 +517,7 @@ def is_tool_visible_to_current_user(tool: Any) -> bool:
             return False
 
         if not current_app.config.get("MCP_RBAC_ENABLED", True):
-            return True
+            return check_tool_permission(tool_func, log_denial=False)
 
         from superset.mcp_service.privacy import (
             tool_requires_data_model_metadata_access,
@@ -476,10 +530,6 @@ def is_tool_visible_to_current_user(tool: Any) -> bool:
         ):
             return False
 
-        class_permission_name = getattr(tool_func, CLASS_PERMISSION_ATTR, None)
-        if not class_permission_name:
-            return True
-
         return check_tool_permission(tool_func, log_denial=False)
 
     except (AttributeError, RuntimeError, ValueError):

superset/mcp_service/composite_token_verifier.py

@@ -113,15 +113,19 @@ class CompositeTokenVerifier(TokenVerifier):
             )
         self._api_key_prefixes = tuple(valid)
 
-    def _validate_api_key_sync(self, token: str) -> str | None:
-        """Validate an API key against FAB and return the user's username.
+    def _validate_api_key_sync(self, token: str) -> tuple[str, list[str]] | None:
+        """Validate an API key against FAB and return (username, scopes).
 
         Runs synchronously inside a thread executor. Pushes a fresh Flask
         app context so that FAB's SecurityManager can access the database.
 
-        Returns the username on success, or ``None`` if the key is invalid,
-        FAB does not support ``validate_api_key``, or an unexpected error
-        occurs (fail closed).
+        ``scopes`` is the key's own ``ApiKey.scopes`` column, parsed from
+        FAB's comma-separated string storage format into a list (empty list
+        if the key has no scopes set, matching the "no scopes advertised"
+        convention used elsewhere in this module and in ``auth.py``).
+
+        Returns ``None`` if the key is invalid, FAB does not support
+        ``validate_api_key``, or an unexpected error occurs (fail closed).
         """
         if self._app is None:
             return None
@@ -135,12 +139,21 @@ class CompositeTokenVerifier(TokenVerifier):
                     )
                     return None
                 user = sm.validate_api_key(token)
-                username = user.username if user else None
-                # Unbind the local reference so this frame no longer points at
-                # the raw token (defense-in-depth). Python does not zero the
-                # underlying string memory on rebind.
-                token = ""  # noqa: S105
-                return username
+                if user is None:
+                    return None
+                username = user.username
+                scopes_str = (
+                    sm.get_api_key_scopes(token)
+                    if hasattr(sm, "get_api_key_scopes")
+                    else None
+                )
+                scopes = (
+                    [s.strip() for s in scopes_str.split(",") if s.strip()]
+                    if scopes_str
+                    else []
+                )
+                token = ""  # noqa: S105 -- unbind raw token, defense-in-depth
+                return username, scopes
         except Exception:  # noqa: BLE001 — catch-all: DB errors, FAB internals, etc.
             logger.warning(
                 "API key transport validation failed unexpectedly; rejecting token",
@@ -168,21 +181,25 @@ class CompositeTokenVerifier(TokenVerifier):
         if any(token.startswith(prefix) for prefix in self._api_key_prefixes):
             if self._app is not None:
                 loop = asyncio.get_running_loop()
-                username = await loop.run_in_executor(
+                result = await loop.run_in_executor(
                     None, self._validate_api_key_sync, token
                 )
-                if username is None:
+                if result is None:
                     logger.debug(
                         "API key rejected at transport layer (invalid or expired)"
                     )
                     return None
+                username, key_scopes = result
                 logger.debug(
                     "API key validated at transport layer for user=%s", username
                 )
                 return AccessToken(
                     token=token,
                     client_id="api_key",
-                    scopes=list(self.required_scopes or []),
+                    # Preserve the key's own scopes exactly. An empty list
+                    # means "no scopes advertised" and therefore retains the
+                    # RBAC-only behavior for existing unscoped API keys.
+                    scopes=key_scopes,
                     claims={
                         API_KEY_PASSTHROUGH_CLAIM: True,
                         API_KEY_VALIDATED_USERNAME_CLAIM: username,
@@ -190,10 +207,11 @@ class CompositeTokenVerifier(TokenVerifier):
                 )
 
             # No app configured: fall back to prefix-only pass-through so
-            # ``_resolve_user_from_api_key`` handles DB validation.
-            # NOTE: ``MCP_REQUIRED_SCOPES`` is intentionally not enforced for
-            # API-key auth — FAB API keys do not carry scopes. Authorization is
-            # enforced downstream via ``check_tool_permission`` (RBAC).
+            # ``_resolve_user_from_api_key`` handles DB validation. Without an
+            # app there is no DB access here, so the key's own ApiKey.scopes
+            # cannot be read — the verifier-global required_scopes are used
+            # instead. Authorization is still enforced downstream via
+            # ``check_tool_permission`` (RBAC).
             logger.debug("API key token detected (prefix match), passing through")
             return AccessToken(
                 token=token,

superset/mcp_service/mcp_config.py

@@ -653,10 +653,9 @@ def _build_composite_verifier(
     if api_key_enabled:
         if required_scopes := app.config.get("MCP_REQUIRED_SCOPES", []):
             logger.warning(
-                "MCP_REQUIRED_SCOPES is configured but API key tokens bypass "
-                "scope enforcement. API key holders gain access regardless of "
-                "MCP_REQUIRED_SCOPES=%r. Enforce per-key authorization via FAB "
-                "roles/RBAC instead.",
+                "MCP_REQUIRED_SCOPES=%r is configured, but API key tokens use "
+                "the scopes stored on each key instead. Unscoped API keys "
+                "retain legacy RBAC-only behavior.",
                 required_scopes,
             )
         raw_prefixes: str | Sequence[str] = app.config.get(

superset/mcp_service/system/tool/get_schema.py

@@ -30,7 +30,7 @@ from fastmcp import Context
 from superset_core.mcp.decorators import tool, ToolAnnotations
 
 from superset.extensions import event_logger
-from superset.mcp_service.auth import MCPPermissionDeniedError
+from superset.mcp_service.auth import _token_scope_allows, MCPPermissionDeniedError
 from superset.mcp_service.common.schema_discovery import (
     CHART_DEFAULT_COLUMNS,
     CHART_SEARCH_COLUMNS,
@@ -235,9 +235,10 @@ async def get_schema(
 
         from superset import security_manager
 
-        if current_app.config.get("MCP_RBAC_ENABLED", True) and not (
-            security_manager.can_access("can_read", class_permission)
-        ):
+        rbac_allows = not current_app.config.get(
+            "MCP_RBAC_ENABLED", True
+        ) or security_manager.can_access("can_read", class_permission)
+        if not (rbac_allows and _token_scope_allows("read", class_permission)):
             user_str = getattr(getattr(g, "user", None), "username", None)
             logger.warning(
                 "get_schema RBAC denied: user=%s type=%s view=%s",

superset/security/api_key_scopes.py

@@ -0,0 +1,77 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Canonical resource and action mappings for scoped API keys."""
+
+# Map FAB method permissions used by MCP tools to the coarser actions supported
+# by API-key scopes. Keep this explicit so an unknown permission fails closed.
+METHOD_PERMISSION_SCOPE_ACTION: dict[str, str] = {
+    "read": "read",
+    "get": "read",
+    "write": "write",
+    "update": "write",
+    "delete": "write",
+    "execute_sql_query": "write",
+}
+
+# Map MCP/FAB class permission names to stable public resource slugs. These
+# cannot be derived by lowercasing because several names contain spaces or use
+# public spellings that differ from their internal class names.
+RESOURCE_SCOPE_NAME: dict[str, str] = {
+    "Annotation": "annotation",
+    "Chart": "chart",
+    "Dashboard": "dashboard",
+    "Database": "database",
+    "Dataset": "dataset",
+    "Explore": "explore",
+    "Query": "query",
+    "ReportSchedule": "report",
+    "Role": "role",
+    "Row Level Security": "rls",
+    "SavedQuery": "savedquery",
+    "SQLLab": "sqllab",
+    "Tag": "tag",
+    "Task": "task",
+    "Theme": "theme",
+    "User": "user",
+}
+
+RESOURCE_SCOPE_CLASS: dict[str, str] = {
+    resource: class_name for class_name, resource in RESOURCE_SCOPE_NAME.items()
+}
+RESOURCE_SCOPE_ACTIONS: frozenset[str] = frozenset(
+    METHOD_PERMISSION_SCOPE_ACTION.values()
+)
+SCOPE_ACTION_METHOD_PERMISSIONS: dict[str, tuple[str, ...]] = {
+    action: tuple(
+        method
+        for method, mapped_action in METHOD_PERMISSION_SCOPE_ACTION.items()
+        if mapped_action == action
+    )
+    for action in RESOURCE_SCOPE_ACTIONS
+}
+
+
+def get_resource_scope(
+    class_permission_name: str, method_permission_name: str
+) -> str | None:
+    """Return the resource scope required by a FAB class/method permission."""
+    resource = RESOURCE_SCOPE_NAME.get(class_permission_name)
+    action = METHOD_PERMISSION_SCOPE_ACTION.get(method_permission_name)
+    if resource is None or action is None:
+        return None
+    return f"superset:{resource}:{action}"

superset/security/manager.py

@@ -17,6 +17,7 @@
 # pylint: disable=too-many-lines
 """A set of constants and methods to manage permissions and security"""
 
+import datetime
 import logging
 import re
 import time
@@ -4926,6 +4927,123 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
             raw_token, secret, algorithms=[algo], audience=audience
         )
 
+    def get_api_key_scopes(self, api_key_string: str) -> Optional[str]:
+        """Return the ``scopes`` value for a validated API key.
+
+        FAB's ``validate_api_key`` resolves the matching ``ApiKey`` row
+        internally (by lookup hash) but only returns the associated
+        ``User`` — the row's ``scopes`` column is otherwise unreachable by
+        callers. This repeats the same cheap, indexed lookup so MCP's
+        ``CompositeTokenVerifier`` can propagate per-key scopes instead of
+        silently falling back to verifier-global scopes. Call only after
+        ``validate_api_key`` has already succeeded for this token — this
+        method does not itself verify the key hash or active status.
+        """
+        lookup = self._compute_lookup_hash(api_key_string)  # type: ignore[attr-defined]
+        api_key = (
+            self.session.query(self.api_key_model)  # type: ignore[attr-defined]
+            .filter(self.api_key_model.lookup_hash == lookup)
+            .one_or_none()
+        )
+        return api_key.scopes if api_key else None
+
+    def _validate_requested_api_key_scopes(
+        self, user: Any, scopes: Optional[str]
+    ) -> None:
+        """Raise if ``scopes`` would grant a user more than their own RBAC.
+
+        Enforces the "intersection, never broader" rule confirmed for this
+        feature: a user must never be able to mint a token scoped beyond
+        what their own role already permits, even if they hand-author the
+        scopes string themselves at issuance time.
+
+        Per-resource scopes (``superset:<resource>:<action>``) are checked
+        against the user's actual ``can_<method>`` RBAC grant for that
+        resource. Flat scopes (``superset:read``/``superset:write``, the
+        pre-per-resource form) can only be self-issued by Admins — a flat
+        scope grants a method across every resource, and there's no single
+        RBAC check that soundly proves a non-Admin has that for "every
+        resource," so it's rejected for anyone else rather than guessed at.
+        Unrecognized scope strings are rejected outright (fail closed).
+
+        NOTE: this only prevents the request from being honored; it does
+        not (yet) produce a clean 400 response, since FAB's ``ApiKeyApi``
+        has no validation hook this can plug into without replacing the API
+        registration entirely. Raising here surfaces as a 500 via FAB's
+        ``@safe`` decorator until that's addressed — tracked as a known
+        follow-up, not silently accepted.
+        """
+        if not scopes:
+            return
+        # pylint: disable-next=import-outside-toplevel
+        from superset.security.api_key_scopes import (
+            RESOURCE_SCOPE_ACTIONS,
+            RESOURCE_SCOPE_CLASS,
+            SCOPE_ACTION_METHOD_PERMISSIONS,
+        )
+
+        admin_role_name = get_conf()["AUTH_ROLE_ADMIN"]
+        is_admin = any(
+            role.name == admin_role_name for role in getattr(user, "roles", [])
+        )
+        for raw_scope in scopes.split(","):
+            scope = raw_scope.strip()
+            if not scope:
+                continue
+            parts = scope.split(":")
+            if len(parts) == 3 and parts[0] == "superset":
+                _, resource_slug, action = parts
+                class_permission_name = RESOURCE_SCOPE_CLASS.get(resource_slug)
+                if class_permission_name is None:
+                    raise ValueError(
+                        f"Requested scope '{scope}' names an unrecognized "
+                        f"resource '{resource_slug}'"
+                    )
+                if action not in RESOURCE_SCOPE_ACTIONS:
+                    raise ValueError(
+                        f"Requested scope '{scope}' names an unrecognized "
+                        f"action '{action}'"
+                    )
+                if any(
+                    self._has_view_access(user, f"can_{method}", class_permission_name)
+                    for method in SCOPE_ACTION_METHOD_PERMISSIONS[action]
+                ):
+                    continue
+                raise ValueError(
+                    f"Requested scope '{scope}' exceeds the issuing user's "
+                    "own permissions"
+                )
+            if (
+                len(parts) == 2
+                and parts[0] == "superset"
+                and parts[1] in RESOURCE_SCOPE_ACTIONS
+                and is_admin
+            ):
+                continue
+            raise ValueError(
+                f"Requested scope '{scope}' is not a recognized "
+                "superset:<resource>:<action> scope, or requires Admin to "
+                "self-issue as a flat scope"
+            )
+
+    def create_api_key(
+        self,
+        user: Any,
+        name: str,
+        scopes: Optional[str] = None,
+        expires_on: Optional[datetime.datetime] = None,
+    ) -> Optional[dict[str, Any]]:
+        """Create a new API key, enforcing the scope-intersection rule.
+
+        Thin wrapper around FAB's ``SecurityManager.create_api_key`` — see
+        ``_validate_requested_api_key_scopes`` for the actual check. FAB's
+        base implementation is otherwise unchanged.
+        """
+        self._validate_requested_api_key_scopes(user, scopes)
+        return super().create_api_key(  # type: ignore[misc]
+            user=user, name=name, scopes=scopes, expires_on=expires_on
+        )
+
     @staticmethod
     def is_guest_user(user: Optional[Any] = None) -> bool:
         # pylint: disable=import-outside-toplevel

tests/integration_tests/queries/saved_queries/api_tests.py

@@ -18,7 +18,6 @@
 # isort:skip_file
 """Unit tests for Superset"""
 
-from datetime import datetime
 from io import BytesIO
 from typing import Optional
 from unittest.mock import Mock, patch
@@ -606,7 +605,10 @@ class TestSavedQueryApi(SupersetTestCase):
             db.session.query(SavedQuery).filter(SavedQuery.label == "label1").all()[0]
         )
         self.login(ADMIN_USERNAME)
-        with freeze_time(datetime.now()):
+        # Freeze relative to the persisted timestamp so database-specific
+        # timestamp precision cannot make the humanized value age into the
+        # next bucket while the request is being handled.
+        with freeze_time(saved_query.changed_on):
             uri = f"api/v1/saved_query/{saved_query.id}"
             rv = self.get_assert_metric(uri, "get")
             assert rv.status_code == 200

tests/unit_tests/mcp_service/system/tool/test_get_schema.py

@@ -66,7 +66,7 @@ def mock_auth():
 
 
 @pytest.fixture(autouse=True)
-def allow_data_model_metadata():
+def allow_data_model_metadata():  # noqa: PT004
     """Keep the standalone get_schema suite in the unrestricted default path."""
     with patch.object(
         get_schema_module,
@@ -606,3 +606,40 @@ class TestGetSchemaPermissionMap:
         factories = set(get_schema_module._SCHEMA_CORE_FACTORIES.keys())
         perms = set(get_schema_module._MODEL_TYPE_CLASS_PERMISSION.keys())
         assert factories == perms
+
+    @pytest.mark.asyncio
+    async def test_resource_scope_is_enforced(self, app, mcp_server):
+        """RBAC access alone cannot bypass a scoped token's resource limit."""
+        with (
+            patch.dict(app.config, {"MCP_RBAC_ENABLED": True}),
+            patch("superset.security_manager.can_access", return_value=True),
+            patch.object(
+                get_schema_module, "_token_scope_allows", return_value=False
+            ) as scope_allows,
+        ):
+            async with Client(mcp_server) as client:
+                with pytest.raises(ToolError, match="Permission denied"):
+                    await client.call_tool(
+                        "get_schema", {"request": {"model_type": "chart"}}
+                    )
+
+        scope_allows.assert_called_once_with("read", "Chart")
+
+    @pytest.mark.asyncio
+    async def test_resource_scope_is_enforced_when_rbac_disabled(self, app, mcp_server):
+        """The RBAC feature flag does not disable credential scopes."""
+        with (
+            patch.dict(app.config, {"MCP_RBAC_ENABLED": False}),
+            patch("superset.security_manager.can_access") as can_access,
+            patch.object(
+                get_schema_module, "_token_scope_allows", return_value=False
+            ) as scope_allows,
+        ):
+            async with Client(mcp_server) as client:
+                with pytest.raises(ToolError, match="Permission denied"):
+                    await client.call_tool(
+                        "get_schema", {"request": {"model_type": "chart"}}
+                    )
+
+        can_access.assert_not_called()
+        scope_allows.assert_called_once_with("read", "Chart")

tests/unit_tests/mcp_service/test_auth_rbac.py

@@ -23,12 +23,14 @@ import pytest
 from flask import g
 
 from superset.mcp_service.auth import (
+    _required_resource_scope,
     check_tool_permission,
     CLASS_PERMISSION_ATTR,
     is_tool_visible_to_current_user,
     MCPPermissionDeniedError,
     METHOD_PERMISSION_ATTR,
     PERMISSION_PREFIX,
+    RESOURCE_SCOPE_NAME,
 )
 
 
@@ -108,6 +110,17 @@ def test_check_tool_permission_no_class_permission_allows(app_context) -> None:
     assert check_tool_permission(func) is True
 
 
+def test_scoped_token_constrains_permissionless_tool(app_context) -> None:
+    """Resource-only scopes do not grant permission-less tools."""
+    g.user = MagicMock(username="admin")
+    func = _make_tool_func()
+
+    with _patch_token_scopes(["superset:dashboard:read"]):
+        assert check_tool_permission(func) is False
+    with _patch_token_scopes(["superset:read"]):
+        assert check_tool_permission(func) is True
+
+
 def test_check_tool_permission_no_user_denies(app_context) -> None:
     """If no g.user, permission check should deny."""
     g.user = None
@@ -170,6 +183,19 @@ def test_check_tool_permission_disabled_via_config(app_context, app) -> None:
         app.config["MCP_RBAC_ENABLED"] = True
 
 
+def test_disabled_rbac_still_enforces_token_scopes(app_context, app) -> None:
+    """Disabling user RBAC does not disable credential restrictions."""
+    func = _make_tool_func(class_perm="Chart", method_perm="write")
+    app.config["MCP_RBAC_ENABLED"] = False
+    try:
+        with _patch_token_scopes(["superset:dashboard:read"]):
+            assert check_tool_permission(func) is False
+        with _patch_token_scopes(["superset:chart:write"]):
+            assert check_tool_permission(func) is True
+    finally:
+        app.config["MCP_RBAC_ENABLED"] = True
+
+
 # -- Permission constants --
 
 
@@ -289,6 +315,19 @@ def test_visibility_public_tool_no_class_permission(app_context) -> None:
     assert is_tool_visible_to_current_user(tool) is True
 
 
+def test_visibility_hides_permissionless_tool_from_resource_scoped_token(
+    app_context,
+) -> None:
+    """Permission-less tools require a flat scope in tools/list too."""
+    g.user = MagicMock(username="viewer")
+    tool = _make_mock_tool(fn=_make_tool_func())
+
+    with _patch_token_scopes(["superset:dashboard:read"]):
+        assert is_tool_visible_to_current_user(tool) is False
+    with _patch_token_scopes(["superset:read"]):
+        assert is_tool_visible_to_current_user(tool) is True
+
+
 def test_visibility_allowed_tool(app_context) -> None:
     """Tools where security_manager grants access are visible."""
     g.user = MagicMock(username="admin")
@@ -431,6 +470,23 @@ def test_scope_falls_back_to_rbac_when_no_jwt_context(app_context) -> None:
     assert result is True
 
 
+def test_scope_context_error_fails_closed(app_context) -> None:
+    """An unexpected token lookup failure cannot erase token restrictions."""
+    g.user = MagicMock(username="editor")
+    func = _make_tool_func(class_perm="Chart", method_perm="read")
+
+    mock_sm = MagicMock()
+    mock_sm.can_access = MagicMock(return_value=True)
+    with (
+        patch("superset.mcp_service.auth.security_manager", mock_sm),
+        patch(
+            "fastmcp.server.dependencies.get_access_token",
+            side_effect=TypeError("invalid token context"),
+        ),
+    ):
+        assert check_tool_permission(func) is False
+
+
 def test_scope_read_denied_when_token_lacks_read_scope(app_context) -> None:
     """A read tool is denied when the token only carries an unrelated scope."""
     g.user = MagicMock(username="viewer")
@@ -447,7 +503,9 @@ def test_scope_read_denied_when_token_lacks_read_scope(app_context) -> None:
     assert result is False
 
 
-def test_scope_denies_unmapped_method_for_scoped_token(app_context) -> None:
+def test_scope_denies_unmapped_method_for_scoped_token(
+    app_context, caplog: pytest.LogCaptureFixture
+) -> None:
     """A scoped token presented for a method permission that is NOT in the
     scope map fails closed (denied), even when RBAC grants, so an unmapped
     custom permission cannot silently bypass scope enforcement."""
@@ -463,6 +521,8 @@ def test_scope_denies_unmapped_method_for_scoped_token(app_context) -> None:
         result = check_tool_permission(func)
 
     assert result is False
+    assert "unmapped method permission 'some_custom_perm'" in caplog.text
+    assert "required scope 'None'" not in caplog.text
 
 
 def test_scope_execute_sql_query_requires_write_scope(app_context) -> None:
@@ -480,6 +540,103 @@ def test_scope_execute_sql_query_requires_write_scope(app_context) -> None:
             assert check_tool_permission(func) is True
 
 
+# -- Per-resource scopes (superset:<resource>:<action>) --
+
+
+def test_required_resource_scope_special_names() -> None:
+    """The explicit resource map handles names a naive lower() would break:
+    'Row Level Security' (spaces) and 'ReportSchedule'/'SQLLab' (misnames)."""
+    assert _required_resource_scope("Row Level Security", "read") == "superset:rls:read"
+    assert _required_resource_scope("ReportSchedule", "write") == (
+        "superset:report:write"
+    )
+    assert _required_resource_scope("SQLLab", "execute_sql_query") == (
+        "superset:sqllab:write"
+    )
+    assert _required_resource_scope("Chart", "update") == "superset:chart:write"
+
+
+def test_required_resource_scope_unmapped_returns_none() -> None:
+    """An unmapped resource or method yields None (no per-resource scope),
+    which callers must NOT treat as a grant."""
+    assert _required_resource_scope("NotAResource", "read") is None
+    assert _required_resource_scope("Chart", "not_a_method") is None
+
+
+def test_resource_scope_name_covers_all_tool_resource_classes() -> None:
+    """RESOURCE_SCOPE_NAME must cover every class_permission_name declared by
+    MCP tools. If a new resource class is added, add it to the map."""
+    assert set(RESOURCE_SCOPE_NAME.keys()) == {
+        "Annotation",
+        "Chart",
+        "Dashboard",
+        "Database",
+        "Dataset",
+        "Explore",
+        "Query",
+        "ReportSchedule",
+        "Role",
+        "Row Level Security",
+        "SavedQuery",
+        "SQLLab",
+        "Tag",
+        "Task",
+        "Theme",
+        "User",
+    }
+
+
+def test_per_resource_scope_grants_matching_tool(app_context) -> None:
+    """A token scoped ONLY to superset:chart:write (no flat superset:write)
+    still grants a Chart/write tool via the per-resource grant path."""
+    g.user = MagicMock(username="editor")
+    func = _make_tool_func(class_perm="Chart", method_perm="write")
+
+    mock_sm = MagicMock()
+    mock_sm.can_access = MagicMock(return_value=True)
+    with (
+        patch("superset.mcp_service.auth.security_manager", mock_sm),
+        _patch_token_scopes(["superset:chart:write"]),
+    ):
+        result = check_tool_permission(func)
+
+    assert result is True
+
+
+def test_per_resource_scope_does_not_leak_across_resources(app_context) -> None:
+    """A token scoped to superset:chart:write does NOT grant a Dashboard/write
+    tool (resource isolation)."""
+    g.user = MagicMock(username="editor")
+    func = _make_tool_func(class_perm="Dashboard", method_perm="write")
+
+    mock_sm = MagicMock()
+    mock_sm.can_access = MagicMock(return_value=True)
+    with (
+        patch("superset.mcp_service.auth.security_manager", mock_sm),
+        _patch_token_scopes(["superset:chart:write"]),
+    ):
+        result = check_tool_permission(func)
+
+    assert result is False
+
+
+def test_per_resource_scope_enforces_action(app_context) -> None:
+    """A token scoped to superset:chart:read does NOT grant a Chart/write tool
+    (action still enforced within the resource)."""
+    g.user = MagicMock(username="editor")
+    func = _make_tool_func(class_perm="Chart", method_perm="write")
+
+    mock_sm = MagicMock()
+    mock_sm.can_access = MagicMock(return_value=True)
+    with (
+        patch("superset.mcp_service.auth.security_manager", mock_sm),
+        _patch_token_scopes(["superset:chart:read"]),
+    ):
+        result = check_tool_permission(func)
+
+    assert result is False
+
+
 # ---------------------------------------------------------------------------
 # User/Role tools must request a permission FAB actually registers.
 #

tests/unit_tests/mcp_service/test_composite_token_verifier.py

@@ -233,13 +233,22 @@ async def test_api_key_passthrough_propagates_required_scopes() -> None:
 # -- Transport-layer DB validation (app configured) --
 
 
-def _make_app_with_api_key(username: str | None) -> MagicMock:
-    """Return a mock Flask app whose SecurityManager validates to ``username``."""
+def _make_app_with_api_key(
+    username: str | None, scopes: str | None = None
+) -> MagicMock:
+    """Return a mock Flask app whose SecurityManager validates to ``username``.
+
+    ``scopes`` is what ``get_api_key_scopes`` returns (FAB stores scopes as a
+    comma-separated string, or None). It must be configured explicitly — an
+    unconfigured MagicMock return value would raise on ``.split(",")`` inside
+    the verifier's broad except-block and silently read as a rejected key.
+    """
     mock_user = MagicMock()
     mock_user.username = username
 
     mock_sm = MagicMock()
     mock_sm.validate_api_key = MagicMock(return_value=mock_user if username else None)
+    mock_sm.get_api_key_scopes = MagicMock(return_value=scopes)
 
     mock_app = MagicMock()
     mock_app.app_context.return_value.__enter__ = MagicMock(return_value=None)
@@ -264,6 +273,41 @@ async def test_transport_validation_valid_key_returns_access_token() -> None:
     assert result.claims.get(API_KEY_VALIDATED_USERNAME_CLAIM) == "alice"
 
 
+@pytest.mark.asyncio
+async def test_transport_validation_uses_keys_own_scopes() -> None:
+    """A key with its own ApiKey.scopes carries them on the AccessToken,
+    parsed from FAB's comma-separated storage format."""
+    mock_app = _make_app_with_api_key(
+        "alice", scopes="superset:dashboard:read, superset:chart:read"
+    )
+    verifier = CompositeTokenVerifier(
+        jwt_verifier=None, api_key_prefixes=["sst_"], app=mock_app
+    )
+
+    result = await verifier.verify_token("sst_valid_key")
+
+    assert result is not None
+    assert result.scopes == ["superset:dashboard:read", "superset:chart:read"]
+
+
+@pytest.mark.asyncio
+async def test_transport_validation_no_key_scopes_remains_unscoped() -> None:
+    """A key without scopes remains unscoped despite global JWT requirements."""
+    mock_app = _make_app_with_api_key("alice", scopes=None)
+    jwt_verifier = MagicMock()
+    jwt_verifier.required_scopes = ["superset:read"]
+    jwt_verifier.verify_token = AsyncMock()
+
+    verifier = CompositeTokenVerifier(
+        jwt_verifier=jwt_verifier, api_key_prefixes=["sst_"], app=mock_app
+    )
+
+    result = await verifier.verify_token("sst_valid_key")
+
+    assert result is not None
+    assert result.scopes == []
+
+
 @pytest.mark.asyncio
 async def test_transport_validation_invalid_key_returns_none() -> None:
     """An invalid API key is rejected at transport (returns None → HTTP 401)."""

tests/unit_tests/security/test_api_key_scopes.py

@@ -0,0 +1,251 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Tests for API key scope validation in SupersetSecurityManager.
+
+Covers the "intersection, never broader" rule: a user must not be able to
+mint an API key scoped beyond what their own RBAC already permits.
+"""
+
+import re
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from superset.extensions import appbuilder
+from superset.security.api_key_scopes import (
+    RESOURCE_SCOPE_ACTIONS,
+    RESOURCE_SCOPE_CLASS,
+)
+from superset.security.manager import SupersetSecurityManager
+
+
+def _make_user(*role_names: str) -> MagicMock:
+    """Build a mock user whose roles carry the given names."""
+    user = MagicMock()
+    roles = []
+    for role_name in role_names:
+        role = MagicMock()
+        role.name = role_name
+        roles.append(role)
+    user.roles = roles
+    return user
+
+
+@pytest.fixture
+def sm(app_context: None) -> SupersetSecurityManager:
+    return SupersetSecurityManager(appbuilder)
+
+
+def test_frontend_scope_catalog_matches_backend_contract() -> None:
+    """Keep the UI picker aligned with the canonical enforcement vocabulary."""
+    frontend_catalog = (
+        Path(__file__).parents[3]
+        / "superset-frontend/src/features/apiKeys/apiKeyScopes.ts"
+    ).read_text()
+    resources_source = re.search(
+        r"const API_KEY_SCOPE_RESOURCES = \[(.*?)\] as const;",
+        frontend_catalog,
+        re.DOTALL,
+    )
+    actions_source = re.search(
+        r"const API_KEY_SCOPE_ACTIONS = \[(.*?)\] as const;",
+        frontend_catalog,
+        re.DOTALL,
+    )
+
+    assert resources_source is not None
+    assert actions_source is not None
+    assert set(re.findall(r"'([^']+)'", resources_source.group(1))) == set(
+        RESOURCE_SCOPE_CLASS
+    )
+    assert set(re.findall(r"'([^']+)'", actions_source.group(1))) == set(
+        RESOURCE_SCOPE_ACTIONS
+    )
+
+
+def test_no_scopes_is_a_noop(sm: SupersetSecurityManager) -> None:
+    """No scopes requested: nothing to validate, no RBAC lookups."""
+    sm._has_view_access = MagicMock()
+    sm._validate_requested_api_key_scopes(_make_user("Gamma"), None)
+    sm._validate_requested_api_key_scopes(_make_user("Gamma"), "")
+    sm._has_view_access.assert_not_called()
+
+
+def test_per_resource_scope_allowed_when_user_has_permission(
+    sm: SupersetSecurityManager,
+) -> None:
+    """A per-resource scope the user's RBAC covers is allowed, and is checked
+    against the matching can_<method> grant."""
+    sm._has_view_access = MagicMock(return_value=True)
+    user = _make_user("Gamma")
+    sm._validate_requested_api_key_scopes(user, "superset:dashboard:read")
+    sm._has_view_access.assert_called_once_with(user, "can_read", "Dashboard")
+
+
+def test_per_resource_scope_rejected_when_user_lacks_permission(
+    sm: SupersetSecurityManager,
+) -> None:
+    """A per-resource scope beyond the user's RBAC is rejected."""
+    sm._has_view_access = MagicMock(return_value=False)
+    with pytest.raises(ValueError, match="exceeds the issuing user's own"):
+        sm._validate_requested_api_key_scopes(
+            _make_user("Gamma"), "superset:dashboard:write"
+        )
+
+
+@pytest.mark.parametrize(
+    ("scope", "registered_permission"),
+    [
+        ("superset:user:read", "can_get"),
+        ("superset:role:read", "can_get"),
+        ("superset:sqllab:write", "can_execute_sql_query"),
+    ],
+)
+def test_scope_issuance_uses_runtime_method_mapping(
+    sm: SupersetSecurityManager, scope: str, registered_permission: str
+) -> None:
+    """Issuance accepts the FAB method permission used by runtime tools."""
+    user = _make_user("Gamma")
+    sm._has_view_access = MagicMock(
+        side_effect=lambda _user, permission, _view: permission == registered_permission
+    )
+
+    sm._validate_requested_api_key_scopes(user, scope)
+
+    assert any(
+        call.args[1] == registered_permission
+        for call in sm._has_view_access.call_args_list
+    )
+
+
+def test_custom_admin_role_can_issue_flat_scope(
+    sm: SupersetSecurityManager,
+) -> None:
+    """Flat-scope issuance honors AUTH_ROLE_ADMIN rather than a fixed name."""
+    with patch("superset.security.manager.get_conf") as get_conf:
+        get_conf.return_value = {"AUTH_ROLE_ADMIN": "PlatformAdmin"}
+        sm._validate_requested_api_key_scopes(
+            _make_user("PlatformAdmin"), "superset:write"
+        )
+
+
+@pytest.mark.parametrize("action", ["delete", "update", "garbage"])
+def test_unrecognized_actions_are_rejected(
+    sm: SupersetSecurityManager, action: str
+) -> None:
+    """Actions that runtime enforcement cannot consume are rejected."""
+    sm._has_view_access = MagicMock()
+    with pytest.raises(ValueError, match="unrecognized action"):
+        sm._validate_requested_api_key_scopes(
+            _make_user("Gamma"), f"superset:chart:{action}"
+        )
+    sm._has_view_access.assert_not_called()
+
+
+def test_unrecognized_resource_slug_rejected_without_rbac_lookup(
+    sm: SupersetSecurityManager,
+) -> None:
+    """An unknown resource slug is rejected outright (fail closed) and never
+    consults RBAC."""
+    sm._has_view_access = MagicMock()
+    with pytest.raises(ValueError, match="unrecognized resource"):
+        sm._validate_requested_api_key_scopes(
+            _make_user("Admin"), "superset:notathing:read"
+        )
+    sm._has_view_access.assert_not_called()
+
+
+def test_flat_scope_allowed_for_admin(sm: SupersetSecurityManager) -> None:
+    """A flat scope (superset:write) may be self-issued by an Admin, with no
+    per-resource RBAC lookups."""
+    sm._has_view_access = MagicMock()
+    sm._validate_requested_api_key_scopes(_make_user("Admin"), "superset:write")
+    sm._has_view_access.assert_not_called()
+
+
+def test_unrecognized_flat_scope_rejected_for_admin(
+    sm: SupersetSecurityManager,
+) -> None:
+    """Admins cannot mint undefined flat scopes."""
+    sm._has_view_access = MagicMock()
+    with pytest.raises(ValueError, match="not a recognized"):
+        sm._validate_requested_api_key_scopes(_make_user("Admin"), "superset:garbage")
+    sm._has_view_access.assert_not_called()
+
+
+def test_flat_scope_rejected_for_non_admin(sm: SupersetSecurityManager) -> None:
+    """A flat scope grants a method across every resource; non-Admins cannot
+    self-issue it."""
+    sm._has_view_access = MagicMock()
+    with pytest.raises(ValueError, match="requires Admin"):
+        sm._validate_requested_api_key_scopes(_make_user("Gamma"), "superset:write")
+
+
+def test_any_failing_scope_rejects_the_whole_request(
+    sm: SupersetSecurityManager,
+) -> None:
+    """With multiple comma-separated scopes, one failure rejects the request
+    even when other scopes are individually allowed."""
+    sm._has_view_access = MagicMock(
+        side_effect=lambda user, perm, view: view == "Chart"
+    )
+    with pytest.raises(ValueError, match="exceeds the issuing user's own"):
+        sm._validate_requested_api_key_scopes(
+            _make_user("Gamma"),
+            "superset:chart:read, superset:dashboard:write",
+        )
+
+
+def test_create_api_key_rejects_before_delegating_to_fab(
+    sm: SupersetSecurityManager,
+) -> None:
+    """create_api_key validates scopes BEFORE calling FAB's implementation:
+    a rejected request never reaches FAB."""
+    sm._has_view_access = MagicMock(return_value=False)
+    with patch(
+        "flask_appbuilder.security.sqla.manager.SecurityManager.create_api_key"
+    ) as fab_create:
+        with pytest.raises(ValueError, match="exceeds the issuing user's own"):
+            sm.create_api_key(
+                user=_make_user("Gamma"),
+                name="my key",
+                scopes="superset:dashboard:write",
+            )
+        fab_create.assert_not_called()
+
+
+def test_create_api_key_delegates_to_fab_on_success(
+    sm: SupersetSecurityManager,
+) -> None:
+    """A validated request is delegated to FAB's create_api_key unchanged."""
+    sm._has_view_access = MagicMock(return_value=True)
+    user = _make_user("Gamma")
+    with patch(
+        "flask_appbuilder.security.sqla.manager.SecurityManager.create_api_key",
+        return_value={"key": "sst_secret"},
+    ) as fab_create:
+        result = sm.create_api_key(
+            user=user,
+            name="my key",
+            scopes="superset:dashboard:read",
+        )
+        fab_create.assert_called_once_with(
+            user=user, name="my key", scopes="superset:dashboard:read", expires_on=None
+        )
+    assert result == {"key": "sst_secret"}