diff --git a/rustfs/src/storage/helper.rs b/rustfs/src/storage/helper.rs index e2d01b13c..97f2cd06e 100644 --- a/rustfs/src/storage/helper.rs +++ b/rustfs/src/storage/helper.rs @@ -15,6 +15,7 @@ use crate::server::{convert_ecstore_object_info, is_audit_module_enabled, is_notify_module_enabled}; use crate::storage::access::{ReqInfo, request_context_from_req}; use crate::storage::request_context::RequestContext; +use crate::storage::sse::KmsRequestAuditScope; use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; use hashbrown::HashMap; use http::StatusCode; @@ -94,6 +95,9 @@ pub struct EnabledOperationHelper { event_builder: Option, start_time: std::time::Instant, request_context: RequestContext, + /// Open for the lifetime of the audit entry so the SSE data path can summarise + /// its KMS work onto it. `None` when this request is not audited. + kms_audit: Option, } impl OperationHelper { @@ -220,6 +224,8 @@ impl OperationHelper { None }; + let kms_audit = audit_enabled.then(|| KmsRequestAuditScope::register(&request_id)); + Self::Enabled(Box::new(EnabledOperationHelper { audit_enabled, notify_enabled, @@ -228,6 +234,7 @@ impl OperationHelper { event_builder, start_time: request_context.start_time, request_context, + kms_audit, })) } @@ -326,14 +333,20 @@ impl OperationHelper { } // Inject OpenTelemetry trace context into audit tags for distributed tracing correlation - if state.request_context.trace_id.is_some() || state.request_context.span_id.is_some() { - let mut tags = HashMap::new(); - if let Some(ref tid) = state.request_context.trace_id { - tags.insert("traceId".to_string(), Value::String(tid.clone())); - } - if let Some(ref sid) = state.request_context.span_id { - tags.insert("spanId".to_string(), Value::String(sid.clone())); - } + let mut tags = HashMap::new(); + if let Some(ref tid) = state.request_context.trace_id { + tags.insert("traceId".to_string(), Value::String(tid.clone())); + } + if let Some(ref sid) = state.request_context.span_id { + tags.insert("spanId".to_string(), Value::String(sid.clone())); + } + // Summarise the KMS work the SSE data path did for this request, so the + // KMS outcome inherits this entry's principal and request ID instead of + // needing an event of its own. + if let Some(scope) = state.kms_audit.as_ref() { + tags.extend(scope.audit_tags().into_iter().map(|(key, value)| (key.to_string(), value))); + } + if !tags.is_empty() { final_builder = final_builder.tags(tags); } @@ -415,6 +428,7 @@ mod tests { use crate::server::{refresh_audit_module_enabled, refresh_notify_module_enabled}; use crate::storage::access::ReqInfo; use crate::storage::request_context::RequestContext; + use base64::Engine as _; use http::{Extensions, HeaderMap, HeaderValue, Method, Uri}; use metrics::{Counter, CounterFn, Gauge, GaugeFn, Histogram, HistogramFn, Key, KeyName, Metadata, SharedString, Unit}; use rustfs_credentials::Credentials; @@ -424,7 +438,7 @@ mod tests { use s3s::dto::{DeleteObjectTaggingInput, DeleteObjectTaggingOutput}; use s3s::{S3Request, S3Response}; use std::sync::{Arc, Mutex}; - use temp_env::with_vars; + use temp_env::{async_with_vars, with_vars}; fn build_request(input: T, method: Method, uri: Uri) -> S3Request { S3Request { @@ -683,6 +697,137 @@ mod tests { ); } + /// SSE-KMS object metadata whose IV is missing, so unwrapping fails inside the + /// SSE layer without depending on a KMS service being reachable from a unit test. + fn unreadable_sse_kms_metadata() -> std::collections::HashMap { + std::collections::HashMap::from([ + ("x-amz-server-side-encryption".to_string(), "aws:kms".to_string()), + ("x-rustfs-encryption-key-id".to_string(), "finance-key".to_string()), + ( + "x-rustfs-encryption-key".to_string(), + base64::engine::general_purpose::STANDARD.encode([7u8; 48]), + ), + ("x-rustfs-encryption-algorithm".to_string(), "aws:kms".to_string()), + ]) + } + + #[tokio::test] + async fn operation_helper_attaches_the_sse_kms_summary_to_the_audit_entry() { + async_with_vars( + [ + (rustfs_config::ENV_NOTIFY_ENABLE, Some("false")), + (rustfs_config::ENV_AUDIT_ENABLE, Some("true")), + ], + async { + refresh_notify_module_enabled(); + refresh_audit_module_enabled(); + + let input = DeleteObjectTaggingInput::builder() + .bucket("finance".to_string()) + .key("ledger.csv".to_string()) + .build() + .unwrap(); + let mut req = build_request(input, Method::GET, Uri::from_static("/finance/ledger.csv")); + req.extensions.insert(RequestContext { + request_id: "kms-tagged-request".to_string(), + x_amz_request_id: "kms-tagged-request".to_string(), + trace_id: None, + span_id: None, + start_time: std::time::Instant::now(), + }); + req.extensions.insert(ReqInfo { + cred: Some(Credentials { + access_key: "analyst".to_string(), + ..Default::default() + }), + bucket: Some("finance".to_string()), + object: Some("ledger.csv".to_string()), + ..Default::default() + }); + + let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject); + + // Drive the real data path: the principal built from this request is + // what carries the audit slot into the SSE code. + let metadata = unreadable_sse_kms_metadata(); + let principal = + crate::storage::sse::SseKmsPrincipal::from_request(&req).expect("authenticated request has a principal"); + crate::storage::sse::sse_decryption(crate::storage::sse::DecryptionRequest { + bucket: "finance", + key: "ledger.csv", + metadata: &metadata, + sse_customer_key: None, + sse_customer_key_md5: None, + principal: Some(&principal), + }) + .await + .expect_err("metadata without an IV cannot be unwrapped"); + + let result = Ok(S3Response::new(DeleteObjectTaggingOutput::default())); + let mut helper = helper.complete(&result); + let OperationHelper::Enabled(state) = &mut helper else { + panic!("helper should be enabled when the audit switch is on"); + }; + let entry = state.audit_builder.take().expect("audit builder should exist").build(); + let tags = entry.tags.expect("a request that used KMS must carry its summary"); + + assert_eq!(tags.get("sseType").and_then(|value| value.as_str()), Some("SSE-KMS")); + assert_eq!(tags.get("kmsKeyId").and_then(|value| value.as_str()), Some("finance-key")); + assert_eq!(tags.get("kmsOutcome").and_then(|value| value.as_str()), Some("failure")); + assert!(tags.contains_key("kmsErrorClass"), "a failed KMS interaction must be classified"); + + let rendered = serde_json::to_string(&tags).expect("audit tags serialize"); + assert!( + !rendered.contains(&base64::engine::general_purpose::STANDARD.encode([7u8; 48])), + "the audit entry must not carry the wrapped data key: {rendered}" + ); + }, + ) + .await; + } + + #[tokio::test] + async fn operation_helper_omits_kms_tags_for_requests_that_used_no_sse() { + async_with_vars( + [ + (rustfs_config::ENV_NOTIFY_ENABLE, Some("false")), + (rustfs_config::ENV_AUDIT_ENABLE, Some("true")), + ], + async { + refresh_notify_module_enabled(); + refresh_audit_module_enabled(); + + let input = DeleteObjectTaggingInput::builder() + .bucket("finance".to_string()) + .key("plain.txt".to_string()) + .build() + .unwrap(); + let mut req = build_request(input, Method::GET, Uri::from_static("/finance/plain.txt")); + req.extensions.insert(RequestContext { + request_id: "plain-request".to_string(), + x_amz_request_id: "plain-request".to_string(), + trace_id: None, + span_id: None, + start_time: std::time::Instant::now(), + }); + + let result = Ok(S3Response::new(DeleteObjectTaggingOutput::default())); + let mut helper = + OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).complete(&result); + let OperationHelper::Enabled(state) = &mut helper else { + panic!("helper should be enabled when the audit switch is on"); + }; + let entry = state.audit_builder.take().expect("audit builder should exist").build(); + + assert!( + entry.tags.is_none_or(|tags| tags.keys().all(|key| !key.starts_with("kms"))), + "a request that never reached KMS must not be tagged as if it had" + ); + }, + ) + .await; + } + #[test] fn operation_helper_still_records_s3_ops_when_audit_and_notify_are_disabled() { with_vars( diff --git a/rustfs/src/storage/sse.rs b/rustfs/src/storage/sse.rs index 024247f7c..6023143cd 100644 --- a/rustfs/src/storage/sse.rs +++ b/rustfs/src/storage/sse.rs @@ -74,7 +74,7 @@ use super::storage_api::ecstore_object::{ EncryptionResolutionError, EncryptionResolutionErrorKind, ObjectEncryptionResolver, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest, }; -use crate::storage::access::{ReqInfo, resource_free_condition_values}; +use crate::storage::access::{ReqInfo, request_context_from_req, resource_free_condition_values}; use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; #[cfg(feature = "rio-v2")] use aes_gcm::aead::Payload; @@ -105,8 +105,8 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; #[cfg(feature = "rio-v2")] use sha2::Sha256; -use std::collections::HashMap; -use std::sync::{Arc, LazyLock, RwLock}; +use std::collections::{BTreeSet, HashMap}; +use std::sync::{Arc, LazyLock, Mutex, RwLock, Weak}; use tracing::{debug, error}; const LOG_COMPONENT_STORAGE: &str = "storage"; @@ -800,6 +800,17 @@ pub enum SSEType { SseC, } +impl SSEType { + /// Stable scheme name for audit consumers. + fn audit_label(self) -> &'static str { + match self { + SSEType::SseS3 => "SSE-S3", + SSEType::SseKms => "SSE-KMS", + SSEType::SseC => "SSE-C", + } + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum EncryptionKeyKind { Direct, @@ -823,6 +834,10 @@ pub struct SseKmsPrincipal { is_owner: bool, claims: HashMap, conditions: HashMap>, + /// Audit slot of the S3 request this principal was built for, when the request + /// is being audited. The principal is the only request-derived value that reaches + /// the managed-SSE code paths, so it doubles as the carrier for the slot. + request_audit: Option>, /// Test-only decision overrides. Carried per principal rather than in a global slot so /// concurrent tests cannot observe each other's injection. #[cfg(test)] @@ -845,6 +860,7 @@ impl SseKmsPrincipal { is_owner: req_info.is_owner, claims: cred.claims_or_empty().clone(), conditions: resource_free_condition_values(req, cred), + request_audit: request_context_from_req(req).and_then(|context| kms_request_audit(&context.request_id)), #[cfg(test)] test_hooks: None, }) @@ -858,9 +874,18 @@ impl SseKmsPrincipal { is_owner: false, claims: HashMap::new(), conditions: HashMap::new(), + request_audit: None, test_hooks: Some(TestAuthorizationHooks { enforced, authorizer }), } } + + /// Bind this principal to `audit` so managed-SSE operations performed for it are + /// summarised onto the request's S3 audit entry. + #[cfg(test)] + fn with_request_audit(mut self, audit: Arc) -> Self { + self.request_audit = Some(audit); + self + } } #[cfg(test)] @@ -1029,7 +1054,14 @@ pub async fn authorize_sse_kms_object_read( return Ok(()); }; - authorize_sse_kms_key(principal, sse_type, KmsAction::DecryptAction, &key_id).await + let result = authorize_sse_kms_key(principal, sse_type, KmsAction::DecryptAction, &key_id).await; + // Only a denial is recorded here: an allowed read goes on to unwrap the key, + // and that operation reports its own outcome. + if let Err(error) = &result { + record_managed_kms_outcome(principal, sse_type, Some(&key_id), Err(error)); + } + + result } /// Resolve the scheme and KMS key a stored managed-SSE object was wrapped with. @@ -1054,6 +1086,234 @@ fn stored_managed_encryption_key(metadata: &HashMap) -> Option<( Some((sse_type, key_id)) } +// ============================================================================ +// Data-plane KMS audit attachment (SSE-S3 / SSE-KMS) +// ============================================================================ + +/// Tag keys carrying the KMS summary on an S3 audit entry. +/// +/// Audit consumers key off these strings, so they are a wire contract: append +/// new keys rather than renaming existing ones. +const KMS_AUDIT_TAG_SSE_TYPE: &str = "sseType"; +const KMS_AUDIT_TAG_KEY_ID: &str = "kmsKeyId"; +const KMS_AUDIT_TAG_KEY_VERSION: &str = "kmsKeyVersion"; +const KMS_AUDIT_TAG_OUTCOME: &str = "kmsOutcome"; +const KMS_AUDIT_TAG_ERROR_CLASS: &str = "kmsErrorClass"; + +/// Separator for the rare request that touched more than one key or scheme. +const KMS_AUDIT_VALUE_SEPARATOR: &str = ","; + +/// What the data path did with KMS while serving one S3 request. +/// +/// This rides on the request's existing S3 audit entry instead of becoming its +/// own event: the entry already carries the principal, request ID and API, so +/// attaching the KMS outcome costs no extra event volume and needs no second +/// correlation key. +/// +/// # Redaction +/// +/// Only the fields below may ever be recorded. The S3 audit entry fans out to +/// every configured target, several of which sit outside the KMS trust +/// boundary, so nothing derived from a data key — plaintext, ciphertext, +/// envelope, nonce — nor any caller-supplied encryption-context value belongs +/// here. Key identifiers and scheme names are configuration, not secrets. +#[derive(Debug, Default)] +struct KmsRequestAuditState { + sse_types: BTreeSet<&'static str>, + key_ids: BTreeSet, + key_versions: BTreeSet, + /// Set by the first failure. A request that failed any KMS interaction is + /// reported as failed even if others succeeded: a partially completed + /// envelope operation is not a success for an audit reader. + error_class: Option<&'static str>, + recorded: bool, +} + +/// Shared accumulator for one request's KMS summary. +/// +/// Written by the managed-SSE code paths through the request's principal, read +/// once by the [`OperationHelper`](crate::storage::helper::OperationHelper) that +/// owns the request's audit entry. +#[derive(Debug, Default)] +pub(crate) struct KmsRequestAudit(Mutex); + +impl KmsRequestAudit { + fn record(&self, sse_type: SSEType, key_id: Option<&str>, key_version: Option, error_class: Option<&'static str>) { + // A poisoned lock costs the audit entry its KMS tags; it must never cost + // the request its result. + let Ok(mut state) = self.0.lock() else { + return; + }; + + state.recorded = true; + state.sse_types.insert(sse_type.audit_label()); + if let Some(key_id) = key_id.filter(|value| !value.is_empty()) { + state.key_ids.insert(key_id.to_string()); + } + if let Some(key_version) = key_version { + state.key_versions.insert(key_version); + } + if state.error_class.is_none() { + state.error_class = error_class; + } + } + + /// Render the summary as audit tags, or nothing when the request performed + /// no KMS work at all (SSE-C and unencrypted objects never reach KMS). + pub(crate) fn audit_tags(&self) -> Vec<(&'static str, Value)> { + let Ok(state) = self.0.lock() else { + return Vec::new(); + }; + if !state.recorded { + return Vec::new(); + } + + let mut tags = Vec::with_capacity(5); + tags.push((KMS_AUDIT_TAG_SSE_TYPE, join_audit_values(state.sse_types.iter().copied()))); + if !state.key_ids.is_empty() { + tags.push((KMS_AUDIT_TAG_KEY_ID, join_audit_values(state.key_ids.iter().map(String::as_str)))); + } + if !state.key_versions.is_empty() { + tags.push(( + KMS_AUDIT_TAG_KEY_VERSION, + join_audit_values(state.key_versions.iter().map(u32::to_string)), + )); + } + let outcome = if state.error_class.is_some() { "failure" } else { "success" }; + tags.push((KMS_AUDIT_TAG_OUTCOME, Value::String(outcome.to_string()))); + if let Some(error_class) = state.error_class { + tags.push((KMS_AUDIT_TAG_ERROR_CLASS, Value::String(error_class.to_string()))); + } + + tags + } +} + +/// Render a set of values as one tag value. +/// +/// Always a string, including for the single-value case: a consumer that would +/// have to branch on the JSON type of a tag is a consumer that will get it +/// wrong for the request that happens to touch two keys. +fn join_audit_values(values: impl Iterator>) -> Value { + let joined = values.map(|value| value.as_ref().to_string()).collect::>(); + Value::String(joined.join(KMS_AUDIT_VALUE_SEPARATOR)) +} + +/// Slots of the requests currently being audited, keyed by canonical request ID. +/// +/// Only a weak reference is held: the owning [`KmsRequestAuditScope`] decides the +/// lifetime, so a scope that is somehow not dropped cannot keep a slot alive. +static KMS_REQUEST_AUDITS: LazyLock>>> = LazyLock::new(|| Mutex::new(HashMap::new())); + +/// Registration of one request's KMS audit slot, dropped with the request's +/// audit entry. +pub(crate) struct KmsRequestAuditScope { + request_id: String, + audit: Arc, +} + +impl KmsRequestAuditScope { + /// Open a slot for `request_id`, so managed-SSE operations served under it can + /// be summarised onto its audit entry. + pub(crate) fn register(request_id: &str) -> Self { + let audit = Arc::new(KmsRequestAudit::default()); + if let Ok(mut registry) = KMS_REQUEST_AUDITS.lock() { + registry.insert(request_id.to_string(), Arc::downgrade(&audit)); + } + + Self { + request_id: request_id.to_string(), + audit, + } + } + + /// Tags describing the KMS work recorded under this request. + pub(crate) fn audit_tags(&self) -> Vec<(&'static str, Value)> { + self.audit.audit_tags() + } +} + +impl Drop for KmsRequestAuditScope { + fn drop(&mut self) { + if let Ok(mut registry) = KMS_REQUEST_AUDITS.lock() { + registry.remove(&self.request_id); + } + } +} + +/// Resolve the audit slot opened for `request_id`, if the request is being audited. +fn kms_request_audit(request_id: &str) -> Option> { + KMS_REQUEST_AUDITS.lock().ok()?.get(request_id)?.upgrade() +} + +/// Failure classes for data-path failures that never reached a KMS backend. +/// +/// Backend failures carry the classification defined by the KMS audit contract; +/// everything else is classified at the S3 boundary it surfaced through. +fn kms_data_plane_error_class(error: &ApiError) -> &'static str { + if let Some(failure) = error + .source + .as_ref() + .and_then(|source| source.downcast_ref::()) + { + return failure.class; + } + + match error.code { + S3ErrorCode::AccessDenied => "access_denied", + S3ErrorCode::InvalidArgument | S3ErrorCode::InvalidRequest => "invalid_argument", + _ => "sse_internal", + } +} + +/// Carries a KMS failure class across the conversion to an S3-facing error. +/// +/// The class is decided while the `KmsError` is still typed; re-deriving it from +/// a rendered message downstream would be guesswork. +#[derive(Debug)] +struct KmsDataPlaneFailure { + class: &'static str, + source: Option>, +} + +impl std::fmt::Display for KmsDataPlaneFailure { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match &self.source { + Some(source) => source.fmt(formatter), + None => formatter.write_str(self.class), + } + } +} + +impl std::error::Error for KmsDataPlaneFailure { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + self.source + .as_ref() + .map(|source| source.as_ref() as &(dyn std::error::Error + 'static)) + } +} + +/// Record the outcome of one managed-SSE operation on the request's audit entry. +/// +/// A `None` principal marks an internal caller — replication, lifecycle, heal — +/// which has no S3 audit entry to attach to. +fn record_managed_kms_outcome( + principal: Option<&SseKmsPrincipal>, + sse_type: SSEType, + key_id: Option<&str>, + result: Result<(), &ApiError>, +) { + let Some(audit) = principal.and_then(|principal| principal.request_audit.as_ref()) else { + return; + }; + + // The KMS key version is not observable on the data path: neither the + // generated data key nor the stored envelope surfaces the master-key version + // that wrapped it. Recording a fabricated version would be worse than + // omitting the tag, so it stays absent until KMS reports it. + audit.record(sse_type, key_id, None, result.err().map(kms_data_plane_error_class)); +} + pub(crate) struct SseObjectEncryptionResolver; #[async_trait] @@ -2015,6 +2275,48 @@ async fn apply_managed_encryption_material( ssekms_context: Option>, content_size: i64, principal: Option<&SseKmsPrincipal>, +) -> Result { + let requested_sse_type = managed_sse_type(server_side_encryption.as_str()); + let requested_key_id = kms_key_id.clone(); + let result = apply_managed_encryption_material_inner( + bucket, + key, + server_side_encryption, + kms_key_id, + ssekms_context, + content_size, + principal, + ) + .await; + + match &result { + // The resolved key is only known on success: it may come from the request, + // the bucket default or the KMS service default. On failure the audit entry + // records what the caller asked for, which is what a reader needs to see. + Ok(material) => record_managed_kms_outcome(principal, material.sse_type, material.kms_key_id.as_deref(), Ok(())), + Err(error) => record_managed_kms_outcome(principal, requested_sse_type, requested_key_id.as_deref(), Err(error)), + } + + result +} + +/// Scheme a managed-SSE header names, defaulting to SSE-S3 the same way the +/// encryption and decryption paths do. +fn managed_sse_type(server_side_encryption: &str) -> SSEType { + match server_side_encryption { + ServerSideEncryption::AWS_KMS => SSEType::SseKms, + _ => SSEType::SseS3, + } +} + +async fn apply_managed_encryption_material_inner( + bucket: &str, + key: &str, + server_side_encryption: ServerSideEncryption, + kms_key_id: Option, + ssekms_context: Option>, + content_size: i64, + principal: Option<&SseKmsPrincipal>, ) -> Result { if !is_managed_sse(&server_side_encryption) { return Err(ApiError::from(StorageError::other(format!( @@ -2101,6 +2403,29 @@ async fn apply_managed_decryption_material( key: &str, metadata: &HashMap, principal: Option<&SseKmsPrincipal>, +) -> Result, ApiError> { + let result = apply_managed_decryption_material_inner(bucket, key, metadata, principal).await; + + match &result { + // `None` means the object carries no managed-SSE metadata — SSE-C and + // plaintext objects never reach KMS and must not appear in the summary. + Ok(None) => {} + Ok(Some(material)) => record_managed_kms_outcome(principal, material.sse_type, material.kms_key_id.as_deref(), Ok(())), + Err(error) => { + if let Some((sse_type, key_id)) = stored_managed_encryption_key(metadata) { + record_managed_kms_outcome(principal, sse_type, Some(&key_id), Err(error)); + } + } + } + + result +} + +async fn apply_managed_decryption_material_inner( + bucket: &str, + key: &str, + metadata: &HashMap, + principal: Option<&SseKmsPrincipal>, ) -> Result, ApiError> { #[cfg(not(feature = "rio-v2"))] let _ = (bucket, key); @@ -2332,7 +2657,13 @@ struct KmsSseDekProvider { } fn kms_operation_error(error: rustfs_kms::KmsError) -> ApiError { - ApiError::from(StorageError::other(error)) + let class = rustfs_kms::audit::error_class(&error); + let mut api_error = ApiError::from(StorageError::other(error)); + // Wrap rather than replace the source so the original chain stays intact for + // logging; the wrapper only adds the class the audit attachment needs. + let source = api_error.source.take(); + api_error.source = Some(Box::new(KmsDataPlaneFailure { class, source })); + api_error } impl KmsSseDekProvider { @@ -5712,4 +6043,192 @@ mod tests { assert!(authorizer.calls().is_empty()); } + + // ======================================================================== + // Data-plane KMS audit attachment + // ======================================================================== + + fn audited_principal(enforced: bool, allowed: bool) -> (SseKmsPrincipal, Arc) { + let audit = Arc::new(super::KmsRequestAudit::default()); + let authorizer = RecordingKmsKeyAuthorizer::new(allowed); + let principal = SseKmsPrincipal::for_test("analyst", enforced, authorizer).with_request_audit(audit.clone()); + (principal, audit) + } + + fn audit_tag(tags: &[(&'static str, serde_json::Value)], key: &str) -> Option { + tags.iter() + .find(|(candidate, _)| *candidate == key) + .and_then(|(_, value)| value.as_str().map(str::to_string)) + } + + #[tokio::test] + async fn sse_kms_write_and_read_summarise_key_id_and_outcome_for_the_audit_entry() { + use rustfs_kms::types::{CreateKeyRequest, KeyUsage}; + let _guard = lock_sse_test_state().await; + + reset_sse_dek_provider(); + let manager = configure_test_global_local_kms().await; + manager + .get_encryption_service() + .await + .expect("encryption service should exist") + .create_key(CreateKeyRequest { + key_name: Some("audit-key".to_string()), + key_usage: KeyUsage::EncryptDecrypt, + description: None, + policy: None, + tags: HashMap::new(), + origin: None, + }) + .await + .expect("kms test key should be created"); + let provider = KmsSseDekProvider::new_with_service_manager(manager.clone()) + .await + .expect("kms provider should initialize from the configured test manager"); + super::set_sse_dek_provider_for_test(Arc::new(provider)); + + let (write_principal, write_audit) = audited_principal(false, true); + let material = sse_encryption(EncryptionRequest { + bucket: "finance", + key: "ledger.csv", + server_side_encryption: Some(ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS)), + ssekms_key_id: Some("audit-key".to_string()), + ssekms_context: Some(HashMap::from([("tenant".to_string(), "acct-4711".to_string())])), + sse_customer_algorithm: None, + sse_customer_key: None, + sse_customer_key_md5: None, + content_size: 128, + principal: Some(&write_principal), + }) + .await + .expect("sse-kms encryption should succeed") + .expect("managed sse-kms material"); + + let write_tags = write_audit.audit_tags(); + assert_eq!(audit_tag(&write_tags, "sseType").as_deref(), Some("SSE-KMS")); + assert_eq!(audit_tag(&write_tags, "kmsKeyId").as_deref(), Some("audit-key")); + assert_eq!(audit_tag(&write_tags, "kmsOutcome").as_deref(), Some("success")); + assert_eq!(audit_tag(&write_tags, "kmsErrorClass"), None, "a successful request has no error class"); + + // The audit entry reaches every configured target: no encoding of the data + // key, and no caller-supplied encryption-context value, may appear in it. + let rendered = format!("{write_tags:?}"); + let encrypted_data_key = material.encrypted_data_key.clone().expect("managed sse wraps a data key"); + for secret in [ + BASE64_STANDARD.encode(&encrypted_data_key), + BASE64_STANDARD.encode(material.key_bytes), + format!("{:?}", material.key_bytes), + format!("{encrypted_data_key:?}"), + "acct-4711".to_string(), + ] { + assert!(!rendered.contains(&secret), "audit tags must not carry key material: {rendered}"); + } + + let metadata = encryption_material_to_metadata(&material).expect("kms metadata should serialize"); + let (read_principal, read_audit) = audited_principal(false, true); + sse_decryption(DecryptionRequest { + bucket: "finance", + key: "ledger.csv", + metadata: &metadata, + sse_customer_key: None, + sse_customer_key_md5: None, + principal: Some(&read_principal), + }) + .await + .expect("sse-kms decryption should succeed") + .expect("managed sse-kms material"); + + let read_tags = read_audit.audit_tags(); + assert_eq!(audit_tag(&read_tags, "sseType").as_deref(), Some("SSE-KMS")); + assert_eq!(audit_tag(&read_tags, "kmsKeyId").as_deref(), Some("audit-key")); + assert_eq!(audit_tag(&read_tags, "kmsOutcome").as_deref(), Some("success")); + } + + #[tokio::test] + async fn sse_c_requests_produce_no_kms_audit_summary() { + let key = [0x21u8; 32]; + let (principal, audit) = audited_principal(false, true); + + sse_encryption(EncryptionRequest { + bucket: "finance", + key: "ledger.csv", + server_side_encryption: None, + ssekms_key_id: None, + ssekms_context: None, + sse_customer_algorithm: Some(SSECustomerAlgorithm::from("AES256".to_string())), + sse_customer_key: Some(SSECustomerKey::from(BASE64_STANDARD.encode(key))), + sse_customer_key_md5: Some(SSECustomerKeyMD5::from(md5_base64(key))), + content_size: 128, + principal: Some(&principal), + }) + .await + .expect("sse-c encryption should succeed") + .expect("sse-c material"); + + assert!( + audit.audit_tags().is_empty(), + "SSE-C keys never reach KMS, so the request has no KMS outcome to report" + ); + } + + #[tokio::test] + async fn denied_sse_kms_read_is_summarised_as_an_access_denied_failure() { + let (principal, audit) = audited_principal(true, false); + + authorize_sse_kms_object_read(Some(&principal), &sse_kms_object_metadata()) + .await + .expect_err("an unauthorized principal must not read the source key"); + + let tags = audit.audit_tags(); + assert_eq!(audit_tag(&tags, "sseType").as_deref(), Some("SSE-KMS")); + assert_eq!(audit_tag(&tags, "kmsKeyId").as_deref(), Some("finance-key")); + assert_eq!(audit_tag(&tags, "kmsOutcome").as_deref(), Some("failure")); + assert_eq!(audit_tag(&tags, "kmsErrorClass").as_deref(), Some("access_denied")); + } + + #[test] + fn kms_backend_failures_keep_the_kms_audit_error_class() { + assert_eq!( + super::kms_data_plane_error_class(&kms_operation_error(rustfs_kms::KmsError::key_not_found("finance-key"))), + "key_not_found" + ); + assert_eq!( + super::kms_data_plane_error_class(&kms_operation_error(rustfs_kms::KmsError::access_denied("nope"))), + "access_denied" + ); + // Failures raised by the SSE layer itself never saw a KmsError; they are + // classified by the boundary they surfaced through. + assert_eq!( + super::kms_data_plane_error_class(&ApiError { + code: S3ErrorCode::AccessDenied, + message: "Access Denied".to_string(), + source: None, + }), + "access_denied" + ); + assert_eq!( + super::kms_data_plane_error_class(&ApiError::from(StorageError::other("no KMS key available"))), + "sse_internal" + ); + } + + #[test] + fn a_request_audit_slot_is_reachable_only_while_its_scope_lives() { + let scope = super::KmsRequestAuditScope::register("request-under-audit"); + let slot = super::kms_request_audit("request-under-audit").expect("a registered request must resolve its slot"); + slot.record(SSEType::SseKms, Some("finance-key"), None, None); + assert_eq!(audit_tag(&scope.audit_tags(), "kmsKeyId").as_deref(), Some("finance-key")); + + drop(scope); + assert!( + super::kms_request_audit("request-under-audit").is_none(), + "the slot must not outlive the audit entry it belongs to" + ); + } + + #[test] + fn a_request_that_did_no_kms_work_reports_no_tags() { + let scope = super::KmsRequestAuditScope::register("quiet-request"); + assert!(scope.audit_tags().is_empty()); + } }