sdgoij/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-01-16 17:20:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Implement zero-downtime reconfiguration for KMS service

Browse Source 
- 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.
This commit is contained in:
reatang
2025-12-31 00:08:09 +08:00
parent 2ab6f8c029
commit 6d3bdc0b3e
12 changed files with 1304 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -7942,6 +7942,7 @@ name = "rustfs-kms"
version = "0.0.5"
dependencies = [
"aes-gcm 0.11.0-rc.2",
"arc-swap",
"async-trait",
"base64",
"chacha20poly1305",

1
crates/kms/.gitignore vendored Normal file
View File

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

1
crates/kms/Cargo.toml
View File

@@ -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 }

243
crates/kms/examples/demo1.rs Normal file
View File

@@ -0,0 +1,243 @@
// 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 std::fs;
use rustfs_kms::{
init_global_kms_service_manager, CreateKeyRequest, DescribeKeyRequest, EncryptionAlgorithm,
GenerateDataKeyRequest, KmsConfig, KeySpec, KeyUsage, ListKeysRequest,
};
use std::collections::HashMap;
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(())
}

287
crates/kms/examples/demo2.rs Normal file
View File

@@ -0,0 +1,287 @@
// 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::{
init_global_kms_service_manager, CreateKeyRequest, DescribeKeyRequest, EncryptionAlgorithm,
GenerateDataKeyRequest, KmsConfig, KmsError, KeySpec, KeyUsage, ListKeysRequest,
};
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(())
}

125
crates/kms/src/backends/local.rs
View File

@@ -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::{
@@ -39,6 +40,8 @@ pub struct LocalKmsClient {
key_cache: RwLock<HashMap<String, MasterKey>>,
/// 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
@@ -60,17 +63,6 @@ struct StoredMasterKey {
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
@@ -94,6 +86,7 @@ impl LocalKmsClient {
config,
key_cache: RwLock::new(HashMap::new()),
master_cipher,
dek_crypto: AesDekCrypto::new(),
})
}
@@ -209,12 +202,6 @@ 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>> {
@@ -249,42 +236,14 @@ 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
}
}
@@ -293,8 +252,8 @@ impl KmsClient for LocalKmsClient {
async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKey> {
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?;
// Verify master key exists and get its version
let master_key_info = self.describe_key(&request.master_key_id, context).await?;
// Generate random data key material
let key_length = match request.key_spec.as_str() {
@@ -309,10 +268,11 @@ 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(),
master_key_version: master_key_info.version,
key_spec: request.key_spec.clone(),
encrypted_key: encrypted_key.clone(),
nonce,
@@ -358,15 +318,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}'")));
}
}
@@ -395,7 +359,7 @@ 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())
@@ -489,7 +453,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 +470,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 +492,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 +509,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
@@ -564,7 +528,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 +588,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(
key_id.clone(),
"AES_256".to_string(),
algorithm.to_string(),
Some("local-kms".to_string()),
request.description.clone(),
);
@@ -860,8 +825,30 @@ 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 key_path = self.client.master_key_path(key_id);
let content = tokio::fs::read(&key_path)
.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
};
self.client.save_master_key(&master_key, &existing_key_material).await?;
// Update cache
let mut cache = self.client.key_cache.write().await;

184
crates/kms/src/backends/vault.rs
View File

@@ -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,8 +37,11 @@ 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
#[derive(Debug, Clone, Serialize, Deserialize)]
struct VaultKeyData {
@@ -101,6 +104,7 @@ impl VaultKmsClient {
kv_mount: config.kv_mount.clone(),
key_path_prefix: config.key_path_prefix.clone(),
config,
dek_crypto: AesDekCrypto::new(),
})
}
@@ -109,18 +113,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> {
@@ -138,6 +130,61 @@ 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 +200,34 @@ 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
}
@@ -227,33 +289,34 @@ impl KmsClient for VaultKmsClient {
async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKey> {
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?;
// Verify master key exists and get its version
let master_key_info = 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(),
master_key_version: master_key_info.version,
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 = DataKey::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,12 +341,39 @@ 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> {
@@ -295,7 +385,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
@@ -444,7 +534,7 @@ impl KmsClient for VaultKmsClient {
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?;

328
crates/kms/src/encryption/dek.rs Normal file
View File

@@ -0,0 +1,328 @@
// 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,
/// Version of the master key (KEK) used to encrypt this DEK
/// This is critical for key rotation: when a KEK is rotated, we need to know
/// which version was used to encrypt each DEK so we can use the correct KEK version for decryption.
#[serde(default = "default_master_key_version")]
pub master_key_version: u32,
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>,
}
fn default_master_key_version() -> u32 {
1
}
/// 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(),
master_key_version: 1,
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.master_key_version, envelope.master_key_version);
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");
assert_eq!(deserialized.master_key_version, 1); // Should default to 1
}
}

6
crates/kms/src/encryption/mod.rs
View File

@@ -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};

89
crates/kms/src/lib.rs
View File

@@ -62,6 +62,7 @@ mod cache;
pub mod config;
mod encryption;
mod error;
pub mod service;
pub mod manager;
pub mod service_manager;
pub mod types;
@@ -73,8 +74,7 @@ pub use api_types::{
UntagKeyRequest, UntagKeyResponse, UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse,
};
pub use config::*;
pub use encryption::ObjectEncryptionService;
pub use encryption::service::DataKey;
pub use service::{DataKey, ObjectEncryptionService};
pub use error::{KmsError, Result};
pub use manager::KmsManager;
pub use 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,88 @@ 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);
}
}

2
crates/kms/src/encryption/service.rs → crates/kms/src/service.rs
View File

@@ -273,7 +273,7 @@ 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".to_string(), object_key.to_string());
context.insert("object_key".to_string(), object_key.to_string());
context.insert("algorithm".to_string(), algorithm.as_str().to_string());
// Auto-create key for SSE-S3 if it doesn't exist

219
crates/kms/src/service_manager.rs
View File

@@ -16,11 +16,12 @@
use crate::backends::{KmsBackend, local::LocalKmsBackend};
use crate::config::{BackendConfig, KmsConfig};
use crate::encryption::service::ObjectEncryptionService;
use crate::service::ObjectEncryptionService;
use crate::error::{KmsError, Result};
use crate::manager::KmsManager;
use std::sync::{Arc, OnceLock};
use tokio::sync::RwLock;
use arc_swap::ArcSwap;
use std::sync::{Arc, OnceLock, atomic::{AtomicU64, Ordering}};
use tokio::sync::{RwLock, Mutex};
use tracing::{error, info, warn};
/// KMS service status
@@ -36,26 +37,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 +107,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 +129,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 +155,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 +179,101 @@ 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");
// Stop current service if running
if matches!(self.get_status().await, KmsServiceStatus::Running) {
self.stop().await?;
}
let _guard = self.lifecycle_mutex.lock().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()
.and_then(|sv| Some(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 +305,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,
})
}
}