feat(kms): orchestrate Vault restore ordering, verification, and mismatch detection

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/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::<Aes256Gcm>::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<BackupManifest> {
+/// 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<BackupManifest> {
     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<BackupManif
         }
         Err(error) => 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<BackupManifest> {
+    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<u8> {
+pub(crate) fn artifact_aad(backup_id: &str, snapshot_generation: u64, artifact_path: &str) -> Vec<u8> {
     serde_json::to_vec(&(BUNDLE_AAD_CONTEXT, backup_id, snapshot_generation, artifact_path))
         .expect("AAD tuple of strings and integers always serializes")
 }

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<String>,
+}
+
+/// 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<String>,
+    /// 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<VaultTransitReference>,
+    /// Per-key KV generation observed at snapshot time, one entry per key the
+    /// bundle carries a record for.
+    pub kv_records: Vec<VaultKvRecordReference>,
+}
+
+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<LocalKdfDescriptor>,
+    /// 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<VaultExternalReferences>,
     /// 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,161 @@ 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",
+        );
+    }
+
+    #[test]
+    fn external_reference_contradictions_fail_closed() {
+        let cases: [(&str, Box<dyn Fn(&mut VaultExternalReferences)>); 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"]);
+    }
 }

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