Fix phase-1/phase-2 validation gap in RESP decoder

src/parse.rs

@@ -27,15 +27,28 @@ fn count_value(src: &[u8], pos: &mut usize, depth: u8) -> Result<(), RespError>
     *pos += 1;
 
     match prefix {
-        b'+' | b'-' | b':' | b',' | b'(' => {
+        b'+' | b'-' | b':' => {
             read_line(src, pos)?;
         }
+        b',' => {
+            let line = read_line(src, pos)?;
+            parse_double(line)?;
+        }
+        b'(' => {
+            let line = read_line(src, pos)?;
+            validate_bignumber(line)?;
+        }
         b'_' => {
             expect_crlf(src, pos)?;
         }
         b'#' => {
-            require(src, *pos, 3)?;
-            *pos += 3;
+            require(src, *pos, 1)?;
+            match src[*pos] {
+                b't' | b'f' => {}
+                byte => return Err(RespError::invalid_type(byte)),
+            }
+            *pos += 1;
+            expect_crlf(src, pos)?;
         }
         b'$' => {
             let len = read_length(src, pos)?;
@@ -57,13 +70,23 @@ fn count_value(src: &[u8], pos: &mut usize, depth: u8) -> Result<(), RespError>
                 }
             }
         }
-        b'!' | b'=' => {
+        b'!' => {
             let len = read_length(src, pos)?;
             if len < 0 {
                 return Err(RespError::InvalidLength);
             }
             skip_bulk(src, pos, to_count(len)?)?;
         }
+        b'=' => {
+            let len = read_length(src, pos)?;
+            if len < 4 {
+                return Err(RespError::InvalidVerbatim);
+            }
+            let (s, _e) = bulk_range(src, pos, to_count(len)?)?;
+            if src[s + 3] != b':' {
+                return Err(RespError::InvalidVerbatim);
+            }
+        }
         b'%' => {
             let len = read_length(src, pos)?;
             if len < 0 {
@@ -106,8 +129,8 @@ fn count_value(src: &[u8], pos: &mut usize, depth: u8) -> Result<(), RespError>
 ///
 /// # Safety invariant
 /// Every index into `src` is safe without bounds-checking because `frame_len`
-/// (which runs `count_value`) has already confirmed that `src` contains a
-/// structurally complete frame. Do not call this function on a partial buffer.
+/// has already validated structure and semantics. Do not call this on a partial
+/// buffer or without a prior successful `frame_len` pass.
 pub(crate) fn build_value(src: &Bytes, pos: &mut usize, depth: u8) -> Result<Value, RespError> {
     if depth > MAX_DEPTH {
         return Err(RespError::DepthLimitExceeded);
@@ -605,6 +628,14 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn boolean_invalid_byte() {
+        assert!(matches!(
+            parse_one(b"#x\r\n"),
+            Err(RespError::InvalidTypeByte { byte: b'x' })
+        ));
+    }
+
     #[test]
     fn invalid_type_byte() {
         assert!(matches!(

tests/error_paths.rs

@@ -189,6 +189,21 @@ fn error_leaves_buffer_unconsumed() {
     assert!(!buf.is_empty(), "invalid bytes should remain in the buffer after error");
 }
 
+#[test]
+fn semantic_error_leaves_buffer_unconsumed() {
+    for wire in [
+        &b",not-a-double\r\n"[..],
+        &b"(not-a-number\r\n"[..],
+        &b"=4\r\nabcd\r\n"[..],
+        &b"#x\r\n"[..],
+    ] {
+        let mut codec = RespCodec::resp2();
+        let mut buf = BytesMut::from(wire);
+        assert!(codec.decode(&mut buf).is_err(), "wire: {wire:?}");
+        assert_eq!(buf.as_ref(), wire, "semantic error must not consume input");
+    }
+}
+
 #[test]
 fn codec_usable_after_buffer_cleared() {
     let mut codec = RespCodec::resp2();