From 07ed5dac67edf18e227d0051a7d8368cbd1069a5 Mon Sep 17 00:00:00 2001 From: yahya-nusseibeh <254622510+yahya-nusseibeh@users.noreply.github.com> Date: Sat, 7 Nov 2026 07:51:55 +0200 Subject: [PATCH] Relay: a cap on open connections per address, and a per-site join budget Two ways one client could lock every user out of the Play with Friends relay, with no account: (a) The only connection cap was global (4,096), counted before anything is read, and each connection may stay silent for the 10 s first-line wait: about 410 idle connections a second from one address kept the relay full, so hosts couldn't register and guests couldn't join. (b) Every join, known code or not, is refused while the global unknown-code budget (120 a minute) is empty. Each /64 may try 5 unknown codes per 10 minutes, so ~240 /64s keep it empty, and one residential /56 has 256 of them. - Limiter::open_connection / close_connection: connections open at once per address key (IPv6 /64; a /48 four times as many), 64 by default (RELAY_MAX_OPEN_PER_ADDRESS; a host keeps one connection per guest, hence the margin). Counted before anything is read, or right after the PROXY line behind the PROXY protocol; past the cap the connection is closed at once, not held for the first-line wait. An RAII OpenSlot gives the place back however the connection ends. - Each site (IPv4 /24, IPv6 /48) has its own unknown-code budget, 24 a minute with a burst of 6, checked before the global one and spent on every unknown code, so emptying the global budget takes many sites, not one home. - docs/play-with-friends.md: the new setting and the site budget. - Tests: the per-address cap (IPv4, IPv6 /64 and /48, off without per-address limits), the per-site budget (shared within a /24 and a /48, refilling), and a running relay closing a fifth silent connection at once with a cap of 4, then letting new ones in once the held ones go. Found in yahya-nusseibeh's review of Horizon (H2, 7 Nov). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SJ3stK8iXVDXqoi7midcqd --- crates/horizon-relay/src/lib.rs | 51 +++++- crates/horizon-relay/src/limits.rs | 169 ++++++++++++++++++ .../horizon-relay/tests/open_per_address.rs | 59 ++++++ docs/play-with-friends.md | 3 +- 4 files changed, 276 insertions(+), 6 deletions(-) create mode 100644 crates/horizon-relay/tests/open_per_address.rs diff --git a/crates/horizon-relay/src/lib.rs b/crates/horizon-relay/src/lib.rs index 025351c06c..8192bd947b 100644 --- a/crates/horizon-relay/src/lib.rs +++ b/crates/horizon-relay/src/lib.rs @@ -93,6 +93,9 @@ impl Config { if let Some(v) = var("RELAY_MAX_HOSTS_PER_ADDRESS") { c.rates.hosts_per_address = v; } + if let Some(v) = var("RELAY_MAX_OPEN_PER_ADDRESS") { + c.rates.open_per_address = v; + } if let Some(v) = var("RELAY_MAX_BRIDGES") { c.max_bridges = v; } @@ -404,11 +407,38 @@ fn unmap(peer: SocketAddr) -> SocketAddr { } } +/// A connection counted against its address's cap ([`limits::Limiter::open_connection`]), given +/// back however the connection ends. +struct OpenSlot<'a>(&'a State, std::net::IpAddr); +impl<'a> OpenSlot<'a> { + fn take(state: &'a State, ip: std::net::IpAddr) -> Option { + // Decided with the lock released before a slot exists: a refused one must never be built, + // its Drop would give back a place it never took (and lock the limiter again). + let open = locked(&state.limiter).open_connection(ip); + open.then(|| Self(state, ip)) + } +} +impl Drop for OpenSlot<'_> { + fn drop(&mut self) { + locked(&self.0.limiter).close_connection(self.1); + } +} + fn handle(state: &Arc, mut stream: TcpStream, peer: SocketAddr) -> io::Result<()> { if state.stop.load(Ordering::Relaxed) { return Ok(()); } let mut peer = unmap(peer); + // Counted per address before anything is read: past its cap a connection is closed at once, + // not held for `first_line` with one of the relay's global slots. Behind the PROXY protocol the + // client's address is the line the load balancer sends on connect, so it's counted after it. + let mut _open = None; + if !state.config.proxy_protocol { + let Some(slot) = OpenSlot::take(state, peer.ip()) else { + return Ok(()); + }; + _open = Some(slot); + } // The first line (and an HTTP request) within `first_line` in all. let first_line = horizon_friends::deadline(&stream, state.config.first_line)?; stream.set_read_timeout(Some(state.config.first_line))?; @@ -427,6 +457,10 @@ fn handle(state: &Arc, mut stream: TcpStream, peer: SocketAddr) -> io::Re peer = proxy_line(&line) .map(unmap) .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "no PROXY line"))?; + let Some(slot) = OpenSlot::take(state, peer.ip()) else { + return Ok(()); + }; + _open = Some(slot); line = wire::read_line(&mut stream)?.ok_or(io::ErrorKind::UnexpectedEof)?; } if line.starts_with(b"GET ") { @@ -632,10 +666,13 @@ fn join( ) -> io::Result<()> { let now = Instant::now(); state.joins.fetch_add(1, Ordering::Relaxed); - // The global budget for unknown codes is checked before the lookup, so a - // refusal says nothing about whether the nameplate exists. - let allowed = - locked(&state.limiter).allow_join(peer.ip(), now) && locked(&state.global_unknown).has(now); + // The budgets for unknown codes (the site's, then the global one) are checked before the + // lookup, so a refusal says nothing about whether the nameplate exists. The site's comes first: + // emptying the global budget takes many sites, not the /64s of one home's /56. + let allowed = { + let mut limiter = locked(&state.limiter); + limiter.allow_join(peer.ip(), now) && limiter.site_unknown_has(peer.ip(), now) + } && locked(&state.global_unknown).has(now); if !allowed { state.rate_limited.fetch_add(1, Ordering::Relaxed); return refuse(&mut stream, ErrorCode::RateLimited); @@ -679,7 +716,11 @@ fn join( Lookup::Found(tx, slot) => (tx, slot), Lookup::Unknown => { state.unknown.fetch_add(1, Ordering::Relaxed); - locked(&state.limiter).unknown(peer.ip(), now); + { + let mut limiter = locked(&state.limiter); + limiter.unknown(peer.ip(), now); + limiter.site_unknown_take(peer.ip(), now); + } locked(&state.global_unknown).try_take(now); return refuse(&mut stream, ErrorCode::UnknownCode); } diff --git a/crates/horizon-relay/src/limits.rs b/crates/horizon-relay/src/limits.rs index 5c3058315b..86832f125d 100644 --- a/crates/horizon-relay/src/limits.rs +++ b/crates/horizon-relay/src/limits.rs @@ -84,6 +84,15 @@ pub struct Rates { /// Hosts one address may hold open at once (`RELAY_MAX_HOSTS_PER_ADDRESS`); /// an IPv6 /48 (one site, 65536 homes) may hold [`SITE_HOSTS`] times as many. pub hosts_per_address: usize, + /// Connections of any kind one address may hold open at once + /// (`RELAY_MAX_OPEN_PER_ADDRESS`), an IPv6 /48 [`SITE_HOSTS`] times as many. Without it one + /// client could fill the relay's global cap with silent connections, each held for the + /// first-line wait. A host keeps one connection per guest it plays with, hence the margin. + pub open_per_address: usize, + /// Unknown codes a minute (and the burst) from one site, an IPv4 /24 or an IPv6 /48, checked + /// before the global budget, so emptying that takes many sites rather than one home's /64s. + pub site_unknown_per_min: f64, + pub site_unknown_burst: f64, } impl Default for Rates { fn default() -> Self { @@ -96,6 +105,9 @@ impl Default for Rates { unknown_window: Duration::from_secs(600), block: Duration::from_secs(600), hosts_per_address: 4, + open_per_address: 64, + site_unknown_per_min: 24.0, + site_unknown_burst: 6.0, } } } @@ -117,6 +129,25 @@ fn host_keys(ip: IpAddr, per_address: usize) -> Vec<((IpAddr, u8), usize)> { keys } +/// The site an address belongs to for the unknown-code budget: its IPv4 /24, or its IPv6 /48. +pub fn site_key(ip: IpAddr) -> IpAddr { + match address_key(ip) { + IpAddr::V4(v4) => { + let o = v4.octets(); + IpAddr::V4(std::net::Ipv4Addr::new(o[0], o[1], o[2], 0)) + } + IpAddr::V6(v6) => { + let s = v6.segments(); + IpAddr::V6(Ipv6Addr::new(s[0], s[1], s[2], 0, 0, 0, 0, 0)) + } + } +} + +struct Site { + unknown: Bucket, + seen: Instant, +} + struct Address { register: Bucket, join: Bucket, @@ -135,6 +166,10 @@ pub struct Limiter { /// Open hosts by address key (and IPv6 site); never swept, each count /// is given back by [`Limiter::close_host`]. hosts: HashMap<(IpAddr, u8), usize>, + /// Open connections, counted the same way and given back by [`Limiter::close_connection`]. + open: HashMap<(IpAddr, u8), usize>, + /// Unknown-code budgets by [`site_key`], swept with the addresses. + sites: HashMap, } impl Limiter { pub fn new(rates: Rates, enabled: bool) -> Self { @@ -144,6 +179,60 @@ impl Limiter { addresses: HashMap::new(), last_sweep: Instant::now(), hosts: HashMap::new(), + open: HashMap::new(), + sites: HashMap::new(), + } + } + /// Count one more connection open from `ip`, unless its address (or IPv6 site) holds as many + /// as it may already: then the caller closes it at once. + pub fn open_connection(&mut self, ip: IpAddr) -> bool { + if !self.enabled { + return true; + } + let keys = host_keys(ip, self.rates.open_per_address); + if keys + .iter() + .any(|(k, cap)| self.open.get(k).copied().unwrap_or(0) >= *cap) + { + return false; + } + for (k, _) in keys { + *self.open.entry(k).or_insert(0) += 1; + } + true + } + /// A connection [`Limiter::open_connection`] let in has closed. + pub fn close_connection(&mut self, ip: IpAddr) { + if !self.enabled { + return; + } + for (k, _) in host_keys(ip, self.rates.open_per_address) { + if let Some(n) = self.open.get_mut(&k) { + *n -= 1; + if *n == 0 { + self.open.remove(&k); + } + } + } + } + fn site(&mut self, ip: IpAddr, now: Instant) -> &mut Site { + let r = &self.rates; + let site = self.sites.entry(site_key(ip)).or_insert_with(|| Site { + unknown: Bucket::new(r.site_unknown_per_min / 60.0, r.site_unknown_burst, now), + seen: now, + }); + site.seen = now; + site + } + /// Whether `ip`'s site has unknown-code budget left (checked before every join, as the global + /// budget is, so a refusal says nothing about whether a code exists). + pub fn site_unknown_has(&mut self, ip: IpAddr, now: Instant) -> bool { + !self.enabled || self.site(ip, now).unknown.has(now) + } + /// An unknown code from `ip`: spends one from its site's budget. + pub fn site_unknown_take(&mut self, ip: IpAddr, now: Instant) { + if self.enabled { + self.site(ip, now).unknown.try_take(now); } } /// Count one more host open from `ip`, unless its address (or IPv6 site) @@ -186,6 +275,9 @@ impl Limiter { now.saturating_duration_since(a.seen) < keep || a.blocked_until.is_some_and(|b| b > now) }); + // An idle site's bucket has refilled long before: dropping it changes nothing. + self.sites + .retain(|_, s| now.saturating_duration_since(s.seen) < keep); } let r = &self.rates; let a = self @@ -248,6 +340,83 @@ impl Limiter { mod tests { use super::*; + #[test] + fn an_address_holds_only_so_many_connections_open() { + let rates = Rates { + open_per_address: 3, + ..Rates::default() + }; + let mut l = Limiter::new(rates, true); + let home: IpAddr = "203.0.113.7".parse().unwrap(); + let other: IpAddr = "203.0.113.8".parse().unwrap(); + assert!((0..3).all(|_| l.open_connection(home))); + assert!(!l.open_connection(home), "the fourth is refused"); + assert!(l.open_connection(other), "another address isn't"); + l.close_connection(home); + assert!(l.open_connection(home), "a closed one gives its place back"); + // IPv6 counts by /64, and a /48 holds four times as many. + let mut l = Limiter::new( + Rates { + open_per_address: 2, + ..Rates::default() + }, + true, + ); + let in_64 = |i: u16| -> IpAddr { format!("2001:db8:1:{i}::{}", i + 1).parse().unwrap() }; + assert!(l.open_connection(in_64(0)) && l.open_connection(in_64(0))); + assert!( + !l.open_connection("2001:db8:1:0::99".parse().unwrap()), + "same /64" + ); + for i in 1..4 { + assert!(l.open_connection(in_64(i)) && l.open_connection(in_64(i))); + } + assert!(!l.open_connection(in_64(4)), "the /48 is full at 8"); + // Off when the relay can't see client addresses. + let mut off = Limiter::new(Rates::default(), false); + assert!((0..1000).all(|_| off.open_connection(home))); + } + + #[test] + fn a_site_spends_its_own_unknown_code_budget() { + let t = Instant::now(); + let mut l = Limiter::new( + Rates { + site_unknown_per_min: 6.0, + site_unknown_burst: 2.0, + ..Rates::default() + }, + true, + ); + // Two homes of one /24 share a budget: the burst, then nothing until it refills. + let (a, b): (IpAddr, IpAddr) = ( + "198.51.100.1".parse().unwrap(), + "198.51.100.200".parse().unwrap(), + ); + l.site_unknown_take(a, t); + assert!(l.site_unknown_has(b, t)); + l.site_unknown_take(b, t); + assert!(!l.site_unknown_has(a, t) && !l.site_unknown_has(b, t)); + assert!( + l.site_unknown_has("198.51.101.1".parse().unwrap(), t), + "another /24 isn't" + ); + assert!( + l.site_unknown_has(a, t + Duration::from_secs(10)), + "one a ten seconds back" + ); + // IPv6: the /56 of one home is one /48. + let mut v6 = Limiter::new( + Rates { + site_unknown_burst: 1.0, + ..Rates::default() + }, + true, + ); + v6.site_unknown_take("2001:db8:7:100::1".parse().unwrap(), t); + assert!(!v6.site_unknown_has("2001:db8:7:1ff::1".parse().unwrap(), t)); + } + #[test] fn buckets_refill_at_their_rate() { let t = Instant::now(); diff --git a/crates/horizon-relay/tests/open_per_address.rs b/crates/horizon-relay/tests/open_per_address.rs new file mode 100644 index 0000000000..80a6bd1c6c --- /dev/null +++ b/crates/horizon-relay/tests/open_per_address.rs @@ -0,0 +1,59 @@ +//! One client address can't fill the relay with silent connections: past its cap each new one is +//! closed at once, not held for the first-line wait with one of the relay's global slots. +use horizon_relay::{Config, Relay, limits::Rates}; +use std::{ + io::Read, + net::TcpStream, + time::{Duration, Instant}, +}; + +#[test] +fn silent_connections_past_an_address_cap_close_at_once() { + let (_relay, address, _) = Relay::spawn_local(Config { + rates: Rates { + open_per_address: 4, + ..Rates::default() + }, + first_line: Duration::from_secs(10), + ..Config::default() + }) + .unwrap(); + // Four silent connections: held open (the relay waits for their first line). + let held: Vec = (0..4) + .map(|_| TcpStream::connect(address).unwrap()) + .collect(); + std::thread::sleep(Duration::from_millis(300)); + for s in &held { + s.set_read_timeout(Some(Duration::from_millis(200))) + .unwrap(); + let mut b = [0u8; 1]; + assert!( + matches!((&*s).read(&mut b), Err(e) if e.kind() == std::io::ErrorKind::WouldBlock + || e.kind() == std::io::ErrorKind::TimedOut), + "within the cap, still open" + ); + } + // The fifth is closed at once, long before the 10 s wait. + let mut over = TcpStream::connect(address).unwrap(); + over.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + let started = Instant::now(); + assert_eq!(over.read(&mut [0u8; 1]).unwrap_or(0), 0); + assert!( + started.elapsed() < Duration::from_secs(3), + "{:?}", + started.elapsed() + ); + // One leaves: its place comes back once the relay sees it go. + drop(held); + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let mut s = TcpStream::connect(address).unwrap(); + s.set_read_timeout(Some(Duration::from_millis(500))) + .unwrap(); + match s.read(&mut [0u8; 1]) { + Err(_) => break, // still open after half a second: let in + Ok(_) => assert!(Instant::now() < deadline, "the places never came back"), + } + std::thread::sleep(Duration::from_millis(100)); + } +} diff --git a/docs/play-with-friends.md b/docs/play-with-friends.md index eb2b34b903..9854c4c898 100644 --- a/docs/play-with-friends.md +++ b/docs/play-with-friends.md @@ -162,12 +162,13 @@ Settings, all optional: | `RELAY_BIND` | all addresses | listen on one address only (e.g. a tailnet address) | | `RELAY_MAX_HOSTS` | 1000 | short codes at once | | `RELAY_MAX_HOSTS_PER_ADDRESS` | 4 | short codes one client address holds at once (IPv6 by /64; a /48 holds 4 times as many); with `RELAY_PER_ADDRESS_LIMITS` | +| `RELAY_MAX_OPEN_PER_ADDRESS` | 64 | connections of any kind one client address holds open at once (IPv6 by /64; a /48 holds 4 times as many); past it a connection is closed at once; with `RELAY_PER_ADDRESS_LIMITS` | | `RELAY_MAX_BRIDGES` | 200 | relayed streams at once | | `RELAY_MAX_PENDING_PER_HOST` | 8 | guests joining one host at the same time | | `RELAY_MAX_CONNECTIONS` | 4096 | connections open at once, of every kind (more are closed unanswered) | | `RELAY_SESSION_KBPS` | 1024 | KiB/s each way per relayed stream | | `RELAY_TOTAL_MBPS` | 100 | Mbit/s for everything relayed | -| `RELAY_GLOBAL_UNKNOWN_PER_MIN` | 120 | unknown short codes a minute, all clients together | +| `RELAY_GLOBAL_UNKNOWN_PER_MIN` | 120 | unknown short codes a minute, all clients together; each IPv4 /24 or IPv6 /48 also has its own budget, 24 a minute (burst 6), checked first | | `RELAY_PER_ADDRESS_LIMITS` | 1 | per-address rate limits (see below) | | `RELAY_PROXY_PROTOCOL` | 0 | read the client address from a PROXY protocol v1 line | | `RELAY_PROXY_FROM` | none | with `RELAY_PROXY_PROTOCOL=1`, required: the load balancers' addresses or CIDR ranges (`10.0.0.0/8,fd00::/8`). Connections from anywhere else are closed unanswered; the relay refuses to start with the protocol on and this empty |