【转载声明】本文为转载文章，原作者「蒋乐兴」，版权归原作者所有。　原文链接：https://mp.weixin.qq.com/s/T-uHLGbg3U5H1DMXtwKlOg

公众号：杨建荣的学习笔记 · 作者：蒋乐兴 · 发布：2020-06-10 23:59:15 · 原文链接

作者：蒋乐兴

MySQL DBA，擅长 python 和 SQL，目前维护着 github 的两个开源项目：mysqltools 、dbmc 以及独立博客：https://www.sqlpy.com。

本文来源：原创投稿

*爱可生开源社区出品，原创内容未经授权不得随意使用，转载请联系小编并注明来源。

问题

在 Python 语言环境下我们这样连接数据库。

1. In [1]: from mysql import connector

3. In [2]: cnx = connector.connect(host="172.16.192.100",port=3306,user="appuser",password="xxxxxx")

但是连接数据库的背后发生了什么呢？

答案

当我们通过驱动程序（mysql-connector-python,pymysql）连接 MySQL 服务端的时候，就是把连接参数传递给驱动程序，驱动程序再根据参数会发起到 MySQL 服务端的 TCP 连接。

当 TCP 连接建立之后驱动程序与服务端之间会按特定的格式和次序交换数据包，数据包的格式和发送次序由 MySQL 协议 规定。

MySQL 协议：

https://dev.mysql.com/doc/internals/en/client-server-protocol.html

整个连接的过程中 MySQL 服务端与驱动程序之间，按如下的次序发送了这些包。

1. MySQL 服务端向客户端发送一个握手包，包里记录了 MySQL-Server 的版本，默认的授权插件，密码盐值（auth-data）。

2. MySQL 客户端发出 ssl 连接请求包（如果有必要的话）。

3. MySQL 客户端发出握手包的响应包，这个包时记录了用户名，密码加密后的串，客户端属性，等等其它信息。

4. MySQL 服务端发出响应包，这个包里记录了登录是否成功，如果没有成功也会给出错误信息。

祼写 TCP 连接 MySQL

从上面给出的信息可以看出像 mysql-connector-python,pymysql 这类的驱动程序，并不是什么神仙、皇帝，只是一个普普通通的 TCP 客户端。那我们能不能自己一个程序来完成“连接”功能呢？还真可以这么干。

1. #!/usr/bin/env python3

2. """

3. """

5. import os

6. import ssl

7. import sys

8. import time

9. import socket

10. import struct

11. import logging

12. import argparse

14. from plugins import get_auth_plugin

16. logging.basicConfig(level=logging.INFO,

17. format='%(asctime)s - %(name)s - %(threadName)s - %(levelname)s - %(lineno)s - %(message)s')

20. def read_str(packet, ends=None, size=None):

21. """

22. """

23. if ends is None and size is None:

24. raise ValueError("either ends not None or size not None.")

26. if not isinstance(packet, (bytes, bytearray)):

27. raise ValueError('packet must be a bytes or bytearray.')

29. if ends is not None:

30. index = packet.index(ends)

31. return packet[index + 1:], packet[0:index]

33. if size is not None and size < len(packet):

34. return packet[size + 1], packet[0:size]

35. else:

36. raise ValueError('size must less than len(packet)')

39. class MySQLTcpSocket(object):

40. """封装一个到 MySQL 数据库的 TCP 连接(同步IO)

41. """

43. def __init__(self, host=None, port=3306, user=None, password=None):

44. self._host = host

45. self._port = port

46. self._user = user

47. self._password = password

48. self._packet_number = 0

50. addrinfos = socket.getaddrinfo(

51. self._host, self._port, socket.AF_INET, socket.SOCK_STREAM, 0)

53. for info in addrinfos:

54. try:

55. family, socket_type, proto, _, addrs = info

56. self._sock = socket.socket(family, socket_type, proto)

57. self._sock.settimeout(3)

58. self._sock.connect(addrs)

59. break

60. except IOError as err:

61. logging.exception(str(err))

62. if hasattr(self._sock, 'close'):

63. self._sock.close()

64. sys.exit(1)

65. except Exception as err:

66. logging.exception(str(err))

67. if hasattr(self._sock, 'close'):

68. self._sock.close()

69. sys.exit(1)

71. @property

72. def packet_number(self):

73. """

74. """

75. if self._packet_number >= 255:

76. self._packet_number = 0

78. self._packet_number = self._packet_number + 1

80. return self._packet_number

82. def prepare_packets(self, packet, packet_number=None):

83. """如果包大于 16 MB 就拆解成多个。

84. """

85. # 16 MB - 1

86. max_length_packet = (1 << 24) - 1

87. if packet_number is None:

88. packet_number = self.packet_number

90. # 超过 16M 的部分按 16M 打包

91. packets = []

92. while len(packet) > max_length_packet:

93. pct = b'\x00\x00\x00' + struct.pack("B", packet_number)

94. packet_number = packet_number + 1

95. packets.append(pct + packet[0:max_length_packet])

96. packet = packet[max_length_packet:]

98. # 没有超过 16M 的部分按实际大小打包

99. pakcet_len = len(packet)

100. pct = struct.pack("<I", pakcet_len)[0:3] + \

101. struct.pack("<B", packet_number) + packet

103. packets.append(pct)

104. return packets

106. def recv(self):

107. """从 MySQL 服务端接收一个数据包

108. """

109. rst = 4

110. header = b''

111. while rst > 0:

112. chunck = self._sock.recv(rst)

113. header = header + chunck

114. rst = rst - len(chunck)

116. # 执行到这里说明状况接收完成,准备接收 payload 部分的字节

117. packet_length, *_ = struct.unpack("<I", header[0:3] + b'\x00')

118. packet = bytearray(packet_length)

119. pvm = memoryview(packet)

120. rst = packet_length

122. while rst > 0:

123. chunck = self._sock.recv(rst)

124. chunck_len = len(chunck)

125. pvm[0:chunck_len] = chunck

126. pvm = pvm[chunck_len:]

127. rst = rst - chunck_len

129. return header + packet

131. def send(self, packet):

132. """发送数据包到 MySQL-Server(自动分组)

133. """

134. for pct in self.prepare_packets(packet):

135. self._sock.sendall(pct)

137. def change_to_ssl_mode(self):

138. """切换到 SSL-Client 模式

139. """

140. context = ssl.create_default_context()

141. context.check_hostname = False

142. context.verify_mode = ssl.CERT_NONE

143. context.load_default_certs()

144. self._sock = context.wrap_socket(self._sock)

146. def __del__(self):

147. if hasattr(self._sock, 'close'):

148. logging.info("close tcp socket object.")

149. self._sock.close()

152. class MySQLProtocol(object):

153. """实现 MySQL 各种数据包的解包的打包

154. """

156. def parser_init_packet(self, packet):

157. """解析 MySQL-Server 发来的握手包

158. """

159. payload_len, *_ = struct.unpack("<I", packet[0:3] + b'\x00')

160. packet_number = packet[3]

161. packet = packet[4:]

163. # 解析

164. # protocol-version

165. protocal_version = packet[0]

166. packet = packet[1:]

168. # 解析

169. # mysql-version

170. packet, mysql_version = read_str(packet, ends=b'\x00')

171. mysql_version = mysql_version.decode("utf8")

173. #

174. connection_id, auth_data_1, capability_lower = struct.unpack(

175. "<I8sx2s", packet[0:15])

177. packet = packet[15:]

179. if len(packet) > 0:

180. charset, status_flags, capability_uper = struct.unpack(

181. f"<BH2s", packet[0:5])

182. packet = packet[5:]

184. auth_plugin_data_length = packet[0]

185. auth_data_2_len = max(13, (auth_plugin_data_length - 8))

187. # packet = packet[1:]

188. packet = packet[11:]

189. auth_data_2 = packet[0:auth_data_2_len]

190. if auth_data_2.endswith(b'\x00'):

191. auth_data_2 = auth_data_2[0:-1]

193. packet = packet[13:]

194. packet, auth_plugin = read_str(packet, ends=b'\x00')

195. auth_plugin = auth_plugin.decode('utf8')

197. capabilities, * \

198. _ = struct.unpack("<I", capability_lower + capability_uper)

199. res = {

200. 'protocol': protocal_version,

201. 'server_version_original': mysql_version,

202. 'server_threadid': connection_id,

203. 'charset': charset,

204. 'server_status': status_flags,

205. 'auth_plugin': auth_plugin,

206. 'auth_data': auth_data_1 + auth_data_2,

207. 'capabilities': capabilities

208. }

210. return res

212. def make_ssl_request_pakcet(self, client_flags=1813005, charset=45, max_allowed_packet=1073741824):

213. """打包 SSL 连接请求包

214. """

215. packet = struct.pack("IIB" + 'x' * 23, client_flags,

216. max_allowed_packet, charset)

217. return packet

219. def make_auth_response_packet(self, auth_data=None, plugin_name='caching_sha2_password',

220. username=None, password=None, database=None, charset=45,

221. client_flags=0, max_allowed_packet=1073741824, ssl_enabled=False,

222. auth_plugin=None, conn_attrs=None):

223. """

224. """

225. # 打包 client_flags,max_allowed_packet,charset

226. packet = struct.pack("IIB" + "x"*23, client_flags,

227. max_allowed_packet, charset)

229. # 打包 username

230. username_bytes = username.encode("utf8")

231. packet = packet + username_bytes + b'\x00'

233. # capabilities & CLIENT_SECURE_CONNECTION == True

234. auth = get_auth_plugin('caching_sha2_password')(

235. auth_data, username, password, ssl_enabled)

236. auth_response = auth.auth_response()

237. auth_response_len = len(auth_response)

238. packet = packet + struct.pack("B", auth_response_len) + auth_response

240. # capabilities & CLIENT_CONNECT_WITH_DB == False

241. packet = packet + b'\x00'

243. # capabilities & CLIENT_PLUGIN_AUTH == True

244. packet = packet + auth_plugin.encode('utf8') + b'\x00'

246. # capabilities & CLIENT_CONNECT_ATTRS == True

247. attrs_len = sum([(2 + len(name) + len(conn_attrs[name]))

248. for name in conn_attrs])

249. packet = packet + struct.pack("B", attrs_len)

251. for name in conn_attrs:

252. packet = packet + struct.pack("B", len(name)) + name.encode('utf8')

253. packet = packet + \

254. struct.pack("B", len(conn_attrs[name])) + \

255. conn_attrs[name].encode('utf8')

257. #

258. return packet

261. def main(host="sqlstudio", port=3306,

262. user='appuser', password='123456'):

264. # 发起到连接 MySQL-Server 的连接

265. logging.info(f"发起到 {host}:{port} 的 TCP 连接 .")

266. sock = MySQLTcpSocket(host=host, port=port,

267. user=user, password=password)

269. # 收取 MySQL-Server 发来的握手包

270. init_packet = sock.recv()

271. logging.info("收到来自 MySQL-Server 的握手包 .")

273. # 解析握手包

274. protocol = MySQLProtocol()

275. init_dict = protocol.parser_init_packet(init_packet)

276. logging.info(f"握手包解析完成")

278. # 发送 ssl 连接请求

279. ssl_connection_requet_packet = protocol.make_ssl_request_pakcet()

280. sock.send(ssl_connection_requet_packet)

281. sock.change_to_ssl_mode()

282. logging.info("已经切换到 ssl 模式 .")

284. # 打包客户端属性信息，由于它不是重点，我这里直接写死了 AMD Yes!

285. # 随便告诉 MySQL-Server 我的操作系统是 Windows-100

286. conn_attrs = {

287. "_pid": f"{os.getpid()}",

288. "_platform": "x86_64",

289. "_source_host": "AMD-Yes",

290. "_client_name": "pure-socket",

291. "_client_license": "GPL-2.0",

292. "_client_version": "0.0.1",

293. "_os": "Windows-100"

294. }

295. auth_response = protocol.make_auth_response_packet(auth_data=init_dict['auth_data'],

296. auth_plugin='caching_sha2_password', username=user, password=password, database='',

297. charset=45, client_flags=1813005, max_allowed_packet=1073741824, ssl_enabled=True, conn_attrs=conn_attrs)

298. # 发送握手的响应包

299. sock.send(auth_response)

300. logging.info("握手响应包发送完成 .")

302. # 处理二次验证

303. auth_switch_packet = sock.recv()

305. if auth_switch_packet[4] == 1:

306. auth_data = auth_switch_packet[5:]

308. if len(auth_data) == 1 and auth_data[0] == 4:

309. auth_response = password.encode('utf8') + b'\x00'

310. sock._packet_number = 3

311. sock.send(auth_response)

312. logging.info("二次校验完成 .")

314. #

315. packet = sock.recv()

316. logging.info("收到来自 MySQL-Server 的登录确认包.")

317. if packet[4] == 0:

318. logging.info(f"收到的是 OK 包、用户登录成功！")

319. else:

320. logging.info(f"登录 MySQL-Server 失败了！")

321. logging.info(f"{packet}")

323. # sleep 30 秒

324. time.sleep(30)

327. if __name__ == "__main__":

328. parser = argparse.ArgumentParser('mysql-login')

329. parser.add_argument('--host', type=str,

330. default="172.16.192.100", help="mysql-server ip")

331. parser.add_argument('--port', type=int, default=3306,

332. help="mysql-server listening port")

333. parser.add_argument('--user', type=str,

334. default='appuser', help="mysql user")

335. parser.add_argument('--password', type=str,

336. default="123456", help="password of mysql user")

338. args = parser.parse_args()

339. main(host=args.host, port=args.port, user=args.user, password=args.password)

执行程序看输出。

1. python3 mysql-login.py

2. 2020-05-26 15:51:20,494 - root - MainThread - INFO - 265 - 发起到 172.16.192.100:3306 的 TCP 连接 .

3. 2020-05-26 15:51:20,498 - root - MainThread - INFO - 271 - 收到来自 MySQL-Server 的握手包 .

4. 2020-05-26 15:51:20,499 - root - MainThread - INFO - 276 - 握手包解析完成

5. 2020-05-26 15:51:20,503 - root - MainThread - INFO - 282 - 已经切换到 ssl 模式 .

6. 2020-05-26 15:51:20,503 - root - MainThread - INFO - 300 - 握手响应包发送完成 .

7. 2020-05-26 15:51:20,504 - root - MainThread - INFO - 316 - 收到来自 MySQL-Server 的登录确认包.

8. 2020-05-26 15:51:20,504 - root - MainThread - INFO - 318 - 收到的是 OK 包、用户登录成功！

在 MySQL 服务端观察连接属性。

1. mysql> show processlist;

2. +----+---------+--------------------+------+---------+------+----------+------------------+

3. | Id | User | Host | db | Command | Time | State | Info |

4. +----+---------+--------------------+------+---------+------+----------+------------------+

5. | 7 | monitor | 127.0.0.1:45088 | NULL | Sleep | 4 | | NULL |

6. | 8 | root | 127.0.0.1:45090 | NULL | Query | 0 | starting | show processlist |

7. | 12 | appuser | 172.16.192.1:55290 | NULL | Sleep | 4 | | NULL |

8. +----+---------+--------------------+------+---------+------+----------+------------------+

9. 3 rows in set (0.00 sec)

11. mysql> select * from performance_schema.session_connect_attrs where processlist_id=12;

12. +----------------+-----------------+-------------+------------------+

13. | PROCESSLIST_ID | ATTR_NAME | ATTR_VALUE | ORDINAL_POSITION |

14. +----------------+-----------------+-------------+------------------+

15. | 12 | _pid | 4885 | 0 |

16. | 12 | _platform | x86_64 | 1 |

17. | 12 | _source_host | AMD-Yes | 2 |

18. | 12 | _client_name | pure-socket | 3 |

19. | 12 | _client_license | GPL-2.0 | 4 |

20. | 12 | _client_version | 0.0.1 | 5 |

21. | 12 | _os | Windows-100 | 6 |

22. +----------------+-----------------+-------------+------------------+

23. 7 rows in set (0.01 sec)

在不用连接驱动的情况下 300 多行代码才实现连接到 MySQL 这个操作，生活太难了！是不是祼写 TCP 就没有用了呢？如果打算自己开发读写分离中间件的话，这个还是有必要的。

其它

关于 MySQL 协议的更多内容可以看 MySQL 内部文档 。上面的示例代码你也可以在 github 上找到。

MySQL 内部文档： 

https://dev.mysql.com/doc/internals/en/client-server-protocol.html

github：

https://github.com/Neeky/mysql-protocol-packets

社区近期动态

 点一下“阅读原文”了解更多资讯