Restrict sharing a server to administrators in the server API and the servers.json import. #10651

The server dialog only shows the Shared switch to administrators, and
the servers.json import only honoured it for them, but the create and
update endpoints stored whatever 'shared' and 'shared_username' they
were sent. Any user able to register a server could therefore share one
with every other user of a server mode installation. The import had a
similar gap: validate_json_data() removed a non-administrator's "Shared"
entries only after its loop, so an early validation error left them in
the data, and the web path ignored that error, so selecting a shared
entry in a file that also held an invalid one created a shared server.

A non-administrator's request that would share a server is now refused
with a 403. A 'shared' value that changes nothing is still accepted, so
editing a private server is unaffected, and 'shared_username' is dropped
rather than refused because the dialog sends it whenever the username
changes. The import prunes a non-administrator's shared entries before
validating and skips them again for each imported entry, only a JSON
true counts as shared, and validation errors are reported on the web
path too. Copy Server no longer sends shared for a non-administrator
owner.

The existing servers are cleared for "replace" only after the file has
validated, so a refused file no longer deletes them. Servers shared by
non-administrators before the upgrade stay shared until the owner
un-shares them or an administrator changes them; pgAdmin warns about
them at server-mode startup, and the documentation describes this.

Compatibility: a servers.json import in server mode that fails
validation is now refused as a whole on the web path, as it already was
in setup.py.

Co-authored-by: Dave Page <dpage@pgadmin.org>

docs/en_US/server_dialog.rst

@@ -33,6 +33,18 @@ Use the fields in the *General* tab to identify the server:
 * If the *Shared?* switch is moved to *Yes* then that server can be
   shared with all the other users. This option is available only to admin users.
   For more information on users see :ref:`User Management Dialog <user_management>`.
+  Sharing a server, whether in this dialog, through the API or by importing a
+  servers.json file, is restricted to users with the *Administrator* role.
+  Servers that were shared by non-administrator users before this restriction
+  was introduced remain shared until they are un-shared, so administrators
+  should audit them: look for servers with the shared flag set in the
+  configuration database whose owner does not have the *Administrator* role.
+  pgAdmin logs a warning for each such server when it starts in server mode.
+  Such a server can be un-shared by its owner through the server API (a PUT
+  request on the server with ``"shared": false``), or by an administrator
+  changing the ``shared`` column of the server record in the pgAdmin
+  configuration database. The *Shared?* switch in this dialog is shown only
+  to administrators, and only the owner of a server can change it.
   Users can access the shared servers with some restrictions - the following
   operations on shared servers are not permitted:
 

web/pgadmin/__init__.py

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

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

@@ -74,6 +74,26 @@ def _is_non_owner(server):
     return server.shared and server.user_id != current_user.id
 
 
+def prepare_sharing_data(data, server=None):
+    """Refuse a create or update payload sent to the servers API that would
+    share a server unless it comes from an administrator, matching the
+    server dialog. The servers.json import enforces the same rule
+    separately, in validate_json_data and load_database_servers.
+    Returns (data, errormsg); data is a plain dict copy.
+
+    A shared value that changes nothing, such as an unchanged false or a
+    non-owner's true, is let through. shared_username only matters for a
+    shared server, and the dialog can send it for anyone because it follows
+    the username field, so it is dropped rather than refused."""
+    data = dict(data)
+    if not config.SERVER_MODE or current_user.has_role('Administrator'):
+        return data, None
+    if data.get('shared') and not (server is not None and server.shared):
+        return data, gettext('Only an administrator can share a server.')
+    data.pop('shared_username', None)
+    return data, None
+
+
 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,
@@ -1003,6 +1023,10 @@ class ServerNode(PGChildNodeView):
             request.data
         )
 
+        data, sharing_err = prepare_sharing_data(data, server)
+        if sharing_err:
+            return forbidden(errmsg=sharing_err)
+
         data, passexec_err = prepare_passexec_data(
             data, _is_non_owner(server))
         if passexec_err:
@@ -1370,6 +1394,10 @@ class ServerNode(PGChildNodeView):
             if data[item] == '':
                 data[item] = None
 
+        data, sharing_err = prepare_sharing_data(data)
+        if sharing_err:
+            return forbidden(errmsg=sharing_err)
+
         data, passexec_err = prepare_passexec_data(data, False)
         if passexec_err:
             return make_json_response(

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

@@ -8,7 +8,7 @@
 //////////////////////////////////////////////////////////////
 
 import { getNodeListById } from '../../../../static/js/node_ajax';
-import ServerSchema, { passexecForServerCopy } from './server.ui';
+import ServerSchema, { passexecForServerCopy, sharingForServerCopy } from './server.ui';
 import { showServerPassword, showChangeServerPassword, showNamedRestorePoint } from '../../../../../static/js/Dialogs/index';
 import _ from 'lodash';
 import getApiInstance, { parseApiError } from '../../../../../static/js/api_instance';
@@ -66,6 +66,9 @@ define('pgadmin.node.server', [
           d.shared_username = null;
           passexecForServerCopy(d);
         }
+        // Only an administrator can share a server, so the copy made by
+        // anyone else is not shared.
+        sharingForServerCopy(d, current_user);
         return d;
       },
       Init: function() {

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

@@ -58,6 +58,16 @@ export function passexecForServerCopy(d) {
   return d;
 }
 
+// Only an administrator can share a server, so a copy made by anyone else
+// is not shared, and does not carry the shared user name.
+export function sharingForServerCopy(d, user) {
+  if (d?.shared && !user?.is_admin) {
+    d.shared = false;
+    d.shared_username = null;
+  }
+  return d;
+}
+
 export function getConnectionParameters() {
   let conParams = [{
     'value': 'hostaddr', 'label': gettext('Host address'), 'vartype': 'string'

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

@@ -0,0 +1,191 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""Only an administrator may share a server through the API, as through
+the server dialog and the servers.json import."""
+
+import json
+import config
+from pgadmin.model import db, Role, Server, User
+from pgadmin.utils.route import BaseTestGenerator
+from regression.python_test_utils import test_utils as utils
+from .utils import restore_server_fixture_on_cleanup
+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']
+
+
+def _server_count(app):
+    with app.app_context():
+        return Server.query.count()
+
+
+def _grant_user_role(test_case, email):
+    """Give the test user the stock 'User' role, as a newly created user
+    would have, so that it may register servers; the test user is created
+    without any role."""
+    app = test_case.app
+    with app.app_context():
+        user = User.query.filter_by(email=email).first()
+        role = Role.query.filter_by(name='User').first()
+        if role in user.roles:
+            return
+        user.roles.append(role)
+        db.session.commit()
+
+    def revoke():
+        with app.app_context():
+            user = User.query.filter_by(email=email).first()
+            role = Role.query.filter_by(name='User').first()
+            if role in user.roles:
+                user.roles.remove(role)
+                db.session.commit()
+
+    test_case.addCleanup(revoke)
+
+
+def _server_row(app, sid):
+    """The fields of the server row these tests check."""
+    with app.app_context():
+        row = Server.query.filter_by(id=sid).first()
+        return dict(shared=bool(row.shared),
+                    shared_username=row.shared_username,
+                    comment=row.comment)
+
+
+class NonAdminServerSharingTestCase(BaseTestGenerator):
+    """A non-administrator cannot share a server, by creating or updating
+    it, but can still create and edit their own private servers."""
+
+    scenarios = [
+        ('Non-admin cannot create a shared server',
+         dict(create={'shared': True}, expected_create=403)),
+        ('Non-admin can create a server with shared false',
+         dict(create={'shared': False}, expected_create=200,
+              expected_shared=False)),
+        ('Non-admin shared_username is ignored on create',
+         dict(create={'shared': False, 'shared_username': 'someone'},
+              expected_create=200, expected_shared=False,
+              expected_shared_username=None)),
+        ('Non-admin cannot share their own server by update',
+         dict(create={}, expected_create=200,
+              update={'shared': True}, expected_update=403,
+              expected_shared=False)),
+        ('Non-admin can edit their own server with an unchanged shared',
+         dict(create={}, expected_create=200,
+              update={'shared': False, 'comment': 'edited'},
+              expected_update=200, expected_shared=False,
+              expected_comment='edited')),
+        ('Non-admin can edit their own server without shared',
+         dict(create={}, expected_create=200,
+              update={'comment': 'edited'},
+              expected_update=200, expected_shared=False,
+              expected_comment='edited')),
+        ('Non-admin shared_username is ignored on update',
+         dict(create={}, expected_create=200,
+              update={'shared_username': 'someone', 'comment': 'edited'},
+              expected_update=200, expected_shared=False,
+              expected_shared_username=None, expected_comment='edited')),
+    ]
+
+    update = None
+    expected_update = None
+    expected_shared = None
+    expected_shared_username = 'unchecked'
+    expected_comment = None
+
+    def setUp(self):
+        self.server_id = None
+        if not config.SERVER_MODE:
+            self.skipTest('Sharing only applies to server mode.')
+        restore_server_fixture_on_cleanup(self)
+
+    @create_user_wise_test_client(test_user_details)
+    def runTest(self):
+        _grant_user_role(self, test_user_details['login_username'])
+        server = dict(self.server)
+        server.pop('shared', None)
+        server.pop('shared_username', None)
+        server.update(self.create)
+
+        servers_before = _server_count(self.app)
+        response = self.tester.post(
+            '/browser/server/obj/{0}/'.format(utils.SERVER_GROUP),
+            data=json.dumps(server),
+            content_type='html/json'
+        )
+        self.assertEqual(response.status_code, self.expected_create,
+                         response.data)
+        if self.expected_create != 200:
+            self.assertEqual(_server_count(self.app), servers_before)
+            return
+
+        self.server_id = json.loads(
+            response.data.decode('utf-8'))['node']['_id']
+        try:
+            if self.update is not None:
+                data = dict(self.update, id=self.server_id)
+                response = self.tester.put(
+                    '/browser/server/obj/{0}/{1}'.format(
+                        utils.SERVER_GROUP, self.server_id),
+                    data=json.dumps(data),
+                    content_type='html/json'
+                )
+                self.assertEqual(response.status_code,
+                                 self.expected_update, response.data)
+
+            row = _server_row(self.app, self.server_id)
+            self.assertEqual(row['shared'], self.expected_shared)
+            if self.expected_shared_username != 'unchecked':
+                self.assertEqual(row['shared_username'],
+                                 self.expected_shared_username)
+            if self.expected_comment is not None:
+                self.assertEqual(row['comment'], self.expected_comment)
+        finally:
+            utils.delete_server_with_api(self.tester, self.server_id)
+            self.server_id = None
+
+
+class AdminServerSharingTestCase(BaseTestGenerator):
+    """An administrator can still share a server, and name its user."""
+
+    scenarios = [
+        ('Admin can create a shared server', dict()),
+    ]
+
+    def setUp(self):
+        self.server_id = None
+        if not config.SERVER_MODE:
+            self.skipTest('Sharing only applies to server mode.')
+        restore_server_fixture_on_cleanup(self)
+
+    def runTest(self):
+        server = dict(self.server, shared=True, shared_username='someone')
+        response = self.tester.post(
+            '/browser/server/obj/{0}/'.format(utils.SERVER_GROUP),
+            data=json.dumps(server),
+            content_type='html/json'
+        )
+        self.assertEqual(response.status_code, 200, response.data)
+        self.server_id = json.loads(
+            response.data.decode('utf-8'))['node']['_id']
+
+        row = _server_row(self.app, self.server_id)
+        self.assertTrue(row['shared'])
+        self.assertEqual(row['shared_username'], 'someone')
+
+    def tearDown(self):
+        if self.server_id is not None:
+            utils.delete_server_with_api(
+                self.__class__.tester, self.server_id)

web/pgadmin/tools/import_export_servers/__init__.py

@@ -25,7 +25,7 @@ from pgadmin.utils.constants import MIMETYPE_APP_JS
 from pgadmin.utils.ajax import make_json_response, internal_server_error, \
     unauthorized
 from pgadmin.model import ServerGroup, Server
-from pgadmin.utils import clear_database_servers, dump_database_servers,\
+from pgadmin.utils import dump_database_servers, \
     load_database_servers, validate_json_data, \
     filename_with_file_manager_path, get_file_manager_storage_dir
 from urllib.parse import unquote
@@ -195,11 +195,11 @@ def save():
         status, errmsg = \
             dump_database_servers(data['filename'], data['selected_sever_ids'])
     elif data['type'] == 'import':
-        # Clear all the existing servers
-        if 'replace_servers' in data and data['replace_servers']:
-            clear_database_servers()
-        status, errmsg = \
-            load_database_servers(data['filename'], data['selected_sever_ids'])
+        # The existing servers are cleared only once the file has been
+        # validated.
+        status, errmsg = load_database_servers(
+            data['filename'], data['selected_sever_ids'],
+            replace=bool(data.get('replace_servers')))
 
     if not status:
         return internal_server_error(errmsg)

web/pgadmin/tools/import_export_servers/tests/test_import_shared_server_non_admin.py

@@ -0,0 +1,128 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""A non-administrator cannot create a shared server through the
+servers.json import route, as through the server API and dialog."""
+
+import json
+import os
+import shutil
+import tempfile
+from unittest.mock import patch
+
+import config
+from pgadmin.model import db, Role, Server, ServerGroup, User
+from pgadmin.utils.route import BaseTestGenerator
+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']
+
+GROUP = 'Import shared server non-admin test group'
+HOST = 'evil.example'
+
+
+def _grant_user_role(test_case, email):
+    """Give the test user the stock 'User' role, which carries the import
+    and export permission; the test user is created without any role."""
+    app = test_case.app
+    with app.app_context():
+        user = User.query.filter_by(email=email).first()
+        role = Role.query.filter_by(name='User').first()
+        if role in user.roles:
+            return
+        user.roles.append(role)
+        db.session.commit()
+
+    def revoke():
+        with app.app_context():
+            user = User.query.filter_by(email=email).first()
+            role = Role.query.filter_by(name='User').first()
+            if role in user.roles:
+                user.roles.remove(role)
+                db.session.commit()
+
+    test_case.addCleanup(revoke)
+
+
+def _evil_count(app):
+    with app.app_context():
+        return Server.query.filter_by(host=HOST).count()
+
+
+class NonAdminImportSharedServerTestCase(BaseTestGenerator):
+    """The import route refuses a mixed file, and imports nothing from a
+    valid file whose only selected server is shared."""
+
+    scenarios = [
+        ('Shared server next to an invalid one is refused',
+         dict(invalid_entry=True, expected_status=500)),
+        ('Selected shared server is skipped from a valid file',
+         dict(invalid_entry=False, expected_status=200)),
+    ]
+
+    def setUp(self):
+        if not config.SERVER_MODE:
+            self.skipTest('Sharing only applies to server mode.')
+        self.tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+        self.addCleanup(self._cleanup)
+
+    def _cleanup(self):
+        with self.app.app_context():
+            Server.query.filter_by(host=HOST).delete()
+            for group in ServerGroup.query.filter_by(name=GROUP).all():
+                Server.query.filter_by(servergroup_id=group.id).delete()
+                db.session.delete(group)
+            db.session.commit()
+
+    @create_user_wise_test_client(test_user_details)
+    def runTest(self):
+        _grant_user_role(self, test_user_details['login_username'])
+        servers = {
+            '1': {'Group': GROUP, 'Name': 'evil', 'Shared': True,
+                  'Host': HOST, 'Port': 5432, 'SharedUsername': 'postgres',
+                  'MaintenanceDB': 'postgres'},
+        }
+        if self.invalid_entry:
+            servers['2'] = {'Group': GROUP, 'Name': 'no host',
+                            'Port': 5432, 'Username': 'postgres',
+                            'MaintenanceDB': 'postgres'}
+        else:
+            servers['2'] = {'Group': GROUP, 'Name': 'fine',
+                            'Host': 'fine.example', 'Port': 5432,
+                            'Username': 'postgres',
+                            'MaintenanceDB': 'postgres'}
+        path = os.path.join(self.tmp, 'servers.json')
+        with open(path, 'w') as f:
+            json.dump({'Servers': servers}, f)
+
+        # The import opens the file beneath the storage directory, so make
+        # the scratch directory the storage directory.
+        with patch('pgadmin.utils.filename_with_file_manager_path',
+                   return_value=path), \
+                patch('pgadmin.utils.get_file_manager_storage_dir',
+                      return_value=self.tmp):
+            response = self.tester.post(
+                '/import_export_servers/save',
+                data=json.dumps({'type': 'import',
+                                 'filename': 'servers.json',
+                                 'selected_sever_ids': ['1']}),
+                content_type='html/json')
+
+        self.assertEqual(response.status_code, self.expected_status,
+                         response.data)
+        self.assertEqual(_evil_count(self.app), 0)
+        with self.app.app_context():
+            self.assertEqual(
+                ServerGroup.query.filter_by(name=GROUP).count(), 0)

web/pgadmin/utils/__init__.py

@@ -983,22 +983,26 @@ def validate_json_data(data, is_admin):
     :param is_admin:
     :return: error message if any
     """
-    skip_servers = []
     # Loop through the servers...
     if "Servers" not in data:
         return gettext("'Servers' attribute not found in the specified file.")
 
+    # Shared servers can only be imported by administrators. Prune them
+    # before validating, so that they are dropped even when the
+    # validation of the remaining servers fails.
+    if not is_admin:
+        skip_servers = [server for server, obj in data["Servers"].items()
+                        if obj.get("Shared") is True]
+        for server in skip_servers:
+            print("Won't import the server '%s' as it is shared " %
+                  data["Servers"][server].get("Name", server))
+            del data["Servers"][server]
+
     for server in data["Servers"]:
         obj = data["Servers"][server]
 
-        is_shared = obj.get("Shared", None)
-
-        # Check if server is shared.Won't import if user is non-admin
-        if is_shared and not is_admin:
-            print("Won't import the server '%s' as it is shared " %
-                  obj["Name"])
-            skip_servers.append(server)
-            continue
+        # Only the JSON value true means shared, anything else is not.
+        is_shared = obj.get("Shared") is True
 
         def check_attrib(attrib):
             if attrib not in obj:
@@ -1053,11 +1057,47 @@ def validate_json_data(data, is_admin):
             return gettext("'Host' or 'Service' attribute not "
                            "found for server '%s'" % server)
 
-    for server in skip_servers:
-        del data["Servers"][server]
     return None
 
 
+def check_non_admin_shared_servers(logger, cli_mode=False):
+    """Server mode only: log a warning for each server that is shared by a
+    user who is not an administrator. Only administrators can share a
+    server now, but servers shared before that restriction stay shared
+    until they are un-shared. Read-only, and it never raises, so it cannot
+    block the startup. Credentials and host details are not logged.
+    Returns the number of servers found."""
+    if not config.SERVER_MODE or cli_mode:
+        return 0
+    try:
+        rows = db.session.query(Server.id, Server.name, Server.user_id) \
+            .filter(Server.shared.is_(True)).all()
+        found = 0
+        owner_is_admin = {}
+        for server_id, name, user_id in rows:
+            if user_id not in owner_is_admin:
+                owner = db.session.get(User, user_id)
+                owner_is_admin[user_id] = owner is not None and \
+                    owner.has_role('Administrator')
+            if owner_is_admin[user_id]:
+                continue
+            found += 1
+            logger.warning(
+                "Server %s ('%s') is shared by user %s, who is not an "
+                "administrator. The owner can un-share it with a PUT of "
+                "\"shared\": false through the API, or an administrator "
+                "can change its 'shared' column in the configuration "
+                "database.", server_id, name, user_id)
+        return found
+    except Exception as e:
+        try:
+            logger.warning("Unable to check for servers shared by "
+                           "non-administrators: %s", e)
+        except Exception:
+            pass
+        return 0
+
+
 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
@@ -1081,8 +1121,11 @@ def _load_server_passexec(new_server, obj):
 
 def load_database_servers(input_file, selected_servers,
                           load_user=current_user, from_setup=False,
-                          auth_source=INTERNAL):
+                          auth_source=INTERNAL, replace=False):
     """Load server groups and servers.
+
+    :param replace: clear the user's existing servers once the file has
+        been validated, and just before the servers are loaded.
     """
     user = _does_user_exist(load_user, from_setup, auth_source)
     if user is None:
@@ -1149,15 +1192,32 @@ def load_database_servers(input_file, selected_servers,
     groups = ServerGroup.query.filter_by(user_id=user_id)
 
     # Validate server data
-    error_msg = validate_json_data(data, user.has_role("Administrator"))
-    if error_msg is not None and from_setup:
-        print(ADD_SERVERS_MSG % (groups_added, servers_added))
+    is_admin = user.has_role("Administrator")
+    error_msg = validate_json_data(data, is_admin)
+    if error_msg is not None:
+        if from_setup:
+            print(ADD_SERVERS_MSG % (groups_added, servers_added))
         return _handle_error(error_msg, from_setup)
 
+    if replace:
+        # Clear the existing servers only once the file is known to be
+        # valid, so that a refused file does not lose the user's servers.
+        result = clear_database_servers(load_user, from_setup, auth_source)
+        if result is not None and result is not True:
+            return result
+
     for server in data["Servers"]:
         if selected_servers is None or str(server) in selected_servers:
             obj = data["Servers"][server]
 
+            # Defence in depth: only administrators can import shared
+            # servers. Checked before the group is created, so that no
+            # empty group is left behind.
+            if obj.get("Shared") is True and not is_admin:
+                print("Won't import the server '%s' as it is shared " %
+                      obj.get("Name", server))
+                continue
+
             # Get the group. Create if necessary
             group_id = next(
                 (g.id for g in groups if g.name == obj["Group"]), -1)
@@ -1181,7 +1241,8 @@ def load_database_servers(input_file, selected_servers,
                 groups_added = groups_added + 1
                 groups = ServerGroup.query.filter_by(user_id=user_id)
 
-            is_shared = obj.get("Shared", None)
+            # Only the JSON value true means shared, as in validation.
+            is_shared = obj.get("Shared") is True
             username = obj.get("Username", None)
             shared_username = obj.get("SharedUsername", None)
             if is_shared and username is None:
@@ -1266,9 +1327,10 @@ def load_database_servers(input_file, selected_servers,
             new_server.tunnel_keep_alive = \
                 obj.get("TunnelKeepAlive", None)
 
-            new_server.shared = is_shared
+            new_server.shared = is_shared and is_admin
 
-            new_server.shared_username = shared_username
+            new_server.shared_username = \
+                shared_username if new_server.shared else None
 
             new_server.kerberos_conn = obj.get("KerberosAuthentication", None)
 

web/pgadmin/utils/tests/test_check_non_admin_shared_servers.py

@@ -0,0 +1,96 @@
+##########################################################################
+#
+# 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 check_non_admin_shared_servers
+
+GROUP = 'Shared server audit test group'
+
+
+class TestCheckNonAdminSharedServers(BaseTestGenerator):
+    """Servers shared by non-administrators are reported at startup, in
+    server mode only, and never raise."""
+
+    scenarios = [
+        ('Server mode', dict(server_mode=True, cli_mode=False)),
+        ('Server mode, CLI', dict(server_mode=True, cli_mode=True)),
+        ('Desktop mode', dict(server_mode=False, cli_mode=False)),
+    ]
+
+    def setUp(self):
+        self.ctx = self.app.test_request_context()
+        self.ctx.push()
+        # Cleanups run last in, first out: the context pops last.
+        self.addCleanup(self.ctx.pop)
+        from pgadmin.model import User
+        self.user = User.query.first()
+        self.addCleanup(self._cleanup)
+
+    def _cleanup(self):
+        from pgadmin.model import db, Server, ServerGroup
+        db.session.rollback()
+        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(self, group, name, shared):
+        from pgadmin.model import db, Server
+        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',
+                        shared=shared)
+        db.session.add(server)
+        db.session.commit()
+        return server
+
+    def runTest(self):
+        from pgadmin.model import db, ServerGroup
+        group = ServerGroup(user_id=self.user.id, name=GROUP)
+        db.session.add(group)
+        db.session.flush()
+        shared = self._add(group, 'shared one', True)
+        self._add(group, 'private one', False)
+
+        logger = MagicMock()
+        for is_admin in (False, True):
+            logger.reset_mock()
+            with patch('pgadmin.utils.config.SERVER_MODE',
+                       self.server_mode), \
+                    patch.object(type(self.user), 'has_role',
+                                 return_value=is_admin):
+                found = check_non_admin_shared_servers(
+                    logger, self.cli_mode)
+
+            if self.server_mode and not self.cli_mode and not is_admin:
+                # Other shared servers of non-admins may exist in the
+                # test database, so look at ours.
+                self.assertGreaterEqual(found, 1)
+                texts = [c.args[0] % c.args[1:]
+                         for c in logger.warning.call_args_list]
+                mine = [t for t in texts
+                        if t.startswith('Server %s (' % shared.id)]
+                self.assertEqual(len(mine), 1)
+                self.assertIn("'shared one'", mine[0])
+                self.assertIn('user %s' % self.user.id, mine[0])
+                self.assertNotIn('private one', ' '.join(texts))
+                self.assertNotIn('db.example.com', ' '.join(texts))
+            else:
+                self.assertEqual(found, 0)
+                logger.warning.assert_not_called()
+
+        # Failures never propagate.
+        logger.reset_mock()
+        with patch('pgadmin.utils.config.SERVER_MODE', True), \
+                patch('pgadmin.utils.db.session.query',
+                      side_effect=RuntimeError('boom')):
+            self.assertEqual(check_non_admin_shared_servers(logger), 0)

web/pgadmin/utils/tests/test_shared_server_import.py

@@ -0,0 +1,295 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL License
+#
+##########################################################################
+"""Only an administrator can create a shared server through the
+servers.json import (load_database_servers)."""
+
+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
+
+GROUP = 'Shared server import test group'
+SHARED_GROUP = 'Shared server import test shared group'
+GROUPS = (GROUP, SHARED_GROUP)
+
+
+class _SharedImportMixin:
+    """Shared set-up: a scratch directory, and clean test groups."""
+
+    is_admin = False
+
+    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()
+        patcher = patch.object(type(self.user), 'has_role',
+                               return_value=self.is_admin)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    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
+        for name in GROUPS:
+            group = ServerGroup.query.filter_by(
+                user_id=self.user.id, name=name).first()
+            if group:
+                Server.query.filter_by(servergroup_id=group.id).delete()
+                db.session.delete(group)
+        db.session.commit()
+
+    def _entry(self, group=GROUP, **kwargs):
+        return dict({'Group': group, 'Host': 'db.example.com',
+                     'Port': 5432, 'Username': 'test_user',
+                     'MaintenanceDB': 'postgres'}, **kwargs)
+
+    def _write(self, servers):
+        with open(self.path, 'w') as f:
+            json.dump({'Servers': servers}, f)
+
+    def _load(self, selected, from_setup=False, **kwargs):
+        from pgadmin.utils import load_database_servers
+        out = io.StringIO()
+        # The web path opens the file beneath the storage directory, so
+        # make the scratch directory the storage directory.
+        with redirect_stdout(out), \
+                patch('pgadmin.utils.filename_with_file_manager_path',
+                      return_value=self.path), \
+                patch('pgadmin.utils.get_file_manager_storage_dir',
+                      return_value=self.tmp):
+            result = load_database_servers(
+                self.path, selected, self.user, from_setup=from_setup,
+                **kwargs)
+        return result, out.getvalue()
+
+    def _servers(self, group=None):
+        from pgadmin.model import Server, ServerGroup
+        groups = ServerGroup.query.filter(
+            ServerGroup.user_id == self.user.id,
+            ServerGroup.name.in_(GROUPS if group is None else [group]))
+        return Server.query.filter(
+            Server.servergroup_id.in_([g.id for g in groups])).all()
+
+    def _group_exists(self, name):
+        from pgadmin.model import ServerGroup
+        return ServerGroup.query.filter_by(
+            user_id=self.user.id, name=name).count() > 0
+
+
+class TestNonAdminMixedFileRefused(_SharedImportMixin, BaseTestGenerator):
+    """A file with a shared server and an invalid one is refused, and the
+    shared server is not created."""
+
+    def runTest(self):
+        self._write({
+            '1': self._entry(group=SHARED_GROUP, Name='evil', Shared=True,
+                             SharedUsername='postgres'),
+            '2': {'Group': GROUP, 'Name': 'no host', 'Port': 5432,
+                  'Username': 'test_user', 'MaintenanceDB': 'postgres'},
+        })
+        (ok, msg), _ = self._load(['1'])
+
+        self.assertFalse(ok)
+        self.assertIn("'Host' or 'Service' attribute not found", msg)
+        self.assertEqual(self._servers(), [])
+        self.assertFalse(self._group_exists(SHARED_GROUP))
+
+
+class TestNonAdminSharedEntrySkipped(_SharedImportMixin, BaseTestGenerator):
+    """Only the non-shared server is imported, and no empty group is left
+    for the shared one."""
+
+    def runTest(self):
+        self._write({
+            '1': self._entry(group=SHARED_GROUP, Name='shared one',
+                             Shared=True, SharedUsername='postgres'),
+            '2': self._entry(Name='private one'),
+        })
+        (ok, _), out = self._load(['1', '2'])
+
+        self.assertTrue(ok)
+        self.assertIn("Won't import the server 'shared one'", out)
+        servers = self._servers()
+        self.assertEqual([s.name for s in servers], ['private one'])
+        self.assertFalse(servers[0].shared)
+        self.assertFalse(self._group_exists(SHARED_GROUP))
+
+
+class TestNonAdminSharedEntryNotValidated(
+        _SharedImportMixin, BaseTestGenerator):
+    """The loader refuses a shared entry itself, even if validation let
+    it through."""
+
+    def runTest(self):
+        self._write({
+            '1': self._entry(group=SHARED_GROUP, Name='shared one',
+                             Shared=True, SharedUsername='postgres'),
+        })
+        with patch('pgadmin.utils.validate_json_data', return_value=None):
+            (ok, _), out = self._load(['1'])
+
+        self.assertTrue(ok)
+        self.assertIn("Won't import the server 'shared one'", out)
+        self.assertEqual(self._servers(), [])
+        self.assertFalse(self._group_exists(SHARED_GROUP))
+
+
+class TestAdminImportsSharedServer(_SharedImportMixin, BaseTestGenerator):
+    """An administrator can import a shared server."""
+
+    is_admin = True
+
+    def runTest(self):
+        self._write({
+            '1': self._entry(Name='shared one', Shared=True,
+                             SharedUsername='postgres', Username=None),
+            '2': self._entry(Name='private one', SharedUsername='ignored'),
+        })
+        (ok, _), _ = self._load(['1', '2'])
+
+        self.assertTrue(ok)
+        servers = {s.name: s for s in self._servers()}
+        self.assertTrue(servers['shared one'].shared)
+        self.assertEqual(servers['shared one'].shared_username, 'postgres')
+        self.assertEqual(servers['shared one'].username, 'postgres')
+        self.assertFalse(servers['private one'].shared)
+        self.assertIsNone(servers['private one'].shared_username)
+
+
+class TestSetupNonAdmin(_SharedImportMixin, BaseTestGenerator):
+    """The setup.py path skips shared servers for a non-admin and exits on
+    an invalid file."""
+
+    def runTest(self):
+        self._write({
+            '1': self._entry(group=SHARED_GROUP, Name='shared one',
+                             Shared=True, SharedUsername='postgres'),
+            '2': self._entry(Name='private one'),
+        })
+        (ok, _), out = self._load(None, from_setup=True)
+        self.assertTrue(ok)
+        self.assertIn("Won't import the server 'shared one'", out)
+        self.assertEqual([s.name for s in self._servers()], ['private one'])
+
+        self._write({
+            '1': self._entry(group=SHARED_GROUP, Name='shared one',
+                             Shared=True, SharedUsername='postgres'),
+            '2': {'Group': GROUP, 'Name': 'no host', 'Port': 5432,
+                  'Username': 'test_user', 'MaintenanceDB': 'postgres'},
+        })
+        with self.assertRaises(SystemExit):
+            self._load(None, from_setup=True)
+        self.assertFalse(self._group_exists(SHARED_GROUP))
+
+
+class TestReplaceKeepsServersOnInvalidFile(
+        _SharedImportMixin, BaseTestGenerator):
+    """replace=True clears the existing servers only after the file has
+    been validated."""
+
+    def runTest(self):
+        from pgadmin.model import db, Server, ServerGroup
+        group = ServerGroup(user_id=self.user.id, name=GROUP)
+        db.session.add(group)
+        db.session.flush()
+        db.session.add(Server(
+            user_id=self.user.id, servergroup_id=group.id, name='existing',
+            host='db.example.com', port=5432, username='test_user',
+            maintenance_db='postgres'))
+        db.session.commit()
+
+        self._write({
+            '1': {'Group': GROUP, 'Name': 'no host', 'Port': 5432,
+                  'Username': 'test_user', 'MaintenanceDB': 'postgres'},
+        })
+        with patch('pgadmin.utils.clear_database_servers') as mock_clear:
+            (ok, _), _ = self._load(['1'], replace=True)
+            self.assertFalse(ok)
+            mock_clear.assert_not_called()
+
+        self.assertEqual([s.name for s in self._servers()], ['existing'])
+
+
+class TestAdminSharedMustBeJsonTrue(_SharedImportMixin, BaseTestGenerator):
+    """Only the JSON value true means shared. A string such as "false" or
+    a number such as 1 is truthy in Python but must not share a server."""
+
+    is_admin = True
+
+    def runTest(self):
+        self._write({
+            '1': self._entry(Name='string false', Shared='false'),
+            '2': self._entry(Name='number one', Shared=1),
+            '3': self._entry(Name='real true', Shared=True),
+        })
+        (ok, _), _ = self._load(['1', '2', '3'])
+
+        self.assertTrue(ok)
+        servers = {s.name: s for s in self._servers()}
+        self.assertFalse(servers['string false'].shared)
+        self.assertFalse(servers['number one'].shared)
+        self.assertTrue(servers['real true'].shared)
+
+
+class TestNonAdminSharedStringFalseIsPrivate(
+        _SharedImportMixin, BaseTestGenerator):
+    """"Shared": "false" is not shared, so a non-admin imports it as a
+    normal private server, while a real true is still skipped."""
+
+    def runTest(self):
+        self._write({
+            '1': self._entry(Name='string false', Shared='false'),
+            '2': self._entry(group=SHARED_GROUP, Name='real true',
+                             Shared=True, SharedUsername='postgres'),
+        })
+        (ok, _), out = self._load(['1', '2'])
+
+        self.assertTrue(ok)
+        self.assertIn("Won't import the server 'real true'", out)
+        servers = self._servers()
+        self.assertEqual([s.name for s in servers], ['string false'])
+        self.assertFalse(servers[0].shared)
+        self.assertFalse(self._group_exists(SHARED_GROUP))
+
+
+class TestReplaceWithValidFile(_SharedImportMixin, BaseTestGenerator):
+    """replace=True with a valid file calls clear_database_servers, and
+    still imports the new servers.
+
+    clear_database_servers is mocked: the real one deletes every server of
+    the user, including the server the rest of the regression suite
+    connects to.
+    """
+
+    def runTest(self):
+        self._write({'1': self._entry(Name='imported')})
+        with patch('pgadmin.utils.clear_database_servers',
+                   return_value=None) as mock_clear:
+            (ok, _), _ = self._load(['1'], replace=True)
+            mock_clear.assert_called_once()
+            self.assertEqual(mock_clear.call_args[0][0].id, self.user.id)
+
+        self.assertTrue(ok)
+        self.assertEqual([s.name for s in self._servers()], ['imported'])

web/pgadmin/utils/tests/test_validate_json_data.py

@@ -131,6 +131,39 @@ class TestValidateJsonData(BaseTestGenerator):
              expected_error=None,
              expected_servers=["2"]
          )),
+        ('A non-admin shared server is dropped even when another server '
+         'fails validation',
+         dict(
+             servers={"1": server(Shared=True, SharedUsername="postgres"),
+                      "2": {"Group": "Servers", "Name": "No host",
+                            "Port": 5432, "Username": "postgres",
+                            "MaintenanceDB": "postgres"}},
+             is_admin=False,
+             expected_error="'Host' or 'Service' attribute not found",
+             expected_servers=["2"]
+         )),
+        ('A non-admin shared server without a name is dropped quietly',
+         dict(
+             servers={"1": {"Group": "Servers", "Shared": True,
+                            "Host": "127.0.0.1", "Port": 5432,
+                            "SharedUsername": "postgres",
+                            "MaintenanceDB": "postgres"},
+                      "2": server(Username="postgres")},
+             is_admin=False,
+             expected_error=None,
+             expected_servers=["2"]
+         )),
+        ('An admin keeps the shared server when another one fails '
+         'validation',
+         dict(
+             servers={"1": server(Shared=True, SharedUsername="postgres"),
+                      "2": {"Group": "Servers", "Name": "No host",
+                            "Port": 5432, "Username": "postgres",
+                            "MaintenanceDB": "postgres"}},
+             is_admin=True,
+             expected_error="'Host' or 'Service' attribute not found",
+             expected_servers=["1", "2"]
+         )),
     ]
 
     def runTest(self):

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

@@ -11,7 +11,7 @@
 import _ from 'lodash';
 import pgAdmin from 'sources/pgadmin';
 import current_user from 'pgadmin.user_management.current_user';
-import ServerSchema, { passexecForServerCopy, starfleetConnectionParams } from '../../../pgadmin/browser/server_groups/servers/static/js/server.ui';
+import ServerSchema, { passexecForServerCopy, sharingForServerCopy, starfleetConnectionParams } from '../../../pgadmin/browser/server_groups/servers/static/js/server.ui';
 import {genericBeforeEach, getCreateView, getEditView, getPropertiesView} from '../genericFunctions';
 
 describe('ServerSchema', ()=>{
@@ -237,4 +237,31 @@ describe('ServerSchema', ()=>{
         .toBeUndefined();
     });
   });
+
+  describe('sharingForServerCopy', ()=>{
+    it('clears the sharing of a copy made by a non-admin', ()=>{
+      let d = {name: 'x', shared: true, shared_username: 'someone'};
+      expect(sharingForServerCopy(d, {is_admin: false})).toBe(d);
+      expect(d.shared).toBe(false);
+      expect(d.shared_username).toBeNull();
+
+      d = {shared: true, shared_username: 'someone'};
+      sharingForServerCopy(d, undefined);
+      expect(d.shared).toBe(false);
+      expect(d.shared_username).toBeNull();
+    });
+
+    it('keeps the sharing of a copy made by an admin', ()=>{
+      let d = {name: 'x', shared: true, shared_username: 'someone'};
+      expect(sharingForServerCopy(d, {is_admin: true})).toBe(d);
+      expect(d.shared).toBe(true);
+      expect(d.shared_username).toBe('someone');
+    });
+
+    it('leaves a server that is not shared untouched', ()=>{
+      let d = {name: 'x', shared: false, shared_username: 'someone'};
+      sharingForServerCopy(d, {is_admin: false});
+      expect(d).toEqual({name: 'x', shared: false, shared_username: 'someone'});
+    });
+  });
 });

web/setup.py

@@ -40,7 +40,7 @@ else:
 from pgadmin.model import db, Version, User, Role, \
     SCHEMA_VERSION as CURRENT_SCHEMA_VERSION
 from pgadmin import create_app
-from pgadmin.utils import clear_database_servers, dump_database_servers, \
+from pgadmin.utils import dump_database_servers, \
     load_database_servers, _handle_error
 from pgadmin.utils.session import cleanup_session_files
 from pgadmin.setup import db_upgrade, create_app_data_directory
@@ -124,10 +124,8 @@ class ManageServers:
         try:
             app = create_app(config.APP_NAME + '-cli')
             with app.test_request_context():
-                if replace:
-                    clear_database_servers(load_user, True, auth_source)
                 load_database_servers(input_file, None, load_user, True,
-                                      auth_source)
+                                      auth_source, replace=replace)
         except Exception as e:
             print(str(e))