Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/dialects/sqlite/pysqlcipher.py: 36%

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