Fix Query Tool password re-prompt loop for unsaved passwords (#9091)

docs/en_US/release_notes_9_16.rst

@@ -40,6 +40,7 @@ Bug fixes
 *********
 
   | `Issue #6308 <https://github.com/pgadmin-org/pgadmin4/issues/6308>`_ -  Fix the infinite loading spinner after an idle database connection is silently dropped, by detecting stale connections and offering a reconnect dialog.
+  | `Issue #9091 <https://github.com/pgadmin-org/pgadmin4/issues/9091>`_ -  Fix the Query Tool re-prompting for an unsaved password in a loop and rejecting the re-entered password, by caching the entered password on the server manager when the primary connection is already established.
   | `Issue #9595 <https://github.com/pgadmin-org/pgadmin4/issues/9595>`_ -  Fix missing ALTER ... SET DEFAULT statements for inherited columns in the generated table SQL/EDIT script.
   | `Issue #9677 <https://github.com/pgadmin-org/pgadmin4/issues/9677>`_ -  Fix the Unlogged table toggle in table properties not generating any ALTER TABLE ... SET LOGGED/UNLOGGED statement.
   | `Issue #9828 <https://github.com/pgadmin-org/pgadmin4/issues/9828>`_ -  Fix tool calls failing against OpenAI-compatible providers that emit empty/null name, arguments, or id fields in streaming continuation deltas.

web/pgadmin/tools/sqleditor/__init__.py

@@ -46,6 +46,8 @@
 from pgadmin.utils.ajax import make_json_response, bad_request, \
     success_return, internal_server_error, service_unavailable, gone
 from pgadmin.utils.driver import get_driver
+from pgadmin.utils.crypto import encrypt
+from pgadmin.utils.master_password import get_crypt_key
 from pgadmin.utils.exception import ConnectionLost, SSHTunnelConnectionLost, \
     CryptKeyMissing, ObjectGone
 from pgadmin.browser.utils import underscore_escape
@@ -2681,6 +2683,16 @@ def connect_server(sid):
 
     conn = manager.connection()
     if conn.connected():
+        # The server's primary connection is already established.  However,
+        # individual tools (Query Tool, View/Edit Data, etc.) open their own
+        # connections and, when the password is not saved, rely on the
+        # password cached on the server manager.  If that cached password is
+        # missing (e.g. it was never persisted, or the tab was restored from
+        # a workspace) the tool prompts for the password.  Make sure the
+        # password the user just entered at that prompt is cached here so the
+        # tool's connection can use it, instead of being discarded and
+        # re-prompted in a loop.
+        _cache_manager_password_from_request(manager)
         return make_json_response(
             success=1,
             info=gettext("Server connected."),
@@ -2693,6 +2705,43 @@ def connect_server(sid):
     )
 
 
+def _cache_manager_password_from_request(manager):
+    """
+    Cache the password supplied with the current request (from a tool's
+    password prompt) onto the server manager, so that connections opened by
+    tools such as the Query Tool can reuse it without prompting again.
+
+    This is a no-op when no password is supplied or when the encryption key
+    is unavailable.  When a password is supplied it overwrites any cached
+    password, so a freshly entered credential (e.g. a regenerated, short-lived
+    cloud auth token) takes effect immediately.
+
+    This is best-effort: any failure (including malformed request data) is
+    logged and swallowed so it never turns the caller's "Server connected"
+    response into a 500 error.
+    """
+    try:
+        if request.form:
+            data = request.form
+        elif request.data:
+            data = json.loads(request.data)
+        else:
+            return
+
+        password = data.get('password', None)
+        if not password:
+            return
+
+        crypt_key_present, crypt_key = get_crypt_key()
+        if not crypt_key_present:
+            return
+
+        manager._update_password(encrypt(password, crypt_key))
+        manager.update_session()
+    except Exception as e:
+        current_app.logger.exception(e)
+
+
 @blueprint.route(
     '/filter_dialog/<int:trans_id>',
     methods=["PUT"], endpoint='set_filter_data'

web/pgadmin/tools/sqleditor/utils/tests/test_cache_manager_password.py

@@ -0,0 +1,107 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+from pgadmin.utils.route import BaseTestGenerator
+from pgadmin.utils.crypto import decrypt
+from unittest.mock import patch, MagicMock
+
+import pgadmin.tools.sqleditor as sqleditor
+
+CRYPT_KEY = 'test-crypt-key'
+LONG_TOKEN = (
+    'abc.rds.amazonaws.com:5432/?Action=connect&'
+    'X-Amz-Algorithm=AWS4-HMAC-SHA256&'
+    'X-Amz-Credential=ABC%2F20250820%2Fus-east-1&'
+    'X-Amz-Signature=deadbeef+slashes%2F%2F'
+) * 8
+
+
+class CacheManagerPasswordTest(BaseTestGenerator):
+    """
+    Regression test for issue #9091.
+
+    When a tool (Query Tool, View/Edit Data, etc.) prompts for a password
+    that was not saved, the entered password is POSTed to the connect_server
+    endpoint.  If the server's primary connection is already connected the
+    endpoint short-circuits; _cache_manager_password_from_request makes sure
+    the entered password is still cached on the manager (encrypted), so the
+    tool's connection can reuse it instead of being re-prompted in a loop.
+    """
+
+    scenarios = [
+        ('When a password is supplied it is encrypted and cached', dict(
+            form_data={'password': LONG_TOKEN},
+            crypt_key_present=True,
+            expect_cached=True,
+        )),
+        ('When an existing cached password is overwritten', dict(
+            form_data={'password': LONG_TOKEN},
+            crypt_key_present=True,
+            existing_password=b'stale-encrypted-token',
+            expect_cached=True,
+        )),
+        ('When no password is supplied it is a no-op', dict(
+            form_data={},
+            crypt_key_present=True,
+            expect_cached=False,
+        )),
+        ('When an empty password is supplied it is a no-op', dict(
+            form_data={'password': ''},
+            crypt_key_present=True,
+            expect_cached=False,
+        )),
+        ('When the crypt key is missing it is a no-op', dict(
+            form_data={'password': LONG_TOKEN},
+            crypt_key_present=False,
+            expect_cached=False,
+        )),
+        ('When the request body is malformed JSON it is a silent no-op', dict(
+            form_data={},
+            request_data=b'{not-valid-json',
+            crypt_key_present=True,
+            expect_cached=False,
+        )),
+    ]
+
+    def runTest(self):
+        manager = MagicMock()
+        manager.password = getattr(self, 'existing_password', None)
+
+        def _update_password(passwd):
+            manager.password = passwd
+
+        manager._update_password.side_effect = _update_password
+
+        mock_request = MagicMock()
+        mock_request.form = self.form_data
+        mock_request.data = getattr(self, 'request_data', None)
+
+        crypt_key = CRYPT_KEY if self.crypt_key_present else None
+
+        with patch.object(sqleditor, 'request', mock_request), \
+            patch.object(sqleditor, 'current_app', MagicMock()), \
+            patch.object(sqleditor, 'get_crypt_key',
+                         return_value=(self.crypt_key_present, crypt_key)):
+            sqleditor._cache_manager_password_from_request(manager)
+
+        if self.expect_cached:
+            self.assertTrue(manager._update_password.called)
+            self.assertTrue(manager.update_session.called)
+            # The cached value must be the encrypted form of the supplied
+            # password and must decrypt back to the original token intact.
+            cached = manager.password
+            self.assertIsNotNone(cached)
+            self.assertNotEqual(cached, self.form_data['password'])
+            decrypted = decrypt(cached, CRYPT_KEY)
+            if isinstance(decrypted, bytes):
+                decrypted = decrypted.decode()
+            self.assertEqual(decrypted, self.form_data['password'])
+        else:
+            self.assertFalse(manager._update_password.called)
+            self.assertFalse(manager.update_session.called)