[TOC]

MySQL通讯协议（3）连接阶段

MySQL 连接生命周期

1graph TD
2A[开始] --> |连接|B(ConnectionState)
3B --> |认证成功|C(CommandState)
4C --> |复制命令|D(ReplicationMode)
5B --> |复制命令|D
6B --> |错误或断开|End
7C --> |关闭连接|End
8D --> |关闭连接|End[结束]
9

MySQL连接是有状态的，当服务接通后，首先会进入连接状态，进行认证，如交换信息、认证账号密码等。认证成功后，进入命令阶段，提交命令接受响应。同时，在连接阶段和命令阶段受到复制命令，都可以进入复制模式。

连接阶段

连接阶段主要做三件事：

交换客户机和服务器的支持的功能
如果需要，设置SSL通信通道
服务器对客户端进行身份认证

Plain Handshake

1, Initial Handshake Packet

连接建立之后，服务端先发送初始握手包。以最新的HandshakeV10为例：

Type

Name

Description

int<1>

protocol version

协议版本：10

string[NUL]

server version

易读的服务器版本

int<4>

thread id

连接id

string[8]

auth-plugin-data-part-1

认证插件数据第一部分

int<1>

filler

填充位，固定0x00

int<2>

capability_flags_1

功能标志的低位两个字节

int<1>

character_set

服务器默认编码

int<2>

status_flags

服务器状态

int<2>

capability_flags_2

功能标志的高位两个字节

if capabilities & CLIENT_PLUGIN_AUTH {

int<1>

auth_plugin_data_len

认证插件数据长度

} else {

int<1>

固定常量0x00

}

string[10]

reserved

保留部分，用0x00填充

string[$len]

auth-plugin-data-part-2

认证插件数据第二部分, 长度$len=MAX(13, length of auth-plugin-data - 8)

if capabilities & CLIENT_PLUGIN_AUTH {

string[NUL]

auth_plugin_name

认证插件名

}

字段含义如下：

protocol_version ：协议版本，当前版本为：0x0a也就是10
server_version ：服务器版本，如：8.0.19
connection_id ：连接id
auth_plugin_data_part_1 ：认证插件数据（即加密种子）第一部分
filler_1 ：空位：0x00
capability_flag_1 ：服务器功能的低位2个字节，用每个bit代表一种功能，2个字节能保存16种功能。 Protocol::CapabilityFlags
character_set ：服务器默认编码，对应编码这个编码集合的id。Protocol::CharacterSet 。
status_flags ：服务器状态。 Protocol::StatusFlags
capability_flags_2 ：服务器功能的低位2个字节。 Protocol::CapabilityFlags
auth_plugin_data_len ：认证插件数据两部分的总长度。
auth_plugin_name ：认证插件的名字。

知道了包格式和字段意义，就可以写代码了。

1import java.io.IOException;
2import java.io.InputStream;
3import java.net.InetSocketAddress;
4import java.net.Socket;
5
6public class HandshakeV10Console {
7    public static void main(String[] args) throws IOException {
8        Socket socket = new Socket();
9        socket.connect(new InetSocketAddress("127.0.0.1", 3306));
10        InputStream in = socket.getInputStream();
11
12        byte[] head = new byte[4];
13        while (in.read(head) != 4) {
14        }
15
16        final int length = head[0] + (head[1] & 0xff << 8) + (head[2] & 0xff << 16);
17        System.out.println("length:" + length);
18        final int seq = head[3];
19        System.out.println("seq:" + seq);
20
21        byte[] body = new byte[length];
22        while ((in.read(body)) != length) {
23        }
24
25        final int protocolVersion = body[0];
26        System.out.println("protocolVersion:" + protocolVersion);
27
28        int position = 1, p = 0;
29        for (; ; ) {
30            if (body[position + p] == 0) {
31                break;
32            }
33            p++;
34        }
35        byte[] ssa = new byte[p];
36        System.arraycopy(body, position, ssa, 0, ssa.length);
37        final String serverVersion = new String(ssa);
38        System.out.println("serverVersion:" + serverVersion);
39
40        position = position + p + 1;
41
42        byte[] cida = new byte[4];
43        System.arraycopy(body, position, cida, 0, cida.length);
44        final int connectionId = (cida[0] & 0xff) + ((cida[1] & 0xff) << 8) + ((cida[2] & 0xff) << 16) + ((cida[3] & 0xff) << 24);
45        System.out.println("connectionId:" + connectionId);
46
47        position += 4;
48
49        // auth-plugin-data-part-1
50        byte[] apdpa1 = new byte[8];
51        System.arraycopy(body, position, apdpa1, 0, apdpa1.length);
52        final String authPluginDataPart1 = new String(apdpa1);
53        System.out.println("authPluginDataPart1:" + authPluginDataPart1);
54
55        position += 9; //filler(1) == 0x00
56
57
58        //capability_flag_1 (2)
59        byte[] cfa = new byte[2];
60        System.arraycopy(body, position, cfa, 0, cfa.length);
61        final int capabilityFlag = (cfa[0] & 0xff) + ((cfa[1] & 0xff) << 8);
62        System.out.println("capabilityFlag:" + capabilityFlag);//65535 = ffff
63
64        position += 2;
65
66        //character_set (1)
67        final int characterSet = (body[position] & 0xff);
68        System.out.println("characterSet:" + characterSet);//33 = utf8_general_ci
69
70        position += 1;
71
72        //status_flags (2)
73        final int statusFlags = (body[position] & 0xff) + ((body[position + 1] & 0xff) << 8);
74        System.out.println("statusFlags:" + statusFlags);//2 = auto-commit is enabled
75
76        position += 2;
77
78        //capability_flag_2 (2)
79        byte[] cfa2 = new byte[2];
80        System.arraycopy(body, position, cfa2, 0, cfa.length);
81        final int capabilityFlag2 = ((cfa2[0] & 0xff) << 16) + ((cfa2[1] & 0xff) << 24);
82        System.out.println("capabilityFlag2:" + capabilityFlag2);//65535 = ffff
83
84        position += 2;
85
86        //auth_plugin_data_len (1)
87        final int authPluginDataLen = (body[position] & 0xff);
88        System.out.println("authPluginDataLen:" + authPluginDataLen);// 0x00080000
89
90        position += 1;
91
92        position += 10;//reserved (all [00])
93
94        int capabilities = capabilityFlag + capabilityFlag2;
95        if ((capabilities & 0x00008000) != 0) {
96            // auth-plugin-data-part-2
97            int len = Math.max(13, authPluginDataLen - 8);
98            System.out.println("auth-plugin-data-part-2 length:" + len);
99            byte[] apdpa2 = new byte[len];
100            System.arraycopy(body, position, apdpa2, 0, apdpa2.length);
101            String authPluginDataPart2 = new String(apdpa2);
102            System.out.println("authPluginDataPart2:" + authPluginDataPart2);
103            position += len;
104        }
105
106
107        if ((capabilities & 0x00080000) != 0) {
108            //auth-plugin name
109            int p2 = 0;
110            for (; ; ) {
111                if (body[position + p2] == 0) {
112                    break;
113                }
114                p2++;
115            }
116            byte[] apna = new byte[p2];
117            System.arraycopy(body, position, apna, 0, apna.length);
118            final String authPluginName = new String(apna);
119            System.out.println("authPluginName:" + authPluginName);
120            position = position + p2 + 1;
121        }
122        System.out.println(position);
123
124        socket.close();
125    }
126}

运行结果：

1length:74
2seq:0
3protocolVersion:10
4serverVersion:8.0.19
5connectionId:9
6authPluginDataPart1:%N1omd4
7capabilityFlag:65535
8characterSet:33
9statusFlags:2
10capabilityFlag2:-939589632
11authPluginDataLen:21
12auth-plugin-data-part-2 length:13
13authPluginDataPart2:X>Yt%#b#y9 
14authPluginName:mysql_native_password
1574

2,Handshake Response

客户端收到握手请求之后，需要回复响应。以客户端4.1+版本使用的HandshakeResponse41为例：

Type

Name

Description

int<4>

client_flag

功能标志，必须包含_CLIENT_PROTOCOL_41_

int<4>

max_packet_size

最大包大小

int<1>

character_set

客户端编码

string[23]

filler

填充位，固定23个0x00

string<NUL>

username

用户名

if capabilities & CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA {

string<length>

auth_response

认证插件生成的认证数据

} else {

int<1>

auth_response_length

认证数据长度

string<length>

auth_response

认证插件生成的认证数据

}

if capabilities & CLIENT_CONNECT_WITH_DB {

string<NUL>

database

连接默认的database(schema)

}

if capabilities & CLIENT_PLUGIN_AUTH {

string<NUL>

client_plugin_name

客户端用来生成认证数据的插件名，用UTF8编码。

}

if capabilities & CLIENT_CONNECT_ATTRS {

int<lenenc>

length of all key-values

所有属性的长度

string<lenenc>

key1

第一个属性的名字

string<lenenc>

value1

第一个属性的值

.. (如果更多的属性，以kv的形式跟在后面)

}

int<1>

zstd_compression_level

zstd压缩算法的压缩级别

字段含义如下：

capability_flags ：客户端支持的功能标志。Protocol::CapabilityFlags
max_packet_size ：客户端发送到服务端的命令包的最大大小。
character_set ：连接默认的编码。 Protocol::CharacterSet.
username ：用来登陆数据库的账号，用连接（_character_set_字段指定的）编码方式编码。
auth-response ：认证插件加密过的认证数据。
database ：连接默认的数据库（schema），用连接（_character_set_字段指定的）编码方式编码。
auth plugin name ：客户端实际用来生成认证数据的插件名字，这个字段需要用UTF-8编码。

用代码实现，首先，把解析初始化包的代码封装一下：

1import com.mysql.cj.protocol.a.NativeServerSession;
2
3import java.io.IOException;
4import java.nio.ByteBuffer;
5import java.nio.channels.SocketChannel;
6
7public class HandshakeV10Parser {
8
9
10    public HandshakeV10Parser() {
11    }
12
13    public InitialHandshakePayload parse(SocketChannel socket) throws IOException {
14        InitialHandshakePayload packet = new InitialHandshakePayload();
15        ByteBuffer buffer = ByteBuffer.allocate(2048);
16
17        //读取头
18        int read = read(socket, buffer, 4);
19        //计算包长
20        int bodyLength = readFixInt(buffer, 3);
21        //读取剩下的
22        read(socket, buffer, bodyLength - read);
23        buffer.flip();
24        buffer.position(4);//跳过头
25
26        packet.setProtocolVersion(readFixInt(buffer, 1));
27        packet.setServerVersion(readNullString(buffer));
28        packet.setThreadId(readFixInt(buffer, 4));
29        packet.setAuthPluginDataPart1(readFixString(buffer, 8));
30        buffer.get();//filler
31        packet.setCapabilityFlags(readFixInt(buffer, 2));
32        packet.setCharacterSet(readFixInt(buffer, 1));
33        packet.setStatusFlags(readFixInt(buffer, 2));
34        int capabilities = packet.getCapabilityFlags() | (readFixInt(buffer, 2) << 16);
35        packet.setCapabilityFlags(capabilities);
36        if ((capabilities & NativeServerSession.CLIENT_PLUGIN_AUTH) != 0) {
37            int i = readFixInt(buffer, 1);
38            packet.setAuthPluginDataLen(i);
39        } else {
40            buffer.get();
41            packet.setAuthPluginDataLen(0);
42        }
43        buffer.position(buffer.position() + 10);//reserved
44        int apdp2len = Math.max(13, packet.getAuthPluginDataLen() - 8);
45        packet.setAuthPluginDataPart2(readFixString(buffer, apdp2len));
46
47        if ((capabilities & NativeServerSession.CLIENT_PLUGIN_AUTH) != 0) {
48            packet.setAuthPluginName(readNullString(buffer));
49        }
50        return packet;
51    }
52
53    private String readFixString(ByteBuffer buffer, int len) {
54        byte[] data = new byte[len];
55        for (int i = 0; i < len; i++) {
56            data[i] = buffer.get();
57        }
58        return new String(data);
59    }
60
61    private String readNullString(ByteBuffer buffer) {
62        int position = buffer.position();
63        int end = position;
64        while (buffer.get() != 0) {
65            end++;
66        }
67        buffer.position(position);
68        byte[] data = new byte[end - position];
69        buffer.get(data);
70        buffer.get();//skip 00
71        return new String(data);
72    }
73
74    private int readFixInt(ByteBuffer buffer, int len) {
75        int data = 0;
76        for (int i = 0; i < len; i++) {
77            data |= (buffer.get() << (i * 8));
78        }
79        return data;
80    }
81
82    public int read(SocketChannel socket, ByteBuffer buffer, int len) throws IOException {
83        grow(buffer, len);
84        int n = 0;
85        while (n < len) {
86            int count = socket.read(buffer);
87            n += count;
88        }
89        return n;
90    }
91
92    public ByteBuffer grow(ByteBuffer buffer, int len) {
93        if (buffer.remaining() < len) {
94            ByteBuffer nb = ByteBuffer.allocate(buffer.capacity() << 1);
95            buffer.flip();
96            nb.put(buffer);
97            return nb;
98        } else {
99            return buffer;
100        }
101    }
102}
103

然后读取加密种子，混淆密码，加上其他所需参数，编码消息返回：

1import com.mysql.cj.protocol.Security;
2import com.mysql.cj.protocol.a.NativeConstants;
3import com.mysql.cj.protocol.a.NativeServerSession;
4
5import java.io.IOException;
6import java.net.InetSocketAddress;
7import java.nio.ByteBuffer;
8import java.nio.channels.SocketChannel;
9import java.nio.charset.StandardCharsets;
10import java.util.HashMap;
11import java.util.Map;
12
13public class MySqlClient {
14
15    private String username = "root";
16    private String password = "root";
17    private String database = "test";
18
19    public static void main(String[] args) throws Exception {
20        MySqlClient client = new MySqlClient();
21        client.run();
22    }
23
24    public void run() {
25        try (SocketChannel socket = SocketChannel.open(new InetSocketAddress("127.0.0.1", 3306))) {
26            InitialHandshakePayload packet = init(socket);
27
28            HandshakeResponse response = new HandshakeResponse();
29            response.setCapabilityFlags(packet.getCapabilityFlags());
30            response.setMaxPacketSize(NativeConstants.MAX_PACKET_SIZE);
31            response.setCharacterSet(packet.getCharacterSet());
32            response.setUsername(username);
33            //混淆
34            response.setAuthResponse(auth(packet, password));
35            response.setDatabase(database);
36            response.setClientPluginName(packet.getAuthPluginName());
37
38            Map<String, String> attrs = new HashMap<>();
39//            attrs.put("_runtime_version", "1.8.0_181");
40//            attrs.put("_client_version", "8.0.19");
41//            attrs.put("_client_license", "GPL");
42//            attrs.put("_runtime_vendor", "Oracle Corporation");
43//            attrs.put("_client_name", "MySQL Connector/J");
44            response.setAttributes(attrs);
45
46            //客户端响应
47            response(socket, response);
48
49            //服务端响应
50            ByteBuffer buffer = ByteBuffer.allocate(2048);
51            int n = 0;
52            while (n < 4) {
53                int count = socket.read(buffer);
54                n += count;
55            }
56            int bodyLength = readFixInt(buffer, 3);
57            //读取剩下的
58            int len = bodyLength - n;
59            n = 0;
60            while (n < len) {
61                int count = socket.read(buffer);
62                n += count;
63            }
64
65            byte type = buffer.get(4);
66            if (type == 0) {
67                //ok
68                System.out.println("ok");
69            } else if (type == (byte) 0xff) {
70                //err
71                int code = (buffer.get(5) & 0xff) | ((buffer.get(6) & 0xff) << 8);
72                System.out.println("error code:" + code);
73                System.out.println("marker:" + (buffer.get(7) & 0xff));
74                byte[] ssa = new byte[5];
75                ssa[0] = buffer.get(8);
76                ssa[1] = buffer.get(9);
77                ssa[2] = buffer.get(10);
78                ssa[3] = buffer.get(11);
79                ssa[4] = buffer.get(12);
80                System.out.println("code:" + new String(ssa));
81
82                int s = 13;
83                while (true) {
84                    if (buffer.get(s) == 0) {
85                        break;
86                    }
87                    s++;
88                }
89                byte[] msga = new byte[s - 13];
90                for (int i = 0; i < msga.length; i++) {
91                    msga[i] = buffer.get(13 + i);
92                }
93                System.out.println("error:" + new String(msga));
94            }
95
96        } catch (IOException e) {
97            e.printStackTrace();
98        }
99    }
100
101    public void response(SocketChannel socket, HandshakeResponse response) throws IOException {
102        ByteBuffer buffer = ByteBuffer.allocate(1024);
103
104        int capabilityFlags = response.getCapabilityFlags();
105        writeFixInt(buffer, 20881935, 4);
106        int maxPacketSize = response.getMaxPacketSize();
107        writeFixInt(buffer, maxPacketSize, 4);
108        int characterSet = response.getCharacterSet();
109        writeFixInt(buffer, characterSet, 1);
110        //filler [00]*23
111        writeFixInt(buffer, 0, 23);
112        String username = response.getUsername();
113        writeNullString(buffer, username.getBytes(StandardCharsets.UTF_8));
114
115        if ((capabilityFlags & NativeServerSession.CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA) != 0) {
116            byte[] authResponse = response.getAuthResponse();
117            writeLengthString(buffer, authResponse);
118        } else {
119            byte[] authResponse = response.getAuthResponse();
120            writeLengthInt(buffer, authResponse.length);
121            writeFixString(buffer, authResponse);
122        }
123
124        if ((capabilityFlags & NativeServerSession.CLIENT_CONNECT_WITH_DB) != 0) {
125            String database = response.getDatabase();
126            writeNullString(buffer, database.getBytes(StandardCharsets.UTF_8));
127        }
128
129        if ((capabilityFlags & NativeServerSession.CLIENT_PLUGIN_AUTH) != 0) {
130            String clientPluginName = response.getClientPluginName();
131            writeNullString(buffer, clientPluginName.getBytes(StandardCharsets.UTF_8));
132        }
133
134        if ((capabilityFlags & NativeServerSession.CLIENT_CONNECT_ATTRS) != 0) {
135            Map<String, String> attributes = response.getAttributes();
136            ByteBuffer attrBuffer = ByteBuffer.allocate(1024);
137            for (Map.Entry<String, String> entry : attributes.entrySet()) {
138                writeLengthString(attrBuffer, entry.getKey().getBytes(StandardCharsets.UTF_8));
139                writeLengthString(attrBuffer, entry.getValue().getBytes(StandardCharsets.UTF_8));
140            }
141            attrBuffer.flip();
142
143            writeLengthInt(buffer, attrBuffer.limit());
144            grow(buffer, attrBuffer.limit());
145            buffer.put(attrBuffer);
146        }
147
148        buffer.flip();
149
150        int bodySize = buffer.limit();
151        ByteBuffer packet = ByteBuffer.allocate(bodySize + 4);
152        writeFixInt(packet, bodySize, 3);
153        packet.put((byte) 1);
154        packet.put(buffer);
155        packet.flip();
156        socket.write(packet);
157
158    }
159
160    private void writeFixString(ByteBuffer buffer, byte[] data) {
161        grow(buffer, data.length);
162        for (byte b : data) {
163            buffer.put(b);
164        }
165    }
166
167    private void writeLengthString(ByteBuffer buffer, byte[] data) {
168        grow(buffer, data.length + 9);
169        writeLengthInt(buffer, data.length);
170        writeFixString(buffer, data);
171    }
172
173    /**
174     * If the value is < 251, it is stored as a 1-byte integer.
175     * If the value is ≥ 251 and < (2^16), it is stored as fc + 2-byte integer.
176     * If the value is ≥ (2^16) and < (2^24), it is stored as fd + 3-byte integer.
177     * If the value is ≥ (2^24) and < (2^64) it is stored as fe + 8-byte integer.
178     */
179    private void writeLengthInt(ByteBuffer buffer, int v) {
180        if (v < 251) {
181            grow(buffer, 1);
182            writeFixInt(buffer, v, 1);
183        } else if (v < 65536L) {
184            grow(buffer, 3);
185            writeFixInt(buffer, 0xfc, 1);
186            writeFixInt(buffer, v, 2);
187        } else if (v < 16777216L) {
188            grow(buffer, 4);
189            writeFixInt(buffer, 0xfd, 1);
190            writeFixInt(buffer, v, 3);
191
192        } else {
193            grow(buffer, 9);
194            writeFixInt(buffer, 0xfe, 1);
195            writeFixInt(buffer, v, 8);
196        }
197    }
198
199    private void writeNullString(ByteBuffer buffer, byte[] data) {
200        grow(buffer, data.length + 1);
201        for (byte b : data) {
202            buffer.put(b);
203        }
204        buffer.put((byte) 0);
205    }
206
207    public void writeFixInt(ByteBuffer buffer, int v, int len) {
208        grow(buffer, len);
209        for (int i = 0; i < len; i++) {
210            buffer.put((byte) (v >>> (i * 8)));
211        }
212    }
213
214
215    private int readFixInt(ByteBuffer buffer, int len) {
216        int data = 0;
217        for (int i = 0; i < len; i++) {
218            data |= (buffer.get() << (i * 8));
219        }
220        return data;
221    }
222
223    public InitialHandshakePayload init(SocketChannel socket) throws IOException {
224        HandshakeV10Parser parser = new HandshakeV10Parser();
225        InitialHandshakePayload packet = parser.parse(socket);
226        System.out.println(packet);
227        return packet;
228    }
229
230
231    public byte[] auth(InitialHandshakePayload packet, String password) {
232        final String authPluginName = packet.getAuthPluginName();
233
234        if ("mysql_native_password".equals(authPluginName)) {
235            String data = packet.getAuthPluginDataPart1() + packet.getAuthPluginDataPart2();
236            byte[] bytes = data.getBytes();
237            byte[] seed = new byte[20];
238            //去掉最后的0
239            System.arraycopy(bytes, 0, seed, 0, 20);
240            return Security.scramble411(password.getBytes(StandardCharsets.UTF_8), seed);
241        } else {
242            //省略
243            return new byte[0];
244        }
245    }
246
247    public ByteBuffer grow(ByteBuffer buffer, int len) {
248        if (buffer.remaining() < len) {
249            ByteBuffer nb = ByteBuffer.allocate(buffer.capacity() << 1);
250            buffer.flip();
251            nb.put(buffer);
252            return nb;
253        } else {
254            return buffer;
255        }
256    }
257}

如果不出意外，服务端应该会返回OK。报文内容为[7] [0] [0] [2] [0] [0] [0] [2]。

总结

至此，连接阶段完成，接下来会进入命令阶段，客户端可以向服务器提交命令了。

另外，由于MySQL8默认认证插件改成了caching_sha2_password，这里为了测试方便，改回了mysql_native_password。连接阶段还涉及一些其他操作，如认证方法切换、SSL连接等，这里暂时略过。

最后，整理下代码：https://github.com/dingfugui/mysql-protocol/tree/master/src/main/java/prrety

相关配置

MySQL server：8.0.19

JDBC：8.0.19

MySQL配置文件：

1[mysqld]
2port=3306
3basedir=...
4datadir=...
5character_set_server=utf8
6default-storage-engine=INNODB
7sql_mode=NO_ENGINE_SUBSTITUTION,STRICT_TRANS_TABLES
8default_authentication_plugin=mysql_native_password
9[mysql]
10default-character-set=utf8

参考资料：

https://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::HandshakeResponse

https://dev.mysql.com/doc/dev/mysql-server/8.0.19/page_protocol_connection_phase_packets.html

mysql:mysql-connector-java:8.0.19

MySQL通讯协议（3）连接阶段

MySQL通讯协议（3）连接阶段

MySQL 连接生命周期

连接阶段

Plain Handshake

1, Initial Handshake Packet

2,Handshake Response

总结

评论区

相关推荐