Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 28 additions & 9 deletions web/pgadmin/browser/server_groups/servers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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',
})

Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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',
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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',
Expand Down Expand Up @@ -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 '
Expand All @@ -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)
Expand Down
Loading