diff --git a/backend/tests/ai_sessions.rs b/backend/tests/ai_sessions.rs index 6b33a4141e..b2b8406db3 100644 --- a/backend/tests/ai_sessions.rs +++ b/backend/tests/ai_sessions.rs @@ -1580,3 +1580,163 @@ async fn test_backups_fall_back_to_the_instance_storage(db: Pool) -> a Ok(()) } + +/// Runs its closure when dropped, so process-wide state is put back even when an assertion +/// fails. +struct OnDrop(F); +impl Drop for OnDrop { + fn drop(&mut self) { + (self.0)() + } +} + +/// Once every worker reads AES-256-GCM and this server has been up long enough for any server +/// of the previous release to be gone, objects are written in it: a session backed up as CBC +/// before the switch still restores whole with the GCM pieces added after it, a head at its +/// cap and a push token still read back under the larger ciphertext, and relocated GCM +/// ciphertext does not decrypt for another user. +#[sqlx::test(fixtures("base"))] +async fn test_backups_switch_to_gcm_once_every_worker_reads_it( + db: Pool, +) -> anyhow::Result<()> { + use windmill_common::min_version::{ + MIN_VERSION, MIN_VERSION_SUPPORTS_AEAD_SECRETS, MIN_VERSION_SUPPORTS_JOB_TOKEN_SCOPES, + }; + use windmill_common::workspace_crypt::V2_MAGIC; + + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let base = format!( + "http://localhost:{}/api/w/test-workspace", + server.addr.port() + ); + let storage_dir = tempfile::tempdir()?; + configure_primary_lfs(&db, &storage_dir.path().to_string_lossy()).await?; + let before = MIN_VERSION.load_full(); + let _restore = OnDrop(move || MIN_VERSION.store(before.clone())); + let _uptime = OnDrop(|| windmill_common::db::set_process_uptime_offset(Default::default())); + let is_gcm = |b: &[u8]| b.starts_with(&V2_MAGIC); + let object = |path: &str| -> Vec { + std::fs::read(user_root(storage_dir.path(), "test@windmill.dev").join(path)).unwrap() + }; + + MIN_VERSION.store(std::sync::Arc::new( + MIN_VERSION_SUPPORTS_JOB_TOKEN_SCOPES.version().clone(), + )); + let resp = push( + &base, + "SECRET_TOKEN", + json!({ + "owner": "test@windmill.dev", + "sessions": [{ + "id": "s1", + "whole": true, + "head": { "id": "s1", "workspace_id": "test-workspace", "createdAt": 1, "chatId": "c1" }, + "chats": [{ "id": "c1", "record": { "id": "c1", "title": "BEFORE" } }] + }] + }), + ) + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + let cbc_chat = object("sessions/s1/chats/c1.json"); + assert!(!cbc_chat.is_empty() && !is_gcm(&cbc_chat)); + + MIN_VERSION.store(std::sync::Arc::new( + MIN_VERSION_SUPPORTS_AEAD_SECRETS.version().clone(), + )); + // Every worker reads it, but this server was only just started: still CBC. + let resp = push( + &base, + "SECRET_TOKEN", + json!({ + "owner": "test@windmill.dev", + "sessions": [{ "id": "s1", "chats": [{ "id": "c0", "record": { "id": "c0", "title": "FRESH" } }] }] + }), + ) + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + let fresh_chat = object("sessions/s1/chats/c0.json"); + assert!(!fresh_chat.is_empty() && !is_gcm(&fresh_chat)); + + windmill_common::db::set_process_uptime_offset( + windmill_common::db::PREVIOUS_RELEASE_ROLLOUT_UPTIME, + ); + let resp = push( + &base, + "SECRET_TOKEN", + json!({ + "owner": "test@windmill.dev", + "sessions": [{ + "id": "s1", + "chats": [{ "id": "c2", "record": { "id": "c2", "title": "AFTER" } }], + "images": [{ "chat_id": "c2", "id": "i1", "data_url": "data:image/png;base64,AAAA" }] + }] + }), + ) + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + assert_eq!( + object("sessions/s1/chats/c1.json"), + cbc_chat, + "not rewritten" + ); + assert!(is_gcm(&object("sessions/s1/chats/c2.json"))); + assert!(is_gcm(&object("images/s1/c2/i1"))); + + let pulled = pull(&base, "SECRET_TOKEN", &["s1"]).await?; + let mut titles: Vec = pulled["sessions"][0]["chats"] + .as_array() + .unwrap() + .iter() + .map(|c| c["record"]["title"].as_str().unwrap().to_string()) + .collect(); + titles.sort(); + assert_eq!(titles, ["AFTER", "BEFORE", "FRESH"], "{pulled}"); + assert_eq!( + pulled["sessions"][0]["images"][0]["data_url"], + "data:image/png;base64,AAAA" + ); + + // A head at its cap, pushed in two parts: the second reads back the push token. + let mut head = json!({ "id": "s2", "workspace_id": "test-workspace", "createdAt": 2, "chatId": "c", "pad": "" }); + let pad = 1024 * 1024 - head.to_string().len(); + head["pad"] = json!("x".repeat(pad)); + assert_eq!(head.to_string().len(), 1024 * 1024); + for part in [ + json!({ "id": "s2", "whole": true, "push": "p2", "opens": true, "partial": true, "head": head }), + json!({ "id": "s2", "whole": true, "push": "p2", "chats": [{ "id": "c", "record": { "id": "c" } }] }), + ] { + let resp = push( + &base, + "SECRET_TOKEN", + json!({ "owner": "test@windmill.dev", "sessions": [part] }), + ) + .await?; + assert_eq!(resp.status(), 200, "{}", resp.text().await?); + let pushed: Value = resp.json().await?; + assert_eq!(pushed["results"], json!([{ "id": "s2" }]), "{pushed}"); + } + assert!(is_gcm(&object("sessions/s2/head.json"))); + let pulled = pull(&base, "SECRET_TOKEN", &["s2"]).await?; + assert_eq!( + pulled["sessions"][0]["head"]["pad"].as_str().map(str::len), + Some(pad) + ); + assert_eq!(pulled["sessions"][0]["chats"][0]["id"], "c", "{pulled}"); + + let first = user_root(storage_dir.path(), "test@windmill.dev"); + copy_dir( + &first, + &first + .parent() + .unwrap() + .join(calculate_hash("test2@windmill.dev")), + )?; + assert_eq!( + pull(&base, "SECRET_TOKEN_2", &["s1", "s2"]).await?["sessions"], + json!([]), + "relocated ciphertext must not decrypt for another user" + ); + + Ok(()) +} diff --git a/backend/windmill-api/src/ai_sessions.rs b/backend/windmill-api/src/ai_sessions.rs index 9e980a54f4..097e5c4862 100644 --- a/backend/windmill-api/src/ai_sessions.rs +++ b/backend/windmill-api/src/ai_sessions.rs @@ -24,7 +24,6 @@ use axum::{ Extension, Json, Router, }; use futures::{StreamExt, TryStreamExt}; -use magic_crypt::MagicCryptTrait; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use std::sync::Arc; @@ -33,9 +32,11 @@ use windmill_api_workspaces::ai_session_backups::{ fallback_store, generation_prefix, primary_store, storage_id, MAX_OBJECT_BYTES, }; use windmill_api_workspaces::workspaces::sessions_retention_days; +use windmill_common::db::{process_uptime, PREVIOUS_RELEASE_ROLLOUT_UPTIME}; use windmill_common::error::{Error, JsonResult, Result}; use windmill_common::utils::calculate_hash; use windmill_common::variables::{crypt_from_key_with_suffix, get_workspace_key, WorkspaceCrypt}; +use windmill_common::workspace_crypt::{CipherFormat, MAX_BYTES_OVERHEAD}; use windmill_object_store::object_store_reexports::{ ObjectStore, ObjectStoreError, Path as ObjectPath, PutPayload, }; @@ -47,9 +48,9 @@ const PULL_BODY_LIMIT: usize = 64 * 1024; /// A pull answer larger than this hands the remaining ids back as `deferred`. const PULL_RESPONSE_BUDGET: usize = 32 * 1024 * 1024; const MAX_HEAD_BYTES: usize = 1024 * 1024; -/// What the cipher adds to a plaintext at most (a block of padding): an object stored at a -/// cap is that much larger than the cap when read back. -const CIPHER_PADDING: usize = 16; +/// What the cipher adds to a plaintext at most: an object stored at a cap is that much larger +/// than the cap when read back. +const CIPHER_PADDING: usize = MAX_BYTES_OVERHEAD; /// The browser bounds an image to a 1568 px edge and re-encodes past 700 KB; this is /// well above what that produces. const MAX_IMAGE_BYTES: usize = 4 * 1024 * 1024; @@ -120,6 +121,15 @@ pub fn workspaced_service() -> Router { ) } +fn backup_format(current: CipherFormat, uptime: std::time::Duration) -> CipherFormat { + // A previous-release server reads only CBC here: GCM waits for it to be rolled out. + if current == CipherFormat::V2 && uptime >= PREVIOUS_RELEASE_ROLLOUT_UPTIME { + CipherFormat::V2 + } else { + CipherFormat::LegacyCbc + } +} + /// What reading an object yields. `Gone`: not there (deleted since the listing, or never /// pushed). `Grown`: larger than expected, so replaced since the listing (or planted), and /// left unread; a pull answers with a page ending before it rather than without it. @@ -214,7 +224,8 @@ impl Backend { } fn seal(&self, plaintext: &[u8]) -> Vec { - self.mc.encrypt_bytes_to_bytes(plaintext) + let format = backup_format(CipherFormat::current(), process_uptime()); + self.mc.encrypt_bytes_in(plaintext, format) } /// Bytes written. @@ -255,7 +266,7 @@ impl Backend { // apart beyond the cipher's padding check, which a wrong key passes now and then. match self .mc - .decrypt_bytes_to_bytes(&bytes) + .decrypt_bytes(&bytes) .ok() .and_then(|plaintext| String::from_utf8(plaintext).ok()) { @@ -1632,3 +1643,32 @@ async fn push( results, })) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn backups_are_written_in_gcm_only_once_the_previous_release_is_rolled_out() { + let just_short = PREVIOUS_RELEASE_ROLLOUT_UPTIME - std::time::Duration::from_secs(1); + for (current, uptime, expected) in [ + ( + CipherFormat::V2, + PREVIOUS_RELEASE_ROLLOUT_UPTIME, + CipherFormat::V2, + ), + (CipherFormat::V2, just_short, CipherFormat::LegacyCbc), + ( + CipherFormat::LegacyCbc, + PREVIOUS_RELEASE_ROLLOUT_UPTIME * 4, + CipherFormat::LegacyCbc, + ), + ] { + assert_eq!( + backup_format(current, uptime), + expected, + "{current:?} {uptime:?}" + ); + } + } +} diff --git a/backend/windmill-common/src/db.rs b/backend/windmill-common/src/db.rs index 38b89ad3b6..dc22b37157 100644 --- a/backend/windmill-common/src/db.rs +++ b/backend/windmill-common/src/db.rs @@ -328,7 +328,24 @@ pub fn connection_is_retired(age: std::time::Duration) -> bool { cutoff != 0 && (PROCESS_START.elapsed().saturating_sub(age).as_millis() as u64) < cutoff } +/// Servers report no version, so the worker version gate alone cannot see a server of the +/// previous release; its rollout is assumed done once this server has been up this long. +pub const PREVIOUS_RELEASE_ROLLOUT_UPTIME: std::time::Duration = + std::time::Duration::from_secs(30 * 60); + +static PROCESS_UPTIME_OFFSET_SECS: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(0); + /// Time since this process first used its database pool. pub fn process_uptime() -> std::time::Duration { PROCESS_START.elapsed() + + std::time::Duration::from_secs( + PROCESS_UPTIME_OFFSET_SECS.load(std::sync::atomic::Ordering::Relaxed), + ) +} + +/// For tests of what waits on [`process_uptime`]: adds `offset` to it. +#[doc(hidden)] +pub fn set_process_uptime_offset(offset: std::time::Duration) { + PROCESS_UPTIME_OFFSET_SECS.store(offset.as_secs(), std::sync::atomic::Ordering::Relaxed); } diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index f357298243..b7a2f19d86 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -416,11 +416,6 @@ pub fn variable_value_widen_pending() -> bool { CipherFormat::current() == CipherFormat::V2 && !VARIABLE_VALUE_WIDENED.load(Ordering::Relaxed) } -/// Servers report no version, so the worker version gate alone cannot see a server of the -/// previous release; its rollout is assumed done once this server has been up this long. -const VARIABLE_VALUE_WIDEN_MIN_UPTIME: std::time::Duration = - std::time::Duration::from_secs(30 * 60); - /// Widen `variable.value` to [`VARIABLE_VALUE_WIDTH`], or `false` when it is too early to. /// Not a migration: changing a column's type, even its length, fails every statement a live /// process prepared on it until its connection closes. It waits for v2 writes, gated on a @@ -437,7 +432,7 @@ async fn widen_variable_value(db: &DB) -> error::Result { .fetch_one(db) .await?; if typmod >= 0 && typmod - 4 < VARIABLE_VALUE_WIDTH { - if crate::db::process_uptime() < VARIABLE_VALUE_WIDEN_MIN_UPTIME { + if crate::db::process_uptime() < crate::db::PREVIOUS_RELEASE_ROLLOUT_UPTIME { return Ok(false); } tracing::info!("widening variable.value to VARCHAR({VARIABLE_VALUE_WIDTH})"); diff --git a/backend/windmill-common/src/workspace_crypt.rs b/backend/windmill-common/src/workspace_crypt.rs index 26f38e123b..6bdea28391 100644 --- a/backend/windmill-common/src/workspace_crypt.rs +++ b/backend/windmill-common/src/workspace_crypt.rs @@ -32,6 +32,9 @@ const TAG_LEN: usize = 16; const BLOCK: usize = 16; const NARROW_VARIABLE_VALUE_WIDTH: usize = 15000; +/// The most `encrypt_bytes_in` adds to a plaintext, in either format. +pub const MAX_BYTES_OVERHEAD: usize = V2_MAGIC.len() + NONCE_LEN + BLOCK + TAG_LEN; + const AEAD_KEY_INFO: &[u8] = b"windmill workspace secret aes-256-gcm v1"; const B64: base64::engine::GeneralPurpose = base64::engine::general_purpose::STANDARD; @@ -78,8 +81,8 @@ pub struct WorkspaceCrypt { aead: Aes256Gcm, } -/// Code that drives `magic_crypt` directly on this cipher (app link secrets, AI session -/// objects) keeps the legacy format it always had. +/// Code that drives `magic_crypt` directly on this cipher (app link secrets) keeps the legacy +/// format it always had. impl std::ops::Deref for WorkspaceCrypt { type Target = MagicCrypt256; fn deref(&self) -> &MagicCrypt256 { @@ -153,7 +156,14 @@ impl WorkspaceCrypt { (self.decrypt_str(&upgraded).ok()? == plaintext).then_some(upgraded) } - fn decrypt_bytes(&self, bytes: &[u8]) -> Result, DecryptFailure> { + pub fn encrypt_bytes_in(&self, plaintext: &[u8], format: CipherFormat) -> Vec { + match format { + CipherFormat::V2 => self.seal(plaintext), + CipherFormat::LegacyCbc => self.legacy.encrypt_bytes_to_bytes(plaintext), + } + } + + pub fn decrypt_bytes(&self, bytes: &[u8]) -> Result, DecryptFailure> { match self.open(bytes) { Some(plaintext) => Ok(plaintext), // Also for a value with the magic: a CBC ciphertext starts with it one time in 2^32. @@ -259,6 +269,20 @@ mod tests { ); } + #[test] + fn bytes_round_trip_within_the_overhead_in_both_formats() { + let wc = WorkspaceCrypt::new(KEY); + for len in 0..64 { + let plain = vec![b'x'; len]; + for format in [CipherFormat::V2, CipherFormat::LegacyCbc] { + let sealed = wc.encrypt_bytes_in(&plain, format); + assert_eq!(wc.open(&sealed).is_some(), format == CipherFormat::V2); + assert!(sealed.len() <= len + MAX_BYTES_OVERHEAD, "len {len}"); + assert_eq!(wc.decrypt_bytes(&sealed).unwrap(), plain); + } + } + } + #[test] fn v2_too_long_for_a_narrow_column_is_written_in_cbc() { let wc = WorkspaceCrypt::new(KEY);