alksocks 0.3.2

//! The producer-side local backend (feature `local`; producer.md
//! §Producer-Side Local Backend): the same door shape in the other
//! direction: the producer serves from a real `TcpListener`, so
//! vanilla clients reach a *locally-served* SOCKS5 resource (the
//! ssh `-D` capability hosted on the producer side). Each accepted
//! TCP connection is handed to the landed substrate-free entry
//! [`Socks5Producer::serve_socks5_over`] directly; the RFC
//! conversation runs on the accepted socket, the seams dial locally
//! (`tcp_dial()` / `udp_egress()` / `bind_accept()`). No new
//! protocol-layer code: this module is pure accept-loop assembly.
//!
//! **The door surface** (ADR-005): the NoAuth shape is the primary
//! one here: the producer IS the RFC conversation on this path
//! (`serve_socks5_over` runs `accept_no_auth` in-stream); there is no
//! door-side protocol half to re-originate. A cred-bearing variant
//! (RFC 1929) would wrap `accept_password_auth` before
//! `serve_socks5_over` remains a documented follow-up, not landed (the task
//! spec lands the NoAuth shape unless a test drives the other).
//!
//! **The accept path owns the idle defense + the session cap**
//! (ADR-014's accept-path defense; an accepted socket is not a
//! channel; the cap regime does not bound it): the idle-first-byte
//! peek (non-consuming, as the producer's own greeting read picks the
//! byte up) and the max-concurrent-sessions cap (an accept beyond
//! the bound is refused door-side, the socket closed immediately, no
//! session spawned) are the same two knobs the consumer door carries
//! ([`crate::door::FrontDoor`]).
//!
//! **The BIND command gate** on this path is the builder's
//! `bind_policy` (the substrate-free entry's gate; ADR-006's
//! posture). The pump handler's identity-derived gate does **not**
//! apply here: this path has no open-op identity, and the accepted
//! socket is the whole caller; whatever policy governs it is the
//! builder's explicit `bind_policy(bool)` and the two accept-path
//! knobs.
//!
//! **Lifetime** (the alktunnels `TunnelListener` teardown posture):
//! the accept loop runs until the caller ends it; [`ProducerDoor`]
//! hands back a guard whose `close()` awaits the loop's end; dropping
//! the guard is a best-effort abort (the spawned sessions own their
//! sockets; an in-flight session is not force-closed by a drop, only
//! `close()` is the graceful full stop).

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use tokio::net::TcpListener;
use tokio::sync::{watch, Semaphore};

use crate::producer::Socks5Producer;

/// Why one accepted connection's serving task ended (the backend's
/// per-connection teardown diagnostic, logged + isolated; the loop
/// never exits on a per-connection failure).
#[derive(Debug)]
enum BackendExit {
    /// The vanilla client went away before/after the greeting.
    ClientGone,
    /// The RFC conversation ended (its `SocksError`: the mapped
    /// refusal, the protocol garbage, or the session end).
    Session(crate::SocksError),
    /// A plain I/O failure on the accepted socket.
    Io(std::io::Error),
}

impl std::fmt::Display for BackendExit {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            BackendExit::ClientGone => f.write_str("client gone"),
            BackendExit::Session(err) => write!(f, "session: {err:#}"),
            BackendExit::Io(err) => write!(f, "io: {err}"),
        }
    }
}

/// The producer-side local backend (feature `local`): a real
/// `TcpListener` front door for [`Socks5Producer`]; each accepted
/// TCP connection is served via `serve_socks5_over` (the NoAuth
/// in-stream RFC conversation; the seams dial locally). The accept
/// path owns the two ADR-014 defense knobs (the idle-first-byte
/// window + the max-concurrent-sessions cap).
pub struct ProducerDoor {
    bind_addr: Option<SocketAddr>,
    producer: Socks5Producer,
    idle_first_byte: Duration,
    max_sessions: usize,
}

impl ProducerDoor {
    /// The builder (the consumer door's sibling shape).
    pub fn builder() -> ProducerDoorBuilder {
        ProducerDoorBuilder::default()
    }
}

/// The builder for [`ProducerDoor`]. The producer and the bind
/// address are required (the explicit bind decision, never implied);
/// the two defense knobs carry the consumer door's defaults.
#[derive(Default)]
pub struct ProducerDoorBuilder {
    producer: Option<Socks5Producer>,
    bind_addr: Option<SocketAddr>,
    idle_first_byte: Option<Duration>,
    max_sessions: Option<usize>,
}

impl ProducerDoorBuilder {
    /// The producer whose `serve_socks5_over` serves every accepted
    /// connection (the landed assembly shape, with its seams dial/egress/
    /// accept; its `bind_policy` is this path's BIND gate).
    pub fn producer(mut self, producer: Socks5Producer) -> Self {
        self.producer = Some(producer);
        self
    }

    /// The local bind address: the backend's explicit bind decision
    /// (the never-binds posture: the protocol layer never binds; this
    /// is the assembly layer's spelled-out choice, with loopback the
    /// documented posture).
    pub fn bind(mut self, addr: impl Into<SocketAddr>) -> Self {
        self.bind_addr = Some(addr.into());
        self
    }

    /// The idle-first-byte window (ADR-014 knob 1): an accepted
    /// connection that sends no first byte within this window is
    /// closed door-side. The window covers the first byte of the RFC
    /// conversation, the greeting (`05 ..`), read producer-side by
    /// `serve_socks5_over`'s own handshake (the peek is non-consuming).
    pub fn idle_first_byte(mut self, window: Duration) -> Self {
        self.idle_first_byte = Some(window);
        self
    }

    /// The max-concurrent-*connections* cap (ADR-014 knob 2; review
    /// 003 N-3's naming: the mechanism bounds accepted connections,
    /// including ones past the idle gate that are merely slow
    /// readers, not active RFC sessions): an accept beyond the bound
    /// is refused door-side; the socket is closed immediately, no
    /// session spawned.
    pub fn max_sessions(mut self, cap: usize) -> Self {
        self.max_sessions = Some(cap);
        self
    }

    pub fn build(self) -> Result<ProducerDoor, ProducerDoorBuildError> {
        let producer = self
            .producer
            .ok_or(ProducerDoorBuildError::MissingProducer)?;
        Ok(ProducerDoor {
            bind_addr: self.bind_addr,
            producer,
            idle_first_byte: self.idle_first_byte.unwrap_or(Duration::from_secs(30)),
            max_sessions: self.max_sessions.unwrap_or(256),
        })
    }
}

/// Why the producer-door builder refused to construct (the builder's
/// typed construction error, a missing required field; convention 2:
/// no panic path).
#[derive(Debug, thiserror::Error)]
pub enum ProducerDoorBuildError {
    #[error("the backend requires a producer (.producer); there is nothing to serve")]
    MissingProducer,
}

/// Why the producer door's `serve` refused to start (the typed
/// construction error: `serve` carries the bind decision, so an
/// omitted `.bind` is its error, not a panic; convention 2).
#[derive(Debug, thiserror::Error)]
pub enum ProducerDoorError {
    #[error("the backend requires a bind address (.bind) before serve; the bind decision is never implied")]
    MissingBindAddr,
    #[error("tcp bind: {0}")]
    Bind(#[source] std::io::Error),
}

/// The serving guard: [`ProducerDoor::serve`] returns one; `close()`
/// ends the accept loop gracefully (the listener is dropped, the
/// loop's end is awaited; the spawned sessions already running own
/// their sockets and finish on their own), and dropping it aborts the
/// loop task best-effort (the alktunnels `TunnelListener` teardown
/// posture).
pub struct ProducerDoorGuard {
    done_rx: Option<watch::Receiver<()>>,
    task: Option<tokio::task::JoinHandle<Option<std::io::Error>>>,
}

impl ProducerDoorGuard {
    /// End the accept loop gracefully: the loop task returns, and the
    /// guard awaits its end. In-flight sessions are NOT force-closed;
    /// they own their sockets and finish with their clients (per-
    /// connection isolation cuts both ways). Returns the loop task's
    /// fatal outcome, if any (`None` = the loop ended by close).
    ///
    /// The receiver is dropped before the await: the loop's
    /// `done_tx.closed()` fires only when every receiver is gone;
    /// holding it across the await would deadlock the graceful stop.
    pub async fn close(mut self) -> Option<std::io::Error> {
        self.done_rx = None;
        let task = self.task.take();
        drop(self);
        match task {
            Some(task) => task.await.ok().flatten(),
            None => None,
        }
    }
}

impl Drop for ProducerDoorGuard {
    /// Best-effort abort: drop the loop task's handle (the loop ends
    /// when its listener drops, since the listener is owned by the task;
    /// an abort cancels the accept wait). Sync-only, never awaits.
    fn drop(&mut self) {
        if let Some(task) = self.task.take() {
            task.abort();
        }
    }
}

impl ProducerDoor {
    /// Bind the configured address and run the accept loop. Per-
    /// connection failures are logged + isolated; the loop continues
    /// (the alktunnels per-accept isolation posture). The listener's
    /// lifetime is the caller's: the returned [`ProducerDoorGuard`]
    /// ends it (`close()` graceful; drop best-effort abort).
    pub async fn serve(self) -> Result<ProducerDoorGuard, ProducerDoorError> {
        let ProducerDoor {
            bind_addr,
            producer,
            idle_first_byte,
            max_sessions,
        } = self;
        let bind_addr = bind_addr.ok_or(ProducerDoorError::MissingBindAddr)?;
        let listener = TcpListener::bind(bind_addr)
            .await
            .map_err(ProducerDoorError::Bind)?;
        Ok(serve_on_parts(
            listener,
            idle_first_byte,
            max_sessions,
            producer,
        ))
    }

    /// Run the accept loop over an existing listener (the test/
    /// assembly seam: the same accept loop without the bind).
    pub async fn serve_on(self, listener: TcpListener) -> ProducerDoorGuard {
        let ProducerDoor {
            producer,
            idle_first_byte,
            max_sessions,
            ..
        } = self;
        serve_on_parts(listener, idle_first_byte, max_sessions, producer)
    }
}

fn serve_on_parts(
    listener: TcpListener,
    idle_first_byte: Duration,
    max_sessions: usize,
    producer: Socks5Producer,
) -> ProducerDoorGuard {
    let (done_tx, done_rx) = watch::channel(());
    let task = tokio::spawn(async move {
        let live = Arc::new(Semaphore::new(max_sessions));
        loop {
            tokio::select! {
                _ = done_tx.closed() => return None,
                accepted = listener.accept() => {
                    let (stream, peer) = match accepted {
                        Ok(pair) => pair,
                        Err(err) => {
                            tracing::warn!(error = %err, "producer door: accept failed");
                            continue;
                        }
                    };

                    // The session cap (ADR-014 knob 2): refuse
                    // beyond the bound; `try_acquire` keeps the
                    // accept loop free (a permit returns when a
                    // serving task ends and drops its guard).
                    let Ok(permit) = live.clone().try_acquire_owned() else {
                        tracing::warn!(peer = %peer, "producer door: session cap reached, refusing");
                        drop(stream);
                        continue;
                    };

                    let producer = producer.clone();
                    tokio::spawn(async move {
                        let _permit = permit;
                        let exit = serve_accepted(stream, &producer, idle_first_byte).await;
                        tracing::debug!(peer = %peer, "producer door: connection ended: {exit}");
                    });
                }
            }
        }
    });
    ProducerDoorGuard {
        done_rx: Some(done_rx),
        task: Some(task),
    }
}

/// Serve one accepted connection: the idle-first-byte defense first
/// (ADR-014 knob 1: the peek waits for readability without
/// consuming; the producer's own greeting read picks the byte up),
/// then the substrate-free entry (`serve_socks5_over`, the NoAuth
/// in-stream RFC conversation; the seams dial locally).
async fn serve_accepted(
    stream: tokio::net::TcpStream,
    producer: &Socks5Producer,
    idle: Duration,
) -> BackendExit {
    let mut first = [0u8; 1];
    match tokio::time::timeout(idle, stream.peek(&mut first)).await {
        Ok(Ok(_)) => {}
        Ok(Err(err)) => return BackendExit::Io(err),
        Err(_) => return BackendExit::ClientGone,
    }

    match producer.serve_socks5_over(stream).await {
        Ok(()) => BackendExit::ClientGone,
        Err(err) => BackendExit::Session(err),
    }
}