fix

Refactor KMS and SSE integration by introducing a local master key option and enhancing key management. This update includes changes to the KMS configuration, updates to the local backend for key handling, and improvements in the SSE encryption and decryption processes, ensuring better clarity and functionality.
Enhance SSE decryption process by consolidating encryption metadata handling and improving key validation. This update introduces clearer management of server-side encryption parameters and refines the decryption logic for better performance and security.
2026-01-17 09:40:32 +00:00 · 2026-01-17 16:42:09 +08:00 · 2026-01-17 16:19:58 +08:00 · 2026-01-17 12:32:50 +08:00 · 2026-01-17 01:48:37 +08:00 · 2026-01-17 01:15:20 +08:00
30 changed files with 4200 additions and 1078 deletions
--- a/.vscode/launch.json
+++ b/.vscode/launch.json
@@ -121,6 +121,50 @@
                "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/volume/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": "Debug executable target/debug/test",
            "type":  "lldb",
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -710,9 +710,9 @@ dependencies = [

 [[package]]
 name = "aws-runtime"
-version = "1.5.17"
+version = "1.5.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d81b5b2898f6798ad58f484856768bca817e3cd9de0974c24ae0f1113fe88f1b"
+checksum = "959dab27ce613e6c9658eb3621064d0e2027e5f2acb65bc526a43577facea557"
 dependencies = [
 "aws-credential-types",
 "aws-sigv4",
@@ -735,9 +735,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-s3"
-version = "1.119.0"
+version = "1.120.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1d65fddc3844f902dfe1864acb8494db5f9342015ee3ab7890270d36fbd2e01c"
+checksum = "06673901e961f20fa8d7da907da48f7ad6c1b383e3726c22bd418900f015abe1"
 dependencies = [
 "aws-credential-types",
 "aws-runtime",
@@ -747,6 +747,7 @@ dependencies = [
 "aws-smithy-eventstream",
 "aws-smithy-http",
 "aws-smithy-json",
+ "aws-smithy-observability",
 "aws-smithy-runtime",
 "aws-smithy-runtime-api",
 "aws-smithy-types",
@@ -769,15 +770,16 @@ dependencies = [

 [[package]]
 name = "aws-sdk-sso"
-version = "1.91.0"
+version = "1.92.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ee6402a36f27b52fe67661c6732d684b2635152b676aa2babbfb5204f99115d"
+checksum = "b7d63bd2bdeeb49aa3f9b00c15e18583503b778b2e792fc06284d54e7d5b6566"
 dependencies = [
 "aws-credential-types",
 "aws-runtime",
 "aws-smithy-async",
 "aws-smithy-http",
 "aws-smithy-json",
+ "aws-smithy-observability",
 "aws-smithy-runtime",
 "aws-smithy-runtime-api",
 "aws-smithy-types",
@@ -791,15 +793,16 @@ dependencies = [

 [[package]]
 name = "aws-sdk-ssooidc"
-version = "1.93.0"
+version = "1.94.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a45a7f750bbd170ee3677671ad782d90b894548f4e4ae168302c57ec9de5cb3e"
+checksum = "532d93574bf731f311bafb761366f9ece345a0416dbcc273d81d6d1a1205239b"
 dependencies = [
 "aws-credential-types",
 "aws-runtime",
 "aws-smithy-async",
 "aws-smithy-http",
 "aws-smithy-json",
+ "aws-smithy-observability",
 "aws-smithy-runtime",
 "aws-smithy-runtime-api",
 "aws-smithy-types",
@@ -813,15 +816,16 @@ dependencies = [

 [[package]]
 name = "aws-sdk-sts"
-version = "1.95.0"
+version = "1.96.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "55542378e419558e6b1f398ca70adb0b2088077e79ad9f14eb09441f2f7b2164"
+checksum = "357e9a029c7524db6a0099cd77fbd5da165540339e7296cca603531bc783b56c"
 dependencies = [
 "aws-credential-types",
 "aws-runtime",
 "aws-smithy-async",
 "aws-smithy-http",
 "aws-smithy-json",
+ "aws-smithy-observability",
 "aws-smithy-query",
 "aws-smithy-runtime",
 "aws-smithy-runtime-api",
@@ -1631,9 +1635,9 @@ dependencies = [

 [[package]]
 name = "cmov"
-version = "0.4.4"
+version = "0.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "360a5d5b750cd7fb97d5ead6e6e0ef0b288d3c2464a189f04f38670e268842ed"
+checksum = "b1339d398d44e506d9b72c1af2f6f51a41c9c64f9a0738eb9aedede47ed1f683"

 [[package]]
 name = "colorchoice"
@@ -1812,6 +1816,17 @@ dependencies = [
 "rand 0.9.2",
 ]

+[[package]]
+name = "core_affinity"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a034b3a7b624016c6e13f5df875747cc25f884156aad2abd12b6c46797971342"
+dependencies = [
+ "libc",
+ "num_cpus",
+ "winapi",
+]
+
 [[package]]
 name = "cpp_demangle"
 version = "0.4.5"
@@ -3216,6 +3231,7 @@ dependencies = [
 "rustfs-common",
 "rustfs-ecstore",
 "rustfs-filemeta",
+ "rustfs-iam",
 "rustfs-lock",
 "rustfs-madmin",
 "rustfs-protos",
@@ -3597,11 +3613,12 @@ dependencies = [

 [[package]]
 name = "flexi_logger"
-version = "0.31.7"
+version = "0.31.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "31e5335674a3a259527f97e9176a3767dcc9b220b8e29d643daeb2d6c72caf8b"
+checksum = "aea7feddba9b4e83022270d49a58d4a1b3fdad04b34f78cf1ce471f698e42672"
 dependencies = [
 "chrono",
+ "core_affinity",
 "crossbeam-channel",
 "crossbeam-queue",
 "flate2",
@@ -4198,8 +4215,6 @@ version = "0.15.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
 dependencies = [
- "allocator-api2",
- "equivalent",
 "foldhash 0.1.5",
 ]

@@ -5297,11 +5312,11 @@ dependencies = [

 [[package]]
 name = "lru"
-version = "0.12.5"
+version = "0.16.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38"
+checksum = "a1dc47f592c06f33f8e3aea9591776ec7c9f9e4124778ff8a3c3b87159f7e593"
 dependencies = [
- "hashbrown 0.15.5",
+ "hashbrown 0.16.1",
 ]

 [[package]]
@@ -5641,9 +5656,9 @@ dependencies = [

 [[package]]
 name = "notify-debouncer-mini"
-version = "0.6.0"
+version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a689eb4262184d9a1727f9087cd03883ea716682ab03ed24efec57d7716dccb8"
+checksum = "17849edfaabd9a5fef1c606d99cfc615a8e99f7ac4366406d86c7942a3184cf2"
 dependencies = [
 "log",
 "notify",
@@ -7558,9 +7573,9 @@ dependencies = [

 [[package]]
 name = "rustc-demangle"
-version = "0.1.26"
+version = "0.1.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "56f7d92ca342cea22a06f2121d944b4fd82af56988c270852495420f961d4ace"
+checksum = "b50b8869d9fc858ce7266cce0194bd74df58b9d0e3f6df3a9fc8eb470d95c09d"

 [[package]]
 name = "rustc-hash"
@@ -7581,6 +7596,7 @@ dependencies = [
 name = "rustfs"
 version = "0.0.5"
 dependencies = [
+ "aes-gcm 0.11.0-rc.2",
 "astral-tokio-tar",
 "async-trait",
 "atoi",
@@ -7615,6 +7631,7 @@ dependencies = [
 "moka",
 "pin-project-lite",
 "pprof",
+ "rand 0.10.0-rc.6",
 "reqwest",
 "rmp-serde",
 "russh",
@@ -7942,6 +7959,7 @@ name = "rustfs-kms"
 version = "0.0.5"
 dependencies = [
 "aes-gcm 0.11.0-rc.2",
+ "arc-swap",
 "async-trait",
 "base64",
 "chacha20poly1305",
@@ -8390,9 +8408,9 @@ dependencies = [

 [[package]]
 name = "rustls-pki-types"
-version = "1.13.2"
+version = "1.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "21e6f2ab2928ca4291b86736a8bd920a277a399bba1589409d72154ff87c1282"
+checksum = "4910321ebe4151be888e35fe062169554e74aad01beafed60410131420ceffbc"
 dependencies = [
 "web-time",
 "zeroize",
@@ -8739,11 +8757,11 @@ dependencies = [

 [[package]]
 name = "serde_spanned"
-version = "0.6.9"
+version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3"
+checksum = "f8bbf91e5a4d6315eee45e704372590b30e260ee83af6639d64557f51b067776"
 dependencies = [
- "serde",
+ "serde_core",
 ]

 [[package]]
@@ -9881,44 +9899,42 @@ dependencies = [

 [[package]]
 name = "toml"
-version = "0.8.23"
+version = "0.9.11+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362"
-dependencies = [
- "serde",
- "serde_spanned",
- "toml_datetime",
- "toml_edit",
-]
-
-[[package]]
-name = "toml_datetime"
-version = "0.6.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c"
-dependencies = [
- "serde",
-]
-
-[[package]]
-name = "toml_edit"
-version = "0.22.27"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a"
+checksum = "f3afc9a848309fe1aaffaed6e1546a7a14de1f935dc9d89d32afd9a44bab7c46"
 dependencies = [
 "indexmap 2.13.0",
- "serde",
+ "serde_core",
 "serde_spanned",
 "toml_datetime",
- "toml_write",
+ "toml_parser",
+ "toml_writer",
 "winnow",
 ]

 [[package]]
-name = "toml_write"
-version = "0.1.2"
+name = "toml_datetime"
+version = "0.7.5+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801"
+checksum = "92e1cfed4a3038bc5a127e35a2d360f145e1f4b971b551a2ba5fd7aedf7e1347"
+dependencies = [
+ "serde_core",
+]
+
+[[package]]
+name = "toml_parser"
+version = "1.0.6+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a3198b4b0a8e11f09dd03e133c0280504d0801269e9afa46362ffde1cbeebf44"
+dependencies = [
+ "winnow",
+]
+
+[[package]]
+name = "toml_writer"
+version = "1.0.6+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ab16f14aed21ee8bfd8ec22513f7287cd4a91aa92e44edfe2c17ddd004e92607"

 [[package]]
 name = "tonic"
@@ -10438,9 +10454,9 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"

 [[package]]
 name = "wasip2"
-version = "1.0.1+wasi-0.2.4"
+version = "1.0.2+wasi-0.2.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0562428422c63773dad2c345a1882263bbf4d65cf3f42e90921f787ef5ad58e7"
+checksum = "9517f9239f02c069db75e65f174b3da828fe5f5b945c4dd26bd25d89c03ebcf5"
 dependencies = [
 "wit-bindgen",
 ]
@@ -10978,15 +10994,12 @@ name = "winnow"
 version = "0.7.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5a5364e9d77fcdeeaa6062ced926ee3381faa2ee02d3eb83a5c27a8825540829"
-dependencies = [
- "memchr",
-]

 [[package]]
 name = "wit-bindgen"
-version = "0.46.0"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f17a85883d4e6d00e8a97c586de764dabcc06133f7f1d55dce5cdc070ad7fe59"
+checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5"

 [[package]]
 name = "wrapcenum-derive"
--- a/crates/e2e_test/Cargo.toml
+++ b/crates/e2e_test/Cargo.toml
@@ -26,6 +26,7 @@ workspace = true
 [dependencies]
 rustfs-ecstore.workspace = true
 rustfs-common.workspace = true
+rustfs-iam.workspace = true
 flatbuffers.workspace = true
 futures.workspace = true
 rustfs-lock.workspace = true
--- a/crates/ecstore/src/erasure_coding/bitrot.rs
+++ b/crates/ecstore/src/erasure_coding/bitrot.rs
@@ -108,7 +108,6 @@ pin_project! {
        inner: W,
        hash_algo: HashAlgorithm,
        shard_size: usize,
-        buf: Vec<u8>,
        finished: bool,
    }
 }
@@ -124,7 +123,6 @@ where
            inner,
            hash_algo,
            shard_size,
-            buf: Vec::new(),
            finished: false,
        }
    }
@@ -159,19 +157,19 @@ where

        if hash_algo.size() > 0 {
            let hash = hash_algo.hash_encode(buf);
-            self.buf.extend_from_slice(hash.as_ref());
+            if hash.as_ref().is_empty() {
+                error!("bitrot writer write hash error: hash is empty");
+                return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "hash is empty"));
+            }
+            self.inner.write_all(hash.as_ref()).await?;
        }

-        self.buf.extend_from_slice(buf);
+        self.inner.write_all(buf).await?;

-        self.inner.write_all(&self.buf).await?;
-
-        // self.inner.flush().await?;
+        self.inner.flush().await?;

        let n = buf.len();

-        self.buf.clear();
-
        Ok(n)
    }

--- a/crates/kms/.gitignore
+++ b/crates/kms/.gitignore
@@ -0,0 +1 @@
+examples/local_data/*
--- a/crates/kms/Cargo.toml
+++ b/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 }
--- a/crates/kms/examples/kms_local_demo.rs
+++ b/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_ok() {
+        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(())
+}
--- a/crates/kms/examples/kms_vault_kv_demo.rs
+++ b/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(())
+}
--- a/crates/kms/src/backends/local.rs
+++ b/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,6 +25,7 @@ use aes_gcm::{
    aead::{Aead, KeyInit},
 };
 use async_trait::async_trait;
+use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
 use rand::Rng;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -36,9 +38,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
@@ -54,24 +58,12 @@ struct StoredMasterKey {
    created_at: chrono::DateTime<chrono::Utc>,
    rotated_at: Option<chrono::DateTime<chrono::Utc>>,
    created_by: Option<String>,
-    /// Encrypted key material (32 bytes for AES-256)
-    encrypted_key_material: Vec<u8>,
+    /// Encrypted key material (32 bytes encoded in base64 for AES-256)
+    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: chrono::DateTime<chrono::Utc>,
-}
-
 impl LocalKmsClient {
    /// Create a new local KMS client
    pub async fn new(config: LocalConfig) -> Result<Self> {
@@ -94,6 +86,7 @@ impl LocalKmsClient {
            config,
            key_cache: RwLock::new(HashMap::new()),
            master_cipher,
+            dek_crypto: AesDekCrypto::new(),
        })
    }

@@ -115,8 +108,8 @@ impl LocalKmsClient {
        self.config.key_dir.join(format!("{key_id}.key"))
    }

-    /// Load a master key from disk
-    async fn load_master_key(&self, key_id: &str) -> Result<MasterKey> {
+    /// Decode and decrypt a stored key file, returning both the metadata and decrypted key material
+    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));
@@ -126,7 +119,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"));
            }
@@ -135,14 +128,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,
@@ -157,7 +165,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
@@ -169,9 +177,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 {
@@ -209,39 +219,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);
-
-        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
-        };
-
+        let (_stored_key, key_material) = self.decode_stored_key(key_id).await?;
        Ok(key_material)
    }

@@ -249,53 +229,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())
-            .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()))
+        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>> {
-        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())
-            .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)
+        self.dek_crypto.decrypt(&key_material, ciphertext, nonce).await
    }
 }

 #[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,
@@ -309,7 +258,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(),
@@ -323,7 +272,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)
@@ -358,15 +307,19 @@ impl KmsClient for LocalKmsClient {
        let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)?;

        // Verify encryption context matches
-        if !request.encryption_context.is_empty() {
-            for (key, expected_value) in &request.encryption_context {
-                if let Some(actual_value) = envelope.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 {
+        // 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}'")));
                }
            }
@@ -381,7 +334,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
@@ -395,13 +348,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?;
@@ -489,7 +442,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
@@ -506,7 +459,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
@@ -528,7 +481,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
@@ -545,7 +498,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
@@ -556,7 +509,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?;
@@ -564,7 +517,7 @@ impl KmsClient for LocalKmsClient {
        master_key.rotated_at = Some(chrono::Utc::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
@@ -624,12 +577,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(),
            );
@@ -793,28 +747,12 @@ impl KmsBackend for LocalKmsBackend {
        };

        // Save the updated key to disk - preserve existing key material!
-        // Load the stored key from disk to get the existing key material
-        let key_path = self.client.master_key_path(key_id);
-        let content = tokio::fs::read(&key_path)
+        // Load and decode the stored key to get the existing key material
+        let (_stored_key, existing_key_material) = self
+            .client
+            .decode_stored_key(key_id)
            .await
-            .map_err(|e| KmsError::internal_error(format!("Failed to read key file: {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
-        };
+            .map_err(|e| KmsError::internal_error(format!("Failed to decode key: {e}")))?;

        self.client.save_master_key(&master_key, &existing_key_material).await?;

@@ -860,8 +798,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();
-        self.client.save_master_key(&master_key, &key_material).await?;
+        // Preserve existing key material instead of generating new one
+        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;
--- a/crates/kms/src/backends/mod.rs
+++ b/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
    ///
--- a/crates/kms/src/backends/vault.rs
+++ b/crates/kms/src/backends/vault.rs
@@ -16,11 +16,11 @@

 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 rand::RngCore;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use tracing::{debug, info, warn};
@@ -37,6 +37,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
@@ -101,6 +103,7 @@ impl VaultKmsClient {
            kv_mount: config.kv_mount.clone(),
            key_path_prefix: config.key_path_prefix.clone(),
            config,
+            dek_crypto: AesDekCrypto::new(),
        })
    }

@@ -109,19 +112,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
@@ -138,6 +128,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);
@@ -153,19 +201,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: chrono::Utc::now(),
-            status: KeyStatus::Active,
-            version: 1,
+            created_at: existing_key_data.created_at,
+            status: existing_key_data.status,
+            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
    }

@@ -224,36 +289,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
-        let _master_key = self.describe_key(&request.master_key_id, context).await?;
-
-        // 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);
+        // Generate random data key material using the existing method
+        let plaintext_key = generate_key_material(&request.key_spec)?;

        // 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 {
-            key_id: request.master_key_id.clone(),
-            version: 1,
-            plaintext: Some(plaintext_key),
-            ciphertext: general_purpose::STANDARD
-                .decode(&encrypted_key)
-                .map_err(|e| KmsError::cryptographic_error("decode", e.to_string()))?,
+        // 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(),
            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: chrono::Utc::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> {
@@ -278,15 +340,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
-        // In practice, you'd extract this from the ciphertext envelope
-        Err(KmsError::invalid_operation("Decrypt not fully implemented for Vault backend"))
+        // Parse the data key envelope from ciphertext
+        let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)
+            .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
@@ -295,7 +384,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
@@ -314,7 +403,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(),
@@ -437,19 +526,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,
--- a/crates/kms/src/encryption/dek.rs
+++ b/crates/kms/src/encryption/dek.rs
@@ -0,0 +1,313 @@
+// 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 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: chrono::DateTime<chrono::Utc>,
+}
+
+/// 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: chrono::Utc::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 that old envelopes without master_key_version can still be deserialized
+        let old_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:00Z"
+        }"#;
+
+        let deserialized: DataKeyEnvelope = serde_json::from_str(old_envelope_json).expect("Should deserialize old format");
+        assert_eq!(deserialized.key_id, "test-key-id");
+        assert_eq!(deserialized.master_key_id, "master-key-id");
+    }
+}
--- a/crates/kms/src/encryption/mod.rs
+++ b/crates/kms/src/encryption/mod.rs
@@ -14,7 +14,7 @@

 //! Object encryption service implementation

-mod ciphers;
-pub mod service;
+pub mod ciphers;
+pub mod dek;

-pub use service::ObjectEncryptionService;
+pub use dek::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
--- a/crates/kms/src/lib.rs
+++ b/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);
+    }
 }
--- a/crates/kms/src/encryption/service.rs
+++ b/crates/kms/src/encryption/service.rs
@@ -273,6 +273,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());

--- a/crates/kms/src/service_manager.rs
+++ b/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 tokio::sync::RwLock;
+use crate::service::ObjectEncryptionService;
+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)
-    manager: Arc<RwLock<Option<Arc<KmsManager>>>>,
-    /// Current encryption service (if running)
-    encryption_service: Arc<RwLock<Option<Arc<ObjectEncryptionService>>>>,
+    /// Current service version (if running)
+    /// Uses ArcSwap for atomic, lock-free service switching
+    /// This allows instant atomic updates without blocking readers
+    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)),
-            encryption_service: Arc::new(RwLock::new(None)),
+            current_service: ArcSwap::from_pointee(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 {
-            Ok(backend) => {
-                // Create KMS manager
-                let kms_manager = Arc::new(KmsManager::new(backend, config));
-
-                // 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);
-                }
+        match self.create_service_version(&config).await {
+            Ok(service_version) => {
+                // Atomically update to new service version (lock-free, instant)
+                // ArcSwap::store() is a true atomic operation using CAS
+                self.current_service.store(Arc::new(Some(service_version)));

                // 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
-        {
-            let mut manager = self.manager.write().await;
-            *manager = None;
-        }
-        {
-            let mut service = self.encryption_service.write().await;
-            *service = None;
-        }
+        // Atomically clear current service version (lock-free, instant)
+        // Note: Existing Arc references will keep the service alive until operations complete
+        self.current_service.store(Arc::new(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
-        if matches!(self.get_status().await, KmsServiceStatus::Running) {
-            self.stop().await?;
-        }
+        info!("Reconfiguring KMS service (zero-downtime)");

        // 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
-        self.start().await?;
+        // Create new service version without stopping old one
+        // 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");
-        Ok(())
+                // Atomically switch to new service version (lock-free, instant CAS operation)
+                // 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
-    async fn create_backend(&self, config: &KmsConfig) -> Result<Arc<dyn KmsBackend>> {
-        match &config.backend_config {
+    /// Create a new service version from configuration
+    ///
+    /// 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,
+        })
    }
 }

--- a/crates/kms/src/types.rs
+++ b/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: DateTime<Utc>,
 }

-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 {
-    fn from(master_key: MasterKey) -> Self {
+impl From<MasterKeyInfo> for KeyInfo {
+    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();
    }
--- a/crates/rio/src/hash_reader.rs
+++ b/crates/rio/src/hash_reader.rs
@@ -169,8 +169,9 @@ impl HashReader {
        sha256hex: Option<String>,
        diskable_md5: bool,
    ) -> std::io::Result<Self> {
-        // Check if it's already a HashReader and update its parameters
-        if let Some(existing_hash_reader) = inner.as_hash_reader_mut() {
+        if size >= 0
+            && let Some(existing_hash_reader) = inner.as_hash_reader_mut()
+        {
            if existing_hash_reader.bytes_read() > 0 {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
@@ -212,7 +213,8 @@ impl HashReader {
            let content_sha256 = existing_hash_reader.content_sha256().clone();
            let content_sha256_hasher = existing_hash_reader.content_sha256().clone().map(|_| Sha256Hasher::new());
            let inner = existing_hash_reader.take_inner();
-            return Ok(Self {
+
+            Ok(Self {
                inner,
                size,
                checksum: md5hex.clone(),
@@ -225,34 +227,36 @@ impl HashReader {
                content_hasher,
                checksum_on_finish: false,
                trailer_s3s: existing_hash_reader.get_trailer().cloned(),
-            });
-        }
+            })
+        } else {
+            if size > 0 {
+                let hr = HardLimitReader::new(inner, size);
+                inner = Box::new(hr);

-        if size > 0 {
-            let hr = HardLimitReader::new(inner, size);
-            inner = Box::new(hr);
-            if !diskable_md5 && !inner.is_hash_reader() {
+                if !diskable_md5 && !inner.is_hash_reader() {
+                    let er = EtagReader::new(inner, md5hex.clone());
+                    inner = Box::new(er);
+                }
+            } else if !diskable_md5 {
                let er = EtagReader::new(inner, md5hex.clone());
                inner = Box::new(er);
            }
-        } else if !diskable_md5 {
-            let er = EtagReader::new(inner, md5hex.clone());
-            inner = Box::new(er);
+
+            Ok(Self {
+                inner,
+                size,
+                checksum: md5hex,
+                actual_size,
+                diskable_md5,
+                bytes_read: 0,
+                content_hash: None,
+                content_hasher: None,
+                content_sha256: sha256hex.clone(),
+                content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
+                checksum_on_finish: false,
+                trailer_s3s: None,
+            })
        }
-        Ok(Self {
-            inner,
-            size,
-            checksum: md5hex,
-            actual_size,
-            diskable_md5,
-            bytes_read: 0,
-            content_hash: None,
-            content_hasher: None,
-            content_sha256: sha256hex.clone(),
-            content_sha256_hasher: sha256hex.clone().map(|_| Sha256Hasher::new()),
-            checksum_on_finish: false,
-            trailer_s3s: None,
-        })
    }

    pub fn into_inner(self) -> Box<dyn Reader> {
--- a/crates/targets/src/arn.rs
+++ b/crates/targets/src/arn.rs
@@ -129,6 +129,8 @@ impl ARN {
    }

    /// Parsing ARN from string
+    /// Only accepts ARNs with the RustFS prefix: "arn:rustfs:sqs:"
+    /// Format: arn:rustfs:sqs:{region}:{id}:{name}
    pub fn parse(s: &str) -> Result<Self, TargetError> {
        if !s.starts_with(ARN_PREFIX) {
            return Err(TargetError::InvalidARN(s.to_string()));
--- a/crates/utils/src/os/windows.rs
+++ b/crates/utils/src/os/windows.rs
@@ -40,7 +40,7 @@ pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
            Some(&mut total_number_of_bytes),
            Some(&mut total_number_of_free_bytes),
        )
-        .map_err(|e| Error::from_raw_os_error(e.code().0 as i32))?;
+        .map_err(|e| Error::from_raw_os_error(e.code().0))?;
    }

    let total = total_number_of_bytes;
@@ -66,7 +66,7 @@ pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
            Some(&mut number_of_free_clusters),
            Some(&mut total_number_of_clusters),
        )
-        .map_err(|e| Error::from_raw_os_error(e.code().0 as i32))?;
+        .map_err(|e| Error::from_raw_os_error(e.code().0))?;
    }

    Ok(DiskInfo {
@@ -94,7 +94,7 @@ fn get_volume_name(v: &[u16]) -> std::io::Result<Vec<u16>> {

    unsafe {
        GetVolumePathNameW(windows::core::PCWSTR::from_raw(v.as_ptr()), &mut volume_name_buffer)
-            .map_err(|e| Error::from_raw_os_error(e.code().0 as i32))?;
+            .map_err(|e| Error::from_raw_os_error(e.code().0))?;
    }

    let len = volume_name_buffer
@@ -137,7 +137,7 @@ fn get_fs_type(p: &[u16]) -> std::io::Result<String> {
            Some(&mut file_system_flags),
            Some(&mut file_system_name_buffer),
        )
-        .map_err(|e| Error::from_raw_os_error(e.code().0 as i32))?;
+        .map_err(|e| Error::from_raw_os_error(e.code().0))?;
    }

    Ok(utf16_to_string(&file_system_name_buffer))
--- a/rustfs/Cargo.toml
+++ b/rustfs/Cargo.toml
@@ -95,6 +95,7 @@ serde_urlencoded = { workspace = true }
 rustls = { workspace = true }
 subtle = { workspace = true }
 rustls-pemfile = { workspace = true }
+aes-gcm = { workspace = true }

 # Time and Date
 chrono = { workspace = true }
@@ -126,6 +127,7 @@ urlencoding = { workspace = true }
 uuid = { workspace = true }
 zip = { workspace = true }
 libc = { workspace = true }
+rand = { workspace = true }

 # Observability and Metrics
 metrics = { workspace = true }
--- a/rustfs/src/config/mod.rs
+++ b/rustfs/src/config/mod.rs
@@ -138,6 +138,10 @@ pub struct Opt {
    #[arg(long, env = "RUSTFS_KMS_KEY_DIR")]
    pub kms_key_dir: Option<String>,

+    /// KMS local master key for local backend (optional)
+    #[arg(long, env = "RUSTFS_KMS_LOCAL_MASTER_KEY")]
+    pub kms_local_master_key: Option<String>,
+
    /// Vault address for vault backend
    #[arg(long, env = "RUSTFS_KMS_VAULT_ADDRESS")]
    pub kms_vault_address: Option<String>,
@@ -159,6 +163,47 @@ pub struct Opt {
    /// Options: GeneralPurpose, AiTraining, DataAnalytics, WebWorkload, IndustrialIoT, SecureStorage
    #[arg(long, default_value_t = String::from("GeneralPurpose"), env = "RUSTFS_BUFFER_PROFILE")]
    pub buffer_profile: String,
+
+    /// Enable FTPS server
+    #[arg(long, default_value_t = false, env = "RUSTFS_FTPS_ENABLE")]
+    pub ftps_enable: bool,
+
+    /// FTPS server bind address
+    #[arg(long, default_value_t = String::from("0.0.0.0:21"), env = "RUSTFS_FTPS_ADDRESS")]
+    pub ftps_address: String,
+
+    /// FTPS server certificate file path
+    #[arg(long, env = "RUSTFS_FTPS_CERTS_FILE")]
+    pub ftps_certs_file: Option<String>,
+
+    /// FTPS server private key file path
+    #[arg(long, env = "RUSTFS_FTPS_KEY_FILE")]
+    pub ftps_key_file: Option<String>,
+
+    /// FTPS server passive ports range (e.g., "40000-50000")
+    #[arg(long, env = "RUSTFS_FTPS_PASSIVE_PORTS")]
+    pub ftps_passive_ports: Option<String>,
+
+    /// FTPS server external IP address for passive mode (auto-detected if not specified)
+    #[arg(long, env = "RUSTFS_FTPS_EXTERNAL_IP")]
+    pub ftps_external_ip: Option<String>,
+
+    /// Enable SFTP server
+    #[arg(long, default_value_t = false, env = "RUSTFS_SFTP_ENABLE")]
+    pub sftp_enable: bool,
+
+    /// SFTP server bind address
+    #[arg(long, default_value_t = String::from("0.0.0.0:22"), env = "RUSTFS_SFTP_ADDRESS")]
+    pub sftp_address: String,
+
+    /// SFTP server host key file path
+    #[arg(long, env = "RUSTFS_SFTP_HOST_KEY")]
+    pub sftp_host_key: Option<String>,
+
+    /// Path to authorized SSH public keys file for SFTP authentication
+    /// Each line should contain an OpenSSH public key: ssh-rsa AAAA... comment
+    #[arg(long, env = "RUSTFS_SFTP_AUTHORIZED_KEYS")]
+    pub sftp_authorized_keys: Option<String>,
 }

 impl std::fmt::Debug for Opt {
--- a/rustfs/src/init.rs
+++ b/rustfs/src/init.rs
@@ -116,21 +116,29 @@ pub(crate) async fn add_bucket_notification_configuration(buckets: Vec<String>)
                    "Bucket '{}' has existing notification configuration: {:?}", bucket, cfg);

                let mut event_rules = Vec::new();
-                process_queue_configurations(&mut event_rules, cfg.queue_configurations.clone(), |arn_str| {
+                if let Err(e) = process_queue_configurations(&mut event_rules, cfg.queue_configurations.clone(), |arn_str| {
                    ARN::parse(arn_str)
                        .map(|arn| arn.target_id)
                        .map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
-                });
-                process_topic_configurations(&mut event_rules, cfg.topic_configurations.clone(), |arn_str| {
+                }) {
+                    error!("Failed to parse queue notification config for bucket '{}': {:?}", bucket, e);
+                }
+                if let Err(e) = process_topic_configurations(&mut event_rules, cfg.topic_configurations.clone(), |arn_str| {
                    ARN::parse(arn_str)
                        .map(|arn| arn.target_id)
                        .map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
-                });
-                process_lambda_configurations(&mut event_rules, cfg.lambda_function_configurations.clone(), |arn_str| {
-                    ARN::parse(arn_str)
-                        .map(|arn| arn.target_id)
-                        .map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
-                });
+                }) {
+                    error!("Failed to parse topic notification config for bucket '{}': {:?}", bucket, e);
+                }
+                if let Err(e) =
+                    process_lambda_configurations(&mut event_rules, cfg.lambda_function_configurations.clone(), |arn_str| {
+                        ARN::parse(arn_str)
+                            .map(|arn| arn.target_id)
+                            .map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
+                    })
+                {
+                    error!("Failed to parse lambda notification config for bucket '{}': {:?}", bucket, e);
+                }

                if let Err(e) = notifier_global::add_event_specific_rules(bucket, region, &event_rules)
                    .await
@@ -176,11 +184,17 @@ pub(crate) async fn init_kms_system(opt: &config::Opt) -> std::io::Result<()> {
                    .as_ref()
                    .ok_or_else(|| Error::other("KMS key directory is required for local backend"))?;

+                // root key for local backend(optional)
+                let master_key = opt
+                    .kms_local_master_key
+                    .as_ref()
+                    .map_or(None, |k| Some(k.to_string()));
+
                rustfs_kms::config::KmsConfig {
                    backend: rustfs_kms::config::KmsBackend::Local,
                    backend_config: rustfs_kms::config::BackendConfig::Local(rustfs_kms::config::LocalConfig {
                        key_dir: std::path::PathBuf::from(key_dir),
-                        master_key: None,
+                        master_key: master_key,
                        file_permissions: Some(0o600),
                    }),
                    default_key_id: opt.kms_default_key_id.clone(),
--- a/rustfs/src/storage/ecfs.rs
+++ b/rustfs/src/storage/ecfs.rs
--- a/rustfs/src/storage/mod.rs
+++ b/rustfs/src/storage/mod.rs
@@ -18,6 +18,7 @@ pub mod ecfs;
 pub(crate) mod entity;
 pub(crate) mod helper;
 pub mod options;
+pub mod sse;
 pub mod tonic_service;

 #[cfg(test)]
--- a/rustfs/src/storage/sse.rs
+++ b/rustfs/src/storage/sse.rs
--- a/scripts/decrypt_local_kms_key.py
+++ b/scripts/decrypt_local_kms_key.py
@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+"""
+Decrypt RustFS Local KMS Key File
+
+This script decrypts a local KMS key file when you know the master key.
+It replicates the key derivation and decryption logic from the Rust implementation.
+
+Usage:
+    python decrypt_local_kms_key.py <key_file_path> <master_key>
+
+Example:
+    python decrypt_local_kms_key.py target/kms-key-dir/test-key.key "my-master-key"
+
+Requirements:
+    pip install cryptography
+"""
+
+import sys
+import json
+import base64
+import hashlib
+from pathlib import Path
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+
+
+def derive_master_key(master_key_str: str) -> bytes:
+    """
+    Derive a 256-bit AES key from the master key string.
+    
+    This replicates the Rust implementation:
+    - SHA-256 hash of: master_key + salt("rustfs-kms-local")
+    - Returns 32 bytes for AES-256
+    
+    Args:
+        master_key_str: The master key string
+        
+    Returns:
+        32-byte key for AES-256-GCM
+    """
+    hasher = hashlib.sha256()
+    hasher.update(master_key_str.encode('utf-8'))
+    hasher.update(b'rustfs-kms-local')  # Salt to prevent rainbow tables
+    return hasher.digest()  # Returns 32 bytes
+
+
+def decrypt_key_file(key_file_path: str, master_key: str) -> dict:
+    """
+    Decrypt a local KMS key file.
+    
+    Args:
+        key_file_path: Path to the .key file
+        master_key: The master key string used for encryption
+        
+    Returns:
+        Dictionary containing the decrypted key material and metadata
+        
+    Raises:
+        FileNotFoundError: If key file doesn't exist
+        json.JSONDecodeError: If key file is not valid JSON
+        ValueError: If decryption fails or file format is invalid
+    """
+    # Read the key file
+    key_path = Path(key_file_path)
+    if not key_path.exists():
+        raise FileNotFoundError(f"Key file not found: {key_file_path}")
+    
+    with open(key_path, 'r', encoding='utf-8') as f:
+        stored_key = json.load(f)
+    
+    # Validate required fields
+    required_fields = ['encrypted_key_material', 'nonce', 'key_id', 'algorithm']
+    missing_fields = [field for field in required_fields if field not in stored_key]
+    if missing_fields:
+        raise ValueError(f"Missing required fields: {missing_fields}")
+    
+    # Extract encrypted data
+    encrypted_key_material_b64 = stored_key['encrypted_key_material']
+    nonce = bytes(stored_key['nonce'])
+    
+    # Validate nonce length (must be 12 bytes for AES-GCM)
+    if len(nonce) != 12:
+        raise ValueError(f"Invalid nonce length: {len(nonce)}, expected 12 bytes")
+    
+    # Derive the AES key from master key
+    aes_key = derive_master_key(master_key)
+    
+    # Decode the base64 encrypted key material
+    try:
+        encrypted_bytes = base64.b64decode(encrypted_key_material_b64)
+    except Exception as e:
+        raise ValueError(f"Failed to decode base64 encrypted key material: {e}")
+    
+    # Decrypt using AES-256-GCM
+    try:
+        aesgcm = AESGCM(aes_key)
+        plaintext_key_material = aesgcm.decrypt(nonce, encrypted_bytes, None)
+    except Exception as e:
+        raise ValueError(f"Decryption failed. Wrong master key or corrupted file: {e}")
+    
+    # Return decrypted data with metadata
+    return {
+        'key_id': stored_key['key_id'],
+        'algorithm': stored_key['algorithm'],
+        'version': stored_key.get('version', 1),
+        'status': stored_key.get('status'),
+        'usage': stored_key.get('usage'),
+        'created_at': stored_key.get('created_at'),
+        'created_by': stored_key.get('created_by'),
+        'description': stored_key.get('description'),
+        'key_material_hex': plaintext_key_material.hex(),
+        'key_material_base64': base64.b64encode(plaintext_key_material).decode('ascii'),
+        'key_material_length': len(plaintext_key_material),
+    }
+
+
+def main():
+    if len(sys.argv) != 3:
+        print("Usage: python decrypt_local_kms_key.py <key_file_path> <master_key>")
+        print()
+        print("Example:")
+        print('  python decrypt_local_kms_key.py target/kms-key-dir/test-key.key "my-master-key"')
+        sys.exit(1)
+    
+    key_file_path = sys.argv[1]
+    master_key = sys.argv[2]
+    
+    try:
+        result = decrypt_key_file(key_file_path, master_key)
+        
+        print("=" * 70)
+        print("Successfully Decrypted Key File")
+        print("=" * 70)
+        print(f"Key ID:          {result['key_id']}")
+        print(f"Algorithm:       {result['algorithm']}")
+        print(f"Version:         {result['version']}")
+        print(f"Status:          {result['status']}")
+        print(f"Usage:           {result['usage']}")
+        print(f"Created At:      {result['created_at']}")
+        print(f"Created By:      {result['created_by']}")
+        print(f"Description:     {result['description']}")
+        print(f"Key Length:      {result['key_material_length']} bytes")
+        print()
+        print("Decrypted Key Material:")
+        print(f"  Hex:           {result['key_material_hex']}")
+        print(f"  Base64:        {result['key_material_base64']}")
+        print("=" * 70)
+        
+        # Security warning
+        print()
+        print("⚠️  WARNING: The decrypted key material above is sensitive!")
+        print("    Do not share it or store it in plain text.")
+        
+    except FileNotFoundError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+    except json.JSONDecodeError as e:
+        print(f"Error: Invalid JSON format in key file: {e}", file=sys.stderr)
+        sys.exit(1)
+    except ValueError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+    except Exception as e:
+        print(f"Unexpected error: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+if __name__ == '__main__':
+    main()
+
--- a/scripts/decrypt_local_kms_key_README.md
+++ b/scripts/decrypt_local_kms_key_README.md
@@ -0,0 +1,196 @@
+# Local KMS Key Decryption Tool
+
+This directory contains Python scripts to decrypt RustFS Local KMS key files when you know the master key.
+
+## Files
+
+- `decrypt_local_kms_key.py` - Main decryption script
+- `decrypt_local_kms_key_test.py` - Test suite and examples
+- `decrypt_local_kms_key_README.md` - This file
+
+## Requirements
+
+Install the required Python package:
+
+```bash
+pip install cryptography
+```
+
+## Usage
+
+### Basic Usage
+
+```bash
+python scripts/decrypt_local_kms_key.py <key_file_path> <master_key>
+```
+
+### Example
+
+```bash
+# Decrypt a key file with master key "my-secret-key"
+python scripts/decrypt_local_kms_key.py target/kms-key-dir/test-key.key "my-secret-key"
+```
+
+use uv
+
+```
+uv run --with cryptography python scripts\decrypt_local_kms_key.py .\target\kms-key-dir\rustfs-master-key.key "my-secret-key"
+```
+
+### Output
+
+The script will output:
+- Key metadata (ID, algorithm, version, status, etc.)
+- Decrypted key material in both hex and base64 formats
+- Key length in bytes
+
+Example output:
+```
+======================================================================
+Successfully Decrypted Key File
+======================================================================
+Key ID:          test-key-001
+Algorithm:       AES_256
+Version:         1
+Status:          Active
+Usage:           EncryptDecrypt
+Created At:      2024-01-17T10:00:00Z
+Created By:      local-kms
+Description:     Test key for encryption
+Key Length:      32 bytes
+
+Decrypted Key Material:
+  Hex:           a1b2c3d4e5f6...
+  Base64:        obPD1OX2...
+======================================================================
+
+⚠️  WARNING: The decrypted key material above is sensitive!
+    Do not share it or store it in plain text.
+```
+
+## Testing
+
+Run the test suite to verify the implementation:
+
+```bash
+python scripts/decrypt_local_kms_key_test.py
+```
+
+This will:
+1. Test key derivation logic
+2. Test encryption/decryption cycle
+3. Show example key file format
+
+## How It Works
+
+### 1. Key Derivation
+
+The script derives a 256-bit AES key from the master key string using:
+
+```python
+SHA256(master_key_string + "rustfs-kms-local")
+```
+
+This matches the Rust implementation in `crates/kms/src/backends/local.rs`:
+
+```rust
+let mut hasher = Sha256::new();
+hasher.update(master_key.as_bytes());
+hasher.update(b"rustfs-kms-local"); // Salt
+let hash = hasher.finalize();
+```
+
+### 2. Decryption Process
+
+1. Read the `.key` file (JSON format)
+2. Extract `encrypted_key_material` (base64 string) and `nonce` (12 bytes)
+3. Decode the base64 encrypted data
+4. Decrypt using AES-256-GCM with the derived key and nonce
+5. Return the plaintext key material
+
+### 3. Key File Format
+
+The `.key` files are JSON with the following structure:
+
+```json
+{
+  "key_id": "test-key",
+  "version": 1,
+  "algorithm": "AES_256",
+  "usage": "EncryptDecrypt",
+  "status": "Active",
+  "description": "Description",
+  "metadata": {},
+  "created_at": "2024-01-17T10:00:00Z",
+  "rotated_at": null,
+  "created_by": "local-kms",
+  "encrypted_key_material": "base64_encoded_encrypted_bytes",
+  "nonce": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]
+}
+```
+
+## Security Notes
+
+⚠️ **Important Security Considerations:**
+
+1. **Protect the Master Key**: The master key is the root of trust. Anyone with the master key can decrypt all keys.
+
+2. **Sensitive Output**: The decrypted key material is highly sensitive. Do not:
+   - Store it in plain text
+   - Share it over insecure channels
+   - Log it to files
+   - Commit it to version control
+
+3. **Use Cases**: This script is intended for:
+   - Debugging and troubleshooting
+   - Key recovery in emergencies
+   - Migration to other KMS systems
+   - Forensic analysis
+
+4. **Production Use**: In production, keys should only be decrypted by the RustFS KMS service itself.
+
+## Troubleshooting
+
+### "Decryption failed. Wrong master key or corrupted file"
+
+This error occurs when:
+- The master key is incorrect
+- The key file is corrupted
+- The key file format doesn't match expectations
+
+Verify that you're using the same master key that was configured when the key was created.
+
+### "Invalid nonce length"
+
+The nonce must be exactly 12 bytes for AES-GCM. If you see this error, the key file may be corrupted.
+
+### "Missing required fields"
+
+The key file must contain:
+- `encrypted_key_material`
+- `nonce`
+- `key_id`
+- `algorithm`
+
+Check that the file is a valid RustFS Local KMS key file.
+
+## Related Configuration
+
+The master key is configured in RustFS via:
+
+- Environment variable: `RUSTFS_KMS_LOCAL_MASTER_KEY`
+- Config file: `kms.backend_config.master_key`
+
+Example `.env`:
+```bash
+RUSTFS_KMS_BACKEND=local
+RUSTFS_KMS_LOCAL_KEY_DIR=/path/to/key/dir
+RUSTFS_KMS_LOCAL_MASTER_KEY=your-master-key-here
+```
+
+## License
+
+Copyright 2024 RustFS Team
+
+Licensed under the Apache License, Version 2.0.
+
--- a/scripts/generate-sse-keys.ps1
+++ b/scripts/generate-sse-keys.ps1
@@ -0,0 +1,28 @@
+# Generate SSE encryption key (32 bytes base64 encoded)
+# For testing with __RUSTFS_SSE_SIMPLE_CMK environment variable
+# Usage: .\generate-sse-keys.ps1
+
+# Function to generate a random 256-bit key and encode it in base64
+function Generate-Base64Key {
+    # Generate 32 bytes (256 bits) of random data
+    $bytes = New-Object byte[] 32
+    $rng = [System.Security.Cryptography.RandomNumberGenerator]::Create()
+    $rng.GetBytes($bytes)
+    $rng.Dispose()
+    
+    # Convert to base64
+    return [Convert]::ToBase64String($bytes)
+}
+
+# Generate key
+$base64Key = Generate-Base64Key
+
+# Output result
+Write-Host ""
+Write-Host "Generated SSE encryption key (32 bytes, base64 encoded):" -ForegroundColor Green
+Write-Host ""
+Write-Host $base64Key -ForegroundColor Yellow
+Write-Host ""
+Write-Host "You can use this in your environment variable:" -ForegroundColor Cyan
+Write-Host "`$env:__RUSTFS_SSE_SIMPLE_CMK=`"$base64Key`"" -ForegroundColor White
+Write-Host ""
--- a/scripts/generate-sse-keys.sh
+++ b/scripts/generate-sse-keys.sh
@@ -0,0 +1,24 @@
+#!/bin/bash
+# Generate SSE encryption key (32 bytes base64 encoded)
+# For testing with __RUSTFS_SSE_SIMPLE_CMK environment variable
+# Usage: ./generate-sse-keys.sh
+
+set -euo pipefail
+
+# Function to generate a random 256-bit key and encode it in base64
+generate_base64_key() {
+    # Generate 32 bytes (256 bits) of random data and base64 encode it
+    openssl rand -base64 32 | tr -d '\n'
+}
+
+# Generate key
+base64_key=$(generate_base64_key)
+
+# Output result
+echo "Generated SSE encryption key (32 bytes, base64 encoded):"
+echo ""
+echo "$base64_key"
+echo ""
+echo "You can use this in your environment variable:"
+echo "export __RUSTFS_SSE_SIMPLE_CMK=\"$base64_key\""
+
Author	SHA1	Message	Date
reatang	31ac6c20a5	fix	2026-01-17 16:42:09 +08:00
reatang	6ae884c571	Refactor KMS and SSE integration by introducing a local master key option and enhancing key management. This update includes changes to the KMS configuration, updates to the local backend for key handling, and improvements in the SSE encryption and decryption processes, ensuring better clarity and functionality.	2026-01-17 16:19:58 +08:00
reatang	24629a12fc	Enhance SSE decryption process by consolidating encryption metadata handling and improving key validation. This update introduces clearer management of server-side encryption parameters and refines the decryption logic for better performance and security.	2026-01-17 12:32:50 +08:00
reatang	c24728d044	Refactor and unify SSE encryption and decryption processes	2026-01-17 01:48:37 +08:00
reatang	a26ddf3aaa	Resolving the issue of encryption parameters not being stored.	2026-01-17 01:15:20 +08:00
reatang	190ebb2832	chore: update dependencies in Cargo.lock and Cargo.toml	2026-01-16 23:40:23 +08:00
reatang	8588188cac	Refactor the relationship between SSE and KMS, and decouple them through interfaces.	2026-01-16 23:32:39 +08:00
reatang	d00ce55047	Refactor the SSE decoding code to the SSE layer	2026-01-16 23:32:39 +08:00
reatang	680a017759	Refactoring the SSE layer encryption function	2026-01-16 23:32:39 +08:00
reatang	6bbf8b8650	The APIs exported by the SSE module have been refactored.	2026-01-16 23:32:39 +08:00
reatang	d6b9f9557f	feat: implement unified encryption and decryption API This update introduces a unified API for encryption and decryption in the SSE module, consolidating the previous methods into two core functions: `apply_encryption()` and `apply_decryption()`. The new API simplifies the process of applying server-side encryption (SSE-S3, SSE-KMS, and SSE-C) and enhances code readability. Additionally, detailed documentation and examples have been added to guide usage.	2026-01-16 23:32:29 +08:00
reatang	b4c436ffe0	Completely extract the SSE layer from the business logic.	2026-01-16 23:32:29 +08:00
Andreas Nussberger	09a90058ff	helm: add nodeSelector to standalone deployment (#1367 ) Co-authored-by: majinghe <42570491+majinghe@users.noreply.github.com>	2026-01-16 23:32:06 +08:00
houseme	ef24e2b886	chore: upgrade dependencies and migrate to aws-lc-rs (#1333 )	2026-01-16 23:32:06 +08:00
yxrxy	c63fcebfe1	Feat/ftps&sftp (#1308 ) [feat] ftp / sftp	2026-01-16 23:31:07 +08:00
reatang	81accd0bff	Optimize structure names to prevent conflicts	2026-01-16 23:22:43 +08:00
reatang	ec3b75bbb6	Remove useless parameters	2026-01-16 23:22:43 +08:00
reatang	6d3bdc0b3e	feat: Implement zero-downtime reconfiguration for KMS service - Added support for versioned service management in KmsServiceManager, allowing seamless reconfiguration without interrupting ongoing operations. - Introduced ArcSwap for atomic service switching, ensuring instant updates without blocking. - Enhanced service lifecycle management with mutex protection for concurrent operations. - Updated dependencies in Cargo.toml and Cargo.lock to include arc-swap. - Refactored encryption service handling, moving to a new module structure for better organization. This change significantly improves the KMS service's reliability and performance during configuration changes.	2026-01-16 23:22:43 +08:00
LeonWang0735	2ab6f8c029	fix:correctly handle compress object when put object (#1534 )	2026-01-16 23:11:48 +08:00
weisd	0927f937a7	fix: Fix BitrotWriter encode writer implementation (#1531 )	2026-01-16 17:11:54 +08:00
Audric	548a39ffe7	fix: return error instead of silently ignoring invalid ARNs in notification config (#1528 ) Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>	2026-01-16 16:12:55 +08:00