Merge branch 'cve-9.17-rebase' into master

docs/en_US/release_notes_9_17.rst

@@ -49,6 +49,13 @@ Bug fixes
   | `Issue #10117 <https://github.com/pgadmin-org/pgadmin4/issues/10117>`_ -  Fix a "'Response' object is not iterable" crash when expanding a Trigger node under a Table to view its trigger function, and report the correct error message on a node.sql failure in that flow.
   | `Issue #10158 <https://github.com/pgadmin-org/pgadmin4/issues/10158>`_ -  Honor the selected EOL sequence when copying query text to the clipboard.
   | `Issue #10187 <https://github.com/pgadmin-org/pgadmin4/issues/10187>`_ -  Fix the object browser's extension UI breaking under PostgreSQL 19's extension catalog changes.
+  | `Issue #10190 <https://github.com/pgadmin-org/pgadmin4/issues/10190>`_ -  Fix a tool-permission bypass where a user denied the Query Tool, Grant Wizard, or Schema Diff permission could still drive that tool's backend routes and Socket.IO handlers directly, since the permission check was applied only to a single "front door" route per tool. Also fixes a non-owner triggering an adhoc connection against another user's shared server persisting a new server record still owned by that other user (CVE-2026-17350). Reported by LXY.
+  | `Issue #10191 <https://github.com/pgadmin-org/pgadmin4/issues/10191>`_ -  Fix OS command injection in the MASTER_PASSWORD_HOOK feature, where an externally-sourced username (e.g. via OAuth2/OIDC, Kerberos, or webserver authentication) containing shell metacharacters could execute arbitrary commands as the pgAdmin service account when the configured hook string uses ``%u`` (CVE-2026-17347). Reported by Thiago Pereira.
+  | `Issue #10192 <https://github.com/pgadmin-org/pgadmin4/issues/10192>`_ -  Fix a lexer-differential bypass of the AI Assistant's read-only transaction guard, where sqlparse's string-literal lexing disagreed with PostgreSQL's own parser under ``standard_conforming_strings = on``, letting a crafted multi-statement payload smuggle a COMMIT past the intended read-only wrapper, an incomplete fix for CVE-2026-12045 (CVE-2026-17351). Reported by Kai Aizen.
+  | `Issue #10193 <https://github.com/pgadmin-org/pgadmin4/issues/10193>`_ -  Fix SQL injection in the Index Statistics all-indexes listing and the Publications/Subscriptions Dependencies views, where an apostrophe in a table, index, publication, or subscription name broke out of an unescaped template interpolation, an incomplete fix for CVE-2026-12044 (CVE-2026-17346). Reported by Hung Tran.
+  | `Issue #10194 <https://github.com/pgadmin-org/pgadmin4/issues/10194>`_ -  Fix several Constraints, Preferences, Debugger, and Schema Diff routes missing the ``@pga_login_required`` decorator, making them reachable without authentication in server mode, an incomplete fix for CVE-2026-12046 (CVE-2026-17348). Reported by Hung Tran.
+  | `Issue #10200 <https://github.com/pgadmin-org/pgadmin4/issues/10200>`_ -  Fix an adhoc server connection cloning another user's stored database credentials (password, save password flag, tunnel password) alongside ownership, letting a non-owner who cloned another user's shared server connect using that user's saved database password (CVE-2026-17349).
+  | `Issue #10213 <https://github.com/pgadmin-org/pgadmin4/issues/10213>`_ -  Fix OS command injection in the Import/Export Data tool, where a query-based export could pass a crafted query string past the ``\copy (...)`` parenthesis-balance guard by exploiting a backslash-escape mismatch with psql's default ``standard_conforming_strings = on`` behaviour, exposing a live ``TO PROGRAM`` clause for arbitrary command execution (CVE-2026-17566). Reported by Arpit Jain.
 
 Additional changes (no associated issue)
 ****************************************

web/config.py

@@ -656,6 +656,14 @@ USE_OS_SECRET_STORAGE = True
 # You can pass the current username as an argument to the external script
 # by specifying %u in config value.
 # E.g. - MASTER_PASSWORD_HOOK = '<PATH>/passwdgen_script.sh %u'
+#
+# The command is split into arguments and executed directly, without a
+# shell, so shell features (pipes, redirection, environment-variable
+# expansion, globbing) in this value are not interpreted; put any such
+# logic inside the hook script itself. On Windows, a batch file must be
+# invoked via an executable wrapper rather than directly. If the path to
+# the script/program contains spaces, quote it, e.g. -
+# MASTER_PASSWORD_HOOK = '"<PATH WITH SPACES>/passwdgen_script.exe" %u'
 ##########################################################################
 MASTER_PASSWORD_HOOK = None
 

web/pgadmin/authenticate/__init__.py

@@ -82,6 +82,40 @@ def socket_login_required(f):
     return wrapped
 
 
+def socket_permissions_required(*fsperms):
+    """Socket.IO equivalent of flask_security's permissions_required.
+
+    Refuses the event unless the current user is authenticated and holds
+    all of the named pgAdmin permissions (via their roles). This mirrors
+    the HTTP @permissions_required decorator so that Socket.IO handlers
+    enforce the same tool-level RBAC as the routes: an authenticated user
+    must not be able to reach a sensitive handler (e.g. the schema diff
+    comparison or the PSQL pseudo-terminal) for a tool whose permission
+    they have been denied.
+
+    has_permission() is used in preference to flask_principal's
+    Permission().can() because it reads the user's roles directly and
+    does not depend on the principal identity having been loaded onto the
+    socket request context; it also honours pgAdmin's Administrator
+    bypass (see CustomUserMixin).
+    """
+    def wrapper(f):
+        @functools.wraps(f)
+        def wrapped(*args, **kwargs):
+            if not current_user.is_authenticated:
+                disconnect()
+                raise ConnectionRefusedError("Unauthorised !")
+
+            for fsperm in fsperms:
+                if not current_user.has_permission(fsperm):
+                    disconnect()
+                    raise ConnectionRefusedError("Forbidden !")
+
+            return f(*args, **kwargs)
+        return wrapped
+    return wrapper
+
+
 class AuthenticateModule(pga_utils.PgAdminModule):
     def get_exposed_url_endpoints(self):
         return ['authenticate.login']

web/pgadmin/browser/__init__.py

@@ -27,7 +27,7 @@ from flask import current_app, render_template, url_for, make_response, \
     flash, Response, request, redirect, session
 from flask_babel import gettext
 from libgravatar import Gravatar
-from flask_security import current_user
+from flask_security import current_user, permissions_required
 from flask_login.utils import login_url
 from flask_security.changeable import send_password_changed_notice
 from flask_security.decorators import anonymous_user_required
@@ -50,6 +50,8 @@ from pgadmin.utils.ajax import make_json_response, internal_server_error, \
     bad_request
 from pgadmin.utils.csrf import pgCSRFProtect
 from pgadmin.utils.preferences import Preferences
+from pgadmin.tools.user_management.PgAdminPermissions import \
+    AllPermissionTypes
 from pgadmin.browser.register_browser_preferences import \
     register_browser_preferences
 from pgadmin.browser.register_editor_preferences import \
@@ -891,6 +893,7 @@ if hasattr(config, 'SECURITY_CHANGEABLE') and config.SECURITY_CHANGEABLE:
     @blueprint.route("/change_password", endpoint="change_password",
                      methods=['GET', 'POST'])
     @pgCSRFProtect.exempt
+    @permissions_required(AllPermissionTypes.change_password)
     @pga_login_required
     def change_password():
         """View function which handles a change password request."""

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

@@ -1048,7 +1048,7 @@ class PublicationView(PGChildNodeView, SchemaDiffObjectCompare):
         status, pname = self.conn.execute_scalar(get_name_sql)
         table_sql = render_template(
             "/".join([self.template_path, 'dependencies.sql']),
-            pname=pname
+            pname=pname, conn=self.conn
         )
         status, res = self.conn.execute_dict(table_sql)
         if not status:

web/pgadmin/browser/server_groups/servers/databases/publications/templates/publications/pg/default/sql/dependencies.sql

@@ -2,4 +2,4 @@ SELECT cl.oid AS oid,
 pg_catalog.quote_ident(pgb_table.schemaname)||'.'||pg_catalog.quote_ident(pgb_table.tablename) AS pubtable
 FROM pg_catalog.pg_publication_tables pgb_table
 LEFT JOIN pg_catalog.pg_class cl ON pgb_table.tablename = cl.relname
-WHERE pubname = '{{ pname }}' AND pgb_table.schemaname NOT LIKE 'pgagent';
+WHERE pubname = {{ pname|qtLiteral(conn) }} AND pgb_table.schemaname NOT LIKE 'pgagent';

web/pgadmin/browser/server_groups/servers/databases/publications/templates/publications/pg/default/sql/get_position.sql

@@ -1 +1 @@
-SELECT oid, pubname AS name FROM pg_catalog.pg_publication WHERE pubname = '{{ pubname }}';
+SELECT oid, pubname AS name FROM pg_catalog.pg_publication WHERE pubname = {{ pubname|qtLiteral(conn) }};

web/pgadmin/browser/server_groups/servers/databases/publications/templates/publications/ppas/default/sql/dependencies.sql

@@ -2,4 +2,4 @@ SELECT cl.oid AS oid,
 pg_catalog.quote_ident(pgb_table.schemaname)||'.'||pg_catalog.quote_ident(pgb_table.tablename) AS pubtable
 FROM pg_catalog.pg_publication_tables pgb_table
 LEFT JOIN pg_catalog.pg_class cl ON pgb_table.tablename = cl.relname
-WHERE pubname = '{{ pname }}' AND pgb_table.schemaname NOT LIKE 'pgagent';
+WHERE pubname = {{ pname|qtLiteral(conn) }} AND pgb_table.schemaname NOT LIKE 'pgagent';

web/pgadmin/browser/server_groups/servers/databases/publications/templates/publications/ppas/default/sql/get_position.sql

@@ -1 +1 @@
-SELECT oid, pubname AS name FROM pg_catalog.pg_publication WHERE pubname = '{{ pubname }}';
+SELECT oid, pubname AS name FROM pg_catalog.pg_publication WHERE pubname = {{ pubname|qtLiteral(conn) }};

web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/__init__.py

@@ -19,6 +19,7 @@ from flask_babel import gettext
 from pgadmin.browser.collection import CollectionNodeModule
 from pgadmin.utils.ajax import make_json_response, \
     make_response as ajax_response, internal_server_error
+from pgadmin.user_login_check import pga_login_required
 
 from config import PG_DEFAULT_DRIVER
 from .type import ConstraintRegistry
@@ -112,6 +113,7 @@ blueprint = ConstraintsModule(__name__)
 
 
 @blueprint.route('/nodes/<int:gid>/<int:sid>/<int:did>/<int:scid>/<int:tid>/')
+@pga_login_required
 def nodes(**kwargs):
     """
     Returns all constraint as a tree node.
@@ -137,6 +139,7 @@ def nodes(**kwargs):
 
 
 @blueprint.route('/obj/<int:gid>/<int:sid>/<int:did>/<int:scid>/<int:tid>/')
+@pga_login_required
 def proplist(**kwargs):
     """
     Returns all constraint with properties.
@@ -164,6 +167,7 @@ def proplist(**kwargs):
                  methods=['DELETE'])
 @blueprint.route('/delete/<int:gid>/<int:sid>/<int:did>/<int:scid>/<int:tid>/',
                  methods=['DELETE'])
+@pga_login_required
 def delete(**kwargs):
     """
     Delete multiple constraints under the table.

web/pgadmin/browser/server_groups/servers/databases/schemas/tables/templates/indexes/sql/16_plus/coll_stats.sql

@@ -12,7 +12,7 @@ FROM
         AND dep.refclassid=(SELECT oid FROM pg_catalog.pg_class WHERE relname='pg_constraint'))
     LEFT OUTER JOIN pg_catalog.pg_constraint con ON (con.tableoid = dep.refclassid AND con.oid = dep.refobjid)
 WHERE
-    schemaname = '{{schema}}'
-    AND stat.relname = '{{table}}'
+    schemaname = {{schema|qtLiteral(conn)}}
+    AND stat.relname = {{table|qtLiteral(conn)}}
     AND (con.contype IS NULL OR con.contype = 'p')
 ORDER BY indexrelname;

web/pgadmin/browser/server_groups/servers/databases/schemas/tables/templates/indexes/sql/default/coll_stats.sql

@@ -11,7 +11,7 @@ FROM
         AND dep.refclassid=(SELECT oid FROM pg_catalog.pg_class WHERE relname='pg_constraint'))
     LEFT OUTER JOIN pg_catalog.pg_constraint con ON (con.tableoid = dep.refclassid AND con.oid = dep.refobjid)
 WHERE
-    schemaname = '{{schema}}'
-    AND stat.relname = '{{table}}'
+    schemaname = {{schema|qtLiteral(conn)}}
+    AND stat.relname = {{table|qtLiteral(conn)}}
     AND (con.contype IS NULL OR con.contype = 'p')
 ORDER BY indexrelname;

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

@@ -892,7 +892,7 @@ class SubscriptionView(PGChildNodeView, SchemaDiffObjectCompare):
         status, subname = self.conn.execute_scalar(get_name_sql)
         table_sql = render_template(
             "/".join([self.template_path, 'dependencies.sql']),
-            subname=subname
+            subname=subname, conn=self.conn
         )
         status, res = self.conn.execute_dict(table_sql)
         if not status:

web/pgadmin/browser/server_groups/servers/databases/subscriptions/templates/subscriptions/sql/default/dependencies.sql

@@ -1,2 +1,2 @@
 SELECT subpublications AS pub FROM pg_catalog.pg_subscription
-WHERE subname = '{{subname}}';
+WHERE subname = {{subname|qtLiteral(conn)}};

web/pgadmin/browser/server_groups/servers/databases/subscriptions/templates/subscriptions/sql/default/get_position.sql

@@ -1 +1 @@
-SELECT oid, subname AS name FROM pg_catalog.pg_subscription WHERE subname = '{{ subname }}'  and subdbid={{did}} :: oid;
+SELECT oid, subname AS name FROM pg_catalog.pg_subscription WHERE subname = {{ subname|qtLiteral(conn) }}  and subdbid={{did}} :: oid;

web/pgadmin/browser/server_groups/servers/databases/tests/test_name_literal_sql_escaping.py

@@ -0,0 +1,208 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""Regression test for CVE-2026-12044 follow-up: object-name lookups in
+the index Statistics, publication and subscription templates must escape
+the interpolated name with ``qtLiteral(conn)``.
+
+These templates used to embed a browser-tree-sourced object name inside a
+bare ``'{{ name }}'`` literal on the (incorrect) assumption that such
+names could never contain an apostrophe. PostgreSQL permits arbitrary
+characters in quoted identifiers, so a low-privilege user who can
+``CREATE TABLE "x'; ..."`` (or CREATE PUBLICATION / SUBSCRIPTION) could
+plant a name that breaks out of the literal once any user views that
+object's Statistics or Dependencies panel -- the "SQL injection in
+pgAdmin index Statistics (incomplete fix for CVE-2026-12044)" report.
+
+This is the behavioural counterpart to the lint guard in
+``test_sql_string_literal_lint.py`` (which only checks the raw
+``'{{ x }}'`` pattern is absent) and complements
+``test_stats_template_regclass_cast.py`` (which covers the *single-index*
+``stats.sql`` pgstatindex path -- a different template from the
+``coll_stats.sql`` all-indexes listing fixed here).
+
+For each template it renders with a stacked-statement apostrophe payload
+and asserts:
+
+  * POSITIVE -- the name appears exactly as ``qtLiteral`` would escape it
+    (apostrophe doubled, wrapped in single quotes), i.e. the filter was
+    actually applied.
+  * SEMANTIC -- the rendered SQL parses as exactly one statement, so the
+    smuggled ``; <stmt>;`` cannot escape the intended query. This is the
+    property that actually matters; a raw-interpolation regression would
+    parse as several statements and fail here.
+
+No database connection is required: templates are rendered against a
+minimal Flask app with the production Jinja filters, exactly like
+``test_stats_template_regclass_cast.py``.
+"""
+
+import os
+
+import sqlparse
+from flask import Flask, render_template
+from jinja2 import FileSystemLoader
+
+from pgadmin.utils.driver import get_driver
+from pgadmin.utils.route import BaseTestGenerator
+from config import PG_DEFAULT_DRIVER
+
+
+# Stacked-statement payload. Under the old raw-interpolation form this
+# breaks out of the literal and injects a second statement; under
+# qtLiteral the apostrophe is doubled and the whole thing stays inside a
+# single string literal.
+PAYLOAD = "x'; SELECT pg_sleep(5); --"
+
+
+WEB_ROOT = os.path.realpath(
+    os.path.join(os.path.dirname(os.path.realpath(__file__)),
+                 os.pardir, os.pardir, os.pardir, os.pardir, os.pardir,
+                 os.pardir)
+)
+
+
+def _abs(*parts):
+    return os.path.join(WEB_ROOT, *parts)
+
+
+class _FakeConn:
+    """Stand-in for a psycopg connection so qtLiteral/qtIdent resolve
+    without a live server (see test_stats_template_regclass_cast.py)."""
+
+    conn = None
+
+    def __bool__(self):
+        return True
+
+
+class _FakeApp(Flask):
+    """Minimal Flask app mirroring the production Jinja filters."""
+
+    def __init__(self, template_root):
+        super().__init__('')
+        driver = get_driver(PG_DEFAULT_DRIVER, self)
+        self.jinja_env.filters['qtLiteral'] = driver.qtLiteral
+        self.jinja_env.filters['qtIdent'] = driver.qtIdent
+        self.jinja_env.filters['qtTypeIdent'] = driver.qtTypeIdent
+        self.jinja_env.loader = FileSystemLoader([template_root])
+
+
+_TABLES_ROOT = _abs(
+    'pgadmin', 'browser', 'server_groups', 'servers', 'databases',
+    'schemas', 'tables', 'templates')
+_PUB_ROOT = _abs(
+    'pgadmin', 'browser', 'server_groups', 'servers', 'databases',
+    'publications', 'templates')
+_SUB_ROOT = _abs(
+    'pgadmin', 'browser', 'server_groups', 'servers', 'databases',
+    'subscriptions', 'templates')
+
+
+class NameLiteralSQLEscapingTestCase(BaseTestGenerator):
+    """Each name-lookup template must escape the object name via
+    qtLiteral so a planted apostrophe cannot smuggle a second
+    statement."""
+
+    scenarios = [
+        ('Index stats coll_stats (default) escapes schema + table', dict(
+            template_root=_TABLES_ROOT,
+            template='indexes/sql/default/coll_stats.sql',
+            inject=dict(schema='public', table=PAYLOAD),
+            escaped_vars=['table'],
+        )),
+        ('Index stats coll_stats (16_plus) escapes schema + table', dict(
+            template_root=_TABLES_ROOT,
+            template='indexes/sql/16_plus/coll_stats.sql',
+            inject=dict(schema='public', table=PAYLOAD),
+            escaped_vars=['table'],
+        )),
+        ('Index stats coll_stats (default) escapes schema payload', dict(
+            template_root=_TABLES_ROOT,
+            template='indexes/sql/default/coll_stats.sql',
+            inject=dict(schema=PAYLOAD, table='t'),
+            escaped_vars=['schema'],
+        )),
+        ('Publication dependencies (pg) escapes pname', dict(
+            template_root=_PUB_ROOT,
+            template='publications/pg/default/sql/dependencies.sql',
+            inject=dict(pname=PAYLOAD),
+            escaped_vars=['pname'],
+        )),
+        ('Publication get_position (pg) escapes pubname', dict(
+            template_root=_PUB_ROOT,
+            template='publications/pg/default/sql/get_position.sql',
+            inject=dict(pubname=PAYLOAD),
+            escaped_vars=['pubname'],
+        )),
+        ('Publication dependencies (ppas) escapes pname', dict(
+            template_root=_PUB_ROOT,
+            template='publications/ppas/default/sql/dependencies.sql',
+            inject=dict(pname=PAYLOAD),
+            escaped_vars=['pname'],
+        )),
+        ('Publication get_position (ppas) escapes pubname', dict(
+            template_root=_PUB_ROOT,
+            template='publications/ppas/default/sql/get_position.sql',
+            inject=dict(pubname=PAYLOAD),
+            escaped_vars=['pubname'],
+        )),
+        ('Subscription dependencies escapes subname', dict(
+            template_root=_SUB_ROOT,
+            template='subscriptions/sql/default/dependencies.sql',
+            inject=dict(subname=PAYLOAD),
+            escaped_vars=['subname'],
+        )),
+        ('Subscription get_position escapes subname', dict(
+            template_root=_SUB_ROOT,
+            template='subscriptions/sql/default/get_position.sql',
+            inject=dict(subname=PAYLOAD, did=12345),
+            escaped_vars=['subname'],
+        )),
+    ]
+
+    def setUp(self):
+        self.app_under_test = _FakeApp(self.template_root)
+
+    def runTest(self):
+        conn = _FakeConn()
+        driver = get_driver(PG_DEFAULT_DRIVER, self.app_under_test)
+
+        # `_('...')` (gettext) is used for column aliases in coll_stats;
+        # inject an identity stand-in so rendering needs no babel context.
+        ctx = dict(self.inject, conn=conn, _=lambda s: s)
+        with self.app_under_test.app_context():
+            rendered = render_template(self.template, **ctx)
+
+        # POSITIVE: every payload-carrying var appears exactly as
+        # qtLiteral would escape it (apostrophe doubled, single-quoted).
+        for var in self.escaped_vars:
+            expected = driver.qtLiteral(self.inject[var], conn)
+            self.assertIn(
+                expected, rendered,
+                msg=('Expected qtLiteral-escaped {!r} ({!r}) in rendered '
+                     'SQL, so the name is safely quoted.\nRendered:\n{}'
+                     .format(var, expected, rendered)))
+
+        # SEMANTIC: the rendered SQL must be exactly one statement. If the
+        # apostrophe had broken out of the literal, the smuggled
+        # `; SELECT pg_sleep(5); --` would parse as extra statements.
+        statements = [
+            s for s in sqlparse.parse(rendered)
+            if s.token_first(skip_cm=True, skip_ws=True) is not None
+        ]
+        self.assertEqual(
+            len(statements), 1,
+            msg=('Rendered SQL must be a single statement; a payload '
+                 'apostrophe escaped the literal and smuggled additional '
+                 'statements.\nParsed {} statements from:\n{}'
+                 .format(len(statements), rendered)))
+
+    def tearDown(self):
+        pass

web/pgadmin/browser/server_groups/servers/databases/tests/test_sql_string_literal_lint.py

@@ -154,32 +154,6 @@ ALLOWLIST = [
       "'{{constraint_type}}'"), 1,
      "Single-char pg_constraint.contype code, hardcoded by handler."),
 
-    # ------------------------------------------------------------------
-    # Index stat lookups for an already-existing table. ``schema`` and
-    # ``table`` come from the browser tree node and ultimately from
-    # pg_catalog. Cannot contain apostrophes for any object pgAdmin was
-    # able to discover via its own browse queries (which use qtLiteral on
-    # the way in), so this is not a SQL injection sink. A correctness
-    # follow-up (legitimately named objects containing apostrophes) is
-    # tracked separately.
-    # ------------------------------------------------------------------
-    (('pgadmin/browser/server_groups/servers/databases/schemas/tables/'
-      'templates/indexes/sql/default/coll_stats.sql',
-      "'{{schema}}'"), 1,
-     "Schema name from pg_catalog via browser tree, not request input."),
-    (('pgadmin/browser/server_groups/servers/databases/schemas/tables/'
-      'templates/indexes/sql/default/coll_stats.sql',
-      "'{{table}}'"), 1,
-     "Table name from pg_catalog via browser tree, not request input."),
-    (('pgadmin/browser/server_groups/servers/databases/schemas/tables/'
-      'templates/indexes/sql/16_plus/coll_stats.sql',
-      "'{{schema}}'"), 1,
-     "Schema name from pg_catalog via browser tree, not request input."),
-    (('pgadmin/browser/server_groups/servers/databases/schemas/tables/'
-      'templates/indexes/sql/16_plus/coll_stats.sql',
-      "'{{table}}'"), 1,
-     "Table name from pg_catalog via browser tree, not request input."),
-
     # ------------------------------------------------------------------
     # Aggregate initial-condition strings. ``initial_val`` and
     # ``moving_initial_val`` come from the aggregate-creation form. The
@@ -300,36 +274,6 @@ ALLOWLIST = [
       "'{{ data.origin }}'"), 1,
      "Bounded enum (any/none); CREATE SUBSCRIPTION needs superuser."),
 
-    # ------------------------------------------------------------------
-    # Subscription / publication name lookups. ``subname``, ``pubname``
-    # and ``pname`` come from the browser tree (resolved from OID
-    # references in pg_catalog), not from request body input.
-    # ------------------------------------------------------------------
-    (('pgadmin/browser/server_groups/servers/databases/subscriptions/'
-      'templates/subscriptions/sql/default/get_position.sql',
-      "'{{ subname }}'"), 1,
-     "Subscription name from pg_catalog via browser tree."),
-    (('pgadmin/browser/server_groups/servers/databases/subscriptions/'
-      'templates/subscriptions/sql/default/dependencies.sql',
-      "'{{subname}}'"), 1,
-     "Subscription name from pg_catalog via browser tree."),
-    (('pgadmin/browser/server_groups/servers/databases/publications/'
-      'templates/publications/pg/default/sql/get_position.sql',
-      "'{{ pubname }}'"), 1,
-     "Publication name from pg_catalog via browser tree."),
-    (('pgadmin/browser/server_groups/servers/databases/publications/'
-      'templates/publications/pg/default/sql/dependencies.sql',
-      "'{{ pname }}'"), 1,
-     "Publication name from pg_catalog via browser tree."),
-    (('pgadmin/browser/server_groups/servers/databases/publications/'
-      'templates/publications/ppas/default/sql/get_position.sql',
-      "'{{ pubname }}'"), 1,
-     "Publication name from pg_catalog via browser tree."),
-    (('pgadmin/browser/server_groups/servers/databases/publications/'
-      'templates/publications/ppas/default/sql/dependencies.sql',
-      "'{{ pname }}'"), 1,
-     "Publication name from pg_catalog via browser tree."),
-
     # ------------------------------------------------------------------
     # DBMS job scheduler ``CREATE`` templates emit lines that are entirely
     # inside SQL ``--`` line comments (display-only headers, never

web/pgadmin/browser/tests/test_change_password_permission_required.py

@@ -0,0 +1,81 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""RBAC regression: the self-service change-password route must honour the
+'change_password' permission.
+
+AllPermissionTypes.change_password exists in the permission catalogue and
+is presented in Role management as a toggle ("Change Password"), but
+browser.change_password never checked it -- it relied on
+@pga_login_required alone. An administrator who explicitly revoked this
+permission from a role (e.g. to force password changes through an
+external identity provider only) had that restriction silently ignored:
+any authenticated user under that role could still change their own
+password through the in-app form regardless.
+
+Note this is lower severity than the tool-RBAC gaps it sits alongside: the
+route is inherently self-scoped (a user can only ever change their own
+password, never another user's), so the impact is a policy-enforcement
+gap rather than a privilege-escalation or cross-tenant data exposure.
+
+Skipped in DESKTOP mode, where every request is auto-authenticated as the
+all-permissions DESKTOP_USER and no permission decorator is exercisable.
+"""
+
+import config
+
+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']
+
+
+class ChangePasswordPermissionRequiredTestCase(BaseTestGenerator):
+    """A user lacking the change_password permission must get 403 from
+    browser.change_password, not just from tool routes."""
+
+    scenarios = [
+        ('browser.change_password requires change_password permission',
+         dict()),
+    ]
+
+    def setUp(self):
+        if not config.SERVER_MODE:
+            self.skipTest(
+                'Permission decorators are only exercisable in SERVER '
+                'mode; DESKTOP mode auto-authenticates as the '
+                'all-permissions DESKTOP_USER on every request.'
+            )
+        if not (hasattr(config, 'SECURITY_CHANGEABLE') and
+                config.SECURITY_CHANGEABLE):
+            self.skipTest(
+                'browser.change_password is only registered when '
+                'SECURITY_CHANGEABLE is enabled.'
+            )
+
+    @create_user_wise_test_client(test_user_details)
+    def runTest(self):
+        response = self.tester.get('/browser/change_password',
+                                   follow_redirects=False)
+
+        self.assertEqual(
+            response.status_code, 403,
+            'browser.change_password did not enforce the change_password '
+            'permission for a user lacking it: expected 403, got {0}. '
+            'Body: {1!r}'.format(
+                response.status_code, response.data[:200]
+            )
+        )
+
+    def tearDown(self):
+        pass

web/pgadmin/llm/__init__.py

@@ -14,6 +14,7 @@ import logging
 import ssl
 from flask import request
 from flask_babel import gettext
+from flask_security import permissions_required
 from pgadmin.utils import PgAdminModule
 from pgadmin.utils.preferences import Preferences
 from pgadmin.utils.ajax import make_json_response, internal_server_error
@@ -22,6 +23,8 @@ from pgadmin.utils.constants import MIMETYPE_APP_JS
 from pgadmin.utils.csrf import pgCSRFProtect
 import config
 from pgadmin.llm.utils import LLMApiError
+from pgadmin.tools.user_management.PgAdminPermissions import \
+    AllPermissionTypes
 
 # Try to use certifi for proper SSL certificate handling
 try:
@@ -1047,6 +1050,7 @@ def _fetch_docker_models(api_url):
     methods=["GET"],
     endpoint='security_report'
 )
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_security_report(sid):
     """
@@ -1113,6 +1117,7 @@ def generate_security_report(sid):
     endpoint='security_report_stream'
 )
 @pgCSRFProtect.exempt
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_security_report_stream(sid):
     """
@@ -1298,6 +1303,7 @@ def _gather_security_config(conn, manager):
     methods=["GET"],
     endpoint='database_security_report'
 )
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_database_security_report(sid, did):
     """
@@ -1366,6 +1372,7 @@ def generate_database_security_report(sid, did):
     endpoint='database_security_report_stream'
 )
 @pgCSRFProtect.exempt
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_database_security_report_stream(sid, did):
     """
@@ -1426,6 +1433,7 @@ def generate_database_security_report_stream(sid, did):
     methods=["GET"],
     endpoint='schema_security_report'
 )
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_schema_security_report(sid, did, scid):
     """
@@ -1508,6 +1516,7 @@ def generate_schema_security_report(sid, did, scid):
     endpoint='schema_security_report_stream'
 )
 @pgCSRFProtect.exempt
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_schema_security_report_stream(sid, did, scid):
     """
@@ -1582,6 +1591,7 @@ def generate_schema_security_report_stream(sid, did, scid):
     methods=["GET"],
     endpoint='performance_report'
 )
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_performance_report(sid):
     """
@@ -1648,6 +1658,7 @@ def generate_performance_report(sid):
     endpoint='performance_report_stream'
 )
 @pgCSRFProtect.exempt
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_performance_report_stream(sid):
     """
@@ -1706,6 +1717,7 @@ def generate_performance_report_stream(sid):
     methods=["GET"],
     endpoint='database_performance_report'
 )
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_database_performance_report(sid, did):
     """
@@ -1774,6 +1786,7 @@ def generate_database_performance_report(sid, did):
     endpoint='database_performance_report_stream'
 )
 @pgCSRFProtect.exempt
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_database_performance_report_stream(sid, did):
     """
@@ -1834,6 +1847,7 @@ def generate_database_performance_report_stream(sid, did):
     methods=["GET"],
     endpoint='database_design_report'
 )
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_database_design_report(sid, did):
     """
@@ -1902,6 +1916,7 @@ def generate_database_design_report(sid, did):
     endpoint='database_design_report_stream'
 )
 @pgCSRFProtect.exempt
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_database_design_report_stream(sid, did):
     """
@@ -1962,6 +1977,7 @@ def generate_database_design_report_stream(sid, did):
     methods=["GET"],
     endpoint='schema_design_report'
 )
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_schema_design_report(sid, did, scid):
     """
@@ -2044,6 +2060,7 @@ def generate_schema_design_report(sid, did, scid):
     endpoint='schema_design_report_stream'
 )
 @pgCSRFProtect.exempt
+@permissions_required(AllPermissionTypes.tools_ai)
 @pga_login_required
 def generate_schema_design_report_stream(sid, did, scid):
     """

web/pgadmin/llm/tests/test_database_tool_security.py

@@ -10,21 +10,32 @@
 """Security regression tests for execute_sql_query input validation.
 
 These tests pin the multi-statement / leading-keyword guard added in
-``_validate_readonly_query``.  The validator is the load-bearing
-defense against an attacker escaping the ``BEGIN TRANSACTION READ
-ONLY`` wrapper by emitting a transaction-control statement (COMMIT,
-END, ROLLBACK, ABORT) followed by writes -- the original Isaac Chen
-report from 2026-06-08.
+``_validate_readonly_query``.  It rejects the original Isaac Chen
+report from 2026-06-08 (a transaction-control statement such as COMMIT,
+END, ROLLBACK, ABORT followed by writes) whenever sqlparse and
+PostgreSQL agree on where the statement boundary is.
 
-The tests are pure unit tests (no DB, no Flask client) -- they exercise
-the validator directly.  Anything that reaches the connection layer is
-already too late.
+``_validate_readonly_query`` is a fast pre-filter, not the security
+boundary -- see ``ValidateReadonlyQueryLexerDifferentialTestCase``
+below and the module docstring in ``database.py`` for why sqlparse
+cannot be the load-bearing check, and ``ExecuteReadonlyQueryProtocolTestCase``
+for the protocol-level enforcement that actually is (Kai Aizen /
+SnailSploit report, 2026-07-23, on the bf4792444446 fix).
+
+Most of these tests are pure unit tests (no DB, no Flask client) --
+they exercise the validator directly. Anything that reaches the
+connection layer is already too late for validator, which is exactly
+why the protocol-level fix exists.
 """
 
+from unittest.mock import patch, MagicMock
+
 from pgadmin.utils.route import BaseTestGenerator
 from pgadmin.llm.tools.database import (
     DatabaseToolError,
     _validate_readonly_query,
+    _connect_readonly,
+    execute_readonly_query,
 )
 
 
@@ -404,3 +415,170 @@ class ValidateReadonlyQueryRejectTestCase(BaseTestGenerator):
             f"Validator accepted query that should have been "
             f"rejected: {self.query!r}"
         )
+
+
+class ValidateReadonlyQueryLexerDifferentialTestCase(BaseTestGenerator):
+    """Pins a KNOWN, ACCEPTED limitation of the sqlparse-based validator.
+
+    sqlparse's string-literal lexing does not always match PostgreSQL's.
+    Under standard_conforming_strings = on (the default), a backslash
+    inside a '...'-quoted literal is an ordinary character to PostgreSQL
+    but sqlparse treats it as escaping the following quote, so it keeps
+    reading past what PostgreSQL considers the end of the string. That
+    lets a payload like the one below smuggle ``;COMMIT;<write>;`` past
+    the "exactly one statement" check disguised as a single string
+    literal, even though PostgreSQL executes it as four statements --
+    the report from Kai Aizen / SnailSploit (2026-07-23) against the
+    bf4792444446 fix.
+
+    This is NOT something ``_validate_readonly_query`` can be made to
+    catch in general (it would require re-implementing PostgreSQL's
+    lexer, including its dependence on server-side GUCs like
+    standard_conforming_strings). The test below asserts the validator
+    accepts the payload -- documenting that this is expected -- and
+    exists only to point at where the real protection lives: see
+    ExecuteReadonlyQueryProtocolTestCase, which pins that the query is
+    always executed with prepare=True. That forces PostgreSQL's own
+    Parse step -- not a client-side approximation of it -- to reject
+    any text containing more than one statement, regardless of how it
+    is lexed.
+    """
+
+    scenarios = [
+        ('Backslash-quote smuggled COMMIT + DDL', dict(
+            query=(
+                "SELECT '\\';COMMIT;CREATE TABLE pwn(x int);"
+                "SELECT 1 --'"
+            ),
+        )),
+        ('Backslash-quote smuggled COMMIT + COPY TO PROGRAM', dict(
+            query=(
+                "SELECT '\\';COMMIT;COPY (SELECT 1) TO PROGRAM 'id';"
+                "SELECT 1 --'"
+            ),
+        )),
+    ]
+
+    def setUp(self):
+        pass
+
+    def runTest(self):
+        # Documents current, accepted behavior -- must NOT raise.
+        # If this ever starts raising, _validate_readonly_query has
+        # changed in a way that may be worth understanding, but the
+        # query is still safe only because of prepare=True downstream;
+        # don't mistake a change here for the security fix itself.
+        keyword = _validate_readonly_query(self.query)
+        self.assertIsNone(keyword)
+
+
+class ExecuteReadonlyQueryProtocolTestCase(BaseTestGenerator):
+    """Pins that execute_readonly_query always runs the LLM's query with
+    prepare=True.
+
+    This is the actual load-bearing defense against statement-boundary
+    smuggling (see ValidateReadonlyQueryLexerDifferentialTestCase):
+    prepare=True forces psycopg3's extended query protocol, whose Parse
+    step is answered by PostgreSQL's own parser and rejects a query
+    string containing more than one SQL statement -- independent of
+    sqlparse, and independent of any particular payload shape. Verified
+    end-to-end against a live PostgreSQL 16 instance during the
+    SnailSploit report triage (2026-07-23): the smuggled-COMMIT payload
+    that _validate_readonly_query accepts is rejected by PostgreSQL's
+    Parse step with "cannot insert multiple commands into a prepared
+    statement" once prepare=True is set, both before and after the
+    max_rows LIMIT-wrapping applied to SELECT queries.
+
+    This test mocks the connection layer (no DB) and only pins the
+    wiring: that prepare=True is passed on every call, not just for
+    inputs that look suspicious. A future refactor that drops the
+    keyword argument, or only sets it conditionally, must fail this
+    test.
+    """
+
+    scenarios = [
+        ('Benign SELECT', dict(
+            query='SELECT 1',
+        )),
+        ('Backslash-quote smuggled COMMIT + DDL', dict(
+            query=(
+                "SELECT '\\';COMMIT;CREATE TABLE pwn(x int);"
+                "SELECT 1 --'"
+            ),
+        )),
+    ]
+
+    def setUp(self):
+        pass
+
+    def runTest(self):
+        mock_manager = MagicMock()
+        mock_conn = MagicMock()
+        mock_conn.execute_void.return_value = (True, None)
+        mock_conn.execute_2darray.return_value = (
+            True, {'columns': [], 'rows': []}
+        )
+
+        with patch(
+            'pgadmin.llm.tools.database._get_connection',
+            return_value=(mock_manager, mock_conn)
+        ), patch(
+            'pgadmin.llm.tools.database._connect_readonly',
+            return_value=(True, None)
+        ):
+            execute_readonly_query(sid=1, did=1, query=self.query)
+
+        mock_conn.execute_2darray.assert_called_once()
+        _args, kwargs = mock_conn.execute_2darray.call_args
+        self.assertTrue(
+            kwargs.get('prepare') is True,
+            "execute_readonly_query must run the LLM-supplied query "
+            "with prepare=True so PostgreSQL's own Parse step -- not "
+            "the sqlparse pre-filter -- enforces the single-statement "
+            "guarantee."
+        )
+
+
+class ConnectReadonlyForcesPrepareThresholdTestCase(BaseTestGenerator):
+    """Pins that _connect_readonly forces conn.conn.prepare_threshold = 0.
+
+    prepare=True on the execute_2darray call is necessary but NOT
+    sufficient: psycopg3's PrepareManager.get() returns Prepare.NO --
+    silently falling back to the multi-statement-capable simple query
+    protocol -- whenever the connection's prepare_threshold is None,
+    *before* it even inspects the prepare argument. pgAdmin's per-server
+    "Prepare threshold" field defaults to blank/None, so on a default
+    server the prepare=True guarantee never engages and the
+    CVE-2026-12045 smuggled-COMMIT bypass stays live (Kai Aizen /
+    SnailSploit follow-up, verified live against PostgreSQL on
+    2026-07-24).
+
+    _connect_readonly() therefore forces prepare_threshold = 0 on the
+    LLM's single-use connection. This test pins that override so a
+    future refactor that drops it -- reopening the bypass -- fails here,
+    without needing a live server.
+    """
+
+    def setUp(self):
+        pass
+
+    def runTest(self):
+        mock_manager = MagicMock()
+        mock_conn = MagicMock()
+        mock_conn.connected.return_value = True
+        mock_conn.execute_void.return_value = (True, None)
+        # Emulate pgAdmin's default: extended protocol disabled.
+        mock_conn.conn.prepare_threshold = None
+
+        status, msg = _connect_readonly(
+            mock_manager, mock_conn, 'llm_test_conn')
+
+        self.assertTrue(status, msg)
+        self.assertEqual(
+            mock_conn.conn.prepare_threshold, 0,
+            "_connect_readonly must force prepare_threshold=0 on the "
+            "LLM connection; otherwise psycopg3 ignores prepare=True "
+            "(when the server's Prepare threshold is blank/None, the "
+            "default) and falls back to the multi-statement-capable "
+            "simple query protocol, reopening CVE-2026-12045."
+        )

web/pgadmin/llm/tools/database.py

@@ -44,10 +44,21 @@ LLM_APP_NAME_PREFIX = 'pgAdmin 4 - LLM'
 # effective if the LLM cannot exit that transaction. A multi-statement
 # payload such as `COMMIT; <write>; SELECT 1` would otherwise terminate
 # the read-only transaction (via COMMIT/END/ROLLBACK/ABORT) and run
-# subsequent statements in autocommit mode. The allowlist below, in
-# combination with the single-statement check, ensures the LLM cannot
-# emit transaction-control statements, SET/RESET, DML/DDL, CALL, COPY,
-# or anything else that could escape or weaken the read-only sandbox.
+# subsequent statements in autocommit mode.
+#
+# The allowlist below rejects transaction-control statements, SET/RESET,
+# DML/DDL, CALL, COPY, and anything else that could weaken the read-only
+# sandbox, *for queries sqlparse correctly recognises as a single
+# statement*. It is a fast, user-friendly pre-filter, not the boundary
+# enforcement: sqlparse's lexer can disagree with PostgreSQL's own
+# parser about where one statement ends and another begins (e.g. string
+# literal escaping under standard_conforming_strings), so a payload can
+# look like one safe statement to sqlparse while PostgreSQL executes it
+# as several. The actual guarantee that only one statement reaches the
+# server is enforced at the protocol level -- see _execute_readonly_query,
+# which runs the query with prepare=True so PostgreSQL's own Parse step
+# (not a client-side approximation of it) rejects multi-statement text.
+#
 # EXPLAIN ANALYZE on a SELECT remains supported; EXPLAIN ANALYZE on a
 # write statement is blocked by PostgreSQL itself inside the read-only
 # transaction.
@@ -126,6 +137,24 @@ def _connect_readonly(
             if not status:
                 return False, msg
 
+        # Force the extended query protocol on this connection,
+        # regardless of the server's configured "Prepare threshold"
+        # (which defaults to blank/None, meaning "never prepare"). The
+        # per-call prepare=True passed to execute_2darray() in
+        # _execute_readonly_query() is silently ignored by psycopg3
+        # whenever conn.prepare_threshold is None -- PrepareManager.get()
+        # checks that before it looks at the prepare argument at all, and
+        # falls back to the simple query protocol, which is exactly the
+        # multi-statement-capable protocol this defense must avoid.
+        # Setting the threshold to 0 ("always prepare") on this
+        # single-use, per-query connection (see conn_id generation in
+        # execute_readonly_query) makes the extended protocol -- and
+        # therefore PostgreSQL's single-statement Parse-step guarantee --
+        # unconditional here, without touching the server-wide setting or
+        # any other connection.
+        if getattr(conn, 'conn', None) is not None:
+            conn.conn.prepare_threshold = 0
+
         # Set application name via SQL - this is thread-safe and doesn't
         # require environment variables. The name will be visible in
         # pg_stat_activity to identify LLM connections.
@@ -139,6 +168,16 @@ def _connect_readonly(
             # Non-fatal - connection still works without custom app name
             pass
 
+        # Defense-in-depth: make READ ONLY the session default, not just a
+        # property of the transaction started by BEGIN TRANSACTION READ
+        # ONLY. If a statement ever managed to end that transaction early
+        # (e.g. a smuggled COMMIT), the next transaction on this connection
+        # -- implicit or explicit -- would still be read-only rather than
+        # falling back to a writable default.
+        conn.execute_void(
+            "SET SESSION CHARACTERISTICS AS TRANSACTION READ ONLY"
+        )
+
         return True, None
 
     except Exception as e:
@@ -166,20 +205,28 @@ def _first_real_keyword(statement) -> str:
 
 def _validate_readonly_query(query: str) -> None:
     """
-    Ensure an LLM-supplied query is a single, read-only statement.
+    Reject LLM-supplied queries that are obviously not a single,
+    read-only statement.
 
-    This is the load-bearing defense against an attacker escaping the
-    read-only transaction wrapper by emitting multiple statements. The
-    PostgreSQL simple-query protocol cheerfully runs a whole
-    semicolon-separated batch in one round-trip, so a payload such as
-    ``COMMIT; <write>; SELECT 1`` would terminate the read-only
-    transaction (via the leading ``COMMIT``) and execute the remaining
-    statements in autocommit mode. The trailing ``ROLLBACK`` would then
-    be a no-op.
+    This is a fast pre-filter, not the security boundary: it rejects
+    plainly-disallowed statement types and multi-statement input *as
+    sqlparse's lexer sees it*. sqlparse's notion of where a string
+    literal (and therefore a statement) ends can disagree with
+    PostgreSQL's own parser -- e.g. a backslash before a quote is an
+    escape to sqlparse but an ordinary character to PostgreSQL when
+    standard_conforming_strings is on (the default) -- so a payload can
+    pass this check as a single SELECT while PostgreSQL would actually
+    execute it as several statements, including a COMMIT that ends the
+    read-only transaction. That class of attack is closed at the
+    protocol level, not here: see _execute_readonly_query, which
+    executes the query with prepare=True so PostgreSQL's own Parse step
+    -- the actual authority on statement boundaries -- rejects any text
+    containing more than one statement.
 
     Validation rules:
 
-    * The input must contain exactly one non-empty statement.
+    * The input must contain exactly one non-empty statement (as far as
+      sqlparse can tell).
     * The leading keyword must be in :data:`_ALLOWED_LEADING_KEYWORDS`.
 
     PostgreSQL is left to enforce the rest -- ``EXPLAIN ANALYZE`` on a
@@ -242,6 +289,19 @@ def _execute_readonly_query(conn, query: str) -> dict:
     The query is wrapped in a read-only transaction to ensure
     no data modifications can occur.
 
+    The query is executed with prepare=True, and the connection's
+    prepare_threshold is forced to 0 in _connect_readonly(), which
+    together force it through PostgreSQL's extended query protocol.
+    (prepare=True alone is not enough: psycopg3 ignores it whenever
+    prepare_threshold is None -- pgAdmin's blank-by-default server
+    setting -- and falls back to the multi-statement-capable simple
+    query protocol.) The server's Parse step accepts only a single SQL
+    statement in the extended protocol, so this is what actually
+    guarantees the LLM cannot smuggle a second statement (e.g. a COMMIT
+    to end the read-only transaction early) past
+    _validate_readonly_query's sqlparse-based check -- see the note on
+    _validate_readonly_query for why that check alone is not sufficient.
+
     Args:
         conn: Database connection
         query: SQL query to execute
@@ -266,8 +326,10 @@ def _execute_readonly_query(conn, query: str) -> dict:
             )
 
         try:
-            # Execute the actual query
-            status, result = conn.execute_2darray(query)
+            # Execute the actual query. prepare=True, combined with the
+            # prepare_threshold=0 forced in _connect_readonly(), is
+            # load-bearing -- see the docstring above.
+            status, result = conn.execute_2darray(query, prepare=True)
 
             if not status:
                 raise DatabaseToolError(

web/pgadmin/misc/workspaces/__init__.py

@@ -176,6 +176,23 @@ def adhoc_connect_server():
                 # Clone the server object
                 server = server.clone()
 
+                # Server.clone() copies every column from the source row,
+                # including user_id/shared/shared_username. When the source
+                # is another user's shared server, the clone must not inherit
+                # that ownership: force the new adhoc record to belong to the
+                # current user and to be private, otherwise a non-owner ends
+                # up persisting a cross-tenant, administrator-owned server
+                # row.
+                server.user_id = current_user.id
+                server.shared = False
+                server.shared_username = None
+                # The clone also inherits the source server's stored
+                # credentials; drop them so a non-owner can't persist
+                # another user's secret material under their own row.
+                server.password = None
+                server.save_password = False
+                server.tunnel_password = None
+
                 # Replace the following with the new/changed value.
                 server.maintenance_db = new_db
                 server.username = new_username

web/pgadmin/misc/workspaces/tests/__init__.py

@@ -0,0 +1,8 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################

web/pgadmin/misc/workspaces/tests/test_adhoc_connect_server_ownership.py

@@ -0,0 +1,138 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""Regression: adhoc connect must not persist cross-tenant server rows.
+
+/misc/workspace/adhoc_connect_server clones an existing server when a
+'sid' is supplied. Server.clone() copies every column of the source row,
+including user_id/shared/shared_username. When a non-owner triggers an
+adhoc connect against an administrator-owned *shared* server, the clone
+must be re-homed to the current user and made private; otherwise pgAdmin
+persists a new, administrator-owned (user_id of the admin), shared adhoc
+server row created at the behest of another user, a cross-tenant
+integrity problem.
+
+This test creates a shared server as the admin, then, as a non-admin
+user, calls adhoc_connect_server with that server's id and asserts that
+every resulting adhoc server row belongs to the non-admin user and is not
+shared.
+
+Skipped in DESKTOP mode (single user; no cross-tenant boundary).
+"""
+
+import json
+
+import config
+
+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']
+
+
+class AdhocConnectServerOwnershipTestCase(BaseTestGenerator):
+    """A non-owner adhoc connect against a shared server must not leave a
+    persisted server row owned by another (admin) user."""
+
+    scenarios = [
+        ('adhoc clone of a shared server is re-homed to the caller',
+         dict()),
+    ]
+
+    def setUp(self):
+        self.shared_sid = None
+        if not config.SERVER_MODE:
+            self.skipTest(
+                'Adhoc ownership isolation only applies to server mode.')
+
+        # Create a shared server as the admin user.
+        self.server['shared'] = True
+        url = "/browser/server/obj/{0}/".format(utils.SERVER_GROUP)
+        response = self.tester.post(
+            url, data=json.dumps(self.server), content_type='html/json')
+        self.assertEqual(response.status_code, 200)
+        self.shared_sid = json.loads(
+            response.data.decode('utf-8'))['node']['_id']
+
+    def _user_id(self, email):
+        from pgadmin.model import User
+        with self.app.app_context():
+            user = User.query.filter_by(username=email).first()
+            return user.id if user else None
+
+    def _adhoc_servers(self):
+        from pgadmin.model import Server
+        with self.app.app_context():
+            return [
+                dict(id=s.id, user_id=s.user_id, shared=bool(s.shared))
+                for s in Server.query.filter_by(is_adhoc=1).all()
+            ]
+
+    @create_user_wise_test_client(test_user_details)
+    def runTest(self):
+        if not self.shared_sid:
+            raise Exception('Shared server was not created.')
+
+        admin_email = \
+            config_data['pgAdmin4_login_credentials']['login_username']
+        non_admin_email = test_user_details['login_username']
+        admin_id = self._user_id(admin_email)
+        non_admin_id = self._user_id(non_admin_email)
+        self.assertIsNotNone(non_admin_id)
+
+        # As the non-admin, trigger an adhoc connect that clones the
+        # admin-owned shared server. The connection attempt itself may
+        # fail; what matters is the persisted row.
+        data = dict(
+            server_name='adhoc_isolation_probe',
+            did=self.server.get('did', 1),
+            sid=self.shared_sid,
+            host=self.server['host'],
+            port=self.server['port'],
+            user=self.server['username'],
+        )
+        self.tester.post(
+            '/misc/workspace/adhoc_connect_server',
+            data=json.dumps(data), content_type='application/json')
+
+        adhoc = self._adhoc_servers()
+        self.assertGreaterEqual(
+            len(adhoc), 1,
+            'Expected an adhoc server row to have been persisted.')
+        for row in adhoc:
+            self.assertNotEqual(
+                row['user_id'], admin_id,
+                'Adhoc server row {0} is owned by the administrator '
+                '(user_id={1}); a non-owner created a cross-tenant '
+                'server record.'.format(row['id'], row['user_id']))
+            self.assertEqual(
+                row['user_id'], non_admin_id,
+                'Adhoc server row {0} should be owned by the calling '
+                'user (id={1}), got user_id={2}.'.format(
+                    row['id'], non_admin_id, row['user_id']))
+            self.assertFalse(
+                row['shared'],
+                'Adhoc server row {0} must not be shared.'.format(
+                    row['id']))
+
+    def tearDown(self):
+        # Remove any adhoc rows left behind, then the shared server.
+        from pgadmin.model import db, Server
+        with self.app.app_context():
+            for s in Server.query.filter_by(is_adhoc=1).all():
+                db.session.delete(s)
+            db.session.commit()
+        if self.shared_sid:
+            utils.delete_server_with_api(
+                self.__class__.tester, self.shared_sid)

web/pgadmin/preferences/__init__.py

@@ -185,6 +185,7 @@ def preferences_s():
 
 
 @blueprint.route("/get_all_cli", methods=["GET"], endpoint='get_all_cli')
+@pga_login_required
 def get_all_cli():
     """Fetch all preferences for caching."""
     # Load Preferences

web/pgadmin/tools/backup/__init__.py

@@ -511,6 +511,7 @@ def create_backup_objects_job(sid):
 @blueprint.route(
     '/utility_exists/<int:sid>/<backup_obj_type>', endpoint='utility_exists'
 )
+@permissions_required(AllPermissionTypes.tools_backup)
 @pga_login_required
 def check_utility_exists(sid, backup_obj_type):
     """
@@ -553,6 +554,7 @@ def check_utility_exists(sid, backup_obj_type):
 @blueprint.route(
     '/objects/<int:sid>/<int:did>/<int:scid>', endpoint='schema_objects'
 )
+@permissions_required(AllPermissionTypes.tools_backup)
 @pga_login_required
 def objects(sid, did, scid=None):
     """

web/pgadmin/tools/debugger/__init__.py

@@ -943,6 +943,7 @@ def initialize_target(debug_type, trans_id, sid, did,
 @blueprint.route(
     '/close/<int:trans_id>', methods=["DELETE"], endpoint='close'
 )
+@pga_login_required
 def close(trans_id):
     """
     close(trans_id)
@@ -1786,6 +1787,7 @@ def select_frame(trans_id, frame_id):
     '/get_arguments/<int:sid>/<int:did>/<int:scid>/<int:func_id>',
     methods=['GET'], endpoint='get_arguments'
 )
+@permissions_required(AllPermissionTypes.tools_debugger)
 @pga_login_required
 def get_arguments_sqlite(sid, did, scid, func_id):
     """
@@ -1879,6 +1881,7 @@ def get_array_string(data, i):
     '/set_arguments/<int:sid>/<int:did>/<int:scid>/<int:func_id>',
     methods=['POST'], endpoint='set_arguments'
 )
+@permissions_required(AllPermissionTypes.tools_debugger)
 @pga_login_required
 def set_arguments_sqlite(sid, did, scid, func_id):
     """
@@ -1977,6 +1980,7 @@ def set_arguments_sqlite(sid, did, scid, func_id):
     '/clear_arguments/<int:sid>/<int:did>/<int:scid>/<int:func_id>',
     methods=['POST'], endpoint='clear_arguments'
 )
+@permissions_required(AllPermissionTypes.tools_debugger)
 @pga_login_required
 def clear_arguments_sqlite(sid, did, scid, func_id):
     """

web/pgadmin/tools/erd/__init__.py

@@ -32,7 +32,7 @@ from pgadmin.utils.constants import PREF_LABEL_KEYBOARD_SHORTCUTS, \
     PREF_LABEL_OPTIONS
 from .utils import ERDHelper
 from pgadmin.utils.exception import ConnectionLost
-from pgadmin.authenticate import socket_login_required
+from pgadmin.authenticate import socket_permissions_required
 from pgadmin.tools.user_management.PgAdminPermissions import AllPermissionTypes
 from ... import socketio
 
@@ -585,6 +585,7 @@ def panel(trans_id):
     '/initialize/<int:trans_id>/<int:sgid>/<int:sid>/<int:did>',
     methods=["POST"], endpoint='initialize'
 )
+@permissions_required(AllPermissionTypes.tools_erd_tool)
 @pga_login_required
 def initialize_erd(trans_id, sgid, sid, did):
     """
@@ -659,6 +660,7 @@ def _get_connection(sid, did, trans_id, db_name=None):
 @blueprint.route('/prequisite/<int:trans_id>/<int:sgid>/<int:sid>/<int:did>',
                  methods=["GET"],
                  endpoint='prequisite')
+@permissions_required(AllPermissionTypes.tools_erd_tool)
 @pga_login_required
 def prequisite(trans_id, sgid, sid, did):
     conn = _get_connection(sid, did, trans_id)
@@ -722,6 +724,7 @@ def translate_foreign_keys(tab_fks, tab_data, all_nodes):
 @blueprint.route('/sql/<int:trans_id>/<int:sgid>/<int:sid>/<int:did>',
                  methods=["POST"],
                  endpoint='sql')
+@permissions_required(AllPermissionTypes.tools_erd_tool)
 @pga_login_required
 def sql(trans_id, sgid, sid, did):
     data = json.loads(request.data)
@@ -776,7 +779,7 @@ def connect():
 
 
 @socketio.on('tables', namespace=SOCKETIO_NAMESPACE)
-@socket_login_required
+@socket_permissions_required(AllPermissionTypes.tools_erd_tool)
 def tables(params):
     try:
         helper = ERDHelper(params['trans_id'], params['sid'], params['did'])

web/pgadmin/tools/grant_wizard/__init__.py

@@ -248,6 +248,7 @@ def get_node_sql_with_type(node_id, node_type, server_prop,
     '/<int:sid>/<int:did>/<int:node_id>/<node_type>/',
     methods=['GET'], endpoint='objects'
 )
+@permissions_required(AllPermissionTypes.tools_grant_wizard)
 @pga_login_required
 @check_precondition
 def properties(sid, did, node_id, node_type):
@@ -404,6 +405,7 @@ def set_priv_for_package(server_prop, data, acls):
     '/sql/<int:sid>/<int:did>/',
     methods=['POST'], endpoint='modified_sql'
 )
+@permissions_required(AllPermissionTypes.tools_grant_wizard)
 @pga_login_required
 @check_precondition
 def msql(sid, did):
@@ -538,6 +540,7 @@ def parse_priv(data, acls, server_prop):
 @blueprint.route(
     '/<int:sid>/<int:did>/', methods=['POST'], endpoint='apply'
 )
+@permissions_required(AllPermissionTypes.tools_grant_wizard)
 @pga_login_required
 @check_precondition
 def save(sid, did):

web/pgadmin/tools/import_export/__init__.py

@@ -281,12 +281,18 @@ def _is_query_parens_balanced(query):
     skip content inside those constructs either, or an attacker can
     hide an unbalanced ) inside them to bypass validation.
 
-    Backslash escapes (\\x) are always handled inside single-quoted
-    strings. psql enables this for E'...' strings unconditionally
-    and for all strings when standard_conforming_strings=off. By
-    always handling backslashes, we match psql in all modes. This
-    may reject some valid queries (e.g. '\\') with scs=on) but
-    will never accept a dangerous one.
+    Backslash handling inside single-quoted strings is ambiguous:
+    psql treats \\ as a plain character when
+    standard_conforming_strings=on (the default since PG 9.1, and
+    the default on every PostgreSQL version pgAdmin 4 supports), but
+    as an escape character when scs=off or inside E'...' strings.
+    Silently picking one interpretation can let a crafted query be
+    accepted here while psql parses it differently (e.g. a \\'
+    that we treat as "still inside the string" while psql, under
+    the default scs=on, treats it as the string's closing quote,
+    exposing a real ')' immediately after). Rather than guess the
+    server's scs setting, we reject any query containing a
+    backslash inside a single-quoted string outright.
     """
     depth = 0
     i = 0
@@ -296,14 +302,15 @@ def _is_query_parens_balanced(query):
         ch = query[i]
 
         # Single-quoted string literal: skip to closing quote.
-        # Handles both '' and \' escape sequences to match psql's
-        # strtokx which enables backslash escaping for E-strings
-        # (always) and all strings when scs=off.
+        # Handles the '' escape sequence to match psql's strtokx.
+        # A bare backslash is rejected outright (see docstring):
+        # its meaning depends on standard_conforming_strings, which
+        # we cannot reliably know here, so we refuse to guess.
         if ch == "'":
             i += 1
             while i < length:
                 if query[i] == '\\':
-                    i += 2  # backslash escape: skip next char
+                    return False  # ambiguous escape: reject
                 elif query[i] == "'":
                     if i + 1 < length and query[i + 1] == "'":
                         i += 2  # '' escape
@@ -561,6 +568,7 @@ def create_import_export_job(sid):
 
 
 @blueprint.route('/get_settings/', methods=['GET'], endpoint='get_settings')
+@permissions_required(AllPermissionTypes.tools_import_export_data)
 @pga_login_required
 def get_import_export_settings():
     settings = get_setting('import_export_setting', None)
@@ -574,6 +582,7 @@ def get_import_export_settings():
 @blueprint.route(
     '/utility_exists/<int:sid>', endpoint='utility_exists'
 )
+@permissions_required(AllPermissionTypes.tools_import_export_data)
 @pga_login_required
 def check_utility_exists(sid):
     """

web/pgadmin/tools/import_export/tests/test_import_export_create_job_unit_test.py

@@ -86,25 +86,35 @@ class IEQueryParensBalancedTest(BaseTestGenerator):
          dict(query="SELECT ')", expected=False)),
         ('Rejected: unterminated double-quoted identifier',
          dict(query='SELECT "col)', expected=False)),
-        # The parser handles backslash escapes inside single-quoted
-        # strings unconditionally (matching E-string and scs=off
-        # semantics), so the close paren is correctly identified as
-        # being inside the string literal. Original test labelled this
-        # as a "false positive" expecting over-rejection; in fact the
-        # parser is precise here. Keep the case to document handling.
-        ('Balanced: E-string with escaped quote and paren inside literal',
-         dict(query="SELECT E'\\')'", expected=True)),
+        # Whether a backslash inside a single-quoted string is an
+        # escape character depends on standard_conforming_strings,
+        # which the parser cannot know. Rather than guess (and risk
+        # disagreeing with psql's actual parse), any backslash inside
+        # a single-quoted string is rejected outright.
+        ('Rejected: backslash inside single-quoted string is ambiguous',
+         dict(query="SELECT E'\\')'", expected=False)),
         ('Unbalanced: E-string RCE bypass',
          dict(query="E'\\'' ) TO PROGRAM 'cmd' --'",
               expected=False)),
         ('Unbalanced: scs=off backslash escape bypass',
          dict(query="'\\'' ) TO PROGRAM 'cmd' --'",
               expected=False)),
-        ('Balanced: backslash-backslash then close quote',
-         dict(query="SELECT '\\\\' FROM t", expected=True)),
+        ('Rejected: backslash-backslash inside string is ambiguous',
+         dict(query="SELECT '\\\\' FROM t", expected=False)),
         ('Unbalanced: injection split across lines',
          dict(query="SELECT 1)\nTO PROGRAM 'id' --(",
               expected=False)),
+        # CVE regression: real psql (standard_conforming_strings=on,
+        # the default) treats \\ as a literal character, so 'a\\'
+        # closes the string literal at that second quote -- it does
+        # NOT escape it. The unescaped ')' immediately after is then
+        # live and closes the wrapping \\copy (...) context, exposing
+        # a TO PROGRAM clause for RCE. The old parser treated \\'
+        # as "still inside the string" unconditionally and missed
+        # this. Reported by Arpit Jain (arpitjain099), 2026-07-27.
+        ('Unbalanced: scs=on backslash-quote RCE bypass (CVE)',
+         dict(query="SELECT 'a\\') TO PROGRAM 'echo pwned' x'",
+              expected=False)),
     ]
 
     def runTest(self):

web/pgadmin/tools/import_export_servers/__init__.py

@@ -61,6 +61,7 @@ def index():
 
 
 @blueprint.route('/get_servers', methods=['GET'], endpoint='get_servers')
+@permissions_required(AllPermissionTypes.tools_import_export_servers)
 @pga_login_required
 def get_servers():
     """
@@ -92,6 +93,7 @@ def get_servers():
 
 
 @blueprint.route('/load_servers', methods=['POST'], endpoint='load_servers')
+@permissions_required(AllPermissionTypes.tools_import_export_servers)
 @pga_login_required
 def load_servers():
     """

web/pgadmin/tools/maintenance/__init__.py

@@ -292,6 +292,7 @@ def create_maintenance_job(sid, did):
 @blueprint.route(
     '/utility_exists/<int:sid>', endpoint='utility_exists'
 )
+@permissions_required(AllPermissionTypes.tools_maintenance)
 @pga_login_required
 def check_utility_exists(sid):
     """

web/pgadmin/tools/psql/__init__.py

@@ -20,16 +20,18 @@ from flask import Response, request
 from flask import render_template, copy_current_request_context, \
     current_app as app
 from flask_babel import gettext
-from flask_security import current_user
+from flask_security import current_user, permissions_required
 from pgadmin.user_login_check import pga_login_required
 from pgadmin.browser.utils import underscore_unescape, underscore_escape
 from pgadmin.utils import PgAdminModule
 from pgadmin.utils.driver import get_driver
 from ... import socketio as sio
 from pgadmin.utils import get_complete_file_path
-from pgadmin.authenticate import socket_login_required
+from pgadmin.authenticate import socket_login_required, \
+    socket_permissions_required
 from pgadmin.model import Server
 from pgadmin.utils.server_access import get_server
+from pgadmin.tools.user_management.PgAdminPermissions import AllPermissionTypes
 
 if _platform == 'win32':
     # Check Windows platform support for WinPty api, Disable psql
@@ -76,6 +78,7 @@ blueprint = PSQLModule('psql', __name__, static_url_path='/static')
 @blueprint.route('/panel/<int:trans_id>',
                  methods=["POST"],
                  endpoint="panel")
+@permissions_required(AllPermissionTypes.tools_psql_tool)
 @pga_login_required
 def panel(trans_id):
     """
@@ -292,7 +295,7 @@ def pty_handel_io(connection_data, data, sid):
 
 
 @sio.on('start_process', namespace='/pty')
-@socket_login_required
+@socket_permissions_required(AllPermissionTypes.tools_psql_tool)
 def start_process(data):
     """
     Start the pty terminal and execute psql command and emit results to user.
@@ -475,6 +478,7 @@ def other_key_press(data):
 
 
 @sio.on('socket_input', namespace='/pty')
+@socket_permissions_required(AllPermissionTypes.tools_psql_tool)
 def socket_input(data):
     """
     This get the user input through socket.
@@ -503,6 +507,7 @@ def socket_input(data):
 
 
 @sio.on('socket_set_role', namespace='/pty')
+@socket_permissions_required(AllPermissionTypes.tools_psql_tool)
 def socket_set_role(data):
     """
     This function sets the role used to connect to server.
@@ -537,6 +542,7 @@ def socket_set_role(data):
 
 
 @sio.on('resize', namespace='/pty')
+@socket_permissions_required(AllPermissionTypes.tools_psql_tool)
 def resize(data):
     """
     Resize the pty terminal as per the UI terminal.

web/pgadmin/tools/restore/__init__.py

@@ -492,6 +492,7 @@ def create_restore_job(sid):
 @blueprint.route(
     '/utility_exists/<int:sid>', endpoint='utility_exists'
 )
+@permissions_required(AllPermissionTypes.tools_restore)
 @pga_login_required
 def check_utility_exists(sid):
     """

web/pgadmin/tools/schema_diff/__init__.py

@@ -28,7 +28,7 @@ from pgadmin.utils.driver import get_driver
 from pgadmin.utils.constants import PREF_LABEL_DISPLAY, \
     ERROR_MSG_TRANS_ID_NOT_FOUND
 from sqlalchemy import or_
-from pgadmin.authenticate import socket_login_required
+from pgadmin.authenticate import socket_permissions_required
 from pgadmin import socketio
 from pgadmin.tools.user_management.PgAdminPermissions import AllPermissionTypes
 from pgadmin.utils.server_access import \
@@ -207,6 +207,7 @@ def update_session_diff_transaction(trans_id, session_obj, diff_model_obj):
     methods=["GET"],
     endpoint="initialize"
 )
+@permissions_required(AllPermissionTypes.tools_schema_diff)
 @pga_login_required
 def initialize(trans_id):
     """
@@ -237,6 +238,7 @@ def initialize(trans_id):
 @blueprint.route('/close/<int:trans_id>',
                  methods=["DELETE"],
                  endpoint='close')
+@pga_login_required
 def close(trans_id):
     """
     Remove the session details for the particular transaction id.
@@ -270,6 +272,7 @@ def close(trans_id):
     methods=["GET"],
     endpoint="servers"
 )
+@permissions_required(AllPermissionTypes.tools_schema_diff)
 @pga_login_required
 def servers():
     """
@@ -323,6 +326,7 @@ def servers():
     methods=["GET"],
     endpoint="get_server"
 )
+@permissions_required(AllPermissionTypes.tools_schema_diff)
 @pga_login_required
 def get_server(sid, did):
     """
@@ -366,6 +370,7 @@ def get_server(sid, did):
     methods=["POST"],
     endpoint="connect_server"
 )
+@permissions_required(AllPermissionTypes.tools_schema_diff)
 @pga_login_required
 def connect_server(sid):
     # Check if server is already connected then no need to reconnect again.
@@ -394,6 +399,7 @@ def connect_server(sid):
     methods=["POST"],
     endpoint="connect_database"
 )
+@permissions_required(AllPermissionTypes.tools_schema_diff)
 @pga_login_required
 def connect_database(sid, did):
     server = get_server_access(sid)
@@ -411,6 +417,7 @@ def connect_database(sid, did):
     methods=["GET"],
     endpoint="databases"
 )
+@permissions_required(AllPermissionTypes.tools_schema_diff)
 @pga_login_required
 def databases(sid):
     """
@@ -454,6 +461,7 @@ def databases(sid):
     methods=["GET"],
     endpoint="schemas"
 )
+@permissions_required(AllPermissionTypes.tools_schema_diff)
 @pga_login_required
 def schemas(sid, did):
     """
@@ -478,7 +486,7 @@ def schemas(sid, did):
 
 
 @socketio.on('compare_database', namespace=SOCKETIO_NAMESPACE)
-@socket_login_required
+@socket_permissions_required(AllPermissionTypes.tools_schema_diff)
 def compare_database(params):
     """
     This function will compare the two databases.
@@ -637,7 +645,7 @@ def compare_database(params):
 
 
 @socketio.on('compare_schema', namespace=SOCKETIO_NAMESPACE)
-@socket_login_required
+@socket_permissions_required(AllPermissionTypes.tools_schema_diff)
 def compare_schema(params):
     """
     This function will compare the two schema.
@@ -707,6 +715,7 @@ def compare_schema(params):
     methods=["GET"],
     endpoint="ddl_compare"
 )
+@permissions_required(AllPermissionTypes.tools_schema_diff)
 @pga_login_required
 def ddl_compare(trans_id, source_sid, source_did, source_scid,
                 target_sid, target_did, target_scid, source_oid,

web/pgadmin/tools/search_objects/__init__.py

@@ -62,6 +62,7 @@ def index():
 
 
 @blueprint.route("types/<int:sid>/<int:did>", endpoint='types')
+@permissions_required(AllPermissionTypes.tools_search_objects)
 @pga_login_required
 def types(sid, did):
     so_obj = SearchObjectsHelper(sid, did, blueprint.show_system_objects())

web/pgadmin/tools/sqleditor/__init__.py

@@ -193,6 +193,7 @@ def index():
     methods=["PUT", "POST"],
     endpoint="initialize_viewdata"
 )
+@permissions_required(AllPermissionTypes.tools_query_tool)
 @pga_login_required
 def initialize_viewdata(trans_id, cmd_type, obj_type, sgid, sid, did, obj_id):
     """

web/pgadmin/tools/sqleditor/tests/test_nlq_chat_tool_dispatch_regression.py

@@ -0,0 +1,228 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""End-to-end regression test for the CVE-2026-12045 bypass fix, driven
+through the REAL /sqleditor/nlq/chat/<trans_id>/stream HTTP route.
+
+The existing NLQ tests in test_nlq_chat.py all patch
+``pgadmin.llm.chat.chat_with_database_stream`` directly, which stubs out
+the entire tool-call loop and never touches ``execute_tool`` /
+``execute_readonly_query`` in ``pgadmin.llm.tools.database``. That leaves
+a gap: nothing exercises the actual production code path a real AI
+Assistant request takes -- Flask route -> chat_with_database_stream's
+tool-dispatch loop -> execute_tool -> execute_readonly_query ->
+conn.execute_2darray -- to confirm prepare=True is still wired through
+when the "chosen" query originates from an LLM tool call rather than a
+direct Python call (as in test_database_tool_security.py's
+ExecuteReadonlyQueryProtocolTestCase).
+
+This test closes that gap without needing a real LLM API key: it patches
+only ``pgadmin.llm.chat.get_llm_client`` (the same patch point the
+project's own EXPLAIN-analysis and NLQ tests already use for other
+scenarios) to return a fake client whose ``chat_stream()`` simulates the
+model choosing to call ``execute_sql_query`` with the smuggled-COMMIT
+payload from the Kai Aizen / SnailSploit report (2026-07-23). Everything
+below that -- chat.py's real tool-dispatch loop, execute_tool,
+execute_readonly_query -- runs for real. Only the DB connection object
+itself is mocked (as in ExecuteReadonlyQueryProtocolTestCase), since a
+real PostgreSQL connection isn't available in this test environment.
+"""
+
+import json
+from unittest.mock import patch, MagicMock
+
+from pgadmin.utils.route import BaseTestGenerator
+from pgadmin.llm.models import LLMResponse, ToolCall, StopReason
+
+
+BYPASS_PAYLOAD = (
+    "SELECT '\\';COMMIT;CREATE TABLE pwn_marker(x int);SELECT 1 --'"
+)
+
+
+def _make_mock_llm_client(tool_query, final_text='Here is what I found.'):
+    """A fake LLM client whose chat_stream() first requests a tool call
+    with `tool_query`, then (once the tool result is fed back) ends the
+    turn with `final_text`. Mirrors the shape client.chat_stream() is
+    expected to yield: str chunks and/or a terminal LLMResponse.
+    """
+    state = {'iteration': 0}
+
+    def chat_stream_side_effect(*_args, **_kwargs):
+        state['iteration'] += 1
+        if state['iteration'] == 1:
+            yield LLMResponse(
+                content='',
+                tool_calls=[ToolCall(
+                    id='tc-1',
+                    name='execute_sql_query',
+                    arguments={'query': tool_query},
+                )],
+                stop_reason=StopReason.TOOL_USE,
+            )
+        else:
+            yield final_text
+            yield LLMResponse(
+                content=final_text,
+                stop_reason=StopReason.END_TURN,
+            )
+
+    mock_client = MagicMock()
+    mock_client.chat_stream.side_effect = chat_stream_side_effect
+    return mock_client
+
+
+class NLQToolDispatchProtocolTestCase(BaseTestGenerator):
+    """Pins that a real request through /sqleditor/nlq/chat/.../stream,
+    when the model chooses to call execute_sql_query with the
+    backslash-quote smuggled-COMMIT payload, still reaches
+    conn.execute_2darray with prepare=True -- through the full
+    production dispatch path, not just a direct Python call.
+
+    The DB connection is mocked (no real PostgreSQL available here), so
+    this does not re-prove that PostgreSQL itself rejects the payload
+    -- that is already covered by ExecuteReadonlyQueryProtocolTestCase
+    (mock-based) and by manual verification against a live server. This
+    test's job is narrower and complementary: prove the HTTP route
+    really drives execute_readonly_query with this exact attacker-
+    controlled string and that prepare=True is not lost or made
+    conditional anywhere along the way.
+    """
+
+    scenarios = [
+        ('Tool call with bypass payload reaches DB layer with '
+         'prepare=True', dict(
+             mock_execute_2darray_result=(True, {
+                 'columns': [], 'rows': []
+             }),
+         )),
+        ('Tool call failure (simulating real Postgres Parse-step '
+         'rejection) is surfaced, not a crash', dict(
+             mock_execute_2darray_result=(
+                 False,
+                 'cannot insert multiple commands into a prepared '
+                 'statement'
+             ),
+         )),
+    ]
+
+    def setUp(self):
+        pass
+
+    def runTest(self):
+        trans_id = 12345
+
+        mock_trans_obj = MagicMock()
+        mock_trans_obj.sid = 1
+        mock_trans_obj.did = 1
+
+        mock_session_conn = MagicMock()
+        mock_session_conn.connected.return_value = True
+
+        mock_session = {'sid': 1, 'did': 1}
+
+        # The connection object *inside* pgadmin.llm.tools.database --
+        # this is what execute_2darray(prepare=True) is called on.
+        mock_llm_conn = MagicMock()
+        mock_llm_conn.execute_void.return_value = (True, None)
+        mock_llm_conn.execute_2darray.return_value = \
+            self.mock_execute_2darray_result
+
+        mock_client = _make_mock_llm_client(BYPASS_PAYLOAD)
+
+        patches = [
+            patch('pgadmin.llm.utils.is_llm_enabled', return_value=True),
+            patch('pgadmin.llm.chat.is_llm_available', return_value=True),
+            patch(
+                'pgadmin.llm.chat.get_llm_client',
+                return_value=mock_client
+            ),
+            patch(
+                'pgadmin.tools.sqleditor.check_transaction_status',
+                return_value=(
+                    True, None, mock_session_conn, mock_trans_obj,
+                    mock_session
+                )
+            ),
+            patch(
+                'pgadmin.llm.tools.database._get_connection',
+                return_value=(MagicMock(), mock_llm_conn)
+            ),
+            patch(
+                'pgadmin.llm.tools.database._connect_readonly',
+                return_value=(True, None)
+            ),
+            patch(
+                'pgadmin.authenticate.mfa.utils.mfa_required',
+                lambda f: f
+            ),
+        ]
+
+        for p in patches:
+            p.start()
+
+        try:
+            response = self.tester.post(
+                f'/sqleditor/nlq/chat/{trans_id}/stream',
+                data=json.dumps({'message': 'please run my query'}),
+                content_type='application/json',
+                follow_redirects=True
+            )
+
+            self.assertEqual(response.status_code, 200)
+            self.assertIn('text/event-stream', response.content_type)
+
+            # Consume the stream so the generator (and therefore the
+            # real tool-dispatch loop) actually executes.
+            raw = response.data.decode('utf-8')
+
+            # The route must not blow up with an unhandled exception --
+            # either a 'complete' event (tool succeeded from the
+            # model's perspective) or the loop simply continuing to a
+            # final answer after a tool-error message is acceptable;
+            # what must NOT happen is the request dying before
+            # execute_2darray is ever reached.
+            events = []
+            for line in raw.split('\n'):
+                if line.startswith('data: '):
+                    try:
+                        events.append(json.loads(line[6:]))
+                    except json.JSONDecodeError:
+                        pass
+            event_types = [e.get('type') for e in events]
+            self.assertIn('complete', event_types)
+
+            # The actual security-critical assertion: regardless of
+            # whether the simulated DB call "succeeded" or "failed",
+            # it was invoked with the attacker's exact payload and
+            # prepare=True -- proving the fix is wired all the way
+            # through the real HTTP route, not bypassed by some
+            # earlier short-circuit.
+            mock_llm_conn.execute_2darray.assert_called_once()
+            call_args, call_kwargs = \
+                mock_llm_conn.execute_2darray.call_args
+            # execute_readonly_query wraps SELECT-prefixed queries with
+            # a LIMIT subquery before executing (see database.py) -- the
+            # attacker's payload must still be present verbatim inside
+            # that wrapper, and it's the wrapped string that actually
+            # gets sent with prepare=True.
+            self.assertIn(BYPASS_PAYLOAD, call_args[0])
+            self.assertTrue(
+                call_kwargs.get('prepare') is True,
+                "The real /sqleditor/nlq/chat/.../stream route must "
+                "still invoke execute_2darray(query, prepare=True) "
+                "for an LLM-chosen query, even when that query is the "
+                "backslash-quote smuggled-COMMIT bypass payload."
+            )
+        finally:
+            for p in patches:
+                p.stop()
+
+    def tearDown(self):
+        pass

web/pgadmin/tools/tests/__init__.py

@@ -0,0 +1,8 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################

web/pgadmin/tools/tests/test_tool_permissions_required.py

@@ -0,0 +1,239 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""RBAC regression: tool backend routes must enforce the tool permission.
+
+pgAdmin gates each tool behind a permission (tools_query_tool,
+tools_grant_wizard, tools_schema_diff, tools_erd_tool, tools_psql_tool,
+tools_debugger, ...). Historically the @permissions_required decorator was
+applied only to a single 'front door' route per tool, while alternate
+initialisation routes, object-discovery/SQL/apply routes and Socket.IO
+handlers relied on @pga_login_required alone. That let a user who had been
+denied a tool still drive the rest of that tool's backend workflow:
+
+  * View/Edit Data via sqleditor.initialize_viewdata (Query Tool),
+  * object discovery, SQL preview and the real GRANT via the grant wizard,
+  * schema diff initialisation, enumeration and comparison,
+  * ERD initialisation and DDL generation,
+  * the PSQL panel,
+  * the debugger's stored-argument routes.
+
+This test logs in as a user with no roles (therefore no tool permissions)
+and asserts that every one of those backend routes returns HTTP 403, i.e.
+the permission gate fires before the route body runs. It is deliberately a
+single consolidated 'blanket' check so that a newly-added route in any of
+these blueprints which forgets the decorator is caught here.
+
+Also covers three further front-door-only gaps found auditing the same
+pattern beyond the original CVE fix's scope:
+
+  * AI Reports (tools_ai): the Security/Performance/Design report routes
+    in the llm blueprint had no server-side permission check at all --
+    only the frontend menu was hidden behind AllPermissionTypes.TOOLS_AI
+    (see llm/static/js/ai_tools.js). A denied user could call any of the
+    14 report/report-stream routes directly and have real server/
+    database/schema content sent to the configured external LLM
+    provider.
+  * import_export_servers: only 'save' was gated; 'get_servers' and
+    'load_servers' relied on @pga_login_required alone.
+  * search_objects: only 'search' was gated; 'types' was not.
+
+The permission check is the outermost decorator, so it runs before any
+connection/transaction lookup; dummy path parameters (ids of 1, a random
+trans_id) are sufficient to reach and trip it.
+
+Skipped in DESKTOP mode, where every request is auto-authenticated as the
+all-permissions DESKTOP_USER and no permission decorator is exercisable.
+"""
+
+import secrets
+
+import config
+import flask
+
+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']
+
+
+# A throwaway transaction id; these routes never get far enough to use it.
+_TRANS = secrets.choice(range(1, 9999999))
+
+
+class ToolPermissionRequiredTestCase(BaseTestGenerator):
+    """A user lacking a tool's permission must get 403 from every backend
+    route in that tool, not just its primary entry point."""
+
+    scenarios = [
+        # --- Query Tool: View/Edit Data alternate init (AC-001) ---
+        ('sqleditor.initialize_viewdata requires tools_query_tool',
+         dict(method='post', endpoint='sqleditor.initialize_viewdata',
+              url_kwargs=dict(trans_id=_TRANS, cmd_type=1, obj_type='table',
+                              sgid=1, sid=1, did=1, obj_id=1))),
+
+        # --- Grant Wizard: discovery, SQL preview, apply (AC-002) ---
+        ('grant_wizard.objects requires tools_grant_wizard',
+         dict(method='get', endpoint='grant_wizard.objects',
+              url_kwargs=dict(sid=1, did=1, node_id=1, node_type='table'))),
+        ('grant_wizard.modified_sql requires tools_grant_wizard',
+         dict(method='post', endpoint='grant_wizard.modified_sql',
+              url_kwargs=dict(sid=1, did=1))),
+        ('grant_wizard.apply requires tools_grant_wizard',
+         dict(method='post', endpoint='grant_wizard.apply',
+              url_kwargs=dict(sid=1, did=1))),
+
+        # --- Schema Diff: init, enumeration, connect, ddl (AC-005) ---
+        ('schema_diff.initialize requires tools_schema_diff',
+         dict(method='get', endpoint='schema_diff.initialize',
+              url_kwargs=dict(trans_id=_TRANS))),
+        ('schema_diff.servers requires tools_schema_diff',
+         dict(method='get', endpoint='schema_diff.servers',
+              url_kwargs=dict())),
+        ('schema_diff.get_server requires tools_schema_diff',
+         dict(method='get', endpoint='schema_diff.get_server',
+              url_kwargs=dict(sid=1, did=1))),
+        ('schema_diff.connect_server requires tools_schema_diff',
+         dict(method='post', endpoint='schema_diff.connect_server',
+              url_kwargs=dict(sid=1))),
+        ('schema_diff.connect_database requires tools_schema_diff',
+         dict(method='post', endpoint='schema_diff.connect_database',
+              url_kwargs=dict(sid=1, did=1))),
+        ('schema_diff.databases requires tools_schema_diff',
+         dict(method='get', endpoint='schema_diff.databases',
+              url_kwargs=dict(sid=1))),
+        ('schema_diff.schemas requires tools_schema_diff',
+         dict(method='get', endpoint='schema_diff.schemas',
+              url_kwargs=dict(sid=1, did=1))),
+        ('schema_diff.ddl_compare requires tools_schema_diff',
+         dict(method='get', endpoint='schema_diff.ddl_compare',
+              url_kwargs=dict(trans_id=_TRANS, source_sid=1, source_did=1,
+                              source_scid=1, target_sid=1, target_did=1,
+                              target_scid=1, source_oid=1, target_oid=1,
+                              node_type='table', comp_status='Different'))),
+
+        # --- ERD: init and DDL generation (audit, new) ---
+        ('erd.initialize requires tools_erd_tool',
+         dict(method='post', endpoint='erd.initialize',
+              url_kwargs=dict(trans_id=_TRANS, sgid=1, sid=1, did=1))),
+        ('erd.prequisite requires tools_erd_tool',
+         dict(method='get', endpoint='erd.prequisite',
+              url_kwargs=dict(trans_id=_TRANS, sgid=1, sid=1, did=1))),
+        ('erd.sql requires tools_erd_tool',
+         dict(method='post', endpoint='erd.sql',
+              url_kwargs=dict(trans_id=_TRANS, sgid=1, sid=1, did=1))),
+
+        # --- PSQL: panel (audit, new) ---
+        ('psql.panel requires tools_psql_tool',
+         dict(method='post', endpoint='psql.panel',
+              url_kwargs=dict(trans_id=_TRANS))),
+
+        # --- Debugger: directly-addressable stored-argument routes ---
+        ('debugger.get_arguments requires tools_debugger',
+         dict(method='get', endpoint='debugger.get_arguments',
+              url_kwargs=dict(sid=1, did=1, scid=1, func_id=1))),
+        ('debugger.set_arguments requires tools_debugger',
+         dict(method='post', endpoint='debugger.set_arguments',
+              url_kwargs=dict(sid=1, did=1, scid=1, func_id=1))),
+        ('debugger.clear_arguments requires tools_debugger',
+         dict(method='post', endpoint='debugger.clear_arguments',
+              url_kwargs=dict(sid=1, did=1, scid=1, func_id=1))),
+
+        # --- AI Reports: Security/Performance/Design, all levels ---
+        ('llm.security_report requires tools_ai',
+         dict(method='get', endpoint='llm.security_report',
+              url_kwargs=dict(sid=1))),
+        ('llm.security_report_stream requires tools_ai',
+         dict(method='get', endpoint='llm.security_report_stream',
+              url_kwargs=dict(sid=1))),
+        ('llm.database_security_report requires tools_ai',
+         dict(method='get', endpoint='llm.database_security_report',
+              url_kwargs=dict(sid=1, did=1))),
+        ('llm.database_security_report_stream requires tools_ai',
+         dict(method='get',
+              endpoint='llm.database_security_report_stream',
+              url_kwargs=dict(sid=1, did=1))),
+        ('llm.schema_security_report requires tools_ai',
+         dict(method='get', endpoint='llm.schema_security_report',
+              url_kwargs=dict(sid=1, did=1, scid=1))),
+        ('llm.schema_security_report_stream requires tools_ai',
+         dict(method='get', endpoint='llm.schema_security_report_stream',
+              url_kwargs=dict(sid=1, did=1, scid=1))),
+        ('llm.performance_report requires tools_ai',
+         dict(method='get', endpoint='llm.performance_report',
+              url_kwargs=dict(sid=1))),
+        ('llm.performance_report_stream requires tools_ai',
+         dict(method='get', endpoint='llm.performance_report_stream',
+              url_kwargs=dict(sid=1))),
+        ('llm.database_performance_report requires tools_ai',
+         dict(method='get', endpoint='llm.database_performance_report',
+              url_kwargs=dict(sid=1, did=1))),
+        ('llm.database_performance_report_stream requires tools_ai',
+         dict(method='get',
+              endpoint='llm.database_performance_report_stream',
+              url_kwargs=dict(sid=1, did=1))),
+        ('llm.database_design_report requires tools_ai',
+         dict(method='get', endpoint='llm.database_design_report',
+              url_kwargs=dict(sid=1, did=1))),
+        ('llm.database_design_report_stream requires tools_ai',
+         dict(method='get', endpoint='llm.database_design_report_stream',
+              url_kwargs=dict(sid=1, did=1))),
+        ('llm.schema_design_report requires tools_ai',
+         dict(method='get', endpoint='llm.schema_design_report',
+              url_kwargs=dict(sid=1, did=1, scid=1))),
+        ('llm.schema_design_report_stream requires tools_ai',
+         dict(method='get', endpoint='llm.schema_design_report_stream',
+              url_kwargs=dict(sid=1, did=1, scid=1))),
+
+        # --- Import/Export Servers: front-door-only gap ---
+        ('import_export_servers.get_servers requires '
+         'tools_import_export_servers',
+         dict(method='get', endpoint='import_export_servers.get_servers',
+              url_kwargs=dict())),
+        ('import_export_servers.load_servers requires '
+         'tools_import_export_servers',
+         dict(method='post', endpoint='import_export_servers.load_servers',
+              url_kwargs=dict())),
+
+        # --- Search Objects: front-door-only gap ---
+        ('search_objects.types requires tools_search_objects',
+         dict(method='get', endpoint='search_objects.types',
+              url_kwargs=dict(sid=1, did=1))),
+    ]
+
+    def setUp(self):
+        if not config.SERVER_MODE:
+            self.skipTest(
+                'Tool permission decorators are only exercisable in SERVER '
+                'mode; DESKTOP mode auto-authenticates as the '
+                'all-permissions DESKTOP_USER on every request.'
+            )
+
+    @create_user_wise_test_client(test_user_details)
+    def runTest(self):
+        # Build the URL through the URL map so we exercise the registered
+        # route exactly and don't hard-code blueprint prefixes.
+        with self.app.test_request_context():
+            url = flask.url_for(self.endpoint, **self.url_kwargs)
+
+        http = getattr(self.tester, self.method)
+        response = http(url, follow_redirects=False)
+
+        self.assertEqual(
+            response.status_code, 403,
+            'Route {0} ({1} {2}) did not enforce its tool permission for a '
+            'user lacking it: expected 403, got {3}. Body: {4!r}'.format(
+                self.endpoint, self.method.upper(), url,
+                response.status_code, response.data[:200]
+            )
+        )

web/pgadmin/tools/tests/test_tool_socket_permissions_required.py

@@ -0,0 +1,122 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""RBAC regression: Socket.IO tool handlers must enforce the permission.
+
+The HTTP routes of the schema diff, ERD and PSQL tools are guarded by
+@permissions_required, but those tools also expose Socket.IO event
+handlers (schema_diff 'compare_database'/'compare_schema', erd 'tables',
+the psql '/pty' namespace). Those handlers previously carried only
+@socket_login_required (or nothing at all), so an authenticated user who
+had been denied the tool could still drive them. They now use
+@socket_permissions_required, which refuses the connection when the user
+lacks the permission.
+
+This test connects to each namespace as a user with no roles (hence no
+tool permissions) and asserts that emitting a guarded event causes the
+handler to refuse and disconnect the socket, rather than executing.
+
+Skipped in DESKTOP mode, where every request is auto-authenticated as the
+all-permissions DESKTOP_USER.
+"""
+
+import sys
+
+import config
+
+from flask_socketio import ConnectionRefusedError
+
+from pgadmin.utils.route import BaseTestGenerator
+from pgadmin import socketio
+from regression.test_setup import config_data
+from regression.python_test_utils import test_utils as utils
+
+test_user_details = None
+if config.SERVER_MODE:
+    test_user_details = config_data['pgAdmin4_test_non_admin_credentials']
+
+
+class ToolSocketPermissionRequiredTestCase(BaseTestGenerator):
+    """A guarded Socket.IO handler must refuse a user lacking the tool
+    permission instead of running."""
+
+    scenarios = [
+        ('schema_diff compare_database requires tools_schema_diff',
+         dict(namespace='/schema_diff', event='compare_database',
+              params=dict(trans_id=1, source_sid=1, source_did=1,
+                          target_sid=1, target_did=1,
+                          ignore_owner=0, ignore_whitespaces=0,
+                          ignore_tablespace=0, ignore_grants=0))),
+        ('schema_diff compare_schema requires tools_schema_diff',
+         dict(namespace='/schema_diff', event='compare_schema',
+              params=dict(trans_id=1, source_sid=1, source_did=1,
+                          source_scid=1, target_sid=1, target_did=1,
+                          target_scid=1, ignore_owner=0,
+                          ignore_whitespaces=0, ignore_tablespace=0,
+                          ignore_grants=0))),
+        ('psql start_process requires tools_psql_tool',
+         dict(namespace='/pty', event='start_process',
+              params=dict(sid=1, did=1))),
+        ('psql socket_input requires tools_psql_tool',
+         dict(namespace='/pty', event='socket_input',
+              params=dict(key_name='a'))),
+        ('psql socket_set_role requires tools_psql_tool',
+         dict(namespace='/pty', event='socket_set_role',
+              params=dict(role='public'))),
+        ('psql resize requires tools_psql_tool',
+         dict(namespace='/pty', event='resize',
+              params=dict(rows=24, cols=80))),
+        ('erd tables requires tools_erd_tool',
+         dict(namespace='/erd', event='tables',
+              params=dict(trans_id=1, sid=1, did=1))),
+    ]
+
+    def setUp(self):
+        if not config.SERVER_MODE:
+            self.skipTest(
+                'Socket permission decorators are only exercisable in '
+                'SERVER mode; DESKTOP mode auto-authenticates as the '
+                'all-permissions DESKTOP_USER.'
+            )
+        if self.namespace == '/pty' and sys.platform == 'win32':
+            self.skipTest('PSQL is disabled on Windows.')
+        config.ENABLE_PSQL = True
+
+    def runTest(self):
+        # Log in a brand-new user that has no roles, therefore no tool
+        # permissions, and bind a Socket.IO test client to that session.
+        non_admin_client = utils.get_test_user(self, test_user_details)
+        self.assertIsNotNone(
+            non_admin_client, 'Could not create the non-admin test user.')
+
+        sclient = socketio.test_client(
+            self.app, namespace=self.namespace,
+            flask_test_client=non_admin_client)
+
+        # The namespace 'connect' handler is not permission-gated, so the
+        # connection itself is allowed; the guard is on the event handler.
+        self.assertTrue(
+            sclient.is_connected(self.namespace),
+            'Expected the namespace connection to succeed for an '
+            'authenticated user.')
+
+        # Emitting the guarded event must trip socket_permissions_required,
+        # which raises ConnectionRefusedError instead of running the
+        # handler. The flask-socketio test client surfaces that as an
+        # exception out of emit(); if the handler had run unguarded, no
+        # exception would be raised here.
+        try:
+            sclient.emit(self.event, self.params, namespace=self.namespace)
+        except ConnectionRefusedError:
+            pass
+        else:
+            self.fail(
+                'Socket handler {0} on {1} did not refuse a user lacking '
+                'the tool permission; the handler ran instead of raising '
+                'ConnectionRefusedError.'.format(self.event, self.namespace))

web/pgadmin/utils/driver/abstract.py

@@ -197,7 +197,7 @@ class BaseConnection(metaclass=ABCMeta):
 
     @abstractmethod
     def execute_2darray(self, query, params=None,
-                        formatted_exception_msg=False):
+                        formatted_exception_msg=False, prepare=None):
         pass
 
     @abstractmethod

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

@@ -830,7 +830,7 @@ WHERE db.datname = current_database()""")
                 25,
                 'Psycopg3 Cursor: {0}'.format(str(e)))
 
-    def __internal_blocking_execute(self, cur, query, params):
+    def __internal_blocking_execute(self, cur, query, params, prepare=None):
         """
         This function executes the query using cursor's execute function,
         but in case of asynchronous connection we need to wait for the
@@ -841,10 +841,17 @@ WHERE db.datname = current_database()""")
             cur: Cursor object
             query: SQL query to run.
             params: Extra parameters
+            prepare: force the query through PostgreSQL's PREPARE step
+                (extended query protocol) when True. Unlike the simple
+                query protocol, the server rejects more than one SQL
+                statement in a single Parse message, so this is used by
+                callers that must guarantee a single statement is
+                executed regardless of how a client-side SQL lexer
+                would have classified the text.
         """
 
         query = query.encode(self.python_encoding)
-        cur.execute(query, params)
+        cur.execute(query, params, prepare=prepare)
 
     def execute_on_server_as_csv(self, records=2000):
         """
@@ -1246,7 +1253,7 @@ WHERE db.datname = current_database()""")
         )
 
     def execute_2darray(self, query, params=None,
-                        formatted_exception_msg=False):
+                        formatted_exception_msg=False, prepare=None):
         status, cur = self.__cursor()
         self.row_count = 0
 
@@ -1270,14 +1277,16 @@ WHERE db.datname = current_database()""")
             )
         )
         try:
-            self.__internal_blocking_execute(cur, query, params)
+            self.__internal_blocking_execute(
+                cur, query, params, prepare=prepare
+            )
         except psycopg.Error as pe:
             cur.close_cursor()
             if not self.connected() and self.auto_reconnect and \
                     not self.reconnecting:
                 return self.__attempt_execution_reconnect(
                     self.execute_2darray, query, params,
-                    formatted_exception_msg
+                    formatted_exception_msg, prepare
                 )
             errmsg = self._formatted_exception_msg(pe, formatted_exception_msg)
             current_app.logger.error(

web/pgadmin/utils/master_password.py

@@ -179,16 +179,44 @@ def process_masterpass_disabled():
 
 def get_master_password_from_master_hook():
     """
-    This method executes specified command & returns output.
-    :param command: Shell command with absolute path
-    :return: Output of command.
+    This method executes the configured MASTER_PASSWORD_HOOK command and
+    returns its output.
+
+    The hook is parsed into an argument vector and executed *without* a shell
+    (shell=False). The username (which may originate from an external
+    authentication source such as OAuth/OIDC, Kerberos or webserver auth, and
+    is therefore untrusted) is substituted for %u only after the trusted hook
+    string has been tokenised, so it is always confined to a single argument.
+    Any shell metacharacters it contains are then inert rather than
+    interpreted, which prevents OS command injection via the username.
+
+    :return: Output of the command, or None on failure.
     """
+    import shlex
     import subprocess
+
     cmd = config.MASTER_PASSWORD_HOOK
-    command = cmd.replace('%u', current_user.username) \
-        if '%u' in cmd else cmd
+    if not cmd:
+        return None
     try:
-        p = subprocess.Popen(command, stdout=subprocess.PIPE, shell=True)
+        # Tokenise the admin-configured hook first, then substitute the
+        # untrusted username into the individual arguments. Substituting
+        # before tokenising would let metacharacters in the username alter
+        # the command, so the order here is security-critical.
+        #
+        # posix=True strips quotes correctly on both platforms (unlike
+        # posix=False, which never strips quotes at all and breaks any
+        # quoted path or argument on Windows). escape='' disables backslash
+        # escaping so Windows paths (e.g. C:\Program Files\hook.exe) are not
+        # mangled.
+        lex = shlex.shlex(cmd, posix=True)
+        lex.whitespace_split = True
+        lex.escape = ''
+        args = [token.replace('%u', current_user.username) for token in lex]
+        if not args:
+            return None
+
+        p = subprocess.Popen(args, stdout=subprocess.PIPE, shell=False)
         out, err = p.communicate()
         if p.returncode == 0:
             output = out.decode() if hasattr(out, 'decode') else out
@@ -196,7 +224,7 @@ def get_master_password_from_master_hook():
             return output
         else:
             error = "Command '{0}' failed, exit-code={1} error = {2}".format(
-                command, p.returncode, str(err))
+                cmd, p.returncode, str(err))
             current_app.logger.error(error)
     except Exception as e:
         current_app.logger.exception(

web/pgadmin/utils/tests/test_master_password_hook.py

@@ -0,0 +1,163 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""Regression tests for the MASTER_PASSWORD_HOOK OS command injection.
+
+The hook substitutes the externally-supplied username for %u. The fix
+tokenises the (trusted) hook string first and only then substitutes the
+(untrusted) username into the individual arguments, executing the result
+with shell=False. As a result, shell metacharacters in the username are
+passed through as a single, literal argv element rather than being
+interpreted by a shell."""
+
+import os
+import shutil
+import tempfile
+from unittest.mock import patch, MagicMock
+
+from pgadmin.utils.route import BaseTestGenerator
+import pgadmin.utils.master_password as mp
+
+
+class _FakeProc:
+    """Minimal stand-in for a subprocess.Popen result."""
+    returncode = 0
+
+    def communicate(self):
+        return (b'secret-key', b'')
+
+
+class TestMasterPasswordHookConfinesUsername(BaseTestGenerator):
+    """The username, whatever shell syntax it contains, must reach the hook
+    as exactly one argv element, and the hook must run with shell=False. This
+    is verified by capturing the arguments passed to subprocess.Popen, so the
+    check is deterministic and platform-independent (nothing is executed)."""
+
+    scenarios = [
+        ('semicolon command separator',
+         dict(username='attacker; touch /tmp/x')),
+        ('command substitution $()',
+         dict(username='x$(touch /tmp/x)')),
+        ('backtick command substitution',
+         dict(username='x`touch /tmp/x`')),
+        ('pipe to another command',
+         dict(username='x | touch /tmp/x')),
+        ('logical-and chain',
+         dict(username='x && touch /tmp/x')),
+        ('embedded newline',
+         dict(username='x\ntouch /tmp/x')),
+        ('embedded whitespace only',
+         dict(username='first last')),
+    ]
+
+    def runTest(self):
+        captured = {}
+
+        def fake_popen(args, *a, **kw):
+            captured['args'] = args
+            captured['shell'] = kw.get('shell')
+            return _FakeProc()
+
+        fake_user = MagicMock()
+        fake_user.username = self.username
+
+        with patch.object(mp, 'current_user', fake_user), \
+                patch.object(mp.config, 'MASTER_PASSWORD_HOOK',
+                             '/opt/get-secret %u'), \
+                patch('subprocess.Popen', side_effect=fake_popen):
+            output = mp.get_master_password_from_master_hook()
+
+        # A shell must never be involved.
+        self.assertIs(captured.get('shell'), False)
+        # The command must be passed as an argument vector, not a string.
+        self.assertIsInstance(captured.get('args'), list)
+        # The hook program token is preserved untouched as argv[0].
+        self.assertEqual(captured['args'][0], '/opt/get-secret')
+        # The entire username lands in a single argv element ...
+        self.assertEqual(captured['args'][1], self.username)
+        # ... and nothing was split out into extra tokens.
+        self.assertEqual(len(captured['args']), 2)
+        # The hook output is still returned to the caller.
+        self.assertEqual(output, 'secret-key')
+
+
+class TestMasterPasswordHookQuotedPath(BaseTestGenerator):
+    """A quoted hook program path (needed when the path contains spaces,
+    e.g. 'C:\\Program Files\\hook.exe' on Windows) must be tokenised as a
+    single argv element with the quotes stripped, and any backslashes in
+    the path must survive untouched."""
+
+    scenarios = [
+        ('quoted path with spaces, posix-style',
+         dict(cmd='"/opt/my secrets/get-secret" %u',
+              expected_argv0='/opt/my secrets/get-secret')),
+        ('quoted windows path with backslashes and spaces',
+         dict(cmd='"C:\\Program Files\\hook.exe" %u',
+              expected_argv0='C:\\Program Files\\hook.exe')),
+        ('unquoted windows path, no spaces',
+         dict(cmd='C:\\hook.exe %u', expected_argv0='C:\\hook.exe')),
+    ]
+
+    def runTest(self):
+        captured = {}
+
+        def fake_popen(args, *a, **kw):
+            captured['args'] = args
+            return _FakeProc()
+
+        fake_user = MagicMock()
+        fake_user.username = 'alice'
+
+        with patch.object(mp, 'current_user', fake_user), \
+                patch.object(mp.config, 'MASTER_PASSWORD_HOOK', self.cmd), \
+                patch('subprocess.Popen', side_effect=fake_popen):
+            mp.get_master_password_from_master_hook()
+
+        self.assertEqual(captured['args'][0], self.expected_argv0)
+        self.assertEqual(captured['args'][1], 'alice')
+        self.assertEqual(len(captured['args']), 2)
+
+
+class TestMasterPasswordHookNoShellExecution(BaseTestGenerator):
+    """End-to-end PoC regression (POSIX only): a username containing shell
+    syntax must not cause command execution. Mirrors the reported proof of
+    concept using a harmless marker file: if a shell were involved, the ';'
+    would start a second command that creates the marker."""
+
+    def runTest(self):
+        if os.name == 'nt':
+            self.skipTest('POSIX shell-injection PoC; skipped on Windows.')
+
+        echo = shutil.which('echo')
+        if echo is None:
+            self.skipTest('echo binary not available on PATH.')
+
+        tmpdir = tempfile.mkdtemp(prefix='pgadmin-hook-test-')
+        marker = os.path.join(tmpdir, 'master_hook_rce')
+        username = 'attacker; touch {0}; #'.format(marker)
+
+        fake_user = MagicMock()
+        fake_user.username = username
+
+        try:
+            with patch.object(mp, 'current_user', fake_user), \
+                    patch.object(mp.config, 'MASTER_PASSWORD_HOOK',
+                                 '{0} hook-%u'.format(echo)):
+                output = mp.get_master_password_from_master_hook()
+
+            self.assertFalse(
+                os.path.exists(marker),
+                'Injected command executed: the marker file was created, '
+                'which means the username was interpreted by a shell.')
+            # The username was handled as inert data and echoed verbatim.
+            self.assertEqual(output, 'hook-' + username)
+        finally:
+            if os.path.exists(marker):
+                os.remove(marker)
+            os.rmdir(tmpdir)