diff --git a/services/backup/src/constants.rs b/services/backup/src/constants.rs index 506931d12..6eb32c6ee 100644 --- a/services/backup/src/constants.rs +++ b/services/backup/src/constants.rs @@ -1,43 +1,44 @@ // Assorted constants pub const MPSC_CHANNEL_BUFFER_CAPACITY: usize = 1; pub const ID_SEPARATOR: &str = ":"; pub const ATTACHMENT_HOLDER_SEPARATOR: &str = ";"; pub const WS_FRAME_SIZE: usize = 1_048_576; // 1MiB pub const LOG_DEFAULT_PAGE_SIZE: i32 = 20; pub const LOG_BACKUP_ID_SEPARATOR: &str = "#"; // Configuration defaults pub const DEFAULT_HTTP_PORT: u16 = 50052; pub const DEFAULT_BLOB_SERVICE_URL: &str = "http://localhost:50053"; // Environment variable names pub const LOG_LEVEL_ENV_VAR: &str = tracing_subscriber::filter::EnvFilter::DEFAULT_ENV; // DynamoDB constants pub mod backup_table { pub const TABLE_NAME: &str = "backup-service-backup"; pub const CREATED_INDEX: &str = "userID-created-index"; pub mod attr { pub const USER_ID: &str = "userID"; pub const BACKUP_ID: &str = "backupID"; pub const CREATED: &str = "created"; pub const USER_DATA: &str = "userData"; pub const USER_KEYS: &str = "userKeys"; pub const ATTACHMENTS: &str = "attachments"; + pub const SIWE_BACKUP_MSG: &str = "siweBackupMsg"; } } pub mod log_table { pub const TABLE_NAME: &str = "backup-service-log"; pub mod attr { pub const BACKUP_ID: &str = "backupID"; pub const LOG_ID: &str = "logID"; pub const CONTENT_DB: &str = "content"; pub const CONTENT_BLOB_INFO: &str = "blobInfo"; pub const ATTACHMENTS: &str = "attachments"; } } diff --git a/services/backup/src/database/backup_item.rs b/services/backup/src/database/backup_item.rs index 21b48b008..efcf591ed 100644 --- a/services/backup/src/database/backup_item.rs +++ b/services/backup/src/database/backup_item.rs @@ -1,203 +1,218 @@ use crate::constants::backup_table; use aws_sdk_dynamodb::types::AttributeValue; use chrono::{DateTime, Utc}; use comm_lib::{ blob::{client::BlobServiceClient, types::BlobInfo}, - database::{AttributeTryInto, DBItemError, TryFromAttribute}, + database::{ + AttributeExtractor, AttributeTryInto, DBItemError, TryFromAttribute, + }, }; use std::collections::HashMap; #[derive(Clone, Debug)] pub struct BackupItem { pub user_id: String, pub backup_id: String, pub created: DateTime, pub user_keys: BlobInfo, pub user_data: BlobInfo, pub attachments: Vec, + pub siwe_backup_msg: Option, } impl BackupItem { pub fn new( user_id: String, backup_id: String, user_keys: BlobInfo, user_data: BlobInfo, attachments: Vec, + siwe_backup_msg: Option, ) -> Self { BackupItem { user_id, backup_id, created: chrono::Utc::now(), user_keys, user_data, attachments, + siwe_backup_msg, } } pub fn revoke_holders(&self, blob_client: &BlobServiceClient) { blob_client.schedule_revoke_holder( &self.user_keys.blob_hash, &self.user_keys.holder, ); blob_client.schedule_revoke_holder( &self.user_data.blob_hash, &self.user_data.holder, ); for attachment_info in &self.attachments { blob_client.schedule_revoke_holder( &attachment_info.blob_hash, &attachment_info.holder, ); } } pub fn item_key( user_id: &str, backup_id: &str, ) -> HashMap { HashMap::from([ ( backup_table::attr::USER_ID.to_string(), AttributeValue::S(user_id.to_string()), ), ( backup_table::attr::BACKUP_ID.to_string(), AttributeValue::S(backup_id.to_string()), ), ]) } } impl From for HashMap { fn from(value: BackupItem) -> Self { let mut attrs = HashMap::from([ ( backup_table::attr::USER_ID.to_string(), AttributeValue::S(value.user_id), ), ( backup_table::attr::BACKUP_ID.to_string(), AttributeValue::S(value.backup_id), ), ( backup_table::attr::CREATED.to_string(), AttributeValue::S(value.created.to_rfc3339()), ), ( backup_table::attr::USER_KEYS.to_string(), value.user_keys.into(), ), ( backup_table::attr::USER_DATA.to_string(), value.user_data.into(), ), ]); if !value.attachments.is_empty() { attrs.insert( backup_table::attr::ATTACHMENTS.to_string(), AttributeValue::L( value .attachments .into_iter() .map(AttributeValue::from) .collect(), ), ); } + if let Some(siwe_backup_msg_value) = value.siwe_backup_msg { + attrs.insert( + backup_table::attr::SIWE_BACKUP_MSG.to_string(), + AttributeValue::S(siwe_backup_msg_value), + ); + } attrs } } impl TryFrom> for BackupItem { type Error = DBItemError; fn try_from( mut value: HashMap, ) -> Result { let user_id = String::try_from_attr( backup_table::attr::USER_ID, value.remove(backup_table::attr::USER_ID), )?; let backup_id = String::try_from_attr( backup_table::attr::BACKUP_ID, value.remove(backup_table::attr::BACKUP_ID), )?; let created = DateTime::::try_from_attr( backup_table::attr::CREATED, value.remove(backup_table::attr::CREATED), )?; let user_keys = BlobInfo::try_from_attr( backup_table::attr::USER_KEYS, value.remove(backup_table::attr::USER_KEYS), )?; let user_data = BlobInfo::try_from_attr( backup_table::attr::USER_DATA, value.remove(backup_table::attr::USER_DATA), )?; let attachments = value.remove(backup_table::attr::ATTACHMENTS); let attachments = if attachments.is_some() { attachments.attr_try_into(backup_table::attr::ATTACHMENTS)? } else { Vec::new() }; + let siwe_backup_msg: Option = + value.take_attr(backup_table::attr::SIWE_BACKUP_MSG)?; + Ok(BackupItem { user_id, backup_id, created, user_keys, user_data, attachments, + siwe_backup_msg, }) } } /// Corresponds to the items in the [`crate::constants::BACKUP_TABLE_INDEX_USERID_CREATED`] /// global index #[derive(Clone, Debug)] pub struct OrderedBackupItem { pub user_id: String, pub created: DateTime, pub backup_id: String, pub user_keys: BlobInfo, } impl TryFrom> for OrderedBackupItem { type Error = DBItemError; fn try_from( mut value: HashMap, ) -> Result { let user_id = String::try_from_attr( backup_table::attr::USER_ID, value.remove(backup_table::attr::USER_ID), )?; let created = DateTime::::try_from_attr( backup_table::attr::CREATED, value.remove(backup_table::attr::CREATED), )?; let backup_id = String::try_from_attr( backup_table::attr::BACKUP_ID, value.remove(backup_table::attr::BACKUP_ID), )?; let user_keys = BlobInfo::try_from_attr( backup_table::attr::USER_KEYS, value.remove(backup_table::attr::USER_KEYS), )?; Ok(OrderedBackupItem { user_id, created, backup_id, user_keys, }) } } diff --git a/services/backup/src/http/handlers/backup.rs b/services/backup/src/http/handlers/backup.rs index 45eea5e29..1e976d720 100644 --- a/services/backup/src/http/handlers/backup.rs +++ b/services/backup/src/http/handlers/backup.rs @@ -1,314 +1,327 @@ use actix_web::{ error::ErrorBadRequest, web::{self, Bytes}, HttpResponse, Responder, }; use comm_lib::{ auth::UserIdentity, backup::LatestBackupIDResponse, blob::{client::BlobServiceClient, types::BlobInfo}, http::multipart::{get_named_text_field, get_text_field}, tools::Defer, }; use std::convert::Infallible; use tokio_stream::{wrappers::ReceiverStream, StreamExt}; use tracing::{info, instrument, trace, warn}; use crate::{ database::{backup_item::BackupItem, DatabaseClient}, error::BackupError, }; #[instrument(skip_all, fields(backup_id))] pub async fn upload( user: UserIdentity, blob_client: web::Data, db_client: web::Data, mut multipart: actix_multipart::Multipart, ) -> actix_web::Result { let backup_id = get_named_text_field("backup_id", &mut multipart).await?; tracing::Span::current().record("backup_id", &backup_id); info!("Backup data upload started"); let (user_keys_blob_info, user_keys_revoke) = forward_field_to_blob( &mut multipart, &blob_client, "user_keys_hash", "user_keys", ) .await?; let (user_data_blob_info, user_data_revoke) = forward_field_to_blob( &mut multipart, &blob_client, "user_data_hash", "user_data", ) .await?; let attachments_hashes: Vec = match get_text_field(&mut multipart).await? { Some((name, attachments)) => { if name != "attachments" { warn!( name, "Malformed request: 'attachments' text field expected." ); return Err(ErrorBadRequest("Bad request")); } attachments.lines().map(ToString::to_string).collect() } None => Vec::new(), }; let mut attachments = Vec::new(); let mut attachments_revokes = Vec::new(); for attachment_hash in attachments_hashes { let (holder, revoke) = create_attachment_holder(&attachment_hash, &blob_client).await?; attachments.push(BlobInfo { blob_hash: attachment_hash, holder, }); attachments_revokes.push(revoke); } + let siwe_backup_msg_option: Option = + match get_text_field(&mut multipart).await? { + Some((name, siwe_backup_msg)) => { + if name == "siwe_backup_msg" { + Some(siwe_backup_msg) + } else { + None + } + } + _ => None, + }; + let item = BackupItem::new( user.user_id.clone(), backup_id, user_keys_blob_info, user_data_blob_info, attachments, + siwe_backup_msg_option, ); db_client .put_backup_item(item) .await .map_err(BackupError::from)?; user_keys_revoke.cancel(); user_data_revoke.cancel(); for attachment_revoke in attachments_revokes { attachment_revoke.cancel(); } db_client .remove_old_backups(&user.user_id, &blob_client) .await .map_err(BackupError::from)?; Ok(HttpResponse::Ok().finish()) } #[instrument(skip_all, fields(hash_field_name, data_field_name))] async fn forward_field_to_blob<'revoke, 'blob: 'revoke>( multipart: &mut actix_multipart::Multipart, blob_client: &'blob web::Data, hash_field_name: &str, data_field_name: &str, ) -> actix_web::Result<(BlobInfo, Defer<'revoke>)> { trace!("Reading blob fields: {hash_field_name:?}, {data_field_name:?}"); let blob_hash = get_named_text_field(hash_field_name, multipart).await?; let Some(mut field) = multipart.try_next().await? else { warn!("Malformed request: expected a field."); return Err(ErrorBadRequest("Bad request"))?; }; if field.name() != data_field_name { warn!( hash_field_name, "Malformed request: '{data_field_name}' data field expected." ); return Err(ErrorBadRequest("Bad request"))?; } let blob_info = BlobInfo { blob_hash, holder: uuid::Uuid::new_v4().to_string(), }; // [`actix_multipart::Multipart`] isn't [`std::marker::Send`], and so we cannot pass it to the blob client directly. // Instead we have to forward it to a channel and create stream from the receiver. let (tx, rx) = tokio::sync::mpsc::channel(1); let receive_promise = async move { trace!("Receiving blob data"); // [`actix_multipart::MultipartError`] isn't [`std::marker::Send`] so we return it here, and pass [`Infallible`] // as the error to the channel while let Some(chunk) = field.try_next().await? { if let Err(err) = tx.send(Result::::Ok(chunk)).await { warn!("Error when sending data through a channel: '{err}'"); // Error here means that the channel has been closed from the blob client side. We don't want to return an error // here, because `tokio::try_join!` only returns the first error it receives and we want to prioritize the backup // client error. break; } } trace!("Finished receiving blob data"); Result::<(), actix_web::Error>::Ok(()) }; let data_stream = ReceiverStream::new(rx); let send_promise = async { blob_client .simple_put(&blob_info.blob_hash, &blob_info.holder, data_stream) .await .map_err(BackupError::from)?; Ok(()) }; tokio::try_join!(receive_promise, send_promise)?; let revoke_info = blob_info.clone(); let revoke_holder = Defer::new(|| { blob_client .schedule_revoke_holder(revoke_info.blob_hash, revoke_info.holder) }); Ok((blob_info, revoke_holder)) } #[instrument(skip_all)] async fn create_attachment_holder<'revoke, 'blob: 'revoke>( attachment: &str, blob_client: &'blob web::Data, ) -> Result<(String, Defer<'revoke>), BackupError> { let holder = uuid::Uuid::new_v4().to_string(); if !blob_client .assign_holder(attachment, &holder) .await .map_err(BackupError::from)? { warn!("Blob attachment with hash {attachment:?} doesn't exist"); } let revoke_hash = attachment.to_string(); let revoke_holder = holder.clone(); let revoke_holder = Defer::new(|| { blob_client.schedule_revoke_holder(revoke_hash, revoke_holder) }); Ok((holder, revoke_holder)) } #[instrument(skip_all, fields(backup_id = %path))] pub async fn download_user_keys( user: UserIdentity, path: web::Path, blob_client: web::Data, db_client: web::Data, ) -> actix_web::Result { info!("Download user keys request"); let backup_id = path.into_inner(); download_user_blob( |item| &item.user_keys, &user.user_id, &backup_id, blob_client, db_client, ) .await } #[instrument(skip_all, fields(backup_id = %path))] pub async fn download_user_data( user: UserIdentity, path: web::Path, blob_client: web::Data, db_client: web::Data, ) -> actix_web::Result { info!("Download user data request"); let backup_id = path.into_inner(); download_user_blob( |item| &item.user_data, &user.user_id, &backup_id, blob_client, db_client, ) .await } pub async fn download_user_blob( data_extractor: impl FnOnce(&BackupItem) -> &BlobInfo, user_id: &str, backup_id: &str, blob_client: web::Data, db_client: web::Data, ) -> actix_web::Result { let backup_item = db_client .find_backup_item(user_id, backup_id) .await .map_err(BackupError::from)? .ok_or(BackupError::NoBackup)?; let stream = blob_client .get(&data_extractor(&backup_item).blob_hash) .await .map_err(BackupError::from)?; Ok( HttpResponse::Ok() .content_type("application/octet-stream") .streaming(stream), ) } #[instrument(skip_all, fields(username = %path))] pub async fn get_latest_backup_id( path: web::Path, db_client: web::Data, ) -> actix_web::Result { let username = path.into_inner(); // Treat username as user_id in the initial version let user_id = username; let Some(backup_item) = db_client .find_last_backup_item(&user_id) .await .map_err(BackupError::from)? else { return Err(BackupError::NoBackup.into()); }; let response = LatestBackupIDResponse { backup_id: backup_item.backup_id, }; Ok(web::Json(response)) } #[instrument(skip_all, fields(username = %path))] pub async fn download_latest_backup_keys( path: web::Path, db_client: web::Data, blob_client: web::Data, ) -> actix_web::Result { let username = path.into_inner(); // Treat username as user_id in the initial version let user_id = username; let Some(backup_item) = db_client .find_last_backup_item(&user_id) .await .map_err(BackupError::from)? else { return Err(BackupError::NoBackup.into()); }; let stream = blob_client .get(&backup_item.user_keys.blob_hash) .await .map_err(BackupError::from)?; Ok( HttpResponse::Ok() .content_type("application/octet-stream") .streaming(stream), ) }