alksocks/tests/local-backend-producer.rs

//! The producer-side local backend's integration tests (feature
//! `local`; tasks/implementation/local-backend-producer.md): a vanilla
//! RFC 1928 client against the locally-served producer — the whole
//! stack on real sockets (`ProducerDoor` → `serve_socks5_over` →
//! `local::tcp_dial()` / `local::bind_accept()`), BIND through the
//! local backend (the faithful-RFC variant — real reply#1/reply#2
//! addresses), the idle-first-byte timeout, the session cap,
//! per-connection failure isolation, and the `bind_policy(false)`
//! gate's RFC-faithful BIND refusal (0x02).
#![cfg(feature = "local")]
#![allow(dead_code)]

use std::time::Duration;

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::time::timeout;

use alksocks::local::ProducerDoor;
use alksocks::producer::Socks5Producer;
use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};

const BOUND: Duration = Duration::from_secs(10);
const IDLE: Duration = Duration::from_millis(300);

/// Spawn a real echo TCP server on an ephemeral loopback port
/// (CONNECT's target — the dial seam dials it for real).
async fn echo_tcp_server() -> std::net::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 producer under test: local socket-backed seams all three
/// (the whole stack on real sockets), BIND policy as given.
fn local_producer(bind_policy: Option<bool>) -> Socks5Producer {
    let builder = Socks5Producer::builder()
        .dial(alksocks::local::tcp_dial())
        .egress(alksocks::local::udp_egress())
        .accept(alksocks::local::bind_accept(std::net::IpAddr::from([
            127, 0, 0, 1,
        ])));
    let builder = match bind_policy {
        Some(can) => builder.bind_policy(can),
        None => builder,
    };
    builder.build()
}

/// Serve the producer behind a real listener (ephemeral loopback
/// port); returns the vanilla client's door address and the guard.
async fn serve(
    producer: Socks5Producer,
    idle: Duration,
    cap: usize,
) -> (std::net::SocketAddr, alksocks::local::ProducerDoorGuard) {
    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
    let addr = listener.local_addr().expect("local_addr");
    let door = ProducerDoor::builder()
        .producer(producer)
        .idle_first_byte(idle)
        .max_sessions(cap)
        .build()
        .expect("the door builds");
    let guard = door.serve_on(listener).await;
    (addr, guard)
}

/// A vanilla client: connect to the door, run the full NoAuth
/// greeting + CONNECT request, read the reply, and echo round-trip
/// through the dialed target. Returns the reply code.
async fn vanilla_connect(
    door: std::net::SocketAddr,
    target: std::net::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];
    req.extend_from_slice(&[127, 0, 0, 1]);
    req.extend_from_slice(&target.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)
}

#[tokio::test]
async fn vanilla_client_connect_round_trips_through_local_seams() {
    let target = echo_tcp_server().await;
    let (door, _guard) = serve(local_producer(None), IDLE, 16).await;

    let (mut stream, code, _addr) = timeout(BOUND, vanilla_connect(door, target))
        .await
        .expect("round-trip");
    assert_eq!(
        code, 0x00,
        "the CONNECT resolved through the real dial seam"
    );

    // The data plane is raw pass-through to the real echo server.
    stream.write_all(b"ping").await.expect("probe");
    let mut buf = [0u8; 4];
    timeout(BOUND, stream.read_exact(&mut buf))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&buf, b"ping");

    drop(stream);
    _guard.close().await;
}

/// BIND through the local backend — the faithful-RFC variant: a real
/// inbound peer connects to the reply#1-advertised listener, and
/// reply#2 carries the real peer address.
#[tokio::test]
async fn vanilla_client_bind_round_trips_with_real_addresses() {
    let (door, _guard) = serve(local_producer(None), IDLE, 16).await;

    let mut client = TcpStream::connect(door).await.expect("door connect");
    client
        .write_all(&[0x05, 0x01, 0x00])
        .await
        .expect("greeting");
    let mut chosen = [0u8; 2];
    client
        .read_exact(&mut chosen)
        .await
        .expect("method selection");
    assert_eq!(chosen, [0x05, 0x00]);

    // BIND for the ephemeral port (ATYP v4, port 0).
    client
        .write_all(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
        .await
        .expect("bind request");

    // Reply#1: success + the real bound listener address.
    let mut head = [0u8; 4];
    client.read_exact(&mut head).await.expect("reply#1 head");
    assert_eq!(head[1], 0x00, "BIND reply#1 succeeded");
    let mut bnd = [0u8; 6];
    client.read_exact(&mut bnd).await.expect("reply#1 BND.ADDR");
    assert_eq!(bnd[0], 127, "the reply carries the real loopback bind");
    let bound_port = u16::from_be_bytes([bnd[4], bnd[5]]);
    assert!(bound_port != 0, "a real listener was bound");

    // The inbound peer connects to the advertised address.
    let advertised = std::net::SocketAddr::from(([127, 0, 0, 1], bound_port));
    let mut peer = TcpStream::connect(advertised).await.expect("peer connects");
    peer.write_all(b"hello-bind").await.expect("peer writes");

    // Reply#2: success + the peer's real address.
    let mut head2 = [0u8; 4];
    client.read_exact(&mut head2).await.expect("reply#2 head");
    assert_eq!(head2[1], 0x00, "BIND reply#2 succeeded");
    let mut peer_addr = [0u8; 6];
    client
        .read_exact(&mut peer_addr)
        .await
        .expect("reply#2 DST.ADDR");
    assert_eq!(peer_addr[0], 127, "reply#2 carries the real peer address");
    let peer_port = u16::from_be_bytes([peer_addr[4], peer_addr[5]]);
    assert!(peer_port != 0, "the peer's ephemeral port is real");

    // The data plane is raw pass-through to the accepted peer.
    let mut buf = [0u8; 10];
    timeout(BOUND, client.read_exact(&mut buf))
        .await
        .expect("data plane")
        .expect("read");
    assert_eq!(&buf, b"hello-bind");

    client.write_all(b"pong").await.expect("client writes");
    let mut echo = [0u8; 4];
    timeout(BOUND, peer.read_exact(&mut echo))
        .await
        .expect("reverse data plane")
        .expect("read");
    assert_eq!(&echo, b"pong");

    drop(peer);
    drop(client);
    _guard.close().await;
}

/// The idle-first-byte defense (ADR-014 knob 1): an accepted
/// connection that sends nothing within the window is closed
/// door-side — EOF arrives, no greeting ever answered.
#[tokio::test]
async fn idle_first_byte_timeout_closes_the_accepted_connection() {
    let (door, _guard) = serve(local_producer(None), IDLE, 16).await;
    let mut stream = TcpStream::connect(door).await.expect("door connect");

    let mut buf = [0u8; 2];
    match timeout(IDLE * 4, stream.read(&mut buf)).await {
        Err(_) => panic!("the idle bound must fire"),
        Ok(Ok(0)) | Ok(Err(_)) => {}
        Ok(Ok(n)) => panic!("EOF expected after the idle bound — got {n} bytes"),
    }
    _guard.close().await;
}

/// The session cap (ADR-014 knob 2): an accept beyond the bound is
/// refused door-side (immediate EOF), while the sessions under the
/// bound keep serving.
#[tokio::test]
async fn session_cap_refuses_beyond_the_bound_while_live_sessions_serve() {
    let target = echo_tcp_server().await;
    let (door, _guard) = serve(local_producer(None), IDLE, 1).await;

    // One live session fills the cap (holding the reply open).
    let (mut held, code, _) = timeout(BOUND, vanilla_connect(door, target))
        .await
        .expect("held session");
    assert_eq!(code, 0x00, "the cap's first session serves");

    // The second accept is beyond the bound: closed immediately.
    let mut refused = TcpStream::connect(door).await.expect("door connect");
    let mut buf = [0u8; 2];
    match timeout(BOUND, refused.read(&mut buf)).await {
        Err(_) => panic!("the capped accept must end"),
        Ok(Ok(0)) | Ok(Err(_)) => {}
        Ok(Ok(n)) => panic!("the capped accept is refused — got {n} bytes"),
    }

    // The held session still serves (isolation: the refusal did not
    // touch it).
    held.write_all(b"still-alive").await.expect("held writes");
    let mut echo = [0u8; 11];
    timeout(BOUND, held.read_exact(&mut echo))
        .await
        .expect("held echoes")
        .expect("read");
    assert_eq!(&echo, b"still-alive");

    drop(held);
    _guard.close().await;
}

/// Per-connection failure isolation: a connection speaking protocol
/// garbage ends loud door-side while the loop continues — the next
/// connection on the same door serves normally.
#[tokio::test]
async fn per_connection_failure_is_isolated_and_the_loop_continues() {
    let target = echo_tcp_server().await;
    let (door, _guard) = serve(local_producer(None), IDLE, 16).await;

    // Garbage: a valid greeting, then a malformed command head.
    let mut bad = TcpStream::connect(door).await.expect("door connect");
    bad.write_all(&[0x05, 0x01, 0x00]).await.expect("greeting");
    let mut chosen = [0u8; 2];
    bad.read_exact(&mut chosen).await.expect("method selection");
    bad.write_all(&[0x05, 0x01, 0x00, 0xFF])
        .await
        .expect("garbage head");
    drop(bad);
    tokio::time::sleep(Duration::from_millis(50)).await;

    // The loop continues: a well-formed client serves normally.
    let (mut stream, code, _) = timeout(BOUND, vanilla_connect(door, target))
        .await
        .expect("next session");
    assert_eq!(code, 0x00, "the loop survived the failed connection");

    stream.write_all(b"ping").await.expect("probe");
    let mut buf = [0u8; 4];
    timeout(BOUND, stream.read_exact(&mut buf))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&buf, b"ping");

    drop(stream);
    _guard.close().await;
}

/// The `bind_policy(false)` gate is honored on this path (ADR-006's
/// posture): BIND refuses RFC-faithfully with 0x02 — the command is
/// known, the caller is not allowed — while CONNECT still serves.
#[tokio::test]
async fn bind_policy_false_refuses_bind_rfc_faithfully_but_connect_serves() {
    let target = echo_tcp_server().await;
    let (door, _guard) = serve(local_producer(Some(false)), IDLE, 16).await;

    // BIND → 0x02 (ConnectionNotAllowed), through the producer's own
    // in-stream gate (no open-op identity here — the builder's
    // bind_policy is the whole gate on this path).
    let mut client = TcpStream::connect(door).await.expect("door connect");
    client
        .write_all(&[0x05, 0x01, 0x00])
        .await
        .expect("greeting");
    let mut chosen = [0u8; 2];
    client
        .read_exact(&mut chosen)
        .await
        .expect("method selection");
    client
        .write_all(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
        .await
        .expect("bind request");
    let mut head = [0u8; 4];
    client.read_exact(&mut head).await.expect("reply head");
    assert_eq!(head[1], 0x02, "bind_policy(false) → ConnectionNotAllowed");
    drop(client);

    // CONNECT still serves through the same closed-gate producer.
    let (stream, code, _) = timeout(BOUND, vanilla_connect(door, target))
        .await
        .expect("connect session");
    assert_eq!(code, 0x00, "the gate closes BIND alone");

    drop(stream);
    _guard.close().await;
}

/// The default BIND posture holds too: with the accept seam wired and
/// no explicit policy, BIND is permitted (the backward-compat
/// reading) and serves through the real listener seam.
#[tokio::test]
async fn default_bind_posture_serves_through_the_accept_seam() {
    let (door, _guard) = serve(local_producer(None), IDLE, 16).await;

    let mut client = TcpStream::connect(door).await.expect("door connect");
    client
        .write_all(&[0x05, 0x01, 0x00])
        .await
        .expect("greeting");
    let mut chosen = [0u8; 2];
    client
        .read_exact(&mut chosen)
        .await
        .expect("method selection");
    client
        .write_all(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
        .await
        .expect("bind request");
    let mut head = [0u8; 4];
    client.read_exact(&mut head).await.expect("reply#1 head");
    assert_eq!(head[1], 0x00, "the default posture permits BIND");
    drop(client);
    _guard.close().await;
}

/// UDP ASSOCIATE through the local backend: the sentinel reply rides
/// the door, the datagram stage relays through the real UDP egress
/// (a real loopback UDP echo target).
#[tokio::test]
async fn vanilla_client_udp_associate_relays_through_the_local_egress() {
    let sock = tokio::net::UdpSocket::bind("127.0.0.1:0")
        .await
        .expect("bind");
    let udp_target = sock.local_addr().expect("local_addr");
    tokio::spawn(async move {
        let mut buf = [0u8; 65535];
        loop {
            let Ok((n, from)) = sock.recv_from(&mut buf).await else {
                return;
            };
            if sock.send_to(&buf[..n], from).await.is_err() {
                return;
            }
        }
    });

    let (door, _guard) = serve(local_producer(None), IDLE, 16).await;

    let mut client = TcpStream::connect(door).await.expect("door connect");
    client
        .write_all(&[0x05, 0x01, 0x00])
        .await
        .expect("greeting");
    let mut chosen = [0u8; 2];
    client
        .read_exact(&mut chosen)
        .await
        .expect("method selection");
    client
        .write_all(&[0x05, 0x03, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
        .await
        .expect("associate request");
    let mut head = [0u8; 4];
    client.read_exact(&mut head).await.expect("reply head");
    assert_eq!(
        head[1], 0x00,
        "the associate resolved through the real egress seam"
    );
    let mut bnd = [0u8; 6];
    client.read_exact(&mut bnd).await.expect("BND.ADDR");
    assert_eq!(&bnd, &[127, 0, 0, 1, 0, 0], "the sentinel BND.ADDR");

    // One datagram frame (the in-band wire shape) echoes through the
    // real UDP egress.
    let dst = TargetAddr::Ip(udp_target);
    let frame = alksocks::wire::frame(&dst, b"udp-probe").expect("frame");
    client.write_all(&frame).await.expect("datagram frame");

    let mut reply = vec![0u8; frame.len()];
    timeout(BOUND, client.read_exact(&mut reply))
        .await
        .expect("echo reply")
        .expect("read");
    // The reply's inner carries the *sender's* address (the echo's
    // source is the egress socket, not the target) — decode and match
    // the payload only.
    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"udp-probe", "the datagram echoed");

    drop(client);
    _guard.close().await;
}

/// The door's address target form: a domain-form CONNECT target
/// resolves at the seam (the ssh `-D` semantic — ADR-012) — the echo
/// server is reachable as `localhost`.
#[tokio::test]
async fn domain_form_targets_resolve_at_the_seam() {
    let target = echo_tcp_server().await;
    let (door, _guard) = serve(local_producer(None), IDLE, 16).await;

    let mut client = TcpStream::connect(door).await.expect("door connect");
    client
        .write_all(&[0x05, 0x01, 0x00])
        .await
        .expect("greeting");
    let mut chosen = [0u8; 2];
    client
        .read_exact(&mut chosen)
        .await
        .expect("method selection");

    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(&target.port().to_be_bytes());
    client.write_all(&req).await.expect("request");
    client.flush().await.expect("flush");

    let mut head = [0u8; 4];
    client.read_exact(&mut head).await.expect("reply head");
    assert_eq!(head[1], 0x00, "the domain resolved at the seam");
    let mut addr = [0u8; 6];
    client.read_exact(&mut addr).await.expect("BND.ADDR");

    client.write_all(b"dom").await.expect("probe");
    let mut buf = [0u8; 3];
    timeout(BOUND, client.read_exact(&mut buf))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&buf, b"dom");

    drop(client);
    _guard.close().await;
}

/// The `ToTargetAddr` import stays exercised (the compile-time shape
/// the domain test rides).
#[allow(dead_code)]
fn _target_addr_shape() {
    let _: TargetAddr = std::net::SocketAddr::from(([127, 0, 0, 1], 1))
        .to_target_addr()
        .expect("encode");
}