From afc1832fb77e159e01f4ba06f88d01ce5b772b14 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Sat, 29 Aug 2026 12:49:10 +0200 Subject: [PATCH] [Fix] Connectivity --- flake.nix | 9 - iota-connection/src/message_handlers.rs | 449 ++++++++++++- iota-connection/src/relay.rs | 22 + iota-storage/src/users/user_manager.rs | 4 + iota-storage/src/util/chats_util.rs | 11 + iota-storage/src/util/communities_util.rs | 32 +- iota-storage/src/util/db.rs | 48 +- iota-storage/src/util/mod.rs | 1 + iota-storage/src/util/sync.rs | 188 +++++- iota-storage/src/util/synced_settings.rs | 659 ++++++++++++++++++++ mtp-type-maps | 2 +- omikron-connector/Cargo.toml | 1 + omikron-connector/src/omikron_connection.rs | 223 +++---- systemd/iota-daemon.service | 1 - 14 files changed, 1462 insertions(+), 188 deletions(-) create mode 100644 iota-storage/src/util/synced_settings.rs diff --git a/flake.nix b/flake.nix index 5d64dc6..e2ca8a1 100644 --- a/flake.nix +++ b/flake.nix @@ -140,12 +140,6 @@ description = "Environment files to load for the Iota service."; }; - identitySecretFile = lib.mkOption { - type = lib.types.nullOr lib.types.path; - default = null; - description = "Owner-readable file containing the passphrase for the protected Iota identity."; - }; - openFirewall = lib.mkOption { type = lib.types.bool; default = true; @@ -267,9 +261,6 @@ } // lib.optionalAttrs (cfg.environmentFiles != []) { EnvironmentFile = cfg.environmentFiles; - } - // lib.optionalAttrs (cfg.identitySecretFile != null) { - LoadCredential = "iota-identity:${cfg.identitySecretFile}"; }; }; diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index d692be4..ae7e392 100644 --- a/iota-connection/src/message_handlers.rs +++ b/iota-connection/src/message_handlers.rs @@ -1,9 +1,10 @@ use crate::message_common::*; use iota_storage::util::chat_files::{self, MessageState}; -use iota_storage::util::chats_util::{self, get_user, mod_user}; +use iota_storage::util::chats_util::{self, get_user, has_user, mod_user}; use iota_storage::util::communities_util::CommunitiesUtil; use iota_storage::util::e2ee_storage::{self, ChatSecretQuery}; use iota_storage::util::settings; +use iota_storage::util::synced_settings::{self, SettingScope, SyncedSetting}; use mtp::codec::{ CommunicationType, CommunicationValue, DataType, DataValue, TypeMap, VerifiedRelayContent, }; @@ -17,6 +18,18 @@ pub struct MessageMutation { pub send_time: i64, } +#[derive(Debug)] +pub struct SettingMutation { + pub response: CommunicationValue, + pub changed: Option, +} + +struct SettingLocator { + scope: SettingScope, + scope_key: String, + name: String, +} + fn required_sender_id(cv: &CommunicationValue) -> Result { let sender = cv .require_sender() @@ -435,6 +448,29 @@ fn stored_message_value( typed_container(stored_message_fields(message, storage_owner, partner_id)) } +fn synced_setting_value(setting: &SyncedSetting) -> DataValue { + typed_container(vec![ + ( + DataType::SettingId, + DataValue::SignedNumber(setting.id.into()), + ), + ( + DataType::SettingScope, + DataValue::Str(setting.scope.as_str().to_string()), + ), + ( + DataType::SettingTarget, + DataValue::Str(setting.scope_key.clone()), + ), + (DataType::SettingsName, DataValue::Str(setting.name.clone())), + (DataType::Payload, DataValue::Str(setting.payload.clone())), + ( + DataType::VersionNumber, + DataValue::SignedNumber(setting.revision.into()), + ), + ]) +} + pub fn handle_get_chat_secret(cv: &CommunicationValue) -> CommunicationValue { let Some(user_id) = data_string(cv, DataType::UserId) else { return error_response(cv, CommunicationType::ErrorInvalidData); @@ -671,32 +707,56 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { || !known_session || reported_version > head || schema != CACHE_SCHEMA_VERSION; - let (contacts, messages, deleted_messages, deleted_contacts, mode) = if full { - ( - chats_util::get_users(user_id), - chat_files::get_all_messages(user_id), - Vec::new(), - Vec::new(), - "full", - ) - } else { - match sync::delta(user_id, reported_version, head) { - Ok(delta) => ( - chats_util::get_users_by_ids(user_id, &delta.contact_upserts), - chat_files::get_messages_by_ids(user_id, &delta.message_upserts), - delta.deleted_message_ids, - delta.deleted_contact_ids, - "delta", - ), - Err(_) => ( + let (contacts, messages, settings, deleted_messages, deleted_contacts, deleted_settings, mode) = + if full { + let settings = match synced_settings::list(user_id) { + Ok(settings) => settings, + Err(_) => return sync_error(cv), + }; + ( chats_util::get_users(user_id), chat_files::get_all_messages(user_id), + settings, + Vec::new(), Vec::new(), Vec::new(), "full", - ), - } - }; + ) + } else { + match sync::delta(user_id, reported_version, head) { + Ok(delta) => { + let settings = + match synced_settings::list_by_ids(user_id, &delta.setting_upserts) { + Ok(settings) => settings, + Err(_) => return sync_error(cv), + }; + ( + chats_util::get_users_by_ids(user_id, &delta.contact_upserts), + chat_files::get_messages_by_ids(user_id, &delta.message_upserts), + settings, + delta.deleted_message_ids, + delta.deleted_contact_ids, + delta.deleted_setting_ids, + "delta", + ) + } + Err(_) => { + let settings = match synced_settings::list(user_id) { + Ok(settings) => settings, + Err(_) => return sync_error(cv), + }; + ( + chats_util::get_users(user_id), + chat_files::get_all_messages(user_id), + settings, + Vec::new(), + Vec::new(), + Vec::new(), + "full", + ) + } + } + }; let message_values = messages .iter() .map(|message| stored_message_value(message, user_id, message.external_user)) @@ -727,6 +787,10 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { ), ) .add_typed_default(DataType::Messages, DataValue::Array(message_values)) + .add_typed_default( + DataType::Settings, + DataValue::Array(settings.iter().map(synced_setting_value).collect()), + ) .add_typed_default( DataType::Communities, DataValue::Array(community_values(user_id)), @@ -749,6 +813,15 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { .collect(), ), ) + .add_typed_default( + DataType::DeletedSettingIds, + DataValue::Array( + deleted_settings + .into_iter() + .map(|id| DataValue::SignedNumber(id as i128)) + .collect(), + ), + ) .add_typed_default(DataType::UserIds, current_contact_ids(user_id)) .add_typed_default(DataType::Calls, DataValue::Array(Vec::new())) } @@ -995,7 +1068,9 @@ pub fn handle_remove_community(cv: &CommunicationValue) -> CommunicationValue { let Some(address) = cv.get_data(DataType::CommunityAddress).as_str() else { return error_response(cv, CommunicationType::ErrorInvalidData); }; - CommunitiesUtil::remove_community(sender_id, address.to_string()); + if CommunitiesUtil::remove_community(sender_id, address.to_string()).is_err() { + return error_response(cv, CommunicationType::ErrorInternal); + } CommunicationValue::new(CommunicationType::RemoveCommunity) .with_request_id(cv) .with_receiver(sender_wire_id(sender_id)) @@ -1337,3 +1412,331 @@ pub fn handle_settings_list( DataValue::SignedNumber(session_id as i128), ) } + +fn setting_response( + cv: &CommunicationValue, + response_type: CommunicationType, + setting: &SyncedSetting, +) -> CommunicationValue { + CommunicationValue::new(response_type) + .with_request_id(cv) + .with_receiver(sender_wire_id(setting.user_id)) + .add_typed_default( + DataType::SettingId, + DataValue::SignedNumber(setting.id.into()), + ) + .add_typed_default( + DataType::SettingScope, + DataValue::Str(setting.scope.as_str().to_string()), + ) + .add_typed_default( + DataType::SettingTarget, + DataValue::Str(setting.scope_key.clone()), + ) + .add_typed_default(DataType::SettingsName, DataValue::Str(setting.name.clone())) + .add_typed_default(DataType::Payload, DataValue::Str(setting.payload.clone())) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(setting.revision.into()), + ) +} + +fn setting_changed(setting: &SyncedSetting) -> CommunicationValue { + CommunicationValue::new(CommunicationType::SyncedSettingChanged) + .with_receiver(sender_wire_id(setting.user_id)) + .add_typed_default( + DataType::SettingId, + DataValue::SignedNumber(setting.id.into()), + ) + .add_typed_default( + DataType::SettingScope, + DataValue::Str(setting.scope.as_str().to_string()), + ) + .add_typed_default( + DataType::SettingTarget, + DataValue::Str(setting.scope_key.clone()), + ) + .add_typed_default(DataType::SettingsName, DataValue::Str(setting.name.clone())) + .add_typed_default(DataType::Payload, DataValue::Str(setting.payload.clone())) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(setting.revision.into()), + ) +} + +fn setting_deleted(user_id: i64, deleted: &synced_settings::DeletedSetting) -> CommunicationValue { + CommunicationValue::new(CommunicationType::SyncedSettingChanged) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default( + DataType::DeletedSettingIds, + DataValue::Array(vec![DataValue::SignedNumber(deleted.id.into())]), + ) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(deleted.revision.into()), + ) +} + +fn parse_setting_locator(cv: &CommunicationValue) -> Result { + let Some(scope_name) = cv.get_data(DataType::SettingScope).as_str() else { + return Err(error_response(cv, CommunicationType::ErrorInvalidData)); + }; + let Some(scope) = SettingScope::parse(scope_name) else { + return Err(error_response(cv, CommunicationType::ErrorInvalidData)); + }; + let Some(scope_key) = cv.get_data(DataType::SettingTarget).as_str() else { + return Err(error_response(cv, CommunicationType::ErrorInvalidData)); + }; + let Some(name) = cv.get_data(DataType::SettingsName).as_str() else { + return Err(error_response(cv, CommunicationType::ErrorInvalidData)); + }; + if !synced_settings::is_valid_name(name) { + return Err(error_response(cv, CommunicationType::ErrorInvalidData)); + } + match scope { + SettingScope::User if !scope_key.is_empty() => { + Err(error_response(cv, CommunicationType::ErrorInvalidData)) + } + SettingScope::Contact if !scope_key.parse::().is_ok_and(|id| id > 0) => { + Err(error_response(cv, CommunicationType::ErrorInvalidData)) + } + SettingScope::Community if scope_key.is_empty() => { + Err(error_response(cv, CommunicationType::ErrorInvalidData)) + } + _ => Ok(SettingLocator { + scope, + scope_key: scope_key.to_string(), + name: name.to_string(), + }), + } +} + +fn validate_setting_target( + user_id: i64, + locator: &SettingLocator, +) -> Result<(), CommunicationType> { + match locator.scope { + SettingScope::User => Ok(()), + SettingScope::Contact => { + let contact_id = locator + .scope_key + .parse::() + .map_err(|_| CommunicationType::ErrorInvalidData)?; + match has_user(user_id, contact_id) { + Ok(true) => Ok(()), + Ok(false) => Err(CommunicationType::ErrorInvalidData), + Err(_) => Err(CommunicationType::ErrorInternal), + } + } + SettingScope::Community => { + match CommunitiesUtil::has_community(user_id, &locator.scope_key) { + Ok(true) => Ok(()), + Ok(false) => Err(CommunicationType::ErrorInvalidData), + Err(_) => Err(CommunicationType::ErrorInternal), + } + } + } +} + +fn setting_mutation_error( + cv: &CommunicationValue, + error_type: CommunicationType, +) -> SettingMutation { + SettingMutation { + response: error_response(cv, error_type), + changed: None, + } +} + +pub fn handle_synced_setting_set(cv: &CommunicationValue) -> SettingMutation { + let user_id = match required_sender_id(cv) { + Ok(user_id) if user_id > 0 => user_id, + _ => return setting_mutation_error(cv, CommunicationType::ErrorInvalidData), + }; + let locator = match parse_setting_locator(cv) { + Ok(locator) => locator, + Err(response) => { + return SettingMutation { + response, + changed: None, + }; + } + }; + if let Err(error_type) = validate_setting_target(user_id, &locator) { + return setting_mutation_error(cv, error_type); + } + let Some(payload) = cv.get_data(DataType::Payload).as_str() else { + return setting_mutation_error(cv, CommunicationType::ErrorInvalidData); + }; + match synced_settings::set( + user_id, + locator.scope, + &locator.scope_key, + &locator.name, + payload, + ) { + Ok(setting) => SettingMutation { + response: setting_response(cv, CommunicationType::SyncedSettingSet, &setting), + changed: Some(setting_changed(&setting)), + }, + Err(_) => setting_mutation_error(cv, CommunicationType::ErrorInternal), + } +} + +pub fn handle_synced_setting_get(cv: &CommunicationValue) -> CommunicationValue { + let user_id = match required_sender_id(cv) { + Ok(user_id) if user_id > 0 => user_id, + _ => return error_response(cv, CommunicationType::ErrorInvalidData), + }; + let locator = match parse_setting_locator(cv) { + Ok(locator) => locator, + Err(response) => return response, + }; + if let Err(error_type) = validate_setting_target(user_id, &locator) { + return error_response(cv, error_type); + } + match synced_settings::get(user_id, locator.scope, &locator.scope_key, &locator.name) { + Ok(Some(setting)) => setting_response(cv, CommunicationType::SyncedSettingGet, &setting), + Ok(None) => error_response(cv, CommunicationType::ErrorNotFound), + Err(_) => error_response(cv, CommunicationType::ErrorInternal), + } +} + +pub fn handle_synced_setting_delete(cv: &CommunicationValue) -> SettingMutation { + let user_id = match required_sender_id(cv) { + Ok(user_id) if user_id > 0 => user_id, + _ => return setting_mutation_error(cv, CommunicationType::ErrorInvalidData), + }; + let locator = match parse_setting_locator(cv) { + Ok(locator) => locator, + Err(response) => { + return SettingMutation { + response, + changed: None, + }; + } + }; + if let Err(error_type) = validate_setting_target(user_id, &locator) { + return setting_mutation_error(cv, error_type); + } + match synced_settings::delete(user_id, locator.scope, &locator.scope_key, &locator.name) { + Ok(Some(deleted)) => SettingMutation { + response: CommunicationValue::new(CommunicationType::SyncedSettingDelete) + .with_request_id(cv) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default( + DataType::SettingId, + DataValue::SignedNumber(deleted.id.into()), + ) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(deleted.revision.into()), + ), + changed: deleted.changed.then(|| setting_deleted(user_id, &deleted)), + }, + Ok(None) => SettingMutation { + response: CommunicationValue::new(CommunicationType::SyncedSettingDelete) + .with_request_id(cv) + .with_receiver(sender_wire_id(user_id)), + changed: None, + }, + Err(_) => setting_mutation_error(cv, CommunicationType::ErrorInternal), + } +} + +pub fn handle_synced_settings_list(cv: &CommunicationValue) -> CommunicationValue { + let user_id = match required_sender_id(cv) { + Ok(user_id) if user_id > 0 => user_id, + _ => return error_response(cv, CommunicationType::ErrorInvalidData), + }; + match synced_settings::list(user_id) { + Ok(settings) => CommunicationValue::new(CommunicationType::SyncedSettingsList) + .with_request_id(cv) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default( + DataType::Settings, + DataValue::Array(settings.iter().map(synced_setting_value).collect()), + ), + Err(_) => error_response(cv, CommunicationType::ErrorInternal), + } +} + +#[cfg(test)] +mod synced_settings_tests { + use super::{handle_synced_setting_get, handle_synced_setting_set, parse_setting_locator}; + use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; + + fn request() -> CommunicationValue { + CommunicationValue::new(CommunicationType::SyncedSettingSet) + .with_id(1) + .with_sender(7) + .add_typed_default(DataType::SettingScope, DataValue::Str("user".to_string())) + .add_typed_default(DataType::SettingTarget, DataValue::Str(String::new())) + .add_typed_default( + DataType::SettingsName, + DataValue::Str("notifications.enabled".to_string()), + ) + } + + #[test] + fn missing_sender_is_rejected_for_synced_settings() { + let response = handle_synced_setting_get(&request().without_sender()); + + assert!(response.is_type(CommunicationType::ErrorInvalidData)); + assert_eq!(response.id(), Some(1)); + assert_eq!(response.receiver(), None); + } + + #[test] + fn user_scope_rejects_a_non_empty_target() { + let request = + request().add_typed_default(DataType::SettingTarget, DataValue::Str("123".to_string())); + + let response = handle_synced_setting_set(&request).response; + + assert!(response.is_type(CommunicationType::ErrorInvalidData)); + } + + #[test] + fn contact_scope_rejects_a_malformed_target() { + let request = request() + .add_typed_default( + DataType::SettingScope, + DataValue::Str("contact".to_string()), + ) + .add_typed_default( + DataType::SettingTarget, + DataValue::Str("not-a-user".to_string()), + ); + + let response = handle_synced_setting_set(&request).response; + + assert!(response.is_type(CommunicationType::ErrorInvalidData)); + } + + #[test] + fn community_scope_requires_an_address() { + let request = request() + .add_typed_default( + DataType::SettingScope, + DataValue::Str("community".to_string()), + ) + .add_typed_default(DataType::SettingTarget, DataValue::Str(String::new())); + + let response = handle_synced_setting_set(&request).response; + + assert!(response.is_type(CommunicationType::ErrorInvalidData)); + } + + #[test] + fn invalid_setting_name_is_rejected() { + let request = request().add_typed_default( + DataType::SettingsName, + DataValue::Str("notifications..enabled".to_string()), + ); + + let response = parse_setting_locator(&request); + + assert!(response.is_err()); + } +} diff --git a/iota-connection/src/relay.rs b/iota-connection/src/relay.rs index 161fb93..8dde6c3 100644 --- a/iota-connection/src/relay.rs +++ b/iota-connection/src/relay.rs @@ -53,6 +53,28 @@ pub fn message_security_class(frame: &CommunicationValue) -> MessageSecurityClas } } +#[cfg(test)] +mod security_tests { + use super::{MessageSecurityClass, message_security_class}; + use mtp::codec::{CommunicationType, CommunicationValue}; + + #[test] + fn synchronized_setting_requests_are_authenticated_local_requests() { + for setting_type in [ + CommunicationType::SyncedSettingSet, + CommunicationType::SyncedSettingGet, + CommunicationType::SyncedSettingDelete, + CommunicationType::SyncedSettingsList, + CommunicationType::SyncedSettingChanged, + ] { + assert_eq!( + message_security_class(&CommunicationValue::new(setting_type)), + MessageSecurityClass::AuthenticatedLocalRequest + ); + } + } +} + #[derive(Debug, Clone)] pub struct UserIdentity { pub user_id: u64, diff --git a/iota-storage/src/users/user_manager.rs b/iota-storage/src/users/user_manager.rs index ed2294f..cad4a4e 100644 --- a/iota-storage/src/users/user_manager.rs +++ b/iota-storage/src/users/user_manager.rs @@ -276,6 +276,10 @@ pub fn purge_user_data(user_id: i64) -> Result<(), crate::storage_error::Storage params![user_id], )?; tx.execute("DELETE FROM settings WHERE user_id = ?1", params![user_id])?; + tx.execute( + "DELETE FROM synced_settings WHERE user_id = ?1", + params![user_id], + )?; tx.execute( "DELETE FROM sync_events WHERE user_id = ?1", params![user_id], diff --git a/iota-storage/src/util/chats_util.rs b/iota-storage/src/util/chats_util.rs index 634b8f6..e5e158a 100644 --- a/iota-storage/src/util/chats_util.rs +++ b/iota-storage/src/util/chats_util.rs @@ -1,8 +1,19 @@ +use crate::storage_error::StorageError; use crate::users::contact::Contact; use crate::util::db; use crate::util::sync::{self, EntityType, Operation}; use rusqlite::params; +pub fn has_user(storage_owner: i64, user_id: i64) -> Result { + db::with_db(|conn| { + Ok(conn.query_row( + "SELECT EXISTS(SELECT 1 FROM contacts WHERE storage_owner = ?1 AND user_id = ?2)", + params![storage_owner, user_id], + |row| row.get(0), + )?) + }) +} + pub fn mod_user(storage_owner: i64, contact: &Contact) { if let Err(e) = db::with_db(|conn| { let tx = conn.unchecked_transaction()?; diff --git a/iota-storage/src/util/communities_util.rs b/iota-storage/src/util/communities_util.rs index c560056..3b3be8e 100644 --- a/iota-storage/src/util/communities_util.rs +++ b/iota-storage/src/util/communities_util.rs @@ -1,4 +1,6 @@ +use crate::storage_error::StorageError; use crate::util::db; +use crate::util::synced_settings::{self, SettingScope}; use rusqlite::params; #[derive(Debug, Clone)] @@ -11,6 +13,16 @@ pub struct StoredCommunity { pub struct CommunitiesUtil; impl CommunitiesUtil { + pub fn has_community(storage_owner: i64, address: &str) -> Result { + db::with_db(|conn| { + Ok(conn.query_row( + "SELECT EXISTS(SELECT 1 FROM communities WHERE storage_owner = ?1 AND address = ?2)", + params![storage_owner, address], + |row| row.get(0), + )?) + }) + } + pub fn add_community(storage_owner: i64, address: String, title: String, position: String) { if let Err(e) = db::with_db(|conn| { conn.execute( @@ -29,16 +41,22 @@ impl CommunitiesUtil { } } - pub fn remove_community(storage_owner: i64, community_address: String) { - if let Err(e) = db::with_db(|conn| { - conn.execute( + pub fn remove_community( + storage_owner: i64, + community_address: String, + ) -> Result<(), StorageError> { + db::with_immediate_transaction(|tx| { + tx.execute( "DELETE FROM communities WHERE storage_owner = ?1 AND address = ?2", params![storage_owner, community_address], )?; - Ok(()) - }) { - eprintln!("Failed to remove_community: {}", e); - } + synced_settings::delete_scope_in_tx( + tx, + storage_owner, + SettingScope::Community, + &community_address, + ) + }) } pub fn get_communities(storage_owner: i64) -> Vec { diff --git a/iota-storage/src/util/db.rs b/iota-storage/src/util/db.rs index 260c667..ad540af 100644 --- a/iota-storage/src/util/db.rs +++ b/iota-storage/src/util/db.rs @@ -464,6 +464,31 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { )?; } + if current_version < 12 { + conn.execute_batch( + r#" + CREATE TABLE IF NOT EXISTS synced_settings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + scope_type TEXT NOT NULL + CHECK (scope_type IN ('user', 'contact', 'community')), + scope_key TEXT NOT NULL, + name TEXT NOT NULL, + payload TEXT NOT NULL, + revision INTEGER NOT NULL, + deleted INTEGER NOT NULL DEFAULT 0 + CHECK (deleted IN (0, 1)), + UNIQUE(user_id, scope_type, scope_key, name) + ); + CREATE INDEX IF NOT EXISTS idx_synced_settings_owner + ON synced_settings (user_id, deleted); + CREATE INDEX IF NOT EXISTS idx_synced_settings_scope + ON synced_settings (user_id, scope_type, scope_key, deleted); + PRAGMA user_version = 12; + "#, + )?; + } + Ok(()) } @@ -533,7 +558,7 @@ mod tests { run_migrations_on_connection(&conn)?; let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; - assert_eq!(version, 11); + assert_eq!(version, 12); for column in ["height", "reply_to", "edited_count", "deleted_by_external"] { let mut statement = conn.prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = ?1")?; @@ -552,7 +577,7 @@ mod tests { run_migrations_on_connection(&conn)?; run_migrations_on_connection(&conn)?; let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; - assert_eq!(version, 11); + assert_eq!(version, 12); for table in [ "sync_heads", "sync_events", @@ -561,6 +586,7 @@ mod tests { "relay_replay", "pending_relays", "relay_inbox", + "synced_settings", ] { let exists: i64 = conn.query_row( "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1", @@ -576,4 +602,22 @@ mod tests { } Ok(()) } + + #[test] + fn adds_synced_settings_to_a_version_eleven_schema() -> Result<(), StorageError> { + let conn = Connection::open_in_memory()?; + conn.execute_batch("PRAGMA user_version = 11;")?; + + run_migrations_on_connection(&conn)?; + + let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; + assert_eq!(version, 12); + for column in ["id", "user_id", "scope_type", "scope_key", "name", "payload", "revision", "deleted"] { + let mut statement = + conn.prepare("SELECT 1 FROM pragma_table_info('synced_settings') WHERE name = ?1")?; + assert!(statement.exists([column])?); + } + + Ok(()) + } } diff --git a/iota-storage/src/util/mod.rs b/iota-storage/src/util/mod.rs index 2a1eec6..0731077 100644 --- a/iota-storage/src/util/mod.rs +++ b/iota-storage/src/util/mod.rs @@ -8,3 +8,4 @@ pub mod relay_queue; pub mod relay_replay; pub mod settings; pub mod sync; +pub mod synced_settings; diff --git a/iota-storage/src/util/sync.rs b/iota-storage/src/util/sync.rs index a358a1d..b9303d0 100644 --- a/iota-storage/src/util/sync.rs +++ b/iota-storage/src/util/sync.rs @@ -1,21 +1,23 @@ -//! Durable per-user state journal used by device cache synchronization. +/* Durable per-user state journal used by device cache synchronization. */ use crate::storage_error::StorageError; use crate::util::db; use rusqlite::{Transaction, params}; use std::collections::BTreeMap; -pub const CACHE_SCHEMA_VERSION: i64 = 2; +pub const CACHE_SCHEMA_VERSION: i64 = 3; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum EntityType { Message, Contact, + Setting, } impl EntityType { fn as_str(self) -> &'static str { match self { Self::Message => "message", Self::Contact => "contact", + Self::Setting => "setting", } } } @@ -40,6 +42,8 @@ pub struct Delta { pub deleted_message_ids: Vec, pub contact_upserts: Vec, pub deleted_contact_ids: Vec, + pub setting_upserts: Vec, + pub deleted_setting_ids: Vec, } pub fn now_millis() -> i64 { @@ -145,16 +149,174 @@ pub fn delta(user_id: i64, from_version: i64, captured_head: i64) -> Result out.deleted_message_ids.push(id), - ("message", _) => out.message_upserts.push(id), - ("contact", "delete") => out.deleted_contact_ids.push(id), - ("contact", _) => out.contact_upserts.push(id), - _ => {} - } - } - Ok(out) + Ok(reduce_events(final_events)) }) } + +fn reduce_events(events: BTreeMap<(String, i64), String>) -> Delta { + let mut out = Delta::default(); + for ((kind, id), operation) in events { + match (kind.as_str(), operation.as_str()) { + ("message", "delete") => out.deleted_message_ids.push(id), + ("message", _) => out.message_upserts.push(id), + ("contact", "delete") => out.deleted_contact_ids.push(id), + ("contact", _) => out.contact_upserts.push(id), + ("setting", "delete") => out.deleted_setting_ids.push(id), + ("setting", _) => out.setting_upserts.push(id), + _ => {} + } + } + out +} + +#[cfg(test)] +fn reduce_event_sequence(events: I) -> Delta +where + I: IntoIterator, +{ + let mut final_events = BTreeMap::new(); + for (kind, id, operation) in events { + final_events.insert((kind, id), operation); + } + reduce_events(final_events) +} + +#[cfg(test)] +mod tests { + use super::{EntityType, Operation, reduce_event_sequence, reduce_events}; + use rusqlite::Connection; + use std::collections::BTreeMap; + + #[test] + fn setting_upsert_is_included_in_delta() { + let mut events = BTreeMap::new(); + events.insert( + (EntityType::Setting.as_str().to_string(), 7), + Operation::Upsert.as_str().to_string(), + ); + + let delta = reduce_events(events); + + assert_eq!(delta.setting_upserts, vec![7]); + assert!(delta.deleted_setting_ids.is_empty()); + } + + #[test] + fn setting_delete_is_included_in_delta() { + let mut events = BTreeMap::new(); + events.insert( + (EntityType::Setting.as_str().to_string(), 7), + Operation::Delete.as_str().to_string(), + ); + + let delta = reduce_events(events); + + assert_eq!(delta.deleted_setting_ids, vec![7]); + assert!(delta.setting_upserts.is_empty()); + } + + #[test] + fn final_setting_operation_wins() { + let mut events = BTreeMap::new(); + events.insert( + (EntityType::Setting.as_str().to_string(), 7), + Operation::Upsert.as_str().to_string(), + ); + events.insert( + (EntityType::Setting.as_str().to_string(), 8), + Operation::Delete.as_str().to_string(), + ); + + let delta = reduce_events(events); + + assert_eq!(delta.setting_upserts, vec![7]); + assert_eq!(delta.deleted_setting_ids, vec![8]); + } + + #[test] + fn setting_upsert_then_delete_resolves_to_delete() { + let delta = reduce_event_sequence([ + ( + EntityType::Setting.as_str().to_string(), + 7, + Operation::Upsert.as_str().to_string(), + ), + ( + EntityType::Setting.as_str().to_string(), + 7, + Operation::Delete.as_str().to_string(), + ), + ]); + + assert_eq!(delta.deleted_setting_ids, vec![7]); + assert!(delta.setting_upserts.is_empty()); + } + + #[test] + fn setting_delete_then_upsert_resolves_to_upsert() { + let delta = reduce_event_sequence([ + ( + EntityType::Setting.as_str().to_string(), + 7, + Operation::Delete.as_str().to_string(), + ), + ( + EntityType::Setting.as_str().to_string(), + 7, + Operation::Upsert.as_str().to_string(), + ), + ]); + + assert_eq!(delta.setting_upserts, vec![7]); + assert!(delta.deleted_setting_ids.is_empty()); + } + + #[test] + fn setting_events_share_the_user_sync_head() { + let connection = Connection::open_in_memory().unwrap(); + connection + .execute_batch( + " + CREATE TABLE sync_heads (user_id INTEGER PRIMARY KEY, version INTEGER NOT NULL); + CREATE TABLE sync_events ( + user_id INTEGER NOT NULL, + version INTEGER NOT NULL, + entity_type TEXT NOT NULL, + entity_id INTEGER NOT NULL, + operation TEXT NOT NULL, + created_at INTEGER NOT NULL, + PRIMARY KEY (user_id, version) + ); + ", + ) + .unwrap(); + let transaction = connection.unchecked_transaction().unwrap(); + let message = super::record_event( + &transaction, + 1, + EntityType::Message, + 10, + Operation::Upsert, + ) + .unwrap(); + let setting = super::record_event( + &transaction, + 1, + EntityType::Setting, + 11, + Operation::Upsert, + ) + .unwrap(); + let contact = super::record_event( + &transaction, + 1, + EntityType::Contact, + 12, + Operation::Upsert, + ) + .unwrap(); + transaction.commit().unwrap(); + + assert_eq!((message, setting, contact), (1, 2, 3)); + } +} diff --git a/iota-storage/src/util/synced_settings.rs b/iota-storage/src/util/synced_settings.rs new file mode 100644 index 0000000..61c884c --- /dev/null +++ b/iota-storage/src/util/synced_settings.rs @@ -0,0 +1,659 @@ +use crate::storage_error::StorageError; +use crate::util::db; +use crate::util::sync::{self, EntityType, Operation}; +use rusqlite::{params, Connection, OptionalExtension, Row, Transaction}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SettingScope { + User, + Contact, + Community, +} + +impl SettingScope { + pub fn as_str(self) -> &'static str { + match self { + Self::User => "user", + Self::Contact => "contact", + Self::Community => "community", + } + } + + pub fn parse(value: &str) -> Option { + match value { + "user" => Some(Self::User), + "contact" => Some(Self::Contact), + "community" => Some(Self::Community), + _ => None, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SyncedSetting { + pub id: i64, + pub user_id: i64, + pub scope: SettingScope, + pub scope_key: String, + pub name: String, + pub payload: String, + pub revision: i64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DeletedSetting { + pub id: i64, + pub revision: i64, + pub changed: bool, +} + +pub fn is_valid_name(name: &str) -> bool { + !name.is_empty() + && name + .chars() + .all(|character| character.is_alphanumeric() || "_-.".contains(character)) + && !name.contains("..") +} + +fn validate_locator(scope: SettingScope, scope_key: &str, name: &str) -> Result<(), StorageError> { + if !is_valid_name(name) { + return Err(StorageError::Other( + "invalid synchronized setting name".into(), + )); + } + + match scope { + SettingScope::User if !scope_key.is_empty() => Err(StorageError::Other( + "user settings must not have a target".into(), + )), + SettingScope::Contact => { + let valid_contact = scope_key.parse::().is_ok_and(|id| id > 0); + if valid_contact { + Ok(()) + } else { + Err(StorageError::Other("invalid contact setting target".into())) + } + } + SettingScope::Community if scope_key.is_empty() => Err(StorageError::Other( + "community settings require a target".into(), + )), + _ => Ok(()), + } +} + +fn normalized_scope_key(scope: SettingScope, scope_key: &str) -> Result { + match scope { + SettingScope::Contact => scope_key + .parse::() + .map(|id| id.to_string()) + .map_err(|_| StorageError::Other("invalid contact setting target".into())), + SettingScope::User | SettingScope::Community => Ok(scope_key.to_string()), + } +} + +fn setting_from_parts( + id: i64, + user_id: i64, + scope_type: String, + scope_key: String, + name: String, + payload: String, + revision: i64, +) -> Result { + let scope = SettingScope::parse(&scope_type) + .ok_or_else(|| StorageError::Other("database contains an invalid setting scope".into()))?; + Ok(SyncedSetting { + id, + user_id, + scope, + scope_key, + name, + payload, + revision, + }) +} + +fn row_parts(row: &Row<'_>) -> rusqlite::Result<(i64, i64, String, String, String, String, i64)> { + Ok(( + row.get(0)?, + row.get(1)?, + row.get(2)?, + row.get(3)?, + row.get(4)?, + row.get(5)?, + row.get(6)?, + )) +} + +fn load_setting_from_tx( + tx: &Transaction<'_>, + setting_id: i64, +) -> Result { + let parts = tx.query_row( + "SELECT id, user_id, scope_type, scope_key, name, payload, revision + FROM synced_settings + WHERE id = ?1 AND deleted = 0", + [setting_id], + row_parts, + )?; + setting_from_parts( + parts.0, parts.1, parts.2, parts.3, parts.4, parts.5, parts.6, + ) +} + +pub fn set( + user_id: i64, + scope: SettingScope, + scope_key: &str, + name: &str, + payload: &str, +) -> Result { + if user_id <= 0 { + return Err(StorageError::Other("invalid setting owner".into())); + } + validate_locator(scope, scope_key, name)?; + let scope_key = normalized_scope_key(scope, scope_key)?; + + db::with_immediate_transaction(|tx| set_in_tx(tx, user_id, scope, &scope_key, name, payload)) +} + +fn set_in_tx( + tx: &Transaction<'_>, + user_id: i64, + scope: SettingScope, + scope_key: &str, + name: &str, + payload: &str, +) -> Result { + tx.execute( + "INSERT INTO synced_settings + (user_id, scope_type, scope_key, name, payload, revision, deleted) + VALUES (?1, ?2, ?3, ?4, ?5, 0, 0) + ON CONFLICT(user_id, scope_type, scope_key, name) DO NOTHING", + params![user_id, scope.as_str(), scope_key, name, payload], + )?; + let setting_id: i64 = tx.query_row( + "SELECT id FROM synced_settings + WHERE user_id = ?1 AND scope_type = ?2 AND scope_key = ?3 AND name = ?4", + params![user_id, scope.as_str(), scope_key, name], + |row| row.get(0), + )?; + let revision = sync::record_event( + tx, + user_id, + EntityType::Setting, + setting_id, + Operation::Upsert, + )?; + tx.execute( + "UPDATE synced_settings + SET payload = ?2, deleted = 0, revision = ?3 + WHERE id = ?1", + params![setting_id, payload, revision], + )?; + load_setting_from_tx(tx, setting_id) +} + +pub fn get( + user_id: i64, + scope: SettingScope, + scope_key: &str, + name: &str, +) -> Result, StorageError> { + if user_id <= 0 { + return Err(StorageError::Other("invalid setting owner".into())); + } + validate_locator(scope, scope_key, name)?; + let scope_key = normalized_scope_key(scope, scope_key)?; + db::with_db(|conn| get_from_connection(conn, user_id, scope, &scope_key, name)) +} + +fn get_from_connection( + conn: &Connection, + user_id: i64, + scope: SettingScope, + scope_key: &str, + name: &str, +) -> Result, StorageError> { + let parts = conn + .query_row( + "SELECT id, user_id, scope_type, scope_key, name, payload, revision + FROM synced_settings + WHERE user_id = ?1 AND scope_type = ?2 AND scope_key = ?3 + AND name = ?4 AND deleted = 0", + params![user_id, scope.as_str(), scope_key, name], + row_parts, + ) + .optional()?; + parts + .map(|parts| { + setting_from_parts( + parts.0, parts.1, parts.2, parts.3, parts.4, parts.5, parts.6, + ) + }) + .transpose() +} + +pub fn list(user_id: i64) -> Result, StorageError> { + if user_id <= 0 { + return Err(StorageError::Other("invalid setting owner".into())); + } + db::with_db(|conn| list_from_connection(conn, user_id)) +} + +fn list_from_connection( + conn: &Connection, + user_id: i64, +) -> Result, StorageError> { + let mut statement = conn.prepare( + "SELECT id, user_id, scope_type, scope_key, name, payload, revision + FROM synced_settings + WHERE user_id = ?1 AND deleted = 0 + ORDER BY id ASC", + )?; + let rows = statement.query_map([user_id], row_parts)?; + let mut settings = Vec::new(); + for row in rows { + let parts = row?; + settings.push(setting_from_parts( + parts.0, parts.1, parts.2, parts.3, parts.4, parts.5, parts.6, + )?); + } + Ok(settings) +} + +pub fn list_by_ids(user_id: i64, ids: &[i64]) -> Result, StorageError> { + if user_id <= 0 { + return Err(StorageError::Other("invalid setting owner".into())); + } + if ids.is_empty() { + return Ok(Vec::new()); + } + db::with_db(|conn| list_by_ids_from_connection(conn, user_id, ids)) +} + +fn list_by_ids_from_connection( + conn: &Connection, + user_id: i64, + ids: &[i64], +) -> Result, StorageError> { + let placeholders = std::iter::repeat_n("?", ids.len()) + .collect::>() + .join(", "); + let query = format!( + "SELECT id, user_id, scope_type, scope_key, name, payload, revision + FROM synced_settings + WHERE user_id = ? AND deleted = 0 AND id IN ({placeholders}) + ORDER BY id ASC" + ); + let mut values = Vec::with_capacity(ids.len() + 1); + values.push(user_id); + values.extend_from_slice(ids); + let mut statement = conn.prepare(&query)?; + let rows = statement.query_map(rusqlite::params_from_iter(values), row_parts)?; + let mut settings = Vec::new(); + for row in rows { + let parts = row?; + settings.push(setting_from_parts( + parts.0, parts.1, parts.2, parts.3, parts.4, parts.5, parts.6, + )?); + } + Ok(settings) +} + +pub fn delete( + user_id: i64, + scope: SettingScope, + scope_key: &str, + name: &str, +) -> Result, StorageError> { + if user_id <= 0 { + return Err(StorageError::Other("invalid setting owner".into())); + } + validate_locator(scope, scope_key, name)?; + let scope_key = normalized_scope_key(scope, scope_key)?; + db::with_immediate_transaction(|tx| delete_in_tx(tx, user_id, scope, &scope_key, name)) +} + +fn delete_in_tx( + tx: &Transaction<'_>, + user_id: i64, + scope: SettingScope, + scope_key: &str, + name: &str, +) -> Result, StorageError> { + let existing = tx + .query_row( + "SELECT id, revision, deleted FROM synced_settings + WHERE user_id = ?1 AND scope_type = ?2 AND scope_key = ?3 AND name = ?4", + params![user_id, scope.as_str(), scope_key, name], + |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, i64>(1)?, + row.get::<_, i64>(2)?, + )) + }, + ) + .optional()?; + let Some((setting_id, current_revision, deleted)) = existing else { + return Ok(None); + }; + if deleted != 0 { + return Ok(Some(DeletedSetting { + id: setting_id, + revision: current_revision, + changed: false, + })); + } + + let revision = sync::record_event( + tx, + user_id, + EntityType::Setting, + setting_id, + Operation::Delete, + )?; + tx.execute( + "UPDATE synced_settings SET deleted = 1, revision = ?2 WHERE id = ?1", + params![setting_id, revision], + )?; + Ok(Some(DeletedSetting { + id: setting_id, + revision, + changed: true, + })) +} + +pub(crate) fn delete_scope_in_tx( + tx: &Transaction<'_>, + user_id: i64, + scope: SettingScope, + scope_key: &str, +) -> Result<(), StorageError> { + let setting_ids = { + let mut statement = tx.prepare( + "SELECT id FROM synced_settings + WHERE user_id = ?1 AND scope_type = ?2 AND scope_key = ?3 AND deleted = 0", + )?; + let rows = statement.query_map(params![user_id, scope.as_str(), scope_key], |row| { + row.get::<_, i64>(0) + })?; + rows.collect::, _>>()? + }; + + for setting_id in setting_ids { + let revision = sync::record_event( + tx, + user_id, + EntityType::Setting, + setting_id, + Operation::Delete, + )?; + tx.execute( + "UPDATE synced_settings SET deleted = 1, revision = ?2 WHERE id = ?1", + params![setting_id, revision], + )?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::{ + delete_in_tx, get_from_connection, is_valid_name, list_by_ids_from_connection, + list_from_connection, set_in_tx, SettingScope, + }; + use rusqlite::Connection; + + fn connection() -> Connection { + let connection = Connection::open_in_memory().unwrap(); + connection + .execute_batch( + " + CREATE TABLE sync_heads ( + user_id INTEGER PRIMARY KEY, + version INTEGER NOT NULL + ); + CREATE TABLE sync_events ( + user_id INTEGER NOT NULL, + version INTEGER NOT NULL, + entity_type TEXT NOT NULL, + entity_id INTEGER NOT NULL, + operation TEXT NOT NULL, + created_at INTEGER NOT NULL, + PRIMARY KEY (user_id, version) + ); + CREATE TABLE synced_settings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + scope_type TEXT NOT NULL, + scope_key TEXT NOT NULL, + name TEXT NOT NULL, + payload TEXT NOT NULL, + revision INTEGER NOT NULL, + deleted INTEGER NOT NULL DEFAULT 0, + UNIQUE(user_id, scope_type, scope_key, name) + ); + ", + ) + .unwrap(); + connection + } + + fn set( + connection: &mut Connection, + user_id: i64, + scope: SettingScope, + scope_key: &str, + name: &str, + payload: &str, + ) -> super::SyncedSetting { + let transaction = connection.transaction().unwrap(); + let setting = set_in_tx(&transaction, user_id, scope, scope_key, name, payload).unwrap(); + transaction.commit().unwrap(); + setting + } + + #[test] + fn parses_supported_setting_scopes() { + assert_eq!(SettingScope::parse("user"), Some(SettingScope::User)); + assert_eq!(SettingScope::parse("contact"), Some(SettingScope::Contact)); + assert_eq!( + SettingScope::parse("community"), + Some(SettingScope::Community) + ); + } + + #[test] + fn rejects_unknown_setting_scope() { + assert_eq!(SettingScope::parse("device"), None); + } + + #[test] + fn validates_setting_name_syntax() { + assert!(is_valid_name("notifications.enabled")); + assert!(!is_valid_name("notifications..enabled")); + assert!(!is_valid_name("")); + assert!(!is_valid_name("notifications/enabled")); + } + + #[test] + fn stores_and_loads_a_user_setting() { + let mut connection = connection(); + let stored = set( + &mut connection, + 1, + SettingScope::User, + "", + "notifications.enabled", + "true", + ); + let loaded = get_from_connection( + &connection, + 1, + SettingScope::User, + "", + "notifications.enabled", + ) + .unwrap(); + + assert_eq!(loaded, Some(stored)); + } + + #[test] + fn contact_and_community_targets_are_distinct() { + let mut connection = connection(); + set( + &mut connection, + 1, + SettingScope::Contact, + "123", + "notifications.enabled", + "false", + ); + set( + &mut connection, + 1, + SettingScope::Community, + "community-a", + "notifications.enabled", + "true", + ); + + assert_eq!(list_from_connection(&connection, 1).unwrap().len(), 2); + } + + #[test] + fn users_store_same_setting_independently() { + let mut connection = connection(); + set( + &mut connection, + 1, + SettingScope::User, + "", + "receipts.user_read", + "true", + ); + set( + &mut connection, + 2, + SettingScope::User, + "", + "receipts.user_read", + "false", + ); + + assert_eq!( + get_from_connection(&connection, 1, SettingScope::User, "", "receipts.user_read") + .unwrap() + .unwrap() + .payload, + "true" + ); + assert_eq!( + get_from_connection(&connection, 2, SettingScope::User, "", "receipts.user_read") + .unwrap() + .unwrap() + .payload, + "false" + ); + } + + #[test] + fn update_retains_id_and_advances_revision() { + let mut connection = connection(); + let first = set( + &mut connection, + 1, + SettingScope::User, + "", + "notifications.enabled", + "true", + ); + let second = set( + &mut connection, + 1, + SettingScope::User, + "", + "notifications.enabled", + "false", + ); + + assert_eq!(second.id, first.id); + assert!(second.revision > first.revision); + } + + #[test] + fn delete_tombstones_setting_and_records_delta_delete() { + let mut connection = connection(); + let stored = set( + &mut connection, + 1, + SettingScope::User, + "", + "notifications.enabled", + "true", + ); + let transaction = connection.transaction().unwrap(); + let deleted = delete_in_tx( + &transaction, + 1, + SettingScope::User, + "", + "notifications.enabled", + ) + .unwrap() + .unwrap(); + transaction.commit().unwrap(); + let journal_operation: String = connection + .query_row( + "SELECT operation FROM sync_events WHERE entity_id = ?1 ORDER BY version DESC LIMIT 1", + [stored.id], + |row| row.get(0), + ) + .unwrap(); + + assert_eq!(deleted.id, stored.id); + assert!(deleted.changed); + assert_eq!(journal_operation, "delete"); + assert!(list_from_connection(&connection, 1).unwrap().is_empty()); + assert!(list_by_ids_from_connection(&connection, 1, &[stored.id]) + .unwrap() + .is_empty()); + } + + #[test] + fn setting_can_be_recreated_with_the_same_id() { + let mut connection = connection(); + let first = set( + &mut connection, + 1, + SettingScope::User, + "", + "notifications.enabled", + "true", + ); + let transaction = connection.transaction().unwrap(); + delete_in_tx( + &transaction, + 1, + SettingScope::User, + "", + "notifications.enabled", + ) + .unwrap(); + transaction.commit().unwrap(); + let recreated = set( + &mut connection, + 1, + SettingScope::User, + "", + "notifications.enabled", + "false", + ); + + assert_eq!(recreated.id, first.id); + assert_eq!(recreated.payload, "false"); + } +} diff --git a/mtp-type-maps b/mtp-type-maps index f3c5037..f4e45aa 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit f3c5037b0a099ae5389486eec77471bd5addab4a +Subproject commit f4e45aa3a3ad0e3c3a257f66857b904a1af7901c diff --git a/omikron-connector/Cargo.toml b/omikron-connector/Cargo.toml index c4773f7..5902891 100644 --- a/omikron-connector/Cargo.toml +++ b/omikron-connector/Cargo.toml @@ -14,6 +14,7 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = [ "client", "crypto", "files", + "raw", ] } dashmap = "6.2.1" diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 0c37013..119c9a4 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -12,7 +12,6 @@ use mtp::crypto::{Keyring, PublicKeyBundle}; use rand_core::RngCore; use std::env; use std::fs; -use std::io::ErrorKind; use std::path::{Path, PathBuf}; use std::sync::{Arc, LazyLock}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -38,9 +37,6 @@ use iota_util::route_target::RouteTarget; // ============================================================================ const IOTA_KEYRING_PATH: &str = "iota.mk"; -const IDENTITY_SECRET_ENV: &str = "IOTA_IDENTITY_SECRET"; -const IDENTITY_SECRET_FILE_ENV: &str = "IOTA_IDENTITY_SECRET_FILE"; -const SYSTEMD_IDENTITY_CREDENTIAL: &str = "iota-identity"; static IDENTITY_PATH: std::sync::OnceLock = std::sync::OnceLock::new(); static OMIKRON_PUBLIC_KEY_PATH: std::sync::OnceLock = std::sync::OnceLock::new(); @@ -107,7 +103,6 @@ const RELAY_RETENTION_MILLIS: i64 = 30 * 24 * 60 * 60 * 1000; pub enum IdentityError { Storage(mtp::files::FileError), Directory(std::io::Error), - Secret(String), InvalidLegacyIdentity, Verification(String), } @@ -117,7 +112,6 @@ impl std::fmt::Display for IdentityError { match self { Self::Storage(error) => write!(f, "identity storage error: {error}"), Self::Directory(error) => write!(f, "unable to create identity directory: {error}"), - Self::Secret(error) => write!(f, "unable to load identity secret: {error}"), Self::InvalidLegacyIdentity => f.write_str("legacy identity is invalid"), Self::Verification(error) => { write!(f, "persisted identity could not be verified: {error}") @@ -154,68 +148,13 @@ fn wire_user_id(user_id: i64) -> u64 { u64::try_from(user_id).expect("validated user ID is non-negative") } -fn load_identity_secret() -> Result, IdentityError> { - if let Some(path) = env::var_os(IDENTITY_SECRET_FILE_ENV) { - let path = PathBuf::from(path); - let mut secret = fs::read(&path) - .map_err(|error| IdentityError::Secret(format!("{}: {error}", path.display())))?; - while matches!(secret.last(), Some(b'\n' | b'\r')) { - secret.pop(); - } - if secret.is_empty() { - return Err(IdentityError::Secret(format!( - "{} is empty", - path.display() - ))); - } - return Ok(secret); - } - - if let Ok(secret) = env::var(IDENTITY_SECRET_ENV) { - if secret.is_empty() { - return Err(IdentityError::Secret(format!( - "{IDENTITY_SECRET_ENV} is empty" - ))); - } - return Ok(secret.into_bytes()); - } - - if let Ok(credentials_dir) = env::var("CREDENTIALS_DIRECTORY") { - let path = Path::new(&credentials_dir).join(SYSTEMD_IDENTITY_CREDENTIAL); - let mut secret = fs::read(&path) - .map_err(|error| IdentityError::Secret(format!("{}: {error}", path.display())))?; - while matches!(secret.last(), Some(b'\n' | b'\r')) { - secret.pop(); - } - if secret.is_empty() { - return Err(IdentityError::Secret(format!( - "{} is empty", - path.display() - ))); - } - return Ok(secret); - } - - Err(IdentityError::Secret(format!( - "set {IDENTITY_SECRET_FILE_ENV}, {IDENTITY_SECRET_ENV}, or a systemd identity credential" - ))) -} - -fn load_legacy_raw_keyring(path: &Path) -> Result { - let bytes = fs::read(path).map_err(|error| IdentityError::Storage(error.into()))?; - if bytes.len() < 5 || bytes[..4] != *b"MTMK" || bytes[4] != 1 { - return Err(IdentityError::InvalidLegacyIdentity); - } - Keyring::from_bytes(&bytes[5..]).map_err(|_| IdentityError::InvalidLegacyIdentity) -} - -fn save_protected_keyring_verified( - keyring: &Keyring, - path: &Path, - passphrase: &[u8], -) -> Result<(), IdentityError> { - mtp::files::save_keyring(keyring, path, passphrase).map_err(IdentityError::Storage)?; - let persisted = mtp::files::load_keyring(path, passphrase).map_err(IdentityError::Storage)?; +/* + * The identity is stored in the Iota state directory as raw keyring bytes so + * daemon restarts do not depend on a separately managed passphrase. + */ +fn save_keyring_verified(keyring: &Keyring, path: &Path) -> Result<(), IdentityError> { + mtp::files::save_keyring_raw(keyring, path).map_err(IdentityError::Storage)?; + let persisted = mtp::files::load_keyring_raw(path).map_err(IdentityError::Storage)?; let expected = keyring .try_to_bytes() .map_err(|error| IdentityError::Verification(error.to_string()))?; @@ -233,7 +172,6 @@ fn save_protected_keyring_verified( fn load_or_migrate_keyring_at( path: &Path, legacy: Option, - passphrase: &[u8], ) -> Result { if let Some(parent) = path .parent() @@ -242,14 +180,9 @@ fn load_or_migrate_keyring_at( fs::create_dir_all(parent).map_err(IdentityError::Directory)?; } - match mtp::files::load_keyring(path, passphrase) { + match mtp::files::load_keyring_raw(path) { Ok(keyring) => return Ok(keyring), - Err(mtp::files::FileError::Io(error)) if error.kind() == ErrorKind::NotFound => {} - Err(mtp::files::FileError::UnprotectedKeyring) => { - let keyring = load_legacy_raw_keyring(path)?; - save_protected_keyring_verified(&keyring, path, passphrase)?; - return Ok(keyring); - } + Err(mtp::files::FileError::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => {} Err(error) => return Err(IdentityError::Storage(error)), } @@ -266,7 +199,7 @@ fn load_or_migrate_keyring_at( } }; - save_protected_keyring_verified(&keyring, path, passphrase)?; + save_keyring_verified(&keyring, path)?; Ok(keyring) } @@ -486,9 +419,8 @@ impl OmikronConnection { self.set_state(ConnectionState::Connecting).await; log_t!("omikron_connecting"); - let identity_secret = load_identity_secret().map_err(|error| error.to_string())?; let keyring = Arc::new( - self.load_or_migrate_keyring(&identity_secret) + self.load_or_migrate_keyring() .await .map_err(|error| format!("Iota identity initialization failed: {error}"))?, ); @@ -597,8 +529,8 @@ impl OmikronConnection { // Identity (own Keyring, migrated from the legacy base64-in-config format) // ------------------------------------------------------------------------- - async fn load_or_migrate_keyring(&self, passphrase: &[u8]) -> Result { - load_or_migrate_keyring_at(identity_path(), CONFIG.load().keyring.clone(), passphrase) + async fn load_or_migrate_keyring(&self) -> Result { + load_or_migrate_keyring_at(identity_path(), CONFIG.load().keyring.clone()) } // ------------------------------------------------------------------------- @@ -1063,7 +995,10 @@ impl OmikronConnection { ) { Ok(value) => value, Err(error) => { - log!("Relay shared-Iota origin content verification failed: {}", error); + log!( + "Relay shared-Iota origin content verification failed: {}", + error + ); let _ = relay_replay::mark_rejected( verified.context.signer_id, &verified.context.message_id, @@ -1103,7 +1038,10 @@ impl OmikronConnection { &verified.context.message_id, accepted_at, ) { - log!("Relay shared-Iota destination timestamp storage failed: {}", error); + log!( + "Relay shared-Iota destination timestamp storage failed: {}", + error + ); self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) .await; return; @@ -1188,10 +1126,9 @@ impl OmikronConnection { .await; return; } - if let Err(error) = relay_replay::mark_queued( - verified.context.signer_id, - &verified.context.message_id, - ) { + if let Err(error) = + relay_replay::mark_queued(verified.context.signer_id, &verified.context.message_id) + { log!("Relay origin state update failed: {}", error); } match self @@ -1199,9 +1136,7 @@ impl OmikronConnection { .await { Ok(response) if response.is_type(CommunicationType::Success) => { - let returned_id = response - .get_data(DataType::RelayMessageId) - .as_str(); + let returned_id = response.get_data(DataType::RelayMessageId).as_str(); let destination_accepted_at = response .get_data(DataType::RelayAcceptedAt) .as_number() @@ -1228,7 +1163,10 @@ impl OmikronConnection { &verified.context.message_id, destination_accepted_at, ) { - log!("Relay destination acknowledgement storage failed: {}", error); + log!( + "Relay destination acknowledgement storage failed: {}", + error + ); } } if let Err(error) = relay_queue::acknowledge_iota(router, frame_id) { @@ -1337,26 +1275,21 @@ impl OmikronConnection { return; } }; - if let Err(error) = - message_handlers::apply_verified_relay_content( - &verified.context, - &content, - accepted_at, - match i64::try_from(destination) { - Ok(value) => value, - Err(_) => { - log!("Relay destination ID exceeds storage range"); - self.send_relay_response( - frame.id(), - CommunicationType::ErrorInvalidData, - ) + if let Err(error) = message_handlers::apply_verified_relay_content( + &verified.context, + &content, + accepted_at, + match i64::try_from(destination) { + Ok(value) => value, + Err(_) => { + log!("Relay destination ID exceeds storage range"); + self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData) .await; - return; - } - }, - false, - ) - { + return; + } + }, + false, + ) { log!("Relay application dispatch failed: {}", error); if let Err(queue_error) = relay_queue::remove_for_frame(RouteTarget::User(destination), frame_id) @@ -1371,10 +1304,9 @@ impl OmikronConnection { .await; return; } - if let Err(error) = relay_replay::mark_applied( - verified.context.signer_id, - &verified.context.message_id, - ) { + if let Err(error) = + relay_replay::mark_applied(verified.context.signer_id, &verified.context.message_id) + { log!("Relay application state update failed: {}", error); self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) .await; @@ -1382,10 +1314,9 @@ impl OmikronConnection { } } - if let Err(error) = relay_replay::mark_queued( - verified.context.signer_id, - &verified.context.message_id, - ) { + if let Err(error) = + relay_replay::mark_queued(verified.context.signer_id, &verified.context.message_id) + { log!("Relay queue state update failed: {}", error); } self.send_relay_success( @@ -1564,6 +1495,10 @@ impl OmikronConnection { dispatch!(SettingsSave, handle_settings_save); dispatch!(SettingsLoad, handle_settings_load); dispatch!(SettingsList, handle_settings_list); + dispatch!(SyncedSettingSet, handle_synced_setting_set); + dispatch!(SyncedSettingGet, handle_synced_setting_get); + dispatch!(SyncedSettingDelete, handle_synced_setting_delete); + dispatch!(SyncedSettingsList, handle_synced_settings_list); dispatch!(EraseHostedUserData, handle_erase_hosted_user_data); } @@ -2143,6 +2078,34 @@ impl OmikronConnection { .await; } + async fn handle_synced_setting_set(self: Arc, cv: &CommunicationValue) { + let mutation = message_handlers::handle_synced_setting_set(cv); + let _ = self.send_message(&mutation.response).await; + if let Some(changed) = mutation.changed { + let _ = self.send_message(&changed).await; + } + } + + async fn handle_synced_setting_get(self: Arc, cv: &CommunicationValue) { + let _ = self + .send_message(&message_handlers::handle_synced_setting_get(cv)) + .await; + } + + async fn handle_synced_setting_delete(self: Arc, cv: &CommunicationValue) { + let mutation = message_handlers::handle_synced_setting_delete(cv); + let _ = self.send_message(&mutation.response).await; + if let Some(changed) = mutation.changed { + let _ = self.send_message(&changed).await; + } + } + + async fn handle_synced_settings_list(self: Arc, cv: &CommunicationValue) { + let _ = self + .send_message(&message_handlers::handle_synced_settings_list(cv)) + .await; + } + // ------------------------------------------------------------------------- // Public API // ------------------------------------------------------------------------- @@ -2329,8 +2292,6 @@ impl OmikronConnection { /// recovery does not silently destroy the user's previous identity. pub async fn rotate_identity(self: &Arc) -> Result<(), OmikronError> { log!("Iota identity rotation requested"); - let identity_secret = - load_identity_secret().map_err(|error| OmikronError::Internal(error.to_string()))?; self.stop().await; let path = identity_path(); @@ -2361,7 +2322,7 @@ impl OmikronConnection { )) })?; } - save_protected_keyring_verified(&keyring, path, &identity_secret).map_err(|error| { + save_keyring_verified(&keyring, path).map_err(|error| { OmikronError::Internal(format!( "could not save new identity {}: {error}", path.display() @@ -2542,16 +2503,15 @@ mod tests { } #[test] - fn generated_identity_is_protected_and_survives_reload() { + fn generated_identity_is_unprotected_and_survives_reload() { let path = test_path("reload"); - let passphrase = b"test identity secret"; - let keyring = load_or_migrate_keyring_at(&path, None, passphrase).expect("identity saves"); - let reloaded = load_or_migrate_keyring_at(&path, None, passphrase).expect("identity loads"); + let keyring = load_or_migrate_keyring_at(&path, None).expect("identity saves"); + let reloaded = load_or_migrate_keyring_at(&path, None).expect("identity loads"); assert_eq!( keyring.try_to_bytes().expect("keyring serializes"), reloaded.try_to_bytes().expect("keyring serializes") ); - assert!(mtp::files::load_keyring(&path, b"wrong secret").is_err()); + assert!(mtp::files::load_keyring_raw(&path).is_ok()); let _ = fs::remove_file(path); } @@ -2559,14 +2519,14 @@ mod tests { fn corrupt_existing_identity_does_not_generate_a_replacement() { let path = test_path("corrupt"); fs::write(&path, b"not a keyring").expect("corrupt fixture writes"); - let error = load_or_migrate_keyring_at(&path, None, b"test identity secret") - .expect_err("corrupt identity must fail"); + let error = + load_or_migrate_keyring_at(&path, None).expect_err("corrupt identity must fail"); assert!(matches!(error, IdentityError::Storage(_))); let _ = fs::remove_file(path); } #[test] - fn legacy_raw_identity_is_migrated_only_when_the_raw_format_is_valid() { + fn legacy_raw_identity_is_loaded_only_when_the_raw_format_is_valid() { let path = test_path("legacy"); let keyring = crypto_helper::generate_keyring(); let mut raw = b"MTMK".to_vec(); @@ -2574,8 +2534,7 @@ mod tests { raw.extend_from_slice(&keyring.try_to_bytes().expect("keyring serializes")); fs::write(&path, raw).expect("legacy fixture writes"); - let migrated = load_or_migrate_keyring_at(&path, None, b"test identity secret") - .expect("legacy identity migrates"); + let migrated = load_or_migrate_keyring_at(&path, None).expect("legacy identity loads"); assert_eq!( migrated.try_to_bytes().expect("keyring serializes"), keyring.try_to_bytes().expect("keyring serializes") @@ -2588,7 +2547,7 @@ mod tests { let parent = test_path("parent-file"); fs::write(&parent, b"not a directory").expect("parent fixture writes"); let path = parent.join("iota.mk"); - let error = load_or_migrate_keyring_at(&path, None, b"test identity secret") + let error = load_or_migrate_keyring_at(&path, None) .expect_err("directory failure must be returned"); assert!(matches!(error, IdentityError::Directory(_))); let _ = fs::remove_file(parent); diff --git a/systemd/iota-daemon.service b/systemd/iota-daemon.service index 2c6340b..a5a2903 100644 --- a/systemd/iota-daemon.service +++ b/systemd/iota-daemon.service @@ -17,7 +17,6 @@ Environment=IOTA_SOCKET=/run/iota/iota.sock Environment=IOTA_DATA_DIR=/var/lib/iota Environment=IOTA_DEPLOYMENT_MODE=system_always_on Environment=IOTA_SUPERVISOR=systemd -LoadCredential=iota-identity:/etc/iota/iota-identity.secret # Exit code 75 = restart requested (daemon-specific convention) RestartPreventExitStatus=0