# Copyright (c) 2009, 2022, Oracle and/or its affiliates.
|
#
|
# This program is free software; you can redistribute it and/or modify
|
# it under the terms of the GNU General Public License, version 2.0, as
|
# published by the Free Software Foundation.
|
#
|
# This program is also distributed with certain software (including
|
# but not limited to OpenSSL) that is licensed under separate terms,
|
# as designated in a particular file or component or in included license
|
# documentation. The authors of MySQL hereby grant you an
|
# additional permission to link the program and your derivative works
|
# with the separately licensed software that they have included with
|
# MySQL.
|
#
|
# Without limiting anything contained in the foregoing, this file,
|
# which is part of MySQL Connector/Python, is also subject to the
|
# Universal FOSS Exception, version 1.0, a copy of which can be found at
|
# http://oss.oracle.com/licenses/universal-foss-exception.
|
#
|
# This program is distributed in the hope that it will be useful, but
|
# WITHOUT ANY WARRANTY; without even the implied warranty of
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
|
# See the GNU General Public License, version 2.0, for more details.
|
#
|
# You should have received a copy of the GNU General Public License
|
# along with this program; if not, write to the Free Software Foundation, Inc.,
|
# 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
|
|
# mypy: disable-error-code="assignment"
|
|
"""Implements the MySQL Client/Server protocol."""
|
|
import datetime
|
import struct
|
|
from decimal import Decimal, DecimalException
|
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
|
|
from . import utils
|
from .authentication import get_auth_plugin
|
from .constants import (
|
PARAMETER_COUNT_AVAILABLE,
|
ClientFlag,
|
FieldFlag,
|
FieldType,
|
ServerCmd,
|
)
|
from .errors import DatabaseError, InterfaceError, ProgrammingError, get_exception
|
from .types import (
|
ConnAttrsType,
|
DescriptionType,
|
EofPacketType,
|
HandShakeType,
|
OkPacketType,
|
ParseValueFromBinaryResultPacketTypes,
|
QueryAttrType,
|
SocketType,
|
StatsPacketType,
|
StrOrBytes,
|
SupportedMysqlBinaryProtocolTypes,
|
)
|
|
PROTOCOL_VERSION: int = 10
|
|
|
class MySQLProtocol:
|
"""Implements MySQL client/server protocol
|
|
Create and parses MySQL packets.
|
"""
|
|
@staticmethod
|
def _connect_with_db(client_flags: int, database: Optional[str]) -> bytes:
|
"""Prepare database string for handshake response"""
|
if client_flags & ClientFlag.CONNECT_WITH_DB and database:
|
return database.encode("utf8") + b"\x00"
|
return b"\x00"
|
|
@staticmethod
|
def _auth_response(
|
client_flags: int,
|
username: Optional[StrOrBytes],
|
password: Optional[str],
|
database: Optional[str],
|
auth_plugin_class: str,
|
auth_plugin: str,
|
auth_data: Optional[bytes],
|
ssl_enabled: bool,
|
) -> bytes:
|
"""Prepare the authentication response"""
|
if not password:
|
return b"\x00"
|
|
try:
|
auth = get_auth_plugin(auth_plugin, auth_plugin_class)(
|
auth_data,
|
username=username,
|
password=password,
|
database=database,
|
ssl_enabled=ssl_enabled,
|
)
|
plugin_auth_response = auth.auth_response()
|
# # We could receive a NULL response, an Interface error is called when so
|
# if plugin_auth_response is None:
|
# raise InterfaceError
|
except (TypeError, InterfaceError) as err:
|
raise InterfaceError(f"Failed authentication: {err}") from err
|
|
if client_flags & ClientFlag.SECURE_CONNECTION:
|
resplen = len(plugin_auth_response)
|
auth_response = struct.pack("<B", resplen) + plugin_auth_response
|
else:
|
auth_response = plugin_auth_response + b"\x00"
|
return auth_response
|
|
def make_auth(
|
self,
|
handshake: Optional[HandShakeType],
|
username: Optional[StrOrBytes] = None,
|
password: Optional[str] = None,
|
database: Optional[str] = None,
|
charset: int = 45,
|
client_flags: int = 0,
|
max_allowed_packet: int = 1073741824,
|
ssl_enabled: bool = False,
|
auth_plugin: Optional[str] = None,
|
conn_attrs: Optional[ConnAttrsType] = None,
|
auth_plugin_class: Optional[str] = None,
|
) -> bytes:
|
"""Make a MySQL Authentication packet"""
|
if handshake is None:
|
handshake = {}
|
try:
|
auth_data: Optional[bytes] = handshake["auth_data"]
|
auth_plugin = auth_plugin or handshake["auth_plugin"]
|
except (TypeError, KeyError) as err:
|
raise ProgrammingError(
|
f"Handshake misses authentication info ({err})"
|
) from None
|
|
if not username:
|
username = b""
|
try:
|
username_bytes = username.encode("utf8") # type: ignore[union-attr]
|
except AttributeError:
|
# Username is already bytes
|
username_bytes = username
|
|
filler = "x" * 22
|
username_len = len(username_bytes)
|
|
packet = struct.pack(
|
f"<IIH{filler}{username_len}sx",
|
client_flags,
|
max_allowed_packet,
|
charset,
|
username_bytes,
|
)
|
|
packet += self._auth_response(
|
client_flags,
|
username,
|
password,
|
database,
|
auth_plugin_class,
|
auth_plugin,
|
auth_data,
|
ssl_enabled,
|
)
|
|
packet += self._connect_with_db(client_flags, database)
|
|
if client_flags & ClientFlag.PLUGIN_AUTH:
|
packet += auth_plugin.encode("utf8") + b"\x00"
|
|
if (client_flags & ClientFlag.CONNECT_ARGS) and conn_attrs is not None:
|
packet += self.make_conn_attrs(conn_attrs)
|
|
return packet
|
|
@staticmethod
|
def make_conn_attrs(conn_attrs: ConnAttrsType) -> bytes:
|
"""Encode the connection attributes"""
|
for attr_name in conn_attrs:
|
if conn_attrs[attr_name] is None:
|
conn_attrs[attr_name] = ""
|
conn_attrs_len = (
|
sum(len(x) + len(conn_attrs[x]) for x in conn_attrs)
|
+ len(conn_attrs.keys())
|
+ len(conn_attrs.values())
|
)
|
|
conn_attrs_packet = struct.pack("<B", conn_attrs_len)
|
for attr_name in conn_attrs:
|
conn_attrs_packet += struct.pack("<B", len(attr_name))
|
conn_attrs_packet += attr_name.encode("utf8")
|
conn_attrs_packet += struct.pack("<B", len(conn_attrs[attr_name]))
|
conn_attrs_packet += conn_attrs[attr_name].encode("utf8") # type: ignore[union-attr]
|
return conn_attrs_packet
|
|
@staticmethod
|
def make_auth_ssl(
|
charset: int = 45, client_flags: int = 0, max_allowed_packet: int = 1073741824
|
) -> bytearray:
|
"""Make a SSL authentication packet"""
|
return (
|
utils.int4store(client_flags)
|
+ utils.int4store(max_allowed_packet)
|
+ utils.int2store(charset)
|
+ b"\x00" * 22
|
)
|
|
@staticmethod
|
def make_command(command: int, argument: Optional[bytes] = None) -> bytearray:
|
"""Make a MySQL packet containing a command"""
|
data = utils.int1store(command)
|
if argument is not None:
|
data += argument
|
return data
|
|
@staticmethod
|
def make_stmt_fetch(statement_id: int, rows: int = 1) -> bytearray:
|
"""Make a MySQL packet with Fetch Statement command"""
|
return utils.int4store(statement_id) + utils.int4store(rows)
|
|
def make_change_user(
|
self,
|
handshake: HandShakeType,
|
username: Optional[StrOrBytes] = None,
|
password: Optional[str] = None,
|
database: Optional[str] = None,
|
charset: int = 45,
|
client_flags: int = 0,
|
ssl_enabled: bool = False,
|
auth_plugin: Optional[str] = None,
|
conn_attrs: Optional[ConnAttrsType] = None,
|
auth_plugin_class: Optional[str] = None,
|
) -> bytes:
|
"""Make a MySQL packet with the Change User command"""
|
|
try:
|
auth_data: Optional[bytes] = handshake["auth_data"]
|
auth_plugin = auth_plugin or handshake["auth_plugin"]
|
except (TypeError, KeyError) as err:
|
raise ProgrammingError(
|
f"Handshake misses authentication info ({err})"
|
) from None
|
|
if not username:
|
username = b""
|
try:
|
username_bytes = username.encode("utf8") # type: ignore[union-attr]
|
except AttributeError:
|
# Username is already bytes
|
username_bytes = username
|
|
username_len = len(username_bytes)
|
packet = struct.pack(
|
f"<B{username_len}sx",
|
ServerCmd.CHANGE_USER,
|
username_bytes,
|
)
|
|
packet += self._auth_response(
|
client_flags,
|
username,
|
password,
|
database,
|
auth_plugin_class,
|
auth_plugin,
|
auth_data,
|
ssl_enabled,
|
)
|
|
packet += self._connect_with_db(client_flags, database)
|
|
packet += struct.pack("<H", charset)
|
|
if client_flags & ClientFlag.PLUGIN_AUTH:
|
packet += auth_plugin.encode("utf8") + b"\x00"
|
|
if (client_flags & ClientFlag.CONNECT_ARGS) and conn_attrs is not None:
|
packet += self.make_conn_attrs(conn_attrs)
|
|
return packet
|
|
@staticmethod
|
def parse_handshake(packet: bytes) -> HandShakeType:
|
"""Parse a MySQL Handshake-packet"""
|
res = {}
|
res["protocol"] = struct.unpack("<xxxxB", packet[0:5])[0]
|
if res["protocol"] != PROTOCOL_VERSION:
|
raise DatabaseError(
|
f"Protocol mismatch; server version = {res['protocol']}, "
|
f"client version = {PROTOCOL_VERSION}"
|
)
|
packet, res["server_version_original"] = utils.read_string(
|
packet[5:], end=b"\x00"
|
)
|
|
(
|
res["server_threadid"],
|
auth_data1,
|
capabilities1,
|
res["charset"],
|
res["server_status"],
|
capabilities2,
|
auth_data_length,
|
) = struct.unpack("<I8sx2sBH2sBxxxxxxxxxx", packet[0:31])
|
res["server_version_original"] = res["server_version_original"].decode()
|
|
packet = packet[31:]
|
|
capabilities = utils.intread(capabilities1 + capabilities2)
|
auth_data2 = b""
|
if capabilities & ClientFlag.SECURE_CONNECTION:
|
size = min(13, auth_data_length - 8) if auth_data_length else 13
|
auth_data2 = packet[0:size]
|
packet = packet[size:]
|
if auth_data2[-1] == 0:
|
auth_data2 = auth_data2[:-1]
|
|
if capabilities & ClientFlag.PLUGIN_AUTH:
|
if b"\x00" not in packet and res["server_version_original"].startswith(
|
"5.5.8"
|
):
|
# MySQL server 5.5.8 has a bug where end byte is not send
|
(packet, res["auth_plugin"]) = (b"", packet)
|
else:
|
(packet, res["auth_plugin"]) = utils.read_string(packet, end=b"\x00")
|
res["auth_plugin"] = res["auth_plugin"].decode("utf-8")
|
else:
|
res["auth_plugin"] = "mysql_native_password"
|
|
res["auth_data"] = auth_data1 + auth_data2
|
res["capabilities"] = capabilities
|
return res
|
|
@staticmethod
|
def parse_auth_next_factor(packet: bytes) -> Tuple[bytes, str]:
|
"""Parse a MySQL AuthNextFactor packet."""
|
packet, status = utils.read_int(packet, 1)
|
if status != 2:
|
raise InterfaceError("Failed parsing AuthNextFactor packet (invalid)")
|
packet, auth_plugin = utils.read_string(packet, end=b"\x00")
|
return packet, auth_plugin.decode("utf-8")
|
|
@staticmethod
|
def parse_ok(packet: bytes) -> OkPacketType:
|
"""Parse a MySQL OK-packet"""
|
if not packet[4] == 0:
|
raise InterfaceError("Failed parsing OK packet (invalid).")
|
|
ok_packet = {}
|
try:
|
ok_packet["field_count"] = struct.unpack("<xxxxB", packet[0:5])[0]
|
packet, ok_packet["affected_rows"] = utils.read_lc_int(packet[5:])
|
packet, ok_packet["insert_id"] = utils.read_lc_int(packet)
|
(
|
ok_packet["status_flag"],
|
ok_packet["warning_count"],
|
) = struct.unpack("<HH", packet[0:4])
|
packet = packet[4:]
|
if packet:
|
packet, ok_packet["info_msg"] = utils.read_lc_string(packet)
|
ok_packet["info_msg"] = ok_packet["info_msg"].decode("utf-8")
|
except ValueError as err:
|
raise InterfaceError("Failed parsing OK packet.") from err
|
return ok_packet
|
|
@staticmethod
|
def parse_column_count(packet: bytes) -> Optional[int]:
|
"""Parse a MySQL packet with the number of columns in result set"""
|
try:
|
count = utils.read_lc_int(packet[4:])[1]
|
return count
|
except (struct.error, ValueError) as err:
|
raise InterfaceError("Failed parsing column count") from err
|
|
@staticmethod
|
def parse_column(packet: bytes, encoding: str = "utf-8") -> DescriptionType:
|
"""Parse a MySQL column-packet"""
|
packet, _ = utils.read_lc_string(packet[4:]) # catalog
|
packet, _ = utils.read_lc_string(packet) # db
|
packet, _ = utils.read_lc_string(packet) # table
|
packet, _ = utils.read_lc_string(packet) # org_table
|
packet, name = utils.read_lc_string(packet) # name
|
packet, _ = utils.read_lc_string(packet) # org_name
|
|
try:
|
(
|
charset,
|
_,
|
column_type,
|
flags,
|
_,
|
) = struct.unpack("<xHIBHBxx", packet)
|
except struct.error:
|
raise InterfaceError("Failed parsing column information") from None
|
|
return (
|
name.decode(encoding),
|
column_type,
|
None, # display_size
|
None, # internal_size
|
None, # precision
|
None, # scale
|
~flags & FieldFlag.NOT_NULL, # null_ok
|
flags, # MySQL specific
|
charset,
|
)
|
|
def parse_eof(self, packet: bytes) -> EofPacketType:
|
"""Parse a MySQL EOF-packet"""
|
if packet[4] == 0:
|
# EOF packet deprecation
|
return self.parse_ok(packet)
|
|
err_msg = "Failed parsing EOF packet."
|
res = {}
|
try:
|
unpacked = struct.unpack("<xxxBBHH", packet)
|
except struct.error as err:
|
raise InterfaceError(err_msg) from err
|
|
if not (unpacked[1] == 254 and len(packet) <= 9):
|
raise InterfaceError(err_msg)
|
|
res["warning_count"] = unpacked[2]
|
res["status_flag"] = unpacked[3]
|
return res
|
|
@staticmethod
|
def parse_statistics(packet: bytes, with_header: bool = True) -> StatsPacketType:
|
"""Parse the statistics packet"""
|
errmsg = "Failed getting COM_STATISTICS information"
|
res: Dict[str, Union[int, Decimal]] = {}
|
# Information is separated by 2 spaces
|
pairs: List[bytes] = [b""]
|
lbl: StrOrBytes = b""
|
if with_header:
|
pairs = packet[4:].split(b"\x20\x20")
|
else:
|
pairs = packet.split(b"\x20\x20")
|
for pair in pairs:
|
try:
|
lbl, val = [v.strip() for v in pair.split(b":", 2)]
|
except ValueError as err:
|
raise InterfaceError(errmsg) from err
|
|
# It's either an integer or a decimal
|
lbl = lbl.decode("utf-8")
|
try:
|
res[lbl] = int(val)
|
except (KeyError, ValueError):
|
try:
|
res[lbl] = Decimal(val.decode("utf-8"))
|
except DecimalException as err:
|
raise InterfaceError(f"{errmsg} ({lbl}:{repr(val)})") from err
|
return res
|
|
def read_text_result(
|
self, sock: SocketType, version: Tuple[int, ...], count: int = 1
|
) -> Tuple[List[Tuple[Optional[bytes], ...]], Optional[EofPacketType],]:
|
"""Read MySQL text result
|
|
Reads all or given number of rows from the socket.
|
|
Returns a tuple with 2 elements: a list with all rows and
|
the EOF packet.
|
"""
|
# Keep unused 'version' for API backward compatibility
|
_ = version
|
rows = []
|
eof = None
|
rowdata = None
|
i = 0
|
while True:
|
if eof or i == count:
|
break
|
packet = sock.recv()
|
if packet.startswith(b"\xff\xff\xff"):
|
datas = [packet[4:]]
|
packet = sock.recv()
|
while packet.startswith(b"\xff\xff\xff"):
|
datas.append(packet[4:])
|
packet = sock.recv()
|
datas.append(packet[4:])
|
rowdata = utils.read_lc_string_list(bytearray(b"").join(datas))
|
elif packet[4] == 254 and packet[0] < 7:
|
eof = self.parse_eof(packet)
|
rowdata = None
|
else:
|
eof = None
|
rowdata = utils.read_lc_string_list(packet[4:])
|
if eof is None and rowdata is not None:
|
rows.append(rowdata)
|
elif eof is None and rowdata is None:
|
raise get_exception(packet)
|
i += 1
|
return rows, eof
|
|
@staticmethod
|
def _parse_binary_integer(
|
packet: bytes, field: DescriptionType
|
) -> Tuple[bytes, int]:
|
"""Parse an integer from a binary packet"""
|
if field[1] == FieldType.TINY:
|
format_ = "<b"
|
length = 1
|
elif field[1] == FieldType.SHORT:
|
format_ = "<h"
|
length = 2
|
elif field[1] in (FieldType.INT24, FieldType.LONG):
|
format_ = "<i"
|
length = 4
|
elif field[1] == FieldType.LONGLONG:
|
format_ = "<q"
|
length = 8
|
|
if field[7] & FieldFlag.UNSIGNED:
|
format_ = format_.upper()
|
|
return (packet[length:], struct.unpack(format_, packet[0:length])[0])
|
|
@staticmethod
|
def _parse_binary_float(
|
packet: bytes, field: DescriptionType
|
) -> Tuple[bytes, float]:
|
"""Parse a float/double from a binary packet"""
|
if field[1] == FieldType.DOUBLE:
|
length = 8
|
format_ = "<d"
|
else:
|
length = 4
|
format_ = "<f"
|
|
return (packet[length:], struct.unpack(format_, packet[0:length])[0])
|
|
@staticmethod
|
def _parse_binary_new_decimal(
|
packet: bytes, charset: str = "utf8"
|
) -> Tuple[bytes, Decimal]:
|
"""Parse a New Decimal from a binary packet"""
|
(packet, value) = utils.read_lc_string(packet)
|
return (packet, Decimal(value.decode(charset)))
|
|
@staticmethod
|
def _parse_binary_timestamp(
|
packet: bytes,
|
field_type: int,
|
) -> Tuple[bytes, Optional[Union[datetime.date, datetime.datetime]]]:
|
"""Parse a timestamp from a binary packet"""
|
length = packet[0]
|
value = None
|
if length == 4:
|
year = struct.unpack("<H", packet[1:3])[0]
|
month = packet[3]
|
day = packet[4]
|
if field_type in (FieldType.DATETIME, FieldType.TIMESTAMP):
|
value = datetime.datetime(year=year, month=month, day=day)
|
else:
|
value = datetime.date(year=year, month=month, day=day)
|
elif length >= 7:
|
mcs = 0
|
if length == 11:
|
mcs = struct.unpack("<I", packet[8 : length + 1])[0]
|
value = datetime.datetime(
|
year=struct.unpack("<H", packet[1:3])[0],
|
month=packet[3],
|
day=packet[4],
|
hour=packet[5],
|
minute=packet[6],
|
second=packet[7],
|
microsecond=mcs,
|
)
|
|
return (packet[length + 1 :], value)
|
|
@staticmethod
|
def _parse_binary_time(packet: bytes) -> Tuple[bytes, datetime.timedelta]:
|
"""Parse a time value from a binary packet"""
|
length = packet[0]
|
if not length:
|
return (packet[1:], datetime.timedelta())
|
data = packet[1 : length + 1]
|
mcs = 0
|
if length > 8:
|
mcs = struct.unpack("<I", data[8:])[0]
|
days = struct.unpack("<I", data[1:5])[0]
|
if data[0] == 1:
|
days *= -1
|
tmp = datetime.timedelta(
|
days=days,
|
seconds=data[7],
|
microseconds=mcs,
|
minutes=data[6],
|
hours=data[5],
|
)
|
|
return (packet[length + 1 :], tmp)
|
|
def _parse_binary_values(
|
self,
|
fields: List[DescriptionType],
|
packet: bytes,
|
charset: str = "utf-8",
|
) -> Tuple[ParseValueFromBinaryResultPacketTypes, ...]:
|
"""Parse values from a binary result packet"""
|
null_bitmap_length = (len(fields) + 7 + 2) // 8
|
null_bitmap = [int(i) for i in packet[0:null_bitmap_length]]
|
packet = packet[null_bitmap_length:]
|
|
values: List[Any] = []
|
value: Any = None
|
for pos, field in enumerate(fields):
|
if null_bitmap[int((pos + 2) / 8)] & (1 << (pos + 2) % 8):
|
values.append(None)
|
continue
|
if field[1] in (
|
FieldType.TINY,
|
FieldType.SHORT,
|
FieldType.INT24,
|
FieldType.LONG,
|
FieldType.LONGLONG,
|
):
|
packet, value = self._parse_binary_integer(packet, field)
|
values.append(value)
|
elif field[1] in (FieldType.DOUBLE, FieldType.FLOAT):
|
packet, value = self._parse_binary_float(packet, field)
|
values.append(value)
|
elif field[1] in (FieldType.DECIMAL, FieldType.NEWDECIMAL):
|
packet, value = self._parse_binary_new_decimal(packet, charset)
|
values.append(value)
|
elif field[1] in (
|
FieldType.DATETIME,
|
FieldType.DATE,
|
FieldType.TIMESTAMP,
|
):
|
(packet, value) = self._parse_binary_timestamp(packet, field[1])
|
values.append(value)
|
elif field[1] == FieldType.TIME:
|
(packet, value) = self._parse_binary_time(packet)
|
values.append(value)
|
else:
|
(packet, value) = utils.read_lc_string(packet)
|
try:
|
values.append(value.decode(charset))
|
except UnicodeDecodeError:
|
values.append(value)
|
|
return tuple(values)
|
|
def read_binary_result(
|
self,
|
sock: SocketType,
|
columns: List[DescriptionType],
|
count: int = 1,
|
charset: str = "utf-8",
|
) -> Tuple[
|
List[Tuple[ParseValueFromBinaryResultPacketTypes, ...]],
|
Optional[EofPacketType],
|
]:
|
"""Read MySQL binary protocol result
|
|
Reads all or given number of binary resultset rows from the socket.
|
"""
|
rows = []
|
eof = None
|
values = None
|
i = 0
|
while True:
|
if eof is not None:
|
break
|
if i == count:
|
break
|
packet = sock.recv()
|
if packet[4] == 254:
|
eof = self.parse_eof(packet)
|
values = None
|
elif packet[4] == 0:
|
eof = None
|
values = self._parse_binary_values(columns, packet[5:], charset)
|
if eof is None and values is not None:
|
rows.append(values)
|
elif eof is None and values is None:
|
raise get_exception(packet)
|
i += 1
|
return (rows, eof)
|
|
@staticmethod
|
def parse_binary_prepare_ok(packet: bytes) -> Dict[str, int]:
|
"""Parse a MySQL Binary Protocol OK packet"""
|
if not packet[4] == 0:
|
raise InterfaceError("Failed parsing Binary OK packet")
|
|
ok_pkt = {}
|
try:
|
packet, ok_pkt["statement_id"] = utils.read_int(packet[5:], 4)
|
packet, ok_pkt["num_columns"] = utils.read_int(packet, 2)
|
packet, ok_pkt["num_params"] = utils.read_int(packet, 2)
|
packet = packet[1:] # Filler 1 * \x00
|
packet, ok_pkt["warning_count"] = utils.read_int(packet, 2)
|
except ValueError as err:
|
raise InterfaceError("Failed parsing Binary OK packet") from err
|
|
return ok_pkt
|
|
@staticmethod
|
def prepare_binary_integer(value: int) -> Tuple[bytes, int, int]:
|
"""Prepare an integer for the MySQL binary protocol"""
|
field_type = None
|
flags = 0
|
if value < 0:
|
if value >= -128:
|
format_ = "<b"
|
field_type = FieldType.TINY
|
elif value >= -32768:
|
format_ = "<h"
|
field_type = FieldType.SHORT
|
elif value >= -2147483648:
|
format_ = "<i"
|
field_type = FieldType.LONG
|
else:
|
format_ = "<q"
|
field_type = FieldType.LONGLONG
|
else:
|
flags = 128
|
if value <= 255:
|
format_ = "<B"
|
field_type = FieldType.TINY
|
elif value <= 65535:
|
format_ = "<H"
|
field_type = FieldType.SHORT
|
elif value <= 4294967295:
|
format_ = "<I"
|
field_type = FieldType.LONG
|
else:
|
field_type = FieldType.LONGLONG
|
format_ = "<Q"
|
return (struct.pack(format_, value), field_type, flags)
|
|
@staticmethod
|
def prepare_binary_timestamp(
|
value: Union[datetime.date, datetime.datetime]
|
) -> Tuple[bytearray, int]:
|
"""Prepare a timestamp object for the MySQL binary protocol
|
|
This method prepares a timestamp of type datetime.datetime or
|
datetime.date for sending over the MySQL binary protocol.
|
A tuple is returned with the prepared value and field type
|
as elements.
|
|
Raises ValueError when the argument value is of invalid type.
|
|
Returns a tuple.
|
"""
|
if isinstance(value, datetime.datetime):
|
field_type = FieldType.DATETIME
|
elif isinstance(value, datetime.date):
|
field_type = FieldType.DATE
|
else:
|
raise ValueError("Argument must a datetime.datetime or datetime.date")
|
|
packed = (
|
utils.int2store(value.year)
|
+ utils.int1store(value.month)
|
+ utils.int1store(value.day)
|
)
|
|
if isinstance(value, datetime.datetime):
|
packed = (
|
packed
|
+ utils.int1store(value.hour)
|
+ utils.int1store(value.minute)
|
+ utils.int1store(value.second)
|
)
|
if value.microsecond > 0:
|
packed += utils.int4store(value.microsecond)
|
|
packed = utils.int1store(len(packed)) + packed
|
return (packed, field_type)
|
|
@staticmethod
|
def prepare_binary_time(
|
value: Union[datetime.timedelta, datetime.time]
|
) -> Tuple[bytearray, int]:
|
"""Prepare a time object for the MySQL binary protocol
|
|
This method prepares a time object of type datetime.timedelta or
|
datetime.time for sending over the MySQL binary protocol.
|
A tuple is returned with the prepared value and field type
|
as elements.
|
|
Raises ValueError when the argument value is of invalid type.
|
|
Returns a tuple.
|
"""
|
if not isinstance(value, (datetime.timedelta, datetime.time)):
|
raise ValueError("Argument must a datetime.timedelta or datetime.time")
|
|
field_type = FieldType.TIME
|
negative = 0
|
mcs = None
|
packed = b""
|
|
if isinstance(value, datetime.timedelta):
|
if value.days < 0:
|
negative = 1
|
(hours, remainder) = divmod(value.seconds, 3600)
|
(mins, secs) = divmod(remainder, 60)
|
packed += (
|
utils.int4store(abs(value.days))
|
+ utils.int1store(hours)
|
+ utils.int1store(mins)
|
+ utils.int1store(secs)
|
)
|
mcs = value.microseconds
|
else:
|
packed += (
|
utils.int4store(0)
|
+ utils.int1store(value.hour)
|
+ utils.int1store(value.minute)
|
+ utils.int1store(value.second)
|
)
|
mcs = value.microsecond
|
if mcs:
|
packed += utils.int4store(mcs)
|
|
packed = utils.int1store(negative) + packed
|
packed = utils.int1store(len(packed)) + packed
|
|
return (packed, field_type)
|
|
@staticmethod
|
def prepare_stmt_send_long_data(
|
statement: int, param: int, data: bytes
|
) -> bytearray:
|
"""Prepare long data for prepared statements
|
|
Returns a string.
|
"""
|
packet: bytearray = utils.int4store(statement) + utils.int2store(param) + data
|
return packet
|
|
def make_stmt_execute(
|
self,
|
statement_id: int,
|
data: Sequence[SupportedMysqlBinaryProtocolTypes] = (),
|
parameters: Sequence[Any] = (),
|
flags: int = 0,
|
long_data_used: Optional[Dict[int, Tuple[bool]]] = None,
|
charset: str = "utf8",
|
query_attrs: Optional[QueryAttrType] = None,
|
converter_str_fallback: bool = False,
|
) -> bytearray:
|
"""Make a MySQL packet with the Statement Execute command"""
|
iteration_count = 1
|
null_bitmap = [0] * ((len(data) + 7) // 8)
|
values = []
|
types = []
|
packed = b""
|
data_len = len(data)
|
query_attr_names = []
|
flags = flags if not query_attrs else flags + PARAMETER_COUNT_AVAILABLE
|
if charset == "utf8mb4":
|
charset = "utf8"
|
if long_data_used is None:
|
long_data_used = {}
|
if query_attrs:
|
data = list(data)
|
for _, attr_val in query_attrs:
|
data.append(attr_val)
|
null_bitmap = [0] * ((len(data) + 7) // 8)
|
if parameters or data:
|
if data_len != len(parameters):
|
raise InterfaceError(
|
"Failed executing prepared statement: data values does not"
|
" match number of parameters"
|
)
|
for pos, value in enumerate(data):
|
_flags = 0
|
if value is None:
|
null_bitmap[(pos // 8)] |= 1 << (pos % 8)
|
types.append(
|
utils.int1store(FieldType.NULL) + utils.int1store(_flags)
|
)
|
continue
|
if pos in long_data_used:
|
if long_data_used[pos][0]:
|
# We suppose binary data
|
field_type = FieldType.BLOB
|
else:
|
# We suppose text data
|
field_type = FieldType.STRING
|
elif isinstance(value, int):
|
(
|
packed,
|
field_type,
|
_flags,
|
) = self.prepare_binary_integer(value)
|
values.append(packed)
|
elif isinstance(value, str):
|
value = value.encode(charset)
|
values.append(utils.lc_int(len(value)) + value)
|
field_type = FieldType.VARCHAR
|
elif isinstance(value, bytes):
|
values.append(utils.lc_int(len(value)) + value)
|
field_type = FieldType.BLOB
|
elif isinstance(value, Decimal):
|
values.append(
|
utils.lc_int(len(str(value).encode(charset)))
|
+ str(value).encode(charset)
|
)
|
field_type = FieldType.DECIMAL
|
elif isinstance(value, float):
|
values.append(struct.pack("<d", value))
|
field_type = FieldType.DOUBLE
|
elif isinstance(value, (datetime.datetime, datetime.date)):
|
(packed, field_type) = self.prepare_binary_timestamp(value)
|
values.append(packed)
|
elif isinstance(value, (datetime.timedelta, datetime.time)):
|
(packed, field_type) = self.prepare_binary_time(value)
|
values.append(packed)
|
elif converter_str_fallback:
|
value = str(value).encode(charset)
|
values.append(utils.lc_int(len(value)) + value)
|
field_type = FieldType.STRING
|
else:
|
raise ProgrammingError(
|
"MySQL binary protocol can not handle "
|
f"'{value.__class__.__name__}' objects"
|
)
|
types.append(utils.int1store(field_type) + utils.int1store(_flags))
|
if query_attrs and pos + 1 > data_len:
|
name = query_attrs[pos - data_len][0].encode(charset)
|
query_attr_names.append(utils.lc_int(len(name)) + name)
|
packet = (
|
utils.int4store(statement_id)
|
+ utils.int1store(flags)
|
+ utils.int4store(iteration_count)
|
)
|
|
# if (num_params > 0 || (CLIENT_QUERY_ATTRIBUTES \
|
# && (flags & PARAMETER_COUNT_AVAILABLE)) {
|
if query_attrs is not None:
|
parameter_count = data_len + len(query_attrs)
|
else:
|
parameter_count = data_len
|
if parameter_count:
|
# if CLIENT_QUERY_ATTRIBUTES is on
|
if query_attrs is not None:
|
packet += utils.lc_int(parameter_count)
|
|
packet += b"".join(
|
[struct.pack("B", bit) for bit in null_bitmap]
|
) + utils.int1store(1)
|
count = 0
|
for a_type in types:
|
packet += a_type
|
# if CLIENT_QUERY_ATTRIBUTES is on {
|
# string<lenenc> parameter_name Name of the parameter
|
# or empty if not present
|
# } if CLIENT_QUERY_ATTRIBUTES is on
|
if query_attrs is not None:
|
if count + 1 > data_len:
|
packet += query_attr_names[count - data_len]
|
else:
|
packet += b"\x00"
|
count += 1
|
|
for a_value in values:
|
packet += a_value
|
|
return packet
|
|
@staticmethod
|
def parse_auth_switch_request(packet: bytes) -> Tuple[str, bytes]:
|
"""Parse a MySQL AuthSwitchRequest-packet"""
|
if not packet[4] == 254:
|
raise InterfaceError("Failed parsing AuthSwitchRequest packet")
|
|
packet, plugin_name = utils.read_string(packet[5:], end=b"\x00")
|
if packet and packet[-1] == 0:
|
packet = packet[:-1]
|
|
return plugin_name.decode("utf8"), packet
|
|
@staticmethod
|
def parse_auth_more_data(packet: bytes) -> bytes:
|
"""Parse a MySQL AuthMoreData-packet"""
|
if not packet[4] == 1:
|
raise InterfaceError("Failed parsing AuthMoreData packet")
|
|
return packet[5:]
|
