diff --git a/Cargo.lock b/Cargo.lock index 75af5ff..baa33b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2167,6 +2167,7 @@ dependencies = [ "libc", "mtp", "omikron-connector", + "serde_json", "serde_yaml", "sysinfo", "tempfile", @@ -2218,6 +2219,8 @@ name = "iota-process-manager" version = "0.1.0" dependencies = [ "async-trait", + "libc", + "tempfile", "tokio", ] diff --git a/README.md b/README.md index aee09ab..5a8c16a 100644 --- a/README.md +++ b/README.md @@ -52,8 +52,12 @@ document without accepting it. # Linux daemon installation The system-managed daemon runs as the dedicated `iota` account and listens on -`/run/iota/iota.sock` through socket activation. Operator access is granted -through the `iota-operators` group. After installing, add an account with: +`/run/iota/iota.sock` through socket activation. The system IPC socket is the +privilege boundary. Operator access is granted through the `iota-operators` +group, and every account admitted through that socket is authorized for the +full operator-console role, including user management, identity rotation, +configuration, and daemon lifecycle commands. After installing, add an +account with: ```text usermod -aG iota-operators USER diff --git a/client/src/client_connection.rs b/client/src/client_connection.rs index a207471..0dda93b 100644 --- a/client/src/client_connection.rs +++ b/client/src/client_connection.rs @@ -140,7 +140,11 @@ impl ClientConnection { return; } - let _msg_id = cv.get_id(); + if cv.require_id().is_err() { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } if cv.is_type(CommunicationType::Challenge) { self.handle_challenge(&cv).await; @@ -154,7 +158,14 @@ impl ClientConnection { } if cv.is_type(CommunicationType::SaveAppData) { - let sender_id = cv.get_sender(); + let sender_id = match cv.require_sender() { + Ok(sender_id) => sender_id, + Err(_) => { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } + }; let _app_data = cv .get_data(DataType::AppData) .as_str() @@ -162,18 +173,25 @@ impl ClientConnection { .to_string(); let res = CommunicationValue::new(CommunicationType::SaveAppData) - .with_id(cv.get_id()) + .with_request_id(&cv) .with_receiver(sender_id); self.send_message(&res).await; return; } if cv.is_type(CommunicationType::LoadAppData) { - let sender_id = cv.get_sender(); + let sender_id = match cv.require_sender() { + Ok(sender_id) => sender_id, + Err(_) => { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } + }; let app_data = String::new(); let res = CommunicationValue::new(CommunicationType::LoadAppData) - .with_id(cv.get_id()) + .with_request_id(&cv) .with_receiver(sender_id) .add_typed_default(DataType::AppData, DataValue::Str(app_data)); self.send_message(&res).await; @@ -287,12 +305,32 @@ impl ClientConnection { } if cv.is_type(CommunicationType::SettingsSave) { - let my_id = cv.get_sender(); - let Some(settings_name) = cv.get_data(DataType::SettingsName).as_str() else { return }; - let Some(settings_value) = cv.get_data(DataType::Payload).as_str() else { return }; + let my_id = match cv.require_sender() { + Ok(my_id) => my_id, + Err(_) => { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } + }; + let Ok(my_id_i64) = i64::try_from(my_id) else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + }; + let Some(settings_name) = cv.get_data(DataType::SettingsName).as_str() else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + }; + let Some(settings_value) = cv.get_data(DataType::Payload).as_str() else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + }; let _ = iota_storage::util::settings::save( - my_id as i64, + my_id_i64, iota_storage::util::settings::GLOBAL_SESSION_ID, settings_name, settings_value, @@ -300,17 +338,33 @@ impl ClientConnection { let response = CommunicationValue::new(CommunicationType::SettingsSave) .with_receiver(my_id) - .with_id(cv.get_id()); + .with_request_id(&cv); self.send_message(&response).await; return; } if cv.is_type(CommunicationType::SettingsLoad) { - let my_id = cv.get_sender(); - let Some(settings_name) = cv.get_data(DataType::SettingsName).as_string() else { return }; + let my_id = match cv.require_sender() { + Ok(my_id) => my_id, + Err(_) => { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } + }; + let Ok(my_id_i64) = i64::try_from(my_id) else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + }; + let Some(settings_name) = cv.get_data(DataType::SettingsName).as_string() else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + }; let settings_value_str = iota_storage::util::settings::load( - my_id as i64, + my_id_i64, iota_storage::util::settings::GLOBAL_SESSION_ID, &settings_name, ) @@ -318,7 +372,7 @@ impl ClientConnection { .flatten() .unwrap_or_default(); let response = CommunicationValue::new(CommunicationType::SettingsLoad) - .with_id(cv.get_id()) + .with_request_id(&cv) .with_receiver(my_id) .add_typed_default(DataType::Payload, DataValue::Str(settings_value_str)) .add_typed_default(DataType::SettingsName, DataValue::Str(settings_name)); @@ -328,15 +382,27 @@ impl ClientConnection { } if cv.is_type(CommunicationType::SettingsList) { - let my_id = cv.get_sender(); + let my_id = match cv.require_sender() { + Ok(my_id) => my_id, + Err(_) => { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } + }; + let Ok(my_id_i64) = i64::try_from(my_id) else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + }; let settings = iota_storage::util::settings::list( - my_id as i64, + my_id_i64, iota_storage::util::settings::GLOBAL_SESSION_ID, ) .unwrap_or_default(); let settings_json = settings.into_iter().map(DataValue::Str).collect(); let response = CommunicationValue::new(CommunicationType::SettingsList) - .with_id(cv.get_id()) + .with_request_id(&cv) .with_receiver(my_id) .add_typed_default(DataType::Settings, DataValue::Array(settings_json)); @@ -356,7 +422,7 @@ impl ClientConnection { if let Some(solved) = solved { let response = CommunicationValue::new(CommunicationType::ChallengeResponse) - .with_id(cv.get_id()) + .with_request_id(&cv) .add_typed_default(DataType::Challenge, DataValue::Str(solved)); self.send_message(&response).await; @@ -405,7 +471,9 @@ impl ClientConnection { timeout_duration: Option, ) -> Result { let (tx, mut rx) = mpsc::channel(1); - let msg_id = cv.get_id(); + let msg_id = cv + .require_id() + .map_err(|error| format!("cannot await response without a message id: {error}"))?; let task_tx = tx.clone(); self.waiting_tasks.insert( diff --git a/communities/src/community_connection.rs b/communities/src/community_connection.rs index b607a28..ff0aff2 100644 --- a/communities/src/community_connection.rs +++ b/communities/src/community_connection.rs @@ -2,7 +2,6 @@ use crate::auth::auth_user::AuthUser; use crate::communities::community::Community; use crate::communities::interactables::interactable::Interactable; use crate::users::user_manager::get_user; -use iota_util::mtp_compat::CommunicationValueCompat; use aes_gcm::{Aes256Gcm, KeyInit, Nonce, aead::Aead}; use base64::{Engine as _, engine::general_purpose::STANDARD}; use futures::SinkExt; @@ -22,6 +21,20 @@ use tungstenite::Message; use tungstenite::Utf8Bytes; use uuid::Uuid; use x448::PublicKey; + +trait CommunicationResponseExt { + fn with_request_id(self, request: &CommunicationValue) -> Self; +} + +impl CommunicationResponseExt for CommunicationValue { + fn with_request_id(mut self, request: &CommunicationValue) -> Self { + self = self.without_id(); + if let Some(id) = request.id() { + self = self.with_id(id); + } + self + } +} pub struct CommunityConnection { pub sender: Arc>, Message>>>, pub receiver: Arc>>>>, @@ -118,7 +131,7 @@ impl CommunityConnection { .unwrap_or(0); let Some(user) = get_user(user_id) else { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidUserId) + self.send_error_response(&cv, CommunicationType::ErrorInvalidUserId) .await; return; }; @@ -148,7 +161,7 @@ impl CommunityConnection { let user_public_key_bytes = match STANDARD.decode(&user.public_key) { Ok(bytes) => bytes, Err(_) => { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidUserId) + self.send_error_response(&cv, CommunicationType::ErrorInvalidUserId) .await; return; } @@ -157,14 +170,14 @@ impl CommunityConnection { let user_pub_key: PublicKey = match PublicKey::from_bytes(&user_public_key_bytes) { Some(key) => key, __ => { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidUserId) + self.send_error_response(&cv, CommunicationType::ErrorInvalidUserId) .await; return; } }; let Some(community) = self.community.read().await.clone() else { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal) + self.send_error_response(&cv, CommunicationType::ErrorInternal) .await; return; }; @@ -175,7 +188,7 @@ impl CommunityConnection { let shared_secret = match community_private_key.to_diffie_hellman(&user_pub_key) { Some(secret) => secret, _ => { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal) + self.send_error_response(&cv, CommunicationType::ErrorInternal) .await; return; } @@ -197,7 +210,7 @@ impl CommunityConnection { let encrypted_challenge = match cipher.encrypt(nonce, challenge_str.as_bytes()) { Ok(data) => data, Err(_) => { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal) + self.send_error_response(&cv, CommunicationType::ErrorInternal) .await; return; } @@ -212,7 +225,7 @@ impl CommunityConnection { STANDARD.encode(community_public_key.as_bytes()), ) .add_data_str(DataType::Challenge, STANDARD.encode(&encrypted_out)) - .with_id(cv.get_id()); + .with_request_id(&cv); self.send_message(&response).await; } @@ -220,7 +233,7 @@ impl CommunityConnection { let client_challenge_response_b64 = match cv.get_data(DataType::Challenge) { Some(data) => data.to_string(), _ => { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidData) + self.send_error_response(&cv, CommunicationType::ErrorInvalidData) .await; return; } @@ -229,38 +242,38 @@ impl CommunityConnection { let challenge_response_bytes = match STANDARD.decode(&client_challenge_response_b64) { Ok(bytes) => bytes, Err(_) => { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidData) + self.send_error_response(&cv, CommunicationType::ErrorInvalidData) .await; return; } }; if challenge_response_bytes.len() < 12 { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidData) + self.send_error_response(&cv, CommunicationType::ErrorInvalidData) .await; return; } let Some(user) = self.auth.read().await.clone() else { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal) + self.send_error_response(&cv, CommunicationType::ErrorInternal) .await; return; }; let Some(user_pub_bytes) = STANDARD.decode(&user.public_key).ok() else { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidData) + self.send_error_response(&cv, CommunicationType::ErrorInvalidData) .await; return; }; let Some(user_pub_key) = PublicKey::from_bytes(&user_pub_bytes) else { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidPublicKey) + self.send_error_response(&cv, CommunicationType::ErrorInvalidPublicKey) .await; return; }; let Some(community) = self.community.read().await.clone() else { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal) + self.send_error_response(&cv, CommunicationType::ErrorInternal) .await; return; }; @@ -270,7 +283,7 @@ impl CommunityConnection { let shared_secret = match community_private_key.to_diffie_hellman(&user_pub_key) { Some(secret) => secret, _ => { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal) + self.send_error_response(&cv, CommunicationType::ErrorInternal) .await; return; } @@ -294,7 +307,7 @@ impl CommunityConnection { let decrypted_bytes = match cipher.decrypt(nonce, ciphertext) { Ok(pt) => pt, Err(_) => { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidChallenge) + self.send_error_response(&cv, CommunicationType::ErrorInvalidChallenge) .await; return; } @@ -303,7 +316,7 @@ impl CommunityConnection { let client_response = match String::from_utf8(decrypted_bytes) { Ok(str) => str, Err(_) => { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidData) + self.send_error_response(&cv, CommunicationType::ErrorInvalidData) .await; return; } @@ -312,7 +325,7 @@ impl CommunityConnection { let expected_challenge = self.challenge.read().await.clone(); if client_response != expected_challenge { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidChallenge) + self.send_error_response(&cv, CommunicationType::ErrorInvalidChallenge) .await; self.close().await; return; @@ -324,7 +337,7 @@ impl CommunityConnection { } let Some(community) = self.community.read().await.clone() else { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal) + self.send_error_response(&cv, CommunicationType::ErrorInternal) .await; return; }; @@ -332,7 +345,7 @@ impl CommunityConnection { let user_id = self.get_user_id().await; if user_id == 0 { - self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidUserId) + self.send_error_response(&cv, CommunicationType::ErrorInvalidUserId) .await; return; } @@ -351,13 +364,17 @@ impl CommunityConnection { } c }) - .with_id(cv.get_id()); + .with_request_id(&cv); self.send_message(&response).await; } - async fn send_error_response(&self, message_id: &Uuid, error_type: CommunicationType) { - let error = CommunicationValue::new(error_type).with_id(*message_id); + async fn send_error_response( + &self, + request: &CommunicationValue, + error_type: CommunicationType, + ) { + let error = CommunicationValue::new(error_type).with_request_id(request); self.send_message(&error).await; } pub async fn close(&self) { diff --git a/communities/src/interactables/text_chat.rs b/communities/src/interactables/text_chat.rs index 6b69baa..780d8ed 100644 --- a/communities/src/interactables/text_chat.rs +++ b/communities/src/interactables/text_chat.rs @@ -8,7 +8,7 @@ use crate::{ }; use async_trait::async_trait; use json::{JsonValue, array, object}; -use iota_util::mtp_compat::{CommunicationValueCompat, OptionalDataValueExt}; +use iota_util::mtp_compat::{OptionalDataValueExt, RequiredCommunicationFields}; use std::fs; use std::path::Path; use std::sync::Arc; @@ -31,6 +31,10 @@ impl TextChat { } } pub fn add_message(&self, send_time: u128, sender: i64, message: &str) { + let Ok(send_time) = i64::try_from(send_time) else { + log!("Message timestamp exceeds local storage range"); + return; + }; let user_dir = &format!( "communities/{}/interactables/{}/{}", self.get_community().get_name(), @@ -74,7 +78,7 @@ impl TextChat { } let json_obj = object! { - "timestamp" => send_time as i64, + "timestamp" => send_time, "content" => message, "sender" => sender.to_string(), }; @@ -202,7 +206,7 @@ impl Interactable for TextChat { let mut payload = JsonValue::new_object(); payload["messages"] = messages; return CommunicationValue::new(CommunicationType::Function) - .with_id(cv.get_id()) + .with_request_id(&cv) .add_data_str(DataType::Name, self.name.clone()) .add_data_str(DataType::Path, self.path.clone()) .add_data_str(DataType::Result, "message_chunk".to_string()) @@ -210,19 +214,28 @@ impl Interactable for TextChat { } if cv.get_data(DataType::Function).unwrap().as_str().unwrap() == "send_message" { let message = payload["message"].as_str().unwrap(); + let sender = match cv.require_sender() { + Ok(sender) => sender, + Err(_) => return CommunicationValue::new(CommunicationType::ErrorInvalidData) + .with_request_id(&cv), + }; let milliseconds_timestamp: u128 = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_millis(); - self.add_message(milliseconds_timestamp, cv.get_sender(), message); + let Ok(sender) = i64::try_from(sender) else { + return CommunicationValue::new(CommunicationType::ErrorInvalidData) + .with_request_id(&cv); + }; + self.add_message(milliseconds_timestamp, sender, message); let mut distribution_payload = JsonValue::new_object(); distribution_payload["message"] = JsonValue::String(message.to_string()); - distribution_payload["sender_id"] = JsonValue::String(cv.get_sender().to_string()); + distribution_payload["sender_id"] = JsonValue::String(sender.to_string()); distribution_payload["send_time"] = JsonValue::String(milliseconds_timestamp.to_string()); let distribution = CommunicationValue::new(CommunicationType::Update) - .with_id(cv.get_id()) + .with_request_id(&cv) .add_data_str(DataType::Name, self.name.clone()) .add_data_str(DataType::Path, self.path.clone()) .add_data_str(DataType::Result, "message_live".to_string()) @@ -238,13 +251,13 @@ impl Interactable for TextChat { } } return CommunicationValue::new(CommunicationType::Function) - .with_id(cv.get_id()) + .with_request_id(&cv) .add_data_str(DataType::Name, self.name.clone()) .add_data_str(DataType::Path, self.path.clone()) .add_data_str(DataType::Result, "message_received".to_string()) .add_data(DataType::Payload, JsonValue::new_object()); } - CommunicationValue::new(CommunicationType::ErrorInternal).with_id(cv.get_id()) + CommunicationValue::new(CommunicationType::ErrorInternal).with_request_id(&cv) } fn to_json(&self) -> JsonValue { JsonValue::new_object() diff --git a/communities/src/interactables/voice_chat.rs b/communities/src/interactables/voice_chat.rs index 7e20022..7681765 100644 --- a/communities/src/interactables/voice_chat.rs +++ b/communities/src/interactables/voice_chat.rs @@ -1,7 +1,7 @@ use crate::communities::{community::Community, interactables::interactable::Interactable}; use async_trait::async_trait; use json::JsonValue; -use iota_util::mtp_compat::{CommunicationValueCompat, OptionalDataValueExt}; +use iota_util::mtp_compat::OptionalDataValueExt; use std::sync::Arc; use std::{any::Any, sync::RwLock}; use uuid::Uuid; @@ -131,7 +131,7 @@ impl Interactable for VoiceChat { response_payload["send_time"] = JsonValue::String(send_time.to_string()); return CommunicationValue::new(CommunicationType::Function) - .with_id(cv.get_id()) + .with_request_id(&cv) .add_data_str(DataType::Name, self.name.clone()) .add_data_str(DataType::Path, self.path.clone()) .add_data_str(DataType::Result, "getting_call".to_string()) @@ -159,13 +159,13 @@ impl Interactable for VoiceChat { response_payload["streaming"] = JsonValue::Boolean(streaming); return CommunicationValue::new(CommunicationType::Update) - .with_id(cv.get_id()) + .with_request_id(&cv) .add_data_str(DataType::Name, self.name.clone()) .add_data_str(DataType::Path, self.path.clone()) .add_data_str(DataType::Result, "user_changed".to_string()) .add_data(DataType::Payload, response_payload); } - CommunicationValue::new(CommunicationType::ErrorInternal).with_id(cv.get_id()) + CommunicationValue::new(CommunicationType::ErrorInternal).with_request_id(&cv) } fn to_json(&self) -> JsonValue { diff --git a/flake.nix b/flake.nix index e2ca8a1..5d64dc6 100644 --- a/flake.nix +++ b/flake.nix @@ -140,6 +140,12 @@ 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; @@ -261,6 +267,9 @@ } // lib.optionalAttrs (cfg.environmentFiles != []) { EnvironmentFile = cfg.environmentFiles; + } + // lib.optionalAttrs (cfg.identitySecretFile != null) { + LoadCredential = "iota-identity:${cfg.identitySecretFile}"; }; }; diff --git a/iota-cli/src/ipc_client.rs b/iota-cli/src/ipc_client.rs index ca912fc..45ca8fc 100644 --- a/iota-cli/src/ipc_client.rs +++ b/iota-cli/src/ipc_client.rs @@ -572,7 +572,7 @@ impl IpcClient { iota_ipc::IpcErrorCode::Timeout => "The daemon request timed out.", iota_ipc::IpcErrorCode::Cancelled => "The daemon request was cancelled.", iota_ipc::IpcErrorCode::Unauthorized => { - "The daemon rejected this operation as unauthorized." + "The daemon rejected this operation: the connected IPC account lacks the required role. Use the configured operator socket or ask an administrator to grant access." } iota_ipc::IpcErrorCode::InternalFailure => "The daemon reported an internal failure.", } diff --git a/iota-connection/src/message_common.rs b/iota-connection/src/message_common.rs index fff5bf6..742d9b6 100644 --- a/iota-connection/src/message_common.rs +++ b/iota-connection/src/message_common.rs @@ -2,7 +2,21 @@ use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use mtp::type_map::TypeMap; use std::time::{SystemTime, UNIX_EPOCH}; -pub use iota_util::mtp_compat::{CommunicationValueCompat, OptionalDataValueExt}; +pub use iota_util::mtp_compat::{MtpFieldError, OptionalDataValueExt, RequiredCommunicationFields}; + +pub trait CommunicationResponseExt { + fn with_request_id(self, request: &CommunicationValue) -> Self; +} + +impl CommunicationResponseExt for CommunicationValue { + fn with_request_id(mut self, request: &CommunicationValue) -> Self { + self = self.without_id(); + if let Some(id) = request.id() { + self = self.with_id(id); + } + self + } +} pub fn typed_container(items: Vec<(DataType, DataValue)>) -> DataValue { use mtp::type_map::{DataTypeId, TypeMap}; @@ -95,16 +109,48 @@ pub fn chat_secret_recipients(cv: &CommunicationValue) -> Option i64 { - SystemTime::now() + let millis = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() - .as_millis() as i64 + .as_millis(); + i64::try_from(millis).unwrap_or(i64::MAX) } pub fn error_response(request: &CommunicationValue, ty: CommunicationType) -> CommunicationValue { - let mut response = CommunicationValue::new(ty).with_id(request.id().unwrap_or_default()); + let mut response = CommunicationValue::new(ty).without_id(); + if let Some(id) = request.id() { + response = response.with_id(id); + } if let Some(sender) = request.sender() { response = response.with_receiver(sender); } response } + +#[cfg(test)] +mod tests { + use super::error_response; + use mtp::codec::{CommunicationType, CommunicationValue}; + + #[test] + fn error_response_preserves_an_absent_request_id() { + let request = CommunicationValue::new(CommunicationType::GetChats) + .without_id() + .with_sender(42); + let response = error_response(&request, CommunicationType::ErrorInvalidData); + + assert_eq!(response.id(), None); + assert_eq!(response.receiver(), Some(42)); + } + + #[test] + fn error_response_copies_an_existing_request_id() { + let request = CommunicationValue::new(CommunicationType::GetChats) + .with_id(7) + .with_sender(42); + let response = error_response(&request, CommunicationType::ErrorInvalidData); + + assert_eq!(response.id(), Some(7)); + assert_eq!(response.receiver(), Some(42)); + } +} diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index 4d7144f..4be17f6 100644 --- a/iota-connection/src/message_handlers.rs +++ b/iota-connection/src/message_handlers.rs @@ -10,15 +10,26 @@ use mtp::codec::{ use crate::relay::VerifiedRelayContext; +#[derive(Debug)] pub struct MessageMutation { pub sender_id: i64, pub partner_id: i64, pub send_time: i64, } -pub fn message_mutation(cv: &CommunicationValue) -> Result { - let sender_id = i64::try_from(cv.get_sender()) +fn required_sender_id(cv: &CommunicationValue) -> Result { + let sender = cv + .require_sender() .map_err(|_| error_response(cv, CommunicationType::ErrorInvalidData))?; + i64::try_from(sender).map_err(|_| error_response(cv, CommunicationType::ErrorInvalidData)) +} + +fn sender_wire_id(sender_id: i64) -> u64 { + u64::try_from(sender_id).expect("validated authenticated sender is non-negative") +} + +pub fn message_mutation(cv: &CommunicationValue) -> Result { + let sender_id = required_sender_id(cv)?; let partner_id = data_i64(cv, DataType::ChatPartnerId) .filter(|id| *id > 0) .ok_or_else(|| error_response(cv, CommunicationType::ErrorInvalidData))?; @@ -115,7 +126,7 @@ pub fn apply_verified_relay_content( match content.message_type.as_str() { "MessageSend" => { - let message = relay_string(&content.content, DataType::Content, &context.type_map) + let message = relay_string(&content.content, DataType::AppContent, &context.type_map) .ok_or_else(|| "Relay MessageSend is missing Content".to_string())?; let send_time = relay_number(&content.content, DataType::SendTime, &context.type_map) .and_then(|value| i64::try_from(value).ok()) @@ -138,7 +149,7 @@ pub fn apply_verified_relay_content( Ok(()) } "MessageEdit" => { - let message = relay_string(&content.content, DataType::Content, &context.type_map) + let message = relay_string(&content.content, DataType::AppContent, &context.type_map) .ok_or_else(|| "Relay MessageEdit is missing Content".to_string())?; let send_time = relay_number(&content.content, DataType::SendTime, &context.type_map) .and_then(|value| i64::try_from(value).ok()) @@ -206,7 +217,7 @@ pub fn handle_message_edit(cv: &CommunicationValue) -> CommunicationValue { Ok(mutation) => mutation, Err(response) => return response, }; - let Some(content) = cv.get_data(DataType::Content).as_str() else { + let Some(content) = cv.get_data(DataType::AppContent).as_str() else { return error_response(cv, CommunicationType::ErrorInvalidData); }; @@ -277,14 +288,17 @@ fn stored_message_fields( ) -> Vec<(DataType, DataValue)> { let mut fields = vec![ ( - DataType::MessageId, + DataType::AppMessageId, DataValue::SignedNumber(message.id as i128), ), ( DataType::SendTime, DataValue::SignedNumber(message.message_time as i128), ), - (DataType::Content, DataValue::Str(message.content.clone())), + ( + DataType::AppContent, + DataValue::Str(message.content.clone()), + ), ( DataType::MessageState, DataValue::Str(message.message_state.clone()), @@ -293,22 +307,22 @@ fn stored_message_fields( DataType::Height, DataValue::SignedNumber(message.height as i128), ), - ( - DataType::SenderId, - DataValue::UnsignedNumber(if message.sent_by_self { - storage_owner as u128 - } else { - partner_id as u128 - }), - ), ]; + let sender_id = if message.sent_by_self { + storage_owner + } else { + partner_id + }; + if let Ok(sender_id) = u128::try_from(sender_id) { + fields.push((DataType::SenderId, DataValue::UnsignedNumber(sender_id))); + } if message.edited { fields.push((DataType::Edited, DataValue::Bool(true))); } - if let Some(reply_to) = message.reply_to { + if let Some(reply_to) = message.reply_to.and_then(|id| u64::try_from(id).ok()) { fields.push(( DataType::ReplyId, - DataValue::UnsignedNumber(reply_to as u64 as u128), + DataValue::UnsignedNumber(u128::from(reply_to)), )); } if !message.reactions.is_empty() { @@ -345,7 +359,11 @@ 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); }; - if user_id != cv.get_sender().to_string() { + let sender_id = match required_sender_id(cv) { + Ok(sender_id) => sender_id, + Err(response) => return response, + }; + if user_id != sender_id.to_string() { return error_response(cv, CommunicationType::ErrorNotFound); } let Some(chat_id) = data_string(cv, DataType::ChatId) else { @@ -358,8 +376,8 @@ pub fn handle_get_chat_secret(cv: &CommunicationValue) -> CommunicationValue { secret_id: data_string(cv, DataType::SecretId), }) { Ok(Some(record)) => CommunicationValue::new(CommunicationType::ChatSecretResponse) - .with_id(cv.get_id()) - .with_receiver(cv.get_sender()) + .with_request_id(cv) + .with_receiver(sender_wire_id(sender_id)) .add_typed_default(DataType::UserId, DataValue::Str(record.user_id)) .add_typed_default(DataType::ChatId, DataValue::Str(record.chat_id)) .add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id)) @@ -380,7 +398,7 @@ pub fn handle_get_chat_secret(cv: &CommunicationValue) -> CommunicationValue { DataValue::Str(record.wrapping_scheme), ) .add_typed_default( - DataType::CreatedAt, + DataType::AppCreatedAt, DataValue::SignedNumber(record.created_at as i128), ) .add_typed_default( @@ -393,7 +411,10 @@ pub fn handle_get_chat_secret(cv: &CommunicationValue) -> CommunicationValue { } pub fn handle_create_app(cv: &CommunicationValue) -> CommunicationValue { - let sender_id = cv.get_sender() as i64; + let sender_id = match required_sender_id(cv) { + Ok(sender_id) => sender_id, + Err(response) => return response, + }; let app_identifier = cv .get_data(DataType::AppIdentifier) .as_str() @@ -415,12 +436,15 @@ pub fn handle_create_app(cv: &CommunicationValue) -> CommunicationValue { } CommunicationValue::new(CommunicationType::CreateApp) - .with_id(cv.get_id()) - .with_receiver(sender_id as u64) + .with_request_id(cv) + .with_receiver(sender_wire_id(sender_id)) } pub fn handle_delete_app(cv: &CommunicationValue) -> CommunicationValue { - let sender_id = cv.get_sender() as i64; + let sender_id = match required_sender_id(cv) { + Ok(sender_id) => sender_id, + Err(response) => return response, + }; let app_identifier = cv .get_data(DataType::AppIdentifier) .as_str() @@ -437,8 +461,8 @@ pub fn handle_delete_app(cv: &CommunicationValue) -> CommunicationValue { } CommunicationValue::new(CommunicationType::DeleteApp) - .with_id(cv.get_id()) - .with_receiver(sender_id as u64) + .with_request_id(cv) + .with_receiver(sender_wire_id(sender_id)) } fn contact_value( @@ -495,8 +519,8 @@ fn contact_ids_value(ids: impl IntoIterator) -> DataValue { #[cfg(test)] mod presence_tests { - use super::contact_ids_value; - use mtp::codec::DataValue; + use super::{contact_ids_value, handle_get_chats, message_mutation}; + use mtp::codec::{CommunicationType, CommunicationValue, DataValue}; #[test] fn contact_snapshot_is_sorted_and_deduplicated() { @@ -509,6 +533,26 @@ mod presence_tests { ]) ); } + + #[test] + fn message_mutation_rejects_a_missing_authenticated_sender() { + let request = CommunicationValue::new(CommunicationType::MessageEdit).with_id(11); + let response = message_mutation(&request).expect_err("missing sender must be rejected"); + + assert!(response.is_type(CommunicationType::ErrorInvalidData)); + assert_eq!(response.id(), Some(11)); + assert_eq!(response.receiver(), None); + } + + #[test] + fn read_handler_rejects_a_missing_authenticated_sender() { + let request = CommunicationValue::new(CommunicationType::GetChats).with_id(12); + let response = handle_get_chats(&request); + + assert!(response.is_type(CommunicationType::ErrorInvalidData)); + assert_eq!(response.id(), Some(12)); + assert_eq!(response.receiver(), None); + } } fn sync_error(cv: &CommunicationValue) -> CommunicationValue { @@ -523,7 +567,7 @@ fn sync_error(cv: &CommunicationValue) -> CommunicationValue { /// The sender is authenticated by MTP; a UserId embedded by a client is never trusted here. pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { use iota_storage::util::sync::{self, CACHE_SCHEMA_VERSION}; - let user_id = match i64::try_from(cv.get_sender()) { + let user_id = match required_sender_id(cv) { Ok(id) if id > 0 => id, _ => return sync_error(cv), }; @@ -578,8 +622,8 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { .map(|message| stored_message_value(message, user_id, message.external_user)) .collect(); CommunicationValue::new(CommunicationType::ClientStateSync) - .with_id(cv.get_id()) - .with_receiver(cv.get_sender()) + .with_request_id(cv) + .with_receiver(sender_wire_id(user_id)) .add_typed_default( DataType::SessionId, DataValue::SignedNumber(session_id as i128), @@ -603,6 +647,10 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { ), ) .add_typed_default(DataType::Messages, DataValue::Array(message_values)) + .add_typed_default( + DataType::Communities, + DataValue::Array(community_values(user_id)), + ) .add_typed_default( DataType::DeletedMessageIds, DataValue::Array( @@ -627,7 +675,7 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { pub fn handle_client_state_ack(cv: &CommunicationValue) -> CommunicationValue { use iota_storage::util::sync::{self, CACHE_SCHEMA_VERSION}; - let user_id = match i64::try_from(cv.get_sender()) { + let user_id = match required_sender_id(cv) { Ok(id) if id > 0 => id, _ => return sync_error(cv), }; @@ -654,46 +702,50 @@ pub fn handle_client_state_ack(cv: &CommunicationValue) -> CommunicationValue { } pub fn handle_message_state(cv: &CommunicationValue) { - let sender_id = &cv.get_sender(); - let receiver_id = match cv.get_data(DataType::ChatPartnerId).as_number() { - Some(id) => id, + let sender_id = match required_sender_id(cv) { + Ok(sender_id) => sender_id, + Err(_) => return, + }; + let receiver_id = match data_i64(cv, DataType::ChatPartnerId) { + Some(id) if id > 0 => id, _ => return, }; - let timestamp_i64 = if let Some(n) = cv.get_data(DataType::SendTime).as_number() { - n as i64 - } else if let Some(s) = cv.get_data(DataType::SendTime).as_str() { - s.parse::().unwrap_or_else(|_| now_millis_i64()) - } else { - now_millis_i64() - }; + let timestamp_i64 = data_i64(cv, DataType::SendTime).unwrap_or_else(now_millis_i64); let _ = chat_files::change_message_state( timestamp_i64, - receiver_id as i64, - *sender_id as i64, + receiver_id, + sender_id, MessageState::from_str(cv.get_data(DataType::MessageState).as_str().unwrap_or("")), ); } pub fn handle_messages_get(cv: &CommunicationValue) -> CommunicationValue { - let my_id = cv.get_sender(); - let partner_id = cv.get_data(DataType::UserId).as_number().unwrap_or(0); - let offset = cv.get_data(DataType::Offset).as_number().unwrap_or(0); - let amount = cv.get_data(DataType::Amount).as_number().unwrap_or(0); - let messages = chat_files::get_messages( - my_id as i64, - partner_id as i64, - offset as i64, - amount as i64, - ); + let my_id = match cv.require_sender() { + Ok(my_id) => my_id, + Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData), + }; + let Ok(my_id_i64) = i64::try_from(my_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let Some(partner_id) = data_i64(cv, DataType::UserId).filter(|id| *id > 0) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let Some(offset) = data_i64(cv, DataType::Offset).filter(|offset| *offset >= 0) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let Some(amount) = data_i64(cv, DataType::Amount).filter(|amount| *amount > 0) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let messages = chat_files::get_messages(my_id_i64, partner_id, offset, amount); let mut msg_array: Vec = Vec::new(); for m in &messages { - msg_array.push(stored_message_value(m, my_id as i64, partner_id as i64)); + msg_array.push(stored_message_value(m, my_id_i64, partner_id)); } CommunicationValue::new(CommunicationType::MessagesGet) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default(DataType::Messages, DataValue::Array(msg_array)) } @@ -703,7 +755,10 @@ pub fn handle_message_get(cv: &CommunicationValue) -> CommunicationValue { return error_response(cv, CommunicationType::ErrorInvalidData); }; let partner_id = data_i64(cv, DataType::ChatPartnerId); - let owner = cv.get_sender() as i64; + let owner = match required_sender_id(cv) { + Ok(owner) => owner, + Err(response) => return response, + }; let message = match chat_files::get_message(owner, send_time, partner_id) { Ok(Some(message)) => message, @@ -712,8 +767,8 @@ pub fn handle_message_get(cv: &CommunicationValue) -> CommunicationValue { }; let mut response = CommunicationValue::new(CommunicationType::MessageGet) - .with_id(cv.get_id()) - .with_receiver(cv.get_sender()); + .with_request_id(cv) + .with_receiver(u64::try_from(owner).expect("authenticated sender is non-negative")); for (data_type, value) in stored_message_fields(&message, owner, message.external_user) { response = response.add_typed_default(data_type, value); } @@ -721,8 +776,14 @@ pub fn handle_message_get(cv: &CommunicationValue) -> CommunicationValue { } pub fn handle_get_chats(cv: &CommunicationValue) -> CommunicationValue { - let user_id = cv.get_sender(); - let users = chats_util::get_users(user_id as i64); + let user_id = match cv.require_sender() { + Ok(user_id) => user_id, + Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData), + }; + let Ok(user_id_i64) = i64::try_from(user_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let users = chats_util::get_users(user_id_i64); let mut user_array = Vec::new(); for user in users { let mut container = Vec::new(); @@ -739,27 +800,28 @@ pub fn handle_get_chats(cv: &CommunicationValue) -> CommunicationValue { user_array.push(typed_container(container)); } CommunicationValue::new(CommunicationType::GetChats) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(user_id) .add_typed_default(DataType::UserIds, DataValue::Array(user_array)) } pub fn handle_add_conversation(cv: &CommunicationValue) -> CommunicationValue { - let user_id = cv.get_sender(); + let user_id = match cv.require_sender() { + Ok(user_id) => user_id, + Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData), + }; + let Ok(user_id_i64) = i64::try_from(user_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; let session_id = match data_i64(cv, DataType::SessionId) { Some(id) if id > 0 => id, _ => return sync_error(cv), }; - let other_id = match cv.get_data(DataType::ChatPartnerId).as_number() { - Some(n) => n as i64, - None => cv - .get_data(DataType::ChatPartnerId) - .as_str() - .unwrap_or("0") - .parse() - .unwrap_or(0), + let other_id = match data_i64(cv, DataType::ChatPartnerId) { + Some(id) if id > 0 => id, + _ => return error_response(cv, CommunicationType::ErrorInvalidData), }; - let mut contact = get_user(user_id as i64, other_id) + let mut contact = get_user(user_id_i64, other_id) .unwrap_or(iota_storage::users::contact::Contact::new(other_id)); if let Some(name) = cv.get_data(DataType::ChatPartnerName).as_str() { @@ -767,75 +829,97 @@ pub fn handle_add_conversation(cv: &CommunicationValue) -> CommunicationValue { } contact.set_last_message_at(now_millis_i64()); - mod_user(user_id as i64, &contact); + mod_user(user_id_i64, &contact); CommunicationValue::new(CommunicationType::AddConversation) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(user_id) .add_typed_default( DataType::SessionId, DataValue::SignedNumber(session_id as i128), ) - .add_typed_default(DataType::UserIds, current_contact_ids(user_id as i64)) + .add_typed_default(DataType::UserIds, current_contact_ids(user_id_i64)) } pub fn handle_add_community(cv: &CommunicationValue) -> CommunicationValue { + let sender_id = match required_sender_id(cv) { + Ok(sender_id) => sender_id, + Err(response) => return response, + }; + let Some(address) = cv.get_data(DataType::CommunityAddress).as_str() else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let Some(title) = cv.get_data(DataType::CommunityTitle).as_str() else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let Some(position) = cv.get_data(DataType::Position).as_str() else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; CommunitiesUtil::add_community( - cv.get_sender() as i64, - cv.get_data(DataType::CommunityAddress) - .as_str() - .unwrap() - .to_string(), - cv.get_data(DataType::CommunityTitle) - .as_str() - .unwrap() - .to_string(), - cv.get_data(DataType::Position) - .as_str() - .unwrap() - .to_string(), + sender_id, + address.to_string(), + title.to_string(), + position.to_string(), ); CommunicationValue::new(CommunicationType::AddCommunity) - .with_id(cv.get_id()) - .with_receiver(cv.get_sender()) + .with_request_id(cv) + .with_receiver(sender_wire_id(sender_id)) } pub fn handle_get_communities(cv: &CommunicationValue) -> CommunicationValue { - let mut comm_array = Vec::new(); - for c in CommunitiesUtil::get_communities(cv.get_sender() as i64) { - let mut container: Vec<(DataType, DataValue)> = Vec::new(); - container.push(( - DataType::CommunityAddress, - DataValue::Str(c.address.clone()), - )); - container.push((DataType::CommunityTitle, DataValue::Str(c.title.clone()))); - container.push((DataType::Position, DataValue::Str(c.position.clone()))); - comm_array.push(typed_container(container)); - } - + let sender_id = match required_sender_id(cv) { + Ok(sender_id) => sender_id, + Err(response) => return response, + }; CommunicationValue::new(CommunicationType::GetCommunities) - .with_id(cv.get_id()) - .with_receiver(cv.get_sender()) - .add_typed_default(DataType::Communities, DataValue::Array(comm_array)) + .with_request_id(cv) + .with_receiver(sender_wire_id(sender_id)) + .add_typed_default( + DataType::Communities, + DataValue::Array(community_values(sender_id)), + ) +} + +fn community_values(storage_owner: i64) -> Vec { + CommunitiesUtil::get_communities(storage_owner) + .into_iter() + .map(|community| { + typed_container(vec![ + ( + DataType::CommunityAddress, + DataValue::Str(community.address), + ), + (DataType::CommunityTitle, DataValue::Str(community.title)), + (DataType::Position, DataValue::Str(community.position)), + ]) + }) + .collect() } pub fn handle_remove_community(cv: &CommunicationValue) -> CommunicationValue { - CommunitiesUtil::remove_community( - cv.get_sender() as i64, - cv.get_data(DataType::CommunityAddress) - .as_str() - .unwrap() - .to_string(), - ); + let sender_id = match required_sender_id(cv) { + Ok(sender_id) => sender_id, + Err(response) => return response, + }; + 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()); CommunicationValue::new(CommunicationType::RemoveCommunity) - .with_id(cv.get_id()) - .with_receiver(cv.get_sender()) + .with_request_id(cv) + .with_receiver(sender_wire_id(sender_id)) } pub fn handle_global_settings_save(cv: &CommunicationValue) -> CommunicationValue { - let my_id = cv.get_sender(); + let my_id = match cv.require_sender() { + Ok(my_id) => my_id, + Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData), + }; + let Ok(my_id_i64) = i64::try_from(my_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; let Some(settings_value) = cv.get_data(DataType::Payload).as_str() else { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -843,13 +927,13 @@ pub fn handle_global_settings_save(cv: &CommunicationValue) -> CommunicationValu ); }; - if settings::save_global(my_id as i64, settings_value).is_err() { + if settings::save_global(my_id_i64, settings_value).is_err() { return error_response(cv, CommunicationType::ErrorInvalidData); } let mut response = CommunicationValue::new(CommunicationType::GlobalSettingsSave) .with_receiver(my_id) - .with_id(cv.get_id()); + .with_request_id(cv); if let Some(session_id) = cv.get_data(DataType::SessionId).as_number() { response = response.add_typed_default( @@ -862,13 +946,19 @@ pub fn handle_global_settings_save(cv: &CommunicationValue) -> CommunicationValu } pub fn handle_global_settings_load(cv: &CommunicationValue) -> CommunicationValue { - let my_id = cv.get_sender(); - let Ok(settings_value) = settings::load_global(my_id as i64) else { + let my_id = match cv.require_sender() { + Ok(my_id) => my_id, + Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData), + }; + let Ok(my_id_i64) = i64::try_from(my_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let Ok(settings_value) = settings::load_global(my_id_i64) else { return error_response(cv, CommunicationType::ErrorInvalidData); }; let Some(settings_value_str) = settings_value else { let mut response = CommunicationValue::new(CommunicationType::ErrorNotFound) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Path, @@ -886,7 +976,7 @@ pub fn handle_global_settings_load(cv: &CommunicationValue) -> CommunicationValu }; let mut response = CommunicationValue::new(CommunicationType::GlobalSettingsLoad) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default(DataType::Payload, DataValue::Str(settings_value_str)); @@ -904,10 +994,16 @@ pub fn handle_settings_save( cv: &CommunicationValue, _expected_session_id: i128, ) -> CommunicationValue { - let my_id = cv.get_sender(); + let my_id = match cv.require_sender() { + Ok(my_id) => my_id, + Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData), + }; + let Ok(my_id_i64) = i64::try_from(my_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; let Some(session_id) = cv.get_data(DataType::SessionId).as_number() else { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -916,7 +1012,7 @@ pub fn handle_settings_save( }; if session_id == 0 || session_id > 1_000_000 { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -929,7 +1025,7 @@ pub fn handle_settings_save( }; let Some(settings_name) = cv.get_data(DataType::SettingsName).as_str() else { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -942,7 +1038,7 @@ pub fn handle_settings_save( }; let Some(settings_value) = cv.get_data(DataType::Payload).as_str() else { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -960,7 +1056,7 @@ pub fn handle_settings_save( || settings_name.contains("..") { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -975,21 +1071,17 @@ pub fn handle_settings_save( DataValue::SignedNumber(session_id as i128), ); } + let Ok(session_id_i64) = i64::try_from(session_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; - if settings::save( - my_id as i64, - session_id as i64, - settings_name, - settings_value, - ) - .is_err() - { + if settings::save(my_id_i64, session_id_i64, settings_name, settings_value).is_err() { return error_response(cv, CommunicationType::ErrorInvalidData); } CommunicationValue::new(CommunicationType::SettingsSave) .with_receiver(my_id) - .with_id(cv.get_id()) + .with_request_id(cv) .add_typed_default( DataType::SettingsName, DataValue::Str(settings_name.to_string()), @@ -1004,10 +1096,16 @@ pub fn handle_settings_load( cv: &CommunicationValue, _expected_session_id: i128, ) -> CommunicationValue { - let my_id = cv.get_sender(); + let my_id = match cv.require_sender() { + Ok(my_id) => my_id, + Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData), + }; + let Ok(my_id_i64) = i64::try_from(my_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; let Some(session_id) = cv.get_data(DataType::SessionId).as_number() else { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -1016,7 +1114,7 @@ pub fn handle_settings_load( }; if session_id == 0 || session_id > 1_000_000 { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -1027,9 +1125,12 @@ pub fn handle_settings_load( DataValue::SignedNumber(session_id as i128), ); } + let Ok(session_id_i64) = i64::try_from(session_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; let Some(settings_name) = cv.get_data(DataType::SettingsName).as_str() else { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -1047,7 +1148,7 @@ pub fn handle_settings_load( || settings_name.contains("..") { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -1063,12 +1164,12 @@ pub fn handle_settings_load( ); } - let Ok(settings_value) = settings::load(my_id as i64, session_id as i64, settings_name) else { + let Ok(settings_value) = settings::load(my_id_i64, session_id_i64, settings_name) else { return error_response(cv, CommunicationType::ErrorInvalidData); }; let Some(settings_value_str) = settings_value else { return CommunicationValue::new(CommunicationType::ErrorNotFound) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::SettingsName, @@ -1081,7 +1182,7 @@ pub fn handle_settings_load( }; CommunicationValue::new(CommunicationType::SettingsLoad) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default(DataType::Payload, DataValue::Str(settings_value_str)) .add_typed_default( @@ -1098,10 +1199,16 @@ pub fn handle_settings_list( cv: &CommunicationValue, _expected_session_id: i128, ) -> CommunicationValue { - let my_id = cv.get_sender(); + let my_id = match cv.require_sender() { + Ok(my_id) => my_id, + Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData), + }; + let Ok(my_id_i64) = i64::try_from(my_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; let Some(session_id) = cv.get_data(DataType::SessionId).as_number() else { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -1110,7 +1217,7 @@ pub fn handle_settings_list( }; if session_id == 0 || session_id > 1_000_000 { return CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default( DataType::Message, @@ -1121,13 +1228,16 @@ pub fn handle_settings_list( DataValue::SignedNumber(session_id as i128), ); } + let Ok(session_id_i64) = i64::try_from(session_id) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; - let Ok(settings) = settings::list(my_id as i64, session_id as i64) else { + let Ok(settings) = settings::list(my_id_i64, session_id_i64) else { return error_response(cv, CommunicationType::ErrorInvalidData); }; let settings_json = settings.into_iter().map(DataValue::Str).collect(); CommunicationValue::new(CommunicationType::SettingsList) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(my_id) .add_typed_default(DataType::Settings, DataValue::Array(settings_json)) .add_typed_default( diff --git a/iota-connection/src/relay.rs b/iota-connection/src/relay.rs index a75c5a5..f9ef1fe 100644 --- a/iota-connection/src/relay.rs +++ b/iota-connection/src/relay.rs @@ -4,6 +4,8 @@ use mtp::codec::{ VerifiedRelayContent, VerifiedRelayMetadata, forward_relay_frame, open_relay_content_with_keyrings, open_relay_metadata_with_without_replay, relay_metadata_claimed_signer_id, + open_relay_content_with_limits_without_replay, open_relay_metadata_with_without_replay, + relay_metadata_claimed_signer_id_with_options, }; use mtp::crypto::{Keyring, PublicKeyBundle}; use std::fmt; @@ -147,7 +149,13 @@ where return Err(RelayValidationError::OuterSenderNotAllowed); } - let claimed_signer = relay_metadata_claimed_signer_id(frame, &[keyring])?; + let open_options = RelayOpenOptions::new(RELAY_PROTECTION_POLICY); + let claimed_signer = relay_metadata_claimed_signer_id_with_options( + frame, + &[keyring], + open_options.decode_limits, + open_options.protected_limits, + )?; let signing_keys = resolve_signing_keys(claimed_signer).await?; if signing_keys.is_empty() { return Err(RelayValidationError::MissingSigningKeys(claimed_signer)); @@ -186,12 +194,17 @@ pub fn open_verified_relay_content( keyrings: &[&Keyring], expected_recipient_id: u64, ) -> Result { - Ok(open_relay_content_with_keyrings( + Ok(open_relay_content_with_limits_without_replay( &relay.metadata, keyrings, &relay.signing_keys, Some(expected_recipient_id), - RELAY_PROTECTION_POLICY, + RelayOpenOptions { + policy: RELAY_PROTECTION_POLICY, + decode_limits: relay.metadata.decode_limits(), + encode_limits: relay.metadata.encode_limits(), + protected_limits: relay.metadata.protected_limits(), + }, )?) } diff --git a/iota-daemon-lib/Cargo.toml b/iota-daemon-lib/Cargo.toml index 1aec94c..c9dc6eb 100644 --- a/iota-daemon-lib/Cargo.toml +++ b/iota-daemon-lib/Cargo.toml @@ -16,6 +16,7 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git" } libc = "0.2" sysinfo = "0.38.0" serde_yaml = "0.9" +serde_json = "1" tokio = { version = "1.50.0", features = ["full"] } tokio-util = { version = "0.7", features = ["rt"] } uuid = { version = "*", features = ["v4"] } diff --git a/iota-daemon-lib/src/command_router.rs b/iota-daemon-lib/src/command_router.rs index 7f9f1c7..a6a2718 100644 --- a/iota-daemon-lib/src/command_router.rs +++ b/iota-daemon-lib/src/command_router.rs @@ -1,10 +1,10 @@ use crate::log_buffer::LogBuffer; use crate::{DaemonRuntime, DaemonServices}; use iota_ipc::{ - CommunitySummary, ComponentStatusResponse, ConfigResponse, ExitIntent, IpcErrorCode, - LocalRequest, LogEntriesResponse, OmikronStatusResponse, ResponseEnvelope, ResponsePayload, - ResponseResult, StatusResponse, TaskSummary, UpdateStatusResponse, UserDetailResponse, - UserSummary, + CommunitySummary, ComponentStatusResponse, ConfigResponse, DaemonMessage, ExitIntent, + IpcErrorCode, LocalRequest, LogEntriesResponse, LogEntry, MAX_MESSAGE_SIZE, + OmikronStatusResponse, ResponseEnvelope, ResponsePayload, ResponseResult, StatusResponse, + TaskSummary, UpdateStatusResponse, UserDetailResponse, UserSummary, }; use iota_logger::{log, log_command}; use iota_storage::users::user_manager; @@ -15,6 +15,37 @@ use std::time::Duration; use crate::daemon_state::{ShutdownReason, StartupPhase}; +pub use iota_ipc::IpcRole; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct PeerContext { + pub pid: i32, + pub uid: u32, + pub role: IpcRole, +} + +const MAX_LOG_ENTRIES_PER_RESPONSE: usize = 512; + +fn bounded_log_entries(mut entries: Vec) -> Vec { + entries.truncate(MAX_LOG_ENTRIES_PER_RESPONSE); + while !entries.is_empty() { + let response = DaemonMessage::Response(ResponseEnvelope { + request_id: u64::MAX, + result: ResponseResult::Ok(ResponsePayload::LogEntries(LogEntriesResponse { + entries: entries.clone(), + })), + }); + let fits = serde_json::to_vec(&response) + .map(|encoded| encoded.len() <= MAX_MESSAGE_SIZE) + .unwrap_or(false); + if fits { + return entries; + } + entries.remove(0); + } + entries +} + #[derive(Clone)] pub struct CommandRouter { runtime: Arc, @@ -35,8 +66,33 @@ impl CommandRouter { } } - pub async fn route(&self, request_id: u64, request: LocalRequest) -> ResponseEnvelope { - log_command!("{:?}", request); + pub async fn route( + &self, + peer: &PeerContext, + request_id: u64, + request: LocalRequest, + ) -> ResponseEnvelope { + if !peer.role.allows(request.required_role()) { + log!( + "IPC authorization denied: pid={}, uid={}, role={:?}, request={:?}", + peer.pid, + peer.uid, + peer.role, + request + ); + return ResponseEnvelope { + request_id, + result: ResponseResult::Error(IpcErrorCode::Unauthorized), + }; + } + + log_command!( + "pid={} uid={} role={:?} request={:?}", + peer.pid, + peer.uid, + peer.role, + request + ); let result = self.execute(request).await; ResponseEnvelope { request_id, result } } @@ -284,7 +340,7 @@ impl CommandRouter { { return ResponseResult::Error(IpcErrorCode::Conflict); } - self.runtime.shutdown(match intent { + self.runtime.request_shutdown(match intent { ExitIntent::Stop => ShutdownReason::Stop, ExitIntent::Restart => ShutdownReason::Restart, }); @@ -298,13 +354,13 @@ impl CommandRouter { }, )), LocalRequest::RestartDaemon => { - self.runtime.shutdown(ShutdownReason::Restart); + self.runtime.request_shutdown(ShutdownReason::Restart); ResponseResult::Ok(ResponsePayload::Acknowledged { message: "Daemon restart requested".into(), }) } LocalRequest::StopDaemon => { - self.runtime.shutdown(ShutdownReason::Stop); + self.runtime.request_shutdown(ShutdownReason::Stop); ResponseResult::Ok(ResponsePayload::Acknowledged { message: "Daemon shutdown requested".into(), }) @@ -378,7 +434,7 @@ impl CommandRouter { LocalRequest::ImportUser { .. } => ResponseResult::Error(IpcErrorCode::InvalidRequest), LocalRequest::GetLogs { limit } => { let entries = if let Ok(buf) = self.log_buffer.lock() { - buf.recent(limit) + bounded_log_entries(buf.recent(limit.min(MAX_LOG_ENTRIES_PER_RESPONSE))) } else { Vec::new() }; @@ -393,11 +449,10 @@ impl CommandRouter { Err(_e) => ResponseResult::Error(IpcErrorCode::InternalFailure), }, LocalRequest::ListCommunities => { - let iota_id = config_util::CONFIG - .load() - .iota_id - .map(|id| id as i64) - .unwrap_or(0); + let iota_id = config_util::CONFIG.load().iota_id; + let Ok(iota_id) = iota_id.map(i64::try_from).unwrap_or(Ok(0)) else { + return ResponseResult::Ok(ResponsePayload::Communities(Vec::new())); + }; let stored = iota_storage::util::communities_util::CommunitiesUtil::get_communities(iota_id); let summaries: Vec = stored @@ -412,3 +467,77 @@ impl CommandRouter { } } } + +#[cfg(test)] +mod tests { + use super::{IpcRole, LocalRequest, bounded_log_entries}; + use iota_ipc::{ExitIntent, LogEntry, SecretString}; + + #[test] + fn every_request_has_an_explicit_role_policy() { + let requests = [ + LocalRequest::GetStatus, + LocalRequest::ListTasks, + LocalRequest::ListUsers, + LocalRequest::CreateUser { + username: "alice".into(), + }, + LocalRequest::AttachUserFromTu { + credential: SecretString("credential".into()), + }, + LocalRequest::PurgeUserData { user_id: 1 }, + LocalRequest::ReleaseUser { user_id: 1 }, + LocalRequest::CompleteDeleteUser { + user_id: 1, + credential: None, + }, + LocalRequest::RemoveUser { user_id: 1 }, + LocalRequest::ReconnectOmikron, + LocalRequest::RotateIotaIdentity, + LocalRequest::RequestProcessExit { + intent: ExitIntent::Stop, + }, + LocalRequest::GetDaemonStatus, + LocalRequest::RestartDaemon, + LocalRequest::StopDaemon, + LocalRequest::GetConfig, + LocalRequest::SetConfig { + key: "port".into(), + value: "1984".into(), + }, + LocalRequest::ReloadConfig, + LocalRequest::GetOmikronStatus, + LocalRequest::ListComponents, + LocalRequest::GetUser { user_id: 1 }, + LocalRequest::ImportUser { + username: "alice".into(), + }, + LocalRequest::GetLogs { limit: 10 }, + LocalRequest::CheckUpdate, + LocalRequest::ListCommunities, + ]; + + assert_eq!(requests.len(), 25); + for request in requests { + let required = request.required_role(); + assert!(IpcRole::Admin.allows(required)); + assert_eq!( + IpcRole::Operate.allows(required), + required != IpcRole::Admin + ); + assert_eq!(IpcRole::Read.allows(required), required == IpcRole::Read); + } + } + + #[test] + fn log_responses_drop_entries_that_cannot_fit_one_ipc_frame() { + let entries = vec![LogEntry { + timestamp_ms: 0, + sender: "test".into(), + message: "x".repeat(2 * 1024 * 1024), + is_error: false, + }]; + + assert!(bounded_log_entries(entries).is_empty()); + } +} diff --git a/iota-daemon-lib/src/daemon_state.rs b/iota-daemon-lib/src/daemon_state.rs index 4c7347c..634cbe2 100644 --- a/iota-daemon-lib/src/daemon_state.rs +++ b/iota-daemon-lib/src/daemon_state.rs @@ -55,7 +55,7 @@ impl From for iota_ipc::StartupPhase { /* This wrapper exposes daemon state as IPC-safe snapshots while preserving a * single owned state instance for all daemon subsystems. The cancellation token - * is the single lifecycle signal — all subsystems check it instead of a + * is the single lifecycle signal, and all subsystems check it instead of a * separate boolean. */ pub struct DaemonRuntime { pub state: Arc, @@ -131,12 +131,20 @@ impl DaemonRuntime { } pub fn shutdown(&self, reason: ShutdownReason) { + self.request_shutdown(reason); + self.begin_shutdown(); + } + + pub fn request_shutdown(&self, reason: ShutdownReason) { if self.shutdown_tx.borrow().is_none() { let _ = self.shutdown_tx.send(Some(reason)); - self.cancellation.cancel(); } } + pub fn begin_shutdown(&self) { + self.cancellation.cancel(); + } + pub fn shutdown_reason(&self) -> Option { self.shutdown_tx.borrow().clone() } diff --git a/iota-daemon-lib/src/ipc_server.rs b/iota-daemon-lib/src/ipc_server.rs index 97b5dd6..a472348 100644 --- a/iota-daemon-lib/src/ipc_server.rs +++ b/iota-daemon-lib/src/ipc_server.rs @@ -1,23 +1,26 @@ use crate::deployment::from_environment; use crate::log_buffer::LogBuffer; -use crate::{CommandRouter, DaemonRuntime, DaemonServices}; +use crate::{CommandRouter, DaemonRuntime, DaemonServices, IpcRole, PeerContext}; use iota_ipc::{ ClientMessage, DaemonMessage, HelloAck, MIN_PROTOCOL_VERSION, PROTOCOL_VERSION, read_msg, write_msg, }; use iota_logger::log; +use iota_storage::util::config_util; use std::io::Result; -use std::os::unix::fs::{FileTypeExt, MetadataExt, OpenOptionsExt}; +use std::os::unix::fs::{FileTypeExt, MetadataExt, OpenOptionsExt, PermissionsExt}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::{env, fs::File, os::fd::FromRawFd, os::unix::net::UnixListener as StdUnixListener}; +use tokio::io::AsyncWriteExt; use tokio::net::{UnixListener, UnixStream}; -use tokio::sync::{broadcast, mpsc, watch}; +use tokio::sync::{Semaphore, broadcast, mpsc, watch}; use tokio::time::timeout; use uuid::Uuid; /// Per-client outbound queue capacity. const CLIENT_CHANNEL_SIZE: usize = 256; +const MAX_CONFIGURED_IPC_CLIENTS: usize = 4096; /// Maximum handshake retries before giving up. const MAX_HANDSHAKE_RETRIES: u32 = 1; @@ -36,6 +39,20 @@ struct ClientSubscription { metric_interval_ms: u64, } +enum WriterCommand { + Message(DaemonMessage), + Flush { + complete: tokio::sync::oneshot::Sender<()>, + }, +} + +fn configured_client_limit() -> usize { + config_util::CONFIG + .load() + .max_ipc_clients + .clamp(1, MAX_CONFIGURED_IPC_CLIENTS) +} + pub struct IpcServer { listener: UnixListener, runtime: Arc, @@ -45,6 +62,7 @@ pub struct IpcServer { state_rx: watch::Receiver, instance_id: String, _instance_lock: File, + client_limit: Arc, } impl IpcServer { @@ -96,11 +114,18 @@ impl IpcServer { } remove_stale_socket(&path).await?; let listener = UnixListener::bind(&path)?; - let _ = tokio::fs::set_permissions( - &path, - std::os::unix::fs::PermissionsExt::from_mode(0o600), - ) - .await; + if let Err(error) = + tokio::fs::set_permissions(&path, PermissionsExt::from_mode(0o600)).await + { + drop(listener); + let _ = tokio::fs::remove_file(&path).await; + return Err(error); + } + if let Err(error) = validate_manual_socket(&path).await { + drop(listener); + let _ = tokio::fs::remove_file(&path).await; + return Err(error); + } return Ok(Self { listener, runtime, @@ -110,6 +135,7 @@ impl IpcServer { state_rx, instance_id: Uuid::new_v4().to_string(), _instance_lock: lock, + client_limit: Arc::new(Semaphore::new(configured_client_limit())), }); } }; @@ -122,12 +148,21 @@ impl IpcServer { state_rx, instance_id: Uuid::new_v4().to_string(), _instance_lock: File::options().read(true).open("/dev/null")?, + client_limit: Arc::new(Semaphore::new(configured_client_limit())), }) } pub async fn serve(self) -> Result<()> { loop { let (stream, _addr) = self.listener.accept().await?; + let permit = match self.client_limit.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + eprintln!("IPC connection rejected: active client limit reached"); + drop(stream); + continue; + } + }; eprintln!("IPC client accepted"); let runtime = self.runtime.clone(); let services = self.services.clone(); @@ -136,6 +171,7 @@ impl IpcServer { let state_rx = self.state_rx.clone(); let instance_id = self.instance_id.clone(); tokio::spawn(async move { + let _permit = permit; if let Err(error) = handle_client( stream, runtime, @@ -232,6 +268,33 @@ async fn remove_stale_socket(path: &Path) -> Result<()> { } } +async fn validate_manual_socket(path: &Path) -> Result<()> { + let metadata = tokio::fs::symlink_metadata(path).await?; + if metadata.file_type().is_symlink() || !metadata.file_type().is_socket() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "bound IPC path is no longer a Unix socket", + )); + } + + let mode = metadata.permissions().mode() & 0o777; + if mode != 0o600 { + return Err(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + format!("IPC socket has unexpected mode {mode:o}"), + )); + } + + let expected_uid = unsafe { libc::geteuid() } as u32; + if metadata.uid() != expected_uid { + return Err(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + "IPC socket ownership changed after bind", + )); + } + Ok(()) +} + #[derive(Clone, Debug)] struct PeerIdentity { pid: i32, @@ -274,6 +337,14 @@ fn peer_credentials(stream: &UnixStream) -> Result { } } +fn role_for_peer(_peer: &PeerIdentity) -> IpcRole { + // This deployment has one IPC listener. Its Unix socket permissions are + // the admission boundary: systemd grants access to root, the daemon, and + // members of iota-operators. Once a peer has passed that boundary, it is + // an administrator for the operator console protocol. + IpcRole::Admin +} + async fn handle_client( stream: UnixStream, runtime: Arc, @@ -283,16 +354,17 @@ async fn handle_client( mut state_rx: watch::Receiver, instance_id: String, ) -> Result<()> { - let peer = peer_credentials(&stream)?; - // Access control belongs to the Unix socket. The systemd socket grants - // iota-operators group access (0660); rejecting every UID other than the - // service account here would make that authorization ineffective. Manual - // sockets remain owner-only (0600) at bind time. + let peer_identity = peer_credentials(&stream)?; + let peer = PeerContext { + pid: peer_identity.pid, + uid: peer_identity.uid, + role: role_for_peer(&peer_identity), + }; let (mut reader, mut writer) = stream.into_split(); // A failed writer must stop the reader and any subsequent command work // for this client; otherwise the reader can remain parked forever. let session_cancellation = runtime.cancellation.child_token(); - let (directed_tx, directed_rx) = mpsc::channel::(CLIENT_CHANNEL_SIZE); + let (directed_tx, directed_rx) = mpsc::channel::(CLIENT_CHANNEL_SIZE); eprintln!("IPC handshake started (pid={}, uid={})", peer.pid, peer.uid); // --- Handshake --- @@ -343,7 +415,7 @@ async fn handle_client( break; } Ok(_) => { - // Unexpected first message — send error and close. + // Unexpected first message, send an error and close. return Err(std::io::Error::new( std::io::ErrorKind::InvalidData, "Expected Hello as first message", @@ -353,7 +425,7 @@ async fn handle_client( }, } } - let _version = negotiated_version.ok_or_else(|| { + let negotiated_version = negotiated_version.ok_or_else(|| { std::io::Error::new(std::io::ErrorKind::Other, "Handshake failed after retries") })?; @@ -361,7 +433,7 @@ async fn handle_client( // --- Send initial state snapshot --- let initial = DaemonMessage::StateUpdate(runtime.snapshot()); - let _ = directed_tx.send(initial).await; + let _ = directed_tx.send(WriterCommand::Message(initial)).await; // --- Writer task: merge directed responses + shared log events --- let mut log_rx = log_tx.subscribe(); @@ -375,19 +447,29 @@ async fn handle_client( tokio::spawn(async move { let mut directed_rx = directed_rx; let mut last_metric_sent = tokio::time::Instant::now(); + let mut state_updates_open = true; loop { let metric_interval = sub_rx.borrow().metric_interval_ms; tokio::select! { + _ = session_cancellation.cancelled() => break, // Directed messages (responses to this client's requests) - msg = directed_rx.recv() => { - match msg { - Some(message) => { + command = directed_rx.recv() => { + match command { + Some(WriterCommand::Message(message)) => { if let Err(error) = write_client_message(&mut writer, &message).await { eprintln!("IPC client writer stopped while sending directed message: {error}"); session_cancellation.cancel(); break; } } + Some(WriterCommand::Flush { complete }) => { + if let Err(error) = writer.flush().await { + eprintln!("IPC client writer stopped while flushing: {error}"); + session_cancellation.cancel(); + break; + } + let _ = complete.send(()); + } None => break, } } @@ -435,12 +517,16 @@ async fn handle_client( break; } } - Err(broadcast::error::RecvError::Closed) => break, + Err(broadcast::error::RecvError::Closed) => { + session_cancellation.cancel(); + break; + } } } - changed = state_rx.changed() => { + changed = state_rx.changed(), if state_updates_open => { if changed.is_err() { - break; + state_updates_open = false; + continue; } let snapshot = state_rx.borrow().clone(); if let Err(error) = write_client_message(&mut writer, &DaemonMessage::StateUpdate(snapshot)).await { @@ -475,9 +561,14 @@ async fn handle_client( iota_ipc::LocalRequest::StopDaemon => Some("shutdown requested"), _ => None, }; - let response = if envelope.protocol_version < MIN_PROTOCOL_VERSION - || envelope.protocol_version > PROTOCOL_VERSION - { + let response = if envelope.protocol_version != negotiated_version { + log!( + "IPC protocol mismatch: pid={}, uid={}, negotiated={}, request={}", + peer.pid, + peer.uid, + negotiated_version, + envelope.protocol_version + ); iota_ipc::ResponseEnvelope { request_id: envelope.request_id, result: iota_ipc::ResponseResult::Error( @@ -485,21 +576,42 @@ async fn handle_client( ), } } else { - router.route(envelope.request_id, envelope.request).await + router + .route(&peer, envelope.request_id, envelope.request) + .await }; - let _ = directed_tx.send(DaemonMessage::Response(response)).await; - if let Some(reason) = shutdown_reason { + let should_shutdown = shutdown_reason.is_some() + && matches!(&response.result, iota_ipc::ResponseResult::Ok(_)); + let _ = directed_tx + .send(WriterCommand::Message(DaemonMessage::Response(response))) + .await; + if let Some(reason) = shutdown_reason.filter(|_| should_shutdown) { let _ = directed_tx - .send(DaemonMessage::LifecycleEvent( + .send(WriterCommand::Message(DaemonMessage::LifecycleEvent( iota_ipc::LifecycleEvent::Shutdown { reason: reason.into(), }, - )) + ))) .await; - // The request itself initiates daemon cancellation. Give - // the dedicated writer a chance to flush the response - // and lifecycle event before this session is torn down. - tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let (flush_tx, flush_rx) = tokio::sync::oneshot::channel(); + let _ = directed_tx + .send(WriterCommand::Flush { complete: flush_tx }) + .await; + timeout(CLIENT_IO_TIMEOUT, flush_rx) + .await + .map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::TimedOut, + "IPC shutdown response flush timed out", + ) + })? + .map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "IPC writer stopped before shutdown flush", + ) + })?; + runtime.begin_shutdown(); break; } } @@ -515,26 +627,51 @@ async fn handle_client( metric_interval_ms: interval, }); let snapshot = DaemonMessage::StateUpdate(runtime.snapshot()); - let _ = directed_tx.send(snapshot).await; - let _ = directed_tx.send(DaemonMessage::Subscribed).await; + let _ = directed_tx.send(WriterCommand::Message(snapshot)).await; + let _ = directed_tx + .send(WriterCommand::Message(DaemonMessage::Subscribed)) + .await; } Ok(ClientMessage::Ping { seq }) => { - let _ = directed_tx.send(DaemonMessage::Pong { seq }).await; + let _ = directed_tx + .send(WriterCommand::Message(DaemonMessage::Pong { seq })) + .await; } Ok(ClientMessage::Hello { .. }) => { // Re-handshake on existing connection: treat as resubscribe let snapshot = DaemonMessage::StateUpdate(runtime.snapshot()); - let _ = directed_tx.send(snapshot).await; + let _ = directed_tx.send(WriterCommand::Message(snapshot)).await; } Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => break, Err(error) => { - writer_task.abort(); + session_cancellation.cancel(); + drop(directed_tx); + let mut writer_task = writer_task; + match timeout(CLIENT_IO_TIMEOUT, &mut writer_task).await { + Ok(_) => {} + Err(_) => { + writer_task.abort(); + let _ = writer_task.await; + } + } return Err(error); } } } + drop(directed_tx); session_cancellation.cancel(); - writer_task.abort(); + let mut writer_task = writer_task; + match timeout(CLIENT_IO_TIMEOUT, &mut writer_task).await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + eprintln!("IPC client writer task failed: {error}"); + } + Err(_) => { + eprintln!("IPC client writer did not stop before timeout"); + writer_task.abort(); + let _ = writer_task.await; + } + } log!( "IPC client disconnected (pid={}, uid={})", peer.pid, @@ -563,4 +700,46 @@ mod tests { .is_err() ); } + + #[tokio::test] + async fn manual_socket_validation_requires_owner_only_mode() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("ipc.sock"); + let listener = StdUnixListener::bind(&path).expect("test socket binds"); + tokio::fs::set_permissions(&path, PermissionsExt::from_mode(0o600)) + .await + .expect("test socket permissions apply"); + + validate_manual_socket(&path) + .await + .expect("manual socket validation succeeds"); + drop(listener); + } + + #[tokio::test] + async fn manual_socket_validation_rejects_unexpected_mode() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("ipc.sock"); + let listener = StdUnixListener::bind(&path).expect("test socket binds"); + tokio::fs::set_permissions(&path, PermissionsExt::from_mode(0o660)) + .await + .expect("test socket permissions apply"); + + let error = validate_manual_socket(&path) + .await + .expect_err("group-accessible manual socket must be rejected"); + assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied); + drop(listener); + } + + #[test] + fn an_admitted_operator_peer_receives_administrator_role() { + let peer = PeerIdentity { + pid: 123, + uid: 1000, + _gid: 1000, + }; + + assert_eq!(role_for_peer(&peer), IpcRole::Admin); + } } diff --git a/iota-daemon-lib/src/lib.rs b/iota-daemon-lib/src/lib.rs index 6ccd977..3afeff8 100644 --- a/iota-daemon-lib/src/lib.rs +++ b/iota-daemon-lib/src/lib.rs @@ -7,7 +7,7 @@ pub mod log_buffer; pub mod services; pub mod task_registry; -pub use command_router::CommandRouter; +pub use command_router::{CommandRouter, IpcRole, PeerContext}; pub use daemon_state::{DaemonRuntime, ShutdownReason, StartupPhase}; pub use ipc_server::IpcServer; pub use services::DaemonServices; diff --git a/iota-daemon-lib/tests/command_router.rs b/iota-daemon-lib/tests/command_router.rs index 2bf8bd7..99a2aaa 100644 --- a/iota-daemon-lib/tests/command_router.rs +++ b/iota-daemon-lib/tests/command_router.rs @@ -1,7 +1,7 @@ use async_trait::async_trait; use iota_daemon_lib::log_buffer::LogBuffer; -use iota_daemon_lib::{CommandRouter, DaemonRuntime, DaemonServices}; -use iota_ipc::{LocalRequest, ResponseResult}; +use iota_daemon_lib::{CommandRouter, DaemonRuntime, DaemonServices, IpcRole, PeerContext}; +use iota_ipc::{IpcErrorCode, LocalRequest, ResponseResult}; use mtp::codec::CommunicationValue; use omikron_connector::{OmikronClient, OmikronError}; use std::sync::{ @@ -13,6 +13,22 @@ use std::time::Duration; struct FakeOmikron { reconnects: AtomicUsize, } + +fn admin_peer() -> PeerContext { + PeerContext { + pid: 1, + uid: 0, + role: IpcRole::Admin, + } +} + +fn read_peer() -> PeerContext { + PeerContext { + pid: 2, + uid: 1000, + role: IpcRole::Read, + } +} #[async_trait] impl OmikronClient for FakeOmikron { async fn send_message(&self, _: &CommunicationValue) -> Result<(), OmikronError> { @@ -55,7 +71,10 @@ async fn reconnect_uses_the_injected_client() { Arc::new(Mutex::new(LogBuffer::new(100))), ); assert!(matches!( - router.route(1, LocalRequest::ReconnectOmikron).await.result, + router + .route(&admin_peer(), 1, LocalRequest::ReconnectOmikron) + .await + .result, ResponseResult::Ok(_) )); assert_eq!(fake.reconnects.load(Ordering::SeqCst), 1); @@ -79,10 +98,44 @@ async fn identity_rotation_is_available_while_omikron_is_offline() { ); assert!(matches!( router - .route(1, LocalRequest::RotateIotaIdentity) + .route(&admin_peer(), 1, LocalRequest::RotateIotaIdentity) .await .result, ResponseResult::Ok(_) )); assert_eq!(fake.reconnects.load(Ordering::SeqCst), 1); } + +#[tokio::test] +async fn read_role_cannot_execute_an_administrative_request() { + let fake = Arc::new(FakeOmikron { + reconnects: AtomicUsize::new(0), + }); + let services = Arc::new(DaemonServices { + omikron: fake.clone(), + users: Default::default(), + config: Default::default(), + active: true, + }); + let router = CommandRouter::new( + Arc::new(DaemonRuntime::new()), + services, + Arc::new(Mutex::new(LogBuffer::new(100))), + ); + + assert!(matches!( + router + .route( + &read_peer(), + 9, + LocalRequest::SetConfig { + key: "port".into(), + value: "1984".into(), + }, + ) + .await + .result, + ResponseResult::Error(IpcErrorCode::Unauthorized) + )); + assert_eq!(fake.reconnects.load(Ordering::SeqCst), 0); +} diff --git a/iota-daemon-lib/tests/ipc_server.rs b/iota-daemon-lib/tests/ipc_server.rs new file mode 100644 index 0000000..7540ea4 --- /dev/null +++ b/iota-daemon-lib/tests/ipc_server.rs @@ -0,0 +1,252 @@ +use async_trait::async_trait; +use iota_daemon_lib::{DaemonRuntime, DaemonServices, IpcServer}; +use iota_ipc::{ + ClientMessage, DaemonMessage, ExitIntent, IpcErrorCode, LocalRequest, PROTOCOL_VERSION, + RequestEnvelope, ResponseResult, read_msg, write_msg, +}; +use iota_storage::util::config_util::{self, IotaConfig}; +use mtp::codec::CommunicationValue; +use omikron_connector::{OmikronClient, OmikronError}; +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; +use tokio::net::UnixStream; +use tokio::sync::{broadcast, watch}; + +struct ConfigRestore(Arc); + +impl Drop for ConfigRestore { + fn drop(&mut self) { + config_util::CONFIG.store(self.0.clone()); + } +} + +fn set_client_limit(limit: usize) -> ConfigRestore { + let previous = config_util::CONFIG.load_full(); + let mut config = (*previous).clone(); + config.max_ipc_clients = limit; + config_util::CONFIG.store(Arc::new(config)); + ConfigRestore(previous) +} + +async fn start_server( + path: &Path, + services: Arc, +) -> (Arc, tokio::task::JoinHandle<()>) { + let runtime = Arc::new(DaemonRuntime::new()); + let (log_tx, _) = broadcast::channel(32); + let log_buffer = Arc::new(std::sync::Mutex::new( + iota_daemon_lib::log_buffer::LogBuffer::new(32), + )); + let (_, state_rx) = watch::channel(runtime.snapshot()); + let server = IpcServer::bind( + path.to_owned(), + runtime.clone(), + services, + log_tx, + log_buffer, + state_rx, + ) + .await + .expect("IPC server binds"); + let task = tokio::spawn(async move { + let _ = server.serve().await; + }); + (runtime, task) +} + +async fn try_connect_and_await_hello(path: &Path) -> std::io::Result { + let mut stream = UnixStream::connect(path).await?; + write_msg( + &mut stream, + &ClientMessage::Hello { + supported_versions: vec![PROTOCOL_VERSION], + }, + ) + .await?; + let message: DaemonMessage = read_msg(&mut stream).await?; + if !matches!(message, DaemonMessage::HelloAck(_)) { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "expected HelloAck", + )); + } + Ok(stream) +} + +async fn connect_and_await_hello(path: &Path) -> UnixStream { + try_connect_and_await_hello(path) + .await + .expect("IPC connection completes the Hello exchange") +} + +#[tokio::test] +async fn active_client_limit_rejects_excess_clients_and_releases_permits() { + let _config = set_client_limit(1); + let directory = tempfile::tempdir().unwrap(); + let socket = directory.path().join("ipc.sock"); + let (_runtime, server_task) = start_server(&socket, DaemonServices::inactive()).await; + + let first = connect_and_await_hello(&socket).await; + let mut rejected = UnixStream::connect(&socket) + .await + .expect("second connection reaches the Unix listener"); + let rejected_result = tokio::time::timeout( + Duration::from_secs(2), + read_msg::<_, DaemonMessage>(&mut rejected), + ) + .await + .expect("rejected client is closed promptly"); + assert!(rejected_result.is_err()); + + drop(first); + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let _released = loop { + match try_connect_and_await_hello(&socket).await { + Ok(stream) => break stream, + Err(_error) if tokio::time::Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(10)).await; + } + Err(error) => panic!("client permit was not released: {error}"), + } + }; + server_task.abort(); + let _ = server_task.await; +} + +#[tokio::test] +async fn request_with_version_different_from_hello_is_rejected() { + let directory = tempfile::tempdir().unwrap(); + let socket = directory.path().join("ipc.sock"); + let (_runtime, server_task) = start_server(&socket, DaemonServices::inactive()).await; + let mut stream = connect_and_await_hello(&socket).await; + + write_msg( + &mut stream, + &ClientMessage::Request(RequestEnvelope { + request_id: 7, + protocol_version: PROTOCOL_VERSION + 1, + request: LocalRequest::GetStatus, + }), + ) + .await + .expect("request sends"); + + let response = loop { + match read_msg::<_, DaemonMessage>(&mut stream) + .await + .expect("daemon response arrives") + { + DaemonMessage::Response(response) => break response, + _ => continue, + } + }; + assert_eq!(response.request_id, 7); + assert!(matches!( + response.result, + ResponseResult::Error(IpcErrorCode::UnsupportedVersion) + )); + + drop(stream); + server_task.abort(); + let _ = server_task.await; +} + +struct TestOmikron; + +#[async_trait] +impl OmikronClient for TestOmikron { + async fn send_message(&self, _: &CommunicationValue) -> Result<(), OmikronError> { + Ok(()) + } + + async fn await_response( + &self, + _: &CommunicationValue, + _: Duration, + ) -> Result { + Err(OmikronError::Disconnected("test client".into())) + } + + async fn reconnect(&self) -> Result<(), OmikronError> { + Ok(()) + } + + async fn rotate_identity(&self) -> Result<(), OmikronError> { + Ok(()) + } + + async fn is_connected(&self) -> bool { + true + } +} + +fn active_services() -> Arc { + Arc::new(DaemonServices { + omikron: Arc::new(TestOmikron), + users: Default::default(), + config: Default::default(), + active: true, + }) +} + +#[tokio::test] +async fn shutdown_delivers_response_and_lifecycle_event_before_eof() { + let directory = tempfile::tempdir().unwrap(); + let socket = directory.path().join("ipc.sock"); + let (runtime, server_task) = start_server(&socket, active_services()).await; + let mut stream = connect_and_await_hello(&socket).await; + + write_msg( + &mut stream, + &ClientMessage::Request(RequestEnvelope { + request_id: 8, + protocol_version: PROTOCOL_VERSION, + request: LocalRequest::RequestProcessExit { + intent: ExitIntent::Stop, + }, + }), + ) + .await + .expect("shutdown request sends"); + + let mut response_seen = false; + let mut lifecycle_seen = false; + for _ in 0..4 { + match tokio::time::timeout( + Duration::from_secs(2), + read_msg::<_, DaemonMessage>(&mut stream), + ) + .await + .expect("shutdown message arrives") + .expect("shutdown stream remains readable") + { + DaemonMessage::Response(response) => { + assert_eq!(response.request_id, 8); + assert!(matches!(response.result, ResponseResult::Ok(_))); + response_seen = true; + } + DaemonMessage::LifecycleEvent(iota_ipc::LifecycleEvent::Shutdown { .. }) => { + lifecycle_seen = true; + } + _ => {} + } + if response_seen && lifecycle_seen { + break; + } + } + + assert!(response_seen); + assert!(lifecycle_seen); + let eof = tokio::time::timeout( + Duration::from_secs(2), + read_msg::<_, DaemonMessage>(&mut stream), + ) + .await + .expect("shutdown connection closes after flush"); + assert!(eof.is_err()); + assert!(runtime.is_shutting_down()); + + server_task.abort(); + let _ = server_task.await; +} diff --git a/iota-ipc/src/lib.rs b/iota-ipc/src/lib.rs index af695f2..acf72b1 100644 --- a/iota-ipc/src/lib.rs +++ b/iota-ipc/src/lib.rs @@ -5,13 +5,13 @@ pub mod transport; pub use protocol::{ ClientMessage, CommunitySummary, ComponentHealth, ComponentId, ComponentStatusResponse, ConfigResponse, ConnectionStatus, DaemonMessage, DaemonStatusResponse, DeploymentMode, - ExitIntent, HealthStatus, HelloAck, IpcErrorCode, LifecycleEvent, LifecyclePhase, LocalRequest, - LocalUserState, LogEntriesResponse, LogEntry, MetricSample, OmikronStatusResponse, - RequestEnvelope, ResponseEnvelope, ResponsePayload, ResponseResult, SecretString, StartupPhase, - StateSnapshot, StatusResponse, SupervisorKind, TaskSummary, UpdateStatusResponse, - UserDetailResponse, UserSummary, + ExitIntent, HealthStatus, HelloAck, IpcErrorCode, IpcRole, LifecycleEvent, LifecyclePhase, + LocalRequest, LocalUserState, LogEntriesResponse, LogEntry, MetricSample, + OmikronStatusResponse, RequestEnvelope, ResponseEnvelope, ResponsePayload, ResponseResult, + SecretString, StartupPhase, StateSnapshot, StatusResponse, SupervisorKind, TaskSummary, + UpdateStatusResponse, UserDetailResponse, UserSummary, }; -pub use transport::{read_msg, write_msg}; +pub use transport::{MAX_MESSAGE_SIZE, read_msg, write_msg}; /// Current IPC protocol version. pub const PROTOCOL_VERSION: u16 = 2; diff --git a/iota-ipc/src/protocol.rs b/iota-ipc/src/protocol.rs index 01cd14e..4006583 100644 --- a/iota-ipc/src/protocol.rs +++ b/iota-ipc/src/protocol.rs @@ -97,6 +97,59 @@ pub enum LocalRequest { ListCommunities, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum IpcRole { + Read, + Operate, + Admin, +} + +impl IpcRole { + pub fn allows(self, required: IpcRole) -> bool { + matches!( + (self, required), + (IpcRole::Admin, _) + | (IpcRole::Operate, IpcRole::Operate | IpcRole::Read) + | (IpcRole::Read, IpcRole::Read) + ) + } +} + +impl LocalRequest { + /// Return the minimum authenticated local role required to execute a + /// request. New request variants must be assigned explicitly here. + pub fn required_role(&self) -> IpcRole { + match self { + Self::GetStatus + | Self::ListTasks + | Self::ListUsers + | Self::GetDaemonStatus + | Self::GetOmikronStatus + | Self::ListComponents + | Self::GetUser { .. } + | Self::GetLogs { .. } + | Self::CheckUpdate + | Self::ListCommunities => IpcRole::Read, + + Self::ReconnectOmikron | Self::ReloadConfig => IpcRole::Operate, + + Self::CreateUser { .. } + | Self::AttachUserFromTu { .. } + | Self::PurgeUserData { .. } + | Self::ReleaseUser { .. } + | Self::CompleteDeleteUser { .. } + | Self::RemoveUser { .. } + | Self::RotateIotaIdentity + | Self::RequestProcessExit { .. } + | Self::RestartDaemon + | Self::StopDaemon + | Self::GetConfig + | Self::SetConfig { .. } + | Self::ImportUser { .. } => IpcRole::Admin, + } + } +} + #[derive(Clone, Copy, Debug, Deserialize, Serialize)] #[serde(rename_all = "snake_case")] pub enum ExitIntent { @@ -296,7 +349,9 @@ impl std::fmt::Display for IpcErrorCode { Self::Disconnected => "the daemon connection was lost", Self::Timeout => "the daemon did not respond in time", Self::Cancelled => "the daemon cancelled the request", - Self::Unauthorized => "the daemon denied this operation", + Self::Unauthorized => { + "the daemon denied this operation because the IPC account lacks the required role" + } Self::InternalFailure => "the daemon encountered an internal failure", }) } @@ -317,6 +372,11 @@ mod error_tests { .to_string() .contains("InternalFailure") ); + assert!( + IpcErrorCode::Unauthorized + .to_string() + .contains("required role") + ); } } diff --git a/iota-ipc/src/transport.rs b/iota-ipc/src/transport.rs index 2793451..6a4382f 100644 --- a/iota-ipc/src/transport.rs +++ b/iota-ipc/src/transport.rs @@ -3,7 +3,10 @@ use serde::de::DeserializeOwned; use std::io::{Error, ErrorKind, Result}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; -const MAX_MESSAGE_SIZE: usize = 1024 * 1024; +/// Maximum encoded payload size for a single IPC frame. +/// +/// This is a wire-level contract shared by both sides of the connection. +pub const MAX_MESSAGE_SIZE: usize = 1024 * 1024; /* Length-prefixing preserves message boundaries on a byte stream and bounds * allocations before JSON is deserialized. */ @@ -14,6 +17,12 @@ where { let payload = serde_json::to_vec(message).map_err(|error| Error::new(ErrorKind::InvalidData, error))?; + if payload.len() > MAX_MESSAGE_SIZE { + return Err(Error::new( + ErrorKind::InvalidData, + "IPC message exceeds limit", + )); + } let len = u32::try_from(payload.len()) .map_err(|_| Error::new(ErrorKind::InvalidData, "IPC message is too large"))?; writer.write_u32(len).await?; @@ -40,7 +49,7 @@ where #[cfg(test)] mod tests { - use super::{read_msg, write_msg}; + use super::{MAX_MESSAGE_SIZE, read_msg, write_msg}; use crate::protocol::{ClientMessage, LocalRequest, RequestEnvelope}; #[tokio::test] @@ -57,4 +66,31 @@ mod tests { let received: ClientMessage = read_msg(&mut reader).await.expect("read succeeds"); assert!(matches!(received, ClientMessage::Request(req) if req.request_id == 4)); } + + #[tokio::test] + async fn write_rejects_message_above_frame_limit() { + let (mut writer, _reader) = tokio::io::duplex(MAX_MESSAGE_SIZE + 16); + let message = "x".repeat(MAX_MESSAGE_SIZE + 1); + + let error = write_msg(&mut writer, &message) + .await + .expect_err("oversized payload must be rejected before framing"); + + assert_eq!(error.kind(), std::io::ErrorKind::InvalidData); + assert!(error.to_string().contains("exceeds limit")); + } + + #[tokio::test] + async fn read_rejects_frame_above_limit_before_allocating_payload() { + let (mut writer, mut reader) = tokio::io::duplex(16); + tokio::io::AsyncWriteExt::write_u32(&mut writer, (MAX_MESSAGE_SIZE + 1) as u32) + .await + .expect("length prefix write succeeds"); + + let error = read_msg::<_, ClientMessage>(&mut reader) + .await + .expect_err("oversized frame must be rejected"); + + assert_eq!(error.kind(), std::io::ErrorKind::InvalidData); + } } diff --git a/iota-process-manager/Cargo.toml b/iota-process-manager/Cargo.toml index c159ef8..6977698 100644 --- a/iota-process-manager/Cargo.toml +++ b/iota-process-manager/Cargo.toml @@ -6,3 +6,7 @@ edition = "2024" [dependencies] async-trait = "0.1" tokio = { version = "1.50", features = ["process", "time", "io-util", "macros", "rt"] } + +[dev-dependencies] +libc = "0.2" +tempfile = "3" diff --git a/iota-process-manager/src/lib.rs b/iota-process-manager/src/lib.rs index c0ebb0d..3512a90 100644 --- a/iota-process-manager/src/lib.rs +++ b/iota-process-manager/src/lib.rs @@ -167,26 +167,66 @@ pub async fn detect() -> Option> { mod systemd { use super::*; use std::{path::Path, process::Stdio}; - use tokio::{process::Command, time::timeout}; + use tokio::{io::AsyncRead, process::Command, time::timeout}; const SERVICE: &str = "iota-daemon.service"; const SOCKET: &str = "iota-daemon.socket"; const COMMON: [&str; 2] = ["--no-pager", "--no-ask-password"]; + const MAX_COMMAND_OUTPUT_BYTES: usize = 1024 * 1024; pub struct RealExecutor; - #[async_trait] - impl CommandExecutor for RealExecutor { - async fn output( + + async fn read_bounded(reader: R) -> std::io::Result> + where + R: AsyncRead + Unpin, + { + use tokio::io::AsyncReadExt; + + let mut output = Vec::new(); + reader + .take((MAX_COMMAND_OUTPUT_BYTES + 1) as u64) + .read_to_end(&mut output) + .await?; + if output.len() > MAX_COMMAND_OUTPUT_BYTES { + output.truncate(MAX_COMMAND_OUTPUT_BYTES); + } + Ok(output) + } + + async fn collect_output( + stdout: tokio::process::ChildStdout, + stderr: tokio::process::ChildStderr, + ) -> Result<(Vec, Vec), ProcessManagerError> { + let (stdout_result, stderr_result) = + tokio::join!(read_bounded(stdout), read_bounded(stderr)); + let stdout = stdout_result.map_err(|error| { + ProcessManagerError::new( + ProcessManagerErrorKind::CommandFailed, + format!("stdout read failed: {error}"), + ) + })?; + let stderr = stderr_result.map_err(|error| { + ProcessManagerError::new( + ProcessManagerErrorKind::CommandFailed, + format!("stderr read failed: {error}"), + ) + })?; + Ok((stdout, stderr)) + } + + impl RealExecutor { + async fn output_with_timeout( &self, program: &str, args: &[&str], + process_timeout: std::time::Duration, ) -> Result { - let child = Command::new(program) + let mut child = Command::new(program) .args(args) .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) - .kill_on_drop(false) + .kill_on_drop(true) .spawn() .map_err(|e| { ProcessManagerError::new( @@ -194,28 +234,79 @@ mod systemd { format!("Could not run {program}: {e}"), ) })?; - let output = timeout(PROCESS_MANAGER_TIMEOUT, child.wait_with_output()) - .await - .map_err(|_| { - ProcessManagerError::new( + + let stdout = child.stdout.take().ok_or_else(|| { + ProcessManagerError::new( + ProcessManagerErrorKind::CommandFailed, + "command stdout pipe was not created", + ) + })?; + let stderr = child.stderr.take().ok_or_else(|| { + ProcessManagerError::new( + ProcessManagerErrorKind::CommandFailed, + "command stderr pipe was not created", + ) + })?; + let output_task = tokio::spawn(collect_output(stdout, stderr)); + + let status = match timeout(process_timeout, child.wait()).await { + Ok(result) => result.map_err(|e| { + ProcessManagerError::new(ProcessManagerErrorKind::CommandFailed, e.to_string()) + })?, + Err(_) => { + // Keep the Child alive across the timeout. Explicitly + // terminate it and await wait() so the OS child is + // reaped before reporting the timeout. + let kill_error = child.start_kill().err(); + let wait_error = child.wait().await.err(); + output_task.abort(); + let _ = output_task.await; + + if let Some(error) = wait_error { + return Err(ProcessManagerError::new( + ProcessManagerErrorKind::CommandFailed, + format!("{program} timed out and could not be reaped: {error}"), + )); + } + let termination_detail = kill_error + .map(|error| format!("; termination request reported: {error}")) + .unwrap_or_default(); + return Err(ProcessManagerError::new( ProcessManagerErrorKind::TimedOut, format!( - "{program} timed out after {} seconds", - PROCESS_MANAGER_TIMEOUT.as_secs() + "{program} timed out after {} seconds{termination_detail}", + process_timeout.as_secs(), ), - ) - })? - .map_err(|e| { - ProcessManagerError::new(ProcessManagerErrorKind::CommandFailed, e.to_string()) - })?; + )); + } + }; + + let (stdout, stderr) = output_task.await.map_err(|error| { + ProcessManagerError::new( + ProcessManagerErrorKind::CommandFailed, + format!("command output task failed: {error}"), + ) + })??; Ok(CommandOutput { - success: output.status.success(), - stdout: String::from_utf8_lossy(&output.stdout).into_owned(), - stderr: String::from_utf8_lossy(&output.stderr).into_owned(), + success: status.success(), + stdout: String::from_utf8_lossy(&stdout).into_owned(), + stderr: String::from_utf8_lossy(&stderr).into_owned(), }) } } + #[async_trait] + impl CommandExecutor for RealExecutor { + async fn output( + &self, + program: &str, + args: &[&str], + ) -> Result { + self.output_with_timeout(program, args, PROCESS_MANAGER_TIMEOUT) + .await + } + } + pub struct SystemdManager { executor: Arc, service: &'static str, @@ -461,6 +552,51 @@ mod systemd { assert!(call.contains(&"--no-pager".into())); assert!(call.contains(&"--no-ask-password".into())); } + + #[tokio::test] + async fn timed_out_real_child_is_terminated_and_reaped() { + use std::fs; + use std::time::Duration; + + let directory = tempfile::tempdir().unwrap(); + let pid_file = directory.path().join("child.pid"); + let script = format!( + "printf '%s' \"$$\" > '{}'; exec sleep 60", + pid_file.display() + ); + let executor = RealExecutor; + let task = tokio::spawn(async move { + executor + .output_with_timeout("sh", &["-c", &script], Duration::from_millis(50)) + .await + }); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let pid = loop { + if let Ok(contents) = fs::read_to_string(&pid_file) { + if let Ok(pid) = contents.parse::() { + break pid; + } + } + assert!(tokio::time::Instant::now() < deadline); + tokio::task::yield_now().await; + }; + + let result = task.await.unwrap(); + assert_eq!( + result.unwrap_err().kind(), + ProcessManagerErrorKind::TimedOut + ); + assert!(!std::path::Path::new(&format!("/proc/{pid}")).exists()); + + let mut status = 0; + let wait_result = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) }; + assert_eq!(wait_result, -1); + assert_eq!( + std::io::Error::last_os_error().raw_os_error(), + Some(libc::ECHILD) + ); + } } } diff --git a/iota-state/Cargo.toml b/iota-state/Cargo.toml index 4db4406..e4acfc2 100644 --- a/iota-state/Cargo.toml +++ b/iota-state/Cargo.toml @@ -12,4 +12,8 @@ dashmap = "6.1.0" once_cell = "1.21.3" tokio = { version = "1.50.0", features = ["full"] } json = "*" +<<<<<<< HEAD sysinfo = "0.38.0" +======= +sysinfo = "0.39.0" +>>>>>>> refs/remotes/origin/main diff --git a/iota-storage/Cargo.toml b/iota-storage/Cargo.toml index a579f7f..ac8783e 100644 --- a/iota-storage/Cargo.toml +++ b/iota-storage/Cargo.toml @@ -7,7 +7,10 @@ edition = "2024" iota-logger = { path = "../iota-logger" } iota-util = { path = "../iota-util" } iota-paths = { path = "../iota-paths" } +<<<<<<< HEAD +======= +>>>>>>> refs/remotes/origin/main base64 = "0.22.1" json = "*" arc-swap = "1" diff --git a/iota-storage/src/util/config_util.rs b/iota-storage/src/util/config_util.rs index 62b65e9..b3dfe08 100644 --- a/iota-storage/src/util/config_util.rs +++ b/iota-storage/src/util/config_util.rs @@ -29,6 +29,8 @@ pub struct IotaConfig { pub private_key: Option, #[serde(default = "default_read_receipts_enabled")] pub read_receipts_enabled: bool, + #[serde(default = "default_max_ipc_clients")] + pub max_ipc_clients: usize, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -87,6 +89,10 @@ const fn default_read_receipts_enabled() -> bool { true } +const fn default_max_ipc_clients() -> usize { + 64 +} + impl Default for IotaConfig { fn default() -> Self { Self { @@ -99,6 +105,7 @@ impl Default for IotaConfig { public_key: None, private_key: None, read_receipts_enabled: default_read_receipts_enabled(), + max_ipc_clients: default_max_ipc_clients(), } } } @@ -198,6 +205,14 @@ pub fn modify_config_value(key: &str, value: &str) -> Result<(), &'static str> { modify_config(|cfg| cfg.read_receipts_enabled = parsed); Ok(()) } + "max_ipc_clients" => { + let parsed: usize = value.parse().map_err(|_| "invalid max_ipc_clients")?; + if parsed == 0 { + return Err("max_ipc_clients must be greater than zero"); + } + modify_config(|cfg| cfg.max_ipc_clients = parsed); + Ok(()) + } "web.mode" => { let mode = match value { "disabled" => WebMode::Disabled, diff --git a/iota-updater/Cargo.toml b/iota-updater/Cargo.toml index 8467afd..9e19437 100644 --- a/iota-updater/Cargo.toml +++ b/iota-updater/Cargo.toml @@ -5,7 +5,10 @@ edition = "2024" [dependencies] iota-paths = { path = "../iota-paths" } +<<<<<<< HEAD +======= +>>>>>>> refs/remotes/origin/main tokio = { version = "1.50.0", features = ["full"] } sha2 = "0.11.0" hex = "*" diff --git a/iota-util/src/crypto_helper.rs b/iota-util/src/crypto_helper.rs index 339dd1f..1931da7 100644 --- a/iota-util/src/crypto_helper.rs +++ b/iota-util/src/crypto_helper.rs @@ -6,10 +6,17 @@ pub fn generate_keyring() -> Keyring { } pub fn keyring_to_base64(keyring: &Keyring) -> String { +<<<<<<< HEAD keyring .try_to_bytes() .map(|bytes| STANDARD.encode(bytes)) .unwrap_or_default() +======= + let bytes = keyring + .try_to_bytes() + .expect("keyring fields must fit the wire format"); + STANDARD.encode(bytes) +>>>>>>> refs/remotes/origin/main } pub fn keyring_from_base64(s: &str) -> Option { @@ -18,10 +25,17 @@ pub fn keyring_from_base64(s: &str) -> Option { } pub fn public_key_bundle_to_base64(bundle: &PublicKeyBundle) -> String { +<<<<<<< HEAD bundle .try_as_bytes() .map(|bytes| STANDARD.encode(bytes)) .unwrap_or_default() +======= + let bytes = bundle + .try_as_bytes() + .expect("public key bundle fields must fit the wire format"); + STANDARD.encode(bytes) +>>>>>>> refs/remotes/origin/main } pub fn public_key_bundle_from_base64(s: &str) -> Option { diff --git a/iota-util/src/mtp_compat.rs b/iota-util/src/mtp_compat.rs index 02aced4..adc47d3 100644 --- a/iota-util/src/mtp_compat.rs +++ b/iota-util/src/mtp_compat.rs @@ -1,28 +1,42 @@ use mtp::codec::{CommunicationValue, DataValue}; use mtp::type_map::DataTypeId; -/* - * Keep legacy control-plane handlers source-compatible while they migrate to - * MTP's explicit optional routing fields. Relay handlers must use sender() and - * receiver() directly so an absent outer sender cannot become an identity. - */ -pub trait CommunicationValueCompat { - fn get_id(&self) -> u32; - fn get_sender(&self) -> u64; - fn get_receiver(&self) -> u64; +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum MtpFieldError { + MissingId, + MissingSender, + MissingReceiver, } -impl CommunicationValueCompat for CommunicationValue { - fn get_id(&self) -> u32 { - self.id().unwrap_or_default() +impl std::fmt::Display for MtpFieldError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::MissingId => "missing message id", + Self::MissingSender => "missing sender", + Self::MissingReceiver => "missing receiver", + }) + } +} + +impl std::error::Error for MtpFieldError {} + +pub trait RequiredCommunicationFields { + fn require_id(&self) -> Result; + fn require_sender(&self) -> Result; + fn require_receiver(&self) -> Result; +} + +impl RequiredCommunicationFields for CommunicationValue { + fn require_id(&self) -> Result { + self.id().ok_or(MtpFieldError::MissingId) } - fn get_sender(&self) -> u64 { - self.sender().unwrap_or_default() + fn require_sender(&self) -> Result { + self.sender().ok_or(MtpFieldError::MissingSender) } - fn get_receiver(&self) -> u64 { - self.receiver().unwrap_or_default() + fn require_receiver(&self) -> Result { + self.receiver().ok_or(MtpFieldError::MissingReceiver) } } @@ -65,3 +79,33 @@ impl<'a> OptionalDataValueExt<'a> for Option<&'a DataValue> { self.and_then(DataValue::as_container) } } + +#[cfg(test)] +mod tests { + use super::{MtpFieldError, RequiredCommunicationFields}; + use mtp::codec::{CommunicationType, CommunicationValue}; + + #[test] + fn missing_routing_fields_are_reported_instead_of_defaulted() { + let message = CommunicationValue::new(CommunicationType::Success).without_id(); + + assert_eq!(message.require_id(), Err(MtpFieldError::MissingId)); + assert_eq!(message.require_sender(), Err(MtpFieldError::MissingSender)); + assert_eq!( + message.require_receiver(), + Err(MtpFieldError::MissingReceiver) + ); + } + + #[test] + fn present_routing_fields_are_returned_unchanged() { + let message = CommunicationValue::new(CommunicationType::Success) + .with_id(7) + .with_sender(8) + .with_receiver(9); + + assert_eq!(message.require_id(), Ok(7)); + assert_eq!(message.require_sender(), Ok(8)); + assert_eq!(message.require_receiver(), Ok(9)); + } +} diff --git a/omikron-connector/Cargo.toml b/omikron-connector/Cargo.toml index 5902891..c4773f7 100644 --- a/omikron-connector/Cargo.toml +++ b/omikron-connector/Cargo.toml @@ -14,7 +14,6 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = [ "client", "crypto", "files", - "raw", ] } dashmap = "6.2.1" diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs old mode 100755 new mode 100644 index 1aae831..9599cb6 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -9,12 +9,12 @@ use iota_util::crypto_util::{self}; use mtp::client::{Client, ClientConfig, MTPConnection, Policy, SendMode, Sender}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; 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, - atomic::{AtomicU32, Ordering}, -}; +use std::sync::{Arc, LazyLock}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::sync::{Mutex, RwLock, Semaphore, oneshot, watch}; use tokio::task::JoinHandle; @@ -38,6 +38,9 @@ use iota_util::route_target::RouteTarget; // ============================================================================ const IOTA_KEYRING_PATH: &str = "iota.mk"; +const IDENTITY_SECRET_ENV: &str = "IOTA_IDENTITY_SECRET"; +const IDENTITY_SECRET_FILE_ENV: &str = "IOTA_IDENTITY_SECRET_FILE"; +const SYSTEMD_IDENTITY_CREDENTIAL: &str = "iota-identity"; static IDENTITY_PATH: std::sync::OnceLock = std::sync::OnceLock::new(); static OMIKRON_PUBLIC_KEY_PATH: std::sync::OnceLock = std::sync::OnceLock::new(); @@ -111,7 +114,173 @@ const TASK_CLEANUP_INTERVAL: Duration = Duration::from_secs(60); const TASK_MAX_AGE: Duration = Duration::from_secs(60); const MAX_CONCURRENT_HANDLERS: usize = 20; const RELAY_RETENTION_MILLIS: i64 = 30 * 24 * 60 * 60 * 1000; -static NEXT_RELAY_FRAME_ID: AtomicU32 = AtomicU32::new(1); + +#[derive(Debug)] +pub enum IdentityError { + Storage(mtp::files::FileError), + Directory(std::io::Error), + Secret(String), + InvalidLegacyIdentity, + Verification(String), +} + +impl std::fmt::Display for IdentityError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + 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}") + } + } + } +} + +impl std::error::Error for IdentityError {} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ConnectionAttemptResult { + became_healthy: bool, +} + +fn jittered_reconnect_delay(delay: Duration) -> Duration { + let ceiling_ms = u64::try_from(MAX_RECONNECT_DELAY.as_millis()) + .expect("reconnect ceiling must fit in milliseconds"); + let base_ms = u64::try_from(delay.as_millis().min(u128::from(ceiling_ms))) + .expect("bounded reconnect delay must fit in milliseconds"); + let jitter_span = base_ms / 5; + if jitter_span == 0 { + return Duration::from_millis(base_ms); + } + + let mut rng = rand_core::OsRng; + let range = jitter_span.saturating_mul(2).saturating_add(1); + let offset = (rng.next_u64() % range) as i128 - jitter_span as i128; + let jittered = (base_ms as i128 + offset).clamp(0, i128::from(ceiling_ms)); + Duration::from_millis(u64::try_from(jittered).expect("bounded jitter must be non-negative")) +} + +fn wire_user_id(user_id: i64) -> u64 { + u64::try_from(user_id).expect("validated user ID is non-negative") +} + +fn load_identity_secret() -> Result, IdentityError> { + if let Some(path) = env::var_os(IDENTITY_SECRET_FILE_ENV) { + let path = PathBuf::from(path); + let mut secret = fs::read(&path) + .map_err(|error| IdentityError::Secret(format!("{}: {error}", path.display())))?; + while matches!(secret.last(), Some(b'\n' | b'\r')) { + secret.pop(); + } + if secret.is_empty() { + return Err(IdentityError::Secret(format!( + "{} is empty", + path.display() + ))); + } + return Ok(secret); + } + + if let Ok(secret) = env::var(IDENTITY_SECRET_ENV) { + if secret.is_empty() { + return Err(IdentityError::Secret(format!( + "{IDENTITY_SECRET_ENV} is empty" + ))); + } + return Ok(secret.into_bytes()); + } + + if let Ok(credentials_dir) = env::var("CREDENTIALS_DIRECTORY") { + let path = Path::new(&credentials_dir).join(SYSTEMD_IDENTITY_CREDENTIAL); + let mut secret = fs::read(&path) + .map_err(|error| IdentityError::Secret(format!("{}: {error}", path.display())))?; + while matches!(secret.last(), Some(b'\n' | b'\r')) { + secret.pop(); + } + if secret.is_empty() { + return Err(IdentityError::Secret(format!( + "{} is empty", + path.display() + ))); + } + return Ok(secret); + } + + Err(IdentityError::Secret(format!( + "set {IDENTITY_SECRET_FILE_ENV}, {IDENTITY_SECRET_ENV}, or a systemd identity credential" + ))) +} + +fn load_legacy_raw_keyring(path: &Path) -> Result { + let bytes = fs::read(path).map_err(|error| IdentityError::Storage(error.into()))?; + if bytes.len() < 5 || bytes[..4] != *b"MTMK" || bytes[4] != 1 { + return Err(IdentityError::InvalidLegacyIdentity); + } + Keyring::from_bytes(&bytes[5..]).map_err(|_| IdentityError::InvalidLegacyIdentity) +} + +fn save_protected_keyring_verified( + keyring: &Keyring, + path: &Path, + passphrase: &[u8], +) -> Result<(), IdentityError> { + mtp::files::save_keyring(keyring, path, passphrase).map_err(IdentityError::Storage)?; + let persisted = mtp::files::load_keyring(path, passphrase).map_err(IdentityError::Storage)?; + let expected = keyring + .try_to_bytes() + .map_err(|error| IdentityError::Verification(error.to_string()))?; + let actual = persisted + .try_to_bytes() + .map_err(|error| IdentityError::Verification(error.to_string()))?; + if expected != actual { + return Err(IdentityError::Verification( + "persisted keyring differs from the requested identity".into(), + )); + } + Ok(()) +} + +fn load_or_migrate_keyring_at( + path: &Path, + legacy: Option, + passphrase: &[u8], +) -> Result { + if let Some(parent) = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + fs::create_dir_all(parent).map_err(IdentityError::Directory)?; + } + + match mtp::files::load_keyring(path, passphrase) { + 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(error) => return Err(IdentityError::Storage(error)), + } + + let keyring = match legacy { + Some(encoded) => { + keyring_from_base64(&encoded).ok_or(IdentityError::InvalidLegacyIdentity)? + } + None => { + log!( + "No existing Iota identity found at {}; generating a new identity", + path.display() + ); + crypto_helper::generate_keyring() + } + }; + + save_protected_keyring_verified(&keyring, path, passphrase)?; + Ok(keyring) +} // ============================================================================ // Waiting Task System @@ -179,14 +348,6 @@ pub struct OmikronConnection { pub(crate) app: Arc>, } -fn ensure_relay_frame_id(frame: CommunicationValue) -> CommunicationValue { - if frame.id().is_some_and(|id| id != 0) { - return frame; - } - let id = NEXT_RELAY_FRAME_ID.fetch_add(1, Ordering::Relaxed).max(1); - frame.with_id(id) -} - impl OmikronConnection { pub fn new(active_tasks: Arc>, app: Arc>) -> Self { Self::with_cancellation(CancellationToken::new(), active_tasks, app) @@ -299,29 +460,29 @@ impl OmikronConnection { break; } - match self.clone().connect_once().await { - Ok(()) => { - if *self.reconnect_on_close.read().await { - log!("Connection lost, reconnecting in {:?}...", reconnect_delay); - } else { + let retry_reason = match self.clone().connect_once().await { + Ok(result) => { + if result.became_healthy { + reconnect_delay = RECONNECT_DELAY; + } + if !*self.reconnect_on_close.read().await { break; } + "Connection lost".to_string() } Err(e) => { if self.auth_failure.read().await.is_some() { log!("Authentication failed, stopping reconnection: {}", e); break; } - log!( - "Connection failed: {}, retrying in {:?}...", - e, - reconnect_delay - ); + format!("Connection failed: {e}") } - } + }; + let delay = jittered_reconnect_delay(reconnect_delay); + log!("{}, retrying in {:?}...", retry_reason, delay); tokio::select! { - _ = sleep(reconnect_delay) => {} + _ = sleep(delay) => {} _ = shutdown_rx.changed() => { if *shutdown_rx.borrow() { break; @@ -333,11 +494,16 @@ impl OmikronConnection { } } - async fn connect_once(self: Arc) -> Result<(), String> { + async fn connect_once(self: Arc) -> Result { self.set_state(ConnectionState::Connecting).await; log_t!("omikron_connecting"); - let keyring = Arc::new(self.load_or_migrate_keyring().await); + let identity_secret = load_identity_secret().map_err(|error| error.to_string())?; + let keyring = Arc::new( + self.load_or_migrate_keyring(&identity_secret) + .await + .map_err(|error| format!("Iota identity initialization failed: {error}"))?, + ); *self.keyring.write().await = Some(keyring.clone()); let existing_iota_id = CONFIG.load().iota_id; @@ -349,29 +515,21 @@ impl OmikronConnection { log!("Connecting to Omikron at {}", addr_str); + let policy = Policy::default() + .with_send_mode(SendMode::SingleStreamPerMessage) + .with_timeouts( + Duration::from_millis(2_000), + Duration::from_millis(2_000), + Duration::from_millis(30_000), + ) + .with_keep_alive(Some(Duration::from_secs(6))) + .with_receiver_queue_capacity(1000) + .with_max_concurrent_stream_tasks(10) + .with_persistent_stream_retries(5, Duration::from_secs(5)); let client_config = ClientConfig::new(&addr_str) .with_description("iota") - .with_policy(Policy { - send_mode: SendMode::SingleStreamPerMessage, - max_message_size: 1_000_000_000, - handshake_max_message_size: 1_000_000_000, - close_frame_len: u32::MAX, - application_close_code: 0, - open_stream_timeout: Duration::from_millis(2_000), - write_timeout: Duration::from_millis(2_000), - accept_stream_timeout: Duration::from_millis(10_000), - read_timeout: Duration::from_millis(30_000), - keep_alive_interval: Some(Duration::from_secs(6)), - max_idle_timeout: None, - force_close_delay: Duration::from_millis(300), - receiver_queue_capacity: 1000, - max_concurrent_stream_tasks: 10, - persistent_stream_max_retries: 5, - persistent_stream_retry_backoff: Duration::from_secs(5), - max_frames_per_stream: None, - }) - .with_ping_interval(MAINTENANCE_INTERVAL) - .with_max_missed_pings(0); + .with_policy(policy) + .with_ping_interval(MAINTENANCE_INTERVAL); let connection = match Client::auth_connect_or_register( client_config, @@ -440,13 +598,9 @@ impl OmikronConnection { } match result { - Ok(()) => { - if *self.reconnect_on_close.read().await { - Err("Connection closed, will reconnect".to_string()) - } else { - Ok(()) - } - } + Ok(()) => Ok(ConnectionAttemptResult { + became_healthy: true, + }), Err(e) => Err(format!("Read loop error: {}", e)), } } @@ -455,6 +609,7 @@ impl OmikronConnection { // Identity (own Keyring, migrated from the legacy base64-in-config format) // ------------------------------------------------------------------------- +<<<<<<< HEAD /* * `iota.mk` is now the source of truth for this Iota's identity. A * pre-existing base64 keyring in config.json (from before the MTP auth @@ -490,6 +645,10 @@ impl OmikronConnection { } keyring +======= + async fn load_or_migrate_keyring(&self, passphrase: &[u8]) -> Result { + load_or_migrate_keyring_at(identity_path(), CONFIG.load().keyring.clone(), passphrase) +>>>>>>> refs/remotes/origin/main } // ------------------------------------------------------------------------- @@ -562,8 +721,12 @@ impl OmikronConnection { }; let (host, port, public_key) = if let Some(endpoint) = discovered { + let discovered_key_bytes = endpoint.public_key.try_as_bytes().map_err(|error| { + format!("Failed to serialize discovered Omikron public key: {error}") + })?; match &cached_key { Some(cached) => { +<<<<<<< HEAD let keys_match = match (cached.try_as_bytes(), endpoint.public_key.try_as_bytes()) { (Ok(cached_bytes), Ok(discovered_bytes)) => { @@ -589,6 +752,20 @@ impl OmikronConnection { } else { (endpoint.host, endpoint.port, cached.clone()) } +======= + let cached_key_bytes = cached.try_as_bytes().map_err(|error| { + format!("Failed to serialize cached Omikron public key: {error}") + })?; + if cached_key_bytes != discovered_key_bytes { + log!( + "Fetched Omikron public key differs from the cached {} - keeping the \ + cached key. Delete {} manually if this is an expected key rotation.", + OMIKRON_PUBLIC_KEY_PATH, + OMIKRON_PUBLIC_KEY_PATH + ); + } + (endpoint.host, endpoint.port, cached.clone()) +>>>>>>> refs/remotes/origin/main } None => { if let Err(e) = save_omikron_public_key(&endpoint.public_key, key_path) { @@ -636,7 +813,13 @@ impl OmikronConnection { }); continue; } - let msg_id = cv.get_id(); + let Some(msg_id) = cv.id() else { + let self_clone = self.clone(); + tokio::spawn(async move { + self_clone.handle_message_impl(cv).await; + }); + continue; + }; if let Some((_, task)) = WAITING_TASKS.remove(&msg_id) { if (task.task)(cv.clone()) { continue; @@ -686,7 +869,7 @@ impl OmikronConnection { } if let Some(ping) = connection.get_ping() { - let ping_ms = ping.as_millis() as i64; + let ping_ms = i64::try_from(ping.as_millis()).unwrap_or(i64::MAX); *self.last_ping.lock().await = ping_ms; self.app.lock().unwrap().push_ping_val(ping_ms as f64); } @@ -704,7 +887,10 @@ impl OmikronConnection { &self, signer_id: u64, ) -> Result, RelayValidationError> { - if let Some(user) = iota_storage::users::user_manager::get_user(signer_id as i64) { + let signer_id_i64 = i64::try_from(signer_id).map_err(|_| { + RelayValidationError::KeyLookup("signer ID exceeds local storage range".into()) + })?; + if let Some(user) = iota_storage::users::user_manager::get_user(signer_id_i64) { let key = iota_util::crypto_helper::public_key_bundle_from_base64(&user.public_key) .ok_or_else(|| { RelayValidationError::KeyLookup("stored user key is invalid".into()) @@ -714,7 +900,7 @@ impl OmikronConnection { let request = CommunicationValue::new(CommunicationType::GetUserData).add_typed_default( DataType::UserId, - DataValue::UnsignedNumber(signer_id as u128), + DataValue::UnsignedNumber(u128::from(signer_id)), ); let response = self .await_response(&request, Some(Duration::from_secs(10))) @@ -736,15 +922,19 @@ impl OmikronConnection { } pub async fn hosting_iota_for_user(&self, user_id: u64) -> Result { - if iota_storage::users::user_manager::get_user(user_id as i64).is_some() { + let user_id_i64 = i64::try_from(user_id) + .map_err(|_| "user ID exceeds local storage range".to_string())?; + if iota_storage::users::user_manager::get_user(user_id_i64).is_some() { return CONFIG .load() .iota_id .ok_or_else(|| "Iota identity is not configured".into()); } - let request = CommunicationValue::new(CommunicationType::GetUserData) - .add_typed_default(DataType::UserId, DataValue::UnsignedNumber(user_id as u128)); + let request = CommunicationValue::new(CommunicationType::GetUserData).add_typed_default( + DataType::UserId, + DataValue::UnsignedNumber(u128::from(user_id)), + ); let response = self .await_response(&request, Some(Duration::from_secs(10))) .await?; @@ -766,17 +956,19 @@ impl OmikronConnection { } async fn handle_relay(self: Arc, frame: CommunicationValue) { - let frame = ensure_relay_frame_id(frame); - let incoming_frame_id = frame.id(); + let Some(incoming_frame_id) = frame.id() else { + log!("Rejecting Relay without a message id"); + return; + }; let Some(local_iota_id) = CONFIG.load().iota_id else { log!("Rejecting Relay because this Iota has no registered identity"); - self.send_relay_response(incoming_frame_id, CommunicationType::ErrorInvalidData) + self.send_relay_response(Some(incoming_frame_id), CommunicationType::ErrorInvalidData) .await; return; }; let Some(keyring) = self.keyring.read().await.as_ref().cloned() else { log!("Rejecting Relay because the Iota keyring is unavailable"); - self.send_relay_response(incoming_frame_id, CommunicationType::ErrorInternal) + self.send_relay_response(Some(incoming_frame_id), CommunicationType::ErrorInternal) .await; return; }; @@ -798,24 +990,29 @@ impl OmikronConnection { Ok(value) => value, Err(error) => { log!("Relay metadata verification failed: {}", error); - self.send_relay_response(incoming_frame_id, CommunicationType::ErrorInvalidData) - .await; + self.send_relay_response( + Some(incoming_frame_id), + CommunicationType::ErrorInvalidData, + ) + .await; return; } }; - let signer_is_local = - iota_storage::users::user_manager::get_user(verified.context.signer_id as i64) - .is_some(); - let recipient_is_local = - iota_storage::users::user_manager::get_user(verified.context.final_recipient_id as i64) - .is_some(); + let signer_is_local = i64::try_from(verified.context.signer_id) + .ok() + .and_then(iota_storage::users::user_manager::get_user) + .is_some(); + let recipient_is_local = i64::try_from(verified.context.final_recipient_id) + .ok() + .and_then(iota_storage::users::user_manager::get_user) + .is_some(); if !signer_is_local && !recipient_is_local { log!( "Rejecting Relay with no local origin or destination: signer {}, recipient {}", verified.context.signer_id, verified.context.final_recipient_id, ); - self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData) + self.send_relay_response(Some(incoming_frame_id), CommunicationType::ErrorInvalidData) .await; return; } @@ -833,7 +1030,7 @@ impl OmikronConnection { } }; let type_map_version = verified.context.type_map.version.to_string(); - let frame_id = frame.id().unwrap_or_default(); + let frame_id = incoming_frame_id; let reservation = match relay_replay::reserve( verified.context.signer_id, &verified.context.message_id, @@ -1000,7 +1197,7 @@ impl OmikronConnection { RouteTarget::User(destination), &bytes, now_millis_i64(), - forwarded.id().unwrap_or_default(), + frame_id, &type_map_version, ) { log!("Relay could not be queued for client delivery: {}", error); @@ -1142,7 +1339,7 @@ impl OmikronConnection { } // ------------------------------------------------------------------------- - // Message Handling — Dispatch + // Message Handling - Dispatch // ------------------------------------------------------------------------- pub async fn handle_message(self: Arc, cv: CommunicationValue) { @@ -1169,7 +1366,10 @@ impl OmikronConnection { } } - let msg_id = cv.get_id(); + let Some(msg_id) = cv.id() else { + self.handle_message_impl(cv).await; + return; + }; if let Some((_, task)) = WAITING_TASKS.remove(&msg_id) { if (task.task)(cv.clone()) { @@ -1185,6 +1385,12 @@ impl OmikronConnection { self.handle_relay(cv).await; return; } + if cv.require_id().is_err() { + let _ = self + .send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } if matches!( iota_connection::relay::message_security_class(&cv), @@ -1264,7 +1470,7 @@ impl OmikronConnection { } let acknowledgement = CommunicationValue::new(CommunicationType::EraseHostedUserDataAck) - .with_id(cv.get_id()) + .with_request_id(cv) .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())); let _ = self.send_message(&acknowledgement).await; } @@ -1276,7 +1482,15 @@ impl OmikronConnection { } async fn handle_app_identification(self: Arc, cv: &CommunicationValue) { - let sender_id = cv.get_sender(); + let sender_id = match cv.require_sender() { + Ok(sender_id) => sender_id, + Err(_) => { + let _ = self + .send_message(&error_response(cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } + }; let app_identifier = cv .get_data(DataType::AppIdentifier) .as_str() @@ -1287,7 +1501,12 @@ impl OmikronConnection { .as_str() .unwrap_or("") .to_string(); - let user_id = cv.get_data(DataType::UserId).as_number().unwrap_or(0) as i64; + let Some(user_id) = data_i64(cv, DataType::UserId).filter(|id| *id > 0) else { + let _ = self + .send_message(&error_response(cv, CommunicationType::ErrorInvalidData)) + .await; + return; + }; let mut trusted = false; if let Some(user) = iota_storage::users::user_manager::get_user(user_id) { @@ -1308,7 +1527,8 @@ impl OmikronConnection { if let Some(app_pub_bundle) = iota_util::crypto_helper::public_key_bundle_from_base64(&app_public_key) { - if let Ok(keyring) = mtp::files::load_keyring_raw(identity_path()) { + let keyring = self.keyring.read().await.as_ref().cloned(); + if let Some(keyring) = keyring { if let Ok(encrypted_challenge) = crypto_util::encrypt_challenge(&challenge, &app_pub_bundle) { @@ -1316,7 +1536,7 @@ impl OmikronConnection { let pub_k_b64 = crypto_helper::public_key_bundle_to_base64(&bundle); let res = CommunicationValue::new(CommunicationType::AppChallenge) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(sender_id) .add_typed_default(DataType::PublicKey, DataValue::Str(pub_k_b64)) .add_typed_default( @@ -1332,18 +1552,26 @@ impl OmikronConnection { } let res = CommunicationValue::new(CommunicationType::ErrorInvalidChallenge) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(sender_id); let _ = self.send_message(&res).await; } async fn handle_app_challenge_response(self: Arc, cv: &CommunicationValue) { - let sender_id = cv.get_sender(); + let sender_id = match cv.require_sender() { + Ok(sender_id) => sender_id, + Err(_) => { + let _ = self + .send_message(&error_response(cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } + }; if let Some((_, expected_challenge)) = self.app_challenges.remove(&sender_id) { if let Some(DataValue::Str(response)) = cv.get_data(DataType::Challenge) { if expected_challenge == *response { let res = CommunicationValue::new(CommunicationType::AppIdentificationResponse) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(sender_id); let _ = self.send_message(&res).await; return; @@ -1351,13 +1579,21 @@ impl OmikronConnection { } } let res = CommunicationValue::new(CommunicationType::ErrorInvalidChallenge) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(sender_id); let _ = self.send_message(&res).await; } async fn handle_save_app_data(self: Arc, cv: &CommunicationValue) { - let sender_id = cv.get_sender(); + let sender_id = match cv.require_sender() { + Ok(sender_id) => sender_id, + Err(_) => { + let _ = self + .send_message(&error_response(cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } + }; let app_data = cv .get_data(DataType::AppData) .as_str() @@ -1370,13 +1606,21 @@ impl OmikronConnection { } let res = CommunicationValue::new(CommunicationType::SaveAppData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(sender_id); let _ = self.send_message(&res).await; } async fn handle_load_app_data(self: Arc, cv: &CommunicationValue) { - let sender_id = cv.get_sender(); + let sender_id = match cv.require_sender() { + Ok(sender_id) => sender_id, + Err(_) => { + let _ = self + .send_message(&error_response(cv, CommunicationType::ErrorInvalidData)) + .await; + return; + } + }; let mut app_data = String::new(); if let Some(session) = self.app_sessions.get(&sender_id) { @@ -1385,7 +1629,7 @@ impl OmikronConnection { } let res = CommunicationValue::new(CommunicationType::LoadAppData) - .with_id(cv.get_id()) + .with_request_id(cv) .with_receiver(sender_id) .add_typed_default(DataType::AppData, DataValue::Str(app_data)); let _ = self.send_message(&res).await; @@ -1426,9 +1670,9 @@ impl OmikronConnection { extra: Vec<(DataType, DataValue)>, ) -> CommunicationValue { let mut message = CommunicationValue::new(ty) - .with_id(request.get_id()) - .with_sender(mutation.sender_id as u64) - .with_receiver(mutation.partner_id as u64) + .with_request_id(request) + .with_sender(wire_user_id(mutation.sender_id)) + .with_receiver(wire_user_id(mutation.partner_id)) .add_typed_default( DataType::ChatPartnerId, DataValue::SignedNumber(mutation.sender_id as i128), @@ -1444,18 +1688,26 @@ impl OmikronConnection { } async fn persist_and_deliver_remote_edit(&self, cv: &CommunicationValue) { - let sender_id = match i64::try_from(cv.get_sender()) { - Ok(sender_id) => sender_id, - Err(_) => return, + let sender_id = match cv + .require_sender() + .ok() + .and_then(|id| i64::try_from(id).ok()) + { + Some(sender_id) => sender_id, + None => return, }; - let receiver_id = match i64::try_from(cv.get_receiver()) { - Ok(receiver_id) if receiver_id > 0 => receiver_id, + let receiver_id = match cv + .require_receiver() + .ok() + .and_then(|id| i64::try_from(id).ok()) + { + Some(receiver_id) if receiver_id > 0 => receiver_id, _ => return, }; let Some(send_time) = data_i64(cv, DataType::SendTime).filter(|time| *time > 0) else { return; }; - let Some(content) = cv.get_data(DataType::Content).as_str() else { + let Some(content) = cv.get_data(DataType::AppContent).as_str() else { return; }; if chat_files::apply_remote_edit(receiver_id, sender_id, send_time, sender_id, content) @@ -1466,12 +1718,20 @@ impl OmikronConnection { } async fn persist_and_deliver_remote_reaction(&self, cv: &CommunicationValue, add: bool) { - let sender_id = match i64::try_from(cv.get_sender()) { - Ok(sender_id) => sender_id, - Err(_) => return, + let sender_id = match cv + .require_sender() + .ok() + .and_then(|id| i64::try_from(id).ok()) + { + Some(sender_id) => sender_id, + None => return, }; - let receiver_id = match i64::try_from(cv.get_receiver()) { - Ok(receiver_id) if receiver_id > 0 => receiver_id, + let receiver_id = match cv + .require_receiver() + .ok() + .and_then(|id| i64::try_from(id).ok()) + { + Some(receiver_id) if receiver_id > 0 => receiver_id, _ => return, }; let Some(send_time) = data_i64(cv, DataType::SendTime).filter(|time| *time > 0) else { @@ -1494,12 +1754,20 @@ impl OmikronConnection { } async fn persist_and_deliver_remote_delete(&self, cv: &CommunicationValue) { - let sender_id = match i64::try_from(cv.get_sender()) { - Ok(sender_id) => sender_id, - Err(_) => return, + let sender_id = match cv + .require_sender() + .ok() + .and_then(|id| i64::try_from(id).ok()) + { + Some(sender_id) => sender_id, + None => return, }; - let receiver_id = match i64::try_from(cv.get_receiver()) { - Ok(receiver_id) if receiver_id > 0 => receiver_id, + let receiver_id = match cv + .require_receiver() + .ok() + .and_then(|id| i64::try_from(id).ok()) + { + Some(receiver_id) if receiver_id > 0 => receiver_id, _ => return, }; let Some(send_time) = data_i64(cv, DataType::SendTime).filter(|time| *time > 0) else { @@ -1522,7 +1790,7 @@ impl OmikronConnection { .await; return; }; - let Some(content) = cv.get_data(DataType::Content).as_str() else { + let Some(content) = cv.get_data(DataType::AppContent).as_str() else { let _ = self .send_message(&error_response(cv, CommunicationType::ErrorInvalidData)) .await; @@ -1532,7 +1800,7 @@ impl OmikronConnection { CommunicationType::MessageEditLive, cv, &mutation, - vec![(DataType::Content, DataValue::Str(content.to_string()))], + vec![(DataType::AppContent, DataValue::Str(content.to_string()))], ); if iota_storage::users::user_manager::get_user(mutation.partner_id).is_some() && chat_files::apply_remote_edit( @@ -1631,9 +1899,13 @@ impl OmikronConnection { } async fn handle_message_delete_live(self: Arc, cv: &CommunicationValue) { - let sender_id = match i64::try_from(cv.get_sender()) { - Ok(sender_id) => sender_id, - Err(_) => return, + let sender_id = match cv + .require_sender() + .ok() + .and_then(|id| i64::try_from(id).ok()) + { + Some(sender_id) => sender_id, + None => return, }; if iota_storage::users::user_manager::get_user(sender_id).is_none() { self.persist_and_deliver_remote_delete(cv).await; @@ -1814,7 +2086,9 @@ impl OmikronConnection { timeout_duration: Option, ) -> Result { let (tx, rx) = oneshot::channel(); - let msg_id = cv.get_id(); + let msg_id = cv + .require_id() + .map_err(|error| format!("cannot await response without a message id: {error}"))?; WAITING_TASKS.insert( msg_id, @@ -1931,6 +2205,8 @@ impl OmikronConnection { /// recovery does not silently destroy the user's previous identity. pub async fn rotate_identity(self: &Arc) -> Result<(), OmikronError> { log!("Iota identity rotation requested"); + let identity_secret = + load_identity_secret().map_err(|error| OmikronError::Internal(error.to_string()))?; self.stop().await; let path = identity_path(); @@ -1950,7 +2226,10 @@ impl OmikronConnection { } let keyring = crypto_helper::generate_keyring(); - if let Some(parent) = path.parent() { + if let Some(parent) = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { std::fs::create_dir_all(parent).map_err(|error| { OmikronError::Internal(format!( "could not create identity directory {}: {error}", @@ -1958,7 +2237,11 @@ impl OmikronConnection { )) })?; } +<<<<<<< HEAD save_keyring(&keyring, path).map_err(|error| { +======= + save_protected_keyring_verified(&keyring, path, &identity_secret).map_err(|error| { +>>>>>>> refs/remotes/origin/main OmikronError::Internal(format!( "could not save new identity {}: {error}", path.display() @@ -2127,6 +2410,7 @@ impl OmikronClient for OmikronConnection { } #[cfg(test)] +<<<<<<< HEAD mod tests { use super::*; @@ -2145,5 +2429,80 @@ mod tests { loaded.try_to_bytes().unwrap() ); std::fs::remove_dir_all(directory).unwrap(); +======= +mod identity_tests { + use super::*; + + fn test_path(name: &str) -> PathBuf { + std::env::temp_dir().join(format!( + "iota-identity-{name}-{}-{}", + std::process::id(), + Uuid::new_v4() + )) + } + + #[test] + fn generated_identity_is_protected_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"); + 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()); + let _ = fs::remove_file(path); + } + + #[test] + 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"); + assert!(matches!(error, IdentityError::Storage(_))); + let _ = fs::remove_file(path); + } + + #[test] + fn legacy_raw_identity_is_migrated_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(); + raw.push(1); + 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"); + assert_eq!( + migrated.try_to_bytes().expect("keyring serializes"), + keyring.try_to_bytes().expect("keyring serializes") + ); + let _ = fs::remove_file(path); + } + + #[test] + fn identity_directory_failure_is_returned() { + 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") + .expect_err("directory failure must be returned"); + assert!(matches!(error, IdentityError::Directory(_))); + let _ = fs::remove_file(parent); + } + + #[test] + fn reconnect_jitter_stays_bounded_by_the_exponential_delay_ceiling() { + for _ in 0..32 { + let delay = jittered_reconnect_delay(Duration::from_secs(5)); + assert!(delay >= Duration::from_secs(4)); + assert!(delay <= Duration::from_secs(6)); + } + + assert!(jittered_reconnect_delay(Duration::from_secs(600)) <= MAX_RECONNECT_DELAY); +>>>>>>> refs/remotes/origin/main } } diff --git a/omikron-connector/src/user_ops.rs b/omikron-connector/src/user_ops.rs index 3008120..60f8429 100644 --- a/omikron-connector/src/user_ops.rs +++ b/omikron-connector/src/user_ops.rs @@ -377,7 +377,7 @@ mod tests { use super::{CreateUserError, request_user_id, valid_username}; use crate::{OmikronClient, OmikronError}; use async_trait::async_trait; - use iota_util::mtp_compat::CommunicationValueCompat; + use iota_connection::message_common::CommunicationResponseExt; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use std::time::Duration; @@ -397,7 +397,7 @@ mod tests { _: Duration, ) -> Result { assert!(request.is_type(CommunicationType::GetRegister)); - Ok(self.response.clone().with_id(request.get_id())) + Ok(self.response.clone().with_request_id(request)) } async fn reconnect(&self) -> Result<(), OmikronError> { diff --git a/systemd/iota-daemon.service b/systemd/iota-daemon.service index a5a2903..2c6340b 100644 --- a/systemd/iota-daemon.service +++ b/systemd/iota-daemon.service @@ -17,6 +17,7 @@ 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