From 61679638672b42a90b5ff4452d481462fc1c730c Mon Sep 17 00:00:00 2001 From: Dave Page Date: Tue, 13 Oct 2026 00:47:32 +0100 Subject: [PATCH] Refresh a stale ServerManager when a server id is reused. #6090 (#10312) A cached manager whose server definition no longer matches is rebuilt, and the owner's settings are deep-copied before being overlaid for shared-server users. --- .../browser/server_groups/servers/__init__.py | 8 +- .../tests/test_server_data_isolation.py | 226 ++++++++++ web/pgadmin/utils/driver/psycopg3/__init__.py | 193 ++++++++- .../utils/driver/psycopg3/server_manager.py | 16 + .../psycopg3/tests/test_manager_is_stale.py | 394 ++++++++++++++++++ 5 files changed, 827 insertions(+), 10 deletions(-) create mode 100644 web/pgadmin/utils/driver/psycopg3/tests/test_manager_is_stale.py diff --git a/web/pgadmin/browser/server_groups/servers/__init__.py b/web/pgadmin/browser/server_groups/servers/__init__.py index c64556fde42..71613d7e26a 100644 --- a/web/pgadmin/browser/server_groups/servers/__init__.py +++ b/web/pgadmin/browser/server_groups/servers/__init__.py @@ -279,9 +279,11 @@ def get_shared_server_properties(server, sharedserver): # Override per-user connection_params keys. Use the # SharedServer value whenever it is present, regardless of - # whether the owner's Server has the same key. - s_conn = getattr(server, 'connection_params', None) \ - or {} + # whether the owner's Server has the same key. Work on a copy: + # the dict may be a MutableDict still attached to the owner's + # session-bound row, and mutating it in place would mark that + # row dirty for the next commit. + s_conn = dict(getattr(server, 'connection_params', None) or {}) ss_conn = getattr(sharedserver, 'connection_params', None) or {} for key in SENSITIVE_CONN_KEYS: diff --git a/web/pgadmin/browser/server_groups/servers/tests/test_server_data_isolation.py b/web/pgadmin/browser/server_groups/servers/tests/test_server_data_isolation.py index e1a4992938b..fc3c5ee8e11 100644 --- a/web/pgadmin/browser/server_groups/servers/tests/test_server_data_isolation.py +++ b/web/pgadmin/browser/server_groups/servers/tests/test_server_data_isolation.py @@ -9,9 +9,11 @@ """Tests for server data isolation between users in server mode.""" +import copy import json import sqlite3 import config +from pgadmin.model import db, SharedServer, User from pgadmin.utils.route import BaseTestGenerator from regression.python_test_utils import test_utils as utils from regression.test_setup import config_data @@ -24,6 +26,25 @@ config_data['pgAdmin4_test_non_admin_credentials'] +def restore_server_fixture_on_cleanup(test): + """Restore the shared server fixture once the test finishes. + + The runner passes the same server dict from get_config_data() to + every test, so keys these tests add (such as shared and + shared_username) or change in place would otherwise leak into + later tests. The dict is restored in place, which removes any keys + that were absent originally, and the restore is registered as a + cleanup so that it also runs when setUp fails part-way. + """ + original = copy.deepcopy(test.server) + + def restore(): + test.server.clear() + test.server.update(original) + + test.addCleanup(restore) + + class ServerDataIsolationGetTestCase(BaseTestGenerator): """Verify that a non-admin user cannot access another user's private (non-shared) server by ID.""" @@ -39,6 +60,7 @@ def setUp(self): self.skipTest( 'Data isolation tests only apply to server mode.' ) + restore_server_fixture_on_cleanup(self) # Create a private (non-shared) server as the admin user self.server['shared'] = False @@ -95,6 +117,7 @@ def setUp(self): self.skipTest( 'Data isolation tests only apply to server mode.' ) + restore_server_fixture_on_cleanup(self) # Create a shared server as the admin user self.server['shared'] = True @@ -147,6 +170,7 @@ def setUp(self): self.skipTest( 'Data isolation tests only apply to server mode.' ) + restore_server_fixture_on_cleanup(self) # Create a shared server with sensitive owner-only fields self.server['shared'] = True @@ -232,6 +256,7 @@ def setUp(self): self.skipTest( 'Data isolation tests only apply to server mode.' ) + restore_server_fixture_on_cleanup(self) # Create shared server with owner SSL paths self.server['shared'] = True @@ -308,6 +333,7 @@ def setUp(self): self.skipTest( 'Data isolation tests only apply to server mode.' ) + restore_server_fixture_on_cleanup(self) # Save admin tester BEFORE the decorator replaces it. self.admin_tester = self.tester @@ -364,3 +390,203 @@ def tearDown(self): return utils.delete_server_with_api( self.__class__.tester, self.server_id) + + +class SharedServerNonOwnerConnectionPersistsTestCase(BaseTestGenerator): + """Verify that a non-owner's connection to a shared server survives + later requests. The non-owner connects with their own shared + username, which differs from the owner's, and the cached server + manager must not be mistaken for a stale one (and released) when it + is next looked up.""" + + scenarios = [ + ('Non-owner connection to shared server stays connected', + dict(is_positive_test=True)), + ] + + def setUp(self): + self.server_id = None + if not config.SERVER_MODE: + self.skipTest( + 'Data isolation tests only apply to server mode.' + ) + restore_server_fixture_on_cleanup(self) + + # The owner's username is never used to connect; only the + # non-owner's shared username has to be a real role. The + # fixture dict is shared with every later test, so the real + # username is restored in tearDown. + self.server['shared'] = True + self.server['shared_username'] = self.server['username'] + self.server['username'] = 'pga_test_owner_only_role' + url = "/browser/server/obj/{0}/".format(utils.SERVER_GROUP) + response = self.tester.post( + url, + data=json.dumps(self.server), + content_type='html/json' + ) + self.assertEqual(response.status_code, 200) + response_data = json.loads(response.data.decode('utf-8')) + self.assertIn('node', response_data) + self.server_id = response_data['node']['_id'] + + @create_user_wise_test_client(test_user_details) + def runTest(self): + """The non-owner stays connected after connecting.""" + if not self.server_id: + raise Exception("Server not found to test shared access") + + url = '/browser/server/connect/{0}/{1}'.format( + utils.SERVER_GROUP, self.server_id) + response = self.tester.post( + url, + data=json.dumps({'password': self.server['db_password']}), + content_type='html/json' + ) + self.assertEqual(response.status_code, 200) + data = json.loads(response.data.decode('utf-8')) + self.assertTrue( + data.get('data', {}).get('connected'), + 'Non-owner should be able to connect to the shared server.' + ' Got {0}'.format(data)) + + response = self.tester.get(url) + self.assertEqual(response.status_code, 200) + data = json.loads(response.data.decode('utf-8')) + self.assertTrue( + data.get('data', {}).get('connected'), + 'Non-owner connection to the shared server should still be' + ' open on the next request. Got {0}'.format(data)) + + self.tester.delete(url) + + def tearDown(self): + if self.server_id is None: + return + utils.delete_server_with_api( + self.__class__.tester, self.server_id) + + +class SharedServerNonOwnerSavePasswordKeepsOwnerParamsTestCase( + BaseTestGenerator): + """Verify that a non-owner connecting to a shared server with + 'save password' ticked does not rewrite the owner's + connection_params. The non-owner's SharedServer carries its own + sslcert; the overlay built for the non-owner's server manager must + not share the owner's connection_params dict, or the commit that + saves the non-owner's password flushes the overlaid values into the + owner's row.""" + + scenarios = [ + ('Non-owner save_password leaves owner connection_params intact', + dict(is_positive_test=True)), + ] + + owner_conn_params = { + 'sslcert': '/home/owner/.ssl/cert.pem', + 'sslkey': '/home/owner/.ssl/key.pem', + 'sslrootcert': '/home/owner/.ssl/ca.pem', + } + + def setUp(self): + self.server_id = None + if not config.SERVER_MODE: + self.skipTest( + 'Data isolation tests only apply to server mode.' + ) + restore_server_fixture_on_cleanup(self) + + self.server['shared'] = True + self.server['shared_username'] = self.server['username'] + conn_params = dict(self.server.get('connection_params') or {}) + conn_params.update(self.owner_conn_params) + self.server['connection_params'] = conn_params + url = "/browser/server/obj/{0}/".format(utils.SERVER_GROUP) + response = self.tester.post( + url, + data=json.dumps(self.server), + content_type='html/json' + ) + self.assertEqual(response.status_code, 200) + response_data = json.loads(response.data.decode('utf-8')) + self.assertIn('node', response_data) + self.server_id = response_data['node']['_id'] + + @staticmethod + def _conn_param(data, key): + conn_params = data.get('connection_params', {}) + if isinstance(conn_params, list): + for item in conn_params: + if item.get('name') == key: + return item.get('value') + return None + return conn_params.get(key) + + @create_user_wise_test_client(test_user_details) + def _connect_as_non_owner(self): + """Give the non-owner's SharedServer its own sslcert, then + connect with save_password so the connect endpoint commits.""" + # Fetching the properties creates the non-owner's SharedServer + # record. The test users carry no pgAdmin role, so they cannot + # edit it through the API; set the sslcert on the row directly. + obj_url = '/browser/server/obj/{0}/{1}'.format( + utils.SERVER_GROUP, self.server_id) + response = self.tester.get(obj_url, follow_redirects=True) + self.assertEqual(response.status_code, 200) + with self.app.app_context(): + non_owner = User.query.filter_by( + email=test_user_details['login_username']).first() + shared_server = SharedServer.query.filter_by( + user_id=non_owner.id, osid=self.server_id).first() + self.assertIsNotNone(shared_server) + shared_server.connection_params = dict( + shared_server.connection_params or {}, + sslcert='/home/nonowner/cert.pem') + db.session.commit() + + url = '/browser/server/connect/{0}/{1}'.format( + utils.SERVER_GROUP, self.server_id) + response = self.tester.post( + url, + data=json.dumps({'password': self.server['db_password'], + 'save_password': True}), + content_type='html/json' + ) + self.assertEqual(response.status_code, 200) + data = json.loads(response.data.decode('utf-8')) + self.assertTrue( + data.get('data', {}).get('connected'), + 'Non-owner should be able to connect to the shared server.' + ' Got {0}'.format(data)) + + # A second request as the non-owner exercises the cached + # manager's staleness check and the overlay it is built from. + response = self.tester.get(url) + self.assertEqual(response.status_code, 200) + self.tester.delete(url) + + def runTest(self): + """The owner's connection_params survive a non-owner's + connect with save_password.""" + if not self.server_id: + raise Exception("Server not found to test shared access") + + self._connect_as_non_owner() + + url = '/browser/server/obj/{0}/{1}'.format( + utils.SERVER_GROUP, self.server_id) + response = self.tester.get(url, follow_redirects=True) + self.assertEqual(response.status_code, 200) + data = json.loads(response.data.decode('utf-8')) + for key, value in self.owner_conn_params.items(): + self.assertEqual( + self._conn_param(data, key), value, + 'Owner connection_params "{0}" should be unchanged after ' + 'a non-owner connected with save_password. Got: {1}' + .format(key, self._conn_param(data, key))) + + def tearDown(self): + if self.server_id is None: + return + utils.delete_server_with_api( + self.__class__.tester, self.server_id) diff --git a/web/pgadmin/utils/driver/psycopg3/__init__.py b/web/pgadmin/utils/driver/psycopg3/__init__.py index 96bbc212d8c..aabc6da2deb 100644 --- a/web/pgadmin/utils/driver/psycopg3/__init__.py +++ b/web/pgadmin/utils/driver/psycopg3/__init__.py @@ -13,20 +13,23 @@ object. """ +import copy import datetime import re from flask import session from flask_babel import gettext from flask_login import current_user +from sqlalchemy import inspect as sa_inspect from werkzeug.exceptions import InternalServerError import psycopg from threading import Lock import config -from pgadmin.model import Server +from pgadmin.model import Server, SharedServer from pgadmin.utils.server_access import get_server, \ get_user_server_query from pgadmin.utils.exception import ObjectGone +from pgadmin.utils.passexec import build_passexec from .keywords import scan_keyword from ..abstract import BaseDriver from .connection import Connection @@ -81,17 +84,159 @@ def _restore_connections_from_session(self): session['__pgsql_server_managers'].copy() servers = get_user_server_query().filter( Server.is_adhoc == 0) + pga_user = self._current_pga_user() for server in servers: + server = self._manager_source(server) manager = managers[str(server.id)] = \ ServerManager(server) + manager.pga_user = pga_user if server.id in session_managers: - manager._restore( - session_managers[server.id]) - manager.update_session() + saved = session_managers[server.id] + if self._saved_state_is_stale( + saved, server, pga_user): + # The persisted blob was serialized under + # this numeric id by whatever Server row + # held it before (e.g. the configuration + # database was reset or restored without + # restarting pgAdmin), so it no longer + # describes this row. Restoring it would + # hand the new row the previous row's + # password/connection state. The same goes + # for state saved by a different pgAdmin + # user on this browser session. Drop it and + # let the manager start clean. + manager.update_session() + else: + manager._restore(saved) + manager.update_session() return managers return {} + @staticmethod + def _current_pga_user(): + """ + The fs_uniquifier of the logged-in pgAdmin user, which cached and + serialized ServerManager state is bound to. Unlike User.id it is + random, so it is not reused when the configuration database is + reset and a new user is created. + """ + return getattr(current_user, 'fs_uniquifier', None) + + @staticmethod + def _manager_source(server_data): + """ + The Server a ServerManager for this server should be built from, + and compared against by _manager_is_stale and + _saved_state_is_stale. + + For a non-owner of a shared server that is not the owner's row: + the connect endpoint builds the manager from the user's + SharedServer overlay (see + ServerModule.get_shared_server_properties), which replaces the + username, service, tunnel host, password and so on with the + user's own values. Building or comparing against the owner's row + instead would treat every such manager as stale and drop its + live connection on every request. + + The overlay is applied to a transient copy of the row's columns, + so the session-bound instance other code may be holding is never + detached or modified. The SharedServer record is only read here, + never created; until the user has one (it is created when they + first connect), the row itself is returned. + """ + if not (config.SERVER_MODE and server_data.shared and + server_data.user_id != current_user.id): + return server_data + + shared_server = SharedServer.query.filter_by( + user_id=current_user.id, osid=server_data.id).first() + if shared_server is None: + return server_data + + # Imported here, as the servers module imports this driver. + from pgadmin.browser.server_groups.servers import ServerModule + + # Columns only: Server.clone() would also copy the servergroup + # relationship, which cascades the copy into the session. The + # values are deep-copied so the JSON columns (connection_params, + # tags) are not shared by reference with the session-bound row: + # MutableDict reports an in-place change to every parent it + # belongs to, so the overlay's edits would otherwise dirty the + # owner's row and be flushed by the next commit. + overlay = Server(**{ + attr.key: copy.deepcopy(getattr(server_data, attr.key)) + for attr in sa_inspect(Server).column_attrs + }) + return ServerModule.get_shared_server_properties( + overlay, shared_server) + + @staticmethod + def _effective_tunnel_host(server_data): + """ + The tunnel host a ServerManager built from this row would carry. + ServerManager.update() ignores the row's tunnel columns when + SUPPORT_SSH_TUNNEL is off, so a row that still has a tunnel host + (imported, or saved before the setting was turned off) must be + compared against None rather than being reported stale forever. + """ + return server_data.tunnel_host if config.SUPPORT_SSH_TUNNEL \ + else None + + @staticmethod + def _saved_state_is_stale(saved, server_data, pga_user=None): + """ + Same identity check as _manager_is_stale, applied to the + serialized ServerManager state carried across worker + restarts/new sessions in the Flask session + ('__pgsql_server_managers'), before it is restored onto a + manager that was just built fresh from the current Server row. + Without this, a reused server id would have its old serialized + password/connections restored onto the new row on the very + first request, before any manager exists to run + _manager_is_stale against. + + The state is also stale if it was saved by a different pgAdmin + user: nothing rotates the session id at login, so a new user + logging in on the same browser session must not inherit the + previous user's password or connections, even for a server with + identical connection details. + """ + return ( + saved.get('pga_user') != pga_user or + saved.get('host') != server_data.host or + saved.get('port') != server_data.port or + saved.get('db') != server_data.maintenance_db or + saved.get('user') != server_data.username or + saved.get('service') != server_data.service or + saved.get('tunnel_host') != + Driver._effective_tunnel_host(server_data) + ) + + @staticmethod + def _manager_is_stale(manager, server_data, pga_user=None): + """ + A cached manager is normally kept in sync with edits to its + Server row via explicit manager.update() calls from the + server-edit endpoints. It can still go stale in place if the + row itself was swapped out from under it, e.g. a numeric + server id reused by an unrelated row after the configuration + database was reset or restored without restarting pgAdmin, so + compare against what actually identifies the target rather + than trusting the id match alone. It is also stale if it was + built for a different pgAdmin user on the same browser session + (see _saved_state_is_stale). + """ + return ( + getattr(manager, 'pga_user', None) != pga_user or + manager.host != server_data.host or + manager.port != server_data.port or + manager.db != server_data.maintenance_db or + manager.user != server_data.username or + manager.service != server_data.service or + manager.tunnel_host != Driver._effective_tunnel_host(server_data) + ) + def connection_manager(self, sid=None): """ connection_manager(...) @@ -137,15 +282,49 @@ def connection_manager(self, sid=None): managers = self.managers[session.sid] if str(sid) in managers: manager = managers[str(sid)] + pga_user = self._current_pga_user() + manager_source = self._manager_source(server_data) with connection_restore_lock: - manager._restore_connections() - manager.update_session() + if self._manager_is_stale( + manager, manager_source, pga_user): + # The id has been reused by an unrelated Server + # row (e.g. the configuration database was reset + # or restored without restarting pgAdmin), so the + # cached manager still points at whatever server + # it was originally built from. Drop it rather + # than report a live connection to a server that, + # from this row's perspective, was never opened. + # The same applies to a manager built for another + # pgAdmin user on this browser session. + manager.release() + manager.update(manager_source) + manager.pga_user = pga_user + else: + manager._restore_connections() + manager.update_session() + # Identity (host/port/db/user/service/tunnel) + # still matches, so the live connection is kept, + # but access-control-relevant metadata such as + # shared/ownership is not part of that identity + # check and manager.update() was skipped above - + # refresh it here too, otherwise a row whose + # sharing/ownership changed via the same reused-id + # path could keep serving the previous owner's + # passexec to a new, non-owning user. The password + # exec command is only rebuilt when sharing + # changes, so its cached password survives + # ordinary requests. + if manager.shared != server_data.shared: + manager.shared = server_data.shared + manager.passexec = build_passexec( + manager_source) managers['pinged'] = datetime.datetime.now() if str(sid) not in managers: # server_data was already access-checked above; # it cannot be None at this point. - manager = ServerManager(server_data) + manager = ServerManager(self._manager_source(server_data)) + manager.pga_user = self._current_pga_user() managers[str(sid)] = manager return manager diff --git a/web/pgadmin/utils/driver/psycopg3/server_manager.py b/web/pgadmin/utils/driver/psycopg3/server_manager.py index 3fa2baddfcd..f282ea9e406 100644 --- a/web/pgadmin/utils/driver/psycopg3/server_manager.py +++ b/web/pgadmin/utils/driver/psycopg3/server_manager.py @@ -54,6 +54,9 @@ def __init__(self, server): self.tunnel_object = None self.tunnel_created = False self.display_connection_string = '' + # fs_uniquifier of the pgAdmin user this manager was built for; + # set by the driver, see Driver._current_pga_user. + self.pga_user = None self.update(server) @@ -151,6 +154,19 @@ def as_dict(self): res['ver'] = self.ver res['sversion'] = self.sversion + # Persisted alongside the connection state so a later restore + # (e.g. after a worker restart) can tell whether this blob still + # belongs to the Server row for this id, or whether the id was + # reused by an unrelated row after the configuration database + # was reset/restored - see Driver._manager_is_stale. + res['host'] = self.host + res['port'] = self.port + res['db'] = self.db + res['user'] = self.user + res['service'] = self.service + res['tunnel_host'] = self.tunnel_host + res['pga_user'] = self.pga_user + self._set_password(res) if self.use_ssh_tunnel: diff --git a/web/pgadmin/utils/driver/psycopg3/tests/test_manager_is_stale.py b/web/pgadmin/utils/driver/psycopg3/tests/test_manager_is_stale.py new file mode 100644 index 00000000000..4f3b63c5eb3 --- /dev/null +++ b/web/pgadmin/utils/driver/psycopg3/tests/test_manager_is_stale.py @@ -0,0 +1,394 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2026, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +""" +Unit tests for Driver._manager_is_stale, Driver._saved_state_is_stale and +Driver._manager_source. + +These are pure attribute-comparison tests, run as plain unittest +TestCases without needing a Postgres server connection. +""" + +import unittest +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from sqlalchemy import inspect as sa_inspect + +from pgadmin.model import Server, SharedServer +from pgadmin.utils.route import BaseTestGenerator +from pgadmin.utils.driver.psycopg3 import Driver + + +def make_manager(**overrides): + fields = dict( + host='old-host', port=5432, db='postgres', user='old-user', + service=None, tunnel_host=None, + ) + fields.update(overrides) + return SimpleNamespace(**fields) + + +def make_server_data(**overrides): + fields = dict( + host='old-host', port=5432, maintenance_db='postgres', + username='old-user', service=None, tunnel_host=None, + ) + fields.update(overrides) + return SimpleNamespace(**fields) + + +def make_saved_state(**overrides): + """Mimics the identity fields ServerManager.as_dict() persists into + the Flask session ('__pgsql_server_managers') alongside the + serialized password/connections.""" + fields = dict( + host='old-host', port=5432, db='postgres', user='old-user', + service=None, tunnel_host=None, + ) + fields.update(overrides) + return fields + + +class _PureUnitTestSetupMixin: + """setUp here calls unittest.TestCase.setUp directly, skipping + BaseTestGenerator.setUp's Postgres connection.""" + + def setUp(self): + unittest.TestCase.setUp(self) + + +class TestManagerIsStaleMatchesUnchanged( + _PureUnitTestSetupMixin, BaseTestGenerator): + """A manager whose identity fields still match the current Server + row is not stale, even if the row was legitimately edited via the + normal manager.update() flow elsewhere.""" + + scenarios = [('default', dict())] + + def runTest(self): + manager = make_manager() + server_data = make_server_data() + + self.assertFalse(Driver._manager_is_stale(manager, server_data)) + + +class TestManagerIsStaleDetectsReusedId( + _PureUnitTestSetupMixin, BaseTestGenerator): + """A manager built from a Server row that no longer matches the + current row for this id (e.g. the id was reused after the + configuration database was reset) must be treated as stale.""" + + scenarios = [('default', dict())] + + def runTest(self): + manager = make_manager(host='deleted-server.example.com') + server_data = make_server_data(host='new-server.example.com') + + self.assertTrue(Driver._manager_is_stale(manager, server_data)) + + +class TestManagerIsStaleChecksEachIdentityField( + _PureUnitTestSetupMixin, BaseTestGenerator): + """Any one of host/port/db/user/service/tunnel_host differing is + enough to mark the manager stale.""" + + scenarios = [('default', dict())] + + def runTest(self): + server_data = make_server_data() + + for field, value in ( + ('port', 5433), + ('maintenance_db', 'template1'), + ('username', 'new-user'), + ('service', 'myservice'), + ('tunnel_host', 'bastion.example.com'), + ): + manager = make_manager() + changed_server_data = make_server_data(**{field: value}) + self.assertTrue( + Driver._manager_is_stale(manager, changed_server_data), + "expected stale manager when %s changes" % field) + + +class TestSavedStateIsStaleMatchesUnchanged( + _PureUnitTestSetupMixin, BaseTestGenerator): + """Serialized session state whose identity fields still match the + current Server row is safe to restore onto a freshly built + manager.""" + + scenarios = [('default', dict())] + + def runTest(self): + saved = make_saved_state() + server_data = make_server_data() + + self.assertFalse(Driver._saved_state_is_stale(saved, server_data)) + + +class TestSavedStateIsStaleDetectsReusedId( + _PureUnitTestSetupMixin, BaseTestGenerator): + """Serialized state left over from a deleted Server row (e.g. after + the configuration database was reset/restored without restarting + pgAdmin) must not be restored onto the row that reused its id.""" + + scenarios = [('default', dict())] + + def runTest(self): + saved = make_saved_state(host='deleted-server.example.com') + server_data = make_server_data(host='new-server.example.com') + + self.assertTrue(Driver._saved_state_is_stale(saved, server_data)) + + +class TestSavedStateIsStaleMissingFieldsAreStale( + _PureUnitTestSetupMixin, BaseTestGenerator): + """State serialized before identity fields were added to + ServerManager.as_dict() (i.e. a dict without host/port/etc. keys) + cannot be verified, so it must be treated as stale rather than + trusted blindly.""" + + scenarios = [('default', dict())] + + def runTest(self): + saved = {'sid': 1, 'ver': '18.0', 'sversion': 180000, + 'connections': {}} + server_data = make_server_data() + + self.assertTrue(Driver._saved_state_is_stale(saved, server_data)) + + +class TestSavedStateIsStaleChecksEachIdentityField( + _PureUnitTestSetupMixin, BaseTestGenerator): + """Any one of host/port/db/user/service/tunnel_host differing is + enough to discard the serialized state.""" + + scenarios = [('default', dict())] + + def runTest(self): + server_data = make_server_data() + + for field, value in ( + ('port', 5433), + ('maintenance_db', 'template1'), + ('username', 'new-user'), + ('service', 'myservice'), + ('tunnel_host', 'bastion.example.com'), + ): + saved = make_saved_state() + changed_server_data = make_server_data(**{field: value}) + self.assertTrue( + Driver._saved_state_is_stale(saved, changed_server_data), + "expected stale saved state when %s changes" % field) + + +class TestManagerIsStaleDetectsDifferentPgAdminUser( + _PureUnitTestSetupMixin, BaseTestGenerator): + """A manager built for one pgAdmin user must not be reused for + another user who logs in on the same browser session, even when + every connection field matches (issue #6090: the same servers + re-imported by a new user after the configuration database was + reset).""" + + scenarios = [('default', dict())] + + def runTest(self): + manager = make_manager(pga_user='old-user-uniquifier') + server_data = make_server_data() + + self.assertTrue(Driver._manager_is_stale( + manager, server_data, 'new-user-uniquifier')) + self.assertFalse(Driver._manager_is_stale( + manager, server_data, 'old-user-uniquifier')) + + +class TestSavedStateIsStaleDetectsDifferentPgAdminUser( + _PureUnitTestSetupMixin, BaseTestGenerator): + """Serialized state saved by one pgAdmin user must not be restored + for another user on the same browser session, even when every + connection field matches.""" + + scenarios = [('default', dict())] + + def runTest(self): + saved = make_saved_state(pga_user='old-user-uniquifier') + server_data = make_server_data() + + self.assertTrue(Driver._saved_state_is_stale( + saved, server_data, 'new-user-uniquifier')) + self.assertFalse(Driver._saved_state_is_stale( + saved, server_data, 'old-user-uniquifier')) + + +def make_shared_server_row(**overrides): + """A transient (never added to the session) shared Server row owned + by user 1.""" + fields = dict( + id=7, user_id=1, servergroup_id=1, name='shared', + host='old-host', port=5432, maintenance_db='postgres', + username='owner-user', service='owner-service', + tunnel_host='owner-bastion', shared=True, + ) + fields.update(overrides) + return Server(**fields) + + +class TestManagerSourceUsesSharedServerOverlay( + _PureUnitTestSetupMixin, BaseTestGenerator): + """A non-owner's manager for a shared server is built from their + SharedServer overlay, so its username/service/tunnel host differ + from the owner's Server row. The manager source must carry the + overlay, so such a manager compares as not stale rather than having + its live connection released on every request, and the owner's row + must be left untouched.""" + + scenarios = [('default', dict())] + + def runTest(self): + server_data = make_shared_server_row() + shared_server = SharedServer( + osid=7, user_id=2, servergroup_id=1, name='mine', + username='other-user', service=None, + tunnel_host='other-bastion') + # The manager as the connect endpoint leaves it for the + # non-owner, after manager.update() with the overlay. + manager = make_manager( + user='other-user', service=None, tunnel_host='other-bastion', + pga_user='other-uniquifier') + + shared_model = MagicMock() + shared_model.query.filter_by.return_value.first.return_value = \ + shared_server + module = 'pgadmin.utils.driver.psycopg3' + with patch(module + '.config.SERVER_MODE', True), \ + patch(module + '.current_user', SimpleNamespace(id=2)), \ + patch(module + '.SharedServer', shared_model): + source = Driver._manager_source(server_data) + + shared_model.query.filter_by.assert_called_once_with( + user_id=2, osid=7) + self.assertIsNot(source, server_data) + # Never added to a session, so nothing can flush it. + self.assertTrue(sa_inspect(source).transient) + self.assertEqual(source.id, 7) + self.assertEqual(source.username, 'other-user') + self.assertIsNone(source.service) + self.assertEqual(source.tunnel_host, 'other-bastion') + self.assertFalse(Driver._manager_is_stale( + manager, source, 'other-uniquifier')) + self.assertFalse(Driver._saved_state_is_stale( + make_saved_state(user='other-user', service=None, + tunnel_host='other-bastion', + pga_user='other-uniquifier'), + source, 'other-uniquifier')) + + # The owner's row itself is unchanged. + self.assertEqual(server_data.username, 'owner-user') + self.assertEqual(server_data.service, 'owner-service') + self.assertEqual(server_data.tunnel_host, 'owner-bastion') + self.assertTrue(Driver._manager_is_stale( + manager, server_data, 'other-uniquifier')) + + +class TestManagerSourceOwnerUsesServerRow( + _PureUnitTestSetupMixin, BaseTestGenerator): + """The owner of a shared server, anyone in desktop mode, and a + non-owner with no SharedServer record yet all get the Server row + itself.""" + + scenarios = [('default', dict())] + + def runTest(self): + server_data = make_shared_server_row() + module = 'pgadmin.utils.driver.psycopg3' + + for server_mode, user_id, shared_row in ( + (True, 1, None), (False, 2, None), (True, 2, None)): + shared_model = MagicMock() + shared_model.query.filter_by.return_value.first.return_value \ + = shared_row + with patch(module + '.config.SERVER_MODE', server_mode), \ + patch(module + '.current_user', + SimpleNamespace(id=user_id)), \ + patch(module + '.SharedServer', shared_model): + self.assertIs( + Driver._manager_source(server_data), server_data) + + +class TestManagerSourceDoesNotShareOwnerJsonColumns( + _PureUnitTestSetupMixin, BaseTestGenerator): + """The overlay must not share the owner row's connection_params (a + MutableDict) or tags by reference: get_shared_server_properties + rewrites the sensitive keys in place, and a MutableDict reports that + change to every row it belongs to, so a shared reference would dirty + the owner's session-bound row and let a non-owner's commit overwrite + the owner's SSL and passfile paths.""" + + scenarios = [('default', dict())] + + def runTest(self): + owner_params = { + 'sslcert': '/home/owner/.ssl/cert.pem', + 'sslkey': '/home/owner/.ssl/key.pem', + 'passfile': '/home/owner/.pgpass', + 'sslmode': 'require', + } + server_data = make_shared_server_row( + connection_params=dict(owner_params), tags=['owner-tag']) + shared_server = SharedServer( + osid=7, user_id=2, servergroup_id=1, name='mine', + username='other-user', + connection_params={'sslcert': '/home/other/cert.pem'}) + + shared_model = MagicMock() + shared_model.query.filter_by.return_value.first.return_value = \ + shared_server + module = 'pgadmin.utils.driver.psycopg3' + with patch(module + '.config.SERVER_MODE', True), \ + patch(module + '.current_user', SimpleNamespace(id=2)), \ + patch(module + '.SharedServer', shared_model): + source = Driver._manager_source(server_data) + + self.assertIsNot(source.connection_params, + server_data.connection_params) + self.assertIsNot(source.tags, server_data.tags) + self.assertEqual(source.connection_params, + {'sslcert': '/home/other/cert.pem', + 'sslmode': 'require'}) + # The owner's row still holds exactly what it held before. + self.assertEqual(server_data.connection_params, owner_params) + self.assertEqual(server_data.tags, ['owner-tag']) + + +class TestStalePredicatesIgnoreTunnelHostWhenTunnelsUnsupported( + _PureUnitTestSetupMixin, BaseTestGenerator): + """With SUPPORT_SSH_TUNNEL off, ServerManager.update() stores + tunnel_host = None whatever the row says, so a row that still + carries a tunnel host (imported, or saved before the setting was + turned off) must compare against None rather than being stale on + every request.""" + + scenarios = [('default', dict())] + + def runTest(self): + module = 'pgadmin.utils.driver.psycopg3' + server_data = make_server_data(tunnel_host='bastion.example.com') + manager = make_manager(tunnel_host=None) + saved = make_saved_state(tunnel_host=None) + + with patch(module + '.config.SUPPORT_SSH_TUNNEL', False): + self.assertFalse(Driver._manager_is_stale(manager, server_data)) + self.assertFalse( + Driver._saved_state_is_stale(saved, server_data)) + + with patch(module + '.config.SUPPORT_SSH_TUNNEL', True): + self.assertTrue(Driver._manager_is_stale(manager, server_data)) + self.assertTrue( + Driver._saved_state_is_stale(saved, server_data))