feat(admin): expose KMS backend capabilities in status response

Surface the backend capability matrix as an optional, additive
'capabilities' field on the /v3/kms/status response so management
clients can discover supported lifecycle operations. The field is
skipped when unset, keeping the response shape unchanged for existing
consumers, and legacy payloads without it still deserialize.

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

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