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

1# sqlite/pysqlcipher.py 

2# Copyright (C) 2005-2023 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 

8""" 

9.. dialect:: sqlite+pysqlcipher 

10 :name: pysqlcipher 

11 :dbapi: sqlcipher 3 or pysqlcipher 

12 :connectstring: sqlite+pysqlcipher://:passphrase@/file_path[?kdf_iter=<iter>] 

13 

14 Dialect for support of DBAPIs that make use of the 

15 `SQLCipher <https://www.zetetic.net/sqlcipher>`_ backend. 

16 

17 

18Driver 

19------ 

20 

21Current dialect selection logic is: 

22 

23* If the :paramref:`_sa.create_engine.module` parameter supplies a DBAPI module, 

24 that module is used. 

25* Otherwise for Python 3, choose https://pypi.org/project/sqlcipher3/ 

26* If not available, fall back to https://pypi.org/project/pysqlcipher3/ 

27* For Python 2, https://pypi.org/project/pysqlcipher/ is used. 

28 

29.. warning:: The ``pysqlcipher3`` and ``pysqlcipher`` DBAPI drivers are no 

30 longer maintained; the ``sqlcipher3`` driver as of this writing appears 

31 to be current. For future compatibility, any pysqlcipher-compatible DBAPI 

32 may be used as follows:: 

33 

34 import sqlcipher_compatible_driver 

35 

36 from sqlalchemy import create_engine 

37 

38 e = create_engine( 

39 "sqlite+pysqlcipher://:password@/dbname.db", 

40 module=sqlcipher_compatible_driver 

41 ) 

42 

43These drivers make use of the SQLCipher engine. This system essentially 

44introduces new PRAGMA commands to SQLite which allows the setting of a 

45passphrase and other encryption parameters, allowing the database file to be 

46encrypted. 

47 

48 

49Connect Strings 

50--------------- 

51 

52The format of the connect string is in every way the same as that 

53of the :mod:`~sqlalchemy.dialects.sqlite.pysqlite` driver, except that the 

54"password" field is now accepted, which should contain a passphrase:: 

55 

56 e = create_engine('sqlite+pysqlcipher://:testing@/foo.db') 

57 

58For an absolute file path, two leading slashes should be used for the 

59database name:: 

60 

61 e = create_engine('sqlite+pysqlcipher://:testing@//path/to/foo.db') 

62 

63A selection of additional encryption-related pragmas supported by SQLCipher 

64as documented at https://www.zetetic.net/sqlcipher/sqlcipher-api/ can be passed 

65in the query string, and will result in that PRAGMA being called for each 

66new connection. Currently, ``cipher``, ``kdf_iter`` 

67``cipher_page_size`` and ``cipher_use_hmac`` are supported:: 

68 

69 e = create_engine('sqlite+pysqlcipher://:testing@/foo.db?cipher=aes-256-cfb&kdf_iter=64000') 

70 

71.. warning:: Previous versions of sqlalchemy did not take into consideration 

72 the encryption-related pragmas passed in the url string, that were silently 

73 ignored. This may cause errors when opening files saved by a 

74 previous sqlalchemy version if the encryption options do not match. 

75 

76 

77Pooling Behavior 

78---------------- 

79 

80The driver makes a change to the default pool behavior of pysqlite 

81as described in :ref:`pysqlite_threading_pooling`. The pysqlcipher driver 

82has been observed to be significantly slower on connection than the 

83pysqlite driver, most likely due to the encryption overhead, so the 

84dialect here defaults to using the :class:`.SingletonThreadPool` 

85implementation, 

86instead of the :class:`.NullPool` pool used by pysqlite. As always, the pool 

87implementation is entirely configurable using the 

88:paramref:`_sa.create_engine.poolclass` parameter; the :class:`. 

89StaticPool` may 

90be more feasible for single-threaded use, or :class:`.NullPool` may be used 

91to prevent unencrypted connections from being held open for long periods of 

92time, at the expense of slower startup time for new connections. 

93 

94 

95""" # noqa 

96 

97from __future__ import absolute_import 

98 

99from .pysqlite import SQLiteDialect_pysqlite 

100from ... import pool 

101from ... import util 

102 

103 

104class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite): 

105 driver = "pysqlcipher" 

106 supports_statement_cache = True 

107 

108 pragmas = ("kdf_iter", "cipher", "cipher_page_size", "cipher_use_hmac") 

109 

110 @classmethod 

111 def dbapi(cls): 

112 if util.py3k: 

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 else: 

123 from pysqlcipher import dbapi2 as sqlcipher 

124 

125 return sqlcipher 

126 

127 @classmethod 

128 def get_pool_class(cls, url): 

129 return pool.SingletonThreadPool 

130 

131 def on_connect_url(self, url): 

132 super_on_connect = super( 

133 SQLiteDialect_pysqlcipher, self 

134 ).on_connect_url(url) 

135 

136 # pull the info we need from the URL early. Even though URL 

137 # is immutable, we don't want any in-place changes to the URL 

138 # to affect things 

139 passphrase = url.password or "" 

140 url_query = dict(url.query) 

141 

142 def on_connect(conn): 

143 cursor = conn.cursor() 

144 cursor.execute('pragma key="%s"' % passphrase) 

145 for prag in self.pragmas: 

146 value = url_query.get(prag, None) 

147 if value is not None: 

148 cursor.execute('pragma %s="%s"' % (prag, value)) 

149 cursor.close() 

150 

151 if super_on_connect: 

152 super_on_connect(conn) 

153 

154 return on_connect 

155 

156 def create_connect_args(self, url): 

157 plain_url = url._replace(password=None) 

158 plain_url = plain_url.difference_update_query(self.pragmas) 

159 return super(SQLiteDialect_pysqlcipher, self).create_connect_args( 

160 plain_url 

161 ) 

162 

163 

164dialect = SQLiteDialect_pysqlcipher