Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy_utils/types/password.py: 32%

1import weakref 

2 

3from sqlalchemy import types 

4from sqlalchemy.dialects import oracle, postgresql, sqlite 

5from sqlalchemy.ext.mutable import Mutable 

6 

7from ..exceptions import ImproperlyConfigured 

8from .scalar_coercible import ScalarCoercible 

9 

10passlib = None 

11try: 

12 import passlib 

13 from passlib.context import LazyCryptContext 

14except ImportError: 

15 pass 

16 

17 

18class Password(Mutable): 

19 @classmethod 

20 def coerce(cls, key, value): 

21 if isinstance(value, Password): 

22 return value 

23 

24 if isinstance(value, (str, bytes)): 

25 return cls(value, secret=True) 

26 

27 super().coerce(key, value) 

28 

29 def __init__(self, value, context=None, secret=False): 

30 # Store the hash (if it is one). 

31 self.hash = value if not secret else None 

32 

33 # Store the secret if we have one. 

34 self.secret = value if secret else None 

35 

36 # The hash should be bytes. 

37 if isinstance(self.hash, str): 

38 self.hash = self.hash.encode('utf8') 

39 

40 # Save weakref of the password context (if we have one) 

41 self.context = weakref.proxy(context) if context is not None else None 

42 

43 def __eq__(self, value): 

44 if self.hash is None or value is None: 

45 # If either the hash or the value is None, 

46 # equivalence depends on *all* values being None, 

47 # including the secret. 

48 return self.hash is None and self.secret is None and value is None 

49 

50 if isinstance(value, Password): 

51 # Comparing 2 hashes isn't very useful; but this equality 

52 # method breaks otherwise. 

53 return value.hash == self.hash 

54 

55 if self.context is None: 

56 # Compare 2 hashes again as we don't know how to validate. 

57 return value == self 

58 

59 if isinstance(value, (str, bytes)): 

60 valid, new = self.context.verify_and_update(value, self.hash) 

61 if valid and new: 

62 # New hash was calculated due to various reasons; stored one 

63 # wasn't optimal, etc. 

64 self.hash = new 

65 

66 # The hash should be bytes. 

67 if isinstance(self.hash, str): 

68 self.hash = self.hash.encode('utf8') 

69 self.changed() 

70 

71 return valid 

72 

73 return False 

74 

75 def __ne__(self, value): 

76 return not (self == value) 

77 

78 

79class PasswordType(ScalarCoercible, types.TypeDecorator): 

80 """ 

81 PasswordType hashes passwords as they come into the database and allows 

82 verifying them using a Pythonic interface. This Pythonic interface 

83 relies on setting up automatic data type coercion using the 

84 :func:`~sqlalchemy_utils.listeners.force_auto_coercion` function. 

85 

86 All keyword arguments (aside from max_length) are forwarded to the 

87 construction of a `passlib.context.LazyCryptContext` object, which 

88 also supports deferred configuration via the `onload` callback. 

89 

90 The following usage will create a password column that will 

91 automatically hash new passwords as `pbkdf2_sha512` but still compare 

92 passwords against pre-existing `md5_crypt` hashes. As passwords are 

93 compared; the password hash in the database will be updated to 

94 be `pbkdf2_sha512`. 

95 

96 :: 

97 

98 

99 class Model(Base): 

100 password = sa.Column(PasswordType( 

101 schemes=[ 

102 'pbkdf2_sha512', 

103 'md5_crypt' 

104 ], 

105 

106 deprecated=['md5_crypt'] 

107 )) 

108 

109 

110 Verifying password is as easy as: 

111 

112 :: 

113 

114 target = Model() 

115 target.password = 'b' 

116 # '$5$rounds=80000$H.............' 

117 

118 target.password == 'b' 

119 # True 

120 

121 

122 Lazy configuration of the type with Flask config: 

123 

124 :: 

125 

126 

127 import flask 

128 from sqlalchemy_utils import PasswordType, force_auto_coercion 

129 

130 force_auto_coercion() 

131 

132 class User(db.Model): 

133 __tablename__ = 'user' 

134 

135 password = db.Column( 

136 PasswordType( 

137 # The returned dictionary is forwarded to the CryptContext 

138 onload=lambda **kwargs: dict( 

139 schemes=flask.current_app.config['PASSWORD_SCHEMES'], 

140 **kwargs 

141 ), 

142 ), 

143 unique=False, 

144 nullable=False, 

145 ) 

146 

147 """ 

148 

149 impl = types.VARBINARY(1024) 

150 cache_ok = True 

151 

152 def __init__(self, max_length=None, **kwargs): 

153 # Fail if passlib is not found. 

154 if passlib is None: 

155 raise ImproperlyConfigured("'passlib' is required to use 'PasswordType'") 

156 

157 # Construct the passlib crypt context. 

158 self.context = LazyCryptContext(**kwargs) 

159 self._max_length = max_length 

160 

161 @property 

162 def hashing_method(self): 

163 return 'hash' if hasattr(self.context, 'hash') else 'encrypt' 

164 

165 @property 

166 def length(self): 

167 """Get column length.""" 

168 if self._max_length is None: 

169 self._max_length = self.calculate_max_length() 

170 

171 return self._max_length 

172 

173 def calculate_max_length(self): 

174 # Calculate the largest possible encoded password. 

175 # name + rounds + salt + hash + ($ * 4) of largest hash 

176 max_lengths = [1024] 

177 for name in self.context.schemes(): 

178 scheme = getattr(__import__('passlib.hash').hash, name) 

179 length = 4 + len(scheme.name) 

180 length += len(str(getattr(scheme, 'max_rounds', ''))) 

181 length += getattr(scheme, 'max_salt_size', 0) or 0 

182 length += getattr(scheme, 'encoded_checksum_size', scheme.checksum_size) 

183 max_lengths.append(length) 

184 

185 # Return the maximum calculated max length. 

186 return max(max_lengths) 

187 

188 def load_dialect_impl(self, dialect): 

189 if dialect.name == 'postgresql': 

190 # Use a BYTEA type for postgresql. 

191 impl = postgresql.BYTEA(self.length) 

192 elif dialect.name == 'oracle': 

193 # Use a RAW type for oracle. 

194 impl = oracle.RAW(self.length) 

195 elif dialect.name == 'sqlite': 

196 # Use a BLOB type for sqlite 

197 impl = sqlite.BLOB(self.length) 

198 else: 

199 # Use a VARBINARY for all other dialects. 

200 impl = types.VARBINARY(self.length) 

201 return dialect.type_descriptor(impl) 

202 

203 def process_bind_param(self, value, dialect): 

204 if isinstance(value, Password): 

205 # If were given a password secret; hash it. 

206 if value.secret is not None: 

207 return self._hash(value.secret).encode('utf8') 

208 

209 # Value has already been hashed. 

210 return value.hash 

211 

212 if isinstance(value, str): 

213 # Assume value has not been hashed. 

214 return self._hash(value).encode('utf8') 

215 

216 def process_result_value(self, value, dialect): 

217 if value is not None: 

218 return Password(value, self.context) 

219 

220 def _hash(self, value): 

221 return getattr(self.context, self.hashing_method)(value) 

222 

223 def _coerce(self, value): 

224 if value is None: 

225 return 

226 

227 if not isinstance(value, Password): 

228 # Hash the password using the default scheme. 

229 value = self._hash(value).encode('utf8') 

230 return Password(value, context=self.context) 

231 

232 else: 

233 # If were given a password object; ensure the context is right. 

234 value.context = weakref.proxy(self.context) 

235 

236 # If were given a password secret; hash it. 

237 if value.secret is not None: 

238 value.hash = self._hash(value.secret).encode('utf8') 

239 value.secret = None 

240 

241 return value 

242 

243 @property 

244 def python_type(self): 

245 return self.impl.type.python_type 

246 

247 

248Password.associate_with(PasswordType)