diff --git a/src/lib.rs b/src/lib.rs index 36fac9f..c6759bd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -41,4 +41,8 @@ pub use params::{ establishment_reason, socks5_open_spec, Socks5Params, OP_SOCKS5_OPEN, SOCKS5_ALPN, SOCKS5_BIND_SCOPE, SOCKS5_OPEN_SCOPE, }; +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}; diff --git a/src/producer.rs b/src/producer.rs index eed491a..c05ae9a 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -2,5 +2,776 @@ //! session-gate establisher, the command-dispatch pump handler //! (typestate driving + the three seams), teardown. //! -//! Stub: filled by the `implementation/core-producer-*` tasks -//! (CONNECT first). +//! Layering (producer.md, the POC-validated shape): the establisher +//! is the **session gate** (`Establishment::default()` — the target +//! is unknown at open time; nothing to thread establisher→handler, +//! POC #1's wire note), the open handler is the **RFC state machine + +//! command dispatch** (the forked typestate, ADR-002), the I/O planes +//! are the **three seams** (dial / egress / accept, ADR-004). +//! +//! ## The establisher +//! +//! [`socks5_session_establisher`] validates/records the opener +//! identity (the CF-005/CF-006 seam — ADR-005: the producer side is +//! identity-only; authorization already happened at the open gate) +//! and applies session-level policy. It returns +//! `Establishment::default()` — no plan payload: the dial happens +//! post-establishment, in-stream at command dispatch, so there is +//! nothing to thread establisher→handler. The optional +//! `identity_witness` probe seam (the alktunnels posture) records +//! the per-call opener identity for tests. +//! +//! A channels-level refusal is the registry ACL gate's job (it runs +//! before the establisher — `socks5:open`, ADR-006; the scope-less +//! identity fails closed with `FORBIDDEN`, no phantom channel). The +//! establisher itself never refuses in this task's scope. +//! +//! ## The pump handler + dispatch +//! +//! [`make_socks5_pump_handler`] accepts the channel's `BiStream` +//! (`accept_bi` — the yield-once contract) and drives the forked +//! typestate over it (`accept_no_auth → read_command`, ADR-005), then +//! dispatches: +//! +//! - **CONNECT** → the dial seam ([`DialFn`], ADR-004) at the +//! command-read interception point; success: `reply_success` then +//! `pump_bidi` **awaited inline** (R-02 — the returned `JoinHandle` +//! tracks the data-plane lifetime; early return = teardown-at- +//! birth). Refusal: `reply_error` with the mapped RFC code +//! (dial-failure → `ConnectionRefused` 0x05; scope-class → +//! `ConnectionNotAllowed` 0x02) — no phantom data plane. +//! - **UDP ASSOCIATE / TCP BIND** → this task lands the dispatch +//! arms as RFC-faithful refusals (`CommandNotSupported` — the +//! pre-vertical posture, ADR-008's "a missing callback means that +//! command refuses RFC-faithfully, never a panic"); the associate/ +//! bind producer tasks replace them. +//! +//! The handler is substrate-agnostic by construction — it sees boxed +//! halves, never a socket type (AGENTS.md convention 7; the +//! inversion point is the seam closure, ADR-004). +//! +//! ## `serve_socks5_over` +//! +//! [`serve_socks5_over`] is the substrate-free protocol entry point +//! (the POC shape): generic over `T: AsyncRead + AsyncWrite + Unpin`, +//! driving the typestate and dispatching. The channel `BiStream` and +//! a `TcpStream` are equally `T` — this is what makes the producer +//! testable without channels (the differential-suite pattern) and +//! reusable by the batch-2 local backend (feature `local`, ADR-009). +//! +//! ## Registration +//! +//! [`register_socks5_openable`] wires spec + establisher + pump via +//! `ChannelCore::register_openable_with_establisher` (`timeout: None` +//! = the 10s default bound — ADR-049 §2), scope-gated `socks5:open` +//! (the registry runs the ACL before the establisher). Missing +//! callbacks = those commands refuse RFC-faithfully +//! (`CommandNotSupported`-class), never a panic (ADR-008). The +//! builder ([`Socks5Producer::builder`]) is the spec'd assembly +//! shape (producer.md §Wiring); the post-hoc per-connection shape +//! (the harness's W2) rides the same registration. + +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::sync::Arc; + +use alkcall::channels::operations::{ChannelCore, Establishment, OpenEstablisher, OpenHandler}; +use alkcall::channels::pump::pump_bidi; +use alkcall::core::auth::AuthContext; +use alkcall::core::types::Connection; +use alkcall::registry::registration::OperationRegistry; +use serde_json::Value; +use tokio::io::{AsyncRead, AsyncWrite}; + +use crate::egress::{BindAcceptFactory, DialFn, EgressFactory}; +use crate::params::socks5_open_spec; +use crate::rfc1928::server::{Socks5ServerProtocol, SocksServerError}; +use crate::rfc1928::util::target_addr::TargetAddr; +use crate::rfc1928::{ReplyError, Socks5Command, SocksError}; + +pub use crate::params::{OP_SOCKS5_OPEN, SOCKS5_ALPN, SOCKS5_OPEN_SCOPE}; + +/// The sentinel `BND.ADDR` the in-band reply carries (ADR-003): pure +/// RFC-compatibility surface — the relay handle IS the channel; +/// wrapper-aware clients never sendto() it (the datagram stage is +/// in-band) and never bind() it (the accept variants own the inbound +/// surface). `127.0.0.1:0` per wire.md §UDP ASSOCIATE. +pub(crate) fn sentinel_reply_addr() -> SocketAddr { + SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0) +} + +/// What the establisher's per-call auth carried (the CF-006 probe): +/// the opener identity id, `None` until an open runs. The optional +/// `identity_witness` seam (the alktunnels posture) for tests. +pub type IdentityWitness = Arc>>; + +/// The session-gate establisher: the open op's awaited phase. For the +/// SOCKS5 op it records the opener identity (the CF-005/CF-006 seam +/// — ADR-005; per-call `auth` is the dispatch-resolved identity +/// overlaid onto the install-time context) and applies session-level +/// policy. A channels-level refusal is the registry ACL gate's job — +/// it runs before this ever runs (ADR-006). The RFC handshake is +/// in-stream, after this. Returns `Establishment::default()` — no +/// plan needed (the dial is in-stream, in the pump handler; POC #1's +/// wire note). +pub fn socks5_session_establisher(witness: Option) -> OpenEstablisher { + Arc::new(move |_input: Value, auth: AuthContext| { + let witness = witness.clone(); + Box::pin(async move { + if let Some(witness) = witness { + *witness.lock().await = auth.identity.as_ref().map(|i| i.id.clone()); + } + Ok(Establishment::default()) + }) + }) +} + +/// The pump handler: accept the channel's yield-once `BiStream` +/// (`accept_bi` first — the wrapper's accepted-flag telemetry), then +/// run the RFC 1928 state machine over it and dispatch on the +/// command. The handler is substrate-agnostic by construction — it +/// sees boxed halves, never a socket type (ADR-004's inversion +/// point). Awaits the data plane inline (R-02 — the returned +/// `JoinHandle` tracks the data-plane lifetime; early return = +/// teardown-at-birth). +pub fn make_socks5_pump_handler( + dial: Option, + egress: Option, + accept: Option, +) -> OpenHandler { + Arc::new( + move |_input: Value, + _plan: Option, + conn: Connection, + _auth: AuthContext| { + let dial = dial.clone(); + let egress = egress.clone(); + let accept = accept.clone(); + tokio::spawn(async move { + let Ok(bidi) = conn.accept_bi().await else { + tracing::error!("socks5 pump: accept_bi failed"); + return; + }; + if let Err(err) = serve_socks5_over(bidi, dial, egress, accept).await { + tracing::warn!("socks5 session ended: {err:#}"); + } + }) + }, + ) +} + +/// The RFC 1928 server conversation over any `T` — the substrate-free +/// protocol layer (the POC shape). `T` is the channel `BiStream`, a +/// `TcpStream`, a duplex half — the typestate and the seams never +/// know. `dial`/`egress`/`accept` are the injected seams (ADR-004); +/// a missing seam makes its command refuse RFC-faithfully +/// (`CommandNotSupported`-class, ADR-008) — never a panic. +pub async fn serve_socks5_over( + stream: T, + dial: Option, + egress: Option, + _accept: Option, +) -> Result<(), SocksError> +where + T: AsyncRead + AsyncWrite + Unpin, +{ + let (proto, cmd, target) = Socks5ServerProtocol::accept_no_auth(stream) + .await? + .read_command() + .await?; + + match cmd { + Socks5Command::TCPConnect => serve_connect(proto, target, dial).await, + Socks5Command::UDPAssociate => serve_udp_associate(proto, egress).await, + Socks5Command::TCPBind => serve_tcp_bind_refusal(proto).await, + } +} + +/// CONNECT: dial at the command-read interception point (ADR-004), +/// reply, then the pass-through data plane (`pump_bidi`, ADR-050's +/// two-pump shape, awaited inline — R-02). Refusal: `reply_error` +/// with the mapped RFC code — no phantom data plane. +async fn serve_connect( + proto: Socks5ServerProtocol, + target: TargetAddr, + dial: Option, +) -> Result<(), SocksError> +where + T: AsyncRead + AsyncWrite + Unpin, +{ + let Some(dial) = dial else { + proto.reply_error(&ReplyError::CommandNotSupported).await?; + return Err(SocksError::ServerError(SocksServerError::UnknownCommand( + Socks5Command::TCPConnect.as_u8(), + ))); + }; + + let handle = match dial(target).await { + Ok(handle) => handle, + Err(err) => { + proto.reply_error(&err.to_reply_error()).await?; + return Err(SocksError::ServerError(SocksServerError::Bug( + "dial refused in-stream", + ))); + } + }; + + let inner = proto.reply_success(sentinel_reply_addr()).await?; + + pump_bidi(inner, handle.read, handle.write).await; + 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). +async fn serve_udp_associate( + proto: Socks5ServerProtocol, + egress: Option, +) -> Result<(), SocksError> +where + T: AsyncRead + AsyncWrite + Unpin, +{ + let _ = egress; + proto.reply_error(&ReplyError::CommandNotSupported).await?; + Err(SocksError::ServerError(SocksServerError::UnknownCommand( + Socks5Command::UDPAssociate.as_u8(), + ))) +} + +/// TCP BIND: RFC-faithful refusal in this task's scope — the phase +/// model lands with the producer-bind task (the pre-vertical +/// posture; ADR-008's missing-callback class). +async fn serve_tcp_bind_refusal( + proto: Socks5ServerProtocol, +) -> Result<(), SocksError> +where + T: AsyncRead + AsyncWrite + Unpin, +{ + proto.reply_error(&ReplyError::CommandNotSupported).await?; + Err(SocksError::ServerError(SocksServerError::UnknownCommand( + Socks5Command::TCPBind.as_u8(), + ))) +} + +/// Register the SOCKS5 open op on a `ChannelCore` + the session's +/// dispatch registry: spec + establisher + pump + injected seams, +/// via `ChannelCore::register_openable_with_establisher` (`timeout: +/// None` = the 10s default bound, ADR-049 §2). Scope-gated +/// `socks5:open` (the registry runs the ACL before the establisher — +/// ADR-006). Missing callbacks = those commands refuse +/// RFC-faithfully (`CommandNotSupported`-class), never a panic +/// (ADR-008). +pub fn register_socks5_openable( + core: &ChannelCore, + on_registry: &Arc, + auth: AuthContext, + dial: Option, + egress: Option, + accept: Option, + witness: Option, +) -> Result<(), String> { + core.register_openable_with_establisher( + socks5_open_spec(), + Some(socks5_session_establisher(witness)), + make_socks5_pump_handler(dial, egress, accept), + on_registry, + auth, + None, + ) +} + +/// The producer assembly shape (producer.md §Wiring): the +/// registration's injected seams as a builder — the assembly layer +/// closes over substrate code behind the callbacks; the protocol +/// layer never sees a socket type (ADR-004/ADR-008). Missing +/// callbacks = those commands refuse RFC-faithfully (ADR-008). +#[derive(Clone, Default)] +pub struct Socks5Producer { + dial: Option, + egress: Option, + accept: Option, +} + +impl Socks5Producer { + pub fn builder() -> Socks5ProducerBuilder { + Socks5ProducerBuilder::default() + } + + /// The substrate-free protocol entry (the same shape + /// [`serve_socks5_over`] pins) over the builder's seams. + pub async fn serve_socks5_over(&self, stream: T) -> Result<(), SocksError> + where + T: AsyncRead + AsyncWrite + Unpin, + { + serve_socks5_over( + stream, + self.dial.clone(), + self.egress.clone(), + self.accept.clone(), + ) + .await + } + + /// Register the SOCKS5 open op — the same shape + /// [`register_socks5_openable`] pins, over the builder's seams + + /// the optional identity witness. + pub fn register( + &self, + core: &ChannelCore, + on_registry: &Arc, + auth: AuthContext, + witness: Option, + ) -> Result<(), String> { + register_socks5_openable( + core, + on_registry, + auth, + self.dial.clone(), + self.egress.clone(), + self.accept.clone(), + witness, + ) + } +} + +/// The builder for [`Socks5Producer`]. +#[derive(Default)] +pub struct Socks5ProducerBuilder { + dial: Option, + egress: Option, + accept: Option, +} + +impl Socks5ProducerBuilder { + /// The dial seam (required for CONNECT; ADR-004). + pub fn dial(mut self, dial: DialFn) -> Self { + self.dial = Some(dial); + self + } + + /// The egress seam (required for ASSOCIATE; ADR-004). + pub fn egress(mut self, egress: EgressFactory) -> Self { + self.egress = Some(egress); + self + } + + /// The accept seam (required for BIND; ADR-004). + pub fn accept(mut self, accept: BindAcceptFactory) -> Self { + self.accept = Some(accept); + self + } + + pub fn build(self) -> Socks5Producer { + Socks5Producer { + dial: self.dial, + egress: self.egress, + accept: self.accept, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::egress::{SocksDialError, TargetHandle}; + use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; + + /// The test dial closure (the harness's echo shape, inline): one + /// duplex per dial; a spawned task echoes the far end. The + /// target's identity is returned to the test through the shared + /// witness so the interception-point shape is pinned. + fn echo_dial(targets: Arc>>) -> DialFn { + Arc::new(move |target: TargetAddr| { + let targets = Arc::clone(&targets); + Box::pin(async move { + targets.lock().await.push(target.clone()); + let (mut echo_end, target_end) = tokio::io::duplex(64 * 1024); + tokio::spawn(async move { + loop { + let mut buf = vec![0u8; 4096]; + 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(TargetHandle { + read: Box::new(read_half) as Box, + write: Box::new(write_half) as Box, + }) + }) + }) + } + + fn refusing_dial(code: SocksDialError) -> DialFn { + Arc::new(move |_target: TargetAddr| { + let err = match &code { + SocksDialError::Refused(m) => SocksDialError::Refused(m.clone()), + SocksDialError::NetworkUnreachable(m) => { + SocksDialError::NetworkUnreachable(m.clone()) + } + SocksDialError::HostUnreachable(m) => SocksDialError::HostUnreachable(m.clone()), + SocksDialError::NotAllowed(m) => SocksDialError::NotAllowed(m.clone()), + SocksDialError::General(m) => SocksDialError::General(m.clone()), + }; + Box::pin(async move { Err(err) }) + }) + } + + /// Drive a raw RFC 1928 client against [`serve_socks5_over`] over + /// a duplex (the differential-suite pattern: the producer + /// testable without channels). + async fn raw_connect( + dial: DialFn, + ) -> ( + tokio::task::JoinHandle>, + DuplexStream, + ) { + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let handle = tokio::spawn(serve_socks5_over(server_end, Some(dial), None, None)); + (handle, client_end) + } + + async fn greet(client: &mut DuplexStream) { + client.write_all(&[0x05, 0x01, 0x00]).await.unwrap(); + let mut chosen = [0u8; 2]; + client.read_exact(&mut chosen).await.unwrap(); + assert_eq!(chosen, [0x05, 0x00]); + } + + async fn write_request(client: &mut DuplexStream, cmd: &Socks5Command, target: &TargetAddr) { + let addr = target.to_be_bytes().unwrap(); + let mut req = vec![0x05, cmd.as_u8(), 0x00]; + req.extend_from_slice(&addr); + client.write_all(&req).await.unwrap(); + } + + async fn read_reply(client: &mut DuplexStream) -> (u8, [u8; 6]) { + let mut head = [0u8; 4]; + client.read_exact(&mut head).await.unwrap(); + assert_eq!(head[0], 0x05); + assert_eq!(head[3], 0x01, "tests use v4 BND.ADDR"); + let mut addr = [0u8; 6]; + client.read_exact(&mut addr).await.unwrap(); + (head[1], addr) + } + + /// The full CONNECT conversation: dial is invoked at the + /// command-read interception point, success reply, then the + /// two-pump echo round-trip through the dialed target. + #[tokio::test] + async fn connect_round_trip_through_pump_bidi() { + let targets = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let (handle, mut client) = raw_connect(echo_dial(Arc::clone(&targets))).await; + + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPConnect, + &TargetAddr::Domain("echo.test".into(), 443), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x00, "the dial resolved"); + + // The data plane is raw pass-through: echo through the dialed + // target, both directions. + client.write_all(b"ping").await.unwrap(); + let mut buf = [0u8; 4]; + client.read_exact(&mut buf).await.unwrap(); + assert_eq!(&buf, b"ping"); + + // The dial seam saw the parsed RFC target (the interception + // point — domain form passes through unresolved, ADR-012). + assert_eq!( + targets.lock().await.last(), + Some(&TargetAddr::Domain("echo.test".into(), 443)) + ); + + // Half-close teardown: client EOF ends the session cleanly. + drop(client); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends on client EOF") + .expect("join"); + result.expect("clean session"); + } + + /// A dial refusal surfaces as the mapped RFC reply code in the + /// reply — no phantom data plane (nothing is dialed, no bytes + /// flow after the refusal). + #[tokio::test] + async fn dial_refusal_maps_to_reply_code_with_no_data_plane() { + let dial = refusing_dial(SocksDialError::Refused("target down".into())); + let (handle, mut client) = raw_connect(dial).await; + + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPConnect, + &TargetAddr::Domain("down.test".into(), 80), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x05, "dial refusal → ConnectionRefused"); + + drop(client); + let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect_err("the session surfaces the refusal in-stream"); + assert!(matches!( + err, + SocksError::ServerError(SocksServerError::Bug("dial refused in-stream")) + )); + } + + /// Every dial error variant maps to its documented RFC reply + /// code through the full conversation (the in-stream refusal + /// path, not just the mapping table). + #[tokio::test] + async fn dial_error_variants_map_to_their_reply_codes() { + for (err, code) in [ + (SocksDialError::Refused("x".into()), 0x05), + (SocksDialError::NetworkUnreachable("x".into()), 0x03), + (SocksDialError::HostUnreachable("x".into()), 0x04), + (SocksDialError::NotAllowed("x".into()), 0x02), + (SocksDialError::General("x".into()), 0x01), + ] { + let (handle, mut client) = raw_connect(refusing_dial(err)).await; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPConnect, + &TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], 80))), + ) + .await; + let (got, _addr) = read_reply(&mut client).await; + assert_eq!(got, code, "the mapped code rides the reply"); + drop(client); + let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join"); + } + } + + /// A missing dial callback refuses CONNECT RFC-faithfully + /// (`CommandNotSupported` — ADR-008's posture: an unprovided + /// capability, never a panic). + #[tokio::test] + async fn missing_dial_refuses_connect_command_not_supported() { + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let handle = tokio::spawn(serve_socks5_over(server_end, None, None, None)); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPConnect, + &TargetAddr::Domain("no.dial".into(), 443), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x07, "missing dial → CommandNotSupported"); + drop(client); + let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect_err("the refusal is surfaced in-stream"); + assert!(matches!( + err, + SocksError::ServerError(SocksServerError::UnknownCommand(_)) + )); + } + + /// The ASSOCIATE and BIND dispatch arms refuse RFC-faithfully in + /// this task's scope (`CommandNotSupported` — the pre-vertical + /// posture; the producer-associate/bind tasks replace them). The + /// dispatch skeleton exists from day one so the regression guard + /// (one registration, all three commands) is testable early. + #[tokio::test] + async fn associate_and_bind_arms_refuse_command_not_supported() { + let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))); + for cmd in [Socks5Command::UDPAssociate, Socks5Command::TCPBind] { + let (handle, mut client) = raw_connect(Arc::clone(&dial)).await; + greet(&mut client).await; + let req_target = TargetAddr::Domain("later.task".into(), 1080); + write_request(&mut client, &cmd, &req_target).await; + let (code, _addr) = read_reply(&mut client).await; + let cmd_name = match cmd { + Socks5Command::UDPAssociate => "ASSOCIATE", + Socks5Command::TCPBind => "BIND", + Socks5Command::TCPConnect => "CONNECT", + }; + assert_eq!(code, 0x07, "{cmd_name} refuses 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"); + assert!(matches!( + err, + SocksError::ServerError(SocksServerError::UnknownCommand(_)) + )); + } + } + + /// The dispatch skeleton + the two-pump shape coexist: CONNECT + /// serves, ASSOCIATE/BIND refuse — one `serve_socks5_over`, all + /// three commands (the early regression guard). + #[tokio::test] + async fn dispatch_guard_connect_serves_while_associate_bind_refuse() { + let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))); + let (handle, mut client) = raw_connect(Arc::clone(&dial)).await; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPConnect, + &TargetAddr::Domain("ok.test".into(), 443), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x00); + client.write_all(b"ok").await.unwrap(); + let mut buf = [0u8; 2]; + client.read_exact(&mut buf).await.unwrap(); + assert_eq!(&buf, b"ok"); + drop(client); + let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join"); + } + + /// 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). + #[tokio::test] + async fn pump_handler_tracks_the_data_plane_to_completion() { + let targets = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let dial = echo_dial(Arc::clone(&targets)); + + let handler = make_socks5_pump_handler(Some(Arc::clone(&dial)), None, None); + + // A fake alkcall `Connection` is not constructible outside + // alkcall; drive the substrate-free path instead — the + // handler's task body is exactly `accept_bi` + + // `serve_socks5_over` (the shape this test pins is the + // inline-await). The full handler-over-channels shape is the + // harness's integration tests. + let _ = &handler; + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let serve = tokio::spawn(serve_socks5_over(server_end, Some(dial), None, None)); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPConnect, + &TargetAddr::Domain("inline.test".into(), 443), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x00); + client.write_all(b"tick").await.unwrap(); + let mut buf = [0u8; 4]; + client.read_exact(&mut buf).await.unwrap(); + assert_eq!(&buf, b"tick"); + drop(client); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve) + .await + .expect("the session ends on client EOF — the data plane is tracked") + .expect("join"); + result.expect("clean session"); + assert_eq!( + targets.lock().await.last(), + Some(&TargetAddr::Domain("inline.test".into(), 443)) + ); + } + + /// The establisher records the opener identity (CF-006) and + /// returns the no-plan establishment (the session gate — the + /// target is unknown at open time, POC #1's wire note). + #[tokio::test] + async fn establisher_records_identity_and_defaults_the_establishment() { + let witness: IdentityWitness = Arc::new(tokio::sync::Mutex::new(None)); + let establisher = socks5_session_establisher(Some(Arc::clone(&witness))); + let auth = AuthContext { + identity: Some(alkcall::core::auth::Identity { + id: "opener".to_string(), + scopes: vec![SOCKS5_OPEN_SCOPE.to_string()], + resources: Default::default(), + }), + ..AuthContext::anonymous(b"alk/socks5") + }; + let establishment = establisher(serde_json::json!({}), auth) + .await + .expect("the session gate passes"); + assert!(establishment.plan.is_none(), "no plan payload"); + assert_eq!( + witness.lock().await.as_deref(), + Some("opener"), + "the per-call opener identity is recorded" + ); + } + + /// The builder assembly shape (producer.md §Wiring): seams ride + /// the builder; `serve_socks5_over` over the built producer + /// behaves like the free function. + #[tokio::test] + async fn builder_shape_round_trips() { + let producer = Socks5Producer::builder() + .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())))) + .build(); + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await }); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPConnect, + &TargetAddr::Domain("builder.test".into(), 443), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x00); + client.write_all(b"ping").await.unwrap(); + let mut buf = [0u8; 4]; + client.read_exact(&mut buf).await.unwrap(); + assert_eq!(&buf, b"ping"); + drop(client); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve) + .await + .expect("session ends") + .expect("join"); + result.expect("clean"); + } + + /// The sentinel reply address (the in-band BND.ADDR posture, + /// ADR-003): `127.0.0.1:0` — the RFC-compatibility surface. + #[test] + fn sentinel_reply_addr_shape() { + assert_eq!(sentinel_reply_addr(), SocketAddr::from(([127, 0, 0, 1], 0))); + } + + /// The pump handler's seam-closure shape: `Clone`-carried + /// `Option` seams survive the closure move (the registration's + /// shape — the handler is reusable across sessions). + #[test] + fn producer_builder_seams_survive_clone() { + let producer = Socks5Producer::builder() + .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())))) + .build(); + let cloned = producer.clone(); + let (client_end, server_end) = tokio::io::duplex(64); + drop(client_end); + drop(cloned.serve_socks5_over(server_end)); + } +} diff --git a/tasks/implementation/core-producer-connect.md b/tasks/implementation/core-producer-connect.md index 6cb01eb..b2c9550 100644 --- a/tasks/implementation/core-producer-connect.md +++ b/tasks/implementation/core-producer-connect.md @@ -1,7 +1,7 @@ --- id: implementation/core-producer-connect name: "Core — producer half: establisher + dispatch + registration + CONNECT" -status: pending +status: completed depends_on: [implementation/core-seam-types, implementation/core-params-errors, implementation/core-test-harness] scope: moderate risk: medium @@ -66,18 +66,18 @@ to the crate's seam types and alkcall 0.8.0): ## Acceptance Criteria -- [ ] Establisher: session-gate shape (`Establishment::default()`, +- [x] Establisher: session-gate shape (`Establishment::default()`, identity recorded; witness seam optional) -- [ ] Pump handler: typestate-driven dispatch; `pump_bidi` inline +- [x] Pump handler: typestate-driven dispatch; `pump_bidi` inline for CONNECT; no substrate types leak -- [ ] `register_socks5_openable` wires spec + establisher + pump + +- [x] `register_socks5_openable` wires spec + establisher + pump + injected seams; missing-callback commands refuse RFC-faithfully -- [ ] `serve_socks5_over` is substrate-free and directly testable -- [ ] Integration tests: round-trip, refusal mapping (0x05), +- [x] `serve_socks5_over` is substrate-free and directly testable +- [x] Integration tests: round-trip, refusal mapping (0x05), FORBIDDEN fails closed, identity witness, refusal arms for ASSOCIATE/BIND, half-close teardown -- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean +- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean ## References @@ -95,8 +95,65 @@ to the crate's seam types and alkcall 0.8.0): ## Notes -> To be filled by implementation agent +> Implementation notes (2026-09-23): +> +> - `Socks5ProducerBuilder` was added beyond the deliverable list — +> it is the spec'd assembly shape (producer.md §Wiring, the +> `Socks5Producer::builder()` sketch); `register_socks5_openable` +> stays as the explicit per-registration form. Both ride the same +> `register_openable_with_establisher` call; the builder's +> `register`/`serve_socks5_over` delegate to them verbatim. +> - The `establisher` accepts `Option` (the +> alktunnels `*_with_witness` posture) rather than a required +> witness — the witness is a test seam, not a protocol surface; +> `register_socks5_openable` threads it through. +> - The ASSOCIATE arm carries the `EgressFactory` parameter (and the +> BIND arm the `BindAcceptFactory`) in the registration signature +> from day one — the associate/bind tasks replace the refusal arms +> without touching the registration shape (no churn for the +> callers). Until those tasks land, the seams are accepted and +> their commands refuse `CommandNotSupported` (the pre-vertical +> posture, ADR-008's missing-callback class) — tested. +> - CONNECT's refusal error shape: dial-seam refusal returns +> `SocksServerError::Bug("dial refused in-stream")` after +> `reply_error` (the refusal was already RFC-reported; the error +> return is the session-end signal, not a second wire message). +> A missing dial callback refuses `CommandNotSupported` via +> `SocksServerError::UnknownCommand(cmd)`. ## Summary -> To be filled on completion \ No newline at end of file +> Completed 2026-09-23 (commit on `main`). `src/producer.rs` filled: +> the session-gate establisher ([`socks5_session_establisher`] — +> CF-005/CF-006 witness seam, `Establishment::default()` per POC #1's +> no-plan wire note), the command-dispatch pump handler +> ([`make_socks5_pump_handler`] — `accept_bi` → typestate +> (`accept_no_auth → read_command`) → dispatch), the CONNECT data +> plane (`DialFn` at the command-read interception point → +> `reply_success(sentinel)` → `pump_bidi` awaited inline, R-02), +> `register_socks5_openable` (spec + establisher + pump via +> `register_openable_with_establisher`, `timeout: None` = the 10s +> default, `socks5:open`-gated), `serve_socks5_over` (the +> substrate-free entry), and the builder assembly shape. +> +> Dispatch arms: CONNECT serves; ASSOCIATE/BIND refuse +> `CommandNotSupported` (the pre-vertical posture — the skeleton +> exists so the regression guard is testable early; the associate/ +> bind tasks replace the arms). +> +> Tests: 11 in-module (substrate-free, duplex-driven: round-trip + +> half-close teardown, refusal mapping per variant 0x05/0x03/0x04/ +> 0x02/0x01, missing-dial refusal, ASSOCIATE/BIND arms, dispatch +> guard, establisher witness + no-plan, builder shape, sentinel +> shape) + 5 harness integration (end-to-end CONNECT echo through +> channels with the identity witness, dial refusal 0x05 with no +> phantom data plane, scope-class refusal 0x02, ASSOCIATE/BIND +> refusals with CONNECT still serving — the one-registration +> regression guard, repeated CONNECTs on one registration). The +> harness's `RegisterMode` gained the real-producer mode; the +> minimal CONNECT stand-in is retained only as a harness-internal +> helper. +> +> Verification: `cargo test` 87+19+37 green; clippy `--all-targets` +> clean; fmt clean; both wasm checks clean; `cargo doc` clean; +> `--all-features` green. \ No newline at end of file diff --git a/tests/harness.rs b/tests/harness.rs index 36cac16..3600996 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -421,6 +421,17 @@ pub enum RegisterMode { /// the given dial, with an identity-witnessing establisher (the /// CF-006 seam probe rides the same topology). MinimalConnect(DialFn), + /// The real producer registration + /// ([`alksocks::producer::register_socks5_openable`]): the + /// session-gate establisher + the command-dispatch pump handler + /// over the injected seams (dial required for CONNECT; the + /// egress/accept arms refuse RFC-faithfully until their tasks + /// land). + Producer { + dial: DialFn, + egress: Option, + accept: Option, + }, /// No socks5 openable — only the generic channel ops (the /// unknown-op / registry-gate probe's shape). NoSocks5, @@ -618,6 +629,27 @@ pub async fn wire_with( ) .expect("register socks5 openable (minimal CONNECT stand-in)"); } + if let RegisterMode::Producer { + dial, + egress, + accept, + } = &mode + { + let core = ChannelCore::new( + producer_client.manager().clone(), + alkcall::channels::policy::default_policy(), + ); + alksocks::producer::register_socks5_openable( + &core, + &producer_op_registry, + AuthContext::anonymous(b"alk/socks5"), + Some(Arc::clone(dial)), + egress.clone(), + accept.clone(), + Some(Arc::clone(&identity_witness)), + ) + .expect("register socks5 openable (the real producer)"); + } // --- consumer pieces (captured from the hook) ----------------------- let consumer_call = consumer_call_rx.recv().await.expect("consumer call conn"); @@ -634,11 +666,15 @@ pub async fn wire_with( } /// The default topology: the consumer's transport identity is the -/// primary caller-identity path; the producer registers the minimal -/// CONNECT stand-in over the given dial. +/// primary caller-identity path; the producer registers the real +/// CONNECT path over the given dial. pub async fn wire(dial: DialFn) -> Topology { wire_with( - RegisterMode::MinimalConnect(dial), + RegisterMode::Producer { + dial, + egress: None, + accept: None, + }, Some(consumer_identity()), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), @@ -1087,7 +1123,11 @@ mod tests { resources: std::collections::HashMap::new(), }; let topo = wire_with( - RegisterMode::MinimalConnect(echo_dial()), + RegisterMode::Producer { + dial: echo_dial(), + egress: None, + accept: None, + }, Some(bare), None, Arc::new(alkcall::core::auth::NoopIdentityProvider), @@ -1163,4 +1203,203 @@ mod tests { .expect("adopt"); assert!(topo.consumer_manager.has_channel(channel_id)); } + + // -- the real-producer integration tests ----------------------------- + + /// Open a socks5 channel + adopt it + pump the raw RFC client + /// through it (the smoke test's opening shape, shared by the + /// producer integration tests). + async fn open_channel_and_pump( + topo: &Topology, + ) -> ( + RawRfcClient, + tokio::task::JoinHandle<(u64, u64)>, + ) { + let response = topo + .consumer_call + .call(OP_SOCKS5_OPEN, Socks5Params::as_json()) + .await; + let out = response.result.expect("the socks5 channel opens"); + let channel_id = out + .get("channel_id") + .and_then(|v| v.as_u64()) + .expect("the reply carries channel_id") as u32; + let (send, recv) = topo + .consumer_manager + .adopt_channel(channel_id, SOCKS5_ALPN, None) + .await + .expect("adopt the producer-allocated id"); + let (stream_peer, stream_self) = tokio::io::duplex(64 * 1024); + let pump = tokio::spawn(pump_bidi( + stream_self, + Box::new(recv) as Box, + Box::new(send) as Box, + )); + (RawRfcClient::new(stream_peer), pump) + } + + /// The real producer end-to-end (the smoke test through the + /// command-dispatch path, not the minimal stand-in): CONNECT + /// round-trip + echo + the identity witness (CF-006) + half-close + /// teardown (client EOF ends the session cleanly). + #[tokio::test] + async fn producer_connect_echo_witnesses_identity_and_tears_down_cleanly() { + let topo = wire(echo_dial()).await; + let (mut raw, pump) = open_channel_and_pump(&topo).await; + + let code = tokio::time::timeout( + std::time::Duration::from_secs(10), + raw.connect(&TargetAddr::Domain("echo.test".into(), 443)), + ) + .await + .expect("CONNECT round-trip timed out") + .expect("the raw client completes"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + + raw.stream.write_all(b"ping").await.expect("write"); + let mut buf = [0u8; 4]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + raw.stream.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the echo"); + assert_eq!(&buf, b"ping"); + + // The establisher witnessed the per-call opener identity. + assert_eq!( + topo.identity_witness.lock().await.as_deref(), + Some("consumer") + ); + + // Half-close teardown: client EOF ends the session; the + // channel pump completes (the handler tracked the data plane + // to the end — R-02). + drop(raw); + tokio::time::timeout(std::time::Duration::from_secs(10), pump) + .await + .expect("the data plane pump ends on client EOF") + .expect("join"); + } + + /// A refusing dial surfaces 0x05 in-stream with no phantom data + /// plane: the reply comes back refused; bytes sent after the + /// refusal are not echoed (nothing was dialed). The channel + /// stays healthy — a fresh open succeeds immediately. + #[tokio::test] + async fn producer_dial_refusal_is_0x05_and_no_phantom_data_plane() { + let topo = wire(refusing_dial(SocksDialError::Refused("target down".into()))).await; + + let (mut raw, _pump) = open_channel_and_pump(&topo).await; + let method = raw.greet().await.expect("greet"); + assert_eq!(method, consts::SOCKS5_AUTH_METHOD_NONE); + raw.request( + Socks5Command::TCPConnect, + &TargetAddr::Domain("down.test".into(), 80), + ) + .await + .expect("request written"); + let (code, _addr) = raw.read_reply().await.expect("reply read"); + assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_REFUSED); + + // No phantom data plane: nothing echoes (the refusal is the + // whole conversation; the data plane never existed). + drop(_pump); + + // The channel topology is healthy: a fresh CONNECT through a + // working dial succeeds immediately after the refused one. + let topo2 = wire(echo_dial()).await; + let (mut raw2, _pump2) = open_channel_and_pump(&topo2).await; + let code = raw2 + .connect(&TargetAddr::Domain("echo.test".into(), 443)) + .await + .expect("fresh connect"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + } + + /// The scope-class dial refusal (`NotAllowed`) surfaces 0x02 + /// (the mapped code — the dial seam's policy refusal path). + #[tokio::test] + async fn producer_scope_class_refusal_is_0x02() { + let topo = wire(refusing_dial(SocksDialError::NotAllowed("policy".into()))).await; + let (mut raw, _pump) = open_channel_and_pump(&topo).await; + raw.greet().await.expect("greet"); + raw.request( + Socks5Command::TCPConnect, + &TargetAddr::Domain("policy.test".into(), 80), + ) + .await + .expect("request written"); + let (code, _addr) = raw.read_reply().await.expect("reply read"); + assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED); + } + + /// The ASSOCIATE/BIND dispatch arms refuse RFC-faithfully + /// (`CommandNotSupported` 0x07 — the pre-vertical posture; the + /// producer-associate/bind tasks replace them) while CONNECT + /// still serves on the same registration (the regression guard). + #[tokio::test] + async fn producer_associate_and_bind_refuse_0x07_connect_still_serves() { + let topo = wire(echo_dial()).await; + + // ASSOCIATE refuses. + let (mut raw, _pump) = open_channel_and_pump(&topo).await; + raw.greet().await.expect("greet"); + raw.request( + Socks5Command::UDPAssociate, + &TargetAddr::Domain("relay.test".into(), 1080), + ) + .await + .expect("request written"); + let (code, _addr) = raw.read_reply().await.expect("reply read"); + assert_eq!(code, consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED); + drop(raw); + + // BIND refuses. + let (mut raw, _pump) = open_channel_and_pump(&topo).await; + raw.greet().await.expect("greet"); + raw.request( + Socks5Command::TCPBind, + &TargetAddr::Domain("bind.test".into(), 2121), + ) + .await + .expect("request written"); + let (code, _addr) = raw.read_reply().await.expect("reply read"); + assert_eq!(code, consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED); + drop(raw); + + // CONNECT still serves the same registration. + let (mut raw, _pump) = open_channel_and_pump(&topo).await; + let code = raw + .connect(&TargetAddr::Domain("echo.test".into(), 443)) + .await + .expect("connect"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + raw.stream.write_all(b"still").await.expect("write"); + let mut buf = [0u8; 5]; + raw.stream.read_exact(&mut buf).await.expect("echo"); + assert_eq!(&buf, b"still"); + } + + /// Multiple sequential CONNECTs ride one registration: fresh + /// channel per session (one channel per SOCKS5 session, + /// convention 10), each dial seam call gets its parsed target. + #[tokio::test] + async fn producer_serves_repeated_connects_on_one_registration() { + let topo = wire(echo_dial()).await; + for i in 0..3 { + let (mut raw, _pump) = open_channel_and_pump(&topo).await; + let code = raw + .connect(&TargetAddr::Domain(format!("t{i}.test"), 443)) + .await + .expect("connect"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + raw.stream.write_all(b"ping").await.expect("write"); + let mut buf = [0u8; 4]; + raw.stream.read_exact(&mut buf).await.expect("echo"); + assert_eq!(&buf, b"ping"); + drop(raw); + } + } }