diff --git a/db-async-common/src/main/java/com/github/jasync/sql/db/Configuration.kt b/db-async-common/src/main/java/com/github/jasync/sql/db/Configuration.kt
index 13c242cc..8bd3f6d5 100644
--- a/db-async-common/src/main/java/com/github/jasync/sql/db/Configuration.kt
+++ b/db-async-common/src/main/java/com/github/jasync/sql/db/Configuration.kt
@@ -10,6 +10,7 @@ import io.netty.channel.nio.NioEventLoopGroup
 import io.netty.util.CharsetUtil
 import mu.KotlinLogging
 import java.nio.charset.Charset
+import java.nio.file.Path
 import java.time.Duration
 import java.util.concurrent.CompletionStage
 import java.util.concurrent.Executor
@@ -43,6 +44,7 @@ private val logger = KotlinLogging.logger {}
  * @param currentSchema optional database schema - postgresql only.
  * @param socketPath path to unix domain socket file (on the local machine)
  * @param credentialsProvider a credential provider used to inject credentials on demand
+ * @param rsaPublicKey path to the RSA public key, used for password encryption over unsafe connections
  *
  */
 class Configuration @JvmOverloads constructor(
@@ -63,7 +65,8 @@ class Configuration @JvmOverloads constructor(
     val executionContext: Executor = ExecutorServiceUtils.CommonPool,
     val currentSchema: String? = null,
     val socketPath: String? = null,
-    val credentialsProvider: CredentialsProvider? = null
+    val credentialsProvider: CredentialsProvider? = null,
+    val rsaPublicKey: Path? = null,
 ) {
     init {
         if (socketPath != null && eventLoopGroup is NioEventLoopGroup) {
@@ -96,6 +99,7 @@ class Configuration @JvmOverloads constructor(
         currentSchema: String? = null,
         socketPath: String? = null,
         credentialsProvider: CredentialsProvider? = null,
+        rsaPublicKey: Path? = null,
     ): Configuration {
         return Configuration(
             username = username ?: this.username,
@@ -116,6 +120,7 @@ class Configuration @JvmOverloads constructor(
             currentSchema = currentSchema ?: this.currentSchema,
             socketPath = socketPath ?: this.socketPath,
             credentialsProvider = credentialsProvider ?: this.credentialsProvider,
+            rsaPublicKey = rsaPublicKey ?: this.rsaPublicKey,
         )
     }
 
@@ -143,6 +148,7 @@ class Configuration @JvmOverloads constructor(
         if (currentSchema != other.currentSchema) return false
         if (socketPath != other.socketPath) return false
         if (credentialsProvider != other.credentialsProvider) return false
+        if (rsaPublicKey != other.rsaPublicKey) return false
 
         return true
     }
@@ -166,11 +172,12 @@ class Configuration @JvmOverloads constructor(
         result = 31 * result + (currentSchema?.hashCode() ?: 0)
         result = 31 * result + (socketPath?.hashCode() ?: 0)
         result = 31 * result + (credentialsProvider?.hashCode() ?: 0)
+        result = 31 * result + (rsaPublicKey?.hashCode() ?: 0)
         return result
     }
 
     override fun toString(): String {
-        return "Configuration(username='$username', host='$host', port=$port, password=****, database=$database, ssl=$ssl, charset=$charset, maximumMessageSize=$maximumMessageSize, allocator=$allocator, connectionTimeout=$connectionTimeout, queryTimeout=$queryTimeout, applicationName=$applicationName, interceptors=$interceptors, eventLoopGroup=$eventLoopGroup, executionContext=$executionContext, currentSchema=$currentSchema, socketPath=$socketPath, credentialsProvider=$credentialsProvider)"
+        return "Configuration(username='$username', host='$host', port=$port, password=****, database=$database, ssl=$ssl, charset=$charset, maximumMessageSize=$maximumMessageSize, allocator=$allocator, connectionTimeout=$connectionTimeout, queryTimeout=$queryTimeout, applicationName=$applicationName, interceptors=$interceptors, eventLoopGroup=$eventLoopGroup, executionContext=$executionContext, currentSchema=$currentSchema, socketPath=$socketPath, credentialsProvider=$credentialsProvider, rsaPublicKey=$rsaPublicKey)"
     }
 }
 
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/MySQLConnection.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/MySQLConnection.kt
index 594f6b38..c1814731 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/MySQLConnection.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/MySQLConnection.kt
@@ -13,10 +13,12 @@ import com.github.jasync.sql.db.exceptions.InsufficientParametersException
 import com.github.jasync.sql.db.interceptor.PreparedStatementParams
 import com.github.jasync.sql.db.mysql.codec.MySQLConnectionHandler
 import com.github.jasync.sql.db.mysql.codec.MySQLHandlerDelegate
+import com.github.jasync.sql.db.mysql.encoder.auth.AuthenticationMethod
 import com.github.jasync.sql.db.mysql.exceptions.MySQLException
 import com.github.jasync.sql.db.mysql.message.client.AuthenticationSwitchResponse
 import com.github.jasync.sql.db.mysql.message.client.CapabilityRequestMessage
 import com.github.jasync.sql.db.mysql.message.client.HandshakeResponseMessage
+import com.github.jasync.sql.db.mysql.message.server.AuthMoreDataMessage
 import com.github.jasync.sql.db.mysql.message.server.AuthenticationSwitchRequest
 import com.github.jasync.sql.db.mysql.message.server.EOFMessage
 import com.github.jasync.sql.db.mysql.message.server.ErrorMessage
@@ -93,6 +95,8 @@ class MySQLConnection @JvmOverloads constructor(
     private var connected = false
     private var lastException: Throwable? = null
     private var serverVersion: Version? = null
+    private var authenticationMethod: String? = null
+    private var authenticationSeed: ByteArray? = null
 
     object StatusFlags {
         // https://dev.mysql.com/doc/internals/en/status-flags.html
@@ -259,6 +263,8 @@ class MySQLConnection @JvmOverloads constructor(
     override fun onHandshake(message: HandshakeMessage) {
         this.serverVersion = parseVersion(message.serverVersion)
         this.serverStatus = message.statusFlags
+        this.authenticationMethod = message.authenticationMethod
+        this.authenticationSeed = message.seed
 
         val switchToSsl = when (this.configuration.ssl.mode) {
             SSLConfiguration.Mode.Disable -> false
@@ -298,7 +304,9 @@ class MySQLConnection @JvmOverloads constructor(
             message.authenticationMethod,
             database = configuration.database,
             password = configuration.password,
-            appName = configuration.applicationName
+            appName = configuration.applicationName,
+            sslConfiguration = configuration.ssl,
+            rsaPublicKey = configuration.rsaPublicKey,
         )
 
         if (!switchToSsl) {
@@ -336,7 +344,37 @@ class MySQLConnection @JvmOverloads constructor(
     }
 
     override fun switchAuthentication(message: AuthenticationSwitchRequest) {
-        this.connectionHandler.write(AuthenticationSwitchResponse(configuration.password, message))
+        val response = AuthenticationSwitchResponse(
+            configuration.password,
+            configuration.ssl,
+            configuration.rsaPublicKey,
+            message
+        )
+
+        this.connectionHandler.write(response)
+    }
+
+    override fun onAuthMoreData(message: AuthMoreDataMessage) {
+        if (message.isSuccess()) {
+            // Do nothing. This message will be followed by an `OkMessage`.
+            return
+        }
+
+        if (authenticationMethod != AuthenticationMethod.CachingSha2) {
+            throw IllegalStateException(
+                "AuthMoreDataMessage is only supported for '${AuthenticationMethod.CachingSha2}' method"
+            )
+        }
+
+        val request = AuthenticationSwitchRequest(AuthenticationMethod.Sha256, authenticationSeed!!)
+        val response = AuthenticationSwitchResponse(
+            configuration.password,
+            configuration.ssl,
+            configuration.rsaPublicKey,
+            request
+        )
+
+        this.connectionHandler.write(response)
     }
 
     override fun sendQueryDirect(query: String): CompletableFuture<QueryResult> {
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLConnectionHandler.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLConnectionHandler.kt
index 4688f847..d8ded74d 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLConnectionHandler.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLConnectionHandler.kt
@@ -4,7 +4,6 @@ import com.github.jasync.sql.db.Configuration
 import com.github.jasync.sql.db.exceptions.DatabaseException
 import com.github.jasync.sql.db.general.MutableResultSet
 import com.github.jasync.sql.db.mysql.binary.BinaryRowDecoder
-import com.github.jasync.sql.db.mysql.encoder.auth.AuthenticationMethod
 import com.github.jasync.sql.db.mysql.message.client.AuthenticationSwitchResponse
 import com.github.jasync.sql.db.mysql.message.client.CapabilityRequestMessage
 import com.github.jasync.sql.db.mysql.message.client.CloseStatementMessage
@@ -132,18 +131,7 @@ class MySQLConnectionHandler(
                         this.handleEOF(message)
                     }
                     ServerMessage.AuthMoreData -> {
-                        val m = message as AuthMoreDataMessage
-
-                        if (!m.isSuccess()) {
-                            if (!sslEstablished) {
-                                throw IllegalStateException(
-                                    "Full authentication mode for ${AuthenticationMethod.CachingSha2} requires SSL"
-                                )
-                            }
-
-                            val request = AuthenticationSwitchRequest(AuthenticationMethod.CachingSha2, null)
-                            handlerDelegate.switchAuthentication(request)
-                        }
+                        handlerDelegate.onAuthMoreData(message as AuthMoreDataMessage)
                     }
                     ServerMessage.ColumnDefinition -> {
                         val m = message as ColumnDefinitionMessage
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLHandlerDelegate.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLHandlerDelegate.kt
index 9c9e0c33..1556fca1 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLHandlerDelegate.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLHandlerDelegate.kt
@@ -1,6 +1,7 @@
 package com.github.jasync.sql.db.mysql.codec
 
 import com.github.jasync.sql.db.ResultSet
+import com.github.jasync.sql.db.mysql.message.server.AuthMoreDataMessage
 import com.github.jasync.sql.db.mysql.message.server.AuthenticationSwitchRequest
 import com.github.jasync.sql.db.mysql.message.server.EOFMessage
 import com.github.jasync.sql.db.mysql.message.server.ErrorMessage
@@ -18,5 +19,6 @@ interface MySQLHandlerDelegate {
     fun connected(ctx: ChannelHandlerContext)
     fun onResultSet(resultSet: ResultSet, message: EOFMessage)
     fun switchAuthentication(message: AuthenticationSwitchRequest)
+    fun onAuthMoreData(message: AuthMoreDataMessage)
     fun unregistered()
 }
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/AuthenticationSwitchResponseEncoder.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/AuthenticationSwitchResponseEncoder.kt
index ead77f90..f470d99b 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/AuthenticationSwitchResponseEncoder.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/AuthenticationSwitchResponseEncoder.kt
@@ -20,8 +20,13 @@ class AuthenticationSwitchResponseEncoder(val charset: Charset) : MessageEncoder
 
         val buffer = ByteBufferUtils.packetBuffer()
 
-        val bytes =
-            authenticator.generateAuthentication(charset, switch.password, switch.request.seed)
+        val bytes = authenticator.generateAuthentication(
+            charset,
+            switch.password,
+            switch.request.seed,
+            switch.sslConfiguration,
+            switch.rsaPublicKey,
+        )
         buffer.writeBytes(bytes)
 
         return buffer
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/HandshakeResponseEncoder.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/HandshakeResponseEncoder.kt
index 554265f3..6054ef52 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/HandshakeResponseEncoder.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/HandshakeResponseEncoder.kt
@@ -31,7 +31,7 @@ class HandshakeResponseEncoder(private val charset: Charset, private val headerE
             val authenticator = this.authenticationMethods.getOrElse(
                 method
             ) { throw UnsupportedAuthenticationMethodException(method) }
-            val bytes = authenticator.generateAuthentication(charset, m.password, m.seed)
+            val bytes = authenticator.generateAuthentication(charset, m.password, m.seed, m.sslConfiguration, m.rsaPublicKey)
             buffer.writeByte(bytes.length)
             buffer.writeBytes(bytes)
         } else {
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationMethod.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationMethod.kt
index ae3eb915..3ba84cfa 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationMethod.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationMethod.kt
@@ -1,10 +1,18 @@
 package com.github.jasync.sql.db.mysql.encoder.auth
 
+import com.github.jasync.sql.db.SSLConfiguration
 import java.nio.charset.Charset
+import java.nio.file.Path
 
 interface AuthenticationMethod {
 
-    fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray?): ByteArray
+    fun generateAuthentication(
+        charset: Charset,
+        password: String?,
+        seed: ByteArray,
+        sslConfiguration: SSLConfiguration,
+        rsaPublicKey: Path?,
+    ): ByteArray
 
     companion object {
         const val CachingSha2 = "caching_sha2_password"
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationScrambler.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationScrambler.kt
index 7ff4afd9..a43132ca 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationScrambler.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationScrambler.kt
@@ -1,6 +1,5 @@
 package com.github.jasync.sql.db.mysql.encoder.auth
 
-import com.github.jasync.sql.db.util.length
 import java.nio.charset.Charset
 import java.security.MessageDigest
 import kotlin.experimental.xor
@@ -32,11 +31,8 @@ object AuthenticationScrambler {
         }
 
         val result = messageDigest.digest()
-        var counter = 0
-
-        while (counter < result.length) {
-            result[counter] = (result[counter] xor initialDigest[counter])
-            counter += 1
+        for ((index, byte) in result.withIndex()) {
+            result[index] = byte xor initialDigest[index]
         }
 
         return result
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/CachingSha2PasswordAuthentication.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/CachingSha2PasswordAuthentication.kt
index 1b7d1525..dcacae2c 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/CachingSha2PasswordAuthentication.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/CachingSha2PasswordAuthentication.kt
@@ -1,22 +1,22 @@
 package com.github.jasync.sql.db.mysql.encoder.auth
 
+import com.github.jasync.sql.db.SSLConfiguration
 import java.nio.charset.Charset
+import java.nio.file.Path
 
 object CachingSha2PasswordAuthentication : AuthenticationMethod {
 
     private val EmptyArray = ByteArray(0)
 
-    override fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray?): ByteArray {
+    override fun generateAuthentication(
+        charset: Charset,
+        password: String?,
+        seed: ByteArray,
+        sslConfiguration: SSLConfiguration,
+        rsaPublicKey: Path?,
+    ): ByteArray {
         return if (password != null) {
-            if (seed != null) {
-                // Fast authentication mode. Requires seed, but not SSL.
-                AuthenticationScrambler.scramble411("SHA-256", password, charset, seed, false)
-            } else {
-                // Full authentication mode.
-                // Since this sends the plaintext password, SSL is required.
-                // Without SSL, the server always rejects the password.
-                Sha256PasswordAuthentication.generateAuthentication(charset, password, null)
-            }
+            AuthenticationScrambler.scramble411("SHA-256", password, charset, seed, false)
         } else {
             EmptyArray
         }
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/MySQLNativePasswordAuthentication.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/MySQLNativePasswordAuthentication.kt
index d47018c5..6101c6b8 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/MySQLNativePasswordAuthentication.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/MySQLNativePasswordAuthentication.kt
@@ -1,14 +1,20 @@
 package com.github.jasync.sql.db.mysql.encoder.auth
 
+import com.github.jasync.sql.db.SSLConfiguration
 import java.nio.charset.Charset
+import java.nio.file.Path
 
 object MySQLNativePasswordAuthentication : AuthenticationMethod {
 
     private val EmptyArray = ByteArray(0)
 
-    override fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray?): ByteArray {
-        requireNotNull(seed) { "Seed should not be null" }
-
+    override fun generateAuthentication(
+        charset: Charset,
+        password: String?,
+        seed: ByteArray,
+        sslConfiguration: SSLConfiguration,
+        rsaPublicKey: Path?,
+    ): ByteArray {
         return if (password != null) {
             AuthenticationScrambler.scramble411("SHA-1", password, charset, seed, true)
         } else {
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/OldPasswordAuthentication.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/OldPasswordAuthentication.kt
index 9dafaea6..199c7e98 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/OldPasswordAuthentication.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/OldPasswordAuthentication.kt
@@ -1,7 +1,9 @@
 package com.github.jasync.sql.db.mysql.encoder.auth
 
+import com.github.jasync.sql.db.SSLConfiguration
 import com.github.jasync.sql.db.util.length
 import java.nio.charset.Charset
+import java.nio.file.Path
 import kotlin.math.floor
 
 @Suppress("RedundantExplicitType", "UNUSED_VALUE", "VARIABLE_WITH_REDUNDANT_INITIALIZER")
@@ -9,9 +11,13 @@ object OldPasswordAuthentication : AuthenticationMethod {
 
     private val EmptyArray = ByteArray(0)
 
-    override fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray?): ByteArray {
-        requireNotNull(seed) { "Seed should not be null" }
-
+    override fun generateAuthentication(
+        charset: Charset,
+        password: String?,
+        seed: ByteArray,
+        sslConfiguration: SSLConfiguration,
+        rsaPublicKey: Path?,
+    ): ByteArray {
         return when {
             !password.isNullOrEmpty() -> {
                 // The native authentication handshake will provide a 20-byte challenge.
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/README.md b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/README.md
new file mode 100644
index 00000000..3eef0b05
--- /dev/null
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/README.md
@@ -0,0 +1,60 @@
+# Authentication methods
+
+This driver implements multiple authentication methods available in MySQL/MariaDB and PostgresSQL.
+The step-by-step authentication flow and implementation details are described below.
+
+## `caching_sha2_password`
+
+This is the default authentication method since MySQL 8.0.
+Official documentation can be found [here][caching-sha2-password].
+
+The fast authentication flow (using password scrambling) is as follows:
+1. During the handshake, MySQL server sends the authentication seed (nonce).
+2. The driver scrambles the password using SHA-256 with `AuthenticationScrambler`, and sends `HandshakeResponse`.
+3. If the password entry is cached on the server, it performs fast authentication, and returns `AuthMoreData`
+   message indicating success (`data=3`). This is followed by `OkMessage` and the authentication flow completes.
+4. In case the password is not cached, the server requires us to switch to full authentication, and returns
+   `AuthMoreData` with `data=4`.
+
+If we need to perform the full authentication flow (using SHA-256 hashing), the process is as follows:
+1. If we're connected over SSL, we can send `AuthenticationSwitchResponse` with a plaintext password.
+   Note that if we try to do the same over an unsafe connection, the server always rejects the password.
+2. If we are not connected over SSL, we can use the provided `rsaPublicKey` (used by the server) to encrypt the
+   password, and send it as `AuthenticationSwitchResponse`. See `Sha256PasswordAuthentication` for
+   implementation details.
+3. If `rsaPublicKey` is not specified, the public key used to encrypt the password can be fetched from the
+   server. **This is currently not supported by the driver.**
+4. If the authentication was successful, the server caches the password entry, and returns `OkMessage`.
+   The next authentication request for the specified user can therefore be done with fast authentication.
+
+## `sha256_password`
+
+This authentication method has been deprecated in favor of `caching_sha2_password` in MySQL 8.0, and works the
+same as its full authentication flow.
+
+## `mysql_native_password`
+
+This was the default authentication method until MySQL 8.0.
+Official documentation can be found [here][mysql-native-password].
+
+The authentication flow is as follows:
+1. During the handshake, MySQL server sends the authentication seed (nonce).
+2. The driver scrambles the password using SHA-1 with `AuthenticationScrambler`, and sends `HandshakeResponse`.
+3. If the password is correct the server sends `OkMessage`, and `ErrorMessage` otherwise.
+
+## `mysql_old_password`
+
+This method was mainly used before MySQL 4.1. It was deprecated in MySQL 5.6 and removed in MySQL 5.7.
+Official documentation can be found [here][mysql-old-password].
+
+The authentication flow is as follows:
+1. During the handshake, MySQL server sends the authentication seed (nonce). This can either be 8 bytes on older
+   versions of MySQL, or 20 bytes if the server uses the `mysql_native_password` method as the default. In the
+   latter case, the driver uses the first 8 bytes of the seed.
+2. The driver hashes the password using a proprietary algorithm, and sends `HandshakeResponse`. See
+   `OldPasswordAuthentication` for implementation details.
+3. If the password is correct the server sends `OkMessage`, and `ErrorMessage` otherwise.
+
+[caching-sha2-password]: https://dev.mysql.com/doc/dev/mysql-server/8.0.32/page_caching_sha2_authentication_exchanges.html
+[mysql-native-password]: https://dev.mysql.com/doc/dev/mysql-server/8.0.32/page_protocol_connection_phase_authentication_methods_native_password_authentication.html
+[mysql-old-password]: https://dev.mysql.com/doc/dev/mysql-server/8.0.32/page_protocol_connection_phase_authentication_methods.html#page_protocol_connection_phase_authentication_methods_old_password_authentication
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/Sha256PasswordAuthentication.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/Sha256PasswordAuthentication.kt
index 5ae0edca..a9c618ca 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/Sha256PasswordAuthentication.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/Sha256PasswordAuthentication.kt
@@ -1,21 +1,74 @@
 package com.github.jasync.sql.db.mysql.encoder.auth
 
+import com.github.jasync.sql.db.SSLConfiguration
 import com.github.jasync.sql.db.util.length
+import mu.KotlinLogging
 import java.nio.charset.Charset
+import java.nio.file.Files
+import java.nio.file.Path
+import java.security.KeyFactory
+import java.security.PublicKey
+import java.security.spec.X509EncodedKeySpec
+import java.util.Base64
+import javax.crypto.Cipher
+import kotlin.experimental.xor
+
+private val logger = KotlinLogging.logger {}
 
-// TODO: Implement public key encryption.
 object Sha256PasswordAuthentication : AuthenticationMethod {
 
     private val EmptyArray = ByteArray(0)
+    private val RsaHeaderRegex = Regex("(-+BEGIN PUBLIC KEY-+|-+END PUBLIC KEY-+|\\r?\\n)")
+
+    override fun generateAuthentication(
+        charset: Charset,
+        password: String?,
+        seed: ByteArray,
+        sslConfiguration: SSLConfiguration,
+        rsaPublicKey: Path?,
+    ): ByteArray {
+        if (password == null) {
+            return EmptyArray
+        }
+
+        val bytes = password.toByteArray(charset)
+        val result = ByteArray(bytes.length + 1)
+        bytes.copyInto(result)
+
+        // We can send the plaintext password in SSL mode.
+        if (sslConfiguration.mode != SSLConfiguration.Mode.Disable) {
+            return result
+        }
+
+        // Otherwise we need to encrypt the password with an RSA public key.
+        if (rsaPublicKey == null) {
+            throw IllegalStateException(
+                "Authentication is not possible over an unsafe connection. Please use SSL or specify 'rsaPublicKey'"
+            )
+        }
+
+        for ((index, byte) in result.withIndex()) {
+            result[index] = byte xor seed[index % seed.length]
+        }
 
-    override fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray?): ByteArray {
-        return if (password != null) {
-            val bytes = password.toByteArray(charset)
-            val result = ByteArray(bytes.length + 1)
-            bytes.copyInto(result)
-            result
-        } else {
-            EmptyArray
+        val publicKey = try {
+            getPublicKey(rsaPublicKey)
+        } catch (e: Exception) {
+            logger.error(e) { "Unable to read the RSA public key at '$rsaPublicKey'" }
+            throw e
         }
+
+        val cipher = Cipher.getInstance("RSA/ECB/OAEPWithSHA-1AndMGF1Padding")
+        cipher.init(Cipher.ENCRYPT_MODE, publicKey)
+        return cipher.doFinal(result)
+    }
+
+    private fun getPublicKey(path: Path): PublicKey {
+        val data = Files.readAllBytes(path).toString(Charsets.UTF_8).replace(RsaHeaderRegex, "")
+        val bytes = Base64.getDecoder().decode(data)
+        val keySpec = X509EncodedKeySpec(bytes)
+
+        val factory = KeyFactory.getInstance("RSA")
+        return factory.generatePublic(keySpec)
     }
 }
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/client/AuthenticationSwitchResponse.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/client/AuthenticationSwitchResponse.kt
index b5554991..ffd6d222 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/client/AuthenticationSwitchResponse.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/client/AuthenticationSwitchResponse.kt
@@ -1,6 +1,12 @@
 package com.github.jasync.sql.db.mysql.message.client
 
+import com.github.jasync.sql.db.SSLConfiguration
 import com.github.jasync.sql.db.mysql.message.server.AuthenticationSwitchRequest
+import java.nio.file.Path
 
-data class AuthenticationSwitchResponse(val password: String?, val request: AuthenticationSwitchRequest) :
-    ClientMessage(AuthSwitchResponse)
+data class AuthenticationSwitchResponse(
+    val password: String?,
+    val sslConfiguration: SSLConfiguration,
+    val rsaPublicKey: Path?,
+    val request: AuthenticationSwitchRequest,
+) : ClientMessage(AuthSwitchResponse)
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/client/HandshakeResponseMessage.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/client/HandshakeResponseMessage.kt
index 99689045..291f2326 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/client/HandshakeResponseMessage.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/client/HandshakeResponseMessage.kt
@@ -1,6 +1,8 @@
 package com.github.jasync.sql.db.mysql.message.client
 
+import com.github.jasync.sql.db.SSLConfiguration
 import java.nio.charset.Charset
+import java.nio.file.Path
 
 data class HandshakeResponseMessage(
     val header: CapabilityRequestMessage,
@@ -11,5 +13,7 @@ data class HandshakeResponseMessage(
     val authenticationMethod: String,
     val password: String? = null,
     val database: String? = null,
-    val appName: String? = null
+    val appName: String? = null,
+    val sslConfiguration: SSLConfiguration,
+    val rsaPublicKey: Path? = null,
 ) : ClientMessage(ClientProtocolVersion)
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthenticationSwitchRequest.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthenticationSwitchRequest.kt
index b06843dd..33dfe837 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthenticationSwitchRequest.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthenticationSwitchRequest.kt
@@ -2,5 +2,5 @@ package com.github.jasync.sql.db.mysql.message.server
 
 data class AuthenticationSwitchRequest(
     val method: String,
-    val seed: ByteArray?,
+    val seed: ByteArray,
 ) : ServerMessage(ServerMessage.EOF)
diff --git a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/AuthenticationSpec.kt b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/AuthenticationSpec.kt
index 7103420c..4e57f8cb 100644
--- a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/AuthenticationSpec.kt
+++ b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/AuthenticationSpec.kt
@@ -7,7 +7,10 @@ import com.github.jasync.sql.db.invoke
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.junit.Test
+import org.testcontainers.containers.BindMode
 import org.testcontainers.containers.MySQLContainer
+import java.nio.file.Path
+import java.nio.file.Paths
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.TimeUnit
 
@@ -15,28 +18,45 @@ class AuthenticationSpec {
 
     @Test
     fun cachingSha2PasswordAuthentication() {
-        val container = createContainer("mysql:8.0.31")
+        val container = createContainer("mysql:8.0.32")
 
         withConnection(container, "root", "test") { connection ->
-            connection.sendQuery("CREATE USER 'user' IDENTIFIED WITH caching_sha2_password BY 'foo'").await()
-            connection.sendQuery("GRANT ALL PRIVILEGES ON *.* to 'user'").await()
+            connection.sendQuery("CREATE USER 'user1' IDENTIFIED WITH caching_sha2_password BY 'foo'").await()
+            connection.sendQuery("GRANT ALL PRIVILEGES ON *.* to 'user1'").await()
+
+            connection.sendQuery("CREATE USER 'user2' IDENTIFIED WITH caching_sha2_password BY 'bar'").await()
+            connection.sendQuery("GRANT ALL PRIVILEGES ON *.* to 'user2'").await()
         }
 
         // First connection without SSL fails because we need to perform full authentication.
         assertThatThrownBy {
-            withConnection(container, "user", "foo") { /* Empty */ }
-        }.hasCauseInstanceOf(IllegalStateException::class.java)
-            .hasRootCauseMessage("Full authentication mode for caching_sha2_password requires SSL")
+            withConnection(container, "user1", "foo") { /* Empty */ }
+        }.hasRootCauseInstanceOf(IllegalStateException::class.java)
+            .hasRootCauseMessage("Authentication is not possible over an unsafe connection. Please use SSL or specify 'rsaPublicKey'")
 
         // Perform full authentication with SSL.
-        withConnection(container, "user", "foo", SSL_MODE) { connection ->
+        withConnection(container, "user1", "foo", SSL_MODE) { connection ->
             val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
             assertThat(result.rows).hasSize(1)
             assertThat(result.rows[0]("plugin")).isEqualTo("caching_sha2_password")
         }
 
         // Perform fast authentication without SSL.
-        withConnection(container, "user", "foo") { connection ->
+        withConnection(container, "user1", "foo") { connection ->
+            val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
+            assertThat(result.rows).hasSize(1)
+            assertThat(result.rows[0]("plugin")).isEqualTo("caching_sha2_password")
+        }
+
+        // Perform full authentication without SSL, with an RSA public key.
+        withConnection(container, "user2", "bar", rsaPublicKey = PUBLIC_KEY) { connection ->
+            val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
+            assertThat(result.rows).hasSize(1)
+            assertThat(result.rows[0]("plugin")).isEqualTo("caching_sha2_password")
+        }
+
+        // Perform fast authentication without SSL.
+        withConnection(container, "user2", "bar") { connection ->
             val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
             assertThat(result.rows).hasSize(1)
             assertThat(result.rows[0]("plugin")).isEqualTo("caching_sha2_password")
@@ -67,7 +87,7 @@ class AuthenticationSpec {
 
     @Test
     fun nativePasswordAuthentication() {
-        val container = createContainer("mysql:8.0.31")
+        val container = createContainer("mysql:8.0.32")
 
         withConnection(container, "root", "test") { connection ->
             connection.sendQuery("CREATE USER 'user' IDENTIFIED WITH mysql_native_password BY 'foo'").await()
@@ -92,12 +112,26 @@ class AuthenticationSpec {
             connection.sendQuery("GRANT ALL PRIVILEGES ON *.* to 'user'").await()
         }
 
+        // We can send the plaintext password over SSL.
         withConnection(container, "user", "foo", SSL_MODE) { connection ->
             val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
             assertThat(result.rows).hasSize(1)
             assertThat(result.rows[0]("plugin")).isEqualTo("sha256_password")
         }
 
+        // Prevent authentication over unsafe connections.
+        assertThatThrownBy {
+            withConnection(container, "user", "foo") { /* Empty */ }
+        }.hasRootCauseInstanceOf(IllegalStateException::class.java)
+            .hasRootCauseMessage("Authentication is not possible over an unsafe connection. Please use SSL or specify 'rsaPublicKey'")
+
+        // But allow password encryption using an RSA public key.
+        withConnection(container, "user", "foo", rsaPublicKey = PUBLIC_KEY) { connection ->
+            val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
+            assertThat(result.rows).hasSize(1)
+            assertThat(result.rows[0]("plugin")).isEqualTo("sha256_password")
+        }
+
         container.stop()
     }
 
@@ -110,6 +144,10 @@ class AuthenticationSpec {
             container.withCommand(command)
         }
 
+        for (file in ContainerHelper.configurationFiles) {
+            container.withClasspathResourceMapping(file, "/docker-entrypoint-initdb.d/$file", BindMode.READ_ONLY)
+        }
+
         container.start()
         return container
     }
@@ -119,6 +157,7 @@ class AuthenticationSpec {
         username: String,
         password: String,
         sslConfiguration: SSLConfiguration = SSLConfiguration(Mode.Disable),
+        rsaPublicKey: Path? = null,
         fn: (MySQLConnection) -> T,
     ): T {
         val configuration = Configuration(
@@ -126,6 +165,7 @@ class AuthenticationSpec {
             password = password,
             port = container.firstMappedPort,
             ssl = sslConfiguration,
+            rsaPublicKey = rsaPublicKey,
         )
 
         val connection = MySQLConnection(configuration)
@@ -144,5 +184,6 @@ class AuthenticationSpec {
     private companion object {
         const val QUERY_CURRENT_PLUGIN = "SELECT plugin FROM mysql.user WHERE CURRENT_USER() = CONCAT(user, '@', host);"
         val SSL_MODE = SSLConfiguration(Mode.Prefer)
+        val PUBLIC_KEY: Path = Paths.get(AuthenticationSpec::class.java.getResource("/public_key.pem")!!.toURI())
     }
 }
diff --git a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/ContainerHelper.java b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/ContainerHelper.java
index 0a761f7d..cc1c9708 100644
--- a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/ContainerHelper.java
+++ b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/ContainerHelper.java
@@ -18,6 +18,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -59,6 +60,14 @@ public static Integer getPort() {
         "mysql_async_tests",
         sslConfiguration);
 
+    public static List<String> configurationFiles = Arrays.asList(
+        "ca.pem",
+        "server-key.pem",
+        "server-cert.pem",
+        "private_key.pem",
+        "public_key.pem",
+        "update-config.sh");
+
     private static boolean isLocalMySQLRunning() {
         try {
             Connection connection = new MySQLConnection(rootConfiguration);
@@ -90,7 +99,7 @@ private static boolean isLocalMySQLRunning() {
 
     private static void startMySQLDocker() throws IOException {
         if (mysql == null) {
-            mysql = new MySQLContainer("mysql:8.0.31") {
+            mysql = new MySQLContainer("mysql:8.0.32") {
                 @Override
                 protected void configure() {
                     super.configure();
@@ -105,9 +114,10 @@ protected void configure() {
              .withPassword("root")
              .withDatabaseName("mysql_async_tests");
 
-            for (String file : Arrays.asList("ca.pem", "server-key.pem", "server-cert.pem", "update-config.sh")) {
+            for (String file : configurationFiles) {
                 mysql.withClasspathResourceMapping(file, "/docker-entrypoint-initdb.d/" + file, BindMode.READ_ONLY);
             }
+
             // expose unix domain socket
             Path domainSocketDirectoryPath = Files.createTempDirectory("mysqld");
             File domainSocketDirectory = domainSocketDirectoryPath.toFile();
diff --git a/mysql-async/src/test/resources/readme.md b/mysql-async/src/test/resources/README.md
similarity index 91%
rename from mysql-async/src/test/resources/readme.md
rename to mysql-async/src/test/resources/README.md
index b41bcc25..1815e29f 100644
--- a/mysql-async/src/test/resources/readme.md
+++ b/mysql-async/src/test/resources/README.md
@@ -20,4 +20,7 @@ openssl pkcs8 -topk8 -nocrypt -in client-key.pem -out client-key2.pem
 mv client-key2.pem client-key.pem
 rm *-req.pem
 rm *.ext
+# Generate the private/public RSA key pair.
+openssl genrsa -out private_key.pem 2048
+openssl rsa -in private_key.pem -pubout -out public_key.pem
 ```
diff --git a/mysql-async/src/test/resources/private_key.pem b/mysql-async/src/test/resources/private_key.pem
new file mode 100644
index 00000000..517dafac
--- /dev/null
+++ b/mysql-async/src/test/resources/private_key.pem
@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEAruu0RLfaLertaDzR5KOCSrQqGTYAC/BJUCfhLmGUdcDmawYV
+gtHcjCm7giySmVu3cXVs5ThULlxBrmihgZ5k+DQdOeH6s8HpbEHoE/1aK0ijZ2kF
+dv42SK8lp7Uh3gptECjC7HRjk3FMMmWZylZ5s3nkSUPhYUSNfOwJRO1uzGTpaYYC
+oZBFFWf5Kud7eZQeZiNvnQkDvOXyZnUTyvrUeojo2Kr0ARjgVYOp17lNvAJIar/m
+EmEdyQ+jUWgFrubCANclNuhEUIyR99PABESD9rg8zjFU6EcRvP62T4e5zMxMLTw4
+ZDSrAj9Zx14mfIcko1HXUnKTpmzMI1wUGFlEoQIDAQABAoIBACa/xEFs0QJXBpyO
+zJhvuE9ANgs7sSrz4enFy9Zwe2jSgLi4sk82mjrai9U8doVOkgaqUqM8pTlX6pzu
+RMjoA8oQEWFfNCBaFpGK0BSNjUoFX11rIHDJw9C6LAdI0uXTPXHU9clXxy9Ea/LG
+Gxl9qpwdpnqsdOMNQqKnzy3bwy231x6qFrEqx9TgG0JYgUncb/PKvYXxrOsJqTTM
+clpnZec9cWjZi0J9hvMyfO9g9oPcZzTKSrVoHXIx5KsUCwr2UXSEkgNUEb8iG8fs
+YRDnjRQ0yUlNQrlUy6YHi0C+TN42uK57qBJiyut1ZoyxtT7jPneUu5pShTkrPDoC
+sxtf1zkCgYEA4ikO9hh70qegwTde177D6KR5oYa0mVL1RxJkS8cYPXJ6zxR/uvDe
+FJB+a8l4Md4bxtgbk/0bCMgrDvu4l+VcdSrE9V7Ny8+KtHKJZNF7mWrh3v2Dp1sZ
+wnKjgkDWIxxvo9i6QwoK6OLUev5ogy76WDtapneNW1kJX+taucGrlxMCgYEAxf/w
+8QppYvPu7SSX/KpiSZJ/66Dn/bWut2qtc0o2BnZqasS3XsxK2iVF2TIWQwiOiFKt
+W5M7PXvqENgEmHEsacZ7ltwNQxbSfHdtiOqeWYjOi9Yilg2yY3LJNmFCt6dyXa9Q
+Jvi005xKbMQ6sN672v3u4d/mrILYKmy2ZEhO5/sCgYBmay+iVR4mHNGZDqk9zN9N
+iMaoVqeM09vODs8q3gPlN+XTx7W8g/4ek/0cdQWgl7Q+jXmXwESw6m1NgMNszmrb
+iL3fXMqX5Ooso74C0TdHF/coE+i+Lmxw+ZeAkjondmY8bhaT64VbR/XvqSzNU8X9
+lQuNXZC/cIflT2ErxEAfzwKBgHqEsinlIGsruFejgizFncYa0e8TBRq4FqHGcAfc
+DuwW0Ci7CyCs11B/KeaJGL9oBxKR3lXBGDImgCLmGInf1fSp7gXeqpIuCBceq3fC
+fjO29OAQpBwn44+oEpwEuQz9n0YbWkSTNwmON3twPUT+vk13Ph8ktg9fc7VkycPp
+/nGLAoGBAJLbGpD8jQkKhRj9pIi8YeX9Acxi5oAHjgV1LdxQ9zbplBQwx60jZmrw
+UHyxrWkpBVdRc8A5xpps7LFNBLXbwXwaPYXhu+OkOidXz648ODkpi28ZUfEd1GBo
+kCKhTgPGD742XfMsdEraWYCHsI6TdbUUTq1rdGnjKfNq92jY0i5h
+-----END RSA PRIVATE KEY-----
diff --git a/mysql-async/src/test/resources/public_key.pem b/mysql-async/src/test/resources/public_key.pem
new file mode 100644
index 00000000..f0d2d6ed
--- /dev/null
+++ b/mysql-async/src/test/resources/public_key.pem
@@ -0,0 +1,9 @@
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAruu0RLfaLertaDzR5KOC
+SrQqGTYAC/BJUCfhLmGUdcDmawYVgtHcjCm7giySmVu3cXVs5ThULlxBrmihgZ5k
++DQdOeH6s8HpbEHoE/1aK0ijZ2kFdv42SK8lp7Uh3gptECjC7HRjk3FMMmWZylZ5
+s3nkSUPhYUSNfOwJRO1uzGTpaYYCoZBFFWf5Kud7eZQeZiNvnQkDvOXyZnUTyvrU
+eojo2Kr0ARjgVYOp17lNvAJIar/mEmEdyQ+jUWgFrubCANclNuhEUIyR99PABESD
+9rg8zjFU6EcRvP62T4e5zMxMLTw4ZDSrAj9Zx14mfIcko1HXUnKTpmzMI1wUGFlE
+oQIDAQAB
+-----END PUBLIC KEY-----
diff --git a/mysql-async/src/test/resources/update-config.sh b/mysql-async/src/test/resources/update-config.sh
index d56e28ec..325c0879 100644
--- a/mysql-async/src/test/resources/update-config.sh
+++ b/mysql-async/src/test/resources/update-config.sh
@@ -2,6 +2,8 @@
 cp -f /docker-entrypoint-initdb.d/ca.pem /var/lib/mysql/ca.pem
 cp -f /docker-entrypoint-initdb.d/server-cert.pem /var/lib/mysql/server-cert.pem
 cp -f /docker-entrypoint-initdb.d/server-key.pem /var/lib/mysql/server-key.pem
+cp -f /docker-entrypoint-initdb.d/private_key.pem /var/lib/mysql/private_key.pem
+cp -f /docker-entrypoint-initdb.d/public_key.pem /var/lib/mysql/public_key.pem
 cat /etc/my.cnf || true
 cat /etc/mysql/my.cnf || true
 ls -lah /var/lib/mysql
diff --git a/r2dbc-mysql/src/test/java/com/github/jasync/r2dbc/mysql/integ/R2dbcContainerHelper.java b/r2dbc-mysql/src/test/java/com/github/jasync/r2dbc/mysql/integ/R2dbcContainerHelper.java
index 1ca5dadf..edf0caba 100644
--- a/r2dbc-mysql/src/test/java/com/github/jasync/r2dbc/mysql/integ/R2dbcContainerHelper.java
+++ b/r2dbc-mysql/src/test/java/com/github/jasync/r2dbc/mysql/integ/R2dbcContainerHelper.java
@@ -13,6 +13,7 @@
 import org.testcontainers.containers.MySQLContainer;
 
 import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -52,6 +53,14 @@ public static Integer getPort() {
         "mysql_async_tests",
         sslConfiguration);
 
+    public static List<String> configurationFiles = Arrays.asList(
+        "ca.pem",
+        "server-key.pem",
+        "server-cert.pem",
+        "private_key.pem",
+        "public_key.pem",
+        "update-config.sh");
+
     private static boolean isLocalMySQLRunning() {
         try {
             new MySQLConnection(rootConfiguration)
@@ -67,7 +76,7 @@ private static boolean isLocalMySQLRunning() {
 
     private static void startMySQLDocker() {
         if (mysql == null) {
-            mysql = new MySQLContainer("mysql:8.0.31") {
+            mysql = new MySQLContainer("mysql:8.0.32") {
                 @Override
                 protected void configure() {
                     super.configure();
@@ -82,7 +91,7 @@ protected void configure() {
              .withPassword("root")
              .withDatabaseName("mysql_async_tests");
 
-            for (String file : Arrays.asList("ca.pem", "server-key.pem", "server-cert.pem", "update-config.sh")) {
+            for (String file : configurationFiles) {
                 mysql.withClasspathResourceMapping(file, "/docker-entrypoint-initdb.d/" + file, BindMode.READ_ONLY);
             }
         }
diff --git a/r2dbc-mysql/src/test/resources/readme.md b/r2dbc-mysql/src/test/resources/README.md
similarity index 91%
rename from r2dbc-mysql/src/test/resources/readme.md
rename to r2dbc-mysql/src/test/resources/README.md
index b41bcc25..1815e29f 100644
--- a/r2dbc-mysql/src/test/resources/readme.md
+++ b/r2dbc-mysql/src/test/resources/README.md
@@ -20,4 +20,7 @@ openssl pkcs8 -topk8 -nocrypt -in client-key.pem -out client-key2.pem
 mv client-key2.pem client-key.pem
 rm *-req.pem
 rm *.ext
+# Generate the private/public RSA key pair.
+openssl genrsa -out private_key.pem 2048
+openssl rsa -in private_key.pem -pubout -out public_key.pem
 ```
diff --git a/r2dbc-mysql/src/test/resources/private_key.pem b/r2dbc-mysql/src/test/resources/private_key.pem
new file mode 100644
index 00000000..517dafac
--- /dev/null
+++ b/r2dbc-mysql/src/test/resources/private_key.pem
@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEAruu0RLfaLertaDzR5KOCSrQqGTYAC/BJUCfhLmGUdcDmawYV
+gtHcjCm7giySmVu3cXVs5ThULlxBrmihgZ5k+DQdOeH6s8HpbEHoE/1aK0ijZ2kF
+dv42SK8lp7Uh3gptECjC7HRjk3FMMmWZylZ5s3nkSUPhYUSNfOwJRO1uzGTpaYYC
+oZBFFWf5Kud7eZQeZiNvnQkDvOXyZnUTyvrUeojo2Kr0ARjgVYOp17lNvAJIar/m
+EmEdyQ+jUWgFrubCANclNuhEUIyR99PABESD9rg8zjFU6EcRvP62T4e5zMxMLTw4
+ZDSrAj9Zx14mfIcko1HXUnKTpmzMI1wUGFlEoQIDAQABAoIBACa/xEFs0QJXBpyO
+zJhvuE9ANgs7sSrz4enFy9Zwe2jSgLi4sk82mjrai9U8doVOkgaqUqM8pTlX6pzu
+RMjoA8oQEWFfNCBaFpGK0BSNjUoFX11rIHDJw9C6LAdI0uXTPXHU9clXxy9Ea/LG
+Gxl9qpwdpnqsdOMNQqKnzy3bwy231x6qFrEqx9TgG0JYgUncb/PKvYXxrOsJqTTM
+clpnZec9cWjZi0J9hvMyfO9g9oPcZzTKSrVoHXIx5KsUCwr2UXSEkgNUEb8iG8fs
+YRDnjRQ0yUlNQrlUy6YHi0C+TN42uK57qBJiyut1ZoyxtT7jPneUu5pShTkrPDoC
+sxtf1zkCgYEA4ikO9hh70qegwTde177D6KR5oYa0mVL1RxJkS8cYPXJ6zxR/uvDe
+FJB+a8l4Md4bxtgbk/0bCMgrDvu4l+VcdSrE9V7Ny8+KtHKJZNF7mWrh3v2Dp1sZ
+wnKjgkDWIxxvo9i6QwoK6OLUev5ogy76WDtapneNW1kJX+taucGrlxMCgYEAxf/w
+8QppYvPu7SSX/KpiSZJ/66Dn/bWut2qtc0o2BnZqasS3XsxK2iVF2TIWQwiOiFKt
+W5M7PXvqENgEmHEsacZ7ltwNQxbSfHdtiOqeWYjOi9Yilg2yY3LJNmFCt6dyXa9Q
+Jvi005xKbMQ6sN672v3u4d/mrILYKmy2ZEhO5/sCgYBmay+iVR4mHNGZDqk9zN9N
+iMaoVqeM09vODs8q3gPlN+XTx7W8g/4ek/0cdQWgl7Q+jXmXwESw6m1NgMNszmrb
+iL3fXMqX5Ooso74C0TdHF/coE+i+Lmxw+ZeAkjondmY8bhaT64VbR/XvqSzNU8X9
+lQuNXZC/cIflT2ErxEAfzwKBgHqEsinlIGsruFejgizFncYa0e8TBRq4FqHGcAfc
+DuwW0Ci7CyCs11B/KeaJGL9oBxKR3lXBGDImgCLmGInf1fSp7gXeqpIuCBceq3fC
+fjO29OAQpBwn44+oEpwEuQz9n0YbWkSTNwmON3twPUT+vk13Ph8ktg9fc7VkycPp
+/nGLAoGBAJLbGpD8jQkKhRj9pIi8YeX9Acxi5oAHjgV1LdxQ9zbplBQwx60jZmrw
+UHyxrWkpBVdRc8A5xpps7LFNBLXbwXwaPYXhu+OkOidXz648ODkpi28ZUfEd1GBo
+kCKhTgPGD742XfMsdEraWYCHsI6TdbUUTq1rdGnjKfNq92jY0i5h
+-----END RSA PRIVATE KEY-----
diff --git a/r2dbc-mysql/src/test/resources/public_key.pem b/r2dbc-mysql/src/test/resources/public_key.pem
new file mode 100644
index 00000000..f0d2d6ed
--- /dev/null
+++ b/r2dbc-mysql/src/test/resources/public_key.pem
@@ -0,0 +1,9 @@
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAruu0RLfaLertaDzR5KOC
+SrQqGTYAC/BJUCfhLmGUdcDmawYVgtHcjCm7giySmVu3cXVs5ThULlxBrmihgZ5k
++DQdOeH6s8HpbEHoE/1aK0ijZ2kFdv42SK8lp7Uh3gptECjC7HRjk3FMMmWZylZ5
+s3nkSUPhYUSNfOwJRO1uzGTpaYYCoZBFFWf5Kud7eZQeZiNvnQkDvOXyZnUTyvrU
+eojo2Kr0ARjgVYOp17lNvAJIar/mEmEdyQ+jUWgFrubCANclNuhEUIyR99PABESD
+9rg8zjFU6EcRvP62T4e5zMxMLTw4ZDSrAj9Zx14mfIcko1HXUnKTpmzMI1wUGFlE
+oQIDAQAB
+-----END PUBLIC KEY-----
diff --git a/r2dbc-mysql/src/test/resources/update-config.sh b/r2dbc-mysql/src/test/resources/update-config.sh
index d56e28ec..325c0879 100644
--- a/r2dbc-mysql/src/test/resources/update-config.sh
+++ b/r2dbc-mysql/src/test/resources/update-config.sh
@@ -2,6 +2,8 @@
 cp -f /docker-entrypoint-initdb.d/ca.pem /var/lib/mysql/ca.pem
 cp -f /docker-entrypoint-initdb.d/server-cert.pem /var/lib/mysql/server-cert.pem
 cp -f /docker-entrypoint-initdb.d/server-key.pem /var/lib/mysql/server-key.pem
+cp -f /docker-entrypoint-initdb.d/private_key.pem /var/lib/mysql/private_key.pem
+cp -f /docker-entrypoint-initdb.d/public_key.pem /var/lib/mysql/public_key.pem
 cat /etc/my.cnf || true
 cat /etc/mysql/my.cnf || true
 ls -lah /var/lib/mysql