permessage-deflate

Author: willrnch

Cargo.lock

@@ -54,6 +54,7 @@ utf-8 = "0.7.5"
 rand = "0.8.4"
 thiserror = "1.0.40"
 bytes = "1.5.0"
+miniz_oxide = "0.8.9"
 
 # Axum integration
 axum-core = { version = "0.5.0", optional = true }

Cargo.toml

@@ -1,7 +1,7 @@
 AUTOBAHN_TESTSUITE_DOCKER := crossbario/autobahn-testsuite:0.8.2@sha256:5d4ba3aa7d6ab2fdbf6606f3f4ecbe4b66f205ce1cbc176d6cdf650157e52242
 
 build-server:
-	sudo cargo build --release --example echo_server --features "upgrade"
+	cargo build --release --example echo_server --features "upgrade"
 
 run-server: build-server
 	echo ${PWD}
@@ -18,7 +18,7 @@ run-server: build-server
 	../target/release/examples/echo_server
 
 build-client:
-	sudo cargo build --release --example autobahn_client --features "upgrade"
+	cargo build --release --example autobahn_client --features "upgrade"
 
 run-client: build-client
 	echo ${PWD}
@@ -34,4 +34,4 @@ run-client: build-client
 	sleep 5
 	../target/release/examples/autobahn_client
 
-.PHONY: build-server run-server build-client run-client
+.PHONY: build-server run-server build-client run-client

autobahn/Makefile

@@ -1 +1 @@
-1.76.0
+1.91.1

rust-toolchain

@@ -34,6 +34,8 @@ pub enum WebSocketError {
   InvalidSecWebsocketVersion,
   #[error("Invalid value")]
   InvalidValue,
+  #[error("Invalid encoding")]
+  InvalidEncoding,
   #[error("Sec-WebSocket-Key header is missing")]
   MissingSecWebSocketKey,
   #[error(transparent)]

src/error.rs

@@ -222,7 +222,7 @@ impl Fragments {
           if self.fragments.is_some() {
             return Err(WebSocketError::InvalidFragment);
           }
-          return Ok(Some(Frame::new(true, frame.opcode, None, frame.payload)));
+          return Ok(Some(Frame::new(true, frame.opcode, None, frame.payload, frame.compressed)));
         } else {
           self.fragments = match frame.opcode {
             OpCode::Text => match utf8::decode(&frame.payload) {
@@ -292,6 +292,7 @@ impl Fragments {
               self.opcode,
               None,
               self.fragments.take().unwrap().take_buffer().into(),
+              false,
             )));
           }
         }
@@ -303,6 +304,7 @@ impl Fragments {
               self.opcode,
               None,
               self.fragments.take().unwrap().take_buffer().into(),
+              false,
             )));
           }
         }

src/fragment.rs

@@ -14,11 +14,16 @@
 
 use tokio::io::AsyncWriteExt;
 
+use miniz_oxide::{MZFlush, MZStatus};
+use miniz_oxide::inflate::stream::{InflateState, inflate};
+
 use bytes::BytesMut;
 use core::ops::Deref;
 
 use crate::WebSocketError;
 
+const TRAILER: [u8; 4] = [0x00, 0x00, 0xff, 0xff];
+
 macro_rules! repr_u8 {
     ($(#[$meta:meta])* $vis:vis enum $name:ident {
       $($(#[$vmeta:meta])* $vname:ident $(= $val:expr)?,)*
@@ -136,6 +141,8 @@ pub struct Frame<'f> {
   mask: Option<[u8; 4]>,
   /// The payload of the frame.
   pub payload: Payload<'f>,
+  /// Is the frame payload compressed
+  pub compressed: bool,
 }
 
 const MAX_HEAD_SIZE: usize = 16;
@@ -147,12 +154,14 @@ impl<'f> Frame<'f> {
     opcode: OpCode,
     mask: Option<[u8; 4]>,
     payload: Payload<'f>,
+    compressed: bool,
   ) -> Self {
     Self {
       fin,
       opcode,
       mask,
       payload,
+      compressed,
     }
   }
 
@@ -167,6 +176,7 @@ impl<'f> Frame<'f> {
       opcode: OpCode::Text,
       mask: None,
       payload,
+      compressed: false,
     }
   }
 
@@ -179,6 +189,7 @@ impl<'f> Frame<'f> {
       opcode: OpCode::Binary,
       mask: None,
       payload,
+      compressed: false,
     }
   }
 
@@ -197,6 +208,7 @@ impl<'f> Frame<'f> {
       opcode: OpCode::Close,
       mask: None,
       payload: payload.into(),
+      compressed: false,
     }
   }
 
@@ -211,6 +223,7 @@ impl<'f> Frame<'f> {
       opcode: OpCode::Close,
       mask: None,
       payload,
+      compressed: false,
     }
   }
 
@@ -223,6 +236,7 @@ impl<'f> Frame<'f> {
       opcode: OpCode::Pong,
       mask: None,
       payload,
+      compressed: false,
     }
   }
 
@@ -334,6 +348,33 @@ impl<'f> Frame<'f> {
     buf[size..size + len].copy_from_slice(&self.payload);
     &buf[..size + len]
   }
+
+  pub fn inflate(&self, state: &mut InflateState) -> Result<Self, WebSocketError>
+  {
+      let payload = [self.payload.to_vec().as_slice(), &TRAILER].concat();
+
+      let max_output_size = usize::max_value();
+      let mut out: Vec<u8> = vec![0; payload.len().saturating_mul(2).min(max_output_size)];
+
+      let res = inflate(state, &payload, &mut out, MZFlush::None);
+
+      if res.status != Ok(MZStatus::Ok) {
+        return Err(WebSocketError::InvalidEncoding);
+      }
+
+      out.truncate(res.bytes_written);
+
+      let payload = Payload::Owned(out);
+
+      Ok(Self {
+        fin: self.fin,
+        opcode: self.opcode,
+        mask: self.mask,
+        payload,
+        compressed: false
+      })
+  }
+
 }
 
 repr_u8! {

src/frame.rs

@@ -175,6 +175,9 @@ use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWrite;
 use tokio::io::AsyncWriteExt;
 
+use miniz_oxide::DataFormat;
+use miniz_oxide::inflate::stream::InflateState;
+
 pub use crate::close::CloseCode;
 pub use crate::error::WebSocketError;
 pub use crate::fragment::FragmentCollector;
@@ -208,6 +211,8 @@ pub(crate) struct ReadHalf {
   writev_threshold: usize,
   max_message_size: usize,
   buffer: BytesMut,
+
+  state: InflateState,
 }
 
 #[cfg(feature = "unstable-split")]
@@ -364,6 +369,7 @@ pub struct WebSocket<S> {
   stream: S,
   write_half: WriteHalf,
   read_half: ReadHalf,
+
 }
 
 impl<'f, S> WebSocket<S> {
@@ -577,6 +583,8 @@ impl ReadHalf {
   pub fn after_handshake(role: Role) -> Self {
     let buffer = BytesMut::with_capacity(8192);
 
+    let state = InflateState::new(DataFormat::Raw);
+
     Self {
       role,
       auto_apply_mask: true,
@@ -585,6 +593,7 @@ impl ReadHalf {
       writev_threshold: 1024,
       max_message_size: 64 << 20,
       buffer,
+      state,
     }
   }
 
@@ -610,6 +619,13 @@ impl ReadHalf {
       frame.unmask()
     };
 
+    if frame.compressed {
+        frame = match frame.inflate(&mut self.state) {
+            Ok(frame) => frame,
+            Err(e) => return (Err(e), None),
+        }
+    }
+
     match frame.opcode {
       OpCode::Close if self.auto_close => {
         match frame.payload.len() {
@@ -681,7 +697,11 @@ impl ReadHalf {
     let rsv2 = self.buffer[0] & 0b00100000 != 0;
     let rsv3 = self.buffer[0] & 0b00010000 != 0;
 
-    if rsv1 || rsv2 || rsv3 {
+    let mut compressed = false;
+
+    if rsv1 && !rsv2 && !rsv3 {
+      compressed = true;
+    } else if rsv1 || rsv2 || rsv3 {
       return Err(WebSocketError::ReservedBitsNotZero);
     }
 
@@ -744,7 +764,7 @@ impl ReadHalf {
 
     // if we read too much it will stay in the buffer, for the next call to this method
     let payload = self.buffer.split_to(payload_len);
-    let frame = Frame::new(fin, opcode, mask, Payload::Bytes(payload));
+    let frame = Frame::new(fin, opcode, mask, Payload::Bytes(payload), compressed);
     Ok(frame)
   }
 }