From caa0572c3aef1a6c6e3ffc63a0111b81d56c85ad Mon Sep 17 00:00:00 2001 From: Alois Date: Sun, 5 Jul 2026 21:45:44 +0200 Subject: [PATCH] (feat): crypto migrations --- client/src/client_connection.rs | 479 +++++++++----------- iota-logger/src/lib.rs | 3 +- iota-storage/src/util/e2ee_storage.rs | 411 +++++++++++++++++ iota-storage/src/util/mod.rs | 1 + omikron-connector/src/omikron_connection.rs | 222 +++++++++ type-maps.yaml | 29 ++ 6 files changed, 890 insertions(+), 255 deletions(-) create mode 100644 iota-storage/src/util/e2ee_storage.rs diff --git a/client/src/client_connection.rs b/client/src/client_connection.rs index a386a4e..390a969 100644 --- a/client/src/client_connection.rs +++ b/client/src/client_connection.rs @@ -7,11 +7,16 @@ use iota_storage::util::chats_util::{get_user, mod_user}; use iota_storage::util::communities_util::CommunitiesUtil; use iota_storage::util::config_util::CONFIG; use iota_storage::util::{chat_files, chats_util}; +use iota_storage::util::e2ee_storage::{ + self, EncryptedDeviceSecretQuery, EncryptedMessageQuery, + StoredEncryptedDeviceSecret, StoredEncryptedMessage, +}; use iota_util::crypto_helper::keyring_from_base64; use iota_util::crypto_util::{self}; use iota_util::file_util::{get_children, load_file, save_file}; use mtp::client::{Receiver, Sender}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; + use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::{Mutex, RwLock, mpsc, watch}; @@ -29,6 +34,41 @@ fn typed_container(items: Vec<(DataType, DataValue)>) -> DataValue { ) } + + +fn data_string(cv: &CommunicationValue, dt: DataType) -> Option { + cv.get_data(dt) + .as_str() + .map(|s| s.to_string()) + .or_else(|| cv.get_data(dt).as_number().map(|n| n.to_string())) + .or_else(|| cv.get_data(dt).as_signed_number().map(|n| n.to_string())) +} + +fn data_i64(cv: &CommunicationValue, dt: DataType) -> Option { + cv.get_data(dt) + .as_number() + .and_then(|n| i64::try_from(n).ok()) + .or_else(|| cv.get_data(dt).as_signed_number().and_then(|n| i64::try_from(n).ok())) + .or_else(|| cv.get_data(dt).as_str().and_then(|s| s.parse::().ok())) +} + +fn data_bytes(cv: &CommunicationValue, dt: DataType) -> Option> { + cv.get_bytes(dt).map(|bytes| bytes.to_vec()) +} + +fn now_millis_i64() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 +} + +fn error_response(request: &CommunicationValue, ty: CommunicationType) -> CommunicationValue { + CommunicationValue::new(ty) + .with_id(request.get_id()) + .with_receiver(request.get_sender()) +} + // ============================================================================ // Waiting Task System // ============================================================================ @@ -142,6 +182,190 @@ impl ClientConnection { return; } + if cv.is_type(CommunicationType::SetEncryptedDeviceSecret) { + let sender_id = cv.get_sender().to_string(); + if data_string(&cv, DataType::UserId).as_deref() != Some(sender_id.as_str()) { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + return; + } + let now = now_millis_i64(); + let record = data_string(&cv, DataType::UserId) + .zip(data_string(&cv, DataType::DeviceId)) + .zip(data_string(&cv, DataType::SecretId)) + .zip(data_i64(&cv, DataType::VersionNumber)) + .zip(data_bytes(&cv, DataType::EncryptedSecret)) + .zip(data_string(&cv, DataType::WrappingScheme)) + .map(|(((((user_id, device_id), secret_id), version), encrypted_secret), wrapping_scheme)| { + StoredEncryptedDeviceSecret { + user_id, + device_id, + secret_id, + version, + encrypted_secret, + wrapping_public_key_id: data_string(&cv, DataType::WrappingPublicKeyId), + wrapping_scheme, + created_at: data_i64(&cv, DataType::CreatedAt).unwrap_or(now), + updated_at: now, + } + }); + + match record.map(e2ee_storage::put_encrypted_device_secret) { + Some(Ok(())) => { + self.send_message(&error_response(&cv, CommunicationType::Success)).await; + } + _ => { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + } + } + return; + } + + if cv.is_type(CommunicationType::GetEncryptedDeviceSecret) { + let Some(user_id) = data_string(&cv, DataType::UserId) else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + return; + }; + let sender_id = cv.get_sender().to_string(); + if user_id != sender_id { + self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound)).await; + return; + } + + match e2ee_storage::get_encrypted_device_secret(EncryptedDeviceSecretQuery { + user_id, + device_id: data_string(&cv, DataType::DeviceId), + secret_id: data_string(&cv, DataType::SecretId), + }) { + Ok(Some(record)) => { + let mut response = CommunicationValue::new(CommunicationType::EncryptedDeviceSecretResponse) + .with_id(cv.get_id()) + .with_receiver(cv.get_sender()) + .add_typed_default(DataType::UserId, DataValue::Str(record.user_id)) + .add_typed_default(DataType::DeviceId, DataValue::Str(record.device_id)) + .add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id)) + .add_typed_default(DataType::VersionNumber, DataValue::SignedNumber(record.version as i128)) + .add_typed_default(DataType::EncryptedSecret, DataValue::Bytes(record.encrypted_secret)) + .add_typed_default(DataType::WrappingScheme, DataValue::Str(record.wrapping_scheme)) + .add_typed_default(DataType::CreatedAt, DataValue::SignedNumber(record.created_at as i128)) + .add_typed_default(DataType::UpdatedAt, DataValue::SignedNumber(record.updated_at as i128)); + if let Some(value) = record.wrapping_public_key_id { + response = response.add_typed_default(DataType::WrappingPublicKeyId, DataValue::Str(value)); + } + self.send_message(&response).await; + } + Ok(None) => { + self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound)).await; + } + Err(_) => { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + } + } + return; + } + + if cv.is_type(CommunicationType::EncryptedMessage) { + let sender_id = cv.get_sender().to_string(); + if data_string(&cv, DataType::SenderUserId).as_deref() != Some(sender_id.as_str()) { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + return; + } + let record = data_string(&cv, DataType::MessageId) + .zip(data_string(&cv, DataType::ConversationId)) + .zip(data_string(&cv, DataType::SenderClientId)) + .zip(data_string(&cv, DataType::RecipientClientId)) + .zip(data_bytes(&cv, DataType::EncryptedPayload)) + .map(|((((message_id, conversation_id), sender_client_id), recipient_client_id), encrypted_payload)| { + StoredEncryptedMessage { + message_id, + conversation_id, + sender_client_id, + recipient_client_id, + sender_user_id: data_string(&cv, DataType::SenderUserId), + recipient_user_id: data_string(&cv, DataType::RecipientUserId), + created_at: data_i64(&cv, DataType::CreatedAt).unwrap_or_else(now_millis_i64), + encryption_version: data_i64(&cv, DataType::EncryptionVersion).unwrap_or(1), + encrypted_payload, + } + }); + + if let Some(record) = record { + let message_id = record.message_id.clone(); + let conversation_id = record.conversation_id.clone(); + let recipient_client_id = record.recipient_client_id.clone(); + match e2ee_storage::put_encrypted_message(record) { + Ok(()) => { + self.send_message(&CommunicationValue::new(CommunicationType::EncryptedMessageAck) + .with_id(cv.get_id()) + .with_receiver(cv.get_sender()) + .add_typed_default(DataType::MessageId, DataValue::Str(message_id)) + .add_typed_default(DataType::ConversationId, DataValue::Str(conversation_id)) + .add_typed_default(DataType::RecipientClientId, DataValue::Str(recipient_client_id)) + .add_typed_default(DataType::GetTime, DataValue::SignedNumber(now_millis_i64() as i128)) + ).await; + } + Err(_) => { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + } + } + } else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + } + return; + } + + if cv.is_type(CommunicationType::EncryptedMessagesGet) { + let requester_user_id = cv.get_sender().to_string(); + let limit = data_i64(&cv, DataType::Limit).map(|v| v as i64); + let since = data_i64(&cv, DataType::Since); + let conversation_id = data_string(&cv, DataType::ConversationId); + let peer_client_id = data_string(&cv, DataType::PeerClientId); + + match e2ee_storage::get_encrypted_messages(EncryptedMessageQuery { + sender_user_id: data_string(&cv, DataType::SenderUserId), + recipient_client_id: None, + recipient_user_id: Some(requester_user_id.clone()), + conversation_id, + limit, + offset: since.map(|v| v.max(0)), + }) { + Ok(records) => { + let messages = records + .into_iter() + .filter(|record| { + peer_client_id + .as_ref() + .map(|peer| &record.sender_client_id == peer || &record.recipient_client_id == peer) + .unwrap_or(true) + }) + .map(|record| { + typed_container(vec![ + (DataType::MessageId, DataValue::Str(record.message_id)), + (DataType::ConversationId, DataValue::Str(record.conversation_id)), + (DataType::SenderClientId, DataValue::Str(record.sender_client_id)), + (DataType::RecipientClientId, DataValue::Str(record.recipient_client_id)), + (DataType::SenderUserId, DataValue::Str(record.sender_user_id.unwrap_or_default())), + (DataType::RecipientUserId, DataValue::Str(record.recipient_user_id.unwrap_or_default())), + (DataType::CreatedAt, DataValue::SignedNumber(record.created_at as i128)), + (DataType::EncryptionVersion, DataValue::SignedNumber(record.encryption_version as i128)), + (DataType::EncryptedPayload, DataValue::Bytes(record.encrypted_payload)), + ]) + }) + .collect::>(); + + self.send_message(&CommunicationValue::new(CommunicationType::EncryptedMessagesResponse) + .with_id(cv.get_id()) + .with_receiver(cv.get_sender()) + .add_typed_default(DataType::Messages, DataValue::Array(messages)) + .add_typed_default(DataType::HasMore, DataValue::Bool(false)) + ).await; + } + Err(_) => { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + } + } + return; + } + if cv.is_type(CommunicationType::SaveAppData) { let sender_id = cv.get_sender(); let _app_data = cv @@ -327,259 +551,8 @@ impl ClientConnection { // Incoming storsed message: store for the recipient, attempt local delivery, notify sender. if cv.is_type(CommunicationType::MessageSend) { - let sender_id: u64 = cv.get_sender(); - - // parse receiver_id (the storage owner for this incoming message) - let receiver_id: i64 = if let Some(n) = cv.get_data(DataType::ReceiverId).as_number() { - n as i64 - } else if let Some(s) = cv.get_data(DataType::ReceiverId).as_str() { - s.parse::().unwrap_or(0) - } else { - 0 - }; - - // parse send_time robustly (number or string), fallback to now - let send_time_val = cv.get_data(DataType::SendTime); - let now_i64 = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64; - let timestamp_i64 = if let Some(n) = send_time_val.as_number() { - n as i64 - } else if let Some(s) = send_time_val.as_str() { - s.parse::().unwrap_or(now_i64) - } else { - now_i64 - }; - let timestamp_u128 = timestamp_i64 as u128; - - // content may be missing; default to empty string - let content = cv - .get_data(DataType::Content) - .as_str() - .unwrap_or("") - .to_string(); - - let height = cv.get_data(DataType::Height).as_number().unwrap_or(0) as i64; - - let is_local = iota_storage::users::user_manager::get_user(receiver_id).is_some(); - - if is_local { - // persist message for the receiver (storage_owner = receiver_id) - chat_files::add_message( - timestamp_u128, - false, - receiver_id as i64, - sender_id as i64, - &content, - height, - ); - } - - // persist message for the sender (storage_owner = sender_id) - chat_files::add_message( - timestamp_u128, - true, - sender_id as i64, - receiver_id as i64, - &content, - height, - ); - - // send confirmation back to sender - let conf_msg = CommunicationValue::new(CommunicationType::MessageSend) - .with_id(cv.get_id()) - .with_receiver(sender_id as u64); - self.send_message(&conf_msg).await; - - if !is_local { - let fw_msg = CommunicationValue::new(CommunicationType::MessageOtherIota) - .with_id(cv.get_id()) - .with_receiver(receiver_id as u64) - .with_sender(sender_id as u64) - .add_typed_default(DataType::Height, DataValue::SignedNumber(height as i128)) - .add_typed_default(DataType::Content, DataValue::Str(content)) - .add_typed_default( - DataType::SendTime, - DataValue::SignedNumber(timestamp_i64 as i128), - ); - - let other_iota_resp = self - .clone() - .await_response(&fw_msg, Some(Duration::from_secs(10))) - .await; - - if let Ok(resp) = other_iota_resp { - let ms_raw = resp - .get_data(DataType::MessageState) - .as_string() - .unwrap_or_else(|| "".to_string()); - let ms = MessageState::from_str(&ms_raw).upgrade(MessageState::Received); - - let _ = chat_files::change_message_state( - timestamp_i64, - sender_id as i64, - receiver_id as i64, - ms.clone(), - ); - - self.send_message( - &CommunicationValue::new(CommunicationType::MessageState) - .with_id(cv.get_id()) - .with_receiver(sender_id as u64) - .with_sender(receiver_id as u64) - .add_typed_default( - DataType::ChatPartnerId, - DataValue::SignedNumber(receiver_id as i128), - ) - .add_typed_default( - DataType::SendTime, - DataValue::SignedNumber(timestamp_i64 as i128), - ) - .add_typed_default( - DataType::MessageState, - DataValue::Str(ms.as_str().to_string()), - ), - ) - .await; - } else { - let _ = chat_files::change_message_state( - timestamp_i64, - sender_id as i64, - receiver_id as i64, - MessageState::Sent, - ); - - self.send_message( - &CommunicationValue::new(CommunicationType::MessageState) - .with_id(cv.get_id()) - .with_receiver(sender_id as u64) - .with_sender(receiver_id as u64) - .add_typed_default( - DataType::ChatPartnerId, - DataValue::SignedNumber(receiver_id as i128), - ) - .add_typed_default( - DataType::SendTime, - DataValue::SignedNumber(timestamp_i64 as i128), - ) - .add_typed_default( - DataType::MessageState, - DataValue::Str(MessageState::Sent.as_str().to_string()), - ), - ) - .await; - } - return; - } else { - // Build a live-delivery message for the local client (recipient) - let user_forward = CommunicationValue::new(CommunicationType::MessageLive) - .with_id(cv.get_id()) - .with_receiver(receiver_id as u64) - .add_typed_default( - DataType::SenderId, - DataValue::SignedNumber(sender_id as i128), - ) - .add_typed_default( - DataType::Message, - typed_container(vec![ - (DataType::Content, DataValue::Str(content.clone())), - ( - DataType::SendTime, - DataValue::SignedNumber(timestamp_i64 as i128), - ), - (DataType::Height, DataValue::SignedNumber(height as i128)), - ]), - ); - - // Attempt delivery and await a response from the local client - let user_resp = self - .clone() - .await_response(&user_forward, Some(Duration::from_secs(10))) - .await; - - if let Ok(user_resp) = user_resp { - let ms_raw = user_resp - .get_data(DataType::MessageState) - .as_string() - .unwrap_or_else(|| "".to_string()); - let ms = MessageState::from_str(&ms_raw).upgrade(MessageState::Received); - - // update stored message state for receiver - let _ = chat_files::change_message_state( - timestamp_i64, - receiver_id as i64, - sender_id as i64, - ms.clone(), - ); - - // update stored message state for sender - let _ = chat_files::change_message_state( - timestamp_i64, - sender_id as i64, - receiver_id as i64, - ms.clone(), - ); - - // notify original sender about the delivered/read state - self.send_message( - &CommunicationValue::new(CommunicationType::MessageState) - .with_id(cv.get_id()) - .with_receiver(sender_id as u64) - .with_sender(receiver_id as u64) - .add_typed_default( - DataType::ChatPartnerId, - DataValue::SignedNumber(receiver_id as i128), - ) - .add_typed_default( - DataType::SendTime, - DataValue::SignedNumber(timestamp_i64 as i128), - ) - .add_typed_default( - DataType::MessageState, - DataValue::Str(ms.as_str().to_string()), - ), - ) - .await; - } else { - // Delivery failed or timed out; mark as Sent - let _ = chat_files::change_message_state( - timestamp_i64, - receiver_id as i64, - sender_id as i64, - MessageState::Sent, - ); - - let _ = chat_files::change_message_state( - timestamp_i64, - sender_id as i64, - receiver_id as i64, - MessageState::Sent, - ); - - // notify sender - self.send_message( - &CommunicationValue::new(CommunicationType::MessageState) - .with_id(cv.get_id()) - .with_receiver(sender_id as u64) - .with_sender(receiver_id as u64) - .add_typed_default( - DataType::SendTime, - DataValue::SignedNumber(timestamp_i64 as i128), - ) - .add_typed_default( - DataType::ChatPartnerId, - DataValue::SignedNumber(receiver_id as i128), - ) - .add_typed_default( - DataType::MessageState, - DataValue::Str(MessageState::Sent.as_str().to_string()), - ), - ) - .await; - } - return; - } + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + return; } if cv.is_type(CommunicationType::MessageOtherIota) { diff --git a/iota-logger/src/lib.rs b/iota-logger/src/lib.rs index bfde4ce..01ebbff 100644 --- a/iota-logger/src/lib.rs +++ b/iota-logger/src/lib.rs @@ -1,5 +1,4 @@ use std::{ - collections::BTreeMap, fs::{self, OpenOptions}, io::Write, path::Path, @@ -8,7 +7,7 @@ use std::{ time::{SystemTime, UNIX_EPOCH}, }; -use mtp::codec::{CommunicationValue, DataType, DataTypeId, DataValue, Version}; +use mtp::codec::{CommunicationValue, DataTypeId, DataValue, Version}; use ratatui::style::Color; use iota_state::{APP_STATE, UNIQUE, UiLogEntry}; diff --git a/iota-storage/src/util/e2ee_storage.rs b/iota-storage/src/util/e2ee_storage.rs new file mode 100644 index 0000000..4872e4d --- /dev/null +++ b/iota-storage/src/util/e2ee_storage.rs @@ -0,0 +1,411 @@ +use crate::util::db; +use rusqlite::{params, OptionalExtension}; +use std::sync::{Arc, LazyLock, Mutex}; + +pub type StorageError = String; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StoredEncryptedMessage { + pub message_id: String, + pub conversation_id: String, + pub sender_client_id: String, + pub recipient_client_id: String, + pub sender_user_id: Option, + pub recipient_user_id: Option, + pub created_at: i64, + pub encryption_version: i64, + pub encrypted_payload: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StoredEncryptedDeviceSecret { + pub user_id: String, + pub device_id: String, + pub secret_id: String, + pub version: i64, + pub encrypted_secret: Vec, + pub wrapping_public_key_id: Option, + pub wrapping_scheme: String, + pub created_at: i64, + pub updated_at: i64, +} + +#[derive(Debug, Clone, Default)] +pub struct EncryptedMessageQuery { + pub sender_user_id: Option, + pub recipient_client_id: Option, + pub recipient_user_id: Option, + pub conversation_id: Option, + pub limit: Option, + pub offset: Option, +} + +#[derive(Debug, Clone, Default)] +pub struct EncryptedDeviceSecretQuery { + pub user_id: String, + pub device_id: Option, + pub secret_id: Option, +} + +static E2EE_DB: LazyLock>> = LazyLock::new(|| { + db::create_shared_connection( + "e2ee", + r#" + PRAGMA journal_mode = WAL; + PRAGMA synchronous = NORMAL; + + CREATE TABLE IF NOT EXISTS encrypted_messages ( + message_id TEXT NOT NULL PRIMARY KEY, + conversation_id TEXT NOT NULL, + sender_client_id TEXT NOT NULL, + recipient_client_id TEXT NOT NULL, + sender_user_id TEXT, + recipient_user_id TEXT, + created_at INTEGER NOT NULL, + encryption_version INTEGER NOT NULL, + encrypted_payload BLOB NOT NULL, + acked_at INTEGER + ); + + CREATE INDEX IF NOT EXISTS idx_encrypted_messages_recipient + ON encrypted_messages (recipient_client_id, created_at ASC); + CREATE INDEX IF NOT EXISTS idx_encrypted_messages_recipient_user + ON encrypted_messages (recipient_user_id, created_at ASC); + CREATE INDEX IF NOT EXISTS idx_encrypted_messages_conversation + ON encrypted_messages (conversation_id, created_at ASC); + + CREATE TABLE IF NOT EXISTS encrypted_device_secrets ( + user_id TEXT NOT NULL, + device_id TEXT NOT NULL, + secret_id TEXT NOT NULL, + version INTEGER NOT NULL, + encrypted_secret BLOB NOT NULL, + wrapping_public_key_id TEXT, + wrapping_scheme TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + PRIMARY KEY (user_id, device_id, secret_id) + ); + + CREATE INDEX IF NOT EXISTS idx_encrypted_device_secrets_owner + ON encrypted_device_secrets (user_id, device_id, secret_id); + "#, + ) + .expect("Failed to create or initialize E2EE DB") +}); + +pub fn put_encrypted_message(record: StoredEncryptedMessage) -> Result<(), StorageError> { + db::with_conn(&E2EE_DB, |conn| { + conn.execute( + r#" + INSERT OR REPLACE INTO encrypted_messages ( + message_id, conversation_id, sender_client_id, recipient_client_id, + sender_user_id, recipient_user_id, created_at, encryption_version, + encrypted_payload + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) + "#, + params![ + record.message_id, + record.conversation_id, + record.sender_client_id, + record.recipient_client_id, + record.sender_user_id, + record.recipient_user_id, + record.created_at, + record.encryption_version, + record.encrypted_payload, + ], + )?; + Ok(()) + }) +} + +pub fn get_encrypted_messages( + query: EncryptedMessageQuery, +) -> Result, StorageError> { + let limit = query.limit.unwrap_or(100).clamp(1, 500); + let offset = query.offset.unwrap_or(0).max(0); + + db::with_conn(&E2EE_DB, |conn| { + let mut stmt = conn.prepare( + r#" + SELECT message_id, conversation_id, sender_client_id, recipient_client_id, + sender_user_id, recipient_user_id, created_at, encryption_version, + encrypted_payload + FROM encrypted_messages + WHERE (?2 IS NULL OR recipient_client_id = ?2) + AND ( + (?1 IS NULL AND (?3 IS NULL OR sender_user_id = ?3 OR recipient_user_id = ?3)) + OR (?1 IS NOT NULL AND ?3 IS NOT NULL AND ( + (sender_user_id = ?1 AND recipient_user_id = ?3) + OR (sender_user_id = ?3 AND recipient_user_id = ?1) + )) + OR (?1 IS NOT NULL AND ?3 IS NULL AND (sender_user_id = ?1 OR recipient_user_id = ?1)) + ) + AND (?4 IS NULL OR conversation_id = ?4) + ORDER BY created_at ASC + LIMIT ?5 OFFSET ?6 + "#, + )?; + + let rows = stmt.query_map( + params![ + query.sender_user_id, + query.recipient_client_id, + query.recipient_user_id, + query.conversation_id, + limit, + offset, + ], + encrypted_message_from_row, + )?; + + let mut out = Vec::new(); + for row in rows { + out.push(row?); + } + Ok(out) + }) +} + +pub fn ack_encrypted_message( + message_id: &str, + recipient_client_id: &str, +) -> Result<(), StorageError> { + let now = now_millis(); + db::with_conn(&E2EE_DB, |conn| { + conn.execute( + r#" + UPDATE encrypted_messages + SET acked_at = ?1 + WHERE message_id = ?2 AND recipient_client_id = ?3 + "#, + params![now, message_id, recipient_client_id], + )?; + Ok(()) + }) +} + +pub fn put_encrypted_device_secret( + record: StoredEncryptedDeviceSecret, +) -> Result<(), StorageError> { + db::with_conn(&E2EE_DB, |conn| { + conn.execute( + r#" + INSERT INTO encrypted_device_secrets ( + user_id, device_id, secret_id, version, encrypted_secret, + wrapping_public_key_id, wrapping_scheme, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) + ON CONFLICT(user_id, device_id, secret_id) DO UPDATE SET + version = excluded.version, + encrypted_secret = excluded.encrypted_secret, + wrapping_public_key_id = excluded.wrapping_public_key_id, + wrapping_scheme = excluded.wrapping_scheme, + created_at = excluded.created_at, + updated_at = excluded.updated_at + "#, + params![ + record.user_id, + record.device_id, + record.secret_id, + record.version, + record.encrypted_secret, + record.wrapping_public_key_id, + record.wrapping_scheme, + record.created_at, + record.updated_at, + ], + )?; + Ok(()) + }) +} + +pub fn get_encrypted_device_secret( + query: EncryptedDeviceSecretQuery, +) -> Result, StorageError> { + if query.user_id.is_empty() { + return Ok(None); + } + + db::with_conn(&E2EE_DB, |conn| { + conn.query_row( + r#" + SELECT user_id, device_id, secret_id, version, encrypted_secret, + wrapping_public_key_id, wrapping_scheme, created_at, updated_at + FROM encrypted_device_secrets + WHERE user_id = ?1 + AND (?2 IS NULL OR device_id = ?2) + AND (?3 IS NULL OR secret_id = ?3) + ORDER BY updated_at DESC + LIMIT 1 + "#, + params![query.user_id, query.device_id, query.secret_id], + encrypted_device_secret_from_row, + ) + .optional() + }) +} + +fn encrypted_message_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(StoredEncryptedMessage { + message_id: row.get(0)?, + conversation_id: row.get(1)?, + sender_client_id: row.get(2)?, + recipient_client_id: row.get(3)?, + sender_user_id: row.get(4)?, + recipient_user_id: row.get(5)?, + created_at: row.get(6)?, + encryption_version: row.get(7)?, + encrypted_payload: row.get(8)?, + }) +} + +fn encrypted_device_secret_from_row( + row: &rusqlite::Row<'_>, +) -> rusqlite::Result { + Ok(StoredEncryptedDeviceSecret { + user_id: row.get(0)?, + device_id: row.get(1)?, + secret_id: row.get(2)?, + version: row.get(3)?, + encrypted_secret: row.get(4)?, + wrapping_public_key_id: row.get(5)?, + wrapping_scheme: row.get(6)?, + created_at: row.get(7)?, + updated_at: row.get(8)?, + }) +} + +fn now_millis() -> i64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 +} + +#[cfg(test)] +mod tests { + use super::*; + + fn suffix(name: &str) -> String { + format!("{name}-{}", now_millis()) + } + + #[test] + fn stores_encrypted_message_as_opaque_ciphertext() { + let message_id = suffix("msg"); + put_encrypted_message(StoredEncryptedMessage { + message_id: message_id.clone(), + conversation_id: suffix("conversation"), + sender_client_id: "sender".to_string(), + recipient_client_id: "recipient".to_string(), + sender_user_id: Some("sender-user".to_string()), + recipient_user_id: Some(suffix("recipient-user")), + created_at: now_millis(), + encryption_version: 1, + encrypted_payload: vec![9, 8, 7], + }) + .unwrap(); + + let messages = get_encrypted_messages(EncryptedMessageQuery { + recipient_client_id: Some("recipient".to_string()), + limit: Some(10), + ..Default::default() + }) + .unwrap(); + assert!(messages.iter().any(|m| m.message_id == message_id && m.encrypted_payload == vec![9, 8, 7])); + } + + #[test] + fn encrypted_history_query_returns_both_directions_for_peer() { + let user = suffix("user"); + let peer = suffix("peer"); + let incoming_id = suffix("incoming"); + let outgoing_id = suffix("outgoing"); + let other_id = suffix("other"); + + put_encrypted_message(StoredEncryptedMessage { + message_id: incoming_id.clone(), + conversation_id: suffix("conversation"), + sender_client_id: "peer-client".to_string(), + recipient_client_id: "user-client".to_string(), + sender_user_id: Some(peer.clone()), + recipient_user_id: Some(user.clone()), + created_at: now_millis(), + encryption_version: 1, + encrypted_payload: vec![1], + }) + .unwrap(); + put_encrypted_message(StoredEncryptedMessage { + message_id: outgoing_id.clone(), + conversation_id: suffix("conversation"), + sender_client_id: "user-client".to_string(), + recipient_client_id: "peer-client".to_string(), + sender_user_id: Some(user.clone()), + recipient_user_id: Some(peer.clone()), + created_at: now_millis() + 1, + encryption_version: 1, + encrypted_payload: vec![2], + }) + .unwrap(); + put_encrypted_message(StoredEncryptedMessage { + message_id: other_id.clone(), + conversation_id: suffix("conversation"), + sender_client_id: "other-client".to_string(), + recipient_client_id: "user-client".to_string(), + sender_user_id: Some(suffix("other")), + recipient_user_id: Some(user.clone()), + created_at: now_millis() + 2, + encryption_version: 1, + encrypted_payload: vec![3], + }) + .unwrap(); + + let messages = get_encrypted_messages(EncryptedMessageQuery { + sender_user_id: Some(peer), + recipient_user_id: Some(user), + limit: Some(10), + ..Default::default() + }) + .unwrap(); + + assert!(messages.iter().any(|m| m.message_id == incoming_id)); + assert!(messages.iter().any(|m| m.message_id == outgoing_id)); + assert!(!messages.iter().any(|m| m.message_id == other_id)); + } + + #[test] + fn stores_and_retrieves_encrypted_device_secret_blob_for_owner_only() { + let user_id = suffix("user"); + put_encrypted_device_secret(StoredEncryptedDeviceSecret { + user_id: user_id.clone(), + device_id: "device".to_string(), + secret_id: "main".to_string(), + version: 1, + encrypted_secret: vec![42, 43], + wrapping_public_key_id: None, + wrapping_scheme: "mtp-chacha20poly1305-hkdf-sha256-v1".to_string(), + created_at: 1, + updated_at: 2, + }) + .unwrap(); + + let found = get_encrypted_device_secret(EncryptedDeviceSecretQuery { + user_id: user_id.clone(), + device_id: Some("device".to_string()), + secret_id: Some("main".to_string()), + }) + .unwrap() + .unwrap(); + assert_eq!(found.encrypted_secret, vec![42, 43]); + + let denied = get_encrypted_device_secret(EncryptedDeviceSecretQuery { + user_id: suffix("other-user"), + device_id: Some("device".to_string()), + secret_id: Some("main".to_string()), + }) + .unwrap(); + assert!(denied.is_none()); + } +} diff --git a/iota-storage/src/util/mod.rs b/iota-storage/src/util/mod.rs index dff0d65..d5b2e4d 100644 --- a/iota-storage/src/util/mod.rs +++ b/iota-storage/src/util/mod.rs @@ -3,3 +3,4 @@ pub mod chats_util; pub mod communities_util; pub mod config_util; pub mod db; +pub mod e2ee_storage; diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 0d07db9..95dcd22 100755 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -6,6 +6,10 @@ use iota_storage::util::chat_files::{self, MessageState, change_message_state}; use iota_storage::util::chats_util::{self, get_user, mod_user}; use iota_storage::util::communities_util::CommunitiesUtil; use iota_storage::util::config_util::{modify_config, CONFIG}; +use iota_storage::util::e2ee_storage::{ + self, EncryptedDeviceSecretQuery, EncryptedMessageQuery, + StoredEncryptedDeviceSecret, StoredEncryptedMessage, +}; use iota_util::crypto_helper::{self, keyring_from_base64}; use iota_util::crypto_util::{self}; use iota_util::file_util::{get_children, has_file, load_file, save_file}; @@ -34,6 +38,39 @@ fn typed_container(items: Vec<(DataType, DataValue)>) -> DataValue { ) } +fn data_string(cv: &CommunicationValue, dt: DataType) -> Option { + cv.get_data(dt) + .as_str() + .map(|s| s.to_string()) + .or_else(|| cv.get_data(dt).as_number().map(|n| n.to_string())) + .or_else(|| cv.get_data(dt).as_signed_number().map(|n| n.to_string())) +} + +fn data_i64(cv: &CommunicationValue, dt: DataType) -> Option { + cv.get_data(dt) + .as_number() + .and_then(|n| i64::try_from(n).ok()) + .or_else(|| cv.get_data(dt).as_signed_number().and_then(|n| i64::try_from(n).ok())) + .or_else(|| cv.get_data(dt).as_str().and_then(|s| s.parse::().ok())) +} + +fn data_bytes(cv: &CommunicationValue, dt: DataType) -> Option> { + cv.get_bytes(dt).map(|bytes| bytes.to_vec()) +} + +fn now_millis_i64() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 +} + +fn error_response(request: &CommunicationValue, ty: CommunicationType) -> CommunicationValue { + CommunicationValue::new(ty) + .with_id(request.get_id()) + .with_receiver(request.get_sender()) +} + // Helper function to check if read receipts are enabled globally async fn is_read_receipts_enabled() -> bool { CONFIG.load().read_receipts_enabled @@ -553,6 +590,191 @@ impl OmikronConnection { return; } + if cv.is_type(CommunicationType::SetEncryptedDeviceSecret) { + let sender_id = cv.get_sender().to_string(); + if data_string(&cv, DataType::UserId).as_deref() != Some(sender_id.as_str()) { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + return; + } + let now = now_millis_i64(); + let record = data_string(&cv, DataType::UserId) + .zip(data_string(&cv, DataType::DeviceId)) + .zip(data_string(&cv, DataType::SecretId)) + .zip(data_i64(&cv, DataType::VersionNumber)) + .zip(data_bytes(&cv, DataType::EncryptedSecret)) + .zip(data_string(&cv, DataType::WrappingScheme)) + .map(|(((((user_id, device_id), secret_id), version), encrypted_secret), wrapping_scheme)| { + StoredEncryptedDeviceSecret { + user_id, + device_id, + secret_id, + version, + encrypted_secret, + wrapping_public_key_id: data_string(&cv, DataType::WrappingPublicKeyId), + wrapping_scheme, + created_at: data_i64(&cv, DataType::CreatedAt).unwrap_or(now), + updated_at: now, + } + }); + + match record.map(e2ee_storage::put_encrypted_device_secret) { + Some(Ok(())) => self.send_message(&error_response(&cv, CommunicationType::Success)).await, + _ => self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await, + } + return; + } + + if cv.is_type(CommunicationType::GetEncryptedDeviceSecret) { + let Some(user_id) = data_string(&cv, DataType::UserId) else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + return; + }; + if user_id != cv.get_sender().to_string() { + self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound)).await; + return; + } + + match e2ee_storage::get_encrypted_device_secret(EncryptedDeviceSecretQuery { + user_id, + device_id: data_string(&cv, DataType::DeviceId), + secret_id: data_string(&cv, DataType::SecretId), + }) { + Ok(Some(record)) => { + let mut response = CommunicationValue::new(CommunicationType::EncryptedDeviceSecretResponse) + .with_id(cv.get_id()) + .with_receiver(cv.get_sender()) + .add_typed_default(DataType::UserId, DataValue::Str(record.user_id)) + .add_typed_default(DataType::DeviceId, DataValue::Str(record.device_id)) + .add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id)) + .add_typed_default(DataType::VersionNumber, DataValue::SignedNumber(record.version as i128)) + .add_typed_default(DataType::EncryptedSecret, DataValue::Bytes(record.encrypted_secret)) + .add_typed_default(DataType::WrappingScheme, DataValue::Str(record.wrapping_scheme)) + .add_typed_default(DataType::CreatedAt, DataValue::SignedNumber(record.created_at as i128)) + .add_typed_default(DataType::UpdatedAt, DataValue::SignedNumber(record.updated_at as i128)); + if let Some(value) = record.wrapping_public_key_id { + response = response.add_typed_default(DataType::WrappingPublicKeyId, DataValue::Str(value)); + } + self.send_message(&response).await; + } + Ok(None) => self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound)).await, + Err(_) => self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await, + } + return; + } + + if cv.is_type(CommunicationType::EncryptedMessage) { + let sender_id = cv.get_sender().to_string(); + if data_string(&cv, DataType::SenderUserId).as_deref() != Some(sender_id.as_str()) { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + return; + } + let record = data_string(&cv, DataType::MessageId) + .zip(data_string(&cv, DataType::ConversationId)) + .zip(data_string(&cv, DataType::SenderClientId)) + .zip(data_string(&cv, DataType::RecipientClientId)) + .zip(data_bytes(&cv, DataType::EncryptedPayload)) + .map(|((((message_id, conversation_id), sender_client_id), recipient_client_id), encrypted_payload)| { + StoredEncryptedMessage { + message_id, + conversation_id, + sender_client_id, + recipient_client_id, + sender_user_id: data_string(&cv, DataType::SenderUserId), + recipient_user_id: data_string(&cv, DataType::RecipientUserId), + created_at: data_i64(&cv, DataType::CreatedAt).unwrap_or_else(now_millis_i64), + encryption_version: data_i64(&cv, DataType::EncryptionVersion).unwrap_or(1), + encrypted_payload, + } + }); + + if let Some(record) = record { + let message_id = record.message_id.clone(); + let conversation_id = record.conversation_id.clone(); + let recipient_client_id = record.recipient_client_id.clone(); + let sender_client_id = record.sender_client_id.clone(); + let sender_user_id = record.sender_user_id.clone().unwrap_or_default(); + let recipient_user_id = record.recipient_user_id.clone().unwrap_or_default(); + match e2ee_storage::put_encrypted_message(record) { + Ok(()) => { + self.send_message(&CommunicationValue::new(CommunicationType::EncryptedMessageAck) + .with_id(cv.get_id()) + .with_receiver(cv.get_sender()) + .add_typed_default(DataType::MessageId, DataValue::Str(message_id)) + .add_typed_default(DataType::ConversationId, DataValue::Str(conversation_id)) + .add_typed_default(DataType::RecipientClientId, DataValue::Str(recipient_client_id.clone())) + .add_typed_default(DataType::GetTime, DataValue::SignedNumber(now_millis_i64() as i128)) + ).await; + + if !recipient_user_id.is_empty() + && recipient_user_id != sender_user_id + && recipient_client_id != sender_client_id + { + if cv.get_receiver().to_string() != recipient_user_id { + self.send_message(&cv.clone().with_receiver(recipient_user_id.parse::().unwrap_or(0))).await; + } else { + self.send_message(&cv).await; + } + } + } + Err(_) => self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await, + } + } else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + } + return; + } + + if cv.is_type(CommunicationType::EncryptedMessagesGet) { + let requester_user_id = cv.get_sender().to_string(); + let limit = data_i64(&cv, DataType::Limit).map(|v| v as i64); + let since = data_i64(&cv, DataType::Since); + let conversation_id = data_string(&cv, DataType::ConversationId); + let peer_client_id = data_string(&cv, DataType::PeerClientId); + + match e2ee_storage::get_encrypted_messages(EncryptedMessageQuery { + sender_user_id: data_string(&cv, DataType::SenderUserId), + recipient_client_id: None, + recipient_user_id: Some(requester_user_id.clone()), + conversation_id, + limit, + offset: since.map(|v| v.max(0)), + }) { + Ok(records) => { + let messages = records + .into_iter() + .filter(|record| { + peer_client_id + .as_ref() + .map(|peer| &record.sender_client_id == peer || &record.recipient_client_id == peer) + .unwrap_or(true) + }) + .map(|record| { + typed_container(vec![ + (DataType::MessageId, DataValue::Str(record.message_id)), + (DataType::ConversationId, DataValue::Str(record.conversation_id)), + (DataType::SenderClientId, DataValue::Str(record.sender_client_id)), + (DataType::RecipientClientId, DataValue::Str(record.recipient_client_id)), + (DataType::SenderUserId, DataValue::Str(record.sender_user_id.unwrap_or_default())), + (DataType::RecipientUserId, DataValue::Str(record.recipient_user_id.unwrap_or_default())), + (DataType::CreatedAt, DataValue::SignedNumber(record.created_at as i128)), + (DataType::EncryptionVersion, DataValue::SignedNumber(record.encryption_version as i128)), + (DataType::EncryptedPayload, DataValue::Bytes(record.encrypted_payload)), + ]) + }) + .collect::>(); + + self.send_message(&CommunicationValue::new(CommunicationType::EncryptedMessagesResponse) + .with_id(cv.get_id()) + .with_receiver(cv.get_sender()) + .add_typed_default(DataType::Messages, DataValue::Array(messages)) + .add_typed_default(DataType::HasMore, DataValue::Bool(false)) + ).await; + } + Err(_) => self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await, + } + return; + } + if cv.is_type(CommunicationType::AppIdentification) { let sender_id = cv.get_sender(); let app_identifier = cv diff --git a/type-maps.yaml b/type-maps.yaml index 7fecac8..354f186 100644 --- a/type-maps.yaml +++ b/type-maps.yaml @@ -99,6 +99,13 @@ type_maps: AppChallengeResponse: 133 AppIdentificationResponse: 134 LoadTxtRecord: 135 + SetEncryptedDeviceSecret: 139 + GetEncryptedDeviceSecret: 140 + EncryptedDeviceSecretResponse: 141 + EncryptedMessage: 142 + EncryptedMessageAck: 143 + EncryptedMessagesGet: 144 + EncryptedMessagesResponse: 145 DataTypes: ErrorType: 32 ErrorProtocol: 33 @@ -199,3 +206,25 @@ type_maps: AppData: 131 TauriToken: 132 Challenge: 133 + EncryptedPayload: 134 + SecurePayload: 135 + DeviceId: 136 + ClientId: 137 + SecretId: 142 + VersionNumber: 143 + EncryptedSecret: 144 + WrappingPublicKeyId: 145 + WrappingScheme: 146 + UpdatedAt: 147 + MessageId: 148 + ConversationId: 149 + SenderClientId: 150 + RecipientClientId: 151 + SenderUserId: 152 + RecipientUserId: 153 + EncryptionVersion: 154 + HasMore: 155 + NextCursor: 156 + Since: 157 + Limit: 158 + PeerClientId: 159