review-003 N-8: the worthwhile residue probed — the BIND Domain-reply wire helper + table, the refusal-recode unit pins, the serve() wrappers, the len/is_empty accessors, the bridge later-batch garbage

src/door.rs

@@ -639,7 +639,23 @@ async fn pipe_bind_plane(
         session.close().await;
         return DoorExit::PlaneEnded;
     };
-    let wire = match &bind_addr {
+    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![
@@ -653,13 +669,7 @@ async fn pipe_bind_plane(
             wire.extend_from_slice(&port.to_be_bytes());
             wire
         }
-    };
-    if write_vanilla_reply(&mut stream, &wire).await.is_err() {
-        session.close().await;
-        return DoorExit::ClientGone;
     }
-
-    bind_quiet_and_pipe(stream, session).await
 }
 
 /// The BIND door's quiet wait + reply#2 + the data plane: await the
@@ -1178,3 +1188,149 @@ mod tests {
         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; 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 — 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 — 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 — 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 — 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:?}"
+            );
+        }
+    }
+}