Merge branch 'master' into update-rustls

serprex · web-flow · commit 0d62e1c2fb64 · 2024-11-11T15:31:48.000Z
diff --git a/Cargo.toml b/Cargo.toml
@@ -7,7 +7,7 @@ license = "MIT/Apache-2.0"
 name = "mysql_async"
 readme = "README.md"
 repository = "https://github.com/blackbeam/mysql_async"
-version = "0.34.1"
+version = "0.34.2"
 exclude = ["test/*"]
 edition = "2021"
 categories = ["asynchronous", "database"]
@@ -20,25 +20,22 @@ futures-core = "0.3"
 futures-util = "0.3"
 futures-sink = "0.3"
 keyed_priority_queue = "0.4"
-lazy_static = "1"
 lru = "0.12.0"
-mio = { version = "0.8.0", features = ["os-poll", "net"] }
-mysql_common = { version = "0.32", default-features = false }
-once_cell = "1.7.2"
+mysql_common = { version = "0.33", default-features = false }
 pem = "3.0"
 percent-encoding = "2.1.0"
 pin-project = "1.0.2"
 rand = "0.8.5"
 serde = "1"
 serde_json = "1"
 socket2 = "0.5.2"
-thiserror = "1.0.4"
+thiserror = "2"
 tokio = { version = "1.0", features = ["io-util", "fs", "net", "time", "rt"] }
 tokio-util = { version = "0.7.2", features = ["codec", "io"] }
 tracing = { version = "0.1.37", default-features = false, features = [
     "attributes",
 ], optional = true }
-twox-hash = "1"
+twox-hash = "2"
 url = "2.1"
 
 [dependencies.tokio-rustls]
diff --git a/src/conn/mod.rs b/src/conn/mod.rs
@@ -102,6 +102,7 @@ struct ConnInner {
     status: StatusFlags,
     last_ok_packet: Option<OkPacket<'static>>,
     last_err_packet: Option<mysql_common::packets::ServerError<'static>>,
+    handshake_complete: bool,
     pool: Option<Pool>,
     pending_result: std::result::Result<Option<PendingResult>, ServerError>,
     tx_status: TxStatus,
@@ -147,6 +148,7 @@ impl ConnInner {
             status: StatusFlags::empty(),
             last_ok_packet: None,
             last_err_packet: None,
+            handshake_complete: false,
             stream: None,
             is_mariadb: false,
             version: (0, 0, 0),
@@ -581,10 +583,11 @@ impl Conn {
         );
 
         // Serialize here to satisfy borrow checker.
-        let mut buf = crate::BUFFER_POOL.get();
+        let mut buf = crate::buffer_pool().get();
         handshake_response.serialize(buf.as_mut());
 
         self.write_packet(buf).await?;
+        self.inner.handshake_complete = true;
         Ok(())
     }
 
@@ -633,7 +636,7 @@ impl Conn {
             if let Some(plugin_data) = plugin_data {
                 self.write_struct(&plugin_data.into_owned()).await?;
             } else {
-                self.write_packet(crate::BUFFER_POOL.get()).await?;
+                self.write_packet(crate::buffer_pool().get()).await?;
             }
 
             self.continue_auth().await?;
@@ -701,7 +704,7 @@ impl Conn {
                 }
                 Some(0x04) => {
                     let pass = self.inner.opts.pass().unwrap_or_default();
-                    let mut pass = crate::BUFFER_POOL.get_with(pass.as_bytes());
+                    let mut pass = crate::buffer_pool().get_with(pass.as_bytes());
                     pass.as_mut().push(0);
 
                     if self.is_secure() || self.is_socket() {
@@ -789,7 +792,19 @@ impl Conn {
         if let Ok(ok_packet) = ok_packet {
             self.handle_ok(ok_packet.into_owned());
         } else {
-            let err_packet = ParseBuf(packet).parse::<ErrPacket>(self.capabilities());
+            // If we haven't completed the handshake the server will not be aware of our
+            // capabilities and so it will behave as if we have none. In particular, the error
+            // packet will not contain a SQL State field even if our capabilities do contain the
+            // `CLIENT_PROTOCOL_41` flag. Therefore it is necessary to parse an incoming packet
+            // with no capability assumptions if we have not completed the handshake.
+            //
+            // https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_connection_phase.html
+            let capabilities = if self.inner.handshake_complete {
+                self.capabilities()
+            } else {
+                CapabilityFlags::empty()
+            };
+            let err_packet = ParseBuf(packet).parse::<ErrPacket>(capabilities);
             if let Ok(err_packet) = err_packet {
                 self.handle_err(err_packet)?;
                 return Ok(true);
@@ -838,13 +853,13 @@ impl Conn {
 
     /// Writes bytes to a server.
     pub(crate) async fn write_bytes(&mut self, bytes: &[u8]) -> Result<()> {
-        let buf = crate::BUFFER_POOL.get_with(bytes);
+        let buf = crate::buffer_pool().get_with(bytes);
         self.write_packet(buf).await
     }
 
     /// Sends a serializable structure to a server.
     pub(crate) async fn write_struct<T: MySerialize>(&mut self, x: &T) -> Result<()> {
-        let mut buf = crate::BUFFER_POOL.get();
+        let mut buf = crate::buffer_pool().get();
         x.serialize(buf.as_mut());
         self.write_packet(buf).await
     }
@@ -870,7 +885,7 @@ impl Conn {
         T: AsRef<[u8]>,
     {
         let cmd_data = cmd_data.as_ref();
-        let mut buf = crate::BUFFER_POOL.get();
+        let mut buf = crate::buffer_pool().get();
         let body = buf.as_mut();
         body.push(cmd as u8);
         body.extend_from_slice(cmd_data);
@@ -1270,10 +1285,11 @@ mod test {
     use futures_util::stream::{self, StreamExt};
     use mysql_common::constants::MAX_PAYLOAD_LEN;
     use rand::Fill;
+    use tokio::{io::AsyncWriteExt, net::TcpListener};
 
     use crate::{
         from_row, params, prelude::*, test_misc::get_opts, ChangeUserOpts, Conn, Error,
-        OptsBuilder, Pool, Value, WhiteListFsHandler,
+        OptsBuilder, Pool, ServerError, Value, WhiteListFsHandler,
     };
 
     #[tokio::test]
@@ -1400,16 +1416,18 @@ mod test {
         .filter(|variant| plugins.iter().any(|p| p == variant.0));
 
         for (plug, val, pass) in variants {
+            dbg!((plug, val, pass, conn.inner.version));
+
+            if plug == "mysql_native_password" && conn.inner.version >= (9, 0, 0) {
+                continue;
+            }
+
             let _ = conn.query_drop("DROP USER 'test_user'@'%'").await;
 
             let query = format!("CREATE USER 'test_user'@'%' IDENTIFIED WITH {}", plug);
             conn.query_drop(query).await.unwrap();
 
-            if (8, 0, 11) <= conn.inner.version && conn.inner.version <= (9, 0, 0) {
-                conn.query_drop(format!("SET PASSWORD FOR 'test_user'@'%' = '{}'", pass))
-                    .await
-                    .unwrap();
-            } else {
+            if conn.inner.version <= (8, 0, 11) {
                 conn.query_drop(format!("SET old_passwords = {}", val))
                     .await
                     .unwrap();
@@ -1419,6 +1437,10 @@ mod test {
                 ))
                 .await
                 .unwrap();
+            } else {
+                conn.query_drop(format!("SET PASSWORD FOR 'test_user'@'%' = '{}'", pass))
+                    .await
+                    .unwrap();
             };
 
             let opts = get_opts()
@@ -1548,6 +1570,10 @@ mod test {
         };
 
         for (i, plugin) in plugins.iter().enumerate() {
+            if *plugin == "mysql_native_password" && conn.server_version() >= (9, 0, 0) {
+                continue;
+            }
+
             let mut rng = rand::thread_rng();
             let mut pass = [0u8; 10];
             pass.try_fill(&mut rng).unwrap();
@@ -2189,6 +2215,45 @@ mod test {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn should_handle_initial_error_packet() {
+        let header = [
+            0x68, 0x00, 0x00, // packet_length
+            0x00, // sequence
+            0xff, // error_header
+            0x69, 0x04, // error_code
+        ];
+        let error_message = "Host '172.17.0.1' is blocked because of many connection errors; unblock with 'mysqladmin flush-hosts'";
+
+        // Create a fake MySQL server that immediately replies with an error packet.
+        let listener = TcpListener::bind("127.0.0.1:0000").await.unwrap();
+
+        let listen_addr = listener.local_addr().unwrap();
+
+        tokio::task::spawn(async move {
+            let (mut stream, _) = listener.accept().await.unwrap();
+            stream.write_all(&header).await.unwrap();
+            stream.write_all(error_message.as_bytes()).await.unwrap();
+            stream.shutdown().await.unwrap();
+        });
+
+        let opts = OptsBuilder::default()
+            .ip_or_hostname(listen_addr.ip().to_string())
+            .tcp_port(listen_addr.port());
+        let server_err = match Conn::new(opts).await {
+            Err(Error::Server(server_err)) => server_err,
+            other => panic!("expected server error but got: {:?}", other),
+        };
+        assert_eq!(
+            server_err,
+            ServerError {
+                code: 1129,
+                state: "HY000".to_owned(),
+                message: error_message.to_owned(),
+            }
+        );
+    }
+
     #[cfg(feature = "nightly")]
     mod bench {
         use crate::{conn::Conn, queryable::Queryable, test_misc::get_opts};
diff --git a/src/conn/stmt_cache.rs b/src/conn/stmt_cache.rs
@@ -7,7 +7,7 @@
 // modified, or distributed except according to those terms.
 
 use lru::LruCache;
-use twox_hash::XxHash;
+use twox_hash::XxHash64;
 
 use std::{
     borrow::Borrow,
@@ -42,7 +42,7 @@ pub struct Entry {
 pub struct StmtCache {
     cap: usize,
     cache: LruCache<u32, Entry>,
-    query_map: HashMap<QueryString, u32, BuildHasherDefault<XxHash>>,
+    query_map: HashMap<QueryString, u32, BuildHasherDefault<XxHash64>>,
 }
 
 impl StmtCache {
diff --git a/src/io/mod.rs b/src/io/mod.rs
@@ -75,7 +75,7 @@ impl Default for PacketCodec {
     fn default() -> Self {
         Self {
             inner: Default::default(),
-            decode_buf: crate::BUFFER_POOL.get(),
+            decode_buf: crate::buffer_pool().get(),
         }
     }
 }
@@ -100,7 +100,7 @@ impl Decoder for PacketCodec {
 
     fn decode(&mut self, src: &mut BytesMut) -> std::result::Result<Option<Self::Item>, IoError> {
         if self.inner.decode(src, self.decode_buf.as_mut())? {
-            let new_buf = crate::BUFFER_POOL.get();
+            let new_buf = crate::buffer_pool().get();
             Ok(Some(replace(&mut self.decode_buf, new_buf)))
         } else {
             Ok(None)
diff --git a/src/lib.rs b/src/lib.rs
@@ -453,8 +453,11 @@ mod queryable;
 
 type BoxFuture<'a, T> = futures_core::future::BoxFuture<'a, Result<T>>;
 
-static BUFFER_POOL: once_cell::sync::Lazy<Arc<crate::buffer_pool::BufferPool>> =
-    once_cell::sync::Lazy::new(Default::default);
+fn buffer_pool() -> &'static Arc<crate::buffer_pool::BufferPool> {
+    static BUFFER_POOL: std::sync::OnceLock<Arc<crate::buffer_pool::BufferPool>> =
+        std::sync::OnceLock::new();
+    BUFFER_POOL.get_or_init(Default::default)
+}
 
 #[cfg(feature = "binlog")]
 #[doc(inline)]
@@ -562,6 +565,8 @@ pub mod prelude {
     #[doc(inline)]
     pub use crate::queryable::Queryable;
     #[doc(inline)]
+    pub use mysql_common::prelude::ColumnIndex;
+    #[doc(inline)]
     pub use mysql_common::prelude::FromRow;
     #[doc(inline)]
     pub use mysql_common::prelude::{FromValue, ToValue};
@@ -608,9 +613,8 @@ pub mod prelude {
 
 #[doc(hidden)]
 pub mod test_misc {
-    use lazy_static::lazy_static;
-
     use std::env;
+    use std::sync::OnceLock;
 
     use crate::opts::{Opts, OptsBuilder, SslOpts};
 
@@ -621,26 +625,17 @@ pub mod test_misc {
         _dummy(err);
     }
 
-    lazy_static! {
-        pub static ref DATABASE_URL: String = {
+    pub fn get_opts() -> OptsBuilder {
+        static DATABASE_OPTS: OnceLock<Opts> = OnceLock::new();
+        let database_opts = DATABASE_OPTS.get_or_init(|| {
             if let Ok(url) = env::var("DATABASE_URL") {
-                let opts = Opts::from_url(&url).expect("DATABASE_URL invalid");
-                if opts
-                    .db_name()
-                    .expect("a database name is required")
-                    .is_empty()
-                {
-                    panic!("database name is empty");
-                }
-                url
+                Opts::from_url(&url).expect("DATABASE_URL invalid")
             } else {
-                "mysql://root:password@localhost:3307/mysql".into()
+                Opts::from_url("mysql://root:password@localhost:3307/mysql").unwrap()
             }
-        };
-    }
+        });
 
-    pub fn get_opts() -> OptsBuilder {
-        let mut builder = OptsBuilder::from_opts(Opts::from_url(&DATABASE_URL).unwrap());
+        let mut builder = OptsBuilder::from_opts(database_opts.clone());
         if test_ssl() {
             let ssl_opts = SslOpts::default()
                 .with_danger_skip_domain_validation(true)

Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ impl Default for PacketCodec {`
`75`	`75`	`fn default() -> Self {`
`76`	`76`	`Self {`
`77`	`77`	`inner: Default::default(),`
`78`		`- decode_buf: crate::BUFFER_POOL.get(),`
	`78`	`+ decode_buf: crate::buffer_pool().get(),`
`79`	`79`	`}`
`80`	`80`	`}`
`81`	`81`	`}`
`@@ -100,7 +100,7 @@ impl Decoder for PacketCodec {`
`100`	`100`
`101`	`101`	`fn decode(&mut self, src: &mut BytesMut) -> std::result::Result<Option<Self::Item>, IoError> {`
`102`	`102`	`if self.inner.decode(src, self.decode_buf.as_mut())? {`
`103`		`- let new_buf = crate::BUFFER_POOL.get();`
	`103`	`+ let new_buf = crate::buffer_pool().get();`
`104`	`104`	`Ok(Some(replace(&mut self.decode_buf, new_buf)))`
`105`	`105`	`} else {`
`106`	`106`	`Ok(None)`