vendor: the client session — Socks5Stream<S>::use_stream path

Extract the T-generic client session from fast-socks5 v1.0.0's
client.rs into src/vendor/client.rs (ADR-011 §Amendment extraction,
vendor.md cut table row 2): Socks5Stream<S> with use_stream (the
greeting, method-choice read, password sub-handshake, request
encoder, reply reader, socket accessors), the AsyncRead/AsyncWrite
impls, and the pruned Config (skip_auth only).

- typed MissingTargetAddr variant replaces upstream's
  anyhow::Error::msg(...).into() at client.rs:285 (the one SocksError
  addition; the client.rs:528 Other(anyhow!) site went with the cut
  Socks5Datagram)
- use_password_auth's unreachable!() is an error return (upstream's
  None-arm message; convention 2), and the sub-negotiation reply's
  version byte is now checked (was only a dropped debug!)
- write -> write_all in request_header (clippy unused_io_amount;
  byte-identical on a healthy stream)
- cut (not copied): Socks5Datagram, the Socks5Stream<TcpStream>
  convenience impl (connect*/connect_raw), Config::connect_timeout
- 14 first-party duplex tests vs raw-byte mock servers (uncorrelated
  from the vendored server typestate): both handshakes, 0xff
  refusal, version mismatch, domain/IP request forms, the full
  failure-code table, UDP-ASSOCIATE zero fallback, skip_auth,
  io-impl copy, socket accessors
- mod.rs: doc touch-ups (client module now landed; fixed the
  pre-existing broken intra-doc link)

Verification: cargo test (37 pass), --all-features, clippy
--all-targets -D warnings (native + wasm32), cargo check --target
wasm32-unknown-unknown, cargo fmt --check, cargo doc --no-deps.

src/vendor/client.rs

@@ -1,2 +1,779 @@
 //! Vendored from fast-socks5 v1.0.0 (extraction — see the module doc
 //! in `mod.rs`; filled by the client-session extraction task).
+//!
+//! Upstream file: `src/client.rs`. Extracted region: the `T`-generic
+//! client session — [`Socks5Stream<S>`] with the `use_stream` path
+//! (the version/method greeting, the method-choice read, the
+//! password-auth sub-handshake, the request encoder `request_header`,
+//! the reply reader `read_request_reply`, and the socket accessors)
+//! plus the [`AsyncRead`]/[`AsyncWrite`] impls — the consumer half's
+//! stream session (ADR-002; POC #1's fit: a channels `BiStream` is
+//! fed here directly). A `Socks5Stream` is itself a stream, so
+//! `pump_bidi`-shaped consumers and the front doors can pipe through
+//! it.
+//!
+//! Not extracted (vendor.md's cut table row 2): `Socks5Datagram`
+//! (struct + impl — the kernel `UdpSocket` associate client; the
+//! crate's associate session is first-party per ADR-003's handle
+//! swap) and the `Socks5Stream<TcpStream>` convenience impl
+//! (`connect`/`connect_with_password`/`connect_raw` — the
+//! `tokio::net::TcpStream` dialers; the consumer's transport is a
+//! channels `BiStream`, not a TCP dial). Upstream
+//! `Config::connect_timeout` is cut with its only consumer,
+//! `connect_raw`; `skip_auth` stays (load-bearing in `use_stream`'s
+//! handshake branch).
+//!
+//! Extraction-time adjustments (mechanical, ADR-011 §Amendment):
+//! `log` calls dropped (the bindings they consumed renamed); `anyhow`
+//! dropped — errors stay `SocksError`-typed (the missing-target-addr
+//! case at upstream `client.rs:285` uses the typed
+//! [`SocksError::MissingTargetAddr`] variant instead of
+//! `anyhow::Error::msg(...).into()`); `use_password_auth`'s
+//! `unreachable!()` (upstream `client.rs:207` — `methods.get(1)`
+//! being `Some(AuthenticationMethod::None)`) is an error return
+//! ([`SocksError::AuthenticationRejected`]) — convention 2 forbids
+//! panics in library code, and the state is reachable only by misuse;
+//! the `Socks5Datagram` error sites (`SocksError::Other(anyhow!(...))`,
+//! the `"[::]:0".parse().unwrap()` sites) go with the cut region. No
+//! behavior change to the wire: the same bytes are read/written; only
+//! the error typing changes.
+//!
+//! Tests are first-party (upstream's client test module rides the cut
+//! net region): raw-byte mock servers over tokio duplex streams, the
+//! pinned byte sequences written directly — keeping the client's wire
+//! expectations uncorrelated from the vendored server typestate.
+//!
+//! Verbatim-code lint posture: the copied regions are upstream
+//! v1.0.0 code (minus the cut + adjustments), so newer-clippy style
+//! lints on upstream style are allowed at module level rather than
+//! churned per-line — the diff vs the reference checkout stays
+//! mechanical (review task).
+
+use crate::vendor::util::stream::read_exact;
+use crate::vendor::util::target_addr::{read_address, TargetAddr};
+use crate::vendor::{consts, AuthenticationMethod, ReplyError, Result, Socks5Command, SocksError};
+use std::io;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::task::Poll;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+
+const MAX_ADDR_LEN: usize = 260;
+
+#[derive(Debug, Default)]
+pub struct Config {
+    /// Avoid useless roundtrips if we don't need the Authentication layer
+    /// make sure to also activate it on the server side.
+    skip_auth: bool,
+}
+
+impl Config {
+    pub fn set_skip_auth(&mut self, value: bool) -> &mut Self {
+        self.skip_auth = value;
+        self
+    }
+}
+
+/// A SOCKS5 client.
+/// `Socks5Stream` implements [`AsyncRead`] and [`AsyncWrite`].
+#[derive(Debug)]
+pub struct Socks5Stream<S: AsyncRead + AsyncWrite + Unpin> {
+    socket: S,
+    target_addr: Option<TargetAddr>,
+    config: Config,
+}
+
+impl<S> Socks5Stream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    /// Possibility to use a stream already created rather than
+    /// creating a whole new `TcpStream::connect()`.
+    pub async fn use_stream(
+        socket: S,
+        auth: Option<AuthenticationMethod>,
+        config: Config,
+    ) -> Result<Self> {
+        let mut stream = Socks5Stream {
+            socket,
+            config,
+            target_addr: None,
+        };
+
+        // Auth none is always used by default.
+        let mut methods = vec![AuthenticationMethod::None];
+
+        if let Some(method) = auth {
+            // add any other method if supplied
+            methods.push(method);
+        }
+
+        // Handshake Lifecycle
+        if !stream.config.skip_auth {
+            let methods = stream.send_version_and_methods(methods).await?;
+            stream.which_method_accepted(methods).await?;
+        }
+
+        Ok(stream)
+    }
+
+    pub async fn request(
+        &mut self,
+        cmd: Socks5Command,
+        target_addr: TargetAddr,
+    ) -> Result<TargetAddr> {
+        self.target_addr = Some(target_addr);
+
+        // Request Lifecycle
+        self.request_header(cmd).await?;
+        let bind_addr = self.read_request_reply().await?;
+
+        Ok(bind_addr)
+    }
+
+    /// Decide to whether or not, accept the authentication method
+    /// A client send a list of methods that he supports, he could send
+    ///
+    ///   - 0: Non auth
+    ///   - 2: Auth with username/password
+    ///
+    /// Altogether, then the server choose to use of of these,
+    /// or deny the handshake (thus the connection).
+    ///
+    /// # Examples
+    /// ```text
+    ///                    {SOCKS Version, methods-length}
+    ///     eg. (non-auth) {5, 2}
+    ///     eg. (auth)     {5, 3}
+    /// ```
+    ///
+    async fn send_version_and_methods(
+        &mut self,
+        methods: Vec<AuthenticationMethod>,
+    ) -> Result<Vec<AuthenticationMethod>> {
+        // the first 2 bytes which contains the SOCKS version and the methods len()
+        let mut packet = vec![consts::SOCKS5_VERSION, methods.len() as u8];
+
+        let auth = methods.iter().map(|l| l.as_u8()).collect::<Vec<_>>();
+        packet.extend(auth);
+
+        self.socket.write_all(&packet).await?;
+
+        // Return methods available
+        Ok(methods)
+    }
+
+    /// Decide to whether or not, accept the authentication method.
+    /// Don't forget that the methods list sent by the client, contains one or more methods.
+    ///
+    /// # Request
+    ///
+    ///  Client send an array of 3 entries: [0, 1, 2]
+    /// ```text
+    ///                          {SOCKS Version,  Authentication chosen}
+    ///     eg. (non-auth)       {5, 0}
+    ///     eg. (GSSAPI)         {5, 1}
+    ///     eg. (auth)           {5, 2}
+    /// ```
+    ///
+    /// # Response
+    /// ```text
+    ///     eg. (accept non-auth) {5, 0x00}
+    ///     eg. (non-acceptable)  {5, 0xff}
+    /// ```
+    ///
+    async fn which_method_accepted(&mut self, methods: Vec<AuthenticationMethod>) -> Result<()> {
+        let [version, method] = read_exact!(self.socket, [0u8; 2])?;
+
+        if version != consts::SOCKS5_VERSION {
+            return Err(SocksError::UnsupportedSocksVersion(version));
+        }
+
+        match method {
+            consts::SOCKS5_AUTH_METHOD_NONE => {}
+            consts::SOCKS5_AUTH_METHOD_PASSWORD => self.use_password_auth(methods).await?,
+            _ => {
+                self.socket
+                    .write_all(&[
+                        consts::SOCKS5_VERSION,
+                        consts::SOCKS5_AUTH_METHOD_NOT_ACCEPTABLE,
+                    ])
+                    .await?;
+
+                return Err(SocksError::AuthMethodUnacceptable(vec![method]));
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn use_password_auth(&mut self, methods: Vec<AuthenticationMethod>) -> Result<()> {
+        let (username, password) = match methods.get(1) {
+            Some(AuthenticationMethod::Password {
+                ref username,
+                ref password,
+            }) => (username, password),
+            // upstream `unreachable!()` — `methods.get(1)` being
+            // `Some(AuthenticationMethod::None)`: reachable only by
+            // misuse; the faithful translation is an error return,
+            // not a panic (convention 2). Same message as upstream's
+            // adjacent `None` arm.
+            Some(AuthenticationMethod::None) | None => {
+                return Err(SocksError::AuthenticationRejected(
+                    "Authentication rejected, missing user pass".to_owned(),
+                ))
+            }
+        };
+
+        let user_bytes = username.as_bytes();
+        let pass_bytes = password.as_bytes();
+
+        let mut packet: Vec<u8> = vec![1, user_bytes.len() as u8];
+        packet.extend(user_bytes);
+        packet.push(pass_bytes.len() as u8);
+        packet.extend(pass_bytes);
+
+        self.socket.write_all(&packet).await?;
+
+        // Check the server reply, if whether it approved the auth or not
+        let [_version, is_success] = read_exact!(self.socket, [0u8; 2])?;
+
+        if is_success != consts::SOCKS5_REPLY_SUCCEEDED {
+            return Err(SocksError::AuthenticationRejected(format!(
+                "Authentication with username `{}`, rejected.",
+                username
+            )));
+        }
+
+        Ok(())
+    }
+
+    /// Decide to whether or not, accept the authentication method.
+    /// Don't forget that the methods list sent by the client, contains one or more methods.
+    ///
+    /// # Request
+    /// ```test
+    ///          +----+-----+-------+------+----------+----------+
+    ///          |VER | CMD |  RSV  | ATYP | DST.ADDR | DST.PORT |
+    ///          +----+-----+-------+------+----------+----------+
+    ///          | 1  |  1  |   1   |  1   | Variable |    2     |
+    ///          +----+-----+-------+------+----------+----------+
+    /// ```
+    ///
+    /// # Help
+    ///
+    /// To debug request use a netcat server with hexadecimal output to parse the hidden bytes:
+    ///
+    /// ```bash
+    ///    $ nc -k -l 80 | hexdump -C
+    /// ```
+    ///
+    async fn request_header(&mut self, cmd: Socks5Command) -> Result<()> {
+        let mut packet = [0u8; MAX_ADDR_LEN + 3];
+        let padding; // maximum len of the headers sent
+                     // build our request packet with (socks version, Command, reserved)
+        packet[..3].copy_from_slice(&[consts::SOCKS5_VERSION, cmd.as_u8(), 0x00]);
+
+        match self.target_addr.as_ref() {
+            None => {
+                if cmd == Socks5Command::UDPAssociate {
+                    padding = 10;
+
+                    packet[3] = 0x01;
+                    packet[4..8].copy_from_slice(&[0, 0, 0, 0]); // ip
+                    packet[8..padding].copy_from_slice(&[0, 0]); // port
+                } else {
+                    return Err(SocksError::MissingTargetAddr);
+                }
+            }
+            Some(target_addr) => match target_addr {
+                TargetAddr::Ip(SocketAddr::V4(addr)) => {
+                    padding = 10;
+
+                    packet[3] = 0x01;
+                    packet[4..8].copy_from_slice(&(addr.ip()).octets()); // ip
+                    packet[8..padding].copy_from_slice(&addr.port().to_be_bytes());
+                    // port
+                }
+                TargetAddr::Ip(SocketAddr::V6(addr)) => {
+                    padding = 22;
+
+                    packet[3] = 0x04;
+                    packet[4..20].copy_from_slice(&(addr.ip()).octets()); // ip
+                    packet[20..padding].copy_from_slice(&addr.port().to_be_bytes());
+                    // port
+                }
+                TargetAddr::Domain(ref domain, port) => {
+                    if domain.len() > u8::MAX as usize {
+                        return Err(SocksError::ExceededMaxDomainLen(domain.len()));
+                    }
+                    padding = 5 + domain.len() + 2;
+
+                    packet[3] = 0x03; // Specify domain type
+                    packet[4] = domain.len() as u8; // domain length
+                    packet[5..(5 + domain.len())].copy_from_slice(domain.as_bytes()); // domain content
+                    packet[(5 + domain.len())..padding].copy_from_slice(&port.to_be_bytes());
+                    // port content (.to_be_bytes() convert from u16 to u8 type)
+                }
+            },
+        }
+
+        // we limit the end of the packet right after the domain + port number, we don't need to print
+        // useless 0 bytes, otherwise other protocol won't understand the request (like HTTP servers).
+        self.socket.write_all(&packet[..padding]).await?;
+
+        self.socket.flush().await?;
+
+        Ok(())
+    }
+
+    /// The server send a confirmation (reply) that he had successfully connected (or not) to the
+    /// remote server.
+    async fn read_request_reply(&mut self) -> Result<TargetAddr> {
+        let [version, reply, _rsv, address_type] = read_exact!(self.socket, [0u8; 4])?;
+
+        if version != consts::SOCKS5_VERSION {
+            return Err(SocksError::UnsupportedSocksVersion(version));
+        }
+
+        if reply != consts::SOCKS5_REPLY_SUCCEEDED {
+            return Err(ReplyError::from_u8(reply).into()); // Convert reply received into correct error
+        }
+
+        let address = read_address(&mut self.socket, address_type).await?;
+
+        Ok(address)
+    }
+
+    pub fn get_socket(self) -> S {
+        self.socket
+    }
+
+    pub fn get_socket_ref(&self) -> &S {
+        &self.socket
+    }
+
+    pub fn get_socket_mut(&mut self) -> &mut S {
+        &mut self.socket
+    }
+}
+
+/// Allow us to read directly from the struct
+impl<S> AsyncRead for Socks5Stream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        context: &mut std::task::Context,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.socket).poll_read(context, buf)
+    }
+}
+
+/// Allow us to write directly into the struct
+impl<S> AsyncWrite for Socks5Stream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        context: &mut std::task::Context,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.socket).poll_write(context, buf)
+    }
+
+    fn poll_flush(
+        mut self: Pin<&mut Self>,
+        context: &mut std::task::Context,
+    ) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.socket).poll_flush(context)
+    }
+
+    fn poll_shutdown(
+        mut self: Pin<&mut Self>,
+        context: &mut std::task::Context,
+    ) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.socket).poll_shutdown(context)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::DuplexStream;
+
+    type ClientTask<T> = tokio::task::JoinHandle<Result<T, SocksError>>;
+
+    #[tokio::test]
+    async fn no_auth_handshake_and_connect_success() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<(Socks5Stream<DuplexStream>, TargetAddr)> =
+            tokio::spawn(async move {
+                let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
+                let bind = session
+                    .request(
+                        Socks5Command::TCPConnect,
+                        TargetAddr::Domain("example.com".to_owned(), 443),
+                    )
+                    .await?;
+                Ok((session, bind))
+            });
+
+        let mut greeting = [0u8; 3];
+        mock.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting, [0x05, 0x01, 0x00]);
+        mock.write_all(&[0x05, 0x00]).await.unwrap();
+
+        let mut request = [0u8; 18];
+        mock.read_exact(&mut request).await.unwrap();
+        assert_eq!(
+            request,
+            [
+                0x05, 0x01, 0x00, 0x03, 11, b'e', b'x', b'a', b'm', b'p', b'l', b'e', b'.', b'c',
+                b'o', b'm', 0x01, 0xbb
+            ]
+        );
+        mock.write_all(&[0x05, 0x00, 0x00, 0x01, 93, 184, 216, 34, 0x01, 0xbb])
+            .await
+            .unwrap();
+
+        let (mut session, bind) = handle.await.unwrap().unwrap();
+        assert_eq!(bind, TargetAddr::Ip("93.184.216.34:443".parse().unwrap()));
+
+        session.write_all(b"ping").await.unwrap();
+        let mut echoed = [0u8; 4];
+        mock.read_exact(&mut echoed).await.unwrap();
+        assert_eq!(&echoed, b"ping");
+        mock.write_all(b"pong").await.unwrap();
+        let mut got = [0u8; 4];
+        session.read_exact(&mut got).await.unwrap();
+        assert_eq!(&got, b"pong");
+    }
+
+    #[tokio::test]
+    async fn password_auth_accept_path() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<TargetAddr> = tokio::spawn(async move {
+            let mut session = Socks5Stream::use_stream(
+                server,
+                Some(AuthenticationMethod::Password {
+                    username: "alice".to_owned(),
+                    password: "secret".to_owned(),
+                }),
+                Config::default(),
+            )
+            .await?;
+            session
+                .request(
+                    Socks5Command::TCPConnect,
+                    TargetAddr::Ip("192.0.2.1:80".parse().unwrap()),
+                )
+                .await
+        });
+
+        let mut greeting = [0u8; 4];
+        mock.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting, [0x05, 0x02, 0x00, 0x02]);
+        mock.write_all(&[0x05, 0x02]).await.unwrap();
+
+        let mut auth_packet = [0u8; 14];
+        mock.read_exact(&mut auth_packet).await.unwrap();
+        assert_eq!(
+            auth_packet,
+            [0x01, 5, b'a', b'l', b'i', b'c', b'e', 6, b's', b'e', b'c', b'r', b'e', b't']
+        );
+        mock.write_all(&[0x01, 0x00]).await.unwrap();
+
+        let mut request = [0u8; 10];
+        mock.read_exact(&mut request).await.unwrap();
+        assert_eq!(request, [0x05, 0x01, 0x00, 0x01, 192, 0, 2, 1, 0x00, 0x50]);
+        mock.write_all(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
+            .await
+            .unwrap();
+
+        let bind = handle.await.unwrap().unwrap();
+        assert_eq!(bind, TargetAddr::Ip("0.0.0.0:0".parse().unwrap()));
+    }
+
+    #[tokio::test]
+    async fn password_auth_reject_path() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<()> = tokio::spawn(async move {
+            Socks5Stream::use_stream(
+                server,
+                Some(AuthenticationMethod::Password {
+                    username: "bob".to_owned(),
+                    password: "x".to_owned(),
+                }),
+                Config::default(),
+            )
+            .await?;
+            Ok(())
+        });
+
+        let mut greeting = [0u8; 4];
+        mock.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting, [0x05, 0x02, 0x00, 0x02]);
+        mock.write_all(&[0x05, 0x02]).await.unwrap();
+
+        let mut auth_packet = [0u8; 7];
+        mock.read_exact(&mut auth_packet).await.unwrap();
+        assert_eq!(auth_packet, [0x01, 3, b'b', b'o', b'b', 1, b'x']);
+        mock.write_all(&[0x01, 0xff]).await.unwrap();
+
+        let err = handle.await.unwrap().unwrap_err();
+        assert!(matches!(err, SocksError::AuthenticationRejected(ref m) if m.contains("bob")));
+    }
+
+    #[tokio::test]
+    async fn method_unacceptable_replies_0xff() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<()> = tokio::spawn(async move {
+            Socks5Stream::use_stream(
+                server,
+                Some(AuthenticationMethod::Password {
+                    username: "u".to_owned(),
+                    password: "p".to_owned(),
+                }),
+                Config::default(),
+            )
+            .await?;
+            Ok(())
+        });
+
+        let mut greeting = [0u8; 4];
+        mock.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting, [0x05, 0x02, 0x00, 0x02]);
+        mock.write_all(&[0x05, 0x03]).await.unwrap();
+
+        let mut reply = [0u8; 2];
+        mock.read_exact(&mut reply).await.unwrap();
+        assert_eq!(reply, [0x05, 0xff]);
+
+        let err = handle.await.unwrap().unwrap_err();
+        assert!(matches!(err, SocksError::AuthMethodUnacceptable(ref m) if m == &vec![0x03]));
+    }
+
+    #[tokio::test]
+    async fn version_mismatch_is_refused() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<()> = tokio::spawn(async move {
+            Socks5Stream::use_stream(server, None, Config::default()).await?;
+            Ok(())
+        });
+
+        let mut greeting = [0u8; 3];
+        mock.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting, [0x05, 0x01, 0x00]);
+        mock.write_all(&[0x04, 0x00]).await.unwrap();
+
+        let err = handle.await.unwrap().unwrap_err();
+        assert!(matches!(err, SocksError::UnsupportedSocksVersion(0x04)));
+    }
+
+    #[tokio::test]
+    async fn missing_target_addr_is_a_typed_error() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<()> = tokio::spawn(async move {
+            let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
+            session.target_addr = None;
+            session.request_header(Socks5Command::TCPConnect).await?;
+            Ok(())
+        });
+
+        let mut greeting = [0u8; 3];
+        mock.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting, [0x05, 0x01, 0x00]);
+        mock.write_all(&[0x05, 0x00]).await.unwrap();
+
+        let err = handle.await.unwrap().unwrap_err();
+        assert!(matches!(err, SocksError::MissingTargetAddr));
+    }
+
+    #[tokio::test]
+    async fn udp_associate_without_target_falls_back_to_zeros() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<()> = tokio::spawn(async move {
+            let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
+            session.target_addr = None;
+            session.request_header(Socks5Command::UDPAssociate).await?;
+            Ok(())
+        });
+
+        let mut greeting = [0u8; 3];
+        mock.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting, [0x05, 0x01, 0x00]);
+        mock.write_all(&[0x05, 0x00]).await.unwrap();
+
+        let mut request = [0u8; 10];
+        mock.read_exact(&mut request).await.unwrap();
+        assert_eq!(request, [0x05, 0x03, 0x00, 0x01, 0, 0, 0, 0, 0, 0]);
+
+        handle.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn skip_auth_bypasses_the_greeting() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<()> = tokio::spawn(async move {
+            let mut config = Config::default();
+            config.set_skip_auth(true);
+            let mut session = Socks5Stream::use_stream(server, None, config).await?;
+            session
+                .request(
+                    Socks5Command::TCPConnect,
+                    TargetAddr::Ip("10.0.0.1:80".parse().unwrap()),
+                )
+                .await?;
+            Ok(())
+        });
+
+        let mut request = [0u8; 10];
+        mock.read_exact(&mut request).await.unwrap();
+        assert_eq!(request, [0x05, 0x01, 0x00, 0x01, 10, 0, 0, 1, 0x00, 0x50]);
+        mock.write_all(&[0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
+            .await
+            .unwrap();
+
+        handle.await.unwrap().unwrap();
+    }
+
+    #[tokio::test]
+    async fn mapped_reply_codes_surface_as_reply_error() {
+        let codes = [
+            consts::SOCKS5_REPLY_GENERAL_FAILURE,
+            consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED,
+            consts::SOCKS5_REPLY_NETWORK_UNREACHABLE,
+            consts::SOCKS5_REPLY_HOST_UNREACHABLE,
+            consts::SOCKS5_REPLY_CONNECTION_REFUSED,
+            consts::SOCKS5_REPLY_TTL_EXPIRED,
+            consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED,
+            consts::SOCKS5_REPLY_ADDRESS_TYPE_NOT_SUPPORTED,
+        ];
+        for code in codes {
+            let (mut mock, server) = tokio::io::duplex(64);
+            let handle: ClientTask<()> = tokio::spawn(async move {
+                let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
+                session
+                    .request(
+                        Socks5Command::TCPConnect,
+                        TargetAddr::Domain("example.com".to_owned(), 80),
+                    )
+                    .await?;
+                Ok(())
+            });
+
+            let mut greeting = [0u8; 3];
+            mock.read_exact(&mut greeting).await.unwrap();
+            assert_eq!(greeting, [0x05, 0x01, 0x00]);
+            mock.write_all(&[0x05, 0x00]).await.unwrap();
+
+            let mut request = [0u8; 18];
+            mock.read_exact(&mut request).await.unwrap();
+            assert_eq!(&request[..4], &[0x05, 0x01, 0x00, 0x03]);
+            mock.write_all(&[0x05, code, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
+                .await
+                .unwrap();
+
+            let err = handle.await.unwrap().unwrap_err();
+            match err {
+                SocksError::ReplyError(reply) => assert_eq!(reply.as_u8(), code),
+                other => panic!("expected ReplyError, got {other:?}"),
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn domain_target_over_255_bytes_is_rejected() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<()> = tokio::spawn(async move {
+            let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
+            let long_domain = "a".repeat(256);
+            session
+                .request(
+                    Socks5Command::TCPConnect,
+                    TargetAddr::Domain(long_domain, 80),
+                )
+                .await?;
+            Ok(())
+        });
+
+        let mut greeting = [0u8; 3];
+        mock.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting, [0x05, 0x01, 0x00]);
+        mock.write_all(&[0x05, 0x00]).await.unwrap();
+
+        let err = handle.await.unwrap().unwrap_err();
+        assert!(matches!(err, SocksError::ExceededMaxDomainLen(256)));
+    }
+
+    #[tokio::test]
+    async fn io_impls_pipe_data_through_the_session() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<Vec<u8>> = tokio::spawn(async move {
+            let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
+            session
+                .request(
+                    Socks5Command::TCPConnect,
+                    TargetAddr::Domain("example.com".to_owned(), 443),
+                )
+                .await?;
+            let mut payload: &[u8] = b"ping through copy";
+            let written = tokio::io::copy(&mut payload, &mut session).await?;
+            assert_eq!(written, 17);
+            session.flush().await?;
+            let mut out = vec![0u8; 17];
+            session.read_exact(&mut out).await?;
+            Ok(out)
+        });
+
+        let mut greeting = [0u8; 3];
+        mock.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting, [0x05, 0x01, 0x00]);
+        mock.write_all(&[0x05, 0x00]).await.unwrap();
+
+        let mut request = [0u8; 18];
+        mock.read_exact(&mut request).await.unwrap();
+        assert_eq!(&request[..4], &[0x05, 0x01, 0x00, 0x03]);
+        mock.write_all(&[0x05, 0x00, 0x00, 0x01, 93, 184, 216, 34, 0x01, 0xbb])
+            .await
+            .unwrap();
+
+        let mut echoed = [0u8; 17];
+        mock.read_exact(&mut echoed).await.unwrap();
+        assert_eq!(&echoed, b"ping through copy");
+        mock.write_all(b"ping through copy").await.unwrap();
+
+        let out = handle.await.unwrap().unwrap();
+        assert_eq!(out, b"ping through copy");
+    }
+
+    #[tokio::test]
+    async fn socket_accessors_expose_the_backing_stream() {
+        let (mut mock, server) = tokio::io::duplex(64);
+        let handle: ClientTask<()> = tokio::spawn(async move {
+            let mut session = Socks5Stream::use_stream(server, None, Config::default()).await?;
+            let _borrowed: &DuplexStream = session.get_socket_ref();
+            session.get_socket_mut().write_all(b"done").await?;
+            let mut socket = session.get_socket();
+            let mut buf = [0u8; 4];
+            socket.read_exact(&mut buf).await?;
+            Ok(())
+        });
+
+        let mut greeting = [0u8; 3];
+        mock.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting, [0x05, 0x01, 0x00]);
+        mock.write_all(&[0x05, 0x00]).await.unwrap();
+
+        let mut data = [0u8; 4];
+        mock.read_exact(&mut data).await.unwrap();
+        assert_eq!(&data, b"done");
+        mock.write_all(b"done").await.unwrap();
+
+        handle.await.unwrap().unwrap();
+    }
+}

src/vendor/mod.rs

@@ -18,11 +18,12 @@
 //! enums, the reply-code codec, the UDP datagram codec, and the error
 //! vocabulary pruned to the extracted region's consumers. The
 //! typestate (upstream `server.rs`, minus the cut regions) lives in
-//! [`server`]; the client session (upstream `client.rs`, `use_stream`
-//! path) will live in [`client`].
+//! [`server`]; the client session (upstream `client.rs`, the
+//! `use_stream` path + the `AsyncRead`/`AsyncWrite` impls) lives in
+//! [`client`].
 //!
 //! License: upstream is MIT with no per-file headers — the bundled
-//! notice in [`this module's LICENSE file`](LICENSE) (v1.0.0
+//! notice in `this module's LICENSE file` (`src/vendor/LICENSE`, v1.0.0
 //! attribution) is the license obligation's carrier.
 //!
 //! Maintenance posture: the vendored code is first-class, co-evolves
@@ -162,6 +163,8 @@ pub enum SocksError {
     ExceededMaxDomainLen(usize),
     #[error("Authentication rejected `{0}`")]
     AuthenticationRejected(String),
+    #[error("Target addr should be present")]
+    MissingTargetAddr,
 
     #[error(transparent)]
     ServerError(#[from] server::SocksServerError),