diff --git a/Cargo.lock b/Cargo.lock index 0a6595954..780f7b61f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9523,6 +9523,7 @@ dependencies = [ "moka", "rand 0.10.2", "reqwest", + "rustfs-s3-types", "rustfs-security-governance", "rustfs-utils", "rustify", diff --git a/crates/kms/Cargo.toml b/crates/kms/Cargo.toml index 1dd5b9385..3d8c80876 100644 --- a/crates/kms/Cargo.toml +++ b/crates/kms/Cargo.toml @@ -64,6 +64,10 @@ md-5 = { workspace = true } arc-swap = { workspace = true } rustfs-utils = { workspace = true } rustfs-security-governance = { workspace = true } +# `EventName` for KMS audit records. A leaf crate with no rustfs dependencies, +# so the audit sink can live outside this crate without a second, drifting +# copy of the event vocabulary. +rustfs-s3-types = { workspace = true } # HTTP client for Vault reqwest = { workspace = true } diff --git a/crates/kms/src/audit.rs b/crates/kms/src/audit.rs new file mode 100644 index 000000000..a73d94fb2 --- /dev/null +++ b/crates/kms/src/audit.rs @@ -0,0 +1,423 @@ +// Copyright 2024 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. + +//! Audit contract for KMS management operations. +//! +//! [`KmsManager`](crate::manager::KmsManager) builds a [`KmsAuditRecord`] for +//! every management operation it serves and hands it to the installed +//! [`KmsAuditSink`]. No sink is installed by default, so a deployment that +//! does not consume KMS audit records pays nothing beyond the `Option` check. +//! +//! The record is deliberately a KMS-side type rather than an audit-crate +//! entry: keeping the dependency edge out of this crate lets the server map +//! records onto its existing audit pipeline (and its delivery semantics) +//! without KMS having to know that pipeline exists. +//! +//! # Redaction +//! +//! A record has no field that can hold key material, and every value that +//! originates from a caller passes through [`redact_encryption_context`] +//! before it is stored. Adding a field here means re-checking that invariant. + +use crate::error::KmsError; +use crate::types::OperationContext; +use rustfs_s3_types::EventName; +use sha2::{Digest, Sha256}; +use std::collections::{BTreeMap, HashMap}; +use std::time::Duration; +use uuid::Uuid; + +/// Encryption-context keys that describe *where* an object lives rather than +/// anything secret about it. Their values are recorded verbatim; every other +/// key is reduced to a digest. +const ENCRYPTION_CONTEXT_ALLOWLIST: [&str; 5] = ["bucket", "object", "object_key", "algorithm", "sse_type"]; + +/// Prefix marking a value that was replaced by a digest of itself. +const DIGEST_PREFIX: &str = "sha256:"; + +/// Hex characters of the digest that are kept. Enough to correlate repeated +/// values across records without being reversible in practice. +const DIGEST_LEN: usize = 16; + +/// A KMS management operation that produces an audit record. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum KmsAuditOperation { + /// Creation of a new master key. + CreateKey, + /// Metadata lookup for a single key. + DescribeKey, + /// Enumeration of keys. + ListKeys, + /// Scheduling a key for deletion after the pending window. + ScheduleKeyDeletion, + /// Cancelling a previously scheduled deletion. + CancelKeyDeletion, + /// Returning a disabled key to service. + EnableKey, + /// Taking a key out of service without destroying it. + DisableKey, + /// Rotating a key to a new version. + RotateKey, + /// Irreversible removal of key material once the pending window expired. + DeleteKey, +} + +impl KmsAuditOperation { + /// Stable operation name for audit consumers. + pub fn as_str(&self) -> &'static str { + match self { + KmsAuditOperation::CreateKey => "CreateKey", + KmsAuditOperation::DescribeKey => "DescribeKey", + KmsAuditOperation::ListKeys => "ListKeys", + KmsAuditOperation::ScheduleKeyDeletion => "ScheduleKeyDeletion", + KmsAuditOperation::CancelKeyDeletion => "CancelKeyDeletion", + KmsAuditOperation::EnableKey => "EnableKey", + KmsAuditOperation::DisableKey => "DisableKey", + KmsAuditOperation::RotateKey => "RotateKey", + KmsAuditOperation::DeleteKey => "DeleteKey", + } + } + + /// Notification/audit event name carried by records for this operation. + pub fn event_name(&self) -> EventName { + match self { + KmsAuditOperation::CreateKey => EventName::KmsKeyCreated, + KmsAuditOperation::DescribeKey | KmsAuditOperation::ListKeys => EventName::KmsKeyAccessed, + KmsAuditOperation::ScheduleKeyDeletion => EventName::KmsKeyDeletionScheduled, + KmsAuditOperation::CancelKeyDeletion => EventName::KmsKeyDeletionCancelled, + KmsAuditOperation::EnableKey => EventName::KmsKeyEnabled, + KmsAuditOperation::DisableKey => EventName::KmsKeyDisabled, + KmsAuditOperation::RotateKey => EventName::KmsKeyRotated, + KmsAuditOperation::DeleteKey => EventName::KmsKeyDeleted, + } + } +} + +/// Whether the audited operation completed. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum KmsAuditOutcome { + /// The operation completed and its effect is durable. + Success, + /// The operation failed; `error_class` carries the reason category. + Failure, +} + +impl KmsAuditOutcome { + /// Stable outcome name for audit consumers. + pub fn as_str(&self) -> &'static str { + match self { + KmsAuditOutcome::Success => "success", + KmsAuditOutcome::Failure => "failure", + } + } +} + +/// Coarse, stable classification of a failed KMS operation. +/// +/// Audit consumers alert on these, so the strings are a wire contract: add +/// new classes rather than renaming existing ones. +pub fn error_class(error: &KmsError) -> &'static str { + match error { + KmsError::ConfigurationError { .. } => "configuration", + KmsError::KeyNotFound { .. } => "key_not_found", + KmsError::InvalidKey { .. } => "invalid_key", + KmsError::CryptographicError { .. } => "cryptographic", + KmsError::BackendError { .. } => "backend", + KmsError::AccessDenied { .. } => "access_denied", + KmsError::KeyAlreadyExists { .. } => "key_already_exists", + KmsError::InvalidOperation { .. } => "invalid_operation", + KmsError::InternalError { .. } => "internal", + KmsError::SerializationError { .. } => "serialization", + KmsError::IoError { .. } => "io", + KmsError::CacheError { .. } => "cache", + KmsError::ValidationError { .. } => "validation", + KmsError::UnsupportedAlgorithm { .. } => "unsupported_algorithm", + KmsError::InvalidKeySize { .. } => "invalid_key_size", + KmsError::ContextMismatch { .. } => "context_mismatch", + KmsError::OperationTimedOut { .. } => "timeout", + KmsError::OperationCancelled { .. } => "cancelled", + KmsError::MaterialMissing { .. } => "material_missing", + KmsError::MaterialCorrupt { .. } => "material_corrupt", + KmsError::MaterialAuthenticationFailed { .. } => "material_authentication_failed", + KmsError::UnsupportedFormatVersion { .. } => "unsupported_format_version", + KmsError::KeyVersionNotFound { .. } => "key_version_not_found", + KmsError::Backup(_) => "backup", + KmsError::UnsupportedCapability { .. } => "unsupported_capability", + KmsError::CredentialsUnavailable { .. } => "credentials_unavailable", + } +} + +/// One audited KMS management operation. +/// +/// Everything an audit consumer needs to answer "who did what to which key, +/// against which backend, and did it work" — and nothing that could carry key +/// material. See the module docs for the redaction rules. +/// +/// Tenant attribution is intentionally absent: multi-tenancy is not modelled +/// yet, and an invented tenant value is worse than a missing one. It will +/// arrive as an entry in [`Self::context`] once tenancy lands. +#[derive(Debug, Clone)] +pub struct KmsAuditRecord { + /// Correlates every record emitted for one logical request. + pub operation_id: Uuid, + /// The audited operation. + pub operation: KmsAuditOperation, + /// Event name published to audit consumers. + pub event: EventName, + /// Authenticated identity that requested the operation, or + /// [`OperationContext::INTERNAL_PRINCIPAL`] for server-initiated work. + pub principal: String, + /// Client address, when the caller supplied one. + pub source_ip: Option, + /// Client user agent, when the caller supplied one. + pub user_agent: Option, + /// Key the operation acted on. `None` for operations that span keys, such + /// as listing. + pub key_id: Option, + /// Key version the operation resolved to, when the result identifies one. + /// Operations on a key as a whole leave this unset. + pub key_version: Option, + /// Whether the operation succeeded. + pub outcome: KmsAuditOutcome, + /// Failure category; `None` on success. See [`error_class`]. + pub error_class: Option<&'static str>, + /// Backend that served the operation, e.g. `local` or `vault-transit`. + pub backend: &'static str, + /// Wall-clock time spent in the audited operation. + pub latency: Duration, + /// Retries performed above the audit point. `None` means the number is + /// not observable here: backend-internal retries are accounted for by the + /// `rustfs_kms_backend_operation_attempts` metric instead of being + /// duplicated (and possibly contradicted) in the audit trail. + pub retry_count: Option, + /// Server-supplied correlation values carried by the operation context. + /// Also the reserved slot for tenant attribution. + pub context: BTreeMap, + /// Caller-supplied encryption context, after [`redact_encryption_context`]. + pub encryption_context: BTreeMap, +} + +impl KmsAuditRecord { + /// Start a record for `operation` performed under `context`. + /// + /// The outcome defaults to failure so that a record which somehow escapes + /// without [`Self::with_result`] under-reports success rather than + /// inventing it. + pub fn new(operation: KmsAuditOperation, context: &OperationContext, backend: &'static str) -> Self { + Self { + operation_id: context.operation_id, + operation, + event: operation.event_name(), + principal: context.principal.clone(), + source_ip: context.source_ip.clone(), + user_agent: context.user_agent.clone(), + key_id: None, + key_version: None, + outcome: KmsAuditOutcome::Failure, + error_class: None, + backend, + latency: Duration::ZERO, + retry_count: None, + context: context + .additional_context + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + encryption_context: BTreeMap::new(), + } + } + + /// Attach the key the operation acted on. + pub fn with_key_id(mut self, key_id: Option>) -> Self { + self.key_id = key_id.map(Into::into); + self + } + + /// Attach the key version the operation resolved to. + pub fn with_key_version(mut self, key_version: Option) -> Self { + self.key_version = key_version; + self + } + + /// Attach the measured duration of the operation. + pub fn with_latency(mut self, latency: Duration) -> Self { + self.latency = latency; + self + } + + /// Derive outcome and error class from the operation result. + pub fn with_result(mut self, result: &crate::error::Result) -> Self { + match result { + Ok(_) => { + self.outcome = KmsAuditOutcome::Success; + self.error_class = None; + } + Err(error) => { + self.outcome = KmsAuditOutcome::Failure; + self.error_class = Some(error_class(error)); + } + } + self + } + + /// Attach the caller-supplied encryption context, redacted. + /// + /// Redaction happens here rather than at the call site so no caller can + /// place a raw context value into a record by accident. + pub fn with_encryption_context(mut self, encryption_context: &HashMap) -> Self { + self.encryption_context = redact_encryption_context(encryption_context); + self + } +} + +/// Reduce a caller-supplied encryption context to something safe to persist. +/// +/// Keys naming the object's location are kept verbatim because they are the +/// reason the context is audited at all. Every other value is replaced by a +/// truncated digest, which still correlates repeated values across records +/// but does not reproduce whatever the caller put there. +pub fn redact_encryption_context(encryption_context: &HashMap) -> BTreeMap { + encryption_context + .iter() + .map(|(key, value)| { + let recorded = if ENCRYPTION_CONTEXT_ALLOWLIST.contains(&key.as_str()) { + value.clone() + } else { + digest_value(value) + }; + (key.clone(), recorded) + }) + .collect() +} + +fn digest_value(value: &str) -> String { + let digest = hex::encode(Sha256::digest(value.as_bytes())); + format!("{DIGEST_PREFIX}{}", &digest[..DIGEST_LEN]) +} + +/// Receives KMS audit records. +/// +/// Implementations must not block: they are called on the task that served +/// the KMS operation. Delivery failures are the sink's problem — the KMS +/// operation has already completed by the time a record is emitted, and its +/// result is never changed by what the sink does. +pub trait KmsAuditSink: Send + Sync { + /// Handle one audit record. + fn emit(&self, record: KmsAuditRecord); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn every_operation_maps_to_a_kms_event() { + let operations = [ + KmsAuditOperation::CreateKey, + KmsAuditOperation::DescribeKey, + KmsAuditOperation::ListKeys, + KmsAuditOperation::ScheduleKeyDeletion, + KmsAuditOperation::CancelKeyDeletion, + KmsAuditOperation::EnableKey, + KmsAuditOperation::DisableKey, + KmsAuditOperation::RotateKey, + KmsAuditOperation::DeleteKey, + ]; + + for operation in operations { + let event = operation.event_name(); + assert!(event.is_kms(), "{} must map to a KMS event, got {event}", operation.as_str()); + } + } + + #[test] + fn record_defaults_to_failure_until_a_result_is_attached() { + let context = OperationContext::new("tester".to_string()); + let record = KmsAuditRecord::new(KmsAuditOperation::CreateKey, &context, "local"); + + assert_eq!(record.outcome, KmsAuditOutcome::Failure); + assert!(record.error_class.is_none()); + + let ok: crate::error::Result<()> = Ok(()); + assert_eq!(record.clone().with_result(&ok).outcome, KmsAuditOutcome::Success); + + let denied: crate::error::Result<()> = Err(KmsError::access_denied("nope")); + let failed = record.with_result(&denied); + assert_eq!(failed.outcome, KmsAuditOutcome::Failure); + assert_eq!(failed.error_class, Some("access_denied")); + } + + #[test] + fn record_carries_the_full_operation_context() { + let context = OperationContext::new("arn:aws:iam::user/alice".to_string()) + .with_source_ip("10.0.0.7".to_string()) + .with_user_agent("aws-cli/2".to_string()) + .with_context("requestID".to_string(), "req-1".to_string()); + + let record = KmsAuditRecord::new(KmsAuditOperation::RotateKey, &context, "vault-transit") + .with_key_id(Some("key-1")) + .with_key_version(Some(3)) + .with_latency(Duration::from_millis(12)); + + assert_eq!(record.operation_id, context.operation_id); + assert_eq!(record.principal, "arn:aws:iam::user/alice"); + assert_eq!(record.source_ip.as_deref(), Some("10.0.0.7")); + assert_eq!(record.user_agent.as_deref(), Some("aws-cli/2")); + assert_eq!(record.key_id.as_deref(), Some("key-1")); + assert_eq!(record.key_version, Some(3)); + assert_eq!(record.backend, "vault-transit"); + assert_eq!(record.latency, Duration::from_millis(12)); + assert_eq!(record.event, EventName::KmsKeyRotated); + assert_eq!(record.context.get("requestID").map(String::as_str), Some("req-1")); + } + + #[test] + fn encryption_context_keeps_only_allowlisted_values_verbatim() { + let secret = "eyJhbGciOiJIUzI1NiJ9.super-secret-grant-token"; + let context = HashMap::from([ + ("bucket".to_string(), "photos".to_string()), + ("object_key".to_string(), "2026/cat.png".to_string()), + ("algorithm".to_string(), "AES256".to_string()), + ("grant_token".to_string(), secret.to_string()), + ("customer_reference".to_string(), "acct-4711".to_string()), + ]); + + let redacted = redact_encryption_context(&context); + + assert_eq!(redacted.get("bucket").map(String::as_str), Some("photos")); + assert_eq!(redacted.get("object_key").map(String::as_str), Some("2026/cat.png")); + assert_eq!(redacted.get("algorithm").map(String::as_str), Some("AES256")); + + for key in ["grant_token", "customer_reference"] { + let value = redacted.get(key).expect("non-allowlisted key should be kept as a digest"); + assert!(value.starts_with(DIGEST_PREFIX), "{key} should be digested, got {value}"); + } + + let rendered = format!("{redacted:?}"); + assert!(!rendered.contains(secret), "redacted context must not reproduce the raw value"); + assert!(!rendered.contains("acct-4711"), "redacted context must not reproduce the raw value"); + } + + #[test] + fn identical_values_digest_identically_across_records() { + // Correlation is the reason a digest is preferable to a fixed + // placeholder; assert it actually holds. + let first = redact_encryption_context(&HashMap::from([("tenant".to_string(), "alpha".to_string())])); + let second = redact_encryption_context(&HashMap::from([("tenant".to_string(), "alpha".to_string())])); + let other = redact_encryption_context(&HashMap::from([("tenant".to_string(), "beta".to_string())])); + + assert_eq!(first.get("tenant"), second.get("tenant")); + assert_ne!(first.get("tenant"), other.get("tenant")); + } +} diff --git a/crates/kms/src/config.rs b/crates/kms/src/config.rs index be4376ca5..fe09bcc9b 100644 --- a/crates/kms/src/config.rs +++ b/crates/kms/src/config.rs @@ -130,6 +130,21 @@ pub enum KmsBackend { Static, } +impl KmsBackend { + /// Stable identifier for logs, metrics, and audit records. + /// + /// External consumers key off these values, so treat them as a wire + /// contract rather than a rendering detail. + pub fn as_str(&self) -> &'static str { + match self { + KmsBackend::VaultKv2 => "vault-kv2", + KmsBackend::VaultTransit => "vault-transit", + KmsBackend::Local => "local", + KmsBackend::Static => "static", + } + } +} + /// Main KMS configuration #[derive(Debug, Clone, Serialize, Deserialize)] pub struct KmsConfig { diff --git a/crates/kms/src/deletion_worker.rs b/crates/kms/src/deletion_worker.rs index 3fca15b25..49594499a 100644 --- a/crates/kms/src/deletion_worker.rs +++ b/crates/kms/src/deletion_worker.rs @@ -22,12 +22,14 @@ //! every node of a deployment concurrently — a key is only ever removed while //! its (re-read) record is an expired pending deletion or a tombstone. +use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink}; use crate::backends::{ExpiredKeyRemoval, KmsBackend}; -use crate::types::{KeyStatus, ListKeysRequest}; +use crate::error::Result; +use crate::types::{KeyStatus, ListKeysRequest, OperationContext}; use async_trait::async_trait; use jiff::Zoned; use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio_util::sync::CancellationToken; use tracing::{debug, info, warn}; @@ -68,6 +70,8 @@ pub(crate) struct DeletionWorker { default_key_id: Option, reference_checker: Option>, interval: Duration, + backend_kind: &'static str, + audit_sink: Option>, } impl DeletionWorker { @@ -75,15 +79,24 @@ impl DeletionWorker { backend: Arc, default_key_id: Option, reference_checker: Option>, + backend_kind: &'static str, ) -> Self { Self { backend, default_key_id, reference_checker, + backend_kind, + audit_sink: None, interval: DEFAULT_SWEEP_INTERVAL, } } + /// Audit each irreversible key removal to `sink`. + pub(crate) fn with_audit_sink(mut self, sink: Option>) -> Self { + self.audit_sink = sink; + self + } + pub(crate) fn spawn(self, cancel: CancellationToken) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { self.run(cancel).await }) } @@ -168,15 +181,43 @@ impl DeletionWorker { // The backend re-checks state and deadline under its own write // synchronization, so a cancellation racing this sweep wins there. - match self.backend.remove_expired_key(key_id, now).await { - Ok(ExpiredKeyRemoval::Removed) => report.removed.push(key_id.to_string()), + let started = Instant::now(); + let outcome = self.backend.remove_expired_key(key_id, now).await; + match &outcome { + Ok(ExpiredKeyRemoval::Removed) => { + report.removed.push(key_id.to_string()); + self.audit_removal(key_id, started, &outcome); + } Ok(ExpiredKeyRemoval::StateChanged | ExpiredKeyRemoval::NotExpired) => report.skipped += 1, Err(error) => { warn!(key_id, %error, "failed to remove expired KMS key; will retry next sweep"); report.failed += 1; + self.audit_removal(key_id, started, &outcome); } } } + + /// Record an attempted destruction of key material. + /// + /// Only attempts that reached the backend are audited: a key the sweep + /// declined to touch (not yet due, still referenced, state changed under + /// us) was never at risk, and recording it as a deletion event would + /// drown the real ones. + fn audit_removal(&self, key_id: &str, started: Instant, outcome: &Result) { + let Some(sink) = self.audit_sink.as_ref() else { + return; + }; + + // Removal runs on a background sweep, so there is no request + // principal to attribute it to. + let context = OperationContext::internal(); + sink.emit( + KmsAuditRecord::new(KmsAuditOperation::DeleteKey, &context, self.backend_kind) + .with_key_id(Some(key_id)) + .with_latency(started.elapsed()) + .with_result(outcome), + ); + } } #[cfg(test)] @@ -216,7 +257,7 @@ mod tests { } fn worker(backend: Arc) -> DeletionWorker { - DeletionWorker::new(backend, None, None) + DeletionWorker::new(backend, None, None, "local") } fn after_window() -> Zoned { @@ -307,7 +348,7 @@ mod tests { schedule(&backend, &key_id).await; // Blocked while it is the configured default key. - let as_default = DeletionWorker::new(backend.clone(), Some(key_id.clone()), None); + let as_default = DeletionWorker::new(backend.clone(), Some(key_id.clone()), None, "local"); let report = as_default.sweep(&after_window()).await; assert_eq!(report.blocked, vec![key_id.clone()]); assert!(report.removed.is_empty()); @@ -317,6 +358,7 @@ mod tests { backend.clone(), None, Some(Arc::new(StaticReferences(vec!["bucket:sse-bucket".to_string()]))), + "local", ); let report = with_references.sweep(&after_window()).await; assert_eq!(report.blocked, vec![key_id.clone()]); @@ -327,11 +369,53 @@ mod tests { .expect("blocked key must still exist"); // Removed once nothing references it anymore. - let unreferenced = DeletionWorker::new(backend.clone(), None, Some(Arc::new(StaticReferences(Vec::new())))); + let unreferenced = DeletionWorker::new(backend.clone(), None, Some(Arc::new(StaticReferences(Vec::new()))), "local"); let report = unreferenced.sweep(&after_window()).await; assert_eq!(report.removed, vec![key_id.clone()]); } + #[tokio::test] + async fn only_attempted_removals_are_audited() { + #[derive(Default)] + struct CapturingSink { + records: std::sync::Mutex>, + } + + impl KmsAuditSink for CapturingSink { + fn emit(&self, record: KmsAuditRecord) { + self.records.lock().expect("sink lock should not be poisoned").push(record); + } + } + + let temp_dir = tempfile::tempdir().expect("temp dir"); + let backend = local_backend(&temp_dir).await; + let key_id = create_key(&backend, "audited-removal").await; + schedule(&backend, &key_id).await; + + let sink = Arc::new(CapturingSink::default()); + let worker = DeletionWorker::new(backend.clone(), None, None, "local").with_audit_sink(Some(sink.clone())); + + // A key that is not yet due was never at risk, so it produces no record. + worker.sweep(&Zoned::now()).await; + assert!( + sink.records.lock().expect("sink lock").is_empty(), + "a key the sweep declined to touch must not be audited as a deletion" + ); + + worker.sweep(&after_window()).await; + + let records = sink.records.lock().expect("sink lock"); + assert_eq!(records.len(), 1, "the removal should be audited exactly once"); + let record = &records[0]; + assert_eq!(record.operation, crate::audit::KmsAuditOperation::DeleteKey); + assert_eq!(record.event, rustfs_s3_types::EventName::KmsKeyDeleted); + assert_eq!(record.outcome, crate::audit::KmsAuditOutcome::Success); + assert_eq!(record.key_id.as_deref(), Some(key_id.as_str())); + assert_eq!(record.backend, "local"); + // Background work has no request principal to attribute. + assert_eq!(record.principal, OperationContext::INTERNAL_PRINCIPAL); + } + #[tokio::test] async fn deadline_survives_backend_restart_and_sweep_completes_it() { let temp_dir = tempfile::tempdir().expect("temp dir"); diff --git a/crates/kms/src/lib.rs b/crates/kms/src/lib.rs index 892ed5a35..a52ef3d52 100644 --- a/crates/kms/src/lib.rs +++ b/crates/kms/src/lib.rs @@ -65,6 +65,7 @@ // Core modules pub mod api_types; +pub mod audit; pub mod backends; pub mod backup; mod cache; @@ -86,6 +87,7 @@ pub use api_types::{ StartKmsResponse, StopKmsResponse, TagKeyRequest, TagKeyResponse, UntagKeyRequest, UntagKeyResponse, UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse, }; +pub use audit::{KmsAuditOperation, KmsAuditOutcome, KmsAuditRecord, KmsAuditSink, redact_encryption_context}; pub use config::*; pub use deletion_worker::DeletionReferenceChecker; pub use encryption::is_data_key_envelope; diff --git a/crates/kms/src/manager.rs b/crates/kms/src/manager.rs index 9d77e07b4..30355b75d 100644 --- a/crates/kms/src/manager.rs +++ b/crates/kms/src/manager.rs @@ -14,6 +14,7 @@ //! KMS manager for handling key operations and backend coordination +use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink}; use crate::backends::KmsBackend; use crate::cache::KmsCache; use crate::config::KmsConfig; @@ -21,9 +22,10 @@ use crate::error::Result; use crate::types::{ CancelKeyDeletionRequest, CancelKeyDeletionResponse, CreateKeyRequest, CreateKeyResponse, DecryptRequest, DecryptResponse, DeleteKeyRequest, DeleteKeyResponse, DescribeKeyRequest, DescribeKeyResponse, EncryptRequest, EncryptResponse, - GenerateDataKeyRequest, GenerateDataKeyResponse, ListKeysRequest, ListKeysResponse, + GenerateDataKeyRequest, GenerateDataKeyResponse, ListKeysRequest, ListKeysResponse, OperationContext, }; use std::sync::Arc; +use std::time::Instant; use tokio::sync::RwLock; /// KMS Manager coordinates operations between backends and caching @@ -33,6 +35,8 @@ pub struct KmsManager { cache: Arc>, default_key_id: Option, enable_cache: bool, + backend_kind: &'static str, + audit_sink: Option>, } impl KmsManager { @@ -44,16 +48,72 @@ impl KmsManager { cache, default_key_id: config.default_key_id, enable_cache: config.enable_cache, + backend_kind: config.backend.as_str(), + audit_sink: None, } } + /// Send an audit record for every management operation to `sink`. + /// + /// Without a sink the manager builds no records at all, so a deployment + /// that does not consume KMS audit records is unaffected. + pub fn with_audit_sink(mut self, sink: Arc) -> Self { + self.audit_sink = Some(sink); + self + } + /// Get the default key ID if configured pub fn get_default_key_id(&self) -> Option<&String> { self.default_key_id.as_ref() } + /// Emit an audit record for a completed management operation. + /// + /// Called after the operation resolved, so nothing here can change its + /// result; the record is built only when a sink is installed. + fn audit( + &self, + operation: KmsAuditOperation, + context: &OperationContext, + key_id: Option<&str>, + started: Instant, + result: &Result, + ) { + let Some(sink) = self.audit_sink.as_ref() else { + return; + }; + + sink.emit( + KmsAuditRecord::new(operation, context, self.backend_kind) + .with_key_id(key_id) + .with_latency(started.elapsed()) + .with_result(result), + ); + } + /// Create a new master key + /// + /// Audited as an internal operation; callers serving an authenticated + /// request should use [`Self::create_key_with_context`]. pub async fn create_key(&self, request: CreateKeyRequest) -> Result { + self.create_key_with_context(request, &OperationContext::internal()).await + } + + /// Create a new master key on behalf of `context`'s principal + pub async fn create_key_with_context( + &self, + request: CreateKeyRequest, + context: &OperationContext, + ) -> Result { + let started = Instant::now(); + let key_name = request.key_name.clone(); + let result = self.create_key_inner(request).await; + let key_id = result.as_ref().ok().map(|r| r.key_id.as_str()).or(key_name.as_deref()); + self.audit(KmsAuditOperation::CreateKey, context, key_id, started, &result); + result + } + + async fn create_key_inner(&self, request: CreateKeyRequest) -> Result { let response = self.backend.create_key(request).await?; // Cache the key metadata if enabled @@ -84,7 +144,27 @@ impl KmsManager { } /// Describe a key + /// + /// Audited as an internal operation; callers serving an authenticated + /// request should use [`Self::describe_key_with_context`]. pub async fn describe_key(&self, request: DescribeKeyRequest) -> Result { + self.describe_key_with_context(request, &OperationContext::internal()).await + } + + /// Describe a key on behalf of `context`'s principal + pub async fn describe_key_with_context( + &self, + request: DescribeKeyRequest, + context: &OperationContext, + ) -> Result { + let started = Instant::now(); + let key_id = request.key_id.clone(); + let result = self.describe_key_inner(request).await; + self.audit(KmsAuditOperation::DescribeKey, context, Some(&key_id), started, &result); + result + } + + async fn describe_key_inner(&self, request: DescribeKeyRequest) -> Result { // Check cache first if enabled if self.enable_cache { let cache = self.cache.read().await; @@ -109,8 +189,20 @@ impl KmsManager { } /// List keys + /// + /// Audited as an internal operation; callers serving an authenticated + /// request should use [`Self::list_keys_with_context`]. pub async fn list_keys(&self, request: ListKeysRequest) -> Result { - self.backend.list_keys(request).await + self.list_keys_with_context(request, &OperationContext::internal()).await + } + + /// List keys on behalf of `context`'s principal + pub async fn list_keys_with_context(&self, request: ListKeysRequest, context: &OperationContext) -> Result { + let started = Instant::now(); + let result = self.backend.list_keys(request).await; + // Listing spans keys, so the record carries no key id. + self.audit(KmsAuditOperation::ListKeys, context, None, started, &result); + result } /// Get cache statistics @@ -133,7 +225,27 @@ impl KmsManager { } /// Delete a key + /// + /// Audited as an internal operation; callers serving an authenticated + /// request should use [`Self::delete_key_with_context`]. pub async fn delete_key(&self, request: DeleteKeyRequest) -> Result { + self.delete_key_with_context(request, &OperationContext::internal()).await + } + + /// Delete a key on behalf of `context`'s principal + pub async fn delete_key_with_context( + &self, + request: DeleteKeyRequest, + context: &OperationContext, + ) -> Result { + let started = Instant::now(); + let key_id = request.key_id.clone(); + let result = self.delete_key_inner(request).await; + self.audit(KmsAuditOperation::ScheduleKeyDeletion, context, Some(&key_id), started, &result); + result + } + + async fn delete_key_inner(&self, request: DeleteKeyRequest) -> Result { let response = self.backend.delete_key(request).await?; // Remove from cache if enabled and key is being deleted @@ -146,7 +258,28 @@ impl KmsManager { } /// Cancel key deletion + /// + /// Audited as an internal operation; callers serving an authenticated + /// request should use [`Self::cancel_key_deletion_with_context`]. pub async fn cancel_key_deletion(&self, request: CancelKeyDeletionRequest) -> Result { + self.cancel_key_deletion_with_context(request, &OperationContext::internal()) + .await + } + + /// Cancel key deletion on behalf of `context`'s principal + pub async fn cancel_key_deletion_with_context( + &self, + request: CancelKeyDeletionRequest, + context: &OperationContext, + ) -> Result { + let started = Instant::now(); + let key_id = request.key_id.clone(); + let result = self.cancel_key_deletion_inner(request).await; + self.audit(KmsAuditOperation::CancelKeyDeletion, context, Some(&key_id), started, &result); + result + } + + async fn cancel_key_deletion_inner(&self, request: CancelKeyDeletionRequest) -> Result { let response = self.backend.cancel_key_deletion(request).await?; // Update cache if enabled @@ -159,24 +292,60 @@ impl KmsManager { } /// Enable a disabled key + /// + /// Audited as an internal operation; callers serving an authenticated + /// request should use [`Self::enable_key_with_context`]. pub async fn enable_key(&self, key_id: &str) -> Result<()> { - self.backend.enable_key(key_id).await?; - self.invalidate_cached_metadata(key_id).await; - Ok(()) + self.enable_key_with_context(key_id, &OperationContext::internal()).await + } + + /// Enable a disabled key on behalf of `context`'s principal + pub async fn enable_key_with_context(&self, key_id: &str, context: &OperationContext) -> Result<()> { + let started = Instant::now(); + let result = self.backend.enable_key(key_id).await; + if result.is_ok() { + self.invalidate_cached_metadata(key_id).await; + } + self.audit(KmsAuditOperation::EnableKey, context, Some(key_id), started, &result); + result } /// Disable a key; existing data remains decryptable + /// + /// Audited as an internal operation; callers serving an authenticated + /// request should use [`Self::disable_key_with_context`]. pub async fn disable_key(&self, key_id: &str) -> Result<()> { - self.backend.disable_key(key_id).await?; - self.invalidate_cached_metadata(key_id).await; - Ok(()) + self.disable_key_with_context(key_id, &OperationContext::internal()).await + } + + /// Disable a key on behalf of `context`'s principal + pub async fn disable_key_with_context(&self, key_id: &str, context: &OperationContext) -> Result<()> { + let started = Instant::now(); + let result = self.backend.disable_key(key_id).await; + if result.is_ok() { + self.invalidate_cached_metadata(key_id).await; + } + self.audit(KmsAuditOperation::DisableKey, context, Some(key_id), started, &result); + result } /// Rotate a key to a new version + /// + /// Audited as an internal operation; callers serving an authenticated + /// request should use [`Self::rotate_key_with_context`]. pub async fn rotate_key(&self, key_id: &str) -> Result<()> { - self.backend.rotate_key(key_id).await?; - self.invalidate_cached_metadata(key_id).await; - Ok(()) + self.rotate_key_with_context(key_id, &OperationContext::internal()).await + } + + /// Rotate a key on behalf of `context`'s principal + pub async fn rotate_key_with_context(&self, key_id: &str, context: &OperationContext) -> Result<()> { + let started = Instant::now(); + let result = self.backend.rotate_key(key_id).await; + if result.is_ok() { + self.invalidate_cached_metadata(key_id).await; + } + self.audit(KmsAuditOperation::RotateKey, context, Some(key_id), started, &result); + result } /// Drop cached metadata after a state mutation so the next describe @@ -208,11 +377,470 @@ impl KmsManager { #[cfg(test)] mod tests { use super::*; + use crate::audit::KmsAuditOutcome; use crate::backends::local::LocalKmsBackend; - use crate::types::{KeySpec, KeyState, KeyUsage}; + use crate::error::KmsError; + use crate::types::{KeyMetadata, KeySpec, KeyState, KeyUsage}; + use async_trait::async_trait; + use base64::Engine as _; + use jiff::Zoned; use std::collections::HashMap; + use std::sync::Mutex; use tempfile::tempdir; + /// Sink that keeps every record so tests can assert on the audit trail. + #[derive(Default)] + struct CapturingSink { + records: Mutex>, + } + + impl KmsAuditSink for CapturingSink { + fn emit(&self, record: KmsAuditRecord) { + self.records + .lock() + .expect("audit records lock should not be poisoned") + .push(record); + } + } + + impl CapturingSink { + fn records(&self) -> Vec { + self.records + .lock() + .expect("audit records lock should not be poisoned") + .clone() + } + + fn take_one(&self) -> KmsAuditRecord { + let mut records = self.records.lock().expect("audit records lock should not be poisoned"); + assert_eq!(records.len(), 1, "expected exactly one audit record, got {records:?}"); + records.remove(0) + } + } + + /// Backend whose management operations all succeed or all fail, so a + /// single test can drive every audited operation down both paths — + /// including operations no real backend supports on both. + struct ScriptedBackend { + failure: Option, + } + + impl ScriptedBackend { + fn succeeding() -> Self { + Self { failure: None } + } + + fn failing(failure: KmsError) -> Self { + Self { failure: Some(failure) } + } + + fn check(&self) -> Result<()> { + match &self.failure { + Some(failure) => Err(failure.clone()), + None => Ok(()), + } + } + + fn metadata(key_id: &str) -> KeyMetadata { + KeyMetadata { + key_id: key_id.to_string(), + key_state: KeyState::Enabled, + key_usage: KeyUsage::EncryptDecrypt, + description: None, + creation_date: Zoned::now(), + deletion_date: None, + origin: "RUSTFS_KMS".to_string(), + key_manager: "RUSTFS".to_string(), + tags: HashMap::new(), + } + } + } + + #[async_trait] + impl KmsBackend for ScriptedBackend { + async fn create_key(&self, request: CreateKeyRequest) -> Result { + self.check()?; + let key_id = request.key_name.unwrap_or_else(|| "scripted-key".to_string()); + Ok(CreateKeyResponse { + key_metadata: Self::metadata(&key_id), + key_id, + }) + } + + async fn encrypt(&self, _request: EncryptRequest) -> Result { + unimplemented!("data plane is not audited by the manager") + } + + async fn decrypt(&self, _request: DecryptRequest) -> Result { + unimplemented!("data plane is not audited by the manager") + } + + async fn generate_data_key(&self, _request: GenerateDataKeyRequest) -> Result { + unimplemented!("data plane is not audited by the manager") + } + + async fn describe_key(&self, request: DescribeKeyRequest) -> Result { + self.check()?; + Ok(DescribeKeyResponse { + key_metadata: Self::metadata(&request.key_id), + }) + } + + async fn list_keys(&self, _request: ListKeysRequest) -> Result { + self.check()?; + Ok(ListKeysResponse { + keys: Vec::new(), + next_marker: None, + truncated: false, + }) + } + + async fn delete_key(&self, request: DeleteKeyRequest) -> Result { + self.check()?; + Ok(DeleteKeyResponse { + key_metadata: Self::metadata(&request.key_id), + key_id: request.key_id, + deletion_date: None, + }) + } + + async fn cancel_key_deletion(&self, request: CancelKeyDeletionRequest) -> Result { + self.check()?; + Ok(CancelKeyDeletionResponse { + key_metadata: Self::metadata(&request.key_id), + key_id: request.key_id, + }) + } + + async fn enable_key(&self, _key_id: &str) -> Result<()> { + self.check() + } + + async fn disable_key(&self, _key_id: &str) -> Result<()> { + self.check() + } + + async fn rotate_key(&self, _key_id: &str) -> Result<()> { + self.check() + } + + async fn health_check(&self) -> Result { + Ok(true) + } + } + + const AUDITED_KEY_ID: &str = "audited-key"; + + /// Prefix length used when checking that a record embedded no *part* of a + /// secret. Long enough that a collision with unrelated text is not a real + /// concern. + const FRAGMENT_LEN: usize = 24; + + fn scripted_manager(backend: ScriptedBackend) -> (KmsManager, Arc) { + let temp_dir = tempdir().expect("Failed to create temp dir"); + let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults(); + let sink = Arc::new(CapturingSink::default()); + let manager = KmsManager::new(Arc::new(backend), config).with_audit_sink(sink.clone()); + (manager, sink) + } + + fn request_context() -> OperationContext { + OperationContext::new("arn:aws:iam::user/alice".to_string()) + .with_source_ip("192.0.2.10".to_string()) + .with_user_agent("rustfs-admin/1".to_string()) + .with_context("requestID".to_string(), "req-42".to_string()) + } + + /// Drive one management operation and return the single record it emitted. + async fn run_operation(manager: &KmsManager, operation: KmsAuditOperation, context: &OperationContext) -> Result<()> { + match operation { + KmsAuditOperation::CreateKey => manager + .create_key_with_context( + CreateKeyRequest { + key_name: Some(AUDITED_KEY_ID.to_string()), + ..Default::default() + }, + context, + ) + .await + .map(|_| ()), + KmsAuditOperation::DescribeKey => manager + .describe_key_with_context( + DescribeKeyRequest { + key_id: AUDITED_KEY_ID.to_string(), + }, + context, + ) + .await + .map(|_| ()), + KmsAuditOperation::ListKeys => manager + .list_keys_with_context(ListKeysRequest::default(), context) + .await + .map(|_| ()), + KmsAuditOperation::ScheduleKeyDeletion => manager + .delete_key_with_context( + DeleteKeyRequest { + key_id: AUDITED_KEY_ID.to_string(), + pending_window_in_days: None, + force_immediate: None, + }, + context, + ) + .await + .map(|_| ()), + KmsAuditOperation::CancelKeyDeletion => manager + .cancel_key_deletion_with_context( + CancelKeyDeletionRequest { + key_id: AUDITED_KEY_ID.to_string(), + }, + context, + ) + .await + .map(|_| ()), + KmsAuditOperation::EnableKey => manager.enable_key_with_context(AUDITED_KEY_ID, context).await, + KmsAuditOperation::DisableKey => manager.disable_key_with_context(AUDITED_KEY_ID, context).await, + KmsAuditOperation::RotateKey => manager.rotate_key_with_context(AUDITED_KEY_ID, context).await, + // Physical removal happens on the background sweep, not here. + KmsAuditOperation::DeleteKey => unreachable!("removal is audited by the deletion worker"), + } + } + + /// Every management operation the manager serves, in audit terms. + const AUDITED_OPERATIONS: [KmsAuditOperation; 8] = [ + KmsAuditOperation::CreateKey, + KmsAuditOperation::DescribeKey, + KmsAuditOperation::ListKeys, + KmsAuditOperation::ScheduleKeyDeletion, + KmsAuditOperation::CancelKeyDeletion, + KmsAuditOperation::EnableKey, + KmsAuditOperation::DisableKey, + KmsAuditOperation::RotateKey, + ]; + + #[tokio::test] + async fn every_management_operation_emits_a_complete_success_record() { + for operation in AUDITED_OPERATIONS { + let (manager, sink) = scripted_manager(ScriptedBackend::succeeding()); + let context = request_context(); + + let outer = Instant::now(); + run_operation(&manager, operation, &context) + .await + .unwrap_or_else(|error| panic!("{} should succeed: {error}", operation.as_str())); + let outer_elapsed = outer.elapsed(); + + let record = sink.take_one(); + assert_eq!(record.operation, operation); + assert_eq!(record.event, operation.event_name()); + assert_eq!(record.outcome, KmsAuditOutcome::Success); + assert_eq!(record.error_class, None); + assert_eq!(record.operation_id, context.operation_id); + assert_eq!(record.principal, "arn:aws:iam::user/alice"); + assert_eq!(record.source_ip.as_deref(), Some("192.0.2.10")); + assert_eq!(record.user_agent.as_deref(), Some("rustfs-admin/1")); + assert_eq!(record.backend, "local"); + assert_eq!(record.context.get("requestID").map(String::as_str), Some("req-42")); + assert!( + record.latency <= outer_elapsed, + "{} reported a latency larger than the call it measured", + operation.as_str() + ); + + // Listing spans keys; every other operation names the key it touched. + if operation == KmsAuditOperation::ListKeys { + assert_eq!(record.key_id, None); + } else { + assert_eq!(record.key_id.as_deref(), Some(AUDITED_KEY_ID)); + } + } + } + + #[tokio::test] + async fn every_management_operation_emits_a_failure_record() { + for operation in AUDITED_OPERATIONS { + let (manager, sink) = scripted_manager(ScriptedBackend::failing(KmsError::access_denied("denied by policy"))); + let context = request_context(); + + let error = run_operation(&manager, operation, &context) + .await + .expect_err("scripted backend should reject the operation"); + assert!(matches!(error, KmsError::AccessDenied { .. })); + + let record = sink.take_one(); + assert_eq!(record.operation, operation); + assert_eq!(record.event, operation.event_name()); + assert_eq!(record.outcome, KmsAuditOutcome::Failure); + assert_eq!(record.error_class, Some("access_denied")); + assert_eq!(record.operation_id, context.operation_id); + assert_eq!(record.principal, "arn:aws:iam::user/alice"); + assert_eq!(record.source_ip.as_deref(), Some("192.0.2.10")); + + // A denied create still has to name the key the caller asked for, + // otherwise the record cannot answer "what were they after". + if operation != KmsAuditOperation::ListKeys { + assert_eq!(record.key_id.as_deref(), Some(AUDITED_KEY_ID)); + } + } + } + + #[tokio::test] + async fn operations_are_unaffected_when_no_sink_is_installed() { + // The audit trail is optional; without a sink the manager must behave + // exactly as it did before records existed. + let temp_dir = tempdir().expect("Failed to create temp dir"); + let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults(); + let manager = KmsManager::new(Arc::new(ScriptedBackend::succeeding()), config); + + for operation in AUDITED_OPERATIONS { + run_operation(&manager, operation, &request_context()) + .await + .unwrap_or_else(|error| panic!("{} should succeed without a sink: {error}", operation.as_str())); + } + } + + #[tokio::test] + async fn context_free_calls_are_attributed_to_the_internal_principal() { + // Callers that have no authenticated identity must still be + // distinguishable from an identity we failed to record. + let (manager, sink) = scripted_manager(ScriptedBackend::succeeding()); + + manager + .describe_key(DescribeKeyRequest { + key_id: AUDITED_KEY_ID.to_string(), + }) + .await + .expect("describe should succeed"); + + let record = sink.take_one(); + assert_eq!(record.principal, OperationContext::INTERNAL_PRINCIPAL); + assert_eq!(record.source_ip, None); + assert_eq!(record.user_agent, None); + } + + #[tokio::test] + async fn unsupported_lifecycle_operations_are_audited_with_their_own_class() { + // The local backend has no version history, so rotation is a capability + // gap rather than a policy denial; the audit trail must say so. + let temp_dir = tempdir().expect("Failed to create temp dir"); + let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults(); + let backend = Arc::new(LocalKmsBackend::new(config.clone()).await.expect("Failed to create backend")); + let sink = Arc::new(CapturingSink::default()); + let manager = KmsManager::new(backend, config).with_audit_sink(sink.clone()); + + let key_id = manager + .create_key_with_context( + CreateKeyRequest { + key_name: Some("rotate-me".to_string()), + ..Default::default() + }, + &request_context(), + ) + .await + .expect("create should succeed") + .key_id; + + manager + .rotate_key_with_context(&key_id, &request_context()) + .await + .expect_err("local rotation must be rejected"); + + let records = sink.records(); + let rotate = records.last().expect("rotation should be audited"); + assert_eq!(rotate.operation, KmsAuditOperation::RotateKey); + assert_eq!(rotate.outcome, KmsAuditOutcome::Failure); + assert_eq!(rotate.error_class, Some("unsupported_capability")); + assert_eq!(rotate.key_id.as_deref(), Some(key_id.as_str())); + } + + /// Negative assertion: no audit record may reproduce key material. Driven + /// against the real local backend so the assertion covers whatever the + /// backend actually hands back, not a hand-written stand-in. + #[tokio::test] + async fn audit_records_never_reproduce_key_material() { + let temp_dir = tempdir().expect("Failed to create temp dir"); + let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults(); + let backend = Arc::new(LocalKmsBackend::new(config.clone()).await.expect("Failed to create backend")); + let sink = Arc::new(CapturingSink::default()); + let manager = KmsManager::new(backend, config).with_audit_sink(sink.clone()); + + let grant_token = "grant-token-cec4d4b5a1"; + let context = request_context(); + + let key_id = manager + .create_key_with_context( + CreateKeyRequest { + key_name: Some("material-key".to_string()), + ..Default::default() + }, + &context, + ) + .await + .expect("create should succeed") + .key_id; + + // Produce real key material, then keep driving the management plane so + // any record built afterwards is covered by the assertions below. + let data_key = manager + .generate_data_key(GenerateDataKeyRequest { + key_id: key_id.clone(), + key_spec: KeySpec::Aes256, + encryption_context: HashMap::from([("bucket".to_string(), "secrets".to_string())]), + }) + .await + .expect("data key generation should succeed"); + let decrypted = manager + .decrypt(DecryptRequest { + ciphertext: data_key.ciphertext_blob.clone(), + encryption_context: HashMap::from([("bucket".to_string(), "secrets".to_string())]), + grant_tokens: vec![grant_token.to_string()], + }) + .await + .expect("decrypt should succeed"); + + manager + .describe_key_with_context(DescribeKeyRequest { key_id: key_id.clone() }, &context) + .await + .expect("describe should succeed"); + manager + .list_keys_with_context(ListKeysRequest::default(), &context) + .await + .expect("list should succeed"); + manager + .disable_key_with_context(&key_id, &context) + .await + .expect("disable should succeed"); + manager + .enable_key_with_context(&key_id, &context) + .await + .expect("enable should succeed"); + + let base64 = base64::engine::general_purpose::STANDARD; + let encodings = |bytes: &[u8]| vec![hex::encode(bytes), base64.encode(bytes)]; + let mut forbidden = vec![grant_token.to_string()]; + forbidden.extend(encodings(&data_key.plaintext_key)); + forbidden.extend(encodings(&decrypted.plaintext)); + forbidden.extend(encodings(&data_key.ciphertext_blob)); + // Fragments catch a record that embedded only part of a blob. + let fragments: Vec = forbidden + .iter() + .filter(|secret| secret.len() > FRAGMENT_LEN) + .map(|secret| secret[..FRAGMENT_LEN].to_string()) + .collect(); + forbidden.extend(fragments); + + let records = sink.records(); + assert!(!records.is_empty(), "management operations should have been audited"); + for record in &records { + let rendered = format!("{record:?}"); + for secret in &forbidden { + assert!( + !rendered.contains(secret.as_str()), + "audit record leaked key material or a grant token: {rendered}" + ); + } + } + } + #[tokio::test] async fn test_manager_operations() { let temp_dir = tempdir().expect("Failed to create temp dir"); diff --git a/crates/kms/src/service.rs b/crates/kms/src/service.rs index 01902541b..20a10dd1c 100644 --- a/crates/kms/src/service.rs +++ b/crates/kms/src/service.rs @@ -112,6 +112,23 @@ impl ObjectEncryptionService { self.kms_manager.create_key(request).await } + /// Create a new master key on behalf of `context`'s principal + /// + /// # Arguments + /// * `request` - CreateKeyRequest with key parameters + /// * `context` - Identity and correlation data recorded in the audit trail + /// + /// # Returns + /// CreateKeyResponse with created key details + /// + pub async fn create_key_with_context( + &self, + request: CreateKeyRequest, + context: &OperationContext, + ) -> Result { + self.kms_manager.create_key_with_context(request, context).await + } + /// Describe a master key (delegates to KMS manager) /// /// # Arguments @@ -124,6 +141,23 @@ impl ObjectEncryptionService { self.kms_manager.describe_key(request).await } + /// Describe a master key on behalf of `context`'s principal + /// + /// # Arguments + /// * `request` - DescribeKeyRequest with key ID + /// * `context` - Identity and correlation data recorded in the audit trail + /// + /// # Returns + /// DescribeKeyResponse with key metadata + /// + pub async fn describe_key_with_context( + &self, + request: DescribeKeyRequest, + context: &OperationContext, + ) -> Result { + self.kms_manager.describe_key_with_context(request, context).await + } + /// List master keys (delegates to KMS manager) /// /// # Arguments @@ -136,6 +170,19 @@ impl ObjectEncryptionService { self.kms_manager.list_keys(request).await } + /// List master keys on behalf of `context`'s principal + /// + /// # Arguments + /// * `request` - ListKeysRequest with listing parameters + /// * `context` - Identity and correlation data recorded in the audit trail + /// + /// # Returns + /// ListKeysResponse with list of keys + /// + pub async fn list_keys_with_context(&self, request: ListKeysRequest, context: &OperationContext) -> Result { + self.kms_manager.list_keys_with_context(request, context).await + } + /// Generate a data encryption key (delegates to KMS manager) /// /// # Arguments diff --git a/crates/kms/src/service_manager.rs b/crates/kms/src/service_manager.rs index e80d7bc77..5acd1a15a 100644 --- a/crates/kms/src/service_manager.rs +++ b/crates/kms/src/service_manager.rs @@ -14,6 +14,7 @@ //! KMS service manager for dynamic configuration and runtime management +use crate::audit::KmsAuditSink; use crate::backends::vault_credentials::CredentialTaskHandle; use crate::backends::{KmsBackend, local::LocalKmsBackend}; use crate::config::{BackendConfig, KmsConfig}; @@ -171,6 +172,8 @@ pub struct KmsServiceManager { lifecycle_mutex: Arc>, /// External reference checker consulted before expired keys are removed deletion_reference_checker: std::sync::RwLock>>, + /// External sink receiving audit records for management operations + audit_sink: std::sync::RwLock>>, } impl KmsServiceManager { @@ -185,9 +188,26 @@ impl KmsServiceManager { version_counter: Arc::new(AtomicU64::new(0)), lifecycle_mutex: Arc::new(Mutex::new(())), deletion_reference_checker: std::sync::RwLock::new(None), + audit_sink: std::sync::RwLock::new(None), } } + /// Install the sink that receives an audit record for every KMS + /// management operation. Takes effect for services built by the next + /// start or reconfigure. + /// + /// Without a sink no records are built, so KMS operations are unaffected + /// when auditing is not configured. + pub fn set_audit_sink(&self, sink: Arc) { + if let Ok(mut slot) = self.audit_sink.write() { + *slot = Some(sink); + } + } + + fn audit_sink(&self) -> Option> { + self.audit_sink.read().ok().and_then(|slot| slot.clone()) + } + /// Install the reference checker consulted before the deletion worker /// removes an expired key. Takes effect for workers spawned by the next /// start or reconfigure. @@ -585,7 +605,11 @@ impl KmsServiceManager { }; // Create KMS manager - let kms_manager = Arc::new(KmsManager::new(backend, config.clone())); + let mut kms_manager = KmsManager::new(backend, config.clone()); + if let Some(sink) = self.audit_sink() { + kms_manager = kms_manager.with_audit_sink(sink); + } + let kms_manager = Arc::new(kms_manager); // Create encryption service let encryption_service = Arc::new(ObjectEncryptionService::new((*kms_manager).clone())); @@ -629,7 +653,13 @@ impl KmsServiceManager { return None; } let cancel = CancellationToken::new(); - let worker = DeletionWorker::new(backend, config.default_key_id.clone(), self.deletion_reference_checker()); + let worker = DeletionWorker::new( + backend, + config.default_key_id.clone(), + self.deletion_reference_checker(), + config.backend.as_str(), + ) + .with_audit_sink(self.audit_sink()); let task = worker.spawn(cancel.clone()); Some(Arc::new(DeletionWorkerHandle { cancel, diff --git a/crates/kms/src/types.rs b/crates/kms/src/types.rs index f469b5936..c0cdb6f9e 100644 --- a/crates/kms/src/types.rs +++ b/crates/kms/src/types.rs @@ -484,6 +484,20 @@ impl OperationContext { } } + /// Principal recorded for work the server starts on its own behalf. + /// + /// Namespaced so it can never collide with an access key or an IAM ARN, + /// which keeps "no human did this" distinguishable from "we lost track of + /// who did this" in the audit trail. + pub const INTERNAL_PRINCIPAL: &'static str = "rustfs:internal"; + + /// Context for an operation with no authenticated caller, such as + /// background maintenance or a call made while serving a data-plane + /// request that carries its own audit entry. + pub fn internal() -> Self { + Self::new(Self::INTERNAL_PRINCIPAL.to_string()) + } + /// Add additional context /// /// # Arguments diff --git a/crates/notify/src/rules/mod.rs b/crates/notify/src/rules/mod.rs index 28608618a..fe2c87e25 100644 --- a/crates/notify/src/rules/mod.rs +++ b/crates/notify/src/rules/mod.rs @@ -22,6 +22,8 @@ mod pattern_rules_test; #[cfg(test)] mod pattern_test; mod rules_map; +#[cfg(test)] +mod rules_map_test; mod subscriber_index; mod subscriber_snapshot; mod target_id_set; diff --git a/crates/notify/src/rules/rules_map_test.rs b/crates/notify/src/rules/rules_map_test.rs new file mode 100644 index 000000000..93ff68cf3 --- /dev/null +++ b/crates/notify/src/rules/rules_map_test.rs @@ -0,0 +1,104 @@ +// Copyright 2024 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. + +//! Regressions for rule matching against the KMS event namespace. +//! +//! KMS events (backlog#1583) are appended to `EventName` for the audit sink, +//! but they must stay invisible to bucket notification rules: a subscriber +//! asking for `s3:ObjectCreated:*` — or for everything — must never receive +//! key management activity. + +use super::RulesMap; +use rustfs_s3_types::EventName; +use rustfs_targets::arn::TargetID; + +const KMS_EVENTS: &[EventName] = &[ + EventName::KmsKeyCreated, + EventName::KmsKeyRotated, + EventName::KmsKeyEnabled, + EventName::KmsKeyDisabled, + EventName::KmsKeyDeletionScheduled, + EventName::KmsKeyDeletionCancelled, + EventName::KmsKeyDeleted, + EventName::KmsKeyAccessed, +]; + +fn test_target() -> TargetID { + TargetID::new("primary".to_string(), "webhook".to_string()) +} + +fn rules_for(events: &[EventName]) -> RulesMap { + let mut rules = RulesMap::new(); + rules.add_rule_config(events, String::new(), test_target()); + rules +} + +#[test] +fn s3_wildcard_subscriptions_do_not_match_kms_events() { + let rules = rules_for(&[ + EventName::ObjectCreatedAll, + EventName::ObjectAccessedAll, + EventName::ObjectRemovedAll, + EventName::ObjectTaggingAll, + EventName::ObjectReplicationAll, + EventName::ObjectRestoreAll, + EventName::ObjectTransitionAll, + EventName::LifecycleExpirationAll, + EventName::ObjectScannerAll, + ]); + + for event in KMS_EVENTS { + assert!(!rules.has_subscriber(event), "{event} must not have an S3 wildcard subscriber"); + assert!( + rules.match_rules(*event, "any/object").is_empty(), + "{event} must not match any S3 wildcard rule" + ); + } +} + +#[test] +fn everything_subscription_does_not_match_kms_events() { + let rules = rules_for(&[EventName::Everything]); + + // Sanity: the catch-all really is subscribed to the S3 surface. + assert!(rules.has_subscriber(&EventName::ObjectCreatedPut)); + + for event in KMS_EVENTS { + assert!(!rules.has_subscriber(event), "{event} must stay outside the s3 catch-all"); + assert!( + rules.match_rules(*event, "any/object").is_empty(), + "{event} must not match the s3 catch-all rule" + ); + } +} + +#[test] +fn kms_subscription_does_not_leak_into_s3_matching() { + // The inverse direction: even an explicit KMS subscription must not widen + // the mask so that S3 events start matching a KMS-only rule. + let rules = rules_for(KMS_EVENTS); + + for event in [ + EventName::ObjectCreatedPut, + EventName::ObjectRemovedDelete, + EventName::ObjectAccessedGet, + EventName::BucketCreated, + ] { + assert!(!rules.has_subscriber(&event), "{event} must not match a KMS-only rule"); + assert!( + rules.match_rules(event, "any/object").is_empty(), + "{event} must not match a KMS-only rule" + ); + } +} diff --git a/crates/s3-types/src/event_name.rs b/crates/s3-types/src/event_name.rs index fe0ed12f4..663201905 100644 --- a/crates/s3-types/src/event_name.rs +++ b/crates/s3-types/src/event_name.rs @@ -86,6 +86,20 @@ pub enum EventName { ObjectRemovedAbortMultipartUpload, ObjectCreatedCreateMultipartUpload, ObjectRemovedDeleteObjects, + + // KMS management-plane events. They travel to the audit sink only and are + // never produced by the bucket notification path, so no compound `s3:` + // event expands to them. New variants must keep being appended here: the + // discriminant of every preceding variant is a `mask()` bit position, and + // inserting in the middle would silently renumber existing bits. + KmsKeyCreated, + KmsKeyRotated, + KmsKeyEnabled, + KmsKeyDisabled, + KmsKeyDeletionScheduled, + KmsKeyDeletionCancelled, + KmsKeyDeleted, + KmsKeyAccessed, } // Single event type sequential array for Everything.expand() @@ -186,6 +200,17 @@ impl EventName { "s3:Scanner:LargeVersions" => Ok(EventName::ScannerLargeVersions), "s3:Scanner:BigPrefix" => Ok(EventName::ScannerBigPrefix), "s3:Scanner:*" => Ok(EventName::ObjectScannerAll), + // KMS events use their own namespace so a `s3:` wildcard in a bucket + // notification config can never select them. They still round-trip + // because audit entries are persisted and replayed by the store targets. + "kms:Key:Created" => Ok(EventName::KmsKeyCreated), + "kms:Key:Rotated" => Ok(EventName::KmsKeyRotated), + "kms:Key:Enabled" => Ok(EventName::KmsKeyEnabled), + "kms:Key:Disabled" => Ok(EventName::KmsKeyDisabled), + "kms:Key:DeletionScheduled" => Ok(EventName::KmsKeyDeletionScheduled), + "kms:Key:DeletionCancelled" => Ok(EventName::KmsKeyDeletionCancelled), + "kms:Key:Deleted" => Ok(EventName::KmsKeyDeleted), + "kms:Key:Accessed" => Ok(EventName::KmsKeyAccessed), // `Everything` has no string representation (`as_str` yields ""), so it // cannot be parsed back from a string. Every other variant round-trips. _ => Err(ParseEventNameError(s.to_string())), @@ -251,6 +276,14 @@ impl EventName { EventName::ObjectRemovedAbortMultipartUpload => "s3:ObjectRemoved:AbortMultipartUpload", EventName::ObjectCreatedCreateMultipartUpload => "s3:ObjectCreated:CreateMultipartUpload", EventName::ObjectRemovedDeleteObjects => "s3:ObjectRemoved:DeleteObjects", + EventName::KmsKeyCreated => "kms:Key:Created", + EventName::KmsKeyRotated => "kms:Key:Rotated", + EventName::KmsKeyEnabled => "kms:Key:Enabled", + EventName::KmsKeyDisabled => "kms:Key:Disabled", + EventName::KmsKeyDeletionScheduled => "kms:Key:DeletionScheduled", + EventName::KmsKeyDeletionCancelled => "kms:Key:DeletionCancelled", + EventName::KmsKeyDeleted => "kms:Key:Deleted", + EventName::KmsKeyAccessed => "kms:Key:Accessed", } } @@ -357,6 +390,25 @@ impl EventName { | EventName::ObjectRemovedDeleteObjects ) } + + /// Returns `true` for KMS management-plane events. + /// + /// These are audit-only: they are never emitted through the bucket + /// notification pipeline, and no `s3:` event selector expands to them. + #[inline] + pub fn is_kms(&self) -> bool { + matches!( + self, + EventName::KmsKeyCreated + | EventName::KmsKeyRotated + | EventName::KmsKeyEnabled + | EventName::KmsKeyDisabled + | EventName::KmsKeyDeletionScheduled + | EventName::KmsKeyDeletionCancelled + | EventName::KmsKeyDeleted + | EventName::KmsKeyAccessed + ) + } } /// Returns the S3 notification event schema version for a given event. @@ -570,6 +622,26 @@ mod tests { EventName::ObjectRemovedAbortMultipartUpload, EventName::ObjectCreatedCreateMultipartUpload, EventName::ObjectRemovedDeleteObjects, + EventName::KmsKeyCreated, + EventName::KmsKeyRotated, + EventName::KmsKeyEnabled, + EventName::KmsKeyDisabled, + EventName::KmsKeyDeletionScheduled, + EventName::KmsKeyDeletionCancelled, + EventName::KmsKeyDeleted, + EventName::KmsKeyAccessed, + ]; + + /// Every KMS management-plane event. + const KMS_EVENT_NAMES: &[EventName] = &[ + EventName::KmsKeyCreated, + EventName::KmsKeyRotated, + EventName::KmsKeyEnabled, + EventName::KmsKeyDisabled, + EventName::KmsKeyDeletionScheduled, + EventName::KmsKeyDeletionCancelled, + EventName::KmsKeyDeleted, + EventName::KmsKeyAccessed, ]; /// Regression for backlog#965: `mask()` used to recurse forever for the @@ -682,6 +754,54 @@ mod tests { } } + /// KMS events are audit-plane only. No `s3:` selector — including the + /// catch-all `Everything` and every compound "All" type — may share a bit + /// with them, otherwise a bucket notification rule would silently start + /// matching KMS activity. + #[test] + fn test_kms_event_masks_are_disjoint_from_every_s3_selector() { + let s3_selectors: Vec = ALL_EVENT_NAMES.iter().copied().filter(|ev| !ev.is_kms()).collect(); + + let mut seen = 0u64; + for kms in KMS_EVENT_NAMES { + let mask = kms.mask(); + assert_ne!(mask, 0, "KMS event {kms} must have a non-zero mask"); + assert_eq!(seen & mask, 0, "KMS event {kms} mask overlaps another KMS event"); + seen |= mask; + + for s3 in &s3_selectors { + assert_eq!(s3.mask() & mask, 0, "KMS event {kms} mask collides with S3 selector {s3}"); + } + } + } + + /// KMS event names must live in their own namespace so that neither a + /// `s3:` prefix filter nor an `s3:...:*` wildcard can select them. + #[test] + fn test_kms_event_names_are_outside_the_s3_namespace() { + for ev in KMS_EVENT_NAMES { + assert!(ev.is_kms(), "{ev} should be classified as a KMS event"); + assert!(ev.as_str().starts_with("kms:"), "unexpected KMS event name {:?}", ev.as_str()); + assert_eq!(EventName::parse(ev.as_str()).as_ref(), Ok(ev), "KMS event {ev} must round-trip"); + assert_eq!(ev.expand(), vec![*ev], "KMS event {ev} must expand to itself only"); + } + + for ev in ALL_EVENT_NAMES.iter().filter(|ev| !ev.is_kms()) { + assert!(!ev.as_str().starts_with("kms:"), "{ev} must not claim the KMS namespace"); + } + } + + /// `mask()` shifts by `discriminant - 1`, so the enum may hold at most 64 + /// mask-bearing variants. Appending past that overflows the shift instead + /// of failing loudly, so keep the budget check next to the variants. + #[test] + fn test_mask_bit_budget_is_not_exhausted() { + for ev in ALL_EVENT_NAMES { + let value = *ev as u32; + assert!(value <= 64, "{ev} has discriminant {value}; mask() only has 64 bits to hand out"); + } + } + /// `Everything` must cover every sequential single-type bit. #[test] fn test_everything_mask_covers_all_single_types() { diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index 4272afa7d..5f22c1f65 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -27,6 +27,9 @@ checked_files=( "rustfs/src/storage/rpc/http_service.rs" "rustfs/src/storage/rpc/node_service.rs" "crates/kms/src/config.rs" + "crates/kms/src/audit.rs" + "crates/kms/src/manager.rs" + "crates/kms/src/deletion_worker.rs" "crates/audit/src/pipeline.rs" "crates/audit/src/system.rs" "crates/audit/src/global.rs"