diff --git a/fuzz/shared/src/ws_chunk_framing.rs b/fuzz/shared/src/ws_chunk_framing.rs index 31e9794..a6b61e4 100644 --- a/fuzz/shared/src/ws_chunk_framing.rs +++ b/fuzz/shared/src/ws_chunk_framing.rs @@ -102,16 +102,36 @@ fn split_walk(data: &[u8]) { match split.feed(piece) { Walk::Emitted(messages) => emitted.extend(messages.into_iter().flatten()), Walk::Incomplete => {} - Walk::Oversize | Walk::PendingOverCap => { + Walk::Oversize(pre_pieces) | Walk::PendingOverCap(pre_pieces) => { // A split-drive failure must be reachable on the // whole drive too (the walk is boundary-insensitive: // failures depend on the declared header at the - // accumulation position, not on the span edges). + // accumulation position, not on the span edges). On + // the split drive the pre-failure frames drained on + // EARLIER feeds (already in `emitted`); the whole + // drive carries them in the failure-carry — the + // concatenations must agree (consumption accounting + // across both drives). let mut whole = WriteChunkWalk::new(); let replay_result = whole.feed(data); - assert!( - matches!(replay_result, Walk::Oversize | Walk::PendingOverCap), - "a split-drive failure must be reachable on the whole drive too" + let replay_pre = match replay_result { + Walk::Oversize(p) | Walk::PendingOverCap(p) => p, + other => panic!( + "a split-drive failure must be reachable on the whole drive too, got {other:?}" + ), + }; + let split_parsed: Vec = emitted + .iter() + .copied() + .chain(pre_pieces.iter().flatten().copied()) + .collect(); + + let whole_parsed: Vec = + replay_pre.iter().flatten().copied().collect(); + assert_eq!( + split_parsed, whole_parsed, + "the parsed-before-failure bytes must agree across \ + the whole and split drives (consumption accounting)" ); assert_failure_shape(data); return; @@ -196,7 +216,7 @@ fn assert_outcome(walk: &mut WriteChunkWalk, result: Walk, fed: &[u8], declared: "Incomplete requires no complete frame in the fed bytes" ); } - Walk::Oversize => { + Walk::Oversize(pre_pieces) => { assert!( declared.oversize, "Oversize requires a header declaring > MAX_CHUNK_LEN" @@ -206,8 +226,20 @@ fn assert_outcome(walk: &mut WriteChunkWalk, result: Walk, fed: &[u8], declared: fed.len() - declared.stop, "the over-length header's bytes onward stay un-consumed" ); + // Consumption accounting holds through the failure: the + // carry's bytes equal every complete frame the walk + // consumed before the bad header. + let carry: Vec = pre_pieces.iter().flatten().copied().collect(); + let pre_stop: Vec = declared.payloads + [..declared.payloads.len().min(carry.len())] + .to_vec(); + assert_eq!( + carry, pre_stop, + "the failure-carry must equal the pre-failure declared stream" + ); } - Walk::PendingOverCap => { + + Walk::PendingOverCap(_) => { assert!( fed.len() > PENDING_BUFFER_CAP, "PendingOverCap requires an over-cap accumulator; the \ @@ -274,6 +306,15 @@ mod corpus_replay { over.extend_from_slice(b"partial-payload"); fuzz_ws_chunk_framing(&over); fuzz_ws_chunk_framing(&h(4, u32::MAX)); + // The drain-then-failure shape: a complete chunk followed by + // an oversize header IN THE SAME FEED — the failure-carry + // path (post-fix contract: the carried pieces equal the + // pre-failure frames; nothing parsed is dropped). + let mut drain_then_fail = h(7, 4); + drain_then_fail.extend_from_slice(b"abcd"); + drain_then_fail.extend_from_slice(&h(7, MAX_CHUNK_LEN + 1)); + drain_then_fail.extend_from_slice(b"tail"); + fuzz_ws_chunk_framing(&drain_then_fail); // len just under / at the cap (small stand-ins — full 16 MiB // cap frames are valid but expensive; the fuzzer explores the // near-cap shapes and the cap partition is asserted exactly in diff --git a/src/websocket/byte_adapter.rs b/src/websocket/byte_adapter.rs index bb34aba..38ee9a1 100644 --- a/src/websocket/byte_adapter.rs +++ b/src/websocket/byte_adapter.rs @@ -180,9 +180,9 @@ pub const DEFAULT_WS_IDLE_TIMEOUT: Duration = Duration::from_secs(60); /// demux's TooLarge arm and parsing continues), with O(1) state: no /// byte copying, and the forwarded byte stream is unmodified. struct InboundChunkProgress { - header: [u8; 8], - header_fill: u8, - payload_remaining: u64, + pub(crate) header: [u8; 8], + pub(crate) header_fill: u8, + pub(crate) payload_remaining: u64, } impl InboundChunkProgress { @@ -194,7 +194,7 @@ impl InboundChunkProgress { } } - fn observe(&mut self, bytes: &[u8]) -> u32 { + pub(crate) fn observe(&mut self, bytes: &[u8]) -> u32 { let mut completed = 0u32; for &byte in bytes { if self.payload_remaining > 0 { @@ -478,12 +478,14 @@ async fn fail_write_pump( /// - [`Walk::Incomplete`] — fewer than a full header plus payload is /// buffered; the pump awaits the next span. /// - [`Walk::Oversize`] — a header claiming -/// `> MAX_CHUNK_LEN` reached the front of the buffer; the pump fails -/// the stream loudly (close 1011 + `InvalidData`). +/// `> MAX_CHUNK_LEN` reached the front of the buffer; the pump +/// emits the carried pre-failure pieces, then fails the stream +/// loudly (close 1011 + `InvalidData`). /// - [`Walk::PendingOverCap`] — the accumulator exceeded -/// `PENDING_BUFFER_CAP` without completing a chunk; the pump fails -/// the stream loudly (defense-in-depth behind the pre-queue -/// `poll_write` rejection, WS-14). +/// `PENDING_BUFFER_CAP` without completing a chunk; the pump +/// emits the carried pieces, then fails the stream loudly +/// (defense-in-depth behind the pre-queue `poll_write` rejection, +/// WS-14). #[derive(Default)] pub struct WriteChunkWalk { pending: Vec, @@ -520,7 +522,7 @@ impl WriteChunkWalk { ]; let len = u32::from_be_bytes(len_bytes); if len > MAX_CHUNK_LEN { - return Walk::Oversize; + return Walk::Oversize(messages); } let total = 8usize.saturating_add(len as usize); if self.pending.len() < total { @@ -532,7 +534,7 @@ impl WriteChunkWalk { } } if self.pending.len() > PENDING_BUFFER_CAP { - return Walk::PendingOverCap; + return Walk::PendingOverCap(messages); } if messages.is_empty() { Walk::Incomplete @@ -564,10 +566,19 @@ pub enum Walk { Incomplete, /// A header claiming a payload over `MAX_CHUNK_LEN` reached the /// front of the buffer (WS-04/HY-09): the stream fails loudly. - Oversize, + /// Carries the pieces drained from complete chunks BEFORE the + /// bad header — the consumption-accounting contract (every + /// parsed byte is either emitted or retained, wave-1 invariant 6) + /// walks them through the emit path first; the pump sends them + /// and then fails the stream. Empty when the bad header was the + /// first frame. + Oversize(Vec>), /// The accumulator exceeded `PENDING_BUFFER_CAP` without a - /// complete chunk (WS-05): the stream fails loudly. - PendingOverCap, + /// complete chunk (WS-05): the stream fails loudly. Carries the + /// pieces drained from complete chunks before the cap bound + /// tripped (same contract as [`Walk::Oversize`]; the walk drops + /// nothing it parsed). + PendingOverCap(Vec>), } /// The write pump: drains the queued byte spans through @@ -618,7 +629,23 @@ async fn run_write_pump( } } Walk::Incomplete => {} - Walk::Oversize => { + // The failure arms carry the pieces drained before the + // violation: emit them (consumption accounting — the + // walk parses nothing it does not surface), then end + // the stream loudly. + Walk::Oversize(messages) => { + for piece in &messages { + if !send_bounded::( + &mut ws_sink, + piece, + write_timeout, + &write_error_slot, + ) + .await + { + return; + } + } fail_write_pump::( &mut ws_sink, &write_error_slot, @@ -627,7 +654,19 @@ async fn run_write_pump( .await; return; } - Walk::PendingOverCap => { + Walk::PendingOverCap(messages) => { + for piece in &messages { + if !send_bounded::( + &mut ws_sink, + piece, + write_timeout, + &write_error_slot, + ) + .await + { + return; + } + } fail_write_pump::( &mut ws_sink, &write_error_slot, @@ -1180,6 +1219,304 @@ where ) } +/// Fuzz-harness drive surface (`fuzz/shared/src/ws_session_opseq.rs`, +/// target 4): the generic pump bodies over a harness-controlled mock +/// socket flavor, plus the `InboundChunkProgress` walk (WS-01) for +/// direct invariant probes. The mock lives on the harness side; the +/// pumps run unchanged — the same shape as the test-module fake-socket +/// drives, exposed so the fuzzer interleaves write/read/close ops +/// (the alkcall 0.8.1 stateful bug class) against the production +/// assembly. +#[cfg(any(feature = "server", feature = "wss"))] +pub mod fuzzing { + use super::{InboundChunkProgress, WriteMsg}; + use futures::channel::mpsc as futures_mpsc; + use futures::{Sink, SinkExt, Stream, StreamExt}; + use std::sync::{Arc, Mutex as StdMutex}; + use tokio::sync::mpsc; + use tokio::sync::oneshot; + + /// The WS-01 chunk-progress walk (the read pump's idle-deadline + /// input), wrapped for the harness. + pub struct FuzzInboundChunkProgress(InboundChunkProgress); + + impl Default for FuzzInboundChunkProgress { + fn default() -> Self { + Self::new() + } + } + + impl FuzzInboundChunkProgress { + /// A fresh walk (empty header window, no pending payload). + pub fn new() -> Self { + Self(InboundChunkProgress::new()) + } + + /// Feed one arrived byte span; returns the completed-chunk + /// count the deadline resets on (WS-01 progress semantics). + pub fn observe(&mut self, bytes: &[u8]) -> u32 { + self.0.observe(bytes) + } + } + + /// The write-pump failure channel, re-declared here (same shape as + /// the production alias) so the hub's `pub` signatures never name + /// the crate-private alias. + pub type FuzzWriteErrorSlot = Arc>>>; + + /// The mock flavor's message type (binary carries bytes; text and + /// close are the pumps' error/close arms). + #[derive(Debug, Clone, PartialEq)] + pub enum FuzzMsg { + /// A binary message — the byte stream's carrier. + Binary(Vec), + /// A text message — the protocol-error arm (close 1002). + Text, + /// A close frame with code + the per-cause canonical reason. + Close { + /// The close code. + code: u16, + /// The per-cause reason string (WS-15). + reason: &'static str, + }, + } + + /// The mock WS socket halves wired to the generic pumps: a + /// harness-owned message stream (the inbound side) and the sink + /// half the write pump emits through (the outbound side drained + /// by the harness's model). + pub struct FuzzMockSocket { + /// The inbound message queue (what `ws_stream.next()` yields). + pub inbound_rx: futures_mpsc::Receiver>, + /// The outbound message queue (what the write pump sends into). + pub outbound_tx: futures_mpsc::Sender, + } + + impl Stream for FuzzMockSocket { + type Item = Result; + + fn poll_next( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + self.inbound_rx.poll_next_unpin(cx) + } + } + + impl Sink for FuzzMockSocket { + type Error = (); + + fn poll_ready( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + self.outbound_tx.poll_ready_unpin(cx).map_err(|_| ()) + } + + fn start_send( + mut self: std::pin::Pin<&mut Self>, + item: FuzzMsg, + ) -> Result<(), Self::Error> { + self.outbound_tx.start_send_unpin(item).map_err(|_| ()) + } + + fn poll_flush( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + self.outbound_tx.poll_flush_unpin(cx).map_err(|_| ()) + } + + fn poll_close( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + self.outbound_tx.poll_close_unpin(cx).map_err(|_| ()) + } + } + + /// Split the mock socket exactly like the production splits. + pub fn fuzz_mock_splits( + socket: FuzzMockSocket, + ) -> ( + futures::stream::SplitSink, + futures::stream::SplitStream, + ) { + socket.split() + } + + /// The production-assembly channel bundle with scalable read + /// slots, opaque over the crate-private `WriteMsg`: the read side + /// forwards inbound bytes through `tokio_read_tx/read_rx`; the + /// write side's queue is reached only through the [`FuzzWriteSide`] + /// handle; the failure slot pairs with `FuzzChannels::write_error_rx`. + pub fn fuzz_channel_set_with_read_slots(read_slots: usize) -> FuzzChannels { + FuzzChannels::new(read_slots) + } + + /// The assembled channel bundle (see + /// [`fuzz_channel_set_with_read_slots`]). + #[allow(dead_code)] + pub struct FuzzChannels { + /// The read side's forward channel: bytes delivered to the + /// demux. + pub tokio_read_tx: mpsc::Sender>, + /// The receiving half (harness-owned or handed to the demux). + pub tokio_read_rx: mpsc::Receiver>, + /// The write side: queue writes and drain close requests. + pub write: FuzzWriteSide, + /// The pump failure slot (both pumps' fatal-reason channel). + pub write_error_slot: FuzzWriteErrorSlot, + /// The one-shot failure receiver. + pub write_error_rx: oneshot::Receiver<&'static str>, + } + + /// The write-side queue handle, opaque over `WriteMsg`. + pub struct FuzzWriteSide { + tx: Option>, + write_rx: Option>, + } + + impl FuzzWriteSide { + fn new() -> Self { + let (tx, rx) = futures_mpsc::channel::(super::WRITE_SLOTS); + Self { + tx: Some(tx), + write_rx: Some(rx), + } + } + + /// Queue one byte span (the mux's `AsyncWrite` shape). + pub fn queue_bytes(&mut self, bytes: Vec) -> Result<(), String> { + match self.tx.as_mut() { + Some(tx) => tx + .try_send(WriteMsg::Bytes(bytes)) + .map_err(|e| format!("write queue rejected: {e:?}")), + None => Err("write queue closed".to_string()), + } + } + + /// Request a close frame with the given code + reason. + pub fn queue_close(&mut self, code: u16, reason: &'static str) -> Result<(), String> { + match self.tx.as_mut() { + Some(tx) => tx + .try_send(WriteMsg::CloseWith(code, reason)) + .map_err(|e| format!("write queue rejected: {e:?}")), + None => Err("write queue closed".to_string()), + } + } + + /// Whether the queue still has a live sender half (the harness + /// checks before queueing; try_send errors name the + /// disposition either way). + pub fn has_tx(&self) -> bool { + self.tx.is_some() + } + + /// Drop the sending half (the queue-close trigger; also the + /// stream-side shutdown shape). + pub fn drop_tx(&mut self) { + self.tx = None; + } + } + + impl FuzzChannels { + fn new(read_slots: usize) -> Self { + let (tokio_read_tx, tokio_read_rx) = mpsc::channel(read_slots); + let (write_error_slot, write_error_rx) = super::make_write_error_slot(); + Self { + tokio_read_tx, + tokio_read_rx, + write: FuzzWriteSide::new(), + write_error_slot, + write_error_rx, + } + } + + /// Try to take the fatal reason (oneshot try_recv semantics; + /// `None` = nothing to observe yet or the channel already + /// drained). + pub fn try_take_write_error(&mut self) -> Option<&'static str> { + self.write_error_rx.try_recv().ok() + } + } + + /// Run the generic read pump body (WS-01 idle semantics included) + /// over the mock flavor, drawn from the channel bundle: the read + /// side forwards into the bundle's `tokio_read_tx`; the fatal + /// close request rides the bundle's write queue (opaque inside + /// the bundle — the harness observes the queue through + /// `write_error_rx` and the outbound socket only). + pub async fn fuzz_spawn_read_pump( + channels: &FuzzChannels, + stream: futures::stream::SplitStream, + idle_timeout: Option, + ) -> tokio::task::JoinHandle<()> { + let write_tx_for_read = channels + .write + .tx + .clone() + .expect("the read pump's clone is taken before any drop_tx"); + tokio::spawn(super::run_read_pump::( + stream, + channels.tokio_read_tx.clone(), + write_tx_for_read, + channels.write_error_slot.clone(), + idle_timeout, + || {}, + )) + } + + /// Run the generic write pump body (the queued chunk parse + + /// `WriteChunkWalk` + WS-18 bounded sends) over the mock flavor, + /// drawn from the channel bundle. + pub async fn fuzz_spawn_write_pump( + channels: &mut FuzzChannels, + sink: futures::stream::SplitSink, + write_timeout: Option, + ) -> Result, String> { + let Some(write_rx) = channels.write.write_rx.take() else { + return Err("write pump already spawned".to_string()); + }; + Ok(tokio::spawn(super::run_write_pump::( + sink, + write_rx, + channels.write_error_slot.clone(), + write_timeout, + ))) + } + + struct FuzzFramingMock; + + impl super::WsFraming for FuzzFramingMock { + type Bytes = Vec; + type Msg = FuzzMsg; + + fn binary(msg: &Self::Msg) -> Option<&Self::Bytes> { + match msg { + FuzzMsg::Binary(b) => Some(b), + _ => None, + } + } + + fn is_text(msg: &Self::Msg) -> bool { + matches!(msg, FuzzMsg::Text) + } + + fn is_close(msg: &Self::Msg) -> bool { + matches!(msg, FuzzMsg::Close { .. }) + } + + fn close_message(code: u16, reason: &'static str) -> Self::Msg { + FuzzMsg::Close { code, reason } + } + + fn binary_message(bytes: Vec) -> Self::Msg { + FuzzMsg::Binary(bytes) + } + } +} + #[cfg(test)] mod tests { use super::*;