feat(kms): add a disaster-recovery drill harness for the Local backend

Rehearse the full backup/restore loop offline and return machine-readable
evidence: seed a sandbox deployment, seal sample objects through the
production encryption path, export a bundle, destroy the persistence layer,
preflight, restore, and decrypt every pre-disaster object again.

The evidence records the measured recovery point (one key is written past the
snapshot fence and must stay unrecoverable), the recovery time by phase, the
manifest digest before and after, and whether the restore treated its bundle
as read-only.

crates/kms/src/backup/drill.rs

@@ -0,0 +1,976 @@
+// 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.
+
+//! Disaster-recovery drill harness for the Local backend.
+//!
+//! A backup that has never been restored is a hypothesis. This module turns
+//! it into evidence: [`run_local_drill`] rehearses the complete loop —
+//! seed a deployment, seal real objects through the production encryption
+//! path, export a bundle, destroy the persistence layer, preflight, restore,
+//! and then *decrypt the pre-disaster ciphertext again* — and returns a
+//! machine-readable [`DrillEvidence`] record of what happened and how long it
+//! took.
+//!
+//! # Why the object probe is the acceptance criterion
+//!
+//! Manifest digests and artifact counts only prove the bundle is intact. The
+//! question a disaster actually asks is whether historical objects are still
+//! readable, so the drill keeps the ciphertext and the encryption metadata of
+//! every object it sealed before the disaster and, after the restore, decrypts
+//! each one through a freshly opened backend. A drill whose probes do not
+//! reproduce the original plaintext byte for byte is a failed drill regardless
+//! of how clean the bundle looked.
+//!
+//! # Sandbox only
+//!
+//! The drill owns everything under its workspace: it creates the deployment it
+//! later destroys. It never reads, writes, or points a restore at a running
+//! deployment's key directory, and the destructive step is applied to the
+//! workspace copy only. Sizing the dataset to production is what makes the
+//! recovery-time number transferable — the code path is identical either way.
+//!
+//! # RPO and RTO
+//!
+//! The drill writes one key *after* the export fence closed and proves it is
+//! absent after the restore: the recovery point is the snapshot generation,
+//! not the moment of the disaster, and the drill measures that window instead
+//! of asserting it. The recovery time is the sum of the phases an operator
+//! actually waits on (quarantine, preflight, restore, verification); human
+//! decision time is deliberately excluded and belongs to the runbook.
+//!
+//! # What this module does not re-test
+//!
+//! Bundle-side defects (tampered, truncated, wrong-KEK, incomplete,
+//! unknown-version) and every commit-point crash are covered exhaustively by
+//! the unit tests of [`crate::backup::local_export`] and
+//! [`crate::backup::local_restore`]. The drill exercises the operator-facing
+//! loop on top of them rather than restating them.
+
+use crate::backends::local::{LOCAL_KMS_MASTER_KEY_SALT_FILE, LOCAL_RESTORE_COMMIT_MARKER_FILE, LocalKmsBackend};
+use crate::backends::{KmsBackend as KmsBackendTrait, local::validate_key_id};
+use crate::backup::capability::BackupBackendKind;
+use crate::backup::dry_run::RestoreDryRunReport;
+use crate::backup::local_export::{BackupKek, LocalBackupExportRequest, export_local_backup, read_local_bundle_manifest};
+use crate::backup::local_restore::{
+    LocalRestoreReport, LocalRestoreRequest, RestoreConflictPolicy, dry_run_local_restore, restore_local_backup,
+};
+use crate::backup::manifest::ContentDigest;
+use crate::config::{BackendConfig, KmsConfig, LocalConfig};
+use crate::error::{KmsError, Result};
+use crate::manager::KmsManager;
+use crate::service::ObjectEncryptionService;
+use crate::types::{CreateKeyRequest, EncryptionAlgorithm, EncryptionMetadata, KeyUsage};
+use jiff::Zoned;
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use std::collections::HashMap;
+use std::io::Cursor;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::{Instant, UNIX_EPOCH};
+use tokio::fs;
+use tokio::io::AsyncReadExt;
+
+/// Format version of [`DrillEvidence`], so an evidence consumer can reject a
+/// shape it does not know instead of guessing.
+pub const DRILL_EVIDENCE_FORMAT_VERSION: u32 = 1;
+
+/// Bucket name the drill seals its sample objects under.
+const DRILL_BUCKET: &str = "rustfs-kms-drill";
+/// Workspace subdirectory holding the deployment the drill destroys.
+const LIVE_DIR: &str = "live";
+/// Workspace subdirectory holding the exported bundle.
+const BUNDLE_DIR: &str = "bundle";
+/// Workspace subdirectory the damaged key directory is moved aside into.
+const QUARANTINE_DIR: &str = "quarantine";
+/// Upper bound on a single drill run, guarding against a dataset that would
+/// take an operator by surprise mid-incident-rehearsal.
+const MAX_DRILL_OBJECTS: usize = 10_000;
+
+/// Size of the deployment a drill rehearses against.
+///
+/// The numbers are evidence: a recovery time only transfers to production if
+/// the drill restored a comparable number of keys.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct DrillDataset {
+    /// Master keys created before the backup.
+    pub keys: usize,
+    /// Objects sealed under each key and re-read after the restore.
+    pub objects_per_key: usize,
+    /// Plaintext size of each sample object.
+    pub object_bytes: usize,
+}
+
+impl Default for DrillDataset {
+    fn default() -> Self {
+        Self {
+            keys: 4,
+            objects_per_key: 2,
+            object_bytes: 4096,
+        }
+    }
+}
+
+/// How the drill destroys the KMS persistence layer.
+///
+/// Each variant is one row of the Local disaster matrix. All of them converge
+/// on the same recovery procedure — quarantine what survived, restore into a
+/// clean directory — which is exactly the point: the operator does not have to
+/// diagnose the failure mode before acting.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum DrillDisaster {
+    /// The whole key directory is gone (lost volume, wiped host).
+    KeyDirectoryLost,
+    /// The key records survive but the KDF salt is gone, so none of their
+    /// material can be unwrapped.
+    MasterKeySaltLost,
+    /// One key record is truncated in place (torn write, partial media
+    /// failure).
+    KeyRecordCorrupted,
+}
+
+/// One drill phase, in execution order.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum DrillPhase {
+    /// Create the deployment and its master keys.
+    Seed,
+    /// Seal the sample objects that the restore will later have to make
+    /// readable again.
+    SealObjects,
+    /// Export the backup bundle.
+    Export,
+    /// Write past the snapshot fence, establishing the recovery point.
+    PostSnapshotWrite,
+    /// Destroy the persistence layer.
+    Disaster,
+    /// Move what survived aside so the restore target is clean.
+    Quarantine,
+    /// Zero-write restore preflight.
+    DryRun,
+    /// Execute the restore.
+    Restore,
+    /// Decrypt every pre-disaster sample object again.
+    Verify,
+    /// Re-run the restore against the recovered target.
+    RepeatRestore,
+}
+
+impl DrillPhase {
+    /// Whether the phase is part of the recovery an operator waits on.
+    ///
+    /// Seeding, sealing, exporting and the disaster itself are drill
+    /// scaffolding, not recovery work, so they stay out of the recovery-time
+    /// total.
+    fn counts_towards_recovery_time(self) -> bool {
+        matches!(
+            self,
+            DrillPhase::Quarantine | DrillPhase::DryRun | DrillPhase::Restore | DrillPhase::Verify
+        )
+    }
+}
+
+/// Wall-clock cost of one phase.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct DrillPhaseTiming {
+    /// Which phase.
+    pub phase: DrillPhase,
+    /// Duration in milliseconds.
+    pub millis: u64,
+}
+
+/// Outcome of decrypting one pre-disaster sample object after the restore.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct EnvelopeProbe {
+    /// Master key the object's data key was wrapped under.
+    pub key_id: String,
+    /// Bucket the object was sealed under.
+    pub bucket: String,
+    /// Object key.
+    pub object_key: String,
+    /// Digest of the plaintext as it was before the disaster.
+    pub plaintext_digest: ContentDigest,
+    /// Whether the restored deployment reproduced that plaintext exactly.
+    pub verified: bool,
+    /// Failure detail when it did not. Identifiers and error text only.
+    pub detail: Option<String>,
+}
+
+/// What the bundle was, and what it still was after the recovery.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct DrillBundleEvidence {
+    /// Bundle identifier.
+    pub backup_id: String,
+    /// Snapshot generation the bundle belongs to.
+    pub snapshot_generation: u64,
+    /// Sealed manifest digest recorded at export time.
+    pub manifest_digest: ContentDigest,
+    /// Number of artifacts in the bundle.
+    pub artifact_count: usize,
+    /// Total bundle size on disk.
+    pub bundle_bytes: u64,
+    /// Manifest digest re-read after the whole recovery.
+    pub manifest_digest_after_recovery: ContentDigest,
+    /// Whether every byte of the bundle directory survived the recovery
+    /// unchanged. Restore must treat its source as strictly read-only.
+    pub source_unchanged: bool,
+}
+
+/// What the disaster destroyed and how the recovery proceeded.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct DrillRecoveryEvidence {
+    /// Top-level names the disaster removed or damaged.
+    pub destroyed_files: Vec<String>,
+    /// Where the damaged directory was preserved. Nothing is ever destroyed
+    /// by the recovery itself: the pre-restore state stays available for
+    /// forensics and rollback.
+    pub quarantine_path: PathBuf,
+    /// Top-level names that were moved into quarantine.
+    pub quarantined_files: Vec<String>,
+    /// The preflight report, verbatim.
+    pub dry_run: RestoreDryRunReport,
+    /// Whether the target was byte-for-byte identical before and after the
+    /// preflight.
+    pub dry_run_zero_write: bool,
+    /// The restore report, verbatim. Absent when the preflight refused the
+    /// recovery and the drill therefore never restored.
+    pub restore: Option<LocalRestoreReport>,
+    /// Whether the cutover commit marker was gone once the restore returned.
+    /// A surviving marker means the target is still mid-cutover.
+    pub commit_marker_cleared: bool,
+    /// Whether re-running the same restore against the recovered target was
+    /// refused.
+    pub repeat_restore_refused: bool,
+    /// Whether that refused re-run left the recovered target untouched.
+    pub repeat_restore_left_target_unchanged: bool,
+}
+
+/// The measured recovery point.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct DrillRpoEvidence {
+    /// When the export sealed the bundle.
+    #[serde(with = "crate::time_serde::zoned")]
+    pub snapshot_completed_at: Zoned,
+    /// When the disaster struck.
+    #[serde(with = "crate::time_serde::zoned")]
+    pub disaster_at: Zoned,
+    /// The data-loss window: work committed inside it is not in the bundle.
+    pub rpo_window_millis: u64,
+    /// Keys created after the fence closed, which the bundle cannot contain.
+    pub post_snapshot_key_ids: Vec<String>,
+    /// Objects sealed under those keys.
+    pub post_snapshot_objects: usize,
+    /// How many of them were readable again after the restore. Anything above
+    /// zero means the bundle was not a point-in-time snapshot.
+    pub post_snapshot_objects_recovered: usize,
+}
+
+/// Overall drill outcome.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum DrillVerdict {
+    /// Every check held.
+    Passed,
+    /// At least one check did not; see the findings.
+    Failed,
+}
+
+/// Machine-readable record of one drill run.
+///
+/// Identifiers, digests, and durations only: no key material, no master key,
+/// no plaintext, no ciphertext. It is meant to be archived next to the
+/// incident record as the immutable evidence a drill actually happened and
+/// what it proved.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct DrillEvidence {
+    /// Evidence schema version.
+    pub format_version: u32,
+    /// Caller-chosen identifier of this run.
+    pub drill_id: String,
+    /// Backend the drill exercised.
+    pub backend: BackupBackendKind,
+    /// Deployment identity the rehearsal deployment carried.
+    pub deployment_identity: String,
+    /// RustFS version recorded in the bundle manifest.
+    pub rustfs_version: String,
+    /// Disaster that was simulated.
+    pub disaster: DrillDisaster,
+    /// Size of the rehearsed deployment.
+    pub dataset: DrillDataset,
+    /// When the run started.
+    #[serde(with = "crate::time_serde::zoned")]
+    pub started_at: Zoned,
+    /// Bundle evidence.
+    pub bundle: DrillBundleEvidence,
+    /// Recovery evidence.
+    pub recovery: DrillRecoveryEvidence,
+    /// Recovery-point evidence.
+    pub rpo: DrillRpoEvidence,
+    /// Per-phase durations.
+    pub timings: Vec<DrillPhaseTiming>,
+    /// Recovery time: the phases an operator waits on.
+    pub rto_millis: u64,
+    /// One entry per sample object re-read after the restore.
+    pub envelope_probes: Vec<EnvelopeProbe>,
+    /// Every check that did not hold, in the order they were observed.
+    pub findings: Vec<String>,
+    /// Whether the drill passed.
+    pub verdict: DrillVerdict,
+}
+
+impl DrillEvidence {
+    /// Serialize the evidence for archival.
+    pub fn encode(&self) -> Result<Vec<u8>> {
+        serde_json::to_vec_pretty(self)
+            .map_err(|error| KmsError::internal_error(format!("drill evidence does not serialize: {error}")))
+    }
+}
+
+/// Parameters of one drill run.
+pub struct DrillRequest {
+    /// Identifier of this run; also the prefix of every key the drill creates.
+    pub drill_id: String,
+    /// Scratch directory the drill owns entirely. Created if absent; must not
+    /// already contain a drill workspace.
+    pub workspace: PathBuf,
+    /// Deployment identity stamped into the bundle and required to match on
+    /// restore.
+    pub deployment_identity: String,
+    /// RustFS version string recorded in the manifest.
+    pub rustfs_version: String,
+    /// Snapshot generation for the exported bundle.
+    pub snapshot_generation: u64,
+    /// Size of the rehearsed deployment.
+    pub dataset: DrillDataset,
+    /// Disaster to simulate.
+    pub disaster: DrillDisaster,
+    /// At-rest master key of the rehearsal deployment. Supply a throwaway
+    /// value: the drill's own deployment is the only thing it protects.
+    pub master_key: String,
+    /// Unix permission bits for key files.
+    pub file_permissions: Option<u32>,
+}
+
+impl std::fmt::Debug for DrillRequest {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DrillRequest")
+            .field("drill_id", &self.drill_id)
+            .field("workspace", &self.workspace)
+            .field("deployment_identity", &self.deployment_identity)
+            .field("rustfs_version", &self.rustfs_version)
+            .field("snapshot_generation", &self.snapshot_generation)
+            .field("dataset", &self.dataset)
+            .field("disaster", &self.disaster)
+            .field("master_key", &"<redacted>")
+            .field("file_permissions", &self.file_permissions)
+            .finish()
+    }
+}
+
+impl DrillRequest {
+    fn validate(&self) -> Result<()> {
+        for (field, value) in [
+            ("drill_id", &self.drill_id),
+            ("deployment_identity", &self.deployment_identity),
+            ("rustfs_version", &self.rustfs_version),
+            ("master_key", &self.master_key),
+        ] {
+            if value.is_empty() {
+                return Err(KmsError::validation_error(format!("drill {field} must not be empty")));
+            }
+        }
+        // Key ids are derived from the drill id, so an id that cannot name a
+        // key file has to fail here rather than half way through seeding.
+        validate_key_id(&self.drill_id)?;
+        if !self.workspace.is_absolute() {
+            return Err(KmsError::validation_error("drill workspace must be an absolute path"));
+        }
+        if self.dataset.keys == 0 || self.dataset.objects_per_key == 0 || self.dataset.object_bytes == 0 {
+            return Err(KmsError::validation_error(
+                "drill dataset must contain at least one key, one object per key, and one byte per object",
+            ));
+        }
+        let objects = self
+            .dataset
+            .keys
+            .checked_mul(self.dataset.objects_per_key)
+            .ok_or_else(|| KmsError::validation_error("drill dataset size overflows"))?;
+        if objects > MAX_DRILL_OBJECTS {
+            return Err(KmsError::validation_error(format!(
+                "drill dataset of {objects} objects exceeds the {MAX_DRILL_OBJECTS} object limit"
+            )));
+        }
+        Ok(())
+    }
+
+    fn key_id(&self, index: usize) -> String {
+        format!("{}-key-{index:04}", self.drill_id)
+    }
+
+    fn post_snapshot_key_id(&self) -> String {
+        format!("{}-post-snapshot", self.drill_id)
+    }
+
+    fn local_config(&self, key_dir: PathBuf) -> KmsConfig {
+        let mut config = KmsConfig::local(key_dir.clone());
+        config.backend_config = BackendConfig::Local(LocalConfig {
+            key_dir,
+            master_key: Some(self.master_key.clone()),
+            file_permissions: self.file_permissions,
+        });
+        // The drill re-opens the backend after the restore to prove the
+        // material on disk decrypts; a warm cache would prove nothing.
+        config.enable_cache = false;
+        config
+    }
+}
+
+/// One object sealed before the disaster, kept so the restore can be asked to
+/// make it readable again.
+struct SealedObject {
+    key_id: String,
+    object_key: String,
+    plaintext_digest: ContentDigest,
+    ciphertext: Vec<u8>,
+    metadata: EncryptionMetadata,
+}
+
+/// Collects per-phase wall-clock cost.
+struct PhaseClock {
+    timings: Vec<DrillPhaseTiming>,
+    started: Instant,
+}
+
+impl PhaseClock {
+    fn new() -> Self {
+        Self {
+            timings: Vec::new(),
+            started: Instant::now(),
+        }
+    }
+
+    /// Close the current phase and open the next one.
+    fn finish(&mut self, phase: DrillPhase) {
+        let now = Instant::now();
+        self.timings.push(DrillPhaseTiming {
+            phase,
+            millis: now.duration_since(self.started).as_millis().min(u128::from(u64::MAX)) as u64,
+        });
+        self.started = now;
+    }
+
+    fn recovery_time_millis(&self) -> u64 {
+        self.timings
+            .iter()
+            .filter(|timing| timing.phase.counts_towards_recovery_time())
+            .map(|timing| timing.millis)
+            .sum()
+    }
+}
+
+/// Run one disaster-recovery drill and return its evidence.
+///
+/// Environment failures (an unusable workspace, a backend that will not start)
+/// are errors. Verdicts about the recovery itself — a probe that did not
+/// reproduce its plaintext, a preflight that wrote, a bundle the restore
+/// modified — are findings inside the returned evidence, so a failed drill
+/// still produces the record of *how* it failed.
+pub async fn run_local_drill(kek: &BackupKek, request: &DrillRequest) -> Result<DrillEvidence> {
+    request.validate()?;
+    let started_at = Zoned::now();
+    let mut clock = PhaseClock::new();
+    let mut findings = Vec::new();
+
+    let live_dir = request.workspace.join(LIVE_DIR);
+    let bundle_dir = request.workspace.join(BUNDLE_DIR);
+    let quarantine_dir = request.workspace.join(QUARANTINE_DIR);
+    prepare_workspace(&request.workspace, &live_dir).await?;
+
+    let config = request.local_config(live_dir.clone());
+
+    // The seeded deployment and its bundle. The backend is dropped before the
+    // disaster so no live handle outlives the state it points at.
+    let (sealed, rpo_sealed, manifest, snapshot_completed_at) = {
+        let backend = Arc::new(LocalKmsBackend::new(config.clone()).await?);
+        let service = ObjectEncryptionService::new(KmsManager::new(backend.clone(), config.clone()));
+
+        let mut key_ids = Vec::with_capacity(request.dataset.keys);
+        for index in 0..request.dataset.keys {
+            let key_id = request.key_id(index);
+            create_drill_key(&service, &key_id).await?;
+            key_ids.push(key_id);
+        }
+        clock.finish(DrillPhase::Seed);
+
+        let mut sealed = Vec::with_capacity(request.dataset.keys * request.dataset.objects_per_key);
+        for key_id in &key_ids {
+            for index in 0..request.dataset.objects_per_key {
+                sealed.push(seal_object(&service, key_id, index, request.dataset.object_bytes).await?);
+            }
+        }
+        clock.finish(DrillPhase::SealObjects);
+
+        let client = backend
+            .local_backup_client()
+            .ok_or_else(|| KmsError::internal_error("local backend did not expose its backup client"))?;
+        let manifest = export_local_backup(
+            client,
+            kek,
+            &LocalBackupExportRequest {
+                backup_id: request.drill_id.clone(),
+                deployment_identity: request.deployment_identity.clone(),
+                rustfs_version: request.rustfs_version.clone(),
+                snapshot_generation: request.snapshot_generation,
+                destination: bundle_dir.clone(),
+                sanitized_config: None,
+            },
+        )
+        .await?;
+        let snapshot_completed_at = Zoned::now();
+        clock.finish(DrillPhase::Export);
+
+        // Past the fence: this key belongs to the window the bundle cannot
+        // cover, and the restore must not resurrect it.
+        let post_snapshot_key = request.post_snapshot_key_id();
+        create_drill_key(&service, &post_snapshot_key).await?;
+        let mut rpo_sealed = Vec::with_capacity(request.dataset.objects_per_key);
+        for index in 0..request.dataset.objects_per_key {
+            rpo_sealed.push(seal_object(&service, &post_snapshot_key, index, request.dataset.object_bytes).await?);
+        }
+        clock.finish(DrillPhase::PostSnapshotWrite);
+
+        (sealed, rpo_sealed, manifest, snapshot_completed_at)
+    };
+
+    let bundle_digest_before = tree_digest(&bundle_dir).await?;
+
+    let destroyed_files = apply_disaster(&live_dir, request.disaster).await?;
+    let disaster_at = Zoned::now();
+    clock.finish(DrillPhase::Disaster);
+
+    // Recovery step one is always the same: preserve what survived, then
+    // restore into a clean directory. Nothing damaged is deleted.
+    let quarantined_files = quarantine(&live_dir, &quarantine_dir).await?;
+    clock.finish(DrillPhase::Quarantine);
+
+    let restore_request = LocalRestoreRequest {
+        bundle_dir: bundle_dir.clone(),
+        target_key_dir: live_dir.clone(),
+        target_deployment_identity: request.deployment_identity.clone(),
+        observed_generation: None,
+        master_key: Some(request.master_key.clone()),
+        conflict_policy: RestoreConflictPolicy::RestoreIntoEmptyTarget,
+        file_permissions: request.file_permissions,
+    };
+
+    let target_before_dry_run = tree_digest(&live_dir).await?;
+    let dry_run = dry_run_local_restore(kek, &restore_request).await?;
+    let dry_run_zero_write = tree_digest(&live_dir).await? == target_before_dry_run;
+    clock.finish(DrillPhase::DryRun);
+    if !dry_run_zero_write {
+        findings.push("restore preflight modified the target; a dry-run must be zero-write".to_string());
+    }
+    // A refused preflight ends the drill: proceeding anyway is exactly the
+    // behavior the restore contract forbids. The evidence is still produced,
+    // because *why* a rehearsed recovery was refused is the finding.
+    let permitted = dry_run.restore_permitted();
+    if !permitted {
+        findings.push(format!(
+            "restore preflight refused the recovery: {} blocker(s), {} conflict(s)",
+            dry_run.blockers.len(),
+            dry_run.conflicts.len()
+        ));
+    }
+
+    let mut restore_report = None;
+    let mut commit_marker_cleared = false;
+    let mut envelope_probes = Vec::new();
+    let mut post_snapshot_objects_recovered = 0usize;
+    let mut repeat_restore_refused = false;
+    let mut repeat_restore_left_target_unchanged = false;
+
+    if permitted {
+        let restore = restore_local_backup(kek, &restore_request).await?;
+        commit_marker_cleared = !fs::try_exists(live_dir.join(LOCAL_RESTORE_COMMIT_MARKER_FILE)).await?;
+        clock.finish(DrillPhase::Restore);
+        if !commit_marker_cleared {
+            findings.push("the cutover commit marker survived a completed restore".to_string());
+        }
+        if restore.backup_id != manifest.backup_id || restore.snapshot_generation != manifest.snapshot_generation {
+            findings.push(format!(
+                "restore reported bundle '{}' at generation {}, the exported bundle was '{}' at generation {}",
+                restore.backup_id, restore.snapshot_generation, manifest.backup_id, manifest.snapshot_generation
+            ));
+        }
+        restore_report = Some(restore);
+
+        // The acceptance criterion: a freshly opened backend over the restored
+        // directory must reproduce every pre-disaster plaintext.
+        {
+            let backend = Arc::new(LocalKmsBackend::new(config.clone()).await?);
+            let service = ObjectEncryptionService::new(KmsManager::new(backend, config.clone()));
+            for object in &sealed {
+                envelope_probes.push(probe_object(&service, object).await);
+            }
+            for object in &rpo_sealed {
+                if probe_object(&service, object).await.verified {
+                    post_snapshot_objects_recovered += 1;
+                }
+            }
+        }
+        clock.finish(DrillPhase::Verify);
+
+        for probe in &envelope_probes {
+            if !probe.verified {
+                findings.push(format!(
+                    "object '{}' under key '{}' did not decrypt to its pre-disaster plaintext",
+                    probe.object_key, probe.key_id
+                ));
+            }
+        }
+        if post_snapshot_objects_recovered != 0 {
+            findings.push(format!(
+                "{post_snapshot_objects_recovered} object(s) written after the snapshot fence were readable again; \
+                 the bundle is not a point-in-time snapshot"
+            ));
+        }
+
+        // A second restore into the recovered target must be refused without
+        // touching it: repeating a drill, or an operator running the runbook
+        // twice, must never be able to damage a healthy deployment.
+        let target_before_repeat = tree_digest(&live_dir).await?;
+        repeat_restore_refused = restore_local_backup(kek, &restore_request).await.is_err();
+        repeat_restore_left_target_unchanged = tree_digest(&live_dir).await? == target_before_repeat;
+        clock.finish(DrillPhase::RepeatRestore);
+        if !repeat_restore_refused {
+            findings.push("re-running the restore against the recovered target was accepted".to_string());
+        }
+        if !repeat_restore_left_target_unchanged {
+            findings.push("the refused re-run modified the recovered target".to_string());
+        }
+    }
+
+    let manifest_after = read_local_bundle_manifest(&bundle_dir).await?;
+    let source_unchanged = tree_digest(&bundle_dir).await? == bundle_digest_before;
+    if !source_unchanged || manifest_after.manifest_digest != manifest.manifest_digest {
+        findings.push("the bundle changed during the recovery; a restore source must be read-only".to_string());
+    }
+
+    let verdict = if findings.is_empty() {
+        DrillVerdict::Passed
+    } else {
+        DrillVerdict::Failed
+    };
+
+    Ok(DrillEvidence {
+        format_version: DRILL_EVIDENCE_FORMAT_VERSION,
+        drill_id: request.drill_id.clone(),
+        backend: BackupBackendKind::Local,
+        deployment_identity: request.deployment_identity.clone(),
+        rustfs_version: request.rustfs_version.clone(),
+        disaster: request.disaster,
+        dataset: request.dataset,
+        started_at,
+        bundle: DrillBundleEvidence {
+            backup_id: manifest.backup_id.clone(),
+            snapshot_generation: manifest.snapshot_generation,
+            manifest_digest: manifest.manifest_digest.clone(),
+            artifact_count: manifest.artifacts.len(),
+            bundle_bytes: directory_bytes(&bundle_dir).await?,
+            manifest_digest_after_recovery: manifest_after.manifest_digest,
+            source_unchanged,
+        },
+        recovery: DrillRecoveryEvidence {
+            destroyed_files,
+            quarantine_path: quarantine_dir,
+            quarantined_files,
+            dry_run,
+            dry_run_zero_write,
+            restore: restore_report,
+            commit_marker_cleared,
+            repeat_restore_refused,
+            repeat_restore_left_target_unchanged,
+        },
+        rpo: DrillRpoEvidence {
+            rpo_window_millis: elapsed_millis(&snapshot_completed_at, &disaster_at),
+            snapshot_completed_at,
+            disaster_at,
+            post_snapshot_key_ids: vec![request.post_snapshot_key_id()],
+            post_snapshot_objects: rpo_sealed.len(),
+            post_snapshot_objects_recovered,
+        },
+        rto_millis: clock.recovery_time_millis(),
+        timings: clock.timings,
+        envelope_probes,
+        findings,
+        verdict,
+    })
+}
+
+async fn prepare_workspace(workspace: &Path, live_dir: &Path) -> Result<()> {
+    fs::create_dir_all(workspace).await?;
+    if fs::try_exists(live_dir).await? {
+        return Err(KmsError::invalid_operation(format!(
+            "drill workspace {} already holds a rehearsal deployment; point the drill at a fresh directory",
+            workspace.display()
+        )));
+    }
+    fs::create_dir_all(live_dir).await?;
+    Ok(())
+}
+
+async fn create_drill_key(service: &ObjectEncryptionService, key_id: &str) -> Result<()> {
+    service
+        .create_key(CreateKeyRequest {
+            key_name: Some(key_id.to_string()),
+            key_usage: KeyUsage::EncryptDecrypt,
+            description: Some("rustfs kms disaster-recovery drill".to_string()),
+            policy: None,
+            tags: HashMap::new(),
+            origin: Some("kms-dr-drill".to_string()),
+        })
+        .await?;
+    Ok(())
+}
+
+/// Seal one sample object through the production encryption path and keep
+/// exactly what a real deployment would keep: the ciphertext and its
+/// encryption metadata.
+async fn seal_object(service: &ObjectEncryptionService, key_id: &str, index: usize, bytes: usize) -> Result<SealedObject> {
+    let object_key = format!("{key_id}/object-{index:04}");
+    let plaintext = drill_payload(&object_key, bytes);
+    let plaintext_digest = ContentDigest::sha256_of(&plaintext);
+    let result = service
+        .encrypt_object(
+            DRILL_BUCKET,
+            &object_key,
+            Cursor::new(plaintext),
+            &EncryptionAlgorithm::Aes256,
+            Some(key_id),
+            None,
+        )
+        .await?;
+    Ok(SealedObject {
+        key_id: key_id.to_string(),
+        object_key,
+        plaintext_digest,
+        ciphertext: result.ciphertext,
+        metadata: result.metadata,
+    })
+}
+
+/// Ask the restored deployment to unwrap one object's data key and reproduce
+/// its plaintext.
+async fn probe_object(service: &ObjectEncryptionService, object: &SealedObject) -> EnvelopeProbe {
+    let mut probe = EnvelopeProbe {
+        key_id: object.key_id.clone(),
+        bucket: DRILL_BUCKET.to_string(),
+        object_key: object.object_key.clone(),
+        plaintext_digest: object.plaintext_digest.clone(),
+        verified: false,
+        detail: None,
+    };
+
+    let decrypted = service
+        .decrypt_object(
+            DRILL_BUCKET,
+            &object.object_key,
+            object.ciphertext.clone(),
+            &object.metadata,
+            None,
+        )
+        .await;
+    let mut reader = match decrypted {
+        Ok(reader) => reader,
+        Err(error) => {
+            probe.detail = Some(error.to_string());
+            return probe;
+        }
+    };
+
+    let mut plaintext = Vec::new();
+    if let Err(error) = reader.read_to_end(&mut plaintext).await {
+        probe.detail = Some(error.to_string());
+        return probe;
+    }
+    if ContentDigest::sha256_of(&plaintext) == object.plaintext_digest {
+        probe.verified = true;
+    } else {
+        probe.detail = Some("decrypted plaintext does not match the pre-disaster digest".to_string());
+    }
+    probe
+}
+
+/// Deterministic, incompressible-enough sample payload derived from the object
+/// key, so a drill never needs a fixture file and two runs of the same drill
+/// seal the same bytes.
+fn drill_payload(object_key: &str, bytes: usize) -> Vec<u8> {
+    let mut payload = Vec::with_capacity(bytes);
+    let mut block = Sha256::digest(object_key.as_bytes());
+    while payload.len() < bytes {
+        let take = block.len().min(bytes - payload.len());
+        payload.extend_from_slice(&block[..take]);
+        block = Sha256::digest(block);
+    }
+    payload
+}
+
+/// Destroy the persistence layer and report what was destroyed.
+async fn apply_disaster(live_dir: &Path, disaster: DrillDisaster) -> Result<Vec<String>> {
+    match disaster {
+        DrillDisaster::KeyDirectoryLost => {
+            let names = top_level_names(live_dir).await?;
+            fs::remove_dir_all(live_dir).await?;
+            Ok(names)
+        }
+        DrillDisaster::MasterKeySaltLost => {
+            let salt = live_dir.join(LOCAL_KMS_MASTER_KEY_SALT_FILE);
+            if !fs::try_exists(&salt).await? {
+                return Err(KmsError::internal_error(
+                    "drill deployment has no master-key salt to destroy; the seed phase did not produce one",
+                ));
+            }
+            fs::remove_file(&salt).await?;
+            Ok(vec![LOCAL_KMS_MASTER_KEY_SALT_FILE.to_string()])
+        }
+        DrillDisaster::KeyRecordCorrupted => {
+            let mut records: Vec<String> = top_level_names(live_dir)
+                .await?
+                .into_iter()
+                .filter(|name| name.ends_with(".key"))
+                .collect();
+            records.sort();
+            let victim = records
+                .first()
+                .ok_or_else(|| KmsError::internal_error("drill deployment has no key record to corrupt"))?;
+            let path = live_dir.join(victim);
+            let mut content = fs::read(&path).await?;
+            content.truncate(content.len() / 2);
+            fs::write(&path, &content).await?;
+            Ok(vec![victim.clone()])
+        }
+    }
+}
+
+/// Move whatever survived the disaster aside, leaving a clean restore target.
+///
+/// The damaged state is preserved, never deleted: a restore that turns out to
+/// be the wrong decision must remain reversible, and forensics need the
+/// original bytes.
+async fn quarantine(live_dir: &Path, quarantine_dir: &Path) -> Result<Vec<String>> {
+    if !fs::try_exists(live_dir).await? {
+        fs::create_dir_all(live_dir).await?;
+        return Ok(Vec::new());
+    }
+    let names = top_level_names(live_dir).await?;
+    if names.is_empty() {
+        return Ok(Vec::new());
+    }
+    fs::rename(live_dir, quarantine_dir).await?;
+    fs::create_dir_all(live_dir).await?;
+    Ok(names)
+}
+
+async fn top_level_names(dir: &Path) -> Result<Vec<String>> {
+    let mut names = Vec::new();
+    let mut entries = match fs::read_dir(dir).await {
+        Ok(entries) => entries,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(names),
+        Err(error) => return Err(error.into()),
+    };
+    while let Some(entry) = entries.next_entry().await? {
+        names.push(entry.file_name().to_string_lossy().into_owned());
+    }
+    names.sort();
+    Ok(names)
+}
+
+async fn directory_bytes(dir: &Path) -> Result<u64> {
+    let dir = dir.to_path_buf();
+    tokio::task::spawn_blocking(move || {
+        let mut total = 0u64;
+        for path in walk_files(&dir)? {
+            total = total.saturating_add(std::fs::metadata(&path)?.len());
+        }
+        Ok(total)
+    })
+    .await
+    .map_err(|error| KmsError::internal_error(format!("drill directory sizing failed: {error}")))?
+}
+
+/// Canonical digest over a directory tree: relative path, length, content
+/// digest, and modification time of every file. Equality across an operation
+/// is proof that the operation performed no persistent write.
+async fn tree_digest(root: &Path) -> Result<ContentDigest> {
+    let root = root.to_path_buf();
+    tokio::task::spawn_blocking(move || {
+        let mut lines = Vec::new();
+        for path in walk_files(&root)? {
+            let metadata = std::fs::metadata(&path)?;
+            let modified = metadata
+                .modified()?
+                .duration_since(UNIX_EPOCH)
+                .map_or(0, |since| since.as_nanos());
+            let relative = path.strip_prefix(&root).unwrap_or(path.as_path()).to_string_lossy().into_owned();
+            let content = std::fs::read(&path)?;
+            lines.push(format!(
+                "{relative}\u{1f}{}\u{1f}{modified}\u{1f}{}",
+                metadata.len(),
+                hex::encode(Sha256::digest(&content))
+            ));
+        }
+        lines.sort();
+        Ok(ContentDigest::sha256_of(lines.join("\n").as_bytes()))
+    })
+    .await
+    .map_err(|error| KmsError::internal_error(format!("drill tree digest failed: {error}")))?
+}
+
+/// Every regular file under `root`, recursively. A missing root is an empty
+/// tree, not an error: the drill digests directories the disaster may have
+/// removed.
+fn walk_files(root: &Path) -> std::io::Result<Vec<PathBuf>> {
+    let mut files = Vec::new();
+    let mut stack = vec![root.to_path_buf()];
+    while let Some(dir) = stack.pop() {
+        let entries = match std::fs::read_dir(&dir) {
+            Ok(entries) => entries,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
+            Err(error) => return Err(error),
+        };
+        for entry in entries {
+            let entry = entry?;
+            let path = entry.path();
+            if entry.file_type()?.is_dir() {
+                stack.push(path);
+            } else {
+                files.push(path);
+            }
+        }
+    }
+    Ok(files)
+}
+
+fn elapsed_millis(from: &Zoned, to: &Zoned) -> u64 {
+    let nanos = to.timestamp().as_nanosecond() - from.timestamp().as_nanosecond();
+    u64::try_from(nanos / 1_000_000).unwrap_or(0)
+}

crates/kms/src/backup/mod.rs

@@ -21,7 +21,8 @@
 //! [`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.
+//! follow-up changes. [`drill`] rehearses the whole loop end to end and turns
+//! it into machine-readable evidence.
 //!
 //! # Bundle model
 //!
@@ -51,6 +52,7 @@
 //! format version.
 
 mod capability;
+pub mod drill;
 mod dry_run;
 mod error;
 pub mod local_export;
@@ -59,6 +61,10 @@ mod manifest;
 pub mod vault_restore;
 
 pub use capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
+pub use drill::{
+    DRILL_EVIDENCE_FORMAT_VERSION, DrillBundleEvidence, DrillDataset, DrillDisaster, DrillEvidence, DrillPhase,
+    DrillPhaseTiming, DrillRecoveryEvidence, DrillRequest, DrillRpoEvidence, DrillVerdict, EnvelopeProbe, run_local_drill,
+};
 pub use dry_run::{
     ExternalDependencyMismatch, RestoreBlocker, RestoreBlockerCode, RestoreConflict, RestoreConflictKind, RestoreDryRunReport,
 };