diff --git a/crates/context_aware_config/src/api/secrets/handlers.rs b/crates/context_aware_config/src/api/secrets/handlers.rs index 539bdade5..c9a2f5250 100644 --- a/crates/context_aware_config/src/api/secrets/handlers.rs +++ b/crates/context_aware_config/src/api/secrets/handlers.rs @@ -8,9 +8,9 @@ use service_utils::{ encryption::{ decrypt_workspace_key, encrypt_secret, rotate_workspace_encryption_key_helper, }, + redis::put_workspace_in_redis, service::types::{ - AppState, DbConnection, EncryptionKey, OrganisationId, SchemaName, - WorkspaceContext, WorkspaceId, + AppState, DbConnection, EncryptionKey, SchemaName, WorkspaceContext, }, }; use superposition_derives::{authorized, declare_resource}; @@ -48,11 +48,9 @@ pub fn master_key_endpoints() -> Scope { } fn get_workspace_encryption_key( - workspace_context: &WorkspaceContext, + workspace: &Workspace, master_encryption_key: &Option, ) -> superposition::Result { - let workspace: &Workspace = &workspace_context.settings; - let Some(master_encryption_key) = master_encryption_key else { log::error!("Master encryption key not configured"); return Err(bad_argument!( @@ -157,31 +155,45 @@ async fn create_handler( let req = req.into_inner(); let DbConnection(mut conn) = db_conn; + let user_email = user.get_email(); - let encryption_key = - get_workspace_encryption_key(&workspace_context, &state.master_encryption_key)?; - - let encrypted_secret_value = encrypt_secret(&req.value, &encryption_key) - .map_err(|e| bad_argument!("Encryption failed: {}", e))?; - - let now = chrono::Utc::now(); - - let new_secret = Secret { - name: req.name, - encrypted_value: encrypted_secret_value, - description: req.description, - change_reason: req.change_reason.clone(), - created_at: now, - last_modified_at: now, - created_by: user.get_email(), - last_modified_by: user.get_email(), - }; - - let created_secret = diesel::insert_into(secrets::table) - .values(&new_secret) - .returning(Secret::as_returning()) - .schema_name(&workspace_context.schema_name) - .get_result(&mut conn)?; + let created_secret = + conn.transaction::(|conn| { + // Share the workspace-row lock with key rotation and reload the key from the + // database. Holding this lock through the insert prevents a secret from being + // written with an old cached key while a rotation is in flight. + let workspace = workspaces::table + .find(( + &workspace_context.organisation_id.0, + &workspace_context.workspace_id.0, + )) + .for_share() + .get_result::(conn)?; + let encryption_key = + get_workspace_encryption_key(&workspace, &state.master_encryption_key)?; + let encrypted_secret_value = encrypt_secret(&req.value, &encryption_key) + .map_err(|e| bad_argument!("Encryption failed: {}", e))?; + let schema_name = SchemaName(workspace.workspace_schema_name); + let now = chrono::Utc::now(); + + let new_secret = Secret { + name: req.name, + encrypted_value: encrypted_secret_value, + description: req.description, + change_reason: req.change_reason.clone(), + created_at: now, + last_modified_at: now, + created_by: user_email.clone(), + last_modified_by: user_email, + }; + + diesel::insert_into(secrets::table) + .values(&new_secret) + .returning(Secret::as_returning()) + .schema_name(&schema_name) + .get_result(conn) + .map_err(Into::into) + })?; Ok(Json(SecretResponse::from(created_secret))) } @@ -218,35 +230,47 @@ async fn update_handler( let DbConnection(mut conn) = db_conn; let secret_name = path.into_inner(); let req_inner = req.into_inner(); + let user_email = user.get_email(); - let encrypted_value_opt = if let Some(ref plaintext_value) = req_inner.value { - let encryption_key = get_workspace_encryption_key( - &workspace_context, - &state.master_encryption_key, - )?; - Some( - encrypt_secret(plaintext_value, &encryption_key) - .map_err(|e| bad_argument!("Encryption failed: {}", e))?, - ) - } else { - None - }; - - let changeset = UpdateSecretChangeset { - encrypted_value: encrypted_value_opt, - description: req_inner.description, - change_reason: req_inner.change_reason, - }; - - let updated_secret = diesel::update(secrets::table) - .filter(secrets::name.eq(secret_name)) - .set(( - changeset, - secrets::last_modified_at.eq(chrono::Utc::now()), - secrets::last_modified_by.eq(user.get_email()), - )) - .schema_name(&workspace_context.schema_name) - .get_result::(&mut conn)?; + let updated_secret = + conn.transaction::(|conn| { + let workspace = workspaces::table + .find(( + &workspace_context.organisation_id.0, + &workspace_context.workspace_id.0, + )) + .for_share() + .get_result::(conn)?; + let encrypted_value = if let Some(ref plaintext_value) = req_inner.value { + let encryption_key = get_workspace_encryption_key( + &workspace, + &state.master_encryption_key, + )?; + Some( + encrypt_secret(plaintext_value, &encryption_key) + .map_err(|e| bad_argument!("Encryption failed: {}", e))?, + ) + } else { + None + }; + let schema_name = SchemaName(workspace.workspace_schema_name); + let changeset = UpdateSecretChangeset { + encrypted_value, + description: req_inner.description, + change_reason: req_inner.change_reason, + }; + + diesel::update(secrets::table) + .filter(secrets::name.eq(secret_name)) + .set(( + changeset, + secrets::last_modified_at.eq(chrono::Utc::now()), + secrets::last_modified_by.eq(user_email), + )) + .schema_name(&schema_name) + .get_result::(conn) + .map_err(Into::into) + })?; Ok(Json(SecretResponse::from(updated_secret))) } @@ -302,46 +326,42 @@ pub async fn rotate_master_key_handler( ) })?; - let all_workspaces: Vec = workspaces::table.load(&mut conn)?; - let user_email = user.get_email(); - let (workspaces_rotated, total_secrets_re_encrypted) = conn - .transaction::<(i64, i64), superposition::AppError, _>(|conn| { - let mut workspaces_rotated = 0i64; + let (rotated_workspaces, total_secrets_re_encrypted) = conn + .transaction::<(Vec, i64), superposition::AppError, _>(|conn| { + let workspace_targets: Vec<(String, String)> = workspaces::table + .select((workspaces::organisation_id, workspaces::workspace_name)) + .order(( + workspaces::organisation_id.asc(), + workspaces::workspace_name.asc(), + )) + .load(conn)?; + + let mut rotated_workspaces = Vec::with_capacity(workspace_targets.len()); let mut total_secrets_re_encrypted = 0i64; - for workspace in all_workspaces { - let workspace_context = WorkspaceContext { - workspace_id: WorkspaceId(workspace.workspace_name.clone()), - organisation_id: OrganisationId(workspace.organisation_id.clone()), - schema_name: SchemaName(workspace.workspace_schema_name.clone()), - settings: workspace, - }; - match rotate_workspace_encryption_key_helper( - &workspace_context, + for (organisation_id, workspace_name) in workspace_targets { + let (secrets_count, workspace) = rotate_workspace_encryption_key_helper( + &organisation_id, + &workspace_name, conn, master_encryption_key, &user_email, - ) { - Ok(secrets_count) => { - workspaces_rotated += 1; - total_secrets_re_encrypted += secrets_count; - } - Err(e) => { - log::error!( - "Failed to rotate keys for workspace {}: {}", - workspace_context.schema_name.0, - e - ); - return Err(e); - } - } + )?; + total_secrets_re_encrypted += secrets_count; + rotated_workspaces.push(workspace); } - Ok((workspaces_rotated, total_secrets_re_encrypted)) + Ok((rotated_workspaces, total_secrets_re_encrypted)) })?; + for workspace in &rotated_workspaces { + put_workspace_in_redis(workspace, &state.redis).await; + } + + let workspaces_rotated = rotated_workspaces.len() as i64; + log::info!( "Successfully rotated master encryption key. Rotated {} workspaces, re-encrypted {} secrets.", workspaces_rotated, diff --git a/crates/service_utils/src/encryption.rs b/crates/service_utils/src/encryption.rs index 355a2bb37..e9b9cc878 100644 --- a/crates/service_utils/src/encryption.rs +++ b/crates/service_utils/src/encryption.rs @@ -6,14 +6,17 @@ use base64::{Engine, engine::general_purpose}; use diesel::{ExpressionMethods, QueryDsl, RunQueryDsl}; use rand::RngCore; use secrecy::{ExposeSecret, SecretString}; -use superposition_macros::bad_argument; +use superposition_macros::unexpected_error; use superposition_types::{ DBConnection, - database::{schema::secrets, superposition_schema::superposition::workspaces}, + database::{ + models::Workspace, schema::secrets, + superposition_schema::superposition::workspaces, + }, result, }; -use crate::service::types::{AppEnv, EncryptionKey, SchemaName, WorkspaceContext}; +use crate::service::types::{AppEnv, EncryptionKey, SchemaName}; const NONCE_SIZE: usize = 12; const ENCRYPTION_KEY_BYTE_LENGTH: usize = 32; @@ -235,17 +238,14 @@ fn re_encrypt_secrets( return Ok(0); } + // Complete every cryptographic operation before mutating the database. This keeps + // crypto failures side-effect free and lets the surrounding transaction roll back + // database failures without leaving a partially rotated workspace. + let re_encrypted_secrets = + re_encrypt_secret_values(&all_secrets, current_key, new_key)?; let now = chrono::Utc::now(); - for (name, encrypted_value) in &all_secrets { - let decrypted_value = decrypt_secret(encrypted_value, current_key) - .map_err(|e| bad_argument!("Failed to decrypt secret '{}': {}", name, e))?; - - let new_encrypted_value = - encrypt_secret(decrypted_value.expose_secret(), new_key).map_err(|e| { - bad_argument!("Failed to encrypt secret '{}' with new key: {}", name, e) - })?; - + for (name, new_encrypted_value) in &re_encrypted_secrets { diesel::update(secrets::table.find(&name)) .set(( secrets::encrypted_value.eq(&new_encrypted_value), @@ -260,68 +260,109 @@ fn re_encrypt_secrets( Ok(all_secrets.len() as i64) } +fn re_encrypt_secret_values( + secrets: &[(String, String)], + current_key: &SecretString, + new_key: &SecretString, +) -> result::Result> { + secrets + .iter() + .map(|(name, encrypted_value)| { + let decrypted_value = + decrypt_secret(encrypted_value, current_key).map_err(|e| { + log::error!( + "Failed to decrypt secret '{}' during key rotation: {}", + name, + e + ); + unexpected_error!("Failed to decrypt secret during key rotation") + })?; + + let new_encrypted_value = + encrypt_secret(decrypted_value.expose_secret(), new_key).map_err( + |e| { + log::error!( + "Failed to encrypt secret '{}' during key rotation: {}", + name, + e + ); + unexpected_error!("Failed to encrypt secret during key rotation") + }, + )?; + + Ok((name.clone(), new_encrypted_value)) + }) + .collect() +} + +/// Rotate a workspace encryption key while holding a row lock on the canonical +/// workspace record. +/// +/// This function must be called from inside a database transaction. The row lock +/// serializes rotations for the same workspace and is held until that transaction +/// commits or rolls back. pub fn rotate_workspace_encryption_key_helper( - workspace_context: &WorkspaceContext, + organisation_id: &str, + workspace_name: &str, conn: &mut DBConnection, master_encryption_key: &EncryptionKey, user_email: &str, -) -> result::Result { - let current_key = decrypt_workspace_key( - &workspace_context.settings.encryption_key, - master_encryption_key, - ) - .map_err(|e| { - log::error!("Failed to decrypt current workspace key for {}", e); - bad_argument!("Failed to decrypt workspace encryption key") - })?; +) -> result::Result<(i64, Workspace)> { + let workspace = workspaces::dsl::workspaces + .find((organisation_id, workspace_name)) + .for_update() + .get_result::(conn)?; + let schema_name = SchemaName(workspace.workspace_schema_name.clone()); + + let current_key = + decrypt_workspace_key(&workspace.encryption_key, master_encryption_key).map_err( + |e| { + log::error!( + "Failed to decrypt current key for workspace '{}': {}", + workspace_name, + e + ); + unexpected_error!("Failed to decrypt workspace encryption key") + }, + )?; let new_key = generate_encryption_key(); let encrypted_new_key = encrypt_workspace_key(&new_key, &master_encryption_key.current_key).map_err( |e| { - log::error!("Failed to encrypt new workspace key: {}", e); - bad_argument!("Failed to encrypt new workspace key: {}", e) + log::error!( + "Failed to encrypt new key for workspace '{}': {}", + workspace_name, + e + ); + unexpected_error!("Failed to encrypt new workspace key") }, )?; - // Re-encrypt secrets if we have an existing key, otherwise this is initialization - let total_secrets_re_encrypted = re_encrypt_secrets( - conn, - &workspace_context.schema_name, - ¤t_key, - &new_key, - user_email, - ) - .unwrap_or_else(|_| { - log::info!( - "Initializing encryption key for workspace {}", - workspace_context.settings.workspace_name - ); - 0 // Just return i64, not Result - }); + let total_secrets_re_encrypted = + re_encrypt_secrets(conn, &schema_name, ¤t_key, &new_key, user_email)?; let rotation_time = chrono::Utc::now(); // Update workspace with new encrypted key - diesel::update(workspaces::dsl::workspaces.find(( - &workspace_context.organisation_id.0, - &workspace_context.settings.workspace_name, - ))) + let updated_workspace = diesel::update( + workspaces::dsl::workspaces.find((organisation_id, workspace_name)), + ) .set(( workspaces::dsl::encryption_key.eq(&encrypted_new_key), workspaces::dsl::key_rotated_at.eq(Some(rotation_time)), workspaces::dsl::last_modified_at.eq(rotation_time), workspaces::dsl::last_modified_by.eq(user_email), )) - .execute(conn)?; + .get_result::(conn)?; log::info!( "Rotated encryption key for workspace {}. total number of re-encrypted secrets {}", - workspace_context.settings.workspace_name, + workspace_name, total_secrets_re_encrypted ); - Ok(total_secrets_re_encrypted) + Ok((total_secrets_re_encrypted, updated_workspace)) } #[cfg(test)] @@ -376,4 +417,59 @@ mod tests { let decrypted = decrypt_with_fallback(&encrypted, &encryption_key).unwrap(); assert_eq!(plaintext, decrypted.expose_secret()); } + + #[test] + fn test_re_encrypt_secret_values() { + let current_key = generate_encryption_key(); + let new_key = generate_encryption_key(); + let secrets = vec![ + ( + "first".to_string(), + encrypt_secret("first value", ¤t_key).unwrap(), + ), + ( + "second".to_string(), + encrypt_secret("second value", ¤t_key).unwrap(), + ), + ]; + + let Ok(re_encrypted) = re_encrypt_secret_values(&secrets, ¤t_key, &new_key) + else { + panic!("secret re-encryption should succeed"); + }; + + assert_eq!(re_encrypted.len(), 2); + assert_eq!( + decrypt_secret(&re_encrypted[0].1, &new_key) + .unwrap() + .expose_secret(), + "first value" + ); + assert_eq!( + decrypt_secret(&re_encrypted[1].1, &new_key) + .unwrap() + .expose_secret(), + "second value" + ); + } + + #[test] + fn test_re_encrypt_secret_values_propagates_crypto_failure() { + let current_key = generate_encryption_key(); + let new_key = generate_encryption_key(); + let secrets = vec![ + ( + "valid".to_string(), + encrypt_secret("valid value", ¤t_key).unwrap(), + ), + ("corrupt".to_string(), "not-valid-ciphertext".to_string()), + ]; + + let rotation_result = re_encrypt_secret_values(&secrets, ¤t_key, &new_key); + + assert!(matches!( + rotation_result, + Err(result::AppError::UnexpectedError(_)) + )); + } } diff --git a/crates/service_utils/src/redis.rs b/crates/service_utils/src/redis.rs index 590dfefef..f885c8b15 100644 --- a/crates/service_utils/src/redis.rs +++ b/crates/service_utils/src/redis.rs @@ -4,7 +4,9 @@ use fred::{ }; use serde::{Serialize, de::DeserializeOwned}; use superposition_macros::unexpected_error; -use superposition_types::{DBConnection, result as superposition}; +use superposition_types::{ + DBConnection, database::models::Workspace, result as superposition, +}; use crate::{ db::{PgSchemaConnectionPool, run_query}, @@ -23,6 +25,41 @@ pub const EXPERIMENT_GROUPS_LAST_MODIFIED_KEY_SUFFIX: &str = pub const EXPERIMENT_CONFIG_LAST_MODIFIED_KEY_SUFFIX: &str = "::experiment_config::last_modified_at"; +/// Best-effort update of the canonical workspace cache entry. +pub async fn put_workspace_in_redis( + workspace: &Workspace, + redis_pool: &Option, +) { + let Some(pool) = redis_pool else { + log::debug!("Redis not configured, skipping workspace cache update"); + return; + }; + + let serialized = match serde_json::to_string(workspace) { + Ok(serialized) => serialized, + Err(error) => { + log::error!("Failed to serialize workspace for Redis: {}", error); + return; + } + }; + let key_ttl: i64 = get_from_env_or_default("REDIS_KEY_TTL", 604800); + + if let Err(error) = redis_set_data( + pool, + workspace.workspace_schema_name.clone(), + serialized, + Some(Expiration::EX(key_ttl)), + ) + .await + { + log::warn!( + "Failed to update Redis cache for workspace '{}': {}", + workspace.workspace_schema_name, + error + ); + } +} + /// Fetch data from Redis if available, else fall back to database call and write back to Redis /// if redis is disabled read from the database directly /// the fallback function is expected to return Result diff --git a/crates/superposition/src/workspace/handlers.rs b/crates/superposition/src/workspace/handlers.rs index f00df8c31..c52ba7117 100644 --- a/crates/superposition/src/workspace/handlers.rs +++ b/crates/superposition/src/workspace/handlers.rs @@ -11,7 +11,6 @@ use diesel::{ connection::SimpleConnection, r2d2::{ConnectionManager, PooledConnection}, }; -use fred::{prelude::KeysInterface, types::Expiration}; use regex::Regex; use service_utils::{ encryption::{ @@ -21,9 +20,8 @@ use service_utils::{ helpers::get_workspace, kronos_dispatch::setup_dispatcher, middlewares::auth_z::AuthZHandler, - service::types::{ - AppState, DbConnection, OrganisationId, SchemaName, WorkspaceContext, WorkspaceId, - }, + redis::put_workspace_in_redis, + service::types::{AppState, DbConnection, OrganisationId, SchemaName}, }; use superposition_derives::{authorized, declare_resource}; use superposition_macros::{bad_argument, db_error, unexpected_error, validation_error}; @@ -176,7 +174,7 @@ async fn create_handler( Ok(inserted_workspace.remove(0)) })?; - put_workspace_in_redis(&created_workspace, &state, &workspace_schema_name.0).await; + put_workspace_in_redis(&created_workspace, &state.redis).await; if state.kronos_workspace.is_none() { setup_dispatcher( @@ -284,7 +282,7 @@ async fn update_handler( })?; let workspace = get_workspace(&schema_name, &mut conn)?; - put_workspace_in_redis(&workspace, &app_state, &schema_name.0).await; + put_workspace_in_redis(&workspace, &app_state.redis).await; if let Some(old_email) = old_email { let _ = authz_handler @@ -305,35 +303,6 @@ async fn update_handler( Ok(Json(response)) } -async fn put_workspace_in_redis( - workspace: &Workspace, - state: &Data, - schema_name: &str, -) { - let redis_pool = match &state.redis { - Some(pool) => pool, - None => { - log::debug!("Redis not configured, skipping workspace cache update"); - return; - } - }; - - let key_ttl: i64 = - service_utils::helpers::get_from_env_or_default("REDIS_KEY_TTL", 604800); - let expiration = Some(Expiration::EX(key_ttl)); - - if let Ok(serialized) = serde_json::to_string(workspace) { - let client = redis_pool.next_connected(); - - if let Err(e) = client - .set::<(), &str, String>(schema_name, serialized, expiration, None, false) - .await - { - log::warn!("Failed to update Redis cache with workspace: {}", e); - } - }; -} - // TODO: Add ABAC based filtering for this endpoint, currently it returns all the workspaces in the org, we need to filter it based on the permissions of the user #[authorized] #[get("")] @@ -483,7 +452,7 @@ async fn migrate_schema_handler( } let workspace = get_workspace(&schema_name, &mut conn)?; - put_workspace_in_redis(&workspace, &state, &schema_name.0).await; + put_workspace_in_redis(&workspace, &state.redis).await; let response = WorkspaceResponse::from(workspace); Ok(Json(response)) @@ -507,27 +476,21 @@ pub async fn rotate_encryption_key_handler( )); }; - let schema_name = SchemaName(format!("{}_{}", *org_id, workspace_name.into_inner())); - let workspace = get_workspace(&schema_name, &mut conn)?; - let workspace_context = WorkspaceContext { - schema_name: schema_name.clone(), - organisation_id: org_id, - workspace_id: WorkspaceId(workspace.workspace_name.clone()), - settings: workspace, - }; + let workspace_name = workspace_name.into_inner(); + let user_email = user.get_email(); - let total_secrets_re_encrypted = conn - .transaction::(|conn| { + let (total_secrets_re_encrypted, workspace) = conn + .transaction::<(i64, Workspace), superposition::AppError, _>(|conn| { rotate_workspace_encryption_key_helper( - &workspace_context, + &org_id.0, + &workspace_name, conn, master_encryption_key, - &user.get_email(), + &user_email, ) })?; - let workspace = get_workspace(&schema_name, &mut conn)?; - put_workspace_in_redis(&workspace, &state, &schema_name.0).await; + put_workspace_in_redis(&workspace, &state.redis).await; Ok(Json(KeyRotationResponse { total_secrets_re_encrypted,