fix: refresh stale ServerManager when a server id is reused

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:

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)