From 41826c8e1ae751a933b7c30819fd14a7f48554f3 Mon Sep 17 00:00:00 2001 From: reverse Date: Wed, 12 Aug 2026 17:05:24 +0000 Subject: [PATCH] =?UTF-8?q?socketengine:=20per-IP=20accept-rate=20limiter?= =?UTF-8?q?=20(token=20bucket,=20accept=5Frate/accept=5Fburst,=20off=20by?= =?UTF-8?q?=20default)=20=E2=80=94=20drops=20connection-churn=20floods=20a?= =?UTF-8?q?t=20the=20accept=20edge=20before=20any=20state=20is=20allocated?= =?UTF-8?q?;=20exempts=20trusted=20proxies=20and=20server=20links?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- echoircd.conf.example | 6 +++ src/main.rs | 12 +++++- src/socketengine.rs | 89 ++++++++++++++++++++++++++++++++++++++++--- tests/integration.rs | 36 +++++++++++++++++ 4 files changed, 136 insertions(+), 7 deletions(-) diff --git a/echoircd.conf.example b/echoircd.conf.example index 59a3007..b680fc7 100644 --- a/echoircd.conf.example +++ b/echoircd.conf.example @@ -145,6 +145,12 @@ amu_target = both # round-robins connections across the pool; the core stays single-threaded and lock-free. # io_threads = 0 # reactor workers; 0 = auto (one per core, capped at 4) # tls_handshake_timeout = 15 # drop a TLS conn that stalls mid-handshake (secs; 0 = off) +# Per-IP accept-rate limit: drop connection-churn floods at the accept edge, before any +# per-connection state is allocated (complements the connclass concurrent clone caps). +# Off by default; a generous value never affects real clients but stops a flooder +# opening/closing connections in a loop. Trusted proxies and server links are exempt. +# accept_rate = 0 # max NEW connections/sec per source IP (0 = off) +# accept_burst = 0 # instantaneous burst allowed per IP (0 = same as accept_rate) # --- ident (RFC1413): off by default; a connection class can also enable it --- # useident = yes # look up every client's ident (adds connect latency) diff --git a/src/main.rs b/src/main.rs index ac54d75..7286bba 100644 --- a/src/main.rs +++ b/src/main.rs @@ -64,6 +64,9 @@ fn main() { // against connections that open but never negotiate let hs = raw_num("tls_handshake_timeout", 15); let handshake_timeout = (hs > 0).then(|| Duration::from_secs(hs as u64)); + // per-IP accept-rate limit (0 = off): drop connection-churn floods at the edge + let accept_limiter = + socketengine::AcceptLimiter::from_conf(raw_num("accept_rate", 0), raw_num("accept_burst", 0)); // trusted PROXY-protocol source globs (reactor rewrites the client IP from them) let proxy_trust: Vec = cfg.raw.get("proxy").cloned().unwrap_or_default(); @@ -123,6 +126,7 @@ fn main() { let tls_counter = counter.clone(); let tls_proxy_trust = proxy_trust.clone(); let tls_reactors = reactors.clone(); + let tls_limiter = accept_limiter.clone(); thread::spawn(move || { socketengine::accept_loop( tls_listener, @@ -133,6 +137,7 @@ fn main() { max_line, tls_proxy_trust, tls_reactors, + tls_limiter, ) }); } @@ -150,7 +155,7 @@ fn main() { let s_tx = tx.clone(); let s_counter = counter.clone(); thread::spawn(move || { - // links stay on the thread path: no reactor handoff + // links stay on the thread path: no reactor handoff, no rate limit socketengine::accept_loop( sl, s_tx, @@ -160,6 +165,7 @@ fn main() { max_line, Vec::new(), Vec::new(), + None, ) }); } @@ -185,6 +191,8 @@ fn main() { } // client plaintext connections: the acceptor round-robins them across the pool - thread::spawn(move || socketengine::run_acceptor(client_listener, reactors, counter, proxy_trust)); + thread::spawn(move || { + socketengine::run_acceptor(client_listener, reactors, counter, proxy_trust, accept_limiter) + }); let _ = core.join(); } diff --git a/src/socketengine.rs b/src/socketengine.rs index 71f1d91..b42cfff 100644 --- a/src/socketengine.rs +++ b/src/socketengine.rs @@ -17,10 +17,10 @@ use std::collections::{HashMap, HashSet}; use std::io::{self, BufRead, BufReader, Read, Write}; -use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream}; +use std::net::{IpAddr, Shutdown, SocketAddr, TcpListener, TcpStream}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::mpsc::{self, Receiver, Sender, TryRecvError}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::thread; use std::time::{Duration, Instant}; @@ -283,6 +283,69 @@ fn dispatch(reactors: &[ReactorHandle], rr: &mut usize, a: Accepted) { } } +/// A token-bucket rate limiter keyed by source IP, checked at the accept edge so a +/// connection-churn flood is dropped before any per-connection state is allocated — +/// the cheapest possible rejection. Shared (behind a mutex) by the plaintext and TLS +/// acceptors, so one IP can't earn a fresh budget per listener. Off unless +/// `accept_rate` is configured. Complements the connclass *concurrent* clone caps with +/// a *rate* cap, and skips connections from trusted proxies (whose peer IP is the proxy). +pub struct AcceptLimiter { + rate: f64, // sustained new connections/sec per IP + burst: f64, // bucket capacity — the instantaneous burst allowed per IP + inner: Mutex, +} + +struct LimiterState { + buckets: HashMap, // ip -> (tokens, last refill) + last_prune: Instant, +} + +impl AcceptLimiter { + /// Build a limiter from config, or `None` when disabled (`rate` 0). `burst` 0 + /// defaults to `rate` (one second's worth). + pub fn from_conf(rate: usize, burst: usize) -> Option> { + if rate == 0 { + return None; + } + let burst = if burst == 0 { rate } else { burst }; + Some(Arc::new(AcceptLimiter { + rate: rate as f64, + burst: burst.max(1) as f64, + inner: Mutex::new(LimiterState { + buckets: HashMap::new(), + last_prune: Instant::now(), + }), + })) + } + + /// Whether a new connection from `ip` is allowed now, consuming one token. + fn allow(&self, ip: IpAddr) -> bool { + let now = Instant::now(); + let mut st = self.inner.lock().unwrap_or_else(|e| e.into_inner()); + // periodically forget IPs idle for a while, so memory tracks only active sources + if now.duration_since(st.last_prune) >= Duration::from_secs(30) { + st.buckets + .retain(|_, &mut (_, last)| now.duration_since(last) < Duration::from_secs(60)); + st.last_prune = now; + } + let entry = st.buckets.entry(ip).or_insert((self.burst, now)); + let refilled = + (entry.0 + self.rate * now.duration_since(entry.1).as_secs_f64()).min(self.burst); + if refilled >= 1.0 { + *entry = (refilled - 1.0, now); + true + } else { + *entry = (refilled, now); + false + } + } +} + +/// True when a limiter is configured and this IP is over its accept rate. +fn rate_limited(limiter: &Option>, ip: IpAddr) -> bool { + limiter.as_ref().map(|l| !l.allow(ip)).unwrap_or(false) +} + /// The plaintext client acceptor: owns the listener and round-robins each new /// connection onto a reactor worker. pub fn run_acceptor( @@ -290,6 +353,7 @@ pub fn run_acceptor( reactors: Vec, counter: Arc, proxy_trust: Vec, + limiter: Option>, ) { if reactors.is_empty() { eprintln!("acceptor: no worker threads; plaintext clients disabled"); @@ -321,18 +385,23 @@ pub fn run_acceptor( loop { match listener.accept() { Ok((stream, _addr)) => { - let _ = stream.set_nodelay(true); - let uid = counter.fetch_add(1, Ordering::Relaxed); let addr = stream .peer_addr() .unwrap_or_else(|_| "0.0.0.0:0".parse().unwrap()); - let local_port = stream.local_addr().map(|a| a.port()).unwrap_or(0); // a connection from a trusted proxy leads with a PROXY header; // the worker holds its Connect until that header is consumed so // the core sees the real client IP. let via_proxy = proxy_trust.iter().any(|g| { crate::modules::connclass::ip_matches(g, &addr.ip().to_string()) }); + // rate-limit direct clients at the edge; drop before allocating + // anything. Proxied clients carry the proxy's IP, so skip them. + if !via_proxy && rate_limited(&limiter, addr.ip()) { + continue; // stream drops here, nothing else touched + } + let _ = stream.set_nodelay(true); + let uid = counter.fetch_add(1, Ordering::Relaxed); + let local_port = stream.local_addr().map(|a| a.port()).unwrap_or(0); dispatch( &reactors, &mut rr, @@ -855,6 +924,7 @@ pub fn accept_loop( max_line: usize, proxy_trust: Vec, reactors: Vec, + limiter: Option>, ) { let mut rr: usize = 0; for conn in listener.incoming() { @@ -862,6 +932,15 @@ pub fn accept_loop( let Ok(addr) = stream.peer_addr() else { continue; }; + // rate-limit direct client connections at the edge (not S2S links, not proxied) + if !link { + let via_proxy = proxy_trust + .iter() + .any(|g| crate::modules::connclass::ip_matches(g, &addr.ip().to_string())); + if !via_proxy && rate_limited(&limiter, addr.ip()) { + continue; // drop before any per-connection work + } + } let _ = stream.set_nodelay(true); let local_port = stream.local_addr().map(|a| a.port()).unwrap_or(0); let uid = counter.fetch_add(1, Ordering::Relaxed); diff --git a/tests/integration.rs b/tests/integration.rs index b3f5455..c4cf893 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -76,6 +76,10 @@ impl Drop for Server { impl Server { fn start(io_threads: usize, tls: bool, hs_timeout: u32) -> Server { + Server::start_full(io_threads, tls, hs_timeout, 0) + } + + fn start_full(io_threads: usize, tls: bool, hs_timeout: u32, accept_rate: usize) -> Server { let (plain, tlsp, s2s) = (free_port(), free_port(), free_port()); let dir = std::env::temp_dir().join(format!("echoircd-it-{}-{plain}", std::process::id())); std::fs::create_dir_all(&dir).unwrap(); @@ -83,6 +87,9 @@ impl Server { "servername = it.test\nnetwork = itNet\nbind = 127.0.0.1:{plain}\n\ bind_server = 127.0.0.1:{s2s}\nsid = 1AA\nmotd = hi\nio_threads = {io_threads}\n" ); + if accept_rate > 0 { + conf.push_str(&format!("accept_rate = {accept_rate}\naccept_burst = {accept_rate}\n")); + } if tls { let (cert, key) = gen_cert(); let (cp, kp) = (dir.join("cert.pem"), dir.join("key.pem")); @@ -265,6 +272,35 @@ fn tls_in_reactor_handshake_and_cross_transport() { ); } +#[test] +fn accept_rate_limit_drops_connection_churn() { + // rate/burst = 5: a rapid burst of 20 connections from one IP must be partly dropped + // at the accept edge — some register, but not all 20. + let srv = Server::start_full(2, false, 5, 5); + let mut socks = Vec::new(); + for i in 0..20 { + if let Ok(mut s) = TcpStream::connect(("127.0.0.1", srv.plain)) { + s.set_read_timeout(Some(Duration::from_millis(600))).unwrap(); + let _ = s.write_all(format!("NICK n{i}\r\nUSER n{i} 0 * :n\r\n").as_bytes()); + socks.push(s); + } + } + let mut registered = 0; + for s in socks.iter_mut() { + if read_until(s, " 001 ", Duration::from_millis(800)) { + registered += 1; + } + } + assert!( + registered < 20, + "rate limit didn't drop any of a 20-connection burst ({registered} registered)" + ); + assert!( + registered >= 3, + "rate limit dropped too much — burst of 5 should let at least a few through ({registered})" + ); +} + #[test] fn tls_stalled_handshake_is_reaped() { // 2s handshake timeout: a raw TCP connection to the TLS port that never negotiates