-- 查看当前连接列表
SHOW PROCESSLIST;
-- Id 列就是 connection_id

-- 终止某个连接
KILL 123;

# 典型的客户端能力标志
CLIENT_FLAGS = (
    CLIENT_PROTOCOL_41 |              # 4.1 协议格式
    CLIENT_SECURE_CONNECTION |        # 安全连接
    CLIENT_PLUGIN_AUTH |              # 认证插件
    CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA |  # 长度编码认证数据
    CLIENT_CONNECT_WITH_DB |          # 可选：指定数据库
    CLIENT_CONNECT_ATTRS |            # 可选：连接属性
    CLIENT_DEPRECATE_EOF |            # 可选：使用 OK 替代 EOF
    CLIENT_MULTI_STATEMENTS |         # 可选：多语句
    CLIENT_MULTI_RESULTS              # 可选：多结果集
)

"""
mysql_ssl_connection.py
演示 MySQL 协议的 SSL/TLS 升级过程
"""
import socket
import ssl
import struct

def mysql_ssl_connect(host='127.0.0.1', port=3306,
                      username='root', password='password',
                      ca_cert='/path/to/ca.pem'):
    """建立带 SSL 的 MySQL 连接"""

    # 第一步：建立普通 TCP 连接
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((host, port))
    print(f"[+] TCP 连接建立")

    # 第二步：接收服务器握手包
    header = sock.recv(4)
    payload_length = struct.unpack('<I', header[0:3] + b'\x00')[0]
    sequence = header[3]
    server_handshake = sock.recv(payload_length)
    print(f"[+] 收到 HandshakeV10 (seq={sequence}, len={payload_length})")

    # 提取 scramble（简化版）
    # 跳过 protocol_version(1) + server_version + null
    null_pos = server_handshake.index(b'\x00', 1)
    offset = null_pos + 1
    # connection_id(4) + auth_data_part1(8) + filler(1)
    offset += 13
    cap_low = struct.unpack('<H', server_handshake[offset:offset+2])[0]

    # 第三步：发送 SSLRequest（仅前 32 字节）
    CLIENT_SSL = 1 << 11
    CLIENT_PROTOCOL_41 = 1 << 9
    CLIENT_SECURE_CONNECTION = 1 << 15
    CLIENT_PLUGIN_AUTH = 1 << 19

    capability_flags = (
        CLIENT_PROTOCOL_41 |
        CLIENT_SECURE_CONNECTION |
        CLIENT_SSL |
        CLIENT_PLUGIN_AUTH
    )

    ssl_request = struct.pack('<IIB23s',
        capability_flags,     # capability_flags
        1024 * 1024 * 16,     # max_packet_size
        33,                   # character_set (utf8)
        b'\x00' * 23          # reserved
    )

    # 添加包头 (3 字节长度 + 1 字节序列号)
    packet = struct.pack('<I', len(ssl_request))[:3] + struct.pack('B', 1) + ssl_request
    sock.send(packet)
    print(f"[+] 发送 SSLRequest")

    # 第四步：升级到 TLS
    context = ssl.create_default_context()
    context.load_verify_locations(ca_cert)
    # 如需跳过证书验证（仅测试环境）:
    # context.check_hostname = False
    # context.verify_mode = ssl.CERT_NONE

    ssl_sock = context.wrap_socket(sock, server_hostname=host)
    print(f"[+] TLS 升级成功: {ssl_sock.version()}")

    # 第五步：通过加密通道发送 HandshakeResponse41
    # （此处省略完整的认证响应构造，实际实现请参考 PyMySQL 源码）
    print("[*] 后续所有通信已加密")

    ssl_sock.close()


if __name__ == '__main__':
    mysql_ssl_connect()

-- 查看 SSL 配置
SHOW VARIABLES LIKE '%ssl%';

-- 强制所有连接使用 SSL
ALTER USER 'app_user'@'%' REQUIRE SSL;

-- 要求特定的 SSL 证书
ALTER USER 'app_user'@'%' REQUIRE ISSUER '/CN=MyCA';

"""
mysql_minimal_handshake.py
实现最简单的 MySQL 协议握手（使用 mysql_native_password 认证）
"""
import socket
import struct
import hashlib

def sha1(data):
    return hashlib.sha1(data).digest()

def mysql_native_password_hash(password, scramble):
    """
    mysql_native_password 认证算法:
    hash = SHA1(password) XOR SHA1(scramble + SHA1(SHA1(password)))
    """
    if not password:
        return b''

    password_sha1 = sha1(password.encode('utf-8'))
    password_sha1_sha1 = sha1(password_sha1)
    scramble_and_hash = sha1(scramble + password_sha1_sha1)

    # XOR 操作
    result = bytes(a ^ b for a, b in zip(password_sha1, scramble_and_hash))
    return result


def read_null_terminated_string(data, offset):
    """读取以 null 结尾的字符串"""
    end = data.index(b'\x00', offset)
    return data[offset:end].decode('ascii'), end + 1


def read_length_encoded_integer(data, offset):
    """读取长度编码的整数"""
    first = data[offset]
    if first < 0xFB:
        return first, offset + 1
    elif first == 0xFC:
        return struct.unpack('<H', data[offset+1:offset+3])[0], offset + 3
    elif first == 0xFD:
        return struct.unpack('<I', data[offset+1:offset+4] + b'\x00')[0], offset + 4
    elif first == 0xFE:
        return struct.unpack('<Q', data[offset+1:offset+9])[0], offset + 9


def minimal_handshake(host='127.0.0.1', port=3306,
                      username='root', password='', database=None):
    """实现最小化的 MySQL 协议握手"""

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(5)

    try:
        # 1. TCP 连接
        sock.connect((host, port))
        print(f"[+] TCP 连接到 {host}:{port}")

        # 2. 接收 HandshakeV10
        header = b''
        while len(header) < 4:
            header += sock.recv(4 - len(header))

        pkt_len = struct.unpack('<I', header[0:3] + b'\x00')[0]
        seq = header[3]

        payload = b''
        while len(payload) < pkt_len:
            payload += sock.recv(pkt_len - len(payload))

        print(f"[+] 收到 HandshakeV10 (seq={seq}, len={pkt_len})")

        # 解析 HandshakeV10
        offset = 0
        protocol_version = payload[offset]
        offset += 1

        server_version, offset = read_null_terminated_string(payload, offset)

        connection_id = struct.unpack('<I', payload[offset:offset+4])[0]
        offset += 4

        auth_data_part1 = payload[offset:offset+8]
        offset += 8

        filler = payload[offset]
        offset += 1

        cap_low = struct.unpack('<H', payload[offset:offset+2])[0]
        offset += 2

        charset = payload[offset]
        offset += 1

        status = struct.unpack('<H', payload[offset:offset+2])[0]
        offset += 2

        cap_high = struct.unpack('<H', payload[offset:offset+2])[0]
        offset += 2

        server_caps = (cap_high << 16) | cap_low

        auth_data_len = payload[offset]
        offset += 1
        offset += 10  # reserved

        # auth_data_part2
        part2_len = max(13, auth_data_len - 8)
        auth_data_part2 = payload[offset:offset+part2_len]
        offset += part2_len + 1  # +1 for null terminator

        scramble = auth_data_part1 + auth_data_part2

        auth_plugin, offset = read_null_terminated_string(payload, offset)

        print(f"  服务器版本: {server_version}")
        print(f"  连接 ID: {connection_id}")
        print(f"  字符集: {charset}")
        print(f"  认证插件: {auth_plugin}")
        print(f"  Scramble: {scramble.hex()}")

        # 3. 构造 HandshakeResponse41
        # 选择 mysql_native_password 以简化演示
        auth_method = 'mysql_native_password'
        auth_data = mysql_native_password_hash(password, scramble)

        CLIENT_PROTOCOL_41 = 1 << 9
        CLIENT_SECURE_CONNECTION = 1 << 15
        CLIENT_PLUGIN_AUTH = 1 << 19
        CLIENT_CONNECT_WITH_DB = 1 << 3

        client_caps = (
            CLIENT_PROTOCOL_41 |
            CLIENT_SECURE_CONNECTION |
            CLIENT_PLUGIN_AUTH
        )
        if database:
            client_caps |= CLIENT_CONNECT_WITH_DB

        # 能力标志取交集
        client_caps &= server_caps

        # 构建响应体
        body = b''
        body += struct.pack('<I', client_caps)          # capability_flags
        body += struct.pack('<I', 16 * 1024 * 1024)     # max_packet_size
        body += struct.pack('<B', 33)                    # character_set (utf8mb4=33)
        body += b'\x00' * 23                             # reserved
        body += username.encode('utf-8') + b'\x00'      # username
        body += bytes([len(auth_data)]) + auth_data      # auth_response (length-encoded)
        if database:
            body += database.encode('utf-8') + b'\x00'   # database
        body += auth_method.encode('utf-8') + b'\x00'    # auth_plugin_name

        # 添加包头
        packet = struct.pack('<I', len(body))[:3] + struct.pack('B', seq + 1) + body
        sock.send(packet)
        print(f"[+] 发送 HandshakeResponse41 (seq={seq+1})")

        # 4. 接收认证结果
        header = sock.recv(4)
        resp_len = struct.unpack('<I', header[0:3] + b'\x00')[0]
        resp_seq = header[3]

        response = b''
        while len(response) < resp_len:
            response += sock.recv(resp_len - len(response))

        print(f"[+] 收到响应 (seq={resp_seq}, len={resp_len})")

        if response[0] == 0x00:
            print("[✓] 认证成功! (OK_Packet)")
            # 解析 OK 包
            affected_rows, _ = read_length_encoded_integer(response, 1)
            print(f"  affected_rows: {affected_rows}")
        elif response[0] == 0xFF:
            # ERR_Packet
            error_code = struct.unpack('<H', response[1:3])[0]
            error_msg = response[9:].decode('utf-8', errors='replace')
            print(f"[✗] 认证失败! (ERR_Packet)")
            print(f"  错误代码: {error_code}")
            print(f"  错误消息: {error_msg}")
        elif response[0] == 0xFE:
            print("[*] 需要 AuthMoreData (额外认证步骤)")
        else:
            print(f"[?] 未知响应: {response.hex()}")

    except Exception as e:
        print(f"[-] 错误: {e}")
    finally:
        sock.close()


if __name__ == '__main__':
    minimal_handshake(
        host='127.0.0.1',
        port=3306,
        username='root',
        password='your_password_here',
        database='test'
    )

# 确保 MySQL 服务器允许 mysql_native_password 认证
mysql -u root -p -e "ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY 'your_password_here';"

# 运行脚本
python3 mysql_minimal_handshake.py

// mysql_handshake.go
package main

import (
	"crypto/sha1"
	"encoding/binary"
	"fmt"
	"net"
)

func main() {
	conn, err := net.Dial("tcp", "127.0.0.1:3306")
	if err != nil {
		panic(err)
	}
	defer conn.Close()

	fmt.Println("[+] TCP 连接建立")

	// 读取 HandshakeV10
	header := make([]byte, 4)
	conn.Read(header)
	pktLen := uint32(header[0]) | uint32(header[1])<<8 | uint32(header[2])<<16
	seq := header[3]

	payload := make([]byte, pktLen)
	conn.Read(payload)
	fmt.Printf("[+] HandshakeV10: seq=%d, len=%d\n", seq, pktLen)

	// 解析协议版本
	protoVer := payload[0]
	fmt.Printf("  协议版本: %d\n", protoVer)

	// 解析服务器版本
	end := 1
	for payload[end] != 0 {
		end++
	}
	version := string(payload[1:end])
	fmt.Printf("  服务器版本: %s\n", version)

	// 解析连接 ID
	connID := binary.LittleEndian.Uint32(payload[end+1 : end+5])
	fmt.Printf("  连接 ID: %d\n", connID)

	fmt.Println("[✓] 握手包解析完成")
}

func mysqlNativePassword(password string, scramble []byte) []byte {
	if len(password) == 0 {
		return nil
	}
	h1 := sha1.Sum([]byte(password))
	h2 := sha1.Sum(h1[:])
	h3 := sha1.Sum(append(scramble, h2[:]...))
	result := make([]byte, 20)
	for i := 0; i < 20; i++ {
		result[i] = h1[i] ^ h3[i]
	}
	return result
}