alkdev/alksocks
6
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tests: consolidate the duplicated harness shapes (review-002 N-5)

Browse Source 
- new tests/common/mod.rs — the single-sourced shapes, compiled into
  the lib test crate (#[cfg(test)] #[path] mod in src/lib.rs +
  extern crate self as alksocks) and the integration suite
  (tests/harness.rs declares it and re-exports under the consumed
  names; the per-test use-harness surface is unchanged)
- single-sourced: the echo dial (plain + witnessing), the refusing
  dial, the queue egress + echo factory (the harness ChanEgress /
  producer QueueEgress mutex-wrapper divergence collapsed), the
  refusing egress, the raw RFC 1928 client trio (generic over the
  stream; harness RawRfcClient delegates), the datagram frame
  codec helpers (three copies to one), the parked accept factory
- deliberately left test-local: HotNoneEgress / PureHotNoneEgress
  (R-1's two shapes), Witness / FailingSendWitness (drop flags),
  core-integration's HotNoneEgress (concurrency-load variant),
  the typed accept-refusal factories
- zero behavior change: same test population (145 lib / 38
  associate / 41 bind / 40 connect / 41 integration / 30 harness /
  37 differential / 3 doc-tests), no tests folded, net -307 lines

Verification: cargo test + --all-features green (same counts),
clippy -D warnings clean (native + wasm), fmt clean, doc clean,
both wasm checks clean, publish dry-run completes.
This commit is contained in:
glm-5.3-flash committed 2026-09-25 03:39:01 +00:00
1 parent 8dbe73d49b
commit 07b8a74241
6 files changed
+519 -396
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/lib.rs
+17
View File
@@ -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,
src/producer.rs
+21 -141
View File
@@ -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<tokio::sync::Mutex<Vec<TargetAddr>>>) -> 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<dyn AsyncRead + Send + Sync + Unpin>,
write: Box::new(write_half) as Box<dyn AsyncWrite + Send + Sync + Unpin>,
})
})
})
}
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<Bytes>,
outbound: tokio::sync::Mutex<tokio::sync::mpsc::UnboundedReceiver<Bytes>>,
}
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<Option<Bytes>, 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<u8> {
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<dyn Egress + Send>)
})
})
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::<Result<BindTargetHandle, BindAcceptError>>(),
))
})
})
common::parked_accept_factory()
}
/// The BIND happy path through the phase model: reply#1 carries
tasks/implementation/fix-test-helper-consolidation.md
+124 -8
View File
@@ -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.
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.
tests/common/mod.rs
+314
View File
@@ -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<Mutex<Vec<TargetAddr>>>) -> 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<dyn AsyncRead + Send + Sync + Unpin>,
write: Box::new(write_half) as Box<dyn AsyncWrite + Send + Sync + Unpin>,
})
})
})
}
/// 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<Bytes>,
pub outbound: Mutex<tokio::sync::mpsc::UnboundedReceiver<Bytes>>,
}
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<Option<Bytes>, 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::<Bytes>();
Ok(Box::new(QueueEgress {
inbound,
outbound: Mutex::new(outbound),
}) as Box<dyn Egress + Send>)
})
}
/// 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::<Result<BindTargetHandle, BindAcceptError>>(),
))
})
})
}
// --- 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<S>(stream: &mut S) -> Result<u8, std::io::Error>
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<S>(
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<S>(stream: &mut S) -> Result<(u8, u8, Vec<u8>), 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<u8> = 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<S>(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<S>(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<S>(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<u8> {
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")
}
tests/core-integration.rs
+6 -33
View File
@@ -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<u8> {
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::<Result<BindTargetHandle, BindAcceptError>>(),
))
})
})
}
// 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
tests/harness.rs
+37 -214
View File
@@ -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::<Bytes>();
Ok(Box::new(ChanEgress {
tx,
rx: Mutex::new(rx),
}) as Box<dyn Egress + Send>)
})
}
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<Bytes>,
rx: Mutex<tokio::sync::mpsc::UnboundedReceiver<Bytes>>,
}
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<Option<Bytes>, 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<tokio::sync::Mutex<Vec<TargetAddr>>>) -> 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<u8, std::io::Error> {
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<u8>), std::io::Error> {
let mut head = [0u8; 4];
self.stream.read_exact(&mut head).await?;
let addr_bytes: Vec<u8> = 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<u8> {
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