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

1# Python implementation of low level MySQL client-server protocol 

2# http://dev.mysql.com/doc/internals/en/client-server-protocol.html 

3 

4import struct 

5import sys 

6 

7from . import err 

8from .charset import MBLENGTH 

9from .constants import FIELD_TYPE, SERVER_STATUS 

10 

11DEBUG = False 

12 

13NULL_COLUMN = 251 

14UNSIGNED_CHAR_COLUMN = 251 

15UNSIGNED_SHORT_COLUMN = 252 

16UNSIGNED_INT24_COLUMN = 253 

17UNSIGNED_INT64_COLUMN = 254 

18 

19 

20def dump_packet(data): # pragma: no cover 

21 def printable(data): 

22 if 32 <= data < 127: 

23 return chr(data) 

24 return "." 

25 

26 try: 

27 print("packet length:", len(data)) 

28 for i in range(1, 7): 

29 f = sys._getframe(i) 

30 print("call[%d]: %s (line %d)" % (i, f.f_code.co_name, f.f_lineno)) 

31 print("-" * 66) 

32 except ValueError: 

33 pass 

34 dump_data = [data[i : i + 16] for i in range(0, min(len(data), 256), 16)] 

35 for d in dump_data: 

36 print( 

37 " ".join(f"{x:02X}" for x in d) 

38 + " " * (16 - len(d)) 

39 + " " * 2 

40 + "".join(printable(x) for x in d) 

41 ) 

42 print("-" * 66) 

43 print() 

44 

45 

46class MysqlPacket: 

47 """Representation of a MySQL response packet. 

48 

49 Provides an interface for reading/parsing the packet results. 

50 """ 

51 

52 __slots__ = ("_data", "_position") 

53 

54 def __init__(self, data, encoding): 

55 self._position = 0 

56 self._data = data 

57 

58 def get_all_data(self): 

59 return self._data 

60 

61 def read(self, size): 

62 """Read the first 'size' bytes in packet and advance cursor past them.""" 

63 result = self._data[self._position : (self._position + size)] 

64 if len(result) != size: 

65 error = ( 

66 "Result length not requested length:\n" 

67 f"Expected={size}. Actual={len(result)}. Position: {self._position}. Data Length: {len(self._data)}" 

68 ) 

69 if DEBUG: 

70 print(error) 

71 self.dump() 

72 raise AssertionError(error) 

73 self._position += size 

74 return result 

75 

76 def read_all(self): 

77 """Read all remaining data in the packet. 

78 

79 (Subsequent read() will return errors.) 

80 """ 

81 result = self._data[self._position :] 

82 self._position = None # ensure no subsequent read() 

83 return result 

84 

85 def advance(self, length): 

86 """Advance the cursor in data buffer 'length' bytes.""" 

87 new_position = self._position + length 

88 if new_position < 0 or new_position > len(self._data): 

89 raise Exception( 

90 f"Invalid advance amount ({length}) for cursor. Position={new_position}" 

91 ) 

92 self._position = new_position 

93 

94 def rewind(self, position=0): 

95 """Set the position of the data buffer cursor to 'position'.""" 

96 if position < 0 or position > len(self._data): 

97 raise Exception("Invalid position to rewind cursor to: %s." % position) 

98 self._position = position 

99 

100 def get_bytes(self, position, length=1): 

101 """Get 'length' bytes starting at 'position'. 

102 

103 Position is start of payload (first four packet header bytes are not 

104 included) starting at index '0'. 

105 

106 No error checking is done. If requesting outside end of buffer 

107 an empty string (or string shorter than 'length') may be returned! 

108 """ 

109 return self._data[position : (position + length)] 

110 

111 def read_uint8(self): 

112 result = self._data[self._position] 

113 self._position += 1 

114 return result 

115 

116 def read_uint16(self): 

117 result = struct.unpack_from("<H", self._data, self._position)[0] 

118 self._position += 2 

119 return result 

120 

121 def read_uint24(self): 

122 low, high = struct.unpack_from("<HB", self._data, self._position) 

123 self._position += 3 

124 return low + (high << 16) 

125 

126 def read_uint32(self): 

127 result = struct.unpack_from("<I", self._data, self._position)[0] 

128 self._position += 4 

129 return result 

130 

131 def read_uint64(self): 

132 result = struct.unpack_from("<Q", self._data, self._position)[0] 

133 self._position += 8 

134 return result 

135 

136 def read_string(self): 

137 end_pos = self._data.find(b"\0", self._position) 

138 if end_pos < 0: 

139 return None 

140 result = self._data[self._position : end_pos] 

141 self._position = end_pos + 1 

142 return result 

143 

144 def read_length_encoded_integer(self): 

145 """Read a 'Length Coded Binary' number from the data buffer. 

146 

147 Length coded numbers can be anywhere from 1 to 9 bytes depending 

148 on the value of the first byte. 

149 """ 

150 c = self._data[self._position] 

151 self._position += 1 

152 if c == NULL_COLUMN: 

153 return None 

154 if c < UNSIGNED_CHAR_COLUMN: 

155 return c 

156 elif c == UNSIGNED_SHORT_COLUMN: 

157 return self.read_uint16() 

158 elif c == UNSIGNED_INT24_COLUMN: 

159 return self.read_uint24() 

160 elif c == UNSIGNED_INT64_COLUMN: 

161 return self.read_uint64() 

162 

163 def read_length_coded_string(self): 

164 """Read a 'Length Coded String' from the data buffer. 

165 

166 A 'Length Coded String' consists first of a length coded 

167 (unsigned, positive) integer represented in 1-9 bytes followed by 

168 that many bytes of binary data. (For example "cat" would be "3cat".) 

169 """ 

170 length = self.read_length_encoded_integer() 

171 if length is None: 

172 return None 

173 return self.read(length) 

174 

175 def read_struct(self, fmt): 

176 s = struct.Struct(fmt) 

177 result = s.unpack_from(self._data, self._position) 

178 self._position += s.size 

179 return result 

180 

181 def is_ok_packet(self): 

182 # https://dev.mysql.com/doc/internals/en/packet-OK_Packet.html 

183 return self._data[0] == 0 and len(self._data) >= 7 

184 

185 def is_eof_packet(self): 

186 # http://dev.mysql.com/doc/internals/en/generic-response-packets.html#packet-EOF_Packet 

187 # Caution: \xFE may be LengthEncodedInteger. 

188 # If \xFE is LengthEncodedInteger header, 8bytes followed. 

189 return self._data[0] == 0xFE and len(self._data) < 9 

190 

191 def is_auth_switch_request(self): 

192 # http://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::AuthSwitchRequest 

193 return self._data[0] == 0xFE 

194 

195 def is_extra_auth_data(self): 

196 # https://dev.mysql.com/doc/internals/en/successful-authentication.html 

197 return self._data[0] == 1 

198 

199 def is_resultset_packet(self): 

200 field_count = self._data[0] 

201 return 1 <= field_count <= 250 

202 

203 def is_load_local_packet(self): 

204 return self._data[0] == 0xFB 

205 

206 def is_error_packet(self): 

207 return self._data[0] == 0xFF 

208 

209 def check_error(self): 

210 if self.is_error_packet(): 

211 self.raise_for_error() 

212 

213 def raise_for_error(self): 

214 self.rewind() 

215 self.advance(1) # field_count == error (we already know that) 

216 errno = self.read_uint16() 

217 if DEBUG: 

218 print("errno =", errno) 

219 err.raise_mysql_exception(self._data) 

220 

221 def dump(self): 

222 dump_packet(self._data) 

223 

224 

225class FieldDescriptorPacket(MysqlPacket): 

226 """A MysqlPacket that represents a specific column's metadata in the result. 

227 

228 Parsing is automatically done and the results are exported via public 

229 attributes on the class such as: db, table_name, name, length, type_code. 

230 """ 

231 

232 def __init__(self, data, encoding): 

233 MysqlPacket.__init__(self, data, encoding) 

234 self._parse_field_descriptor(encoding) 

235 

236 def _parse_field_descriptor(self, encoding): 

237 """Parse the 'Field Descriptor' (Metadata) packet. 

238 

239 This is compatible with MySQL 4.1+ (not compatible with MySQL 4.0). 

240 """ 

241 self.catalog = self.read_length_coded_string() 

242 self.db = self.read_length_coded_string() 

243 self.table_name = self.read_length_coded_string().decode(encoding) 

244 self.org_table = self.read_length_coded_string().decode(encoding) 

245 self.name = self.read_length_coded_string().decode(encoding) 

246 self.org_name = self.read_length_coded_string().decode(encoding) 

247 ( 

248 self.charsetnr, 

249 self.length, 

250 self.type_code, 

251 self.flags, 

252 self.scale, 

253 ) = self.read_struct("<xHIBHBxx") 

254 # 'default' is a length coded binary and is still in the buffer? 

255 # not used for normal result sets... 

256 

257 def description(self): 

258 """Provides a 7-item tuple compatible with the Python PEP249 DB Spec.""" 

259 return ( 

260 self.name, 

261 self.type_code, 

262 None, # TODO: display_length; should this be self.length? 

263 self.get_column_length(), # 'internal_size' 

264 self.get_column_length(), # 'precision' # TODO: why!?!? 

265 self.scale, 

266 self.flags % 2 == 0, 

267 ) 

268 

269 def get_column_length(self): 

270 if self.type_code == FIELD_TYPE.VAR_STRING: 

271 mblen = MBLENGTH.get(self.charsetnr, 1) 

272 return self.length // mblen 

273 return self.length 

274 

275 def __str__(self): 

276 return f"{self.__class__} {self.db!r}.{self.table_name!r}.{self.name!r}, type={self.type_code}, flags={self.flags:x}" 

277 

278 

279class OKPacketWrapper: 

280 """ 

281 OK Packet Wrapper. It uses an existing packet object, and wraps 

282 around it, exposing useful variables while still providing access 

283 to the original packet objects variables and methods. 

284 """ 

285 

286 def __init__(self, from_packet): 

287 if not from_packet.is_ok_packet(): 

288 raise ValueError( 

289 "Cannot create " 

290 + str(self.__class__.__name__) 

291 + " object from invalid packet type" 

292 ) 

293 

294 self.packet = from_packet 

295 self.packet.advance(1) 

296 

297 self.affected_rows = self.packet.read_length_encoded_integer() 

298 self.insert_id = self.packet.read_length_encoded_integer() 

299 self.server_status, self.warning_count = self.read_struct("<HH") 

300 self.message = self.packet.read_all() 

301 self.has_next = self.server_status & SERVER_STATUS.SERVER_MORE_RESULTS_EXISTS 

302 

303 def __getattr__(self, key): 

304 return getattr(self.packet, key) 

305 

306 

307class EOFPacketWrapper: 

308 """ 

309 EOF Packet Wrapper. It uses an existing packet object, and wraps 

310 around it, exposing useful variables while still providing access 

311 to the original packet objects variables and methods. 

312 """ 

313 

314 def __init__(self, from_packet): 

315 if not from_packet.is_eof_packet(): 

316 raise ValueError( 

317 f"Cannot create '{self.__class__}' object from invalid packet type" 

318 ) 

319 

320 self.packet = from_packet 

321 self.warning_count, self.server_status = self.packet.read_struct("<xhh") 

322 if DEBUG: 

323 print("server_status=", self.server_status) 

324 self.has_next = self.server_status & SERVER_STATUS.SERVER_MORE_RESULTS_EXISTS 

325 

326 def __getattr__(self, key): 

327 return getattr(self.packet, key) 

328 

329 

330class LoadLocalPacketWrapper: 

331 """ 

332 Load Local Packet Wrapper. It uses an existing packet object, and wraps 

333 around it, exposing useful variables while still providing access 

334 to the original packet objects variables and methods. 

335 """ 

336 

337 def __init__(self, from_packet): 

338 if not from_packet.is_load_local_packet(): 

339 raise ValueError( 

340 f"Cannot create '{self.__class__}' object from invalid packet type" 

341 ) 

342 

343 self.packet = from_packet 

344 self.filename = self.packet.get_all_data()[1:] 

345 if DEBUG: 

346 print("filename=", self.filename) 

347 

348 def __getattr__(self, key): 

349 return getattr(self.packet, key)