Fix Query Tool state save/restore failing on installations with a legacy SECRET_KEY. #10573 (#10574)

web/pgadmin/settings/__init__.py

@@ -24,6 +24,7 @@ from pgadmin.model import db, Setting, ApplicationState
 from pgadmin.utils.constants import MIMETYPE_APP_JS
 from .utils import get_dialog_type, get_file_type_setting
 from cryptography.fernet import Fernet
+import base64
 import hashlib
 
 MODULE_NAME = 'settings'
@@ -294,6 +295,25 @@ def get_file_format_setting():
                               info=get_file_type_setting(list(data.values())))
 
 
+def get_application_state_fernet():
+    """
+    Return the Fernet instance used to encrypt the stored application state.
+
+    SECRET_KEY is normally auto-generated as a valid Fernet key, but
+    installations that set SECRET_KEY in their configuration before it moved
+    into the configuration database have that arbitrary string stored
+    verbatim, which Fernet rejects. In that case, derive a valid key from it;
+    no state can have been stored with such a key, so nothing existing
+    becomes unreadable.
+    """
+    secret = current_app.config['SECRET_KEY']
+    key = secret.encode() if isinstance(secret, str) else secret
+    try:
+        return Fernet(key)
+    except ValueError:
+        return Fernet(base64.urlsafe_b64encode(hashlib.sha256(key).digest()))
+
+
 @blueprint.route(
     '/save_application_state',
     methods=["POST"], endpoint='save_application_state'
@@ -306,7 +326,7 @@ def save_application_state():
     """
     data = json.loads(request.data)
     trans_id = data['trans_id']
-    fernet = Fernet(current_app.config['SECRET_KEY'].encode())
+    fernet = get_application_state_fernet()
     tool_data = fernet.encrypt(json.dumps(data['tool_data']).encode())
     connection_info = data['connection_info'] \
         if 'connection_info' in data else None
@@ -363,7 +383,7 @@ def get_application_state():
     """
     Returns application state if any stored.
     """
-    fernet = Fernet(current_app.config['SECRET_KEY'].encode())
+    fernet = get_application_state_fernet()
     result = db.session \
         .query(ApplicationState) \
         .filter(ApplicationState.uid == current_user.id) \
@@ -404,7 +424,7 @@ def get_application_state():
     methods=["GET"], endpoint='get_tool_data')
 @pga_login_required
 def get_tool_data(trans_id):
-    fernet = Fernet(current_app.config['SECRET_KEY'].encode())
+    fernet = get_application_state_fernet()
     result = db.session \
         .query(ApplicationState) \
         .filter(ApplicationState.uid == current_user.id,

web/pgadmin/settings/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/settings/tests/test_application_state_fernet.py

@@ -0,0 +1,60 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+import base64
+import os
+
+from cryptography.fernet import Fernet
+
+from pgadmin.settings import get_application_state_fernet
+from pgadmin.utils.route import BaseTestGenerator
+
+
+VALID_KEY = base64.urlsafe_b64encode(os.urandom(32)).decode()
+
+
+class TestApplicationStateFernet(BaseTestGenerator):
+    """Check that the application state can be encrypted whatever form the
+    SECRET_KEY takes, including legacy keys that are not valid Fernet
+    keys."""
+
+    scenarios = [
+        ('A valid Fernet key is used as-is',
+         dict(secret_key=VALID_KEY, used_as_is=True)),
+        ('A legacy plain-text key is accepted',
+         dict(secret_key='my-old-secret-key', used_as_is=False)),
+        ('A legacy key of 44 characters that is not base64 is accepted',
+         dict(secret_key='!' * 44, used_as_is=False)),
+        ('A legacy bytes key is accepted',
+         dict(secret_key=b'my-old-secret-key', used_as_is=False)),
+    ]
+
+    def setUp(self):
+        self.saved_secret_key = self.app.config.get('SECRET_KEY')
+        self.app.config['SECRET_KEY'] = self.secret_key
+
+    def tearDown(self):
+        self.app.config['SECRET_KEY'] = self.saved_secret_key
+
+    def runTest(self):
+        with self.app.app_context():
+            fernet = get_application_state_fernet()
+            token = fernet.encrypt(b'{"sql": "SELECT 1"}')
+
+            # A fresh instance must decrypt what an earlier one encrypted,
+            # as state is saved and restored by separate requests.
+            self.assertEqual(
+                get_application_state_fernet().decrypt(token),
+                b'{"sql": "SELECT 1"}')
+
+            if self.used_as_is:
+                # Existing installations must still be able to read the
+                # state they have already stored.
+                self.assertEqual(
+                    Fernet(self.secret_key.encode()).decrypt(token),
+                    b'{"sql": "SELECT 1"}')