← Back
Editing: authentication.py
# MySQL Connector/Python - MySQL driver written in Python. # Copyright (c) 2014, 2015, Oracle and/or its affiliates. All rights reserved. # MySQL Connector/Python is licensed under the terms of the GPLv2 # <http://www.gnu.org/licenses/old-licenses/gpl-2.0.html>, like most # MySQL Connectors. There are special exceptions to the terms and # conditions of the GPLv2 as it is applied to this software, see the # FOSS License Exception # <http://www.mysql.com/about/legal/licensing/foss-exception.html>. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation. # # 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 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 """Implementing support for MySQL Authentication Plugins""" from hashlib import sha1 import struct from . import errors from .catch23 import PY2, isstr class BaseAuthPlugin(object): """Base class for authentication plugins Classes inheriting from BaseAuthPlugin should implement the method prepare_password(). When instantiating, auth_data argument is required. The username, password and database are optional. The ssl_enabled argument can be used to tell the plugin whether SSL is active or not. The method auth_response() method is used to retrieve the password which was prepared by prepare_password(). """ requires_ssl = False plugin_name = '' def __init__(self, auth_data, username=None, password=None, database=None, ssl_enabled=False): """Initialization""" self._auth_data = auth_data self._username = username self._password = password self._database = database self._ssl_enabled = ssl_enabled def prepare_password(self): """Prepares and returns password to be send to MySQL This method needs to be implemented by classes inheriting from this class. It is used by the auth_response() method. Raises NotImplementedError. """ raise NotImplementedError def auth_response(self): """Returns the prepared password to send to MySQL Raises InterfaceError on errors. For example, when SSL is required by not enabled. Returns str """ if self.requires_ssl and not self._ssl_enabled: raise errors.InterfaceError("{name} requires SSL".format( name=self.plugin_name)) return self.prepare_password() class MySQLNativePasswordAuthPlugin(BaseAuthPlugin): """Class implementing the MySQL Native Password authentication plugin""" requires_ssl = False plugin_name = 'mysql_native_password' def prepare_password(self): """Prepares and returns password as native MySQL 4.1+ password""" if not self._auth_data: raise errors.InterfaceError("Missing authentication data (seed)") if not self._password: return b'' password = self._password if isstr(self._password): password = self._password.encode('utf-8') else: password = self._password if PY2: password = buffer(password) # pylint: disable=E0602 try: auth_data = buffer(self._auth_data) # pylint: disable=E0602 except TypeError: raise errors.InterfaceError("Authentication data incorrect") else: password = password auth_data = self._auth_data hash4 = None try: hash1 = sha1(password).digest() hash2 = sha1(hash1).digest() hash3 = sha1(auth_data + hash2).digest() if PY2: xored = [ord(h1) ^ ord(h3) for (h1, h3) in zip(hash1, hash3)] else: xored = [h1 ^ h3 for (h1, h3) in zip(hash1, hash3)] hash4 = struct.pack('20B', *xored) except Exception as exc: raise errors.InterfaceError( "Failed scrambling password; {0}".format(exc)) return hash4 class MySQLClearPasswordAuthPlugin(BaseAuthPlugin): """Class implementing the MySQL Clear Password authentication plugin""" requires_ssl = True plugin_name = 'mysql_clear_password' def prepare_password(self): """Returns password as as clear text""" if not self._password: return b'\x00' password = self._password if PY2: if isinstance(password, unicode): # pylint: disable=E0602 password = password.encode('utf8') elif isinstance(password, str): password = password.encode('utf8') return password + b'\x00' class MySQLSHA256PasswordAuthPlugin(BaseAuthPlugin): """Class implementing the MySQL SHA256 authentication plugin Note that encrypting using RSA is not supported since the Python Standard Library does not provide this OpenSSL functionality. """ requires_ssl = True plugin_name = 'sha256_password' def prepare_password(self): """Returns password as as clear text""" if not self._password: return b'\x00' password = self._password if PY2: if isinstance(password, unicode): # pylint: disable=E0602 password = password.encode('utf8') elif isinstance(password, str): password = password.encode('utf8') return password + b'\x00' def get_auth_plugin(plugin_name): """Return authentication class based on plugin name This function returns the class for the authentication plugin plugin_name. The returned class is a subclass of BaseAuthPlugin. Raises errors.NotSupportedError when plugin_name is not supported. Returns subclass of BaseAuthPlugin. """ for authclass in BaseAuthPlugin.__subclasses__(): # pylint: disable=E1101 if authclass.plugin_name == plugin_name: return authclass raise errors.NotSupportedError( "Authentication plugin '{0}' is not supported".format(plugin_name))
Save File
Cancel