【Mysql协议解析处理流程】

Mysql协议客户端与服务端交互流程

当Mysql客户端连接服务端后，服务端会发送Server Greeting，而后客户端根据服务端的Server Greeting确认服务端版本，而后发送登录请求；服务端接收到登录请求后，根据参数选择是否发送Auth Swith Request，客户端而后发送Auth Swith Response，如果服务端确认鉴权信息后，会发送Response Ok信息，而后，客户端即可执行sql语句了。
sql语句执行，可顺序执行多条，一条sql语句，返回该语句的结果。
当客户端，需要退出时，发送Requst Quit命令。
一个简单的处理流程，如下图所示：

协议处理内部框架

根据Mysql客户端与服务端交互流程，协议处理过程中，可定义内部状态，而后根据状态不同，调用不同的解析函数进行处理；Mysql内部的状态包括：

typedef enum mysql_state {UNDEFINED,LOGIN,REQUEST,RESPONSE_OK,RESPONSE_ERROR,RESPONSE_EOF,INTERMEDIATE_EOF,RESPONSE_MESSAGE,RESPONSE_TABULAR,RESPONSE_SHOW_FIELDS,FIELD_PACKET,ROW_PACKET,COLUMN_COUNT,RESPONSE_PREPARE,PREPARED_PARAMETERS,PREPARED_FIELDS,AUTH_SWITCH_REQUEST,AUTH_SWITCH_RESPONSE,AUTH_SHA2,AUTH_PUBKEY,AUTH_SHA2_RESPONSE,BINLOG_DUMP,CLONE_INIT,CLONE_ACTIVE,CLONE_EXIT,RESPONSE_LOCALINFILE,INFILE_DATA
} mysql_state_t;

rust代码可以定义为：

pub enum MysqlStateProgress {#[default]UNDEFINED,LOGIN,REQUEST,RESPONSE_OK,RESPONSE_ERROR,RESPONSE_EOF,INTERMEDIATE_EOF,RESPONSE_MESSAGE,RESPONSE_TABULAR,RESPONSE_SHOW_FIELDS,FIELD_PACKET,ROW_PACKET,COLUMN_COUNT,RESPONSE_PREPARE,PREPARED_PARAMETERS,PREPARED_FIELDS,AUTH_SWITCH_REQUEST,AUTH_SWITCH_RESPONSE,AUTH_SHA2,AUTH_PUBKEY,AUTH_SHA2_RESPONSE,BINLOG_DUMP,CLONE_INIT,CLONE_ACTIVE,CLONE_EXIT,RESPONSE_LOCALINFILE,INFILE_DATA
}

在协议处理框架
中介绍了rs_mysql_parse_request及rs_mysql_parse_request，为mysql，请求及相应的入口；对于新来的请求/响应数据，根据当前状态进行解析，解析完一个消息后，根据该消息可以改变协议解析的状态，而后如果数据流中是否存在数据，如果有则可重复上述过程进行数据的解析。

大体流程代码如下：

fn parse_request(&mut self, input: &[u8], flow: *const Flow) -> AppLayerResult {// ... ...let mut start = input;while !start.is_empty() {match MysqlState::state_based_req_parsing(self.state_progress, start) {Ok((rem, response)) => {start = rem;let next_state = self.request_process_next_state(&response, flow);if let Some(state) = next_state {self.state_progress = state;};}Err(nom::Err::Incomplete(_)) => {let consumed = input.len() - start.len();let needed = start.len() + 1;return AppLayerResult::incomplete(consumed as u32, needed as u32);}Err(_) => {return AppLayerResult::err();}}}// Input was fully consumed.return AppLayerResult::ok();}

其中MysqlState::state_based_req_parsing则对应单个消息的处理流程，其处理步骤，需要针对单个消息的组织结构，而后进行解析；
Mysql单个消息遵循3个字节长度+1个字节序列号+消息体的格式进行传输；对于消息体的内容需要根据消息类型进行解析。

消息体解析示例

service greeting解析

wireshark抓包展示的结构如下：

参考wireshark源码中的处理

static int
mysql_dissect_greeting(tvbuff_t *tvb, packet_info *pinfo, int offset,proto_tree *tree, mysql_conn_data_t *conn_data,const mysql_frame_data_t *my_frame_data)
{int protocol;int lenstr;int ver_offset;proto_item *tf;proto_item *greeting_tree;char buffer[7];protocol= tvb_get_uint8(tvb, offset);if (protocol == 0xff) {return mysql_dissect_error_packet(tvb, pinfo, offset+1, tree, my_frame_data);}mysql_set_conn_state(pinfo, conn_data, LOGIN);tf = proto_tree_add_item(tree, hf_mysql_server_greeting, tvb, offset, -1, ENC_NA);greeting_tree = proto_item_add_subtree(tf, ett_server_greeting);col_append_fstr(pinfo->cinfo, COL_INFO, " proto=%d", protocol) ;proto_tree_add_item(greeting_tree, hf_mysql_protocol, tvb, offset, 1, ENC_NA);offset += 1;/* version string */lenstr = tvb_strsize(tvb,offset);/* check if it is a MariaDB Server: MariaDB always sends 5.5.5- before real version number */tvb_get_raw_bytes_as_string(tvb, offset, buffer, 7);if (lenstr > 6 && strncmp(buffer, MARIADB_RPL_VERSION_HACK, sizeof(MARIADB_RPL_VERSION_HACK) - 1) == 0){conn_data->is_mariadb_server= 1;col_append_fstr(pinfo->cinfo, COL_INFO, " version=%s ",tvb_format_text(pinfo->pool, tvb, offset + 6, lenstr - 7));} else {col_append_fstr(pinfo->cinfo, COL_INFO, " version=%s ",tvb_format_text(pinfo->pool, tvb, offset, lenstr-1));}col_set_fence(pinfo->cinfo, COL_INFO);proto_tree_add_item(greeting_tree, hf_mysql_version, tvb, offset, lenstr, ENC_ASCII);conn_data->major_version = 0;for (ver_offset = 0; ver_offset < lenstr; ver_offset++) {uint8_t ver_char = tvb_get_uint8(tvb, offset + ver_offset);if (ver_char == '.') break;conn_data->major_version = conn_data->major_version * 10 + ver_char - '0';}offset += lenstr;/* 4 bytes little endian thread_id */proto_tree_add_item(greeting_tree, hf_mysql_thread_id, tvb, offset, 4, ENC_LITTLE_ENDIAN);offset += 4;/* salt string */lenstr = tvb_strsize(tvb,offset);proto_tree_add_item(greeting_tree, hf_mysql_salt, tvb, offset, lenstr, ENC_ASCII);offset += lenstr;/* rest is optional */if (!tvb_reported_length_remaining(tvb, offset)) return offset;/* 2 bytes CAPS */offset = mysql_dissect_caps(tvb, offset, greeting_tree, hf_mysql_caps_server, &conn_data->srv_caps);/* MariaDB server don't have the CLIENT_MYSQL/CLIENT_LONG_PASSWORD capability */if (!(conn_data->srv_caps & MYSQL_CAPS_LP)){conn_data->is_mariadb_server= 1;}...return offset;
}

对应rust源码如下：

pub fn parse_server_greeting_message(i: &[u8], length:u32, packet_number:u8) -> IResult<&[u8], MysqlBEMessage> {let (i, protocol) = verify(be_u8, |&x| x == 10)(i)?;let (i, version) = take_until("\x00")(i)?;let (i, _) = tag("\x00")(i)?;let (i, thread_id) = le_u32(i)?;let (i, salt1) = take_until("\x00")(i)?;let (i, _) = tag("\x00")(i)?;let (i, server_capabilities) = le_u16(i)?;let (i, character_set) = be_u8(i)?;let (i, status_flags) = le_u16(i)?;let (i, extended_server_capabilities) = le_u16(i)?;let (mut i, auth_plugin_data_len) = le_u8(i)?;let mut is_mairiadb_server = false;let mut mariadb_server_ext_caps = 0_u32;if version.len() >= 6 && &version[0..6] == b"5.5.5-" {is_mairiadb_server = true;SCLogDebug!("Detected MariaDB server");} else {SCLogDebug!("Detected MySQL server");}if server_capabilities & MYSQL_CAPS_LP != 0 {let (i1, _) = take(10_usize)(i)?;i = i1} else {is_mairiadb_server = true;let (i1, _) = take(6_usize)(i)?;(i, mariadb_server_ext_caps) = le_u32(i1)?;}let (i, salt2) = take_until("\x00")(i)?;let (i, _) = tag("\x00")(i)?;let (i, auth_plugin_name) = take_until("\x00")(i)?;let (i, _) = tag("\x00")(i)?;...
}

login request解析

wireshark login request 抓包情况如下：

wireshark对应源码为：

static int
mysql_dissect_login(tvbuff_t *tvb, packet_info *pinfo, int offset,proto_tree *tree, mysql_conn_data_t *conn_data)
{int lenstr;proto_item *tf;proto_item *login_tree;/* after login there can be OK or DENIED */if (conn_data->clnt_caps & MYSQL_CAPS_SL) {mysql_set_conn_state(pinfo, conn_data, LOGIN);} else if (!(conn_data->clnt_caps == 0)) {mysql_set_conn_state(pinfo, conn_data, RESPONSE_OK);}tf = proto_tree_add_item(tree, hf_mysql_login_request, tvb, offset, -1, ENC_NA);login_tree = proto_item_add_subtree(tf, ett_login_request);offset = mysql_dissect_caps(tvb, offset, login_tree, hf_mysql_caps_client, &conn_data->clnt_caps);/* MariaDB clients don't have the CLIENT_MYSQL/CLIENT_LONG_PASSWORD capability */if (!(conn_data->clnt_caps & MYSQL_CAPS_LP)){conn_data->is_mariadb_client= 1;}if (!(conn_data->frame_start_ssl) && conn_data->clnt_caps & MYSQL_CAPS_SL) /* Next packet will be use SSL */{col_set_str(pinfo->cinfo, COL_INFO, "Response: SSL Handshake");conn_data->frame_start_ssl = pinfo->num;ssl_starttls_ack(tls_handle, pinfo, mysql_handle);}if (conn_data->clnt_caps & MYSQL_CAPS_CU) /* 4.1 protocol */{offset = mysql_dissect_extcaps(tvb, offset, login_tree, hf_mysql_extcaps_client, &conn_data->clnt_caps_ext);proto_tree_add_item(login_tree, hf_mysql_max_packet, tvb, offset, 4, ENC_LITTLE_ENDIAN);offset += 4;uint32_t collation;proto_tree_add_item_ret_uint(login_tree, conn_data->is_mariadb_server ? hf_mariadb_collation : hf_mysql_collation, tvb, offset, 1, ENC_NA, &collation);unsigned encoding = collation_to_encoding(collation, conn_data->is_mariadb_server);mysql_set_encoding_client(pinfo, conn_data, encoding);mysql_set_encoding_results(pinfo, conn_data, encoding);offset += 1; /* for charset */if (conn_data->is_mariadb_client){/* 19 bytes unused */proto_tree_add_item(login_tree, hf_mysql_unused, tvb, offset, 19, ENC_NA);offset += 19;offset= mariadb_dissect_caps_or_flags(tvb, offset, FT_UINT32, login_tree, hf_mariadb_extcaps_client, mariadb_extcaps_flags, &conn_data->mariadb_client_ext_caps);} else {/* 23 bytes unused */proto_tree_add_item(login_tree, hf_mysql_unused, tvb, offset, 23, ENC_NA);offset += 23;}} else { /* pre-4.1 */proto_tree_add_item(login_tree, hf_mysql_max_packet, tvb, offset, 3, ENC_LITTLE_ENDIAN);offset += 3;}/* User name */lenstr = my_tvb_strsize(tvb, offset);col_append_fstr(pinfo->cinfo, COL_INFO, " user=%s ",tvb_format_text(pinfo->pool, tvb, offset, lenstr-1));proto_tree_add_item(login_tree, hf_mysql_user, tvb, offset, lenstr, ENC_ASCII);offset += lenstr;...return offset;
}

rust代码如下：

pub fn parse_auth_request(i: &[u8], length:u32, packet_number:u8) -> IResult<&[u8], MysqlFEMessage> {let (mut i, client_capabilities) = le_u16(i)?;let mut ext_client_capabilities = 0_u16;if client_capabilities & MYSQL_CAPS_SL != 0 {SCLogDebug!("SSL is requested by the client");// Handle SSL request here if needed}if client_capabilities & MYSQL_CAPS_CU != 0 {(i, ext_client_capabilities) = le_u16(i)?;(i, _) = le_u32(i)?;(i, _) = be_u8(i)?;(i, _) = take(23_usize)(i)?;} else {// If CLIENT_PROTOCOL_41 is not set, we don't have extended capabilitiesext_client_capabilities = 0;(i, _) = le_u24(i)?;}let (i, user_name) = take_until("\x00")(i)?;let (mut i, _) = tag("\x00")(i)?;if (client_capabilities & MYSQL_CAPS_SC) != 0 {let (i1, passwd_length) = le_u8(i)?;let (i1, password) = take(passwd_length as usize)(i1)?;i = i1;} else {let (i1, password) = take_until("\x00")(i)?;let ( i1, _) = tag("\x00")(i1)?;i = i1;}let mut schema: &[u8] = &[];if (client_capabilities & MYSQL_CAPS_CD) != 0 {let (i1, schema_) = take_until("\x00")(i)?;let (i1, _) = tag("\x00")(i1)?;i = i1;schema = schema_;} if (ext_client_capabilities & MYSQL_CAPS_PA) !=0 {let (i1, auth_plugin_) = take_until("\x00")(i)?;let (i1, _) = tag("\x00")(i1)?;i = i1;}if (ext_client_capabilities & MYSQL_CAPS_CA) !=0 {let (i1, auth_plugin_data_len) = le_u8(i)?;let (i1, auth_plugin_data) = take(auth_plugin_data_len as usize)(i1)?;i = i1;} if (ext_client_capabilities & MYSQL_CAPS_ZS) !=0 {let (i1, zstd_compression_level) = le_u8(i)?;i = i1;SCLogDebug!("ZSTD compression level: {}", zstd_compression_level);} else {SCLogDebug!("ZSTD compression not requested by the client");}...}

FixedLengthInteger的解析

参考Integer Type Mysql协议解析中，会经常出现该类型数据的解析
Mysql源码如下

static int
tvb_get_fle(tvbuff_t *tvb, proto_tree *tree _U_, int offset, uint64_t *res, uint8_t *is_null)
{uint8_t prefix;int num_bytes;uint64_t length;prefix = tvb_get_uint8(tvb, offset);if (is_null) {*is_null = 0;}switch (prefix) {case 251:if (res)*res = 0;if (is_null)*is_null = 1;return 1;case 252: // 0xFCnum_bytes = 3;offset++;length = (uint64_t)tvb_get_uint16(tvb, offset, ENC_LITTLE_ENDIAN);break;case 253: // 0xFDnum_bytes = 4;offset++;length = (uint64_t)tvb_get_uint24(tvb, offset, ENC_LITTLE_ENDIAN);break;case 254: // 0xFEnum_bytes = 9;offset++;length = tvb_get_uint64(tvb, offset, ENC_LITTLE_ENDIAN);break;default:num_bytes = 1;length = tvb_get_uint8(tvb, offset);}if (res) {*res = length;}return num_bytes;
}

对应rust源码定义为：

pub fn parse_fle(i:&[u8]) -> IResult<&[u8], FleMessage> {let (i, tag) = le_u8(i)?;match (tag) {251 => {Ok((i, FleMessage {value: 0,length: 1,is_null: 1, // This indicates that the value is NULL}))}252 => {let (i, number) = le_u16(i)?;Ok((i, FleMessage{value: number as u64,length: 3,is_null: 0, // This indicates that the value is not NULL}))}253 => {let (i, number) = le_u32(i)?;Ok((i, FleMessage {value: number as u64,length: 5,is_null: 0, // This indicates that the value is not NULL}))}254 => {let (i, number) = le_u64(i)?;Ok((i, FleMessage {value: number,length: 9,is_null: 0, // This indicates that the value is not NULL}))}_ => {Ok((i, FleMessage {value: tag as u64,length: 1,is_null: 0, // This indicates that the value is not NULL}))}}}