Refactor: refactor SSE layer and KMS subsystem (#1703)

Co-authored-by: houseme <housemecn@gmail.com>

.vscode/launch.json

@@ -170,6 +170,81 @@
             "sourceLanguages": [
                 "rust"
             ],
-        }
+        },
+        {
+            "name": "Debug executable target/debug/rustfs with sse",
+            "type":  "lldb",
+            "request": "launch",
+            "program": "${workspaceFolder}/target/debug/rustfs",
+            "args": [],
+            "cwd": "${workspaceFolder}",
+            //"stopAtEntry": false,
+            //"preLaunchTask": "cargo build",
+            "env": {
+                "RUSTFS_ACCESS_KEY": "rustfsadmin",
+                "RUSTFS_SECRET_KEY": "rustfsadmin",
+                "RUSTFS_VOLUMES": "./target/volumes/test{1...4}",
+                "RUSTFS_ADDRESS": ":9000",
+                "RUSTFS_CONSOLE_ENABLE": "true",
+                "RUSTFS_CONSOLE_ADDRESS": "127.0.0.1:9001",
+                "RUSTFS_OBS_LOG_DIRECTORY": "./target/logs",
+                // "RUSTFS_OBS_TRACE_ENDPOINT": "http://127.0.0.1:4318/v1/traces", // jeager otlp http endpoint
+                // "RUSTFS_OBS_METRIC_ENDPOINT": "http://127.0.0.1:4318/v1/metrics", // default otlp http endpoint
+                // "RUSTFS_OBS_LOG_ENDPOINT": "http://127.0.0.1:4318/v1/logs", // default otlp http endpoint
+                // "RUSTFS_COMPRESS_ENABLE": "true",
+
+                // 1. simple sse test key (no kms system)
+                // "__RUSTFS_SSE_SIMPLE_CMK": "2dfNXGHlsEflGVCxb+5DIdGEl1sIvtwX+QfmYasi5QM=",
+
+                // 2. kms local backend test key
+                "RUSTFS_KMS_ENABLE": "true",
+                "RUSTFS_KMS_BACKEND": "local",
+                "RUSTFS_KMS_KEY_DIR": "./target/kms-key-dir",
+                "RUSTFS_KMS_LOCAL_MASTER_KEY": "my-secret-key", // Some Password
+                "RUSTFS_KMS_DEFAULT_KEY_ID": "rustfs-master-key",
+
+                // 3. kms vault backend test key
+                // "RUSTFS_KMS_ENABLE": "true",
+                // "RUSTFS_KMS_BACKEND": "vault",
+                // "RUSTFS_KMS_VAULT_ADDRESS": "http://127.0.0.1:8200",
+                // "RUSTFS_KMS_VAULT_TOKEN": "Dev Token",
+                // "RUSTFS_KMS_DEFAULT_KEY_ID": "rustfs-master-key",
+
+            },
+            "sourceLanguages": [
+                "rust"
+            ],
+        },
+        {
+            "name": "E2E test executable target/debug/rustfs",
+            "type":  "lldb",
+            "request": "launch",
+            "program": "${workspaceFolder}/target/debug/rustfs",
+            "args": [],
+            "cwd": "${workspaceFolder}",
+            //"stopAtEntry": false,
+            //"preLaunchTask": "cargo build",
+            "env": {
+                "RUST_LOG": "rustfs=debug,ecstore=info,s3s=debug,iam=debug",
+                "RUST_BACKTRACE": "full",
+                "RUSTFS_ACCESS_KEY": "rustfsadmin",
+                "RUSTFS_SECRET_KEY": "rustfsadmin",
+                "RUSTFS_VOLUMES": "./target/e2e-test/test{1...4}",
+                "RUSTFS_REGION": "us-east-1",
+                "RUSTFS_ADDRESS": ":9000",
+                "RUSTFS_CONSOLE_ENABLE": "true",
+                "RUSTFS_CONSOLE_ADDRESS": "127.0.0.1:9001",
+                "RUSTFS_OBS_LOG_DIRECTORY": "./target/logs",
+
+                "RUSTFS_KMS_ENABLE": "true",
+                "RUSTFS_KMS_BACKEND": "local",
+                "RUSTFS_KMS_KEY_DIR": "./target/e2e-key-dir",
+                "RUSTFS_KMS_LOCAL_MASTER_KEY": "my-secret-key", // Some Password
+                "RUSTFS_KMS_DEFAULT_KEY_ID": "rustfs-master-key",
+            },
+            "sourceLanguages": [
+                "rust"
+            ],
+        },
     ]
 }

Cargo.lock

@@ -7628,6 +7628,7 @@ dependencies = [
 name = "rustfs"
 version = "0.0.5"
 dependencies = [
+ "aes-gcm 0.11.0-rc.2",
  "astral-tokio-tar",
  "async-trait",
  "atoi",
@@ -7990,6 +7991,7 @@ name = "rustfs-kms"
 version = "0.0.5"
 dependencies = [
  "aes-gcm 0.11.0-rc.2",
+ "arc-swap",
  "async-trait",
  "base64",
  "chacha20poly1305",

crates/ecstore/src/store_api.rs

@@ -715,7 +715,16 @@ impl ObjectInfo {
             return Ok(actual_size);
         }
 
-        // TODO: IsEncrypted
+        // Check if object is encrypted
+        // Encrypted objects store original size in x-rustfs-encryption-original-size metadata
+        if let Some(size_str) = self.user_defined.get("x-rustfs-encryption-original-size")
+            && !size_str.is_empty()
+        {
+            let size = size_str
+                .parse::<i64>()
+                .map_err(|e| std::io::Error::other(format!("Failed to parse encryption original size: {e}")))?;
+            return Ok(size);
+        }
 
         Ok(self.size)
     }

crates/kms/.gitignore

@@ -0,0 +1 @@
+examples/local_data/*

crates/kms/Cargo.toml

@@ -55,6 +55,7 @@ moka = { workspace = true, features = ["future"] }
 
 # Additional dependencies
 md5 = { workspace = true }
+arc-swap = { workspace = true }
 
 # HTTP client for Vault
 reqwest = { workspace = true }

crates/kms/examples/kms_local_demo.rs

@@ -0,0 +1,251 @@
+// 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.
+
+//! KMS Demo - Comprehensive example demonstrating RustFS KMS capabilities
+//!
+//! This example demonstrates:
+//! - Initializing and configuring KMS service
+//! - Creating master keys
+//! - Generating data encryption keys
+//! - Encrypting and decrypting data using high-level APIs
+//! - Key management operations
+//! - Cache statistics
+//!
+//! Run with: `cargo run --example demo1`
+
+use rustfs_kms::{
+    CreateKeyRequest, DescribeKeyRequest, EncryptionAlgorithm, GenerateDataKeyRequest, KeySpec, KeyUsage, KmsConfig,
+    ListKeysRequest, init_global_kms_service_manager,
+};
+use std::collections::HashMap;
+use std::fs;
+use std::io::Cursor;
+use tokio::io::AsyncReadExt;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Note: Tracing is optional - if tracing-subscriber is not available,
+    // the example will still work but with less detailed logging
+
+    println!("=== RustFS KMS Demo ===\n");
+
+    // Step 1: Initialize global KMS service manager
+    println!("1. Initializing KMS service manager...");
+    let service_manager = init_global_kms_service_manager();
+    println!("   ✓ Service manager initialized\n");
+
+    // Step 2: Create a temporary directory for local backend
+    println!("2. Setting up local backend...");
+    if fs::metadata("examples/local_data").is_err() {
+        fs::create_dir_all("examples/local_data")?;
+    }
+    let data_dir = std::path::PathBuf::from("examples/local_data");
+    println!("   ✓ Using data directory: {}\n", data_dir.display());
+
+    // Step 3: Configure KMS with local backend
+    println!("3. Configuring KMS with local backend...");
+    let config = KmsConfig::local(data_dir)
+        .with_default_key("demo-key-default-1".to_string())
+        .with_cache(true);
+    service_manager.configure(config).await?;
+    println!("   ✓ KMS configured\n");
+
+    // Step 4: Start the KMS service
+    println!("4. Starting KMS service...");
+    service_manager.start().await?;
+    println!("   ✓ KMS service started\n");
+
+    // Step 5: Get the encryption service
+    println!("5. Getting encryption service...");
+    let encryption_service = rustfs_kms::get_global_encryption_service()
+        .await
+        .ok_or("Encryption service not available")?;
+    println!("   ✓ Encryption service obtained\n");
+
+    // Step 6: Create a master key
+    println!("6. Creating a master key...");
+    let create_request = CreateKeyRequest {
+        key_name: Some("demo-key-master-1".to_string()),
+        key_usage: KeyUsage::EncryptDecrypt,
+        description: Some("Demo master key for encryption".to_string()),
+        policy: None,
+        tags: {
+            let mut tags = HashMap::new();
+            tags.insert("environment".to_string(), "demo".to_string());
+            tags.insert("purpose".to_string(), "testing".to_string());
+            tags
+        },
+        origin: Some("demo1.rs".to_string()),
+    };
+
+    let create_response = encryption_service.create_key(create_request).await?;
+    println!("   ✓ Master key created:");
+    println!("     - Key ID: {}", create_response.key_id);
+    println!("     - Key State: {:?}", create_response.key_metadata.key_state);
+    println!("     - Key Usage: {:?}", create_response.key_metadata.key_usage);
+    println!("     - Created: {}\n", create_response.key_metadata.creation_date);
+
+    let master_key_id = create_response.key_id.clone();
+
+    // Step 7: Describe the key
+    println!("7. Describing the master key...");
+    let describe_request = DescribeKeyRequest {
+        key_id: master_key_id.clone(),
+    };
+    let describe_response = encryption_service.describe_key(describe_request).await?;
+    let metadata = describe_response.key_metadata;
+    println!("   ✓ Key details:");
+    println!("     - Key ID: {}", metadata.key_id);
+    println!("     - Description: {:?}", metadata.description);
+    println!("     - Key Usage: {:?}", metadata.key_usage);
+    println!("     - Key State: {:?}", metadata.key_state);
+    println!("     - Tags: {:?}\n", metadata.tags);
+
+    // Step 8: Generate a data encryption key (OPTIONAL - for demonstration only)
+    // NOTE: This step is OPTIONAL and only for educational purposes!
+    // In real usage, you can skip this step and go directly to Step 9.
+    // encrypt_object() will automatically generate a data key internally.
+    println!("8. [OPTIONAL] Generating a data encryption key (for demonstration)...");
+    println!("   ⚠️  This step is OPTIONAL - only for understanding the two-layer key architecture:");
+    println!("   - Master Key (CMK): Used to encrypt/decrypt data keys");
+    println!("   - Data Key (DEK): Used to encrypt/decrypt actual data");
+    println!("   In production, you can skip this and use encrypt_object() directly!\n");
+
+    let data_key_request = GenerateDataKeyRequest {
+        key_id: master_key_id.clone(),
+        key_spec: KeySpec::Aes256,
+        encryption_context: {
+            let mut context = HashMap::new();
+            context.insert("bucket".to_string(), "demo-bucket".to_string());
+            context.insert("object_key".to_string(), "demo-object.txt".to_string());
+            context
+        },
+    };
+
+    let data_key_response = encryption_service.generate_data_key(data_key_request).await?;
+    println!("   ✓ Data key generated (for demonstration):");
+    println!("     - Master Key ID: {}", data_key_response.key_id);
+    println!("     - Data Key (plaintext) length: {} bytes", data_key_response.plaintext_key.len());
+    println!(
+        "     - Encrypted Data Key (ciphertext blob) length: {} bytes",
+        data_key_response.ciphertext_blob.len()
+    );
+    println!("     - Note: This data key is NOT used in Step 9 - encrypt_object() generates its own!\n");
+
+    // Step 9: Encrypt some data using high-level API
+    // This is the RECOMMENDED way to encrypt data - everything is handled automatically!
+    println!("9. Encrypting data using object encryption service (RECOMMENDED)...");
+    println!("   ✅ This is all you need! encrypt_object() handles everything:");
+    println!("   1. Validates/creates the master key (if needed)");
+    println!("   2. Generates a NEW data key using the master key (independent of Step 8)");
+    println!("   3. Uses the data key to encrypt the actual data");
+    println!("   4. Stores the encrypted data key (ciphertext blob) in metadata");
+    println!("   You only need to provide the master_key_id - everything else is handled!\n");
+
+    let plaintext = b"Hello, RustFS KMS! This is a test message for encryption.";
+    println!("   Plaintext: {}", String::from_utf8_lossy(plaintext));
+
+    let reader = Cursor::new(plaintext);
+    // Just provide the master_key_id - encrypt_object() handles everything internally!
+    let encryption_result = encryption_service
+        .encrypt_object(
+            "demo-bucket",
+            "demo-object.txt",
+            reader,
+            &EncryptionAlgorithm::Aes256,
+            Some(&master_key_id), // Only need to provide master key ID
+            None,
+        )
+        .await?;
+
+    println!("   ✓ Data encrypted:");
+    println!("     - Encrypted data length: {} bytes", encryption_result.ciphertext.len());
+    println!("     - Algorithm: {}", encryption_result.metadata.algorithm);
+    println!(
+        "     - Master Key ID: {} (used to encrypt the data key)",
+        encryption_result.metadata.key_id
+    );
+    println!(
+        "     - Encrypted Data Key length: {} bytes (stored in metadata)",
+        encryption_result.metadata.encrypted_data_key.len()
+    );
+    println!("     - Original size: {} bytes\n", encryption_result.metadata.original_size);
+
+    // Step 10: Decrypt the data using high-level API
+    println!("10. Decrypting data...");
+    println!("   Note: decrypt_object() has the ENTIRE decryption flow built-in:");
+    println!("   1. Extracts the encrypted data key from metadata");
+    println!("   2. Uses master key to decrypt the data key");
+    println!("   3. Uses the decrypted data key to decrypt the actual data");
+    println!("   You only need to provide the encrypted data and metadata!\n");
+
+    let mut decrypted_reader = encryption_service
+        .decrypt_object(
+            "demo-bucket",
+            "demo-object.txt",
+            encryption_result.ciphertext.clone(),
+            &encryption_result.metadata, // Contains everything needed for decryption
+            None,
+        )
+        .await?;
+
+    let mut decrypted_data = Vec::new();
+    decrypted_reader.read_to_end(&mut decrypted_data).await?;
+
+    println!("   ✓ Data decrypted:");
+    println!("     - Decrypted text: {}\n", String::from_utf8_lossy(&decrypted_data));
+
+    // Verify decryption
+    assert_eq!(plaintext, decrypted_data.as_slice());
+    println!("   ✓ Decryption verified: plaintext matches original\n");
+
+    // Step 11: List all keys
+    println!("11. Listing all keys...");
+    let list_request = ListKeysRequest {
+        limit: Some(10),
+        marker: None,
+        usage_filter: None,
+        status_filter: None,
+    };
+    let list_response = encryption_service.list_keys(list_request).await?;
+    println!("   ✓ Keys found: {}", list_response.keys.len());
+    for (idx, key_info) in list_response.keys.iter().enumerate() {
+        println!("     {}. {} ({:?})", idx + 1, key_info.key_id, key_info.status);
+    }
+    println!();
+
+    // Step 12: Check cache statistics
+    println!("12. Checking cache statistics...");
+    if let Some((hits, misses)) = encryption_service.cache_stats().await {
+        println!("   ✓ Cache statistics:");
+        println!("     - Cache hits: {}", hits);
+        println!("     - Cache misses: {}\n", misses);
+    } else {
+        println!("   - Cache is disabled\n");
+    }
+
+    // Step 13: Health check
+    println!("13. Performing health check...");
+    let is_healthy = encryption_service.health_check().await?;
+    println!("   ✓ KMS backend is healthy: {}\n", is_healthy);
+
+    // Step 14: Stop the service
+    println!("14. Stopping KMS service...");
+    service_manager.stop().await?;
+    println!("   ✓ KMS service stopped\n");
+
+    println!("=== Demo completed successfully! ===");
+
+    Ok(())
+}

crates/kms/examples/kms_vault_kv_demo.rs

@@ -0,0 +1,292 @@
+// 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.
+
+//! KMS Demo 2 - Comprehensive example demonstrating RustFS KMS with Vault backend
+//!
+//! This example demonstrates:
+//! - Initializing and configuring KMS service with Vault backend
+//! - Creating master keys stored in Vault
+//! - Generating data encryption keys
+//! - Encrypting and decrypting data using high-level APIs
+//! - Key management operations with Vault
+//! - Cache statistics
+//!
+//! Prerequisites:
+//! - Vault server running at http://127.0.0.1:8200 (or set RUSTFS_KMS_VAULT_ADDRESS)
+//! - Vault token (set RUSTFS_KMS_VAULT_TOKEN environment variable, or use default "dev-token" for dev mode)
+//!
+//! Run with: `cargo run --example demo2`
+//! Or with custom Vault settings:
+//!   RUSTFS_KMS_VAULT_ADDRESS=http://127.0.0.1:8200 RUSTFS_KMS_VAULT_TOKEN=your-token cargo run --example demo2
+
+use rustfs_kms::{
+    CreateKeyRequest, DescribeKeyRequest, EncryptionAlgorithm, GenerateDataKeyRequest, KeySpec, KeyUsage, KmsConfig, KmsError,
+    ListKeysRequest, init_global_kms_service_manager,
+};
+use std::collections::HashMap;
+use std::io::Cursor;
+use tokio::io::AsyncReadExt;
+use url::Url;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Note: Tracing is optional - if tracing-subscriber is not available,
+    // the example will still work but with less detailed logging
+
+    println!("=== RustFS KMS Demo 2 (Vault Backend) ===\n");
+
+    // Step 1: Initialize global KMS service manager
+    println!("1. Initializing KMS service manager...");
+    let service_manager = init_global_kms_service_manager();
+    println!("   ✓ Service manager initialized\n");
+
+    // Step 2: Get Vault configuration from environment or use defaults
+    println!("2. Configuring Vault backend...");
+    let vault_address = std::env::var("RUSTFS_KMS_VAULT_ADDRESS").unwrap_or_else(|_| "http://127.0.0.1:8200".to_string());
+    let vault_token = std::env::var("RUSTFS_KMS_VAULT_TOKEN").unwrap_or_else(|_| {
+        println!("   ⚠️  No RUSTFS_KMS_VAULT_TOKEN found, using default 'dev-token'");
+        println!("      For production, set RUSTFS_KMS_VAULT_TOKEN environment variable");
+        "dev-token".to_string()
+    });
+
+    let vault_url = Url::parse(&vault_address).map_err(|e| format!("Invalid Vault address '{}': {}", vault_address, e))?;
+
+    println!("   ✓ Vault address: {}", vault_address);
+    println!("   ✓ Using token authentication\n");
+
+    // Step 3: Configure KMS with Vault backend
+    println!("3. Configuring KMS with Vault backend...");
+    let config = KmsConfig::vault(vault_url, vault_token)
+        .with_default_key("demo-key-master-1".to_string())
+        .with_cache(true);
+    service_manager.configure(config).await?;
+    println!("   ✓ KMS configured with Vault backend\n");
+
+    // Step 4: Start the KMS service
+    println!("4. Starting KMS service...");
+    service_manager.start().await?;
+    println!("   ✓ KMS service started\n");
+
+    // Step 5: Get the encryption service
+    println!("5. Getting encryption service...");
+    let encryption_service = rustfs_kms::get_global_encryption_service()
+        .await
+        .ok_or("Encryption service not available")?;
+    println!("   ✓ Encryption service obtained\n");
+
+    // Step 6: Create a master key (stored in Vault) or use existing one
+    println!("6. Checking for existing master key in Vault...");
+    let master_key_id = "demo-key-master-1".to_string();
+    let describe_request = DescribeKeyRequest {
+        key_id: master_key_id.clone(),
+    };
+
+    let master_key_id = match encryption_service.describe_key(describe_request).await {
+        Ok(describe_response) => {
+            // Key already exists, use it
+            println!("   ✓ Master key already exists in Vault:");
+            println!("     - Key ID: {}", describe_response.key_metadata.key_id);
+            println!("     - Key State: {:?}", describe_response.key_metadata.key_state);
+            println!("     - Key Usage: {:?}", describe_response.key_metadata.key_usage);
+            println!("     - Created: {}\n", describe_response.key_metadata.creation_date);
+            describe_response.key_metadata.key_id
+        }
+        Err(KmsError::KeyNotFound { .. }) => {
+            // Key doesn't exist, create it
+            println!("   Key not found, creating new master key in Vault...");
+            let create_request = CreateKeyRequest {
+                key_name: Some(master_key_id.clone()),
+                key_usage: KeyUsage::EncryptDecrypt,
+                description: Some("Demo master key for encryption (stored in Vault)".to_string()),
+                policy: None,
+                tags: {
+                    let mut tags = HashMap::new();
+                    tags.insert("environment".to_string(), "demo".to_string());
+                    tags.insert("purpose".to_string(), "testing".to_string());
+                    tags.insert("backend".to_string(), "vault".to_string());
+                    tags
+                },
+                origin: Some("demo2.rs".to_string()),
+            };
+
+            let create_response = encryption_service.create_key(create_request).await?;
+            println!("   ✓ Master key created in Vault:");
+            println!("     - Key ID: {}", create_response.key_id);
+            println!("     - Key State: {:?}", create_response.key_metadata.key_state);
+            println!("     - Key Usage: {:?}", create_response.key_metadata.key_usage);
+            println!("     - Created: {}\n", create_response.key_metadata.creation_date);
+            create_response.key_id
+        }
+        Err(e) => {
+            // Other error, return it
+            return Err(Box::new(e) as Box<dyn std::error::Error>);
+        }
+    };
+
+    // Step 7: Describe the key (retrieved from Vault)
+    println!("7. Describing the master key (from Vault)...");
+    let describe_request = DescribeKeyRequest {
+        key_id: master_key_id.clone(),
+    };
+    let describe_response = encryption_service.describe_key(describe_request).await?;
+    let metadata = describe_response.key_metadata;
+    println!("   ✓ Key details (from Vault):");
+    println!("     - Key ID: {}", metadata.key_id);
+    println!("     - Description: {:?}", metadata.description);
+    println!("     - Key Usage: {:?}", metadata.key_usage);
+    println!("     - Key State: {:?}", metadata.key_state);
+    println!("     - Tags: {:?}\n", metadata.tags);
+
+    // Step 8: Generate a data encryption key (OPTIONAL - for demonstration only)
+    // NOTE: This step is OPTIONAL and only for educational purposes!
+    // In real usage, you can skip this step and go directly to Step 9.
+    // encrypt_object() will automatically generate a data key internally.
+    println!("8. [OPTIONAL] Generating a data encryption key (for demonstration)...");
+    println!("   ⚠️  This step is OPTIONAL - only for understanding the two-layer key architecture:");
+    println!("   - Master Key (CMK): Stored in Vault, used to encrypt/decrypt data keys");
+    println!("   - Data Key (DEK): Generated per object, encrypted by master key");
+    println!("   In production, you can skip this and use encrypt_object() directly!\n");
+
+    let data_key_request = GenerateDataKeyRequest {
+        key_id: master_key_id.clone(),
+        key_spec: KeySpec::Aes256,
+        encryption_context: {
+            let mut context = HashMap::new();
+            context.insert("bucket".to_string(), "demo-bucket".to_string());
+            context.insert("object_key".to_string(), "demo-object.txt".to_string());
+            context
+        },
+    };
+
+    let data_key_response = encryption_service.generate_data_key(data_key_request).await?;
+    println!("   ✓ Data key generated (for demonstration):");
+    println!("     - Master Key ID: {}", data_key_response.key_id);
+    println!("     - Data Key (plaintext) length: {} bytes", data_key_response.plaintext_key.len());
+    println!(
+        "     - Encrypted Data Key (ciphertext blob) length: {} bytes",
+        data_key_response.ciphertext_blob.len()
+    );
+    println!("     - Note: This data key is NOT used in Step 9 - encrypt_object() generates its own!\n");
+
+    // Step 9: Encrypt some data using high-level API
+    // This is the RECOMMENDED way to encrypt data - everything is handled automatically!
+    println!("9. Encrypting data using object encryption service (RECOMMENDED)...");
+    println!("   ✅ This is all you need! encrypt_object() handles everything:");
+    println!("   1. Validates/creates the master key in Vault (if needed)");
+    println!("   2. Generates a NEW data key using the master key from Vault (independent of Step 8)");
+    println!("   3. Uses the data key to encrypt the actual data");
+    println!("   4. Stores the encrypted data key (ciphertext blob) in metadata");
+    println!("   You only need to provide the master_key_id - everything else is handled!\n");
+
+    let plaintext = b"Hello, RustFS KMS with Vault! This is a test message for encryption.";
+    println!("   Plaintext: {}", String::from_utf8_lossy(plaintext));
+
+    let reader = Cursor::new(plaintext);
+    // Just provide the master_key_id - encrypt_object() handles everything internally!
+    let encryption_result = encryption_service
+        .encrypt_object(
+            "demo-bucket",
+            "demo-object.txt",
+            reader,
+            &EncryptionAlgorithm::Aes256,
+            Some(&master_key_id), // Only need to provide master key ID
+            None,
+        )
+        .await?;
+
+    println!("   ✓ Data encrypted:");
+    println!("     - Encrypted data length: {} bytes", encryption_result.ciphertext.len());
+    println!("     - Algorithm: {}", encryption_result.metadata.algorithm);
+    println!(
+        "     - Master Key ID: {} (stored in Vault, used to encrypt the data key)",
+        encryption_result.metadata.key_id
+    );
+    println!(
+        "     - Encrypted Data Key length: {} bytes (stored in metadata)",
+        encryption_result.metadata.encrypted_data_key.len()
+    );
+    println!("     - Original size: {} bytes\n", encryption_result.metadata.original_size);
+
+    // Step 10: Decrypt the data using high-level API
+    println!("10. Decrypting data...");
+    println!("   Note: decrypt_object() has the ENTIRE decryption flow built-in:");
+    println!("   1. Extracts the encrypted data key from metadata");
+    println!("   2. Uses master key from Vault to decrypt the data key");
+    println!("   3. Uses the decrypted data key to decrypt the actual data");
+    println!("   You only need to provide the encrypted data and metadata!\n");
+
+    let mut decrypted_reader = encryption_service
+        .decrypt_object(
+            "demo-bucket",
+            "demo-object.txt",
+            encryption_result.ciphertext.clone(),
+            &encryption_result.metadata, // Contains everything needed for decryption
+            None,
+        )
+        .await?;
+
+    let mut decrypted_data = Vec::new();
+    decrypted_reader.read_to_end(&mut decrypted_data).await?;
+
+    println!("   ✓ Data decrypted:");
+    println!("     - Decrypted text: {}\n", String::from_utf8_lossy(&decrypted_data));
+
+    // Verify decryption
+    assert_eq!(plaintext, decrypted_data.as_slice());
+    println!("   ✓ Decryption verified: plaintext matches original\n");
+
+    // Step 11: List all keys (from Vault)
+    println!("11. Listing all keys (from Vault)...");
+    let list_request = ListKeysRequest {
+        limit: Some(10),
+        marker: None,
+        usage_filter: None,
+        status_filter: None,
+    };
+    let list_response = encryption_service.list_keys(list_request).await?;
+    println!("   ✓ Keys found in Vault: {}", list_response.keys.len());
+    for (idx, key_info) in list_response.keys.iter().enumerate() {
+        println!("     {}. {} ({:?})", idx + 1, key_info.key_id, key_info.status);
+    }
+    println!();
+
+    // Step 12: Check cache statistics
+    println!("12. Checking cache statistics...");
+    if let Some((hits, misses)) = encryption_service.cache_stats().await {
+        println!("   ✓ Cache statistics:");
+        println!("     - Cache hits: {}", hits);
+        println!("     - Cache misses: {}\n", misses);
+    } else {
+        println!("   - Cache is disabled\n");
+    }
+
+    // Step 13: Health check (verifies Vault connectivity)
+    println!("13. Performing health check (Vault connectivity)...");
+    let is_healthy = encryption_service.health_check().await?;
+    println!("   ✓ KMS backend (Vault) is healthy: {}\n", is_healthy);
+
+    // Step 14: Stop the service
+    println!("14. Stopping KMS service...");
+    service_manager.stop().await?;
+    println!("   ✓ KMS service stopped\n");
+
+    println!("=== Demo 2 (Vault Backend) completed successfully! ===");
+    println!("\n💡 Tips:");
+    println!("   - Keys are now stored in Vault at: {}/v1/secret/data/rustfs/kms/keys/", vault_address);
+    println!("   - You can verify keys in Vault using: vault kv list secret/rustfs/kms/keys/");
+    println!("   - For production, use proper Vault authentication (AppRole, etc.)");
+    println!("   - See examples/VAULT_SETUP.md for detailed Vault configuration guide");
+
+    Ok(())
+}

crates/kms/src/api_types.rs

@@ -271,7 +271,7 @@ impl ConfigureVaultKmsRequest {
         KmsConfig {
             backend: KmsBackend::Vault,
             default_key_id: self.default_key_id.clone(),
-            backend_config: BackendConfig::Vault(VaultConfig {
+            backend_config: BackendConfig::Vault(Box::new(VaultConfig {
                 address: self.address.clone(),
                 auth_method: self.auth_method.clone(),
                 namespace: self.namespace.clone(),
@@ -288,7 +288,7 @@ impl ConfigureVaultKmsRequest {
                 } else {
                     None
                 },
-            }),
+            })),
             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),

crates/kms/src/backends/local.rs

@@ -17,6 +17,7 @@
 use crate::backends::{BackendInfo, KmsBackend, KmsClient};
 use crate::config::KmsConfig;
 use crate::config::LocalConfig;
+use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
 use crate::error::{KmsError, Result};
 use crate::types::*;
 use aes_gcm::{
@@ -24,11 +25,13 @@ use aes_gcm::{
     aead::{Aead, KeyInit},
 };
 use async_trait::async_trait;
+use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
 use jiff::Zoned;
 use rand::Rng;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::path::PathBuf;
+use std::time::Duration;
 use tokio::fs;
 use tokio::sync::RwLock;
 use tracing::{debug, info, warn};
@@ -37,9 +40,11 @@ use tracing::{debug, info, warn};
 pub struct LocalKmsClient {
     config: LocalConfig,
     /// In-memory cache of loaded keys for performance
-    key_cache: RwLock<HashMap<String, MasterKey>>,
+    key_cache: RwLock<HashMap<String, MasterKeyInfo>>,
     /// Master encryption key for encrypting stored keys
     master_cipher: Option<Aes256Gcm>,
+    /// DEK encryption implementation
+    dek_crypto: AesDekCrypto,
 }
 
 /// Serializable representation of a master key stored on disk
@@ -55,24 +60,12 @@ struct StoredMasterKey {
     created_at: Zoned,
     rotated_at: Option<Zoned>,
     created_by: Option<String>,
-    /// Encrypted key material (32 bytes for AES-256)
+    /// Encrypted key material (32 bytes encoded in base64 for AES-256)
-    encrypted_key_material: Vec<u8>,
+    encrypted_key_material: String,
     /// Nonce used for encryption
     nonce: Vec<u8>,
 }
 
-/// Data key envelope stored with each data key generation
-#[derive(Debug, Clone, Serialize, Deserialize)]
-struct DataKeyEnvelope {
-    key_id: String,
-    master_key_id: String,
-    key_spec: String,
-    encrypted_key: Vec<u8>,
-    nonce: Vec<u8>,
-    encryption_context: HashMap<String, String>,
-    created_at: Zoned,
-}
-
 impl LocalKmsClient {
     /// Create a new local KMS client
     pub async fn new(config: LocalConfig) -> Result<Self> {
@@ -95,6 +88,7 @@ impl LocalKmsClient {
             config,
             key_cache: RwLock::new(HashMap::new()),
             master_cipher,
+            dek_crypto: AesDekCrypto::new(),
         })
     }
 
@@ -116,8 +110,8 @@ impl LocalKmsClient {
         self.config.key_dir.join(format!("{key_id}.key"))
     }
 
-    /// Load a master key from disk
+    /// Decode and decrypt a stored key file, returning both the metadata and decrypted key material
-    async fn load_master_key(&self, key_id: &str) -> Result<MasterKey> {
+    async fn decode_stored_key(&self, key_id: &str) -> Result<(StoredMasterKey, Vec<u8>)> {
         let key_path = self.master_key_path(key_id);
         if !key_path.exists() {
             return Err(KmsError::key_not_found(key_id));
@@ -127,7 +121,7 @@ impl LocalKmsClient {
         let stored_key: StoredMasterKey = serde_json::from_slice(&content)?;
 
         // Decrypt key material if master cipher is available
-        let _key_material = if let Some(ref cipher) = self.master_cipher {
+        let key_material = if let Some(ref cipher) = self.master_cipher {
             if stored_key.nonce.len() != 12 {
                 return Err(KmsError::cryptographic_error("nonce", "Invalid nonce length"));
             }
@@ -136,14 +130,29 @@ impl LocalKmsClient {
             nonce_array.copy_from_slice(&stored_key.nonce);
             let nonce = Nonce::from(nonce_array);
 
+            // Decode base64 string to bytes
+            let encrypted_bytes = BASE64
+                .decode(&stored_key.encrypted_key_material)
+                .map_err(|e| KmsError::cryptographic_error("base64_decode", e.to_string()))?;
+
             cipher
-                .decrypt(&nonce, stored_key.encrypted_key_material.as_ref())
+                .decrypt(&nonce, encrypted_bytes.as_ref())
                 .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))?
         } else {
-            stored_key.encrypted_key_material
+            // Decode base64 string to bytes when no encryption
+            BASE64
+                .decode(&stored_key.encrypted_key_material)
+                .map_err(|e| KmsError::cryptographic_error("base64_decode", e.to_string()))?
         };
 
-        Ok(MasterKey {
+        Ok((stored_key, key_material))
+    }
+
+    /// Load a master key from disk
+    async fn load_master_key(&self, key_id: &str) -> Result<MasterKeyInfo> {
+        let (stored_key, _key_material) = self.decode_stored_key(key_id).await?;
+
+        Ok(MasterKeyInfo {
             key_id: stored_key.key_id,
             version: stored_key.version,
             algorithm: stored_key.algorithm,
@@ -158,7 +167,7 @@ impl LocalKmsClient {
     }
 
     /// Save a master key to disk
-    async fn save_master_key(&self, master_key: &MasterKey, key_material: &[u8]) -> Result<()> {
+    async fn save_master_key(&self, master_key: &MasterKeyInfo, key_material: &[u8]) -> Result<()> {
         let key_path = self.master_key_path(&master_key.key_id);
 
         // Encrypt key material if master cipher is available
@@ -170,9 +179,11 @@ impl LocalKmsClient {
             let encrypted = cipher
                 .encrypt(&nonce, key_material)
                 .map_err(|e| KmsError::cryptographic_error("encrypt", e.to_string()))?;
-            (encrypted, nonce.to_vec())
+            // Encode encrypted bytes to base64 string
+            (BASE64.encode(&encrypted), nonce.to_vec())
         } else {
-            (key_material.to_vec(), Vec::new())
+            // Encode key material to base64 string when no encryption
+            (BASE64.encode(key_material), Vec::new())
         };
 
         let stored_key = StoredMasterKey {
@@ -210,39 +221,9 @@ impl LocalKmsClient {
         Ok(())
     }
 
-    /// Generate a random 256-bit key
-    fn generate_key_material() -> Vec<u8> {
-        let mut key_material = vec![0u8; 32]; // 256 bits
-        rand::rng().fill(&mut key_material[..]);
-        key_material
-    }
-
     /// Get the actual key material for a master key
     async fn get_key_material(&self, key_id: &str) -> Result<Vec<u8>> {
-        let key_path = self.master_key_path(key_id);
+        let (_stored_key, key_material) = self.decode_stored_key(key_id).await?;
-
-        if !key_path.exists() {
-            return Err(KmsError::key_not_found(key_id));
-        }
-
-        let content = fs::read(&key_path).await?;
-        let stored_key: StoredMasterKey = serde_json::from_slice(&content)?;
-
-        // Decrypt key material if master cipher is available
-        let key_material = if let Some(ref cipher) = self.master_cipher {
-            if stored_key.nonce.len() != 12 {
-                return Err(KmsError::cryptographic_error("nonce", "Invalid nonce length"));
-            }
-            let mut nonce_array = [0u8; 12];
-            nonce_array.copy_from_slice(&stored_key.nonce);
-            let nonce = Nonce::from(nonce_array);
-            cipher
-                .decrypt(&nonce, stored_key.encrypted_key_material.as_ref())
-                .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))?
-        } else {
-            stored_key.encrypted_key_material
-        };
-
         Ok(key_material)
     }
 
@@ -250,53 +231,22 @@ impl LocalKmsClient {
     async fn encrypt_with_master_key(&self, key_id: &str, plaintext: &[u8]) -> Result<(Vec<u8>, Vec<u8>)> {
         // Load the actual master key material
         let key_material = self.get_key_material(key_id).await?;
-        let key = Key::<Aes256Gcm>::try_from(key_material.as_slice())
+        self.dek_crypto.encrypt(&key_material, plaintext).await
-            .map_err(|_| KmsError::cryptographic_error("key", "Invalid key length"))?;
-        let cipher = Aes256Gcm::new(&key);
-
-        let mut nonce_bytes = [0u8; 12];
-        rand::rng().fill(&mut nonce_bytes[..]);
-
-        let nonce = Nonce::from(nonce_bytes);
-
-        let ciphertext = cipher
-            .encrypt(&nonce, plaintext)
-            .map_err(|e| KmsError::cryptographic_error("encrypt", e.to_string()))?;
-
-        Ok((ciphertext, nonce_bytes.to_vec()))
     }
 
     /// Decrypt data using a master key
     async fn decrypt_with_master_key(&self, key_id: &str, ciphertext: &[u8], nonce: &[u8]) -> Result<Vec<u8>> {
-        if nonce.len() != 12 {
-            return Err(KmsError::cryptographic_error("nonce", "Invalid nonce length"));
-        }
         // Load the actual master key material
         let key_material = self.get_key_material(key_id).await?;
-        let key = Key::<Aes256Gcm>::try_from(key_material.as_slice())
+        self.dek_crypto.decrypt(&key_material, ciphertext, nonce).await
-            .map_err(|_| KmsError::cryptographic_error("key", "Invalid key length"))?;
-        let cipher = Aes256Gcm::new(&key);
-
-        let mut nonce_array = [0u8; 12];
-        nonce_array.copy_from_slice(nonce);
-        let nonce_ref = Nonce::from(nonce_array);
-
-        let plaintext = cipher
-            .decrypt(&nonce_ref, ciphertext)
-            .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))?;
-
-        Ok(plaintext)
     }
 }
 
 #[async_trait]
 impl KmsClient for LocalKmsClient {
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKey> {
+    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
         debug!("Generating data key for master key: {}", request.master_key_id);
 
-        // Verify master key exists
-        let _master_key = self.describe_key(&request.master_key_id, context).await?;
-
         // Generate random data key material
         let key_length = match request.key_spec.as_str() {
             "AES_256" => 32,
@@ -310,7 +260,7 @@ impl KmsClient for LocalKmsClient {
         // Encrypt the data key with the master key
         let (encrypted_key, nonce) = self.encrypt_with_master_key(&request.master_key_id, &plaintext_key).await?;
 
-        // Create data key envelope
+        // Create data key envelope with master key version for rotation support
         let envelope = DataKeyEnvelope {
             key_id: uuid::Uuid::new_v4().to_string(),
             master_key_id: request.master_key_id.clone(),
@@ -324,7 +274,7 @@ impl KmsClient for LocalKmsClient {
         // Serialize the envelope as the ciphertext
         let ciphertext = serde_json::to_vec(&envelope)?;
 
-        let data_key = DataKey::new(envelope.key_id, 1, Some(plaintext_key), ciphertext, request.key_spec.clone());
+        let data_key = DataKeyInfo::new(envelope.key_id, 1, Some(plaintext_key), ciphertext, request.key_spec.clone());
 
         info!("Generated data key for master key: {}", request.master_key_id);
         Ok(data_key)
@@ -359,15 +309,19 @@ impl KmsClient for LocalKmsClient {
         let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)?;
 
         // Verify encryption context matches
-        if !request.encryption_context.is_empty() {
+        // Check that all keys in envelope.encryption_context are present in request.encryption_context
-            for (key, expected_value) in &request.encryption_context {
+        // and their values match. This ensures the context used for decryption matches what was used for encryption.
-                if let Some(actual_value) = envelope.encryption_context.get(key) {
+        for (key, expected_value) in &envelope.encryption_context {
-                    if actual_value != expected_value {
+            if let Some(actual_value) = request.encryption_context.get(key) {
-                        return Err(KmsError::context_mismatch(format!(
+                if actual_value != expected_value {
-                            "Context mismatch for key '{key}': expected '{expected_value}', got '{actual_value}'"
+                    return Err(KmsError::context_mismatch(format!(
-                        )));
+                        "Context mismatch for key '{key}': expected '{expected_value}', got '{actual_value}'"
-                    }
+                    )));
-                } else {
+                }
+            } else {
+                // If request.encryption_context is empty, allow decryption (backward compatibility)
+                // Otherwise, require all envelope context keys to be present
+                if !request.encryption_context.is_empty() {
                     return Err(KmsError::context_mismatch(format!("Missing context key '{key}'")));
                 }
             }
@@ -382,7 +336,7 @@ impl KmsClient for LocalKmsClient {
         Ok(plaintext)
     }
 
-    async fn create_key(&self, key_id: &str, algorithm: &str, context: Option<&OperationContext>) -> Result<MasterKey> {
+    async fn create_key(&self, key_id: &str, algorithm: &str, context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
         debug!("Creating master key: {}", key_id);
 
         // Check if key already exists
@@ -396,13 +350,13 @@ impl KmsClient for LocalKmsClient {
         }
 
         // Generate key material
-        let key_material = Self::generate_key_material();
+        let key_material = generate_key_material(algorithm)?;
 
         let created_by = context
             .map(|ctx| ctx.principal.clone())
             .unwrap_or_else(|| "local-kms".to_string());
 
-        let master_key = MasterKey::new_with_description(key_id.to_string(), algorithm.to_string(), Some(created_by), None);
+        let master_key = MasterKeyInfo::new_with_description(key_id.to_string(), algorithm.to_string(), Some(created_by), None);
 
         // Save to disk
         self.save_master_key(&master_key, &key_material).await?;
@@ -490,7 +444,7 @@ impl KmsClient for LocalKmsClient {
 
         // For simplicity, we'll regenerate key material
         // In a real implementation, we'd preserve the original key material
-        let key_material = Self::generate_key_material();
+        let key_material = generate_key_material(&master_key.algorithm)?;
         self.save_master_key(&master_key, &key_material).await?;
 
         // Update cache
@@ -507,7 +461,7 @@ impl KmsClient for LocalKmsClient {
         let mut master_key = self.load_master_key(key_id).await?;
         master_key.status = KeyStatus::Disabled;
 
-        let key_material = Self::generate_key_material();
+        let key_material = generate_key_material(&master_key.algorithm)?;
         self.save_master_key(&master_key, &key_material).await?;
 
         // Update cache
@@ -529,7 +483,7 @@ impl KmsClient for LocalKmsClient {
         let mut master_key = self.load_master_key(key_id).await?;
         master_key.status = KeyStatus::PendingDeletion;
 
-        let key_material = Self::generate_key_material();
+        let key_material = generate_key_material(&master_key.algorithm)?;
         self.save_master_key(&master_key, &key_material).await?;
 
         // Update cache
@@ -546,7 +500,7 @@ impl KmsClient for LocalKmsClient {
         let mut master_key = self.load_master_key(key_id).await?;
         master_key.status = KeyStatus::Active;
 
-        let key_material = Self::generate_key_material();
+        let key_material = generate_key_material(&master_key.algorithm)?;
         self.save_master_key(&master_key, &key_material).await?;
 
         // Update cache
@@ -557,7 +511,7 @@ impl KmsClient for LocalKmsClient {
         Ok(())
     }
 
-    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKey> {
+    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
         debug!("Rotating key: {}", key_id);
 
         let mut master_key = self.load_master_key(key_id).await?;
@@ -565,7 +519,7 @@ impl KmsClient for LocalKmsClient {
         master_key.rotated_at = Some(Zoned::now());
 
         // Generate new key material
-        let key_material = Self::generate_key_material();
+        let key_material = generate_key_material(&master_key.algorithm)?;
         self.save_master_key(&master_key, &key_material).await?;
 
         // Update cache
@@ -625,12 +579,13 @@ impl KmsBackend for LocalKmsBackend {
 
         // Create master key with description directly
         let _master_key = {
+            let algorithm = "AES_256";
             // Generate key material
-            let key_material = LocalKmsClient::generate_key_material();
+            let key_material = generate_key_material(algorithm)?;
 
-            let master_key = MasterKey::new_with_description(
+            let master_key = MasterKeyInfo::new_with_description(
                 key_id.clone(),
-                "AES_256".to_string(),
+                algorithm.to_string(),
                 Some("local-kms".to_string()),
                 request.description.clone(),
             );
@@ -787,35 +742,19 @@ impl KmsBackend for LocalKmsBackend {
                 return Err(KmsError::invalid_parameter("pending_window_in_days must be between 7 and 30".to_string()));
             }
 
-            let deletion_date = Zoned::now() + jiff::Span::new().days(days as i64);
+            let deletion_date = Zoned::now() + Duration::from_secs(days as u64 * 86400);
             master_key.status = KeyStatus::PendingDeletion;
 
             (Some(deletion_date.to_string()), Some(deletion_date))
         };
 
         // Save the updated key to disk - preserve existing key material!
-        // Load the stored key from disk to get the existing key material
+        // Load and decode the stored key to get the existing key material
-        let key_path = self.client.master_key_path(key_id);
+        let (_stored_key, existing_key_material) = self
-        let content = tokio::fs::read(&key_path)
+            .client
+            .decode_stored_key(key_id)
             .await
-            .map_err(|e| KmsError::internal_error(format!("Failed to read key file: {e}")))?;
+            .map_err(|e| KmsError::internal_error(format!("Failed to decode key: {e}")))?;
-        let stored_key: StoredMasterKey =
-            serde_json::from_slice(&content).map_err(|e| KmsError::internal_error(format!("Failed to parse stored key: {e}")))?;
-
-        // Decrypt the existing key material to preserve it
-        let existing_key_material = if let Some(ref cipher) = self.client.master_cipher {
-            if stored_key.nonce.len() != 12 {
-                return Err(KmsError::cryptographic_error("nonce", "Invalid nonce length"));
-            }
-            let mut nonce_array = [0u8; 12];
-            nonce_array.copy_from_slice(&stored_key.nonce);
-            let nonce = Nonce::from(nonce_array);
-            cipher
-                .decrypt(&nonce, stored_key.encrypted_key_material.as_ref())
-                .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))?
-        } else {
-            stored_key.encrypted_key_material
-        };
 
         self.client.save_master_key(&master_key, &existing_key_material).await?;
 
@@ -861,8 +800,14 @@ impl KmsBackend for LocalKmsBackend {
         master_key.status = KeyStatus::Active;
 
         // Save the updated key to disk - this is the missing critical step!
-        let key_material = LocalKmsClient::generate_key_material();
+        // Preserve existing key material instead of generating new one
-        self.client.save_master_key(&master_key, &key_material).await?;
+        let (_stored_key, existing_key_material) = self
+            .client
+            .decode_stored_key(key_id)
+            .await
+            .map_err(|e| KmsError::internal_error(format!("Failed to decode key: {e}")))?;
+
+        self.client.save_master_key(&master_key, &existing_key_material).await?;
 
         // Update cache
         let mut cache = self.client.key_cache.write().await;

crates/kms/src/backends/mod.rs

@@ -36,7 +36,7 @@ pub trait KmsClient: Send + Sync {
     ///
     /// # Returns
     /// Returns a DataKey containing both plaintext and encrypted key material
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKey>;
+    async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKeyInfo>;
 
     /// Encrypt data directly using a master key
     ///
@@ -67,7 +67,7 @@ pub trait KmsClient: Send + Sync {
     /// * `key_id` - Unique identifier for the new key
     /// * `algorithm` - Key algorithm (e.g., "AES_256")
     /// * `context` - Optional operation context for auditing
-    async fn create_key(&self, key_id: &str, algorithm: &str, context: Option<&OperationContext>) -> Result<MasterKey>;
+    async fn create_key(&self, key_id: &str, algorithm: &str, context: Option<&OperationContext>) -> Result<MasterKeyInfo>;
 
     /// Get information about a specific key
     ///
@@ -139,7 +139,7 @@ pub trait KmsClient: Send + Sync {
     /// # Arguments
     /// * `key_id` - The key identifier
     /// * `context` - Optional operation context for auditing
-    async fn rotate_key(&self, key_id: &str, context: Option<&OperationContext>) -> Result<MasterKey>;
+    async fn rotate_key(&self, key_id: &str, context: Option<&OperationContext>) -> Result<MasterKeyInfo>;
 
     /// Health check
     ///

crates/kms/src/backends/vault.rs

@@ -16,14 +16,15 @@
 
 use crate::backends::{BackendInfo, KmsBackend, KmsClient};
 use crate::config::{KmsConfig, VaultConfig};
+use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
 use crate::error::{KmsError, Result};
 use crate::types::*;
 use async_trait::async_trait;
 use base64::{Engine as _, engine::general_purpose};
 use jiff::Zoned;
-use rand::RngCore;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use std::time::Duration;
 use tracing::{debug, info, warn};
 use vaultrs::{
     client::{VaultClient, VaultClientSettingsBuilder},
@@ -38,6 +39,8 @@ pub struct VaultKmsClient {
     kv_mount: String,
     /// Path prefix for storing keys
     key_path_prefix: String,
+    /// DEK encryption implementation
+    dek_crypto: AesDekCrypto,
 }
 
 /// Key data stored in Vault
@@ -102,6 +105,7 @@ impl VaultKmsClient {
             kv_mount: config.kv_mount.clone(),
             key_path_prefix: config.key_path_prefix.clone(),
             config,
+            dek_crypto: AesDekCrypto::new(),
         })
     }
 
@@ -110,19 +114,6 @@ impl VaultKmsClient {
         format!("{}/{}", self.key_path_prefix, key_id)
     }
 
-    /// Generate key material for the given algorithm
-    fn generate_key_material(algorithm: &str) -> Result<Vec<u8>> {
-        let key_size = match algorithm {
-            "AES_256" => 32,
-            "AES_128" => 16,
-            _ => return Err(KmsError::unsupported_algorithm(algorithm)),
-        };
-
-        let mut key_material = vec![0u8; key_size];
-        rand::rng().fill_bytes(&mut key_material);
-        Ok(key_material)
-    }
-
     /// Encrypt key material using Vault's transit engine
     async fn encrypt_key_material(&self, key_material: &[u8]) -> Result<String> {
         // For simplicity, we'll base64 encode the key material
@@ -139,6 +130,64 @@ impl VaultKmsClient {
             .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))
     }
 
+    /// Get the actual key material for a master key
+    async fn get_key_material(&self, key_id: &str) -> Result<Vec<u8>> {
+        let mut key_data = self.get_key_data(key_id).await?;
+
+        // If encrypted_key_material is empty, generate and store it (fix for old keys)
+        if key_data.encrypted_key_material.is_empty() {
+            warn!("Key {} has empty encrypted_key_material, generating and storing new key material", key_id);
+            let key_material = generate_key_material(&key_data.algorithm)?;
+            key_data.encrypted_key_material = self.encrypt_key_material(&key_material).await?;
+            // Store the updated key data back to Vault
+            self.store_key_data(key_id, &key_data).await?;
+            return Ok(key_material);
+        }
+
+        let key_material = match self.decrypt_key_material(&key_data.encrypted_key_material).await {
+            Ok(km) => km,
+            Err(e) => {
+                warn!("Failed to decrypt key material for key {}: {}, generating new key material", key_id, e);
+                let new_key_material = generate_key_material(&key_data.algorithm)?;
+                key_data.encrypted_key_material = self.encrypt_key_material(&new_key_material).await?;
+                // Store the updated key data back to Vault
+                self.store_key_data(key_id, &key_data).await?;
+                return Ok(new_key_material);
+            }
+        };
+
+        // Validate key material length (should be 32 bytes for AES-256)
+        if key_material.len() != 32 {
+            // Try to fix: generate new key material if length is wrong
+            warn!(
+                "Key {} has invalid key material length ({} bytes), generating new key material",
+                key_id,
+                key_material.len()
+            );
+            let new_key_material = generate_key_material(&key_data.algorithm)?;
+            key_data.encrypted_key_material = self.encrypt_key_material(&new_key_material).await?;
+            // Store the updated key data back to Vault
+            self.store_key_data(key_id, &key_data).await?;
+            return Ok(new_key_material);
+        }
+
+        Ok(key_material)
+    }
+
+    /// Encrypt data using a master key
+    async fn encrypt_with_master_key(&self, key_id: &str, plaintext: &[u8]) -> Result<(Vec<u8>, Vec<u8>)> {
+        // Load the actual master key material
+        let key_material = self.get_key_material(key_id).await?;
+        self.dek_crypto.encrypt(&key_material, plaintext).await
+    }
+
+    /// Decrypt data using a master key
+    async fn decrypt_with_master_key(&self, key_id: &str, ciphertext: &[u8], nonce: &[u8]) -> Result<Vec<u8>> {
+        // Load the actual master key material
+        let key_material = self.get_key_material(key_id).await?;
+        self.dek_crypto.decrypt(&key_material, ciphertext, nonce).await
+    }
+
     /// Store key data in Vault
     async fn store_key_data(&self, key_id: &str, key_data: &VaultKeyData) -> Result<()> {
         let path = self.key_path(key_id);
@@ -154,19 +203,36 @@ impl VaultKmsClient {
     async fn store_key_metadata(&self, key_id: &str, request: &CreateKeyRequest) -> Result<()> {
         debug!("Storing key metadata for {}, input tags: {:?}", key_id, request.tags);
 
+        // Get existing key data to preserve encrypted_key_material and other fields
+        // This is called after create_key, so the key should already exist
+        let mut existing_key_data = self.get_key_data(key_id).await?;
+
+        // If encrypted_key_material is empty, generate it (this handles the case where
+        // an old key was created without proper key material)
+        if existing_key_data.encrypted_key_material.is_empty() {
+            warn!("Key {} has empty encrypted_key_material, generating new key material", key_id);
+            let key_material = generate_key_material(&existing_key_data.algorithm)?;
+            existing_key_data.encrypted_key_material = self.encrypt_key_material(&key_material).await?;
+        }
+
+        // Update only the metadata fields, preserving the encrypted_key_material
         let key_data = VaultKeyData {
-            algorithm: "AES_256".to_string(),
+            algorithm: existing_key_data.algorithm.clone(),
             usage: request.key_usage.clone(),
-            created_at: Zoned::now(),
+            created_at: existing_key_data.created_at,
-            status: KeyStatus::Active,
+            status: existing_key_data.status,
-            version: 1,
+            version: existing_key_data.version,
             description: request.description.clone(),
-            metadata: HashMap::new(),
+            metadata: existing_key_data.metadata.clone(),
             tags: request.tags.clone(),
-            encrypted_key_material: String::new(), // Not used for transit keys
+            encrypted_key_material: existing_key_data.encrypted_key_material.clone(), // Preserve the key material
         };
 
-        debug!("VaultKeyData tags before storage: {:?}", key_data.tags);
+        debug!(
+            "VaultKeyData tags before storage: {:?}, encrypted_key_material length: {}",
+            key_data.tags,
+            key_data.encrypted_key_material.len()
+        );
         self.store_key_data(key_id, &key_data).await
     }
 
@@ -225,36 +291,33 @@ impl VaultKmsClient {
 
 #[async_trait]
 impl KmsClient for VaultKmsClient {
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKey> {
+    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
         debug!("Generating data key for master key: {}", request.master_key_id);
 
-        // Verify master key exists
+        // Generate random data key material using the existing method
-        let _master_key = self.describe_key(&request.master_key_id, context).await?;
+        let plaintext_key = generate_key_material(&request.key_spec)?;
-
-        // Generate data key material
-        let key_length = match request.key_spec.as_str() {
-            "AES_256" => 32,
-            "AES_128" => 16,
-            _ => return Err(KmsError::unsupported_algorithm(&request.key_spec)),
-        };
-
-        let mut plaintext_key = vec![0u8; key_length];
-        rand::rng().fill_bytes(&mut plaintext_key);
 
         // Encrypt the data key with the master key
-        let encrypted_key = self.encrypt_key_material(&plaintext_key).await?;
+        let (encrypted_key, nonce) = self.encrypt_with_master_key(&request.master_key_id, &plaintext_key).await?;
 
-        Ok(DataKey {
+        // Create data key envelope with master key version for rotation support
-            key_id: request.master_key_id.clone(),
+        let envelope = DataKeyEnvelope {
-            version: 1,
+            key_id: uuid::Uuid::new_v4().to_string(),
-            plaintext: Some(plaintext_key),
+            master_key_id: request.master_key_id.clone(),
-            ciphertext: general_purpose::STANDARD
-                .decode(&encrypted_key)
-                .map_err(|e| KmsError::cryptographic_error("decode", e.to_string()))?,
             key_spec: request.key_spec.clone(),
-            metadata: request.encryption_context.clone(),
+            encrypted_key: encrypted_key.clone(),
+            nonce,
+            encryption_context: request.encryption_context.clone(),
             created_at: Zoned::now(),
-        })
+        };
+
+        // Serialize the envelope as the ciphertext
+        let ciphertext = serde_json::to_vec(&envelope)?;
+
+        let data_key = DataKeyInfo::new(envelope.key_id, 1, Some(plaintext_key), ciphertext, request.key_spec.clone());
+
+        info!("Generated data key for master key: {}", request.master_key_id);
+        Ok(data_key)
     }
 
     async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
@@ -279,15 +342,42 @@ impl KmsClient for VaultKmsClient {
         })
     }
 
-    async fn decrypt(&self, _request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
         debug!("Decrypting data");
 
-        // For this simple implementation, we assume the key ID is embedded in the ciphertext metadata
+        // Parse the data key envelope from ciphertext
-        // In practice, you'd extract this from the ciphertext envelope
+        let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)
-        Err(KmsError::invalid_operation("Decrypt not fully implemented for Vault backend"))
+            .map_err(|e| KmsError::cryptographic_error("parse", format!("Failed to parse data key envelope: {e}")))?;
+
+        // Verify encryption context matches
+        // Check that all keys in envelope.encryption_context are present in request.encryption_context
+        // and their values match. This ensures the context used for decryption matches what was used for encryption.
+        for (key, expected_value) in &envelope.encryption_context {
+            if let Some(actual_value) = request.encryption_context.get(key) {
+                if actual_value != expected_value {
+                    return Err(KmsError::context_mismatch(format!(
+                        "Context mismatch for key '{key}': expected '{expected_value}', got '{actual_value}'"
+                    )));
+                }
+            } else {
+                // If request.encryption_context is empty, allow decryption (backward compatibility)
+                // Otherwise, require all envelope context keys to be present
+                if !request.encryption_context.is_empty() {
+                    return Err(KmsError::context_mismatch(format!("Missing context key '{key}'")));
+                }
+            }
+        }
+
+        // Decrypt the data key
+        let plaintext = self
+            .decrypt_with_master_key(&envelope.master_key_id, &envelope.encrypted_key, &envelope.nonce)
+            .await?;
+
+        info!("Successfully decrypted data");
+        Ok(plaintext)
     }
 
-    async fn create_key(&self, key_id: &str, algorithm: &str, _context: Option<&OperationContext>) -> Result<MasterKey> {
+    async fn create_key(&self, key_id: &str, algorithm: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
         debug!("Creating master key: {} with algorithm: {}", key_id, algorithm);
 
         // Check if key already exists
@@ -296,7 +386,7 @@ impl KmsClient for VaultKmsClient {
         }
 
         // Generate key material
-        let key_material = Self::generate_key_material(algorithm)?;
+        let key_material = generate_key_material(algorithm)?;
         let encrypted_material = self.encrypt_key_material(&key_material).await?;
 
         // Create key data
@@ -315,7 +405,7 @@ impl KmsClient for VaultKmsClient {
         // Store in Vault
         self.store_key_data(key_id, &key_data).await?;
 
-        let master_key = MasterKey {
+        let master_key = MasterKeyInfo {
             key_id: key_id.to_string(),
             version: key_data.version,
             algorithm: key_data.algorithm.clone(),
@@ -438,19 +528,19 @@ impl KmsClient for VaultKmsClient {
         Ok(())
     }
 
-    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKey> {
+    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
         debug!("Rotating key: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
         key_data.version += 1;
 
         // Generate new key material
-        let key_material = Self::generate_key_material(&key_data.algorithm)?;
+        let key_material = generate_key_material(&key_data.algorithm)?;
         key_data.encrypted_key_material = self.encrypt_key_material(&key_material).await?;
 
         self.store_key_data(key_id, &key_data).await?;
 
-        let master_key = MasterKey {
+        let master_key = MasterKeyInfo {
             key_id: key_id.to_string(),
             version: key_data.version,
             algorithm: key_data.algorithm,
@@ -506,7 +596,7 @@ impl VaultKmsBackend {
     /// Create a new VaultKmsBackend
     pub async fn new(config: KmsConfig) -> Result<Self> {
         let vault_config = match &config.backend_config {
-            crate::config::BackendConfig::Vault(vault_config) => vault_config.clone(),
+            crate::config::BackendConfig::Vault(vault_config) => (**vault_config).clone(),
             _ => return Err(KmsError::configuration_error("Expected Vault backend configuration")),
         };
 
@@ -681,7 +771,7 @@ impl KmsBackend for VaultKmsBackend {
                 ));
             }
 
-            let deletion_date = Zoned::now() + jiff::Span::new().days(days as i64);
+            let deletion_date = Zoned::now() + Duration::from_secs(days as u64 * 86400);
             key_metadata.key_state = KeyState::PendingDeletion;
             key_metadata.deletion_date = Some(deletion_date.clone());
 

crates/kms/src/config.rs

@@ -69,7 +69,7 @@ pub enum BackendConfig {
     /// Local backend configuration
     Local(LocalConfig),
     /// Vault backend configuration
-    Vault(VaultConfig),
+    Vault(Box<VaultConfig>),
 }
 
 impl Default for BackendConfig {
@@ -194,11 +194,11 @@ impl KmsConfig {
     pub fn vault(address: Url, token: String) -> Self {
         Self {
             backend: KmsBackend::Vault,
-            backend_config: BackendConfig::Vault(VaultConfig {
+            backend_config: BackendConfig::Vault(Box::new(VaultConfig {
                 address: address.to_string(),
                 auth_method: VaultAuthMethod::Token { token },
                 ..Default::default()
-            }),
+            })),
             ..Default::default()
         }
     }
@@ -207,11 +207,11 @@ impl KmsConfig {
     pub fn vault_approle(address: Url, role_id: String, secret_id: String) -> Self {
         Self {
             backend: KmsBackend::Vault,
-            backend_config: BackendConfig::Vault(VaultConfig {
+            backend_config: BackendConfig::Vault(Box::new(VaultConfig {
                 address: address.to_string(),
                 auth_method: VaultAuthMethod::AppRole { role_id, secret_id },
                 ..Default::default()
-            }),
+            })),
             ..Default::default()
         }
     }
@@ -353,7 +353,7 @@ impl KmsConfig {
                 let address = std::env::var("RUSTFS_KMS_VAULT_ADDRESS").unwrap_or_else(|_| "http://localhost:8200".to_string());
                 let token = std::env::var("RUSTFS_KMS_VAULT_TOKEN").unwrap_or_else(|_| "dev-token".to_string());
 
-                config.backend_config = BackendConfig::Vault(VaultConfig {
+                config.backend_config = BackendConfig::Vault(Box::new(VaultConfig {
                     address,
                     auth_method: VaultAuthMethod::Token { token },
                     namespace: std::env::var("RUSTFS_KMS_VAULT_NAMESPACE").ok(),
@@ -362,7 +362,7 @@ impl KmsConfig {
                     key_path_prefix: std::env::var("RUSTFS_KMS_VAULT_KEY_PREFIX")
                         .unwrap_or_else(|_| "rustfs/kms/keys".to_string()),
                     tls: None,
-                });
+                }));
             }
         }
 

crates/kms/src/encryption/dek.rs

@@ -0,0 +1,314 @@
+// 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.
+
+//! Data Encryption Key (DEK) encryption interface and implementations
+//!
+//! This module provides a unified interface for encrypting and decrypting
+//! data encryption keys using master keys. It abstracts the encryption
+//! operations so that different backends can share the same encryption logic.
+
+#![allow(dead_code)] // Trait methods may be used by implementations
+
+use crate::error::{KmsError, Result};
+use async_trait::async_trait;
+use jiff::Zoned;
+use rand::RngCore;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+/// Data key envelope for encrypting/decrypting data keys
+///
+/// This structure stores the encrypted DEK along with metadata needed for decryption.
+/// The `master_key_version` field records which version of the KEK (Key Encryption Key)
+/// was used to encrypt this DEK, enabling proper key rotation support.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DataKeyEnvelope {
+    pub key_id: String,
+    pub master_key_id: String,
+    pub key_spec: String,
+    pub encrypted_key: Vec<u8>,
+    pub nonce: Vec<u8>,
+    pub encryption_context: HashMap<String, String>,
+    pub created_at: Zoned,
+}
+
+/// Trait for encrypting and decrypting data encryption keys (DEK)
+///
+/// This trait abstracts the encryption operations used to protect
+/// data encryption keys with master keys. Different implementations
+/// can use different encryption algorithms (e.g., AES-256-GCM).
+#[async_trait]
+pub trait DekCrypto: Send + Sync {
+    /// Encrypt plaintext data using a master key material
+    ///
+    /// # Arguments
+    /// * `key_material` - The master key material (raw bytes)
+    /// * `plaintext` - The data to encrypt
+    ///
+    /// # Returns
+    /// A tuple of (ciphertext, nonce) where:
+    /// - `ciphertext` - The encrypted data
+    /// - `nonce` - The nonce used for encryption (should be stored with ciphertext)
+    async fn encrypt(&self, key_material: &[u8], plaintext: &[u8]) -> Result<(Vec<u8>, Vec<u8>)>;
+
+    /// Decrypt ciphertext data using a master key material
+    ///
+    /// # Arguments
+    /// * `key_material` - The master key material (raw bytes)
+    /// * `ciphertext` - The encrypted data
+    /// * `nonce` - The nonce used for encryption
+    ///
+    /// # Returns
+    /// The decrypted plaintext data
+    async fn decrypt(&self, key_material: &[u8], ciphertext: &[u8], nonce: &[u8]) -> Result<Vec<u8>>;
+
+    /// Get the algorithm name used by this implementation
+    #[allow(dead_code)] // May be used by implementations or for debugging
+    fn algorithm(&self) -> &'static str;
+
+    /// Get the required key material size in bytes
+    #[allow(dead_code)] // May be used by implementations or for debugging
+    fn key_size(&self) -> usize;
+}
+
+/// AES-256-GCM implementation of DEK encryption
+pub struct AesDekCrypto;
+
+impl AesDekCrypto {
+    /// Create a new AES-256-GCM DEK crypto instance
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+#[async_trait]
+impl DekCrypto for AesDekCrypto {
+    async fn encrypt(&self, key_material: &[u8], plaintext: &[u8]) -> Result<(Vec<u8>, Vec<u8>)> {
+        use aes_gcm::{
+            Aes256Gcm, Key, Nonce,
+            aead::{Aead, KeyInit},
+        };
+
+        // Validate key material length
+        if key_material.len() != 32 {
+            return Err(KmsError::cryptographic_error(
+                "key",
+                format!("Invalid key length: expected 32 bytes, got {}", key_material.len()),
+            ));
+        }
+
+        // Create cipher from key material
+        let key =
+            Key::<Aes256Gcm>::try_from(key_material).map_err(|_| KmsError::cryptographic_error("key", "Invalid key length"))?;
+        let cipher = Aes256Gcm::new(&key);
+
+        // Generate random nonce (12 bytes for GCM)
+        let mut nonce_bytes = [0u8; 12];
+        rand::rng().fill_bytes(&mut nonce_bytes);
+        let nonce = Nonce::from(nonce_bytes);
+
+        // Encrypt plaintext
+        let ciphertext = cipher
+            .encrypt(&nonce, plaintext)
+            .map_err(|e| KmsError::cryptographic_error("encrypt", e.to_string()))?;
+
+        Ok((ciphertext, nonce_bytes.to_vec()))
+    }
+
+    async fn decrypt(&self, key_material: &[u8], ciphertext: &[u8], nonce: &[u8]) -> Result<Vec<u8>> {
+        use aes_gcm::{
+            Aes256Gcm, Key, Nonce,
+            aead::{Aead, KeyInit},
+        };
+
+        // Validate nonce length
+        if nonce.len() != 12 {
+            return Err(KmsError::cryptographic_error("nonce", "Invalid nonce length: expected 12 bytes"));
+        }
+
+        // Validate key material length
+        if key_material.len() != 32 {
+            return Err(KmsError::cryptographic_error(
+                "key",
+                format!("Invalid key length: expected 32 bytes, got {}", key_material.len()),
+            ));
+        }
+
+        // Create cipher from key material
+        let key =
+            Key::<Aes256Gcm>::try_from(key_material).map_err(|_| KmsError::cryptographic_error("key", "Invalid key length"))?;
+        let cipher = Aes256Gcm::new(&key);
+
+        // Convert nonce
+        let mut nonce_array = [0u8; 12];
+        nonce_array.copy_from_slice(nonce);
+        let nonce_ref = Nonce::from(nonce_array);
+
+        // Decrypt ciphertext
+        let plaintext = cipher
+            .decrypt(&nonce_ref, ciphertext)
+            .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))?;
+
+        Ok(plaintext)
+    }
+
+    #[allow(dead_code)] // Trait method, may be used by implementations
+    fn algorithm(&self) -> &'static str {
+        "AES-256-GCM"
+    }
+
+    #[allow(dead_code)] // Trait method, may be used by implementations
+    fn key_size(&self) -> usize {
+        32 // 256 bits
+    }
+}
+
+impl Default for AesDekCrypto {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Generate random key material for the given algorithm
+///
+/// # Arguments
+/// * `algorithm` - The key algorithm (e.g., "AES_256", "AES_128")
+///
+/// # Returns
+/// A vector containing the generated key material
+pub fn generate_key_material(algorithm: &str) -> Result<Vec<u8>> {
+    let key_size = match algorithm {
+        "AES_256" => 32,
+        "AES_128" => 16,
+        _ => return Err(KmsError::unsupported_algorithm(algorithm)),
+    };
+
+    let mut key_material = vec![0u8; key_size];
+    rand::rng().fill_bytes(&mut key_material);
+    Ok(key_material)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_aes_dek_crypto_encrypt_decrypt() {
+        let crypto = AesDekCrypto::new();
+
+        // Generate test key material
+        let key_material = generate_key_material("AES_256").expect("Failed to generate key material");
+        let plaintext = b"Hello, World! This is a test message.";
+
+        // Test encryption
+        let (ciphertext, nonce) = crypto
+            .encrypt(&key_material, plaintext)
+            .await
+            .expect("Encryption should succeed");
+
+        assert!(!ciphertext.is_empty());
+        assert_eq!(nonce.len(), 12);
+        assert_ne!(ciphertext, plaintext);
+
+        // Test decryption
+        let decrypted = crypto
+            .decrypt(&key_material, &ciphertext, &nonce)
+            .await
+            .expect("Decryption should succeed");
+
+        assert_eq!(decrypted, plaintext);
+    }
+
+    #[tokio::test]
+    async fn test_aes_dek_crypto_invalid_key_size() {
+        let crypto = AesDekCrypto::new();
+        let invalid_key = vec![0u8; 16]; // Too short
+        let plaintext = b"test";
+
+        let result = crypto.encrypt(&invalid_key, plaintext).await;
+        assert!(result.is_err());
+    }
+
+    #[tokio::test]
+    async fn test_aes_dek_crypto_invalid_nonce() {
+        let crypto = AesDekCrypto::new();
+        let key_material = generate_key_material("AES_256").expect("Failed to generate key material");
+        let ciphertext = vec![0u8; 16];
+        let invalid_nonce = vec![0u8; 8]; // Too short
+
+        let result = crypto.decrypt(&key_material, &ciphertext, &invalid_nonce).await;
+        assert!(result.is_err());
+    }
+
+    #[tokio::test]
+    async fn test_generate_key_material() {
+        let key_256 = generate_key_material("AES_256").expect("Should generate AES_256 key");
+        assert_eq!(key_256.len(), 32);
+
+        let key_128 = generate_key_material("AES_128").expect("Should generate AES_128 key");
+        assert_eq!(key_128.len(), 16);
+
+        // Keys should be different
+        let key_256_2 = generate_key_material("AES_256").expect("Should generate AES_256 key");
+        assert_ne!(key_256, key_256_2);
+
+        // Invalid algorithm
+        assert!(generate_key_material("INVALID").is_err());
+    }
+
+    #[tokio::test]
+    async fn test_data_key_envelope_serialization() {
+        let envelope = DataKeyEnvelope {
+            key_id: "test-key-id".to_string(),
+            master_key_id: "master-key-id".to_string(),
+            key_spec: "AES_256".to_string(),
+            encrypted_key: vec![1, 2, 3, 4],
+            nonce: vec![5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16],
+            encryption_context: {
+                let mut map = HashMap::new();
+                map.insert("bucket".to_string(), "test-bucket".to_string());
+                map
+            },
+            created_at: Zoned::now(),
+        };
+
+        // Test serialization
+        let serialized = serde_json::to_vec(&envelope).expect("Serialization should succeed");
+        assert!(!serialized.is_empty());
+
+        // Test deserialization
+        let deserialized: DataKeyEnvelope = serde_json::from_slice(&serialized).expect("Deserialization should succeed");
+        assert_eq!(deserialized.key_id, envelope.key_id);
+        assert_eq!(deserialized.master_key_id, envelope.master_key_id);
+        assert_eq!(deserialized.encrypted_key, envelope.encrypted_key);
+    }
+
+    #[tokio::test]
+    async fn test_data_key_envelope_backward_compatibility() {
+        // Test deserialization with current Zoned format (with timezone annotation)
+        let envelope_json = r#"{
+            "key_id": "test-key-id",
+            "master_key_id": "master-key-id",
+            "key_spec": "AES_256",
+            "encrypted_key": [1, 2, 3, 4],
+            "nonce": [5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16],
+            "encryption_context": {"bucket": "test-bucket"},
+            "created_at": "2024-01-01T00:00:00+00:00[UTC]"
+        }"#;
+
+        let deserialized: DataKeyEnvelope = serde_json::from_str(envelope_json).expect("Should deserialize current format");
+        assert_eq!(deserialized.key_id, "test-key-id");
+        assert_eq!(deserialized.master_key_id, "master-key-id");
+    }
+}

crates/kms/src/encryption/mod.rs

@@ -14,7 +14,7 @@
 
 //! Object encryption service implementation
 
-mod ciphers;
+pub mod ciphers;
-pub mod service;
+pub mod dek;
 
-pub use service::ObjectEncryptionService;
+pub use dek::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};

crates/kms/src/lib.rs

@@ -63,6 +63,7 @@ pub mod config;
 mod encryption;
 mod error;
 pub mod manager;
+pub mod service;
 pub mod service_manager;
 pub mod types;
 
@@ -73,10 +74,9 @@ pub use api_types::{
     UntagKeyRequest, UntagKeyResponse, UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse,
 };
 pub use config::*;
-pub use encryption::ObjectEncryptionService;
-pub use encryption::service::DataKey;
 pub use error::{KmsError, Result};
 pub use manager::KmsManager;
+pub use service::{DataKey, ObjectEncryptionService};
 pub use service_manager::{
     KmsServiceManager, KmsServiceStatus, get_global_encryption_service, get_global_kms_service_manager,
     init_global_kms_service_manager,
@@ -112,6 +112,7 @@ pub fn shutdown_global_services() {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::sync::Arc;
     use tempfile::TempDir;
 
     #[tokio::test]
@@ -139,4 +140,91 @@ mod tests {
         // Test stop
         manager.stop().await.expect("Stop should succeed");
     }
+
+    #[tokio::test]
+    async fn test_versioned_service_reconfiguration() {
+        // Test versioned service reconfiguration for zero-downtime
+        let manager = KmsServiceManager::new();
+
+        // Initial state: no version
+        assert!(manager.get_service_version().await.is_none());
+
+        // Start first service
+        let temp_dir1 = TempDir::new().expect("Failed to create temp dir");
+        let config1 = KmsConfig::local(temp_dir1.path().to_path_buf());
+        manager
+            .configure(config1.clone())
+            .await
+            .expect("Configuration should succeed");
+        manager.start().await.expect("Start should succeed");
+
+        // Verify version 1
+        let version1 = manager.get_service_version().await.expect("Service should have version");
+        assert_eq!(version1, 1);
+
+        // Get service reference (simulating ongoing operation)
+        let service1 = manager.get_encryption_service().await.expect("Service should be available");
+
+        // Reconfigure to new service (zero-downtime)
+        let temp_dir2 = TempDir::new().expect("Failed to create temp dir");
+        let config2 = KmsConfig::local(temp_dir2.path().to_path_buf());
+        manager.reconfigure(config2).await.expect("Reconfiguration should succeed");
+
+        // Verify version 2
+        let version2 = manager.get_service_version().await.expect("Service should have version");
+        assert_eq!(version2, 2);
+
+        // Old service reference should still be valid (Arc keeps it alive)
+        // New requests should get version 2
+        let service2 = manager.get_encryption_service().await.expect("Service should be available");
+
+        // Verify they are different instances
+        assert!(!Arc::ptr_eq(&service1, &service2));
+
+        // Old service should still work (simulating long-running operation)
+        // This demonstrates zero-downtime: old operations continue, new operations use new service
+        assert!(service1.health_check().await.is_ok());
+        assert!(service2.health_check().await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn test_concurrent_reconfiguration() {
+        // Test that concurrent reconfiguration requests are serialized
+        let manager = Arc::new(KmsServiceManager::new());
+
+        let temp_dir = TempDir::new().expect("Failed to create temp dir");
+        let base_path = temp_dir.path().to_path_buf();
+
+        // Initial configuration
+        let config1 = KmsConfig::local(base_path.clone());
+        manager.configure(config1).await.expect("Configuration should succeed");
+        manager.start().await.expect("Start should succeed");
+
+        // Spawn multiple concurrent reconfiguration requests
+        let mut handles = Vec::new();
+        for _i in 0..5 {
+            let manager_clone = manager.clone();
+            let path = base_path.clone();
+            let handle = tokio::spawn(async move {
+                let config = KmsConfig::local(path);
+                manager_clone.reconfigure(config).await
+            });
+            handles.push(handle);
+        }
+
+        // Wait for all reconfigurations to complete
+        let mut results = Vec::new();
+        for handle in handles {
+            results.push(handle.await);
+        }
+
+        // All should succeed (serialized by mutex)
+        for result in results {
+            assert!(result.expect("Task should complete").is_ok());
+        }
+
+        // Final version should be 6 (1 initial + 5 reconfigurations)
+        let final_version = manager.get_service_version().await.expect("Service should have version");
+        assert_eq!(final_version, 6);
+    }
 }

crates/kms/src/service.rs

@@ -274,6 +274,8 @@ impl ObjectEncryptionService {
         // Build encryption context
         let mut context = encryption_context.cloned().unwrap_or_default();
         context.insert("bucket".to_string(), bucket.to_string());
+        context.insert("object_key".to_string(), object_key.to_string());
+        // Backward compatibility: also include legacy "object" context key
         context.insert("object".to_string(), object_key.to_string());
         context.insert("algorithm".to_string(), algorithm.as_str().to_string());
 

crates/kms/src/service_manager.rs

@@ -16,11 +16,15 @@
 
 use crate::backends::{KmsBackend, local::LocalKmsBackend};
 use crate::config::{BackendConfig, KmsConfig};
-use crate::encryption::service::ObjectEncryptionService;
 use crate::error::{KmsError, Result};
 use crate::manager::KmsManager;
-use std::sync::{Arc, OnceLock};
+use crate::service::ObjectEncryptionService;
-use tokio::sync::RwLock;
+use arc_swap::ArcSwap;
+use std::sync::{
+    Arc, OnceLock,
+    atomic::{AtomicU64, Ordering},
+};
+use tokio::sync::{Mutex, RwLock};
 use tracing::{error, info, warn};
 
 /// KMS service status
@@ -36,26 +40,43 @@ pub enum KmsServiceStatus {
     Error(String),
 }
 
-/// Dynamic KMS service manager
+/// Service version information for zero-downtime reconfiguration
+#[derive(Clone)]
+struct ServiceVersion {
+    /// Service version number (monotonically increasing)
+    version: u64,
+    /// The encryption service instance
+    service: Arc<ObjectEncryptionService>,
+    /// The KMS manager instance
+    manager: Arc<KmsManager>,
+}
+
+/// Dynamic KMS service manager with versioned services for zero-downtime reconfiguration
 pub struct KmsServiceManager {
-    /// Current KMS manager (if running)
+    /// Current service version (if running)
-    manager: Arc<RwLock<Option<Arc<KmsManager>>>>,
+    /// Uses ArcSwap for atomic, lock-free service switching
-    /// Current encryption service (if running)
+    /// This allows instant atomic updates without blocking readers
-    encryption_service: Arc<RwLock<Option<Arc<ObjectEncryptionService>>>>,
+    current_service: ArcSwap<Option<ServiceVersion>>,
     /// Current configuration
     config: Arc<RwLock<Option<KmsConfig>>>,
     /// Current status
     status: Arc<RwLock<KmsServiceStatus>>,
+    /// Version counter (monotonically increasing)
+    version_counter: Arc<AtomicU64>,
+    /// Mutex to protect lifecycle operations (start, stop, reconfigure)
+    /// This ensures only one lifecycle operation happens at a time
+    lifecycle_mutex: Arc<Mutex<()>>,
 }
 
 impl KmsServiceManager {
     /// Create a new KMS service manager (not configured)
     pub fn new() -> Self {
         Self {
-            manager: Arc::new(RwLock::new(None)),
+            current_service: ArcSwap::from_pointee(None),
-            encryption_service: Arc::new(RwLock::new(None)),
             config: Arc::new(RwLock::new(None)),
             status: Arc::new(RwLock::new(KmsServiceStatus::NotConfigured)),
+            version_counter: Arc::new(AtomicU64::new(0)),
+            lifecycle_mutex: Arc::new(Mutex::new(())),
         }
     }
 
@@ -89,6 +110,12 @@ impl KmsServiceManager {
 
     /// Start KMS service with current configuration
     pub async fn start(&self) -> Result<()> {
+        let _guard = self.lifecycle_mutex.lock().await;
+        self.start_internal().await
+    }
+
+    /// Internal start implementation (called within lifecycle mutex)
+    async fn start_internal(&self) -> Result<()> {
         let config = {
             let config_guard = self.config.read().await;
             match config_guard.as_ref() {
@@ -105,23 +132,11 @@ impl KmsServiceManager {
 
         info!("Starting KMS service with backend: {:?}", config.backend);
 
-        match self.create_backend(&config).await {
+        match self.create_service_version(&config).await {
-            Ok(backend) => {
+            Ok(service_version) => {
-                // Create KMS manager
+                // Atomically update to new service version (lock-free, instant)
-                let kms_manager = Arc::new(KmsManager::new(backend, config));
+                // ArcSwap::store() is a true atomic operation using CAS
-
+                self.current_service.store(Arc::new(Some(service_version)));
-                // Create encryption service
-                let encryption_service = Arc::new(ObjectEncryptionService::new((*kms_manager).clone()));
-
-                // Update manager and service
-                {
-                    let mut manager = self.manager.write().await;
-                    *manager = Some(kms_manager);
-                }
-                {
-                    let mut service = self.encryption_service.write().await;
-                    *service = Some(encryption_service);
-                }
 
                 // Update status
                 {
@@ -143,18 +158,21 @@ impl KmsServiceManager {
     }
 
     /// Stop KMS service
+    ///
+    /// Note: This stops accepting new operations, but existing operations using
+    /// the service will continue until they complete (due to Arc reference counting).
     pub async fn stop(&self) -> Result<()> {
+        let _guard = self.lifecycle_mutex.lock().await;
+        self.stop_internal().await
+    }
+
+    /// Internal stop implementation (called within lifecycle mutex)
+    async fn stop_internal(&self) -> Result<()> {
         info!("Stopping KMS service");
 
-        // Clear manager and service
+        // Atomically clear current service version (lock-free, instant)
-        {
+        // Note: Existing Arc references will keep the service alive until operations complete
-            let mut manager = self.manager.write().await;
+        self.current_service.store(Arc::new(None));
-            *manager = None;
-        }
-        {
-            let mut service = self.encryption_service.write().await;
-            *service = None;
-        }
 
         // Update status (keep configuration)
         {
@@ -164,37 +182,96 @@ impl KmsServiceManager {
             }
         }
 
-        info!("KMS service stopped successfully");
+        info!("KMS service stopped successfully (existing operations may continue)");
         Ok(())
     }
 
-    /// Reconfigure and restart KMS service
+    /// Reconfigure and restart KMS service with zero-downtime
+    ///
+    /// This method implements versioned service switching:
+    /// 1. Creates a new service version without stopping the old one
+    /// 2. Atomically switches to the new version
+    /// 3. Old operations continue using the old service (via Arc reference counting)
+    /// 4. New operations automatically use the new service
+    ///
+    /// This ensures zero downtime during reconfiguration, even for long-running
+    /// operations like encrypting large files.
     pub async fn reconfigure(&self, new_config: KmsConfig) -> Result<()> {
-        info!("Reconfiguring KMS service");
+        let _guard = self.lifecycle_mutex.lock().await;
 
-        // Stop current service if running
+        info!("Reconfiguring KMS service (zero-downtime)");
-        if matches!(self.get_status().await, KmsServiceStatus::Running) {
-            self.stop().await?;
-        }
 
         // Configure with new config
-        self.configure(new_config).await?;
+        {
+            let mut config = self.config.write().await;
+            *config = Some(new_config.clone());
+        }
 
-        // Start with new configuration
+        // Create new service version without stopping old one
-        self.start().await?;
+        // This allows existing operations to continue while new operations use new service
+        match self.create_service_version(&new_config).await {
+            Ok(new_service_version) => {
+                // Get old version for logging (lock-free read)
+                let old_version = self.current_service.load().as_ref().as_ref().map(|sv| sv.version);
 
-        info!("KMS service reconfigured successfully");
+                // Atomically switch to new service version (lock-free, instant CAS operation)
-        Ok(())
+                // This is a true atomic operation - no waiting for locks, instant switch
+                // Old service will be dropped when no more Arc references exist
+                self.current_service.store(Arc::new(Some(new_service_version.clone())));
+
+                // Update status
+                {
+                    let mut status = self.status.write().await;
+                    *status = KmsServiceStatus::Running;
+                }
+
+                if let Some(old_ver) = old_version {
+                    info!(
+                        "KMS service reconfigured successfully: version {} -> {} (old service will be cleaned up when operations complete)",
+                        old_ver, new_service_version.version
+                    );
+                } else {
+                    info!(
+                        "KMS service reconfigured successfully: version {} (service started)",
+                        new_service_version.version
+                    );
+                }
+                Ok(())
+            }
+            Err(e) => {
+                let err_msg = format!("Failed to reconfigure KMS: {e}");
+                error!("{}", err_msg);
+                let mut status = self.status.write().await;
+                *status = KmsServiceStatus::Error(err_msg.clone());
+                Err(KmsError::backend_error(&err_msg))
+            }
+        }
     }
 
     /// Get KMS manager (if running)
+    ///
+    /// Returns the manager from the current service version.
+    /// Uses lock-free atomic load for optimal performance.
     pub async fn get_manager(&self) -> Option<Arc<KmsManager>> {
-        self.manager.read().await.clone()
+        self.current_service.load().as_ref().as_ref().map(|sv| sv.manager.clone())
     }
 
-    /// Get encryption service (if running)  
+    /// Get encryption service (if running)
+    ///
+    /// Returns the service from the current service version.
+    /// Uses lock-free atomic load - no blocking, instant access.
+    /// This ensures new operations always use the latest service version,
+    /// while existing operations continue using their Arc references.
     pub async fn get_encryption_service(&self) -> Option<Arc<ObjectEncryptionService>> {
-        self.encryption_service.read().await.clone()
+        self.current_service.load().as_ref().as_ref().map(|sv| sv.service.clone())
+    }
+
+    /// Get current service version number
+    ///
+    /// Useful for monitoring and debugging.
+    /// Uses lock-free atomic load.
+    pub async fn get_service_version(&self) -> Option<u64> {
+        self.current_service.load().as_ref().as_ref().map(|sv| sv.version)
     }
 
     /// Health check for the KMS service
@@ -226,20 +303,40 @@ impl KmsServiceManager {
         }
     }
 
-    /// Create backend from configuration
+    /// Create a new service version from configuration
-    async fn create_backend(&self, config: &KmsConfig) -> Result<Arc<dyn KmsBackend>> {
+    ///
-        match &config.backend_config {
+    /// This creates a new backend, manager, and service, and assigns it a new version number.
+    async fn create_service_version(&self, config: &KmsConfig) -> Result<ServiceVersion> {
+        // Increment version counter
+        let version = self.version_counter.fetch_add(1, Ordering::Relaxed) + 1;
+
+        info!("Creating KMS service version {} with backend: {:?}", version, config.backend);
+
+        // Create backend
+        let backend = match &config.backend_config {
             BackendConfig::Local(_) => {
-                info!("Creating Local KMS backend");
+                info!("Creating Local KMS backend for version {}", version);
                 let backend = LocalKmsBackend::new(config.clone()).await?;
-                Ok(Arc::new(backend))
+                Arc::new(backend) as Arc<dyn KmsBackend>
             }
             BackendConfig::Vault(_) => {
-                info!("Creating Vault KMS backend");
+                info!("Creating Vault KMS backend for version {}", version);
                 let backend = crate::backends::vault::VaultKmsBackend::new(config.clone()).await?;
-                Ok(Arc::new(backend))
+                Arc::new(backend) as Arc<dyn KmsBackend>
             }
-        }
+        };
+
+        // Create KMS manager
+        let kms_manager = Arc::new(KmsManager::new(backend, config.clone()));
+
+        // Create encryption service
+        let encryption_service = Arc::new(ObjectEncryptionService::new((*kms_manager).clone()));
+
+        Ok(ServiceVersion {
+            version,
+            service: encryption_service,
+            manager: kms_manager,
+        })
     }
 }
 

crates/kms/src/types.rs

@@ -22,7 +22,7 @@ use zeroize::Zeroize;
 
 /// Data encryption key (DEK) used for encrypting object data
 #[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct DataKey {
+pub struct DataKeyInfo {
     /// Key identifier
     pub key_id: String,
     /// Key version
@@ -40,7 +40,7 @@ pub struct DataKey {
     pub created_at: Zoned,
 }
 
-impl DataKey {
+impl DataKeyInfo {
     /// Create a new data key
     ///
     /// # Arguments
@@ -96,7 +96,7 @@ impl DataKey {
 
 /// Master key stored in KMS backend
 #[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct MasterKey {
+pub struct MasterKeyInfo {
     /// Unique key identifier
     pub key_id: String,
     /// Key version
@@ -119,7 +119,7 @@ pub struct MasterKey {
     pub created_by: Option<String>,
 }
 
-impl MasterKey {
+impl MasterKeyInfo {
     /// Create a new master key
     ///
     /// # Arguments
@@ -226,8 +226,8 @@ pub struct KeyInfo {
     pub created_by: Option<String>,
 }
 
-impl From<MasterKey> for KeyInfo {
+impl From<MasterKeyInfo> for KeyInfo {
-    fn from(master_key: MasterKey) -> Self {
+    fn from(master_key: MasterKeyInfo) -> Self {
         Self {
             key_id: master_key.key_id,
             description: master_key.description,
@@ -913,7 +913,7 @@ pub struct CancelKeyDeletionResponse {
 }
 
 // SECURITY: Implement Drop to automatically zero sensitive data when DataKey is dropped
-impl Drop for DataKey {
+impl Drop for DataKeyInfo {
     fn drop(&mut self) {
         self.clear_plaintext();
     }

rustfs/Cargo.toml

@@ -126,6 +126,8 @@ urlencoding = { workspace = true }
 uuid = { workspace = true }
 zip = { workspace = true }
 libc = { workspace = true }
+rand = { workspace = true }
+aes-gcm = { workspace = true }
 
 # Observability and Metrics
 metrics = { workspace = true }

rustfs/src/init.rs

@@ -209,7 +209,7 @@ pub(crate) async fn init_kms_system(opt: &config::Opt) -> std::io::Result<()> {
 
                 rustfs_kms::config::KmsConfig {
                     backend: rustfs_kms::config::KmsBackend::Vault,
-                    backend_config: rustfs_kms::config::BackendConfig::Vault(rustfs_kms::config::VaultConfig {
+                    backend_config: rustfs_kms::config::BackendConfig::Vault(Box::new(rustfs_kms::config::VaultConfig {
                         address: vault_address.clone(),
                         auth_method: rustfs_kms::config::VaultAuthMethod::Token {
                             token: vault_token.clone(),
@@ -219,7 +219,7 @@ pub(crate) async fn init_kms_system(opt: &config::Opt) -> std::io::Result<()> {
                         kv_mount: "secret".to_string(),
                         key_path_prefix: "rustfs/kms/keys".to_string(),
                         tls: None,
-                    }),
+                    })),
                     default_key_id: opt.kms_default_key_id.clone(),
                     timeout: std::time::Duration::from_secs(30),
                     retry_attempts: 3,

rustfs/src/storage/ecfs_extend.rs

@@ -17,7 +17,7 @@ use crate::config::workload_profiles::{
 };
 use crate::error::ApiError;
 use crate::server::cors;
-use crate::storage::ecfs::{InMemoryAsyncReader, ListObjectUnorderedQuery};
+use crate::storage::ecfs::ListObjectUnorderedQuery;
 use axum::body::Body;
 use http::{HeaderMap, HeaderValue, StatusCode};
 use metrics::counter;
@@ -28,9 +28,6 @@ use rustfs_ecstore::bucket::replication::ReplicationConfigurationExt;
 use rustfs_ecstore::error::StorageError;
 use rustfs_ecstore::store_api::{BucketOptions, ObjectInfo, ObjectToDelete};
 use rustfs_ecstore::{StorageAPI, new_object_layer_fn};
-use rustfs_filemeta::ObjectPartInfo;
-use rustfs_kms::{EncryptionMetadata, ObjectEncryptionContext, get_global_encryption_service};
-use rustfs_rio::{DecryptReader, Reader, WarpReader};
 use rustfs_targets::EventName;
 use rustfs_targets::arn::{TargetID, TargetIDError};
 use rustfs_utils::http::{
@@ -40,7 +37,7 @@ use rustfs_utils::http::{
 use s3s::dto::{
     Delimiter, LambdaFunctionConfiguration, NotificationConfigurationFilter, ObjectLockConfiguration, ObjectLockEnabled,
     ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, ObjectLockRetentionMode, QueueConfiguration,
-    ServerSideEncryption, TopicConfiguration,
+    TopicConfiguration,
 };
 use s3s::{S3Error, S3ErrorCode, S3Response, S3Result};
 use serde_urlencoded::from_bytes;
@@ -50,7 +47,6 @@ use std::sync::Arc;
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
 use time::{format_description::FormatItem, macros::format_description};
-use tokio::io::AsyncRead;
 use tracing::{debug, warn};
 
 pub const RFC1123: &[FormatItem<'_>] =
@@ -326,180 +322,6 @@ pub(crate) fn get_buffer_size_opt_in(file_size: i64) -> usize {
     buffer_size
 }
 
-pub(crate) async fn create_managed_encryption_material(
-    bucket: &str,
-    key: &str,
-    algorithm: &ServerSideEncryption,
-    kms_key_id: Option<String>,
-    original_size: i64,
-) -> Result<crate::storage::ecfs::ManagedEncryptionMaterial, ApiError> {
-    let Some(service) = get_global_encryption_service().await else {
-        return Err(ApiError::from(StorageError::other("KMS encryption service is not initialized")));
-    };
-
-    if !is_managed_sse(algorithm) {
-        return Err(ApiError::from(StorageError::other(format!(
-            "Unsupported server-side encryption algorithm: {}",
-            algorithm.as_str()
-        ))));
-    }
-
-    let algorithm_str = algorithm.as_str();
-
-    let mut context = ObjectEncryptionContext::new(bucket.to_string(), key.to_string());
-    if original_size >= 0 {
-        context = context.with_size(original_size as u64);
-    }
-
-    let mut kms_key_candidate = kms_key_id;
-    if kms_key_candidate.is_none() {
-        kms_key_candidate = service.get_default_key_id().cloned();
-    }
-
-    let kms_key_to_use = kms_key_candidate
-        .clone()
-        .ok_or_else(|| ApiError::from(StorageError::other("No KMS key available for managed server-side encryption")))?;
-
-    let (data_key, encrypted_data_key) = service
-        .create_data_key(&kms_key_candidate, &context)
-        .await
-        .map_err(|e| ApiError::from(StorageError::other(format!("Failed to create data key: {e}"))))?;
-
-    let metadata = EncryptionMetadata {
-        algorithm: algorithm_str.to_string(),
-        key_id: kms_key_to_use.clone(),
-        key_version: 1,
-        iv: data_key.nonce.to_vec(),
-        tag: None,
-        encryption_context: context.encryption_context.clone(),
-        encrypted_at: jiff::Zoned::now(),
-        original_size: if original_size >= 0 { original_size as u64 } else { 0 },
-        encrypted_data_key,
-    };
-
-    let mut headers = service.metadata_to_headers(&metadata);
-    headers.insert("x-rustfs-encryption-original-size".to_string(), metadata.original_size.to_string());
-
-    Ok(crate::storage::ecfs::ManagedEncryptionMaterial {
-        data_key,
-        headers,
-        kms_key_id: kms_key_to_use,
-    })
-}
-
-pub(crate) async fn decrypt_managed_encryption_key(
-    bucket: &str,
-    key: &str,
-    metadata: &HashMap<String, String>,
-) -> Result<Option<([u8; 32], [u8; 12], Option<i64>)>, ApiError> {
-    if !metadata.contains_key("x-rustfs-encryption-key") {
-        return Ok(None);
-    }
-
-    let Some(service) = get_global_encryption_service().await else {
-        return Err(ApiError::from(StorageError::other("KMS encryption service is not initialized")));
-    };
-
-    let parsed = service
-        .headers_to_metadata(metadata)
-        .map_err(|e| ApiError::from(StorageError::other(format!("Failed to parse encryption metadata: {e}"))))?;
-
-    if parsed.iv.len() != 12 {
-        return Err(ApiError::from(StorageError::other("Invalid encryption nonce length; expected 12 bytes")));
-    }
-
-    let context = ObjectEncryptionContext::new(bucket.to_string(), key.to_string());
-    let data_key = service
-        .decrypt_data_key(&parsed.encrypted_data_key, &context)
-        .await
-        .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt data key: {e}"))))?;
-
-    let key_bytes = data_key.plaintext_key;
-    let mut nonce = [0u8; 12];
-    nonce.copy_from_slice(&parsed.iv[..12]);
-
-    let original_size = metadata
-        .get("x-rustfs-encryption-original-size")
-        .and_then(|s| s.parse::<i64>().ok());
-
-    Ok(Some((key_bytes, nonce, original_size)))
-}
-
-pub(crate) fn derive_part_nonce(base: [u8; 12], part_number: usize) -> [u8; 12] {
-    let mut nonce = base;
-    let current = u32::from_be_bytes([nonce[8], nonce[9], nonce[10], nonce[11]]);
-    let incremented = current.wrapping_add(part_number as u32);
-    nonce[8..12].copy_from_slice(&incremented.to_be_bytes());
-    nonce
-}
-
-pub(crate) async fn decrypt_multipart_managed_stream(
-    mut encrypted_stream: Box<dyn AsyncRead + Unpin + Send + Sync>,
-    parts: &[ObjectPartInfo],
-    key_bytes: [u8; 32],
-    base_nonce: [u8; 12],
-) -> Result<(Box<dyn Reader>, i64), StorageError> {
-    let total_plain_capacity: usize = parts.iter().map(|part| part.actual_size.max(0) as usize).sum();
-
-    let mut plaintext = Vec::with_capacity(total_plain_capacity);
-
-    for part in parts {
-        if part.size == 0 {
-            continue;
-        }
-
-        let mut encrypted_part = vec![0u8; part.size];
-        tokio::io::AsyncReadExt::read_exact(&mut encrypted_stream, &mut encrypted_part)
-            .await
-            .map_err(|e| StorageError::other(format!("failed to read encrypted multipart segment {}: {}", part.number, e)))?;
-
-        let part_nonce = derive_part_nonce(base_nonce, part.number);
-        let cursor = std::io::Cursor::new(encrypted_part);
-        let mut decrypt_reader = DecryptReader::new(WarpReader::new(cursor), key_bytes, part_nonce);
-
-        tokio::io::AsyncReadExt::read_to_end(&mut decrypt_reader, &mut plaintext)
-            .await
-            .map_err(|e| StorageError::other(format!("failed to decrypt multipart segment {}: {}", part.number, e)))?;
-    }
-
-    let total_plain_size = plaintext.len() as i64;
-    let reader = Box::new(WarpReader::new(InMemoryAsyncReader::new(plaintext))) as Box<dyn Reader>;
-
-    Ok((reader, total_plain_size))
-}
-
-pub(crate) fn strip_managed_encryption_metadata(metadata: &mut HashMap<String, String>) {
-    const KEYS: [&str; 7] = [
-        "x-amz-server-side-encryption",
-        "x-amz-server-side-encryption-aws-kms-key-id",
-        "x-rustfs-encryption-iv",
-        "x-rustfs-encryption-tag",
-        "x-rustfs-encryption-key",
-        "x-rustfs-encryption-context",
-        "x-rustfs-encryption-original-size",
-    ];
-
-    for key in KEYS.iter() {
-        metadata.remove(*key);
-    }
-}
-
-/// Check if the given server-side encryption algorithm is a managed SSE type
-///
-/// This function checks if the provided ServerSideEncryption algorithm
-/// corresponds to a managed server-side encryption method, specifically
-/// "AES256" or "aws:kms".
-///
-/// # Arguments
-/// * `algorithm` - A reference to the ServerSideEncryption enum to check.
-///
-/// # Returns
-/// * `true` if the algorithm is "AES256" or "aws:kms", otherwise `false`.
-///
-pub(crate) fn is_managed_sse(algorithm: &ServerSideEncryption) -> bool {
-    matches!(algorithm.as_str(), "AES256" | "aws:kms")
-}
-
 /// Validate object key for control characters and log special characters
 ///
 /// This function:

rustfs/src/storage/mod.rs

@@ -26,5 +26,8 @@ mod ecfs_extend;
 #[cfg(test)]
 mod ecfs_test;
 pub(crate) mod head_prefix;
+mod sse;
+#[cfg(test)]
+mod sse_test;
 
 pub(crate) use ecfs_extend::*;

rustfs/src/storage/sse_test.rs

@@ -0,0 +1,250 @@
+// 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.
+
+#[cfg(test)]
+mod tests {
+    use crate::storage::sse::SseDekProvider;
+    use crate::storage::sse::TestSseDekProvider;
+    use rustfs_rio::{DecryptReader, EncryptReader, WarpReader};
+    use std::io::Cursor;
+    use tokio::io::AsyncReadExt;
+
+    /// Test EncryptReader encryption and DecryptReader decryption integration without KMS
+    /// This test verifies the complete encryption/decryption flow:
+    /// 1. Create TestSseDekProvider with a test master key
+    /// 2. Generate data encryption key (DEK) using the provider
+    /// 3. Encrypt data using EncryptReader
+    /// 4. Decrypt data using DecryptReader
+    /// 5. Verify decrypted data matches original plaintext
+    #[tokio::test]
+    async fn test_encrypt_reader_decrypt_reader_integration_without_kms() {
+        // Step 1: Create TestSseDekProvider with test master key
+        let provider = TestSseDekProvider::new_with_key([0x42u8; 32]);
+
+        // Step 2: Generate a data encryption key
+        let bucket = "test-bucket";
+        let key = "test-key";
+        let kms_key_id = "default"; // Key ID is ignored in test provider
+
+        let (data_key, _encrypted_dek) = provider
+            .generate_sse_dek(bucket, key, kms_key_id)
+            .await
+            .expect("Failed to generate DEK");
+
+        // Verify data key properties
+        assert_eq!(data_key.plaintext_key.len(), 32);
+        assert_eq!(data_key.nonce.len(), 12);
+
+        // Step 3: Prepare test data
+        let plaintext = b"Hello, World! This is a test message for encryption and decryption.";
+        println!("Original plaintext: {:?}", String::from_utf8_lossy(plaintext));
+        println!("Plaintext length: {} bytes", plaintext.len());
+
+        // Step 4: Encrypt using EncryptReader (wrap Cursor with WarpReader)
+        let plaintext_reader = WarpReader::new(Cursor::new(plaintext.to_vec()));
+        let mut encrypt_reader = EncryptReader::new(plaintext_reader, data_key.plaintext_key, data_key.nonce);
+
+        // Read encrypted data
+        let mut encrypted_data = Vec::new();
+        encrypt_reader
+            .read_to_end(&mut encrypted_data)
+            .await
+            .expect("Failed to read encrypted data");
+
+        println!("Encrypted data length: {} bytes", encrypted_data.len());
+        println!(
+            "First 16 bytes of encrypted data: {:02x?}",
+            &encrypted_data[..16.min(encrypted_data.len())]
+        );
+
+        // Verify encrypted data is different from plaintext
+        assert_ne!(
+            &encrypted_data[..plaintext.len().min(encrypted_data.len())],
+            plaintext,
+            "Encrypted data should be different from plaintext"
+        );
+
+        // Step 5: Decrypt using DecryptReader (wrap Cursor with WarpReader)
+        let encrypted_reader = WarpReader::new(Cursor::new(encrypted_data));
+        let mut decrypt_reader = DecryptReader::new(encrypted_reader, data_key.plaintext_key, data_key.nonce);
+
+        // Read decrypted data
+        let mut decrypted_data = Vec::new();
+        decrypt_reader
+            .read_to_end(&mut decrypted_data)
+            .await
+            .expect("Failed to read decrypted data");
+
+        println!("Decrypted data: {:?}", String::from_utf8_lossy(&decrypted_data));
+        println!("Decrypted length: {} bytes", decrypted_data.len());
+
+        // Step 6: Verify decrypted data matches original plaintext
+        assert_eq!(decrypted_data, plaintext, "Decrypted data should match original plaintext");
+
+        println!("✅ EncryptReader/DecryptReader integration test passed!");
+    }
+
+    /// Test EncryptReader with large data (10MB) without KMS
+    #[tokio::test]
+    async fn test_encrypt_reader_large_data_without_kms() {
+        // Create TestSseDekProvider with test master key
+        let provider = TestSseDekProvider::new_with_key([0x42u8; 32]);
+
+        let bucket = "test-bucket";
+        let key = "test-key-large";
+        let kms_key_id = "default";
+
+        let (data_key, _encrypted_dek) = provider
+            .generate_sse_dek(bucket, key, kms_key_id)
+            .await
+            .expect("Failed to generate DEK");
+
+        // Create 1MB of test data
+        let plaintext_size = 1024 * 1024 * 10;
+        let plaintext: Vec<u8> = (0..plaintext_size).map(|i| (i % 256) as u8).collect();
+        println!("Testing with {} bytes of data", plaintext.len());
+
+        // Encrypt (wrap with WarpReader)
+        let plaintext_reader = WarpReader::new(Cursor::new(plaintext.clone()));
+        let mut encrypt_reader = EncryptReader::new(plaintext_reader, data_key.plaintext_key, data_key.nonce);
+
+        let mut encrypted_data = Vec::new();
+        encrypt_reader
+            .read_to_end(&mut encrypted_data)
+            .await
+            .expect("Failed to encrypt large data");
+
+        println!("Encrypted {} bytes to {} bytes", plaintext.len(), encrypted_data.len());
+
+        // Decrypt (wrap with WarpReader)
+        let encrypted_reader = WarpReader::new(Cursor::new(encrypted_data));
+        let mut decrypt_reader = DecryptReader::new(encrypted_reader, data_key.plaintext_key, data_key.nonce);
+
+        let mut decrypted_data = Vec::new();
+        decrypt_reader
+            .read_to_end(&mut decrypted_data)
+            .await
+            .expect("Failed to decrypt large data");
+
+        // Verify
+        assert_eq!(decrypted_data.len(), plaintext.len(), "Decrypted size should match original");
+        assert_eq!(decrypted_data, plaintext, "Decrypted data should match original plaintext");
+
+        println!("✅ Large data encryption/decryption test passed!");
+    }
+
+    /// Test EncryptReader with different nonces produce different ciphertexts
+    #[tokio::test]
+    async fn test_encrypt_reader_different_nonces_produce_different_ciphertext() {
+        // Create TestSseDekProvider with test master key
+        let provider = TestSseDekProvider::new_with_key([0x42u8; 32]);
+
+        let bucket = "test-bucket";
+        let key = "test-key";
+        let kms_key_id = "default";
+
+        // Generate two different keys (with different nonces)
+        let (data_key1, _) = provider
+            .generate_sse_dek(bucket, key, kms_key_id)
+            .await
+            .expect("Failed to generate DEK 1");
+
+        let (data_key2, _) = provider
+            .generate_sse_dek(bucket, key, kms_key_id)
+            .await
+            .expect("Failed to generate DEK 2");
+
+        // Verify nonces are different
+        assert_ne!(data_key1.nonce, data_key2.nonce, "Different keys should have different nonces");
+
+        // Same plaintext
+        let plaintext = b"Same plaintext";
+
+        // Encrypt with first key (wrap with WarpReader)
+        let reader1 = WarpReader::new(Cursor::new(plaintext.to_vec()));
+        let mut encrypt_reader1 = EncryptReader::new(reader1, data_key1.plaintext_key, data_key1.nonce);
+        let mut encrypted1 = Vec::new();
+        encrypt_reader1.read_to_end(&mut encrypted1).await.unwrap();
+
+        // Encrypt with second key (wrap with WarpReader)
+        let reader2 = WarpReader::new(Cursor::new(plaintext.to_vec()));
+        let mut encrypt_reader2 = EncryptReader::new(reader2, data_key2.plaintext_key, data_key2.nonce);
+        let mut encrypted2 = Vec::new();
+        encrypt_reader2.read_to_end(&mut encrypted2).await.unwrap();
+
+        // Verify ciphertexts are different (due to different nonces/keys)
+        assert_ne!(
+            encrypted1, encrypted2,
+            "Same plaintext with different nonces should produce different ciphertext"
+        );
+
+        println!("✅ Different nonces produce different ciphertext - test passed!");
+    }
+
+    /// Test EncryptReader with decrypted DEK (simulating full cycle)
+    #[tokio::test]
+    async fn test_encrypt_reader_with_decrypted_dek() {
+        // Create TestSseDekProvider with test master key
+        let provider = TestSseDekProvider::new_with_key([0x42u8; 32]);
+
+        let bucket = "test-bucket";
+        let key = "test-key";
+        let kms_key_id = "default";
+
+        // Step 1: Generate DEK and get encrypted DEK
+        let (data_key, encrypted_dek) = provider
+            .generate_sse_dek(bucket, key, kms_key_id)
+            .await
+            .expect("Failed to generate DEK");
+
+        let original_plaintext_key = data_key.plaintext_key;
+        let original_nonce = data_key.nonce;
+
+        // Step 2: Later, decrypt the DEK (simulating GET operation)
+        let decrypted_plaintext_key = provider
+            .decrypt_sse_dek(&encrypted_dek, kms_key_id)
+            .await
+            .expect("Failed to decrypt DEK");
+
+        // Step 3: Verify decrypted key matches original
+        assert_eq!(
+            decrypted_plaintext_key, original_plaintext_key,
+            "Decrypted DEK should match original plaintext key"
+        );
+
+        // Step 4: Use decrypted key to encrypt/decrypt data
+        let plaintext = b"Test data with decrypted DEK";
+
+        // Encrypt with original key (wrap with WarpReader)
+        let reader = WarpReader::new(Cursor::new(plaintext.to_vec()));
+        let mut encrypt_reader = EncryptReader::new(reader, original_plaintext_key, original_nonce);
+        let mut encrypted_data = Vec::new();
+        encrypt_reader.read_to_end(&mut encrypted_data).await.unwrap();
+
+        // Decrypt with recovered key (simulating GET operation) (wrap with WarpReader)
+        let reader = WarpReader::new(Cursor::new(encrypted_data));
+        let mut decrypt_reader = DecryptReader::new(
+            reader,
+            decrypted_plaintext_key,
+            original_nonce, // In real scenario, read from metadata
+        );
+        let mut decrypted_data = Vec::new();
+        decrypt_reader.read_to_end(&mut decrypted_data).await.unwrap();
+
+        // Step 5: Verify
+        assert_eq!(decrypted_data, plaintext, "Data decrypted with recovered key should match original");
+
+        println!("✅ Full cycle (generate -> encrypt DEK -> decrypt DEK -> decrypt data) test passed!");
+    }
+}