# Python implementation of the MySQL client-server protocol # http://dev.mysql.com/doc/internals/en/client-server-protocol.html from __future__ import print_function from ._compat import PY2, range_type, text_type, str_type, JYTHON, IRONPYTHON from functools import partial import os import hashlib import socket try: import ssl SSL_ENABLED = True except ImportError: SSL_ENABLED = False import struct import sys if PY2: import ConfigParser as configparser else: import configparser import io try: import getpass DEFAULT_USER = getpass.getuser() except ImportError: DEFAULT_USER = None from .charset import MBLENGTH, charset_by_name, charset_by_id from .cursors import Cursor from .constants import FIELD_TYPE from .constants import SERVER_STATUS from .constants.CLIENT import * from .constants.COMMAND import * from .util import byte2int, int2byte from .converters import escape_item, encoders, decoders, escape_string from .err import ( raise_mysql_exception, Warning, Error, InterfaceError, DataError, DatabaseError, OperationalError, IntegrityError, InternalError, NotSupportedError, ProgrammingError) _py_version = sys.version_info[:2] # socket.makefile() in Python 2 is not usable because very inefficient and # bad behavior about timeout. # XXX: ._socketio doesn't work under IronPython. if _py_version == (2, 7) and not IRONPYTHON: # read method of file-like returned by sock.makefile() is very slow. # So we copy io-based one from Python 3. from ._socketio import SocketIO def _makefile(sock, mode): return io.BufferedReader(SocketIO(sock, mode)) elif _py_version == (2, 6): # Python 2.6 doesn't have fast io module. # So we make original one. class SockFile(object): def __init__(self, sock): self._sock = sock def read(self, n): read = self._sock.recv(n) if len(read) == n: return read while True: data = self._sock.recv(n-len(read)) if not data: return read read += data if len(read) == n: return read def _makefile(sock, mode): assert mode == 'rb' return SockFile(sock) else: # socket.makefile in Python 3 is nice. def _makefile(sock, mode): return sock.makefile(mode) TEXT_TYPES = set([ FIELD_TYPE.BIT, FIELD_TYPE.BLOB, FIELD_TYPE.LONG_BLOB, FIELD_TYPE.MEDIUM_BLOB, FIELD_TYPE.STRING, FIELD_TYPE.TINY_BLOB, FIELD_TYPE.VAR_STRING, FIELD_TYPE.VARCHAR]) sha_new = partial(hashlib.new, 'sha1') DEBUG = False NULL_COLUMN = 251 UNSIGNED_CHAR_COLUMN = 251 UNSIGNED_SHORT_COLUMN = 252 UNSIGNED_INT24_COLUMN = 253 UNSIGNED_INT64_COLUMN = 254 UNSIGNED_CHAR_LENGTH = 1 UNSIGNED_SHORT_LENGTH = 2 UNSIGNED_INT24_LENGTH = 3 UNSIGNED_INT64_LENGTH = 8 DEFAULT_CHARSET = 'latin1' MAX_PACKET_LEN = 2**24-1 def dump_packet(data): def is_ascii(data): if 65 <= byte2int(data) <= 122: #data.isalnum(): if isinstance(data, int): return chr(data) return data return '.' try: print("packet length:", len(data)) print("method call[1]:", sys._getframe(1).f_code.co_name) print("method call[2]:", sys._getframe(2).f_code.co_name) print("method call[3]:", sys._getframe(3).f_code.co_name) print("method call[4]:", sys._getframe(4).f_code.co_name) print("method call[5]:", sys._getframe(5).f_code.co_name) print("-" * 88) except ValueError: pass dump_data = [data[i:i+16] for i in range_type(0, min(len(data), 256), 16)] for d in dump_data: print(' '.join(map(lambda x:"{:02X}".format(byte2int(x)), d)) + ' ' * (16 - len(d)) + ' ' * 2 + ' '.join(map(lambda x:"{}".format(is_ascii(x)), d))) print("-" * 88) print() def _scramble(password, message): if not password: return b'\0' if DEBUG: print('password=' + password) stage1 = sha_new(password).digest() stage2 = sha_new(stage1).digest() s = sha_new() s.update(message) s.update(stage2) result = s.digest() return _my_crypt(result, stage1) def _my_crypt(message1, message2): length = len(message1) result = struct.pack('B', length) for i in range_type(length): x = (struct.unpack('B', message1[i:i+1])[0] ^ struct.unpack('B', message2[i:i+1])[0]) result += struct.pack('B', x) return result # old_passwords support ported from libmysql/password.c SCRAMBLE_LENGTH_323 = 8 class RandStruct_323(object): def __init__(self, seed1, seed2): self.max_value = 0x3FFFFFFF self.seed1 = seed1 % self.max_value self.seed2 = seed2 % self.max_value def my_rnd(self): self.seed1 = (self.seed1 * 3 + self.seed2) % self.max_value self.seed2 = (self.seed1 + self.seed2 + 33) % self.max_value return float(self.seed1) / float(self.max_value) def _scramble_323(password, message): hash_pass = _hash_password_323(password) hash_message = _hash_password_323(message[:SCRAMBLE_LENGTH_323]) hash_pass_n = struct.unpack(">LL", hash_pass) hash_message_n = struct.unpack(">LL", hash_message) rand_st = RandStruct_323(hash_pass_n[0] ^ hash_message_n[0], hash_pass_n[1] ^ hash_message_n[1]) outbuf = io.BytesIO() for _ in range_type(min(SCRAMBLE_LENGTH_323, len(message))): outbuf.write(int2byte(int(rand_st.my_rnd() * 31) + 64)) extra = int2byte(int(rand_st.my_rnd() * 31)) out = outbuf.getvalue() outbuf = io.BytesIO() for c in out: outbuf.write(int2byte(byte2int(c) ^ byte2int(extra))) return outbuf.getvalue() def _hash_password_323(password): nr = 1345345333 add = 7 nr2 = 0x12345671 for c in [byte2int(x) for x in password if x not in (' ', '\t')]: nr^= (((nr & 63)+add)*c)+ (nr << 8) & 0xFFFFFFFF nr2= (nr2 + ((nr2 << 8) ^ nr)) & 0xFFFFFFFF add= (add + c) & 0xFFFFFFFF r1 = nr & ((1 << 31) - 1) # kill sign bits r2 = nr2 & ((1 << 31) - 1) # pack return struct.pack(">LL", r1, r2) def pack_int24(n): return struct.pack(' len(self.__data): raise Exception('Invalid advance amount (%s) for cursor. ' 'Position=%s' % (length, new_position)) self._position = new_position def rewind(self, position=0): """Set the position of the data buffer cursor to 'position'.""" if position < 0 or position > len(self.__data): raise Exception("Invalid position to rewind cursor to: %s." % position) self._position = position def get_bytes(self, position, length=1): """Get 'length' bytes starting at 'position'. Position is start of payload (first four packet header bytes are not included) starting at index '0'. No error checking is done. If requesting outside end of buffer an empty string (or string shorter than 'length') may be returned! """ return self.__data[position:(position+length)] def read_length_encoded_integer(self): """Read a 'Length Coded Binary' number from the data buffer. Length coded numbers can be anywhere from 1 to 9 bytes depending on the value of the first byte. """ c = ord(self.read(1)) if c == NULL_COLUMN: return None if c < UNSIGNED_CHAR_COLUMN: return c elif c == UNSIGNED_SHORT_COLUMN: return unpack_uint16(self.read(UNSIGNED_SHORT_LENGTH)) elif c == UNSIGNED_INT24_COLUMN: return unpack_int24(self.read(UNSIGNED_INT24_LENGTH)) elif c == UNSIGNED_INT64_COLUMN: return unpack_int64(self.read(UNSIGNED_INT64_LENGTH)) def read_length_coded_string(self): """Read a 'Length Coded String' from the data buffer. A 'Length Coded String' consists first of a length coded (unsigned, positive) integer represented in 1-9 bytes followed by that many bytes of binary data. (For example "cat" would be "3cat".) """ length = self.read_length_encoded_integer() if length is None: return None return self.read(length) def is_ok_packet(self): return self.__data[0:1] == b'\0' def is_eof_packet(self): # http://dev.mysql.com/doc/internals/en/generic-response-packets.html#packet-EOF_Packet # Caution: \xFE may be LengthEncodedInteger. # If \xFE is LengthEncodedInteger header, 8bytes followed. return len(self.__data) < 9 and self.__data[0:1] == b'\xfe' def is_resultset_packet(self): field_count = ord(self.__data[0:1]) return 1 <= field_count <= 250 def is_error_packet(self): return self.__data[0:1] == b'\xff' def check_error(self): if self.is_error_packet(): self.rewind() self.advance(1) # field_count == error (we already know that) errno = unpack_uint16(self.read(2)) if DEBUG: print("errno = {}".format(errno)) raise_mysql_exception(self.__data) def dump(self): dump_packet(self.__data) class FieldDescriptorPacket(MysqlPacket): """A MysqlPacket that represents a specific column's metadata in the result. Parsing is automatically done and the results are exported via public attributes on the class such as: db, table_name, name, length, type_code. """ def __init__(self, connection): MysqlPacket.__init__(self, connection) self.__parse_field_descriptor(connection.encoding) def __parse_field_descriptor(self, encoding): """Parse the 'Field Descriptor' (Metadata) packet. This is compatible with MySQL 4.1+ (not compatible with MySQL 4.0). """ self.catalog = self.read_length_coded_string() self.db = self.read_length_coded_string() self.table_name = self.read_length_coded_string().decode(encoding) self.org_table = self.read_length_coded_string().decode(encoding) self.name = self.read_length_coded_string().decode(encoding) self.org_name = self.read_length_coded_string().decode(encoding) self.advance(1) # non-null filler self.charsetnr = struct.unpack(' 2: use_unicode = True if db is not None and database is None: database = db if compress or named_pipe: raise NotImplementedError("compress and named_pipe arguments are not supported") if ssl and ('capath' in ssl or 'cipher' in ssl): raise NotImplementedError('ssl options capath and cipher are not supported') self.ssl = False if ssl: if not SSL_ENABLED: raise NotImplementedError("ssl module not found") self.ssl = True client_flag |= SSL for k in ('key', 'cert', 'ca'): v = None if k in ssl: v = ssl[k] setattr(self, k, v) if read_default_group and not read_default_file: if sys.platform.startswith("win"): read_default_file = "c:\\my.ini" else: read_default_file = "/etc/my.cnf" if read_default_file: if not read_default_group: read_default_group = "client" cfg = configparser.RawConfigParser() cfg.read(os.path.expanduser(read_default_file)) def _config(key, default): try: return cfg.get(read_default_group, key) except Exception: return default user = _config("user", user) passwd = _config("password", passwd) host = _config("host", host) database = _config("database", database) unix_socket = _config("socket", unix_socket) port = int(_config("port", port)) charset = _config("default-character-set", charset) self.host = host self.port = port self.user = user or DEFAULT_USER self.password = passwd self.db = database self.no_delay = no_delay self.unix_socket = unix_socket if charset: self.charset = charset self.use_unicode = True else: self.charset = DEFAULT_CHARSET self.use_unicode = False if use_unicode is not None: self.use_unicode = use_unicode self.encoding = charset_by_name(self.charset).encoding client_flag |= CAPABILITIES client_flag |= MULTI_STATEMENTS if self.db: client_flag |= CONNECT_WITH_DB self.client_flag = client_flag self.cursorclass = cursorclass self.connect_timeout = connect_timeout self._result = None self._affected_rows = 0 self.host_info = "Not connected" #: specified autocommit mode. None means use server default. self.autocommit_mode = autocommit self.encoders = encoders # Need for MySQLdb compatibility. self.decoders = conv self.sql_mode = sql_mode self.init_command = init_command self._connect() def close(self): ''' Send the quit message and close the socket ''' if self.socket is None: raise Error("Already closed") send_data = struct.pack('= i + 6: lang, stat, cap_h, salt_len = struct.unpack('= i + salt_len: self.salt += data[i:i+salt_len] # salt_len includes auth_plugin_data_part_1 and filler #TODO: AUTH PLUGIN NAME may appeare here. def get_server_info(self): return self.server_version Warning = Warning Error = Error InterfaceError = InterfaceError DatabaseError = DatabaseError DataError = DataError OperationalError = OperationalError IntegrityError = IntegrityError InternalError = InternalError ProgrammingError = ProgrammingError NotSupportedError = NotSupportedError # TODO: move OK and EOF packet parsing/logic into a proper subclass # of MysqlPacket like has been done with FieldDescriptorPacket. class MySQLResult(object): def __init__(self, connection): from weakref import proxy self.connection = proxy(connection) self.affected_rows = None self.insert_id = None self.server_status = None self.warning_count = 0 self.message = None self.field_count = 0 self.description = None self.rows = None self.has_next = None self.unbuffered_active = False def __del__(self): if self.unbuffered_active: self._finish_unbuffered_query() def read(self): first_packet = self.connection._read_packet() # TODO: use classes for different packet types? if first_packet.is_ok_packet(): self._read_ok_packet(first_packet) else: self._read_result_packet(first_packet) def init_unbuffered_query(self): self.unbuffered_active = True first_packet = self.connection._read_packet() if first_packet.is_ok_packet(): self._read_ok_packet(first_packet) self.unbuffered_active = False else: self.field_count = first_packet.read_length_encoded_integer() self._get_descriptions() # Apparently, MySQLdb picks this number because it's the maximum # value of a 64bit unsigned integer. Since we're emulating MySQLdb, # we set it to this instead of None, which would be preferred. self.affected_rows = 18446744073709551615 def _read_ok_packet(self, first_packet): ok_packet = OKPacketWrapper(first_packet) self.affected_rows = ok_packet.affected_rows self.insert_id = ok_packet.insert_id self.server_status = ok_packet.server_status self.warning_count = ok_packet.warning_count self.message = ok_packet.message def _check_packet_is_eof(self, packet): if packet.is_eof_packet(): eof_packet = EOFPacketWrapper(packet) self.warning_count = eof_packet.warning_count self.has_next = eof_packet.has_next return True return False def _read_result_packet(self, first_packet): self.field_count = first_packet.read_length_encoded_integer() self._get_descriptions() self._read_rowdata_packet() def _read_rowdata_packet_unbuffered(self): # Check if in an active query if not self.unbuffered_active: return # EOF packet = self.connection._read_packet() if self._check_packet_is_eof(packet): self.unbuffered_active = False self.rows = None return row = self._read_row_from_packet(packet) self.affected_rows = 1 self.rows = (row,) # rows should tuple of row for MySQL-python compatibility. return row def _finish_unbuffered_query(self): # After much reading on the MySQL protocol, it appears that there is, # in fact, no way to stop MySQL from sending all the data after # executing a query, so we just spin, and wait for an EOF packet. while self.unbuffered_active: packet = self.connection._read_packet() if self._check_packet_is_eof(packet): self.unbuffered_active = False def _read_rowdata_packet(self): """Read a rowdata packet for each data row in the result set.""" rows = [] while True: packet = self.connection._read_packet() if self._check_packet_is_eof(packet): break rows.append(self._read_row_from_packet(packet)) self.affected_rows = len(rows) self.rows = tuple(rows) def _read_row_from_packet(self, packet): use_unicode = self.connection.use_unicode row = [] for field in self.fields: data = packet.read_length_coded_string() if data is not None: field_type = field.type_code if use_unicode: if field_type in TEXT_TYPES: charset = charset_by_id(field.charsetnr) if use_unicode and not charset.is_binary: # TEXTs with charset=binary means BINARY types. data = data.decode(charset.encoding) else: data = data.decode() converter = self.connection.decoders.get(field_type) if DEBUG: print("DEBUG: field={}, converter={}".format(field, converter)) if DEBUG: print("DEBUG: DATA = ", data) if converter is not None: data = converter(data) row.append(data) return tuple(row) def _get_descriptions(self): """Read a column descriptor packet for each column in the result.""" self.fields = [] description = [] for i in range_type(self.field_count): field = self.connection._read_packet(FieldDescriptorPacket) self.fields.append(field) description.append(field.description()) eof_packet = self.connection._read_packet() assert eof_packet.is_eof_packet(), 'Protocol error, expecting EOF' self.description = tuple(description)