diff --git a/src/muxed.rs b/src/muxed.rs index 72e2ed2..ee440c5 100644 --- a/src/muxed.rs +++ b/src/muxed.rs @@ -8,8 +8,8 @@ use data_encoding::BASE32; use serde::{Deserialize, Serialize}; /// Stellar strkey version bytes -const VERSION_ACCOUNT_ID: u8 = 6; // G-address -const VERSION_MUXED_ACCOUNT: u8 = 12; // M-address +const VERSION_ACCOUNT_ID: u8 = 6 << 3; // G-address +const VERSION_MUXED_ACCOUNT: u8 = 12 << 3; // M-address /// Length of a Stellar M-address (`MUXED_ACCOUNT` strkey) pub const MUXED_ADDRESS_LEN: usize = 69; @@ -84,7 +84,7 @@ pub fn parse_muxed_address(addr: &str) -> Option { if decoded[0] != VERSION_MUXED_ACCOUNT { return None; } - let checksum = u16::from_be_bytes([decoded[41], decoded[42]]); + let checksum = u16::from_le_bytes([decoded[41], decoded[42]]); let payload = &decoded[0..41]; if crc16(payload) != checksum { return None; @@ -97,9 +97,8 @@ pub fn parse_muxed_address(addr: &str) -> Option { let mut g_payload = [0u8; 35]; g_payload[0] = VERSION_ACCOUNT_ID; g_payload[1..33].copy_from_slice(account_id); - let c = crc16(&g_payload); - g_payload[33] = (c >> 8) as u8; - g_payload[34] = (c & 0xff) as u8; + let checksum = crc16(&g_payload[..33]).to_le_bytes(); + g_payload[33..35].copy_from_slice(&checksum); let base_account = BASE32.encode(&g_payload); Some(MuxedAccountInfo { @@ -160,13 +159,15 @@ mod tests { parse_muxed_address("GA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVSGZ") .is_none() ); - // Valid M-address format: parse may succeed (if checksum/version match) or None + // SEP-0023 regression: decode a real Stellar M-address exactly. let m = "MAAAAAAAAAAAAAB7BQ2L7E5NBWMXDUCMZSIPOBKRDSBYVLMXGSSKF6YNPIB7Y77ITLVL6"; - let info = parse_muxed_address(m); - if let Some(ref i) = info { - assert!(i.muxed_address == m); - assert!(i.base_account.as_ref().is_none_or(|g| g.starts_with('G'))); - } + let info = parse_muxed_address(m).expect("valid Stellar muxed account"); + assert_eq!(info.muxed_address, m); + assert_eq!( + info.base_account.as_deref(), + Some("GAAAAAAAAAAAAAB7BQ2L7E5NBWMXDUCMZSIPOBKRDSBYVLMXGSSKFK3W") + ); + assert_eq!(info.muxed_id, Some(18_090_249_435_756_882_074)); // Too short M string assert!(parse_muxed_address("M").is_none()); } diff --git a/src/rate_limit.rs b/src/rate_limit.rs index 41dc380..3634f4b 100644 --- a/src/rate_limit.rs +++ b/src/rate_limit.rs @@ -1,3 +1,6 @@ +pub mod bucket; +pub mod policy; + use anyhow::Context; use axum::{ extract::{ConnectInfo, Request, State}, diff --git a/src/rate_limit/bucket.rs b/src/rate_limit/bucket.rs new file mode 100644 index 0000000..5627038 --- /dev/null +++ b/src/rate_limit/bucket.rs @@ -0,0 +1,81 @@ +//! Generic in-memory token bucket used by rate-limit policies. + +use std::{collections::HashMap, hash::Hash, time::Duration}; + +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct BucketSnapshot { + pub allowed: bool, + pub remaining: u32, +} + +#[derive(Debug, Clone, Copy)] +struct BucketState { + tokens: f64, + last_refill: Duration, +} + +/// Small deterministic token bucket keyed by any hashable identity. +/// +/// Callers provide the timestamp so tests and higher-level policies can make +/// atomic decisions across multiple buckets without sleeping. +#[derive(Debug)] +pub struct TokenBucket { + capacity: u32, + refill_per_second: f64, + states: HashMap, +} + +impl TokenBucket +where + K: Eq + Hash + Clone, +{ + #[must_use] + pub fn new(capacity: u32, refill_per_second: f64) -> Self { + Self { + capacity, + refill_per_second: refill_per_second.max(0.0), + states: HashMap::new(), + } + } + + fn state_at(&mut self, key: &K, now: Duration) -> &mut BucketState { + let capacity = f64::from(self.capacity); + let refill_per_second = self.refill_per_second; + let state = self.states.entry(key.clone()).or_insert(BucketState { + tokens: capacity, + last_refill: now, + }); + + if now > state.last_refill { + let elapsed = now.saturating_sub(state.last_refill).as_secs_f64(); + state.tokens = (state.tokens + elapsed * refill_per_second).min(capacity); + state.last_refill = now; + } + + state + } + + pub fn can_take(&mut self, key: &K, now: Duration) -> bool { + self.state_at(key, now).tokens >= 1.0 + } + + pub fn remaining(&mut self, key: &K, now: Duration) -> u32 { + self.state_at(key, now).tokens.floor().max(0.0) as u32 + } + + pub fn take(&mut self, key: &K, now: Duration) -> BucketSnapshot { + let state = self.state_at(key, now); + if state.tokens < 1.0 { + return BucketSnapshot { + allowed: false, + remaining: state.tokens.floor().max(0.0) as u32, + }; + } + + state.tokens -= 1.0; + BucketSnapshot { + allowed: true, + remaining: state.tokens.floor().max(0.0) as u32, + } + } +} diff --git a/src/rate_limit/policy.rs b/src/rate_limit/policy.rs new file mode 100644 index 0000000..4d63c43 --- /dev/null +++ b/src/rate_limit/policy.rs @@ -0,0 +1,134 @@ +//! Two-tier rate limiting for Stellar muxed accounts. +//! +//! A muxed address has two useful identities: the full M-address and its shared +//! base G-address. Enforcing both prevents muxed-ID rotation from bypassing a +//! limit while keeping one busy sub-account from consuming every sibling's +//! per-muxed budget. + +use std::time::Duration; + +use thiserror::Error; + +use crate::muxed::parse_muxed_address; + +use super::bucket::TokenBucket; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MuxedRateLimitTier { + MuxedAddress, + BaseAccount, +} + +impl MuxedRateLimitTier { + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::MuxedAddress => "muxed_address", + Self::BaseAccount => "base_account", + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct MuxedRateLimitConfig { + pub per_muxed_capacity: u32, + pub per_muxed_refill_per_second: f64, + pub per_base_capacity: u32, + pub per_base_refill_per_second: f64, +} + +// Deliberately no Default: base-account fan-out is a product decision, not an engineering guess. + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum MuxedRateLimitError { + #[error("invalid Stellar muxed account address")] + InvalidMuxedAddress, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MuxedRateLimitDecision { + pub allowed: bool, + pub hit_tier: Option, + pub muxed_remaining: u32, + pub base_remaining: u32, +} + +impl MuxedRateLimitDecision { + /// Stable response value suitable for a JSON rejection field/header. + #[must_use] + pub const fn rejection_tier(&self) -> Option<&'static str> { + match self.hit_tier { + Some(tier) => Some(tier.as_str()), + None => None, + } + } +} + +#[derive(Debug)] +pub struct MuxedAccountRateLimiter { + per_muxed: TokenBucket, + per_base: TokenBucket, +} + +impl MuxedAccountRateLimiter { + #[must_use] + pub fn new(config: MuxedRateLimitConfig) -> Self { + Self { + per_muxed: TokenBucket::new( + config.per_muxed_capacity, + config.per_muxed_refill_per_second, + ), + per_base: TokenBucket::new( + config.per_base_capacity, + config.per_base_refill_per_second, + ), + } + } + + /// Enforce full-M-address and base-G-address budgets atomically. + /// + /// Both tiers are previewed before either token is consumed, so a request + /// rejected by the shared base tier does not burn the caller's muxed token. + pub fn check( + &mut self, + address: &str, + now: Duration, + ) -> Result { + let parsed = parse_muxed_address(address).ok_or(MuxedRateLimitError::InvalidMuxedAddress)?; + let base = parsed + .base_account + .ok_or(MuxedRateLimitError::InvalidMuxedAddress)?; + let muxed = parsed.muxed_address; + + let muxed_allowed = self.per_muxed.can_take(&muxed, now); + let base_allowed = self.per_base.can_take(&base, now); + + if !muxed_allowed { + return Ok(MuxedRateLimitDecision { + allowed: false, + hit_tier: Some(MuxedRateLimitTier::MuxedAddress), + muxed_remaining: self.per_muxed.remaining(&muxed, now), + base_remaining: self.per_base.remaining(&base, now), + }); + } + + if !base_allowed { + return Ok(MuxedRateLimitDecision { + allowed: false, + hit_tier: Some(MuxedRateLimitTier::BaseAccount), + muxed_remaining: self.per_muxed.remaining(&muxed, now), + base_remaining: self.per_base.remaining(&base, now), + }); + } + + let muxed_snapshot = self.per_muxed.take(&muxed, now); + let base_snapshot = self.per_base.take(&base, now); + + Ok(MuxedRateLimitDecision { + allowed: muxed_snapshot.allowed && base_snapshot.allowed, + hit_tier: None, + muxed_remaining: muxed_snapshot.remaining, + base_remaining: base_snapshot.remaining, + }) + } +} diff --git a/tests/muxed_bypass_test.rs b/tests/muxed_bypass_test.rs new file mode 100644 index 0000000..84cd249 --- /dev/null +++ b/tests/muxed_bypass_test.rs @@ -0,0 +1,56 @@ +use std::time::Duration; + +use data_encoding::BASE32; +use stellar_analysis_backend::rate_limit::policy::{ + MuxedAccountRateLimiter, MuxedRateLimitConfig, MuxedRateLimitTier, +}; + +const VERSION_MUXED_ACCOUNT: u8 = 12 << 3; +const CRC16_POLY: u16 = 0x1021; + +fn crc16(data: &[u8]) -> u16 { + let mut crc = 0u16; + for &byte in data { + crc ^= u16::from(byte) << 8; + for _ in 0..8 { + crc = if crc & 0x8000 != 0 { + (crc << 1) ^ CRC16_POLY + } else { + crc << 1 + }; + } + } + crc +} + +fn muxed_address(account_id: [u8; 32], muxed_id: u64) -> String { + let mut raw = [0u8; 43]; + raw[0] = VERSION_MUXED_ACCOUNT; + raw[1..33].copy_from_slice(&account_id); + raw[33..41].copy_from_slice(&muxed_id.to_be_bytes()); + let checksum = crc16(&raw[..41]).to_le_bytes(); + raw[41..43].copy_from_slice(&checksum); + BASE32.encode(&raw) +} + +#[test] +fn rotating_muxed_ids_cannot_bypass_shared_base_limit() { + let mut limiter = MuxedAccountRateLimiter::new(MuxedRateLimitConfig { + per_muxed_capacity: 2, + per_muxed_refill_per_second: 0.0, + per_base_capacity: 3, + per_base_refill_per_second: 0.0, + }); + let account = [7u8; 32]; + let now = Duration::ZERO; + + for id in 1..=3 { + let decision = limiter.check(&muxed_address(account, id), now).unwrap(); + assert!(decision.allowed); + } + + let rejected = limiter.check(&muxed_address(account, 4), now).unwrap(); + assert!(!rejected.allowed); + assert_eq!(rejected.hit_tier, Some(MuxedRateLimitTier::BaseAccount)); + assert_eq!(rejected.rejection_tier(), Some("base_account")); +} diff --git a/tests/muxed_noisy_neighbor_test.rs b/tests/muxed_noisy_neighbor_test.rs new file mode 100644 index 0000000..eb83f37 --- /dev/null +++ b/tests/muxed_noisy_neighbor_test.rs @@ -0,0 +1,63 @@ +use std::time::Duration; + +use data_encoding::BASE32; +use stellar_analysis_backend::rate_limit::policy::{ + MuxedAccountRateLimiter, MuxedRateLimitConfig, MuxedRateLimitTier, +}; + +const VERSION_MUXED_ACCOUNT: u8 = 12 << 3; +const CRC16_POLY: u16 = 0x1021; + +fn crc16(data: &[u8]) -> u16 { + let mut crc = 0u16; + for &byte in data { + crc ^= u16::from(byte) << 8; + for _ in 0..8 { + crc = if crc & 0x8000 != 0 { + (crc << 1) ^ CRC16_POLY + } else { + crc << 1 + }; + } + } + crc +} + +fn muxed_address(account_id: [u8; 32], muxed_id: u64) -> String { + let mut raw = [0u8; 43]; + raw[0] = VERSION_MUXED_ACCOUNT; + raw[1..33].copy_from_slice(&account_id); + raw[33..41].copy_from_slice(&muxed_id.to_be_bytes()); + let checksum = crc16(&raw[..41]).to_le_bytes(); + raw[41..43].copy_from_slice(&checksum); + BASE32.encode(&raw) +} + +#[test] +fn one_noisy_muxed_account_does_not_exhaust_sibling_budgets() { + let mut limiter = MuxedAccountRateLimiter::new(MuxedRateLimitConfig { + per_muxed_capacity: 2, + per_muxed_refill_per_second: 0.0, + per_base_capacity: 10, + per_base_refill_per_second: 0.0, + }); + let account = [11u8; 32]; + let now = Duration::ZERO; + let noisy = muxed_address(account, 1); + + assert!(limiter.check(&noisy, now).unwrap().allowed); + assert!(limiter.check(&noisy, now).unwrap().allowed); + + let noisy_rejection = limiter.check(&noisy, now).unwrap(); + assert!(!noisy_rejection.allowed); + assert_eq!( + noisy_rejection.hit_tier, + Some(MuxedRateLimitTier::MuxedAddress) + ); + assert_eq!(noisy_rejection.rejection_tier(), Some("muxed_address")); + + for id in 2..=6 { + let sibling = limiter.check(&muxed_address(account, id), now).unwrap(); + assert!(sibling.allowed, "sibling muxed id {id} should remain available"); + } +}