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

@@ -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

@@ -175,6 +175,9 @@ use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWrite;
 use tokio::io::AsyncWriteExt;
 
+use miniz_oxide::{DataFormat, MZFlush};
+use miniz_oxide::inflate::stream::{InflateState, inflate};
+
 pub use crate::close::CloseCode;
 pub use crate::error::WebSocketError;
 pub use crate::fragment::FragmentCollector;
@@ -681,7 +684,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);
     }
 
@@ -743,8 +750,13 @@ 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 mut payload = self.buffer.split_to(payload_len);
+    if compressed {
+      payload = BytesMut::from(inflate_payload(&payload.to_vec())?.as_slice());
+    }
+
     let frame = Frame::new(fin, opcode, mask, Payload::Bytes(payload));
+
     Ok(frame)
   }
 }
@@ -820,3 +832,25 @@ mod tests {
     assert_unsync::<WebSocket<tokio::net::TcpStream>>();
   };
 }
+
+fn inflate_payload(
+  payload: &Vec<u8>
+) -> Result<Vec<u8>, WebSocketError>
+{
+  let max_output_size = usize::max_value();
+  let mut out: Vec<u8> = vec![0; payload.len().saturating_mul(2).min(max_output_size)];
+  let mut state = InflateState::new_boxed(DataFormat::Raw);
+
+  let payload = [payload.as_slice(), [0x00, 0x00, 0xff, 0xff].as_slice()].concat();
+  let res = inflate(&mut state, &payload, &mut out, MZFlush::Partial);
+
+  match res.status {
+    Ok(_) => {
+      out.truncate(res.bytes_written);
+      Ok(out)
+    }
+    Err(_) => {
+      Err(WebSocketError::InvalidEncoding)
+    }
+  }
+}