@@ -42,6 +42,13 @@ impl AuthPlugin {
     ) -> Result<bool, Error> {
         match self {
             AuthPlugin::CachingSha2Password if packet[0] == 0x01 => {
+                if packet.len() < 2 {
+                    return Err(err_protocol!(
+                        "expected at least 2 bytes in caching_sha2_password authentication packet but got {}",
+                        packet.len()
+                    ));
+                }
+
                 match packet[1] {
                     // fast_auth_success — the server still sends a trailing
                     // OK_Packet, so yield back to the handshake loop and let
 
@@ -435,6 +435,15 @@ async fn recv_result_metadata(
     // the result-set metadata is primarily a listing of each output
     // column in the result-set
 
+    // MySQL (and MariaDB) enforce a hard limit of 4096 columns per result set;
+    // reject larger claimed counts before they are used to size allocations,
+    // as a hostile server could otherwise force an arbitrarily large allocation
+    if num_columns > 4096 {
+        return Err(err_protocol!(
+            "result-set metadata claims {num_columns} columns but the maximum supported is 4096"
+        ));
+    }
+
     let mut column_names = HashMap::with_capacity(num_columns);
 
     columns.clear();
 
@@ -11,6 +11,13 @@ use crate::protocol::response::{EofPacket, ErrPacket, OkPacket, Status};
 use crate::protocol::{Capabilities, Packet};
 use crate::{MySqlConnectOptions, MySqlDatabaseError};
 
+/// The maximum total payload that we accept for a single (possibly reassembled)
+/// packet: 1 GiB, MySQL's own hard maximum for `max_allowed_packet`.
+///
+/// Without this cap, a hostile server could stream an unbounded number of
+/// 16 MiB packet parts and exhaust the client's memory.
+const MAX_PACKET_PAYLOAD_LEN: usize = 1024 * 1024 * 1024;
+
 pub struct MySqlStream<S = Box<dyn Socket>> {
     // Wrapping the socket in `Box` allows us to unsize in-place.
     pub(crate) socket: BufferedSocket<S>,
@@ -155,6 +162,14 @@ impl<S: Socket> MySqlStream<S> {
             while last_read == 0xFF_FF_FF {
                 let part = self.recv_packet_part().await?;
                 last_read = part.len();
+
+                if final_payload.len() + part.len() > MAX_PACKET_PAYLOAD_LEN {
+                    return Err(err_protocol!(
+                        "reassembled packet payload of {} bytes exceeds maximum packet payload of {MAX_PACKET_PAYLOAD_LEN} bytes",
+                        final_payload.len() + part.len()
+                    ));
+                }
+
                 final_payload.extend_from_slice(&part);
             }
             final_payload.into()
 
@@ -62,6 +62,13 @@ impl MySqlBufExt for Bytes {
         let size = usize::try_from(size)
             .map_err(|_| err_protocol!("string length overflows usize: {size}"))?;
 
+        if size > self.remaining() {
+            return Err(err_protocol!(
+                "length-encoded string declares {size} bytes but only {} bytes remain",
+                self.remaining()
+            ));
+        }
+
         self.get_str(size)
     }
 
@@ -70,6 +77,54 @@ impl MySqlBufExt for Bytes {
         let size = usize::try_from(size)
             .map_err(|_| err_protocol!("string length overflows usize: {size}"))?;
 
+        if size > self.remaining() {
+            return Err(err_protocol!(
+                "length-encoded byte string declares {size} bytes but only {} bytes remain",
+                self.remaining()
+            ));
+        }
+
         Ok(self.split_to(size))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn get_str_lenenc_reads_within_bounds() {
+        let mut buf = Bytes::from_static(&[3, b'a', b'b', b'c']);
+
+        assert_eq!(buf.get_str_lenenc().unwrap(), "abc");
+        assert!(buf.is_empty());
+    }
+
+    #[test]
+    fn get_str_lenenc_rejects_oversized_length() {
+        // 0xfc prefixed length of 0x0100 (256) but only 3 bytes of data
+        let mut buf = Bytes::from_static(&[0xfc, 0x00, 0x01, b'a', b'b', b'c']);
+
+        let err = buf.get_str_lenenc().unwrap_err();
+
+        assert!(err.to_string().contains("256 bytes"), "{err}");
+    }
+
+    #[test]
+    fn get_bytes_lenenc_reads_within_bounds() {
+        let mut buf = Bytes::from_static(&[3, b'a', b'b', b'c']);
+
+        assert_eq!(&*buf.get_bytes_lenenc().unwrap(), b"abc");
+        assert!(buf.is_empty());
+    }
+
+    #[test]
+    fn get_bytes_lenenc_rejects_oversized_length() {
+        // 0xfc prefixed length of 0xffff (65535) but only 1 byte of data
+        let mut buf = Bytes::from_static(&[0xfc, 0xff, 0xff, b'a']);
+
+        let err = buf.get_bytes_lenenc().unwrap_err();
+
+        assert!(err.to_string().contains("65535 bytes"), "{err}");
+    }
+}
@@ -29,29 +29,74 @@ pub(crate) struct Handshake {
 
 impl ProtocolDecode<'_> for Handshake {
     fn decode_with(mut buf: Bytes, _: ()) -> Result<Self, Error> {
+        // NOTE: the handshake packet is received before authentication, so a
+        // hostile server can send anything here; every fixed-width read is
+        // validated against the remaining buffer to avoid panicking on
+        // truncated packets
+        if buf.is_empty() {
+            return Err(err_protocol!(
+                "expected protocol version (1 byte) in Handshake packet but got 0 bytes"
+            ));
+        }
+
         let protocol_version = buf.get_u8(); // int<1>
         let server_version = buf.get_str_nul()?; // string<NUL>
+
+        if buf.remaining() < 13 {
+            return Err(err_protocol!(
+                "expected connection id (4 bytes), auth-plugin-data-part-1 (8 bytes) and reserved byte (13 bytes total) in Handshake packet but got {} bytes",
+                buf.remaining()
+            ));
+        }
+
         let connection_id = buf.get_u32_le(); // int<4>
         let auth_plugin_data_1 = buf.get_bytes(8); // string<8>
 
         buf.advance(1); // reserved: string<1>
 
+        if buf.remaining() < 2 {
+            return Err(err_protocol!(
+                "expected capability flags part 1 (2 bytes) in Handshake packet but got {} bytes",
+                buf.remaining()
+            ));
+        }
+
         let capabilities_1 = buf.get_u16_le(); // int<2>
         let mut capabilities = Capabilities::from_bits_truncate(capabilities_1.into());
 
+        if buf.remaining() < 5 {
+            return Err(err_protocol!(
+                "expected collation (1 byte), status (2 bytes) and capability flags part 2 (2 bytes) in Handshake packet but got {} bytes",
+                buf.remaining()
+            ));
+        }
+
         let collation = buf.get_u8(); // int<1>
         let status = Status::from_bits_truncate(buf.get_u16_le());
 
         let capabilities_2 = buf.get_u16_le(); // int<2>
         capabilities |= Capabilities::from_bits_truncate(((capabilities_2 as u32) << 16).into());
 
+        if buf.is_empty() {
+            return Err(err_protocol!(
+                "expected auth-plugin-data length (1 byte) in Handshake packet but got 0 bytes"
+            ));
+        }
+
         let auth_plugin_data_len = if capabilities.contains(Capabilities::PLUGIN_AUTH) {
             buf.get_u8()
         } else {
             buf.advance(1); // int<1>
             0
         };
 
+        if buf.remaining() < 10 {
+            return Err(err_protocol!(
+                "expected reserved bytes (6 bytes) and reserved bytes or capability flags part 3 (4 bytes) in Handshake packet but got {} bytes",
+                buf.remaining()
+            ));
+        }
+
         buf.advance(6); // reserved: string<6>
 
         if capabilities.contains(Capabilities::MYSQL) {
@@ -62,8 +107,18 @@ impl ProtocolDecode<'_> for Handshake {
         }
 
         let auth_plugin_data_2 = if capabilities.contains(Capabilities::SECURE_CONNECTION) {
-            let len = cmp::max(auth_plugin_data_len.saturating_sub(9), 12);
-            let v = buf.get_bytes(len as usize);
+            let len = cmp::max(auth_plugin_data_len.saturating_sub(9), 12) as usize;
+
+            // the attacker-controlled length must fit in the remaining buffer
+            // *with* the trailing NUL-terminator, i.e. `len + 1` bytes
+            if buf.remaining() <= len {
+                return Err(err_protocol!(
+                    "expected auth-plugin-data-part-2 ({len} bytes) and NUL-terminator in Handshake packet but got {} bytes",
+                    buf.remaining()
+                ));
+            }
+
+            let v = buf.get_bytes(len);
             buf.advance(1); // NUL-terminator
 
             v
@@ -197,3 +252,47 @@ fn test_decode_handshake_mariadb_10_4_7() {
         &[116, 54, 76, 92, 106, 34, 100, 83, 85, 49, 52, 79, 112, 104, 57, 34, 60, 72, 53, 110,]
     );
 }
+
+#[test]
+fn test_decode_handshake_truncated_never_panics() {
+    const MYSQL_8_0_18: &[u8] = b"\n8.0.18\x00\x19\x00\x00\x00\x114aB0c\x06g\x00\xff\xff\xff\x02\x00\xff\xc7\x15\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00tL\x03s\x0f[4\rl4. \x00caching_sha2_password\x00";
+    const MARIA_DB_10_4_7: &[u8] = b"\n5.5.5-10.4.7-MariaDB-1:10.4.7+maria~bionic\x00\x0b\x00\x00\x00t6L\\j\"dS\x00\xfe\xf7\x08\x02\x00\xff\x81\x15\x00\x00\x00\x00\x00\x00\x07\x00\x00\x00U14Oph9\"<H5n\x00mysql_native_password\x00";
+
+    for packet in [MYSQL_8_0_18, MARIA_DB_10_4_7] {
+        // every strict prefix of a valid handshake is an error, never a panic
+        for len in 0..packet.len() {
+            assert!(
+                Handshake::decode(packet[..len].into()).is_err(),
+                "expected error for handshake truncated to {len} bytes"
+            );
+        }
+
+        // and the full packet still decodes
+        assert!(Handshake::decode(packet.into()).is_ok());
+    }
+}
+
+#[test]
+fn test_decode_handshake_rejects_oversized_auth_plugin_data_len() {
+    // 0xc7ff capability flags part 2 sets SECURE_CONNECTION and PLUGIN_AUTH,
+    // so the auth-plugin-data length byte governs the part-2 read; it claims
+    // 255 bytes (246 after subtracting 9) but only a few bytes follow
+    let mut packet = Vec::new();
+    packet.push(10_u8); // protocol version
+    packet.extend_from_slice(b"8.0.0\0"); // server version
+    packet.extend_from_slice(&1_u32.to_le_bytes()); // connection id
+    packet.extend_from_slice(b"01234567"); // auth-plugin-data-part-1
+    packet.push(0); // reserved
+    packet.extend_from_slice(&0xffff_u16.to_le_bytes()); // capability flags part 1
+    packet.push(0); // collation
+    packet.extend_from_slice(&2_u16.to_le_bytes()); // status
+    packet.extend_from_slice(&0xc7ff_u16.to_le_bytes()); // capability flags part 2
+    packet.push(255); // auth-plugin-data length
+    packet.extend_from_slice(&[0; 6]); // reserved
+    packet.extend_from_slice(&[0; 4]); // reserved
+    packet.extend_from_slice(b"short-junk\0");
+
+    let err = Handshake::decode(packet.into()).unwrap_err();
+
+    assert!(err.to_string().contains("auth-plugin-data-part-2"), "{err}");
+}
@@ -28,9 +28,36 @@ impl ProtocolDecode<'_, Capabilities> for EofPacket {
             ));
         }
 
+        if buf.remaining() < 4 {
+            return Err(err_protocol!(
+                "expected warnings (2 bytes) and status flags (2 bytes) in EOF_Packet but got {} bytes",
+                buf.remaining()
+            ));
+        }
+
         let warnings = buf.get_u16_le();
         let status = Status::from_bits_truncate(buf.get_u16_le());
 
         Ok(Self { status, warnings })
     }
 }
+
+#[test]
+fn test_decode_eof_packet() {
+    let p = EofPacket::decode_with(
+        b"\xfe\x00\x00\x02\x00".as_ref().into(),
+        Capabilities::empty(),
+    )
+    .unwrap();
+
+    assert_eq!(p.warnings, 0);
+    assert!(p.status.contains(Status::SERVER_STATUS_AUTOCOMMIT));
+}
+
+#[test]
+fn test_decode_eof_packet_truncated() {
+    // only 2 bytes after the 0xfe header, need 4
+    let p = EofPacket::decode_with(b"\xfe\x00\x00".as_ref().into(), Capabilities::empty());
+
+    assert!(p.is_err());
+}
@@ -25,13 +25,28 @@ impl ProtocolDecode<'_, Capabilities> for ErrPacket {
             ));
         }
 
+        if buf.remaining() < 2 {
+            return Err(err_protocol!(
+                "expected error code (2 bytes) in ERR_Packet but got {} bytes",
+                buf.remaining()
+            ));
+        }
+
         let error_code = buf.get_u16_le();
         let mut sql_state = None;
 
         if capabilities.contains(Capabilities::PROTOCOL_41) {
             // If the next byte is '#' then we have a SQL STATE
             if buf.starts_with(b"#") {
                 buf.advance(1);
+
+                if buf.remaining() < 5 {
+                    return Err(err_protocol!(
+                        "expected SQL state (5 bytes) after '#' marker in ERR_Packet but got {} bytes",
+                        buf.remaining()
+                    ));
+                }
+
                 sql_state = Some(buf.get_str(5)?);
             }
         }
@@ -69,3 +84,25 @@ fn test_decode_err_packet_unknown_database() {
     assert_eq!(p.sql_state.as_deref(), Some("42000"));
     assert_eq!(&p.error_message, "Unknown database \'unknown\'");
 }
+
+#[test]
+fn test_decode_err_packet_truncated_error_code() {
+    // only 1 byte of the 2 byte error code is present
+    const ERR_TRUNCATED_CODE: &[u8] = b"\xff\x84";
+
+    let err =
+        ErrPacket::decode_with(ERR_TRUNCATED_CODE.into(), Capabilities::PROTOCOL_41).unwrap_err();
+
+    assert!(err.to_string().contains("expected error code"), "{err}");
+}
+
+#[test]
+fn test_decode_err_packet_truncated_sql_state() {
+    // '#' marker but fewer than 5 bytes of SQL state
+    const ERR_TRUNCATED_STATE: &[u8] = b"\xff\x84\x04#420";
+
+    let err =
+        ErrPacket::decode_with(ERR_TRUNCATED_STATE.into(), Capabilities::PROTOCOL_41).unwrap_err();
+
+    assert!(err.to_string().contains("expected SQL state"), "{err}");
+}