diff --git a/docs/operator-user-guide.md b/docs/operator-user-guide.md index 556464a..5cd46bd 100644 --- a/docs/operator-user-guide.md +++ b/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=` 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. diff --git a/docs/operator-user-guide.zh-CN.md b/docs/operator-user-guide.zh-CN.md index 3328d9f..58fa598 100644 --- a/docs/operator-user-guide.zh-CN.md +++ b/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=`，这样资源变化可以触发 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，迁移客户端后再移除旧条目。 diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs index 12bb2f8..275bcc2 100644 --- a/src/reconcile/provisioning.rs +++ b/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#"InvalidRequestinvalid resource: unknown "*""#, + ) + }), + ); + 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(); diff --git a/src/status.rs b/src/status.rs index 8c30b4b..a6ea308 100644 --- a/src/status.rs +++ b/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}{quote}"); - } - } - "".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::*; diff --git a/src/sts/admin_ops.rs b/src/sts/admin_ops.rs index b08efb8..707f2bd 100644 --- a/src/sts/admin_ops.rs +++ b/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( diff --git a/src/sts/core_ops.rs b/src/sts/core_ops.rs index 9db2fdc..ffdb9ae 100644 --- a/src/sts/core_ops.rs +++ b/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 diff --git a/src/sts/rustfs_client.rs b/src/sts/rustfs_client.rs index 9839bff..3f0aa60 100644 --- a/src/sts/rustfs_client.rs +++ b/src/sts/rustfs_client.rs @@ -16,9 +16,10 @@ use std::{collections::BTreeMap, time::Duration}; use k8s_openapi::api::core::v1 as corev1; use kube::{Api, Client}; -use reqwest::{Certificate, Client as HttpClient, StatusCode}; +use reqwest::{Certificate, Client as HttpClient, Response, StatusCode}; use crate::Tenant; +use crate::utils::sanitize::redact_sensitive_pairs; /// admin_ops: tenant admin operations (user/policy APIs). #[path = "admin_ops.rs"] @@ -57,6 +58,8 @@ const ADMIN_SIGNING_SERVICE: &str = "s3"; const STS_SIGNING_SERVICE: &str = "sts"; const ADMIN_HTTP_CONNECT_TIMEOUT: Duration = Duration::from_secs(3); const ADMIN_HTTP_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +const MAX_UPSTREAM_ERROR_BODY_BYTES: usize = 8 * 1024; +const MAX_UPSTREAM_ERROR_DETAIL_CHARS: usize = 512; /// Credentials read from Tenant `.spec.credsSecret`. #[derive(Debug, Clone, PartialEq, Eq)] @@ -173,22 +176,36 @@ pub struct RustfsErasureSetInfo { pub enum RustfsClientError { MissingTenantNamespace, MissingCredsSecret, - MissingCredentialKey { key: &'static str }, - EmptyCredentialValue { key: &'static str }, - InvalidCredentialValue { key: &'static str }, + MissingCredentialKey { + key: &'static str, + }, + EmptyCredentialValue { + key: &'static str, + }, + InvalidCredentialValue { + key: &'static str, + }, TenantSecretLookupFailed, InvalidPolicyName, InvalidPolicyDocument, TenantTlsRequired, TenantTlsNotReady, TenantTlsClientCertificateRequired, - MissingTenantTlsCaKey { secret: String, key: String }, - TenantTlsCaSecretLookupFailed { secret: String }, + MissingTenantTlsCaKey { + secret: String, + key: String, + }, + TenantTlsCaSecretLookupFailed { + secret: String, + }, InvalidTenantTlsCa, TlsClientBuildFailed, RequestBuildFailed, RequestFailed, - UnexpectedStatus(StatusCode), + UnexpectedStatus { + status: StatusCode, + detail: Option, + }, ParseResponseFailed, SigningFailed, } @@ -223,7 +240,13 @@ impl std::fmt::Display for RustfsClientError { Self::TlsClientBuildFailed => write!(f, "failed to build TLS HTTP client"), Self::RequestBuildFailed => write!(f, "failed to construct request"), Self::RequestFailed => write!(f, "request failed"), - Self::UnexpectedStatus(status) => write!(f, "upstream returned {status}"), + Self::UnexpectedStatus { status, detail } => { + write!(f, "upstream returned {status}")?; + if let Some(detail) = detail { + write!(f, ": {detail}")?; + } + Ok(()) + } Self::ParseResponseFailed => write!(f, "failed to parse AssumeRole response"), Self::SigningFailed => write!(f, "failed to compute request signature"), } @@ -232,6 +255,146 @@ impl std::fmt::Display for RustfsClientError { impl std::error::Error for RustfsClientError {} +impl RustfsClientError { + pub(super) async fn unexpected_response(response: Response) -> Self { + let status = response.status(); + let (body, truncated) = read_limited_response_body(response).await; + Self::unexpected_status_with_limited_body(status, &body, truncated) + } + + pub(super) async fn limited_response_body(response: Response) -> (String, bool) { + read_limited_response_body(response).await + } + + fn unexpected_status_with_limited_body( + status: StatusCode, + body: &str, + body_truncated: bool, + ) -> Self { + Self::UnexpectedStatus { + status, + detail: summarize_upstream_error_body(body, body_truncated), + } + } + + #[cfg(test)] + pub(super) fn unexpected_status_with_body(status: StatusCode, body: &str) -> Self { + Self::unexpected_status_with_limited_body(status, body, false) + } +} + +async fn read_limited_response_body(mut response: Response) -> (String, bool) { + let mut body = Vec::new(); + let read_limit = MAX_UPSTREAM_ERROR_BODY_BYTES.saturating_add(1); + + loop { + let remaining = read_limit.saturating_sub(body.len()); + if remaining == 0 { + break; + } + + let chunk = match response.chunk().await { + Ok(Some(chunk)) => chunk, + Ok(None) => break, + Err(_) => break, + }; + if chunk.len() > remaining { + body.extend_from_slice(&chunk[..remaining]); + break; + } + body.extend_from_slice(&chunk); + } + + let truncated = body.len() > MAX_UPSTREAM_ERROR_BODY_BYTES; + if truncated { + body.truncate(MAX_UPSTREAM_ERROR_BODY_BYTES); + } + + (String::from_utf8_lossy(&body).into_owned(), truncated) +} + +fn summarize_upstream_error_body(body: &str, body_truncated: bool) -> Option { + let body = body.trim(); + if body.is_empty() { + return None; + } + + if let Some(message) = helpers::extract_xml_tag(body, "Message") { + let message = decode_basic_xml_entities(&message); + let detail = match helpers::extract_xml_tag(body, "Code") { + Some(code) if !code.trim().is_empty() => { + format!("{}: {message}", decode_basic_xml_entities(&code)) + } + _ => message, + }; + return Some(sanitize_error_detail(&detail)); + } + + if let Ok(value) = serde_json::from_str::(body) + && let Some(detail) = summarize_json_error(&value) + { + return Some(sanitize_error_detail(&detail)); + } + + if body_truncated { + return Some(format!( + "response body exceeded {MAX_UPSTREAM_ERROR_BODY_BYTES} bytes" + )); + } + + Some(sanitize_error_detail(body)) +} + +fn summarize_json_error(value: &serde_json::Value) -> Option { + if let Some(message) = value.as_str() { + return Some(message.to_string()); + } + + let object = value.as_object()?; + let message = ["message", "Message", "error", "Error"] + .iter() + .find_map(|key| object.get(*key).and_then(serde_json::Value::as_str))?; + let code = ["code", "Code"] + .iter() + .find_map(|key| object.get(*key).and_then(serde_json::Value::as_str)); + + Some(match code { + Some(code) if !code.trim().is_empty() => format!("{code}: {message}"), + _ => message.to_string(), + }) +} + +fn collapse_whitespace(value: &str) -> String { + value.split_whitespace().collect::>().join(" ") +} + +fn sanitize_error_detail(value: &str) -> String { + let detail = collapse_whitespace(value); + let detail = redact_sensitive_pairs(&detail); + truncate_error_detail(detail) +} + +fn truncate_error_detail(value: String) -> String { + let mut truncated = String::new(); + for (index, ch) in value.chars().enumerate() { + if index >= MAX_UPSTREAM_ERROR_DETAIL_CHARS { + truncated.push_str("..."); + return truncated; + } + truncated.push(ch); + } + truncated +} + +fn decode_basic_xml_entities(value: &str) -> String { + value + .replace("<", "<") + .replace(">", ">") + .replace(""", "\"") + .replace("'", "'") + .replace("&", "&") +} + #[derive(Debug)] struct SignedRequest { amz_date: String, diff --git a/src/sts/s3_ops.rs b/src/sts/s3_ops.rs index 531fa5e..e49eb1e 100644 --- a/src/sts/s3_ops.rs +++ b/src/sts/s3_ops.rs @@ -81,12 +81,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 bucket_already_exists(status, &body) { return Ok(CreateBucketResult::AlreadyExists); } - Err(RustfsClientError::UnexpectedStatus(status)) + Err(RustfsClientError::unexpected_status_with_limited_body( + status, &body, truncated, + )) } pub async fn bucket_object_lock_enabled( @@ -119,11 +121,13 @@ impl RustfsAdminClient { if !response.status().is_success() { 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); } - return Err(RustfsClientError::UnexpectedStatus(status)); + return Err(RustfsClientError::unexpected_status_with_limited_body( + status, &body, truncated, + )); } let body = response diff --git a/src/sts/sts_ops.rs b/src/sts/sts_ops.rs index 21c10b3..f30cc20 100644 --- a/src/sts/sts_ops.rs +++ b/src/sts/sts_ops.rs @@ -71,7 +71,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 diff --git a/src/sts/tests.rs b/src/sts/tests.rs index 6f99fe7..dc98d04 100644 --- a/src/sts/tests.rs +++ b/src/sts/tests.rs @@ -27,9 +27,9 @@ use std::{collections::BTreeMap, sync::Arc}; use tokio::sync::Mutex; use super::{ - ADD_USER_PATH, CreateBucketResult, LIST_CANNED_POLICIES_PATH, POOLS_DECOMMISSION_PATH, - POOLS_LIST_PATH, POOLS_STATUS_PATH, RustfsAdminClient, RustfsClientError, SERVER_INFO_PATH, - SET_POLICY_PATH, + ADD_USER_PATH, CreateBucketResult, LIST_CANNED_POLICIES_PATH, MAX_UPSTREAM_ERROR_BODY_BYTES, + POOLS_DECOMMISSION_PATH, POOLS_LIST_PATH, POOLS_STATUS_PATH, RustfsAdminClient, + RustfsClientError, SERVER_INFO_PATH, SET_POLICY_PATH, USER_INFO_PATH, helpers::{extract_canned_policy_document, extract_credentials, parse_assume_role_response}, }; @@ -45,6 +45,15 @@ fn secret_with_fields(fields: Vec<(&str, &[u8])>) -> corev1::Secret { } } +fn assert_oversized_upstream_body_hidden(err: RustfsClientError) { + assert_eq!( + err.to_string(), + format!( + "upstream returned 502 Bad Gateway: response body exceeded {MAX_UPSTREAM_ERROR_BODY_BYTES} bytes" + ) + ); +} + #[test] fn extract_credentials_reports_missing_access_key() { let secret = secret_with_fields(vec![("secretkey", b"sekret")]); @@ -114,6 +123,125 @@ fn parse_assume_role_xml_success_and_failure() { assert!(parse_assume_role_response("").is_none()); } +#[test] +fn unexpected_status_includes_upstream_xml_error_summary() { + let err = RustfsClientError::unexpected_status_with_body( + StatusCode::BAD_REQUEST, + r#"InvalidRequestinvalid resource: unknown "*"abc"#, + ); + + let message = err.to_string(); + assert_eq!( + message, + r#"upstream returned 400 Bad Request: InvalidRequest: invalid resource: unknown "*""# + ); + assert!(!message.contains("")); +} + +#[test] +fn unexpected_status_includes_upstream_json_error_summary() { + let err = RustfsClientError::unexpected_status_with_body( + StatusCode::BAD_REQUEST, + r#"{"code":"InvalidRequest","message":"policy Resource must use ARN form"}"#, + ); + + assert_eq!( + err.to_string(), + "upstream returned 400 Bad Request: InvalidRequest: policy Resource must use ARN form" + ); +} + +#[test] +fn unexpected_status_redacts_sensitive_upstream_error_summary() { + let err = RustfsClientError::unexpected_status_with_body( + StatusCode::BAD_REQUEST, + r#"{"code":"InvalidRequest","message":"secretkey: SK_TEST clientSecret: oidc-secret AKIA_XML"}"#, + ); + + let message = err.to_string(); + assert!(message.contains("secretkey: ")); + assert!(message.contains("clientSecret: ")); + assert!(message.contains("")); + assert!(!message.contains("SK_TEST")); + assert!(!message.contains("oidc-secret")); + assert!(!message.contains("AKIA_XML")); +} + +#[test] +fn unexpected_status_hides_truncated_unstructured_response_body() { + let retained_body = "x".repeat(MAX_UPSTREAM_ERROR_BODY_BYTES); + let err = RustfsClientError::unexpected_status_with_limited_body( + StatusCode::BAD_GATEWAY, + &retained_body, + true, + ); + + assert_eq!( + err.to_string(), + format!( + "upstream returned 502 Bad Gateway: response body exceeded {MAX_UPSTREAM_ERROR_BODY_BYTES} bytes" + ) + ); +} + +#[tokio::test] +async fn unexpected_response_preserves_exact_limit_unstructured_response_body() { + let body = "x".repeat(MAX_UPSTREAM_ERROR_BODY_BYTES); + let router = Router::new().route( + ADD_USER_PATH, + put(move || { + let body = body.clone(); + async move { (StatusCode::BAD_GATEWAY, body) } + }), + ); + + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let err = client + .add_user("app-user", "secret123") + .await + .expect_err("exact limit body should still report the retained body"); + + let message = err.to_string(); + assert!(message.contains("upstream returned 502 Bad Gateway")); + assert!(!message.contains("response body exceeded")); + + server.abort(); +} + +#[tokio::test] +async fn unexpected_response_hides_over_limit_unstructured_response_body() { + let body = "x".repeat(MAX_UPSTREAM_ERROR_BODY_BYTES + 1); + let router = Router::new().route( + ADD_USER_PATH, + put(move || { + let body = body.clone(); + async move { (StatusCode::BAD_GATEWAY, body) } + }), + ); + + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let err = client + .add_user("app-user", "secret123") + .await + .expect_err("oversized body should be hidden"); + + assert_oversized_upstream_body_hidden(err); + + server.abort(); +} + #[derive(Clone, Default)] struct Capture { path: Arc>, @@ -383,6 +511,38 @@ async fn add_canned_policy_uses_expected_path_query_body_and_admin_signing() { server.abort(); } +#[tokio::test] +async fn add_canned_policy_reports_upstream_policy_parse_error() { + let router = Router::new().route( + "/rustfs/admin/v3/add-canned-policy", + put(|| async { + ( + StatusCode::BAD_REQUEST, + r#"InvalidRequestinvalid resource: unknown "*""#, + ) + }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let policy = r#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:*","Resource":"*"}]}"#; + let err = client + .add_canned_policy("tenant-policy", policy) + .await + .expect_err("invalid RustFS policy should include upstream parse details"); + + let message = err.to_string(); + assert!(message.contains("upstream returned 400 Bad Request")); + assert!(message.contains(r#"InvalidRequest: invalid resource: unknown "*""#)); + assert!(!message.contains("")); + + server.abort(); +} + #[tokio::test] async fn server_info_uses_expected_path_and_parses_health_fields() { let capture = Capture::default(); @@ -628,6 +788,34 @@ async fn add_user_uses_expected_path_query_and_body() { server.abort(); } +#[tokio::test] +async fn user_exists_limits_unexpected_error_response_body() { + let body = "x".repeat(MAX_UPSTREAM_ERROR_BODY_BYTES + 1); + let router = Router::new().route( + USER_INFO_PATH, + get(move || { + let body = body.clone(); + async move { (StatusCode::BAD_GATEWAY, body) } + }), + ); + + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let err = client + .user_exists("app-user") + .await + .expect_err("unexpected user lookup error should hide oversized body"); + + assert_oversized_upstream_body_hidden(err); + + server.abort(); +} + #[tokio::test] async fn set_user_policy_uses_single_authoritative_mapping_call() { let capture = Capture::default(); @@ -708,6 +896,37 @@ async fn bucket_object_lock_enabled_parses_enabled_response() { server.abort(); } +#[tokio::test] +async fn bucket_object_lock_enabled_limits_unexpected_error_response_body() { + let body = "x".repeat(MAX_UPSTREAM_ERROR_BODY_BYTES + 1); + let router = Router::new().route( + "/app-data", + get(move |req: Request| { + let body = body.clone(); + async move { + assert_eq!(req.uri().query().unwrap_or(""), "object-lock="); + (StatusCode::BAD_GATEWAY, body) + } + }), + ); + + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let err = client + .bucket_object_lock_enabled("app-data") + .await + .expect_err("unexpected object-lock error should hide oversized body"); + + assert_oversized_upstream_body_hidden(err); + + server.abort(); +} + #[tokio::test] async fn create_bucket_sends_object_lock_header_and_region_body() { let capture = Capture::default(); @@ -761,6 +980,34 @@ async fn create_bucket_sends_object_lock_header_and_region_body() { server.abort(); } +#[tokio::test] +async fn create_bucket_limits_unexpected_error_response_body() { + let body = "x".repeat(MAX_UPSTREAM_ERROR_BODY_BYTES + 1); + let router = Router::new().route( + "/app-data", + put(move || { + let body = body.clone(); + async move { (StatusCode::BAD_GATEWAY, body) } + }), + ); + + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let err = client + .create_bucket("app-data", None, false) + .await + .expect_err("unexpected bucket create error should hide oversized body"); + + assert_oversized_upstream_body_hidden(err); + + server.abort(); +} + #[test] fn extract_canned_policy_document_accepts_raw_policy_document() { let raw_policy = diff --git a/src/utils.rs b/src/utils.rs index 394d907..c227851 100755 --- a/src/utils.rs +++ b/src/utils.rs @@ -12,4 +12,5 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub(crate) mod sanitize; pub mod tls; diff --git a/src/utils/sanitize.rs b/src/utils/sanitize.rs new file mode 100644 index 0000000..f1fc5bc --- /dev/null +++ b/src/utils/sanitize.rs @@ -0,0 +1,322 @@ +// Copyright 2025 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +const SENSITIVE_KEYS: [&str; 16] = [ + "token", + "password", + "accesskey", + "access_key", + "access-key", + "secretkey", + "secret_key", + "secret-key", + "clientsecret", + "client_secret", + "client-secret", + "sessiontoken", + "session_token", + "session-token", + "credential", + "credentials", +]; + +pub(crate) fn redact_sensitive_pairs(message: &str) -> String { + let message = redact_sensitive_xml_tags(message); + redact_sensitive_key_value_pairs(&message) +} + +fn is_sensitive_key(key: &str) -> bool { + matches!( + normalize_key(key).as_str(), + "token" + | "password" + | "accesskey" + | "secretkey" + | "clientsecret" + | "sessiontoken" + | "credential" + | "credentials" + ) +} + +fn normalize_key(raw: &str) -> String { + raw.trim_matches(|ch: char| !ch.is_ascii_alphanumeric() && ch != '-' && ch != '_') + .chars() + .filter(|ch| ch.is_ascii_alphanumeric()) + .collect::() + .to_ascii_lowercase() +} + +fn redact_sensitive_xml_tags(message: &str) -> String { + let mut output = String::with_capacity(message.len()); + let mut cursor = 0usize; + + while cursor < message.len() { + let Some(ch) = message[cursor..].chars().next() else { + break; + }; + + if ch == '<' + && let Some(replacement) = redact_xml_tag_at(message, cursor) + { + output.push_str(&replacement.redacted); + cursor = replacement.end; + continue; + } + + output.push(ch); + cursor += ch.len_utf8(); + } + + output +} + +struct XmlRedaction { + redacted: String, + end: usize, +} + +fn redact_xml_tag_at(message: &str, cursor: usize) -> Option { + let tag_end = cursor + message[cursor..].find('>')?; + let tag_content = &message[cursor + 1..tag_end]; + if tag_content.starts_with('/') || tag_content.starts_with('?') || tag_content.starts_with('!') + { + return None; + } + let tag_name_end = tag_content + .find(|ch: char| ch.is_whitespace() || ch == '/') + .unwrap_or(tag_content.len()); + let tag_name = &tag_content[..tag_name_end]; + if tag_name.is_empty() || !is_sensitive_key(tag_name) { + return None; + } + + let open_end = tag_end + 1; + let close = format!(""); + let close_start = open_end + message[open_end..].find(&close)?; + let close_end = close_start + close.len(); + + Some(XmlRedaction { + redacted: format!( + "{}{}", + &message[cursor..open_end], + &message[close_start..close_end] + ), + end: close_end, + }) +} + +fn redact_sensitive_key_value_pairs(message: &str) -> 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 || !is_sensitive_key(candidate) { + 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 +} + +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}{quote}"); + } + } + "".to_string() +} + +#[cfg(test)] +mod tests { + use super::redact_sensitive_pairs; + + #[test] + fn preserves_required_key_names() { + let message = "Vault backend requires kmsSecret referencing a Secret with key vault-token"; + + assert_eq!(redact_sensitive_pairs(message), message); + } + + #[test] + fn redacts_colon_and_json_secret_values() { + let message = + "kms config token: tok_123 password: p@ss accesskey: AKIA_TEST secretkey: SK_TEST"; + + let sanitized = redact_sensitive_pairs(message); + + assert!(sanitized.contains("token")); + assert!(sanitized.contains("password")); + assert!(sanitized.contains("accesskey")); + assert!(sanitized.contains("secretkey")); + assert!(!sanitized.contains("tok_123")); + assert!(!sanitized.contains("p@ss")); + assert!(!sanitized.contains("AKIA_TEST")); + assert!(!sanitized.contains("SK_TEST")); + } + + #[test] + fn redacts_key_name_variants_and_xml_tags() { + let message = + r#"clientSecret: oidc-secret {"access_key":"AKIA_JSON"} SK_XML"#; + + let sanitized = redact_sensitive_pairs(message); + + assert!(sanitized.contains("clientSecret: ")); + assert!(sanitized.contains(r#""access_key":"""#)); + assert!(sanitized.contains("")); + assert!(!sanitized.contains("oidc-secret")); + assert!(!sanitized.contains("AKIA_JSON")); + assert!(!sanitized.contains("SK_XML")); + } + + #[test] + fn handles_unicode_without_panicking() { + let message = "错误🔐 token: tok_123 用户=测试 secretkey: SK_TEST 完成"; + + let sanitized = redact_sensitive_pairs(message); + + assert!(sanitized.contains("错误🔐")); + assert!(sanitized.contains("用户=测试")); + assert!(sanitized.contains("完成")); + assert!(sanitized.contains("token: ")); + assert!(sanitized.contains("secretkey: ")); + assert!(!sanitized.contains("tok_123")); + assert!(!sanitized.contains("SK_TEST")); + } + + #[test] + fn redacts_unicode_quoted_values() { + let message = "{\"说明\":\"🔐\",\"secretkey\":\"秘密值\"}"; + + let sanitized = redact_sensitive_pairs(message); + + assert!(sanitized.contains("\"说明\":\"🔐\"")); + assert!(sanitized.contains("\"secretkey\":\"\"")); + assert!(!sanitized.contains("秘密值")); + } + + #[test] + fn redacts_after_unicode_whitespace() { + let message = "token:\u{3000}tok_123 secretkey:\u{2003}SK_TEST"; + + let sanitized = redact_sensitive_pairs(message); + + assert!(sanitized.contains("token:\u{3000}")); + assert!(sanitized.contains("secretkey:\u{2003}")); + assert!(!sanitized.contains("tok_123")); + assert!(!sanitized.contains("SK_TEST")); + } +}