diff --git a/docs/architecture/README.md b/docs/architecture/README.md index 5104b9f..91bd455 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -1,7 +1,8 @@ --- status: draft -last_updated: 2026-09-24 (the protocol core landed + the cross-cutting -integration suite; the rfc1928 doc's targeted fork 2026-09-22; the +last_updated: 2026-09-25 (the assembly layer landed + the cross-cutting +assembly integration suite; the protocol core landed + the cross-cutting +integration suite 2026-09-24; the rfc1928 doc's targeted fork 2026-09-22; the vendor extraction amendment ADR-011 §Amendment 2026-09-21; Phase 1 initial draft 2026-09-20 — the OQ-SK ledger promoted into ADRs 001-012 and open-questions.md) @@ -20,17 +21,17 @@ architecture spec docs and the ADRs. ## Current State -Phase 1 (Architecture) — the **protocol core is implemented**. Phase +Phase 1 (Architecture) — the **protocol core and the assembly layer +are implemented**. Phase 0 (`docs/research/phase-0.md`) is converged (2026-09-14) and POC-complete (all numbered POCs passed or were dropped with the scope decision); the OQ-SK ledger's Phase 1 inputs are collected and -promoted into ADRs 001–013. All docs are `draft` — they transition +promoted into ADRs 001–014. All docs are `draft` — they transition to `reviewed` when their open questions resolve and the architecture review passes. -The landed core (the protocol-only batch; no `local`/`composed` code -exists yet): the RFC 1928 core in `src/rfc1928/` (the targeted fork, -ADR-013 — the vendor group's six tasks landed and passed the +The landed surface: the RFC 1928 core in `src/rfc1928/` (the targeted +fork, ADR-013 — the vendor group's six tasks landed and passed the extraction review), the ASSOCIATE datagram frame codec (`src/wire.rs`, ADR-003), the open-op params + typed error surface (`src/params.rs` / `src/error.rs`, ADR-001), the three seam types (`src/egress.rs`, @@ -38,9 +39,17 @@ ADR-004), the producer half (`src/producer.rs` — the session-gate establisher, the command-dispatch pump, the three commands, the scope gates), and the consumer half (`src/consumer.rs` — the connect/associate/bind session types with teardown ownership). The -batch-1 cross-cutting integration suite (`tests/core-integration.rs`) -pins the core's whole behavior end to end; the `local` and -`composed` provider groups follow (batch 2). +assembly layer (batch 2) landed with it: the `local` providers +(`src/local/` — tcp_dial / udp_egress / bind_accept, the producer-side +`ProducerDoor`, ADR-008/009), the front doors (`src/door.rs` — the +NoAuth raw-pipe + RFC 1929 cred-bearing TCP door, the BIND door, the +UDP door, ADR-005/003/014), and the composed providers +(`src/composed/` — the direct-op egress + the bind-first accept over +alktunnels 0.2.0, ADR-007/008). Both cross-cutting integration suites +pin the behavior end to end: `tests/core-integration.rs` (the +protocol core) and `tests/assembly-integration.rs` (the assembly +layer — the full `-D` story through the doors, the composed variant +set, mixed assembly, the assembly teardown matrix). ## Documents diff --git a/docs/architecture/consumer.md b/docs/architecture/consumer.md index f1d8372..881515d 100644 --- a/docs/architecture/consumer.md +++ b/docs/architecture/consumer.md @@ -1,6 +1,8 @@ --- status: draft -last_updated: 2026-09-24 (the BIND spent posture reconciled — the +last_updated: 2026-09-25 (the cred-bearing door's call-connection +contract reconciled — TokenClient's door-only posture, one command per +accepted connection; the BIND spent posture reconciled — the take-guarded reply#2 reader, review 002 C-1/D-1; the landed session signatures reconciled 2026-09-24; Phase 1 initial draft 2026-09-20) --- @@ -123,7 +125,17 @@ BIND door): crosses the channel as a SOCKS password; the channel identity never reaches the vanilla client. The door checks credentials itself (no forked `AuthMethod` impl needed — the door owns the - accept loop). + accept loop). The mapped identity rides the open-op call's + `auth_token` payload (CF-005 precedence (1)) via + `TokenClient::wrap` — the client's channel-0 `CallConnection` is + taken once per door (the door-only contract: a cred-bearing + door's client is a door client; other shapes — the UDP door's + association, channels-native sessions — ride their own + connections to the same producer registration, one registration, + many connections). The vanilla conversation is + **one command per accepted connection** (the door's RFC half + consumes the greeting + one command; the data plane rides to + EOF). Door variant selection (NoAuth vs cred-bearing; the UDP door's bind address) is assembly-layer configuration. diff --git a/src/composed/mod.rs b/src/composed/mod.rs index be23685..aa27b7a 100644 --- a/src/composed/mod.rs +++ b/src/composed/mod.rs @@ -194,6 +194,14 @@ pub struct ComposedEgress { /// The factory `Sender` alone does not close the queue (the /// egress holds it) — the count is the design. live_relays: usize, + /// Whether any tunnel has ever been opened. The lazy table starts + /// empty: a fresh association's `next_reply` parks until the first + /// `send` opens a tunnel (the producer's stage selects the channel + /// branch first on a quiescent association — the association + /// serves until a datagram arrives or the channel EOFs). The + /// ended-relay `Err` fires only once at least one tunnel existed + /// and every relay has ended (the closed-queue condition). + ever_opened: bool, } /// The shared inbound queue's message: a tunnel's datagram (keyed by @@ -281,6 +289,7 @@ impl ComposedEgress { inbound: rx, inbound_tx: tx, live_relays: 0, + ever_opened: false, } } @@ -306,6 +315,7 @@ impl ComposedEgress { // bug). The relay task owns the bare read half. tokio::spawn(relay_read(taken.read, self.inbound_tx.clone(), key.clone())); self.live_relays += 1; + self.ever_opened = true; self.entries.insert( key.clone(), TableEntry { @@ -368,11 +378,14 @@ impl Egress for ComposedEgress { Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => break, } } - if self.live_relays == 0 { + if self.ever_opened && self.live_relays == 0 { // Every relay ended (the tunnels ended): a closed // relay set must NOT spin the stage's select (an // instantly-ready None would starve the channel - // branch) — end the association instead. + // branch) — end the association instead. A fresh + // association (no tunnel ever opened) parks here — + // the channel branch governs until the first + // datagram arrives. return Err(EgressError::General("all tunnel relays ended".into())); } match self.inbound.recv().await { diff --git a/src/lib.rs b/src/lib.rs index 588a3bb..7940969 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -40,21 +40,30 @@ //! sessions — [`Socks5ChannelSession`] (CONNECT), //! [`AssociateSession`] (the framed datagram stage), //! [`BindSession`] (the reply#1/reply#2 reader) — each owning its -//! channel's teardown; no `Clone`. +//! channel's teardown; no `Clone`. The cred-bearing front door's +//! re-origination rides the token-carrying surface (`TokenClient` +//! + the `*_with_token` constructor twins). //! -//! `local` (the socket-backed assembly-layer providers — -//! [`crate::local::tcp_dial`] / [`crate::local::udp_egress`] / -//! [`crate::local::bind_accept`], ADR-008/009) is feature-gated: the -//! default crate is protocol-only and wasm-clean by construction, and -//! the composed providers ride `composed` -//! ([`crate::composed::direct_egress`] — the alktunnels direct-op -//! egress, ADR-008/009). +//! The assembly layer (feature-gated, ADR-008/009): `local` (the +//! socket-backed providers — `local::tcp_dial` / `local::udp_egress` +//! / `local::bind_accept` — plus the producer-side +//! `local::ProducerDoor`) and `door` (the consumer's front doors — +//! the NoAuth raw-pipe + RFC 1929 cred-bearing `FrontDoor`, the +//! `UdpFrontDoor` with the RFC codec at the door boundary only, +//! ADR-005/003/014) ride `local`; the composed providers +//! (`composed::direct_egress` — the alktunnels direct-op egress, +//! `composed::bindfirst_accept` — the bind-first accept, +//! ADR-007/008) ride `composed`. The default crate is protocol-only +//! and wasm-clean by construction. //! //! Architecture: `docs/architecture/overview.md` (spec set: //! wire.md, producer.md, consumer.md, rfc1928.md); the cross-cutting -//! behavior is pinned end to end by `tests/core-integration.rs` -//! (one registration, all three commands; concurrency; the teardown -//! matrix; the identity/scope matrix; the readiness contract). +//! behavior is pinned end to end by `tests/core-integration.rs` (the +//! protocol core: one registration, all three commands; concurrency; +//! the teardown matrix; the identity/scope matrix; the readiness +//! contract) and `tests/assembly-integration.rs` (the assembly +//! layer: the full `-D` story through the doors, the composed +//! variant set, mixed assembly, the assembly teardown matrix). #![forbid(unsafe_code)] diff --git a/tasks/implementation/assembly-integration-suite.md b/tasks/implementation/assembly-integration-suite.md index 239c9ab..d352fe5 100644 --- a/tasks/implementation/assembly-integration-suite.md +++ b/tasks/implementation/assembly-integration-suite.md @@ -1,7 +1,7 @@ --- id: implementation/assembly-integration-suite name: Assembly — the cross-cutting integration suite + docs sync -status: pending +status: completed depends_on: [implementation/door-udp, implementation/local-backend-producer, implementation/composed-accept] scope: moderate risk: medium @@ -61,17 +61,17 @@ Deliverables: ## Acceptance Criteria -- [ ] The full `-D` story (vanilla client → door → session → +- [x] The full `-D` story (vanilla client → door → session → producer → real target) pinned end to end for all three commands -- [ ] The composed variant set pinned end to end (all three commands) -- [ ] Mixed assembly (doors + channels-native + composed) concurrent +- [x] The composed variant set pinned end to end (all three commands) +- [x] Mixed assembly (doors + channels-native + composed) concurrent on one producer — pinned -- [ ] The assembly teardown matrix pinned (door disconnects, idle +- [x] The assembly teardown matrix pinned (door disconnects, idle bounds, fresh-session-after-teardown) -- [ ] All four verification targets green (default, all-features, +- [x] All four verification targets green (default, all-features, wasm default, wasm composed) -- [ ] Docs synced (README Current State, lib.rs crate docs, any +- [x] Docs synced (README Current State, lib.rs crate docs, any drift reconciled) ## References @@ -86,8 +86,118 @@ Deliverables: ## Notes -> To be filled by implementation agent +> Filled by the implementation agent (2026-09-25). + +- **The spec drift the implementation surfaced (one real drift, + reconciled in `src/composed/mod.rs`):** `ComposedEgress::next_reply` + returned the association-end `Err` ("all tunnel relays ended") on a + *fresh* association too — the lazy table starts with `live_relays == + 0`, so the producer's datagram stage (which parks on `next_reply` + when no datagram has arrived) ended every quiescent association at + birth. The composed-egress suite drove the egress send-first and + never hit it; the assembly suite's `AssociateSession` (open → the + stage parks → the first datagram) did. Fix: an `ever_opened` flag — + the ended-relay `Err` fires only once at least one tunnel existed + and every relay has ended; a never-opened table parks (the readiness + contract's `Pending`, the channel branch governs). The landed spec + wording (producer.md: "`Pending` when nothing is ready") already + stated the correct contract — the implementation was the divergent + side. Regression-pinned in `tests/composed-egress.rs` + (`fresh_association_parks_before_any_tunnel_is_opened`). +- **Two assembly-shape findings recorded in the tests (not code + changes — the landed contracts are correct):** + - The cred-bearing door's `TokenClient::wrap` takes the client's + channel-0 `CallConnection` once (the door-only contract, + door-tcp's Notes) — the UDP door's association (and any other + consumption shape) rides its own channels connection to the same + producer registration. The suite pins the assembly story as "one + registration, many connections" (consumer.md reconciled). + - The RFC conversation is **one command per connection** at every + door shape (the cred-bearing TCP door's RFC half consumes the + greeting + one command; the producer-side `ProducerDoor`'s + `serve_socks5_over` likewise serves one command per accepted + socket — a second command's bytes ride the first session's data + plane). The vanilla one-command-per-connection shape is pinned by + the suite's structure (each command opens its own door + connection). +- **Harness consolidation (review 002 N-5's discipline):** the + composed-accept machinery the assembly suite also consumes (the + test bind, the accept queue + loop, the channel-set probe, the + identities) moved from `tests/composed-accept.rs` into + `tests/harness.rs` (single-sourced; `composed-accept.rs` consumes + the same shapes via the harness re-exports — its test bodies are + unchanged in behavior). The five-gate composed identity + (`composed_full_opener_identity` — `tunnel:open` + `tunnel:direct` + + `tunnel:forwarded` + `socks5:open` + `socks5:bind`) is new. +- **The wasm gate, observed**: `cargo check --features local --target + wasm32-unknown-unknown` fails as designed (the alktty posture) — + the first error is `error: This wasm target is unsupported by mio. + If using Tokio, disable the net feature.` (mio v1.2.3, via + `tokio/net` which `local` enables), followed by the mio `sys` + module's unresolved imports (`crate::sys::IoSourceState`, + `crate::sys::tcp`, `Selector`, `event`). Both default and + `--features composed` wasm checks are green. +- **The `-D` story's door wiring**: the cred-bearing TCP door + the + UDP door each take their own `ChannelClient` (two connections to + the one registration); the tests' `DoorTokenProvider` resolves the + mapped `door-`-prefixed identity ids (the same producer-side + resolution policy shape the door-tcp suite pins). The producer-door + test drives the producer's own front door (`ProducerDoor` → + `serve_socks5_over` over the `local` provider set) through all + three commands. ## Summary -> To be filled on completion \ No newline at end of file +> Filled on completion (2026-09-25). + +Landed the batch-2 closing task: `tests/assembly-integration.rs` — +seven cross-cutting integration tests pinning the whole assembly layer +end to end (feature `local` + `composed`): + +1. **The full `-D` story** (`full_dash_d_story_...`): a vanilla + tun2proxy-shaped client through the cred-bearing RFC 1929 TCP door + + the UDP door against one producer registration with the `local` + providers and real echo targets — CONNECT + BIND (the app server + dials the advertised address; reply#2 carries the real peer) + + ASSOCIATE (the RFC-header'd datagram through the relay socket) + live simultaneously, each exercised mid-life of the others, the + door-client disconnect ending only its own sessions, fresh + sessions succeeding after (no connection poisoning). +2. **The composed story** (`composed_story_...`): the producer with + `composed::direct_egress` + `composed::bindfirst_accept` + a real + TCP dial — CONNECT, ASSOCIATE (the lazy table; the direct-op dial + count witnesses the lazy open), BIND (the real `bound` + the real + `peer` + the pass-through plane) — all three commands, all live. +3. **Mixed assembly, twice**: channels-native sessions + door clients + concurrent on one registration (the doors on their own + connections), and composed seams beside channels-native sessions + (the composed suite's producer; teardowns end only their own + sessions; the channel-set probe after the settle). +4. **The teardown matrix, assembly edition**: the UDP door's idle + bound fires on the vanished door client, the association's channel + reaps (the channel count witnesses), a fresh association succeeds + immediately (no connection poisoning). +5. **The producer's own front door** (`ProducerDoor`) drives all + three commands through one door + one registration (the `-D` + story's other half; the faithful-RFC BIND variant). +6. **The `-D` domain-target pin**: a domain-form CONNECT through the + NoAuth raw-pipe door resolves at the producer's seam (ADR-012). + +The reconciled drift: `ComposedEgress`'s never-opened-table association +end (fixed in `src/composed/mod.rs` — the `ever_opened` guard; +regression-pinned in `tests/composed-egress.rs`). Docs synced: +`README.md` Current State (the assembly layer landed; both suites +named), `lib.rs` crate docs (the assembly-layer paragraph + both +suites named), `consumer.md` §Door variants (the `TokenClient` +door-only contract + one-command-per-connection posture reconciled — +the drift the suite surfaced, recorded in Notes). Harness: the +composed-accept machinery single-sourced into `tests/harness.rs` (the +N-5 discipline). + +Verification: `cargo test` (375 green across 12 targets), `cargo test +--all-features` (615 green), `cargo clippy --all-targets -- -D +warnings` + `cargo clippy --all-targets --all-features -- -D warnings` +clean, `cargo fmt --check` clean, `cargo doc --no-deps` clean (0 +warnings), both wasm checks green (default + `--features composed`, +check + clippy), and the `local`-on-wasm build failure observed + +recorded (the alktty posture — mio's unsupported-wasm error). \ No newline at end of file diff --git a/tests/assembly-integration.rs b/tests/assembly-integration.rs new file mode 100644 index 0000000..1e573b7 --- /dev/null +++ b/tests/assembly-integration.rs @@ -0,0 +1,1501 @@ +//! The cross-cutting assembly integration suite — the whole assembly +//! layer's behavior pinned in one place (the batch-2 closing task; the +//! architecture-sync check's test half). +//! +//! The per-component tasks own their verticals (tests/door-*.rs, +//! tests/local-*.rs, tests/composed-*.rs); this suite owns the +//! compositions and the cross-shape invariants: +//! +//! - **The full `-D` story end to end**: a vanilla tun2proxy-shaped +//! client → the cred-bearing TCP door → the channels-native session +//! → the producer with `local` providers → real echo targets — +//! CONNECT + UDP (the RFC-header'd datagram through the UDP door) + +//! BIND (the vanilla BIND conversation through the door) all through +//! one door process against one producer registration (the +//! one-registration-three-commands regression, now through the +//! doors). +//! - **The composed story end to end**: the producer with +//! `composed::direct_egress` + `composed::bindfirst_accept` — +//! CONNECT through the direct op, ASSOCIATE through the lazy table, +//! BIND through the bind-first listener (the ADR-008 composed +//! variant set, all three commands). +//! - **Mixed assembly**: one producer registration serving +//! channels-native sessions AND door clients AND composed seams +//! concurrently (the producer/consumer-roles-not-sides posture — +//! both shapes on one connection). +//! - **The teardown matrix, assembly edition**: door-client disconnect +//! tears the association/channel down (no leaks — the channel count +//! witnesses); the idle bounds fire; fresh sessions succeed +//! immediately after each teardown (no connection poisoning). +#![cfg(all(feature = "local", feature = "composed"))] + +mod harness; + +fn init_trace() { + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::level_filters::LevelFilter::DEBUG) + .with_test_writer() + .try_init(); +} + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use alkcall::channels::operations::ChannelCore; +use alkcall::core::auth::{Identity, IdentityProvider}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream, UdpSocket}; +use tokio::time::timeout; + +use alksocks::composed::{bindfirst_accept, direct_egress, register_forwarded_serving}; +use alksocks::consumer::{AssociateSession, BindSession, Socks5ChannelSession}; +use alksocks::rfc1928::consts; +use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; +use alksocks::{CredentialTable, DoorCredential, FrontDoor, UdpFrontDoor, SOCKS5_BIND_SCOPE}; + +use harness::{ + composed_full_opener_identity, composed_push_accept, wire_bindfirst_composed, + wire_consumer_side_with, wire_consumer_side_with_provider, COMPOSED_BOUND_PORT, + COMPOSED_PEER_PORT, COMPOSED_RESOURCE, +}; + +const BOUND: Duration = Duration::from_secs(10); + +// --- Real targets ------------------------------------------------------------ + +/// A real TCP echo server on an ephemeral loopback port. +async fn echo_tcp_server() -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + tokio::spawn(async move { + let mut buf = [0u8; 4096]; + loop { + match stream.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => { + if stream.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + } + }); + } + }); + addr +} + +/// A real UDP echo server on an ephemeral loopback port. +async fn echo_udp_server() -> SocketAddr { + let sock = UdpSocket::bind(("127.0.0.1", 0)) + .await + .expect("bind udp echo"); + let addr = sock.local_addr().expect("addr"); + tokio::spawn(async move { + let mut buf = vec![0u8; 65535]; + loop { + let Ok((n, peer)) = sock.recv_from(&mut buf).await else { + return; + }; + if sock.send_to(&buf[..n], peer).await.is_err() { + return; + } + } + }); + addr +} + +// --- Vanilla-client helpers (the door boundary's RFC forms) ------------------ + +/// The RFC 1928 cred-bearing conversation's CONNECT: greeting (the +/// password method offered) → method selection → credentials → the +/// CONNECT command for a v4 target → the reply → (code, stream, BND). +/// The vanilla tun2proxy shape (`--proxy socks5://user:pass@host:port`). +async fn vanilla_cred_connect(door: SocketAddr, target: SocketAddr) -> (TcpStream, u8, [u8; 6]) { + let mut stream = vanilla_with_creds(door).await; + let mut req = vec![0x05, 0x01, 0x00, 0x01]; + let SocketAddr::V4(v4) = target else { + panic!("v4 targets only") + }; + req.extend_from_slice(&v4.ip().octets()); + req.extend_from_slice(&v4.port().to_be_bytes()); + stream.write_all(&req).await.expect("request"); + stream.flush().await.expect("flush"); + + let mut head = [0u8; 4]; + stream.read_exact(&mut head).await.expect("reply head"); + assert_eq!(head[0], 0x05, "reply is v5"); + assert_eq!(head[3], 0x01, "v4 BND.ADDR"); + let mut addr = [0u8; 6]; + stream.read_exact(&mut addr).await.expect("BND.ADDR"); + (stream, head[1], addr) +} + +/// The RFC 1928 NoAuth conversation against a TCP door: greet, send a +/// CONNECT command for a v4 target, read the reply → (code, stream). +async fn vanilla_no_auth_connect(door: SocketAddr, target: SocketAddr) -> (TcpStream, u8, [u8; 6]) { + let mut stream = TcpStream::connect(door).await.expect("door connect"); + stream + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting"); + let mut chosen = [0u8; 2]; + stream + .read_exact(&mut chosen) + .await + .expect("method selection"); + assert_eq!(chosen, [0x05, 0x00], "no-auth selected"); + + let mut req = vec![0x05, 0x01, 0x00, 0x01]; + let SocketAddr::V4(v4) = target else { + panic!("v4 targets only") + }; + req.extend_from_slice(&v4.ip().octets()); + req.extend_from_slice(&v4.port().to_be_bytes()); + stream.write_all(&req).await.expect("request"); + stream.flush().await.expect("flush"); + + let mut head = [0u8; 4]; + stream.read_exact(&mut head).await.expect("reply head"); + assert_eq!(head[0], 0x05, "reply is v5"); + assert_eq!(head[3], 0x01, "v4 BND.ADDR"); + let mut addr = [0u8; 6]; + stream.read_exact(&mut addr).await.expect("BND.ADDR"); + (stream, head[1], addr) +} + +/// An RFC-header'd datagram (the vanilla client's sendto form — the +/// door boundary's codec, hand-rolled per the RFC wire form). +fn rfc_datagram(target: SocketAddr, payload: &[u8]) -> Vec { + let SocketAddr::V4(v4) = target else { + panic!("v4 targets only") + }; + let mut dg = vec![0u8, 0, 0, 0x01]; + dg.extend_from_slice(&v4.ip().octets()); + dg.extend_from_slice(&v4.port().to_be_bytes()); + dg.extend_from_slice(payload); + dg +} + +/// Parse an RFC-header'd datagram back → (addr, payload). +fn parse_rfc_datagram(wire: &[u8]) -> (SocketAddr, &[u8]) { + assert_eq!(&wire[..2], &[0, 0], "RSV zero"); + assert_eq!(wire[3], 0x01, "v4 ATYP"); + let ip = std::net::Ipv4Addr::new(wire[4], wire[5], wire[6], wire[7]); + let port = u16::from_be_bytes([wire[8], wire[9]]); + (SocketAddr::from((ip, port)), &wire[10..]) +} + +// --- The token provider (the cred-bearing door's CF-005 seam) ---------------- + +/// Resolves the door's mapped identity ids (`door-`-prefixed) to +/// identities carrying the requested scopes; anything else resolves to +/// none (the CF-005 precedence fall-through). +#[derive(Clone)] +struct DoorTokenProvider { + bind_scoped: Vec, +} + +impl IdentityProvider for DoorTokenProvider { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option { + let id = String::from_utf8_lossy(&token.raw).into_owned(); + if id.starts_with("door-") { + let mut scopes = vec![alksocks::SOCKS5_OPEN_SCOPE.to_string()]; + if self.bind_scoped.contains(&id) { + scopes.push(SOCKS5_BIND_SCOPE.to_string()); + } + return Some(Identity { + id, + scopes, + resources: Default::default(), + }); + } + None + } +} + +fn cred_table() -> CredentialTable { + CredentialTable::new(vec![DoorCredential { + username: "vanilla".to_string(), + password: "hunter2".to_string(), + identity: Identity { + id: "door-vanilla".to_string(), + scopes: vec![ + alksocks::SOCKS5_OPEN_SCOPE.to_string(), + SOCKS5_BIND_SCOPE.to_string(), + ], + resources: Default::default(), + }, + }]) +} + +/// The RFC 1929 cred-bearing conversation: greeting (password method +/// offered) → method selection → credentials → command. Returns the +/// post-auth stream. +async fn vanilla_with_creds(door: SocketAddr) -> TcpStream { + let mut stream = TcpStream::connect(door).await.expect("door connect"); + stream + .write_all(&[0x05, 0x01, 0x02]) + .await + .expect("greeting"); + let mut chosen = [0u8; 2]; + stream + .read_exact(&mut chosen) + .await + .expect("method selected"); + assert_eq!(chosen, [0x05, 0x02], "password auth selected"); + let mut packet = vec![0x01, 7]; + packet.extend_from_slice(b"vanilla"); + packet.push(7); + packet.extend_from_slice(b"hunter2"); + stream.write_all(&packet).await.expect("creds sent"); + let mut status = [0u8; 2]; + stream.read_exact(&mut status).await.expect("status"); + assert_eq!(status, [0x01, 0x00], "RFC 1929 accepted"); + stream +} + +// --- The composed suite (the direct-op + bind-first producer) ---------------- + +/// The composed topology: the SOCKS5 producer serves the direct op +/// (CONNECT + ASSOCIATE's egress) AND the bind-first op (BIND's +/// listener) on one connection — the ADR-008 composed variant set. +/// The direct-op dial is a real TCP/UDP-capable stand-in: a TCP dial +/// dials the real echo target (CONNECT rides it); a UDP dial yields a +/// framed echo pipe (ASSOCIATE's destinations echo datagrams). +struct ComposedSuite { + topo: harness::BindFirstTopology, + queue: harness::ComposedAcceptQueue, + dial_count: Arc, +} + +async fn wire_composed_suite() -> ComposedSuite { + init_trace(); + let resource_registry = alktunnels::producer::ResourceRegistry::new(); + resource_registry + .register( + COMPOSED_RESOURCE, + alktunnels::params::Substrate::Tcp, + "assembly-owned", + ) + .await; + let queue = harness::composed_fresh_queue(); + let dial_count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + + // The direct-op dial: a TCP dial rides the real sockets via + // `local::tcp_dial`-shaped behavior re-expressed on the alktunnels + // `DialFn` (the direct op's TargetHandle); a UDP dial yields the + // framed echo pipe (the composed-egress suite's shape). + let dial = { + let dial_count = Arc::clone(&dial_count); + Arc::new( + move |substrate: alktunnels::params::Substrate, + backing: &str| + -> futures::future::BoxFuture< + 'static, + Result, + > { + let dial_count = Arc::clone(&dial_count); + let backing = backing.to_string(); + Box::pin(async move { + use alktunnels::params::Substrate as S; + use alktunnels::producer::TargetHandle as TunnelTargetHandle; + use alktunnels::TunnelEstablishError; + dial_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + match substrate { + S::Tcp => { + // A real TCP dial: `host:port` → TcpStream → split. + let (host, port) = backing.rsplit_once(':').ok_or_else(|| { + TunnelEstablishError::DialFailed("bad target".into()) + })?; + let addr: SocketAddr = + format!("{host}:{port}").parse().map_err(|e| { + TunnelEstablishError::DialFailed(format!("parse: {e}")) + })?; + let stream = TcpStream::connect(addr).await.map_err(|e| { + TunnelEstablishError::DialFailed(format!("connect: {e}")) + })?; + let (r, w) = stream.into_split(); + Ok(TunnelTargetHandle { + read: Box::new(r), + write: Box::new(w), + }) + } + S::Udp => { + // The framed UDP echo pipe (the composed-egress + // suite's shape): the "target" decodes the + // tunnel codec and echoes each datagram. + let (consumer_side, target_side) = tokio::io::duplex(64 * 1024); + tokio::spawn(async move { + use tokio::io::AsyncReadExt; + let (mut t_read, mut t_write) = tokio::io::split(target_side); + let mut reader = alktunnels::wire::DatagramReader::new(); + let mut buf = vec![0u8; 16 * 1024]; + loop { + let n = match t_read.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => n, + }; + for dg in reader.feed(&buf[..n]) { + let framed = match alktunnels::wire::frame_datagram(&dg) { + Ok(f) => f, + Err(_) => return, + }; + if t_write.write_all(&framed).await.is_err() { + return; + } + let _ = t_write.flush().await; + } + } + }); + let (c_read, c_write) = tokio::io::split(consumer_side); + Ok(TunnelTargetHandle { + read: Box::new(c_read), + write: Box::new(c_write), + }) + } + other => Err(TunnelEstablishError::DialFailed(format!( + "substrate not wired: {other:?}" + ))), + } + }) + }, + ) as alktunnels::producer::DialFn + }; + + let topo = wire_bindfirst_composed( + resource_registry, + harness::composed_test_bind(), + harness::composed_queue_loop(queue.clone()), + None, // the SOCKS5 openable registers post-hoc, below + composed_full_opener_identity(), + ) + .await; + + // The forwarded serving half rides the opener's registry. + let registry = register_forwarded_serving( + &ChannelCore::new( + topo.opener.manager().clone(), + alkcall::channels::policy::default_policy(), + ), + &topo.opener_registry, + alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"), + ); + + // The SOCKS5 producer: the composed accept seam (BIND) + the + // composed direct-op egress (ASSOCIATE) + a real TCP dial + // (CONNECT — the direct-op dial's TCP arm shares the real target; + // the CONNECT dial here is the crate's own local dial over the + // SOCKS5 seam, distinct from the tunnel dial). + let factory = bindfirst_accept(topo.opener.clone(), registry, COMPOSED_RESOURCE); + let core = ChannelCore::new( + topo.producer_manager.clone(), + alkcall::channels::policy::default_policy(), + ); + alksocks::producer::Socks5Producer::builder() + .dial(alksocks::local::tcp_dial()) + .egress(direct_egress(topo.opener.clone(), 8)) + .accept(factory) + .build() + .register( + &core, + &topo.producer_registry, + alkcall::core::auth::AuthContext::anonymous(b"alk/socks5"), + None, + ) + .expect("register socks5 openable (the composed suite's producer)"); + + // The direct op rides the producer's registry (the direct-op + // caller is the opener's client; the dial is the tunnel dial + // above). + let direct_core = ChannelCore::new( + topo.producer_manager.clone(), + alkcall::channels::policy::default_policy(), + ); + alktunnels::producer::register_tunnel_direct_openable( + &direct_core, + &topo.producer_registry, + alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"), + dial, + None, + ) + .expect("register the tunnel direct op (the composed suite)"); + + ComposedSuite { + topo, + queue, + dial_count, + } +} + +// --- The suite ----------------------------------------------------------------- + +/// THE full `-D` story (the one-registration-three-commands regression +/// through the doors): one producer registration (the consumer-side +/// topology over the `local` providers) + one cred-bearing TCP door + +/// one UDP door. A vanilla client drives all three commands through +/// the doors against the one registration: +/// +/// - CONNECT: the RFC 1929 door conversation → the CONNECT request → +/// the real TCP echo target round-trips. +/// - BIND: the door's BIND conversation (its own RFC session — the +/// vanilla client's one-command-per-connection shape) → the app +/// server dials the advertised address → reply#2's real peer → the +/// data plane. +/// - ASSOCIATE: the UDP door's relay socket ← the RFC-header'd +/// datagram → the real UDP echo target → back RFC-header'd. +/// +/// All three sessions are live simultaneously mid-life of each other; +/// each teardown ends only its own session. The two doors ride two +/// channels connections to the same producer registration (the +/// assembly story: one registration, many connections — the cred- +/// bearing door's `TokenClient` takes its client's call connection, +/// the door-only contract, so the UDP door's association rides its +/// own). +#[tokio::test] +async fn full_dash_d_story_one_registration_three_commands_through_the_doors() { + init_trace(); + let tcp_target = echo_tcp_server().await; + let udp_target = echo_udp_server().await; + + // One producer registration, two door connections: the cred- + // bearing TCP door rides one client (its TokenClient takes the + // call connection); the UDP door's association rides a second + // connection to the same registration (the UDP relay is a + // separate datagram plane by construction — one relay socket = one + // association). + let tcp_dial = alksocks::local::tcp_dial(); + let tcp_topo = wire_consumer_side_with_provider( + tcp_dial, + Some(alksocks::local::udp_egress()), + Some(alksocks::local::bind_accept([127, 0, 0, 1])), + Some(harness::consumer_identity()), + Arc::new(DoorTokenProvider { + bind_scoped: vec!["door-vanilla".to_string()], + }), + ) + .await; + let udp_dial = alksocks::local::tcp_dial(); + let udp_topo = wire_consumer_side_with_provider( + udp_dial, + Some(alksocks::local::udp_egress()), + Some(alksocks::local::bind_accept([127, 0, 0, 1])), + Some(harness::consumer_identity()), + Arc::new(DoorTokenProvider { + bind_scoped: vec!["door-vanilla".to_string()], + }), + ) + .await; + + // The cred-bearing TCP door (the CONNECT + BIND shape) + the UDP + // door (the datagram plane) — one door process per plane, the + // assembly wiring consumer.md documents. + let tcp_listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let door_addr = tcp_listener.local_addr().expect("door addr"); + let door = FrontDoor::builder() + .credentials(cred_table()) + .bind(door_addr) + .build(); + let _tcp_serve = tokio::spawn({ + let client = Arc::clone(&tcp_topo.consumer); + async move { + door.serve_on(tcp_listener, client) + .await + .expect("the door serves") + } + }); + let udp_relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); + let udp_relay_addr = udp_relay.local_addr().expect("relay addr"); + let udp_door = UdpFrontDoor::builder().bind(udp_relay_addr).build(); + let _udp_serve = tokio::spawn({ + let client = Arc::clone(&udp_topo.consumer); + async move { + udp_door + .serve_on(udp_relay, client) + .await + .expect("the udp door serves") + } + }); + + // --- CONNECT through the cred-bearing door (live) --------------- + let mut connect_stream = timeout(BOUND, vanilla_with_creds(door_addr)) + .await + .expect("the cred conversation"); + let mut req = vec![0x05, 0x01, 0x00, 0x01]; + let SocketAddr::V4(v4) = tcp_target else { + panic!("v4 target") + }; + req.extend_from_slice(&v4.ip().octets()); + req.extend_from_slice(&v4.port().to_be_bytes()); + connect_stream + .write_all(&req) + .await + .expect("connect request"); + let mut head = [0u8; 4]; + timeout(BOUND, connect_stream.read_exact(&mut head)) + .await + .expect("reply#1") + .expect("read"); + assert_eq!( + head[1], + consts::SOCKS5_REPLY_SUCCEEDED, + "CONNECT through the door" + ); + let mut connect_bnd = [0u8; 6]; + timeout(BOUND, connect_stream.read_exact(&mut connect_bnd)) + .await + .expect("the CONNECT reply's BND.ADDR") + .expect("read"); + + // --- BIND through the door (live, mid-life of the CONNECT) ------ + // Its own door connection (the vanilla one-command-per-connection + // shape — the door's RFC half reads one command per conversation). + let mut bind_stream = timeout(BOUND, vanilla_with_creds(door_addr)) + .await + .expect("the second cred conversation"); + bind_stream + .write_all(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) + .await + .expect("bind request"); + let mut head1 = [0u8; 4]; + timeout(BOUND, bind_stream.read_exact(&mut head1)) + .await + .expect("bind reply#1") + .expect("read"); + assert_eq!( + head1[1], + consts::SOCKS5_REPLY_SUCCEEDED, + "BIND through the door" + ); + let mut bnd = [0u8; 6]; + timeout(BOUND, bind_stream.read_exact(&mut bnd)) + .await + .expect("reply#1 BND.ADDR") + .expect("read"); + let advertised = SocketAddr::from(( + [bnd[0], bnd[1], bnd[2], bnd[3]], + u16::from_be_bytes([bnd[4], bnd[5]]), + )); + assert!( + advertised.port() != 0, + "reply#1 carries the real listener: {advertised}" + ); + let mut app = timeout(BOUND, TcpStream::connect(advertised)) + .await + .expect("dial the advertised address") + .expect("app connected"); + app.write_all(b"vanilla-bind").await.expect("app writes"); + { + let mut head2 = [0u8; 4]; + timeout(BOUND, bind_stream.read_exact(&mut head2)) + .await + .expect("bind reply#2 head") + .expect("read"); + assert_eq!(head2[1], consts::SOCKS5_REPLY_SUCCEEDED, "BIND reply#2"); + let mut peer_addr = [0u8; 6]; + timeout(BOUND, bind_stream.read_exact(&mut peer_addr)) + .await + .expect("reply#2 DST.ADDR") + .expect("read"); + let port = u16::from_be_bytes([peer_addr[4], peer_addr[5]]); + assert_eq!( + port, + app.local_addr().expect("app local").port(), + "reply#2 carries the real peer" + ); + } + + // --- ASSOCIATE through the UDP door (live, mid-life of both) ---- + let client_sock = UdpSocket::bind("127.0.0.1:0").await.expect("client socket"); + client_sock + .send_to(&rfc_datagram(udp_target, b"udp-door"), udp_relay_addr) + .await + .expect("vanilla sendto the relay"); + let mut rx = vec![0u8; 65535]; + let (n, from) = timeout(BOUND, client_sock.recv_from(&mut rx)) + .await + .expect("udp reply within bound") + .expect("recv"); + assert_eq!( + from, udp_relay_addr, + "the reply comes from the relay socket" + ); + let (src, payload) = parse_rfc_datagram(&rx[..n]); + assert_eq!(src, udp_target, "the echo reports the target"); + assert_eq!(payload, b"udp-door"); + + // --- The data planes, all three live: exercise each mid-life ---- + connect_stream + .write_all(b"tcp-live") + .await + .expect("tcp write"); + let mut buf = [0u8; 8]; + timeout(BOUND, connect_stream.read_exact(&mut buf)) + .await + .expect("tcp echo") + .expect("read"); + assert_eq!(&buf, b"tcp-live"); + + let mut app_buf = [0u8; 12]; + timeout(BOUND, bind_stream.read_exact(&mut app_buf)) + .await + .expect("bind data plane") + .expect("read"); + assert_eq!(&app_buf, b"vanilla-bind"); + + client_sock + .send_to(&rfc_datagram(udp_target, b"udp-live"), udp_relay_addr) + .await + .expect("send"); + let (n, _) = timeout(BOUND, client_sock.recv_from(&mut rx)) + .await + .expect("udp echo") + .expect("recv"); + assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"udp-live"); + + // --- Teardown: the door-client disconnect ends only its own ----- + // The CONNECT + BIND connections drop (the vanilla client's + // disconnect); the association on the UDP door is a separate + // session. + drop(connect_stream); + drop(bind_stream); + drop(app); + + // The UDP door's association survives the door-client disconnect + // (mux/demux isolation): another datagram round-trips. + client_sock + .send_to(&rfc_datagram(udp_target, b"still"), udp_relay_addr) + .await + .expect("send after the tcp teardown"); + let (n, _) = timeout(BOUND, client_sock.recv_from(&mut rx)) + .await + .expect("post-teardown echo") + .expect("recv"); + assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"still"); + + // Fresh sessions of every command succeed immediately after (no + // connection poisoning). + let (mut fresh, code, _) = timeout(BOUND, vanilla_cred_connect(door_addr, tcp_target)) + .await + .expect("fresh CONNECT through the door"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + fresh.write_all(b"again").await.expect("write"); + let mut buf = [0u8; 5]; + timeout(BOUND, fresh.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"again"); + + client_sock + .send_to(&rfc_datagram(udp_target, b"fresh"), udp_relay_addr) + .await + .expect("send fresh"); + let (n, _) = timeout(BOUND, client_sock.recv_from(&mut rx)) + .await + .expect("fresh udp echo") + .expect("recv"); + assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"fresh"); +} + +/// The teardown matrix, door edition (the no-leak witness): the door +/// client's disconnect tears the association's channel down (the +/// channel count witnesses); a fresh association succeeds immediately +/// after (no connection poisoning). +#[tokio::test] +async fn door_client_disconnect_tears_the_association_down_no_leaks() { + init_trace(); + let udp_target = echo_udp_server().await; + let topo = wire_consumer_side_with( + alksocks::local::tcp_dial(), + Some(alksocks::local::udp_egress()), + None, + Some(harness::consumer_identity()), + ) + .await; + + let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); + let relay_addr = relay.local_addr().expect("relay addr"); + let door = UdpFrontDoor::builder().bind(relay_addr).build(); + let serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { door.serve_on(relay, client).await.expect("the door serves") } + }); + + // The association is live: a datagram round-trips and the channel + // exists. + let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket"); + client + .send_to(&rfc_datagram(udp_target, b"pre"), relay_addr) + .await + .expect("send"); + let mut rx = vec![0u8; 65535]; + let (n, _) = timeout(BOUND, client.recv_from(&mut rx)) + .await + .expect("reply") + .expect("recv"); + assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"pre"); + assert!( + topo.consumer + .manager() + .channel_ids() + .iter() + .any(|&id| id != 0), + "the association rode a channel" + ); + + // The door client vanishes: the idle bound (the constructor knob) + // is the door's teardown trigger for a silent relay — with the + // socket gone, no datagrams arrive, the bound fires, the + // association ends (association lifetime = channel lifetime, + // ADR-003), and the door's task returns. + drop(client); + timeout(Duration::from_secs(35), serve) + .await + .expect("the door's task ended on the idle bound") + .expect("clean exit"); + + // No leak: the association's channel reaped. + tokio::time::sleep(Duration::from_millis(200)).await; + assert!( + topo.consumer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0), + "the teardown reaped the association's channel: {:?}", + topo.consumer.manager().channel_ids() + ); + + // Fresh association succeeds immediately (no connection + // poisoning). + let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2"); + let relay2_addr = relay2.local_addr().expect("relay addr 2"); + let door2 = UdpFrontDoor::builder().bind(relay2_addr).build(); + let _serve2 = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door2 + .serve_on(relay2, client) + .await + .expect("the fresh door serves") + } + }); + let client2 = UdpSocket::bind("127.0.0.1:0").await.expect("client 2"); + client2 + .send_to(&rfc_datagram(udp_target, b"fresh"), relay2_addr) + .await + .expect("send"); + let (n, _) = timeout(BOUND, client2.recv_from(&mut rx)) + .await + .expect("fresh reply") + .expect("recv"); + assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"fresh"); +} + +/// The composed story end to end (the ADR-008 composed variant set, +/// all three commands): the SOCKS5 producer serves +/// `composed::direct_egress` (ASSOCIATE — the lazy destination→tunnel +/// table) + `composed::bindfirst_accept` (BIND — the bind-first +/// listener) + a real TCP dial (CONNECT). The channels-native +/// sessions drive all three against the one registration; the +/// direct-op dial count witnesses the table's lazy shape. +#[tokio::test] +async fn composed_story_all_three_commands_on_one_registration() { + let suite = wire_composed_suite().await; + + // CONNECT through the dial seam (a real TCP echo target — + // dialed by the producer's local dial). + let tcp_target = echo_tcp_server().await; + let mut session = timeout( + BOUND, + Socks5ChannelSession::connect(suite.topo.opener.as_ref(), TargetAddr::Ip(tcp_target)), + ) + .await + .expect("connect within bound") + .expect("the CONNECT session"); + let (mut read, mut write) = session.split().expect("pipe"); + write.write_all(b"composed-connect").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = [0u8; 16]; + timeout(BOUND, read.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"composed-connect"); + + // ASSOCIATE through the composed egress (the direct op): one + // destination → one tunnel (the dial count witnesses the lazy + // open), the framed datagram round-trips through it. + let mut assoc = timeout( + BOUND, + AssociateSession::associate(suite.topo.opener.as_ref()), + ) + .await + .expect("associate within bound") + .expect("the association"); + let dest = TargetAddr::Ip(SocketAddr::from(([10, 9, 9, 9], 5353))); + assoc + .send_datagram(&dest, b"composed-udp") + .await + .expect("send"); + let (src, payload) = timeout(BOUND, assoc.recv_datagram()) + .await + .expect("recv within bound") + .expect("a datagram") + .expect("a datagram (not eof)"); + assert_eq!(src, dest, "the echo reports the destination"); + assert_eq!(&payload[..], b"composed-udp"); + assert_eq!( + suite.dial_count.load(std::sync::atomic::Ordering::SeqCst), + 1, + "one direct-op dial for the one destination" + ); + + // BIND through the composed accept (the bind-first listener): + // reply#1 = the real bound; the queue accept; reply#2 = the real + // peer; the pass-through data plane. + let mut bind = timeout( + BOUND, + BindSession::bind( + suite.topo.opener.as_ref(), + ("bind.test", 2121).to_target_addr().unwrap(), + ), + ) + .await + .expect("bind within bound") + .expect("the BIND session"); + let bind_addr = bind.reply1().expect("reply#1").bind_addr; + assert_eq!( + bind_addr, + TargetAddr::Ip(SocketAddr::from(([127, 0, 0, 1], COMPOSED_BOUND_PORT))), + "reply#1 advertises the real OS-chosen bound" + ); + composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await; + let mut plane = timeout(BOUND, bind.await_inbound()) + .await + .expect("reply#2") + .expect("await_inbound ok") + .expect("the plane"); + assert_eq!( + plane.peer, + TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], COMPOSED_PEER_PORT))), + "reply#2 carries the forwarded op's real peer" + ); + plane.write_all(b"composed-bind").await.expect("write"); + plane.flush().await.expect("flush"); + let mut bbuf = [0u8; 13]; + timeout(BOUND, plane.read_exact(&mut bbuf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&bbuf, b"composed-bind"); + + // All three sessions were live simultaneously; the teardowns end + // only their own sessions (mux/demux isolation) — exercised by + // the interleaving above. Drop + close. + drop(read); + drop(write); + drop(assoc); + drop(plane); + assert!(session.close().await, "the CONNECT channel reaped"); + assert!(bind.close().await, "the BIND channel reaped"); +} + +/// Mixed assembly (the producer/consumer-roles-not-sides posture): one +/// producer registration serving channels-native sessions AND door +/// clients AND composed seams concurrently. The composed suite's +/// producer gains a door-shaped second consumer topology over the same +/// registration — the channels-native sessions and the door clients +/// interleave on the one registration (both shapes, one connection). +#[tokio::test] +async fn mixed_assembly_channels_native_and_door_clients_concurrent() { + init_trace(); + let tcp_target = echo_tcp_server().await; + let udp_target = echo_udp_server().await; + + let topo = wire_consumer_side_with_provider( + alksocks::local::tcp_dial(), + Some(alksocks::local::udp_egress()), + Some(alksocks::local::bind_accept([127, 0, 0, 1])), + Some(harness::consumer_identity()), + Arc::new(DoorTokenProvider { + bind_scoped: vec!["door-mixed".to_string()], + }), + ) + .await; + let door_topo = wire_consumer_side_with_provider( + alksocks::local::tcp_dial(), + Some(alksocks::local::udp_egress()), + Some(alksocks::local::bind_accept([127, 0, 0, 1])), + Some(harness::consumer_identity()), + Arc::new(DoorTokenProvider { + bind_scoped: vec!["door-mixed".to_string()], + }), + ) + .await; + let udp_door_topo = wire_consumer_side_with_provider( + alksocks::local::tcp_dial(), + Some(alksocks::local::udp_egress()), + Some(alksocks::local::bind_accept([127, 0, 0, 1])), + Some(harness::consumer_identity()), + Arc::new(DoorTokenProvider { + bind_scoped: vec!["door-mixed".to_string()], + }), + ) + .await; + + // The door fronts (the vanilla shape) on their own connections to + // the same registration (the cred-bearing door's TokenClient takes + // its client's call connection — the door-only contract; the UDP + // door's association rides a second connection). + let tcp_listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let door_addr = tcp_listener.local_addr().expect("door addr"); + let door = FrontDoor::builder() + .credentials(cred_table()) + .bind(door_addr) + .build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&door_topo.consumer); + async move { + door.serve_on(tcp_listener, client) + .await + .expect("the door serves") + } + }); + let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); + let relay_addr = relay.local_addr().expect("relay addr"); + let udp_door = UdpFrontDoor::builder().bind(relay_addr).build(); + let _udp_serve = tokio::spawn({ + let client = Arc::clone(&udp_door_topo.consumer); + async move { + udp_door + .serve_on(relay, client) + .await + .expect("the udp door serves") + } + }); + + // The three shapes, concurrent on the one registration: + // + // 1. A channels-native CONNECT + ASSOCIATE + BIND session trio + // (the wrapper-aware shape). + // 2. A door-client CONNECT round-trip (the vanilla RFC 1929 + // shape through the cred-bearing door). + // 3. A door-client ASSOCIATE datagram (the vanilla RFC-header'd + // shape through the UDP door). + // + // All six run interleaved (join_all), then each is verified + // mid-life of the others. + + let native = async { + // CONNECT native. + let mut s = + Socks5ChannelSession::connect(topo.consumer.as_ref(), TargetAddr::Ip(tcp_target)) + .await + .map_err(|e| format!("native connect: {e}"))?; + let (mut read, mut write) = s.split().ok_or_else(|| "no pipe".to_string())?; + write + .write_all(b"native-tcp") + .await + .map_err(|e| e.to_string())?; + write.flush().await.map_err(|e| e.to_string())?; + let mut buf = [0u8; 10]; + read.read_exact(&mut buf).await.map_err(|e| e.to_string())?; + if &buf != b"native-tcp" { + return Err("native connect echo mismatch".to_string()); + } + drop(read); + drop(write); + s.close().await; + Ok(()) + }; + + let native_udp = async { + let mut a = AssociateSession::associate(topo.consumer.as_ref()) + .await + .map_err(|e| format!("native associate: {e}"))?; + let target = TargetAddr::Ip(udp_target); + a.send_datagram(&target, b"native-udp") + .await + .map_err(|e| e.to_string())?; + let (src, payload) = a + .recv_datagram() + .await + .map_err(|e| e.to_string())? + .ok_or_else(|| "eof before a datagram".to_string())?; + if src != target || &payload[..] != b"native-udp" { + return Err("native associate echo mismatch".to_string()); + } + a.close().await; + Ok(()) + }; + + let door_tcp = async { + let mut stream = timeout(BOUND, vanilla_with_creds(door_addr)) + .await + .map_err(|e| format!("door connect: {e}"))?; + let mut req = vec![0x05, 0x01, 0x00, 0x01]; + let SocketAddr::V4(v4) = tcp_target else { + panic!("v4") + }; + req.extend_from_slice(&v4.ip().octets()); + req.extend_from_slice(&v4.port().to_be_bytes()); + stream.write_all(&req).await.map_err(|e| e.to_string())?; + let mut head = [0u8; 4]; + timeout(BOUND, stream.read_exact(&mut head)) + .await + .map_err(|e| e.to_string())? + .map_err(|e| e.to_string())?; + if head[1] != consts::SOCKS5_REPLY_SUCCEEDED { + return Err(format!("door connect reply {}", head[1])); + } + let mut bnd = [0u8; 6]; + timeout(BOUND, stream.read_exact(&mut bnd)) + .await + .map_err(|e| e.to_string())? + .map_err(|e| e.to_string())?; + stream + .write_all(b"door-tcp") + .await + .map_err(|e| e.to_string())?; + let mut buf = [0u8; 8]; + timeout(BOUND, stream.read_exact(&mut buf)) + .await + .map_err(|e| e.to_string())? + .map_err(|e| e.to_string())?; + if &buf != b"door-tcp" { + return Err("door connect echo mismatch".to_string()); + } + Ok(()) + }; + + let door_udp = async { + let sock = UdpSocket::bind("127.0.0.1:0") + .await + .map_err(|e| e.to_string())?; + sock.send_to(&rfc_datagram(udp_target, b"door-udp"), relay_addr) + .await + .map_err(|e| e.to_string())?; + let mut rx = vec![0u8; 65535]; + let (n, _) = timeout(BOUND, sock.recv_from(&mut rx)) + .await + .map_err(|e| format!("door udp recv: {e}"))? + .map_err(|e| e.to_string())?; + if parse_rfc_datagram(&rx[..n]).1 != b"door-udp" { + return Err("door udp echo mismatch".to_string()); + } + Ok(()) + }; + + let (r1, r2, r3, r4) = tokio::join!(native, native_udp, door_tcp, door_udp); + r1.expect("native connect"); + r2.expect("native associate"); + r3.expect("door connect"); + r4.expect("door associate"); + + // The registration is healthy after the mixed burst: one more of + // each shape succeeds. + let (mut fresh, code, _) = timeout(BOUND, vanilla_cred_connect(door_addr, tcp_target)) + .await + .expect("fresh door connect"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + fresh.write_all(b"again").await.expect("write"); + let mut buf = [0u8; 5]; + timeout(BOUND, fresh.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"again"); + + let mut fresh_native = + Socks5ChannelSession::connect(topo.consumer.as_ref(), TargetAddr::Ip(tcp_target)) + .await + .expect("fresh native connect"); + let (mut read, mut write) = fresh_native.split().expect("pipe"); + write.write_all(b"again2").await.expect("write"); + let mut buf2 = [0u8; 6]; + timeout(BOUND, read.read_exact(&mut buf2)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf2, b"again2"); + drop(read); + drop(write); + fresh_native.close().await; +} + +/// The mixed-composed variant of the mixed-assembly pin: channels- +/// native sessions AND composed seams AND door clients on one +/// registration — the composed suite's producer + a door consumer +/// topology, both shapes live on one connection (roles-not-sides: the +/// producer serves every shape through its seams; the connection +/// direction is independent of the service direction). +#[tokio::test] +async fn mixed_assembly_composed_seams_beside_channels_native_sessions() { + let suite = wire_composed_suite().await; + let tcp_target = echo_tcp_server().await; + + // Channels-native CONNECT + composed BIND + composed ASSOCIATE, + // all opened before any is exercised (live simultaneously), then + // each exercised mid-life of the others; teardowns end only their + // own sessions. + let mut connect = timeout( + BOUND, + Socks5ChannelSession::connect(suite.topo.opener.as_ref(), TargetAddr::Ip(tcp_target)), + ) + .await + .expect("connect") + .expect("the CONNECT session"); + let mut assoc = timeout( + BOUND, + AssociateSession::associate(suite.topo.opener.as_ref()), + ) + .await + .expect("associate") + .expect("the association"); + let mut bind = timeout( + BOUND, + BindSession::bind( + suite.topo.opener.as_ref(), + ("bind.test", 2121).to_target_addr().unwrap(), + ), + ) + .await + .expect("bind") + .expect("the BIND session"); + + // The BIND's quiet wait is live (the listener exists); complete it + // mid-life. + composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await; + let mut plane = timeout(BOUND, bind.await_inbound()) + .await + .expect("reply#2") + .expect("await_inbound ok") + .expect("the plane"); + plane.write_all(b"mid").await.expect("write"); + let mut bbuf = [0u8; 3]; + timeout(BOUND, plane.read_exact(&mut bbuf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&bbuf, b"mid"); + + // The CONNECT + ASSOCIATE sessions still serve (no connection + // poisoning). + let (mut read, mut write) = connect.split().expect("pipe"); + write.write_all(b"alive").await.expect("write"); + let mut buf = [0u8; 5]; + timeout(BOUND, read.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"alive"); + + let dest = TargetAddr::Ip(SocketAddr::from(([10, 9, 9, 9], 5353))); + assoc.send_datagram(&dest, b"alive").await.expect("send"); + let (src, payload) = timeout(BOUND, assoc.recv_datagram()) + .await + .expect("recv") + .expect("a datagram") + .expect("a datagram (not eof)"); + assert_eq!(src, dest); + assert_eq!(&payload[..], b"alive"); + + // Teardowns end only their own sessions; the registration serves + // a fresh CONNECT after. + drop(read); + drop(write); + drop(assoc); + drop(plane); + assert!(connect.close().await); + assert!(bind.close().await); + let mut fresh = timeout( + BOUND, + Socks5ChannelSession::connect(suite.topo.opener.as_ref(), TargetAddr::Ip(tcp_target)), + ) + .await + .expect("fresh connect") + .expect("the fresh session"); + let (mut read, mut write) = fresh.split().expect("pipe"); + write.write_all(b"again").await.expect("write"); + let mut buf = [0u8; 5]; + timeout(BOUND, read.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"again"); + drop(read); + drop(write); + fresh.close().await; + + // The handler-exit teardowns reap asynchronously (the open + // wrapper's post-handler teardown) — settle past the race, then + // probe (the composed suites' shape). + for _ in 0..50 { + tokio::time::sleep(Duration::from_millis(100)).await; + let ids = suite.topo.producer_manager.channel_ids(); + if ids.iter().all(|&id| id == 0 || id % 2 == 1) { + break; + } + } + // No channel leaks (the composed suites' probe; the producer side + // may retain the adopted forwarded entries — the documented + // concession). + harness::assert_composed_channels(&suite.topo, &[0], &[]).await; +} + +/// The producer-side local backend as the `-D` story's other half (the +/// one-registration-three-commands regression through the producer's +/// own front door): a vanilla NoAuth client drives CONNECT + BIND + +/// ASSOCIATE through one `ProducerDoor` against one producer +/// registration (the `local` provider set), all three live +/// simultaneously — one accepted connection per command (the RFC +/// conversation is one command per accepted socket, the vanilla +/// server's shape). +#[tokio::test] +async fn producer_door_one_registration_three_commands() { + init_trace(); + let tcp_target = echo_tcp_server().await; + let udp_target = echo_udp_server().await; + + let producer = alksocks::producer::Socks5Producer::builder() + .dial(alksocks::local::tcp_dial()) + .egress(alksocks::local::udp_egress()) + .accept(alksocks::local::bind_accept([127, 0, 0, 1])) + .build(); + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let door_addr = listener.local_addr().expect("door addr"); + let door = alksocks::local::ProducerDoor::builder() + .producer(producer) + .build(); + let _guard = door.serve_on(listener).await; + + // CONNECT (live). + let (mut connect_stream, code, _) = + timeout(BOUND, vanilla_no_auth_connect(door_addr, tcp_target)) + .await + .expect("connect round trip"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + + // BIND (live, mid-life of the CONNECT) — its own door connection. + let mut bind_stream = TcpStream::connect(door_addr).await.expect("door connect"); + bind_stream + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting"); + let mut chosen = [0u8; 2]; + timeout(BOUND, bind_stream.read_exact(&mut chosen)) + .await + .expect("method selection") + .expect("read"); + assert_eq!(chosen, [0x05, 0x00]); + bind_stream + .write_all(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) + .await + .expect("bind request"); + let mut head1 = [0u8; 4]; + timeout(BOUND, bind_stream.read_exact(&mut head1)) + .await + .expect("bind reply#1") + .expect("read"); + assert_eq!(head1[1], consts::SOCKS5_REPLY_SUCCEEDED); + let mut bnd = [0u8; 6]; + timeout(BOUND, bind_stream.read_exact(&mut bnd)) + .await + .expect("reply#1 BND.ADDR") + .expect("read"); + let advertised = SocketAddr::from(( + [bnd[0], bnd[1], bnd[2], bnd[3]], + u16::from_be_bytes([bnd[4], bnd[5]]), + )); + let mut app = timeout(BOUND, TcpStream::connect(advertised)) + .await + .expect("dial the advertised address") + .expect("app connected"); + app.write_all(b"prod-bind").await.expect("app writes"); + let mut head2 = [0u8; 4]; + timeout(BOUND, bind_stream.read_exact(&mut head2)) + .await + .expect("bind reply#2") + .expect("read"); + assert_eq!(head2[1], consts::SOCKS5_REPLY_SUCCEEDED, "BIND reply#2"); + let mut peer_addr = [0u8; 6]; + timeout(BOUND, bind_stream.read_exact(&mut peer_addr)) + .await + .expect("reply#2 DST.ADDR") + .expect("read"); + assert_eq!( + u16::from_be_bytes([peer_addr[4], peer_addr[5]]), + app.local_addr().expect("app local").port(), + "reply#2 carries the real peer" + ); + + // ASSOCIATE (live, mid-life of both) — its own door connection: + // the sentinel reply + the in-band framed datagram through the + // real egress (the producer door path speaks the in-band wire — + // the ASSOCIATE datagram plane has no door-side RFC codec here; + // the vanilla client is wrapper-aware on this path). + let mut assoc_stream = TcpStream::connect(door_addr).await.expect("door connect"); + assoc_stream + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting"); + let mut chosen3 = [0u8; 2]; + timeout(BOUND, assoc_stream.read_exact(&mut chosen3)) + .await + .expect("method selection") + .expect("read"); + assert_eq!(chosen3, [0x05, 0x00]); + assoc_stream + .write_all(&[0x05, 0x03, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) + .await + .expect("associate request"); + let mut head3 = [0u8; 4]; + timeout(BOUND, assoc_stream.read_exact(&mut head3)) + .await + .expect("associate reply") + .expect("read"); + assert_eq!( + head3[1], + consts::SOCKS5_REPLY_SUCCEEDED, + "ASSOCIATE through the door" + ); + let mut bnd3 = [0u8; 6]; + timeout(BOUND, assoc_stream.read_exact(&mut bnd3)) + .await + .expect("the sentinel BND.ADDR") + .expect("read"); + assert_eq!(&bnd3, &[127, 0, 0, 1, 0, 0], "the sentinel"); + + // All three data planes, live simultaneously. + connect_stream.write_all(b"prod-tcp").await.expect("write"); + let mut buf = [0u8; 8]; + timeout(BOUND, connect_stream.read_exact(&mut buf)) + .await + .expect("tcp echo") + .expect("read"); + assert_eq!(&buf, b"prod-tcp"); + + let mut app_buf = [0u8; 9]; + timeout(BOUND, bind_stream.read_exact(&mut app_buf)) + .await + .expect("bind data plane") + .expect("read"); + assert_eq!(&app_buf, b"prod-bind"); + + let dst = TargetAddr::Ip(udp_target); + let frame = alksocks::wire::frame(&dst, b"prod-udp").expect("frame"); + assoc_stream.write_all(&frame).await.expect("datagram"); + let mut reply = vec![0u8; frame.len()]; + timeout(BOUND, assoc_stream.read_exact(&mut reply)) + .await + .expect("udp echo") + .expect("read"); + let n = u16::from_be_bytes([reply[0], reply[1]]) as usize; + let inner = &reply[2..2 + n]; + let atyp = inner[0]; + let addr_len = match atyp { + 0x01 => 1 + 4 + 2, + 0x03 => 1 + 1 + inner[1] as usize + 2, + 0x04 => 1 + 16 + 2, + other => panic!("unexpected ATYP {other}"), + }; + assert_eq!(&inner[addr_len..], b"prod-udp", "the datagram echoed"); + + // The teardown: the client drops; the door stays serving (the + // loop continues — the per-accept isolation); a fresh session of + // every command succeeds (fresh CONNECT here; the door is + // healthy). + drop(connect_stream); + drop(app); + let (_, code, _) = timeout(BOUND, vanilla_no_auth_connect(door_addr, tcp_target)) + .await + .expect("fresh connect after the teardown"); + assert_eq!( + code, + consts::SOCKS5_REPLY_SUCCEEDED, + "no connection poisoning" + ); +} + +/// The `-D` story's vanilla-client-fidelity pin (tun2proxy-shaped): a +/// domain-form CONNECT through the cred-bearing door resolves at the +/// producer's seam (the ssh `-D` semantic, ADR-012) — `localhost` is +/// reachable as the echo target. +#[tokio::test] +async fn dash_d_domain_targets_resolve_at_the_producer_seam() { + init_trace(); + let tcp_target = echo_tcp_server().await; + let topo = wire_consumer_side_with( + alksocks::local::tcp_dial(), + None, + None, + Some(harness::consumer_identity()), + ) + .await; + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let door_addr = listener.local_addr().expect("door addr"); + let door = FrontDoor::builder().bind(door_addr).build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door.serve_on(listener, client) + .await + .expect("the door serves") + } + }); + + // The NoAuth door is the raw pipe: the vanilla client's whole RFC + // conversation (domain-form CONNECT included) crosses it verbatim + // and resolves at the producer's dial seam. + let mut stream = TcpStream::connect(door_addr).await.expect("door connect"); + stream + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting"); + let mut chosen = [0u8; 2]; + stream + .read_exact(&mut chosen) + .await + .expect("method selection"); + assert_eq!(chosen, [0x05, 0x00]); + + let domain = "localhost".to_owned(); + let mut req = vec![0x05, 0x01, 0x00, 0x03, domain.len() as u8]; + req.extend_from_slice(domain.as_bytes()); + req.extend_from_slice(&tcp_target.port().to_be_bytes()); + stream.write_all(&req).await.expect("request"); + let mut head = [0u8; 4]; + timeout(BOUND, stream.read_exact(&mut head)) + .await + .expect("reply") + .expect("read"); + assert_eq!( + head[1], + consts::SOCKS5_REPLY_SUCCEEDED, + "the domain resolved" + ); + let mut bnd = [0u8; 6]; + timeout(BOUND, stream.read_exact(&mut bnd)) + .await + .expect("BND.ADDR") + .expect("read"); + stream.write_all(b"dom").await.expect("write"); + let mut buf = [0u8; 3]; + timeout(BOUND, stream.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"dom"); +} diff --git a/tests/composed-accept.rs b/tests/composed-accept.rs index 42c07b7..5c50a19 100644 --- a/tests/composed-accept.rs +++ b/tests/composed-accept.rs @@ -17,151 +17,43 @@ 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 alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; +use alktunnels::params::Substrate; +use alktunnels::producer::ResourceRegistry; 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(); -} +use harness::{ + assert_composed_channels, composed_fresh_queue, composed_opener_identity, composed_push_accept, + composed_queue_loop, composed_scopeless_opener_identity, composed_test_bind, + wire_bindfirst_composed, COMPOSED_BOUND_PORT, COMPOSED_PEER_PORT, COMPOSED_RESOURCE, +}; /// 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, + queue: harness::ComposedAcceptQueue, } async fn wire_suite() -> Suite { - wire_suite_with_identity(harness::composed_opener_identity()).await + wire_suite_with_identity(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") + .register(COMPOSED_RESOURCE, Substrate::Tcp, "assembly-owned") .await; - let queue = fresh_queue(); - let topo = harness::wire_bindfirst_composed( + let queue = composed_fresh_queue(); + let topo = wire_bindfirst_composed( resource_registry, - test_bind(), - queue_loop(queue.clone()), + composed_test_bind(), + composed_queue_loop(queue.clone()), None, // the SOCKS5 openable registers post-hoc, below — the composed factory needs the opener's client identity, ) @@ -180,7 +72,7 @@ async fn wire_suite_with_identity(identity: alkcall::core::auth::Identity) -> Su // 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 factory = bindfirst_accept(topo.opener.clone(), registry, COMPOSED_RESOURCE); let core = alkcall::channels::operations::ChannelCore::new( topo.producer_manager.clone(), alkcall::channels::policy::default_policy(), @@ -212,33 +104,6 @@ async fn bind_session(suite: &Suite) -> BindSession { .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 @@ -268,10 +133,7 @@ async fn composed_bind_full_story_bound_peer_and_round_trip() { 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 - ))), + TargetAddr::Ip(SocketAddr::from(([127, 0, 0, 1], COMPOSED_BOUND_PORT))), "reply#1 advertises the real OS-chosen bound" ); @@ -279,7 +141,7 @@ async fn composed_bind_full_story_bound_peer_and_round_trip() { // 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; + composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await; let mut plane = tokio::time::timeout(Duration::from_secs(10), session.await_inbound()) .await .expect("reply#2 within timeout") @@ -287,10 +149,7 @@ async fn composed_bind_full_story_bound_peer_and_round_trip() { .expect("the plane exists"); assert_eq!( plane.peer, - alksocks::rfc1928::util::target_addr::TargetAddr::Ip(SocketAddr::from(( - [10, 0, 0, 1], - PEER_PORT - ))), + TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], COMPOSED_PEER_PORT))), "reply#2 carries the forwarded op's real peer" ); @@ -324,10 +183,10 @@ async fn composed_bind_full_story_bound_peer_and_round_trip() { // 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` + // retain the adopted forwarded channel (see `assert_composed_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; + assert_composed_channels(&suite.topo, &[0], &[channel_id]).await; } /// The client-gone teardown (the drop-witness): dropping the BIND @@ -382,7 +241,7 @@ async fn client_gone_teardown_releases_the_listener() { suite.queue.0.lock().await.is_empty(), "the dropped listener consumed nothing" ); - assert_only_channels(&suite.topo, &[0], &[0]).await; + assert_composed_channels(&suite.topo, &[0], &[0]).await; } /// Two concurrent BIND sessions get distinct tokens (the mint is @@ -401,8 +260,8 @@ async fn concurrent_bind_sessions_get_distinct_tokens() { // 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; + composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await; + composed_push_accept(&suite.queue, ("10.0.0.2", COMPOSED_PEER_PORT + 1)).await; let mut p1 = tokio::time::timeout(Duration::from_secs(10), s1.await_inbound()) .await @@ -442,7 +301,7 @@ async fn concurrent_bind_sessions_get_distinct_tokens() { assert!(s1.close().await); assert!(s2.close().await); tokio::time::sleep(Duration::from_millis(300)).await; - assert_only_channels(&suite.topo, &[0], &[0]).await; + assert_composed_channels(&suite.topo, &[0], &[0]).await; } /// A forwarded-open failure is per-accept isolated: an opener identity @@ -452,13 +311,13 @@ async fn concurrent_bind_sessions_get_distinct_tokens() { /// 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 suite = wire_suite_with_identity(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; + composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await; // The forwarded-open failure is per-accept isolated AT THE TUNNEL // LAYER: the producer's accept loop drops that accepted connection @@ -468,7 +327,7 @@ async fn forwarded_open_failure_is_per_accept_isolated() { // 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; + composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await; tokio::time::sleep(Duration::from_millis(300)).await; assert!( suite.queue.0.lock().await.is_empty(), @@ -493,10 +352,10 @@ async fn late_forwarded_open_after_client_gone_drops_the_channel() { // 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; + composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_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; + assert_composed_channels(&suite.topo, &[0], &[0]).await; } diff --git a/tests/composed-egress.rs b/tests/composed-egress.rs index 439bcb4..87ec2d4 100644 --- a/tests/composed-egress.rs +++ b/tests/composed-egress.rs @@ -161,6 +161,33 @@ async fn datagrams_round_trip_per_destination_multiple_tunnels() { .expect_err("next_reply stays Pending when nothing is ready"); } +/// A fresh association (no tunnel ever opened) parks in `next_reply` +/// — the stage's select is governed by the channel branch until the +/// first datagram opens a tunnel (the assembly suite's drift finding: +/// the ended-relay `Err` fired on the never-opened table too, killing +/// every quiescent association at birth). Pinned on its own. +#[tokio::test] +async fn fresh_association_parks_before_any_tunnel_is_opened() { + let dial_count = Arc::new(AtomicUsize::new(0)); + let topo = wire_direct(counting_udp_echo_dial(Arc::clone(&dial_count))).await; + let mut egress = egress_for(&topo, 8).await; + + // The never-opened table: `next_reply` parks (Pending), never an + // instantly-ready `Err` (the stage would end the association + // before its first datagram). + { + let mut poll = egress.next_reply(); + tokio::time::timeout(std::time::Duration::from_millis(100), &mut poll) + .await + .expect_err("the fresh association's next_reply parks"); + } + + // After the first datagram opens a tunnel, the normal contract + // holds (the round-trip works). + round_trip(egress.as_mut(), &dest(53), b"after").await; + assert_eq!(dial_count.load(Ordering::SeqCst), 1); +} + #[tokio::test] async fn lru_eviction_reopens_evicted_destinations() { let dial_count = Arc::new(AtomicUsize::new(0)); diff --git a/tests/harness.rs b/tests/harness.rs index 5f96950..a28cf34 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -1211,6 +1211,182 @@ pub fn composed_scopeless_opener_identity() -> Identity { } } +/// The composed suite's full identity: the five gates the composed +/// three-command story crosses (`tunnel:open` + `tunnel:direct` + +/// `tunnel:forwarded` + `socks5:open` + `socks5:bind`) — the assembly +/// suite's producer registration serves the direct op (CONNECT + +/// ASSOCIATE's tunnels) beside the bind-first op (BIND's listener). +pub fn composed_full_opener_identity() -> Identity { + Identity { + id: "composed-opener-full".to_string(), + scopes: vec![ + alktunnels::TUNNEL_OPEN_SCOPE.to_string(), + alktunnels::TUNNEL_DIRECT_SCOPE.to_string(), + alktunnels::TUNNEL_FORWARDED_SCOPE.to_string(), + SOCKS5_OPEN_SCOPE.to_string(), + SOCKS5_BIND_SCOPE.to_string(), + ], + resources: HashMap::new(), + } +} + +// --- The composed accept's shared machinery (review 002 N-5) ---------------- +// +// The bind/accept-queue stand-ins and the channel-set probe the +// composed suites consume — single-sourced here now that two suites +// (composed-accept + assembly-integration) drive the same shapes. + +/// The resource the bind-first producer serves (the composed suites' +/// only subscription — the registry namespace gate). +pub const COMPOSED_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). +pub const COMPOSED_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). +pub const COMPOSED_PEER_PORT: u16 = 55231; + +/// The test bind (the assembly's bind seam): reports the test's +/// bound address + an opaque listener witness. +pub fn composed_test_bind() -> alktunnels::producer::BindFn { + Arc::new(|_substrate, _backing| { + Box::pin(async { + Ok(alktunnels::producer::BoundListener { + listener: Arc::new(()) as Arc, + bound: alktunnels::params::SubstrateAddr::HostPort { + host: "127.0.0.1".to_string(), + port: COMPOSED_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). +pub type ComposedAcceptQueue = ( + Arc< + tokio::sync::Mutex< + std::collections::VecDeque<( + alktunnels::params::SubstrateAddr, + alktunnels::producer::TargetHandle, + )>, + >, + >, + Arc, +); + +pub fn composed_fresh_queue() -> ComposedAcceptQueue { + ( + Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new())), + Arc::new(tokio::sync::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 composed_accepted_halves( + peer_addr: (&str, u16), +) -> ( + alktunnels::params::SubstrateAddr, + alktunnels::producer::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); + ( + alktunnels::params::SubstrateAddr::HostPort { + host: peer_addr.0.to_string(), + port: peer_addr.1, + }, + alktunnels::producer::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. +pub fn composed_queue_loop( + queue: ComposedAcceptQueue, +) -> alktunnels::producer::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(alktunnels::TunnelEstablishError::DialFailed( + "listener closed".to_string(), + )) + }) + }) +} + +/// Push an accepted pair into the test's accept queue. +pub async fn composed_push_accept(queue: &ComposedAcceptQueue, peer: (&str, u16)) { + let (peer_addr, handle) = composed_accepted_halves(peer); + queue.0.lock().await.push_back((peer_addr, handle)); + queue.1.notify_waiters(); +} + +/// The composed suites' 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`). +pub async fn assert_composed_channels( + topo: &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:?}" + ); +} + #[cfg(test)] mod tests { use super::*;