//! The TCP front door's integration tests (feature `local`): the //! NoAuth raw pipe with a vanilla RFC 1928 client (the POC #1 curl //! shape), the cred-bearing door's RFC 1929 (good creds → CONNECT //! round-trip; bad creds → rejection with no channel opened; the //! command bytes never reach the producer as greeting bytes), the //! BIND door (reply#1/reply#2 fidelity), the ADR-014 accept-path //! defense (the idle-first-byte timeout + the session cap), and the //! per-connection failure isolation. #![cfg(feature = "local")] mod harness; use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; use tokio::time::timeout; use alkcall::core::auth::{Identity, IdentityProvider}; use alksocks::consumer::{BindSession, Socks5ChannelSession}; use alksocks::rfc1928::consts; use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; use alksocks::{FrontDoor, SOCKS5_BIND_SCOPE}; use harness::consumer_identity; const BOUND: Duration = Duration::from_secs(10); /// The token→identity provider the cred-bearing door's producer runs /// (CF-005 precedence (1)): the token is the mapped identity's id, /// the resolution seam's vocabulary; the password never rides it. /// Resolves the credential-table identity ids (the `door-` prefixed /// forms the tests configure) to identities carrying `socks5:open` /// (plus `socks5:bind` when the entry is bind-scoped); anything else /// resolves to none (the dispatch falls through to the transport /// identity per the CF-005 precedence order). #[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 } } /// An echo TCP server on an ephemeral loopback port (the vanilla /// client's CONNECT target, the POC #1 curl shape). 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 } /// The door's raw RFC 1928 client (the differential-suite pattern): a /// `TcpStream` dialed to the door, greeting + method selection + /// command + reply read. struct VanillaClient { stream: TcpStream, } impl VanillaClient { async fn dial(addr: SocketAddr) -> Self { Self { stream: TcpStream::connect(addr).await.expect("dial the door"), } } /// The dial without the panic (the `serve` wrapper's bind races /// the test's connect, the retry loop's probe). async fn try_dial(addr: SocketAddr) -> std::io::Result { Ok(Self { stream: TcpStream::connect(addr).await?, }) } /// Greeting `05 02 00` (password auth offered, the cred-bearing /// door's shape) → the method-selection read. async fn greet_password(&mut self) -> u8 { self.stream .write_all(&[0x05, 0x01, 0x02]) .await .expect("greeting written"); let mut chosen = [0u8; 2]; self.stream .read_exact(&mut chosen) .await .expect("method selected"); assert_eq!(chosen[0], consts::SOCKS5_VERSION); chosen[1] } /// Greeting `05 01 00` (no-auth offered) → the method-selection /// read. async fn greet_no_auth(&mut self) -> u8 { self.stream .write_all(&[0x05, 0x01, 0x00]) .await .expect("greeting written"); let mut chosen = [0u8; 2]; self.stream .read_exact(&mut chosen) .await .expect("method selected"); assert_eq!(chosen[0], consts::SOCKS5_VERSION); chosen[1] } /// The RFC 1929 credential sub-negotiation → the status byte. async fn send_credentials(&mut self, user: &str, pass: &str) -> u8 { let mut packet = vec![0x01, user.len() as u8]; packet.extend_from_slice(user.as_bytes()); packet.push(pass.len() as u8); packet.extend_from_slice(pass.as_bytes()); self.stream.write_all(&packet).await.expect("creds sent"); let mut status = [0u8; 2]; self.stream.read_exact(&mut status).await.expect("status"); assert_eq!(status[0], 0x01, "the sub-negotiation version is 1"); status[1] } /// A CONNECT request (v4 target) → the reply code. async fn connect_v4(&mut self, target: SocketAddr) -> u8 { let mut req = [0u8; 10]; req[..3].copy_from_slice(&[0x05, consts::SOCKS5_CMD_TCP_CONNECT, 0x00]); req[3] = 0x01; let SocketAddr::V4(v4) = target else { panic!("v4 targets only"); }; req[4..8].copy_from_slice(&v4.ip().octets()); req[8..10].copy_from_slice(&v4.port().to_be_bytes()); self.stream.write_all(&req).await.expect("request sent"); let (code, _addr) = self.read_reply_addr().await; code } /// A BIND request (v4 target) → reply#1's code + the advertised /// BND.ADDR. async fn bind_v4(&mut self, ip: [u8; 4], port: u16) -> (u8, TargetAddr) { let mut req = [0u8; 10]; req[..3].copy_from_slice(&[0x05, consts::SOCKS5_CMD_TCP_BIND, 0x00]); req[3] = 0x01; req[4..8].copy_from_slice(&ip); req[8..10].copy_from_slice(&port.to_be_bytes()); self.stream.write_all(&req).await.expect("bind sent"); self.read_reply_addr().await } /// Read one reply (any BND.ADDR form) → (code, addr). async fn read_reply_addr(&mut self) -> (u8, TargetAddr) { let mut head = [0u8; 4]; self.stream.read_exact(&mut head).await.expect("reply head"); assert_eq!(head[0], consts::SOCKS5_VERSION, "reply version 5"); let addr = match head[3] { consts::SOCKS5_ADDR_TYPE_IPV4 => { let mut rest = [0u8; 6]; self.stream.read_exact(&mut rest).await.expect("v4 addr"); TargetAddr::Ip(SocketAddr::from(( std::net::Ipv4Addr::from([rest[0], rest[1], rest[2], rest[3]]), u16::from_be_bytes([rest[4], rest[5]]), ))) } consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME => { let mut len = [0u8; 1]; self.stream.read_exact(&mut len).await.expect("domain len"); let mut rest = vec![0u8; len[0] as usize + 2]; self.stream.read_exact(&mut rest).await.expect("domain"); TargetAddr::Domain( String::from_utf8(rest[..rest.len() - 2].to_vec()).expect("domain utf8"), u16::from_be_bytes([rest[rest.len() - 2], rest[rest.len() - 1]]), ) } other => panic!("unknown reply ATYP {other:#x}"), }; (head[1], addr) } } /// The NoAuth door end to end (the POC #1 curl shape): a vanilla RFC /// 1928 client through the raw pipe → CONNECT round-trip to a real /// TCP echo target. Zero protocol translation pinned: the door adds /// nothing; the producer runs the whole RFC conversation through the /// pipe (the vanilla client's greeting crosses it verbatim). #[tokio::test] async fn noauth_door_raw_pipe_vanilla_connect_round_trip() { let echo = echo_tcp_server().await; let dial = alksocks::tcp_dial(); let topo = harness::wire_consumer_side_with(dial, None, None, None).await; // The door takes a pre-bound listener (`serve_on`), so the test // owns the port (the assembly seam: the same accept loop without // the door's own bind). let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); let door = FrontDoor::builder() .bind(addr) .build() .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(listener, client) .await .expect("the door serves") } }); let mut vanilla = VanillaClient::dial(addr).await; let method = timeout(BOUND, vanilla.greet_no_auth()) .await .expect("greeting round-trips through the raw pipe"); assert_eq!(method, consts::SOCKS5_AUTH_METHOD_NONE); let code = timeout(BOUND, vanilla.connect_v4(echo)) .await .expect("CONNECT round-trips through the raw pipe"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); // The data plane is the literal pipe: echo both directions. vanilla.stream.write_all(b"vanilla-echo").await.unwrap(); let mut buf = [0u8; 12]; timeout(BOUND, vanilla.stream.read_exact(&mut buf)) .await .expect("echo within bound") .expect("read"); assert_eq!(&buf, b"vanilla-echo", "the raw pipe round-trips"); topo.consumer.manager(); } /// The cred-bearing door: good creds → RFC 1929 success → CONNECT /// round-trip. The mapped identity reaches the producer (the /// establisher witness sees the mapped identity id, never the /// password); the in-band conversation is re-originated (the vanilla /// client's command bytes were consumed by the door's own RFC half). #[tokio::test] async fn cred_door_good_creds_connect_round_trip_with_mapped_identity() { let echo = echo_tcp_server().await; let dial = alksocks::tcp_dial(); let topo = harness::wire_consumer_side_with_provider( dial, None, None, Some(consumer_identity()), Arc::new(DoorTokenProvider { bind_scoped: vec![], }), ) .await; let _ = echo; let table = alksocks::CredentialTable::new(vec![alksocks::DoorCredential { username: "alice".to_string(), password: "secret".to_string(), identity: Identity { id: "door-alice".to_string(), scopes: vec![alksocks::SOCKS5_OPEN_SCOPE.to_string()], resources: Default::default(), }, }]); let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); let door = FrontDoor::builder() .credentials(table) .bind(addr) .build() .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(listener, client) .await .expect("the door serves") } }); let mut vanilla = VanillaClient::dial(addr).await; let method = timeout(BOUND, vanilla.greet_password()) .await .expect("greeting"); assert_eq!(method, consts::SOCKS5_AUTH_METHOD_PASSWORD); let status = timeout(BOUND, vanilla.send_credentials("alice", "secret")) .await .expect("credential exchange"); assert_eq!(status, 0x00, "accepted"); assert_eq!(status, 0x00, "RFC 1929 success"); let code = timeout(BOUND, vanilla.connect_v4(echo)) .await .expect("CONNECT after auth"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); vanilla.stream.write_all(b"cred-echo").await.unwrap(); let mut buf = [0u8; 9]; timeout(BOUND, vanilla.stream.read_exact(&mut buf)) .await .expect("echo within bound") .expect("read"); assert_eq!(&buf, b"cred-echo"); // The establisher witness saw the mapped identity (CF-006 via the // token resolution, CF-005 precedence (1)), never the password. timeout(BOUND, async { loop { let seen = topo.identity_witness.lock().await.clone(); if seen.as_deref() == Some("door-alice") { return; } tokio::time::sleep(Duration::from_millis(10)).await; } }) .await .expect("the establisher witnessed the mapped identity"); } /// The cred-bearing door: bad creds → the RFC 1929 rejection (status /// 0xff), and no channel opened (the identity witness never fires; /// the consumer's manager holds only channel 0 (the ACL boundary /// held door-side). #[tokio::test] async fn cred_door_bad_creds_rejected_no_channel_opened() { let dial = alksocks::tcp_dial(); let table = alksocks::CredentialTable::new(vec![alksocks::DoorCredential { username: "alice".to_string(), password: "secret".to_string(), identity: Identity { id: "door-alice".to_string(), scopes: vec![alksocks::SOCKS5_OPEN_SCOPE.to_string()], resources: Default::default(), }, }]); let topo = harness::wire_consumer_side_with_provider( dial, None, None, Some(consumer_identity()), Arc::new(DoorTokenProvider { bind_scoped: vec![], }), ) .await; let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); let door = FrontDoor::builder() .credentials(table) .bind(addr) .build() .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(listener, client) .await .expect("the door serves") } }); let mut vanilla = VanillaClient::dial(addr).await; let method = timeout(BOUND, vanilla.greet_password()) .await .expect("greeting"); assert_eq!(method, consts::SOCKS5_AUTH_METHOD_PASSWORD); let status = timeout(BOUND, vanilla.send_credentials("alice", "wrong")) .await .expect("credential exchange"); assert_eq!(status, 0xff, "RFC 1929 rejection"); // No channel opened: the witness never fired (the door-side // rejection never reached the producer), and the consumer's // manager holds only channel 0. tokio::time::sleep(Duration::from_millis(200)).await; assert_eq!( topo.identity_witness.lock().await.as_deref(), None, "the rejection never opened a channel" ); assert!( topo.consumer .manager() .channel_ids() .iter() .all(|&id| id == 0), "no channel entry exists opener-side" ); } /// The pin: a vanilla client's command bytes must not reach the /// producer as greeting bytes (the door never pipes raw post-auth). /// The producer's in-band conversation is re-originated: the /// producer's `accept_no_auth` reads a fresh greeting written by the /// *consumer session* (`Socks5Stream::use_stream`), not the vanilla /// client's consumed bytes. Observable as: the cred-bearing CONNECT /// round-trip works at all (raw post-auth piping would misparse the /// command bytes as a greeting (the producer would reply 0xff-shaped /// garbage), and the witness carries the mapped identity (the /// password never crossed). #[tokio::test] async fn cred_door_never_pipes_raw_post_auth() { let echo = echo_tcp_server().await; let dial = alksocks::tcp_dial(); let table = alksocks::CredentialTable::new(vec![alksocks::DoorCredential { username: "bob".to_string(), password: "pw".to_string(), identity: Identity { id: "door-bob".to_string(), scopes: vec![alksocks::SOCKS5_OPEN_SCOPE.to_string()], resources: Default::default(), }, }]); let topo = harness::wire_consumer_side_with_provider( dial, None, None, Some(consumer_identity()), Arc::new(DoorTokenProvider { bind_scoped: vec![], }), ) .await; let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); let door = FrontDoor::builder() .credentials(table) .bind(addr) .build() .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(listener, client) .await .expect("the door serves") } }); // The vanilla client speaks greeting → password-auth → command. // If the door piped these bytes raw post-auth, the producer's // in-band greeting read would consume `05 01 00 01 ...` (the // command request) as a method list and the conversation would // fail. The round-trip succeeding IS the pin (the re-origination // re-spoke the bytes in-band). let mut vanilla = VanillaClient::dial(addr).await; let method = timeout(BOUND, vanilla.greet_password()) .await .expect("greeting"); assert_eq!(method, consts::SOCKS5_AUTH_METHOD_PASSWORD, "password auth"); let status = timeout(BOUND, vanilla.send_credentials("bob", "pw")) .await .expect("creds"); assert_eq!(status, 0x00, "accepted"); let code = timeout(BOUND, vanilla.connect_v4(echo)) .await .expect("the in-band conversation is fresh (never the consumed bytes)"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); vanilla.stream.write_all(b"pin").await.unwrap(); let mut buf = [0u8; 3]; timeout(BOUND, vanilla.stream.read_exact(&mut buf)) .await .expect("echo") .expect("read"); assert_eq!(&buf, b"pin"); } /// The BIND door (cred-bearing variant): a vanilla BIND conversation /// through the door; reply#1's advertised address passes through /// unchanged (the door adds nothing), reply#2 carries the real peer, /// and the data plane round-trips (the POC #5 door shape). #[tokio::test] async fn bind_door_vanilla_conversation_reply_fidelity() { let dial = alksocks::tcp_dial(); let egress = None; let accept = Some(alksocks::bind_accept([127, 0, 0, 1])); let table = alksocks::CredentialTable::new(vec![alksocks::DoorCredential { username: "carol".to_string(), password: "pw".to_string(), identity: Identity { id: "door-carol".to_string(), scopes: vec![ alksocks::SOCKS5_OPEN_SCOPE.to_string(), SOCKS5_BIND_SCOPE.to_string(), ], resources: Default::default(), }, }]); let topo = harness::wire_consumer_side_with_provider( dial, egress, accept, Some(consumer_identity()), Arc::new(DoorTokenProvider { bind_scoped: vec!["door-carol".to_string()], }), ) .await; let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); let door = FrontDoor::builder() .credentials(table) .bind(addr) .build() .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(listener, client) .await .expect("the door serves") } }); let mut vanilla = VanillaClient::dial(addr).await; let method = timeout(BOUND, vanilla.greet_password()) .await .expect("greeting"); assert_eq!(method, consts::SOCKS5_AUTH_METHOD_PASSWORD, "password auth"); let status = timeout(BOUND, vanilla.send_credentials("carol", "pw")) .await .expect("creds"); assert_eq!(status, 0x00, "accepted"); // BIND request: 127.0.0.1:0 (ephemeral; the local accept // provider binds the real listener). let (code1, advertised) = timeout(BOUND, vanilla.bind_v4([127, 0, 0, 1], 0)) .await .expect("reply#1"); assert_eq!( code1, consts::SOCKS5_REPLY_SUCCEEDED, "BIND served under the gate" ); let TargetAddr::Ip(advertised_sa) = advertised else { panic!("an IP advertised address, got {advertised:?}") }; assert!( advertised_sa.ip().is_loopback() && advertised_sa.port() != 0, "reply#1 carries the real listener address: {advertised_sa}" ); // The app server dials the advertised address. let mut app = timeout(BOUND, TcpStream::connect(advertised_sa)) .await .expect("dial within bound") .expect("app dials the advertised address"); app.write_all(b"vanilla-bind").await.unwrap(); app.flush().await.unwrap(); // Reply#2: success + the real peer (the door adds nothing). let (code2, peer) = timeout(BOUND, vanilla.read_reply_addr()) .await .expect("reply#2"); assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED, "reply#2 success"); let TargetAddr::Ip(peer_sa) = peer else { panic!("an IP peer, got {peer:?}") }; assert_eq!( peer_sa.port(), app.local_addr().expect("app local").port(), "reply#2 carries the real peer port (byte-exact)" ); // The data plane through the door. let mut buf = [0u8; 12]; timeout(BOUND, vanilla.stream.read_exact(&mut buf)) .await .expect("data plane round trip") .expect("read"); assert_eq!(&buf, b"vanilla-bind"); drop(app); } /// The in-band refusal's reply code is re-encoded to the vanilla /// client (the RFC codec at the door boundary): an identity carrying /// `socks5:open` only issues BIND through the cred-bearing door; /// the in-band command gate refuses 0x02, and the door re-encodes /// that code as reply#1 (never a bare EOF; the refused reply carries /// the placeholder zero address). #[tokio::test] async fn cred_door_in_band_refusal_surfaces_the_reply_code() { let dial = alksocks::tcp_dial(); let table = alksocks::CredentialTable::new(vec![alksocks::DoorCredential { username: "open-only".to_string(), password: "pw".to_string(), identity: Identity { id: "door-open-only".to_string(), scopes: vec![alksocks::SOCKS5_OPEN_SCOPE.to_string()], resources: Default::default(), }, }]); let topo = harness::wire_consumer_side_with_provider( dial, None, None, Some(consumer_identity()), Arc::new(DoorTokenProvider { bind_scoped: vec![], }), ) .await; let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); let door = FrontDoor::builder() .credentials(table) .bind(addr) .build() .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(listener, client) .await .expect("the door serves") } }); let mut vanilla = VanillaClient::dial(addr).await; let method = timeout(BOUND, vanilla.greet_password()) .await .expect("greeting"); assert_eq!(method, consts::SOCKS5_AUTH_METHOD_PASSWORD); let status = timeout(BOUND, vanilla.send_credentials("open-only", "pw")) .await .expect("credential exchange"); assert_eq!(status, 0x00, "accepted"); // BIND through the open-only identity: the in-band command gate's // 0x02 rides the door's re-encoded reply (the placeholder BND.ADDR: // a refused reply carries a zero address). let (code, advertised) = timeout(BOUND, vanilla.bind_v4([127, 0, 0, 1], 0)) .await .expect("the refusal reply arrives (never a bare EOF)"); assert_eq!( code, consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED, "the in-band gate's 0x02 is re-encoded door-side" ); assert!( matches!(advertised, TargetAddr::Ip(sa) if sa.port() == 0 && sa.ip().is_unspecified()), "a refused reply carries the placeholder BND.ADDR: {advertised:?}" ); } /// The idle-first-byte timeout (ADR-014 knob 1): an accepted /// connection that sends no first byte within the window is closed /// door-side. #[tokio::test] async fn idle_first_byte_timeout_closes_a_silent_connection() { let dial = alksocks::tcp_dial(); let topo = harness::wire_consumer_side_with(dial, None, None, None).await; let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); let door = FrontDoor::builder() .bind(addr) .idle_first_byte(Duration::from_millis(300)) .build() .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(listener, client) .await .expect("the door serves") } }); // Connect and stay silent: the door must close the socket within // (window + slack). let mut silent = TcpStream::connect(addr).await.expect("dial the door"); let mut probe = [0u8; 1]; let start = std::time::Instant::now(); let outcome = timeout(Duration::from_secs(3), silent.read(&mut probe)).await; match outcome { Err(_) => panic!("the silent connection must be closed door-side"), Ok(Ok(0)) | Ok(Err(_)) => {} Ok(Ok(n)) => panic!("EOF expected, got {n} bytes"), } let elapsed = start.elapsed(); assert!( elapsed >= Duration::from_millis(250), "the door waited the window before closing (not instant): {elapsed:?}" ); assert!( elapsed < Duration::from_secs(2), "the close arrived near the window, not arbitrarily late: {elapsed:?}" ); // No channel was opened for the silent connection. tokio::time::sleep(Duration::from_millis(100)).await; assert!( topo.consumer .manager() .channel_ids() .iter() .all(|&id| id == 0), "the idle timeout fired before any channel opened" ); } /// The session cap (ADR-014 knob 2): an accept beyond the bound is /// refused door-side: no channel opened for the refused accept, and /// the door keeps serving (the loop continues). #[tokio::test] async fn session_cap_refuses_beyond_the_bound() { let dial = alksocks::tcp_dial(); let topo = harness::wire_consumer_side_with(dial, None, None, None).await; let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); let door = FrontDoor::builder() .bind(addr) .idle_first_byte(Duration::from_secs(30)) .max_sessions(1) .build() .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(listener, client) .await .expect("the door serves") } }); // Session 1: connects and stays silent (holds the one slot). let mut holder = TcpStream::connect(addr).await.expect("first dial"); let _ = &mut holder; // Session 2: beyond the cap, so refused. The socket closes without // a reply (the door-side refusal; no RFC bytes ever flow). let refused = TcpStream::connect(addr).await.expect("second dial"); let mut probe = [0u8; 1]; let outcome = timeout(Duration::from_secs(3), { let mut refused = refused; async move { refused.read(&mut probe).await } }) .await; match outcome { Err(_) => panic!("the beyond-cap connection must be refused"), Ok(Ok(0)) | Ok(Err(_)) => {} Ok(Ok(n)) => panic!("no bytes for a beyond-cap accept, got {n}"), } // Exactly one channel exists (the holder's, opened when it went // silent-past... wait, no: the holder sent no bytes either, so the // idle timeout governs it; the cap test needs the holder to stay // within the window while the second dial arrives. The holder's // channel opens only after its first byte (silence keeps it // channel-less. Assert: no channels at all for either connection // (both silent), and the door is still serving (a fresh dial // succeeds). tokio::time::sleep(Duration::from_millis(100)).await; assert!( topo.consumer .manager() .channel_ids() .iter() .all(|&id| id == 0), "neither silent connection opened a channel" ); // The door keeps serving (the loop continued). let fresh = TcpStream::connect(addr).await; assert!(fresh.is_ok(), "the door keeps accepting after refusals"); drop(fresh); } /// Per-connection failure isolation (the alktunnels per-accept /// posture): one dead client doesn't kill the door. A client that /// speaks garbage (an invalid greeting) gets its connection ended, /// and the door keeps serving fresh clients end to end. #[tokio::test] async fn per_connection_failure_isolation() { let echo = echo_tcp_server().await; let dial = alksocks::tcp_dial(); let topo = harness::wire_consumer_side_with(dial, None, None, None).await; let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); let door = FrontDoor::builder() .bind(addr) .build() .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(listener, client) .await .expect("the door serves") } }); // Client A: protocol garbage (a bad version byte through the raw // pipe; the producer's typestate refuses in-band, and the door's // pipe ends with the session). let mut garbage = VanillaClient::dial(addr).await; garbage.stream.write_all(&[0x04, 0x01, 0x00]).await.unwrap(); // Give the producer's refusal a beat; the door's pipe tears down. tokio::time::sleep(Duration::from_millis(200)).await; drop(garbage); // Client B: the full vanilla CONNECT round-trip still works (the // door survived the dead client). let mut good = VanillaClient::dial(addr).await; let method = timeout(BOUND, good.greet_no_auth()) .await .expect("the door is alive"); assert_eq!(method, consts::SOCKS5_AUTH_METHOD_NONE); let code = timeout(BOUND, good.connect_v4(echo)) .await .expect("the door survived the dead client"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); good.stream.write_all(b"alive").await.unwrap(); let mut buf = [0u8; 5]; timeout(BOUND, good.stream.read_exact(&mut buf)) .await .expect("echo") .expect("read"); assert_eq!(&buf, b"alive"); } /// The channels-native sessions still work beside the door (the /// regression guard: the door's `TokenClient::wrap` took the client's /// call connection: a NoAuth door never wraps, so the client's own /// sessions keep serving; a cred-bearing door's client is door-only /// by design (the wrap's contract, asserted here at the surface /// level via a NoAuth topology). #[tokio::test] async fn native_sessions_work_beside_a_noauth_door() { let echo = echo_tcp_server().await; let dial = alksocks::tcp_dial(); let topo = harness::wire_consumer_side_with(dial, None, None, None).await; let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); let door = FrontDoor::builder() .bind(addr) .build() .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(listener, client) .await .expect("the door serves") } }); // The channels-native CONNECT session rides the same client. let mut session = Socks5ChannelSession::connect(topo.consumer.as_ref(), TargetAddr::Ip(echo)) .await .expect("the native session connects"); let (mut read, mut write) = session.split().expect("post-reply pipe"); write.write_all(b"native").await.unwrap(); write.flush().await.unwrap(); let mut buf = [0u8; 6]; timeout(BOUND, read.read_exact(&mut buf)) .await .expect("echo") .expect("read"); assert_eq!(&buf, b"native"); session.close().await; } /// The BIND door via the channels-native session (the regression /// guard on the door's BIND path machinery: the same /// `BindSession::bind_with_token` surface the door rides, driven /// directly with the token provider wired). #[tokio::test] async fn bind_session_with_token_round_trip() { let dial = alksocks::tcp_dial(); let accept = Some(alksocks::bind_accept([127, 0, 0, 1])); let topo = harness::wire_consumer_side_with_provider( dial, None, accept, Some(consumer_identity()), Arc::new(DoorTokenProvider { bind_scoped: vec!["door-native-bind".to_string()], }), ) .await; let token_client = Arc::new( alksocks::consumer::TokenClient::wrap(topo.consumer.as_ref()) .await .expect("the door's token surface wraps"), ); let mut session = BindSession::bind_with_token( token_client.as_ref(), ("bind.token", 0u16).to_target_addr().unwrap(), b"door-native-bind".to_vec(), ) .await .expect("the token-carrying BIND session binds"); let TargetAddr::Ip(sa) = session.reply1().expect("reply#1").bind_addr.clone() else { panic!("an IP BND.ADDR") }; assert!(sa.ip().is_loopback() && sa.port() != 0, "a real listener"); let mut app = timeout(BOUND, TcpStream::connect(sa)) .await .expect("dial") .expect("app dials"); let mut plane = timeout(BOUND, session.await_inbound()) .await .expect("reply#2") .expect("plane") .expect("accept resolved"); plane.write_all(b"token-bind").await.unwrap(); plane.flush().await.unwrap(); let mut buf = [0u8; 10]; timeout(BOUND, app.read_exact(&mut buf)) .await .expect("read") .expect("read"); assert_eq!(&buf, b"token-bind"); session.close().await; } /// The ASSOCIATE token twin driven directly (review 003 D-1: the /// `associate_with_token` constructor is the session family's /// completeness twin: no door calls it, so it needs its own probe; /// the same shape as `bind_session_with_token_round_trip`): the /// open rides the payload `auth_token` (CF-005 precedence (1)), /// the ASSOCIATE's DST.ADDR is the unspecified address (the /// vanilla shape), the reply's BND.ADDR is the in-band sentinel /// (ADR-003), and a datagram round-trips through the echo egress. #[tokio::test] async fn associate_session_with_token_round_trip() { let topo = harness::wire_consumer_side_with_provider( alksocks::tcp_dial(), Some(harness::echo_egress()), None, Some(consumer_identity()), Arc::new(DoorTokenProvider { bind_scoped: vec![], }), ) .await; let token_client = Arc::new( alksocks::consumer::TokenClient::wrap(topo.consumer.as_ref()) .await .expect("the door's token surface wraps"), ); let mut session = alksocks::consumer::AssociateSession::associate_with_token( token_client.as_ref(), b"door-native-associate".to_vec(), ) .await .expect("the token-carrying ASSOCIATE session associates"); let relay = session .relay_addr() .expect("the reply carried BND.ADDR") .into_string_and_port(); assert_eq!(relay.0, "127.0.0.1", "the in-band sentinel BND.ADDR"); assert_eq!(relay.1, 0); let target = TargetAddr::Ip("10.0.0.9:53".parse().unwrap()); session .send_datagram(&target, b"token-assoc") .await .expect("send datagram"); let (_, payload) = timeout(BOUND, session.recv_datagram()) .await .expect("recv within timeout") .expect("a datagram (not eof)") .expect("a datagram"); assert_eq!(&payload[..], b"token-assoc"); assert!( topo.consumer.manager().has_channel(session.channel_id()), "the adopted channel entry exists while associated" ); session.shutdown().await.expect("shutdown"); } /// The `serve()` wrapper (review 003 N-8: the other tests all drove /// `serve_on`; the bind-then-serve wrapper round-trips too; the NoAuth raw-pipe /// shape, the port probed up front so the wrapper's own bind lands on /// it). #[tokio::test] async fn front_door_serve_wrapper_round_trips() { let echo = echo_tcp_server().await; let dial = alksocks::tcp_dial(); let topo = harness::wire_consumer_side_with(dial, None, None, None).await; // A free loopback port, probed up front (the door binds exactly // it, the wrapper's own bind decision). The connect retries: // the wrapper's bind races the probe's teardown (the OS may hold // the freed port briefly). let mut addr = None; for _ in 0..50 { let probe = TcpListener::bind("127.0.0.1:0").await.expect("probe bind"); let candidate = probe.local_addr().expect("probe addr"); drop(probe); tokio::time::sleep(Duration::from_millis(10)).await; if TcpStream::connect(candidate).await.is_err() { addr = Some(candidate); break; } } let addr = addr.expect("a genuinely free port"); let door = FrontDoor::builder() .bind(addr) .build() .expect("the door builds"); let serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve(client).await.expect("the door serves") } }); let mut vanilla = loop { if let Ok(v) = VanillaClient::try_dial(addr).await { break v; } tokio::time::sleep(Duration::from_millis(20)).await; }; let method = timeout(BOUND, vanilla.greet_no_auth()) .await .expect("greeting"); assert_eq!(method, consts::SOCKS5_AUTH_METHOD_NONE); let code = timeout(BOUND, vanilla.connect_v4(echo)) .await .expect("CONNECT"); assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); vanilla.stream.write_all(b"served").await.unwrap(); let mut buf = [0u8; 6]; timeout(BOUND, vanilla.stream.read_exact(&mut buf)) .await .expect("echo") .expect("read"); assert_eq!(&buf, b"served"); serve.abort(); } /// The credential table's observability accessors (review 003 N-8: /// `len`/`is_empty` had no caller; one probe pins both). #[test] fn credential_table_len_and_is_empty_accessors() { let empty = alksocks::CredentialTable::new(vec![]); assert!(empty.is_empty()); assert_eq!(empty.len(), 0); let one = alksocks::CredentialTable::new(vec![alksocks::DoorCredential { username: "u".to_string(), password: "p".to_string(), identity: harness::consumer_identity(), }]); assert!(!one.is_empty()); assert_eq!(one.len(), 1); }