diff --git a/keyserver/addons/rust-node-addon/src/identity_client/login.rs b/keyserver/addons/rust-node-addon/src/identity_client/login.rs index 1deaf364b..6434a195b 100644 --- a/keyserver/addons/rust-node-addon/src/identity_client/login.rs +++ b/keyserver/addons/rust-node-addon/src/identity_client/login.rs @@ -1,148 +1,149 @@ use super::*; use comm_opaque2::client::Login; use grpc_clients::identity::protos::{ unauth::SecondaryDeviceKeysUploadRequest, unauthenticated::{OpaqueLoginFinishRequest, OpaqueLoginStartRequest}, }; use tracing::debug; #[napi] #[instrument(skip_all)] +#[allow(clippy::too_many_arguments)] pub async fn login_user( username: String, password: String, signed_identity_keys_blob: SignedIdentityKeysBlob, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, force: Option, ) -> Result { debug!("Attempting to log in user: {}", username); // Set up the gRPC client that will be used to talk to the Identity service let mut identity_client = get_identity_client().await?; // Start OPAQUE registration and send initial registration request let mut client_login = Login::new(); let opaque_login_request = client_login .start(&password) .map_err(|_| Error::from_reason("Failed to create opaque login request"))?; let login_start_request = OpaqueLoginStartRequest { opaque_login_request, username, device_key_upload: Some(DeviceKeyUpload { device_key_info: Some(IdentityKeyInfo { payload: signed_identity_keys_blob.payload, payload_signature: signed_identity_keys_blob.signature, }), content_upload: Some(Prekey { prekey: content_prekey, prekey_signature: content_prekey_signature, }), notif_upload: Some(Prekey { prekey: notif_prekey, prekey_signature: notif_prekey_signature, }), one_time_content_prekeys: Vec::new(), one_time_notif_prekeys: Vec::new(), device_type: DeviceType::Keyserver.into(), }), force, }; debug!("Starting login to identity service"); let response = identity_client .log_in_password_user_start(login_start_request) .await .map_err(handle_grpc_error)?; debug!("Received login response from identity service"); let login_start_response = response.into_inner(); let opaque_login_upload = client_login .finish(&login_start_response.opaque_login_response) .map_err(|_| Error::from_reason("Failed to finish opaque login request"))?; let login_finish_request = OpaqueLoginFinishRequest { session_id: login_start_response.session_id, opaque_login_upload, }; debug!("Attempting to finalize opaque login exchange with identity service"); let login_finish_response = identity_client .log_in_password_user_finish(login_finish_request) .await .map_err(handle_grpc_error)? .into_inner(); debug!("Finished login with identity service"); let user_info = UserLoginInfo { user_id: login_finish_response.user_id, access_token: login_finish_response.access_token, }; Ok(user_info) } #[napi] #[instrument(skip_all)] #[allow(clippy::too_many_arguments)] pub async fn upload_secondary_device_keys_and_log_in( user_id: String, nonce: String, nonce_signature: String, signed_identity_keys_blob: SignedIdentityKeysBlob, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, ) -> Result { let device_key_upload = DeviceKeyUpload { device_key_info: Some(IdentityKeyInfo { payload: signed_identity_keys_blob.payload, payload_signature: signed_identity_keys_blob.signature, }), content_upload: Some(Prekey { prekey: content_prekey, prekey_signature: content_prekey_signature, }), notif_upload: Some(Prekey { prekey: notif_prekey, prekey_signature: notif_prekey_signature, }), one_time_content_prekeys: content_one_time_keys, one_time_notif_prekeys: notif_one_time_keys, device_type: DeviceType::Keyserver.into(), }; let mut identity_client = get_identity_client().await?; let request = SecondaryDeviceKeysUploadRequest { user_id, nonce, nonce_signature, device_key_upload: Some(device_key_upload), }; let response = identity_client .upload_keys_for_registered_device_and_log_in(request) .await .map_err(|_| { Error::from_reason( "Failed to upload keys for registered device and log in", ) })? .into_inner(); let user_info = UserLoginInfo { user_id: response.user_id, access_token: response.access_token, }; Ok(user_info) } diff --git a/keyserver/addons/rust-node-addon/src/identity_client/mod.rs b/keyserver/addons/rust-node-addon/src/identity_client/mod.rs index 526e60596..593b6ea31 100644 --- a/keyserver/addons/rust-node-addon/src/identity_client/mod.rs +++ b/keyserver/addons/rust-node-addon/src/identity_client/mod.rs @@ -1,218 +1,219 @@ pub mod add_reserved_usernames; pub mod config; pub mod get_farcaster_users; pub mod get_inbound_keys_for_user; pub mod login; pub mod nonce; pub mod prekey; pub mod register_user; pub mod remove_reserved_usernames; pub mod upload_one_time_keys; use client_proto::identity_client_service_client::IdentityClientServiceClient; use client_proto::{ AddReservedUsernamesRequest, DeviceKeyUpload, DeviceType, IdentityKeyInfo, Prekey, RegistrationFinishRequest, RegistrationStartRequest, RemoveReservedUsernameRequest, }; use config::get_identity_service_config; use generated::CODE_VERSION; use grpc_clients::identity::authenticated::ChainedInterceptedAuthClient; use grpc_clients::identity::protos::authenticated::{ InboundKeyInfo, UploadOneTimeKeysRequest, }; use grpc_clients::identity::protos::unauthenticated as client_proto; use grpc_clients::identity::shared::CodeVersionLayer; use lazy_static::lazy_static; use napi::bindgen_prelude::*; use serde::{Deserialize, Serialize}; use std::env::var; use tonic::codegen::InterceptedService; use tonic::transport::Channel; use tracing::{self, info, instrument, warn, Level}; use tracing_subscriber::EnvFilter; mod generated { // We get the CODE_VERSION from this generated file include!(concat!(env!("OUT_DIR"), "/version.rs")); } const DEVICE_TYPE: &str = "keyserver"; const ENV_NODE_ENV: &str = "NODE_ENV"; const ENV_DEVELOPMENT: &str = "development"; lazy_static! { static ref IDENTITY_SERVICE_CONFIG: IdentityServiceConfig = { let filter = EnvFilter::builder() .with_default_directive(Level::INFO.into()) .with_env_var(EnvFilter::DEFAULT_ENV) .from_env_lossy(); let subscriber = tracing_subscriber::fmt().with_env_filter(filter).finish(); tracing::subscriber::set_global_default(subscriber) .expect("Unable to configure tracing"); get_identity_service_config() .unwrap_or_else(|_| IdentityServiceConfig::default()) }; } #[derive(Serialize, Deserialize)] #[serde(rename_all = "camelCase")] struct IdentityServiceConfig { identity_socket_addr: String, } impl Default for IdentityServiceConfig { fn default() -> Self { info!("Using default identity configuration based on NODE_ENV"); const DEV_SOCKET_ADDR: &str = "https://identity.staging.commtechnologies.org:50054"; const PROD_SOCKET_ADDR: &str = "https://identity.commtechnologies.org:50054"; let default_socket_addr = match var(ENV_NODE_ENV) { Ok(val) if val == ENV_DEVELOPMENT => DEV_SOCKET_ADDR, _ => PROD_SOCKET_ADDR, } .to_string(); Self { identity_socket_addr: default_socket_addr, } } } async fn get_identity_client() -> Result< IdentityClientServiceClient>, > { info!("Connecting to identity service"); grpc_clients::identity::get_unauthenticated_client( &IDENTITY_SERVICE_CONFIG.identity_socket_addr, CODE_VERSION, DEVICE_TYPE.to_string(), ) .await .map_err(|e| { Error::new( Status::GenericFailure, format!("Unable to connect to identity service: {}", e), ) }) } async fn get_authenticated_identity_client( user_id: String, device_id: String, access_token: String, ) -> Result { info!("Connecting to identity service"); grpc_clients::identity::get_auth_client( &IDENTITY_SERVICE_CONFIG.identity_socket_addr, user_id, device_id, access_token, CODE_VERSION, DEVICE_TYPE.to_string(), ) .await .map_err(|e| { Error::new( Status::GenericFailure, format!("Unable to connect to identity service: {}", e), ) }) } #[napi(object)] pub struct SignedIdentityKeysBlob { pub payload: String, pub signature: String, } #[napi(object)] pub struct UserLoginInfo { pub user_id: String, pub access_token: String, } pub struct DeviceInboundKeyInfo { pub payload: String, pub payload_signature: String, pub content_prekey: String, pub content_prekey_signature: String, pub notif_prekey: String, pub notif_prekey_signature: String, } impl TryFrom for DeviceInboundKeyInfo { type Error = Error; fn try_from(key_info: InboundKeyInfo) -> Result { let identity_info = key_info .identity_info .ok_or(Error::from_status(Status::GenericFailure))?; let IdentityKeyInfo { payload, payload_signature, } = identity_info; let content_prekey = key_info .content_prekey .ok_or(Error::from_status(Status::GenericFailure))?; let Prekey { prekey: content_prekey_value, prekey_signature: content_prekey_signature, } = content_prekey; let notif_prekey = key_info .notif_prekey .ok_or(Error::from_status(Status::GenericFailure))?; let Prekey { prekey: notif_prekey_value, prekey_signature: notif_prekey_signature, } = notif_prekey; Ok(Self { payload, payload_signature, content_prekey: content_prekey_value, content_prekey_signature, notif_prekey: notif_prekey_value, notif_prekey_signature, }) } } #[napi(object)] pub struct InboundKeyInfoResponse { pub payload: String, pub payload_signature: String, pub content_prekey: String, pub content_prekey_signature: String, pub notif_prekey: String, pub notif_prekey_signature: String, pub username: Option, pub wallet_address: Option, } pub fn handle_grpc_error(error: tonic::Status) -> napi::Error { warn!("Received error: {}", error.message()); Error::new(Status::GenericFailure, error.message()) } #[cfg(test)] mod tests { use super::CODE_VERSION; #[test] + #[allow(clippy::assertions_on_constants)] fn test_code_version_exists() { assert!(CODE_VERSION > 0); } } diff --git a/keyserver/addons/rust-node-addon/src/identity_client/register_user.rs b/keyserver/addons/rust-node-addon/src/identity_client/register_user.rs index 25de31236..7834b1273 100644 --- a/keyserver/addons/rust-node-addon/src/identity_client/register_user.rs +++ b/keyserver/addons/rust-node-addon/src/identity_client/register_user.rs @@ -1,86 +1,87 @@ use super::*; use tracing::{debug, warn}; #[napi] #[instrument(skip_all)] +#[allow(clippy::too_many_arguments)] pub async fn register_user( username: String, password: String, signed_identity_keys_blob: SignedIdentityKeysBlob, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, ) -> Result { debug!("Attempting to register user: {}", username); // Set up the gRPC client that will be used to talk to the Identity service let mut identity_client = get_identity_client().await?; // Start OPAQUE registration and send initial registration request let mut opaque_registration = comm_opaque2::client::Registration::new(); let opaque_registration_request = opaque_registration .start(&password) .map_err(|_| Error::from_status(Status::GenericFailure))?; let device_key_upload = DeviceKeyUpload { device_key_info: Some(IdentityKeyInfo { payload: signed_identity_keys_blob.payload, payload_signature: signed_identity_keys_blob.signature, }), content_upload: Some(Prekey { prekey: content_prekey, prekey_signature: content_prekey_signature, }), notif_upload: Some(Prekey { prekey: notif_prekey, prekey_signature: notif_prekey_signature, }), one_time_content_prekeys: content_one_time_keys, one_time_notif_prekeys: notif_one_time_keys, device_type: DeviceType::Keyserver.into(), }; let registration_start_request = RegistrationStartRequest { opaque_registration_request, username, device_key_upload: Some(device_key_upload), farcaster_id: None, initial_device_list: "".to_string(), }; // Finish OPAQUE registration and send final registration request let response = identity_client .register_password_user_start(registration_start_request) .await .map_err(handle_grpc_error)?; debug!("Received registration start response"); let registration_start_response = response.into_inner(); let opaque_registration_upload = opaque_registration .finish( &password, ®istration_start_response.opaque_registration_response, ) .map_err(|_| Error::from_status(Status::GenericFailure))?; let registration_finish_request = RegistrationFinishRequest { session_id: registration_start_response.session_id, opaque_registration_upload, }; let registration_response = identity_client .register_password_user_finish(registration_finish_request) .await .map_err(handle_grpc_error)? .into_inner(); let user_info = UserLoginInfo { user_id: registration_response.user_id, access_token: registration_response.access_token, }; Ok(user_info) } diff --git a/native/native_rust_library/src/backup.rs b/native/native_rust_library/src/backup.rs index d33dade10..2ef08bbf8 100644 --- a/native/native_rust_library/src/backup.rs +++ b/native/native_rust_library/src/backup.rs @@ -1,508 +1,508 @@ mod compaction_upload_promises; mod file_info; mod upload_handler; use crate::argon2_tools::{compute_backup_key, compute_backup_key_str}; use crate::constants::{aes, secure_store}; use crate::ffi::{ create_main_compaction, get_backup_directory_path, get_backup_user_keys_file_path, get_siwe_backup_message_path, restore_from_backup_log, restore_from_main_compaction, secure_store_get, string_callback, void_callback, }; use crate::utils::future_manager; use crate::utils::jsi_callbacks::handle_string_result_as_callback; use crate::BACKUP_SOCKET_ADDR; use crate::RUNTIME; use backup_client::{ BackupClient, BackupDescriptor, LatestBackupIDResponse, RequestedData, TryStreamExt, UserIdentity, }; use serde::{Deserialize, Serialize}; use siwe::Message; use std::error::Error; use std::path::PathBuf; pub mod ffi { use super::*; pub use upload_handler::ffi::*; pub fn create_backup( backup_id: String, backup_secret: String, pickle_key: String, pickled_account: String, siwe_backup_msg: String, promise_id: u32, ) { compaction_upload_promises::insert(backup_id.clone(), promise_id); RUNTIME.spawn(async move { let result = create_userkeys_compaction( backup_id.clone(), backup_secret, pickle_key, pickled_account, ) .await .map_err(|err| err.to_string()); if let Err(err) = result { compaction_upload_promises::resolve(&backup_id, Err(err)); return; } if !siwe_backup_msg.is_empty() { if let Err(err) = create_siwe_backup_msg_compaction(&backup_id, siwe_backup_msg).await { compaction_upload_promises::resolve(&backup_id, Err(err.to_string())); return; } } let (future_id, future) = future_manager::new_future::<()>().await; create_main_compaction(&backup_id, future_id); if let Err(err) = future.await { compaction_upload_promises::resolve(&backup_id, Err(err)); tokio::spawn(upload_handler::compaction::cleanup_files(backup_id)); return; } trigger_backup_file_upload(); // The promise will be resolved when the backup is uploaded }); } pub fn restore_backup( backup_secret: String, backup_id: String, promise_id: u32, ) { RUNTIME.spawn(async move { let backup_id = Option::from(backup_id).filter(|it| !it.is_empty()); let result = download_backup(backup_secret, backup_id) .await .map_err(|err| err.to_string()); let result = match result { Ok(result) => result, Err(error) => { void_callback(error, promise_id); return; } }; let (future_id, future) = future_manager::new_future::<()>().await; restore_from_main_compaction( &result.backup_restoration_path.to_string_lossy(), &result.backup_data_key, future_id, ); if let Err(error) = future.await { void_callback(error, promise_id); return; } if let Err(error) = download_and_apply_logs(&result.backup_id, result.backup_log_data_key) .await { void_callback(error.to_string(), promise_id); return; } void_callback(String::new(), promise_id); }); } pub fn retrieve_backup_keys(backup_secret: String, promise_id: u32) { RUNTIME.spawn(async move { let latest_backup_id_response = download_latest_backup_id() .await .map_err(|err| err.to_string()); let backup_id = match latest_backup_id_response { Ok(result) => result.backup_id, Err(error) => { string_callback(error, promise_id, "".to_string()); return; } }; let result = download_backup_keys(backup_id, backup_secret) .await .map_err(|err| err.to_string()); let result = match result { Ok(result) => result, Err(error) => { string_callback(error, promise_id, "".to_string()); return; } }; let serialize_result = serde_json::to_string(&result); handle_string_result_as_callback(serialize_result, promise_id); }); } pub fn restore_backup_data( backup_id: String, backup_data_key: String, backup_log_data_key: String, promise_id: u32, ) { RUNTIME.spawn(async move { let backup_keys = BackupKeysResult { backup_id, backup_data_key, backup_log_data_key, }; let result = download_backup_data(backup_keys) .await .map_err(|err| err.to_string()); let result = match result { Ok(result) => result, Err(error) => { void_callback(error, promise_id); return; } }; let (future_id, future) = future_manager::new_future::<()>().await; restore_from_main_compaction( &result.backup_restoration_path.to_string_lossy(), &result.backup_data_key, future_id, ); if let Err(error) = future.await { void_callback(error, promise_id); return; } if let Err(error) = download_and_apply_logs(&result.backup_id, result.backup_log_data_key) .await { void_callback(error.to_string(), promise_id); return; } void_callback(String::new(), promise_id); }); } pub fn retrieve_latest_siwe_backup_data(promise_id: u32) { RUNTIME.spawn(async move { let result = download_latest_backup_id() .await .map_err(|err| err.to_string()); let result = match result { Ok(result) => result, Err(error) => { string_callback(error, promise_id, "".to_string()); return; } }; let LatestBackupIDResponse { backup_id, siwe_backup_msg, } = result; let siwe_backup_msg_string = match siwe_backup_msg { Some(siwe_backup_msg_value) => siwe_backup_msg_value, None => { string_callback( "Backup message unavailable".to_string(), promise_id, "".to_string(), ); return; } }; let siwe_backup_msg_obj: Message = match siwe_backup_msg_string.parse() { Ok(siwe_backup_msg_obj) => siwe_backup_msg_obj, Err(error) => { string_callback(error.to_string(), promise_id, "".to_string()); return; } }; let siwe_backup_msg_nonce = siwe_backup_msg_obj.nonce; let siwe_backup_msg_statement = match siwe_backup_msg_obj.statement { Some(statement) => statement, None => { string_callback( "Backup message invalid: missing statement".to_string(), promise_id, "".to_string(), ); return; } }; let siwe_backup_msg_issued_at = siwe_backup_msg_obj.issued_at.to_string(); let siwe_backup_data = SIWEBackupData { backup_id: backup_id, siwe_backup_msg: siwe_backup_msg_string, siwe_backup_msg_nonce: siwe_backup_msg_nonce, siwe_backup_msg_statement: siwe_backup_msg_statement, siwe_backup_msg_issued_at: siwe_backup_msg_issued_at, }; let serialize_result = serde_json::to_string(&siwe_backup_data); handle_string_result_as_callback(serialize_result, promise_id); }); } } pub async fn create_userkeys_compaction( backup_id: String, backup_secret: String, pickle_key: String, pickled_account: String, ) -> Result<(), Box> { let mut backup_key = compute_backup_key(backup_secret.as_bytes(), backup_id.as_bytes())?; let backup_data_key = secure_store_get(secure_store::SECURE_STORE_ENCRYPTION_KEY_ID)?; let backup_log_data_key = secure_store_get(secure_store::SECURE_STORE_BACKUP_LOGS_ENCRYPTION_KEY_ID)?; let user_keys = UserKeys { backup_data_key, backup_log_data_key, pickle_key, pickled_account, }; let encrypted_user_keys = user_keys.encrypt(&mut backup_key)?; let user_keys_file = get_backup_user_keys_file_path(&backup_id)?; tokio::fs::write(user_keys_file, encrypted_user_keys).await?; Ok(()) } pub async fn create_siwe_backup_msg_compaction( - backup_id: &String, + backup_id: &str, siwe_backup_msg: String, ) -> Result<(), Box> { - let siwe_backup_msg_file = get_siwe_backup_message_path(&backup_id)?; + let siwe_backup_msg_file = get_siwe_backup_message_path(backup_id)?; tokio::fs::write(siwe_backup_msg_file, siwe_backup_msg).await?; Ok(()) } async fn download_backup( backup_secret: String, backup_id: Option, ) -> Result> { let backup_id = match backup_id { Some(backup_id) => backup_id, None => { let latest_backup_id_response = download_latest_backup_id().await?; latest_backup_id_response.backup_id } }; let backup_keys = download_backup_keys(backup_id, backup_secret).await?; download_backup_data(backup_keys).await } async fn download_latest_backup_id( ) -> Result> { let backup_client = BackupClient::new(BACKUP_SOCKET_ADDR)?; let user_identity = get_user_identity_from_secure_store()?; let latest_backup_descriptor = BackupDescriptor::Latest { username: user_identity.user_id.clone(), }; let backup_id_response = backup_client .download_backup_data(&latest_backup_descriptor, RequestedData::BackupID) .await?; let LatestBackupIDResponse { backup_id, siwe_backup_msg, } = serde_json::from_slice(&backup_id_response)?; Ok(LatestBackupIDResponse { backup_id, siwe_backup_msg, }) } async fn download_backup_keys( backup_id: String, backup_secret: String, ) -> Result> { let backup_client = BackupClient::new(BACKUP_SOCKET_ADDR)?; let user_identity = get_user_identity_from_secure_store()?; let latest_backup_descriptor = BackupDescriptor::Latest { username: user_identity.user_id.clone(), }; let mut encrypted_user_keys = backup_client .download_backup_data(&latest_backup_descriptor, RequestedData::UserKeys) .await?; let mut backup_key = compute_backup_key_str(&backup_secret, &backup_id)?; let user_keys = UserKeys::from_encrypted(&mut encrypted_user_keys, &mut backup_key)?; Ok(BackupKeysResult { backup_id, backup_data_key: user_keys.backup_data_key, backup_log_data_key: user_keys.backup_log_data_key, }) } async fn download_backup_data( backup_keys: BackupKeysResult, ) -> Result> { let backup_client = BackupClient::new(BACKUP_SOCKET_ADDR)?; let user_identity = get_user_identity_from_secure_store()?; let BackupKeysResult { backup_id, backup_data_key, backup_log_data_key, } = backup_keys; let backup_data_descriptor = BackupDescriptor::BackupID { backup_id: backup_id.clone(), user_identity: user_identity.clone(), }; let encrypted_user_data = backup_client .download_backup_data(&backup_data_descriptor, RequestedData::UserData) .await?; let backup_restoration_path = PathBuf::from(get_backup_directory_path()?).join("restore_compaction"); tokio::fs::write(&backup_restoration_path, encrypted_user_data).await?; Ok(CompactionDownloadResult { backup_restoration_path, backup_id, backup_data_key, backup_log_data_key, }) } async fn download_and_apply_logs( backup_id: &str, backup_log_data_key: String, ) -> Result<(), Box> { let mut backup_log_data_key = backup_log_data_key.into_bytes(); let backup_client = BackupClient::new(BACKUP_SOCKET_ADDR)?; let user_identity = get_user_identity_from_secure_store()?; let stream = backup_client.download_logs(&user_identity, backup_id).await; let mut stream = Box::pin(stream); while let Some(mut log) = stream.try_next().await? { let data = decrypt( backup_log_data_key.as_mut_slice(), log.content.as_mut_slice(), )?; let (future_id, future) = future_manager::new_future::<()>().await; restore_from_backup_log(data, future_id); future.await?; } Ok(()) } fn get_user_identity_from_secure_store() -> Result { Ok(UserIdentity { user_id: secure_store_get(secure_store::USER_ID)?, access_token: secure_store_get(secure_store::COMM_SERVICES_ACCESS_TOKEN)?, device_id: secure_store_get(secure_store::DEVICE_ID)?, }) } #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct BackupKeysResult { #[serde(rename = "backupID")] backup_id: String, backup_data_key: String, backup_log_data_key: String, } #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct SIWEBackupData { #[serde(rename = "backupID")] backup_id: String, siwe_backup_msg: String, siwe_backup_msg_statement: String, siwe_backup_msg_nonce: String, siwe_backup_msg_issued_at: String, } struct CompactionDownloadResult { backup_id: String, backup_restoration_path: PathBuf, backup_data_key: String, backup_log_data_key: String, } #[derive(Debug, Serialize, Deserialize)] struct UserKeys { backup_data_key: String, backup_log_data_key: String, pickle_key: String, pickled_account: String, } impl UserKeys { fn encrypt(&self, backup_key: &mut [u8]) -> Result, Box> { let mut json = serde_json::to_vec(self)?; encrypt(backup_key, &mut json) } fn from_encrypted( data: &mut [u8], backup_key: &mut [u8], ) -> Result> { let decrypted = decrypt(backup_key, data)?; Ok(serde_json::from_slice(&decrypted)?) } } fn encrypt(key: &mut [u8], data: &mut [u8]) -> Result, Box> { let encrypted_len = data.len() + aes::IV_LENGTH + aes::TAG_LENGTH; let mut encrypted = vec![0; encrypted_len]; crate::ffi::encrypt(key, data, &mut encrypted)?; Ok(encrypted) } fn decrypt(key: &mut [u8], data: &mut [u8]) -> Result, Box> { let decrypted_len = data.len() - aes::IV_LENGTH - aes::TAG_LENGTH; let mut decrypted = vec![0; decrypted_len]; crate::ffi::decrypt(key, data, &mut decrypted)?; Ok(decrypted) } diff --git a/native/native_rust_library/src/identity/login.rs b/native/native_rust_library/src/identity/login.rs index 6f858c40e..9a2c795bd 100644 --- a/native/native_rust_library/src/identity/login.rs +++ b/native/native_rust_library/src/identity/login.rs @@ -1,279 +1,280 @@ use comm_opaque2::client::Login; use grpc_clients::identity::{ get_unauthenticated_client, protos::unauth::{ DeviceKeyUpload, ExistingDeviceLoginRequest, IdentityKeyInfo, OpaqueLoginFinishRequest, OpaqueLoginStartRequest, Prekey, SecondaryDeviceKeysUploadRequest, WalletAuthRequest, }, }; use tracing::instrument; use super::{IdentityAuthResult, LogInPasswordUserInfo, LogInWalletUserInfo}; use crate::utils::jsi_callbacks::handle_string_result_as_callback; use crate::{Error, CODE_VERSION, DEVICE_TYPE, IDENTITY_SOCKET_ADDR, RUNTIME}; +#[allow(clippy::too_many_arguments)] pub mod ffi { use crate::identity::{ DeviceKeys, LogInPasswordUserInfo, LogInWalletUserInfo, }; use super::*; #[instrument] pub fn log_in_password_user( username: String, password: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, promise_id: u32, ) { RUNTIME.spawn(async move { let password_user_info = LogInPasswordUserInfo { username, password, device_keys: DeviceKeys { key_payload, key_payload_signature, content_prekey, content_prekey_signature, notif_prekey, notif_prekey_signature, content_one_time_keys: Vec::new(), notif_one_time_keys: Vec::new(), }, }; let result = log_in_password_user_helper(password_user_info).await; handle_string_result_as_callback(result, promise_id); }); } #[instrument] pub fn log_in_wallet_user( siwe_message: String, siwe_signature: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, promise_id: u32, ) { RUNTIME.spawn(async move { let wallet_user_info = LogInWalletUserInfo { siwe_message, siwe_signature, device_keys: DeviceKeys { key_payload, key_payload_signature, content_prekey, content_prekey_signature, notif_prekey, notif_prekey_signature, content_one_time_keys: Vec::new(), notif_one_time_keys: Vec::new(), }, }; let result = log_in_wallet_user_helper(wallet_user_info).await; handle_string_result_as_callback(result, promise_id); }); } // QR code device log in pub fn upload_secondary_device_keys_and_log_in( user_id: String, nonce: String, nonce_signature: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, promise_id: u32, ) { RUNTIME.spawn(async move { let device_key_upload = DeviceKeyUpload { device_key_info: Some(IdentityKeyInfo { payload: key_payload, payload_signature: key_payload_signature, }), content_upload: Some(Prekey { prekey: content_prekey, prekey_signature: content_prekey_signature, }), notif_upload: Some(Prekey { prekey: notif_prekey, prekey_signature: notif_prekey_signature, }), one_time_content_prekeys: content_one_time_keys, one_time_notif_prekeys: notif_one_time_keys, device_type: DEVICE_TYPE.into(), }; let result = upload_secondary_device_keys_and_log_in_helper( user_id, nonce, nonce_signature, device_key_upload, ) .await; handle_string_result_as_callback(result, promise_id); }); } pub fn log_in_existing_device( user_id: String, device_id: String, nonce: String, nonce_signature: String, promise_id: u32, ) { RUNTIME.spawn(async move { let result = log_in_existing_device_helper( user_id, device_id, nonce, nonce_signature, ) .await; handle_string_result_as_callback(result, promise_id); }); } } async fn log_in_password_user_helper( password_user_info: LogInPasswordUserInfo, ) -> Result { let mut client_login = Login::new(); let opaque_login_request = client_login .start(&password_user_info.password) .map_err(crate::handle_error)?; let login_start_request = OpaqueLoginStartRequest { opaque_login_request, username: password_user_info.username, device_key_upload: Some(password_user_info.device_keys.into()), force: None, }; let mut identity_client = get_unauthenticated_client( IDENTITY_SOCKET_ADDR, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let response = identity_client .log_in_password_user_start(login_start_request) .await?; let login_start_response = response.into_inner(); let opaque_login_upload = client_login .finish(&login_start_response.opaque_login_response) .map_err(crate::handle_error)?; let login_finish_request = OpaqueLoginFinishRequest { session_id: login_start_response.session_id, opaque_login_upload, }; let login_finish_response = identity_client .log_in_password_user_finish(login_finish_request) .await? .into_inner(); let auth_result = IdentityAuthResult::from(login_finish_response); Ok(serde_json::to_string(&auth_result)?) } async fn log_in_wallet_user_helper( wallet_user_info: LogInWalletUserInfo, ) -> Result { let login_request = WalletAuthRequest { siwe_message: wallet_user_info.siwe_message, siwe_signature: wallet_user_info.siwe_signature, device_key_upload: Some(wallet_user_info.device_keys.into()), farcaster_id: None, initial_device_list: "".to_string(), }; let mut identity_client = get_unauthenticated_client( IDENTITY_SOCKET_ADDR, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let login_response = identity_client .log_in_wallet_user(login_request) .await? .into_inner(); let auth_result = IdentityAuthResult::from(login_response); Ok(serde_json::to_string(&auth_result)?) } async fn upload_secondary_device_keys_and_log_in_helper( user_id: String, nonce: String, nonce_signature: String, device_key_upload: DeviceKeyUpload, ) -> Result { let mut identity_client = get_unauthenticated_client( IDENTITY_SOCKET_ADDR, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let request = SecondaryDeviceKeysUploadRequest { user_id, nonce, nonce_signature, device_key_upload: Some(device_key_upload), }; let response = identity_client .upload_keys_for_registered_device_and_log_in(request) .await? .into_inner(); let auth_result = IdentityAuthResult::from(response); Ok(serde_json::to_string(&auth_result)?) } async fn log_in_existing_device_helper( user_id: String, device_id: String, nonce: String, nonce_signature: String, ) -> Result { let mut identity_client = get_unauthenticated_client( IDENTITY_SOCKET_ADDR, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let request = ExistingDeviceLoginRequest { user_id, device_id, nonce, nonce_signature, }; let response = identity_client .log_in_existing_device(request) .await? .into_inner(); let auth_result = IdentityAuthResult::from(response); Ok(serde_json::to_string(&auth_result)?) } diff --git a/native/native_rust_library/src/identity/registration.rs b/native/native_rust_library/src/identity/registration.rs index 26992c1e7..a878eeae7 100644 --- a/native/native_rust_library/src/identity/registration.rs +++ b/native/native_rust_library/src/identity/registration.rs @@ -1,336 +1,337 @@ use crate::{ utils::jsi_callbacks::handle_string_result_as_callback, Error, CODE_VERSION, DEVICE_TYPE, IDENTITY_SOCKET_ADDR, RUNTIME, }; use comm_opaque2::client::Registration; use grpc_clients::identity::{ get_unauthenticated_client, protos::unauth::{ RegistrationFinishRequest, RegistrationStartRequest, ReservedRegistrationStartRequest, ReservedWalletRegistrationRequest, WalletAuthRequest, }, }; use tracing::instrument; use super::{ farcaster::farcaster_id_string_to_option, IdentityAuthResult, RegisterPasswordUserInfo, RegisterReservedPasswordUserInfo, RegisterReservedWalletUserInfo, RegisterWalletUserInfo, }; +#[allow(clippy::too_many_arguments)] pub mod ffi { use crate::identity::{ DeviceKeys, RegisterPasswordUserInfo, RegisterReservedPasswordUserInfo, RegisterWalletUserInfo, }; use super::*; #[instrument] pub fn register_password_user( username: String, password: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, farcaster_id: String, initial_device_list: String, promise_id: u32, ) { RUNTIME.spawn(async move { let password_user_info = RegisterPasswordUserInfo { username, password, device_keys: DeviceKeys { key_payload, key_payload_signature, content_prekey, content_prekey_signature, notif_prekey, notif_prekey_signature, content_one_time_keys, notif_one_time_keys, }, farcaster_id: farcaster_id_string_to_option(&farcaster_id), initial_device_list, }; let result = register_password_user_helper(password_user_info).await; handle_string_result_as_callback(result, promise_id); }); } #[instrument] pub fn register_reserved_password_user( username: String, password: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, keyserver_message: String, keyserver_signature: String, initial_device_list: String, promise_id: u32, ) { RUNTIME.spawn(async move { let password_user_info = RegisterReservedPasswordUserInfo { username, password, device_keys: DeviceKeys { key_payload, key_payload_signature, content_prekey, content_prekey_signature, notif_prekey, notif_prekey_signature, content_one_time_keys, notif_one_time_keys, }, keyserver_message, keyserver_signature, initial_device_list, }; let result = register_reserved_password_user_helper(password_user_info).await; handle_string_result_as_callback(result, promise_id); }); } #[instrument] pub fn register_wallet_user( siwe_message: String, siwe_signature: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, farcaster_id: String, initial_device_list: String, promise_id: u32, ) { RUNTIME.spawn(async move { let wallet_user_info = RegisterWalletUserInfo { siwe_message, siwe_signature, device_keys: DeviceKeys { key_payload, key_payload_signature, content_prekey, content_prekey_signature, notif_prekey, notif_prekey_signature, content_one_time_keys, notif_one_time_keys, }, farcaster_id: farcaster_id_string_to_option(&farcaster_id), initial_device_list, }; let result = register_wallet_user_helper(wallet_user_info).await; handle_string_result_as_callback(result, promise_id); }); } #[instrument] pub fn register_reserved_wallet_user( siwe_message: String, siwe_signature: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, keyserver_message: String, keyserver_signature: String, initial_device_list: String, promise_id: u32, ) { RUNTIME.spawn(async move { let wallet_user_info = RegisterReservedWalletUserInfo { siwe_message, siwe_signature, device_keys: DeviceKeys { key_payload, key_payload_signature, content_prekey, content_prekey_signature, notif_prekey, notif_prekey_signature, content_one_time_keys, notif_one_time_keys, }, keyserver_message, keyserver_signature, initial_device_list, }; let result = register_reserved_wallet_user_helper(wallet_user_info).await; handle_string_result_as_callback(result, promise_id); }); } } async fn register_password_user_helper( password_user_info: RegisterPasswordUserInfo, ) -> Result { let mut client_registration = Registration::new(); let opaque_registration_request = client_registration .start(&password_user_info.password) .map_err(crate::handle_error)?; let registration_start_request = RegistrationStartRequest { opaque_registration_request, username: password_user_info.username, device_key_upload: Some(password_user_info.device_keys.into()), farcaster_id: password_user_info.farcaster_id, initial_device_list: password_user_info.initial_device_list, }; let mut identity_client = get_unauthenticated_client( IDENTITY_SOCKET_ADDR, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let response = identity_client .register_password_user_start(registration_start_request) .await?; let registration_start_response = response.into_inner(); let opaque_registration_upload = client_registration .finish( &password_user_info.password, ®istration_start_response.opaque_registration_response, ) .map_err(crate::handle_error)?; let registration_finish_request = RegistrationFinishRequest { session_id: registration_start_response.session_id, opaque_registration_upload, }; let registration_finish_response = identity_client .register_password_user_finish(registration_finish_request) .await? .into_inner(); let auth_result = IdentityAuthResult::from(registration_finish_response); Ok(serde_json::to_string(&auth_result)?) } async fn register_reserved_password_user_helper( password_user_info: RegisterReservedPasswordUserInfo, ) -> Result { let mut client_registration = Registration::new(); let opaque_registration_request = client_registration .start(&password_user_info.password) .map_err(crate::handle_error)?; let registration_start_request = ReservedRegistrationStartRequest { opaque_registration_request, username: password_user_info.username, device_key_upload: Some(password_user_info.device_keys.into()), keyserver_message: password_user_info.keyserver_message, keyserver_signature: password_user_info.keyserver_signature, initial_device_list: password_user_info.initial_device_list, }; let mut identity_client = get_unauthenticated_client( IDENTITY_SOCKET_ADDR, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let response = identity_client .register_reserved_password_user_start(registration_start_request) .await?; let registration_start_response = response.into_inner(); let opaque_registration_upload = client_registration .finish( &password_user_info.password, ®istration_start_response.opaque_registration_response, ) .map_err(crate::handle_error)?; let registration_finish_request = RegistrationFinishRequest { session_id: registration_start_response.session_id, opaque_registration_upload, }; let registration_finish_response = identity_client .register_password_user_finish(registration_finish_request) .await? .into_inner(); let auth_result = IdentityAuthResult::from(registration_finish_response); Ok(serde_json::to_string(&auth_result)?) } async fn register_wallet_user_helper( wallet_user_info: RegisterWalletUserInfo, ) -> Result { let registration_request = WalletAuthRequest { siwe_message: wallet_user_info.siwe_message, siwe_signature: wallet_user_info.siwe_signature, device_key_upload: Some(wallet_user_info.device_keys.into()), farcaster_id: wallet_user_info.farcaster_id, initial_device_list: wallet_user_info.initial_device_list, }; let mut identity_client = get_unauthenticated_client( IDENTITY_SOCKET_ADDR, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let registration_response = identity_client .register_wallet_user(registration_request) .await? .into_inner(); let auth_result = IdentityAuthResult::from(registration_response); Ok(serde_json::to_string(&auth_result)?) } async fn register_reserved_wallet_user_helper( wallet_user_info: RegisterReservedWalletUserInfo, ) -> Result { let registration_request = ReservedWalletRegistrationRequest { siwe_message: wallet_user_info.siwe_message, siwe_signature: wallet_user_info.siwe_signature, device_key_upload: Some(wallet_user_info.device_keys.into()), keyserver_message: wallet_user_info.keyserver_message, keyserver_signature: wallet_user_info.keyserver_signature, initial_device_list: wallet_user_info.initial_device_list, }; let mut identity_client = get_unauthenticated_client( IDENTITY_SOCKET_ADDR, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let registration_response = identity_client .register_reserved_wallet_user(registration_request) .await? .into_inner(); let auth_result = IdentityAuthResult::from(registration_response); Ok(serde_json::to_string(&auth_result)?) } diff --git a/native/native_rust_library/src/identity/x3dh.rs b/native/native_rust_library/src/identity/x3dh.rs index 261af7b92..52c3e850e 100644 --- a/native/native_rust_library/src/identity/x3dh.rs +++ b/native/native_rust_library/src/identity/x3dh.rs @@ -1,394 +1,395 @@ use grpc_clients::identity::get_auth_client; use grpc_clients::identity::protos::authenticated::{ InboundKeyInfo, InboundKeysForUserRequest, KeyserverKeysResponse, OutboundKeyInfo, OutboundKeysForUserRequest, RefreshUserPrekeysRequest, UploadOneTimeKeysRequest, }; use grpc_clients::identity::protos::unauth::{IdentityKeyInfo, Prekey}; use serde::Serialize; use tracing::instrument; use crate::identity::AuthInfo; use crate::utils::jsi_callbacks::{ handle_string_result_as_callback, handle_void_result_as_callback, }; use crate::{Error, CODE_VERSION, DEVICE_TYPE, IDENTITY_SOCKET_ADDR, RUNTIME}; pub mod ffi { use super::*; pub fn get_outbound_keys_for_user( auth_user_id: String, auth_device_id: String, auth_access_token: String, user_id: String, promise_id: u32, ) { RUNTIME.spawn(async move { let get_outbound_keys_request_info = GetOutboundKeysRequestInfo { user_id }; let auth_info = AuthInfo { access_token: auth_access_token, user_id: auth_user_id, device_id: auth_device_id, }; let result = get_outbound_keys_for_user_helper( get_outbound_keys_request_info, auth_info, ) .await; handle_string_result_as_callback(result, promise_id); }); } pub fn get_inbound_keys_for_user( auth_user_id: String, auth_device_id: String, auth_access_token: String, user_id: String, promise_id: u32, ) { RUNTIME.spawn(async move { let get_inbound_keys_request_info = GetInboundKeysRequestInfo { user_id }; let auth_info = AuthInfo { access_token: auth_access_token, user_id: auth_user_id, device_id: auth_device_id, }; let result = get_inbound_keys_for_user_helper( get_inbound_keys_request_info, auth_info, ) .await; handle_string_result_as_callback(result, promise_id); }); } pub fn get_keyserver_keys( user_id: String, device_id: String, access_token: String, keyserver_id: String, promise_id: u32, ) { RUNTIME.spawn(async move { let get_keyserver_keys_request = OutboundKeysForUserRequest { user_id: keyserver_id, }; let auth_info = AuthInfo { access_token, user_id, device_id, }; let result = get_keyserver_keys_helper(get_keyserver_keys_request, auth_info).await; handle_string_result_as_callback(result, promise_id); }); } + #[allow(clippy::too_many_arguments)] pub fn refresh_user_prekeys( auth_user_id: String, auth_device_id: String, auth_access_token: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, promise_id: u32, ) { RUNTIME.spawn(async move { let refresh_request = RefreshUserPrekeysRequest { new_content_prekeys: Some(Prekey { prekey: content_prekey, prekey_signature: content_prekey_signature, }), new_notif_prekeys: Some(Prekey { prekey: notif_prekey, prekey_signature: notif_prekey_signature, }), }; let auth_info = AuthInfo { access_token: auth_access_token, user_id: auth_user_id, device_id: auth_device_id, }; let result = refresh_user_prekeys_helper(refresh_request, auth_info).await; handle_void_result_as_callback(result, promise_id); }); } #[instrument] pub fn upload_one_time_keys( auth_user_id: String, auth_device_id: String, auth_access_token: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, promise_id: u32, ) { RUNTIME.spawn(async move { let upload_request = UploadOneTimeKeysRequest { content_one_time_prekeys: content_one_time_keys, notif_one_time_prekeys: notif_one_time_keys, }; let auth_info = AuthInfo { access_token: auth_access_token, user_id: auth_user_id, device_id: auth_device_id, }; let result = upload_one_time_keys_helper(auth_info, upload_request).await; handle_void_result_as_callback(result, promise_id); }); } } // helper structs struct GetOutboundKeysRequestInfo { user_id: String, } struct GetInboundKeysRequestInfo { user_id: String, } // This struct should not be altered without also updating // OutboundKeyInfoResponse in lib/types/identity-service-types.js #[derive(Serialize)] #[serde(rename_all = "camelCase")] struct OutboundKeyInfoResponse { pub payload: String, pub payload_signature: String, pub content_prekey: String, pub content_prekey_signature: String, pub notif_prekey: String, pub notif_prekey_signature: String, pub one_time_content_prekey: Option, pub one_time_notif_prekey: Option, } // This struct should not be altered without also updating // InboundKeyInfoResponse in lib/types/identity-service-types.js #[derive(Serialize)] #[serde(rename_all = "camelCase")] struct InboundKeyInfoResponse { pub payload: String, pub payload_signature: String, pub content_prekey: String, pub content_prekey_signature: String, pub notif_prekey: String, pub notif_prekey_signature: String, } impl TryFrom for OutboundKeyInfoResponse { type Error = Error; fn try_from(key_info: OutboundKeyInfo) -> Result { let identity_info = key_info.identity_info.ok_or(Error::MissingResponseData)?; let IdentityKeyInfo { payload, payload_signature, } = identity_info; let content_prekey = key_info.content_prekey.ok_or(Error::MissingResponseData)?; let Prekey { prekey: content_prekey_value, prekey_signature: content_prekey_signature, } = content_prekey; let notif_prekey = key_info.notif_prekey.ok_or(Error::MissingResponseData)?; let Prekey { prekey: notif_prekey_value, prekey_signature: notif_prekey_signature, } = notif_prekey; let one_time_content_prekey = key_info.one_time_content_prekey; let one_time_notif_prekey = key_info.one_time_notif_prekey; Ok(Self { payload, payload_signature, content_prekey: content_prekey_value, content_prekey_signature, notif_prekey: notif_prekey_value, notif_prekey_signature, one_time_content_prekey, one_time_notif_prekey, }) } } impl TryFrom for OutboundKeyInfoResponse { type Error = Error; fn try_from(response: KeyserverKeysResponse) -> Result { let key_info = response.keyserver_info.ok_or(Error::MissingResponseData)?; Self::try_from(key_info) } } impl TryFrom for InboundKeyInfoResponse { type Error = Error; fn try_from(key_info: InboundKeyInfo) -> Result { let identity_info = key_info.identity_info.ok_or(Error::MissingResponseData)?; let IdentityKeyInfo { payload, payload_signature, } = identity_info; let content_prekey = key_info.content_prekey.ok_or(Error::MissingResponseData)?; let Prekey { prekey: content_prekey_value, prekey_signature: content_prekey_signature, } = content_prekey; let notif_prekey = key_info.notif_prekey.ok_or(Error::MissingResponseData)?; let Prekey { prekey: notif_prekey_value, prekey_signature: notif_prekey_signature, } = notif_prekey; Ok(Self { payload, payload_signature, content_prekey: content_prekey_value, content_prekey_signature, notif_prekey: notif_prekey_value, notif_prekey_signature, }) } } async fn get_outbound_keys_for_user_helper( get_outbound_keys_request_info: GetOutboundKeysRequestInfo, auth_info: AuthInfo, ) -> Result { let mut identity_client = get_auth_client( IDENTITY_SOCKET_ADDR, auth_info.user_id, auth_info.device_id, auth_info.access_token, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let response = identity_client .get_outbound_keys_for_user(OutboundKeysForUserRequest { user_id: get_outbound_keys_request_info.user_id, }) .await? .into_inner(); let outbound_key_info: Vec = response .devices .into_values() .map(OutboundKeyInfoResponse::try_from) .collect::, _>>()?; Ok(serde_json::to_string(&outbound_key_info)?) } async fn get_inbound_keys_for_user_helper( get_inbound_keys_request_info: GetInboundKeysRequestInfo, auth_info: AuthInfo, ) -> Result { let mut identity_client = get_auth_client( IDENTITY_SOCKET_ADDR, auth_info.user_id, auth_info.device_id, auth_info.access_token, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let response = identity_client .get_inbound_keys_for_user(InboundKeysForUserRequest { user_id: get_inbound_keys_request_info.user_id, }) .await? .into_inner(); let inbound_key_info: Vec = response .devices .into_values() .map(InboundKeyInfoResponse::try_from) .collect::, _>>()?; Ok(serde_json::to_string(&inbound_key_info)?) } async fn get_keyserver_keys_helper( get_keyserver_keys_request: OutboundKeysForUserRequest, auth_info: AuthInfo, ) -> Result { let mut identity_client = get_auth_client( IDENTITY_SOCKET_ADDR, auth_info.user_id, auth_info.device_id, auth_info.access_token, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; let response = identity_client .get_keyserver_keys(get_keyserver_keys_request) .await? .into_inner(); let keyserver_keys = OutboundKeyInfoResponse::try_from(response)?; Ok(serde_json::to_string(&keyserver_keys)?) } async fn refresh_user_prekeys_helper( refresh_request: RefreshUserPrekeysRequest, auth_info: AuthInfo, ) -> Result<(), Error> { get_auth_client( IDENTITY_SOCKET_ADDR, auth_info.user_id, auth_info.device_id, auth_info.access_token, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await? .refresh_user_prekeys(refresh_request) .await?; Ok(()) } async fn upload_one_time_keys_helper( auth_info: AuthInfo, upload_request: UploadOneTimeKeysRequest, ) -> Result<(), Error> { let mut identity_client = get_auth_client( IDENTITY_SOCKET_ADDR, auth_info.user_id, auth_info.device_id, auth_info.access_token, CODE_VERSION, DEVICE_TYPE.as_str_name().to_lowercase(), ) .await?; identity_client.upload_one_time_keys(upload_request).await?; Ok(()) } diff --git a/native/native_rust_library/src/lib.rs b/native/native_rust_library/src/lib.rs index a4a072b0e..b00dde03b 100644 --- a/native/native_rust_library/src/lib.rs +++ b/native/native_rust_library/src/lib.rs @@ -1,532 +1,533 @@ use comm_opaque2::grpc::opaque_error_to_grpc_status as handle_error; use grpc_clients::identity::protos::unauth::DeviceType; use lazy_static::lazy_static; use std::sync::Arc; use tokio::runtime::{Builder, Runtime}; use tonic::Status; mod argon2_tools; mod backup; mod constants; mod identity; mod utils; use crate::argon2_tools::compute_backup_key_str; use crate::utils::jsi_callbacks::{ handle_string_result_as_callback, handle_void_result_as_callback, }; mod generated { // We get the CODE_VERSION from this generated file include!(concat!(env!("OUT_DIR"), "/version.rs")); // We get the IDENTITY_SOCKET_ADDR from this generated file include!(concat!(env!("OUT_DIR"), "/socket_config.rs")); } pub use generated::CODE_VERSION; pub use generated::{BACKUP_SOCKET_ADDR, IDENTITY_SOCKET_ADDR}; #[cfg(not(target_os = "android"))] pub const DEVICE_TYPE: DeviceType = DeviceType::Ios; #[cfg(target_os = "android")] pub const DEVICE_TYPE: DeviceType = DeviceType::Android; lazy_static! { static ref RUNTIME: Arc = Arc::new(Builder::new_multi_thread().enable_all().build().unwrap()); } // ffi uses use backup::ffi::*; use identity::ffi::*; use utils::future_manager::ffi::*; +#[allow(clippy::too_many_arguments)] #[cxx::bridge] mod ffi { // Identity Service APIs extern "Rust" { #[cxx_name = "identityRegisterPasswordUser"] fn register_password_user( username: String, password: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, farcaster_id: String, initial_device_list: String, promise_id: u32, ); #[cxx_name = "identityRegisterReservedPasswordUser"] fn register_reserved_password_user( username: String, password: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, keyserver_message: String, keyserver_signature: String, initial_device_list: String, promise_id: u32, ); #[cxx_name = "identityLogInPasswordUser"] fn log_in_password_user( username: String, password: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, promise_id: u32, ); #[cxx_name = "identityRegisterWalletUser"] fn register_wallet_user( siwe_message: String, siwe_signature: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, farcaster_id: String, initial_device_list: String, promise_id: u32, ); #[cxx_name = "identityRegisterReservedWalletUser"] fn register_reserved_wallet_user( siwe_message: String, siwe_signature: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, keyserver_message: String, keyserver_signature: String, initial_device_list: String, promise_id: u32, ); #[cxx_name = "identityLogInWalletUser"] fn log_in_wallet_user( siwe_message: String, siwe_signature: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, promise_id: u32, ); #[cxx_name = "identityUpdateUserPassword"] fn update_user_password( user_id: String, device_id: String, access_token: String, password: String, promise_id: u32, ); #[cxx_name = "identityDeleteWalletUser"] fn delete_wallet_user( user_id: String, device_id: String, access_token: String, promise_id: u32, ); #[cxx_name = "identityDeletePasswordUser"] fn delete_password_user( user_id: String, device_id: String, access_token: String, password: String, promise_id: u32, ); #[cxx_name = "identityLogOut"] fn log_out( user_id: String, device_id: String, access_token: String, promise_id: u32, ); #[cxx_name = "identityLogOutSecondaryDevice"] fn log_out_secondary_device( user_id: String, device_id: String, access_token: String, promise_id: u32, ); #[cxx_name = "identityGetOutboundKeysForUser"] fn get_outbound_keys_for_user( auth_user_id: String, auth_device_id: String, auth_access_token: String, user_id: String, promise_id: u32, ); #[cxx_name = "identityGetInboundKeysForUser"] fn get_inbound_keys_for_user( auth_user_id: String, auth_device_id: String, auth_access_token: String, user_id: String, promise_id: u32, ); #[cxx_name = "identityRefreshUserPrekeys"] fn refresh_user_prekeys( auth_user_id: String, auth_device_id: String, auth_access_token: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, promise_id: u32, ); #[cxx_name = "identityGenerateNonce"] fn generate_nonce(promise_id: u32); #[cxx_name = "identityVersionSupported"] fn version_supported(promise_id: u32); #[cxx_name = "identityUploadOneTimeKeys"] fn upload_one_time_keys( auth_user_id: String, auth_device_id: String, auth_access_token: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, promise_id: u32, ); #[cxx_name = "identityGetKeyserverKeys"] fn get_keyserver_keys( user_id: String, device_id: String, access_token: String, keyserver_id: String, promise_id: u32, ); #[cxx_name = "identityGetDeviceListForUser"] fn get_device_list_for_user( auth_user_id: String, auth_device_id: String, auth_access_token: String, user_id: String, since_timestamp: i64, promise_id: u32, ); #[cxx_name = "identityGetDeviceListsForUsers"] fn get_device_lists_for_users( auth_user_id: String, auth_device_id: String, auth_access_token: String, user_ids: Vec, promise_id: u32, ); #[cxx_name = "identityUpdateDeviceList"] fn update_device_list( auth_user_id: String, auth_device_id: String, auth_access_token: String, update_payload: String, promise_id: u32, ); #[cxx_name = "identityUploadSecondaryDeviceKeysAndLogIn"] fn upload_secondary_device_keys_and_log_in( user_id: String, nonce: String, nonce_signature: String, key_payload: String, key_payload_signature: String, content_prekey: String, content_prekey_signature: String, notif_prekey: String, notif_prekey_signature: String, content_one_time_keys: Vec, notif_one_time_keys: Vec, promise_id: u32, ); #[cxx_name = "identityLogInExistingDevice"] fn log_in_existing_device( user_id: String, device_id: String, nonce: String, nonce_signature: String, promise_id: u32, ); #[cxx_name = "identityFindUserIDForWalletAddress"] fn find_user_id_for_wallet_address(wallet_address: String, promise_id: u32); #[cxx_name = "identityFindUserIDForUsername"] fn find_user_id_for_username(username: String, promise_id: u32); // Farcaster #[cxx_name = "identityGetFarcasterUsers"] fn get_farcaster_users(farcaster_ids: Vec, promise_id: u32); #[cxx_name = "identityLinkFarcasterAccount"] fn link_farcaster_account( user_id: String, device_id: String, access_token: String, farcaster_id: String, promise_id: u32, ); #[cxx_name = "identityUnlinkFarcasterAccount"] fn unlink_farcaster_account( user_id: String, device_id: String, access_token: String, promise_id: u32, ); #[cxx_name = "identityFindUserIdentities"] fn find_user_identities( user_id: String, device_id: String, access_token: String, user_ids: Vec, promise_id: u32, ); // Argon2 #[cxx_name = "compute_backup_key"] fn compute_backup_key_str( password: &str, backup_id: &str, ) -> Result<[u8; 32]>; } unsafe extern "C++" { include!("RustCallback.h"); #[namespace = "comm"] #[cxx_name = "stringCallback"] fn string_callback(error: String, promise_id: u32, ret: String); #[namespace = "comm"] #[cxx_name = "voidCallback"] fn void_callback(error: String, promise_id: u32); #[namespace = "comm"] #[cxx_name = "boolCallback"] fn bool_callback(error: String, promise_id: u32, ret: bool); } // AES cryptography #[namespace = "comm"] unsafe extern "C++" { include!("RustAESCrypto.h"); #[allow(unused)] #[cxx_name = "aesGenerateKey"] fn generate_key(buffer: &mut [u8]) -> Result<()>; /// The first two argument aren't mutated but creation of Java ByteBuffer /// requires the underlying bytes to be mutable. #[allow(unused)] #[cxx_name = "aesEncrypt"] fn encrypt( key: &mut [u8], plaintext: &mut [u8], sealed_data: &mut [u8], ) -> Result<()>; /// The first two argument aren't mutated but creation of Java ByteBuffer /// requires the underlying bytes to be mutable. #[allow(unused)] #[cxx_name = "aesDecrypt"] fn decrypt( key: &mut [u8], sealed_data: &mut [u8], plaintext: &mut [u8], ) -> Result<()>; } // Comm Services Auth Metadata Emission #[namespace = "comm"] unsafe extern "C++" { include!("RustCSAMetadataEmitter.h"); #[allow(unused)] #[cxx_name = "sendAuthMetadataToJS"] fn send_auth_metadata_to_js( access_token: String, user_id: String, ) -> Result<()>; } // Backup extern "Rust" { #[cxx_name = "startBackupHandler"] fn start_backup_handler() -> Result<()>; #[cxx_name = "stopBackupHandler"] fn stop_backup_handler() -> Result<()>; #[cxx_name = "triggerBackupFileUpload"] fn trigger_backup_file_upload(); #[cxx_name = "createBackup"] fn create_backup( backup_id: String, backup_secret: String, pickle_key: String, pickled_account: String, siwe_backup_msg: String, promise_id: u32, ); #[cxx_name = "restoreBackup"] fn restore_backup( backup_secret: String, backup_id: String, promise_id: u32, ); #[cxx_name = "restoreBackupData"] fn restore_backup_data( backup_id: String, backup_data_key: String, backup_log_data_key: String, promise_id: u32, ); #[cxx_name = "retrieveBackupKeys"] fn retrieve_backup_keys(backup_secret: String, promise_id: u32); #[cxx_name = "retrieveLatestSIWEBackupData"] fn retrieve_latest_siwe_backup_data(promise_id: u32); } // Secure store #[namespace = "comm"] unsafe extern "C++" { include!("RustSecureStore.h"); #[allow(unused)] #[cxx_name = "secureStoreSet"] fn secure_store_set(key: &str, value: String) -> Result<()>; #[cxx_name = "secureStoreGet"] fn secure_store_get(key: &str) -> Result; } // C++ Backup creation #[namespace = "comm"] unsafe extern "C++" { include!("RustBackupExecutor.h"); #[cxx_name = "getBackupDirectoryPath"] fn get_backup_directory_path() -> Result; #[cxx_name = "getBackupFilePath"] fn get_backup_file_path( backup_id: &str, is_attachments: bool, ) -> Result; #[cxx_name = "getBackupLogFilePath"] fn get_backup_log_file_path( backup_id: &str, log_id: &str, is_attachments: bool, ) -> Result; #[cxx_name = "getBackupUserKeysFilePath"] fn get_backup_user_keys_file_path(backup_id: &str) -> Result; #[cxx_name = "getSIWEBackupMessagePath"] fn get_siwe_backup_message_path(backup_id: &str) -> Result; #[cxx_name = "createMainCompaction"] fn create_main_compaction(backup_id: &str, future_id: usize); #[cxx_name = "restoreFromMainCompaction"] fn restore_from_main_compaction( main_compaction_path: &str, main_compaction_encryption_key: &str, future_id: usize, ); #[cxx_name = "restoreFromBackupLog"] fn restore_from_backup_log(backup_log: Vec, future_id: usize); } // Future handling from C++ extern "Rust" { #[cxx_name = "resolveUnitFuture"] fn resolve_unit_future(future_id: usize); #[cxx_name = "rejectFuture"] fn reject_future(future_id: usize, error: String); } } #[derive( Debug, derive_more::Display, derive_more::From, derive_more::Error, )] pub enum Error { #[display(fmt = "{}", "_0.message()")] TonicGRPC(Status), #[display(fmt = "{}", "_0")] SerdeJson(serde_json::Error), #[display(fmt = "Missing response data")] MissingResponseData, #[display(fmt = "{}", "_0")] GRPClient(grpc_clients::error::Error), } #[cfg(test)] #[allow(clippy::assertions_on_constants)] mod tests { use super::{BACKUP_SOCKET_ADDR, CODE_VERSION, IDENTITY_SOCKET_ADDR}; #[test] fn test_code_version_exists() { assert!(CODE_VERSION > 0); } #[test] fn test_identity_socket_addr_exists() { assert!(!IDENTITY_SOCKET_ADDR.is_empty()); assert!(!BACKUP_SOCKET_ADDR.is_empty()); } } diff --git a/services/blob/src/service.rs b/services/blob/src/service.rs index 0e3a42743..8d68195aa 100644 --- a/services/blob/src/service.rs +++ b/services/blob/src/service.rs @@ -1,629 +1,629 @@ #![allow(unused)] use regex::RegexSet; use std::collections::{BTreeMap, HashSet}; use std::ops::{Bound, Range, RangeBounds, RangeInclusive}; use std::sync::Arc; use async_stream::try_stream; use chrono::Duration; use comm_lib::http::ByteStream; use comm_lib::shared::reserved_users::RESERVED_USERNAME_SET; use comm_lib::tools::BoxedError; use once_cell::sync::Lazy; use tokio_stream::StreamExt; use tonic::codegen::futures_core::Stream; use tracing::{debug, error, info, trace, warn}; use crate::config::{CONFIG, OFFENSIVE_INVITE_LINKS}; use crate::constants::{ INVITE_LINK_BLOB_HASH_PREFIX, S3_MULTIPART_UPLOAD_MINIMUM_CHUNK_SIZE, }; use crate::database::types::{ BlobItemInput, BlobItemRow, PrimaryKey, UncheckedKind, }; use crate::database::DBError; use crate::s3::{Error as S3Error, S3Client, S3Path}; use crate::tools::MemOps; use crate::{constants::BLOB_DOWNLOAD_CHUNK_SIZE, database::DatabaseClient}; #[derive( Debug, derive_more::Display, derive_more::From, derive_more::Error, )] pub enum InviteLinkError { Reserved, Offensive, } #[derive( Debug, derive_more::Display, derive_more::From, derive_more::Error, )] pub enum BlobServiceError { BlobNotFound, BlobAlreadyExists, InvalidState, DB(DBError), S3(S3Error), InputError(#[error(ignore)] BoxedError), InviteLinkError(InviteLinkError), } type BlobServiceResult = Result; #[derive(Clone, Debug)] pub struct BlobServiceConfig { /// Blob data is streamed from S3 in chunks of this size. pub download_chunk_size: usize, /// If enabled, orphaned blobs will be deleted immediately after /// last holder is removed. This option should be enabled /// if maintenance garbage collection tasks are not run. pub instant_delete_orphaned_blobs: bool, /// Minimum age that a orphan must stay unmodified /// before it can be deleted by a garbage collection task /// This option is ignored if `instant_delete_orphaned_blobs` is `true` pub orphan_protection_period: chrono::Duration, } static OFFENSIVE_INVITE_LINKS_REGEX_SET: Lazy = Lazy::new(|| { RegexSet::new(OFFENSIVE_INVITE_LINKS.iter().collect::>()).unwrap() }); impl Default for BlobServiceConfig { fn default() -> Self { BlobServiceConfig { download_chunk_size: BLOB_DOWNLOAD_CHUNK_SIZE as usize, instant_delete_orphaned_blobs: false, orphan_protection_period: Duration::hours(1), } } } #[derive(Clone)] pub struct BlobService { db: Arc, s3: S3Client, config: BlobServiceConfig, } impl BlobService { pub fn new( db: DatabaseClient, s3: S3Client, config: BlobServiceConfig, ) -> Self { Self { db: Arc::new(db), s3, config, } } /// Retrieves blob object metadata and returns a download object /// that can be used to download the blob data. pub async fn create_download( &self, blob_hash: impl Into, ) -> BlobServiceResult { // 1. Get S3 path let s3_path = match self.db.get_blob_item(blob_hash.into()).await { Ok(Some(BlobItemRow { s3_path, .. })) => Ok(s3_path), Ok(None) => { debug!("Blob not found"); Err(BlobServiceError::BlobNotFound) } Err(err) => Err(BlobServiceError::DB(err)), }?; debug!("S3 path: {:?}", s3_path); // 2. Get S3 Object metadata trace!("Getting S3 object metadata..."); let object_metadata = self.s3.get_object_metadata(&s3_path).await?; let blob_size: u64 = object_metadata.content_length().try_into().map_err(|err| { error!("Failed to parse S3 object content length: {:?}", err); BlobServiceError::InvalidState })?; debug!("S3 object size: {} bytes", blob_size); // 3. Create download session let session = BlobDownloadObject { s3_path, blob_size, byte_range: 0..blob_size, chunk_size: self.config.download_chunk_size as u64, s3_client: self.s3.clone(), }; Ok(session) } fn validate_invite_link_blob_hash( invite_secret: &str, ) -> Result<(), BlobServiceError> { let lowercase_secret = invite_secret.to_lowercase(); if (RESERVED_USERNAME_SET.contains(&lowercase_secret)) { debug!("Reserved invite link"); return Err(BlobServiceError::InviteLinkError(InviteLinkError::Reserved)); } if (OFFENSIVE_INVITE_LINKS_REGEX_SET.is_match(&lowercase_secret)) { debug!("Offensive invite link"); return Err(BlobServiceError::InviteLinkError( InviteLinkError::Offensive, )); } Ok(()) } pub async fn put_blob( &self, blob_hash: impl Into, mut blob_data_stream: impl ByteStream, ) -> Result<(), BlobServiceError> { let blob_hash: String = blob_hash.into(); let blob_item = BlobItemInput::new(&blob_hash); if self.db.get_blob_item(&blob_hash).await?.is_some() { debug!("Blob already exists"); return Err(BlobServiceError::BlobAlreadyExists); } if let Some(invite_secret) = blob_hash.strip_prefix(INVITE_LINK_BLOB_HASH_PREFIX) { - Self::validate_invite_link_blob_hash(&invite_secret)?; + Self::validate_invite_link_blob_hash(invite_secret)?; } let mut upload_session = self.s3.start_upload_session(&blob_item.s3_path).await?; trace!(?blob_item, "Started S3 upload session"); tokio::pin!(blob_data_stream); let mut s3_chunk: Vec = Vec::new(); while let Some(chunk) = blob_data_stream.try_next().await.map_err(|err| { warn!("Failed to get data chunk: {:?}", err); BlobServiceError::InputError(err) })? { s3_chunk.extend_from_slice(&chunk); // New parts should be added to AWS only if they exceed minimum part size, // Otherwise AWS returns error if s3_chunk.len() as u64 > S3_MULTIPART_UPLOAD_MINIMUM_CHUNK_SIZE { trace!( chunk_size = s3_chunk.len(), "Chunk size exceeded, adding new S3 part" ); upload_session .add_part(s3_chunk.take_out()) .await .map_err(BlobServiceError::from)?; } } trace!("Upload stream drained"); // add the remaining data as the last S3 part if !s3_chunk.is_empty() { trace!("Uploading remaining {} bytes", s3_chunk.len()); upload_session.add_part(s3_chunk).await?; } // Complete the upload session upload_session.finish_upload().await?; trace!("S3 upload complete, putting item to db"); self.db.put_blob_item(blob_item).await?; Ok(()) } pub async fn assign_holder( &self, blob_hash: impl Into, holder: impl Into, ) -> BlobServiceResult { let blob_hash: String = blob_hash.into(); trace!(blob_hash, "Attempting to assign holder"); self .db .put_holder_assignment(blob_hash.clone(), holder.into()) .await?; trace!("Holder assigned. Checking if data exists"); let data_exists = self.db.get_blob_item(blob_hash).await?.is_some(); Ok(data_exists) } pub async fn revoke_holder( &self, blob_hash: impl Into, holder: impl Into, instant_delete: bool, ) -> BlobServiceResult<()> { let blob_hash: String = blob_hash.into(); let holder: String = holder.into(); trace!(blob_hash, holder, "Attempting to revoke holder"); self.db.delete_holder_assignment(&blob_hash, holder).await?; if instant_delete || self.config.instant_delete_orphaned_blobs { trace!("Instant orphan deletion enabled. Looking for holders"); let is_orphan = self .db .list_blob_holders(&blob_hash, Some(1)) .await? .is_empty(); if !is_orphan { trace!("Found holders, nothing to do"); return Ok(()); } debug!("No holders left, deleting blob if exists"); trace!("Getting blob item"); let Some(blob_item) = self.db.get_blob_item(&blob_hash).await? else { trace!("Blob item not found, nothing to do"); return Ok(()); }; trace!("Deleting S3 object"); self.s3.delete_object(&blob_item.s3_path).await?; trace!("Deleting blob item entry from DB"); self.db.delete_blob_item(blob_hash).await?; } Ok(()) } pub async fn perform_cleanup(&self) -> anyhow::Result<()> { info!("Starting cleanup..."); // 1. Fetch blobs and holders marked as "unchecked" debug!("Querying for unchecked blobs and holders..."); let protection_periond = self.config.orphan_protection_period; let (unchecked_blobs, unchecked_holders) = tokio::try_join!( self .db .find_unchecked_items(UncheckedKind::Blob, protection_periond), self .db .find_unchecked_items(UncheckedKind::Holder, protection_periond) )?; debug!( "Found {} unchecked blobs and {} unchecked holders", unchecked_blobs.len(), unchecked_holders.len() ); let mut unchecked_items = UncheckedCollection::new(); // 2. construct structures of possibly orphaned blobs debug!("Creating structures of possibly orphaned items..."); for PrimaryKey { blob_hash, .. } in unchecked_blobs { trace!("Creating unchecked item for blob hash '{}'", &blob_hash); unchecked_items.insert( blob_hash.clone(), UncheckedItem { blob_hash: Some(blob_hash), holders: Vec::new(), }, ); } // 3. iterate over possibly orphaned holders and fill the structs for PrimaryKey { blob_hash, holder } in unchecked_holders { if let Some(item) = unchecked_items.get_mut(&blob_hash) { trace!( "Inserting holder '{}' for blob hash '{}'", &holder, &blob_hash ); item.holders.push(holder); } else { trace!( "Creating empty item for holder '{}' (blob hash: '{}')", &holder, &blob_hash ); unchecked_items.insert( blob_hash.clone(), UncheckedItem { blob_hash: None, holders: vec![holder], }, ); } } let mut orphans = HashSet::new(); let mut checked = HashSet::new(); // 4. Filter out items that are for sure not orphaned let checked_items = unchecked_items.filter_out_checked(); debug!("Filtered out {} checked items", checked_items.len()); checked.extend(checked_items); // 5. Query DDB for additional blobs and holders to check if they exist let mut fetch_results = Vec::new(); // 5a. Query holders - Find if possibly orphan blobs have at least one holder debug!("Querying holders for possibly orphaned blobs..."); for blob_hash in unchecked_items.blobs_to_find_holders() { let holders = self .db .list_blob_holders(blob_hash, Some(1)) .await? .into_iter() .map(|holder| PrimaryKey::new(blob_hash.to_string(), holder)); let len_before = fetch_results.len(); fetch_results.extend(holders); trace!( "Found {} holders for blob hash '{}'", fetch_results.len() - len_before, blob_hash ); } // 5b. Query blobs - Find if possibly orphaned holders have blobs debug!("Querying blobs for possibly orphaned holders..."); let blobs_to_get = unchecked_items.blobs_to_check_existence(); let queried_blobs_len = blobs_to_get.len(); let existing_blobs = self.db.list_existing_keys(blobs_to_get).await?; debug!( "Found {} existing blobs out of {} queried", existing_blobs.len(), queried_blobs_len ); fetch_results.extend(existing_blobs); // 6. Update the struct with query results // Then do 2nd pass of filtering out checked items (repeat step 4) debug!("Feeding data structure with query results and filtering again..."); unchecked_items.feed_with_query_results(fetch_results); let checked_items = unchecked_items.filter_out_checked(); debug!("Filtered out {} checked items", checked_items.len()); checked.extend(checked_items); // 7. Perform actual cleanup orphans.extend(unchecked_items.into_primary_keys()); let s3_paths: Vec = orphans .iter() .filter(|pk| pk.is_blob_item()) .map(|PrimaryKey { blob_hash, .. }| S3Path { bucket_name: CONFIG.s3_bucket_name.clone(), object_name: blob_hash.to_string(), }) .collect(); let num_orphans = orphans.len(); let num_checked = checked.len(); let num_s3_blobs = s3_paths.len(); // 7a. Make changes to database debug!("Cleaning up database... Marking {} items as checked and deleting {} orphans", num_checked, num_orphans); tokio::try_join!( self.db.batch_delete_rows(orphans), self.db.batch_mark_checked(checked) )?; // 7b. Delete orphaned blobs from S3 debug!("Cleaning up S3... Deleting {} blobs", num_s3_blobs); self.s3.batch_delete_objects(s3_paths).await?; info!( "Cleanup complete. Deleted orphaned {} DB items and marked {} items as checked. {} blobs were deleted from S3", num_orphans, num_checked, num_s3_blobs ); Ok(()) } } // A B-tree map performs well for both random and sequential access. type BlobHash = String; type UncheckedCollection = BTreeMap; /// Represents an "unchecked" blob entity. It might miss either /// blob hash or holders. #[derive(Debug)] struct UncheckedItem { blob_hash: Option, holders: Vec, } impl UncheckedItem { fn has_blob_hash(&self) -> bool { self.blob_hash.is_some() } fn has_holders(&self) -> bool { !self.holders.is_empty() } /// Returns primary keys for this item. It contains primary heys for holders /// and for blob item (if it has hash). /// A fallback hash is required for holders if item's blob hash is None. fn as_primary_keys(&self, fallback_blob_hash: &str) -> Vec { if !self.has_holders() && !self.has_blob_hash() { warn!( fallback_blob_hash, "Item has no hash and no holders, this should never happen!" ); return Vec::new(); } let hash_for_holders = self.blob_hash.as_deref().unwrap_or(fallback_blob_hash); let mut keys = self .holders .iter() .map(|holder| PrimaryKey { blob_hash: hash_for_holders.to_string(), holder: holder.to_string(), }) .collect::>(); if let Some(blob_hash) = &self.blob_hash { keys.push(PrimaryKey::for_blob_item(blob_hash.to_string())); } keys } } trait CleanupOperations { /// Retains only items that should remain unchecked /// (missing blob hash or holders). /// /// Returns removed items - these items are checked /// (contain both blob hash and at least one holder). fn filter_out_checked(&mut self) -> Vec; /// Returns list of blob hashes for which we need to query if they contain /// at least one holder fn blobs_to_find_holders(&self) -> Vec<&BlobHash>; /// Returns primary keys for blob items that need to be checked if they exist /// /// Technically, this returns all items that have holders but no hash. fn blobs_to_check_existence(&self) -> Vec; /// Updates the structure after fetching additional data from database. fn feed_with_query_results( &mut self, fetched_items: impl IntoIterator, ); /// Turns this collection into a list of DB primary keys fn into_primary_keys(self) -> Vec; } impl CleanupOperations for UncheckedCollection { /// Retains only items that should remain unchecked /// (missing blob hash or holders). /// /// Returns removed items - these items are checked /// (contain both blob hash and at least one holder). fn filter_out_checked(&mut self) -> Vec { let mut checked = Vec::new(); self.retain(|blob_hash, item| { if !item.has_blob_hash() || !item.has_holders() { // blob hash or holder missing, leave unchecked return true; } checked.extend(item.as_primary_keys(blob_hash)); false }); checked } /// Returns list of blob hashes for which we need to query if they contain /// at least one holder fn blobs_to_find_holders(&self) -> Vec<&BlobHash> { self .iter() .filter_map(|(blob_hash, item)| { if item.has_blob_hash() && !item.has_holders() { Some(blob_hash) } else { None } }) .collect() } /// Returns primary keys for blob items that need to be checked if they exist /// /// Technically, this returns all blob items that have holders but no hash. fn blobs_to_check_existence(&self) -> Vec { self .iter() .filter_map(|(blob_hash, item)| { if item.has_holders() && !item.has_blob_hash() { Some(PrimaryKey::for_blob_item(blob_hash)) } else { None } }) .collect() } /// Updates the structure after fetching additional data from database. fn feed_with_query_results( &mut self, fetched_items: impl IntoIterator, ) { for pk in fetched_items.into_iter() { let Some(item) = self.get_mut(&pk.blob_hash) else { warn!("Got fetched item that was not requested: {:?}", pk); continue; }; if pk.is_blob_item() { item.blob_hash = Some(pk.blob_hash) } else { item.holders.push(pk.holder); } } } fn into_primary_keys(self) -> Vec { self .into_iter() .flat_map(|(blob_hash, item)| item.as_primary_keys(&blob_hash)) .collect() } } pub struct BlobDownloadObject { /// Size of the whole blob object in bytes. pub blob_size: u64, /// Range of bytes to be downloaded (exclusive end). byte_range: Range, chunk_size: u64, s3_client: S3Client, s3_path: S3Path, } impl BlobDownloadObject { pub fn set_byte_range(&mut self, range: impl RangeBounds) { let range_start = match range.start_bound() { Bound::Included(&start) => start, Bound::Excluded(&start) => start + 1, Bound::Unbounded => 0, }; let range_end = match range.end_bound() { Bound::Included(&end) => end + 1, Bound::Excluded(&end) => end, Bound::Unbounded => self.blob_size, }; // Clamp range to blob size let start = std::cmp::max(range_start, 0); let end_exclusive = std::cmp::min(range_end, self.blob_size); self.byte_range = start..end_exclusive; debug!("Requested byte range: {}..{}", start, end_exclusive); } /// Size of the data to be downloaded in bytes. pub fn download_size(&self) -> u64 { self.byte_range.end - self.byte_range.start } pub fn into_stream(self) -> impl Stream>> { let BlobDownloadObject { byte_range, chunk_size, s3_path, s3_client, .. } = self; try_stream! { trace!("Starting download stream"); let mut offset: u64 = byte_range.start; while offset < byte_range.end { let next_size = std::cmp::min(chunk_size, byte_range.end - offset); let range = offset..(offset + next_size); trace!(?range, "Getting {} bytes of data", next_size); yield s3_client.get_object_bytes(&s3_path, range).await?; offset += next_size; } } } } diff --git a/services/commtest/src/identity/mod.rs b/services/commtest/src/identity/mod.rs index 241171e63..fdc6413e0 100644 --- a/services/commtest/src/identity/mod.rs +++ b/services/commtest/src/identity/mod.rs @@ -1,51 +1,51 @@ use base64::Engine; use ed25519_dalek::{ed25519::signature::Signer, Keypair, Signature}; use rand::rngs::OsRng; use self::olm_account_infos::{ ClientPublicKeys, IdentityPublicKeys, DEFAULT_CLIENT_KEYS, }; pub mod device; pub mod olm_account_infos; pub struct SigningCapableAccount { signing_key: Keypair, } impl SigningCapableAccount { pub fn new() -> Self { let mut rng = OsRng {}; let signing_key = Keypair::generate(&mut rng); Self { signing_key } } /// returns device public keys, required for device key upload pub fn public_keys(&self) -> ClientPublicKeys { let default = DEFAULT_CLIENT_KEYS.clone(); let signing_public_key = self.signing_key.public.to_bytes(); let ed25519 = base64::engine::general_purpose::STANDARD_NO_PAD .encode(signing_public_key); ClientPublicKeys { primary_identity_public_keys: IdentityPublicKeys { ed25519, ..default.primary_identity_public_keys }, ..default } } /// signs message, returns signature - pub fn sign_message(&mut self, message: &str) -> String { + pub fn sign_message(&self, message: &str) -> String { let signature: Signature = self.signing_key.sign(message.as_bytes()); base64::engine::general_purpose::STANDARD_NO_PAD .encode(signature.to_bytes()) } } impl Default for SigningCapableAccount { fn default() -> Self { Self::new() } } diff --git a/services/commtest/tests/identity_access_tokens_tests.rs b/services/commtest/tests/identity_access_tokens_tests.rs index 45f155515..dd186924b 100644 --- a/services/commtest/tests/identity_access_tokens_tests.rs +++ b/services/commtest/tests/identity_access_tokens_tests.rs @@ -1,89 +1,89 @@ use commtest::identity::device::{ register_user_device, DEVICE_TYPE, PLACEHOLDER_CODE_VERSION, }; use commtest::identity::SigningCapableAccount; use commtest::service_addr; use grpc_clients::identity::protos::unauth::{ Empty, ExistingDeviceLoginRequest, }; use grpc_clients::identity::{ get_unauthenticated_client, protos::unauth::VerifyUserAccessTokenRequest, }; #[tokio::test] async fn verify_access_token() { let identity_grpc_endpoint = service_addr::IDENTITY_GRPC.to_string(); let device_info = register_user_device(None, None).await; let mut identity_client = get_unauthenticated_client( &identity_grpc_endpoint, PLACEHOLDER_CODE_VERSION, DEVICE_TYPE.to_string(), ) .await .expect("Couldn't connect to identity service"); let verify_request = VerifyUserAccessTokenRequest::from(&device_info); let response = identity_client .verify_user_access_token(verify_request) .await .unwrap(); assert!(response.into_inner().token_valid); } #[tokio::test] async fn refresh_token_test() { let identity_grpc_endpoint = service_addr::IDENTITY_GRPC.to_string(); let mut client = get_unauthenticated_client( &identity_grpc_endpoint, PLACEHOLDER_CODE_VERSION, DEVICE_TYPE.to_string(), ) .await .expect("Couldn't connect to identity service"); - let mut account = SigningCapableAccount::new(); + let account = SigningCapableAccount::new(); let client_keys = account.public_keys(); let user = register_user_device(Some(&client_keys), None).await; // refresh session let nonce = client .generate_nonce(Empty {}) .await .expect("failed to generate nonce") .into_inner() .nonce; let nonce_signature = account.sign_message(&nonce); let new_credentials = client .log_in_existing_device(ExistingDeviceLoginRequest { user_id: user.user_id.clone(), device_id: user.device_id.clone(), nonce, nonce_signature, }) .await .expect("LogInExistingDevice call failed") .into_inner(); // old token should now be invalid let old_token_result = client .verify_user_access_token(VerifyUserAccessTokenRequest::from(&user)) .await .expect("failed to verify token") .into_inner(); assert!(!old_token_result.token_valid); // new token should be valid let new_token_result = client .verify_user_access_token(VerifyUserAccessTokenRequest { user_id: new_credentials.user_id, access_token: new_credentials.access_token, device_id: user.device_id, }) .await .expect("failed to verify token") .into_inner(); assert!(new_token_result.token_valid); } diff --git a/services/commtest/tests/identity_one_time_key_tests.rs b/services/commtest/tests/identity_one_time_key_tests.rs index d7b88293f..9857f82b4 100644 --- a/services/commtest/tests/identity_one_time_key_tests.rs +++ b/services/commtest/tests/identity_one_time_key_tests.rs @@ -1,189 +1,189 @@ use commtest::identity::device::{ register_user_device, DEVICE_TYPE, PLACEHOLDER_CODE_VERSION, }; use commtest::identity::olm_account_infos::generate_random_olm_key; use commtest::service_addr; use grpc_clients::identity::{ get_auth_client, protos::authenticated::OutboundKeysForUserRequest, protos::authenticated::UploadOneTimeKeysRequest, }; #[tokio::test] async fn upload_one_time_keys() { let device_info = register_user_device(None, None).await; let mut identity_client = get_auth_client( &service_addr::IDENTITY_GRPC.to_string(), device_info.user_id, device_info.device_id, device_info.access_token, PLACEHOLDER_CODE_VERSION, DEVICE_TYPE.to_string(), ) .await .expect("Couldn't connect to identity service"); let upload_request = UploadOneTimeKeysRequest { content_one_time_prekeys: vec![ generate_random_olm_key(), generate_random_olm_key(), ], notif_one_time_prekeys: vec![ generate_random_olm_key(), generate_random_olm_key(), ], }; identity_client .upload_one_time_keys(upload_request) .await .unwrap(); } #[tokio::test] async fn max_hundred_keys_in_ddb() { let device_info = register_user_device(None, None).await; let mut identity_client = get_auth_client( &service_addr::IDENTITY_GRPC.to_string(), device_info.user_id.clone(), device_info.device_id, device_info.access_token, PLACEHOLDER_CODE_VERSION, DEVICE_TYPE.to_string(), ) .await .expect("Couldn't connect to identity service"); // We expect these keys to be removed by the identity service before we // retrieve any OTKs let first_upload_request = UploadOneTimeKeysRequest { content_one_time_prekeys: vec![generate_random_olm_key()], notif_one_time_prekeys: vec![generate_random_olm_key()], }; identity_client .upload_one_time_keys(first_upload_request) .await .unwrap(); let mut expected_first_retrieved_content_key = None; let mut expected_first_retrieved_notif_key = None; let mut expected_second_retrieved_content_key = None; let mut expected_second_retrieved_notif_key = None; // Upload 100 content and notif one-time keys in batches of 20 keys for request_num in 0..5 { let content_keys: Vec<_> = (0..20).map(|_| generate_random_olm_key()).collect(); let notif_keys: Vec<_> = (0..20).map(|_| generate_random_olm_key()).collect(); if request_num == 0 { - expected_first_retrieved_content_key = content_keys.get(0).cloned(); - expected_first_retrieved_notif_key = notif_keys.get(0).cloned(); + expected_first_retrieved_content_key = content_keys.first().cloned(); + expected_first_retrieved_notif_key = notif_keys.first().cloned(); expected_second_retrieved_content_key = content_keys.get(5).cloned(); expected_second_retrieved_notif_key = notif_keys.get(5).cloned(); } let upload_request = UploadOneTimeKeysRequest { content_one_time_prekeys: content_keys, notif_one_time_prekeys: notif_keys, }; identity_client .upload_one_time_keys(upload_request) .await .unwrap(); } let keyserver_request = OutboundKeysForUserRequest { user_id: device_info.user_id, }; let first_keyserver_response = identity_client .get_keyserver_keys(keyserver_request.clone()) .await .unwrap() .into_inner() .keyserver_info .unwrap(); assert!(first_keyserver_response.one_time_content_prekey.is_some()); assert!(first_keyserver_response.one_time_notif_prekey.is_some()); assert_eq!( expected_first_retrieved_content_key, first_keyserver_response.one_time_content_prekey ); assert_eq!( expected_first_retrieved_notif_key, first_keyserver_response.one_time_notif_prekey ); // Upload 5 more keys for each account let content_keys: Vec<_> = (0..5).map(|_| generate_random_olm_key()).collect(); let notif_keys: Vec<_> = (0..5).map(|_| generate_random_olm_key()).collect(); let final_upload_request = UploadOneTimeKeysRequest { content_one_time_prekeys: content_keys, notif_one_time_prekeys: notif_keys, }; identity_client .upload_one_time_keys(final_upload_request) .await .unwrap(); let second_keyserver_response = identity_client .get_keyserver_keys(keyserver_request) .await .unwrap() .into_inner() .keyserver_info .unwrap(); assert!(second_keyserver_response.one_time_content_prekey.is_some()); assert!(second_keyserver_response.one_time_notif_prekey.is_some()); assert_eq!( expected_second_retrieved_content_key, second_keyserver_response.one_time_content_prekey ); assert_eq!( expected_second_retrieved_notif_key, second_keyserver_response.one_time_notif_prekey ); } #[tokio::test] async fn max_24_keys_per_account_per_upload() { let device_info = register_user_device(None, None).await; let mut identity_client = get_auth_client( &service_addr::IDENTITY_GRPC.to_string(), device_info.user_id, device_info.device_id, device_info.access_token, PLACEHOLDER_CODE_VERSION, DEVICE_TYPE.to_string(), ) .await .expect("Couldn't connect to identity service"); // The limit is 24 keys per account per upload, so this should fail let content_keys = (0..26).map(|_| generate_random_olm_key()).collect(); let notif_keys = (0..20).map(|_| generate_random_olm_key()).collect(); let upload_request = UploadOneTimeKeysRequest { content_one_time_prekeys: content_keys, notif_one_time_prekeys: notif_keys, }; assert!(identity_client .upload_one_time_keys(upload_request) .await .is_err()); } diff --git a/services/identity/src/database.rs b/services/identity/src/database.rs index 0bb210b7a..9e97ab3b8 100644 --- a/services/identity/src/database.rs +++ b/services/identity/src/database.rs @@ -1,1224 +1,1225 @@ use comm_lib::aws::ddb::{ operation::{ delete_item::DeleteItemOutput, get_item::GetItemOutput, put_item::PutItemOutput, query::QueryOutput, }, primitives::Blob, types::{AttributeValue, PutRequest, WriteRequest}, }; use comm_lib::aws::{AwsConfig, DynamoDBClient}; use comm_lib::database::{ AttributeExtractor, AttributeMap, DBItemAttributeError, DBItemError, TryFromAttribute, }; use std::collections::{HashMap, HashSet}; use std::str::FromStr; use std::sync::Arc; pub use crate::database::device_list::DeviceIDAttribute; pub use crate::database::one_time_keys::OTKRow; use crate::{ constants::{error_types, USERS_TABLE_SOCIAL_PROOF_ATTRIBUTE_NAME}, ddb_utils::EthereumIdentity, device_list::SignedDeviceList, reserved_users::UserDetail, siwe::SocialProof, }; use crate::{ ddb_utils::{DBIdentity, OlmAccountType}, grpc_services::protos, }; use crate::{error::Error, grpc_utils::DeviceKeysInfo}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use tracing::{debug, error, info, warn}; use crate::client_service::{FlattenedDeviceKeyUpload, UserRegistrationInfo}; use crate::config::CONFIG; use crate::constants::{ NONCE_TABLE, NONCE_TABLE_CREATED_ATTRIBUTE, NONCE_TABLE_EXPIRATION_TIME_ATTRIBUTE, NONCE_TABLE_EXPIRATION_TIME_UNIX_ATTRIBUTE, NONCE_TABLE_PARTITION_KEY, RESERVED_USERNAMES_TABLE, RESERVED_USERNAMES_TABLE_PARTITION_KEY, RESERVED_USERNAMES_TABLE_USER_ID_ATTRIBUTE, USERS_TABLE, USERS_TABLE_DEVICES_MAP_DEVICE_TYPE_ATTRIBUTE_NAME, USERS_TABLE_FARCASTER_ID_ATTRIBUTE_NAME, USERS_TABLE_PARTITION_KEY, USERS_TABLE_REGISTRATION_ATTRIBUTE, USERS_TABLE_USERNAME_ATTRIBUTE, USERS_TABLE_USERNAME_INDEX, USERS_TABLE_WALLET_ADDRESS_ATTRIBUTE, USERS_TABLE_WALLET_ADDRESS_INDEX, }; use crate::id::generate_uuid; use crate::nonce::NonceData; use crate::token::AuthType; pub use grpc_clients::identity::DeviceType; mod device_list; mod farcaster; mod one_time_keys; mod token; mod workflows; pub use device_list::{DeviceListRow, DeviceListUpdate, DeviceRow}; use self::device_list::Prekey; #[derive(Serialize, Deserialize)] pub struct OlmKeys { pub curve25519: String, pub ed25519: String, } #[derive(Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct KeyPayload { pub notification_identity_public_keys: OlmKeys, pub primary_identity_public_keys: OlmKeys, } impl FromStr for KeyPayload { type Err = serde_json::Error; // The payload is held in the database as an escaped JSON payload. // Escaped double quotes need to be trimmed before attempting to serialize fn from_str(payload: &str) -> Result { serde_json::from_str(&payload.replace(r#"\""#, r#"""#)) } } pub struct DBDeviceTypeInt(pub i32); impl TryFrom for DeviceType { type Error = crate::error::Error; fn try_from(value: DBDeviceTypeInt) -> Result { let device_result = DeviceType::try_from(value.0); device_result.map_err(|_| { Error::Attribute(DBItemError { attribute_name: USERS_TABLE_DEVICES_MAP_DEVICE_TYPE_ATTRIBUTE_NAME .to_string(), attribute_value: Some(AttributeValue::N(value.0.to_string())).into(), attribute_error: DBItemAttributeError::InvalidValue, }) }) } } pub struct OutboundKeys { pub key_payload: String, pub key_payload_signature: String, pub content_prekey: Prekey, pub notif_prekey: Prekey, pub content_one_time_key: Option, pub notif_one_time_key: Option, } impl From for protos::auth::OutboundKeyInfo { fn from(db_keys: OutboundKeys) -> Self { use protos::unauth::IdentityKeyInfo; Self { identity_info: Some(IdentityKeyInfo { payload: db_keys.key_payload, payload_signature: db_keys.key_payload_signature, }), content_prekey: Some(db_keys.content_prekey.into()), notif_prekey: Some(db_keys.notif_prekey.into()), one_time_content_prekey: db_keys.content_one_time_key, one_time_notif_prekey: db_keys.notif_one_time_key, } } } #[derive(Clone)] pub struct DatabaseClient { client: Arc, } impl DatabaseClient { pub fn new(aws_config: &AwsConfig) -> Self { let client = match &CONFIG.localstack_endpoint { Some(endpoint) => { info!( "Configuring DynamoDB client to use LocalStack endpoint: {}", endpoint ); let ddb_config_builder = comm_lib::aws::ddb::config::Builder::from(aws_config) .endpoint_url(endpoint); DynamoDBClient::from_conf(ddb_config_builder.build()) } None => DynamoDBClient::new(aws_config), }; DatabaseClient { client: Arc::new(client), } } pub async fn add_password_user_to_users_table( &self, registration_state: UserRegistrationInfo, password_file: Vec, code_version: u64, access_token_creation_time: DateTime, ) -> Result { let device_key_upload = registration_state.flattened_device_key_upload; let user_id = self .add_user_to_users_table( Some((registration_state.username, Blob::new(password_file))), None, registration_state.user_id, registration_state.farcaster_id, ) .await?; // When initial device list is present, we should apply it // instead of auto-creating one. if let Some(device_list) = registration_state.initial_device_list { let initial_device_list = DeviceListUpdate::try_from(device_list)?; self .register_primary_device( &user_id, device_key_upload.clone(), code_version, access_token_creation_time, initial_device_list, ) .await?; } else { self .add_device( &user_id, device_key_upload.clone(), code_version, access_token_creation_time, ) .await?; } self .append_one_time_prekeys( &user_id, &device_key_upload.device_id_key, &device_key_upload.content_one_time_keys, &device_key_upload.notif_one_time_keys, ) .await?; Ok(user_id) } + #[allow(clippy::too_many_arguments)] pub async fn add_wallet_user_to_users_table( &self, flattened_device_key_upload: FlattenedDeviceKeyUpload, wallet_address: String, social_proof: SocialProof, user_id: Option, code_version: u64, access_token_creation_time: DateTime, farcaster_id: Option, initial_device_list: Option, ) -> Result { let wallet_identity = EthereumIdentity { wallet_address, social_proof, }; let user_id = self .add_user_to_users_table( None, Some(wallet_identity), user_id, farcaster_id, ) .await?; // When initial device list is present, we should apply it // instead of auto-creating one. if let Some(device_list) = initial_device_list { let initial_device_list = DeviceListUpdate::try_from(device_list)?; self .register_primary_device( &user_id, flattened_device_key_upload.clone(), code_version, access_token_creation_time, initial_device_list, ) .await?; } else { self .add_device( &user_id, flattened_device_key_upload.clone(), code_version, access_token_creation_time, ) .await?; } self .append_one_time_prekeys( &user_id, &flattened_device_key_upload.device_id_key, &flattened_device_key_upload.content_one_time_keys, &flattened_device_key_upload.notif_one_time_keys, ) .await?; Ok(user_id) } async fn add_user_to_users_table( &self, username_and_password_file: Option<(String, Blob)>, wallet_identity: Option, user_id: Option, farcaster_id: Option, ) -> Result { let user_id = user_id.unwrap_or_else(generate_uuid); let mut user = HashMap::from([( USERS_TABLE_PARTITION_KEY.to_string(), AttributeValue::S(user_id.clone()), )]); if let Some((username, password_file)) = username_and_password_file { user.insert( USERS_TABLE_USERNAME_ATTRIBUTE.to_string(), AttributeValue::S(username), ); user.insert( USERS_TABLE_REGISTRATION_ATTRIBUTE.to_string(), AttributeValue::B(password_file), ); } if let Some(eth_identity) = wallet_identity { user.insert( USERS_TABLE_WALLET_ADDRESS_ATTRIBUTE.to_string(), AttributeValue::S(eth_identity.wallet_address), ); user.insert( USERS_TABLE_SOCIAL_PROOF_ATTRIBUTE_NAME.to_string(), eth_identity.social_proof.into(), ); } if let Some(fid) = farcaster_id { user.insert( USERS_TABLE_FARCASTER_ID_ATTRIBUTE_NAME.to_string(), AttributeValue::S(fid), ); } self .client .put_item() .table_name(USERS_TABLE) .set_item(Some(user)) // make sure we don't accidentaly overwrite existing row .condition_expression("attribute_not_exists(#pk)") .expression_attribute_names("#pk", USERS_TABLE_PARTITION_KEY) .send() .await .map_err(|e| Error::AwsSdk(e.into()))?; Ok(user_id) } pub async fn add_user_device( &self, user_id: String, flattened_device_key_upload: FlattenedDeviceKeyUpload, code_version: u64, access_token_creation_time: DateTime, ) -> Result<(), Error> { let content_one_time_keys = flattened_device_key_upload.content_one_time_keys.clone(); let notif_one_time_keys = flattened_device_key_upload.notif_one_time_keys.clone(); // add device to the device list if not exists let device_id = flattened_device_key_upload.device_id_key.clone(); let device_exists = self .device_exists(user_id.clone(), device_id.clone()) .await?; if device_exists { self .update_device_login_time( user_id.clone(), device_id, access_token_creation_time, ) .await?; return Ok(()); } // add device to the new device list self .add_device( &user_id, flattened_device_key_upload, code_version, access_token_creation_time, ) .await?; self .append_one_time_prekeys( &user_id, &device_id, &content_one_time_keys, ¬if_one_time_keys, ) .await?; Ok(()) } pub async fn get_keyserver_keys_for_user( &self, user_id: &str, ) -> Result, Error> { use crate::grpc_services::protos::unauth::DeviceType as GrpcDeviceType; let user_devices = self.get_current_devices(user_id).await?; let maybe_keyserver_device = user_devices .into_iter() .find(|device| device.device_type == GrpcDeviceType::Keyserver); let Some(keyserver) = maybe_keyserver_device else { return Ok(None); }; debug!( "Found keyserver in devices table (ID={})", &keyserver.device_id ); let (notif_one_time_key, requested_more_keys) = self .get_one_time_key( user_id, &keyserver.device_id, OlmAccountType::Notification, true, ) .await?; let (content_one_time_key, _) = self .get_one_time_key( user_id, &keyserver.device_id, OlmAccountType::Content, !requested_more_keys, ) .await?; debug!( "Able to get notif one-time key for keyserver {}: {}", &keyserver.device_id, notif_one_time_key.is_some() ); debug!( "Able to get content one-time key for keyserver {}: {}", &keyserver.device_id, content_one_time_key.is_some() ); let outbound_payload = OutboundKeys { key_payload: keyserver.device_key_info.key_payload, key_payload_signature: keyserver.device_key_info.key_payload_signature, content_prekey: keyserver.content_prekey, notif_prekey: keyserver.notif_prekey, content_one_time_key, notif_one_time_key, }; Ok(Some(outbound_payload)) } pub async fn get_keyserver_device_id_for_user( &self, user_id: &str, ) -> Result, Error> { use crate::grpc_services::protos::unauth::DeviceType as GrpcDeviceType; let user_devices = self.get_current_devices(user_id).await?; let maybe_keyserver_device_id = user_devices .into_iter() .find(|device| device.device_type == GrpcDeviceType::Keyserver) .map(|device| device.device_id); Ok(maybe_keyserver_device_id) } pub async fn update_user_password( &self, user_id: String, password_file: Vec, ) -> Result<(), Error> { let update_expression = format!("SET {} = :p", USERS_TABLE_REGISTRATION_ATTRIBUTE); let expression_attribute_values = HashMap::from([( ":p".to_string(), AttributeValue::B(Blob::new(password_file)), )]); self .client .update_item() .table_name(USERS_TABLE) .key(USERS_TABLE_PARTITION_KEY, AttributeValue::S(user_id)) .update_expression(update_expression) .set_expression_attribute_values(Some(expression_attribute_values)) .send() .await .map_err(|e| Error::AwsSdk(e.into()))?; Ok(()) } #[tracing::instrument(skip_all)] pub async fn delete_user( &self, user_id: String, ) -> Result { // We must delete the one-time keys first because doing so requires device // IDs from the devices table debug!(user_id, "Attempting to delete user's one-time keys"); self.delete_otks_table_rows_for_user(&user_id).await?; debug!(user_id, "Attempting to delete user's devices"); self.delete_devices_table_rows_for_user(&user_id).await?; debug!(user_id, "Attempting to delete user's access tokens"); self.delete_all_tokens_for_user(&user_id).await?; debug!(user_id, "Attempting to delete user"); match self .client .delete_item() .table_name(USERS_TABLE) .key( USERS_TABLE_PARTITION_KEY, AttributeValue::S(user_id.clone()), ) .send() .await { Ok(out) => { info!("User has been deleted {}", user_id); Ok(out) } Err(e) => { error!( errorType = error_types::GENERIC_DB_LOG, "DynamoDB client failed to delete user {}", user_id ); Err(Error::AwsSdk(e.into())) } } } pub async fn wallet_address_taken( &self, wallet_address: String, ) -> Result { let result = self .get_user_id_from_user_info(wallet_address, &AuthType::Wallet) .await?; Ok(result.is_some()) } pub async fn username_taken(&self, username: String) -> Result { let result = self .get_user_id_from_user_info(username, &AuthType::Password) .await?; Ok(result.is_some()) } pub async fn filter_out_taken_usernames( &self, user_details: Vec, ) -> Result, Error> { let db_usernames = self.get_all_usernames().await?; let db_usernames_set: HashSet = db_usernames.into_iter().collect(); let available_user_details: Vec = user_details .into_iter() .filter(|user_detail| !db_usernames_set.contains(&user_detail.username)) .collect(); Ok(available_user_details) } #[tracing::instrument(skip_all)] async fn get_user_from_user_info( &self, user_info: String, auth_type: &AuthType, ) -> Result>, Error> { let (index, attribute_name) = match auth_type { AuthType::Password => { (USERS_TABLE_USERNAME_INDEX, USERS_TABLE_USERNAME_ATTRIBUTE) } AuthType::Wallet => ( USERS_TABLE_WALLET_ADDRESS_INDEX, USERS_TABLE_WALLET_ADDRESS_ATTRIBUTE, ), }; match self .client .query() .table_name(USERS_TABLE) .index_name(index) .key_condition_expression(format!("{} = :u", attribute_name)) .expression_attribute_values(":u", AttributeValue::S(user_info.clone())) .send() .await { Ok(QueryOutput { items: Some(items), .. }) => { let num_items = items.len(); if num_items == 0 { return Ok(None); } if num_items > 1 { warn!( "{} user IDs associated with {} {}: {:?}", num_items, attribute_name, user_info, items ); } let first_item = items[0].clone(); let user_id = first_item .get(USERS_TABLE_PARTITION_KEY) .ok_or(DBItemError { attribute_name: USERS_TABLE_PARTITION_KEY.to_string(), attribute_value: None.into(), attribute_error: DBItemAttributeError::Missing, })? .as_s() .map_err(|_| DBItemError { attribute_name: USERS_TABLE_PARTITION_KEY.to_string(), attribute_value: first_item .get(USERS_TABLE_PARTITION_KEY) .cloned() .into(), attribute_error: DBItemAttributeError::IncorrectType, })?; let result = self.get_item_from_users_table(user_id).await?; Ok(result.item) } Ok(_) => { info!( "No item found for {} {} in users table", attribute_name, user_info ); Ok(None) } Err(e) => { error!( errorType = error_types::GENERIC_DB_LOG, "DynamoDB client failed to get user from {} {}: {}", attribute_name, user_info, e ); Err(Error::AwsSdk(e.into())) } } } pub async fn get_keys_for_user( &self, user_id: &str, get_one_time_keys: bool, ) -> Result, Error> { let mut devices_response = self.get_keys_for_user_devices(user_id).await?; if devices_response.is_empty() { debug!("No devices found for user {}", user_id); return Ok(None); } if get_one_time_keys { for (device_id_key, device_keys) in devices_response.iter_mut() { let requested_more_keys; (device_keys.notif_one_time_key, requested_more_keys) = self .get_one_time_key( user_id, device_id_key, OlmAccountType::Notification, true, ) .await?; (device_keys.content_one_time_key, _) = self .get_one_time_key( user_id, device_id_key, OlmAccountType::Content, !requested_more_keys, ) .await?; } } Ok(Some(devices_response)) } pub async fn get_user_id_from_user_info( &self, user_info: String, auth_type: &AuthType, ) -> Result, Error> { match self .get_user_from_user_info(user_info.clone(), auth_type) .await { Ok(Some(mut user)) => user .take_attr(USERS_TABLE_PARTITION_KEY) .map(Some) .map_err(Error::Attribute), Ok(_) => Ok(None), Err(e) => Err(e), } } #[tracing::instrument(skip_all)] pub async fn get_user_id_and_password_file_from_username( &self, username: &str, ) -> Result)>, Error> { match self .get_user_from_user_info(username.to_string(), &AuthType::Password) .await { Ok(Some(mut user)) => { let user_id = user.take_attr(USERS_TABLE_PARTITION_KEY)?; let password_file = parse_registration_data_attribute( user.remove(USERS_TABLE_REGISTRATION_ATTRIBUTE), )?; Ok(Some((user_id, password_file))) } Ok(_) => { info!( "No item found for user {} in PAKE registration table", username ); Ok(None) } Err(e) => { error!( errorType = error_types::GENERIC_DB_LOG, "DynamoDB client failed to get registration data for user {}: {}", username, e ); Err(e) } } } pub async fn get_username_and_password_file( &self, user_id: &str, ) -> Result)>, Error> { let Some(mut user) = self.get_item_from_users_table(user_id).await?.item else { return Ok(None); }; let username = user.take_attr(USERS_TABLE_USERNAME_ATTRIBUTE)?; let password_file = parse_registration_data_attribute( user.remove(USERS_TABLE_REGISTRATION_ATTRIBUTE), )?; Ok(Some((username, password_file))) } async fn get_item_from_users_table( &self, user_id: &str, ) -> Result { let primary_key = create_simple_primary_key(( USERS_TABLE_PARTITION_KEY.to_string(), user_id.to_string(), )); self .client .get_item() .table_name(USERS_TABLE) .set_key(Some(primary_key)) .consistent_read(true) .send() .await .map_err(|e| Error::AwsSdk(e.into())) } pub async fn find_db_user_identities( &self, user_ids: impl IntoIterator, ) -> Result, Error> { use comm_lib::database::batch_operations::{ batch_get, ExponentialBackoffConfig, }; let primary_keys = user_ids.into_iter().map(|user_id| { create_simple_primary_key(( USERS_TABLE_PARTITION_KEY.to_string(), user_id, )) }); let projection_expression = [ USERS_TABLE_PARTITION_KEY, USERS_TABLE_USERNAME_ATTRIBUTE, USERS_TABLE_WALLET_ADDRESS_ATTRIBUTE, USERS_TABLE_SOCIAL_PROOF_ATTRIBUTE_NAME, USERS_TABLE_FARCASTER_ID_ATTRIBUTE_NAME, ] .join(", "); debug!( num_requests = primary_keys.size_hint().0, "Attempting to batch get user identifiers" ); let responses = batch_get( &self.client, USERS_TABLE, primary_keys, Some(projection_expression), ExponentialBackoffConfig::default(), ) .await .map_err(Error::from)?; debug!("Found {} matching user identifiers in DDB", responses.len()); let mut results = HashMap::with_capacity(responses.len()); for response in responses { let user_id = response.get_attr(USERS_TABLE_PARTITION_KEY)?; // if this fails, it means that projection expression didnt have all attrs it needed let identity = DBIdentity::try_from(response)?; results.insert(user_id, identity); } Ok(results) } /// Retrieves username for password users or wallet address for wallet users /// Returns `None` if user not found #[tracing::instrument(skip_all)] pub async fn get_user_identity( &self, user_id: &str, ) -> Result, Error> { self .get_item_from_users_table(user_id) .await? .item .map(DBIdentity::try_from) .transpose() .map_err(|e| { error!( user_id, errorType = error_types::GENERIC_DB_LOG, "Database item is missing an identifier" ); e }) } async fn get_all_usernames(&self) -> Result, Error> { let scan_output = self .client .scan() .table_name(USERS_TABLE) .projection_expression(USERS_TABLE_USERNAME_ATTRIBUTE) .send() .await .map_err(|e| Error::AwsSdk(e.into()))?; let mut result = Vec::new(); if let Some(attributes) = scan_output.items { for mut attribute in attributes { if let Ok(username) = attribute.take_attr(USERS_TABLE_USERNAME_ATTRIBUTE) { result.push(username); } } } Ok(result) } pub async fn get_all_user_details(&self) -> Result, Error> { let scan_output = self .client .scan() .table_name(USERS_TABLE) .projection_expression(format!( "{USERS_TABLE_USERNAME_ATTRIBUTE}, {USERS_TABLE_PARTITION_KEY}" )) .send() .await .map_err(|e| Error::AwsSdk(e.into()))?; let mut result = Vec::new(); if let Some(attributes) = scan_output.items { for mut attribute in attributes { if let (Ok(username), Ok(user_id)) = ( attribute.take_attr(USERS_TABLE_USERNAME_ATTRIBUTE), attribute.take_attr(USERS_TABLE_PARTITION_KEY), ) { result.push(UserDetail { username, user_id }); } } } Ok(result) } pub async fn get_all_reserved_user_details( &self, ) -> Result, Error> { let scan_output = self .client .scan() .table_name(RESERVED_USERNAMES_TABLE) .projection_expression(format!( "{RESERVED_USERNAMES_TABLE_PARTITION_KEY},\ {RESERVED_USERNAMES_TABLE_USER_ID_ATTRIBUTE}" )) .send() .await .map_err(|e| Error::AwsSdk(e.into()))?; let mut result = Vec::new(); if let Some(attributes) = scan_output.items { for mut attribute in attributes { if let (Ok(username), Ok(user_id)) = ( attribute.take_attr(USERS_TABLE_USERNAME_ATTRIBUTE), attribute.take_attr(RESERVED_USERNAMES_TABLE_USER_ID_ATTRIBUTE), ) { result.push(UserDetail { username, user_id }); } } } Ok(result) } pub async fn add_nonce_to_nonces_table( &self, nonce_data: NonceData, ) -> Result { let item = HashMap::from([ ( NONCE_TABLE_PARTITION_KEY.to_string(), AttributeValue::S(nonce_data.nonce), ), ( NONCE_TABLE_CREATED_ATTRIBUTE.to_string(), AttributeValue::S(nonce_data.created.to_rfc3339()), ), ( NONCE_TABLE_EXPIRATION_TIME_ATTRIBUTE.to_string(), AttributeValue::S(nonce_data.expiration_time.to_rfc3339()), ), ( NONCE_TABLE_EXPIRATION_TIME_UNIX_ATTRIBUTE.to_string(), AttributeValue::N(nonce_data.expiration_time.timestamp().to_string()), ), ]); self .client .put_item() .table_name(NONCE_TABLE) .set_item(Some(item)) .send() .await .map_err(|e| Error::AwsSdk(e.into())) } pub async fn get_nonce_from_nonces_table( &self, nonce_value: impl Into, ) -> Result, Error> { let get_response = self .client .get_item() .table_name(NONCE_TABLE) .key( NONCE_TABLE_PARTITION_KEY, AttributeValue::S(nonce_value.into()), ) .send() .await .map_err(|e| Error::AwsSdk(e.into()))?; let Some(mut item) = get_response.item else { return Ok(None); }; let nonce = item.take_attr(NONCE_TABLE_PARTITION_KEY)?; let created = DateTime::::try_from_attr( NONCE_TABLE_CREATED_ATTRIBUTE, item.remove(NONCE_TABLE_CREATED_ATTRIBUTE), )?; let expiration_time = DateTime::::try_from_attr( NONCE_TABLE_EXPIRATION_TIME_ATTRIBUTE, item.remove(NONCE_TABLE_EXPIRATION_TIME_ATTRIBUTE), )?; Ok(Some(NonceData { nonce, created, expiration_time, })) } pub async fn remove_nonce_from_nonces_table( &self, nonce: impl Into, ) -> Result<(), Error> { self .client .delete_item() .table_name(NONCE_TABLE) .key(NONCE_TABLE_PARTITION_KEY, AttributeValue::S(nonce.into())) .send() .await .map_err(|e| Error::AwsSdk(e.into()))?; Ok(()) } pub async fn add_usernames_to_reserved_usernames_table( &self, user_details: Vec, ) -> Result<(), Error> { // A single call to BatchWriteItem can consist of up to 25 operations for user_chunk in user_details.chunks(25) { let write_requests = user_chunk .iter() .map(|user_detail| { let put_request = PutRequest::builder() .item( RESERVED_USERNAMES_TABLE_PARTITION_KEY, AttributeValue::S(user_detail.username.to_string()), ) .item( RESERVED_USERNAMES_TABLE_USER_ID_ATTRIBUTE, AttributeValue::S(user_detail.user_id.to_string()), ) .build(); WriteRequest::builder().put_request(put_request).build() }) .collect(); self .client .batch_write_item() .request_items(RESERVED_USERNAMES_TABLE, write_requests) .send() .await .map_err(|e| Error::AwsSdk(e.into()))?; } info!("Batch write item to reserved usernames table succeeded"); Ok(()) } #[tracing::instrument(skip_all)] pub async fn delete_username_from_reserved_usernames_table( &self, username: String, ) -> Result { debug!( "Attempting to delete username {} from reserved usernames table", username ); match self .client .delete_item() .table_name(RESERVED_USERNAMES_TABLE) .key( RESERVED_USERNAMES_TABLE_PARTITION_KEY, AttributeValue::S(username.clone()), ) .send() .await { Ok(out) => { info!( "Username {} has been deleted from reserved usernames table", username ); Ok(out) } Err(e) => { error!(errorType = error_types::GENERIC_DB_LOG, "DynamoDB client failed to delete username {} from reserved usernames table", username); Err(Error::AwsSdk(e.into())) } } } pub async fn username_in_reserved_usernames_table( &self, username: &str, ) -> Result { match self .client .get_item() .table_name(RESERVED_USERNAMES_TABLE) .key( RESERVED_USERNAMES_TABLE_PARTITION_KEY.to_string(), AttributeValue::S(username.to_string()), ) .consistent_read(true) .send() .await { Ok(GetItemOutput { item: Some(_), .. }) => Ok(true), Ok(_) => Ok(false), Err(e) => Err(Error::AwsSdk(e.into())), } } } type AttributeName = String; type Devices = HashMap; fn create_simple_primary_key( partition_key: (AttributeName, String), ) -> HashMap { HashMap::from([(partition_key.0, AttributeValue::S(partition_key.1))]) } fn create_composite_primary_key( partition_key: (AttributeName, String), sort_key: (AttributeName, String), ) -> HashMap { let mut primary_key = create_simple_primary_key(partition_key); primary_key.insert(sort_key.0, AttributeValue::S(sort_key.1)); primary_key } fn parse_registration_data_attribute( attribute: Option, ) -> Result, DBItemError> { match attribute { Some(AttributeValue::B(server_registration_bytes)) => { Ok(server_registration_bytes.into_inner()) } Some(_) => Err(DBItemError::new( USERS_TABLE_REGISTRATION_ATTRIBUTE.to_string(), attribute.into(), DBItemAttributeError::IncorrectType, )), None => Err(DBItemError::new( USERS_TABLE_REGISTRATION_ATTRIBUTE.to_string(), attribute.into(), DBItemAttributeError::Missing, )), } } #[deprecated(note = "Use `comm_lib` counterpart instead")] #[allow(dead_code)] fn parse_map_attribute( attribute_name: &str, attribute_value: Option, ) -> Result { match attribute_value { Some(AttributeValue::M(map)) => Ok(map), Some(_) => { error!( attribute = attribute_name, value = ?attribute_value, error_type = "IncorrectType", errorType = error_types::GENERIC_DB_LOG, "Unexpected attribute type when parsing map attribute" ); Err(DBItemError::new( attribute_name.to_string(), attribute_value.into(), DBItemAttributeError::IncorrectType, )) } None => { error!( attribute = attribute_name, error_type = "Missing", errorType = error_types::GENERIC_DB_LOG, "Attribute is missing" ); Err(DBItemError::new( attribute_name.to_string(), attribute_value.into(), DBItemAttributeError::Missing, )) } } } #[cfg(test)] mod tests { use super::*; #[test] fn test_create_simple_primary_key() { let partition_key_name = "userID".to_string(); let partition_key_value = "12345".to_string(); let partition_key = (partition_key_name.clone(), partition_key_value.clone()); let mut primary_key = create_simple_primary_key(partition_key); assert_eq!(primary_key.len(), 1); let attribute = primary_key.remove(&partition_key_name); assert!(attribute.is_some()); assert_eq!(attribute, Some(AttributeValue::S(partition_key_value))); } #[test] fn test_create_composite_primary_key() { let partition_key_name = "userID".to_string(); let partition_key_value = "12345".to_string(); let partition_key = (partition_key_name.clone(), partition_key_value.clone()); let sort_key_name = "deviceID".to_string(); let sort_key_value = "54321".to_string(); let sort_key = (sort_key_name.clone(), sort_key_value.clone()); let mut primary_key = create_composite_primary_key(partition_key, sort_key); assert_eq!(primary_key.len(), 2); let partition_key_attribute = primary_key.remove(&partition_key_name); assert!(partition_key_attribute.is_some()); assert_eq!( partition_key_attribute, Some(AttributeValue::S(partition_key_value)) ); let sort_key_attribute = primary_key.remove(&sort_key_name); assert!(sort_key_attribute.is_some()); assert_eq!(sort_key_attribute, Some(AttributeValue::S(sort_key_value))) } #[test] fn validate_keys() { // Taken from test user let example_payload = r#"{\"notificationIdentityPublicKeys\":{\"curve25519\":\"DYmV8VdkjwG/VtC8C53morogNJhpTPT/4jzW0/cxzQo\",\"ed25519\":\"D0BV2Y7Qm36VUtjwyQTJJWYAycN7aMSJmhEsRJpW2mk\"},\"primaryIdentityPublicKeys\":{\"curve25519\":\"Y4ZIqzpE1nv83kKGfvFP6rifya0itRg2hifqYtsISnk\",\"ed25519\":\"cSlL+VLLJDgtKSPlIwoCZg0h0EmHlQoJC08uV/O+jvg\"}}"#; let serialized_payload = KeyPayload::from_str(example_payload).unwrap(); assert_eq!( serialized_payload .notification_identity_public_keys .curve25519, "DYmV8VdkjwG/VtC8C53morogNJhpTPT/4jzW0/cxzQo" ); } #[test] fn test_int_to_device_type() { let valid_result = DeviceType::try_from(3); assert!(valid_result.is_ok()); assert_eq!(valid_result.unwrap(), DeviceType::Android); let invalid_result = DeviceType::try_from(6); assert!(invalid_result.is_err()); } } diff --git a/services/identity/src/grpc_services/authenticated.rs b/services/identity/src/grpc_services/authenticated.rs index 33dd565e5..11afb67aa 100644 --- a/services/identity/src/grpc_services/authenticated.rs +++ b/services/identity/src/grpc_services/authenticated.rs @@ -1,745 +1,746 @@ use std::collections::HashMap; use crate::config::CONFIG; use crate::database::DeviceListUpdate; use crate::device_list::SignedDeviceList; use crate::{ client_service::{handle_db_error, UpdateState, WorkflowInProgress}, constants::{error_types, request_metadata}, database::DatabaseClient, grpc_services::shared::get_value, }; use chrono::DateTime; use comm_opaque2::grpc::protocol_error_to_grpc_status; use tonic::{Request, Response, Status}; use tracing::{debug, error, trace, warn}; use super::protos::auth::{ identity_client_service_server::IdentityClientService, DeletePasswordUserFinishRequest, DeletePasswordUserStartRequest, DeletePasswordUserStartResponse, GetDeviceListRequest, GetDeviceListResponse, InboundKeyInfo, InboundKeysForUserRequest, InboundKeysForUserResponse, KeyserverKeysResponse, LinkFarcasterAccountRequest, OutboundKeyInfo, OutboundKeysForUserRequest, OutboundKeysForUserResponse, PeersDeviceListsRequest, PeersDeviceListsResponse, RefreshUserPrekeysRequest, UpdateDeviceListRequest, UpdateUserPasswordFinishRequest, UpdateUserPasswordStartRequest, UpdateUserPasswordStartResponse, UploadOneTimeKeysRequest, UserIdentitiesRequest, UserIdentitiesResponse, }; use super::protos::unauth::Empty; #[derive(derive_more::Constructor)] pub struct AuthenticatedService { db_client: DatabaseClient, } fn get_auth_info(req: &Request<()>) -> Option<(String, String, String)> { trace!("Retrieving auth info for request: {:?}", req); let user_id = get_value(req, request_metadata::USER_ID)?; let device_id = get_value(req, request_metadata::DEVICE_ID)?; let access_token = get_value(req, request_metadata::ACCESS_TOKEN)?; Some((user_id, device_id, access_token)) } pub fn auth_interceptor( req: Request<()>, db_client: &DatabaseClient, ) -> Result, Status> { trace!("Intercepting request to check auth info: {:?}", req); let (user_id, device_id, access_token) = get_auth_info(&req) .ok_or_else(|| Status::unauthenticated("Missing credentials"))?; let handle = tokio::runtime::Handle::current(); let new_db_client = db_client.clone(); // This function cannot be `async`, yet must call the async db call // Force tokio to resolve future in current thread without an explicit .await let valid_token = tokio::task::block_in_place(move || { handle .block_on(new_db_client.verify_access_token( user_id, device_id, access_token, )) .map_err(handle_db_error) })?; if !valid_token { return Err(Status::aborted("Bad Credentials")); } Ok(req) } pub fn get_user_and_device_id( request: &Request, ) -> Result<(String, String), Status> { let user_id = get_value(request, request_metadata::USER_ID) .ok_or_else(|| Status::unauthenticated("Missing user_id field"))?; let device_id = get_value(request, request_metadata::DEVICE_ID) .ok_or_else(|| Status::unauthenticated("Missing device_id field"))?; Ok((user_id, device_id)) } #[tonic::async_trait] impl IdentityClientService for AuthenticatedService { #[tracing::instrument(skip_all)] async fn refresh_user_prekeys( &self, request: Request, ) -> Result, Status> { let (user_id, device_id) = get_user_and_device_id(&request)?; let message = request.into_inner(); debug!("Refreshing prekeys for user: {}", user_id); let content_keys = message .new_content_prekeys .ok_or_else(|| Status::invalid_argument("Missing content keys"))?; let notif_keys = message .new_notif_prekeys .ok_or_else(|| Status::invalid_argument("Missing notification keys"))?; self .db_client .update_device_prekeys( user_id, device_id, content_keys.into(), notif_keys.into(), ) .await .map_err(handle_db_error)?; let response = Response::new(Empty {}); Ok(response) } #[tracing::instrument(skip_all)] async fn get_outbound_keys_for_user( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); let user_id = &message.user_id; let devices_map = self .db_client .get_keys_for_user(user_id, true) .await .map_err(handle_db_error)? .ok_or_else(|| tonic::Status::not_found("user not found"))?; let transformed_devices = devices_map .into_iter() .map(|(key, device_info)| (key, OutboundKeyInfo::from(device_info))) .collect::>(); Ok(tonic::Response::new(OutboundKeysForUserResponse { devices: transformed_devices, })) } #[tracing::instrument(skip_all)] async fn get_inbound_keys_for_user( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); let user_id = &message.user_id; let devices_map = self .db_client .get_keys_for_user(user_id, false) .await .map_err(handle_db_error)? .ok_or_else(|| tonic::Status::not_found("user not found"))?; let transformed_devices = devices_map .into_iter() .map(|(key, device_info)| (key, InboundKeyInfo::from(device_info))) .collect::>(); let identifier = self .db_client .get_user_identity(user_id) .await .map_err(handle_db_error)? .ok_or_else(|| tonic::Status::not_found("user not found"))?; Ok(tonic::Response::new(InboundKeysForUserResponse { devices: transformed_devices, identity: Some(identifier.into()), })) } #[tracing::instrument(skip_all)] async fn get_keyserver_keys( &self, request: Request, ) -> Result, Status> { let message = request.into_inner(); let identifier = self .db_client .get_user_identity(&message.user_id) .await .map_err(handle_db_error)? .ok_or_else(|| tonic::Status::not_found("user not found"))?; let Some(keyserver_info) = self .db_client .get_keyserver_keys_for_user(&message.user_id) .await .map_err(handle_db_error)? else { return Err(Status::not_found("keyserver not found")); }; let primary_device_data = self .db_client .get_primary_device_data(&message.user_id) .await .map_err(handle_db_error)?; let primary_device_keys = primary_device_data.device_key_info; let response = Response::new(KeyserverKeysResponse { keyserver_info: Some(keyserver_info.into()), identity: Some(identifier.into()), primary_device_identity_info: Some(primary_device_keys.into()), }); return Ok(response); } #[tracing::instrument(skip_all)] async fn upload_one_time_keys( &self, request: tonic::Request, ) -> Result, tonic::Status> { let (user_id, device_id) = get_user_and_device_id(&request)?; let message = request.into_inner(); debug!("Attempting to update one time keys for user: {}", user_id); self .db_client .append_one_time_prekeys( &user_id, &device_id, &message.content_one_time_prekeys, &message.notif_one_time_prekeys, ) .await .map_err(handle_db_error)?; Ok(tonic::Response::new(Empty {})) } #[tracing::instrument(skip_all)] async fn update_user_password_start( &self, request: tonic::Request, ) -> Result, tonic::Status> { let (user_id, _) = get_user_and_device_id(&request)?; let message = request.into_inner(); let server_registration = comm_opaque2::server::Registration::new(); let server_message = server_registration .start( &CONFIG.server_setup, &message.opaque_registration_request, user_id.as_bytes(), ) .map_err(protocol_error_to_grpc_status)?; let update_state = UpdateState { user_id }; let session_id = self .db_client .insert_workflow(WorkflowInProgress::Update(update_state)) .await .map_err(handle_db_error)?; let response = UpdateUserPasswordStartResponse { session_id, opaque_registration_response: server_message, }; Ok(Response::new(response)) } #[tracing::instrument(skip_all)] async fn update_user_password_finish( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); let Some(WorkflowInProgress::Update(state)) = self .db_client .get_workflow(message.session_id) .await .map_err(handle_db_error)? else { return Err(tonic::Status::not_found("session not found")); }; let server_registration = comm_opaque2::server::Registration::new(); let password_file = server_registration .finish(&message.opaque_registration_upload) .map_err(protocol_error_to_grpc_status)?; self .db_client .update_user_password(state.user_id, password_file) .await .map_err(handle_db_error)?; let response = Empty {}; Ok(Response::new(response)) } #[tracing::instrument(skip_all)] async fn log_out_user( &self, request: tonic::Request, ) -> Result, tonic::Status> { let (user_id, device_id) = get_user_and_device_id(&request)?; self .db_client .remove_device(&user_id, &device_id) .await .map_err(handle_db_error)?; self .db_client .delete_otks_table_rows_for_user_device(&user_id, &device_id) .await .map_err(handle_db_error)?; self .db_client .delete_access_token_data(user_id, device_id) .await .map_err(handle_db_error)?; let response = Empty {}; Ok(Response::new(response)) } #[tracing::instrument(skip_all)] async fn log_out_secondary_device( &self, request: tonic::Request, ) -> Result, tonic::Status> { let (user_id, device_id) = get_user_and_device_id(&request)?; debug!( "Secondary device logout request for user_id={}, device_id={}", user_id, device_id ); self .verify_device_on_device_list( &user_id, &device_id, DeviceListItemKind::Secondary, ) .await?; self .db_client .delete_access_token_data(&user_id, &device_id) .await .map_err(handle_db_error)?; self .db_client .remove_device_data(&user_id, &device_id) .await .map_err(handle_db_error)?; self .db_client .delete_otks_table_rows_for_user_device(&user_id, &device_id) .await .map_err(handle_db_error)?; let response = Empty {}; Ok(Response::new(response)) } #[tracing::instrument(skip_all)] async fn delete_wallet_user( &self, request: tonic::Request, ) -> Result, tonic::Status> { let (user_id, _) = get_user_and_device_id(&request)?; debug!("Attempting to delete wallet user: {}", user_id); self .db_client .delete_user(user_id) .await .map_err(handle_db_error)?; let response = Empty {}; Ok(Response::new(response)) } #[tracing::instrument(skip_all)] async fn delete_password_user_start( &self, request: tonic::Request, ) -> Result, tonic::Status> { let (user_id, _) = get_user_and_device_id(&request)?; let message = request.into_inner(); debug!("Attempting to start deleting password user: {}", user_id); let maybe_username_and_password_file = self .db_client .get_username_and_password_file(&user_id) .await .map_err(handle_db_error)?; let Some((username, password_file_bytes)) = maybe_username_and_password_file else { return Err(tonic::Status::not_found("user not found")); }; let mut server_login = comm_opaque2::server::Login::new(); let server_response = server_login .start( &CONFIG.server_setup, &password_file_bytes, &message.opaque_login_request, username.as_bytes(), ) .map_err(protocol_error_to_grpc_status)?; let delete_state = construct_delete_password_user_info(server_login); let session_id = self .db_client .insert_workflow(WorkflowInProgress::PasswordUserDeletion(Box::new( delete_state, ))) .await .map_err(handle_db_error)?; let response = Response::new(DeletePasswordUserStartResponse { session_id, opaque_login_response: server_response, }); Ok(response) } #[tracing::instrument(skip_all)] async fn delete_password_user_finish( &self, request: tonic::Request, ) -> Result, tonic::Status> { let (user_id, _) = get_user_and_device_id(&request)?; let message = request.into_inner(); debug!("Attempting to finish deleting password user: {}", user_id); let Some(WorkflowInProgress::PasswordUserDeletion(state)) = self .db_client .get_workflow(message.session_id) .await .map_err(handle_db_error)? else { return Err(tonic::Status::not_found("session not found")); }; let mut server_login = state.opaque_server_login; server_login .finish(&message.opaque_login_upload) .map_err(protocol_error_to_grpc_status)?; self .db_client .delete_user(user_id) .await .map_err(handle_db_error)?; let response = Empty {}; Ok(Response::new(response)) } #[tracing::instrument(skip_all)] async fn get_device_list_for_user( &self, request: tonic::Request, ) -> Result, tonic::Status> { let GetDeviceListRequest { user_id, since_timestamp, } = request.into_inner(); let since = since_timestamp .map(|timestamp| { DateTime::from_timestamp_millis(timestamp) .ok_or_else(|| tonic::Status::invalid_argument("Invalid timestamp")) }) .transpose()?; let mut db_result = self .db_client .get_device_list_history(user_id, since) .await .map_err(handle_db_error)?; // these should be sorted already, but just in case db_result.sort_by_key(|list| list.timestamp); let device_list_updates: Vec = db_result .into_iter() .map(SignedDeviceList::try_from) .collect::, _>>()?; let stringified_updates = device_list_updates .iter() .map(SignedDeviceList::as_json_string) .collect::, _>>()?; Ok(Response::new(GetDeviceListResponse { device_list_updates: stringified_updates, })) } #[tracing::instrument(skip_all)] async fn get_device_lists_for_users( &self, request: tonic::Request, ) -> Result, tonic::Status> { let PeersDeviceListsRequest { user_ids } = request.into_inner(); // do all fetches concurrently let mut fetch_tasks = tokio::task::JoinSet::new(); let mut device_lists = HashMap::with_capacity(user_ids.len()); for user_id in user_ids { let db_client = self.db_client.clone(); fetch_tasks.spawn(async move { let result = db_client.get_current_device_list(&user_id).await; (user_id, result) }); } while let Some(task_result) = fetch_tasks.join_next().await { match task_result { Ok((user_id, Ok(Some(device_list_row)))) => { let signed_list = SignedDeviceList::try_from(device_list_row)?; let serialized_list = signed_list.as_json_string()?; device_lists.insert(user_id, serialized_list); } Ok((user_id, Ok(None))) => { warn!(user_id, "User has no device list, skipping!"); } Ok((user_id, Err(err))) => { error!( user_id, errorType = error_types::GRPC_SERVICES_LOG, "Failed fetching device list: {err}" ); // abort fetching other users fetch_tasks.abort_all(); return Err(handle_db_error(err)); } Err(join_error) => { error!( errorType = error_types::GRPC_SERVICES_LOG, "Failed to join device list task: {join_error}" ); fetch_tasks.abort_all(); return Err(Status::aborted("unexpected error")); } } } let response = PeersDeviceListsResponse { users_device_lists: device_lists, }; Ok(Response::new(response)) } #[tracing::instrument(skip_all)] async fn update_device_list( &self, request: tonic::Request, ) -> Result, tonic::Status> { let (user_id, _device_id) = get_user_and_device_id(&request)?; // TODO: when we stop doing "primary device rotation" (migration procedure) // we should verify if this RPC is called by primary device only let new_list = SignedDeviceList::try_from(request.into_inner())?; let update = DeviceListUpdate::try_from(new_list)?; self .db_client .apply_devicelist_update( &user_id, update, crate::device_list::validation::update_device_list_rpc_validator, ) .await .map_err(handle_db_error)?; Ok(Response::new(Empty {})) } #[tracing::instrument(skip_all)] async fn link_farcaster_account( &self, request: tonic::Request, ) -> Result, tonic::Status> { let (user_id, _) = get_user_and_device_id(&request)?; let message = request.into_inner(); let mut get_farcaster_users_response = self .db_client .get_farcaster_users(vec![message.farcaster_id.clone()]) .await .map_err(handle_db_error)?; if get_farcaster_users_response.len() > 1 { error!( errorType = error_types::GRPC_SERVICES_LOG, "multiple users associated with the same Farcaster ID" ); return Err(Status::failed_precondition("cannot link Farcaster ID")); } if let Some(u) = get_farcaster_users_response.pop() { if u.0.user_id == user_id { return Ok(Response::new(Empty {})); } else { return Err(Status::already_exists( "farcaster ID already associated with different user", )); } } self .db_client .add_farcaster_id(user_id, message.farcaster_id) .await .map_err(handle_db_error)?; let response = Empty {}; Ok(Response::new(response)) } #[tracing::instrument(skip_all)] async fn unlink_farcaster_account( &self, request: tonic::Request, ) -> Result, tonic::Status> { let (user_id, _) = get_user_and_device_id(&request)?; self .db_client .remove_farcaster_id(user_id) .await .map_err(handle_db_error)?; let response = Empty {}; Ok(Response::new(response)) } #[tracing::instrument(skip_all)] async fn find_user_identities( &self, request: tonic::Request, ) -> Result, tonic::Status> { let message = request.into_inner(); let results = self .db_client .find_db_user_identities(message.user_ids) .await .map_err(handle_db_error)?; let mapped_results = results .into_iter() .map(|(user_id, identifier)| (user_id, identifier.into())) .collect(); let response = UserIdentitiesResponse { identities: mapped_results, }; return Ok(Response::new(response)); } } +#[allow(dead_code)] enum DeviceListItemKind { Any, Primary, Secondary, } impl AuthenticatedService { async fn verify_device_on_device_list( &self, user_id: &String, device_id: &String, device_kind: DeviceListItemKind, ) -> Result<(), tonic::Status> { let device_list = self .db_client .get_current_device_list(user_id) .await .map_err(|err| { error!( user_id, errorType = error_types::GRPC_SERVICES_LOG, "Failed fetching device list: {err}" ); handle_db_error(err) })?; let Some(device_list) = device_list else { error!( user_id, errorType = error_types::GRPC_SERVICES_LOG, "User has no device list!" ); return Err(Status::failed_precondition("no device list")); }; use DeviceListItemKind as DeviceKind; let device_on_list = match device_kind { DeviceKind::Any => device_list.has_device(device_id), DeviceKind::Primary => device_list.is_primary_device(device_id), DeviceKind::Secondary => device_list.has_secondary_device(device_id), }; if !device_on_list { debug!( "Device {} not on device list for user {}", device_id, user_id ); return Err(Status::permission_denied("device not on device list")); } Ok(()) } } #[derive(Clone, serde::Serialize, serde::Deserialize)] pub struct DeletePasswordUserInfo { pub opaque_server_login: comm_opaque2::server::Login, } fn construct_delete_password_user_info( opaque_server_login: comm_opaque2::server::Login, ) -> DeletePasswordUserInfo { DeletePasswordUserInfo { opaque_server_login, } } diff --git a/services/identity/src/grpc_services/shared.rs b/services/identity/src/grpc_services/shared.rs index e1f3bc4e0..9507d70b8 100644 --- a/services/identity/src/grpc_services/shared.rs +++ b/services/identity/src/grpc_services/shared.rs @@ -1,35 +1,35 @@ use grpc_clients::error::unsupported_version; -use tonic::{IntoRequest, Request, Status}; -use tracing::{trace, Instrument}; +use tonic::{Request, Status}; +use tracing::trace; use crate::constants::{request_metadata, MIN_SUPPORTED_NATIVE_VERSION}; pub fn version_interceptor(req: Request<()>) -> Result, Status> { trace!("Intercepting request to check version: {:?}", req); match get_version_info(&req) { Some((version, platform)) if (platform == "ios" || platform == "android") && version < MIN_SUPPORTED_NATIVE_VERSION => { Err(unsupported_version()) } _ => Ok(req), } } fn get_version_info(req: &Request<()>) -> Option<(u64, String)> { trace!("Retrieving version info for request: {:?}", req); let code_version: u64 = get_value(req, request_metadata::CODE_VERSION)? .parse() .ok()?; let device_type = get_value(req, request_metadata::DEVICE_TYPE)?; Some((code_version, device_type)) } pub fn get_value(req: &Request, key: &str) -> Option { let raw_value = req.metadata().get(key)?; raw_value.to_str().ok().map(|s| s.to_string()) } diff --git a/services/identity/src/regex.rs b/services/identity/src/regex.rs index dfa1888b3..197ed265c 100644 --- a/services/identity/src/regex.rs +++ b/services/identity/src/regex.rs @@ -1,54 +1,58 @@ use regex::Regex; use crate::constants::VALID_USERNAME_REGEX_STRING; pub fn is_valid_username(candidate: &str) -> bool { let valid_username_regex = Regex::new(VALID_USERNAME_REGEX_STRING) .expect("regex pattern should be valid"); valid_username_regex.is_match(candidate) } #[cfg(test)] mod tests { use super::*; #[test] fn test_valid_username_with_dash_and_underscore() { assert!(is_valid_username("WRx16rC_VWX-"),); } #[test] fn test_valid_username_with_minimum_length() { assert!(is_valid_username("a")); } #[test] fn test_valid_username_with_maximum_length() { assert!(is_valid_username( "Dg75raz1KL0rzqFbnUDAtSp0KfvDThPGcYGHiEin9m7LCWYJ7vQYNkXpbTNJ7MYaaTeqApgYDr3XdNX0BmKetP0iJeKRZuDdSEuybnGrdivwSQPLrJ8rv2WMTGvNjjDypvMMbDXmQ7wt0jPFfKD41Uc07SkPFcBJkwBJ8WJuJWmm7m0nvxhFRieQn319qHk" )); } + #[allow(clippy::bool_assert_comparison)] #[test] fn test_invalid_username_too_short() { assert_eq!(is_valid_username(""), false); } + #[allow(clippy::bool_assert_comparison)] #[test] fn test_invalid_username_too_long() { assert_eq!( is_valid_username("XVnYhmjPrVpSwLZnSGjdnx5mmRFLbSh5RjjqU7Wf62LXe9Hyy0CfJQNXgTy62Uxzip538MMYtwvaqqA0fukYijWAuZdVVpYRMuaZ0gzxLgV6hzF9amvnjZGYgdKeEX45WErYq2nN7Q2ivcFfftWnpudPcPr0pxWRLdmqS37jNVHWMX919vXz0PdzeLNHx0TW"), false ); } + #[allow(clippy::bool_assert_comparison)] #[test] fn test_invalid_username_first_char_non_alphanumeric() { assert_eq!(is_valid_username("-asdf"), false); } + #[allow(clippy::bool_assert_comparison)] #[test] fn test_invalid_username_invalid_symbol() { assert_eq!(is_valid_username("asdf$"), false); } } diff --git a/services/identity/src/siwe.rs b/services/identity/src/siwe.rs index 45680da02..fcdd89514 100644 --- a/services/identity/src/siwe.rs +++ b/services/identity/src/siwe.rs @@ -1,169 +1,170 @@ use std::collections::HashMap; use chrono::Utc; use comm_lib::{ aws::ddb::types::AttributeValue, database::{AttributeExtractor, AttributeMap, TryFromAttribute}, }; use regex::Regex; use siwe::Message; use tonic::Status; use tracing::error; use crate::constants::{ error_types, SOCIAL_PROOF_MESSAGE_ATTRIBUTE, SOCIAL_PROOF_SIGNATURE_ATTRIBUTE, }; pub fn parse_and_verify_siwe_message( siwe_message: &str, siwe_signature: &str, ) -> Result { let siwe_message: Message = siwe_message.parse().map_err(|e| { error!( errorType = error_types::SIWE_LOG, "Failed to parse SIWE message: {}", e ); Status::invalid_argument("invalid message") })?; let decoded_signature = hex::decode(siwe_signature.trim_start_matches("0x")) .map_err(|e| { error!( errorType = error_types::SIWE_LOG, "Failed to decode SIWE signature: {}", e ); Status::invalid_argument("invalid signature") })?; let signature = decoded_signature.try_into().map_err(|e| { error!( errorType = error_types::SIWE_LOG, "Conversion to SIWE signature failed: {:?}", e ); Status::invalid_argument("invalid message") })?; siwe_message .verify(signature, None, None, Some(&Utc::now())) .map_err(|e| { error!( errorType = error_types::SIWE_LOG, "Signature verification failed: {}", e ); Status::unauthenticated("message not authenticated") })?; Ok(siwe_message) } pub fn is_valid_ethereum_address(candidate: &str) -> bool { let ethereum_address_regex = Regex::new(r"^0x[a-fA-F0-9]{40}$").unwrap(); ethereum_address_regex.is_match(candidate) } #[derive(derive_more::Constructor)] pub struct SocialProof { pub message: String, pub signature: String, } impl From for AttributeValue { fn from(value: SocialProof) -> Self { AttributeValue::M(HashMap::from([ ( SOCIAL_PROOF_MESSAGE_ATTRIBUTE.to_string(), AttributeValue::S(value.message), ), ( SOCIAL_PROOF_SIGNATURE_ATTRIBUTE.to_string(), AttributeValue::S(value.signature), ), ])) } } impl TryFrom for SocialProof { type Error = comm_lib::database::DBItemError; fn try_from(mut attrs: AttributeMap) -> Result { let message = attrs.take_attr(SOCIAL_PROOF_MESSAGE_ATTRIBUTE)?; let signature = attrs.take_attr(SOCIAL_PROOF_SIGNATURE_ATTRIBUTE)?; Ok(Self { message, signature }) } } impl TryFromAttribute for SocialProof { fn try_from_attr( attribute_name: impl Into, attribute: Option, ) -> Result { AttributeMap::try_from_attr(attribute_name, attribute) .and_then(SocialProof::try_from) } } #[cfg(test)] mod tests { use crate::constants::USERS_TABLE_SOCIAL_PROOF_ATTRIBUTE_NAME; use super::*; #[test] fn test_valid_ethereum_address() { assert!(is_valid_ethereum_address( "0x1234567890123456789012345678901234567890" ),); assert!(is_valid_ethereum_address( "0xABCDEF123456789012345678901234567890ABCD" )); assert!(is_valid_ethereum_address( "0xabcdef123456789012345678901234567890abcd" )); } + #[allow(clippy::bool_assert_comparison)] #[test] fn test_invalid_ethereum_address() { // Shorter than 42 characters assert_eq!( is_valid_ethereum_address("0x12345678901234567890123456789012345678"), false ); // Longer than 42 characters assert_eq!( is_valid_ethereum_address("0x123456789012345678901234567890123456789012"), false ); // Missing 0x prefix assert_eq!( is_valid_ethereum_address("1234567890123456789012345678901234567890"), false ); // Contains invalid characters assert_eq!( is_valid_ethereum_address("0x1234567890GHIJKL9012345678901234567890"), false ); // Empty string assert_eq!(is_valid_ethereum_address(""), false); } #[test] fn test_social_proof_ddb_format() { let message = "foo"; let signature = "bar"; let social_proof = SocialProof::new(message.to_string(), signature.to_string()); let mut user_item = AttributeMap::from([( USERS_TABLE_SOCIAL_PROOF_ATTRIBUTE_NAME.to_string(), social_proof.into(), )]); let social_proof_from_attr: SocialProof = user_item .take_attr(USERS_TABLE_SOCIAL_PROOF_ATTRIBUTE_NAME) .expect("social proof fetch failed"); assert_eq!(social_proof_from_attr.message, message); assert_eq!(social_proof_from_attr.signature, signature); } } diff --git a/shared/comm-lib/src/lib.rs b/shared/comm-lib/src/lib.rs index e8bd3b2f5..340569854 100644 --- a/shared/comm-lib/src/lib.rs +++ b/shared/comm-lib/src/lib.rs @@ -1,33 +1,36 @@ pub mod auth; pub mod backup; pub mod blob; pub mod constants; #[cfg(feature = "crypto")] pub mod crypto; #[cfg(feature = "aws")] pub mod database; #[cfg(feature = "http")] pub mod http; pub mod shared; pub mod tools; +#[allow(unused_imports)] mod reexports { #[cfg(feature = "blob-client")] pub use {bytes, reqwest}; #[cfg(feature = "http")] pub use {actix_web, http}; #[cfg(feature = "aws")] pub mod aws { pub use aws_config as config; pub use aws_sdk_dynamodb as ddb; pub use aws_sdk_secretsmanager as secretsmanager; // commonly used types pub use config::SdkConfig as AwsConfig; pub use ddb::Client as DynamoDBClient; pub use ddb::Error as DynamoDBError; } } + +#[allow(unused_imports)] pub use reexports::*;