Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 50 additions & 0 deletions crates/aisix-core/src/models/apikey.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,24 @@ pub struct ApiKey {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub rate_limit: Option<RateLimit>,

/// Team this API key belongs to. Used as a limiter bucket key
/// (`team:<id>`) for team-level rate limiting.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub team_id: Option<String>,

/// Rate limit inherited from the owning team.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub team_rate_limit: Option<RateLimit>,

/// Org member who owns this key. Used as a limiter bucket key
/// (`member:<id>`) for member-level rate limiting.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub owner_id: Option<String>,

/// Rate limit inherited from the owning member.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub owner_rate_limit: Option<RateLimit>,

/// etcd-key uuid; filled by the loader, never in the JSON payload.
#[serde(skip)]
pub(crate) runtime_id: String,
Expand Down Expand Up @@ -144,6 +162,10 @@ mod tests {
key_hash: "abc".into(),
allowed_models: vec![],
rate_limit: None,
team_id: None,
team_rate_limit: None,
owner_id: None,
owner_rate_limit: None,
runtime_id: String::new(),
};
assert!(!k.can_access("my-gpt4"));
Expand Down Expand Up @@ -207,4 +229,32 @@ mod tests {
// Resource::name now returns key_hash, not plaintext.
assert_eq!(k.name(), SAMPLE_HASH);
}

#[test]
fn deserialises_with_team_and_owner_fields() {
let k: ApiKey = serde_json::from_str(&format!(
r#"{{
"key_hash": "{SAMPLE_HASH}",
"allowed_models": ["gpt-4o"],
"team_id": "team-uuid-1",
"team_rate_limit": {{"rpm": 600}},
"owner_id": "member-uuid-1",
"owner_rate_limit": {{"tpm": 200000}}
}}"#
))
.unwrap();
assert_eq!(k.team_id.as_deref(), Some("team-uuid-1"));
assert_eq!(k.team_rate_limit.as_ref().unwrap().rpm, Some(600));
assert_eq!(k.owner_id.as_deref(), Some("member-uuid-1"));
assert_eq!(k.owner_rate_limit.as_ref().unwrap().tpm, Some(200000));
}

#[test]
fn absent_team_owner_fields_default_to_none() {
let k = sample();
assert!(k.team_id.is_none());
assert!(k.team_rate_limit.is_none());
assert!(k.owner_id.is_none());
assert!(k.owner_rate_limit.is_none());
}
}
10 changes: 4 additions & 6 deletions crates/aisix-proxy/src/audio.rs
Original file line number Diff line number Diff line change
Expand Up @@ -287,9 +287,8 @@ async fn multipart_dispatch(
return Err(ProxyError::ModelForbidden(model_name.clone()));
}

// Budget + rate-limit gate (issue #107). Audio transcriptions /
// translations bypassed both — Whisper customers ran unmetered.
let _reservation = crate::quota::enforce(state, auth).await?;
let model_rl = crate::quota::ModelRateLimit::from_model(&model_name, &model_entry.value);
let _reservation = crate::quota::enforce(state, auth, model_rl).await?;

let model = &model_entry.value;
let provider = crate::dispatch::require_provider(model)?;
Expand Down Expand Up @@ -388,9 +387,8 @@ async fn speech_dispatch(
return Err(ProxyError::ModelForbidden(model_name.clone()));
}

// Budget + rate-limit gate (issue #107). TTS/speech bypassed
// both pre-fix.
let _reservation = crate::quota::enforce(state, auth).await?;
let model_rl = crate::quota::ModelRateLimit::from_model(&model_name, &model_entry.value);
let _reservation = crate::quota::enforce(state, auth, model_rl).await?;

let model = &model_entry.value;
let provider = crate::dispatch::require_provider(model)?;
Expand Down
28 changes: 12 additions & 16 deletions crates/aisix-proxy/src/chat.rs
Original file line number Diff line number Diff line change
Expand Up @@ -460,12 +460,10 @@ async fn dispatch(
}
}

let rl_key = auth.entry.id.clone();
let rl_limits = auth.key().rate_limit.clone().unwrap_or_default();
let reservation = state
.limiter
.pre_commit(&rl_key, &rl_limits)
.map_err(|e| with_model(ProxyError::from(e)))?;
// Multi-layer rate-limit reservation (api_key + model + team + member).
let model_rl = crate::quota::ModelRateLimit::from_model(&req.model, &virtual_entry.value);
let reservation =
crate::quota::enforce_rate_limit(state, auth, model_rl).map_err(&with_model)?;

let now = created_ts();

Expand All @@ -488,13 +486,10 @@ async fn dispatch(
.chat_stream(req, &ctx)
.await
.map_err(|e| with_model(ProxyError::Bridge(e)))?;
// Drop the reservation now: concurrency releases (the SSE
// stream that follows is driven by the client, not by the
// proxy holding open an upstream-bound future), and RPM was
// already counted by pre_commit. TPM is updated retroactively
// on stream-end by `add_tokens_post_stream` — see issue #108.
// Pre-fix this path called commit_tokens(0) and never came
// back, leaving TPM caps blind for all streaming traffic.
// Drop the reservation now: concurrency releases on all layers.
// RPM was already counted by pre_commit. TPM is updated
// retroactively on stream-end by `add_tokens_post_stream`.
let post_stream_keys = reservation.keys();
drop(reservation);
// Capture everything the stream-completion callback needs so
// it can fire `emit_usage_event` once the terminal SSE chunk
Expand All @@ -503,7 +498,6 @@ async fn dispatch(
// populate `usage` on the last chunk; emitting at handler
// return (the non-streaming path's spot) would record zeros.
let limiter = Arc::clone(&state.limiter);
let post_stream_key = rl_key.clone();
let state_for_telem = state.clone();
let request_id_for_telem = request_id.to_string();
let model_id_for_telem = model_id.clone();
Expand Down Expand Up @@ -534,8 +528,10 @@ async fn dispatch(
now,
stream_guardrail,
move |comp: StreamCompletion| {
// Existing: rate-limit accounting (TPM cap) — see #108.
limiter.add_tokens_post_stream(&post_stream_key, comp.total_tokens);
// Rate-limit accounting (TPM cap) for all layers.
for key in &post_stream_keys {
limiter.add_tokens_post_stream(key, comp.total_tokens);
}
// Telemetry: emit with the actual upstream-reported counts.
// cost_usd stays 0.0; cp-api recomputes server-side from
// its model_pricing catalog (same pattern as the non-
Expand Down
7 changes: 2 additions & 5 deletions crates/aisix-proxy/src/completions.rs
Original file line number Diff line number Diff line change
Expand Up @@ -108,11 +108,8 @@ async fn dispatch(
return Err(ProxyError::ModelForbidden(model_name.to_string()));
}

// Budget + rate-limit gate (issue #107). Reservation drops at
// end of dispatch — RPM counts on pre_commit, concurrency
// releases on drop, TPM is left at 0 (this handler doesn't
// surface a uniform token count today).
let _reservation = crate::quota::enforce(state, auth).await?;
let model_rl = crate::quota::ModelRateLimit::from_model(model_name, &model_entry.value);
let _reservation = crate::quota::enforce(state, auth, model_rl).await?;

let model = &model_entry.value;
let provider = crate::dispatch::require_provider(model)?;
Expand Down
9 changes: 2 additions & 7 deletions crates/aisix-proxy/src/embeddings.rs
Original file line number Diff line number Diff line change
Expand Up @@ -141,13 +141,8 @@ async fn dispatch(
.get(provider)
.ok_or(ProxyError::ProviderUnavailable)?;

// Budget + rate-limit gate (issue #107). Pre-fix this endpoint
// bypassed both. The reservation is held until commit_tokens at
// the end of dispatch — embeddings don't surface a stable token
// count across providers, so we commit 0 for now (RPM counts,
// TPM doesn't). Plumbing per-provider token totals through is a
// follow-up.
let reservation = crate::quota::enforce(state, auth).await?;
let model_rl = crate::quota::ModelRateLimit::from_model(&body.model, &model_entry.value);
let reservation = crate::quota::enforce(state, auth, model_rl).await?;

let upstream_model_id = crate::dispatch::require_upstream_model(model)?.to_string();

Expand Down
4 changes: 2 additions & 2 deletions crates/aisix-proxy/src/images.rs
Original file line number Diff line number Diff line change
Expand Up @@ -104,8 +104,8 @@ async fn dispatch(
return Err(ProxyError::ModelForbidden(model_name.to_string()));
}

// Budget + rate-limit gate (issue #107).
let _reservation = crate::quota::enforce(state, auth).await?;
let model_rl = crate::quota::ModelRateLimit::from_model(model_name, &model_entry.value);
let _reservation = crate::quota::enforce(state, auth, model_rl).await?;

let model = &model_entry.value;

Expand Down
10 changes: 2 additions & 8 deletions crates/aisix-proxy/src/messages.rs
Original file line number Diff line number Diff line change
Expand Up @@ -159,14 +159,8 @@ async fn dispatch(
return Err(ProxyError::ModelForbidden(model_name.clone()));
}

// Budget + rate-limit gate (issue #107). Pre-fix this endpoint
// bypassed both — Anthropic-API customers ran unmetered. Held
// until end of dispatch via Drop, which releases the concurrency
// permit. RPM is recorded immediately on pre_commit; TPM stays
// 0 for now (the streaming + cross-provider paths emit token
// counts in different shapes — plumbing them uniformly is a
// follow-up).
let _reservation = crate::quota::enforce(state, auth).await?;
let model_rl = crate::quota::ModelRateLimit::from_model(&model_name, &model_entry.value);
let _reservation = crate::quota::enforce(state, auth, model_rl).await?;

let model = &model_entry.value;
let pk_entry = crate::dispatch::resolve_provider_key(&snapshot, model)?;
Expand Down
8 changes: 1 addition & 7 deletions crates/aisix-proxy/src/passthrough.rs
Original file line number Diff line number Diff line change
Expand Up @@ -162,13 +162,7 @@ async fn dispatch(
req: Request,
request_id: &str,
) -> Result<(Response, String), ProxyError> {
// Budget + rate-limit gate (issue #107). The previous _auth
// binding ignored the AuthenticatedKey entirely — passthrough
// ran completely unmetered, with no per-key budget cap and no
// RPM/TPM limit. This was the most exploitable gap because the
// /passthrough/* family covers everything OpenAI ships *plus*
// every provider's own API. Held for the dispatch lifetime.
let _reservation = crate::quota::enforce(&state, auth).await?;
let _reservation = crate::quota::enforce(&state, auth, None).await?;
let snapshot = state.snapshot.load();

// Find a model for this provider so we can borrow its provider_key.
Expand Down
151 changes: 106 additions & 45 deletions crates/aisix-proxy/src/quota.rs
Original file line number Diff line number Diff line change
@@ -1,65 +1,126 @@
//! Pre-dispatch quota gate shared by every LLM endpoint.
//!
//! Before this gate landed, only `/v1/chat/completions` ran budget +
//! rate-limit checks (`chat::dispatch`). Every other LLM endpoint —
//! `/v1/embeddings`, `/v1/messages`, `/v1/audio/*`,
//! `/v1/images/generations`, `/v1/responses`, `/v1/rerank`,
//! `/v1/completions`, the `/passthrough/...` family — went straight
//! from auth into the upstream Bridge, silently bypassing both. A
//! customer running the gateway as their org's LLM proxy expected
//! RPM/TPM caps and budget cutoffs to apply uniformly across
//! endpoints; the gap was visible to anyone using `/v1/messages`
//! (Anthropic API-shape) or `/v1/embeddings`. See issue #107.
//! Applies budget + multi-layer rate limiting:
//! 1. Budget pre-check (cp-api cached decision)
//! 2. API-key rate limit (`auth.entry.id`)
//! 3. Model rate limit (`model:<name>`) — when the resolved Model has one
//! 4. Team rate limit (`team:<id>`) — when the ApiKey carries team info
//! 5. Member rate limit (`member:<id>`) — when the ApiKey carries owner info
//!
//! This module hosts the minimum check every non-chat handler now
//! performs: a budget pre-check via cp-api, then a rate-limit
//! reservation. Guardrails are *not* applied here — they need
//! per-handler text extraction (chat reads messages, embeddings reads
//! `input` strings, audio reads transcripts) and that wiring is a
//! larger, separate change. The chat handler still has its own
//! guardrail path; this gate runs in parallel for every other
//! endpoint.
//!
//! Returning a [`Reservation`] (not just a permit) lets the caller
//! commit token usage post-dispatch. Non-chat handlers that don't
//! track upstream tokens uniformly call
//! [`aisix_ratelimit::Reservation::commit_tokens`] with `0`, which
//! still releases the concurrency permit and counts the request
//! against RPM but skips TPM. Future work can plumb per-endpoint
//! token totals through.
//! All layers use AND logic — every layer must pass or the request gets
//! 429. The returned [`MultiReservation`] commits token usage to all
//! layers and releases all concurrency permits on drop.

use aisix_ratelimit::Reservation;
use aisix_core::RateLimit;
use aisix_ratelimit::MultiReservation;

use crate::auth::AuthenticatedKey;
use crate::error::ProxyError;
use crate::state::ProxyState;

/// Apply budget + rate-limit checks for one request. Call this before
/// touching the Bridge in every LLM endpoint handler. The returned
/// [`Reservation`] is alive until the caller commits or drops it; on
/// commit, RPM is finalised and TPM accounted for the supplied total.
/// Optional model rate-limit info resolved by the caller before enforce.
pub(crate) struct ModelRateLimit {
pub name: String,
pub limits: RateLimit,
}

impl ModelRateLimit {
/// Build from a resolved model entry. Returns `None` when the model
/// has no rate limit configured or has an unrestricted one (all fields
/// are `None`).
pub fn from_model(model_name: &str, model: &aisix_core::Model) -> Option<Self> {
model
.rate_limit
.as_ref()
.filter(|rl| !rl.is_unrestricted())
.map(|rl| Self {
Comment thread
jarvis9443 marked this conversation as resolved.
name: model_name.to_owned(),
limits: rl.clone(),
})
}
}

/// Reserve across all applicable rate-limit layers (api_key, model, team, member).
fn reserve_layers<'a>(
state: &'a ProxyState,
auth: &AuthenticatedKey,
model_rl: Option<ModelRateLimit>,
) -> Result<MultiReservation<'a, aisix_ratelimit::SystemClock>, ProxyError> {
let mut reservations = Vec::with_capacity(4);

// Layer 1: API key rate limit.
let key_limits = auth.key().rate_limit.clone().unwrap_or_default();
if !key_limits.is_unrestricted() {
let r = state
.limiter
.pre_commit(&auth.entry.id, &key_limits)
.map_err(ProxyError::from)?;
reservations.push(r);
Comment thread
jarvis9443 marked this conversation as resolved.
}

// Layer 2: Model rate limit.
if let Some(mrl) = model_rl {
if !mrl.limits.is_unrestricted() {
let key = format!("model:{}", mrl.name);
let r = state
.limiter
.pre_commit(&key, &mrl.limits)
.map_err(ProxyError::from)?;
reservations.push(r);
}
}

// Layer 3: Team rate limit.
if let (Some(tid), Some(trl)) = (&auth.key().team_id, &auth.key().team_rate_limit) {
if !tid.is_empty() && !trl.is_unrestricted() {
let key = format!("team:{tid}");
let r = state
.limiter
.pre_commit(&key, trl)
.map_err(ProxyError::from)?;
reservations.push(r);
}
}

// Layer 4: Member/owner rate limit.
if let (Some(oid), Some(orl)) = (&auth.key().owner_id, &auth.key().owner_rate_limit) {
if !oid.is_empty() && !orl.is_unrestricted() {
let key = format!("member:{oid}");
let r = state
.limiter
.pre_commit(&key, orl)
.map_err(ProxyError::from)?;
reservations.push(r);
}
}

Ok(MultiReservation::new(reservations))
}

/// Apply budget + multi-layer rate-limit checks for one request.
/// `model_rl` is the resolved Model's rate_limit (if any). Pass `None`
/// for endpoints that don't resolve a model (e.g. passthrough).
pub(crate) async fn enforce<'a>(
state: &'a ProxyState,
auth: &AuthenticatedKey,
) -> Result<Reservation<'a, aisix_ratelimit::SystemClock>, ProxyError> {
// Budget pre-check via cp-api. Mirrors chat::dispatch — the DP no
// longer owns budget state; cp-api returns a cached/live decision
// per api_key.
model_rl: Option<ModelRateLimit>,
) -> Result<MultiReservation<'a, aisix_ratelimit::SystemClock>, ProxyError> {
let decision = state.budgets.check(&auth.entry.id).await;
if !decision.allowed {
return Err(ProxyError::BudgetExceeded(
decision.reason.unwrap_or_else(|| auth.entry.id.clone()),
));
}

// Rate-limit reservation. The reservation holds a concurrency
// permit until it's committed (or dropped). Commit at the end of
// dispatch with whatever token count the upstream returned (0 if
// the handler doesn't track tokens — RPM still counts).
let rl_key = auth.entry.id.clone();
let rl_limits = auth.key().rate_limit.clone().unwrap_or_default();
state
.limiter
.pre_commit(&rl_key, &rl_limits)
.map_err(ProxyError::from)
reserve_layers(state, auth, model_rl)
}

/// Rate-limit-only enforcement (no budget check). Used by `chat.rs`
/// which handles budget separately.
pub(crate) fn enforce_rate_limit<'a>(
state: &'a ProxyState,
auth: &AuthenticatedKey,
model_rl: Option<ModelRateLimit>,
) -> Result<MultiReservation<'a, aisix_ratelimit::SystemClock>, ProxyError> {
reserve_layers(state, auth, model_rl)
}
Loading
Loading