Skip to content

Commit 73fc0ec

Browse files
jmecomjmecom
authored
[codex] Enforce shared relay admission limits (BUZZ-SEC-019) (#1917)
1 parent caa195c commit 73fc0ec

8 files changed

Lines changed: 383 additions & 3 deletions

File tree

‎.env.example‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,16 @@ RELAY_URL=ws://localhost:3000
5151
# (use `just web` for Vite HMR instead).
5252
# BUZZ_WEB_DIR=./web/dist
5353

54+
# Shared Redis-backed admission limits. Defaults shown below; each value must
55+
# be a positive integer.
56+
# BUZZ_RATE_LIMIT_HUMAN_MESSAGES_PER_MIN=60
57+
# BUZZ_RATE_LIMIT_HUMAN_API_CALLS_PER_MIN=300
58+
# BUZZ_RATE_LIMIT_HUMAN_WS_EVENTS_PER_SEC=10
59+
# BUZZ_RATE_LIMIT_AGENT_STANDARD_MESSAGES_PER_MIN=120
60+
# BUZZ_RATE_LIMIT_AGENT_STANDARD_API_CALLS_PER_MIN=600
61+
# BUZZ_RATE_LIMIT_AGENT_ELEVATED_MESSAGES_PER_MIN=300
62+
# BUZZ_RATE_LIMIT_AGENT_PLATFORM_MESSAGES_PER_MIN=600
63+
5464
# -----------------------------------------------------------------------------
5565
# Git (NIP-34 bare repositories)
5666
# -----------------------------------------------------------------------------

‎.github/workflows/ci.yml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -459,6 +459,9 @@ jobs:
459459
BUZZ_BIND_ADDR=0.0.0.0:3000 \
460460
BUZZ_REQUIRE_AUTH_TOKEN=false \
461461
BUZZ_RECONCILE_CHANNELS=true \
462+
BUZZ_RATE_LIMIT_HUMAN_MESSAGES_PER_MIN=100000 \
463+
BUZZ_RATE_LIMIT_HUMAN_API_CALLS_PER_MIN=100000 \
464+
BUZZ_RATE_LIMIT_HUMAN_WS_EVENTS_PER_SEC=10000 \
462465
BUZZ_GIT_PROBE_WRITERS=8 \
463466
SPROUT_REMINDER_SCHEDULER_INTERVAL_SECS=1 \
464467
./target/ci/buzz-relay > /tmp/buzz-relay.log 2>&1 &

‎crates/buzz-relay/src/admission.rs‎

Lines changed: 158 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,158 @@
1+
use buzz_auth::{LimitType, RateLimiter};
2+
use buzz_core::TenantContext;
3+
use nostr::PublicKey;
4+
5+
// Desktop startup establishes several independent live subscriptions at once.
6+
// Preserve the configured average rate while allowing that bounded burst. This
7+
// is still a fixed-window limiter, so a Redis-backed token bucket would be a
8+
// better long-term fit for smoother refill behavior.
9+
const WS_BURST_WINDOW_SECS: u64 = 5;
10+
11+
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12+
pub(crate) enum AdmissionError {
13+
Exceeded { reset_in_secs: u64 },
14+
Unavailable,
15+
}
16+
17+
pub(crate) async fn check_principal<L: RateLimiter>(
18+
limiter: &L,
19+
tenant: &TenantContext,
20+
pubkey: &PublicKey,
21+
limit_type: LimitType,
22+
window_secs: u64,
23+
limit: u64,
24+
) -> Result<(), AdmissionError> {
25+
match limiter
26+
.check_and_increment(tenant, pubkey, limit_type, window_secs, limit)
27+
.await
28+
{
29+
Ok(result) if result.allowed => Ok(()),
30+
Ok(result) => Err(AdmissionError::Exceeded {
31+
reset_in_secs: result.reset_in_secs,
32+
}),
33+
Err(error) => {
34+
tracing::warn!(error = %error, "shared rate-limit admission unavailable");
35+
Err(AdmissionError::Unavailable)
36+
}
37+
}
38+
}
39+
40+
pub(crate) fn ws_admission_budget(per_second_limit: u64) -> (u64, u64) {
41+
(
42+
WS_BURST_WINDOW_SECS,
43+
per_second_limit.saturating_mul(WS_BURST_WINDOW_SECS),
44+
)
45+
}
46+
47+
#[cfg(test)]
48+
mod tests {
49+
use std::net::IpAddr;
50+
use std::sync::atomic::{AtomicUsize, Ordering};
51+
52+
use buzz_auth::{AuthError, RateLimitResult, RateLimiter};
53+
use buzz_core::CommunityId;
54+
use nostr::Keys;
55+
use uuid::Uuid;
56+
57+
use super::*;
58+
59+
enum StubOutcome {
60+
Denied,
61+
Failed,
62+
}
63+
64+
struct StubLimiter {
65+
outcome: StubOutcome,
66+
calls: AtomicUsize,
67+
}
68+
69+
impl RateLimiter for StubLimiter {
70+
async fn check_and_increment(
71+
&self,
72+
_ctx: &TenantContext,
73+
_pubkey: &PublicKey,
74+
_limit_type: LimitType,
75+
_window_secs: u64,
76+
_limit: u64,
77+
) -> Result<RateLimitResult, AuthError> {
78+
self.calls.fetch_add(1, Ordering::Relaxed);
79+
match self.outcome {
80+
StubOutcome::Denied => Ok(RateLimitResult::denied(11, 10, 1)),
81+
StubOutcome::Failed => Err(AuthError::Internal("redis unavailable".to_owned())),
82+
}
83+
}
84+
85+
async fn check_ip_connection(
86+
&self,
87+
_ip: &IpAddr,
88+
_window_secs: u64,
89+
_limit: u64,
90+
) -> Result<RateLimitResult, AuthError> {
91+
match self.outcome {
92+
StubOutcome::Denied => Ok(RateLimitResult::denied(11, 10, 1)),
93+
StubOutcome::Failed => Err(AuthError::Internal("redis unavailable".to_owned())),
94+
}
95+
}
96+
}
97+
98+
fn tenant() -> TenantContext {
99+
TenantContext::resolved(
100+
CommunityId::from_uuid(Uuid::from_u128(1)),
101+
"relay.example.com",
102+
)
103+
}
104+
105+
#[test]
106+
fn websocket_budget_preserves_rate_with_a_bounded_burst() {
107+
assert_eq!(ws_admission_budget(10), (5, 50));
108+
}
109+
110+
#[test]
111+
fn websocket_budget_saturates_on_overflow() {
112+
assert_eq!(ws_admission_budget(u64::MAX), (5, u64::MAX));
113+
}
114+
115+
#[tokio::test]
116+
async fn denied_shared_counter_rejects_admission() {
117+
let limiter = StubLimiter {
118+
outcome: StubOutcome::Denied,
119+
calls: AtomicUsize::new(0),
120+
};
121+
let keys = Keys::generate();
122+
123+
let result = check_principal(
124+
&limiter,
125+
&tenant(),
126+
&keys.public_key(),
127+
LimitType::WsEvents,
128+
1,
129+
10,
130+
)
131+
.await;
132+
133+
assert_eq!(result, Err(AdmissionError::Exceeded { reset_in_secs: 1 }));
134+
assert_eq!(limiter.calls.load(Ordering::Relaxed), 1);
135+
}
136+
137+
#[tokio::test]
138+
async fn shared_counter_failure_rejects_admission() {
139+
let limiter = StubLimiter {
140+
outcome: StubOutcome::Failed,
141+
calls: AtomicUsize::new(0),
142+
};
143+
let keys = Keys::generate();
144+
145+
let result = check_principal(
146+
&limiter,
147+
&tenant(),
148+
&keys.public_key(),
149+
LimitType::ApiCalls,
150+
60,
151+
300,
152+
)
153+
.await;
154+
155+
assert_eq!(result, Err(AdmissionError::Unavailable));
156+
assert_eq!(limiter.calls.load(Ordering::Relaxed), 1);
157+
}
158+
}

‎crates/buzz-relay/src/api/bridge.rs‎

Lines changed: 38 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,14 +13,48 @@ use axum::{
1313
use base64::Engine;
1414
use serde_json::Value;
1515

16-
use buzz_auth::{Nip98ReplayGuard, DEFAULT_REPLAY_TTL_SECS};
16+
use buzz_auth::{LimitType, Nip98ReplayGuard, DEFAULT_REPLAY_TTL_SECS};
1717
use buzz_core::TenantContext;
1818

1919
use crate::handlers::ingest::{IngestAuth, IngestError};
2020
use crate::state::AppState;
2121

2222
use super::{api_error, internal_error, not_found};
2323

24+
async fn enforce_http_admission(
25+
state: &AppState,
26+
tenant: &TenantContext,
27+
pubkey: &nostr::PublicKey,
28+
) -> Result<(), (StatusCode, Json<Value>)> {
29+
let limit = state.auth.config().rate_limits.human_api_calls_per_min;
30+
match crate::admission::check_principal(
31+
state.admission_rate_limiter.as_ref(),
32+
tenant,
33+
pubkey,
34+
LimitType::ApiCalls,
35+
60,
36+
limit,
37+
)
38+
.await
39+
{
40+
Ok(()) => Ok(()),
41+
Err(crate::admission::AdmissionError::Exceeded { reset_in_secs }) => {
42+
metrics::counter!("buzz_admission_rejections_total", "transport" => "http", "reason" => "quota").increment(1);
43+
Err(api_error(
44+
StatusCode::TOO_MANY_REQUESTS,
45+
&format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"),
46+
))
47+
}
48+
Err(crate::admission::AdmissionError::Unavailable) => {
49+
metrics::counter!("buzz_admission_rejections_total", "transport" => "http", "reason" => "unavailable").increment(1);
50+
Err(api_error(
51+
StatusCode::SERVICE_UNAVAILABLE,
52+
"rate-limited: shared admission unavailable",
53+
))
54+
}
55+
}
56+
}
57+
2458
/// Verify bridge auth: NIP-98 (production) or X-Pubkey (dev mode).
2559
///
2660
/// Returns the authenticated public key and an event ID for replay detection.
@@ -584,6 +618,7 @@ pub async fn submit_event(
584618
Some(&body),
585619
state.config.require_auth_token,
586620
)?;
621+
enforce_http_admission(&state, &tenant, &pubkey).await?;
587622
check_nip98_replay(&state, &tenant, event_id_bytes).await?;
588623
let pubkey_bytes = pubkey.to_bytes().to_vec();
589624

@@ -653,6 +688,7 @@ pub async fn query_events(
653688
Some(&body),
654689
state.config.require_auth_token,
655690
)?;
691+
enforce_http_admission(&state, &tenant, &pubkey).await?;
656692
check_nip98_replay(&state, &tenant, event_id_bytes).await?;
657693
let pubkey_bytes = pubkey.to_bytes().to_vec();
658694

@@ -1036,6 +1072,7 @@ pub async fn count_events(
10361072
Some(&body),
10371073
state.config.require_auth_token,
10381074
)?;
1075+
enforce_http_admission(&state, &tenant, &pubkey).await?;
10391076
check_nip98_replay(&state, &tenant, event_id_bytes).await?;
10401077
let pubkey_bytes = pubkey.to_bytes().to_vec();
10411078

‎crates/buzz-relay/src/config.rs‎

Lines changed: 82 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -233,6 +233,54 @@ fn parse_bind_addr(raw: &str) -> Result<SocketAddr, ConfigError> {
233233
.map_err(|e| ConfigError::InvalidBindAddr(e.to_string()))
234234
}
235235

236+
fn positive_u64_from_env(name: &str, default: u64) -> Result<u64, ConfigError> {
237+
match std::env::var(name) {
238+
Ok(raw) => raw
239+
.parse::<u64>()
240+
.ok()
241+
.filter(|value| *value > 0)
242+
.ok_or_else(|| ConfigError::InvalidValue(format!("{name} must be a positive integer"))),
243+
Err(std::env::VarError::NotPresent) => Ok(default),
244+
Err(std::env::VarError::NotUnicode(_)) => Err(ConfigError::InvalidValue(format!(
245+
"{name} must be valid Unicode"
246+
))),
247+
}
248+
}
249+
250+
fn rate_limit_config_from_env() -> Result<buzz_auth::RateLimitConfig, ConfigError> {
251+
let defaults = buzz_auth::RateLimitConfig::default();
252+
Ok(buzz_auth::RateLimitConfig {
253+
human_messages_per_min: positive_u64_from_env(
254+
"BUZZ_RATE_LIMIT_HUMAN_MESSAGES_PER_MIN",
255+
defaults.human_messages_per_min,
256+
)?,
257+
human_api_calls_per_min: positive_u64_from_env(
258+
"BUZZ_RATE_LIMIT_HUMAN_API_CALLS_PER_MIN",
259+
defaults.human_api_calls_per_min,
260+
)?,
261+
human_ws_events_per_sec: positive_u64_from_env(
262+
"BUZZ_RATE_LIMIT_HUMAN_WS_EVENTS_PER_SEC",
263+
defaults.human_ws_events_per_sec,
264+
)?,
265+
agent_standard_messages_per_min: positive_u64_from_env(
266+
"BUZZ_RATE_LIMIT_AGENT_STANDARD_MESSAGES_PER_MIN",
267+
defaults.agent_standard_messages_per_min,
268+
)?,
269+
agent_standard_api_calls_per_min: positive_u64_from_env(
270+
"BUZZ_RATE_LIMIT_AGENT_STANDARD_API_CALLS_PER_MIN",
271+
defaults.agent_standard_api_calls_per_min,
272+
)?,
273+
agent_elevated_messages_per_min: positive_u64_from_env(
274+
"BUZZ_RATE_LIMIT_AGENT_ELEVATED_MESSAGES_PER_MIN",
275+
defaults.agent_elevated_messages_per_min,
276+
)?,
277+
agent_platform_messages_per_min: positive_u64_from_env(
278+
"BUZZ_RATE_LIMIT_AGENT_PLATFORM_MESSAGES_PER_MIN",
279+
defaults.agent_platform_messages_per_min,
280+
)?,
281+
})
282+
}
283+
236284
fn parse_operator_api_origin(raw: &str) -> Result<String, ConfigError> {
237285
let raw = raw.trim();
238286
let url = url::Url::parse(raw).map_err(|e| {
@@ -477,7 +525,9 @@ impl Config {
477525
));
478526
}
479527

480-
let auth = buzz_auth::AuthConfig::default();
528+
let auth = buzz_auth::AuthConfig {
529+
rate_limits: rate_limit_config_from_env()?,
530+
};
481531

482532
if !require_auth_token {
483533
warn!(
@@ -842,6 +892,37 @@ mod tests {
842892
));
843893
}
844894

895+
#[test]
896+
fn rate_limits_can_be_overridden() {
897+
let _guard = ENV_MUTEX.lock().unwrap();
898+
std::env::set_var("BUZZ_RATE_LIMIT_HUMAN_MESSAGES_PER_MIN", "1001");
899+
std::env::set_var("BUZZ_RATE_LIMIT_HUMAN_API_CALLS_PER_MIN", "1002");
900+
std::env::set_var("BUZZ_RATE_LIMIT_HUMAN_WS_EVENTS_PER_SEC", "1003");
901+
902+
let config = Config::from_env().expect("config");
903+
904+
std::env::remove_var("BUZZ_RATE_LIMIT_HUMAN_MESSAGES_PER_MIN");
905+
std::env::remove_var("BUZZ_RATE_LIMIT_HUMAN_API_CALLS_PER_MIN");
906+
std::env::remove_var("BUZZ_RATE_LIMIT_HUMAN_WS_EVENTS_PER_SEC");
907+
assert_eq!(config.auth.rate_limits.human_messages_per_min, 1001);
908+
assert_eq!(config.auth.rate_limits.human_api_calls_per_min, 1002);
909+
assert_eq!(config.auth.rate_limits.human_ws_events_per_sec, 1003);
910+
}
911+
912+
#[test]
913+
fn rate_limit_overrides_reject_zero() {
914+
let _guard = ENV_MUTEX.lock().unwrap();
915+
std::env::set_var("BUZZ_RATE_LIMIT_HUMAN_WS_EVENTS_PER_SEC", "0");
916+
let result = Config::from_env();
917+
std::env::remove_var("BUZZ_RATE_LIMIT_HUMAN_WS_EVENTS_PER_SEC");
918+
919+
assert!(matches!(
920+
result,
921+
Err(ConfigError::InvalidValue(ref message))
922+
if message.contains("BUZZ_RATE_LIMIT_HUMAN_WS_EVENTS_PER_SEC")
923+
));
924+
}
925+
845926
#[test]
846927
fn relay_operator_pubkeys_parse_dedupe_and_normalize() {
847928
let _guard = ENV_MUTEX.lock().unwrap();

0 commit comments

Comments
 (0)