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

Compare commits

base: tensamin:c2b15b9890b8a4b1387f4cb0b7716d596f78bff4
Branches Tags
tensamin:main
tensamin:renovate/zip-8.x
tensamin:renovate/rand_core-0.x
tensamin:renovate/rand-0.x
tensamin:renovate/base64-0.x
tensamin:0.1.0-dev-f252724
tensamin:1.0.0
...
compare: tensamin:ea2df05dc8dfb08729bb47a3e07eeee4d6a3cee0
Branches Tags
tensamin:renovate/zip-8.x
tensamin:renovate/rand_core-0.x
tensamin:renovate/rand-0.x
tensamin:renovate/base64-0.x
tensamin:main
tensamin:0.1.0-dev-f252724
tensamin:1.0.0
Author SHA1 Message Date
Rasensprenger
ea2df05dc8 Update Rust crate base64 to 0.23.0
All checks were successful
renovate/stability-days Updates have met minimum release age requirement
Details
2026-08-29 14:02:27 +03:00
Alex Emmet
afc1832fb7
 		[Fix] Connectivity 2026-08-29 12:49:10 +02:00
17 changed files with 1468 additions and 194 deletions
Download patch file Download diff file Expand all files Collapse all files

6
Cargo.lock generated
View file

@ -2240,7 +2240,7 @@ name = "iota-storage"
version = "0.1.0"
dependencies = [
"arc-swap",
"base64 0.22.1",
"base64 0.23.1",
"iota-logger",
"iota-paths",
"iota-util",
@ -2283,7 +2283,7 @@ dependencies = [
name = "iota-util"
version = "0.1.0"
dependencies = [
"base64 0.22.1",
"base64 0.23.1",
"hex",
"iota-paths",
"mtp",
@ -2996,7 +2996,7 @@ name = "omikron-connector"
version = "0.1.0"
dependencies = [
"async-trait",
"base64 0.22.1",
"base64 0.23.1",
"dashmap",
"iota-connection",
"iota-logger",

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,

2
iota-storage/Cargo.toml
View file

@ -7,7 +7,7 @@ edition = "2024"
iota-logger = { path = "../iota-logger" }
iota-util = { path = "../iota-util" }
iota-paths = { path = "../iota-paths" }
base64 = "0.22.1"
base64 = "0.23.0"
json = "*"
arc-swap = "1"
once_cell = "1.21.3"

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
iota-util/Cargo.toml
View file

@ -15,7 +15,7 @@ sysinfo = "0.38.0"
uuid = { version = "*", features = ["v4"] }
walkdir = "2.5.0"
zip = "6.0.0"
base64 = "0.22.1"
base64 = "0.23.0"
hex = "*"
[dev-dependencies]

2
mtp-type-maps

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

3
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"
@ -22,5 +23,5 @@ reqwest = "0.13.2"
tokio = { version = "1.50.0", features = ["full"] }
tokio-util = { version = "0.7", features = ["rt"] }
uuid = { version = "*", features = ["v4"] }
base64 = "0.22.1"
base64 = "0.23.0"
rand_core = { version = "0.6", features = ["getrandom", "std"] }

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