producer: the UDP ASSOCIATE datagram stage — the select!-driven relay (ADR-003)
- serve_udp_associate: sentinel reply (127.0.0.1:0 — the handle-swap surface) then the datagram stage; missing egress refuses RFC-faithfully (CommandNotSupported, ADR-008); egress-factory failure maps to reply#1's code - run_datagram_stage: select! over channel reads ∥ egress replies (the concurrent full-duplex shape), typed StageExit diagnostics (EOF = the clean exit; undecodable frames fail loud; the egress readiness contract tolerated as spurious wakeups) - wire: frame_inner — the reply-path half of framing-once (identical wire to frame(), pinned); egress inner bytes never re-prefixed - tests: the readiness-contract test (instant-Ok(None) egress does not starve the stage), empty-payload datagram legal (never EOF), undecodable-frame loud fail, EOF teardown + fresh associate, the egress drop witness (stage owns it), missing-egress refusal + CONNECT guard — in-process (duplex) and over channels (harness); the three-command regression guard drives all commands serving Verified: cargo test 174/174; clippy --all-targets -D warnings; fmt --check; wasm check + clippy (wasm32-unknown-unknown); doc.
This commit is contained in:
glm-5.3-flash committed
1 parent
048493fefa
commit
df772ccbeb
5 files changed
+1181
-54
No files matched your search
+1
-1
@@ -45,4 +45,4 @@ pub use producer::{
|
||||
make_socks5_pump_handler, register_socks5_openable, serve_socks5_over,
|
||||
socks5_session_establisher, IdentityWitness, Socks5Producer, Socks5ProducerBuilder,
|
||||
};
|
||||
pub use wire::{addr_encoding_len, frame, DatagramReader, WireError, MAX_INNER_LEN};
|
||||
pub use wire::{addr_encoding_len, frame, frame_inner, DatagramReader, WireError, MAX_INNER_LEN};
|
||||
+763
-23
@@ -40,9 +40,16 @@
|
||||
//! birth). Refusal: `reply_error` with the mapped RFC code
|
||||
//! (dial-failure → `ConnectionRefused` 0x05; scope-class →
|
||||
//! `ConnectionNotAllowed` 0x02) — no phantom data plane.
|
||||
//! - **UDP ASSOCIATE** → RFC-faithful refusal in this task's scope
|
||||
//! (`CommandNotSupported` — ADR-008's missing-callback class); the
|
||||
//! producer-associate task replaces it.
|
||||
//! - **UDP ASSOCIATE** → the datagram stage (ADR-003): the sentinel
|
||||
//! reply (`127.0.0.1:0` — the relay handle IS the channel; the
|
||||
//! address is pure RFC-compatibility surface), then the
|
||||
//! `select!`-driven stage (wire.md §UDP ASSOCIATE): channel reads ∥
|
||||
//! egress replies, framing applied exactly once by the wire layer,
|
||||
//! the egress readiness contract (`Pending` when nothing is ready),
|
||||
//! association lifetime = channel lifetime. Missing egress seam:
|
||||
//! RFC-faithful refusal (`CommandNotSupported` — ADR-008's
|
||||
//! missing-callback class). Egress-factory failure: reply#1 with
|
||||
//! the mapped code.
|
||||
//! - **TCP BIND** → the command gate first (`socks5:bind`, ADR-006,
|
||||
//! from the per-call opener identity — a scope-less identity gets
|
||||
//! `ConnectionNotAllowed` 0x02 while CONNECT still serves it), then
|
||||
@@ -97,11 +104,14 @@ use alkcall::registry::registration::OperationRegistry;
|
||||
use serde_json::Value;
|
||||
use tokio::io::{AsyncRead, AsyncWrite};
|
||||
|
||||
use crate::egress::{BindAccept, BindAcceptFactory, BindTargetHandle, DialFn, EgressFactory};
|
||||
use crate::egress::{
|
||||
BindAccept, BindAcceptFactory, BindTargetHandle, DialFn, Egress, EgressFactory,
|
||||
};
|
||||
use crate::params::{socks5_open_spec, SOCKS5_BIND_SCOPE};
|
||||
use crate::rfc1928::server::{new_reply, Socks5ServerProtocol, SocksServerError};
|
||||
use crate::rfc1928::util::target_addr::TargetAddr;
|
||||
use crate::rfc1928::{ReplyError, Socks5Command, SocksError};
|
||||
use crate::wire::{frame_inner, DatagramReader};
|
||||
|
||||
pub use crate::params::{OP_SOCKS5_OPEN, SOCKS5_ALPN, SOCKS5_OPEN_SCOPE};
|
||||
|
||||
@@ -260,9 +270,16 @@ where
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// UDP ASSOCIATE: RFC-faithful refusal in this task's scope — the
|
||||
/// datagram stage lands with the producer-associate task (the
|
||||
/// pre-vertical posture; ADR-008's missing-callback class).
|
||||
/// UDP ASSOCIATE: the datagram stage (ADR-003, POC #2-validated).
|
||||
/// The typestate is driven directly (no `run_udp_proxy_custom` — its
|
||||
/// unconditional kernel bind is unusable on channels); the reply
|
||||
/// carries the sentinel address (the relay handle IS the channel;
|
||||
/// the address is pure RFC-compatibility surface for wrapper-aware
|
||||
/// clients, who never sendto() it); the post-reply stream is the
|
||||
/// framed `{addr, data}` datagram stage. A missing egress seam
|
||||
/// refuses RFC-faithfully (ADR-008's missing-callback class — an
|
||||
/// unprovided capability, never a panic); an egress-factory failure
|
||||
/// refuses with the mapped code.
|
||||
async fn serve_udp_associate<T>(
|
||||
proto: Socks5ServerProtocol<T, crate::rfc1928::server::states::CommandRead>,
|
||||
egress: Option<EgressFactory>,
|
||||
@@ -270,11 +287,146 @@ async fn serve_udp_associate<T>(
|
||||
where
|
||||
T: AsyncRead + AsyncWrite + Unpin,
|
||||
{
|
||||
let _ = egress;
|
||||
proto.reply_error(&ReplyError::CommandNotSupported).await?;
|
||||
Err(SocksError::ServerError(SocksServerError::UnknownCommand(
|
||||
Socks5Command::UDPAssociate.as_u8(),
|
||||
)))
|
||||
let Some(egress) = egress else {
|
||||
proto.reply_error(&ReplyError::CommandNotSupported).await?;
|
||||
return Err(SocksError::ServerError(SocksServerError::UnknownCommand(
|
||||
Socks5Command::UDPAssociate.as_u8(),
|
||||
)));
|
||||
};
|
||||
|
||||
let mut egress = match egress().await {
|
||||
Ok(egress) => egress,
|
||||
Err(err) => {
|
||||
proto.reply_error(&err.to_reply_error()).await?;
|
||||
return Err(SocksError::ServerError(SocksServerError::Bug(
|
||||
"egress refused in-stream",
|
||||
)));
|
||||
}
|
||||
};
|
||||
|
||||
let inner = proto.reply_success(sentinel_reply_addr()).await?;
|
||||
|
||||
run_datagram_stage(inner, egress.as_mut()).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The post-reply datagram stage (wire.md §UDP ASSOCIATE; the POC #2
|
||||
/// shape). Three POC invariants, load-bearing:
|
||||
///
|
||||
/// - **The concurrent select shape** — channel reads and the egress's
|
||||
/// reply stream run concurrently (the honest full-duplex shape; a
|
||||
/// poll-then-block loop loses replies arriving while the channel
|
||||
/// read is parked — the POC's first iteration, measured).
|
||||
/// - **The egress readiness contract** (`Egress::next_reply`): it
|
||||
/// must be `Pending` when no reply is ready; an instantly-ready
|
||||
/// `Ok(None)` starves a biased select into livelock. The stage
|
||||
/// tolerates a spurious `Ok(None)` by re-parking (never a hot
|
||||
/// loop); an `Err` ends the association (the egress drops with it).
|
||||
/// - **Framing applied exactly once, by the wire layer** — the stage
|
||||
/// reads frames via `DatagramReader` (channel → egress `send` gets
|
||||
/// the payload) and writes frames via `frame_inner` (egress
|
||||
/// `next_reply`'s inner bytes → the channel). The egress never sees
|
||||
/// a length prefix.
|
||||
///
|
||||
/// Association lifetime = channel lifetime: channel EOF ends the
|
||||
/// stage; no extra control vocabulary. A undecodable frame (garbage
|
||||
/// ATYP — any post-reply control byte) fails the association loud
|
||||
/// (the RFC's "post-reply control data is garbage" semantic). A
|
||||
/// stream end mid-frame is fail-loud too (never a silent partial
|
||||
/// datagram — the reader state is the teardown diagnostic). The
|
||||
/// stage owns the egress: on every exit path the egress binding
|
||||
/// drops here (teardown-at-birth — no `_` bindings of owning
|
||||
/// values).
|
||||
///
|
||||
/// Per-datagram addressing in the frame does the demultiplexing (no
|
||||
/// flow table); one relay loop, no spawn-per-datagram. A frame's
|
||||
/// egress `send` is awaited inline before the next select iteration —
|
||||
/// the stage interleaves egress sends with reply drains (the POC's
|
||||
/// measured shape; sufficient for relay postures).
|
||||
async fn run_datagram_stage<T>(mut inner: T, egress: &mut (dyn Egress + Send)) -> StageExit
|
||||
where
|
||||
T: AsyncRead + AsyncWrite + Unpin,
|
||||
{
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
let mut reader = DatagramReader::new();
|
||||
let mut chunk = vec![0u8; 16 * 1024];
|
||||
loop {
|
||||
tokio::select! {
|
||||
// The egress's reply stream (its own pump; `None` = a
|
||||
// spurious wakeup, the stage re-parks — never a hot loop).
|
||||
reply = egress.next_reply() => {
|
||||
match reply {
|
||||
Ok(Some(frame)) => {
|
||||
match frame_inner(&frame) {
|
||||
Ok(wire) => {
|
||||
if inner.write_all(&wire).await.is_err() {
|
||||
return StageExit::Io;
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
tracing::debug!("udp stage: reply frame rejected: {err}");
|
||||
return StageExit::ReplyFrameRejected;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(None) => {}
|
||||
Err(err) => {
|
||||
tracing::debug!("udp egress relay ended: {err:#}");
|
||||
return StageExit::EgressEnded;
|
||||
}
|
||||
}
|
||||
}
|
||||
// The channel's read side: frames in, EOF = end.
|
||||
read = inner.read(&mut chunk) => {
|
||||
match read {
|
||||
// Channel EOF = association end (the clean exit —
|
||||
// the association lifetime is the channel's).
|
||||
Ok(0) => return StageExit::ChannelEof,
|
||||
Ok(n) => {
|
||||
for dg in reader.feed(&chunk[..n]) {
|
||||
match dg {
|
||||
Err(err) => {
|
||||
tracing::debug!("udp stage: undecodable frame: {err}");
|
||||
return StageExit::UndecodableFrame;
|
||||
}
|
||||
Ok((addr, payload)) => {
|
||||
if let Err(err) = egress.send(addr, payload).await {
|
||||
tracing::debug!("udp egress send failed: {err:#}");
|
||||
return StageExit::EgressEnded;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
tracing::debug!("udp stage: channel read error: {err}");
|
||||
return StageExit::Io;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Why the datagram stage ended (the stage's teardown diagnostic —
|
||||
/// each exit drops the egress with the stage; the channel EOF is the
|
||||
/// one clean exit, per the association-lifetime contract).
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum StageExit {
|
||||
/// Channel EOF — the association's clean end (the one the
|
||||
/// association-lifetime contract names).
|
||||
ChannelEof,
|
||||
/// A frame the codec refused (garbage ATYP / a `len == 0` frame —
|
||||
/// post-reply control bytes). Loud fail: never a silent drop.
|
||||
UndecodableFrame,
|
||||
/// An egress `next_reply` `Err` or a `send` failure (the egress
|
||||
/// relay ended; the lossy-datagram posture).
|
||||
EgressEnded,
|
||||
/// A reply the wire codec rejected (the egress violated the
|
||||
/// joint bound — never a wire overflow).
|
||||
ReplyFrameRejected,
|
||||
/// The channel stream itself failed (I/O).
|
||||
Io,
|
||||
}
|
||||
|
||||
/// TCP BIND: the phase model (ADR-007; POC #5-validated). Phases:
|
||||
@@ -546,6 +698,7 @@ impl Socks5ProducerBuilder {
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::egress::{BindAcceptError, BindAcceptFactoryError, SocksDialError, TargetHandle};
|
||||
use bytes::Bytes;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream};
|
||||
|
||||
/// The test dial closure (the harness's echo shape, inline): one
|
||||
@@ -779,29 +932,616 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
/// The ASSOCIATE dispatch arm refuses RFC-faithfully in this
|
||||
/// task's scope (`CommandNotSupported` — the pre-vertical
|
||||
/// posture; the producer-associate task replaces it). BIND now
|
||||
/// serves under the gate — its tests live below.
|
||||
#[tokio::test]
|
||||
async fn associate_arm_refuses_command_not_supported() {
|
||||
// -- UDP ASSOCIATE: the datagram stage (ADR-003, POC #2 shape,
|
||||
// in-process over duplexes) ------------------------------------
|
||||
|
||||
/// The test egress (the harness's ChanEgress shape, inline): one
|
||||
/// unbounded queue; `send` frames the reply inner bytes for echo
|
||||
/// (`addr + data` — the egress speaks inner bytes, no length
|
||||
/// prefix); `next_reply` parks on the queue (`Pending` when
|
||||
/// nothing is ready — the readiness contract).
|
||||
struct QueueEgress {
|
||||
inbound: tokio::sync::mpsc::UnboundedSender<Bytes>,
|
||||
outbound: tokio::sync::Mutex<tokio::sync::mpsc::UnboundedReceiver<Bytes>>,
|
||||
}
|
||||
|
||||
impl crate::egress::Egress for QueueEgress {
|
||||
fn send(
|
||||
&mut self,
|
||||
addr: TargetAddr,
|
||||
payload: Bytes,
|
||||
) -> futures::future::BoxFuture<'_, Result<(), crate::egress::EgressError>> {
|
||||
Box::pin(async move {
|
||||
let mut inner = addr.to_be_bytes().map_err(|e| {
|
||||
crate::egress::EgressError::General(format!("addr encode: {e}"))
|
||||
})?;
|
||||
inner.extend_from_slice(&payload);
|
||||
self.inbound
|
||||
.send(Bytes::from(inner))
|
||||
.map_err(|_| crate::egress::EgressError::General("egress closed".into()))
|
||||
})
|
||||
}
|
||||
|
||||
fn next_reply(
|
||||
&mut self,
|
||||
) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, crate::egress::EgressError>>
|
||||
{
|
||||
Box::pin(async move {
|
||||
match self.outbound.lock().await.recv().await {
|
||||
Some(inner) => Ok(Some(inner)),
|
||||
None => Err(crate::egress::EgressError::General("egress closed".into())),
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
use crate::egress::EgressError;
|
||||
|
||||
/// The frame a raw client sends on the datagram stage:
|
||||
/// `[len: u16 BE][addr + data]`.
|
||||
fn datagram_frame(addr: &TargetAddr, payload: &[u8]) -> Vec<u8> {
|
||||
crate::wire::frame(addr, payload).unwrap().to_vec()
|
||||
}
|
||||
|
||||
/// Split the wire into (addr, payload) — the raw client's reply
|
||||
/// decode (one frame per read batch, in order).
|
||||
fn decode_datagram(wire: &[u8]) -> (TargetAddr, Bytes) {
|
||||
let mut reader = crate::wire::DatagramReader::new();
|
||||
let mut got = None;
|
||||
for dg in reader.feed(wire) {
|
||||
got = Some(dg);
|
||||
}
|
||||
got.expect("one datagram decoded").expect("frame decodes")
|
||||
}
|
||||
|
||||
/// The full ASSOCIATE conversation over a duplex: greet →
|
||||
/// associate request → sentinel reply. Returns (server handle,
|
||||
/// client duplex). `egress_factory_err` simulates a refusing
|
||||
/// factory; `None` wires the echo egress.
|
||||
async fn raw_associate(
|
||||
egress: Option<EgressFactory>,
|
||||
) -> (
|
||||
tokio::task::JoinHandle<Result<(), SocksError>>,
|
||||
tokio::io::DuplexStream,
|
||||
) {
|
||||
let (client_end, server_end) = tokio::io::duplex(64 * 1024);
|
||||
let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())));
|
||||
let (handle, mut client) = raw_connect(dial).await;
|
||||
let handle = tokio::spawn(serve_socks5_over(
|
||||
server_end,
|
||||
Some(dial),
|
||||
egress,
|
||||
None,
|
||||
false,
|
||||
));
|
||||
let mut client = client_end;
|
||||
greet(&mut client).await;
|
||||
let req_target = TargetAddr::Domain("later.task".into(), 1080);
|
||||
write_request(&mut client, &Socks5Command::UDPAssociate, &req_target).await;
|
||||
write_request(
|
||||
&mut client,
|
||||
&Socks5Command::UDPAssociate,
|
||||
&TargetAddr::Domain("relay.test".into(), 1080),
|
||||
)
|
||||
.await;
|
||||
let (code, addr) = read_reply(&mut client).await;
|
||||
assert_eq!(code, 0x00, "the associate reply succeeds");
|
||||
assert_eq!(
|
||||
(addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]),
|
||||
(127, 0, 0, 1, 0, 0),
|
||||
"the reply carries the sentinel BND.ADDR 127.0.0.1:0"
|
||||
);
|
||||
(handle, client)
|
||||
}
|
||||
|
||||
/// The echo egress over the queue shape (the harness's
|
||||
/// `echo_egress`, inline): datagrams sent echo back as inner
|
||||
/// bytes.
|
||||
fn echo_egress_factory() -> EgressFactory {
|
||||
Arc::new(|| {
|
||||
Box::pin(async {
|
||||
let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel();
|
||||
Ok(Box::new(QueueEgress {
|
||||
inbound,
|
||||
outbound: tokio::sync::Mutex::new(outbound),
|
||||
}) as Box<dyn Egress + Send>)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/// The ASSOCIATE happy path: sentinel reply → datagram
|
||||
/// round-trip through the echo egress (the frame shape both
|
||||
/// directions, in-order across multiple datagrams on one
|
||||
/// association).
|
||||
#[tokio::test]
|
||||
async fn udp_associate_round_trip_through_the_echo_egress() {
|
||||
let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await;
|
||||
|
||||
// Multiple datagrams, in order, one association.
|
||||
let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], 53)));
|
||||
for i in 0..5u8 {
|
||||
let payload = vec![i; 10];
|
||||
client
|
||||
.write_all(&datagram_frame(&target, &payload))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
for i in 0..5u8 {
|
||||
let mut buf = vec![0u8; 2 + 7 + 10];
|
||||
client.read_exact(&mut buf).await.expect("reply frame");
|
||||
let (src, echoed) = decode_datagram(&buf);
|
||||
assert_eq!(src, target, "the echo reports the destination");
|
||||
assert_eq!(&echoed[..], &vec![i; 10][..], "in-order echo");
|
||||
}
|
||||
|
||||
// The association is alive: another datagram round-trips.
|
||||
client
|
||||
.write_all(&datagram_frame(&target, b"alive"))
|
||||
.await
|
||||
.unwrap();
|
||||
let mut buf = vec![0u8; 2 + 7 + 5];
|
||||
client.read_exact(&mut buf).await.expect("reply frame");
|
||||
let (_, echoed) = decode_datagram(&buf);
|
||||
assert_eq!(&echoed[..], b"alive");
|
||||
|
||||
// Channel EOF ends the stage cleanly (the association
|
||||
// lifetime = the channel's).
|
||||
drop(client);
|
||||
let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
|
||||
.await
|
||||
.expect("session ends on channel EOF")
|
||||
.expect("join");
|
||||
result.expect("clean session");
|
||||
}
|
||||
|
||||
/// An empty-payload datagram is a legal datagram — never EOF (the
|
||||
/// F-2 invariant at payload level, end to end): it round-trips
|
||||
/// through the echo egress and the association survives it.
|
||||
#[tokio::test]
|
||||
async fn udp_empty_payload_datagram_is_legal_not_eof() {
|
||||
let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await;
|
||||
let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 2], 53)));
|
||||
|
||||
// pre → empty (len = 7 inner: address only) → post, in order;
|
||||
// the empty datagram must echo as an empty datagram (a real
|
||||
// datagram), and the association must survive it.
|
||||
client
|
||||
.write_all(&datagram_frame(&target, b"pre"))
|
||||
.await
|
||||
.unwrap();
|
||||
let empty = datagram_frame(&target, b"");
|
||||
assert_eq!(
|
||||
empty.len(),
|
||||
2 + 7,
|
||||
"the empty payload's frame is addr-only inner"
|
||||
);
|
||||
client.write_all(&empty).await.unwrap();
|
||||
client
|
||||
.write_all(&datagram_frame(&target, b"post"))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let mut buf = vec![0u8; 2 + 7 + 3];
|
||||
client.read_exact(&mut buf).await.expect("pre echo");
|
||||
let (_, echoed) = decode_datagram(&buf);
|
||||
assert_eq!(&echoed[..], b"pre");
|
||||
|
||||
let mut buf = vec![0u8; 2 + 7];
|
||||
client.read_exact(&mut buf).await.expect("empty echo");
|
||||
let (src, echoed) = decode_datagram(&buf);
|
||||
assert_eq!(src, target);
|
||||
assert!(
|
||||
echoed.is_empty(),
|
||||
"the empty-payload datagram round-trips (never EOF)"
|
||||
);
|
||||
|
||||
let mut buf = vec![0u8; 2 + 7 + 4];
|
||||
client.read_exact(&mut buf).await.expect("post echo");
|
||||
let (_, echoed) = decode_datagram(&buf);
|
||||
assert_eq!(&echoed[..], b"post");
|
||||
|
||||
drop(client);
|
||||
let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
|
||||
.await
|
||||
.expect("session ends")
|
||||
.expect("join");
|
||||
result.expect("clean session");
|
||||
}
|
||||
|
||||
/// A undecodable frame (garbage ATYP — any post-reply control
|
||||
/// byte parses as one) fails the association loud: EOF on the
|
||||
/// next read, no silent drop.
|
||||
#[tokio::test]
|
||||
async fn udp_undecodable_frame_ends_association() {
|
||||
let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await;
|
||||
|
||||
// Hand-craft a frame whose inner bytes start with an
|
||||
// unassigned ATYP.
|
||||
let inner = vec![0x7Fu8, 1, 2, 3];
|
||||
let mut garbage_frame = (inner.len() as u16).to_be_bytes().to_vec();
|
||||
garbage_frame.extend_from_slice(&inner);
|
||||
client.write_all(&garbage_frame).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;
|
||||
match outcome {
|
||||
Err(_) => panic!("the association must end after an undecodable frame"),
|
||||
Ok(Ok(0)) | Ok(Err(_)) => {}
|
||||
Ok(Ok(n)) => panic!("EOF expected after an undecodable frame — 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");
|
||||
}
|
||||
|
||||
/// Channel EOF ends the association and the connection stays
|
||||
/// healthy (the POC's test 4, in-process shape): after the client
|
||||
/// half-closes one association, a fresh ASSOCIATE on a fresh
|
||||
/// duplex succeeds immediately (the same registration is
|
||||
/// unpoisoned).
|
||||
#[tokio::test]
|
||||
async fn udp_channel_eof_ends_association_and_connection_stays_healthy() {
|
||||
let egress = echo_egress_factory();
|
||||
let (handle, mut client) = raw_associate(Some(Arc::clone(&egress))).await;
|
||||
|
||||
// Half-close: EOF to the stage → the association ends.
|
||||
use tokio::io::AsyncWriteExt as _;
|
||||
client.shutdown().await.unwrap();
|
||||
let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
|
||||
.await
|
||||
.expect("session ends on channel EOF")
|
||||
.expect("join");
|
||||
result.expect("clean session (EOF is the clean exit)");
|
||||
drop(client);
|
||||
|
||||
// A fresh associate succeeds immediately after (the same
|
||||
// registration, unpoisoned).
|
||||
let (handle, mut client) = raw_associate(Some(Arc::clone(&egress))).await;
|
||||
let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 3], 53)));
|
||||
client
|
||||
.write_all(&datagram_frame(&target, b"fresh"))
|
||||
.await
|
||||
.unwrap();
|
||||
let mut buf = vec![0u8; 2 + 7 + 5];
|
||||
client.read_exact(&mut buf).await.expect("fresh echo");
|
||||
let (_, echoed) = decode_datagram(&buf);
|
||||
assert_eq!(&echoed[..], b"fresh");
|
||||
drop(client);
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), handle)
|
||||
.await
|
||||
.expect("session ends")
|
||||
.expect("join")
|
||||
.expect("clean");
|
||||
}
|
||||
|
||||
/// The readiness contract (the POC's load-bearing API finding,
|
||||
/// re-pinned by this suite per ADR-003): an egress whose
|
||||
/// `next_reply` is an instantly-ready `Ok(None)` must not starve
|
||||
/// the stage's channel branch — datagrams still egress and echo.
|
||||
/// (The stage tolerates spurious `Ok(None)`s by re-parking.)
|
||||
#[tokio::test]
|
||||
async fn udp_instant_ok_none_egress_does_not_starve_the_stage() {
|
||||
struct HotNoneEgress {
|
||||
inbound: tokio::sync::mpsc::UnboundedSender<Bytes>,
|
||||
outbound: tokio::sync::Mutex<tokio::sync::mpsc::UnboundedReceiver<Bytes>>,
|
||||
}
|
||||
|
||||
impl crate::egress::Egress for HotNoneEgress {
|
||||
fn send(
|
||||
&mut self,
|
||||
addr: TargetAddr,
|
||||
payload: Bytes,
|
||||
) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
|
||||
Box::pin(async move {
|
||||
let mut inner = addr
|
||||
.to_be_bytes()
|
||||
.map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
|
||||
inner.extend_from_slice(&payload);
|
||||
self.inbound
|
||||
.send(Bytes::from(inner))
|
||||
.map_err(|_| EgressError::General("egress closed".into()))
|
||||
})
|
||||
}
|
||||
|
||||
fn next_reply(
|
||||
&mut self,
|
||||
) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
|
||||
Box::pin(async move {
|
||||
// Drain one queued reply if any, then — instead of
|
||||
// parking — return an instantly-ready `Ok(None)`
|
||||
// (the contract violation the stage must tolerate;
|
||||
// an actual biased-select livelock would hang this
|
||||
// test's channel branch).
|
||||
match self.outbound.lock().await.try_recv() {
|
||||
Ok(inner) => Ok(Some(inner)),
|
||||
Err(_) => Ok(None),
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
let factory: EgressFactory = Arc::new(|| {
|
||||
Box::pin(async {
|
||||
let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel();
|
||||
Ok(Box::new(HotNoneEgress {
|
||||
inbound,
|
||||
outbound: tokio::sync::Mutex::new(outbound),
|
||||
}) as Box<dyn Egress + Send>)
|
||||
})
|
||||
});
|
||||
|
||||
let (handle, mut client) = raw_associate(Some(factory)).await;
|
||||
let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 4], 53)));
|
||||
|
||||
// The datagrams must still egress (the channel branch
|
||||
// progresses) despite the hot `Ok(None)`s.
|
||||
client
|
||||
.write_all(&datagram_frame(&target, b"hot"))
|
||||
.await
|
||||
.unwrap();
|
||||
let mut buf = vec![0u8; 2 + 7 + 3];
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), async {
|
||||
client.read_exact(&mut buf).await
|
||||
})
|
||||
.await
|
||||
.expect("the channel branch is not starved — the echo arrived")
|
||||
.expect("read the echo");
|
||||
let (_, echoed) = decode_datagram(&buf);
|
||||
assert_eq!(&echoed[..], b"hot");
|
||||
|
||||
drop(client);
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), handle)
|
||||
.await
|
||||
.expect("session ends")
|
||||
.expect("join")
|
||||
.expect("clean");
|
||||
}
|
||||
|
||||
/// A missing egress callback refuses ASSOCIATE RFC-faithfully
|
||||
/// (`CommandNotSupported` 0x07 — ADR-008's missing-callback class,
|
||||
/// never a panic), and CONNECT still serves the same registration
|
||||
/// (the regression guard).
|
||||
#[tokio::test]
|
||||
async fn missing_egress_refuses_associate_command_not_supported() {
|
||||
let (client_end, server_end) = tokio::io::duplex(64 * 1024);
|
||||
let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())));
|
||||
let handle = tokio::spawn(serve_socks5_over(server_end, Some(dial), None, None, false));
|
||||
let mut client = client_end;
|
||||
greet(&mut client).await;
|
||||
write_request(
|
||||
&mut client,
|
||||
&Socks5Command::UDPAssociate,
|
||||
&TargetAddr::Domain("no.egress".into(), 1080),
|
||||
)
|
||||
.await;
|
||||
let (code, _addr) = read_reply(&mut client).await;
|
||||
assert_eq!(code, 0x07, "ASSOCIATE refuses CommandNotSupported");
|
||||
assert_eq!(code, 0x07, "missing egress → CommandNotSupported");
|
||||
drop(client);
|
||||
let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
|
||||
.await
|
||||
.expect("session ends")
|
||||
.expect("join")
|
||||
.expect_err("refusal surfaced in-stream");
|
||||
.expect_err("the refusal surfaces in-stream");
|
||||
assert!(matches!(
|
||||
err,
|
||||
SocksError::ServerError(SocksServerError::UnknownCommand(_))
|
||||
));
|
||||
|
||||
// CONNECT still serves the same registration shape (the
|
||||
// regression guard).
|
||||
let (handle, mut client) =
|
||||
raw_connect(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())))).await;
|
||||
greet(&mut client).await;
|
||||
let (code, _addr) = read_reply_after_connect(&mut client).await;
|
||||
assert_eq!(code, 0x00, "CONNECT still serves");
|
||||
drop(client);
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), handle)
|
||||
.await
|
||||
.expect("session ends")
|
||||
.expect("join")
|
||||
.expect("clean");
|
||||
}
|
||||
|
||||
/// The egress-factory failure (post-command, pre-reply) maps to
|
||||
/// reply#1's code — no phantom datagram plane.
|
||||
#[tokio::test]
|
||||
async fn egress_factory_failure_maps_to_reply_code() {
|
||||
let factory: EgressFactory =
|
||||
Arc::new(|| Box::pin(async { Err(EgressError::Unreachable("no egress".into())) }));
|
||||
let (client_end, server_end) = tokio::io::duplex(64 * 1024);
|
||||
let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())));
|
||||
let handle = tokio::spawn(serve_socks5_over(
|
||||
server_end,
|
||||
Some(dial),
|
||||
Some(factory),
|
||||
None,
|
||||
false,
|
||||
));
|
||||
let mut client = client_end;
|
||||
greet(&mut client).await;
|
||||
write_request(
|
||||
&mut client,
|
||||
&Socks5Command::UDPAssociate,
|
||||
&TargetAddr::Domain("down.egress".into(), 1080),
|
||||
)
|
||||
.await;
|
||||
let (code, _addr) = read_reply(&mut client).await;
|
||||
assert_eq!(code, 0x03, "egress-factory refusal → NetworkUnreachable");
|
||||
drop(client);
|
||||
let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
|
||||
.await
|
||||
.expect("session ends")
|
||||
.expect("join")
|
||||
.expect_err("the refusal surfaces in-stream");
|
||||
assert!(matches!(
|
||||
err,
|
||||
SocksError::ServerError(SocksServerError::Bug("egress refused in-stream"))
|
||||
));
|
||||
}
|
||||
|
||||
/// The egress lifecycle: the factory is called once per
|
||||
/// ASSOCIATE, the stage owns the egress, and stage exit drops it
|
||||
/// (the teardown-at-birth witness — the drop flag flips when the
|
||||
/// channel EOFs, not before).
|
||||
#[tokio::test]
|
||||
async fn egress_factory_called_once_and_dropped_with_the_stage() {
|
||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
||||
|
||||
static CALLS: AtomicUsize = AtomicUsize::new(0);
|
||||
let dropped = Arc::new(AtomicBool::new(false));
|
||||
let dropped_for_factory = Arc::clone(&dropped);
|
||||
|
||||
let factory: EgressFactory = Arc::new(move || {
|
||||
CALLS.fetch_add(1, Ordering::SeqCst);
|
||||
let dropped = Arc::clone(&dropped_for_factory);
|
||||
Box::pin(async move {
|
||||
let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel();
|
||||
struct Witness {
|
||||
flag: Arc<AtomicBool>,
|
||||
inbound: tokio::sync::mpsc::UnboundedSender<Bytes>,
|
||||
outbound: tokio::sync::Mutex<tokio::sync::mpsc::UnboundedReceiver<Bytes>>,
|
||||
}
|
||||
impl crate::egress::Egress for Witness {
|
||||
fn send(
|
||||
&mut self,
|
||||
addr: TargetAddr,
|
||||
payload: Bytes,
|
||||
) -> futures::future::BoxFuture<'_, Result<(), EgressError>>
|
||||
{
|
||||
let mut inner = addr.to_be_bytes().unwrap().to_vec();
|
||||
inner.extend_from_slice(&payload);
|
||||
Box::pin(async move {
|
||||
self.inbound
|
||||
.send(Bytes::from(inner))
|
||||
.map_err(|_| EgressError::General("egress closed".into()))
|
||||
})
|
||||
}
|
||||
|
||||
fn next_reply(
|
||||
&mut self,
|
||||
) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>>
|
||||
{
|
||||
Box::pin(async move {
|
||||
match self.outbound.lock().await.recv().await {
|
||||
Some(inner) => Ok(Some(inner)),
|
||||
None => Err(EgressError::General("egress closed".into())),
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
impl Drop for Witness {
|
||||
fn drop(&mut self) {
|
||||
self.flag.store(true, Ordering::SeqCst);
|
||||
}
|
||||
}
|
||||
Ok(Box::new(Witness {
|
||||
flag: dropped,
|
||||
inbound,
|
||||
outbound: tokio::sync::Mutex::new(outbound),
|
||||
}) as Box<dyn Egress + Send>)
|
||||
})
|
||||
});
|
||||
|
||||
let (handle, mut client) = raw_associate(Some(factory)).await;
|
||||
assert_eq!(CALLS.load(Ordering::SeqCst), 1, "the factory ran once");
|
||||
assert!(
|
||||
!dropped.load(Ordering::SeqCst),
|
||||
"the egress lives while the stage runs"
|
||||
);
|
||||
|
||||
// One round-trip (the stage is alive).
|
||||
let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 4], 53)));
|
||||
client
|
||||
.write_all(&datagram_frame(&target, b"witness"))
|
||||
.await
|
||||
.unwrap();
|
||||
let mut buf = vec![0u8; 2 + 7 + 7];
|
||||
client.read_exact(&mut buf).await.expect("echo");
|
||||
let (_, echoed) = decode_datagram(&buf);
|
||||
assert_eq!(&echoed[..], b"witness");
|
||||
|
||||
// Channel EOF → stage exit → the egress drops.
|
||||
drop(client);
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), handle)
|
||||
.await
|
||||
.expect("session ends")
|
||||
.expect("join")
|
||||
.expect("clean");
|
||||
assert!(
|
||||
dropped.load(Ordering::SeqCst),
|
||||
"stage exit drops the egress (teardown-at-birth)"
|
||||
);
|
||||
}
|
||||
|
||||
/// 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.
|
||||
#[tokio::test]
|
||||
async fn connect_and_associate_serve_on_the_same_registration() {
|
||||
let targets = Arc::new(tokio::sync::Mutex::new(Vec::new()));
|
||||
let dial = echo_dial(Arc::clone(&targets));
|
||||
|
||||
// CONNECT first.
|
||||
let (client_end, server_end) = tokio::io::duplex(64 * 1024);
|
||||
let serve = tokio::spawn(serve_socks5_over(
|
||||
server_end,
|
||||
Some(Arc::clone(&dial)),
|
||||
Some(echo_egress_factory()),
|
||||
None,
|
||||
false,
|
||||
));
|
||||
let mut client = client_end;
|
||||
greet(&mut client).await;
|
||||
write_request(
|
||||
&mut client,
|
||||
&Socks5Command::TCPConnect,
|
||||
&TargetAddr::Domain("tcp.test".into(), 443),
|
||||
)
|
||||
.await;
|
||||
let (code, _addr) = read_reply(&mut client).await;
|
||||
assert_eq!(code, 0x00);
|
||||
client.write_all(b"tcp").await.unwrap();
|
||||
let mut buf = [0u8; 3];
|
||||
client.read_exact(&mut buf).await.unwrap();
|
||||
assert_eq!(&buf, b"tcp");
|
||||
drop(client);
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), serve)
|
||||
.await
|
||||
.expect("session ends")
|
||||
.expect("join")
|
||||
.expect("clean");
|
||||
|
||||
// ASSOCIATE second — the same seams.
|
||||
let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await;
|
||||
let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 5], 53)));
|
||||
client
|
||||
.write_all(&datagram_frame(&target, b"udp"))
|
||||
.await
|
||||
.unwrap();
|
||||
let mut buf = vec![0u8; 2 + 7 + 3];
|
||||
client.read_exact(&mut buf).await.expect("echo");
|
||||
let (_, echoed) = decode_datagram(&buf);
|
||||
assert_eq!(&echoed[..], b"udp");
|
||||
drop(client);
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), handle)
|
||||
.await
|
||||
.expect("session ends")
|
||||
.expect("join")
|
||||
.expect("clean");
|
||||
}
|
||||
|
||||
/// The CONNECT conversation's front half (the regression guard's
|
||||
/// helper: greet + request + reply).
|
||||
async fn read_reply_after_connect(client: &mut tokio::io::DuplexStream) -> (u8, [u8; 6]) {
|
||||
write_request(
|
||||
client,
|
||||
&Socks5Command::TCPConnect,
|
||||
&TargetAddr::Domain("echo.test".into(), 443),
|
||||
)
|
||||
.await;
|
||||
read_reply(client).await
|
||||
}
|
||||
|
||||
/// The dispatch skeleton + the two-pump shape coexist: CONNECT
|
||||
|
||||
+57
@@ -158,6 +158,23 @@ pub fn frame(addr: &TargetAddr, payload: &[u8]) -> Result<Bytes, WireError> {
|
||||
Ok(buf.freeze())
|
||||
}
|
||||
|
||||
/// Frame pre-encoded inner bytes (`addr + data` — exactly what
|
||||
/// [`crate::egress::Egress::next_reply`] returns) as one
|
||||
/// `[len: u16 BE][inner]` frame. The reply-direction half of the
|
||||
/// framing-once invariant: the egress speaks inner bytes (no length
|
||||
/// prefix), and this adds the single prefix at the wire layer. Rejects
|
||||
/// [`WireError::Oversize`] when the inner bytes exceed
|
||||
/// [`MAX_INNER_LEN`] (the u16 length field would wrap).
|
||||
pub fn frame_inner(inner: &[u8]) -> Result<Bytes, WireError> {
|
||||
if inner.len() > MAX_INNER_LEN {
|
||||
return Err(WireError::Oversize(inner.len(), MAX_INNER_LEN));
|
||||
}
|
||||
let mut buf = BytesMut::with_capacity(FRAME_LEN_FIELD + inner.len());
|
||||
buf.put_u16(inner.len() as u16);
|
||||
buf.put_slice(inner);
|
||||
Ok(buf.freeze())
|
||||
}
|
||||
|
||||
/// Incremental datagram decoder over the framed stage's read
|
||||
/// direction. Feed it whatever `read()` yields; it reassembles
|
||||
/// `[len: u16 BE][addr + data]` frames across chunk boundaries — a
|
||||
@@ -507,6 +524,46 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn frame_inner_frames_the_egress_inner_bytes_exactly_once() {
|
||||
let inner = {
|
||||
let mut v = v4(53).to_be_bytes().unwrap().to_vec();
|
||||
v.extend_from_slice(b"answer");
|
||||
v
|
||||
};
|
||||
let framed = frame_inner(&inner).unwrap();
|
||||
assert_eq!(&framed[..2], &(inner.len() as u16).to_be_bytes());
|
||||
assert_eq!(&framed[2..], &inner[..]);
|
||||
// The read direction decodes it back: the same (addr, payload)
|
||||
// pair the egress encoded (the round-trip shape).
|
||||
let mut r = DatagramReader::new();
|
||||
let dgs = r.feed(&framed);
|
||||
assert_eq!(dgs.len(), 1);
|
||||
let (src, payload) = dgs.into_iter().next().unwrap().unwrap();
|
||||
assert_eq!(src, v4(53));
|
||||
assert_eq!(&payload[..], b"answer");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn frame_inner_rejects_inner_bytes_over_the_u16_bound() {
|
||||
let err = frame_inner(&vec![0u8; MAX_INNER_LEN + 1]).unwrap_err();
|
||||
assert!(matches!(err, WireError::Oversize(65536, 65535)));
|
||||
let framed = frame_inner(&vec![0u8; MAX_INNER_LEN]).unwrap();
|
||||
assert_eq!(framed.len(), 2 + MAX_INNER_LEN);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn frame_and_frame_inner_produce_identical_wires() {
|
||||
// The reply path (frame_inner over pre-encoded inner bytes)
|
||||
// and the forward path (frame over addr + payload) agree
|
||||
// byte-for-byte — one codec, two callers.
|
||||
let addr = domain("example.com", 53);
|
||||
let payload = b"payload";
|
||||
let mut inner = addr.to_be_bytes().unwrap().to_vec();
|
||||
inner.extend_from_slice(payload);
|
||||
assert_eq!(frame_inner(&inner).unwrap(), frame(&addr, payload).unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn oversize_rejected_at_frame_time() {
|
||||
let addr = v4(80);
|
||||
|
||||
Reference in new issue
Block a user