diff --git a/src/lib.rs b/src/lib.rs index 9391c3e..748cac2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -64,6 +64,23 @@ pub mod params; pub mod producer; pub mod wire; +/// The shared test helpers (review 002 N-5): the single-sourced +/// harness shapes, compiled into the lib test crate. The file lives +/// under `tests/common/` so the integration suite reaches the same +/// copy through its own `mod` declaration (`tests/harness.rs` +/// re-exports the consumed subset). `#[cfg(test)]` keeps it out of +/// the published crate; the `#[path]` resolves relative to `src/`. +#[cfg(test)] +#[path = "../tests/common/mod.rs"] +pub(crate) mod test_common; + +// The shared module is written once against the extern-crate paths +// (`alksocks::`) the integration suite resolves; inside the lib test +// crate those same paths resolve via this alias (`extern crate` +// self-linkage, the standard `mod common` companion). +#[cfg(test)] +extern crate self as alksocks; + pub use consumer::{ AssociateSession, BindDataPlane, BindReply1, BindSession, Socks5ChannelSession, SpentSessionError, TakenAssociateInner, diff --git a/src/producer.rs b/src/producer.rs index dd63f3e..3b0eff4 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -751,57 +751,23 @@ impl Socks5ProducerBuilder { #[cfg(test)] mod tests { use super::*; - use crate::egress::{BindAcceptError, BindAcceptFactoryError, SocksDialError, TargetHandle}; + use crate::test_common as common; + + use crate::egress::{BindAcceptError, BindAcceptFactoryError, SocksDialError}; use bytes::Bytes; + use common::{ + datagram_frame, decode_datagram, echo_dial_witnessed, read_reply_v4, refusing_dial, + write_request, + }; use std::pin::Pin; use std::task::{Context, Poll}; use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; - /// The test dial closure (the harness's echo shape, inline): one - /// duplex per dial; a spawned task echoes the far end. The - /// target's identity is returned to the test through the shared - /// witness so the interception-point shape is pinned. + /// The test dial closure (the common module's witnessed echo + /// shape — the target's identity is returned to the test through + /// the shared witness so the interception-point shape is pinned). fn echo_dial(targets: Arc>>) -> DialFn { - Arc::new(move |target: TargetAddr| { - let targets = Arc::clone(&targets); - Box::pin(async move { - targets.lock().await.push(target.clone()); - let (mut echo_end, target_end) = tokio::io::duplex(64 * 1024); - tokio::spawn(async move { - loop { - let mut buf = vec![0u8; 4096]; - match echo_end.read(&mut buf).await { - Ok(0) | Err(_) => return, - Ok(n) => { - if echo_end.write_all(&buf[..n]).await.is_err() { - return; - } - } - } - } - }); - let (read_half, write_half) = tokio::io::split(target_end); - Ok(TargetHandle { - read: Box::new(read_half) as Box, - write: Box::new(write_half) as Box, - }) - }) - }) - } - - fn refusing_dial(code: SocksDialError) -> DialFn { - Arc::new(move |_target: TargetAddr| { - let err = match &code { - SocksDialError::Refused(m) => SocksDialError::Refused(m.clone()), - SocksDialError::NetworkUnreachable(m) => { - SocksDialError::NetworkUnreachable(m.clone()) - } - SocksDialError::HostUnreachable(m) => SocksDialError::HostUnreachable(m.clone()), - SocksDialError::NotAllowed(m) => SocksDialError::NotAllowed(m.clone()), - SocksDialError::General(m) => SocksDialError::General(m.clone()), - }; - Box::pin(async move { Err(err) }) - }) + echo_dial_witnessed(targets) } /// Drive a raw RFC 1928 client against [`serve_socks5_over`] over @@ -835,27 +801,11 @@ mod tests { } async fn greet(client: &mut DuplexStream) { - client.write_all(&[0x05, 0x01, 0x00]).await.unwrap(); - let mut chosen = [0u8; 2]; - client.read_exact(&mut chosen).await.unwrap(); - assert_eq!(chosen, [0x05, 0x00]); - } - - async fn write_request(client: &mut DuplexStream, cmd: &Socks5Command, target: &TargetAddr) { - let addr = target.to_be_bytes().unwrap(); - let mut req = vec![0x05, cmd.as_u8(), 0x00]; - req.extend_from_slice(&addr); - client.write_all(&req).await.unwrap(); + common::greet(client).await } async fn read_reply(client: &mut DuplexStream) -> (u8, [u8; 6]) { - let mut head = [0u8; 4]; - client.read_exact(&mut head).await.unwrap(); - assert_eq!(head[0], 0x05); - assert_eq!(head[3], 0x01, "tests use v4 BND.ADDR"); - let mut addr = [0u8; 6]; - client.read_exact(&mut addr).await.unwrap(); - (head[1], addr) + read_reply_v4(client).await } /// The full CONNECT conversation: dial is invoked at the @@ -990,66 +940,11 @@ mod tests { // -- UDP ASSOCIATE: the datagram stage (ADR-003, POC #2 shape, // in-process over duplexes) ------------------------------------ - - /// The test egress (the harness's ChanEgress shape, inline): one - /// unbounded queue; `send` frames the reply inner bytes for echo - /// (`addr + data` — the egress speaks inner bytes, no length - /// prefix); `next_reply` parks on the queue (`Pending` when - /// nothing is ready — the readiness contract). - struct QueueEgress { - inbound: tokio::sync::mpsc::UnboundedSender, - outbound: tokio::sync::Mutex>, - } - - impl crate::egress::Egress for QueueEgress { - fn send( - &mut self, - addr: TargetAddr, - payload: Bytes, - ) -> futures::future::BoxFuture<'_, Result<(), crate::egress::EgressError>> { - Box::pin(async move { - let mut inner = addr.to_be_bytes().map_err(|e| { - crate::egress::EgressError::General(format!("addr encode: {e}")) - })?; - inner.extend_from_slice(&payload); - self.inbound - .send(Bytes::from(inner)) - .map_err(|_| crate::egress::EgressError::General("egress closed".into())) - }) - } - - fn next_reply( - &mut self, - ) -> futures::future::BoxFuture<'_, Result, crate::egress::EgressError>> - { - Box::pin(async move { - match self.outbound.lock().await.recv().await { - Some(inner) => Ok(Some(inner)), - None => Err(crate::egress::EgressError::General("egress closed".into())), - } - }) - } - } + // (the queue egress + its echo factory + the frame helpers ride + // the common module — the harness's single-sourced shapes.) use crate::egress::EgressError; - /// The frame a raw client sends on the datagram stage: - /// `[len: u16 BE][addr + data]`. - fn datagram_frame(addr: &TargetAddr, payload: &[u8]) -> Vec { - crate::wire::frame(addr, payload).unwrap().to_vec() - } - - /// Split the wire into (addr, payload) — the raw client's reply - /// decode (one frame per read batch, in order). - fn decode_datagram(wire: &[u8]) -> (TargetAddr, Bytes) { - let mut reader = crate::wire::DatagramReader::new(); - let mut got = None; - for dg in reader.feed(wire) { - got = Some(dg); - } - got.expect("one datagram decoded").expect("frame decodes") - } - /// The full ASSOCIATE conversation over a duplex: greet → /// associate request → sentinel reply. Returns (server handle, /// client duplex). `egress_factory_err` simulates a refusing @@ -1087,19 +982,11 @@ mod tests { (handle, client) } - /// The echo egress over the queue shape (the harness's - /// `echo_egress`, inline): datagrams sent echo back as inner - /// bytes. + /// The echo egress over the queue shape (the common module's + /// `echo_egress_factory` — the harness's `echo_egress`, same + /// shape): datagrams sent echo back as inner bytes. fn echo_egress_factory() -> EgressFactory { - Arc::new(|| { - Box::pin(async { - let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel(); - Ok(Box::new(QueueEgress { - inbound, - outbound: tokio::sync::Mutex::new(outbound), - }) as Box) - }) - }) + common::echo_egress_factory() } /// The ASSOCIATE happy path: sentinel reply → datagram @@ -2057,16 +1944,9 @@ mod tests { /// An accept factory whose future parks forever (the quiet wait's /// other exits are then deterministic — garbage/EOF win, the - /// accept never does). + /// accept never does): the common module's shape. fn parked_accept_factory() -> BindAcceptFactory { - Arc::new(move |_target: TargetAddr| { - Box::pin(async move { - Ok(BindAccept::new( - SocketAddr::from(([127, 0, 0, 1], 2121)), - std::future::pending::>(), - )) - }) - }) + common::parked_accept_factory() } /// The BIND happy path through the phase model: reply#1 carries diff --git a/tasks/implementation/fix-test-helper-consolidation.md b/tasks/implementation/fix-test-helper-consolidation.md index 3f8948b..8e820d4 100644 --- a/tasks/implementation/fix-test-helper-consolidation.md +++ b/tasks/implementation/fix-test-helper-consolidation.md @@ -1,7 +1,7 @@ --- id: implementation/fix-test-helper-consolidation name: "Tests — consolidate the duplicated harness shapes (review-002 N-5)" -status: pending +status: completed depends_on: [implementation/fix-coverage-residue] scope: narrow risk: low @@ -47,17 +47,17 @@ introduce fresh duplicates of harness shapes, fold them here too. ## Acceptance Criteria -- [ ] The shared module exists and both the lib tests and the +- [x] The shared module exists and both the lib tests and the integration suite compile against it (the `mod common` pattern; no `tests/harness.rs` import from `src/`) -- [ ] `QueueEgress`/echo-dial and the `greet`/`write_request`/ +- [x] `QueueEgress`/echo-dial and the `greet`/`write_request`/ `read_reply` trio are single-sourced (the byte-near-identical shapes) -- [ ] `HotNoneEgress`/`PureHotNoneEgress`/`Witness` remain +- [x] `HotNoneEgress`/`PureHotNoneEgress`/`Witness` remain test-local (deliberately distinct shapes — not collapsed) -- [ ] Zero behavior change: the full suite green with the same test +- [x] Zero behavior change: the full suite green with the same test count (minus any folded duplicates, recorded in the Notes) -- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean +- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean ## References @@ -67,8 +67,124 @@ introduce fresh duplicates of harness shapes, fold them here too. ## Notes -> Agent fills this during implementation. +> Filled by the implementation agent (2026-09-25): + +- **The shared module is `tests/common/mod.rs`** (314 lines), the + standard `mod common` pattern with two wiring details worth + recording: + - **The lib-test include rides `lib.rs`, not the test module**: + `src/lib.rs` declares `#[cfg(test)] #[path = + "../tests/common/mod.rs"] pub(crate) mod test_common;` plus + `#[cfg(test)] extern crate self as alksocks;`. The `#[path]` + from a nested module (`src/producer.rs`'s tests) does not + resolve — `#[path]` is relative to the containing module's + *directory* (`src/producer/tests/` for an inline child), and + `#[path]` rejects non-literal expressions (no + `concat!(env!("CARGO_MANIFEST_DIR"), ...)`), so the only clean + mount point is the crate root, where `../tests/common/mod.rs` + resolves. The `extern crate self` alias is the standard + companion: the module is written once against the extern-crate + paths (`alksocks::…`) the integration suite resolves; inside + the lib test crate the alias makes the same paths resolve. + `src/producer.rs` consumes it as `use crate::test_common as + common;` — no `tests/harness.rs` import from `src/`. + - **The integration suite reaches the same copy through + `harness`**: `tests/harness.rs` declares `#[path = + "common/mod.rs"] pub mod common;` and re-exports the consumed + subset under the names its callers already use (`echo_egress` + for `echo_egress_factory`, `udp_frame`/`udp_unframe` for + `datagram_frame`/`decode_datagram`, + `witnessing_echo_dial` for `echo_dial_witnessed`) — the + per-test `use harness::{…}` surface is unchanged, so the + consumer-* suites needed no edits at all. +- **Single-sourced shapes** (the task's named targets plus the + byte-identical stragglers the consolidation surfaced): + - The echo dial: harness `echo_dial` + `witnessing_echo_dial` + + producer's `echo_dial(targets)` → common's + `echo_dial_witnessed` + `echo_dial` (a throwaway-witness + wrapper); the producer keeps a one-line `echo_dial(targets)` + adapter since its witness arg is per-test. + - The refusing dial (all five `SocksDialError` variants + reconstructed per call) — three copies → one. + - The queue egress + echo factory: harness `ChanEgress` + + producer's `QueueEgress` (the review's mutex-wrapper-only + divergence) + `echo_egress`/`echo_egress_factory` → common's + `QueueEgress` + `echo_egress_factory` (the mutex-wrapped + shape both had converged on; `next_reply` parks on `recv()` — + the readiness contract preserved verbatim). + - The refusing egress factory → common's `refusing_egress` + (harness copy; the producer's inline + `Err(EgressError::Unreachable(…))` closures stayed — they are + one-liners, not the shape). + - The raw RFC 1928 client trio: harness `RawRfcClient`'s + `greet`/`request`/`read_reply` bodies + producer's + `greet`/`write_request`/`read_reply` → common's + `raw_greet`/`raw_request`/`raw_read_reply` (generic over + `S: AsyncRead + AsyncWrite + Unpin`, substrate-agnostic per + convention 7) + the unwrapping producer-shape wrappers + (`greet`/`write_request`/`read_reply_v4`, the v4-only reply + shape the producer tests' BND.ADDR assertions index). The + harness's `RawRfcClient` now delegates to the common trio; its + `connect()` convenience and two-ATYP-unaware API stay (its + callers use it). + - The datagram frame helpers: harness `udp_frame`/`udp_unframe` + (test module) + core-integration's `udp_frame`/`udp_unframe` + + producer's `datagram_frame`/`decode_datagram` → common's + `datagram_frame`/`decode_datagram` (re-exported under both + names). + - The parked accept factory: harness-era producer + `parked_accept_factory` + core-integration's `parked_accept` → + common's `parked_accept_factory` (bind addr `127.0.0.1:2121` + — the ECHO_BIND_PORT carrier; core-integration imports it + aliased as `parked_accept`). +- **Deliberately not collapsed** (per the task; verified + test-local): producer's `HotNoneEgress` (mutex-yielding hot + `Ok(None)`) and `PureHotNoneEgress` (zero-yield — R-1's two + distinct violation shapes), `Witness`/`FailingSendWitness` (the + drop flags), core-integration's `HotNoneEgress` (same-named but + the concurrency-load variant), and the refusing + accept-factory/accept-future pair (typed per error class; used by + the harness and consumer-bind only). Also left: harness's + `echo_accept` + producer's `bind_accept_factory` (the echo-pipe + accept shape) — the producer copy carries the per-test requested- + target witness, so the bodies differ beyond the mutex wrapper; + noted for the next touch but out of this task's named scope. +- **Zero behavior change, same test count**: 145 lib / 38 associate + / 41 bind / 40 connect / 41 integration / 30 harness / 37 + differential / 3 doc-tests — byte-identical to the coverage + task's post-N-8 population (no folded duplicates; the coverage + task's new tests were all consumer-side unit shapes with no + harness duplicates to fold). Net −307 lines (81 insertions / 388 + deletions across the four touched files + the new 314-line + module). +- **Verification**: `cargo test` + `--all-features` green (same + counts), clippy `-D warnings` clean (native + wasm), fmt clean, + `cargo doc --no-deps` clean, both wasm checks clean + (`check` + `clippy` on `wasm32-unknown-unknown`), `cargo publish + --dry-run --allow-dirty` completes. The `tests/common/` path + rides the existing `exclude = ["tasks/", …]`-style packaging + posture automatically (dev-only; `#[cfg(test)]` keeps it out of + the published lib; the integration binaries are not packaged). ## Summary -> Agent fills this on completion. \ No newline at end of file +Consolidated review 002 N-5's duplicated harness shapes into +`tests/common/mod.rs` — the standard `mod common` pattern, mounted +from `src/lib.rs` (`#[cfg(test)] #[path] pub(crate) mod test_common` ++ `extern crate self as alksocks`) for the lib test crate and from +`tests/harness.rs` (`#[path = "common/mod.rs"]` + name-preserving +re-exports) for the integration suite, so the per-test `use +harness::{…}` surface and the producer tests' local names are +unchanged. Single-sourced: the echo dial (plain + witnessing), the +refusing dial, the queue egress + its echo factory (the harness/ +producer mutex-wrapper divergence collapsed to one shape), the +refusing egress, the raw RFC 1928 client trio (generic over the +stream type; the harness's `RawRfcClient` delegates to it), the +datagram frame codec helpers (three copies → one), and the parked +accept factory. Left deliberately distinct: the hot-none egress +variants (R-1's two shapes), the drop-witness egresses, and the +typed accept-refusal factories. Zero behavior change — full suite +green with the identical test population (375 + 2 ignored doc), no +tests folded. Verification: `cargo test` + `--all-features` green, +clippy `-D warnings` clean (native + wasm), fmt clean, doc clean, +both wasm checks clean, publish dry-run completes. \ No newline at end of file diff --git a/tests/common/mod.rs b/tests/common/mod.rs new file mode 100644 index 0000000..b119875 --- /dev/null +++ b/tests/common/mod.rs @@ -0,0 +1,314 @@ +//! Shared test helpers — the harness shapes single-sourced for both +//! the `src/` unit suites and the integration suite (review 002 N-5; +//! the standard `mod common` pattern). The lib tests include this +//! file via `#[path = "../tests/common/mod.rs"]` (unit tests in `src/` +//! cannot import `tests/harness.rs` — separate crates); the +//! integration suite reaches it through `harness`'s re-exports. +//! +//! Single-sourced here (the byte-near-identical shapes): the echo +//! dial (plain + target-witnessing), the refusing dial, the queue +//! egress + its echo factory, the raw RFC 1928 client trio +//! (`greet`/`request`/`read_reply`), the datagram frame codec +//! helpers, and the parked accept factory. Deliberately NOT collapsed +//! (distinct test shapes, stay test-local): the hot-none egress +//! variants (R-1's readiness-contract violations) and the drop-witness +//! egresses. +//! +//! `allow(dead_code)` rides the harness posture: each test binary +//! (and the lib test crate) uses a different subset of the helpers. + +#![allow(dead_code)] + +use std::net::SocketAddr; +use std::sync::Arc; + +use alksocks::egress::{ + BindAccept, BindAcceptError, BindAcceptFactory, BindTargetHandle, DialFn, Egress, EgressError, + EgressFactory, SocksDialError, TargetHandle, +}; +use alksocks::rfc1928::consts; +use alksocks::rfc1928::util::target_addr::TargetAddr; +use alksocks::rfc1928::Socks5Command; +use bytes::Bytes; +use futures::future::BoxFuture; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::sync::Mutex; + +// --- The dial seam stand-ins ------------------------------------------------- + +/// An echo dial closure that records every dialed target (the +/// interception-point probe: domain form arrives unresolved — +/// ADR-012; IP-string forms arrive as parsed IP targets). One duplex +/// per dial — the "target" echoes bytes until EOF. The producer-side +/// pump gets split `DuplexStream` halves; a spawned task echoes the +/// other end. No sockets, no `local` feature. +pub fn echo_dial_witnessed(targets: Arc>>) -> DialFn { + Arc::new(move |target: TargetAddr| { + let targets = Arc::clone(&targets); + Box::pin(async move { + targets.lock().await.push(target); + let (mut echo_end, target_end) = tokio::io::duplex(64 * 1024); + tokio::spawn(async move { + loop { + let mut buf = vec![0u8; 4096]; + match echo_end.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => { + if echo_end.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + } + }); + let (read_half, write_half) = tokio::io::split(target_end); + Ok(TargetHandle { + read: Box::new(read_half) as Box, + write: Box::new(write_half) as Box, + }) + }) + }) +} + +/// An echo dial with a throwaway witness (the tests that never +/// inspect the dialed targets). +pub fn echo_dial() -> DialFn { + echo_dial_witnessed(Arc::new(Mutex::new(Vec::new()))) +} + +/// A dial that always refuses with the given [`SocksDialError`] +/// variant (the refusal-mapping tests' dial — the refusal rides +/// in-stream as the mapped RFC reply code). The variant is +/// reconstructed per call (the seam errors are not `Clone`). +pub fn refusing_dial(err: SocksDialError) -> DialFn { + Arc::new(move |_target: TargetAddr| { + let err = match &err { + SocksDialError::Refused(m) => SocksDialError::Refused(m.clone()), + SocksDialError::NetworkUnreachable(m) => SocksDialError::NetworkUnreachable(m.clone()), + SocksDialError::HostUnreachable(m) => SocksDialError::HostUnreachable(m.clone()), + SocksDialError::NotAllowed(m) => SocksDialError::NotAllowed(m.clone()), + SocksDialError::General(m) => SocksDialError::General(m.clone()), + }; + Box::pin(async move { Err(err) }) + }) +} + +// --- The egress seam stand-ins ------------------------------------------------ + +/// The queue egress behind [`echo_egress_factory`]: `send` queues the +/// inner bytes for echo (`addr + data` — the egress speaks inner +/// bytes, no length prefix); `next_reply` parks on the queue +/// (`Pending` until a datagram arrives — the readiness contract, +/// never an instantly-ready `Ok(None)`). The receiver rides a mutex +/// so `&mut self` methods can lock it (the shape both the harness's +/// and the producer's inline copies had converged on). +pub struct QueueEgress { + pub inbound: tokio::sync::mpsc::UnboundedSender, + pub outbound: Mutex>, +} + +impl Egress for QueueEgress { + fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>> { + Box::pin(async move { + let mut inner = addr + .to_be_bytes() + .map_err(|e| EgressError::General(format!("addr encode: {e}")))?; + inner.extend_from_slice(&payload); + self.inbound + .send(Bytes::from(inner)) + .map_err(|_| EgressError::General("egress closed".into())) + }) + } + + fn next_reply(&mut self) -> BoxFuture<'_, Result, EgressError>> { + Box::pin(async move { + // Real async park: `Pending` until a datagram arrives (the + // readiness contract — never an instantly-ready `Ok(None)`, + // which would starve a biased `select!` into livelock). + match self.outbound.lock().await.recv().await { + Some(inner) => Ok(Some(inner)), + // All senders dropped: end the association (an `Err`, + // never a hot `Ok(None)`). + None => Err(EgressError::General("egress closed".into())), + } + }) + } +} + +/// An echo egress factory: datagrams sent to the egress echo back +/// (the POC #2 shape). The readiness contract is honored (see +/// [`QueueEgress`]). +pub fn echo_egress_factory() -> EgressFactory { + alksocks::egress::egress_factory(|| { + let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel::(); + Ok(Box::new(QueueEgress { + inbound, + outbound: Mutex::new(outbound), + }) as Box) + }) +} + +/// A failing egress factory (the ASSOCIATE-refusal probe): refuses +/// with the given error, which the producer maps to reply#1's code. +/// The variant is reconstructed per call (the seam errors are not +/// `Clone`). +pub fn refusing_egress(err: EgressError) -> EgressFactory { + Arc::new(move || { + let err = match &err { + EgressError::Refused(m) => EgressError::Refused(m.clone()), + EgressError::Unreachable(m) => EgressError::Unreachable(m.clone()), + EgressError::General(m) => EgressError::General(m.clone()), + }; + Box::pin(async move { Err(err) }) + }) +} + +// --- The accept seam stand-ins ------------------------------------------------- + +/// A parked accept factory: the accept future never resolves (the +/// deterministic quiet-wait probe — the client-gone exit is the only +/// end this session can take). The advertised bind address matches +/// the harness's `ECHO_BIND_PORT` (2121 — the reply#1 carrier only). +pub fn parked_accept_factory() -> BindAcceptFactory { + Arc::new(move |_target: TargetAddr| { + Box::pin(async move { + Ok(BindAccept::new( + SocketAddr::from(([127, 0, 0, 1], 2121)), + std::future::pending::>(), + )) + }) + }) +} + +// --- The raw RFC 1928 client trio ---------------------------------------------- + +/// Greeting (`05 01 00` — version 5, one method, no-auth) → the +/// method-selection read. Returns the selected method byte; asserts +/// the method-selection reply's version byte is 5. +pub async fn raw_greet(stream: &mut S) -> Result +where + S: AsyncRead + AsyncWrite + Unpin, +{ + stream.write_all(&[0x05, 0x01, 0x00]).await?; + let mut chosen = [0u8; 2]; + stream.read_exact(&mut chosen).await?; + assert_eq!( + chosen[0], + consts::SOCKS5_VERSION, + "the method-selection reply is version 5" + ); + Ok(chosen[1]) +} + +/// Write a command request for `target` (the request form: +/// `VER CMD RSV ATYP DST.ADDR DST.PORT`), then flush. +pub async fn raw_request( + stream: &mut S, + cmd: &Socks5Command, + target: &TargetAddr, +) -> Result<(), std::io::Error> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let addr = target + .to_be_bytes() + .map_err(|e| std::io::Error::other(format!("target encode: {e}")))?; + let mut req = vec![consts::SOCKS5_VERSION, cmd.as_u8(), 0x00]; + req.extend_from_slice(&addr); + stream.write_all(&req).await?; + stream.flush().await +} + +/// Read one reply (the 4-byte head + the BND.ADDR per its ATYP). +/// Returns `(reply_code, atyp, addr_bytes)` — the address bytes as +/// the RFC form carries them (v4/v6: addr + port; domain: len + +/// name + port). +pub async fn raw_read_reply(stream: &mut S) -> Result<(u8, u8, Vec), std::io::Error> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let mut head = [0u8; 4]; + stream.read_exact(&mut head).await?; + assert_eq!(head[0], consts::SOCKS5_VERSION, "the reply is version 5"); + let addr_bytes: Vec = match head[3] { + consts::SOCKS5_ADDR_TYPE_IPV4 => { + let mut a = [0u8; 6]; + stream.read_exact(&mut a).await?; + a.to_vec() + } + consts::SOCKS5_ADDR_TYPE_IPV6 => { + let mut a = [0u8; 18]; + stream.read_exact(&mut a).await?; + a.to_vec() + } + consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME => { + let mut len = [0u8; 1]; + stream.read_exact(&mut len).await?; + let mut a = vec![0u8; len[0] as usize + 2]; + stream.read_exact(&mut a).await?; + let mut full = vec![len[0]]; + full.extend_from_slice(&a); + full + } + other => { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("unknown address type {other}"), + )) + } + }; + Ok((head[1], head[3], addr_bytes)) +} + +/// The producer-test trio's greet: unwraps and asserts no-auth was +/// selected. +pub async fn greet(client: &mut S) +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let method = raw_greet(client).await.unwrap(); + assert_eq!(method, consts::SOCKS5_AUTH_METHOD_NONE); +} + +/// The producer-test trio's request writer (unwrapping). +pub async fn write_request(client: &mut S, cmd: &Socks5Command, target: &TargetAddr) +where + S: AsyncRead + AsyncWrite + Unpin, +{ + raw_request(client, cmd, target).await.unwrap(); +} + +/// The producer-test trio's reply reader: the v4-only shape every +/// producer test relies on (the BND.ADDR assertions index a +/// six-byte `addr + port`). +pub async fn read_reply_v4(client: &mut S) -> (u8, [u8; 6]) +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let (code, atyp, addr) = raw_read_reply(client).await.unwrap(); + assert_eq!(atyp, consts::SOCKS5_ADDR_TYPE_IPV4, "tests use v4 BND.ADDR"); + let mut out = [0u8; 6]; + out.copy_from_slice(&addr); + (code, out) +} + +// --- The datagram frame helpers ------------------------------------------------- + +/// Frame one datagram the way the in-band wire carries it +/// (`[len: u16 BE][addr + data]` — the wire codec's shape). +pub fn datagram_frame(addr: &TargetAddr, payload: &[u8]) -> Vec { + alksocks::wire::frame(addr, payload) + .expect("frame") + .to_vec() +} + +/// Decode a datagram from the wire (one frame per read batch, in +/// order). +pub fn decode_datagram(wire: &[u8]) -> (TargetAddr, Bytes) { + let mut reader = alksocks::wire::DatagramReader::new(); + let mut got = None; + for dg in reader.feed(wire) { + got = Some(dg); + } + got.expect("one datagram decoded").expect("frame decodes") +} diff --git a/tests/core-integration.rs b/tests/core-integration.rs index b5e56d9..ebe5102 100644 --- a/tests/core-integration.rs +++ b/tests/core-integration.rs @@ -40,10 +40,7 @@ use std::sync::Arc; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use alksocks::consumer::{AssociateSession, BindSession, Socks5ChannelSession}; -use alksocks::egress::{ - BindAccept, BindAcceptError, BindAcceptFactory, BindTargetHandle, Egress, EgressError, - EgressFactory, -}; +use alksocks::egress::{Egress, EgressError, EgressFactory}; use alksocks::error::SocksSessionError; use alksocks::params::Socks5Params; use alksocks::rfc1928::consts; @@ -248,39 +245,15 @@ fn bind_fut<'a>(topo: &'a Topology, host: String, tag: &'static [u8]) -> RoundTr Box::pin(bind_round_trip(topo, host, tag)) } -// --- The ASSOCIATE wire-frame helpers (the harness's shape) ------------------ +// --- The ASSOCIATE wire-frame helpers (the common module's shapes) ----------- -fn udp_frame(addr: &TargetAddr, payload: &[u8]) -> Vec { - alksocks::wire::frame(addr, payload) - .expect("frame") - .to_vec() -} - -fn udp_unframe(wire: &[u8]) -> (TargetAddr, bytes::Bytes) { - let mut reader = alksocks::wire::DatagramReader::new(); - let mut got = None; - for dg in reader.feed(wire) { - got = Some(dg); - } - got.expect("one datagram").expect("frame decodes") -} +use harness::{parked_accept_factory as parked_accept, udp_frame, udp_unframe}; // --- Seam stand-ins ---------------------------------------------------------- -/// A parked accept factory: the accept future never resolves (the -/// deterministic quiet-wait probe — the client-gone exit is the only -/// end this session can take). -fn parked_accept() -> BindAcceptFactory { - Arc::new(move |_target: TargetAddr| { - Box::pin(async move { - let bind_addr = SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT)); - Ok(BindAccept::new( - bind_addr, - std::future::pending::>(), - )) - }) - }) -} +// The parked accept factory rides the common module (imported as +// `parked_accept` above — the deterministic quiet-wait probe; the +// advertised bind address is the common module's 2121). /// The hot egress (the readiness contract's violator, the /// producer-associate unit test's shape): `next_reply` drains one diff --git a/tests/harness.rs b/tests/harness.rs index e669a88..fdf5e09 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -49,8 +49,8 @@ use alkcall::protocol::dispatch::Dispatcher; use alkcall::registry::registration::OperationRegistry; use alksocks::egress::{ - BindAccept, BindAcceptError, BindAcceptFactory, BindTargetHandle, DialFn, Egress, EgressError, - EgressFactory, SocksDialError, TargetHandle, + BindAccept, BindAcceptError, BindAcceptFactory, BindTargetHandle, DialFn, EgressError, + EgressFactory, SocksDialError, }; use alksocks::params::{ socks5_open_spec, OP_SOCKS5_OPEN, SOCKS5_ALPN, SOCKS5_BIND_SCOPE, SOCKS5_OPEN_SCOPE, @@ -61,6 +61,28 @@ use alksocks::rfc1928::util::target_addr::TargetAddr; use alksocks::rfc1928::{ReplyError, Socks5Command, SocksError}; use alksocks::Socks5Params; +// --- Shared helpers (the single-sourced shapes; review 002 N-5) ----------- +// +// The dial/egress/accept stand-ins and the raw RFC 1928 client trio +// live in `tests/common/mod.rs` — the same file the lib test crate +// includes via `#[path]` (unit tests in `src/` cannot import +// `tests/harness.rs`, but they can share this module). The harness +// declares the module (path-included from `tests/`) and re-exports +// the shapes its tests consume so the per-test `use harness::{...}` +// surface is unchanged. + +#[path = "common/mod.rs"] +#[allow(unused_imports)] +pub mod common; + +#[allow(unused_imports)] +pub use common::{ + datagram_frame as udp_frame, decode_datagram as udp_unframe, echo_dial, + echo_dial_witnessed as witnessing_echo_dial, echo_egress_factory as echo_egress, + parked_accept_factory, raw_greet, raw_read_reply, raw_request, read_reply_v4, refusing_dial, + refusing_egress, write_request, QueueEgress, +}; + // --- Identities ----------------------------------------------------------- /// The consumer identity the transport carries: `socks5:open` only @@ -123,128 +145,13 @@ impl IdentityProvider for TestIdProvider { } // --- In-process seam closures --------------------------------------------- - -/// An echo dial closure (the ADR-004 injection point under test): a -/// duplex per dial — the "target" echoes bytes until EOF. The -/// producer-side pump gets split `DuplexStream` halves; a spawned -/// task echoes the other end. No sockets, no `local` feature. -pub fn echo_dial() -> DialFn { - Arc::new(move |_target: TargetAddr| { - Box::pin(async { - let (mut echo_end, target_end) = tokio::io::duplex(64 * 1024); - tokio::spawn(async move { - loop { - let mut buf = vec![0u8; 4096]; - match echo_end.read(&mut buf).await { - Ok(0) | Err(_) => return, - Ok(n) => { - if echo_end.write_all(&buf[..n]).await.is_err() { - return; - } - } - } - } - }); - let (read_half, write_half) = tokio::io::split(target_end); - Ok(TargetHandle { - read: Box::new(read_half), - write: Box::new(write_half), - }) - }) - }) -} - -/// A dial that always refuses with the given [`SocksDialError`] -/// variant (the refusal-mapping tests' dial — the refusal rides -/// in-stream as the mapped RFC reply code). The variant is -/// reconstructed per call (the seam errors are not `Clone`). -pub fn refusing_dial(err: SocksDialError) -> DialFn { - Arc::new(move |_target: TargetAddr| { - let err = match &err { - SocksDialError::Refused(m) => SocksDialError::Refused(m.clone()), - SocksDialError::NetworkUnreachable(m) => SocksDialError::NetworkUnreachable(m.clone()), - SocksDialError::HostUnreachable(m) => SocksDialError::HostUnreachable(m.clone()), - SocksDialError::NotAllowed(m) => SocksDialError::NotAllowed(m.clone()), - SocksDialError::General(m) => SocksDialError::General(m.clone()), - }; - Box::pin(async move { Err(err) }) - }) -} - -/// An echo egress: datagrams sent to it echo back (the POC #2 -/// shape). The readiness contract is honored: `next_reply` parks on -/// the queue's `recv()` — `Pending` when nothing is ready — never an -/// instantly-ready `Ok(None)`. -pub fn echo_egress() -> EgressFactory { - alksocks::egress::egress_factory(|| { - let (tx, rx) = tokio::sync::mpsc::unbounded_channel::(); - Ok(Box::new(ChanEgress { - tx, - rx: Mutex::new(rx), - }) as Box) - }) -} - -use tokio::sync::Mutex; - -/// The in-process egress behind [`echo_egress`]: `send` queues the -/// inner bytes for echo; `next_reply` parks on the queue (`Pending` -/// until a datagram arrives — the readiness contract, harness-modeled). -struct ChanEgress { - tx: tokio::sync::mpsc::UnboundedSender, - rx: Mutex>, -} - -impl Egress for ChanEgress { - fn send( - &mut self, - addr: TargetAddr, - payload: Bytes, - ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> { - Box::pin(async move { - // Echo: the reply's inner bytes are `addr + data` — - // exactly what `next_reply` returns (the wire layer's - // length prefix is applied exactly once, by the frame - // writer, never here). - let mut inner = addr - .to_be_bytes() - .map_err(|e| EgressError::General(format!("addr encode: {e}")))?; - inner.extend_from_slice(&payload); - self.tx - .send(Bytes::from(inner)) - .map_err(|_| EgressError::General("egress closed".into())) - }) - } - - fn next_reply(&mut self) -> futures::future::BoxFuture<'_, Result, EgressError>> { - Box::pin(async move { - // Real async park: `Pending` until a datagram arrives (the - // readiness contract — never an instantly-ready `Ok(None)`, - // which would starve a biased `select!` into livelock). - match self.rx.lock().await.recv().await { - Some(inner) => Ok(Some(inner)), - // All senders dropped: end the association (an `Err`, - // never a hot `Ok(None)`). - None => Err(EgressError::General("egress closed".into())), - } - }) - } -} - -/// A failing egress factory (the ASSOCIATE-refusal probe): refuses -/// with the given error, which the producer maps to reply#1's code. -/// The variant is reconstructed per call (the seam errors are not -/// `Clone`). -pub fn refusing_egress(err: EgressError) -> EgressFactory { - Arc::new(move || { - let err = match &err { - EgressError::Refused(m) => EgressError::Refused(m.clone()), - EgressError::Unreachable(m) => EgressError::Unreachable(m.clone()), - EgressError::General(m) => EgressError::General(m.clone()), - }; - Box::pin(async move { Err(err) }) - }) -} +// +// The dial/egress/accept stand-ins are single-sourced in +// `tests/common/mod.rs` (declared + re-exported above): `echo_dial`, +// `refusing_dial`, `echo_egress` (the queue egress), and +// `parked_accept_factory`. The shapes below are the ones deliberately +// different per suite: the accept factories (echo + refusing) and the +// producer-side registrations. /// The advertised BND.ADDR of [`echo_accept`] (in-process, no real /// listener — the address is the reply#1 carrier only). @@ -860,37 +767,6 @@ pub async fn wire_consumer_side(dial: DialFn) -> ConsumerSideTopology { wire_consumer_side_with(dial, None, None, Some(consumer_identity())).await } -/// An echo dial that records every dialed target (the -/// interception-point probe: domain form arrives unresolved — -/// ADR-012; IP-string forms arrive as parsed IP targets). -pub fn witnessing_echo_dial(targets: Arc>>) -> DialFn { - Arc::new(move |target: TargetAddr| { - let targets = Arc::clone(&targets); - Box::pin(async move { - targets.lock().await.push(target); - let (mut echo_end, target_end) = tokio::io::duplex(64 * 1024); - tokio::spawn(async move { - loop { - let mut buf = vec![0u8; 4096]; - match echo_end.read(&mut buf).await { - Ok(0) | Err(_) => return, - Ok(n) => { - if echo_end.write_all(&buf[..n]).await.is_err() { - return; - } - } - } - } - }); - let (read_half, write_half) = tokio::io::split(target_end); - Ok(TargetHandle { - read: Box::new(read_half), - write: Box::new(write_half), - }) - }) - }) -} - // --- The raw RFC 1928 client helper ----------------------------------------- /// A hand-rolled minimal RFC 1928 client over any @@ -916,10 +792,7 @@ where /// method-selection read (`05 00`). Returns the selected method /// byte. pub async fn greet(&mut self) -> Result { - self.stream.write_all(&[0x05, 0x01, 0x00]).await?; - let mut chosen = [0u8; 2]; - self.stream.read_exact(&mut chosen).await?; - Ok(chosen[1]) + raw_greet(&mut self.stream).await } /// Write a command request for `target` (the request form: @@ -929,49 +802,15 @@ where cmd: Socks5Command, target: &TargetAddr, ) -> Result<(), std::io::Error> { - let addr = target - .to_be_bytes() - .map_err(|e| std::io::Error::other(format!("target encode: {e}")))?; - let mut req = vec![consts::SOCKS5_VERSION, cmd.as_u8(), 0x00]; - req.extend_from_slice(&addr); - self.stream.write_all(&req).await?; - self.stream.flush().await + raw_request(&mut self.stream, &cmd, target).await } /// Read one reply (the 4-byte head + the BND.ADDR per its ATYP). /// Returns `(reply_code, bnd_addr_bytes)` — the address bytes as /// the RFC form carries them (ATYP + addr + port). pub async fn read_reply(&mut self) -> Result<(u8, Vec), std::io::Error> { - let mut head = [0u8; 4]; - self.stream.read_exact(&mut head).await?; - let addr_bytes: Vec = match head[3] { - consts::SOCKS5_ADDR_TYPE_IPV4 => { - let mut a = [0u8; 6]; - self.stream.read_exact(&mut a).await?; - a.to_vec() - } - consts::SOCKS5_ADDR_TYPE_IPV6 => { - let mut a = [0u8; 18]; - self.stream.read_exact(&mut a).await?; - a.to_vec() - } - consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME => { - let mut len = [0u8; 1]; - self.stream.read_exact(&mut len).await?; - let mut a = vec![0u8; len[0] as usize + 2]; - self.stream.read_exact(&mut a).await?; - let mut full = vec![len[0]]; - full.extend_from_slice(&a); - full - } - other => { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidData, - format!("unknown address type {other}"), - )) - } - }; - Ok((head[1], addr_bytes)) + let (code, _atyp, addr) = raw_read_reply(&mut self.stream).await?; + Ok((code, addr)) } /// The full CONNECT round-trip: greet, request, reply. Returns @@ -1586,24 +1425,8 @@ mod tests { } // -- the producer ASSOCIATE datagram stage over channels ------------- - - /// Frame one datagram the way the in-band wire carries it - /// (`[len: u16 BE][addr + data]` — the wire codec's shape). - fn udp_frame(addr: &TargetAddr, payload: &[u8]) -> Vec { - alksocks::wire::frame(addr, payload) - .expect("frame") - .to_vec() - } - - /// Decode a datagram from the wire (one frame; in-order batches). - fn udp_unframe(wire: &[u8]) -> (TargetAddr, Bytes) { - let mut reader = alksocks::wire::DatagramReader::new(); - let mut got = None; - for dg in reader.feed(wire) { - got = Some(dg); - } - got.expect("one datagram").expect("frame decodes") - } + // (the frame helpers ride the common module: `udp_frame` is + // `datagram_frame`, `udp_unframe` is `decode_datagram`.) /// The ASSOCIATE conversation over a channel: open + adopt + /// pump, greet, request, reply — asserting the sentinel