codekingpro/portable-devtools

1# dialects/sqlite/pysqlcipher.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
8
9
10"""
11.. dialect:: sqlite+pysqlcipher
12    :name: pysqlcipher
13    :dbapi: sqlcipher 3 or pysqlcipher
14    :connectstring: sqlite+pysqlcipher://:passphrase@/file_path[?kdf_iter=<iter>]
15
16    Dialect for support of DBAPIs that make use of the
17    `SQLCipher <https://www.zetetic.net/sqlcipher>`_ backend.
18
19
20Driver
21------
22
23Current dialect selection logic is:
24
25* If the :paramref:`_sa.create_engine.module` parameter supplies a DBAPI module,
26  that module is used.
27* Otherwise for Python 3, choose https://pypi.org/project/sqlcipher3/
28* If not available, fall back to https://pypi.org/project/pysqlcipher3/
29* For Python 2, https://pypi.org/project/pysqlcipher/ is used.
30
31.. warning:: The ``pysqlcipher3`` and ``pysqlcipher`` DBAPI drivers are no
32   longer maintained; the ``sqlcipher3`` driver as of this writing appears
33   to be current.  For future compatibility, any pysqlcipher-compatible DBAPI
34   may be used as follows::
35
36        import sqlcipher_compatible_driver
37
38        from sqlalchemy import create_engine
39
40        e = create_engine(
41            "sqlite+pysqlcipher://:password@/dbname.db",
42            module=sqlcipher_compatible_driver,
43        )
44
45These drivers make use of the SQLCipher engine. This system essentially
46introduces new PRAGMA commands to SQLite which allows the setting of a
47passphrase and other encryption parameters, allowing the database file to be
48encrypted.
49
50
51Connect Strings
52---------------
53
54The format of the connect string is in every way the same as that
55of the :mod:`~sqlalchemy.dialects.sqlite.pysqlite` driver, except that the
56"password" field is now accepted, which should contain a passphrase::
57
58    e = create_engine("sqlite+pysqlcipher://:testing@/foo.db")
59
60For an absolute file path, two leading slashes should be used for the
61database name::
62
63    e = create_engine("sqlite+pysqlcipher://:testing@//path/to/foo.db")
64
65A selection of additional encryption-related pragmas supported by SQLCipher
66as documented at https://www.zetetic.net/sqlcipher/sqlcipher-api/ can be passed
67in the query string, and will result in that PRAGMA being called for each
68new connection.  Currently, ``cipher``, ``kdf_iter``
69``cipher_page_size`` and ``cipher_use_hmac`` are supported::
70
71    e = create_engine(
72        "sqlite+pysqlcipher://:testing@/foo.db?cipher=aes-256-cfb&kdf_iter=64000"
73    )
74
75.. warning:: Previous versions of sqlalchemy did not take into consideration
76   the encryption-related pragmas passed in the url string, that were silently
77   ignored. This may cause errors when opening files saved by a
78   previous sqlalchemy version if the encryption options do not match.
79
80
81Pooling Behavior
82----------------
83
84The driver makes a change to the default pool behavior of pysqlite
85as described in :ref:`pysqlite_threading_pooling`.   The pysqlcipher driver
86has been observed to be significantly slower on connection than the
87pysqlite driver, most likely due to the encryption overhead, so the
88dialect here defaults to using the :class:`.SingletonThreadPool`
89implementation,
90instead of the :class:`.NullPool` pool used by pysqlite.  As always, the pool
91implementation is entirely configurable using the
92:paramref:`_sa.create_engine.poolclass` parameter; the :class:`.
93StaticPool` may
94be more feasible for single-threaded use, or :class:`.NullPool` may be used
95to prevent unencrypted connections from being held open for long periods of
96time, at the expense of slower startup time for new connections.
97
98
99"""  # noqa
100
101from .pysqlite import SQLiteDialect_pysqlite
102from ... import pool
103
104
105class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite):
106    driver = "pysqlcipher"
107    supports_statement_cache = True
108
109    pragmas = ("kdf_iter", "cipher", "cipher_page_size", "cipher_use_hmac")
110
111    @classmethod
112    def import_dbapi(cls):
113        try:
114            import sqlcipher3 as sqlcipher
115        except ImportError:
116            pass
117        else:
118            return sqlcipher
119
120        from pysqlcipher3 import dbapi2 as sqlcipher
121
122        return sqlcipher
123
124    @classmethod
125    def get_pool_class(cls, url):
126        return pool.SingletonThreadPool
127
128    def on_connect_url(self, url):
129        super_on_connect = super().on_connect_url(url)
130
131        # pull the info we need from the URL early.  Even though URL
132        # is immutable, we don't want any in-place changes to the URL
133        # to affect things
134        passphrase = url.password or ""
135        url_query = dict(url.query)
136
137        def on_connect(conn):
138            cursor = conn.cursor()
139            cursor.execute('pragma key="%s"' % passphrase)
140            for prag in self.pragmas:
141                value = url_query.get(prag, None)
142                if value is not None:
143                    cursor.execute('pragma %s="%s"' % (prag, value))
144            cursor.close()
145
146            if super_on_connect:
147                super_on_connect(conn)
148
149        return on_connect
150
151    def create_connect_args(self, url):
152        plain_url = url._replace(password=None)
153        plain_url = plain_url.difference_update_query(self.pragmas)
154        return super().create_connect_args(plain_url)
155
156
157dialect = SQLiteDialect_pysqlcipher
158