Stop libpq using the pgAdmin service account's Kerberos credentials in server mode. #10655

When a connection was not meant to use a Kerberos user's delegated
credentials, pgAdmin removed KRB5CCNAME from its environment, so the
Kerberos library fell back to the default credential cache of the account
pgAdmin runs as. libpq then answered any server that asked for GSSAPI
authentication with that account's ticket, whatever user name the
pgAdmin user gave, so any user could connect as the service account's
principal; gssencmode=disable does not prevent this. A client keytab for
that account had the same effect, even with no cache, as the library
fills the cache from it. On Windows, SSPI used the service account's
logon in the same way. psql was also started with a copy of pgAdmin's
environment, which could hold another user's credential cache whilst that
user was connecting, and the connection made to list a publisher's
publications was not made under the lock that guards it.

In server mode KRB5CCNAME and KRB5_CLIENT_KTNAME now always point at a
file that cannot exist, except whilst a user who logged in with Kerberos
connects to a server with Kerberos authentication enabled, when
KRB5CCNAME is that user's delegated cache as before. Because environment
isolation alone does not stop SSPI or gssproxy, every other connection
also gets require_auth=!gss,!sspi and gssencmode=disable, merged with the
server's own require_auth, which is refused if it allowed only GSS or
SSPI. A Kerberos-allowed connection forces gsslib=gssapi, so libpq uses
the delegated cache rather than the Windows logon. psql and the utilities
receive the same settings as PGREQUIREAUTH, PGGSSENCMODE and PGGSSLIB;
background processes now pass the server's own kerberos_conn instead of
always assuming a Kerberos connection, and the publication lookup
connects under the lock. Desktop mode is unchanged.

The forced require_auth, gssencmode and gsslib are serialised last, so
they win over any value a crafted connection parameter key places before
them, and the final connection string is parsed and refused if any of
them differs from the value forced (or is present where it is left to
the environment), so a smuggled value cannot steer the PSQL tool or a
utility to GSSAPI or SSPI however the parameters were serialised.

A pg_service.conf entry ranks above the PG* environment variables, so
only a service the administrator lists in SERVER_MODE_ALLOWED_SERVICES
can override these settings for the utilities; the documentation says
such an entry must not set them.

Compatibility: in server mode, connecting with the service account's
Kerberos ticket, client keytab or Windows logon (SSPI) no longer works;
only users who log in to pgAdmin with Kerberos can use GSSAPI, with their
own delegated credentials, as the documentation already described. A
server whose require_auth lists only gss or sspi is refused unless the
user logged in with Kerberos (SSPI-only is never usable). require_auth
needs libpq 16 or later; with an older libpq only the KRB5CCNAME and
KRB5_CLIENT_KTNAME isolation applies, which does not cover Windows SSPI
or gssproxy.

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

docs/en_US/kerberos.rst

@@ -126,6 +126,62 @@ PostgreSQL Server settings to configure Kerberos Authentication
 * Note that, you have to login into pgAdmin with Kerberos authentication to
   then connect to PostgreSQL using Kerberos.
 
+* In server mode, the only Kerberos credentials used to connect to a
+  PostgreSQL server are those delegated by a user who logged in to pgAdmin
+  with Kerberos, and only for a server with "Kerberos authentication"
+  enabled. The default credential cache and client keytab of the operating
+  system account that pgAdmin runs as are never used, so a user cannot
+  connect to a server with that account's Kerberos identity.
+
+* To enforce this for Windows SSPI, which uses the logon of the account
+  pgAdmin runs as, and for gssproxy, which environment variables do not
+  stop, in server mode pgAdmin also tells libpq what to use and what to
+  refuse.
+
+  * For a Kerberos-login user's connection to a server with Kerberos
+    authentication enabled, pgAdmin forces ``gsslib=gssapi``, replacing any
+    ``gsslib`` the user set. Without it, a libpq built with both GSSAPI and
+    SSPI answers the server's request with SSPI, which uses the Windows
+    logon of the account pgAdmin runs as and ignores the delegated
+    credentials, whereas ``gsslib=gssapi`` makes libpq use the delegated
+    credential cache. This helps only with a Windows libpq build that
+    includes both GSSAPI and SSPI, as pgAdmin's Windows packages do. With
+    libpq 16 or later, SSPI is also refused with ``require_auth=!sspi``;
+    with an older libpq it is not refused.
+
+  * Every other connection uses ``require_auth=!gss,!sspi`` and
+    ``gssencmode=disable``, merged with the server's own "Require
+    authentication" (``require_auth``) parameter. A server whose
+    ``require_auth`` allows only ``gss`` can be used only by a
+    Kerberos-login user connecting to a server with Kerberos authentication
+    enabled, and one that allows only ``sspi`` cannot be used in server
+    mode.
+
+  * The ``require_auth`` protection needs libpq 16 or later in pgAdmin.
+    Without it only the ``KRB5CCNAME`` and ``KRB5_CLIENT_KTNAME`` isolation
+    applies to connections that may not use Kerberos, which does not cover
+    Windows SSPI or gssproxy. ``gssencmode`` needs libpq 12 or later;
+    ``gsslib`` is known to every libpq that pgAdmin supports.
+
+  * psql and the utilities (such as pg_dump) get no ``require_auth`` in their
+    connection string unless the user set one for the server, which is then
+    merged as above. Their protection is the ``PGREQUIREAUTH``,
+    ``PGGSSENCMODE`` and ``PGGSSLIB`` environment variables. Only a libpq
+    16 or later in the utility honours ``PGREQUIREAUTH``; an older one
+    ignores it silently.
+
+  * A service file entry's settings take precedence over these environment
+    variables for psql and the utilities, whereas pgAdmin's own connections
+    are not affected, because their connection string takes precedence over
+    the service file. Only a service listed in
+    ``SERVER_MODE_ALLOWED_SERVICES`` can be used in server mode, so only an
+    entry the administrator has written can do this. Do not set
+    ``require_auth``, ``gssencmode`` or ``gsslib`` in such an entry unless
+    every pgAdmin user may use what it allows: the ``KRB5CCNAME`` and
+    ``KRB5_CLIENT_KTNAME`` isolation cannot be overridden by a service file,
+    but ``require_auth`` and ``gsslib`` in a listed service can re-enable
+    Windows SSPI or gssproxy for the utilities.
+
 
 Master Password
 ===============

docs/en_US/server_dialog.rst

@@ -141,6 +141,10 @@ Click on the *+* button to add a new parameter. Some of the parameters are:
   in seconds. Zero or not specified means wait indefinitely. It is not
   recommended to use a timeout of less than 2 seconds. By default it is set to
   10 seconds.
+* *Require authentication* (*require_auth*) field to specify the
+  authentication methods the server may use. In server mode, pgAdmin
+  refuses the GSSAPI and SSPI methods for a connection that may not use
+  Kerberos, as described in :ref:`kerberos`.
 * *SSL mode* field to select the type of SSL connection the server should use.
   For more information about using SSL encryption, see
   `Section 33.18 of the Postgres documentation <https://www.postgresql.org/docs/current/libpq-ssl.html>`_.

web/config.py

@@ -1142,7 +1142,9 @@ SERVER_PASSEXEC_ENV_PASSTHROUGH = []
 # user. Any user can connect using a listed service, including any
 # password, password file or client certificate its entry names, and can
 # override the entry's host to send that password to a server of their
-# own choosing.
+# own choosing. psql and the utilities also take the entry's require_auth,
+# gssencmode and gsslib in preference to the settings pgAdmin gives them
+# for Kerberos, so do not set those in a listed service.
 #############################################################################
 SERVER_MODE_ALLOWED_SERVICES = []
 

web/pgadmin/__init__.py

@@ -43,6 +43,8 @@ from pgadmin.model import db, Role, Server, SharedServer, ServerGroup, \
 from pgadmin.utils import PgAdminModule, driver, KeyManager, heartbeat, \
     remove_inherited_libpq_credentials
 from pgadmin.utils.db_utils import normalize_database_uri
+from pgadmin.utils.locker import isolate_kerberos_credentials, \
+    warn_if_no_auth_refusal
 from pgadmin.utils.preferences import Preferences
 from pgadmin.utils.session import create_session_interface, pga_unauthorised
 from pgadmin.utils.versioned_template_loader import VersionedTemplateLoader
@@ -333,6 +335,8 @@ def create_app(app_name=None):
     # pick up a password or service from the service account's environment.
     if config.SERVER_MODE is True:
         remove_inherited_libpq_credentials(app.logger)
+        isolate_kerberos_credentials()
+        warn_if_no_auth_refusal(app.logger)
 
     ##########################################################################
     # Setup i18n

web/pgadmin/misc/bgprocess/processes.py

@@ -26,13 +26,14 @@ import shutil
 from pgadmin.utils import u_encode, file_quote, fs_encoding, \
     get_complete_file_path, get_storage_directory, IS_WIN, \
     connection_params_to_env, utility_connection_params, check_service_allowed
-from pgadmin.utils.constants import (KERBEROS, UTILITIES_ARRAY,
+from pgadmin.utils.constants import (UTILITIES_ARRAY,
                                      BG_PROCESS_ERROR_MSGS)
-from pgadmin.utils.locker import ConnectionLocker
+from pgadmin.utils.locker import ConnectionLocker, kerberos_environ, \
+    kerberos_libpq_env
 from pgadmin.utils.preferences import Preferences
 
 from dateutil import parser
-from flask import current_app, session
+from flask import current_app
 from flask_babel import gettext as _
 from flask_security import current_user
 
@@ -140,6 +141,7 @@ class BatchProcess:
             self.stdout = self.stderr = self.stime = self.etime = \
             self.ecode = self.manager_obj = None
         self.env = dict()
+        self.kerberos_conn = False
 
         if 'id' in kwargs:
             self._retrieve_process(kwargs['id'])
@@ -333,11 +335,9 @@ class BatchProcess:
         env['PROCID'] = self.id
         env['OUTDIR'] = self.log_dir
         env['PGA_BGP_FOREGROUND'] = "1"
-        if config.SERVER_MODE and session and \
-                session.get('auth_source_manager', {}).get(
-                    'current_source') == KERBEROS and \
-                'KRB5CCNAME' in session:
-            env['KRB5CCNAME'] = session['KRB5CCNAME']
+        # A user who logged in with Kerberos has their own credentials;
+        # in server mode, no other user may have any.
+        env.update(kerberos_environ(self.kerberos_conn))
 
         if self.env:
             env.update(self.env)
@@ -911,6 +911,13 @@ class BatchProcess:
         if self.manager_obj and self.manager_obj.use_ssh_tunnel:
             conn_env.pop('PGHOSTADDR', None)
 
+        # Whether the server has Kerberos enabled decides if the utility may
+        # use Kerberos, and so which authentication libpq must refuse.
+        self.kerberos_conn = bool(getattr(
+            self.manager_obj if self.manager_obj else server,
+            'kerberos_conn', False))
+        kerberos_libpq_env(conn_env, self.kerberos_conn)
+
         self.env.update(conn_env)
 
         # Take the service name from the manager too: for a shared server

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

@@ -34,6 +34,7 @@ def _make_process():
     p.log_dir = '/nonexistent/log/dir'
     p.stime = p.etime = None
     p.env = dict()
+    p.kerberos_conn = False
     return p
 
 

web/pgadmin/misc/bgprocess/tests/test_set_env_variables_kerberos.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
+#
+##########################################################################
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from flask import session
+
+from pgadmin.misc.bgprocess.processes import BatchProcess
+from pgadmin.utils.constants import INTERNAL, KERBEROS
+from pgadmin.utils.locker import KerberosNotAllowed
+from pgadmin.utils.route import BaseTestGenerator
+
+
+class TestSetEnvVariablesKerberos(BaseTestGenerator):
+    """Check that in server mode a utility's environment makes libpq refuse
+    GSSAPI and SSPI where the connection may not use Kerberos, merged with
+    the server's own require_auth, and is left alone in desktop mode.
+    """
+
+    scenarios = [
+        ('An internal user has GSSAPI and SSPI refused',
+         dict(params={},
+              expected={'PGREQUIREAUTH': '!gss,!sspi',
+                        'PGGSSENCMODE': 'disable'},
+              absent=('PGGSSLIB',))),
+        ('The server\'s require_auth is merged',
+         dict(params={'require_auth': 'scram-sha-256,gss',
+                      'gssencmode': 'require'},
+              expected={'PGREQUIREAUTH': 'scram-sha-256',
+                        'PGGSSENCMODE': 'disable'})),
+        ('Negated methods of the server are kept',
+         dict(params={'require_auth': '!password'},
+              expected={'PGREQUIREAUTH': '!password,!gss,!sspi',
+                        'PGGSSENCMODE': 'disable'})),
+        ('A server accepting only GSSAPI is refused',
+         dict(params={'require_auth': 'gss'}, error=True)),
+        ('A Kerberos user connecting with Kerberos has only SSPI refused',
+         dict(source=KERBEROS, kerberos_conn=True, params={},
+              expected={'PGREQUIREAUTH': '!sspi', 'PGGSSLIB': 'gssapi'},
+              absent=('PGGSSENCMODE',))),
+        ('A Kerberos user\'s PGGSSLIB is forced to gssapi',
+         dict(source=KERBEROS, kerberos_conn=True,
+              params={'gsslib': 'sspi'},
+              expected={'PGGSSLIB': 'gssapi'})),
+        ('A Kerberos user may use a server accepting only GSSAPI',
+         dict(source=KERBEROS, kerberos_conn=True,
+              params={'require_auth': 'gss'},
+              expected={'PGREQUIREAUTH': 'gss', 'PGGSSLIB': 'gssapi'})),
+        ('A Kerberos user on a server without Kerberos is as any other',
+         dict(source=KERBEROS, kerberos_conn=False, params={},
+              expected={'PGREQUIREAUTH': '!gss,!sspi',
+                        'PGGSSENCMODE': 'disable'},
+              absent=('PGGSSLIB',))),
+        ('Desktop mode sets nothing',
+         dict(server_mode=False, params={},
+              absent=('PGREQUIREAUTH', 'PGGSSENCMODE', 'PGGSSLIB'))),
+        ('Desktop mode leaves the server\'s require_auth alone',
+         dict(server_mode=False, params={'require_auth': 'gss'},
+              expected={'PGREQUIREAUTH': 'gss'})),
+    ]
+
+    server_mode = True
+    source = INTERNAL
+    kerberos_conn = False
+    expected = {}
+    absent = ()
+
+    def runTest(self):
+        # Skip __init__, which would create a Process row.
+        p = BatchProcess.__new__(BatchProcess)
+        p.id = 'test'
+        p.env = dict()
+        p.manager_obj = MagicMock(
+            connection_params=self.params, use_ssh_tunnel=False,
+            service=None, kerberos_conn=self.kerberos_conn)
+        server = SimpleNamespace(service=None, connection_params={},
+                                 kerberos_conn=self.kerberos_conn)
+
+        config = {'SERVER_MODE': self.server_mode}
+
+        with self.app.test_request_context(), \
+                patch.dict(self.app.config, config), \
+                patch('pgadmin.utils.locker.config.SERVER_MODE',
+                      self.server_mode), \
+                patch('pgadmin.utils.locker.libpq_version',
+                      return_value=160000), \
+                patch.object(self.app, 'PGADMIN_RUNTIME', False):
+            session['auth_source_manager'] = {'current_source': self.source}
+            session['KRB5CCNAME'] = 'FILE:/tmp/pgadmin_cache_alice'
+            if getattr(self, 'error', False):
+                with self.assertRaises(KerberosNotAllowed):
+                    p.set_env_variables(server)
+                return
+            p.set_env_variables(server)
+
+        for key, value in self.expected.items():
+            self.assertEqual(p.env.get(key), value, key)
+        for key in self.absent:
+            self.assertNotIn(key, p.env)

web/pgadmin/tools/psql/__init__.py

@@ -28,6 +28,7 @@ from pgadmin.utils import PgAdminModule
 from pgadmin.utils.driver import get_driver
 from ... import socketio as sio
 from pgadmin.utils import get_complete_file_path
+from pgadmin.utils.locker import kerberos_environ, kerberos_libpq_env
 from pgadmin.authenticate import socket_login_required, \
     socket_permissions_required
 from pgadmin.model import Server
@@ -513,13 +514,20 @@ def get_connection_str(psql_utility, db, manager, env):
     :param psql_utility: PostgreSQL binary path.
     :param db: database name to connect specific db.
     :param env: a dict to put the environment psql needs in, which is
-        the password.
+        the password and the Kerberos settings.
     :return: connection attribute list for PSQL connection.
     """
     manager.export_password_env(env, 'PGPASSWORD')
+    # Set explicitly, as pgAdmin's own environment holds another user's
+    # Kerberos credential cache whilst that user is connecting.
+    env.update(kerberos_environ(manager.kerberos_conn))
+    # psql has its own libpq, so tell it through its environment to refuse
+    # GSSAPI and SSPI where it may not use them.
+    kerberos_libpq_env(env, manager.kerberos_conn)
     database = db if db != '' else 'postgres'
     user = underscore_unescape(manager.user) if manager.user else None
-    conn_attr = manager.create_connection_string(database, user)
+    conn_attr = manager.create_connection_string(
+        database, user, for_utility=True)
 
     conn_attr_list = list()
     conn_attr_list.append(psql_utility)

web/pgadmin/tools/psql/tests/test_password_env.py

@@ -37,6 +37,7 @@ class PSQLPasswordEnvTestCase(BaseTestGenerator):
         manager.password = b'encrypted'
         manager.passexec = None
         manager.user = 'alice'
+        manager.kerberos_conn = False
         manager.create_connection_string = MagicMock(
             return_value='host=127.0.0.1 dbname=postgres user=alice')
 

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

@@ -30,6 +30,8 @@ from pgadmin.utils.server_access import get_server, \
     get_user_server_query
 from pgadmin.utils.exception import ObjectGone
 from pgadmin.utils import isolate_client_credential_files
+from pgadmin.utils.locker import ConnectionLocker, \
+    REQUIRE_AUTH_MIN_LIBPQ, apply_kerberos_conn_params, libpq_version
 from pgadmin.utils.passexec import build_passexec
 from .keywords import scan_keyword
 from ..abstract import BaseDriver
@@ -418,8 +420,19 @@ class Driver(BaseDriver):
                 sslcompression=True if kwargs[
                     'sslcompression'] else False
             )
-            return psycopg.Connection.connect(
-                **isolate_client_credential_files(params))
+            # This connection never uses Kerberos, so libpq is told to
+            # refuse it.
+            apply_kerberos_conn_params(params, is_kerberos_conn=False)
+            params = isolate_client_credential_files(params)
+            if libpq_version() >= REQUIRE_AUTH_MIN_LIBPQ:
+                # GSSAPI and SSPI are refused before any credentials are
+                # used, so there is nothing for the lock to guard.
+                return psycopg.Connection.connect(**params)
+
+            # Under the lock, so that pgAdmin's environment does not hold
+            # another user's Kerberos credential cache meanwhile.
+            with ConnectionLocker():
+                return psycopg.Connection.connect(**params)
 
     def release_connection(self, sid, database=None, conn_id=None):
         """

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

@@ -24,6 +24,8 @@ from pgadmin.utils import get_complete_file_path, \
     check_connection_params_allowed
 from pgadmin.utils.crypto import decrypt
 from pgadmin.utils.master_password import process_masterpass_disabled
+from pgadmin.utils.locker import KerberosNotAllowed, \
+    apply_kerberos_conn_params, check_kerberos_conn_params_unchanged
 from .connection import Connection
 from pgadmin.model import Server, User
 from pgadmin.utils.exception import ConnectionLost, SSHTunnelConnectionLost,\
@@ -703,10 +705,15 @@ WHERE db.oid = {0}""".format(did))
 
         return value
 
-    def create_connection_string(self, database, user, password=None):
+    def create_connection_string(self, database, user, password=None,
+                                 for_utility=False):
         """
         This function is used to create connection string based on the
         parameters.
+
+        :param for_utility: the string is for a utility, such as psql, with
+            its own libpq, so parameters that only a newer libpq knows are
+            not added to it; they are in its environment instead.
         """
         dsn_args = dict()
         dsn_args['host'] = self.host
@@ -808,6 +815,18 @@ WHERE db.oid = {0}""".format(did))
             if k not in dsn_args:
                 dsn_args[k] = v
 
+        # In server mode, libpq must refuse GSSAPI and SSPI where this
+        # connection may not use Kerberos; the environment alone does not
+        # stop SSPI or gssproxy.
+        kerberos_error = None
+        for args in (dsn_args, display_dsn_args):
+            try:
+                apply_kerberos_conn_params(
+                    args, getattr(self, 'kerberos_conn', False),
+                    add_missing=not for_utility)
+            except KerberosNotAllowed as e:
+                kerberos_error = kerberos_error or e
+
         isolate_client_credential_files(dsn_args)
 
         # The server's service and a 'service' connection parameter are
@@ -837,6 +856,9 @@ WHERE db.oid = {0}""".format(did))
         if path_error is not None:
             raise path_error
 
+        if kerberos_error is not None:
+            raise kerberos_error
+
         conn_string = make_conninfo(**dsn_args)
 
         # make_conninfo() quotes parameter values but not their keys, so a
@@ -847,4 +869,12 @@ WHERE db.oid = {0}""".format(did))
         # allowlist holds regardless of how the parameters were serialised.
         check_connection_params_allowed(conninfo_to_dict(conn_string))
 
+        # A crafted parameter key (make_conninfo() does not escape keys) can
+        # serialise its own require_auth/gssencmode/gsslib into the string,
+        # which libpq, or psql whose conninfo outranks PGREQUIREAUTH, would
+        # honour. Re-check the parsed final string so the forced values (or
+        # their absence, for a utility that gets them from the environment)
+        # are what libpq actually acts on, however the parameters were built.
+        check_kerberos_conn_params_unchanged(conn_string, dsn_args)
+
         return conn_string

web/pgadmin/utils/driver/psycopg3/tests/test_kerberos_conninfo.py

@@ -0,0 +1,235 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""In server mode, libpq is told to refuse GSSAPI and SSPI, and GSSAPI
+encryption, for a connection that may not use Kerberos, as the environment
+variables that isolate Kerberos credentials do not stop Windows SSPI or
+gssproxy. Desktop mode is unchanged."""
+
+import os
+import shutil
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+import psycopg
+from psycopg.conninfo import conninfo_to_dict
+
+import config
+from flask import session
+from pgadmin.utils.constants import INTERNAL, KERBEROS
+from pgadmin.utils.locker import KerberosNotAllowed
+from pgadmin.utils.route import BaseTestGenerator
+
+import pgadmin.utils.driver.psycopg3.server_manager as server_manager
+from pgadmin.utils.driver.psycopg3.connection import Connection
+
+USER_CACHE = 'FILE:/tmp/pgadmin_cache_alice'
+
+# The libpq that parses a connection string is the one installed, which may
+# be older than 16 and then rejects require_auth. So most scenarios use a
+# stand-in for make_conninfo, which does not validate; only one uses the
+# real parser, and it needs libpq 16 or later.
+LIBPQ_HAS_REQUIRE_AUTH = psycopg.pq.version() >= 160000
+
+
+def _fake_make_conninfo(conninfo='', **kwargs):
+    """A deterministic stand-in for psycopg's make_conninfo: key=value
+    pairs of the keyword arguments that are not None."""
+    return ' '.join('{0}={1}'.format(k, v) for k, v in kwargs.items()
+                    if v is not None)
+
+
+def _parse_fake(conninfo):
+    return dict(item.split('=', 1) for item in conninfo.split(' ') if item)
+
+
+def _login(source, ccache=USER_CACHE):
+    session['auth_source_manager'] = {'current_source': source}
+    if ccache:
+        session['KRB5CCNAME'] = ccache
+
+
+class ConnectionStringKerberosTestCase(BaseTestGenerator):
+    """The connection string refuses GSSAPI and SSPI where Kerberos may
+    not be used, merged with the server's own require_auth."""
+
+    scenarios = [
+        ('An internal user is refused GSSAPI and SSPI',
+         dict(params={},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('Negated methods of the server are kept',
+         dict(params={'require_auth': '!password'},
+              expected={'require_auth': '!password,!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('Accepted methods of the server lose GSSAPI',
+         dict(params={'require_auth': 'scram-sha-256,gss'},
+              expected={'require_auth': 'scram-sha-256',
+                        'gssencmode': 'disable'})),
+        ('A server accepting only GSSAPI and SSPI is refused',
+         dict(params={'require_auth': 'gss,sspi'}, error=True)),
+        ('A Kerberos user connecting with Kerberos has only SSPI refused',
+         dict(source=KERBEROS, kerberos_conn=True, params={},
+              expected={'require_auth': '!sspi'},
+              absent=('gssencmode',))),
+        ('A Kerberos user on a server without Kerberos is as any other',
+         dict(source=KERBEROS, kerberos_conn=False, params={},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('A Kerberos user connecting with Kerberos is forced to GSSAPI',
+         dict(source=KERBEROS, kerberos_conn=True, libpq=150000, params={},
+              expected={'gsslib': 'gssapi'},
+              absent=('require_auth', 'gssencmode'))),
+        ('A Kerberos user\'s own gsslib is overridden with GSSAPI',
+         dict(source=KERBEROS, kerberos_conn=True, libpq=150000,
+              params={'gsslib': 'sspi'}, expected={'gsslib': 'gssapi'})),
+        ('A Kerberos user\'s utility string has gsslib overridden',
+         dict(source=KERBEROS, kerberos_conn=True, libpq=150000,
+              for_utility=True, params={'gsslib': 'sspi'},
+              expected={'gsslib': 'gssapi'})),
+        ('No gsslib is added where Kerberos is refused',
+         dict(libpq=150000, params={}, expected={'gssencmode': 'disable'},
+              absent=('gsslib',))),
+        ('Desktop mode changes nothing',
+         dict(server_mode=False, params={},
+              absent=('require_auth', 'gssencmode'))),
+        ('An old libpq has only GSSAPI encryption disabled',
+         dict(libpq=150000, params={}, expected={'gssencmode': 'disable'},
+              absent=('require_auth',))),
+        ('A utility string does not get require_auth',
+         dict(for_utility=True, params={},
+              absent=('require_auth', 'gssencmode'))),
+        ('A utility string has the server\'s require_auth changed',
+         dict(for_utility=True,
+              params={'require_auth': 'scram-sha-256,gss'},
+              expected={'require_auth': 'scram-sha-256'},
+              absent=('gssencmode',))),
+        ('The user\'s gssencmode does not apply',
+         dict(params={'gssencmode': 'require'},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('The user\'s gssencmode does not apply with an old libpq',
+         dict(libpq=150000, params={'gssencmode': 'require'},
+              expected={'gssencmode': 'disable'},
+              absent=('require_auth',))),
+        ('The real libpq parser accepts the connection string',
+         dict(real_parser=True, params={'require_auth': '!password'},
+              expected={'require_auth': '!password,!gss,!sspi',
+                        'gssencmode': 'disable'})),
+    ]
+
+    server_mode = True
+    source = INTERNAL
+    kerberos_conn = False
+    libpq = 160000
+    for_utility = False
+    real_parser = False
+    expected = {}
+    absent = ()
+
+    def setUp(self):
+        # Filesystem-only tests: skip BaseTestGenerator's server setup.
+        unittest.TestCase.setUp(self)
+        self.storage = os.path.realpath(
+            tempfile.mkdtemp(prefix='pga_kerberos_conninfo_'))
+
+    def tearDown(self):
+        shutil.rmtree(self.storage, ignore_errors=True)
+
+    def runTest(self):
+        if self.real_parser and not LIBPQ_HAS_REQUIRE_AUTH:
+            self.skipTest('The installed libpq cannot parse require_auth.')
+
+        manager = server_manager.ServerManager.__new__(
+            server_manager.ServerManager)
+        manager.host = '127.0.0.1'
+        manager.port = 5432
+        manager.use_ssh_tunnel = False
+        manager.service = None
+        manager.connection_params = dict(self.params)
+        manager.kerberos_conn = self.kerberos_conn
+
+        patches = [
+            patch.dict(self.app.config, {'SERVER_MODE': self.server_mode}),
+            patch.object(config, 'SERVER_MODE', self.server_mode),
+            patch.object(self.app, 'PGADMIN_RUNTIME', False),
+            patch('pgadmin.utils.get_storage_directory',
+                  return_value=self.storage),
+            patch('pgadmin.utils.locker.libpq_version',
+                  return_value=self.libpq),
+        ]
+        if not self.real_parser:
+            patches.append(patch.object(
+                server_manager, 'make_conninfo', _fake_make_conninfo))
+        with self.app.test_request_context():
+            for p in patches:
+                p.start()
+            try:
+                _login(self.source)
+                if getattr(self, 'error', False):
+                    with self.assertRaises(PermissionError) as cm:
+                        manager.create_connection_string(
+                            'postgres', 'alice',
+                            for_utility=self.for_utility)
+                    self.assertIn('require_auth', str(cm.exception))
+                    # The display string is still built.
+                    self.assertTrue(manager.display_connection_string)
+                    return
+                dsn = manager.create_connection_string(
+                    'postgres', 'alice', for_utility=self.for_utility)
+            finally:
+                for p in reversed(patches):
+                    p.stop()
+
+        for conninfo in (dsn, manager.display_connection_string):
+            values = conninfo_to_dict(conninfo) if self.real_parser \
+                else _parse_fake(conninfo)
+            for key, value in self.expected.items():
+                self.assertEqual(values.get(key), value, key)
+            for key in self.absent:
+                self.assertNotIn(key, values)
+
+
+class ConnectKerberosNotAllowedTestCase(BaseTestGenerator):
+    """A server whose require_auth allows only GSSAPI or SSPI, for a
+    connection that may not use them, is reported by Connection.connect()
+    as a failed connection with the message, not as an exception."""
+
+    scenarios = [('KerberosNotAllowed is a failed connection', dict())]
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        message = 'require_auth allows only GSSAPI or SSPI'
+        manager = MagicMock(use_ssh_tunnel=0, kerberos_conn=False,
+                            passexec=None, password=None)
+        manager.get_connection_param_value.return_value = None
+        manager.create_connection_string.side_effect = \
+            KerberosNotAllowed(message)
+
+        conn = Connection.__new__(Connection)
+        conn.conn = None
+        conn.manager = manager
+        conn.db = 'postgres'
+        conn.conn_id = 'test'
+        conn.password = None
+        conn.reconnecting = False
+        conn.async_ = 0
+
+        with patch('pgadmin.utils.driver.psycopg3.connection.'
+                   'get_crypt_key', return_value=(True, 'key')), \
+                patch.object(config, 'SERVER_MODE', True), \
+                self.app.test_request_context():
+            status, msg = conn.connect()
+
+        self.assertFalse(status)
+        self.assertEqual(msg, message)
+        manager.stop_ssh_tunnel.assert_called_once()

web/pgadmin/utils/locker.py

@@ -14,10 +14,287 @@ Kerberos Environment Locker class
 from threading import Lock
 from os import environ
 from flask import session, current_app
+from flask_babel import gettext
+from psycopg.conninfo import conninfo_to_dict
 
 import config
+from pgadmin.utils import NO_CLIENT_FILE
 from pgadmin.utils.constants import KERBEROS
 
+# In server mode, the Kerberos environment for a connection that must not
+# use any Kerberos credentials: a credential cache and a client keytab that
+# cannot exist or be created (see NO_CLIENT_FILE). Without KRB5CCNAME, the
+# Kerberos library uses the default credential cache of the account pgAdmin
+# runs as, and libpq answers a server's request for GSSAPI authentication
+# with that account's ticket, whatever user name the pgAdmin user gave;
+# gssencmode=disable does not prevent that. A cache that does not exist but
+# could be created is not enough either, as the library fills it from the
+# account's client keytab, if it has one, so that is disabled too.
+NO_KERBEROS_CREDENTIALS = {
+    'KRB5CCNAME': 'FILE:' + NO_CLIENT_FILE,
+    'KRB5_CLIENT_KTNAME': 'FILE:' + NO_CLIENT_FILE,
+}
+
+
+# The first libpq versions to know the require_auth and gssencmode
+# connection parameters.
+REQUIRE_AUTH_MIN_LIBPQ = 160000
+GSSENCMODE_MIN_LIBPQ = 120000
+
+
+class KerberosNotAllowed(PermissionError):
+    """The server's require_auth setting would allow only GSSAPI or SSPI
+    authentication, which this connection may not use."""
+
+
+def libpq_version():
+    """
+    Return the version of the libpq that pgAdmin's own connections use.
+    This is deliberately not cached.
+    """
+    # Imported here as modules of pgAdmin are imported where libpq is
+    # not installed.
+    import psycopg
+    return psycopg.pq.version()
+
+
+def kerberos_allowed(is_kerberos_conn):
+    """
+    Return True if, in server mode, the current user may use Kerberos for a
+    connection: they logged in to pgAdmin with Kerberos, and have delegated
+    credentials, and the connection is to a server with Kerberos
+    authentication enabled.
+    """
+    if not (config.SERVER_MODE and is_kerberos_conn):
+        return False
+    try:
+        source = session.get('auth_source_manager', {}).get(
+            'current_source')
+        return source == KERBEROS and 'KRB5CCNAME' in session
+    except RuntimeError:
+        # No request context: no user, so no credentials.
+        return False
+
+
+def kerberos_environ(is_kerberos_conn=False):
+    """
+    Return the Kerberos environment variables for a connection, or for a
+    process that connects, on behalf of the current user.
+
+    In server mode only a user who logged in to pgAdmin with Kerberos has
+    credentials, the ones delegated at login, and they are used only for a
+    connection that is to use Kerberos. Otherwise, and when there is no
+    request context, no credentials are available at all. Desktop mode
+    leaves the environment alone, so the user's own credentials are used.
+    """
+    if not config.SERVER_MODE:
+        return {}
+
+    env = dict(NO_KERBEROS_CREDENTIALS)
+    if kerberos_allowed(is_kerberos_conn):
+        env['KRB5CCNAME'] = session['KRB5CCNAME']
+    return env
+
+
+def refused_auth_methods(is_kerberos_conn):
+    """
+    Return the authentication methods libpq must refuse for a connection.
+
+    The environment variables above do not stop libpq on Windows using
+    SSPI, with the logon of the account pgAdmin runs as, or a gssproxy
+    service supplying credentials, so in server mode libpq is also told to
+    refuse them. SSPI is never wanted, as it cannot use a user's delegated
+    credentials.
+    """
+    if not config.SERVER_MODE:
+        return ()
+    if kerberos_allowed(is_kerberos_conn):
+        return ('sspi',)
+    return ('gss', 'sspi')
+
+
+def merge_require_auth(value, refused):
+    """
+    Return the require_auth value to use in place of the server's own,
+    so that the authentication methods in refused are not accepted.
+
+    A value that lists methods to refuse, each starting with '!', has
+    those added to. A value that lists methods to accept has the refused
+    ones removed from it, and the connection is not allowed if none are
+    left. A value mixing the two is returned unchanged, as libpq rejects
+    it.
+    """
+    if not refused:
+        return value
+
+    tokens = [t.strip() for t in str(value or '').split(',')]
+    tokens = [t for t in tokens if t]
+
+    if not tokens:
+        return ','.join('!' + m for m in refused)
+
+    negated = [t.startswith('!') for t in tokens]
+    if all(negated):
+        present = {t[1:].strip().lower() for t in tokens}
+        return ','.join(
+            tokens + ['!' + m for m in refused if m not in present])
+
+    if any(negated):
+        return value
+
+    kept = [t for t in tokens if t.lower() not in refused]
+    if not kept:
+        raise KerberosNotAllowed(gettext(
+            'The "Require authentication" (require_auth) parameter of the '
+            'server allows only GSSAPI or SSPI authentication. In server '
+            'mode, a server that allows only GSSAPI can be used only by a '
+            'user who logged in with Kerberos, connecting to a server with '
+            'Kerberos authentication enabled, and a server that allows '
+            'only SSPI cannot be used. Change require_auth in the '
+            'properties of the server.'))
+    return ','.join(kept)
+
+
+def apply_kerberos_conn_params(params, is_kerberos_conn, add_missing=True,
+                               names=None):
+    """
+    In server mode, update the libpq connection parameters in params, in
+    place, so that libpq refuses GSSAPI and SSPI authentication and GSSAPI
+    encryption for a connection that may not use Kerberos, and SSPI always,
+    and so that a connection that may use Kerberos does it with GSSAPI.
+
+    :param params: a dict of connection parameters, or of environment
+        variables if names is given.
+    :param is_kerberos_conn: whether the server has Kerberos enabled.
+    :param add_missing: add a parameter that is not in params only if
+        pgAdmin's own libpq knows it, as a libpq that does not fails to
+        parse a connection string containing it. It does not apply to the
+        environment, as a utility with an older libpq ignores the variable
+        and one with a newer libpq does not depend on pgAdmin's.
+    :param names: maps 'require_auth', 'gssencmode' and 'gsslib' to the
+        keys to use instead, for environment variables.
+    """
+    if not config.SERVER_MODE:
+        return
+
+    names = names or {}
+    in_env = bool(names)
+    require_auth = names.get('require_auth', 'require_auth')
+    gssencmode = names.get('gssencmode', 'gssencmode')
+    gsslib = names.get('gsslib', 'gsslib')
+    ver = libpq_version() if add_missing and not in_env else None
+    allowed = kerberos_allowed(is_kerberos_conn)
+
+    # Each forced parameter is popped before it is set, so it is re-inserted
+    # at the end of params. make_conninfo() does not escape keys, so a
+    # crafted key such as 'application_name=c require_auth' serialises its
+    # own require_auth=; libpq takes the last of a duplicated keyword, so
+    # the forced value must come last to win over any such smuggled value,
+    # whatever position the user's own require_auth/gssencmode/gsslib held.
+    #
+    # A parameter the user left empty is as good as unset. It is dropped
+    # where it is not replaced, as an empty value is invalid for a libpq
+    # that does not know the parameter.
+    #
+    # GSSAPI encryption is left as the user set it only where Kerberos is
+    # allowed.
+    if not allowed and (
+            in_env or params.get(gssencmode) or
+            (ver is not None and ver >= GSSENCMODE_MIN_LIBPQ)):
+        params.pop(gssencmode, None)
+        params[gssencmode] = 'disable'
+    elif not params.get(gssencmode):
+        params.pop(gssencmode, None)
+
+    # A libpq built with both GSSAPI and SSPI answers a server's GSS
+    # request with SSPI, the Windows logon of the account pgAdmin runs
+    # as, which ignores KRB5CCNAME and which require_auth does not stop.
+    # So where Kerberos is allowed, GSSAPI is forced, whatever the user set.
+    # Every libpq that pgAdmin supports knows gsslib. Where Kerberos is not
+    # allowed GSS is refused, so it is not needed.
+    if allowed and (in_env or gsslib in params or ver is not None):
+        params.pop(gsslib, None)
+        params[gsslib] = 'gssapi'
+
+    refused = refused_auth_methods(is_kerberos_conn)
+    if in_env or params.get(require_auth) or \
+            (ver is not None and ver >= REQUIRE_AUTH_MIN_LIBPQ):
+        merged = merge_require_auth(params.get(require_auth), refused)
+        params.pop(require_auth, None)
+        params[require_auth] = merged
+    else:
+        params.pop(require_auth, None)
+
+
+# The connection parameters that decide whether libpq attempts Kerberos.
+KERBEROS_CONN_PARAMS = ('require_auth', 'gssencmode', 'gsslib')
+
+
+def check_kerberos_conn_params_unchanged(conn_string, expected):
+    """
+    Raise KerberosNotAllowed if the final connection string, as libpq will
+    parse it, carries a require_auth, gssencmode or gsslib that differs from
+    the value apply_kerberos_conn_params() forced into expected (or from its
+    absence).
+
+    make_conninfo() does not escape parameter keys, so a crafted key such as
+    'application_name=c require_auth' serialises its own value for one of
+    these, which libpq, or the PSQL tool whose conninfo outranks
+    PGREQUIREAUTH/PGGSSENCMODE, would honour. Checking the parsed string
+    makes the guarantee independent of how the parameters were serialised.
+    """
+    if not config.SERVER_MODE:
+        return
+    parsed = conninfo_to_dict(conn_string)
+    for key in KERBEROS_CONN_PARAMS:
+        if parsed.get(key) != expected.get(key):
+            raise KerberosNotAllowed(gettext(
+                'The connection parameters contain a {0} that is not '
+                'allowed in server mode.').format(key))
+
+
+def kerberos_libpq_env(env, is_kerberos_conn):
+    """
+    Add the libpq environment variables that stop a utility using
+    Kerberos where it may not, to env, and return it. Only a utility with
+    a libpq that knows them is affected.
+    """
+    apply_kerberos_conn_params(
+        env, is_kerberos_conn,
+        names={'require_auth': 'PGREQUIREAUTH',
+               'gssencmode': 'PGGSSENCMODE',
+               'gsslib': 'PGGSSLIB'})
+    return env
+
+
+def isolate_kerberos_credentials():
+    """
+    In server mode, stop libpq in pgAdmin's own process, and the processes
+    it starts, from using the Kerberos credentials of the account pgAdmin
+    runs as, except where kerberos_environ() provides a user's own.
+    """
+    if config.SERVER_MODE:
+        environ.update(NO_KERBEROS_CREDENTIALS)
+
+
+def warn_if_no_auth_refusal(logger):
+    """
+    In server mode, warn if pgAdmin's libpq cannot refuse SSPI and
+    gssproxy, which it needs require_auth for. Never raises.
+    """
+    if not config.SERVER_MODE:
+        return
+    try:
+        if libpq_version() < REQUIRE_AUTH_MIN_LIBPQ:
+            logger.warning(
+                'The libpq used by pgAdmin is older than version 16, so '
+                'it cannot refuse SSPI (Windows) or gssproxy '
+                'authentication for connections that may not use '
+                'Kerberos. Only the environment-based isolation of '
+                'Kerberos credentials applies; use libpq 16 or later.')
+    except Exception as e:
+        logger.debug('Unable to check the libpq version: %s', e)
+
 
 class ConnectionLocker:
     """Implementing lock while setting/unsetting
@@ -33,27 +310,16 @@ class ConnectionLocker:
             self.lock.acquire()
             current_app.logger.info("Acquired a lock.")
 
-            try:
-                # session access requires a request context. Test code that
-                # calls start() under app_context (no request) would raise
-                # RuntimeError here, leaking the lock since __exit__ is not
-                # called when __enter__ raises.
-                source = session.get('auth_source_manager', {}).get(
-                    'current_source')
-                if source == KERBEROS and 'KRB5CCNAME' in session and \
-                        self.is_kerberos_conn:
-                    environ['KRB5CCNAME'] = session['KRB5CCNAME']
-                else:
-                    environ.pop('KRB5CCNAME', None)
-            except RuntimeError:
-                # No request context — nothing Kerberos-related to set up.
-                environ.pop('KRB5CCNAME', None)
+            # kerberos_environ() handles the lack of a request context
+            # itself; an exception here would leak the lock, since
+            # __exit__ is not called when __enter__ raises.
+            environ.update(kerberos_environ(self.is_kerberos_conn))
 
         return self
 
     def __exit__(self, type, value, traceback):
         if config.SERVER_MODE:
-            environ.pop('KRB5CCNAME', None)
+            environ.update(NO_KERBEROS_CREDENTIALS)
             if self.lock.locked():
                 current_app.logger.info("Released a lock.")
                 self.lock.release()

web/pgadmin/utils/tests/test_kerberos_isolation.py

@@ -0,0 +1,663 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""In server mode, libpq must never use the Kerberos credentials of the
+account pgAdmin runs as: its default credential cache or client keytab.
+Only a user who logged in to pgAdmin with Kerberos has credentials, the
+ones delegated at login. Desktop mode is unchanged."""
+
+import os
+import unittest
+from contextlib import nullcontext
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from flask import session
+from psycopg.conninfo import make_conninfo, conninfo_to_dict
+
+import config
+from pgadmin.utils.constants import INTERNAL, KERBEROS
+from pgadmin.utils.locker import NO_KERBEROS_CREDENTIALS, \
+    ConnectionLocker, KerberosNotAllowed, apply_kerberos_conn_params, \
+    isolate_kerberos_credentials, kerberos_environ, kerberos_libpq_env, \
+    merge_require_auth
+from pgadmin.utils.route import BaseTestGenerator
+
+USER_CACHE = 'FILE:/tmp/pgadmin_cache_alice'
+SERVICE_ACCOUNT = {'KRB5CCNAME': 'FILE:/tmp/krb5cc_service_account'}
+
+
+def _login(source, ccache=USER_CACHE):
+    session['auth_source_manager'] = {'current_source': source}
+    if ccache:
+        session['KRB5CCNAME'] = ccache
+
+
+class KerberosEnvironTestCase(BaseTestGenerator):
+    """kerberos_environ() gives a user's own delegated credentials to a
+    Kerberos connection, and in server mode nothing else."""
+
+    scenarios = [
+        ('Desktop mode leaves the environment alone',
+         dict(server_mode=False, source=KERBEROS, kerberos_conn=True,
+              expected={})),
+        ('An internal user has no Kerberos credentials',
+         dict(source=INTERNAL, kerberos_conn=True,
+              expected=NO_KERBEROS_CREDENTIALS)),
+        ('A Kerberos user gets their own cache for a Kerberos connection',
+         dict(source=KERBEROS, kerberos_conn=True,
+              expected=dict(NO_KERBEROS_CREDENTIALS,
+                            KRB5CCNAME=USER_CACHE))),
+        ('A Kerberos user has none for a connection without Kerberos',
+         dict(source=KERBEROS, kerberos_conn=False,
+              expected=NO_KERBEROS_CREDENTIALS)),
+        ('A Kerberos user without a delegated cache has none',
+         dict(source=KERBEROS, kerberos_conn=True, ccache=None,
+              expected=NO_KERBEROS_CREDENTIALS)),
+        ('Without a request context there are none',
+         dict(source=None, kerberos_conn=True,
+              expected=NO_KERBEROS_CREDENTIALS)),
+    ]
+
+    server_mode = True
+    ccache = USER_CACHE
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        with patch.object(config, 'SERVER_MODE', self.server_mode):
+            if self.source is None:
+                env = kerberos_environ(self.kerberos_conn)
+            else:
+                with self.app.test_request_context():
+                    _login(self.source, self.ccache)
+                    env = kerberos_environ(self.kerberos_conn)
+        self.assertEqual(env, self.expected)
+
+
+class ConnectionLockerTestCase(BaseTestGenerator):
+    """Whilst a connection is made, pgAdmin's environment holds only the
+    connecting user's own Kerberos credentials, if any, and afterwards
+    none; never the service account's."""
+
+    scenarios = [
+        ('An internal user connects without Kerberos credentials',
+         dict(source=INTERNAL, kerberos_conn=True,
+              during=NO_KERBEROS_CREDENTIALS)),
+        ('A Kerberos user connects with their own cache',
+         dict(source=KERBEROS, kerberos_conn=True,
+              during=dict(NO_KERBEROS_CREDENTIALS, KRB5CCNAME=USER_CACHE))),
+        ('A Kerberos user connects without Kerberos to another server',
+         dict(source=KERBEROS, kerberos_conn=False,
+              during=NO_KERBEROS_CREDENTIALS)),
+        ('Desktop mode keeps the service account\'s cache',
+         dict(server_mode=False, source=INTERNAL, kerberos_conn=True,
+              during=SERVICE_ACCOUNT, after=SERVICE_ACCOUNT)),
+    ]
+
+    server_mode = True
+    after = NO_KERBEROS_CREDENTIALS
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    @staticmethod
+    def _kerberos_vars():
+        return {k: os.environ[k] for k in NO_KERBEROS_CREDENTIALS
+                if k in os.environ}
+
+    def runTest(self):
+        with patch.object(config, 'SERVER_MODE', self.server_mode), \
+                patch.dict(os.environ, SERVICE_ACCOUNT), \
+                self.app.test_request_context():
+            os.environ.pop('KRB5_CLIENT_KTNAME', None)
+            _login(self.source)
+            with ConnectionLocker(self.kerberos_conn):
+                during = self._kerberos_vars()
+            after = self._kerberos_vars()
+
+        self.assertEqual(during, self.during)
+        self.assertEqual(after, self.after)
+        self.assertFalse(ConnectionLocker.lock.locked())
+
+
+class IsolateKerberosCredentialsTestCase(BaseTestGenerator):
+    """At startup in server mode pgAdmin's environment, which libpq in its
+    own process and every process it starts inherit, is pointed away from
+    the service account's credentials."""
+
+    scenarios = [
+        ('Server mode', dict(server_mode=True,
+                             expected=NO_KERBEROS_CREDENTIALS)),
+        ('Desktop mode', dict(server_mode=False, expected=SERVICE_ACCOUNT)),
+    ]
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        with patch.object(config, 'SERVER_MODE', self.server_mode), \
+                patch.dict(os.environ, SERVICE_ACCOUNT):
+            os.environ.pop('KRB5_CLIENT_KTNAME', None)
+            isolate_kerberos_credentials()
+            env = {k: os.environ[k] for k in NO_KERBEROS_CREDENTIALS
+                   if k in os.environ}
+        self.assertEqual(env, self.expected)
+
+
+class _Captured(Exception):
+    pass
+
+
+class BatchProcessKerberosTestCase(BaseTestGenerator):
+    """A utility started by a user gets that user's Kerberos credentials,
+    if they logged in with Kerberos and it connects to a server with
+    Kerberos authentication enabled, and in server mode no others."""
+
+    scenarios = [
+        ('An internal user\'s utility has no Kerberos credentials',
+         dict(source=INTERNAL, kerberos_conn=True,
+              expected=NO_KERBEROS_CREDENTIALS)),
+        ('A Kerberos user\'s utility has their own cache',
+         dict(source=KERBEROS, kerberos_conn=True,
+              expected=dict(NO_KERBEROS_CREDENTIALS,
+                            KRB5CCNAME=USER_CACHE))),
+        ('A Kerberos user\'s utility has none for a server without '
+         'Kerberos',
+         dict(source=KERBEROS, kerberos_conn=False,
+              expected=NO_KERBEROS_CREDENTIALS)),
+        ('Desktop mode keeps the service account\'s cache',
+         dict(server_mode=False, source=INTERNAL, kerberos_conn=False,
+              expected=SERVICE_ACCOUNT)),
+    ]
+
+    server_mode = True
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        from pgadmin.misc.bgprocess.processes import BatchProcess
+
+        process = BatchProcess.__new__(BatchProcess)
+        process.id = 'kerberos_test'
+        process.log_dir = '/nonexistent'
+        process.cmd = '/bin/true'
+        process.args = []
+        process.env = {'PGPASSWORD': 'x'}
+        process.manager_obj = None
+        process.check_start_end_time = MagicMock()
+        captured = {}
+
+        def capture(env):
+            captured.update(env)
+            raise _Captured()
+
+        server = SimpleNamespace(service=None, connection_params={},
+                                 kerberos_conn=self.kerberos_conn)
+
+        with patch.object(config, 'SERVER_MODE', self.server_mode), \
+                patch.dict(os.environ, SERVICE_ACCOUNT), \
+                self.app.test_request_context():
+            os.environ.pop('KRB5_CLIENT_KTNAME', None)
+            _login(self.source)
+            process.set_env_variables(server)
+            with self.assertRaises(_Captured):
+                process.start(cb=capture)
+
+        self.assertEqual(
+            {k: captured[k] for k in NO_KERBEROS_CREDENTIALS
+             if k in captured}, self.expected)
+
+
+class PSQLKerberosTestCase(BaseTestGenerator):
+    """psql gets the user's own Kerberos credentials for a Kerberos
+    connection, and in server mode none otherwise, whatever pgAdmin's
+    environment holds whilst another user connects."""
+
+    scenarios = [
+        ('An internal user\'s psql has no Kerberos credentials',
+         dict(source=INTERNAL, kerberos_conn=True,
+              expected=NO_KERBEROS_CREDENTIALS,
+              gssencmode='disable', require_auth='!gss,!sspi')),
+        ('A Kerberos user\'s psql has their own cache',
+         dict(source=KERBEROS, kerberos_conn=True,
+              expected=dict(NO_KERBEROS_CREDENTIALS,
+                            KRB5CCNAME=USER_CACHE),
+              gssencmode=None, require_auth='!sspi')),
+        ('A Kerberos user\'s psql has none for a server without Kerberos',
+         dict(source=KERBEROS, kerberos_conn=False,
+              expected=NO_KERBEROS_CREDENTIALS,
+              gssencmode='disable', require_auth='!gss,!sspi')),
+    ]
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        from pgadmin.tools import psql
+        import pgadmin.utils.driver.psycopg3.server_manager as server_manager
+
+        manager = server_manager.ServerManager.__new__(
+            server_manager.ServerManager)
+        manager.password = None
+        manager.passexec = None
+        manager.user = 'alice'
+        manager.kerberos_conn = self.kerberos_conn
+        manager.create_connection_string = MagicMock(
+            return_value='host=127.0.0.1 dbname=postgres user=alice')
+
+        env = {}
+        with patch.object(config, 'SERVER_MODE', True), \
+                patch.dict(os.environ, {'KRB5CCNAME': 'FILE:/tmp/other'}), \
+                self.app.test_request_context():
+            _login(self.source)
+            psql.get_connection_str('/usr/bin/psql', 'postgres', manager,
+                                    env)
+            # As get_user_env() builds it, without the per-user psqlrc.
+            child_env = dict(os.environ, **env)
+
+        self.assertEqual(
+            {k: child_env[k] for k in NO_KERBEROS_CREDENTIALS
+             if k in child_env}, self.expected)
+        # psql runs on its own libpq, so is told what to refuse.
+        manager.create_connection_string.assert_called_once_with(
+            'postgres', 'alice', for_utility=True)
+        self.assertEqual(env.get('PGGSSENCMODE'), self.gssencmode)
+        self.assertEqual(env['PGREQUIREAUTH'], self.require_auth)
+
+
+class SubscriptionConnectionKerberosTestCase(BaseTestGenerator):
+    """The direct connection made to list a publisher's publications, for
+    a subscription, may not use Kerberos in server mode. With libpq 16 or
+    later it is told to refuse GSSAPI and SSPI, so needs no lock; with an
+    older libpq it is made under the lock, so that pgAdmin's environment
+    holds no other user's Kerberos credential cache whilst it is made."""
+
+    scenarios = [
+        ('libpq 16 refuses Kerberos itself',
+         dict(libpq=160000, locked=False,
+              expected={'gssencmode': 'disable',
+                        'require_auth': '!gss,!sspi'})),
+        ('libpq 15 has the lock',
+         dict(libpq=150000, locked=True,
+              expected={'gssencmode': 'disable'})),
+        ('libpq 11 has the lock and no gssencmode',
+         dict(libpq=110000, locked=True, expected={})),
+    ]
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        from pgadmin.utils.driver import get_driver
+        import pgadmin.utils.driver.psycopg3 as psycopg3_driver
+
+        captured = {}
+        locked = []
+
+        def connect(**kwargs):
+            captured.update({k: os.environ.get(k)
+                             for k in NO_KERBEROS_CREDENTIALS})
+            captured['params'] = kwargs
+            locked.append(ConnectionLocker.lock.locked())
+            return 'connection'
+
+        with patch.object(config, 'SERVER_MODE', True), \
+                patch.dict(os.environ, {'KRB5CCNAME': USER_CACHE}), \
+                patch('pgadmin.utils.locker.libpq_version',
+                      return_value=self.libpq), \
+                patch.object(psycopg3_driver, 'libpq_version',
+                             return_value=self.libpq), \
+                patch.object(psycopg3_driver.psycopg.Connection, 'connect',
+                             side_effect=connect), \
+                self.app.test_request_context():
+            _login(INTERNAL)
+            conn = get_driver(config.PG_DEFAULT_DRIVER).get_connection(
+                sid=0, host='127.0.0.1', database='postgres', user='alice',
+                port=5432, passfile=None, sslmode='prefer', sslcert=None,
+                sslkey=None, sslrootcert=None, sslcompression=False)
+
+        self.assertEqual(conn, 'connection')
+        self.assertEqual(locked, [self.locked])
+        params = captured.pop('params')
+        self.assertEqual(
+            {k: params[k] for k in ('gssencmode', 'require_auth')
+             if k in params}, self.expected)
+        if self.locked:
+            self.assertEqual(captured, NO_KERBEROS_CREDENTIALS)
+
+
+class MergeRequireAuthTestCase(BaseTestGenerator):
+    """The server's own require_auth is combined with the methods that
+    may not be used, or the connection is refused if nothing else would
+    be left."""
+
+    scenarios = [
+        ('Nothing refused leaves the value alone',
+         dict(value='gss', refused=(), expected='gss')),
+        ('No value refuses the methods',
+         dict(value=None, refused=('gss', 'sspi'),
+              expected='!gss,!sspi')),
+        ('An empty value refuses the methods',
+         dict(value='', refused=('gss', 'sspi'), expected='!gss,!sspi')),
+        ('Only commas and spaces refuses the methods',
+         dict(value=' , ', refused=('gss', 'sspi'),
+              expected='!gss,!sspi')),
+        ('Negated methods are added to',
+         dict(value='!password', refused=('gss', 'sspi'),
+              expected='!password,!gss,!sspi')),
+        ('A negated method already there is not repeated',
+         dict(value='!sspi,!md5', refused=('gss', 'sspi'),
+              expected='!sspi,!md5,!gss')),
+        ('A negated method is matched whatever its case',
+         dict(value='!GSS', refused=('gss', 'sspi'),
+              expected='!GSS,!sspi')),
+        ('Accepted methods lose those refused',
+         dict(value='scram-sha-256,gss', refused=('gss', 'sspi'),
+              expected='scram-sha-256')),
+        ('Accepted methods are matched whatever their case',
+         dict(value='SCRAM-SHA-256,GSS,Sspi', refused=('gss', 'sspi'),
+              expected='SCRAM-SHA-256')),
+        ('Whitespace is removed',
+         dict(value=' password , gss ,, md5 ', refused=('gss', 'sspi'),
+              expected='password,md5')),
+        ('Whitespace in negated methods is dealt with',
+         dict(value=' !password , ', refused=('gss', 'sspi'),
+              expected='!password,!gss,!sspi')),
+        ('Accepting only the methods refused is an error',
+         dict(value='gss,sspi', refused=('gss', 'sspi'), error=True)),
+        ('Accepting only GSS is an error',
+         dict(value='gss', refused=('gss', 'sspi'), error=True)),
+        ('Accepting GSS is fine where only SSPI is refused',
+         dict(value='gss', refused=('sspi',), expected='gss')),
+        ('Accepting only SSPI is an error where it is refused',
+         dict(value='sspi', refused=('sspi',), error=True)),
+        ('A mix of negated and accepted methods is left for libpq',
+         dict(value='password,!gss', refused=('gss', 'sspi'),
+              expected='password,!gss')),
+    ]
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        if getattr(self, 'error', False):
+            with self.assertRaises(KerberosNotAllowed) as cm:
+                merge_require_auth(self.value, self.refused)
+            self.assertIn('require_auth', str(cm.exception))
+            self.assertIsInstance(cm.exception, PermissionError)
+            return
+        self.assertEqual(
+            merge_require_auth(self.value, self.refused), self.expected)
+
+
+class ApplyKerberosConnParamsTestCase(BaseTestGenerator):
+    """The libpq parameters, or environment variables, that make libpq
+    refuse Kerberos for a connection that may not use it."""
+
+    scenarios = [
+        ('Desktop mode changes nothing',
+         dict(server_mode=False, params={'require_auth': 'gss'},
+              expected={'require_auth': 'gss'})),
+        ('An internal user gets both on libpq 16',
+         dict(params={},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('An old libpq gets only gssencmode',
+         dict(libpq=150000, params={}, expected={'gssencmode': 'disable'})),
+        ('A very old libpq gets nothing',
+         dict(libpq=110000, params={}, expected={})),
+        ('An old libpq has the user\'s require_auth merged',
+         dict(libpq=150000, params={'require_auth': 'scram-sha-256,gss'},
+              expected={'require_auth': 'scram-sha-256',
+                        'gssencmode': 'disable'})),
+        ('A utility connection has nothing added',
+         dict(add_missing=False, params={}, expected={})),
+        ('A utility connection has a user\'s values changed',
+         dict(add_missing=False,
+              params={'require_auth': 'scram-sha-256,gss',
+                      'gssencmode': 'prefer'},
+              expected={'require_auth': 'scram-sha-256',
+                        'gssencmode': 'disable'})),
+        ('The user\'s gssencmode is overridden',
+         dict(params={'gssencmode': 'require'},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('A Kerberos user connecting with Kerberos only has SSPI refused',
+         dict(source=KERBEROS, kerberos_conn=True,
+              params={'gssencmode': 'require'},
+              expected={'require_auth': '!sspi',
+                        'gssencmode': 'require',
+                        'gsslib': 'gssapi'})),
+        ('A Kerberos user with a server accepting only GSS is allowed',
+         dict(source=KERBEROS, kerberos_conn=True,
+              params={'require_auth': 'gss'},
+              expected={'require_auth': 'gss', 'gsslib': 'gssapi'})),
+        ('A Kerberos user\'s gsslib is overridden with gssapi',
+         dict(source=KERBEROS, kerberos_conn=True,
+              params={'gsslib': 'sspi'},
+              expected={'require_auth': '!sspi', 'gsslib': 'gssapi'})),
+        ('A utility connection of a Kerberos user has gsslib changed',
+         dict(source=KERBEROS, kerberos_conn=True, add_missing=False,
+              params={'gsslib': 'sspi'},
+              expected={'gsslib': 'gssapi'})),
+        ('A utility connection of a Kerberos user has no gsslib added',
+         dict(source=KERBEROS, kerberos_conn=True, add_missing=False,
+              params={}, expected={})),
+        ('No gsslib is added where Kerberos is not allowed',
+         dict(source=KERBEROS, kerberos_conn=False, params={},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('The user\'s gsslib is left alone where Kerberos is refused',
+         dict(params={'gsslib': 'sspi'},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable', 'gsslib': 'sspi'})),
+        ('An empty require_auth and gssencmode count as unset',
+         dict(params={'require_auth': '', 'gssencmode': None},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('An empty require_auth is replaced, not merged, for Kerberos',
+         dict(source=KERBEROS, kerberos_conn=True,
+              params={'require_auth': None, 'gssencmode': ''},
+              expected={'require_auth': '!sspi', 'gsslib': 'gssapi'})),
+        ('Empty keys are dropped for an old libpq, not passed on',
+         dict(libpq=110000, params={'require_auth': '', 'gssencmode': ''},
+              expected={})),
+        ('Empty keys are dropped for a utility string',
+         dict(add_missing=False,
+              params={'require_auth': None, 'gssencmode': ''},
+              expected={})),
+        ('A Kerberos user with a server accepting only SSPI is refused',
+         dict(source=KERBEROS, kerberos_conn=True,
+              params={'require_auth': 'sspi'}, error=True)),
+        ('A Kerberos user on a server without Kerberos is refused GSS',
+         dict(source=KERBEROS, kerberos_conn=False, params={},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('A Kerberos user without a delegated cache is refused GSS',
+         dict(source=KERBEROS, kerberos_conn=True, ccache=None, params={},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('An internal user is refused GSS and an only-GSS server',
+         dict(params={'require_auth': 'gss'}, error=True)),
+        ('The environment is always set, as the utility has its own libpq',
+         dict(libpq=110000, names=True, params={},
+              expected={'PGREQUIREAUTH': '!gss,!sspi',
+                        'PGGSSENCMODE': 'disable'})),
+        ('The environment has the server\'s require_auth merged',
+         dict(names=True, params={'PGREQUIREAUTH': '!password'},
+              expected={'PGREQUIREAUTH': '!password,!gss,!sspi',
+                        'PGGSSENCMODE': 'disable'})),
+        ('The environment of a Kerberos connection refuses SSPI',
+         dict(source=KERBEROS, kerberos_conn=True, names=True, params={},
+              expected={'PGREQUIREAUTH': '!sspi',
+                        'PGGSSLIB': 'gssapi'})),
+        ('The environment overrides a PGGSSLIB of a Kerberos connection',
+         dict(source=KERBEROS, kerberos_conn=True, names=True,
+              params={'PGGSSLIB': 'sspi'},
+              expected={'PGREQUIREAUTH': '!sspi', 'PGGSSLIB': 'gssapi'})),
+        ('The environment has no PGGSSLIB where Kerberos is refused',
+         dict(source=INTERNAL, kerberos_conn=True, names=True, params={},
+              expected={'PGREQUIREAUTH': '!gss,!sspi',
+                        'PGGSSENCMODE': 'disable'})),
+    ]
+
+    server_mode = True
+    source = INTERNAL
+    kerberos_conn = True
+    ccache = USER_CACHE
+    libpq = 160000
+    add_missing = True
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        params = dict(self.params)
+        with patch.object(config, 'SERVER_MODE', self.server_mode), \
+                patch('pgadmin.utils.locker.libpq_version',
+                      return_value=self.libpq), \
+                self.app.test_request_context():
+            _login(self.source, self.ccache)
+            if getattr(self, 'names', False):
+                call = kerberos_libpq_env
+            else:
+                def call(p, k):
+                    apply_kerberos_conn_params(
+                        p, k, add_missing=self.add_missing)
+
+            if getattr(self, 'error', False):
+                with self.assertRaises(KerberosNotAllowed):
+                    call(params, self.kerberos_conn)
+                return
+            call(params, self.kerberos_conn)
+        self.assertEqual(params, self.expected)
+
+
+class ApplyKerberosConnParamsKeyInjectionTestCase(BaseTestGenerator):
+    """make_conninfo() does not escape parameter keys, so a crafted key
+    such as 'application_name=c require_auth' serialises its own
+    require_auth=; libpq takes the last of a duplicated keyword. The forced
+    require_auth/gssencmode must be re-inserted last so they win over any
+    such smuggled value, whatever position the user's own parameter held."""
+
+    scenarios = [
+        ('A crafted key cannot override the forced parameters',
+         dict(params={
+             'require_auth': '!password', 'gssencmode': 'disable',
+             'application_name=c require_auth': 'gss',
+             'fallback_application_name=d gssencmode': 'prefer'})),
+        ('A crafted key with a leading space cannot either',
+         dict(params={
+             'require_auth': '!password', 'gssencmode': 'disable',
+             ' require_auth': 'gss', ' gssencmode': 'prefer'})),
+    ]
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        params = dict(self.params)
+        with patch.object(config, 'SERVER_MODE', True), \
+                patch('pgadmin.utils.locker.libpq_version',
+                      return_value=160000), \
+                self.app.test_request_context():
+            _login(INTERNAL)
+            apply_kerberos_conn_params(params, is_kerberos_conn=False)
+
+        # libpq acts on the last value of a duplicated keyword, which is
+        # what conninfo_to_dict() returns for the serialised string.
+        effective = conninfo_to_dict(make_conninfo(**params))
+        self.assertEqual(effective.get('require_auth'), '!password,!gss,!sspi')
+        self.assertEqual(effective.get('gssencmode'), 'disable')
+
+
+class ConnectionStringKerberosKeyInjectionTestCase(BaseTestGenerator):
+    """create_connection_string() must not let a crafted parameter key steer
+    the connection to GSSAPI or SSPI. In the utility string the forced values
+    live in the environment (nothing is added to the string), so a smuggled
+    require_auth/gssencmode/gsslib is refused; in the in-process string the
+    forced value is serialised last and wins, so the safe value is what libpq
+    sees. A Kerberos user can try the same with gsslib to reach SSPI.
+
+    Such a key is never a libpq keyword, so the connection-parameter key
+    validation refuses it first. The Kerberos scenarios turn that validation
+    off, so that they show this check holds on its own."""
+
+    scenarios = [
+        ('A utility string with only a crafted require_auth key is refused',
+         dict(source=INTERNAL, kerberos_conn=False, for_utility=True,
+              params={'application_name=c require_auth': 'gss',
+                      'fallback_application_name=d gssencmode': 'prefer'},
+              refused=True)),
+        ('The in-process string forces the safe require_auth/gssencmode',
+         dict(source=INTERNAL, kerberos_conn=False, for_utility=False,
+              params={'application_name=c require_auth': 'gss',
+                      'fallback_application_name=d gssencmode': 'prefer'},
+              expected={'require_auth': '!gss,!sspi',
+                        'gssencmode': 'disable'})),
+        ('A utility string with only a crafted gsslib key is refused for a '
+         'Kerberos user',
+         dict(source=KERBEROS, kerberos_conn=True, for_utility=True,
+              params={'application_name=c gsslib': 'sspi'}, refused=True)),
+        ('The in-process string forces gsslib for a Kerberos user',
+         dict(source=KERBEROS, kerberos_conn=True, for_utility=False,
+              params={'application_name=c gsslib': 'sspi'},
+              expected={'gsslib': 'gssapi', 'require_auth': '!sspi'})),
+        ('The key validation refuses a crafted key before the Kerberos check',
+         dict(source=INTERNAL, kerberos_conn=False, for_utility=True,
+              params={'application_name=c require_auth': 'gss'},
+              key_check=True, refused=True, exc=PermissionError,
+              message='Invalid connection parameter')),
+    ]
+
+    refused = False
+    key_check = False
+    exc = KerberosNotAllowed
+    message = None
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        import pgadmin.utils.driver.psycopg3.server_manager as server_manager
+
+        manager = server_manager.ServerManager.__new__(
+            server_manager.ServerManager)
+        manager.host = '127.0.0.1'
+        manager.port = 5432
+        manager.use_ssh_tunnel = False
+        manager.service = None
+        manager.kerberos_conn = self.kerberos_conn
+        manager.connection_params = dict(self.params)
+
+        key_check = nullcontext() if self.key_check else patch.object(
+            server_manager, 'is_valid_connection_param_key',
+            return_value=True)
+        with patch.object(config, 'SERVER_MODE', True), \
+                patch('pgadmin.utils.locker.libpq_version',
+                      return_value=160000), \
+                key_check, \
+                self.app.test_request_context():
+            _login(self.source)
+            if self.refused:
+                with self.assertRaises(self.exc) as cm:
+                    manager.create_connection_string(
+                        'postgres', 'alice', for_utility=self.for_utility)
+                self.assertIsInstance(cm.exception, PermissionError)
+                if self.message:
+                    self.assertIn(self.message, str(cm.exception))
+            else:
+                dsn = manager.create_connection_string(
+                    'postgres', 'alice', for_utility=self.for_utility)
+                effective = conninfo_to_dict(dsn)
+                for key, value in self.expected.items():
+                    self.assertEqual(effective.get(key), value)