alksocks 0.1.0

//! The `local` feature's socket-backed seam providers (ADR-008's local
//! variants, ADR-009's feature gate): the three seams' faithful-RFC
//! assembly-layer constructors — [`tcp_dial`] (CONNECT), [`udp_egress`]
//! (UDP ASSOCIATE), [`bind_accept`] (TCP BIND). These are convenience
//! constructors over the landed seam types ([`crate::egress`]); the
//! protocol layer never sees them, and **the providers ARE the explicit
//! bind decisions** — the never-binds posture (convention 9) holds
//! because the protocol layer never references this module: every bind
//! below (the dial's connect, the egress socket's `127.0.0.1:0`, the
//! BIND listener) is the caller's explicit choice to enable this
//! feature and wire the constructor.
//!
//! The seam contracts each constructor honors (the landed types'
//! normative doc comments, `crate::egress`):
//!
//! - **Ownership rides the handle structurally** — the payloads own
//!   their socket halves; dropping the handle drops the connection.
//! - **The egress speaks inner bytes** — [`Egress::send`] takes the
//!   payload only; [`Egress::next_reply`] returns
//!   `TargetAddr::to_be_bytes() + data` (the *sender's* address, the
//!   RFC's per-datagram source semantics); framing-once — the length
//!   prefix is the wire layer's ([`crate::wire`]), added exactly once
//!   downstream.
//! - **The readiness contract** ([`Egress::next_reply`]) — parks on
//!   `readable()` (a real async wait on the socket's receive
//!   readiness), then `try_recv_from`; `WouldBlock` → the
//!   spurious-wakeup `Ok(None)` shape; never a hot `Ok(None)`.
//! - **The accept future owns its resources end-to-end** — the
//!   listener lives inside the accept future; the quiet wait's drop IS
//!   the client-gone teardown.
//!
//! Resolution is the provider's job (ADR-012): domain targets/destinations
//! resolve here, at the seam — the ssh `-D` semantic, via the
//! non-blocking [`tokio::net::lookup_host`] resolver.

use std::io::ErrorKind;
use std::net::SocketAddr;
use std::sync::Arc;

use bytes::Bytes;
use futures::future::BoxFuture;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::net::{TcpListener, TcpStream, UdpSocket};

pub use crate::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
use crate::{
    BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError, BindTargetHandle,
    DialFn, Egress, EgressError, EgressFactory, SocksDialError, TargetHandle,
};

/// Resolve a [`TargetAddr`] to a `SocketAddr` (egress-side DNS; the
/// ssh `-D` semantic — ADR-012: resolution is the provider's job). IP
/// targets resolve directly; domain targets resolve via
/// [`tokio::net::lookup_host`] (the non-blocking resolver — review 003
/// S-1's blocking-`to_socket_addrs` residue retired; the first address
/// wins, upstream parity).
pub(crate) async fn resolve_target(addr: &TargetAddr) -> Result<SocketAddr, EgressError> {
    match addr {
        TargetAddr::Ip(sa) => Ok(*sa),
        TargetAddr::Domain(domain, port) => {
            let addrs = tokio::net::lookup_host((domain.as_str(), *port))
                .await
                .map_err(|e| EgressError::Unreachable(format!("resolve {domain}: {e}")))?
                .next()
                .ok_or_else(|| {
                    EgressError::Unreachable(format!("resolve {domain}: no addresses"))
                })?;
            Ok(addrs)
        }
    }
}

/// Map an io error kind to the dial vocabulary (faithful-RFC mapping;
/// `NotAllowed` is never synthesized by this provider — it is a
/// ruleset concept, not an io one).
pub(crate) fn dial_error_of(what: &str, err: &std::io::Error) -> SocksDialError {
    match err.kind() {
        ErrorKind::ConnectionRefused => SocksDialError::Refused(format!("{what}: {err}")),
        ErrorKind::NetworkUnreachable => {
            SocksDialError::NetworkUnreachable(format!("{what}: {err}"))
        }
        ErrorKind::HostUnreachable => SocksDialError::HostUnreachable(format!("{what}: {err}")),
        ErrorKind::TimedOut => SocksDialError::NetworkUnreachable(format!("{what}: {err}")),
        ErrorKind::PermissionDenied => SocksDialError::General(format!("{what}: {err}")),
        _ => SocksDialError::General(format!("{what}: {err}")),
    }
}

/// The CONNECT dial seam's local-socket variant (ADR-008): a real
/// `TcpStream` per CONNECT target; domain targets resolve here (the
/// ssh `-D` semantic). The connected stream splits into the
/// [`TargetHandle`] halves (`TcpStream::into_split` — the halves are
/// `Send + Sync + Unpin`, boxed per the seam's shape). Dial failures
/// map to [`SocksDialError`] faithfully ([`dial_error_of`]);
/// `NotAllowed` is never synthesized by this provider.
pub fn tcp_dial() -> DialFn {
    Arc::new(|target: TargetAddr| Box::pin(async move { local_tcp_dial(target).await }))
}

async fn local_tcp_dial(target: TargetAddr) -> Result<TargetHandle, SocksDialError> {
    let addrs: Vec<SocketAddr> = match &target {
        TargetAddr::Ip(sa) => vec![*sa],
        TargetAddr::Domain(domain, port) => tokio::net::lookup_host((domain.as_str(), *port))
            .await
            .map_err(|e| SocksDialError::HostUnreachable(format!("resolve {domain}: {e}")))?
            .collect(),
    };
    if addrs.is_empty() {
        return Err(SocksDialError::HostUnreachable(format!(
            "resolve {target}: no addresses"
        )));
    }
    let mut last: Option<SocksDialError> = None;
    for addr in addrs {
        match TcpStream::connect(addr).await {
            Ok(stream) => {
                let (read, write) = stream.into_split();
                return Ok(TargetHandle {
                    read: Box::new(read) as Box<dyn AsyncRead + Send + Sync + Unpin>,
                    write: Box::new(write) as Box<dyn AsyncWrite + Send + Sync + Unpin>,
                });
            }
            Err(err) => last = Some(dial_error_of(&format!("connect {target}"), &err)),
        }
    }
    Err(last.unwrap_or_else(|| {
        SocksDialError::General(format!("connect {target}: no addresses tried"))
    }))
}

/// The ASSOCIATE egress seam's local-socket variant (ADR-008): one
/// real `UdpSocket` per association, bound `127.0.0.1:0` (the
/// assembly-layer bind decision, explicit here); per-datagram
/// `send_to`/`recv_from`, no flow table; domain destinations resolve
/// at the egress ([`resolve_target`]).
///
/// **The readiness contract is load-bearing**: [`LocalEgress`]'s
/// `next_reply` parks on `sock.readable().await` (a real async park on
/// receive readiness — `writable()` is always ready for UDP and would
/// spin), then `try_recv_from` (`WouldBlock` → the spurious-wakeup
/// `Ok(None)` shape; a datagram arriving makes the socket readable).
/// Replies carry *inner* bytes (`TargetAddr::to_be_bytes() + data` —
/// the sender's address per the RFC's per-datagram source semantics;
/// framing-once — the length prefix is the wire layer's). `send` takes
/// the payload only.
pub fn udp_egress() -> EgressFactory {
    Arc::new(|| {
        Box::pin(async {
            LocalEgress::bind()
                .await
                .map(|e| Box::new(e) as Box<dyn Egress + Send>)
        })
    })
}

/// One real UDP socket per association (the POC #2 `LocalEgress`
/// shape, ported). See [`udp_egress`] for the contracts.
pub struct LocalEgress {
    sock: Arc<UdpSocket>,
}

impl LocalEgress {
    /// Bind the association's socket (the explicit bind decision:
    /// loopback ephemeral — a wildcard bind is the caller's variant
    /// to add, never this constructor's default).
    pub async fn bind() -> Result<Self, EgressError> {
        let sock = UdpSocket::bind("127.0.0.1:0")
            .await
            .map_err(|e| EgressError::General(format!("udp bind: {e}")))?;
        Ok(Self {
            sock: Arc::new(sock),
        })
    }
}

impl Egress for LocalEgress {
    fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>> {
        let sock = Arc::clone(&self.sock);
        Box::pin(async move {
            let dest = resolve_target(&addr).await?;
            sock.send_to(&payload, dest)
                .await
                .map(|_| ())
                .map_err(|e| EgressError::General(format!("udp send_to {dest}: {e}")))
        })
    }

    fn next_reply(&mut self) -> BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
        let sock = Arc::clone(&self.sock);
        Box::pin(async move {
            // A real async wait on the socket's receive readiness (a
            // datagram arriving makes it readable): the stage's
            // select! parks here with the channel read — no busy-loop,
            // no lost replies.
            sock.readable()
                .await
                .map_err(|e| EgressError::General(format!("udp readable: {e}")))?;
            let mut scratch = vec![0u8; 65535];
            match sock.try_recv_from(&mut scratch) {
                Ok((n, from)) => {
                    // The INNER bytes (`addr + data` — the RFC's
                    // per-datagram source semantics): the length
                    // prefix is the wire layer's, applied exactly
                    // once downstream.
                    let mut inner = TargetAddr::Ip(from)
                        .to_be_bytes()
                        .map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
                    inner.extend_from_slice(&scratch[..n]);
                    Ok(Some(Bytes::from(inner)))
                }
                // A spurious wakeup (the readiness contract's
                // tolerated shape — the stage re-parks; this happens
                // async, never a hot loop).
                Err(e) if e.kind() == ErrorKind::WouldBlock => Ok(None),
                Err(e) => Err(EgressError::General(format!("udp try_recv: {e}"))),
            }
        })
    }
}

/// The BIND accept seam's local-socket variant (ADR-008, the faithful
/// RFC shape, ADR-007 variant 1): a real `TcpListener` per BIND
/// session. The requested target's port is honored when non-zero (FTP
/// active mode); port 0 binds an ephemeral port; `local_addr()` is
/// reply#1's `bind_addr`. The accept future owns the listener
/// end-to-end (the quiet wait's drop IS the client-gone teardown — the
/// listener lives inside the future, never in a separate binding);
/// `accept()`'s peer is reply#2's [`BindTargetHandle::peer`]
/// (byte-exact). Factory failures map to [`BindAcceptFactoryError`]
/// (bind refused/held port → `Refused`/`Shortage` per the io error);
/// accept-stage failures → [`BindAcceptError`].
///
/// **Address policy is the caller's**: `bind_host` is taken
/// explicitly — loopback (`127.0.0.1`) is the POC's posture; a
/// wildcard bind is the caller's explicit choice, spelled here.
pub fn bind_accept(bind_host: impl Into<std::net::IpAddr>) -> BindAcceptFactory {
    let bind_host = bind_host.into();
    Arc::new(move |target: TargetAddr| {
        Box::pin(async move { local_bind_accept(bind_host, &target).await })
    })
}

async fn local_bind_accept(
    bind_host: std::net::IpAddr,
    target: &TargetAddr,
) -> Result<BindAccept, BindAcceptFactoryError> {
    let requested_port = match target {
        TargetAddr::Ip(addr) => addr.port(),
        TargetAddr::Domain(_, port) => *port,
    };
    let listener = TcpListener::bind(SocketAddr::new(bind_host, requested_port))
        .await
        .map_err(|e| match e.kind() {
            ErrorKind::PermissionDenied | ErrorKind::AddrInUse => {
                BindAcceptFactoryError::Shortage(format!("tcp bind: {e}"))
            }
            ErrorKind::AddrNotAvailable => {
                BindAcceptFactoryError::Refused(format!("tcp bind: {e}"))
            }
            _ => BindAcceptFactoryError::General(format!("tcp bind: {e}")),
        })?;
    let bind_addr = listener
        .local_addr()
        .map_err(|e| BindAcceptFactoryError::General(format!("local_addr: {e}")))?;
    Ok(BindAccept::new(bind_addr, async move {
        let (stream, peer) = listener
            .accept()
            .await
            .map_err(|e| BindAcceptError::General(format!("accept: {e}")))?;
        let peer = peer
            .to_target_addr()
            .map_err(|e| BindAcceptError::General(format!("peer addr: {e}")))?;
        let (read, write) = stream.into_split();
        Ok(BindTargetHandle {
            peer,
            read: Box::new(read) as Box<dyn AsyncRead + Send + Sync + Unpin>,
            write: Box::new(write) as Box<dyn AsyncWrite + Send + Sync + Unpin>,
        })
    }))
}

#[cfg(test)]
mod tests;

mod producer_door;

pub use producer_door::{
    ProducerDoor, ProducerDoorBuildError, ProducerDoorBuilder, ProducerDoorError, ProducerDoorGuard,
};