fix(codecs): correct operand order in octet-counting framer underflow

changelog.d/octet_counting_discard_underflow.fix.md

@@ -0,0 +1,3 @@
+Fixed an integer underflow in the octet-counting framer (used by TCP `syslog` sources) that occurred when an over-length, length-prefixed message was split across multiple reads. Previously the decoder could panic in debug builds, or in release builds wrap the remaining-bytes counter to a huge value, wedging the decoder and silently dropping all subsequent input on that connection.
+
+authors: hhh6593

lib/codecs/src/decoding/framing/octet_counting.rs

@@ -105,7 +105,7 @@ impl OctetCountingDecoder {
                 //
                 // There aren't enough in this frame so we need to discard the
                 // entire frame and adjust the amount to discard accordingly.
-                self.octet_decoding = Some(State::Discarding(src.len() - chars));
+                self.octet_decoding = Some(State::Discarding(chars - src.len()));
                 src.advance(src.len());
                 Ok(None)
             }
@@ -407,4 +407,23 @@ mod tests {
         assert!(result.is_err());
         assert_eq!(b"32 something valid"[..], buffer);
     }
+
+    #[test]
+    fn octet_decode_discard_partial_frame_underflow() {
+        let mut decoder = OctetCountingDecoder::new_with_max_length(16);
+        let mut buffer = BytesMut::with_capacity(32);
+
+        // A length prefix of 26 exceeds the max length of 16, so the decoder
+        // enters the discarding state with 26 bytes to discard, leaving "abc".
+        buffer.put(&b"26 abc"[..]);
+        let _ = decoder.decode(&mut buffer);
+        assert_eq!(decoder.octet_decoding, Some(State::Discarding(26)));
+
+        // Only three more bytes arrive, so the buffer holds fewer bytes than
+        // remain to be discarded. This is the branch that previously underflowed.
+        buffer.put(&b"def"[..]);
+        let result = decoder.decode(&mut buffer);
+        assert_eq!(Ok(None), result.map_err(|_| false));
+        assert_eq!(decoder.octet_decoding, Some(State::Discarding(20)));
+    }
 }