tensamin/iota
Watch
6
0
Fork
You've already forked iota
 0
Code Issues Pull requests 3 Projects Releases 1 Packages Wiki Activity Actions

[Fix] Connectivity

Browse source
This commit is contained in:
Alex Emmet 2026-08-29 12:49:10 +02:00
commit afc1832fb7
No known key found for this signature in database
14 changed files with 1462 additions and 188 deletions
Download patch file Download diff file Expand all files Collapse all files

9
flake.nix
View file

@ -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}";
};
};

449
iota-connection/src/message_handlers.rs
View file

@ -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<CommunicationValue>,
}
struct SettingLocator {
scope: SettingScope,
scope_key: String,
name: String,
}
fn required_sender_id(cv: &CommunicationValue) -> Result<i64, CommunicationValue> {
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<SettingLocator, CommunicationValue> {
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::<i64>().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::<i64>()
.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());
}
}

22
iota-connection/src/relay.rs
View file

@ -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,

4
iota-storage/src/users/user_manager.rs
View file

@ -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],

11
iota-storage/src/util/chats_util.rs
View file

@ -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<bool, StorageError> {
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()?;

32
iota-storage/src/util/communities_util.rs
View file

@ -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<bool, StorageError> {
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<StoredCommunity> {

48
iota-storage/src/util/db.rs
View file

@ -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(())
}
}

1
iota-storage/src/util/mod.rs
View file

@ -8,3 +8,4 @@ pub mod relay_queue;
pub mod relay_replay;
pub mod settings;
pub mod sync;
pub mod synced_settings;

188
iota-storage/src/util/sync.rs
View file

@ -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<i64>,
pub contact_upserts: Vec<i64>,
pub deleted_contact_ids: Vec<i64>,
pub setting_upserts: Vec<i64>,
pub deleted_setting_ids: Vec<i64>,
}
pub fn now_millis() -> i64 {
@ -145,16 +149,174 @@ pub fn delta(user_id: i64, from_version: i64, captured_head: i64) -> Result<Delt
let (kind, id, operation) = row?;
final_events.insert((kind, id), operation);
}
let mut out = Delta::default();
for ((kind, id), operation) in final_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),
_ => {}
}
}
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<I>(events: I) -> Delta
where
I: IntoIterator<Item = (String, i64, String)>,
{
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));
}
}

659
iota-storage/src/util/synced_settings.rs Normal file
View file

@ -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<Self> {
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::<i64>().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<String, StorageError> {
match scope {
SettingScope::Contact => scope_key
.parse::<i64>()
.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<SyncedSetting, StorageError> {
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<SyncedSetting, StorageError> {
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<SyncedSetting, 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| 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<SyncedSetting, StorageError> {
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<Option<SyncedSetting>, 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<Option<SyncedSetting>, 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<Vec<SyncedSetting>, 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<Vec<SyncedSetting>, 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<Vec<SyncedSetting>, 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<Vec<SyncedSetting>, StorageError> {
let placeholders = std::iter::repeat_n("?", ids.len())
.collect::<Vec<_>>()
.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<Option<DeletedSetting>, 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<Option<DeletedSetting>, 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::<Result<Vec<_>, _>>()?
};
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");
}
}

2
mtp-type-maps

@ -1 +1 @@
Subproject commit f3c5037b0a099ae5389486eec77471bd5addab4a
Subproject commit f4e45aa3a3ad0e3c3a257f66857b904a1af7901c

1
omikron-connector/Cargo.toml
View file

@ -14,6 +14,7 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = [
"client",
"crypto",
"files",
"raw",
] }
dashmap = "6.2.1"

223
omikron-connector/src/omikron_connection.rs
View file

@ -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<PathBuf> = std::sync::OnceLock::new();
static OMIKRON_PUBLIC_KEY_PATH: std::sync::OnceLock<PathBuf> = 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<Vec<u8>, 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<Keyring, IdentityError> {
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<String>,
passphrase: &[u8],
) -> Result<Keyring, IdentityError> {
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<Keyring, IdentityError> {
load_or_migrate_keyring_at(identity_path(), CONFIG.load().keyring.clone(), passphrase)
async fn load_or_migrate_keyring(&self) -> Result<Keyring, IdentityError> {
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<Self>, 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<Self>, cv: &CommunicationValue) {
let _ = self
.send_message(&message_handlers::handle_synced_setting_get(cv))
.await;
}
async fn handle_synced_setting_delete(self: Arc<Self>, 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<Self>, 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<Self>) -> 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);

1
systemd/iota-daemon.service
View file

@ -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