use super::{Alphabet, DecodeError, EncodeError, Engine};
use std::io::{self, Read, Write};
struct OutputQueue<const CAP: usize> {
buffer: [u8; CAP],
start: usize,
len: usize,
}
impl<const CAP: usize> OutputQueue<CAP> {
const fn new() -> Self {
Self {
buffer: [0; CAP],
start: 0,
len: 0,
}
}
const fn is_empty(&self) -> bool {
self.len == 0
}
const fn len(&self) -> usize {
self.len
}
const fn capacity(&self) -> usize {
self.len + self.available_capacity()
}
fn push_slice(&mut self, input: &[u8]) -> io::Result<()> {
if input.len() > self.available_capacity() {
return Err(io::Error::other(
"base64 stream output queue capacity exceeded",
));
}
let mut read = 0;
while read < input.len() {
let write = (self.start + self.len) % CAP;
self.buffer[write] = input[read];
self.len += 1;
read += 1;
}
Ok(())
}
fn copy_front(&self, output: &mut [u8]) -> usize {
let count = core::cmp::min(self.len, output.len());
let first = core::cmp::min(count, CAP - self.start);
output[..first].copy_from_slice(&self.buffer[self.start..self.start + first]);
let second = count - first;
if second > 0 {
output[first..first + second].copy_from_slice(&self.buffer[..second]);
}
count
}
fn discard_front(&mut self, count: usize) {
let count = core::cmp::min(count, self.len);
let first = core::cmp::min(count, CAP - self.start);
crate::wipe_bytes(&mut self.buffer[self.start..self.start + first]);
let second = count - first;
if second > 0 {
crate::wipe_bytes(&mut self.buffer[..second]);
}
self.start = (self.start + count) % CAP;
self.len -= count;
if self.len == 0 {
self.start = 0;
}
}
fn pop_slice(&mut self, output: &mut [u8]) -> usize {
let count = self.copy_front(output);
self.discard_front(count);
count
}
fn clear_all(&mut self) {
crate::wipe_bytes(&mut self.buffer);
self.start = 0;
self.len = 0;
}
const fn available_capacity(&self) -> usize {
CAP - self.len
}
}
pub struct Encoder<W, A, const PAD: bool>
where
A: Alphabet,
{
inner: Option<W>,
engine: Engine<A, PAD>,
pending: [u8; 2],
pending_len: usize,
output: OutputQueue<1024>,
finalized: bool,
failed: bool,
}
impl<W, A, const PAD: bool> Encoder<W, A, PAD>
where
A: Alphabet,
{
#[must_use]
pub const fn new(inner: W, engine: Engine<A, PAD>) -> Self {
Self {
inner: Some(inner),
engine,
pending: [0; 2],
pending_len: 0,
output: OutputQueue::new(),
finalized: false,
failed: false,
}
}
#[must_use]
pub fn get_ref(&self) -> &W {
self.inner_ref()
}
pub fn get_mut(&mut self) -> &mut W {
self.inner_mut()
}
#[must_use]
pub const fn engine(&self) -> Engine<A, PAD> {
self.engine
}
#[must_use]
pub const fn is_padded(&self) -> bool {
PAD
}
#[must_use]
pub const fn pending_len(&self) -> usize {
self.pending_len
}
#[must_use]
pub const fn has_pending_input(&self) -> bool {
self.pending_len != 0
}
#[must_use]
pub const fn pending_input_needed_len(&self) -> usize {
if self.has_pending_input() {
3 - self.pending_len
} else {
0
}
}
#[must_use]
pub const fn buffered_output_len(&self) -> usize {
self.output.len()
}
#[must_use]
pub const fn buffered_output_capacity(&self) -> usize {
self.output.capacity()
}
#[must_use]
pub const fn buffered_output_remaining_capacity(&self) -> usize {
self.output.available_capacity()
}
#[must_use]
pub const fn has_buffered_output(&self) -> bool {
!self.output.is_empty()
}
#[must_use]
pub const fn is_finalized(&self) -> bool {
self.finalized
}
#[must_use]
pub const fn is_failed(&self) -> bool {
self.failed
}
#[must_use]
pub const fn can_into_inner(&self) -> bool {
!self.is_failed() && !self.has_pending_input() && !self.has_buffered_output()
}
#[must_use]
pub fn into_inner(mut self) -> W {
self.take_inner()
}
#[allow(clippy::result_large_err)]
pub fn try_into_inner(mut self) -> Result<W, Self> {
if !self.can_into_inner() {
return Err(self);
}
Ok(self.take_inner())
}
fn inner_ref(&self) -> &W {
match &self.inner {
Some(inner) => inner,
None => unreachable!("stream encoder inner writer was already taken"),
}
}
fn inner_mut(&mut self) -> &mut W {
match &mut self.inner {
Some(inner) => inner,
None => unreachable!("stream encoder inner writer was already taken"),
}
}
fn take_inner(&mut self) -> W {
match self.inner.take() {
Some(inner) => inner,
None => unreachable!("stream encoder inner writer was already taken"),
}
}
fn clear_pending(&mut self) {
crate::wipe_bytes(&mut self.pending);
self.pending_len = 0;
}
fn clear_output(&mut self) {
self.output.clear_all();
}
}
impl<W, A, const PAD: bool> Drop for Encoder<W, A, PAD>
where
A: Alphabet,
{
fn drop(&mut self) {
self.clear_pending();
self.clear_output();
}
}
impl<W, A, const PAD: bool> core::fmt::Debug for Encoder<W, A, PAD>
where
A: Alphabet,
{
fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
formatter
.debug_struct("Encoder")
.field("inner", &redacted_inner_state(self.inner.is_some()))
.field("engine", &self.engine)
.field("pending", &"<redacted>")
.field("pending_len", &self.pending_len)
.field("pending_input_needed_len", &self.pending_input_needed_len())
.field("buffered_output_len", &self.output.len())
.field("buffered_output_capacity", &self.output.capacity())
.field(
"buffered_output_remaining_capacity",
&self.output.available_capacity(),
)
.field("can_into_inner", &self.can_into_inner())
.field("finalized", &self.finalized)
.field("failed", &self.failed)
.finish()
}
}
impl<W, A, const PAD: bool> Encoder<W, A, PAD>
where
W: Write,
A: Alphabet,
{
pub fn try_finish(&mut self) -> io::Result<()> {
if self.failed {
return Err(stream_encoder_failed_error());
}
if !self.finalized {
self.queue_pending_final()?;
self.finalized = true;
}
self.flush()
}
pub fn finish(mut self) -> io::Result<W> {
self.try_finish()?;
Ok(self.take_inner())
}
fn queue_pending_final(&mut self) -> io::Result<()> {
if self.pending_len == 0 {
return Ok(());
}
let mut pending = [0u8; 2];
pending[..self.pending_len].copy_from_slice(&self.pending[..self.pending_len]);
let pending_len = self.pending_len;
let mut encoded = [0u8; 4];
let result = self.queue_encoded_temp(&pending[..pending_len], &mut encoded);
crate::wipe_bytes(&mut pending);
result?;
self.clear_pending();
Ok(())
}
fn queue_encoded_temp(&mut self, input: &[u8], encoded: &mut [u8]) -> io::Result<()> {
let written = match self.engine.encode_slice(input, encoded) {
Ok(written) => written,
Err(err) => {
crate::wipe_bytes(encoded);
self.failed = true;
return Err(encode_error_to_io(err));
}
};
let result = self.output.push_slice(&encoded[..written]);
crate::wipe_bytes(encoded);
if result.is_err() {
self.failed = true;
}
result
}
fn drain_output(&mut self) -> io::Result<()> {
let mut chunk = [0u8; 1024];
while !self.output.is_empty() {
let pending = self.output.copy_front(&mut chunk);
let result = self.inner_mut().write(&chunk[..pending]);
crate::wipe_bytes(&mut chunk[..pending]);
match result {
Ok(0) => {
return Err(io::Error::new(
io::ErrorKind::WriteZero,
"base64 stream encoder could not drain buffered output",
));
}
Ok(written) => {
if written > pending {
self.failed = true;
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"wrapped writer reported more bytes than provided",
));
}
self.output.discard_front(written);
}
Err(err) => return Err(err),
}
}
Ok(())
}
}
impl<W, A, const PAD: bool> Write for Encoder<W, A, PAD>
where
W: Write,
A: Alphabet,
{
fn write(&mut self, input: &[u8]) -> io::Result<usize> {
if self.failed {
return Err(stream_encoder_failed_error());
}
self.drain_output()?;
if self.finalized {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"base64 stream encoder received input after finalization",
));
}
if input.is_empty() {
return Ok(0);
}
let mut consumed = 0;
if self.pending_len > 0 {
let needed = 3 - self.pending_len;
if input.len() < needed {
self.pending[self.pending_len..self.pending_len + input.len()]
.copy_from_slice(input);
self.pending_len += input.len();
return Ok(input.len());
}
let mut chunk = [0u8; 3];
chunk[..self.pending_len].copy_from_slice(&self.pending[..self.pending_len]);
chunk[self.pending_len..].copy_from_slice(&input[..needed]);
let mut encoded = [0u8; 4];
let result = self.queue_encoded_temp(&chunk, &mut encoded);
crate::wipe_bytes(&mut chunk);
result?;
self.clear_pending();
consumed += needed;
}
let remaining = &input[consumed..];
let full_len = remaining.len() / 3 * 3;
if full_len > 0 {
let max_by_queue = self.output.available_capacity() / 4 * 3;
let mut take = core::cmp::min(full_len, core::cmp::min(768, max_by_queue));
take -= take % 3;
if take == 0 {
return Ok(consumed);
}
let mut encoded = [0u8; 1024];
self.queue_encoded_temp(&remaining[..take], &mut encoded)?;
consumed += take;
if take < full_len {
return Ok(consumed);
}
}
let tail = &input[consumed..];
self.pending[..tail.len()].copy_from_slice(tail);
self.pending_len = tail.len();
consumed += tail.len();
Ok(consumed)
}
fn flush(&mut self) -> io::Result<()> {
if self.failed {
return Err(stream_encoder_failed_error());
}
self.drain_output()?;
self.inner_mut().flush()
}
}
fn encode_error_to_io(err: EncodeError) -> io::Error {
io::Error::new(io::ErrorKind::InvalidInput, err)
}
pub struct Decoder<W, A, const PAD: bool>
where
A: Alphabet,
{
inner: Option<W>,
engine: Engine<A, PAD>,
pending: [u8; 4],
pending_len: usize,
output: OutputQueue<1024>,
finished: bool,
failed: bool,
finalized: bool,
}
impl<W, A, const PAD: bool> Decoder<W, A, PAD>
where
A: Alphabet,
{
#[must_use]
pub const fn new(inner: W, engine: Engine<A, PAD>) -> Self {
Self {
inner: Some(inner),
engine,
pending: [0; 4],
pending_len: 0,
output: OutputQueue::new(),
finished: false,
finalized: false,
failed: false,
}
}
#[must_use]
pub fn get_ref(&self) -> &W {
self.inner_ref()
}
pub fn get_mut(&mut self) -> &mut W {
self.inner_mut()
}
#[must_use]
pub const fn engine(&self) -> Engine<A, PAD> {
self.engine
}
#[must_use]
pub const fn is_padded(&self) -> bool {
PAD
}
#[must_use]
pub const fn pending_len(&self) -> usize {
self.pending_len
}
#[must_use]
pub const fn has_pending_input(&self) -> bool {
self.pending_len != 0
}
#[must_use]
pub const fn pending_input_needed_len(&self) -> usize {
if self.has_pending_input() {
4 - self.pending_len
} else {
0
}
}
#[must_use]
pub const fn buffered_output_len(&self) -> usize {
self.output.len()
}
#[must_use]
pub const fn buffered_output_capacity(&self) -> usize {
self.output.capacity()
}
#[must_use]
pub const fn buffered_output_remaining_capacity(&self) -> usize {
self.output.available_capacity()
}
#[must_use]
pub const fn has_buffered_output(&self) -> bool {
!self.output.is_empty()
}
#[must_use]
pub const fn has_terminal_padding(&self) -> bool {
self.finished
}
#[must_use]
pub const fn is_finalized(&self) -> bool {
self.finalized
}
#[must_use]
pub const fn is_failed(&self) -> bool {
self.failed
}
#[must_use]
pub const fn can_into_inner(&self) -> bool {
!self.is_failed() && !self.has_pending_input() && !self.has_buffered_output()
}
#[must_use]
pub fn into_inner(mut self) -> W {
self.take_inner()
}
#[allow(clippy::result_large_err)]
pub fn try_into_inner(mut self) -> Result<W, Self> {
if !self.can_into_inner() {
return Err(self);
}
Ok(self.take_inner())
}
fn inner_ref(&self) -> &W {
match &self.inner {
Some(inner) => inner,
None => unreachable!("stream decoder inner writer was already taken"),
}
}
fn inner_mut(&mut self) -> &mut W {
match &mut self.inner {
Some(inner) => inner,
None => unreachable!("stream decoder inner writer was already taken"),
}
}
fn take_inner(&mut self) -> W {
match self.inner.take() {
Some(inner) => inner,
None => unreachable!("stream decoder inner writer was already taken"),
}
}
fn clear_pending(&mut self) {
crate::wipe_bytes(&mut self.pending);
self.pending_len = 0;
}
fn clear_output(&mut self) {
self.output.clear_all();
}
}
impl<W, A, const PAD: bool> Drop for Decoder<W, A, PAD>
where
A: Alphabet,
{
fn drop(&mut self) {
self.clear_pending();
self.clear_output();
}
}
impl<W, A, const PAD: bool> core::fmt::Debug for Decoder<W, A, PAD>
where
A: Alphabet,
{
fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
formatter
.debug_struct("Decoder")
.field("inner", &redacted_inner_state(self.inner.is_some()))
.field("engine", &self.engine)
.field("pending", &"<redacted>")
.field("pending_len", &self.pending_len)
.field("pending_input_needed_len", &self.pending_input_needed_len())
.field("buffered_output_len", &self.output.len())
.field("buffered_output_capacity", &self.output.capacity())
.field(
"buffered_output_remaining_capacity",
&self.output.available_capacity(),
)
.field("can_into_inner", &self.can_into_inner())
.field("terminal_padding", &self.finished)
.field("finalized", &self.finalized)
.field("failed", &self.failed)
.finish()
}
}
impl<W, A, const PAD: bool> Decoder<W, A, PAD>
where
W: Write,
A: Alphabet,
{
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;
}
self.flush()
}
pub fn finish(mut self) -> io::Result<W> {
self.try_finish()?;
Ok(self.take_inner())
}
fn queue_pending_final(&mut self) -> io::Result<()> {
if self.pending_len == 0 {
return Ok(());
}
let mut pending = [0u8; 4];
pending[..self.pending_len].copy_from_slice(&self.pending[..self.pending_len]);
let pending_len = self.pending_len;
let mut decoded = [0u8; 3];
let result = self.queue_decoded_temp(&pending[..pending_len], &mut decoded);
crate::wipe_bytes(&mut pending);
if let Err(err) = result {
self.clear_pending();
return Err(err);
}
self.clear_pending();
Ok(())
}
fn queue_full_quad(&mut self, mut input: [u8; 4]) -> io::Result<()> {
let mut decoded = [0u8; 3];
let result = self.queue_decoded_temp(&input, &mut decoded);
crate::wipe_bytes(&mut input);
let written = result?;
if written < 3 {
self.finished = true;
}
Ok(())
}
fn queue_decoded_temp(&mut self, input: &[u8], decoded: &mut [u8]) -> io::Result<usize> {
let written = match self.engine.decode_slice(input, decoded) {
Ok(written) => written,
Err(err) => {
crate::wipe_bytes(decoded);
self.failed = true;
return Err(decode_error_to_io(err));
}
};
let result = self.output.push_slice(&decoded[..written]);
crate::wipe_bytes(decoded);
result?;
Ok(written)
}
fn drain_output(&mut self) -> io::Result<()> {
let mut chunk = [0u8; 1024];
while !self.output.is_empty() {
let pending = self.output.copy_front(&mut chunk);
let result = self.inner_mut().write(&chunk[..pending]);
crate::wipe_bytes(&mut chunk[..pending]);
match result {
Ok(0) => {
return Err(io::Error::new(
io::ErrorKind::WriteZero,
"base64 stream decoder could not drain buffered output",
));
}
Ok(written) => {
if written > pending {
self.failed = true;
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"wrapped writer reported more bytes than provided",
));
}
self.output.discard_front(written);
}
Err(err) => return Err(err),
}
}
Ok(())
}
}
impl<W, A, const PAD: bool> Write for Decoder<W, A, PAD>
where
W: Write,
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);
}
self.drain_output()?;
if self.finalized {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"base64 stream decoder received input after finalization",
));
}
if self.finished {
self.failed = true;
return Err(trailing_input_after_padding_error());
}
let mut consumed = 0;
if self.pending_len > 0 {
let needed = 4 - self.pending_len;
if input.len() < needed {
self.pending[self.pending_len..self.pending_len + input.len()]
.copy_from_slice(input);
self.pending_len += input.len();
return Ok(input.len());
}
let mut quad = [0u8; 4];
quad[..self.pending_len].copy_from_slice(&self.pending[..self.pending_len]);
quad[self.pending_len..].copy_from_slice(&input[..needed]);
let result = self.queue_full_quad(quad);
crate::wipe_bytes(&mut quad);
if let Err(err) = result {
self.clear_pending();
return Err(err);
}
self.clear_pending();
consumed += needed;
if self.finished {
return Ok(consumed);
}
}
while input.len() - consumed >= 4 {
if self.output.available_capacity() < 3 {
return Ok(consumed);
}
let mut quad = [
input[consumed],
input[consumed + 1],
input[consumed + 2],
input[consumed + 3],
];
let mut decoded = [0u8; 3];
let written = match self.engine.decode_slice(&quad, &mut decoded) {
Ok(written) => written,
Err(err) => {
crate::wipe_bytes(&mut quad);
crate::wipe_bytes(&mut decoded);
self.failed = true;
if consumed > 0 {
return Ok(consumed);
}
return Err(decode_error_to_io(err));
}
};
let result = self.output.push_slice(&decoded[..written]);
crate::wipe_bytes(&mut quad);
crate::wipe_bytes(&mut decoded);
result?;
consumed += 4;
if written < 3 {
self.finished = true;
return Ok(consumed);
}
}
let tail = &input[consumed..];
self.pending[..tail.len()].copy_from_slice(tail);
self.pending_len = tail.len();
consumed += tail.len();
Ok(consumed)
}
fn flush(&mut self) -> io::Result<()> {
if self.failed {
return Err(stream_decoder_failed_error());
}
self.drain_output()?;
self.inner_mut().flush()
}
}
fn decode_error_to_io(err: DecodeError) -> io::Error {
io::Error::new(io::ErrorKind::InvalidInput, err)
}
fn trailing_input_after_padding_error() -> io::Error {
io::Error::new(
io::ErrorKind::InvalidInput,
"base64 decoder received trailing input after padding",
)
}
fn stream_decoder_failed_error() -> io::Error {
io::Error::new(
io::ErrorKind::InvalidInput,
"base64 stream decoder is failed after malformed input",
)
}
fn stream_encoder_failed_error() -> io::Error {
io::Error::new(
io::ErrorKind::InvalidInput,
"base64 stream encoder is failed after internal error",
)
}
pub struct DecoderReader<R, A, const PAD: bool>
where
A: Alphabet,
{
inner: Option<R>,
engine: Engine<A, PAD>,
pending: [u8; 4],
pending_len: usize,
output: OutputQueue<3>,
finished: bool,
terminal_seen: bool,
failed: bool,
}
impl<R, A, const PAD: bool> DecoderReader<R, A, PAD>
where
A: Alphabet,
{
#[must_use]
pub fn new(inner: R, engine: Engine<A, PAD>) -> Self {
Self {
inner: Some(inner),
engine,
pending: [0; 4],
pending_len: 0,
output: OutputQueue::new(),
finished: false,
terminal_seen: false,
failed: false,
}
}
#[must_use]
pub fn get_ref(&self) -> &R {
self.inner_ref()
}
pub fn get_mut(&mut self) -> &mut R {
self.inner_mut()
}
#[must_use]
pub const fn engine(&self) -> Engine<A, PAD> {
self.engine
}
#[must_use]
pub const fn is_padded(&self) -> bool {
PAD
}
#[must_use]
pub const fn pending_len(&self) -> usize {
self.pending_len
}
#[must_use]
pub const fn has_pending_input(&self) -> bool {
self.pending_len != 0
}
#[must_use]
pub const fn pending_input_needed_len(&self) -> usize {
if self.has_pending_input() {
4 - self.pending_len
} else {
0
}
}
#[must_use]
pub const fn buffered_output_len(&self) -> usize {
self.output.len()
}
#[must_use]
pub const fn buffered_output_capacity(&self) -> usize {
self.output.capacity()
}
#[must_use]
pub const fn buffered_output_remaining_capacity(&self) -> usize {
self.output.available_capacity()
}
#[must_use]
pub const fn has_buffered_output(&self) -> bool {
!self.output.is_empty()
}
#[must_use]
pub const fn has_terminal_padding(&self) -> bool {
self.terminal_seen
}
#[must_use]
pub const fn has_finished_input(&self) -> bool {
self.finished
}
#[must_use]
pub const fn is_finished(&self) -> bool {
self.finished && self.output.is_empty()
}
#[must_use]
pub const fn is_failed(&self) -> bool {
self.failed
}
#[must_use]
pub const fn can_into_inner(&self) -> bool {
!self.is_failed() && self.is_finished()
}
#[must_use]
pub fn into_inner(mut self) -> R {
self.take_inner()
}
#[allow(clippy::result_large_err)]
pub fn try_into_inner(mut self) -> Result<R, Self> {
if !self.can_into_inner() {
return Err(self);
}
Ok(self.take_inner())
}
fn inner_ref(&self) -> &R {
match &self.inner {
Some(inner) => inner,
None => unreachable!("stream decoder reader inner reader was already taken"),
}
}
fn inner_mut(&mut self) -> &mut R {
match &mut self.inner {
Some(inner) => inner,
None => unreachable!("stream decoder reader inner reader was already taken"),
}
}
fn take_inner(&mut self) -> R {
match self.inner.take() {
Some(inner) => inner,
None => unreachable!("stream decoder reader inner reader was already taken"),
}
}
fn clear_pending(&mut self) {
crate::wipe_bytes(&mut self.pending);
self.pending_len = 0;
}
}
impl<R, A, const PAD: bool> Drop for DecoderReader<R, A, PAD>
where
A: Alphabet,
{
fn drop(&mut self) {
self.clear_pending();
self.output.clear_all();
}
}
impl<R, A, const PAD: bool> core::fmt::Debug for DecoderReader<R, A, PAD>
where
A: Alphabet,
{
fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
formatter
.debug_struct("DecoderReader")
.field("inner", &redacted_inner_state(self.inner.is_some()))
.field("engine", &self.engine)
.field("pending", &"<redacted>")
.field("pending_len", &self.pending_len)
.field("pending_input_needed_len", &self.pending_input_needed_len())
.field("buffered_output_len", &self.output.len())
.field("buffered_output_capacity", &self.output.capacity())
.field(
"buffered_output_remaining_capacity",
&self.output.available_capacity(),
)
.field("can_into_inner", &self.can_into_inner())
.field("finished", &self.finished)
.field("terminal_padding", &self.terminal_seen)
.field("failed", &self.failed)
.finish()
}
}
impl<R, A, const PAD: bool> Read for DecoderReader<R, A, PAD>
where
R: Read,
A: Alphabet,
{
fn read(&mut self, output: &mut [u8]) -> io::Result<usize> {
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()?;
}
Ok(self.output.pop_slice(output))
}
}
impl<R, A, const PAD: bool> DecoderReader<R, A, PAD>
where
R: Read,
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(());
}
let mut input = [0u8; 4];
let available = 4 - self.pending_len;
let read = match self.inner_mut().read(&mut input[..available]) {
Ok(read) => read,
Err(err) => {
crate::wipe_bytes(&mut input);
return Err(err);
}
};
if read == 0 {
crate::wipe_bytes(&mut input);
self.finished = true;
self.push_final_pending()?;
return Ok(());
}
self.pending[self.pending_len..self.pending_len + read].copy_from_slice(&input[..read]);
crate::wipe_bytes(&mut input);
self.pending_len += read;
if self.pending_len < 4 {
return Ok(());
}
let mut quad = self.pending;
self.clear_pending();
let result = self.push_decoded(&quad);
crate::wipe_bytes(&mut quad);
result?;
if self.terminal_seen {
self.finished = true;
}
Ok(())
}
fn push_final_pending(&mut self) -> io::Result<()> {
if self.pending_len == 0 {
return Ok(());
}
let mut pending = [0u8; 4];
pending[..self.pending_len].copy_from_slice(&self.pending[..self.pending_len]);
let pending_len = self.pending_len;
self.clear_pending();
let result = self.push_decoded(&pending[..pending_len]);
crate::wipe_bytes(&mut pending);
result
}
fn push_decoded(&mut self, input: &[u8]) -> io::Result<()> {
let mut decoded = [0u8; 3];
let written = match self.engine.decode_slice(input, &mut decoded) {
Ok(written) => written,
Err(err) => {
crate::wipe_bytes(&mut decoded);
self.failed = true;
return Err(decode_error_to_io(err));
}
};
let result = self.output.push_slice(&decoded[..written]);
crate::wipe_bytes(&mut decoded);
result?;
if input.len() == 4 && written < 3 {
self.terminal_seen = true;
}
Ok(())
}
}
pub struct EncoderReader<R, A, const PAD: bool>
where
A: Alphabet,
{
inner: Option<R>,
engine: Engine<A, PAD>,
pending: [u8; 2],
pending_len: usize,
output: OutputQueue<1024>,
finished: bool,
failed: bool,
}
impl<R, A, const PAD: bool> EncoderReader<R, A, PAD>
where
A: Alphabet,
{
#[must_use]
pub fn new(inner: R, engine: Engine<A, PAD>) -> Self {
Self {
inner: Some(inner),
engine,
pending: [0; 2],
pending_len: 0,
output: OutputQueue::new(),
finished: false,
failed: false,
}
}
#[must_use]
pub fn get_ref(&self) -> &R {
self.inner_ref()
}
pub fn get_mut(&mut self) -> &mut R {
self.inner_mut()
}
#[must_use]
pub const fn engine(&self) -> Engine<A, PAD> {
self.engine
}
#[must_use]
pub const fn is_padded(&self) -> bool {
PAD
}
#[must_use]
pub const fn pending_len(&self) -> usize {
self.pending_len
}
#[must_use]
pub const fn has_pending_input(&self) -> bool {
self.pending_len != 0
}
#[must_use]
pub const fn pending_input_needed_len(&self) -> usize {
if self.has_pending_input() {
3 - self.pending_len
} else {
0
}
}
#[must_use]
pub const fn buffered_output_len(&self) -> usize {
self.output.len()
}
#[must_use]
pub const fn buffered_output_capacity(&self) -> usize {
self.output.capacity()
}
#[must_use]
pub const fn buffered_output_remaining_capacity(&self) -> usize {
self.output.available_capacity()
}
#[must_use]
pub const fn has_buffered_output(&self) -> bool {
!self.output.is_empty()
}
#[must_use]
pub const fn has_finished_input(&self) -> bool {
self.finished
}
#[must_use]
pub const fn is_finished(&self) -> bool {
self.finished && self.output.is_empty()
}
#[must_use]
pub const fn is_failed(&self) -> bool {
self.failed
}
#[must_use]
pub const fn can_into_inner(&self) -> bool {
self.is_finished() && !self.failed
}
#[must_use]
pub fn into_inner(mut self) -> R {
self.take_inner()
}
#[allow(clippy::result_large_err)]
pub fn try_into_inner(mut self) -> Result<R, Self> {
if !self.can_into_inner() {
return Err(self);
}
Ok(self.take_inner())
}
fn inner_ref(&self) -> &R {
match &self.inner {
Some(inner) => inner,
None => unreachable!("stream encoder reader inner reader was already taken"),
}
}
fn inner_mut(&mut self) -> &mut R {
match &mut self.inner {
Some(inner) => inner,
None => unreachable!("stream encoder reader inner reader was already taken"),
}
}
fn take_inner(&mut self) -> R {
match self.inner.take() {
Some(inner) => inner,
None => unreachable!("stream encoder reader inner reader was already taken"),
}
}
fn clear_pending(&mut self) {
crate::wipe_bytes(&mut self.pending);
self.pending_len = 0;
}
}
impl<R, A, const PAD: bool> Drop for EncoderReader<R, A, PAD>
where
A: Alphabet,
{
fn drop(&mut self) {
self.clear_pending();
self.output.clear_all();
}
}
impl<R, A, const PAD: bool> core::fmt::Debug for EncoderReader<R, A, PAD>
where
A: Alphabet,
{
fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
formatter
.debug_struct("EncoderReader")
.field("inner", &redacted_inner_state(self.inner.is_some()))
.field("engine", &self.engine)
.field("pending", &"<redacted>")
.field("pending_len", &self.pending_len)
.field("pending_input_needed_len", &self.pending_input_needed_len())
.field("buffered_output_len", &self.output.len())
.field("buffered_output_capacity", &self.output.capacity())
.field(
"buffered_output_remaining_capacity",
&self.output.available_capacity(),
)
.field("can_into_inner", &self.can_into_inner())
.field("finished", &self.finished)
.field("failed", &self.failed)
.finish()
}
}
impl<R, A, const PAD: bool> Read for EncoderReader<R, A, PAD>
where
R: Read,
A: Alphabet,
{
fn read(&mut self, output: &mut [u8]) -> io::Result<usize> {
if self.failed {
return Err(stream_encoder_failed_error());
}
if output.is_empty() {
return Ok(0);
}
while self.output.is_empty() && !self.finished {
self.fill_output()?;
}
Ok(self.output.pop_slice(output))
}
}
impl<R, A, const PAD: bool> EncoderReader<R, A, PAD>
where
R: Read,
A: Alphabet,
{
fn fill_output(&mut self) -> io::Result<()> {
let mut input = [0u8; 768];
let read = match self.inner_mut().read(&mut input) {
Ok(read) => read,
Err(err) => {
crate::wipe_bytes(&mut input);
return Err(err);
}
};
if read == 0 {
crate::wipe_bytes(&mut input);
self.finished = true;
if let Err(err) = self.push_final_pending() {
self.failed = true;
return Err(err);
}
return Ok(());
}
let mut consumed = 0;
if self.pending_len > 0 {
let needed = 3 - self.pending_len;
if read < needed {
self.pending[self.pending_len..self.pending_len + read]
.copy_from_slice(&input[..read]);
self.pending_len += read;
crate::wipe_bytes(&mut input);
return Ok(());
}
let mut chunk = [0u8; 3];
chunk[..self.pending_len].copy_from_slice(&self.pending[..self.pending_len]);
chunk[self.pending_len..].copy_from_slice(&input[..needed]);
let result = self.push_encoded(&chunk);
crate::wipe_bytes(&mut chunk);
if let Err(err) = result {
crate::wipe_bytes(&mut input);
self.failed = true;
return Err(err);
}
self.clear_pending();
consumed += needed;
}
let remaining = &input[consumed..read];
let full_len = remaining.len() / 3 * 3;
let tail_len = remaining.len() - full_len;
let mut tail = [0u8; 2];
tail[..tail_len].copy_from_slice(&remaining[full_len..]);
let result = if full_len > 0 {
self.push_encoded(&remaining[..full_len])
} else {
Ok(())
};
crate::wipe_bytes(&mut input);
if let Err(err) = result {
crate::wipe_bytes(&mut tail);
self.failed = true;
return Err(err);
}
self.pending[..tail_len].copy_from_slice(&tail[..tail_len]);
crate::wipe_bytes(&mut tail);
self.pending_len = tail_len;
Ok(())
}
fn push_final_pending(&mut self) -> io::Result<()> {
if self.pending_len == 0 {
return Ok(());
}
let mut pending = [0u8; 2];
pending[..self.pending_len].copy_from_slice(&self.pending[..self.pending_len]);
let pending_len = self.pending_len;
self.clear_pending();
let result = self.push_encoded(&pending[..pending_len]);
crate::wipe_bytes(&mut pending);
result
}
fn push_encoded(&mut self, input: &[u8]) -> io::Result<()> {
let mut encoded = [0u8; 1024];
let written = match self.engine.encode_slice(input, &mut encoded) {
Ok(written) => written,
Err(err) => {
crate::wipe_bytes(&mut encoded);
return Err(encode_error_to_io(err));
}
};
let result = self.output.push_slice(&encoded[..written]);
crate::wipe_bytes(&mut encoded);
result
}
}
const fn redacted_inner_state(present: bool) -> &'static str {
if present { "<present>" } else { "<taken>" }
}