Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pymysql/_auth.py: 16%

1""" 

2Implements auth methods 

3""" 

4 

5from .err import OperationalError 

6 

7 

8try: 

9 from cryptography.hazmat.backends import default_backend 

10 from cryptography.hazmat.primitives import serialization, hashes 

11 from cryptography.hazmat.primitives.asymmetric import padding 

12 

13 _have_cryptography = True 

14except ImportError: 

15 _have_cryptography = False 

16 

17from functools import partial 

18import hashlib 

19 

20 

21DEBUG = False 

22SCRAMBLE_LENGTH = 20 

23sha1_new = partial(hashlib.new, "sha1") 

24 

25 

26# mysql_native_password 

27# https://dev.mysql.com/doc/internals/en/secure-password-authentication.html#packet-Authentication::Native41 

28 

29 

30def scramble_native_password(password, message): 

31 """Scramble used for mysql_native_password""" 

32 if not password: 

33 return b"" 

34 

35 stage1 = sha1_new(password).digest() 

36 stage2 = sha1_new(stage1).digest() 

37 s = sha1_new() 

38 s.update(message[:SCRAMBLE_LENGTH]) 

39 s.update(stage2) 

40 result = s.digest() 

41 return _my_crypt(result, stage1) 

42 

43 

44def _my_crypt(message1, message2): 

45 result = bytearray(message1) 

46 

47 for i in range(len(result)): 

48 result[i] ^= message2[i] 

49 

50 return bytes(result) 

51 

52 

53# MariaDB's client_ed25519-plugin 

54# https://mariadb.com/kb/en/library/connection/#client_ed25519-plugin 

55 

56_nacl_bindings = False 

57 

58 

59def _init_nacl(): 

60 global _nacl_bindings 

61 try: 

62 from nacl import bindings 

63 

64 _nacl_bindings = bindings 

65 except ImportError: 

66 raise RuntimeError( 

67 "'pynacl' package is required for ed25519_password auth method" 

68 ) 

69 

70 

71def _scalar_clamp(s32): 

72 ba = bytearray(s32) 

73 ba0 = bytes(bytearray([ba[0] & 248])) 

74 ba31 = bytes(bytearray([(ba[31] & 127) | 64])) 

75 return ba0 + bytes(s32[1:31]) + ba31 

76 

77 

78def ed25519_password(password, scramble): 

79 """Sign a random scramble with elliptic curve Ed25519. 

80 

81 Secret and public key are derived from password. 

82 """ 

83 # variable names based on rfc8032 section-5.1.6 

84 # 

85 if not _nacl_bindings: 

86 _init_nacl() 

87 

88 # h = SHA512(password) 

89 h = hashlib.sha512(password).digest() 

90 

91 # s = prune(first_half(h)) 

92 s = _scalar_clamp(h[:32]) 

93 

94 # r = SHA512(second_half(h) || M) 

95 r = hashlib.sha512(h[32:] + scramble).digest() 

96 

97 # R = encoded point [r]B 

98 r = _nacl_bindings.crypto_core_ed25519_scalar_reduce(r) 

99 R = _nacl_bindings.crypto_scalarmult_ed25519_base_noclamp(r) 

100 

101 # A = encoded point [s]B 

102 A = _nacl_bindings.crypto_scalarmult_ed25519_base_noclamp(s) 

103 

104 # k = SHA512(R || A || M) 

105 k = hashlib.sha512(R + A + scramble).digest() 

106 

107 # S = (k * s + r) mod L 

108 k = _nacl_bindings.crypto_core_ed25519_scalar_reduce(k) 

109 ks = _nacl_bindings.crypto_core_ed25519_scalar_mul(k, s) 

110 S = _nacl_bindings.crypto_core_ed25519_scalar_add(ks, r) 

111 

112 # signature = R || S 

113 return R + S 

114 

115 

116# sha256_password 

117 

118 

119def _roundtrip(conn, send_data): 

120 conn.write_packet(send_data) 

121 pkt = conn._read_packet() 

122 pkt.check_error() 

123 return pkt 

124 

125 

126def _xor_password(password, salt): 

127 # Trailing NUL character will be added in Auth Switch Request. 

128 # See https://github.com/mysql/mysql-server/blob/7d10c82196c8e45554f27c00681474a9fb86d137/sql/auth/sha2_password.cc#L939-L945 

129 salt = salt[:SCRAMBLE_LENGTH] 

130 password_bytes = bytearray(password) 

131 # salt = bytearray(salt) # for PY2 compat. 

132 salt_len = len(salt) 

133 for i in range(len(password_bytes)): 

134 password_bytes[i] ^= salt[i % salt_len] 

135 return bytes(password_bytes) 

136 

137 

138def sha2_rsa_encrypt(password, salt, public_key): 

139 """Encrypt password with salt and public_key. 

140 

141 Used for sha256_password and caching_sha2_password. 

142 """ 

143 if not _have_cryptography: 

144 raise RuntimeError( 

145 "'cryptography' package is required for sha256_password or" 

146 + " caching_sha2_password auth methods" 

147 ) 

148 message = _xor_password(password + b"\0", salt) 

149 rsa_key = serialization.load_pem_public_key(public_key, default_backend()) 

150 return rsa_key.encrypt( 

151 message, 

152 padding.OAEP( 

153 mgf=padding.MGF1(algorithm=hashes.SHA1()), 

154 algorithm=hashes.SHA1(), 

155 label=None, 

156 ), 

157 ) 

158 

159 

160def sha256_password_auth(conn, pkt): 

161 if conn._secure: 

162 if DEBUG: 

163 print("sha256: Sending plain password") 

164 data = conn.password + b"\0" 

165 return _roundtrip(conn, data) 

166 

167 if pkt.is_auth_switch_request(): 

168 conn.salt = pkt.read_all() 

169 if conn.salt.endswith(b"\0"): 

170 conn.salt = conn.salt[:-1] 

171 if not conn.server_public_key and conn.password: 

172 # Request server public key 

173 if DEBUG: 

174 print("sha256: Requesting server public key") 

175 pkt = _roundtrip(conn, b"\1") 

176 

177 if pkt.is_extra_auth_data(): 

178 conn.server_public_key = pkt._data[1:] 

179 if DEBUG: 

180 print("Received public key:\n", conn.server_public_key.decode("ascii")) 

181 

182 if conn.password: 

183 if not conn.server_public_key: 

184 raise OperationalError("Couldn't receive server's public key") 

185 

186 data = sha2_rsa_encrypt(conn.password, conn.salt, conn.server_public_key) 

187 else: 

188 data = b"" 

189 

190 return _roundtrip(conn, data) 

191 

192 

193def scramble_caching_sha2(password, nonce): 

194 # (bytes, bytes) -> bytes 

195 """Scramble algorithm used in cached_sha2_password fast path. 

196 

197 XOR(SHA256(password), SHA256(SHA256(SHA256(password)), nonce)) 

198 """ 

199 if not password: 

200 return b"" 

201 

202 p1 = hashlib.sha256(password).digest() 

203 p2 = hashlib.sha256(p1).digest() 

204 p3 = hashlib.sha256(p2 + nonce).digest() 

205 

206 res = bytearray(p1) 

207 for i in range(len(p3)): 

208 res[i] ^= p3[i] 

209 

210 return bytes(res) 

211 

212 

213def caching_sha2_password_auth(conn, pkt): 

214 # No password fast path 

215 if not conn.password: 

216 return _roundtrip(conn, b"") 

217 

218 if pkt.is_auth_switch_request(): 

219 # Try from fast auth 

220 conn.salt = pkt.read_all() 

221 if conn.salt.endswith(b"\0"): # str.removesuffix is available in 3.9 

222 conn.salt = conn.salt[:-1] 

223 if DEBUG: 

224 print(f"caching sha2: Trying fast path. salt={conn.salt.hex()!r}") 

225 scrambled = scramble_caching_sha2(conn.password, conn.salt) 

226 pkt = _roundtrip(conn, scrambled) 

227 # else: fast auth is tried in initial handshake 

228 

229 if not pkt.is_extra_auth_data(): 

230 raise OperationalError( 

231 "caching sha2: Unknown packet for fast auth: %s" % pkt._data[:1] 

232 ) 

233 

234 # magic numbers: 

235 # 2 - request public key 

236 # 3 - fast auth succeeded 

237 # 4 - need full auth 

238 

239 pkt.advance(1) 

240 n = pkt.read_uint8() 

241 

242 if n == 3: 

243 if DEBUG: 

244 print("caching sha2: succeeded by fast path.") 

245 pkt = conn._read_packet() 

246 pkt.check_error() # pkt must be OK packet 

247 return pkt 

248 

249 if n != 4: 

250 raise OperationalError("caching sha2: Unknown result for fast auth: %s" % n) 

251 

252 if DEBUG: 

253 print("caching sha2: Trying full auth...") 

254 

255 if conn._secure: 

256 if DEBUG: 

257 print("caching sha2: Sending plain password via secure connection") 

258 return _roundtrip(conn, conn.password + b"\0") 

259 

260 if not conn.server_public_key: 

261 pkt = _roundtrip(conn, b"\x02") # Request public key 

262 if not pkt.is_extra_auth_data(): 

263 raise OperationalError( 

264 "caching sha2: Unknown packet for public key: %s" % pkt._data[:1] 

265 ) 

266 

267 conn.server_public_key = pkt._data[1:] 

268 if DEBUG: 

269 print(conn.server_public_key.decode("ascii")) 

270 

271 data = sha2_rsa_encrypt(conn.password, conn.salt, conn.server_public_key) 

272 pkt = _roundtrip(conn, data)