test(kms): pin rotation contracts across backends and document retention (#5486)

* feat(kms): retain historical master key versions for Vault KV2 rotation

Rotation previously had to be rejected outright because replacing the
stored material would orphan every DEK wrapped by earlier versions.
Vault KV2 now keeps each version's material in an immutable, create-only
record at {prefix}/{key_id}/versions/{N} and treats the top-level record
as the current-version pointer plus fast-path material copy:

- decrypt resolves the envelope's master_key_version to its version
  record; a missing version fails closed with KeyVersionNotFound and
  never falls back to the current material
- envelopes without a version (pre-versioning writers) resolve to the
  baseline_version frozen at the key's first rotation, so never-rotated
  keys behave exactly as before
- generate_data_key stamps the wrapping version from the same key record
  snapshot that supplied the material
- rotate_key commits in check-and-set order: freeze baseline, persist
  the next version's material, then switch the current pointer; any
  failure leaves the current pointer untouched, and concurrent rotations
  serialize on the CAS writes with monotonically unique versions
- key listings drop the versions/ directory entries and physical key
  deletion purges version records before the key record

Refs rustfs/backlog#1565

* test(kms): pin rotation contracts across backends and document retention

Completes the backlog#1565 series with the cross-backend regression net
and operator documentation:

- Vault Transit: ignored integration test proving version-prefixed
  historical ciphertext still decrypts after rotation, with no
  RustFS-side version bookkeeping in the envelope
- Local: offline mixed-format test interleaving pre-versioning and
  versioned envelopes through a rejected rotation; the existing
  rotation-rejection pinning test already covers material immutability
- docs: kms-backend-security.md gains the KV2 versioned retention
  model, version record retention/destruction preconditions, and the
  upgrade-before-first-rotation cluster constraint

Refs rustfs/backlog#1565

crates/kms/src/backends/local.rs

@@ -1910,6 +1910,64 @@ mod tests {
         );
     }
 
+    /// Mixed-format regression for rustfs/backlog#1565: a batch interleaving
+    /// pre-versioning envelopes (no master_key_version field) with versioned ones
+    /// must route and decrypt in full, and a rejected rotation in the middle must
+    /// not disturb either format.
+    #[tokio::test]
+    async fn mixed_format_envelopes_decrypt_across_rejected_rotation() {
+        let (client, _temp_dir) = create_test_client().await;
+        let key_id = "mixed-format-key";
+        client.create_key(key_id, "AES_256", None).await.expect("create key");
+
+        let request = GenerateKeyRequest::new(key_id.to_string(), "AES_256".to_string());
+        let mut batch = Vec::new();
+        for index in 0..4 {
+            let data_key = client.generate_data_key(&request, None).await.expect("generate data key");
+            let ciphertext = if index % 2 == 0 {
+                // Legacy shape: the local backend already omits master_key_version.
+                let envelope: serde_json::Value = serde_json::from_slice(&data_key.ciphertext).expect("parse envelope");
+                assert!(
+                    !envelope
+                        .as_object()
+                        .expect("envelope is an object")
+                        .contains_key("master_key_version"),
+                    "local envelopes must keep the pre-versioning shape"
+                );
+                data_key.ciphertext.clone()
+            } else {
+                // Versioned shape, as a rotation-aware writer would emit it.
+                let mut envelope: serde_json::Value = serde_json::from_slice(&data_key.ciphertext).expect("parse envelope");
+                envelope
+                    .as_object_mut()
+                    .expect("envelope is an object")
+                    .insert("master_key_version".to_string(), serde_json::json!(1));
+                serde_json::to_vec(&envelope).expect("serialize versioned envelope")
+            };
+            assert!(
+                crate::encryption::is_data_key_envelope(&ciphertext),
+                "batch member {index} must still route as a KMS envelope"
+            );
+            batch.push((ciphertext, data_key.plaintext.clone().expect("plaintext")));
+        }
+
+        // A rejected rotation in the middle of the batch's lifetime must leave
+        // every already-issued envelope decryptable.
+        let error = client
+            .rotate_key(key_id, None)
+            .await
+            .expect_err("local rotation must stay rejected");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }));
+
+        for (index, (ciphertext, plaintext)) in batch.iter().enumerate() {
+            let decrypted = client
+                .decrypt(&DecryptRequest::new(ciphertext.clone()), None)
+                .await
+                .unwrap_or_else(|error| panic!("batch member {index} must decrypt: {error}"));
+            assert_eq!(&decrypted, plaintext, "batch member {index} plaintext must round-trip");
+        }
+    }
+
     #[tokio::test]
     async fn startup_rejects_key_file_with_mismatched_embedded_id() {
         let (client, temp_dir) = create_dev_mode_client().await;

crates/kms/src/backends/vault_transit.rs

@@ -907,4 +907,70 @@ mod tests {
             "after restart, a pending-deletion key must not be usable for new data keys"
         );
     }
+
+    /// Contract regression for rustfs/backlog#1565.
+    ///
+    /// Transit rotation is delegated entirely to Vault's own key versioning: the
+    /// ciphertext self-describes the wrapping version ("vault:vN:..."), so historical
+    /// ciphertext must keep decrypting after rotation without any RustFS-side
+    /// version bookkeeping in the envelope.
+    #[tokio::test]
+    #[ignore] // Requires a running Vault instance with transit engine enabled
+    async fn test_transit_old_ciphertext_decrypts_after_rotate() {
+        let client = VaultTransitKmsClient::new(test_vault_transit_config(), Duration::from_secs(30))
+            .await
+            .expect("Failed to create VaultTransit client");
+
+        let key_id = format!("regression-1565-rotate-{}", uuid::Uuid::new_v4());
+        client.create_key(&key_id, "AES_256", None).await.expect("create_key");
+
+        let request = GenerateKeyRequest {
+            master_key_id: key_id.clone(),
+            key_spec: "AES_256".to_string(),
+            key_length: Some(32),
+            encryption_context: HashMap::new(),
+            grant_tokens: Vec::new(),
+        };
+
+        let dk_v1 = client.generate_data_key(&request, None).await.expect("generate under v1");
+        let env_v1: DataKeyEnvelope = serde_json::from_slice(&dk_v1.ciphertext).expect("parse v1 envelope");
+        assert!(
+            env_v1.encrypted_key.starts_with(b"vault:v1:"),
+            "first-version Transit ciphertext must carry the vault:v1: prefix"
+        );
+        assert_eq!(
+            env_v1.master_key_version, None,
+            "Transit envelopes must not carry a RustFS-side master key version"
+        );
+
+        let rotated = client.rotate_key(&key_id, None).await.expect("rotate_key");
+        assert_eq!(rotated.version, 2, "rotation must advance the Transit key version");
+
+        let dk_v2 = client.generate_data_key(&request, None).await.expect("generate under v2");
+        let env_v2: DataKeyEnvelope = serde_json::from_slice(&dk_v2.ciphertext).expect("parse v2 envelope");
+        assert!(
+            env_v2.encrypted_key.starts_with(b"vault:v2:"),
+            "post-rotation Transit ciphertext must carry the vault:v2: prefix"
+        );
+
+        // Historical ciphertext keeps decrypting per Vault's version semantics,
+        // interleaved with post-rotation ciphertext.
+        for (data_key, label) in [(&dk_v1, "v1"), (&dk_v2, "v2"), (&dk_v1, "v1 again")] {
+            let plaintext = client
+                .decrypt(
+                    &DecryptRequest {
+                        ciphertext: data_key.ciphertext.clone(),
+                        encryption_context: HashMap::new(),
+                        grant_tokens: Vec::new(),
+                    },
+                    None,
+                )
+                .await
+                .unwrap_or_else(|error| panic!("{label} ciphertext must stay decryptable after rotation: {error}"));
+            assert_eq!(Some(plaintext), data_key.plaintext, "{label} plaintext must round-trip");
+        }
+
+        // Cleanup so repeated runs against the same Vault do not accumulate keys.
+        let _ = client.schedule_key_deletion(&key_id, 7, None).await;
+    }
 }