Add FAB option to log users out after a maximum session lifetime

providers/fab/provider.yaml

@@ -222,11 +222,38 @@ config:
       session_lifetime_minutes:
         description: |
           The UI cookie lifetime in minutes. User will be logged out from UI after
-          ``[fab] session_lifetime_minutes`` of non-activity
+          ``[fab] session_lifetime_minutes`` of inactivity: the deadline slides forward on every
+          request, so it is only reached once the session has been idle for the whole period.
+
+          Note that leaving an Airflow UI tab open counts as activity even when nobody is at the
+          keyboard. The UI polls the API in the background and silently re-authenticates whenever
+          its API token expires, which keeps sliding the deadline, so a session with an open tab is
+          never idle and never expires. Use ``[fab] session_max_lifetime_minutes`` to log users out
+          after a fixed period regardless of activity.
         version_added: 2.0.0
         type: integer
         example: ~
         default: "43200"
+      session_max_lifetime_minutes:
+        description: |
+          Maximum lifetime of a UI session in minutes, counted from the login time and never
+          extended by activity. Unlike ``[fab] session_lifetime_minutes``, this deadline is
+          reached even when the user keeps working in the UI, so it forces periodic
+          re-authentication. Set to ``0`` (the default) to disable it.
+
+          The API tokens the UI receives are capped so that they never outlive the deadline: their
+          expiry is the shorter of ``[api_auth] jwt_expiration_time`` and the time left in the
+          session. Without that cap the deadline would only be noticed the next time the UI came
+          back to the auth manager — which it does when its token expires — and an already-issued
+          token would keep working against the API in the meantime.
+
+          This applies to UI sessions only. Tokens minted for programmatic clients by
+          ``POST /auth/token`` belong to no session and always last ``[api_auth]
+          jwt_expiration_time``.
+        version_added: 3.9.0
+        type: integer
+        example: "480"
+        default: "0"
       enable_proxy_fix:
         description: |
           Enable werkzeug ``ProxyFix`` middleware for reverse proxy

providers/fab/src/airflow/providers/fab/get_provider_info.py

@@ -136,12 +136,19 @@ def get_provider_info():
                         "default": "database",
                     },
                     "session_lifetime_minutes": {
-                        "description": "The UI cookie lifetime in minutes. User will be logged out from UI after\n``[fab] session_lifetime_minutes`` of non-activity\n",
+                        "description": "The UI cookie lifetime in minutes. User will be logged out from UI after\n``[fab] session_lifetime_minutes`` of inactivity: the deadline slides forward on every\nrequest, so it is only reached once the session has been idle for the whole period.\n\nNote that leaving an Airflow UI tab open counts as activity even when nobody is at the\nkeyboard. The UI polls the API in the background and silently re-authenticates whenever\nits API token expires, which keeps sliding the deadline, so a session with an open tab is\nnever idle and never expires. Use ``[fab] session_max_lifetime_minutes`` to log users out\nafter a fixed period regardless of activity.\n",
                         "version_added": "2.0.0",
                         "type": "integer",
                         "example": None,
                         "default": "43200",
                     },
+                    "session_max_lifetime_minutes": {
+                        "description": "Maximum lifetime of a UI session in minutes, counted from the login time and never\nextended by activity. Unlike ``[fab] session_lifetime_minutes``, this deadline is\nreached even when the user keeps working in the UI, so it forces periodic\nre-authentication. Set to ``0`` (the default) to disable it.\n\nThe API tokens the UI receives are capped so that they never outlive the deadline: their\nexpiry is the shorter of ``[api_auth] jwt_expiration_time`` and the time left in the\nsession. Without that cap the deadline would only be noticed the next time the UI came\nback to the auth manager — which it does when its token expires — and an already-issued\ntoken would keep working against the API in the meantime.\n\nThis applies to UI sessions only. Tokens minted for programmatic clients by\n``POST /auth/token`` belong to no session and always last ``[api_auth]\njwt_expiration_time``.\n",
+                        "version_added": "3.9.0",
+                        "type": "integer",
+                        "example": "480",
+                        "default": "0",
+                    },
                     "enable_proxy_fix": {
                         "description": "Enable werkzeug ``ProxyFix`` middleware for reverse proxy\n",
                         "version_added": "2.1.0",

providers/fab/src/airflow/providers/fab/www/app.py

@@ -35,7 +35,10 @@
 from airflow.providers.fab.www.extensions.init_jinja_globals import init_jinja_globals
 from airflow.providers.fab.www.extensions.init_manifest_files import configure_manifest_files
 from airflow.providers.fab.www.extensions.init_security import init_api_auth
-from airflow.providers.fab.www.extensions.init_session import init_airflow_session_interface
+from airflow.providers.fab.www.extensions.init_session import (
+    init_airflow_session_interface,
+    init_session_max_lifetime,
+)
 from airflow.providers.fab.www.extensions.init_views import (
     init_error_handlers,
     init_plugins,
@@ -127,6 +130,7 @@ def remove_duplicate_date_header(response):
             init_plugins(flask_app)
         elif isinstance(get_auth_manager(), FabAuthManager):
             init_airflow_session_interface(flask_app, db)
+            init_session_max_lifetime(flask_app)
         init_jinja_globals(flask_app, enable_plugins=enable_plugins)
         init_wsgi_middleware(flask_app)
     return flask_app

providers/fab/src/airflow/providers/fab/www/extensions/init_session.py

@@ -16,7 +16,11 @@
 # under the License.
 from __future__ import annotations
 
+import logging
+import time
+
 from flask import session as builtin_flask_session
+from flask_login import current_user, logout_user, user_logged_in
 
 from airflow.exceptions import AirflowConfigException
 from airflow.providers.common.compat.sdk import conf
@@ -25,6 +29,10 @@
     AirflowSecureCookieSessionInterface,
 )
 
+log = logging.getLogger(__name__)
+
+SESSION_LOGIN_TIME_KEY = "_login_at"
+
 
 def init_airflow_session_interface(app, db):
     """Set airflow session interface."""
@@ -62,3 +70,54 @@ def make_session_permanent():
             f"[fab] session_backend: '{selected_backend}'. Please set "
             "this to either 'database' or 'securecookie'."
         )
+
+
+def get_max_session_lifetime_seconds() -> int:
+    """Return ``[fab] session_max_lifetime_minutes`` in seconds, or ``0`` when the cap is disabled."""
+    return max(conf.getint("fab", "session_max_lifetime_minutes", fallback=0), 0) * 60
+
+
+def get_remaining_session_lifetime() -> float | None:
+    """
+    Return how many seconds are left before the current session hits its maximum lifetime.
+
+    ``None`` means the session is not capped, either because ``[fab]
+    session_max_lifetime_minutes`` is disabled or because the session carries no login stamp.
+    """
+    max_lifetime_seconds = get_max_session_lifetime_seconds()
+    if not max_lifetime_seconds:
+        return None
+    login_time = builtin_flask_session.get(SESSION_LOGIN_TIME_KEY)
+    if login_time is None:
+        return None
+    return login_time + max_lifetime_seconds - time.time()
+
+
+def init_session_max_lifetime(app):
+    """Expire sessions ``[fab] session_max_lifetime_minutes`` after login, regardless of activity."""
+    if not get_max_session_lifetime_seconds():
+        return
+
+    # ``weak=False``: the receiver is a local function, so a weak subscription could be collected.
+    @user_logged_in.connect_via(app, weak=False)
+    def stamp_login_time(sender, user, **kwargs):
+        # Wall clock rather than ``time.monotonic()``: the stamp is persisted in the session and
+        # read back by other API server processes, which share no monotonic clock origin.
+        builtin_flask_session[SESSION_LOGIN_TIME_KEY] = time.time()
+
+    @app.before_request
+    def expire_session_past_max_lifetime():
+        if SESSION_LOGIN_TIME_KEY not in builtin_flask_session:
+            # Sessions that predate this setting have no stamp; cap them from now on rather than
+            # leaving them exempt forever.
+            if current_user.is_authenticated:
+                builtin_flask_session[SESSION_LOGIN_TIME_KEY] = time.time()
+            return
+        remaining = get_remaining_session_lifetime()
+        if remaining is not None and remaining < 0:
+            log.debug("Session reached [fab] session_max_lifetime_minutes, expiring it.")
+            # ``logout_user`` rather than emptying the session: it also invalidates the
+            # remember-me cookie and drops the user from the request context, so the request that
+            # crossed the deadline is itself unauthenticated.
+            logout_user()
+            builtin_flask_session.pop(SESSION_LOGIN_TIME_KEY, None)

providers/fab/src/airflow/providers/fab/www/views.py

@@ -33,6 +33,7 @@
 from airflow.api_fastapi.auth.managers.base_auth_manager import COOKIE_NAME_JWT_TOKEN
 from airflow.providers.common.compat.sdk import conf
 from airflow.providers.fab.version_compat import AIRFLOW_V_3_1_1_PLUS, AIRFLOW_V_3_1_8_PLUS
+from airflow.providers.fab.www.extensions.init_session import get_remaining_session_lifetime
 
 if AIRFLOW_V_3_1_8_PLUS:
     from airflow.api_fastapi.app import get_cookie_path
@@ -114,9 +115,30 @@ def get_safe_url(url):
     return redirect_url.geturl()
 
 
+def get_token_expiration_seconds() -> int:
+    """
+    Return how long the API token handed to the browser should live.
+
+    The UI only comes back to the auth manager once its token expires, so the token expiry — not
+    the session cookie — is what actually forces a re-authentication. Capping it at whatever is
+    left of the session keeps ``[fab] session_max_lifetime_minutes`` an exact deadline instead of
+    one the user overshoots by up to ``[api_auth] jwt_expiration_time``.
+    """
+    expiration_seconds = conf.getint("api_auth", "jwt_expiration_time")
+    remaining_session_lifetime = get_remaining_session_lifetime()
+    if remaining_session_lifetime is None:
+        return expiration_seconds
+    # Truncate rather than round so the token never survives the deadline. At least a second: a
+    # session already past its deadline is logged out on its next request anyway, and a
+    # non-positive expiry would be rejected as malformed rather than as expired.
+    return max(min(expiration_seconds, int(remaining_session_lifetime)), 1)
+
+
 def redirect(*args, **kwargs):
     if g.user is not None and g.user.is_authenticated:
-        token = get_auth_manager().generate_jwt(g.user)
+        token = get_auth_manager().generate_jwt(
+            g.user, expiration_time_in_seconds=get_token_expiration_seconds()
+        )
         response = make_response(flask_redirect(*args, **kwargs))
 
         secure = request.scheme == "https" or bool(conf.get("api", "ssl_cert", fallback=""))

providers/fab/tests/unit/fab/www/extensions/__init__.py

@@ -0,0 +1,16 @@
+# 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.

providers/fab/tests/unit/fab/www/extensions/test_init_session.py

@@ -0,0 +1,157 @@
+# 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
+
+import datetime
+import time
+from contextlib import contextmanager
+from datetime import timedelta
+
+import pytest
+import time_machine
+from flask import Flask, session as builtin_flask_session
+from flask_login import LoginManager, current_user, login_user
+
+from airflow.providers.fab.www.extensions.init_session import (
+    SESSION_LOGIN_TIME_KEY,
+    get_remaining_session_lifetime,
+    init_session_max_lifetime,
+)
+
+from tests_common.test_utils.config import conf_vars
+
+LOGIN_TIME = datetime.datetime(2026, 9, 9, 12, 0, tzinfo=datetime.timezone.utc)
+
+
+class FakeUser:
+    is_authenticated = True
+    is_active = True
+    is_anonymous = False
+
+    def get_id(self):
+        return "1"
+
+
+@contextmanager
+def build_client(max_lifetime_minutes: int, remember_cookie_name: str | None = None):
+    """Yield a test client for an app wired up with ``session_max_lifetime_minutes`` in force."""
+    app = Flask(__name__)
+    app.secret_key = "test-secret-key"
+    if remember_cookie_name:
+        app.config["REMEMBER_COOKIE_NAME"] = remember_cookie_name
+
+    login_manager = LoginManager(app)
+    login_manager.session_protection = None
+    login_manager.user_loader(lambda user_id: FakeUser() if user_id == "1" else None)
+
+    @app.route("/login")
+    def login():
+        login_user(FakeUser(), remember=bool(remember_cookie_name))
+        return ""
+
+    @app.route("/whoami")
+    def whoami():
+        return "authenticated" if current_user.is_authenticated else "anonymous"
+
+    with conf_vars({("fab", "session_max_lifetime_minutes"): str(max_lifetime_minutes)}):
+        init_session_max_lifetime(app)
+        yield app.test_client()
+
+
+def get_body(response) -> str:
+    return response.get_data(as_text=True)
+
+
+@pytest.mark.parametrize(
+    ("minutes_since_login", "expected"),
+    [(30, "authenticated"), (31, "anonymous")],
+)
+def test_session_expires_at_max_lifetime_despite_activity(minutes_since_login, expected):
+    with build_client(30) as client, time_machine.travel(LOGIN_TIME, tick=False) as traveller:
+        client.get("/login")
+        # Requesting throughout the window must not push the deadline back.
+        traveller.shift(timedelta(minutes=minutes_since_login - 1))
+        assert get_body(client.get("/whoami")) == "authenticated"
+        traveller.shift(timedelta(minutes=1))
+        assert get_body(client.get("/whoami")) == expected
+
+
+def test_session_never_expires_when_max_lifetime_is_disabled():
+    with build_client(0) as client, time_machine.travel(LOGIN_TIME, tick=False) as traveller:
+        client.get("/login")
+        traveller.shift(timedelta(days=30))
+        assert get_body(client.get("/whoami")) == "authenticated"
+
+
+def test_session_predating_the_setting_is_capped_from_its_next_request():
+    with build_client(30) as client, time_machine.travel(LOGIN_TIME, tick=False) as traveller:
+        with client.session_transaction() as flask_session:
+            flask_session["_user_id"] = "1"
+
+        assert get_body(client.get("/whoami")) == "authenticated"
+        traveller.shift(timedelta(minutes=31))
+        assert get_body(client.get("/whoami")) == "anonymous"
+
+
+def test_expiring_a_session_clears_the_remember_me_cookie():
+    with (
+        build_client(30, remember_cookie_name="remember_token") as client,
+        time_machine.travel(LOGIN_TIME, tick=False) as traveller,
+    ):
+        login = client.get("/login")
+        assert any(h.startswith("remember_token=") for h in login.headers.getlist("Set-Cookie"))
+        traveller.shift(timedelta(minutes=31))
+
+        response = client.get("/whoami")
+
+        assert get_body(response) == "anonymous"
+        # Read the headers rather than the client cookie jar: the jar accessors differ between the
+        # Werkzeug versions we support.
+        assert any(h.startswith("remember_token=;") for h in response.headers.getlist("Set-Cookie"))
+
+
+@pytest.mark.parametrize(
+    ("max_lifetime_minutes", "minutes_since_login", "expected"),
+    [
+        pytest.param(0, 0, None, id="uncapped-when-disabled"),
+        pytest.param(30, 0, 1800, id="full-window-at-login"),
+        pytest.param(30, 10, 1200, id="shrinks-as-the-session-ages"),
+        pytest.param(30, 31, -60, id="negative-once-past-the-deadline"),
+    ],
+)
+def test_get_remaining_session_lifetime(max_lifetime_minutes, minutes_since_login, expected):
+    app = Flask(__name__)
+    app.secret_key = "test-secret-key"
+
+    with (
+        conf_vars({("fab", "session_max_lifetime_minutes"): str(max_lifetime_minutes)}),
+        time_machine.travel(LOGIN_TIME, tick=False) as traveller,
+        app.test_request_context(),
+    ):
+        builtin_flask_session[SESSION_LOGIN_TIME_KEY] = time.time()
+        traveller.shift(timedelta(minutes=minutes_since_login))
+
+        assert get_remaining_session_lifetime() == expected
+
+
+def test_remaining_session_lifetime_is_unknown_without_a_login_stamp():
+    app = Flask(__name__)
+    app.secret_key = "test-secret-key"
+
+    with conf_vars({("fab", "session_max_lifetime_minutes"): "30"}), app.test_request_context():
+        assert get_remaining_session_lifetime() is None