Fail closed on malformed stream decode

Author: eldryoth

CHANGELOG.md

@@ -107,6 +107,8 @@
   short-write regressions for encoder and decoder adapters.
 - Changed reader-side stream output draining to copy queued data into caller
   buffers in bounded slices while clearing queue slots as bytes are consumed.
+- Hardened stream decoders to fail closed after malformed Base64 input, with
+  `is_failed()` state inspection and regressions for writer and reader adapters.
 
 ## 0.8.0 - 2026-05-16
 

README.md

@@ -576,6 +576,9 @@ including recovery readiness, pending quantum state, and fixed output queue
 capacity. As with other `std::io::Write` implementations, direct `write()`
 calls may accept only part of the provided input while buffering encoded or
 decoded output; use `write_all()` when the whole input slice must be consumed.
+Decoder writer and reader adapters fail closed after malformed Base64 input;
+`is_failed()` exposes that state, while unchecked `into_inner()` remains
+available for explicit recovery of the wrapped object.
 
 URL-safe, no-padding encoding:
 

docs/RELEASE_EVIDENCE.md

@@ -117,6 +117,8 @@ The release gate runs:
   is retained until the wrapped writer reports bytes accepted
 - stream reader output queues drain into caller buffers in bounded slices while
   consumed queue slots are cleared
+- stream decoder fail-closed tests proving malformed Base64 input poisons the
+  adapter while preserving explicit unchecked inner recovery
 - stream fuzz coverage for chunked writers, fragmented reader sources, and
   adjacent framed payload boundaries
 - Kani proofs through `scripts/check_kani.sh` when Kani is installed and its

src/lib.rs

@@ -959,6 +959,7 @@ pub mod stream {
         output: OutputQueue<1024>,
         finished: bool,
         finalized: bool,
+        failed: bool,
     }
 
     impl<W, A, const PAD: bool> Decoder<W, A, PAD>
@@ -976,6 +977,7 @@ pub mod stream {
                 output: OutputQueue::new(),
                 finished: false,
                 finalized: false,
+                failed: false,
             }
         }
 
@@ -1075,11 +1077,21 @@ pub mod stream {
             self.finalized
         }
 
+        /// Returns whether this decoder has rejected malformed Base64 input.
+        ///
+        /// Once this returns `true`, later writes, flushes, and finalization
+        /// attempts return an error. The unchecked [`Self::into_inner`] method
+        /// can still be used for explicit recovery of the wrapped writer.
+        #[must_use]
+        pub const fn is_failed(&self) -> bool {
+            self.failed
+        }
+
         /// Returns whether [`Self::try_into_inner`] can recover the wrapped
         /// writer without discarding pending encoded input.
         #[must_use]
         pub const fn can_into_inner(&self) -> bool {
-            !self.has_pending_input() && !self.has_buffered_output()
+            !self.is_failed() && !self.has_pending_input() && !self.has_buffered_output()
         }
 
         /// Consumes the decoder without flushing pending input.
@@ -1165,6 +1177,7 @@ pub mod stream {
                 .field("can_into_inner", &self.can_into_inner())
                 .field("terminal_padding", &self.finished)
                 .field("finalized", &self.finalized)
+                .field("failed", &self.failed)
                 .finish()
         }
     }
@@ -1184,6 +1197,9 @@ pub mod stream {
         /// the caller still owns the decoder for diagnostics or explicit
         /// recovery.
         pub fn try_finish(&mut self) -> io::Result<()> {
+            if self.failed {
+                return Err(stream_decoder_failed_error());
+            }
             if !self.finalized {
                 self.queue_pending_final()?;
                 self.finalized = true;
@@ -1208,7 +1224,10 @@ pub mod stream {
             let mut decoded = [0u8; 3];
             let result = self.queue_decoded_temp(&pending[..pending_len], &mut decoded);
             crate::wipe_bytes(&mut pending);
-            result?;
+            if let Err(err) = result {
+                self.clear_pending();
+                return Err(err);
+            }
             self.clear_pending();
             Ok(())
         }
@@ -1229,6 +1248,7 @@ pub mod stream {
                 Ok(written) => written,
                 Err(err) => {
                     crate::wipe_bytes(decoded);
+                    self.failed = true;
                     return Err(decode_error_to_io(err));
                 }
             };
@@ -1275,6 +1295,9 @@ pub mod stream {
         A: Alphabet,
     {
         fn write(&mut self, input: &[u8]) -> io::Result<usize> {
+            if self.failed {
+                return Err(stream_decoder_failed_error());
+            }
             if input.is_empty() {
                 self.drain_output()?;
                 return Ok(0);
@@ -1287,6 +1310,7 @@ pub mod stream {
                 ));
             }
             if self.finished {
+                self.failed = true;
                 return Err(trailing_input_after_padding_error());
             }
 
@@ -1305,7 +1329,10 @@ pub mod stream {
                 quad[self.pending_len..].copy_from_slice(&input[..needed]);
                 let result = self.queue_full_quad(quad);
                 crate::wipe_bytes(&mut quad);
-                result?;
+                if let Err(err) = result {
+                    self.clear_pending();
+                    return Err(err);
+                }
                 self.clear_pending();
                 consumed += needed;
                 return Ok(consumed);
@@ -1330,6 +1357,9 @@ pub mod stream {
         }
 
         fn flush(&mut self) -> io::Result<()> {
+            if self.failed {
+                return Err(stream_decoder_failed_error());
+            }
             self.drain_output()?;
             self.inner_mut().flush()
         }
@@ -1346,6 +1376,13 @@ pub mod stream {
         )
     }
 
+    fn stream_decoder_failed_error() -> io::Error {
+        io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "base64 stream decoder is failed after malformed input",
+        )
+    }
+
     /// A streaming Base64 decoder for `std::io::Read`.
     ///
     /// For padded engines, this reader stops at the terminal padded Base64
@@ -1363,6 +1400,7 @@ pub mod stream {
         output: OutputQueue<3>,
         finished: bool,
         terminal_seen: bool,
+        failed: bool,
     }
 
     impl<R, A, const PAD: bool> DecoderReader<R, A, PAD>
@@ -1380,6 +1418,7 @@ pub mod stream {
                 output: OutputQueue::new(),
                 finished: false,
                 terminal_seen: false,
+                failed: false,
             }
         }
 
@@ -1489,11 +1528,22 @@ pub mod stream {
             self.finished && self.output.is_empty()
         }
 
+        /// Returns whether this decoder reader has rejected malformed Base64
+        /// input.
+        ///
+        /// Once this returns `true`, later reads return an error. The unchecked
+        /// [`Self::into_inner`] method can still be used for explicit recovery
+        /// of the wrapped reader.
+        #[must_use]
+        pub const fn is_failed(&self) -> bool {
+            self.failed
+        }
+
         /// Returns whether [`Self::try_into_inner`] can recover the wrapped
         /// reader without discarding buffered decoded output.
         #[must_use]
         pub const fn can_into_inner(&self) -> bool {
-            self.is_finished()
+            !self.is_failed() && self.is_finished()
         }
 
         /// Consumes the decoder reader and returns the wrapped reader.
@@ -1575,6 +1625,7 @@ pub mod stream {
                 .field("can_into_inner", &self.can_into_inner())
                 .field("finished", &self.finished)
                 .field("terminal_padding", &self.terminal_seen)
+                .field("failed", &self.failed)
                 .finish()
         }
     }
@@ -1588,6 +1639,9 @@ pub mod stream {
             if output.is_empty() {
                 return Ok(0);
             }
+            if self.failed {
+                return Err(stream_decoder_failed_error());
+            }
 
             while self.output.is_empty() && !self.finished {
                 self.fill_output()?;
@@ -1603,6 +1657,9 @@ pub mod stream {
         A: Alphabet,
     {
         fn fill_output(&mut self) -> io::Result<()> {
+            if self.failed {
+                return Err(stream_decoder_failed_error());
+            }
             if self.terminal_seen {
                 self.finished = true;
                 return Ok(());
@@ -1656,6 +1713,7 @@ pub mod stream {
                 Ok(written) => written,
                 Err(err) => {
                     crate::wipe_bytes(&mut decoded);
+                    self.failed = true;
                     return Err(decode_error_to_io(err));
                 }
             };

tests/rfc4648.rs

@@ -3341,14 +3341,20 @@ fn stream_decoder_try_finish_reports_bad_final_pending_input() {
     let err = decoder.try_finish().unwrap_err();
     assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
     assert!(!decoder.is_finalized());
-    assert_eq!(decoder.pending_len(), 1);
-    assert!(decoder.has_pending_input());
+    assert!(decoder.is_failed());
+    assert_eq!(decoder.pending_len(), 0);
+    assert!(!decoder.has_pending_input());
+    assert!(!decoder.can_into_inner());
     assert!(decoder.get_ref().is_empty());
 
-    decoder.write_all(b"Gk=").unwrap();
-    decoder.try_finish().unwrap();
-    assert!(decoder.is_finalized());
-    assert_eq!(decoder.get_ref(), b"hi");
+    assert_eq!(
+        decoder.write_all(b"Gk=").unwrap_err().kind(),
+        std::io::ErrorKind::InvalidInput
+    );
+    assert_eq!(
+        decoder.try_finish().unwrap_err().kind(),
+        std::io::ErrorKind::InvalidInput
+    );
 }
 
 #[cfg(feature = "stream")]
@@ -3357,6 +3363,8 @@ fn stream_decoder_rejects_trailing_input_after_padding() {
     let mut decoder = Decoder::new(Vec::new(), STANDARD);
     let err = decoder.write_all(b"aGk=AA").unwrap_err();
     assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+    assert!(decoder.is_failed());
+    assert!(!decoder.can_into_inner());
 }
 
 #[cfg(feature = "stream")]
@@ -3366,6 +3374,31 @@ fn stream_decoder_rejects_short_trailing_input_after_pending_padding() {
     decoder.write_all(b"aG").unwrap();
     let err = decoder.write_all(b"k=A").unwrap_err();
     assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+    assert!(decoder.is_failed());
+    assert_eq!(decoder.pending_len(), 0);
+    assert!(!decoder.can_into_inner());
+}
+
+#[cfg(feature = "stream")]
+#[test]
+fn stream_decoder_fails_closed_after_malformed_input() {
+    let mut decoder = Decoder::new(Vec::new(), STANDARD);
+
+    let err = decoder.write(b"!!!!").unwrap_err();
+    assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+    assert!(decoder.is_failed());
+    assert!(!decoder.is_finalized());
+    assert_eq!(decoder.pending_len(), 0);
+    assert!(!decoder.can_into_inner());
+
+    assert_eq!(
+        decoder.write(b"aGk=").unwrap_err().kind(),
+        std::io::ErrorKind::InvalidInput
+    );
+    assert_eq!(
+        decoder.flush().unwrap_err().kind(),
+        std::io::ErrorKind::InvalidInput
+    );
 }
 
 #[cfg(feature = "stream")]
@@ -3542,6 +3575,32 @@ fn stream_decoder_reader_rejects_bad_final_pending_input() {
     let mut reader = DecoderReader::new(&b"a"[..], STANDARD);
     let mut decoded = Vec::new();
     assert!(reader.read_to_end(&mut decoded).is_err());
+    assert!(reader.is_failed());
+    assert_eq!(reader.pending_len(), 0);
+    assert!(!reader.can_into_inner());
+
+    let mut output = [0u8; 1];
+    assert_eq!(
+        reader.read(&mut output).unwrap_err().kind(),
+        std::io::ErrorKind::InvalidInput
+    );
+}
+
+#[cfg(feature = "stream")]
+#[test]
+fn stream_decoder_reader_fails_closed_after_malformed_input() {
+    let mut reader = DecoderReader::new(Cursor::new(&b"!!!!aGk="[..]), STANDARD);
+    let mut output = [0u8; 1];
+
+    let err = reader.read(&mut output).unwrap_err();
+    assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+    assert!(reader.is_failed());
+    assert!(!reader.can_into_inner());
+
+    assert_eq!(
+        reader.read(&mut output).unwrap_err().kind(),
+        std::io::ErrorKind::InvalidInput
+    );
 }
 
 #[cfg(feature = "stream")]