diff --git a/web/pgadmin/browser/server_groups/servers/__init__.py b/web/pgadmin/browser/server_groups/servers/__init__.py index cfcb324c4d8..60adc24ec39 100644 --- a/web/pgadmin/browser/server_groups/servers/__init__.py +++ b/web/pgadmin/browser/server_groups/servers/__init__.py @@ -53,8 +53,14 @@ # File-path keys in connection_params that are per-user and must # not be copied from the owner to a new SharedServer or leaked # through the property merge. -SENSITIVE_CONN_KEYS = frozenset({ - 'passfile', 'sslcert', 'sslkey', +# +# 'passfile' is a file path too, but it is deliberately not in here: it +# is the mechanism by which a shared server's owner (an administrator +# provisioning servers.json, say) lets every user of that server +# authenticate automatically, so it is inherited like any other +# connection parameter rather than being stripped away. +PER_USER_CONN_KEYS = frozenset({ + 'sslcert', 'sslkey', 'sslrootcert', 'sslcrl', 'sslcrldir', }) @@ -211,13 +217,18 @@ def get_shared_server_properties(server, sharedserver): or {} ss_conn = getattr(sharedserver, 'connection_params', None) or {} - for key in SENSITIVE_CONN_KEYS: + # 'passfile' is absent from PER_USER_CONN_KEYS, so the owner's + # value survives this loop and reaches the non-owner, including + # on SharedServer rows created before it started being copied. + for key in PER_USER_CONN_KEYS: if key in ss_conn: s_conn[key] = ss_conn[key] elif key in s_conn: # Owner has this key but non-owner doesn't — # remove it so the owner's path doesn't leak. del s_conn[key] + if 'passfile' in ss_conn: + s_conn['passfile'] = ss_conn['passfile'] server.connection_params = s_conn server.servergroup_id = sharedserver.servergroup_id @@ -228,7 +239,14 @@ def get_shared_server_properties(server, sharedserver): server.passexec_cmd = sharedserver.passexec_cmd server.passexec_expiration = sharedserver.passexec_expiration server.kerberos_conn = sharedserver.kerberos_conn - server.tags = sharedserver.tags + # A SharedServer row created before tags were copied across has + # NULL tags, which is not the same thing as a user who has + # cleared every tag they had: the latter leaves an empty list. + # Fall back to the owner's tags only for the former, so an + # existing row picks them up without overriding a deliberate + # choice. + server.tags = sharedserver.tags \ + if sharedserver.tags is not None else server.tags server.post_connection_sql = sharedserver.post_connection_sql return server @@ -437,15 +455,16 @@ def create_shared_server(data, gid): db.session.rollback() user = User.query.filter_by(id=data.user_id).first() - # Strip owner's sensitive file paths from - # connection_params — each user should configure - # their own SSL/passfile paths. + # Strip the owner's per-user SSL file paths from + # connection_params — each user should configure their own + # SSL certificate/key paths. 'passfile' is not among them, + # so it is copied like any other connection parameter. safe_conn_params = {} if data.connection_params: safe_conn_params = { k: v for k, v in data.connection_params.items() - if k not in SENSITIVE_CONN_KEYS + if k not in PER_USER_CONN_KEYS } shared_server = SharedServer( @@ -480,7 +499,7 @@ def create_shared_server(data, gid): passexec_cmd=None, passexec_expiration=None, kerberos_conn=False, - tags=None, + tags=data.tags, post_connection_sql=None ) db.session.add(shared_server) diff --git a/web/pgadmin/browser/server_groups/servers/tests/test_shared_server_unit.py b/web/pgadmin/browser/server_groups/servers/tests/test_shared_server_unit.py index 49f6a2b93eb..90b00c1cbd1 100644 --- a/web/pgadmin/browser/server_groups/servers/tests/test_shared_server_unit.py +++ b/web/pgadmin/browser/server_groups/servers/tests/test_shared_server_unit.py @@ -110,6 +110,14 @@ class TestGetSharedServerProperties(BaseTestGenerator): dict(test_method='test_overlays_kerberos_tags')), ('Merge strips owner SSL paths not in SharedServer', dict(test_method='test_strips_owner_ssl_paths')), + ('Merge inherits the owner passfile when SharedServer has none', + dict(test_method='test_inherits_owner_passfile')), + ('Merge prefers the SharedServer passfile when it has one', + dict(test_method='test_shared_passfile_wins')), + ('Merge falls back to owner tags when SharedServer tags are NULL', + dict(test_method='test_tags_fall_back_to_owner')), + ('Merge respects tags a user has deliberately cleared', + dict(test_method='test_cleared_tags_are_respected')), ('Merge applies SharedServer SSL paths', dict(test_method='test_applies_ss_ssl_paths')), ('Merge overrides service from SharedServer', @@ -176,13 +184,48 @@ def test_overlays_kerberos_tags(self): def test_strips_owner_ssl_paths(self): result = self._merge() cp = result.connection_params - # Owner had sslkey, sslrootcert, sslcrl, sslcrldir, - # passfile — SharedServer did not — should be removed. + # Owner had sslkey, sslrootcert, sslcrl, sslcrldir — + # SharedServer did not — should be removed. self.assertNotIn('sslkey', cp) self.assertNotIn('sslcrl', cp) self.assertNotIn('sslcrldir', cp) self.assertNotIn('sslrootcert', cp) - self.assertNotIn('passfile', cp) + + def test_inherits_owner_passfile(self): + # A SharedServer row created before passfile started being + # copied has none of its own, and must still end up with the + # owner's: it is how the owner lets every user of the shared + # server authenticate without a password of their own. + result = self._merge() + self.assertEqual( + result.connection_params['passfile'], + '/home/owner/.pgpass') + + def test_shared_passfile_wins(self): + ss = _make_shared_server(connection_params={ + 'passfile': '/home/nonowner/.pgpass'}) + result = self._merge(ss=ss) + self.assertEqual( + result.connection_params['passfile'], + '/home/nonowner/.pgpass') + + def test_tags_fall_back_to_owner(self): + # NULL tags on the SharedServer means the row predates tags + # being copied across, so the owner's should show through. + owner_tags = [{'text': 'prod', 'color': '#f00'}] + result = self._merge( + server=_make_server(tags=owner_tags), + ss=_make_shared_server(tags=None)) + self.assertEqual(result.tags, owner_tags) + + def test_cleared_tags_are_respected(self): + # An empty list is a user who has removed every tag they had, + # which is not the same thing as never having had any, so the + # owner's tags must not come back. + result = self._merge( + server=_make_server(tags=[{'text': 'prod', 'color': '#f00'}]), + ss=_make_shared_server(tags=[])) + self.assertEqual(result.tags, []) def test_applies_ss_ssl_paths(self): result = self._merge() @@ -236,6 +279,10 @@ class TestCreateSharedServerSanitization(BaseTestGenerator): scenarios = [ ('Sanitizes connection_params on creation', dict(test_method='test_sanitizes_conn_params')), + ('Copies passfile on creation', + dict(test_method='test_copies_passfile')), + ('Copies tags from owner on creation', + dict(test_method='test_copies_tags')), ('Copies tunnel_port from owner', dict(test_method='test_copies_tunnel_port')), ('Copies tunnel_keep_alive from owner', @@ -273,9 +320,10 @@ def _create(self, server=None): def test_sanitizes_conn_params(self): self._create() cp = self.captured_kwargs.get('connection_params', {}) - # Sensitive keys must be stripped + # Personal SSL client cert/key paths must be stripped - + # each user configures their own. for key in ('sslcert', 'sslkey', 'sslrootcert', - 'sslcrl', 'sslcrldir', 'passfile'): + 'sslcrl', 'sslcrldir'): self.assertNotIn( key, cp, 'Sensitive key "{0}" should be stripped ' @@ -284,6 +332,25 @@ def test_sanitizes_conn_params(self): self.assertEqual(cp.get('sslmode'), 'verify-full') self.assertEqual(cp.get('connect_timeout'), '10') + def test_copies_passfile(self): + # passfile is how the owner (e.g. an admin provisioning + # servers.json) lets every user of a shared server + # authenticate automatically - it must be copied, unlike + # the other, genuinely personal, SSL file paths (#10137). + self._create() + cp = self.captured_kwargs.get('connection_params', {}) + self.assertEqual(cp.get('passfile'), '/home/owner/.pgpass') + + def test_copies_tags(self): + # Tags configured on the owner's server (e.g. via + # servers.json) must be visible to non-owners too (#10136). + server = _make_server( + tags=[{'text': 'prod', 'color': '#f00'}]) + self._create(server) + self.assertEqual( + self.captured_kwargs.get('tags'), + [{'text': 'prod', 'color': '#f00'}]) + def test_copies_tunnel_port(self): server = _make_server(tunnel_port=2222) self._create(server)