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 

7try: 

8 from cryptography.hazmat.backends import default_backend 

9 from cryptography.hazmat.primitives import hashes, serialization 

10 from cryptography.hazmat.primitives.asymmetric import padding 

11 

12 _have_cryptography = True 

13except ImportError: 

14 _have_cryptography = False 

15 

16import hashlib 

17from functools import partial 

18 

19DEBUG = False 

20SCRAMBLE_LENGTH = 20 

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

22 

23 

24# mysql_native_password 

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

26 

27 

28def scramble_native_password(password, message): 

29 """Scramble used for mysql_native_password""" 

30 if not password: 

31 return b"" 

32 

33 stage1 = sha1_new(password).digest() 

34 stage2 = sha1_new(stage1).digest() 

35 s = sha1_new() 

36 s.update(message[:SCRAMBLE_LENGTH]) 

37 s.update(stage2) 

38 result = s.digest() 

39 return _my_crypt(result, stage1) 

40 

41 

42def _my_crypt(message1, message2): 

43 result = bytearray(message1) 

44 

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

46 result[i] ^= message2[i] 

47 

48 return bytes(result) 

49 

50 

51# MariaDB's client_ed25519-plugin 

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

53 

54_nacl_bindings = False 

55 

56 

57def _init_nacl(): 

58 global _nacl_bindings 

59 try: 

60 from nacl import bindings 

61 

62 _nacl_bindings = bindings 

63 except ImportError: 

64 raise RuntimeError( 

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

66 ) 

67 

68 

69def _scalar_clamp(s32): 

70 ba = bytearray(s32) 

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

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

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

74 

75 

76def ed25519_password(password, scramble): 

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

78 

79 Secret and public key are derived from password. 

80 """ 

81 # variable names based on rfc8032 section-5.1.6 

82 # 

83 if not _nacl_bindings: 

84 _init_nacl() 

85 

86 # h = SHA512(password) 

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

88 

89 # s = prune(first_half(h)) 

90 s = _scalar_clamp(h[:32]) 

91 

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

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

94 

95 # R = encoded point [r]B 

96 r = _nacl_bindings.crypto_core_ed25519_scalar_reduce(r) 

97 R = _nacl_bindings.crypto_scalarmult_ed25519_base_noclamp(r) 

98 

99 # A = encoded point [s]B 

100 A = _nacl_bindings.crypto_scalarmult_ed25519_base_noclamp(s) 

101 

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

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

104 

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

106 k = _nacl_bindings.crypto_core_ed25519_scalar_reduce(k) 

107 ks = _nacl_bindings.crypto_core_ed25519_scalar_mul(k, s) 

108 S = _nacl_bindings.crypto_core_ed25519_scalar_add(ks, r) 

109 

110 # signature = R || S 

111 return R + S 

112 

113 

114# sha256_password 

115 

116 

117def _roundtrip(conn, send_data): 

118 conn.write_packet(send_data) 

119 pkt = conn._read_packet() 

120 pkt.check_error() 

121 return pkt 

122 

123 

124def _xor_password(password, salt): 

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

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

127 salt = salt[:SCRAMBLE_LENGTH] 

128 password_bytes = bytearray(password) 

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

130 salt_len = len(salt) 

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

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

133 return bytes(password_bytes) 

134 

135 

136def sha2_rsa_encrypt(password, salt, public_key): 

137 """Encrypt password with salt and public_key. 

138 

139 Used for sha256_password and caching_sha2_password. 

140 """ 

141 if not _have_cryptography: 

142 raise RuntimeError( 

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

144 + " caching_sha2_password auth methods" 

145 ) 

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

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

148 return rsa_key.encrypt( 

149 message, 

150 padding.OAEP( 

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

152 algorithm=hashes.SHA1(), 

153 label=None, 

154 ), 

155 ) 

156 

157 

158def sha256_password_auth(conn, pkt): 

159 if conn._secure: 

160 if DEBUG: 

161 print("sha256: Sending plain password") 

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

163 return _roundtrip(conn, data) 

164 

165 if pkt.is_auth_switch_request(): 

166 conn.salt = pkt.read_all() 

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

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

169 if not conn.server_public_key and conn.password: 

170 # Request server public key 

171 if DEBUG: 

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

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

174 

175 if pkt.is_extra_auth_data(): 

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

177 if DEBUG: 

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

179 

180 if conn.password: 

181 if not conn.server_public_key: 

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

183 

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

185 else: 

186 data = b"" 

187 

188 return _roundtrip(conn, data) 

189 

190 

191def scramble_caching_sha2(password, nonce): 

192 # (bytes, bytes) -> bytes 

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

194 

195 XOR(SHA256(password), SHA256(SHA256(SHA256(password)), nonce)) 

196 """ 

197 if not password: 

198 return b"" 

199 

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

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

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

203 

204 res = bytearray(p1) 

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

206 res[i] ^= p3[i] 

207 

208 return bytes(res) 

209 

210 

211def caching_sha2_password_auth(conn, pkt): 

212 # No password fast path 

213 if not conn.password: 

214 return _roundtrip(conn, b"") 

215 

216 if pkt.is_auth_switch_request(): 

217 # Try from fast auth 

218 conn.salt = pkt.read_all() 

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

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

221 if DEBUG: 

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

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

224 pkt = _roundtrip(conn, scrambled) 

225 # else: fast auth is tried in initial handshake 

226 

227 if not pkt.is_extra_auth_data(): 

228 raise OperationalError( 

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

230 ) 

231 

232 # magic numbers: 

233 # 2 - request public key 

234 # 3 - fast auth succeeded 

235 # 4 - need full auth 

236 

237 pkt.advance(1) 

238 n = pkt.read_uint8() 

239 

240 if n == 3: 

241 if DEBUG: 

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

243 pkt = conn._read_packet() 

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

245 return pkt 

246 

247 if n != 4: 

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

249 

250 if DEBUG: 

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

252 

253 if conn._secure: 

254 if DEBUG: 

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

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

257 

258 if not conn.server_public_key: 

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

260 if not pkt.is_extra_auth_data(): 

261 raise OperationalError( 

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

263 ) 

264 

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

266 if DEBUG: 

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

268 

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

270 pkt = _roundtrip(conn, data)