chore(deps-dev): bump databricks-sqlalchemy from 1.0.5 to 2.0.9

Bumps [databricks-sqlalchemy](https://github.com/databricks/databricks-sqlalchemy) from 1.0.5 to 2.0.9.
- [Release notes](https://github.com/databricks/databricks-sqlalchemy/releases)
- [Changelog](https://github.com/databricks/databricks-sqlalchemy/blob/main/CHANGELOG.md)
- [Commits](https://github.com/databricks/databricks-sqlalchemy/compare/v1.0.5...v2.0.9)

---
updated-dependencies:
- dependency-name: databricks-sqlalchemy
  dependency-version: 2.0.9
  dependency-type: direct:development
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>

pyproject.toml

@@ -137,10 +137,10 @@ d1 = [
     "sqlalchemy-d1>=0.1.0",
     "dbapi-d1>=0.1.0",
 ]
-databend = ["databend-sqlalchemy>=0.3.2, <1.0"]
+databend = ["databend-sqlalchemy>=0.5.5, <1.0"]
 databricks = [
     "databricks-sql-connector==4.2.6",
-    "databricks-sqlalchemy==1.0.5",
+    "databricks-sqlalchemy==2.0.9",
 ]
 db2 = ["ibm-db-sa>0.3.8, <=0.4.4"]
 denodo = ["denodo-sqlalchemy>=2.0.5,<2.1.0"]

superset-frontend/playwright/pages/DashboardPage.ts

@@ -94,39 +94,6 @@ export class DashboardPage {
     );
   }
 
-  /**
-   * Wait for every chart on the dashboard to finish rendering and return how
-   * many charts rendered.
-   *
-   * A chart's `#chart-id-<id>` element only becomes visible once the chart has
-   * fetched its data and rendered (the same signal the legacy Cypress
-   * `waitForChartLoad` helper relied on). The chart set is derived from the
-   * dashboard itself via the `[data-test="chart-grid-component"]` holders, so
-   * this adapts to any dashboard without hard-coding chart names or counts.
-   */
-  async waitForAllChartsRendered(options?: {
-    timeout?: number;
-  }): Promise<number> {
-    // Charts issue real backend queries; allow generous time for slow viz types.
-    const timeout = options?.timeout ?? TIMEOUT.API_RESPONSE * 2;
-    const holders = this.page.locator('[data-test="chart-grid-component"]');
-    await holders.first().waitFor({ state: 'attached', timeout });
-
-    const count = await holders.count();
-    for (let i = 0; i < count; i += 1) {
-      const chartId = await holders.nth(i).getAttribute('data-test-chart-id');
-      if (!chartId) {
-        throw new Error(
-          `Chart holder ${i} is missing its data-test-chart-id attribute`,
-        );
-      }
-      await this.page
-        .locator(`#chart-id-${chartId}`)
-        .waitFor({ state: 'visible', timeout });
-    }
-    return count;
-  }
-
   /**
    * Open the dashboard header actions menu (three-dot menu)
    */

superset-frontend/playwright/tests/dashboard/dashboard-load.spec.ts

@@ -1,192 +0,0 @@
-/**
- * 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.
- */
-
-/**
- * E2E migration of the Cypress "Dashboard load" suite (dashboard/load.test.ts).
- *
- * Only the "should load dashboard" case is a genuine end-to-end test: it loads a
- * multi-chart dashboard and proves every chart renders by issuing real backend
- * queries. The remaining legacy cases (edit/standalone URL-param rendering,
- * send-log-data) only assert DOM/URL state with no backend round-trip and belong
- * in component/RTL coverage instead.
- *
- * The dashboard is built from scratch via the API (rather than relying on a
- * seeded example) so the test is hermetic, self-cleaning, and deterministic.
- *
- * CI green => the dashboard route mounts, every chart POSTs /api/v1/chart/data
- *             successfully, and each chart's render marker becomes visible.
- * CI red   => the dashboard failed to load or a chart never rendered.
- */
-import { testWithAssets, expect } from '../../helpers/fixtures';
-import { apiPostChart, apiPutChart } from '../../helpers/api/chart';
-import { apiPostDashboard } from '../../helpers/api/dashboard';
-import { getDatasetByName } from '../../helpers/api/dataset';
-import { TIMEOUT } from '../../utils/constants';
-import { DashboardPage } from '../../pages/DashboardPage';
-
-const DATASET_NAME = 'birth_names';
-
-testWithAssets(
-  'dashboard loads and every chart renders via real queries',
-  async ({ page, testAssets }) => {
-    // Building + loading a multi-chart dashboard chains several slow queries.
-    testWithAssets.setTimeout(TIMEOUT.SLOW_TEST);
-
-    const dataset = await getDatasetByName(page, DATASET_NAME);
-    if (!dataset) {
-      throw new Error(`Dataset ${DATASET_NAME} not found`);
-    }
-    const datasetId = dataset.id;
-    const datasource = `${datasetId}__table`;
-
-    // A spread of viz types that all render cleanly from the birth_names dataset.
-    const chartSpecs = [
-      {
-        viz_type: 'big_number_total',
-        params: { datasource, viz_type: 'big_number_total', metric: 'count' },
-      },
-      {
-        viz_type: 'table',
-        params: {
-          datasource,
-          viz_type: 'table',
-          query_mode: 'aggregate',
-          groupby: ['name'],
-          metrics: ['count'],
-          row_limit: 100,
-        },
-      },
-      {
-        viz_type: 'echarts_timeseries_line',
-        params: {
-          datasource,
-          viz_type: 'echarts_timeseries_line',
-          x_axis: 'ds',
-          time_grain_sqla: 'P1Y',
-          metrics: ['count'],
-          groupby: [],
-          row_limit: 100,
-        },
-      },
-    ];
-
-    // Parallel-safe suffix so chart/dashboard names never collide across workers.
-    const uniqueSuffix = `${Date.now()}_${testWithAssets.info().parallelIndex}`;
-
-    // Create each chart via the API.
-    const charts: { id: number; sliceName: string }[] = [];
-    for (const spec of chartSpecs) {
-      const sliceName = `load_smoke_${spec.viz_type}_${uniqueSuffix}`;
-      const resp = await apiPostChart(page, {
-        slice_name: sliceName,
-        viz_type: spec.viz_type,
-        datasource_id: datasetId,
-        datasource_type: 'table',
-        params: JSON.stringify(spec.params),
-      });
-      expect(resp.ok()).toBe(true);
-      const body = await resp.json();
-      const chartId: number = body.result?.id ?? body.id;
-      if (!chartId) {
-        throw new Error(
-          `Chart creation for ${spec.viz_type} returned no id: ${JSON.stringify(body)}`,
-        );
-      }
-      testAssets.trackChart(chartId);
-      charts.push({ id: chartId, sliceName });
-    }
-    const chartIds = charts.map(chart => chart.id);
-
-    // Lay all charts out in a single row.
-    const chartKeys = chartIds.map(id => `CHART-${id}`);
-    const positionJson: Record<string, unknown> = {
-      DASHBOARD_VERSION_KEY: 'v2',
-      ROOT_ID: { type: 'ROOT', id: 'ROOT_ID', children: ['GRID_ID'] },
-      GRID_ID: {
-        type: 'GRID',
-        id: 'GRID_ID',
-        children: ['ROW-1'],
-        parents: ['ROOT_ID'],
-      },
-      'ROW-1': {
-        type: 'ROW',
-        id: 'ROW-1',
-        children: chartKeys,
-        parents: ['ROOT_ID', 'GRID_ID'],
-        meta: { background: 'BACKGROUND_TRANSPARENT' },
-      },
-    };
-    chartIds.forEach((chartId, index) => {
-      positionJson[chartKeys[index]] = {
-        type: 'CHART',
-        id: chartKeys[index],
-        children: [],
-        parents: ['ROOT_ID', 'GRID_ID', 'ROW-1'],
-        meta: {
-          chartId,
-          width: 4,
-          height: 50,
-          sliceName: charts[index].sliceName,
-        },
-      };
-    });
-
-    const dashResp = await apiPostDashboard(page, {
-      dashboard_title: `load_smoke_${uniqueSuffix}`,
-      published: true,
-      position_json: JSON.stringify(positionJson),
-    });
-    expect(dashResp.ok()).toBe(true);
-    const dashBody = await dashResp.json();
-    const dashboardId: number = dashBody.result?.id ?? dashBody.id;
-    testAssets.trackDashboard(dashboardId);
-
-    // Associate every chart with the dashboard so they actually render.
-    for (const chartId of chartIds) {
-      await apiPutChart(page, chartId, { dashboards: [dashboardId] });
-    }
-
-    // Record the real chart-data round-trips the dashboard makes on load.
-    const chartDataStatuses: number[] = [];
-    page.on('response', response => {
-      const request = response.request();
-      if (
-        request.method() === 'POST' &&
-        response.url().includes('/api/v1/chart/data')
-      ) {
-        chartDataStatuses.push(response.status());
-      }
-    });
-
-    const dashboard = new DashboardPage(page);
-    await dashboard.gotoById(dashboardId);
-    await dashboard.waitForLoad();
-
-    // Every chart grid component must reach its rendered state.
-    const renderedCount = await dashboard.waitForAllChartsRendered();
-    expect(renderedCount).toBe(chartIds.length);
-
-    // The render came from real backend queries, and all of them succeeded.
-    expect(chartDataStatuses.length).toBeGreaterThan(0);
-    expect(
-      chartDataStatuses.every(status => status === 200),
-      `all /api/v1/chart/data responses should be 200, got [${chartDataStatuses}]`,
-    ).toBe(true);
-  },
-);

superset/mcp_service/CLAUDE.md

@@ -250,11 +250,17 @@ async def health_check(ctx: Context) -> dict:
 ```
 
 **Authentication priority order** (in `auth.py`):
-1. JWT context (per-request ContextVar from FastMCP)
+1. JWT context (per-request ContextVar from FastMCP). Also resolves a verified
+   embedded **guest token** to a `GuestUser` when `MCP_EMBEDDED_GUEST_AUTH_ENABLED`
+   + `EMBEDDED_SUPERSET` are on (a guest is never downgraded to a lower priority).
 2. API Key authentication (via FAB SecurityManager)
 3. `MCP_DEV_USERNAME` config (development only)
 4. `g.user` fallback (set by external middleware)
 
+Guest tokens are verified by `GuestTokenVerifier` (in the `CompositeTokenVerifier`,
+before the JWT verifier) using the shared core `GUEST_TOKEN_JWT_*` config, then
+built into a `GuestUser` in `_resolve_user_from_jwt_context`. See `SECURITY.md`.
+
 **`@mcp_auth_hook`** is only used directly on **resources** — tools get auth wrapping from `@tool(protect=True)`.
 
 ### 4. Use Pydantic Schemas
@@ -453,6 +459,11 @@ MCP_USER_RESOLVER = None         # Custom function to extract username from JWT
 # RBAC
 MCP_RBAC_ENABLED = True          # Enable permission checking (default: True)
 
+# Embedded guest auth (opt-in; requires the EMBEDDED_SUPERSET feature flag).
+# Reuses core GUEST_TOKEN_JWT_* config — no MCP-specific guest secret/audience.
+MCP_EMBEDDED_GUEST_AUTH_ENABLED = False
+MCP_GUEST_DENIED_TOOLS = {"find_users", "get_instance_info"}  # tools guests cannot call
+
 
 # Response Caching (optional, uses in-memory store by default; Redis when MCP_STORE_CONFIG enabled)
 MCP_CACHE_CONFIG = {

superset/mcp_service/SECURITY.md

@@ -209,6 +209,54 @@ superset fab create-user \
 # (Use Superset UI or FAB CLI to configure role permissions)
 ```
 
+### Embedded Guest Authentication
+
+The MCP service can accept Superset **embedded guest tokens** (the kind minted by
+`SupersetSecurityManager.create_guest_access_token` for embedded dashboards), so an
+embedded guest can drive MCP tools scoped to the dashboards/resources in its token.
+This is **opt-in** and reuses the existing core guest-token configuration — no
+MCP-specific guest secret or audience is introduced.
+
+**Enable it** (both are required):
+
+```python
+# superset_config.py
+FEATURE_FLAGS = {"EMBEDDED_SUPERSET": True}   # prerequisite — guest tokens only exist here
+MCP_EMBEDDED_GUEST_AUTH_ENABLED = True        # opt-in for the MCP transport (default False)
+```
+
+**How it works**
+
+- The guest token is presented as `Authorization: Bearer <guest_token>` (the MCP
+  transport reads only the Bearer slot, not the web `X-GuestToken` header).
+- A dedicated `GuestTokenVerifier` validates it against the shared core config —
+  `GUEST_TOKEN_JWT_SECRET` / `GUEST_TOKEN_JWT_ALGO` / `GUEST_TOKEN_JWT_AUDIENCE` —
+  replays the embedded structural checks, and enforces **revocation**
+  (`GUEST_TOKEN_REVOCATION_ENABLED` version bumps and per-dashboard
+  `guest_token_revoked_before` cutoffs). It runs **before** the MCP JWT verifier
+  (which pins its own algorithm/keys and would otherwise reject the guest token).
+- A verified guest resolves to a `GuestUser` as the **highest-priority** identity
+  source, so a guest is never downgraded to API-key / `MCP_DEV_USERNAME` / `g.user`.
+- Data access is scoped by core's existing `raise_for_access` (dataset allowlist,
+  dashboard access, RLS) — guests only reach their token's resources.
+- Sensitive enumeration tools are denied to guests via `MCP_GUEST_DENIED_TOOLS`
+  (default `{"find_users", "get_instance_info"}`), enforced at both tool listing
+  and call time regardless of `MCP_RBAC_ENABLED`.
+
+**Deployment requirements**
+
+- The MCP and web/minting services must **share `GUEST_TOKEN_JWT_SECRET` and
+  `GUEST_TOKEN_JWT_AUDIENCE`** (set `GUEST_TOKEN_JWT_AUDIENCE` explicitly; if unset,
+  audience validation falls back to the URL host, which may differ between services).
+  A startup warning is logged if guest auth is enabled while the secret is the
+  insecure default or the audience is unset.
+- The `GUEST_ROLE_NAME` role (default `Public`) must exist; a guest token is
+  rejected if it does not.
+- Do **not** set `MCP_DEV_USERNAME` on a guest-serving deployment.
+- Guest auth is wired at MCP startup; toggling `EMBEDDED_SUPERSET` or
+  `MCP_EMBEDDED_GUEST_AUTH_ENABLED` at runtime requires an MCP service restart
+  to take effect.
+
 ## Authorization
 
 ### RBAC Integration

superset/mcp_service/auth.py

@@ -46,16 +46,17 @@ Configuration:
 
 import logging
 from contextlib import AbstractContextManager, nullcontext
-from typing import Any, Callable, TYPE_CHECKING, TypeVar
+from typing import Any, Callable, cast, TYPE_CHECKING, TypeAlias, TypeVar
 
 from flask import current_app, g, has_app_context, has_request_context
 from flask_appbuilder.security.sqla.models import User
 
-from superset import security_manager
+from superset import is_feature_enabled, security_manager
 from superset.mcp_service.composite_token_verifier import (
     API_KEY_PASSTHROUGH_CLAIM,
     API_KEY_VALIDATED_USERNAME_CLAIM,
 )
+from superset.mcp_service.guest_token_verifier import GUEST_TOKEN_CLAIM
 from superset.mcp_service.mcp_config import (
     default_user_resolver,
     get_mcp_api_key_enabled,
@@ -63,16 +64,23 @@ from superset.mcp_service.mcp_config import (
 from superset.mcp_service.utils.error_sanitization import (
     sanitize_for_log as _sanitize_for_log,
 )
+from superset.security.guest_token import GuestUser
 
 if TYPE_CHECKING:
     from superset.connectors.sqla.models import SqlaTable
     from superset.mcp_service.chart.chart_utils import DatasetValidationResult
+    from superset.security.guest_token import GuestToken
 
 # Type variable for decorated functions
 F = TypeVar("F", bound=Callable[..., Any])
 
 logger = logging.getLogger(__name__)
 
+# An MCP request resolves to a real DB ``User`` or, for embedded guests, a
+# ``GuestUser`` (an AnonymousUserMixin, not a ``User`` subclass). Both are valid
+# authenticated principals for tool execution.
+MCPUser: TypeAlias = User | GuestUser
+
 # Constants for RBAC permission attributes (mirrors FAB conventions)
 PERMISSION_PREFIX = "can_"
 CLASS_PERMISSION_ATTR = "_class_permission_name"
@@ -228,7 +236,37 @@ def _log_scope_denial(
         )
 
 
-def check_tool_permission(func: Callable[..., Any], *, log_denial: bool = True) -> bool:
+# Guest deny-list default (when MCP_GUEST_DENIED_TOOLS is unset); blocks tools
+# with no RBAC class that would otherwise fall open. Sync with mcp_config.py.
+_DEFAULT_GUEST_DENIED_TOOLS: frozenset[str] = frozenset(
+    {"find_users", "get_instance_info"}
+)
+
+
+def _tool_denied_for_guest(func: Callable[..., Any]) -> bool:
+    """True when the current user is a guest and ``func`` is on the guest
+    deny-list — barring enumeration tools (user listing, instance metadata) that
+    may declare no RBAC class. ``isinstance`` (not ``is_guest_user``) keeps this
+    cheap and off the feature-flag path."""
+    if not isinstance(getattr(g, "user", None), GuestUser):
+        return False
+    denied = current_app.config.get("MCP_GUEST_DENIED_TOOLS")
+    # A str would make ``in`` do substring matching and corrupt the decision;
+    # require a real collection, else fall back to the default.
+    if not isinstance(denied, (set, frozenset, list, tuple)):
+        if denied is not None:
+            logger.warning(
+                "MCP_GUEST_DENIED_TOOLS must be a set/list of tool names, got %s; "
+                "using the default deny-list",
+                type(denied).__name__,
+            )
+        denied = _DEFAULT_GUEST_DENIED_TOOLS
+    return getattr(func, "__name__", None) in denied
+
+
+def check_tool_permission(  # noqa: C901
+    func: Callable[..., Any], *, log_denial: bool = True
+) -> bool:
     """Check if the current user has RBAC permission for an MCP tool.
 
     Reads permission metadata stored on the function by the @tool decorator
@@ -247,6 +285,22 @@ def check_tool_permission(func: Callable[..., Any], *, log_denial: bool = True)
         True if user has permission or no permission is required.
     """
     try:
+        # Embedded guests are barred from sensitive enumeration tools regardless
+        # of RBAC config (the deny-list is a guest restriction, not a FAB
+        # permission, so it must hold even when MCP_RBAC_ENABLED is False).
+        if _tool_denied_for_guest(func):
+            if log_denial:
+                logger.warning(
+                    "Tool %s denied for embedded guest (MCP_GUEST_DENIED_TOOLS)",
+                    func.__name__,
+                )
+            else:
+                logger.debug(
+                    "Tool %s hidden for embedded guest (MCP_GUEST_DENIED_TOOLS)",
+                    func.__name__,
+                )
+            return False
+
         if not current_app.config.get("MCP_RBAC_ENABLED", True):
             return True
 
@@ -344,13 +398,19 @@ def is_tool_visible_to_current_user(tool: Any) -> bool:
         True if the tool is visible to the current user, False otherwise.
     """
     try:
-        if not current_app.config.get("MCP_RBAC_ENABLED", True):
-            return True
-
         tool_func = getattr(tool, "fn", None)
         if tool_func is None:
             return True
 
+        # Hide guest-denied tools from tools/list regardless of RBAC config
+        # (enforced again at call time in check_tool_permission, including for
+        # permission-less tools).
+        if _tool_denied_for_guest(tool_func):
+            return False
+
+        if not current_app.config.get("MCP_RBAC_ENABLED", True):
+            return True
+
         from superset.mcp_service.privacy import (
             tool_requires_data_model_metadata_access,
             user_can_view_data_model_metadata,
@@ -394,7 +454,7 @@ def load_user_with_relationships(
     return security_manager.find_user_with_relationships(username=username, email=email)
 
 
-def _resolve_user_from_jwt_context(app: Any) -> User | None:
+def _resolve_user_from_jwt_context(app: Any) -> MCPUser | None:  # noqa: C901
     """
     Resolve the current user from the MCP SDK's per-request JWT context.
 
@@ -408,6 +468,8 @@ def _resolve_user_from_jwt_context(app: Any) -> User | None:
     Returns:
         User object with relationships loaded, or None if no JWT context
         (i.e. no token present — caller should fall through to next source).
+        For a verified embedded guest token (``client_id == "guest"``) returns
+        the corresponding ``GuestUser`` built from the token's resources/RLS.
 
     Raises:
         ValueError: If JWT resolves a username that doesn't exist in the DB
@@ -430,6 +492,35 @@ def _resolve_user_from_jwt_context(app: Any) -> User | None:
     # 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)
+
+    # Embedded guest token (already admitted by the GuestTokenVerifier): resolve
+    # as the highest-priority identity so a valid guest is never downgraded.
+    # Anti-forgery: only the GuestTokenVerifier sets the marker (the composite
+    # verifier strips it from JWT tokens) and this branch requires guest auth
+    # enabled, so a crafted IdP JWT with the marker can't pose as a guest.
+    if (
+        isinstance(claims, dict)
+        and claims.get(GUEST_TOKEN_CLAIM)
+        and getattr(access_token, "client_id", None) == "guest"
+    ):
+        if not (
+            is_feature_enabled("EMBEDDED_SUPERSET")
+            and app.config.get("MCP_EMBEDDED_GUEST_AUTH_ENABLED", False)
+        ):
+            logger.warning(
+                "Guest-marked token presented but embedded guest auth is not "
+                "enabled; rejecting"
+            )
+            return None
+        logger.debug("Resolving MCP request as embedded guest user")
+        # Drop the internal marker so it does not leak into GuestUser.guest_token.
+        guest_claims: dict[str, Any] = {
+            k: v for k, v in claims.items() if k != GUEST_TOKEN_CLAIM
+        }
+        return security_manager.get_guest_user_from_token(
+            cast("GuestToken", guest_claims)
+        )
+
     if isinstance(claims, dict) and claims.get(API_KEY_PASSTHROUGH_CLAIM):
         if getattr(access_token, "client_id", None) == "api_key":
             logger.debug(
@@ -607,12 +698,15 @@ def _resolve_user_from_api_key(app: Any) -> User | None:
     return _validate_api_key_fallback(app, api_key_string)
 
 
-def get_user_from_request() -> User:
+def get_user_from_request() -> MCPUser:
     """
     Get the current user for the MCP tool request.
 
     Priority order:
-    1. JWT auth context (per-request ContextVar from MCP SDK) — safest
+    1. JWT auth context (per-request ContextVar from MCP SDK) — safest. This
+       also resolves a verified embedded guest token to a ``GuestUser`` (when
+       ``MCP_EMBEDDED_GUEST_AUTH_ENABLED`` + ``EMBEDDED_SUPERSET`` are on), so a
+       guest can never be downgraded to a lower-priority source.
     2. API key from Authorization header (via FAB SecurityManager)
     3. MCP_DEV_USERNAME from configuration (for development/testing)
     4. g.user fallback (for external middleware like Preset's
@@ -740,7 +834,7 @@ def _log_user_resolution_failure(exc: ValueError | PermissionError) -> None:
         logger.error("MCP user resolution failed, denying request: %s", exc)
 
 
-def _assert_user_active(user: User | None) -> None:
+def _assert_user_active(user: MCPUser | None) -> None:
     """Raise ValueError if the user account is disabled (no-op for None)."""
     if user is None:
         return
@@ -750,7 +844,7 @@ def _assert_user_active(user: User | None) -> None:
         )
 
 
-def _setup_user_context() -> User | None:
+def _setup_user_context() -> MCPUser | None:
     """
     Set up user context for MCP tool execution.
 

superset/mcp_service/composite_token_verifier.py

@@ -34,6 +34,8 @@ from typing import Any
 from fastmcp.server.auth import AccessToken
 from fastmcp.server.auth.providers.jwt import TokenVerifier
 
+from superset.mcp_service.guest_token_verifier import GUEST_TOKEN_CLAIM
+
 logger = logging.getLogger(__name__)
 
 # Namespaced claim that flags an AccessToken as an API-key token.
@@ -61,11 +63,16 @@ class CompositeTokenVerifier(TokenVerifier):
             Bearer tokens are rejected at the transport layer (used when
             ``MCP_AUTH_ENABLED=False`` but ``FAB_API_KEY_ENABLED=True``).
         api_key_prefixes: List of prefixes that identify API key tokens
-            (e.g. ``["sst_"]``).
+            (e.g. ``["sst_"]``). Pass an empty list to disable API-key
+            acceptance (e.g. a guest-only deployment).
         app: Flask application instance used to push an app context for
             FAB SecurityManager access during token validation. When
             ``None``, prefix matching is used without DB validation
             (backward-compatible / test mode).
+        guest_verifier: Optional verifier for Superset embedded guest tokens.
+            When provided, non-API-key tokens are offered to it BEFORE the JWT
+            verifier (guest tokens are signed with a different key and would
+            otherwise be rejected by the JWT verifier).
     """
 
     def __init__(
@@ -73,12 +80,14 @@ class CompositeTokenVerifier(TokenVerifier):
         jwt_verifier: TokenVerifier | None,
         api_key_prefixes: list[str],
         app: Any = None,
+        guest_verifier: TokenVerifier | None = None,
     ) -> None:
         super().__init__(
             base_url=getattr(jwt_verifier, "base_url", None),
             required_scopes=getattr(jwt_verifier, "required_scopes", None) or [],
         )
         self._jwt_verifier = jwt_verifier
+        self._guest_verifier = guest_verifier
         self._app = app
         if app is None:
             logger.warning(
@@ -149,6 +158,10 @@ class CompositeTokenVerifier(TokenVerifier):
         - When no app is configured (test/compat mode), falls back to prefix-
           only acceptance and defers DB validation to the Flask layer.
 
+        For embedded guest tokens (when a guest verifier is configured), the
+        token is validated against the core ``GUEST_TOKEN_JWT_*`` config before
+        the JWT verifier is tried.
+
         For all other tokens, delegates to the wrapped JWT verifier when one
         is configured; rejects if no JWT verifier is configured.
         """
@@ -189,6 +202,14 @@ class CompositeTokenVerifier(TokenVerifier):
                 claims={API_KEY_PASSTHROUGH_CLAIM: True},
             )
 
+        # Guest tokens are tried before the JWT verifier (they're HS256-signed
+        # with a different key and would otherwise be rejected). A non-guest
+        # token returns None and falls through to the JWT path.
+        if self._guest_verifier is not None:
+            guest_access_token = await self._guest_verifier.verify_token(token)
+            if guest_access_token is not None:
+                return guest_access_token
+
         if self._jwt_verifier is None:
             logger.debug(
                 "Bearer token does not match any API key prefix and no JWT "
@@ -196,4 +217,19 @@ class CompositeTokenVerifier(TokenVerifier):
             )
             return None
 
-        return await self._jwt_verifier.verify_token(token)
+        jwt_access_token = await self._jwt_verifier.verify_token(token)
+        # Anti-forgery: only the GuestTokenVerifier may set the guest marker.
+        # Strip it from JWT-verified tokens so a crafted IdP JWT can't pose as
+        # a verified guest.
+        if jwt_access_token is not None:
+            jwt_claims = getattr(jwt_access_token, "claims", None)
+            if isinstance(jwt_claims, dict) and GUEST_TOKEN_CLAIM in jwt_claims:
+                # Rebuild, not mutate: claims may be immutable/copied,
+                # so pop() can no-op.
+                stripped = {
+                    k: v for k, v in jwt_claims.items() if k != GUEST_TOKEN_CLAIM
+                }
+                jwt_access_token = jwt_access_token.model_copy(
+                    update={"claims": stripped}
+                )
+        return jwt_access_token

superset/mcp_service/guest_token_verifier.py

@@ -0,0 +1,180 @@
+# 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.
+
+"""
+Guest-token verifier for the MCP service.
+
+Recognizes Superset embedded *guest tokens* presented as MCP Bearer tokens and
+validates them by reusing core's guest-token machinery
+(``SupersetSecurityManager``): signature/expiry/audience via
+``parse_jwt_guest_token`` (HS256 against ``GUEST_TOKEN_JWT_SECRET``), the same
+structural claim checks the web embedded flow runs, and revocation via
+``_is_guest_token_revoked``.
+
+This verifier runs BEFORE the JWT verifier in ``CompositeTokenVerifier``: the MCP
+JWT verifier pins its own algorithm/keys (default RS256 against the MCP
+JWKS/keys) and would reject an HS256 guest token before any resolution code runs,
+so guest tokens must be recognized at a verifier that runs first.
+
+Gated behind ``MCP_EMBEDDED_GUEST_AUTH_ENABLED`` (opt-in, default False) AND the
+``EMBEDDED_SUPERSET`` feature flag. On any failure the verifier returns ``None``
+so the token falls through to the next verifier — mirroring the web request
+loader, which returns ``None`` for an invalid guest token rather than raising.
+"""
+
+import asyncio
+import logging
+from typing import Any
+
+from fastmcp.server.auth import AccessToken
+from fastmcp.server.auth.providers.jwt import TokenVerifier
+
+logger = logging.getLogger(__name__)
+
+# Namespaced claim flagging an AccessToken as a verified guest token. The paired
+# ``client_id == "guest"`` check keeps an external IdP JWT from posing as a guest.
+GUEST_TOKEN_CLAIM: str = "_superset_mcp_guest_token"  # noqa: S105
+
+
+class GuestTokenVerifier(TokenVerifier):
+    """Verifies Superset embedded guest tokens for the MCP transport.
+
+    Args:
+        app: Flask application used to push an app context so the core
+            SecurityManager (and the metadata DB, for revocation) is reachable
+            during validation. When ``None`` the verifier is a no-op and
+            returns ``None`` for every token.
+    """
+
+    def __init__(self, app: Any = None) -> None:
+        super().__init__(base_url=None, required_scopes=[])
+        self._app = app
+        if app is None:
+            logger.warning(
+                "GuestTokenVerifier created without a Flask app; guest tokens "
+                "cannot be validated and will be rejected."
+            )
+
+    def _verify_guest_sync(self, token: str) -> dict[str, Any] | None:
+        """Validate a guest token against core's guest-token machinery.
+
+        Runs in a thread executor under a fresh app context (SecurityManager +
+        metadata DB reachable). Returns the validated claims, or ``None`` on any
+        failure (defer to the next verifier). Revocation is checked here, matching
+        the web flow.
+        """
+        if self._app is None:
+            return None
+        try:
+            with self._app.app_context():
+                # Deferred: is_feature_enabled isn't bound until app init completes.
+                from superset import is_feature_enabled
+
+                # Defense-in-depth: the verifier is only constructed when these
+                # are enabled, but never honor a guest token if embedding is off.
+                if not is_feature_enabled("EMBEDDED_SUPERSET"):
+                    return None
+                if not self._app.config.get("MCP_EMBEDDED_GUEST_AUTH_ENABLED", False):
+                    return None
+
+                sm = self._app.appbuilder.sm
+                try:
+                    parsed = sm.parse_jwt_guest_token(token)
+                except Exception:  # noqa: BLE001 — not a guest token / bad sig / exp
+                    # Most Bearer tokens aren't guest tokens (e.g. a regular OIDC
+                    # JWT for the JWT verifier); keep quiet and fall through.
+                    logger.debug("Bearer token is not a valid guest token; deferring")
+                    return None
+
+                # Mirror the web embedded flow's structural validation
+                # (SupersetSecurityManager.get_guest_user_from_request).
+                if (
+                    not isinstance(parsed, dict)
+                    or parsed.get("user") is None
+                    or parsed.get("resources") is None
+                    or parsed.get("rls_rules") is None
+                    or parsed.get("type") != "guest"
+                ):
+                    logger.debug("Guest token failed structural validation; deferring")
+                    return None
+
+                if sm._is_guest_token_revoked(parsed):  # noqa: SLF001
+                    logger.debug("Guest token has been revoked; rejecting")
+                    return None
+
+                # GuestUser is built from GUEST_ROLE_NAME; a missing role fails
+                # RBAC confusingly, so reject up front with an actionable message.
+                role_name = self._app.config["GUEST_ROLE_NAME"]
+                if sm.find_role(role_name) is None:
+                    logger.error(
+                        "Guest token is valid but the guest role GUEST_ROLE_NAME=%r "
+                        "does not exist; rejecting. Create the role to enable "
+                        "embedded guest access over MCP.",
+                        role_name,
+                    )
+                    return None
+
+                return parsed
+        except Exception:  # noqa: BLE001 — DB errors, FAB internals, etc.
+            logger.warning(
+                "Guest token transport validation failed unexpectedly; rejecting",
+                exc_info=True,
+            )
+            return None
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """Return a guest ``AccessToken`` when ``token`` is a valid guest token.
+
+        Returns ``None`` (defer to the next verifier) when guest auth is disabled
+        or the token is not a valid, non-revoked guest token.
+        """
+        if self._app is None:
+            return None
+        # Cheap opt-in gate before paying for a thread dispatch + DB work.
+        if not self._app.config.get("MCP_EMBEDDED_GUEST_AUTH_ENABLED", False):
+            return None
+
+        try:
+            loop = asyncio.get_running_loop()
+            parsed = await loop.run_in_executor(None, self._verify_guest_sync, token)
+        except Exception:  # noqa: BLE001 — asyncio/executor machinery failure
+            # Honor the "any failure -> defer" contract even if the dispatch
+            # itself fails (e.g. executor shutdown, no running loop).
+            logger.warning(
+                "Guest token verification dispatch failed; deferring", exc_info=True
+            )
+            return None
+
+        if parsed is None:
+            return None
+
+        try:
+            expires_at = int(parsed["exp"])
+        except (KeyError, TypeError, ValueError):
+            # parse_jwt_guest_token validates exp, but never raise here on a
+            # missing/malformed claim — defer like every other failure path.
+            logger.warning("Guest token lacks a valid exp claim; deferring")
+            return None
+
+        logger.debug("Guest token validated at transport layer for MCP")
+        return AccessToken(
+            token=token,
+            client_id="guest",
+            scopes=[],
+            expires_at=expires_at,
+            claims={GUEST_TOKEN_CLAIM: True, **parsed},
+        )

superset/mcp_service/mcp_config.py

@@ -24,11 +24,13 @@ from typing import Any, Dict, Optional, Sequence
 from fastmcp.server.auth.providers.jwt import JWTVerifier
 from flask import Flask
 
+from superset.constants import CHANGE_ME_GUEST_TOKEN_JWT_SECRET
 from superset.mcp_service.composite_token_verifier import CompositeTokenVerifier
 from superset.mcp_service.constants import (
     DEFAULT_TOKEN_LIMIT,
     DEFAULT_WARN_THRESHOLD_PCT,
 )
+from superset.mcp_service.guest_token_verifier import GuestTokenVerifier
 from superset.mcp_service.jwt_verifier import DetailedJWTVerifier, MCPJWTVerifier
 
 logger = logging.getLogger(__name__)
@@ -147,6 +149,15 @@ MCP_API_KEY_ENABLED: bool | None = None
 # own key-management UI without forking the auth code.
 MCP_API_KEY_CREATE_URL = "/profile/"
 
+# Accept Superset embedded guest tokens on the MCP transport (opt-in, default
+# False). Requires the EMBEDDED_SUPERSET flag and the shared core
+# GUEST_TOKEN_JWT_* config. See SECURITY.md "Embedded Guest Authentication".
+MCP_EMBEDDED_GUEST_AUTH_ENABLED: bool = False
+
+# Tools a guest may never call (enforced at tools/list and call time, regardless
+# of RBAC). Sync with _DEFAULT_GUEST_DENIED_TOOLS in auth.py.
+MCP_GUEST_DENIED_TOOLS: set[str] = {"find_users", "get_instance_info"}
+
 
 # Session configuration for local development
 MCP_SESSION_CONFIG = {
@@ -408,8 +419,9 @@ def create_default_mcp_auth_factory(app: Flask) -> Optional[Any]:
     """
     auth_enabled = app.config.get("MCP_AUTH_ENABLED", False)
     api_key_enabled = get_mcp_api_key_enabled(app, startup_warning=True)
+    guest_enabled = _is_mcp_guest_auth_enabled(app)
 
-    if not (auth_enabled or api_key_enabled):
+    if not (auth_enabled or api_key_enabled or guest_enabled):
         return None
 
     # When JWT auth is enabled, an audience must be configured so issued tokens
@@ -435,7 +447,7 @@ def create_default_mcp_auth_factory(app: Flask) -> Optional[Any]:
 
         if not (jwks_uri or public_key or secret):
             logger.warning("MCP_AUTH_ENABLED is True but no JWT keys/secret configured")
-            if not api_key_enabled:
+            if not (api_key_enabled or guest_enabled):
                 return None
         else:
             try:
@@ -448,45 +460,117 @@ def create_default_mcp_auth_factory(app: Flask) -> Optional[Any]:
             except Exception:
                 # Do not log the exception — it may contain secrets (e.g., key material)
                 logger.error("Failed to create MCP JWT verifier")
-                if not api_key_enabled:
+                if not (api_key_enabled or guest_enabled):
                     return None
 
-    if api_key_enabled:
-        return _build_composite_verifier(app, jwt_verifier)
+    # A composite verifier is needed whenever API-key OR guest auth is on, so
+    # those token types are recognized before (or instead of) the JWT verifier.
+    if api_key_enabled or guest_enabled:
+        return _build_composite_verifier(
+            app,
+            jwt_verifier,
+            api_key_enabled=api_key_enabled,
+            guest_enabled=guest_enabled,
+        )
 
     return jwt_verifier
 
 
-def _build_composite_verifier(app: Flask, jwt_verifier: Any) -> CompositeTokenVerifier:
-    """Build a CompositeTokenVerifier with API key prefixes from config."""
-    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.",
-            required_scopes,
-        )
-    raw_prefixes: str | Sequence[str] = app.config.get("FAB_API_KEY_PREFIXES", ["sst_"])
-    # Normalize: a plain string (e.g. "sst_") would iterate as characters;
-    # wrap it in a list so CompositeTokenVerifier receives a proper sequence.
-    # Guard against non-iterable config values (e.g. None, integers) that
-    # would raise TypeError and cause _create_auth_provider to fail open.
-    if isinstance(raw_prefixes, str):
-        api_key_prefixes: list[str] = [raw_prefixes]
-    else:
-        try:
-            api_key_prefixes = list(raw_prefixes)
-        except TypeError:
+def _is_mcp_guest_auth_enabled(app: Flask) -> bool:
+    """Return True when embedded guest auth should be active for the MCP transport.
+
+    Requires the opt-in ``MCP_EMBEDDED_GUEST_AUTH_ENABLED`` config AND the
+    ``EMBEDDED_SUPERSET`` feature flag — guest tokens only exist, and
+    ``is_guest_user`` only returns True, when embedding is enabled.
+    """
+    if not app.config.get("MCP_EMBEDDED_GUEST_AUTH_ENABLED", False):
+        return False
+    with app.app_context():
+        # Deferred: is_feature_enabled isn't bound until app init completes.
+        from superset import is_feature_enabled
+
+        if not is_feature_enabled("EMBEDDED_SUPERSET"):
             logger.warning(
-                "FAB_API_KEY_PREFIXES must be a string or list; using default"
+                "MCP_EMBEDDED_GUEST_AUTH_ENABLED is True but the EMBEDDED_SUPERSET "
+                "feature flag is disabled; embedded guest auth for MCP will not be "
+                "enabled. Enable EMBEDDED_SUPERSET to accept guest tokens over MCP."
             )
-            api_key_prefixes = ["sst_"]
-    logger.info("API key auth enabled for MCP")
+            return False
+    return True
+
+
+def _validate_guest_config(app: Flask) -> None:
+    """Hard-fail on the default GUEST_TOKEN_JWT_SECRET; warn on an unset audience."""
+    if app.config.get("GUEST_TOKEN_JWT_SECRET") == CHANGE_ME_GUEST_TOKEN_JWT_SECRET:
+        # MCPAuthConfigError specifically: the bootstrap re-raises this type to
+        # refuse startup but swallows others. See _create_auth_provider.
+        raise MCPAuthConfigError(
+            "MCP_EMBEDDED_GUEST_AUTH_ENABLED is set but GUEST_TOKEN_JWT_SECRET is "
+            "the insecure default; refusing to wire guest auth. Set a strong "
+            "GUEST_TOKEN_JWT_SECRET shared with the guest-token minting service."
+        )
+    if not app.config.get("GUEST_TOKEN_JWT_AUDIENCE"):
+        # Don't interpolate the fallback host: CodeQL flags logging config-derived
+        # values as clear-text secrets, and the warning alone suffices.
+        logger.warning(
+            "MCP embedded guest auth enabled but GUEST_TOKEN_JWT_AUDIENCE is unset; "
+            "audience validation falls back to the request URL host. Set "
+            "GUEST_TOKEN_JWT_AUDIENCE consistently across the web and MCP services."
+        )
+
+
+def _build_composite_verifier(
+    app: Flask,
+    jwt_verifier: Any,
+    *,
+    api_key_enabled: bool = True,
+    guest_enabled: bool = False,
+) -> CompositeTokenVerifier:
+    """Build a CompositeTokenVerifier wiring API-key and/or guest verification.
+
+    ``api_key_prefixes`` is left empty when API-key auth is disabled (e.g. a
+    guest-only deployment) so API-key tokens are not silently accepted.
+    """
+    api_key_prefixes: list[str] = []
+    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.",
+                required_scopes,
+            )
+        raw_prefixes: str | Sequence[str] = app.config.get(
+            "FAB_API_KEY_PREFIXES", ["sst_"]
+        )
+        # Normalize: a plain string (e.g. "sst_") would iterate as characters;
+        # wrap it in a list so CompositeTokenVerifier receives a proper sequence.
+        # Guard against non-iterable config values (e.g. None, integers) that
+        # would raise TypeError and cause _create_auth_provider to fail open.
+        if isinstance(raw_prefixes, str):
+            api_key_prefixes = [raw_prefixes]
+        else:
+            try:
+                api_key_prefixes = list(raw_prefixes)
+            except TypeError:
+                logger.warning(
+                    "FAB_API_KEY_PREFIXES must be a string or list; using default"
+                )
+                api_key_prefixes = ["sst_"]
+        logger.info("API key auth enabled for MCP")
+
+    guest_verifier: GuestTokenVerifier | None = None
+    if guest_enabled:
+        _validate_guest_config(app)
+        guest_verifier = GuestTokenVerifier(app=app)
+        logger.info("Embedded guest token auth enabled for MCP")
+
     return CompositeTokenVerifier(
         jwt_verifier=jwt_verifier,
         api_key_prefixes=api_key_prefixes,
         app=app,
+        guest_verifier=guest_verifier,
     )
 
 
@@ -587,6 +671,8 @@ def get_mcp_config(app_config: dict[str, Any] | None = None) -> dict[str, Any]:
         "MCP_DISABLED_TOOLS": set(MCP_DISABLED_TOOLS),
         "MCP_DISABLED_CHART_PLUGINS": MCP_DISABLED_CHART_PLUGINS,
         "MCP_CHART_PLUGIN_ENABLED_FUNC": MCP_CHART_PLUGIN_ENABLED_FUNC,
+        "MCP_EMBEDDED_GUEST_AUTH_ENABLED": MCP_EMBEDDED_GUEST_AUTH_ENABLED,
+        "MCP_GUEST_DENIED_TOOLS": set(MCP_GUEST_DENIED_TOOLS),
         **MCP_SESSION_CONFIG,
         **MCP_CSRF_CONFIG,
     }

superset/mcp_service/server.py

@@ -747,6 +747,7 @@ def _create_auth_provider(flask_app: Any) -> Any | None:
         flask_app.config.get("MCP_AUTH_ENABLED", False)
         or flask_app.config.get("MCP_API_KEY_ENABLED", False)
         or flask_app.config.get("FAB_API_KEY_ENABLED", False)
+        or flask_app.config.get("MCP_EMBEDDED_GUEST_AUTH_ENABLED", False)
     ):
         from superset.mcp_service.mcp_config import (
             create_default_mcp_auth_factory,

superset/models/helpers.py

@@ -2912,6 +2912,17 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
             )
         )
 
+        # Honor the dataset "Hour Offset". Result timestamps are displayed shifted
+        # by +offset hours (see normalize_df / DateColumn in superset.utils.core),
+        # but the time filter compares the raw stored values. Shifting the filter
+        # bounds by -offset keeps the filter consistent with what is displayed;
+        # otherwise a date selection lands on the wrong calendar day (#104810).
+        if offset_hours := getattr(self, "offset", 0) or 0:
+            if start_dttm is not None:
+                start_dttm = start_dttm - timedelta(hours=offset_hours)
+            if end_dttm is not None:
+                end_dttm = end_dttm - timedelta(hours=offset_hours)
+
         l = []  # noqa: E741
         if start_dttm:
             l.append(

tests/integration_tests/query_context_tests.py

@@ -124,6 +124,10 @@ def test_time_offset_comparison_queries_use_chart_row_limit(
     def cache_timeout_fn() -> int:
         return query_context._processor.get_cache_timeout()
 
+    # A non-zero dataset Hour Offset shifts temporal filter bounds (#104810) and
+    # other tests can leave one set on the shared birth_names table; pin it to 0
+    # so the comparison-window literals below stay deterministic.
+    query_context.datasource.offset = 0
     time_offsets_obj = query_context.datasource.processing_time_offsets(
         df, query_object, cache_key_fn, cache_timeout_fn, query_context.force
     )

tests/integration_tests/sqla_models_tests.py

@@ -304,6 +304,11 @@ class TestDatabaseModel(SupersetTestCase):
             ),
         )
         table = self.get_table(name="birth_names")
+        # This test targets filter operators, not the dataset Hour Offset. A
+        # non-zero offset shifts temporal filter bounds (#104810) and other tests
+        # in the suite can leave one set on the shared birth_names table, so pin
+        # it to 0 here to keep the temporal-range literal deterministic.
+        table.offset = 0
         for filter_ in filters:
             query_obj = {
                 "granularity": None,

tests/unit_tests/mcp_service/test_guest_token_auth.py

@@ -0,0 +1,553 @@
+# 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 embedded guest-token authentication on the MCP transport.
+
+Covers the three layers of the feature:
+- ``GuestTokenVerifier`` (transport): validates a guest token by reusing core's
+  guest-token machinery and emits a marked ``AccessToken``.
+- ``CompositeTokenVerifier`` routing: guests are tried before the JWT verifier.
+- ``_resolve_user_from_jwt_context`` (resolution): builds the ``GuestUser`` from
+  the verified claims, and ignores look-alike tokens lacking the marker.
+- The guest deny-list for sensitive enumeration tools.
+"""
+
+from contextlib import contextmanager
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import jwt
+import pytest
+from fastmcp.server.auth import AccessToken
+from flask import g
+
+from superset.app import SupersetApp
+from superset.constants import CHANGE_ME_GUEST_TOKEN_JWT_SECRET
+from superset.mcp_service.auth import (
+    _resolve_user_from_jwt_context,
+    _tool_denied_for_guest,
+    check_tool_permission,
+    CLASS_PERMISSION_ATTR,
+    is_tool_visible_to_current_user,
+    METHOD_PERMISSION_ATTR,
+)
+from superset.mcp_service.composite_token_verifier import CompositeTokenVerifier
+from superset.mcp_service.guest_token_verifier import (
+    GUEST_TOKEN_CLAIM,
+    GuestTokenVerifier,
+)
+from superset.mcp_service.mcp_config import (
+    _is_mcp_guest_auth_enabled,
+    _validate_guest_config,
+    MCPAuthConfigError,
+)
+from superset.security.guest_token import GuestUser
+
+
+def _access_token(client_id: str, claims: dict[str, Any]) -> AccessToken:
+    """Build a fastmcp AccessToken for routing tests."""
+    fake_token = "fake-token"  # noqa: S105 — test fixture, not a real credential
+    return AccessToken(token=fake_token, client_id=client_id, scopes=[], claims=claims)
+
+
+def _parsed_guest_claims(**overrides: Any) -> dict[str, Any]:
+    """A decoded guest-token claims dict shaped like core's mint output."""
+    claims = {
+        "user": {"username": "embed-guest"},
+        "resources": [{"type": "dashboard", "id": "abc-uuid"}],
+        "rls_rules": [],
+        "iat": 1,
+        "exp": 9_999_999_999,
+        "aud": "superset",
+        "type": "guest",
+    }
+    claims.update(overrides)
+    return claims
+
+
+class _FakeApp:
+    """Minimal Flask-app stand-in for unit-testing GuestTokenVerifier."""
+
+    def __init__(self, config: dict[str, Any], sm: Any) -> None:
+        self.config = config
+        self.appbuilder = SimpleNamespace(sm=sm)
+
+    @contextmanager
+    def app_context(self) -> Any:
+        yield
+
+
+def _make_guest_verifier(
+    *,
+    parsed: dict[str, Any] | None = None,
+    parse_error: Exception | None = None,
+    revoked: bool = False,
+    role_exists: bool = True,
+    mcp_enabled: bool = True,
+) -> tuple[GuestTokenVerifier, MagicMock]:
+    sm = MagicMock()
+    if parse_error is not None:
+        sm.parse_jwt_guest_token.side_effect = parse_error
+    else:
+        sm.parse_jwt_guest_token.return_value = (
+            _parsed_guest_claims() if parsed is None else parsed
+        )
+    sm._is_guest_token_revoked.return_value = revoked
+    sm.find_role.return_value = object() if role_exists else None
+    config = {
+        "MCP_EMBEDDED_GUEST_AUTH_ENABLED": mcp_enabled,
+        "GUEST_ROLE_NAME": "Public",
+    }
+    return GuestTokenVerifier(app=_FakeApp(config, sm)), sm
+
+
+# -- GuestTokenVerifier --
+
+
+@pytest.mark.asyncio
+async def test_guest_verifier_accepts_valid_token() -> None:
+    verifier, _ = _make_guest_verifier()
+    with patch("superset.is_feature_enabled", return_value=True):
+        token = await verifier.verify_token("raw-guest-token")
+
+    assert token is not None
+    assert token.client_id == "guest"
+    assert token.claims[GUEST_TOKEN_CLAIM] is True
+    # The parsed guest claims are carried through for resolution.
+    assert token.claims["user"] == {"username": "embed-guest"}
+    assert token.claims["type"] == "guest"
+
+
+@pytest.mark.asyncio
+async def test_guest_verifier_noop_when_mcp_flag_off() -> None:
+    verifier, sm = _make_guest_verifier(mcp_enabled=False)
+    with patch("superset.is_feature_enabled", return_value=True):
+        token = await verifier.verify_token("raw-guest-token")
+
+    assert token is None
+    # Short-circuits before doing any parsing work.
+    sm.parse_jwt_guest_token.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_guest_verifier_defers_when_embedded_disabled() -> None:
+    verifier, _ = _make_guest_verifier()
+    with patch("superset.is_feature_enabled", return_value=False):
+        token = await verifier.verify_token("raw-guest-token")
+
+    assert token is None
+
+
+@pytest.mark.asyncio
+async def test_guest_verifier_defers_on_bad_signature() -> None:
+    verifier, _ = _make_guest_verifier(parse_error=ValueError("bad signature"))
+    with patch("superset.is_feature_enabled", return_value=True):
+        token = await verifier.verify_token("not-a-guest-token")
+
+    assert token is None
+
+
+@pytest.mark.asyncio
+async def test_guest_verifier_rejects_non_guest_type() -> None:
+    verifier, _ = _make_guest_verifier(parsed=_parsed_guest_claims(type="access"))
+    with patch("superset.is_feature_enabled", return_value=True):
+        token = await verifier.verify_token("raw-token")
+
+    assert token is None
+
+
+@pytest.mark.asyncio
+async def test_guest_verifier_rejects_missing_structural_claims() -> None:
+    bad = _parsed_guest_claims()
+    del bad["resources"]
+    verifier, _ = _make_guest_verifier(parsed=bad)
+    with patch("superset.is_feature_enabled", return_value=True):
+        token = await verifier.verify_token("raw-token")
+
+    assert token is None
+
+
+@pytest.mark.asyncio
+async def test_guest_verifier_rejects_revoked_token() -> None:
+    verifier, _ = _make_guest_verifier(revoked=True)
+    with patch("superset.is_feature_enabled", return_value=True):
+        token = await verifier.verify_token("raw-guest-token")
+
+    assert token is None
+
+
+@pytest.mark.asyncio
+async def test_guest_verifier_rejects_when_guest_role_missing() -> None:
+    verifier, _ = _make_guest_verifier(role_exists=False)
+    with patch("superset.is_feature_enabled", return_value=True):
+        token = await verifier.verify_token("raw-guest-token")
+
+    assert token is None
+
+
+# -- CompositeTokenVerifier routing --
+
+
+class _StubVerifier:
+    base_url = None
+    required_scopes: list[str] = []
+
+    def __init__(self, return_value: Any) -> None:
+        self._return_value = return_value
+        self.called = False
+
+    async def verify_token(self, token: str) -> Any:
+        self.called = True
+        return self._return_value
+
+
+@pytest.mark.asyncio
+async def test_composite_tries_guest_before_jwt() -> None:
+    guest_at = _access_token("guest", {GUEST_TOKEN_CLAIM: True})
+    guest = _StubVerifier(guest_at)
+    jwt = _StubVerifier(_access_token("idp", {}))
+    composite = CompositeTokenVerifier(
+        jwt_verifier=jwt, api_key_prefixes=[], app=None, guest_verifier=guest
+    )
+
+    result = await composite.verify_token("some-token")
+
+    assert result is guest_at
+    assert guest.called is True
+    assert jwt.called is False  # guest short-circuits the JWT path
+
+
+@pytest.mark.asyncio
+async def test_composite_falls_through_to_jwt_for_non_guest() -> None:
+    jwt_result = _access_token("idp", {})
+    guest = _StubVerifier(None)  # not a guest token
+    jwt = _StubVerifier(jwt_result)
+    composite = CompositeTokenVerifier(
+        jwt_verifier=jwt, api_key_prefixes=[], app=None, guest_verifier=guest
+    )
+
+    result = await composite.verify_token("some-token")
+
+    assert result is jwt_result
+    assert guest.called is True
+    assert jwt.called is True
+
+
+# -- _resolve_user_from_jwt_context --
+
+
+def test_resolve_builds_guest_user_from_token(app: SupersetApp) -> None:
+    token = MagicMock()
+    token.claims = {GUEST_TOKEN_CLAIM: True, **_parsed_guest_claims()}
+    token.client_id = "guest"
+    fake_guest = MagicMock()
+    fake_guest.username = "embed-guest"
+
+    with app.app_context():
+        with (
+            patch.dict(app.config, {"MCP_EMBEDDED_GUEST_AUTH_ENABLED": True}),
+            patch("fastmcp.server.dependencies.get_access_token", return_value=token),
+            patch("superset.mcp_service.auth.is_feature_enabled", return_value=True),
+            patch("superset.mcp_service.auth.security_manager") as mock_sm,
+        ):
+            mock_sm.get_guest_user_from_token = MagicMock(return_value=fake_guest)
+            result = _resolve_user_from_jwt_context(app)
+
+    assert result is fake_guest
+    # The internal marker must not leak into the GuestUser's token dict.
+    passed_token = mock_sm.get_guest_user_from_token.call_args.args[0]
+    assert GUEST_TOKEN_CLAIM not in passed_token
+
+
+def test_resolve_ignores_guest_type_without_marker(app: SupersetApp) -> None:
+    """An external IdP JWT that merely carries ``type==guest`` must NOT be
+    treated as an embedded guest — it lacks the namespaced marker and the
+    ``client_id == "guest"`` signal, so normal resolution applies."""
+    mock_user = MagicMock()
+    mock_user.username = "alice"
+    mock_user.roles = []
+    mock_user.groups = []
+    token = MagicMock()
+    token.claims = {"type": "guest", "sub": "alice"}
+    token.client_id = "idp"
+
+    with app.app_context():
+        with (
+            patch("fastmcp.server.dependencies.get_access_token", return_value=token),
+            patch(
+                "superset.mcp_service.auth.load_user_with_relationships",
+                return_value=mock_user,
+            ),
+        ):
+            result = _resolve_user_from_jwt_context(app)
+
+    assert result is not None
+    assert result.username == "alice"
+
+
+# -- Guest deny-list --
+
+
+def _make_guest_user() -> GuestUser:
+    return GuestUser(
+        token={
+            "user": {"username": "g"},
+            "resources": [],
+            "rls_rules": [],
+            "iat": 1,
+            "exp": 9_999_999_999,
+        },
+        roles=[],
+    )
+
+
+def test_tool_denied_for_guest_helper(app: SupersetApp) -> None:
+    denied = MagicMock()
+    denied.__name__ = "find_users"
+    allowed = MagicMock()
+    allowed.__name__ = "list_charts"
+
+    with app.app_context():
+        # Non-guest: deny-list never applies.
+        g.user = MagicMock(spec=[])
+        assert _tool_denied_for_guest(denied) is False
+
+        # Guest: denied tool blocked, non-denied tool allowed.
+        g.user = _make_guest_user()
+        assert _tool_denied_for_guest(denied) is True
+        assert _tool_denied_for_guest(allowed) is False
+
+
+def test_guest_denied_tool_blocked_at_call_time(app: SupersetApp) -> None:
+    func = MagicMock()
+    func.__name__ = "find_users"
+
+    with app.app_context():
+        g.user = _make_guest_user()
+        # Denied regardless of MCP_RBAC_ENABLED (it runs before the gate).
+        assert check_tool_permission(func) is False
+
+
+def test_guest_denied_tool_hidden_from_listing(app: SupersetApp) -> None:
+    tool = MagicMock()
+    tool.fn = MagicMock()
+    tool.fn.__name__ = "get_instance_info"
+
+    with app.app_context():
+        g.user = _make_guest_user()
+        assert is_tool_visible_to_current_user(tool) is False
+
+
+def test_guest_allowed_tool_permitted_with_rbac(app: SupersetApp) -> None:
+    """A guest passes the deny-list and is granted a non-denied tool through the
+    full RBAC chain (deny-list runs, RBAC grants, scopes allow)."""
+    func = MagicMock()
+    func.__name__ = "list_charts"
+    setattr(func, CLASS_PERMISSION_ATTR, "Chart")
+    setattr(func, METHOD_PERMISSION_ATTR, "read")
+
+    with app.app_context():
+        with (
+            patch.dict(app.config, {"MCP_RBAC_ENABLED": True}),
+            patch("superset.mcp_service.auth.security_manager") as mock_sm,
+            patch("superset.mcp_service.auth._token_scope_allows", return_value=True),
+        ):
+            mock_sm.can_access = MagicMock(return_value=True)
+            g.user = _make_guest_user()
+            assert check_tool_permission(func) is True
+
+
+def test_denylist_string_misconfig_falls_back_to_default(app: SupersetApp) -> None:
+    """A misconfigured string deny-list must not cause substring matching; the
+    type guard falls back to the safe default set."""
+    func = MagicMock()
+    func.__name__ = "find_user"  # substring of "find_users"
+
+    with app.app_context():
+        with patch.dict(
+            app.config, {"MCP_GUEST_DENIED_TOOLS": "find_users,get_instance_info"}
+        ):
+            g.user = _make_guest_user()
+            # Naive `in` on the string would wrongly match "find_user"; the guard
+            # falls back to the default set, so "find_user" is allowed...
+            assert _tool_denied_for_guest(func) is False
+            # ...while a genuinely-denied tool is still blocked via the default.
+            func.__name__ = "find_users"
+            assert _tool_denied_for_guest(func) is True
+
+
+def test_guest_denied_tools_operator_override(app: SupersetApp) -> None:
+    func = MagicMock()
+    func.__name__ = "execute_sql"
+
+    with app.app_context():
+        with patch.dict(app.config, {"MCP_GUEST_DENIED_TOOLS": {"execute_sql"}}):
+            g.user = _make_guest_user()
+            assert _tool_denied_for_guest(func) is True
+
+
+# -- Hardening: expiry, forgery, and the resolution gate --
+
+
+@pytest.mark.asyncio
+async def test_guest_verifier_defers_on_expired_token() -> None:
+    verifier, _ = _make_guest_verifier(parse_error=jwt.ExpiredSignatureError("expired"))
+    with patch("superset.is_feature_enabled", return_value=True):
+        token = await verifier.verify_token("expired-guest-token")
+
+    assert token is None
+
+
+@pytest.mark.asyncio
+async def test_verify_token_defers_on_missing_exp() -> None:
+    """A successfully-parsed token with no usable ``exp`` must defer, not raise."""
+    bad = _parsed_guest_claims()
+    del bad["exp"]
+    verifier, _ = _make_guest_verifier(parsed=bad)
+    with patch("superset.is_feature_enabled", return_value=True):
+        token = await verifier.verify_token("raw-token")
+
+    assert token is None
+
+
+@pytest.mark.asyncio
+async def test_composite_strips_guest_marker_from_jwt_token() -> None:
+    """A crafted IdP JWT carrying the guest marker must have it stripped by the
+    composite, so it cannot be mistaken for a verified guest at resolution. The
+    strip rebuilds the token rather than mutating in place, so it holds even when
+    ``AccessToken.claims`` is an immutable or copied mapping."""
+    forged = _access_token("guest", {GUEST_TOKEN_CLAIM: True, "sub": "attacker"})
+    jwt_verifier = _StubVerifier(forged)
+    composite = CompositeTokenVerifier(
+        jwt_verifier=jwt_verifier, api_key_prefixes=[], app=None, guest_verifier=None
+    )
+
+    result = await composite.verify_token("crafted-idp-jwt")
+
+    assert result is not None
+    assert GUEST_TOKEN_CLAIM not in result.claims
+    assert result.claims["sub"] == "attacker"
+    # Rebuilt, not mutated: the original token still carries the marker.
+    assert forged.claims[GUEST_TOKEN_CLAIM] is True
+
+
+def test_resolve_rejects_guest_marker_when_guest_auth_disabled(
+    app: SupersetApp,
+) -> None:
+    """Even with the marker + client_id, a token is not treated as a guest when
+    embedded guest auth is disabled (defense against marker forgery)."""
+    token = MagicMock()
+    token.claims = {GUEST_TOKEN_CLAIM: True, **_parsed_guest_claims()}
+    token.client_id = "guest"
+
+    with app.app_context():
+        with (
+            patch.dict(app.config, {"MCP_EMBEDDED_GUEST_AUTH_ENABLED": False}),
+            patch("fastmcp.server.dependencies.get_access_token", return_value=token),
+            patch("superset.mcp_service.auth.is_feature_enabled", return_value=True),
+        ):
+            result = _resolve_user_from_jwt_context(app)
+
+    assert result is None
+
+
+def test_resolve_rejects_guest_marker_when_embedded_flag_off(app: SupersetApp) -> None:
+    """The other half of the gate: with the marker + client_id + the MCP guest
+    flag on, a token is still not treated as a guest when the EMBEDDED_SUPERSET
+    feature flag is off (both gates are required)."""
+    token = MagicMock()
+    token.claims = {GUEST_TOKEN_CLAIM: True, **_parsed_guest_claims()}
+    token.client_id = "guest"
+
+    with app.app_context():
+        with (
+            patch.dict(app.config, {"MCP_EMBEDDED_GUEST_AUTH_ENABLED": True}),
+            patch("fastmcp.server.dependencies.get_access_token", return_value=token),
+            patch("superset.mcp_service.auth.is_feature_enabled", return_value=False),
+        ):
+            result = _resolve_user_from_jwt_context(app)
+
+    assert result is None
+
+
+def test_resolve_ignores_guest_marker_without_guest_client_id(app: SupersetApp) -> None:
+    """The marker alone is not enough — ``client_id == "guest"`` is also required,
+    so a normal JWT that carries the marker resolves as its own user."""
+    mock_user = MagicMock()
+    mock_user.username = "alice"
+    mock_user.roles = []
+    mock_user.groups = []
+    token = MagicMock()
+    token.claims = {GUEST_TOKEN_CLAIM: True, "sub": "alice"}
+    token.client_id = "idp"  # not "guest"
+
+    with app.app_context():
+        with (
+            patch.dict(app.config, {"MCP_EMBEDDED_GUEST_AUTH_ENABLED": True}),
+            patch("fastmcp.server.dependencies.get_access_token", return_value=token),
+            patch(
+                "superset.mcp_service.auth.load_user_with_relationships",
+                return_value=mock_user,
+            ),
+        ):
+            result = _resolve_user_from_jwt_context(app)
+
+    assert result is not None
+    assert result.username == "alice"
+
+
+# -- guest config validation --
+
+
+def test_validate_guest_config_raises_on_default_secret() -> None:
+    app = _FakeApp(
+        {"GUEST_TOKEN_JWT_SECRET": CHANGE_ME_GUEST_TOKEN_JWT_SECRET}, MagicMock()
+    )
+    with pytest.raises(MCPAuthConfigError, match="insecure default"):
+        _validate_guest_config(app)
+
+
+def test_validate_guest_config_passes_with_strong_secret() -> None:
+    app = _FakeApp(
+        {
+            "GUEST_TOKEN_JWT_SECRET": "a-strong-shared-secret",
+            "GUEST_TOKEN_JWT_AUDIENCE": "superset",
+        },
+        MagicMock(),
+    )
+    _validate_guest_config(app)  # does not raise
+
+
+# -- guest auth enablement gate --
+
+
+def test_mcp_guest_auth_disabled_when_embedded_flag_off() -> None:
+    app = _FakeApp({"MCP_EMBEDDED_GUEST_AUTH_ENABLED": True}, MagicMock())
+    with patch("superset.is_feature_enabled", return_value=False):
+        assert _is_mcp_guest_auth_enabled(app) is False
+
+
+def test_mcp_guest_auth_enabled_when_flag_and_feature_on() -> None:
+    app = _FakeApp({"MCP_EMBEDDED_GUEST_AUTH_ENABLED": True}, MagicMock())
+    with patch("superset.is_feature_enabled", return_value=True):
+        assert _is_mcp_guest_auth_enabled(app) is True
+
+
+def test_mcp_guest_auth_disabled_when_opt_in_off() -> None:
+    app = _FakeApp({"MCP_EMBEDDED_GUEST_AUTH_ENABLED": False}, MagicMock())
+    assert _is_mcp_guest_auth_enabled(app) is False

tests/unit_tests/mcp_service/test_mcp_server.py

@@ -209,3 +209,32 @@ def test_create_auth_provider_propagates_auth_config_error() -> None:
     ):
         with pytest.raises(MCPAuthConfigError):
             _create_auth_provider(flask_app)
+
+
+def test_create_auth_provider_fails_closed_on_insecure_guest_secret() -> None:
+    """Guest-only deployment with an insecure GUEST_TOKEN_JWT_SECRET must abort.
+
+    When only MCP_EMBEDDED_GUEST_AUTH_ENABLED is on and the default factory
+    raises MCPAuthConfigError (insecure default guest secret), _create_auth_provider
+    must re-raise it — otherwise the server would boot with no authentication.
+    """
+    from superset.mcp_service.mcp_config import MCPAuthConfigError
+    from superset.mcp_service.server import _create_auth_provider
+
+    flask_app = MagicMock()
+    flask_app.config.get.side_effect = lambda key, default=None: {
+        "MCP_AUTH_FACTORY": None,
+        "MCP_AUTH_ENABLED": False,
+        "MCP_API_KEY_ENABLED": False,
+        "FAB_API_KEY_ENABLED": False,
+        "MCP_EMBEDDED_GUEST_AUTH_ENABLED": True,
+    }.get(key, default)
+
+    with patch(
+        "superset.mcp_service.mcp_config.create_default_mcp_auth_factory",
+        side_effect=MCPAuthConfigError(
+            "GUEST_TOKEN_JWT_SECRET is the insecure default"
+        ),
+    ):
+        with pytest.raises(MCPAuthConfigError):
+            _create_auth_provider(flask_app)

tests/unit_tests/models/test_time_filter_hour_offset.py

@@ -0,0 +1,99 @@
+# 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 that the dataset Hour Offset is honored by the time filter (#104810)."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from flask import Flask
+from pytest_mock import MockerFixture
+
+from superset.connectors.sqla.models import SqlaTable, TableColumn
+from superset.models.core import Database
+from superset.superset_typing import QueryObjectDict
+
+
+def _build_dataset(offset: int) -> SqlaTable:
+    database = Database(
+        id=1,
+        database_name="test_db",
+        sqlalchemy_uri="sqlite://",
+    )
+    columns = [
+        TableColumn(column_name="dttm", is_dttm=1, type="TIMESTAMP"),
+        TableColumn(column_name="value", type="INTEGER"),
+    ]
+    return SqlaTable(
+        table_name="test_table",
+        columns=columns,
+        main_dttm_col="dttm",
+        database=database,
+        offset=offset,
+    )
+
+
+def _generated_sql(dataset: SqlaTable, mocker: MockerFixture, app: Flask) -> str:
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.get_guest_rls_filters",
+        return_value=[],
+    )
+    mocker.patch(
+        "superset.connectors.sqla.models.security_manager.is_guest_user",
+        return_value=False,
+    )
+    query_obj: QueryObjectDict = {
+        "granularity": "dttm",
+        "from_dttm": datetime(2024, 1, 1),
+        "to_dttm": datetime(2024, 1, 31),
+        "is_timeseries": False,
+        "filter": [
+            {"col": "dttm", "op": "TEMPORAL_RANGE", "val": "2024-01-01 : 2024-01-31"}
+        ],
+        "metrics": [],
+        "columns": ["value"],
+    }
+    with app.test_request_context():
+        return dataset.get_query_str_extended(query_obj, mutate=False).sql
+
+
+def test_time_filter_without_offset_uses_raw_bounds(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    sql = _generated_sql(_build_dataset(0), mocker, app)
+    # The requested start bound (2024-01-01) is used verbatim; no day shift.
+    assert "2024-01-01" in sql
+    assert "2023-12-31" not in sql
+
+
+def test_time_filter_shifts_bounds_by_dataset_hour_offset(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    # Offset of +4h: displayed values are value + 4h, so the filter bounds must
+    # shift back by 4h. Start 2024-01-01 00:00 -> 2023-12-31 20:00 (#104810).
+    sql = _generated_sql(_build_dataset(4), mocker, app)
+    assert "2023-12-31 20:00:00" in sql
+    # The unshifted start bound must NOT appear as the filter boundary.
+    assert "2024-01-01 00:00:00" not in sql
+
+
+def test_time_filter_negative_offset_shifts_forward(
+    mocker: MockerFixture, app: Flask
+) -> None:
+    # Offset of -4h shifts the start bound forward to 2024-01-01 04:00.
+    sql = _generated_sql(_build_dataset(-4), mocker, app)
+    assert "2024-01-01 04:00:00" in sql