codekingpro/portable-devtools

1# dialects/sqlite/pysqlcipher.py
2# Copyright (C) 2005-2024 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('sqlite+pysqlcipher://:testing@/foo.db?cipher=aes-256-cfb&kdf_iter=64000')
72
73.. warning:: Previous versions of sqlalchemy did not take into consideration
74   the encryption-related pragmas passed in the url string, that were silently
75   ignored. This may cause errors when opening files saved by a
76   previous sqlalchemy version if the encryption options do not match.
77
78
79Pooling Behavior
80----------------
81
82The driver makes a change to the default pool behavior of pysqlite
83as described in :ref:`pysqlite_threading_pooling`.   The pysqlcipher driver
84has been observed to be significantly slower on connection than the
85pysqlite driver, most likely due to the encryption overhead, so the
86dialect here defaults to using the :class:`.SingletonThreadPool`
87implementation,
88instead of the :class:`.NullPool` pool used by pysqlite.  As always, the pool
89implementation is entirely configurable using the
90:paramref:`_sa.create_engine.poolclass` parameter; the :class:`.
91StaticPool` may
92be more feasible for single-threaded use, or :class:`.NullPool` may be used
93to prevent unencrypted connections from being held open for long periods of
94time, at the expense of slower startup time for new connections.
95
96
97"""  # noqa
98
99from .pysqlite import SQLiteDialect_pysqlite
100from ... import pool
101
102
103class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite):
104    driver = "pysqlcipher"
105    supports_statement_cache = True
106
107    pragmas = ("kdf_iter", "cipher", "cipher_page_size", "cipher_use_hmac")
108
109    @classmethod
110    def import_dbapi(cls):
111        try:
112            import sqlcipher3 as sqlcipher
113        except ImportError:
114            pass
115        else:
116            return sqlcipher
117
118        from pysqlcipher3 import dbapi2 as sqlcipher
119
120        return sqlcipher
121
122    @classmethod
123    def get_pool_class(cls, url):
124        return pool.SingletonThreadPool
125
126    def on_connect_url(self, url):
127        super_on_connect = super().on_connect_url(url)
128
129        # pull the info we need from the URL early.  Even though URL
130        # is immutable, we don't want any in-place changes to the URL
131        # to affect things
132        passphrase = url.password or ""
133        url_query = dict(url.query)
134
135        def on_connect(conn):
136            cursor = conn.cursor()
137            cursor.execute('pragma key="%s"' % passphrase)
138            for prag in self.pragmas:
139                value = url_query.get(prag, None)
140                if value is not None:
141                    cursor.execute('pragma %s="%s"' % (prag, value))
142            cursor.close()
143
144            if super_on_connect:
145                super_on_connect(conn)
146
147        return on_connect
148
149    def create_connect_args(self, url):
150        plain_url = url._replace(password=None)
151        plain_url = plain_url.difference_update_query(self.pragmas)
152        return super().create_connect_args(plain_url)
153
154
155dialect = SQLiteDialect_pysqlcipher
156