diff --git a/.gitignore b/.gitignore
index cd93a4b0..0b2c85be 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,12 +1,14 @@
 *.pyc
 *.pyo
-__pycache__
-.coverage
-/dist
-/PyMySQL.egg-info
+/.cache
+/.coverage
+/.idea
 /.tox
+/.venv
+/.vscode
+/PyMySQL.egg-info
 /build
+/dist
+/docs/build
 /pymysql/tests/databases.json
-
-/.idea
-docs/build
+__pycache__
diff --git a/.travis.yml b/.travis.yml
index 2822cd05..8d960249 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -35,13 +35,14 @@ matrix:
       python: "3.4"
     - env:
         - DB=mysql:8.0
+        - TEST_AUTH=yes
       python: "3.7-dev"
 
 # different py version from 5.6 and 5.7 as cache seems to be based on py version
 # http://dev.mysql.com/downloads/mysql/5.7.html has latest development release version
 # really only need libaio1 for DB builds however libaio-dev is whitelisted for container builds and liaio1 isn't
 install:
-  - pip install -U coveralls unittest2 coverage
+  - pip install -U coveralls unittest2 coverage cryptography pytest
 
 before_script:
   - ./.travis/initializedb.sh
@@ -51,6 +52,9 @@ before_script:
 
 script:
   - coverage run ./runtests.py
+  - if [ "${TEST_AUTH}" = "yes" ];
+    then pytest -v tests;
+    fi
   - if [ ! -z "${DB}" ];
     then docker logs mysqld;
     fi
diff --git a/.travis/initializedb.sh b/.travis/initializedb.sh
index 18c00eca..d9897e49 100755
--- a/.travis/initializedb.sh
+++ b/.travis/initializedb.sh
@@ -37,6 +37,16 @@ if [ ! -z "${DB}" ]; then
         docker cp mysqld:/var/lib/mysql/server-cert.pem "${HOME}"
         docker cp mysqld:/var/lib/mysql/client-key.pem "${HOME}"
         docker cp mysqld:/var/lib/mysql/client-cert.pem "${HOME}"
+
+        # Test user for auth test
+        mysql -e '
+            CREATE USER
+                user_sha256   IDENTIFIED WITH "sha256_password" BY "pass_sha256",
+                nopass_sha256 IDENTIFIED WITH "sha256_password",
+                user_caching_sha2   IDENTIFIED WITH "caching_sha2_password" BY "pass_caching_sha2",
+                nopass_caching_sha2 IDENTIFIED WITH "caching_sha2_password"
+                PASSWORD EXPIRE NEVER;'
+        mysql -e 'GRANT RELOAD ON *.* TO user_caching_sha2;'
     else
         WITH_PLUGIN=''
     fi
diff --git a/pymysql/_auth.py b/pymysql/_auth.py
new file mode 100644
index 00000000..ddf6e4e5
--- /dev/null
+++ b/pymysql/_auth.py
@@ -0,0 +1,252 @@
+"""
+Implements auth methods
+"""
+from ._compat import text_type
+from .constants import CLIENT
+from .err import OperationalError
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization, hashes
+from cryptography.hazmat.primitives.asymmetric import padding
+
+from functools import partial
+import hashlib
+import struct
+
+
+DEBUG = True
+SCRAMBLE_LENGTH = 20
+sha1_new = partial(hashlib.new, 'sha1')
+
+
+# mysql_native_password
+# https://dev.mysql.com/doc/internals/en/secure-password-authentication.html#packet-Authentication::Native41
+
+
+def scramble_native_password(password, message):
+    """Scramble used for mysql_native_password"""
+    if not password:
+        return b''
+
+    stage1 = sha1_new(password).digest()
+    stage2 = sha1_new(stage1).digest()
+    s = sha1_new()
+    s.update(message[:SCRAMBLE_LENGTH])
+    s.update(stage2)
+    result = s.digest()
+    return _my_crypt(result, stage1)
+
+
+def _my_crypt(message1, message2):
+    length = len(message1)
+    result = b''
+    for i in range(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
+# https://dev.mysql.com/doc/internals/en/old-password-authentication.html
+
+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_old_password(password, message):
+    """Scramble for old_password"""
+    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(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
+
+    # x in py3 is numbers, p27 is chars
+    for c in [byte2int(x) for x in password if x not in (' ', '\t', 32, 9)]:
+        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)
+    return struct.pack(">LL", r1, r2)
+
+
+# sha256_password
+
+
+def _roundtrip(conn, send_data):
+    conn.write_packet(send_data)
+    pkt = conn._read_packet()
+    pkt.check_error()
+    return pkt
+
+
+def _xor_password(password, salt):
+    password_bytes = bytearray(password)
+    salt = bytearray(salt)  # for PY2 compat.
+    salt_len = len(salt)
+    for i in range(len(password_bytes)):
+        password_bytes[i] ^= salt[i % salt_len]
+    return bytes(password_bytes)
+
+
+def sha2_rsa_encrypt(password, salt, public_key):
+    """Encrypt password with salt and public_key.
+
+    Used for sha256_password and caching_sha2_password.
+    """
+    message = _xor_password(password + b'\0', salt)
+    rsa_key = serialization.load_pem_public_key(public_key, default_backend())
+    return rsa_key.encrypt(
+        message,
+        padding.OAEP(
+            mgf=padding.MGF1(algorithm=hashes.SHA1()),
+            algorithm=hashes.SHA1(),
+            label=None,
+        ),
+    )
+
+
+def sha256_password_auth(conn, pkt):
+    if conn.ssl and conn.server_capabilities & CLIENT.SSL:
+        if DEBUG:
+            print("sha256: Sending plain password")
+        data = conn.password + b'\0'
+        return _roundtrip(conn, data)
+
+    if pkt.is_auth_switch_request():
+        conn.salt = pkt.read_all()
+        if not conn.server_public_key and conn.password:
+            # Request server public key
+            if DEBUG:
+                print("sha256: Requesting server public key")
+            pkt = _roundtrip(conn, b'\1')
+
+    if pkt.is_extra_auth_data():
+        conn.server_public_key = pkt._data[1:]
+        if DEBUG:
+            print("Received public key:\n", conn.server_public_key.decode('ascii'))
+
+    if conn.password:
+        if not conn.server_public_key:
+            raise OperationalError("Couldn't receive server's public key")
+
+        data = sha2_rsa_encrypt(conn.password, conn.salt, conn.server_public_key)
+    else:
+        data = b''
+
+    return _roundtrip(conn, data)
+
+
+def scramble_caching_sha2(password, nonce):
+    # (bytes, bytes) -> bytes
+    """Scramble algorithm used in cached_sha2_password fast path.
+
+    XOR(SHA256(password), SHA256(SHA256(SHA256(password)), nonce))
+    """
+    if not password:
+        return b''
+
+    p1 = hashlib.sha256(password).digest()
+    p2 = hashlib.sha256(p1).digest()
+    p3 = hashlib.sha256(p2 + nonce).digest()
+
+    res = bytearray(p1)
+    for i in range(len(p3)):
+        res[i] ^= p3[i]
+
+    return bytes(res)
+
+
+def caching_sha2_password_auth(conn, pkt):
+    # No password fast path
+    if not conn.password:
+        return _roundtrip(conn, b'')
+
+    if pkt.is_auth_switch_request():
+        # Try from fast auth
+        if DEBUG:
+            print("caching sha2: Trying fast path")
+        conn.salt = pkt.read_all()
+        scrambled = scramble_caching_sha2(conn.password, conn.salt)
+        pkt = _roundtrip(conn, scrambled)
+    # else: fast auth is tried in initial handshake
+
+    if not pkt.is_extra_auth_data():
+        raise OperationalError(
+            "caching sha2: Unknown packet for fast auth: %s" % pkt._data[:1]
+        )
+
+    # magic numbers:
+    # 2 - request public key
+    # 3 - fast auth succeeded
+    # 4 - need full auth
+
+    pkt.advance(1)
+    n = pkt.read_uint8()
+
+    if n == 3:
+        if DEBUG:
+            print("caching sha2: succeeded by fast path.")
+        pkt = conn._read_packet()
+        pkt.check_error()  # pkt must be OK packet
+        return pkt
+
+    if n != 4:
+        raise OperationalError("caching sha2: Unknwon result for fast auth: %s" % n)
+
+    if DEBUG:
+        print("caching sha2: Trying full auth...")
+
+    if conn.ssl and conn.server_capabilities & CLIENT.SSL:
+        if DEBUG:
+            print("caching sha2: Sending plain password via SSL")
+        return _roundtrip(conn, conn.password + b'\0')
+
+    if not conn.server_public_key:
+        pkt = _roundtrip(conn, b'\x02')  # Request public key
+        if not pkt.is_extra_auth_data():
+            raise OperationalError(
+                "caching sha2: Unknown packet for public key: %s" % pkt._data[:1]
+            )
+
+        conn.server_public_key = pkt._data[1:]
+        if DEBUG:
+            print(conn.server_public_key.decode('ascii'))
+
+    data = sha2_rsa_encrypt(conn.password, conn.salt, conn.server_public_key)
+    pkt = _roundtrip(conn, data)
diff --git a/pymysql/connections.py b/pymysql/connections.py
index d5dc0aff..14ae76d9 100644
--- a/pymysql/connections.py
+++ b/pymysql/connections.py
@@ -6,8 +6,6 @@
 from ._compat import PY2, range_type, text_type, str_type, JYTHON, IRONPYTHON
 
 import errno
-from functools import partial
-import hashlib
 import io
 import os
 import socket
@@ -16,6 +14,8 @@
 import traceback
 import warnings
 
+from . import _auth
+
 from .charset import charset_by_name, charset_by_id
 from .constants import CLIENT, COMMAND, CR, FIELD_TYPE, SERVER_STATUS
 from . import converters
@@ -43,7 +43,6 @@
     # KeyError occurs when there's no entry in OS database for a current user.
     DEFAULT_USER = None
 
-
 DEBUG = False
 
 _py_version = sys.version_info[:2]
@@ -87,90 +86,16 @@ def _makefile(sock, mode):
     FIELD_TYPE.VARCHAR,
     FIELD_TYPE.GEOMETRY])
 
-sha_new = partial(hashlib.new, 'sha1')
 
-DEFAULT_CHARSET = 'latin1'
+DEFAULT_CHARSET = 'latin1'  # TODO: change to utf8mb4
 
 MAX_PACKET_LEN = 2**24-1
 
-SCRAMBLE_LENGTH = 20
-
-def _scramble(password, message):
-    if not password:
-        return b''
-    if DEBUG: print('password=' + str(password))
-    stage1 = sha_new(password).digest()
-    stage2 = sha_new(stage1).digest()
-    s = sha_new()
-    s.update(message[:SCRAMBLE_LENGTH]) 
-    s.update(stage2)
-    result = s.digest()
-    return _my_crypt(result, stage1)
-
-
-def _my_crypt(message1, message2):
-    length = len(message1)
-    result = b''
-    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
-
-    # x in py3 is numbers, p27 is chars
-    for c in [byte2int(x) for x in password if x not in (' ', '\t', 32, 9)]:
-        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)
-    return struct.pack(">LL", r1, r2)
-
 
 def pack_int24(n):
     return struct.pack('<I', n)[:3]
 
+
 # https://dev.mysql.com/doc/internals/en/integer.html#packet-Protocol::LengthEncodedInteger
 def lenenc_int(i):
     if (i < 0):
@@ -186,6 +111,7 @@ def lenenc_int(i):
     else:
         raise ValueError("Encoding %x is larger than %x - no representation in LengthEncodedInteger" % (i, (1 << 64)))
 
+
 class Connection(object):
     """
     Representation of a socket with a mysql server.
@@ -240,6 +166,7 @@ class Connection(object):
         The class needs an authenticate method taking an authentication packet as
         an argument.  For the dialog plugin, a prompt(echo, prompt) method can be used
         (if no authenticate method) for returning a string from the user. (experimental)
+    :param server_public_key: SHA256 authenticaiton plugin public key value. (default: None)
     :param db: Alias for database. (for compatibility to MySQLdb)
     :param passwd: Alias for password. (for compatibility to MySQLdb)
     :param binary_prefix: Add _binary prefix on bytes and bytearray. (default: False)
@@ -261,8 +188,9 @@ def __init__(self, host=None, user=None, password="",
                  compress=None, named_pipe=None, no_delay=None,
                  autocommit=False, db=None, passwd=None, local_infile=False,
                  max_allowed_packet=16*1024*1024, defer_connect=False,
-                 auth_plugin_map={}, read_timeout=None, write_timeout=None,
-                 bind_address=None, binary_prefix=False, program_name=None):
+                 auth_plugin_map=None, read_timeout=None, write_timeout=None,
+                 bind_address=None, binary_prefix=False, program_name=None,
+                 server_public_key=None):
         if no_delay is not None:
             warnings.warn("no_delay option is deprecated", DeprecationWarning)
 
@@ -329,7 +257,9 @@ def _config(key, arg):
         self.host = host or "localhost"
         self.port = port or 3306
         self.user = user or DEFAULT_USER
-        self.password = password or ""
+        self.password = password or b""
+        if isinstance(self.password, text_type):
+            self.password = self.password.encode('latin1')
         self.db = database
         self.unix_socket = unix_socket
         self.bind_address = bind_address
@@ -378,15 +308,15 @@ def _config(key, arg):
         self.sql_mode = sql_mode
         self.init_command = init_command
         self.max_allowed_packet = max_allowed_packet
-        self._auth_plugin_map = auth_plugin_map
+        self._auth_plugin_map = auth_plugin_map or {}
         self._binary_prefix = binary_prefix
+        self.server_public_key = server_public_key
 
         self._connect_attrs = {
             '_client_name': 'pymysql',
             '_pid': str(os.getpid()),
             '_client_version': VERSION_STRING,
         }
-
         if program_name:
             self._connect_attrs["program_name"] = program_name
         elif sys.argv:
@@ -420,7 +350,7 @@ def close(self):
 
         See `Connection.close() <https://www.python.org/dev/peps/pep-0249/#Connection.close>`_
         in the specification.
-        
+
         :raise Error: If the connection is already closed.
         """
         if self._closed:
@@ -446,7 +376,7 @@ def _force_close(self):
         if self._sock:
             try:
                 self._sock.close()
-            except:
+            except:  # noqa
                 pass
         self._sock = None
         self._rfile = None
@@ -485,7 +415,7 @@ def begin(self):
     def commit(self):
         """
         Commit changes to stable storage.
-        
+
         See `Connection.commit() <https://www.python.org/dev/peps/pep-0249/#commit>`_
         in the specification.
         """
@@ -495,7 +425,7 @@ def commit(self):
     def rollback(self):
         """
         Roll back the current transaction.
-        
+
         See `Connection.rollback() <https://www.python.org/dev/peps/pep-0249/#rollback>`_
         in the specification.
         """
@@ -512,7 +442,7 @@ def show_warnings(self):
     def select_db(self, db):
         """
         Set current db.
-        
+
         :param db: The name of the db.
         """
         self._execute_command(COMMAND.COM_INIT_DB, db)
@@ -520,7 +450,7 @@ def select_db(self, db):
 
     def escape(self, obj, mapping=None):
         """Escape whatever value you pass to it.
-        
+
         Non-standard, for internal use; do not use this in your applications.
         """
         if isinstance(obj, str_type):
@@ -534,7 +464,7 @@ def escape(self, obj, mapping=None):
 
     def literal(self, obj):
         """Alias for escape()
-        
+
         Non-standard, for internal use; do not use this in your applications.
         """
         return self.escape(obj, self.encoders)
@@ -554,7 +484,7 @@ def _quote_bytes(self, s):
     def cursor(self, cursor=None):
         """
         Create a new cursor to execute queries with.
-        
+
         :param cursor: The type of cursor to create; one of :py:class:`Cursor`,
             :py:class:`SSCursor`, :py:class:`DictCursor`, or :py:class:`SSDictCursor`.
             None means use Cursor.
@@ -602,7 +532,7 @@ def kill(self, thread_id):
     def ping(self, reconnect=True):
         """
         Check if the server is alive.
-        
+
         :param reconnect: If the connection is closed, reconnect.
         :raise Error: If the connection is closed and reconnect=False.
         """
@@ -684,7 +614,7 @@ def connect(self, sock=None):
             if sock is not None:
                 try:
                     sock.close()
-                except:
+                except:  # noqa
                     pass
 
             if isinstance(e, (OSError, IOError, socket.error)):
@@ -811,7 +741,6 @@ def _execute_command(self, command, sql):
         :raise InterfaceError: If the connection is closed.
         :raise ValueError: If no username was specified.
         """
-        
         if not self._sock:
             raise err.InterfaceError("(0, '')")
 
@@ -861,7 +790,7 @@ def _request_authentication(self):
         if isinstance(self.user, text_type):
             self.user = self.user.encode(self.encoding)
 
-        data_init = struct.pack('<iIB23s', self.client_flag, 1, charset_id, b'')
+        data_init = struct.pack('<iIB23s', self.client_flag, MAX_PACKET_LEN, charset_id, b'')
 
         if self.ssl and self.server_capabilities & CLIENT.SSL:
             self.write_packet(data_init)
@@ -872,8 +801,27 @@ def _request_authentication(self):
         data = data_init + self.user + b'\0'
 
         authresp = b''
+        plugin_name = None
+
         if self._auth_plugin_name in ('', 'mysql_native_password'):
-            authresp = _scramble(self.password.encode('latin1'), self.salt)
+            authresp = _auth.scramble_native_password(self.password, self.salt)
+        elif self._auth_plugin_name == 'caching_sha2_password':
+            plugin_name = b'caching_sha2_password'
+            if self.password:
+                if DEBUG:
+                    print("caching_sha2: trying fast path")
+                authresp = _auth.scramble_caching_sha2(self.password, self.salt)
+            else:
+                if DEBUG:
+                    print("caching_sha2: empty password")
+        elif self._auth_plugin_name == 'sha256_password':
+            plugin_name = b'sha256_password'
+            if self.ssl and self.server_capabilities & CLIENT.SSL:
+                authresp = self.password + b'\0'
+            elif self.password:
+                authresp = b'\1'  # request public key
+            else:
+                authresp = b'\0'  # empty password
 
         if self.server_capabilities & CLIENT.PLUGIN_AUTH_LENENC_CLIENT_DATA:
             data += lenenc_int(len(authresp)) + authresp
@@ -888,10 +836,7 @@ def _request_authentication(self):
             data += self.db + b'\0'
 
         if self.server_capabilities & CLIENT.PLUGIN_AUTH:
-            name = self._auth_plugin_name
-            if isinstance(name, text_type):
-                name = name.encode('ascii')
-            data += name + b'\0'
+            data += (plugin_name or b'') + b'\0'
 
         if self.server_capabilities & CLIENT.CONNECT_ATTRS:
             connect_attrs = b''
@@ -908,6 +853,7 @@ def _request_authentication(self):
         # if authentication method isn't accepted the first byte
         # will have the octet 254
         if auth_packet.is_auth_switch_request():
+            if DEBUG: print("received auth switch")
             # https://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::AuthSwitchRequest
             auth_packet.read_uint8() # 0xfe packet identifier
             plugin_name = auth_packet.read_string()
@@ -915,36 +861,42 @@ def _request_authentication(self):
                 auth_packet = self._process_auth(plugin_name, auth_packet)
             else:
                 # send legacy handshake
-                data = _scramble_323(self.password.encode('latin1'), self.salt) + b'\0'
+                data = _auth.scramble_old_password(self.password, self.salt) + b'\0'
                 self.write_packet(data)
                 auth_packet = self._read_packet()
+        elif auth_packet.is_extra_auth_data():
+            if DEBUG:
+                print("received extra data")
+            # https://dev.mysql.com/doc/internals/en/successful-authentication.html
+            if self._auth_plugin_name == "caching_sha2_password":
+                auth_packet = _auth.caching_sha2_password_auth(self, auth_packet)
+            elif self._auth_plugin_name == "sha256_password":
+                auth_packet = _auth.sha256_password_auth(self, auth_packet)
+            else:
+                raise err.OperationalError("Received extra packet for auth method %r", self._auth_plugin_name)
+
+        if DEBUG: print("Succeed to auth")
 
     def _process_auth(self, plugin_name, auth_packet):
-        plugin_class = self._auth_plugin_map.get(plugin_name)
-        if not plugin_class:
-            plugin_class = self._auth_plugin_map.get(plugin_name.decode('ascii'))
-        if plugin_class:
+        handler = self._get_auth_plugin_handler(plugin_name)
+        if handler:
             try:
-                handler = plugin_class(self)
                 return handler.authenticate(auth_packet)
             except AttributeError:
                 if plugin_name != b'dialog':
-                    raise err.OperationalError(2059, "Authentication plugin '%s'" \
-                              " not loaded: - %r missing authenticate method" % (plugin_name, plugin_class))
-            except TypeError:
-                raise err.OperationalError(2059, "Authentication plugin '%s'" \
-                    " not loaded: - %r cannot be constructed with connection object" % (plugin_name, plugin_class))
-        else:
-            handler = None
-        if plugin_name == b"mysql_native_password":
-            # https://dev.mysql.com/doc/internals/en/secure-password-authentication.html#packet-Authentication::Native41
-            data = _scramble(self.password.encode('latin1'), auth_packet.read_all())
+                    raise err.OperationalError(2059, "Authentication plugin '%s'"
+                              " not loaded: - %r missing authenticate method" % (plugin_name, type(handler)))
+        if plugin_name == b"caching_sha2_password":
+            return _auth.caching_sha2_password_auth(self, auth_packet)
+        elif plugin_name == b"sha256_password":
+            return _auth.sha256_password_auth(self, auth_packet)
+        elif plugin_name == b"mysql_native_password":
+            data = _auth.scramble_native_password(self.password, auth_packet.read_all())
         elif plugin_name == b"mysql_old_password":
-            # https://dev.mysql.com/doc/internals/en/old-password-authentication.html
-            data = _scramble_323(self.password.encode('latin1'), auth_packet.read_all()) + b'\0'
+            data = _auth.scramble_old_password(self.password, auth_packet.read_all()) + b'\0'
         elif plugin_name == b"mysql_clear_password":
             # https://dev.mysql.com/doc/internals/en/clear-text-authentication.html
-            data = self.password.encode('latin1') + b'\0'
+            data = self.password + b'\0'
         elif plugin_name == b"dialog":
             pkt = auth_packet
             while True:
@@ -954,7 +906,7 @@ def _process_auth(self, plugin_name, auth_packet):
                 prompt = pkt.read_all()
 
                 if prompt == b"Password: ":
-                    self.write_packet(self.password.encode('latin1') + b'\0')
+                    self.write_packet(self.password + b'\0')
                 elif handler:
                     resp = 'no response - TypeError within plugin.prompt method'
                     try:
@@ -981,6 +933,20 @@ def _process_auth(self, plugin_name, auth_packet):
         pkt.check_error()
         return pkt
 
+    def _get_auth_plugin_handler(self, plugin_name):
+        plugin_class = self._auth_plugin_map.get(plugin_name)
+        if not plugin_class and isinstance(plugin_name, bytes):
+            plugin_class = self._auth_plugin_map.get(plugin_name.decode('ascii'))
+        if plugin_class:
+            try:
+                handler = plugin_class(self)
+            except TypeError:
+                raise err.OperationalError(2059, "Authentication plugin '%s'"
+                    " not loaded: - %r cannot be constructed with connection object" % (plugin_name, plugin_class))
+        else:
+            handler = None
+        return handler
+
     # _mysql support
     def thread_id(self):
         return self.server_thread_id[0]
@@ -1053,9 +1019,9 @@ def _get_server_information(self):
             server_end = data.find(b'\0', i)
             if server_end < 0: # pragma: no cover - very specific upstream bug
                 # not found \0 and last field so take it all
-                self._auth_plugin_name = data[i:].decode('latin1')
+                self._auth_plugin_name = data[i:].decode('utf-8')
             else:
-                self._auth_plugin_name = data[i:server_end].decode('latin1')
+                self._auth_plugin_name = data[i:server_end].decode('utf-8')
 
     def get_server_info(self):
         return self.server_version
@@ -1254,7 +1220,7 @@ def _get_descriptions(self):
                     # This behavior is different from TEXT / BLOB.
                     # We should decode result by connection encoding regardless charsetnr.
                     # See https://github.com/PyMySQL/PyMySQL/issues/488
-                    encoding = conn_encoding  # SELECT CAST(... AS JSON) 
+                    encoding = conn_encoding  # SELECT CAST(... AS JSON)
                 elif field_type in TEXT_TYPES:
                     if field.charsetnr == 63:  # binary
                         # TEXTs with charset=binary means BINARY types.
diff --git a/pymysql/protocol.py b/pymysql/protocol.py
index e872a0eb..8ccf7c4d 100644
--- a/pymysql/protocol.py
+++ b/pymysql/protocol.py
@@ -11,6 +11,7 @@
 import struct
 import sys
 
+
 DEBUG = False
 
 NULL_COLUMN = 251
@@ -21,8 +22,8 @@
 
 
 def dump_packet(data):  # pragma: no cover
-    def is_ascii(data):
-        if 65 <= byte2int(data) <= 122:
+    def printable(data):
+        if 32 <= byte2int(data) < 127:
             if isinstance(data, int):
                 return chr(data)
             return data
@@ -30,7 +31,7 @@ def is_ascii(data):
 
     try:
         print("packet length:", len(data))
-        for i in range(1, 6):
+        for i in range(1, 7):
             f = sys._getframe(i)
             print("call[%d]: %s (line %d)" % (i, f.f_code.co_name, f.f_lineno))
         print("-" * 66)
@@ -38,9 +39,9 @@ def is_ascii(data):
         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)) +
+        print(' '.join("{:02X}".format(byte2int(x)) for x in d) +
               '   ' * (16 - len(d)) + ' ' * 2 +
-              ''.join(map(lambda x: "{}".format(is_ascii(x)), d)))
+              ''.join(printable(x) for x in d))
     print("-" * 66)
     print()
 
@@ -196,6 +197,10 @@ def is_auth_switch_request(self):
         # http://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::AuthSwitchRequest
         return self._data[0:1] == b'\xfe'
 
+    def is_extra_auth_data(self):
+        # https://dev.mysql.com/doc/internals/en/successful-authentication.html
+        return self._data[0:1] == b'\x01'
+
     def is_resultset_packet(self):
         field_count = ord(self._data[0:1])
         return 1 <= field_count <= 250
diff --git a/pymysql/tests/test_connection.py b/pymysql/tests/test_connection.py
index c626a0d3..28091be2 100644
--- a/pymysql/tests/test_connection.py
+++ b/pymysql/tests/test_connection.py
@@ -360,11 +360,12 @@ def testAuthSHA256(self):
             else:
                 c.execute('SET old_passwords = 2')
                 c.execute("SET PASSWORD FOR 'pymysql_sha256'@'localhost' = PASSWORD('Sh@256Pa33')")
+            c.execute("FLUSH PRIVILEGES")
             db = self.db.copy()
             db['password'] = "Sh@256Pa33"
-            # not implemented yet so thows error
+            # Although SHA256 is supported, need the configuration of public key of the mysql server. Currently will get error by this test. 
             with self.assertRaises(pymysql.err.OperationalError):
-                pymysql.connect(user='pymysql_256', **db)
+                pymysql.connect(user='pymysql_sha256', **db)
 
 class TestConnection(base.PyMySQLTestCase):
 
diff --git a/runtests.py b/runtests.py
index 00e492b0..ea3d9e8d 100755
--- a/runtests.py
+++ b/runtests.py
@@ -3,6 +3,10 @@
 
 from pymysql._compat import PYPY, JYTHON, IRONPYTHON
 
+#import pymysql
+#pymysql.connections.DEBUG = True
+#pymysql._auth.DEBUG = True
+
 if not (PYPY or JYTHON or IRONPYTHON):
     import atexit
     import gc
diff --git a/setup.py b/setup.py
index f9903258..0c0357cc 100755
--- a/setup.py
+++ b/setup.py
@@ -22,6 +22,9 @@
     long_description=readme,
     license="MIT",
     packages=find_packages(),
+    install_requires=[
+        "cryptography",
+    ],
     classifiers=[
         'Development Status :: 5 - Production/Stable',
         'Programming Language :: Python :: 2',
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/test_auth.py b/tests/test_auth.py
new file mode 100644
index 00000000..7d857344
--- /dev/null
+++ b/tests/test_auth.py
@@ -0,0 +1,63 @@
+"""Test for auth methods supported by MySQL 8"""
+
+import os
+import pymysql
+
+# pymysql.connections.DEBUG = True
+# pymysql._auth.DEBUG = True
+
+host = "127.0.0.1"
+port = 3306
+
+ca = os.path.expanduser("~/ca.pem")
+ssl = {'ca': ca, 'check_hostname': False}
+
+
+def test_sha256_no_password():
+    con = pymysql.connect(user="nopass_sha256", host=host, port=port, ssl=None)
+    con.close()
+
+
+def test_sha256_no_passowrd_ssl():
+    con = pymysql.connect(user="nopass_sha256", host=host, port=port, ssl=ssl)
+    con.close()
+
+
+def test_sha256_password():
+    con = pymysql.connect(user="user_sha256", password="pass_sha256", host=host, port=port, ssl=None)
+    con.close()
+
+
+def test_sha256_password_ssl():
+    con = pymysql.connect(user="user_sha256", password="pass_sha256", host=host, port=port, ssl=ssl)
+    con.close()
+
+
+def test_caching_sha2_no_password():
+    con = pymysql.connect(user="nopass_caching_sha2", host=host, port=port, ssl=None)
+    con.close()
+
+
+def test_caching_sha2_no_password():
+    con = pymysql.connect(user="nopass_caching_sha2", host=host, port=port, ssl=ssl)
+    con.close()
+
+
+def test_caching_sha2_password():
+    con = pymysql.connect(user="user_caching_sha2", password="pass_caching_sha2", host=host, port=port, ssl=None)
+    con.close()
+
+    # Fast path of caching sha2
+    con = pymysql.connect(user="user_caching_sha2", password="pass_caching_sha2", host=host, port=port, ssl=None)
+    con.query("FLUSH PRIVILEGES")
+    con.close()
+
+
+def test_caching_sha2_password_ssl():
+    con = pymysql.connect(user="user_caching_sha2", password="pass_caching_sha2", host=host, port=port, ssl=ssl)
+    con.close()
+
+    # Fast path of caching sha2
+    con = pymysql.connect(user="user_caching_sha2", password="pass_caching_sha2", host=host, port=port, ssl=None)
+    con.query("FLUSH PRIVILEGES")
+    con.close()