diff --git a/src/websocket/byte_adapter.rs b/src/websocket/byte_adapter.rs index 91452b8..319b23e 100644 --- a/src/websocket/byte_adapter.rs +++ b/src/websocket/byte_adapter.rs @@ -1110,7 +1110,7 @@ mod tests { /// An 8-byte bogus chunk header claiming `MAX_CHUNK_LEN + 1` /// payload bytes (WS-04/HY-09 acceptance: the parser must fail the /// stream loudly instead of silently waiting for ~4 GiB). - fn oversize_header() -> Vec { + pub(crate) fn oversize_header() -> Vec { let mut header = vec![0u8; 8]; header[4..8].copy_from_slice(&(MAX_CHUNK_LEN + 1).to_be_bytes()); header @@ -1745,3 +1745,170 @@ mod tests { .expect("peer close accepted"); } } + +/// WS-19: the axum-flavor `AxumFraming` arms have no direct unit test — +/// the cap-trip and text→1002 closes are asserted on tungstenite only +/// (via the shared generic pumps). These drive the same generic pumps +/// with `AxumFraming` over a fake axum `WebSocket` sink/stream pair +/// (no server needed): a `mpsc`-backed sink/stream pair standing in +/// for the split halves of `axum::extract::ws::WebSocket`. +#[cfg(test)] +mod axum_framing_tests { + use super::*; + use futures::channel::mpsc as fut_mpsc; + + /// In-process stand-in for the split halves of an axum + /// `WebSocket`: messages flow stream→`rx` and `tx`→sink, so the + /// generic pumps run against `AxumFraming` unchanged. + struct AxumFakeSocket { + sink_tx: futures_mpsc::Sender, + stream_rx: + std::pin::Pin> + Send>>, + } + + impl futures::Stream for AxumFakeSocket { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.stream_rx.as_mut().poll_next(cx) + } + } + + impl futures::Sink for AxumFakeSocket { + type Error = (); + + fn poll_ready( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + Pin::new(&mut self.sink_tx).poll_ready(cx).map_err(|_| ()) + } + + fn start_send(mut self: Pin<&mut Self>, item: AxumMessage) -> Result<(), Self::Error> { + self.sink_tx.start_send(item).map_err(|_| ()) + } + + fn poll_flush( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + futures::Sink::poll_flush(Pin::new(&mut self.sink_tx), cx).map_err(|_| ()) + } + + fn poll_close( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + futures::Sink::poll_close(Pin::new(&mut self.sink_tx), cx).map_err(|_| ()) + } + } + + type AxumPairParts = ( + futures::stream::SplitSink, + futures::stream::SplitStream, + futures_mpsc::Receiver, + fut_mpsc::Sender>, + ); + + fn axum_pair() -> AxumPairParts { + let (outbound_tx, outbound_rx) = fut_mpsc::channel::(4); + let (inbound_tx, inbound_rx) = fut_mpsc::channel::>(4); + let socket = AxumFakeSocket { + sink_tx: outbound_tx, + stream_rx: Box::pin(inbound_rx), + }; + let (sink, stream) = socket.split(); + (sink, stream, outbound_rx, inbound_tx) + } + + /// WS-19 mirror over `AxumFraming`: a text message on the read + /// side triggers the 1002 protocol-error close carrying the text + /// reason (the same generic read-pump arm the tungstenite tests + /// exercise — asserted here on the axum message types). + #[tokio::test] + async fn axum_framing_text_message_requests_protocol_close_with_reason() { + let (_sink, stream, _outbound_rx, mut inbound_tx) = axum_pair(); + + let (write_tx_for_read, mut write_rx_for_read) = fut_mpsc::channel::(WRITE_SLOTS); + let read_task = tokio::spawn(run_read_pump::( + stream, + mpsc::channel::>(READ_SLOTS).0, + write_tx_for_read, + make_write_error_slot().0, + None, + || {}, + )); + + inbound_tx + .send(Ok(AxumMessage::Text("text frame".into()))) + .await + .expect("inbound message accepted"); + + let close = tokio::time::timeout(std::time::Duration::from_secs(5), async { + while let Some(msg) = write_rx_for_read.next().await { + if let WriteMsg::CloseWith(code, reason) = msg { + return ::close_message(code, reason); + } + } + AxumMessage::Text("queue closed".into()) + }) + .await + .expect("close requested"); + + let AxumMessage::Close(Some(frame)) = close else { + panic!("expected a close frame, got {close:?}"); + }; + assert_eq!(frame.code, WS_PROTOCOL_ERROR, "text closes with 1002"); + assert_eq!( + frame.reason, "text messages not supported", + "close reason names the text cause" + ); + read_task.abort(); + } + + /// WS-19 mirror of the over-cap close on the axum flavor: a + /// header claiming above-`MAX_CHUNK_LEN` payload fails the pump — + /// the peer sees the 1011 close naming the violation and the + /// pump's error slot carries the reason. + #[tokio::test] + async fn axum_framing_cap_trip_fails_the_pump_with_the_internal_close() { + let (sink, _stream, mut outbound_rx, _inbound_tx) = axum_pair(); + let (slot, _rx) = make_write_error_slot(); + + let write_task = tokio::spawn(run_write_pump::( + sink, + { + let (mut tx, rx) = fut_mpsc::channel::(WRITE_SLOTS); + tx.send(WriteMsg::Bytes(tests::oversize_header())) + .await + .expect("queue write accepted"); + rx + }, + slot, + None, + )); + + let close = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + match outbound_rx.next().await { + Some(m @ AxumMessage::Close(_)) => return m, + Some(_) => continue, + None => return AxumMessage::Text("stream ended".into()), + } + } + }) + .await + .expect("close observed"); + + let AxumMessage::Close(Some(frame)) = close else { + panic!("expected a close frame, got {close:?}"); + }; + assert_eq!(frame.code, WS_INTERNAL_ERROR, "cap trip closes with 1011"); + assert!( + frame.reason.contains("MAX_CHUNK_LEN"), + "close reason names the violation: {}", + frame.reason + ); + write_task.abort(); + } +}