feat(kms): wire OperationContext into an audit event contract

`OperationContext` carried identity and correlation data that nothing
ever constructed or read, so key creation, rotation and deletion left no
trace of who requested them.

Give every management operation a `*_with_context` entry point that
builds a `KmsAuditRecord` and hands it to an installed `KmsAuditSink`.
The record answers who acted on which key, against which backend, with
what outcome and error class, and how long it took. Existing entry points
delegate with an explicitly internal principal, so an unattributed call
is distinguishable from an identity that was lost.

The backend trait is untouched: auditing sits at the layer that has an
identity, mirroring how metrics sit at the backend choke point. The sink
is optional and no records are built without one, so a deployment that
does not consume them is unaffected. Records also cover the background
sweep that destroys expired key material, which is otherwise the least
observable operation in the crate.

Redaction is structural rather than advisory: a record has no field that
can hold key material, and caller-supplied encryption context passes
through an allowlist that digests everything it does not recognise.
Tenant attribution is deliberately left out until tenancy is modelled.

Refs rustfs/backlog#1583 (part of rustfs/backlog#1562)

Cargo.lock

@@ -9523,6 +9523,7 @@ dependencies = [
  "moka",
  "rand 0.10.2",
  "reqwest",
+ "rustfs-s3-types",
  "rustfs-security-governance",
  "rustfs-utils",
  "rustify",

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 }

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<String>,
+    /// Client user agent, when the caller supplied one.
+    pub user_agent: Option<String>,
+    /// Key the operation acted on. `None` for operations that span keys, such
+    /// as listing.
+    pub key_id: Option<String>,
+    /// 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<u32>,
+    /// 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<u32>,
+    /// Server-supplied correlation values carried by the operation context.
+    /// Also the reserved slot for tenant attribution.
+    pub context: BTreeMap<String, String>,
+    /// Caller-supplied encryption context, after [`redact_encryption_context`].
+    pub encryption_context: BTreeMap<String, String>,
+}
+
+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<impl Into<String>>) -> 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<u32>) -> 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<T>(mut self, result: &crate::error::Result<T>) -> 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<String, String>) -> 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<String, String>) -> BTreeMap<String, String> {
+    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"));
+    }
+}

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 {

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<String>,
     reference_checker: Option<Arc<dyn DeletionReferenceChecker>>,
     interval: Duration,
+    backend_kind: &'static str,
+    audit_sink: Option<Arc<dyn KmsAuditSink>>,
 }
 
 impl DeletionWorker {
@@ -75,15 +79,24 @@ impl DeletionWorker {
         backend: Arc<dyn KmsBackend>,
         default_key_id: Option<String>,
         reference_checker: Option<Arc<dyn DeletionReferenceChecker>>,
+        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<Arc<dyn KmsAuditSink>>) -> 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<ExpiredKeyRemoval>) {
+        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<LocalKmsBackend>) -> 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<Vec<KmsAuditRecord>>,
+        }
+
+        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");

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;

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<RwLock<KmsCache>>,
     default_key_id: Option<String>,
     enable_cache: bool,
+    backend_kind: &'static str,
+    audit_sink: Option<Arc<dyn KmsAuditSink>>,
 }
 
 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<dyn KmsAuditSink>) -> 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<T>(
+        &self,
+        operation: KmsAuditOperation,
+        context: &OperationContext,
+        key_id: Option<&str>,
+        started: Instant,
+        result: &Result<T>,
+    ) {
+        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<CreateKeyResponse> {
+        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<CreateKeyResponse> {
+        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<CreateKeyResponse> {
         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<DescribeKeyResponse> {
+        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<DescribeKeyResponse> {
+        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<DescribeKeyResponse> {
         // 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<ListKeysResponse> {
-        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<ListKeysResponse> {
+        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<DeleteKeyResponse> {
+        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<DeleteKeyResponse> {
+        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<DeleteKeyResponse> {
         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<CancelKeyDeletionResponse> {
+        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<CancelKeyDeletionResponse> {
+        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<CancelKeyDeletionResponse> {
         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<Vec<KmsAuditRecord>>,
+    }
+
+    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<KmsAuditRecord> {
+            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<KmsError>,
+    }
+
+    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<CreateKeyResponse> {
+            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<EncryptResponse> {
+            unimplemented!("data plane is not audited by the manager")
+        }
+
+        async fn decrypt(&self, _request: DecryptRequest) -> Result<DecryptResponse> {
+            unimplemented!("data plane is not audited by the manager")
+        }
+
+        async fn generate_data_key(&self, _request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> {
+            unimplemented!("data plane is not audited by the manager")
+        }
+
+        async fn describe_key(&self, request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
+            self.check()?;
+            Ok(DescribeKeyResponse {
+                key_metadata: Self::metadata(&request.key_id),
+            })
+        }
+
+        async fn list_keys(&self, _request: ListKeysRequest) -> Result<ListKeysResponse> {
+            self.check()?;
+            Ok(ListKeysResponse {
+                keys: Vec::new(),
+                next_marker: None,
+                truncated: false,
+            })
+        }
+
+        async fn delete_key(&self, request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
+            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<CancelKeyDeletionResponse> {
+            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<bool> {
+            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<CapturingSink>) {
+        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<String> = 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");

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<CreateKeyResponse> {
+        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<DescribeKeyResponse> {
+        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<ListKeysResponse> {
+        self.kms_manager.list_keys_with_context(request, context).await
+    }
+
     /// Generate a data encryption key (delegates to KMS manager)
     ///
     /// # Arguments

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<Mutex<()>>,
     /// External reference checker consulted before expired keys are removed
     deletion_reference_checker: std::sync::RwLock<Option<Arc<dyn DeletionReferenceChecker>>>,
+    /// External sink receiving audit records for management operations
+    audit_sink: std::sync::RwLock<Option<Arc<dyn KmsAuditSink>>>,
 }
 
 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<dyn KmsAuditSink>) {
+        if let Ok(mut slot) = self.audit_sink.write() {
+            *slot = Some(sink);
+        }
+    }
+
+    fn audit_sink(&self) -> Option<Arc<dyn KmsAuditSink>> {
+        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,

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

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;

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"
+        );
+    }
+}

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<EventName> = 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() {

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"