From: Federico Caselli <cfederico87@gmail.com>
Date: Mon, 13 Apr 2026 20:26:48 +0000 (+0200)
Subject: Improve escaping in pysqlcipher
X-Git-Tag: rel_2_1_0b2~6
X-Git-Url: http://git.ipfire.org/index.cgi?a=commitdiff_plain;h=87f17c2e59cd27e20e7badc685c482e788fb2711;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

Improve escaping in pysqlcipher

Escape key and pragma values when utilizing the pysqlcipher dialect.

Fixes: #13230
Change-Id: I7583577a3e00e2f2986e50f32136a9ef005eb28a
---

diff --git a/doc/build/changelog/unreleased_20/13230.rst b/doc/build/changelog/unreleased_20/13230.rst
new file mode 100644
index 0000000000..a9cb60e670
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/13230.rst
@@ -0,0 +1,5 @@
+.. change::
+    :tags: bug, sqlite
+    :tickets: 13230
+
+    Escape key and pragma values when utilizing the pysqlcipher dialect.
diff --git a/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py b/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py
index 1a4c251a47..7f2c3d4b79 100644
--- a/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py
+++ b/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py
@@ -131,16 +131,20 @@ class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite):
         # pull the info we need from the URL early.  Even though URL
         # is immutable, we don't want any in-place changes to the URL
         # to affect things
-        passphrase = url.password or ""
-        url_query = dict(url.query)
+        ip = self.identifier_preparer
+        passphrase = ip.quote_identifier(url.password or "")
+        query_pragmas = {
+            prag: ip.quote_identifier(url.query[prag])
+            for prag in self.pragmas
+            if url.query.get(prag) is not None
+        }
 
         def on_connect(conn):
             cursor = conn.cursor()
-            cursor.execute('pragma key="%s"' % passphrase)
-            for prag in self.pragmas:
-                value = url_query.get(prag, None)
-                if value is not None:
-                    cursor.execute('pragma %s="%s"' % (prag, value))
+            cursor.execute(f"pragma key={passphrase}")
+            for prag, value in query_pragmas.items():
+                cursor.execute(f"pragma {prag}={value}")
+            print(query_pragmas)
             cursor.close()
 
             if super_on_connect:
diff --git a/test/dialect/sqlite/test_dialect.py b/test/dialect/sqlite/test_dialect.py
index 27392fc079..5deb799ea7 100644
--- a/test/dialect/sqlite/test_dialect.py
+++ b/test/dialect/sqlite/test_dialect.py
@@ -1,6 +1,7 @@
 """SQLite-specific tests."""
 
 import os
+from tempfile import mkstemp
 
 from sqlalchemy import and_
 from sqlalchemy import Column
@@ -197,12 +198,35 @@ class DialectTest(
     @testing.only_on("sqlite+pysqlcipher")
     def test_pysqlcipher_connects(self):
         """test #6586"""
-        str_url = str(testing.db.url)
+        f, name = mkstemp()
+        str_url = str(testing.db.url.set(database=name))
         e = create_engine(str_url)
 
         with e.connect() as conn:
             eq_(conn.scalar(text("select 1")), 1)
 
+        e.dispose()
+        os.close(f)
+        os.unlink(name)
+
+    @testing.only_on("sqlite+pysqlcipher")
+    def test_pysqlcipher_escaped(self):
+        """test #6586"""
+        f, name = mkstemp()
+        new_url = testing.db.url.set(
+            password='tes"ting',
+            database=name,
+            query={"cipher": 'aes-"select 1', "kdf_iter": "64000"},
+        )
+        e = create_engine(new_url)
+
+        with e.connect() as conn:
+            eq_(conn.scalar(text("select 1")), 1)
+
+        e.dispose()
+        os.close(f)
+        os.unlink(name)
+
     @testing.provide_metadata
     def test_extra_reserved_words(self, connection):
         """Tests reserved words in identifiers.