fix(reports): commit permalink before Playwright navigates to tab URL

CreateDashboardPermalinkCommand only flushes the INSERT when invoked inside an outer @transaction() (the nesting guard skips the commit). The row is therefore not visible to Playwright's separate DB connection, causing dashboard_permalink to return a 404, the <body class="standalone"> element to never appear, and the screenshot to time out after 600s. Commit immediately after CreateDashboardPermalinkCommand.run() returns, matching the pattern already used by create_log() in the same class. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
fix(bigquery): set default dataset from schema in adjust_engine_params (#40776 )
2026-06-16 21:19:18 +00:00 · 2026-06-15 23:50:30 +00:00 · 2026-06-15 18:37:06 -04:00 · 2026-06-15 15:22:21 -07:00 · 2026-06-15 14:45:41 -07:00 · 2026-06-15 12:17:14 -07:00
25 changed files with 1878 additions and 16 deletions
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -121,6 +121,20 @@ This change is backward compatible. The feature is off by default, and even when

 Disabling a user account (setting `active` to `False`, via the admin UI, REST API, or CLI) now terminates that user's outstanding sessions on their next request, instead of relying on a passive check. This works for both client-side cookie sessions and server-side session stores via a per-user invalidation epoch (`user_attribute.sessions_invalidated_at`, added by a migration). The mechanism is inert for users that were never disabled (NULL epoch), so there is no behavior change for active users. Re-enabling an account and logging in again starts a fresh, valid session. The migration backfills the epoch for accounts that are already disabled at upgrade time, so re-enabling such an account does not revive a session that predates this feature.

+### Opt-in SSH tunnel server host key verification
+
+SSH tunnels can now optionally pin the expected SSH server host key as a defense-in-depth measure against man-in-the-middle attacks. paramiko's transport performs no known-hosts checking by default, so previously the SSH server's identity was not verified. This feature is opt-in and off by default; existing tunnels are unaffected.
+
+- A new nullable `server_host_key` column on the `ssh_tunnels` table stores the expected host key in authorized-key form (e.g. `ssh-ed25519 AAAA...`). It is a public key and is stored in plaintext. It can be set via the SSH tunnel POST/PUT payloads (`ssh_tunnel.server_host_key`).
+- When a tunnel has `server_host_key` set, Superset connects to the SSH server, reads the host key it presents, and rejects the tunnel if it does not match.
+- A new config flag `SSH_TUNNEL_STRICT_HOST_KEY_CHECKING` (default `False`) controls fail-closed behavior. When `True`, every tunnel must declare a `server_host_key`; a tunnel without one is rejected.
+
+Runbook to adopt:
+
+1. Capture the SSH server's host key, e.g. `ssh-keyscan -t ed25519 ssh.example.com` (verify it out-of-band).
+2. Set that value on the tunnel's `server_host_key` (via the database/SSH tunnel API or UI payload).
+3. Optionally set `SSH_TUNNEL_STRICT_HOST_KEY_CHECKING = True` in `superset_config.py` to require host-key verification on all tunnels.
+
 ### Dataset import validates catalog against the target connection

 Importing a dataset now validates the `catalog` field against the target database connection. When the connection has multi-catalog disabled (`allow_multi_catalog` off) and the dataset's catalog is not the connection's default catalog, the import fails instead of silently persisting the non-default catalog. This matches the validation already enforced on the dataset update path and prevents imported datasets from querying an unintended database.
--- a/superset-frontend/src/SqlLab/actions/sqlLab.test.ts
+++ b/superset-frontend/src/SqlLab/actions/sqlLab.test.ts
@@ -1420,6 +1420,39 @@ describe('async actions', () => {
      });
    });

+    // eslint-disable-next-line no-restricted-globals -- TODO: Migrate from describe blocks
+    describe('syncTable', () => {
+      test('updates the table schema state in the backend', () => {
+        expect.assertions(4);
+
+        const tableName = 'table';
+        const schemaName = 'schema';
+        const store = mockStore(initialState);
+        const expectedActionTypes = [
+          actions.MERGE_TABLE, // syncTable
+        ];
+        const request = actions.syncTable(
+          query as any,
+          tableName as any,
+          schemaName,
+        );
+        return request(store.dispatch, store.getState, undefined).then(() => {
+          expect(store.getActions().map(a => a.type)).toEqual(
+            expectedActionTypes,
+          );
+          expect(store.getActions()[0].prepend).toBeFalsy();
+          expect(
+            fetchMock.callHistory.calls(updateTableSchemaEndpoint),
+          ).toHaveLength(1);
+
+          // tab state is not updated, since no query was run
+          expect(
+            fetchMock.callHistory.calls(updateTabStateEndpoint),
+          ).toHaveLength(0);
+        });
+      });
+    });
+
    // eslint-disable-next-line no-restricted-globals -- TODO: Migrate from describe blocks
    describe('runTablePreviewQuery', () => {
      const results = {
--- a/superset-frontend/src/SqlLab/actions/sqlLab.ts
+++ b/superset-frontend/src/SqlLab/actions/sqlLab.ts
@@ -1346,6 +1346,52 @@ export function runTablePreviewQuery(
  };
 }

+export interface TableMetaData {
+  columns?: unknown[];
+  selectStar?: string;
+  primaryKey?: unknown;
+  foreignKeys?: unknown[];
+  indexes?: unknown[];
+}
+
+export function syncTable(
+  table: Table,
+  tableMetadata: TableMetaData,
+  finalQueryEditorId?: string,
+): SqlLabThunkAction<Promise<unknown>> {
+  return function (dispatch: AppDispatch) {
+    const finalTable = { ...table, queryEditorId: finalQueryEditorId };
+    const sync = isFeatureEnabled(FeatureFlag.SqllabBackendPersistence)
+      ? SupersetClient.post({
+          endpoint: encodeURI('/tableschemaview/'),
+          postPayload: { table: { ...tableMetadata, ...finalTable } },
+        })
+      : Promise.resolve({ json: { id: table.id } });
+
+    return sync
+      .then(({ json: resultJson }) => {
+        const newTable = { ...table, id: `${resultJson.id}` };
+        dispatch(
+          mergeTable({
+            ...newTable,
+            expanded: true,
+            initialized: true,
+          }),
+        );
+      })
+      .catch(() =>
+        dispatch(
+          addDangerToast(
+            t(
+              'An error occurred while fetching table metadata. ' +
+                'Please contact your administrator.',
+            ),
+          ),
+        ),
+      );
+  };
+}
+
 export function changeDataPreviewId(
  oldQueryId: string,
  newQuery: Query,
--- a/superset-frontend/src/SqlLab/components/TableElement/TableElement.test.tsx
+++ b/superset-frontend/src/SqlLab/components/TableElement/TableElement.test.tsx
@@ -0,0 +1,464 @@
+/**
+ * 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 { isValidElement } from 'react';
+import fetchMock from 'fetch-mock';
+import { FeatureFlag, isFeatureEnabled } from '@superset-ui/core';
+import TableElement, { Column } from 'src/SqlLab/components/TableElement';
+import { table, initialState } from 'src/SqlLab/fixtures';
+import { render, waitFor, fireEvent } from 'spec/helpers/testing-library';
+import * as sqlLabActions from 'src/SqlLab/actions/sqlLab';
+import { QueryEditor } from 'src/SqlLab/types';
+
+jest.mock('@superset-ui/core', () => ({
+  ...jest.requireActual('@superset-ui/core'),
+  isFeatureEnabled: jest.fn(),
+}));
+
+const mockedIsFeatureEnabled = isFeatureEnabled as jest.Mock;
+
+jest.mock('@superset-ui/core/components/Loading', () => ({
+  Loading: () => <div data-test="mock-loading" />,
+}));
+jest.mock('@superset-ui/core/components/IconTooltip', () => ({
+  IconTooltip: ({
+    onClick,
+    tooltip,
+  }: {
+    onClick: () => void;
+    tooltip: string;
+  }) => (
+    <button type="button" data-test="mock-icon-tooltip" onClick={onClick}>
+      {tooltip}
+    </button>
+  ),
+}));
+jest.mock(
+  'src/SqlLab/components/ColumnElement',
+  () =>
+    ({ column }: { column: Column }) => (
+      <div data-test="mock-column-element">{column.name}</div>
+    ),
+);
+const getTableMetadataEndpoint =
+  /\/api\/v1\/database\/\d+\/table_metadata\/(?:\?.*)?$/;
+const getExtraTableMetadataEndpoint =
+  /\/api\/v1\/database\/\d+\/table_metadata\/extra\/(?:\?.*)?$/;
+const updateTableSchemaExpandedEndpoint = 'glob:*/tableschemaview/*/expanded';
+const updateTableSchemaEndpoint = 'glob:*/tableschemaview/';
+
+beforeEach(() => {
+  fetchMock.get(getTableMetadataEndpoint, table);
+  fetchMock.get(getExtraTableMetadataEndpoint, {});
+  fetchMock.post(updateTableSchemaExpandedEndpoint, {});
+  fetchMock.post(updateTableSchemaEndpoint, {});
+});
+
+afterEach(() => fetchMock.clearHistory().removeRoutes());
+
+const mockedProps = {
+  table: {
+    ...table,
+    initialized: true,
+  },
+  activeKey: [table.id],
+};
+
+const createStateWithQueryEditor = (queryEditor: Partial<QueryEditor>) => ({
+  ...initialState,
+  sqlLab: {
+    ...initialState.sqlLab,
+    queryEditors: [queryEditor],
+  },
+});
+
+const setupSyncTableTest = () => {
+  const spy = jest.spyOn(sqlLabActions, 'syncTable');
+  mockedIsFeatureEnabled.mockImplementation(
+    featureFlag => featureFlag === FeatureFlag.SqllabBackendPersistence,
+  );
+  fetchMock.removeRoute(updateTableSchemaEndpoint);
+  fetchMock.post(
+    updateTableSchemaEndpoint,
+    { id: 100 },
+    { name: updateTableSchemaEndpoint },
+  );
+  return spy;
+};
+
+test('renders', () => {
+  expect(isValidElement(<TableElement table={table} />)).toBe(true);
+});
+
+test('renders with props', () => {
+  expect(isValidElement(<TableElement {...mockedProps} />)).toBe(true);
+});
+
+test('has 4 IconTooltip elements', async () => {
+  const { getAllByTestId } = render(<TableElement {...mockedProps} />, {
+    useRedux: true,
+    initialState,
+  });
+  await waitFor(() =>
+    expect(getAllByTestId('mock-icon-tooltip')).toHaveLength(6),
+  );
+});
+
+test('has 14 columns', async () => {
+  const { getAllByTestId } = render(<TableElement {...mockedProps} />, {
+    useRedux: true,
+    initialState,
+  });
+  await waitFor(() =>
+    expect(getAllByTestId('mock-column-element')).toHaveLength(14),
+  );
+});
+
+test('fades table', async () => {
+  const { getAllByTestId } = render(<TableElement {...mockedProps} />, {
+    useRedux: true,
+    initialState,
+  });
+  await waitFor(() =>
+    expect(getAllByTestId('mock-icon-tooltip')).toHaveLength(6),
+  );
+  const style = window.getComputedStyle(getAllByTestId('fade')[0]);
+  expect(style.opacity).toBe('0');
+  fireEvent.mouseEnter(getAllByTestId('table-element-header-container')[0]);
+  await waitFor(() =>
+    expect(window.getComputedStyle(getAllByTestId('fade')[0]).opacity).toBe(
+      '1',
+    ),
+  );
+});
+
+test('sorts columns', async () => {
+  const { getAllByTestId, getByText } = render(
+    <TableElement {...mockedProps} />,
+    {
+      useRedux: true,
+      initialState,
+    },
+  );
+  await waitFor(() =>
+    expect(getAllByTestId('mock-icon-tooltip')).toHaveLength(6),
+  );
+  expect(
+    getAllByTestId('mock-column-element').map(el => el.textContent),
+  ).toEqual(table.columns.map(col => col.name));
+  fireEvent.click(getByText('Sort columns alphabetically'));
+  const sorted = table.columns.map(col => col.name).sort();
+  expect(
+    getAllByTestId('mock-column-element').map(el => el.textContent),
+  ).toEqual(sorted);
+  expect(getAllByTestId('mock-column-element')[0]).toHaveTextContent('active');
+});
+
+test('removes the table', async () => {
+  const updateTableSchemaEndpoint = 'glob:*/tableschemaview/*';
+  fetchMock.delete(updateTableSchemaEndpoint, {});
+  mockedIsFeatureEnabled.mockImplementation(
+    featureFlag => featureFlag === FeatureFlag.SqllabBackendPersistence,
+  );
+  const { getAllByTestId, getByText } = render(
+    <TableElement {...mockedProps} />,
+    {
+      useRedux: true,
+      initialState,
+    },
+  );
+  await waitFor(() =>
+    expect(getAllByTestId('mock-icon-tooltip')).toHaveLength(6),
+  );
+  expect(fetchMock.callHistory.calls(updateTableSchemaEndpoint)).toHaveLength(
+    0,
+  );
+  fireEvent.click(getByText('Remove table preview'));
+  await waitFor(() =>
+    expect(fetchMock.callHistory.calls(updateTableSchemaEndpoint)).toHaveLength(
+      1,
+    ),
+  );
+  mockedIsFeatureEnabled.mockClear();
+});
+
+test('fetches table metadata when expanded', async () => {
+  render(<TableElement {...mockedProps} />, {
+    useRedux: true,
+    initialState,
+  });
+  expect(fetchMock.callHistory.calls(getTableMetadataEndpoint)).toHaveLength(0);
+  expect(
+    fetchMock.callHistory.calls(getExtraTableMetadataEndpoint),
+  ).toHaveLength(0);
+  await waitFor(() =>
+    expect(fetchMock.callHistory.calls(getTableMetadataEndpoint)).toHaveLength(
+      1,
+    ),
+  );
+  expect(
+    fetchMock.callHistory.calls(updateTableSchemaExpandedEndpoint),
+  ).toHaveLength(0);
+  expect(
+    fetchMock.callHistory.calls(getExtraTableMetadataEndpoint),
+  ).toHaveLength(1);
+});
+
+test('refreshes table metadata when triggered', async () => {
+  const { getAllByTestId, getByText } = render(
+    <TableElement {...mockedProps} />,
+    {
+      useRedux: true,
+      initialState,
+    },
+  );
+  await waitFor(() =>
+    expect(getAllByTestId('mock-icon-tooltip')).toHaveLength(6),
+  );
+  expect(fetchMock.callHistory.calls(updateTableSchemaEndpoint)).toHaveLength(
+    0,
+  );
+  expect(fetchMock.callHistory.calls(getTableMetadataEndpoint)).toHaveLength(1);
+
+  fireEvent.click(getByText('Refresh table schema'));
+  await waitFor(() =>
+    expect(fetchMock.callHistory.calls(getTableMetadataEndpoint)).toHaveLength(
+      2,
+    ),
+  );
+  await waitFor(() =>
+    expect(fetchMock.callHistory.calls(updateTableSchemaEndpoint)).toHaveLength(
+      1,
+    ),
+  );
+});
+
+test('calls syncTable with valid backend ID when query editor has tabViewId', async () => {
+  const syncTableSpy = setupSyncTableTest();
+  const testTable = {
+    ...table,
+    initialized: false,
+    queryEditorId: 'temp-id-123',
+  };
+
+  const state = createStateWithQueryEditor({
+    id: 'temp-id-123',
+    tabViewId: '42',
+    inLocalStorage: false,
+    name: 'Test Editor',
+  });
+
+  render(<TableElement table={testTable} activeKey={[testTable.id]} />, {
+    useRedux: true,
+    initialState: state,
+  });
+
+  await waitFor(() => {
+    expect(syncTableSpy).toHaveBeenCalledWith(
+      expect.any(Object),
+      expect.any(Object),
+      '42', // finalQueryEditorId
+    );
+  });
+
+  syncTableSpy.mockRestore();
+});
+
+test('does not call syncTable when query editor is in localStorage', async () => {
+  const syncTableSpy = setupSyncTableTest();
+  const testTable = {
+    ...table,
+    initialized: false,
+    queryEditorId: 'local-id',
+  };
+
+  const state = createStateWithQueryEditor({
+    id: 'local-id',
+    tabViewId: undefined,
+    inLocalStorage: true,
+    name: 'Local Editor',
+  });
+
+  render(<TableElement table={testTable} activeKey={[testTable.id]} />, {
+    useRedux: true,
+    initialState: state,
+  });
+
+  await waitFor(() => {
+    expect(fetchMock.callHistory.calls(getTableMetadataEndpoint)).toHaveLength(
+      1,
+    );
+  });
+
+  await new Promise(resolve => setTimeout(resolve, 100));
+  expect(syncTableSpy).not.toHaveBeenCalled();
+
+  syncTableSpy.mockRestore();
+});
+
+test('does not call syncTable with non-numeric queryEditorId', async () => {
+  const syncTableSpy = setupSyncTableTest();
+  const testTable = {
+    ...table,
+    initialized: false,
+    queryEditorId: 'not-a-number',
+  };
+
+  const state = createStateWithQueryEditor({
+    id: 'not-a-number',
+    tabViewId: 'also-not-a-number',
+    inLocalStorage: false,
+    name: 'Invalid Editor',
+  });
+
+  render(<TableElement table={testTable} activeKey={[testTable.id]} />, {
+    useRedux: true,
+    initialState: state,
+  });
+
+  await waitFor(() => {
+    expect(fetchMock.callHistory.calls(getTableMetadataEndpoint)).toHaveLength(
+      1,
+    );
+  });
+
+  await new Promise(resolve => setTimeout(resolve, 100));
+  expect(syncTableSpy).not.toHaveBeenCalled();
+
+  syncTableSpy.mockRestore();
+});
+
+test('does not call syncTable for already initialized tables', async () => {
+  const syncTableSpy = setupSyncTableTest();
+  const testTable = {
+    ...table,
+    initialized: true, // Already initialized
+    queryEditorId: '789',
+  };
+
+  const state = createStateWithQueryEditor({
+    id: '789',
+    tabViewId: '789',
+    inLocalStorage: false,
+    name: 'Initialized Editor',
+  });
+
+  render(<TableElement table={testTable} activeKey={[testTable.id]} />, {
+    useRedux: true,
+    initialState: state,
+  });
+
+  await waitFor(() => {
+    expect(fetchMock.callHistory.calls(getTableMetadataEndpoint)).toHaveLength(
+      1,
+    );
+  });
+
+  await new Promise(resolve => setTimeout(resolve, 100));
+  expect(syncTableSpy).not.toHaveBeenCalled();
+
+  syncTableSpy.mockRestore();
+});
+
+test('calls syncTable after query editor is migrated from localStorage', async () => {
+  const syncTableSpy = setupSyncTableTest();
+  const testTable = {
+    ...table,
+    initialized: false,
+    queryEditorId: 'temp-editor-id',
+  };
+
+  // Start with editor in localStorage
+  const localState = createStateWithQueryEditor({
+    id: 'temp-editor-id',
+    tabViewId: undefined,
+    inLocalStorage: true,
+    name: 'Temp Editor',
+  });
+
+  const { rerender } = render(
+    <TableElement table={testTable} activeKey={[testTable.id]} />,
+    {
+      useRedux: true,
+      initialState: localState,
+    },
+  );
+
+  await new Promise(resolve => setTimeout(resolve, 100));
+  expect(syncTableSpy).not.toHaveBeenCalled();
+
+  const migratedState = createStateWithQueryEditor({
+    id: 'temp-editor-id',
+    tabViewId: '999',
+    inLocalStorage: false,
+    name: 'Temp Editor',
+  });
+
+  rerender(<TableElement table={testTable} activeKey={[testTable.id]} />);
+
+  const { unmount } = render(
+    <TableElement table={testTable} activeKey={[testTable.id]} />,
+    {
+      useRedux: true,
+      initialState: migratedState,
+    },
+  );
+
+  await waitFor(() => {
+    expect(syncTableSpy).toHaveBeenCalledWith(
+      expect.any(Object),
+      expect.any(Object),
+      '999',
+    );
+  });
+
+  unmount();
+  syncTableSpy.mockRestore();
+});
+
+test('passes numeric queryEditorId validation', async () => {
+  const syncTableSpy = setupSyncTableTest();
+  const testTable = {
+    ...table,
+    initialized: false,
+    queryEditorId: 'editor-123',
+  };
+
+  const state = createStateWithQueryEditor({
+    id: 'editor-123',
+    tabViewId: '456',
+    inLocalStorage: false,
+    name: 'Valid Editor',
+  });
+
+  render(<TableElement table={testTable} activeKey={[testTable.id]} />, {
+    useRedux: true,
+    initialState: state,
+  });
+
+  await waitFor(() => {
+    expect(syncTableSpy).toHaveBeenCalled();
+    const [, , finalQueryEditorId] = syncTableSpy.mock.calls[0];
+    // Verify it's a valid numeric string
+    expect(Number.isNaN(Number(finalQueryEditorId))).toBe(false);
+    expect(typeof finalQueryEditorId).toBe('string');
+    expect(finalQueryEditorId).toMatch(/^\d+$/);
+  });
+
+  syncTableSpy.mockRestore();
+});
--- a/superset-frontend/src/SqlLab/components/TableElement/index.tsx
+++ b/superset-frontend/src/SqlLab/components/TableElement/index.tsx
@@ -0,0 +1,420 @@
+/**
+ * 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 { useState, useRef, useEffect } from 'react';
+import { useSelector } from 'react-redux';
+import { useAppDispatch } from 'src/SqlLab/hooks/useAppDispatch';
+import type { QueryEditor, SqlLabRootState, Table } from 'src/SqlLab/types';
+import {
+  ButtonGroup,
+  Card,
+  Collapse,
+  Tooltip,
+  Flex,
+  IconTooltip,
+  Loading,
+  ModalTrigger,
+  type CollapseProps,
+} from '@superset-ui/core/components';
+import { CopyToClipboard } from 'src/components';
+import { t } from '@apache-superset/core/translation';
+import { styled, useTheme } from '@apache-superset/core/theme';
+import { debounce } from 'lodash';
+
+import {
+  removeDataPreview,
+  removeTables,
+  addDangerToast,
+  syncTable,
+} from 'src/SqlLab/actions/sqlLab';
+import {
+  tableApiUtil,
+  useTableExtendedMetadataQuery,
+  useTableMetadataQuery,
+} from 'src/hooks/apiResources';
+import useEffectEvent from 'src/hooks/useEffectEvent';
+import { ActionType } from 'src/types/Action';
+import { Icons } from '@superset-ui/core/components/Icons';
+import { Space } from '@superset-ui/core/components/Space';
+import ColumnElement, { ColumnKeyTypeType } from '../ColumnElement';
+import ShowSQL from '../ShowSQL';
+
+export interface Column {
+  name: string;
+  keys?: { type: ColumnKeyTypeType }[];
+  type: string;
+}
+
+export interface TableElementProps extends CollapseProps {
+  table: Table;
+}
+
+const StyledSpan = styled.span`
+  cursor: pointer;
+`;
+
+const Fade = styled.div`
+  transition: all ${({ theme }) => theme.motionDurationMid};
+  opacity: ${(props: { hovered: boolean }) => (props.hovered ? 1 : 0)};
+`;
+
+const TableElement = ({ table, ...props }: TableElementProps) => {
+  const { dbId, catalog, schema, name, expanded, id } = table;
+  const theme = useTheme();
+  const dispatch = useAppDispatch();
+  const {
+    currentData: tableMetadata,
+    isSuccess: isMetadataSuccess,
+    isFetching: isMetadataFetching,
+    isError: hasMetadataError,
+  } = useTableMetadataQuery(
+    {
+      dbId,
+      catalog,
+      schema,
+      table: name,
+    },
+    { skip: !expanded },
+  );
+  const {
+    currentData: tableExtendedMetadata,
+    isSuccess: isExtraMetadataSuccess,
+    isLoading: isExtraMetadataLoading,
+    isError: hasExtendedMetadataError,
+  } = useTableExtendedMetadataQuery(
+    {
+      dbId,
+      catalog,
+      schema,
+      table: name,
+    },
+    { skip: !expanded },
+  );
+  const tableData = {
+    ...tableMetadata,
+    ...tableExtendedMetadata,
+  };
+  const queryEditors = useSelector<SqlLabRootState, QueryEditor[]>(
+    state => state.sqlLab.queryEditors,
+  );
+  const currentTable = { ...tableData, ...table };
+  const { queryEditorId } = currentTable;
+  const queryEditor = queryEditors.find(
+    qe => qe.id === queryEditorId || qe.tabViewId === queryEditorId,
+  );
+  const currentQueryEditorId = queryEditor?.tabViewId || queryEditorId;
+
+  useEffect(() => {
+    if (hasMetadataError || hasExtendedMetadataError) {
+      dispatch(
+        addDangerToast(t('An error occurred while fetching table metadata')),
+      );
+    }
+  }, [hasMetadataError, hasExtendedMetadataError, dispatch]);
+
+  // TODO: migrate syncTable logic by SIP-93
+  const syncTableMetadata = useEffectEvent(() => {
+    const { initialized } = table;
+    // if not a valid number, wait for backend to assign one
+    const hasFinalQueryEditorId =
+      currentQueryEditorId &&
+      !Number.isNaN(Number(currentQueryEditorId)) &&
+      currentTable.queryEditorId !== currentQueryEditorId;
+    if (!initialized && hasFinalQueryEditorId) {
+      dispatch(syncTable(currentTable, tableData, currentQueryEditorId));
+    }
+  });
+
+  useEffect(() => {
+    if (isMetadataSuccess && isExtraMetadataSuccess) {
+      syncTableMetadata();
+    }
+  }, [
+    isMetadataSuccess,
+    isExtraMetadataSuccess,
+    currentQueryEditorId,
+    syncTableMetadata,
+  ]);
+
+  const [sortColumns, setSortColumns] = useState(false);
+  const [hovered, setHovered] = useState(false);
+  const tableNameRef = useRef<HTMLInputElement>(null);
+
+  const setHover = (hovered: boolean) => {
+    debounce(() => setHovered(hovered), 100)();
+  };
+
+  const removeTable = () => {
+    dispatch(removeDataPreview(table));
+    dispatch(removeTables([table]));
+  };
+
+  const toggleSortColumns = () => {
+    setSortColumns(prevState => !prevState);
+  };
+
+  const refreshTableMetadata = () => {
+    dispatch(
+      tableApiUtil.invalidateTags([{ type: 'TableMetadatas', id: name }]),
+    );
+    dispatch(syncTable(table, tableData, table.queryEditorId));
+  };
+
+  const renderWell = () => {
+    let partitions;
+    let metadata;
+    if (tableData.partitions) {
+      let partitionQuery;
+      let partitionClipBoard;
+      if (tableData.partitions.partitionQuery) {
+        ({ partitionQuery } = tableData.partitions);
+        const tt = t('Copy partition query to clipboard');
+        partitionClipBoard = (
+          <CopyToClipboard
+            text={partitionQuery}
+            shouldShowText={false}
+            tooltipText={tt}
+            copyNode={<Icons.CopyOutlined iconSize="s" />}
+          />
+        );
+      }
+      const latest = Object.entries(tableData.partitions?.latest || [])
+        .map(([key, value]) => `${key}=${value}`)
+        .join('/');
+
+      partitions = (
+        <div>
+          <small>
+            {t('latest partition:')} {latest}
+          </small>{' '}
+          {partitionClipBoard}
+        </div>
+      );
+    }
+
+    if (tableData.metadata) {
+      metadata = Object.entries(tableData.metadata).map(([key, value]) => (
+        <div>
+          <small>
+            <strong>{key}:</strong> {value}
+          </small>
+        </div>
+      ));
+      if (!metadata?.length) {
+        // hide metadata card view
+        return null;
+      }
+    }
+
+    if (!partitions) {
+      // hide partition card view
+      return null;
+    }
+
+    return (
+      <Card size="small">
+        {partitions}
+        {metadata}
+      </Card>
+    );
+  };
+
+  const renderControls = () => {
+    let keyLink;
+    const KEYS_FOR_TABLE_TEXT = t('Keys for table');
+    if (tableData?.indexes?.length) {
+      keyLink = (
+        <ModalTrigger
+          modalTitle={`${KEYS_FOR_TABLE_TEXT} ${name}`}
+          modalBody={tableData.indexes.map((ix, i) => (
+            <pre key={i}>{JSON.stringify(ix, null, '  ')}</pre>
+          ))}
+          triggerNode={
+            <IconTooltip
+              className="pull-left"
+              tooltip={t('View keys & indexes (%s)', tableData.indexes.length)}
+            >
+              <Icons.TableOutlined
+                iconSize="m"
+                iconColor={theme.colorPrimary}
+              />
+            </IconTooltip>
+          }
+        />
+      );
+    }
+    return (
+      <Flex style={{ height: 22 }} align="center">
+        {isMetadataFetching || isExtraMetadataLoading ? (
+          <Loading position="inline" />
+        ) : (
+          <Fade
+            data-test="fade"
+            hovered={hovered}
+            onClick={e => e.stopPropagation()}
+          >
+            <ButtonGroup>
+              <Space size="small">
+                <IconTooltip
+                  className="pull-left pointer"
+                  onClick={refreshTableMetadata}
+                  tooltip={t('Refresh table schema')}
+                >
+                  <Icons.SyncOutlined
+                    iconSize="m"
+                    iconColor={theme.colorIcon}
+                  />
+                </IconTooltip>
+                {keyLink}
+                <IconTooltip
+                  onClick={toggleSortColumns}
+                  tooltip={
+                    sortColumns
+                      ? t('Original table column order')
+                      : t('Sort columns alphabetically')
+                  }
+                >
+                  <Icons.SortAscendingOutlined
+                    iconSize="m"
+                    aria-hidden
+                    iconColor={
+                      sortColumns ? theme.colorIcon : theme.colorTextDisabled
+                    }
+                  />
+                </IconTooltip>
+                {tableData.selectStar && (
+                  <CopyToClipboard
+                    copyNode={
+                      <IconTooltip
+                        aria-label={t('Copy')}
+                        tooltip={t('Copy SELECT statement to the clipboard')}
+                      >
+                        <Icons.CopyOutlined
+                          iconSize="m"
+                          aria-hidden
+                          iconColor={theme.colorIcon}
+                        />
+                      </IconTooltip>
+                    }
+                    text={tableData.selectStar}
+                    shouldShowText={false}
+                  />
+                )}
+                {tableData.view && (
+                  <ShowSQL
+                    sql={tableData.view}
+                    tooltipText={t('Show CREATE VIEW statement')}
+                    title={t('CREATE VIEW statement')}
+                  />
+                )}
+                <IconTooltip
+                  className=" table-remove pull-left pointer"
+                  onClick={removeTable}
+                  tooltip={t('Remove table preview')}
+                >
+                  <Icons.CloseOutlined
+                    iconSize="m"
+                    aria-hidden
+                    iconColor={theme.colorIcon}
+                  />
+                </IconTooltip>
+              </Space>
+            </ButtonGroup>
+          </Fade>
+        )}
+      </Flex>
+    );
+  };
+
+  const renderHeader = () => {
+    const element: HTMLInputElement | null = tableNameRef.current;
+    let trigger = [] as ActionType[];
+    if (element && element.offsetWidth < element.scrollWidth) {
+      trigger = ['hover'];
+    }
+
+    return (
+      <div
+        data-test="table-element-header-container"
+        className="clearfix header-container"
+      >
+        <Tooltip
+          id="copy-to-clipboard-tooltip"
+          style={{ cursor: 'pointer' }}
+          title={name}
+          trigger={trigger}
+        >
+          <StyledSpan
+            data-test="collapse"
+            ref={tableNameRef}
+            className="table-name"
+          >
+            <strong>{name}</strong>
+          </StyledSpan>
+        </Tooltip>
+      </div>
+    );
+  };
+
+  const renderBody = () => {
+    let cols;
+    if (tableData.columns) {
+      cols = tableData.columns.slice();
+      if (sortColumns) {
+        cols.sort((a: Column, b: Column) => {
+          const colA = a.name.toUpperCase();
+          const colB = b.name.toUpperCase();
+          return colA < colB ? -1 : colA > colB ? 1 : 0;
+        });
+      }
+    }
+
+    const metadata = (
+      <div data-test="table-element" css={{ paddingTop: 6 }}>
+        {renderWell()}
+        <div>
+          {cols?.map(col => (
+            <ColumnElement column={col} key={col.name} />
+          ))}
+        </div>
+      </div>
+    );
+    return metadata;
+  };
+
+  return (
+    <Collapse
+      activeKey={props.activeKey}
+      expandIconPosition="end"
+      onChange={props.onChange}
+      ghost
+      items={[
+        {
+          key: id,
+          label: renderHeader(),
+          children: renderBody(),
+          extra: renderControls(),
+          onMouseEnter: () => setHover(true),
+          onMouseLeave: () => setHover(false),
+        },
+      ]}
+    />
+  );
+};
+
+export default TableElement;
--- a/superset/charts/schemas.py
+++ b/superset/charts/schemas.py
@@ -1454,6 +1454,21 @@ class ChartDataQueryObjectSchema(Schema):
        allow_none=True,
    )

+    @post_load
+    def rename_deprecated_fields(
+        self, data: dict[str, Any], **kwargs: Any
+    ) -> dict[str, Any]:
+        _renames = (
+            ("groupby", "columns"),
+            ("granularity_sqla", "granularity"),
+            ("timeseries_limit", "series_limit"),
+            ("timeseries_limit_metric", "series_limit_metric"),
+        )
+        for old, new in _renames:
+            if value := data.pop(old, None):
+                data[new] = value
+        return data
+

 class ChartDataQueryContextSchema(Schema):
    query_context_factory: QueryContextFactory | None = None
--- a/superset/commands/database/create.py
+++ b/superset/commands/database/create.py
@@ -33,6 +33,7 @@ from superset.commands.database.exceptions import (
 from superset.commands.database.ssh_tunnel.exceptions import (
    SSHTunnelCreateFailedError,
    SSHTunnelDatabasePortError,
+    SSHTunnelHostKeyVerificationError,
    SSHTunnelingNotEnabledError,
    SSHTunnelInvalidError,
 )
@@ -75,6 +76,7 @@ class CreateDatabaseCommand(BaseCommand):
            SupersetErrorsException,
            SSHTunnelingNotEnabledError,
            SSHTunnelDatabasePortError,
+            SSHTunnelHostKeyVerificationError,
        ) as ex:
            event_logger.log_with_context(
                action=f"db_creation_failed.{ex.__class__.__name__}",
--- a/superset/commands/database/ssh_tunnel/exceptions.py
+++ b/superset/commands/database/ssh_tunnel/exceptions.py
@@ -75,3 +75,11 @@ class SSHTunnelMissingCredentials(CommandInvalidError, SSHTunnelError):  # noqa:

 class SSHTunnelInvalidCredentials(CommandInvalidError, SSHTunnelError):  # noqa: N818
    message = _("Cannot have multiple credentials for the SSH Tunnel")
+
+
+class SSHTunnelHostKeyVerificationError(CommandInvalidError, SSHTunnelError):
+    """The SSH server's host key failed opt-in verification for a tunnel."""
+
+    message = _(
+        "The SSH server host key could not be verified against the expected key."
+    )
--- a/superset/commands/database/test_connection.py
+++ b/superset/commands/database/test_connection.py
@@ -29,6 +29,7 @@ from superset.commands.database.exceptions import (
 )
 from superset.commands.database.ssh_tunnel.exceptions import (
    SSHTunnelDatabasePortError,
+    SSHTunnelHostKeyVerificationError,
    SSHTunnelingNotEnabledError,
 )
 from superset.commands.database.utils import ping
@@ -221,7 +222,11 @@ class TestConnectionDatabaseCommand(BaseCommand):
                engine=engine_name,
            )
            raise DatabaseSecurityUnsafeError(message=str(ex)) from ex
-        except (SupersetTimeoutException, SSHTunnelingNotEnabledError) as ex:
+        except (
+            SupersetTimeoutException,
+            SSHTunnelingNotEnabledError,
+            SSHTunnelHostKeyVerificationError,
+        ) as ex:
            event_logger.log_with_context(
                action=get_log_connection_action(
                    "test_connection_error",
@@ -230,7 +235,8 @@ class TestConnectionDatabaseCommand(BaseCommand):
                ),
                engine=engine_name,
            )
-            # bubble up the exception to return proper status code
+            # bubble up the exception (preserving its specific message and status)
+            # instead of flattening it into a generic connection failure
            raise
        except Exception as ex:
            if not database:
--- a/superset/commands/report/execute.py
+++ b/superset/commands/report/execute.py
@@ -358,6 +358,11 @@ class BaseReportState:
            dashboard_id=str(self._report_schedule.dashboard.uuid),
            state=dashboard_state,
        ).run()
+        # Commit the permalink immediately so Playwright's separate DB connection
+        # can resolve the URL. CreateDashboardPermalinkCommand only flushes when
+        # called inside an outer @transaction(), leaving the row invisible to
+        # other connections until we explicitly commit here.
+        db.session.commit()  # pylint: disable=consider-using-transaction

        return get_url_path(
            "Superset.dashboard_permalink",
--- a/superset/config.py
+++ b/superset/config.py
@@ -895,6 +895,15 @@ SSH_TUNNEL_TIMEOUT_SEC = 10.0
 #: Timeout (seconds) for transport socket (``socket.settimeout``)
 SSH_TUNNEL_PACKET_TIMEOUT_SEC = 1.0

+#: Opt-in defense-in-depth: when enabled, every SSH tunnel must declare an expected
+#: server host key (``server_host_key`` on the tunnel) and the SSH server's presented
+#: host key is verified against it before the tunnel is opened. A mismatch, or a
+#: missing expected key while this flag is enabled, fails closed and the tunnel is
+#: rejected. When disabled (the default), tunnels without a ``server_host_key`` open
+#: without host-key verification, preserving existing behavior; tunnels that do set a
+#: ``server_host_key`` are still verified regardless of this flag.
+SSH_TUNNEL_STRICT_HOST_KEY_CHECKING: bool = False
+

 # Feature flags may also be set via 'SUPERSET_FEATURE_' prefixed environment vars.
 DEFAULT_FEATURE_FLAGS.update(
--- a/superset/databases/api.py
+++ b/superset/databases/api.py
@@ -56,6 +56,7 @@ from superset.commands.database.importers.dispatcher import ImportDatabasesComma
 from superset.commands.database.oauth2 import OAuth2StoreTokenCommand
 from superset.commands.database.ssh_tunnel.exceptions import (
    SSHTunnelDatabasePortError,
+    SSHTunnelHostKeyVerificationError,
    SSHTunnelingNotEnabledError,
 )
 from superset.commands.database.sync_permissions import SyncPermissionsCommand
@@ -484,7 +485,11 @@ class DatabaseRestApi(BaseSupersetModelRestApi):
                exc_info=True,
            )
            return self.response_422(message=str(ex))
-        except (SSHTunnelingNotEnabledError, SSHTunnelDatabasePortError) as ex:
+        except (
+            SSHTunnelingNotEnabledError,
+            SSHTunnelDatabasePortError,
+            SSHTunnelHostKeyVerificationError,
+        ) as ex:
            return self.response_400(message=str(ex))
        except SupersetException as ex:
            return self.response(ex.status, message=ex.message)
@@ -569,7 +574,11 @@ class DatabaseRestApi(BaseSupersetModelRestApi):
                exc_info=True,
            )
            return self.response_422(message=str(ex))
-        except (SSHTunnelingNotEnabledError, SSHTunnelDatabasePortError) as ex:
+        except (
+            SSHTunnelingNotEnabledError,
+            SSHTunnelDatabasePortError,
+            SSHTunnelHostKeyVerificationError,
+        ) as ex:
            return self.response_400(message=str(ex))

    @expose("/<int:pk>", methods=("DELETE",))
@@ -1291,7 +1300,11 @@ class DatabaseRestApi(BaseSupersetModelRestApi):
        try:
            TestConnectionDatabaseCommand(item).run()
            return self.response(200, message="OK")
-        except (SSHTunnelingNotEnabledError, SSHTunnelDatabasePortError) as ex:
+        except (
+            SSHTunnelingNotEnabledError,
+            SSHTunnelDatabasePortError,
+            SSHTunnelHostKeyVerificationError,
+        ) as ex:
            return self.response_400(message=str(ex))

    @expose("/<int:pk>/related_objects/", methods=("GET",))
--- a/superset/databases/schemas.py
+++ b/superset/databases/schemas.py
@@ -477,6 +477,22 @@ class DatabaseSSHTunnel(Schema):
    private_key = fields.String(required=False, load_only=True)
    private_key_password = fields.String(required=False, load_only=True)

+    # Optional expected SSH server host key in authorized-key form
+    # (e.g. "ssh-rsa AAAA...", "ssh-ed25519 AAAA..."). When set, the SSH server's
+    # presented host key is verified against it before the tunnel is opened. This is
+    # a public key, so it is not sensitive and is not masked.
+    server_host_key = fields.String(
+        required=False,
+        allow_none=True,
+        metadata={
+            "description": (
+                "Expected SSH server host key in authorized-key form "
+                "(e.g. 'ssh-ed25519 AAAA...'). When set, the server's host key is "
+                "verified against it before the tunnel is opened."
+            )
+        },
+    )
+
    @validates_schema
    def validate_authentication(self, data: dict[str, Any], **kwargs: Any) -> None:
        errors: dict[str, str] = {}
--- a/superset/databases/ssh_tunnel/models.py
+++ b/superset/databases/ssh_tunnel/models.py
@@ -72,6 +72,12 @@ class SSHTunnel(AuditMixinNullable, ExtraJSONMixin, ImportExportMixin, Model):
        encrypted_field_factory.create(Text), nullable=True
    )

+    # Optional expected SSH server host key, in authorized-key form
+    # (e.g. "ssh-rsa AAAA...", "ssh-ed25519 AAAA..."). When set, the SSH server's
+    # presented host key is verified against this value before the tunnel is opened.
+    # This is a public key, so it is stored in plaintext (not encrypted).
+    server_host_key = sa.Column(sa.Text, nullable=True)
+
    export_fields = [
        "server_address",
        "server_port",
@@ -79,6 +85,7 @@ class SSHTunnel(AuditMixinNullable, ExtraJSONMixin, ImportExportMixin, Model):
        "password",
        "private_key",
        "private_key_password",
+        "server_host_key",
    ]

    extra_import_fields = [
@@ -93,6 +100,9 @@ class SSHTunnel(AuditMixinNullable, ExtraJSONMixin, ImportExportMixin, Model):
            "server_port": self.server_port,
            "username": self.username,
        }
+        if self.server_host_key is not None:
+            # public key, not sensitive: returned in cleartext
+            output["server_host_key"] = self.server_host_key
        if self.password is not None:
            output["password"] = PASSWORD_MASK
        if self.private_key is not None:
--- a/superset/db_engine_specs/bigquery.py
+++ b/superset/db_engine_specs/bigquery.py
@@ -195,6 +195,7 @@ class BigQueryEngineSpec(BaseEngineSpec):  # pylint: disable=too-many-public-met
    allows_hidden_cc_in_orderby = True

    supports_catalog = supports_dynamic_catalog = supports_cross_catalog_queries = True
+    supports_dynamic_schema = True

    # when editing the database, mask this field in `encrypted_extra`
    # pylint: disable=invalid-name
@@ -740,11 +741,41 @@ class BigQueryEngineSpec(BaseEngineSpec):  # pylint: disable=too-many-public-met
        catalog: str | None = None,
        schema: str | None = None,
    ) -> tuple[URL, dict[str, Any]]:
-        if catalog:
-            uri = uri.set(host=catalog, database="")
+        if not uri.host:
+            # Triple-slash form (e.g., bigquery:///project): project is in database.
+            default_catalog = uri.database
+            default_schema = None
+        else:
+            # Standard forms: bigquery://project, bigquery://project/dataset
+            default_catalog = uri.host
+            default_schema = uri.database or None  # coerce empty string to None
+
+        uri = uri.set(
+            host=catalog or default_catalog,
+            database=schema or default_schema,
+        )

        return uri, connect_args

+    @classmethod
+    def get_schema_from_engine_params(
+        cls,
+        sqlalchemy_uri: URL,
+        connect_args: dict[str, Any],
+    ) -> str | None:
+        """
+        Return the default dataset encoded in a ``bigquery://project/dataset`` URI.
+
+        The BigQuery SQLAlchemy driver uses the URL ``database`` component as the
+        default dataset, but only when ``host`` (the project) is also present.
+        The triple-slash form ``bigquery:///project`` puts the project in
+        ``database`` with no host, so we guard against misidentifying it as a
+        dataset.
+        """
+        if sqlalchemy_uri.host and sqlalchemy_uri.database:
+            return sqlalchemy_uri.database
+        return None
+
    @classmethod
    def get_allow_cost_estimate(cls, extra: dict[str, Any]) -> bool:
        return True
--- a/superset/extensions/ssh.py
+++ b/superset/extensions/ssh.py
@@ -15,26 +15,63 @@
 # specific language governing permissions and limitations
 # under the License.

+import base64
+import binascii
 import logging
+import socket
 from io import StringIO
 from typing import TYPE_CHECKING

+import paramiko
 import sshtunnel
 from flask import Flask
 from paramiko import RSAKey
+from paramiko.pkey import UnknownKeyType

-from superset.commands.database.ssh_tunnel.exceptions import SSHTunnelDatabasePortError
+from superset.commands.database.ssh_tunnel.exceptions import (
+    SSHTunnelDatabasePortError,
+    SSHTunnelHostKeyVerificationError,
+)
 from superset.databases.utils import make_url_safe
 from superset.utils.class_utils import load_class_from_name

 if TYPE_CHECKING:
    from superset.databases.ssh_tunnel.models import SSHTunnel

+logger = logging.getLogger(__name__)
+
+
+def _parse_authorized_key(authorized_key: str) -> paramiko.PKey:
+    """
+    Parse a host key in authorized-key form (``"<type> <base64>[ comment]"``) into a
+    :class:`paramiko.PKey`. The optional trailing comment field and surrounding
+    whitespace are ignored.
+
+    :raises ValueError: if the value is empty or cannot be parsed as a host key.
+    """
+    fields = authorized_key.strip().split()
+    if len(fields) < 2:
+        raise ValueError("Host key must be in 'ssh-<type> <base64>' form")
+    key_type, key_b64 = fields[0], fields[1]
+    try:
+        # validate=True so malformed characters raise instead of being silently
+        # dropped, which could otherwise pin an unintended key value.
+        key_bytes = base64.b64decode(key_b64, validate=True)
+    except (binascii.Error, ValueError) as ex:
+        raise ValueError("Host key base64 payload could not be decoded") from ex
+    try:
+        return paramiko.PKey.from_type_string(key_type, key_bytes)
+    except (paramiko.SSHException, UnknownKeyType) as ex:
+        raise ValueError(f"Host key could not be parsed: {ex}") from ex
+

 class SSHManager:
    def __init__(self, app: Flask) -> None:
        super().__init__()
        self.local_bind_address = app.config["SSH_TUNNEL_LOCAL_BIND_ADDRESS"]
+        self.strict_host_key_checking = app.config.get(
+            "SSH_TUNNEL_STRICT_HOST_KEY_CHECKING", False
+        )
        sshtunnel.TUNNEL_TIMEOUT = app.config["SSH_TUNNEL_TIMEOUT_SEC"]
        sshtunnel.SSH_TIMEOUT = app.config["SSH_TUNNEL_PACKET_TIMEOUT_SEC"]

@@ -48,6 +85,87 @@ class SSHManager:
            port=server.local_bind_port,
        )

+    def _verify_host_key(self, ssh_tunnel: "SSHTunnel") -> "paramiko.PKey | None":
+        """
+        Opt-in defense-in-depth: verify the SSH server's host key before opening the
+        tunnel, to resist man-in-the-middle attacks (paramiko's ``Transport`` does no
+        known-hosts checking by default).
+
+        Behavior:
+
+        - If the tunnel declares an expected ``server_host_key``, connect to the SSH
+          server, read the host key it presents, and compare. On mismatch (or if the
+          expected key cannot be parsed) raise
+          :class:`SSHTunnelHostKeyVerificationError`.
+        - If no expected key is set and ``SSH_TUNNEL_STRICT_HOST_KEY_CHECKING`` is
+          enabled, fail closed and raise.
+        - If no expected key is set and strict checking is disabled, do nothing,
+          preserving existing (unverified) behavior.
+
+        :returns: the parsed expected host key when one is configured (so the caller
+            can pin it on the tunnel's own connection), or ``None`` when no key is
+            configured.
+        """
+        expected_raw = ssh_tunnel.server_host_key
+
+        if not expected_raw or not expected_raw.strip():
+            if self.strict_host_key_checking:
+                raise SSHTunnelHostKeyVerificationError(
+                    message=(
+                        "SSH_TUNNEL_STRICT_HOST_KEY_CHECKING is enabled but no "
+                        "expected server host key is configured for this tunnel."
+                    )
+                )
+            return None
+
+        try:
+            expected_key = _parse_authorized_key(expected_raw)
+        except ValueError as ex:
+            raise SSHTunnelHostKeyVerificationError(
+                message=f"The configured expected server host key is invalid: {ex}"
+            ) from ex
+
+        # Build the socket ourselves with an explicit timeout so the TCP connect
+        # phase is bounded too. ``paramiko.Transport((host, port))`` would connect
+        # synchronously with no timeout, leaving ``start_client(timeout=...)`` to
+        # govern only the SSH handshake; an unreachable host could then block for the
+        # full OS-level TCP timeout.
+        try:
+            sock = socket.create_connection(
+                (ssh_tunnel.server_address, ssh_tunnel.server_port),
+                timeout=sshtunnel.SSH_TIMEOUT,
+            )
+        except OSError as ex:
+            raise SSHTunnelHostKeyVerificationError(
+                message=f"Could not connect to the SSH server: {ex}"
+            ) from ex
+
+        transport = paramiko.Transport(sock)
+        try:
+            transport.start_client(timeout=sshtunnel.SSH_TIMEOUT)
+            remote_key = transport.get_remote_server_key()
+        except Exception as ex:  # noqa: BLE001
+            raise SSHTunnelHostKeyVerificationError(
+                message=f"Could not retrieve the SSH server host key: {ex}"
+            ) from ex
+        finally:
+            transport.close()
+
+        if remote_key != expected_key:
+            logger.warning(
+                "SSH host key mismatch for %s:%s",
+                ssh_tunnel.server_address,
+                ssh_tunnel.server_port,
+            )
+            raise SSHTunnelHostKeyVerificationError(
+                message=(
+                    "The SSH server presented a host key that does not match the "
+                    "expected server host key configured for this tunnel."
+                )
+            )
+
+        return expected_key
+
    def create_tunnel(
        self,
        ssh_tunnel: "SSHTunnel",
@@ -60,6 +178,12 @@ class SSHManager:
        port = url.port or get_default_port(backend)
        if not port:
            raise SSHTunnelDatabasePortError()
+
+        # Opt-in host-key verification runs before the tunnel is opened. It returns
+        # the parsed expected key (or None) so we can also pin it on the tunnel's own
+        # connection below.
+        expected_host_key = self._verify_host_key(ssh_tunnel)
+
        params = {
            "ssh_address_or_host": (ssh_tunnel.server_address, ssh_tunnel.server_port),
            "ssh_username": ssh_tunnel.username,
@@ -68,6 +192,14 @@ class SSHManager:
            "debug_level": logging.getLogger("flask_appbuilder").level,
        }

+        if expected_host_key is not None:
+            # Pin the expected key on the tunnel's own connection, so paramiko verifies
+            # the host that actually carries traffic on the same transport. The probe
+            # above and the tunnel open separate connections, so verifying only the
+            # probe would leave a TOCTOU gap (DNS re-resolution, selective
+            # interception); pinning here closes it.
+            params["ssh_host_key"] = expected_host_key
+
        if ssh_tunnel.password:
            params["ssh_password"] = ssh_tunnel.password
        elif ssh_tunnel.private_key:
--- a/superset/migrations/shared/catalogs.py
+++ b/superset/migrations/shared/catalogs.py
@@ -25,7 +25,7 @@ import sqlalchemy as sa
 from alembic import op
 from flask import current_app
 from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import lazyload, Session

 from superset import db, security_manager
 from superset.db_engine_specs.base import GenericDBException
@@ -379,7 +379,15 @@ def upgrade_catalog_perms(engines: set[str] | None = None) -> None:
    bind = op.get_bind()
    session = db.Session(bind=bind)

-    for database in session.query(Database).all():
+    # The Database model has an eager-loaded (``lazy="joined"``) ``ssh_tunnel``
+    # backref. Eager-loading it here would SELECT every column on ``ssh_tunnels``,
+    # including columns added by later migrations that do not yet exist at the
+    # revision this helper runs in (e.g. on a fresh DB upgraded in one pass). The
+    # catalog upgrade only needs scalar ``Database`` columns, so disable the eager
+    # join to keep the query schema-safe across migration revisions.
+    for database in (
+        session.query(Database).options(lazyload(Database.ssh_tunnel)).all()
+    ):
        db_engine_spec = database.db_engine_spec
        if (
            engines and db_engine_spec.engine not in engines
@@ -576,7 +584,11 @@ def downgrade_catalog_perms(engines: set[str] | None = None) -> None:
    bind = op.get_bind()
    session = db.Session(bind=bind)

-    for database in session.query(Database).all():
+    # See upgrade_catalog_perms: avoid eager-loading the ``ssh_tunnel`` backref so the
+    # query stays schema-safe across migration revisions.
+    for database in (
+        session.query(Database).options(lazyload(Database.ssh_tunnel)).all()
+    ):
        db_engine_spec = database.db_engine_spec
        if (
            engines and db_engine_spec.engine not in engines
--- a/superset/migrations/versions/2026-06-03_10-00_78a40c08b4be_add_server_host_key_to_ssh_tunnels.py
+++ b/superset/migrations/versions/2026-06-03_10-00_78a40c08b4be_add_server_host_key_to_ssh_tunnels.py
@@ -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.
+"""add server_host_key to ssh_tunnels
+
+Adds a nullable ``server_host_key`` column to the ``ssh_tunnels`` table. It stores the
+expected SSH server host key in authorized-key form (e.g. "ssh-ed25519 AAAA...") so
+operators can opt in to verifying the SSH server's host key before a tunnel is opened.
+This is a public key and is stored in plaintext (not encrypted). The column is
+nullable, so existing tunnels are unaffected.
+
+Revision ID: 78a40c08b4be
+Revises: b7c9d1e2f3a4
+Create Date: 2026-06-03 10:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+
+from superset.migrations.shared.utils import add_columns, drop_columns
+
+# revision identifiers, used by Alembic.
+revision = "78a40c08b4be"
+down_revision = "b7c9d1e2f3a4"
+
+
+def upgrade() -> None:
+    """Add the nullable ``server_host_key`` column to ``ssh_tunnels``."""
+    add_columns(
+        "ssh_tunnels",
+        sa.Column("server_host_key", sa.Text(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    """Drop the ``server_host_key`` column from ``ssh_tunnels``."""
+    drop_columns("ssh_tunnels", "server_host_key")
--- a/superset/security/manager.py
+++ b/superset/security/manager.py
@@ -342,11 +342,43 @@ PermissionModelView.include_route_methods = {RouteMethod.LIST}
 ViewMenuModelView.include_route_methods = {RouteMethod.LIST}


+# Keys on an adhoc column/metric that a guest may legitimately change through a
+# supported native filter, and which therefore must not count as payload
+# tampering. The time grain of a temporal x-axis is baked into its `BASE_AXIS`
+# column by `normalizeTimeColumn` on the frontend (it copies
+# `extras.time_grain_sqla` onto the column), so a Time Grain filter alters the
+# column payload without changing which data is queried.
+GUEST_OVERRIDABLE_VALUE_KEYS = frozenset({"timeGrain"})
+
+
+def _strip_overridable_keys(value: Any) -> Any:
+    """
+    Recursively drop guest-overridable keys from a value.
+
+    Adhoc columns/metrics can be nested inside sequences (e.g. an ``orderby``
+    entry is a ``(column, bool)`` tuple), so the overridable keys must be
+    stripped at every level rather than only from a top-level dict.
+    """
+    if isinstance(value, dict):
+        return {
+            key: _strip_overridable_keys(val)
+            for key, val in value.items()
+            if key not in GUEST_OVERRIDABLE_VALUE_KEYS
+        }
+    if isinstance(value, (list, tuple)):
+        return [_strip_overridable_keys(item) for item in value]
+    return value
+
+
 def freeze_value(value: Any) -> str:
    """
    Used to compare column and metric sets.
+
+    Guest-overridable keys (e.g. the time grain baked into a temporal x-axis
+    column) are dropped so that legitimate native-filter changes don't read as
+    payload tampering.
    """
-    return json.dumps(value, sort_keys=True)
+    return json.dumps(_strip_overridable_keys(value), sort_keys=True)


 def _native_filter_allowed_targets(
--- a/tests/integration_tests/db_engine_specs/bigquery_tests.py
+++ b/tests/integration_tests/db_engine_specs/bigquery_tests.py
@@ -305,6 +305,13 @@ class TestBigQueryDbEngineSpec(SupersetTestCase):

    @mock.patch("superset.models.core.Database.db_engine_spec", BigQueryEngineSpec)
    @mock.patch("sqlalchemy_bigquery._helpers.create_bigquery_client", mock.Mock)
+    @mock.patch(
+        "superset.db_engine_specs.bigquery.BigQueryEngineSpec.adjust_engine_params",
+        new=lambda uri, connect_args, catalog=None, schema=None, **kw: (
+            uri,
+            connect_args,
+        ),
+    )
    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_calculated_column_in_order_by(self):
        table = self.get_table(name="birth_names")
--- a/tests/unit_tests/charts/test_schemas.py
+++ b/tests/unit_tests/charts/test_schemas.py
@@ -213,6 +213,53 @@ def test_chart_data_query_object_schema_time_grain_sqla_validation(
    assert result["extras"]["time_grain_sqla"] is None


+def test_chart_data_query_object_schema_deprecated_fields_renamed(
+    app_context: None,
+) -> None:
+    """Deprecated query object fields are renamed to their canonical names."""
+    schema = ChartDataQueryObjectSchema()
+
+    # groupby alone → becomes columns
+    result = schema.load({"groupby": ["country_name"]})
+    assert result.get("columns") == ["country_name"]
+    assert "groupby" not in result
+
+    # groupby overwrites columns when both are provided
+    result = schema.load({"groupby": ["region"], "columns": ["country_name"]})
+    assert result.get("columns") == ["region"]
+    assert "groupby" not in result
+
+    # empty groupby is discarded; existing columns is preserved
+    result = schema.load({"groupby": [], "columns": ["country_name"]})
+    assert result.get("columns") == ["country_name"]
+    assert "groupby" not in result
+
+    # null groupby is discarded; existing columns is preserved (allow_none=True)
+    result = schema.load({"groupby": None, "columns": ["country_name"]})
+    assert result.get("columns") == ["country_name"]
+    assert "groupby" not in result
+
+    # no groupby → columns passes through unchanged
+    result = schema.load({"columns": ["country_name"]})
+    assert result.get("columns") == ["country_name"]
+    assert "groupby" not in result
+
+    # granularity_sqla → granularity
+    result = schema.load({"granularity_sqla": "ds"})
+    assert result.get("granularity") == "ds"
+    assert "granularity_sqla" not in result
+
+    # timeseries_limit → series_limit
+    result = schema.load({"timeseries_limit": 5})
+    assert result.get("series_limit") == 5
+    assert "timeseries_limit" not in result
+
+    # timeseries_limit_metric → series_limit_metric
+    result = schema.load({"timeseries_limit_metric": "count"})
+    assert result.get("series_limit_metric") == "count"
+    assert "timeseries_limit_metric" not in result
+
+
@pytest.mark.parametrize(
    "app",
    [{"TIME_GRAIN_ADDONS": {"PT10M": "10 minutes"}}],
--- a/tests/unit_tests/databases/filters_test.py
+++ b/tests/unit_tests/databases/filters_test.py
@@ -119,5 +119,5 @@ def test_database_filter(mocker: MockerFixture) -> None:
    )
    assert (
        str(compiled_query)
-        == "SELECT dbs.uuid, dbs.created_on, dbs.changed_on, dbs.id, dbs.verbose_name, dbs.database_name, dbs.sqlalchemy_uri, dbs.password, dbs.cache_timeout, dbs.select_as_create_table_as, dbs.expose_in_sqllab, dbs.configuration_method, dbs.allow_run_async, dbs.allow_file_upload, dbs.allow_ctas, dbs.allow_cvas, dbs.allow_dml, dbs.force_ctas_schema, dbs.extra, dbs.encrypted_extra, dbs.impersonate_user, dbs.server_cert, dbs.is_managed_externally, dbs.external_url, dbs.created_by_fk, dbs.changed_by_fk, ssh_tunnels_1.uuid AS uuid_1, ssh_tunnels_1.created_on AS created_on_1, ssh_tunnels_1.changed_on AS changed_on_1, ssh_tunnels_1.extra_json, ssh_tunnels_1.id AS id_1, ssh_tunnels_1.database_id, ssh_tunnels_1.server_address, ssh_tunnels_1.server_port, ssh_tunnels_1.username, ssh_tunnels_1.password AS password_1, ssh_tunnels_1.private_key, ssh_tunnels_1.private_key_password, ssh_tunnels_1.created_by_fk AS created_by_fk_1, ssh_tunnels_1.changed_by_fk AS changed_by_fk_1 \nFROM dbs LEFT OUTER JOIN ssh_tunnels AS ssh_tunnels_1 ON dbs.id = ssh_tunnels_1.database_id \nWHERE '[' || dbs.database_name || '].(id:' || CAST(dbs.id AS VARCHAR) || ')' IN ('[my_db].(id:42)', '[my_other_db].(id:43)') OR dbs.database_name IN ('my_db', 'my_other_db', 'third_db')"  # noqa: E501
+        == "SELECT dbs.uuid, dbs.created_on, dbs.changed_on, dbs.id, dbs.verbose_name, dbs.database_name, dbs.sqlalchemy_uri, dbs.password, dbs.cache_timeout, dbs.select_as_create_table_as, dbs.expose_in_sqllab, dbs.configuration_method, dbs.allow_run_async, dbs.allow_file_upload, dbs.allow_ctas, dbs.allow_cvas, dbs.allow_dml, dbs.force_ctas_schema, dbs.extra, dbs.encrypted_extra, dbs.impersonate_user, dbs.server_cert, dbs.is_managed_externally, dbs.external_url, dbs.created_by_fk, dbs.changed_by_fk, ssh_tunnels_1.uuid AS uuid_1, ssh_tunnels_1.created_on AS created_on_1, ssh_tunnels_1.changed_on AS changed_on_1, ssh_tunnels_1.extra_json, ssh_tunnels_1.id AS id_1, ssh_tunnels_1.database_id, ssh_tunnels_1.server_address, ssh_tunnels_1.server_port, ssh_tunnels_1.username, ssh_tunnels_1.password AS password_1, ssh_tunnels_1.private_key, ssh_tunnels_1.private_key_password, ssh_tunnels_1.server_host_key, ssh_tunnels_1.created_by_fk AS created_by_fk_1, ssh_tunnels_1.changed_by_fk AS changed_by_fk_1 \nFROM dbs LEFT OUTER JOIN ssh_tunnels AS ssh_tunnels_1 ON dbs.id = ssh_tunnels_1.database_id \nWHERE '[' || dbs.database_name || '].(id:' || CAST(dbs.id AS VARCHAR) || ')' IN ('[my_db].(id:42)', '[my_other_db].(id:43)') OR dbs.database_name IN ('my_db', 'my_other_db', 'third_db')"  # noqa: E501
    )
--- a/tests/unit_tests/db_engine_specs/test_bigquery.py
+++ b/tests/unit_tests/db_engine_specs/test_bigquery.py
@@ -448,7 +448,92 @@ def test_adjust_engine_params_catalog_as_host() -> None:
        {},
        catalog="other-project",
    )[0]
-    assert str(uri) == "bigquery://other-project/"
+    assert uri.host == "other-project"
+    assert not uri.database  # no dataset when only catalog is overridden
+
+
+def test_adjust_engine_params_schema_as_dataset() -> None:
+    """
+    Test that passing a schema sets it as the BigQuery default dataset.
+
+    BigQuery requires table names to be fully qualified (project.dataset.table)
+    unless a default dataset is set via the URL database component. When schema
+    is provided, the URL database should be updated so unqualified table names
+    resolve to schema.table_name.
+    """
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+
+    url = make_url("bigquery://project")
+
+    # Without schema, URL is unchanged
+    uri = BigQueryEngineSpec.adjust_engine_params(url, {})[0]
+    assert str(uri) == "bigquery://project"
+
+    # With schema, database component is set to enable default dataset
+    uri = BigQueryEngineSpec.adjust_engine_params(
+        url,
+        {},
+        schema="my_dataset",
+    )[0]
+    assert uri.database == "my_dataset"
+
+    # catalog + schema: catalog goes to host, schema goes to database
+    uri = BigQueryEngineSpec.adjust_engine_params(
+        url,
+        {},
+        catalog="other-project",
+        schema="my_dataset",
+    )[0]
+    assert uri.host == "other-project"
+    assert uri.database == "my_dataset"
+
+    # Triple-slash form (bigquery:///project): project must not be overwritten
+    triple_slash_url = make_url("bigquery:///my_project")
+    uri = BigQueryEngineSpec.adjust_engine_params(
+        triple_slash_url,
+        {},
+        schema="my_dataset",
+    )[0]
+    assert uri.host == "my_project"
+    assert uri.database == "my_dataset"
+
+
+def test_get_schema_from_engine_params() -> None:
+    """
+    Test that get_schema_from_engine_params returns the dataset from
+    bigquery://project/dataset URIs and None for all other URL forms.
+    """
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+
+    # Standard form: project in host, dataset in database
+    assert (
+        BigQueryEngineSpec.get_schema_from_engine_params(
+            make_url("bigquery://project/my_dataset"), {}
+        )
+        == "my_dataset"
+    )
+
+    # Project-only URI — no default dataset configured
+    assert (
+        BigQueryEngineSpec.get_schema_from_engine_params(
+            make_url("bigquery://project"), {}
+        )
+        is None
+    )
+
+    # Triple-slash form — database component is the project, not a dataset
+    assert (
+        BigQueryEngineSpec.get_schema_from_engine_params(
+            make_url("bigquery:///my_project"), {}
+        )
+        is None
+    )
+
+    # Bare URI — no project, no dataset
+    assert (
+        BigQueryEngineSpec.get_schema_from_engine_params(make_url("bigquery://"), {})
+        is None
+    )


 def test_get_materialized_view_names() -> None:
--- a/tests/unit_tests/extensions/ssh_test.py
+++ b/tests/unit_tests/extensions/ssh_test.py
@@ -14,11 +14,44 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from unittest.mock import Mock
+from unittest.mock import Mock, patch

+import paramiko
+import pytest
 import sshtunnel

-from superset.extensions.ssh import SSHManagerFactory
+from superset.commands.database.ssh_tunnel.exceptions import (
+    SSHTunnelHostKeyVerificationError,
+)
+from superset.extensions.ssh import SSHManager, SSHManagerFactory
+
+
+def _make_manager(strict: bool = False) -> SSHManager:
+    """Build an ``SSHManager`` test instance with configurable strict checking."""
+    app = Mock()
+    app.config = {
+        "SSH_TUNNEL_MAX_RETRIES": 2,
+        "SSH_TUNNEL_LOCAL_BIND_ADDRESS": "127.0.0.1",
+        "SSH_TUNNEL_TIMEOUT_SEC": 123.0,
+        "SSH_TUNNEL_PACKET_TIMEOUT_SEC": 321.0,
+        "SSH_TUNNEL_MANAGER_CLASS": "superset.extensions.ssh.SSHManager",
+        "SSH_TUNNEL_STRICT_HOST_KEY_CHECKING": strict,
+    }
+    return SSHManager(app)
+
+
+def _authorized_key(key: paramiko.PKey) -> str:
+    """Render a paramiko key in authorized-key (``"<type> <base64>"``) form."""
+    return f"{key.get_name()} {key.get_base64()}"
+
+
+def _ssh_tunnel(server_host_key: str | None) -> Mock:
+    """Create a mocked SSH tunnel with server connection fields populated."""
+    tunnel = Mock()
+    tunnel.server_address = "ssh.example.com"
+    tunnel.server_port = 22
+    tunnel.server_host_key = server_host_key
+    return tunnel


 def test_ssh_tunnel_timeout_setting() -> None:
@@ -34,3 +67,199 @@ def test_ssh_tunnel_timeout_setting() -> None:
    factory.init_app(app)
    assert sshtunnel.TUNNEL_TIMEOUT == 123.0
    assert sshtunnel.SSH_TIMEOUT == 321.0
+
+
+@patch("superset.extensions.ssh.socket.create_connection")
+@patch("superset.extensions.ssh.paramiko.Transport")
+def test_verify_host_key_match(
+    mock_transport_cls: Mock, mock_create_connection: Mock
+) -> None:
+    """The server presents the same key we expect: verification passes."""
+    server_key = paramiko.RSAKey.generate(2048)
+    manager = _make_manager(strict=False)
+    tunnel = _ssh_tunnel(_authorized_key(server_key))
+
+    transport = mock_transport_cls.return_value
+    transport.get_remote_server_key.return_value = server_key
+
+    result = manager._verify_host_key(tunnel)  # should not raise
+
+    # The TCP connect is bounded by an explicit timeout, and the resulting
+    # socket is handed to Transport.
+    mock_create_connection.assert_called_once_with(
+        ("ssh.example.com", 22), timeout=321.0
+    )
+    mock_transport_cls.assert_called_once_with(mock_create_connection.return_value)
+    transport.start_client.assert_called_once()
+    transport.close.assert_called_once()
+    # The parsed expected key is returned so the caller can pin it on the tunnel.
+    assert result == server_key
+
+
+@patch("superset.extensions.ssh.socket.create_connection")
+@patch("superset.extensions.ssh.paramiko.Transport")
+def test_verify_host_key_mismatch_raises(
+    mock_transport_cls: Mock, mock_create_connection: Mock
+) -> None:
+    """The server presents a different key than expected: verification fails."""
+    expected_key = paramiko.RSAKey.generate(2048)
+    presented_key = paramiko.RSAKey.generate(2048)
+    manager = _make_manager(strict=False)
+    tunnel = _ssh_tunnel(_authorized_key(expected_key))
+
+    transport = mock_transport_cls.return_value
+    transport.get_remote_server_key.return_value = presented_key
+
+    with pytest.raises(SSHTunnelHostKeyVerificationError):
+        manager._verify_host_key(tunnel)
+
+    mock_create_connection.assert_called_once()
+    transport.close.assert_called_once()
+
+
+@patch("superset.extensions.ssh.socket.create_connection")
+def test_verify_host_key_connect_failure_raises(
+    mock_create_connection: Mock,
+) -> None:
+    """A bounded TCP connect failure surfaces as a host-key verification error."""
+    manager = _make_manager(strict=False)
+    server_key = paramiko.RSAKey.generate(2048)
+    tunnel = _ssh_tunnel(_authorized_key(server_key))
+
+    mock_create_connection.side_effect = OSError("connection refused")
+
+    with pytest.raises(SSHTunnelHostKeyVerificationError):
+        manager._verify_host_key(tunnel)
+
+
+@patch("superset.extensions.ssh.paramiko.Transport")
+def test_verify_host_key_unset_non_strict_skips(mock_transport_cls: Mock) -> None:
+    """Back-compat: no expected key + strict checking off => no verification at all."""
+    manager = _make_manager(strict=False)
+    tunnel = _ssh_tunnel(None)
+
+    assert manager._verify_host_key(tunnel) is None  # should not raise
+
+    mock_transport_cls.assert_not_called()
+
+
+@patch("superset.extensions.ssh.paramiko.Transport")
+def test_verify_host_key_unset_strict_raises(mock_transport_cls: Mock) -> None:
+    """Fail-closed: no expected key + strict checking on => reject."""
+    manager = _make_manager(strict=True)
+    tunnel = _ssh_tunnel(None)
+
+    with pytest.raises(SSHTunnelHostKeyVerificationError):
+        manager._verify_host_key(tunnel)
+
+    mock_transport_cls.assert_not_called()
+
+
+@patch("superset.extensions.ssh.socket.create_connection")
+@patch("superset.extensions.ssh.paramiko.Transport")
+def test_verify_host_key_match_ignores_comment_and_whitespace(
+    mock_transport_cls: Mock,
+    mock_create_connection: Mock,
+) -> None:
+    # The stored key may carry a trailing comment and extra whitespace.
+    server_key = paramiko.RSAKey.generate(2048)
+    manager = _make_manager(strict=False)
+    stored = f"  {_authorized_key(server_key)} user@host  "
+    tunnel = _ssh_tunnel(stored)
+
+    transport = mock_transport_cls.return_value
+    transport.get_remote_server_key.return_value = server_key
+
+    manager._verify_host_key(tunnel)  # should not raise
+
+    # Whitespace/comment stripping must not short-circuit verification: the
+    # bounded TCP connect and Transport handshake still run as in the plain
+    # match case.
+    mock_create_connection.assert_called_once_with(
+        ("ssh.example.com", 22), timeout=321.0
+    )
+    mock_transport_cls.assert_called_once_with(mock_create_connection.return_value)
+    transport.start_client.assert_called_once()
+    transport.close.assert_called_once()
+
+
+def test_verify_host_key_invalid_expected_raises() -> None:
+    # A malformed expected key is rejected before any network connection.
+    manager = _make_manager(strict=False)
+    tunnel = _ssh_tunnel("not-a-valid-key")
+
+    with pytest.raises(SSHTunnelHostKeyVerificationError):
+        manager._verify_host_key(tunnel)
+
+
+def test_verify_host_key_unknown_key_type_raises() -> None:
+    """An unsupported key type is wrapped in the verification error, not leaked."""
+    manager = _make_manager(strict=False)
+    server_key = paramiko.RSAKey.generate(2048)
+    tunnel = _ssh_tunnel(f"ssh-bogus {server_key.get_base64()}")
+
+    with pytest.raises(SSHTunnelHostKeyVerificationError):
+        manager._verify_host_key(tunnel)
+
+
+@patch("superset.extensions.ssh.sshtunnel.open_tunnel")
+@patch("superset.extensions.ssh.socket.create_connection")
+@patch("superset.extensions.ssh.paramiko.Transport")
+def test_create_tunnel_pins_verified_host_key(
+    mock_transport_cls: Mock,
+    mock_create_connection: Mock,
+    mock_open_tunnel: Mock,
+) -> None:
+    """A verified expected key is also pinned on the tunnel's own connection.
+
+    When an expected host key is configured and verified, it is also pinned on the
+    tunnel's own connection (``ssh_host_key``) so paramiko verifies the host that
+    actually carries traffic on the same transport — closing the probe-vs-tunnel
+    TOCTOU gap rather than trusting only the pre-flight probe.
+    """
+    server_key = paramiko.RSAKey.generate(2048)
+    manager = _make_manager(strict=False)
+    tunnel = _ssh_tunnel(_authorized_key(server_key))
+    tunnel.username = "user"
+    tunnel.password = None
+    tunnel.private_key = None
+
+    mock_transport_cls.return_value.get_remote_server_key.return_value = server_key
+
+    manager.create_tunnel(tunnel, "postgresql://u:p@db:5432/ex")
+
+    _, kwargs = mock_open_tunnel.call_args
+    assert kwargs["ssh_host_key"] == server_key
+
+
+@patch("superset.extensions.ssh.sshtunnel.open_tunnel")
+def test_create_tunnel_without_host_key_does_not_pin(mock_open_tunnel: Mock) -> None:
+    # No expected key configured (non-strict): nothing is pinned, preserving the
+    # prior behavior.
+    manager = _make_manager(strict=False)
+    tunnel = _ssh_tunnel(None)
+    tunnel.username = "user"
+    tunnel.password = None
+    tunnel.private_key = None
+
+    manager.create_tunnel(tunnel, "postgresql://u:p@db:5432/ex")
+
+    _, kwargs = mock_open_tunnel.call_args
+    assert "ssh_host_key" not in kwargs
+
+
+def test_ssh_tunnel_schema_round_trips_server_host_key() -> None:
+    """The schema accepts and preserves the public host key field."""
+    from superset.databases.schemas import DatabaseSSHTunnel
+
+    server_key = paramiko.RSAKey.generate(2048)
+    authorized = _authorized_key(server_key)
+    payload = {
+        "server_address": "ssh.example.com",
+        "server_port": 22,
+        "username": "user",
+        "password": "secret",
+        "server_host_key": authorized,
+    }
+    loaded = DatabaseSSHTunnel().load(payload)
+    assert loaded["server_host_key"] == authorized
--- a/tests/unit_tests/security/manager_test.py
+++ b/tests/unit_tests/security/manager_test.py
@@ -1218,6 +1218,182 @@ def test_query_context_modified_orderby(mocker: MockerFixture) -> None:
    assert query_context_modified(query_context)


+def test_query_context_modified_time_grain_native_filter(
+    mocker: MockerFixture,
+) -> None:
+    """
+    Test `query_context_modified` when a guest applies a Time Grain native filter.
+
+    Reproduces https://github.com/apache/superset/issues/32768.
+
+    On a chart that uses a generic x-axis, the selected time grain is baked into the
+    ``BASE_AXIS`` adhoc column as a ``timeGrain`` property (see
+    ``normalizeTimeColumn`` on the frontend, which copies ``extras.time_grain_sqla``
+    onto the column). A Time Grain native filter is a supported, read-only guest
+    interaction: it only changes the granularity at which the *same* dimension is
+    bucketed, never which metrics or columns are queried.
+
+    Previously, because the changed time grain travels inside the ``columns``
+    payload, the subset comparison treated the request as tampering and
+    ``query_context_modified`` returned ``True`` -- so guests hit "Guest user cannot
+    modify chart payload" whenever they picked a grain other than the chart default.
+
+    ``freeze_value`` now drops the guest-overridable ``timeGrain`` key before
+    comparing, so a pure time-grain change is no longer flagged as a modification.
+    This test guards that behavior.
+    """
+    # The chart was saved with a monthly grain on its x-axis column.
+    stored_axis_column: AdhocColumn = {
+        "label": "order_date",
+        "sqlExpression": "order_date",
+        "columnType": "BASE_AXIS",
+        "timeGrain": "P1M",
+    }
+    # The guest picked a daily grain via the dashboard Time Grain native filter;
+    # `normalizeTimeColumn` rewrote the otherwise-identical column accordingly.
+    requested_axis_column: AdhocColumn = {
+        **stored_axis_column,
+        "timeGrain": "P1D",
+    }
+
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.params_dict = {
+        "metrics": ["count"],
+    }
+    query_context.slice_.query_context = json.dumps(
+        {
+            "queries": [
+                {
+                    "columns": [stored_axis_column],
+                    "metrics": ["count"],
+                }
+            ],
+        }
+    )
+    # Native-filter data requests don't carry the mutated columns at the top level;
+    # the grain change only shows up inside the query's columns.
+    query_context.form_data = {
+        "slice_id": 42,
+        "metrics": ["count"],
+    }
+    query_context.queries = [
+        QueryObject(
+            columns=[requested_axis_column],
+            metrics=["count"],
+        ),
+    ]
+
+    assert not query_context_modified(query_context)
+
+
+def test_query_context_modified_time_grain_with_tampered_column(
+    mocker: MockerFixture,
+) -> None:
+    """
+    Test that relaxing the time grain comparison does not open a tamper hole.
+
+    Only the ``timeGrain`` key is guest-overridable. A request that changes the
+    grain *and* also swaps a non-overridable attribute (here ``sqlExpression``,
+    which selects which column is queried) must still be flagged as tampering --
+    otherwise a guest could query an arbitrary column under cover of a Time Grain
+    filter.
+    """
+    stored_axis_column: AdhocColumn = {
+        "label": "order_date",
+        "sqlExpression": "order_date",
+        "columnType": "BASE_AXIS",
+        "timeGrain": "P1M",
+    }
+    # Guest changes the grain (allowed) but also rewrites the SQL expression to a
+    # different column (not allowed) -- this must still read as a modification.
+    tampered_axis_column: AdhocColumn = {
+        **stored_axis_column,
+        "sqlExpression": "secret_column",
+        "timeGrain": "P1D",
+    }
+
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.params_dict = {
+        "metrics": ["count"],
+    }
+    query_context.slice_.query_context = json.dumps(
+        {
+            "queries": [
+                {
+                    "columns": [stored_axis_column],
+                    "metrics": ["count"],
+                }
+            ],
+        }
+    )
+    query_context.form_data = {
+        "slice_id": 42,
+        "metrics": ["count"],
+    }
+    query_context.queries = [
+        QueryObject(
+            columns=[tampered_axis_column],
+            metrics=["count"],
+        ),
+    ]
+
+    assert query_context_modified(query_context)
+
+
+def test_query_context_modified_time_grain_in_orderby(
+    mocker: MockerFixture,
+) -> None:
+    """
+    Test `query_context_modified` when the time grain travels inside `orderby`.
+
+    Each ``orderby`` entry is an ``(column, bool)`` tuple, so a temporal x-axis
+    adhoc column carrying the guest-overridable ``timeGrain`` is nested one level
+    deep rather than sitting at the top level. The overridable key must still be
+    stripped before comparing, otherwise sorting by the temporal axis would make
+    a pure time-grain change read as tampering.
+    """
+    stored_axis_column: AdhocColumn = {
+        "label": "order_date",
+        "sqlExpression": "order_date",
+        "columnType": "BASE_AXIS",
+        "timeGrain": "P1M",
+    }
+    requested_axis_column: AdhocColumn = {
+        **stored_axis_column,
+        "timeGrain": "P1D",
+    }
+
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.params_dict = {
+        "metrics": ["count"],
+    }
+    query_context.slice_.query_context = json.dumps(
+        {
+            "queries": [
+                {
+                    "orderby": [[stored_axis_column, True]],
+                    "metrics": ["count"],
+                }
+            ],
+        }
+    )
+    query_context.form_data = {
+        "slice_id": 42,
+        "metrics": ["count"],
+    }
+    query_context.queries = [
+        QueryObject(
+            orderby=[(requested_axis_column, True)],
+            metrics=["count"],
+        ),
+    ]
+
+    assert not query_context_modified(query_context)
+
+
 def test_get_catalog_perm() -> None:
    """
    Test the `get_catalog_perm` method.
Author	SHA1	Message	Date
Elizabeth Thompson	9058b49971	fix(reports): commit permalink before Playwright navigates to tab URL CreateDashboardPermalinkCommand only flushes the INSERT when invoked inside an outer @transaction() (the nesting guard skips the commit). The row is therefore not visible to Playwright's separate DB connection, causing dashboard_permalink to return a 404, the <body class="standalone"> element to never appear, and the screenshot to time out after 600s. Commit immediately after CreateDashboardPermalinkCommand.run() returns, matching the pattern already used by create_log() in the same class. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>	2026-06-15 23:50:30 +00:00
Amin Ghadersohi	c1b5d05f83	fix(bigquery): set default dataset from schema in adjust_engine_params (#40776 )	2026-06-15 18:37:06 -04:00
Evan Rusackas	e16bb29faf	fix(embedded): allow guests to apply a Time Grain native filter (#32768 ) (#41017 ) Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>	2026-06-15 15:22:21 -07:00
Elizabeth Thompson	09b4bc51a3	fix(charts): rename deprecated query object fields in schema before QueryObject construction (#41056 )	2026-06-15 14:45:41 -07:00
Evan Rusackas	379435b7eb	feat(ssh_tunnel): add opt-in server host key verification (#40673 ) Co-authored-by: Claude Code <noreply@anthropic.com>	2026-06-15 12:17:14 -07:00