feat(kms): add master key version to data key envelope contract (#5480)

* fix(kms): restore vault backend test compilation after timeout parameter

PR #5472 added an attempt_timeout parameter to VaultKmsClient::new while
PR #5474 landed tests still using the one-argument form, leaving
'cargo test -p rustfs-kms' unable to compile on main. Pass the same
30-second timeout the surrounding integration tests already use.

* feat(kms): add master key version to data key envelope contract

DataKeyEnvelope gains an optional master_key_version field recording
which KEK version wrapped the DEK, so rotation-aware backends can load
the matching historical material on decrypt. The field is skipped when
None, keeping envelopes from non-rotating backends byte-identical to
the historical seven-field JSON shape, and legacy envelopes without the
field deserialize to None. The envelope discriminator marker is
untouched, so mixed-format routing is unchanged in both directions.

Adds the KeyVersionNotFound typed error for version-addressed material
lookups that must fail closed instead of falling back to the current
version.

Refs rustfs/backlog#1565

crates/kms/src/backends/local.rs

@@ -942,7 +942,8 @@ 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 with master key version for rotation support
+        // Local rotation is rejected, so every envelope is wrapped by the key's sole
+        // material and needs no master key version.
         let envelope = DataKeyEnvelope {
             key_id: uuid::Uuid::new_v4().to_string(),
             master_key_id: request.master_key_id.clone(),
@@ -951,6 +952,7 @@ impl KmsClient for LocalKmsClient {
             nonce,
             encryption_context: request.encryption_context.clone(),
             created_at: Zoned::now(),
+            master_key_version: None,
         };
 
         // Serialize the envelope as the ciphertext

crates/kms/src/backends/static_kms.rs

@@ -137,6 +137,8 @@ impl KmsClient for StaticKmsBackend {
             nonce: nonce_bytes.to_vec(),
             encryption_context: request.encryption_context.clone(),
             created_at: Zoned::now(),
+            // The static backend has a single fixed key with no rotation.
+            master_key_version: None,
         };
         let ciphertext = serde_json::to_vec(&envelope)?;
 
@@ -181,6 +183,8 @@ impl KmsClient for StaticKmsBackend {
             nonce: nonce_bytes.to_vec(),
             encryption_context: request.encryption_context.clone(),
             created_at: Zoned::now(),
+            // The static backend has a single fixed key with no rotation.
+            master_key_version: None,
         };
         let ciphertext = serde_json::to_vec(&envelope)?;
 

crates/kms/src/backends/vault.rs

@@ -311,6 +311,7 @@ impl KmsClient for VaultKmsClient {
             nonce,
             encryption_context: request.encryption_context.clone(),
             created_at: Zoned::now(),
+            master_key_version: None,
         };
 
         // Serialize the envelope as the ciphertext
@@ -938,7 +939,9 @@ mod tests {
     async fn test_vault_kv2_rotate_key_rejected_without_touching_storage() {
         // No Vault instance needed: rotation must be rejected before any storage access,
         // so the call cannot read or overwrite key material.
-        let client = VaultKmsClient::new(integration_vault_config()).await.expect("client");
+        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+            .await
+            .expect("client");
 
         let err = client
             .rotate_key("any-key", None)
@@ -950,7 +953,9 @@ mod tests {
 
     #[tokio::test]
     async fn test_vault_kv2_backend_info_reports_at_rest_protection() {
-        let client = VaultKmsClient::new(integration_vault_config()).await.expect("client");
+        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+            .await
+            .expect("client");
 
         let info = client.backend_info();
         assert_eq!(info.backend_type, "vault-kv2");
@@ -962,7 +967,9 @@ mod tests {
     #[tokio::test]
     #[ignore] // Requires a running Vault instance (dev mode)
     async fn test_vault_kv2_rotate_rejected_and_material_untouched() {
-        let client = VaultKmsClient::new(integration_vault_config()).await.expect("client");
+        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+            .await
+            .expect("client");
 
         let key_id = format!("rotate-{}", uuid::Uuid::new_v4());
         client.create_key(&key_id, "AES_256", None).await.expect("create");

crates/kms/src/backends/vault_transit.rs

@@ -406,6 +406,9 @@ impl KmsClient for VaultTransitKmsClient {
             nonce: Vec::new(),
             encryption_context: request.encryption_context.clone(),
             created_at: Zoned::now(),
+            // Transit ciphertext already self-describes its key version
+            // ("vault:vN:..."), so the envelope never carries one.
+            master_key_version: None,
         };
 
         let ciphertext = serde_json::to_vec(&envelope)?;

crates/kms/src/encryption/dek.rs

@@ -32,7 +32,10 @@ use std::collections::HashMap;
 ///
 /// 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.
+/// wrapped this DEK so rotation-aware backends can load the matching historical
+/// material. Envelopes written before versioning carry `None`; backends must resolve
+/// `None` to a deterministic baseline version recorded in key metadata, never
+/// implicitly to whatever version is current.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DataKeyEnvelope {
     pub key_id: String,
@@ -43,6 +46,12 @@ pub struct DataKeyEnvelope {
     pub encryption_context: HashMap<String, String>,
     #[serde(with = "crate::time_serde::zoned")]
     pub created_at: Zoned,
+    /// KEK version that wrapped `encrypted_key`; `None` on pre-versioning envelopes.
+    ///
+    /// Optional and omitted when `None` so envelopes from non-rotating backends stay
+    /// byte-identical to the historical seven-field JSON shape.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub master_key_version: Option<u32>,
 }
 
 #[derive(Deserialize)]
@@ -307,6 +316,7 @@ mod tests {
                 map
             },
             created_at: Zoned::now(),
+            master_key_version: None,
         };
 
         // Test serialization
@@ -336,6 +346,8 @@ mod tests {
         let deserialized: DataKeyEnvelope = serde_json::from_str(envelope_json).expect("Should deserialize current format");
         assert_eq!(deserialized.key_id, "test-key-id");
         assert_eq!(deserialized.master_key_id, "master-key-id");
+        // Envelopes persisted before versioning must parse with no master key version.
+        assert_eq!(deserialized.master_key_version, None);
     }
 
     #[tokio::test]
@@ -353,6 +365,50 @@ mod tests {
         let deserialized: DataKeyEnvelope = serde_json::from_str(envelope_json).expect("Should deserialize legacy format");
         assert_eq!(deserialized.key_id, "test-key-id");
         assert_eq!(deserialized.master_key_id, "master-key-id");
+        assert_eq!(deserialized.master_key_version, None);
+    }
+
+    #[test]
+    fn test_data_key_envelope_none_version_serializes_without_field() {
+        // A `None` version must keep the serialized envelope on the historical
+        // seven-field JSON shape so non-rotating backends emit byte-compatible
+        // envelopes that older readers accept unchanged.
+        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: HashMap::new(),
+            created_at: Zoned::now(),
+            master_key_version: None,
+        };
+
+        let value = serde_json::to_value(&envelope).expect("serialize envelope");
+        let object = value.as_object().expect("envelope serializes to an object");
+        assert!(!object.contains_key("master_key_version"));
+        assert_eq!(object.len(), 7, "None version must not change the seven-field JSON shape");
+    }
+
+    #[test]
+    fn test_data_key_envelope_version_round_trip() {
+        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: HashMap::new(),
+            created_at: Zoned::now(),
+            master_key_version: Some(7),
+        };
+
+        let serialized = serde_json::to_vec(&envelope).expect("serialize envelope");
+        let value: serde_json::Value = serde_json::from_slice(&serialized).expect("parse serialized envelope");
+        assert_eq!(value.get("master_key_version"), Some(&serde_json::json!(7)));
+
+        let deserialized: DataKeyEnvelope = serde_json::from_slice(&serialized).expect("deserialize envelope");
+        assert_eq!(deserialized.master_key_version, Some(7));
     }
 
     #[test]
@@ -368,8 +424,19 @@ mod tests {
         }"#;
         let minio_legacy = br#"{"aead":"AES-256-GCM-HMAC-SHA-256","iv":[1],"nonce":[2],"bytes":[3]}"#;
         let duplicate_key_id = [b"{\"key_id\":\"duplicate\",".as_slice(), &kms_envelope[1..]].concat();
+        // Rotation-aware envelope: the optional master_key_version field must not
+        // change how mixed batches of old and new envelopes are routed.
+        let versioned_envelope = {
+            let mut value: serde_json::Value = serde_json::from_slice(kms_envelope).expect("parse KMS envelope fixture");
+            value
+                .as_object_mut()
+                .expect("KMS envelope fixture is an object")
+                .insert("master_key_version".to_string(), serde_json::json!(2));
+            serde_json::to_vec(&value).expect("serialize versioned envelope")
+        };
 
         assert!(is_data_key_envelope(kms_envelope));
+        assert!(is_data_key_envelope(&versioned_envelope));
         assert!(is_data_key_envelope(&[b" \n".as_slice(), kms_envelope].concat()));
         assert!(!is_data_key_envelope(&duplicate_key_id));
         assert!(!is_data_key_envelope(b"bm9uY2U=:Y2lwaGVydGV4dA=="));

crates/kms/src/error.rs

@@ -120,6 +120,10 @@ pub enum KmsError {
     /// Persisted key record uses a format version unknown to this build
     #[error("Unsupported key format version {version:?} for key {key_id}")]
     UnsupportedFormatVersion { key_id: String, version: String },
+
+    /// Requested master key version has no persisted material for the key
+    #[error("Key version {version} not found for key {key_id}")]
+    KeyVersionNotFound { key_id: String, version: u32 },
 }
 
 impl KmsError {
@@ -253,6 +257,14 @@ impl KmsError {
             version: version.into(),
         }
     }
+
+    /// Create a key version not found error
+    pub fn key_version_not_found<S: Into<String>>(key_id: S, version: u32) -> Self {
+        Self::KeyVersionNotFound {
+            key_id: key_id.into(),
+            version,
+        }
+    }
 }
 
 /// Convert from standard library errors