From 76717f3944f8619d8dbc363d7560986aaf0b96c5 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Wed, 23 Sep 2026 23:50:01 +0000 Subject: [PATCH] =?UTF-8?q?consumer:=20Socks5ChannelSession=20(CONNECT)=20?= =?UTF-8?q?=E2=80=94=20teardown=20ownership=20+=20e2e=20suite?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Socks5ChannelSession::connect: open_channel(OP_SOCKS5_OPEN, {}, SOCKS5_ALPN) then the forked client typestate's use_stream path over the channel BiStream (POC #1's fit); post-reply the session is the raw byte pipe (0 B framing) - Target construction via ToTargetAddr: IP-string forms parse to IP targets, domain form passes through unresolved (ADR-012) - split() borrowed halves (SplitPipe) / into_inner() owned pipe (TakenInner — the session rides along as the teardown owner); channel_id() for correlation - Teardown ownership (alktunnels ADR-005 posture): Drop sync-reaps the adopted entry (teardown_channel is sync; Drop cannot await) + the pipe drop EOFs the producer pump structurally; close() is the graceful path (write-half shutdown, then reap); the in-stream refusal tears the channel down in connect (teardown-at-birth — no phantom channel); no Clone (compile_fail doctest) - Error surface: open failures are the typed SocksOpenError reasons (ADR-049 §4); a refused target is the RFC reply code through SocksSessionError::Rfc (consumer.md §Error surface) - Harness: the consumer-side topology (wire_consumer_side[_with] — the consumer dials with a real ChannelClient; the producer accepts with the real registration) + witnessing_echo_dial; the allow(dead_code) attr rides the alktunnels harness posture - tests/consumer-connect.rs: 9 e2e tests — echo round-trip + CF-006 witness, target forms at the dial seam (ADR-012), refused dial → typed RFC error + no phantom channel, FORBIDDEN open → typed open error + no session, drop teardown + connection health, half-close EOF semantics, into_inner pairing, Send shape Verified: cargo test (219 green incl. the compile_fail doctest), clippy --all-targets -D warnings, fmt --check, check/clippy --target wasm32-unknown-unknown --- src/consumer.rs | 241 +++++++++- src/lib.rs | 1 + tasks/implementation/core-consumer-connect.md | 49 +- tests/consumer-connect.rs | 432 ++++++++++++++++++ tests/harness.rs | 160 +++++++ 5 files changed, 876 insertions(+), 7 deletions(-) create mode 100644 tests/consumer-connect.rs diff --git a/src/consumer.rs b/src/consumer.rs index ee89dde..5cbedfa 100644 --- a/src/consumer.rs +++ b/src/consumer.rs @@ -1,5 +1,238 @@ -//! The consumer half: the channels-native session types (CONNECT / -//! ASSOCIATE / the BIND reply#2 reader) and teardown ownership. +//! The consumer half: the channels-native session types and teardown +//! ownership (consumer.md §The Channels-Native Sessions). //! -//! Stub: filled by the `implementation/core-consumer-*` tasks -//! (CONNECT first). +//! [`Socks5ChannelSession`] is the CONNECT session: it opens +//! `channels/socks5/sub` via `ChannelClient::open_channel` (trivial +//! params — ADR-001), drives the forked client typestate's +//! `use_stream` path over the channel's `BiStream` (POC #1's fit — +//! `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`). +//! +//! Target construction rides [`ToTargetAddr`]: IP-string forms parse +//! to IP targets; the domain form passes through unresolved (ADR-012 +//! — resolution is the egress's job, never the protocol's). +//! +//! Teardown ownership (alktunnels ADR-005 posture, inherited via +//! ADR-004): the session owns the adopted channel entry. On Drop it +//! reaps it (`ChannelManager::teardown_channel` — sync, best-effort; +//! `Drop` cannot await) and the data plane EOFs structurally (the +//! mux's implicit-EOF sentinel on the dropped write half). The +//! teardown-at-birth discipline covers the in-stream refusal path: +//! a refused CONNECT tears the adopted channel down before the +//! error returns — a refused session never leaves a phantom channel. +//! The session does **not** implement `Clone` (compile-asserted on +//! the struct): aliasing a session would alias its teardown. +//! +//! Error surface (consumer.md §Error surface): open failures are the +//! typed [`SocksOpenError`] reasons (ADR-049 §4 — a refused session +//! never yields a phantom channel); a refused target is an RFC reply +//! code surfaced through [`SocksSessionError::Rfc`]. + +use alkcall::channels::client::ChannelClient; +use alkcall::channels::manager::ChannelManager; +use alkcall::core::types::BiStream; +use tokio::io::{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; + +/// A completed SOCKS5 CONNECT session on the consumer side: the +/// adopted channel + the post-reply byte pipe to the target. +/// +/// The session is a handle to one channel (one session = one channel +/// = one CONNECT, convention 10). It does **not** implement `Clone`: +/// +/// ```compile_fail +/// use alksocks::consumer::Socks5ChannelSession; +/// fn assert_clone() {} +/// assert_clone::(); +/// ``` +/// +/// The inner stream is `Option` so `Drop`/`close` can empty the +/// type (the alktunnels `Drop`-impl posture); a spent session drops +/// as a teardown-only no-op. +pub struct Socks5ChannelSession { + channel_id: u32, + manager: Option, + inner: Option>, +} + +impl std::fmt::Debug for Socks5ChannelSession { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Socks5ChannelSession") + .field("channel_id", &self.channel_id) + .field("connected", &self.inner.is_some()) + .finish() + } +} + +impl Socks5ChannelSession { + /// CONNECT to `target` 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 CONNECT request in-stream over the channel's + /// `BiStream`, and returns the session owning both. + /// + /// Error surface: an open failure is [`SocksSessionError::Open`] + /// (the typed ADR-049 reasons; no channel was ever returned); a + /// refused target 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 connect( + 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?; + inner.request(Socks5Command::TCPConnect, target).await?; + Result::, SocksSessionError>::Ok(inner) + }; + + match negotiated.await { + Ok(inner) => Ok(Self { + channel_id, + manager: Some(client.manager().clone()), + inner: Some(inner), + }), + 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 + } + + /// The post-CONNECT pipe as split read/write halves (0 B framing + /// — raw bytes to/from the target). The halves borrow the + /// session: teardown stays owned (`close`/`Drop`). `None` after + /// [`Self::into_inner`] took the pipe. + pub fn split(&mut self) -> Option> { + let inner = self.inner.as_mut()?; + Some(tokio::io::split(inner)) + } + + /// Take the underlying [`Socks5Stream`] (advanced use — + /// the forked client type's socket accessors, owned-half pumps). + /// The session rides along still owning the channel's teardown: + /// drop the pair (or `close` the session) when done — dropping + /// `inner` EOFs the data plane structurally. + pub fn into_inner(mut self) -> TakenInner { + let inner = self.inner.take(); + TakenInner { + session: self, + inner: inner.expect("into_inner on a spent session"), + } + } + + /// 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 Socks5ChannelSession { + /// 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(); + } +} + +/// The owned pipe [`Socks5ChannelSession::into_inner`] yields, paired +/// with the session that still owns the channel's teardown. +pub struct TakenInner { + /// The teardown owner — reap via `close`/`Drop`. + pub session: Socks5ChannelSession, + /// The post-CONNECT pipe (the forked client type over the + /// channel's `BiStream`). + pub inner: Socks5Stream, +} + +/// The borrowed halves [`Socks5ChannelSession::split`] yields (the +/// `tokio::io::split` pair over the pipe). A half-close rides the +/// write half's [`AsyncWriteExt::shutdown`]. +pub type SplitPipe<'s> = ( + ReadHalf<&'s mut Socks5Stream>, + WriteHalf<&'s mut Socks5Stream>, +); + +#[cfg(test)] +mod tests { + use super::*; + use alkcall::channels::client::ChannelOpenError; + + fn assert_send() {} + + #[test] + fn session_shape_is_send_and_not_clone() { + assert_send::(); + assert_send::(); + } + + #[test] + fn debug_impl_names_the_channel_and_liveness() { + let session = Socks5ChannelSession { + channel_id: 7, + manager: None, + inner: None, + }; + let rendered = format!("{session:?}"); + assert!(rendered.contains("channel_id: 7"), "{rendered}"); + assert!(rendered.contains("connected: false"), "{rendered}"); + } + + #[test] + fn spent_session_reap_is_a_no_op() { + let mut session = Socks5ChannelSession { + channel_id: 9, + manager: None, + inner: None, + }; + assert!(!session.reap(), "no manager — nothing reaped"); + assert!(session.split().is_none(), "no pipe after teardown"); + } + + #[test] + fn open_errors_carry_the_typed_surface_through_connects_mapping() { + // The mapping `connect` applies to an open failure: the typed + // `SocksOpenError` vocabulary rides `SocksSessionError::Open`. + let open = SocksOpenError::Open(ChannelOpenError::MissingChannelId); + let session: SocksSessionError = open.into(); + assert!(matches!(session, SocksSessionError::Open(_))); + } +} diff --git a/src/lib.rs b/src/lib.rs index 3bb6d59..03908da 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -29,6 +29,7 @@ pub mod params; pub mod producer; pub mod wire; +pub use consumer::Socks5ChannelSession; pub use egress::{ egress_factory, BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError, BindTargetHandle, DialFn, Egress, EgressError, EgressFactory, SocksDialError, TargetHandle, diff --git a/tasks/implementation/core-consumer-connect.md b/tasks/implementation/core-consumer-connect.md index 74af9a4..7f67fb8 100644 --- a/tasks/implementation/core-consumer-connect.md +++ b/tasks/implementation/core-consumer-connect.md @@ -1,7 +1,7 @@ --- id: implementation/core-consumer-connect name: "Core — consumer half: Socks5ChannelSession (CONNECT) + e2e tests" -status: pending +status: completed depends_on: [implementation/core-producer-connect] scope: narrow risk: low @@ -70,8 +70,51 @@ consumer.rs`, adapted to the crate's params/error surface): ## Notes -> To be filled by implementation agent +> Filled by the implementation agent (2026-09-23): + +- The session shape is exactly the POC's, adapted to the crate + surface: `Socks5Stream::use_stream(bidi, None, Config::default())` + over `BiStream::from_joined(recv, send)`, then `request(TCPConnect, + target)`. `Config::default()` is the no-auth posture (the + `skip_auth` knob stays internal to the fork; the identity seam is + the open op's, CF-005/CF-006 — no RFC-facing auth on this path). +- `split()` returns borrowed halves (`tokio::io::split` over a + `&mut Socks5Stream`; the `SplitPipe` type alias), not + owned ones: the session keeps teardown ownership for every shape, + including the split-driven one. The owned path is `into_inner()` + (`TakenInner { session, inner }` — the session rides along as the + teardown owner). +- Refusal teardown: an in-stream refusal (Rfc error) tears the + adopted channel down inside `connect` before the error returns + (teardown-at-birth; the phantom-channel property, tested both + sides). The producer's own handler-exit teardown races the + consumer's — the leak-check probe polls past that race. +- Drop is the sync-reap posture (`teardown_channel` is sync; + `Drop` cannot await) + the structural EOF of the dropped pipe; the + graceful path is `close()` (write-half shutdown first, then reap). +- Harness: added the consumer-side topology (`wire_consumer_side[_with]` + — the consumer dials with a real `ChannelClient`; the producer + accepts with the real registration) plus `witnessing_echo_dial` + (the dial-witness target-form probe). The `allow(dead_code)` attr + rides the alktunnels harness posture (each test binary compiles the + harness and uses a different subset). +- No `Clone` is compile-asserted by the doctest; a runtime probe + pins `Send` (the `Clone` bound itself is the doctest's job). ## Summary -> To be filled on completion \ No newline at end of file +Implemented `Socks5ChannelSession` (CONNECT) in `src/consumer.rs` — +`connect`/`channel_id`/`split`/`into_inner`/`close` + `Drop` teardown +ownership and the `SplitPipe`/`TakenInner` companions — with the +typed error surface (`SocksOpenError` for opens, the RFC reply code +through `SocksSessionError::Rfc` for refused targets) and +teardown-at-birth on the in-stream refusal path. E2E suite +(`tests/consumer-connect.rs`, 9 tests over the new consumer-side +harness topology): echo round-trip + CF-006 witness, target forms +(IP-string parses / domain passes through — ADR-012, wire-observed +at the dial), refused dial → typed RFC error + no phantom channel, +FORBIDDEN open → typed open error + no session, drop teardown + +connection health, half-close EOF semantics, `into_inner` pairing, +and the no-`Clone`/`Send` compile shape (the `compile_fail` doctest +pins `!Clone`). Verification: `cargo test` (219 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-connect.rs b/tests/consumer-connect.rs new file mode 100644 index 0000000..ed2b381 --- /dev/null +++ b/tests/consumer-connect.rs @@ -0,0 +1,432 @@ +//! The consumer-half integration tests — `Socks5ChannelSession` +//! (CONNECT) end to end through the harness's consumer-side topology +//! (the producer's real command-dispatch path over channels; the +//! alkcall 0.8.0 establishment surface consumed from the consumer's +//! own `ChannelClient`). +//! +//! Covered: connect → echo round-trip (payload bytes both +//! directions); the dial-witness target forms (IP-string parses to an +//! IP target, domain passes through unresolved — ADR-012); refused +//! dial → the RFC reply code surfaced as a typed session error; the +//! scope-less FORBIDDEN open → typed open error, no session; drop +//! teardown (no leaked channel entries, the connection stays +//! healthy); half-close semantics (the write-half shutdown EOFs the +//! producer's pump cleanly); the compile-asserted no-`Clone` posture. + +mod harness; + +use std::sync::Arc; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use alksocks::consumer::{Socks5ChannelSession, TakenInner}; +use alksocks::egress::SocksDialError; +use alksocks::error::SocksSessionError; +use alksocks::rfc1928::consts; +use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; +use alksocks::rfc1928::SocksError as SocksErrorKind; + +use harness::{ + echo_dial, refusing_dial, wire_consumer_side, wire_consumer_side_with, ConsumerSideTopology, +}; + +/// A `SocksOpenError`-carrying open failure (the FORBIDDEN probe's +/// expectation shape). +fn expect_forbidden(err: &SocksSessionError) -> bool { + match err { + SocksSessionError::Open(open) => match open.open_ref() { + alkcall::channels::client::ChannelOpenError::CallFailed { error } => { + error.code == "FORBIDDEN" + } + _ => false, + }, + _ => false, + } +} + +/// A dial that always refuses (the refusal-riding-in-stream probe's +/// dial — the RFC reply code must surface through the session type, +/// not as an open failure). +fn refusing_session_dial() -> alksocks::egress::DialFn { + refusing_dial(SocksDialError::Refused("target down".into())) +} + +/// Both managers hold only channel 0 (the phantom-channel property, +/// asserted from both sides). The producer's entry is reaped by its +/// open-op wrapper when the pump handler exits — an await-able +/// teardown that races the consumer-side reap; this probe polls past +/// the race instead of asserting instantly. +async fn no_leaked_channels(topo: &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"); +} + +/// The happy path: `connect` → the session is a raw byte pipe → echo +/// round-trip both directions; the dial seam saw the parsed target; +/// the establisher witnessed the opener identity (CF-006). +#[tokio::test] +async fn connect_echo_round_trip_and_dial_witness() { + let topo = wire_consumer_side(echo_dial()).await; + + let mut session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("echo.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("the session connects"); + + // The `split()`-returned halves are the last use of the session; + // the tail test's alias import lives here. + session + .split() + .expect("post-reply pipe") + .1 + .write_all(b"ping") + .await + .expect("write the probe"); + let (mut read, _) = session.split().expect("post-reply pipe"); + let mut buf = [0u8; 4]; + 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"ping"); + + // The establisher witnessed the per-call opener identity. + assert_eq!( + topo.identity_witness.lock().await.as_deref(), + Some("consumer") + ); + + session.close().await; + no_leaked_channels(&topo).await; +} + +/// Target construction (ADR-012): an IP-string form parses to an IP +/// target (the dial seam receives `TargetAddr::Ip`); the domain form +/// passes through unresolved (`TargetAddr::Domain`). Both ride the +/// same happy path. +#[tokio::test] +async fn target_construction_ip_strings_parse_domains_pass_through() { + let targets = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let topo = wire_consumer_side(harness::witnessing_echo_dial(Arc::clone(&targets))).await; + + // The domain form: passes through unresolved (the dial's job). + let session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("echo.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("domain connect"); + session.close().await; + + // The IP-string form: parsed to an IP target by `ToTargetAddr`. + let session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("10.0.0.1", 8443u16).to_target_addr().unwrap(), + ) + .await + .expect("ip-string connect"); + session.close().await; + + let dialed = targets.lock().await.clone(); + assert_eq!( + dialed, + vec![ + TargetAddr::Domain("echo.test".to_owned(), 443), + TargetAddr::Ip(std::net::SocketAddr::from(([10, 0, 0, 1], 8443))) + ], + "the dial seam received the parsed target forms (ADR-012)" + ); + no_leaked_channels(&topo).await; +} + +/// A refused dial: the RFC reply code surfaces through the session +/// type (`SocksSessionError::Rfc` — the in-stream refusal, never an +/// open failure), and no phantom channel remains on either side. +#[tokio::test] +async fn refused_dial_surfaces_rfc_reply_code_no_phantom_channel() { + let topo = wire_consumer_side(refusing_session_dial()).await; + + let err = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("down.test", 80u16).to_target_addr().unwrap(), + ) + .await + .expect_err("the dial refused — the reply code rides the session error"); + + match &err { + SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => { + assert_eq!(code.as_u8(), consts::SOCKS5_REPLY_CONNECTION_REFUSED); + } + other => panic!("expected an Rfc reply-code refusal, got {other:?}"), + } + + // Teardown-at-birth: the adopted channel is reaped before the + // error returns (no phantom channel, either side). + no_leaked_channels(&topo).await; + + // The connection is healthy: a fresh connect through a working + // dial succeeds immediately. + let topo2 = wire_consumer_side(echo_dial()).await; + let session = Socks5ChannelSession::connect( + topo2.consumer.as_ref(), + ("echo.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("fresh connect after the refusal"); + session.close().await; +} + +/// A scope-less identity: the registry's FORBIDDEN open failure is +/// the typed `SocksOpenError` through `SocksSessionError::Open` — +/// no session, no phantom channel (the open gate is the boundary). +#[tokio::test] +async fn forbidden_open_is_a_typed_open_error_with_no_session() { + let bare = alkcall::core::auth::Identity { + id: "scopeless".to_string(), + scopes: vec![], + resources: Default::default(), + }; + let topo = wire_consumer_side_with(echo_dial(), None, None, Some(bare)).await; + + let err = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("gated.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect_err("a scope-less identity fails closed"); + + assert!( + expect_forbidden(&err), + "FORBIDDEN is the open error: {err:?}" + ); + no_leaked_channels(&topo).await; +} + +/// Drop teardown: dropping a live session mid-pipe reaps the adopted +/// channel entry (no leak) and the connection stays healthy — a +/// fresh connect succeeds immediately after. +#[tokio::test] +async fn drop_teardown_reaps_and_connection_stays_healthy() { + let topo = wire_consumer_side(echo_dial()).await; + + let mut session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("dropped.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("the session connects"); + session + .split() + .expect("pipe") + .1 + .write_all(b"mid") + .await + .expect("write before the drop"); + let (mut read, _) = session.split().expect("pipe"); + let mut buf = [0u8; 3]; + read.read_exact(&mut buf).await.expect("echo before drop"); + assert_eq!(&buf, b"mid"); + + // Drop mid-stream (the teardown-ownership posture). + drop(session); + no_leaked_channels(&topo).await; + + // The connection is healthy: a fresh session round-trips. + let mut fresh = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("fresh.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("fresh connect after the drop"); + fresh + .split() + .expect("pipe") + .1 + .write_all(b"again") + .await + .expect("write"); + let (mut read, _) = fresh.split().expect("pipe"); + let mut buf = [0u8; 5]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + read.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the fresh echo"); + assert_eq!(&buf, b"again"); + fresh.close().await; + no_leaked_channels(&topo).await; +} + +/// Half-close semantics: shutting the write half EOFs the producer's +/// pump cleanly (the EOF sentinel), while the read half stays alive +/// until the target's data flows. The pump's completion is the +/// producer-side evidence (the handler tracks the data plane — R-02). +#[tokio::test] +async fn half_close_write_eof_ends_the_producer_pump() { + let topo = wire_consumer_side(echo_dial()).await; + + let session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("halfclose.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("the session connects"); + let channel_id = session.channel_id(); + assert!(channel_id >= 1, "a producer-allocated id"); + assert!(topo.consumer.manager().has_channel(channel_id)); + + // Split into owned halves via `into_inner` (the advanced-use + // accessor), then half-close the write half. + let TakenInner { + session: _owner, + inner, + } = session.into_inner(); + let (mut read, mut write) = tokio::io::split(inner); + write.write_all(b"pre").await.expect("write"); + let mut buf = [0u8; 3]; + read.read_exact(&mut buf).await.expect("echo"); + assert_eq!(&buf, b"pre"); + write.shutdown().await.expect("half-close the pipe"); + + // The producer's pump observes the EOF sentinel and ends (the + // two-pump shape's shutdown-on-completion; the echo target's + // pump exits with it). The session-side observable: the read + // half drains to EOF. + drop(write); + let mut eof = [0u8; 1]; + tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + match read.read(&mut eof).await { + Ok(0) | Err(_) => break, + Ok(_) => continue, + } + } + }) + .await + .expect("the read resolves after the half-close"); + drop(read); + drop(_owner); + no_leaked_channels(&topo).await; +} + +/// `into_inner` yields the forked client type paired with the +/// teardown-owning session; the pair's `close` reaps the channel. +#[tokio::test] +async fn into_inner_pairs_the_pipe_with_the_teardown_owner() { + let topo = wire_consumer_side(echo_dial()).await; + + let session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("inner.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("the session connects"); + let channel_id = session.channel_id(); + let TakenInner { session, inner } = session.into_inner(); + assert!(topo.consumer.manager().has_channel(channel_id)); + + let (mut read, write) = tokio::io::split(inner); + let mut buf = [0u8; 1]; + // The pipe is idle; the close below reaps the entry. + let _ = tokio::time::timeout(std::time::Duration::from_millis(50), read.read(&mut buf)).await; + drop(write); + drop(read); + + let reaped = session.close().await; + assert!(reaped, "the adopted channel entry existed"); + no_leaked_channels(&topo).await; +} + +/// The no-`Clone` posture: the type is `!Clone` because its fields +/// are (the `compile_fail` doctest pins the type-level posture). +#[test] +fn session_type_is_not_clone() { + // A `Clone`-bound assertion on a non-`Clone` type cannot be a + // runtime test; the doctest is the compile-assert. This probe + // pins the Send shape the type promises. + fn assert_send() {} + assert_send::(); +} + +/// The read half stays alive across a half-close (the echo flows +/// after the write half's EOF was already seen by the producer): a +/// payload written before the shutdown round-trips, and the read +/// half observes the target's own EOF after it (the two-pump +/// half-close semantics, both directions). +#[tokio::test] +async fn both_directions_survive_across_halves() { + let topo = wire_consumer_side(echo_dial()).await; + + let mut session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("halves.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("the session connects"); + + // Write half: two writes, flush each (the chunk stream carries + // them in order). + let (_, mut write) = session.split().expect("pipe"); + write.write_all(b"one").await.expect("write one"); + write.flush().await.expect("flush"); + write.write_all(b"two").await.expect("write two"); + write.flush().await.expect("flush"); + drop(write); + + // Read half: the echo arrives in order. + let (mut read, _) = session.split().expect("pipe"); + let mut buf = [0u8; 6]; + 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"onetwo"); + + session.close().await; + no_leaked_channels(&topo).await; +} + +/// The `split()` accessor's `Option` shape: `None` after the pipe +/// was taken (the spent-session posture; the borrow checker makes +/// the direct post-`into_inner` call the only reachable shape). +#[tokio::test] +async fn close_after_into_inner_reaps() { + let topo = wire_consumer_side(echo_dial()).await; + let session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("spent.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("the session connects"); + let channel_id = session.channel_id(); + let TakenInner { session, .. } = session.into_inner(); + // The inner pipe was taken; `close` still reaps the entry. + assert!(session.close().await, "the entry existed before the close"); + no_leaked_channels(&topo).await; + assert!(channel_id >= 1); +} diff --git a/tests/harness.rs b/tests/harness.rs index 800f4f2..22d54b0 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -1,5 +1,6 @@ //! Test harness — the producer↔consumer wiring over a //! `tokio::io::duplex` carrying the channels 8-byte chunk format. +#![allow(dead_code)] //! //! Topology (the alktunnels harness shape, adapted to this crate's //! seams; the alkcall 0.8.0 establishment surface under test — this @@ -26,6 +27,8 @@ //! //! Everything is in-process: no sockets, no `local` feature. The //! harness is dev-only (`tests/`) — it stays out of the wasm build. +//! The `allow(dead_code)` rides the alktunnels posture: each test +//! binary compiles the harness and uses a different subset. use std::collections::HashMap; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; @@ -689,6 +692,163 @@ pub async fn wire(dial: DialFn) -> Topology { .await } +// --- The consumer-side (forward) topology ------------------------------------- + +/// The consumer-session topology (`Socks5ChannelSession::connect` is +/// normative for this shape): the consumer dials the transport +/// (connect side — `ChannelClient::from_connection`, a pure consumer +/// serving nothing) and the producer accepts (the adapter + an +/// install hook registering the SOCKS5 openable over the injected +/// seams on the producer's serving registry). +pub struct ConsumerSideTopology { + /// The consumer's own client (connect side, pure consumer). Its + /// manager adopted the producer-allocated channel IDs; the + /// session reaps via it. + pub consumer: Arc, + /// The producer's manager (accept side, serving) — for asserting + /// the handler-exit reaping (no leaked channel entries). + pub producer_manager: ChannelManager, + /// What the establisher's per-call auth carried (the CF-006 + /// probe — `None` until an open runs). + pub identity_witness: IdentityWitness, +} + +/// Wire the consumer-side topology over one duplex transport: +/// consumer dials (identity attached before `from_connection` — the +/// CF-005 (b) transport-identity path), producer accepts (the hook +/// attaches the resolved identity to channel 0 before the +/// `CallConnection` — the load-bearing POC #1 wiring requirement). +pub async fn wire_consumer_side_with( + dial: DialFn, + egress: Option, + accept: Option, + identity: Option, +) -> ConsumerSideTopology { + let producer_op_registry = Arc::new(OperationRegistry::new()); + let (producer_manager_tx, mut producer_manager_rx) = + tokio::sync::mpsc::channel::(1); + let identity_witness: IdentityWitness = Arc::new(tokio::sync::Mutex::new(None)); + let witness_for_hook = Arc::clone(&identity_witness); + let resolved = Arc::new(identity.unwrap_or_else(consumer_identity)); + + let resolved_for_hook = Arc::clone(&resolved); + let dial_for_hook = Arc::clone(&dial); + let install_hook: InstallChannelZero = Arc::new( + move |manager: ChannelManager, channel0_conn: CoreConnection, _auth: AuthContext| { + let registry = Arc::clone(&producer_op_registry); + let manager_tx = producer_manager_tx.clone(); + let witness = Arc::clone(&witness_for_hook); + let resolved = Arc::clone(&resolved_for_hook); + let dial = Arc::clone(&dial_for_hook); + let egress = egress.clone(); + let accept = accept.clone(); + tokio::spawn(async move { + let channel0_bidi = match channel0_conn.accept_bi().await { + Ok(s) => s, + Err(_) => return, + }; + let (writer, reader) = split_single_stream(channel0_bidi); + let _ = channel0_conn.set_identity((*resolved).clone()); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + let core = + ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy()); + let ops = ChannelOperations::with_default_policy(manager.clone()); + ops.register_on(®istry) + .expect("register channel ops on producer"); + alksocks::producer::register_socks5_openable( + &core, + ®istry, + AuthContext::anonymous(b"alk/socks5"), + Some(dial), + egress, + accept, + Some(witness), + ) + .expect("register socks5 openable (the consumer-side topology)"); + let _ = manager_tx.send(manager).await; + let dp = Dispatcher::new( + registry, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ); + dp.serve_single_stream(call_connection, reader, writer) + .await; + }) + }, + ); + + let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024); + let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None); + let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None); + consumer_conn + .set_identity((*resolved).clone()) + .expect("transport identity set once"); + + let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap)); + let producer_transport_auth = AuthContext::anonymous(b"alk/channels"); + let _adapter_task = tokio::spawn(async move { + let _ = alkcall::core::types::ProtocolHandler::handle( + &adapter, + producer_conn, + &producer_transport_auth, + ) + .await; + }); + + let consumer_client = ChannelClient::from_connection(consumer_conn) + .await + .expect("consumer channel client init"); + let producer_manager = producer_manager_rx + .recv() + .await + .expect("producer manager captured"); + + ConsumerSideTopology { + consumer: Arc::new(consumer_client), + producer_manager, + identity_witness, + } +} + +/// The default consumer-side topology: the CONNECT path over the +/// given dial, the consumer identity on the transport. +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