feat(guardrails): Aliyun AI Guardrails kind (MultiModalGuard) with suggestion verdicts and mask write-back

crates/aisix-admin/src/openapi.rs

@@ -4266,6 +4266,7 @@ fn add_variant_titles(doc: &mut Value) {
                 "Azure AI Content Safety Prompt Shield",
                 "Azure AI Content Safety Text Moderation",
                 "Aliyun Text Moderation",
+                "Aliyun AI Guardrails",
                 "PII Detection and Redaction",
                 "Lakera Guard",
                 "OpenAI Moderation",

crates/aisix-core/src/models/guardrail.rs

@@ -339,6 +339,87 @@ fn default_aliyun_risk_level_threshold() -> String {
     "high".to_owned()
 }
 
+/// Config block for `kind: "aliyun_ai_guardrail"`. Calls Aliyun's AI
+/// Guardrails product (AI 安全护栏, action `MultiModalGuard`, version
+/// `2022-03-02`) on the `green-cip.<region>.aliyuncs.com` endpoint for
+/// policy-driven moderation on input and/or output (including streaming
+/// output).
+///
+/// A DIFFERENT Aliyun product from `aliyun_text_moderation`
+/// (TextModerationPlus / Content Moderation): AI Guardrails is activated,
+/// billed, and policy-configured separately (commodity
+/// `lvwang_guardrail_public_cn`), and its calls appear in the AI
+/// Guardrails console. The input hook uses the `query_security_check_pro`
+/// service code (`query_security_check` at `service_level: "basic"`), the
+/// output hook `response_security_check_pro` / `response_security_check`.
+///
+/// Verdicts follow the returned `Data.Suggestion`, which Aliyun computes
+/// from the check/block policies configured in its console — there is no
+/// local risk threshold. `block` blocks; anything else passes (detection
+/// detail lands in logs/telemetry).
+///
+/// Only `access_key_secret` is a secret (decrypted by cp-api before kine
+/// projection. It is plaintext in DP memory only and never logged). Every
+/// other field travels in the clear.
+#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct AliyunAiGuardrailConfig {
+    /// Aliyun region the guardrail lives in, e.g. `cn-shanghai`. The data plane
+    /// builds the endpoint `https://green-cip.<region>.aliyuncs.com`.
+    #[schemars(length(min = 1))]
+    pub region: String,
+    /// Explicit endpoint override as a full URL with no trailing slash. When set, it takes precedence over `region`.
+    #[serde(default)]
+    #[schemars(length(min = 1))]
+    pub endpoint: Option<String>,
+    /// Aliyun AccessKey ID.
+    #[schemars(length(min = 1))]
+    pub access_key_id: String,
+    /// Aliyun AccessKey secret. Decrypted before projection. Plaintext is held
+    /// in memory only and is not logged. Used to sign the request.
+    #[schemars(length(min = 1))]
+    pub access_key_secret: String,
+    /// Which AI Guardrails service tier to call: `pro` uses
+    /// `query_security_check_pro` / `response_security_check_pro`, `basic`
+    /// uses `query_security_check` / `response_security_check`. Must match
+    /// the tier activated on the Aliyun account.
+    #[serde(default = "default_aliyun_aig_service_level")]
+    pub service_level: String,
+    /// HTTP call timeout in milliseconds. `fail_open` and `output_fail_open`
+    /// govern the verdict when it elapses. A value of `0` triggers the timeout
+    /// immediately.
+    #[serde(default = "default_acs_timeout_ms")]
+    #[schemars(range(max = 4_294_967_295u32))]
+    pub timeout_ms: u32,
+    /// Fail-open policy for the output hook. When disabled, an Aliyun outage does not release unscanned model output.
+    #[serde(default)]
+    pub output_fail_open: bool,
+
+    // --- streaming-output controls (consumed by aisix-proxy build_sse_stream) ---
+    /// Streaming output moderation mode: sliding-window incremental release
+    /// or whole-response hold-back.
+    #[serde(default = "default_acs_stream_processing_mode")]
+    pub stream_processing_mode: String,
+    /// Sliding-window size in characters when window mode is used. Aliyun limits each MultiModalGuard call to 2,000 characters of text.
+    #[serde(default = "default_aliyun_window_size")]
+    #[schemars(range(min = 1, max = 2_000))]
+    pub window_size: u32,
+    /// Chars carried between windows so a span split across a boundary is still caught.
+    #[serde(default = "default_aliyun_window_overlap_size")]
+    pub window_overlap_size: u32,
+    /// Max bytes buffered in `buffer_full` mode before `on_buffer_exceeded` applies.
+    #[serde(default = "default_acs_max_buffer_bytes")]
+    #[schemars(range(min = 1))]
+    pub max_buffer_bytes: u64,
+    /// Buffer-overflow policy for streamed output when the buffer cap is hit.
+    #[serde(default = "default_acs_on_buffer_exceeded")]
+    pub on_buffer_exceeded: String,
+}
+
+fn default_aliyun_aig_service_level() -> String {
+    "pro".to_owned()
+}
+
 /// One built-in detector selection for `kind: "pii"`. The `type` names the
 /// detector to enable; `action` optionally overrides the guardrail-level
 /// `default_action` for this detector only.
@@ -662,6 +743,11 @@ pub enum GuardrailKind {
     /// `TextModerationPlus` action on `green-cip.<region>.aliyuncs.com`, on
     /// input and/or output, including streaming output.
     AliyunTextModeration(AliyunTextModerationConfig),
+    /// Aliyun AI Guardrails (AI 安全护栏). Policy-driven moderation via the
+    /// `MultiModalGuard` action on `green-cip.<region>.aliyuncs.com`; the
+    /// verdict follows the console-configured policy's `Suggestion`. On
+    /// input and/or output, including streaming output.
+    AliyunAiGuardrail(AliyunAiGuardrailConfig),
     /// Built-in sensitive-data detection and redaction inside AISIX. Built-in
     /// detectors and custom regex patterns can mask matched spans or block
     /// traffic on input, output, or both, including buffered streaming output.
@@ -696,6 +782,7 @@ impl GuardrailKind {
                 "azure_content_safety_text_moderation"
             }
             GuardrailKind::AliyunTextModeration(_) => "aliyun_text_moderation",
+            GuardrailKind::AliyunAiGuardrail(_) => "aliyun_ai_guardrail",
             GuardrailKind::Pii(_) => "pii",
             GuardrailKind::Lakera(_) => "lakera",
             GuardrailKind::OpenaiModeration(_) => "openai_moderation",
@@ -1260,6 +1347,68 @@ mod tests {
         }
     }
 
+    #[test]
+    fn aliyun_ai_guardrail_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": "aig-guard",
+            "kind": "aliyun_ai_guardrail",
+            "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::AliyunAiGuardrail(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.service_level, "pro");
+                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 AliyunAiGuardrail variant"),
+        }
+    }
+
+    #[test]
+    fn aliyun_ai_guardrail_kind_round_trips_set_fields() {
+        let v = json!({
+            "name": "aig-guard",
+            "kind": "aliyun_ai_guardrail",
+            "region": "cn-beijing",
+            "endpoint": "http://127.0.0.1:8080",
+            "access_key_id": "id",
+            "access_key_secret": "secret",
+            "service_level": "basic",
+            "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::AliyunAiGuardrail(ref c) => {
+                assert_eq!(c.endpoint.as_deref(), Some("http://127.0.0.1:8080"));
+                assert_eq!(c.service_level, "basic");
+                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 AliyunAiGuardrail variant"),
+        }
+    }
+
     #[test]
     fn pii_kind_parses_with_defaults() {
         // cp-api omits unset fields (omitempty); the DP must apply the

crates/aisix-core/src/models/mod.rs

@@ -38,12 +38,12 @@ pub use cache_policy::{AppliesTo, CacheBackend, CachePolicy};
 pub use embedding::EmbeddingConfig;
 pub use ensemble::{EnsembleConfig, Judge, PanelMember};
 pub use guardrail::{
-    AliyunTextModerationConfig, AppliedGuardrail, AzureContentSafetyConfig,
-    AzureContentSafetyTextModerationConfig, BedrockAWSCredentials, BedrockConfig,
-    BedrockLatencyMode, Guardrail, GuardrailAttachment, GuardrailExecution, GuardrailHookPoint,
-    GuardrailKind, GuardrailMetricsSink, GuardrailMonitorHit, GuardrailScopeType, KeywordConfig,
-    KeywordPattern, LakeraConfig, OpenaiModerationConfig, PiiConfig, PiiCustomPattern,
-    PiiDetectorConfig, PresidioConfig, PresidioEntityConfig,
+    AliyunAiGuardrailConfig, AliyunTextModerationConfig, AppliedGuardrail,
+    AzureContentSafetyConfig, AzureContentSafetyTextModerationConfig, BedrockAWSCredentials,
+    BedrockConfig, BedrockLatencyMode, Guardrail, GuardrailAttachment, GuardrailExecution,
+    GuardrailHookPoint, GuardrailKind, GuardrailMetricsSink, GuardrailMonitorHit,
+    GuardrailScopeType, KeywordConfig, KeywordPattern, LakeraConfig, OpenaiModerationConfig,
+    PiiConfig, PiiCustomPattern, PiiDetectorConfig, PresidioConfig, PresidioEntityConfig,
 };
 pub use mcp_server::{McpAuthType, McpServer, McpTransport};
 pub use model::{

crates/aisix-guardrails/src/aliyun.rs

@@ -386,7 +386,7 @@ impl AliyunTextModerationGuardrail {
 ///
 /// Only ever held transiently to pull one symbolic token out; nothing of it is
 /// logged.
-const MAX_ERROR_BODY_PARSE_BYTES: usize = 64 * 1024;
+pub(crate) const MAX_ERROR_BODY_PARSE_BYTES: usize = 64 * 1024;
 
 /// Best-effort pull of the error `Code` out of an Aliyun error body.
 ///
@@ -400,7 +400,7 @@ const MAX_ERROR_BODY_PARSE_BYTES: usize = 64 * 1024;
 /// The XML arm is a substring scan rather than a parser: one tag out of a
 /// fixed vendor envelope doesn't justify an XML dependency, and a wrong guess
 /// costs a blank log field, not a leak.
-fn extract_error_code(body: &str) -> String {
+pub(crate) fn extract_error_code(body: &str) -> String {
     if let Ok(v) = serde_json::from_str::<serde_json::Value>(body) {
         if let Some(code) = v.get("Code") {
             // A string on this path; render a stray number rather than drop it.
@@ -436,7 +436,7 @@ fn risk_rank(level: &str) -> u8 {
 /// (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 {
+pub(crate) fn percent_encode(s: &str) -> String {
     let mut out = String::with_capacity(s.len() * 3);
     for b in s.bytes() {
         match b {
@@ -455,7 +455,7 @@ fn percent_encode(s: &str) -> String {
 /// 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 {
+pub(crate) 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() {
@@ -473,7 +473,10 @@ fn string_to_sign(params: &std::collections::BTreeMap<&str, String>) -> String {
 }
 
 /// Compute the RPC v1 signature: `Base64(HMAC-SHA1(secret + "&", StringToSign))`.
-fn sign(params: &std::collections::BTreeMap<&str, String>, access_key_secret: &str) -> String {
+pub(crate) 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 =
@@ -485,7 +488,7 @@ fn sign(params: &std::collections::BTreeMap<&str, String>, access_key_secret: &s
 /// Failure cause buckets. `bypass_tag()` maps to the strings stored in
 /// `usage_events.guardrail_bypassed_reason`.
 #[derive(Debug)]
-enum AliyunFailure {
+pub(crate) enum AliyunFailure {
     Timeout,
     Throttled,
     IoError,
@@ -499,7 +502,7 @@ enum AliyunFailure {
 }
 
 impl AliyunFailure {
-    fn bypass_tag(&self) -> &'static str {
+    pub(crate) fn bypass_tag(&self) -> &'static str {
         match self {
             Self::Timeout => "aliyun_timeout",
             Self::Throttled => "aliyun_throttled",
@@ -516,7 +519,7 @@ impl AliyunFailure {
 /// XML rather than JSON. It is therefore a strictly more reliable source
 /// for the id than the body's `RequestId`, which is unreachable exactly
 /// when the body doesn't parse.
-const ACS_REQUEST_ID_HEADER: &str = "x-acs-request-id";
+pub(crate) const ACS_REQUEST_ID_HEADER: &str = "x-acs-request-id";
 
 /// What one `TextModerationPlus` call reported about itself, for operator
 /// triage (AISIX-Cloud#1060).
@@ -970,6 +973,8 @@ mod tests {
         F: FnOnce() -> Fut,
         Fut: std::future::Future<Output = ()>,
     {
+        // One capture test at a time, process-wide (see TRACING_CAPTURE_LOCK).
+        let _capture_guard = crate::TRACING_CAPTURE_LOCK.lock().await;
         let buf = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
         let subscriber = tracing_subscriber::fmt()
             .with_ansi(false)