alksocks 0.3.2

//! 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<String>,
}

impl IdentityProvider for DoorTokenProvider {
    fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
        None
    }
    fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
        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<Self> {
        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);
}