Replace free-text server-mode password exec commands with an administrator allowlist. (#10506)

In server mode, users choose a password exec command by name from
SERVER_PASSEXEC_COMMANDS rather than entering a command line. Existing
commands are converted on upgrade once the allowlist is configured; if
it is empty the conversion is postponed and a warning is logged.

docs/en_US/import_export_servers.rst

@@ -138,6 +138,15 @@ imported server.
 
 Password fields cannot be imported or exported.
 
+In desktop mode, ``PasswordExecCommand`` is the shell command to run, as shown
+in the example below. In server mode it must instead be the name of a command
+defined in the ``SERVER_PASSEXEC_COMMANDS`` setting (see :ref:`config_py`), and
+the same name is written when servers are exported. A server whose
+``PasswordExecCommand`` is not the name of a configured command is still
+imported, but without a password exec command, and a warning naming the server
+is printed. ``PasswordExecExpiration`` is imported only together with a valid
+command name.
+
 The following example shows both a minimally defined and a fully defined server:
 
 .. code-block:: python

docs/en_US/server_dialog.rst

@@ -211,7 +211,7 @@ Use the fields in the *Advanced* tab to configure a connection:
   that will be used against the pg_database table to limit the databases that you see.
   For example, you might enter: *live_db test_db* so that only live_db and test_db
   are shown in the pgAdmin object explorer.
-* Use the *Password exec command* field to specify a shell command to be executed
+* In desktop mode, use the *Password exec command* field to specify a shell command to be executed
   to retrieve a password to be used for SQL authentication. The ``stdout`` of the
   command will be used as the SQL password. This may be useful when the password
   should be generated as a transient authorization token instead of providing a
@@ -219,8 +219,36 @@ Use the fields in the *Advanced* tab to configure a connection:
   You can pass server hostname, port and DB username to the password exec command as variable by providing placeholders
   like ``%HOSTNAME%``, ``%PORT%`` and ``%USERNAME%`` which will be replace with the server connection information.
   Example: ``/path/to/script --hostname %HOSTNAME% --port %PORT% --username %USERNAME%``
+* In server mode, use the *Password exec command* drop-down list to select one
+  of the commands that the administrator has defined in the
+  ``SERVER_PASSEXEC_COMMANDS`` setting (see :ref:`config_py`). Users cannot
+  enter their own command. The list offers *None* and each configured command
+  name, and is hidden if no commands have been configured. On a shared server,
+  users who do not own the server also see *Inherit from owner*, which is the
+  default for them and uses whatever command the owner has selected; selecting
+  *None* or a named command overrides it for that user only. If a selected
+  command is later removed from the configuration, the server is treated as
+  having no command and a warning is logged.
+
+  The command is run directly, without a shell, and does not receive the full
+  environment of the pgAdmin process. Only ``PATH``, ``HOME``, ``LANG``,
+  ``LANGUAGE``, ``TZ``, ``TMPDIR``, ``TMP``, ``TEMP``, ``USER``, ``LOGNAME``
+  and any ``LC_*`` variable are passed through (and, on Windows,
+  ``SYSTEMROOT``, ``SYSTEMDRIVE``, ``PATHEXT``, ``COMSPEC`` and ``WINDIR``),
+  together with any variables the administrator has named in the
+  ``SERVER_PASSEXEC_ENV_PASSTHROUGH`` setting. The command also receives the
+  following variables describing the connection, which take precedence over
+  any variable of the same name:
+  ``PGADMIN_PASSEXEC_HOST``, ``PGADMIN_PASSEXEC_PORT``,
+  ``PGADMIN_PASSEXEC_USERNAME`` (the username that connection uses, so a
+  non-owner's own username on a shared server), ``PGADMIN_PASSEXEC_DATABASE``
+  (the maintenance database), ``PGADMIN_PASSEXEC_PGADMIN_USER`` (the pgAdmin
+  user making the connection) and ``PGADMIN_PASSEXEC_AUTH_SOURCE`` (how that
+  user authenticated to pgAdmin). The ``stdout`` of the command is used as the
+  SQL password.
 * Use the *Password exec expiration* field to specify a maximum age, in seconds,
-  of the password generated with a *Password exec command*. If not specified,
+  of the password generated with a *Password exec command*. In server mode, this field is
+  enabled only when a named command is selected. If not specified,
   the password will not expire until your pgAdmin session does.
   Zero means the command will be executed for each new connection or reconnection that is made.
   If the generated password is not valid indefinitely, set this value to slightly before it will expire.
@@ -234,7 +262,34 @@ Use the fields in the *Advanced* tab to configure a connection:
 .. note:: The password file option is only supported when pgAdmin is using libpq
     v10.0 or later to connect to the server.
 
-.. note:: The Password exec option is only supported when pgAdmin is run in desktop mode.
+.. note:: In server mode, the *Password exec command* setting is available
+    only when the administrator has defined at least one command in
+    ``SERVER_PASSEXEC_COMMANDS``. Free-text commands are never run in server
+    mode, and when pgAdmin is upgraded, an existing free-text command is kept
+    only if it exactly matches a configured command (its arguments joined by
+    single spaces), in which case it is converted to that command's name; all
+    others are cleared and a warning is logged. This conversion cannot be
+    undone, so ``SERVER_PASSEXEC_COMMANDS`` must be configured before the
+    first start after upgrading. If it is still empty at that point, the
+    conversion is postponed: existing commands are left in the configuration
+    database unused, a warning giving the number of affected servers is
+    logged at each start, and they are converted or cleared on the first
+    start at which at least one command is configured. Placeholders such as
+    ``%HOSTNAME%``, ``%PORT%`` and ``%USERNAME%`` are not substituted in
+    server mode, so a command containing them is converted only if the
+    configured arguments contain the same literal text. On a shared server, a
+    user who does not own the server and whose own command is cleared will
+    then inherit the owner's command, if the owner has one. The
+    ``ENABLE_SERVER_PASS_EXEC_CMD`` setting is deprecated and ignored.
+
+.. warning:: In server mode, only ``PGADMIN_PASSEXEC_PGADMIN_USER`` and
+    ``PGADMIN_PASSEXEC_AUTH_SOURCE`` identify the pgAdmin user on whose behalf
+    the command is run. ``PGADMIN_PASSEXEC_HOST``, ``PGADMIN_PASSEXEC_PORT``,
+    ``PGADMIN_PASSEXEC_USERNAME`` and ``PGADMIN_PASSEXEC_DATABASE`` are set by
+    whoever edits the server definition; on a shared server, a user who does
+    not own the server sets their own username. Commands should therefore
+    decide whether to issue a password based on the pgAdmin identity, and
+    treat the other four variables as untrusted input.
 
 Click the *Post Connection SQL* tab to continue.
 

web/config.py

@@ -1067,10 +1067,54 @@ AUTO_DISCOVER_SERVERS = True
 SERVER_HEARTBEAT_TIMEOUT = 30  # In seconds
 
 #############################################################################
-# ENABLE_SERVER_PASS_EXEC_CMD is used to enable/disable Password exec command
-# field in server properties. This is used to specify a shell command to be
-# executed to retrieve a password to be used for server authentication.
-# This setting is applicable only for server mode.
+# SERVER_PASSEXEC_COMMANDS is an allowlist of named commands that may be used
+# to retrieve a password for server authentication. Users pick a command by
+# name in the server properties; they cannot supply their own. Each value is
+# a list holding the executable followed by its arguments, for example:
+#
+#   SERVER_PASSEXEC_COMMANDS = {
+#       'vault': ['/usr/local/bin/get-pg-pass', '--ttl', '300']
+#   }
+#
+# Commands are run directly, without a shell, so no quoting or substitution
+# is applied to the arguments. Connection details are passed in these
+# environment variables: PGADMIN_PASSEXEC_HOST, PGADMIN_PASSEXEC_PORT,
+# PGADMIN_PASSEXEC_USERNAME, PGADMIN_PASSEXEC_DATABASE,
+# PGADMIN_PASSEXEC_PGADMIN_USER and PGADMIN_PASSEXEC_AUTH_SOURCE. Only the
+# last two identify the pgAdmin user; the host, port, username and database
+# are set by whoever edits the server, so commands must authorise on the
+# pgAdmin identity and treat the other four as untrusted input.
+# Names may not start with "__". An empty dictionary (the default) turns the
+# feature off. This setting is applicable only for server mode.
+#
+# Upgrading from a release that allowed free-text commands in server mode:
+# configure this BEFORE the first start after upgrading. On that start, an
+# existing free-text command is converted to the name of the entry whose
+# arguments, joined by single spaces, match it exactly, and every other
+# command is cleared; the clearing cannot be undone. If the dictionary is
+# empty at that start, the conversion is postponed: existing commands stay
+# in the configuration database unused, a warning is logged at each start,
+# and they are converted or cleared on the first start with at least one
+# entry configured.
+#############################################################################
+SERVER_PASSEXEC_COMMANDS = {}
+
+#############################################################################
+# SERVER_PASSEXEC_ENV_PASSTHROUGH lists the names of additional environment
+# variables to pass from pgAdmin's environment to the commands defined in
+# SERVER_PASSEXEC_COMMANDS, for example ['VAULT_ADDR']. The commands do not
+# receive pgAdmin's full environment; they get only PATH, HOME, LANG,
+# LANGUAGE, TZ, TMPDIR, TMP, TEMP, USER, LOGNAME, any LC_* variable (and on
+# Windows, SYSTEMROOT, SYSTEMDRIVE, PATHEXT, COMSPEC and WINDIR), the
+# variables named here, and the PGADMIN_PASSEXEC_* variables, which always
+# take precedence. Do not list variables that hold secrets the commands do
+# not need. This setting is applicable only for server mode.
+#############################################################################
+SERVER_PASSEXEC_ENV_PASSTHROUGH = []
+
+#############################################################################
+# DEPRECATED and ignored: free-text password exec commands are never run in
+# server mode. Use SERVER_PASSEXEC_COMMANDS instead.
 #############################################################################
 ENABLE_SERVER_PASS_EXEC_CMD = False
 

web/migrations/versions/server_passexec_name_.py

@@ -0,0 +1,50 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""Add passexec_name to the server and sharedserver tables
+
+In server mode the password exec command is chosen by name from the
+administrator-defined SERVER_PASSEXEC_COMMANDS setting, rather than
+being free text. This adds the column that holds the chosen name.
+Existing passexec_cmd values are converted at application start up,
+because that needs the configuration, which a migration cannot see.
+
+Revision ID: server_passexec_name
+Revises: normalize_locked_text_default
+Create Date: 2026-10-06
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = 'server_passexec_name'
+down_revision = 'normalize_locked_text_default'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    for table in ('server', 'sharedserver'):
+        if not inspector.has_table(table):
+            continue
+        existing_cols = {c['name'] for c in inspector.get_columns(table)}
+        if 'passexec_name' in existing_cols:
+            continue
+        with op.batch_alter_table(table) as batch_op:
+            batch_op.add_column(
+                sa.Column('passexec_name', sa.Text(), nullable=True))
+
+
+def downgrade():
+    # pgAdmin only upgrades, downgrade not implemented.
+    pass

web/pgadmin/__init__.py

@@ -543,6 +543,9 @@ def create_app(app_name=None):
         if not cli_mode:
             delete_adhoc_servers()
 
+        from pgadmin.utils.passexec import server_passexec_startup
+        server_passexec_startup(app, cli_mode)
+
     Mail(app)
 
     # Don't bother paths when running in cli mode

web/pgadmin/browser/__init__.py

@@ -68,6 +68,7 @@ from pgadmin.utils.constants import MIMETYPE_APP_JS, PGADMIN_NODE, \
     VW_EDT_DEFAULT_PLACEHOLDER, NO_CACHE_CONTROL
 from pgadmin.authenticate import AuthSourceManager
 from pgadmin.utils.exception import CryptKeyMissing
+from pgadmin.utils.passexec import get_server_passexec_commands
 
 from pgadmin.user_login_check import pga_login_required
 
@@ -563,7 +564,8 @@ def utils():
             shared_storage_list=shared_storage_list,
             restricted_shared_storage_list=[] if current_user.has_role(
                 "Administrator") else restricted_shared_storage_list,
-            enable_server_passexec_cmd=config.ENABLE_SERVER_PASS_EXEC_CMD,
+            server_passexec_commands=sorted(get_server_passexec_commands())
+            if config.SERVER_MODE else [],
             max_server_tags_allowed=config.MAX_SERVER_TAGS_ALLOWED,
             llm_enabled=is_llm_enabled(),
         ), 200)

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

@@ -30,6 +30,8 @@ import config
 from config import PG_DEFAULT_DRIVER
 from pgadmin.model import db, Server, ServerGroup, User, SharedServer
 from pgadmin.utils.driver import get_driver
+from pgadmin.utils.passexec import normalise_passexec_name, \
+    passexec_name_for_api, PASSEXEC_INHERIT
 from pgadmin.utils.master_password import get_crypt_key
 from pgadmin.utils.exception import CryptKeyMissing, ConnectionLost
 from pgadmin.tools.schema_diff.node_registry import SchemaDiffRegistry
@@ -65,6 +67,59 @@ def _is_non_owner(server):
     return server.shared and server.user_id != current_user.id
 
 
+def prepare_passexec_data(data, is_non_owner):
+    """Validate and normalise the password exec fields of a create or
+    update payload. Returns (data, errormsg); data is a plain dict copy,
+    since request.form is immutable.
+
+    Server mode: a free-text passexec_cmd is refused (an empty one is
+    dropped), and passexec_name is converted to its stored value.
+    Desktop mode: passexec_name is dropped; the command is untouched."""
+    data = dict(data)
+    if not config.SERVER_MODE:
+        data.pop('passexec_name', None)
+        return data, None
+    if data.get('passexec_cmd'):
+        return data, gettext(
+            'Free-text password exec commands are not allowed in server '
+            'mode; choose one of the commands configured by the '
+            'administrator.')
+    data.pop('passexec_cmd', None)
+    if 'passexec_name' in data:
+        if data['passexec_name'] == PASSEXEC_INHERIT and not is_non_owner:
+            return data, gettext(
+                'Only a user of a shared server can inherit the password '
+                'exec command from its owner.')
+        try:
+            data['passexec_name'] = normalise_passexec_name(
+                data['passexec_name'], is_non_owner)
+        except ValueError:
+            return data, gettext(
+                'The password exec command is not one of those configured '
+                'by the administrator.')
+    return data, None
+
+
+def passexec_api_fields(server, shared_server, is_non_owner):
+    """The password exec fields for the properties response. In server
+    mode the free-text command is never returned, and a non-owner sees
+    only their own SharedServer values (the overlaid server object
+    carries the owner's)."""
+    if not config.SERVER_MODE:
+        return {'passexec_cmd': server.passexec_cmd,
+                'passexec_expiration': server.passexec_expiration}
+    source = shared_server if is_non_owner else server
+    if source is None:
+        stored, expiration = None, None
+    else:
+        stored, expiration = source.passexec_name, \
+            source.passexec_expiration
+    return {
+        'passexec_name': passexec_name_for_api(stored, is_non_owner),
+        'passexec_expiration': expiration,
+    }
+
+
 def is_using_passfile(server, manager):
     """
     True if the connection authenticates through the configured passfile,
@@ -243,8 +298,6 @@ class ServerModule(sg.ServerGroupPluginModule):
         server.server_owner = sharedserver.server_owner
         server.password = sharedserver.password
         server.prepare_threshold = sharedserver.prepare_threshold
-        server.passexec_cmd = sharedserver.passexec_cmd
-        server.passexec_expiration = sharedserver.passexec_expiration
         server.kerberos_conn = sharedserver.kerberos_conn
         server.tags = sharedserver.tags
         server.post_connection_sql = sharedserver.post_connection_sql
@@ -499,6 +552,7 @@ class ServerModule(sg.ServerGroupPluginModule):
                 connection_params=safe_conn_params,
                 prepare_threshold=data.prepare_threshold,
                 passexec_cmd=None,
+                passexec_name=None,
                 passexec_expiration=None,
                 kerberos_conn=False,
                 tags=None,
@@ -890,6 +944,7 @@ class ServerNode(PGChildNodeView):
             'db_res': 'db_res',
             'db_res_type': 'db_res_type',
             'passexec_cmd': 'passexec_cmd',
+            'passexec_name': 'passexec_name',
             'passexec_expiration': 'passexec_expiration',
             'bgcolor': 'bgcolor',
             'fgcolor': 'fgcolor',
@@ -924,6 +979,12 @@ class ServerNode(PGChildNodeView):
             request.data
         )
 
+        data, passexec_err = prepare_passexec_data(
+            data, _is_non_owner(server))
+        if passexec_err:
+            return make_json_response(
+                success=0, status=400, errormsg=passexec_err)
+
         if 'db_res' in data and isinstance(data['db_res'], list):
             data['db_res'] = ','.join(data['db_res'])
 
@@ -978,10 +1039,6 @@ class ServerNode(PGChildNodeView):
         # which will affect the connections.
         if not conn.connected():
             manager.update(server)
-            # Suppress passexec for non-owners so the manager
-            # never holds the owner's password-exec command.
-            if _is_non_owner(server):
-                manager.passexec = None
 
         return jsonify(
             node=self.blueprint.generate_browser_node(
@@ -1010,7 +1067,10 @@ class ServerNode(PGChildNodeView):
     @staticmethod
     def _update_server_details(server, sharedserver,
                                config_param_map, arg, value):
-        if value == '':
+        # '' is meaningful for a non-owner's passexec_name: it means
+        # explicitly no command, as opposed to NULL, which inherits.
+        if value == '' and not (arg == 'passexec_name' and
+                                _is_non_owner(server)):
             value = None
 
         if _is_non_owner(server):
@@ -1031,7 +1091,7 @@ class ServerNode(PGChildNodeView):
         # SharedServer — they enable command/SQL execution
         # or are owner-level concepts not on SharedServer.
         _owner_only_fields = frozenset({
-            'passexec_cmd', 'passexec_expiration',
+            'passexec_cmd',
             'db_res', 'db_res_type',
         })
 
@@ -1149,7 +1209,9 @@ class ServerNode(PGChildNodeView):
         # port and user when server is connected
         display_connection_str = self.update_connection_string(manager, server)
 
-        if _is_non_owner(server):
+        shared_server = None
+        non_owner = bool(_is_non_owner(server))
+        if non_owner:
             shared_server = ServerModule.get_shared_server(server, gid)
             server = ServerModule.get_shared_server_properties(server,
                                                                shared_server)
@@ -1201,8 +1263,7 @@ class ServerNode(PGChildNodeView):
             'fgcolor': server.fgcolor,
             'db_res': get_db_restriction(server.db_res_type, server.db_res),
             'db_res_type': server.db_res_type,
-            'passexec_cmd': server.passexec_cmd,
-            'passexec_expiration': server.passexec_expiration,
+            **passexec_api_fields(server, shared_server, non_owner),
             'service': server.service if server.service else None,
             'use_ssh_tunnel': use_ssh_tunnel,
             'tunnel_host': tunnel_host,
@@ -1261,10 +1322,16 @@ class ServerNode(PGChildNodeView):
         # Loop through data and if found any value is blank string then
         # convert it to None as after porting into React, from frontend
         # '' blank string is coming as a value instead of null.
+        data = dict(data)
         for item in data:
             if data[item] == '':
                 data[item] = None
 
+        data, passexec_err = prepare_passexec_data(data, False)
+        if passexec_err:
+            return make_json_response(
+                success=0, status=400, errormsg=passexec_err)
+
         # Get enc key
         crypt_key_present, crypt_key = get_crypt_key()
         if not crypt_key_present:
@@ -1353,6 +1420,7 @@ class ServerNode(PGChildNodeView):
                 shared=data.get('shared', None),
                 shared_username=data.get('shared_username', None),
                 passexec_cmd=data.get('passexec_cmd', None),
+                passexec_name=data.get('passexec_name', None),
                 passexec_expiration=data.get('passexec_expiration', None),
                 kerberos_conn=1 if data.get('kerberos_conn', False) else 0,
                 connection_params=connection_params,
@@ -1649,12 +1717,6 @@ class ServerNode(PGChildNodeView):
         # the API call is not made from SQL Editor or View/Edit Data tool
         if not manager.connection().connected() and not is_qt:
             manager.update(server)
-            # Re-suppress passexec after update() which rebuilds
-            # from the (overlaid) server object.  Belt-and-suspenders:
-            # the overlay already defaults passexec to None, but this
-            # guards against direct DB edits.
-            if _is_non_owner(server):
-                manager.passexec = None
         conn = manager.connection()
 
         # Get enc key
@@ -1696,7 +1758,7 @@ class ServerNode(PGChildNodeView):
             conn_passwd = getattr(conn, 'password', None)
             if conn_passwd is None and not server.save_password and \
                     passfile_param is None and \
-                    server.passexec_cmd is None and \
+                    manager.passexec is None and \
                     server.service is None:
                 prompt_password = True
             elif passfile_param and passfile_param != '' and \

web/pgadmin/browser/server_groups/servers/static/js/server.js

@@ -8,7 +8,7 @@
 //////////////////////////////////////////////////////////////
 
 import { getNodeListById } from '../../../../static/js/node_ajax';
-import ServerSchema from './server.ui';
+import ServerSchema, { passexecForServerCopy } from './server.ui';
 import { showServerPassword, showChangeServerPassword, showNamedRestorePoint } from '../../../../../static/js/Dialogs/index';
 import _ from 'lodash';
 import getApiInstance, { parseApiError } from '../../../../../static/js/api_instance';
@@ -64,6 +64,7 @@ define('pgadmin.node.server', [
           d.shared = false;
           d.server_owner = null;
           d.shared_username = null;
+          passexecForServerCopy(d);
         }
         return d;
       },

web/pgadmin/browser/server_groups/servers/static/js/server.ui.js

@@ -47,6 +47,17 @@ class TagsSchema extends BaseUISchema {
   }
 }
 
+// A copy of another user's shared server becomes the current user's own
+// server, which cannot inherit a password exec command, so an inherited
+// command becomes none. Free-text commands never apply in server mode.
+export function passexecForServerCopy(d) {
+  if (d.passexec_name === '__inherit__') {
+    d.passexec_name = '__none__';
+  }
+  delete d.passexec_cmd;
+  return d;
+}
+
 export function getConnectionParameters() {
   let conParams = [{
     'value': 'hostaddr', 'label': gettext('Host address'), 'vartype': 'string'
@@ -193,6 +204,7 @@ export default class ServerSchema extends BaseUISchema {
       db_res: undefined,
       db_res_type: 'databases',
       passexec: undefined,
+      passexec_name: pgAdmin.server_mode == 'True' ? '__none__' : undefined,
       passexec_expiration: undefined,
       service: undefined,
       shared_username: '',
@@ -565,14 +577,40 @@ export default class ServerSchema extends BaseUISchema {
         id: 'passexec_cmd', label: gettext('Password exec command'), type: 'text',
         group: gettext('Advanced'), controlProps: {maxLength: null},
         mode: ['properties', 'edit', 'create'],
-        disabled: pgAdmin.server_mode == 'True' && pgAdmin.enable_server_passexec_cmd == 'False',
+        visible: () => pgAdmin.server_mode != 'True',
         helpMessage: gettext('The server hostname, port, and username can be passed as variables by using the placeholders %HOSTNAME%, %PORT%, and %USERNAME%, which will be replaced with the corresponding server connection information.')
       },
+      {
+        id: 'passexec_name', label: gettext('Password exec command'),
+        group: gettext('Advanced'),
+        mode: ['properties', 'edit', 'create'],
+        controlProps: {allowClear: false},
+        visible: () => pgAdmin.server_mode == 'True' &&
+          (pgAdmin.server_passexec_commands || []).length > 0,
+        type: (state) => {
+          let options = [{label: gettext('None'), value: '__none__'}];
+          // Only a non-owner of a shared server can inherit the owner's command.
+          if (obj.isShared(state)) {
+            options.unshift({label: gettext('Inherit from owner'), value: '__inherit__'});
+          }
+          (pgAdmin.server_passexec_commands || []).forEach((name) => {
+            options.push({label: name, value: name});
+          });
+          return {type: 'select', options: options};
+        },
+        helpMessage: gettext('The command is chosen from those configured by the administrator. The server hostname, port, username and maintenance database, and your pgAdmin username and authentication source, are passed to it in the PGADMIN_PASSEXEC_* environment variables.')
+      },
       {
         id: 'passexec_expiration', label: gettext('Password exec expiration (seconds)'), type: 'int',
         group: gettext('Advanced'),
         mode: ['properties', 'edit', 'create'],
+        visible: () => pgAdmin.server_mode != 'True' ||
+          (pgAdmin.server_passexec_commands || []).length > 0,
         disabled: function(state) {
+          if (pgAdmin.server_mode == 'True') {
+            return !state.passexec_name ||
+              ['__none__', '__inherit__'].includes(state.passexec_name);
+          }
           return isEmptyString(state.passexec_cmd);
         },
       },

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

@@ -10,6 +10,7 @@
 """Tests for server data isolation between users in server mode."""
 
 import json
+import sqlite3
 import config
 from pgadmin.utils.route import BaseTestGenerator
 from regression.python_test_utils import test_utils as utils
@@ -135,7 +136,7 @@ class SharedServerFieldSuppressionTestCase(BaseTestGenerator):
     when a non-owner accesses a shared server's properties."""
 
     scenarios = [
-        ('Shared server suppresses passexec_cmd and '
+        ('Shared server suppresses '
          'post_connection_sql for non-owner',
          dict(is_positive_test=True)),
     ]
@@ -149,7 +150,6 @@ class SharedServerFieldSuppressionTestCase(BaseTestGenerator):
 
         # Create a shared server with sensitive owner-only fields
         self.server['shared'] = True
-        self.server['passexec_cmd'] = '/usr/bin/get-secret'
         self.server['passexec_expiration'] = 100
         self.server['post_connection_sql'] = 'SET role admin;'
         url = "/browser/server/obj/{0}/".format(utils.SERVER_GROUP)
@@ -163,10 +163,22 @@ class SharedServerFieldSuppressionTestCase(BaseTestGenerator):
         self.assertIn('node', response_data)
         self.server_id = response_data['node']['_id']
 
+        # The API no longer accepts a free-text command in server mode,
+        # so plant one as a row left over from an older release would be.
+        conn = sqlite3.connect(config.TEST_SQLITE_PATH)
+        try:
+            conn.execute(
+                'UPDATE server SET passexec_cmd=?, passexec_name=? '
+                'WHERE id=?',
+                ('/usr/bin/get-secret', 'vault', self.server_id))
+            conn.commit()
+        finally:
+            conn.close()
+
     @create_user_wise_test_client(test_user_details)
     def runTest(self):
-        """Non-owner should NOT see passexec_cmd or
-        post_connection_sql in properties response."""
+        """Non-owner should NOT see post_connection_sql in the
+        properties response."""
         if not self.server_id:
             raise Exception("Server not found to test suppression")
 
@@ -176,17 +188,19 @@ class SharedServerFieldSuppressionTestCase(BaseTestGenerator):
         self.assertEqual(response.status_code, 200)
         data = json.loads(response.data.decode('utf-8'))
 
-        # passexec_cmd must be None/null for non-owners
-        self.assertIsNone(
-            data.get('passexec_cmd'),
-            'passexec_cmd should be suppressed for non-owners.'
-            ' Got: {0}'.format(data.get('passexec_cmd'))
-        )
+        # The free-text command is never returned in server mode, and
+        # the owner's expiration must not reach a non-owner (they see
+        # only their own SharedServer value, None until they set one).
+        self.assertNotIn(
+            'passexec_cmd', data,
+            'passexec_cmd should not be returned in server mode.')
         self.assertIsNone(
             data.get('passexec_expiration'),
-            'passexec_expiration should be suppressed for '
-            'non-owners.'
-        )
+            'The owner passexec_expiration should not be exposed to '
+            'non-owners.')
+        # With no choice of their own the non-owner inherits; they are
+        # not told which command the owner uses.
+        self.assertEqual(data.get('passexec_name'), '__inherit__')
         # post_connection_sql must be None/null for non-owners
         self.assertIsNone(
             data.get('post_connection_sql'),

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

@@ -0,0 +1,428 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""Tests for named password exec commands in the servers API.
+
+The server-mode API cases run only when the suite itself runs in server
+mode (with a second, non-admin user for the non-owner cases), because
+patching config.SERVER_MODE on a desktop-mode app also switches the login
+checks in before_request and fails every request. The desktop cases patch
+SERVER_MODE off, which works in either mode, and the view helpers are
+tested with mocks in both modes."""
+
+import json
+import sqlite3
+from unittest.mock import MagicMock, patch
+
+import config
+from pgadmin.browser.server_groups.servers import (
+    prepare_passexec_data, passexec_api_fields)
+from pgadmin.utils.passexec import (
+    normalise_passexec_name, passexec_name_for_api,
+    PASSEXEC_INHERIT, PASSEXEC_NONE)
+from pgadmin.utils.route import BaseTestGenerator
+from regression.python_test_utils import test_utils as utils
+from regression.test_setup import config_data
+from regression.python_test_utils.test_utils import \
+    create_user_wise_test_client
+
+test_user_details = None
+if config.SERVER_MODE:
+    test_user_details = \
+        config_data['pgAdmin4_test_non_admin_credentials']
+
+COMMANDS = {'vault': ['/usr/bin/vault-get', '--short']}
+CMDS_PATH = 'pgadmin.utils.passexec.get_server_passexec_commands'
+
+
+class PassexecNameConversionTestCase(BaseTestGenerator):
+    """Pure tests for normalise_passexec_name."""
+
+    scenarios = [
+        ('owner none', dict(owner=True, value=None, expect=None)),
+        ('owner empty', dict(owner=True, value='', expect=None)),
+        ('owner __none__',
+         dict(owner=True, value=PASSEXEC_NONE, expect=None)),
+        ('owner name', dict(owner=True, value='vault', expect='vault')),
+        ('owner unknown name',
+         dict(owner=True, value='nope', expect=ValueError)),
+        ('owner __inherit__',
+         dict(owner=True, value=PASSEXEC_INHERIT, expect=ValueError)),
+        ('non-owner __inherit__',
+         dict(owner=False, value=PASSEXEC_INHERIT, expect=None)),
+        ('non-owner none', dict(owner=False, value=None, expect='')),
+        ('non-owner empty', dict(owner=False, value='', expect='')),
+        ('non-owner __none__',
+         dict(owner=False, value=PASSEXEC_NONE, expect='')),
+        ('non-owner name',
+         dict(owner=False, value='vault', expect='vault')),
+        ('non-owner unknown name',
+         dict(owner=False, value='nope', expect=ValueError)),
+    ]
+
+    def runTest(self):
+        with patch(CMDS_PATH, return_value=COMMANDS):
+            if self.expect is ValueError:
+                with self.assertRaises(ValueError):
+                    normalise_passexec_name(self.value, not self.owner)
+            else:
+                self.assertEqual(
+                    normalise_passexec_name(self.value, not self.owner),
+                    self.expect)
+
+
+class PassexecNameForApiTestCase(BaseTestGenerator):
+    """passexec_name_for_api is the inverse of normalise_passexec_name."""
+
+    scenarios = [
+        ('owner none', dict(non_owner=False, stored=None,
+                            expect=PASSEXEC_NONE)),
+        ('owner name', dict(non_owner=False, stored='vault',
+                            expect='vault')),
+        ('non-owner none', dict(non_owner=True, stored=None,
+                                expect=PASSEXEC_INHERIT)),
+        ('non-owner empty', dict(non_owner=True, stored='',
+                                 expect=PASSEXEC_NONE)),
+        ('non-owner name', dict(non_owner=True, stored='vault',
+                                expect='vault')),
+    ]
+
+    def runTest(self):
+        self.assertEqual(
+            passexec_name_for_api(self.stored, self.non_owner),
+            self.expect)
+
+
+class PassexecPayloadHelpersTestCase(BaseTestGenerator):
+    """prepare_passexec_data and passexec_api_fields, which carry the
+    owner and non-owner rules for the create, update and properties
+    views."""
+
+    scenarios = [('helpers', dict())]
+
+    def runTest(self):
+        with patch(CMDS_PATH, return_value=COMMANDS):
+            # Server mode: free-text command refused.
+            with patch.object(config, 'SERVER_MODE', True):
+                for non_owner in (False, True):
+                    _, err = prepare_passexec_data(
+                        {'passexec_cmd': 'echo x'}, non_owner)
+                    self.assertIsNotNone(err)
+                # Empty command key is dropped.
+                data, err = prepare_passexec_data(
+                    {'passexec_cmd': '', 'name': 'a'}, False)
+                self.assertIsNone(err)
+                self.assertEqual(data, {'name': 'a'})
+                # Non-owner values map to SharedServer semantics.
+                data, err = prepare_passexec_data(
+                    {'passexec_name': PASSEXEC_NONE}, True)
+                self.assertEqual(data['passexec_name'], '')
+                data, err = prepare_passexec_data(
+                    {'passexec_name': PASSEXEC_INHERIT}, True)
+                self.assertIsNone(data['passexec_name'])
+                data, err = prepare_passexec_data(
+                    {'passexec_name': 'vault',
+                     'passexec_expiration': 30}, True)
+                self.assertEqual(data['passexec_name'], 'vault')
+                # Owner may not inherit; unknown names refused.
+                _, err = prepare_passexec_data(
+                    {'passexec_name': PASSEXEC_INHERIT}, False)
+                self.assertIsNotNone(err)
+                _, err = prepare_passexec_data(
+                    {'passexec_name': 'nope'}, True)
+                self.assertIsNotNone(err)
+
+            # Desktop mode: name dropped, command kept.
+            with patch.object(config, 'SERVER_MODE', False):
+                data, err = prepare_passexec_data(
+                    {'passexec_name': 'vault', 'passexec_cmd': 'echo x'},
+                    False)
+                self.assertIsNone(err)
+                self.assertEqual(data, {'passexec_cmd': 'echo x'})
+
+        # Properties for a non-owner come from their SharedServer row,
+        # never from the overlaid owner object.
+        owner_obj = MagicMock(passexec_cmd='secret', passexec_name='vault',
+                              passexec_expiration=99)
+        with patch.object(config, 'SERVER_MODE', True):
+            ss = MagicMock(passexec_name=None, passexec_expiration=None)
+            fields = passexec_api_fields(owner_obj, ss, True)
+            self.assertNotIn('passexec_cmd', fields)
+            self.assertEqual(fields['passexec_name'], PASSEXEC_INHERIT)
+            self.assertIsNone(fields['passexec_expiration'])
+            ss = MagicMock(passexec_name='', passexec_expiration=5)
+            fields = passexec_api_fields(owner_obj, ss, True)
+            self.assertEqual(fields['passexec_name'], PASSEXEC_NONE)
+            self.assertEqual(fields['passexec_expiration'], 5)
+            fields = passexec_api_fields(owner_obj, None, False)
+            self.assertNotIn('passexec_cmd', fields)
+            self.assertEqual(fields['passexec_name'], 'vault')
+            self.assertEqual(fields['passexec_expiration'], 99)
+        with patch.object(config, 'SERVER_MODE', False):
+            fields = passexec_api_fields(owner_obj, None, False)
+            self.assertEqual(fields, {'passexec_cmd': 'secret',
+                                      'passexec_expiration': 99})
+
+
+class PassexecServerApiTestCase(BaseTestGenerator):
+    """Owner create/update/properties through the real views in server
+    mode."""
+
+    scenarios = [('api', dict())]
+
+    def setUp(self):
+        self.created = []
+        if not config.SERVER_MODE:
+            self.skipTest('Server-mode rules need a server-mode run.')
+        self.url = '/browser/server/obj/{0}/'.format(utils.SERVER_GROUP)
+
+    def tearDown(self):
+        for sid in self.created:
+            utils.delete_server_with_api(self.tester, sid)
+
+    def _post(self, **extra):
+        payload = dict(self.server)
+        payload.pop('passexec_cmd', None)
+        payload.update(extra)
+        response = self.tester.post(
+            self.url, data=json.dumps(payload), content_type='html/json')
+        if response.status_code == 200:
+            self.created.append(
+                json.loads(response.data.decode('utf-8'))['node']['_id'])
+        return response
+
+    def _row(self, sid):
+        conn = sqlite3.connect(config.TEST_SQLITE_PATH)
+        try:
+            conn.row_factory = sqlite3.Row
+            return conn.execute(
+                'SELECT passexec_cmd, passexec_name FROM server '
+                'WHERE id=?', (sid,)).fetchone()
+        finally:
+            conn.close()
+
+    def _props(self, sid):
+        response = self.tester.get(self.url + str(sid))
+        self.assertEqual(response.status_code, 200)
+        return json.loads(response.data.decode('utf-8'))
+
+    def runTest(self):
+        with patch(CMDS_PATH, return_value=COMMANDS), \
+                patch.object(config, 'SERVER_MODE', True):
+            # 1. valid name stored and returned.
+            r = self._post(passexec_name='vault')
+            self.assertEqual(r.status_code, 200)
+            sid = self.created[-1]
+            self.assertEqual(self._row(sid)['passexec_name'], 'vault')
+            props = self._props(sid)
+            self.assertEqual(props['passexec_name'], 'vault')
+            self.assertNotIn('passexec_cmd', props)
+
+            # 2. unknown name refused.
+            self.assertEqual(
+                self._post(passexec_name='nope').status_code, 400)
+
+            # 3a. free-text command refused in server mode.
+            self.assertEqual(
+                self._post(passexec_cmd='echo x').status_code, 400)
+
+            # 4. owner may not choose inherit.
+            r = self.tester.put(
+                self.url + str(sid),
+                data=json.dumps({'passexec_name': PASSEXEC_INHERIT}),
+                content_type='html/json')
+            self.assertEqual(r.status_code, 400)
+            self.assertEqual(self._row(sid)['passexec_name'], 'vault')
+
+            # Owner update to none clears it.
+            r = self.tester.put(
+                self.url + str(sid),
+                data=json.dumps({'passexec_name': PASSEXEC_NONE}),
+                content_type='html/json')
+            self.assertEqual(r.status_code, 200)
+            self.assertIsNone(self._row(sid)['passexec_name'])
+            self.assertEqual(
+                self._props(sid)['passexec_name'], PASSEXEC_NONE)
+
+
+class PassexecDesktopApiTestCase(PassexecServerApiTestCase):
+    """Desktop rules through the real views: the free-text command is
+    stored and a name is ignored."""
+
+    scenarios = [('desktop api', dict())]
+
+    def setUp(self):
+        self.created = []
+        self.url = '/browser/server/obj/{0}/'.format(utils.SERVER_GROUP)
+
+    def runTest(self):
+        with patch.object(config, 'SERVER_MODE', False):
+            # 3b. desktop mode stores the command; 8. ignores the name.
+            r = self._post(passexec_cmd='echo x', passexec_name='vault')
+            self.assertEqual(r.status_code, 200)
+            row = self._row(self.created[-1])
+            self.assertEqual(row['passexec_cmd'], 'echo x')
+            self.assertIsNone(row['passexec_name'])
+
+
+class PassexecSharedServerUpdateTestCase(BaseTestGenerator):
+    """Non-owner updates: _set_valid_attr_value stores the name and
+    expiration on the SharedServer and leaves the owner's row alone."""
+
+    scenarios = [('shared update', dict())]
+
+    def runTest(self):
+        from pgadmin.browser.server_groups.servers import ServerNode
+        node = ServerNode.__new__(ServerNode)
+        cmap = {'passexec_name': 'passexec_name',
+                'passexec_expiration': 'passexec_expiration',
+                'passexec_cmd': 'passexec_cmd'}
+        srv_mod = 'pgadmin.browser.server_groups.servers'
+        for api, stored in ((PASSEXEC_NONE, ''), (PASSEXEC_INHERIT, None),
+                            ('vault', 'vault')):
+            owner = MagicMock(shared=True, user_id=1, passexec_name='keep',
+                              passexec_cmd=None, passexec_expiration=None)
+            ss = MagicMock(passexec_name='old', passexec_expiration=None)
+            with patch(CMDS_PATH, return_value=COMMANDS), \
+                    patch.object(config, 'SERVER_MODE', True), \
+                    patch(srv_mod + '.current_user', MagicMock(id=2)), \
+                    patch(srv_mod + '.get_crypt_key',
+                          return_value=(True, b'k')):
+                data, err = prepare_passexec_data(
+                    {'passexec_name': api, 'passexec_expiration': 30}, True)
+                self.assertIsNone(err)
+                node._set_valid_attr_value(1, data, cmap, owner, ss)
+            self.assertEqual(ss.passexec_name, stored)
+            self.assertEqual(ss.passexec_expiration, 30)
+            self.assertEqual(owner.passexec_name, 'keep')
+
+
+class PassexecNonOwnerApiTestCase(BaseTestGenerator):
+    """A second user updating and reading a shared server's password exec
+    settings through the real views."""
+
+    scenarios = [('non-owner api', dict())]
+
+    def setUp(self):
+        self.server_ids = []
+        if not config.SERVER_MODE:
+            self.skipTest('Shared servers only apply to server mode.')
+        url = '/browser/server/obj/{0}/'.format(utils.SERVER_GROUP)
+        with patch(CMDS_PATH, return_value=COMMANDS):
+            for _ in range(2):
+                payload = dict(self.server)
+                payload.pop('passexec_cmd', None)
+                payload.pop('passexec_expiration', None)
+                payload.update(shared=True, passexec_name='vault')
+                response = self.tester.post(
+                    url, data=json.dumps(payload),
+                    content_type='html/json')
+                self.assertEqual(response.status_code, 200)
+                self.server_ids.append(json.loads(
+                    response.data.decode('utf-8'))['node']['_id'])
+
+    def tearDown(self):
+        for sid in self.server_ids:
+            utils.delete_server_with_api(self.__class__.tester, sid)
+
+    @staticmethod
+    def _query(sql, args):
+        conn = sqlite3.connect(config.TEST_SQLITE_PATH)
+        try:
+            return conn.execute(sql, args).fetchall()
+        finally:
+            conn.close()
+
+    def _shared_row(self, sid):
+        rows = self._query(
+            'SELECT passexec_name, passexec_expiration FROM sharedserver '
+            'WHERE osid=?', (sid,))
+        self.assertEqual(len(rows), 1)
+        return rows[0]
+
+    def _owner_row(self, sid):
+        return self._query(
+            'SELECT passexec_cmd, passexec_name, passexec_expiration '
+            'FROM server WHERE id=?', (sid,))[0]
+
+    @staticmethod
+    def _set_user_role(grant):
+        """The test user is created without a role, so it cannot update
+        servers; give it the standard 'User' role while the test runs."""
+        conn = sqlite3.connect(config.TEST_SQLITE_PATH)
+        try:
+            args = (test_user_details['login_username'],)
+            if grant:
+                conn.execute(
+                    'INSERT INTO roles_users (user_id, role_id) '
+                    'SELECT u.id, r.id FROM user u, role r '
+                    'WHERE u.username=? AND r.name=\'User\'', args)
+            else:
+                conn.execute(
+                    'DELETE FROM roles_users WHERE user_id IN '
+                    '(SELECT id FROM user WHERE username=?)', args)
+            conn.commit()
+        finally:
+            conn.close()
+
+    @create_user_wise_test_client(test_user_details)
+    def runTest(self):
+        self._set_user_role(True)
+        try:
+            self._run_as_non_owner()
+        finally:
+            self._set_user_role(False)
+
+    def _run_as_non_owner(self):
+        sid, other = self.server_ids
+        base = '/browser/server/obj/{0}/'.format(utils.SERVER_GROUP)
+        owner_before = self._owner_row(sid)
+        self.assertEqual(owner_before, (None, 'vault', None))
+
+        def put(payload):
+            return self.tester.put(
+                base + str(sid), data=json.dumps(payload),
+                content_type='html/json')
+
+        with patch(CMDS_PATH, return_value=COMMANDS):
+            r = put({'passexec_name': PASSEXEC_NONE})
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(self._shared_row(sid)[0], '')
+            self.assertEqual(self._owner_row(sid), owner_before)
+
+            r = put({'passexec_name': PASSEXEC_INHERIT})
+            self.assertEqual(r.status_code, 200)
+            self.assertIsNone(self._shared_row(sid)[0])
+            self.assertEqual(self._owner_row(sid), owner_before)
+
+            r = put({'passexec_name': 'vault', 'passexec_expiration': 30})
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(self._shared_row(sid), ('vault', 30))
+            self.assertEqual(self._owner_row(sid), owner_before)
+
+            r = put({'passexec_cmd': 'echo x'})
+            self.assertEqual(r.status_code, 400)
+            self.assertEqual(self._shared_row(sid), ('vault', 30))
+            self.assertEqual(self._owner_row(sid), owner_before)
+
+            # Properties for a user with no SharedServer row yet.
+            self.assertEqual(self._query(
+                'SELECT 1 FROM sharedserver WHERE osid=?', (other,)), [])
+            r = self.tester.get(base + str(other), follow_redirects=True)
+            self.assertEqual(r.status_code, 200)
+            data = json.loads(r.data.decode('utf-8'))
+            self.assertEqual(data['passexec_name'], PASSEXEC_INHERIT)
+            self.assertNotIn('passexec_cmd', data)
+            self.assertIsNone(data['passexec_expiration'])
+
+            # And their own choices are what properties then reports.
+            r = self.tester.get(base + str(sid), follow_redirects=True)
+            data = json.loads(r.data.decode('utf-8'))
+            self.assertEqual(data['passexec_name'], 'vault')
+            self.assertEqual(data['passexec_expiration'], 30)

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

@@ -102,7 +102,7 @@ class TestGetSharedServerProperties(BaseTestGenerator):
     using mock objects."""
 
     scenarios = [
-        ('Merge overlays passexec_cmd from SharedServer',
+        ('Merge keeps the owner passexec fields',
          dict(test_method='test_overlays_passexec')),
         ('Merge overlays post_connection_sql from SharedServer',
          dict(test_method='test_overlays_post_sql')),
@@ -138,17 +138,15 @@ class TestGetSharedServerProperties(BaseTestGenerator):
             server, ss)
 
     def test_overlays_passexec(self):
-        # SharedServer defaults have None - overlay copies that.
-        result = self._merge()
-        self.assertIsNone(result.passexec_cmd)
-        self.assertIsNone(result.passexec_expiration)
-        # If SharedServer has a value, it should appear.
+        # The overlay must not copy the SharedServer's passexec fields;
+        # resolve_server_passexec() reads them from the SharedServer row.
         ss = _make_shared_server(
             passexec_cmd='/usr/bin/get-pw',
             passexec_expiration=120)
         result = self._merge(ss=ss)
-        self.assertEqual(result.passexec_cmd, '/usr/bin/get-pw')
-        self.assertEqual(result.passexec_expiration, 120)
+        self.assertEqual(result.passexec_cmd,
+                         '/usr/bin/vault-get-secret')
+        self.assertEqual(result.passexec_expiration, 300)
 
     def test_overlays_post_sql(self):
         # SharedServer defaults have None - overlay copies that.
@@ -339,8 +337,9 @@ class TestMergeExpungesServer(BaseTestGenerator):
             # Should not crash
             result = ServerModule.get_shared_server_properties(
                 server, ss)
-        # SharedServer defaults passexec_cmd to None
-        self.assertIsNone(result.passexec_cmd)
+        # The owner's passexec fields are left alone by the overlay
+        self.assertEqual(result.passexec_cmd,
+                         '/usr/bin/vault-get-secret')
 
 
 class TestUpdateConnectionParameter(BaseTestGenerator):

web/pgadmin/browser/templates/browser/js/utils.js

@@ -69,8 +69,8 @@ define('pgadmin.browser.utils',
   /* Minimum password length */
   pgAdmin['password_length_min'] = '{{password_length_min}}';
 
-  /* Enable server password exec command */
-  pgAdmin['enable_server_passexec_cmd'] = '{{enable_server_passexec_cmd}}';
+  /* Named server-mode password exec commands */
+  pgAdmin['server_passexec_commands'] = {{ server_passexec_commands|tojson }};
 
   /* LLM/AI features enabled */
   pgAdmin['llm_enabled'] = '{{llm_enabled}}' == 'True';

web/pgadmin/model/__init__.py

@@ -34,7 +34,7 @@ import config
 #
 ##########################################################################
 
-SCHEMA_VERSION = 52
+SCHEMA_VERSION = 53
 
 ##########################################################################
 #
@@ -319,6 +319,7 @@ class Server(db.Model, UserScopedMixin):
     db_res_type = db.Column(db.String(32), default='databases')
     passexec_cmd = db.Column(db.Text(), nullable=True)
     passexec_expiration = db.Column(db.Integer(), nullable=True)
+    passexec_name = db.Column(db.Text(), nullable=True)
     bgcolor = db.Column(db.String(10), nullable=True)
     fgcolor = db.Column(db.String(10), nullable=True)
     service = db.Column(db.Text(), nullable=True)
@@ -601,6 +602,7 @@ class SharedServer(db.Model, UserScopedMixin):
     prepare_threshold = db.Column(db.Integer(), nullable=True)
     passexec_cmd = db.Column(db.Text(), nullable=True)
     passexec_expiration = db.Column(db.Integer(), nullable=True)
+    passexec_name = db.Column(db.Text(), nullable=True)
     kerberos_conn = db.Column(
         db.Boolean(), nullable=False, default=False
     )

web/pgadmin/utils/__init__.py

@@ -580,14 +580,18 @@ def dump_database_servers(output_file, selected_servers,
             add_value(attr_dict, "PostConnectionSQL",
                       server.post_connection_sql)
 
-            # if desktop mode or server mode with
-            # ENABLE_SERVER_PASS_EXEC_CMD flag is True
-            if not current_app.config['SERVER_MODE'] or \
-                    current_app.config['ENABLE_SERVER_PASS_EXEC_CMD']:
+            if not current_app.config['SERVER_MODE']:
                 add_value(attr_dict, "PasswordExecCommand",
                           server.passexec_cmd)
                 add_value(attr_dict, "PasswordExecExpiration",
                           server.passexec_expiration)
+            elif server.passexec_name:
+                # Server mode never exports free-text commands, only
+                # the name of an entry in SERVER_PASSEXEC_COMMANDS.
+                add_value(attr_dict, "PasswordExecCommand",
+                          server.passexec_name)
+                add_value(attr_dict, "PasswordExecExpiration",
+                          server.passexec_expiration)
 
             servers_dumped = servers_dumped + 1
 
@@ -705,6 +709,27 @@ def validate_json_data(data, is_admin):
     return None
 
 
+def _load_server_passexec(new_server, obj):
+    """Apply PasswordExecCommand from an import file in server mode,
+    where it must be the name of a configured command. Anything else
+    is ignored with a warning that names the server only, as the text
+    could be a secret."""
+    from pgadmin.utils.passexec import get_server_passexec_commands
+
+    name = obj.get("PasswordExecCommand", None)
+    if not name:
+        return
+
+    if isinstance(name, str) and name in get_server_passexec_commands():
+        new_server.passexec_name = name
+        new_server.passexec_expiration = obj.get(
+            "PasswordExecExpiration", None)
+    else:
+        print("Ignoring the password exec command for server '%s' as it "
+              "is not the name of a command in SERVER_PASSEXEC_COMMANDS." %
+              obj.get("Name"))
+
+
 def load_database_servers(input_file, selected_servers,
                           load_user=current_user, from_setup=False,
                           auth_source=INTERNAL):
@@ -859,13 +884,12 @@ def load_database_servers(input_file, selected_servers,
 
             new_server.post_connection_sql = obj.get("PostConnectionSQL", None)
 
-            # if desktop mode or server mode with
-            # ENABLE_SERVER_PASS_EXEC_CMD flag is True
-            if not current_app.config['SERVER_MODE'] or \
-                    current_app.config['ENABLE_SERVER_PASS_EXEC_CMD']:
+            if not current_app.config['SERVER_MODE']:
                 new_server.passexec_cmd = obj.get("PasswordExecCommand", None)
                 new_server.passexec_expiration = obj.get(
                     "PasswordExecExpiration", None)
+            else:
+                _load_server_passexec(new_server, obj)
 
             db.session.add(new_server)
 

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

@@ -84,12 +84,6 @@ class Driver(BaseDriver):
                 for server in servers:
                     manager = managers[str(server.id)] = \
                         ServerManager(server)
-                    # Suppress passexec for non-owners of shared
-                    # servers — it runs commands on the client
-                    # machine and must not inherit the owner's.
-                    if config.SERVER_MODE and server.shared and \
-                            server.user_id != current_user.id:
-                        manager.passexec = None
                     if server.id in session_managers:
                         manager._restore(
                             session_managers[server.id])
@@ -152,12 +146,6 @@ class Driver(BaseDriver):
             # server_data was already access-checked above;
             # it cannot be None at this point.
             manager = ServerManager(server_data)
-            # Suppress passexec for non-owners of shared
-            # servers — it runs commands on the client machine
-            # and must not inherit the owner's.
-            if config.SERVER_MODE and server_data.shared and \
-                    server_data.user_id != current_user.id:
-                manager.passexec = None
             managers[str(sid)] = manager
 
             return manager

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

@@ -28,7 +28,7 @@ from pgadmin.utils.exception import ConnectionLost, SSHTunnelConnectionLost,\
     CryptKeyMissing
 from pgadmin.utils.master_password import get_crypt_key
 from pgadmin.utils.exception import ObjectGone
-from pgadmin.utils.passexec import PasswordExec
+from pgadmin.utils.passexec import build_passexec
 from psycopg.conninfo import make_conninfo
 
 if config.SUPPORT_SSH_TUNNEL:
@@ -83,10 +83,7 @@ class ServerManager(object):
         self.db_res = server.db_res
         self.db_res_type = server.db_res_type
         self.name = server.name
-        self.passexec = \
-            PasswordExec(server.passexec_cmd, server.host, server.port,
-                         server.username, server.passexec_expiration) \
-            if server.passexec_cmd else None
+        self.passexec = build_passexec(server)
         self.service = server.service
 
         if config.SUPPORT_SSH_TUNNEL:

web/pgadmin/utils/passexec.py

@@ -7,22 +7,197 @@
 #
 ##########################################################################
 import logging
+import os
 import subprocess
 from datetime import datetime, timedelta, timezone
 from threading import Lock
 
 from flask import current_app
+from flask_security import current_user
 
 import config
 from pgadmin.utils.driver import get_driver
 
+PASSEXEC_INHERIT = '__inherit__'
+PASSEXEC_NONE = '__none__'
+
+
+def _iter_server_passexec_entries():
+    """Yield (name, argv_or_None, reason) for each configured entry."""
+    cmds = getattr(config, 'SERVER_PASSEXEC_COMMANDS', None) or {}
+    if not isinstance(cmds, dict):
+        yield None, None, 'SERVER_PASSEXEC_COMMANDS must be a dict'
+        return
+    for name, argv in cmds.items():
+        if not isinstance(name, str) or not name.strip():
+            yield name, None, 'name must be a non-empty string'
+        elif name.startswith('__'):
+            yield name, None, 'names starting with "__" are reserved'
+        elif not isinstance(argv, (list, tuple)) or not argv or \
+                not all(isinstance(a, str) for a in argv):
+            yield name, None, 'value must be a non-empty list of strings'
+        else:
+            yield name, list(argv), None
+
+
+def get_server_passexec_commands():
+    """Return the valid entries of SERVER_PASSEXEC_COMMANDS as
+    {name: argv}, with fresh lists so callers cannot alter config."""
+    return {name: argv for name, argv, reason
+            in _iter_server_passexec_entries() if reason is None}
+
+
+def normalise_passexec_name(value, is_non_owner):
+    """Map the name sent by the API to the value stored in the database.
+
+    Owner (Server row): None, '' and '__none__' store None (no command);
+    a configured name stores itself. '__inherit__' is meaningless for an
+    owner. Non-owner (SharedServer row): '__inherit__' stores None,
+    None, '' and '__none__' store '' (explicitly no command), and a
+    configured name stores itself. Anything else raises ValueError."""
+    if value is None or value == '' or value == PASSEXEC_NONE:
+        return '' if is_non_owner else None
+    if value == PASSEXEC_INHERIT:
+        if is_non_owner:
+            return None
+        raise ValueError('Inheriting is only available for shared servers.')
+    if not isinstance(value, str) or \
+            value not in get_server_passexec_commands():
+        raise ValueError('Unknown password exec command.')
+    return value
+
+
+def passexec_name_for_api(stored, is_non_owner):
+    """The inverse of normalise_passexec_name, for the properties API."""
+    if is_non_owner:
+        if stored is None:
+            return PASSEXEC_INHERIT
+        return stored or PASSEXEC_NONE
+    return stored or PASSEXEC_NONE
+
+
+def check_server_passexec_config(logger):
+    """Log configuration problems once, at startup."""
+    for name, _argv, reason in _iter_server_passexec_entries():
+        if reason is not None:
+            logger.error('Ignoring SERVER_PASSEXEC_COMMANDS entry %r: %s',
+                         name, reason)
+    extra = getattr(config, 'SERVER_PASSEXEC_ENV_PASSTHROUGH', None) or []
+    if not isinstance(extra, (list, tuple)) or \
+            not all(isinstance(n, str) for n in extra):
+        logger.error('SERVER_PASSEXEC_ENV_PASSTHROUGH must be a list of '
+                     'environment variable names; ignoring any entry that '
+                     'is not a string.')
+    if getattr(config, 'ENABLE_SERVER_PASS_EXEC_CMD', False):
+        logger.warning(
+            'ENABLE_SERVER_PASS_EXEC_CMD is deprecated and no longer has '
+            'any effect; define the permitted commands in '
+            'SERVER_PASSEXEC_COMMANDS instead.')
+
+
+def convert_legacy_server_passexec(logger):
+    """Server mode only: convert free-text passexec_cmd values that
+    exactly match a configured command into its name, and clear every
+    passexec_cmd, since free-text commands no longer run in server mode.
+    Command text is never logged because it may contain secrets.
+    Returns the number of rows changed.
+
+    The conversion is irreversible, so it does not run until at least
+    one command is configured: with an empty SERVER_PASSEXEC_COMMANDS
+    nothing could match and every command would be cleared. Until then
+    the legacy values stay in the database, where nothing in server
+    mode reads, runs, returns or exports them, and a warning is logged
+    at each start so the administrator knows to configure the setting
+    before they are converted."""
+    from pgadmin.model import db, Server, SharedServer
+    by_cmd = {' '.join(argv): name
+              for name, argv in get_server_passexec_commands().items()}
+    if not by_cmd:
+        for model in (Server, SharedServer):
+            pending = model.query.filter(
+                model.passexec_cmd.isnot(None),
+                model.passexec_cmd != '').count()
+            if pending:
+                logger.warning(
+                    'SERVER_PASSEXEC_COMMANDS is not configured, but %s '
+                    'row(s) in the %s table still hold a password exec '
+                    'command from before the upgrade. They are not run in '
+                    'server mode. Configure SERVER_PASSEXEC_COMMANDS '
+                    'before the next start: commands that match a '
+                    'configured entry are then converted to its name, '
+                    'and the rest are cleared.',
+                    pending, model.__tablename__)
+        return 0
+    changed = 0
+    for model in (Server, SharedServer):
+        # Normalise empty strings quietly; they carry no command.
+        model.query.filter(model.passexec_cmd == '').update(
+            {model.passexec_cmd: None}, synchronize_session=False)
+        rows = model.query.filter(model.passexec_cmd.isnot(None)).all()
+        for row in rows:
+            name = by_cmd.get(row.passexec_cmd.strip())
+            if name is not None:
+                row.passexec_name = name
+            else:
+                logger.warning(
+                    'Cleared the password exec command on %s %s (user %s): '
+                    'free-text commands are not run in server mode and it '
+                    'does not match any SERVER_PASSEXEC_COMMANDS entry.',
+                    model.__tablename__, row.id, row.user_id)
+            row.passexec_cmd = None
+            changed += 1
+    db.session.commit()
+    return changed
+
+
+# Variables copied from pgAdmin's environment into a server-mode password
+# exec command's environment. Everything else is withheld, because the
+# process environment can hold secrets such as another user's PGPASSWORD
+# (set by export_password_env) or the container's setup password.
+PASSEXEC_BASE_ENV = ('PATH', 'HOME', 'LANG', 'LANGUAGE', 'TZ', 'TMPDIR',
+                     'TMP', 'TEMP', 'USER', 'LOGNAME')
+PASSEXEC_WINDOWS_ENV = ('SYSTEMROOT', 'SYSTEMDRIVE', 'PATHEXT', 'COMSPEC',
+                        'WINDIR')
+
+
+def server_passexec_base_env():
+    """Return the reduced environment for a server-mode password exec
+    command: the base allowlist, any LC_* variable, the Windows system
+    variables on Windows, and the names in
+    SERVER_PASSEXEC_ENV_PASSTHROUGH, each copied only if it is set."""
+    names = set(PASSEXEC_BASE_ENV)
+    if os.name == 'nt':
+        names.update(PASSEXEC_WINDOWS_ENV)
+    extra = getattr(config, 'SERVER_PASSEXEC_ENV_PASSTHROUGH', None) or []
+    if isinstance(extra, (list, tuple)):
+        names.update(n for n in extra if isinstance(n, str))
+    if os.name == 'nt':
+        # os.environ upper-cases its keys on Windows.
+        names = {n.upper() for n in names}
+    return {k: v for k, v in os.environ.items()
+            if k in names or k.startswith('LC_')}
+
+
+def server_passexec_startup(app, cli_mode):
+    """Run the server-mode password exec checks when the app starts.
+
+    The legacy conversion is skipped in CLI mode: setup.py does not set
+    SERVER_MODE, so it would otherwise clear the free-text commands in a
+    desktop configuration database passed with --sqlite-path."""
+    if not config.SERVER_MODE:
+        return
+    check_server_passexec_config(app.logger)
+    if not cli_mode:
+        convert_legacy_server_passexec(app.logger)
+
 
 class PasswordExec:
 
-    lock = Lock()
-
     def __init__(self, cmd, host, port, username, expiration_seconds=None,
                  timeout=60):
+        # One lock per instance, so a slow command only delays callers
+        # waiting for the same cached password, not every user.
+        self.lock = Lock()
         self.host = host
         self.port = port
         self.username = username
@@ -34,7 +209,7 @@ class PasswordExec:
         self.last_result = None
 
     def get(self):
-        if config.SERVER_MODE and not config.ENABLE_SERVER_PASS_EXEC_CMD:
+        if config.SERVER_MODE:
             # Arbitrary shell execution on server is a security risk
             raise NotImplementedError('Passexec not available in server mode')
         driver = get_driver(config.PG_DEFAULT_DRIVER)
@@ -45,21 +220,27 @@ class PasswordExec:
         self.cmd = self.cmd.replace(
             '%USERNAME%',
             driver.qtIdent(None, self.username) if self.username else '')
+        return self._get_cached(lambda: subprocess.run(
+            self.cmd,
+            shell=True,
+            timeout=self.timeout,
+            capture_output=True,
+            text=True,
+            check=True,
+        ), skip=not self.cmd)
+
+    def _get_cached(self, run, skip=False):
+        """Return the cached password, calling run() when it is missing
+        or expired. If skip is true and a run would be needed, return
+        None instead."""
         with self.lock:
             if not self.password or self.is_expired():
-                if not self.cmd:
+                if skip:
                     return None
                 current_app.logger.info('Calling passexec')
                 now = datetime.now(timezone.utc)
                 try:
-                    p = subprocess.run(
-                        self.cmd,
-                        shell=True,
-                        timeout=self.timeout,
-                        capture_output=True,
-                        text=True,
-                        check=True,
-                    )
+                    p = run()
                 except subprocess.CalledProcessError as e:
                     if e.stderr:
                         self.create_logger().error(e.stderr)
@@ -82,3 +263,116 @@ class PasswordExec:
         for h in current_app.logger.handlers:
             logger.addHandler(h)
         return logger
+
+
+class ServerPasswordExec(PasswordExec):
+    """Runs an administrator-defined command without a shell, passing
+    connection details in environment variables so that no value can be
+    interpreted as part of the command."""
+
+    def __init__(self, name, argv, host, port, username, database,
+                 pgadmin_user, auth_source, expiration_seconds=None,
+                 timeout=60):
+        super().__init__(None, host, port, username, expiration_seconds,
+                         timeout)
+        self.name = name
+        self.argv = list(argv)
+        self.env = server_passexec_base_env()
+        self.env.update({
+            'PGADMIN_PASSEXEC_HOST': host or '',
+            'PGADMIN_PASSEXEC_PORT': str(port) if port is not None else '',
+            'PGADMIN_PASSEXEC_USERNAME': username or '',
+            'PGADMIN_PASSEXEC_DATABASE': database or '',
+            'PGADMIN_PASSEXEC_PGADMIN_USER': pgadmin_user or '',
+            'PGADMIN_PASSEXEC_AUTH_SOURCE': auth_source or '',
+        })
+
+    def get(self):
+        return self._get_cached(self._run)
+
+    def _run(self):
+        """Run the command, replacing any failure with an exception that
+        names only the command. The subprocess exceptions include the
+        full argument list in their text, and that may contain secrets,
+        so it must reach neither the user nor the log."""
+        logger = self.create_logger()
+        try:
+            return subprocess.run(
+                self.argv, shell=False, env=self.env, timeout=self.timeout,
+                capture_output=True, text=True, check=True)
+        except subprocess.CalledProcessError as e:
+            logger.error('Password exec command %r exited with status %s.',
+                         self.name, e.returncode)
+            self._log_stderr(logger, e.stderr)
+        except subprocess.TimeoutExpired as e:
+            logger.error('Password exec command %r timed out after %s '
+                         'seconds.', self.name, self.timeout)
+            self._log_stderr(logger, e.stderr)
+        except OSError as e:
+            logger.error('Password exec command %r could not be run '
+                         '(errno %s: %s).', self.name, e.errno, e.strerror)
+        raise Exception(
+            "Password exec command '{0}' failed.".format(self.name)) \
+            from None
+
+    def _log_stderr(self, logger, stderr):
+        if isinstance(stderr, bytes):
+            stderr = stderr.decode('utf-8', 'replace')
+        if stderr:
+            logger.error('Password exec command %r stderr: %s',
+                         self.name, stderr)
+
+
+def resolve_server_passexec(server, user):
+    """Work out which named command applies to this server for this
+    pgAdmin user. Returns (name, expiration, db_username) or None.
+
+    A non-owner of a shared server inherits the owner's command unless
+    their SharedServer row says otherwise (NULL inherits, '' is none,
+    anything else is their own choice). The owner's values are read from
+    the Server row's passexec fields, which get_shared_server_properties
+    deliberately leaves alone."""
+    from pgadmin.model import SharedServer
+    if server.shared and server.user_id != user.id:
+        ss = SharedServer.query.filter_by(osid=server.id,
+                                          user_id=user.id).first()
+        db_user = ss.username if ss else server.shared_username
+        own = ss.passexec_name if ss else None
+        if own == '':
+            return None
+        if own is not None:
+            return own, ss.passexec_expiration, db_user
+        if not server.passexec_name:
+            return None
+        return server.passexec_name, server.passexec_expiration, db_user
+    if not server.passexec_name:
+        return None
+    return server.passexec_name, server.passexec_expiration, server.username
+
+
+def build_passexec(server):
+    """Build the password exec object for a ServerManager, or None."""
+    if not config.SERVER_MODE:
+        if not server.passexec_cmd:
+            return None
+        return PasswordExec(server.passexec_cmd, server.host, server.port,
+                            server.username, server.passexec_expiration)
+    commands = get_server_passexec_commands()
+    if not commands:
+        return None
+    if not current_user or not current_user.is_authenticated:
+        return None
+    resolved = resolve_server_passexec(server, current_user)
+    if resolved is None:
+        return None
+    name, expiration, db_user = resolved
+    argv = commands.get(name)
+    if argv is None:
+        current_app.logger.warning(
+            'Server %s uses password exec command %r, which is not in '
+            'SERVER_PASSEXEC_COMMANDS; ignoring it.', server.id, name)
+        return None
+    return ServerPasswordExec(
+        name, argv, server.host, server.port, db_user,
+        server.maintenance_db, current_user.username,
+        getattr(current_user, 'auth_source', None), expiration)

web/pgadmin/utils/tests/test_passexec_convert.py

@@ -0,0 +1,154 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL License
+#
+##########################################################################
+from unittest.mock import patch, MagicMock
+
+from pgadmin.utils.route import BaseTestGenerator
+from pgadmin.utils import passexec
+from pgadmin.utils.passexec import convert_legacy_server_passexec
+
+VAULT_CMD = '/bin/get-pass --ttl 300'
+SECRET_CMD = 'echo SECRET-TOKEN-123'
+
+
+class TestConvertLegacyServerPassexec(BaseTestGenerator):
+    """Free-text commands become names where they match the allowlist
+    and are cleared otherwise, without logging command text. With no
+    commands configured nothing is touched, and a warning counts the
+    rows still waiting to be converted."""
+
+    scenarios = [
+        ('Commands configured', dict(configured=True)),
+        ('No commands configured', dict(configured=False)),
+    ]
+
+    def setUp(self):
+        self.ctx = self.app.app_context()
+        self.ctx.push()
+        from pgadmin.model import db, User, ServerGroup, Server, \
+            SharedServer
+        user = User.query.first()
+        self.group = ServerGroup(user_id=user.id,
+                                 name='Passexec convert test group')
+        db.session.add(self.group)
+        db.session.flush()
+        self.group_id = self.group.id
+
+        def make(cmd):
+            s = Server(user_id=user.id, servergroup_id=self.group_id,
+                       name='passexec-test', host='db.example.com',
+                       port=5432, username='test_user',
+                       maintenance_db='postgres', passexec_cmd=cmd)
+            db.session.add(s)
+            db.session.flush()
+            return s
+
+        self.a = make(VAULT_CMD).id
+        self.b = make(SECRET_CMD).id
+        self.c = make(None).id
+        self.d = make('').id
+        shared = SharedServer(
+            osid=self.a, user_id=user.id, server_owner=user.username,
+            servergroup_id=self.group_id, name='passexec-test-shared',
+            host='db.example.com', port=5432, username='test_user',
+            maintenance_db='postgres', shared=True,
+            passexec_cmd=VAULT_CMD)
+        db.session.add(shared)
+        db.session.flush()
+        self.shared = shared.id
+        db.session.commit()
+
+    def runTest(self):
+        if self.configured:
+            self.check_converted()
+        else:
+            self.check_postponed()
+
+    def check_postponed(self):
+        from pgadmin.model import db, Server, SharedServer
+        logger = MagicMock()
+        with patch.object(passexec.config, 'SERVER_PASSEXEC_COMMANDS',
+                          {}, create=True):
+            self.assertEqual(convert_legacy_server_passexec(logger), 0)
+            db.session.expire_all()
+
+            for sid, cmd in ((self.a, VAULT_CMD), (self.b, SECRET_CMD),
+                             (self.c, None), (self.d, '')):
+                row = db.session.get(Server, sid)
+                self.assertEqual((row.passexec_name, row.passexec_cmd),
+                                 (None, cmd))
+            s = db.session.get(SharedServer, self.shared)
+            self.assertEqual((s.passexec_name, s.passexec_cmd),
+                             (None, VAULT_CMD))
+
+            # One warning per table with pending rows, counting at least
+            # this test's rows (the shared test DB may hold others), and
+            # never the command text.
+            warned = {c[0][2]: c[0][1]
+                      for c in logger.warning.call_args_list}
+            self.assertGreaterEqual(warned.get(Server.__tablename__, 0), 2)
+            self.assertGreaterEqual(
+                warned.get(SharedServer.__tablename__, 0), 1)
+            for call in logger.method_calls:
+                self.assertNotIn('SECRET-TOKEN-123', repr(call))
+                self.assertNotIn('get-pass', repr(call))
+
+    def check_converted(self):
+        from pgadmin.model import db, Server, SharedServer
+        logger = MagicMock()
+        with patch.object(passexec.config, 'SERVER_PASSEXEC_COMMANDS',
+                          {'vault': ['/bin/get-pass', '--ttl', '300']},
+                          create=True):
+            # Other rows in the shared test DB may also be converted, so
+            # only require that ours were counted.
+            changed = convert_legacy_server_passexec(logger)
+            self.assertGreaterEqual(changed, 3)
+            db.session.expire_all()
+
+            a = db.session.get(Server, self.a)
+            self.assertEqual((a.passexec_name, a.passexec_cmd),
+                             ('vault', None))
+            b = db.session.get(Server, self.b)
+            self.assertEqual((b.passexec_name, b.passexec_cmd),
+                             (None, None))
+            c = db.session.get(Server, self.c)
+            self.assertEqual((c.passexec_name, c.passexec_cmd),
+                             (None, None))
+            d = db.session.get(Server, self.d)
+            self.assertIsNone(d.passexec_cmd)
+            self.assertIsNone(d.passexec_name)
+            s = db.session.get(SharedServer, self.shared)
+            self.assertEqual((s.passexec_name, s.passexec_cmd),
+                             ('vault', None))
+
+            # Other rows in the shared test DB may also be warned about,
+            # so look only at warnings naming this test's rows: one for b,
+            # none for the converted, empty or NULL rows.
+            warned = [(c[0][1], c[0][2])
+                      for c in logger.warning.call_args_list]
+            ours = [w for w in warned
+                    if w in {(Server.__tablename__, i) for i in
+                             (self.a, self.b, self.c, self.d)} or
+                    w == (SharedServer.__tablename__, self.shared)]
+            self.assertEqual(ours, [(Server.__tablename__, self.b)])
+            for call in logger.method_calls:
+                self.assertNotIn('SECRET-TOKEN-123', repr(call))
+                self.assertNotIn('get-pass', repr(call))
+
+            # Idempotent.
+            self.assertEqual(convert_legacy_server_passexec(logger), 0)
+
+    def tearDown(self):
+        from pgadmin.model import db, Server, SharedServer, ServerGroup
+        db.session.rollback()
+        SharedServer.query.filter_by(id=self.shared).delete()
+        for sid in (self.a, self.b, self.c, self.d):
+            Server.query.filter_by(id=sid).delete()
+        ServerGroup.query.filter_by(id=self.group_id).delete()
+        db.session.commit()
+        self.ctx.pop()

web/pgadmin/utils/tests/test_passexec_import_export.py

@@ -0,0 +1,174 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL License
+#
+##########################################################################
+import io
+import json
+import os
+import shutil
+import tempfile
+from contextlib import redirect_stdout
+from unittest.mock import patch
+
+from pgadmin.utils.route import BaseTestGenerator
+from pgadmin.utils import passexec
+
+COMMANDS = {'vault': ['/bin/get-pass', '--ttl', '300']}
+FREE_TEXT = 'echo SECRET-TOKEN-123'
+GROUP = 'Passexec import export test group'
+
+
+class _ImportExportMixin:
+    """Shared set-up: a scratch directory, and a clean test group."""
+
+    def setUp(self):
+        # gettext needs a request context to pick a locale.
+        self.ctx = self.app.test_request_context()
+        self.ctx.push()
+        from pgadmin.model import User
+        self.user = User.query.first()
+        self.tmp = tempfile.mkdtemp()
+        self.path = os.path.join(self.tmp, 'servers.json')
+        self._cleanup()
+
+    def tearDown(self):
+        from pgadmin.model import db
+        db.session.rollback()
+        self._cleanup()
+        shutil.rmtree(self.tmp, ignore_errors=True)
+        self.ctx.pop()
+
+    def _cleanup(self):
+        from pgadmin.model import db, Server, ServerGroup
+        group = ServerGroup.query.filter_by(
+            user_id=self.user.id, name=GROUP).first()
+        if group:
+            Server.query.filter_by(servergroup_id=group.id).delete()
+            db.session.delete(group)
+        db.session.commit()
+
+    def _add_server(self, name, **kwargs):
+        from pgadmin.model import db, Server, ServerGroup
+        group = ServerGroup.query.filter_by(
+            user_id=self.user.id, name=GROUP).first()
+        if group is None:
+            group = ServerGroup(user_id=self.user.id, name=GROUP)
+            db.session.add(group)
+            db.session.flush()
+        server = Server(user_id=self.user.id, servergroup_id=group.id,
+                        name=name, host='db.example.com', port=5432,
+                        username='test_user', maintenance_db='postgres',
+                        **kwargs)
+        db.session.add(server)
+        db.session.commit()
+        return server
+
+    def _dump(self):
+        from pgadmin.utils import dump_database_servers
+        with redirect_stdout(io.StringIO()):
+            ok, _ = dump_database_servers(
+                self.path, None, self.user, from_setup=True)
+        self.assertTrue(ok)
+        with open(self.path) as f:
+            servers = json.load(f)['Servers']
+        return {s['Name']: s for s in servers.values()
+                if s['Group'] == GROUP}
+
+    def _load(self, servers):
+        from pgadmin.utils import load_database_servers
+        servers = {str(i): dict({'Group': GROUP, 'Host': 'db.example.com',
+                                 'Port': 5432, 'Username': 'test_user',
+                                 'MaintenanceDB': 'postgres'}, **s)
+                   for i, s in enumerate(servers, 1)}
+        with open(self.path, 'w') as f:
+            json.dump({'Servers': servers}, f)
+        out = io.StringIO()
+        with redirect_stdout(out):
+            ok, _ = load_database_servers(
+                self.path, None, self.user, from_setup=True)
+        self.assertTrue(ok)
+        return out.getvalue()
+
+    def _loaded(self, name):
+        from pgadmin.model import Server, ServerGroup
+        group = ServerGroup.query.filter_by(
+            user_id=self.user.id, name=GROUP).first()
+        return Server.query.filter_by(
+            servergroup_id=group.id, name=name).one()
+
+
+class TestServerModePasswordExecExport(
+        _ImportExportMixin, BaseTestGenerator):
+    """Server mode exports the command name, never free text."""
+
+    def runTest(self):
+        self._add_server('with-name', passexec_name='vault',
+                         passexec_expiration=100)
+        self._add_server('no-name', passexec_expiration=50)
+        self._add_server('legacy-text', passexec_cmd=FREE_TEXT)
+        with patch.dict(self.app.config, {'SERVER_MODE': True}):
+            dumped = self._dump()
+
+        self.assertEqual(dumped['with-name']['PasswordExecCommand'],
+                         'vault')
+        self.assertEqual(dumped['with-name']['PasswordExecExpiration'], 100)
+        for name in ('no-name', 'legacy-text'):
+            self.assertNotIn('PasswordExecCommand', dumped[name])
+            self.assertNotIn('PasswordExecExpiration', dumped[name])
+        self.assertNotIn('SECRET-TOKEN-123', json.dumps(dumped))
+
+
+class TestServerModePasswordExecImport(
+        _ImportExportMixin, BaseTestGenerator):
+    """Server mode imports only names that are configured."""
+
+    def runTest(self):
+        with patch.dict(self.app.config, {'SERVER_MODE': True}), \
+                patch.object(passexec.config, 'SERVER_PASSEXEC_COMMANDS',
+                             COMMANDS, create=True):
+            out = self._load([
+                {'Name': 'by-name', 'PasswordExecCommand': 'vault',
+                 'PasswordExecExpiration': 120},
+                {'Name': 'by-text', 'PasswordExecCommand': FREE_TEXT,
+                 'PasswordExecExpiration': 120},
+                {'Name': 'unknown', 'PasswordExecCommand': 'missing'},
+            ])
+
+        s = self._loaded('by-name')
+        self.assertEqual((s.passexec_name, s.passexec_cmd,
+                          s.passexec_expiration), ('vault', None, 120))
+
+        s = self._loaded('by-text')
+        self.assertEqual((s.passexec_name, s.passexec_cmd,
+                          s.passexec_expiration), (None, None, None))
+
+        self.assertIn("'by-text'", out)
+        self.assertIn("'unknown'", out)
+        self.assertNotIn('SECRET-TOKEN-123', out)
+        self.assertNotIn('echo', out)
+
+
+class TestDesktopModePasswordExecRoundTrip(
+        _ImportExportMixin, BaseTestGenerator):
+    """Desktop mode still exports and imports the free-text command."""
+
+    def runTest(self):
+        self._add_server('desktop', passexec_cmd=FREE_TEXT,
+                         passexec_expiration=30)
+        with patch.dict(self.app.config, {'SERVER_MODE': False}):
+            dumped = self._dump()
+            self.assertEqual(dumped['desktop']['PasswordExecCommand'],
+                             FREE_TEXT)
+            self.assertEqual(dumped['desktop']['PasswordExecExpiration'],
+                             30)
+
+            self._load([{'Name': 'desktop-copy',
+                         'PasswordExecCommand': FREE_TEXT,
+                         'PasswordExecExpiration': 30}])
+        s = self._loaded('desktop-copy')
+        self.assertEqual((s.passexec_cmd, s.passexec_name,
+                          s.passexec_expiration), (FREE_TEXT, None, 30))

web/pgadmin/utils/tests/test_passexec_resolve.py

@@ -0,0 +1,164 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL License
+#
+##########################################################################
+
+"""Tests for resolve_server_passexec() and build_passexec()."""
+
+from unittest.mock import patch, MagicMock
+
+from pgadmin.utils.route import BaseTestGenerator
+from pgadmin.utils.passexec import (
+    resolve_server_passexec, build_passexec, PasswordExec,
+    ServerPasswordExec)
+
+CMDS = {'vault': ['/bin/get-pass'], 'other': ['/bin/other']}
+
+
+def _server(**kw):
+    s = MagicMock(id=7, user_id=1, shared=True, host='db.example.com',
+                  port=5432, username='owner_db', shared_username='shr',
+                  maintenance_db='postgres', passexec_name='vault',
+                  passexec_expiration=60, passexec_cmd=None)
+    for k, v in kw.items():
+        setattr(s, k, v)
+    return s
+
+
+OWNER = MagicMock(id=1, username='owner@example.com',
+                  auth_source='internal')
+OTHER = MagicMock(id=2, username='jane.doe@example.com',
+                  auth_source='ldap')
+
+
+class TestResolveServerPassexec(BaseTestGenerator):
+    """Which named command applies to which pgAdmin user."""
+
+    scenarios = [
+        ('Owner gets own command',
+         dict(srv={}, user=OWNER, row=None,
+              expected=('vault', 60, 'owner_db'))),
+        ('Owner with no command',
+         dict(srv=dict(passexec_name=None), user=OWNER, row=None,
+              expected=None)),
+        ('Unshared server uses the owner values for any user',
+         dict(srv=dict(shared=False), user=OTHER, row=None,
+              expected=('vault', 60, 'owner_db'))),
+        ('Non-owner without a row inherits',
+         dict(srv={}, user=OTHER, row=None,
+              expected=('vault', 60, 'shr'))),
+        ('Non-owner row with NULL inherits',
+         dict(srv={}, user=OTHER,
+              row=dict(passexec_name=None, username='jane_db'),
+              expected=('vault', 60, 'jane_db'))),
+        ('Non-owner row with empty string means none',
+         dict(srv={}, user=OTHER,
+              row=dict(passexec_name='', username='jane_db'),
+              expected=None)),
+        ('Non-owner row with its own command',
+         dict(srv={}, user=OTHER,
+              row=dict(passexec_name='other', passexec_expiration=5,
+                       username='jane_db'),
+              expected=('other', 5, 'jane_db'))),
+        ('Non-owner inherits from an owner with none',
+         dict(srv=dict(passexec_name=None), user=OTHER,
+              row=dict(passexec_name=None, username='jane_db'),
+              expected=None)),
+    ]
+
+    def runTest(self):
+        server = _server(**self.srv)
+        row = MagicMock(**self.row) if self.row is not None else None
+        with patch('pgadmin.model.SharedServer') as ss:
+            ss.query.filter_by.return_value.first.return_value = row
+            self.assertEqual(resolve_server_passexec(server, self.user),
+                             self.expected)
+
+
+class TestBuildPassexec(BaseTestGenerator):
+    """build_passexec() in server and desktop mode."""
+
+    scenarios = [
+        ('Builds a ServerPasswordExec for a non-owner',
+         dict(test_method='test_builds')),
+        ('Unknown name returns None and warns',
+         dict(test_method='test_unknown_name')),
+        ('Server mode never uses free text',
+         dict(test_method='test_server_free_text')),
+        ('Desktop mode keeps free text',
+         dict(test_method='test_desktop')),
+        ('Unauthenticated user gets None',
+         dict(test_method='test_unauthenticated')),
+        ('No configured commands returns None without a query',
+         dict(test_method='test_no_commands')),
+    ]
+
+    def runTest(self):
+        getattr(self, self.test_method)()
+
+    def _build(self, server, user=OTHER, row=None, server_mode=True,
+               cmds=CMDS):
+        with self.app.test_request_context(), \
+                patch('pgadmin.utils.passexec.config.SERVER_MODE',
+                      server_mode), \
+                patch('pgadmin.utils.passexec.config.'
+                      'SERVER_PASSEXEC_COMMANDS', cmds, create=True), \
+                patch('pgadmin.utils.passexec.current_user', user), \
+                patch('pgadmin.model.SharedServer') as ss:
+            self.shared_server = ss
+            ss.query.filter_by.return_value.first.return_value = row
+            return build_passexec(server)
+
+    def test_builds(self):
+        res = self._build(
+            _server(),
+            row=MagicMock(passexec_name=None, username='jane_db'))
+        self.assertIsInstance(res, ServerPasswordExec)
+        self.assertEqual(res.env['PGADMIN_PASSEXEC_USERNAME'], 'jane_db')
+        self.assertEqual(res.env['PGADMIN_PASSEXEC_PGADMIN_USER'],
+                         'jane.doe@example.com')
+        self.assertEqual(res.env['PGADMIN_PASSEXEC_AUTH_SOURCE'], 'ldap')
+        self.assertEqual(res.argv, ['/bin/get-pass'])
+        self.assertEqual(res.name, 'vault')
+
+    def test_unknown_name(self):
+        server = _server(passexec_name='gone', user_id=2)
+        app = MagicMock()
+        with patch('pgadmin.utils.passexec.current_app', app):
+            res = self._build(server)
+        self.assertIsNone(res)
+        self.assertTrue(app.logger.warning.called)
+        logged = ' '.join(str(a) for a in
+                          app.logger.warning.call_args[0])
+        self.assertIn('gone', logged)
+        self.assertIn('7', logged)
+
+    def test_server_free_text(self):
+        res = self._build(_server(passexec_cmd='echo x', passexec_name=None,
+                                  user_id=2))
+        self.assertIsNone(res)
+
+    def test_desktop(self):
+        res = self._build(_server(passexec_cmd='echo x'),
+                          server_mode=False)
+        self.assertIs(type(res), PasswordExec)
+        self.assertEqual(res.cmd, 'echo x')
+        self.assertIsNone(self._build(
+            _server(passexec_cmd=None, passexec_name='vault'),
+            server_mode=False))
+
+    def test_unauthenticated(self):
+        anon = MagicMock(is_authenticated=False)
+        self.assertIsNone(self._build(_server(), user=anon))
+
+    def test_no_commands(self):
+        res = self._build(
+            _server(),
+            row=MagicMock(passexec_name='vault', username='jane_db'),
+            cmds={})
+        self.assertIsNone(res)
+        self.shared_server.query.filter_by.assert_not_called()

web/pgadmin/utils/tests/test_passexec_server.py

@@ -0,0 +1,293 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL License
+#
+##########################################################################
+import os
+from unittest.mock import patch, MagicMock
+
+from pgadmin.utils.route import BaseTestGenerator
+from pgadmin.utils import passexec
+from pgadmin.utils.passexec import (
+    PasswordExec, ServerPasswordExec, get_server_passexec_commands,
+    check_server_passexec_config)
+
+
+class TestServerPassexecCommandsValidation(BaseTestGenerator):
+    """Only well-formed allowlist entries are returned."""
+
+    def runTest(self):
+        cfg = {
+            'vault': ['/bin/get-pass', '--ttl', '300'],
+            'tuple': ('/bin/x',),
+            '': ['/bin/empty-name'],
+            '  ': ['/bin/blank-name'],
+            '__none__': ['/bin/reserved'],
+            '__x': ['/bin/reserved-prefix'],
+            'empty': [],
+            'str': '/bin/a-string-not-a-list',
+            'mixed': ['/bin/x', 3],
+            5: ['/bin/int-name'],
+        }
+        with patch.object(passexec.config, 'SERVER_PASSEXEC_COMMANDS',
+                          cfg, create=True):
+            cmds = get_server_passexec_commands()
+            self.assertEqual(cmds, {
+                'vault': ['/bin/get-pass', '--ttl', '300'],
+                'tuple': ['/bin/x'],
+            })
+            # Returned lists are copies.
+            cmds['vault'].append('evil')
+            self.assertEqual(get_server_passexec_commands()['vault'],
+                             ['/bin/get-pass', '--ttl', '300'])
+            logger = MagicMock()
+            check_server_passexec_config(logger)
+            self.assertEqual(logger.error.call_count, 8)
+
+
+class TestServerPassexecCommandsNotADict(BaseTestGenerator):
+    """A non-dict setting yields no commands and one logged error."""
+
+    def runTest(self):
+        with patch.object(passexec.config, 'SERVER_PASSEXEC_COMMANDS',
+                          ['/bin/x'], create=True):
+            self.assertEqual(get_server_passexec_commands(), {})
+            logger = MagicMock()
+            check_server_passexec_config(logger)
+            self.assertEqual(logger.error.call_count, 1)
+
+
+class TestPassthroughNotAList(BaseTestGenerator):
+    """A passthrough setting that is not a list of names is logged."""
+
+    def runTest(self):
+        for value in ('VAULT_ADDR', ['VAULT_ADDR', 1]):
+            logger = MagicMock()
+            with patch.object(passexec.config, 'SERVER_PASSEXEC_COMMANDS',
+                              {}, create=True), \
+                    patch.object(passexec.config,
+                                 'SERVER_PASSEXEC_ENV_PASSTHROUGH', value,
+                                 create=True):
+                check_server_passexec_config(logger)
+            logger.error.assert_called_once()
+            self.assertIn('SERVER_PASSEXEC_ENV_PASSTHROUGH',
+                          logger.error.call_args[0][0])
+
+
+class TestDeprecatedFlagWarning(BaseTestGenerator):
+    def runTest(self):
+        logger = MagicMock()
+        with patch.object(passexec.config, 'SERVER_PASSEXEC_COMMANDS', {},
+                          create=True), \
+                patch.object(passexec.config,
+                             'ENABLE_SERVER_PASS_EXEC_CMD', True):
+            check_server_passexec_config(logger)
+        logger.warning.assert_called_once()
+        self.assertIn('SERVER_PASSEXEC_COMMANDS',
+                      logger.warning.call_args[0][0])
+
+
+class TestServerPasswordExecNoShell(BaseTestGenerator):
+    """Values travel in env vars verbatim; argv is never altered."""
+
+    def runTest(self):
+        host = 'h;touch /tmp/x'
+        user = '$(touch /tmp/y)`id`\nz'
+        pexec = ServerPasswordExec(
+            'vault', ['/bin/get-pass', '%USERNAME%'], host, 5432, user,
+            'postgres', 'jane.doe@example.com', 'internal')
+        fake_proc = MagicMock(stdout='secret\n')
+        with self.app.app_context(), \
+                patch('pgadmin.utils.passexec.subprocess.run',
+                      return_value=fake_proc) as mock_run, \
+                patch('pgadmin.utils.passexec.config.SERVER_MODE', True):
+            self.assertEqual(pexec.get(), 'secret')
+        args, kwargs = mock_run.call_args
+        self.assertEqual(args[0], ['/bin/get-pass', '%USERNAME%'])
+        self.assertFalse(kwargs.get('shell', False))
+        env = kwargs['env']
+        self.assertEqual(env['PGADMIN_PASSEXEC_HOST'], host)
+        self.assertEqual(env['PGADMIN_PASSEXEC_PORT'], '5432')
+        self.assertEqual(env['PGADMIN_PASSEXEC_USERNAME'], user)
+        self.assertEqual(env['PGADMIN_PASSEXEC_DATABASE'], 'postgres')
+        self.assertEqual(env['PGADMIN_PASSEXEC_PGADMIN_USER'],
+                         'jane.doe@example.com')
+        self.assertEqual(env['PGADMIN_PASSEXEC_AUTH_SOURCE'], 'internal')
+
+
+class TestServerPasswordExecRealProcess(BaseTestGenerator):
+    """End to end with a real child: metacharacters are inert."""
+
+    def runTest(self):
+        import os
+        import shutil
+        import sys
+        import tempfile
+        d = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, d, ignore_errors=True)
+        marker = os.path.join(d, 'pwned')
+        user = '$(touch {0})`touch {0}`;touch {0}'.format(marker)
+        argv = [sys.executable, '-c',
+                'import os;print(os.environ["PGADMIN_PASSEXEC_USERNAME"])']
+        pexec = ServerPasswordExec('t', argv, 'h', None, user, None,
+                                   'u', 'internal')
+        with self.app.app_context(), \
+                patch('pgadmin.utils.passexec.config.SERVER_MODE', True):
+            self.assertEqual(pexec.get(), user)
+        self.assertFalse(os.path.exists(marker))
+
+
+class TestServerPasswordExecUnsetValues(BaseTestGenerator):
+    def runTest(self):
+        pexec = ServerPasswordExec('t', ['/bin/x'], None, None, None, None,
+                                   None, None)
+        with self.app.app_context(), \
+                patch('pgadmin.utils.passexec.subprocess.run',
+                      return_value=MagicMock(stdout='p')) as mock_run:
+            pexec.get()
+        env = mock_run.call_args[1]['env']
+        for k in ('HOST', 'PORT', 'USERNAME', 'DATABASE', 'PGADMIN_USER',
+                  'AUTH_SOURCE'):
+            self.assertEqual(env['PGADMIN_PASSEXEC_' + k], '')
+
+
+class TestFreeTextRefusedInServerMode(BaseTestGenerator):
+    def runTest(self):
+        pexec = PasswordExec('echo x', 'h', 5432, 'u')
+        with self.app.app_context(), \
+                patch('pgadmin.utils.passexec.config.SERVER_MODE', True), \
+                patch('pgadmin.utils.passexec.config.'
+                      'ENABLE_SERVER_PASS_EXEC_CMD', True), \
+                patch('pgadmin.utils.passexec.subprocess.run') as mock_run:
+            with self.assertRaises(NotImplementedError):
+                pexec.get()
+        mock_run.assert_not_called()
+
+
+class TestPasswordExecLockPerInstance(BaseTestGenerator):
+    """A command running for one instance does not block another."""
+
+    def runTest(self):
+        a = ServerPasswordExec('a', ['/bin/a'], 'h', 5432, 'u', 'postgres',
+                               'jane.doe@example.com', 'internal')
+        b = ServerPasswordExec('b', ['/bin/b'], 'h', 5432, 'u', 'postgres',
+                               'john.doe@example.com', 'internal')
+        c = PasswordExec('echo x', 'h', 5432, 'u')
+        d = PasswordExec('echo y', 'h', 5432, 'u')
+        self.assertIsNot(a.lock, b.lock)
+        self.assertIsNot(c.lock, d.lock)
+        import threading
+        result = []
+        app = self.app
+
+        def fetch():
+            with app.app_context():
+                result.append(b.get())
+
+        with patch('pgadmin.utils.passexec.subprocess.run',
+                   return_value=MagicMock(stdout='p')):
+            # Holding a's lock, as a running command would, must not stop
+            # b fetching its password in another thread.
+            with a.lock:
+                t = threading.Thread(target=fetch, daemon=True)
+                t.start()
+                t.join(timeout=10)
+        self.assertFalse(t.is_alive())
+        self.assertEqual(result, ['p'])
+
+
+class TestServerPasswordExecReducedEnv(BaseTestGenerator):
+    """The child gets a reduced environment: secrets in pgAdmin's
+    environment stay out, PATH and configured passthrough names get in,
+    and the PGADMIN_PASSEXEC_* values cannot be overridden."""
+
+    def runTest(self):
+        import json
+        import sys
+        fake_env = {
+            'PATH': '/usr/bin:/bin',
+            'PGPASSWORD': 'other-users-password',
+            'PGADMIN_SETUP_PASSWORD': 'setup-password',
+            'VAULT_ADDR': 'https://vault.example.com',
+            'LC_ALL': 'C.UTF-8',
+            'PGADMIN_PASSEXEC_HOST': 'spoofed.example.com',
+        }
+        if os.name == 'nt':
+            # A Windows Python child may fail to start without SYSTEMROOT.
+            fake_env['SYSTEMROOT'] = os.environ.get('SYSTEMROOT', '')
+        argv = [sys.executable, '-c',
+                'import json, os; print(json.dumps(dict(os.environ)))']
+        with self.app.app_context(), \
+                patch.dict('os.environ', fake_env, clear=True), \
+                patch.object(passexec.config,
+                             'SERVER_PASSEXEC_ENV_PASSTHROUGH',
+                             ['VAULT_ADDR', 'NOT_SET_ANYWHERE'],
+                             create=True), \
+                patch('pgadmin.utils.passexec.config.SERVER_MODE', True):
+            pexec = ServerPasswordExec('t', argv, 'db.example.com', 5432,
+                                       'u', 'postgres',
+                                       'jane.doe@example.com', 'internal')
+            env = json.loads(pexec.get())
+        self.assertNotIn('PGPASSWORD', env)
+        self.assertNotIn('PGADMIN_SETUP_PASSWORD', env)
+        self.assertNotIn('NOT_SET_ANYWHERE', env)
+        self.assertEqual(env['PATH'], '/usr/bin:/bin')
+        self.assertEqual(env['VAULT_ADDR'], 'https://vault.example.com')
+        self.assertEqual(env['LC_ALL'], 'C.UTF-8')
+        self.assertEqual(env['PGADMIN_PASSEXEC_HOST'], 'db.example.com')
+
+
+class TestServerPasswordExecFailureHidesArgv(BaseTestGenerator):
+    """Failures name the command but never reveal its arguments, in
+    either the raised message or the log."""
+
+    scenarios = [
+        ('Non-zero exit', dict(kind='exit')),
+        ('Missing executable', dict(kind='missing')),
+        ('Timeout', dict(kind='timeout')),
+    ]
+
+    def runTest(self):
+        import os
+        import shutil
+        import sys
+        import tempfile
+        secret = 'S3CRET-ARG'
+        timeout = 60
+        if self.kind == 'exit':
+            argv = [sys.executable, '-c',
+                    'import sys; sys.stderr.write("boom"); sys.exit(3)',
+                    secret]
+            expect_log = 'status 3'
+        elif self.kind == 'missing':
+            d = tempfile.mkdtemp()
+            self.addCleanup(shutil.rmtree, d, ignore_errors=True)
+            argv = [os.path.join(d, 'no-such-command'), secret]
+            expect_log = 'could not be run'
+        else:
+            argv = [sys.executable, '-c', 'import time; time.sleep(30)',
+                    secret]
+            timeout = 1
+            expect_log = 'timed out'
+        pexec = ServerPasswordExec('vault-test', argv, 'h', 5432, 'u',
+                                   'postgres', 'jane.doe@example.com',
+                                   'internal', timeout=timeout)
+        with self.app.app_context(), \
+                patch('pgadmin.utils.passexec.config.SERVER_MODE', True), \
+                self.assertLogs('passexec', level='INFO') as logs:
+            with self.assertRaises(Exception) as ctx:
+                pexec.get()
+        exc = ctx.exception
+        self.assertIs(type(exc), Exception)
+        self.assertEqual(str(exc),
+                         "Password exec command 'vault-test' failed.")
+        self.assertTrue(exc.__suppress_context__)
+        output = '\n'.join(logs.output)
+        self.assertIn('vault-test', output)
+        self.assertIn(expect_log, output)
+        self.assertNotIn(secret, output)
+        if self.kind == 'exit':
+            self.assertIn('boom', output)

web/pgadmin/utils/tests/test_passexec_startup.py

@@ -0,0 +1,43 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL License
+#
+##########################################################################
+from unittest.mock import patch, MagicMock
+
+from pgadmin.utils.route import BaseTestGenerator
+from pgadmin.utils import passexec
+from pgadmin.utils.passexec import server_passexec_startup
+
+
+class TestServerPassexecStartup(BaseTestGenerator):
+    """The legacy conversion runs only for the server-mode web app, never
+    from the CLI (setup.py), which could be pointed at a desktop DB."""
+
+    scenarios = [
+        ('Server mode, web app', dict(server_mode=True, cli_mode=False,
+                                      check=True, convert=True)),
+        ('Server mode, CLI', dict(server_mode=True, cli_mode=True,
+                                  check=True, convert=False)),
+        ('Desktop mode, web app', dict(server_mode=False, cli_mode=False,
+                                       check=False, convert=False)),
+        ('Desktop mode, CLI', dict(server_mode=False, cli_mode=True,
+                                   check=False, convert=False)),
+    ]
+
+    def runTest(self):
+        app = MagicMock()
+        with patch.object(passexec.config, 'SERVER_MODE',
+                          self.server_mode), \
+                patch.object(passexec, 'check_server_passexec_config') \
+                as mock_check, \
+                patch.object(passexec, 'convert_legacy_server_passexec') \
+                as mock_convert:
+            server_passexec_startup(app, self.cli_mode)
+        self.assertEqual(mock_check.called, self.check)
+        self.assertEqual(mock_convert.called, self.convert)
+        if self.convert:
+            mock_convert.assert_called_once_with(app.logger)

web/regression/javascript/schema_ui_files/server.ui.spec.js

@@ -8,8 +8,10 @@
 //////////////////////////////////////////////////////////////
 
 
+import _ from 'lodash';
 import pgAdmin from 'sources/pgadmin';
-import ServerSchema from '../../../pgadmin/browser/server_groups/servers/static/js/server.ui';
+import current_user from 'pgadmin.user_management.current_user';
+import ServerSchema, { passexecForServerCopy } from '../../../pgadmin/browser/server_groups/servers/static/js/server.ui';
 import {genericBeforeEach, getCreateView, getEditView, getPropertiesView} from '../genericFunctions';
 
 describe('ServerSchema', ()=>{
@@ -84,4 +86,111 @@ describe('ServerSchema', ()=>{
     state.tunnel_keep_alive = 0;
     expect(schemaObj.validate(state, setError)).toBe(false);
   });
+
+  describe('password exec command', ()=>{
+    let origMode, origCommands, origUserId;
+    const field = (schema, id) => _.find(schema.fields, (f)=>f.id==id);
+    const setup = (serverMode, commands) => {
+      pgAdmin.server_mode = serverMode;
+      pgAdmin.server_passexec_commands = commands;
+      return createSchemaObject();
+    };
+
+    beforeEach(()=>{
+      origMode = pgAdmin.server_mode;
+      origCommands = pgAdmin.server_passexec_commands;
+      origUserId = current_user.id;
+    });
+
+    afterEach(()=>{
+      pgAdmin.server_mode = origMode;
+      pgAdmin.server_passexec_commands = origCommands;
+      current_user.id = origUserId;
+    });
+
+    it('offers None and the configured names to an owner', ()=>{
+      let schema = setup('True', ['vault']);
+      let f = field(schema, 'passexec_name');
+      expect(f.mode).toEqual(['properties', 'edit', 'create']);
+      expect(f.visible({})).toBe(true);
+      let type = f.type({id: 1, shared: false});
+      expect(type.type).toBe('select');
+      expect(type.options).toEqual([
+        {label: 'None', value: '__none__'},
+        {label: 'vault', value: 'vault'},
+      ]);
+      expect(field(schema, 'passexec_cmd').visible({})).toBe(false);
+    });
+
+    it('offers Inherit from owner to a non-owner of a shared server', ()=>{
+      let schema = setup('True', ['vault']);
+      current_user.id = 2;
+      schema.userId = 1;
+      let type = field(schema, 'passexec_name').type({id: 5, shared: true});
+      expect(type.options).toEqual([
+        {label: 'Inherit from owner', value: '__inherit__'},
+        {label: 'None', value: '__none__'},
+        {label: 'vault', value: 'vault'},
+      ]);
+    });
+
+    it('does not offer Inherit to the owner of a shared server', ()=>{
+      let schema = setup('True', ['vault']);
+      current_user.id = 1;
+      schema.userId = 1;
+      let type = field(schema, 'passexec_name').type({id: 5, shared: true});
+      expect(type.options.map((o)=>o.value)).toEqual(['__none__', 'vault']);
+    });
+
+    it('hides the select when no commands are configured', ()=>{
+      let schema = setup('True', []);
+      expect(field(schema, 'passexec_name').visible({})).toBe(false);
+      expect(field(schema, 'passexec_cmd').visible({})).toBe(false);
+      expect(field(schema, 'passexec_expiration').visible({})).toBe(false);
+    });
+
+    it('shows the expiration in server mode when commands are configured', ()=>{
+      let schema = setup('True', ['vault']);
+      expect(field(schema, 'passexec_expiration').visible({})).toBe(true);
+    });
+
+    it('shows the free text command only in desktop mode', ()=>{
+      let schema = setup('False', []);
+      expect(field(schema, 'passexec_cmd').visible({})).toBe(true);
+      expect(field(schema, 'passexec_name').visible({})).toBe(false);
+      expect(field(schema, 'passexec_expiration').visible({})).toBe(true);
+    });
+
+    it('disables the expiration according to the mode', ()=>{
+      let schema = setup('True', ['vault']);
+      let disabled = field(schema, 'passexec_expiration').disabled;
+      expect(disabled({passexec_name: '__none__'})).toBe(true);
+      expect(disabled({passexec_name: '__inherit__'})).toBe(true);
+      expect(disabled({passexec_name: 'vault'})).toBe(false);
+
+      schema = setup('False', []);
+      disabled = field(schema, 'passexec_expiration').disabled;
+      expect(disabled({passexec_cmd: ''})).toBe(true);
+      expect(disabled({passexec_cmd: 'echo x'})).toBe(false);
+    });
+
+    it('defaults to None in server mode', ()=>{
+      expect(setup('True', ['vault']).defaults.passexec_name).toBe('__none__');
+      expect(setup('False', []).defaults.passexec_name).toBeUndefined();
+    });
+
+    it('turns an inherited command into none when copying a shared server', ()=>{
+      let d = {name: 'x', passexec_name: '__inherit__', passexec_cmd: 'echo x'};
+      expect(passexecForServerCopy(d)).toBe(d);
+      expect(d.passexec_name).toBe('__none__');
+      expect(d).not.toHaveProperty('passexec_cmd');
+    });
+
+    it('keeps a named or none command when copying a shared server', ()=>{
+      expect(passexecForServerCopy({passexec_name: 'vault'}).passexec_name)
+        .toBe('vault');
+      expect(passexecForServerCopy({passexec_name: '__none__'}).passexec_name)
+        .toBe('__none__');
+    });
+  });
 });