use std::{
collections::HashMap,
time::{SystemTime, UNIX_EPOCH},
};
use chia_sdk_client::{RateLimit, RateLimits, V2_RATE_LIMITS};
use chia_traits::Streamable;
use crate::DigMessage;
#[derive(Debug, Clone)]
pub struct OpcodeRateLimits {
default_settings: RateLimit,
non_tx_frequency: f64,
non_tx_max_total_size: f64,
tx: HashMap<u8, RateLimit>,
other: HashMap<u8, RateLimit>,
}
impl From<&RateLimits> for OpcodeRateLimits {
fn from(limits: &RateLimits) -> Self {
let rekey = |map: &HashMap<chia_protocol::ProtocolMessageTypes, RateLimit>| {
map.iter()
.filter_map(|(msg_type, limit)| Some((*msg_type.to_bytes().ok()?.first()?, *limit)))
.collect()
};
Self {
default_settings: limits.default_settings,
non_tx_frequency: limits.non_tx_frequency,
non_tx_max_total_size: limits.non_tx_max_total_size,
tx: rekey(&limits.tx),
other: rekey(&limits.other),
}
}
}
impl Default for OpcodeRateLimits {
fn default() -> Self {
Self::from(&*V2_RATE_LIMITS)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Admission {
Admitted,
Deferred,
Unsendable,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Direction {
Inbound,
Outbound,
}
#[derive(Debug, Clone)]
pub struct OpcodeRateLimiter {
direction: Direction,
reset_seconds: u64,
period: u64,
limit_factor: f64,
counts: HashMap<u8, f64>,
cumulative_sizes: HashMap<u8, f64>,
non_tx_count: f64,
non_tx_size: f64,
limits: OpcodeRateLimits,
}
impl OpcodeRateLimiter {
#[must_use]
pub fn new(
direction: Direction,
reset_seconds: u64,
limit_factor: f64,
limits: OpcodeRateLimits,
) -> Self {
Self {
direction,
reset_seconds,
period: now_seconds() / reset_seconds,
limit_factor,
counts: HashMap::new(),
cumulative_sizes: HashMap::new(),
non_tx_count: 0.0,
non_tx_size: 0.0,
limits,
}
}
pub fn allow(&mut self, message: &DigMessage) -> bool {
self.admit(message) == Admission::Admitted
}
pub fn admit(&mut self, message: &DigMessage) -> Admission {
self.roll_window();
let size = f64::from(u32::try_from(message.data.len()).unwrap_or(u32::MAX));
let opcode = message.msg_type;
let mut limit = self.limits.default_settings;
let mut counts_against_non_tx = false;
if let Some(tx_limit) = self.limits.tx.get(&opcode) {
limit = *tx_limit;
} else if let Some(other_limit) = self.limits.other.get(&opcode) {
limit = *other_limit;
counts_against_non_tx = true;
}
let max_total = limit
.max_total_size
.unwrap_or(limit.frequency * limit.max_size);
let fits_an_empty_window = size <= limit.max_size
&& size <= max_total * self.limit_factor
&& 1.0 <= limit.frequency * self.limit_factor
&& (!counts_against_non_tx
|| (1.0 <= self.limits.non_tx_frequency * self.limit_factor
&& size <= self.limits.non_tx_max_total_size * self.limit_factor));
let new_count = self.counts.get(&opcode).unwrap_or(&0.0) + 1.0;
let new_cumulative = self.cumulative_sizes.get(&opcode).unwrap_or(&0.0) + size;
let (new_non_tx_count, new_non_tx_size) = if counts_against_non_tx {
(self.non_tx_count + 1.0, self.non_tx_size + size)
} else {
(self.non_tx_count, self.non_tx_size)
};
let fits_this_window = new_non_tx_count <= self.limits.non_tx_frequency * self.limit_factor
&& new_non_tx_size <= self.limits.non_tx_max_total_size * self.limit_factor
&& new_count <= limit.frequency * self.limit_factor
&& new_cumulative <= max_total * self.limit_factor;
let verdict = match (fits_an_empty_window, fits_this_window) {
(false, _) => Admission::Unsendable,
(true, false) => Admission::Deferred,
(true, true) => Admission::Admitted,
};
let charge = self.direction == Direction::Inbound || verdict == Admission::Admitted;
if charge {
self.counts.insert(opcode, new_count);
self.cumulative_sizes.insert(opcode, new_cumulative);
self.non_tx_count = new_non_tx_count;
self.non_tx_size = new_non_tx_size;
}
verdict
}
fn roll_window(&mut self) {
let period = now_seconds() / self.reset_seconds;
if self.period == period {
return;
}
self.period = period;
self.counts.clear();
self.cumulative_sizes.clear();
self.non_tx_count = 0.0;
self.non_tx_size = 0.0;
}
}
fn now_seconds() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("system clock is before the unix epoch")
.as_secs()
}
#[cfg(test)]
mod tests {
use super::{Admission, Direction, OpcodeRateLimiter, OpcodeRateLimits};
use crate::{Bytes, DigMessage, DIG_MESSAGE};
use chia_protocol::ProtocolMessageTypes;
use chia_sdk_client::{RateLimit, RateLimits, V2_RATE_LIMITS};
use chia_traits::Streamable;
fn message(opcode: u8, payload_len: usize) -> DigMessage {
DigMessage::new(opcode, None, Bytes::new(vec![0u8; payload_len]))
}
fn handshake_opcode() -> u8 {
*ProtocolMessageTypes::Handshake
.to_bytes()
.expect("encode")
.first()
.expect("one byte")
}
fn handshake_capped_at_two() -> RateLimits {
let mut limits = V2_RATE_LIMITS.clone();
limits.other.insert(
ProtocolMessageTypes::Handshake,
RateLimit::new(2.0, 10.0 * 1024.0, None),
);
limits
}
fn admit_handshakes(limits: OpcodeRateLimits, count: usize) -> Vec<Admission> {
let mut limiter = OpcodeRateLimiter::new(Direction::Outbound, 60, 1.0, limits);
(0..count)
.map(|_| limiter.admit(&message(handshake_opcode(), 16)))
.collect()
}
fn new_peak_opcode() -> u8 {
*ProtocolMessageTypes::NewPeak
.to_bytes()
.expect("encode")
.first()
.expect("one byte")
}
const OVERSIZED_HANDSHAKE: usize = 10 * 1024 + 1;
fn flood_then_probe(direction: Direction) -> Admission {
let mut limiter = OpcodeRateLimiter::new(
direction,
60,
1.0,
OpcodeRateLimits::from(&handshake_capped_at_two()),
);
for i in 0..2 {
assert_eq!(
limiter.admit(&message(handshake_opcode(), OVERSIZED_HANDSHAKE)),
Admission::Unsendable,
"flood frame {i} was not refused on size -- the fixture is not exercising a refusal"
);
}
limiter.admit(&message(handshake_opcode(), 16))
}
#[test]
fn an_inbound_refusal_is_charged_against_the_window() {
assert_eq!(
flood_then_probe(Direction::Inbound),
Admission::Deferred,
"a legal inbound frame was admitted after two refused frames -- the refusals were not \
charged, so a rejected flood is free"
);
}
#[test]
fn an_outbound_refusal_is_not_charged_against_the_window() {
assert_eq!(
flood_then_probe(Direction::Outbound),
Admission::Admitted,
"a refused outbound message was charged -- a backing-off caller is now penalised"
);
}
#[test]
fn an_inbound_refusal_is_charged_against_the_non_tx_count_aggregate() {
let probe = |direction| {
let mut table = V2_RATE_LIMITS.clone();
table.non_tx_frequency = 2.0;
table.other.insert(
ProtocolMessageTypes::Handshake,
RateLimit::new(100.0, 10.0 * 1024.0, None),
);
table.other.insert(
ProtocolMessageTypes::NewPeak,
RateLimit::new(100.0, 10.0 * 1024.0, None),
);
let mut limiter =
OpcodeRateLimiter::new(direction, 60, 1.0, OpcodeRateLimits::from(&table));
for _ in 0..2 {
assert_eq!(
limiter.admit(&message(handshake_opcode(), OVERSIZED_HANDSHAKE)),
Admission::Unsendable
);
}
limiter.admit(&message(new_peak_opcode(), 16))
};
assert_eq!(
probe(Direction::Inbound),
Admission::Deferred,
"a second `other` opcode was admitted after two refused frames -- the non_tx COUNT \
aggregate was not charged, so an oversized flood cannot exhaust the shared budget"
);
assert_eq!(
probe(Direction::Outbound),
Admission::Admitted,
"the fixture cannot distinguish the directions"
);
}
#[test]
fn an_inbound_refusal_is_charged_against_the_non_tx_size_aggregate() {
let probe = |direction| {
let mut table = V2_RATE_LIMITS.clone();
table.non_tx_frequency = 1000.0;
table.non_tx_max_total_size = 15.0 * 1024.0;
table.other.insert(
ProtocolMessageTypes::Handshake,
RateLimit::new(100.0, 10.0 * 1024.0, None),
);
table.other.insert(
ProtocolMessageTypes::NewPeak,
RateLimit::new(100.0, 10.0 * 1024.0, None),
);
let mut limiter =
OpcodeRateLimiter::new(direction, 60, 1.0, OpcodeRateLimits::from(&table));
for _ in 0..2 {
assert_eq!(
limiter.admit(&message(handshake_opcode(), OVERSIZED_HANDSHAKE)),
Admission::Unsendable
);
}
limiter.admit(&message(new_peak_opcode(), 16))
};
assert_eq!(
probe(Direction::Inbound),
Admission::Deferred,
"the non_tx SIZE aggregate was not charged for refused frames -- a flood of large \
rejected frames still costs the peer nothing"
);
assert_eq!(
probe(Direction::Outbound),
Admission::Admitted,
"the fixture cannot distinguish the directions"
);
}
#[test]
fn chia_opcodes_keep_their_upstream_limits() {
let limits = OpcodeRateLimits::default();
let handshake = *ProtocolMessageTypes::Handshake
.to_bytes()
.expect("encode")
.first()
.expect("one byte");
let upstream = chia_sdk_client::V2_RATE_LIMITS
.other
.get(&ProtocolMessageTypes::Handshake)
.expect("upstream defines a handshake limit");
let ours = limits
.other
.get(&handshake)
.expect("re-keyed table kept the handshake limit");
assert_eq!(ours.frequency, upstream.frequency);
assert_eq!(ours.max_size, upstream.max_size);
}
#[test]
fn every_upstream_opcode_survives_the_rekey_with_its_limits() {
let ours = OpcodeRateLimits::default();
let upstream = &*V2_RATE_LIMITS;
for (label, upstream_map, our_map) in [
("tx", &upstream.tx, &ours.tx),
("other", &upstream.other, &ours.other),
] {
assert_eq!(
our_map.len(),
upstream_map.len(),
"{label}: re-key changed the entry count, so an opcode was dropped or collided",
);
assert!(
!upstream_map.is_empty(),
"{label}: upstream table is empty, so this test proves nothing",
);
for (msg_type, expected) in upstream_map.iter() {
let opcode = *msg_type
.to_bytes()
.expect("ProtocolMessageTypes encodes")
.first()
.expect("one byte");
let got = our_map.get(&opcode).unwrap_or_else(|| {
panic!("{label}: {msg_type:?} (opcode {opcode}) missing after re-key")
});
assert_eq!(got.frequency, expected.frequency, "{label}: {msg_type:?}");
assert_eq!(got.max_size, expected.max_size, "{label}: {msg_type:?}");
}
}
}
#[test]
fn a_caller_supplied_table_governs_the_limiter() {
let verdicts = admit_handshakes(OpcodeRateLimits::from(&handshake_capped_at_two()), 3);
assert_eq!(
verdicts,
vec![
Admission::Admitted,
Admission::Admitted,
Admission::Deferred
],
"the custom frequency of 2 did not govern"
);
}
#[test]
fn default_still_derives_from_the_upstream_table() {
let via_default = admit_handshakes(OpcodeRateLimits::default(), 3);
let via_upstream = admit_handshakes(OpcodeRateLimits::from(&*V2_RATE_LIMITS), 3);
assert_eq!(
via_default, via_upstream,
"Default no longer agrees with the table it is documented to derive from"
);
assert_eq!(
via_default[2],
Admission::Admitted,
"upstream admits a third handshake (frequency 5); this probe cannot distinguish tables \
if it does not"
);
}
#[test]
fn dig_opcodes_fall_back_to_the_default_budget() {
let mut limiter =
OpcodeRateLimiter::new(Direction::Outbound, 60, 1.0, OpcodeRateLimits::default());
assert!(limiter.allow(&message(DIG_MESSAGE, 16)));
}
#[test]
fn frequency_budget_admits_up_to_the_bound_and_refuses_past_it() {
let limits = OpcodeRateLimits::default();
let allowance = limits.default_settings.frequency as usize;
let mut limiter = OpcodeRateLimiter::new(Direction::Outbound, 60, 1.0, limits);
for i in 0..allowance {
assert!(
limiter.allow(&message(DIG_MESSAGE, 1)),
"message {i} refused below the bound"
);
}
assert!(
!limiter.allow(&message(DIG_MESSAGE, 1)),
"one message over the bound was admitted"
);
}
#[test]
fn a_deferrable_refusal_is_distinguished_from_a_permanent_one() {
let limits = OpcodeRateLimits::default();
let allowance = limits.default_settings.frequency as usize;
let max_size = limits.default_settings.max_size as usize;
let mut exhausted = OpcodeRateLimiter::new(Direction::Outbound, 60, 1.0, limits);
for _ in 0..allowance {
assert_eq!(
exhausted.admit(&message(DIG_MESSAGE, 1)),
Admission::Admitted
);
}
assert_eq!(
exhausted.admit(&message(DIG_MESSAGE, 1)),
Admission::Deferred,
"an exhausted frequency budget resets on the next window, so waiting can help"
);
let mut fresh =
OpcodeRateLimiter::new(Direction::Outbound, 60, 1.0, OpcodeRateLimits::default());
assert_eq!(
fresh.admit(&message(DIG_MESSAGE, max_size + 1)),
Admission::Unsendable,
"an oversized message is refused identically in every window"
);
}
#[test]
fn size_cap_is_pinned_from_both_sides() {
let max_size = OpcodeRateLimits::default().default_settings.max_size as usize;
let mut at_bound =
OpcodeRateLimiter::new(Direction::Outbound, 60, 1.0, OpcodeRateLimits::default());
assert!(at_bound.allow(&message(DIG_MESSAGE, max_size)));
let mut over_bound =
OpcodeRateLimiter::new(Direction::Outbound, 60, 1.0, OpcodeRateLimits::default());
assert!(!over_bound.allow(&message(DIG_MESSAGE, max_size + 1)));
}
#[test]
fn the_rekeyed_table_pins_upstream_limits_at_absolute_values() {
let limits = OpcodeRateLimits::default();
let handshake = limits
.other
.get(&1)
.expect("opcode 1 (Handshake) kept its entry");
assert_eq!(handshake.frequency, 5.0, "Handshake frequency");
assert_eq!(handshake.max_size, 10.0 * 1024.0, "Handshake max_size");
let tx_ack = limits
.tx
.get(&49)
.expect("opcode 49 (TransactionAck) kept its tx entry");
assert_eq!(tx_ack.frequency, 5000.0, "TransactionAck frequency");
assert_eq!(tx_ack.max_size, 2048.0, "TransactionAck max_size");
let new_tx = limits
.tx
.get(&21)
.expect("opcode 21 (NewTransaction) kept its tx entry");
assert_eq!(new_tx.frequency, 5000.0, "NewTransaction frequency");
assert_eq!(new_tx.max_size, 100.0, "NewTransaction max_size");
assert_eq!(limits.non_tx_frequency, 1000.0);
assert_eq!(limits.non_tx_max_total_size, 100.0 * 1024.0 * 1024.0);
assert_eq!(limits.default_settings.frequency, 100.0);
assert_eq!(limits.default_settings.max_size, 1024.0 * 1024.0);
}
#[test]
fn falling_back_to_the_default_would_be_a_detectable_loosening() {
let limits = OpcodeRateLimits::default();
let handshake = limits
.other
.get(&1)
.expect("opcode 1 (Handshake) kept its entry");
assert!(
handshake.frequency < limits.default_settings.frequency,
"Handshake ({}) is not tighter than default ({}) -- the pin above can no longer distinguish a re-keyed table from a collapsed one",
handshake.frequency,
limits.default_settings.frequency
);
assert!(
handshake.max_size < limits.default_settings.max_size,
"Handshake max_size is not tighter than default"
);
}
#[test]
fn the_rekeyed_table_retains_the_bulk_of_the_upstream_entries() {
let limits = OpcodeRateLimits::default();
assert!(
limits.other.len() >= 30,
"other map holds only {} entries -- the re-key lost most of the table",
limits.other.len()
);
assert!(
limits.tx.len() >= 5,
"tx map holds only {} entries -- the re-key lost most of the table",
limits.tx.len()
);
for opcode in limits.other.keys().chain(limits.tx.keys()) {
assert!(
*opcode < 200,
"opcode {opcode} is outside the chia band -- the re-key is keying off a different ProtocolMessageTypes than the wire uses"
);
}
}
}