From 4e1673eefcf7328a4f834da901aff27dfef5a852 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Fri, 25 Sep 2026 22:13:17 +0000 Subject: [PATCH] =?UTF-8?q?composed:=20the=20bind-first=20accept=20?= =?UTF-8?q?=E2=80=94=20the=20listener=20on=20a=20listen=20resource?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lands the composed accept (ADR-007 variant 2 / ADR-008): the BIND accept whose listener lives on an alktunnels 0.2.0 bind-first listen resource, at the graduation's vanilla fidelity. - `composed::bindfirst_accept` — the BindAcceptFactory: opens the listen channel (reply#1 advertises the open-op reply's real `bound`, the 0.8.0 reply projection — no expected-addr stand-in), inserts the minted `listen` token into the correlation registry, and returns the accept future parked on the token's oneshot. - `composed::register_forwarded_serving` — the forwarded-op serving half on the SOCKS5 producer's connection: the shared handler correlates by token and owns the middle pump (either-EOF completion; a three-party per-direction shutdown cascade would deadlock). - The Sync-bound wrinkle: `SyncHalves`/`OwnedWrite` (mutex Sync-proof, no unsafe) over the forwarded halves; the owner (`ForwardedOwner`, the TunnelListener) rides inside the write payload — dropping the handle drops the session, halves, and listener together (teardown structural, compiler-enforced). - `tests/composed-accept.rs`: the full story (real bound, real peer, raw pass-through round trip), client-gone teardown drop-witness, distinct tokens for concurrent BINDs, per-accept isolation of a FORBIDDEN forwarded open, late-forwarded correlation miss. - The producer-side adopted forwarded entry rides the per-adopter contract (transport-EOF reap), as upstream's bind-first suites concede; the no-leak probe reflects that shape. Verification: cargo test (default + --all-features), wasm check/clippy (hard gate), clippy --all-targets --all-features -D warnings, fmt --check — all clean. --- src/composed/mod.rs | 469 +++++++++++++++++++++- src/lib.rs | 3 +- tasks/implementation/composed-accept.md | 74 +++- tests/composed-accept.rs | 502 ++++++++++++++++++++++++ tests/harness.rs | 260 ++++++++++++ 5 files changed, 1284 insertions(+), 24 deletions(-) create mode 100644 tests/composed-accept.rs diff --git a/src/composed/mod.rs b/src/composed/mod.rs index 8a48712..be23685 100644 --- a/src/composed/mod.rs +++ b/src/composed/mod.rs @@ -1,8 +1,11 @@ -//! The `composed` feature's egress provider (ADR-008's composed -//! variants, ADR-009's feature gate): [`direct_egress`] — the UDP -//! ASSOCIATE egress that dials each destination through an alktunnels -//! 0.2.0 direct-op channel ([`alktunnels::TunnelSession::open_direct`], -//! `Substrate::Udp`). +//! The `composed` feature's egress + accept providers (ADR-008's +//! composed variants, ADR-009's feature gate): [`direct_egress`] — the +//! UDP ASSOCIATE egress that dials each destination through an +//! alktunnels 0.2.0 direct-op channel +//! ([`alktunnels::TunnelSession::open_direct`], `Substrate::Udp`) — +//! and [`bindfirst_accept`] — the TCP BIND accept whose listener lives +//! on a bind-first listen resource (ADR-007 variant 2 / ADR-008's +//! composed accept). //! //! **Per-target ACL for free** (ADR-008's composed benefit): each //! destination rides its own `channels/tunnel/direct`-governed channel @@ -49,30 +52,44 @@ //! **The Sync-bound wrinkle** (the batch-1 review's seam contract vs //! alktunnels' surface): the landed `TargetHandle`/`BindTargetHandle` //! halves are `Box`, but -//! alktunnels' `TakenHalves` yields `Send + Unpin` only. The -//! owner-wrapper posture (egress.rs's documented contract: "keep-alive -//! state a substrate needs rides inside the halves — a provider-side -//! owner wrapper") is designed here as [`SessionOwner`] and shared -//! with the composed accept (the sibling task): an alktunnels half -//! wrapped with its session inside, so the handle payload's `Sync` -//! bound is met by ownership and the channel entry reaps when the -//! wrapper (and its session) drops. For the egress the wrapper is not -//! the handle payload's burden — the egress owns the session in the -//! table entry (the POC's shape); the relay task owns the bare read -//! half. The accept variant (the sibling task) wraps the accepted -//! halves in this wrapper. +//! alktunnels' `TakenHalves` yields `Send + Unpin` only — and +//! `alkcall::BiStream` is `Send + Unpin` but NOT `Sync` (its inner is +//! a `dyn AsyncReadWrite` trait object). The owner-wrapper posture +//! (egress.rs's documented contract: "keep-alive state a substrate +//! needs rides inside the halves — a provider-side owner wrapper") is +//! designed here as [`SessionOwner`] and shared with the composed +//! accept (the sibling task): an alktunnels half wrapped with its +//! session inside, so the handle payload's `Sync` bound is met by +//! ownership and the channel entry reaps when the wrapper (and its +//! session) drops. For the egress the wrapper is not the handle +//! payload's burden — the egress owns the session in the table entry +//! (the POC's shape); the relay task owns the bare read half. The +//! accept variant meets the `+ Sync` bound with the [`SyncHalves`] +//! adapter (a `std::sync::Mutex` around the joined halves — a +//! `Mutex` is `Sync`, no unsafe) with the channel session +//! owned inside the payloads ([`ForwardedOwner`] — dropping the handle +//! drops the session AND the halves together). use std::collections::HashMap; +use std::net::SocketAddr; use std::sync::Arc; use alkcall::channels::client::ChannelClient; -use alktunnels::params::{Substrate, SubstrateAddr}; -use alktunnels::TunnelSession; +use alkcall::channels::operations::ChannelCore; +use alkcall::channels::pump::pump_bidi; +use alkcall::core::auth::AuthContext; +use alkcall::core::types::BiStream; +use alktunnels::params::{Substrate, SubstrateAddr, TunnelForwardedParams}; +use alktunnels::{TunnelListener, TunnelSession}; use bytes::Bytes; use futures::future::BoxFuture; use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::sync::oneshot; -use crate::egress::{Egress, EgressError, EgressFactory}; +use crate::egress::{ + BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError, BindTargetHandle, + Egress, EgressError, EgressFactory, +}; pub use crate::rfc1928::util::target_addr::TargetAddr; use crate::wire::MAX_INNER_LEN; @@ -404,3 +421,415 @@ pub fn direct_egress(client: Arc, max_tunnels: usize) -> EgressFa }) }) } +// --- The composed accept (ADR-007 variant 2, the sibling task's shape) ------ + +/// The correlation registry (the assembly-side bookkeeping): `listen` +/// token → the oneshot delivering the forwarded session (the adopted +/// channel's `BiStream` + the forwarded op's `peer` param). One entry +/// per live BIND session; the forwarded handler (the provider's shared +/// opener half) removes and completes the entry on correlation. A +/// dropped receiver (client-gone teardown) makes completion fail — the +/// handler drops the channel stream (teardown-at-birth). +pub type CorrelationRegistry = + Arc>>>; + +/// What a correlated forwarded session delivers: the forwarded +/// channel's `BiStream` (raw pass-through — the SOCKS5 BIND data plane +/// is raw on both ends) + the forwarded op's `peer` param (reply#2's +/// real peer — the graduation's win). Public because the +/// [`CorrelationRegistry`] it travels through is public API surface +/// (the assembly holds the handle [`register_forwarded_serving`] +/// returns); constructing one is the forwarded handler's job alone. +pub struct ForwardedSession { + pub read: tokio::io::ReadHalf, + pub write: tokio::io::WriteHalf, + pub peer: SubstrateAddr, +} + +/// The forwarded-op serving handler (the provider's shared opener +/// half): correlates the arriving forwarded open by `listen` token and +/// OWNS the data-plane pump (R-02 — the `register_openable` wrapper +/// awaits this task inline, then tears the forwarded channel down; a +/// handler that returns early reaps the channel at birth). The pump +/// bridges the forwarded channel's `BiStream` to a local duplex pair: +/// the near end pumps here (this task lives exactly as long as the +/// data plane — either end's EOF ends it), the far end travels through +/// the token's oneshot into the accept future's [`BindTargetHandle`] +/// payloads (dropping the handle drops the far end → the pump sees EOF +/// → the task returns → the wrapper reaps the forwarded channel — +/// teardown-at-birth, structural). A correlation miss (no waiting +/// session — client gone, or a token nobody minted) drops the stream: +/// the task returns immediately, the same reaping shape. A dead +/// oneshot receiver makes the send fail — the task drops the stream +/// and returns (same reaping). +fn make_forwarded_handler(registry: CorrelationRegistry) -> alktunnels::ForwardedHandler { + Arc::new(move |forwarded: TunnelForwardedParams, bidi: BiStream| { + let registry = Arc::clone(®istry); + tracing::debug!( + listen = forwarded.listen, + "composed accept: forwarded open arrived" + ); + tokio::spawn(async move { + let tx = registry.lock().await.remove(&forwarded.listen); + let Some(tx) = tx else { + tracing::warn!( + listen = forwarded.listen, + "composed accept: forwarded open with no waiting session; dropping" + ); + return; + }; + let (near, far) = tokio::io::duplex(64 * 1024); + let (far_read, far_write) = tokio::io::split(far); + if tx + .send(ForwardedSession { + read: far_read, + write: far_write, + peer: forwarded.peer, + }) + .is_err() + { + tracing::warn!( + listen = forwarded.listen, + "composed accept: waiting session vanished; dropping the forwarded channel" + ); + return; + } + // The middle pump (three-party half-close would cycle: + // pump_bidi's per-direction shutdown-on-completion + // deadlocks a bridge — the SOCKS5 pump's far.write shutdown + // needs near.read EOF; the wire's bidi.read EOF needs the + // producer pump's far.read EOF; far.read needs near.write + // shutdown; near.write needs bidi.read EOF). The bridge + // therefore completes on EITHER direction's EOF and drops + // everything: the far halves (the SOCKS5 pump sees EOF), + // the near halves, and the BiStream (the forwarded channel + // tears down; the wrapper's inline await returns — R-02 — + // and reaps the entry). + let (near_read, near_write) = tokio::io::split(near); + pump_bidi(bidi, near_read, near_write).await; + }) + }) +} + +/// The forwarded channel's `BiStream` reassembled after the +/// payload-split: the read half rides the [`SyncHalves`] adapter (the +/// mutex-joined form), the write half rides [`OwnedWrite`] — and the +/// owner rides inside the write payload so dropping the handle drops +/// everything: the teardown-at-birth discipline, compiler-enforced. +/// The owner is the [`TunnelListener`] (NOT an adopted session — the +/// forwarded channel's entry is the producer's, adopted inside the +/// producer's `ForwardedOpener`): the listener's keep-alive holds the +/// far accept loop (and every per-accept pump it spawned — R-02) +/// alive; dropping the handle drops the listener → the listen channel +/// EOFs → the far teardown cascades. The edition-2021 disjoint-capture +/// trap (an earlier draft took `listener.channel_id` — a `u32` — and +/// dropped the listener at the factory's return, EOF-ing the listen +/// channel at birth) is the reason the listener is forced into the +/// future with a whole-struct rebinding. +struct ForwardedOwner { + _listener: TunnelListener, +} + +/// The `BindTargetHandle` read-half adapter over the forwarded +/// channel's read half: `BiStream` is `Send + Unpin` but NOT `Sync` +/// (its inner is a `dyn AsyncReadWrite` trait object — no `Sync` +/// supertrait), so the raw halves cannot meet the handle payloads' +/// `+ Sync` bound directly. A `std::sync::Mutex` guard is the +/// no-unsafe Sync proof (`Mutex` is `Sync`); the RFC 1928 +/// BIND data plane is a raw pass-through pumped by ONE `pump_bidi` +/// task at a time, so contention never occurs at runtime — the mutex +/// is the Sync-proof shape, not a concurrency mechanism. A poisoned +/// lock (a panic in one poll) unwraps via `into_inner` — a panic in +/// one operation does not cascade (AGENTS.md convention 2). +struct SyncHalves { + inner: std::sync::Mutex>, +} + +impl AsyncRead for SyncHalves { + fn poll_read( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + let this = self.get_mut(); + let mut inner = match this.inner.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + std::pin::Pin::new(&mut *inner).poll_read(cx, buf) + } +} + +/// The write-half payload: the owner ([`ForwardedOwner`] — the +/// adopted channel session) and the write half ride inside one +/// `std::sync::Mutex` (the same no-unsafe Sync proof as +/// [`SyncHalves`] — `TunnelSession` and `WriteHalf` are +/// `Send` but not `Sync`). Contention never occurs at runtime (one +/// pump task); a poisoned lock unwraps via `into_inner`. +struct OwnedWrite { + inner: std::sync::Mutex, +} + +struct OwnedInner { + _owner: ForwardedOwner, + write: tokio::io::WriteHalf, +} + +impl AsyncWrite for OwnedWrite { + fn poll_write( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + let this = self.get_mut(); + let mut inner = match this.inner.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + std::pin::Pin::new(&mut inner.write).poll_write(cx, buf) + } + + fn poll_flush( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + let this = self.get_mut(); + let mut inner = match this.inner.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + std::pin::Pin::new(&mut inner.write).poll_flush(cx) + } + + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + let this = self.get_mut(); + let mut inner = match this.inner.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + std::pin::Pin::new(&mut inner.write).poll_shutdown(cx) + } +} + +/// The forwarded op's serving registration (the opener half — ADR-008 +/// §3, `register_tunnel_forwarded_openable` + this provider's +/// correlation handler): register on the SOCKS5 producer's serving +/// registry so its connection serves the forwarded op. Call once per +/// channels connection at assembly time — the registry is shared +/// across all BIND sessions every [`bindfirst_accept`] factory over +/// the same client fabricates (one token space per producer +/// connection). +/// +/// The scope gate is `tunnel:forwarded` on the PRODUCER's per-call +/// identity (the ingress grant — ADR-006 Amendment 1; never implied by +/// `tunnel:open`). The returned registry handle feeds the factories; +/// keep it alongside the client they ride. +/// +/// # Panics +/// +/// If the underlying registration fails (a duplicate +/// `channels/tunnel/forwarded` op id on the registry — an assembly +/// bug, not a runtime condition). `register_openable`'s failure type +/// is a `String` diagnostic, not a typed error. +pub fn register_forwarded_serving( + core: &ChannelCore, + on_registry: &Arc, + auth: AuthContext, +) -> CorrelationRegistry { + let registry: CorrelationRegistry = Arc::new(tokio::sync::Mutex::new(HashMap::new())); + alktunnels::producer::register_tunnel_forwarded_openable( + core, + on_registry, + auth, + make_forwarded_handler(Arc::clone(®istry)), + ) + .expect("register forwarded openable (the composed accept's serving half)"); + registry +} + +/// The composed accept factory (ADR-007 variant 2, producer.md +/// §Wiring `composed::bindfirst_accept`): the `BindAcceptFactory` the +/// producer registration consumes — the BIND accept whose listener +/// lives on an alktunnels 0.2.0 bind-first listen resource +/// (`register_tunnel_bindfirst_openable` + `BindFirstInjections` — +/// per-listener ACL for free, hub-relayable). +/// +/// Per session the factory: +/// +/// 1. Opens the listen channel (`TunnelListener::open` — the +/// establishment bound covers only this open-op round trip, +/// ADR-007); a failed open is a factory failure (reply#1's mapped +/// code — no phantom listener). +/// 2. Advertises reply#1's BND.ADDR as the REAL `bound` (the open-op +/// reply's `bound` field, the alkcall 0.8.0 reply projection — the +/// graduation's vanilla-fidelity win; no expected-addr stand-in). +/// A producer that supplied no `bound`/`listen` (not a bind-first +/// resource) fails the factory — no stand-in on this variant. +/// 3. Inserts the minted `listen` token → oneshot into the +/// correlation registry and returns the accept future parked on +/// it. +/// +/// The accept future waits for the ONE forwarded session matching the +/// token (the correlation registry is assembly-side bookkeeping — the +/// listener does not own the receipt queue), then builds the +/// [`BindTargetHandle`]: reply#2's peer is the forwarded op's `peer` +/// param (the graduation's win — no sentinel), and the data-plane +/// halves are the [`SyncHalves`]/[`OwnedWrite`] adapters over the +/// adopted channel's `BiStream` with the [`ForwardedOwner`] (the +/// adopted channel session) riding inside the write payload — dropping +/// the handle drops the session AND the halves together. +/// +/// **The client-gone teardown** (the seam contract, end-to-end): the +/// listener and the oneshot receiver ride the accept future — the +/// quiet wait's `select!` drop of that future drops the +/// [`TunnelListener`] (its held write half EOFs the listen channel — +/// the far accept dispatcher tears down) AND drops the oneshot +/// receiver (a later forwarded open for that token finds a dead +/// oneshot and drops the channel stream) — no leaked listener, no +/// leaked channel. +/// +/// Only TCP-stream listen resources are supported (the SOCKS5 BIND +/// data plane is raw on both ends; a UDP listen substrate would frame +/// — a datagram BIND variant is a different task). A non-TCP `bound` +/// fails the factory (`NotAllowed`). +pub fn bindfirst_accept( + client: Arc, + registry: CorrelationRegistry, + resource: impl Into, +) -> BindAcceptFactory { + let resource: Arc = resource.into().into(); + Arc::new(move |_target: TargetAddr| { + let client = Arc::clone(&client); + let registry = Arc::clone(®istry); + let resource = Arc::clone(&resource); + Box::pin(async move { + let params = alktunnels::TunnelParams { + resource: resource.to_string(), + substrate: Substrate::Tcp, + }; + let listener = TunnelListener::open(client.as_ref(), params) + .await + .map_err(|e| BindAcceptFactoryError::Refused(format!("listen open: {e}")))?; + let bind_addr = match listener.bound() { + Some(SubstrateAddr::HostPort { host, port }) => resolve_bound(host, *port) + .map_err(|e| BindAcceptFactoryError::General(format!("bound render: {e}")))?, + Some(SubstrateAddr::Path { path }) => { + return Err(BindAcceptFactoryError::NotAllowed(format!( + "non-tcp bound address `{path}` — this provider serves TCP listeners only" + ))); + } + None => { + return Err(BindAcceptFactoryError::Shortage( + "the producer supplied no `bound` reply field — not a bind-first \ + resource; no stand-in on this variant" + .into(), + )); + } + }; + let token = listener.listen().ok_or_else(|| { + BindAcceptFactoryError::Shortage( + "the producer supplied no `listen` correlation token — not a bind-first \ + resource; no stand-in on this variant" + .into(), + ) + })?; + let (tx, rx) = oneshot::channel(); + registry.lock().await.insert(token, tx); + tracing::debug!(token, "composed accept: registry entry inserted"); + + let fut = async move { + // Force the whole-struct capture: edition-2021 disjoint + // capture would take `listener.channel_id` (a `u32`) + // and drop the listener itself at the factory's return + // — EOF-ing the listen channel at birth. The listener + // rides the future end-to-end (the seam contract). + let listener = listener; + let ForwardedSession { read, write, peer } = match rx.await { + Ok(session) => session, + Err(_) => { + tracing::warn!( + token, + "composed accept: oneshot sender dropped; listener closing" + ); + // The registry entry was removed without + // delivery (the forwarded serving half tore + // down — the whole provider is going away). + // The listener's drop tears the listen + // channel down (the held write half EOFs). + return Err(BindAcceptError::General( + "the forwarded-op serving half went away; the listener closed".into(), + )); + } + }; + let peer_addr = peer_to_socket_addr(&peer); + Ok(BindTargetHandle { + peer: TargetAddr::Ip(peer_addr), + read: Box::new(SyncHalves { + inner: std::sync::Mutex::new(read), + }) as Box, + write: Box::new(OwnedWrite { + inner: std::sync::Mutex::new(OwnedInner { + // The listener rides INSIDE the payload: + // its keep-alive holds the far accept loop + // (and every per-accept pump it spawned — + // R-02) alive for the data plane's + // lifetime; dropping the handle drops the + // listener → the listen channel EOFs → + // the teardown cascades (teardown + // structural, compiler-enforced). The + // edition-2021 disjoint-capture trap (an + // earlier draft took `listener.channel_id` + // — a `u32` — and dropped the listener at + // the factory's return, EOF-ing the listen + // channel at birth) is why the whole + // struct is rebound into the future above. + _owner: ForwardedOwner { + _listener: listener, + }, + write, + }), + }), + }) + }; + Ok(BindAccept::new(bind_addr, fut)) + }) + }) +} + +/// Render the `bound` host (an IP literal or an unresolved host name) +/// + port into reply#1's `SocketAddr`: +/// +/// - IP literals resolve directly. +/// - A non-IP host fails (reply#1's address is a real socket address +/// on this variant — a hostname bound is the producer's resolution +/// surface, not the advertised address's). +/// +/// The parse error is a `String` diagnostic (the factory maps it to +/// `General`). +fn resolve_bound(host: &str, port: u16) -> Result { + host.parse::() + .map(|ip| SocketAddr::new(ip, port)) + .map_err(|e| format!("`{host}`: {e}")) +} + +/// The forwarded op's `peer` param rendered to reply#2's socket +/// address (the graduation's peer fidelity — the param is the real +/// peer). A `Path` peer (a unix-socket accept) has no RFC-shaped +/// address; the RFC sentinel `0.0.0.0:0` carries it (the data plane +/// is unaffected either way). +fn peer_to_socket_addr(peer: &SubstrateAddr) -> SocketAddr { + match peer { + SubstrateAddr::HostPort { host, port } => host + .parse::() + .map(|ip| SocketAddr::new(ip, *port)) + .unwrap_or_else(|_| sentinel_peer()), + SubstrateAddr::Path { .. } => sentinel_peer(), + } +} + +fn sentinel_peer() -> SocketAddr { + SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), 0) +} diff --git a/src/lib.rs b/src/lib.rs index 3944e07..588a3bb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -123,7 +123,8 @@ pub use local::{ProducerDoor, ProducerDoorBuilder, ProducerDoorGuard}; #[cfg(feature = "composed")] pub use composed::{ - direct_egress, key_of, substrate_addr_of, ComposedEgress, SessionOwner, DEFAULT_MAX_TUNNELS, + bindfirst_accept, direct_egress, key_of, register_forwarded_serving, substrate_addr_of, + ComposedEgress, CorrelationRegistry, ForwardedSession, SessionOwner, DEFAULT_MAX_TUNNELS, }; #[cfg(feature = "local")] diff --git a/tasks/implementation/composed-accept.md b/tasks/implementation/composed-accept.md index ac1a9aa..7e59c11 100644 --- a/tasks/implementation/composed-accept.md +++ b/tasks/implementation/composed-accept.md @@ -1,7 +1,7 @@ --- id: implementation/composed-accept name: Composed — the bind-first accept (the listener on a listen resource) -status: pending +status: completed depends_on: [implementation/composed-egress] scope: moderate risk: medium @@ -115,8 +115,76 @@ Deliverables: ## Notes -> To be filled by implementation agent +- **The correlation shape** (the design center): a + `CorrelationRegistry` (`Arc>>>`) keyed on the `listen` token. + The factory inserts the token → oneshot right after + `TunnelListener::open`; the provider's shared forwarded handler + (`make_forwarded_handler`, registered via + `register_forwarded_serving` on the SOCKS5 producer's connection) + removes the entry on the arriving forwarded open and delivers + `ForwardedSession { far duplex halves, peer }`. A dropped receiver + (client-gone) or a miss makes the handler drop the forwarded + channel stream — teardown-at-birth, R-02 (the inline `pump_bidi` + await in `register_openable`'s wrapper then reaps the entry). +- **The Sync-bound wrinkle**: alktunnels' halves are `Send + Unpin` + only; the landed `BindTargetHandle` payloads carry `+ Sync`. Two + adapters pin the no-unsafe Sync proof (`Mutex` is `Sync`; + contention never occurs — one pump task at a time): `SyncHalves` + (read half) and `OwnedWrite` (write half + the owner inside). +- **The owner-wrapper rides inside the payloads** + (`ForwardedOwner`): the write payload owns the `TunnelListener` — + dropping the handle drops the listener → the listen channel EOFs → + the far teardown cascades (teardown structural, + compiler-enforced). The edition-2021 disjoint-capture trap (an + earlier draft captured `listener.channel_id`, dropping the + listener at the factory's return — EOF at birth) is why the whole + struct is rebound into the accept future. +- **The three-party bridge pump**: the forwarded handler bridges the + forwarded `BiStream` to a local duplex pair whose far end travels + into the accept future's payloads. The bridge completes on EITHER + direction's EOF (a `pump_bidi`-then-drop, not a per-direction + shutdown cascade — a three-party half-close cycle would deadlock: + each pump's completion-shutdown needs the next hop's EOF, and the + chain closes on itself). Dropping everything EOFs the SOCKS5 + producer-side pump, ends the per-accept pump, and tears the + forwarded channel down. +- **Test concession (producer-side adopted forwarded entry)**: the + producer's `open_forwarded` caller adopts the opener-allocated + forwarded channel id on the producer-side manager; that entry's + reap rides the per-adopter contract + transport EOF, NOT the + listener's teardown (alktunnels ADR-005; upstream's own bind-first + suites make the same concession — `producer_bindfirst.rs` + `no_stray_channels` documents the listen channel, and the + forwarded entries are adopted sessions owned by their adopter). + `assert_only_channels` therefore treats odd producer-side ids as + the adopted-forwarded shape. +- **Forwarded-open failure semantics**: a FORBIDDEN forwarded open is + per-accept isolated AT THE TUNNEL LAYER (the producer's accept loop + drops that connection and keeps serving; the alktunnels + producer-suite pin). The SOCKS5 accept future stays parked — the + SOCKS5 producer never learns of the tunnel-layer failure — so no + reply#2 arrives; the quiet wait is unaffected. Pinned by test. ## Summary -> To be filled on completion \ No newline at end of file +Landed `composed::bindfirst_accept` + `composed::register_forwarded_serving` +(the composed accept, ADR-007 variant 2 / ADR-008) on the alktunnels 0.2.0 +bind-first listen resource, at the graduation's vanilla fidelity: reply#1 +advertises the open-op reply's real `bound` (the alkcall 0.8.0 reply +projection; no expected-addr stand-in), reply#2 carries the forwarded op's +real `peer` param. The correlation registry (token → oneshot) is assembly +bookkeeping; the provider owns the forwarded-op serving half, the token→waiter +map, and the per-forwarded middle pump (either-EOF completion shape). The +Sync-bound wrinkle is met by `SyncHalves`/`OwnedWrite` (mutex Sync-proof, no +unsafe) with the owner (`ForwardedOwner`, the `TunnelListener`) riding inside +the write payload — dropping the handle drops everything (teardown structural, +compiler-enforced). `tests/composed-accept.rs` pins: the full story (real +`bound`, real `peer`, raw pass-through round trip), the client-gone teardown +(drop-witness: the accept future owns the listener end-to-end; no leaked +listen channel), two concurrent BIND sessions' distinct tokens, per-accept +isolation of a FORBIDDEN forwarded open, and the late-forwarded correlation +miss dropping the channel at birth. Verified: `cargo test`, `cargo test +--all-features`, `cargo check/clippy --target wasm32-unknown-unknown` (hard +gate), `cargo clippy --all-targets --all-features -- -D warnings`, +`cargo fmt --check` — all clean. \ No newline at end of file diff --git a/tests/composed-accept.rs b/tests/composed-accept.rs new file mode 100644 index 0000000..42c07b7 --- /dev/null +++ b/tests/composed-accept.rs @@ -0,0 +1,502 @@ +//! The composed accept integration tests (ADR-007 variant 2 — the +//! BIND accept whose listener lives on an alktunnels 0.2.0 bind-first +//! listen resource, end to end over channels): the graduation's +//! vanilla fidelity (reply#1 = the real `bound`; reply#2 = the +//! forwarded op's real `peer` — no stand-ins), the raw pass-through +//! data plane, the client-gone teardown (drop-witness), two concurrent +//! BIND sessions' distinct tokens, and the per-accept isolation. +//! +//! The topology: the opener (the SOCKS5 consumer side) dials the +//! transport; the tunnel producer serves BOTH the alktunnels bind-first +//! open op and the SOCKS5 open op on one registry. The forwarded op +//! serves on the opener side (post-hoc `register_forwarded_serving` — +//! the correlation registry the composed accept factories ride). + +#![cfg(feature = "composed")] + +mod harness; + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use alksocks::composed::{bindfirst_accept, register_forwarded_serving}; +use alksocks::consumer::BindSession; +use alksocks::rfc1928::util::target_addr::ToTargetAddr; +use alktunnels::params::{Substrate, SubstrateAddr}; +use alktunnels::producer::{ + BindFirstAcceptLoop, BindFn, BoundListener, ResourceRegistry, TargetHandle, + TunnelEstablishError, +}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::sync::Notify; + +/// The resource the bind-first producer serves (the tests' only +/// subscription — the registry namespace gate). +const RESOURCE: &str = "bind"; + +/// The bound address the test bind reports (the "OS-chosen" stand-in — +/// the test's queue listener owns it; fidelity is the assert's job). +const BOUND_PORT: u16 = 42113; + +/// The peer address the accepted pairs report (the "app server" — the +/// forwarded op's `peer` param carries it to reply#2). +const PEER_PORT: u16 = 55231; + +/// The test bind (the assembly's bind seam): reports the test's +/// bound address + an opaque listener witness. +fn test_bind() -> BindFn { + Arc::new(|_substrate, _backing| { + Box::pin(async { + Ok(BoundListener { + listener: Arc::new(()) as Arc, + bound: SubstrateAddr::HostPort { + host: "127.0.0.1".to_string(), + port: BOUND_PORT, + }, + }) + }) + }) +} + +/// The accept-queue pair the test accept loop pops from (the alktunnels +/// listener test's shape: a VecDeque + Notify — one accepted pair per +/// dispatcher call). +type AcceptQueue = ( + Arc>>, + Arc, +); + +fn fresh_queue() -> AcceptQueue { + ( + Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new())), + Arc::new(Notify::new()), + ) +} + +/// An accepted "connection" for the test accept loop: a duplex pair +/// with an echo task on the far end (the app server's socket stand-in +/// — the data plane's echo confirms the full round trip). +fn accepted_halves(peer_addr: (&str, u16)) -> (SubstrateAddr, TargetHandle) { + let (consumer_side, target_side) = tokio::io::duplex(64 * 1024); + let (mut t_read, mut t_write) = tokio::io::split(target_side); + tokio::spawn(async move { + let mut buf = vec![0u8; 8192]; + loop { + match t_read.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => { + if t_write.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + } + }); + let (c_read, c_write) = tokio::io::split(consumer_side); + ( + SubstrateAddr::HostPort { + host: peer_addr.0.to_string(), + port: peer_addr.1, + }, + TargetHandle { + read: Box::new(c_read), + write: Box::new(c_write), + }, + ) +} + +/// The test accept loop (the dispatcher invokes the closure per +/// accept): pop one accepted pair, or park until notified. +fn queue_loop(queue: AcceptQueue) -> BindFirstAcceptLoop { + let (queue, notify) = queue; + Arc::new(move |_listener| { + let queue = Arc::clone(&queue); + let notify = Arc::clone(¬ify); + Box::pin(async move { + let popped = { + let notified = std::pin::pin!(notify.notified()); + let mut q = queue.lock().await; + if let Some(pair) = q.pop_front() { + Some(pair) + } else { + drop(q); + notified.await; + let mut q = queue.lock().await; + q.pop_front() + } + }; + popped.ok_or(TunnelEstablishError::DialFailed( + "listener closed".to_string(), + )) + }) + }) +} + +/// Push an accepted pair into the test's accept queue. +async fn push_accept(queue: &AcceptQueue, peer: (&str, u16)) { + let (peer_addr, handle) = accepted_halves(peer); + queue.0.lock().await.push_back((peer_addr, handle)); + queue.1.notify_waiters(); +} + +/// The full wiring: the bind-first topology with the composed accept +/// factory + the forwarded serving registered post-hoc on the opener's +/// registry. Returns everything the tests drive. +struct Suite { + topo: harness::BindFirstTopology, + queue: AcceptQueue, +} + +async fn wire_suite() -> Suite { + wire_suite_with_identity(harness::composed_opener_identity()).await +} + +async fn wire_suite_with_identity(identity: alkcall::core::auth::Identity) -> Suite { + let resource_registry = ResourceRegistry::new(); + resource_registry + .register(RESOURCE, Substrate::Tcp, "assembly-owned") + .await; + let queue = fresh_queue(); + let topo = harness::wire_bindfirst_composed( + resource_registry, + test_bind(), + queue_loop(queue.clone()), + None, // the SOCKS5 openable registers post-hoc, below — the composed factory needs the opener's client + identity, + ) + .await; + // The correlation registry rides the opener's serving registry + // (the forwarded op's serving half — the provider's opener half). + let registry = register_forwarded_serving( + &alkcall::channels::operations::ChannelCore::new( + topo.opener.manager().clone(), + alkcall::channels::policy::default_policy(), + ), + &topo.opener_registry, + alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"), + ); + // The SOCKS5 openable registers post-hoc (W2's shape) on the + // producer's registry — with the REAL composed factory, which + // closes over the opener's client + the correlation registry (both + // only exist now). + let factory = bindfirst_accept(Arc::clone(&topo.opener), Arc::clone(®istry), RESOURCE); + let core = alkcall::channels::operations::ChannelCore::new( + topo.producer_manager.clone(), + alkcall::channels::policy::default_policy(), + ); + alksocks::producer::Socks5Producer::builder() + .accept(factory) + .build() + .register( + &core, + &topo.producer_registry, + alkcall::core::auth::AuthContext::anonymous(b"alk/socks5"), + None, + ) + .expect("register socks5 openable post-hoc (the composed accept factory)"); + Suite { topo, queue } +} + +/// Open a BIND session through the composed accept (the SOCKS5 BIND +/// client over channels — the consumer session type). The session +/// drives the whole pipeline: open → negotiate → BIND → reply#1 (the +/// factory's real `bound`) → quiet wait (the factory's accept future) +/// → reply#2 (the real peer). +async fn bind_session(suite: &Suite) -> BindSession { + BindSession::bind( + suite.topo.opener.as_ref(), + ("bind.test", 2121).to_target_addr().unwrap(), + ) + .await + .expect("the session binds through the composed accept") +} + +/// The no-leak probe: the opener side ends with channel 0 (+ the +/// live channels the test names); the producer side ends with channel +/// 0, the SOCKS5 channels the test names, AND any odd id — the +/// producer-side ADOPTED forwarded channels (opener-allocated ids, +/// adopted inside the producer's `open_forwarded` caller; per-session +/// ownership per alktunnels ADR-005 — their entries reap at transport +/// EOF, not at the listener's teardown; upstream's own bind-first +/// suites make the same concession in `no_stray_channels`). +async fn assert_only_channels( + topo: &harness::BindFirstTopology, + opener_extra: &[u32], + producer_extra: &[u32], +) { + let opener_ids = topo.opener.manager().channel_ids(); + let producer_ids = topo.producer_manager.channel_ids(); + let opener_ok = opener_ids + .iter() + .all(|id| *id == 0 || opener_extra.contains(id)); + let producer_ok = producer_ids + .iter() + .all(|id| *id == 0 || *id % 2 == 1 || producer_extra.contains(id)); + assert!( + opener_ok && producer_ok, + "channel sets: opener={opener_ids:?} producer={producer_ids:?}" + ); +} + +/// THE gate: the full composed BIND story — reply#1 carries the real +/// `bound` (the graduation's vanilla fidelity), an app server "dials" +/// (the queue accept), reply#2 carries the real peer (the forwarded +/// op's `peer` param), and the pass-through data plane round-trips +/// through the forwarded channel to the echo task and back. +#[tokio::test] +async fn composed_bind_full_story_bound_peer_and_round_trip() { + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::level_filters::LevelFilter::TRACE) + .with_test_writer() + .try_init(); + let suite = wire_suite().await; + + // The BIND session rides the real composed factory (the SOCKS5 + // registration's accept seam — replaced post-hoc per-suite; the + // registration reads through the shared registry per dispatch, so + // the session's open dispatches with the factory in place). + let mut session = bind_session(&suite).await; + let channel_id = session.channel_id(); + assert!( + suite.topo.producer_manager.has_channel(channel_id), + "the SOCKS5 channel is live on the producer side" + ); + + // Reply#1: the REAL bound (the open-op reply's `bound` field — no + // expected-addr stand-in; the test bind's reported address). + let bind_addr = session.reply1().expect("reply#1").bind_addr; + assert_eq!( + bind_addr, + alksocks::rfc1928::util::target_addr::TargetAddr::Ip(SocketAddr::from(( + [127, 0, 0, 1], + BOUND_PORT + ))), + "reply#1 advertises the real OS-chosen bound" + ); + + // The app server dials the far listener (the queue accept) — the + // producer's accept loop opens the forwarded channel toward the + // opener; the correlation registry delivers it into the accept + // future; reply#2 carries the real peer. + push_accept(&suite.queue, ("10.0.0.1", PEER_PORT)).await; + let mut plane = tokio::time::timeout(Duration::from_secs(10), session.await_inbound()) + .await + .expect("reply#2 within timeout") + .expect("await_inbound ok") + .expect("the plane exists"); + assert_eq!( + plane.peer, + alksocks::rfc1928::util::target_addr::TargetAddr::Ip(SocketAddr::from(( + [10, 0, 0, 1], + PEER_PORT + ))), + "reply#2 carries the forwarded op's real peer" + ); + + // The pass-through data plane: client bytes cross the SOCKS5 + // channel → the producer's pump → the forwarded channel → the + // accepted halves' echo → and back. + plane.write_all(b"bind-ping").await.expect("write"); + plane.flush().await.expect("flush"); + let mut buf = [0u8; 9]; + tokio::time::timeout(Duration::from_secs(10), plane.read_exact(&mut buf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&buf, b"bind-ping"); + + // Teardown: the session reaps its SOCKS5 channel; the data plane's + // halves drop (the owner wrapper reaps the forwarded channel). + drop(plane); + assert!(session.close().await, "the SOCKS5 channel reaped"); + // The teardown cascade: the data plane's halves drop → the + // forwarded handler's pump EOFs → the wrapper reaps the forwarded + // channel → the listener dropped with the handle → the listen + // channel EOFs → the far accept loop exits. Settle, then assert. + for _ in 0..50 { + tokio::time::sleep(Duration::from_millis(100)).await; + let opener_ids = suite.topo.opener.manager().channel_ids(); + if opener_ids.iter().all(|&id| id == 0) { + break; + } + } + // The opener side holds only channel 0 (the producer-allocated + // forwarded channel's entry rides the opener's manager per its own + // adopter contract — alktunnels ADR-005). The producer side may + // retain the adopted forwarded channel (see `assert_only_channels` + // — its adopter's session dropped with the plane; the entry reaps + // at transport EOF, not at the listener's). + assert_only_channels(&suite.topo, &[0], &[channel_id]).await; +} + +/// The client-gone teardown (the drop-witness): dropping the BIND +/// session mid-quiet-wait releases the far listener — the accept +/// future owns it end-to-end (the quiet wait's `select!` drop IS the +/// teardown). +#[tokio::test] +async fn client_gone_teardown_releases_the_listener() { + let suite = wire_suite().await; + let mut session = bind_session(&suite).await; + let socks5_channel = session.channel_id(); + let _ = session.reply1().expect("reply#1"); + + // The client vanishes (drop = the quiet wait's select! observes + // the channel EOF → the accept future drops → the TunnelListener + // inside it drops → the listen channel EOFs → the far accept + // dispatcher tears down). + drop(session); + + // The SOCKS5 channel reaps; the listen channel (the far listener's + // carrier) reaps too — no leaked listener, no leaked channel. + let mut socks5_gone = false; + let mut listen_gone = false; + for _ in 0..50 { + tokio::time::sleep(Duration::from_millis(200)).await; + socks5_gone = !suite.topo.producer_manager.has_channel(socks5_channel); + // The listen channel's id is unknown here (the accept future + // held it); the aggregate leak probe is the witness: after the + // teardown cascade every channel is gone on both sides. + listen_gone = suite + .topo + .producer_manager + .channel_ids() + .iter() + .all(|&id| id == 0); + if socks5_gone && listen_gone { + break; + } + } + assert!( + socks5_gone, + "the SOCKS5 channel reaped after the client dropped" + ); + assert!( + listen_gone, + "the listen channel reaped (no leaked listener): producer={:?}", + suite.topo.producer_manager.channel_ids() + ); + // The accept queue was never consumed (the listener died before + // any app server arrived). + assert!( + suite.queue.0.lock().await.is_empty(), + "the dropped listener consumed nothing" + ); + assert_only_channels(&suite.topo, &[0], &[0]).await; +} + +/// Two concurrent BIND sessions get distinct tokens (the mint is +/// per-open); each accept future correlates only ITS forwarded +/// session. +#[tokio::test] +async fn concurrent_bind_sessions_get_distinct_tokens() { + let suite = wire_suite().await; + let mut s1 = bind_session(&suite).await; + let mut s2 = bind_session(&suite).await; + assert_ne!(s1.channel_id(), s2.channel_id(), "distinct SOCKS5 channels"); + let _ = s1.reply1().expect("reply#1 s1"); + let _ = s2.reply1().expect("reply#2 s2"); + + // Two accepts: the tokens are distinct (two listen opens minted + // distinct tokens — observably: both sessions resolve their own + // forwarded sessions, the correlation delivered each to the right + // waiter). + push_accept(&suite.queue, ("10.0.0.1", PEER_PORT)).await; + push_accept(&suite.queue, ("10.0.0.2", PEER_PORT + 1)).await; + + let mut p1 = tokio::time::timeout(Duration::from_secs(10), s1.await_inbound()) + .await + .expect("s1 reply#2") + .expect("s1 plane ok") + .expect("s1 plane"); + let mut p2 = tokio::time::timeout(Duration::from_secs(10), s2.await_inbound()) + .await + .expect("s2 reply#2") + .expect("s2 plane ok") + .expect("s2 plane"); + + // Each plane pumps only its own accepted connection (the + // correlation routed by token): distinct peer ports arrive, each + // on the session whose listener accepted it. + let peers = [p1.peer.clone(), p2.peer.clone()]; + assert_ne!(peers[0], peers[1], "the two accepts routed distinctly"); + + // Both data planes round-trip independently. + p1.write_all(b"one").await.expect("p1 write"); + p2.write_all(b"two").await.expect("p2 write"); + let mut b1 = [0u8; 3]; + let mut b2 = [0u8; 3]; + tokio::time::timeout(Duration::from_secs(10), p1.read_exact(&mut b1)) + .await + .expect("p1 echo") + .expect("p1 read"); + tokio::time::timeout(Duration::from_secs(10), p2.read_exact(&mut b2)) + .await + .expect("p2 echo") + .expect("p2 read"); + assert_eq!(&b1, b"one"); + assert_eq!(&b2, b"two"); + + drop(p1); + drop(p2); + assert!(s1.close().await); + assert!(s2.close().await); + tokio::time::sleep(Duration::from_millis(300)).await; + assert_only_channels(&suite.topo, &[0], &[0]).await; +} + +/// A forwarded-open failure is per-accept isolated: an opener identity +/// without `tunnel:forwarded` makes the forwarded open fail FORBIDDEN +/// — the BIND session's accept future never resolves a forwarded +/// session (reply#2's mapped refusal), the listen channel stays +/// usable, and the failure does not tear the listener down. +#[tokio::test] +async fn forwarded_open_failure_is_per_accept_isolated() { + let suite = wire_suite_with_identity(harness::composed_scopeless_opener_identity()).await; + let mut session = bind_session(&suite).await; + let _ = session.reply1().expect("reply#1"); + + // The accept fires (the queue accept consumed — the producer's + // accept loop is alive), but the forwarded open fails FORBIDDEN. + push_accept(&suite.queue, ("10.0.0.1", PEER_PORT)).await; + + // The forwarded-open failure is per-accept isolated AT THE TUNNEL + // LAYER: the producer's accept loop drops that accepted connection + // and keeps serving. The accept future stays parked (the SOCKS5 + // producer never learns of the tunnel-layer failure — the + // quiet-wait shape is unaffected), so the queue drains both + // accepts and the listener survives (the alktunnels producer + // suite's `forwarded_open_failure_isolates_the_accept` pin, end to + // end). + push_accept(&suite.queue, ("10.0.0.1", PEER_PORT)).await; + tokio::time::sleep(Duration::from_millis(300)).await; + assert!( + suite.queue.0.lock().await.is_empty(), + "the accept loop consumed both accepts (alive throughout)" + ); + drop(session); + tokio::time::sleep(Duration::from_millis(300)).await; +} + +/// The dropped listener (client gone mid-quiet-wait) leaves a +/// forwarded open nowhere to go: the correlation miss drops the +/// forwarded channel (teardown-at-birth — pinned by the dead-oneshot +/// probe's aggregate-leak witness). +#[tokio::test] +async fn late_forwarded_open_after_client_gone_drops_the_channel() { + let suite = wire_suite().await; + let mut session = bind_session(&suite).await; + let _ = session.reply1().expect("reply#1"); + drop(session); + + // The client is gone; the listener is being torn down. An accept + // arriving in the race window may reach the dispatcher before the + // EOF sentinel does; the forwarded open then finds a dead oneshot + // (the correlation miss) and drops the channel. + push_accept(&suite.queue, ("10.0.0.1", PEER_PORT)).await; + tokio::time::sleep(Duration::from_millis(500)).await; + // Either the listener was already gone (the accept queued + // unconsumed) or the forwarded channel dropped at birth — either + // way: no leaked channels on the opener side. + assert_only_channels(&suite.topo, &[0], &[0]).await; +} diff --git a/tests/harness.rs b/tests/harness.rs index 1bd9f9d..5f96950 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -951,6 +951,266 @@ where // --- Tests ------------------------------------------------------------------ +// --- The composed accept's topology (the bind-first tunnel producer) --- + +/// The bind-first topology (the composed accept suite's shape — +/// alktunnels' `wire_listener_bindfirst` adapted): the opener dials +/// the transport (the SOCKS5 consumer side — its `ChannelClient` is +/// what `TunnelListener::open` and the SOCKS5 BIND client ride) and +/// the tunnel producer accepts (the adapter + an install hook +/// registering the alktunnels bind-first openable over the given +/// bind/accept-loop injections AND the SOCKS5 producer openable with +/// the composed accept factory). One connection carries both +/// protocols' ops — the producer serves `alk/tunnel`'s open op and the +/// forwarded op, plus the SOCKS5 open op, on one registry. +/// +/// The opener identity carries `tunnel:open` + `tunnel:forwarded` + +/// `socks5:open` + `socks5:bind` (the four gates the composed BIND +/// story crosses: the listen open, the forwarded ingress, the SOCKS5 +/// session open, the BIND command gate). +pub struct BindFirstTopology { + /// The opener's client — `TunnelListener::open` + the SOCKS5 BIND + /// session + the composed accept factory all ride it. + pub opener: Arc, + /// The opener's serving registry — the forwarded op lands here + /// (via `register_forwarded_serving`). + pub opener_registry: Arc, + /// The producer's manager (accept side, serving) — reaping probes. + pub producer_manager: ChannelManager, + /// The producer's serving registry (probes). + pub producer_registry: Arc, + /// What the bind-first establisher's per-call auth carried (the + /// CF-006 probe). + pub identity_witness: Arc>>, +} + +/// Wire the bind-first topology over one duplex transport. The +/// producer's install hook registers — on its serving registry — the +/// alktunnels bind-first openable (over the given `bind` + the given +/// accept loop) and the SOCKS5 openable over a composed accept factory +/// built from `register_forwarded_serving` on the OPENER side... but +/// the forwarded op serves on the opener's registry, which only +/// exists after the opener's client is built. The hook therefore +/// defers the forwarded registration: the test wires it post-hoc via +/// [`alksocks::composed::register_forwarded_serving`] on +/// `opener_registry` (W2's post-hoc posture — the dispatcher reads +/// through the shared `Arc` per dispatch). +pub async fn wire_bindfirst_composed( + registry: alktunnels::producer::ResourceRegistry, + bind: alktunnels::producer::BindFn, + accept_loop: alktunnels::producer::BindFirstAcceptLoop, + accept_factory: Option, + opener_identity: Identity, +) -> BindFirstTopology { + let producer_op_registry = Arc::new(OperationRegistry::new()); + let opener_op_registry = Arc::new(OperationRegistry::new()); + let registry_for_hook = Arc::clone(&producer_op_registry); + let opener_registry_for_struct = Arc::clone(&opener_op_registry); + let (producer_manager_tx, mut producer_manager_rx) = + tokio::sync::mpsc::channel::(1); + let identity_witness: Arc>> = + Arc::new(tokio::sync::Mutex::new(None)); + let bind_witness = Arc::clone(&identity_witness); + let opener_identity_for_hook = opener_identity.clone(); + + // --- producer (accept side, serving): adapter + install hook --- + let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| { + let registry_arc = Arc::clone(®istry_for_hook); + let resource_registry = registry.clone(); + let bind = Arc::clone(&bind); + let accept_loop = Arc::clone(&accept_loop); + let accept = accept_factory.clone(); + let witness = Arc::clone(&bind_witness); + let manager_tx = producer_manager_tx.clone(); + let caller = opener_identity_for_hook.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); + // The accept-side identity posture (the load-bearing POC #1 + // requirement): attach the resolved identity to channel 0 + // BEFORE building the `CallConnection`. + let _ = channel0_conn.set_identity(caller); + 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_arc) + .expect("register channel ops on producer"); + + // The forwarded-open caller (the call plane — alktunnels + // harness's shape): call + adopt on the serving session's + // `CallConnection`. + let call_for_opener = Arc::clone(&call_connection); + let manager_for_opener = manager.clone(); + let open_forwarded: alktunnels::producer::ForwardedOpener = + Arc::new(move |forwarded: alktunnels::TunnelForwardedParams| { + let call = Arc::clone(&call_for_opener); + let manager = manager_for_opener.clone(); + Box::pin(async move { + let input = serde_json::to_value(&forwarded).map_err(|e| { + alktunnels::TunnelEstablishError::HandlerError(format!( + "forwarded params serialize: {e}" + )) + })?; + let response = call.call(alktunnels::OP_TUNNEL_FORWARDED, input).await; + let out = response.result.map_err(|e| { + alktunnels::TunnelEstablishError::HandlerError(format!( + "forwarded open: {}", + e.message + )) + })?; + let channel_id = + out.get("channel_id") + .and_then(|v| v.as_u64()) + .ok_or_else(|| { + alktunnels::TunnelEstablishError::HandlerError( + "forwarded open reply missing channel_id".to_string(), + ) + })? as u32; + let (send, recv) = manager + .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None) + .await + .map_err(|e| { + alktunnels::TunnelEstablishError::HandlerError(format!( + "forwarded adopt: {e}" + )) + })?; + Ok((send, recv)) + }) + }); + + alktunnels::producer::register_tunnel_bindfirst_openable( + &core, + &resource_registry, + ®istry_arc, + AuthContext::anonymous(b"alk/tunnel"), + alktunnels::producer::BindFirstInjections { + bind, + accept_loop, + open_forwarded, + }, + Some(witness), + ) + .expect("register tunnel bindfirst openable"); + + // The SOCKS5 openable rides the same registry when the + // accept factory is available at hook time (the + // static-factory shape). When the factory closes over + // topology pieces that only exist after the opener's + // client is built (the composed accept's factory needs the + // opener's `ChannelClient`), the registration defers to + // the post-hoc W2 shape — the caller registers it on the + // returned `producer_registry` (the dispatcher reads + // through the shared `Arc` per + // dispatch). + if let Some(accept) = accept { + let producer = alksocks::producer::Socks5Producer::builder() + .accept(accept) + .build(); + producer + .register( + &core, + ®istry_arc, + AuthContext::anonymous(b"alk/socks5"), + None, + ) + .expect("register socks5 openable (the composed accept topology)"); + } + + let _ = manager_tx.send(manager).await; + let dp = Dispatcher::new( + registry_arc, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ); + dp.serve_single_stream(call_connection, reader, writer) + .await; + }) + }); + + // --- transport ------------------------------------------------------ + let (opener_end, producer_end) = tokio::io::duplex(64 * 1024); + let opener_conn = CoreConnection::from_bidi(opener_end, b"alk/channels".to_vec(), None); + let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None); + opener_conn + .set_identity(opener_identity.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; + }); + + // --- opener (connect side): client + serving (the forwarded op + // registers post-hoc, per the type docs) ----------------------------- + let opener_client = ChannelClient::from_connection_with_serving( + opener_conn, + Some(ServingConfig { + registry: Arc::clone(&opener_op_registry), + identity_provider: Arc::new(alkcall::core::auth::NoopIdentityProvider), + identity: None, + }), + ) + .await + .expect("opener channel client init"); + + let producer_manager = producer_manager_rx + .recv() + .await + .expect("producer manager captured"); + + BindFirstTopology { + opener: Arc::new(opener_client), + opener_registry: opener_registry_for_struct, + producer_manager, + producer_registry: producer_op_registry, + identity_witness, + } +} + +/// The composed suite's opener identity: the four gates the composed +/// BIND story crosses (`tunnel:open` + `tunnel:forwarded` + `socks5:open` +/// + `socks5:bind`). +pub fn composed_opener_identity() -> Identity { + Identity { + id: "composed-opener".to_string(), + scopes: vec![ + alktunnels::TUNNEL_OPEN_SCOPE.to_string(), + alktunnels::TUNNEL_FORWARDED_SCOPE.to_string(), + SOCKS5_OPEN_SCOPE.to_string(), + SOCKS5_BIND_SCOPE.to_string(), + ], + resources: HashMap::new(), + } +} + +/// [`composed_opener_identity`] minus the forwarded grant — the +/// per-accept isolation probe's identity (the forwarded open fails +/// FORBIDDEN; the accept is consumed, the listener lives). +pub fn composed_scopeless_opener_identity() -> Identity { + Identity { + id: "composed-opener-scopeless".to_string(), + scopes: vec![ + alktunnels::TUNNEL_OPEN_SCOPE.to_string(), + SOCKS5_OPEN_SCOPE.to_string(), + SOCKS5_BIND_SCOPE.to_string(), + ], + resources: HashMap::new(), + } +} + #[cfg(test)] mod tests { use super::*;