From 0d30b623a709d57db5c99455d2d81440f76ca3ae Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 1 Aug 2026 12:03:11 +0800 Subject: [PATCH] feat(kms): orchestrate Vault restore ordering, verification, and mismatch detection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restoring a Vault-backed deployment is an operator action for the trust root and a RustFS action for everything else. This adds the orchestration around that split. Ordering is structural, not documented: VaultRestoreSequence refuses any stage that is not the next one, so material and metadata can never be published before the Transit/HSM trust root is verified back. Every way the recovered Vault can disagree with the bundle is a distinct typed mismatch: a missing Transit key, a min_decryption_version raised above what the bundle still needs, a Transit version history restored below the snapshot point, a rolled-back KV generation, and cluster, namespace, or mount drift. A dry-run reports them all and issues reads only; an actual restore refuses on the first one, before any KV path is probed. Writes are create-only check-and-set throughout, so version regression, generation rollback, and reviving deleted state are impossible rather than merely rejected, and a lost race stops the restore instead of overwriting. A reserved KV record is the single commit point: published before the first write and removed after the last, it names exactly the records this restore owns, so re-running the same bundle rolls forward idempotently and abort takes back precisely what was written — never a record that pre-existed. Every interruption converges to the complete old or the complete new state. Tests drive the whole surface offline through the scripted Vault responder, including the zero-write contract asserted at the wire; the live-Vault drill is ignored but kept compiling. Refs rustfs/backlog#1572 (part of rustfs/backlog#1562) --- crates/kms/src/backup/mod.rs | 18 +- crates/kms/src/backup/vault_restore.rs | 1908 ++++++++++++++++++++++++ 2 files changed, 1921 insertions(+), 5 deletions(-) create mode 100644 crates/kms/src/backup/vault_restore.rs diff --git a/crates/kms/src/backup/mod.rs b/crates/kms/src/backup/mod.rs index 53b0fa741..26b850441 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,10 @@ 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::{ + 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..3d33fcd96 --- /dev/null +++ b/crates/kms/src/backup/vault_restore.rs @@ -0,0 +1,1908 @@ +// 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. +//! +//! # Writes +//! +//! [`dry_run_vault_restore`] issues reads only. [`restore_vault_backup`] +//! writes exclusively 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. +//! +//! # 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 and [`abort_vault_restore`] rolls back exactly the records the +//! marker names. Every interruption converges to the complete old or the +//! complete new state. + +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}; +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 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 generation = 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?; + let Some(generation) = generation 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(false)` means a concurrent writer got there + /// first, which the caller surfaces instead of clobbering. + 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(_) => Ok(true), + Err(error) if is_cas_conflict(&error) => Ok(false), + Err(error) => Err(AttemptError::from_vaultrs(error, |error| { + KmsError::backend_error(format!("Failed to create KV record {path} during restore: {error}")) + })), + } + }) + .await + } + + /// Remove a KV record and all its versions. Only ever applied to records + /// this restore created (the commit marker names 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. It names the exact bundle and the exact key ids this +/// restore publishes, so re-entry completes the same work and an abort takes +/// back precisely what was written — never a record that pre-existed. +#[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, +} + +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, + } + } + + /// 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)?; + } + Ok(marker) + } + + fn matches_bundle(&self, manifest: &BackupManifest) -> bool { + self.backup_id == manifest.backup_id && self.manifest_digest == manifest.manifest_digest + } +} + +/// 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}") +} + +/// Verify the recovered external trust root against the bundle's references. +/// +/// Static coordinates (cluster, namespace, mounts) are compared first because +/// they need no Vault call and because a mount mismatch would make every +/// subsequent probe read the wrong thing. Reads only. +async fn verify_trust_root( + client: &VaultRestoreClient, + references: &VaultExternalReferences, + target: &VaultRestoreTarget, +) -> Result> { + 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 Some(transit) = references.transit.as_ref() else { + return Ok(mismatches); + }; + let Some(target_mount) = target.transit_mount.as_ref() else { + mismatches.push(VaultRestoreMismatch::Mount { + dependency: "vault transit mount".to_string(), + expected: transit.mount_path.clone(), + observed: "".to_string(), + }); + return Ok(mismatches); + }; + if &transit.mount_path != target_mount { + mismatches.push(VaultRestoreMismatch::Mount { + dependency: "vault transit mount".to_string(), + expected: transit.mount_path.clone(), + observed: target_mount.clone(), + }); + return Ok(mismatches); + } + + match client.read_transit_key(target_mount, &transit.key_name).await? { + None => mismatches.push(VaultRestoreMismatch::TransitKeyMissing { + mount_path: target_mount.clone(), + 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(VaultRestoreCommitMarker::from_value(record.content)?)), + 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(marker) = &marker + && !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 ('{}')", marker.backup_id), + }); + return Ok(report); + } + + for record in selected_records(&decoded, marker.as_ref()) { + 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(marker) = &existing_marker + && !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", + marker.backup_id + ))); + } + + let mut plans = Vec::new(); + for record in selected_records(&decoded, existing_marker.as_ref()) { + 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 (marker, resumed) = match existing_marker { + Some(marker) => (marker, 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 content = serde_json::to_value(&marker) + .map_err(|error| KmsError::internal_error(format!("restore commit marker serialization failed: {error}")))?; + if !client.create_kv_record(&marker_kv_path(prefix), &content).await? { + return Err(KmsError::invalid_operation( + "another restore published a commit marker for this target concurrently; \ + resolve it before retrying", + )); + } + (marker, 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; + } + if !client + .create_kv_record(&record_kv_path(prefix, &record.key_id), &record.content) + .await? + { + 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 + ))); + } + } + 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 exactly the records the published marker names — never one the +/// restore found already present — then the marker itself. 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(marker) = read_marker(client, prefix).await? else { + return Err(KmsError::invalid_operation(format!( + "no vault restore is in progress under {}/{prefix}", + target.kv_mount + ))); + }; + for key_id in &marker.key_ids { + client.delete_kv_record(&record_kv_path(prefix, key_id)).await?; + } + client.delete_kv_record(&marker_kv_path(prefix)).await +} + +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; + + 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() -> ScriptedResponse { + ScriptedResponse::ok(serde_json::json!({ + "created_time": "2026-08-01T00:00:00Z", + "deletion_time": "", + "custom_metadata": null, + "destroyed": false, + "version": 1, + })) + } + + fn not_found() -> ScriptedResponse { + ScriptedResponse::error(404, "not found") + } + + fn empty_ok() -> ScriptedResponse { + ScriptedResponse::ok(serde_json::json!({})) + } + + fn marker_value(key_ids: &[&str], 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, + }) + } + + 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 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 + 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 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}"), + format!("POST /v1/{KV_MOUNT}/data/{KV_PREFIX}/{VAULT_RESTORE_MARKER_KEY}"), + format!("POST /v1/{KV_MOUNT}/data/{KV_PREFIX}/{KEY_ID}"), + format!("DELETE /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{VAULT_RESTORE_MARKER_KEY}"), + ], + "the trust root is verified first and the marker brackets every write" + ); + + // Every write is create-only: restore can structurally not overwrite. + for body in vault.request_bodies().iter().filter(|body| !body.is_empty()) { + let parsed: serde_json::Value = serde_json::from_str(body).expect("write body is JSON"); + assert_eq!(parsed["options"]["cas"], 0, "write was not create-only: {body}"); + } + } + + #[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; + let marker = marker_value(&[KEY_ID], &manifest.manifest_digest); + let (vault, client, target) = scripted_client(vec![ + healthy_transit_key(), + kv_metadata_response(1), // marker present + kv_read_response(marker, 1), + 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_marked_records() { + let (vault, client, target) = scripted_client(vec![ + kv_metadata_response(1), + kv_read_response( + marker_value( + &[KEY_ID], + &ContentDigest { + algorithm: DigestAlgorithm::Sha256, + hex: "a".repeat(64), + }, + ), + 1, + ), + empty_ok(), // record removed + empty_ok(), // marker removed + ]) + .await; + + abort_vault_restore(&client, &target).await.expect("abort should succeed"); + 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!("DELETE /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{KEY_ID}"), + format!("DELETE /v1/{KV_MOUNT}/metadata/{KV_PREFIX}/{VAULT_RESTORE_MARKER_KEY}"), + ] + ); + } + + #[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]); + } +}