local: the producer-side local backend — the producer's front door

- ProducerDoor (feature local, src/local/producer_door.rs): a real
  TcpListener accept loop handing each accepted connection to the
  landed Socks5Producer::serve_socks5_over — the NoAuth in-stream RFC
  conversation with the local seam providers dialing/egressing/accepting;
  no new protocol-layer code.
- The accept path owns ADR-014's two knobs: the idle-first-byte peek
  (non-consuming) and the max-concurrent-sessions cap (try_acquire —
  an accept beyond the bound is refused door-side).
- ProducerDoorGuard: the listener's lifetime is the caller's — graceful
  close() awaits the loop's end; Drop aborts best-effort (the
  alktunnels TunnelListener teardown posture). In-flight sessions are
  never force-closed.
- Per-connection failures are logged + isolated (the loop continues).
- The BIND gate on this path is the builder's bind_policy (no open-op
  identity here; the pump handler's identity-derived gate does not
  apply) — pinned by the 0x02-refusal test.
- tests/local-backend-producer.rs (9): vanilla CONNECT round-trip on
  real sockets, faithful-RFC BIND (real reply#1/reply#2 addresses),
  UDP ASSOCIATE through the real egress, domain resolution at the
  seam, idle timeout, session cap, failure isolation, bind_policy(false).

Verification: cargo test + --all-features green; clippy clean (default,
all-features, wasm32 targets); cargo fmt --check clean; wasm checks
green (backend is local-gated).

src/lib.rs

@@ -113,6 +113,9 @@ pub use wire::{addr_encoding_len, frame, frame_inner, DatagramReader, WireError,
 #[cfg(feature = "local")]
 pub use local::{bind_accept, tcp_dial, udp_egress, LocalEgress};
 
+#[cfg(feature = "local")]
+pub use local::{ProducerDoor, ProducerDoorBuilder, ProducerDoorGuard};
+
 #[cfg(feature = "local")]
 pub use consumer::TokenClient;
 #[cfg(feature = "local")]

src/local/mod.rs

@@ -280,3 +280,7 @@ async fn local_bind_accept(
 
 #[cfg(test)]
 mod tests;
+
+mod producer_door;
+
+pub use producer_door::{ProducerDoor, ProducerDoorBuilder, ProducerDoorGuard};

src/local/producer_door.rs

@@ -0,0 +1,297 @@
+//! 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` — 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 — 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 — 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 — 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 — loopback is 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-sessions cap (ADR-014 knob 2): 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) -> ProducerDoor {
+        ProducerDoor {
+            bind_addr: self.bind_addr,
+            producer: self.producer.expect("the backend serves a producer"),
+            idle_first_byte: self.idle_first_byte.unwrap_or(Duration::from_secs(30)),
+            max_sessions: self.max_sessions.unwrap_or(256),
+        }
+    }
+}
+
+/// 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).
+    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 — 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, std::io::Error> {
+        let ProducerDoor {
+            bind_addr,
+            producer,
+            idle_first_byte,
+            max_sessions,
+        } = self;
+        let bind_addr = bind_addr.expect("the backend binds an address (builder .bind)");
+        let listener = TcpListener::bind(bind_addr).await?;
+        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),
+    }
+}