fix possible panic reading data from server

Found by inspection. It'd be nice to fuzz the tokio module to more
reliably find problems like this, but it looks like I'd have to make
a bunch more stuff public for that. Maybe later...

CHANGELOG.md

@@ -3,6 +3,7 @@
 *   fix depacketization of fragmented AAC frames
 *   [#52](https://github.com/scottlamb/retina/issues/52): allow compatibility
     with cameras that incorrectly omit the SDP origin line.
+*   fix panic if RTSP server precedes a data message with a CRLF.
 
 ## `v0.3.7` (2022-01-28)
 

src/tokio.rs

@@ -152,6 +152,7 @@ struct Codec {
 /// An intermediate error type that exists because [`Framed`] expects the
 /// codec's error type to implement `From<std::io::Error>`, and [`Error`]
 /// takes additional context.
+#[derive(Debug)]
 enum CodecError {
     IoError(std::io::Error),
     ParseError { description: String, pos: u64 },
@@ -165,6 +166,13 @@ impl std::convert::From<std::io::Error> for CodecError {
 
 impl Codec {
     fn parse_msg(&self, src: &mut BytesMut) -> Result<Option<(usize, Message<Bytes>)>, CodecError> {
+        // Skip whitespace as `rtsp-types` does. It's important to also do it here, or we might
+        // skip the our own data message encoding (next if) then hit
+        // unreachable! after rtsp-types returns Message::Data.
+        while src.starts_with(b"\r\n") {
+            src.advance(2);
+        }
+
         if !src.is_empty() && src[0] == b'$' {
             // Fast path for interleaved data, avoiding MessageRef -> Message<&[u8]> ->
             // Message<Bytes> conversion. This speeds things up quite a bit in practice,
@@ -314,3 +322,21 @@ impl UdpPair {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use tokio_util::codec::Decoder;
+
+    use super::*;
+
+    #[test]
+    fn crlf_data() {
+        let mut codec = Codec {
+            ctx: ConnectionContext::dummy(),
+            read_pos: 0,
+        };
+        let mut buf = BytesMut::from(&b"\r\n$\x00\x00\x04asdfrest"[..]);
+        codec.decode(&mut buf).unwrap();
+        assert_eq!(&buf[..], b"rest");
+    }
+}