use std::collections::VecDeque;
use std::io;
use puffin::codec;
use crate::tls::rustls::msgs::message::{MessageError, OpaqueMessage};
pub struct MessageDeframer {
pub frames: VecDeque<OpaqueMessage>,
pub desynced: bool,
buf: Box<[u8; OpaqueMessage::MAX_WIRE_SIZE]>,
used: usize,
}
enum BufferContents {
Invalid,
Partial,
Valid,
}
impl Default for MessageDeframer {
fn default() -> Self {
Self::new()
}
}
impl MessageDeframer {
pub fn new() -> Self {
Self {
frames: VecDeque::new(),
desynced: false,
buf: Box::new([0u8; OpaqueMessage::MAX_WIRE_SIZE]),
used: 0,
}
}
pub fn read(&mut self, rd: &mut dyn io::Read) -> io::Result<usize> {
debug_assert!(self.used <= OpaqueMessage::MAX_WIRE_SIZE);
let new_bytes = rd.read(&mut self.buf[self.used..])?;
self.used += new_bytes;
loop {
match self.try_deframe_one() {
BufferContents::Invalid => {
self.desynced = true;
break;
}
BufferContents::Valid => continue,
BufferContents::Partial => break,
}
}
Ok(new_bytes)
}
pub fn has_pending(&self) -> bool {
!self.frames.is_empty() || self.used > 0
}
fn try_deframe_one(&mut self) -> BufferContents {
let mut rd = codec::Reader::init(&self.buf[..self.used]);
match OpaqueMessage::read(&mut rd) {
Ok(m) => {
let used = rd.used();
self.frames.push_back(m);
self.buf_consume(used);
BufferContents::Valid
}
Err(MessageError::TooShortForHeader) | Err(MessageError::TooShortForLength) => {
BufferContents::Partial
}
Err(_) => BufferContents::Invalid,
}
}
#[allow(clippy::comparison_chain)]
fn buf_consume(&mut self, taken: usize) {
if taken < self.used {
self.buf.copy_within(taken..self.used, 0);
self.used -= taken;
} else if taken == self.used {
self.used = 0;
}
}
}
#[cfg(test)]
mod tests {
use std::io;
use super::MessageDeframer;
use crate::tls::rustls::msgs;
use crate::tls::rustls::msgs::message::Message;
const FIRST_MESSAGE: &[u8] = include_bytes!("../testdata/deframer-test.1.bin");
const SECOND_MESSAGE: &[u8] = include_bytes!("../testdata/deframer-test.2.bin");
const EMPTY_APPLICATIONDATA_MESSAGE: &[u8] =
include_bytes!("../testdata/deframer-empty-applicationdata.bin");
const INVALID_EMPTY_MESSAGE: &[u8] = include_bytes!("../testdata/deframer-invalid-empty.bin");
const INVALID_CONTENTTYPE_MESSAGE: &[u8] =
include_bytes!("../testdata/deframer-invalid-contenttype.bin");
const INVALID_VERSION_MESSAGE: &[u8] =
include_bytes!("../testdata/deframer-invalid-version.bin");
const INVALID_LENGTH_MESSAGE: &[u8] = include_bytes!("../testdata/deframer-invalid-length.bin");
struct ByteRead<'a> {
buf: &'a [u8],
offs: usize,
}
impl<'a> ByteRead<'a> {
fn new(bytes: &'a [u8]) -> Self {
ByteRead {
buf: bytes,
offs: 0,
}
}
}
impl<'a> io::Read for ByteRead<'a> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let mut len = 0;
while len < buf.len() && len < self.buf.len() - self.offs {
buf[len] = self.buf[self.offs + len];
len += 1;
}
self.offs += len;
Ok(len)
}
}
fn input_bytes(d: &mut MessageDeframer, bytes: &[u8]) -> io::Result<usize> {
let mut rd = ByteRead::new(bytes);
d.read(&mut rd)
}
fn input_bytes_concat(
d: &mut MessageDeframer,
bytes1: &[u8],
bytes2: &[u8],
) -> io::Result<usize> {
let mut bytes = vec![0u8; bytes1.len() + bytes2.len()];
bytes[..bytes1.len()].clone_from_slice(bytes1);
bytes[bytes1.len()..].clone_from_slice(bytes2);
let mut rd = ByteRead::new(&bytes);
d.read(&mut rd)
}
struct ErrorRead {
error: Option<io::Error>,
}
impl ErrorRead {
fn new(error: io::Error) -> Self {
Self { error: Some(error) }
}
}
impl io::Read for ErrorRead {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
for (i, b) in buf.iter_mut().enumerate() {
*b = i as u8;
}
let error = self.error.take().unwrap();
Err(error)
}
}
fn input_error(d: &mut MessageDeframer) {
let error = io::Error::from(io::ErrorKind::TimedOut);
let mut rd = ErrorRead::new(error);
d.read(&mut rd).expect_err("error not propagated");
}
fn input_whole_incremental(d: &mut MessageDeframer, bytes: &[u8]) {
let frames_before = d.frames.len();
for i in 0..bytes.len() {
assert_len(1, input_bytes(d, &bytes[i..i + 1]));
assert!(d.has_pending());
if i < bytes.len() - 1 {
assert_eq!(frames_before, d.frames.len());
}
}
assert_eq!(frames_before + 1, d.frames.len());
}
fn assert_len(want: usize, got: io::Result<usize>) {
if let Ok(gotval) = got {
assert_eq!(gotval, want);
} else {
panic!("read failed, expected {:?} bytes", want);
}
}
fn pop_first(d: &mut MessageDeframer) {
let m = d.frames.pop_front().unwrap();
assert_eq!(m.typ, msgs::enums::ContentType::Handshake);
Message::try_from(m.into_plain_message()).unwrap();
}
fn pop_second(d: &mut MessageDeframer) {
let m = d.frames.pop_front().unwrap();
assert_eq!(m.typ, msgs::enums::ContentType::Alert);
Message::try_from(m.into_plain_message()).unwrap();
}
#[test_log::test]
fn check_incremental() {
let mut d = MessageDeframer::new();
assert!(!d.has_pending());
input_whole_incremental(&mut d, FIRST_MESSAGE);
assert!(d.has_pending());
assert_eq!(1, d.frames.len());
pop_first(&mut d);
assert!(!d.has_pending());
assert!(!d.desynced);
}
#[test_log::test]
fn check_incremental_2() {
let mut d = MessageDeframer::new();
assert!(!d.has_pending());
input_whole_incremental(&mut d, FIRST_MESSAGE);
assert!(d.has_pending());
input_whole_incremental(&mut d, SECOND_MESSAGE);
assert!(d.has_pending());
assert_eq!(2, d.frames.len());
pop_first(&mut d);
assert!(d.has_pending());
pop_second(&mut d);
assert!(!d.has_pending());
assert!(!d.desynced);
}
#[test_log::test]
fn check_whole() {
let mut d = MessageDeframer::new();
assert!(!d.has_pending());
assert_len(FIRST_MESSAGE.len(), input_bytes(&mut d, FIRST_MESSAGE));
assert!(d.has_pending());
assert_eq!(d.frames.len(), 1);
pop_first(&mut d);
assert!(!d.has_pending());
assert!(!d.desynced);
}
#[test_log::test]
fn check_whole_2() {
let mut d = MessageDeframer::new();
assert!(!d.has_pending());
assert_len(FIRST_MESSAGE.len(), input_bytes(&mut d, FIRST_MESSAGE));
assert_len(SECOND_MESSAGE.len(), input_bytes(&mut d, SECOND_MESSAGE));
assert_eq!(d.frames.len(), 2);
pop_first(&mut d);
pop_second(&mut d);
assert!(!d.has_pending());
assert!(!d.desynced);
}
#[test_log::test]
fn test_two_in_one_read() {
let mut d = MessageDeframer::new();
assert!(!d.has_pending());
assert_len(
FIRST_MESSAGE.len() + SECOND_MESSAGE.len(),
input_bytes_concat(&mut d, FIRST_MESSAGE, SECOND_MESSAGE),
);
assert_eq!(d.frames.len(), 2);
pop_first(&mut d);
pop_second(&mut d);
assert!(!d.has_pending());
assert!(!d.desynced);
}
#[test_log::test]
fn test_two_in_one_read_shortest_first() {
let mut d = MessageDeframer::new();
assert!(!d.has_pending());
assert_len(
FIRST_MESSAGE.len() + SECOND_MESSAGE.len(),
input_bytes_concat(&mut d, SECOND_MESSAGE, FIRST_MESSAGE),
);
assert_eq!(d.frames.len(), 2);
pop_second(&mut d);
pop_first(&mut d);
assert!(!d.has_pending());
assert!(!d.desynced);
}
#[test_log::test]
fn test_incremental_with_nonfatal_read_error() {
let mut d = MessageDeframer::new();
assert_len(3, input_bytes(&mut d, &FIRST_MESSAGE[..3]));
input_error(&mut d);
assert_len(
FIRST_MESSAGE.len() - 3,
input_bytes(&mut d, &FIRST_MESSAGE[3..]),
);
assert_eq!(d.frames.len(), 1);
pop_first(&mut d);
assert!(!d.has_pending());
assert!(!d.desynced);
}
#[test_log::test]
fn test_invalid_contenttype_errors() {
let mut d = MessageDeframer::new();
assert_len(
INVALID_CONTENTTYPE_MESSAGE.len(),
input_bytes(&mut d, INVALID_CONTENTTYPE_MESSAGE),
);
assert!(d.desynced);
}
#[test_log::test]
fn test_invalid_version_errors() {
let mut d = MessageDeframer::new();
assert_len(
INVALID_VERSION_MESSAGE.len(),
input_bytes(&mut d, INVALID_VERSION_MESSAGE),
);
assert!(d.desynced);
}
#[test_log::test]
fn test_invalid_length_errors() {
let mut d = MessageDeframer::new();
assert_len(
INVALID_LENGTH_MESSAGE.len(),
input_bytes(&mut d, INVALID_LENGTH_MESSAGE),
);
assert!(d.desynced);
}
#[test_log::test]
fn test_empty_applicationdata() {
let mut d = MessageDeframer::new();
assert_len(
EMPTY_APPLICATIONDATA_MESSAGE.len(),
input_bytes(&mut d, EMPTY_APPLICATIONDATA_MESSAGE),
);
let m = d.frames.pop_front().unwrap();
assert_eq!(m.typ, msgs::enums::ContentType::ApplicationData);
assert_eq!(m.payload.0.len(), 0);
assert!(!d.has_pending());
assert!(!d.desynced);
}
#[test_log::test]
fn test_invalid_empty_errors() {
let mut d = MessageDeframer::new();
assert_len(
INVALID_EMPTY_MESSAGE.len(),
input_bytes(&mut d, INVALID_EMPTY_MESSAGE),
);
assert!(d.desynced);
}
}