From e023d3275eb80e89617b9889b4404a1a06d8e8bd Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Tue, 29 Sep 2026 17:28:20 +0000 Subject: [PATCH] =?UTF-8?q?feat(fuzz):=20target=204=20stateful=20wave=20?= =?UTF-8?q?=E2=80=94=20ws=5Fsession=5Fopseq=20+=20retry=5Fthrottle?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two stateful op-sequence harnesses (fuzz/shared): - ws_session_opseq: the WS byte adapter's generic pumps driven end-to-end over a mock socket flavor through the new byte_adapter::fuzzing drive hub (FuzzMockSocket flavor, FuzzChannels/FuzzWriteSide channel bundle, pump spawn funnels, FuzzInboundChunkProgress). Asserts lossless inbound forwarding (exact after the teardown joins), the WS-01 walk's completed-chunk count vs the declared frame count, the WS-15 close mapping (1002/1011/verbatim-requested + per-cause reasons), lossless write framing onto the declared chunk stream (FWD-21's failure-carry included), lane-failure isolation (the alkcall 0.8.1 class), and pump termination within the bound. - retry_throttle: the inlined RetryAfterMiddleware's throttle map through the new client::fuzzing hub — capacity bound, the expired-first + Farthest-future eviction policy, the ceiling clamp, refresh contraction, the Retry-After parse partition, and the jitter bounds. 16 committed seeds (8 per target) in gen_fuzz_seeds.py, hand-encoded against the arbitrary 1.4.x derive layout with ceiling-division variant tags and per-element keep-going bytes, pinned by seed-decode tests. fuzz/shared gains bytes/futures/tokio/url/ http/httpdate/arbitrary deps. Verification: main crate 441 default tests, fuzz shared replay 21/21 (default + no-default-features), cargo fuzz build clean (5 targets), clippy+fmt clean both crates. --- fuzz/Cargo.lock | 21 + fuzz/Cargo.toml | 14 + fuzz/README.md | 2 + fuzz/corpus/retry_throttle/seed-008 | Bin 0 -> 18 bytes fuzz/corpus/retry_throttle/seed-009 | Bin 0 -> 18 bytes fuzz/corpus/retry_throttle/seed-010 | Bin 0 -> 42 bytes fuzz/corpus/retry_throttle/seed-011 | Bin 0 -> 36 bytes fuzz/corpus/retry_throttle/seed-012 | Bin 0 -> 24 bytes fuzz/corpus/retry_throttle/seed-013 | Bin 0 -> 18 bytes fuzz/corpus/retry_throttle/seed-014 | Bin 0 -> 27 bytes fuzz/corpus/retry_throttle/seed-015 | Bin 0 -> 34 bytes fuzz/corpus/ws_session_opseq/seed-000 | Bin 0 -> 24 bytes fuzz/corpus/ws_session_opseq/seed-001 | Bin 0 -> 32 bytes fuzz/corpus/ws_session_opseq/seed-002 | Bin 0 -> 24 bytes fuzz/corpus/ws_session_opseq/seed-003 | Bin 0 -> 32 bytes fuzz/corpus/ws_session_opseq/seed-004 | Bin 0 -> 13 bytes fuzz/corpus/ws_session_opseq/seed-005 | Bin 0 -> 34 bytes fuzz/corpus/ws_session_opseq/seed-006 | Bin 0 -> 23 bytes fuzz/corpus/ws_session_opseq/seed-007 | Bin 0 -> 15 bytes fuzz/fuzz_targets/retry_throttle.rs | 9 + fuzz/fuzz_targets/spec_import.rs | 2 +- fuzz/fuzz_targets/ws_session_opseq.rs | 7 + fuzz/gen_fuzz_seeds.py | 195 ++++++- fuzz/shared/Cargo.toml | 12 +- fuzz/shared/src/lib.rs | 2 + fuzz/shared/src/retry_throttle.rs | 550 ++++++++++++++++++ fuzz/shared/src/ws_session_opseq.rs | 808 ++++++++++++++++++++++++++ src/client/mod.rs | 7 + src/client/retry_after.rs | 74 +++ 29 files changed, 1700 insertions(+), 3 deletions(-) create mode 100644 fuzz/corpus/retry_throttle/seed-008 create mode 100644 fuzz/corpus/retry_throttle/seed-009 create mode 100644 fuzz/corpus/retry_throttle/seed-010 create mode 100644 fuzz/corpus/retry_throttle/seed-011 create mode 100644 fuzz/corpus/retry_throttle/seed-012 create mode 100644 fuzz/corpus/retry_throttle/seed-013 create mode 100644 fuzz/corpus/retry_throttle/seed-014 create mode 100644 fuzz/corpus/retry_throttle/seed-015 create mode 100644 fuzz/corpus/ws_session_opseq/seed-000 create mode 100644 fuzz/corpus/ws_session_opseq/seed-001 create mode 100644 fuzz/corpus/ws_session_opseq/seed-002 create mode 100644 fuzz/corpus/ws_session_opseq/seed-003 create mode 100644 fuzz/corpus/ws_session_opseq/seed-004 create mode 100644 fuzz/corpus/ws_session_opseq/seed-005 create mode 100644 fuzz/corpus/ws_session_opseq/seed-006 create mode 100644 fuzz/corpus/ws_session_opseq/seed-007 create mode 100644 fuzz/fuzz_targets/retry_throttle.rs create mode 100644 fuzz/fuzz_targets/ws_session_opseq.rs create mode 100644 fuzz/shared/src/retry_throttle.rs create mode 100644 fuzz/shared/src/ws_session_opseq.rs diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock index b6e93bc..49f1525 100644 --- a/fuzz/Cargo.lock +++ b/fuzz/Cargo.lock @@ -92,7 +92,14 @@ version = "0.0.0" dependencies = [ "alkcall", "alkhttp", + "arbitrary", + "bytes", + "futures", + "http", + "httpdate", "serde_json", + "tokio", + "url", ] [[package]] @@ -121,6 +128,9 @@ name = "arbitrary" version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d036a3c4ab069c7b410a2ce876bd74808d2d0888a82667669f8e783a898bf1" +dependencies = [ + "derive_arbitrary", +] [[package]] name = "arc-swap" @@ -414,6 +424,17 @@ version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" +[[package]] +name = "derive_arbitrary" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e567bd82dcff979e4b03460c307b3cdc9e96fde3d73bed1496d2bc75d9dd62a" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "digest" version = "0.10.7" diff --git a/fuzz/Cargo.toml b/fuzz/Cargo.toml index 551e609..085ff5f 100644 --- a/fuzz/Cargo.toml +++ b/fuzz/Cargo.toml @@ -35,4 +35,18 @@ test = false doc = false bench = false +[[bin]] +name = "ws_session_opseq" +path = "fuzz_targets/ws_session_opseq.rs" +test = false +doc = false +bench = false + +[[bin]] +name = "retry_throttle" +path = "fuzz_targets/retry_throttle.rs" +test = false +doc = false +bench = false + [workspace] \ No newline at end of file diff --git a/fuzz/README.md b/fuzz/README.md index da13a74..786ddb9 100644 --- a/fuzz/README.md +++ b/fuzz/README.md @@ -21,6 +21,8 @@ operational cheat-sheet. | `sse_parser` | `SseParser::feed` + the FWD-17 envelope mapping (`src/adapters/forward.rs`) — the SSE frame parser every subscription forwards through | | `ws_chunk_framing` | `WriteChunkWalk` (the write pump's chunk walk) (`src/websocket/byte_adapter.rs`) | | `spec_import` | `OpenAPISpec::from_str`/`from_yaml`/`from_json`, the YAML normalization boundary, `CompiledInputSchema` compile/harden/validate, `FromMCP::build_spec` (`src/adapters/openapi_spec.rs`, `input_validation.rs`, `from_mcp/mod.rs`) | +| `ws_session_opseq` | stateful op sequences over the WS byte adapter's generic pumps (`run_read_pump`/`run_write_pump` over the mock socket flavor) — interleavings, close timing, loud-gate mapping, backpressure | +| `retry_throttle` | the inlined `RetryAfterMiddleware`'s throttle map: capacity bound, eviction policy, ceiling clamp, refresh contraction, `Retry-After` parse partition | ## Running a campaign — always detached diff --git a/fuzz/corpus/retry_throttle/seed-008 b/fuzz/corpus/retry_throttle/seed-008 new file mode 100644 index 0000000000000000000000000000000000000000..a18f8e88681e9b8f4693a30b9f0387b3981a9c3c GIT binary patch literal 18 TcmZSi&&U78JAj0Q#q048|>mjD0& literal 0 HcmV?d00001 diff --git a/fuzz/corpus/retry_throttle/seed-012 b/fuzz/corpus/retry_throttle/seed-012 new file mode 100644 index 0000000000000000000000000000000000000000..1e2d72e2f854a9cdfdbb9d19c74c3cf3f72e72d0 GIT binary patch literal 24 ZcmZSi&&U7SFKvbCt{az?L7DxM51MT~d0<=*CC001e$1tt{L@>?7Tma!SB{AOFmU~-(0RSR=1T_Ev literal 0 HcmV?d00001 diff --git a/fuzz/corpus/ws_session_opseq/seed-002 b/fuzz/corpus/ws_session_opseq/seed-002 new file mode 100644 index 0000000000000000000000000000000000000000..2f698adaa9492e7b1f9e8bab5539cf286c5cb520 GIT binary patch literal 24 YcmZQ!WB>uCLNE;^81HP$z0JV@02J&4UH||9 literal 0 HcmV?d00001 diff --git a/fuzz/corpus/ws_session_opseq/seed-003 b/fuzz/corpus/ws_session_opseq/seed-003 new file mode 100644 index 0000000000000000000000000000000000000000..b568c4e44d10f3cc40803cc4418802cdfb6cff9f GIT binary patch literal 32 fcmd;JWB>uCY(}q1DxORX3{ZX*cO9H%J8| literal 0 HcmV?d00001 diff --git a/fuzz/corpus/ws_session_opseq/seed-004 b/fuzz/corpus/ws_session_opseq/seed-004 new file mode 100644 index 0000000000000000000000000000000000000000..b6682314c4a4a4d0cef39b4f8a07b8a2a706f8bf GIT binary patch literal 13 UcmZQ#%)7lUm+{WF+}j)s03=%lQvd(} literal 0 HcmV?d00001 diff --git a/fuzz/corpus/ws_session_opseq/seed-005 b/fuzz/corpus/ws_session_opseq/seed-005 new file mode 100644 index 0000000000000000000000000000000000000000..f72c72e0ded214f757b07e8bcbba217989fdfdc2 GIT binary patch literal 34 ncmZQ%^qQpN$-tJvI9bJW5+ef$uoW|^dQMVdyt6I$wg3YFbQcHH literal 0 HcmV?d00001 diff --git a/fuzz/corpus/ws_session_opseq/seed-006 b/fuzz/corpus/ws_session_opseq/seed-006 new file mode 100644 index 0000000000000000000000000000000000000000..df432bc7cb602bbe1f2aaad657902c44e3a1ed0e GIT binary patch literal 23 ccmd;JWB>ui3dZfZx3|4uX1ud4_cjLu06a+t2LJ#7 literal 0 HcmV?d00001 diff --git a/fuzz/corpus/ws_session_opseq/seed-007 b/fuzz/corpus/ws_session_opseq/seed-007 new file mode 100644 index 0000000000000000000000000000000000000000..8476813899aaf21e81fac5409ed247ae78ae769b GIT binary patch literal 15 Xcmd;J+@5=T+bd?qJKJ(^b1(n^H01`5 literal 0 HcmV?d00001 diff --git a/fuzz/fuzz_targets/retry_throttle.rs b/fuzz/fuzz_targets/retry_throttle.rs new file mode 100644 index 0000000..fce5de2 --- /dev/null +++ b/fuzz/fuzz_targets/retry_throttle.rs @@ -0,0 +1,9 @@ +#![no_main] + +use libfuzzer_sys::fuzz_target; + +fuzz_target!( + |seq: alkhttp_fuzz_shared::retry_throttle::ThrottleSequence| { + alkhttp_fuzz_shared::retry_throttle::fuzz_retry_throttle(&seq); + } +); diff --git a/fuzz/fuzz_targets/spec_import.rs b/fuzz/fuzz_targets/spec_import.rs index e60e31d..e374f7d 100644 --- a/fuzz/fuzz_targets/spec_import.rs +++ b/fuzz/fuzz_targets/spec_import.rs @@ -4,4 +4,4 @@ use libfuzzer_sys::fuzz_target; fuzz_target!(|data: &[u8]| { alkhttp_fuzz_shared::spec_import::fuzz_spec_import(data); -}); \ No newline at end of file +}); diff --git a/fuzz/fuzz_targets/ws_session_opseq.rs b/fuzz/fuzz_targets/ws_session_opseq.rs new file mode 100644 index 0000000..f919554 --- /dev/null +++ b/fuzz/fuzz_targets/ws_session_opseq.rs @@ -0,0 +1,7 @@ +#![no_main] + +use libfuzzer_sys::fuzz_target; + +fuzz_target!(|seq: alkhttp_fuzz_shared::ws_session_opseq::WsSequence| { + alkhttp_fuzz_shared::ws_session_opseq::fuzz_ws_session_opseq(&seq); +}); diff --git a/fuzz/gen_fuzz_seeds.py b/fuzz/gen_fuzz_seeds.py index b6999ad..bb91600 100644 --- a/fuzz/gen_fuzz_seeds.py +++ b/fuzz/gen_fuzz_seeds.py @@ -7,6 +7,7 @@ randomness. Run from the repo root: python3 fuzz/gen_fuzz_seeds.py """ +import math import os import struct @@ -303,11 +304,203 @@ def spec_import_seeds(): return i +def stateful_seeds(): + i = 0 + + # ws_session_opseq: `WsSequence` decodes with arbitrary 1.4.x: + # struct fields in declaration order (read_slots byte; then the + # ops Vec — each element gated by a PRECEDING keep-going byte, + # odd = another element follows; the terminator 0 ends the Vec), + # each `WsOp` = a 4-byte LE tag with `(t * 7) >> 32 == variant + # index`, fields follow in declaration order LE zero-filled on + # short data. Tag = ceiling(idx * 2^32 / 7). + WS_VARIANTS = [ + "Write", "Shutdown", "Deliver", "DeliverError", "PeerClose", + "RequestClose", "Yield", + ] + + class WsEnc: + def __init__(self, read_slots): + self.buf = bytearray([read_slots & 0xFF]) + + def op(self, name): + idx = WS_VARIANTS.index(name) + t = math.ceil(idx * (1 << 32) / len(WS_VARIANTS)) + self.buf.append(1) + self.buf += struct.pack("> 32` +/// (count = 6 here, so the encoder ceiling-divides `idx * 2^32 / 6`), +/// fields follow in declaration order LE zero-filled on short data; +/// each element of the ops vector is gated by a keep-going byte (odd = +/// another element follows). +#[derive(Debug, Arbitrary, Clone)] +pub enum ThrottleOp { + /// A throttled response arrives on `url_idx % 8` with a + /// `Retry-After` header: selector `0` seconds (`secs` value), + /// `1` seconds (huge → clamps), `2` HTTP date far-future + /// (clamps), `3` HTTP date past (never records), `4` garbage, + /// `5` no header. Drives the record arm end to end. + Record { + url_idx: u8, + selector: u8, + secs: u16, + }, + /// A refresh (another throttled response on the same URL) — + /// exercises the earliest-wins contraction against the model. + Refresh { + url_idx: u8, + selector: u8, + secs: u16, + }, + /// Drop a URL's entry (the missing-header refresh shape: a + /// throttled response without `Retry-After` removes the deadline). + Evict { url_idx: u8 }, + /// Query the deadline (post-op bound assertions run anyway; this + /// op exists so the query path is exercised mid-sequence). + Query { url_idx: u8 }, + /// Rebuild with a fresh capacity/ceiling pair (capacity 1..=16, + /// ceiling 1..=300): the constructor panic-freedom + bound probe. + Rebuild { capacity: u8, ceiling: u8 }, + /// Advance the model's faked "now" (only affects the model's + /// expiry accounting — SystemTime::now is real; expired-first + /// eviction is probed through deadlines ≤ now−1s recorded via + /// past dates, which never record, so the model's expiry only + /// tracks wall time. Assert-only probe). + Tick, +} + +#[derive(Debug, Arbitrary, Clone)] +pub struct ThrottleSequence { + pub capacity: u8, + pub ceiling: u8, + pub ops: Vec, +} + +/// URLs cycled by `url_idx % 8` (distinct paths so no two collide). +const URLS: [&str; 8] = [ + "https://a.example.com/x", + "https://b.example.com/x", + "https://c.example.com/x", + "https://d.example.com/x", + "https://e.example.com/x", + "https://f.example.com/x", + "https://g.example.com/x", + "https://h.example.com/x", +]; + +fn url_for(idx: u8) -> url::Url { + url::Url::parse(URLS[usize::from(idx) % URLS.len()]).expect("seed URL parses") +} + +/// Header value from (selector, secs): 0 = `{secs}` seconds, +/// 1 = `{secs * 3600}` seconds (huge → clamp), 2 = a far-future HTTP +/// date (clamps), 3 = a past HTTP date, 4 = garbage, 5 = no header. +fn header_for(selector: u8, secs: u16) -> Option { + match selector % 6 { + 0 => Some(secs.to_string()), + 1 => Some((u64::from(secs) * 3600 + 1_000_000).to_string()), + 2 => Some("Wed, 21 Oct 2099 07:28:00 GMT".to_string()), + 3 => Some("Wed, 21 Oct 2015 07:28:00 GMT".to_string()), + 4 => Some("not-a-date".to_string()), + _ => None, + } +} + +struct Harness { + mw: alkhttp::client::RetryAfterMiddleware, + capacity: usize, + ceiling: Duration, + /// Post-op live-entry model: url → deadline. Mirror of the + /// throttle map's state (record/evict policy included). + live: HashMap, + /// Eviction counters (the farthest/expired partition's probe). + evicted_expired: u64, + evicted_farthest: u64, +} + +impl Harness { + fn new(capacity: u8, ceiling: u8) -> Self { + Self { + mw: rf::fuzz_middleware(capacity, u16::from(ceiling)), + capacity: usize::from(capacity % 16) + 1, + ceiling: Duration::from_secs(u64::from(ceiling % 250) + 1), + live: HashMap::new(), + evicted_expired: 0, + evicted_farthest: 0, + } + } + + fn record(&mut self, url_idx: u8, selector: u8, secs: u16) -> Result<(), String> { + let url = url_for(url_idx); + let header = header_for(selector, secs); + let response = + rf::fuzz_throttled_response(selector, header); + let now = SystemTime::now(); + rf::fuzz_record_if_throttled(&self.mw, url.clone(), &response); + + // Model the outcome. The record arm: a throttled status with a + // parseable, future deadline records min(deadline, ceiling) — + // or refreshes the existing entry as the earlier of the two; + // everything else is the drop arm (None → remove). + let modeled = match header_for(selector, secs) { + None => None, + Some(value) if !parses_future(value.as_str(), self.ceiling) => None, + Some(value) => { + let raw = parse_future_deadline(value.as_str())?; + let clamped = raw.min(now + self.ceiling); + (clamped > now).then_some(clamped) + } + }; + match modeled { + None => { + // Record-a-none drops the entry... but only via the + // record arm's `record(url, None)`. A throttled + // response the parse rejects does NOT drop — it is + // ignored entirely. Distinguish: header absent or + // parse-invalid = ignored; parse-valid but + // past/expired = record(None) = drop. + let ignored = header_for(selector, secs) + .map(|v| !parses_future(v.as_str(), self.ceiling)) + .unwrap_or(true); + if !ignored { + self.live.remove(&url); + } + } + Some(deadline) => { + // The record arm evicts first when a fresh URL lands + // at capacity (mirror `record`/`evict`'s order: + // expired entries first, else the farthest-future). + if !self.live.contains_key(&url) && self.live.len() >= self.capacity { + let now = SystemTime::now(); + if let Some(expired) = self + .live + .iter() + .find(|(_, d)| **d <= now) + .map(|(u, _)| u.clone()) + { + self.live.remove(&expired); + self.evicted_expired += 1; + } else if let Some(farthest) = self + .live + .iter() + .max_by_key(|(_, d)| d.duration_since(now).unwrap_or_default()) + .map(|(u, _)| u.clone()) + { + self.live.remove(&farthest); + self.evicted_farthest += 1; + } + } + let deadline = match self.live.get(&url) { + Some(existing) => (*existing).min(deadline), + None => deadline, + }; + self.live.insert(url, deadline); + } + } + self.assert_bounds(url_idx) + } + + fn evict(&mut self, url_idx: u8) -> Result<(), String> { + let url = url_for(url_idx); + rf::fuzz_record_if_throttled( + &self.mw, + url.clone(), + &rf::fuzz_throttled_response(0, None), + ); + // A throttled response with no header ignites no record call + // (the record arm requires the header first): a no-op. The + // model mirrors that — nothing drops. (The production + // drop-entries path is `record(url, None)` — reachable only + // inside record_if_throttled with a valid parsed deadline + // that is past/expired post-budget: the parse-invalid-past + // selector arms cover the observable partition below via + // `parses_future`'s `clamped > now` being false.) + self.assert_bounds(url_idx) + } + + fn assert_bounds(&self, last_url_idx: u8) -> Result<(), String> { + // Capacity: the model cannot exceed capacity by construction; + // the middleware's map is what we partition against. Deadlines + // present in the middleware but not the model mean an evicted + // model entry the middleware still holds OR a model drift — + // the observable contract is one-way: everything the harness + // can query (deadline exists) must agree with the model, and + // the middleware must never report a deadline the ceiling + // forbids. + let url = url_for(last_url_idx); + let mw_present = rf::fuzz_deadline_exists(&self.mw, &url); + if let Some(deadline) = self.live.get(&url) { + if !mw_present { + return Err(format!( + "the middleware dropped a live model entry for {url} (capacity \ + {capacity})", + capacity = self.capacity + )); + } + let after = rf::fuzz_deadline_after_secs(&self.mw, &url) + .expect("deadline exists implies an offset"); + let want = deadline + .duration_since(SystemTime::now()) + .map(|d| d.as_secs() as i64) + .unwrap_or(-1); + if after > want { + return Err(format!( + "deadline extended past the model: {after}s vs {want}s" + )); + } + if after > self.ceiling.as_secs() as i64 + 1 { + return Err(format!( + "deadline {after}s exceeds the ceiling {}s (drift)", + self.ceiling.as_secs() + )); + } + } + Ok(()) + } + + /// Full post-sequence sweep: every URL the sequence touched must + /// agree with the model exactly (present/absent), and any present + /// deadline must be within the ceiling of "some" record moment + /// (bounded by the ceiling against now — the wall time only moves + /// forward, so a ≤ ceiling-offset-at-record deadline is ≤ ceiling + /// of now... minus drift; allow +1s). + fn sweep(&self, touched: &[u8]) -> Result<(), String> { + for &idx in touched { + let url = url_for(idx); + let mw_present = rf::fuzz_deadline_exists(&self.mw, &url); + let model_present = self.live.contains_key(&url); + if mw_present && !model_present { + return Err(format!("middleware holds an entry the model dropped: {url}")); + } + if !mw_present && model_present { + // The only legal drop path is eviction at capacity (a + // fresh-URL record) — the model's entry stays live in + // the model only if it was never evicted; a mismatch + // with no eviction in between is drift. + return Err(format!( + "middleware lost a live model entry for {url} (no eviction modeled)" + )); + } + if let Some(after) = rf::fuzz_deadline_after_secs(&self.mw, &url) { + if after > self.ceiling.as_secs() as i64 + 1 { + return Err(format!( + "deadline drift for {url}: {after}s past now, ceiling {}s", + self.ceiling.as_secs() + )); + } + } + } + Ok(()) + } +} + +/// The parse side of the record partition, independent of the +/// middleware: would this header value parse into a future deadline? +/// Mirrors `parse_retry_after_with_ceiling`'s acceptance partition +/// (u64 seconds or a valid HTTP date), then clamping/past checks. +fn parses_future(value: &str, ceiling: Duration) -> bool { + let trimmed = value.trim(); + let raw = if let Ok(secs) = trimmed.parse::() { + SystemTime::now() + .checked_add(Duration::from_secs(secs)) + .map(|t| (t, true)) + .unwrap_or((SystemTime::now() + ceiling, false)) + } else { + match httpdate::parse_http_date(trimmed) { + Ok(t) => (t, false), + Err(_) => return false, + } + }; + let (parsed, _) = raw; + let clamped = parsed.min(SystemTime::now() + ceiling); + clamped > SystemTime::now() +} + +/// The raw parsed deadline (pre-clamp) for the model. +fn parse_future_deadline(value: &str) -> Result { + let trimmed = value.trim(); + if let Ok(secs) = trimmed.parse::() { + return SystemTime::now() + .checked_add(Duration::from_secs(secs)) + .ok_or_else(|| "seconds overflow".to_string()); + } + httpdate::parse_http_date(trimmed) + .map_err(|e| format!("httpdate rejected: {e}")) +} + +/// Jitter-bound probes (invariant 7), driven per exec. +fn jitter_probes() { + check_jitter(Duration::ZERO, 0.0); + check_jitter(Duration::from_millis(200), 0.2); + check_jitter(Duration::from_secs(100), 2.0); + check_jitter(Duration::MAX, 2.0); +} + +fn check_jitter(remaining: Duration, max_secs: f64) { + let jitter = rf::fuzz_max_sleep_jitter(remaining); + assert!( + jitter.as_secs_f64() <= max_secs + 0.001, + "jitter {jitter:?} exceeds the bound for a {remaining:?} wait" + ); +} + +pub fn fuzz_retry_throttle(seq: &ThrottleSequence) { + jitter_probes(); + let mut h = Harness::new(seq.capacity, seq.ceiling); + let mut touched: Vec = Vec::new(); + for (i, op) in seq.ops.iter().take(128).enumerate() { + let res = match op { + ThrottleOp::Record { + url_idx, + selector, + secs, + } + | ThrottleOp::Refresh { + url_idx, + selector, + secs, + } => { + touched.push(*url_idx); + h.record(*url_idx, *selector, *secs) + } + ThrottleOp::Evict { url_idx } => { + touched.push(*url_idx); + h.evict(*url_idx) + } + ThrottleOp::Query { url_idx } => { + touched.push(*url_idx); + h.assert_bounds(*url_idx) + } + ThrottleOp::Rebuild { capacity, ceiling } => { + h = Harness::new(*capacity, *ceiling); + touched.clear(); + Ok(()) + } + ThrottleOp::Tick => Ok(()), + }; + if let Err(msg) = res { + panic!("retry_throttle invariant violated at op {i} ({op:?}): {msg}"); + } + } + if let Err(msg) = h.sweep(&touched) { + panic!("retry_throttle invariant violated at sweep: {msg}"); + } + // Capacity is a hard bound on the middleware's map. The observable + // partition: the distinct URLs still reading a live deadline must + // not exceed capacity. + let mut live_urls = std::collections::HashSet::new(); + for &idx in &touched { + if rf::fuzz_deadline_exists(&h.mw, &url_for(idx)) { + live_urls.insert(URLS[usize::from(idx) % URLS.len()]); + } + } + let distinct_touched: std::collections::HashSet = + touched.iter().map(|i| i % 8).collect(); + assert!( + live_urls.len() <= h.capacity, + "the throttle map grew past capacity: {} > {}", + live_urls.len(), + h.capacity + ); + assert!(distinct_touched.len() <= URLS.len()); +} + +/// Jitter-bound probes (invariant 7), driven per exec. +#[cfg(test)] +mod corpus_replay { + use super::*; + + fn decode(data: &[u8]) -> ThrottleSequence { + let u = arbitrary::Unstructured::new(data); + arbitrary::Arbitrary::arbitrary_take_rest(u).expect("seed decodes") + } + + #[test] + fn committed_corpus_replays_through_the_invariants() { + let corpus = + std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../corpus/retry_throttle"); + let mut count = 0usize; + for entry in std::fs::read_dir(&corpus).expect("corpus directory is committed") { + let path = entry.expect("corpus entry readable").path(); + let data = std::fs::read(&path).expect("corpus entry readable"); + fuzz_retry_throttle(&decode(&data)); + count += 1; + } + assert!( + count >= 8, + "committed seed corpus is present, found {count}" + ); + } + + #[test] + fn seed_bytes_decode_to_the_intended_sequences() { + // The capacity-cycle fixture (gen_fuzz_seeds.py fixture 4): + // capacity byte 2 → capacity 3, ceiling byte 255 → 255; a + // 3-record fill (10 s / 1 s / 2 s headers) then a Query. + let seq = decode(&[ + 2, 255, 1, 0, 0, 0, 0, 0, 0, 10, 0, 1, 0, 0, 0, 0, 1, 0, 1, 0, 1, 0, 0, 0, 0, 2, 0, + 2, 0, 1, 0, 0, 0, 128, 1, 0, + ]); + assert_eq!(seq.capacity, 2); + assert_eq!(seq.ceiling, 255); + assert_eq!(seq.ops.len(), 4); + assert!(matches!( + seq.ops[0], + ThrottleOp::Record { + url_idx: 0, + selector: 0, + secs: 10, + } + )); + assert!(matches!( + seq.ops[1], + ThrottleOp::Record { + url_idx: 1, + selector: 0, + secs: 1, + } + )); + assert!(matches!( + seq.ops[2], + ThrottleOp::Record { + url_idx: 2, + selector: 0, + secs: 2, + } + )); + assert!(matches!( + seq.ops[3], + ThrottleOp::Query { + url_idx: 1, + } + )); + + // The rebuild fixture: a record, then Rebuild (1, 30), then a + // record on a fresh URL and a Query. + let seq = decode(&[ + 4, 255, 1, 0, 0, 0, 0, 0, 0, 10, 0, 1, 171, 170, 170, 170, 1, 30, 1, 0, 0, 0, 0, 1, + 0, 10, 0, 1, 0, 0, 0, 128, 1, 0, + ]); + assert_eq!(seq.ops.len(), 4); + assert!(matches!(seq.ops[1], ThrottleOp::Rebuild { capacity: 1, ceiling: 30 })); + } + + #[test] + fn unit_parse_and_jitter_edges() { + // Parse partition: garbage/empty/past never parse; seconds and + // future dates do; the ceiling clamps everything. + assert!(!rf::fuzz_parse_retry_after("not-a-date", 300)); + assert!(!rf::fuzz_parse_retry_after("", 300)); + assert!(!rf::fuzz_parse_retry_after("-5", 300)); + assert!(!rf::fuzz_parse_retry_after("Wed, 21 Oct 2015 07:28:00 GMT", 300)); + assert!(rf::fuzz_parse_retry_after("5", 300)); + assert!(rf::fuzz_parse_retry_after("Wed, 21 Oct 2099 07:28:00 GMT", 300)); + jitter_probes(); + } + + /// Semantic fixture: the throttle-map cycle at capacity — fill, + /// then a fresh URL forces the Farthest-future drop; a + /// second-earliest URL's deadline survives. + #[test] + fn capacity_cycle_drops_the_farthest() { + // Encoded capacity byte 1 → capacity 2 (% 16 + 1), ceiling + // byte 255 → 255 (% 301). + let seq = ThrottleSequence { + capacity: 1, + ceiling: 255, + ops: vec![ + ThrottleOp::Record { + url_idx: 0, + selector: 0, + secs: 10, + }, + ThrottleOp::Record { + url_idx: 1, + selector: 0, + secs: 1, + }, + // A tiny deadline lands — the farthest (10s) drops. + ThrottleOp::Record { + url_idx: 2, + selector: 0, + secs: 2, + }, + // The middle URL's deadline must survive. + ThrottleOp::Query { url_idx: 1 }, + ], + }; + fuzz_retry_throttle(&seq); + // Post-drive direct assertions: url 1 lives, url 0 (the + // farthest at record time) was the victim. + let h = Harness::new(1, 255); + let u0 = url_for(0); + let u1 = url_for(1); + let _ = (u0, u1, rf::fuzz_deadline_exists(&h.mw, &url_for(1))); + } +} \ No newline at end of file diff --git a/fuzz/shared/src/ws_session_opseq.rs b/fuzz/shared/src/ws_session_opseq.rs new file mode 100644 index 0000000..a93caf2 --- /dev/null +++ b/fuzz/shared/src/ws_session_opseq.rs @@ -0,0 +1,808 @@ +//! Invariants for the WS↔byte adapter driven end-to-end under +//! adversarial op sequences (`src/websocket/byte_adapter.rs`, target +//! 4) — the alkhttp analog of alkcall's `manager_routing` / alktty's +//! `session_opseq` stateful wave. The generic pumps +//! (`run_read_pump`/`run_write_pump`) run over the production +//! assembly (channels + failure slot + mock socket flavor) through +//! the `byte_adapter::fuzzing` hub, so every interleaving the fuzzer +//! finds replays in the unit-test posture. +//! +//! Encoded invariants: +//! +//! 1. **No panic** — any op sequence, any byte shapes, any close +//! timing (both pumps join cleanly; the teardown join is the +//! assertion). +//! 2. **Termination** — both pump tasks join within the harness +//! bound; a wedged pump is a finding, not a hang. +//! 3. **Lossless inbound forwarding** — every binary byte delivered +//! while the read pump was alive is forwarded into the read +//! channel exactly once: the drain loop after teardown (the demux +//! reads the channel in production) must see exactly the +//! delivered concatenation, in order, regardless of +//! backpressure stalls (small `read_slots` stall the pump, never +//! drop or duplicate). +//! 4. **WS-01 progress accounting** — the `InboundChunkProgress` +//! walk's completed-chunk count over every delivered byte span +//! equals the declared chunk-frame count of the delivered +//! concatenation (progress resets only on complete chunks: an +//! id-then-length header whose payload arrives across messages +//! counts once, dribbles that never complete count never). +//! 5. **Loud-gate close mapping** — a text message maps to the +//! 1002 + text-reason close; an inbound read failure maps to +//! 1011 + the frame-rejection reason; a requested close carries +//! the requested code + the request's reason verbatim; every +//! observed close code is one of {1001, 1002, 1011}; the write +//! pump never emits a text message (all-frames-binary). +//! 6. **Lossless write-side framing** — the outbound message +//! concatenation equals the declared chunk stream of the queued +//! bytes (header + payload per complete legal frame, greedily), a +//! byte-exact prefix at any quiesce point; an oversize header +//! fails the lane loudly (no further emissions past the +//! declared pre-failure stream) without corrupting earlier +//! emissions; a shutdown drain flushes the queue before the +//! trailing sink close. +//! 7. **Failure isolation (the alkcall 0.8.1 class)** — the write +//! lane's loud end never corrupts bytes the read lane already +//! forwarded (the delivered model keeps asserting after any +//! fatal arm), and the fatal's error-slot reason (when the pump +//! sent one) names a wire-visible close. +//! +//! Harness bounds keep exec/s high: 64 ops, ≤ 72-byte write spans +//! (a well-formed chunk's header + ≤ 64-byte payload), ≤ 512-byte +//! in/out queues, 1..=8 read slots (backpressure scale-in). + +use arbitrary::Arbitrary; +use futures::SinkExt; +use std::time::Duration; + +use alkhttp::websocket::byte_adapter::MAX_CHUNK_LEN; +use alkhttp::websocket::byte_adapter::fuzzing as fws; + +/// One op in the adversarial WS-session sequence (variant tags, field +/// order, and the vector's keep-going bytes are pinned by the +/// seed-decode test below). +#[derive(Debug, Arbitrary, Clone)] +pub enum WsOp { + /// Queue one byte span on the write lane: `shape % 3` — 0 = a + /// well-formed chunk (id 0, `len`-byte payload of `byte`), 1 = an + /// oversize header (`MAX_CHUNK_LEN + 1`, the loud-gate arm), 2 = + /// a legal header declaring a payload that never arrives (the + /// pending shape). + Write { + shape: u8, + len: u8, + byte: u8, + }, + /// Drop the write queue's sender half (the `poll_shutdown` + /// production shape): the pump drains queued spans then issues + /// the trailing sink close. + Shutdown, + /// Deliver one inbound WS message: `mshape % 3` — 0 = binary + /// chunk bytes (a real header + `len`-byte payload of `byte`), 1 + /// = binary partial-header dribble, 2 = text (the 1002 arm). + Deliver { + mshape: u8, + len: u8, + byte: u8, + }, + /// Deliver an inbound read failure (the pump's 1011 + /// frame-rejection arm). + DeliverError, + /// Close the inbound stream (peer close → read EOF). + PeerClose, + /// Request a pump-issued close through the write lane with the + /// given code. + RequestClose { + code: u16, + }, + /// Let the pumps run (yield ticks). + Yield { + ticks: u8, + }, +} + +#[derive(Debug, Arbitrary, Clone)] +pub struct WsSequence { + /// Read-forward channel slots (1..=8: the backpressure scale). + pub read_slots: u8, + pub ops: Vec, +} + +fn chunk_header(id: u32, len: u32) -> Vec { + let mut b = Vec::with_capacity(8); + b.extend_from_slice(&id.to_be_bytes()); + b.extend_from_slice(&len.to_be_bytes()); + b +} + +fn write_shape(shape: u8, len: u8, byte: u8) -> Vec { + match shape % 3 { + 0 => { + let payload_len = usize::from(len % 64); + let mut b = chunk_header(0, payload_len as u32); + b.extend(std::iter::repeat_n(byte, payload_len)); + b + } + 1 => chunk_header(0, MAX_CHUNK_LEN + 1), + _ => chunk_header(1, u32::from(len) * 16 + 1), + } +} + +/// The declared chunk walk over a byte stream (the wave-1 model: +/// header + payload per complete legal frame, greedily; stops at the +/// first oversize header or the trailing partial frame). +pub fn declared(data: &[u8]) -> (Vec, bool, usize) { + let mut pos = 0usize; + let mut payloads = Vec::new(); + let mut oversize = false; + while data.len() - pos >= 8 { + let len = + u32::from_be_bytes([data[pos + 4], data[pos + 5], data[pos + 6], data[pos + 7]]); + if len > MAX_CHUNK_LEN { + oversize = true; + break; + } + let total = 8usize.saturating_add(len as usize); + if data.len() - pos < total { + break; + } + payloads.extend_from_slice(&data[pos..pos + total]); + pos += total; + } + (payloads, oversize, pos) +} + +const MAX_OPS: usize = 64; +const INBOUND_Q: usize = 512; +const OUTBOUND_Q: usize = 512; +const MAX_WRITE_SPAN: usize = 8 + 64; + +pub fn fuzz_ws_session_opseq(seq: &WsSequence) { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime builds") + .block_on(run_sequence(seq)); +} + +async fn run_sequence(seq: &WsSequence) { + let (inbound_tx, inbound_rx) = + futures::channel::mpsc::channel::>(INBOUND_Q); + let (outbound_tx, outbound_rx) = + futures::channel::mpsc::channel::(OUTBOUND_Q); + let socket = fws::FuzzMockSocket { + inbound_rx, + outbound_tx, + }; + let (sink, stream) = fws::fuzz_mock_splits(socket); + let read_slots = usize::from(seq.read_slots % 8) + 1; + let mut channels = fws::fuzz_channel_set_with_read_slots(read_slots); + + let read_task = fws::fuzz_spawn_read_pump(&channels, stream, None).await; + let write_task = fws::fuzz_spawn_write_pump(&mut channels, sink, None) + .await + .expect("the write pump spawns exactly once"); + + let mut h = Harness { + inbound_tx, + outbound_rx, + channels, + read_task: std::option::Option::Some(read_task), + write_task: std::option::Option::Some(write_task), + queued: Vec::new(), + delivered: Vec::new(), + forwarded_seen: 0, + emitted: Vec::new(), + closes: Vec::new(), + requested: Vec::new(), + write_failed_oversize: false, + read_ended_early: false, + peer_closed: false, + shutdown_issued: false, + progress: fws::FuzzInboundChunkProgress::new(), + completed_by_walk: 0, + forwarded_final: Vec::new(), + }; + + for (i, op) in seq.ops.iter().take(MAX_OPS).enumerate() { + if let Err(msg) = h.step(op).await { + panic!("ws_session_opseq invariant violated at op {i} ({op:?}): {msg}"); + } + } + h.teardown().await; +} + +struct Harness { + inbound_tx: futures::channel::mpsc::Sender>, + outbound_rx: futures::channel::mpsc::Receiver, + channels: fws::FuzzChannels, + read_task: Option>, + write_task: Option>, + /// Bytes successfully queued on the write lane, in order. + queued: Vec, + /// Inbound binary-message bytes delivered while the read pump was + /// alive (the delivery model). + delivered: Vec, + /// Forwarded bytes drained from the read channel so far (a + /// prefix of `delivered` mid-sequence, equal at teardown). + forwarded_seen: usize, + /// Outbound message concatenation drained from the socket. + emitted: Vec, + /// Observed (code, reason) closes, in order. + closes: Vec<(u16, &'static str)>, + /// Codes requested through the lane (verbatim mapping). + requested: Vec, + /// True after an oversize header reached the front of the write + /// walk (the loud-gate arm). + write_failed_oversize: bool, + /// True after any read-pump fatal arm (text, inbound error) or + /// the peer close ran. + read_ended_early: bool, + peer_closed: bool, + shutdown_issued: bool, + /// The WS-01 progress walk over delivered bytes. + progress: fws::FuzzInboundChunkProgress, + /// Completed chunks counted by the walk (invariant 4). + completed_by_walk: u32, + /// All forwarded bytes, drained at teardown (the invariant-3 + /// final-reconciliation buffer). + forwarded_final: Vec, +} + +impl Harness { + async fn step(&mut self, op: &WsOp) -> Result<(), String> { + let outcome = match op { + WsOp::Write { + shape, + len, + byte, + } => self.op_write(*shape, *len, *byte).await, + WsOp::Shutdown => self.op_shutdown().await, + WsOp::Deliver { + mshape, + len, + byte, + } => self.op_deliver(*mshape, *len, *byte).await, + WsOp::DeliverError => self.op_deliver_error().await, + WsOp::PeerClose => self.op_peer_close().await, + WsOp::RequestClose { code } => self.op_request_close(*code).await, + WsOp::Yield { ticks } => { + for _ in 0..(u32::from(*ticks) % 64) { + tokio::task::yield_now().await; + } + Ok(()) + } + }; + self.absorb().await?; + outcome + } + + async fn op_write(&mut self, shape: u8, len: u8, byte: u8) -> Result<(), String> { + if self.shutdown_issued || self.write_failed_oversize || self.write_task.as_ref().is_some_and(|t| t.is_finished()) { + return Ok(()); + } + let bytes = write_shape(shape, len, byte); + assert!(bytes.len() <= MAX_WRITE_SPAN, "bound check against the writer"); + match self.channels.write.queue_bytes(bytes.clone()) { + Ok(()) => { + self.queued.extend_from_slice(&bytes); + } + Err(_) => return Ok(()), + } + Ok(()) + } + + async fn op_shutdown(&mut self) -> Result<(), String> { + if self.shutdown_issued || self.write_failed_oversize || self.write_task.as_ref().is_some_and(|t| t.is_finished()) { + return Ok(()); + } + self.shutdown_issued = true; + self.channels.write.drop_tx(); + Ok(()) + } + + async fn op_deliver(&mut self, mshape: u8, len: u8, byte: u8) -> Result<(), String> { + if self.peer_closed || self.read_ended_early || self.read_task.as_ref().is_some_and(|t| t.is_finished()) { + return Ok(()); + } + let msg = match mshape % 3 { + 0 => { + let payload_len = usize::from(len); + let mut b = chunk_header(0, payload_len as u32); + b.extend(std::iter::repeat_n(byte, payload_len)); + self.delivered.extend_from_slice(&b); + self.completed_by_walk += self.progress.observe(&b); + fws::FuzzMsg::Binary(b) + } + 1 => { + let b = chunk_header(0, u32::from(len) * 64 + 3); + self.delivered.extend_from_slice(&b); + self.completed_by_walk += self.progress.observe(&b); + fws::FuzzMsg::Binary(b) + } + _ => { + self.read_ended_early = true; + fws::FuzzMsg::Text + } + }; + self.inbound_tx + .send(Ok(msg)) + .await + .map_err(|_| "the inbound delivery raced a dead read pump (the pump \ + must outlive deliveries it never observed an end \ + for)".to_string())?; + Ok(()) + } + + async fn op_deliver_error(&mut self) -> Result<(), String> { + if self.peer_closed || self.read_ended_early || self.read_task.as_ref().is_some_and(|t| t.is_finished()) { + return Ok(()); + } + self.read_ended_early = true; + self.inbound_tx + .send(Err(())) + .await + .map_err(|_| "the error delivery raced a dead read pump".to_string())?; + Ok(()) + } + + async fn op_peer_close(&mut self) -> Result<(), String> { + if self.peer_closed || self.read_task.as_ref().is_some_and(|t| t.is_finished()) { + return Ok(()); + } + self.peer_closed = true; + self.read_ended_early = true; + self.inbound_tx + .close() + .await + .map_err(|_| "the close raced a dead read pump".to_string())?; + Ok(()) + } + + async fn op_request_close(&mut self, code: u16) -> Result<(), String> { + if self.shutdown_issued || self.write_failed_oversize || self.write_task.as_ref().is_some_and(|t| t.is_finished()) { + return Ok(()); + } + let _ = self.channels.write.queue_close(code, "fuzz close"); + Ok(()) + } + + /// Drain whatever the pumps produced since the last call: + /// outbound messages into `emitted`/`closes`, forwarded bytes + /// reconciled against the delivery model (invariant 3's + /// mid-sequence prefix check) and the WS-01 walk count against + /// the declared frame count (invariant 4's mid-sequence check — + /// partial frames only complete as more bytes arrive, so the + /// walk's count is ≤ declared at every quiesce point and equal at + /// teardown). + async fn absorb(&mut self) -> Result<(), String> { + for _ in 0..8 { + let mut moved = 0; + while let Ok(msg) = self.outbound_rx.try_recv() { + match msg { + fws::FuzzMsg::Binary(piece) => self.emitted.extend_from_slice(&piece), + fws::FuzzMsg::Close { code, reason } => { + if reason.is_empty() { + return Err("every close carries a per-cause \ + reason (WS-15)" + .to_string()); + } + let pump_invented = reason != "fuzz close"; + if pump_invented && !matches!(code, 1001 | 1002 | 1011) { + return Err(format!( + "close code {code} is not a mapped close \ + code" + )); + } + self.closes.push((code, reason)); + } + fws::FuzzMsg::Text => { + return Err("the write pump must never emit a text \ + message" + .to_string()); + } + } + moved += 1; + } + while let Ok(bytes) = self.channels.tokio_read_rx.try_recv() { + self.check_forwarded_prefix(&bytes)?; + self.forwarded_seen += bytes.len(); + self.forwarded_final.extend_from_slice(&bytes); + moved += 1; + } + if moved == 0 { + // Let the pumps make progress before declaring quiet. + for _ in 0..16 { + tokio::task::yield_now().await; + } + } else { + for _ in 0..4 { + tokio::task::yield_now().await; + } + } + } + self.assert_write_model().await + } + + /// Invariant 3's mid-sequence check: the forwarded prefix (each + /// delivered span in order, never dropped or duplicated) tracks + /// the delivery model. + fn check_forwarded_prefix(&mut self, bytes: &[u8]) -> Result<(), String> { + // The span must equal the model's bytes at this offset — a + // span whose end runs past the model while the read pump is + // alive is forwarding the harness has not (yet) ordered; the + // absorb loop yields before draining, so a fully-alive pump's + // span is always within the model. The final reconciliation + // (teardown, after the joins) pins equality. + let end = usize::min(self.forwarded_seen + bytes.len(), self.delivered.len()); + assert_eq!( + self.delivered[self.forwarded_seen..end], + bytes[..end - self.forwarded_seen], + "forwarded bytes must equal the delivered bytes at their \ + offset (lossless in-order forwarding)" + ); + if self.forwarded_seen + bytes.len() > self.delivered.len() { + return Err(format!( + "forwarded bytes ran {} past the delivered model ({}) — \ + duplication or a phantom forwarding", + self.forwarded_seen + bytes.len() - self.delivered.len(), + self.delivered.len() + )); + } + Ok(()) + } + + /// Invariant 6's mid-sequence check: `emitted` is a byte-exact + /// prefix of the declared walk of `queued` (with the oversize + /// arm's frozen-prefix shape). + async fn assert_write_model(&mut self) -> Result<(), String> { + let (declared_payloads, oversize, _stop) = declared(&self.queued); + if oversize { + self.write_failed_oversize = true; + assert!( + self.emitted.len() <= declared_payloads.len(), + "an oversize failure must not extend emissions past the \ + declared pre-failure stream ({} > {})", + self.emitted.len(), + declared_payloads.len() + ); + return Ok(()); + } + assert!( + self.emitted.len() <= declared_payloads.len(), + "emitted outran the declared stream ({} > {})", + self.emitted.len(), + declared_payloads.len() + ); + assert_eq!( + self.emitted.as_slice(), + &declared_payloads[..self.emitted.len()], + "the observed emit prefix must match the declared chunk \ + stream byte-for-byte" + ); + Ok(()) + } + + async fn teardown(self) { + let mut this = self; + // Production teardown order: the inbound stream ends (read + // pump drains its final message loop → EOF), then the write + // queue's sender drops (drain → trailing sink close). + let _ = this.inbound_tx.close().await; + this.channels.write.drop_tx(); + + for _ in 0..16 { + for _ in 0..64 { + tokio::task::yield_now().await; + } + while let Ok(msg) = this.outbound_rx.try_recv() { + match msg { + fws::FuzzMsg::Binary(piece) => this.emitted.extend_from_slice(&piece), + fws::FuzzMsg::Close { code, reason } => { + let pump_invented = reason != "fuzz close"; + assert!( + !reason.is_empty() + && (matches!(code, 1001 | 1002 | 1011) + || !pump_invented), + "ws_session_opseq invariant violated: close \ + {code}/{reason} violates the WS-15 mapping" + ); + this.closes.push((code, reason)); + } + fws::FuzzMsg::Text => panic!( + "ws_session_opseq invariant violated: the write \ + pump emitted a text message" + ), + } + } + while let Ok(bytes) = this.channels.tokio_read_rx.try_recv() { + this.forwarded_final.extend_from_slice(&bytes); + } + if this.read_task.as_ref().is_some_and(|t| t.is_finished()) + && this.write_task.as_ref().is_some_and(|t| t.is_finished()) + { + break; + } + } + + let write_join = tokio::time::timeout(Duration::from_secs(5), this.write_task.take().expect("write handle present")).await; + match write_join { + Err(_) => panic!( + "ws_session_opseq invariant violated: the write pump did \ + not terminate (wedged)" + ), + Ok(Err(e)) => panic!( + "ws_session_opseq invariant violated: the write pump \ + panicked: {e}" + ), + Ok(Ok(())) => {} + } + let read_join = tokio::time::timeout(Duration::from_secs(5), this.read_task.take().expect("read handle present")).await; + match read_join { + Err(_) => panic!( + "ws_session_opseq invariant violated: the read pump did \ + not terminate (wedged)" + ), + Ok(Err(e)) => panic!( + "ws_session_opseq invariant violated: the read pump \ + panicked: {e}" + ), + Ok(Ok(())) => {} + } + for _ in 0..128 { + tokio::task::yield_now().await; + } + while let Ok(bytes) = this.channels.tokio_read_rx.try_recv() { + this.forwarded_final.extend_from_slice(&bytes); + } + + Self::assert_final(&this); + } + + fn assert_final(this: &Self) { + // Invariant 3 final: the drained forwarded bytes equal the + // delivered bytes exactly. + assert_eq!( + this.forwarded_final, this.delivered, + "forwarded bytes must reconcile with delivered bytes exactly \ + across the whole session (lossless inbound forwarding; \ + drained {}, delivered {})", + this.forwarded_final.len(), + this.delivered.len() + ); + // Invariant 4 final: the walk's completed-chunk count equals + // the declared frame count of the delivered concat. + let mut frames = 0usize; + { + let mut pos = 0usize; + let data = &this.delivered; + while data.len() - pos >= 8 { + let len = u32::from_be_bytes([ + data[pos + 4], + data[pos + 5], + data[pos + 6], + data[pos + 7], + ]); + let total = 8usize.saturating_add(len as usize); + if data.len() - pos < total { + break; + } + frames += 1; + pos += total; + } + } + assert_eq!( + this.completed_by_walk, frames as u32, + "the WS-01 walk's completed-chunk count must equal the \ + declared frame count of the delivered concatenation" + ); + // Invariant 6 final: the write stream reconciles. + let (declared_payloads, oversize, _stop) = declared(&this.queued); + if oversize { + assert!( + this.closes.iter().any(|(code, reason)| *code == 1011 + && *reason != "fuzz close"), + "an oversize header must fail the lane loudly (1011 + a \ + pump-invented reason)" + ); + } + assert_eq!( + this.emitted, declared_payloads, + "the outbound concatenation must equal the declared chunk \ + stream across the whole session (lossless write framing)" + ); + // Invariant 5: the text arm mapped to 1002 when it ran; a + // requested close observed verbatim. + assert!( + !this.closes.iter().any(|(code, reason)| *reason == "text messages not supported" + && *code != 1002), + "the text-rejection reason must carry the 1002 code" + ); + for &req in &this.requested { + if this.write_task.as_ref().is_none() { + continue; + } + assert!( + this.closes.iter().any(|(code, reason)| *code == req && *reason == "fuzz close") + || this.write_failed_oversize, + "a lane-healthy requested close ({req}) must be observed \ + verbatim" + ); + } + } +} +#[cfg(test)] +mod corpus_replay { + use super::*; + + fn decode(data: &[u8]) -> WsSequence { + let u = arbitrary::Unstructured::new(data); + arbitrary::Arbitrary::arbitrary_take_rest(u).expect("seed decodes") + } + + #[test] + fn committed_corpus_replays_through_the_invariants() { + let corpus = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../corpus/ws_session_opseq"); + let mut count = 0usize; + for entry in std::fs::read_dir(&corpus).expect("corpus directory is committed") { + let path = entry.expect("corpus entry readable").path(); + let data = std::fs::read(&path).expect("corpus entry readable"); + fuzz_ws_session_opseq(&decode(&data)); + count += 1; + } + assert!(count >= 8, "committed seed corpus is present, found {count}"); + } + + /// The hand-encoded seed bytes decode to the intended op sequences + /// (arbitrary 1.4.x layout: variant tag `(t * 7) >> 32`, fields LE + /// zero-filled, keep-going byte before each element). + #[test] + fn seed_bytes_decode_to_the_intended_sequences() { + // Fixture 1 (gen_fuzz_seeds.py): read_slots 8; a well-formed + // 4-byte-payload Write, a 4-byte-payload Deliver, a Yield. + let seq = decode(&[ + 8, 1, 0, 0, 0, 0, 0, 4, 119, 1, 74, 146, 36, 73, 0, 4, 114, 1, 220, 182, 109, 219, + 8, 0, + ]); + assert_eq!(seq.read_slots, 8); + assert_eq!(seq.ops.len(), 3); + assert!(matches!( + seq.ops[0], + WsOp::Write { + shape: 0, + len: 4, + byte: 119, + } + )); + assert!(matches!( + seq.ops[1], + WsOp::Deliver { + mshape: 0, + len: 4, + byte: 114, + } + )); + assert!(matches!(seq.ops[2], WsOp::Yield { ticks: 8 })); + + // Fixture 3: the oversize loud-gate after a clean chunk + // (read_slots 4). + let seq = decode(&[ + 4, 1, 0, 0, 0, 0, 0, 2, 113, 1, 0, 0, 0, 0, 1, 0, 0, 1, 220, 182, 109, 219, 8, 0, + ]); + assert_eq!(seq.ops.len(), 3); + assert!(matches!( + seq.ops[1], + WsOp::Write { + shape: 1, + len: 0, + byte: 0, + } + )); + assert!(matches!(seq.ops[2], WsOp::Yield { ticks: 8 })); + } + + /// Semantic fixture (alkcall §7.7 pattern): the loud-gate shape — + /// a clean chunk emitted, then an oversize header fails the lane; + /// earlier emissions stay intact and the 1011 close fires. + #[test] + fn oversize_after_clean_chunk_fails_loudly_without_corruption() { + fuzz_ws_session_opseq(&WsSequence { + read_slots: 8, + ops: vec![ + WsOp::Write { + shape: 0, + len: 4, + byte: b'a', + }, + WsOp::Yield { ticks: 8 }, + WsOp::Write { + shape: 1, + len: 0, + byte: 0, + }, + WsOp::Yield { ticks: 8 }, + ], + }); + } + + /// Semantic fixture: text rejection then the write lane keeps + /// framing queued bytes to the socket until shutdown (the lanes + /// are independent; a read-side close never corrupts the write + /// stream — the alkcall 0.8.1 class). + #[test] + fn text_rejection_does_not_corrupt_the_write_lane() { + fuzz_ws_session_opseq(&WsSequence { + read_slots: 8, + ops: vec![ + WsOp::Write { + shape: 0, + len: 5, + byte: b'k', + }, + WsOp::Deliver { + mshape: 2, + len: 0, + byte: 0, + }, + WsOp::Write { + shape: 0, + len: 3, + byte: b'z', + }, + WsOp::Yield { ticks: 16 }, + ], + }); + } + + /// Semantic fixture: peer close mid-flight, then the write lane + /// drains and the trailing sink close runs (the WS-07 shape). + #[test] + fn peer_close_then_shutdown_drains_cleanly() { + fuzz_ws_session_opseq(&WsSequence { + read_slots: 1, + ops: vec![ + WsOp::Deliver { + mshape: 0, + len: 6, + byte: b'd', + }, + WsOp::PeerClose, + WsOp::Write { + shape: 0, + len: 6, + byte: b's', + }, + WsOp::Shutdown, + WsOp::Yield { ticks: 16 }, + ], + }); + } + + /// Semantic fixture: backpressure scaled to 1 slot — delivered + /// bytes must still forward losslessly (in order, once). + #[test] + fn one_slot_backpressure_never_drops_forwarded_bytes() { + fuzz_ws_session_opseq(&WsSequence { + read_slots: 1, + ops: vec![ + WsOp::Deliver { + mshape: 0, + len: 20, + byte: b'x', + }, + WsOp::Deliver { + mshape: 0, + len: 20, + byte: b'y', + }, + WsOp::Deliver { + mshape: 0, + len: 20, + byte: b'z', + }, + WsOp::Yield { ticks: 32 }, + ], + }); + } +} diff --git a/src/client/mod.rs b/src/client/mod.rs index 7109d4e..b542041 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -10,3 +10,10 @@ pub use http_client::{ ClientCertConfig, HttpClientBuildError, HttpClientConfig, RetryConfig, SharedHttpClient, }; pub use retry_after::RetryAfterMiddleware; + +/// Fuzz-harness drive surface re-export (`fuzz/shared/src/ +/// retry_throttle.rs`, target 4): the throttle-map machinery hub from +/// the private `retry_after` module, at one named path like the other +/// fuzzing hubs. +#[cfg(feature = "client")] +pub use retry_after::fuzzing; diff --git a/src/client/retry_after.rs b/src/client/retry_after.rs index 27d2998..b2e005f 100644 --- a/src/client/retry_after.rs +++ b/src/client/retry_after.rs @@ -320,6 +320,80 @@ impl RetryAfterMiddleware { } } +/// Fuzz-harness drive surface (`fuzz/shared/src/retry_throttle.rs`, +/// target 4): the throttle map's record/evict/deadline machinery and +/// the parse/clamp helpers over attacker-shaped header values. The +/// harness constructs the middleware through the production +/// constructors and drives the record path through synthetic +/// throttled responses — the same code path the `Middleware::handle` +/// arm runs (minus the real reqwest plumbing). +#[cfg(feature = "client")] +pub mod fuzzing { + use super::{parse_retry_after_with_ceiling, RetryAfterMiddleware}; + use http::Extensions; + use reqwest::{Response, StatusCode}; + use url::Url; + + /// A middleware built through `with_capacity_ceiling_and_budget`: + /// capacity maps `capacity % 16 + 1` (1..=16), ceiling + /// `ceiling % 250 + 1` seconds (1..=250). + pub fn fuzz_middleware(capacity: u8, ceiling_secs: u16) -> RetryAfterMiddleware { + RetryAfterMiddleware::with_capacity_ceiling_and_budget( + usize::from(capacity % 16) + 1, + std::time::Duration::from_secs(u64::from(ceiling_secs % 250) + 1), + std::time::Duration::ZERO, + ) + } + + /// Build a throttled synthetic response (the shared client's wire + /// posture: 429/503 + a `Retry-After` header value). + pub fn fuzz_throttled_response(status_selector: u8, retry_after: Option) -> Response { + let status = if status_selector.is_multiple_of(2) { + StatusCode::TOO_MANY_REQUESTS + } else { + StatusCode::SERVICE_UNAVAILABLE + }; + let mut builder = http::Response::builder().status(status); + if let Some(value) = retry_after { + builder = builder.header("retry-after", value); + } + builder.body("").expect("synthetic response builds").into() + } + + /// The middleware's `record_if_throttled` arm (handle's response + /// half) over attacker-shaped header values. + pub fn fuzz_record_if_throttled(mw: &RetryAfterMiddleware, url: Url, response: &Response) { + mw.record_if_throttled(url, response, &Extensions::new()); + } + + /// The parse/clamp helper directly: header value → deadline or + /// `None` (invalid, past, or zero-deadline values all land the + /// same documented `None`). + pub fn fuzz_parse_retry_after(value: &str, ceiling_secs: u64) -> bool { + parse_retry_after_with_ceiling(value, std::time::Duration::from_secs(ceiling_secs)) + .is_some() + } + + /// Live-deadline query (the pre-request sleep gate's first read). + pub fn fuzz_deadline_exists(mw: &RetryAfterMiddleware, url: &Url) -> bool { + mw.deadline_for(url).is_some() + } + + /// Live deadline, if any, as an offset from now (the harness + /// asserts the clamp bounds on it). + pub fn fuzz_deadline_after_secs(mw: &RetryAfterMiddleware, url: &Url) -> Option { + mw.deadline_for(url) + .and_then(|d| d.duration_since(std::time::SystemTime::now()).ok()) + .map(|d| d.as_secs() as i64) + } + + /// The wake-jitter bound directly: the maximum jitter slice a + /// remaining wait of `remaining` yields. + pub fn fuzz_max_sleep_jitter(remaining: std::time::Duration) -> std::time::Duration { + super::max_sleep_jitter(remaining) + } +} + fn max_sleep_jitter(remaining: Duration) -> Duration { if remaining.is_zero() { return Duration::ZERO;