lyrebird/

lib.rs

//! # Lyrebird - Pluggable Transport Proxy Applications
//!
//! This crate provides a PT-manager loop usable as a library
//! (`lyrebird::run()`) or as a standalone binary that implements the
//! Tor pluggable-transport spec on top of the [`ptrs`] interface.
//!
//! **Stability**: the public API is unstable and subject to change
//! without notice; do not rely on this crate for security-critical
//! applications. The server-side path is incomplete and gated behind
//! the `experimental-server` cargo feature — it must not be enabled in
//! production.
//!
//! ## Lyrebird Pluggable Transport Bridge
//!
//! ['lyrebird'] provides an executable program designed to manage the calling
//! interface used by the Tor libraries when launching pluggable transports (see `pt-spec.txt`).
//!
//! `... [tor_client] <---> [pt_client] <====> [pt_bridge] <---> [tor_orport] ...`
//!
//! Usage info:
//!
//! ```txt
//! Tunnel Tor SOCKS5 traffic through pluggable transport connections
//!
//! Usage: lyrebird [OPTIONS]
//!
//! Options:
//!       --enable-logging         Log to {TOR_PT_STATE_LOCATION}/obfs4proxy.log
//!       --log-level <LOG_LEVEL>  Log Level (ERROR/WARN/INFO/DEBUG/TRACE) [default: ERROR]
//!       --unsafe-logging         Disable the address scrubber on logging
//!   -h, --help                   Print help
//!   -V, --version                Print version
//! ```
//!
//! ### Installation
//!
//! To install:
//!
//! `cargo install lyrebird`
//!
//! This installs in the configured Rust location (i.e. `$HOME/.cargo/bin`). You may
//! wish to copy `./lyrebird` to a permanent location (e.g. `/usr/local/bin`).
//!
//! Client side torrc configuration:
//! ```text
//! ClientTransportPlugin obfs4 exec /usr/local/bin/lyrebird
//! ```
//!
//! Bridge side torrc configuration:
//! ```text
//! # Act as a bridge relay.
//! BridgeRelay 1
//!
//! # Enable the Extended ORPort
//! ExtORPort auto
//!
//! # Use lyrebird to provide the obfs4 protocol.
//! ServerTransportPlugin obfs4 exec /usr/local/bin/lyrebird
//!
//! # (Optional) Listen on the specified address/port for obfs4 connections as
//! # opposed to picking a port automatically.
//! #ServerTransportListenAddr obfs4 0.0.0.0:443
//! ```
//!
//! ### Tips and tricks
//!
//!  * On modern Linux systems it is possible to have lyrebird bind to reserved
//!    ports (<=1024) even when not running as root by granting the
//!    `CAP_NET_BIND_SERVICE` capability with setcap:
//!
//!    `# setcap 'cap_net_bind_service=+ep' /usr/local/bin/lyrebird`
//!
//!  * The autogenerated obfs4 bridge parameters are placed in
//!    `DataDir/pt_state/obfs4_state.json`.  To ease deployment, the client side
//!    bridge line is written to `DataDir/pt_state/obfs4_bridgeline.txt`.
//!
// Follow-up work tracked in the issue tracker, not in rustdoc:
//   - tunnel-level metrics (failures, byte counters).
//   - rate-limiting for the bidirectional copy.
//   - replace `tokio::io::copy_bidirectional` with an interactive copy
//     that flushes only when the reader stalls.

#![allow(unused, dead_code)]
#![deny(missing_docs)]

use obfs4::{ClientBuilder, Obfs4PT};
use ptrs::{error, info, warn};
use ptrs::{
    ClientBuilder as _, ClientTransport, PluggableTransport, ServerBuilder, ServerTransport,
};

use anyhow::{anyhow, Context, Result};
use clap::Parser;
use fast_socks5::{
    server::{DenyAuthentication, SimpleUserPassword},
    util::target_addr::TargetAddr,
    AuthenticationMethod,
};
use safelog::sensitive;
use tokio::task::JoinSet;
use tokio::{
    io::{copy_bidirectional, AsyncRead, AsyncWrite, AsyncWriteExt},
    net::{TcpListener, TcpStream},
    sync::oneshot,
};
use tokio_util::sync::CancellationToken;
use tracing::Level;
use tracing_subscriber::{filter::LevelFilter, prelude::*};

use std::{env, net::SocketAddr, pin::Pin, str::FromStr, sync::Arc};

/// Maximum number of concurrently handled client connections.
/// Bounds memory/task growth under connection floods (DoS mitigation).
const MAX_CONCURRENT_CONNS: usize = 1024;

/// Client Socks address to listen on.
const CLIENT_SOCKS_ADDR: &str = "127.0.0.1:0";

/// Error defined to denote a failure to get the bridge line
#[derive(Debug, thiserror::Error)]
#[error("Error while obtaining bridge line data")]
struct BridgeLineParseError;

#[derive(Parser)]
#[command(author, version, long_about = None, about="Tunnel Tor SOCKS5 traffic through pluggable transport connections")]
struct Args {
    /// Log to {TOR_PT_STATE_LOCATION}/obfs4proxy.log
    #[arg(long, default_value_t = false)]
    enable_logging: bool,

    /// Log Level (ERROR/WARN/INFO/DEBUG/TRACE)
    #[arg(long, default_value_t=String::from("ERROR"))]
    log_level: String,

    /// Disable the address scrubber on logging
    #[arg(long, default_value_t = false)]
    unsafe_logging: bool,
}

/// initialize the logging receiver(s) for things to be logged into using the
/// tracing / tracing_subscriber libraries
// TODO: GeoIP. Json for file log writer.
fn init_logging_recvr(
    enable: bool,
    should_scrub: bool,
    level_str: &str,
    statedir: &str,
) -> Result<()> {
    if should_scrub {
        safelog::enforce_safe_logging();
    } else {
        safelog::disable_safe_logging();
    }

    // The PT protocol owns stdout for the parent ↔ PT control channel.
    // Logs go to stderr; the file layer below (when enabled) writes
    // separately into the state directory.
    //
    // Compact (single-line) layout matches the parent process's
    // tracing-subscriber default — we share a terminal with it when
    // launched via the busybox dispatch, so consistent formatting
    // matters more than the pretty multi-line output. Per-target
    // filter mutes the chatty `fast_socks5` connection log; arti is
    // the one bashing through dozens of PT connections during
    // bootstrap and the per-conn lines drown the relevant signal.
    let console_filter =
        tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| {
            tracing_subscriber::EnvFilter::new(
                "info,fast_socks5=warn,obfs4::sessions=warn,lyrebird::handshake=info",
            )
        });
    let console_layer = tracing_subscriber::fmt::layer()
        .with_writer(std::io::stderr)
        .with_filter(console_filter);

    let log_layers = if enable {
        let level = Level::from_str(level_str)?;

        let file = std::fs::File::create(format!("{statedir}/obfs4proxy.log"))?;

        let state_dir_layer = tracing_subscriber::fmt::layer()
            .with_writer(file)
            .with_filter(LevelFilter::from_level(level));

        console_layer.and_then(state_dir_layer).boxed()
    } else {
        console_layer.boxed()
    };

    tracing_subscriber::registry().with(log_layers).init();

    Ok(())
}

/// Resolve a `fast_socks5::util::TargetAddr` to a concrete `SocketAddr`.
/// `Ip(_)` variants pass through; `Domain` variants always fail because
/// the PT spec forbids the transport from doing DNS — that's the calling
/// Tor client's responsibility.
pub fn resolve_target_addr(addr: &TargetAddr) -> Result<SocketAddr> {
    match addr {
        TargetAddr::Ip(sa) => Ok(*sa),
        TargetAddr::Domain(_, _) => {
            // this will always fail because ptrs does not do dns lookups.
            ptrs::resolve_addr(format!("{addr}")).context("domain resolution failed")
        }
    }
}

/// Run the lyrebird pluggable transport. Expects the standard
/// `TOR_PT_*` environment variables set by the parent process (arti /
/// tor) and speaks the PT-managed-transport protocol on stdin/stdout.
///
/// # Cancel safety
///
/// This function is **not cancel-safe**. It manages long-lived server
/// state and open connections. Dropping the future will abandon all
/// active tunnels without graceful shutdown.
pub async fn run() -> Result<()> {
    let args = Args::parse();

    // Make state directory
    let statedir = ptrs::make_state_dir()?;

    // launch tracing subscriber with filter level
    init_logging_recvr(
        args.enable_logging,
        !args.unsafe_logging,
        &args.log_level,
        &statedir,
    )?;

    let cancel_token = tokio_util::sync::CancellationToken::new();

    // launch runners
    let mut exit_rx = if ptrs::is_client()? {
        // running as CLIENT
        client_setup(&statedir, cancel_token.clone()).await?
    } else {
        // running as SERVER
        //
        // Server-side is gated behind `experimental-server` because the PT
        // handshake is not yet wired and the ExtORPort dial is missing.
        // Without the feature we refuse to start rather than silently
        // exposing an unauthenticated proxy.
        #[cfg(feature = "experimental-server")]
        {
            server_setup(&statedir, cancel_token.clone()).await?
        }
        #[cfg(not(feature = "experimental-server"))]
        {
            let _ = &statedir;
            let _ = &cancel_token;
            return Err(anyhow!(
                "lyrebird server-side is not implemented; rebuild with \
                 `--features experimental-server` for development use only"
            ));
        }
    };

    info!("accepting connections");

    // At this point, the pt config protocol is finished, and incoming
    // connections will be processed.  Wait till the parent dies
    // (immediate exit), a SIGTERM / Ctrl+Break is received (immediate
    // exit), or a SIGINT / Ctrl+C is received.
    tokio::select! {
        _ = &mut exit_rx => {
            info!("proxy closed");
            return Ok(())
        }
        sig = shutdown_signal() => {
            if sig.is_terminate() {
                info!("proxy terminated");
                return Ok(())
            }
            info!("received interrupt, shutting down");
            cancel_token.cancel();
        }
    }

    // Ok, it was the first interrupt, close all listeners, and wait till
    // the parent dies, all current connections are closed, or either
    // a second interrupt/terminate is received.
    tokio::select! {
        _ = exit_rx => {}
        _ = shutdown_signal() => {}
    }

    Ok(())
}

/// Cross-platform shutdown-signal helper. On Unix maps `SIGTERM` →
/// `Terminate`, `SIGINT` → `Interrupt`. On Windows uses
/// `Ctrl+Break` → `Terminate`, `Ctrl+C` → `Interrupt`.
#[derive(Clone, Copy)]
enum Shutdown {
    Interrupt,
    Terminate,
}

impl Shutdown {
    fn is_terminate(self) -> bool {
        matches!(self, Shutdown::Terminate)
    }
}

#[cfg(unix)]
async fn shutdown_signal() -> Shutdown {
    use tokio::signal::unix::{signal, SignalKind};
    let mut sigint = signal(SignalKind::interrupt()).expect("install SIGINT handler");
    let mut sigterm = signal(SignalKind::terminate()).expect("install SIGTERM handler");
    tokio::select! {
        _ = sigterm.recv() => Shutdown::Terminate,
        _ = sigint.recv() => Shutdown::Interrupt,
    }
}

#[cfg(not(unix))]
async fn shutdown_signal() -> Shutdown {
    use tokio::signal::windows::{ctrl_break, ctrl_c};
    let mut c_c = ctrl_c().expect("install Ctrl+C handler");
    let mut c_break = ctrl_break().expect("install Ctrl+Break handler");
    tokio::select! {
        _ = c_break.recv() => Shutdown::Terminate,
        _ = c_c.recv() => Shutdown::Interrupt,
    }
}

// ================================================================ //
//                            Client                                //
// ================================================================ //

async fn client_setup(
    statedir: &str,
    cancel_token: CancellationToken,
) -> Result<oneshot::Receiver<bool>> {
    let obfs4_name = Obfs4PT::name();
    let webtunnel_name = webtunnel::WEBTUNNEL_NAME.to_string();
    let client_pt_info = ptrs::ClientInfo::new()?;
    // TOR_PT_PROXY is optional. The downstream code never reads this
    // value (see `client_handle_connection`'s `_proxy_uri`), so we
    // substitute a placeholder URL when tor did not supply one.
    let proxy_uri = client_pt_info
        .uri
        .unwrap_or_else(|| url::Url::parse("data:,").expect("placeholder url"));
    let (tx, rx) = oneshot::channel::<bool>();

    // PT spec §3.3.3: announce our protocol version to the parent.
    pt_proto::print_version();

    let mut listeners: Vec<
        std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>>,
    > = Vec::new();

    for name in client_pt_info.methods {
        info!(name);

        if name == obfs4_name {
            let builder = Obfs4PT::client_builder();
            let listener = tokio::net::TcpListener::bind(CLIENT_SOCKS_ADDR).await?;
            let local_addr = listener.local_addr()?;
            pt_proto::print_cmethod(&name, "socks5", local_addr);
            listeners.push(Box::pin(client_accept_loop(
                listener,
                builder,
                proxy_uri.clone(),
                cancel_token.clone(),
            )));
        } else if name == webtunnel_name {
            let builder = webtunnel::WebTunnelBuilder::default();
            let listener = tokio::net::TcpListener::bind(CLIENT_SOCKS_ADDR).await?;
            let local_addr = listener.local_addr()?;
            pt_proto::print_cmethod(&name, "socks5", local_addr);
            listeners.push(Box::pin(client_accept_loop(
                listener,
                builder,
                proxy_uri.clone(),
                cancel_token.clone(),
            )));
        } else {
            pt_proto::print_cmethod_error(&name, "no such transport is supported");
            warn!("no such transport is supported");
            continue;
        }
    }

    // PT spec §3.3.3: end of method announcements.
    pt_proto::print_cmethods_done();

    // spawn a task that runs and monitors the progress of the listeners.
    tokio::spawn(async move {
        let total_len = listeners.len();
        let mut running = total_len;

        // launch all listener futures
        let mut pt_set = JoinSet::new();
        for fut in listeners {
            pt_set.spawn(fut);
        }

        // if any of the listeners exit, handle it
        while let Some(res) = pt_set.join_next().await {
            running -= 1;
            if let Err(e) = res {
                warn!("listener failed: {e}");
            }
            info!("{running}/{total_len} listeners running");
        }

        // if all listeners exit then we can send the tx signal.
        // Best-effort: the receiver may already be dropped if the
        // parent select! moved on (e.g. signal-driven shutdown).
        let _ = tx.send(true);
    });

    Ok(rx)
}

async fn client_accept_loop<C>(
    listener: TcpListener,
    builder: impl ptrs::ClientBuilder<TcpStream, ClientPT = C> + Send + 'static,
    proxy_uri: url::Url,
    cancel_token: CancellationToken,
) -> Result<()>
where
    // the provided client builder should build the C ClientTransport.
    C: ptrs::ClientTransport<TcpStream, std::io::Error> + Send + 'static,
{
    let pt_name = C::method_name();
    let sem = Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_CONNS));
    loop {
        tokio::select! {
            _ = cancel_token.cancelled() => {
                info!("{pt_name} received shutdown signal");
                break; // exit the loop so graceful shutdown actually stops accepting
            }
            res = listener.accept() => {
                let (conn, client_addr) = match res {
                    Err(e) => {
                        error!("failed to accept tcp connection {e}");
                        break;
                    }
                    Ok(c) => c,
                };
                // Acquire a concurrency permit before spawning, bounding
                // the number of in-flight connection tasks (DoS mitigation).
                let permit = match Arc::clone(&sem).acquire_owned().await {
                    Ok(p) => p,
                    Err(_) => break, // semaphore closed — shutting down
                };
                let builder_clone = builder.clone();
                let proxy_clone = proxy_uri.clone();
                tokio::spawn(async move {
                    let _permit = permit; // held for the task's lifetime, freed on drop
                    let _ = client_handle_connection(conn, builder_clone, proxy_clone, client_addr).await;
                });
            }
        }
    }

    Ok(())
}

/// This function assumes that the provided connection / socket manages reconstruction
/// and reliability before passing to this layer.
async fn client_handle_connection<In, C, B>(
    conn: In,
    mut builder: B,
    _proxy_uri: url::Url,
    client_addr: SocketAddr,
) -> Result<()>
where
    // the provided T must be usable as a connection in an async context
    In: AsyncRead + AsyncWrite + Send + Unpin,
    // the provided client builder should build the C ClientTransport.
    C: ptrs::ClientTransport<TcpStream, std::io::Error> + Send,
    B: ptrs::ClientBuilder<TcpStream, ClientPT = C>,
{
    // PT-spec §3.5 requires the parent to negotiate USERNAME/PASSWORD
    // (auth method 0x02) and pack the per-bridge PT arguments into the
    // UNAME/PASSWD fields. We accept any creds (and also accept clients
    // that legitimately want NO_AUTH) — see `PtArgsAuth` below.
    let mut config =
        fast_socks5::server::Config::<fast_socks5::server::DenyAuthentication>::default()
            .with_authentication(PtArgsAuth);
    config.set_allow_no_auth(true);
    // Critical: we are a pluggable transport, not a plain TCP proxy. With
    // the default `execute_command = true`, `upgrade_to_socks5()` would
    // itself open a *plain* TCP connection to the bridge, send the SOCKS
    // reply, and proxy the parent against that — bypassing obfs4 entirely
    // and consuming the parent socket. We must only *read* the request
    // here and do the obfs4 dial + reply ourselves below.
    config.set_execute_command(false);
    let socks5_conn = fast_socks5::server::Socks5Socket::new(conn, Arc::new(config));

    let mut socks5_conn = socks5_conn.upgrade_to_socks5().await?;
    let creds = socks5_conn.take_credentials();

    let target_addr = socks5_conn
        .target_addr()
        .ok_or(BridgeLineParseError)
        .context("missing remote address in request")?;

    // Reconstruct the PT-spec argument string from SOCKS5 user/pass and
    // hand it to the transport's `options(&Args)` so the obfs4
    // ClientBuilder picks up `cert=...` / `iat-mode=...` etc.
    let arg_string = arg_string_from_creds(creds);
    let args = if arg_string.is_empty() {
        ptrs::args::Args::default()
    } else {
        ptrs::args::Args::from_str(&arg_string).context("parsing PT arg string")?
    };
    <B as ptrs::ClientBuilder<TcpStream>>::options(&mut builder, &args)
        .map_err(|e| anyhow::anyhow!("applying PT args to builder: {e}"))?;

    let remote_addr = resolve_target_addr(target_addr).context("no remote address")?;

    // The outgoing OR-port dial. Boxing it as a `FutureResult` is the seam
    // exercised by tests: `establish_pt_conn` only ever sees a pinned
    // future yielding the underlying socket, so an in-memory
    // `tokio::io::duplex()` half can stand in for the real `TcpStream`.
    let remote: Pin<ptrs::FutureResult<TcpStream, std::io::Error>> =
        Box::pin(tokio::net::TcpStream::connect(remote_addr));

    // build the pluggable transport client and then dial, completing the
    // connection and handshake when the future is await-ed.
    let pt_client = builder.build();
    let mut pt_conn = establish_pt_conn(pt_client, remote, client_addr).await?;

    // The obfs4 tunnel to the bridge is up, so report CONNECT success to
    // the PT parent (arti/tor). Because we disabled `execute_command`,
    // fast-socks5 has NOT sent any reply — we must. This is the canonical
    // SOCKS5 success frame: VER=5, REP=0 (succeeded), RSV=0, ATYP=1
    // (IPv4), BND.ADDR=0.0.0.0, BND.PORT=0. The parent ignores BND.*.
    let mut parent = socks5_conn.into_inner();
    parent
        .write_all(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
        .await
        .context("writing SOCKS5 success reply to PT parent")?;
    parent
        .flush()
        .await
        .context("flushing SOCKS5 success reply")?;

    if let Err(e) = copy_bidirectional(&mut parent, &mut pt_conn).await {
        warn!(
            addres = sensitive(client_addr).to_string(),
            "tunnel closed with error: {e:#?}"
        );
    }
    Ok(())
}

/// Drive a built pluggable-transport client to completion against a dial
/// future, returning the wrapped tunnel stream.
///
/// This is the outgoing-connection seam carved out of
/// [`client_handle_connection`]. The dial is supplied as an already-pinned
/// [`ptrs::FutureResult`] (the same shape `establish` consumes) rather than
/// being created inline with `TcpStream::connect`, so tests can inject an
/// in-memory `tokio::io::duplex()` half — or a future that fails — and
/// observe the handshake behaviour without touching the network.
///
/// On handshake failure the client address is logged (scrubbed) and a typed
/// error is returned; the function never panics on a dial or handshake
/// error reachable from the network.
///
/// # Cancel safety
///
/// cancel-safe: NO — `establish` runs the obfs4/webtunnel handshake, which
/// writes to the underlying socket. Dropping this future mid-handshake can
/// leave the peer expecting bytes that were never sent. The caller
/// (`client_handle_connection`) runs inside a dedicated `tokio::spawn` task,
/// so it is not composed under a `select!`/`timeout` cancellation point.
pub(crate) async fn establish_pt_conn<In2, C2>(
    pt_client: C2,
    dial: Pin<ptrs::FutureResult<In2, std::io::Error>>,
    client_addr: SocketAddr,
) -> Result<C2::OutRW>
where
    // The dialed socket must be usable as an async connection. These are the
    // same bounds the transport traits require of their `InRW` parameter.
    In2: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
    C2: ptrs::ClientTransport<In2, std::io::Error> + Send,
{
    match ptrs::ClientTransport::<In2, std::io::Error>::establish(pt_client, dial).await {
        Err(e) => {
            warn!(
                address = sensitive(client_addr).to_string(),
                "handshake failed: {e:#?}"
            );
            Err(obfs4::Error::from(e.to_string())).context("handshake failed")
        }
        Ok(c) => Ok(c),
    }
}

/// This function assumes that the provided connection / socket manages reconstruction
/// and reliability before passing to this layer.
///
/// proxy_uri (currently unused) is meant to indicate that the outgoing connection
/// should be made through _another_ proxy based on the proxy uri. This is relatively
/// easy in golang, but I am not sure how easy it will be here. I believe this is
/// a rather uncommon option so it is left unimplemented for now.
async fn client_handle_connection_clientpt<In, C>(
    conn: In,
    pt_client: C,
    _proxy_uri: url::Url,
    client_addr: SocketAddr,
) -> Result<()>
where
    // the provided T must be usable as a connection in an async context
    In: AsyncRead + AsyncWrite + Send + Unpin,
    // the provided B must implement the Client Builder interface for T
    C: ptrs::ClientTransport<TcpStream, std::io::Error>,
{
    let mut config: fast_socks5::server::Config<SimpleUserPassword> =
        fast_socks5::server::Config::default();
    // config.set_skip_auth(true);
    let mut socks5_conn = fast_socks5::server::Socks5Socket::new(conn, Arc::new(config));

    // let mut socks5_conn = socks5_conn.upgrade_to_socks5().await?;
    let target_addr = socks5_conn
        .target_addr()
        .ok_or(BridgeLineParseError)
        .context("missing remote address in request")?;

    // TODO: get args from the socks request username:Password if it exists.
    // This seems non-trivial to match against the golang obfs4 implementation.
    // Maybe implement my own thing that implements the `Authenticate` trait?
    // Maybe work with the tor_socksproto package?
    //
    // Pluggable transports use the username/password field to pass
    // per-connection arguments.  The fields contain ASCII strings that
    // are combined and then parsed into key/value pairs.
    // argStr := string(uname)
    // if !(plen == 1 && passwd[0] == 0x00) {
    let args: Option<ptrs::args::Args> = match socks5_conn.auth() {
        AuthenticationMethod::Password { username, password } => {
            if username.is_empty() {
                socks5_conn.flush().await?;
                socks5_conn.shutdown().await?;
                return Err(anyhow!("username with 0 length"));
            }
            if password.is_empty() {
                socks5_conn.flush().await?;
                socks5_conn.shutdown().await?;
                return Err(anyhow!("password with 0 length"));
            }

            let mut arg_string = username.clone();
            // tor will set the password to 'NUL', if the field doesn't contain any
            // actual argument data.
            if !(password.len() == 1 && password.as_bytes().first().copied() == Some(0x00)) {
                arg_string.push_str(password);
            }

            match ptrs::args::Args::from_str(&arg_string) {
                Ok(a) => Some(a),
                Err(e) => {
                    return Err(anyhow!(
                        "failed to parse provided args \"{arg_string}\": {e}"
                    ))
                }
            }
        }
        AuthenticationMethod::None => None,
        _ => return Err(anyhow!("negotiated unsupported authentication method")),
    };

    let remote_addr = resolve_target_addr(target_addr).context("no remote address")?;

    let remote = tokio::net::TcpStream::connect(remote_addr);

    // build the pluggable transport client and then dial, completing the
    // connection and handshake when the `wrap(..)` is await-ed.
    let mut pt_conn = match pt_client.establish(Box::pin(remote)).await {
        Ok(c) => c,
        Err(e) => {
            warn!(
                address = sensitive(client_addr).to_string(),
                "handshake failed: {e:#?}"
            );
            return Err(obfs4::Error::from(e.to_string())).context("handshake failed");
        }
    };

    if let Err(e) = copy_bidirectional(&mut socks5_conn.into_inner(), &mut pt_conn).await {
        warn!(
            addres = sensitive(client_addr).to_string(),
            "tunnel closed with error: {e:#?}"
        );
    }

    Ok(())
}

// ================================================================ //
//                            Server                                //
// ================================================================ //
//
// All server-side plumbing is gated behind the `experimental-server`
// cargo feature. The PT handshake and ExtORPort dial are not wired
// (see `server_handle_connection`), so compiling this in by default
// would risk operators standing up an unauthenticated proxy.

#[cfg(feature = "experimental-server")]
async fn server_setup(
    statedir: &str,
    cancel_token: CancellationToken,
) -> Result<oneshot::Receiver<bool>> {
    let obfs4_name = Obfs4PT::name();

    let server_info = ptrs::ServerInfo::new()?;
    let (tx, rx) = oneshot::channel::<bool>();

    let mut listeners = Vec::new();

    for bind_addr in server_info.bind_addrs {
        info!(bind_addr.method_name);
        if bind_addr.method_name != obfs4_name {
            warn!("no such transport is supported");
            continue;
        }

        let mut builder = Obfs4PT::server_builder();
        let server = builder
            .statefile_location(statedir)?
            .options(&bind_addr.options)?
            .build();

        let listener = tokio::net::TcpListener::bind(bind_addr.addr).await?;
        listeners.push(server_listen_loop::<TcpStream, _>(
            listener,
            server,
            cancel_token.clone(),
        ));
    }

    // spawn a task that runs and monitors the progress of the listeners.
    tokio::spawn(async move {
        let total_len = listeners.len();
        let mut running = total_len;

        // launch all listener futures
        let mut pt_set = JoinSet::new();
        for fut in listeners {
            pt_set.spawn(fut);
        }

        // if any of the listeners exit, handle it
        while let Some(res) = pt_set.join_next().await {
            running -= 1;
            if let Err(e) = res {
                warn!("listener failed: {e}");
            }
            info!("{running}/{total_len} listeners running");
        }

        // if all listeners exit then we can send the tx signal.
        // Best-effort: the receiver may already be dropped if the
        // parent select! moved on (e.g. signal-driven shutdown).
        let _ = tx.send(true);
    });

    Ok(rx)
}

#[cfg(feature = "experimental-server")]
async fn server_listen_loop<In, S>(
    listener: TcpListener,
    server: S,
    cancel_token: CancellationToken,
) -> Result<()>
where
    // the provided In must be usable as a connection in an async context
    In: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
    // The provided S must be usable as a Pluggable Transport Server.
    S: ptrs::ServerTransport<In> + Send + Sync + ptrs::ServerTransport<TcpStream> + 'static,
    <S as ptrs::ServerTransport<In>>::OutErr: 'static,
{
    let method_name = <S as ServerTransport<In>>::method_name();
    let server = Arc::new(server);
    loop {
        tokio::select! {
            _ = cancel_token.cancelled() => {
                info!("{method_name} received shutdown signal - closing listener");
                break
            }
            res = listener.accept() => {
                let (mut conn, client_addr) = match res {
                    Err(e) => {
                       error!("{method_name} closing listener - failed to accept tcp connection {e}");
                       break;
                   }
                   Ok(c) => c,
               };
               tokio::spawn(server_handle_connection(
                   conn,
                   server.clone(),
                   client_addr,
               ));
            }
        }
    }

    Ok(())
}

#[cfg(feature = "experimental-server")]
async fn server_handle_connection<In, S>(
    mut conn: In,
    server: Arc<S>,
    client_addr: SocketAddr,
) -> Result<()>
where
    // the provided In must be usable as a connection in an async context
    In: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
    // The provided S must be usable as a Pluggable Transport Server.
    S: ptrs::ServerTransport<In> + Send + Sync + ptrs::ServerTransport<TcpStream>,
    <S as ptrs::ServerTransport<In>>::OutErr: 'static,
{
    let _ = (&mut conn, server, client_addr);
    // Two pieces are still missing here:
    //   1) server.reveal(conn) to complete the PT handshake;
    //   2) a real ExtORPort dial to the parent ORPort.
    // The previous code shipped neither and instead unconditionally
    // dialed a hardcoded 127.0.0.1:8000, which would have stood up an
    // unauthenticated TCP proxy on any host running this build.
    unimplemented!(
        "lyrebird server-side PT handshake is not implemented; \
         do not enable experimental-server in production"
    );
}

// ================================================================ //
//        PT-spec helpers (auth, args, parent-channel messages)     //
// ================================================================ //

/// Custom `fast_socks5` authenticator. Per pt-spec §3.5 the parent
/// MUST negotiate USERNAME/PASSWORD when it wants to pass per-bridge PT
/// args; we accept any (or no) credentials and just propagate them so
/// `client_handle_connection` can grab them via `take_credentials()`.
#[derive(Clone, Copy, Default)]
struct PtArgsAuth;

#[async_trait::async_trait]
impl fast_socks5::server::Authentication for PtArgsAuth {
    type Item = (String, String);

    async fn authenticate(&self, credentials: Option<(String, String)>) -> Option<Self::Item> {
        // Propagate creds untouched. `None` (NO_AUTH path) is also fine —
        // we let it through as empty strings so the same plumbing applies.
        Some(credentials.unwrap_or_default())
    }
}

/// Reconstruct the PT-spec argument string from the SOCKS5
/// USERNAME/PASSWORD fields, mirroring Go lyrebird `rfc1929.go:88-100`.
///
/// PT spec §3.5 mandates this peculiar split when the arg string
/// exceeds 255 bytes: the parent process packs it into
/// `(username, password)` with `username` capped at 255 bytes.
///
/// * `passwd == [0x00]` is the "no passwd; uname is the whole arg list"
///   marker — return just `uname`.
/// * Otherwise concatenate `uname` and `passwd` as raw bytes, no
///   separator.
/// * `None` means the client connected via NO_AUTH (no args).
pub fn arg_string_from_creds(creds: Option<(String, String)>) -> String {
    match creds {
        None => String::new(),
        Some((uname, passwd)) if passwd.as_bytes() == [0x00] => uname,
        Some((uname, passwd)) => {
            let mut s = uname;
            s.push_str(&passwd);
            s
        }
    }
}

/// Minimal subset of the PT spec's parent-channel protocol (the lines
/// the PT prints on stdout for `tor`/`arti` to read). Without these
/// arti waits forever and reports "PT binary gone".
mod pt_proto {
    use std::io::Write;
    use std::net::SocketAddr;

    const VERSION: &str = "1";

    pub fn print_version() {
        emit(format!("VERSION {VERSION}"));
    }

    pub fn print_cmethod(transport: &str, proto: &str, addr: SocketAddr) {
        emit(format!("CMETHOD {transport} {proto} {addr}"));
    }

    pub fn print_cmethod_error(transport: &str, reason: &str) {
        emit(format!("CMETHOD-ERROR {transport} {reason}"));
    }

    pub fn print_cmethods_done() {
        emit("CMETHODS DONE".to_string());
    }

    fn emit(line: String) {
        let mut out = std::io::stdout().lock();
        let _ = writeln!(out, "{line}");
        let _ = out.flush();
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn arg_string_uname_only_when_passwd_is_nul() {
        let creds = Some(("cert=AAA;iat-mode=0".to_string(), "\0".to_string()));
        assert_eq!(arg_string_from_creds(creds), "cert=AAA;iat-mode=0");
    }

    #[test]
    fn arg_string_concat_when_passwd_nonempty() {
        let creds = Some(("cert=".to_string(), "AAA;iat-mode=0".to_string()));
        assert_eq!(arg_string_from_creds(creds), "cert=AAA;iat-mode=0");
    }

    #[test]
    fn arg_string_empty_when_no_creds() {
        assert_eq!(arg_string_from_creds(None), "");
    }

    #[test]
    fn arg_string_then_parse_yields_kv_map() {
        // 300-char value split across the two SOCKS5 fields (uname=255,
        // passwd=remainder) — the canonical case the spec is written for.
        let big = "cert=".to_string() + &"A".repeat(250);
        let tail = ";iat-mode=0".to_string();
        let creds = Some((big.clone(), tail.clone()));
        let arg_string = arg_string_from_creds(creds);
        assert_eq!(arg_string, big + &tail);

        let args = ptrs::args::Args::from_str(&arg_string).expect("parse");
        assert!(args.retrieve("cert").is_some());
        assert_eq!(args.retrieve("iat-mode").as_deref(), Some("0"));
    }

    #[tokio::test]
    async fn pt_args_auth_propagates_creds() {
        use fast_socks5::server::Authentication;
        let auth = PtArgsAuth;
        let got = auth
            .authenticate(Some(("u".to_string(), "p".to_string())))
            .await;
        assert_eq!(got, Some(("u".to_string(), "p".to_string())));
    }

    #[tokio::test]
    async fn pt_args_auth_accepts_no_creds() {
        use fast_socks5::server::Authentication;
        let auth = PtArgsAuth;
        let got = auth.authenticate(None).await;
        assert_eq!(got, Some((String::new(), String::new())));
    }

    // -- resolve_target_addr --

    #[test]
    fn resolve_target_addr_ip() {
        let addr = TargetAddr::Ip("127.0.0.1:9050".parse().unwrap());
        let resolved = resolve_target_addr(&addr).unwrap();
        assert_eq!(resolved.to_string(), "127.0.0.1:9050");
    }

    #[test]
    fn resolve_target_addr_ipv6() {
        let addr = TargetAddr::Ip("[::1]:443".parse().unwrap());
        let resolved = resolve_target_addr(&addr).unwrap();
        assert_eq!(resolved.to_string(), "[::1]:443");
    }

    #[test]
    fn resolve_target_addr_domain_fails() {
        let addr = TargetAddr::Domain("example.com".into(), 443);
        let err = resolve_target_addr(&addr);
        assert!(
            err.is_err(),
            "domain resolution should fail (PT doesn't do DNS)"
        );
    }

    // -- arg_string edge cases --

    #[test]
    fn arg_string_empty_uname_and_passwd() {
        let creds = Some((String::new(), String::new()));
        assert_eq!(arg_string_from_creds(creds), "");
    }

    #[test]
    fn arg_string_passwd_is_nul_only() {
        let creds = Some((String::new(), "\0".to_string()));
        assert_eq!(arg_string_from_creds(creds), "");
    }

    // -- establish_pt_conn (outgoing-dial seam) --
    //
    // These exercise the client outgoing-connection path
    // (`client_handle_connection` → `establish_pt_conn`) that was previously
    // unreachable in tests because it created a real `TcpStream::connect`
    // inline. The dial is now an injectable `Pin<FutureResult<_, _>>`, so a
    // `tokio::io::duplex()` half (or a deliberately-failing future) stands in
    // for the OR-port socket. We drive the *ptrs trait* surface end to end:
    // build an obfs4 client from a bridge-line arg string exactly the way
    // `client_handle_connection` does (`Args` → `ClientBuilder::options` →
    // `build`), then run the real obfs4 handshake over the duplex against a
    // matching obfs4 `Server`.

    use ptrs::ClientBuilder as _;
    use tokio::io::DuplexStream;
    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

    /// Build an obfs4 client transport from a bridge-line arg string via the
    /// same `ptrs` builder path lyrebird uses for a real SOCKS connection:
    /// parse the args, apply them through `ptrs::ClientBuilder::options`, then
    /// `build`. The obfs4 builder/transport impls are generic over the socket
    /// type, so we pin `InRW = DuplexStream` here (an in-memory stand-in for
    /// the OR-port `TcpStream`).
    fn obfs4_client_from_args(arg_string: &str) -> obfs4::Client {
        let args = ptrs::args::Args::from_str(arg_string).expect("parse bridge-line args");
        let mut builder = obfs4::ClientBuilder::default();
        <obfs4::ClientBuilder as ptrs::ClientBuilder<DuplexStream>>::options(&mut builder, &args)
            .expect("apply obfs4 args to builder");
        <obfs4::ClientBuilder as ptrs::ClientBuilder<DuplexStream>>::build(&builder)
    }

    #[tokio::test]
    async fn establish_pt_conn_obfs4_handshake_and_proxies_bytes() {
        // A fresh obfs4 server with a random identity and the bridge-line
        // arg string that a client must present to reach it.
        let server_builder = obfs4::ServerBuilder::<DuplexStream>::default();
        let arg_string = server_builder.client_params();
        let server = server_builder.build();

        // In-memory stand-in for the OR-port socket.
        let (client_side, server_side) = tokio::io::duplex(65_536);

        // Server peer: complete the obfs4 handshake, then echo one message.
        let server_task = tokio::spawn(async move {
            let mut s = server.wrap(server_side).await.expect("server handshake");
            let mut buf = [0u8; 64];
            let n = s.read(&mut buf).await.expect("server read");
            s.write_all(&buf[..n]).await.expect("server echo write");
            s.flush().await.expect("server flush");
        });

        // Client: build through the ptrs trait, then drive the seam with a
        // dial future that yields the duplex half instead of a TcpStream.
        let client = obfs4_client_from_args(&arg_string);
        let dial: Pin<ptrs::FutureResult<DuplexStream, std::io::Error>> =
            Box::pin(async move { Ok(client_side) });
        let client_addr: SocketAddr = "127.0.0.1:9050".parse().unwrap();

        let mut tunnel = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            establish_pt_conn(client, dial, client_addr),
        )
        .await
        .expect("establish_pt_conn timed out")
        .expect("establish_pt_conn should complete the obfs4 handshake");

        // Bytes written into the obfs4 tunnel must come back through the
        // server echo — proving `establish` consumed the injected dial
        // stream and a real encrypted session is in place.
        let msg = b"through-the-obfs4-tunnel";
        tunnel.write_all(msg).await.expect("client write");
        tunnel.flush().await.expect("client flush");

        let mut got = vec![0u8; msg.len()];
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            tunnel.read_exact(&mut got),
        )
        .await
        .expect("client read timed out")
        .expect("client read");
        assert_eq!(&got, msg, "data must round-trip through the obfs4 tunnel");

        tokio::time::timeout(std::time::Duration::from_secs(5), server_task)
            .await
            .expect("server task timed out")
            .expect("server task panicked");
    }

    #[tokio::test]
    async fn establish_pt_conn_dial_failure_is_error_not_panic() {
        // The dial future itself fails (e.g. OR-port connection refused).
        // `establish` must surface this as an `Err` without panicking — this
        // path is reachable from the network, so a regression to `unwrap`
        // would be a remote DoS.
        let server_builder = obfs4::ServerBuilder::<DuplexStream>::default();
        let arg_string = server_builder.client_params();
        let client = obfs4_client_from_args(&arg_string);

        let dial: Pin<ptrs::FutureResult<DuplexStream, std::io::Error>> = Box::pin(async {
            Err(std::io::Error::new(
                std::io::ErrorKind::ConnectionRefused,
                "dial refused",
            ))
        });
        let client_addr: SocketAddr = "127.0.0.1:9050".parse().unwrap();

        let result = establish_pt_conn(client, dial, client_addr).await;
        assert!(
            result.is_err(),
            "a failed dial must produce an error, not a tunnel"
        );
    }

    #[tokio::test]
    async fn establish_pt_conn_handshake_eof_is_error_not_panic() {
        // The dial succeeds (a socket is produced) but the OR-port peer
        // immediately closes — e.g. the bridge dropped the connection during
        // the handshake. The obfs4 client's first handshake read then hits
        // EOF, which must surface as an `Err` from `establish_pt_conn`,
        // promptly and without panicking. (The client returns `UnexpectedEof`
        // on a 0-byte read rather than blocking until its handshake timeout.)
        // This complements the positive end-to-end test: that one proves the
        // bridge-line args reach the handshake crypto (a wrong/empty cert
        // would make it fail); this one proves the failure branch is handled.
        let server_builder = obfs4::ServerBuilder::<DuplexStream>::default();
        let arg_string = server_builder.client_params();
        let client = obfs4_client_from_args(&arg_string);

        let (client_side, server_side) = tokio::io::duplex(65_536);
        // Close the peer half before the handshake reads anything.
        drop(server_side);

        let dial: Pin<ptrs::FutureResult<DuplexStream, std::io::Error>> =
            Box::pin(async move { Ok(client_side) });
        let client_addr: SocketAddr = "127.0.0.1:9050".parse().unwrap();

        let result = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            establish_pt_conn(client, dial, client_addr),
        )
        .await
        .expect("establish_pt_conn should fail fast on EOF, not block until timeout");

        assert!(
            result.is_err(),
            "a peer that closes mid-handshake must produce an error, not a tunnel"
        );
    }

    // -- Bug 2 regression: cancel arm must break the accept loop --

    /// Verify that `client_accept_loop` exits promptly when the
    /// `CancellationToken` is already cancelled before the loop starts.
    /// Without the `break` in the cancel arm, `cancelled()` is
    /// immediately ready on every iteration and the loop spins forever,
    /// causing this test to hit the timeout and fail.
    #[tokio::test]
    async fn client_accept_loop_exits_on_pre_cancelled_token() {
        // Bind a real listener so the function signature is satisfied.
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind listener for test");

        // Cancel BEFORE entering the loop — simulates shutdown race.
        let cancel = CancellationToken::new();
        cancel.cancel();

        let builder = Obfs4PT::client_builder();
        let proxy_uri = url::Url::parse("data:,").expect("placeholder url");

        // The loop must return within 2 s.  On the old code (no break)
        // it would spin indefinitely and the timeout would fire.
        let result = tokio::time::timeout(
            std::time::Duration::from_secs(2),
            client_accept_loop(listener, builder, proxy_uri, cancel),
        )
        .await;

        assert!(
            result.is_ok(),
            "client_accept_loop must exit promptly when the token is cancelled, not spin"
        );
        assert!(
            result.unwrap().is_ok(),
            "client_accept_loop should return Ok(()) on graceful cancel"
        );
    }

    // -- Bridge-connection regression (fast_socks5 `execute_command`) --
    //
    // `client_handle_connection` must use fast-socks5 only to *parse* the
    // request, never to execute it. With the default `execute_command = true`,
    // `upgrade_to_socks5()` would itself open a *plain* TCP connection to the
    // bridge and send the SOCKS reply — bypassing obfs4, so no bridge could
    // ever be reached through the transport. The fix sets
    // `execute_command(false)` and has lyrebird dial obfs4 itself, replying to
    // the parent only once the tunnel is up. These tests drive the real SOCKS5
    // client protocol against `client_handle_connection` pointed at a real
    // loopback "bridge".

    /// Play the SOCKS5 client (as arti/tor would) over `parent`: user/pass
    /// auth carrying the PT arg string, then a CONNECT to `bridge`. Returns
    /// after the CONNECT request is sent; the caller asserts on the reply.
    async fn socks5_client_connect(
        parent: &mut DuplexStream,
        arg_string: &str,
        bridge: SocketAddr,
    ) {
        // greeting: VER=5, 1 method, user/pass (0x02)
        parent.write_all(&[0x05, 0x01, 0x02]).await.unwrap();
        parent.flush().await.unwrap();
        let mut sel = [0u8; 2];
        parent.read_exact(&mut sel).await.unwrap();
        assert_eq!(sel, [0x05, 0x02], "server must select user/pass auth");

        // RFC 1929 user/pass: pack the arg string into UNAME, PASSWD = single
        // NUL (the `arg_string_from_creds` "uname only" form).
        let uname = arg_string.as_bytes();
        assert!(
            uname.len() <= 255,
            "this test packs the arg string into one SOCKS field"
        );
        let mut auth = vec![0x01, uname.len() as u8];
        auth.extend_from_slice(uname);
        auth.extend_from_slice(&[0x01, 0x00]); // PLEN=1, PASSWD=0x00
        parent.write_all(&auth).await.unwrap();
        parent.flush().await.unwrap();
        let mut authresp = [0u8; 2];
        parent.read_exact(&mut authresp).await.unwrap();
        assert_eq!(authresp, [0x01, 0x00], "user/pass auth must succeed");

        // CONNECT: VER=5, CMD=1, RSV=0, ATYP=1 (IPv4), addr, port.
        let (octets, port) = match bridge {
            SocketAddr::V4(v4) => (v4.ip().octets(), v4.port()),
            SocketAddr::V6(_) => unreachable!("test binds IPv4 loopback"),
        };
        let mut req = vec![0x05, 0x01, 0x00, 0x01];
        req.extend_from_slice(&octets);
        req.extend_from_slice(&port.to_be_bytes());
        parent.write_all(&req).await.unwrap();
        parent.flush().await.unwrap();
    }

    #[tokio::test]
    async fn client_handle_connection_tunnels_through_obfs4_and_replies_itself() {
        // A real loopback "bridge" running an obfs4 *server*. `server.wrap()`
        // completes only if an obfs4 *client* handshake arrives — i.e. the
        // connection that reached the bridge was obfs4, not a plain TCP proxy.
        // On the buggy `execute_command = true` path the bridge would instead
        // receive fast-socks5's plain relay and `wrap()` would never complete,
        // failing this test.
        let server_builder = obfs4::ServerBuilder::<TcpStream>::default();
        let arg_string = server_builder.client_params();
        let server = server_builder.build();

        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let bridge_addr = listener.local_addr().unwrap();

        let bridge = tokio::spawn(async move {
            let (sock, _peer) = listener.accept().await.expect("bridge accept");
            let mut s = server.wrap(sock).await.expect("bridge obfs4 handshake");
            let mut buf = [0u8; 64];
            let n = s.read(&mut buf).await.expect("bridge read");
            s.write_all(&buf[..n]).await.expect("bridge echo");
            s.flush().await.expect("bridge flush");
        });

        // Parent (arti/tor) side over an in-memory duplex; `lyrebird_side` is
        // the connection `client_handle_connection` serves SOCKS on.
        let (mut parent, lyrebird_side) = tokio::io::duplex(65_536);
        let builder = Obfs4PT::client_builder();
        let client_addr: SocketAddr = "127.0.0.1:9050".parse().unwrap();
        let handler = tokio::spawn(client_handle_connection(
            lyrebird_side,
            builder,
            url::Url::parse("data:,").unwrap(),
            client_addr,
        ));

        socks5_client_connect(&mut parent, &arg_string, bridge_addr).await;

        // The reply is the canonical success frame lyrebird writes itself
        // (fast-socks5 sends nothing — execute_command is disabled), and it
        // arrives only because the obfs4 tunnel to the bridge came up.
        let mut reply = [0u8; 10];
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            parent.read_exact(&mut reply),
        )
        .await
        .expect("SOCKS5 reply timed out")
        .expect("read SOCKS5 reply");
        assert_eq!(
            reply,
            [0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0],
            "lyrebird must send the SOCKS5 success reply itself after the obfs4 handshake"
        );

        // End-to-end: a probe must round-trip through the obfs4 tunnel.
        let probe = b"bridge-tunnel-probe";
        parent.write_all(probe).await.unwrap();
        parent.flush().await.unwrap();
        let mut got = vec![0u8; probe.len()];
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            parent.read_exact(&mut got),
        )
        .await
        .expect("probe round-trip timed out")
        .expect("read probe echo");
        assert_eq!(
            &got, probe,
            "probe must round-trip through the obfs4 tunnel"
        );

        tokio::time::timeout(std::time::Duration::from_secs(5), bridge)
            .await
            .expect("bridge task timed out")
            .expect("bridge task panicked");
        drop(parent); // let copy_bidirectional see EOF and the handler finish
        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handler).await;
    }

    #[tokio::test]
    async fn client_handle_connection_no_success_reply_when_bridge_not_obfs4() {
        // The bridge accepts the TCP connection but is NOT an obfs4 server: it
        // closes immediately, so the obfs4 client handshake fails. lyrebird
        // must therefore NOT report CONNECT success to the parent. The old
        // `execute_command = true` path replied success on the bare TCP
        // connect regardless of whether obfs4 could be established, which is
        // exactly the bug — so this test would fail on it.
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let bridge_addr = listener.local_addr().unwrap();
        let bridge = tokio::spawn(async move {
            let (sock, _peer) = listener.accept().await.expect("bridge accept");
            drop(sock); // not obfs4 — close before any handshake byte
        });

        // A well-formed cert so `options()` succeeds and the failure is the
        // handshake, not arg parsing; this throwaway identity is never honored.
        let arg_string = obfs4::ServerBuilder::<TcpStream>::default().client_params();

        let (mut parent, lyrebird_side) = tokio::io::duplex(65_536);
        let builder = Obfs4PT::client_builder();
        let client_addr: SocketAddr = "127.0.0.1:9050".parse().unwrap();
        let handler = tokio::spawn(client_handle_connection(
            lyrebird_side,
            builder,
            url::Url::parse("data:,").unwrap(),
            client_addr,
        ));

        socks5_client_connect(&mut parent, &arg_string, bridge_addr).await;

        // No success reply: the handler returns Err before writing one, so its
        // side of the duplex drops and the parent's read hits EOF rather than a
        // 10-byte success frame.
        let mut reply = [0u8; 10];
        let read = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            parent.read_exact(&mut reply),
        )
        .await
        .expect("the read should resolve (EOF), not hang");
        assert!(
            read.is_err(),
            "no SOCKS5 success reply must be sent when the obfs4 handshake fails"
        );

        let outcome = tokio::time::timeout(std::time::Duration::from_secs(5), handler)
            .await
            .expect("handler should finish")
            .expect("handler task panicked");
        assert!(
            outcome.is_err(),
            "client_handle_connection must surface the failed obfs4 dial as an error"
        );

        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), bridge).await;
    }
}