diff --git a/Cargo.lock b/Cargo.lock index b24d8012..3f42dd9b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -158,13 +158,18 @@ dependencies = [ "aws-smithy-async", "aws-smithy-runtime-api", "aws-smithy-types", + "base64 0.22.1", + "chrono", + "hmac 0.12.1", "regex", "reqwest", "serde", "serde_json", + "sha1", "thiserror 1.0.69", "tokio", "tracing", + "uuid", "wiremock", ] @@ -754,7 +759,7 @@ dependencies = [ "bytes", "form_urlencoded", "hex", - "hmac", + "hmac 0.13.0", "http 0.2.12", "http 1.4.0", "percent-encoding", @@ -1602,6 +1607,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer 0.10.4", "crypto-common 0.1.7", + "subtle", ] [[package]] @@ -2109,6 +2115,15 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest 0.10.7", +] + [[package]] name = "hmac" version = "0.13.0" diff --git a/Cargo.toml b/Cargo.toml index 54c2a69e..f73895e2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -100,6 +100,8 @@ base64 = "0.22" # Crypto (M6 v3 CP register flow + ApiKey hash auth) sha2 = "0.10" +sha1 = "0.10" +hmac = "0.12" rcgen = { version = "0.13", default-features = false, features = ["pem", "ring"] } hex = "0.4" diff --git a/crates/aisix-core/src/models/guardrail.rs b/crates/aisix-core/src/models/guardrail.rs index 0737757d..f3214e37 100644 --- a/crates/aisix-core/src/models/guardrail.rs +++ b/crates/aisix-core/src/models/guardrail.rs @@ -32,6 +32,10 @@ //! * `azure_content_safety` — calls Azure AI Content Safety Prompt //! Shield (`/contentsafety/text:shieldPrompt`). Detects jailbreak //! and indirect injection attacks. P1 (PRD-09c §6 P1). +//! * `aliyun_text_moderation` — calls Aliyun's content-safety +//! guardrail (`TextModerationPlus` on `green-cip..aliyuncs.com`). +//! Risk-level moderation on input (`llm_query_moderation`) and output +//! (`llm_response_moderation`). #603. //! //! See `aisix-guardrails/src/keyword.rs` for the runtime semantics //! the snapshot is parsed into. @@ -256,6 +260,83 @@ fn default_acs_on_buffer_exceeded() -> String { "fail_closed".to_owned() } +/// Config block for `kind: "aliyun_text_moderation"`. Calls Aliyun's +/// content-safety guardrail (`TextModerationPlus`, action version +/// `2022-03-02`) on the `green-cip..aliyuncs.com` endpoint for +/// category-risk moderation on input and/or output (including streaming +/// output). Issue #603. +/// +/// The input hook uses the `llm_query_moderation` service code, the +/// output hook `llm_response_moderation`. Aliyun grades each call with a +/// `RiskLevel` (`none`/`low`/`medium`/`high`); the DP blocks when the +/// returned level reaches `risk_level_threshold`. +/// +/// Only `access_key_secret` is a secret (decrypted by cp-api before kine +/// projection — plaintext in DP memory only, never logged); every other +/// field travels in the clear. +#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct AliyunTextModerationConfig { + /// Aliyun region the guardrail lives in, e.g. `cn-shanghai`. The DP + /// builds the endpoint `https://green-cip..aliyuncs.com`. + pub region: String, + /// Explicit endpoint override (full URL, no trailing slash). When set + /// it wins over `region`; used by tests/dev to point at a mock server. + #[serde(default)] + pub endpoint: Option, + /// Aliyun AccessKey ID. + pub access_key_id: String, + /// Aliyun AccessKey secret. Decrypted by cp-api before kine projection; + /// plaintext in memory only, never logged. Used to sign the request. + pub access_key_secret: String, + /// Minimum risk level that triggers a block: `low`, `medium`, or + /// `high` (default). A returned level at or above this blocks. + #[serde(default = "default_aliyun_risk_level_threshold")] + pub risk_level_threshold: String, + /// HTTP call timeout (ms); `fail_open` / `output_fail_open` govern the + /// verdict when it elapses. See the `AzureContentSafetyConfig` note on + /// why `0` means "fire immediately", not "no timeout". + #[serde(default = "default_acs_timeout_ms")] + pub timeout_ms: u32, + /// Fail-open policy for the OUTPUT hook. Defaults `false` (fail-closed) + /// so an Aliyun outage can't release unscanned model output. + #[serde(default)] + pub output_fail_open: bool, + + // --- streaming-output controls (consumed by aisix-proxy build_sse_stream) --- + /// `window` (sliding-window incremental release; default) or + /// `buffer_full` (whole-response hold-back). + #[serde(default = "default_acs_stream_processing_mode")] + pub stream_processing_mode: String, + /// Sliding-window size in chars (window mode). Default 2 000 — Aliyun's + /// `llm_response_moderation` per-call content cap. + #[serde(default = "default_aliyun_window_size")] + pub window_size: u32, + /// Chars carried between windows so a span split across a boundary is + /// still caught. Default 128. + #[serde(default = "default_aliyun_window_overlap_size")] + pub window_overlap_size: u32, + /// Max bytes buffered in `buffer_full` mode before `on_buffer_exceeded` + /// applies. Default 262 144. + #[serde(default = "default_acs_max_buffer_bytes")] + pub max_buffer_bytes: u64, + /// `fail_closed` (default) or `fail_open` when the buffer cap is hit. + #[serde(default = "default_acs_on_buffer_exceeded")] + pub on_buffer_exceeded: String, +} + +fn default_aliyun_risk_level_threshold() -> String { + "high".to_owned() +} + +fn default_aliyun_window_size() -> u32 { + 2_000 +} + +fn default_aliyun_window_overlap_size() -> u32 { + 128 +} + /// Config block for `kind: "bedrock"`. Phase 1 stores the shape + /// passes it through `aisix-guardrails::build` which logs /// `bedrock not yet implemented` and skips the row. @@ -294,6 +375,10 @@ pub enum GuardrailKind { /// on input and/or output (including streaming output). P2 /// (PRD-09c §6 P2, #379). AzureContentSafetyTextModeration(AzureContentSafetyTextModerationConfig), + /// Aliyun content-safety guardrail. Risk-level moderation via the + /// `TextModerationPlus` action on `green-cip..aliyuncs.com`, + /// on input and/or output (including streaming output). #603. + AliyunTextModeration(AliyunTextModerationConfig), } /// Top-level `Guardrail` resource shape. Mirrors what cp-api writes @@ -736,6 +821,68 @@ mod tests { } } + #[test] + fn aliyun_text_moderation_kind_parses_with_defaults() { + // cp-api omits unset fields (omitempty); the DP must apply the + // documented defaults so a minimal row still moderates correctly. + let v = json!({ + "name": "aliyun-guard", + "kind": "aliyun_text_moderation", + "region": "cn-shanghai", + "access_key_id": "LTAI_EXAMPLE", + "access_key_secret": "PLAINTEXT_FOR_TEST" + }); + let g: Guardrail = serde_json::from_value(v).unwrap(); + match g.config { + GuardrailKind::AliyunTextModeration(ref c) => { + assert_eq!(c.region, "cn-shanghai"); + assert_eq!(c.access_key_id, "LTAI_EXAMPLE"); + assert_eq!(c.access_key_secret, "PLAINTEXT_FOR_TEST"); + assert_eq!(c.endpoint, None); + assert_eq!(c.risk_level_threshold, "high"); + assert_eq!(c.timeout_ms, 5_000); + assert!(!c.output_fail_open); + assert_eq!(c.stream_processing_mode, "window"); + assert_eq!(c.window_size, 2_000); + assert_eq!(c.window_overlap_size, 128); + assert_eq!(c.max_buffer_bytes, 262_144); + assert_eq!(c.on_buffer_exceeded, "fail_closed"); + } + _ => panic!("expected AliyunTextModeration variant"), + } + } + + #[test] + fn aliyun_text_moderation_kind_round_trips_set_fields() { + let v = json!({ + "name": "aliyun-guard", + "kind": "aliyun_text_moderation", + "region": "cn-beijing", + "endpoint": "http://127.0.0.1:8080", + "access_key_id": "id", + "access_key_secret": "secret", + "risk_level_threshold": "medium", + "timeout_ms": 3000, + "output_fail_open": true, + "stream_processing_mode": "buffer_full", + "window_size": 1000, + "window_overlap_size": 0 + }); + let g: Guardrail = serde_json::from_value(v).unwrap(); + match g.config { + GuardrailKind::AliyunTextModeration(ref c) => { + assert_eq!(c.endpoint.as_deref(), Some("http://127.0.0.1:8080")); + assert_eq!(c.risk_level_threshold, "medium"); + assert_eq!(c.timeout_ms, 3000); + assert!(c.output_fail_open); + assert_eq!(c.stream_processing_mode, "buffer_full"); + assert_eq!(c.window_size, 1000); + assert_eq!(c.window_overlap_size, 0, "explicit 0 overlap must survive"); + } + _ => panic!("expected AliyunTextModeration variant"), + } + } + #[test] fn resource_trait_uses_name_and_guardrails_kind() { let mut g: Guardrail = serde_json::from_value(json!({ diff --git a/crates/aisix-core/src/models/mod.rs b/crates/aisix-core/src/models/mod.rs index 6ca7b166..a82accd9 100644 --- a/crates/aisix-core/src/models/mod.rs +++ b/crates/aisix-core/src/models/mod.rs @@ -30,9 +30,9 @@ pub mod snapshot; pub use apikey::ApiKey; pub use cache_policy::{AppliesTo, CacheBackend, CachePolicy}; pub use guardrail::{ - AzureContentSafetyConfig, AzureContentSafetyTextModerationConfig, BedrockAWSCredentials, - BedrockConfig, BedrockLatencyMode, Guardrail, GuardrailAttachment, GuardrailHookPoint, - GuardrailKind, GuardrailScopeType, KeywordConfig, KeywordPattern, + AliyunTextModerationConfig, AzureContentSafetyConfig, AzureContentSafetyTextModerationConfig, + BedrockAWSCredentials, BedrockConfig, BedrockLatencyMode, Guardrail, GuardrailAttachment, + GuardrailHookPoint, GuardrailKind, GuardrailScopeType, KeywordConfig, KeywordPattern, }; pub use model::{ Adapter, BackgroundModelCheck, CooldownConfig, Model, DEFAULT_COOLDOWN_TRIGGER_STATUSES, diff --git a/crates/aisix-core/src/models/schema.rs b/crates/aisix-core/src/models/schema.rs index 39043918..3616a2b3 100644 --- a/crates/aisix-core/src/models/schema.rs +++ b/crates/aisix-core/src/models/schema.rs @@ -409,7 +409,7 @@ fn guardrail_schema() -> Value { "enabled": { "type": "boolean" }, "hook_point": { "enum": ["input", "output", "both"] }, "fail_open": { "type": "boolean" }, - "kind": { "enum": ["keyword", "bedrock", "azure_content_safety", "azure_content_safety_text_moderation"] } + "kind": { "enum": ["keyword", "bedrock", "azure_content_safety", "azure_content_safety_text_moderation", "aliyun_text_moderation"] } }, "oneOf": [ { @@ -482,6 +482,31 @@ fn guardrail_schema() -> Value { "on_buffer_exceeded": { "enum": ["fail_closed", "fail_open"] }, "output_fail_open": { "type": "boolean" } } + }, + { + // kind=aliyun_text_moderation — Aliyun content-safety + // guardrail (TextModerationPlus). Mirrors + // AliyunTextModerationConfig in guardrail.rs: region + + // access keys required, endpoint override + threshold + + // streaming params optional (cp-api applies defaults + + // strict validation on write). #603. + "type": "object", + "required": ["kind", "region", "access_key_id", "access_key_secret"], + "properties": { + "kind": { "const": "aliyun_text_moderation" }, + "region": { "type": "string", "minLength": 1 }, + "endpoint": { "type": "string", "minLength": 1 }, + "access_key_id": { "type": "string", "minLength": 1 }, + "access_key_secret": { "type": "string", "minLength": 1 }, + "risk_level_threshold": { "enum": ["low", "medium", "high"] }, + "timeout_ms": { "type": "integer", "minimum": 0, "maximum": 4_294_967_295u64 }, + "output_fail_open": { "type": "boolean" }, + "stream_processing_mode": { "enum": ["window", "buffer_full"] }, + "window_size": { "type": "integer", "minimum": 1, "maximum": 2_000 }, + "window_overlap_size": { "type": "integer", "minimum": 0 }, + "max_buffer_bytes": { "type": "integer", "minimum": 1 }, + "on_buffer_exceeded": { "enum": ["fail_closed", "fail_open"] } + } } ], "$defs": { @@ -1213,6 +1238,61 @@ mod tests { assert!(validate_guardrail(&v).is_err()); } + #[test] + fn guardrail_aliyun_text_moderation_passes() { + // Minimal row: region + access keys. Optional fields (endpoint, + // threshold, streaming params) omitted — the struct applies defaults. + let v = json!({ + "name": "aliyun-guard", + "kind": "aliyun_text_moderation", + "hook_point": "both", + "region": "cn-shanghai", + "access_key_id": "LTAI_EXAMPLE", + "access_key_secret": "plaintext-secret" + }); + validate_guardrail(&v).unwrap(); + } + + #[test] + fn guardrail_aliyun_text_moderation_with_optional_fields_passes() { + let v = json!({ + "name": "aliyun-guard", + "kind": "aliyun_text_moderation", + "region": "cn-beijing", + "endpoint": "http://127.0.0.1:8080", + "access_key_id": "id", + "access_key_secret": "secret", + "risk_level_threshold": "medium", + "timeout_ms": 3000, + "stream_processing_mode": "buffer_full" + }); + validate_guardrail(&v).unwrap(); + } + + #[test] + fn guardrail_aliyun_text_moderation_missing_secret_rejected() { + let v = json!({ + "name": "g", + "kind": "aliyun_text_moderation", + "region": "cn-shanghai", + "access_key_id": "id" + }); + assert!(validate_guardrail(&v).is_err()); + } + + #[test] + fn guardrail_aliyun_text_moderation_bad_threshold_rejected() { + let v = json!({ + "name": "g", + "kind": "aliyun_text_moderation", + "region": "cn-shanghai", + "access_key_id": "id", + "access_key_secret": "s", + "risk_level_threshold": "none" + }); + assert!(validate_guardrail(&v).is_err()); + } + // ---- rate_limit_policy schema tests ---- #[test] diff --git a/crates/aisix-guardrails/Cargo.toml b/crates/aisix-guardrails/Cargo.toml index 0ed12e4f..09a223e4 100644 --- a/crates/aisix-guardrails/Cargo.toml +++ b/crates/aisix-guardrails/Cargo.toml @@ -35,8 +35,18 @@ aws-smithy-types = { workspace = true, optional = true } # so a slim build can opt out. reqwest = { workspace = true, optional = true } +# kind=aliyun_text_moderation guardrail (#603). The Aliyun green-cip +# TextModerationPlus call is signed (RPC signature v1, HMAC-SHA1) and POSTed +# over the shared reqwest client. There is no official Aliyun Rust SDK, so the +# RPC signature is hand-rolled with these primitives (all workspace deps). +hmac = { workspace = true, optional = true } +sha1 = { workspace = true, optional = true } +base64 = { workspace = true, optional = true } +chrono = { workspace = true, optional = true } +uuid = { workspace = true, optional = true } + [features] -default = ["bedrock", "azure-content-safety"] +default = ["bedrock", "azure-content-safety", "aliyun-text-moderation"] bedrock = [ "dep:aws-config", "dep:aws-sdk-bedrockruntime", @@ -46,6 +56,14 @@ bedrock = [ "dep:aws-smithy-types", ] azure-content-safety = ["dep:reqwest"] +aliyun-text-moderation = [ + "dep:reqwest", + "dep:hmac", + "dep:sha1", + "dep:base64", + "dep:chrono", + "dep:uuid", +] [dev-dependencies] tokio = { workspace = true, features = ["macros", "rt"] } diff --git a/crates/aisix-guardrails/src/aliyun.rs b/crates/aisix-guardrails/src/aliyun.rs new file mode 100644 index 00000000..676aa369 --- /dev/null +++ b/crates/aisix-guardrails/src/aliyun.rs @@ -0,0 +1,767 @@ +//! kind=aliyun_text_moderation guardrail dispatcher — calls Aliyun's +//! content-safety guardrail (`TextModerationPlus`) on chat input and/or +//! output and translates the returned `RiskLevel` into a +//! [`GuardrailVerdict`]. +//! +//! Issue #603. +//! +//! API reference (action version 2022-03-02, RPC-style): +//! POST `https://green-cip..aliyuncs.com/` +//! Source: +//! +//! Wire shape: +//! ```text +//! // Request (form-urlencoded, RPC signature v1): +//! // Action=TextModerationPlus&Version=2022-03-02&Service=llm_query_moderation +//! // &ServiceParameters={"content":"...","sessionId":"..."}&Signature=... +//! // Response (HTTP 200): +//! { "Code": 200, "Data": { "RiskLevel": "high|medium|low|none", +//! "Result": [ { "Label": "..." } ] }, "RequestId": "..." } +//! ``` +//! +//! Block decision: the returned `RiskLevel` rank (none; + +const ACTION: &str = "TextModerationPlus"; +const API_VERSION: &str = "2022-03-02"; +const SERVICE_INPUT: &str = "llm_query_moderation"; +const SERVICE_OUTPUT: &str = "llm_response_moderation"; + +/// Per-call content cap (chars). Aliyun caps `llm_query_moderation` at +/// 2 000 and `llm_response_moderation` at 5 000; 2 000 is the safe shared +/// bound and matches the default streaming window. +const MAX_CONTENT_CHARS: usize = 2_000; + +/// One Aliyun Text Moderation row, materialised into a request-time +/// dispatcher. +pub struct AliyunTextModerationGuardrail { + row_name: String, + /// Full endpoint base, no trailing slash (e.g. + /// `https://green-cip.cn-shanghai.aliyuncs.com`). + endpoint: String, + region: String, + access_key_id: String, + access_key_secret: String, + pub(crate) hook_point: GuardrailHookPoint, + /// Fail-open policy for the INPUT hook (from the outer `Guardrail`). + fail_open: bool, + /// Fail-open policy for the OUTPUT hook. Defaults `false` (fail-closed) + /// so an Aliyun outage can't release unscanned model output. + output_fail_open: bool, + /// Minimum returned risk rank that blocks (none=0 … high=3). + threshold_rank: u8, + pub(crate) timeout: Duration, + client: Arc, + + // --- streaming-output controls (surfaced via stream_output_policy) --- + stream_processing_mode: String, + window_size: u32, + window_overlap_size: u32, + max_buffer_bytes: u64, + on_buffer_exceeded: String, +} + +impl AliyunTextModerationGuardrail { + pub fn new( + row_name: impl Into, + cfg: &AliyunTextModerationConfig, + hook_point: GuardrailHookPoint, + fail_open: bool, + ) -> Self { + let client = reqwest::Client::builder() + .build() + .expect("reqwest::Client::builder() failed; this should never happen"); + let endpoint = cfg + .endpoint + .clone() + .unwrap_or_else(|| format!("https://green-cip.{}.aliyuncs.com", cfg.region)); + Self { + row_name: row_name.into(), + endpoint: endpoint.trim_end_matches('/').to_owned(), + region: cfg.region.clone(), + access_key_id: cfg.access_key_id.clone(), + access_key_secret: cfg.access_key_secret.clone(), + hook_point, + fail_open, + output_fail_open: cfg.output_fail_open, + threshold_rank: risk_rank(&cfg.risk_level_threshold), + timeout: Duration::from_millis(cfg.timeout_ms as u64), + client: Arc::new(client), + stream_processing_mode: cfg.stream_processing_mode.clone(), + window_size: cfg.window_size, + window_overlap_size: cfg.window_overlap_size, + max_buffer_bytes: cfg.max_buffer_bytes, + on_buffer_exceeded: cfg.on_buffer_exceeded.clone(), + } + } + + /// Moderate one piece of text with the given service code. `session_id` + /// (when set) is forwarded as `ServiceParameters.sessionId` so Aliyun + /// correlates the chunks of one streamed response. + async fn moderate( + &self, + service: &str, + text: &str, + session_id: Option<&str>, + fail_open: bool, + ) -> GuardrailVerdict { + // Aliyun caps content per call; truncate to the cap. Streaming + // already windows to MAX_CONTENT_CHARS; non-streaming long inputs + // are clamped (the leading content carries the risk in practice). + let content: String = text.chars().take(MAX_CONTENT_CHARS).collect(); + match self.call(service, &content, session_id).await { + Ok(level) => { + if risk_rank(&level) >= self.threshold_rank { + GuardrailVerdict::Block { + reason: format!( + "aliyun text moderation: risk level {} >= threshold (row: {})", + level, self.row_name + ), + } + } else { + GuardrailVerdict::Allow + } + } + Err(failure) => self.handle_failure(failure, fail_open), + } + } + + /// Sign + POST one `TextModerationPlus` call; return the response + /// `RiskLevel` (lowercased, `"none"` when absent). + async fn call( + &self, + service: &str, + content: &str, + session_id: Option<&str>, + ) -> Result { + let mut svc_params = serde_json::Map::new(); + svc_params.insert( + "content".into(), + serde_json::Value::String(content.to_owned()), + ); + if let Some(sid) = session_id { + if !sid.is_empty() { + svc_params.insert( + "sessionId".into(), + serde_json::Value::String(sid.to_owned()), + ); + } + } + let service_parameters = serde_json::Value::Object(svc_params).to_string(); + + let nonce = uuid::Uuid::new_v4().to_string(); + let timestamp = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); + + // Common + business params. BTreeMap keeps them sorted by key, which + // is exactly the canonicalization order the v1 signature requires. + let mut params: std::collections::BTreeMap<&str, String> = + std::collections::BTreeMap::new(); + params.insert("AccessKeyId", self.access_key_id.clone()); + params.insert("Action", ACTION.to_owned()); + params.insert("Format", "JSON".to_owned()); + params.insert("RegionId", self.region.clone()); + params.insert("Service", service.to_owned()); + params.insert("ServiceParameters", service_parameters); + params.insert("SignatureMethod", "HMAC-SHA1".to_owned()); + params.insert("SignatureNonce", nonce); + params.insert("SignatureVersion", "1.0".to_owned()); + params.insert("Timestamp", timestamp); + params.insert("Version", API_VERSION.to_owned()); + + let signature = sign(¶ms, &self.access_key_secret); + + // Body = signed params + Signature, form-urlencoded (RFC3986 — the + // same encoding used to build the signature, so the server re-derives + // an identical StringToSign). + let mut body = String::new(); + for (k, v) in ¶ms { + if !body.is_empty() { + body.push('&'); + } + body.push_str(k); + body.push('='); + body.push_str(&percent_encode(v)); + } + body.push_str("&Signature="); + body.push_str(&percent_encode(&signature)); + + let future = self + .client + .post(format!("{}/", self.endpoint)) + .header("Content-Type", "application/x-www-form-urlencoded") + .header("Accept", "application/json") + .body(body) + .send(); + + let resp = match tokio::time::timeout(self.timeout, future).await { + Err(_elapsed) => return Err(AliyunFailure::Timeout), + Ok(Err(_e)) => return Err(AliyunFailure::IoError), + Ok(Ok(r)) => r, + }; + + let status = resp.status(); + if status == reqwest::StatusCode::TOO_MANY_REQUESTS { + return Err(AliyunFailure::Throttled); + } + if status.is_server_error() { + return Err(AliyunFailure::ServerError); + } + if !status.is_success() { + tracing::error!( + row = %self.row_name, + http_status = status.as_u16(), + "aliyun TextModerationPlus returned 4xx — check region/access keys configuration", + ); + return Err(AliyunFailure::ConfigError); + } + + let body: AliyunResponse = resp.json().await.map_err(|_| AliyunFailure::ServerError)?; + + // Aliyun signals app-level errors via the JSON `Code` (200 = OK) + // even on HTTP 200. + match body.code { + 200 => Ok(body + .data + .and_then(|d| d.risk_level) + .unwrap_or_else(|| "none".to_owned()) + .to_lowercase()), + 408 | 401 | 403 | 400 => { + tracing::error!( + row = %self.row_name, + aliyun_code = body.code, + "aliyun TextModerationPlus auth/permission error — check access keys", + ); + Err(AliyunFailure::ConfigError) + } + other => { + tracing::warn!( + row = %self.row_name, + aliyun_code = other, + "aliyun TextModerationPlus non-200 Code", + ); + Err(AliyunFailure::ServerError) + } + } + } + + fn handle_failure(&self, failure: AliyunFailure, fail_open: bool) -> GuardrailVerdict { + let tag = failure.bypass_tag(); + if !matches!(failure, AliyunFailure::ConfigError) { + tracing::warn!( + row = %self.row_name, + failure = ?failure, + fail_open, + "aliyun text moderation call failed", + ); + } + if fail_open { + GuardrailVerdict::Bypass { reason: tag.into() } + } else { + GuardrailVerdict::Block { + reason: format!("aliyun text moderation unavailable ({tag})"), + } + } + } +} + +/// Rank a risk level so thresholds compare numerically. An unrecognized +/// level ranks as `none` (0) — fail toward allowing rather than blocking +/// on an unexpected label, and the call site logs nothing because Aliyun +/// only ever returns the four known levels. +fn risk_rank(level: &str) -> u8 { + match level.to_ascii_lowercase().as_str() { + "high" => 3, + "medium" => 2, + "low" => 1, + _ => 0, + } +} + +/// RFC3986 percent-encoding with Aliyun's tweaks: unreserved chars +/// (`A-Za-z0-9-_.~`) pass through, every other byte becomes `%XX` +/// (uppercase). Space → `%20`. (Aliyun additionally maps `+`→`%20`, +/// `*`→`%2A`, `%7E`→`~`; we never emit `+` or a literal `*`, and `~` is +/// already unreserved, so encoding each non-unreserved byte covers it.) +fn percent_encode(s: &str) -> String { + let mut out = String::with_capacity(s.len() * 3); + for b in s.bytes() { + match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + out.push(b as char); + } + _ => { + out.push('%'); + out.push_str(&format!("{b:02X}")); + } + } + } + out +} + +/// Build the RPC v1 `StringToSign` from the (already key-sorted) params. +/// Factored out so the canonicalization is unit-testable independent of +/// the HMAC step. +fn string_to_sign(params: &std::collections::BTreeMap<&str, String>) -> String { + let mut canonical = String::new(); + for (k, v) in params { + if !canonical.is_empty() { + canonical.push('&'); + } + canonical.push_str(&percent_encode(k)); + canonical.push('='); + canonical.push_str(&percent_encode(v)); + } + format!( + "POST&{}&{}", + percent_encode("/"), + percent_encode(&canonical) + ) +} + +/// Compute the RPC v1 signature: `Base64(HMAC-SHA1(secret + "&", StringToSign))`. +fn sign(params: &std::collections::BTreeMap<&str, String>, access_key_secret: &str) -> String { + let sts = string_to_sign(params); + let key = format!("{access_key_secret}&"); + let mut mac = + HmacSha1::new_from_slice(key.as_bytes()).expect("HMAC accepts keys of any length"); + mac.update(sts.as_bytes()); + base64::engine::general_purpose::STANDARD.encode(mac.finalize().into_bytes()) +} + +/// Failure cause buckets. `bypass_tag()` maps to the strings stored in +/// `usage_events.guardrail_bypassed_reason`. +#[derive(Debug)] +enum AliyunFailure { + Timeout, + Throttled, + IoError, + ServerError, + ConfigError, +} + +impl AliyunFailure { + fn bypass_tag(&self) -> &'static str { + match self { + Self::Timeout => "aliyun_timeout", + Self::Throttled => "aliyun_throttled", + Self::IoError | Self::ServerError => "aliyun_5xx", + Self::ConfigError => "aliyun_config_error", + } + } +} + +// --- serde shapes for the wire protocol ------------------------------------ + +#[derive(Deserialize)] +struct AliyunResponse { + #[serde(rename = "Code", default)] + code: i32, + #[serde(rename = "Data", default)] + data: Option, +} + +#[derive(Deserialize)] +struct AliyunData { + #[serde(rename = "RiskLevel", default)] + risk_level: Option, +} + +// --- Guardrail trait impl -------------------------------------------------- + +#[async_trait] +impl Guardrail for AliyunTextModerationGuardrail { + fn name(&self) -> &'static str { + "aliyun_text_moderation" + } + + fn stream_output_policy(&self) -> StreamOutputPolicy { + match self.stream_processing_mode.as_str() { + "buffer_full" => StreamOutputPolicy::BufferFull { + max_buffer_bytes: self.max_buffer_bytes as usize, + on_exceeded_fail_open: self.on_buffer_exceeded == "fail_open", + }, + // "window" (default) and any unexpected value → sliding window. + _ => StreamOutputPolicy::Window { + size_chars: self.window_size as usize, + overlap_chars: self.window_overlap_size as usize, + }, + } + } + + async fn check_input(&self, req: &ChatFormat) -> GuardrailVerdict { + if !matches!( + self.hook_point, + GuardrailHookPoint::Input | GuardrailHookPoint::Both + ) { + return GuardrailVerdict::Allow; + } + let text = collect_input_text(req); + if text.is_empty() { + return GuardrailVerdict::Allow; + } + self.moderate(SERVICE_INPUT, &text, None, self.fail_open) + .await + } + + async fn check_output(&self, resp: &ChatResponse) -> GuardrailVerdict { + if !matches!( + self.hook_point, + GuardrailHookPoint::Output | GuardrailHookPoint::Both + ) { + return GuardrailVerdict::Allow; + } + let text = resp.guardrail_output_text(); + if text.is_empty() { + return GuardrailVerdict::Allow; + } + // The upstream provider's request id is stable across all windows + // of one streamed response, so it doubles as the per-stream Aliyun + // sessionId; a fresh uuid keeps non-streaming calls correlated to + // themselves when the provider omits an id. + let session = if resp.id.is_empty() { + uuid::Uuid::new_v4().to_string() + } else { + resp.id.clone() + }; + // Output uses its own fail policy (default fail-closed) so an + // Aliyun outage can't release unscanned model output. + self.moderate(SERVICE_OUTPUT, &text, Some(&session), self.output_fail_open) + .await + } +} + +/// Concatenate the request's user-visible message contents into one blob. +/// (Same collector shape as the Bedrock dispatcher — keeps the families +/// scanning identical text.) +fn collect_input_text(req: &ChatFormat) -> String { + req.messages + .iter() + .map(crate::message_scan_text) + .filter(|s| !s.is_empty()) + .collect::>() + .join("\n") +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use aisix_gateway::{ChatFormat, ChatMessage, ChatResponse, FinishReason, UsageStats}; + use serde_json::json; + use wiremock::matchers::{body_string_contains, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use super::*; + + fn cfg(endpoint: &str, threshold: &str) -> AliyunTextModerationConfig { + serde_json::from_value(json!({ + "region": "cn-shanghai", + "endpoint": endpoint, + "access_key_id": "LTAI_TEST", + "access_key_secret": "test-secret", + "risk_level_threshold": threshold, + "timeout_ms": 5_000, + })) + .unwrap() + } + + fn build(endpoint: &str, threshold: &str, fail_open: bool) -> AliyunTextModerationGuardrail { + AliyunTextModerationGuardrail::new( + "wiremock-test", + &cfg(endpoint, threshold), + GuardrailHookPoint::Both, + fail_open, + ) + } + + fn req(msg: &str) -> ChatFormat { + ChatFormat::new("m", vec![ChatMessage::user(msg)]) + } + + fn resp(content: &str) -> ChatResponse { + ChatResponse { + id: "stream-req-1".into(), + model: "m".into(), + message: ChatMessage::assistant(content), + finish_reason: FinishReason::Stop, + usage: UsageStats::new(0, 0), + } + } + + // --- pure signature / encoding tests --- + + #[test] + fn risk_rank_orders_levels() { + assert!(risk_rank("none") < risk_rank("low")); + assert!(risk_rank("low") < risk_rank("medium")); + assert!(risk_rank("medium") < risk_rank("high")); + assert_eq!(risk_rank("HIGH"), 3, "case-insensitive"); + assert_eq!(risk_rank("garbage"), 0, "unknown ranks as none"); + } + + #[test] + fn percent_encode_matches_aliyun_rules() { + assert_eq!(percent_encode("a b"), "a%20b"); + assert_eq!(percent_encode("/"), "%2F"); + assert_eq!(percent_encode("~-_."), "~-_."); + assert_eq!(percent_encode("{\"k\":\"v\"}"), "%7B%22k%22%3A%22v%22%7D"); + } + + #[test] + fn string_to_sign_is_canonical_and_stable() { + let mut p: std::collections::BTreeMap<&str, String> = std::collections::BTreeMap::new(); + p.insert("Action", "TextModerationPlus".into()); + p.insert("Service", "llm_query_moderation".into()); + let sts = string_to_sign(&p); + // "POST&%2F&" + percentEncode("Action=TextModerationPlus&Service=llm_query_moderation") + assert_eq!( + sts, + "POST&%2F&Action%3DTextModerationPlus%26Service%3Dllm_query_moderation" + ); + } + + #[test] + fn sign_is_deterministic_and_known_vector() { + // Pins the full v1 signature against an openssl-computed reference, + // so a regression in canonicalization or the HMAC step fails loud. + let mut p: std::collections::BTreeMap<&str, String> = std::collections::BTreeMap::new(); + p.insert("Action", "TextModerationPlus".into()); + p.insert("Service", "llm_query_moderation".into()); + let sig = sign(&p, "test-secret"); + assert_eq!(sig, KNOWN_SIGNATURE); + // deterministic + assert_eq!(sign(&p, "test-secret"), sig); + } + + // openssl dgst -sha1 -hmac "test-secret&" over the StringToSign above, + // base64-encoded. Recompute with: + // printf '%s' 'POST&%2F&Action%3DTextModerationPlus%26Service%3Dllm_query_moderation' \ + // | openssl dgst -sha1 -hmac 'test-secret&' -binary | base64 + const KNOWN_SIGNATURE: &str = "pu3Hn+zsRIztpT2f7JT5+zHPPVo="; + + // --- wiremock integration --- + + fn risk_body(level: &str) -> serde_json::Value { + json!({ "Code": 200, "Data": { "RiskLevel": level, "Result": [{ "Label": "x" }] }, "RequestId": "r" }) + } + + #[tokio::test] + async fn clean_input_returns_allow_and_signs_request() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/")) + // proves the signed form body carries Action + Service + Signature + .and(body_string_contains("Action=TextModerationPlus")) + .and(body_string_contains("Service=llm_query_moderation")) + .and(body_string_contains("Signature=")) + .respond_with(ResponseTemplate::new(200).set_body_json(risk_body("none"))) + .expect(1) + .mount(&server) + .await; + + let g = build(&server.uri(), "high", true); + assert_eq!(g.check_input(&req("hello")).await, GuardrailVerdict::Allow); + } + + #[tokio::test] + async fn high_risk_input_blocks_at_high_threshold() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(risk_body("high"))) + .mount(&server) + .await; + let g = build(&server.uri(), "high", true); + assert!(g.check_input(&req("bad")).await.is_block()); + } + + #[tokio::test] + async fn medium_risk_allowed_at_high_threshold_blocked_at_medium() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(risk_body("medium"))) + .mount(&server) + .await; + // threshold=high → medium passes + let g_high = build(&server.uri(), "high", true); + assert_eq!(g_high.check_input(&req("x")).await, GuardrailVerdict::Allow); + // threshold=medium → medium blocks + let g_med = build(&server.uri(), "medium", true); + assert!(g_med.check_input(&req("x")).await.is_block()); + } + + #[tokio::test] + async fn output_sends_session_id() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(body_string_contains("Service=llm_response_moderation")) + // sessionId is JSON-encoded inside ServiceParameters, percent-encoded + // in the body: {"content":"...","sessionId":"stream-req-1"} + .and(body_string_contains("sessionId")) + .respond_with(ResponseTemplate::new(200).set_body_json(risk_body("none"))) + .expect(1) + .mount(&server) + .await; + let g = build(&server.uri(), "high", true); + assert_eq!(g.check_output(&resp("ok")).await, GuardrailVerdict::Allow); + } + + #[tokio::test] + async fn http_5xx_fail_open_true_returns_bypass() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + let g = build(&server.uri(), "high", true); + match g.check_input(&req("x")).await { + GuardrailVerdict::Bypass { reason } => assert_eq!(reason, "aliyun_5xx"), + other => panic!("expected Bypass(aliyun_5xx), got {other:?}"), + } + } + + #[tokio::test] + async fn output_5xx_fails_closed_by_default() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + // output_fail_open defaults false → an output-side 5xx must Block. + let g = build(&server.uri(), "high", true); + assert!( + g.check_output(&resp("model output")).await.is_block(), + "output hook must fail closed on Aliyun error by default" + ); + } + + #[tokio::test] + async fn app_level_403_code_is_config_error_block_when_fail_closed() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(json!({ "Code": 403, "Message": "no permission" })), + ) + .mount(&server) + .await; + // input fail_open=false → config error blocks + let g = build(&server.uri(), "high", false); + assert!(g.check_input(&req("x")).await.is_block()); + } + + #[test] + fn stream_policy_reflects_config() { + let g = build("http://unused", "high", true); + assert_eq!( + g.stream_output_policy(), + StreamOutputPolicy::Window { + size_chars: 2_000, + overlap_chars: 128 + } + ); + let mut g2 = build("http://unused", "high", true); + g2.stream_processing_mode = "buffer_full".to_owned(); + g2.max_buffer_bytes = 1000; + g2.on_buffer_exceeded = "fail_open".to_owned(); + assert_eq!( + g2.stream_output_policy(), + StreamOutputPolicy::BufferFull { + max_buffer_bytes: 1000, + on_exceeded_fail_open: true + } + ); + } + + // --- live smoke test against the real green-cip endpoint --- + // + // Ignored by default (requires real Aliyun credentials + network). + // Run manually with: + // + // ALIYUN_AK_ID=... ALIYUN_AK_SECRET=... ALIYUN_REGION=cn-shanghai \ + // cargo test -p aisix-guardrails aliyun::tests::live_smoke \ + // --features aliyun-text-moderation -- --ignored --nocapture + // + // Exercises the real signer + HTTP + response parse against + // TextModerationPlus and prints the returned RiskLevel for a benign + // and a policy-violating prompt. No credentials are hard-coded. + #[tokio::test] + #[ignore = "requires real Aliyun credentials + network"] + async fn live_smoke_real_endpoint() { + let ak_id = std::env::var("ALIYUN_AK_ID").unwrap_or_default(); + let ak_secret = std::env::var("ALIYUN_AK_SECRET").unwrap_or_default(); + if ak_id.is_empty() || ak_secret.is_empty() { + eprintln!("live_smoke: ALIYUN_AK_ID/ALIYUN_AK_SECRET unset — skipping"); + return; + } + let region = std::env::var("ALIYUN_REGION").unwrap_or_else(|_| "cn-shanghai".to_owned()); + let cfg: AliyunTextModerationConfig = serde_json::from_value(json!({ + "region": region, + "endpoint": std::env::var("ALIYUN_ENDPOINT").ok(), + "access_key_id": ak_id, + "access_key_secret": ak_secret, + "risk_level_threshold": "low", + "timeout_ms": 8000, + })) + .unwrap(); + let g = + AliyunTextModerationGuardrail::new("live-smoke", &cfg, GuardrailHookPoint::Both, false); + + // Benign prompt — expect RiskLevel "none". + let benign = g + .call(SERVICE_INPUT, "今天北京的天气怎么样？", None) + .await + .expect("benign call should succeed"); + eprintln!("live_smoke benign -> RiskLevel={benign}"); + + // Policy-violating prompt (abuse + threat) — expect non-"none". + let risky = g + .call(SERVICE_INPUT, "你这个傻逼，我现在就要弄死你全家", None) + .await + .expect("risky call should succeed"); + eprintln!("live_smoke risky -> RiskLevel={risky}"); + + // Output service with a sessionId, same risky text. + let risky_out = g + .call( + SERVICE_OUTPUT, + "你这个傻逼，我现在就要弄死你全家", + Some("live-sess-1"), + ) + .await + .expect("risky output call should succeed"); + eprintln!("live_smoke output -> RiskLevel={risky_out}"); + + assert_eq!(benign, "none", "benign prompt must score none"); + assert_ne!(risky, "none", "policy-violating prompt must score a risk"); + } +} diff --git a/crates/aisix-guardrails/src/build.rs b/crates/aisix-guardrails/src/build.rs index 603e497b..38234c0c 100644 --- a/crates/aisix-guardrails/src/build.rs +++ b/crates/aisix-guardrails/src/build.rs @@ -173,6 +173,24 @@ fn build_one( GuardrailKind::AzureContentSafetyTextModeration(_) => { Err(BuildError::FeatureDisabled("azure-content-safety")) } + #[cfg(feature = "aliyun-text-moderation")] + GuardrailKind::AliyunTextModeration(cfg) => { + // #603: HTTP-based TextModerationPlus dispatcher. cp-api already + // decrypted the access_key_secret at projection time; the config + // carries plaintext. Endpoint is per-row (derived from the row's + // region, or an explicit override for tests/dev). + let g = crate::aliyun::AliyunTextModerationGuardrail::new( + row.name.clone(), + cfg, + row.hook_point, + row.fail_open, + ); + Ok(Some(Arc::new(g))) + } + #[cfg(not(feature = "aliyun-text-moderation"))] + GuardrailKind::AliyunTextModeration(_) => { + Err(BuildError::FeatureDisabled("aliyun-text-moderation")) + } } } diff --git a/crates/aisix-guardrails/src/lib.rs b/crates/aisix-guardrails/src/lib.rs index 16b9ad31..2424458c 100644 --- a/crates/aisix-guardrails/src/lib.rs +++ b/crates/aisix-guardrails/src/lib.rs @@ -20,6 +20,8 @@ #![forbid(unsafe_code)] #![deny(rust_2018_idioms)] +#[cfg(feature = "aliyun-text-moderation")] +mod aliyun; #[cfg(feature = "bedrock")] mod bedrock; mod build; @@ -60,6 +62,8 @@ pub(crate) fn message_scan_text(m: &ChatMessage) -> String { } } +#[cfg(feature = "aliyun-text-moderation")] +pub use aliyun::AliyunTextModerationGuardrail; #[cfg(feature = "bedrock")] pub use bedrock::BedrockGuardrail; pub use build::{ diff --git a/docs/configuration/guardrails.md b/docs/configuration/guardrails.md index 96608743..fef6b6ff 100644 --- a/docs/configuration/guardrails.md +++ b/docs/configuration/guardrails.md @@ -91,10 +91,45 @@ This is the key difference between schema support and dependable runtime support Keep Bedrock runtime support in the roadmap and limited-capability framing, not as fully available behavior. +## Aliyun Text Moderation Guardrails + +`kind: "aliyun_text_moderation"` calls Aliyun's content-safety guardrail +(`TextModerationPlus`) on the `green-cip..aliyuncs.com` endpoint. The +input hook uses the `llm_query_moderation` service, the output hook +`llm_response_moderation`. The data plane blocks when the returned `RiskLevel` +(`none` < `low` < `medium` < `high`) reaches `risk_level_threshold` (default +`high`). It runs on input and output, including streaming output (windowed, +with the response's request id reused as Aliyun's `sessionId` so the chunks of +one stream correlate). + +Example shape: + +```json title="Aliyun text-moderation guardrail" +{ + "name": "aliyun-review", + "kind": "aliyun_text_moderation", + "hook_point": "both", + "fail_open": false, + "region": "cn-shanghai", + "access_key_id": "YOUR_ACCESS_KEY_ID", + "access_key_secret": "YOUR_ACCESS_KEY_SECRET", + "risk_level_threshold": "high" +} +``` + +- `region` builds the endpoint; set `endpoint` to override it (e.g. a private + proxy) — the override wins over `region`. +- `output_fail_open` defaults `false` so an Aliyun outage cannot release + unscanned model output; the request-level `fail_open` governs the input hook. +- streaming controls (`stream_processing_mode`, `window_size`, + `window_overlap_size`, `max_buffer_bytes`, `on_buffer_exceeded`) mirror the + Azure text-moderation guardrail. + ## Operator Guidance - use `keyword` for production behavior you need to rely on today - treat `bedrock` rows as an advanced or staged capability until your own deployment proves the runtime path you want +- use `aliyun_text_moderation` when your safety stack standardizes on Aliyun's content-safety guardrail; tune `risk_level_threshold` to trade off precision vs. recall ## Troubleshooting diff --git a/schemas/resources/guardrail.schema.json b/schemas/resources/guardrail.schema.json index b83a6b73..2dbe069b 100644 --- a/schemas/resources/guardrail.schema.json +++ b/schemas/resources/guardrail.schema.json @@ -227,6 +227,92 @@ "minimum": 0.0 } } + }, + { + "description": "Aliyun content-safety guardrail. Risk-level moderation via the `TextModerationPlus` action on `green-cip..aliyuncs.com`, on input and/or output (including streaming output). #603.", + "type": "object", + "required": [ + "access_key_id", + "access_key_secret", + "kind", + "region" + ], + "properties": { + "access_key_id": { + "description": "Aliyun AccessKey ID.", + "type": "string" + }, + "access_key_secret": { + "description": "Aliyun AccessKey secret. Decrypted by cp-api before kine projection; plaintext in memory only, never logged. Used to sign the request.", + "type": "string" + }, + "endpoint": { + "description": "Explicit endpoint override (full URL, no trailing slash). When set it wins over `region`; used by tests/dev to point at a mock server.", + "default": null, + "type": [ + "string", + "null" + ] + }, + "kind": { + "type": "string", + "enum": [ + "aliyun_text_moderation" + ] + }, + "max_buffer_bytes": { + "description": "Max bytes buffered in `buffer_full` mode before `on_buffer_exceeded` applies. Default 262 144.", + "default": 262144, + "type": "integer", + "format": "uint64", + "minimum": 0.0 + }, + "on_buffer_exceeded": { + "description": "`fail_closed` (default) or `fail_open` when the buffer cap is hit.", + "default": "fail_closed", + "type": "string" + }, + "output_fail_open": { + "description": "Fail-open policy for the OUTPUT hook. Defaults `false` (fail-closed) so an Aliyun outage can't release unscanned model output.", + "default": false, + "type": "boolean" + }, + "region": { + "description": "Aliyun region the guardrail lives in, e.g. `cn-shanghai`. The DP builds the endpoint `https://green-cip..aliyuncs.com`.", + "type": "string" + }, + "risk_level_threshold": { + "description": "Minimum risk level that triggers a block: `low`, `medium`, or `high` (default). A returned level at or above this blocks.", + "default": "high", + "type": "string" + }, + "stream_processing_mode": { + "description": "`window` (sliding-window incremental release; default) or `buffer_full` (whole-response hold-back).", + "default": "window", + "type": "string" + }, + "timeout_ms": { + "description": "HTTP call timeout (ms); `fail_open` / `output_fail_open` govern the verdict when it elapses. See the `AzureContentSafetyConfig` note on why `0` means \"fire immediately\", not \"no timeout\".", + "default": 5000, + "type": "integer", + "format": "uint32", + "minimum": 0.0 + }, + "window_overlap_size": { + "description": "Chars carried between windows so a span split across a boundary is still caught. Default 128.", + "default": 128, + "type": "integer", + "format": "uint32", + "minimum": 0.0 + }, + "window_size": { + "description": "Sliding-window size in chars (window mode). Default 2 000 — Aliyun's `llm_response_moderation` per-call content cap.", + "default": 2000, + "type": "integer", + "format": "uint32", + "minimum": 0.0 + } + } } ], "required": [ diff --git a/tests/e2e/src/cases/guardrail-aliyun-e2e.test.ts b/tests/e2e/src/cases/guardrail-aliyun-e2e.test.ts new file mode 100644 index 00000000..58aa4956 --- /dev/null +++ b/tests/e2e/src/cases/guardrail-aliyun-e2e.test.ts @@ -0,0 +1,377 @@ +import { createServer, type Server } from "node:http"; +import { createHash } from "node:crypto"; +import OpenAI, { APIError } from "openai"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + AdminClient, + EtcdClient, + pickFreePort, + spawnApp, + startOpenAiUpstream, + waitConfigPropagation, + type OpenAiUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +// E2E: the `aliyun_text_moderation` guardrail (#603) moderates chat +// input and output against Aliyun's content-safety guardrail +// (`TextModerationPlus`). We stand up a mock green-cip endpoint that +// grades any text containing RISKY_MARKER as RiskLevel "high" and +// everything else "none", point the guardrail's `endpoint` override at +// it, and verify the full DP journey end-to-end with a real `aisix` +// binary + etcd + mock upstream. No control plane involved. +// +// References: +// - Aliyun TextModerationPlus (llm_query_moderation / llm_response_moderation) +// +// - OpenAI / Azure `error.type: "content_filter"` envelope convention. + +const CALLER_PLAINTEXT = "sk-aliyun-e2e-caller"; +const CALLER_KEY_HASH = createHash("sha256") + .update(CALLER_PLAINTEXT) + .digest("hex"); + +// Letters survive percent-encoding inside the signed form body, so a +// plain-letter marker is detectable in the raw request the mock receives. +const RISKY_MARKER = "aliyunriskymarker"; + +interface AliyunMockRequest { + service: string; + sessionId?: string; + content: string; + raw: string; +} + +interface AliyunMock { + baseUrl: string; + requests: AliyunMockRequest[]; + close(): Promise; +} + +// Minimal mock of the green-cip TextModerationPlus RPC endpoint. Parses +// the form-urlencoded body, extracts Service + the ServiceParameters JSON +// (content + sessionId), and returns "high" when the content carries the +// marker. It does NOT verify the signature — signature correctness is +// pinned by a known-vector unit test in the dispatcher crate. +async function startAliyunMock(): Promise { + const requests: AliyunMockRequest[] = []; + const server: Server = createServer((req, res) => { + let raw = ""; + req.on("data", (c: Buffer) => (raw += c.toString("utf8"))); + req.on("end", () => { + const params = new URLSearchParams(raw); + const service = params.get("Service") ?? ""; + let content = ""; + let sessionId: string | undefined; + try { + const sp = JSON.parse(params.get("ServiceParameters") ?? "{}"); + content = typeof sp.content === "string" ? sp.content : ""; + sessionId = typeof sp.sessionId === "string" ? sp.sessionId : undefined; + } catch { + // leave defaults + } + requests.push({ service, sessionId, content, raw }); + + const risky = content.includes(RISKY_MARKER); + res.statusCode = 200; + res.setHeader("content-type", "application/json"); + res.end( + JSON.stringify({ + Code: 200, + Data: { + RiskLevel: risky ? "high" : "none", + Result: [{ Label: risky ? "violent_content" : "nonLabel" }], + }, + RequestId: "mock-req-1", + }), + ); + }); + }); + const port = await pickFreePort(); + await new Promise((resolve) => server.listen(port, "127.0.0.1", resolve)); + return { + baseUrl: `http://127.0.0.1:${port}`, + requests, + async close() { + await new Promise((resolve, reject) => { + server.close((err) => (err ? reject(err) : resolve())); + }); + }, + }; +} + +describe("aliyun guardrail e2e: TextModerationPlus blocks risky input/output", () => { + let app: SpawnedApp | undefined; + let benignUpstream: OpenAiUpstream | undefined; + let riskyOutputUpstream: OpenAiUpstream | undefined; + let streamUpstream: OpenAiUpstream | undefined; + let aliyun: AliyunMock | undefined; + let admin: AdminClient | undefined; + let etcdReachable = false; + + beforeAll(async () => { + etcdReachable = await new EtcdClient().ping(); + if (!etcdReachable) return; + + aliyun = await startAliyunMock(); + + // Clean upstream for the input-side cases (its output is benign so the + // output hook always passes for these models). + benignUpstream = await startOpenAiUpstream({ + nonStreamBody: { + id: "cmpl-clean", + object: "chat.completion", + created: Math.floor(Date.now() / 1000), + model: "gpt-4o-mini", + choices: [ + { + index: 0, + message: { role: "assistant", content: "a safe and clean reply" }, + finish_reason: "stop", + }, + ], + usage: { prompt_tokens: 5, completion_tokens: 4, total_tokens: 9 }, + }, + }); + + // Upstream whose RESPONSE carries the risky marker — the input is + // innocent, so this exercises the output hook. + riskyOutputUpstream = await startOpenAiUpstream({ + nonStreamBody: { + id: "cmpl-risky-out", + object: "chat.completion", + created: Math.floor(Date.now() / 1000), + model: "gpt-4o-mini", + choices: [ + { + index: 0, + message: { role: "assistant", content: `here it is: ${RISKY_MARKER}` }, + finish_reason: "stop", + }, + ], + usage: { prompt_tokens: 5, completion_tokens: 6, total_tokens: 11 }, + }, + }); + + streamUpstream = await startOpenAiUpstream({ + streamEvents: [ + '{"id":"strm-risky","object":"chat.completion.chunk","model":"gpt-4o-mini","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}', + `{"id":"strm-risky","object":"chat.completion.chunk","model":"gpt-4o-mini","choices":[{"index":0,"delta":{"content":"streamed ${RISKY_MARKER} payload"},"finish_reason":null}]}`, + '{"id":"strm-risky","object":"chat.completion.chunk","model":"gpt-4o-mini","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}', + "[DONE]", + ], + eventDelayMs: 50, + }); + + app = await spawnApp(); + admin = new AdminClient(app.adminUrl, app.adminKey); + + const benignPk = await admin.createProviderKey({ + display_name: "aliyun-e2e-pk", + secret: "sk-mock", + api_base: `${benignUpstream.baseUrl}/v1`, + }); + await admin.createModel({ + display_name: "aliyun-e2e", + provider: "openai", + model_name: "gpt-4o-mini", + provider_key_id: benignPk.id, + }); + + const riskyOutPk = await admin.createProviderKey({ + display_name: "aliyun-out-e2e-pk", + secret: "sk-mock", + api_base: `${riskyOutputUpstream.baseUrl}/v1`, + }); + await admin.createModel({ + display_name: "aliyun-out-e2e", + provider: "openai", + model_name: "gpt-4o-mini", + provider_key_id: riskyOutPk.id, + }); + + const streamPk = await admin.createProviderKey({ + display_name: "aliyun-stream-e2e-pk", + secret: "sk-mock", + api_base: `${streamUpstream.baseUrl}/v1`, + }); + await admin.createModel({ + display_name: "aliyun-stream-e2e", + provider: "openai", + model_name: "gpt-4o-mini", + provider_key_id: streamPk.id, + }); + + await admin.createApiKey({ + key_hash: CALLER_KEY_HASH, + allowed_models: ["aliyun-e2e", "aliyun-out-e2e", "aliyun-stream-e2e"], + }); + + // One env-wide guardrail covering input + output. Small window so the + // streaming case triggers a windowed output call (and reuses the + // stream's sessionId across windows). `endpoint` points at the mock. + await admin.json("POST", "/admin/v1/guardrails", { + name: "aliyun-e2e-guard", + enabled: true, + hook_point: "both", + fail_open: false, + kind: "aliyun_text_moderation", + region: "cn-shanghai", + endpoint: aliyun.baseUrl, + access_key_id: "LTAI_E2E", + access_key_secret: "e2e-secret", + risk_level_threshold: "high", + stream_processing_mode: "window", + window_size: 16, + window_overlap_size: 4, + }); + }); + + afterAll(async () => { + await app?.exit(); + await benignUpstream?.close(); + await riskyOutputUpstream?.close(); + await streamUpstream?.close(); + await aliyun?.close(); + }); + + test("risky input → 422 content_filter, upstream never called", async (ctx) => { + if (!etcdReachable || !app || !benignUpstream) { + ctx.skip(); + return; + } + const client = new OpenAI({ + apiKey: CALLER_PLAINTEXT, + baseURL: `${app.proxyUrl}/v1`, + maxRetries: 0, + }); + + // Gate on the guardrail being live: poll with a risky prompt until 422. + await waitConfigPropagation(async () => { + try { + await client.chat.completions.create({ + model: "aliyun-e2e", + messages: [{ role: "user", content: `probe ${RISKY_MARKER}` }], + }); + return false; + } catch (e) { + return e instanceof APIError && e.status === 422; + } + }); + + // Benign request passes and hits the upstream. + const okBefore = benignUpstream.receivedRequests.length; + const clean = await client.chat.completions.create({ + model: "aliyun-e2e", + messages: [{ role: "user", content: "what is a safe and clean topic" }], + }); + expect(clean.choices[0]?.message.role).toBe("assistant"); + expect(benignUpstream.receivedRequests.length).toBe(okBefore + 1); + + // Risky input is blocked BEFORE the upstream is called. + const upstreamBefore = benignUpstream.receivedRequests.length; + let caught: unknown; + try { + await client.chat.completions.create({ + model: "aliyun-e2e", + messages: [{ role: "user", content: `please do ${RISKY_MARKER} now` }], + }); + } catch (e) { + caught = e; + } + expect(caught).toBeInstanceOf(APIError); + if (!(caught instanceof APIError)) throw new Error("unreachable"); + expect(caught.status).toBe(422); + expect((caught.error as { type?: unknown })?.type).toBe("content_filter"); + // The matched content must not leak back to the caller (#153). + expect(JSON.stringify(caught.error ?? {})).not.toContain(RISKY_MARKER); + expect(benignUpstream.receivedRequests.length).toBe(upstreamBefore); + }); + + test("risky model output → 422 content_filter after upstream call", async (ctx) => { + if (!etcdReachable || !app || !riskyOutputUpstream) { + ctx.skip(); + return; + } + const client = new OpenAI({ + apiKey: CALLER_PLAINTEXT, + baseURL: `${app.proxyUrl}/v1`, + maxRetries: 0, + }); + + const upstreamBefore = riskyOutputUpstream.receivedRequests.length; + let caught: unknown; + try { + await client.chat.completions.create({ + model: "aliyun-out-e2e", + messages: [{ role: "user", content: "an innocent question" }], + }); + } catch (e) { + caught = e; + } + expect(caught).toBeInstanceOf(APIError); + if (!(caught instanceof APIError)) throw new Error("unreachable"); + expect(caught.status).toBe(422); + expect((caught.error as { type?: unknown })?.type).toBe("content_filter"); + expect(JSON.stringify(caught.error ?? {})).not.toContain(RISKY_MARKER); + // Output hook runs AFTER the upstream → the upstream IS hit. + expect(riskyOutputUpstream.receivedRequests.length).toBe(upstreamBefore + 1); + + // The dispatcher called Aliyun's output service with a sessionId + // (derived from the response id) so windows of one stream correlate. + const outCall = aliyun!.requests.find( + (r) => r.service === "llm_response_moderation" && r.content.includes(RISKY_MARKER), + ); + expect(outCall, "expected an llm_response_moderation call").toBeDefined(); + expect(outCall?.sessionId, "output call must carry a sessionId").toBeTruthy(); + }); + + test("streaming risky output → SSE error event, no [DONE] (windowed)", async (ctx) => { + if (!etcdReachable || !app || !streamUpstream) { + ctx.skip(); + return; + } + const outBefore = aliyun!.requests.filter( + (r) => r.service === "llm_response_moderation", + ).length; + const res = await fetch(`${app.proxyUrl}/v1/chat/completions`, { + method: "POST", + headers: { + authorization: `Bearer ${CALLER_PLAINTEXT}`, + "content-type": "application/json", + }, + body: JSON.stringify({ + model: "aliyun-stream-e2e", + messages: [{ role: "user", content: "tell me something" }], + stream: true, + }), + }); + + expect(res.status).toBe(200); + const wire = await res.text(); + expect(wire).toContain("event: error"); + expect(wire).not.toContain("data: [DONE]"); + + const errEventIdx = wire.indexOf("event: error\n"); + const afterErr = wire.slice(errEventIdx + "event: error\n".length); + const dataLine = afterErr + .split("\n") + .find((l: string) => l.startsWith("data: ")); + expect(dataLine).toBeDefined(); + const parsed = JSON.parse(dataLine!.slice("data: ".length)) as { + error?: { type?: unknown }; + }; + expect(parsed.error?.type).toBe("content_filter"); + + // Every windowed output call for this stream must carry one stable + // sessionId (the upstream's request id, "strm-risky"), proving the + // chunks of a single response correlate at Aliyun. + const streamOutCalls = aliyun!.requests + .filter((r) => r.service === "llm_response_moderation") + .slice(outBefore); + expect(streamOutCalls.length).toBeGreaterThan(0); + const sessionIds = new Set(streamOutCalls.map((r) => r.sessionId)); + expect(sessionIds.size).toBe(1); + expect([...sessionIds][0]).toBe("strm-risky"); + }); +});