From afe13b9e47f35d8288762daf43eaea6934bf189b Mon Sep 17 00:00:00 2001 From: cloudhead Date: Tue, 30 Apr 2024 14:40:05 +0200 Subject: [PATCH] node: Don't rate-limit persistent peers --- radicle-node/src/service.rs | 15 ++-- radicle-node/src/service/limiter.rs | 114 ++++++++++++++++++---------- radicle/src/node/config.rs | 4 + 3 files changed, 85 insertions(+), 48 deletions(-) diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 2a1eff3e..e207d905 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -438,6 +438,7 @@ where emitter: Emitter, ) -> Self { let sessions = Sessions::new(rng.clone()); + let limiter = RateLimiter::new(config.peers()); Self { config, @@ -449,7 +450,7 @@ where clock, db, outbox: Outbox::default(), - limiter: RateLimiter::default(), + limiter, sessions, fetching: HashMap::new(), queue: VecDeque::new(), @@ -1155,10 +1156,12 @@ where } let host: HostName = addr.into(); - if self - .limiter - .limit(host.clone(), &self.config.limits.rate.inbound, self.clock) - { + if self.limiter.limit( + host.clone(), + None, + &self.config.limits.rate.inbound, + self.clock, + ) { trace!(target: "service", "Rate limiting inbound connection from {host}.."); return false; } @@ -1633,7 +1636,7 @@ where }; if self .limiter - .limit(peer.addr.clone().into(), limit, self.clock) + .limit(peer.addr.clone().into(), Some(remote), limit, self.clock) { debug!(target: "service", "Rate limiting message from {remote} ({})", peer.addr); return Ok(()); diff --git a/radicle-node/src/service/limiter.rs b/radicle-node/src/service/limiter.rs index 1975da29..d93a7ef3 100644 --- a/radicle-node/src/service/limiter.rs +++ b/radicle-node/src/service/limiter.rs @@ -1,7 +1,7 @@ -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use localtime::LocalTime; -use radicle::node::{address, config, HostName}; +use radicle::node::{address, config, HostName, NodeId}; /// Peer rate limiter. /// @@ -12,15 +12,35 @@ use radicle::node::{address, config, HostName}; #[derive(Debug, Default)] pub struct RateLimiter { buckets: HashMap, + bypass: HashSet, } impl RateLimiter { + /// Create a new rate limiter with a bypass list. Nodes in the bypass list are not limited. + pub fn new(bypass: impl IntoIterator) -> Self { + Self { + buckets: HashMap::default(), + bypass: bypass.into_iter().collect(), + } + } + /// Call this when the address has performed some rate-limited action. /// Returns whether the action is rate-limited or not. /// /// Supplying a different amount of tokens per address is useful if for eg. a peer /// is outbound vs. inbound. - pub fn limit(&mut self, addr: HostName, tokens: &T, now: LocalTime) -> bool { + pub fn limit( + &mut self, + addr: HostName, + nid: Option<&NodeId>, + tokens: &T, + now: LocalTime, + ) -> bool { + if let Some(nid) = nid { + if self.bypass.contains(nid) { + return false; + } + } if let HostName::Ip(ip) = addr { // Don't limit LAN addresses. if !address::is_routable(&ip) { @@ -99,6 +119,8 @@ impl TokenBucket { #[cfg(test)] #[allow(clippy::bool_assert_comparison, clippy::redundant_clone)] mod test { + use radicle::test::arbitrary; + use super::*; impl AsTokens for (usize, f64) { @@ -116,64 +138,72 @@ mod test { let mut r = RateLimiter::default(); let t = (3, 0.2); // Three tokens burst. One token every 5 seconds. let a = HostName::Dns(String::from("seed.radicle.xyz")); + let n = arbitrary::gen::(1); + let n = Some(&n); - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(0)), false); // Burst capacity - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(1)), false); // Burst capacity - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(2)), false); // Burst capacity - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(3)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(4)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(5)), false); // Refilled (1) - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(6)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(7)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(8)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(9)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(10)), false); // Refilled (1) - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(11)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(12)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(13)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(14)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(15)), false); // Refilled (1) - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(16)), true); // Limited - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(60)), false); // Refilled (3) - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(60)), false); // Burst capacity - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(60)), false); // Burst capacity - assert_eq!(r.limit(a.clone(), &t, LocalTime::from_secs(60)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(0)), false); // Burst capacity + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(1)), false); // Burst capacity + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(2)), false); // Burst capacity + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(3)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(4)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(5)), false); // Refilled (1) + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(6)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(7)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(8)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(9)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(10)), false); // Refilled (1) + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(11)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(12)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(13)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(14)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(15)), false); // Refilled (1) + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(16)), true); // Limited + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(60)), false); // Refilled (3) + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(60)), false); // Burst capacity + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(60)), false); // Burst capacity + assert_eq!(r.limit(a.clone(), n, &t, LocalTime::from_secs(60)), true); // Limited } #[test] + #[rustfmt::skip] fn test_limitter_multi() { let t = (1, 1.0); // One token per second. One token burst. + let n = arbitrary::gen::(1); + let n = Some(&n); let mut r = RateLimiter::default(); let addr1 = HostName::Dns(String::from("seed.radicle.xyz")); let addr2 = HostName::Dns(String::from("seed.radicle.net")); - assert_eq!(r.limit(addr1.clone(), &t, LocalTime::from_secs(0)), false); - assert_eq!(r.limit(addr1.clone(), &t, LocalTime::from_secs(0)), true); - assert_eq!(r.limit(addr2.clone(), &t, LocalTime::from_secs(0)), false); - assert_eq!(r.limit(addr2.clone(), &t, LocalTime::from_secs(0)), true); - assert_eq!(r.limit(addr1.clone(), &t, LocalTime::from_secs(1)), false); // Refilled (1) - assert_eq!(r.limit(addr1.clone(), &t, LocalTime::from_secs(1)), true); - assert_eq!(r.limit(addr2.clone(), &t, LocalTime::from_secs(1)), false); - assert_eq!(r.limit(addr2.clone(), &t, LocalTime::from_secs(1)), true); + assert_eq!(r.limit(addr1.clone(), n, &t, LocalTime::from_secs(0)), false); + assert_eq!(r.limit(addr1.clone(), n, &t, LocalTime::from_secs(0)), true); + assert_eq!(r.limit(addr2.clone(), n, &t, LocalTime::from_secs(0)), false); + assert_eq!(r.limit(addr2.clone(), n, &t, LocalTime::from_secs(0)), true); + assert_eq!(r.limit(addr1.clone(), n, &t, LocalTime::from_secs(1)), false); + assert_eq!(r.limit(addr1.clone(), n, &t, LocalTime::from_secs(1)), true); + assert_eq!(r.limit(addr2.clone(), n, &t, LocalTime::from_secs(1)), false); + assert_eq!(r.limit(addr2.clone(), n, &t, LocalTime::from_secs(1)), true); } #[test] + #[rustfmt::skip] fn test_limitter_different_rates() { let t1 = (1, 1.0); // One token per second. One token burst. let t2 = (2, 2.0); // Two tokens per second. Two token burst. + let n = arbitrary::gen::(1); + let n = Some(&n); let mut r = RateLimiter::default(); let addr1 = HostName::Dns(String::from("seed.radicle.xyz")); let addr2 = HostName::Dns(String::from("seed.radicle.net")); - assert_eq!(r.limit(addr1.clone(), &t1, LocalTime::from_secs(0)), false); - assert_eq!(r.limit(addr1.clone(), &t1, LocalTime::from_secs(0)), true); - assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(0)), false); - assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(0)), false); - assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(0)), true); - assert_eq!(r.limit(addr1.clone(), &t1, LocalTime::from_secs(1)), false); // Refilled (1) - assert_eq!(r.limit(addr1.clone(), &t1, LocalTime::from_secs(1)), true); - assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(1)), false); // Refilled (2) - assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(1)), false); - assert_eq!(r.limit(addr2.clone(), &t2, LocalTime::from_secs(1)), true); + assert_eq!(r.limit(addr1.clone(), n, &t1, LocalTime::from_secs(0)), false); + assert_eq!(r.limit(addr1.clone(), n, &t1, LocalTime::from_secs(0)), true); + assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(0)), false); + assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(0)), false); + assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(0)), true); + assert_eq!(r.limit(addr1.clone(), n, &t1, LocalTime::from_secs(1)), false); // Refilled (1) + assert_eq!(r.limit(addr1.clone(), n, &t1, LocalTime::from_secs(1)), true); + assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(1)), false); // Refilled (2) + assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(1)), false); + assert_eq!(r.limit(addr2.clone(), n, &t2, LocalTime::from_secs(1)), true); } } diff --git a/radicle/src/node/config.rs b/radicle/src/node/config.rs index 27b9b2c0..821643cb 100644 --- a/radicle/src/node/config.rs +++ b/radicle/src/node/config.rs @@ -326,6 +326,10 @@ impl Config { .map(|ca| &ca.addr) } + pub fn peers(&self) -> impl Iterator + '_ { + self.connect.iter().cloned().map(|p| p.id) + } + pub fn is_persistent(&self, id: &NodeId) -> bool { self.peer(id).is_some() }