diff --git a/tasks/implementation/core-test-harness.md b/tasks/implementation/core-test-harness.md index 8a6855e..35922ae 100644 --- a/tasks/implementation/core-test-harness.md +++ b/tasks/implementation/core-test-harness.md @@ -1,7 +1,7 @@ --- id: implementation/core-test-harness name: Core — integration test harness (duplex transport + identities + in-process seams) -status: pending +status: completed depends_on: [implementation/core-seam-types, implementation/core-params-errors] scope: moderate risk: medium @@ -81,16 +81,16 @@ Deliverables: ## Acceptance Criteria -- [ ] Harness topology works over alkcall 0.8.0 (the establishment +- [x] Harness topology works over alkcall 0.8.0 (the establishment surface consumed successfully — the first-consumption risk retired) -- [ ] Identity wiring explicit: `set_identity` in the install hook; +- [x] Identity wiring explicit: `set_identity` in the install hook; a scope-less identity gets `FORBIDDEN` (fails closed) — tested -- [ ] All three identities + all in-process seam closures unit-tested -- [ ] The raw RFC client helper round-trips a greeting + command + +- [x] All three identities + all in-process seam closures unit-tested +- [x] The raw RFC client helper round-trips a greeting + command + reply against a hand-rolled server over a duplex -- [ ] No `local` feature, no sockets — everything in-process -- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean +- [x] No `local` feature, no sockets — everything in-process +- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean (the harness is dev-only — verify it stays out of the wasm build) @@ -108,8 +108,108 @@ Deliverables: ## Notes -> To be filled by implementation agent +> Filled by the implementation agent (2026-09-23). + +- `tests/harness.rs`: the producer↔consumer wiring over one + `tokio::io::duplex` carrying the channels 8-byte chunk format, in + the task's topology — **producer = connect side** + (`ChannelClient::from_connection_with_serving`; its manager + allocates the odd channel IDs), **consumer = accept side** (the + `ChannelsAdapter` + capturing install hook; the hook captures the + consumer's channel-0 `CallConnection` + `ChannelManager`). +- The load-bearing POC #1 wiring requirement is explicit and tested: + the hook calls `channel0_conn.set_identity(resolved)` before + building the `CallConnection`; the transport identity is set on + the dialing (producer) connection before + `from_connection_with_serving` (CF-005 (b) primary path). The + scope-less identity test asserts the `FORBIDDEN` code and that no + channel survives on either side (fail-closed). +- Identities: `consumer_identity()` (`socks5:open` only), + `bind_identity()` (`socks5:bind` only), `both_scopes_identity()` + (both, explicit — ADR-006), `TestIdProvider` + + `TEST_AUTH_TOKEN` (the token-resolver path, unit-tested both + directions). +- In-process seams (no sockets, no `local`): `echo_dial()` + (echo-per-dial, exercised through `pump_bidi` — the producer data + plane's shape), `refusing_dial(err)` (variant-preserving per-call + reconstruction — the seam errors are not `Clone`), + `echo_egress()` (a `ChanEgress` whose `next_reply` parks on + `mpsc::UnboundedReceiver::recv()` — `Pending` when idle, the + readiness contract verified with a 50ms timeout probe), + `refusing_egress(err)`, `echo_accept()` (reply#1 advertises a + static in-process BND.ADDR, `ECHO_BIND_PORT` 2121; the accept + future resolves immediately with echo-pipe halves), + `refusing_accept_factory(err)` (pre-reply#1) and + `refusing_accept(err)` (post-reply#1). +- `register_socks5_openable_minimal` + `socks5_session_establisher`: + the producer-side wiring helper over alkcall 0.8.0's + `ChannelCore::register_openable_with_establisher` (spec from + `socks5_open_spec()`, `Establishment::default()` — session gate + only; the dial is in-stream). The establisher witnesses the + per-call opener identity id (CF-006) into an `IdentityWitness`, + asserted end-to-end in the smoke test. +- `RegisterMode` (`MinimalConnect(dial)` / `NoSocks5`) controls what + `wire_with` registers post-hoc (W2) on the producer's serving + registry; generic channel ops always register. +- The smoke test's producer path is the minimal CONNECT stand-in + (`make_minimal_connect_handler` → `serve_connect_minimal`: the + forked typestate's `accept_no_auth` → `read_command` → CONNECT-only + dispatch → the injected `DialFn` → `reply_success` → `pump_bidi` + inline, R-02). The producer task replaces this; the registration + shape (spec/establisher/handler/auth/timeout) stays. +- `RawRfcClient`: hand-rolled greeting (`05 01 00` → `05 00`) + + request encoder (all three commands, all address forms via + `TargetAddr::to_be_bytes()`) + reply reader (v4/v6/domain BND.ADDR + forms; returns `(code, addr_bytes)`). Round-trip-tested against a + hand-rolled byte-level server over a duplex, success + refusal + reply codes; the harness is producer-independent (no consumer + session types). +- Wire note: the smoke test calls the open op via + `consumer_call.call(OP_SOCKS5_OPEN, Socks5Params::as_json())` and + adopts on `consumer_manager` (the alktunnels `wire_with` + + `open_reverse_channel` shape) rather than + `ChannelClient::open_channel` — in this topology the open op is + served by the producer's registry and reached from the accept + side's channel-0 call surface; `open_channel` remains the + consumer-session surface for the consumer task's session types. +- The harness is `tests/harness.rs` only (dev-only) — verified out + of the wasm build by the targeted checks below. ## Summary -> To be filled on completion \ No newline at end of file +> Filled on completion (2026-09-23). + +Built `tests/harness.rs` — the integration test harness: the +producer↔consumer topology over one `tokio::io::duplex` (producer = +connect side/serving; consumer = accept side/adapter + capturing +hook), the alkcall 0.8.0 establishment surface consumed for the +first time by this crate (`register_openable_with_establisher`, +`Establishment::default`, the consumer-side call + adopt dance) — +the first-consumption risk retired with 14 green tests. + +- The three identities + `TestIdProvider` (all unit-tested). +- All in-process seam closures: `echo_dial`/`refusing_dial` (variant- + preserving), `echo_egress` (readiness contract verified — a 50ms + timeout probe proves `next_reply` parks, never a hot `Ok(None)`) / + `refusing_egress`, `echo_accept` (immediate resolve + echo) / + `refusing_accept_factory` / `refusing_accept` (two-stage refusal + shapes, mapping-table-verified through `to_reply_error()`). +- The producer registration helper (`register_socks5_openable_minimal` + + `socks5_session_establisher` + `RegisterMode`) — the single place + the producer-side wiring lives; the CF-006 identity witness rides + it and is asserted. +- `RawRfcClient` — the raw RFC 1928 client helper over any + `AsyncRead + AsyncWrite`, round-tripped against a hand-rolled + byte-level server (success + refusal replies). +- The smoke test: harness up → producer registered → consumer open + + adopt → raw RFC CONNECT → echo round-trip through `pump_bidi`, with + the establisher witness asserted mid-flight. Plus the fail-closed + probe (scope-less identity → `FORBIDDEN`, no channel anywhere) and + the odd-ID allocation/adopt sanity probe. + +Verification: `cargo test` (127 green: 76 lib + 14 harness + 37 +rfc1928-differential), `cargo test --all-features` green, `cargo +clippy --all-targets -- -D warnings` clean, `cargo fmt --check` +clean, `cargo doc --no-deps` zero warnings, both wasm checks +(`cargo check` + `cargo clippy` on `wasm32-unknown-unknown`) clean — +the harness stays out of the wasm build. \ No newline at end of file diff --git a/tests/harness.rs b/tests/harness.rs new file mode 100644 index 0000000..36cac16 --- /dev/null +++ b/tests/harness.rs @@ -0,0 +1,1166 @@ +//! Test harness — the producer↔consumer wiring over a +//! `tokio::io::duplex` carrying the channels 8-byte chunk format. +//! +//! 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. + +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 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), + /// 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)"); + } + + // --- 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 minimal +/// CONNECT stand-in over the given dial. +pub async fn wire(dial: DialFn) -> Topology { + wire_with( + RegisterMode::MinimalConnect(dial), + Some(consumer_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await +} + +// --- 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() {} + + #[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::MinimalConnect(echo_dial()), + 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)); + } +}