mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-14 00:02:55 +00:00
fix: encrypt ai session backups with aes-256-gcm (#11734)
* fix: encrypt ai session backups with aes-256-gcm Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * test: cover ai session backups written with aes-256-gcm Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix: name the release ai session readers ship in Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix: hold ai session gcm writes until old servers are rolled out Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alexander Petric
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
2f9a111bbb
commit
c7ee201655
@@ -1580,3 +1580,163 @@ async fn test_backups_fall_back_to_the_instance_storage(db: Pool<Postgres>) -> a
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Runs its closure when dropped, so process-wide state is put back even when an assertion
|
||||
/// fails.
|
||||
struct OnDrop<F: FnMut()>(F);
|
||||
impl<F: FnMut()> Drop for OnDrop<F> {
|
||||
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<Postgres>,
|
||||
) -> 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<u8> {
|
||||
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<String> = 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(())
|
||||
}
|
||||
|
||||
@@ -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<u8> {
|
||||
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:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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<bool> {
|
||||
.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})");
|
||||
|
||||
@@ -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<Vec<u8>, DecryptFailure> {
|
||||
pub fn encrypt_bytes_in(&self, plaintext: &[u8], format: CipherFormat) -> Vec<u8> {
|
||||
match format {
|
||||
CipherFormat::V2 => self.seal(plaintext),
|
||||
CipherFormat::LegacyCbc => self.legacy.encrypt_bytes_to_bytes(plaintext),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn decrypt_bytes(&self, bytes: &[u8]) -> Result<Vec<u8>, 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);
|
||||
|
||||
Reference in New Issue
Block a user