alksocks 0.3.2

//! The consumer's TCP front door (feature `local`, ADR-005/ADR-014;
//! consumer.md §The Front Doors): the ssh `-D` local exposure, a real
//! `socks5://127.0.0.1:port` endpoint vanilla clients (curl, browsers,
//! tun2proxy) point at. The door is the vanilla↔ACL boundary; the RFC
//! codec lives only here (ADR-003's "codec at front doors only").
//!
//! Two door variants (ADR-005), selected by constructor config:
//!
//! - **The NoAuth door** (POC #1's raw-pipe shape): per accepted
//!   connection, a literal raw pipe onto a channel `BiStream`
//!   (`copy_bidirectional`, **zero protocol translation**). The
//!   producer runs the whole RFC conversation through the pipe, and
//!   the door adds nothing. The loopback-only / wrapper-aware shape.
//! - **The cred-bearing door** (RFC 1929): the door terminates the
//!   local RFC conversation itself, using the forked
//!   `accept_password_auth` shape (`start` →
//!   `negotiate_auth(&[PasswordAuthentication])` →
//!   `read_username_password` → check → `accept`/`reject`), then
//!   maps the accepted credential through the **static config table**
//!   (`user:pass →` channel identity, ADR-005) and **re-originates
//!   the SOCKS conversation in-band** via the token-carrying consumer
//!   sessions (`Socks5ChannelSession::connect_with_token` /
//!   `BindSession::bind_with_token`), re-encoding the in-band replies
//!   to the vanilla client and piping the post-auth data-plane bytes
//!   raw. **The door never pipes raw post-auth**: the vanilla
//!   client's command bytes were consumed by the door's own RFC 1929
//!   half; the in-band path re-speaks them from the parsed target.
//!   The SOCKS password never crosses the channel as a SOCKS
//!   password; the channel identity never reaches the vanilla client.
//!
//! **The identity mechanism** (the CF-005 seam; decided at
//! implementation, recorded in the task Notes): the credential→identity
//! mapping yields an alkcall [`Identity`] (scopes included); the
//! per-credential identity rides the open-op call's `auth_token`
//! payload (CF-005 precedence (1): payload `auth_token` →
//! `identity_provider` resolution, alkcall `dispatch::
//! resolve_identity`, ADR-017 §7), never a per-credential
//! `ChannelClient`. The token is the mapped identity's id, the
//! resolution seam's vocabulary; the password itself never rides it.
//! The producer resolves it through its `ServingConfig
//! .identity_provider` and the ACL gate reads the resolved scopes
//! exactly as for any per-call opener identity (CF-005/CF-006). The
//! door checks credentials itself, with no forked `AuthMethod` impl
//! (ADR-005).
//!
//! **The accept path owns the idle defense** (ADR-014): an accepted
//! connection that sends no first byte within the configured window
//! is closed door-side (an accepted socket is not a channel, so the
//! cap
//! regime does not bound it); plus the max-concurrent-sessions cap
//! (an accept beyond the bound is refused door-side, no channel
//! opened). The two v1 knobs are constructor config.
//!
//! **The BIND door**: the same listener shape; a vanilla BIND
//! conversation rides [`BindSession`] (reply#1's advertised address
//! passes through to the vanilla client unchanged, since the
//! conversation is the client's own; the door adds nothing,
//! consumer.md §The BIND door's reply#2 fidelity). The BIND door rides
//! the same door variant split (NoAuth raw-pipe vs cred-bearing
//! re-origination).
//!
//! Wiring (consumer.md §Wiring): `FrontDoor::builder()
//! .credentials(cred_table).bind("127.0.0.1:1080").build()` →
//! `serve(channel_client)`; the NoAuth door omits `.credentials`.
//! `serve` runs the accept loop; per-connection failures are logged +
//! isolated (the loop continues, per the alktunnels per-accept
//! isolation posture).
//!
//! **The UDP door** ([`UdpFrontDoor`], the POC #2 test-8 shape): the
//! ASSOCIATE front door, a real UDP socket (the relay socket; the
//! vanilla client `sendto()`s it) bridged onto one in-band
//! association. The RFC UDP-header codec lives ONLY here (ADR-003's
//! codec-at-doors-only): inbound datagrams decode through
//! `parse_udp_request` and re-send in-band as `{addr, data}` frames
//! (`send_datagram`); replies encode through `new_udp_header` and go
//! back to the client's source address. **FRAG ≠ 0 → refuse** (the
//! datagram is dropped, the v1 posture; fragmentation cannot be
//! expressed in-band, and the association stays healthy). The
//! datagram pump is the two-pump shape (one pump per direction,
//! joined, per the ADR-050 contract). Zero payload translation. The
//! association's lifetime is the channel's (ADR-003): the association
//! ends when the channel ends; the door's relay socket closes with
//! it. The door-side idle bound (ADR-014): no datagrams from the
//! client within the configured window closes the association (a
//! relay socket is not a channel; the cap regime does not bound
//! it). The reply's sentinel BND.ADDR is received and never used,
//! since the handle is the channel; the real relay address is the door's
//! own bound socket, which the vanilla client learns from the TCP
//! door's ASSOCIATE reply (the assembly layer wires the two; the UDP
//! door serves the datagram plane alone). The bind address is
//! constructor config (consumer.md: "the UDP door's bind address is
//! assembly-layer configuration"). Surface shape (recorded per the
//! task spec): a **sibling** `UdpFrontDoor` with its own builder; the
//! UDP door shares no accept loop with the TCP door (UDP has no
//! connections; one relay socket = one association), and the shared
//! shape would have forced a mode split on every knob.

use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;

use alkcall::channels::client::ChannelClient;
use alkcall::core::auth::Identity;
use tokio::io::AsyncWriteExt as _;
use tokio::net::{TcpListener, TcpStream, UdpSocket};
use tokio::sync::Semaphore;

use crate::consumer::{AssociateSession, TakenInner, TokenClient};
use crate::error::SocksSessionError;
use crate::rfc1928::consts;
use crate::rfc1928::server::{new_reply, states, Socks5ServerProtocol, SocksServerError};
use crate::rfc1928::util::target_addr::TargetAddr;
use crate::rfc1928::{ReplyError, Socks5Command, SocksError};
use crate::wire::{DatagramReader, WireError};
use bytes::Bytes;

/// One static credential-table entry (ADR-005): `user:pass →` channel
/// identity. The table is the door's credential→identity mapping:
/// static, declarative, assembly-layer.
#[derive(Clone, Debug)]
pub struct DoorCredential {
    /// The RFC 1929 username the vanilla client presents.
    pub username: String,
    /// The RFC 1929 password the vanilla client presents.
    pub password: String,
    /// The channel identity the accepted credential maps to (scopes
    /// included; the open gate and the BIND command gate read them
    /// producer-side). The identity's `id` doubles as the open-op
    /// `auth_token` (the resolution seam's vocabulary: the password
    /// never rides it).
    pub identity: Identity,
}

/// The credential→identity static config table (ADR-005): exact
/// `username`/`password` lookup. A miss rejects the RFC 1929
/// conversation door-side (no channel opened, so the ACL boundary is
/// held; the vanilla client got the RFC 1929 rejection).
#[derive(Clone, Debug, Default)]
pub struct CredentialTable {
    entries: HashMap<(String, String), Identity>,
}

impl CredentialTable {
    /// Build the table from [`DoorCredential`] entries.
    pub fn new(entries: impl IntoIterator<Item = DoorCredential>) -> Self {
        Self {
            entries: entries
                .into_iter()
                .map(|c| ((c.username, c.password), c.identity))
                .collect(),
        }
    }

    /// The exact-match lookup: `(username, password) → Option<identity>`.
    pub fn lookup(&self, username: &str, password: &str) -> Option<&Identity> {
        self.entries
            .get(&(username.to_string(), password.to_string()))
    }

    /// The number of entries (observability).
    pub fn len(&self) -> usize {
        self.entries.len()
    }

    /// Whether the table is empty (an empty cred-bearing table
    /// rejects every conversation door-side).
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }
}

/// Why one accepted connection's serving task ended (the door's
/// per-connection teardown diagnostic, logged + isolated).
#[derive(Debug)]
enum DoorExit {
    /// The vanilla client went away (clean end, both variants).
    ClientGone,
    /// The data plane (local pipe ↔ in-band session) ended.
    PlaneEnded,
    /// The local RFC half failed (a rejected credential, a malformed
    /// conversation, protocol garbage); the vanilla client's fault.
    LocalRfc(SocksError),
    /// The in-band open failed (the producer refused the mapped
    /// identity, the target dial refused, the channel open errored).
    Open(String),
    /// A plain I/O failure on the local socket.
    Io(std::io::Error),
}

impl std::fmt::Display for DoorExit {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            DoorExit::ClientGone => f.write_str("client gone"),
            DoorExit::PlaneEnded => f.write_str("data plane ended"),
            DoorExit::LocalRfc(err) => write!(f, "local rfc: {err}"),
            DoorExit::Open(msg) => write!(f, "open failed: {msg}"),
            DoorExit::Io(err) => write!(f, "io: {err}"),
        }
    }
}

impl From<SocksServerError> for DoorExit {
    fn from(err: SocksServerError) -> Self {
        DoorExit::LocalRfc(SocksError::ServerError(err))
    }
}

/// The TCP front door (feature `local`): the vanilla↔ACL boundary.
/// Variant selection is constructor config (ADR-005): credentials
/// configured = the cred-bearing door (RFC 1929 terminates locally,
/// re-origination in-band); credentials absent = the NoAuth door (the
/// literal raw pipe). The accept path's defense knobs (ADR-014) are
/// constructor config: the idle-first-byte window and the
/// max-concurrent-sessions cap.
pub struct FrontDoor {
    bind_addr: SocketAddr,
    credentials: Option<CredentialTable>,
    idle_first_byte: Duration,
    max_sessions: usize,
}

impl FrontDoor {
    /// The builder (consumer.md §Wiring).
    pub fn builder() -> FrontDoorBuilder {
        FrontDoorBuilder::default()
    }
}

/// The builder for [`FrontDoor`] (consumer.md §Wiring: `builder()
/// .credentials(cred_table).bind("127.0.0.1:1080").build()`). The
/// NoAuth door omits `.credentials`.
#[derive(Default)]
pub struct FrontDoorBuilder {
    bind_addr: Option<SocketAddr>,
    credentials: Option<CredentialTable>,
    idle_first_byte: Option<Duration>,
    max_sessions: Option<usize>,
}

impl FrontDoorBuilder {
    /// The credential→identity static config table (ADR-005).
    /// Configuring it selects the cred-bearing door (RFC 1929);
    /// omitting it selects the NoAuth door (the raw-pipe shape).
    pub fn credentials(mut self, table: CredentialTable) -> Self {
        self.credentials = Some(table);
        self
    }

    /// The local bind address, the door's explicit bind decision
    /// (the never-binds posture: the door is the assembly layer's
    /// spelled-out choice; loopback is the documented posture).
    /// Required: `.build()` is a `Result`; omitting it is the
    /// builder's typed construction error (convention 2: no panic
    /// path).
    pub fn bind(mut self, addr: impl Into<SocketAddr>) -> Self {
        self.bind_addr = Some(addr.into());
        self
    }

    /// The idle-first-byte window (ADR-014's accept-path defense
    /// knob 1): an accepted connection that sends no first byte
    /// within this window is closed door-side. The window covers the
    /// first byte of the local conversation: the NoAuth door's
    /// greeting bytes (read producer-side through the pipe) and the
    /// cred-bearing door's greeting (read by the door's own RFC 1929
    /// half).
    pub fn idle_first_byte(mut self, window: Duration) -> Self {
        self.idle_first_byte = Some(window);
        self
    }

    /// The max-concurrent-*connections* cap (ADR-014's accept-path
    /// defense knob 2; review 003 N-3's naming, for the mechanism
    /// bounds accepted connections, including ones past the idle gate
    /// that are merely slow readers, not active RFC sessions): an
    /// accept beyond the bound is refused door-side; the socket is
    /// closed immediately, with no channel opened.
    pub fn max_sessions(mut self, cap: usize) -> Self {
        self.max_sessions = Some(cap);
        self
    }

    pub fn build(self) -> Result<FrontDoor, DoorBuildError> {
        let bind_addr = self.bind_addr.ok_or(DoorBuildError::MissingBindAddr)?;
        Ok(FrontDoor {
            bind_addr,
            credentials: self.credentials,
            idle_first_byte: self.idle_first_byte.unwrap_or(Duration::from_secs(30)),
            max_sessions: self.max_sessions.unwrap_or(256),
        })
    }
}

/// Why a door builder refused to construct (the builder's typed
/// construction error, i.e. a missing required field; convention 2:
/// no
/// panic path on an omitted `.bind`).
#[derive(Debug, thiserror::Error)]
pub enum DoorBuildError {
    #[error("the door requires a bind address (.bind); the bind decision is never implied")]
    MissingBindAddr,
}

impl FrontDoor {
    /// Bind the configured address and run the accept loop
    /// (consumer.md §Wiring: `door.serve(channel_client)`). Per-
    /// connection failures are logged + isolated, and the loop
    /// continues (the alktunnels per-accept isolation posture). Runs
    /// until the listener fails fatally; the caller owns the task's
    /// lifetime.
    pub async fn serve(self, client: Arc<ChannelClient>) -> std::io::Result<()> {
        let listener = TcpListener::bind(self.bind_addr).await?;
        tracing::info!(bind = %self.bind_addr, "front door: listening");
        self.serve_on(listener, client).await
    }

    /// Run the accept loop over an existing listener (the
    /// test/assembly seam, the same accept loop without the bind).
    pub async fn serve_on(
        self,
        listener: TcpListener,
        client: Arc<ChannelClient>,
    ) -> std::io::Result<()> {
        let credentials = self.credentials;
        let idle = self.idle_first_byte;
        let live = Arc::new(Semaphore::new(self.max_sessions));

        // The cred-bearing door's channel surface: the client's call
        // connection is taken once for the door's lifetime (one
        // transport, one call connection, many per-credential
        // identities riding the open-op `auth_token` payloads). The
        // NoAuth door keeps the client's own calling surface
        // (`open_channel`; the raw-pipe shape never carries a
        // token).
        let token_client = match credentials {
            Some(_) => Some(Arc::new(
                TokenClient::wrap(&client)
                    .await
                    .map_err(|err| std::io::Error::other(err.to_string()))?,
            )),
            None => None,
        };

        loop {
            let (stream, peer) = match listener.accept().await {
                Ok(accepted) => accepted,
                Err(err) => {
                    tracing::warn!(error = %err, "front door: accept failed");
                    continue;
                }
            };

            // The session cap (ADR-014 knob 2): refuse beyond the
            // bound; `try_acquire` keeps the accept loop free (a
            // permit returns when a serving task ends and drops its
            // guard).
            let Ok(permit) = live.clone().try_acquire_owned() else {
                tracing::warn!(peer = %peer, "front door: session cap reached; refusing");
                drop(stream);
                continue;
            };

            let client = Arc::clone(&client);
            let per_accept_credentials = credentials.clone();
            let per_accept_tokens = token_client.clone();
            tokio::spawn(async move {
                let _permit = permit;
                let exit = serve_connection(
                    stream,
                    peer,
                    &client,
                    per_accept_credentials.as_ref(),
                    per_accept_tokens.as_ref(),
                    idle,
                )
                .await;
                tracing::debug!(peer = %peer, "front door: connection ended: {exit}");
            });
        }
    }
}

/// Serve one accepted connection through the door variant split: the
/// idle-first-byte gate first (ADR-014 knob 1, since the door's
/// accept
/// path owns the defense, and an accepted socket is not a channel),
/// then the variant.
async fn serve_connection(
    stream: TcpStream,
    peer: SocketAddr,
    client: &Arc<ChannelClient>,
    credentials: Option<&CredentialTable>,
    token_client: Option<&Arc<TokenClient>>,
    idle: Duration,
) -> DoorExit {
    // The idle-first-byte defense: an accepted connection that sends
    // no first byte within the window is closed door-side (ADR-014's
    // slowloris sliver). The peek waits for readability without
    // consuming, and the variant's own protocol read picks the byte
    // up.
    let mut first = [0u8; 1];
    match tokio::time::timeout(idle, stream.peek(&mut first)).await {
        Ok(Ok(_)) => {}
        Ok(Err(err)) => return DoorExit::Io(err),
        Err(_) => {
            tracing::debug!(peer = %peer, "front door: idle-first-byte timeout");
            return DoorExit::ClientGone;
        }
    }

    match (credentials, token_client) {
        // The NoAuth door: the literal raw pipe (POC #1's shape, with
        // zero protocol translation).
        (None, _) => raw_pipe(stream, client).await,
        // The cred-bearing door: RFC 1929 terminates locally,
        // re-origination rides the token sessions.
        (Some(table), Some(tokens)) => cred_bearing(stream, client, table, tokens).await,
        // Unreachable by construction (`serve_on` builds the pair),
        // so the arm exists to keep the match exhaustive without a
        // panic path (convention 2).
        (Some(_), None) => {
            DoorExit::Open("the cred-bearing door's token surface was not built".to_string())
        }
    }
}

/// The NoAuth door's raw pipe (POC #1's `front_door_raw_pipe_vanilla_
/// client` shape): open a channel, then `copy_bidirectional` the
/// accepted socket onto the channel `BiStream`, so the producer
/// runs the whole RFC conversation through the pipe; the door adds
/// nothing.
async fn raw_pipe(mut stream: TcpStream, client: &Arc<ChannelClient>) -> DoorExit {
    let (_channel_id, send, recv) = match client
        .open_channel(
            crate::params::OP_SOCKS5_OPEN,
            crate::params::Socks5Params::as_json(),
            crate::params::SOCKS5_ALPN,
        )
        .await
    {
        Ok(open) => open,
        Err(err) => return DoorExit::Open(err.to_string()),
    };
    let mut bidi = alkcall::core::types::BiStream::from_joined(recv, send);
    let _ = tokio::io::copy_bidirectional(&mut stream, &mut bidi).await;
    DoorExit::PlaneEnded
}

/// The cred-bearing door (ADR-005): terminate the vanilla RFC 1929
/// conversation's auth locally (the forked surface), map the accepted
/// credential through the static table, re-originate the SOCKS
/// conversation in-band via the token-carrying consumer sessions,
/// re-encode the in-band replies to the vanilla client, and pipe the
/// post-auth data-plane bytes raw.
///
/// The door never pipes raw post-auth: the vanilla client's command
/// bytes were consumed by the door's own RFC half; the in-band path
/// re-speaks them from the parsed target. The password never crosses
/// the channel as a SOCKS password (the in-band handshake is NoAuth,
/// so the mapped identity is the authorization); the channel identity
/// never reaches the vanilla client.
async fn cred_bearing(
    mut stream: TcpStream,
    _client: &Arc<ChannelClient>,
    table: &CredentialTable,
    token_client: &Arc<TokenClient>,
) -> DoorExit {
    // The local RFC 1929 half (the forked surface) + the command read.
    let outcome = local_rfc1929_half(&mut stream, table).await;
    let (cmd, target, identity) = match outcome {
        Ok(triple) => triple,
        Err(exit) => return exit,
    };

    // The in-band re-origination: the per-credential identity rides
    // the open-op call's `auth_token` payload (CF-005 precedence
    // (1)). The token is the mapped identity's id, the resolution
    // seam's vocabulary; the password itself never rides it.
    let token = identity.id.as_bytes().to_vec();

    match cmd {
        Socks5Command::TCPConnect => {
            let session = match crate::consumer::Socks5ChannelSession::connect_with_token(
                token_client,
                target,
                token,
            )
            .await
            {
                Ok(session) => session,
                Err(err) => {
                    // The in-band refusal's mapped code is re-encoded to
                    // the vanilla client (the RFC codec at the door
                    // boundary, ADR-003): a dial refusal is reply#1's
                    // code, never a bare EOF.
                    write_refusal_reply(&mut stream, &err).await;
                    return DoorExit::Open(err.to_string());
                }
            };
            pipe_connect_plane(stream, session).await
        }
        Socks5Command::TCPBind => {
            let session =
                match crate::consumer::BindSession::bind_with_token(token_client, target, token)
                    .await
                {
                    Ok(session) => session,
                    Err(err) => {
                        write_refusal_reply(&mut stream, &err).await;
                        return DoorExit::Open(err.to_string());
                    }
                };
            pipe_bind_plane(stream, session).await
        }
        // UDP ASSOCIATE through the TCP door is the UDP door's shape
        // (the datagram plane's RFC codec boundary, ADR-003), not
        // this one's: refuse RFC-faithfully, so the vanilla client
        // gets
        // the refused reply through the door's own codec (no phantom
        // data plane, no in-band session opened).
        Socks5Command::UDPAssociate => {
            let reply = new_reply(
                &ReplyError::CommandNotSupported,
                SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
            );
            let _ = stream.write_all(&reply).await;
            let _ = stream.flush().await;
            DoorExit::LocalRfc(SocksError::ReplyError(ReplyError::CommandNotSupported))
        }
    }
}

/// The door's local RFC 1929 half (the forked surface, ADR-005's
/// accept_password_auth shape): `negotiate_auth(&[
/// PasswordAuthentication])` → `read_username_password` → check
/// through the static table → `accept`/`reject`, then
/// `read_command` for the re-origination's parsed target. The check
/// rides the forked `CheckResult`'s `bool` impl; the mapped identity
/// is recorded by the same table lookup the check ran (one table,
/// one truth) and re-yielded to the caller.
async fn local_rfc1929_half(
    stream: &mut TcpStream,
    table: &CredentialTable,
) -> Result<(Socks5Command, TargetAddr, Identity), DoorExit> {
    // The check closure is sync (`FnMut(String, String) -> bool` over
    // the forked `CheckResult`'s `bool` impl); the mapped identity is
    // recorded through a sync witness (a `std` mutex, since the
    // closure
    // runs once, synchronously).
    let found: Arc<std::sync::Mutex<Option<Identity>>> = Arc::new(std::sync::Mutex::new(None));
    let found_for_check = Arc::clone(&found);

    let (proto, _ok) = Socks5ServerProtocol::<_, states::Authenticated>::accept_password_auth(
        stream,
        |user, pass| {
            let mapped = table.lookup(&user, &pass).cloned();
            // A poisoned lock would mean a panic mid-check; the
            // `unwrap_or_else(into_inner)` convention keeps the door
            // alive (AGENTS.md convention 2).
            let mut slot = found_for_check.lock().unwrap_or_else(|e| e.into_inner());
            let accepted = mapped.is_some();
            *slot = mapped;
            // The check IS the table lookup: a miss rejects (the
            // forked `accept_password_auth` writes the RFC 1929
            // rejection when the checker returns `false`).
            accepted
        },
    )
    .await
    .map_err(DoorExit::from)?;

    let identity = found
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .take()
        .ok_or_else(|| DoorExit::from(SocksServerError::AuthenticationRejected))?;

    let (_proto, cmd, target) = proto.read_command().await.map_err(DoorExit::from)?;
    Ok((cmd, target, identity))
}

/// Pipe the re-originated CONNECT session: the vanilla reply (the
/// in-band reply's BND.ADDR re-encoded (the RFC codec at the door
/// boundary only, ADR-003), then the data plane ↔ the accepted socket.
/// The door already consumed the vanilla client's greeting + command
/// bytes; the post-auth bytes ride the session's data plane raw from
/// here. The pump shape is the taken inner stream
/// (`Socks5Stream<BiStream>` is itself `AsyncRead + AsyncWrite`, the
/// forked client type's intended use) against the local socket via
/// `copy_bidirectional`; the session stays the channel-teardown owner
/// (the `into_inner` pairing; dropped at the end, so the EOF
/// sentinel
/// reaches the producer's pump).
async fn pipe_connect_plane(
    mut stream: TcpStream,
    session: crate::consumer::Socks5ChannelSession,
) -> DoorExit {
    // The vanilla CONNECT reply: the in-band reply's BND.ADDR
    // re-encoded (the vanilla client is waiting for it; the door's
    // RFC half consumed its local bytes; the in-band conversation is
    // fresh and re-originated). The sentinel address is the
    // RFC-compatibility surface the producer already chose; the door
    // adds nothing.
    let reply = match session.reply_addr() {
        Some(TargetAddr::Ip(sa)) => new_reply(&ReplyError::Succeeded, sa),
        Some(TargetAddr::Domain(_, port)) => new_reply(
            &ReplyError::Succeeded,
            SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), port),
        ),
        None => {
            session.close().await;
            return DoorExit::PlaneEnded;
        }
    };
    if stream.write_all(&reply).await.is_err() {
        session.close().await;
        return DoorExit::ClientGone;
    }
    let _ = stream.flush().await;

    let Ok(TakenInner { session, inner }) = session.into_inner() else {
        return DoorExit::PlaneEnded;
    };
    let _ = tokio::io::copy_bidirectional(&mut stream, &mut { inner }).await;
    let _ = session;
    DoorExit::PlaneEnded
}

/// Pipe the re-originated BIND session (the BIND door, cred-bearing
/// variant): reply#1's advertised address is re-encoded to the
/// vanilla client from the session's captured `bind_addr` (the
/// conversation is the vanilla client's own, and the door's RFC half
/// consumed the local greeting + command bytes, and the in-band
/// session's reply#1 address passes through re-encoded; consumer.md
/// §The BIND door's reply#2 fidelity). The quiet wait rides
/// `await_inbound` (the take-guarded reply#2 reader); reply#2's peer
/// is likewise re-encoded; the data plane passes through raw. The
/// session stays the channel-teardown owner.
async fn pipe_bind_plane(
    mut stream: TcpStream,
    mut session: crate::consumer::BindSession,
) -> DoorExit {
    // Reply#1 (the vanilla form): the in-band reply's BND.ADDR
    // re-encoded, and the advertised address passes through
    // unchanged.
    let bind_addr = session.reply1().map(|reply1| reply1.bind_addr.clone());
    let Some(bind_addr) = bind_addr else {
        session.close().await;
        return DoorExit::PlaneEnded;
    };
    let wire = bind_reply1_wire(&bind_addr);
    if write_vanilla_reply(&mut stream, &wire).await.is_err() {
        session.close().await;
        return DoorExit::ClientGone;
    }

    bind_quiet_and_pipe(stream, session).await
}

/// Reply#1's vanilla wire form, re-encoded from the in-band reply's
/// advertised address: an IP BND.ADDR rides the codec (`new_reply`);
/// a Domain BND.ADDR is the hand-written domain frame (the forked
/// codec's `new_reply` takes a `SocketAddr` only; the domain form
/// has no codec path, so the frame is assembled here; the byte shape
/// is pinned by the unit table, review 003 N-8).
fn bind_reply1_wire(bind_addr: &TargetAddr) -> Vec<u8> {
    match bind_addr {
        TargetAddr::Ip(sa) => new_reply(&ReplyError::Succeeded, *sa),
        TargetAddr::Domain(name, port) => {
            let mut wire = vec![
                consts::SOCKS5_VERSION,
                ReplyError::Succeeded.as_u8(),
                0x00,
                consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME,
            ];
            wire.push(name.len() as u8);
            wire.extend_from_slice(name.as_bytes());
            wire.extend_from_slice(&port.to_be_bytes());
            wire
        }
    }
}

/// The BIND door's quiet wait + reply#2 + the data plane: await the
/// inbound accept through the session (`await_inbound`, the one
/// normative reply#2 reader), re-encode reply#2's peer to the vanilla
/// client, then pipe the pass-through plane raw (the plane owns the
/// in-band stream: `BindDataPlane` impls `AsyncRead + AsyncWrite` for
/// the pass-through to the accepted peer; the session stays the
/// teardown owner, dropped at the end).
async fn bind_quiet_and_pipe(
    mut stream: TcpStream,
    mut session: crate::consumer::BindSession,
) -> DoorExit {
    let plane = match session.await_inbound().await {
        Ok(Some(plane)) => plane,
        Ok(None) => return DoorExit::PlaneEnded,
        // An accept-stage refusal (the mapped reply#2 code): the
        // vanilla client gets that code re-encoded door-side (the RFC
        // codec at the boundary, never a bare EOF), then the
        // connection closes, and the conversation ends where the
        // in-band one did.
        Err(err) => {
            write_refusal_reply(&mut stream, &err).await;
            session.close().await;
            return DoorExit::PlaneEnded;
        }
    };

    // Reply#2 (the vanilla form): the accepted peer re-encoded.
    let reply = match plane.peer {
        TargetAddr::Ip(sa) => new_reply(&ReplyError::Succeeded, sa),
        TargetAddr::Domain(_, port) => new_reply(
            &ReplyError::Succeeded,
            SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), port),
        ),
    };
    if write_vanilla_reply(&mut stream, &reply).await.is_err() {
        session.close().await;
        return DoorExit::ClientGone;
    }

    let _ = tokio::io::copy_bidirectional(&mut stream, &mut { plane }).await;
    session.close().await;
    DoorExit::PlaneEnded
}

/// Write one vanilla reply frame (the door's RFC codec at the
/// boundary, ADR-003): write + flush; a failed write is the client
/// gone.
async fn write_vanilla_reply(stream: &mut TcpStream, reply: &[u8]) -> std::io::Result<()> {
    stream.write_all(reply).await?;
    stream.flush().await
}

/// Re-encode an in-band session error's refusal to the vanilla client
/// (the door's RFC codec at the boundary): an error carrying an RFC
/// reply code gets that code as the command's reply (`Succeeded`-form
/// encoding with the error's code + the placeholder BND.ADDR, since a
/// refused reply carries a zero address per the RFC); any other error
/// shape (open/transport/I-O) stays a bare close (nothing to encode).
/// Best-effort: a failed write means the client is gone either way.
async fn write_refusal_reply(stream: &mut TcpStream, err: &SocksSessionError) {
    let code = match err {
        SocksSessionError::Rfc(SocksError::ReplyError(code)) => *code,
        _ => return,
    };
    let reply = new_reply(&code, SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0));
    let _ = write_vanilla_reply(stream, &reply).await;
}

/// The UDP front door (feature `local`, the POC #2 test-8 shape): the
/// ASSOCIATE door: a real UDP relay socket bridged onto one in-band
/// association. See the module docs for the shape and the codec
/// boundary (ADR-003).
pub struct UdpFrontDoor {
    bind_addr: SocketAddr,
    idle: Duration,
}

/// Why one UDP door's serving task ended (the sibling of the TCP
/// door's [`DoorExit`], for the datagram plane).
#[derive(Debug)]
enum UdpDoorExit {
    /// The in-band association ended (channel EOF; the association
    /// lifetime is the channel's, ADR-003).
    AssociationEnded,
    /// The client's idle bound fired (ADR-014's door-side defense).
    Idle,
    /// The in-band session surface errored.
    Session(String),
    /// A plain I/O failure on the relay socket.
    Io(std::io::Error),
}

impl std::fmt::Display for UdpDoorExit {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            UdpDoorExit::AssociationEnded => f.write_str("association ended"),
            UdpDoorExit::Idle => f.write_str("client idle bound fired"),
            UdpDoorExit::Session(msg) => write!(f, "in-band session: {msg}"),
            UdpDoorExit::Io(err) => write!(f, "io: {err}"),
        }
    }
}

impl UdpFrontDoor {
    /// The builder (the TCP door's sibling shape).
    pub fn builder() -> UdpFrontDoorBuilder {
        UdpFrontDoorBuilder::default()
    }
}

/// The builder for [`UdpFrontDoor`]. The bind address is required
/// (the explicit bind decision, never implied); the idle bound is
/// the ADR-014 door-side defense knob.
#[derive(Default)]
pub struct UdpFrontDoorBuilder {
    bind_addr: Option<SocketAddr>,
    idle: Option<Duration>,
}

impl UdpFrontDoorBuilder {
    /// The local bind address, the relay socket's explicit bind
    /// decision (loopback is the documented posture).
    pub fn bind(mut self, addr: impl Into<SocketAddr>) -> Self {
        self.bind_addr = Some(addr.into());
        self
    }

    /// The client idle bound (ADR-014's door-side defense): no
    /// datagrams from the client within this window closes the
    /// association (the relay socket closes with it).
    pub fn idle(mut self, window: Duration) -> Self {
        self.idle = Some(window);
        self
    }

    pub fn build(self) -> Result<UdpFrontDoor, DoorBuildError> {
        let bind_addr = self.bind_addr.ok_or(DoorBuildError::MissingBindAddr)?;
        Ok(UdpFrontDoor {
            bind_addr,
            idle: self.idle.unwrap_or(Duration::from_secs(30)),
        })
    }
}

impl UdpFrontDoor {
    /// Bind the configured relay address and serve the association
    /// over it. Runs until the association ends (the in-band channel
    /// EOF, the client idle bound, or a fatal I/O failure); the relay
    /// socket closes on return, so the association lifetime is the
    /// channel's, and the door's socket closes with it.
    pub async fn serve(self, client: Arc<ChannelClient>) -> std::io::Result<()> {
        let socket = UdpSocket::bind(self.bind_addr).await?;
        tracing::info!(bind = %self.bind_addr, "udp front door: relay bound");
        self.serve_on(socket, client).await
    }

    /// Serve the association over an existing relay socket (the
    /// test/assembly seam, the same pump without the bind).
    ///
    /// Posture (the peer-pin shape, OQ-SK2-09): the client's source
    /// **IP** is learned from the first inbound datagram and pinned for
    /// the association's lifetime; replies go to the **latest-seen
    /// port** from that IP (a NAT port rebind under a stable IP keeps
    /// receiving). Datagrams from a source IP other than the pinned
    /// one are dropped, per RFC 1928's MUST ("MUST drop any datagrams
    /// arriving from any source IP address other than the one recorded
    /// for the particular association"). The residual (a full client
    /// IP change mid-association (a NAT *address* rebind) stops
    /// receiving replies by design: the RFC's expected-peer posture,
    /// revisitable per OQ-SK2-09. The loopback bind is the documented
    /// posture; a wildcard-bind deployment accepts the same-IP
    /// off-path-first-datagram shape (a much weaker primitive than the
    /// pre-fix any-source pin).
    pub async fn serve_on(
        self,
        socket: UdpSocket,
        client: Arc<ChannelClient>,
    ) -> std::io::Result<()> {
        let mut association = AssociateSession::associate(client.as_ref())
            .await
            .map_err(|err| std::io::Error::other(err.to_string()))?;
        // The reply's sentinel BND.ADDR is received and never used,
        // since the handle is the channel (ADR-003). The real relay address
        // is this door's own bound socket; the vanilla client learns
        // it from the TCP door's ASSOCIATE reply (assembly-layer
        // wiring).
        let _ = association.relay_addr();

        // The datagram pump rides the association's split stream
        // halves: one pump owns the in-band write half (socket→
        // channel), the other the read half (channel→socket), so no
        // shared association lock holds (a mutex over the whole
        // session would deadlock: the channel read parks holding it
        // while the socket side's send needs it). The pump is inlined
        // (the
        // halves borrow the association; the borrow ends before the
        // teardown below).
        let Some((mut channel_read, mut channel_write)) = association.split() else {
            return Ok(());
        };
        let socket = Arc::new(socket);
        // The peer pin (OQ-SK2-09): the client's source IP, learned
        // from the first inbound datagram and pinned; the reply port
        // refreshes to the latest-seen source port from that IP (the
        // NAT-port-rebind shape). A source IP mismatch is the RFC's
        // drop case (the datagram is dropped, not forwarded).
        let client_peer: Arc<std::sync::RwLock<Option<std::net::IpAddr>>> =
            Arc::new(std::sync::RwLock::new(None));
        let reply_port: Arc<std::sync::RwLock<Option<u16>>> =
            Arc::new(std::sync::RwLock::new(None));

        // socket→channel: decode the RFC header (the codec boundary),
        // refuse FRAG ≠ 0 (drop the datagram, the v1 posture), pin
        // the peer IP + enforce the RFC's same-IP drop rule (the pin
        // reacts to *parsed* datagrams only, so protocol garbage can
        // never move it), refresh the reply port, send `{addr, data}`
        // in-band. Each datagram resets the idle window.
        let s2c = {
            let socket = Arc::clone(&socket);
            let client_peer = Arc::clone(&client_peer);
            let reply_port = Arc::clone(&reply_port);
            async move {
                let mut scratch = vec![0u8; 65535];
                loop {
                    let read =
                        tokio::time::timeout(self.idle, socket.recv_from(&mut scratch)).await;
                    let (n, from) = match read {
                        Ok(Ok(pair)) => pair,
                        Ok(Err(err)) => return UdpDoorExit::Io(err),
                        Err(_) => return UdpDoorExit::Idle,
                    };
                    let (frag, addr, payload) =
                        match crate::rfc1928::parse_udp_request(&scratch[..n]).await {
                            Ok(parsed) => parsed,
                            Err(_) => continue,
                        };
                    if frag != 0 {
                        tracing::debug!(
                            peer = %from,
                            "udp front door: FRAG != 0 refused (dropped)"
                        );
                        continue;
                    }
                    {
                        let mut known = client_peer.write().unwrap_or_else(|e| e.into_inner());
                        match *known {
                            None => *known = Some(from.ip()),
                            Some(ip) if ip != from.ip() => {
                                tracing::debug!(
                                    peer = %from,
                                    pinned = %ip,
                                    "udp front door: datagram from a non-expected \
                                     source IP dropped (RFC 1928 peer posture)"
                                );
                                continue;
                            }
                            Some(_) => {}
                        }
                    }
                    *reply_port.write().unwrap_or_else(|e| e.into_inner()) = Some(from.port());
                    if let Err(err) = send_inband(&mut channel_write, &addr, payload).await {
                        return UdpDoorExit::Session(err.to_string());
                    }
                }
            }
        };

        // channel→socket: receive the in-band `{addr, data}` frame,
        // encode the RFC reply header (the codec boundary), send the
        // RFC-header'd datagram back to the pinned peer IP's
        // latest-seen reply port. The association's end (the channel
        // EOF) is this pump's clean exit. The decode state (the
        // incremental reader + the batched-datagram pending queue)
        // persists across pump iterations, so the bridge owns it for
        // the association's lifetime.
        let c2s = {
            let socket = Arc::clone(&socket);
            let client_peer = Arc::clone(&client_peer);
            let reply_port = Arc::clone(&reply_port);
            async move {
                let mut bridge = InbandBridge::new();
                loop {
                    let (addr, payload) = match recv_inband(&mut bridge, &mut channel_read).await {
                        Ok(Some(pair)) => pair,
                        Ok(None) => return UdpDoorExit::AssociationEnded,
                        Err(err) => return UdpDoorExit::Session(err.to_string()),
                    };
                    let mut header = match crate::rfc1928::new_udp_header(addr) {
                        Ok(header) => header,
                        Err(err) => return UdpDoorExit::Session(err.to_string()),
                    };
                    header.extend_from_slice(&payload);
                    let ip = *client_peer.read().unwrap_or_else(|e| e.into_inner());
                    let port = *reply_port.read().unwrap_or_else(|e| e.into_inner());
                    let (Some(ip), Some(port)) = (ip, port) else {
                        continue;
                    };
                    let to = SocketAddr::new(ip, port);
                    if let Err(err) = socket.send_to(&header, to).await {
                        return UdpDoorExit::Io(err);
                    }
                }
            }
        };

        let exit = tokio::select! {
            exit = s2c => exit,
            exit = c2s => exit,
        };
        tracing::debug!("udp front door: association ended: {exit}");
        let _ = association.shutdown().await;
        // A socket I/O failure is the door's fatal outcome (surfaced to
        // the caller); the idle bound, the association's end, and a
        // session error are the association's own ends, so the return
        // is clean.
        match exit {
            UdpDoorExit::Io(err) => Err(err),
            _ => Ok(()),
        }
    }
}

/// The in-band frame send (the write half's shape: frame once with
/// the wire codec, write the framed bytes to the channel write half).
async fn send_inband(
    write: &mut tokio::io::WriteHalf<&mut AssociationStream>,
    addr: &TargetAddr,
    payload: &[u8],
) -> Result<(), SocksSessionError> {
    let framed = crate::wire::frame(addr, payload).map_err(wire_error)?;
    write.write_all(&framed).await?;
    write.flush().await?;
    Ok(())
}

/// The in-band frame read (the read half's shape): a fresh-looking
/// async read whose decoder state **persists across calls**: the
/// bridge's [`InbandBridge`] owns the incremental `DatagramReader`
/// and the batched-datagram pending queue across calls (a datagram
/// split across channel chunks reassembles across reads; frames
/// decoded beyond the one returned queue for later reads; this is
/// the POC
/// #2 reader-state finding, the same shape
/// `AssociateSession::recv_datagram` owns).
struct InbandBridge {
    reader: DatagramReader,
    pending: std::collections::VecDeque<(TargetAddr, Bytes)>,
}

impl InbandBridge {
    fn new() -> Self {
        Self {
            reader: DatagramReader::new(),
            pending: std::collections::VecDeque::new(),
        }
    }

    /// The next in-band datagram: the pending queue first (a chunk
    /// may have batched several frames, drained in order), then an
    /// incremental decode off the read half. `Ok(None)` on channel
    /// EOF (the association's end); fail-loud mid-frame (never a
    /// silent partial datagram).
    async fn recv<R>(
        &mut self,
        read: &mut R,
    ) -> Result<Option<(TargetAddr, Bytes)>, SocksSessionError>
    where
        R: tokio::io::AsyncRead + Unpin,
    {
        if let Some(dg) = self.pending.pop_front() {
            return Ok(Some(dg));
        }
        let mut chunk = vec![0u8; 16 * 1024];
        loop {
            let n = tokio::io::AsyncReadExt::read(read, &mut chunk).await?;
            if n == 0 {
                if self.reader.is_mid_frame() {
                    return Err(SocksSessionError::Rfc(SocksError::Io(std::io::Error::new(
                        std::io::ErrorKind::UnexpectedEof,
                        "association ended mid-datagram (fail-loud)",
                    ))));
                }
                return Ok(None);
            }
            let mut dgs = self.reader.feed(&chunk[..n]).into_iter();
            if let Some(first) = dgs.next() {
                for res in dgs {
                    match res {
                        Ok(pair) => self.pending.push_back(pair),
                        Err(err) => return Err(wire_error(err)),
                    }
                }
                match first {
                    Ok(pair) => return Ok(Some(pair)),
                    Err(err) => return Err(wire_error(err)),
                }
            }
        }
    }
}

/// The channel→socket pump's decode call (the bridge the pump loop
/// drives, so the decode state lives in the bridge, not per call).
async fn recv_inband<R>(
    bridge: &mut InbandBridge,
    read: &mut R,
) -> Result<Option<(TargetAddr, Bytes)>, SocksSessionError>
where
    R: tokio::io::AsyncRead + Unpin,
{
    bridge.recv(read).await
}

/// Map a wire-decode error into the RFC vocabulary (the consumer
/// module's mapping, re-declared here, since the private consumer
/// helper is
/// not part of the public surface).
fn wire_error(err: WireError) -> SocksSessionError {
    match err {
        WireError::Address(addr) => SocksError::AddrError(addr).into(),
        other => SocksError::UdpHeaderError(crate::rfc1928::UdpHeaderError::ReadingError(
            std::io::Error::other(other.to_string()),
        ))
        .into(),
    }
}

/// The association's framed stream type (the split halves' target,
/// the forked client type over the channel's `BiStream`).
type AssociationStream = crate::rfc1928::client::Socks5Stream<alkcall::core::types::BiStream>;

#[cfg(test)]
mod tests {
    use super::*;
    use crate::wire::frame as wire_frame;
    use tokio::io::AsyncWriteExt;
    use tokio::io::{AsyncRead, DuplexStream};

    /// Drive one pump-shaped decode call: feed `bytes` from the peer
    /// side, then decode (the recv parks until a frame completes or
    /// the stream ends, matching the pump loop's shape).
    async fn feed_then_recv<R>(
        bridge: &mut InbandBridge,
        read: &mut R,
        writer: &mut DuplexStream,
        bytes: &[u8],
    ) -> Result<Option<(TargetAddr, Bytes)>, SocksSessionError>
    where
        R: AsyncRead + Unpin,
    {
        if !bytes.is_empty() {
            writer.write_all(bytes).await.expect("feed write");
            writer.flush().await.expect("feed flush");
        }
        bridge.recv(read).await
    }

    #[tokio::test]
    async fn batched_frames_queue_not_drop() {
        // Two frames batched into one feed: the first returns, the
        // second queues and drains on the next call (the per-call
        // reader dropped it, review 003 R-1; the pending queue is
        // the state that persists across the call boundary).
        let (mut writer, reader) = tokio::io::duplex(1024);
        let (mut channel_read, _channel_write) = tokio::io::split(reader);
        let addr = TargetAddr::Ip("10.0.0.2:53".parse().expect("addr"));
        let mut wire = wire_frame(&addr, b"first").unwrap().to_vec();
        wire.extend_from_slice(&wire_frame(&addr, b"second").unwrap());
        let mut bridge = InbandBridge::new();

        let first = feed_then_recv(&mut bridge, &mut channel_read, &mut writer, &wire).await;
        match first {
            Ok(Some((_src, payload))) => assert_eq!(&payload[..], b"first"),
            other => panic!("the first datagram returned: {other:?}"),
        }
        let second = bridge.recv(&mut channel_read).await;
        match second {
            Ok(Some((_src, payload))) => assert_eq!(&payload[..], b"second"),
            other => panic!("the batched datagram queued, not lost: {other:?}"),
        }
    }

    #[tokio::test]
    async fn eof_mid_frame_fails_loud_not_silent() {
        // A frame in flight when the stream ends: the decode errors
        // loud (never a silent partial datagram), matching the same
        // shape
        // AssociateSession::recv_datagram pins.
        let (mut writer, reader) = tokio::io::duplex(1024);
        let (mut channel_read, _channel_write) = tokio::io::split(reader);
        let addr = TargetAddr::Ip("10.0.0.3:53".parse().expect("addr"));
        let framed = wire_frame(&addr, b"truncated").expect("frame");
        let mut bridge = InbandBridge::new();

        let feeder = tokio::spawn(async move {
            writer
                .write_all(&framed[..3])
                .await
                .expect("the partial frame");
            writer.flush().await.expect("flush");
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            let _ = writer.shutdown().await;
        });

        let ended = bridge.recv(&mut channel_read).await;
        assert!(
            ended.is_err(),
            "EOF mid-frame is the fail-loud truncated-datagram condition"
        );
        feeder.await.expect("feeder");
    }

    #[tokio::test]
    async fn partial_frame_state_persists_across_calls() {
        // One chunk carries frame 1 complete + the first bytes of
        // frame 2: call 1 returns frame 1 (frame 2's bytes stay in
        // the bridge's reader), and call 2 (after the stream ends)
        // fails loud on the surviving mid-frame state (a per-call
        // reader would report a clean Ok(None) EOF here).
        let (mut writer, reader) = tokio::io::duplex(1024);
        let (mut channel_read, _channel_write) = tokio::io::split(reader);
        let addr = TargetAddr::Ip("10.0.0.4:53".parse().expect("addr"));
        let mut wire = wire_frame(&addr, b"whole").unwrap().to_vec();
        wire.extend_from_slice(&wire_frame(&addr, b"partial").unwrap()[..3]);
        let mut bridge = InbandBridge::new();

        let feeder = tokio::spawn(async move {
            writer.write_all(&wire).await.expect("the mixed chunk");
            writer.flush().await.expect("flush");
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            let _ = writer.shutdown().await;
        });

        let first = bridge.recv(&mut channel_read).await;
        match first {
            Ok(Some((_src, payload))) => assert_eq!(&payload[..], b"whole"),
            other => panic!("frame 1 returned: {other:?}"),
        }
        let second = bridge.recv(&mut channel_read).await;
        assert!(
            second.is_err(),
            "frame 2's partial bytes survived the call boundary (fail-loud at EOF)"
        );
        feeder.await.expect("feeder");
    }
}

#[cfg(test)]
mod refusal_and_wire_unit_tests {
    use super::*;
    use crate::SocksOpenError;
    use alkcall::channels::client::ChannelOpenError;
    use alkcall::protocol::wire::CallError;
    use tokio::io::AsyncReadExt;

    /// A connected TcpStream pair (the loopback listener shape;
    /// `TcpStream` has no `pair` constructor, so the listener dial is
    /// the portable way).
    async fn socket_pair() -> (tokio::net::TcpStream, tokio::net::TcpStream) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("addr");
        let client = tokio::net::TcpStream::connect(addr).await.expect("connect");
        let (server, _peer) = listener.accept().await.expect("accept");
        (client, server)
    }

    /// The BIND reply#1 wire table (review 003 N-8: the Domain arm was
    /// hand-written wire code with no test): the IP form decodes
    /// through the forked codec (`new_reply`'s own table covers it);
    /// the Domain form's bytes are pinned exactly (VER, REP, RSV,
    /// ATYP=domain, len, name, port.
    #[test]
    fn bind_reply1_wire_forms() {
        let ip = std::net::SocketAddr::from(([127, 0, 0, 1], 2121));
        let ip_wire = bind_reply1_wire(&TargetAddr::Ip(ip));
        assert_eq!(ip_wire[0], consts::SOCKS5_VERSION);
        assert_eq!(ip_wire[1], consts::SOCKS5_REPLY_SUCCEEDED);
        assert_eq!(ip_wire[3], consts::SOCKS5_ADDR_TYPE_IPV4);
        assert_eq!(&ip_wire[4..8], &[127, 0, 0, 1]);
        assert_eq!(&ip_wire[8..10], &2121u16.to_be_bytes());

        let domain_wire = bind_reply1_wire(&TargetAddr::Domain("bind.test".into(), 2121));
        let expected = {
            let mut v = vec![
                consts::SOCKS5_VERSION,
                consts::SOCKS5_REPLY_SUCCEEDED,
                0x00,
                consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME,
                9,
            ];
            v.extend_from_slice(b"bind.test");
            v.extend_from_slice(&2121u16.to_be_bytes());
            v
        };
        assert_eq!(domain_wire, expected, "the domain BND.ADDR frame");
    }

    /// `write_refusal_reply`'s non-reply-code arm (review 003 N-8): an
    /// open/transport-shaped session error stays a bare close, so no
    /// bytes are written to the vanilla client.
    #[tokio::test]
    async fn write_refusal_reply_skips_non_reply_code_shapes() {
        let (mut client, mut server) = socket_pair().await;
        let open_err =
            SocksSessionError::Open(SocksOpenError::Open(ChannelOpenError::CallFailed {
                error: CallError::new("channel:open_failed", "refused", false),
            }));
        write_refusal_reply(&mut server, &open_err).await;
        // Nothing written: the client's read finds no bytes (the bare
        // close is the door's caller's act, since the helper itself
        // is silent). A short window proves no reply bytes exist.
        let mut probe = [0u8; 16];
        let n = tokio::time::timeout(
            std::time::Duration::from_millis(200),
            client.read(&mut probe),
        )
        .await;
        assert!(
            matches!(n, Err(_) | Ok(Ok(0))),
            "no reply bytes for a non-reply-code error: {n:?}"
        );
    }

    /// The refusal re-encode's wire shape (the F-1 fix's helper,
    /// byte-pinned end to end): a reply-code-shaped session error
    /// reaches the vanilla client as the RFC reply with the code +
    /// the placeholder BND.ADDR.
    #[tokio::test]
    async fn write_refusal_reply_encodes_the_mapped_code() {
        let (mut client, mut server) = socket_pair().await;
        let refused =
            SocksSessionError::Rfc(SocksError::ReplyError(ReplyError::ConnectionNotAllowed));
        write_refusal_reply(&mut server, &refused).await;
        let mut wire = [0u8; 10];
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.read_exact(&mut wire),
        )
        .await
        .expect("the reply arrives")
        .expect("read");
        assert_eq!(wire[0], consts::SOCKS5_VERSION);
        assert_eq!(wire[1], consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED);
        assert_eq!(&wire[4..8], &[0, 0, 0, 0], "the placeholder BND.ADDR");
        assert_eq!(&wire[8..10], &0u16.to_be_bytes());
    }
}

#[cfg(test)]
mod bridge_later_batch_tests {
    use super::*;
    use tokio::io::AsyncWriteExt;

    /// A decode failure in a *later* batch frame (review 003 N-8's
    /// residual: the unit tests pinned the happy shapes, where a
    /// garbage
    /// frame following a good one in the same chunk surfaces the
    /// error, the good frame is NOT silently delivered first).
    #[tokio::test]
    async fn garbage_in_a_later_batch_frame_fails_loud() {
        let (mut writer, reader) = tokio::io::duplex(1024);
        let (mut channel_read, _channel_write) = tokio::io::split(reader);
        let addr = TargetAddr::Ip("10.0.0.7:53".parse().expect("addr"));
        let mut wire = crate::wire::frame(&addr, b"good").unwrap().to_vec();
        // A later frame whose length field promises more bytes than
        // the chunk carries, then the stream ends, so the reader's
        // state machine fails loud on the truncated later frame.
        wire.extend_from_slice(&[0xff, 0xff]);
        wire.extend_from_slice(b"short");
        let mut bridge = InbandBridge::new();
        writer.write_all(&wire).await.expect("feed");
        writer.flush().await.expect("flush");
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        let _ = writer.shutdown().await;

        let first = bridge.recv(&mut channel_read).await;
        assert!(
            first.is_err() || matches!(&first, Ok(Some((src, payload))) if &payload[..] == b"good"),
            "the first frame or the loud error: {first:?}"
        );
        if matches!(first, Ok(Some(_))) {
            // The good frame delivered, the truncated later frame
            // fails loud at EOF (the reader's mid-frame state).
            let second = bridge.recv(&mut channel_read).await;
            assert!(
                second.is_err(),
                "the truncated later batch frame fails loud: {second:?}"
            );
        }
    }
}