Merge branch 'master' into chore/disable-git-commit-info-capture-in-playwright-e2e-tests

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Co-authored-by: Joe Li <joe@preset.io>

UPDATING.md

@@ -24,6 +24,8 @@ assists people when migrating to a new version.
 
 ## Next
 
+- `SAMPLES_ROW_LIMIT` is now the default for `/datasource/samples` requests without a valid explicit `per_page`, rather than a hard per-request ceiling; explicit limits are honored up to the existing global row-limit ceiling, matching `/chart/data` SAMPLES requests.
+
 ### OAuth2 database callback metrics include their outcome
 
 The unqualified `DatabaseRestApi.oauth2` StatsD counter has been replaced with

superset-frontend/playwright.config.ts

@@ -47,6 +47,10 @@ export default defineConfig({
   // Retry logic - 2 retries in CI, 0 locally
   retries: process.env.CI ? 2 : 0,
 
+  // Disable capturing Git commit info as the project's history is increasingly dense
+  // and breach Playwright's default 3-seconds `git` command timeout limit
+  captureGitInfo: { commit: false, diff: false },
+
   // Reporter configuration - multiple reporters for better visibility
   reporter: process.env.CI
     ? [

superset-frontend/src/dashboard/components/nativeFilters/FilterBar/FilterControls/GroupByFilterCard.test.tsx

@@ -16,8 +16,20 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+import {
+  ChartCustomizationType,
+  type ChartCustomization,
+} from '@superset-ui/core';
 import { LabeledValue } from '@superset-ui/core/components';
-import { createLabelSortComparator } from './GroupByFilterCard';
+import { render, screen } from 'spec/helpers/testing-library';
+import GroupByFilterCard, {
+  createLabelSortComparator,
+} from './GroupByFilterCard';
+
+jest.mock('src/utils/cachedSupersetGet', () => ({
+  // Never resolves, pinning the card in its column-loading state.
+  cachedSupersetGet: jest.fn(() => new Promise(() => {})),
+}));
 
 const apple: LabeledValue = { value: 'a', label: 'Apple' };
 const banana: LabeledValue = { value: 'b', label: 'Banana' };
@@ -39,3 +51,27 @@ test('preserves source order when sortAscending is unset', () => {
   expect(compare(apple, banana)).toBe(0);
   expect(compare(banana, apple)).toBe(0);
 });
+
+const groupByCustomization: ChartCustomization = {
+  id: 'groupby-1',
+  name: 'Group By',
+  filterType: 'filter_groupby',
+  type: ChartCustomizationType.ChartCustomization,
+  targets: [{ datasetId: 1 }],
+  scope: { rootPath: [], excluded: [] },
+  controlValues: {},
+  defaultDataMask: {},
+};
+
+test('renders the column-loading spinner small and muted', async () => {
+  render(<GroupByFilterCard customizationItem={groupByCustomization} />, {
+    useRedux: true,
+    initialState: {
+      dataMask: {},
+      nativeFilters: { filters: {} },
+    },
+  });
+  const spinner = await screen.findByTestId('loading-indicator');
+  expect(spinner).toHaveClass('inline');
+  expect(spinner).toHaveStyle({ opacity: 0.25, width: '40px' });
+});

superset-frontend/src/dashboard/components/nativeFilters/FilterBar/FilterControls/GroupByFilterCard.tsx

@@ -645,7 +645,7 @@ const GroupByFilterCard: FC<GroupByFilterCardProps> = ({
 
       {loading && (
         <div style={{ textAlign: 'center', marginTop: 8 }}>
-          <Loading position="inline" />
+          <Loading position="inline" size="s" muted />
         </div>
       )}
     </div>

superset/security/manager.py

@@ -36,7 +36,7 @@ from urllib.parse import quote
 
 from flask import current_app, Flask, g, has_app_context, Request, Response
 from flask_appbuilder import Model
-from flask_appbuilder.api import expose, protect, safe
+from flask_appbuilder.api import expose, permission_name, protect, safe
 from flask_appbuilder.models.filters import BaseFilter
 from flask_appbuilder.security.manager import AUTH_REMOTE_USER
 from flask_appbuilder.security.sqla.apis import GroupApi, RoleApi, UserApi
@@ -394,8 +394,11 @@ class SupersetUserApi(UserApi):
     """
     Overriding the UserApi to sync Subject rows, filter excluded users,
     handle deletion constraints, and add audit logging.
-    UserApi has custom post/put that bypass hooks, so we override them
-    and sync after the parent method succeeds.
+
+    The Subject sync happens in ``pre_add``/``pre_update``, which FAB calls
+    *before* the commit that ``self.datamodel.add``/``edit`` issues -- so the
+    sync rides that same commit rather than needing one of its own after the
+    fact.
     """
 
     base_filters = [["username", ExcludeUsersFilter, lambda: []]]
@@ -415,6 +418,45 @@ class SupersetUserApi(UserApi):
         "changed_on",
     ]
 
+    def pre_add(self, item: Model) -> None:
+        """Hash the password (FAB's own ``pre_add``), then sync the user's
+        ``Subject`` row before FAB's own commit.
+
+        ``UserApi.post`` calls ``pre_add`` *before* ``self.datamodel.add``,
+        which is what actually issues the commit -- so flushing the new user
+        here (to obtain its id) and syncing its ``Subject`` row alongside it
+        means both writes ride the same transaction and commit together,
+        instead of the subject sync needing a second, separate commit after
+        the fact.
+        """
+        super().pre_add(item)
+        from superset.daos.user import UserDAO
+
+        self.datamodel.session.add(item)
+        self.datamodel.session.flush()
+        UserDAO._sync_subject(item)
+
+    def pre_update(self, item: Model, data: dict[str, Any]) -> None:
+        """Same reasoning as ``pre_add``: ``UserApi.put`` calls ``pre_update``
+        before ``self.datamodel.edit`` commits, so the subject sync lands in
+        that same transaction.
+        """
+        super().pre_update(item, data)
+        from superset.daos.user import UserDAO
+
+        UserDAO._sync_subject(item)
+
+        if data.get("password"):
+            # An admin-initiated password change via this endpoint must
+            # invalidate the target account's other outstanding sessions,
+            # the same as the self-service ``/me/`` path and the two
+            # password-reset views.
+            from superset.security.session_invalidation import (
+                invalidate_sessions_for_user,
+            )
+
+            invalidate_sessions_for_user(item.id)
+
     @expose("/", methods=["POST"])
     @protect()
     @safe
@@ -430,17 +472,7 @@ class SupersetUserApi(UserApi):
             500:
               description: Server error
         """
-        response = super().post()
-        if response.status_code == 201:
-            from superset.daos.user import UserDAO
-
-            user_id = response.json.get("id")
-            if user_id:
-                user = self.datamodel.session.get(self.datamodel.obj, user_id)
-                if user:
-                    UserDAO._sync_subject(user)
-                    self.datamodel.session.commit()  # pylint: disable=consider-using-transaction
-        return response
+        return super().post()
 
     @expose("/<pk>", methods=["PUT"])
     @protect()
@@ -464,15 +496,42 @@ class SupersetUserApi(UserApi):
             500:
               description: Server error
         """
-        response = super().put(pk)
-        if response.status_code == 200:
-            from superset.daos.user import UserDAO
+        return super().put(pk)
 
-            user = self.datamodel.get(pk, self._base_filters)
-            if user:
-                UserDAO._sync_subject(user)
-                self.datamodel.session.commit()  # pylint: disable=consider-using-transaction
-        return response
+    @expose("/<int:pk>/sessions", methods=["DELETE"])
+    @protect()
+    @permission_name("put")
+    @safe
+    def terminate_sessions(self, pk: int) -> Response:
+        """Terminate a user's outstanding sessions without disabling their account.
+        ---
+        delete:
+          parameters:
+            - in: path
+              name: pk
+              schema:
+                type: integer
+          responses:
+            200:
+              description: Sessions terminated
+            404:
+              $ref: '#/components/responses/404'
+            500:
+              $ref: '#/components/responses/500'
+        """
+        from superset.security.session_invalidation import invalidate_sessions_for_user
+
+        user = self.datamodel.get(pk, self._base_filters)
+        if not user:
+            return self.response_404()
+
+        invalidate_sessions_for_user(user.id)
+        self.datamodel.session.commit()  # pylint: disable=consider-using-transaction
+        _log_audit_event(
+            "UserSessionsTerminated",
+            {"target_username": user.username, "target_user_id": user.id},
+        )
+        return self.response(200, message="User sessions terminated.")
 
     def pre_delete(self, item: Model) -> None:
         from superset.daos.user import UserDAO
@@ -1565,9 +1624,23 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
         bypassed. We distinguish the two by comparing the acting user
         (``g.user``) against the target ``userid``: they match for a
         self-service reset and differ for an admin reset.
+
+        Also stamps the session-invalidation epoch for the target user, so
+        any session for the account that predates this reset stops working --
+        regardless of which of the two paths triggered it.
         """
         super().reset_password(userid, password)
 
+        # pylint: disable=import-outside-toplevel
+        from superset import db
+        from superset.security.session_invalidation import invalidate_sessions_for_user
+
+        invalidate_sessions_for_user(int(userid))
+        # ``super().reset_password`` (FAB's ``update_user``) already committed
+        # its own change in a separate transaction, so the epoch stamp above
+        # needs its own commit too, rather than riding an existing one.
+        db.session.commit()  # pylint: disable=consider-using-transaction
+
         acting_user = getattr(g, "user", None)
         acting_user_id = getattr(acting_user, "id", None)
         # ``userid`` arrives as a string (the ``pk`` request arg) on the admin

superset/security/session_invalidation.py

@@ -41,7 +41,7 @@ from typing import Any, Optional
 from flask import flash, session
 from flask_babel import gettext as __
 from flask_login import current_user, logout_user
-from sqlalchemy import event, inspect
+from sqlalchemy import event, inspect, or_
 from sqlalchemy.exc import IntegrityError
 from werkzeug.wrappers import Response
 
@@ -163,9 +163,20 @@ def invalidate_user_sessions(connection: Any, user_id: int) -> None:
     )
 
     def _stamp_existing() -> int:
+        # Guard against two concurrent writers regressing the epoch: a
+        # transaction that computed an earlier ``now`` can reach this UPDATE
+        # after one with a later ``now`` has already committed. Only apply
+        # the write when it would advance (or initialize) the stored value,
+        # so the epoch is monotonic regardless of commit order.
         return connection.execute(
             table.update()
             .where(table.c.user_id == user_id)
+            .where(
+                or_(
+                    table.c.sessions_invalidated_at.is_(None),
+                    table.c.sessions_invalidated_at < now,
+                )
+            )
             .values(sessions_invalidated_at=now, changed_on=now)
         ).rowcount
 
@@ -187,6 +198,23 @@ def invalidate_user_sessions(connection: Any, user_id: int) -> None:
         _stamp_existing()
 
 
+def invalidate_sessions_for_user(user_id: int) -> None:
+    """Stamp the invalidation epoch for ``user_id`` from ordinary application code.
+
+    Convenience wrapper around ``invalidate_user_sessions`` for callers that
+    don't have the raw ``Connection`` the ``after_update`` event listener
+    receives -- e.g. a password-change flow. The stamp is written through the
+    current session's own connection, so it participates in whatever
+    transaction the caller's other pending changes belong to; it is not
+    committed here, so the caller's own commit (or the next flush that
+    triggers one) is what makes it durable.
+    """
+    # pylint: disable=import-outside-toplevel
+    from superset.extensions import db
+
+    invalidate_user_sessions(db.session.connection(), user_id)
+
+
 def _stamp_epoch_on_disable(_mapper: Any, connection: Any, target: Any) -> None:
     history = inspect(target).attrs.active.history
     # Only act when ``active`` actually changed to False — ignore the

superset/views/datasource/utils.py

@@ -28,7 +28,11 @@ from superset.common.query_context_factory import QueryContextFactory
 from superset.common.utils.query_cache_manager import QueryCacheManager
 from superset.constants import CacheRegion
 from superset.daos.datasource import DatasourceDAO
-from superset.utils.core import extract_dataframe_dtypes, QueryStatus
+from superset.utils.core import (
+    apply_max_row_limit,
+    extract_dataframe_dtypes,
+    QueryStatus,
+)
 from superset.views.datasource.schemas import SamplesPayloadSchema
 
 if TYPE_CHECKING:
@@ -45,9 +49,11 @@ def get_limit_clause(page: Optional[int], per_page: Optional[int]) -> dict[str,
 
     if isinstance(page, int) and isinstance(per_page, int):
         limit = int(per_page)
-        if limit < 0 or limit > samples_row_limit:
+        if limit < 0:
             # reset limit value if input is invalid
             limit = samples_row_limit
+        elif limit:
+            limit = apply_max_row_limit(limit)
 
         offset = max((int(page) - 1) * limit, 0)
 

superset/views/users/api.py

@@ -23,11 +23,12 @@ from flask_appbuilder.security.decorators import protect
 from flask_appbuilder.security.sqla.models import User
 from marshmallow import ValidationError
 from sqlalchemy.orm.exc import NoResultFound
-from werkzeug.security import generate_password_hash
+from werkzeug.security import check_password_hash, generate_password_hash
 
 from superset import is_feature_enabled
 from superset.daos.user import UserDAO
 from superset.extensions import db, event_logger
+from superset.security.session_invalidation import invalidate_sessions_for_user
 from superset.utils.slack import get_user_avatar, SlackClientError
 from superset.views.base_api import BaseSupersetApi, requires_json, statsd_metrics
 from superset.views.users.schemas import CurrentUserPutSchema, UserResponseSchema
@@ -49,12 +50,45 @@ class CurrentUserRestApi(BaseSupersetApi):
     def pre_update(self, item: User, data: Dict[str, Any]) -> None:
         item.changed_on = datetime.now()
         item.changed_by_fk = g.user.id
+        # Pop unconditionally: this key is only meaningful for verifying a
+        # password change below, and it isn't a real column on the user
+        # model -- it must never reach ``UserDAO.update``'s ``setattr`` loop.
+        current_password = data.pop("current_password", None)
         if "password" in data and data["password"]:
+            # An account with no password set yet (e.g. provisioned via an
+            # external auth backend) has nothing to prove knowledge of; for
+            # every other account, the caller must confirm the existing
+            # password before it can be replaced.
+            proof_ok = (
+                item.password
+                and current_password
+                and check_password_hash(item.password, current_password)
+            )
+            if item.password and not proof_ok:
+                raise ValidationError(
+                    {"current_password": ["Incorrect current password."]}
+                )
+            # Compute and assign the hash, then drop the plaintext from
+            # ``data`` -- it is passed to ``UserDAO.update`` as ``attributes``
+            # right after this, and ``BaseDAO.update`` sets every key in it
+            # via ``setattr``. Leaving the plaintext in would overwrite the
+            # hash just assigned below with the raw value.
+            new_password = data.pop("password")
             item.password = generate_password_hash(
-                password=data["password"],
+                password=new_password,
                 method=app.config.get("FAB_PASSWORD_HASH_METHOD", "scrypt"),
                 salt_length=app.config.get("FAB_PASSWORD_HASH_SALT_LENGTH", 16),
             )
+            # A changed password invalidates any other outstanding session
+            # for this account.
+            invalidate_sessions_for_user(item.id)
+        elif "password" in data:
+            # A falsy value (e.g. an empty string, which the complexity
+            # validator lets through when password complexity is disabled)
+            # skips the block above, but the key must still never reach
+            # ``UserDAO.update``'s ``setattr`` loop -- it would blank out
+            # the account's stored hash.
+            data.pop("password")
 
     @expose("/", methods=("GET",))
     @protect()

superset/views/users/schemas.py

@@ -14,17 +14,22 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from typing import Any
+
 from flask_appbuilder.security.sqla.apis.user.schema import User
 from flask_appbuilder.security.sqla.apis.user.validator import (
     PasswordComplexityValidator,
 )
-from marshmallow import fields, Schema
+from marshmallow import fields, Schema, validates_schema, ValidationError
 from marshmallow.fields import Boolean, Integer, String
 from marshmallow.validate import Length
 
 first_name_description = "The current user's first name"
 last_name_description = "The current user's last name"
 password_description = "The current user's password for authentication"  # noqa: S105
+# Required, and verified against the account's existing password, whenever
+# ``password`` is included in the payload.
+current_password_description = "The current user's existing password"  # noqa: S105
 
 
 class UserGroupSchema(Schema):
@@ -64,3 +69,24 @@ class CurrentUserPutSchema(Schema):
         validate=[PasswordComplexityValidator()],
         metadata={"description": password_description},
     )
+    current_password = fields.String(
+        required=False,
+        load_only=True,
+        metadata={"description": current_password_description},
+    )
+
+    @validates_schema
+    def validate_current_password_required_with_password(
+        self, data: dict[str, Any], **kwargs: object
+    ) -> None:
+        """Require ``current_password`` whenever ``password`` is being set.
+
+        This only checks that the field was supplied -- whether it actually
+        matches the account's existing password is verified against the
+        database in ``CurrentUserRestApi.pre_update``, which has access to
+        the user record this schema doesn't.
+        """
+        if data.get("password") and not data.get("current_password"):
+            raise ValidationError(
+                {"current_password": ["This field is required to change the password."]}
+            )

tests/unit_tests/security/test_password_change_session_invalidation.py

@@ -0,0 +1,276 @@
+# 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.
+"""Password-change paths and the session-invalidation epoch.
+
+``UserAttribute.sessions_invalidated_at`` (see
+``superset.security.session_invalidation``) is the mechanism that forces
+outstanding sessions to log out. Originally it was stamped exclusively by the
+``after_update`` listener that fires when an account's ``active`` flag flips
+to ``False``; these tests now cover the additional password-change paths --
+self-service reset, admin-initiated reset, and the ``PUT /api/v1/me/``
+self-service update -- which also stamp that epoch, so a session authenticated
+before a password change stops working after it.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from flask_appbuilder.models.sqla.interface import SQLAInterface
+from flask_appbuilder.security.sqla.models import User
+
+from superset import db, security_manager
+from superset.daos.user import UserDAO
+from superset.models.user_attributes import UserAttribute
+from superset.security.manager import SupersetUserApi
+from superset.views.users.api import CurrentUserRestApi
+from tests.unit_tests.fixtures.common import admin_user, after_each  # noqa: F401
+
+
+def _invalidated_at(user_id: int):
+    attr = db.session.query(UserAttribute).filter_by(user_id=user_id).one_or_none()
+    return attr.sessions_invalidated_at if attr else None
+
+
+@pytest.fixture
+def two_admins() -> Iterator[tuple[User, User]]:
+    """Two admin-role users for the reset_password tests below.
+
+    ``SupersetSecurityManager.reset_password`` -> FAB's ``update_user``
+    hard-commits the session (``commit=True`` by default), so the rollback
+    the shared ``after_each``/``admin_user`` fixtures rely on can't undo it.
+    This fixture creates its own users and deletes them again on teardown so
+    a committed reset doesn't leak rows into later tests.
+    """
+    role = db.session.query(security_manager.role_model).filter_by(name="Admin").one()
+    target = User(
+        first_name="Target",
+        last_name="User",
+        email="session_invalidation_target@example.org",
+        username="session_invalidation_target",
+        roles=[role],
+    )
+    actor = User(
+        first_name="Acting",
+        last_name="Admin",
+        email="session_invalidation_actor@example.org",
+        username="session_invalidation_actor",
+        roles=[role],
+    )
+    db.session.add_all([target, actor])
+    db.session.commit()
+
+    yield target, actor
+
+    db.session.query(UserAttribute).filter(
+        UserAttribute.user_id.in_([target.id, actor.id])
+    ).delete(synchronize_session=False)
+    db.session.query(User).filter(User.id.in_([target.id, actor.id])).delete(
+        synchronize_session=False
+    )
+    db.session.commit()
+
+
+def test_self_service_password_reset_invalidates_other_sessions(
+    two_admins: tuple[User, User],
+) -> None:
+    """``SupersetSecurityManager.reset_password`` used for a self-service
+    reset (acting user resets their own password) stamps the session epoch,
+    so any other outstanding session for the account stops working after the
+    password changes.
+    """
+    target, _actor = two_admins
+
+    with patch("superset.security.manager.g") as mock_g:
+        mock_g.user = target
+        security_manager.reset_password(target.id, "BrandNewPassw0rd!")
+
+    assert _invalidated_at(target.id) is not None
+
+
+def test_admin_password_reset_invalidates_target_sessions(
+    two_admins: tuple[User, User],
+) -> None:
+    """An admin-initiated reset of *another* user's password also stamps the
+    epoch, so the target's outstanding sessions stop working -- this is the
+    closest existing action to an explicit "terminate that user's sessions",
+    short of disabling the account.
+    """
+    target, actor = two_admins
+
+    with patch("superset.security.manager.g") as mock_g:
+        mock_g.user = actor  # differs from target: an admin-initiated reset
+        security_manager.reset_password(target.id, "TemporaryPassw0rd!")
+
+    assert _invalidated_at(target.id) is not None
+
+
+def test_update_me_password_change_invalidates_other_sessions(
+    admin_user: User,  # noqa: F811
+    after_each: None,  # noqa: F811
+) -> None:
+    """The ``PUT /api/v1/me/`` self-service password change (``pre_update`` +
+    ``UserDAO.update`` in ``CurrentUserRestApi.update_me``) also stamps the
+    session-invalidation epoch. ``admin_user`` starts with no password set, so
+    no ``current_password`` proof is required for this change to go through.
+    """
+    api = CurrentUserRestApi()
+    data = {"password": "BrandNewPassw0rd!"}
+
+    with patch("superset.views.users.api.g") as mock_g:
+        mock_g.user = admin_user
+        api.pre_update(admin_user, data)
+        UserDAO.update(item=admin_user, attributes=data)
+    db.session.flush()
+
+    assert _invalidated_at(admin_user.id) is not None
+
+
+def test_admin_edit_user_password_via_put_invalidates_target_sessions(
+    after_each: None,  # noqa: F811
+) -> None:
+    """An admin editing another user's password via ``PUT
+    /api/v1/security/users/<pk>`` (``SupersetUserApi.pre_update``, which FAB's
+    ``UserApi.put`` calls before its own commit) must also stamp the target's
+    session-invalidation epoch, the same as the self-service ``/me/`` path and
+    the two password-reset views -- otherwise this admin path is the one way
+    to change a user's password that leaves their other sessions alive.
+    """
+    role = db.session.query(security_manager.role_model).filter_by(name="Admin").one()
+    user = User(
+        first_name="Target",
+        last_name="User",
+        email="admin_edit_password_target@example.org",
+        username="admin_edit_password_target",
+        roles=[role],
+    )
+    db.session.add(user)
+    db.session.commit()
+
+    api = SupersetUserApi()
+    api.datamodel = SQLAInterface(User, db.session)
+    api.appbuilder = SimpleNamespace(
+        sm=SimpleNamespace(current_user=SimpleNamespace(id=1))
+    )
+
+    api.pre_update(user, {"password": "AdminSetPassw0rd!"})
+
+    assert _invalidated_at(user.id) is not None
+
+    db.session.query(UserAttribute).filter_by(user_id=user.id).delete(
+        synchronize_session=False
+    )
+    db.session.query(User).filter_by(id=user.id).delete(synchronize_session=False)
+    db.session.commit()
+
+
+def test_admin_edit_user_without_password_change_does_not_invalidate_sessions(
+    after_each: None,  # noqa: F811
+) -> None:
+    """Editing a user through the same endpoint *without* touching the
+    password (e.g. renaming them) must not stamp the epoch -- only an actual
+    password change should force other sessions to log out.
+    """
+    role = db.session.query(security_manager.role_model).filter_by(name="Admin").one()
+    user = User(
+        first_name="Target",
+        last_name="User",
+        email="admin_edit_no_password_target@example.org",
+        username="admin_edit_no_password_target",
+        roles=[role],
+    )
+    db.session.add(user)
+    db.session.commit()
+
+    api = SupersetUserApi()
+    api.datamodel = SQLAInterface(User, db.session)
+    api.appbuilder = SimpleNamespace(
+        sm=SimpleNamespace(current_user=SimpleNamespace(id=1))
+    )
+
+    api.pre_update(user, {"first_name": "Renamed"})
+
+    assert _invalidated_at(user.id) is None
+
+    db.session.query(User).filter_by(id=user.id).delete(synchronize_session=False)
+    db.session.commit()
+
+
+def _make_api_for_target(user: User) -> SupersetUserApi:
+    """A ``SupersetUserApi`` instance wired to a fake ``datamodel`` that
+    resolves any pk lookup to ``user`` -- enough to exercise
+    ``terminate_sessions`` without going through HTTP/auth plumbing, mirroring
+    the pattern used in ``test_superset_user_api_subject_sync.py``.
+    """
+    api = SupersetUserApi()
+    api.datamodel = SimpleNamespace(
+        session=db.session,
+        obj=User,
+        get=lambda pk, base_filters=None: user,
+    )
+    api._base_filters = None
+    return api
+
+
+def test_terminate_sessions_action_stamps_target_epoch_without_disabling_account(
+    after_each: None,  # noqa: F811
+) -> None:
+    """``SupersetUserApi.terminate_sessions`` -- the direct, explicit
+    "terminate this user's sessions" admin action -- stamps the epoch for the
+    target user without flipping ``active`` or otherwise touching the account,
+    unlike the only other action that has this effect (disabling the user).
+    """
+    role = db.session.query(security_manager.role_model).filter_by(name="Admin").one()
+    user = User(
+        first_name="Target",
+        last_name="User",
+        email="terminate_sessions_target@example.org",
+        username="terminate_sessions_target",
+        roles=[role],
+    )
+    db.session.add(user)
+    db.session.flush()
+
+    with patch.object(security_manager, "has_access", return_value=True):
+        response = _make_api_for_target(user).terminate_sessions(user.id)
+
+    assert response.status_code == 200
+    assert _invalidated_at(user.id) is not None
+    assert user.active
+
+
+def test_terminate_sessions_action_404s_for_unknown_user(
+    after_each: None,  # noqa: F811
+) -> None:
+    """A pk that doesn't resolve to a user (or is filtered out by
+    ``base_filters``) 404s rather than stamping anything.
+    """
+    api = SupersetUserApi()
+    api.datamodel = SimpleNamespace(
+        session=db.session,
+        obj=User,
+        get=lambda pk, base_filters=None: None,
+    )
+    api._base_filters = None
+
+    with patch.object(security_manager, "has_access", return_value=True):
+        response = api.terminate_sessions(999999)
+
+    assert response.status_code == 404

tests/unit_tests/security/test_superset_user_api_subject_sync.py

@@ -0,0 +1,229 @@
+# 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.
+"""``SupersetUserApi`` syncs the ``Subject`` row that mirrors each ``User``.
+
+``UserApi.post``/``put`` call ``self.pre_add``/``self.pre_update`` *before*
+the write that actually commits (``self.datamodel.add``/``edit``). Overriding
+those hooks -- instead of syncing after the fact, in a second commit issued
+once ``post``/``put`` have already returned -- means the subject sync rides
+the same transaction as the user write: one commit persists both, and a
+failure of that commit rolls both back together instead of leaving an
+orphaned user with no matching ``Subject`` row.
+
+These tests exercise ``pre_add``/``pre_update`` directly, plus the exact call
+pairs FAB's ``UserApi.post``/``put`` make (``pre_add``/``pre_update`` followed
+by the real ``SQLAInterface.add``/``edit``), to confirm that pairing shares a
+single commit and a single rollback.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from flask_appbuilder.models.sqla.interface import SQLAInterface
+from flask_appbuilder.security.sqla.models import User
+
+from superset import db, security_manager
+from superset.security.manager import SupersetUserApi
+from superset.subjects.models import Subject
+from superset.subjects.types import SubjectType
+from tests.unit_tests.fixtures.common import after_each  # noqa: F401
+
+
+def _admin_role() -> object:
+    return db.session.query(security_manager.role_model).filter_by(name="Admin").one()
+
+
+def _make_pending_user(username: str) -> User:
+    """A transient ``User``, not yet added to the session.
+
+    Mirrors what FAB's ``UserApi.post()`` builds (``model = User()``, with
+    attributes set from the request payload) just before it calls
+    ``self.pre_add(model)`` and then ``self.datamodel.add(model)``.
+    """
+    return User(
+        first_name="New",
+        last_name="Guy",
+        email=f"{username}@example.org",
+        username=username,
+        roles=[_admin_role()],
+        password="irrelevant-pre-hash-value",  # noqa: S106
+    )
+
+
+def _make_persisted_user(username: str) -> User:
+    """A ``User`` already committed, standing in for one an earlier request
+    created -- the starting point for a ``put``/``pre_update`` flow.
+    """
+    user = User(
+        first_name="New",
+        last_name="Guy",
+        email=f"{username}@example.org",
+        username=username,
+        roles=[_admin_role()],
+    )
+    db.session.add(user)
+    db.session.commit()
+    return user
+
+
+def _api_for(session=db.session) -> SupersetUserApi:  # noqa: ANN001
+    api = SupersetUserApi()
+    api.datamodel = SQLAInterface(User, session)
+    # FAB's own ``pre_update`` (which ``SupersetUserApi.pre_update`` calls via
+    # ``super()``) reads ``self.appbuilder.sm.current_user.id`` to stamp
+    # ``changed_by_fk`` -- stand in for the acting admin so that lookup
+    # succeeds outside of a real request/login context.
+    api.appbuilder = SimpleNamespace(
+        sm=SimpleNamespace(current_user=SimpleNamespace(id=1))
+    )
+    return api
+
+
+def _subject_for(user_id: int) -> Subject | None:
+    return (
+        db.session.query(Subject)
+        .filter_by(user_id=user_id, type=SubjectType.USER)
+        .one_or_none()
+    )
+
+
+def test_pre_add_syncs_subject_without_committing(
+    after_each: None,  # noqa: F811
+) -> None:
+    """``pre_add`` flushes the new user (to obtain its id) and syncs its
+    ``Subject`` row, but does not commit -- that's still FAB's job, in
+    ``self.datamodel.add``, which runs right after.
+    """
+    user = _make_pending_user("new_guy_pre_add")
+    api = _api_for()
+
+    real_commit = db.session.commit
+    commit_calls = MagicMock(wraps=real_commit)
+    with patch.object(db.session, "commit", commit_calls):
+        api.pre_add(user)
+
+    assert commit_calls.call_count == 0
+    assert user.id is not None
+    assert _subject_for(user.id) is not None
+
+
+def test_pre_add_and_datamodel_add_share_a_single_commit(
+    after_each: None,  # noqa: F811
+) -> None:
+    """The exact pair of calls FAB's ``UserApi.post()`` makes --
+    ``self.pre_add(model)`` then ``self.datamodel.add(model)`` -- persist the
+    user and its ``Subject`` row together, via exactly one commit.
+    """
+    user = _make_pending_user("new_guy_shared_commit")
+    api = _api_for()
+
+    real_commit = db.session.commit
+    commit_calls = MagicMock(wraps=real_commit)
+    with patch.object(db.session, "commit", commit_calls):
+        api.pre_add(user)
+        api.datamodel.add(user)
+
+    assert commit_calls.call_count == 1
+    subject = _subject_for(user.id)
+    assert subject is not None
+
+
+def test_pre_add_failure_rolls_back_user_and_subject_together(
+    after_each: None,  # noqa: F811
+) -> None:
+    """If the commit that follows ``pre_add`` fails (standing in: FAB's own
+    ``self.datamodel.add`` raising), the new user and the ``Subject`` row
+    flushed alongside it roll back together -- there is no window where the
+    user persists without a matching ``Subject``.
+    """
+    user = _make_pending_user("new_guy_pre_add_fail")
+    api = _api_for()
+
+    api.pre_add(user)
+    user_id = user.id
+    assert user_id is not None
+    assert _subject_for(user_id) is not None  # flushed, visible pre-rollback
+
+    db.session.rollback()  # stands in for the follow-up commit failing
+
+    assert db.session.query(User).filter_by(id=user_id).one_or_none() is None
+    assert _subject_for(user_id) is None
+
+
+def test_pre_update_syncs_subject_without_committing(
+    after_each: None,  # noqa: F811
+) -> None:
+    """Same reasoning as ``pre_add``, for an edit: ``pre_update`` syncs the
+    ``Subject`` row without committing, ahead of FAB's own
+    ``self.datamodel.edit``.
+    """
+    user = _make_persisted_user("new_guy_pre_update")
+    api = _api_for()
+
+    real_commit = db.session.commit
+    commit_calls = MagicMock(wraps=real_commit)
+    with patch.object(db.session, "commit", commit_calls):
+        api.pre_update(user, {})
+
+    assert commit_calls.call_count == 0
+    assert _subject_for(user.id) is not None
+
+
+def test_pre_update_and_datamodel_edit_share_a_single_commit(
+    after_each: None,  # noqa: F811
+) -> None:
+    """The exact pair of calls FAB's ``UserApi.put()`` makes --
+    ``self.pre_update(model, item)`` then ``self.datamodel.edit(model)`` --
+    persist the edit and the ``Subject`` sync together, via exactly one
+    commit.
+    """
+    user = _make_persisted_user("new_guy_shared_commit_put")
+    api = _api_for()
+
+    real_commit = db.session.commit
+    commit_calls = MagicMock(wraps=real_commit)
+    with patch.object(db.session, "commit", commit_calls):
+        api.pre_update(user, {})
+        api.datamodel.edit(user)
+
+    assert commit_calls.call_count == 1
+    assert _subject_for(user.id) is not None
+
+
+def test_pre_update_failure_rolls_back_subject_sync_without_orphaning(
+    after_each: None,  # noqa: F811
+) -> None:
+    """If the commit that follows ``pre_update`` fails, the ``Subject`` row it
+    flushed rolls back too -- the previously-persisted user row (created by
+    an earlier, already-successful request) is left exactly as it was, with
+    no half-applied sync attached to it.
+    """
+    user = _make_persisted_user("new_guy_pre_update_fail")
+    user_id = user.id
+    api = _api_for()
+
+    api.pre_update(user, {})
+    assert _subject_for(user_id) is not None  # flushed, visible pre-rollback
+
+    db.session.rollback()  # stands in for the follow-up commit failing
+
+    # The user itself predates this (failed) request and survives.
+    assert db.session.query(User).filter_by(id=user_id).one_or_none() is not None
+    # But the subject sync this request attempted never landed.
+    assert _subject_for(user_id) is None

tests/unit_tests/views/datasource/utils_test.py

@@ -19,9 +19,11 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from flask import current_app
 
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.exceptions import SupersetSecurityException
+from superset.views.datasource.utils import get_limit_clause
 
 
 @patch("superset.views.datasource.utils.get_limit_clause")
@@ -215,3 +217,62 @@ def test_get_samples_count_star_access_denied(mock_get_limit_clause: MagicMock):
         mock_samples_context.raise_for_access.assert_called_once()
         # Verify count context was also checked
         mock_count_context.raise_for_access.assert_called_once()
+
+
+@pytest.mark.parametrize("per_page", [5000, 10000])
+def test_get_limit_clause_honors_per_page_above_samples_row_limit(
+    per_page: int,
+) -> None:
+    """Regression guard: the Explore Data panel "Samples" tab silently caps at
+    ``SAMPLES_ROW_LIMIT`` (config default 1000).
+
+    The samples row-limit dropdown offers 5k/10k options and the samples
+    endpoint's ``SamplesRequestSchema`` accepts ``per_page`` up to 10000, yet
+    ``get_limit_clause`` resets any ``per_page`` above ``SAMPLES_ROW_LIMIT``
+    back down to it. A user who selects 5k/10k therefore silently receives at
+    most 1000 rows, with no signal that the requested limit was overridden.
+
+    The rows a user is allowed to request and the rows actually returned must
+    stay consistent: a ``per_page`` the endpoint accepts must not be silently
+    reduced below the request.
+    """
+    assert get_limit_clause(page=1, per_page=per_page) == {
+        "row_offset": 0,
+        "row_limit": per_page,
+    }
+
+
+@pytest.mark.parametrize(
+    "per_page,expected_row_limit",
+    [
+        (0, 0),
+        (-1, 1000),
+    ],
+)
+def test_get_limit_clause_preserves_zero_and_negative_per_page(
+    per_page: int,
+    expected_row_limit: int,
+) -> None:
+    assert get_limit_clause(page=1, per_page=per_page) == {
+        "row_offset": 0,
+        "row_limit": expected_row_limit,
+    }
+
+
+def test_get_limit_clause_caps_per_page_at_sql_max_row(
+    app_context: None,
+) -> None:
+    """When an operator configures ``SQL_MAX_ROW`` below the schema's
+    ``per_page`` maximum, ``apply_max_row_limit`` still reduces the
+    requested limit, and the offset for subsequent pages must be computed
+    from that reduced (effective) limit, not the raw request.
+    """
+    with patch.dict(current_app.config, {"SQL_MAX_ROW": 2000}):
+        assert get_limit_clause(page=1, per_page=10000) == {
+            "row_offset": 0,
+            "row_limit": 2000,
+        }
+        assert get_limit_clause(page=2, per_page=10000) == {
+            "row_offset": 2000,
+            "row_limit": 2000,
+        }

tests/unit_tests/views/test_current_user_api.py

@@ -0,0 +1,160 @@
+# 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.
+"""Unit tests for the ``CurrentUserRestApi`` self-service update flow.
+
+Covers the ``password`` handling in ``PUT /api/v1/me/``: whether the caller
+must prove knowledge of the existing password, and whether the value that
+ends up persisted is the hash computed in ``pre_update`` or the raw value
+from the request payload.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+from flask_appbuilder.security.sqla.models import User
+from marshmallow import ValidationError
+from werkzeug.security import check_password_hash, generate_password_hash
+
+from superset import db
+from superset.daos.user import UserDAO
+from superset.views.users.api import CurrentUserRestApi
+from superset.views.users.schemas import CurrentUserPutSchema
+from tests.unit_tests.fixtures.common import admin_user, after_each  # noqa: F401
+
+
+def _run_update_me(user: User, data: dict[str, Any]) -> None:
+    """Reproduce the body of ``CurrentUserRestApi.update_me`` for ``data``.
+
+    Exercises the same two calls the endpoint makes -- ``pre_update`` followed
+    by ``UserDAO.update`` with the schema-loaded payload as ``attributes`` --
+    without going through HTTP/auth plumbing, since neither call depends on it.
+    """
+    api = CurrentUserRestApi()
+    with patch("superset.views.users.api.g") as mock_g:
+        mock_g.user = user
+        api.pre_update(user, data)
+        UserDAO.update(item=user, attributes=data)
+
+
+def test_current_user_put_schema_has_current_password_field() -> None:
+    """The payload schema for ``PUT /api/v1/me/`` carries a field for proving
+    knowledge of the existing password, required whenever ``password`` is
+    supplied.
+    """
+    schema = CurrentUserPutSchema()
+
+    assert "password" in schema.fields
+    assert "current_password" in schema.fields
+
+    with pytest.raises(ValidationError):
+        schema.load({"password": "BrandNewPassw0rd!"})
+
+    # Present alongside "password", it loads fine (the schema only checks
+    # that it was *supplied*; whether it's actually correct is verified
+    # against the database in ``CurrentUserRestApi.pre_update``).
+    loaded = schema.load(
+        {"password": "BrandNewPassw0rd!", "current_password": "OldPassw0rd!"}
+    )
+    assert loaded["current_password"] == "OldPassw0rd!"  # noqa: S105
+
+
+def test_update_me_rejects_password_change_without_correct_current_password(
+    admin_user: User,  # noqa: F811
+    after_each: None,  # noqa: F811
+) -> None:
+    """A caller can no longer change the password by supplying only the new
+    value: an account with an existing password must prove knowledge of it
+    via ``current_password`` -- omitting the field, or getting it wrong, both
+    reject the change and leave the stored password untouched. Supplying the
+    correct current password lets the change through.
+    """
+    original_hash = generate_password_hash("OldPassw0rd!")
+    admin_user.password = original_hash
+
+    with pytest.raises(ValidationError):
+        _run_update_me(admin_user, {"password": "BrandNewPassw0rd!"})
+    assert admin_user.password == original_hash
+
+    with pytest.raises(ValidationError):
+        _run_update_me(
+            admin_user,
+            {
+                "password": "BrandNewPassw0rd!",
+                "current_password": "WrongPassw0rd!",
+            },
+        )
+    assert admin_user.password == original_hash
+
+    _run_update_me(
+        admin_user,
+        {"password": "BrandNewPassw0rd!", "current_password": "OldPassw0rd!"},
+    )
+    db.session.flush()
+
+    assert admin_user.password != original_hash
+    assert check_password_hash(admin_user.password, "BrandNewPassw0rd!")
+
+
+def test_update_me_password_change_persists_a_hash_not_plaintext(
+    admin_user: User,  # noqa: F811
+    after_each: None,  # noqa: F811
+) -> None:
+    """``pre_update`` computes ``generate_password_hash(data["password"])`` and
+    assigns it to the user, and ``UserDAO.update`` is then called with the
+    *same* ``data`` dict as ``attributes``. ``BaseDAO.update`` blindly
+    ``setattr``s every key in ``attributes``, so ``pre_update`` must remove
+    the plaintext ``password`` key (and any ``current_password``) from that
+    dict once it's done with them, or the plaintext would overwrite the hash
+    that was just computed and reach the database instead of it.
+
+    ``admin_user`` starts with no password set, so this exercises the
+    first-password-set path, which requires no proof of a prior password.
+    """
+    new_password = "BrandNewPassw0rd!"  # noqa: S105
+
+    _run_update_me(admin_user, {"password": new_password})
+    db.session.flush()
+
+    stored_password = admin_user.password
+
+    # The stored value should be a password hash that verifies against the
+    # new password -- not the plaintext value itself.
+    assert stored_password != new_password
+    assert check_password_hash(stored_password, new_password)
+
+
+def test_update_me_falsy_password_does_not_blank_stored_hash(
+    admin_user: User,  # noqa: F811
+    after_each: None,  # noqa: F811
+) -> None:
+    """A falsy ``password`` (e.g. an empty string, which the schema's
+    complexity validator lets through when password complexity validation is
+    disabled) skips the hashing branch entirely -- ``pre_update`` must still
+    drop the key from ``data`` so it never reaches ``UserDAO.update``'s
+    ``setattr`` loop and blanks the account's stored hash.
+    """
+    original_hash = generate_password_hash("OldPassw0rd!")
+    admin_user.password = original_hash
+
+    _run_update_me(admin_user, {"password": "", "first_name": "Foo"})
+    db.session.flush()
+
+    assert admin_user.password == original_hash
+    assert admin_user.first_name == "Foo"