rt: thread the per-IP limiter through the TCP runtimes (F6)

Add Server::max_conns_per_ip (default 0 = unlimited) and wire a
PeerLimiter into the threadpool, tokio, and mio runtimes plus the HTTP
redirect listener. Each runtime admits on accept and holds the PeerGuard
for the connection's whole lifetime (threadpool: around catch_unwind in
the worker; tokio: inside the spawned task; mio: in the Conn struct so it
releases on sweep/done/error). HTTP/3 is untouched (out of scope). With
no cap set, behavior is unchanged.

Addresses audit finding F6 (per-IP connection limits).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: MagicalTux

src/rt/mio.rs

@@ -10,14 +10,15 @@
 use std::collections::HashMap;
 use std::io::{ErrorKind, Read, Write};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use mio::event::Event;
 use mio::net::{TcpListener, TcpStream};
 use mio::{Events, Interest, Poll, Token};
 
 use crate::error::{Error, Result};
-use crate::rt::common::{IO_TIMEOUT, MIN_PROGRESS, READ_BUF};
+use crate::rt::common::{self, IO_TIMEOUT, MIN_PROGRESS, READ_BUF};
 use crate::session::{Session, SessionConfig};
 
 #[cfg(feature = "tls")]
@@ -50,6 +51,9 @@ struct Conn {
     progress_since: Instant,
     /// Bytes received in the current window (reset once it reaches the floor).
     progress_bytes: usize,
+    /// Holds this connection's per-IP slot; releases it when the `Conn` is
+    /// dropped (idle sweep, done, or error).
+    _peer_guard: common::PeerGuard,
 }
 
 impl Conn {
@@ -125,6 +129,7 @@ impl Conn {
 pub(crate) fn run(
     addrs: Vec<SocketAddr>,
     cfg: SessionConfig,
+    limiter: Arc<common::PeerLimiter>,
     #[cfg(feature = "tls")] tls: Option<TlsAcceptor>,
 ) -> Result<()> {
     let mut listener = bind_first(&addrs)?;
@@ -185,6 +190,7 @@ pub(crate) fn run(
                         &listener,
                         &poll,
                         &cfg,
+                        &limiter,
                         #[cfg(feature = "tls")]
                         &tls,
                         &mut conns,
@@ -241,6 +247,7 @@ fn accept_ready(
     listener: &TcpListener,
     poll: &Poll,
     cfg: &SessionConfig,
+    limiter: &Arc<common::PeerLimiter>,
     #[cfg(feature = "tls")] tls: &Option<TlsAcceptor>,
     conns: &mut HashMap<Token, Conn>,
     next_token: &mut usize,
@@ -255,6 +262,14 @@ fn accept_ready(
                     drop(stream);
                     continue;
                 }
+                // Enforce the per-IP cap; dropping `stream` closes the connection.
+                let guard = match limiter.admit(stream.peer_addr().ok().map(|a| a.ip())) {
+                    Some(g) => g,
+                    None => {
+                        drop(stream);
+                        continue;
+                    }
+                };
                 let session = match build_session(
                     cfg,
                     #[cfg(feature = "tls")]
@@ -285,6 +300,7 @@ fn accept_ready(
                         read_closed: false,
                         progress_since: Instant::now(),
                         progress_bytes: 0,
+                        _peer_guard: guard,
                     },
                 );
             }

src/rt/mod.rs

@@ -67,6 +67,8 @@ pub struct Server {
     handler: Arc<dyn Handler>,
     server_name: Option<String>,
     workers: usize,
+    /// Cap on concurrent connections from a single client IP (0 = unlimited).
+    max_conns_per_ip: u32,
     #[cfg(feature = "tls")]
     tls: Option<TlsAcceptor>,
     #[cfg(feature = "compress")]
@@ -101,6 +103,7 @@ impl Server {
             handler: Arc::new(not_found),
             server_name: Some(concat!("httpsd/", env!("CARGO_PKG_VERSION")).to_owned()),
             workers: default_workers(),
+            max_conns_per_ip: 0,
             #[cfg(feature = "tls")]
             tls: None,
             #[cfg(feature = "compress")]
@@ -139,6 +142,15 @@ impl Server {
         self
     }
 
+    /// Cap the number of concurrent connections from a single client IP (0 =
+    /// unlimited, the default). Applies to the TCP runtimes and the HTTP redirect
+    /// listener; a single source that exceeds the cap has further connections
+    /// dropped so it cannot exhaust the global connection budget.
+    pub fn max_conns_per_ip(mut self, max: u32) -> Server {
+        self.max_conns_per_ip = max;
+        self
+    }
+
     /// Set the `Server` response header value (`None` to omit it).
     pub fn server_name(mut self, name: Option<String>) -> Server {
         self.server_name = name;
@@ -258,27 +270,38 @@ impl Server {
         let listener = std::net::TcpListener::bind(self.addrs.as_slice())?;
         let cfg = self.session_config();
         let tls_mode = self.tls_mode();
+        let limiter = common::PeerLimiter::new(self.max_conns_per_ip);
 
         if !self.http_addrs.is_empty() {
             let http = std::net::TcpListener::bind(self.http_addrs.as_slice())?;
             let ctx = self.http_ctx();
-            std::thread::spawn(move || threadpool::run_http_redirect(http, ctx));
+            let redirect_limiter = Arc::clone(&limiter);
+            std::thread::spawn(move || threadpool::run_http_redirect(http, ctx, redirect_limiter));
         }
 
         // Both the main listener and (if any) the redirect listener are now
         // bound; the readiness signal is emitted inside `threadpool::run` right
         // before the accept loop.
-        threadpool::run(listener, cfg, tls_mode, self.workers, self.ready_tx)
+        threadpool::run(
+            listener,
+            cfg,
+            tls_mode,
+            self.workers,
+            limiter,
+            self.ready_tx,
+        )
     }
 
     /// Run on a tokio runtime. Requires being called from within a tokio
     /// runtime context (e.g. under `#[tokio::main]`).
     #[cfg(feature = "rt-tokio")]
     pub async fn run_tokio(self) -> Result<()> {
         let cfg = self.session_config();
+        let limiter = common::PeerLimiter::new(self.max_conns_per_ip);
         tokio::run(
             self.addrs.clone(),
             cfg,
+            limiter,
             #[cfg(feature = "tls")]
             self.tls,
         )
@@ -290,9 +313,11 @@ impl Server {
     #[cfg(feature = "rt-mio")]
     pub fn run_mio(self) -> Result<()> {
         let cfg = self.session_config();
+        let limiter = common::PeerLimiter::new(self.max_conns_per_ip);
         mio::run(
             self.addrs.clone(),
             cfg,
+            limiter,
             #[cfg(feature = "tls")]
             self.tls,
         )

src/rt/threadpool.rs

@@ -26,12 +26,17 @@ struct Shared {
     tls: TlsMode,
 }
 
+/// An accepted connection queued to a worker, paired with the per-IP slot guard
+/// that must live for the connection's whole lifetime.
+type Job = (TcpStream, common::PeerGuard);
+
 /// Run a blocking accept loop, dispatching connections to `workers` threads.
 pub(crate) fn run(
     listener: TcpListener,
     cfg: SessionConfig,
     tls: TlsMode,
     workers: usize,
+    limiter: Arc<common::PeerLimiter>,
     ready: Option<Sender<()>>,
 ) -> Result<()> {
     // On-demand ACME (TLS-ALPN-01) can self-deadlock with a single worker: the
@@ -55,8 +60,7 @@ pub(crate) fn run(
     // the queue is full `send` blocks the accept loop; further connections wait
     // in the kernel backlog and are shed there if it too fills.
     let backlog = workers.saturating_mul(64).clamp(256, 4096);
-    let (tx, rx): (SyncSender<TcpStream>, Receiver<TcpStream>) =
-        std::sync::mpsc::sync_channel(backlog);
+    let (tx, rx): (SyncSender<Job>, Receiver<Job>) = std::sync::mpsc::sync_channel(backlog);
     let rx = Arc::new(Mutex::new(rx));
 
     for _ in 0..workers {
@@ -75,7 +79,14 @@ pub(crate) fn run(
     for incoming in listener.incoming() {
         match incoming {
             Ok(stream) => {
-                if tx.send(stream).is_err() {
+                // Enforce the per-IP cap before queueing. `continue` drops
+                // `stream`, closing the over-limit connection (shed silently,
+                // like the global caps).
+                let guard = match limiter.admit(stream.peer_addr().ok().map(|a| a.ip())) {
+                    Some(g) => g,
+                    None => continue,
+                };
+                if tx.send((stream, guard)).is_err() {
                     break;
                 }
             }
@@ -89,18 +100,22 @@ pub(crate) fn run(
     Ok(())
 }
 
-fn worker_loop(rx: Arc<Mutex<Receiver<TcpStream>>>, shared: Arc<Shared>) {
+fn worker_loop(rx: Arc<Mutex<Receiver<Job>>>, shared: Arc<Shared>) {
     loop {
-        let stream = {
-            let guard = match rx.lock() {
+        let received = {
+            let lock = match rx.lock() {
                 Ok(g) => g,
                 Err(_) => return,
             };
-            guard.recv()
+            lock.recv()
         };
-        let Ok(stream) = stream else {
+        let Ok((stream, guard)) = received else {
             return;
         };
+        // Hold the per-IP slot for the whole connection: keep `guard` alive
+        // across `catch_unwind` and drop it (releasing the slot) only after the
+        // handler returns — even if it panics.
+        let _guard = guard;
         // Isolate each connection: a panic while serving one client must not
         // kill the worker thread, which would permanently shrink the pool and,
         // once every worker died, wedge the whole server.
@@ -167,7 +182,11 @@ fn handle_acme(
 /// Accept loop for the plain-HTTP listener (redirects + ACME HTTP-01). It spawns
 /// a thread per connection — these are short-lived (read a request, reply, close)
 /// — but caps how many run at once so a flood can't spawn unbounded threads.
-pub(crate) fn run_http_redirect(listener: TcpListener, ctx: HttpCtx) {
+pub(crate) fn run_http_redirect(
+    listener: TcpListener,
+    ctx: HttpCtx,
+    limiter: Arc<common::PeerLimiter>,
+) {
     /// Maximum redirect connections served concurrently; excess is shed.
     const MAX_INFLIGHT: usize = 256;
 
@@ -178,6 +197,11 @@ pub(crate) fn run_http_redirect(listener: TcpListener, ctx: HttpCtx) {
             Ok(mut stream) => {
                 common::apply_timeouts(&stream);
                 stream.set_nodelay(true).ok();
+                // Enforce the per-IP cap; `continue` drops the connection.
+                let guard = match limiter.admit(stream.peer_addr().ok().map(|a| a.ip())) {
+                    Some(g) => g,
+                    None => continue,
+                };
                 // Shed load past the cap by closing the connection (drop).
                 if inflight.fetch_add(1, Ordering::Relaxed) >= MAX_INFLIGHT {
                     inflight.fetch_sub(1, Ordering::Relaxed);
@@ -186,6 +210,8 @@ pub(crate) fn run_http_redirect(listener: TcpListener, ctx: HttpCtx) {
                 let ctx = Arc::clone(&ctx);
                 let inflight = Arc::clone(&inflight);
                 thread::spawn(move || {
+                    // Hold the per-IP slot until the redirect finishes.
+                    let _guard = guard;
                     if let Err(e) = redirect::serve(&mut stream, &ctx)
                         && cfg!(debug_assertions)
                     {

src/rt/tokio.rs

@@ -14,7 +14,7 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{TcpListener, TcpStream};
 
 use crate::error::{Error, Result};
-use crate::rt::common::{IO_TIMEOUT, MIN_PROGRESS, READ_BUF};
+use crate::rt::common::{self, IO_TIMEOUT, MIN_PROGRESS, READ_BUF};
 use crate::session::{Session, SessionConfig};
 
 #[cfg(feature = "tls")]
@@ -39,6 +39,7 @@ struct Shared {
 pub(crate) async fn run(
     addrs: Vec<SocketAddr>,
     cfg: SessionConfig,
+    limiter: Arc<common::PeerLimiter>,
     #[cfg(feature = "tls")] tls: Option<TlsAcceptor>,
 ) -> Result<()> {
     let listener = bind_first(&addrs).await?;
@@ -51,15 +52,27 @@ pub(crate) async fn run(
 
     loop {
         match listener.accept().await {
-            Ok((stream, _peer)) => {
+            Ok((stream, peer)) => {
                 // Shed load past the global cap by dropping the connection.
                 if shared.inflight.fetch_add(1, Ordering::Relaxed) >= MAX_INFLIGHT {
                     shared.inflight.fetch_sub(1, Ordering::Relaxed);
                     drop(stream);
                     continue;
                 }
+                // Enforce the per-IP cap. On rejection, release the inflight
+                // slot we just claimed and drop the connection.
+                let guard = match limiter.admit(Some(peer.ip())) {
+                    Some(g) => g,
+                    None => {
+                        shared.inflight.fetch_sub(1, Ordering::Relaxed);
+                        drop(stream);
+                        continue;
+                    }
+                };
                 let shared = Arc::clone(&shared);
                 tokio::spawn(async move {
+                    // Hold the per-IP slot for the whole connection.
+                    let _guard = guard;
                     let outcome = serve(stream, &shared).await;
                     shared.inflight.fetch_sub(1, Ordering::Relaxed);
                     if cfg!(debug_assertions)