Bound the first frame on the shared proxied receive loop

An accepted onion connection took an inbound slot before a byte was read,
so an unauthenticated remote could hold slots open against the 64-slot
cap and deny inbound onion peering. maint took the same first-frame
deadline as TCP; master and next did not, because the deadline could not
be merged onto the loop the Tor and Nym transports share.

The deadline moves into that shared loop, scoped to the first read only,
using TCP's existing constant rather than a second one. An expiry is not
recorded as a receive error, since folding deadline expiries into that
counter corrupts it.

Nym passes None and is unaffected, which is a stronger statement than
choosing a mixnet-safe value: Nym has no inbound path at all, its
accept_connections returns false and no listener is ever bound, and it
counts no slots, so it has no slot for a silent remote to hold. The
reasoning is recorded at the call site so the deadline does not later
drift onto a transport where Sphinx routing makes a slow first frame
legitimate.

This does not close the divergence from maint. The readiness barrier that
orders the pool insert against the receive task's cleanup was lost from
this path in the same refactor and is still missing; it is filed
separately, and a thirty-second deadline neither opens nor widens that
race.

src/transport/nym/mod.rs

@@ -642,6 +642,15 @@ fn parse_target_addr(addr: &TransportAddr) -> Result<SocksTarget, TransportError
 /// counters, so its teardown hook is a no-op. Emits the terminal
 /// "receive loop stopped" debug (without a `direction` field) that the
 /// shared loop deliberately leaves to each transport.
+///
+/// The first-frame deadline is `None` on every nym connection. Nym is
+/// outbound-only (`accept_connections()` is `false` and no listener is ever
+/// bound), so no nym connection is admitted before a byte is read and none
+/// occupies a capped slot: the pool carries `()` metadata and the teardown
+/// hook decrements nothing. There is no resource for a silent remote to
+/// exhaust, and the mixnet's Sphinx routing makes a first frame legitimately
+/// slow, so a TCP-scale deadline here would drop good connections to defend
+/// a cap that does not exist.
 async fn nym_receive_loop(
     reader: tokio::net::tcp::OwnedReadHalf,
     transport_id: TransportId,
@@ -660,6 +669,7 @@ async fn nym_receive_loop(
         mtu,
         stats,
         "Nym",
+        None,
         |_stats, _meta| {},
     )
     .await;

src/transport/socks5/pool.rs

@@ -8,6 +8,7 @@
 
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 
 use futures::FutureExt;
 use tokio::net::TcpStream;
@@ -141,6 +142,13 @@ pub(crate) trait ProxiedStats: Send + Sync + 'static {
 /// The terminal "receive loop stopped" log is **not** emitted here — it is
 /// hoisted into each per-transport wrapper (tor carries a `direction` field
 /// nym lacks), so this loop is silent on exit.
+///
+/// `first_frame_timeout` bounds the wait for the *first* complete frame only.
+/// It is `Some` for a connection that takes a capped inbound slot from accept
+/// — today only tor's onion listener — and `None` everywhere else, which
+/// covers every outbound connection and the whole of the nym transport (nym
+/// is outbound-only and keeps no counted slots). A deadline expiry is not a
+/// receive error and is deliberately not recorded as one.
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn proxied_receive_loop<S: ProxiedStats, M>(
     mut reader: OwnedReadHalf,
@@ -151,6 +159,7 @@ pub(crate) async fn proxied_receive_loop<S: ProxiedStats, M>(
     mtu: u16,
     stats: Arc<S>,
     label: &'static str,
+    first_frame_timeout: Option<Duration>,
     on_remove: impl Fn(&S, &M),
 ) {
     debug!(
@@ -160,8 +169,34 @@ pub(crate) async fn proxied_receive_loop<S: ProxiedStats, M>(
         label
     );
 
+    let mut first = true;
     loop {
-        match read_fmp_packet(&mut reader, mtu).await {
+        let read = match first_frame_timeout {
+            // Bound the first read only. A silent remote otherwise holds its
+            // inbound slot for as long as it keeps the socket open.
+            Some(d) if first => {
+                match tokio::time::timeout(d, read_fmp_packet(&mut reader, mtu)).await {
+                    Ok(result) => result,
+                    Err(_) => {
+                        // Not a recv error: `record_recv_error` means framing
+                        // or I/O failure, and folding deadline expiries into
+                        // it corrupts that counter.
+                        debug!(
+                            transport_id = %transport_id,
+                            remote_addr = %remote_addr,
+                            timeout_secs = d.as_secs_f64(),
+                            "No complete frame within the first-frame deadline, dropping inbound {} connection",
+                            label
+                        );
+                        break;
+                    }
+                }
+            }
+            _ => read_fmp_packet(&mut reader, mtu).await,
+        };
+        first = false;
+
+        match read {
             Ok(data) => {
                 stats.record_recv(data.len());
 

src/transport/tor/mod.rs

@@ -33,6 +33,7 @@ use crate::transport::socks5::{
     ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, Socks5Auth,
     Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop,
 };
+use crate::transport::tcp::INBOUND_FIRST_FRAME_TIMEOUT;
 use control::{ControlAuth, TorControlClient, TorMonitoringInfo};
 use stats::TorStats;
 
@@ -367,6 +368,7 @@ impl TorTransport {
                 pool,
                 mtu,
                 max_inbound,
+                INBOUND_FIRST_FRAME_TIMEOUT,
                 stats,
             )
             .await;
@@ -769,6 +771,7 @@ impl TorTransport {
                 mtu,
                 recv_stats,
                 Direction::Outbound,
+                None,
             )
             .await;
         });
@@ -934,6 +937,7 @@ impl TorTransport {
                 mtu,
                 recv_stats,
                 Direction::Outbound,
+                None,
             )
             .await;
         });
@@ -1046,6 +1050,10 @@ impl Transport for TorTransport {
 /// actually removing it (so a concurrent close/stop never drives the counter
 /// below zero). `direction` is retained for the terminal "receive loop
 /// stopped" debug field the shared loop deliberately leaves to each transport.
+///
+/// `first_frame_timeout` is `Some` for an inbound connection, which holds a
+/// capped pool slot from the moment it is accepted, and `None` for an
+/// outbound one, which holds no such slot.
 #[allow(clippy::too_many_arguments)]
 async fn tor_receive_loop(
     reader: tokio::net::tcp::OwnedReadHalf,
@@ -1056,6 +1064,7 @@ async fn tor_receive_loop(
     mtu: u16,
     stats: Arc<TorStats>,
     direction: Direction,
+    first_frame_timeout: Option<Duration>,
 ) {
     proxied_receive_loop(
         reader,
@@ -1066,6 +1075,7 @@ async fn tor_receive_loop(
         mtu,
         stats,
         "Tor",
+        first_frame_timeout,
         |stats, meta| match meta {
             Direction::Inbound => stats.record_pool_inbound_removed(),
             Direction::Outbound => stats.record_pool_outbound_removed(),
@@ -1091,6 +1101,13 @@ async fn tor_receive_loop(
 /// connections to a local TCP listener; we accept them, configure
 /// socket options, split the stream, and spawn a per-connection
 /// receive task.
+///
+/// `first_frame_timeout` is the deadline from accept to the first complete
+/// inbound frame, handed to each spawned receive loop. An accepted socket
+/// takes an inbound slot against `max_inbound` before any byte is read, so
+/// without it a remote that connects and stays silent holds that slot for as
+/// long as it keeps the socket open.
+#[allow(clippy::too_many_arguments)]
 async fn tor_accept_loop(
     listener: TcpListener,
     transport_id: TransportId,
@@ -1098,6 +1115,7 @@ async fn tor_accept_loop(
     pool: ProxiedPool<Direction>,
     mtu: u16,
     max_inbound: usize,
+    first_frame_timeout: Duration,
     stats: Arc<TorStats>,
 ) {
     debug!(
@@ -1185,6 +1203,7 @@ async fn tor_accept_loop(
                 mtu,
                 recv_stats,
                 Direction::Inbound,
+                Some(first_frame_timeout),
             )
             .await;
         });
@@ -1893,4 +1912,143 @@ mod tests {
         let err = format!("{}", result.unwrap_err());
         assert!(err.contains("directory"));
     }
+
+    // ========================================================================
+    // Inbound first-frame deadline (onion listener)
+    // ========================================================================
+
+    /// Poll `f` every 10ms until it holds or `limit` elapses.
+    async fn wait_until<F: FnMut() -> bool>(mut f: F, limit: Duration) -> bool {
+        let deadline = Instant::now() + limit;
+        loop {
+            if f() {
+                return true;
+            }
+            if Instant::now() >= deadline {
+                return false;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    }
+
+    /// Drives `tor_accept_loop` directly: the only production path to it is
+    /// `start_directory_mode`, which needs a Tor-managed hostname file and a
+    /// running daemon, so it is not reachable from a unit test.
+    fn spawn_onion_accept_loop(
+        listener: TcpListener,
+        packet_tx: PacketTx,
+        first_frame_timeout: Duration,
+    ) -> (ProxiedPool<Direction>, Arc<TorStats>, JoinHandle<()>) {
+        let pool: ProxiedPool<Direction> = Arc::new(Mutex::new(HashMap::new()));
+        let stats = Arc::new(TorStats::new());
+        let handle = tokio::spawn(tor_accept_loop(
+            listener,
+            TransportId::new(1),
+            packet_tx,
+            pool.clone(),
+            1400,
+            64,
+            first_frame_timeout,
+            stats.clone(),
+        ));
+        (pool, stats, handle)
+    }
+
+    /// Mirror of the TCP case: a silent onion-side socket must lose its
+    /// inbound slot at the deadline. Break-check: with the timeout wrapper
+    /// removed from the shared loop the count stays at 1 and the second
+    /// assertion fails.
+    #[tokio::test]
+    async fn idle_inbound_onion_socket_releases_its_slot() {
+        let (tx, _rx) = packet_channel(32);
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let listen = listener.local_addr().unwrap();
+        let (pool, stats, accept) =
+            spawn_onion_accept_loop(listener, tx, Duration::from_millis(200));
+
+        // Held open for the whole test: any release is the deadline's doing.
+        let squatter = TcpStream::connect(listen).await.unwrap();
+
+        assert!(
+            wait_until(|| stats.pool_inbound_count() == 1, Duration::from_secs(2)).await,
+            "an accepted onion socket should take an inbound slot"
+        );
+        assert!(
+            wait_until(|| stats.pool_inbound_count() == 0, Duration::from_secs(2)).await,
+            "a silent onion socket should lose its slot at the first-frame deadline"
+        );
+        assert!(pool.lock().await.is_empty());
+
+        drop(squatter);
+        accept.abort();
+    }
+
+    /// The deadline covers a *complete* first frame, not merely the first
+    /// byte: a remote that dribbles a prefix inside the deadline and the
+    /// remainder after it must still lose its slot, and the late frame must
+    /// not be delivered.
+    #[tokio::test]
+    async fn byte_dripped_first_onion_frame_past_deadline_is_dropped() {
+        let (tx, mut rx) = packet_channel(32);
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let listen = listener.local_addr().unwrap();
+        let (_pool, stats, accept) =
+            spawn_onion_accept_loop(listener, tx, Duration::from_millis(300));
+
+        let frame = build_msg1_frame();
+        let mut peer = TcpStream::connect(listen).await.unwrap();
+        // Prefix inside the deadline, remainder well past it.
+        peer.write_all(&frame[..4]).await.unwrap();
+        tokio::time::sleep(Duration::from_millis(600)).await;
+        let _ = peer.write_all(&frame[4..]).await;
+
+        assert!(
+            tokio::time::timeout(Duration::from_millis(500), rx.recv())
+                .await
+                .is_err(),
+            "a first onion frame completing after the deadline must not be delivered"
+        );
+        assert!(
+            wait_until(|| stats.pool_inbound_count() == 0, Duration::from_secs(2)).await,
+            "the dripped onion connection should have released its slot"
+        );
+
+        drop(peer);
+        accept.abort();
+    }
+
+    /// The healthy path, and a regression guard as for TCP: the deadline is
+    /// scoped to the first iteration, so an established onion connection that
+    /// then goes quiet keeps its slot. It exists so a future general idle
+    /// deadline cannot start reaping quiet onion links without a test going
+    /// red.
+    #[tokio::test]
+    async fn established_onion_connection_survives_long_idle() {
+        let (tx, mut rx) = packet_channel(32);
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let listen = listener.local_addr().unwrap();
+        let (pool, stats, accept) =
+            spawn_onion_accept_loop(listener, tx, Duration::from_millis(200));
+
+        let mut peer = TcpStream::connect(listen).await.unwrap();
+        peer.write_all(&build_msg1_frame()).await.unwrap();
+        let packet = tokio::time::timeout(Duration::from_secs(2), rx.recv())
+            .await
+            .expect("timeout")
+            .expect("packet channel closed");
+        assert_eq!(packet.data, build_msg1_frame());
+
+        // Four deadlines' worth of silence after the first frame.
+        tokio::time::sleep(Duration::from_millis(800)).await;
+
+        assert_eq!(
+            stats.pool_inbound_count(),
+            1,
+            "an established onion connection must not be dropped by the first-frame deadline"
+        );
+        assert!(!pool.lock().await.is_empty());
+
+        drop(peer);
+        accept.abort();
+    }
 }