fix: SharedServer feature parity columns and write guards (#9835)

Add passexec_cmd, passexec_expiration, kerberos_conn, tags, and
post_connection_sql to SharedServer so non-owners get their own
per-user values instead of inheriting the owner's.  Drop the unused
db_res column which was never overlaid or writable by non-owners.

Key changes:
- New Alembic migration (sharedserver_feature_parity) adds 5 columns,
  drops db_res, cleans up orphaned records.  All operations idempotent.
- Overlay copies new fields from SharedServer instead of suppressing
- _owner_only_fields guard blocks non-owners from setting passexec_cmd,
  passexec_expiration, db_res, db_res_type via API
- Non-owners can set post_connection_sql (runs under their own creds)
- update_tags and flag_modified use sharedserver for non-owners
- update() response returns sharedserver tags for non-owners
- ServerManager passexec suppression with config.SERVER_MODE guard
- UI: post_connection_sql editable for non-owners (readonly only when
  connected, not when shared)
- SCHEMA_VERSION bumped to 51
- Comprehensive unit tests for overlay, write guards, and tag deltas

Author: asheshv

web/migrations/versions/sharedserver_feature_parity_.py

@@ -0,0 +1,87 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""SharedServer feature parity columns, orphan cleanup, and db_res removal
+
+Adds passexec_cmd, passexec_expiration, kerberos_conn, tags, and
+post_connection_sql to the sharedserver table so non-owners get their
+own per-user values instead of inheriting the owner's.  Drops db_res
+which was never overlaid or writable by non-owners.  Also cleans up
+orphaned records whose parent server was deleted.
+
+Revision ID: sharedserver_feature_parity
+Revises: add_user_id_dbg_args
+Create Date: 2026-04-13
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = 'sharedserver_feature_parity'
+down_revision = 'add_user_id_dbg_args'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    if not inspector.has_table('sharedserver'):
+        return
+
+    # Clean up orphaned SharedServer records whose osid
+    # references a Server that no longer exists.
+    op.execute(
+        "DELETE FROM sharedserver WHERE osid NOT IN "
+        "(SELECT id FROM server)"
+    )
+
+    # Add missing columns (idempotent — guard against re-runs).
+    existing_cols = {
+        c['name'] for c in inspector.get_columns('sharedserver')
+    }
+    new_columns = [
+        ('passexec_cmd',
+         sa.Column('passexec_cmd', sa.Text(),
+                   nullable=True)),
+        ('passexec_expiration',
+         sa.Column('passexec_expiration', sa.Integer(),
+                   nullable=True)),
+        ('kerberos_conn',
+         sa.Column('kerberos_conn', sa.Boolean(),
+                   nullable=False,
+                   server_default=sa.false())),
+        ('tags',
+         sa.Column('tags', sa.JSON(), nullable=True)),
+        ('post_connection_sql',
+         sa.Column('post_connection_sql', sa.String(),
+                   nullable=True)),
+    ]
+    cols_to_add = [
+        col for name, col in new_columns
+        if name not in existing_cols
+    ]
+    if cols_to_add:
+        with op.batch_alter_table('sharedserver') as batch_op:
+            for col in cols_to_add:
+                batch_op.add_column(col)
+
+    # Drop db_res — database restrictions are an owner-level
+    # concept; the column was never overlaid or writable by
+    # non-owners and its presence invites accidental leakage.
+    if 'db_res' in existing_cols:
+        with op.batch_alter_table('sharedserver') as batch_op:
+            batch_op.drop_column('db_res')
+
+
+def downgrade():
+    # pgAdmin only upgrades, downgrade not implemented.
+    pass

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

@@ -172,16 +172,17 @@ def get_shared_server_properties(server, sharedserver):
         Return shared server properties.
 
         Overlays per-user SharedServer values onto the owner's Server
-        object.  Security-sensitive fields that are absent from the
-        SharedServer model (passexec_cmd, post_connection_sql) are
-        suppressed for non-owners.
+        object so each non-owner sees their own customizations.
 
         The server is expunged from the SQLAlchemy session before
         mutation so that the owner's record is never dirtied.
         :param server:
         :param sharedserver:
         :return: shared server (detached)
         """
+        if sharedserver is None:
+            return server
+
         # Detach from session so in-place mutations are never
         # flushed back to the owner's Server row.
         sess = object_session(server)
@@ -224,13 +225,11 @@ def get_shared_server_properties(server, sharedserver):
         server.server_owner = sharedserver.server_owner
         server.password = sharedserver.password
         server.prepare_threshold = sharedserver.prepare_threshold
-
-        # Suppress owner-only fields that are absent from SharedServer
-        # and dangerous when inherited (privilege escalation / code
-        # execution).
-        server.passexec_cmd = None
-        server.passexec_expiration = None
-        server.post_connection_sql = None
+        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
 
         return server
 
@@ -477,7 +476,12 @@ def create_shared_server(data, gid):
                 tunnel_prompt_password=0,
                 shared=True,
                 connection_params=safe_conn_params,
-                prepare_threshold=data.prepare_threshold
+                prepare_threshold=data.prepare_threshold,
+                passexec_cmd=None,
+                passexec_expiration=None,
+                kerberos_conn=False,
+                tags=None,
+                post_connection_sql=None
             )
             db.session.add(shared_server)
             db.session.commit()
@@ -904,7 +908,7 @@ def update(self, gid, sid):
 
         # Update connection parameter if any.
         self.update_connection_parameter(data, server, sharedserver)
-        self.update_tags(data, server)
+        self.update_tags(data, sharedserver or server)
 
         if 'connection_params' in data and \
             'hostaddr' in data['connection_params'] and \
@@ -937,7 +941,7 @@ def update(self, gid, sid):
 
         # tags is JSON type, sqlalchemy sometimes will not detect change
         if 'tags' in data:
-            flag_modified(server, 'tags')
+            flag_modified(sharedserver or server, 'tags')
 
         try:
             db.session.commit()
@@ -953,6 +957,10 @@ def update(self, gid, sid):
         # 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(
@@ -974,7 +982,7 @@ def update(self, gid, sid):
                 role=server.role,
                 is_password_saved=bool(server.save_password),
                 description=server.comment,
-                tags=server.tags
+                tags=(sharedserver or server).tags
             )
         )
 
@@ -998,8 +1006,20 @@ def _set_valid_attr_value(self, gid, data, config_param_map, server,
         if not crypt_key_present:
             raise CryptKeyMissing
 
+        # Fields that non-owners must never set on their
+        # SharedServer — they enable command/SQL execution
+        # or are owner-level concepts not on SharedServer.
+        _owner_only_fields = frozenset({
+            'passexec_cmd', 'passexec_expiration',
+            'db_res', 'db_res_type',
+        })
+
         for arg in config_param_map:
             if arg in data:
+                # Non-owners cannot set dangerous fields.
+                if _is_non_owner(server) and \
+                        arg in _owner_only_fields:
+                    continue
                 value = data[arg]
                 if arg == 'password':
                     value = encrypt(data[arg], crypt_key)
@@ -1159,14 +1179,8 @@ def properties(self, gid, sid):
             '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
-                if server.passexec_cmd and
-                not _is_non_owner(server) else None,
-            'passexec_expiration':
-                server.passexec_expiration
-                if server.passexec_expiration and
-                not _is_non_owner(server) else None,
+            'passexec_cmd': server.passexec_cmd,
+            'passexec_expiration': server.passexec_expiration,
             'service': server.service if server.service else None,
             'use_ssh_tunnel': use_ssh_tunnel,
             'tunnel_host': tunnel_host,
@@ -1186,8 +1200,7 @@ def properties(self, gid, sid):
             'connection_string': display_connection_str,
             'prepare_threshold': server.prepare_threshold,
             'tags': tags,
-            'post_connection_sql': server.post_connection_sql
-                if not _is_non_owner(server) else None,
+            'post_connection_sql': server.post_connection_sql,
         }
 
         return ajax_response(response)
@@ -1605,6 +1618,12 @@ def connect(self, gid, sid, is_qt=False, server=None):
         # 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

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

@@ -587,7 +587,7 @@ export default class ServerSchema extends BaseUISchema {
         group: gettext('Post Connection SQL'),
         mode: ['properties', 'edit', 'create'],
         type: 'sql', isFullTab: true,
-        readonly: obj.isConnectedOrShared,
+        readonly: obj.isConnected,
         helpMessage: gettext('Any query specified in the control below will be executed with autocommit mode enabled for each connection to any database on this server.'),
       },
       {