crates/gunnar-ui-server/src/throttle.rs
//! **The one throttle every request passes** (lane GUN-GUEST, owner 2026-10-30: "NOBODY
//! sidesteps throttling").
//!
//! Mounted around the WHOLE router ([`crate::serve::router`]), so every route, the static
//! files, the console's shell and every 401 are counted, in every build. What differs is the
//! budget, by who is asking:
//!
//! | who | counted by | budget |
//! |---|---|---|
//! | a guest (no credential) | its address: IPv4, or the IPv6 /64 ([`ClientKey`], RA5-34) | [`Limits::guest`] |
//! | a request with a session cookie or a bearer | its address, a higher budget | [`Limits::credentialed`] |
//! | a signed-in person, once the gate knows them | their account (gunnar principal) | [`Limits::account`] (charged in `require_auth`) |
//! | a machine / API token (the admin bearer, an MCP token) | the token's hash, never the token | [`Limits::token`] |
//! | everybody together | one bucket | [`Limits::global`] |
//!
//! Each tier is a token bucket (burst + refill) AND a cap on requests in flight. A request
//! must find room in every tier it is counted in, or it takes room in none, so one client
//! that is refused spends nobody else's budget. Past a limit the answer is a `429` with
//! `Retry-After`, in words, made here: no handler runs, gunnar is not asked, nothing is
//! rendered. A client refused [`Limits::block_after`] times within [`Limits::block_window`]
//! is blocked for [`Limits::block_for`] (each block is an audit line). The on-demand
//! renders have their own, lower budget per client and in all ([`Throttle::admit_render`]).
//!
//! The address is the console's ONE rule ([`crate::security::client_ip`]): the socket's
//! peer, and `x-forwarded-for` only from a loopback or `GUNNAR_UI_TRUSTED_PROXIES` peer (the
//! front → console hop), so a client cannot pick its own key.
//!
//! The limits are one setting, [`ENV_LIMITS`], with safe defaults ([`Limits::default`]).
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use axum::extract::{Request, State};
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use http::{header, HeaderMap, HeaderValue, StatusCode};
use nornir_auth_webauthn::client_key::ClientKey;
use crate::security::Buckets;
/// The environment setting for the limits: comma-separated `tier=burst/per_second[/in_flight]`
/// entries (tiers `guest`, `credentialed`, `account`, `token`, `global`; `render` and
/// `render_all` take `burst/per_second`), and `block=refusals/window_secs/block_secs`. A tier
/// not named keeps its default; an entry that does not read is skipped with a warning.
pub const ENV_LIMITS: &str = "GUNNAR_UI_THROTTLE";
/// One bucket's size and refill.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Budget {
pub burst: u32,
pub per_second: f64,
}
/// One tier: its bucket and its cap on requests in flight at once.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Tier {
pub budget: Budget,
pub in_flight: usize,
}
/// Every limit.
#[derive(Debug, Clone, PartialEq)]
pub struct Limits {
pub guest: Tier,
pub credentialed: Tier,
pub account: Tier,
pub token: Tier,
pub global: Tier,
/// On-demand renders, per client.
pub render: Budget,
/// On-demand renders, everybody together.
pub render_all: Budget,
/// Refusals within [`Limits::block_window`] that block a client...
pub block_after: u32,
pub block_window: Duration,
/// ...for this long.
pub block_for: Duration,
/// Keys held per tier at most (the least recently seen go first).
pub max_keys: usize,
}
impl Default for Limits {
/// Generous for a person (a console page load is its document, a few assets and some ten
/// API calls), tight for a scraper.
fn default() -> Limits {
let tier = |burst, per_second, in_flight| Tier { budget: Budget { burst, per_second }, in_flight };
Limits {
guest: tier(120, 4.0, 16),
credentialed: tier(600, 20.0, 64),
account: tier(600, 20.0, 64),
token: tier(600, 20.0, 32),
global: tier(5000, 500.0, 1024),
render: Budget { burst: 10, per_second: 0.5 },
render_all: Budget { burst: 60, per_second: 3.0 },
block_after: 50,
block_window: Duration::from_secs(60),
block_for: Duration::from_secs(300),
max_keys: 10_000,
}
}
}
impl Limits {
/// The defaults with [`ENV_LIMITS`] applied.
pub fn from_env() -> Limits {
Limits::parse(std::env::var(ENV_LIMITS).ok().as_deref().unwrap_or(""))
}
/// The defaults with `setting` applied (see [`ENV_LIMITS`]).
pub fn parse(setting: &str) -> Limits {
let mut l = Limits::default();
for entry in setting.split(',').map(str::trim).filter(|e| !e.is_empty()) {
if !l.apply(entry) {
tracing::warn!(entry, "{ENV_LIMITS} has an entry that does not read; that limit keeps its default");
}
}
l
}
fn apply(&mut self, entry: &str) -> bool {
let Some((name, value)) = entry.split_once('=') else { return false };
let parts: Vec<&str> = value.split('/').map(str::trim).collect();
let num = |i: usize| parts.get(i).and_then(|p| p.parse::<f64>().ok()).filter(|v| v.is_finite() && *v >= 0.0);
let budget = || Some(Budget { burst: num(0)? as u32, per_second: num(1)? });
let tier = |old: Tier| -> Option<Tier> {
let budget = budget()?;
let in_flight = match parts.len() {
2 => old.in_flight,
3 => num(2)? as usize,
_ => return None,
};
Some(Tier { budget, in_flight })
};
match name.trim() {
"guest" => tier(self.guest).map(|t| self.guest = t),
"credentialed" => tier(self.credentialed).map(|t| self.credentialed = t),
"account" => tier(self.account).map(|t| self.account = t),
"token" => tier(self.token).map(|t| self.token = t),
"global" => tier(self.global).map(|t| self.global = t),
"render" if parts.len() == 2 => budget().map(|b| self.render = b),
"render_all" if parts.len() == 2 => budget().map(|b| self.render_all = b),
"block" if parts.len() == 3 => (|| {
let (n, w, f) = (num(0)?, num(1)?, num(2)?);
self.block_after = n as u32;
self.block_window = Duration::from_secs_f64(w);
self.block_for = Duration::from_secs_f64(f);
Some(())
})(),
_ => None,
}
.is_some()
}
}
/// The kind of tier a key is counted in (for the in-flight table).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum Kind {
Guest,
Credentialed,
Account,
Token,
}
/// The request's client, as the throttle keyed it: put in the request's extensions for the
/// handlers that charge more ([`Throttle::admit_render`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Client {
/// The [`ClientKey`] of the request's address (IPv4, or the IPv6 /64).
pub key: String,
}
/// A refusal: how long to wait, and whether the client is blocked.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Refused {
pub retry_after: u64,
pub blocked: bool,
}
impl IntoResponse for Refused {
/// The `429`, in words, with when to come back. Never kept by anything in between.
fn into_response(self) -> Response {
let words = if self.blocked {
format!("Too many requests from your address. It is blocked for {} seconds.\n", self.retry_after)
} else {
format!("Too many requests. Please wait {} seconds and try again.\n", self.retry_after)
};
let mut r = (StatusCode::TOO_MANY_REQUESTS, words).into_response();
let h = r.headers_mut();
h.insert(header::RETRY_AFTER, HeaderValue::from(self.retry_after));
h.insert(header::CONTENT_TYPE, HeaderValue::from_static("text/plain; charset=utf-8"));
h.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
r
}
}
#[derive(Debug, Clone, Copy)]
struct Offence {
since: Instant,
count: u32,
blocked_until: Option<Instant>,
}
struct Inner {
guest: Buckets,
credentialed: Buckets,
account: Buckets,
token: Buckets,
global: Buckets,
render: Buckets,
render_all: Buckets,
flight: HashMap<(Kind, String), usize>,
global_flight: usize,
offences: HashMap<String, Offence>,
}
/// **The throttle.** One per router; shared by every route.
pub struct Throttle {
limits: Limits,
inner: Mutex<Inner>,
throttled: AtomicU64,
blocks: AtomicU64,
#[cfg(test)]
frozen: Mutex<Option<Instant>>,
}
/// Room held in flight; given back when dropped (when the handler has answered).
#[must_use]
pub struct Pass {
throttle: Arc<Throttle>,
keys: Vec<(Kind, String)>,
global: bool,
}
impl Drop for Pass {
fn drop(&mut self) {
let mut inner = self.throttle.lock();
for key in self.keys.drain(..) {
if let Some(n) = inner.flight.get_mut(&key) {
*n = n.saturating_sub(1);
if *n == 0 {
inner.flight.remove(&key);
}
}
}
if self.global {
inner.global_flight = inner.global_flight.saturating_sub(1);
}
}
}
const GLOBAL: &str = "";
impl Throttle {
pub fn new(limits: Limits) -> Arc<Throttle> {
let b = |budget: Budget| Buckets::with(f64::from(budget.burst), budget.per_second, limits.max_keys);
let inner = Inner {
guest: b(limits.guest.budget),
credentialed: b(limits.credentialed.budget),
account: b(limits.account.budget),
token: b(limits.token.budget),
global: Buckets::with(f64::from(limits.global.budget.burst), limits.global.budget.per_second, 1),
render: b(limits.render),
render_all: Buckets::with(f64::from(limits.render_all.burst), limits.render_all.per_second, 1),
flight: HashMap::new(),
global_flight: 0,
offences: HashMap::new(),
};
Arc::new(Throttle {
limits,
inner: Mutex::new(inner),
throttled: AtomicU64::new(0),
blocks: AtomicU64::new(0),
#[cfg(test)]
frozen: Mutex::new(None),
})
}
/// **The router's throttle**: [`Limits::from_env`], or (tests) the limits a test asked for.
pub fn for_router() -> Arc<Throttle> {
#[cfg(test)]
if let Some(limits) = tests::NEXT.with(|n| n.borrow_mut().take()) {
return Throttle::new(limits);
}
Throttle::new(Limits::from_env())
}
pub fn limits(&self) -> &Limits {
&self.limits
}
/// Requests answered 429 since the start.
pub fn throttled(&self) -> u64 {
self.throttled.load(Ordering::Relaxed)
}
/// Clients blocked since the start.
pub fn blocks(&self) -> u64 {
self.blocks.load(Ordering::Relaxed)
}
fn lock(&self) -> std::sync::MutexGuard<'_, Inner> {
self.inner.lock().unwrap_or_else(|p| p.into_inner())
}
fn now(&self) -> Instant {
#[cfg(test)]
if let Some(at) = *self.frozen.lock().unwrap_or_else(|p| p.into_inner()) {
return at;
}
Instant::now()
}
/// **One request at the door**: `client` (its [`ClientKey`]), counted as a guest or (with
/// a credential) at the credentialed budget, and `token`'s hash in its own tier when a
/// bearer came.
fn admit(self: &Arc<Self>, client: &str, credentialed: bool, token: Option<&str>) -> Result<Pass, Refused> {
let now = self.now();
let mut inner = self.lock();
if let Some(wait) = blocked(&mut inner, client, now) {
drop(inner);
self.throttled.fetch_add(1, Ordering::Relaxed);
return Err(Refused { retry_after: wait, blocked: true });
}
let (kind, tier) = if credentialed { (Kind::Credentialed, self.limits.credentialed) } else { (Kind::Guest, self.limits.guest) };
let mut wait: Option<u64> = None;
let mut own = false;
let mut worse = |w: u64, mine: bool, wait: &mut Option<u64>| {
*wait = Some(wait.map_or(w, |o| o.max(w)));
own |= mine;
};
if let Err(w) = inner.global.check(GLOBAL, now) {
worse(w, false, &mut wait);
}
if inner.global_flight >= self.limits.global.in_flight {
worse(1, false, &mut wait);
}
let ip = match kind {
Kind::Credentialed => inner.credentialed.check(client, now),
_ => inner.guest.check(client, now),
};
if let Err(w) = ip {
worse(w, true, &mut wait);
}
if flight(&inner, kind, client) >= tier.in_flight {
worse(1, true, &mut wait);
}
if let Some(t) = token {
if let Err(w) = inner.token.check(t, now) {
worse(w, true, &mut wait);
}
if flight(&inner, Kind::Token, t) >= self.limits.token.in_flight {
worse(1, true, &mut wait);
}
}
if let Some(retry_after) = wait {
// A refusal that is only the whole box's is nobody's offence.
let blocked_now = own && self.offend(&mut inner, client, now);
drop(inner);
self.throttled.fetch_add(1, Ordering::Relaxed);
let retry_after = if blocked_now { self.limits.block_for.as_secs().max(1) } else { retry_after };
return Err(Refused { retry_after, blocked: blocked_now });
}
inner.global.spend(GLOBAL);
inner.global_flight += 1;
let mut keys = vec![(kind, client.to_owned())];
match kind {
Kind::Credentialed => inner.credentialed.spend(client),
_ => inner.guest.spend(client),
}
if let Some(t) = token {
inner.token.spend(t);
keys.push((Kind::Token, t.to_owned()));
}
for k in &keys {
*inner.flight.entry(k.clone()).or_insert(0) += 1;
}
drop(inner);
Ok(Pass { throttle: self.clone(), keys, global: true })
}
/// **A signed-in person's own budget**, charged where the console knows who they are
/// (`require_auth`): per account, whatever address they ask from. `client` is the
/// request's address key, for the offence count.
pub fn admit_account(self: &Arc<Self>, account: &str, client: &str) -> Result<Pass, Refused> {
let now = self.now();
let key = format!("a:{account}");
let mut inner = self.lock();
let mut wait = inner.account.check(&key, now).err();
if flight(&inner, Kind::Account, &key) >= self.limits.account.in_flight {
wait = Some(wait.unwrap_or(1).max(1));
}
if let Some(retry_after) = wait {
let blocked_now = self.offend(&mut inner, client, now);
drop(inner);
self.throttled.fetch_add(1, Ordering::Relaxed);
let retry_after = if blocked_now { self.limits.block_for.as_secs().max(1) } else { retry_after };
return Err(Refused { retry_after, blocked: blocked_now });
}
inner.account.spend(&key);
*inner.flight.entry((Kind::Account, key.clone())).or_insert(0) += 1;
drop(inner);
Ok(Pass { throttle: self.clone(), keys: vec![(Kind::Account, key)], global: false })
}
/// **One on-demand render for `key`** (a [`Client::key`]): its own, lower budget, and one
/// for every client together. `Err` is the seconds to wait; the caller answers 429.
pub fn admit_render(&self, key: &str) -> Result<(), u64> {
let now = self.now();
let mut inner = self.lock();
let mine = inner.render.check(key, now);
let all = inner.render_all.check(GLOBAL, now);
match (mine, all) {
(Ok(()), Ok(())) => {
inner.render.spend(key);
inner.render_all.spend(GLOBAL);
Ok(())
}
(a, b) => {
drop(inner);
self.throttled.fetch_add(1, Ordering::Relaxed);
Err(a.err().unwrap_or(0).max(b.err().unwrap_or(0)).max(1))
}
}
}
/// Count one refusal of `client`; true when it blocks the client now.
fn offend(&self, inner: &mut Inner, client: &str, now: Instant) -> bool {
let l = &self.limits;
if inner.offences.len() >= l.max_keys && !inner.offences.contains_key(client) {
inner.offences.retain(|_, o| o.blocked_until.is_some_and(|t| t > now) || now.saturating_duration_since(o.since) < l.block_window);
if inner.offences.len() >= l.max_keys {
inner.offences.retain(|_, o| o.blocked_until.is_some_and(|t| t > now));
}
}
let o = inner.offences.entry(client.to_owned()).or_insert(Offence { since: now, count: 0, blocked_until: None });
if now.saturating_duration_since(o.since) >= l.block_window {
*o = Offence { since: now, count: 0, blocked_until: None };
}
o.count += 1;
if l.block_after > 0 && o.count >= l.block_after {
o.blocked_until = Some(now + l.block_for);
o.count = 0;
o.since = now;
self.blocks.fetch_add(1, Ordering::Relaxed);
tracing::warn!(
target: crate::auth::AUDIT_TARGET,
event = "throttle.blocked",
client = %client,
refusals = l.block_after,
within_secs = l.block_window.as_secs(),
blocked_secs = l.block_for.as_secs(),
"a client was refused too often and is blocked for a while"
);
return true;
}
false
}
/// Clients blocked right now.
pub fn blocked_now(&self) -> usize {
let now = self.now();
self.lock().offences.values().filter(|o| o.blocked_until.is_some_and(|t| t > now)).count()
}
}
fn flight(inner: &Inner, kind: Kind, key: &str) -> usize {
inner.flight.get(&(kind, key.to_owned())).copied().unwrap_or(0)
}
/// Seconds `client` is still blocked for, if it is.
fn blocked(inner: &mut Inner, client: &str, now: Instant) -> Option<u64> {
let o = inner.offences.get_mut(client)?;
match o.blocked_until {
Some(until) if until > now => Some(until.saturating_duration_since(now).as_secs().max(1)),
Some(_) => {
o.blocked_until = None;
None
}
None => None,
}
}
/// The request's [`ClientKey`], from the console's one client-address rule.
pub fn client_key(request: &Request) -> String {
let ip = crate::security::client_ip(request.headers(), request.extensions().get()).0;
ClientKey::of_peer(&ip).as_str().to_owned()
}
/// A bearer's key in the token tier: its hash, never the token.
fn token_key(headers: &HeaderMap) -> Option<String> {
let offered = headers.get(header::AUTHORIZATION)?.to_str().ok()?.strip_prefix("Bearer ")?.trim();
(!offered.is_empty()).then(|| format!("t:{}", &blake3::hash(offered.as_bytes()).to_hex()[..32]))
}
/// **The layer** ([`crate::serve::router`] mounts it around everything): a request with no
/// room is answered 429 here; one with room carries the throttle and its [`Client`] in its
/// extensions, and gives the room back when the inner service has answered.
pub async fn layer(State(throttle): State<Arc<Throttle>>, mut request: Request, next: Next) -> Response {
let client = client_key(&request);
let token = token_key(request.headers());
let credentialed = token.is_some() || crate::auth::session_cookie(request.headers()).is_some();
let pass = match throttle.admit(&client, credentialed, token.as_deref()) {
Ok(pass) => pass,
Err(refused) => return refused.into_response(),
};
request.extensions_mut().insert(throttle.clone());
request.extensions_mut().insert(Client { key: client });
let response = next.run(request).await;
drop(pass);
response
}
/// `GET /api/throttle` (an administrator's read): what the throttle refused and blocked.
pub async fn counters(axum::Extension(throttle): axum::Extension<Arc<Throttle>>) -> axum::Json<serde_json::Value> {
axum::Json(serde_json::json!({
"throttled": throttle.throttled(),
"blocks": throttle.blocks(),
"blocked_now": throttle.blocked_now(),
}))
}
#[cfg(test)]
pub(crate) mod tests;