From ee7dbdecdae1bb6067bbe25b86af43cec326629a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gr=C3=A9goire=20Bellon-Gervais?= Date: Fri, 11 Sep 2026 10:11:45 +0200 Subject: [PATCH] Try to remember username and inject it on logout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Grégoire Bellon-Gervais --- web/pgadmin/__init__.py | 50 ++++-- .../tests/test_log_authenticated_user.py | 154 ++++++++++++++++++ 2 files changed, 190 insertions(+), 14 deletions(-) create mode 100644 web/pgadmin/browser/tests/test_log_authenticated_user.py diff --git a/web/pgadmin/__init__.py b/web/pgadmin/__init__.py index 619f466b24e..f47ebd119c9 100644 --- a/web/pgadmin/__init__.py +++ b/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() diff --git a/web/pgadmin/browser/tests/test_log_authenticated_user.py b/web/pgadmin/browser/tests/test_log_authenticated_user.py new file mode 100644 index 00000000000..26a6db03aba --- /dev/null +++ b/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)