From b8d2f1c84c74588f0e524c704910ac11c7d4f21a Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 1 Aug 2026 20:18:11 +0800 Subject: [PATCH] feat(kms): orchestrate and verify Vault-backed restores (#5549) --- crates/kms/src/backup/local_export.rs | 23 +- crates/kms/src/backup/manifest.rs | 314 ++++ crates/kms/src/backup/mod.rs | 19 +- crates/kms/src/backup/vault_restore.rs | 2312 ++++++++++++++++++++++++ 4 files changed, 2656 insertions(+), 12 deletions(-) create mode 100644 crates/kms/src/backup/vault_restore.rs diff --git a/crates/kms/src/backup/local_export.rs b/crates/kms/src/backup/local_export.rs index c0f393057..fb6da87ad 100644 --- a/crates/kms/src/backup/local_export.rs +++ b/crates/kms/src/backup/local_export.rs @@ -74,7 +74,7 @@ pub const LOCAL_BUNDLE_MANIFEST_FILE: &str = "manifest.json"; const ARTIFACTS_DIR: &str = "artifacts"; const KEYS_DIR: &str = "artifacts/keys"; const SALT_ARTIFACT_PATH: &str = "artifacts/master-key.salt.enc"; -const AEAD_NONCE_LEN: usize = 12; +pub(crate) const AEAD_NONCE_LEN: usize = 12; /// Domain-separation context for the artifact AAD binding. const BUNDLE_AAD_CONTEXT: &str = "rustfs-kms-local-backup:v1"; /// Domain-separation context for the master-key verifier. @@ -118,7 +118,7 @@ impl BackupKek { } } - fn cipher(&self) -> Aes256Gcm { + pub(crate) fn cipher(&self) -> Aes256Gcm { Aes256Gcm::new(&Key::::from(*self.key)) } } @@ -227,11 +227,14 @@ pub async fn export_local_backup( Ok(manifest) } -/// Read and fully validate the manifest of a local bundle directory. +/// Read and fully validate the manifest of a bundle directory, whatever +/// backend produced it. /// /// A directory without a manifest is an interrupted export: the manifest is -/// written last, so its absence means the bundle never sealed. -pub async fn read_local_bundle_manifest(bundle_dir: &Path) -> Result { +/// written last, so its absence means the bundle never sealed. The manifest +/// file name and framing are bundle-wide, not Local-specific, so consumers of +/// other backends' bundles read them through here too. +pub async fn read_bundle_manifest(bundle_dir: &Path) -> Result { let manifest_path = bundle_dir.join(LOCAL_BUNDLE_MANIFEST_FILE); let bytes = match fs::read(&manifest_path).await { Ok(bytes) => bytes, @@ -240,7 +243,12 @@ pub async fn read_local_bundle_manifest(bundle_dir: &Path) -> Result return Err(error.into()), }; - let manifest = BackupManifest::decode(&bytes)?; + Ok(BackupManifest::decode(&bytes)?) +} + +/// Read and fully validate the manifest of a *local* bundle directory. +pub async fn read_local_bundle_manifest(bundle_dir: &Path) -> Result { + let manifest = read_bundle_manifest(bundle_dir).await?; if manifest.backend != BackupBackendKind::Local { return Err( BackupError::corrupted(format!("bundle manifest declares backend {:?}, expected Local", manifest.backend)).into(), @@ -424,6 +432,7 @@ async fn build_and_write_bundle( backup_kek: kek.descriptor(), artifacts, local_kdf: Some(local_kdf_descriptor(snapshot, master_key_verifier)), + external_references: None, key_versions: None, capability_discovery: None, completeness: CompletenessState::InProgress, @@ -502,7 +511,7 @@ async fn encrypt_and_write_artifact( /// AAD binding an artifact to its bundle identity and path. A JSON tuple /// gives unambiguous field boundaries without a hand-rolled framing format. -fn artifact_aad(backup_id: &str, snapshot_generation: u64, artifact_path: &str) -> Vec { +pub(crate) fn artifact_aad(backup_id: &str, snapshot_generation: u64, artifact_path: &str) -> Vec { serde_json::to_vec(&(BUNDLE_AAD_CONTEXT, backup_id, snapshot_generation, artifact_path)) .expect("AAD tuple of strings and integers always serializes") } diff --git a/crates/kms/src/backup/manifest.rs b/crates/kms/src/backup/manifest.rs index 3eff4fad0..28ea05f76 100644 --- a/crates/kms/src/backup/manifest.rs +++ b/crates/kms/src/backup/manifest.rs @@ -306,6 +306,141 @@ impl LocalKdfDescriptor { } } +/// Transit engine reference recorded in a Vault-backed bundle. +/// +/// Transit keys are non-exportable, so a bundle can only ever name the key and +/// the version window its content depends on. Restore compares these values +/// against the Transit key the operator's native Vault restore produced. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct VaultTransitReference { + /// Transit engine mount path. + pub mount_path: String, + /// Transit key name. + pub key_name: String, + /// Lowest Transit key version the bundle's content still needs. A target + /// whose `min_decryption_version` sits above this can no longer decrypt + /// the oldest state in the bundle. + pub required_min_version: u32, + /// Latest Transit key version at snapshot time. A target whose newest + /// version is below this was restored to a point before the bundle. + pub current_version: u32, + /// Operator-recorded immutable reference to the native Vault/HSM snapshot + /// protecting this key. RustFS neither produces nor consumes that + /// snapshot; the reference exists so restore evidence can name it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub native_snapshot_reference: Option, +} + +/// KV generation of one key's record at snapshot time. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct VaultKvRecordReference { + /// Stable key id the record belongs to. + pub key_id: String, + /// KV2 secret version holding the record when the snapshot was taken. + /// Restore compares the target's current version against it: a lower + /// value means the target's Vault was restored to a point before the + /// bundle, a higher value means the target moved ahead of it. + pub kv_version: u64, +} + +/// Immutable references to the external Vault state a bundle depends on. +/// +/// A Vault-backed bundle never carries the cryptographic root: Transit keys +/// cannot be exported and KV2 records live in Vault's own storage. What the +/// bundle can own is a precise description of *which* external state it was +/// captured against, so a restore refuses to proceed when the operator's +/// native Vault restore landed somewhere else. Credentials are structurally +/// absent — no token, AppRole secret id, or TLS material is representable in +/// this type. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct VaultExternalReferences { + /// Opaque identity of the Vault cluster the snapshot was taken from. + pub cluster_id: String, + /// Vault Enterprise namespace, when the deployment uses one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub namespace: Option, + /// KV v2 mount holding the RustFS records. + pub kv_mount: String, + /// Path prefix under `kv_mount` holding the RustFS records. + pub kv_path_prefix: String, + /// Transit engine reference; present exactly when the bundle's material + /// is protected by a Transit key. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub transit: Option, + /// Per-key KV generation observed at snapshot time, one entry per key the + /// bundle carries a record for. + pub kv_records: Vec, +} + +impl VaultExternalReferences { + fn validate(&self, backend: BackupBackendKind, protection: AtRestProtection) -> Result<(), BackupError> { + require_non_empty("external_references.cluster_id", &self.cluster_id)?; + require_non_empty("external_references.kv_mount", &self.kv_mount)?; + require_non_empty("external_references.kv_path_prefix", &self.kv_path_prefix)?; + if self.namespace.as_deref() == Some("") { + return Err(BackupError::corrupted("external Vault namespace must not be empty when present")); + } + + // The Transit reference is required exactly where the cryptographic + // root lives in Transit; a storage-only KV2 bundle that claims one + // would misdescribe its own trust root. + let transit_required = matches!( + (backend, protection), + (BackupBackendKind::VaultTransit, _) | (BackupBackendKind::VaultKv2, AtRestProtection::TransitWrapped) + ); + match (&self.transit, transit_required) { + (None, true) => { + return Err(BackupError::corrupted(format!( + "({backend:?}, {protection:?}) bundles must reference the Transit key protecting them" + ))); + } + (Some(_), false) => { + return Err(BackupError::corrupted(format!( + "({backend:?}, {protection:?}) bundles have no Transit trust root to reference" + ))); + } + (Some(transit), true) => { + require_non_empty("external_references.transit.mount_path", &transit.mount_path)?; + require_non_empty("external_references.transit.key_name", &transit.key_name)?; + if transit.required_min_version == 0 { + return Err(BackupError::corrupted("Transit reference must require at least key version 1")); + } + if transit.current_version < transit.required_min_version { + return Err(BackupError::corrupted(format!( + "Transit reference declares current version {} below the required minimum {}", + transit.current_version, transit.required_min_version + ))); + } + if transit.native_snapshot_reference.as_deref() == Some("") { + return Err(BackupError::corrupted("Transit native snapshot reference must not be empty when present")); + } + } + (None, false) => {} + } + + let mut seen = BTreeSet::new(); + for record in &self.kv_records { + require_non_empty("external_references.kv_records[].key_id", &record.key_id)?; + if record.kv_version == 0 { + return Err(BackupError::corrupted(format!( + "KV generation for key '{}' must be at least 1", + record.key_id + ))); + } + if !seen.insert(record.key_id.as_str()) { + return Err(BackupError::corrupted(format!( + "KV reference for key '{}' appears more than once", + record.key_id + ))); + } + } + Ok(()) + } +} + /// Completeness marker of a bundle. /// /// A producer writes `in-progress` state (or no manifest at all) until the @@ -394,6 +529,10 @@ pub struct BackupManifest { /// Local backend section; required exactly when `backend` is `Local`. #[serde(default, skip_serializing_if = "Option::is_none")] pub local_kdf: Option, + /// External Vault references; required exactly when `backend` is a Vault + /// backend. Never carries credentials — see [`VaultExternalReferences`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub external_references: Option, /// Reserved for the per-key version/envelope inventory defined by the /// backlog#1565 contract. May not carry data in format version 1. #[serde(default, skip_serializing_if = "Option::is_none")] @@ -572,6 +711,7 @@ impl BackupManifest { } self.validate_responsibility()?; self.validate_local_kdf()?; + self.validate_external_references()?; self.validate_artifacts()?; Ok(()) } @@ -601,6 +741,19 @@ impl BackupManifest { } } + fn validate_external_references(&self) -> Result<(), BackupError> { + let vault_backend = matches!(self.backend, BackupBackendKind::VaultKv2 | BackupBackendKind::VaultTransit); + match (&self.external_references, vault_backend) { + (None, true) => Err(BackupError::corrupted("Vault backend manifest must carry external Vault references")), + (Some(_), false) => Err(BackupError::corrupted(format!( + "external Vault references are not valid for backend {:?}", + self.backend + ))), + (Some(references), true) => references.validate(self.backend, self.at_rest_protection), + (None, false) => Ok(()), + } + } + fn validate_artifacts(&self) -> Result<(), BackupError> { let mut paths = BTreeSet::new(); for artifact in &self.artifacts { @@ -865,6 +1018,7 @@ mod tests { vec![AtRestProtection::EncryptedMasterKey], Some("verifier-opaque-1".to_string()), )), + external_references: None, key_versions: None, capability_discovery: None, completeness: CompletenessState::InProgress, @@ -1271,4 +1425,164 @@ mod tests { assert_eq!(decoded, sealed); assert_eq!(decoded.responsibility, BackupResponsibility::ReferenceOnly); } + + fn transit_references() -> VaultExternalReferences { + VaultExternalReferences { + cluster_id: "vault-cluster-a".to_string(), + namespace: Some("tenant-a".to_string()), + kv_mount: "secret".to_string(), + kv_path_prefix: "rustfs/kms/transit-metadata".to_string(), + transit: Some(VaultTransitReference { + mount_path: "transit".to_string(), + key_name: "rustfs-master".to_string(), + required_min_version: 2, + current_version: 5, + native_snapshot_reference: Some("s3://dr/vault/2026-08-01.snap".to_string()), + }), + kv_records: vec![VaultKvRecordReference { + key_id: "object-key".to_string(), + kv_version: 4, + }], + } + } + + fn transit_manifest_unsealed() -> BackupManifest { + BackupManifest { + backend: BackupBackendKind::VaultTransit, + at_rest_protection: AtRestProtection::ExternalNonExportable, + responsibility: BackupResponsibility::MetadataPlusExternalRoot, + artifacts: vec![artifact(ArtifactKind::KeyMetadata, "vault/records/object-key.json.enc")], + local_kdf: None, + external_references: Some(transit_references()), + ..local_manifest_unsealed() + } + } + + #[test] + fn vault_bundle_round_trips_with_external_references() { + let sealed = seal(transit_manifest_unsealed()); + let bytes = sealed.encode().expect("encoding should succeed"); + let decoded = BackupManifest::decode(&bytes).expect("decoding should succeed"); + assert_eq!(decoded, sealed); + } + + #[test] + fn vault_backend_requires_external_references() { + let manifest = BackupManifest { + external_references: None, + ..transit_manifest_unsealed() + }; + expect_corrupted( + BackupManifest::decode(&serde_json::to_vec(&seal(manifest)).expect("serialize")), + "must carry external Vault references", + ); + } + + #[test] + fn non_vault_backend_rejects_external_references() { + let manifest = BackupManifest { + external_references: Some(transit_references()), + ..local_manifest_unsealed() + }; + expect_corrupted( + BackupManifest::decode(&serde_json::to_vec(&seal(manifest)).expect("serialize")), + "not valid for backend", + ); + } + + /// One way to contradict an otherwise valid set of Vault references. + type ReferenceContradiction = Box; + + #[test] + fn external_reference_contradictions_fail_closed() { + let cases: [(&str, ReferenceContradiction); 7] = [ + ("cluster_id", Box::new(|refs| refs.cluster_id.clear())), + ("kv_mount", Box::new(|refs| refs.kv_mount.clear())), + ("namespace must not be empty", Box::new(|refs| refs.namespace = Some(String::new()))), + ( + "must reference the Transit key", + Box::new(|refs: &mut VaultExternalReferences| refs.transit = None), + ), + ( + "at least key version 1", + Box::new(|refs: &mut VaultExternalReferences| { + if let Some(transit) = refs.transit.as_mut() { + transit.required_min_version = 0; + } + }), + ), + ( + "below the required minimum", + Box::new(|refs: &mut VaultExternalReferences| { + if let Some(transit) = refs.transit.as_mut() { + transit.current_version = 1; + } + }), + ), + ( + "appears more than once", + Box::new(|refs: &mut VaultExternalReferences| { + let first = refs.kv_records[0].clone(); + refs.kv_records.push(first); + }), + ), + ]; + for (needle, mutate) in cases { + let mut references = transit_references(); + mutate(&mut references); + let manifest = BackupManifest { + external_references: Some(references), + ..transit_manifest_unsealed() + }; + expect_corrupted(BackupManifest::decode(&serde_json::to_vec(&seal(manifest)).expect("serialize")), needle); + } + } + + /// The reference schema is the only place a Vault coordinate reaches a + /// bundle, so its field set is pinned: a credential-carrying field could + /// only appear by editing this assertion. + #[test] + fn external_reference_field_set_is_frozen() { + let value = serde_json::to_value(transit_references()).expect("serialize"); + let mut top: Vec<&str> = value.as_object().expect("object").keys().map(String::as_str).collect(); + top.sort_unstable(); + assert_eq!( + top, + [ + "cluster_id", + "kv_mount", + "kv_path_prefix", + "kv_records", + "namespace", + "transit" + ] + ); + + let mut transit: Vec<&str> = value["transit"] + .as_object() + .expect("transit object") + .keys() + .map(String::as_str) + .collect(); + transit.sort_unstable(); + assert_eq!( + transit, + [ + "current_version", + "key_name", + "mount_path", + "native_snapshot_reference", + "required_min_version" + ] + ); + + let mut record: Vec<&str> = value["kv_records"][0] + .as_object() + .expect("record object") + .keys() + .map(String::as_str) + .collect(); + record.sort_unstable(); + assert_eq!(record, ["key_id", "kv_version"]); + } } diff --git a/crates/kms/src/backup/mod.rs b/crates/kms/src/backup/mod.rs index 53b0fa741..36ace38d7 100644 --- a/crates/kms/src/backup/mod.rs +++ b/crates/kms/src/backup/mod.rs @@ -17,9 +17,11 @@ //! The contract side defines the versioned backup manifest, the per-backend //! responsibility matrix, typed failure modes, and the restore dry-run //! report. [`local_export`] implements the producer side and -//! [`local_restore`] the consumer side for the Local backend as -//! crate-internal APIs; the admin API builds on these pieces in follow-up -//! changes. +//! [`local_restore`] the consumer side for the Local backend; +//! [`vault_restore`] orchestrates the consumer side for the Vault backends, +//! whose cryptographic root is restored by Vault's own disaster-recovery +//! flow. All are crate-internal APIs; the admin API builds on these pieces in +//! follow-up changes. //! //! # Bundle model //! @@ -54,6 +56,7 @@ mod error; pub mod local_export; pub mod local_restore; mod manifest; +pub mod vault_restore; pub use capability::{AtRestProtection, BackupBackendKind, BackupResponsibility}; pub use dry_run::{ @@ -62,7 +65,7 @@ pub use dry_run::{ pub use error::BackupError; pub use local_export::{ BackupKek, LOCAL_BUNDLE_MANIFEST_FILE, LocalBackupExportRequest, decrypt_bundle_artifact, export_local_backup, - read_local_bundle_manifest, + read_bundle_manifest, read_local_bundle_manifest, }; pub use local_restore::{ LocalRestoreReport, LocalRestoreRequest, RestoreConflictPolicy, abort_local_restore, dry_run_local_restore, @@ -70,5 +73,11 @@ pub use local_restore::{ }; pub use manifest::{ AeadAlgorithm, ArtifactDescriptor, ArtifactKind, BackupKekDescriptor, BackupManifest, CompletenessState, ContentDigest, - DigestAlgorithm, LocalKdfDescriptor, LocalKeyDerivation, ReservedSlot, + DigestAlgorithm, LocalKdfDescriptor, LocalKeyDerivation, ReservedSlot, VaultExternalReferences, VaultKvRecordReference, + VaultTransitReference, +}; +pub use vault_restore::{ + VaultRestoreAbortReport, VaultRestoreAbortSkip, VaultRestoreAbortSkipReason, VaultRestoreClient, VaultRestoreMismatch, + VaultRestoreReport, VaultRestoreRequest, VaultRestoreSequence, VaultRestoreStage, VaultRestoreTarget, abort_vault_restore, + dry_run_vault_restore, restore_vault_backup, }; diff --git a/crates/kms/src/backup/vault_restore.rs b/crates/kms/src/backup/vault_restore.rs new file mode 100644 index 000000000..7cd3f4dbf --- /dev/null +++ b/crates/kms/src/backup/vault_restore.rs @@ -0,0 +1,2312 @@ +// 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. + +//! Vault restore orchestration: stage ordering, external reference +//! verification, and mismatch classification. +//! +//! # What this module does and does not own +//! +//! The cryptographic root of a Vault deployment is restored by Vault's own +//! disaster-recovery flow: Transit keys are non-exportable and RustFS never +//! attempts to import them. What RustFS owns is the *orchestration* around +//! that operator action — proving the recovered Vault is the one the bundle +//! was captured against, publishing the RustFS-side KV records in the right +//! order, and refusing every combination it cannot prove safe. Taking or +//! restoring the native Vault snapshot itself stays an operator step. +//! +//! # Restore order +//! +//! [`VaultRestoreStage`] is walked strictly in order and +//! [`VaultRestoreSequence`] enforces it structurally: +//! +//! 1. [`VaultRestoreStage::ExternalTrustRoot`] — verify the Transit/HSM trust +//! root the operator restored matches the bundle's references. Read-only. +//! 2. [`VaultRestoreStage::MaterialAndVersions`] — publish material and +//! version records ([`ArtifactKind::KeyMaterial`]). +//! 3. [`VaultRestoreStage::KvMetadata`] — publish metadata records +//! ([`ArtifactKind::KeyMetadata`]); alias and policy artifacts are reserved +//! names for features that do not exist yet. +//! 4. [`VaultRestoreStage::ConfigAndCutover`] — drop the commit marker. The +//! sanitized configuration artifact is the admin layer's to apply. +//! +//! # Mismatch classification +//! +//! Every way the recovered Vault can disagree with the bundle is a distinct +//! [`VaultRestoreMismatch`] variant: a missing Transit key, a +//! `min_decryption_version` raised above what the bundle still needs, a +//! Transit version history restored below the bundle's snapshot point, a KV +//! generation rolled back, and cluster/namespace/mount identity drift. A +//! dry-run reports them all; an actual restore refuses on the first one. +//! +//! Verification is split by what it costs: every coordinate that follows from +//! the bundle and the caller's target description — cluster, namespace, KV +//! mount, KV prefix, Transit mount — is settled first, and any disagreement +//! there is returned before a single Vault request goes out. A target whose +//! coordinates already drifted is not the deployment the bundle describes, so +//! it is neither probed nor read through. +//! +//! # Writes +//! +//! [`dry_run_vault_restore`] issues reads only. Every record +//! [`restore_vault_backup`] publishes goes through create-only +//! check-and-set: a record that already exists is never overwritten, so the +//! default [`RestoreConflictPolicy::Fail`] and the empty-target policy differ +//! only in whether writing is attempted at all. Version regression, generation +//! rollback, and reviving deleted state are structurally impossible rather +//! than merely rejected. The single non-create-only write is the restore's own +//! commit marker, updated check-and-set against the exact version the restore +//! last observed. +//! +//! # Crash re-entry +//! +//! The commit marker is a reserved KV record published (create-only) before +//! the first write and removed after the last. Before it, the target is +//! untouched; with it published, re-running the same bundle rolls forward +//! idempotently. Every interruption converges to the complete old or the +//! complete new state. +//! +//! # Abort and ownership +//! +//! The marker names the key ids a restore *intends* to publish, which is not +//! evidence that it published them: the marker goes out first, and a record +//! write can afterwards be lost to a concurrent creator, or never be reached +//! at all. So each successful create-only write appends a receipt to the +//! marker carrying the KV version it landed at, and +//! [`abort_vault_restore`] removes a record only when the path still holds +//! exactly that version. Anything else — no receipt, a different version, an +//! already-empty path — is left alone and reported as skipped. An abort can +//! therefore leave records behind, but it can never delete a record this +//! restore did not write. + +use crate::backends::local::validate_key_id; +use crate::backends::vault_credentials::{ + VaultClientHandle, VaultConnectionSettings, VaultCredentialPolicy, VaultCredentialProvider, token_source_for, +}; +use crate::backup::dry_run::{ + ExternalDependencyMismatch, RestoreBlocker, RestoreBlockerCode, RestoreConflict, RestoreConflictKind, RestoreDryRunReport, +}; +use crate::backup::error::BackupError; +use crate::backup::local_export::{BackupKek, decrypt_bundle_artifact, read_bundle_manifest}; +use crate::backup::local_restore::RestoreConflictPolicy; +use crate::backup::manifest::{ArtifactKind, BackupManifest, ContentDigest, VaultExternalReferences, VaultTransitReference}; +use crate::config::{KmsConfig, VaultAuthMethod}; +use crate::error::{KmsError, Result}; +use crate::policy::{self, AttemptError, OpClass, RetryPolicy}; +use serde::{Deserialize, Serialize}; +use std::future::Future; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use thiserror::Error; +use tokio_util::sync::CancellationToken; +use vaultrs::api::kv2::requests::SetSecretRequestOptions; +use vaultrs::api::transit::responses::ReadKeyData; +use vaultrs::error::ClientError; +use vaultrs::{kv2, transit::key}; + +/// Bundle-relative directory holding one KV record artifact per key. +const VAULT_RECORD_ARTIFACT_DIR: &str = "vault/records"; +/// Bundle-relative suffix of a KV record artifact. +const VAULT_RECORD_ARTIFACT_SUFFIX: &str = ".json.enc"; +/// KV path segment (under the bundle's path prefix) of the restore commit +/// marker. The leading dot keeps it out of the key id space the backends +/// address; a bundle naming a key id equal to it is rejected. +const VAULT_RESTORE_MARKER_KEY: &str = ".rustfs-restore-commit"; +/// Format version of the commit marker record. +const VAULT_RESTORE_MARKER_FORMAT_VERSION: u32 = 1; +/// Sentinel key id for report entries that concern the whole target. +const WHOLE_TARGET_CONFLICT_ID: &str = "*"; + +/// The restore stages, in the only order they may run. +/// +/// Restoring RustFS-side state before the external trust root is back would +/// leave metadata pointing at key versions Vault cannot serve, so the order is +/// enforced rather than documented. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum VaultRestoreStage { + /// Verify the Vault/HSM trust root the operator restored. Read-only: + /// RustFS never imports non-exportable material. + ExternalTrustRoot, + /// Publish key material and version records. + MaterialAndVersions, + /// Publish KV metadata (and, once those features exist, policy and alias) + /// records. + KvMetadata, + /// Apply configuration and complete the cutover. + ConfigAndCutover, +} + +impl VaultRestoreStage { + /// The stages in execution order. + pub const ORDER: [Self; 4] = [ + Self::ExternalTrustRoot, + Self::MaterialAndVersions, + Self::KvMetadata, + Self::ConfigAndCutover, + ]; + + fn index(self) -> usize { + Self::ORDER + .iter() + .position(|stage| *stage == self) + .expect("ORDER lists every stage") + } +} + +/// Guard that makes the restore order structural. +/// +/// A stage may only be entered when its predecessor completed, and only one +/// stage may be open at a time. Callers that skip, repeat, or reorder a stage +/// get a typed refusal instead of a partially ordered restore. +#[derive(Debug, Default)] +pub struct VaultRestoreSequence { + next: usize, + open: bool, +} + +impl VaultRestoreSequence { + /// A sequence positioned before the first stage. + pub fn new() -> Self { + Self::default() + } + + /// The stage that may be entered next, or `None` once all stages ran. + pub fn expected(&self) -> Option { + VaultRestoreStage::ORDER.get(self.next).copied() + } + + /// Begin `stage`, failing closed when it is not the expected one. + pub fn enter(&mut self, stage: VaultRestoreStage) -> Result<()> { + if self.open || stage.index() != self.next { + return Err(self.order_violation(stage, "start")); + } + self.open = true; + Ok(()) + } + + /// Finish `stage`, failing closed when it is not the open one. + pub fn complete(&mut self, stage: VaultRestoreStage) -> Result<()> { + if !self.open || stage.index() != self.next { + return Err(self.order_violation(stage, "complete")); + } + self.open = false; + self.next += 1; + Ok(()) + } + + fn order_violation(&self, stage: VaultRestoreStage, action: &str) -> KmsError { + KmsError::invalid_operation(format!( + "vault restore may not {action} stage {stage:?}: the restore order is {:?} and the next permitted stage is {:?}", + VaultRestoreStage::ORDER, + self.expected() + )) + } +} + +/// One way the recovered Vault disagrees with what the bundle references. +/// +/// Values are identifiers and version numbers only; no credential or key +/// material is representable here. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum VaultRestoreMismatch { + /// The target Vault cluster is not the one the bundle was captured from. + #[error("bundle references Vault cluster '{expected}' but the restore target reports '{observed}'")] + ClusterIdentity { expected: String, observed: String }, + + /// The target namespace differs from the bundle's. + #[error("bundle references Vault namespace {expected:?} but the restore target uses {observed:?}")] + Namespace { + expected: Option, + observed: Option, + }, + + /// A mount path (KV, KV prefix, or Transit) differs from the bundle's. + #[error("bundle references {dependency} '{expected}' but the restore target uses '{observed}'")] + Mount { + dependency: String, + expected: String, + observed: String, + }, + + /// The Transit key the bundle depends on does not exist in the target. + /// The operator's native Vault restore has not completed, or completed + /// against the wrong mount. + #[error("Transit key '{key_name}' is absent from mount '{mount_path}'; restore the Vault trust root first")] + TransitKeyMissing { mount_path: String, key_name: String }, + + /// The Transit key no longer decrypts the oldest content in the bundle. + #[error( + "Transit key '{key_name}' has min_decryption_version {observed}, above the version {required} the bundle still needs" + )] + TransitMinDecryptionVersionTooHigh { key_name: String, required: u32, observed: u64 }, + + /// The Transit key's version history was restored below the bundle's + /// snapshot point. + #[error( + "Transit key '{key_name}' has latest version {observed_latest}, below the version {expected_current} the bundle was captured at" + )] + TransitVersionHistoryRegressed { + key_name: String, + expected_current: u32, + observed_latest: u32, + }, + + /// A KV record's generation is below the bundle's snapshot point: the + /// target's Vault was restored to an earlier moment than the bundle. + #[error( + "KV record for key '{key_id}' is at generation {observed}, below the generation {expected} the bundle was captured at" + )] + KvGenerationRegressed { key_id: String, expected: u64, observed: u64 }, +} + +impl VaultRestoreMismatch { + /// Whether the mismatch means the dependency is absent rather than + /// inconsistent. Absent dependencies are reported as blockers, the rest as + /// external dependency mismatches; either way the restore is refused. + pub fn is_dependency_unavailable(&self) -> bool { + matches!(self, Self::TransitKeyMissing { .. }) + } + + /// Name of the affected dependency, for the dry-run report. + pub fn dependency(&self) -> String { + match self { + Self::ClusterIdentity { .. } => "vault cluster identity".to_string(), + Self::Namespace { .. } => "vault namespace".to_string(), + Self::Mount { dependency, .. } => dependency.clone(), + Self::TransitKeyMissing { mount_path, key_name } => format!("vault transit key {mount_path}/{key_name}"), + Self::TransitMinDecryptionVersionTooHigh { key_name, .. } | Self::TransitVersionHistoryRegressed { key_name, .. } => { + format!("vault transit key {key_name}") + } + Self::KvGenerationRegressed { key_id, .. } => format!("vault kv record for key {key_id}"), + } + } + + fn expected_value(&self) -> String { + match self { + Self::ClusterIdentity { expected, .. } | Self::Mount { expected, .. } => expected.clone(), + Self::Namespace { expected, .. } => optional_value(expected), + Self::TransitKeyMissing { key_name, .. } => format!("key {key_name} present"), + Self::TransitMinDecryptionVersionTooHigh { required, .. } => format!("min_decryption_version<={required}"), + Self::TransitVersionHistoryRegressed { expected_current, .. } => format!("latest_version>={expected_current}"), + Self::KvGenerationRegressed { expected, .. } => format!("kv_generation>={expected}"), + } + } + + fn observed_value(&self) -> String { + match self { + Self::ClusterIdentity { observed, .. } | Self::Mount { observed, .. } => observed.clone(), + Self::Namespace { observed, .. } => optional_value(observed), + Self::TransitKeyMissing { .. } => "absent".to_string(), + Self::TransitMinDecryptionVersionTooHigh { observed, .. } => format!("min_decryption_version={observed}"), + Self::TransitVersionHistoryRegressed { observed_latest, .. } => format!("latest_version={observed_latest}"), + Self::KvGenerationRegressed { observed, .. } => format!("kv_generation={observed}"), + } + } + + /// Record the mismatch in a dry-run report. + fn record_in(&self, report: &mut RestoreDryRunReport) { + if self.is_dependency_unavailable() { + report.blockers.push(RestoreBlocker { + code: RestoreBlockerCode::ExternalDependencyUnavailable, + detail: self.to_string(), + }); + } else { + report.external_mismatches.push(ExternalDependencyMismatch { + dependency: self.dependency(), + expected: self.expected_value(), + observed: self.observed_value(), + }); + } + } +} + +impl From for KmsError { + fn from(mismatch: VaultRestoreMismatch) -> Self { + KmsError::invalid_operation(mismatch.to_string()) + } +} + +fn optional_value(value: &Option) -> String { + value.clone().unwrap_or_else(|| "".to_string()) +} + +/// Coordinates of the Vault deployment a bundle is restored into. +/// +/// `cluster_id` is injected by the caller rather than probed: the admin layer +/// owns how a deployment's Vault cluster identity is established, and this +/// module freezes only that the bundle's recorded identity must equal it. +#[derive(Debug, Clone)] +pub struct VaultRestoreTarget { + /// Vault server URL. + pub address: String, + /// Authentication method for the restore credentials. + pub auth_method: VaultAuthMethod, + /// Vault Enterprise namespace, when the deployment uses one. + pub namespace: Option, + /// Identity of the Vault cluster as the caller established it. + pub cluster_id: String, + /// KV v2 mount the RustFS records live under. + pub kv_mount: String, + /// Path prefix under `kv_mount` for the RustFS records. + pub kv_path_prefix: String, + /// Transit engine mount, for deployments whose trust root is in Transit. + pub transit_mount: Option, +} + +/// Parameters of one Vault restore (or dry-run) attempt. +#[derive(Debug, Clone)] +pub struct VaultRestoreRequest { + /// Bundle directory produced by a Vault export. Read-only throughout. + pub bundle_dir: PathBuf, + /// Where the bundle is being restored to. + pub target: VaultRestoreTarget, + /// Deployment identity of the restore target; must match the manifest. + pub target_deployment_identity: String, + /// Highest snapshot generation the target has already observed, when the + /// caller tracks one. Equal generations stay allowed so a drill can be + /// repeated; a strictly lower bundle generation is refused. + pub observed_generation: Option, + /// Conflict policy; see [`RestoreConflictPolicy`]. + pub conflict_policy: RestoreConflictPolicy, +} + +impl VaultRestoreRequest { + fn validate(&self) -> Result<()> { + if self.target_deployment_identity.is_empty() { + return Err(KmsError::validation_error("restore target_deployment_identity must not be empty")); + } + if self.target.cluster_id.is_empty() { + return Err(KmsError::validation_error("restore target cluster_id must not be empty")); + } + Ok(()) + } +} + +/// Serializable outcome of a completed Vault restore. Identifiers only. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct VaultRestoreReport { + /// Identifier of the restored bundle. + pub backup_id: String, + /// Snapshot generation the target now holds. + pub snapshot_generation: u64, + /// Key ids this restore published. + pub restored_key_ids: Vec, + /// Key ids the target already held with exactly the bundle's content, and + /// which the restore therefore left alone. + pub already_present_key_ids: Vec, + /// Transit key the bundle's trust root was verified against, if any. + pub transit_key_name: Option, + /// Whether this run rolled forward a previously interrupted restore. + pub resumed: bool, +} + +/// Read-only client for the Vault deployment being restored into, plus the one +/// create-only write the restore performs. +/// +/// Deliberately separate from the KMS backends: a restore runs against a Vault +/// that is not yet serving traffic, addresses reserved paths the backends do +/// not know about, and must never inherit a backend's read-repair or +/// lazy-create behavior. +pub struct VaultRestoreClient { + credentials: Arc, + kv_mount: String, + retry: RetryPolicy, + cancel: CancellationToken, +} + +/// Transit key state as the target reports it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ObservedTransitKey { + min_decryption_version: u64, + latest_version: u32, +} + +/// One KV record as the target holds it. +#[derive(Debug, Clone, PartialEq, Eq)] +struct ObservedKvRecord { + generation: u64, + content: serde_json::Value, +} + +impl VaultRestoreClient { + /// Authenticate against the restore target. + pub async fn connect(target: &VaultRestoreTarget, kms_config: &KmsConfig) -> Result { + let settings = VaultConnectionSettings { + address: target.address.clone(), + namespace: target.namespace.clone(), + attempt_timeout: kms_config.effective_timeout(), + }; + let source = token_source_for(&target.auth_method, &settings)?; + let policy = VaultCredentialPolicy::from_kms_config(kms_config, &target.auth_method); + let credentials = Arc::new(VaultCredentialProvider::new(settings, source, policy).await?); + Ok(Self { + credentials, + kv_mount: target.kv_mount.clone(), + retry: RetryPolicy::from_config(kms_config), + cancel: CancellationToken::new(), + }) + } + + fn vault(&self) -> Result> { + self.credentials.current() + } + + async fn run(&self, operation: &'static str, class: OpClass, attempt: F) -> Result + where + F: FnMut() -> Fut, + Fut: Future>, + { + policy::execute(operation, class, &self.retry, &self.cancel, attempt).await + } + + /// Read a Transit key's version window. `None` means the key is absent. + async fn read_transit_key(&self, mount_path: &str, key_name: &str) -> Result> { + let response = self + .run("vault_restore_read_transit_key", OpClass::ReadIdempotent, move || async move { + match key::read(&self.vault().map_err(AttemptError::fatal)?.client, mount_path, key_name).await { + Ok(response) => Ok(Some(response)), + Err(ClientError::ResponseWrapError) | Err(ClientError::APIError { code: 404, .. }) => Ok(None), + Err(error) => Err(AttemptError::from_vaultrs(error, |error| { + KmsError::backend_error(format!("Failed to read Transit key {key_name} during restore: {error}")) + })), + } + }) + .await?; + + Ok(response.map(|response| ObservedTransitKey { + min_decryption_version: response.min_decryption_version, + latest_version: latest_transit_version(&response.keys), + })) + } + + /// Read the version a KV path currently sits at. `None` means it carries + /// no record. + async fn read_kv_generation(&self, path: &str) -> Result> { + self.run("vault_restore_read_kv_generation", OpClass::ReadIdempotent, move || async move { + match kv2::read_metadata(&self.vault().map_err(AttemptError::fatal)?.client, &self.kv_mount, path).await { + Ok(metadata) => Ok(Some(metadata.current_version)), + Err(ClientError::ResponseWrapError) | Err(ClientError::APIError { code: 404, .. }) => Ok(None), + Err(error) => Err(AttemptError::from_vaultrs(error, |error| { + KmsError::backend_error(format!("Failed to read KV generation of {path} during restore: {error}")) + })), + } + }) + .await + } + + /// Read a KV record together with the secret version holding it. `None` + /// means the path carries no readable record. + async fn read_kv_record(&self, path: &str) -> Result> { + let Some(generation) = self.read_kv_generation(path).await? else { + return Ok(None); + }; + + let content = self + .run("vault_restore_read_kv_record", OpClass::ReadIdempotent, move || async move { + match kv2::read::(&self.vault().map_err(AttemptError::fatal)?.client, &self.kv_mount, path) + .await + { + Ok(content) => Ok(Some(content)), + Err(ClientError::ResponseWrapError) | Err(ClientError::APIError { code: 404, .. }) => Ok(None), + Err(error) => Err(AttemptError::from_vaultrs(error, |error| { + KmsError::backend_error(format!("Failed to read KV record {path} during restore: {error}")) + })), + } + }) + .await?; + + Ok(content.map(|content| ObservedKvRecord { generation, content })) + } + + /// Create a KV record, never overwriting: the check-and-set precondition is + /// "no version exists". `Ok(None)` means a concurrent writer got there + /// first, which the caller surfaces instead of clobbering. `Ok(Some(v))` + /// returns the version the write landed at, which is the only evidence + /// that this restore — and not some other writer — owns the path. + async fn create_kv_record(&self, path: &str, content: &serde_json::Value) -> Result> { + self.run("vault_restore_create_kv_record", OpClass::MutatingNonIdempotent, move || async move { + match kv2::set_with_options( + &self.vault().map_err(AttemptError::fatal)?.client, + &self.kv_mount, + path, + content, + SetSecretRequestOptions { cas: 0 }, + ) + .await + { + Ok(metadata) => Ok(Some(metadata.version)), + Err(error) if is_cas_conflict(&error) => Ok(None), + Err(error) => Err(AttemptError::from_vaultrs(error, |error| { + KmsError::backend_error(format!("Failed to create KV record {path} during restore: {error}")) + })), + } + }) + .await + } + + /// Replace a KV record, check-and-set against the exact version the caller + /// last observed. Applied only to the restore's own commit marker: the + /// restore is the marker's sole writer, so losing this race means another + /// actor is writing the reserved path and the run must stop rather than + /// re-read and retry. + async fn update_kv_record(&self, path: &str, content: &serde_json::Value, expected_version: u64) -> Result { + let cas = u32::try_from(expected_version).map_err(|_| { + KmsError::internal_error(format!( + "KV record {path} sits at version {expected_version}, beyond what check-and-set can address" + )) + })?; + self.run("vault_restore_update_kv_record", OpClass::MutatingNonIdempotent, move || async move { + match kv2::set_with_options( + &self.vault().map_err(AttemptError::fatal)?.client, + &self.kv_mount, + path, + content, + SetSecretRequestOptions { cas }, + ) + .await + { + Ok(metadata) => Ok(metadata.version), + Err(error) if is_cas_conflict(&error) => Err(AttemptError::fatal(KmsError::invalid_operation(format!( + "the restore commit marker at {path} was modified concurrently; another restore or \ + operator is acting on this target" + )))), + Err(error) => Err(AttemptError::from_vaultrs(error, |error| { + KmsError::backend_error(format!("Failed to update KV record {path} during restore: {error}")) + })), + } + }) + .await + } + + /// Remove a KV record and all its versions. Only ever applied to the + /// commit marker and to records whose current version the caller has just + /// matched against the marker's write receipt for them. + async fn delete_kv_record(&self, path: &str) -> Result<()> { + self.run("vault_restore_delete_kv_record", OpClass::MutatingNonIdempotent, move || async move { + match kv2::delete_metadata(&self.vault().map_err(AttemptError::fatal)?.client, &self.kv_mount, path).await { + Ok(()) => Ok(()), + Err(ClientError::ResponseWrapError) | Err(ClientError::APIError { code: 404, .. }) => Ok(()), + Err(error) => Err(AttemptError::from_vaultrs(error, |error| { + KmsError::backend_error(format!("Failed to remove KV record {path} during restore: {error}")) + })), + } + }) + .await + } +} + +/// Whether a KV2 write failed its check-and-set precondition. Mirrors the +/// helper of the same name in the Vault backends, which share no private +/// module with this one. +fn is_cas_conflict(error: &ClientError) -> bool { + matches!( + error, + ClientError::APIError { code: 400, errors } if errors.iter().any(|message| message.contains("check-and-set")) + ) +} + +/// Highest version present in a Transit key's version map. +fn latest_transit_version(keys: &ReadKeyData) -> u32 { + let versions: Vec<&String> = match keys { + ReadKeyData::Symmetric(map) => map.keys().collect(), + ReadKeyData::Asymmetric(map) => map.keys().collect(), + }; + versions + .into_iter() + .filter_map(|name| name.parse::().ok()) + .max() + .unwrap_or(0) +} + +/// The commit marker: its create-only publication is the single commit point +/// of a Vault restore, and from then on it doubles as the restore's write +/// ledger. +/// +/// `key_ids` is intent — what this restore set out to publish — and drives +/// re-entry. `written` is evidence: one receipt per record the restore +/// actually created, carrying the KV version its create-only write landed at. +/// Intent alone proves nothing about ownership (a write can be lost to a +/// concurrent creator after the marker is published), so only `written` may +/// authorize an abort to remove anything. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct VaultRestoreCommitMarker { + format_version: u32, + backup_id: String, + manifest_digest: ContentDigest, + key_ids: Vec, + /// Absent in a marker published before the first record write. + #[serde(default)] + written: Vec, +} + +/// Proof that this restore, and nothing else, created one KV record. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct VaultRestoreWriteReceipt { + key_id: String, + /// Version the create-only write landed at. + kv_version: u64, +} + +impl VaultRestoreCommitMarker { + fn new(manifest: &BackupManifest, key_ids: Vec) -> Self { + Self { + format_version: VAULT_RESTORE_MARKER_FORMAT_VERSION, + backup_id: manifest.backup_id.clone(), + manifest_digest: manifest.manifest_digest.clone(), + key_ids, + written: Vec::new(), + } + } + + /// Strict decode: an out-of-contract marker fails closed, because both + /// roll-forward and abort would otherwise guess what it was committing. + fn from_value(value: serde_json::Value) -> Result { + let marker: Self = serde_json::from_value(value) + .map_err(|error| BackupError::corrupted(format!("vault restore commit marker does not decode: {error}")))?; + if marker.format_version != VAULT_RESTORE_MARKER_FORMAT_VERSION { + return Err(BackupError::corrupted(format!( + "vault restore commit marker has unknown format version {}", + marker.format_version + )) + .into()); + } + for key_id in &marker.key_ids { + validate_key_id(key_id)?; + } + for receipt in &marker.written { + validate_key_id(&receipt.key_id)?; + if !marker.key_ids.contains(&receipt.key_id) { + return Err(BackupError::corrupted(format!( + "vault restore commit marker carries a write receipt for key '{}', which it never claimed", + receipt.key_id + )) + .into()); + } + } + Ok(marker) + } + + /// The version this restore wrote key `key_id` at, if it got that far. + fn receipt_for(&self, key_id: &str) -> Option { + self.written + .iter() + .find(|receipt| receipt.key_id == key_id) + .map(|receipt| receipt.kv_version) + } + + fn matches_bundle(&self, manifest: &BackupManifest) -> bool { + self.backup_id == manifest.backup_id && self.manifest_digest == manifest.manifest_digest + } + + fn to_content(&self) -> Result { + serde_json::to_value(self) + .map_err(|error| KmsError::internal_error(format!("restore commit marker serialization failed: {error}"))) + } +} + +/// A commit marker as the target holds it, with the version to check-and-set +/// against when the restore appends its next write receipt. +struct PublishedMarker { + marker: VaultRestoreCommitMarker, + kv_version: u64, +} + +/// One KV record decoded out of the bundle. +struct DecodedVaultRecord { + key_id: String, + /// Stage that publishes this record, derived from its artifact kind. + stage: VaultRestoreStage, + /// KV generation the record was captured at. + kv_version: u64, + content: serde_json::Value, +} + +/// The fully decoded and cross-checked Vault bundle content. +struct DecodedVaultBundle { + manifest: BackupManifest, + references: VaultExternalReferences, + records: Vec, +} + +/// Decode and cross-verify a Vault bundle in memory. +/// +/// Every artifact is read, digest-verified, and AEAD-authenticated; every +/// record is re-parsed, identity-checked against its artifact path, and +/// matched against the manifest's KV references in both directions, so a +/// bundle that references state it does not carry (or carries state it does +/// not reference) is rejected before anything is inspected. +async fn decode_vault_bundle(bundle_dir: &Path, kek: &BackupKek) -> Result { + let manifest = read_bundle_manifest(bundle_dir).await?; + let references = manifest + .external_references + .clone() + .ok_or_else(|| BackupError::corrupted("vault bundle manifest carries no external Vault references"))?; + + let mut records = Vec::with_capacity(manifest.artifacts.len()); + for artifact in &manifest.artifacts { + let stage = match artifact.kind { + ArtifactKind::KeyMaterial => VaultRestoreStage::MaterialAndVersions, + ArtifactKind::KeyMetadata => VaultRestoreStage::KvMetadata, + // Restoring a bundle that carries state this implementation + // cannot apply would silently drop it, so fail closed instead. + other => { + return Err(KmsError::unsupported_capability( + "vault-restore", + format!("bundle carries artifact kind {other:?}, which this restore implementation cannot apply"), + )); + } + }; + + let key_id = vault_record_key_id(&artifact.path)?; + let reference = references + .kv_records + .iter() + .find(|record| record.key_id == key_id) + .ok_or_else(|| { + BackupError::corrupted(format!("bundle carries a record for key '{key_id}' with no KV generation reference")) + })?; + + let plaintext = decrypt_bundle_artifact(bundle_dir, &manifest, artifact, kek).await?; + let content: serde_json::Value = serde_json::from_slice(&plaintext) + .map_err(|error| BackupError::corrupted(format!("bundled record for key '{key_id}' does not decode: {error}")))?; + if !content.is_object() { + return Err(BackupError::corrupted(format!("bundled record for key '{key_id}' is not a KV object")).into()); + } + + records.push(DecodedVaultRecord { + key_id, + stage, + kv_version: reference.kv_version, + content, + }); + } + + for reference in &references.kv_records { + if !records.iter().any(|record| record.key_id == reference.key_id) { + return Err(BackupError::missing_artifact(format!("{VAULT_RECORD_ARTIFACT_DIR}/{}", reference.key_id)).into()); + } + } + + // Deterministic order: stage first (the restore order), then key id, so a + // restore issues the same request sequence on every run. + records.sort_by(|left, right| (left.stage, &left.key_id).cmp(&(right.stage, &right.key_id))); + + Ok(DecodedVaultBundle { + manifest, + references, + records, + }) +} + +/// Extract and validate the key id from a bundle record artifact path. +fn vault_record_key_id(artifact_path: &str) -> Result { + let key_id = artifact_path + .strip_prefix(&format!("{VAULT_RECORD_ARTIFACT_DIR}/")) + .and_then(|name| name.strip_suffix(VAULT_RECORD_ARTIFACT_SUFFIX)) + .ok_or_else(|| { + BackupError::corrupted(format!( + "vault record artifact path {artifact_path:?} does not follow the \ + '{VAULT_RECORD_ARTIFACT_DIR}/{VAULT_RECORD_ARTIFACT_SUFFIX}' layout" + )) + })? + .to_string(); + validate_key_id(&key_id)?; + if key_id == VAULT_RESTORE_MARKER_KEY { + return Err(BackupError::corrupted(format!( + "bundle names a key id colliding with the reserved restore marker path {VAULT_RESTORE_MARKER_KEY:?}" + )) + .into()); + } + Ok(key_id) +} + +fn record_kv_path(prefix: &str, key_id: &str) -> String { + format!("{prefix}/{key_id}") +} + +fn marker_kv_path(prefix: &str) -> String { + format!("{prefix}/{VAULT_RESTORE_MARKER_KEY}") +} + +/// Compare every coordinate that is decidable from the bundle and the caller's +/// target description alone: cluster identity, namespace, and all three mount +/// paths. Issues no request of any kind. +/// +/// Returns the Transit coordinates to probe only when they agree, so a caller +/// cannot accidentally probe a mount the bundle does not describe. +fn compare_coordinates<'a>( + references: &'a VaultExternalReferences, + target: &'a VaultRestoreTarget, +) -> (Vec, Option<(&'a VaultTransitReference, &'a str)>) { + let mut mismatches = Vec::new(); + + if references.cluster_id != target.cluster_id { + mismatches.push(VaultRestoreMismatch::ClusterIdentity { + expected: references.cluster_id.clone(), + observed: target.cluster_id.clone(), + }); + } + if references.namespace != target.namespace { + mismatches.push(VaultRestoreMismatch::Namespace { + expected: references.namespace.clone(), + observed: target.namespace.clone(), + }); + } + for (dependency, expected, observed) in [ + ("vault kv mount", &references.kv_mount, &target.kv_mount), + ("vault kv path prefix", &references.kv_path_prefix, &target.kv_path_prefix), + ] { + if expected != observed { + mismatches.push(VaultRestoreMismatch::Mount { + dependency: dependency.to_string(), + expected: expected.clone(), + observed: observed.clone(), + }); + } + } + + let probe = match (references.transit.as_ref(), target.transit_mount.as_deref()) { + (None, _) => None, + (Some(transit), Some(mount)) if transit.mount_path == mount => Some((transit, mount)), + (Some(transit), observed) => { + mismatches.push(VaultRestoreMismatch::Mount { + dependency: "vault transit mount".to_string(), + expected: transit.mount_path.clone(), + observed: observed.unwrap_or("").to_string(), + }); + None + } + }; + (mismatches, probe) +} + +/// Verify the recovered external trust root against the bundle's references. +/// +/// Locally decidable coordinates are settled first and, when any of them +/// disagrees, the verdict is returned without issuing a single Vault request: +/// a target whose coordinates already drifted is by definition not the one the +/// bundle describes, and probing it would both read through the wrong mount +/// and touch a deployment this restore has no business talking to. Only once +/// they all agree does the Transit read happen. Reads only. +async fn verify_trust_root( + client: &VaultRestoreClient, + references: &VaultExternalReferences, + target: &VaultRestoreTarget, +) -> Result> { + let (mut mismatches, probe) = compare_coordinates(references, target); + if !mismatches.is_empty() { + return Ok(mismatches); + } + // Every remaining check needs the network; nothing above this line did. + let Some((transit, target_mount)) = probe else { + return Ok(mismatches); + }; + + match client.read_transit_key(target_mount, &transit.key_name).await? { + None => mismatches.push(VaultRestoreMismatch::TransitKeyMissing { + mount_path: target_mount.to_string(), + key_name: transit.key_name.clone(), + }), + Some(observed) => { + if observed.min_decryption_version > u64::from(transit.required_min_version) { + mismatches.push(VaultRestoreMismatch::TransitMinDecryptionVersionTooHigh { + key_name: transit.key_name.clone(), + required: transit.required_min_version, + observed: observed.min_decryption_version, + }); + } + if observed.latest_version < transit.current_version { + mismatches.push(VaultRestoreMismatch::TransitVersionHistoryRegressed { + key_name: transit.key_name.clone(), + expected_current: transit.current_version, + observed_latest: observed.latest_version, + }); + } + } + } + Ok(mismatches) +} + +/// What the restore must do with one bundled record. +enum RecordPlan { + /// The target holds nothing at this path; publish it. + Publish, + /// The target already holds exactly this content — either it pre-existed + /// or an interrupted attempt wrote it. Either way, nothing to do. + AlreadyApplied, + /// The target holds different content and the restore must not overwrite. + Conflict(RestoreConflict), + /// The target's Vault sits at an earlier point than the bundle. + Mismatch(VaultRestoreMismatch), +} + +/// Classify one record against the target, reading only. +/// +/// Content equality is checked before generations: after a crash the records +/// this restore already wrote sit at generation 1 while the bundle references +/// a much higher one, and treating that as a rollback would make roll-forward +/// impossible. +async fn plan_record(client: &VaultRestoreClient, prefix: &str, record: &DecodedVaultRecord) -> Result { + let path = record_kv_path(prefix, &record.key_id); + let Some(observed) = client.read_kv_record(&path).await? else { + return Ok(RecordPlan::Publish); + }; + if observed.content == record.content { + return Ok(RecordPlan::AlreadyApplied); + } + + Ok(match observed.generation.cmp(&record.kv_version) { + std::cmp::Ordering::Greater => RecordPlan::Conflict(RestoreConflict { + key_id: record.key_id.clone(), + kind: RestoreConflictKind::VersionRegression, + detail: format!( + "target holds generation {} of key '{}', ahead of the bundle's generation {}; \ + restoring would roll it back", + observed.generation, record.key_id, record.kv_version + ), + }), + std::cmp::Ordering::Less => RecordPlan::Mismatch(VaultRestoreMismatch::KvGenerationRegressed { + key_id: record.key_id.clone(), + expected: record.kv_version, + observed: observed.generation, + }), + std::cmp::Ordering::Equal => RecordPlan::Conflict(RestoreConflict { + key_id: record.key_id.clone(), + kind: RestoreConflictKind::KeyAlreadyExists, + detail: format!( + "target holds generation {} of key '{}' with content that differs from the bundle's", + observed.generation, record.key_id + ), + }), + }) +} + +async fn read_marker(client: &VaultRestoreClient, prefix: &str) -> Result> { + match client.read_kv_record(&marker_kv_path(prefix)).await? { + Some(record) => Ok(Some(PublishedMarker { + marker: VaultRestoreCommitMarker::from_value(record.content)?, + kv_version: record.generation, + })), + None => Ok(None), + } +} + +/// Zero-write restore preflight: evaluate the bundle against the target and +/// report every blocker, conflict, and external dependency mismatch. +/// +/// The bundle is decoded fully in memory (including AEAD verification of every +/// artifact); the target Vault is only read. Environment failures surface as +/// errors, verdicts about the bundle or the target surface inside the report. +pub async fn dry_run_vault_restore( + kek: &BackupKek, + client: &VaultRestoreClient, + request: &VaultRestoreRequest, +) -> Result { + request.validate()?; + let mut report = RestoreDryRunReport { + backup_id: String::new(), + target_deployment_identity: request.target_deployment_identity.clone(), + blockers: Vec::new(), + conflicts: Vec::new(), + external_mismatches: Vec::new(), + }; + + let decoded = match decode_vault_bundle(&request.bundle_dir, kek).await { + Ok(decoded) => decoded, + Err(error) => match blocker_for(&error) { + Some(blocker) => { + report.blockers.push(blocker); + return Ok(report); + } + None => return Err(error), + }, + }; + report.backup_id = decoded.manifest.backup_id.clone(); + + if decoded.manifest.deployment_identity != request.target_deployment_identity { + report.blockers.push(RestoreBlocker { + code: RestoreBlockerCode::DeploymentMismatch, + detail: format!( + "bundle was produced by deployment '{}', restore target is '{}'", + decoded.manifest.deployment_identity, request.target_deployment_identity + ), + }); + } + if let Some(observed) = request.observed_generation + && decoded.manifest.snapshot_generation < observed + { + report.conflicts.push(RestoreConflict { + key_id: WHOLE_TARGET_CONFLICT_ID.to_string(), + kind: RestoreConflictKind::GenerationRegression, + detail: format!( + "target observed snapshot generation {observed}, bundle carries {}; equal generations stay allowed for repeated drills", + decoded.manifest.snapshot_generation + ), + }); + } + + let mismatches = verify_trust_root(client, &decoded.references, &request.target).await?; + let trust_root_intact = mismatches.is_empty(); + for mismatch in &mismatches { + mismatch.record_in(&mut report); + } + + // Probing per-key state through a mount set the bundle does not describe + // would report conflicts about unrelated records, so the target-state + // section is only meaningful once the coordinates agree. + if !trust_root_intact { + return Ok(report); + } + + let prefix = request.target.kv_path_prefix.as_str(); + let marker = match read_marker(client, prefix).await { + Ok(marker) => marker, + Err(error) => { + report.conflicts.push(RestoreConflict { + key_id: VAULT_RESTORE_MARKER_KEY.to_string(), + kind: RestoreConflictKind::KeyAlreadyExists, + detail: format!("target holds an undecodable restore marker: {error}"), + }); + return Ok(report); + } + }; + if let Some(published) = &marker + && !published.marker.matches_bundle(&decoded.manifest) + { + report.conflicts.push(RestoreConflict { + key_id: VAULT_RESTORE_MARKER_KEY.to_string(), + kind: RestoreConflictKind::KeyAlreadyExists, + detail: format!("target holds a restore marker for a different bundle ('{}')", published.marker.backup_id), + }); + return Ok(report); + } + + for record in selected_records(&decoded, marker.as_ref().map(|published| &published.marker)) { + match plan_record(client, prefix, record).await? { + RecordPlan::Publish | RecordPlan::AlreadyApplied => {} + RecordPlan::Conflict(conflict) => report.conflicts.push(conflict), + RecordPlan::Mismatch(mismatch) => mismatch.record_in(&mut report), + } + } + Ok(report) +} + +/// The records an attempt is responsible for: everything in the bundle on a +/// fresh run, and exactly what a published marker names on a re-entry — a +/// record the first run found already present must stay untouched. +fn selected_records<'a>( + decoded: &'a DecodedVaultBundle, + marker: Option<&VaultRestoreCommitMarker>, +) -> Vec<&'a DecodedVaultRecord> { + match marker { + Some(marker) => decoded + .records + .iter() + .filter(|record| marker.key_ids.contains(&record.key_id)) + .collect(), + None => decoded.records.iter().collect(), + } +} + +/// Restore a sealed Vault bundle into the target deployment. +/// +/// Fail-fast counterpart of the dry-run: the first violated precondition +/// aborts with its typed error. Only [`RestoreConflictPolicy::RestoreIntoEmptyTarget`] +/// ever writes, and even then every write is create-only. +pub async fn restore_vault_backup( + kek: &BackupKek, + client: &VaultRestoreClient, + request: &VaultRestoreRequest, +) -> Result { + request.validate()?; + if request.conflict_policy == RestoreConflictPolicy::Fail { + return Err(KmsError::invalid_operation( + "restore conflict policy 'fail' never writes to the target; review the dry-run report and \ + re-run with the 'restore-into-empty-target' policy to proceed", + )); + } + + let decoded = decode_vault_bundle(&request.bundle_dir, kek).await?; + if decoded.manifest.deployment_identity != request.target_deployment_identity { + return Err(KmsError::invalid_operation(format!( + "bundle was produced by deployment '{}' but the restore target is '{}'", + decoded.manifest.deployment_identity, request.target_deployment_identity + ))); + } + if let Some(observed) = request.observed_generation + && decoded.manifest.snapshot_generation < observed + { + return Err(KmsError::invalid_operation(format!( + "restore would regress the snapshot generation: target observed {observed}, bundle carries {}", + decoded.manifest.snapshot_generation + ))); + } + + let mut sequence = VaultRestoreSequence::new(); + + // Stage 1: the external trust root must be back and provably the right + // one before any RustFS-side state is written. + sequence.enter(VaultRestoreStage::ExternalTrustRoot)?; + if let Some(mismatch) = verify_trust_root(client, &decoded.references, &request.target) + .await? + .into_iter() + .next() + { + return Err(mismatch.into()); + } + sequence.complete(VaultRestoreStage::ExternalTrustRoot)?; + + let prefix = request.target.kv_path_prefix.as_str(); + let existing_marker = read_marker(client, prefix).await?; + if let Some(published) = &existing_marker + && !published.marker.matches_bundle(&decoded.manifest) + { + return Err(KmsError::invalid_operation(format!( + "target holds a restore marker for a different bundle ('{}'); roll that restore forward \ + with its own bundle or abort it explicitly", + published.marker.backup_id + ))); + } + + let mut plans = Vec::new(); + for record in selected_records(&decoded, existing_marker.as_ref().map(|published| &published.marker)) { + match plan_record(client, prefix, record).await? { + RecordPlan::Publish => plans.push((record, true)), + RecordPlan::AlreadyApplied => plans.push((record, false)), + RecordPlan::Conflict(conflict) => { + return Err(KmsError::invalid_operation(format!( + "restore refused, {:?} conflict on key '{}': {}", + conflict.kind, conflict.key_id, conflict.detail + ))); + } + RecordPlan::Mismatch(mismatch) => return Err(mismatch.into()), + } + } + + // The marker is the commit point: it names exactly the records this + // restore is responsible for, computed before the first write and never + // recomputed on re-entry. + let (mut marker, mut marker_version, resumed) = match existing_marker { + Some(published) => (published.marker, published.kv_version, true), + None => { + let key_ids = plans + .iter() + .filter(|(_, publish)| *publish) + .map(|(record, _)| record.key_id.clone()) + .collect(); + let marker = VaultRestoreCommitMarker::new(&decoded.manifest, key_ids); + let Some(version) = client + .create_kv_record(&marker_kv_path(prefix), &marker.to_content()?) + .await? + else { + return Err(KmsError::invalid_operation( + "another restore published a commit marker for this target concurrently; \ + resolve it before retrying", + )); + }; + (marker, version, false) + } + }; + + for stage in [VaultRestoreStage::MaterialAndVersions, VaultRestoreStage::KvMetadata] { + sequence.enter(stage)?; + for (record, publish) in plans.iter().filter(|(record, _)| record.stage == stage) { + if !*publish || !marker.key_ids.contains(&record.key_id) { + continue; + } + let Some(kv_version) = client + .create_kv_record(&record_kv_path(prefix, &record.key_id), &record.content) + .await? + else { + return Err(KmsError::invalid_operation(format!( + "a concurrent writer created the Vault record for key '{}' during restore; \ + restore never overwrites, so it stopped before touching it", + record.key_id + ))); + }; + // Record the receipt before moving on: it is what later lets an + // abort tell this restore's record apart from one that merely sits + // at the same path, and it is the one write here that is not + // create-only. Crashing between the two leaves the record without + // a receipt, which an abort then declines to remove — the safe + // direction, and roll-forward still recognizes it by content. + marker.written.push(VaultRestoreWriteReceipt { + key_id: record.key_id.clone(), + kv_version, + }); + marker_version = client + .update_kv_record(&marker_kv_path(prefix), &marker.to_content()?, marker_version) + .await?; + } + sequence.complete(stage)?; + } + + // Stage 4: the cutover completes when the marker is gone. Applying the + // sanitized configuration artifact belongs to the admin layer. + sequence.enter(VaultRestoreStage::ConfigAndCutover)?; + client.delete_kv_record(&marker_kv_path(prefix)).await?; + sequence.complete(VaultRestoreStage::ConfigAndCutover)?; + + let already_present_key_ids = decoded + .records + .iter() + .map(|record| record.key_id.clone()) + .filter(|key_id| !marker.key_ids.contains(key_id)) + .collect(); + Ok(VaultRestoreReport { + backup_id: decoded.manifest.backup_id.clone(), + snapshot_generation: decoded.manifest.snapshot_generation, + restored_key_ids: marker.key_ids, + already_present_key_ids, + transit_key_name: decoded.references.transit.map(|transit| transit.key_name), + resumed, + }) +} + +/// Explicitly abort an interrupted Vault restore, returning the target to its +/// pre-restore state. +/// +/// Removes only records the marker's write receipts prove this restore created +/// and that still sit at the version it wrote them at; everything else the +/// marker merely intended to write is left in place and reported. The marker +/// itself goes last, so an abort that leaves records behind still clears the +/// in-progress state. Fails closed on a marker it cannot strictly decode. +pub async fn abort_vault_restore(client: &VaultRestoreClient, target: &VaultRestoreTarget) -> Result { + let prefix = target.kv_path_prefix.as_str(); + let Some(published) = read_marker(client, prefix).await? else { + return Err(KmsError::invalid_operation(format!( + "no vault restore is in progress under {}/{prefix}", + target.kv_mount + ))); + }; + let marker = published.marker; + + let mut removed_key_ids = Vec::new(); + let mut skipped = Vec::new(); + for key_id in &marker.key_ids { + let Some(written) = marker.receipt_for(key_id) else { + skipped.push(VaultRestoreAbortSkip { + key_id: key_id.clone(), + reason: VaultRestoreAbortSkipReason::NeverWritten, + }); + continue; + }; + let path = record_kv_path(prefix, key_id); + let observed = client.read_kv_generation(&path).await?; + if observed != Some(written) { + skipped.push(VaultRestoreAbortSkip { + key_id: key_id.clone(), + reason: VaultRestoreAbortSkipReason::NotAtWrittenVersion { written, observed }, + }); + continue; + } + client.delete_kv_record(&path).await?; + removed_key_ids.push(key_id.clone()); + } + + client.delete_kv_record(&marker_kv_path(prefix)).await?; + Ok(VaultRestoreAbortReport { + backup_id: marker.backup_id, + removed_key_ids, + skipped, + }) +} + +/// Serializable outcome of an abort. Identifiers and versions only. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct VaultRestoreAbortReport { + /// Bundle the aborted restore was publishing. + pub backup_id: String, + /// Records the abort proved this restore wrote, and took back. + pub removed_key_ids: Vec, + /// Records the marker named that the abort deliberately left in place. + /// A non-empty list means the target still needs an operator's eye. + pub skipped: Vec, +} + +/// One record an abort refused to remove, and why. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct VaultRestoreAbortSkip { + /// Key whose record was left in place. + pub key_id: String, + /// Why ownership could not be proven. + pub reason: VaultRestoreAbortSkipReason, +} + +/// Why an abort could not prove it owns a record the marker names. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum VaultRestoreAbortSkipReason { + /// The marker carries no write receipt: the restore never created this + /// record, so whatever sits at the path is somebody else's. + NeverWritten, + /// The path no longer holds the version the restore wrote — it was + /// modified, or removed, after the fact. `observed` is `None` when the + /// path holds nothing at all. + NotAtWrittenVersion { written: u64, observed: Option }, +} + +fn blocker_for(error: &KmsError) -> Option { + match error { + KmsError::Backup(inner) => Some(RestoreBlocker::from(inner)), + KmsError::UnsupportedCapability { .. } => Some(RestoreBlocker { + code: RestoreBlockerCode::UnsupportedBackend, + detail: error.to_string(), + }), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::backends::scripted_vault::{ScriptedResponse, ScriptedVault}; + use crate::backup::capability::{AtRestProtection, BackupBackendKind, BackupResponsibility}; + use crate::backup::local_export::{AEAD_NONCE_LEN, LOCAL_BUNDLE_MANIFEST_FILE, artifact_aad}; + use crate::backup::manifest::{ + AeadAlgorithm, ArtifactDescriptor, CompletenessState, DigestAlgorithm, VaultKvRecordReference, VaultTransitReference, + }; + use aes_gcm::{ + Nonce, + aead::{Aead, Payload}, + }; + use std::time::Duration; + use tempfile::TempDir; + + const DEPLOYMENT: &str = "deployment-test"; + const BACKUP_ID: &str = "backup-0001"; + const SNAPSHOT_GENERATION: u64 = 7; + const CLUSTER_ID: &str = "vault-cluster-a"; + const KV_MOUNT: &str = "secret"; + const KV_PREFIX: &str = "rustfs/kms/transit-metadata"; + const TRANSIT_MOUNT: &str = "transit"; + const TRANSIT_KEY: &str = "rustfs-master"; + const KEY_ID: &str = "object-key"; + const KV_GENERATION: u64 = 4; + + /// One way a restore target can drift from the bundle's coordinates. + type TargetDrift = fn(&mut VaultRestoreTarget); + + fn test_kek() -> BackupKek { + BackupKek::new("backup-kek-test", 1, [0x42; 32]).expect("kek") + } + + fn record_content() -> serde_json::Value { + serde_json::json!({ "key_state": "Enabled", "current_version": 3 }) + } + + fn references() -> VaultExternalReferences { + VaultExternalReferences { + cluster_id: CLUSTER_ID.to_string(), + namespace: None, + kv_mount: KV_MOUNT.to_string(), + kv_path_prefix: KV_PREFIX.to_string(), + transit: Some(VaultTransitReference { + mount_path: TRANSIT_MOUNT.to_string(), + key_name: TRANSIT_KEY.to_string(), + required_min_version: 2, + current_version: 5, + native_snapshot_reference: None, + }), + kv_records: vec![VaultKvRecordReference { + key_id: KEY_ID.to_string(), + kv_version: KV_GENERATION, + }], + } + } + + /// Seal a Vault bundle on disk the way a Vault export will: artifacts + /// first, sealed manifest last. + async fn write_bundle(dir: &Path, references: VaultExternalReferences, records: &[(&str, serde_json::Value)]) -> PathBuf { + let bundle = dir.join("bundle"); + tokio::fs::create_dir_all(bundle.join(VAULT_RECORD_ARTIFACT_DIR)) + .await + .expect("artifact dir"); + + let kek = test_kek(); + let mut artifacts = Vec::new(); + for (key_id, content) in records { + let path = format!("{VAULT_RECORD_ARTIFACT_DIR}/{key_id}{VAULT_RECORD_ARTIFACT_SUFFIX}"); + let plaintext = serde_json::to_vec(content).expect("serialize record"); + let nonce = [0x11u8; AEAD_NONCE_LEN]; + let ciphertext = kek + .cipher() + .encrypt( + &Nonce::from(nonce), + Payload { + msg: &plaintext, + aad: &artifact_aad(BACKUP_ID, SNAPSHOT_GENERATION, &path), + }, + ) + .expect("seal artifact"); + let mut payload = nonce.to_vec(); + payload.extend_from_slice(&ciphertext); + tokio::fs::write(bundle.join(&path), &payload).await.expect("write artifact"); + artifacts.push(ArtifactDescriptor { + kind: ArtifactKind::KeyMetadata, + path, + len: payload.len() as u64, + aead_algorithm: AeadAlgorithm::Aes256Gcm, + encrypted_digest: ContentDigest::sha256_of(&payload), + }); + } + + let manifest = BackupManifest { + format_version: BackupManifest::FORMAT_VERSION, + backup_id: BACKUP_ID.to_string(), + created_at: "2026-08-01T00:00:00+00:00[UTC]".parse().expect("timestamp"), + rustfs_version: "1.0.0-test".to_string(), + deployment_identity: DEPLOYMENT.to_string(), + backend: BackupBackendKind::VaultTransit, + at_rest_protection: AtRestProtection::ExternalNonExportable, + responsibility: BackupResponsibility::MetadataPlusExternalRoot, + snapshot_generation: SNAPSHOT_GENERATION, + backup_kek: kek.descriptor(), + artifacts, + local_kdf: None, + external_references: Some(references), + key_versions: None, + capability_discovery: None, + completeness: CompletenessState::InProgress, + manifest_digest: ContentDigest { + algorithm: DigestAlgorithm::Sha256, + hex: String::new(), + }, + }; + let sealed = manifest.seal().expect("seal manifest"); + tokio::fs::write(bundle.join(LOCAL_BUNDLE_MANIFEST_FILE), sealed.encode().expect("encode manifest")) + .await + .expect("write manifest"); + bundle + } + + async fn standard_bundle(dir: &Path) -> PathBuf { + write_bundle(dir, references(), &[(KEY_ID, record_content())]).await + } + + fn target(address: &str) -> VaultRestoreTarget { + VaultRestoreTarget { + address: address.to_string(), + auth_method: VaultAuthMethod::Token { + token: "scripted-token".to_string(), + }, + namespace: None, + cluster_id: CLUSTER_ID.to_string(), + kv_mount: KV_MOUNT.to_string(), + kv_path_prefix: KV_PREFIX.to_string(), + transit_mount: Some(TRANSIT_MOUNT.to_string()), + } + } + + async fn scripted_client(responses: Vec) -> (ScriptedVault, VaultRestoreClient, VaultRestoreTarget) { + let vault = ScriptedVault::serve(responses).await; + let target = target(&vault.address); + let kms_config = KmsConfig { + timeout: Duration::from_secs(5), + retry_attempts: 1, + ..KmsConfig::default() + }; + let client = VaultRestoreClient::connect(&target, &kms_config) + .await + .expect("scripted restore client"); + (vault, client, target) + } + + fn request(bundle_dir: PathBuf, target: VaultRestoreTarget, policy: RestoreConflictPolicy) -> VaultRestoreRequest { + VaultRestoreRequest { + bundle_dir, + target, + target_deployment_identity: DEPLOYMENT.to_string(), + observed_generation: None, + conflict_policy: policy, + } + } + + /// Transit key read payload with the given version window. + fn transit_key_response(min_decryption_version: u64, latest_version: u32) -> ScriptedResponse { + let keys: serde_json::Map = (1..=latest_version) + .map(|version| (version.to_string(), serde_json::json!(1_700_000_000u64))) + .collect(); + ScriptedResponse::ok(serde_json::json!({ + "type": "aes256-gcm96", + "deletion_allowed": false, + "derived": false, + "exportable": false, + "allow_plaintext_backup": false, + "keys": keys, + "min_decryption_version": min_decryption_version, + "min_encryption_version": 0, + "name": TRANSIT_KEY, + "supports_encryption": true, + "supports_decryption": true, + "supports_derivation": true, + "supports_signing": false, + })) + } + + fn healthy_transit_key() -> ScriptedResponse { + transit_key_response(1, 5) + } + + fn kv_metadata_response(current_version: u64) -> ScriptedResponse { + ScriptedResponse::ok(serde_json::json!({ + "cas_required": false, + "created_time": "2026-08-01T00:00:00Z", + "current_version": current_version, + "delete_version_after": "0s", + "max_versions": 0, + "oldest_version": 0, + "updated_time": "2026-08-01T00:00:00Z", + "custom_metadata": null, + "versions": {}, + })) + } + + fn kv_read_response(content: serde_json::Value, version: u64) -> ScriptedResponse { + ScriptedResponse::ok(serde_json::json!({ + "data": content, + "metadata": { + "created_time": "2026-08-01T00:00:00Z", + "deletion_time": "", + "custom_metadata": null, + "destroyed": false, + "version": version, + }, + })) + } + + fn kv_write_response_at(version: u64) -> ScriptedResponse { + ScriptedResponse::ok(serde_json::json!({ + "created_time": "2026-08-01T00:00:00Z", + "deletion_time": "", + "custom_metadata": null, + "destroyed": false, + "version": version, + })) + } + + /// The response a create-only write gets: a fresh path lands at version 1. + fn kv_write_response() -> ScriptedResponse { + kv_write_response_at(1) + } + + fn not_found() -> ScriptedResponse { + ScriptedResponse::error(404, "not found") + } + + fn empty_ok() -> ScriptedResponse { + ScriptedResponse::ok(serde_json::json!({})) + } + + /// A marker as the target would hold it: `key_ids` is what the restore set + /// out to publish, `written` the receipts for what it actually created. + fn marker_value(key_ids: &[&str], written: &[(&str, u64)], manifest_digest: &ContentDigest) -> serde_json::Value { + serde_json::json!({ + "format_version": VAULT_RESTORE_MARKER_FORMAT_VERSION, + "backup_id": BACKUP_ID, + "manifest_digest": manifest_digest, + "key_ids": key_ids, + "written": written + .iter() + .map(|(key_id, kv_version)| serde_json::json!({ "key_id": key_id, "kv_version": kv_version })) + .collect::>(), + }) + } + + fn test_digest() -> ContentDigest { + ContentDigest { + algorithm: DigestAlgorithm::Sha256, + hex: "a".repeat(64), + } + } + + async fn bundle_manifest(bundle: &Path) -> BackupManifest { + read_bundle_manifest(bundle).await.expect("manifest") + } + + #[test] + fn stage_order_is_enforced() { + let mut sequence = VaultRestoreSequence::new(); + assert_eq!(sequence.expected(), Some(VaultRestoreStage::ExternalTrustRoot)); + + // Skipping ahead, repeating, and completing an unopened stage all fail. + sequence + .enter(VaultRestoreStage::MaterialAndVersions) + .expect_err("material may not run before the trust root"); + sequence + .complete(VaultRestoreStage::ExternalTrustRoot) + .expect_err("a stage that was never entered may not complete"); + + sequence.enter(VaultRestoreStage::ExternalTrustRoot).expect("first stage"); + sequence + .enter(VaultRestoreStage::ExternalTrustRoot) + .expect_err("a stage may not be entered twice"); + sequence.complete(VaultRestoreStage::ExternalTrustRoot).expect("complete"); + + sequence + .enter(VaultRestoreStage::KvMetadata) + .expect_err("metadata may not run before material"); + for stage in [ + VaultRestoreStage::MaterialAndVersions, + VaultRestoreStage::KvMetadata, + VaultRestoreStage::ConfigAndCutover, + ] { + sequence.enter(stage).expect("in-order stage"); + sequence.complete(stage).expect("in-order completion"); + } + assert_eq!(sequence.expected(), None); + sequence + .enter(VaultRestoreStage::ExternalTrustRoot) + .expect_err("the sequence does not restart"); + } + + #[tokio::test] + async fn dry_run_writes_nothing_and_reports_a_clean_target() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (vault, client, target) = scripted_client(vec![ + healthy_transit_key(), + not_found(), // marker absent + not_found(), // key record absent + ]) + .await; + + let report = dry_run_vault_restore(&test_kek(), &client, &request(bundle, target, RestoreConflictPolicy::Fail)) + .await + .expect("dry-run should succeed"); + assert!(report.restore_permitted(), "clean target must permit restore: {report:?}"); + + // The zero-write contract, asserted at the wire: a dry-run may only + // ever read. + for line in vault.requests() { + assert!( + line.starts_with("GET ") || line.starts_with("LIST "), + "dry-run issued a non-read request: {line}" + ); + } + } + + #[tokio::test] + async fn dry_run_classifies_a_missing_transit_key() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (_vault, client, target) = scripted_client(vec![not_found()]).await; + + let report = dry_run_vault_restore(&test_kek(), &client, &request(bundle, target, RestoreConflictPolicy::Fail)) + .await + .expect("dry-run should succeed"); + assert!(!report.restore_permitted()); + assert_eq!(report.blockers.len(), 1); + assert_eq!(report.blockers[0].code, RestoreBlockerCode::ExternalDependencyUnavailable); + assert!(report.blockers[0].detail.contains(TRANSIT_KEY)); + } + + #[tokio::test] + async fn dry_run_classifies_a_raised_min_decryption_version() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + // The bundle still needs version 2; the recovered key refuses anything + // below 4. + let (_vault, client, target) = scripted_client(vec![transit_key_response(4, 5)]).await; + + let report = dry_run_vault_restore(&test_kek(), &client, &request(bundle, target, RestoreConflictPolicy::Fail)) + .await + .expect("dry-run should succeed"); + assert!(!report.restore_permitted()); + assert_eq!(report.external_mismatches.len(), 1); + assert_eq!(report.external_mismatches[0].expected, "min_decryption_version<=2"); + assert_eq!(report.external_mismatches[0].observed, "min_decryption_version=4"); + } + + #[tokio::test] + async fn dry_run_classifies_a_regressed_transit_version_history() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + // Native restore landed on a snapshot with only 3 of the 5 versions. + let (_vault, client, target) = scripted_client(vec![transit_key_response(1, 3)]).await; + + let report = dry_run_vault_restore(&test_kek(), &client, &request(bundle, target, RestoreConflictPolicy::Fail)) + .await + .expect("dry-run should succeed"); + assert!(!report.restore_permitted()); + assert_eq!(report.external_mismatches[0].expected, "latest_version>=5"); + assert_eq!(report.external_mismatches[0].observed, "latest_version=3"); + } + + #[tokio::test] + async fn dry_run_classifies_namespace_and_mount_drift() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + // No Vault call at all: coordinate drift is decided before any probe. + let (vault, client, mut target) = scripted_client(Vec::new()).await; + target.namespace = Some("tenant-b".to_string()); + target.kv_mount = "other-kv".to_string(); + target.transit_mount = Some("other-transit".to_string()); + + let report = dry_run_vault_restore(&test_kek(), &client, &request(bundle, target, RestoreConflictPolicy::Fail)) + .await + .expect("dry-run should succeed"); + assert!(!report.restore_permitted()); + let dependencies: Vec<&str> = report + .external_mismatches + .iter() + .map(|mismatch| mismatch.dependency.as_str()) + .collect(); + assert_eq!(dependencies, ["vault namespace", "vault kv mount", "vault transit mount"]); + assert!(vault.requests().is_empty(), "coordinate drift must not probe the target"); + } + + #[tokio::test] + async fn every_locally_decidable_drift_is_settled_without_a_vault_request() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + + // Each of these leaves the Transit mount agreeing with the bundle, so + // only the ordering of the checks — coordinates first, network second + // — can keep the Transit read from going out. + let drifts: [(&str, TargetDrift); 4] = [ + ("vault cluster identity", |target| target.cluster_id = "vault-cluster-b".to_string()), + ("vault namespace", |target| target.namespace = Some("tenant-b".to_string())), + ("vault kv mount", |target| target.kv_mount = "other-kv".to_string()), + ("vault kv path prefix", |target| { + target.kv_path_prefix = "rustfs/kms/elsewhere".to_string() + }), + ]; + + for (dependency, drift) in drifts { + let (vault, client, mut target) = scripted_client(Vec::new()).await; + drift(&mut target); + + let report = + dry_run_vault_restore(&test_kek(), &client, &request(bundle.clone(), target, RestoreConflictPolicy::Fail)) + .await + .expect("dry-run should report rather than fail"); + + assert!(!report.restore_permitted(), "{dependency} drift must refuse the restore"); + let dependencies: Vec<&str> = report + .external_mismatches + .iter() + .map(|mismatch| mismatch.dependency.as_str()) + .collect(); + assert_eq!(dependencies, [dependency]); + assert!( + vault.requests().is_empty(), + "{dependency} drift probed the target: {:?}", + vault.requests() + ); + } + } + + #[tokio::test] + async fn dry_run_classifies_a_rolled_back_kv_generation() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (_vault, client, target) = scripted_client(vec![ + healthy_transit_key(), + not_found(), // marker absent + kv_metadata_response(2), + kv_read_response(serde_json::json!({ "key_state": "Enabled", "current_version": 1 }), 2), + ]) + .await; + + let report = dry_run_vault_restore(&test_kek(), &client, &request(bundle, target, RestoreConflictPolicy::Fail)) + .await + .expect("dry-run should succeed"); + assert!(!report.restore_permitted()); + assert_eq!(report.external_mismatches[0].dependency, format!("vault kv record for key {KEY_ID}")); + assert_eq!(report.external_mismatches[0].expected, "kv_generation>=4"); + assert_eq!(report.external_mismatches[0].observed, "kv_generation=2"); + } + + #[tokio::test] + async fn dry_run_reports_a_target_that_moved_ahead_of_the_bundle() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (_vault, client, target) = scripted_client(vec![ + healthy_transit_key(), + not_found(), + kv_metadata_response(9), + kv_read_response(serde_json::json!({ "key_state": "Disabled", "current_version": 8 }), 9), + ]) + .await; + + let report = dry_run_vault_restore(&test_kek(), &client, &request(bundle, target, RestoreConflictPolicy::Fail)) + .await + .expect("dry-run should succeed"); + assert!(!report.restore_permitted()); + assert_eq!(report.conflicts[0].kind, RestoreConflictKind::VersionRegression); + assert_eq!(report.conflicts[0].key_id, KEY_ID); + } + + #[tokio::test] + async fn fail_policy_never_writes() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (vault, client, target) = scripted_client(Vec::new()).await; + + restore_vault_backup(&test_kek(), &client, &request(bundle, target, RestoreConflictPolicy::Fail)) + .await + .expect_err("the default policy refuses to write"); + assert!(vault.requests().is_empty(), "the refusal must precede every Vault call"); + } + + #[tokio::test] + async fn restore_publishes_records_in_stage_order_and_clears_the_marker() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (vault, client, target) = scripted_client(vec![ + healthy_transit_key(), + not_found(), // marker absent + not_found(), // key record absent + kv_write_response(), // marker published + kv_write_response(), // record published + kv_write_response_at(2), // write receipt appended to the marker + empty_ok(), // marker removed + ]) + .await; + + let report = restore_vault_backup( + &test_kek(), + &client, + &request(bundle, target, RestoreConflictPolicy::RestoreIntoEmptyTarget), + ) + .await + .expect("restore should succeed"); + assert_eq!(report.restored_key_ids, [KEY_ID]); + assert!(report.already_present_key_ids.is_empty()); + assert!(!report.resumed); + assert_eq!(report.transit_key_name.as_deref(), Some(TRANSIT_KEY)); + + let marker_path = format!("POST /v1/{KV_MOUNT}/data/{KV_PREFIX}/{VAULT_RESTORE_MARKER_KEY}"); + let record_path = format!("POST /v1/{KV_MOUNT}/data/{KV_PREFIX}/{KEY_ID}"); + let requests = vault.requests(); + assert_eq!( + requests, + [ + format!("GET /v1/{TRANSIT_MOUNT}/keys/{TRANSIT_KEY}"), + format!("GET /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{VAULT_RESTORE_MARKER_KEY}"), + format!("GET /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{KEY_ID}"), + marker_path.clone(), + record_path.clone(), + marker_path, + format!("DELETE /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{VAULT_RESTORE_MARKER_KEY}"), + ], + "the trust root is verified first, the marker brackets every write, \ + and each published record is receipted before the run moves on" + ); + + let writes: Vec<(String, serde_json::Value)> = requests + .iter() + .zip(vault.request_bodies()) + .filter(|(_, body)| !body.is_empty()) + .map(|(line, body)| (line.clone(), serde_json::from_str(&body).expect("write body is JSON"))) + .collect(); + + // Record writes are create-only: restore can structurally not + // overwrite one. The marker is the sole exception, and even it moves + // only by check-and-set against the version just observed. + for (line, body) in &writes { + let expected_cas = if *line == record_path || body["data"]["written"] == serde_json::json!([]) { + 0 + } else { + 1 + }; + assert_eq!(body["options"]["cas"], expected_cas, "unexpected check-and-set on {line}: {body}"); + } + + let receipt = &writes.last().expect("the marker receipt is the last write").1; + assert_eq!( + receipt["data"]["written"], + serde_json::json!([{ "key_id": KEY_ID, "kv_version": 1 }]), + "the marker must record the version the record write landed at" + ); + } + + #[tokio::test] + async fn restore_refuses_the_trust_root_before_touching_kv() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (vault, client, target) = scripted_client(vec![transit_key_response(4, 5)]).await; + + let error = restore_vault_backup( + &test_kek(), + &client, + &request(bundle, target, RestoreConflictPolicy::RestoreIntoEmptyTarget), + ) + .await + .expect_err("a raised min_decryption_version must abort the restore"); + assert!(error.to_string().contains("min_decryption_version"), "{error}"); + assert_eq!( + vault.requests(), + [format!("GET /v1/{TRANSIT_MOUNT}/keys/{TRANSIT_KEY}")], + "the restore must stop at the trust-root stage" + ); + } + + #[tokio::test] + async fn a_lost_check_and_set_race_stops_the_restore() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (vault, client, target) = scripted_client(vec![ + healthy_transit_key(), + not_found(), + not_found(), + kv_write_response(), + ScriptedResponse::error(400, "check-and-set parameter did not match the current version"), + ]) + .await; + + let error = restore_vault_backup( + &test_kek(), + &client, + &request(bundle, target, RestoreConflictPolicy::RestoreIntoEmptyTarget), + ) + .await + .expect_err("a concurrent writer must stop the restore"); + assert!(error.to_string().contains("concurrent writer"), "{error}"); + // The write was attempted exactly once: a create-only write is never + // replayed, and the marker stays published for roll-forward or abort. + assert_eq!( + vault + .requests() + .iter() + .filter(|line| *line == &format!("POST /v1/{KV_MOUNT}/data/{KV_PREFIX}/{KEY_ID}")) + .count(), + 1 + ); + } + + #[tokio::test] + async fn re_entry_after_a_crash_between_writes_is_idempotent() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let manifest = bundle_manifest(&bundle).await; + // The interrupted attempt published the record and receipted it. + let marker = marker_value(&[KEY_ID], &[(KEY_ID, 1)], &manifest.manifest_digest); + let (vault, client, target) = scripted_client(vec![ + healthy_transit_key(), + kv_metadata_response(2), // marker present + kv_read_response(marker, 2), + kv_metadata_response(1), // the record this run already wrote + kv_read_response(record_content(), 1), + empty_ok(), // marker removed + ]) + .await; + + let report = restore_vault_backup( + &test_kek(), + &client, + &request(bundle, target, RestoreConflictPolicy::RestoreIntoEmptyTarget), + ) + .await + .expect("roll-forward should succeed"); + assert!(report.resumed); + assert_eq!(report.restored_key_ids, [KEY_ID]); + // The already-published record is recognized by content, not by + // generation: it sits at generation 1 while the bundle references 4. + assert!( + !vault.requests().iter().any(|line| line.starts_with("POST")), + "a completed record must not be rewritten: {:?}", + vault.requests() + ); + } + + #[tokio::test] + async fn a_marker_for_another_bundle_fails_closed() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let mut foreign = marker_value(&[KEY_ID], &[], &bundle_manifest(&bundle).await.manifest_digest); + foreign["backup_id"] = serde_json::json!("backup-other"); + let (_vault, client, target) = + scripted_client(vec![healthy_transit_key(), kv_metadata_response(1), kv_read_response(foreign, 1)]).await; + + let error = restore_vault_backup( + &test_kek(), + &client, + &request(bundle, target, RestoreConflictPolicy::RestoreIntoEmptyTarget), + ) + .await + .expect_err("a foreign marker must fail closed"); + assert!(error.to_string().contains("backup-other"), "{error}"); + } + + #[tokio::test] + async fn an_undecodable_marker_fails_closed() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (_vault, client, target) = scripted_client(vec![ + healthy_transit_key(), + kv_metadata_response(1), + kv_read_response(serde_json::json!({ "format_version": 99, "backup_id": BACKUP_ID }), 1), + ]) + .await; + + restore_vault_backup( + &test_kek(), + &client, + &request(bundle, target, RestoreConflictPolicy::RestoreIntoEmptyTarget), + ) + .await + .expect_err("an out-of-contract marker must fail closed"); + } + + #[tokio::test] + async fn abort_takes_back_exactly_the_records_it_receipted() { + let (vault, client, target) = scripted_client(vec![ + kv_metadata_response(2), + kv_read_response(marker_value(&[KEY_ID], &[(KEY_ID, 1)], &test_digest()), 2), + kv_metadata_response(1), // the record still sits at the version we wrote + empty_ok(), // record removed + empty_ok(), // marker removed + ]) + .await; + + let report = abort_vault_restore(&client, &target).await.expect("abort should succeed"); + assert_eq!(report.removed_key_ids, [KEY_ID]); + assert!(report.skipped.is_empty(), "{report:?}"); + assert_eq!( + vault.requests(), + [ + format!("GET /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{VAULT_RESTORE_MARKER_KEY}"), + format!("GET /v1/{KV_MOUNT}/data/{KV_PREFIX}/{VAULT_RESTORE_MARKER_KEY}"), + format!("GET /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{KEY_ID}"), + format!("DELETE /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{KEY_ID}"), + format!("DELETE /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{VAULT_RESTORE_MARKER_KEY}"), + ], + "ownership is re-checked at the path before anything is removed" + ); + } + + /// The marker names what a restore *meant* to write, which after a crash + /// or a lost race is not what it wrote. Neither case may be deleted. + #[tokio::test] + async fn abort_skips_every_record_it_cannot_prove_it_wrote() { + const CRASHED_BEFORE: &str = "second-key"; + const OVERWRITTEN: &str = "third-key"; + + let (vault, client, target) = scripted_client(vec![ + kv_metadata_response(3), + kv_read_response( + marker_value( + &[KEY_ID, CRASHED_BEFORE, OVERWRITTEN], + // No receipt for CRASHED_BEFORE: the restore died before + // reaching it. OVERWRITTEN was written, then moved on by + // somebody else. + &[(KEY_ID, 1), (OVERWRITTEN, 1)], + &test_digest(), + ), + 3, + ), + kv_metadata_response(1), // KEY_ID still at the version we wrote + empty_ok(), // KEY_ID removed + kv_metadata_response(4), // OVERWRITTEN has moved past our version + empty_ok(), // marker removed + ]) + .await; + + let report = abort_vault_restore(&client, &target).await.expect("abort should succeed"); + assert_eq!(report.removed_key_ids, [KEY_ID]); + assert_eq!( + report.skipped, + [ + VaultRestoreAbortSkip { + key_id: CRASHED_BEFORE.to_string(), + reason: VaultRestoreAbortSkipReason::NeverWritten, + }, + VaultRestoreAbortSkip { + key_id: OVERWRITTEN.to_string(), + reason: VaultRestoreAbortSkipReason::NotAtWrittenVersion { + written: 1, + observed: Some(4), + }, + }, + ] + ); + + let requests = vault.requests(); + let deletes: Vec<&str> = requests + .iter() + .filter(|line| line.starts_with("DELETE ")) + .map(String::as_str) + .collect(); + assert_eq!( + deletes, + [ + format!("DELETE /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{KEY_ID}"), + format!("DELETE /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{VAULT_RESTORE_MARKER_KEY}"), + ], + "only the receipted, unchanged record and the marker may be removed" + ); + } + + /// The review case: the marker is published, the record write then loses + /// the create-only race, and the abort that follows must leave the + /// winner's record where it is. + #[tokio::test] + async fn abort_after_a_lost_write_race_leaves_the_other_writer_alone() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (vault, client, target) = scripted_client(vec![ + // Restore: trust root, empty target, marker published, then the + // record write is beaten to the path. + healthy_transit_key(), + not_found(), + not_found(), + kv_write_response(), + ScriptedResponse::error(400, "check-and-set parameter did not match the current version"), + // Abort: the marker is there and still carries no receipt. + kv_metadata_response(1), + kv_read_response(marker_value(&[KEY_ID], &[], &bundle_manifest(&bundle).await.manifest_digest), 1), + empty_ok(), // marker removed + ]) + .await; + + restore_vault_backup( + &test_kek(), + &client, + &request(bundle, target.clone(), RestoreConflictPolicy::RestoreIntoEmptyTarget), + ) + .await + .expect_err("a concurrent writer must stop the restore"); + + let report = abort_vault_restore(&client, &target).await.expect("abort should succeed"); + assert!(report.removed_key_ids.is_empty(), "{report:?}"); + assert_eq!( + report.skipped, + [VaultRestoreAbortSkip { + key_id: KEY_ID.to_string(), + reason: VaultRestoreAbortSkipReason::NeverWritten, + }] + ); + assert!( + !vault + .requests() + .contains(&format!("DELETE /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{KEY_ID}")), + "the abort deleted a record this restore never wrote: {:?}", + vault.requests() + ); + } + + #[tokio::test] + async fn abort_without_a_marker_is_refused() { + let (_vault, client, target) = scripted_client(vec![not_found()]).await; + abort_vault_restore(&client, &target) + .await + .expect_err("aborting without an in-flight restore must be refused"); + } + + #[tokio::test] + async fn a_bundle_referencing_a_record_it_does_not_carry_is_rejected() { + let temp = TempDir::new().expect("temp dir"); + let mut references = references(); + references.kv_records.push(VaultKvRecordReference { + key_id: "ghost-key".to_string(), + kv_version: 1, + }); + let bundle = write_bundle(temp.path(), references, &[(KEY_ID, record_content())]).await; + let (_vault, client, target) = scripted_client(Vec::new()).await; + + let report = dry_run_vault_restore(&test_kek(), &client, &request(bundle, target, RestoreConflictPolicy::Fail)) + .await + .expect("dry-run should report rather than fail"); + assert_eq!(report.blockers[0].code, RestoreBlockerCode::MissingArtifact); + } + + #[tokio::test] + async fn a_tampered_artifact_is_rejected_before_the_target_is_probed() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let artifact = bundle.join(format!("{VAULT_RECORD_ARTIFACT_DIR}/{KEY_ID}{VAULT_RECORD_ARTIFACT_SUFFIX}")); + let mut payload = tokio::fs::read(&artifact).await.expect("read artifact"); + let last = payload.len() - 1; + payload[last] ^= 0xff; + tokio::fs::write(&artifact, &payload).await.expect("tamper artifact"); + + let (vault, client, target) = scripted_client(Vec::new()).await; + let report = dry_run_vault_restore(&test_kek(), &client, &request(bundle, target, RestoreConflictPolicy::Fail)) + .await + .expect("dry-run should report rather than fail"); + assert_eq!(report.blockers[0].code, RestoreBlockerCode::BundleCorrupted); + assert!(vault.requests().is_empty(), "a broken bundle must not reach the target"); + } + + #[tokio::test] + async fn a_bundle_from_another_deployment_is_blocked() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (_vault, client, target) = scripted_client(vec![healthy_transit_key(), not_found(), not_found()]).await; + let mut request = request(bundle, target, RestoreConflictPolicy::Fail); + request.target_deployment_identity = "deployment-other".to_string(); + + let report = dry_run_vault_restore(&test_kek(), &client, &request) + .await + .expect("dry-run should succeed"); + assert!( + report + .blockers + .iter() + .any(|blocker| blocker.code == RestoreBlockerCode::DeploymentMismatch) + ); + } + + #[tokio::test] + async fn a_generation_rollback_is_reported_and_refused() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let (_vault, client, target) = scripted_client(vec![healthy_transit_key(), not_found(), not_found()]).await; + let mut request = request(bundle, target, RestoreConflictPolicy::Fail); + request.observed_generation = Some(SNAPSHOT_GENERATION + 1); + + let report = dry_run_vault_restore(&test_kek(), &client, &request) + .await + .expect("dry-run should succeed"); + assert!( + report + .conflicts + .iter() + .any(|conflict| conflict.kind == RestoreConflictKind::GenerationRegression) + ); + } + + /// End-to-end drill against a real Vault. Never run in CI (no server); + /// kept compiling so the wiring cannot rot. + #[tokio::test] + #[ignore = "requires a live Vault with a restored Transit trust root"] + async fn real_vault_restore_round_trip() { + let temp = TempDir::new().expect("temp dir"); + let bundle = standard_bundle(temp.path()).await; + let mut target = target("http://127.0.0.1:8200"); + target.auth_method = VaultAuthMethod::Token { + token: std::env::var("VAULT_TOKEN").expect("VAULT_TOKEN"), + }; + let client = VaultRestoreClient::connect(&target, &KmsConfig::default()) + .await + .expect("connect to Vault"); + + let request = request(bundle, target, RestoreConflictPolicy::RestoreIntoEmptyTarget); + let preflight = dry_run_vault_restore(&test_kek(), &client, &request) + .await + .expect("dry-run should succeed"); + assert!(preflight.restore_permitted(), "{preflight:?}"); + let report = restore_vault_backup(&test_kek(), &client, &request) + .await + .expect("restore should succeed"); + assert_eq!(report.restored_key_ids, [KEY_ID]); + } +}