feat(kms): accept the AWS backend through KMS configuration (#5592)

* feat(kms): accept the AWS backend through KMS configuration

The AWS KMS backend could be constructed but not selected: the admin
configure API had no AWS variant and startup rejected the backend name.

The configure request pins the region rather than defaulting it, because
that configuration is persisted once and replayed on every node: leaving
the region to each node's ambient provider chain would let nodes address
different regions, and therefore different keys, while reporting an
identical configuration. The request accepts no credential fields, so
credentials stay with the aws-config provider chain on each node, and
`deny_unknown_fields` refuses attempts to submit them anyway.

* test(kms): cover AWS backend selection through the service manager

An end-to-end check that an admin configure request selects the AWS
backend, builds a client, and passes the startup health check. Marked
#[ignore]: it needs real AWS credentials, though it creates no key and
is therefore not billable on its own.

crates/kms/src/api_types.rs

@@ -15,9 +15,10 @@
 //! API types for KMS dynamic configuration
 
 use crate::config::{
-    BackendConfig, CacheConfig, DEFAULT_CACHE_TTL, DEFAULT_MAX_CACHED_KEYS, DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX,
-    DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT, KmsBackend, KmsConfig, LocalConfig, StaticConfig, TlsConfig, VaultAuthMethod,
-    VaultConfig, VaultTransitConfig, allow_immediate_deletion_from_env, redacted_secret, redacted_secret_option,
+    AwsKmsConfig, BackendConfig, CacheConfig, DEFAULT_CACHE_TTL, DEFAULT_MAX_CACHED_KEYS,
+    DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX, DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT, KmsBackend, KmsConfig, LocalConfig,
+    StaticConfig, TlsConfig, VaultAuthMethod, VaultConfig, VaultTransitConfig, allow_immediate_deletion_from_env,
+    redacted_secret, redacted_secret_option,
 };
 use crate::service_manager::KmsServiceStatus;
 use crate::types::{KeyMetadata, KeyUsage};
@@ -165,6 +166,47 @@ pub struct ConfigureStaticKmsRequest {
     pub allow_insecure_dev_defaults: Option<bool>,
 }
 
+/// Request to configure KMS with the AWS KMS backend.
+///
+/// Accepts no credential material by design: every node resolves AWS
+/// credentials through the standard `aws-config` provider chain, so nothing
+/// secret is submitted here, persisted with the cluster configuration, or
+/// echoed back by the status API.
+///
+/// Keys are not created by this path. The backend refuses caller-named key
+/// creation because AWS assigns identifiers itself, so `default_key_id` must
+/// name a key that already exists in AWS, by key id or ARN.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ConfigureAwsKmsRequest {
+    /// AWS region hosting the keys.
+    ///
+    /// Mandatory here, unlike the environment-variable path: this
+    /// configuration is persisted once and replayed on every node, so leaving
+    /// the region to each node's ambient provider chain would let nodes
+    /// silently address different regions — and therefore different keys —
+    /// while reporting the same configuration.
+    pub region: String,
+    /// Endpoint override for local emulators and private endpoints. Unset in
+    /// production, where the SDK derives the regional endpoint; a plaintext
+    /// endpoint stays gated behind the development opt-in.
+    pub endpoint_url: Option<String>,
+    /// Default master key ID for auto-encryption, as an AWS key id or ARN
+    pub default_key_id: Option<String>,
+    /// Operation timeout in seconds
+    pub timeout_seconds: Option<u64>,
+    /// Number of retry attempts
+    pub retry_attempts: Option<u32>,
+    /// Enable caching
+    pub enable_cache: Option<bool>,
+    /// Maximum number of keys to cache
+    pub max_cached_keys: Option<usize>,
+    /// Cache TTL in seconds
+    pub cache_ttl_seconds: Option<u64>,
+    /// Allow development-only insecure defaults
+    pub allow_insecure_dev_defaults: Option<bool>,
+}
+
 impl Drop for ConfigureStaticKmsRequest {
     fn drop(&mut self) {
         use zeroize::Zeroize;
@@ -211,6 +253,9 @@ pub enum ConfigureKmsRequest {
     /// Configure with Static single-key backend
     #[serde(rename = "Static", alias = "static")]
     Static(ConfigureStaticKmsRequest),
+    /// Configure with the AWS KMS backend
+    #[serde(rename = "AWS", alias = "AwsKms", alias = "aws", alias = "aws-kms", alias = "aws_kms")]
+    Aws(ConfigureAwsKmsRequest),
 }
 
 /// KMS configuration response
@@ -636,6 +681,33 @@ impl ConfigureStaticKmsRequest {
     }
 }
 
+impl ConfigureAwsKmsRequest {
+    /// Convert to KmsConfig
+    pub fn to_kms_config(&self) -> KmsConfig {
+        KmsConfig {
+            backend: KmsBackend::Aws,
+            default_key_id: self.default_key_id.clone(),
+            backend_config: BackendConfig::Aws(Box::new(AwsKmsConfig {
+                region: Some(self.region.clone()),
+                endpoint_url: self.endpoint_url.clone(),
+            })),
+            allow_insecure_dev_defaults: self.allow_insecure_dev_defaults.unwrap_or(false),
+            // Read from server configuration, never from the request body: the
+            // gate must mean the same thing whether KMS was configured at
+            // startup or through this endpoint.
+            allow_immediate_deletion: allow_immediate_deletion_from_env(),
+            timeout: Duration::from_secs(self.timeout_seconds.unwrap_or(30)),
+            retry_attempts: self.retry_attempts.unwrap_or(3),
+            enable_cache: self.enable_cache.unwrap_or(true),
+            cache_config: CacheConfig {
+                max_keys: self.max_cached_keys.unwrap_or(DEFAULT_MAX_CACHED_KEYS),
+                ttl: self.cache_ttl_seconds.map_or(DEFAULT_CACHE_TTL, Duration::from_secs),
+                ..CacheConfig::default()
+            },
+        }
+    }
+}
+
 impl ConfigureKmsRequest {
     /// Convert to KmsConfig
     pub fn to_kms_config(&self) -> KmsConfig {
@@ -644,6 +716,7 @@ impl ConfigureKmsRequest {
             ConfigureKmsRequest::VaultKv2(req) => req.to_kms_config(),
             ConfigureKmsRequest::VaultTransit(req) => req.to_kms_config(),
             ConfigureKmsRequest::Static(req) => req.to_kms_config(),
+            ConfigureKmsRequest::Aws(req) => req.to_kms_config(),
         }
     }
 }
@@ -829,6 +902,108 @@ mod tests {
         assert!(request.to_kms_config().validate().is_ok());
     }
 
+    #[test]
+    fn test_deserialize_aws_configure_request_accepts_type_aliases() {
+        for backend_type in ["AWS", "AwsKms", "aws", "aws-kms", "aws_kms"] {
+            let raw = serde_json::json!({
+                "backend_type": backend_type,
+                "region": "eu-central-1",
+                "default_key_id": "arn:aws:kms:eu-central-1:111122223333:key/1234abcd"
+            });
+
+            let request: ConfigureKmsRequest = serde_json::from_value(raw).unwrap_or_else(|e| panic!("{backend_type}: {e}"));
+            let config = request.to_kms_config();
+            assert_eq!(config.backend, KmsBackend::Aws, "backend_type={backend_type}");
+            let aws = config.aws_kms_config().expect("aws backend config");
+            assert_eq!(aws.region.as_deref(), Some("eu-central-1"));
+            assert_eq!(aws.endpoint_url, None);
+            assert!(config.validate().is_ok(), "backend_type={backend_type}");
+        }
+    }
+
+    /// A cluster-persisted AWS configuration must pin its own region: a
+    /// request that leaves it to each node's ambient provider chain is refused
+    /// rather than accepted into a configuration every node interprets
+    /// differently.
+    #[test]
+    fn test_aws_configure_request_requires_an_explicit_region() {
+        let missing = serde_json::json!({
+            "backend_type": "AWS",
+            "default_key_id": "arn:aws:kms:eu-central-1:111122223333:key/1234abcd"
+        });
+        let err = serde_json::from_value::<ConfigureKmsRequest>(missing).expect_err("a region-less AWS request must be refused");
+        assert!(err.to_string().contains("region"), "{err}");
+
+        let empty = serde_json::json!({ "backend_type": "AWS", "region": "" });
+        let request: ConfigureKmsRequest = serde_json::from_value(empty).expect("an empty region deserializes");
+        assert!(request.to_kms_config().validate().is_err(), "an empty region must not validate");
+    }
+
+    #[test]
+    fn test_aws_configure_request_rejects_plaintext_endpoint_without_opt_in() {
+        let raw = serde_json::json!({
+            "backend_type": "AWS",
+            "region": "us-east-1",
+            "endpoint_url": "http://localhost:4566"
+        });
+        let request: ConfigureKmsRequest = serde_json::from_value(raw).expect("aws request should deserialize");
+        assert!(request.to_kms_config().validate().is_err());
+
+        let opt_in = serde_json::json!({
+            "backend_type": "AWS",
+            "region": "us-east-1",
+            "endpoint_url": "http://localhost:4566",
+            "allow_insecure_dev_defaults": true
+        });
+        let request: ConfigureKmsRequest = serde_json::from_value(opt_in).expect("aws request should deserialize");
+        assert!(request.to_kms_config().validate().is_ok());
+    }
+
+    /// The AWS summary carries only non-credential settings, because the
+    /// backend never holds AWS credential material to begin with.
+    #[test]
+    fn test_aws_status_summary_reports_only_non_credential_settings() {
+        let config = ConfigureAwsKmsRequest {
+            region: "us-east-1".to_string(),
+            endpoint_url: None,
+            default_key_id: Some("arn:aws:kms:us-east-1:111122223333:key/1234abcd".to_string()),
+            timeout_seconds: None,
+            retry_attempts: None,
+            enable_cache: None,
+            max_cached_keys: None,
+            cache_ttl_seconds: None,
+            allow_insecure_dev_defaults: None,
+        }
+        .to_kms_config();
+
+        let summary = KmsConfigSummary::from(&config);
+        assert_eq!(summary.backend_type, KmsBackend::Aws);
+        match &summary.backend_summary {
+            BackendSummary::Aws { region, endpoint_url } => {
+                assert_eq!(region.as_deref(), Some("us-east-1"));
+                assert_eq!(endpoint_url.as_deref(), None);
+            }
+            other => panic!("expected aws summary, got {other:?}"),
+        }
+
+        let response = KmsStatusResponse {
+            status: KmsServiceStatus::Running,
+            backend_type: Some(config.backend),
+            healthy: Some(true),
+            config_summary: Some(summary),
+        };
+        let rendered = format!(
+            "{}\n{response:?}",
+            serde_json::to_string(&response).expect("kms status response should serialize")
+        );
+        for credential_field in ["access_key", "secret_key", "session_token", "has_stored_credentials"] {
+            assert!(
+                !rendered.contains(credential_field),
+                "aws status output must not describe credential material: {rendered}"
+            );
+        }
+    }
+
     #[test]
     fn test_configure_request_rejects_unknown_fields() {
         let raw = serde_json::json!({
@@ -853,6 +1028,17 @@ mod tests {
 
         let err = serde_json::from_value::<ConfigureKmsRequest>(raw).expect_err("unknown auth field should fail");
         assert!(err.to_string().contains("unknown field"));
+
+        // AWS credentials belong to the provider chain: a request that tries to
+        // smuggle them in must be refused, not silently ignored.
+        let raw = serde_json::json!({
+            "backend_type": "AWS",
+            "region": "us-east-1",
+            "secret_access_key": "AKIA-not-accepted-here"
+        });
+
+        let err = serde_json::from_value::<ConfigureKmsRequest>(raw).expect_err("unknown aws field should fail");
+        assert!(err.to_string().contains("unknown field"));
     }
 
     #[test]

crates/kms/src/lib.rs

@@ -83,9 +83,9 @@ pub mod types;
 
 // Re-export public API
 pub use api_types::{
-    CacheSummary, ConfigureKmsRequest, ConfigureKmsResponse, ConfigureLocalKmsRequest, ConfigureStaticKmsRequest,
-    ConfigureVaultKmsRequest, ConfigureVaultTransitKmsRequest, KmsConfigSummary, KmsStatusResponse, StartKmsRequest,
-    StartKmsResponse, StopKmsResponse, TagKeyRequest, TagKeyResponse, UntagKeyRequest, UntagKeyResponse,
+    CacheSummary, ConfigureAwsKmsRequest, ConfigureKmsRequest, ConfigureKmsResponse, ConfigureLocalKmsRequest,
+    ConfigureStaticKmsRequest, ConfigureVaultKmsRequest, ConfigureVaultTransitKmsRequest, KmsConfigSummary, KmsStatusResponse,
+    StartKmsRequest, StartKmsResponse, StopKmsResponse, TagKeyRequest, TagKeyResponse, UntagKeyRequest, UntagKeyResponse,
     UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse,
 };
 pub use audit::{KmsAuditOperation, KmsAuditOutcome, KmsAuditRecord, KmsAuditSink, redact_encryption_context};

crates/kms/src/service_manager.rs

@@ -749,6 +749,42 @@ mod tests {
         KmsConfig::static_kms(key_id.to_string(), BASE64_STANDARD.encode([fill; 32]))
     }
 
+    /// End-to-end wiring check for the AWS backend: an admin configure request
+    /// must select it, build a real client, and pass the startup health check.
+    ///
+    /// `RUSTFS_KMS_AWS_REGION` names the region; the credential chain supplies
+    /// the rest. No key is created, so this test is not billable on its own.
+    #[tokio::test]
+    #[ignore] // Requires real AWS credentials
+    async fn aws_backend_configure_and_start_end_to_end() {
+        let region = std::env::var("RUSTFS_KMS_AWS_REGION").expect("RUSTFS_KMS_AWS_REGION must name the test region");
+        let config = crate::api_types::ConfigureAwsKmsRequest {
+            region,
+            endpoint_url: None,
+            default_key_id: std::env::var("RUSTFS_KMS_DEFAULT_KEY_ID").ok(),
+            timeout_seconds: None,
+            retry_attempts: None,
+            enable_cache: None,
+            max_cached_keys: None,
+            cache_ttl_seconds: None,
+            allow_insecure_dev_defaults: None,
+        }
+        .to_kms_config();
+
+        let manager = KmsServiceManager::new();
+        manager.configure(config).await.expect("configure the AWS backend");
+        manager.start().await.expect("start the AWS backend");
+
+        assert_eq!(manager.get_status().await, KmsServiceStatus::Running);
+        let capabilities = manager
+            .get_manager()
+            .await
+            .expect("a running service exposes its manager")
+            .backend_capabilities();
+        assert!(capabilities.schedule_deletion);
+        assert!(!capabilities.physical_delete);
+    }
+
     #[tokio::test]
     async fn configure_rejects_insecure_development_defaults_before_state_update() {
         let manager = KmsServiceManager::new();