feat(fuzz): target 4 stateful wave — ws_session_opseq + retry_throttle
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.
This commit is contained in:
glm-5.3-flash committed
1 parent
0ae53d2d27
commit
e023d3275e
29 files changed
+1700
-3
No files matched your search
Generated
+21
@@ -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"
|
||||
|
||||
@@ -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]
|
||||
@@ -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
|
||||
|
||||
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -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);
|
||||
}
|
||||
);
|
||||
@@ -4,4 +4,4 @@ use libfuzzer_sys::fuzz_target;
|
||||
|
||||
fuzz_target!(|data: &[u8]| {
|
||||
alkhttp_fuzz_shared::spec_import::fuzz_spec_import(data);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
+194
-1
@@ -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("<I", t)
|
||||
|
||||
def u8(self, v):
|
||||
self.buf.append(v & 0xFF)
|
||||
|
||||
def u16(self, v):
|
||||
self.buf += struct.pack("<H", v & 0xFFFF)
|
||||
|
||||
def end(self):
|
||||
self.buf.append(0)
|
||||
return bytes(self.buf)
|
||||
|
||||
# Fixture 1: write/echo flow — a well-formed chunk queued, delivered
|
||||
# bytes, both drained (the happy path).
|
||||
e = WsEnc(8)
|
||||
e.op("Write"); e.u8(0); e.u8(4); e.u8(ord("w"))
|
||||
e.op("Deliver"); e.u8(0); e.u8(4); e.u8(ord("r"))
|
||||
e.op("Yield"); e.u8(8)
|
||||
write_seed("ws_session_opseq", i, e.end()); i += 1
|
||||
|
||||
# Fixture 2: chunk framing across separate writes — a complete
|
||||
# chunk, a declared-but-unfinished chunk (the pending shape),
|
||||
# another complete one.
|
||||
e = WsEnc(8)
|
||||
e.op("Write"); e.u8(0); e.u8(4); e.u8(ord("a"))
|
||||
e.op("Write"); e.u8(2); e.u8(3); e.u8(ord("p"))
|
||||
e.op("Write"); e.u8(0); e.u8(2); e.u8(ord("b"))
|
||||
e.op("Yield"); e.u8(16)
|
||||
write_seed("ws_session_opseq", i, e.end()); i += 1
|
||||
|
||||
# Fixture 3: the oversize loud-gate after a clean chunk (earlier
|
||||
# emissions intact, the 1011 close).
|
||||
e = WsEnc(4)
|
||||
e.op("Write"); e.u8(0); e.u8(2); e.u8(ord("q"))
|
||||
e.op("Write"); e.u8(1); e.u8(0); e.u8(0)
|
||||
e.op("Yield"); e.u8(8)
|
||||
write_seed("ws_session_opseq", i, e.end()); i += 1
|
||||
|
||||
# Fixture 4: text rejection (1002) then a late write (no-op on the
|
||||
# dead lane).
|
||||
e = WsEnc(8)
|
||||
e.op("Write"); e.u8(0); e.u8(2); e.u8(ord("k"))
|
||||
e.op("Deliver"); e.u8(2); e.u8(0); e.u8(0)
|
||||
e.op("Write"); e.u8(0); e.u8(2); e.u8(ord("z"))
|
||||
e.op("Yield"); e.u8(8)
|
||||
write_seed("ws_session_opseq", i, e.end()); i += 1
|
||||
|
||||
# Fixture 5: inbound read error (the frame-rejection arm).
|
||||
e = WsEnc(2)
|
||||
e.op("DeliverError")
|
||||
e.op("Yield"); e.u8(8)
|
||||
write_seed("ws_session_opseq", i, e.end()); i += 1
|
||||
|
||||
# Fixture 6: peer close, then a queued write + shutdown on the
|
||||
# surviving write lane.
|
||||
e = WsEnc(1)
|
||||
e.op("Deliver"); e.u8(0); e.u8(6); e.u8(ord("d"))
|
||||
e.op("PeerClose")
|
||||
e.op("Write"); e.u8(0); e.u8(6); e.u8(ord("s"))
|
||||
e.op("Shutdown")
|
||||
e.op("Yield"); e.u8(16)
|
||||
write_seed("ws_session_opseq", i, e.end()); i += 1
|
||||
|
||||
# Fixture 7: a requested close through the lane (unmapped code —
|
||||
# the pump maps it verbatim; absorbed with the request's reason).
|
||||
e = WsEnc(8)
|
||||
e.op("Write"); e.u8(0); e.u8(1); e.u8(ord("x"))
|
||||
e.op("RequestClose"); e.u16(1000)
|
||||
e.op("Yield"); e.u8(8)
|
||||
write_seed("ws_session_opseq", i, e.end()); i += 1
|
||||
|
||||
# Fixture 8: the canonical text-rejection shape requested by the
|
||||
# harness (code 1002).
|
||||
e = WsEnc(8)
|
||||
e.op("RequestClose"); e.u16(1002)
|
||||
e.op("Yield"); e.u8(8)
|
||||
write_seed("ws_session_opseq", i, e.end()); i += 1
|
||||
|
||||
# retry_throttle: `ThrottleSequence` = capacity byte, ceiling byte,
|
||||
# then the ops Vec (each element gated by a PRECEDING keep-going
|
||||
# byte; the terminator 0 ends it); `ThrottleOp` tag = ceiling
|
||||
# (idx * 2^32 / 6).
|
||||
TH_VARIANTS = ["Record", "Refresh", "Evict", "Query", "Rebuild", "Tick"]
|
||||
|
||||
class ThEnc:
|
||||
def __init__(self, capacity, ceiling):
|
||||
self.buf = bytearray([capacity & 0xFF, ceiling & 0xFF])
|
||||
|
||||
def op(self, name):
|
||||
idx = TH_VARIANTS.index(name)
|
||||
t = math.ceil(idx * (1 << 32) / len(TH_VARIANTS))
|
||||
self.buf.append(1)
|
||||
self.buf += struct.pack("<I", t)
|
||||
|
||||
def u8(self, v):
|
||||
self.buf.append(v & 0xFF)
|
||||
|
||||
def u16(self, v):
|
||||
self.buf += struct.pack("<H", v & 0xFFFF)
|
||||
|
||||
def end(self):
|
||||
self.buf.append(0)
|
||||
return bytes(self.buf)
|
||||
|
||||
# Fixture 1: the seconds parse + query happy path.
|
||||
e = ThEnc(4, 255)
|
||||
e.op("Record"); e.u8(0); e.u8(0); e.u16(5)
|
||||
e.op("Query"); e.u8(0)
|
||||
write_seed("retry_throttle", i, e.end()); i += 1
|
||||
|
||||
# Fixture 2: the far-future HTTP date clamps to the ceiling.
|
||||
e = ThEnc(4, 255)
|
||||
e.op("Record"); e.u8(1); e.u8(2); e.u16(0)
|
||||
e.op("Query"); e.u8(1)
|
||||
write_seed("retry_throttle", i, e.end()); i += 1
|
||||
|
||||
# Fixture 3: garbage / past-date / absent headers never record.
|
||||
e = ThEnc(4, 255)
|
||||
e.op("Record"); e.u8(2); e.u8(4); e.u16(0)
|
||||
e.op("Record"); e.u8(3); e.u8(3); e.u16(0)
|
||||
e.op("Record"); e.u8(4); e.u8(5); e.u16(0)
|
||||
e.op("Query"); e.u8(2)
|
||||
e.op("Query"); e.u8(4)
|
||||
write_seed("retry_throttle", i, e.end()); i += 1
|
||||
|
||||
# Fixture 4: fill to capacity, then a fresh URL evicts (the
|
||||
# farthest-future drop; the earliest-deadline URL survives).
|
||||
e = ThEnc(2, 255)
|
||||
e.op("Record"); e.u8(0); e.u8(0); e.u16(10)
|
||||
e.op("Record"); e.u8(1); e.u8(0); e.u16(1)
|
||||
e.op("Record"); e.u8(2); e.u8(0); e.u16(2)
|
||||
e.op("Query"); e.u8(1)
|
||||
write_seed("retry_throttle", i, e.end()); i += 1
|
||||
|
||||
# Fixture 5: the missing-header refresh is a no-op (the record
|
||||
# arm requires the header first; the entry survives).
|
||||
e = ThEnc(4, 255)
|
||||
e.op("Record"); e.u8(0); e.u8(0); e.u16(10)
|
||||
e.op("Evict"); e.u8(0)
|
||||
e.op("Query"); e.u8(0)
|
||||
write_seed("retry_throttle", i, e.end()); i += 1
|
||||
|
||||
# Fixture 6: a huge seconds value clamps (the hostile ceiling).
|
||||
e = ThEnc(4, 255)
|
||||
e.op("Record"); e.u8(0); e.u8(1); e.u16(1000)
|
||||
e.op("Query"); e.u8(0)
|
||||
write_seed("retry_throttle", i, e.end()); i += 1
|
||||
|
||||
# Fixture 7: refresh contraction — a later higher Retry-After
|
||||
# never extends the first-seen deadline.
|
||||
e = ThEnc(4, 255)
|
||||
e.op("Record"); e.u8(0); e.u8(0); e.u16(5)
|
||||
e.op("Refresh"); e.u8(0); e.u8(0); e.u16(200)
|
||||
e.op("Query"); e.u8(0)
|
||||
write_seed("retry_throttle", i, e.end()); i += 1
|
||||
|
||||
# Fixture 8: rebuild mid-sequence (capacity/ceiling switching).
|
||||
e = ThEnc(4, 255)
|
||||
e.op("Record"); e.u8(0); e.u8(0); e.u16(10)
|
||||
e.op("Rebuild"); e.u8(1); e.u8(30)
|
||||
e.op("Record"); e.u8(1); e.u8(0); e.u16(10)
|
||||
e.op("Query"); e.u8(1)
|
||||
write_seed("retry_throttle", i, e.end()); i += 1
|
||||
|
||||
return i
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
n_sse = sse_parser_seeds()
|
||||
n_ws = ws_chunk_framing_seeds()
|
||||
n_spec = spec_import_seeds()
|
||||
n_stateful = stateful_seeds()
|
||||
print(
|
||||
f"sse_parser: {n_sse} seeds, ws_chunk_framing: {n_ws} seeds, "
|
||||
f"spec_import: {n_spec} seeds"
|
||||
f"spec_import: {n_spec} seeds, stateful (ws_session_opseq + "
|
||||
f"retry_throttle): {n_stateful} seeds"
|
||||
)
|
||||
+11
-1
@@ -15,4 +15,14 @@ mcp = ["alkhttp/mcp"]
|
||||
[dependencies]
|
||||
alkhttp = { path = "../.." }
|
||||
alkcall = "0.8.1"
|
||||
serde_json = "1"
|
||||
serde_json = "1"
|
||||
bytes = "1"
|
||||
futures = "0.3"
|
||||
tokio = { version = "1", features = ["rt", "macros", "time", "sync", "io-util"], default-features = false }
|
||||
url = "2"
|
||||
http = "1"
|
||||
httpdate = "1"
|
||||
|
||||
[dependencies.arbitrary]
|
||||
version = "1"
|
||||
features = ["derive"]
|
||||
@@ -4,6 +4,8 @@
|
||||
//! against every committed corpus entry (the quinn CI pattern) without
|
||||
//! a nightly toolchain.
|
||||
|
||||
pub mod retry_throttle;
|
||||
pub mod spec_import;
|
||||
pub mod sse_parser;
|
||||
pub mod ws_chunk_framing;
|
||||
pub mod ws_session_opseq;
|
||||
@@ -0,0 +1,550 @@
|
||||
//! Invariants for the inlined `RetryAfterMiddleware`'s throttle map
|
||||
//! (`src/client/retry_after.rs`, target 4): bounded growth of the
|
||||
//! per-URL `Retry-After` memory under adversarial op sequences — the
|
||||
//! FWD-05/CLI-01 policy surfaces (capacity, eviction order, ceiling
|
||||
//! clamp, deadline refresh contraction, parse partition).
|
||||
//!
|
||||
//! Encoded invariants:
|
||||
//!
|
||||
//! 1. **No panic** — any header value, any status, any capacity ≥ 1.
|
||||
//! 2. **Capacity bound:** the map never holds more than `capacity`
|
||||
//! entries after any op (the URL→deadline map's whole DoS reason
|
||||
//! for existing).
|
||||
//! 3. **Eviction policy partition:** when a fresh URL must evict,
|
||||
//! an expired entry (if any) goes first, else the Farthest-future
|
||||
//! deadline drops. The harness's URL set is disjoint from the
|
||||
//! victim's except through the model, so the observable shape is:
|
||||
//! the victim URL is gone and every other live entry's deadline is
|
||||
//! ≤ the dropped one's.
|
||||
//! 4. **Ceiling clamp:** any recorded deadline is within the ceiling
|
||||
//! of the record moment (and in the future — a past deadline is
|
||||
//! never recorded).
|
||||
//! 5. **Refresh contraction:** re-recording a URL never extends its
|
||||
//! existing deadline (first-seen deadline wins; a `None` refresh
|
||||
//! drops the entry entirely).
|
||||
//! 6. **Parse partition:** `parse_retry_after_with_ceiling` accepts
|
||||
//! exactly `{u64 seconds}` ∪ `{valid HTTP dates in the future}`,
|
||||
//! both post-ceiling-clamp; garbage, negative seconds, empty, and
|
||||
//! past dates all yield `None` — and an accepted value is never
|
||||
//! more than the ceiling past now.
|
||||
//! 7. **Jitter bounds:** `max_sleep_jitter` is 0 for a zero wait, ≤
|
||||
//! 25% of a short wait, and capped at 2 s for a long wait.
|
||||
|
||||
use arbitrary::Arbitrary;
|
||||
use std::collections::HashMap;
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use alkhttp::client::fuzzing as rf;
|
||||
|
||||
/// One op in the adversarial throttle sequence. Committed seeds are
|
||||
/// hand-encoded against the `arbitrary` 1.4.x derive layout: variant
|
||||
/// tag = a 4-byte LE `u32` `t` with `(u64::from(t) * count) >> 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<ThrottleOp>,
|
||||
}
|
||||
|
||||
/// 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<String> {
|
||||
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<url::Url, SystemTime>,
|
||||
/// 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::<u64>() {
|
||||
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<SystemTime, String> {
|
||||
let trimmed = value.trim();
|
||||
if let Ok(secs) = trimmed.parse::<u64>() {
|
||||
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<u8> = 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<u8> =
|
||||
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)));
|
||||
}
|
||||
}
|
||||
@@ -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<WsOp>,
|
||||
}
|
||||
|
||||
fn chunk_header(id: u32, len: u32) -> Vec<u8> {
|
||||
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<u8> {
|
||||
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<u8>, 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::<Result<fws::FuzzMsg, ()>>(INBOUND_Q);
|
||||
let (outbound_tx, outbound_rx) =
|
||||
futures::channel::mpsc::channel::<fws::FuzzMsg>(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<Result<fws::FuzzMsg, ()>>,
|
||||
outbound_rx: futures::channel::mpsc::Receiver<fws::FuzzMsg>,
|
||||
channels: fws::FuzzChannels,
|
||||
read_task: Option<tokio::task::JoinHandle<()>>,
|
||||
write_task: Option<tokio::task::JoinHandle<()>>,
|
||||
/// Bytes successfully queued on the write lane, in order.
|
||||
queued: Vec<u8>,
|
||||
/// Inbound binary-message bytes delivered while the read pump was
|
||||
/// alive (the delivery model).
|
||||
delivered: Vec<u8>,
|
||||
/// 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<u8>,
|
||||
/// Observed (code, reason) closes, in order.
|
||||
closes: Vec<(u16, &'static str)>,
|
||||
/// Codes requested through the lane (verbatim mapping).
|
||||
requested: Vec<u16>,
|
||||
/// 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<u8>,
|
||||
}
|
||||
|
||||
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 },
|
||||
],
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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<String>) -> 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<i64> {
|
||||
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;
|
||||
|
||||
Reference in new issue
Block a user