[v3-1-test] Logout the user when the refresh token is no longer valid (#60781)

airflow-core/src/airflow/api_fastapi/auth/managers/exceptions.py

@@ -0,0 +1,22 @@
+#
+# 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.
+from __future__ import annotations
+
+
+class AuthManagerRefreshTokenExpiredException(Exception):
+    """Exception to throw when the user refresh token is expired."""

airflow-core/src/airflow/api_fastapi/auth/middlewares/refresh_token.py

@@ -23,6 +23,7 @@
 
 from airflow.api_fastapi.app import get_auth_manager
 from airflow.api_fastapi.auth.managers.base_auth_manager import COOKIE_NAME_JWT_TOKEN
+from airflow.api_fastapi.auth.managers.exceptions import AuthManagerRefreshTokenExpiredException
 from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
 from airflow.api_fastapi.core_api.security import resolve_user_from_token
 from airflow.configuration import conf
@@ -40,26 +41,34 @@ class JWTRefreshMiddleware(BaseHTTPMiddleware):
     """
 
     async def dispatch(self, request: Request, call_next):
-        new_user = None
+        new_token = None
         current_token = request.cookies.get(COOKIE_NAME_JWT_TOKEN)
         try:
-            if current_token:
-                new_user, current_user = await self._refresh_user(current_token)
-                if user := (new_user or current_user):
-                    request.state.user = user
+            if current_token is not None:
+                try:
+                    new_user, current_user = await self._refresh_user(current_token)
+                    if user := (new_user or current_user):
+                        request.state.user = user
+                    if new_user:
+                        # If we created a new user, serialize it and set it as a cookie
+                        new_token = get_auth_manager().generate_jwt(new_user)
+                except (HTTPException, AuthManagerRefreshTokenExpiredException):
+                    # Receive a HTTPException when the Airflow token is expired
+                    # Receive a AuthManagerRefreshTokenExpiredException when the potential underlying refresh
+                    # token used by the auth manager is expired
+                    new_token = ""
 
             response = await call_next(request)
 
-            if new_user:
-                # If we created a new user, serialize it and set it as a cookie
-                new_token = get_auth_manager().generate_jwt(new_user)
+            if new_token is not None:
                 secure = bool(conf.get("api", "ssl_cert", fallback=""))
                 response.set_cookie(
                     COOKIE_NAME_JWT_TOKEN,
                     new_token,
                     httponly=True,
                     secure=secure,
                     samesite="lax",
+                    max_age=0 if new_token == "" else None,
                 )
         except HTTPException as exc:
             # If any HTTPException is raised during user resolution or refresh, return it as response
@@ -68,9 +77,5 @@ async def dispatch(self, request: Request, call_next):
 
     @staticmethod
     async def _refresh_user(current_token: str) -> tuple[BaseUser | None, BaseUser | None]:
-        try:
-            user = await resolve_user_from_token(current_token)
-        except HTTPException:
-            return None, None
-
+        user = await resolve_user_from_token(current_token)
         return get_auth_manager().refresh_user(user=user), user

airflow-core/tests/unit/api_fastapi/auth/middlewares/test_refresh_token.py

@@ -61,11 +61,11 @@ async def test_dispatch_no_token(self, mock_refresh_user, middleware, mock_reque
     @pytest.mark.asyncio
     async def test_dispatch_invalid_token(self, mock_refresh_user, middleware, mock_request):
         mock_request.cookies = {COOKIE_NAME_JWT_TOKEN: "valid_token"}
-        call_next = AsyncMock(return_value=Response())
+        call_next = AsyncMock(return_value=Response(status_code=401))
 
         response = await middleware.dispatch(mock_request, call_next)
-        assert response.status_code == 403
-        assert response.body == b'{"detail":"Invalid JWT token"}'
+        assert response.status_code == 401
+        assert '_token=""; HttpOnly; Max-Age=0; Path=/; SameSite=lax' in response.headers.get("set-cookie")
 
     @patch("airflow.api_fastapi.auth.middlewares.refresh_token.get_auth_manager")
     @patch("airflow.api_fastapi.auth.middlewares.refresh_token.resolve_user_from_token")