diff --git a/crates/kms/src/backup/drill.rs b/crates/kms/src/backup/drill.rs new file mode 100644 index 000000000..cfb11c23f --- /dev/null +++ b/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, +} + +/// 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, + /// 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, + /// 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, + /// 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, + /// 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, + /// 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, + /// Every check that did not hold, in the order they were observed. + pub findings: Vec, + /// Whether the drill passed. + pub verdict: DrillVerdict, +} + +impl DrillEvidence { + /// Serialize the evidence for archival. + pub fn encode(&self) -> Result> { + 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, +} + +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", &"") + .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, + metadata: EncryptionMetadata, +} + +/// Collects per-phase wall-clock cost. +struct PhaseClock { + timings: Vec, + 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 { + 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 { + 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 { + 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> { + 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 = 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> { + 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> { + 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 { + 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 { + 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> { + 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) +} diff --git a/crates/kms/src/backup/mod.rs b/crates/kms/src/backup/mod.rs index 36ace38d7..d8ffd1712 100644 --- a/crates/kms/src/backup/mod.rs +++ b/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, };