feat(kms): add backend capability discovery (#5485)

crates/kms/src/backends/local.rs

@@ -14,7 +14,7 @@
 
 //! Local file-based KMS backend implementation
 
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient};
 use crate::config::KmsConfig;
 use crate::config::LocalConfig;
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -1492,6 +1492,17 @@ impl KmsBackend for LocalKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Rotation stays unadvertised until historical key versions can be
+        // retained (see LocalKmsClient::rotate_key); without version history
+        // there is also no versioning capability. Deletion deadlines are not
+        // yet persisted across restarts, but scheduling itself is supported.
+        BackendCapabilities::minimal()
+            .with_enable_disable(true)
+            .with_schedule_deletion(true)
+            .with_physical_delete(true)
+    }
 }
 
 #[cfg(test)]

crates/kms/src/backends/mod.rs

@@ -17,6 +17,7 @@
 use crate::error::Result;
 use crate::types::*;
 use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
 pub mod local;
@@ -184,6 +185,16 @@ pub trait KmsBackend: Send + Sync {
 
     /// Health check
     async fn health_check(&self) -> Result<bool>;
+
+    /// Report which operations this backend actually supports.
+    ///
+    /// The default is conservative: only the operations every backend is
+    /// required to implement by this trait are advertised. Optional lifecycle
+    /// operations (rotation, enable/disable, deletion scheduling, ...) must be
+    /// opted in by overriding this method.
+    fn capabilities(&self) -> BackendCapabilities {
+        BackendCapabilities::minimal()
+    }
 }
 
 /// Information about a KMS backend
@@ -237,3 +248,225 @@ impl BackendInfo {
         self
     }
 }
+
+/// Set of operations a KMS backend supports.
+///
+/// Reported by [`KmsBackend::capabilities`] so callers (manager, admin API)
+/// can discover what the active backend can do without probing individual
+/// operations. Marked `#[non_exhaustive]` so new capability flags can be
+/// added without breaking downstream code; construct values through
+/// [`BackendCapabilities::minimal`] and the `with_*` builders.
+#[non_exhaustive]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BackendCapabilities {
+    /// Direct encryption of caller-provided plaintext with a master key
+    pub encrypt: bool,
+    /// Decryption of previously produced ciphertext
+    pub decrypt: bool,
+    /// Data encryption key (DEK) generation
+    pub generate_data_key: bool,
+    /// Key rotation that retains prior versions for decryption
+    pub rotate: bool,
+    /// Enabling and disabling keys
+    pub enable_disable: bool,
+    /// Scheduling key deletion with a pending window
+    pub schedule_deletion: bool,
+    /// Multiple key versions addressable after rotation
+    pub versioning: bool,
+    /// Irreversible physical deletion of key material
+    pub physical_delete: bool,
+}
+
+impl BackendCapabilities {
+    /// Conservative baseline: only the operations that every [`KmsBackend`]
+    /// implementation is required to provide by the trait. All optional
+    /// lifecycle capabilities default to unsupported.
+    pub const fn minimal() -> Self {
+        Self {
+            encrypt: true,
+            decrypt: true,
+            generate_data_key: true,
+            rotate: false,
+            enable_disable: false,
+            schedule_deletion: false,
+            versioning: false,
+            physical_delete: false,
+        }
+    }
+
+    /// Set whether direct encryption is supported
+    pub const fn with_encrypt(mut self, encrypt: bool) -> Self {
+        self.encrypt = encrypt;
+        self
+    }
+
+    /// Set whether decryption is supported
+    pub const fn with_decrypt(mut self, decrypt: bool) -> Self {
+        self.decrypt = decrypt;
+        self
+    }
+
+    /// Set whether data key generation is supported
+    pub const fn with_generate_data_key(mut self, generate_data_key: bool) -> Self {
+        self.generate_data_key = generate_data_key;
+        self
+    }
+
+    /// Set whether version-retaining key rotation is supported
+    pub const fn with_rotate(mut self, rotate: bool) -> Self {
+        self.rotate = rotate;
+        self
+    }
+
+    /// Set whether enabling/disabling keys is supported
+    pub const fn with_enable_disable(mut self, enable_disable: bool) -> Self {
+        self.enable_disable = enable_disable;
+        self
+    }
+
+    /// Set whether scheduled deletion with a pending window is supported
+    pub const fn with_schedule_deletion(mut self, schedule_deletion: bool) -> Self {
+        self.schedule_deletion = schedule_deletion;
+        self
+    }
+
+    /// Set whether multiple key versions are supported
+    pub const fn with_versioning(mut self, versioning: bool) -> Self {
+        self.versioning = versioning;
+        self
+    }
+
+    /// Set whether physical deletion of key material is supported
+    pub const fn with_physical_delete(mut self, physical_delete: bool) -> Self {
+        self.physical_delete = physical_delete;
+        self
+    }
+}
+
+impl Default for BackendCapabilities {
+    fn default() -> Self {
+        Self::minimal()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::KmsConfig;
+    use base64::Engine as _;
+    use base64::engine::general_purpose::STANDARD as BASE64;
+
+    /// Backend that implements only the trait-mandated operations and relies
+    /// on the default `capabilities` implementation.
+    struct MinimalBackend;
+
+    #[async_trait]
+    impl KmsBackend for MinimalBackend {
+        async fn create_key(&self, _request: CreateKeyRequest) -> Result<CreateKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn encrypt(&self, _request: EncryptRequest) -> Result<EncryptResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn decrypt(&self, _request: DecryptRequest) -> Result<DecryptResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn generate_data_key(&self, _request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn describe_key(&self, _request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn list_keys(&self, _request: ListKeysRequest) -> Result<ListKeysResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn delete_key(&self, _request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn cancel_key_deletion(&self, _request: CancelKeyDeletionRequest) -> Result<CancelKeyDeletionResponse> {
+            unimplemented!("not exercised by capability tests")
+        }
+
+        async fn health_check(&self) -> Result<bool> {
+            Ok(true)
+        }
+    }
+
+    fn capabilities_snapshot(capabilities: BackendCapabilities) -> std::collections::BTreeMap<String, bool> {
+        serde_json::from_value(serde_json::to_value(capabilities).expect("capabilities should serialize"))
+            .expect("capabilities should deserialize into a flat bool map")
+    }
+
+    #[test]
+    fn default_capabilities_are_conservative() {
+        let capabilities = MinimalBackend.capabilities();
+        assert_eq!(capabilities, BackendCapabilities::minimal());
+        assert_eq!(capabilities, BackendCapabilities::default());
+
+        // The conservative baseline advertises only trait-mandated operations.
+        assert!(capabilities.encrypt);
+        assert!(capabilities.decrypt);
+        assert!(capabilities.generate_data_key);
+        assert!(!capabilities.rotate);
+        assert!(!capabilities.enable_disable);
+        assert!(!capabilities.schedule_deletion);
+        assert!(!capabilities.versioning);
+        assert!(!capabilities.physical_delete);
+    }
+
+    #[tokio::test]
+    async fn local_backend_capabilities_golden() {
+        let temp_dir = tempfile::tempdir().expect("temp dir should be created");
+        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+        let backend = local::LocalKmsBackend::new(config).await.expect("local backend should build");
+
+        insta::assert_json_snapshot!("local_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+
+    #[tokio::test]
+    async fn vault_kv2_backend_capabilities_golden() {
+        let config = KmsConfig::vault(
+            url::Url::parse("http://127.0.0.1:8200").expect("vault URL should parse"),
+            "dev-token".to_string(),
+        )
+        .with_insecure_development_defaults();
+        // Constructing the client performs no network I/O with token auth.
+        let backend = vault::VaultKmsBackend::new(config)
+            .await
+            .expect("vault kv2 backend should build");
+
+        insta::assert_json_snapshot!("vault_kv2_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+
+    #[tokio::test]
+    async fn vault_transit_backend_capabilities_golden() {
+        let config = KmsConfig::vault_transit(
+            url::Url::parse("http://127.0.0.1:8200").expect("vault URL should parse"),
+            "dev-token".to_string(),
+        )
+        .with_insecure_development_defaults();
+        // Constructing the client performs no network I/O with token auth.
+        let backend = vault_transit::VaultTransitKmsBackend::new(config)
+            .await
+            .expect("vault transit backend should build");
+
+        insta::assert_json_snapshot!("vault_transit_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+
+    #[tokio::test]
+    async fn static_backend_capabilities_golden() {
+        let config = KmsConfig::static_kms("static-key".to_string(), BASE64.encode([0u8; 32]));
+        let backend = static_kms::StaticKmsBackend::new(config)
+            .await
+            .expect("static backend should build");
+
+        insta::assert_json_snapshot!("static_backend_capabilities", capabilities_snapshot(backend.capabilities()));
+    }
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__local_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": true,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": true,
+  "rotate": false,
+  "schedule_deletion": true,
+  "versioning": false
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__static_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": false,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": false,
+  "rotate": false,
+  "schedule_deletion": false,
+  "versioning": false
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_kv2_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": true,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": true,
+  "rotate": false,
+  "schedule_deletion": true,
+  "versioning": false
+}

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_transit_backend_capabilities.snap

@@ -0,0 +1,14 @@
+---
+source: crates/kms/src/backends/mod.rs
+expression: capabilities_snapshot(backend.capabilities())
+---
+{
+  "decrypt": true,
+  "enable_disable": true,
+  "encrypt": true,
+  "generate_data_key": true,
+  "physical_delete": true,
+  "rotate": true,
+  "schedule_deletion": true,
+  "versioning": true
+}

crates/kms/src/backends/static_kms.rs

@@ -21,7 +21,7 @@
 //!
 //! encrypted_data(plaintext_len+16) || nonce (12 bytes)
 
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient};
 use crate::config::{BackendConfig, KmsConfig};
 use crate::encryption::DataKeyEnvelope;
 use crate::error::{KmsError, Result};
@@ -435,6 +435,12 @@ impl KmsBackend for StaticKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         Ok(true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Static KMS is a read-only single-key backend: it only performs
+        // cryptographic operations and rejects every lifecycle mutation.
+        BackendCapabilities::minimal()
+    }
 }
 
 #[cfg(test)]

crates/kms/src/backends/vault.rs

@@ -15,7 +15,7 @@
 //! Vault-based KMS backend implementation using vaultrs
 
 use crate::backends::vault_credentials::{VaultClientHandle, VaultConnectionSettings, VaultCredentialProvider, token_source_for};
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient};
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
 use crate::error::{KmsError, Result};
@@ -1095,6 +1095,16 @@ impl KmsBackend for VaultKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Rotation is unadvertised: the KV2 backend cannot rotate without
+        // replacing key material in place, and no historical versions are
+        // retained, so versioning is unsupported as well.
+        BackendCapabilities::minimal()
+            .with_enable_disable(true)
+            .with_schedule_deletion(true)
+            .with_physical_delete(true)
+    }
 }
 
 #[cfg(test)]

crates/kms/src/backends/vault_transit.rs

@@ -15,7 +15,7 @@
 //! Vault Transit-based KMS backend.
 
 use crate::backends::vault_credentials::{VaultClientHandle, VaultConnectionSettings, VaultCredentialProvider, token_source_for};
-use crate::backends::{BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient};
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
 use crate::error::{KmsError, Result};
@@ -762,6 +762,18 @@ impl KmsBackend for VaultTransitKmsBackend {
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
+
+    fn capabilities(&self) -> BackendCapabilities {
+        // Vault Transit natively supports version-retaining rotation, keeps
+        // prior versions addressable for decryption, and allows physical
+        // deletion once a key is pending deletion.
+        BackendCapabilities::minimal()
+            .with_rotate(true)
+            .with_enable_disable(true)
+            .with_schedule_deletion(true)
+            .with_versioning(true)
+            .with_physical_delete(true)
+    }
 }
 
 #[cfg(test)]

crates/kms/src/error.rs

@@ -128,6 +128,10 @@ pub enum KmsError {
     /// Backup/restore bundle contract violation; see [`crate::backup::BackupError`]
     #[error(transparent)]
     Backup(#[from] crate::backup::BackupError),
+
+    /// Operation is not supported by the active KMS backend
+    #[error("Operation '{operation}' is not supported by KMS backend '{backend}'")]
+    UnsupportedCapability { backend: String, operation: String },
 }
 
 impl KmsError {
@@ -269,6 +273,14 @@ impl KmsError {
             version,
         }
     }
+
+    /// Create an unsupported capability error
+    pub fn unsupported_capability<S1: Into<String>, S2: Into<String>>(backend: S1, operation: S2) -> Self {
+        Self::UnsupportedCapability {
+            backend: backend.into(),
+            operation: operation.into(),
+        }
+    }
 }
 
 /// Convert from standard library errors

crates/kms/src/manager.rs

@@ -159,6 +159,11 @@ impl KmsManager {
     pub async fn health_check(&self) -> Result<bool> {
         self.backend.health_check().await
     }
+
+    /// Report the capabilities of the configured backend
+    pub fn backend_capabilities(&self) -> crate::backends::BackendCapabilities {
+        self.backend.capabilities()
+    }
 }
 
 #[cfg(test)]

crates/kms/src/service.rs

@@ -184,6 +184,15 @@ impl ObjectEncryptionService {
         self.kms_manager.health_check().await
     }
 
+    /// Report the capabilities of the configured backend
+    ///
+    /// # Returns
+    /// The capability matrix advertised by the active KMS backend
+    ///
+    pub fn backend_capabilities(&self) -> crate::backends::BackendCapabilities {
+        self.kms_manager.backend_capabilities()
+    }
+
     /// Create a data encryption key for object encryption
     ///
     /// # Arguments

rustfs/src/admin/handlers/kms_management.rs

@@ -72,6 +72,10 @@ pub struct KmsStatusResponse {
     pub cache_enabled: bool,
     pub cache_stats: Option<CacheStatsResponse>,
     pub default_key_id: Option<String>,
+    /// Capability matrix of the active backend. Additive field: omitted by
+    /// older servers, so it must stay optional for consumers.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub capabilities: Option<rustfs_kms::backends::BackendCapabilities>,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -204,6 +208,7 @@ impl Operation for KmsStatusHandler {
             cache_enabled: config.as_ref().is_some_and(|cfg| cfg.enable_cache),
             cache_stats,
             default_key_id: service.get_default_key_id().cloned(),
+            capabilities: Some(service.backend_capabilities()),
         };
 
         let data = serde_json::to_vec(&response).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?;
@@ -339,4 +344,34 @@ mod tests {
     fn kms_clear_cache_rejects_server_info_fallback() {
         assert_lacks_action(&kms_clear_cache_actions(), Action::AdminAction(AdminAction::ServerInfoAdminAction));
     }
+
+    /// The `capabilities` field is additive: payloads produced by older
+    /// servers (without the field) must keep deserializing, and the field
+    /// must be omitted from JSON when unset so existing consumers see an
+    /// unchanged response shape.
+    #[test]
+    fn kms_status_response_capabilities_field_is_additive() {
+        let legacy_json = serde_json::json!({
+            "backend_type": "local",
+            "backend_status": "healthy",
+            "cache_enabled": true,
+            "cache_stats": null,
+            "default_key_id": null,
+        });
+        let legacy: super::KmsStatusResponse =
+            serde_json::from_value(legacy_json).expect("legacy status payload should deserialize");
+        assert!(legacy.capabilities.is_none());
+
+        let serialized = serde_json::to_value(&legacy).expect("status response should serialize");
+        assert!(serialized.get("capabilities").is_none(), "unset capabilities must be omitted");
+
+        let with_capabilities = super::KmsStatusResponse {
+            capabilities: Some(rustfs_kms::backends::BackendCapabilities::minimal()),
+            ..legacy
+        };
+        let serialized = serde_json::to_value(&with_capabilities).expect("status response should serialize");
+        let capabilities = serialized.get("capabilities").expect("capabilities must be present when set");
+        assert_eq!(capabilities.get("encrypt"), Some(&serde_json::Value::Bool(true)));
+        assert_eq!(capabilities.get("rotate"), Some(&serde_json::Value::Bool(false)));
+    }
 }