diff --git a/client/Cargo.toml b/client/Cargo.toml index 72cb4d6..c29f662 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2024" [dependencies] -mtp = { git = "https://git.methanium.net/Methanium/mtp.git" } +mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = ["client"] } iota-logger = { path = "../iota-logger" } iota-util = { path = "../iota-util" } iota-storage = { path = "../iota-storage" } diff --git a/client/src/client_connection.rs b/client/src/client_connection.rs index 390a969..d1c7cf2 100644 --- a/client/src/client_connection.rs +++ b/client/src/client_connection.rs @@ -7,10 +7,7 @@ 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_storage::util::e2ee_storage::{self, ChatSecretQuery, StoredChatSecret}; 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}; @@ -34,8 +31,6 @@ fn typed_container(items: Vec<(DataType, DataValue)>) -> DataValue { ) } - - fn data_string(cv: &CommunicationValue, dt: DataType) -> Option { cv.get_data(dt) .as_str() @@ -182,7 +177,7 @@ impl ClientConnection { return; } - if cv.is_type(CommunicationType::SetEncryptedDeviceSecret) { + if cv.is_type(CommunicationType::SetChatSecret) { 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; @@ -190,26 +185,27 @@ impl ClientConnection { } let now = now_millis_i64(); let record = data_string(&cv, DataType::UserId) - .zip(data_string(&cv, DataType::DeviceId)) + .zip(data_string(&cv, DataType::ChatId)) .zip(data_string(&cv, DataType::SecretId)) .zip(data_i64(&cv, DataType::VersionNumber)) .zip(data_bytes(&cv, DataType::EncryptedSecret)) + .zip(data_bytes(&cv, DataType::KemCiphertext)) .zip(data_string(&cv, DataType::WrappingScheme)) - .map(|(((((user_id, device_id), secret_id), version), encrypted_secret), wrapping_scheme)| { - StoredEncryptedDeviceSecret { + .map(|((((((user_id, chat_id), secret_id), version), encrypted_secret), kem_ciphertext), wrapping_scheme)| { + StoredChatSecret { user_id, - device_id, + chat_id, secret_id, version, encrypted_secret, - wrapping_public_key_id: data_string(&cv, DataType::WrappingPublicKeyId), + kem_ciphertext, wrapping_scheme, created_at: data_i64(&cv, DataType::CreatedAt).unwrap_or(now), updated_at: now, } }); - match record.map(e2ee_storage::put_encrypted_device_secret) { + match record.map(e2ee_storage::put_chat_secret) { Some(Ok(())) => { self.send_message(&error_response(&cv, CommunicationType::Success)).await; } @@ -220,7 +216,7 @@ impl ClientConnection { return; } - if cv.is_type(CommunicationType::GetEncryptedDeviceSecret) { + if cv.is_type(CommunicationType::GetChatSecret) { let Some(user_id) = data_string(&cv, DataType::UserId) else { self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; return; @@ -230,27 +226,29 @@ impl ClientConnection { self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound)).await; return; } + let Some(chat_id) = data_string(&cv, DataType::ChatId) else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + return; + }; - match e2ee_storage::get_encrypted_device_secret(EncryptedDeviceSecretQuery { + match e2ee_storage::get_chat_secret(ChatSecretQuery { user_id, - device_id: data_string(&cv, DataType::DeviceId), + chat_id, secret_id: data_string(&cv, DataType::SecretId), }) { Ok(Some(record)) => { - let mut response = CommunicationValue::new(CommunicationType::EncryptedDeviceSecretResponse) + let response = CommunicationValue::new(CommunicationType::ChatSecretResponse) .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::ChatId, DataValue::Str(record.chat_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::KemCiphertext, DataValue::Bytes(record.kem_ciphertext)) .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) => { @@ -263,106 +261,16 @@ impl ClientConnection { return; } - if cv.is_type(CommunicationType::EncryptedMessage) { + if cv.is_type(CommunicationType::ChatSecretForward) { let sender_id = cv.get_sender().to_string(); - if data_string(&cv, DataType::SenderUserId).as_deref() != Some(sender_id.as_str()) { + let recipient_id = data_string(&cv, DataType::RecipientUserId).unwrap_or_default(); + if data_string(&cv, DataType::SenderUserId).as_deref() != Some(sender_id.as_str()) + || recipient_id.is_empty() + { 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; - } - } + self.send_message(&cv.with_receiver(recipient_id.parse::().unwrap_or(0))).await; return; } diff --git a/iota-storage/src/util/e2ee_storage.rs b/iota-storage/src/util/e2ee_storage.rs index 4872e4d..d26a092 100644 --- a/iota-storage/src/util/e2ee_storage.rs +++ b/iota-storage/src/util/e2ee_storage.rs @@ -5,46 +5,36 @@ 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 struct StoredChatSecret { pub user_id: String, - pub device_id: String, + pub chat_id: String, pub secret_id: String, pub version: i64, pub encrypted_secret: Vec, - pub wrapping_public_key_id: Option, + pub kem_ciphertext: Vec, 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, +pub struct ChatSecretQuery { + pub user_id: String, + pub chat_id: String, + pub secret_id: Option, } -#[derive(Debug, Clone, Default)] -pub struct EncryptedDeviceSecretQuery { - pub user_id: String, - pub device_id: Option, - pub secret_id: Option, +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PendingChatSecretForward { + pub recipient_user_id: String, + pub chat_id: String, + pub sender_user_id: String, + pub secret_id: String, + pub version: i64, + pub encrypted_secret: Vec, + pub kem_ciphertext: Vec, + pub wrapping_scheme: String, + pub created_at: i64, } static E2EE_DB: LazyLock>> = LazyLock::new(|| { @@ -54,163 +44,68 @@ static E2EE_DB: LazyLock>> = LazyLock::new(|| { 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 - ); + DROP TABLE IF EXISTS encrypted_messages; + DROP TABLE IF EXISTS encrypted_device_secrets; - 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 ( + CREATE TABLE IF NOT EXISTS chat_secrets ( user_id TEXT NOT NULL, - device_id TEXT NOT NULL, + chat_id TEXT NOT NULL, secret_id TEXT NOT NULL, version INTEGER NOT NULL, encrypted_secret BLOB NOT NULL, - wrapping_public_key_id TEXT, + kem_ciphertext BLOB NOT NULL, wrapping_scheme TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, - PRIMARY KEY (user_id, device_id, secret_id) + PRIMARY KEY (user_id, chat_id, secret_id) ); - CREATE INDEX IF NOT EXISTS idx_encrypted_device_secrets_owner - ON encrypted_device_secrets (user_id, device_id, secret_id); + CREATE INDEX IF NOT EXISTS idx_chat_secrets_owner + ON chat_secrets (user_id, chat_id, secret_id); + + CREATE TABLE IF NOT EXISTS pending_chat_secret_forwards ( + recipient_user_id TEXT NOT NULL, + chat_id TEXT NOT NULL, + sender_user_id TEXT NOT NULL, + secret_id TEXT NOT NULL, + version INTEGER NOT NULL, + encrypted_secret BLOB NOT NULL, + kem_ciphertext BLOB NOT NULL, + wrapping_scheme TEXT NOT NULL, + created_at INTEGER NOT NULL, + PRIMARY KEY (recipient_user_id, chat_id, secret_id) + ); + + CREATE INDEX IF NOT EXISTS idx_pending_chat_secret_forwards_recipient + ON pending_chat_secret_forwards (recipient_user_id, created_at); "#, ) .expect("Failed to create or initialize E2EE DB") }); -pub fn put_encrypted_message(record: StoredEncryptedMessage) -> Result<(), StorageError> { +pub fn put_chat_secret(record: StoredChatSecret) -> 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 + INSERT INTO chat_secrets ( + user_id, chat_id, secret_id, version, encrypted_secret, + kem_ciphertext, wrapping_scheme, created_at, updated_at ) 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 + ON CONFLICT(user_id, chat_id, secret_id) DO UPDATE SET version = excluded.version, encrypted_secret = excluded.encrypted_secret, - wrapping_public_key_id = excluded.wrapping_public_key_id, + kem_ciphertext = excluded.kem_ciphertext, wrapping_scheme = excluded.wrapping_scheme, created_at = excluded.created_at, updated_at = excluded.updated_at "#, params![ record.user_id, - record.device_id, + record.chat_id, record.secret_id, record.version, record.encrypted_secret, - record.wrapping_public_key_id, + record.kem_ciphertext, record.wrapping_scheme, record.created_at, record.updated_at, @@ -220,68 +115,131 @@ pub fn put_encrypted_device_secret( }) } -pub fn get_encrypted_device_secret( - query: EncryptedDeviceSecretQuery, -) -> Result, StorageError> { - if query.user_id.is_empty() { +pub fn put_pending_chat_secret_forward( + record: PendingChatSecretForward, +) -> Result<(), StorageError> { + db::with_conn(&E2EE_DB, |conn| { + conn.execute( + r#" + INSERT INTO pending_chat_secret_forwards ( + recipient_user_id, chat_id, sender_user_id, secret_id, version, + encrypted_secret, kem_ciphertext, wrapping_scheme, created_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) + ON CONFLICT(recipient_user_id, chat_id, secret_id) DO UPDATE SET + sender_user_id = excluded.sender_user_id, + version = excluded.version, + encrypted_secret = excluded.encrypted_secret, + kem_ciphertext = excluded.kem_ciphertext, + wrapping_scheme = excluded.wrapping_scheme, + created_at = excluded.created_at + "#, + params![ + record.recipient_user_id, + record.chat_id, + record.sender_user_id, + record.secret_id, + record.version, + record.encrypted_secret, + record.kem_ciphertext, + record.wrapping_scheme, + record.created_at, + ], + )?; + Ok(()) + }) +} + +pub fn get_pending_chat_secret_forwards( + limit: i64, +) -> Result, StorageError> { + db::with_conn(&E2EE_DB, |conn| { + let mut stmt = conn.prepare( + r#" + SELECT recipient_user_id, chat_id, sender_user_id, secret_id, version, + encrypted_secret, kem_ciphertext, wrapping_scheme, created_at + FROM pending_chat_secret_forwards + ORDER BY created_at ASC + LIMIT ?1 + "#, + )?; + let rows = stmt.query_map(params![limit.clamp(1, 500)], pending_forward_from_row)?; + let mut out = Vec::new(); + for row in rows { + out.push(row?); + } + Ok(out) + }) +} + +pub fn delete_pending_chat_secret_forward( + recipient_user_id: &str, + chat_id: &str, + secret_id: &str, +) -> Result<(), StorageError> { + db::with_conn(&E2EE_DB, |conn| { + conn.execute( + r#" + DELETE FROM pending_chat_secret_forwards + WHERE recipient_user_id = ?1 AND chat_id = ?2 AND secret_id = ?3 + "#, + params![recipient_user_id, chat_id, secret_id], + )?; + Ok(()) + }) +} + +pub fn get_chat_secret(query: ChatSecretQuery) -> Result, StorageError> { + if query.user_id.is_empty() || query.chat_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 + SELECT user_id, chat_id, secret_id, version, encrypted_secret, + kem_ciphertext, wrapping_scheme, created_at, updated_at + FROM chat_secrets WHERE user_id = ?1 - AND (?2 IS NULL OR device_id = ?2) + AND chat_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, + params![query.user_id, query.chat_id, query.secret_id], + chat_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 { +fn chat_secret_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(StoredChatSecret { user_id: row.get(0)?, - device_id: row.get(1)?, + chat_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)?, + kem_ciphertext: 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 +fn pending_forward_from_row( + row: &rusqlite::Row<'_>, +) -> rusqlite::Result { + Ok(PendingChatSecretForward { + recipient_user_id: row.get(0)?, + chat_id: row.get(1)?, + sender_user_id: row.get(2)?, + secret_id: row.get(3)?, + version: row.get(4)?, + encrypted_secret: row.get(5)?, + kem_ciphertext: row.get(6)?, + wrapping_scheme: row.get(7)?, + created_at: row.get(8)?, + }) } #[cfg(test)] @@ -289,120 +247,44 @@ mod tests { use super::*; fn suffix(name: &str) -> String { - format!("{name}-{}", now_millis()) + let unix = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + format!("{name}-{unix}") } #[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() { + fn stores_and_retrieves_chat_secret_blob_for_owner_chat() { let user_id = suffix("user"); - put_encrypted_device_secret(StoredEncryptedDeviceSecret { + let chat_id = suffix("chat"); + + put_chat_secret(StoredChatSecret { user_id: user_id.clone(), - device_id: "device".to_string(), + chat_id: chat_id.clone(), 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(), + kem_ciphertext: vec![9, 8, 7], + wrapping_scheme: "mtp-kem-chacha20poly1305-hkdf-sha256-v1".to_string(), created_at: 1, updated_at: 2, }) .unwrap(); - let found = get_encrypted_device_secret(EncryptedDeviceSecretQuery { + let found = get_chat_secret(ChatSecretQuery { user_id: user_id.clone(), - device_id: Some("device".to_string()), + chat_id: chat_id.clone(), secret_id: Some("main".to_string()), }) .unwrap() .unwrap(); assert_eq!(found.encrypted_secret, vec![42, 43]); + assert_eq!(found.kem_ciphertext, vec![9, 8, 7]); - let denied = get_encrypted_device_secret(EncryptedDeviceSecretQuery { + let denied = get_chat_secret(ChatSecretQuery { user_id: suffix("other-user"), - device_id: Some("device".to_string()), + chat_id, secret_id: Some("main".to_string()), }) .unwrap(); diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 95dcd22..61ce63a 100755 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -7,8 +7,7 @@ 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, + self, ChatSecretQuery, PendingChatSecretForward, StoredChatSecret, }; use iota_util::crypto_helper::{self, keyring_from_base64}; use iota_util::crypto_util::{self}; @@ -65,6 +64,57 @@ fn now_millis_i64() -> i64 { .as_millis() as i64 } +fn pending_chat_secret_forward_from_cv( + cv: &CommunicationValue, +) -> Option { + Some(PendingChatSecretForward { + recipient_user_id: data_string(cv, DataType::RecipientUserId)?, + chat_id: data_string(cv, DataType::ChatId)?, + sender_user_id: data_string(cv, DataType::SenderUserId)?, + secret_id: data_string(cv, DataType::SecretId)?, + version: data_i64(cv, DataType::VersionNumber)?, + encrypted_secret: data_bytes(cv, DataType::EncryptedSecret)?, + kem_ciphertext: data_bytes(cv, DataType::KemCiphertext)?, + wrapping_scheme: data_string(cv, DataType::WrappingScheme)?, + created_at: data_i64(cv, DataType::CreatedAt).unwrap_or_else(now_millis_i64), + }) +} + +fn chat_secret_forward_cv(record: &PendingChatSecretForward) -> CommunicationValue { + CommunicationValue::new(CommunicationType::ChatSecretForward) + .with_sender(record.sender_user_id.parse::().unwrap_or(0)) + .add_typed_default(DataType::ChatId, DataValue::Str(record.chat_id.clone())) + .add_typed_default( + DataType::SenderUserId, + DataValue::Str(record.sender_user_id.clone()), + ) + .add_typed_default( + DataType::RecipientUserId, + DataValue::Str(record.recipient_user_id.clone()), + ) + .add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id.clone())) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(record.version as i128), + ) + .add_typed_default( + DataType::EncryptedSecret, + DataValue::Bytes(record.encrypted_secret.clone()), + ) + .add_typed_default( + DataType::KemCiphertext, + DataValue::Bytes(record.kem_ciphertext.clone()), + ) + .add_typed_default( + DataType::WrappingScheme, + DataValue::Str(record.wrapping_scheme.clone()), + ) + .add_typed_default( + DataType::CreatedAt, + DataValue::SignedNumber(record.created_at as i128), + ) +} + fn error_response(request: &CommunicationValue, ty: CommunicationType) -> CommunicationValue { CommunicationValue::new(ty) .with_id(request.get_id()) @@ -563,10 +613,50 @@ impl OmikronConnection { break; } + self.flush_pending_chat_secret_forwards().await; self.send_ping().await; } } + async fn forward_chat_secret(&self, cv: &CommunicationValue) -> bool { + self.await_response(cv, Some(Duration::from_secs(10))) + .await + .is_ok() + } + + async fn store_pending_chat_secret_forward(&self, cv: &CommunicationValue) { + let Some(record) = pending_chat_secret_forward_from_cv(cv) else { + return; + }; + let _ = e2ee_storage::put_pending_chat_secret_forward(record); + } + + async fn flush_pending_chat_secret_forwards(&self) { + let Ok(records) = e2ee_storage::get_pending_chat_secret_forwards(100) else { + return; + }; + + for record in records { + let Ok(recipient) = record.recipient_user_id.parse::() else { + let _ = e2ee_storage::delete_pending_chat_secret_forward( + &record.recipient_user_id, + &record.chat_id, + &record.secret_id, + ); + continue; + }; + + let message = chat_secret_forward_cv(&record).with_receiver(recipient); + if self.forward_chat_secret(&message).await { + let _ = e2ee_storage::delete_pending_chat_secret_forward( + &record.recipient_user_id, + &record.chat_id, + &record.secret_id, + ); + } + } + } + // ------------------------------------------------------------------------- // Message Handling (Preserved from original) // ------------------------------------------------------------------------- @@ -590,7 +680,7 @@ impl OmikronConnection { return; } - if cv.is_type(CommunicationType::SetEncryptedDeviceSecret) { + if cv.is_type(CommunicationType::SetChatSecret) { 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; @@ -598,33 +688,34 @@ impl OmikronConnection { } let now = now_millis_i64(); let record = data_string(&cv, DataType::UserId) - .zip(data_string(&cv, DataType::DeviceId)) + .zip(data_string(&cv, DataType::ChatId)) .zip(data_string(&cv, DataType::SecretId)) .zip(data_i64(&cv, DataType::VersionNumber)) .zip(data_bytes(&cv, DataType::EncryptedSecret)) + .zip(data_bytes(&cv, DataType::KemCiphertext)) .zip(data_string(&cv, DataType::WrappingScheme)) - .map(|(((((user_id, device_id), secret_id), version), encrypted_secret), wrapping_scheme)| { - StoredEncryptedDeviceSecret { + .map(|((((((user_id, chat_id), secret_id), version), encrypted_secret), kem_ciphertext), wrapping_scheme)| { + StoredChatSecret { user_id, - device_id, + chat_id, secret_id, version, encrypted_secret, - wrapping_public_key_id: data_string(&cv, DataType::WrappingPublicKeyId), + kem_ciphertext, wrapping_scheme, created_at: data_i64(&cv, DataType::CreatedAt).unwrap_or(now), updated_at: now, } }); - match record.map(e2ee_storage::put_encrypted_device_secret) { + match record.map(e2ee_storage::put_chat_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) { + if cv.is_type(CommunicationType::GetChatSecret) { let Some(user_id) = data_string(&cv, DataType::UserId) else { self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; return; @@ -633,27 +724,29 @@ impl OmikronConnection { self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound)).await; return; } + let Some(chat_id) = data_string(&cv, DataType::ChatId) else { + self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData)).await; + return; + }; - match e2ee_storage::get_encrypted_device_secret(EncryptedDeviceSecretQuery { + match e2ee_storage::get_chat_secret(ChatSecretQuery { user_id, - device_id: data_string(&cv, DataType::DeviceId), + chat_id, secret_id: data_string(&cv, DataType::SecretId), }) { Ok(Some(record)) => { - let mut response = CommunicationValue::new(CommunicationType::EncryptedDeviceSecretResponse) + let response = CommunicationValue::new(CommunicationType::ChatSecretResponse) .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::ChatId, DataValue::Str(record.chat_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::KemCiphertext, DataValue::Bytes(record.kem_ciphertext)) .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, @@ -662,115 +755,23 @@ impl OmikronConnection { return; } - if cv.is_type(CommunicationType::EncryptedMessage) { + if cv.is_type(CommunicationType::ChatSecretForward) { let sender_id = cv.get_sender().to_string(); - if data_string(&cv, DataType::SenderUserId).as_deref() != Some(sender_id.as_str()) { + let recipient_user_id = data_string(&cv, DataType::RecipientUserId).unwrap_or_default(); + if data_string(&cv, DataType::SenderUserId).as_deref() != Some(sender_id.as_str()) + || recipient_user_id.is_empty() + || pending_chat_secret_forward_from_cv(&cv).is_none() + { 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, - } + let forward = cv.clone().with_receiver(recipient_user_id.parse::().unwrap_or(0)); + if self.forward_chat_secret(&forward).await { + self.send_message(&error_response(&cv, CommunicationType::Success)).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, + self.store_pending_chat_secret_forward(&cv).await; + self.send_message(&error_response(&cv, CommunicationType::Success)).await; } return; } diff --git a/type-maps.yaml b/type-maps.yaml index 354f186..34d0757 100644 --- a/type-maps.yaml +++ b/type-maps.yaml @@ -99,13 +99,10 @@ type_maps: AppChallengeResponse: 133 AppIdentificationResponse: 134 LoadTxtRecord: 135 - SetEncryptedDeviceSecret: 139 - GetEncryptedDeviceSecret: 140 - EncryptedDeviceSecretResponse: 141 - EncryptedMessage: 142 - EncryptedMessageAck: 143 - EncryptedMessagesGet: 144 - EncryptedMessagesResponse: 145 + SetChatSecret: 139 + GetChatSecret: 140 + ChatSecretResponse: 141 + ChatSecretForward: 142 DataTypes: ErrorType: 32 ErrorProtocol: 33 @@ -213,18 +210,9 @@ type_maps: SecretId: 142 VersionNumber: 143 EncryptedSecret: 144 - WrappingPublicKeyId: 145 WrappingScheme: 146 UpdatedAt: 147 - MessageId: 148 - ConversationId: 149 - SenderClientId: 150 - RecipientClientId: 151 + ChatId: 148 + KemCiphertext: 149 SenderUserId: 152 RecipientUserId: 153 - EncryptionVersion: 154 - HasMore: 155 - NextCursor: 156 - Since: 157 - Limit: 158 - PeerClientId: 159