Reject tokens naming a deactivated account in the FAB auth manager (#72199)

deserialize_user resolved the token subject by id alone, so a bearer issued
before an account was deactivated continued to resolve to that user. The
password path already refuses an inactive account in auth_user_db; the token
path did not. The account state is now re-checked when the user is loaded,
and a null 'active' column is treated as inactive to match auth_user_db.

The check runs when the user is loaded, so it is bounded by the existing
[fab] cache_ttl window (30s by default) rather than being immediate.

test_is_logged_in_with_inactive_user set is_active via return_value, but
is_active is a property, so the mock stayed truthy and the assertion held
regardless of the state under test. It now sets the attribute, and an
active-user counterpart was added alongside it.

Author: potiuk

providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py

@@ -292,9 +292,17 @@ def deserialize_user(self, token: dict[str, Any]) -> User:
         def _fetch_user() -> User:
             with create_session() as session:
                 try:
-                    return session.scalars(select(User).where(User.id == user_id)).one()
+                    user = session.scalars(select(User).where(User.id == user_id)).one()
                 except NoResultFound:
                     raise ValueError(f"User with id {token['sub']} not found")
+                # A token stays syntactically valid until it expires, so the account it
+                # names has to be re-checked on every request rather than trusted from
+                # the signature alone. ``is_active`` reads the nullable ``active``
+                # column, and a null is treated as inactive here for the same reason it
+                # is on the password path in ``auth_user_db``.
+                if not user.is_active:
+                    raise ValueError(f"User with id {token['sub']} is not active")
+                return user
 
         try:
             return _fetch_user()

providers/fab/tests/unit/fab/auth_manager/test_fab_auth_manager.py

@@ -221,6 +221,36 @@ def test_deserialize_user(self, flask_app, auth_manager_with_appbuilder):
 
         assert user.get_id() == result.get_id()
 
+    def test_deserialize_user_rejects_inactive_user(self, flask_app, auth_manager_with_appbuilder):
+        """A token naming a deactivated account must not resolve to a user."""
+        user = create_user(flask_app, "test_inactive")
+        auth_manager_with_appbuilder.cache.clear()
+
+        user.active = False
+        auth_manager_with_appbuilder.session.commit()
+        auth_manager_with_appbuilder.cache.clear()
+
+        with pytest.raises(ValueError, match=f"User with id {user.id} is not active"):
+            auth_manager_with_appbuilder.deserialize_user({"sub": str(user.id)})
+
+    def test_deserialize_user_rejects_null_active(self, flask_app, auth_manager_with_appbuilder):
+        """``active`` is nullable; a null is treated as inactive, as on the password path."""
+        user = create_user(flask_app, "test_null_active")
+        user.active = None
+        auth_manager_with_appbuilder.session.commit()
+        auth_manager_with_appbuilder.cache.clear()
+
+        with pytest.raises(ValueError, match=f"User with id {user.id} is not active"):
+            auth_manager_with_appbuilder.deserialize_user({"sub": str(user.id)})
+
+    def test_deserialize_user_accepts_active_user(self, flask_app, auth_manager_with_appbuilder):
+        user = create_user(flask_app, "test_still_active")
+        auth_manager_with_appbuilder.cache.clear()
+
+        result = auth_manager_with_appbuilder.deserialize_user({"sub": str(user.id)})
+
+        assert result.get_id() == user.get_id()
+
     def test_deserialize_user_not_found(self, flask_app, auth_manager_with_appbuilder):
         """Test that deserialize_user raises ValueError when the user does not exist."""
         non_existent_id = "99999"
@@ -256,13 +286,25 @@ def test_is_logged_in(self, mock_get_user, auth_manager_with_appbuilder):
 
     @mock.patch.object(FabAuthManager, "get_user")
     def test_is_logged_in_with_inactive_user(self, mock_get_user, auth_manager_with_appbuilder):
+        # ``is_anonymous`` and ``is_active`` are properties on the real model, so the
+        # mock has to set attributes rather than ``return_value``; setting the latter
+        # leaves a truthy Mock in place and the assertion passes regardless of state.
         user = Mock()
-        user.is_anonymous.return_value = False
-        user.is_active.return_value = True
+        user.is_anonymous = False
+        user.is_active = False
         mock_get_user.return_value = user
 
         assert auth_manager_with_appbuilder.is_logged_in() is False
 
+    @mock.patch.object(FabAuthManager, "get_user")
+    def test_is_logged_in_with_active_user(self, mock_get_user, auth_manager_with_appbuilder):
+        user = Mock()
+        user.is_anonymous = False
+        user.is_active = True
+        mock_get_user.return_value = user
+
+        assert auth_manager_with_appbuilder.is_logged_in() is True
+
     @mock.patch.object(FabAuthManager, "get_user")
     def test_is_logged_in_with_auth_role_public(self, mock_get_user, flask_app, auth_manager_with_appbuilder):
         """When ``AUTH_ROLE_PUBLIC`` is set on the Flask app, anonymous users are 'logged in'."""