fix: surface RustFS policy apply errors

docs/operator-user-guide.md

@@ -555,6 +555,8 @@ The operator can create RustFS policies, users, and buckets after the Tenant wor
 
 ConfigMaps and user Secrets must live in the Tenant namespace. If managed outside the Operator Console, label them with `rustfs.tenant=<tenant-name>` so updates enqueue the owning Tenant.
 
+Policy documents are parsed by RustFS. Use S3 ARN resource patterns such as `arn:aws:s3:::bucket` and `arn:aws:s3:::bucket/*`; for all buckets, use `arn:aws:s3:::*`. A bare `Resource: "*"` is not accepted by RustFS policy parsing.
+
 For each `spec.users[]` entry, the operator reads a Secret with the same name as the user. The Secret must contain `accesskey` and `secretkey`, or the MinIO-compatible keys `CONSOLE_ACCESS_KEY` and `CONSOLE_SECRET_KEY`. If both key formats are present, their values must match. User access keys must be at least 8 characters and must not contain whitespace, `=`, or `,`; user secret keys must be at least 8 characters.
 
 Updating a user Secret's `secretkey` rotates that RustFS user's credential. The `accesskey` is immutable after the first successful reconciliation; use a new user entry and Secret when it must change, then migrate clients before removing the old entry.

docs/operator-user-guide.zh-CN.md

@@ -557,6 +557,8 @@ Operator 可以在 Tenant workload Ready 后自动创建 RustFS policy、user 
 
 ConfigMap 和 user Secret 必须位于 Tenant namespace。若这些资源不是通过 Operator Console 创建，建议添加 label：`rustfs.tenant=<tenant-name>`，这样资源变化可以触发 owning Tenant reconcile。
 
+Policy document 由 RustFS 解析。请使用 `arn:aws:s3:::bucket` 和 `arn:aws:s3:::bucket/*` 这类 S3 ARN resource 写法；如需匹配所有 bucket，请使用 `arn:aws:s3:::*`。RustFS policy parser 不接受裸 `Resource: "*"`。
+
 每个 `spec.users[]` 条目都会读取一个与 user 名同名的 Secret。Secret 必须包含 `accesskey` 和 `secretkey`，或者 MinIO 兼容 key：`CONSOLE_ACCESS_KEY` 和 `CONSOLE_SECRET_KEY`。如果两种 key 同时存在，值必须一致。user access key 至少 8 个字符，且不能包含空白、`=` 或 `,`；user secret key 至少 8 个字符。
 
 更新 user Secret 的 `secretkey` 会轮换对应 RustFS user 的凭据。首次成功 provisioning 后，`accesskey` 不可变；如需变更，请新建 user 条目和 Secret，迁移客户端后再移除旧条目。

src/reconcile/provisioning.rs

@@ -1520,6 +1520,42 @@ mod tests {
         server.abort();
     }
 
+    #[tokio::test]
+    async fn apply_policy_includes_upstream_policy_parse_error() {
+        let router = Router::new().route(
+            "/rustfs/admin/v3/add-canned-policy",
+            put(|| async {
+                (
+                    StatusCode::BAD_REQUEST,
+                    r#"<Error><Code>InvalidRequest</Code><Message>invalid resource: unknown &quot;*&quot;</Message></Error>"#,
+                )
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
+            .await
+            .expect("test server should bind");
+        let addr = listener.local_addr().expect("listener should have address");
+        let server = tokio::spawn(async move {
+            axum::serve(listener, router)
+                .await
+                .expect("test server should serve")
+        });
+        let client =
+            RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret");
+        let mut live_policies = BTreeMap::new();
+        let raw = r#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:*","Resource":"*"}]}"#;
+
+        let error = apply_policy(&client, &mut live_policies, "tenant-policy", raw)
+            .await
+            .expect_err("RustFS policy parse error should fail provisioning");
+
+        assert_eq!(
+            error,
+            r#"failed to apply RustFS policy 'tenant-policy': upstream returned 400 Bad Request: InvalidRequest: invalid resource: unknown "*""#
+        );
+        server.abort();
+    }
+
     #[tokio::test]
     async fn rotated_user_secret_is_upserted_for_existing_user() {
         let capture = UserCredentialCapture::default();

src/status.rs

@@ -19,6 +19,7 @@ use crate::types::v1alpha1::status::{
     pool, summarize_current_state,
 };
 use crate::types::v1alpha1::tenant::Tenant;
+use crate::utils::sanitize::redact_sensitive_pairs;
 use kube::runtime::events::EventType;
 
 const LEGACY_PROGRESSING_CONDITION: &str = "Progressing";
@@ -593,144 +594,6 @@ fn sanitize_message(message: &str) -> String {
     redact_sensitive_pairs(message)
 }
 
-fn redact_sensitive_pairs(message: &str) -> String {
-    const SENSITIVE_KEYS: [&str; 4] = ["token", "password", "accesskey", "secretkey"];
-
-    fn is_sensitive_key(key: &str) -> bool {
-        matches!(key, "token" | "password" | "accesskey" | "secretkey")
-    }
-
-    fn normalize_key(raw: &str) -> String {
-        raw.trim_matches(|ch: char| !ch.is_ascii_alphanumeric() && ch != '-' && ch != '_')
-            .to_ascii_lowercase()
-    }
-
-    fn parse_value_end(input: &str, start: usize) -> usize {
-        if start >= input.len() {
-            return start;
-        }
-
-        let mut chars = input[start..].char_indices();
-        let Some((_, first)) = chars.next() else {
-            return start;
-        };
-        if first == '"' || first == '\'' {
-            let mut previous = first;
-            for (offset, ch) in chars {
-                if ch == first && previous != '\\' {
-                    return start + offset + ch.len_utf8();
-                }
-                previous = ch;
-            }
-            return input.len();
-        }
-
-        for (offset, ch) in input[start..].char_indices() {
-            if ch.is_whitespace() || matches!(ch, ',' | ';' | '}' | ']' | ')') {
-                return start + offset;
-            }
-        }
-        input.len()
-    }
-
-    fn skip_whitespace(input: &str, start: usize) -> usize {
-        for (offset, ch) in input[start..].char_indices() {
-            if !ch.is_whitespace() {
-                return start + offset;
-            }
-        }
-        input.len()
-    }
-
-    fn matches_key_at(message: &str, start: usize, key: &str) -> bool {
-        let end = start + key.len();
-        end <= message.len()
-            && message.is_char_boundary(start)
-            && message.is_char_boundary(end)
-            && message[start..end].eq_ignore_ascii_case(key)
-    }
-
-    fn redacted_value(original: &str) -> String {
-        if original.len() >= 2 {
-            let bytes = original.as_bytes();
-            let first = bytes[0];
-            let last = bytes[bytes.len() - 1];
-            if (first == b'"' && last == b'"') || (first == b'\'' && last == b'\'') {
-                let quote = first as char;
-                return format!("{quote}<redacted>{quote}");
-            }
-        }
-        "<redacted>".to_string()
-    }
-
-    let bytes = message.as_bytes();
-    let mut output = String::with_capacity(message.len());
-    let mut cursor = 0usize;
-
-    while cursor < bytes.len() {
-        let mut matched = false;
-
-        for key in SENSITIVE_KEYS {
-            let key_len = key.len();
-
-            let unquoted_match = matches_key_at(message, cursor, key);
-            let quoted_match = cursor + key_len + 2 <= bytes.len()
-                && matches!(bytes[cursor] as char, '"' | '\'')
-                && bytes[cursor + key_len + 1] == bytes[cursor]
-                && matches_key_at(message, cursor + 1, key);
-
-            let (key_start, key_end, cursor_after_key) = if unquoted_match {
-                if cursor > 0 {
-                    let prev = bytes[cursor - 1] as char;
-                    if prev.is_ascii_alphanumeric() || prev == '_' || prev == '-' {
-                        continue;
-                    }
-                }
-                (cursor, cursor + key_len, cursor + key_len)
-            } else if quoted_match {
-                let key_start = cursor + 1;
-                (key_start, key_start + key_len, key_start + key_len + 1)
-            } else {
-                continue;
-            };
-
-            let candidate = &message[key_start..key_end];
-
-            let sep_index = skip_whitespace(message, cursor_after_key);
-            if sep_index >= bytes.len() || !matches!(bytes[sep_index] as char, '=' | ':') {
-                continue;
-            }
-
-            let value_start = skip_whitespace(message, sep_index + 1);
-            let value_end = parse_value_end(message, value_start);
-            if value_end <= value_start {
-                continue;
-            }
-
-            let normalized = normalize_key(candidate);
-            if !is_sensitive_key(&normalized) {
-                continue;
-            }
-
-            output.push_str(&message[cursor..value_start]);
-            output.push_str(&redacted_value(&message[value_start..value_end]));
-            cursor = value_end;
-            matched = true;
-            break;
-        }
-
-        if !matched {
-            let Some(ch) = message[cursor..].chars().next() else {
-                break;
-            };
-            output.push(ch);
-            cursor += ch.len_utf8();
-        }
-    }
-
-    output
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

src/sts/admin_ops.rs

@@ -61,7 +61,7 @@ impl RustfsAdminClient {
             .map_err(|_| RustfsClientError::RequestFailed)?;
 
         if !response.status().is_success() {
-            return Err(RustfsClientError::UnexpectedStatus(response.status()));
+            return Err(RustfsClientError::unexpected_response(response).await);
         }
 
         let body = response
@@ -112,7 +112,7 @@ impl RustfsAdminClient {
             .map_err(|_| RustfsClientError::RequestFailed)?;
 
         if !response.status().is_success() {
-            return Err(RustfsClientError::UnexpectedStatus(response.status()));
+            return Err(RustfsClientError::unexpected_response(response).await);
         }
 
         Ok(())
@@ -179,12 +179,14 @@ impl RustfsAdminClient {
         }
 
         let status = response.status();
-        let body = response.text().await.unwrap_or_default();
+        let (body, truncated) = RustfsClientError::limited_response_body(response).await;
         if status == StatusCode::NOT_FOUND || body_mentions_not_found(&body) {
             return Ok(false);
         }
 
-        Err(RustfsClientError::UnexpectedStatus(status))
+        Err(RustfsClientError::unexpected_status_with_limited_body(
+            status, &body, truncated,
+        ))
     }
 
     pub async fn add_user(

src/sts/core_ops.rs

@@ -73,7 +73,7 @@ impl RustfsAdminClient {
             .map_err(|_| RustfsClientError::RequestFailed)?;
 
         if !response.status().is_success() {
-            return Err(RustfsClientError::UnexpectedStatus(response.status()));
+            return Err(RustfsClientError::unexpected_response(response).await);
         }
 
         response