alksocks/src/egress.rs

//! The three I/O-plane seam types (ADR-004) — the dial seam (CONNECT),
//! the egress seam (UDP ASSOCIATE), the accept seam (TCP BIND) — as
//! function-shaped callbacks with their error vocabularies. These types
//! are the inversion point (AGENTS.md convention 7): the protocol layer
//! never names a substrate type; the feature-gated providers
//! (`local`/`composed`, batch 2) and the test harness construct them.
//!
//! **Function, not trait** (ADR-004): [`DialFn`], [`EgressFactory`],
//! and [`BindAcceptFactory`] are function-shaped callbacks — "produce
//! boxed halves for a resource" is a function, not a trait (the
//! alktunnels ADR-004 finding, extended to all three SOCKS5 seams by
//! POC #1/#2/#5). [`Egress`] is the one trait: the per-association
//! egress is a returned object, not a service. The re-evaluation
//! threshold is three real implementations as trait impls in practice
//! (the alktunnels ADR-004 posture).
//!
//! Contracts (POC-pinned, load-bearing API surface):
//!
//! - **Ownership rides the handle structurally** ([`TargetHandle`],
//!   [`BindTargetHandle`]): the payloads own their halves; any
//!   keep-alive state a substrate needs (e.g. a composed variant's
//!   channel session) rides inside the halves — a provider-side owner
//!   wrapper — so dropping the handle drops everything it dialed or
//!   accepted. No `_` bindings of owning values: the
//!   teardown-at-birth discipline, compiler-enforced by the shape.
//! - **The egress speaks inner bytes**: [`Egress::send`] takes the
//!   payload only; [`Egress::next_reply`] returns `addr + data` — the
//!   frame codec lives on the wire side, applied exactly once
//!   ([`crate::wire`]).
//! - **The readiness contract** ([`Egress::next_reply`]): `Pending`
//!   when no reply is ready — an instantly-ready `Ok(None)` starves a
//!   biased `select!` into livelock (POC #2's structural finding).
//! - **The accept future owns its resources end-to-end**
//!   ([`BindAccept`]): the quiet wait's `select!` drop IS the
//!   client-gone teardown (ADR-004/ADR-007).
//! - **Refusal is RFC-faithful**: every error type maps to RFC 1928
//!   reply codes via `to_reply_error()`; a dial/egress/accept failure
//!   is reply-code-shaped (never coerced to a channels-level error —
//!   session-level refusal stays channels-level, target-level refusal
//!   is in-stream).
//!
//! The readiness-contract behavioral tests (an instant-`Ok(None)`
//! egress must not starve the datagram stage — including the
//! zero-yield shape, the gate review's R-1,
//! tasks/implementation/review-protocol-core.md) live in the producer
//! module's test suite, next to the stage they pin. This module
//! carries the error→reply-code mapping tables, the sync-lift factory
//! constructor, and the compile-shape assertions.

use std::future::Future;
use std::net::SocketAddr;
use std::sync::Arc;

use bytes::Bytes;
use futures::future::BoxFuture;
use tokio::io::{AsyncRead, AsyncWrite};

pub use crate::rfc1928::util::target_addr::TargetAddr;
use crate::rfc1928::ReplyError;

/// The dialed CONNECT target — boxed halves, `+ Sync` per the
/// `ChannelPlan` payload bound (ADR-004). Ownership rides the handle
/// structurally: the payloads own their connection halves; keep-alive
/// state a substrate needs must ride inside the halves (a provider-side
/// owner wrapper), never in a separate binding the compiler could see
/// dropped early.
pub struct TargetHandle {
    pub read: Box<dyn AsyncRead + Send + Sync + Unpin>,
    pub write: Box<dyn AsyncWrite + Send + Sync + Unpin>,
}

impl std::fmt::Debug for TargetHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TargetHandle").finish_non_exhaustive()
    }
}

/// The dial seam (ADR-004): the substrate dial function, injected at
/// registration/assembly. Called once per CONNECT at the command-read
/// interception point with the RFC 1928 target the client requested
/// (domain form passes through unresolved — ADR-012; resolution is the
/// provider's job). Per-target refusal is the returned
/// [`SocksDialError`] (the producer replies with the mapped RFC code —
/// no phantom data plane); session-level refusal stays channels-level.
pub type DialFn = Arc<
    dyn Fn(TargetAddr) -> BoxFuture<'static, Result<TargetHandle, SocksDialError>> + Send + Sync,
>;

/// A dial failure, mapped to RFC 1928 reply codes by
/// [`SocksDialError::to_reply_error`] (never a generic error —
/// AGENTS.md convention 2's faithful `ReplyError` mapping).
#[derive(Debug, thiserror::Error)]
pub enum SocksDialError {
    #[error("connection refused: {0}")]
    Refused(String),
    #[error("network unreachable: {0}")]
    NetworkUnreachable(String),
    #[error("host unreachable: {0}")]
    HostUnreachable(String),
    #[error("connection not allowed by ruleset: {0}")]
    NotAllowed(String),
    #[error("general failure: {0}")]
    General(String),
}

impl SocksDialError {
    /// The RFC 1928 reply code for the reply the producer sends on this
    /// refusal. `Refused` → `ConnectionRefused` (0x05),
    /// `NetworkUnreachable` → `NetworkUnreachable` (0x03),
    /// `HostUnreachable` → `HostUnreachable` (0x04), `NotAllowed` →
    /// `ConnectionNotAllowed` (0x02), `General` → `GeneralFailure`
    /// (0x01) — the POC #1 mapping, table-tested in this module.
    pub fn to_reply_error(&self) -> ReplyError {
        match self {
            Self::Refused(_) => ReplyError::ConnectionRefused,
            Self::NetworkUnreachable(_) => ReplyError::NetworkUnreachable,
            Self::HostUnreachable(_) => ReplyError::HostUnreachable,
            Self::NotAllowed(_) => ReplyError::ConnectionNotAllowed,
            Self::General(_) => ReplyError::GeneralFailure,
        }
    }
}

/// One association's egress failure. Mapped to RFC reply codes only at
/// association setup ([`EgressError::to_reply_error`] — an egress that
/// cannot start refuses the ASSOCIATE pre-reply); per-datagram failures
/// are not per-datagram refusals — the producer stage ends the
/// association on an `Err` from the seam (the lossy-datagram posture).
#[derive(Debug, thiserror::Error)]
pub enum EgressError {
    #[error("egress refused: {0}")]
    Refused(String),
    #[error("egress unreachable: {0}")]
    Unreachable(String),
    #[error("egress general failure: {0}")]
    General(String),
}

impl EgressError {
    /// The RFC 1928 reply code for the ASSOCIATE refusal: `Refused` →
    /// `ConnectionRefused` (0x05), `Unreachable` →
    /// `NetworkUnreachable` (0x03), `General` → `GeneralFailure`
    /// (0x01) — the POC #2 mapping, table-tested in this module.
    pub fn to_reply_error(&self) -> ReplyError {
        match self {
            Self::Refused(_) => ReplyError::ConnectionRefused,
            Self::Unreachable(_) => ReplyError::NetworkUnreachable,
            Self::General(_) => ReplyError::GeneralFailure,
        }
    }
}

/// The per-association datagram egress — the ASSOCIATE seam's returned
/// object and the module's one trait (ADR-004: the factory is
/// function-shaped; the instance it returns is a trait object,
/// `Box<dyn Egress + Send>`, owned by the producer's datagram stage —
/// stage exit drops it, so egress state never outlives the channel).
///
/// The stage never sees which substrate backs the association
/// (convention 7 at the egress layer).
pub trait Egress {
    /// Egress one datagram to `addr`. Payload-only: the producer stage
    /// decoded the wire frame and hands the bare datagram here — the
    /// egress never sees a length prefix (framing is applied exactly
    /// once, by the wire layer). A zero-length payload is a legal
    /// empty datagram, not a signal of any kind (the F-2 invariant).
    fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>>;

    /// Yield the next inbound datagram's *inner* bytes (`addr + data`
    /// — the RFC address encoding followed by the payload, exactly
    /// [`TargetAddr::to_be_bytes()`] + data). The length prefix is the
    /// wire layer's, added by the producer's frame writer — an egress
    /// that prefixes again is the double-framing bug class POC #2
    /// caught.
    ///
    /// # The readiness contract (load-bearing API surface)
    ///
    /// Must return `Pending` when no reply is ready. An
    /// instantly-ready `Ok(None)` starves a biased `select!` into
    /// livelock: the stage parks on `next_reply` ∥ the channel read,
    /// and a hot `Ok(None)` never yields to the channel branch (nor to
    /// the runtime's other tasks on a current-thread scheduler).
    /// `Ok(None)` is tolerated as a spurious wakeup (the stage
    /// re-parks), not as a readiness signal — implement the wait as a
    /// real async park. An `Err` ends the association (the stage
    /// returns; the egress drops with it).
    fn next_reply(&mut self) -> BoxFuture<'_, Result<Option<Bytes>, EgressError>>;
}

/// The egress seam (ADR-004): the per-association egress factory — the
/// [`DialFn`] analogue for the datagram stage, injected at
/// registration/assembly. Called once per UDP ASSOCIATE; the stage owns
/// the returned egress; stage exit drops it.
pub type EgressFactory =
    Arc<dyn Fn() -> BoxFuture<'static, Result<Box<dyn Egress + Send>, EgressError>> + Send + Sync>;

/// A convenience constructor: lift a sync factory into the async shape
/// the producer consumes. The sync result is captured and delivered
/// through the boxed future verbatim (an `Err` still refuses the
/// ASSOCIATE with the mapped code).
pub fn egress_factory<F>(f: F) -> EgressFactory
where
    F: Fn() -> Result<Box<dyn Egress + Send>, EgressError> + Send + Sync + 'static,
{
    Arc::new(move || {
        let result = f();
        Box::pin(async move { result })
    })
}

/// The one accepted inbound connection's handle — reply#2's BND.ADDR
/// carrier plus the data-plane halves. Ownership rides the handle
/// structurally: the payloads own their accepted halves; keep-alive
/// state a substrate needs (e.g. a composed variant's channel session)
/// rides inside the halves (a provider-side owner wrapper), so the
/// compiler enforces the teardown contract — no `_` bindings of owning
/// values (the third teardown-at-birth witness, ADR-004).
pub struct BindTargetHandle {
    /// The accepted connection's peer address (reply#2's BND.ADDR —
    /// the RFC's "client IP and port of the incoming connection").
    pub peer: TargetAddr,
    pub read: Box<dyn AsyncRead + Send + Sync + Unpin>,
    pub write: Box<dyn AsyncWrite + Send + Sync + Unpin>,
}

impl std::fmt::Debug for BindTargetHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BindTargetHandle")
            .field("peer", &self.peer)
            .finish_non_exhaustive()
    }
}

/// One BIND session's accept source. `bind_addr` is the address
/// reply#1 advertises (BND.ADDR — the real bound listener on the local
/// variant, the advertised address on the composed one); the accept
/// future is awaited exactly once, in the producer's quiet-wait phase.
///
/// **The accept future owns its resources end-to-end** (ADR-004/
/// ADR-007): whatever the provider opened (the listener, the tunnel)
/// lives inside the future, so the quiet wait's `select!` drop of that
/// future IS the client-gone teardown — the listener closes when the
/// client vanishes, with no extra teardown path.
pub struct BindAccept {
    bind_addr: SocketAddr,
    accept: BoxFuture<'static, Result<BindTargetHandle, BindAcceptError>>,
}

impl BindAccept {
    /// Build the session's accept source: `bind_addr` for reply#1, the
    /// one-inbound-accept future (owned end-to-end — see the type
    /// docs).
    pub fn new(
        bind_addr: SocketAddr,
        accept: impl Future<Output = Result<BindTargetHandle, BindAcceptError>> + Send + 'static,
    ) -> Self {
        Self {
            bind_addr,
            accept: Box::pin(accept),
        }
    }

    /// The address reply#1 advertises (BND.ADDR).
    pub fn bind_addr(&self) -> SocketAddr {
        self.bind_addr
    }

    /// Await the one inbound accept (the quiet wait's accept branch —
    /// exactly once; the future is consumed).
    pub async fn accept(self) -> Result<BindTargetHandle, BindAcceptError> {
        self.accept.await
    }
}

impl std::fmt::Debug for BindAccept {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BindAccept")
            .field("bind_addr", &self.bind_addr)
            .finish_non_exhaustive()
    }
}

/// Accept-stage failures (post-reply#1): the producer maps them to RFC
/// reply codes for reply#2 — no data plane on refusal.
#[derive(Debug, thiserror::Error)]
pub enum BindAcceptError {
    #[error("accept refused: {0}")]
    Refused(String),
    #[error("accept not allowed: {0}")]
    NotAllowed(String),
    #[error("accept general failure: {0}")]
    General(String),
}

impl BindAcceptError {
    /// The RFC 1928 reply code for reply#2 on this refusal: `Refused` →
    /// `ConnectionRefused` (0x05), `NotAllowed` →
    /// `ConnectionNotAllowed` (0x02), `General` → `GeneralFailure`
    /// (0x01) — the POC #5 mapping, table-tested in this module.
    pub fn to_reply_error(&self) -> ReplyError {
        match self {
            Self::Refused(_) => ReplyError::ConnectionRefused,
            Self::NotAllowed(_) => ReplyError::ConnectionNotAllowed,
            Self::General(_) => ReplyError::GeneralFailure,
        }
    }
}

/// Factory failures (pre-reply#1): the producer maps them to RFC reply
/// codes for reply#1 — no phantom listener.
#[derive(Debug, thiserror::Error)]
pub enum BindAcceptFactoryError {
    #[error("bind refused: {0}")]
    Refused(String),
    #[error("bind not allowed: {0}")]
    NotAllowed(String),
    #[error("bind resource shortage: {0}")]
    Shortage(String),
    #[error("bind general failure: {0}")]
    General(String),
}

impl BindAcceptFactoryError {
    /// The RFC 1928 reply code for reply#1 on this refusal: `Refused` →
    /// `ConnectionRefused` (0x05), `NotAllowed` →
    /// `ConnectionNotAllowed` (0x02), `Shortage`/`General` →
    /// `GeneralFailure` (0x01 — the RFC's catch-all; a resource
    /// shortage has no dedicated code) — the POC #5 mapping,
    /// table-tested in this module.
    pub fn to_reply_error(&self) -> ReplyError {
        match self {
            Self::Refused(_) => ReplyError::ConnectionRefused,
            Self::NotAllowed(_) => ReplyError::ConnectionNotAllowed,
            Self::Shortage(_) => ReplyError::GeneralFailure,
            Self::General(_) => ReplyError::GeneralFailure,
        }
    }
}

/// The accept seam (ADR-004): the per-BIND accept factory — the
/// [`DialFn`]'s mirror, injected at registration/assembly. Called once
/// per TCP BIND with the requested target. The requested port is the
/// only field a real bind can honor (domains are meaningless as bind
/// addresses); the address policy is the factory's (ADR-006/ADR-012:
/// the callbacks refuse — RFC reply codes).
///
/// Failure mapping is two-stage (ADR-007): factory failure
/// ([`BindAcceptFactoryError`]) → reply#1 with the mapped code;
/// accept-stage failure ([`BindAcceptError`]) → reply#2 with the
/// mapped code.
pub type BindAcceptFactory = Arc<
    dyn Fn(TargetAddr) -> BoxFuture<'static, Result<BindAccept, BindAcceptFactoryError>>
        + Send
        + Sync,
>;

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::VecDeque;

    fn domain(port: u16) -> TargetAddr {
        TargetAddr::Domain("example.com".to_owned(), port)
    }

    #[test]
    fn dial_error_maps_all_variants_to_rfc_reply_codes() {
        for (err, code) in [
            (
                SocksDialError::Refused("x".into()),
                ReplyError::ConnectionRefused,
            ),
            (
                SocksDialError::NetworkUnreachable("x".into()),
                ReplyError::NetworkUnreachable,
            ),
            (
                SocksDialError::HostUnreachable("x".into()),
                ReplyError::HostUnreachable,
            ),
            (
                SocksDialError::NotAllowed("x".into()),
                ReplyError::ConnectionNotAllowed,
            ),
            (
                SocksDialError::General("x".into()),
                ReplyError::GeneralFailure,
            ),
        ] {
            assert_eq!(
                err.to_reply_error().as_u8(),
                code.as_u8(),
                "{err} must map to {code:?}"
            );
        }
    }

    #[test]
    fn egress_error_maps_all_variants_to_rfc_reply_codes() {
        for (err, code) in [
            (
                EgressError::Refused("x".into()),
                ReplyError::ConnectionRefused,
            ),
            (
                EgressError::Unreachable("x".into()),
                ReplyError::NetworkUnreachable,
            ),
            (EgressError::General("x".into()), ReplyError::GeneralFailure),
        ] {
            assert_eq!(
                err.to_reply_error().as_u8(),
                code.as_u8(),
                "{err} must map to {code:?}"
            );
        }
    }

    #[test]
    fn bind_accept_error_maps_all_variants_to_rfc_reply_codes() {
        for (err, code) in [
            (
                BindAcceptError::Refused("x".into()),
                ReplyError::ConnectionRefused,
            ),
            (
                BindAcceptError::NotAllowed("x".into()),
                ReplyError::ConnectionNotAllowed,
            ),
            (
                BindAcceptError::General("x".into()),
                ReplyError::GeneralFailure,
            ),
        ] {
            assert_eq!(
                err.to_reply_error().as_u8(),
                code.as_u8(),
                "{err} must map to {code:?}"
            );
        }
    }

    #[test]
    fn bind_accept_factory_error_maps_all_variants_to_rfc_reply_codes() {
        for (err, code) in [
            (
                BindAcceptFactoryError::Refused("x".into()),
                ReplyError::ConnectionRefused,
            ),
            (
                BindAcceptFactoryError::NotAllowed("x".into()),
                ReplyError::ConnectionNotAllowed,
            ),
            (
                BindAcceptFactoryError::Shortage("x".into()),
                ReplyError::GeneralFailure,
            ),
            (
                BindAcceptFactoryError::General("x".into()),
                ReplyError::GeneralFailure,
            ),
        ] {
            assert_eq!(
                err.to_reply_error().as_u8(),
                code.as_u8(),
                "{err} must map to {code:?}"
            );
        }
    }

    struct QueueEgress {
        replies: VecDeque<Bytes>,
    }

    impl Egress for QueueEgress {
        fn send(
            &mut self,
            _addr: TargetAddr,
            payload: Bytes,
        ) -> BoxFuture<'_, Result<(), EgressError>> {
            Box::pin(async move {
                self.replies.push_back(payload);
                Ok(())
            })
        }

        fn next_reply(&mut self) -> BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
            Box::pin(async move { Ok(self.replies.pop_front()) })
        }
    }

    #[tokio::test]
    async fn egress_factory_lifts_a_sync_factory_into_the_async_shape() {
        let factory: EgressFactory = egress_factory(|| {
            Ok(Box::new(QueueEgress {
                replies: VecDeque::new(),
            }) as Box<dyn Egress + Send>)
        });
        let mut egress = factory().await.expect("Ok lifts verbatim");
        egress
            .send(domain(53), Bytes::from_static(b"query"))
            .await
            .expect("send succeeds");
        let reply = egress.next_reply().await.expect("next_reply succeeds");
        assert_eq!(
            reply.expect("the queued reply is ready"),
            Bytes::from_static(b"query")
        );
        assert!(egress.next_reply().await.expect("drained").is_none());
    }

    #[tokio::test]
    async fn egress_factory_delivers_a_sync_err_verbatim() {
        let factory: EgressFactory =
            egress_factory(|| Err(EgressError::Unreachable("no egress".into())));
        let err = match factory().await {
            Err(err) => err,
            Ok(_) => panic!("expected the sync Err to lift verbatim"),
        };
        assert_eq!(
            err.to_reply_error().as_u8(),
            ReplyError::NetworkUnreachable.as_u8()
        );
    }

    #[tokio::test]
    async fn dial_fn_shape_accepts_a_closure_and_yields_the_handle() {
        let dial: DialFn = Arc::new(|target: TargetAddr| {
            Box::pin(async move {
                assert_eq!(target, domain(443));
                let (client_read, target_read) = tokio::io::duplex(64);
                let (target_write, client_write) = tokio::io::duplex(64);
                std::mem::forget((client_read, client_write));
                Ok(TargetHandle {
                    read: Box::new(target_read) as Box<dyn AsyncRead + Send + Sync + Unpin>,
                    write: Box::new(target_write) as Box<dyn AsyncWrite + Send + Sync + Unpin>,
                })
            })
        });
        let handle = dial(domain(443)).await.expect("the dial resolves");
        let _ = (&handle.read, &handle.write);
    }

    #[tokio::test]
    async fn bind_accept_factory_shape_yields_bind_addr_then_the_handle() {
        let factory: BindAcceptFactory = Arc::new(|_target: TargetAddr| {
            Box::pin(async move {
                Ok(BindAccept::new(
                    SocketAddr::from(([127, 0, 0, 1], 2121)),
                    async {
                        let (client_read, target_read) = tokio::io::duplex(64);
                        let (target_write, client_write) = tokio::io::duplex(64);
                        std::mem::forget((client_read, client_write));
                        Ok(BindTargetHandle {
                            peer: TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 7], 44444))),
                            read: Box::new(target_read) as Box<dyn AsyncRead + Send + Sync + Unpin>,
                            write: Box::new(target_write)
                                as Box<dyn AsyncWrite + Send + Sync + Unpin>,
                        })
                    },
                ))
            })
        });
        let accept = factory(domain(2121)).await.expect("the factory resolves");
        assert_eq!(accept.bind_addr(), SocketAddr::from(([127, 0, 0, 1], 2121)));
        let handle = accept.accept().await.expect("the accept resolves");
        assert_eq!(
            handle.peer,
            TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 7], 44444)))
        );
    }

    #[test]
    fn seam_types_have_the_poc_compile_shapes() {
        fn assert_send<T: Send>() {}
        fn assert_send_sync<T: Send + Sync>() {}
        fn assert_unpin<T: Unpin>() {}

        assert_send_sync::<TargetHandle>();
        assert_send_sync::<BindTargetHandle>();
        assert_unpin::<TargetHandle>();
        assert_unpin::<BindTargetHandle>();

        assert_send_sync::<DialFn>();
        assert_send_sync::<EgressFactory>();
        assert_send_sync::<BindAcceptFactory>();

        assert_send::<BindAccept>();
        assert_unpin::<BindAccept>();

        assert_send_sync::<SocksDialError>();
        assert_send_sync::<EgressError>();
        assert_send_sync::<BindAcceptError>();
        assert_send_sync::<BindAcceptFactoryError>();
    }

    #[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::<Result<BindTargetHandle, BindAcceptError>>(),
        );
        let rendered = format!("{accept:?}");
        assert!(rendered.contains("BindAccept"), "{rendered}");
        assert!(
            rendered.contains("127.0.0.1:2121"),
            "the bind address is named: {rendered}"
        );
    }
}