From 776844e2fb6519e7f1bb1ad0b9dfdf30604e2034 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Thu, 24 Sep 2026 00:51:46 +0000 Subject: [PATCH] =?UTF-8?q?consumer:=20BindSession=20=E2=80=94=20the=20rep?= =?UTF-8?q?ly#2=20reader=20+=20e2e=20suite?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - BindSession::bind (open → no-auth → BIND → reply#1, capturing BND.ADDR) + reply1() (the BindReply1 handle) + await_inbound (the POC #5 read_reply2 — hand-read with the forked read_address codec, no second codec, ADR-011) returning BindDataPlane (into_inner / get_socket_mut + AsyncRead/AsyncWrite forwards) - Teardown ownership (the CONNECT session's posture): Drop sync-reap + structural EOF, close() graceful, teardown-at-birth on in-stream refusals, no Clone (compile_fail-asserted) - Typed error surface: open failures ride SocksSessionError::Open (ADR-049 reasons); the command gate's 0x02 and factory-failure reply#1 codes surface from bind as Rfc; accept-stage reply#2 codes surface at await_inbound - tests/consumer-bind.rs: 9 e2e tests over the harness (round-trip, data-plane accessors, gate refusal, both failure mappings, quiet-wait garbage, drop teardown + health, the CONNECT+BIND regression guard, the Send probe) Verification: cargo test (259 incl. the doctest), clippy -D warnings (native + wasm), fmt, both wasm checks, cargo doc clean. --- src/consumer.rs | 311 ++++++++++++- src/lib.rs | 2 +- tasks/implementation/core-consumer-bind.md | 84 +++- tests/consumer-bind.rs | 489 +++++++++++++++++++++ 4 files changed, 870 insertions(+), 16 deletions(-) create mode 100644 tests/consumer-bind.rs diff --git a/src/consumer.rs b/src/consumer.rs index 5cbedfa..613e735 100644 --- a/src/consumer.rs +++ b/src/consumer.rs @@ -8,9 +8,12 @@ //! `Socks5Stream` impls `AsyncRead + AsyncWrite` itself, so a session //! over a `BiStream` is the intended use, not a hack), and after the //! reply presents the raw byte pipe (0 B framing — the session -//! reads/writes the target's bytes directly). The ASSOCIATE and BIND -//! sessions land with their tasks (`core-consumer-associate` / -//! `core-consumer-bind`). +//! reads/writes the target's bytes directly). [`BindSession`] is the +//! BIND session: the reply#2 reader (consumer.md §The BIND client, +//! ADR-007) — reply#1 rides the forked client `request()`, reply#2 is +//! hand-read with the forked `read_address` codec (no second codec), +//! then the pass-through data plane. The ASSOCIATE session lands with +//! its task (`core-consumer-associate`). //! //! Target construction rides [`ToTargetAddr`]: IP-string forms parse //! to IP targets; the domain form passes through unresolved (ADR-012 @@ -35,13 +38,14 @@ use alkcall::channels::client::ChannelClient; use alkcall::channels::manager::ChannelManager; use alkcall::core::types::BiStream; -use tokio::io::{AsyncWriteExt, ReadHalf, WriteHalf}; +use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf}; use crate::error::{SocksOpenError, SocksSessionError}; use crate::params::{Socks5Params, OP_SOCKS5_OPEN, SOCKS5_ALPN}; use crate::rfc1928::client::{Config, Socks5Stream}; -use crate::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; -use crate::rfc1928::Socks5Command; +use crate::rfc1928::consts; +use crate::rfc1928::util::target_addr::{read_address, TargetAddr, ToTargetAddr}; +use crate::rfc1928::{ReplyError, Socks5Command, SocksError}; /// A completed SOCKS5 CONNECT session on the consumer side: the /// adopted channel + the post-reply byte pipe to the target. @@ -191,6 +195,301 @@ pub type SplitPipe<'s> = ( WriteHalf<&'s mut Socks5Stream>, ); +/// A completed BIND reply#1 (consumer.md §The BIND client, ADR-007's +/// client-side phase reading): the channel is open, reply#1 carried +/// BND.ADDR — the advertised address the app server dials (the one +/// real address on the channels path: the local variant's bound +/// listener; the composed variant's expected far-side address; the +/// wrapper-aware sentinel `0.0.0.0:0` on the in-band default). The +/// next phase is [`Self::await_inbound`]. +/// +/// The session owns the adopted channel — teardown stays here +/// (reap via the paired `BindSession`; this handle borrows into it). +/// No `Clone` (the compile-assert rides [`BindSession`]). +pub struct BindReply1<'s> { + inner: &'s mut Socks5Stream, + /// Reply#1's BND.ADDR — the advertised address. + pub bind_addr: TargetAddr, +} + +impl<'s> BindReply1<'s> { + /// Await reply#2 (the inbound accept) and take the pass-through + /// data plane. Reply#2 is hand-read with the forked `read_address` + /// codec — reply#1's codec applied again, no second codec (POC #5's + /// `read_reply2` sketch; ADR-011's rationale). A non-success + /// reply#2 errors with the mapped RFC reply code (accept-stage + /// failures surface here — no data plane). + pub async fn await_inbound(self) -> Result, SocksSessionError> { + let peer = read_reply2(self.inner).await?; + Ok(BindDataPlane { + inner: self.inner, + peer, + }) + } + + /// The raw stream (the front-door bridge seam: a raw pipe to a + /// vanilla client, pre-reply#2). + pub fn raw_mut(&mut self) -> &mut Socks5Stream { + self.inner + } +} + +/// The post-reply#2 data plane (the `BindDataPlane`-shaped pass +/// through): the proxied connection to the accepted inbound peer, 0 B +/// framing — reads/writes ride the channel directly. Accessors: +/// [`Self::into_inner`] (owned) / [`Self::get_socket_mut`] (borrowed; +/// the POC shape). The teardown owner is the [`BindSession`] the +/// `into_inner`/`get_socket_mut` borrows came from — drop it (or +/// `close` it) when done; dropping the data plane EOFs the pipe +/// structurally. +pub struct BindDataPlane<'s> { + inner: &'s mut Socks5Stream, + /// Reply#2's peer address (the RFC's "the address of the incoming + /// connection" — the attribution surface, in-band). + pub peer: TargetAddr, +} + +impl<'s> BindDataPlane<'s> { + /// Take the underlying [`Socks5Stream`] out of the data + /// plane (the owned pipe; the forked client type). The session + /// rides along still owning the teardown (`BindSession`'s + /// `close`/`Drop`). + pub fn into_inner(self) -> &'s mut Socks5Stream { + self.inner + } + + /// The underlying stream (the forked client type's socket + /// accessor, borrowed — the POC's `get_socket_mut` shape). + pub fn get_socket_mut(&mut self) -> &mut BiStream { + self.inner.get_socket_mut() + } +} + +impl tokio::io::AsyncRead for BindDataPlane<'_> { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + let inner = &mut *self.inner; + std::pin::Pin::new(inner).poll_read(cx, buf) + } +} + +impl tokio::io::AsyncWrite for BindDataPlane<'_> { + fn poll_write( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + let inner = &mut *self.inner; + std::pin::Pin::new(inner).poll_write(cx, buf) + } + + fn poll_flush( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + let inner = &mut *self.inner; + std::pin::Pin::new(inner).poll_flush(cx) + } + + fn poll_shutdown( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + let inner = &mut *self.inner; + std::pin::Pin::new(inner).poll_shutdown(cx) + } +} + +/// The BIND consumer session (the `bind()` entry namespace — the POC +/// #5 shape): opens the channel, negotiates, sends BIND, reads +/// reply#1 (`bind`), then awaits reply#2 + the data plane +/// (`await_inbound`). The session owns the adopted channel: on Drop +/// it reaps the entry (the CONNECT session's teardown posture). It +/// does **not** implement `Clone`: +/// +/// ```compile_fail +/// use alksocks::consumer::BindSession; +/// fn assert_clone() {} +/// assert_clone::(); +/// ``` +/// +/// Error surface: a refused BIND (the command gate's 0x02) or a +/// factory failure's reply#1 code surfaces from `bind` as +/// [`SocksSessionError::Rfc`] (the adopted channel is torn down +/// before the error returns — teardown-at-birth, the phantom-channel +/// property); accept-stage failures surface at `await_inbound` as +/// reply#2's mapped code ([`SocksSessionError::Rfc`]) — the channel +/// is still open there (the listener existed; the session lives until +/// the app closes it). +pub struct BindSession { + channel_id: u32, + manager: Option, + inner: Option>, + /// Reply#1's BND.ADDR — captured at `bind` (the forked client's + /// `request()` returns it once; the session carries it for the + /// app server's out-of-band dial). + bind_addr: Option, +} + +impl std::fmt::Debug for BindSession { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("BindSession") + .field("channel_id", &self.channel_id) + .field("awaiting_inbound", &self.inner.is_some()) + .field("bind_addr", &self.bind_addr) + .finish() + } +} + +impl BindSession { + /// BIND through a produced SOCKS5 resource: opens the channel + /// (`open_channel(OP_SOCKS5_OPEN, {}, SOCKS5_ALPN)` — the trivial + /// params, ADR-001), drives the no-auth handshake and the BIND + /// request in-stream over the channel's `BiStream`, and reads + /// reply#1 (carrying BND.ADDR — the forked client `request()`'s + /// normal path). The returned session owns the channel; the + /// advertised address is [`BindReply1::bind_addr`]. + /// + /// Error surface: an open failure is [`SocksSessionError::Open`] + /// (the typed ADR-049 reasons; no channel was ever returned); a + /// refused BIND — the command gate's `ConnectionNotAllowed` 0x02 + /// or a factory failure's mapped reply#1 code — surfaces the RFC + /// reply code through [`SocksSessionError::Rfc`], and the adopted + /// channel is torn down before the error returns (no phantom + /// channel). + pub async fn bind( + client: &ChannelClient, + target: impl ToTargetAddr, + ) -> Result { + let target: TargetAddr = target.to_target_addr()?; + let (channel_id, send, recv) = client + .open_channel(OP_SOCKS5_OPEN, Socks5Params::as_json(), SOCKS5_ALPN) + .await + .map_err(SocksOpenError::from)?; + + let bidi = BiStream::from_joined(recv, send); + let negotiated = async { + let mut inner = Socks5Stream::use_stream(bidi, None, Config::default()).await?; + let bind_addr = inner.request(Socks5Command::TCPBind, target).await?; + Result::<(TargetAddr, Socks5Stream), SocksSessionError>::Ok(( + bind_addr, inner, + )) + }; + + match negotiated.await { + Ok((bind_addr, inner)) => Ok(Self { + channel_id, + manager: Some(client.manager().clone()), + inner: Some(inner), + bind_addr: Some(bind_addr), + }), + Err(err) => { + let _ = client.manager().teardown_channel(channel_id); + Err(err) + } + } + } + + /// The adopted channel's id (observability + correlation; one + /// session = one channel). + pub fn channel_id(&self) -> u32 { + self.channel_id + } + + /// Reply#1's view: the advertised BND.ADDR (the app server dials + /// it out-of-band — the RFC's original role for the address). The + /// reply#2 reader is [`BindReply1::await_inbound`]; the returned + /// handle borrows the session (teardown stays owned). + pub fn reply1(&mut self) -> Option> { + let inner = self.inner.as_mut()?; + Some(BindReply1 { + inner, + bind_addr: self.bind_addr.clone()?, + }) + } + + /// Await the inbound accept (reply#2) and take the pass-through + /// data plane — [`BindReply1::await_inbound`] reached through the + /// session (the borrow rides the session; teardown stays owned). + /// `None` after a taken plane (the spent-session posture). + pub async fn await_inbound(&mut self) -> Result>, SocksSessionError> { + let Some(inner) = self.inner.as_mut() else { + return Ok(None); + }; + let peer = read_reply2(inner).await?; + Ok(Some(BindDataPlane { inner, peer })) + } + + /// The pass-through pipe as split read/write halves (0 B framing + /// — raw bytes to/from the accepted peer). Valid both in the + /// reply#1 phase (pre-reply#2 bytes ride the same channel — the + /// front-door bridge seam) and post-[`Self::await_inbound`] (the + /// data plane proper). `None` after a graceful `close` (the pipe + /// is dropped with the reap). + pub fn split(&mut self) -> Option> { + let inner = self.inner.as_mut()?; + Some(tokio::io::split(inner)) + } + + /// Tear the session down gracefully: half-close the pipe (the EOF + /// sentinel reaches the producer's pump), then reap the adopted + /// channel entry. Returns whether the entry existed. + pub async fn close(mut self) -> bool { + if let Some(inner) = self.inner.as_mut() { + let _ = inner.shutdown().await; + } + self.inner.take(); + self.reap() + } + + fn reap(&mut self) -> bool { + self.manager + .take() + .map(|m| m.teardown_channel(self.channel_id).is_ok()) + .unwrap_or(false) + } +} + +impl Drop for BindSession { + /// Sync-reap the adopted channel entry (`teardown_channel` is + /// sync — best-effort; `Drop` cannot await) and drop the pipe + /// (the mux's implicit-EOF sentinel EOFs the producer's pump). + /// Dropping without `close` never leaks the channel entry. + fn drop(&mut self) { + self.inner.take(); + let _ = self.reap(); + } +} + +/// Read reply#2 off the raw stream (the RFC's second reply: VER, REP, +/// RSV, ATYP, addr, port — the POC #5 `read_reply2` sketch): the 4-byte +/// head, the version pin, the reply code mapped through the forked +/// `ReplyError::from_u8`, then the address via the forked public +/// `read_address` codec — reply#1's codec applied again, **no second +/// codec** (ADR-011's rationale). +async fn read_reply2(inner: &mut Socks5Stream) -> Result +where + T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, +{ + let socket = inner.get_socket_mut(); + let mut header = [0u8; 4]; + socket.read_exact(&mut header).await?; + let [version, reply, _rsv, atyp] = header; + if version != consts::SOCKS5_VERSION { + return Err(SocksError::UnsupportedSocksVersion(version).into()); + } + if reply != consts::SOCKS5_REPLY_SUCCEEDED { + return Err(SocksError::ReplyError(ReplyError::from_u8(reply)).into()); + } + let peer = read_address(socket, atyp) + .await + .map_err(SocksError::AddrError)?; + Ok(peer) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/lib.rs b/src/lib.rs index 03908da..3824b36 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -29,7 +29,7 @@ pub mod params; pub mod producer; pub mod wire; -pub use consumer::Socks5ChannelSession; +pub use consumer::{BindDataPlane, BindReply1, BindSession, Socks5ChannelSession}; pub use egress::{ egress_factory, BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError, BindTargetHandle, DialFn, Egress, EgressError, EgressFactory, SocksDialError, TargetHandle, diff --git a/tasks/implementation/core-consumer-bind.md b/tasks/implementation/core-consumer-bind.md index 543835c..620bb55 100644 --- a/tasks/implementation/core-consumer-bind.md +++ b/tasks/implementation/core-consumer-bind.md @@ -1,7 +1,7 @@ --- id: implementation/core-consumer-bind name: "Core — consumer BIND: BindSession (reply#2 reader) + e2e tests" -status: pending +status: completed depends_on: [implementation/core-consumer-connect, implementation/core-producer-bind] scope: narrow risk: medium @@ -45,15 +45,15 @@ adapted to the crate's error surface): ## Acceptance Criteria -- [ ] `BindSession` with `bind`/`await_inbound`/data-plane accessors +- [x] `BindSession` with `bind`/`await_inbound`/data-plane accessors per the POC shape -- [ ] Reply#2 read via the forked `read_address` codec (no second +- [x] Reply#2 read via the forked `read_address` codec (no second codec) -- [ ] Typed error surface (reply-code-carrying session errors) -- [ ] Teardown ownership on Drop; no `Clone` (compile-asserted) -- [ ] E2E tests through the harness (round-trip, gate refusal, +- [x] Typed error surface (reply-code-carrying session errors) +- [x] Teardown ownership on Drop; no `Clone` (compile-asserted) +- [x] E2E tests through the harness (round-trip, gate refusal, failure mappings, regression guard) -- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean +- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean ## References @@ -65,8 +65,74 @@ adapted to the crate's error surface): ## Notes -> To be filled by implementation agent +> Filled by the implementation agent (2026-09-24): + +- The POC's `BindReply1` handle is absorbed into the teardown-owning + session (the alktunnels ADR-005 posture the CONNECT session + pinned): `BindSession` captures reply#1's BND.ADDR at `bind` (the + forked client `request()` returns it once) and exposes the phase + through borrows — `reply1()` yields the `BindReply1` handle + (`bind_addr` + `raw_mut`), `await_inbound(&mut self)` reads reply#2 + and returns the borrowed `BindDataPlane` (`Ok(None)` on a spent + session). The session stays the single teardown owner for every + shape; the POC's `(session, channel_id)` tuple and the + `BindDataPlane` ownership split collapse into one owner. +- `read_reply2` is the POC's port: 4-byte head → version pin → + `ReplyError::from_u8` → the forked public `read_address` codec (no + second codec, ADR-011). Errors ride `SocksSessionError` (the + `SocksError` conversions routed through the forked vocabulary, per + the crate's error-surface convention — `AddrError`/`ReplyError` are + wrapped into `SocksError` first because the session error type + exposes `From`, not the RFC inner enums directly). +- `BindDataPlane<'s>` borrows the session's stream (the CONNECT + session's `SplitPipe` shape, generalized): `into_inner()` yields + the `&mut Socks5Stream` (the teardown owner rides the + session), `get_socket_mut()` is the forked client's socket accessor + (the POC shape), and `AsyncRead`/`AsyncWrite` are forwarded so the + plane is a plain pipe (`pump_bidi`-shaped consumers work without + the accessors). `BiStream` is not `Sync`, so the plane pins `Send` + (runtime probe) — same shape the forked client type has. +- Teardown: Drop = sync reap + structural EOF (the CONNECT posture); + `close()` = half-close (the EOF sentinel reaches the producer's + pump) then reap. Teardown-at-birth on the in-stream refusal path + (command-gate 0x02, factory-failure reply#1): the adopted channel + is reaped before the error returns. Post-reply#1 failures + (accept-stage, quiet-wait garbage) keep the channel — reply#1 + succeeded, the listener existed; `close`/`Drop` reaps. +- The quiet-wait garbage path surfaces at `await_inbound` as an + `Io`-shaped error (the producer ends the session with no reply#2; + the consumer's reply#2 read EOFs) — the client-side mirror of the + producer's `UnexpectedUdpControlGarbage` exit, pinned by test. +- Harness reuse: no harness changes — the consumer-side topology's + `wire_consumer_side_with` + the producer-bind task's + `echo_accept`/`refusing_accept*` closures cover every test; the + parked-accept factory for the quiet-wait garbage test is + test-local (deterministic garbage-vs-accept branch). ## Summary -> To be filled on completion \ No newline at end of file +Implemented the BIND consumer session in `src/consumer.rs` (POC #5's +shape, adapted to the crate's teardown-owning session posture): +`BindSession::bind` (open → no-auth → BIND → reply#1, capturing +BND.ADDR), `reply1()` (the `BindReply1` handle — the app server's +out-of-band dial address + `raw_mut`), `await_inbound` (the POC's +`read_reply2` — hand-read with the forked `read_address` codec, no +second codec) returning the `BindDataPlane` (`into_inner`/ +`get_socket_mut` + `AsyncRead`/`AsyncWrite` forwards), and the +CONNECT session's teardown ownership (Drop sync-reap + structural +EOF, `close()` graceful, teardown-at-birth on in-stream refusals, no +`Clone` — `compile_fail`-asserted). Typed error surface throughout: +open failures are `SocksSessionError::Open` (ADR-049 reasons); the +command gate's 0x02 and factory-failure reply#1 codes surface from +`bind` as `Rfc`; accept-stage reply#2 codes surface at +`await_inbound`. E2E suite (`tests/consumer-bind.rs`, 9 tests over +the harness): the round-trip (reply#1 BND.ADDR → reply#2 peer → +pass-through echo both directions), the data-plane accessors +(`into_inner`/`get_socket_mut`/`raw_mut` on one pipe), the +command-gate refusal (typed 0x02, no phantom channel), the +factory-failure (reply#1 code) and accept-stage (reply#2 code) +mappings, quiet-wait garbage ending the session, drop teardown + +connection health, the CONNECT+BIND regression guard on one +registration, and the `Send` compile-shape probe. Verification: +`cargo test` (259 incl. the doctest), clippy `-D warnings` (all +targets), fmt, and both wasm checks clean. \ No newline at end of file diff --git a/tests/consumer-bind.rs b/tests/consumer-bind.rs new file mode 100644 index 0000000..df7fcd8 --- /dev/null +++ b/tests/consumer-bind.rs @@ -0,0 +1,489 @@ +//! The consumer BIND integration tests — `BindSession` end to end +//! through the harness's consumer-side topology (the producer's real +//! command-dispatch path over channels; the BIND phase model, ADR-007, +//! driven by the wrapper-aware client, POC #5's `read_reply2` shape). +//! +//! Covered: bind → reply#1 (BND.ADDR as the harness accept advertises) +//! → reply#2 (the peer address) → pass-through echo both directions; +//! the command-gate refusal surfaced as a typed error (a +//! `socks5:open`-only identity cannot bind); factory-failure and +//! accept-stage-failure mappings surfaced; quiet-wait garbage from the +//! client side ends the session; the three-command regression guard +//! from the consumer side. + +mod harness; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use alksocks::consumer::{BindDataPlane, BindReply1, BindSession, Socks5ChannelSession}; +use alksocks::egress::{BindAcceptError, BindAcceptFactoryError}; +use alksocks::error::SocksSessionError; +use alksocks::rfc1928::consts; +use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; +use alksocks::SocksError as SocksErrorKind; + +use harness::{ + echo_egress, refusing_accept, refusing_accept_factory, wire_consumer_side_with, ECHO_BIND_PORT, +}; + +/// Both scopes: the open gate (`socks5:open`) AND the command gate +/// (`socks5:bind`) — a BIND session needs both grants explicitly +/// (ADR-006: never implied). +fn bind_scoped() -> alkcall::core::auth::Identity { + harness::both_scopes_identity() +} + +/// The no-leaked-channels probe (the consumer-connect suite's shape, +/// re-pinned for the BIND sessions): both managers hold only channel +/// 0 — the producer's entry reaps by its open-op wrapper; the probe +/// polls past the reap race. +async fn no_leaked_channels(topo: &harness::ConsumerSideTopology) { + let deadline = std::time::Duration::from_secs(5); + let consumer_ids = topo.consumer.manager().channel_ids(); + assert!( + consumer_ids.iter().all(|&id| id == 0), + "consumer leaked channel entries: {consumer_ids:?}" + ); + tokio::time::timeout(deadline, async { + loop { + let producer_ids = topo.producer_manager.channel_ids(); + if producer_ids.iter().all(|&id| id == 0) { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("the producer's handler-exit teardown reaps the entry"); +} + +/// A session error that carries an RFC reply code (the typed +/// reply-code-carrying shape the BIND paths promise). +fn reply_code_of(err: &SocksSessionError) -> Option { + match err { + SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => Some(code.as_u8()), + _ => None, + } +} + +/// The happy path: `bind` → reply#1 carries the harness accept's +/// advertised BND.ADDR (`ECHO_BIND_PORT` — the reply#1 carrier); the +/// app server's out-of-band dial shape is the `BindReply1` handle; +/// `await_inbound` reads reply#2 (the peer address, 10.0.0.7:44444 — +/// the harness's fixed stand-in); the data plane two-pumps raw bytes +/// both directions; `close` reaps the adopted channel. +#[tokio::test] +async fn bind_round_trip_reply1_reply2_then_pass_through() { + let topo = wire_consumer_side_with( + harness::echo_dial(), + None, + Some(harness::echo_accept()), + Some(bind_scoped()), + ) + .await; + + let mut session = BindSession::bind( + topo.consumer.as_ref(), + ("bind.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect("the session binds"); + + // Reply#1: the advertised BND.ADDR (the harness accept's static + // in-process address). + let channel_id = session.channel_id(); + assert!(channel_id >= 1, "a producer-allocated id"); + assert!(topo.consumer.manager().has_channel(channel_id)); + let BindReply1 { bind_addr, .. } = session.reply1().expect("reply#1's handle"); + assert_eq!( + bind_addr, + TargetAddr::Ip(std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT))), + "reply#1 carries the advertised BND.ADDR" + ); + + // Reply#2: the accepted peer (the harness's fixed stand-in), then + // the raw pass-through data plane. + let mut plane = session + .await_inbound() + .await + .expect("reply#2 read") + .expect("the plane exists before reply#2"); + assert_eq!( + plane.peer, + TargetAddr::Ip(std::net::SocketAddr::from(([10, 0, 0, 7], 44444))), + "reply#2 carries the accepted peer" + ); + + // Phase 2: raw pass-through, both directions. + plane.write_all(b"ping").await.expect("write"); + plane.flush().await.expect("flush"); + let mut buf = [0u8; 4]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + plane.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the echo"); + assert_eq!(&buf, b"ping"); + + // Graceful teardown: half-close (the EOF sentinel) + reap. + drop(plane); + assert!(session.close().await, "the adopted channel entry existed"); + no_leaked_channels(&topo).await; +} + +/// The `BindDataPlane` accessors: `into_inner` yields the owned +/// forked client type (read/write rides it directly); the session +/// still owns the teardown — `close` after the plane is drained reaps +/// the entry. +#[tokio::test] +async fn bind_data_plane_accessors_into_inner_and_socket_mut() { + let topo = wire_consumer_side_with( + harness::echo_dial(), + None, + Some(harness::echo_accept()), + Some(bind_scoped()), + ) + .await; + + let mut session = BindSession::bind( + topo.consumer.as_ref(), + ("plane.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect("the session binds"); + session + .reply1() + .expect("reply#1's handle") + .raw_mut() + .get_socket_mut() + .write_all(b"pre") + .await + .expect("pre-reply#2 write"); + let mut plane = session + .await_inbound() + .await + .expect("reply#2 read") + .expect("the plane"); + // `get_socket_mut` is the borrowed socket accessor (the POC + // shape) — the same stream, reached through the plane. + plane + .get_socket_mut() + .write_all(b"raw") + .await + .expect("write"); + let inner = plane.into_inner(); + let mut buf = [0u8; 6]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + inner.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the echo"); + assert_eq!(&buf, b"preraw", "both write paths rode the same pipe"); + let _unused = inner; + + assert!(session.close().await, "the entry existed"); + no_leaked_channels(&topo).await; +} + +/// The command gate (ADR-006): a `socks5:open`-only identity cannot +/// bind — the refusal surfaces as a typed session error carrying the +/// RFC reply code (`ConnectionNotAllowed` 0x02), and no phantom +/// channel remains on either side. +#[tokio::test] +async fn bind_command_gate_refusal_is_a_typed_reply_code_error() { + let topo = wire_consumer_side_with( + harness::echo_dial(), + None, + Some(harness::echo_accept()), + Some(harness::consumer_identity()), + ) + .await; + + let err = BindSession::bind( + topo.consumer.as_ref(), + ("gate.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect_err("the gate refuses a socks5:open-only identity"); + + assert_eq!( + reply_code_of(&err), + Some(consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED), + "the command gate's 0x02 rides the session error: {err:?}" + ); + + // Teardown-at-birth: the adopted channel was reaped before the + // error returned (no phantom channel, either side). + no_leaked_channels(&topo).await; +} + +/// Factory failure (pre-reply#1) maps to reply#1's code, surfaced +/// through the session type — no session lives past the refusal, no +/// phantom listener. +#[tokio::test] +async fn bind_factory_failure_surfaces_reply1_code() { + let topo = wire_consumer_side_with( + harness::echo_dial(), + None, + Some(refusing_accept_factory(BindAcceptFactoryError::Refused( + "port held".into(), + ))), + Some(bind_scoped()), + ) + .await; + + let err = BindSession::bind( + topo.consumer.as_ref(), + ("factory.fail", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect_err("the factory refused"); + + assert_eq!( + reply_code_of(&err), + Some(consts::SOCKS5_REPLY_CONNECTION_REFUSED), + "the factory refusal maps to reply#1's code: {err:?}" + ); + no_leaked_channels(&topo).await; +} + +/// Accept-stage failure (post-reply#1) maps to reply#2's code — +/// surfaced at `await_inbound`, no data plane; the session's channel +/// is still owned (close/drop reaps it). +#[tokio::test] +async fn bind_accept_stage_failure_surfaces_reply2_code() { + let topo = wire_consumer_side_with( + harness::echo_dial(), + None, + Some(refusing_accept(BindAcceptError::NotAllowed( + "policy".into(), + ))), + Some(bind_scoped()), + ) + .await; + + let mut session = BindSession::bind( + topo.consumer.as_ref(), + ("accept.fail", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect("reply#1 succeeded — the factory resolved"); + + let err = match session.await_inbound().await { + Err(err) => err, + Ok(Some(_)) => panic!("the accept-stage refusal must not yield a data plane"), + Ok(None) => panic!("the plane existed before reply#2"), + }; + assert_eq!( + reply_code_of(&err), + Some(consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED), + "reply#2's mapped code surfaces at await_inbound: {err:?}" + ); + + // The channel entry stays adopted (reply#1 succeeded; the session + // owns it until the app closes it) — close reaps it. + assert!(session.close().await, "the entry existed"); + no_leaked_channels(&topo).await; +} + +/// Quiet-wait garbage from the client side: a control byte after +/// reply#1 is protocol garbage — the producer ends the session with +/// no data plane, so the consumer's `await_inbound` read EOFs (an I/O +/// error, not a reply#2). +#[tokio::test] +async fn bind_quiet_wait_garbage_from_the_client_ends_the_session() { + let parked: alksocks::egress::BindAcceptFactory = + std::sync::Arc::new(move |_target: TargetAddr| { + Box::pin(async move { + Ok(alksocks::egress::BindAccept::new( + std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT)), + std::future::pending::< + Result, + >(), + )) + }) + }); + let topo = wire_consumer_side_with( + harness::echo_dial(), + None, + Some(parked), + Some(bind_scoped()), + ) + .await; + + let mut session = BindSession::bind( + topo.consumer.as_ref(), + ("quiet.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect("reply#1 succeeded"); + + // Protocol garbage in the quiet wait (the client's side). + session + .reply1() + .expect("reply#1's handle") + .raw_mut() + .get_socket_mut() + .write_all(&[0x05, 0x02]) + .await + .expect("garbage written"); + + // The session ends: the reply#2 read EOFs (no reply#2, no data + // plane). + let err = match session.await_inbound().await { + Err(err) => err, + Ok(Some(_)) => panic!("the garbage must not yield a data plane"), + Ok(None) => panic!("the plane existed before reply#2"), + }; + assert!( + matches!(err, SocksSessionError::Io(_)), + "the quiet-wait garbage surfaces as the EOF'd reply#2 read: {err:?}" + ); + + // The channel is still owned (reply#1 succeeded); close reaps it. + assert!(session.close().await, "the entry existed"); + no_leaked_channels(&topo).await; +} + +/// The three-command regression guard from the consumer side (POC #5 +/// test 7's consumer-side mirror): one registration, all three +/// commands served through their consumer sessions — CONNECT +/// (`Socks5ChannelSession`), BIND (`BindSession`), ASSOCIATE rides +/// its task; here CONNECT + BIND coexist on one registration with the +/// both-scopes identity, fresh channel per session. +#[tokio::test] +async fn bind_and_connect_serve_on_the_same_registration() { + let topo = wire_consumer_side_with( + harness::echo_dial(), + Some(echo_egress()), + Some(harness::echo_accept()), + Some(bind_scoped()), + ) + .await; + + // CONNECT serves (the CONNECT session over the same registration). + let mut connect_session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("echo.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("the connect session"); + connect_session + .split() + .expect("pipe") + .1 + .write_all(b"tcp") + .await + .expect("write"); + let (mut read, _) = connect_session.split().expect("pipe"); + let mut buf = [0u8; 3]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + read.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the echo"); + assert_eq!(&buf, b"tcp"); + connect_session.close().await; + + // BIND serves the same registration (fresh channel). + let mut bind_session = BindSession::bind( + topo.consumer.as_ref(), + ("bind.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect("the bind session"); + let mut plane = bind_session + .await_inbound() + .await + .expect("reply#2") + .expect("the plane"); + plane.write_all(b"bind").await.expect("write"); + plane.flush().await.expect("flush"); + let mut buf = [0u8; 4]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + plane.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the echo"); + assert_eq!(&buf, b"bind"); + drop(plane); + bind_session.close().await; + no_leaked_channels(&topo).await; +} + +/// Drop teardown: dropping a live BIND session mid-pipe reaps the +/// adopted channel entry (no leak) and the connection stays healthy — +/// a fresh bind succeeds immediately after (the CONNECT session's +/// drop posture, re-pinned for BIND). +#[tokio::test] +async fn bind_drop_teardown_reaps_and_connection_stays_healthy() { + let topo = wire_consumer_side_with( + harness::echo_dial(), + None, + Some(harness::echo_accept()), + Some(bind_scoped()), + ) + .await; + + let mut session = BindSession::bind( + topo.consumer.as_ref(), + ("dropped.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect("the session binds"); + let BindReply1 { bind_addr, .. } = session.reply1().expect("reply#1's handle"); + assert_eq!( + bind_addr, + TargetAddr::Ip(std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT))) + ); + + // Drop mid-pipe (before reply#2 — the quiet-wait phase). + drop(session); + no_leaked_channels(&topo).await; + + // The connection is healthy: a fresh bind round-trips. + let mut fresh = BindSession::bind( + topo.consumer.as_ref(), + ("fresh.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect("fresh bind after the drop"); + let mut plane = fresh + .await_inbound() + .await + .expect("reply#2") + .expect("the plane"); + plane.write_all(b"again").await.expect("write"); + plane.flush().await.expect("flush"); + let mut buf = [0u8; 5]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + plane.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the fresh echo"); + assert_eq!(&buf, b"again"); + drop(plane); + fresh.close().await; + no_leaked_channels(&topo).await; +} + +/// The no-`Clone` posture for the BIND session (the CONNECT session's +/// compile shape, mirrored): the doctest on the type is the +/// compile-assert; this probe pins `Send`. +#[test] +fn bind_session_type_is_send() { + fn assert_send() {} + assert_send::(); + assert_send::>(); + assert_send::>(); +}