Rotate session id during login (#25771)

(cherry picked from commit 9a034e6)

airflow/www/fab_security/manager.py

@@ -25,6 +25,7 @@
 import logging
 import re
 from typing import Any, Dict, List, Optional, Set, Tuple, Type
+from uuid import uuid4
 
 from flask import current_app, g, session, url_for
 from flask_appbuilder import AppBuilder
@@ -70,6 +71,7 @@
 from flask_login import AnonymousUserMixin, LoginManager, current_user
 from werkzeug.security import check_password_hash, generate_password_hash
 
+from airflow.configuration import conf
 from airflow.www.fab_security.sqla.models import Action, Permission, RegisterUser, Resource, Role, User
 from airflow.www.views import ResourceModelView
 
@@ -854,6 +856,14 @@ def update_user_auth_stat(self, user, success=True):
             user.fail_login_count += 1
         self.update_user(user)
 
+    def _rotate_session_id(self):
+        """
+        Upon successful authentication when using the database session backend,
+        we need to rotate the session id
+        """
+        if conf.get('webserver', 'SESSION_BACKEND') == 'database':
+            session.sid = str(uuid4())
+
     def auth_user_db(self, username, password):
         """
         Method for authenticating user, auth db style
@@ -878,6 +888,7 @@ def auth_user_db(self, username, password):
             log.info(LOGMSG_WAR_SEC_LOGIN_FAILED.format(username))
             return None
         elif check_password_hash(user.password, password):
+            self._rotate_session_id()
             self.update_user_auth_stat(user, True)
             return user
         else:
@@ -1173,6 +1184,7 @@ def auth_user_ldap(self, username, password):
 
             # LOGIN SUCCESS (only if user is now registered)
             if user:
+                self._rotate_session_id()
                 self.update_user_auth_stat(user)
                 return user
             else:
@@ -1200,6 +1212,7 @@ def auth_user_oid(self, email):
             log.info(LOGMSG_WAR_SEC_LOGIN_FAILED.format(email))
             return None
         else:
+            self._rotate_session_id()
             self.update_user_auth_stat(user)
             return user
 
@@ -1229,6 +1242,7 @@ def auth_user_remote_user(self, username):
             log.info(LOGMSG_WAR_SEC_LOGIN_FAILED.format(username))
             return None
 
+        self._rotate_session_id()
         self.update_user_auth_stat(user)
         return user
 
@@ -1314,6 +1328,7 @@ def auth_user_oauth(self, userinfo):
 
         # LOGIN SUCCESS (only if user is now registered)
         if user:
+            self._rotate_session_id()
             self.update_user_auth_stat(user)
             return user
         else:

tests/www/views/test_session.py

@@ -15,6 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 
+from unittest import mock
+
 import pytest
 
 from airflow.exceptions import AirflowConfigException
@@ -23,12 +25,16 @@
 from tests.test_utils.decorators import dont_initialize_flask_app_submodules
 
 
+def get_session_cookie(client):
+    return next((cookie for cookie in client.cookie_jar if cookie.name == 'session'), None)
+
+
 def test_session_cookie_created_on_login(user_client):
-    assert any(cookie.name == 'session' for cookie in user_client.cookie_jar)
+    assert get_session_cookie(user_client) is not None
 
 
 def test_session_inaccessible_after_logout(user_client):
-    session_cookie = next((cookie for cookie in user_client.cookie_jar if cookie.name == 'session'), None)
+    session_cookie = get_session_cookie(user_client)
     assert session_cookie is not None
 
     resp = user_client.get('/logout/')
@@ -63,3 +69,21 @@ def poorly_configured_app_factory():
     )
     with pytest.raises(AirflowConfigException, match=expected_exc_regex):
         poorly_configured_app_factory()
+
+
+def test_session_id_rotates(app, user_client):
+    old_session_cookie = get_session_cookie(user_client)
+    assert old_session_cookie is not None
+
+    resp = user_client.get('/logout/')
+    assert resp.status_code == 302
+
+    patch_path = "airflow.www.fab_security.manager.check_password_hash"
+    with mock.patch(patch_path) as check_password_hash:
+        check_password_hash.return_value = True
+        resp = user_client.post("/login/", data={"username": "test_user", "password": "test_user"})
+    assert resp.status_code == 302
+
+    new_session_cookie = get_session_cookie(user_client)
+    assert new_session_cookie is not None
+    assert old_session_cookie.value != new_session_cookie.value