diff --git a/src/consumer.rs b/src/consumer.rs index 841319a..6013f82 100644 --- a/src/consumer.rs +++ b/src/consumer.rs @@ -1158,6 +1158,343 @@ mod tests { )); } + #[tokio::test] + async fn send_datagram_on_a_spent_stream_is_the_io_arm() { + // The spent-stream `Io` arm (review 002 N-8): a session whose + // stream is gone errors loud through the typed I/O surface — + // the mirror of the covered spent-`recv` arm. + let mut session = AssociateSession { + channel_id: 9, + manager: None, + inner: None, + reader: DatagramReader::new(), + pending: Default::default(), + relay_addr: None, + }; + let err = session + .send_datagram(TargetAddr::Ip("10.0.0.1:53".parse().unwrap()), b"payload") + .await + .expect_err("a spent stream must not send"); + assert!( + matches!(err, SocksSessionError::Io(_)), + "the spent-stream send is the Io arm: {err:?}" + ); + + // The spent-session read shape rides the same posture: the + // `recv_datagram` spent arm is `Ok(None)` (the EOF-shape, the + // family's read-side posture). + assert!( + session.recv_datagram().await.expect("spent recv").is_none(), + "a spent stream reads the spent shape, never an error" + ); + + // The spent-session teardown surfaces: `shutdown` on a spent + // session is a no-op reap (no manager — nothing reaped). + let shutdown = session.shutdown().await.expect("spent shutdown"); + assert!(!shutdown, "no manager — nothing reaped"); + } + + #[tokio::test] + async fn spent_associate_close_reaps_a_no_op() { + // The `close`-on-spent arm: the graceful teardown path is a + // no-op reap on a spent session (the session family's Drop + // posture, on the async path). + let session = AssociateSession { + channel_id: 9, + manager: None, + inner: None, + reader: DatagramReader::new(), + pending: Default::default(), + relay_addr: None, + }; + assert!(!session.close().await, "no manager — nothing reaped"); + } + + #[tokio::test] + async fn recv_datagram_mid_batch_later_frame_err_drops_the_pending_oks() { + use tokio::io::AsyncWriteExt as _; + + // The mid-batch later-frame `Err` arm (review 002 N-8): the + // first frame of a batch decodes `Ok`, a later one is garbage + // — the batch is discarded with the session (fail-loud, never + // a partial batch delivered while the session stays open). + let (mut peer_writer, session_reader) = tokio::io::duplex(1024); + let addr = TargetAddr::Ip("10.0.0.1:53".parse().unwrap()); + let mut wire = crate::wire::frame(&addr, b"first").unwrap().to_vec(); + wire.extend_from_slice(&[0x00, 0x04, 0x7f, 1, 2, 3]); + peer_writer.write_all(&wire).await.unwrap(); + drop(peer_writer); + + let (write_peer, session_writer) = tokio::io::duplex(64); + let _ = write_peer; + + let mut session = AssociateSession { + channel_id: 1, + manager: None, + inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)), + reader: DatagramReader::new(), + pending: Default::default(), + relay_addr: None, + }; + let err = session + .recv_datagram() + .await + .expect_err("the later frame's garbage fails loud"); + assert!( + matches!(err, SocksSessionError::Rfc(SocksError::UdpHeaderError(_))), + "the decode garbage rides the UDP-header class: {err:?}" + ); + + // The pending `Ok` was dropped with the session — the queue + // stays empty and the stream is spent for further reads (the + // error ended the caller's loop, the fail-loud posture). + assert!(session.pending.is_empty()); + } + + #[tokio::test] + async fn associate_into_inner_pairs_the_stream_with_the_teardown_owner() { + // The `Ok` arm of `AssociateSession::into_inner` (review 002 + // N-8): the taken stream pairs with the session that stays the + // teardown owner (the `into_inner_pairs_the_pipe_with_the_ + // teardown_owner` shape's ASSOCIATE mirror) — and a second + // take is the spent shape. + let (_write_peer, session_writer) = tokio::io::duplex(64); + let (read_peer, session_reader) = tokio::io::duplex(64); + let _ = read_peer; + + let session = AssociateSession { + channel_id: 1, + manager: None, + inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)), + reader: DatagramReader::new(), + pending: Default::default(), + relay_addr: Some(TargetAddr::Ip("127.0.0.1:0".parse().unwrap())), + }; + + let taken = session.into_inner().expect("the stream is live"); + // The pair carries both halves: the teardown owner (the relay + // address still observable) and the owned stream. + assert!(taken.session.relay_addr().is_some(), "the owner"); + assert!(taken.session.channel_id() == 1); + drop(taken); + } + + #[tokio::test] + async fn associate_debug_renders_channel_liveness_and_pending() { + // The Debug impl body (review 002 N-8, egress.rs class): the + // associate session renders its channel id, liveness, the + // relay address, and the pending count. + let mut session = AssociateSession { + channel_id: 7, + manager: None, + inner: Some(build_skip_auth_bi_stream( + tokio::io::duplex(16).0, + tokio::io::duplex(16).0, + )), + reader: DatagramReader::new(), + pending: Default::default(), + relay_addr: Some(TargetAddr::Ip("127.0.0.1:0".parse().unwrap())), + }; + session.pending.push_back(( + TargetAddr::Ip("10.0.0.1:53".parse().unwrap()), + Bytes::from_static(b"x"), + )); + let rendered = format!("{session:?}"); + assert!(rendered.contains("channel_id: 7"), "{rendered}"); + assert!(rendered.contains("associated: true"), "{rendered}"); + assert!(rendered.contains("relay_addr"), "{rendered}"); + assert!(rendered.contains("pending_datagrams: 1"), "{rendered}"); + drop(session); + + let bind = BindSession { + channel_id: 8, + manager: None, + inner: Some(build_skip_auth_bi_stream( + tokio::io::duplex(16).0, + tokio::io::duplex(16).0, + )), + bind_addr: Some(TargetAddr::Domain("bind.test".into(), 2121)), + }; + let rendered = format!("{bind:?}"); + assert!(rendered.contains("channel_id: 8"), "{rendered}"); + assert!(rendered.contains("awaiting_inbound: true"), "{rendered}"); + assert!(rendered.contains("bind_addr"), "{rendered}"); + } + + #[tokio::test] + async fn associate_split_yields_live_halves_and_survives_the_round_trip() { + use crate::wire::frame as wire_frame; + use tokio::io::AsyncWriteExt as _; + + // The front-door-bridge surface (consumer.md) on a *live* + // stream: the halves borrow the session and the raw framed + // bytes pass through both directions (0 B framing — the door's + // RFC header codec would own the frames at the boundary). + let addr = TargetAddr::Ip("10.0.0.1:53".parse().unwrap()); + let framed = wire_frame(&addr, b"split").unwrap().to_vec(); + + let (mut peer_writer, session_reader) = tokio::io::duplex(1024); + peer_writer.write_all(&framed).await.unwrap(); + drop(peer_writer); + let (mut peer_reader, session_writer) = tokio::io::duplex(1024); + + let mut session = AssociateSession { + channel_id: 1, + manager: None, + inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)), + reader: DatagramReader::new(), + pending: Default::default(), + relay_addr: None, + }; + let (mut read, mut write) = session.split().expect("live halves"); + + // Read the raw frame through the borrowed read half. + let mut header = [0u8; 2]; + read.read_exact(&mut header) + .await + .expect("the frame header"); + let len = u16::from_be_bytes(header) as usize; + let mut rest = vec![0u8; len]; + read.read_exact(&mut rest).await.expect("the frame body"); + assert_eq!(len, 7 + 5, "one v4-address datagram frame"); + drop(read); + + // The write half passes through (the bridge seam's shape) — + // the peer receives the exact bytes. + write.write_all(&framed).await.unwrap(); + write.flush().await.unwrap(); + drop(write); + + let mut got = vec![0u8; framed.len()]; + peer_reader + .read_exact(&mut got) + .await + .expect("the peer read"); + assert_eq!(got, framed, "the write half rode the raw pipe"); + } + + /// An inner `Socks5Stream` over a true `BiStream` (the session/ + /// plane field type) for the BIND-side surfaces: reads come from + /// `session_end` (the peer's writes), writes go out via the duplex + /// whose other end the test holds. Two joined halves, the same + /// shape `open_and_negotiate` assembles. + fn build_skip_auth_bi_stream( + session_end: tokio::io::DuplexStream, + writer_end: tokio::io::DuplexStream, + ) -> Socks5Stream { + let mut config = Config::default(); + config.set_skip_auth(true); + Socks5Stream::use_stream(BiStream::from_joined(session_end, writer_end), None, config) + .now_or_never() + .expect("skip_auth's use_stream is instantly ready") + .expect("skip_auth's use_stream succeeds") + } + + #[tokio::test] + async fn bind_split_yields_live_halves_pre_reply2() { + use tokio::io::AsyncWriteExt as _; + + // The BIND front-door-bridge surface (consumer.md): the split + // halves in the reply#1 phase — pre-reply#2 bytes ride the + // same channel (the doc's claim, exercised on a live stream). + let (mut peer_writer, session_reader) = tokio::io::duplex(1024); + peer_writer.write_all(b"raw").await.unwrap(); + drop(peer_writer); + + let (mut peer_reader, session_writer) = tokio::io::duplex(1024); + let mut session = BindSession { + channel_id: 1, + manager: None, + inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)), + bind_addr: Some(TargetAddr::Ip("127.0.0.1:2121".parse().unwrap())), + }; + { + let (mut read, mut write) = session.split().expect("live pre-reply2 halves"); + write.write_all(b"ack").await.unwrap(); + write.flush().await.expect("flush"); + let mut buf = [0u8; 3]; + read.read_exact(&mut buf).await.expect("the raw bytes"); + assert_eq!(&buf, b"raw"); + } + + // The write half reached the peer (the bridge seam's shape). + let mut buf = [0u8; 3]; + peer_reader.read_exact(&mut buf).await.expect("the ack"); + assert_eq!(&buf, b"ack"); + drop(peer_reader); + + // The take-guard is untouched: the reader still reads the + // reply#2 phase (the split halves were a view, not a + // consumer) — the dropped peer EOFs it loud, never the spent + // shape. + let outcome = session.await_inbound().await; + assert!( + !matches!(outcome, Ok(None)), + "the reply#2 phase was not consumed by the split" + ); + session.close().await; + } + + #[tokio::test] + async fn bind_data_plane_poll_flush_and_poll_shutdown_pass_through() { + use tokio::io::AsyncWriteExt as _; + + // The BIND door's half-close rides the plane's poll_shutdown + // passthrough (and poll_flush rides the write path): write → + // flush → shutdown — the bytes reach the peer and the EOF + // sentinel follows. + let (_session_writer_unused, session_reader) = tokio::io::duplex(64); + let (mut peer_reader, session_writer) = tokio::io::duplex(64); + let mut plane = BindDataPlane { + inner: build_skip_auth_bi_stream(session_reader, session_writer), + peer: TargetAddr::Ip("10.0.0.7:44444".parse().unwrap()), + }; + plane.write_all(b"half").await.expect("write"); + plane.flush().await.expect("the poll_flush passthrough"); + plane + .shutdown() + .await + .expect("the poll_shutdown passthrough"); + drop(plane); + + let mut seen = Vec::new(); + let mut buf = [0u8; 4]; + loop { + match peer_reader.read(&mut buf).await { + Ok(0) | Err(_) => break, + Ok(n) => seen.extend_from_slice(&buf[..n]), + } + } + assert_eq!(&seen, b"half", "the write rode through, then EOF"); + } + + #[tokio::test] + async fn read_reply2_rejects_a_foreign_producers_version() { + use tokio::io::AsyncWriteExt as _; + + // A foreign producer's reply#2 with a non-5 version: the + // version pin is loud (never an aliased peer). + let (mut peer_writer, session_reader) = tokio::io::duplex(64); + peer_writer + .write_all(&[0x04, 0x00, 0x00, 0x01, 10, 0, 0, 1, 0, 80]) + .await + .unwrap(); + drop(peer_writer); + + let (_write_peer_unused, session_writer) = tokio::io::duplex(64); + let mut inner = build_skip_auth_bi_stream(session_reader, session_writer); + let err = read_reply2(&mut inner) + .await + .expect_err("a foreign version is refused"); + assert!( + matches!( + err, + SocksSessionError::Rfc(SocksError::UnsupportedSocksVersion(0x04)) + ), + "the version pin is loud: {err:?}" + ); + } + /// A session over a raw duplex (the reader internals' test /// harness — the handshake-free path: `skip_auth` bypasses the /// greeting, no reply is expected, the datagram stage rides diff --git a/src/egress.rs b/src/egress.rs index 1e73d6c..b3dd3d7 100644 --- a/src/egress.rs +++ b/src/egress.rs @@ -592,4 +592,41 @@ mod tests { assert_send_sync::(); assert_send_sync::(); } + + #[test] + fn debug_impls_render_the_handle_surfaces() { + // The three Debug impl bodies (review 002 N-8): the dialed + // handle renders non-exhaustive; the accept handle carries the + // peer; the accept source carries the bind address. + let dialed = TargetHandle { + read: Box::new(tokio::io::empty()), + write: Box::new(tokio::io::sink()), + }; + let rendered = format!("{dialed:?}"); + assert!(rendered.contains("TargetHandle"), "{rendered}"); + assert!(rendered.contains(".."), "{rendered}"); + + let accepted = BindTargetHandle { + peer: TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 7], 44444))), + read: Box::new(tokio::io::empty()), + write: Box::new(tokio::io::sink()), + }; + let rendered = format!("{accepted:?}"); + assert!(rendered.contains("BindTargetHandle"), "{rendered}"); + assert!( + rendered.contains("10.0.0.7:44444"), + "the peer is named: {rendered}" + ); + + let accept = BindAccept::new( + SocketAddr::from(([127, 0, 0, 1], 2121)), + std::future::pending::>(), + ); + let rendered = format!("{accept:?}"); + assert!(rendered.contains("BindAccept"), "{rendered}"); + assert!( + rendered.contains("127.0.0.1:2121"), + "the bind address is named: {rendered}" + ); + } } diff --git a/src/error.rs b/src/error.rs index f2b4350..73808be 100644 --- a/src/error.rs +++ b/src/error.rs @@ -227,4 +227,25 @@ mod tests { )); assert!(matches!(io, SocksSessionError::Io(_))); } + + #[test] + fn establishment_reason_display_renders_the_wire_code() { + // The Display impl body (review 002 N-8): the typed reason + // renders as its canonical wire code (the `as_str` inverse). + for (typed, wire) in [ + (SocksEstablishmentReason::DialFailed, "dial_failed"), + ( + SocksEstablishmentReason::UnknownResource, + "unknown_resource", + ), + ( + SocksEstablishmentReason::ResourceShortage, + "resource_shortage", + ), + (SocksEstablishmentReason::HandlerError, "handler_error"), + (SocksEstablishmentReason::Timeout, "timeout"), + ] { + assert_eq!(typed.to_string(), wire); + } + } } diff --git a/src/producer.rs b/src/producer.rs index 6ebb4c8..dd63f3e 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -753,6 +753,8 @@ mod tests { use super::*; use crate::egress::{BindAcceptError, BindAcceptFactoryError, SocksDialError, TargetHandle}; use bytes::Bytes; + use std::pin::Pin; + use std::task::{Context, Poll}; use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; /// The test dial closure (the harness's echo shape, inline): one @@ -1626,6 +1628,296 @@ mod tests { ); } + // -- the datagram stage's failure exits (review 002 N-8) ----------- + // The stage's teardown surface, driven directly: each failure exit + // returns its exact `StageExit` (the serve layer discards it, so + // the unit shape is the only place the matrix is pinnable). + + /// A stream half that fails its reads or writes on demand (the + /// stage's `Io`-exit triggers). + struct FailStream { + fail_read: bool, + fail_write: bool, + } + + impl tokio::io::AsyncRead for FailStream { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { + if self.fail_read { + Poll::Ready(Err(std::io::Error::other("stage read failed"))) + } else { + Poll::Pending + } + } + } + + impl tokio::io::AsyncWrite for FailStream { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + if self.fail_write { + Poll::Ready(Err(std::io::Error::other("stage write failed"))) + } else { + Poll::Ready(Ok(buf.len())) + } + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + + /// An egress yielding one pre-set reply then parking forever (the + /// reply branch wins the select once; the stage never advances + /// past it). + struct OneReplyEgress { + reply: Option, + } + + impl crate::egress::Egress for OneReplyEgress { + fn send( + &mut self, + _addr: TargetAddr, + _payload: Bytes, + ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> { + Box::pin(async move { Err(EgressError::General("send refused".into())) }) + } + + fn next_reply( + &mut self, + ) -> futures::future::BoxFuture<'_, Result, EgressError>> { + let reply = self.reply.take(); + Box::pin(async move { + match reply { + Some(inner) => Ok(Some(inner)), + None => Ok(std::future::pending::>().await), + } + }) + } + } + + /// An egress whose relay already ended (`next_reply` errs) — the + /// `EgressEnded`-via-`next_reply` trigger. + struct RelayEndedEgress; + + impl crate::egress::Egress for RelayEndedEgress { + fn send( + &mut self, + _addr: TargetAddr, + _payload: Bytes, + ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> { + Box::pin(async move { Err(EgressError::General("send refused".into())) }) + } + + fn next_reply( + &mut self, + ) -> futures::future::BoxFuture<'_, Result, EgressError>> { + Box::pin(async move { Err(EgressError::General("relay ended".into())) }) + } + } + + /// An egress that parks forever on `next_reply` (the read branch's + /// exits are then deterministic — the parked-accept shape's egress + /// mirror). + struct ParkedEgress; + + impl crate::egress::Egress for ParkedEgress { + fn send( + &mut self, + _addr: TargetAddr, + _payload: Bytes, + ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> { + Box::pin(async move { Err(EgressError::General("send refused".into())) }) + } + + fn next_reply( + &mut self, + ) -> futures::future::BoxFuture<'_, Result, EgressError>> { + Box::pin(async move { Ok(std::future::pending::>().await) }) + } + } + + /// The `ReplyFrameRejected` exit: an egress reply whose inner + /// bytes violate the u16 joint bound — the wire codec refuses it + /// at `frame_inner`, the stage tears the association down (never a + /// wire overflow). + #[tokio::test] + async fn datagram_stage_reply_frame_rejected_exit() { + let mut egress = Box::new(OneReplyEgress { + reply: Some(Bytes::from(vec![0u8; crate::wire::MAX_INNER_LEN + 1])), + }) as Box; + let exit = run_datagram_stage( + FailStream { + fail_read: false, + fail_write: true, + }, + egress.as_mut(), + ) + .await; + assert_eq!(exit, StageExit::ReplyFrameRejected); + } + + /// The `EgressEnded` exit via `next_reply`'s `Err` — the egress + /// relay ended; the association ends with it (the lossy-datagram + /// posture). + #[tokio::test] + async fn datagram_stage_egress_relay_ended_exit_via_next_reply() { + let mut egress = Box::new(RelayEndedEgress) as Box; + let exit = run_datagram_stage( + FailStream { + fail_read: false, + fail_write: false, + }, + egress.as_mut(), + ) + .await; + assert_eq!(exit, StageExit::EgressEnded); + } + + /// The `Io` exit via a failing write: a valid reply frame arrives + /// but the channel's write half fails — the stage tears the + /// association down. + #[tokio::test] + async fn datagram_stage_write_failure_is_the_io_exit() { + let inner = { + let mut v = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], 53))) + .to_be_bytes() + .unwrap() + .to_vec(); + v.extend_from_slice(b"payload"); + Bytes::from(v) + }; + let mut egress = Box::new(OneReplyEgress { reply: Some(inner) }) as Box; + let exit = run_datagram_stage( + FailStream { + fail_read: false, + fail_write: true, + }, + egress.as_mut(), + ) + .await; + assert_eq!(exit, StageExit::Io); + } + + /// The `Io` exit via a failing channel read — the stage tears the + /// association down (a stream error is never a silent end). + #[tokio::test] + async fn datagram_stage_read_failure_is_the_io_exit() { + let mut egress = Box::new(ParkedEgress) as Box; + let exit = run_datagram_stage( + FailStream { + fail_read: true, + fail_write: false, + }, + egress.as_mut(), + ) + .await; + assert_eq!(exit, StageExit::Io); + } + + /// The `EgressEnded` exit via a failing `egress.send`, at the + /// serve layer (the drop-witness treatment + /// `egress_factory_called_once_and_dropped_with_the_stage` gives + /// the happy path): one datagram in, the egress refuses it, the + /// stage exits — EOF reaches the client and the egress drops with + /// the stage (teardown-at-birth on the failure path). + #[tokio::test] + async fn udp_egress_send_failure_ends_the_association_and_drops_the_egress() { + use std::sync::atomic::{AtomicBool, Ordering}; + + let dropped = Arc::new(AtomicBool::new(false)); + let dropped_for_factory = Arc::clone(&dropped); + let factory: EgressFactory = Arc::new(move || { + let dropped = Arc::clone(&dropped_for_factory); + Box::pin(async move { + struct FailingSendWitness { + flag: Arc, + } + impl crate::egress::Egress for FailingSendWitness { + fn send( + &mut self, + _addr: TargetAddr, + _payload: Bytes, + ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> + { + Box::pin( + async move { Err(EgressError::General("egress send failed".into())) }, + ) + } + + fn next_reply( + &mut self, + ) -> futures::future::BoxFuture<'_, Result, EgressError>> + { + Box::pin(async move { Ok(std::future::pending::>().await) }) + } + } + impl Drop for FailingSendWitness { + fn drop(&mut self) { + self.flag.store(true, Ordering::SeqCst); + } + } + Ok(Box::new(FailingSendWitness { flag: dropped }) as Box) + }) + }); + + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let handle = tokio::spawn(serve_socks5_over( + server_end, + None, + Some(factory), + None, + false, + )); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::UDPAssociate, + &TargetAddr::Domain("send.fail".into(), 1080), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x00, "the associate reply succeeds"); + + // One datagram in: the egress refuses it → the stage exits. + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 4], 53))); + client + .write_all(&datagram_frame(&target, b"doomed")) + .await + .unwrap(); + + // The association ends: EOF (no reply datagram follows). + let mut buf = [0u8; 16]; + let outcome = + tokio::time::timeout(std::time::Duration::from_secs(5), client.read(&mut buf)) + .await + .expect("read resolves"); + match outcome { + Ok(0) | Err(_) => {} + Ok(n) => panic!("EOF expected after the egress send failure — got {n} bytes"), + } + drop(client); + tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect("the stage exited"); + assert!( + dropped.load(Ordering::SeqCst), + "the stage exit drops the egress on the failure path too" + ); + } + /// CONNECT and ASSOCIATE coexist on one `serve_socks5_over` (the /// regression guard, extended): both commands serve their data /// planes over separate duplexes on the same seam set. @@ -2163,6 +2455,182 @@ mod tests { .expect("join"); } + /// The quiet wait's `Err(_)` arm (review 002 N-8): a channel read + /// that *errors* (I/O error, not EOF) during the quiet wait ends + /// the wait with a clean `Ok(())` — the accept future drops with + /// the session (the client-gone contract's error-shaped twin). + /// Driven directly on `bind_quiet_wait_and_serve` (a read error is + /// not producible over a duplex). + #[tokio::test] + async fn bind_quiet_wait_read_error_ends_the_wait_cleanly() { + use std::cell::Cell; + use std::sync::atomic::{AtomicBool, Ordering}; + + let dropped = Arc::new(AtomicBool::new(false)); + struct FailReadStream { + polled: Cell, + } + impl tokio::io::AsyncRead for FailReadStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + _buf: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { + // Park once (both branches get polled — the accept + // future starts, its owned state exists), then error + // on the wake: the read branch's error resolves after + // the accept branch has started. + if self.polled.get() { + Poll::Ready(Err(std::io::Error::other("quiet-wait read failed"))) + } else { + self.polled.set(true); + cx.waker().wake_by_ref(); + Poll::Pending + } + } + } + impl tokio::io::AsyncWrite for FailReadStream { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Poll::Ready(Ok(buf.len())) + } + fn poll_flush( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + fn poll_shutdown( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + } + + struct OwnedAccept(Arc); + impl Drop for OwnedAccept { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + let accept = { + let dropped = Arc::clone(&dropped); + BindAccept::new(SocketAddr::from(([127, 0, 0, 1], 2121)), async move { + let owner = OwnedAccept(dropped); + std::future::pending::<()>().await; + drop(owner); + unreachable!("pending never resolves") + }) + }; + + let outcome = bind_quiet_wait_and_serve( + FailReadStream { + polled: Cell::new(false), + }, + accept, + ) + .await; + outcome.expect("an error-shaped quiet wait is not an error exit"); + assert!( + dropped.load(Ordering::SeqCst), + "the accept future dropped with the wait (the client-gone contract, error arm)" + ); + } + + /// The BIND reply#2 Domain-peer fallback (review 002 N-8): a + /// foreign accept handle carrying a `TargetAddr::Domain(_, port)` + /// peer — the producer writes reply#2 with the unspecified + /// address on the peer's port (`0.0.0.0:port`) and serves the + /// data plane. + #[tokio::test] + async fn bind_domain_peer_falls_back_to_the_unspecified_address() { + let factory: BindAcceptFactory = Arc::new(move |_target: TargetAddr| { + Box::pin(async move { + Ok(BindAccept::new( + SocketAddr::from(([127, 0, 0, 1], 2121)), + async { + let (mut echo_end, target_end) = tokio::io::duplex(64); + tokio::spawn(async move { + let mut buf = [0u8; 64]; + loop { + match echo_end.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => { + if echo_end.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + } + }); + let (read_half, write_half) = tokio::io::split(target_end); + Ok(BindTargetHandle { + peer: TargetAddr::Domain("peer.example".into(), 44444), + read: Box::new(read_half) as Box, + write: Box::new(write_half) + as Box, + }) + }, + )) + }) + }); + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let handle = tokio::spawn(serve_socks5_over( + server_end, + None, + None, + Some(factory), + true, + )); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPBind, + &TargetAddr::Domain("domain.peer".into(), 2121), + ) + .await; + + // Reply#1 succeeds; reply#2 carries the Domain peer's fallback + // (the unspecified address, the peer's port). + let (code1, _addr) = read_reply(&mut client).await; + assert_eq!(code1, 0x00, "reply#1 succeeded"); + let (code2, addr) = read_reply(&mut client).await; + assert_eq!(code2, 0x00, "reply#2 succeeded"); + assert_eq!( + SocketAddr::from(( + [addr[0], addr[1], addr[2], addr[3]], + u16::from_be_bytes([addr[4], addr[5]]) + )), + SocketAddr::from(([0, 0, 0, 0], 44444)), + "the Domain peer fell back to 0.0.0.0:port" + ); + + // The data plane is live (the accept resolved with in-process + // halves — the parked probe half stays alive by forget). + client.write_all(b"ping").await.unwrap(); + let mut buf = [0u8; 4]; + tokio::time::timeout( + std::time::Duration::from_secs(5), + client.read_exact(&mut buf), + ) + .await + .expect("the data plane echoes") + .expect("read the echo"); + assert_eq!(&buf, b"ping"); + + drop(client); + tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect("clean"); + } + /// The pump handler's shape: `accept_bi` → the same conversation, /// with the handler's `JoinHandle` tracking the data plane (R-02 /// — the spawn shape is inline-await, not spawn-and-forget). @@ -2232,6 +2700,15 @@ mod tests { Some("opener"), "the per-call opener identity is recorded" ); + + // The None-witness arm (the registration's optional witness + // seam): the session gate passes without recording anything. + let establisher = socks5_session_establisher(None); + let auth = AuthContext::anonymous(b"alk/socks5"); + let establishment = establisher(serde_json::json!({}), auth) + .await + .expect("the session gate passes without a witness"); + assert!(establishment.plan.is_none(), "no plan payload"); } /// The command gate's resolved boolean rides the open-op diff --git a/tasks/implementation/fix-coverage-residue.md b/tasks/implementation/fix-coverage-residue.md index 398c420..62ebaac 100644 --- a/tasks/implementation/fix-coverage-residue.md +++ b/tasks/implementation/fix-coverage-residue.md @@ -1,7 +1,7 @@ --- id: implementation/fix-coverage-residue name: "Tests — the review-002 coverage residue (N-8: stage exits, splits, spent arms)" -status: pending +status: completed depends_on: [implementation/fix-bind-reply2-reentry, implementation/fix-producer-gate-policy, implementation/fix-rfc1928-posture] scope: moderate risk: low @@ -77,19 +77,19 @@ through an artificial shape. ## Acceptance Criteria -- [ ] The stage's failure exits tested (`ReplyFrameRejected`, +- [x] The stage's failure exits tested (`ReplyFrameRejected`, `EgressEnded` both arms, `Io`) — the `send`-failure arm with the drop-witness assertion -- [ ] `AssociateSession::split` + `BindSession::split` tested on a +- [x] `AssociateSession::split` + `BindSession::split` tested on a live stream (or the batch-2-door note recorded per arm) -- [ ] The single-arm residue covered: spent-`send`, mid-batch +- [x] The single-arm residue covered: spent-`send`, mid-batch later-frame `Err`, ASSOCIATE `into_inner`'s `Ok`, `BindDataPlane` shutdown, `read_reply2` version mismatch, the quiet-wait `Err(_)` arm, the Domain-peer fallback, the `Debug`/`Display` renders -- [ ] `cargo llvm-cov` re-run: the named lines covered or recorded +- [x] `cargo llvm-cov` re-run: the named lines covered or recorded as accepted-uncovered with a reason in the task Notes -- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean +- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean ## References @@ -103,8 +103,137 @@ through an artificial shape. ## Notes -> Agent fills this during implementation. +> Filled by the implementation agent (2026-09-24): + +- **All N-8-named lines covered; no accepted-uncovered entries + added.** The `cargo llvm-cov --show-missing-lines` diff before → + after: every line the review attributed (consumer.rs 356-359/ + 419/437/453-456/636-642/771-774/843 pre-fix numbering; producer.rs + 374-386/405-414/560/568; egress.rs 72-74/226-230/275-279; + error.rs 113-115) is now covered. The suite grew 357 → 375 tests + (145 lib unit / 38 associate / 41 bind / 40 connect / 41 + integration / 30 harness / 37 differential / 3 doc-tests); + TOTAL 96.31% regions / 97.22% lines. +- **The stage's failure exits were pinned as direct unit tests on + `run_datagram_stage`** (the serve layer discards `StageExit`, so + the unit shape is the only place the exact exit matrix is + assertable): `datagram_stage_reply_frame_rejected_exit` (an + egress reply whose inner bytes violate the u16 joint bound → + `ReplyFrameRejected`), `datagram_stage_egress_relay_ended_exit_ + via_next_reply` (`next_reply`'s `Err` → `EgressEnded`), + `datagram_stage_write_failure_is_the_io_exit` + `datagram_stage_ + read_failure_is_the_io_exit` (a failing channel write/read → + `Io`). The `send`-failure arm got the serve-layer drop-witness + treatment: `udp_egress_send_failure_ends_the_association_and_ + drops_the_egress` (a full ASSOCIATE conversation, one datagram in, + the egress refuses it, EOF reaches the client, the stage exits, + and the drop flag flips — teardown-at-birth on the failure path). + The tests ride `FailStream` (a poll-failing stream) and + `OneReplyEgress`/`RelayEndedEgress`/`ParkedEgress` (stage-shaped + egress fakes), all inside the producer test module. +- **The two splits tested on live streams.** `associate_split_ + yields_live_halves_and_survives_the_round_trip` and `bind_split_ + yields_live_halves_pre_reply2` exercise `tokio::io::split` on a + live `Socks5Stream` built by a new test helper + (`build_skip_auth_bi_stream` — two duplexes joined via + `BiStream::from_joined`, the same shape `open_and_negotiate` + assembles): raw framed bytes pass through both directions on the + borrowed halves, and (BIND) the split leaves the take-guard + untouched — `await_inbound` still reads the reply#2 phase after + the halves are dropped. No "batch-2 door exercises it" note + needed; the tests stand alone. +- **The single-arm residue, test by test:** + `send_datagram_on_a_spent_stream_is_the_io_arm` (the spent-stream + `Io` arm — and the spent-`recv` shape asserted in the same test: + `Ok(None)`, the family's read-side posture), + `recv_datagram_mid_batch_later_frame_err_drops_the_pending_oks` + (first frame `Ok`, a later one garbage → the pending `Ok`s dropped + with the session, the fail-loud-drops-pending semantics pinned), + `associate_into_inner_pairs_the_stream_with_the_teardown_owner` + (the `TakenAssociateInner` Ok arm — the CONNECT + `into_inner_pairs_the_pipe_with_the_teardown_owner` shape's + ASSOCIATE mirror), + `bind_data_plane_poll_flush_and_poll_shutdown_pass_through` + (write → `poll_flush` → `poll_shutdown` — the bytes reach the peer + and the EOF sentinel follows), + `read_reply2_rejects_a_foreign_producers_version` (a non-5 + version reply#2 → `UnsupportedSocksVersion(0x04)`, loud), + `bind_quiet_wait_read_error_ends_the_wait_cleanly` (the quiet + wait's `Err(_)` arm driven directly on `bind_quiet_wait_and_serve` + — a read *error* ends the wait `Ok(())` and the accept future + drops with its owned state, the client-gone contract's + error-shaped twin), `bind_domain_peer_falls_back_to_the_ + unspecified_address` (a foreign accept handle carrying + `TargetAddr::Domain(_, port)` peer → reply#2 carries + `0.0.0.0:port`, data plane live), + `debug_impls_render_the_handle_surfaces` (the three egress.rs + `Debug` bodies — TargetHandle non-exhaustive, BindTargetHandle + names the peer, BindAccept names the bind address), + `establishment_reason_display_renders_the_wire_code` (the + `SocksEstablishmentReason` `Display` — each variant renders as its + canonical wire code), `associate_debug_renders_channel_liveness_ + and_pending` (the ASSOCIATE + BIND session `Debug` bodies), + `spent_associate_close_reaps_a_no_op` (the `close`-on-spent arm), + and `establisher_records_identity_and_defaults_the_establishment` + extended with the `None`-witness arm. +- **A quiet-wait select-shape discovery (test-only note):** + `tokio::select!` polls branches in random order — a read branch + that errors on its *first* poll can win before the accept branch + is ever polled, dropping the accept future at zero polls (the + owned state never constructed; a drop flag would never fire — the + first cut of the test caught this flake). The test's `FailReadStream` + parks once (with a wake) then errors on the second poll so both + branches are guaranteed started; the assertion then holds + deterministically. Production semantics unaffected (either exit + drops the accept future; the contract is exit-shaped, not + poll-count-shaped). +- **One review-item line numbers shifted post-fix-tasks**: the + review's `consumer.rs:396-398` (the covered spent-`recv` arm) is + now `418` (the `inner.as_mut()` else arm after the C-1/D-1 + changes); it is covered by the same test that pins the spent-send + arm. `consumer.rs:937` (the spent-bind test's `err().map()` + diagnostic closure) stays uncovered — it only evaluates on `Err`, + and the spent shape is `Ok(None)` by design (the C-1 posture); + cosmetic, not chased. +- **Reachability check for the arm-became-unreachable clause**: no + N-8 arm was made unreachable by the three code-fix tasks — every + named arm still exists at its (shifted) location and is now + tested. The explicit out-of-scope items stayed out of scope: + `rfc1928/client.rs:217-219` remains accepted-uncovered (with its + translated-error note), as do the fork's other accepted-uncovered + blocks. +- **Residual uncovered lines after this task (none new)**: producer + .rs 188-189 (the pump handler's `accept_bi` substrate-failure log + arm — pre-existing, not in N-8's list; untestable without a + broken `Connection` outside alkcall's test surface), + rfc1928/client.rs 217-219 + the fork internals (accepted- + uncovered per review-001/002), wire.rs/mod.rs/target_addr.rs + internals (pre-existing, not in N-8's list), and the test modules' + own panic-guard/unreachable lines (coverage-of-tests noise). ## Summary -> Agent fills this on completion. \ No newline at end of file +Closed review 002's N-8 coverage residue with 18 new tests and zero +production-code changes: the datagram stage's three failure exits +pinned as direct unit tests on `run_datagram_stage` +(`ReplyFrameRejected`, `EgressEnded` via `next_reply`, `Io` via a +failing write and a failing read) plus the serve-layer drop-witness +test for the `send`-failure arm (the egress drops with the stage on +the failure path too); the two front-door-bridge `split`s tested on +live `Socks5Stream` streams (raw bytes through both +halves; the BIND split leaves the take-guard untouched); and the +single-arm residue (spent `send_datagram`'s `Io` arm + the spent +`recv` shape, the mid-batch later-frame `Err` dropping pending `Ok`s, +ASSOCIATE `into_inner`'s `Ok` pairing, `BindDataPlane`'s +`poll_flush`/`poll_shutdown` passthroughs, `read_reply2`'s version +pin, the quiet wait's `Err(_)` arm, the BIND reply#2 Domain-peer +fallback `0.0.0.0:port`, the three egress `Debug` renders, the +`SocksEstablishmentReason` `Display`, the ASSOCIATE/BIND session +`Debug` bodies, the spent-`close` no-op, and the establisher's +`None`-witness arm). `cargo llvm-cov` re-run: every N-8-named line +covered, no new accepted-uncovered entries (the residual uncovered +lines are the pre-existing out-of-scope set + producer.rs 188-189's +untestable `accept_bi` log arm). Verification: `cargo test` + +`--all-features` green (375 tests), clippy `-D warnings` clean +(native + wasm), fmt clean, `cargo doc` clean, both wasm checks +clean; coverage 96.31% regions / 97.22% lines. \ No newline at end of file