feat(rate-limit): enforce muxed and base account limits

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<MuxedAccountInfo> {
     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<MuxedAccountInfo> {
     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());
     }

src/rate_limit.rs

@@ -1,3 +1,6 @@
+pub mod bucket;
+pub mod policy;
+
 use anyhow::Context;
 use axum::{
     extract::{ConnectInfo, Request, State},

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<K> {
+    capacity: u32,
+    refill_per_second: f64,
+    states: HashMap<K, BucketState>,
+}
+
+impl<K> TokenBucket<K>
+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,
+        }
+    }
+}

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<MuxedRateLimitTier>,
+    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<String>,
+    per_base: TokenBucket<String>,
+}
+
+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<MuxedRateLimitDecision, MuxedRateLimitError> {
+        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,
+        })
+    }
+}

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"));
+}

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");
+    }
+}