Snappy additional sanity checks (#1625)

## Issue Addressed

N/A

## Proposed Changes

Adds the following check from the spec

> A reader SHOULD NOT read more than max_encoded_len(n) bytes after reading the SSZ length-prefix n from the header.

beacon_node/eth2_libp2p/src/rpc/codec/ssz_snappy.rs

@@ -99,6 +99,7 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyInboundCodec<TSpec> {
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
         if self.len.is_none() {
             // Decode the length of the uncompressed bytes from an unsigned varint
+            // Note: length-prefix of > 10 bytes(uint64) would be a decoding error
             match self.inner.decode(src).map_err(RPCError::from)? {
                 Some(length) => {
                     self.len = Some(length);
@@ -116,7 +117,7 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyInboundCodec<TSpec> {
         let mut reader = FrameDecoder::new(Cursor::new(&src));
         let mut decoded_buffer = vec![0; length];
 
-        match reader.read_exact(&mut decoded_buffer) {
+        match read_exact(&mut reader, &mut decoded_buffer, length) {
             Ok(()) => {
                 // `n` is how many bytes the reader read in the compressed stream
                 let n = reader.get_ref().position();
@@ -194,8 +195,7 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyInboundCodec<TSpec> {
                 }
             }
             Err(e) => match e.kind() {
-                // Haven't received enough bytes to decode yet
-                // TODO: check if this is the only Error variant where we return `Ok(None)`
+                // Haven't received enough bytes to decode yet, wait for more
                 ErrorKind::UnexpectedEof => Ok(None),
                 _ => Err(e).map_err(RPCError::from),
             },
@@ -277,6 +277,7 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyOutboundCodec<TSpec> {
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
         if self.len.is_none() {
             // Decode the length of the uncompressed bytes from an unsigned varint
+            // Note: length-prefix of > 10 bytes(uint64) would be a decoding error
             match self.inner.decode(src).map_err(RPCError::from)? {
                 Some(length) => {
                     self.len = Some(length as usize);
@@ -293,7 +294,7 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyOutboundCodec<TSpec> {
         }
         let mut reader = FrameDecoder::new(Cursor::new(&src));
         let mut decoded_buffer = vec![0; length];
-        match reader.read_exact(&mut decoded_buffer) {
+        match read_exact(&mut reader, &mut decoded_buffer, length) {
             Ok(()) => {
                 // `n` is how many bytes the reader read in the compressed stream
                 let n = reader.get_ref().position();
@@ -364,8 +365,7 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyOutboundCodec<TSpec> {
                 }
             }
             Err(e) => match e.kind() {
-                // Haven't received enough bytes to decode yet
-                // TODO: check if this is the only Error variant where we return `Ok(None)`
+                // Haven't received enough bytes to decode yet, wait for more
                 ErrorKind::UnexpectedEof => Ok(None),
                 _ => Err(e).map_err(RPCError::from),
             },
@@ -398,7 +398,7 @@ impl<TSpec: EthSpec> OutboundCodec<RPCRequest<TSpec>> for SSZSnappyOutboundCodec
         }
         let mut reader = FrameDecoder::new(Cursor::new(&src));
         let mut decoded_buffer = vec![0; length];
-        match reader.read_exact(&mut decoded_buffer) {
+        match read_exact(&mut reader, &mut decoded_buffer, length) {
             Ok(()) => {
                 // `n` is how many bytes the reader read in the compressed stream
                 let n = reader.get_ref().position();
@@ -409,11 +409,67 @@ impl<TSpec: EthSpec> OutboundCodec<RPCRequest<TSpec>> for SSZSnappyOutboundCodec
                 )?)))
             }
             Err(e) => match e.kind() {
-                // Haven't received enough bytes to decode yet
-                // TODO: check if this is the only Error variant where we return `Ok(None)`
+                // Haven't received enough bytes to decode yet, wait for more
                 ErrorKind::UnexpectedEof => Ok(None),
                 _ => Err(e).map_err(RPCError::from),
             },
         }
     }
 }
+
+/// Wrapper over `read` implementation of `FrameDecoder`.
+///
+/// Works like the standard `read_exact` implementation, except that it returns an error if length of
+// compressed bytes read from the underlying reader is greater than worst case compression length for snappy.
+fn read_exact<T: std::convert::AsRef<[u8]>>(
+    reader: &mut FrameDecoder<Cursor<T>>,
+    mut buf: &mut [u8],
+    uncompressed_length: usize,
+) -> Result<(), std::io::Error> {
+    // Calculate worst case compression length for given uncompressed length
+    let max_compressed_len = snap::raw::max_compress_len(uncompressed_length) as u64;
+
+    // Initialize the position of the reader
+    let mut pos = reader.get_ref().position();
+    let mut count = 0;
+    while !buf.is_empty() {
+        match reader.read(buf) {
+            Ok(0) => break,
+            Ok(n) => {
+                let tmp = buf;
+                buf = &mut tmp[n..];
+            }
+            Err(ref e) if e.kind() == ErrorKind::Interrupted => {}
+            Err(e) => return Err(e),
+        }
+        // Get current position of reader
+        let curr_pos = reader.get_ref().position();
+        // Note: reader should always advance forward. However, this behaviour
+        // depends on the implementation of `snap::FrameDecoder`, so it is better
+        // to check to avoid underflow panic.
+        if curr_pos > pos {
+            count += reader.get_ref().position() - pos;
+            pos = curr_pos;
+        } else {
+            return Err(std::io::Error::new(
+                ErrorKind::InvalidData,
+                "snappy: reader is not advanced forward while reading",
+            ));
+        }
+
+        if count > max_compressed_len {
+            return Err(std::io::Error::new(
+                ErrorKind::InvalidData,
+                "snappy: compressed data is > max_compressed_len",
+            ));
+        }
+    }
+    if !buf.is_empty() {
+        Err(std::io::Error::new(
+            ErrorKind::UnexpectedEof,
+            "failed to fill whole buffer",
+        ))
+    } else {
+        Ok(())
+    }
+}