Building your own MySQL connector (X Protocol)

README.md

How to use

$ docker-compose build
$ docker-compose up -d
$ dokcer-compose run client bash test.sh
type: SINT
name: "id"
original_name: "id"
table: "bar"
original_table: "bar"
schema: "foo"
catalog: "def"
length: 11

type: BYTES
name: "name"
original_name: "name"
table: "bar"
original_table: "bar"
schema: "foo"
catalog: "def"
collation: 255
length: 80

field: "\002"
field: "hoge\000"

field: "\024"
field: "piyo\000"

row1_id=1, row1_name='hoge'
row2_id=10, row2_name='piyo'

docker-compose.yml

version: '3'
services:
  client:
    build: .
    tty: true
  db:
    image: mysql:8.0.20
    environment:
      - MYSQL_ROOT_PASSWORD=test
    command: mysqld --default-authentication-plugin=mysql_native_password

Dockerfile

FROM ubuntu:20.04

ENV DEBIAN_FRONTEND=noninteractive
RUN apt-get update && \
    apt-get install -y vim wget git lsb-release gnupg ngrep ipython3 python3-pip protobuf-compiler

RUN pip3 install mysql-connector-python

# MySQL SHELL
RUN wget https://dev.mysql.com/get/mysql-apt-config_0.8.15-1_all.deb && \
    dpkg -i mysql-apt-config_0.8.15-1_all.deb && \
    rm mysql-apt-config_0.8.15-1_all.deb && \
    apt-get update && \
    apt-get install -y mysql-shell

RUN mkdir -p /app && \
    cd /app && \
    git clone --depth 1 -b mysql-8.0.21 https://github.com/mysql/mysql-server && \
    cp -r mysql-server/plugin/x/protocol/protobuf/ . && \
    rm -rf mysql-server && \
    mkdir lib && \
    protoc -I=protobuf --python_out=lib protobuf/*.proto

COPY main.py /app/main.py
COPY test.sh /app/test.sh
ENV PYTHONPATH=$PYTHONPATH:lib

WORKDIR /app

main.py

import socket
import struct
from hashlib import sha1

import mysqlx_session_pb2
import mysqlx_notice_pb2
import mysqlx_sql_pb2
import mysqlx_resultset_pb2

def read_packet(sock):
    size = sock.recv(4)
    size = struct.unpack('<I', size)[0]
    typ = sock.recv(1)
    payload = sock.recv(size - 1)

    if typ == b"\x0b":  # Ignore `Notice`
        return read_packet(sock)

    return typ, payload

host = "db"
user = "root"
password = "test"
port = 33060

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((host, port))

# C -> S: SESS_AUTHENTICATE_START
auth = mysqlx_session_pb2.AuthenticateStart()
auth.mech_name = 'MYSQL41'
payload = auth.SerializeToString()
size = struct.pack("<I", len(payload) + 1)
SESS_AUTHENTICATE_START = 4
typ = struct.pack('B', SESS_AUTHENTICATE_START)
sock.send(size + typ + payload)

# S -> C: AuthenticateContinue
typ, payload = read_packet(sock)
auth_conti = mysqlx_session_pb2.AuthenticateContinue()
auth_conti.ParseFromString(payload)
challenge = auth_conti.auth_data

# C -> S: AuthenticateContinue
x = sha1(password.encode()).digest()
y = sha1(challenge + sha1(x).digest()).digest()
auth_data = bytes([i ^ j for i, j in zip(x,y)]).hex().encode()
auth_data = b"\x00" + user.encode() + b"\x00\x2a" + auth_data + b"\x00"
auth_conti = mysqlx_session_pb2.AuthenticateContinue()
auth_conti.auth_data = auth_data
payload = auth_conti.SerializeToString()
size = len(payload) + 1
size = struct.pack('<I', size)
CLIENT_SESS_AUTHENTICATE_CONTINUE  = 5
typ = struct.pack('B', CLIENT_SESS_AUTHENTICATE_CONTINUE)
sock.send(size + typ + payload)

# S -> C: SESS_AUTHENTICATE_OK = 4
typ, payload = read_packet(sock)
auth_ok = mysqlx_session_pb2.AuthenticateOk()
auth_ok.ParseFromString(payload)

# C -> S: SQL_STMT_EXECUTE
stmt_execute = mysqlx_sql_pb2.StmtExecute()
stmt_execute.stmt = b"select * from foo.bar"
payload = stmt_execute.SerializeToString()
size = len(payload) + 1
size = struct.pack('<I', size)
SQL_STMT_EXECUTE = 12
typ = struct.pack('B', SQL_STMT_EXECUTE)
sock.send(size + typ + payload)

# id
typ, payload = read_packet(sock)
resultset_id = mysqlx_resultset_pb2.ColumnMetaData()
resultset_id.ParseFromString(payload)
print(resultset_id)

# name
typ, payload = read_packet(sock)
resultset_name = mysqlx_resultset_pb2.ColumnMetaData()
resultset_name.ParseFromString(payload)
print(resultset_name)

# Row
typ, payload = read_packet(sock)
row1 = mysqlx_resultset_pb2.Row()
row1.ParseFromString(payload)
print(row1)

typ, payload = read_packet(sock)
row2 = mysqlx_resultset_pb2.Row()
row2.ParseFromString(payload)
print(row2)


def rshift(val, n):
    return val>>n if val >= 0 else (val+0x100000000)>>n

def decode_zigzag(val):
    return rshift(val, 1) ^ - (val & 1)
# https://gist.github.com/mfuerstenau/ba870a29e16536fdbaba#file-zigzag-encoding-readme-L18
# https://stackoverflow.com/a/5833119
row1_id = decode_zigzag(struct.unpack('B', row1.field[0])[0])
row1_name = row1.field[1].decode('utf-8')[0:-1]
print(f"{row1_id=}, {row1_name=}")
row2_id = decode_zigzag(struct.unpack('B', row2.field[0])[0])
row2_name = row2.field[1].decode('utf-8')[0:-1]
print(f"{row2_id=}, {row2_name=}")

# RESULTSET_FETCH_DONE
typ, payload = read_packet(sock)
fetch_done = mysqlx_resultset_pb2.FetchDone()
fetch_done.ParseFromString(payload)

# SQL_STMT_EXECUTE_OK
typ, payload = read_packet(sock)
stmt_execute_ok = mysqlx_sql_pb2.StmtExecuteOk()
stmt_execute_ok.ParseFromString(payload)

test.sh

#!/bin/bash

mysqlsh root@db:33060 -ptest --sql -e "CREATE DATABASE foo; CREATE TABLE foo.bar (id int, name varchar(20)); INSERT INTO foo.bar values (1, 'hoge'), (10, 'piyo')"

python3 main.py