//! 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 //! is the crate's first consumption of it): //! //! - **producer** (connect side of the channels connection — it runs //! `ChannelClient::from_connection_with_serving`): serves the socks5 //! open op + the generic channel ops on its serving registry. Its //! manager allocates the channel IDs (`ChannelSide::Connect` — the //! serving side allocates; the initiator adopts). //! - **consumer** (accept side): the `ChannelsAdapter` + capturing //! install hook. The hook captures the consumer's channel-0 //! `CallConnection` (for calling the producer's open op) and the //! consumer's `ChannelManager` (for adopting the producer- //! allocated IDs). //! //! The POC #1 wiring requirement is load-bearing here //! (producer.md §Harness wiring requirement): the accept-side //! install hook must call `channel0_conn.set_identity(resolved)` //! before building the `CallConnection` — without it, scope-gated //! opens fail closed with `FORBIDDEN "authentication required"` //! (correct behavior, implicit wiring). The harness makes the //! requirement explicit and tested. //! //! 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}; use std::sync::Arc; use bytes::Bytes; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero}; use alkcall::channels::client::{ChannelClient, ServingConfig}; use alkcall::channels::manager::ChannelManager; use alkcall::channels::operations::{ChannelCore, ChannelOperations}; use alkcall::channels::pump::pump_bidi; use alkcall::core::auth::{AuthContext, AuthToken, Identity, IdentityProvider}; use alkcall::core::types::Connection as CoreConnection; use alkcall::protocol::connection::{split_single_stream, CallConnection}; use alkcall::protocol::dispatch::Dispatcher; use alkcall::registry::registration::OperationRegistry; use alksocks::egress::{ BindAccept, BindAcceptError, BindAcceptFactory, BindTargetHandle, DialFn, Egress, EgressError, EgressFactory, SocksDialError, TargetHandle, }; use alksocks::params::{ socks5_open_spec, OP_SOCKS5_OPEN, SOCKS5_ALPN, SOCKS5_BIND_SCOPE, SOCKS5_OPEN_SCOPE, }; use alksocks::rfc1928::consts; use alksocks::rfc1928::server::{new_reply, Socks5ServerProtocol, SocksServerError}; use alksocks::rfc1928::util::target_addr::TargetAddr; use alksocks::rfc1928::{ReplyError, Socks5Command, SocksError}; use alksocks::Socks5Params; // --- Identities ----------------------------------------------------------- /// The consumer identity the transport carries: `socks5:open` only /// (the CONNECT/ASSOCIATE grant; no `socks5:bind`). pub fn consumer_identity() -> Identity { Identity { id: "consumer".to_string(), scopes: vec![SOCKS5_OPEN_SCOPE.to_string()], resources: HashMap::new(), } } /// The BIND-scope identity: `socks5:bind` ONLY — no `socks5:open`. /// The command-gate probe's identity. pub fn bind_identity() -> Identity { Identity { id: "consumer-bind".to_string(), scopes: vec![SOCKS5_BIND_SCOPE.to_string()], resources: HashMap::new(), } } /// Both scopes: `socks5:open` AND `socks5:bind` (a deployment /// granting both grants both explicitly — never implied, ADR-006). pub fn both_scopes_identity() -> Identity { Identity { id: "consumer-both".to_string(), scopes: vec![SOCKS5_OPEN_SCOPE.to_string(), SOCKS5_BIND_SCOPE.to_string()], resources: HashMap::new(), } } /// The BIND-capable identity the BIND integration tests open with: /// both scopes (the open gate needs `socks5:open`, the command gate /// needs `socks5:bind` — a session needs both grants explicitly). pub fn bind_scoped_identity() -> Identity { both_scopes_identity() } /// The scoped token — resolves to [`consumer_identity`] via /// [`TestIdProvider`] (the hub-forwarding path's shape). pub const TEST_AUTH_TOKEN: &str = "test-token-with-socks5-open-scope"; /// Resolve the scoped token to the consumer identity; anything else /// resolves to none (falling through to the transport identity per /// the CF-005 precedence order). pub struct TestIdProvider; impl IdentityProvider for TestIdProvider { fn resolve_from_fingerprint(&self, _: &str) -> Option { None } fn resolve_from_token(&self, token: &AuthToken) -> Option { if token.raw == TEST_AUTH_TOKEN.as_bytes() { Some(consumer_identity()) } else { None } } } // --- In-process seam closures --------------------------------------------- /// An echo dial closure (the ADR-004 injection point under test): a /// duplex per dial — the "target" echoes bytes until EOF. The /// producer-side pump gets split `DuplexStream` halves; a spawned /// task echoes the other end. No sockets, no `local` feature. pub fn echo_dial() -> DialFn { Arc::new(move |_target: TargetAddr| { Box::pin(async { let (mut echo_end, target_end) = tokio::io::duplex(64 * 1024); tokio::spawn(async move { loop { let mut buf = vec![0u8; 4096]; match echo_end.read(&mut buf).await { Ok(0) | Err(_) => return, Ok(n) => { if echo_end.write_all(&buf[..n]).await.is_err() { return; } } } } }); let (read_half, write_half) = tokio::io::split(target_end); Ok(TargetHandle { read: Box::new(read_half), write: Box::new(write_half), }) }) }) } /// A dial that always refuses with the given [`SocksDialError`] /// variant (the refusal-mapping tests' dial — the refusal rides /// in-stream as the mapped RFC reply code). The variant is /// reconstructed per call (the seam errors are not `Clone`). pub fn refusing_dial(err: SocksDialError) -> DialFn { Arc::new(move |_target: TargetAddr| { let err = match &err { SocksDialError::Refused(m) => SocksDialError::Refused(m.clone()), SocksDialError::NetworkUnreachable(m) => SocksDialError::NetworkUnreachable(m.clone()), SocksDialError::HostUnreachable(m) => SocksDialError::HostUnreachable(m.clone()), SocksDialError::NotAllowed(m) => SocksDialError::NotAllowed(m.clone()), SocksDialError::General(m) => SocksDialError::General(m.clone()), }; Box::pin(async move { Err(err) }) }) } /// An echo egress: datagrams sent to it echo back (the POC #2 /// shape). The readiness contract is honored: `next_reply` parks on /// the queue's `recv()` — `Pending` when nothing is ready — never an /// instantly-ready `Ok(None)`. pub fn echo_egress() -> EgressFactory { alksocks::egress::egress_factory(|| { let (tx, rx) = tokio::sync::mpsc::unbounded_channel::(); Ok(Box::new(ChanEgress { tx, rx: Mutex::new(rx), }) as Box) }) } use tokio::sync::Mutex; /// The in-process egress behind [`echo_egress`]: `send` queues the /// inner bytes for echo; `next_reply` parks on the queue (`Pending` /// until a datagram arrives — the readiness contract, harness-modeled). struct ChanEgress { tx: tokio::sync::mpsc::UnboundedSender, rx: Mutex>, } impl Egress for ChanEgress { fn send( &mut self, addr: TargetAddr, payload: Bytes, ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> { Box::pin(async move { // Echo: the reply's inner bytes are `addr + data` — // exactly what `next_reply` returns (the wire layer's // length prefix is applied exactly once, by the frame // writer, never here). let mut inner = addr .to_be_bytes() .map_err(|e| EgressError::General(format!("addr encode: {e}")))?; inner.extend_from_slice(&payload); self.tx .send(Bytes::from(inner)) .map_err(|_| EgressError::General("egress closed".into())) }) } fn next_reply(&mut self) -> futures::future::BoxFuture<'_, Result, EgressError>> { Box::pin(async move { // Real async park: `Pending` until a datagram arrives (the // readiness contract — never an instantly-ready `Ok(None)`, // which would starve a biased `select!` into livelock). match self.rx.lock().await.recv().await { Some(inner) => Ok(Some(inner)), // All senders dropped: end the association (an `Err`, // never a hot `Ok(None)`). None => Err(EgressError::General("egress closed".into())), } }) } } /// A failing egress factory (the ASSOCIATE-refusal probe): refuses /// with the given error, which the producer maps to reply#1's code. /// The variant is reconstructed per call (the seam errors are not /// `Clone`). pub fn refusing_egress(err: EgressError) -> EgressFactory { Arc::new(move || { let err = match &err { EgressError::Refused(m) => EgressError::Refused(m.clone()), EgressError::Unreachable(m) => EgressError::Unreachable(m.clone()), EgressError::General(m) => EgressError::General(m.clone()), }; Box::pin(async move { Err(err) }) }) } /// The advertised BND.ADDR of [`echo_accept`] (in-process, no real /// listener — the address is the reply#1 carrier only). pub const ECHO_BIND_PORT: u16 = 2121; /// An echo accept factory: reply#1 advertises a static in-process /// address; the accept future resolves immediately with in-process /// duplex halves (the POC #5 shape). The future owns its resources /// end-to-end; the peer address is a fixed documented stand-in. pub fn echo_accept() -> BindAcceptFactory { Arc::new(move |_target: TargetAddr| { Box::pin(async move { let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), ECHO_BIND_PORT); let accept_future = async move { // The accepted data plane is an in-process echo pipe: // the producer's pump gets the `target_end` halves; a // spawned task echoes the `echo_end` (the "inbound // server" the BIND client's bytes round-trip through). 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(BindTargetHandle { peer: TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 7], 44444))), read: Box::new(read_half) as Box, write: Box::new(write_half) as Box, }) }; Ok(BindAccept::new(bind_addr, accept_future)) }) }) } /// A failing accept factory (the BIND-refusal probe, pre-reply#1 → /// reply#1's code). The variant is reconstructed per call (the seam /// errors are not `Clone`). pub fn refusing_accept_factory(err: alksocks::egress::BindAcceptFactoryError) -> BindAcceptFactory { Arc::new(move |_target: TargetAddr| { use alksocks::egress::BindAcceptFactoryError as E; let err = match &err { E::Refused(m) => E::Refused(m.clone()), E::NotAllowed(m) => E::NotAllowed(m.clone()), E::Shortage(m) => E::Shortage(m.clone()), E::General(m) => E::General(m.clone()), }; Box::pin(async move { Err(err) }) }) } /// A failing accept future (the BIND-refusal probe, post-reply#1 → /// reply#2's code): the factory resolves (reply#1 succeeds with the /// advertised BND.ADDR), then the accept future refuses. The variant /// is reconstructed per call (the seam errors are not `Clone`). pub fn refusing_accept(err: BindAcceptError) -> BindAcceptFactory { Arc::new(move |_target: TargetAddr| { use BindAcceptError as E; let err = match &err { E::Refused(m) => E::Refused(m.clone()), E::NotAllowed(m) => E::NotAllowed(m.clone()), E::General(m) => E::General(m.clone()), }; Box::pin(async move { let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), ECHO_BIND_PORT); Ok(BindAccept::new(bind_addr, async move { Err(err) })) }) }) } // --- The producer registration helper -------------------------------------- /// What the establisher's per-call auth carried as the opener /// identity id (the CF-006 probe — `None` until an open runs). pub type IdentityWitness = Arc>>; /// The session-gate establisher: records the opener identity (the /// CF-006 seam the harness witnesses); returns /// `Establishment::default()` — no plan needed (the dial is /// in-stream, in the pump handler). A scope refusal is the registry /// ACL gate's job (it runs before this ever runs — ADR-006). pub fn socks5_session_establisher( witness: IdentityWitness, ) -> alkcall::channels::operations::OpenEstablisher { Arc::new(move |_input: serde_json::Value, auth: AuthContext| { let witness = Arc::clone(&witness); Box::pin(async move { *witness.lock().await = auth.identity.as_ref().map(|i| i.id.clone()); Ok(alkcall::channels::operations::Establishment::default()) }) }) } /// The smoke-test's minimal CONNECT producer path (the POC #1 /// `serve_socks5_over` shape, inline): negotiate no-auth → read the /// command → dial via the injected closure → reply → `pump_bidi` /// inline (R-02 — the handler's `JoinHandle` tracks the data-plane /// lifetime; early return = teardown-at-birth). The producer task /// replaces this with the real command-dispatch pump; the harness /// keeps the smoke test producer-independent. async fn serve_connect_minimal(stream: T, dial: DialFn) -> Result<(), SocksError> where T: AsyncRead + AsyncWrite + Unpin, { let (proto, cmd, target) = Socks5ServerProtocol::accept_no_auth(stream) .await .map_err(SocksError::ServerError)? .read_command() .await .map_err(SocksError::ServerError)?; if cmd != Socks5Command::TCPConnect { proto .reply_error(&ReplyError::CommandNotSupported) .await .map_err(SocksError::ServerError)?; return Err(SocksError::ServerError(SocksServerError::UnknownCommand( cmd.as_u8(), ))); } let handle = match dial(target).await { Ok(handle) => handle, Err(err) => { proto .reply_error(&err.to_reply_error()) .await .map_err(SocksError::ServerError)?; return Err(SocksError::ServerError(SocksServerError::Bug( "dial refused in-stream", ))); } }; let inner = proto .reply_success(SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0)) .await .map_err(SocksError::ServerError)?; pump_bidi(inner, handle.read, handle.write).await; Ok(()) } /// The pump handler over the minimal CONNECT path (the smoke test's /// producer stand-in — the POC #1 shape). Awaited inline (R-02). fn make_minimal_connect_handler(dial: DialFn) -> alkcall::channels::operations::OpenHandler { Arc::new( move |_input: serde_json::Value, _plan: Option, conn: CoreConnection, _auth: AuthContext| { let dial = Arc::clone(&dial); tokio::spawn(async move { let Some(bidi) = conn.accept_bi().await.ok() else { tracing::error!("socks5 pump: accept_bi failed"); return; }; if let Err(err) = serve_connect_minimal(bidi, dial).await { tracing::warn!("socks5 session ended: {err:#}"); } }) }, ) } /// The producer-side registration mode (what [`wire_with`] registers /// on the producer's serving registry post-hoc — W2's shape; the /// single place the producer-side wiring lives, so the per-command /// tasks only write tests). #[derive(Clone)] pub enum RegisterMode { /// The minimal CONNECT stand-in (the POC #1 inline shape) over /// the given dial, with an identity-witnessing establisher (the /// CF-006 seam probe rides the same topology). MinimalConnect(DialFn), /// The real producer registration /// ([`alksocks::producer::register_socks5_openable`]): the /// session-gate establisher + the command-dispatch pump handler /// over the injected seams (dial required for CONNECT; the /// egress/accept arms refuse RFC-faithfully until their tasks /// land). Producer { dial: DialFn, egress: Option, accept: Option, }, /// No socks5 openable — only the generic channel ops (the /// unknown-op / registry-gate probe's shape). NoSocks5, } /// Register the SOCKS5 open op on a `ChannelCore` + the session's /// dispatch registry — the alkcall 0.8.0 establishment surface /// (`register_openable_with_establisher`) consumed for this crate's /// shape. The producer task's `register_socks5_openable` succeeds /// this helper (the smoke test's stand-in handler is replaced then; /// the registration shape — spec, establisher, handler, auth, /// timeout — stays). pub fn register_socks5_openable_minimal( core: &ChannelCore, on_registry: &Arc, auth: AuthContext, dial: DialFn, witness: IdentityWitness, ) -> Result<(), String> { core.register_openable_with_establisher( socks5_open_spec(), Some(socks5_session_establisher(witness)), make_minimal_connect_handler(dial), on_registry, auth, None, ) } // --- The topology ----------------------------------------------------------- pub struct Topology { /// The producer's client (connect side, serving). Its manager /// allocated the socks5 channels (`ChannelSide::Connect`). pub producer: Arc, /// The consumer's manager (accept side) — the initiator adopts /// the producer-allocated IDs here; probes reap via it. pub consumer_manager: ChannelManager, /// The consumer's channel-0 call connection — its surface for /// calling the producer's open op (and channel/close). pub consumer_call: Arc, /// The producer's serving registry (the socks5 openable + /// generic channel ops live here — probes). pub producer_registry: Arc, /// The consumer's serving registry (empty by default — the /// consumer serves no ops in the default topology). pub consumer_registry: Arc, /// What the establisher's per-call auth carried (the CF-006 /// probe — `None` until an open runs). pub identity_witness: IdentityWitness, } /// Wire the producer↔consumer topology over one duplex transport. /// /// Identity wiring (the load-bearing POC #1 requirement): the /// accept-side install hook calls `channel0_conn.set_identity` /// before building the `CallConnection` — the serving dispatch reads /// it as the caller identity (CF-005 (b)'s accept-side analogue; /// without it, scope-gated opens fail closed with `FORBIDDEN`). The /// transport identity attaches to the dialing connection before /// `from_connection_with_serving` (the mTLS/QUIC posture) and /// propagates automatically on the connect side. /// /// `serving_identity` overrides the dispatch-resolved identity per /// `ServingConfig.identity` (CF-005 (a) probe shape — `None` in the /// primary path). `provider` resolves payload tokens (the /// hub-forwarding path). pub async fn wire_with( mode: RegisterMode, consumer_identity: Option, serving_identity: Option, provider: Arc, ) -> Topology { let producer_op_registry = Arc::new(OperationRegistry::new()); let consumer_serving_registry = Arc::new(OperationRegistry::new()); let (consumer_call_tx, mut consumer_call_rx) = tokio::sync::mpsc::channel::>(1); let (consumer_manager_tx, mut consumer_manager_rx) = tokio::sync::mpsc::channel::(1); let identity_witness: IdentityWitness = Arc::new(tokio::sync::Mutex::new(None)); let fallback = Identity { id: "consumer-fallback".to_string(), scopes: vec![SOCKS5_OPEN_SCOPE.to_string()], resources: HashMap::new(), }; let resolved = Arc::new(consumer_identity.unwrap_or(fallback)); // --- consumer (accept side): ChannelsAdapter + capturing hook --- let resolved_for_hook = Arc::clone(&resolved); let serving_registry_for_hook = Arc::clone(&consumer_serving_registry); let install_hook: InstallChannelZero = Arc::new( move |manager: ChannelManager, channel0_conn: CoreConnection, _auth: AuthContext| { let consumer_call_tx = consumer_call_tx.clone(); let consumer_manager_tx = consumer_manager_tx.clone(); let serving_registry = Arc::clone(&serving_registry_for_hook); let resolved = Arc::clone(&resolved_for_hook); 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); // The load-bearing POC #1 wiring requirement: attach // the resolved identity to channel 0 BEFORE building // the `CallConnection` — the serving dispatch reads // it as the caller identity; without it, scope-gated // opens fail closed (`FORBIDDEN "authentication // required"`). Explicit here; tested below. let _ = channel0_conn.set_identity((*resolved).clone()); let call_connection = Arc::new(CallConnection::new_single_stream( channel0_conn, Arc::clone(&writer), )); // The consumer serves the ops pre-registered on its // serving registry (empty in the default topology — // the dispatcher keeps the call loop alive either // way; its read loop resolves the producer's // responses to the consumer's pending entries). let _ = consumer_call_tx.send(Arc::clone(&call_connection)).await; let _ = consumer_manager_tx.send(manager).await; let dp = Dispatcher::new( serving_registry, Arc::new(alkcall::core::auth::NoopIdentityProvider), ); dp.serve_single_stream(call_connection, reader, writer) .await; }) }, ); // --- transport ------------------------------------------------------ // The producer dials the transport (connect side): its connection // gets the transport identity BEFORE `from_connection_with_serving` // (the mTLS/QUIC posture; CF-005 (b) — the primary caller-identity // path). The accept side resolves it out-of-band (modeled in the // hook above). let (producer_end, consumer_end) = tokio::io::duplex(64 * 1024); let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None); let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None); let identity_for_transport = (*resolved).clone(); producer_conn .set_identity(identity_for_transport) .expect("transport identity set once"); // The consumer (accept side) handles its end with the adapter + // install hook. let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap)); let producer_transport_auth = AuthContext::anonymous(b"alk/channels"); let consumer_conn_for_adapter = consumer_conn; let _adapter_task = tokio::spawn(async move { let _ = alkcall::core::types::ProtocolHandler::handle( &adapter, consumer_conn_for_adapter, &producer_transport_auth, ) .await; }); // --- producer (connect side, serving) ------------------------------- let producer_client = ChannelClient::from_connection_with_serving( producer_conn, Some(ServingConfig { registry: Arc::clone(&producer_op_registry), identity_provider: provider, identity: serving_identity, }), ) .await .expect("producer channel client init"); let producer_client = Arc::new(producer_client); // The generic channel ops ride the producer's serving registry // (channel/close tears a channel down out-of-band — review 007's // recipe). let ops = ChannelOperations::with_default_policy(producer_client.manager().clone()); ops.register_on(&producer_op_registry) .expect("register channel ops on producer"); // Post-hoc openable registration on the producer's registry (W2): // the alkcall 0.8.0 establishment surface, consumed for this // crate's shape. if let RegisterMode::MinimalConnect(dial) = &mode { let core = ChannelCore::new( producer_client.manager().clone(), alkcall::channels::policy::default_policy(), ); register_socks5_openable_minimal( &core, &producer_op_registry, AuthContext::anonymous(b"alk/socks5"), Arc::clone(dial), Arc::clone(&identity_witness), ) .expect("register socks5 openable (minimal CONNECT stand-in)"); } if let RegisterMode::Producer { dial, egress, accept, } = &mode { let core = ChannelCore::new( producer_client.manager().clone(), alkcall::channels::policy::default_policy(), ); alksocks::producer::register_socks5_openable( &core, &producer_op_registry, AuthContext::anonymous(b"alk/socks5"), Some(Arc::clone(dial)), egress.clone(), accept.clone(), Some(Arc::clone(&identity_witness)), ) .expect("register socks5 openable (the real producer)"); } // --- consumer pieces (captured from the hook) ----------------------- let consumer_call = consumer_call_rx.recv().await.expect("consumer call conn"); let consumer_manager = consumer_manager_rx.recv().await.expect("consumer manager"); Topology { producer: producer_client, consumer_manager, consumer_call, producer_registry: Arc::clone(&producer_op_registry), consumer_registry: consumer_serving_registry, identity_witness, } } /// The default topology: the consumer's transport identity is the /// primary caller-identity path; the producer registers the real /// CONNECT path over the given dial. pub async fn wire(dial: DialFn) -> Topology { wire_with( RegisterMode::Producer { dial, egress: None, accept: None, }, Some(consumer_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .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 /// `AsyncRead + AsyncWrite` (greeting → method selection → command /// with target → reply read). The differential suite's established /// pattern: the raw helper drives the producer without depending on /// the consumer session types (which land later). pub struct RawRfcClient { pub stream: S, } impl RawRfcClient where S: AsyncRead + AsyncWrite + Unpin, { /// Wrap a stream (a channel's `BiStream`, a duplex half — /// substrate-agnostic). pub fn new(stream: S) -> Self { Self { stream } } /// Greeting (`05 01 00` — version 5, one method, no-auth) → the /// method-selection read (`05 00`). Returns the selected method /// byte. pub async fn greet(&mut self) -> Result { self.stream.write_all(&[0x05, 0x01, 0x00]).await?; let mut chosen = [0u8; 2]; self.stream.read_exact(&mut chosen).await?; Ok(chosen[1]) } /// Write a command request for `target` (the request form: /// `VER CMD RSV ATYP DST.ADDR DST.PORT`). pub async fn request( &mut self, 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 } /// Read one reply (the 4-byte head + the BND.ADDR per its ATYP). /// Returns `(reply_code, bnd_addr_bytes)` — the address bytes as /// the RFC form carries them (ATYP + addr + port). pub async fn read_reply(&mut self) -> Result<(u8, Vec), std::io::Error> { let mut head = [0u8; 4]; self.stream.read_exact(&mut head).await?; let addr_bytes: Vec = match head[3] { consts::SOCKS5_ADDR_TYPE_IPV4 => { let mut a = [0u8; 6]; self.stream.read_exact(&mut a).await?; a.to_vec() } consts::SOCKS5_ADDR_TYPE_IPV6 => { let mut a = [0u8; 18]; self.stream.read_exact(&mut a).await?; a.to_vec() } consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME => { let mut len = [0u8; 1]; self.stream.read_exact(&mut len).await?; let mut a = vec![0u8; len[0] as usize + 2]; self.stream.read_exact(&mut a).await?; let mut full = vec![len[0]]; full.extend_from_slice(&a); full } other => { return Err(std::io::Error::new( std::io::ErrorKind::InvalidData, format!("unknown address type {other}"), )) } }; Ok((head[1], addr_bytes)) } /// The full CONNECT round-trip: greet, request, reply. Returns /// the reply code. pub async fn connect(&mut self, target: &TargetAddr) -> Result { let method = self.greet().await?; assert_eq!( method, consts::SOCKS5_AUTH_METHOD_NONE, "harness uses no-auth" ); self.request(Socks5Command::TCPConnect, target).await?; Ok(self.read_reply().await?.0) } } // --- Tests ------------------------------------------------------------------ #[cfg(test)] mod tests { use super::*; use alksocks::egress::{BindAcceptFactoryError, EgressFactory}; fn assert_send_sync() {} fn assert_send() {} fn init_trace() { let _ = tracing_subscriber::fmt() .with_max_level(tracing::level_filters::LevelFilter::TRACE) .with_test_writer() .try_init(); } #[test] fn identities_carry_the_intended_scopes() { let open = consumer_identity(); assert_eq!(open.scopes, vec![SOCKS5_OPEN_SCOPE.to_string()]); assert_eq!(open.id, "consumer"); let bind = bind_identity(); assert_eq!(bind.scopes, vec![SOCKS5_BIND_SCOPE.to_string()]); assert!(!bind.scopes.contains(&SOCKS5_OPEN_SCOPE.to_string())); let both = both_scopes_identity(); assert!(both.scopes.contains(&SOCKS5_OPEN_SCOPE.to_string())); assert!(both.scopes.contains(&SOCKS5_BIND_SCOPE.to_string())); assert_eq!(both.scopes.len(), 2, "both grants, explicitly"); } #[test] fn test_id_provider_resolves_the_scoped_token() { let provider = TestIdProvider; let token = AuthToken { raw: TEST_AUTH_TOKEN.as_bytes().to_vec(), }; assert_eq!( provider.resolve_from_token(&token).map(|i| i.id), Some("consumer".to_string()) ); assert_eq!( provider .resolve_from_token(&token) .and_then(|i| i.scopes.into_iter().next()), Some(SOCKS5_OPEN_SCOPE.to_string()) ); assert!(provider .resolve_from_token(&AuthToken { raw: b"other".to_vec(), }) .is_none()); assert!(provider.resolve_from_fingerprint("any").is_none()); } #[tokio::test] async fn echo_dial_round_trips_through_pump() { let handle = echo_dial()(TargetAddr::Domain("example.com".into(), 443)) .await .expect("dial resolves"); let (mut probe_far, probe_near) = tokio::io::duplex(64 * 1024); // The pump feeds the handle against the probe's near end — // exactly the data-plane shape the producer handler runs. let pump = tokio::spawn(pump_bidi(probe_near, handle.read, handle.write)); probe_far.write_all(b"ping").await.expect("write"); let mut buf = [0u8; 4]; probe_far.read_exact(&mut buf).await.expect("echoed back"); assert_eq!(&buf, b"ping"); drop(probe_far); let (c2p, p2c) = pump.await.expect("pump completes"); assert_eq!((c2p, p2c), (4, 4), "both pumps moved the probe bytes"); } #[tokio::test] async fn refusing_dial_maps_to_its_reply_code() { let dial = refusing_dial(SocksDialError::NotAllowed("policy".into())); let err = dial(TargetAddr::Domain("example.com".into(), 443)) .await .expect_err("dial refuses"); assert_eq!( err.to_reply_error().as_u8(), ReplyError::ConnectionNotAllowed.as_u8() ); } #[tokio::test] async fn echo_egress_echoes_inner_bytes_and_parks_when_idle() { let mut egress = echo_egress()().await.expect("egress builds"); let payload = Bytes::from_static(b"query"); egress .send( TargetAddr::Ip("10.0.0.1:53".parse().unwrap()), payload.clone(), ) .await .expect("send succeeds"); // The echo comes back as inner bytes (addr encoding + data). let reply = egress .next_reply() .await .expect("next_reply") .expect("a reply is ready"); let expected = { let mut v = TargetAddr::Ip("10.0.0.1:53".parse().unwrap()) .to_be_bytes() .expect("addr encodes"); v.extend_from_slice(&payload); Bytes::from(v) }; assert_eq!(reply, expected); // Readiness contract: with nothing in flight, `next_reply` // parks (`Pending`), never a hot `Ok(None)`. let mut poll = egress.next_reply(); tokio::time::timeout(std::time::Duration::from_millis(50), &mut poll) .await .expect_err("next_reply stays Pending when nothing is ready"); } #[tokio::test] async fn echo_accept_resolves_with_halves_and_echoes() { let accept = echo_accept()(TargetAddr::Domain("example.com".into(), 2121)) .await .expect("factory resolves"); assert_eq!( accept.bind_addr(), SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT)) ); let mut handle = accept.accept().await.expect("accept resolves"); assert_eq!( handle.peer, TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 7], 44444))) ); // Echo round-trip through the accepted halves. handle.write.write_all(b"ping").await.expect("write"); let mut buf = [0u8; 4]; handle.read.read_exact(&mut buf).await.expect("echoed"); assert_eq!(&buf, b"ping"); } #[tokio::test] async fn failing_accept_variants_refuse_at_their_stages() { // Factory failure → pre-reply#1. let factory = refusing_accept_factory(BindAcceptFactoryError::Refused("no listener".into())); let err = factory(TargetAddr::Domain("example.com".into(), 2121)) .await .expect_err("factory refuses"); assert_eq!( err.to_reply_error().as_u8(), ReplyError::ConnectionRefused.as_u8() ); // Accept-stage failure → post-reply#1 (reply#2). let factory = refusing_accept(BindAcceptError::NotAllowed("policy".into())); let accept = factory(TargetAddr::Domain("example.com".into(), 2121)) .await .expect("factory resolves; the accept future refuses"); let err = accept.accept().await.expect_err("accept refuses"); assert_eq!( err.to_reply_error().as_u8(), ReplyError::ConnectionNotAllowed.as_u8() ); } #[tokio::test] async fn refusing_egress_refuses_at_associate_setup() { let factory = refusing_egress(EgressError::Unreachable("no egress".into())); let err = match factory().await { Err(err) => err, Ok(_) => panic!("the factory must refuse"), }; assert_eq!( err.to_reply_error().as_u8(), ReplyError::NetworkUnreachable.as_u8() ); } #[test] fn seam_closures_have_the_poc_compile_shapes() { assert_send_sync::(); assert_send_sync::(); assert_send_sync::(); assert_send::(); } // -- the raw RFC client helper --------------------------------------- /// A hand-rolled minimal RFC server over a duplex: greeting → /// method selection (`05 00`) → reply read (the differential /// suite's mock-server pattern) — the raw client's counterparty. async fn hand_rolled_server(mut stream: tokio::io::DuplexStream, reply_code: u8) { let mut greeting = [0u8; 3]; stream.read_exact(&mut greeting).await.expect("greeting"); assert_eq!(greeting, [0x05, 0x01, 0x00]); stream .write_all(&[consts::SOCKS5_VERSION, consts::SOCKS5_AUTH_METHOD_NONE]) .await .expect("method selection"); // Read the command request head (VER CMD RSV ATYP) + the rest // per ATYP (a domain here — the helper's test target form). let mut head = [0u8; 4]; stream.read_exact(&mut head).await.expect("command head"); assert_eq!(head[1], consts::SOCKS5_CMD_TCP_CONNECT); match head[3] { consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME => { let mut len = [0u8; 1]; stream.read_exact(&mut len).await.expect("domain len"); let mut rest = vec![0u8; len[0] as usize + 2]; stream.read_exact(&mut rest).await.expect("domain + port"); } consts::SOCKS5_ADDR_TYPE_IPV4 => { let mut rest = [0u8; 6]; stream.read_exact(&mut rest).await.expect("v4 + port"); } other => panic!("test wrote an unexpected address type {other}"), } // Reply with a static v4 BND.ADDR. let addr = SocketAddr::from(([127, 0, 0, 1], 1080)); stream .write_all(&new_reply(&ReplyError::from_u8(reply_code), addr)) .await .expect("reply written"); } /// The raw client round-trips greeting + command + reply against /// a hand-rolled server over a duplex (the task's acceptance /// criterion — the raw helper is producer-independent). #[tokio::test] async fn raw_rfc_client_round_trips_against_a_hand_rolled_server() { let (client_end, server_end) = tokio::io::duplex(1024); let _server = tokio::spawn(hand_rolled_server( server_end, consts::SOCKS5_REPLY_SUCCEEDED, )); let mut raw = RawRfcClient::new(client_end); let code = tokio::time::timeout( std::time::Duration::from_secs(5), raw.connect(&TargetAddr::Domain("echo.test".into(), 443)), ) .await .expect("round-trip timed out") .expect("the raw client completes"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); // The refusal path: the server's reply code surfaces verbatim. let (client_end, server_end) = tokio::io::duplex(1024); let server = tokio::spawn(hand_rolled_server( server_end, consts::SOCKS5_REPLY_CONNECTION_REFUSED, )); let mut raw = RawRfcClient::new(client_end); let code = raw .connect(&TargetAddr::Ip("127.0.0.1:1".parse().unwrap())) .await .expect("round-trip completes"); assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_REFUSED); let _ = server.await; } // -- the smoke test --------------------------------------------------- /// The full loop: harness up → producer registered → consumer /// opens a channel → raw RFC CONNECT conversation → echo /// round-trip. The producer path is the minimal CONNECT stand-in /// (the POC #1 inline shape) — the producer task replaces it. #[tokio::test] async fn smoke_connect_echo_round_trip_over_the_harness() { let topo = wire(echo_dial()).await; // The consumer calls the producer's open op on channel 0 and // adopts the producer-allocated channel id (the alkcall 0.8.0 // establishment surface, consumed end-to-end). let response = topo .consumer_call .call(OP_SOCKS5_OPEN, Socks5Params::as_json()) .await; let out = response.result.expect("the socks5 channel opens"); let channel_id = out .get("channel_id") .and_then(|v| v.as_u64()) .expect("the reply carries channel_id") as u32; assert!(channel_id >= 1); let (send, recv) = topo .consumer_manager .adopt_channel(channel_id, SOCKS5_ALPN, None) .await .expect("adopt the producer-allocated id"); // The raw RFC client drives the producer through the channel: // pump the RFC conversation between the raw client's duplex // and the channel halves. let (stream_peer, stream_self) = tokio::io::duplex(64 * 1024); tokio::spawn(pump_bidi( stream_self, Box::new(recv) as Box, Box::new(send) as Box, )); let mut raw = RawRfcClient::new(stream_peer); let code = tokio::time::timeout( std::time::Duration::from_secs(10), raw.connect(&TargetAddr::Domain("echo.test".into(), 443)), ) .await .expect("CONNECT round-trip timed out") .expect("the raw client completes"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); // Echo round-trip through the data plane: whatever the client // sends is echoed by the dialed in-process target. raw.stream .write_all(b"ping") .await .expect("write the probe"); let mut buf = [0u8; 4]; tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read_exact(&mut buf), ) .await .expect("echo timed out") .expect("read the echo"); assert_eq!(&buf, b"ping"); // The producer-side channel is alive; the identity the // establisher witnessed is the consumer's (CF-006). assert_eq!( topo.identity_witness.lock().await.as_deref(), Some("consumer") ); } /// The scope-less identity fails closed: an identity with no /// `socks5:open` gets the registry's `FORBIDDEN` (the open gate /// is the boundary; ADR-006) — the POC #1 wiring requirement /// tested explicitly (the hook's `set_identity` is what made the /// allowed path work above; omitting it yields the same /// `FORBIDDEN`, fail-closed). #[tokio::test] async fn scopeless_identity_fails_closed_with_forbidden() { // A bare identity — no scopes at all. let bare = Identity { id: "scopeless".to_string(), scopes: vec![], resources: std::collections::HashMap::new(), }; let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: None, accept: None, }, Some(bare), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; let response = topo .consumer_call .call(OP_SOCKS5_OPEN, Socks5Params::as_json()) .await; let err = response .result .expect_err("a scope-less identity fails closed"); assert_eq!(err.code, "FORBIDDEN"); // Fail-closed is fail-closed: no channel anywhere. assert!(topo .consumer_manager .channel_ids() .into_iter() .all(|id| id == 0)); assert!(topo .producer .manager() .channel_ids() .into_iter() .all(|id| id == 0)); } /// The establisher's per-call auth witness: the opener identity /// id arrives on the establisher's `AuthContext` (CF-006) — the /// seam the producer task maps SOCKS5 credentials/ACL onto. #[tokio::test] async fn establisher_witnesses_the_opener_identity() { let topo = wire(echo_dial()).await; let response = topo .consumer_call .call(OP_SOCKS5_OPEN, Socks5Params::as_json()) .await; assert!(response.result.is_ok(), "the channel opens"); tokio::time::timeout(std::time::Duration::from_secs(5), async { loop { let seen = topo.identity_witness.lock().await.clone(); if seen.is_some() { assert_eq!(seen.as_deref(), Some("consumer")); return; } tokio::time::sleep(std::time::Duration::from_millis(10)).await; } }) .await .expect("the establisher ran and witnessed the opener"); } /// The harness topology sanity: the producer's manager allocated /// the channel id (the serving side allocates; the initiator /// adopts — ADR-047 §5). #[tokio::test] async fn producer_allocates_consumer_adopts() { let topo = wire(echo_dial()).await; let response = topo .consumer_call .call(OP_SOCKS5_OPEN, Socks5Params::as_json()) .await; let out = response.result.expect("open"); let channel_id = out.get("channel_id").and_then(|v| v.as_u64()).expect("id") as u32; assert_eq!(channel_id % 2, 1, "the connect side allocates odd ids"); assert!(!topo.consumer_manager.has_channel(channel_id)); assert!(topo.producer.manager().has_channel(channel_id)); let (_send, _recv) = topo .consumer_manager .adopt_channel(channel_id, SOCKS5_ALPN, None) .await .expect("adopt"); assert!(topo.consumer_manager.has_channel(channel_id)); } // -- the real-producer integration tests ----------------------------- /// Open a socks5 channel + adopt it + pump the raw RFC client /// through it (the smoke test's opening shape, shared by the /// producer integration tests). async fn open_channel_and_pump( topo: &Topology, ) -> ( RawRfcClient, tokio::task::JoinHandle<(u64, u64)>, ) { let response = topo .consumer_call .call(OP_SOCKS5_OPEN, Socks5Params::as_json()) .await; let out = response.result.expect("the socks5 channel opens"); let channel_id = out .get("channel_id") .and_then(|v| v.as_u64()) .expect("the reply carries channel_id") as u32; let (send, recv) = topo .consumer_manager .adopt_channel(channel_id, SOCKS5_ALPN, None) .await .expect("adopt the producer-allocated id"); let (stream_peer, stream_self) = tokio::io::duplex(64 * 1024); let pump = tokio::spawn(pump_bidi( stream_self, Box::new(recv) as Box, Box::new(send) as Box, )); (RawRfcClient::new(stream_peer), pump) } /// The real producer end-to-end (the smoke test through the /// command-dispatch path, not the minimal stand-in): CONNECT /// round-trip + echo + the identity witness (CF-006) + half-close /// teardown (client EOF ends the session cleanly). #[tokio::test] async fn producer_connect_echo_witnesses_identity_and_tears_down_cleanly() { let topo = wire(echo_dial()).await; let (mut raw, pump) = open_channel_and_pump(&topo).await; let code = tokio::time::timeout( std::time::Duration::from_secs(10), raw.connect(&TargetAddr::Domain("echo.test".into(), 443)), ) .await .expect("CONNECT round-trip timed out") .expect("the raw client completes"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); raw.stream.write_all(b"ping").await.expect("write"); let mut buf = [0u8; 4]; tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.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") ); // Half-close teardown: client EOF ends the session; the // channel pump completes (the handler tracked the data plane // to the end — R-02). drop(raw); tokio::time::timeout(std::time::Duration::from_secs(10), pump) .await .expect("the data plane pump ends on client EOF") .expect("join"); } /// A refusing dial surfaces 0x05 in-stream with no phantom data /// plane: the reply comes back refused; bytes sent after the /// refusal are not echoed (nothing was dialed). The channel /// stays healthy — a fresh open succeeds immediately. #[tokio::test] async fn producer_dial_refusal_is_0x05_and_no_phantom_data_plane() { let topo = wire(refusing_dial(SocksDialError::Refused("target down".into()))).await; let (mut raw, _pump) = open_channel_and_pump(&topo).await; let method = raw.greet().await.expect("greet"); assert_eq!(method, consts::SOCKS5_AUTH_METHOD_NONE); raw.request( Socks5Command::TCPConnect, &TargetAddr::Domain("down.test".into(), 80), ) .await .expect("request written"); let (code, _addr) = raw.read_reply().await.expect("reply read"); assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_REFUSED); // No phantom data plane: nothing echoes (the refusal is the // whole conversation; the data plane never existed). drop(_pump); // The channel topology is healthy: a fresh CONNECT through a // working dial succeeds immediately after the refused one. let topo2 = wire(echo_dial()).await; let (mut raw2, _pump2) = open_channel_and_pump(&topo2).await; let code = raw2 .connect(&TargetAddr::Domain("echo.test".into(), 443)) .await .expect("fresh connect"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); } /// The scope-class dial refusal (`NotAllowed`) surfaces 0x02 /// (the mapped code — the dial seam's policy refusal path). #[tokio::test] async fn producer_scope_class_refusal_is_0x02() { let topo = wire(refusing_dial(SocksDialError::NotAllowed("policy".into()))).await; let (mut raw, _pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::TCPConnect, &TargetAddr::Domain("policy.test".into(), 80), ) .await .expect("request written"); let (code, _addr) = raw.read_reply().await.expect("reply read"); assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED); } /// ASSOCIATE refuses RFC-faithfully when the egress seam is not /// wired (`CommandNotSupported` 0x07 — ADR-008's missing-callback /// class) while CONNECT still serves the same registration (the /// missing-callback posture, over channels). #[tokio::test] async fn producer_associate_refuses_0x07_connect_still_serves() { let topo = wire(echo_dial()).await; // ASSOCIATE refuses (no egress seam wired). let (mut raw, _pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::UDPAssociate, &TargetAddr::Domain("relay.test".into(), 1080), ) .await .expect("request written"); let (code, _addr) = raw.read_reply().await.expect("reply read"); assert_eq!(code, consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED); drop(raw); // CONNECT still serves the same registration. let (mut raw, _pump) = open_channel_and_pump(&topo).await; let code = raw .connect(&TargetAddr::Domain("echo.test".into(), 443)) .await .expect("connect"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); raw.stream.write_all(b"still").await.expect("write"); let mut buf = [0u8; 5]; raw.stream.read_exact(&mut buf).await.expect("echo"); assert_eq!(&buf, b"still"); } // -- the producer ASSOCIATE datagram stage over channels ------------- /// Frame one datagram the way the in-band wire carries it /// (`[len: u16 BE][addr + data]` — the wire codec's shape). fn udp_frame(addr: &TargetAddr, payload: &[u8]) -> Vec { alksocks::wire::frame(addr, payload) .expect("frame") .to_vec() } /// Decode a datagram from the wire (one frame; in-order batches). fn udp_unframe(wire: &[u8]) -> (TargetAddr, Bytes) { let mut reader = alksocks::wire::DatagramReader::new(); let mut got = None; for dg in reader.feed(wire) { got = Some(dg); } got.expect("one datagram").expect("frame decodes") } /// The ASSOCIATE conversation over a channel: open + adopt + /// pump, greet, request, reply — asserting the sentinel /// `127.0.0.1:0` BND.ADDR (ADR-003's RFC-compatibility surface). async fn raw_associate( topo: &Topology, ) -> ( RawRfcClient, tokio::task::JoinHandle<(u64, u64)>, ) { let (mut raw, pump) = open_channel_and_pump(topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::UDPAssociate, &TargetAddr::Domain("relay.test".into(), 1080), ) .await .expect("request written"); let (code, addr) = raw.read_reply().await.expect("reply read"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED, "the associate reply"); assert_eq!( (addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]), (127, 0, 0, 1, 0, 0), "the reply carries the sentinel BND.ADDR" ); (raw, pump) } /// The datagram stage end to end over channels (the POC #2 /// shape): sentinel reply → datagrams round-trip through the echo /// egress in order, one association; EOF (channel shutdown) ends /// the association and a fresh associate succeeds immediately /// (the POC's test 4); the empty-payload datagram is legal (never /// EOF); a garbage-ATYP frame ends the association loud. #[tokio::test] async fn producer_udp_associate_datagram_stage_over_channels() { init_trace(); let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: Some(echo_egress()), accept: None, }, Some(consumer_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; // Associate: the sentinel reply, then the datagram stage. let (mut raw, pump) = raw_associate(&topo).await; let target = TargetAddr::Ip("10.0.0.1:53".parse().unwrap()); // Multiple datagrams, in order, one association. for i in 0..5u8 { raw.stream .write_all(&udp_frame(&target, &[i; 10])) .await .expect("datagram sent"); } for i in 0..5u8 { let mut buf = vec![0u8; 2 + 7 + 10]; tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read_exact(&mut buf), ) .await .expect("echo timed out") .expect("reply frame read"); let (src, payload) = udp_unframe(&buf); assert_eq!(src, target, "the echo reports the destination"); assert_eq!(&payload[..], &vec![i; 10][..], "in-order echo"); } // The empty-payload datagram: legal, never EOF (the F-2 // invariant at payload level). raw.stream .write_all(&udp_frame(&target, b"")) .await .expect("empty datagram sent"); let mut buf = vec![0u8; 2 + 7]; tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read_exact(&mut buf), ) .await .expect("empty echo timed out") .expect("reply frame read"); let (_, payload) = udp_unframe(&buf); assert!(payload.is_empty(), "the empty datagram round-trips"); // A garbage-ATYP frame ends the association loud: EOF on the // next read. raw.stream .write_all(&[0x00, 0x04, 0x7f, 1, 2, 3]) .await .expect("garbage frame sent"); let mut buf = [0u8; 16]; let outcome = tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read(&mut buf), ) .await .expect("read resolves"); match outcome { Ok(0) | Err(_) => {} Ok(n) => panic!("EOF expected after an undecodable frame — got {n} bytes"), } drop(raw); tokio::time::timeout(std::time::Duration::from_secs(10), pump) .await .expect("the stage pump ends") .expect("join"); } /// Channel EOF ends the association and the connection stays /// healthy (the POC's test 4, over channels): after one /// association's channel is torn down, a fresh associate succeeds /// immediately and round-trips. #[tokio::test] async fn producer_udp_associate_eof_teardown_connection_stays_healthy() { init_trace(); let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: Some(echo_egress()), accept: None, }, Some(consumer_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; // First association: EOF ends it (half-close the channel). let (raw, pump) = raw_associate(&topo).await; drop(raw); tokio::time::timeout(std::time::Duration::from_secs(10), pump) .await .expect("the stage pump ends on channel EOF") .expect("join"); // The connection is healthy: a fresh associate round-trips. let (mut raw, _pump) = raw_associate(&topo).await; let target = TargetAddr::Ip("10.0.0.2:53".parse().unwrap()); raw.stream .write_all(&udp_frame(&target, b"fresh")) .await .expect("datagram sent"); let mut buf = vec![0u8; 2 + 7 + 5]; tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read_exact(&mut buf), ) .await .expect("fresh echo timed out") .expect("reply frame read"); let (_, payload) = udp_unframe(&buf); assert_eq!(&payload[..], b"fresh"); } /// The egress-factory refusal maps to reply#1's mapped code over /// channels (the pre-reply refusal path) — no phantom datagram /// plane. #[tokio::test] async fn producer_udp_associate_factory_refusal_maps_to_reply_code() { let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: Some(refusing_egress(EgressError::Unreachable( "no egress".into(), ))), accept: None, }, Some(consumer_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; let (mut raw, _pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::UDPAssociate, &TargetAddr::Domain("relay.test".into(), 1080), ) .await .expect("request written"); let (code, _addr) = raw.read_reply().await.expect("reply read"); assert_eq!( code, consts::SOCKS5_REPLY_NETWORK_UNREACHABLE, "the factory refusal maps to reply#1's code" ); } /// CONNECT + ASSOCIATE on the same registration over channels /// (the regression guard, extended — one registration, both /// commands). #[tokio::test] async fn producer_connect_and_associate_on_the_same_registration() { let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: Some(echo_egress()), accept: None, }, Some(consumer_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; // CONNECT serves. let (mut raw, _pump) = open_channel_and_pump(&topo).await; let code = raw .connect(&TargetAddr::Domain("echo.test".into(), 443)) .await .expect("connect"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); raw.stream.write_all(b"tcp").await.expect("write"); let mut buf = [0u8; 3]; raw.stream.read_exact(&mut buf).await.expect("echo"); assert_eq!(&buf, b"tcp"); drop(raw); // ASSOCIATE serves the same registration. let (mut raw, _pump) = raw_associate(&topo).await; let target = TargetAddr::Ip("10.0.0.3:53".parse().unwrap()); raw.stream .write_all(&udp_frame(&target, b"udp")) .await .expect("datagram sent"); let mut buf = vec![0u8; 2 + 7 + 3]; raw.stream.read_exact(&mut buf).await.expect("echo"); let (_, payload) = udp_unframe(&buf); assert_eq!(&payload[..], b"udp"); } /// BIND over channels (the phase model end to end): the accept /// factory resolves at command dispatch, reply#1 carries the /// advertised BND.ADDR, reply#2 carries the accepted peer, and /// the data plane two-pumps raw through the channel (echo both /// directions). #[tokio::test] async fn producer_bind_round_trip_over_channels() { init_trace(); let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: None, accept: Some(echo_accept()), }, Some(bind_scoped_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; let (mut raw, pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::TCPBind, &TargetAddr::Domain("bind.test".into(), ECHO_BIND_PORT), ) .await .expect("request written"); // Reply#1: the advertised BND.ADDR (the harness's static // in-process address). let (code1, addr1) = raw.read_reply().await.expect("reply#1"); assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED, "reply#1 success"); assert_eq!( u16::from_be_bytes([addr1[addr1.len() - 2], addr1[addr1.len() - 1]]), ECHO_BIND_PORT, "reply#1 carries the factory's advertised port" ); // Reply#2: the accepted peer (the harness's fixed stand-in). let (code2, addr2) = raw.read_reply().await.expect("reply#2"); assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED, "reply#2 success"); let peer_port = u16::from_be_bytes([addr2[addr2.len() - 2], addr2[addr2.len() - 1]]); assert_eq!(peer_port, 44444, "reply#2 carries the accepted peer"); // Phase 2: raw pass-through through the channel, both // directions. raw.stream.write_all(b"ping").await.expect("write"); let mut buf = [0u8; 4]; tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read_exact(&mut buf), ) .await .expect("echo timed out") .expect("read the echo"); assert_eq!(&buf, b"ping"); raw.stream.write_all(b"pong").await.expect("write"); tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read_exact(&mut buf), ) .await .expect("echo timed out") .expect("read the echo"); assert_eq!(&buf, b"pong"); // Half-close teardown ends the session cleanly. drop(raw); tokio::time::timeout(std::time::Duration::from_secs(10), pump) .await .expect("the data plane pump ends on client EOF") .expect("join"); } /// The command gate over channels (ADR-006; POC #5 test 4): an /// identity holding only `socks5:open` gets `ConnectionNotAllowed` /// (0x02) on BIND — while CONNECT still serves the same identity /// (the gate distinguishes ingress from egress). #[tokio::test] async fn producer_bind_command_gate_refuses_0x02_connect_unaffected() { let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: None, accept: Some(echo_accept()), }, Some(consumer_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; // BIND refused, RFC-faithfully — no phantom listener (the // accept seam never ran; the session ends). let (mut raw, pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::TCPBind, &TargetAddr::Domain("gate.test".into(), ECHO_BIND_PORT), ) .await .expect("request written"); let (code, _addr) = raw.read_reply().await.expect("reply read"); assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED); drop(raw); drop(pump); // CONNECT still works for the same identity (egress is not // implied by the BIND gate's absence — separate scopes). let (mut raw, _pump) = open_channel_and_pump(&topo).await; let code = raw .connect(&TargetAddr::Domain("echo.test".into(), 443)) .await .expect("connect still allowed"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); raw.stream.write_all(b"egress").await.expect("write"); let mut buf = [0u8; 6]; tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read_exact(&mut buf), ) .await .expect("echo timed out") .expect("read the echo"); assert_eq!(&buf, b"egress"); } /// The BIND-scope-only identity over channels: `socks5:bind` /// alone serves BIND (the open gate is `socks5:open` — the /// registration's identity must carry it; the command gate only /// adds `socks5:bind` on top of a granted session). The both- /// scopes identity is the deployment shape that grants both /// explicitly (ADR-006: never implied). #[tokio::test] async fn producer_bind_serves_a_both_scopes_identity() { let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: None, accept: Some(echo_accept()), }, Some(both_scopes_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; let (mut raw, _pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::TCPBind, &TargetAddr::Domain("bind.test".into(), ECHO_BIND_PORT), ) .await .expect("request written"); let (code1, _addr) = raw.read_reply().await.expect("reply#1"); assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED); let (code2, _addr) = raw.read_reply().await.expect("reply#2"); assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED); raw.stream.write_all(b"scoped").await.expect("write"); let mut buf = [0u8; 6]; tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read_exact(&mut buf), ) .await .expect("echo timed out") .expect("read the echo"); assert_eq!(&buf, b"scoped"); } /// Quiet-wait garbage over channels: a control byte after /// reply#1 ends the session with no data plane — the raw client /// reads EOF (no reply#2, no echo), and the channel pump /// completes. #[tokio::test] async fn producer_bind_quiet_wait_garbage_over_channels() { init_trace(); // The accept future parks (never resolves) so the garbage // branch is deterministic. let parked: BindAcceptFactory = Arc::new(move |_target: TargetAddr| { Box::pin(async move { let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), ECHO_BIND_PORT); Ok(BindAccept::new( bind_addr, std::future::pending::>(), )) }) }); let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: None, accept: Some(parked), }, Some(bind_scoped_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; let (mut raw, pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::TCPBind, &TargetAddr::Domain("quiet.test".into(), ECHO_BIND_PORT), ) .await .expect("request written"); let (code1, _addr) = raw.read_reply().await.expect("reply#1"); assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED); // Protocol garbage in the quiet wait. raw.stream.write_all(&[0x05, 0x02]).await.expect("garbage"); // The session ends: EOF (no reply#2, no echo). let mut buf = [0u8; 10]; let outcome = tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read(&mut buf), ) .await .expect("read resolves"); match outcome { Ok(0) | Err(_) => {} Ok(n) => panic!("no reply#2 after garbage — got {n} bytes"), } // The client vanishes (EOF both ways through the raw pipe); // the session pump ends (R-02 — the handler tracked the data // plane to its end). drop(raw); tokio::time::timeout(std::time::Duration::from_secs(10), pump) .await .expect("the session pump ends on the garbage refusal") .expect("join"); } /// The three-command regression guard (POC #5 test 7): one /// registration, all three commands — CONNECT serves, BIND serves /// under its gate, ASSOCIATE serves with the echo egress. Fresh /// channel per session. #[tokio::test] async fn producer_three_command_regression_guard() { let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: Some(echo_egress()), accept: Some(echo_accept()), }, Some(both_scopes_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; // CONNECT serves. let (mut raw, _pump) = open_channel_and_pump(&topo).await; let code = raw .connect(&TargetAddr::Domain("echo.test".into(), 443)) .await .expect("connect"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); raw.stream.write_all(b"c").await.expect("write"); let mut buf = [0u8; 1]; tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read_exact(&mut buf), ) .await .expect("echo timed out") .expect("read the echo"); assert_eq!(&buf, b"c"); drop(raw); // BIND serves under the gate. let (mut raw, _pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::TCPBind, &TargetAddr::Domain("bind.test".into(), ECHO_BIND_PORT), ) .await .expect("request written"); let (code1, _addr) = raw.read_reply().await.expect("reply#1"); assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED); let (code2, _addr) = raw.read_reply().await.expect("reply#2"); assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED); drop(raw); // ASSOCIATE serves: the sentinel reply + one datagram // round-trip. let (mut raw, _pump) = raw_associate(&topo).await; let target = TargetAddr::Ip("10.0.0.9:53".parse().unwrap()); raw.stream .write_all(&udp_frame(&target, b"all3")) .await .expect("datagram sent"); let mut buf = vec![0u8; 2 + 7 + 4]; tokio::time::timeout( std::time::Duration::from_secs(10), raw.stream.read_exact(&mut buf), ) .await .expect("echo timed out") .expect("reply frame read"); let (_, payload) = udp_unframe(&buf); assert_eq!(&payload[..], b"all3"); } /// BIND two-stage failure mapping over channels: factory failure /// → reply#1's mapped code; accept-stage failure → reply#2's /// mapped code, no data plane (POC #5 test 6's shape). #[tokio::test] async fn producer_bind_failure_mapping_over_channels() { use alksocks::egress::BindAcceptFactoryError; // Factory failure → reply#1 ConnectionRefused (0x05). let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: None, accept: Some(refusing_accept_factory(BindAcceptFactoryError::Refused( "no listener".into(), ))), }, Some(bind_scoped_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; let (mut raw, _pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::TCPBind, &TargetAddr::Domain("factory.fail".into(), ECHO_BIND_PORT), ) .await .expect("request written"); let (code, _addr) = raw.read_reply().await.expect("reply read"); assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_REFUSED); drop(raw); // Accept-stage failure → reply#2 ConnectionNotAllowed (0x02). let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: None, accept: Some(refusing_accept(BindAcceptError::NotAllowed( "policy".into(), ))), }, Some(bind_scoped_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; let (mut raw, _pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::TCPBind, &TargetAddr::Domain("accept.fail".into(), ECHO_BIND_PORT), ) .await .expect("request written"); let (code1, _addr) = raw.read_reply().await.expect("reply#1"); assert_eq!( code1, consts::SOCKS5_REPLY_SUCCEEDED, "the factory resolved" ); let (code2, _addr) = raw.read_reply().await.expect("reply#2"); assert_eq!(code2, consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED); } /// The no-factory default over channels (ADR-007 variant 3, /// wrapper-aware): sentinel reply#1 + sentinel reply#2, the full /// phase model preserved (the quiet wait included). #[tokio::test] async fn producer_bind_no_factory_serves_variant_3_over_channels() { let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), egress: None, accept: None, }, Some(bind_scoped_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), ) .await; let (mut raw, _pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( Socks5Command::TCPBind, &TargetAddr::Domain("variant3.test".into(), 2121), ) .await .expect("request written"); let (code1, addr1) = raw.read_reply().await.expect("reply#1"); assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED); assert!( addr1.iter().all(|b| *b == 0), "reply#1 carries the wrapper-aware sentinel" ); let (code2, addr2) = raw.read_reply().await.expect("reply#2"); assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED); assert!( addr2.iter().all(|b| *b == 0), "reply#2 carries the wrapper-aware sentinel peer" ); } /// Multiple sequential CONNECTs ride one registration: fresh /// channel per session (one channel per SOCKS5 session, /// convention 10), each dial seam call gets its parsed target. #[tokio::test] async fn producer_serves_repeated_connects_on_one_registration() { let topo = wire(echo_dial()).await; for i in 0..3 { let (mut raw, _pump) = open_channel_and_pump(&topo).await; let code = raw .connect(&TargetAddr::Domain(format!("t{i}.test"), 443)) .await .expect("connect"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); raw.stream.write_all(b"ping").await.expect("write"); let mut buf = [0u8; 4]; raw.stream.read_exact(&mut buf).await.expect("echo"); assert_eq!(&buf, b"ping"); drop(raw); } } }