From 68cedff1d9b1ac39fcc25ef41384e7acab03c8c1 Mon Sep 17 00:00:00 2001 From: Alex Date: Thu, 10 Sep 2026 13:48:02 +0200 Subject: [PATCH] [Fix] FS Operations --- iota-connection/src/message_handlers.rs | 76 +++++++---- iota-core/src/main.rs | 18 ++- iota-daemon-lib/src/command_router.rs | 62 +++++---- iota-daemon/src/main.rs | 10 +- iota-storage/src/users/user_manager.rs | 143 ++++++++++---------- iota-storage/src/util/chat_files.rs | 133 ++++++++---------- iota-storage/src/util/communities_util.rs | 34 ++--- iota-storage/src/util/config_util.rs | 124 +++++++++++++++-- iota-storage/src/util/db.rs | 64 +++++---- omikron-connector/src/omikron_connection.rs | 7 +- omikron-connector/src/user_ops.rs | 9 +- web-ui/src/api.rs | 9 +- 12 files changed, 406 insertions(+), 283 deletions(-) diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index 7882c79..d773e46 100644 --- a/iota-connection/src/message_handlers.rs +++ b/iota-connection/src/message_handlers.rs @@ -775,7 +775,7 @@ pub fn handle_get_chat_secret(cv: &CommunicationValue) -> CommunicationValue { DataValue::SignedNumber(record.updated_at as i128), ), Ok(None) => error_response(cv, CommunicationType::ErrorNotSet), - Err(_) => error_response(cv, CommunicationType::ErrorInvalidData), + Err(_) => error_response(cv, CommunicationType::ErrorInternal), } } @@ -803,7 +803,9 @@ pub fn handle_create_app(cv: &CommunicationValue) -> CommunicationValue { if let Some(mut user) = user { if !user.trusted_apps.contains_key(&app_identifier) { user.trusted_apps.insert(app_identifier, app_public_key); - iota_storage::users::user_manager::update_user(user); + if iota_storage::users::user_manager::update_user(user).is_err() { + return error_response(cv, CommunicationType::ErrorInternal); + } } } } @@ -832,7 +834,9 @@ pub fn handle_delete_app(cv: &CommunicationValue) -> CommunicationValue { if let Some(mut user) = user { if user.trusted_apps.contains_key(&app_identifier) { user.trusted_apps.remove(&app_identifier); - iota_storage::users::user_manager::update_user(user); + if iota_storage::users::user_manager::update_user(user).is_err() { + return error_response(cv, CommunicationType::ErrorInternal); + } } } } @@ -954,12 +958,15 @@ pub fn handle_account_state_request(cv: &CommunicationValue) -> CommunicationVal }; let contacts = match chats_util::get_users(user_id) { Ok(contacts) => contacts, - Err(_) => return account_state_error(cv), + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), + }; + let messages = match chat_files::get_all_messages(user_id) { + Ok(messages) => messages, + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), }; - let messages = chat_files::get_all_messages(user_id); let settings = match synced_settings::list(user_id) { Ok(settings) => settings, - Err(_) => return account_state_error(cv), + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), }; let message_values = messages .iter() @@ -967,20 +974,24 @@ pub fn handle_account_state_request(cv: &CommunicationValue) -> CommunicationVal .collect(); let blocked_users = match blocked_users::list(user_id) { Ok(users) => users, - Err(_) => return account_state_error(cv), + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), }; let receipt_policy = match receipt_policy::get(user_id) { Ok(policy) => policy, - Err(_) => return account_state_error(cv), + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), }; let message_storage_policy = match message_storage_policy::get(user_id) { Ok(policy) => policy, - Err(_) => return account_state_error(cv), + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), }; let contact_ids = match current_contact_ids(user_id) { Ok(contact_ids) => contact_ids, Err(_) => return error_response(cv, CommunicationType::ErrorInternal), }; + let communities = match community_values(user_id) { + Ok(communities) => communities, + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), + }; let retention_duration = match message_storage_policy.retention { message_storage_policy::MessageRetention::Forever => None, message_storage_policy::MessageRetention::Duration { duration_ms } => Some(duration_ms), @@ -1028,10 +1039,7 @@ pub fn handle_account_state_request(cv: &CommunicationValue) -> CommunicationVal } }), ) - .add_typed_default( - DataType::Communities, - DataValue::Array(community_values(user_id)), - ) + .add_typed_default(DataType::Communities, DataValue::Array(communities)) .add_typed_default(DataType::UserIds, contact_ids) .add_typed_default(DataType::Calls, DataValue::Array(Vec::new())); if let Some(duration_ms) = retention_duration { @@ -1069,7 +1077,10 @@ pub fn handle_messages_get(cv: &CommunicationValue) -> CommunicationValue { let Some(amount) = data_i64(cv, DataType::Amount).filter(|amount| *amount > 0) else { return error_response(cv, CommunicationType::ErrorInvalidData); }; - let messages = chat_files::get_messages(my_id_i64, partner_id, offset, amount); + let messages = match chat_files::get_messages(my_id_i64, partner_id, offset, amount) { + Ok(messages) => messages, + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), + }; let mut msg_array: Vec = Vec::new(); for m in &messages { msg_array.push(stored_message_value(m, my_id_i64, partner_id)); @@ -1096,12 +1107,12 @@ pub fn handle_message_get(cv: &CommunicationValue) -> CommunicationValue { { Ok(Some((message, offset))) => (message, Some(offset)), Ok(None) => return error_response(cv, CommunicationType::ErrorNotFound), - Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData), + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), }, None => match chat_files::get_message(owner, send_time, None) { Ok(Some(message)) => (message, None), Ok(None) => return error_response(cv, CommunicationType::ErrorNotFound), - Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData), + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), }, }; @@ -1245,12 +1256,16 @@ pub fn handle_add_community(cv: &CommunicationValue) -> CommunicationValue { let Some(position) = cv.get_data(DataType::Position).as_str() else { return error_response(cv, CommunicationType::ErrorInvalidData); }; - CommunitiesUtil::add_community( + if CommunitiesUtil::add_community( sender_id, address.to_string(), title.to_string(), position.to_string(), - ); + ) + .is_err() + { + return error_response(cv, CommunicationType::ErrorInternal); + } CommunicationValue::new(CommunicationType::AddCommunity) .with_request_id(cv) .with_receiver(sender_wire_id(sender_id)) @@ -1261,17 +1276,18 @@ pub fn handle_get_communities(cv: &CommunicationValue) -> CommunicationValue { Ok(sender_id) => sender_id, Err(response) => return response, }; + let communities = match community_values(sender_id) { + Ok(communities) => communities, + Err(_) => return error_response(cv, CommunicationType::ErrorInternal), + }; CommunicationValue::new(CommunicationType::GetCommunities) .with_request_id(cv) .with_receiver(sender_wire_id(sender_id)) - .add_typed_default( - DataType::Communities, - DataValue::Array(community_values(sender_id)), - ) + .add_typed_default(DataType::Communities, DataValue::Array(communities)) } -fn community_values(storage_owner: i64) -> Vec { - CommunitiesUtil::get_communities(storage_owner) +fn community_values(storage_owner: i64) -> Result, StorageError> { + Ok(CommunitiesUtil::get_communities(storage_owner)? .into_iter() .map(|community| { typed_container(vec![ @@ -1283,7 +1299,7 @@ fn community_values(storage_owner: i64) -> Vec { (DataType::Position, DataValue::Str(community.position)), ]) }) - .collect() + .collect()) } pub fn handle_remove_community(cv: &CommunicationValue) -> CommunicationValue { @@ -1321,7 +1337,7 @@ pub fn handle_global_settings_save(cv: &CommunicationValue) -> CommunicationValu }; if settings::save_global(my_id_i64, settings_value).is_err() { - return error_response(cv, CommunicationType::ErrorInvalidData); + return error_response(cv, CommunicationType::ErrorInternal); } let mut response = CommunicationValue::new(CommunicationType::GlobalSettingsSave) @@ -1347,7 +1363,7 @@ pub fn handle_global_settings_load(cv: &CommunicationValue) -> CommunicationValu return error_response(cv, CommunicationType::ErrorInvalidData); }; let Ok(settings_value) = settings::load_global(my_id_i64) else { - return error_response(cv, CommunicationType::ErrorInvalidData); + return error_response(cv, CommunicationType::ErrorInternal); }; let Some(settings_value_str) = settings_value else { let mut response = CommunicationValue::new(CommunicationType::ErrorNotFound) @@ -1469,7 +1485,7 @@ pub fn handle_settings_save( }; if settings::save(my_id_i64, session_id_i64, settings_name, settings_value).is_err() { - return error_response(cv, CommunicationType::ErrorInvalidData); + return error_response(cv, CommunicationType::ErrorInternal); } CommunicationValue::new(CommunicationType::SettingsSave) @@ -1558,7 +1574,7 @@ pub fn handle_settings_load( } let Ok(settings_value) = settings::load(my_id_i64, session_id_i64, settings_name) else { - return error_response(cv, CommunicationType::ErrorInvalidData); + return error_response(cv, CommunicationType::ErrorInternal); }; let Some(settings_value_str) = settings_value else { return CommunicationValue::new(CommunicationType::ErrorNotFound) @@ -1626,7 +1642,7 @@ pub fn handle_settings_list( }; let Ok(settings) = settings::list(my_id_i64, session_id_i64) else { - return error_response(cv, CommunicationType::ErrorInvalidData); + return error_response(cv, CommunicationType::ErrorInternal); }; let settings_json = settings.into_iter().map(DataValue::Str).collect(); CommunicationValue::new(CommunicationType::SettingsList) diff --git a/iota-core/src/main.rs b/iota-core/src/main.rs index 22a942f..d91c9d3 100644 --- a/iota-core/src/main.rs +++ b/iota-core/src/main.rs @@ -96,7 +96,10 @@ async fn main() { logger::startup(); // BASIC CONFIGURATION - iota_storage::util::config_util::load_config(); + if let Err(error) = iota_storage::util::config_util::load_config() { + log!("Cannot load Iota configuration: {}", error); + return; + } // USER MANAGEMENT if let Err(_) = user_manager::load_users_sync() { @@ -108,7 +111,14 @@ async fn main() { let mut sb = "".to_string(); - for up in user_manager::get_users() { + let users = match user_manager::get_users() { + Ok(users) => users, + Err(error) => { + log!("Failed to load users: {}", error); + return; + } + }; + for up in users { sb = sb + "," + &up.user_id.to_string().as_str(); } @@ -180,7 +190,9 @@ async fn main() { sleep(Duration::from_secs(1)).await; } iota_storage::util::config_util::clear_config(); - user_manager::clear(); + if let Err(error) = user_manager::clear() { + log!("Failed to clear users: {}", error); + } // Commhnities have not been implemented yet. /*community_manager::clear();*/ *state.app.lock().unwrap() = AppState::new(); diff --git a/iota-daemon-lib/src/command_router.rs b/iota-daemon-lib/src/command_router.rs index 5047d39..6d0871c 100644 --- a/iota-daemon-lib/src/command_router.rs +++ b/iota-daemon-lib/src/command_router.rs @@ -210,28 +210,31 @@ impl CommandRouter { .collect::>(), Err(_) => return ResponseResult::Error(IpcErrorCode::StorageFailure), }; - let users = user_manager::get_residency() - .into_iter() - .map(|user| { - let user_id = user.user_id; - let profile = user_manager::get_user(user.user_id)?; - Ok(UserSummary { - credential_status: credential_status(&user, profile.as_ref()), - user_id, - username: user.username, - state: match user.state { - user_manager::LocalUserState::Managed => { - iota_ipc::LocalUserState::Managed - } - user_manager::LocalUserState::Released => { - iota_ipc::LocalUserState::Released - } - }, - data_present: user.data_present, - pending_operation: pending.get(&user_id).cloned(), - }) + let users = match user_manager::get_residency() { + Ok(users) => users, + Err(_) => return ResponseResult::Error(IpcErrorCode::StorageFailure), + } + .into_iter() + .map(|user| { + let user_id = user.user_id; + let profile = user_manager::get_user(user.user_id)?; + Ok(UserSummary { + credential_status: credential_status(&user, profile.as_ref()), + user_id, + username: user.username, + state: match user.state { + user_manager::LocalUserState::Managed => { + iota_ipc::LocalUserState::Managed + } + user_manager::LocalUserState::Released => { + iota_ipc::LocalUserState::Released + } + }, + data_present: user.data_present, + pending_operation: pending.get(&user_id).cloned(), }) - .collect::, iota_storage::storage_error::StorageError>>(); + }) + .collect::, iota_storage::storage_error::StorageError>>(); let Ok(users) = users else { return ResponseResult::Error(IpcErrorCode::StorageFailure); }; @@ -628,12 +631,12 @@ impl CommandRouter { Err(_e) => ResponseResult::Error(IpcErrorCode::InvalidRequest), } } - LocalRequest::ReloadConfig => { - config_util::load_config(); - ResponseResult::Ok(ResponsePayload::Acknowledged { + LocalRequest::ReloadConfig => match config_util::load_config() { + Ok(()) => ResponseResult::Ok(ResponsePayload::Acknowledged { message: "Configuration reloaded".into(), - }) - } + }), + Err(_) => ResponseResult::Error(IpcErrorCode::InvalidRequest), + }, LocalRequest::GetOmikronStatus => { let connected = self.services.omikron.is_connected().await; let iota_id = config_util::CONFIG.load().iota_id; @@ -743,7 +746,12 @@ impl CommandRouter { return ResponseResult::Ok(ResponsePayload::Communities(Vec::new())); }; let stored = - iota_storage::util::communities_util::CommunitiesUtil::get_communities(iota_id); + match iota_storage::util::communities_util::CommunitiesUtil::get_communities( + iota_id, + ) { + Ok(stored) => stored, + Err(_) => return ResponseResult::Error(IpcErrorCode::StorageFailure), + }; let summaries: Vec = stored .into_iter() .map(|c| CommunitySummary { diff --git a/iota-daemon/src/main.rs b/iota-daemon/src/main.rs index af5d269..fb1d45c 100644 --- a/iota-daemon/src/main.rs +++ b/iota-daemon/src/main.rs @@ -90,7 +90,10 @@ async fn main() -> ExitCode { } iota_util::file_util::configure_storage_directory(paths.storage_dir.clone()); iota_storage::util::config_util::configure_config_path(paths.config_file.clone()); - iota_storage::util::config_util::load_config_from(&paths.config_file); + if let Err(error) = iota_storage::util::config_util::load_config_from(&paths.config_file) { + eprintln!("Cannot load Iota configuration: {error}"); + return ExitCode::FAILURE; + } omikron_connector::omikron_connection::configure_identity_path(paths.keyring_file()); match paths.scope { iota_paths::Scope::User => logger::startup_with_log_dir(Some(paths.log_dir.clone())), @@ -332,10 +335,7 @@ async fn main() -> ExitCode { iota_storage::util::config_util::WebMode::Loopback => web_server::WebMode::Loopback, iota_storage::util::config_util::WebMode::Network => web_server::WebMode::Network, }, - bind: web - .bind - .parse() - .unwrap_or(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)), + bind: web.bind.parse().expect("web.bind was validated at startup"), port: web.port, asset_dir: resolve_config_path(&paths.config_file, &web.asset_dir, &paths.asset_dir), tls: web diff --git a/iota-storage/src/users/user_manager.rs b/iota-storage/src/users/user_manager.rs index 5fedca7..8cf216f 100644 --- a/iota-storage/src/users/user_manager.rs +++ b/iota-storage/src/users/user_manager.rs @@ -33,10 +33,8 @@ fn now_millis() -> i64 { .as_millis() as i64 } -pub fn add_user(user: UserProfile) { - if let Err(e) = try_add_user(user) { - eprintln!("Failed to add_user: {}", e); - } +pub fn add_user(user: UserProfile) -> Result<(), crate::storage_error::StorageError> { + try_add_user(user) } pub fn try_add_user(user: UserProfile) -> Result<(), crate::storage_error::StorageError> { @@ -89,12 +87,14 @@ pub fn try_add_user_with_credential_origin( }) } -pub fn update_user(user: UserProfile) { - add_user(user); +pub fn update_user(user: UserProfile) -> Result<(), crate::storage_error::StorageError> { + try_add_user(user) } -pub fn get_user_by_username(username: &str) -> Option { - match db::with_db(|conn| { +pub fn get_user_by_username( + username: &str, +) -> Result, crate::storage_error::StorageError> { + let user = db::with_db(|conn| { match conn.query_row( "SELECT user_id, username, public_key, private_key_hash, reset_token, created_at, display_name FROM users WHERE username = ?1 LIMIT 1", params![username], @@ -116,13 +116,12 @@ pub fn get_user_by_username(username: &str) -> Option { Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None), Err(e) => Err(e.into()), } - }) { - Ok(opt) => opt, - Err(e) => { - eprintln!("Error querying user by username: {}", e); - None - } - } + })?; + user.map(|mut user| { + user.trusted_apps = load_trusted_apps(user.user_id)?; + Ok(user) + }) + .transpose() } pub fn get_user(user_id: i64) -> Result, crate::storage_error::StorageError> { @@ -156,8 +155,8 @@ pub fn get_user(user_id: i64) -> Result, crate::storage_erro .transpose() } -pub fn get_users() -> Vec { - match db::with_db(|conn| { +pub fn get_users() -> Result, crate::storage_error::StorageError> { + db::with_db(|conn| { let mut stmt = conn.prepare( r#" SELECT user_id, username, public_key, private_key_hash, reset_token, created_at, display_name @@ -189,41 +188,36 @@ pub fn get_users() -> Vec { let mut out = Vec::new(); for row in rows { - match row { - Ok(mut user) => { - user.trusted_apps = load_trusted_apps(user.user_id)?; - out.push(user); - } - Err(e) => eprintln!("Failed to read user row: {}", e), - } + let mut user = row?; + user.trusted_apps = load_trusted_apps_from(conn, user.user_id)?; + out.push(user); } Ok(out) - }) { - Ok(v) => v, - Err(e) => { - eprintln!("Failed to query users: {}", e); - Vec::new() - } - } + }) } fn load_trusted_apps( user_id: i64, ) -> Result, crate::storage_error::StorageError> { - db::with_db(|conn| { - let mut stmt = - conn.prepare("SELECT app_id, app_secret FROM trusted_apps WHERE user_id = ?1")?; - let rows = stmt.query_map(params![user_id], |r| { - Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)) - })?; + db::with_db(|conn| load_trusted_apps_from(conn, user_id)) +} - let mut map = std::collections::HashMap::new(); - for row in rows { - let (key, value) = row?; - map.insert(key, value); - } - Ok(map) - }) +fn load_trusted_apps_from( + conn: &rusqlite::Connection, + user_id: i64, +) -> Result, crate::storage_error::StorageError> { + let mut stmt = + conn.prepare("SELECT app_id, app_secret FROM trusted_apps WHERE user_id = ?1")?; + let rows = stmt.query_map(params![user_id], |r| { + Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)) + })?; + + let mut map = std::collections::HashMap::new(); + for row in rows { + let (key, value) = row?; + map.insert(key, value); + } + Ok(map) } pub fn revoke_trusted_app( @@ -249,17 +243,15 @@ pub fn revoke_all_trusted_apps(user_id: i64) -> Result Result<(), crate::storage_error::StorageError> { + db::with_immediate_transaction(|conn| { conn.execute( "DELETE FROM trusted_apps WHERE user_id = ?1", params![user_id], )?; conn.execute("DELETE FROM users WHERE user_id = ?1", params![user_id])?; Ok(()) - }) { - eprintln!("Failed to remove_user: {}", e); - } + }) } /// Remove only local management authority. Hosted content is intentionally @@ -281,15 +273,12 @@ pub fn finalize_local_release( user_id: i64, username_hint: Option<&str>, ) -> Result<(), crate::storage_error::StorageError> { - let username = get_user(user_id)? - .map(|user| user.username) - .or_else(|| { - get_residency_by_id(user_id) - .ok() - .flatten() - .map(|residency| residency.username) - }) - .or_else(|| username_hint.map(str::to_owned)); + let username = match get_user(user_id)? { + Some(user) => Some(user.username), + None => get_residency_by_id(user_id)? + .map(|residency| residency.username) + .or_else(|| username_hint.map(str::to_owned)), + }; let Some(username) = username else { return Err(crate::storage_error::StorageError::Other( "user residency was not found".into(), @@ -553,12 +542,13 @@ pub fn purge_user_data(user_id: i64) -> Result<(), crate::storage_error::Storage /// Complete local erasure is idempotent and is the target for a durable /// Omega-hosted erasure request after account deletion. pub fn erase_user_locally(user_id: i64) -> Result<(), crate::storage_error::StorageError> { - let username = get_user(user_id)?.map(|user| user.username).or_else(|| { - get_residency() + let username = match get_user(user_id)? { + Some(user) => Some(user.username), + None => get_residency()? .into_iter() .find(|entry| entry.user_id == user_id) - .map(|entry| entry.username) - }); + .map(|entry| entry.username), + }; purge_user_data(user_id)?; db::with_db(|conn| { conn.execute( @@ -576,20 +566,29 @@ pub fn erase_user_locally(user_id: i64) -> Result<(), crate::storage_error::Stor .map_err(|error| crate::storage_error::StorageError::Other(error.to_string())) } -pub fn get_residency() -> Vec { +pub fn get_residency() -> Result, crate::storage_error::StorageError> { db::with_db(|conn| { let mut stmt = conn.prepare("SELECT user_id, username, lifecycle_state, data_state, credential_origin FROM user_residency ORDER BY username")?; let rows = stmt.query_map([], |row| { let lifecycle: String = row.get(2)?; Ok(UserResidency { - user_id: row.get(0)?, username: row.get(1)?, - state: if lifecycle == "managed" { LocalUserState::Managed } else { LocalUserState::Released }, + user_id: row.get(0)?, + username: row.get(1)?, + state: if lifecycle == "managed" { + LocalUserState::Managed + } else { + LocalUserState::Released + }, data_present: row.get::<_, String>(3)? == "present", - credential_origin: if row.get::<_, String>(4)? == "external" { CredentialOrigin::External } else { CredentialOrigin::Local }, + credential_origin: if row.get::<_, String>(4)? == "external" { + CredentialOrigin::External + } else { + CredentialOrigin::Local + }, }) })?; rows.collect::, _>>().map_err(Into::into) - }).unwrap_or_default() + }) } pub fn get_residency_by_id( @@ -622,15 +621,13 @@ pub fn get_residency_by_id( }) } -pub fn clear() { - if let Err(e) = db::with_db(|conn| { +pub fn clear() -> Result<(), crate::storage_error::StorageError> { + db::with_immediate_transaction(|conn| { conn.execute_batch( "DELETE FROM trusted_apps; DELETE FROM users; DELETE FROM user_residency;", )?; Ok(()) - }) { - eprintln!("Failed to clear users: {}", e); - } + }) } pub fn save_users() { @@ -648,7 +645,7 @@ pub fn load_users_sync() -> std::io::Result<()> { if let json::JsonValue::Array(arr) = parsed { for j in arr.iter() { if let Some(up) = UserProfile::from_json(j) { - add_user(up); + try_add_user(up).map_err(|error| std::io::Error::other(error.to_string()))?; } } } diff --git a/iota-storage/src/util/chat_files.rs b/iota-storage/src/util/chat_files.rs index 3da9699..f4be254 100644 --- a/iota-storage/src/util/chat_files.rs +++ b/iota-storage/src/util/chat_files.rs @@ -2,7 +2,6 @@ use crate::storage_error::StorageError; use crate::util::db; use crate::util::message_storage_policy::{self, MessageRetention}; use crate::util::sync::{self, EntityType, Operation}; -use iota_logger::log; use rusqlite::{OptionalExtension, Transaction, params}; pub const MAX_UNIQUE_REACTIONS_PER_MESSAGE: usize = 10; @@ -907,9 +906,9 @@ pub fn change_message_state( fn load_reactions( conn: &rusqlite::Connection, msg_ids: &[i64], -) -> std::collections::HashMap> { +) -> Result>, StorageError> { if msg_ids.is_empty() { - return std::collections::HashMap::new(); + return Ok(std::collections::HashMap::new()); } let placeholders: Vec = msg_ids @@ -924,34 +923,33 @@ fn load_reactions( let mut map: std::collections::HashMap> = std::collections::HashMap::new(); - if let Ok(mut stmt) = conn.prepare(&query) { - let params: Vec<&dyn rusqlite::types::ToSql> = msg_ids + let mut stmt = conn.prepare(&query)?; + let params: Vec<&dyn rusqlite::types::ToSql> = msg_ids + .iter() + .map(|id| id as &dyn rusqlite::types::ToSql) + .collect(); + let rows = stmt.query_map(params.as_slice(), |row| { + Ok(( + row.get::<_, i64>(0)?, + StoredReaction { + reaction: row.get(1)?, + user_id: row.get(2)?, + }, + )) + })?; + for row in rows { + let row = row?; + let reactions = map.entry(row.0).or_default(); + if reactions .iter() - .map(|id| id as &dyn rusqlite::types::ToSql) - .collect(); - if let Ok(rows) = stmt.query_map(params.as_slice(), |row| { - Ok(( - row.get::<_, i64>(0)?, - StoredReaction { - reaction: row.get(1)?, - user_id: row.get(2)?, - }, - )) - }) { - for row in rows.flatten() { - let reactions = map.entry(row.0).or_default(); - if reactions - .iter() - .any(|stored: &StoredReaction| stored.reaction == row.1.reaction) - { - reactions.push(row.1); - } else if reactions.len() < MAX_UNIQUE_REACTIONS_PER_MESSAGE { - reactions.push(row.1); - } - } + .any(|stored: &StoredReaction| stored.reaction == row.1.reaction) + { + reactions.push(row.1); + } else if reactions.len() < MAX_UNIQUE_REACTIONS_PER_MESSAGE { + reactions.push(row.1); } } - map + Ok(map) } pub fn get_messages( @@ -959,12 +957,12 @@ pub fn get_messages( external_user: i64, loaded_messages: i64, amount: i64, -) -> Vec { +) -> Result, StorageError> { if amount <= 0 || loaded_messages < 0 { - return Vec::new(); + return Ok(Vec::new()); } - match db::with_db(|conn| { + db::with_db(|conn| { let mut stmt = conn.prepare( r#" SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, @@ -1000,10 +998,10 @@ pub fn get_messages( content: row.get(13)?, sent_by_self: row.get::<_, i64>(14)? != 0, message_state: row.get(15)?, - height: row.get(16).unwrap_or(0), - key_version: row.get(17).unwrap_or(1), - reply_to: row.get(18).ok().flatten(), - edited: row.get::<_, i64>(19).unwrap_or(0) > 0, + height: row.get(16)?, + key_version: row.get(17)?, + reply_to: row.get(18)?, + edited: row.get::<_, i64>(19)? > 0, reactions: Vec::new(), }) }, @@ -1011,26 +1009,17 @@ pub fn get_messages( let mut out = Vec::new(); for row in rows { - match row { - Ok(msg) => out.push(msg), - Err(e) => log!("Failed to read row from sqlite: {}", e), - } + out.push(row?); } let msg_ids: Vec = out.iter().map(|m| m.id).collect(); - let reaction_map = load_reactions(conn, &msg_ids); + let reaction_map = load_reactions(conn, &msg_ids)?; for msg in &mut out { msg.reactions = reaction_map.get(&msg.id).cloned().unwrap_or_default(); } Ok(out) - }) { - Ok(v) => v, - Err(e) => { - log!("Failed to query messages: {}", e); - Vec::new() - } - } + }) } pub fn get_message( @@ -1074,10 +1063,10 @@ pub fn get_message( content: row.get(13)?, sent_by_self: row.get::<_, i64>(14)? != 0, message_state: row.get(15)?, - height: row.get(16).unwrap_or(0), - key_version: row.get(17).unwrap_or(1), - reply_to: row.get(18).ok().flatten(), - edited: row.get::<_, i64>(19).unwrap_or(0) > 0, + height: row.get(16)?, + key_version: row.get(17)?, + reply_to: row.get(18)?, + edited: row.get::<_, i64>(19)? > 0, external_user: row.get(20)?, reactions: Vec::new(), }) @@ -1099,7 +1088,7 @@ pub fn get_message( } let mut message = messages.into_iter().next().expect("checked non-empty"); - let reaction_map = load_reactions(conn, &[message.id]); + let reaction_map = load_reactions(conn, &[message.id])?; message.reactions = reaction_map.get(&message.id).cloned().unwrap_or_default(); Ok(Some(message)) }) @@ -1142,14 +1131,17 @@ pub fn get_message_with_offset( Ok(Some((message, offset))) } -pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec { +pub fn get_messages_by_ids( + storage_owner: i64, + ids: &[i64], +) -> Result, StorageError> { if ids.is_empty() { - return Vec::new(); + return Ok(Vec::new()); } let wanted: std::collections::HashSet = ids.iter().copied().collect(); // A journal id uniquely identifies a row. Load all messages for this owner and retain only // those ids; this keeps reaction hydration identical to normal message loading. - match db::with_db(|conn| { + db::with_db(|conn| { let mut stmt = conn.prepare("SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, origin_iota_received_at, destination_iota_received_at, client_received_at, client_received_recorded_at, read_at, read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, key_version, reply_to, edited_count, external_user FROM messages WHERE storage_owner = ?1 AND deleted_by_external = 0 AND history_deleted = 0")?; let rows = stmt.query_map([storage_owner], |row| { let external_user: i64 = row.get(20)?; @@ -1171,10 +1163,10 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec(14)? != 0, message_state: row.get(15)?, - height: row.get(16).unwrap_or(0), - key_version: row.get(17).unwrap_or(1), - reply_to: row.get(18).ok().flatten(), - edited: row.get::<_, i64>(19).unwrap_or(0) > 0, + height: row.get(16)?, + key_version: row.get(17)?, + reply_to: row.get(18)?, + edited: row.get::<_, i64>(19)? > 0, reactions: Vec::new(), }) })?; @@ -1185,35 +1177,24 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec>()); + let reaction_map = + load_reactions(conn, &messages.iter().map(|m| m.id).collect::>())?; for message in &mut messages { message.reactions = reaction_map.get(&message.id).cloned().unwrap_or_default(); } Ok(messages) - }) { - Ok(messages) => messages, - Err(e) => { - log!("Failed to query messages by id: {}", e); - Vec::new() - } - } + }) } -pub fn get_all_messages(storage_owner: i64) -> Vec { - let ids = match db::with_db(|conn| { +pub fn get_all_messages(storage_owner: i64) -> Result, StorageError> { + let ids = db::with_db(|conn| { let mut stmt = conn.prepare( "SELECT id FROM messages WHERE storage_owner = ?1 AND deleted_by_external = 0 AND history_deleted = 0", )?; Ok(stmt .query_map([storage_owner], |row| row.get::<_, i64>(0))? .collect::, _>>()?) - }) { - Ok(ids) => ids, - Err(e) => { - log!("Failed to query all messages: {}", e); - return Vec::new(); - } - }; + })?; get_messages_by_ids(storage_owner, &ids) } diff --git a/iota-storage/src/util/communities_util.rs b/iota-storage/src/util/communities_util.rs index 3b3be8e..de1ef14 100644 --- a/iota-storage/src/util/communities_util.rs +++ b/iota-storage/src/util/communities_util.rs @@ -23,8 +23,13 @@ impl CommunitiesUtil { }) } - pub fn add_community(storage_owner: i64, address: String, title: String, position: String) { - if let Err(e) = db::with_db(|conn| { + pub fn add_community( + storage_owner: i64, + address: String, + title: String, + position: String, + ) -> Result<(), StorageError> { + db::with_db(|conn| { conn.execute( r#" INSERT INTO communities (storage_owner, address, title, position) @@ -36,9 +41,7 @@ impl CommunitiesUtil { params![storage_owner, address, title, position], )?; Ok(()) - }) { - eprintln!("Failed to add_community: {}", e); - } + }) } pub fn remove_community( @@ -59,8 +62,8 @@ impl CommunitiesUtil { }) } - pub fn get_communities(storage_owner: i64) -> Vec { - match db::with_db(|conn| { + pub fn get_communities(storage_owner: i64) -> Result, StorageError> { + db::with_db(|conn| { let mut stmt = conn.prepare( r#" SELECT address, title, position @@ -77,20 +80,7 @@ impl CommunitiesUtil { }) })?; - let mut out = Vec::new(); - for row in rows { - match row { - Ok(community) => out.push(community), - Err(e) => eprintln!("Failed to read community row: {}", e), - } - } - Ok(out) - }) { - Ok(v) => v, - Err(e) => { - eprintln!("Failed to query communities in get_communities: {}", e); - Vec::new() - } - } + rows.collect::, _>>().map_err(Into::into) + }) } } diff --git a/iota-storage/src/util/config_util.rs b/iota-storage/src/util/config_util.rs index ef3e827..bf66f0b 100644 --- a/iota-storage/src/util/config_util.rs +++ b/iota-storage/src/util/config_util.rs @@ -5,11 +5,37 @@ use std::fs; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::OnceLock; +use thiserror::Error; pub static CONFIG: Lazy> = Lazy::new(|| ArcSwap::new(Arc::new(IotaConfig::default()))); +#[derive(Debug, Error)] +pub enum ConfigError { + #[error("cannot read {path}: {source}")] + Read { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("cannot parse {path}: {source}")] + Parse { + path: PathBuf, + #[source] + source: serde_yaml::Error, + }, + #[error("invalid web.bind {bind:?}: {source}")] + InvalidWebBind { + bind: String, + #[source] + source: std::net::AddrParseError, + }, + #[error("max_ipc_clients must be greater than zero")] + InvalidMaxIpcClients, +} + #[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct IotaConfig { #[serde(skip_serializing_if = "Option::is_none")] pub iota_id: Option, @@ -49,6 +75,7 @@ impl Default for WebMode { } #[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct WebSettings { #[serde(default)] pub mode: WebMode, @@ -113,30 +140,54 @@ impl Default for IotaConfig { } } -pub fn load_config() { - load_config_from(&default_config_path()); +pub fn load_config() -> Result<(), ConfigError> { + load_config_from(&default_config_path()) } /// Loading is intentionally side-effect free: a missing configuration means /// documented defaults, not a newly-created file. -pub fn load_config_from(path: &Path) { +pub fn load_config_from(path: &Path) -> Result<(), ConfigError> { let s = match fs::read_to_string(path) { Ok(contents) => contents, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => return, - Err(error) => { - eprintln!("Failed to read {}: {error}", path.display()); - return; + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + CONFIG.store(Arc::new(IotaConfig::default())); + return Ok(()); + } + Err(source) => { + return Err(ConfigError::Read { + path: path.into(), + source, + }); } }; - match serde_yaml::from_str::(&s) { - Ok(parsed) => { - CONFIG.store(Arc::new(parsed)); - } - Err(e) => { - eprintln!("Failed to parse {}: {e}", path.display()); - } + let parsed = parse_config(path, &s)?; + CONFIG.store(Arc::new(parsed)); + Ok(()) +} + +fn parse_config(path: &Path, yaml: &str) -> Result { + let parsed = serde_yaml::from_str::(yaml).map_err(|source| ConfigError::Parse { + path: path.into(), + source, + })?; + validate_config(&parsed)?; + Ok(parsed) +} + +pub fn validate_config(config: &IotaConfig) -> Result<(), ConfigError> { + config + .web + .bind + .parse::() + .map_err(|source| ConfigError::InvalidWebBind { + bind: config.web.bind.clone(), + source, + })?; + if config.max_ipc_clients == 0 { + return Err(ConfigError::InvalidMaxIpcClients); } + Ok(()) } pub fn clear_config() { @@ -232,6 +283,9 @@ pub fn modify_config_value(key: &str, value: &str) -> Result<(), &'static str> { Ok(()) } "web.bind" => { + value + .parse::() + .map_err(|_| "invalid web.bind")?; let bind = value.to_string(); modify_config(|cfg| cfg.web.bind = bind); Ok(()) @@ -244,3 +298,45 @@ static CONFIG_PATH: OnceLock = OnceLock::new(); pub fn configure_config_path(path: PathBuf) { let _ = CONFIG_PATH.set(path); } + +#[cfg(test)] +mod tests { + use super::{ConfigError, IotaConfig, parse_config, validate_config}; + use std::path::Path; + + #[test] + fn malformed_yaml_is_rejected() { + assert!(matches!( + parse_config(Path::new("config.yaml"), "web: ["), + Err(ConfigError::Parse { .. }) + )); + } + + #[test] + fn unknown_explicit_fields_are_rejected() { + assert!(matches!( + parse_config(Path::new("config.yaml"), "unexpected: true\n"), + Err(ConfigError::Parse { .. }) + )); + } + + #[test] + fn invalid_explicit_bind_is_rejected() { + let mut config = IotaConfig::default(); + config.web.bind = "localhost:1984".into(); + assert!(matches!( + validate_config(&config), + Err(ConfigError::InvalidWebBind { .. }) + )); + } + + #[test] + fn zero_ipc_capacity_is_rejected() { + let mut config = IotaConfig::default(); + config.max_ipc_clients = 0; + assert!(matches!( + validate_config(&config), + Err(ConfigError::InvalidMaxIpcClients) + )); + } +} diff --git a/iota-storage/src/util/db.rs b/iota-storage/src/util/db.rs index 0b7553a..3cedb80 100644 --- a/iota-storage/src/util/db.rs +++ b/iota-storage/src/util/db.rs @@ -51,19 +51,42 @@ pub fn with_db(f: F) -> Result where F: FnOnce(&Connection) -> Result, { - let conn = POOL.get().map_err(|e| StorageError::Pool(e.to_string()))?; - f(&conn) + blocking_region(|| { + let conn = POOL.get().map_err(|e| StorageError::Pool(e.to_string()))?; + f(&conn) + }) } pub fn with_immediate_transaction(f: F) -> Result where F: FnOnce(&Transaction<'_>) -> Result, { - let mut conn = POOL.get().map_err(|e| StorageError::Pool(e.to_string()))?; - let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; - let value = f(&tx)?; - tx.commit()?; - Ok(value) + blocking_region(|| { + let mut conn = POOL.get().map_err(|e| StorageError::Pool(e.to_string()))?; + let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; + let value = f(&tx)?; + tx.commit()?; + Ok(value) + }) +} + +/* SQLite remains synchronous, but daemon transports run on Tokio. Entering a + * blocking region lets Tokio replace the worker that is waiting on the bounded + * eight-connection pool, so database latency cannot stall unrelated I/O. A + * current-thread runtime (used by unit tests and small tools) has no worker to + * hand off and therefore executes directly. */ +fn blocking_region(f: impl FnOnce() -> T) -> T { + match tokio::runtime::Handle::try_current() { + Ok(handle) + if matches!( + handle.runtime_flavor(), + tokio::runtime::RuntimeFlavor::MultiThread + ) => + { + tokio::task::block_in_place(f) + } + _ => f(), + } } /* Verify the persistent database before the pool is initialized. A corrupt @@ -156,9 +179,7 @@ fn add_table_column_if_missing( } fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { - let current_version: i64 = conn - .pragma_query_value(None, "user_version", |r| r.get(0)) - .unwrap_or(0); + let current_version: i64 = conn.pragma_query_value(None, "user_version", |r| r.get(0))?; if current_version < 1 { conn.execute_batch( @@ -272,13 +293,11 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { )?; } - let messages_exist: bool = conn - .query_row( - "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages')", - [], - |row| row.get(0), - ) - .unwrap_or(false); + let messages_exist: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages')", + [], + |row| row.get(0), + )?; if messages_exist { add_column_if_missing( conn, @@ -926,19 +945,12 @@ pub fn with_conn(shared: &Arc>, f: F) -> Resu where F: FnOnce(&Connection) -> Result, { - if tokio::runtime::Handle::try_current().is_ok() { - tokio::task::block_in_place(|| { - let guard = shared - .lock() - .map_err(|e| format!("DB mutex poisoned: {:?}", e))?; - f(&*guard).map_err(|e| e.to_string()) - }) - } else { + blocking_region(|| { let guard = shared .lock() .map_err(|e| format!("DB mutex poisoned: {:?}", e))?; f(&*guard).map_err(|e| e.to_string()) - } + }) } /// Legacy - kept for e2ee_storage which uses its own DB. diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 3ce2426..9867fa9 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -763,8 +763,10 @@ impl OmikronConnection { continue; } let Some(msg_id) = cv.id() else { + let permit = self.handler_semaphore.clone().acquire_owned().await; let self_clone = self.clone(); tokio::spawn(async move { + let _permit = permit; self_clone.handle_message_impl(cv).await; }); continue; @@ -2629,10 +2631,7 @@ impl OmikronConnection { .ok() .and_then(|id| i64::try_from(id).ok()); if response.is_type(CommunicationType::AccountStateSnapshot) { - log!( - "AccountStateSnapshot user={:?} stage=generated", - user_id - ); + log!("AccountStateSnapshot user={:?} stage=generated", user_id); if let Some(user_id) = user_id && let Err(error) = relay_queue::pause_client_deliveries(user_id) { diff --git a/omikron-connector/src/user_ops.rs b/omikron-connector/src/user_ops.rs index d81b5ef..e3f9b4a 100644 --- a/omikron-connector/src/user_ops.rs +++ b/omikron-connector/src/user_ops.rs @@ -399,7 +399,14 @@ pub async fn reconcile_managed_users(connection: &dyn OmikronClient) { _ => {} } } - for user in iota_storage::users::user_manager::get_users() { + let users = match iota_storage::users::user_manager::get_users() { + Ok(users) => users, + Err(error) => { + log!("Could not load local users for reconciliation: {}", error); + return; + } + }; + for user in users { let request = CommunicationValue::new(CommunicationType::GetUserData).add_typed_default( DataType::UserId, DataValue::SignedNumber(user.user_id.into()), diff --git a/web-ui/src/api.rs b/web-ui/src/api.rs index 4359c17..c2c52de 100755 --- a/web-ui/src/api.rs +++ b/web-ui/src/api.rs @@ -107,7 +107,10 @@ async fn users_get(req: HttpRequest, ssl: web::Data) -> impl Responder { return forbidden(); } - let users = iota_storage::users::user_manager::get_users(); + let users = match iota_storage::users::user_manager::get_users() { + Ok(users) => users, + Err(_) => return HttpResponse::InternalServerError().finish(), + }; let list: Vec<_> = users .into_iter() @@ -134,7 +137,9 @@ async fn users_remove( Err(response) => return response, }; - iota_storage::users::user_manager::remove_user(uuid); + if iota_storage::users::user_manager::remove_user(uuid).is_err() { + return HttpResponse::InternalServerError().finish(); + } iota_storage::users::user_manager::save_users(); success()