diff --git a/crates/aisix-obs/src/usage.rs b/crates/aisix-obs/src/usage.rs index 1d89aa74..4e2ec60b 100644 --- a/crates/aisix-obs/src/usage.rs +++ b/crates/aisix-obs/src/usage.rs @@ -247,6 +247,17 @@ pub struct UsageEvent { pub cost_usd: f64, /// True when a guardrail rejected the request (input or output). + /// + /// cp-api indexes this and the dashboard's Logs "Guardrail blocks" + /// view is the exact predicate `guardrail_blocked = true`, so it is + /// the ONLY thing that puts a refusal in front of an operator — a + /// refused request whose event leaves the field at its `false` + /// default is still in the unfiltered feed, which makes the empty + /// Blocked view read as "no guardrail activity" rather than as a + /// missing row (AISIX-Cloud#1428). Every emitter on a failure path + /// must therefore set it from `ProxyError::is_guardrail_block`, + /// including a stream refused after its 200 head went out: there the + /// status stays 200 and this bool is the whole record of the block. pub guardrail_blocked: bool, /// Set when at least one remote-API guardrail (today: kind=bedrock) diff --git a/crates/aisix-proxy/src/a2a.rs b/crates/aisix-proxy/src/a2a.rs index cb1d0505..7b7e0923 100644 --- a/crates/aisix-proxy/src/a2a.rs +++ b/crates/aisix-proxy/src/a2a.rs @@ -323,6 +323,9 @@ async fn dispatch( Duration::ZERO, trace.as_ref(), /* dispatched */ false, + // A quota refusal is not a guardrail decision. + /* guardrail_blocked */ + false, ); return response; } @@ -370,6 +373,7 @@ async fn dispatch( latency, trace.as_ref(), /* dispatched */ true, + /* guardrail_blocked */ false, ); axum::Json(response_value).into_response() } @@ -387,6 +391,7 @@ async fn dispatch( latency, trace.as_ref(), /* dispatched */ true, + /* guardrail_blocked */ false, ); a2a_error_response(rpc_id, status, &err.to_string()) } @@ -457,6 +462,7 @@ impl Drop for StreamUsageOnDrop { self.started.elapsed(), self.trace.as_ref(), /* dispatched */ true, + /* guardrail_blocked */ false, ); } } @@ -501,6 +507,7 @@ async fn dispatch_stream( started.elapsed(), trace.as_ref(), /* dispatched */ true, + /* guardrail_blocked */ false, ); return a2a_error_response(rpc_id, status, &err.to_string()); } @@ -781,6 +788,8 @@ async fn guardrail_block_response( Duration::ZERO, trace, /* dispatched */ false, + // This IS the guardrail refusal. + /* guardrail_blocked */ true, ); Some(response) } @@ -833,6 +842,10 @@ fn emit_a2a_usage( // Whether the call reached the upstream agent — false for a quota // rejection, which refuses before any upstream contact. dispatched: bool, + // Whether a guardrail refused the call. `/a2a` emits exactly one event + // per call, so this row is the only place a refusal can appear to the + // dashboard's "Guardrail blocks" view (AISIX-Cloud#1428). + guardrail_blocked: bool, ) { // No model resolves on this endpoint, so the estimator falls back to its // default encoding — the same thing it does for any non-OpenAI model. @@ -873,6 +886,7 @@ fn emit_a2a_usage( .ttfb .map(|d| d.as_millis().min(u32::MAX as u128) as u32) .unwrap_or_default(), + guardrail_blocked, ..Default::default() }; crate::usage_attr::apply_jwt_identity(&mut event, auth.jwt.as_ref()); diff --git a/crates/aisix-proxy/src/audio.rs b/crates/aisix-proxy/src/audio.rs index abaf9620..f6a2ee12 100644 --- a/crates/aisix-proxy/src/audio.rs +++ b/crates/aisix-proxy/src/audio.rs @@ -241,6 +241,7 @@ pub async fn transcriptions( &api_key_id, status, err.kind(), + err.is_guardrail_block(), &client, crate::usage_attr::enforced_hits(&audit), ); @@ -400,6 +401,7 @@ pub async fn translations( &api_key_id, status, err.kind(), + err.is_guardrail_block(), &client, crate::usage_attr::enforced_hits(&audit), ); @@ -568,6 +570,7 @@ pub async fn speech( &api_key_id, status, err.kind(), + err.is_guardrail_block(), &client, crate::usage_attr::enforced_hits(&audit), ); diff --git a/crates/aisix-proxy/src/chat.rs b/crates/aisix-proxy/src/chat.rs index b1187334..416a463d 100644 --- a/crates/aisix-proxy/src/chat.rs +++ b/crates/aisix-proxy/src/chat.rs @@ -459,8 +459,11 @@ pub async fn chat_completions( // `req.model` resolves against the snapshot, so a guardrail / // budget / rate-limit / bridge error after that point still // records which model the request targeted. ContentFiltered - // (guardrail) sets `guardrail_blocked` for the Blocked tab. - let guardrail_blocked = matches!(err, ProxyError::ContentFiltered { .. }); + // (guardrail) sets `guardrail_blocked` for the Blocked tab — + // through the shared predicate every handler now reads, so the + // family cannot answer this question two different ways + // (AISIX-Cloud#1428). + let guardrail_blocked = err.is_guardrail_block(); let model_id_str = resolved_model_id.as_deref().unwrap_or(""); // AISIX-Cloud#1013: failed requests carry the (post-mask) // request body so a 4xx/5xx can be triaged from the log alone. diff --git a/crates/aisix-proxy/src/completions.rs b/crates/aisix-proxy/src/completions.rs index 0bcd938a..add774fc 100644 --- a/crates/aisix-proxy/src/completions.rs +++ b/crates/aisix-proxy/src/completions.rs @@ -257,6 +257,7 @@ pub async fn completions( &api_key_id, status, err.kind(), + err.is_guardrail_block(), &client, crate::usage_attr::enforced_hits(&audit), ); diff --git a/crates/aisix-proxy/src/embeddings.rs b/crates/aisix-proxy/src/embeddings.rs index 5996916d..e89f29fd 100644 --- a/crates/aisix-proxy/src/embeddings.rs +++ b/crates/aisix-proxy/src/embeddings.rs @@ -249,6 +249,7 @@ pub async fn embeddings( &api_key_id, status, err.kind(), + err.is_guardrail_block(), &client, crate::usage_attr::enforced_hits(&audit), ); diff --git a/crates/aisix-proxy/src/error.rs b/crates/aisix-proxy/src/error.rs index a9e32a0c..3cf82171 100644 --- a/crates/aisix-proxy/src/error.rs +++ b/crates/aisix-proxy/src/error.rs @@ -490,6 +490,28 @@ impl ProxyError { } } + /// Whether this failure IS a guardrail refusal — the request was + /// stopped by the guardrail machinery rather than by an upstream, a + /// quota, a credential or a malformed body. + /// + /// Drives `UsageEvent::guardrail_blocked`, which is the indexed column + /// the dashboard's "Guardrail blocks" view and the + /// `guardrail_blocked=true` usage query filter on. Every failure-path + /// emitter reads it from here rather than re-deriving the match, so a + /// new handler cannot join the family with the flag silently left at + /// its `false` default (AISIX-Cloud#1428). + /// + /// A fail-closed refusal (`unavailable: Some(_)` — the guardrail could + /// not evaluate the request and its row refuses what it cannot check) + /// counts too: the guardrail machinery is still what stopped the + /// request, and hiding it from the Blocked view would leave an operator + /// with a 422 that nothing accounts for. Which of the two it was stays + /// legible on `guardrail_enforced_hits.action` + /// (`blocked` vs `blocked_unavailable`). + pub(crate) fn is_guardrail_block(&self) -> bool { + matches!(self, ProxyError::ContentFiltered { .. }) + } + /// Seconds the client should wait before retrying. Only present for /// rate-limit-style rejections so the proxy can emit a `Retry-After` /// header. diff --git a/crates/aisix-proxy/src/guardrail_blocked_telemetry.rs b/crates/aisix-proxy/src/guardrail_blocked_telemetry.rs new file mode 100644 index 00000000..460b24bc --- /dev/null +++ b/crates/aisix-proxy/src/guardrail_blocked_telemetry.rs @@ -0,0 +1,431 @@ +//! Cross-handler guard for one invariant: when a guardrail refuses a +//! request, the request's terminal `UsageEvent` says so. +//! +//! `UsageEvent::guardrail_blocked` is not decorative. cp-api indexes it +//! (`idx_dpmgr_usage_events_env_blocked`) and the dashboard's Logs +//! "Guardrail blocks" view is the exact predicate `guardrail_blocked = +//! true`, so a refusal emitted without it is a 422 the caller definitely +//! saw and the operator cannot find. The failure mode is worse than a +//! missing row: the request IS in the unfiltered feed, so the Blocked view +//! coming back empty reads as "the gateway logged no guardrail activity", +//! which is how AISIX-Cloud#1428 was reported. +//! +//! The flag was set on `/v1/chat/completions` and `/mcp` and nowhere else. +//! Every other handler builds its failure event through a different +//! emitter — `usage_attr::build_error_usage_event` for the single-attempt +//! family, `responses::emit_zero_token_event` and +//! `messages::emit_anthropic_usage_event` for the two retrying ones — and +//! each left the field at its `false` default. So this file drives the +//! surfaces themselves rather than any one emitter: an emitter test would +//! have passed for chat while nine siblings were wrong. +//! +//! Each surface is driven twice against the same keyword guardrail — once +//! with the blocking literal in the field a caller writes, once without. +//! The second run is what makes the first mean anything: these fixtures +//! point at a dead upstream, so a clean request fails too, and a flag that +//! merely tracked "the request failed" would pass the blocked run and fail +//! the clean one. + +use std::sync::Arc; + +use aisix_core::snapshot::SnapshotHandle; +use aisix_core::{AisixSnapshot, ApiKey, ProxyConfig, ResourceEntry}; +use aisix_obs::{UsageEvent, UsageSink}; +use axum::body::Body; +use axum::http::Request; +use tower::ServiceExt; + +/// The literal the guardrail row below refuses. +const BLOCK: &str = "BLOCKME"; +const CALLER: &str = "sk-caller"; +/// SHA-256 of `CALLER`. +const CALLER_HASH: &str = "8b6712790a2089c67aa97a2d80022df18cc65c7814350e33baebe79aab508891"; +const PK_ID: &str = "11111111-1111-1111-1111-111111111111"; +const ANTHROPIC_PK_ID: &str = "22222222-2222-2222-2222-222222222222"; + +/// A port nothing listens on, so a request that gets past the guardrail +/// gate fails at the network instead of hanging. That failure is the point +/// of the clean run: it produces an error event whose flag must stay false. +const DEAD_UPSTREAM: &str = "http://127.0.0.1:1"; + +fn cfg() -> ProxyConfig { + ProxyConfig { + addr: "127.0.0.1:0".into(), + request_body_limit_bytes: 1_048_576, + real_ip: Default::default(), + request_id: Default::default(), + url_rewrites: Vec::new(), + tls: None, + thread_per_core: None, + workers: None, + } +} + +fn snapshot() -> AisixSnapshot { + let snap = AisixSnapshot::new(); + + let key: ApiKey = serde_json::from_value(serde_json::json!({ + "key_hash": CALLER_HASH, + "allowed_models": ["*"], + "allowed_routes": ["*"], + "allowed_agents": ["*"], + })) + .expect("valid api key"); + snap.apikeys.insert(ResourceEntry::new("ak-1", key, 1)); + + for (id, name, provider, adapter, base) in [ + ( + PK_ID, + "openai-up", + "openai", + "openai", + format!("{DEAD_UPSTREAM}/v1"), + ), + ( + ANTHROPIC_PK_ID, + "anthropic-up", + "anthropic", + "anthropic", + DEAD_UPSTREAM.to_string(), + ), + ] { + let pk: aisix_core::ProviderKey = serde_json::from_value(serde_json::json!({ + "display_name": name, + "secret": "sk-unused", + "api_base": base, + "provider": provider, + "adapter": adapter, + })) + .expect("valid provider key"); + snap.provider_keys.insert(ResourceEntry::new(id, pk, 1)); + } + + for (id, name, provider, model_name, pk_id, kind) in [ + ( + "m-openai", + "gpt-under-test", + "openai", + "gpt-4o", + PK_ID, + None, + ), + ( + "m-anthropic", + "claude-under-test", + "anthropic", + "claude-3-haiku-20240307", + ANTHROPIC_PK_ID, + None, + ), + ( + "m-embedding", + "embed-under-test", + "openai", + "text-embedding-3-small", + PK_ID, + Some("embedding"), + ), + ] { + let mut value = serde_json::json!({ + "display_name": name, + "provider": provider, + "model_name": model_name, + "provider_key_id": pk_id, + }); + if let Some(kind) = kind { + value["kind"] = serde_json::Value::String(kind.to_string()); + } + let model: aisix_core::Model = serde_json::from_value(value).expect("valid model"); + snap.models.insert(ResourceEntry::new(id, model, 1)); + } + + let agent: aisix_core::A2aAgent = serde_json::from_value(serde_json::json!({ + "name": "agent-under-test", + "url": format!("{DEAD_UPSTREAM}/a2a"), + "enabled": true, + })) + .expect("valid a2a agent"); + snap.a2a_agents + .insert(ResourceEntry::new("agent-1", agent, 1)); + + let route: aisix_core::PassthroughRoute = serde_json::from_value(serde_json::json!({ + "name": "byo-tunnel", + "path_prefix": "/passthrough/byo", + "target_url": DEAD_UPSTREAM, + "provider_key_id": PK_ID, + })) + .expect("valid passthrough route"); + snap.passthrough_routes + .insert(ResourceEntry::new("route-1", route, 1)); + + // Env-scoped (no attachment row → the backward-compat fallback applies + // it to every request), input hook, fail-closed. A keyword row is the + // deterministic stand-in for any input-hook kind and, unlike a + // text-independent script, keeps the clean run genuinely clean. + let guardrail: aisix_core::Guardrail = serde_json::from_value(serde_json::json!({ + "name": "block-literal", + "enabled": true, + "kind": "keyword", + "hook_point": "input", + "fail_open": false, + "patterns": [{ "kind": "literal", "value": BLOCK }], + })) + .expect("valid guardrail"); + snap.guardrails + .insert(ResourceEntry::new("g-1", guardrail, 1)); + + snap +} + +/// Build the router plus the receiver its usage events land in. +fn router() -> (axum::Router, tokio::sync::mpsc::Receiver) { + let hub = Arc::new(aisix_gateway::Hub::new()); + hub.register_specialized( + "openai", + Arc::new(aisix_provider_openai::OpenAiBridge::new()), + ); + hub.register_specialized( + "anthropic", + Arc::new(aisix_provider_anthropic::AnthropicBridge::new()), + ); + let (tx, rx) = tokio::sync::mpsc::channel(32); + let state = crate::ProxyState::new(SnapshotHandle::new(snapshot()), hub, &cfg()) + .without_cache() + .with_usage_sink(UsageSink::new(tx)); + (crate::build_router(state), rx) +} + +/// One driveable request per surface, with `text` in the field a caller +/// authors — the field an input guardrail screens. +fn fixture(surface: &str, text: &str) -> Request { + let json = |uri: &str, body: serde_json::Value| { + Request::builder() + .method("POST") + .uri(uri.to_string()) + .header("authorization", format!("Bearer {CALLER}")) + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap() + }; + match surface { + "/v1/chat/completions" => json( + surface, + serde_json::json!({ + "model": "gpt-under-test", + "messages": [{ "role": "user", "content": text }], + }), + ), + "/v1/completions" => json( + surface, + serde_json::json!({ "model": "gpt-under-test", "prompt": text }), + ), + "/v1/responses" => json( + surface, + serde_json::json!({ "model": "gpt-under-test", "input": text }), + ), + "/v1/messages" => Request::builder() + .method("POST") + .uri(surface) + .header("x-api-key", CALLER) + .header("anthropic-version", "2023-06-01") + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({ + "model": "claude-under-test", + "max_tokens": 16, + "messages": [{ "role": "user", "content": text }], + }) + .to_string(), + )) + .unwrap(), + "/v1/embeddings" => json( + surface, + serde_json::json!({ "model": "embed-under-test", "input": text }), + ), + "/v1/rerank" => json( + surface, + serde_json::json!({ + "model": "gpt-under-test", + "query": text, + "documents": ["a document"], + }), + ), + "/v1/images/generations" => json( + surface, + serde_json::json!({ "model": "gpt-under-test", "prompt": text }), + ), + "/v1/audio/speech" => json( + surface, + serde_json::json!({ "model": "gpt-under-test", "input": text, "voice": "alloy" }), + ), + "/v1/videos" => json( + surface, + serde_json::json!({ "model": "gpt-under-test", "prompt": text }), + ), + // JSON-RPC rather than an OpenAI envelope: the screened text is + // `params.message`, the only caller-authored content A2A carries. + "/a2a" => Request::builder() + .method("POST") + .uri("/a2a/agent-under-test") + .header("authorization", format!("Bearer {CALLER}")) + .header("content-type", "application/json") + .header("accept", "application/json, text/event-stream") + .body(Body::from( + serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "message/send", + "params": { + "message": { + "role": "user", + "parts": [{ "kind": "text", "text": text }], + "messageId": "m-1", + }, + }, + }) + .to_string(), + )) + .unwrap(), + // The passthrough tunnel forwards the body verbatim, so its + // screened text is whatever the detected envelope carries. + "/passthrough/byo" => json( + "/passthrough/byo/v1/chat/completions", + serde_json::json!({ + "model": "gpt-4o", + "messages": [{ "role": "user", "content": text }], + }), + ), + other => panic!("no fixture for {other}"), + } +} + +/// Every surface whose input hook a keyword row can reach today, i.e. every +/// one that carries caller-authored text in a JSON field. +/// +/// The multipart surfaces (`/v1/audio/transcriptions`, `/v1/images/edits`) +/// and the blob-scanning job surfaces are deliberately absent: their +/// screened text is an uploaded file, so a keyword fixture would assert +/// nothing they don't already share with `/v1/audio/speech` — all four go +/// through the same `usage_attr::build_error_usage_event` emitter this +/// list already covers three times over. +const SURFACES: &[&str] = &[ + "/v1/chat/completions", + "/v1/completions", + "/v1/responses", + "/v1/messages", + "/v1/embeddings", + "/v1/rerank", + "/v1/images/generations", + "/v1/audio/speech", + "/v1/videos", + "/passthrough/byo", + "/a2a", +]; + +/// Drive one fixture and collect every usage event it emitted. +/// +/// Events are drained on a short timeout rather than counted: the number a +/// surface emits is its own business (a retrying family emits one per +/// failed attempt), and pinning it here would make this file fail for +/// reasons that have nothing to do with the flag. +async fn drive(surface: &str, text: &str) -> (u16, Vec) { + let (router, mut rx) = router(); + let response = router + .oneshot(fixture(surface, text)) + .await + .expect("router must answer"); + let status = response.status().as_u16(); + // Drain the body: a streaming surface emits from its end-of-stream + // guard, which only runs once the body is polled to completion. + let _ = axum::body::to_bytes(response.into_body(), 1 << 20).await; + + let mut events = Vec::new(); + while let Ok(Some(event)) = + tokio::time::timeout(std::time::Duration::from_millis(300), rx.recv()).await + { + events.push(event); + } + (status, events) +} + +#[tokio::test] +async fn a_guardrail_refusal_is_marked_on_every_surface() { + let mut wrong = Vec::new(); + for surface in SURFACES { + let (status, events) = drive(surface, &format!("please {BLOCK} now")).await; + if status != 422 { + wrong.push(format!("{surface}: refused with {status}, expected 422")); + continue; + } + if events.is_empty() { + wrong.push(format!("{surface}: refused but emitted no usage event")); + continue; + } + // The refusal is request-scoped, so it rides the request's terminal + // event. These fixtures refuse before any upstream is contacted, so + // that event is the only one. + if !events.iter().any(|e| e.guardrail_blocked) { + wrong.push(format!( + "{surface}: emitted {} usage event(s), none marked guardrail_blocked", + events.len() + )); + continue; + } + // A refusal costs the caller nothing: no upstream ran. `/a2a` is + // exempt because its counters are the gateway's own reading of the + // words, flagged `usage_estimated` and never charged — they are + // filled from the request before the chain even runs. + if *surface != "/a2a" { + for event in &events { + if event.prompt_tokens != 0 || event.completion_tokens != 0 { + wrong.push(format!( + "{surface}: refused request billed {}+{} tokens", + event.prompt_tokens, event.completion_tokens + )); + } + } + } + } + assert!( + wrong.is_empty(), + "a guardrail refusal must reach the Logs \"Guardrail blocks\" view \ + (usage_events.guardrail_blocked = true) on every surface:\n {}", + wrong.join("\n "), + ); +} + +#[tokio::test] +async fn an_ordinary_failure_is_not_marked_as_a_guardrail_block() { + let mut wrong = Vec::new(); + for surface in SURFACES { + // Same guardrail, same fixtures, text it does not match — so the + // request runs on and dies at the dead upstream instead. + let (status, events) = drive(surface, "a perfectly ordinary question").await; + if status == 422 { + wrong.push(format!("{surface}: clean text was refused ({status})")); + continue; + } + // Without this the control is vacuous: a fixture that stopped + // before the handler ran — a stale key hash, a rejected body, a + // missing model row — emits nothing, and "no event is marked" is + // trivially satisfied by having no event. + if events.is_empty() { + wrong.push(format!( + "{surface}: clean run emitted no usage event, so nothing was checked" + )); + continue; + } + if let Some(event) = events.iter().find(|e| e.guardrail_blocked) { + wrong.push(format!( + "{surface}: {} marked guardrail_blocked on a {} that no guardrail refused", + event.error_class, event.status_code, + )); + } + } + assert!( + wrong.is_empty(), + "guardrail_blocked must track guardrail refusals, not failures in general — \ + a flag set by every 4xx/5xx makes the Blocked view useless:\n {}", + wrong.join("\n "), + ); +} diff --git a/crates/aisix-proxy/src/images.rs b/crates/aisix-proxy/src/images.rs index fcc5de00..9b25ee51 100644 --- a/crates/aisix-proxy/src/images.rs +++ b/crates/aisix-proxy/src/images.rs @@ -227,6 +227,7 @@ pub async fn image_generations( &api_key_id, status, err.kind(), + err.is_guardrail_block(), &client, crate::usage_attr::enforced_hits(&audit), ); diff --git a/crates/aisix-proxy/src/images_edits.rs b/crates/aisix-proxy/src/images_edits.rs index 190ad133..83b04543 100644 --- a/crates/aisix-proxy/src/images_edits.rs +++ b/crates/aisix-proxy/src/images_edits.rs @@ -223,6 +223,7 @@ pub async fn image_edits( &api_key_id, status, err.kind(), + err.is_guardrail_block(), &client, crate::usage_attr::enforced_hits(&audit), ); diff --git a/crates/aisix-proxy/src/jobs.rs b/crates/aisix-proxy/src/jobs.rs index ce6cd049..c248f9d4 100644 --- a/crates/aisix-proxy/src/jobs.rs +++ b/crates/aisix-proxy/src/jobs.rs @@ -779,6 +779,7 @@ fn finish( &auth.entry.id, status, err.kind(), + err.is_guardrail_block(), client, enforced_hits, ); diff --git a/crates/aisix-proxy/src/lib.rs b/crates/aisix-proxy/src/lib.rs index 112c5e32..8239487e 100644 --- a/crates/aisix-proxy/src/lib.rs +++ b/crates/aisix-proxy/src/lib.rs @@ -43,6 +43,8 @@ mod ensemble; mod error; mod error_translate; #[cfg(test)] +mod guardrail_blocked_telemetry; +#[cfg(test)] mod guardrail_coverage; mod guardrail_embedder; mod guardrail_stream; diff --git a/crates/aisix-proxy/src/messages.rs b/crates/aisix-proxy/src/messages.rs index d32600bb..421100f2 100644 --- a/crates/aisix-proxy/src/messages.rs +++ b/crates/aisix-proxy/src/messages.rs @@ -249,6 +249,10 @@ pub async fn messages( // ...and the terminal trace spans. /* terminal_last */ false, + // These are the attempts a WINNER superseded — the request + // was served, so no guardrail refused it. + /* guardrail_blocked */ + false, &audit, ); if !usage_handled_by_stream { @@ -293,6 +297,9 @@ pub async fn messages( metrics, &client, attempt, + // The winner served the caller — nothing refused it. + /* guardrail_blocked */ + false, applied_guardrails.clone(), redaction_counts.clone(), monitor_hits.clone(), @@ -349,6 +356,11 @@ pub async fn messages( }, elapsed, ); + // AISIX-Cloud#1428: a guardrail refusal IS this failure, so the + // terminal event must say so — it is what the dashboard's + // "Guardrail blocks" view filters on. Every other 4xx/5xx class + // leaves the flag alone. + let guardrail_blocked = err.is_guardrail_block(); // AISIX-Cloud#1013: failed requests carry the (post-mask) // request body so a 4xx/5xx can be triaged from the log alone. // Same opt-in gate and cap as the success path; 401/403 stay @@ -401,6 +413,7 @@ pub async fn messages( // emission; the pre-dispatch branch below covers empty. /* terminal_last */ !routing.attempts.is_empty(), + guardrail_blocked, &audit, ); // Pre-dispatch failure (model-not-found, auth, budget, guardrail @@ -430,6 +443,7 @@ pub async fn messages( error_class: err.kind().to_string(), ..Default::default() }, + guardrail_blocked, applied_guardrails.clone(), // Input masking may have fired before the failure. redaction_counts.clone(), @@ -479,6 +493,9 @@ fn emit_failed_attempts_anthropic( // event is the request's terminal emission, so it carries the trace's // SERVER + logical spans. False on the success path. terminal_last: bool, + // Whether the request ended in a guardrail refusal (AISIX-Cloud#1428). + // Rides the same event as the audit handle below, for the same reason. + guardrail_blocked: bool, // The request's enforced-guardrail audit handle (AISIX-Cloud#1330); // stamped only on the event this call marks terminal. audit: &crate::usage_attr::GuardrailAudit, @@ -516,6 +533,7 @@ fn emit_failed_attempts_anthropic( AnthropicUsageMetrics::default(), client, AttemptInfo::from_record(rec), + guardrail_blocked, applied_guardrails.to_vec(), // Failed attempts carry no per-request redaction detail; the // terminal (winner / pre-dispatch) event does. @@ -1483,7 +1501,16 @@ async fn anthropic_passthrough_dispatch( // as 499, matching LiteLLM. The upstream work still // happened, so the event is emitted either way — only // its outcome differs. - if usage.reached_end { + // + // A guardrail refusal is not an abandonment, whatever + // `reached_end` says: the hold-back-overflow arm returns + // mid-stream, so the flag is the only thing that tells + // "the gateway ended this" from "the caller went away". + // chat.rs reaches 200 here by `break`ing to its + // end-of-upstream marker instead; same answer, and this + // way `reached_end` keeps meaning what it says + // (AISIX-Cloud#1428). + if usage.reached_end || usage.guardrail_blocked { 200 } else { crate::CLIENT_CLOSED_REQUEST @@ -1494,6 +1521,7 @@ async fn anthropic_passthrough_dispatch( metrics, &client_ctx_c, attempt_c.clone(), + usage.guardrail_blocked, applied_guardrails_c.clone(), // #932: input-side mask counts captured before dispatch, // merged with the hold-back release's output-side counts. @@ -2137,8 +2165,9 @@ async fn cross_provider_dispatch( user_id_for_telem.as_deref(), user_name_for_telem.as_deref(), // See the sibling passthrough path: an abandoned stream - // is reported as 499, matching LiteLLM. - if comp.reached_end { + // is reported as 499, matching LiteLLM — and a guardrail + // refusal is not an abandonment. + if comp.reached_end || comp.guardrail_blocked { 200 } else { crate::CLIENT_CLOSED_REQUEST @@ -2148,6 +2177,7 @@ async fn cross_provider_dispatch( metrics, &client_for_telem, attempt_for_telem.clone(), + comp.guardrail_blocked, applied_guardrails_for_telem.clone(), // #932: input-side mask counts captured before dispatch, // merged with the hold-back release's output-side counts. @@ -2483,6 +2513,7 @@ fn build_anthropic_sse_stream( max_buffer_bytes = max_hold, "streaming /v1/messages response exceeded hold-back cap; failing closed", ); + guard.comp().guardrail_blocked = true; yield Ok(bytes::Bytes::from(guardrail_block_frame(None, Some(crate::error::TAG_OUTPUT_BUFFER_EXCEEDED)))); return; } @@ -2591,6 +2622,7 @@ fn build_anthropic_sse_stream( ); // Hold-back: the held chunks are dropped — the matched // content never reached the wire. + guard.comp().guardrail_blocked = true; let frame = guardrail_block_frame(guardrail_name.as_deref(), unavailable.as_deref()); yield Ok(bytes::Bytes::from(frame)); return; @@ -2712,6 +2744,13 @@ struct AnthropicStreamCompletion { /// check (AISIX-Cloud#562). Merged with the input-side hits by the /// on_complete emit. monitor_hits: Vec, + /// Set when the end-of-stream output check refused the response and the + /// held frames were dropped for a terminal `error` frame — or when the + /// hold-back buffer overflowed and the stream failed closed. The stream + /// had already committed its upstream tokens, so the event keeps them, + /// but it must not read as a clean delivery (AISIX-Cloud#1428). Mirrors + /// `chat::StreamCompletion::guardrail_blocked`. + guardrail_blocked: bool, } struct CompleteAnthropicStreamOnDrop { @@ -2877,6 +2916,13 @@ fn emit_anthropic_usage_event( metrics: AnthropicUsageMetrics, client: &ClientContext, attempt: AttemptInfo, + // Whether a guardrail refused this request — on the input hook before + // dispatch, or on the output hook after the upstream answered. The + // dashboard's "Guardrail blocks" view filters on exactly this bool, so + // an unset one hides a 422 the caller definitely saw + // (AISIX-Cloud#1428). Request-scoped like `guardrail_enforced_hits` + // below, hence terminal-only. + guardrail_blocked: bool, // The `{kind, hook}` set of guardrails that governed this request (#379). // Empty for the guardrail-free path and pre-resolution failures. applied_guardrails: Vec, @@ -2944,6 +2990,8 @@ fn emit_anthropic_usage_event( // the terminal event carries them — a superseded attempt would // otherwise repeat the same hit once per retry. guardrail_enforced_hits: crate::usage_attr::terminal_enforced_hits(terminal, audit), + // Same rule, same reason. + guardrail_blocked: terminal && guardrail_blocked, ..Default::default() }; // Handler label "messages" — Anthropic /v1/messages inbound @@ -3111,6 +3159,13 @@ struct AnthropicStreamUsage { /// check (AISIX-Cloud#562). Merged with the input-side hits by the /// on_complete emit. monitor_hits: Vec, + /// Set when the end-of-stream output check refused the response and the + /// held frames were dropped for a terminal `error` frame — or when the + /// hold-back buffer overflowed and the stream failed closed. The stream + /// had already committed its upstream tokens, so the event keeps them, + /// but it must not read as a clean delivery (AISIX-Cloud#1428). Mirrors + /// `chat::StreamCompletion::guardrail_blocked`. + guardrail_blocked: bool, } /// Update the accumulator from one parsed SSE `data:` JSON object. @@ -3511,6 +3566,7 @@ where max_buffer_bytes = max_hold, "streaming /v1/messages passthrough exceeded hold-back cap; failing closed", ); + guard.usage().guardrail_blocked = true; yield Ok(Bytes::from(guardrail_block_frame(None, Some(crate::error::TAG_OUTPUT_BUFFER_EXCEEDED)))); return; } @@ -3623,6 +3679,7 @@ where "guardrail blocked streaming /v1/messages passthrough response", ); blocked = true; + guard.usage().guardrail_blocked = true; let frame = guardrail_block_frame(guardrail_name.as_deref(), unavailable.as_deref()); yield Ok(Bytes::from(frame)); } @@ -5783,4 +5840,207 @@ event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n"; .expect("usage event sender dropped"); assert_masked_by_eda(&event); } + + /// AISIX-Cloud#1428: a STREAMING `/v1/messages` response the output hook + /// refuses must be recorded as a guardrail block. + /// + /// The refusal happens after the response head is out, so the caller + /// gets a 200 followed by a terminal `error` frame — and the usage row, + /// which is emitted from the stream's Drop guard, therefore also + /// carries 200 with the upstream's tokens. Everything about it read as + /// a clean delivery: the request was refused, the held content dropped, + /// and neither the row's status nor its flag said so. `guardrail_blocked` + /// is the only field that can — the status must stay 200 because that + /// is what the caller was actually sent, which is the same shape + /// `/v1/chat/completions` records. + /// + /// Both streaming relays are driven, because each accumulates into its + /// own struct and so needed the flag wired separately: the Anthropic + /// passthrough (raw upstream SSE bytes, held and released) and the + /// cross-provider bridge (`ChatChunk`s re-encoded into Anthropic SSE). + #[tokio::test] + async fn streaming_output_block_marks_guardrail_blocked_usage_event() { + use aisix_obs::UsageSink; + use aisix_provider_openai::OpenAiBridge; + + // (relay, upstream path, upstream SSE, model entry, snapshot, + // billed prompt/completion tokens) + let anthropic_sse = "\ +event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_block\",\"role\":\"assistant\",\"content\":[],\"model\":\"claude-3-5-haiku-20241022\",\"stop_reason\":null,\"usage\":{\"input_tokens\":11,\"output_tokens\":1}}}\n\n\ +event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n\ +event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"here it is: BLOCKME\"}}\n\n\ +event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n\ +event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":9}}\n\n\ +event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n"; + let openai_sse = "\ +data: {\"id\":\"cmpl-block\",\"object\":\"chat.completion.chunk\",\"created\":1715000000,\"model\":\"gpt-4o\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"},\"finish_reason\":null}]}\n\n\ +data: {\"id\":\"cmpl-block\",\"object\":\"chat.completion.chunk\",\"created\":1715000000,\"model\":\"gpt-4o\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"here it is: BLOCKME\"},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":11,\"completion_tokens\":9,\"total_tokens\":20}}\n\n\ +data: [DONE]\n\n"; + + for (relay, upstream_path, sse) in [ + ("anthropic passthrough", "/v1/messages", anthropic_sse), + ("cross-provider bridge", "/chat/completions", openai_sse), + ] { + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path(upstream_path)) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse), + ) + .mount(&upstream) + .await; + + let anthropic_upstream = upstream_path == "/v1/messages"; + let snap = if anthropic_upstream { + let snap = new_snap_anthropic(&upstream.uri()); + snap.models.insert(anthropic_model("my-claude")); + snap + } else { + let snap = new_snap_openai(&upstream.uri()); + snap.models.insert(openai_model("my-claude")); + snap + }; + snap.apikeys.insert(apikey_entry(&["*"])); + let row: aisix_core::models::Guardrail = serde_json::from_str( + r#"{"name":"out-block","enabled":true,"kind":"keyword","hook_point":"output","fail_open":false,"patterns":[{"kind":"literal","value":"BLOCKME"}]}"#, + ) + .unwrap(); + snap.guardrails.insert(ResourceEntry::new("g-out", row, 1)); + + let (tx, mut rx) = tokio::sync::mpsc::channel(4); + let hub = Arc::new(Hub::new()); + hub.register_specialized("anthropic", Arc::new(AnthropicBridge::new())); + hub.register_specialized("openai", Arc::new(OpenAiBridge::new())); + let state = crate::ProxyState::new(SnapshotHandle::new(snap), hub, &cfg()) + .without_cache() + .with_usage_sink(UsageSink::new(tx)); + + let resp = crate::build_router(state) + .oneshot(make_req(serde_json::json!({ + "model": "my-claude", + "messages": [{"role": "user", "content": "hi"}], + "max_tokens": 100, + "stream": true, + }))) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK, "{relay}"); + let streamed = + String::from_utf8(to_bytes(resp.into_body(), 65536).await.unwrap().to_vec()) + .unwrap(); + // Hold-back: the matched content never reached the wire. + assert!( + !streamed.contains("BLOCKME"), + "{relay}: the blocked content was released: {streamed}" + ); + + let event = tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()) + .await + .expect("usage event was never emitted") + .expect("usage event sender dropped"); + assert!( + event.guardrail_blocked, + "{relay}: a refused stream must be findable under guardrail_blocked=true" + ); + // The upstream generated (and billed) before the hook refused, + // so the tokens stay on the row — under-reporting spend the + // customer was charged for would be the wrong repair. + assert_eq!(event.prompt_tokens, 11, "{relay}"); + assert_eq!(event.completion_tokens, 9, "{relay}"); + // A refusal is not an abandonment. Both relays report the 200 the + // caller's response head already committed, which is what + // `/v1/chat/completions` records for the same event. + assert_eq!(event.status_code, 200, "{relay}"); + } + } + + /// AISIX-Cloud#1428: the hold-back OVERFLOW arm — a response too large to + /// buffer for scanning, which fails closed — is a guardrail refusal too, + /// and must not be filed as a client abandonment. + /// + /// This arm returns mid-stream, before the upstream-EOF marker, so + /// `reached_end` stays false and the row used to report `499`: "the + /// caller went away". Nobody went away — the gateway refused. chat.rs + /// gets 200 here by `break`ing out to its EOF marker; this reaches the + /// same answer off the flag, which leaves `reached_end` meaning what its + /// doc says. + #[tokio::test] + async fn streaming_holdback_overflow_is_a_block_not_an_abandonment() { + use aisix_obs::UsageSink; + + // Past DEFAULT_STREAM_OUTPUT_BUFFER_BYTES (256 KiB) of held text, + // and deliberately clean: the cap, not the content, is what refuses. + let big = "x".repeat(300_000); + let upstream = MockServer::start().await; + let sse = format!( + "\ +event: message_start\ndata: {{\"type\":\"message_start\",\"message\":{{\"id\":\"msg_big\",\"role\":\"assistant\",\"content\":[],\"model\":\"claude-3-5-haiku-20241022\",\"stop_reason\":null,\"usage\":{{\"input_tokens\":11,\"output_tokens\":1}}}}}}\n\n\ +event: content_block_start\ndata: {{\"type\":\"content_block_start\",\"index\":0,\"content_block\":{{\"type\":\"text\",\"text\":\"\"}}}}\n\n\ +event: content_block_delta\ndata: {{\"type\":\"content_block_delta\",\"index\":0,\"delta\":{{\"type\":\"text_delta\",\"text\":\"{big}\"}}}}\n\n\ +event: content_block_stop\ndata: {{\"type\":\"content_block_stop\",\"index\":0}}\n\n\ +event: message_delta\ndata: {{\"type\":\"message_delta\",\"delta\":{{\"stop_reason\":\"end_turn\"}},\"usage\":{{\"output_tokens\":9}}}}\n\n\ +event: message_stop\ndata: {{\"type\":\"message_stop\"}}\n\n" + ); + Mock::given(method("POST")) + .and(path("/v1/messages")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse), + ) + .mount(&upstream) + .await; + + let (tx, mut rx) = tokio::sync::mpsc::channel(4); + let snap = new_snap_anthropic(&upstream.uri()); + snap.models.insert(anthropic_model("my-claude")); + snap.apikeys.insert(apikey_entry(&["*"])); + // A row whose streamed-output policy is the whole-response hold-back; + // the literal never appears, so only the cap can refuse. + let row: aisix_core::models::Guardrail = serde_json::from_str( + r#"{"name":"out-block","enabled":true,"kind":"keyword","hook_point":"output","fail_open":false,"patterns":[{"kind":"literal","value":"NEVERAPPEARS"}]}"#, + ) + .unwrap(); + snap.guardrails.insert(ResourceEntry::new("g-out", row, 1)); + + let hub = Arc::new(Hub::new()); + hub.register_specialized("anthropic", Arc::new(AnthropicBridge::new())); + let state = crate::ProxyState::new(SnapshotHandle::new(snap), hub, &cfg()) + .without_cache() + .with_usage_sink(UsageSink::new(tx)); + + let resp = crate::build_router(state) + .oneshot(make_req(serde_json::json!({ + "model": "my-claude", + "messages": [{"role": "user", "content": "hi"}], + "max_tokens": 100, + "stream": true, + }))) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let streamed = + String::from_utf8(to_bytes(resp.into_body(), 1 << 20).await.unwrap().to_vec()).unwrap(); + assert!( + streamed.contains(crate::error::TAG_OUTPUT_BUFFER_EXCEEDED), + "the oversized stream must fail closed: {}", + &streamed[..streamed.len().min(400)] + ); + assert!( + !streamed.contains(&big), + "unscannable content must not be released" + ); + + let event = tokio::time::timeout(std::time::Duration::from_millis(1000), rx.recv()) + .await + .expect("usage event was never emitted") + .expect("usage event sender dropped"); + assert!(event.guardrail_blocked); + assert_eq!( + event.status_code, 200, + "a fail-closed refusal is not a client abandonment" + ); + } } diff --git a/crates/aisix-proxy/src/passthrough_route.rs b/crates/aisix-proxy/src/passthrough_route.rs index 6b29e9cd..67a13258 100644 --- a/crates/aisix-proxy/src/passthrough_route.rs +++ b/crates/aisix-proxy/src/passthrough_route.rs @@ -353,6 +353,7 @@ pub async fn entry( api_key_id, status, error.kind(), + error.is_guardrail_block(), &client, crate::usage_attr::enforced_hits(&audit), ); diff --git a/crates/aisix-proxy/src/realtime.rs b/crates/aisix-proxy/src/realtime.rs index fc739263..c42b173c 100644 --- a/crates/aisix-proxy/src/realtime.rs +++ b/crates/aisix-proxy/src/realtime.rs @@ -241,6 +241,7 @@ pub(crate) async fn realtime( api_key_id.unwrap_or(""), status, err.kind(), + err.is_guardrail_block(), &client, // Refused before the handshake, so no chain was ever // resolved and no guardrail can have enforced anything. @@ -598,6 +599,10 @@ async fn run_session( &auth.entry.id, 502, "transport", + // Failing to open the upstream socket is not a guardrail + // decision, whatever the chain went on to allow. + /* guardrail_blocked */ + false, &client, crate::usage_attr::enforced_hits(&audit), ); @@ -832,6 +837,13 @@ async fn run_session( inbound_protocol: "realtime".to_string(), client_source_ip: client.source_ip.clone(), client_user_agent: client.user_agent.clone(), + // A frame the chain refused ends the session, so the session's one + // terminal event is where the refusal has to be recorded — this is + // the only realtime row the "Guardrail blocks" view can ever see + // (AISIX-Cloud#1428). + guardrail_blocked: session_error + .as_ref() + .is_some_and(ProxyError::is_guardrail_block), guardrail_monitor_hits: monitor_hits, guardrail_enforced_hits: crate::usage_attr::enforced_hits(&audit), ..Default::default() diff --git a/crates/aisix-proxy/src/rerank.rs b/crates/aisix-proxy/src/rerank.rs index 0d07aa62..a09971ef 100644 --- a/crates/aisix-proxy/src/rerank.rs +++ b/crates/aisix-proxy/src/rerank.rs @@ -227,6 +227,7 @@ pub async fn rerank( &api_key_id, status, err.kind(), + err.is_guardrail_block(), &client, crate::usage_attr::enforced_hits(&audit), ); diff --git a/crates/aisix-proxy/src/responses.rs b/crates/aisix-proxy/src/responses.rs index b498e834..849c4ca6 100644 --- a/crates/aisix-proxy/src/responses.rs +++ b/crates/aisix-proxy/src/responses.rs @@ -292,6 +292,10 @@ pub async fn responses( // The winner's event carries the terminal spans. /* terminal_last */ false, + // These are the attempts a WINNER superseded — the request + // was served, so no guardrail refused it. + /* guardrail_blocked */ + false, &audit, ); // Issue #404: emit UsageEvent so cp-api's budget ledger @@ -413,6 +417,11 @@ pub async fn responses( }, elapsed, ); + // AISIX-Cloud#1428: a guardrail refusal IS this failure, so the + // terminal event must say so — it is what the dashboard's + // "Guardrail blocks" view filters on. Every other 4xx/5xx class + // leaves the flag alone. + let guardrail_blocked = err.is_guardrail_block(); // AISIX-Cloud#1013: failed requests carry the (post-mask) // request body so a 4xx/5xx can be triaged from the log alone. // Same opt-in gate and cap as the success path; 401/403 stay @@ -459,6 +468,7 @@ pub async fn responses( // emission; the pre-dispatch branch below covers empty. /* terminal_last */ !routing.attempts.is_empty(), + guardrail_blocked, &audit, ); // Pre-dispatch failure (model-not-found, auth, budget) records no @@ -483,6 +493,7 @@ pub async fn responses( error_class: err.kind().to_string(), ..Default::default() }, + guardrail_blocked, // Input masking may have fired before the failure. redaction_counts.clone(), monitor_hits.clone(), @@ -3093,6 +3104,11 @@ fn emit_zero_token_event( elapsed: Duration, client: &ClientContext, attempt: AttemptInfo, + // Whether the request ended in a guardrail refusal, from + // [`ProxyError::is_guardrail_block`]. Request-scoped like the enforced + // hits below, so it lands on the terminal event only + // (AISIX-Cloud#1428). + guardrail_blocked: bool, // Per-detector PII mask counts (#932): input masking may have fired // before the failure. Empty for most failure classes. redacted_entity_counts: crate::redact::RedactionCounts, @@ -3140,6 +3156,8 @@ fn emit_zero_token_event( // superseded attempt's event would repeat the same hit per retry. // Only the terminal event carries them. guardrail_enforced_hits: crate::usage_attr::terminal_enforced_hits(terminal, audit), + // Same rule, same reason as the hits above. + guardrail_blocked: terminal && guardrail_blocked, ..Default::default() }; crate::usage_attr::apply_jwt_identity(&mut event, client.jwt.as_ref()); @@ -3177,6 +3195,9 @@ fn emit_failed_attempts( // event is the request's terminal emission, so it carries the trace's // SERVER + logical spans. False on the success path. terminal_last: bool, + // Whether the request ended in a guardrail refusal (AISIX-Cloud#1428). + // Rides the same event as the audit handle below, for the same reason. + guardrail_blocked: bool, // The request's enforced-guardrail audit handle (AISIX-Cloud#1330); // stamped only on the event this call marks terminal. audit: &crate::usage_attr::GuardrailAudit, @@ -3207,6 +3228,7 @@ fn emit_failed_attempts( Duration::from_millis(u64::from(rec.latency_ms)), client, AttemptInfo::from_record(rec), + guardrail_blocked, // Failed attempts carry no per-request redaction detail; the // terminal event does. crate::redact::RedactionCounts::new(), @@ -3465,6 +3487,91 @@ mod tests { assert!(!msg.contains("BLOCKME"), "blocklist literal leaked: {msg}"); } + /// A routing (group) parent over one member, so the same blocked + /// request can be addressed either directly or through the group. + fn routing_model(name: &str, target: &str) -> ResourceEntry { + let json = format!( + r#"{{"display_name":"{name}","routing":{{"strategy":"failover","targets":[{{"model":"{target}"}}]}}}}"# + ); + let m: Model = serde_json::from_str(&json).unwrap(); + ResourceEntry::new(format!("router-{name}"), m, 1) + } + + /// AISIX-Cloud#1428: an input-guardrail refusal on `/v1/responses` must + /// emit a zero-token 422 UsageEvent marked `guardrail_blocked`, which is + /// the exact predicate the dashboard's Logs "Guardrail blocks" view + /// filters on. Before the fix the event carried the flag's `false` + /// default, so the request appeared in the unfiltered feed and vanished + /// from the Blocked one — which reads as the gateway having logged no + /// guardrail activity at all. + /// + /// Driven over all four combinations the report names, because they + /// take different code: a direct model refuses before any attempt is + /// recorded (the terminal event is the pre-dispatch one), a group + /// parent resolves its targets first, and `stream: true` changes which + /// response the handler builds. The input hook runs before target + /// selection either way, so all four must agree. + #[tokio::test] + async fn input_guardrail_block_marks_guardrail_blocked_usage_event() { + use aisix_obs::UsageSink; + + for model in ["gpt-4o-resp", "resp-group"] { + for stream in [false, true] { + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/responses")) + .respond_with(ResponseTemplate::new(200)) + .expect(0) + .mount(&upstream) + .await; + + let snap = new_snap_openai(&upstream.uri()); + snap.models.insert(openai_model("gpt-4o-resp")); + snap.models + .insert(routing_model("resp-group", "gpt-4o-resp")); + snap.apikeys.insert(apikey_entry(&["*"])); + snap.guardrails.insert(keyword_input_guardrail("BLOCKME")); + + let (tx, mut rx) = tokio::sync::mpsc::channel(8); + let hub = Arc::new(Hub::new()); + hub.register_specialized("openai", Arc::new(OpenAiBridge::new())); + let state = crate::ProxyState::new(SnapshotHandle::new(snap), hub, &cfg()) + .without_cache() + .with_usage_sink(UsageSink::new(tx)); + + let resp = crate::build_router(state) + .oneshot(make_req(serde_json::json!({ + "model": model, + "input": "please BLOCKME now", + "stream": stream, + }))) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::UNPROCESSABLE_ENTITY, + "model={model} stream={stream}" + ); + + let event = tokio::time::timeout(std::time::Duration::from_millis(1000), rx.recv()) + .await + .expect("usage event must be emitted") + .expect("usage_sink sender dropped"); + assert!( + event.guardrail_blocked, + "model={model} stream={stream}: the refusal must be findable under \ + guardrail_blocked=true" + ); + assert_eq!(event.status_code, 422, "model={model} stream={stream}"); + // Nothing was sent upstream, so nothing is billed. + assert_eq!(event.prompt_tokens, 0, "model={model} stream={stream}"); + assert_eq!(event.completion_tokens, 0, "model={model} stream={stream}"); + // The caller-addressed entry, group or not (AISIX-Cloud#790). + assert_eq!(event.requested_model, model); + } + } + } + /// #719: the Responses `input` array form (message items with typed /// content parts) must be scanned too — a blocked literal inside an /// `input_text` part blocks the call. diff --git a/crates/aisix-proxy/src/usage_attr.rs b/crates/aisix-proxy/src/usage_attr.rs index d4c5144a..974e27db 100644 --- a/crates/aisix-proxy/src/usage_attr.rs +++ b/crates/aisix-proxy/src/usage_attr.rs @@ -428,6 +428,12 @@ pub(crate) fn apply_jwt_identity( /// `"passthrough"` for `/passthrough/...`, `"realtime"` for `/v1/realtime`, /// `"openai"` for the OpenAI-shaped handlers) so Logs protocol filtering sees /// failures and successes under the same tag. +/// +/// A guardrail refusal reaches this function like any other failure, so +/// `guardrail_blocked` is a REQUIRED argument rather than a default: a +/// handler that forgets it produces a 422 the "Guardrail blocks" view +/// cannot find, which reads to an operator as "the gateway logged no +/// guardrail activity at all" (AISIX-Cloud#1428). #[allow(clippy::too_many_arguments)] pub(crate) fn emit_error_usage_event( state: &ProxyState, @@ -439,6 +445,11 @@ pub(crate) fn emit_error_usage_event( api_key_id: &str, status_code: u16, error_class: &str, + // Whether this failure IS a guardrail refusal, from + // [`ProxyError::is_guardrail_block`]. Every caller but the realtime + // connect failure — which synthesizes its class without a `ProxyError` + // — reads it off the error it is reporting. + guardrail_blocked: bool, client: &ClientContext, // The request's enforced guardrail hits. The failure path is where a // `blocked` hit lands — a guardrail refusal IS the error — so the @@ -454,6 +465,7 @@ pub(crate) fn emit_error_usage_event( api_key_id, status_code, error_class, + guardrail_blocked, client, enforced, ); @@ -485,6 +497,8 @@ pub(crate) fn build_error_usage_event( api_key_id: &str, status_code: u16, error_class: &str, + // See [`emit_error_usage_event`]. + guardrail_blocked: bool, client: &ClientContext, enforced: Vec, ) -> UsageEvent { @@ -498,6 +512,7 @@ pub(crate) fn build_error_usage_event( error_class: error_class.to_string(), client_source_ip: client.source_ip.clone(), client_user_agent: client.user_agent.clone(), + guardrail_blocked, guardrail_enforced_hits: enforced, ..Default::default() }; diff --git a/crates/aisix-proxy/src/videos.rs b/crates/aisix-proxy/src/videos.rs index a4787c17..e46a2ee6 100644 --- a/crates/aisix-proxy/src/videos.rs +++ b/crates/aisix-proxy/src/videos.rs @@ -1596,6 +1596,7 @@ pub async fn create_video( &auth.entry.id, status, err.kind(), + err.is_guardrail_block(), &client, crate::usage_attr::enforced_hits(&audit), ); diff --git a/tests/e2e/src/cases/guardrail-blocked-usage-flag-e2e.test.ts b/tests/e2e/src/cases/guardrail-blocked-usage-flag-e2e.test.ts new file mode 100644 index 00000000..82e72b7f --- /dev/null +++ b/tests/e2e/src/cases/guardrail-blocked-usage-flag-e2e.test.ts @@ -0,0 +1,246 @@ +import { createHash } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + EtcdClient, + ProxyClient, + SeedClient, + slsLogsFor, + spawnApp, + startMockSls, + startOpenAiUpstream, + waitConfigPropagation, + waitForSlsLog, + type MockSls, + type OpenAiUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +// E2E for AISIX-Cloud#1428: a guardrail refusal on /v1/responses must be +// findable under `guardrail_blocked = true`. +// +// The report: an input guardrail attached to the model a caller addresses +// refuses the request with 422 `content_filter`, and the usage row it +// produces carries `guardrail_blocked` at its `false` default. The row IS +// in the unfiltered feed, so the dashboard's Logs "Guardrail blocks" view +// — whose whole predicate is `guardrail_blocked = true` — comes back empty +// while the caller is being refused. To an operator that reads as "the +// gateway records no guardrail activity at all", which is a worse answer +// than a missing row. +// +// Driven over the four combinations the report names — direct model or +// routing-group parent, streaming or not — because they take different +// code inside the handler even though the input hook runs before target +// selection on all four. Read back off a real Aliyun-SLS export from a +// real `aisix` binary, so what is asserted is the row a consumer actually +// receives, not an in-process struct. +// +// The clean control at the end is what gives the four assertions meaning: +// the same models and the same guardrail, text it does not match, must +// produce a row with the flag OFF. A flag that merely tracked "a guardrail +// was configured" — or "the request failed" — would pass the first four +// and fail that one. + +const CALLER_PLAINTEXT = "sk-guardrail-blocked-flag-caller"; +const CALLER_KEY_HASH = createHash("sha256").update(CALLER_PLAINTEXT).digest("hex"); + +const CREDENTIAL_REF = "mock"; +const MOCK_AK_ID = "LTAI_mock_ak"; +const MOCK_AK_SECRET = "mock_ak_secret"; +const SLS_PROJECT = "aisix-e2e-obs"; +const LOGSTORE = "guardrail-blocked-flag"; + +const FORBIDDEN_WORD = "blockedflagsentinel"; +const DIRECT_MODEL = "gbf-direct"; +const GROUP_MODEL = "gbf-group"; + +/** Each blocked probe plants a unique marker so its row can be found. */ +interface Probe { + model: string; + stream: boolean; + marker: string; +} + +const PROBES: Probe[] = [ + { model: DIRECT_MODEL, stream: false, marker: "gbf-direct-nonstream-4a91" }, + { model: DIRECT_MODEL, stream: true, marker: "gbf-direct-stream-7c02" }, + { model: GROUP_MODEL, stream: false, marker: "gbf-group-nonstream-2e58" }, + { model: GROUP_MODEL, stream: true, marker: "gbf-group-stream-9b13" }, +]; + +describe("guardrail_blocked e2e: a /v1/responses refusal reaches the Blocked view (#1428)", () => { + let upstream: OpenAiUpstream | undefined; + let sls: MockSls | undefined; + let app: SpawnedApp | undefined; + let etcdReachable = false; + + async function responses( + model: string, + input: string, + stream: boolean, + ): Promise { + const res = await fetch(`${app!.proxyUrl}/v1/responses`, { + method: "POST", + headers: { + authorization: `Bearer ${CALLER_PLAINTEXT}`, + "content-type": "application/json", + }, + body: JSON.stringify({ model, input, stream }), + }); + // Drain: a streaming response only completes its telemetry once the + // body is consumed. + await res.text(); + return res; + } + + beforeAll(async () => { + const etcd = new EtcdClient(); + etcdReachable = await etcd.ping(); + if (!etcdReachable) return; + + sls = await startMockSls(); + upstream = await startOpenAiUpstream({ + nonStreamBody: { + id: "resp_gbf", + object: "response", + status: "completed", + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "fine" }], + }, + ], + usage: { input_tokens: 3, output_tokens: 1 }, + }, + }); + + app = await spawnApp({ + extraEnv: { + [`SLS_CRED_${CREDENTIAL_REF.toUpperCase()}_AK_ID`]: MOCK_AK_ID, + [`SLS_CRED_${CREDENTIAL_REF.toUpperCase()}_AK_SECRET`]: MOCK_AK_SECRET, + }, + }); + const seed = new SeedClient(etcd, app.etcdPrefix); + + await seed.createObservabilityExporter({ + name: "gbf-sls", + enabled: true, + kind: "aliyun_sls", + endpoint: sls.url, + project: SLS_PROJECT, + logstore: LOGSTORE, + credential_ref: CREDENTIAL_REF, + content_mode: "full", + }); + + const pk = await seed.createProviderKey({ + display_name: "gbf-pk", + secret: "sk-mock", + api_base: `${upstream.baseUrl}/v1`, + }); + await seed.createModel({ + display_name: DIRECT_MODEL, + provider: "openai", + model_name: "gpt-4o-mini", + provider_key_id: pk.id, + }); + // A group over the one direct member. The report's own workaround for + // the separate member-guardrail gap (#1090) is to attach the rule to + // the group instead, so the group parent is the shape an affected + // operator is most likely to be running. + await seed.createModel({ + display_name: GROUP_MODEL, + routing: { + strategy: "failover", + targets: [{ model: DIRECT_MODEL }], + }, + }); + + // Env-scoped input guardrail: it governs the entry the caller + // addresses, group parent included. + await seed.createGuardrail({ + name: "gbf-guard", + enabled: true, + hook_point: "input", + kind: "keyword", + patterns: [{ kind: "literal", value: FORBIDDEN_WORD }], + }); + + // The caller key is written LAST, after every other resource above. + // The gateway runs one etcd watch over one prefix and applies its + // events in revision order (`aisix-etcd`), so the moment this key + // authenticates, everything written ahead of it — models, the routing + // group, the guardrail, the exporter — is already in the snapshot. + await seed.createApiKey({ + key_hash: CALLER_KEY_HASH, + allowed_models: [DIRECT_MODEL, GROUP_MODEL], + }); + + // ...which is why the readiness gate can stay independent of what the + // tests assert. Gating on the guardrail's own 422 would make a + // guardrail regression surface as a propagation timeout in `beforeAll` + // instead of a failed assertion pointing at the cause. + const proxy = new ProxyClient(app.proxyUrl, CALLER_PLAINTEXT); + await waitConfigPropagation(async () => (await proxy.listModels()).status === 200); + }); + + afterAll(async () => { + await app?.exit(); + await upstream?.close(); + await sls?.close(); + }); + + for (const { model, stream, marker } of PROBES) { + test(`input block on ${model} (stream=${stream}) is recorded as a guardrail block`, async (ctx) => { + if (!etcdReachable || !app || !sls) { + ctx.skip(); + return; + } + const res = await responses(model, `${marker} ${FORBIDDEN_WORD}`, stream); + expect(res.status).toBe(422); + + // The row exists in the unfiltered feed… + const log = await waitForSlsLog( + sls, + LOGSTORE, + (l) => (l.get("prompt") ?? "").includes(marker), + `${model} stream=${stream} usage row`, + ); + // …and the Blocked view's predicate finds it. + expect(log.get("guardrail_blocked")).toBe("true"); + expect(log.get("status_code")).toBe("422"); + // Nothing was sent upstream, so nothing is billed. + expect(log.get("prompt_tokens") ?? "0").toBe("0"); + expect(log.get("completion_tokens") ?? "0").toBe("0"); + // The caller-addressed entry, group parent included. + expect(log.get("requested_model")).toBe(model); + + // Exactly one row per refusal: an input block refuses before any + // target is contacted, so a group parent must not also emit a + // per-attempt row. + const rows = slsLogsFor(sls, LOGSTORE).filter((l) => + (l.get("prompt") ?? "").includes(marker), + ); + expect(rows).toHaveLength(1); + }); + } + + test("a request the same guardrail allows is not marked blocked", async (ctx) => { + if (!etcdReachable || !app || !sls) { + ctx.skip(); + return; + } + const marker = "gbf-clean-6d47"; + const res = await responses(DIRECT_MODEL, `${marker} an ordinary question`, false); + expect(res.status).toBe(200); + + const log = await waitForSlsLog( + sls, + LOGSTORE, + (l) => (l.get("prompt") ?? "").includes(marker), + "allowed request usage row", + ); + expect(log.get("guardrail_blocked")).not.toBe("true"); + expect(log.get("status_code")).toBe("200"); + }); +}); diff --git a/tests/e2e/src/cases/sls-failure-content-e2e.test.ts b/tests/e2e/src/cases/sls-failure-content-e2e.test.ts index 83ac2b5d..99287e55 100644 --- a/tests/e2e/src/cases/sls-failure-content-e2e.test.ts +++ b/tests/e2e/src/cases/sls-failure-content-e2e.test.ts @@ -3,11 +3,12 @@ import { afterAll, beforeAll, describe, expect, test } from "vitest"; import { EtcdClient, SeedClient, - lz4DecompressBlock, + slsLogsFor, spawnApp, startMockSls, startOpenAiUpstream, waitConfigPropagation, + waitForSlsLog, type MockSls, type OpenAiUpstream, type SpawnedApp, @@ -51,97 +52,9 @@ const EMAIL = "dana@example.com"; const CN_ID = "11010519491231002X"; // valid ISO 7064 MOD 11-2 check digit const MASKED_BLOCK_SENTINEL = "masked-block-prompt-6a4e8b"; -// --- Minimal SLS LogGroup protobuf reader (see sink/sls.rs encoder) ----- -// LogGroup { Logs = 1 (message) { Time = 1 (varint), Contents = 2 (message) -// { Key = 1 (string), Value = 2 (string) } } }; unknown fields skipped. - -function readVarint(buf: Buffer, pos: number): [number, number] { - let result = 0; - let shift = 0; - for (;;) { - const b = buf[pos]!; - pos += 1; - result += (b & 0x7f) * 2 ** shift; - if ((b & 0x80) === 0) return [result, pos]; - shift += 7; - } -} - -function skipField(buf: Buffer, pos: number, wireType: number): number { - if (wireType === 0) return readVarint(buf, pos)[1]; - if (wireType === 2) { - const [len, p] = readVarint(buf, pos); - return p + len; - } - if (wireType === 5) return pos + 4; - if (wireType === 1) return pos + 8; - throw new Error(`unsupported wire type ${wireType}`); -} - -function parseContentPair(buf: Buffer): [string, string] { - let pos = 0; - let key = ""; - let value = ""; - while (pos < buf.length) { - const [tag, p] = readVarint(buf, pos); - pos = p; - const field = tag >>> 3; - const wireType = tag & 7; - if (wireType === 2) { - const [len, q] = readVarint(buf, pos); - const bytes = buf.subarray(q, q + len); - pos = q + len; - if (field === 1) key = bytes.toString("utf8"); - else if (field === 2) value = bytes.toString("utf8"); - } else { - pos = skipField(buf, pos, wireType); - } - } - return [key, value]; -} - -function parseLog(buf: Buffer): Map { - const out = new Map(); - let pos = 0; - while (pos < buf.length) { - const [tag, p] = readVarint(buf, pos); - pos = p; - const field = tag >>> 3; - const wireType = tag & 7; - if (field === 2 && wireType === 2) { - const [len, q] = readVarint(buf, pos); - const [k, v] = parseContentPair(buf.subarray(q, q + len)); - out.set(k, v); - pos = q + len; - } else { - pos = skipField(buf, pos, wireType); - } - } - return out; -} - /** Decode every log delivered to `logstore` into flat key→value maps. */ function logsFor(sls: MockSls, logstore: string): Map[] { - const logs: Map[] = []; - for (const r of sls.requests) { - if (r.logstore !== logstore || r.rawSize === 0 || r.body.length === 0) continue; - const group = lz4DecompressBlock(r.body, r.rawSize); - let pos = 0; - while (pos < group.length) { - const [tag, p] = readVarint(group, pos); - pos = p; - const field = tag >>> 3; - const wireType = tag & 7; - if (field === 1 && wireType === 2) { - const [len, q] = readVarint(group, pos); - logs.push(parseLog(group.subarray(q, q + len))); - pos = q + len; - } else { - pos = skipField(group, pos, wireType); - } - } - } - return logs; + return slsLogsFor(sls, logstore); } /** Poll until a FULL_LOGSTORE log matching `pred` arrives (or time out). */ @@ -151,13 +64,7 @@ async function waitForLog( what: string, timeoutMs = 10_000, ): Promise> { - const deadline = Date.now() + timeoutMs; - while (Date.now() < deadline) { - const hit = logsFor(sls, FULL_LOGSTORE).find(pred); - if (hit) return hit; - await new Promise((r) => setTimeout(r, 100)); - } - throw new Error(`no SLS log matching: ${what}`); + return waitForSlsLog(sls, FULL_LOGSTORE, pred, what, timeoutMs); } // ------------------------------------------------------------------------- diff --git a/tests/e2e/src/harness/index.ts b/tests/e2e/src/harness/index.ts index 92c31dc8..5596e651 100644 --- a/tests/e2e/src/harness/index.ts +++ b/tests/e2e/src/harness/index.ts @@ -29,6 +29,8 @@ export { waitForLogstore, waitForToken, lz4DecompressBlock, + slsLogsFor, + waitForSlsLog, type MockSls, type CapturedPutLogs, } from "./sls-mock.js"; diff --git a/tests/e2e/src/harness/sls-mock.ts b/tests/e2e/src/harness/sls-mock.ts index 92b51038..40bb4987 100644 --- a/tests/e2e/src/harness/sls-mock.ts +++ b/tests/e2e/src/harness/sls-mock.ts @@ -146,3 +146,133 @@ export async function waitForToken( } throw new Error(`token '${token}' not seen in logstore '${logstore}' within ${timeoutMs}ms`); } + +// --- LogGroup protobuf reader (see aisix-obs sink/sls.rs encoder) -------- +// LogGroup { Logs = 1 (message) { Time = 1 (varint), Contents = 2 (message) +// { Key = 1 (string), Value = 2 (string) } } }; unknown fields skipped. + +function readVarint(buf: Buffer, pos: number): [number, number] { + let result = 0; + let shift = 0; + for (;;) { + // Past the end `buf[pos]` is `undefined`, and `undefined & 0x80` is 0 — + // so the loop would exit with a wrong value and an advanced position, + // and the mis-parse would surface much later as an opaque + // `waitForSlsLog` timeout instead of naming the truncated payload. + if (pos >= buf.length) { + throw new Error(`truncated varint at offset ${pos} of ${buf.length} bytes`); + } + const b = buf[pos]!; + pos += 1; + result += (b & 0x7f) * 2 ** shift; + if ((b & 0x80) === 0) return [result, pos]; + shift += 7; + } +} + +function skipField(buf: Buffer, pos: number, wireType: number): number { + if (wireType === 0) return readVarint(buf, pos)[1]; + if (wireType === 2) { + const [len, p] = readVarint(buf, pos); + return p + len; + } + if (wireType === 5) return pos + 4; + if (wireType === 1) return pos + 8; + throw new Error(`unsupported wire type ${wireType}`); +} + +function parseContentPair(buf: Buffer): [string, string] { + let pos = 0; + let key = ""; + let value = ""; + while (pos < buf.length) { + const [tag, p] = readVarint(buf, pos); + pos = p; + const field = tag >>> 3; + const wireType = tag & 7; + if (wireType === 2) { + const [len, q] = readVarint(buf, pos); + const bytes = buf.subarray(q, q + len); + pos = q + len; + if (field === 1) key = bytes.toString("utf8"); + else if (field === 2) value = bytes.toString("utf8"); + } else { + pos = skipField(buf, pos, wireType); + } + } + return [key, value]; +} + +function parseLog(buf: Buffer): Map { + const out = new Map(); + let pos = 0; + while (pos < buf.length) { + const [tag, p] = readVarint(buf, pos); + pos = p; + const field = tag >>> 3; + const wireType = tag & 7; + if (field === 2 && wireType === 2) { + const [len, q] = readVarint(buf, pos); + const [k, v] = parseContentPair(buf.subarray(q, q + len)); + out.set(k, v); + pos = q + len; + } else { + pos = skipField(buf, pos, wireType); + } + } + return out; +} + +/** + * Every log delivered to `logstore`, decoded into flat key→value maps. + * + * [`decodedTextFor`] answers "did this token reach the logstore"; this + * answers "what does THIS request's row say", which a substring search + * cannot — a field like `guardrail_blocked` is present on every row, so + * only a per-record read can tell one row's value from another's. + * + * `fromIndex` has the same meaning as in [`decodedTextFor`]. + */ +export function slsLogsFor( + sls: MockSls, + logstore: string, + fromIndex = 0, +): Map[] { + const logs: Map[] = []; + for (const r of sls.requests.slice(fromIndex)) { + if (r.logstore !== logstore || r.rawSize === 0 || r.body.length === 0) continue; + const group = lz4DecompressBlock(r.body, r.rawSize); + let pos = 0; + while (pos < group.length) { + const [tag, p] = readVarint(group, pos); + pos = p; + const field = tag >>> 3; + const wireType = tag & 7; + if (field === 1 && wireType === 2) { + const [len, q] = readVarint(group, pos); + logs.push(parseLog(group.subarray(q, q + len))); + pos = q + len; + } else { + pos = skipField(group, pos, wireType); + } + } + } + return logs; +} + +/** Poll until a `logstore` record matching `pred` arrives (or time out). */ +export async function waitForSlsLog( + sls: MockSls, + logstore: string, + pred: (log: Map) => boolean, + what: string, + timeoutMs = 10_000, +): Promise> { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const hit = slsLogsFor(sls, logstore).find(pred); + if (hit) return hit; + await new Promise((r) => setTimeout(r, 100)); + } + throw new Error(`no SLS log in '${logstore}' matching: ${what}`); +}