From 7ff268799cb4a59dcb2bd06eda3f6561185a8d4b Mon Sep 17 00:00:00 2001 From: zhom <2717306+zhom@users.noreply.github.com> Date: Sun, 1 Nov 2026 14:14:25 +0400 Subject: [PATCH] feat: verify-at-rest command --- donut-sync/README.md | 3 +- donut-sync/src/at_rest.rs | 4 + donut-sync/src/lib.rs | 7 +- donut-sync/src/verify.rs | 496 ++++++++++++++++++++++++++++++++++++ donut-sync/tests/at_rest.rs | 19 ++ 5 files changed, 526 insertions(+), 3 deletions(-) create mode 100644 donut-sync/src/verify.rs diff --git a/donut-sync/README.md b/donut-sync/README.md index b2de22f7..51ab1fec 100644 --- a/donut-sync/README.md +++ b/donut-sync/README.md @@ -22,7 +22,8 @@ also read. With `SYNC_AT_REST_KEY` set, the server encrypts every object it stores, and transfers go through it instead of straight to storage. Objects stored before -stay readable; `donut-sync migrate-at-rest --help` encrypts them in place. +stay readable; `donut-sync migrate-at-rest --help` encrypts them in place, and +`donut-sync verify-at-rest` checks that none is left in plaintext. ## Test diff --git a/donut-sync/src/at_rest.rs b/donut-sync/src/at_rest.rs index 0de2ab06..c3b9b4e3 100644 --- a/donut-sync/src/at_rest.rs +++ b/donut-sync/src/at_rest.rs @@ -105,6 +105,10 @@ impl Keyring { &self.keys[0] } + pub fn holds(&self, id: &str) -> bool { + self.find(id).is_some() + } + fn find(&self, id: &str) -> Option<&Key> { self.keys.iter().find(|k| k.id == id) } diff --git a/donut-sync/src/lib.rs b/donut-sync/src/lib.rs index c1777d03..f956914e 100644 --- a/donut-sync/src/lib.rs +++ b/donut-sync/src/lib.rs @@ -24,6 +24,7 @@ pub mod subscribe; pub mod sync; pub mod time; pub mod transfer; +pub mod verify; use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::Arc; @@ -300,8 +301,10 @@ pub async fn run() -> i32 { let _ = dotenvy::dotenv(); let config = Config::from_env(); let args: Vec = std::env::args().skip(1).collect(); - if args.first().map(String::as_str) == Some(migrate::COMMAND) { - return migrate::run(&config, &args[1..]).await; + match args.first().map(String::as_str) { + Some(migrate::COMMAND) => return migrate::run(&config, &args[1..]).await, + Some(verify::COMMAND) => return verify::run(&config, &args[1..]).await, + _ => {} } if let Err(message) = validate_env(&config) { eprintln!("{message}"); diff --git a/donut-sync/src/verify.rs b/donut-sync/src/verify.rs new file mode 100644 index 00000000..721b5eba --- /dev/null +++ b/donut-sync/src/verify.rs @@ -0,0 +1,496 @@ +//! `donut-sync verify-at-rest`: reads the bucket and reports what is sealed, +//! under which key, and what is still plaintext, then opens a random sample of +//! sealed objects end to end. It writes nothing. Exit code 0 means every +//! object is sealed under a key the server holds and every sample opened. + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use futures_util::StreamExt; + +use crate::at_rest::{self, Keyring, MARKER}; +use crate::config::Config; +use crate::log; +use crate::migrate::TMP_PREFIX; +use crate::sealed_store::sealed_by; +use crate::store::{ListInput, Store, StoreError}; + +pub const COMMAND: &str = "verify-at-rest"; +const CONTEXT: &str = "Verify"; +const PROGRESS_EVERY: Duration = Duration::from_secs(10); +/// Plaintext keys named in the log, so a few can be looked at. +const NAMED_PLAINTEXT: u64 = 20; + +const USAGE: &str = "Usage: donut-sync verify-at-rest [--prefix P] [--sample N] [--concurrency N] + +Reads S3_BUCKET and counts plaintext objects and objects sealed under each key +of SYNC_AT_REST_KEY, then decrypts N random sealed objects in full. Writes +nothing. Exits 0 only when nothing is plaintext and every sample decrypts. + + --prefix P Only keys under P. + --sample N Sealed objects to decrypt (default 100). + --concurrency N HEAD requests at a time (default 8)."; + +#[derive(Debug, Clone)] +pub struct Options { + pub prefix: Option, + pub sample: usize, + pub concurrency: usize, +} + +impl Default for Options { + fn default() -> Self { + Self { + prefix: None, + sample: 100, + concurrency: 8, + } + } +} + +fn parse(args: &[String]) -> Result { + let mut options = Options::default(); + let mut args = args.iter(); + while let Some(arg) = args.next() { + let value = args + .next() + .ok_or_else(|| format!("Unknown argument {arg}"))?; + match arg.as_str() { + "--prefix" => options.prefix = Some(value.clone()), + "--sample" => { + options.sample = value + .parse() + .map_err(|_| "--sample must be a whole number")?; + } + "--concurrency" => { + options.concurrency = value + .parse() + .ok() + .filter(|n| (1..=64).contains(n)) + .ok_or("--concurrency must be 1-64")?; + } + _ => return Err(format!("Unknown argument {arg}")), + } + } + Ok(options) +} + +pub async fn run(config: &Config, args: &[String]) -> i32 { + if args.iter().any(|a| a == "--help" || a == "-h") { + println!("{USAGE}"); + return 0; + } + let options = match parse(args) { + Ok(options) => options, + Err(message) => { + eprintln!("{message}\n\n{USAGE}"); + return 2; + } + }; + let Some(keys) = config.at_rest_key.as_deref() else { + eprintln!("SYNC_AT_REST_KEY is not set"); + return 1; + }; + let keyring = match Keyring::parse(keys) { + Ok(keyring) => Arc::new(keyring), + Err(message) => { + eprintln!("{message}"); + return 1; + } + }; + let store: Arc = match crate::raw_store(config) { + Ok(store) => Arc::new(store), + Err(message) => { + eprintln!("{message}"); + return 1; + } + }; + match verify(store, keyring, &options).await { + Ok(report) if report.complete() => 0, + Ok(_) => 1, + Err(message) => { + log::error(CONTEXT, &format!("Verify stopped err=\"{message}\"")); + 1 + } + } +} + +#[derive(Debug, Default, Clone, PartialEq)] +pub struct Sealed { + pub objects: u64, + pub stored_bytes: u64, + pub plain_bytes: u64, +} + +#[derive(Debug, Default, Clone, PartialEq)] +pub struct Report { + pub objects: u64, + pub stored_bytes: u64, + pub plaintext: u64, + pub plaintext_bytes: u64, + /// By key id, including ids the keyring does not hold. + pub sealed: BTreeMap, + /// Sealed under a key id the keyring does not hold. + pub unknown_key: u64, + /// A marker of another format version, or a size no sealed object has. + pub invalid: u64, + /// Leftover temporary copies of an interrupted migration. + pub temporary: u64, + /// Gone between the listing and the HEAD. + pub vanished: u64, + pub sampled: u64, + pub sample_failed: u64, +} + +impl Report { + pub fn complete(&self) -> bool { + self.plaintext == 0 && self.unknown_key == 0 && self.invalid == 0 && self.sample_failed == 0 + } + + fn line(&self) -> String { + let mut line = format!( + "objects={} stored_bytes={} plaintext={} plaintext_bytes={}", + self.objects, self.stored_bytes, self.plaintext, self.plaintext_bytes + ); + for (id, sealed) in &self.sealed { + line.push_str(&format!( + " sealed[{id}]={} sealed[{id}]_bytes={} sealed[{id}]_plain_bytes={}", + sealed.objects, sealed.stored_bytes, sealed.plain_bytes + )); + } + line.push_str(&format!( + " unknown_key={} invalid={} temporary={} vanished={} sampled={} sample_failed={}", + self.unknown_key, + self.invalid, + self.temporary, + self.vanished, + self.sampled, + self.sample_failed + )); + line + } +} + +type Classified = Result<(Kind, u64), StoreError>; + +enum Kind { + Plaintext, + Sealed { id: String, plain: u64 }, + Invalid(String), + Vanished, +} + +async fn classify(store: &dyn Store, key: &str) -> Classified { + let head = match store.head_object(key).await { + Ok(head) => head, + Err(StoreError::NotFound) => return Ok((Kind::Vanished, 0)), + Err(e) => return Err(e), + }; + let stored = head.content_length.unwrap_or(0); + let kind = match (head.metadata_value(MARKER), sealed_by(&head)) { + (None, _) => Kind::Plaintext, + (Some(marker), None) => Kind::Invalid(format!("unknown marker \"{marker}\"")), + (Some(_), Some(id)) => match at_rest::plain_len(id, stored) { + Some(plain) => Kind::Sealed { + id: id.to_string(), + plain, + }, + None => Kind::Invalid(format!("size {stored} fits no sealed object")), + }, + }; + Ok((kind, stored)) +} + +/// Downloads and opens the whole object; `Err` says why it does not open. +async fn open_fully( + store: &dyn Store, + keyring: &Arc, + key: &str, + plain: u64, +) -> Result<(), String> { + let download = store.get_object(key).await.map_err(|e| e.to_string())?; + let mut body = at_rest::open_stream(keyring.clone(), download.body); + let mut read = 0u64; + while let Some(chunk) = body.next().await { + read += chunk.map_err(|e| e.to_string())?.len() as u64; + } + if read != plain { + return Err(format!("opened {read} bytes, expected {plain}")); + } + Ok(()) +} + +pub async fn verify( + store: Arc, + keyring: Arc, + options: &Options, +) -> Result { + let started = Instant::now(); + log::log( + CONTEXT, + &format!( + "Verify started prefix={} sample={} concurrency={}", + options.prefix.as_deref().unwrap_or("-"), + options.sample, + options.concurrency + ), + ); + let mut report = Report::default(); + // A uniform sample of sealed objects under keys the keyring holds. + let mut sample: Vec<(String, u64)> = Vec::new(); + let mut candidates = 0u64; + let mut token: Option = None; + let mut last_progress = Instant::now(); + loop { + let page = store + .list_objects_v2(&ListInput { + prefix: options.prefix.clone(), + max_keys: Some("1000".to_string()), + continuation_token: token.clone(), + ..ListInput::default() + }) + .await + .map_err(|e| format!("Listing failed: {e}"))?; + let (temporary, keys): (Vec, Vec) = page + .contents + .into_iter() + .map(|object| object.key) + .partition(|key| key.starts_with(TMP_PREFIX)); + report.temporary += temporary.len() as u64; + let classified: Vec<(String, Classified)> = futures_util::stream::iter(keys) + .map(|key| { + let store = store.clone(); + async move { + let kind = classify(store.as_ref(), &key).await; + (key, kind) + } + }) + .buffer_unordered(options.concurrency) + .collect() + .await; + for (key, result) in classified { + let (kind, stored) = result.map_err(|e| format!("HEAD failed key={key} err=\"{e}\""))?; + if !matches!(kind, Kind::Vanished) { + report.objects += 1; + report.stored_bytes += stored; + } + match kind { + Kind::Plaintext => { + report.plaintext += 1; + report.plaintext_bytes += stored; + if report.plaintext <= NAMED_PLAINTEXT { + log::warn( + CONTEXT, + &format!("Plaintext object key={key} bytes={stored}"), + ); + } + } + Kind::Sealed { id, plain } => { + let held = keyring.holds(&id); + let entry = report.sealed.entry(id.clone()).or_default(); + entry.objects += 1; + entry.stored_bytes += stored; + entry.plain_bytes += plain; + if !held { + report.unknown_key += 1; + log::warn( + CONTEXT, + &format!("Object under a key not held key={key} kid={id}"), + ); + continue; + } + candidates += 1; + if sample.len() < options.sample { + sample.push((key, plain)); + } else { + let slot = rand::random_range(0..candidates) as usize; + if slot < options.sample { + sample[slot] = (key, plain); + } + } + } + Kind::Invalid(why) => { + report.invalid += 1; + log::warn( + CONTEXT, + &format!("Invalid sealed object key={key} err=\"{why}\""), + ); + } + Kind::Vanished => report.vanished += 1, + } + } + token = page.next_continuation_token.filter(|_| page.is_truncated); + if token.is_none() { + break; + } + if last_progress.elapsed() >= PROGRESS_EVERY { + last_progress = Instant::now(); + log::log( + CONTEXT, + &format!( + "Verify progress {} elapsed_ms={}", + report.line(), + started.elapsed().as_millis() + ), + ); + } + } + let opened: Vec<(String, Result<(), String>)> = futures_util::stream::iter(sample) + .map(|(key, plain)| { + let store = store.clone(); + let keyring = keyring.clone(); + async move { + let result = open_fully(store.as_ref(), &keyring, &key, plain).await; + (key, result) + } + }) + .buffer_unordered(options.concurrency) + .collect() + .await; + for (key, result) in opened { + report.sampled += 1; + if let Err(e) = result { + report.sample_failed += 1; + log::warn( + CONTEXT, + &format!("Sample did not open key={key} err=\"{e}\""), + ); + } + } + let line = format!( + "Verify done complete={} {} elapsed_ms={}", + report.complete(), + report.line(), + started.elapsed().as_millis() + ); + if report.complete() { + log::log(CONTEXT, &line); + } else { + log::warn(CONTEXT, &line); + } + Ok(report) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::memory_store::MemoryStore; + use crate::sealed_store::SealedStore; + + const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; + const K2: &str = "k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE="; + + fn keyring(value: &str) -> Arc { + Arc::new(Keyring::parse(value).unwrap()) + } + + #[tokio::test] + async fn counts_by_key_and_proves_completion_after_migration() { + let raw = Arc::new(MemoryStore::default()); + raw.set("a", "plain-a"); + raw.set("b", "plain-bb"); + SealedStore::new(raw.clone(), keyring(K2)) + .put_object("c", b"ccc".to_vec(), "text/plain", &[]) + .await + .unwrap(); + raw.set(&format!("{TMP_PREFIX}left"), "x"); + let keys = keyring(&format!("{K1},{K2}")); + + let before = verify(raw.clone(), keys.clone(), &Options::default()) + .await + .unwrap(); + assert!(!before.complete()); + assert_eq!( + (before.objects, before.plaintext, before.plaintext_bytes), + (3, 2, 15) + ); + assert_eq!(before.sealed["k2"].objects, 1); + assert_eq!(before.sealed["k2"].plain_bytes, 3); + assert_eq!( + (before.temporary, before.sampled, before.sample_failed), + (1, 1, 0) + ); + + crate::migrate::migrate( + raw.clone(), + keys.clone(), + &crate::migrate::Options { + concurrency: 2, + ..Default::default() + }, + ) + .await + .unwrap(); + let after = verify(raw.clone(), keys, &Options::default()) + .await + .unwrap(); + assert!(after.complete(), "{after:?}"); + assert_eq!(after.sealed["k1"].objects, 3); + assert_eq!(after.sealed["k1"].plain_bytes, 18); + assert_eq!(after.sampled, 3); + } + + #[tokio::test] + async fn a_damaged_sample_unknown_keys_and_bad_markers_fail_it() { + let raw = Arc::new(MemoryStore::default()); + let keys = keyring(K1); + SealedStore::new(raw.clone(), keys.clone()) + .put_object("good", b"hello".to_vec(), "text/plain", &[]) + .await + .unwrap(); + let mut damaged = raw.raw("good").unwrap(); + let last = damaged.body.len() - 1; + damaged.body[last] ^= 1; + raw.insert("bad", damaged.body, damaged.metadata, None); + SealedStore::new(raw.clone(), keyring(K2)) + .put_object("other", b"x".to_vec(), "text/plain", &[]) + .await + .unwrap(); + raw.insert( + "odd", + b"short".to_vec(), + vec![(MARKER.to_string(), "9:k1".to_string())], + None, + ); + let report = verify(raw, keys, &Options::default()).await.unwrap(); + assert!(!report.complete()); + assert_eq!( + (report.plaintext, report.unknown_key, report.invalid), + (0, 1, 1) + ); + assert_eq!((report.sampled, report.sample_failed), (2, 1)); + } + + #[tokio::test] + async fn the_sample_is_bounded() { + let raw = Arc::new(MemoryStore::default()); + let keys = keyring(K1); + let sealed = SealedStore::new(raw.clone(), keys.clone()); + for i in 0..50 { + sealed + .put_object(&format!("k{i}"), vec![1; i], "text/plain", &[]) + .await + .unwrap(); + } + let options = Options { + sample: 5, + ..Options::default() + }; + let report = verify(raw, keys, &options).await.unwrap(); + assert!(report.complete()); + assert_eq!((report.objects, report.sampled), (50, 5)); + } + + #[test] + fn arguments_parse() { + let args: Vec = ["--prefix", "users/", "--sample", "0"] + .iter() + .map(|s| s.to_string()) + .collect(); + let options = parse(&args).unwrap(); + assert_eq!(options.prefix.as_deref(), Some("users/")); + assert_eq!((options.sample, options.concurrency), (0, 8)); + assert!(parse(&["--dry-run".to_string()]).is_err()); + } +} diff --git a/donut-sync/tests/at_rest.rs b/donut-sync/tests/at_rest.rs index 9d031a75..16512943 100644 --- a/donut-sync/tests/at_rest.rs +++ b/donut-sync/tests/at_rest.rs @@ -372,6 +372,17 @@ async fn migration_seals_everything_in_place_and_a_rerun_does_nothing() { env.push(("SYNC_AT_REST_KEY", K1)); let total: usize = objects.iter().map(|(_, v)| v.len()).sum(); + let (code, out) = run_command(&["verify-at-rest"], &env); + assert_eq!(code, 1, "{out}"); + assert!(out.contains("complete=false"), "{out}"); + assert!( + out.contains(&format!( + "plaintext=25 plaintext_bytes={}", + objects.iter().map(|(_, v)| v.len()).sum::() + )), + "{out}" + ); + let (code, out) = run_command(&["migrate-at-rest", "--dry-run"], &env); assert_eq!(code, 0, "{out}"); assert!( @@ -414,6 +425,14 @@ async fn migration_seals_everything_in_place_and_a_rerun_does_nothing() { let (code, out) = run_command(&["migrate-at-rest"], &env); assert_eq!(code, 0, "{out}"); assert!(out.contains("current=25 sealed=0"), "{out}"); + let (code, out) = run_command(&["verify-at-rest", "--sample", "10"], &env); + assert_eq!(code, 0, "{out}"); + assert!(out.contains("complete=true objects=25"), "{out}"); + assert!( + out.contains("plaintext=0 plaintext_bytes=0 sealed[k1]=25"), + "{out}" + ); + assert!(out.contains("sampled=10 sample_failed=0"), "{out}"); // The server reads what the migration wrote. Self-hosted keys are bare. let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(K1))]).await;