Prevent connection file paths and parameter keys escaping the user's storage directory. #10650

In server mode, get_complete_file_path() joined the user's storage
directory with the path they supplied after stripping only leading
slashes, so a password file, SSL certificate, key, CRL or SSH tunnel
identity file set to "../<another user>/.pgpass" (through the server
dialog, the API or a servers.json import) would connect with another
user's credentials or key. The sslkeylogfile parameter was copied into
the connection string unchanged, so a traversal path let libpq create or
append a file as the pgAdmin service account.

The resolved path, with symbolic links followed, must now stay inside
the user's storage directory, or inside the named shared storage
directory for a "Name:/file" path, as the file manager already requires.
sslkeylogfile is confined the same way, is refused in a shared storage
directory that is read-only for the user, and is no longer inherited by
non-owners of a shared server. A path that escapes, or whose storage
directory cannot be determined, raises PermissionError rather than
letting libpq fall back to its default file, and the message says where
the file has to live. Desktop mode is unchanged.

make_conninfo() quotes connection-parameter values but not their keys,
so a key such as 'sslmode=require passfile' smuggled its own
keyword=value pairs past these checks. Only exact libpq keywords, read
from the linked libpq at runtime, are now accepted as connection
parameter keys: when a server is created, updated, imported from
servers.json or used for an ad-hoc connection, and again when the
connection string is built, for rows saved earlier.

Connecting, the SSH tunnel and the utilities' job creation report a
refused path or key instead of proceeding, and a job whose environment
cannot be resolved is removed rather than left in the Processes list. A
server with a refused path or key can still be listed and edited to
correct it. On Windows, the short path of a not-yet-existing file is
resolved via its parent directory instead of being dropped.

Co-authored-by: Dave Page <dpage@pgadmin.org>
Co-authored-by: hiteshjambhale <hitesh.jambhale@enterprisedb.com>

docs/en_US/server_dialog.rst

@@ -156,6 +156,20 @@ icon that is located to the right of each of the following fields.
     means that administrators of the server may be able to access those files;
     appropriate caution should be taken before choosing to use this feature.
 
+.. note:: In Server mode, every connection file path (the *Password file*,
+    client certificate and key, root certificate, certificate revocation list,
+    CRL directory and SSL key log file) must resolve inside your storage
+    directory, or inside a *SHARED_STORAGE* entry (see :ref:`config_py`)
+    written as *Name:/path*. A path that leaves it through *..* or a symbolic
+    link is refused with an "Access denied (...): the path resolves outside
+    your storage directory..." message when connecting or starting a utility.
+    If you used a symbolic link in your storage directory to reach, for
+    example, */etc/pgadmin/certs*, configure that directory as a
+    *SHARED_STORAGE* entry instead (see :ref:`storage_manager`). Every pgAdmin
+    user can use a file in shared storage, so it should hold only shareable
+    files such as a root certificate or CRL, never a private key or a
+    *.pgpass* file.
+
 Click the *SSH Tunnel* tab to continue.
 
 .. image:: images/server_ssh_tunnel.png

web/pgadmin/browser/server_groups/servers/__init__.py

@@ -46,7 +46,8 @@ from sqlalchemy.orm import object_session
 from sqlalchemy.orm.attributes import flag_modified
 from pgadmin.utils.preferences import Preferences
 from .... import socketio as sio
-from pgadmin.utils import get_complete_file_path
+from pgadmin.utils import get_complete_file_path, \
+    get_invalid_connection_param_key
 from pgadmin.settings.utils import with_object_filters
 from pgadmin.utils.server_access import get_server, \
     get_user_server_query, get_server_group
@@ -63,6 +64,7 @@ from pgadmin.utils.server_access import get_server, \
 SENSITIVE_CONN_KEYS = frozenset({
     'passfile', 'sslcert', 'sslkey',
     'sslrootcert', 'sslcrl', 'sslcrldir',
+    'sslkeylogfile',
 })
 
 
@@ -1012,6 +1014,22 @@ class ServerNode(PGChildNodeView):
 
         # Update connection parameter if any.
         self.update_connection_parameter(data, server, sharedserver)
+
+        # Refuse a key that is not an exact libpq keyword (see create()).
+        # The merged parameters live in data['connection_params'] once
+        # update_connection_parameter() has applied the deltas.
+        if 'connection_params' in data:
+            invalid_key = get_invalid_connection_param_key(
+                data['connection_params'])
+            if invalid_key is not None:
+                return make_json_response(
+                    success=0,
+                    status=400,
+                    errormsg=gettext(
+                        'Invalid connection parameter: {0}').format(
+                            invalid_key)
+                )
+
         self.update_tags(data, sharedserver or server,
                          server.tags if sharedserver else None)
 
@@ -1392,6 +1410,19 @@ class ServerNode(PGChildNodeView):
         else:
             connection_params = convert_connection_parameter(raw_params)
 
+        # Refuse a key that is not an exact libpq keyword: make_conninfo()
+        # quotes values but not keys, so a key carrying its own
+        # "keyword=value" pairs would reach libpq as extra connection
+        # directions.
+        invalid_key = get_invalid_connection_param_key(connection_params)
+        if invalid_key is not None:
+            return make_json_response(
+                success=0,
+                status=400,
+                errormsg=gettext(
+                    'Invalid connection parameter: {0}').format(invalid_key)
+            )
+
         if 'hostaddr' in connection_params and \
                 not is_valid_ipaddress(connection_params['hostaddr']):
             return make_json_response(
@@ -1780,14 +1811,21 @@ class ServerNode(PGChildNodeView):
                     'passfile' in server.connection_params:
                 passfile_param = server.connection_params['passfile']
 
+            passfile_path = None
+            if passfile_param:
+                try:
+                    passfile_path = get_complete_file_path(passfile_param)
+                except PermissionError as e:
+                    return self.get_response_for_password(
+                        server, 401, False, prompt_tunnel_password, str(e))
+
             conn_passwd = getattr(conn, 'password', None)
             if conn_passwd is None and not server.save_password and \
                     passfile_param is None and \
                     manager.passexec is None and \
                     server.service is None:
                 prompt_password = True
-            elif passfile_param and passfile_param != '' and \
-                    get_complete_file_path(passfile_param):
+            elif passfile_path:
                 passfile = passfile_param
             else:
                 password = conn_passwd or server.password

web/pgadmin/browser/server_groups/servers/tests/test_connect_refused_passfile.py

@@ -0,0 +1,63 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""Tests that connecting to a server whose password file path is refused
+asks for a password and reports why, rather than connecting without it."""
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from pgadmin.utils.route import BaseTestGenerator
+
+MODULE = 'pgadmin.browser.server_groups.servers'
+PASSFILE = '../bob/.pgpass'
+
+
+class ServerConnectRefusedPassfileTestCase(BaseTestGenerator):
+    """The connect route returns 401 with the refusal message."""
+
+    scenarios = [
+        ('A refused passfile is reported, not connected with', dict()),
+    ]
+
+    def setUp(self):
+        # No test server is needed: everything the route reads is mocked.
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        server = SimpleNamespace(
+            id=1, name='s', username='postgres', shared=False, service=None,
+            use_ssh_tunnel=False, kerberos_conn=False, save_password=False,
+            password=None, connection_params={'passfile': PASSFILE})
+        manager = MagicMock()
+        manager.connection.return_value.connected.return_value = False
+        manager.passexec = None
+        driver = MagicMock()
+        driver.connection_manager.return_value = manager
+        refusal = PermissionError(
+            'Access denied ({0}): the path resolves outside your storage '
+            'directory, or outside the shared storage directory it '
+            'names.'.format(PASSFILE))
+
+        with patch(MODULE + '.get_server', return_value=server), \
+                patch(MODULE + '._is_non_owner', return_value=False), \
+                patch(MODULE + '.get_crypt_key',
+                      return_value=(True, b'k')), \
+                patch(MODULE + '.get_driver', return_value=driver), \
+                patch(MODULE + '.get_complete_file_path',
+                      side_effect=refusal):
+            response = self.tester.post(
+                '/browser/server/connect/1/1', data='{}',
+                content_type='application/json')
+
+        self.assertEqual(response.status_code, 401)
+        errmsg = response.json['result']['errmsg']
+        self.assertIn(PASSFILE, errmsg)
+        manager.connection.return_value.connect.assert_not_called()

web/pgadmin/misc/bgprocess/processes.py

@@ -97,7 +97,10 @@ class IProcessDesc(metaclass=ABCMeta):
                 # do nothing if bfile is not encoded.
                 pass
 
-            path = get_complete_file_path(process_file)
+            try:
+                path = get_complete_file_path(process_file)
+            except PermissionError:
+                path = None
             path = process_file if path is None else path
 
             if IS_WIN:
@@ -261,6 +264,10 @@ class BatchProcess:
         )
         db.session.add(j)
         db.session.commit()
+        # Process.pid is a string column: PostgreSQL does not compare it
+        # with an integer, so keep it as the string that was stored.
+        self._unstarted_pid = str(uid)
+        self._unstarted_uid = current_user.id
 
     def check_start_end_time(self):
         """
@@ -849,6 +856,43 @@ class BatchProcess:
         db.session.commit()
 
     def set_env_variables(self, server, **kwargs):
+        """
+        Set environment variables.
+
+        A job whose environment cannot be resolved (for instance because a
+        connection file path is refused) can never start, so its row and log
+        directory are removed rather than left in the Processes list.
+        """
+        try:
+            self._set_env_variables(server, **kwargs)
+        except Exception:
+            self._discard_unstarted()
+            raise
+        # The environment is resolved: this job is no longer "unstarted",
+        # so a later call can never remove the row of a running job.
+        self._unstarted_pid = None
+
+    def _discard_unstarted(self):
+        """Remove the row and log directory of a job that never started.
+        This must never raise, so as not to mask the original error."""
+        pid = getattr(self, '_unstarted_pid', None)
+        if pid is None:
+            return
+        self._unstarted_pid = None
+        try:
+            p = Process.query.filter_by(
+                pid=pid, user_id=self._unstarted_uid).first()
+            if p is not None:
+                db.session.delete(p)
+                db.session.commit()
+            if self.log_dir:
+                shutil.rmtree(self.log_dir, True)
+        except Exception as e:
+            db.session.rollback()
+            current_app.logger.warning(
+                'Could not discard the unstarted process %s: %s', pid, e)
+
+    def _set_env_variables(self, server, **kwargs):
         """Set environment variables"""
         # The connection parameters configured against the server are what
         # the utility must connect with, so pass every one of them that

web/pgadmin/misc/bgprocess/tests/test_set_env_variables_refused.py

@@ -0,0 +1,105 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""Tests that a job whose environment cannot be resolved (a refused
+connection file path) leaves no never-started row behind."""
+
+import os
+import shutil
+import tempfile
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from pgadmin.misc.bgprocess.processes import BatchProcess, IProcessDesc
+from pgadmin.model import Process, User, db
+from pgadmin.utils.route import BaseTestGenerator
+
+
+class _Desc(IProcessDesc):
+    @property
+    def message(self):
+        return 'refused job'
+
+    def details(self, cmd, args):
+        return {'message': 'refused job', 'cmd': cmd, 'query': None,
+                'info': ''}
+
+
+class TestSetEnvVariablesRefusedDiscardsJob(BaseTestGenerator):
+    """set_env_variables() removes the job's row and log directory when it
+    raises, and keeps them when it succeeds."""
+
+    scenarios = [
+        ('A refused passfile discards the job',
+         dict(passfile='../bob/.pgpass', refused=True)),
+        ('A passfile in storage keeps the job',
+         dict(passfile='/.pgpass', refused=False)),
+    ]
+
+    def setUp(self):
+        # No test server is needed.
+        unittest.TestCase.setUp(self)
+        self.outer = os.path.realpath(
+            tempfile.mkdtemp(prefix='pga_set_env_'))
+        self.alice = os.path.join(self.outer, 'alice')
+        for d in ('alice', 'bob'):
+            os.makedirs(os.path.join(self.outer, d))
+            with open(os.path.join(self.outer, d, '.pgpass'), 'w') as f:
+                f.write('*:*:*:*:secret\n')
+        self.pid = None
+        self.user_id = None
+
+    def tearDown(self):
+        if self.pid is not None:
+            with self.app.app_context():
+                p = Process.query.filter_by(pid=self.pid).first()
+                if p is not None:
+                    db.session.delete(p)
+                    db.session.commit()
+        if getattr(self, 'log_dir', None):
+            shutil.rmtree(self.log_dir, True)
+        shutil.rmtree(self.outer, ignore_errors=True)
+
+    def runTest(self):
+        params = {'passfile': self.passfile}
+        manager = MagicMock(connection_params=params, use_ssh_tunnel=False,
+                            service=None, kerberos_conn=False)
+        with self.app.test_request_context(), \
+                patch.dict(self.app.config, {'SERVER_MODE': True}), \
+                patch.object(self.app, 'PGADMIN_RUNTIME', False), \
+                patch('pgadmin.utils.get_storage_directory',
+                      return_value=self.alice), \
+                patch('pgadmin.misc.bgprocess.processes.current_user',
+                      SimpleNamespace(id=User.query.first().id)):
+            p = BatchProcess(desc=_Desc(), cmd='/bin/true', args=[],
+                             manager_obj=manager)
+            # The stored pid, as a string: Process.pid is a string column.
+            self.pid = p._unstarted_pid
+            self.assertIsInstance(self.pid, str)
+            self.log_dir = p.log_dir
+            self.assertTrue(os.path.isdir(p.log_dir))
+
+            if self.refused:
+                with self.assertRaises(PermissionError):
+                    p.set_env_variables(None)
+                self.assertIsNone(
+                    Process.query.filter_by(pid=self.pid).first())
+                self.assertFalse(os.path.exists(p.log_dir))
+            else:
+                p.set_env_variables(None)
+                self.assertIsNotNone(
+                    Process.query.filter_by(pid=self.pid).first())
+                self.assertTrue(os.path.isdir(p.log_dir))
+                # Once the environment is resolved the job is no longer
+                # "unstarted": a later failing call must leave its row.
+                self.assertIsNone(p._unstarted_pid)
+                p._discard_unstarted()
+                self.assertIsNotNone(
+                    Process.query.filter_by(pid=self.pid).first())

web/pgadmin/misc/workspaces/__init__.py

@@ -15,7 +15,7 @@ from flask import request, current_app
 from pgadmin.user_login_check import pga_login_required
 from flask_babel import gettext
 from flask_security import current_user
-from pgadmin.utils import PgAdminModule
+from pgadmin.utils import PgAdminModule, get_invalid_connection_param_key
 from pgadmin.model import db, Server
 from pgadmin.utils.server_access import get_server
 from pgadmin.utils.driver import get_driver
@@ -103,6 +103,19 @@ def adhoc_connect_server():
         connection_params = convert_connection_parameter(raw_params)
 
     if connection_params is not None:
+        # Refuse a key that is not an exact libpq keyword: make_conninfo()
+        # quotes values but not keys, so a key carrying its own
+        # "keyword=value" pairs would reach libpq as extra connection
+        # directions.
+        invalid_key = get_invalid_connection_param_key(connection_params)
+        if invalid_key is not None:
+            return make_json_response(
+                success=0,
+                status=400,
+                errormsg=gettext(
+                    'Invalid connection parameter: {0}').format(invalid_key)
+            )
+
         if 'hostaddr' in connection_params and \
                 not is_valid_ipaddress(connection_params['hostaddr']):
             return make_json_response(

web/pgadmin/settings/__init__.py

@@ -332,7 +332,7 @@ def save_application_state():
         if 'connection_info' in data else None
     if (connection_info and 'open_file_name' in connection_info and
             connection_info['open_file_name']):
-        file_path = get_complete_file_path(connection_info['open_file_name'])
+        file_path = _get_open_file_path(connection_info['open_file_name'])
         connection_info['last_saved_file_hash'] = (
             get_last_saved_file_hash(file_path, trans_id))
 
@@ -354,6 +354,17 @@ def save_application_state():
         })
 
 
+def _get_open_file_path(open_file_name):
+    """
+    Resolve the file a tool has open, treating one outside the user's
+    storage directory as though it had been deleted.
+    """
+    try:
+        return get_complete_file_path(open_file_name)
+    except PermissionError:
+        return None
+
+
 def get_last_saved_file_hash(file_path, trans_id):
     result = db.session \
         .query(ApplicationState) \
@@ -369,7 +380,7 @@ def get_last_saved_file_hash(file_path, trans_id):
             file_hash_update_require = not connection_info['is_editor_dirty']
             last_saved_file_hash = connection_info['last_saved_file_hash']
 
-    if file_hash_update_require:
+    if file_hash_update_require and file_path:
         last_saved_file_hash = compute_md5_hash_file(file_path)
 
     return last_saved_file_hash
@@ -394,7 +405,7 @@ def get_application_state():
         connection_info = row.connection_info
         if (connection_info and 'open_file_name' in connection_info and
                 connection_info['open_file_name']):
-            file_path = get_complete_file_path(
+            file_path = _get_open_file_path(
                 connection_info['open_file_name'])
             file_deleted = False if file_path else True
             connection_info['file_deleted'] = file_deleted
@@ -438,7 +449,7 @@ def get_tool_data(trans_id):
         if (connection_info and 'open_file_name' in connection_info and
                 connection_info['open_file_name']):
             file_path = (
-                get_complete_file_path(connection_info['open_file_name']))
+                _get_open_file_path(connection_info['open_file_name']))
             file_deleted = False if file_path else True
             connection_info['file_deleted'] = file_deleted
 

web/pgadmin/utils/__init__.py

@@ -15,6 +15,7 @@ import json
 import subprocess
 import re
 from collections import defaultdict
+from functools import lru_cache
 from operator import attrgetter
 
 from pathlib import Path
@@ -272,16 +273,27 @@ def get_directory_and_file_name(drivefilepath):
     return dir_name, file_name
 
 
-def get_complete_file_path(file, validate=True):
+def get_complete_file_path(file, validate=True, for_write=False):
     """
     Args:
         file: File returned by file manager
         validate: if True, return None unless the path names an existing
             file; pass False for a directory, or for a file that may not
             exist yet
+        for_write: set True when libpq will write the file (e.g.
+            sslkeylogfile) rather than read it. A write target must
+            additionally honour a shared storage directory that is
+            read-only for non-administrators, and must resolve its short
+            path on Windows even though the file does not exist yet.
 
     Returns:
          Full path for the file
+
+    Raises:
+        PermissionError: in server mode, if the path resolves to somewhere
+            outside the user's storage directory (or the named shared
+            storage directory), or names a read-only shared storage
+            directory when for_write is set
     """
     if not file:
         return None
@@ -292,19 +304,71 @@ def get_complete_file_path(file, validate=True):
             return file
         return file if os.path.isfile(file) else None
 
+    orig_file = file
+
     # get dir name and file name
     dir_name, file = get_directory_and_file_name(file)
 
     storage_dir = get_storage_directory(shared_storage=dir_name) if dir_name \
         else get_storage_directory()
-    if storage_dir:
-        file = os.path.join(
-            storage_dir,
-            file.lstrip('/').lstrip('\\')
-        )
-        if IS_WIN:
-            file = file.replace('\\', '/')
-            file = fs_short_path(file)
+
+    # Without a storage directory to anchor it, the path would be used
+    # exactly as given, so refuse it rather than let it name any file the
+    # pgAdmin service account can read.
+    if not storage_dir:
+        raise PermissionError(
+            gettext("Access denied ({0}): the storage directory it refers "
+                    "to could not be found.").format(orig_file))
+
+    # A write target (e.g. sslkeylogfile) must honour a shared storage
+    # directory that is read-only for non-administrators; get_storage_
+    # directory() itself does not enforce that flag, so a non-admin could
+    # otherwise create or append a file there.
+    if for_write and dir_name:
+        shared_dir = next(
+            (sdir for sdir in current_app.config['SHARED_STORAGE']
+             if sdir['name'] == dir_name), None)
+        if shared_dir and shared_dir['restricted_access'] and \
+                not current_user.has_role("Administrator"):
+            raise PermissionError(
+                gettext("Access denied ({0}): the shared storage directory "
+                        "it names is read-only.").format(orig_file))
+
+    file = os.path.join(
+        storage_dir,
+        file.lstrip('/').lstrip('\\')
+    )
+
+    # The path is the user's to choose, and is used for things like the
+    # password file and SSL key libpq connects with, so it must not climb
+    # out of their storage directory with "..", an absolute path or a
+    # symbolic link. This mirrors Filemanager.check_access_permission().
+    try:
+        Path(os.path.realpath(file)).relative_to(
+            os.path.realpath(storage_dir))
+    except ValueError:
+        current_app.logger.warning(
+            'Refusing the path %r, which resolves outside the storage '
+            'directory.', orig_file)
+        raise PermissionError(
+            gettext("Access denied ({0}): the path resolves outside your "
+                    "storage directory, or outside the shared storage "
+                    "directory it names.").format(orig_file))
+
+    if IS_WIN:
+        file = file.replace('\\', '/')
+        short_path = fs_short_path(file)
+        # fs_short_path() returns an empty string on Windows when the file
+        # does not exist yet. For a write target (sslkeylogfile) that is
+        # the normal case, so fall back to the short path of the parent
+        # directory joined with the file name, instead of silently
+        # dropping the path.
+        if for_write and short_path == '':
+            base_name = os.path.basename(file)
+            dir_short_path = fs_short_path(os.path.dirname(file))
+            if dir_short_path != '':
+                short_path = os.path.join(dir_short_path, base_name)
+        file = short_path
 
     if validate:
         return file if os.path.isfile(file) else None
@@ -327,6 +391,52 @@ def connection_param_bool_to_int(value):
     return 1 if value else 0
 
 
+@lru_cache(maxsize=1)
+def get_libpq_connection_keywords():
+    """
+    Return the set of connection-parameter keywords recognised by the
+    libpq that psycopg is linked against.
+
+    This is read from libpq itself rather than hard-coded, so it always
+    matches the installed version; a keyword unknown to this libpq is not
+    treated as valid, since make_conninfo() would reject it anyway.
+    """
+    from psycopg.pq import Conninfo
+    return frozenset(
+        defn.keyword.decode() for defn in Conninfo.get_defaults()
+    )
+
+
+def is_valid_connection_param_key(key):
+    """
+    Return True only if key is an exact libpq connection keyword.
+
+    psycopg's make_conninfo() quotes parameter values but not their keys,
+    so a key that smuggles its own "keyword=value" pairs (for example via
+    an '=' sign or embedded/leading/trailing whitespace) would reach libpq
+    as extra, unconfined connection directions. Accepting only exact libpq
+    keywords closes that injection.
+    """
+    return isinstance(key, str) and key in get_libpq_connection_keywords()
+
+
+def get_invalid_connection_param_key(connection_params):
+    """
+    Return the first key in connection_params that is not a valid libpq
+    keyword, or None if every key is valid (or there are no params).
+
+    Callers use this to refuse a server whose connection parameters carry
+    an injected key, before it is saved or used to build a connection
+    string.
+    """
+    if not connection_params or not isinstance(connection_params, dict):
+        return None
+    for key in connection_params:
+        if not is_valid_connection_param_key(key):
+            return key
+    return None
+
+
 def connection_params_to_env(connection_params):
     """
     Translate the connection parameters configured against a server into the
@@ -1116,6 +1226,20 @@ def load_database_servers(input_file, selected_servers,
 
                 new_server.connection_params = conn_param
 
+            # Refuse a key that is not an exact libpq keyword:
+            # make_conninfo() quotes values but not keys, so a key carrying
+            # its own "keyword=value" pairs would reach libpq as extra
+            # connection directions.
+            invalid_key = get_invalid_connection_param_key(
+                new_server.connection_params)
+            if invalid_key is not None:
+                if from_setup:
+                    print(ADD_SERVERS_MSG % (groups_added, servers_added))
+                return _handle_error(
+                    gettext(
+                        "Invalid connection parameter '%s' for server '%s'" %
+                        (invalid_key, new_server.name)), from_setup)
+
             new_server.bgcolor = obj.get("BGColor", None)
 
             new_server.fgcolor = obj.get("FGColor", None)

web/pgadmin/utils/driver/psycopg3/connection.py

@@ -605,6 +605,11 @@ class Connection(BaseConnection):
                         cursor_factory=DictCursor,
                         prepare_threshold=manager.prepare_threshold)
 
+        except PermissionError as e:
+            # A connection file path that escapes the user's storage
+            # directory (see get_complete_file_path()).
+            manager.stop_ssh_tunnel()
+            return False, str(e)
         except psycopg.Error as e:
             manager.stop_ssh_tunnel()
             if hasattr(e, 'pgerror'):
@@ -1786,6 +1791,8 @@ WHERE db.datname = current_database()""")
 
                 pg_conn = psycopg.connect(connection_string,
                                           cursor_factory=DictCursor)
+        except PermissionError as e:
+            return False, str(e)
         except psycopg.Error as e:
             if hasattr(e, 'pgerror'):
                 msg = e.pgerror
@@ -2020,6 +2027,9 @@ Failed to reset the connection to the server due to following error:
                 # Close the connection
                 pg_conn.close()
 
+            except PermissionError as e:
+                status = False
+                msg = str(e)
             except psycopg.Error as e:
                 status = False
                 if hasattr(e, 'pgerror'):

web/pgadmin/utils/driver/psycopg3/server_manager.py

@@ -20,7 +20,7 @@ from flask_babel import gettext
 from werkzeug.exceptions import InternalServerError
 
 from pgadmin.utils import get_complete_file_path, \
-    connection_param_bool_to_int
+    connection_param_bool_to_int, is_valid_connection_param_key
 from pgadmin.utils.crypto import decrypt
 from pgadmin.utils.master_password import process_masterpass_disabled
 from .connection import Connection
@@ -118,7 +118,14 @@ class ServerManager(object):
         self.gss_authenticated = False
         self.gss_encrypted = False
         self.connection_params = server.connection_params
-        self.create_connection_string(self.db, self.user)
+        try:
+            self.create_connection_string(self.db, self.user)
+        except PermissionError:
+            # A file path that escapes the user's storage directory, or a
+            # connection parameter whose key is not a libpq keyword, is
+            # reported when a connection is attempted; it must not stop the
+            # server being listed, or edited to correct the value.
+            pass
         self.prepare_threshold = server.prepare_threshold
         self.post_connection_sql = server.post_connection_sql
 
@@ -610,8 +617,11 @@ WHERE db.oid = {0}""".format(did))
             # preserving the previous behaviour for anyone relying on
             # "Prompt for Password?" or on the agent itself.
             if self.tunnel_authentication == 1:
-                tunnel_identity_file = get_complete_file_path(
-                    self.tunnel_identity_file)
+                try:
+                    tunnel_identity_file = get_complete_file_path(
+                        self.tunnel_identity_file)
+                except PermissionError as e:
+                    return False, str(e)
                 self.tunnel_object = SSHTunnelForwarder(
                     (self.tunnel_host, int(self.tunnel_port)),
                     ssh_username=self.tunnel_username,
@@ -704,9 +714,26 @@ WHERE db.oid = {0}""".format(did))
             display_dsn_args['password'] = 'xxxxxxx'
             dsn_args['password'] = password
 
+        # A file path that escapes the user's storage directory is refused,
+        # but only once the display string below has been built.
+        path_error = None
+        # A connection parameter whose key is not an exact libpq keyword is
+        # refused the same way (see below).
+        key_error = None
+
         # Loop through all the connection parameters set in the server dialog.
         if self.connection_params and isinstance(self.connection_params, dict):
             for key, value in self.connection_params.items():
+                # make_conninfo() quotes values but not keys, so a key that
+                # smuggles its own "keyword=value" pairs would reach libpq
+                # unconfined. Rows already in the database are never
+                # re-validated on save, so refuse any key that is not an
+                # exact libpq keyword here as well, and keep it out of both
+                # the connection string and its display copy.
+                if not is_valid_connection_param_key(key):
+                    key_error = key_error or key
+                    continue
+
                 with_complete_path = False
                 orig_value = value
                 # Getting complete file path if the key is one of the below.
@@ -714,7 +741,23 @@ WHERE db.oid = {0}""".format(did))
                            'sslcrldir'] or \
                         (key == 'sslrootcert' and value != 'system'):
                     with_complete_path = True
-                    value = get_complete_file_path(value)
+                    try:
+                        value = get_complete_file_path(value)
+                    except PermissionError as e:
+                        path_error = path_error or e
+                        value = None
+                elif key == 'sslkeylogfile':
+                    # libpq writes this file (it may not exist yet), so skip
+                    # the existence check, but still confine it to the user's
+                    # storage directory and refuse a read-only shared
+                    # storage directory (for_write).
+                    with_complete_path = True
+                    try:
+                        value = get_complete_file_path(
+                            value, validate=False, for_write=True)
+                    except PermissionError as e:
+                        path_error = path_error or e
+                        value = None
 
                 # If key is hostaddr and ssh tunnel is in use don't overwrite.
                 if key == 'hostaddr' and self.use_ssh_tunnel:
@@ -747,4 +790,18 @@ WHERE db.oid = {0}""".format(did))
 
         self.display_connection_string = make_conninfo(**display_dsn_args)
 
+        # Raised as PermissionError, like a refused path, so the same guard
+        # in __init__ defers it: a server with an invalid saved key can
+        # still be listed and edited to correct it, while an actual
+        # connection attempt is refused.
+        if key_error is not None:
+            raise PermissionError(
+                gettext(
+                    'Invalid connection parameter: {0}'
+                ).format(key_error)
+            )
+
+        if path_error is not None:
+            raise path_error
+
         return make_conninfo(**dsn_args)

web/pgadmin/utils/tests/test_complete_file_path.py

@@ -0,0 +1,412 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""Tests that get_complete_file_path() keeps a path inside the storage
+directory in server mode.
+
+The path is the user's to choose, and it is what libpq is given as the
+password file, SSL key and so on, so a path that climbs out of the user's
+storage directory would let them connect with another user's credentials.
+"""
+
+import os
+import shutil
+import tempfile
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import flask
+
+import config
+from pgadmin.utils import (
+    connection_params_to_env, get_complete_file_path)
+from pgadmin.utils.route import BaseTestGenerator
+
+import pgadmin.utils.driver.psycopg3.server_manager as server_manager
+
+
+class _StorageMixin:
+    """Builds a storage root holding two users and a shared directory."""
+
+    def setUp(self):
+        # Filesystem-only tests: skip BaseTestGenerator's server setup.
+        unittest.TestCase.setUp(self)
+        self.outer = os.path.realpath(
+            tempfile.mkdtemp(prefix='pga_complete_path_'))
+        self.alice = os.path.join(self.outer, 'storage', 'alice')
+        self.bob = os.path.join(self.outer, 'storage', 'bob')
+        self.shared = os.path.join(self.outer, 'shared')
+        for d in (self.alice, self.bob, self.shared):
+            os.makedirs(d)
+        os.makedirs(os.path.join(self.alice, 'sub'))
+        for d in (self.alice, self.bob, self.shared):
+            with open(os.path.join(d, '.pgpass'), 'w') as f:
+                f.write('*:*:*:*:secret\n')
+        self.shared_storage = [{'name': 'Team', 'path': self.shared,
+                                'restricted_access': False}]
+
+    def tearDown(self):
+        shutil.rmtree(self.outer, ignore_errors=True)
+
+    def _storage_dir(self, user=None, shared_storage=''):
+        if shared_storage:
+            for sdir in self.shared_storage:
+                if sdir['name'] == shared_storage:
+                    return sdir['path']
+            return None
+        return self.alice
+
+    def _resolve(self, path, validate=True, server_mode=True):
+        with self.app.test_request_context(), \
+                patch.dict(self.app.config, {'SERVER_MODE': server_mode}), \
+                patch.object(self.app, 'PGADMIN_RUNTIME', False), \
+                patch.object(config, 'SHARED_STORAGE', self.shared_storage), \
+                patch('pgadmin.utils.get_storage_directory',
+                      side_effect=self._storage_dir):
+            return get_complete_file_path(path, validate=validate)
+
+
+class CompleteFilePathContainmentTestCase(_StorageMixin, BaseTestGenerator):
+    """Paths that stay inside the storage directory resolve; any that
+    escape it are refused rather than resolved or silently dropped."""
+
+    scenarios = [
+        ('A file in the user\'s own storage resolves',
+         dict(path='/.pgpass', expected='storage/alice/.pgpass')),
+        ('A leading slash is relative to the storage directory',
+         dict(path='.pgpass', expected='storage/alice/.pgpass')),
+        ('A missing file in the user\'s own storage is None',
+         dict(path='/missing', expected=None)),
+        ('A missing file resolves when not validated',
+         dict(path='/missing', validate=False,
+              expected='storage/alice/missing')),
+        ('".." inside the storage directory is harmless',
+         dict(path='/sub/../.pgpass', expected='storage/alice/.pgpass')),
+        ('A file in the named shared storage resolves',
+         dict(path='Team:/.pgpass', expected='shared/.pgpass')),
+        ('Another user\'s file is refused',
+         dict(path='../bob/.pgpass', denied=True)),
+        ('Another user\'s file is refused with a leading slash',
+         dict(path='/../bob/.pgpass', denied=True)),
+        ('Another user\'s missing file is refused, not None',
+         dict(path='../bob/missing', denied=True)),
+        ('Another user\'s file is refused even unvalidated',
+         dict(path='../bob/.pgpass', validate=False, denied=True)),
+        ('Climbing out of shared storage is refused',
+         dict(path='Team:/../storage/bob/.pgpass', denied=True)),
+        ('A drive name not at the start of the path is refused',
+         dict(path='x/Team:/.pgpass', denied=True)),
+        ('A symbolic link out of the storage directory is refused',
+         dict(path='/link-to-bob/.pgpass', link=('link-to-bob', 'bob'),
+              denied=True)),
+        ('A symbolic link within the storage directory resolves',
+         dict(path='/link-to-sub/f', link=('link-to-sub', 'alice/sub'),
+              expected='storage/alice/link-to-sub/f')),
+    ]
+
+    validate = True
+    denied = False
+    link = None
+
+    def runTest(self):
+        if self.link:
+            if not hasattr(os, 'symlink'):
+                self.skipTest('Symbolic links are not supported here.')
+            name, target = self.link
+            target = os.path.join(self.outer, 'storage', target)
+            os.makedirs(target, exist_ok=True)
+            with open(os.path.join(target, 'f'), 'w') as f:
+                f.write('x')
+            os.symlink(target, os.path.join(self.alice, name))
+
+        if self.denied:
+            with self.assertRaises(PermissionError):
+                self._resolve(self.path, validate=self.validate)
+            return
+
+        result = self._resolve(self.path, validate=self.validate)
+        if self.expected is None:
+            self.assertIsNone(result)
+        else:
+            # Expected paths are given relative to the temporary root.
+            expected = os.path.join(self.outer, *self.expected.split('/'))
+            self.assertEqual(os.path.normpath(result),
+                             os.path.normpath(expected))
+
+
+class CompleteFilePathDesktopTestCase(_StorageMixin, BaseTestGenerator):
+    """Desktop mode has no storage directory, so the path is used as is."""
+
+    scenarios = [
+        ('Desktop mode uses the path as given', dict()),
+    ]
+
+    def runTest(self):
+        path = os.path.join(self.alice, '..', 'bob', '.pgpass')
+        self.assertEqual(self._resolve(path, server_mode=False), path)
+
+
+class ConnectionStringPathRefusalTestCase(_StorageMixin, BaseTestGenerator):
+    """A refused connection file path stops the connection string being
+    built, so libpq never falls back to its default file in its place, but
+    the display string is still available for the server's properties."""
+
+    scenarios = [
+        ('A refused passfile', dict(param='passfile')),
+        ('A refused SSL key', dict(param='sslkey')),
+        ('A refused SSL certificate', dict(param='sslcert')),
+        ('A refused SSL key log file', dict(param='sslkeylogfile')),
+        ('A refused root certificate', dict(param='sslrootcert')),
+        ('A refused CRL', dict(param='sslcrl')),
+        ('A refused CRL directory', dict(param='sslcrldir')),
+    ]
+
+    # The libpq version that first knows each connection keyword; on an
+    # older libpq make_conninfo() rejects it before the containment check,
+    # so there is nothing to confine.
+    min_libpq = {'sslkeylogfile': 180000, 'sslcrldir': 140000}
+
+    def runTest(self):
+        if self.param in self.min_libpq:
+            import psycopg
+            if psycopg.pq.version() < self.min_libpq[self.param]:
+                self.skipTest('{0} needs libpq {1} or later'.format(
+                    self.param, self.min_libpq[self.param]))
+        manager = server_manager.ServerManager.__new__(
+            server_manager.ServerManager)
+        manager.host = 'db.example.com'
+        manager.port = 5432
+        manager.use_ssh_tunnel = False
+        manager.service = None
+        manager.connection_params = {
+            'sslmode': 'require', self.param: '../bob/.pgpass'}
+
+        with self.app.test_request_context(), \
+                patch.dict(self.app.config, {'SERVER_MODE': True}), \
+                patch.object(self.app, 'PGADMIN_RUNTIME', False), \
+                patch('pgadmin.utils.get_storage_directory',
+                      side_effect=self._storage_dir):
+            with self.assertRaises(PermissionError):
+                manager.create_connection_string('postgres', 'alice')
+
+        self.assertIn('../bob/.pgpass', manager.display_connection_string)
+
+
+class ConnectionParamsToEnvPathRefusalTestCase(_StorageMixin,
+                                               BaseTestGenerator):
+    """connection_params_to_env() refuses a connection file path that
+    leaves the storage directory, and accepts the rest."""
+
+    scenarios = [
+        ('A refused passfile',
+         dict(param='passfile', value='../bob/.pgpass', denied=True)),
+        ('A refused SSL certificate',
+         dict(param='sslcert', value='../bob/.pgpass', denied=True)),
+        ('A refused SSL key',
+         dict(param='sslkey', value='../bob/.pgpass', denied=True)),
+        ('A refused CRL',
+         dict(param='sslcrl', value='../bob/.pgpass', denied=True)),
+        ('A refused root certificate',
+         dict(param='sslrootcert', value='../bob/.pgpass', denied=True)),
+        ('A refused CRL directory',
+         dict(param='sslcrldir', value='../bob', denied=True)),
+        ('The system root certificate is not a path',
+         dict(param='sslrootcert', value='system', denied=False)),
+        ('A passfile in storage is accepted',
+         dict(param='passfile', value='/.pgpass', denied=False)),
+    ]
+
+    def runTest(self):
+        with self.app.test_request_context(), \
+                patch.dict(self.app.config, {'SERVER_MODE': True}), \
+                patch.object(self.app, 'PGADMIN_RUNTIME', False), \
+                patch.object(config, 'SHARED_STORAGE', self.shared_storage), \
+                patch('pgadmin.utils.get_storage_directory',
+                      side_effect=self._storage_dir):
+            if self.denied:
+                with self.assertRaises(PermissionError):
+                    connection_params_to_env({self.param: self.value})
+            else:
+                connection_params_to_env({self.param: self.value})
+
+
+class CompleteFilePathRealStorageDirTestCase(_StorageMixin,
+                                             BaseTestGenerator):
+    """get_complete_file_path() against the real get_storage_directory(),
+    so the user's directory and the shared storage lookup are exercised
+    rather than mocked."""
+
+    scenarios = [
+        ('A file in the user\'s storage resolves', dict()),
+    ]
+
+    def _resolve_real(self, path):
+        with patch.object(config, 'SERVER_MODE', True), \
+                patch.object(config, 'STORAGE_DIR',
+                             os.path.join(self.outer, 'storage')), \
+                patch.object(config, 'SHARED_STORAGE', self.shared_storage), \
+                patch.dict(self.app.config, {'SERVER_MODE': True}), \
+                patch.object(self.app, 'PGADMIN_RUNTIME', False), \
+                self.app.test_request_context():
+            # current_user is a proxy for flask.g._login_user.
+            flask.g._login_user = SimpleNamespace(username='alice')
+            return get_complete_file_path(path)
+
+    def runTest(self):
+        self.assertEqual(
+            os.path.normpath(self._resolve_real('/.pgpass')),
+            os.path.join(self.alice, '.pgpass'))
+
+        with self.assertRaises(PermissionError) as cm:
+            self._resolve_real('../bob/.pgpass')
+        self.assertIn('resolves outside your storage directory',
+                      str(cm.exception))
+
+        self.assertEqual(
+            os.path.normpath(self._resolve_real('Team:/.pgpass')),
+            os.path.join(self.shared, '.pgpass'))
+
+        # A name that is not configured is just part of a relative path.
+        self.assertIsNone(self._resolve_real('Nope:/.pgpass'))
+
+        # A shared storage entry without a usable path has no directory.
+        self.shared_storage.append(
+            {'name': 'Gone', 'path': None, 'restricted_access': False})
+        with self.assertRaises(PermissionError) as cm:
+            self._resolve_real('Gone:/.pgpass')
+        self.assertIn('could not be found', str(cm.exception))
+
+        self.assertIsNone(self._resolve_real('/etc/passwd'))
+
+
+class SSHTunnelIdentityFileRefusalTestCase(_StorageMixin, BaseTestGenerator):
+    """A refused tunnel identity file is reported, not used or raised."""
+
+    scenarios = [
+        ('A refused identity file', dict()),
+    ]
+
+    def runTest(self):
+        if not config.SUPPORT_SSH_TUNNEL:
+            self.skipTest('SSH tunnelling is disabled in this configuration.')
+
+        manager = server_manager.ServerManager.__new__(
+            server_manager.ServerManager)
+        manager.tunnel_authentication = 1
+        manager.tunnel_host = 'tunnel.example.com'
+        manager.tunnel_port = 22
+        manager.tunnel_username = 'tunneluser'
+        manager.tunnel_identity_file = '../bob/.pgpass'
+        manager.tunnel_keep_alive = 0
+        manager.host = 'db.example.com'
+        manager.port = 5432
+        manager.tunnel_object = None
+        manager.tunnel_created = False
+        forwarder = MagicMock()
+
+        with self.app.test_request_context(), \
+                patch.dict(self.app.config, {'SERVER_MODE': True}), \
+                patch.object(self.app, 'PGADMIN_RUNTIME', False), \
+                patch('pgadmin.utils.get_storage_directory',
+                      side_effect=self._storage_dir), \
+                patch.object(server_manager, 'SSHTunnelForwarder',
+                             forwarder), \
+                patch.object(server_manager, 'User', MagicMock()), \
+                patch.object(server_manager, 'current_user', MagicMock()):
+            success, error = manager.create_ssh_tunnel(None)
+
+        self.assertFalse(success)
+        self.assertIn('../bob/.pgpass', error)
+        forwarder.assert_not_called()
+
+
+class ConnectionStringInvalidKeyTestCase(_StorageMixin, BaseTestGenerator):
+    """make_conninfo() quotes values but not keys, so a connection
+    parameter whose key is not an exact libpq keyword (e.g. one carrying
+    its own "keyword=value" pairs) is refused when the connection string
+    is built, since rows already in the database are never re-validated."""
+
+    scenarios = [
+        ('A key smuggling a keyword via "="',
+         dict(bad_key='sslmode=require passfile')),
+        ('A key with a leading space',
+         dict(bad_key=' passfile')),
+        ('A key with a trailing space',
+         dict(bad_key='passfile ')),
+        ('A key that is not a libpq keyword at all',
+         dict(bad_key='not_a_real_keyword')),
+    ]
+
+    def runTest(self):
+        manager = server_manager.ServerManager.__new__(
+            server_manager.ServerManager)
+        manager.host = 'db.example.com'
+        manager.port = 5432
+        manager.use_ssh_tunnel = False
+        manager.service = None
+        manager.connection_params = {
+            'sslmode': 'require', self.bad_key: '/../bob/.pgpass'}
+
+        with self.app.test_request_context(), \
+                patch.dict(self.app.config, {'SERVER_MODE': True}), \
+                patch.object(self.app, 'PGADMIN_RUNTIME', False), \
+                patch('pgadmin.utils.get_storage_directory',
+                      side_effect=self._storage_dir):
+            with self.assertRaises(PermissionError):
+                manager.create_connection_string('postgres', 'alice')
+
+        # The injected key must not reach the connection string's display
+        # copy either.
+        self.assertNotIn(self.bad_key, manager.display_connection_string)
+
+
+class CompleteFilePathWriteTargetTestCase(_StorageMixin, BaseTestGenerator):
+    """A write target (sslkeylogfile) must additionally refuse a shared
+    storage directory that is read-only for non-administrators, which an
+    ordinary read path does not check."""
+
+    scenarios = [
+        ('A non-admin cannot write into a read-only shared directory',
+         dict(restricted=True, admin=False, denied=True)),
+        ('An administrator can write into a read-only shared directory',
+         dict(restricted=True, admin=True, denied=False)),
+        ('Anyone can write into an unrestricted shared directory',
+         dict(restricted=False, admin=False, denied=False)),
+    ]
+
+    def runTest(self):
+        self.shared_storage = [{'name': 'Team', 'path': self.shared,
+                                'restricted_access': self.restricted}]
+        user = _user(self.admin)
+
+        with self.app.test_request_context(), \
+                patch.dict(self.app.config, {'SERVER_MODE': True,
+                                             'SHARED_STORAGE':
+                                             self.shared_storage}), \
+                patch.object(self.app, 'PGADMIN_RUNTIME', False), \
+                patch.object(config, 'SHARED_STORAGE', self.shared_storage), \
+                patch('pgadmin.utils.get_storage_directory',
+                      side_effect=self._storage_dir):
+            flask.g._login_user = user
+            if self.denied:
+                with self.assertRaises(PermissionError):
+                    get_complete_file_path('Team:/keylog', validate=False,
+                                           for_write=True)
+            else:
+                result = get_complete_file_path('Team:/keylog', validate=False,
+                                                for_write=True)
+                self.assertEqual(os.path.normpath(result),
+                                 os.path.join(self.shared, 'keylog'))
+
+
+def _user(is_admin):
+    """A minimal current_user stand-in with a has_role() method."""
+    return SimpleNamespace(username='alice',
+                           has_role=lambda role: is_admin)

web/pgadmin/utils/tests/test_connection_param_keys.py

@@ -0,0 +1,83 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+from pgadmin.utils import is_valid_connection_param_key, \
+    get_invalid_connection_param_key
+from pgadmin.utils.route import BaseTestGenerator
+
+
+class TestConnectionParamKeyValidation(BaseTestGenerator):
+    """
+    make_conninfo() quotes connection-parameter values but not their keys,
+    so a key that smuggles its own "keyword=value" pairs (via an '=' sign
+    or embedded/leading/trailing whitespace) would reach libpq as extra,
+    unconfined connection directions. Only exact libpq keywords are allowed
+    as connection-parameter keys.
+    """
+
+    scenarios = [
+        # --- Accepted: exact libpq keywords ---
+        ('A plain keyword is accepted',
+         dict(connection_params={'sslmode': 'require'},
+              invalid_key=None)),
+        ('A file keyword is accepted',
+         dict(connection_params={'passfile': 'pgpass'},
+              invalid_key=None)),
+        ('Several exact keywords are accepted',
+         dict(connection_params={'host': 'db', 'port': '5432',
+                                 'sslmode': 'verify-full'},
+              invalid_key=None)),
+        ('No parameters at all',
+         dict(connection_params=None, invalid_key=None)),
+        ('An empty dict',
+         dict(connection_params={}, invalid_key=None)),
+        ('Something that is not a dict',
+         dict(connection_params=['sslmode=require'], invalid_key=None)),
+
+        # --- Refused: injected keys ---
+        ('A key carrying a second keyword via "=" is refused',
+         dict(connection_params={'sslmode=require passfile':
+                                 '/../bob/.pgpass'},
+              invalid_key='sslmode=require passfile')),
+        ('A key with a leading space is refused',
+         dict(connection_params={' passfile': '/../bob/.pgpass'},
+              invalid_key=' passfile')),
+        ('A key with a trailing space is refused',
+         dict(connection_params={'passfile ': '/../bob/.pgpass'},
+              invalid_key='passfile ')),
+        ('A key smuggling a keyword via whitespace and "=" is refused',
+         dict(connection_params={'keepalives=1 sslkeylogfile': '/tmp/x'},
+              invalid_key='keepalives=1 sslkeylogfile')),
+        ('A key reintroducing GSSAPI is refused',
+         dict(connection_params={'application_name=c require_auth': 'gss'},
+              invalid_key='application_name=c require_auth')),
+        ('A key that is not a libpq keyword at all is refused',
+         dict(connection_params={'not_a_real_keyword': 'x'},
+              invalid_key='not_a_real_keyword')),
+        ('A valid key alongside an injected key is still refused',
+         dict(connection_params={'sslmode': 'require',
+                                 'sslmode=require passfile': '/x'},
+              invalid_key='sslmode=require passfile')),
+    ]
+
+    def runTest(self):
+        invalid = get_invalid_connection_param_key(self.connection_params)
+        self.assertEqual(invalid, self.invalid_key)
+
+        # The injected key must also fail the single-key check, and a
+        # valid key must pass it.
+        if self.invalid_key is not None:
+            self.assertFalse(
+                is_valid_connection_param_key(self.invalid_key))
+        if isinstance(self.connection_params, dict):
+            for key in self.connection_params:
+                if key != self.invalid_key:
+                    self.assertTrue(
+                        is_valid_connection_param_key(key),
+                        'Expected {0!r} to be a valid libpq keyword'.format(
+                            key))