Try to remember username and inject it on logout

web/pgadmin/__init__.py

@@ -21,7 +21,7 @@
 from collections import defaultdict
 from importlib import import_module
 
-from flask import Flask, abort, request, current_app, session, url_for
+from flask import Flask, abort, request, current_app, g, session, url_for
 from flask_socketio import SocketIO
 from werkzeug.exceptions import HTTPException
 from flask_babel import Babel, gettext
@@ -183,6 +183,36 @@ def _find_blueprint():
 current_blueprint = LocalProxy(_find_blueprint)
 
 
+def _remember_authenticated_user():
+    """Stash the username so a logout within this request doesn't lose it."""
+    if current_app.config.get('LOG_AUTHENTICATED_USER') and \
+            current_user.is_authenticated:
+        g.authenticated_user_name = current_user.username
+
+
+def _set_remote_user_header(response):
+    """Report the authenticated user to the HTTP access log."""
+    if not current_app.config.get('LOG_AUTHENTICATED_USER'):
+        return response
+
+    username = current_user.username if current_user.is_authenticated \
+        else g.get('authenticated_user_name')
+
+    if username:
+        # HTTP headers are latin-1 only, so transliterate anything outside
+        # that range to avoid gunicorn 500s for unicode names.
+        safe = username.encode('latin-1', 'replace').decode('latin-1')
+        # CR/LF and other control chars are valid latin-1 but Werkzeug
+        # rejects them in header values (would 500 every request for that
+        # user), so drop any non-printable characters too.
+        safe = ''.join(c for c in safe if c.isprintable())
+        response.headers['X-Remote-User'] = safe
+    else:
+        response.headers.pop('X-Remote-User', None)
+
+    return response
+
+
 def create_app(app_name=None):
     # Configuration settings
     import config
@@ -838,6 +868,10 @@ def limit_host_addr():
     def before_request():
         """Login the default user if running in desktop mode"""
 
+        # current_user is already anonymous by the time after_request runs on
+        # a logout, so remember who made the request while we still can.
+        _remember_authenticated_user()
+
         # Check the auth key is valid, if it's set, and we're not in server
         # mode, and it's not a help file request.
 
@@ -879,19 +913,7 @@ def before_request():
 
     @app.after_request
     def after_request(response):
-        if config.LOG_AUTHENTICATED_USER:
-            if current_user.is_authenticated and current_user.username:
-                # HTTP headers are latin-1 only, so transliterate anything
-                # outside that range to avoid gunicorn 500s for unicode names.
-                safe = current_user.username.encode(
-                    'latin-1', 'replace').decode('latin-1')
-                # CR/LF and other control chars are valid latin-1 but Werkzeug
-                # rejects them in header values (would 500 every request for
-                # that user), so drop any non-printable characters too.
-                safe = ''.join(c for c in safe if c.isprintable())
-                response.headers['X-Remote-User'] = safe
-            else:
-                response.headers.pop('X-Remote-User', None)
+        _set_remote_user_header(response)
 
         if 'key' in request.args:
             domain = dict()

web/pgadmin/browser/tests/test_log_authenticated_user.py

@@ -0,0 +1,154 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""
+Unit tests for the LOG_AUTHENTICATED_USER wiring in pgadmin/__init__.py.
+
+Covers the two request hooks that emit the X-Remote-User response header
+(read by the gunicorn access log, see pkg/docker/gunicorn_config.py):
+
+  * _remember_authenticated_user - stashes the username on flask.g at
+    before_request time, because Flask-Login has already cleared
+    current_user by the time after_request runs on a logout request.
+  * _set_remote_user_header - writes the header (sanitised for the
+    latin-1 only header encoding), falls back to the stashed name, and
+    strips any incoming/spoofed header when there is no user.
+"""
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from flask import Flask
+
+from pgadmin import _remember_authenticated_user, _set_remote_user_header
+from pgadmin.utils.route import BaseTestGenerator
+
+
+class _SkipServerSetUpMixin:
+    """Bypass BaseTestGenerator's Postgres server setUp - these are pure
+    logic/wiring tests that need no live server or HTTP infrastructure."""
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+
+class _StubResponse:
+    """Minimal stand-in for a Flask Response - only .headers is used."""
+
+    def __init__(self, headers=None):
+        self.headers = dict(headers or {})
+
+
+def _make_app(log_authenticated_user=True):
+    app = Flask(__name__)
+    app.secret_key = 'test'
+    app.config['LOG_AUTHENTICATED_USER'] = log_authenticated_user
+    return app
+
+
+def _user(username):
+    return SimpleNamespace(is_authenticated=True, username=username)
+
+
+ANONYMOUS = SimpleNamespace(is_authenticated=False, username=None)
+
+
+class TestHeaderSetForAuthenticatedUser(
+        _SkipServerSetUpMixin, BaseTestGenerator):
+    """A normal authenticated request gets X-Remote-User."""
+
+    scenarios = [('default', dict())]
+
+    def runTest(self):
+        app = _make_app()
+        response = _StubResponse()
+        with app.test_request_context():
+            with patch('pgadmin.current_user', _user('alice')):
+                _remember_authenticated_user()
+                _set_remote_user_header(response)
+
+        self.assertEqual(response.headers['X-Remote-User'], 'alice')
+
+
+class TestHeaderSurvivesLogout(_SkipServerSetUpMixin, BaseTestGenerator):
+    """On a logout request current_user is already anonymous when
+    after_request runs, but the name captured at before_request time must
+    still be reported."""
+
+    scenarios = [('default', dict())]
+
+    def runTest(self):
+        app = _make_app()
+        response = _StubResponse()
+        with app.test_request_context():
+            with patch('pgadmin.current_user', _user('alice')):
+                _remember_authenticated_user()
+            # Flask-Login cleared current_user in between.
+            with patch('pgadmin.current_user', ANONYMOUS):
+                _set_remote_user_header(response)
+
+        self.assertEqual(response.headers['X-Remote-User'], 'alice')
+
+
+class TestHeaderDroppedWhenNoUser(_SkipServerSetUpMixin, BaseTestGenerator):
+    """With no user at all, any pre-existing (potentially spoofed) header
+    is removed rather than passed through."""
+
+    scenarios = [('default', dict())]
+
+    def runTest(self):
+        app = _make_app()
+        response = _StubResponse({'X-Remote-User': 'spoofed'})
+        with app.test_request_context():
+            with patch('pgadmin.current_user', ANONYMOUS):
+                _remember_authenticated_user()
+                _set_remote_user_header(response)
+
+        self.assertNotIn('X-Remote-User', response.headers)
+
+
+class TestHeaderNotSetWhenDisabled(_SkipServerSetUpMixin, BaseTestGenerator):
+    """LOG_AUTHENTICATED_USER = False (the default) emits nothing."""
+
+    scenarios = [('default', dict())]
+
+    def runTest(self):
+        app = _make_app(log_authenticated_user=False)
+        response = _StubResponse()
+        with app.test_request_context():
+            with patch('pgadmin.current_user', _user('alice')):
+                _remember_authenticated_user()
+                _set_remote_user_header(response)
+
+        self.assertNotIn('X-Remote-User', response.headers)
+
+
+class TestUsernameSanitised(_SkipServerSetUpMixin, BaseTestGenerator):
+    """Header values are latin-1 only and Werkzeug rejects control
+    characters, so names are transliterated and stripped rather than
+    blowing up the request."""
+
+    scenarios = [
+        ('latin-1 name kept as is', dict(
+            username='Jos\u00e9', expected='Jos\u00e9')),
+        ('non latin-1 transliterated', dict(
+            username='\u65e5\u672c', expected='??')),
+        ('CRLF injection stripped', dict(
+            username='alice\r\nX-Injected: 1', expected='aliceX-Injected: 1')),
+    ]
+
+    def runTest(self):
+        app = _make_app()
+        response = _StubResponse()
+        with app.test_request_context():
+            with patch('pgadmin.current_user', _user(self.username)):
+                _set_remote_user_header(response)
+
+        self.assertEqual(response.headers['X-Remote-User'], self.expected)