Merge pull request #11739 from yahya-nusseibeh/fix/relay-per-address-limits

Relay: a cap on open connections per address, and a per-site join budget

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<Self> {
+        // 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<State>, 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))?;
@@ -428,6 +458,10 @@ fn handle(state: &Arc<State>, 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 ") {
@@ -633,10 +667,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);
@@ -680,7 +717,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);
         }

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<IpAddr, Site>,
 }
 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();

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<TcpStream> = (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));
+    }
+}

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 |