Connect to MySQL 8 binlog users without requiring TLS (#127)

Author: mumoshu

Cargo.lock

@@ -17,6 +17,7 @@ byteorder = "1.5"
 sha1 = "0.10"
 sha2 = "0.10"
 rsa = { version = "0.9", features = ["sha2"] }
+rand_core = { version = "0.6", features = ["getrandom"] }
 
 hex = "0.4"
 zstd = "0.13.3"

crates/binlog-protocol/Cargo.toml

@@ -9,8 +9,11 @@ use tokio_rustls::TlsConnector;
 
 use crate::error::Error;
 use crate::options::{SslMode, SslOptions};
+use rsa::pkcs8::DecodePublicKey;
 
 const UTF8_MB4_GENERAL_CI: u16 = 0x002d;
+/// MySQL auth scramble / nonce length (trailing NUL from the handshake is trimmed).
+const SCRAMBLE_LENGTH: usize = 20;
 const CLIENT_PROTOCOL_41: u32 = 512;
 const CLIENT_SSL: u32 = 2048;
 const CLIENT_SECURE_CONNECTION: u32 = 32768;
@@ -191,6 +194,7 @@ pub async fn authenticate(
 ) -> Result<(), Error> {
     negotiate_tls(channel, handshake, host, ssl).await?;
 
+    let mut scramble = handshake.scramble.clone();
     let auth = build_auth_packet(handshake, username, password, channel.tls_active())?;
     channel.write_packet(&auth).await?;
     let mut packet = channel.read_packet().await?;
@@ -216,15 +220,16 @@ pub async fn authenticate(
                         return Ok(());
                     }
                     Some(0x04) => {
-                        if !channel.tls_active() {
-                            return Err(Error::Auth(
-                                "caching_sha2_password full authentication requires TLS; use --tls-mode preferred or required"
-                                    .into(),
-                            ));
+                        if channel.tls_active() {
+                            let mut response = password.as_bytes().to_vec();
+                            response.push(0);
+                            channel.write_packet(&response).await?;
+                        } else {
+                            // Non-TLS full auth: RSA-encrypt the password with the
+                            // server's public key (caching_sha2_password).
+                            perform_caching_sha2_rsa_full_auth(channel, password, &scramble)
+                                .await?;
                         }
-                        let mut response = password.as_bytes().to_vec();
-                        response.push(0);
-                        channel.write_packet(&response).await?;
                         packet = channel.read_packet().await?;
                     }
                     other => {
@@ -243,7 +248,8 @@ pub async fn authenticate(
                     .find_map(|(i, &b)| if b == 0 { Some(i) } else { None })
                     .unwrap_or(0);
                 let plugin_data = &packet[1 + plugin.len() + 1..1 + plugin.len() + 1 + data_end];
-                let response = auth_switch_response(&plugin, password, plugin_data)?;
+                scramble = normalize_scramble(plugin_data);
+                let response = auth_switch_response(&plugin, password, &scramble)?;
                 channel.write_packet(&response).await?;
                 packet = channel.read_packet().await?;
             }
@@ -256,6 +262,50 @@ pub async fn authenticate(
     }
 }
 
+/// Request the server RSA public key and send an RSA-OAEP encrypted password
+/// for `caching_sha2_password` full authentication over a plaintext connection.
+async fn perform_caching_sha2_rsa_full_auth(
+    channel: &mut PacketChannel,
+    password: &str,
+    scramble: &[u8],
+) -> Result<(), Error> {
+    // 0x02 = PublicKeyRequest
+    channel.write_packet(&[0x02]).await?;
+    let key_pkt = channel.read_packet().await?;
+    check_error_packet(&key_pkt, "failed to get RSA public key")?;
+    if key_pkt.first() != Some(&0x01) {
+        return Err(Error::Auth(format!(
+            "unexpected RSA key response prefix: {:02x}",
+            key_pkt.first().copied().unwrap_or(0)
+        )));
+    }
+    // AuthMoreData: [0x01, PEM..., optional 0x00]
+    let pem_bytes = key_pkt[1..].strip_suffix(&[0x00]).unwrap_or(&key_pkt[1..]);
+    let pub_key_pem = std::str::from_utf8(pem_bytes)
+        .map_err(|e| Error::Auth(format!("invalid UTF-8 in RSA public key: {e}")))?;
+    let pub_key = rsa::RsaPublicKey::from_public_key_pem(pub_key_pem)
+        .map_err(|e| Error::Auth(format!("failed to parse RSA public key: {e}")))?;
+
+    // XOR null-terminated password with the 20-byte scramble (repeating).
+    let mut plain = password.as_bytes().to_vec();
+    plain.push(0);
+    if scramble.is_empty() {
+        return Err(Error::Auth(
+            "caching_sha2_password RSA full auth requires a non-empty scramble".into(),
+        ));
+    }
+    for (i, byte) in plain.iter_mut().enumerate() {
+        *byte ^= scramble[i % scramble.len()];
+    }
+
+    // MySQL uses RSAES-OAEP with SHA-1 (MySQL 8.0.5+).
+    let padding = rsa::Oaep::new::<sha1::Sha1>();
+    let encrypted = pub_key
+        .encrypt(&mut rand_core::OsRng, padding, &plain)
+        .map_err(|e| Error::Auth(format!("RSA encryption failed: {e}")))?;
+    channel.write_packet(&encrypted).await
+}
+
 async fn negotiate_tls(
     channel: &mut PacketChannel,
     handshake: &Handshake,
@@ -509,6 +559,7 @@ fn parse_handshake(packet: &[u8]) -> Result<Handshake, Error> {
 
     let mut scramble = scramble_part1.to_vec();
     scramble.extend_from_slice(scramble_part2);
+    scramble = normalize_scramble(&scramble);
 
     let auth_plugin = if capabilities & CLIENT_PLUGIN_AUTH != 0 {
         read_null_string(&packet[pos..])
@@ -524,6 +575,14 @@ fn parse_handshake(packet: &[u8]) -> Result<Handshake, Error> {
     })
 }
 
+/// Pad or truncate to the canonical 20-byte MySQL scramble (trailing handshake NUL
+/// is dropped when the concatenated length is 21, matching mysql_async).
+fn normalize_scramble(raw: &[u8]) -> Vec<u8> {
+    let mut scramble = raw.to_vec();
+    scramble.resize(SCRAMBLE_LENGTH, 0);
+    scramble
+}
+
 fn read_null_string(data: &[u8]) -> String {
     let end = data.iter().position(|&b| b == 0).unwrap_or(data.len());
     String::from_utf8_lossy(&data[..end]).into_owned()

crates/binlog-protocol/src/shared/wire.rs

@@ -657,6 +657,69 @@ async fn repl_user_can_stream_changes() -> Result<()> {
     Ok(())
 }
 
+/// MySQL 8 `caching_sha2_password` full auth over plaintext (RSA) must succeed
+/// without TLS. A freshly created user forces a cache miss (AuthMoreData 0x04).
+#[tokio::test]
+async fn caching_sha2_rsa_full_auth_without_tls() -> Result<()> {
+    crate::shared::init_logging();
+    let container = crate::shared::shared_mysql_binlog().await;
+    let db_name = "integ_sha2_rsa";
+    let admin_conn = crate::shared::create_test_db(container, db_name).await?;
+
+    let admin_pool = mysql_async::Pool::from_url(&admin_conn)?;
+    let mut admin = admin_pool.get_conn().await?;
+
+    // Unique user so the server has no cached password hash yet.
+    let user = format!("sha2_rsa_{}", std::process::id());
+    let pass = "sha2_rsa_pass";
+    admin
+        .query_drop(format!("DROP USER IF EXISTS '{user}'@'%'"))
+        .await?;
+    admin
+        .query_drop(format!(
+            "CREATE USER '{user}'@'%' IDENTIFIED WITH caching_sha2_password BY '{pass}'"
+        ))
+        .await?;
+    admin
+        .query_drop(format!(
+            "GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '{user}'@'%'"
+        ))
+        .await?;
+    admin
+        .query_drop(format!("GRANT SELECT ON `{db_name}`.* TO '{user}'@'%'"))
+        .await?;
+    admin.query_drop("FLUSH PRIVILEGES").await?;
+
+    let sha2_conn = admin_conn.replacen("root:testpass@", &format!("{user}:{pass}@"), 1);
+    let (host, port, username, password, _) = crate::shared::parse_mysql_uri(&sha2_conn)?;
+
+    // Plaintext connection: must use RSA full auth, not TLS cleartext.
+    let client = BinlogClient::connect(ReplicaOptions {
+        host,
+        port,
+        username,
+        password,
+        server_id: 9_003_020,
+        ssl: SslMode::Disabled,
+        blocking_poll: Duration::from_millis(200),
+        flavor: None,
+        mariadb_flags: binlog_protocol::MariaDbDumpFlags {
+            send_annotate_rows: true,
+        },
+        mariadb_gtid_strict_mode: binlog_protocol::MariaDbGtidStrictMode::ServerDefault,
+    })
+    .await
+    .map_err(|e| anyhow::anyhow!("caching_sha2 RSA full auth failed: {e}"))?;
+
+    drop(client);
+    admin
+        .query_drop(format!("DROP USER IF EXISTS '{user}'@'%'"))
+        .await?;
+    drop(admin);
+    admin_pool.disconnect().await?;
+    Ok(())
+}
+
 #[tokio::test]
 async fn mysql_runtime_checkpoint_is_gtid() -> Result<()> {
     crate::shared::init_logging();

crates/mysql-binlog-source/tests/suite/integration.rs

@@ -219,6 +219,22 @@ Provide `--tls-mode` to encrypt the replication connection when the server is re
 - `--tls-mode required` — require TLS; fail if unavailable.
 - `--tls-ca <PATH>` / `--tls-cert <PATH>` / `--tls-key <PATH>` — verify the server and/or present a client certificate.
 
+### Authentication plugins
+
+MySQL 8 defaults to `caching_sha2_password`. surreal-sync supports the stages of that plugin (and the older `mysql_native_password` plugin) as follows:
+
+| Path | Meaning | surreal-sync |
+|------|---------|--------------|
+| Fast auth (`0x03`) | Server already has the password hash in cache | Supported |
+| Full auth + TLS (`0x04`) | Cache miss; send password in clear over TLS | Supported |
+| Full auth + RSA (`0x04`, no TLS) | Cache miss; encrypt password with the server RSA key | Supported |
+| `mysql_native_password` | Older SHA1 scramble plugin | Supported |
+| `sha256_password` | Different legacy plugin | Not supported |
+
+With `--tls-mode disabled` (the default), a cache miss uses RSA public-key encryption so the password is not sent in cleartext. Prefer `--tls-mode required` (with a CA) on untrusted networks so the whole replication session is encrypted, not only the password exchange.
+
+You may still create a replication user with `IDENTIFIED WITH mysql_native_password` for compatibility with older clients; it is optional for surreal-sync.
+
 ### Binlog compression
 
 surreal-sync transparently decodes MySQL 8's compressed transaction payloads (`TRANSACTION_PAYLOAD`, zstd). If you use per-column compression or a compression scheme surreal-sync does not yet decode, it fails with a clear error rather than importing corrupt data; disable that feature or open an issue: