diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index 438e744..a75785c 100644 --- a/iota-connection/src/message_handlers.rs +++ b/iota-connection/src/message_handlers.rs @@ -227,6 +227,11 @@ pub fn apply_verified_relay_content( let height = relay_number(&content.content, DataType::Height, &context.type_map) .and_then(|value| i64::try_from(value).ok()) .unwrap_or_default(); + let key_version = + relay_number(&content.content, DataType::VersionNumber, &context.type_map) + .and_then(|value| i64::try_from(value).ok()) + .filter(|value| *value > 0) + .ok_or_else(|| "Relay MessageSend has an invalid VersionNumber".to_string())?; let reply_to = relay_number(&content.content, DataType::ReplyId, &context.type_map) .and_then(|value| i64::try_from(value).ok()); if relay_string( @@ -240,6 +245,22 @@ pub fn apply_verified_relay_content( "Relay MessageSend identity does not match its protected message ID".into(), ); } + let chat_id = if sender_id < recipient_id { + format!("{sender_id}:{recipient_id}") + } else { + format!("{recipient_id}:{sender_id}") + }; + let latest_secret = e2ee_storage::get_chat_secret(ChatSecretQuery { + user_id: storage_owner.to_string(), + chat_id: chat_id.clone(), + secret_id: Some(format!("chat:{chat_id}:main")), + version: None, + }) + .map_err(|error| error.to_string())? + .ok_or_else(|| "Relay MessageSend has no committed chat secret".to_string())?; + if latest_secret.version != key_version { + return Err("Relay MessageSend uses a stale chat secret version".into()); + } chat_files::add_message(chat_files::NewMessage { relay_signer_id: sender_id, relay_message_id: &context.message_id, @@ -254,6 +275,7 @@ pub fn apply_verified_relay_content( sent_by_self, content: message, height, + key_version, reply_to, origin_iota_received_at: sent_by_self.then_some(accepted_at), destination_iota_received_at: (!sent_by_self).then_some(accepted_at), @@ -272,8 +294,29 @@ pub fn apply_verified_relay_content( let send_time = relay_number(&content.content, DataType::SendTime, &context.type_map) .and_then(|value| i64::try_from(value).ok()) .ok_or_else(|| "Relay MessageEdit is missing SendTime".to_string())?; - chat_files::apply_remote_edit(recipient_id, sender_id, send_time, sender_id, message) - .map_err(|error| error.to_string()) + let external_user = if sent_by_self { + recipient_id + } else { + sender_id + }; + let result = if sent_by_self { + chat_files::edit_message( + storage_owner, + external_user, + send_time, + sender_id, + message, + ) + } else { + chat_files::apply_remote_edit( + storage_owner, + external_user, + send_time, + sender_id, + message, + ) + }; + result.map_err(|error| error.to_string()) } "MessageReactionAdd" | "MessageReactionRemove" => { let reaction = relay_string(&content.content, DataType::Reaction, &context.type_map) @@ -282,19 +325,45 @@ pub fn apply_verified_relay_content( let send_time = relay_number(&content.content, DataType::SendTime, &context.type_map) .and_then(|value| i64::try_from(value).ok()) .ok_or_else(|| "Relay reaction is missing SendTime".to_string())?; - let result = if content.message_type == "MessageReactionAdd" { - chat_files::add_reaction(recipient_id, sender_id, send_time, sender_id, reaction) + let external_user = if sent_by_self { + recipient_id } else { - chat_files::remove_reaction(recipient_id, sender_id, send_time, sender_id, reaction) + sender_id + }; + let result = if content.message_type == "MessageReactionAdd" { + chat_files::add_reaction( + storage_owner, + external_user, + send_time, + sender_id, + reaction, + ) + } else { + chat_files::remove_reaction( + storage_owner, + external_user, + send_time, + sender_id, + reaction, + ) }; result.map_err(|error| error.to_string()) } - "MessageDeleteLive" => { + "MessageDelete" | "MessageDeleteLive" => { let send_time = relay_number(&content.content, DataType::SendTime, &context.type_map) .and_then(|value| i64::try_from(value).ok()) .ok_or_else(|| "Relay MessageDeleteLive is missing SendTime".to_string())?; - chat_files::apply_remote_delete(recipient_id, sender_id, send_time, sender_id) - .map_err(|error| error.to_string()) + let external_user = if sent_by_self { + recipient_id + } else { + sender_id + }; + let result = if sent_by_self { + chat_files::delete_message(storage_owner, external_user, send_time) + } else { + chat_files::apply_remote_delete(storage_owner, external_user, send_time, sender_id) + }; + result.map_err(|error| error.to_string()) } "SetChatSecret" => { let frame = CommunicationValue::new(CommunicationType::SetChatSecret) @@ -302,7 +371,7 @@ pub fn apply_verified_relay_content( let recipients = chat_secret_recipients(&frame) .ok_or_else(|| "Relay SetChatSecret has no recipients".to_string())?; let recipient = recipients - .into_iter() + .iter() .find(|value| value.user_id == storage_owner.to_string()) .ok_or_else(|| "Relay SetChatSecret recipient mismatch".to_string())?; let chat_id = data_string(&frame, DataType::ChatId) @@ -310,16 +379,62 @@ pub fn apply_verified_relay_content( let secret_id = data_string(&frame, DataType::SecretId) .ok_or_else(|| "Relay SetChatSecret is missing SecretId".to_string())?; let version = data_i64(&frame, DataType::VersionNumber) + .filter(|value| *value > 0) .ok_or_else(|| "Relay SetChatSecret is missing VersionNumber".to_string())?; let wrapping_scheme = data_string(&frame, DataType::WrappingScheme) .ok_or_else(|| "Relay SetChatSecret is missing WrappingScheme".to_string())?; + let mut participants = chat_id + .split(':') + .map(|value| value.parse::()) + .collect::, _>>() + .map_err(|_| "Relay SetChatSecret ChatId is not a user pair".to_string())?; + if participants.len() != 2 + || participants.contains(&0) + || participants[0] == participants[1] + { + return Err("Relay SetChatSecret ChatId is not a user pair".into()); + } + participants.sort_unstable(); + if chat_id != format!("{}:{}", participants[0], participants[1]) { + return Err("Relay SetChatSecret ChatId is not canonical".into()); + } + if !participants.contains(&context.signer_id) { + return Err("Relay SetChatSecret signer is not in ChatId".into()); + } + if !participants.contains(&context.final_recipient_id) { + return Err("Relay SetChatSecret recipient is not in ChatId".into()); + } + if secret_id != format!("chat:{chat_id}:main") { + return Err("Relay SetChatSecret has an invalid secret namespace".into()); + } + if context.signer_id != participants[0] { + return Err( + "Relay SetChatSecret was signed by a non-authoritative participant".into(), + ); + } + if context.final_recipient_id != participants[1] { + return Err("Relay SetChatSecret recipient is not the peer participant".into()); + } + if recipients.len() != 2 { + return Err("Relay SetChatSecret must contain exactly two recipients".into()); + } + let mut recipient_ids = recipients + .iter() + .map(|value| value.user_id.parse::()) + .collect::, _>>() + .map_err(|_| "Relay SetChatSecret has invalid recipient IDs".to_string())?; + recipient_ids.sort_unstable(); + let expected_recipient_ids = participants.clone(); + if recipient_ids != expected_recipient_ids { + return Err("Relay SetChatSecret recipients do not match relay identities".into()); + } e2ee_storage::put_chat_secret(e2ee_storage::StoredChatSecret { user_id: storage_owner.to_string(), chat_id, secret_id, version, - encrypted_secret: recipient.encrypted_secret, - kem_ciphertext: recipient.kem_ciphertext, + encrypted_secret: recipient.encrypted_secret.clone(), + kem_ciphertext: recipient.kem_ciphertext.clone(), wrapping_scheme, created_at, updated_at: now_millis_i64(), @@ -338,9 +453,20 @@ pub fn apply_verified_relay_content( if user_id <= 0 { return Err("Relay AddConversation has an invalid storage owner".into()); } + if sender_id == recipient_id { + return Err("Relay AddConversation does not support self conversations".into()); + } + if other_id != sender_id && other_id != recipient_id { + return Err("Relay AddConversation does not match relay identities".into()); + } + let expected_partner = if sent_by_self { + recipient_id + } else { + sender_id + }; add_conversation_for_user( user_id, - other_id, + expected_partner, relay_string( &content.content, DataType::ChatPartnerName, @@ -474,13 +600,22 @@ fn stored_message_fields( DataType::Height, DataValue::SignedNumber(message.height as i128), ), + ( + DataType::VersionNumber, + DataValue::SignedNumber(message.key_version as i128), + ), ]; let sender_id = if message.sent_by_self { storage_owner } else { partner_id }; - if let Ok(sender_id) = u128::try_from(sender_id) { + if let Some(relay_signer_id) = message.relay_signer_id { + fields.push(( + DataType::SenderId, + DataValue::SignedNumber(relay_signer_id.into()), + )); + } else if let Ok(sender_id) = u128::try_from(sender_id) { fields.push((DataType::SenderId, DataValue::UnsignedNumber(sender_id))); } if let Some(relay_message_id) = &message.relay_message_id { @@ -489,12 +624,6 @@ fn stored_message_fields( DataValue::Str(relay_message_id.clone()), )); } - if let Some(relay_signer_id) = message.relay_signer_id { - fields.push(( - DataType::SenderId, - DataValue::SignedNumber(relay_signer_id.into()), - )); - } for (data_type, timestamp) in [ (DataType::AuthoredAt, message.authored_at), ( @@ -607,6 +736,7 @@ pub fn handle_get_chat_secret(cv: &CommunicationValue) -> CommunicationValue { user_id, chat_id, secret_id: data_string(cv, DataType::SecretId), + version: data_i64(cv, DataType::VersionNumber), }) { Ok(Some(record)) => CommunicationValue::new(CommunicationType::ChatSecretResponse) .with_request_id(cv) @@ -2326,6 +2456,46 @@ pub fn handle_user_block_check(cv: &CommunicationValue) -> CommunicationValue { } } +#[cfg(test)] +mod stored_message_tests { + use super::stored_message_value; + use iota_storage::util::chat_files::StoredMessage; + use mtp::codec::DataValue; + + #[test] + fn relay_message_value_has_a_single_sender_id() { + let message = StoredMessage { + id: 1, + external_user: 9, + relay_signer_id: Some(9), + relay_message_id: Some("relay-1".to_string()), + message_time: 2, + authored_at: None, + origin_iota_received_at: None, + destination_iota_received_at: None, + client_received_at: None, + client_received_recorded_at: None, + read_at: None, + read_recorded_at: None, + delivery_failed_at: None, + delivery_failure: None, + content: "message".to_string(), + edited: false, + sent_by_self: false, + message_state: "sent".to_string(), + height: 1, + key_version: 1, + reply_to: None, + reactions: Vec::new(), + }; + + let value = stored_message_value(&message, 7, 9); + + assert!(matches!(value, DataValue::Container(_))); + assert!(value.to_bytes().is_ok()); + } +} + #[cfg(test)] mod synced_settings_tests { use super::{handle_synced_setting_get, handle_synced_setting_set, parse_setting_locator}; diff --git a/iota-connection/src/relay.rs b/iota-connection/src/relay.rs index 8dde6c3..b53baae 100644 --- a/iota-connection/src/relay.rs +++ b/iota-connection/src/relay.rs @@ -212,13 +212,12 @@ where pub fn open_verified_relay_content( relay: &VerifiedRelay, keyrings: &[&Keyring], - expected_recipient_id: u64, ) -> Result { Ok(open_relay_content_with_limits_without_replay( &relay.metadata, keyrings, &relay.signing_keys, - Some(expected_recipient_id), + Some(relay.context.final_recipient_id), RelayOpenOptions { policy: RELAY_PROTECTION_POLICY, decode_limits: relay.metadata.decode_limits(), @@ -290,6 +289,24 @@ mod tests { Ok(()) } + #[tokio::test] + async fn content_validation_uses_final_recipient_id() -> Result<(), String> { + let (signer, recipient, frame) = relay("final-recipient")?; + let trusted_key = signer.public_key_bundle(); + let verified = verify_relay_metadata(&frame, 99, &recipient, move |signer_id| async move { + (signer_id == 7) + .then_some(vec![trusted_key]) + .ok_or(RelayValidationError::MissingSigningKeys(signer_id)) + }) + .await + .map_err(|error| error.to_string())?; + + let content = open_verified_relay_content(&verified, &[&recipient]) + .map_err(|error| error.to_string())?; + assert_eq!(content.final_recipient_id, 42); + Ok(()) + } + #[tokio::test] async fn rejects_metadata_signed_by_untrusted_key() -> Result<(), String> { let (_signer, recipient, frame) = relay("wrong-key")?; diff --git a/iota-storage/src/users/user_manager.rs b/iota-storage/src/users/user_manager.rs index 548bbc3..173f77f 100644 --- a/iota-storage/src/users/user_manager.rs +++ b/iota-storage/src/users/user_manager.rs @@ -302,6 +302,10 @@ pub fn finalize_local_release( params![user_id], )?; tx.execute("DELETE FROM users WHERE user_id = ?1", params![user_id])?; + tx.execute( + "UPDATE pending_relays SET delivery_state = 'waiting_client', last_error = 'user_released' WHERE target_kind = 0 AND destination_id = ?1", + params![user_id], + )?; tx.execute( r#"INSERT INTO user_residency (user_id, username, lifecycle_state, data_state, updated_at) VALUES (?1, ?2, 'released', COALESCE((SELECT data_state FROM user_residency WHERE user_id = ?1), 'present'), ?3) diff --git a/iota-storage/src/util/chat_files.rs b/iota-storage/src/util/chat_files.rs index e04406a..aee6032 100644 --- a/iota-storage/src/util/chat_files.rs +++ b/iota-storage/src/util/chat_files.rs @@ -68,6 +68,7 @@ pub struct StoredMessage { pub sent_by_self: bool, pub message_state: String, pub height: i64, + pub key_version: i64, pub reply_to: Option, pub reactions: Vec, } @@ -82,6 +83,7 @@ pub struct NewMessage<'a> { pub sent_by_self: bool, pub content: &'a str, pub height: i64, + pub key_version: i64, pub reply_to: Option, pub origin_iota_received_at: Option, pub destination_iota_received_at: Option, @@ -596,6 +598,7 @@ pub fn add_message(message: NewMessage<'_>) -> Result { sent_by_self, content, height, + key_version, reply_to, origin_iota_received_at, destination_iota_received_at, @@ -617,9 +620,9 @@ pub fn add_message(message: NewMessage<'_>) -> Result { r#" INSERT INTO messages ( storage_owner, external_user, message_time, content, sent_by_self, - message_state, height, reply_to, relay_signer_id, relay_message_id, + message_state, height, key_version, reply_to, relay_signer_id, relay_message_id, authored_at, origin_iota_received_at, destination_iota_received_at, stored_at, expires_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15) + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16) "#, params![ storage_owner, @@ -629,6 +632,7 @@ pub fn add_message(message: NewMessage<'_>) -> Result { i64::from(sent_by_self), initial_state.as_str(), height, + key_version, reply_to, relay_signer_id, relay_message_id, @@ -960,7 +964,7 @@ pub fn get_messages( SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, origin_iota_received_at, destination_iota_received_at, client_received_at, client_received_recorded_at, read_at, - read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, + read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, key_version, reply_to, edited_count FROM messages WHERE storage_owner = ?1 AND external_user = ?2 AND deleted_by_external = 0 AND history_deleted = 0 @@ -991,8 +995,9 @@ pub fn get_messages( sent_by_self: row.get::<_, i64>(14)? != 0, message_state: row.get(15)?, height: row.get(16).unwrap_or(0), - reply_to: row.get(17).ok().flatten(), - edited: row.get::<_, i64>(18).unwrap_or(0) > 0, + key_version: row.get(17).unwrap_or(1), + reply_to: row.get(18).ok().flatten(), + edited: row.get::<_, i64>(19).unwrap_or(0) > 0, reactions: Vec::new(), }) }, @@ -1033,7 +1038,7 @@ pub fn get_message( SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, origin_iota_received_at, destination_iota_received_at, client_received_at, client_received_recorded_at, read_at, - read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, + read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, key_version, reply_to, edited_count, external_user FROM messages WHERE storage_owner = ?1 @@ -1064,9 +1069,10 @@ pub fn get_message( sent_by_self: row.get::<_, i64>(14)? != 0, message_state: row.get(15)?, height: row.get(16).unwrap_or(0), - reply_to: row.get(17).ok().flatten(), - edited: row.get::<_, i64>(18).unwrap_or(0) > 0, - external_user: row.get(19)?, + key_version: row.get(17).unwrap_or(1), + reply_to: row.get(18).ok().flatten(), + edited: row.get::<_, i64>(19).unwrap_or(0) > 0, + external_user: row.get(20)?, reactions: Vec::new(), }) })?; @@ -1138,9 +1144,9 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec Vec(14)? != 0, message_state: row.get(15)?, height: row.get(16).unwrap_or(0), - reply_to: row.get(17).ok().flatten(), - edited: row.get::<_, i64>(18).unwrap_or(0) > 0, + key_version: row.get(17).unwrap_or(1), + reply_to: row.get(18).ok().flatten(), + edited: row.get::<_, i64>(19).unwrap_or(0) > 0, reactions: Vec::new(), }) })?; diff --git a/iota-storage/src/util/db.rs b/iota-storage/src/util/db.rs index c63e481..c12e14c 100644 --- a/iota-storage/src/util/db.rs +++ b/iota-storage/src/util/db.rs @@ -270,6 +270,21 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { )?; } + let messages_exist: bool = conn + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages')", + [], + |row| row.get(0), + ) + .unwrap_or(false); + if messages_exist { + add_column_if_missing( + conn, + "key_version", + "key_version INTEGER NOT NULL DEFAULT 1", + )?; + } + if current_version < 5 { conn.execute_batch( r#" @@ -814,6 +829,35 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { )?; } + if current_version < 24 { + let pending_relays_exist: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'pending_relays')", + [], + |row| row.get(0), + )?; + if pending_relays_exist { + add_table_column_if_missing( + conn, + "pending_relays", + "delivery_state", + "delivery_state TEXT NOT NULL DEFAULT 'ready' CHECK (delivery_state IN ('ready', 'waiting_client', 'quarantined'))", + )?; + add_table_column_if_missing(conn, "pending_relays", "last_error", "last_error TEXT")?; + conn.execute_batch( + r#" + UPDATE pending_relays + SET delivery_state = 'quarantined', last_error = 'target_recipient_mismatch' + WHERE target_kind = 0 + AND relay_destination_user_id IS NOT NULL + AND destination_id <> relay_destination_user_id; + CREATE INDEX IF NOT EXISTS idx_pending_relays_active + ON pending_relays (delivery_state, target_kind, destination_id, id); + "#, + )?; + } + conn.pragma_update(None, "user_version", 24)?; + } + Ok(()) } @@ -883,7 +927,7 @@ mod tests { run_migrations_on_connection(&conn)?; let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; - assert_eq!(version, 23); + assert_eq!(version, 24); for column in ["height", "reply_to", "edited_count", "deleted_by_external"] { let mut statement = conn.prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = ?1")?; @@ -902,7 +946,7 @@ mod tests { run_migrations_on_connection(&conn)?; run_migrations_on_connection(&conn)?; let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; - assert_eq!(version, 23); + assert_eq!(version, 24); for table in [ "sync_heads", "sync_events", @@ -926,7 +970,13 @@ mod tests { )?; assert_eq!(exists, 1); } - for column in ["frame_id", "target_kind", "type_map_version"] { + for column in [ + "frame_id", + "target_kind", + "type_map_version", + "delivery_state", + "last_error", + ] { let mut statement = conn.prepare("SELECT 1 FROM pragma_table_info('pending_relays') WHERE name = ?1")?; assert!(statement.exists([column])?); @@ -942,7 +992,7 @@ mod tests { run_migrations_on_connection(&conn)?; let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; - assert_eq!(version, 23); + assert_eq!(version, 24); for column in [ "id", "user_id", @@ -1037,7 +1087,7 @@ mod tests { )?; let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; assert_eq!(preserved, "remote_committed"); - assert_eq!(version, 23); + assert_eq!(version, 24); Ok(()) } } diff --git a/iota-storage/src/util/e2ee_storage.rs b/iota-storage/src/util/e2ee_storage.rs index 55cf8cd..9e72c94 100644 --- a/iota-storage/src/util/e2ee_storage.rs +++ b/iota-storage/src/util/e2ee_storage.rs @@ -22,10 +22,11 @@ pub struct ChatSecretQuery { pub user_id: String, pub chat_id: String, pub secret_id: Option, + pub version: Option, } static E2EE_DB: LazyLock>> = LazyLock::new(|| { - db::create_shared_connection( + let database = db::create_shared_connection( "e2ee", r#" PRAGMA journal_mode = WAL; @@ -44,7 +45,7 @@ static E2EE_DB: LazyLock>> = LazyLock::new(|| { wrapping_scheme TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, - PRIMARY KEY (user_id, chat_id, secret_id) + PRIMARY KEY (user_id, chat_id, secret_id, version) ); CREATE INDEX IF NOT EXISTS idx_chat_secrets_owner @@ -52,37 +53,81 @@ static E2EE_DB: LazyLock>> = LazyLock::new(|| { "#, ) - .expect("Failed to create or initialize E2EE DB") + .expect("Failed to create or initialize E2EE DB"); + if let Ok(connection) = database.lock() { + let legacy = connection + .query_row( + "SELECT COUNT(*) FROM pragma_table_info('chat_secrets') WHERE name = 'version' AND pk = 4", + [], + |row| row.get::<_, i64>(0), + ) + .unwrap_or(0) + == 0; + if legacy { + connection + .execute_batch( + "ALTER TABLE chat_secrets RENAME TO chat_secrets_legacy; + CREATE TABLE chat_secrets ( + user_id TEXT NOT NULL, chat_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, + updated_at INTEGER NOT NULL, + PRIMARY KEY (user_id, chat_id, secret_id, version) + ); + INSERT INTO chat_secrets SELECT * FROM chat_secrets_legacy; + DROP TABLE chat_secrets_legacy;", + ) + .expect("Failed to migrate E2EE secret history schema"); + } + } + database }); pub fn put_chat_secret(record: StoredChatSecret) -> Result<(), StorageError> { db::with_conn(&E2EE_DB, |conn| { - conn.execute( - r#" - 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) - ON CONFLICT(user_id, chat_id, secret_id) DO UPDATE SET - version = excluded.version, - encrypted_secret = excluded.encrypted_secret, - 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.chat_id, - record.secret_id, - record.version, - record.encrypted_secret, - record.kem_ciphertext, - record.wrapping_scheme, - record.created_at, - record.updated_at, - ], + let tx = conn.unchecked_transaction()?; + let existing = tx.query_row( + "SELECT encrypted_secret, kem_ciphertext, wrapping_scheme FROM chat_secrets WHERE user_id = ?1 AND chat_id = ?2 AND secret_id = ?3 AND version = ?4", + params![&record.user_id, &record.chat_id, &record.secret_id, record.version], + |row| Ok((row.get::<_, Vec>(0)?, row.get::<_, Vec>(1)?, row.get::<_, String>(2)?)), + ).optional()?; + if let Some(existing) = existing { + if existing + != ( + record.encrypted_secret.clone(), + record.kem_ciphertext.clone(), + record.wrapping_scheme.clone(), + ) + { + return Err(rusqlite::Error::InvalidParameterName( + "immutable chat secret version conflict".into(), + )); + } + return Ok(()); + } + let latest = tx.query_row( + "SELECT MAX(version) FROM chat_secrets WHERE user_id = ?1 AND chat_id = ?2 AND secret_id = ?3", + params![&record.user_id, &record.chat_id, &record.secret_id], + |row| row.get::<_, Option>(0), )?; + let expected = match latest { + None => 1, + Some(version) => version.checked_add(1).ok_or_else(|| { + rusqlite::Error::InvalidParameterName("chat secret version overflow".into()) + })?, + }; + if record.version != expected { + return Err(rusqlite::Error::InvalidParameterName(format!( + "chat secret version is out of sequence: expected {expected}, got {}", + record.version + ))); + } + tx.execute( + "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.user_id, record.chat_id, record.secret_id, record.version, record.encrypted_secret, record.kem_ciphertext, record.wrapping_scheme, record.created_at, record.updated_at], + )?; + tx.commit()?; Ok(()) }) } @@ -117,10 +162,11 @@ pub fn get_chat_secret(query: ChatSecretQuery) -> Result, pub relay_destination_user_id: Option, pub relay_message_id: Option, + pub delivery_state: PendingRelayDeliveryState, + pub last_error: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum PendingRelayDeliveryState { + Ready, + WaitingClient, + Quarantined, +} + +impl PendingRelayDeliveryState { + fn from_db(value: String) -> Result { + match value.as_str() { + "ready" => Ok(Self::Ready), + "waiting_client" => Ok(Self::WaitingClient), + "quarantined" => Ok(Self::Quarantined), + _ => Err(rusqlite::Error::InvalidQuery), + } + } } #[derive(Clone, Debug, PartialEq, Eq)] @@ -31,6 +51,15 @@ pub fn enqueue( frame_id: u32, type_map_version: &str, ) -> Result<(), StorageError> { + if let RouteTarget::User(destination) = target { + let destination = i64::try_from(destination) + .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?; + if destination != relay.destination_user_id { + return Err(StorageError::Other( + "user relay target does not match final recipient".into(), + )); + } + } db::with_immediate_transaction(|tx| { enqueue_in_tx( tx, @@ -79,7 +108,7 @@ pub fn enqueue_in_tx( pub fn list(limit: i64) -> Result, StorageError> { db::with_db(|connection| { let mut statement = connection.prepare( - "SELECT id, destination_id, target_kind, frame, created_at, frame_id, type_map_version, relay_signer_id, relay_destination_user_id, relay_message_id FROM pending_relays ORDER BY id LIMIT ?1", + "SELECT id, destination_id, target_kind, frame, created_at, frame_id, type_map_version, relay_signer_id, relay_destination_user_id, relay_message_id, delivery_state, last_error FROM pending_relays ORDER BY id LIMIT ?1", )?; let rows = statement.query_map(params![limit.clamp(1, 500)], |row| { let destination_id = row.get::<_, i64>(1)?; @@ -118,12 +147,26 @@ pub fn list(limit: i64) -> Result, StorageError> { relay_signer_id: row.get(7)?, relay_destination_user_id: row.get(8)?, relay_message_id: row.get(9)?, + delivery_state: PendingRelayDeliveryState::from_db(row.get(10)?)?, + last_error: row.get(11)?, }) })?; rows.collect::, _>>().map_err(Into::into) }) } +/* Select only relays whose target and signer remain locally managed. This is + * the delivery-authority check after user release, independent of row state. */ +pub fn list_active(limit: i64) -> Result, StorageError> { + db::with_db(|connection| { + let mut statement = connection.prepare( + "SELECT p.id, p.destination_id, p.target_kind, p.frame, p.created_at, p.frame_id, p.type_map_version, p.relay_signer_id, p.relay_destination_user_id, p.relay_message_id, p.delivery_state, p.last_error FROM pending_relays p WHERE p.delivery_state = 'ready' AND ((p.target_kind = 0 AND EXISTS (SELECT 1 FROM users u WHERE u.user_id = p.destination_id)) OR (p.target_kind = 1 AND EXISTS (SELECT 1 FROM users u WHERE u.user_id = p.relay_signer_id))) ORDER BY p.id LIMIT ?1", + )?; + let rows = statement.query_map(params![limit.clamp(1, 500)], pending_relay_from_row)?; + rows.collect::, _>>().map_err(Into::into) + }) +} + pub fn list_without_relay_identity() -> Result, StorageError> { list_without_relay_identity_after(0, i64::MAX) } @@ -133,7 +176,7 @@ pub fn list_without_relay_identity_after( limit: i64, ) -> Result, StorageError> { db::with_db(|connection| { - let mut statement = connection.prepare("SELECT id, destination_id, target_kind, frame, created_at, frame_id, type_map_version, relay_signer_id, relay_destination_user_id, relay_message_id FROM pending_relays WHERE id > ?1 AND (relay_signer_id IS NULL OR relay_destination_user_id IS NULL OR relay_message_id IS NULL) ORDER BY id LIMIT ?2")?; + let mut statement = connection.prepare("SELECT id, destination_id, target_kind, frame, created_at, frame_id, type_map_version, relay_signer_id, relay_destination_user_id, relay_message_id, delivery_state, last_error FROM pending_relays WHERE id > ?1 AND (relay_signer_id IS NULL OR relay_destination_user_id IS NULL OR relay_message_id IS NULL) ORDER BY id LIMIT ?2")?; let rows = statement.query_map( params![after_id, limit.clamp(1, 500)], pending_relay_from_row, @@ -161,11 +204,68 @@ fn pending_relay_from_row(row: &rusqlite::Row<'_>) -> Result Result<(), StorageError> { + db::with_immediate_transaction(|tx| { + tx.execute("UPDATE pending_relays SET delivery_state = 'waiting_client', last_error = 'client_offline' WHERE target_kind = 0 AND destination_id = ?1 AND frame_id = ?2", params![destination_user_id, i64::from(frame_id)])?; + Ok(()) + }) +} + +pub fn pause_client_deliveries(destination_user_id: i64) -> Result<(), StorageError> { + db::with_immediate_transaction(|tx| { + tx.execute( + "UPDATE pending_relays SET delivery_state = 'waiting_client', last_error = 'awaiting_state_ack' WHERE target_kind = 0 AND destination_id = ?1 AND delivery_state = 'ready'", + [destination_user_id], + )?; + Ok(()) + }) +} + +pub fn resume_client_deliveries(destination_user_id: i64) -> Result { + db::with_db(|connection| { + Ok(connection.execute("UPDATE pending_relays SET delivery_state = 'ready', last_error = NULL WHERE target_kind = 0 AND destination_id = ?1 AND delivery_state = 'waiting_client'", [destination_user_id])?) + }) +} + +pub fn resume_managed_client_deliveries() -> Result { + db::with_db(|connection| { + Ok(connection.execute("UPDATE pending_relays SET delivery_state = 'ready', last_error = NULL WHERE target_kind = 0 AND delivery_state = 'waiting_client' AND EXISTS (SELECT 1 FROM users u WHERE u.user_id = pending_relays.destination_id)", [])?) + }) +} + +pub fn quarantine_target_mismatch(id: i64) -> Result<(), StorageError> { + db::with_db(|connection| { + connection.execute("UPDATE pending_relays SET delivery_state = 'quarantined', last_error = 'target_recipient_mismatch' WHERE id = ?1", [id])?; + Ok(()) }) } pub fn set_relay_identity(id: i64, relay: &RelayIdentity) -> Result<(), StorageError> { db::with_db(|connection| { + let target_kind = connection.query_row( + "SELECT target_kind FROM pending_relays WHERE id = ?1", + [id], + |row| row.get::<_, i64>(0), + )?; + if target_kind == 0 { + let destination_id = connection.query_row( + "SELECT destination_id FROM pending_relays WHERE id = ?1", + [id], + |row| row.get::<_, i64>(0), + )?; + if destination_id != relay.destination_user_id { + connection.execute( + "UPDATE pending_relays SET delivery_state = 'quarantined', last_error = 'target_recipient_mismatch' WHERE id = ?1", + [id], + )?; + return Ok(()); + } + } connection.execute("UPDATE pending_relays SET relay_signer_id = ?2, relay_destination_user_id = ?3, relay_message_id = ?4 WHERE id = ?1", params![id, relay.signer_id, relay.destination_user_id, relay.message_id])?; Ok(()) }) diff --git a/iota-util/src/route_target.rs b/iota-util/src/route_target.rs index 43e4249..5b9da7b 100644 --- a/iota-util/src/route_target.rs +++ b/iota-util/src/route_target.rs @@ -42,3 +42,12 @@ impl RouteTarget { } } } + +impl std::fmt::Display for RouteTarget { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::User(id) => write!(formatter, "User({id})"), + Self::Iota(id) => write!(formatter, "Iota({id})"), + } + } +} diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 2aafa5a..6de0d0b 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -13,6 +13,7 @@ use rand_core::RngCore; use std::env; use std::fs; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, LazyLock}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::sync::{Mutex, RwLock, Semaphore, oneshot, watch}; @@ -39,6 +40,11 @@ use iota_util::route_target::RouteTarget; const IOTA_KEYRING_PATH: &str = "iota.mk"; static IDENTITY_PATH: std::sync::OnceLock = std::sync::OnceLock::new(); static OMIKRON_TRUST_DIRECTORY: std::sync::OnceLock = std::sync::OnceLock::new(); +static NEXT_CLIENT_EVENT_ID: AtomicU32 = AtomicU32::new(1); + +fn next_client_event_id() -> u32 { + NEXT_CLIENT_EVENT_ID.fetch_add(1, Ordering::Relaxed).max(1) +} fn record_origin_delivery_failure(signer_id: u64, relay_message_id: &str, failure: &str) { let Ok(storage_owner) = i64::try_from(signer_id) else { @@ -480,7 +486,7 @@ impl OmikronConnection { log!("Connecting to Omikron at {}", addr_str); let policy = Policy::default() - .with_send_mode(SendMode::SingleStreamPerMessage) + .with_send_mode(SendMode::PersistentStream) .with_timeouts( Duration::from_millis(2_000), Duration::from_millis(2_000), @@ -538,7 +544,6 @@ impl OmikronConnection { log_t!("omikron_authenticated"); self.classify_legacy_pending_relays().await; - self.flush_pending_relays().await; let maintenance_self = self.clone(); let maintenance_handle = tokio::spawn(async move { @@ -941,6 +946,164 @@ impl OmikronConnection { } } + fn client_event_from_relay( + message_type: &str, + payload: &DataValue, + signer_id: u64, + recipient_id: u64, + relay_message_id: &str, + ) -> Option { + let sender = DataValue::UnsignedNumber(u128::from(signer_id)); + let receiver = recipient_id; + let event_id = next_client_event_id(); + + match message_type { + "MessageSend" => { + let frame = CommunicationValue::new(CommunicationType::MessageSend) + .with_payload(payload.clone()); + let content = frame.get_data(DataType::Content)?.clone(); + let send_time = frame.get_data(DataType::SendTime)?.clone(); + let version = frame.get_data(DataType::VersionNumber)?.clone(); + let mut message = vec![ + (DataType::Content, content), + (DataType::SendTime, send_time), + (DataType::VersionNumber, version), + ( + DataType::RelayMessageId, + DataValue::Str(relay_message_id.to_string()), + ), + (DataType::MessageState, DataValue::Str("sent".to_string())), + ]; + if let Some(reply_id) = frame.get_data(DataType::ReplyId) { + message.push((DataType::ReplyId, reply_id.clone())); + } + Some( + CommunicationValue::new(CommunicationType::MessageLive) + .with_id(event_id) + .with_sender(signer_id) + .with_receiver(receiver) + .add_typed_default(DataType::SenderId, sender) + .add_typed_default(DataType::Message, typed_container(message)), + ) + } + "SetChatSecret" => { + let frame = CommunicationValue::new(CommunicationType::SetChatSecret) + .with_payload(payload.clone()); + let chat_id = frame.get_data(DataType::ChatId)?.clone(); + let secret_id = frame.get_data(DataType::SecretId)?.clone(); + let version = frame.get_data(DataType::VersionNumber)?.clone(); + Some( + CommunicationValue::new(CommunicationType::ChatSecretForward) + .with_id(event_id) + .with_sender(signer_id) + .with_receiver(receiver) + .add_typed_default(DataType::ChatId, chat_id) + .add_typed_default( + DataType::SenderUserId, + DataValue::Str(signer_id.to_string()), + ) + .add_typed_default( + DataType::RecipientUserId, + DataValue::Str(recipient_id.to_string()), + ) + .add_typed_default(DataType::SecretId, secret_id) + .add_typed_default(DataType::VersionNumber, version) + .add_typed_default( + DataType::Payload, + DataValue::Str("available".to_string()), + ), + ) + } + "MessageEdit" => { + let frame = CommunicationValue::new(CommunicationType::MessageEdit) + .with_payload(payload.clone()); + Some( + CommunicationValue::new(CommunicationType::MessageEditLive) + .with_id(event_id) + .with_sender(signer_id) + .with_receiver(receiver) + .add_typed_default( + DataType::Content, + frame.get_data(DataType::Content)?.clone(), + ) + .add_typed_default( + DataType::SendTime, + frame.get_data(DataType::SendTime)?.clone(), + ) + .add_typed_default( + DataType::VersionNumber, + frame.get_data(DataType::VersionNumber)?.clone(), + ) + .add_typed_default(DataType::ChatPartnerId, sender.clone()), + ) + } + "MessageReactionAdd" | "MessageReactionRemove" => { + let frame = CommunicationValue::new(CommunicationType::MessageReactionAdd) + .with_payload(payload.clone()); + Some( + CommunicationValue::new(CommunicationType::MessageReactionLive) + .with_id(event_id) + .with_sender(signer_id) + .with_receiver(receiver) + .add_typed_default( + DataType::Reaction, + frame.get_data(DataType::Reaction)?.clone(), + ) + .add_typed_default( + DataType::SendTime, + frame.get_data(DataType::SendTime)?.clone(), + ) + .add_typed_default(DataType::ChatPartnerId, sender.clone()) + .add_typed_default(DataType::SenderId, sender) + .add_typed_default( + DataType::Accepted, + DataValue::Bool(message_type == "MessageReactionAdd"), + ), + ) + } + "MessageDelete" | "MessageDeleteLive" => { + let frame = CommunicationValue::new(CommunicationType::MessageDelete) + .with_payload(payload.clone()); + Some( + CommunicationValue::new(CommunicationType::MessageDeleteLive) + .with_id(event_id) + .with_sender(signer_id) + .with_receiver(receiver) + .add_typed_default( + DataType::SendTime, + frame.get_data(DataType::SendTime)?.clone(), + ) + .add_typed_default(DataType::ChatPartnerId, sender), + ) + } + "AddConversation" if recipient_id < signer_id => { + let chat_id = format!("{recipient_id}:{signer_id}"); + Some( + CommunicationValue::new(CommunicationType::ChatSecretForward) + .with_id(event_id) + .with_sender(signer_id) + .with_receiver(receiver) + .add_typed_default(DataType::ChatId, DataValue::Str(chat_id.clone())) + .add_typed_default( + DataType::SenderUserId, + DataValue::Str(signer_id.to_string()), + ) + .add_typed_default( + DataType::RecipientUserId, + DataValue::Str(recipient_id.to_string()), + ) + .add_typed_default( + DataType::SecretId, + DataValue::Str(format!("chat:{chat_id}:main")), + ) + .add_typed_default(DataType::VersionNumber, DataValue::SignedNumber(1)) + .add_typed_default(DataType::Payload, DataValue::Str("init".to_string())), + ) + } + _ => None, + } + } + async fn handle_relay(self: Arc, frame: CommunicationValue) { let Some(incoming_frame_id) = frame.id() else { log!("Rejecting Relay without a message id"); @@ -1046,7 +1209,7 @@ impl OmikronConnection { /* Evaluate recipient policy before reserving relay replay state or * persisting the frame, so blocked traffic leaves no durable trace. */ - if recipient_is_local { + if recipient_block_policy_applies(recipient_is_local, recipient_id, signer_id) { match iota_storage::util::blocked_users::is_blocked(recipient_id, signer_id) { Ok(true) => { log!( @@ -1081,11 +1244,7 @@ impl OmikronConnection { /* An origin Iota is authoritative for receipt disclosure. Inspect * local-origin relay content before reserving or queuing the frame. */ if signer_is_local { - let content = match open_verified_relay_content( - &verified, - &[&keyring], - verified.context.signer_id, - ) { + let content = match open_verified_relay_content(&verified, &[&keyring]) { Ok(content) => content, Err(error) => { log!("Relay origin content verification failed: {}", error); @@ -1180,11 +1339,7 @@ impl OmikronConnection { /* A shared Iota owns both independent replicas before delivering to its * local recipient. The destination path below writes the recipient copy. */ if signer_is_local && recipient_is_local && !already_applied { - let content = match open_verified_relay_content( - &verified, - &[&keyring], - verified.context.final_recipient_id, - ) { + let content = match open_verified_relay_content(&verified, &[&keyring]) { Ok(value) => value, Err(error) => { log!( @@ -1241,15 +1396,36 @@ impl OmikronConnection { } if signer_is_local && !recipient_is_local { + /* Chat-secret versions are immutable. Apply them locally only after + * the peer has accepted the same relay, so a peer conflict cannot + * leave a newly generated origin version behind. */ + let defer_chat_secret_commit = frame.is_type(CommunicationType::SetChatSecret); if !already_applied { - let content = match open_verified_relay_content( - &verified, - &[&keyring], - verified.context.signer_id, - ) { - Ok(value) => value, - Err(error) => { - log!("Relay origin content verification failed: {}", error); + if !defer_chat_secret_commit { + let content = match open_verified_relay_content(&verified, &[&keyring]) { + Ok(value) => value, + Err(error) => { + log!("Relay origin content verification failed: {}", error); + let _ = relay_replay::mark_rejected( + verified.context.signer_id, + &verified.context.message_id, + ); + self.send_relay_response( + frame.id(), + CommunicationType::ErrorInvalidData, + ) + .await; + return; + } + }; + if let Err(error) = message_handlers::apply_verified_relay_content( + &verified.context, + &content, + accepted_at, + i64::try_from(verified.context.signer_id).unwrap_or_default(), + true, + ) { + log!("Relay origin application failed: {}", error); let _ = relay_replay::mark_rejected( verified.context.signer_id, &verified.context.message_id, @@ -1258,22 +1434,6 @@ impl OmikronConnection { .await; return; } - }; - if let Err(error) = message_handlers::apply_verified_relay_content( - &verified.context, - &content, - accepted_at, - i64::try_from(verified.context.signer_id).unwrap_or_default(), - true, - ) { - log!("Relay origin application failed: {}", error); - let _ = relay_replay::mark_rejected( - verified.context.signer_id, - &verified.context.message_id, - ); - self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData) - .await; - return; } } let router = match self @@ -1384,6 +1544,40 @@ impl OmikronConnection { .await; return; }; + if defer_chat_secret_commit && !already_applied { + let content = match open_verified_relay_content(&verified, &[&keyring]) { + Ok(value) => value, + Err(error) => { + log!("Relay origin content verification failed: {}", error); + self.send_relay_response( + frame.id(), + CommunicationType::ErrorInvalidData, + ) + .await; + return; + } + }; + let Ok(origin_id) = i64::try_from(verified.context.signer_id) else { + self.send_relay_response( + frame.id(), + CommunicationType::ErrorInvalidData, + ) + .await; + return; + }; + if let Err(error) = message_handlers::apply_verified_relay_content( + &verified.context, + &content, + accepted_at, + origin_id, + true, + ) { + log!("Relay origin chat-secret application failed: {}", error); + self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) + .await; + return; + } + } if let Err(error) = iota_storage::util::downstream_relay::acknowledge_iota_delivery( router, @@ -1417,12 +1611,23 @@ impl OmikronConnection { } Ok(response) => { log!("Relay origin route returned {}", response.get_type()); - if response.is_type(CommunicationType::ErrorInternal) { + if response.is_type(CommunicationType::ErrorInternal) + && !defer_chat_secret_commit + { record_origin_delivery_failure( verified.context.signer_id, &verified.context.message_id, "destination_internal_error", ); + self.send_relay_success( + frame.id(), + local_iota_id, + &verified.context.message_id, + accepted_at, + true, + ) + .await; + return; } else if let Ok(signer_id) = i64::try_from(verified.context.signer_id) { if let Err(error) = iota_storage::util::downstream_relay::reject_iota_delivery( @@ -1456,8 +1661,32 @@ impl OmikronConnection { &verified.context.message_id, "destination_unreachable", ); - self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) - .await; + if defer_chat_secret_commit { + if let Ok(signer_id) = i64::try_from(verified.context.signer_id) { + if let Err(queue_error) = + iota_storage::util::downstream_relay::reject_iota_delivery( + router, + frame_id, + signer_id, + &verified.context.message_id, + "destination_unreachable", + ) + { + log!("Chat-secret relay retry cleanup failed: {}", queue_error); + } + } + self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) + .await; + return; + } + self.send_relay_success( + frame.id(), + local_iota_id, + &verified.context.message_id, + accepted_at, + true, + ) + .await; } } return; @@ -1515,12 +1744,9 @@ impl OmikronConnection { return; } + let mut client_event = None; if !already_applied { - let content = match open_verified_relay_content( - &verified, - &[&keyring], - verified.context.final_recipient_id, - ) { + let content = match open_verified_relay_content(&verified, &[&keyring]) { Ok(value) => value, Err(error) => { log!("Relay content verification failed: {}", error); @@ -1570,6 +1796,13 @@ impl OmikronConnection { .await; return; } + client_event = Self::client_event_from_relay( + &content.message_type, + &content.content, + verified.context.signer_id, + destination, + &verified.context.message_id, + ); if let Err(error) = relay_replay::mark_applied(verified.context.signer_id, &verified.context.message_id) { @@ -1611,8 +1844,10 @@ impl OmikronConnection { signer_is_local, ) .await; - if let Err(error) = self.send_message(&forwarded).await { - log!("Relay delivery to local client failed: {}", error); + if let Some(event) = client_event { + if let Err(error) = self.send_message(&event).await { + log!("Relay client event delivery failed: {}", error); + } } } @@ -1701,6 +1936,15 @@ impl OmikronConnection { continue; } }; + if matches!(record.target, RouteTarget::User(destination) if i64::try_from(destination).ok() != Some(identity.destination_user_id)) + { + log!( + "Quarantining pending Relay {} with a target-recipient mismatch", + record.id + ); + let _ = relay_queue::quarantine_target_mismatch(record.id); + continue; + } if let Err(error) = relay_queue::set_relay_identity(record.id, &identity) { log!( "Pending Relay {} ownership backfill failed: {}", @@ -1713,10 +1957,18 @@ impl OmikronConnection { } async fn flush_pending_relays(&self) { - let Ok(records) = relay_queue::list(100) else { + self.flush_pending_relays_for_user(None).await; + } + + async fn flush_pending_relays_for_user(&self, destination_user_id: Option) { + let Ok(records) = relay_queue::list_active(100) else { return; }; - for record in records { + for record in records.into_iter().filter(|record| { + destination_user_id.is_none_or(|user_id| { + matches!(record.target, RouteTarget::User(destination) if i64::try_from(destination).ok() == Some(user_id)) + }) + }) { if record.relay_signer_id.is_none() || record.relay_destination_user_id.is_none() || record.relay_message_id.is_none() @@ -1831,6 +2083,69 @@ impl OmikronConnection { record.relay_signer_id, ) && relay_id == Some(message_id) { + if frame.is_type(CommunicationType::SetChatSecret) { + let Some(keyring) = self.keyring.read().await.as_ref().cloned() + else { + log!( + "Retaining pending chat-secret Relay {} until the Iota keyring is available", + record.id + ); + continue; + }; + let resolver_connection = self; + let verified = match verify_relay_metadata( + &frame, + destination_iota, + &keyring, + move |signer_id| async move { + resolver_connection + .resolve_relay_signing_keys(signer_id) + .await + }, + ) + .await + { + Ok(value) => value, + Err(error) => { + log!( + "Retaining pending chat-secret Relay {} after verification failure: {}", + record.id, + error + ); + continue; + } + }; + let content = match open_verified_relay_content( + &verified, + &[&keyring], + ) { + Ok(value) => value, + Err(error) => { + log!( + "Retaining pending chat-secret Relay {} after content verification failure: {}", + record.id, + error + ); + continue; + } + }; + if let Err(error) = + message_handlers::apply_verified_relay_content( + &verified.context, + &content, + now_millis_i64(), + signer_id, + true, + ) + { + log!( + "Retaining pending chat-secret Relay {} after origin application failure: {}", + record.id, + error + ); + continue; + } + } if let Err(error) = iota_storage::util::downstream_relay::acknowledge_iota_delivery( destination_iota, @@ -1917,6 +2232,21 @@ impl OmikronConnection { } } + if cv.is_type(CommunicationType::ErrorNoIota) + && cv.get_data(DataType::ErrorType).as_str() == Some("client_offline") + && let (Some(frame_id), Some(destination_id)) = ( + cv.id(), + cv.get_data(DataType::UserId) + .as_number() + .and_then(|id| i64::try_from(id).ok()), + ) + { + if let Err(error) = relay_queue::mark_client_offline(destination_id, frame_id) { + log!("Pending Relay offline state update failed: {}", error); + } + return; + } + let Some(msg_id) = cv.id() else { self.handle_message_impl(cv).await; return; @@ -2271,15 +2601,66 @@ impl OmikronConnection { } async fn handle_client_connected(self: Arc, cv: &CommunicationValue) { - let _ = self - .send_message(&message_handlers::handle_client_connected(cv)) - .await; + let response = message_handlers::handle_client_connected(cv); + let user_id = cv + .require_sender() + .ok() + .and_then(|id| i64::try_from(id).ok()); + let session_id = cv + .get_data(DataType::SessionId) + .as_number() + .and_then(|id| i64::try_from(id).ok()); + + if response.is_type(CommunicationType::ClientStateSync) { + log!( + "ClientStateSync user={:?} session={:?} stage=generated", + user_id, + session_id + ); + if let Some(user_id) = user_id + && let Err(error) = relay_queue::pause_client_deliveries(user_id) + { + log!("Pending Relay state-sync pause failed: {}", error); + } + + if let Err(error) = self.send_message(&response).await { + log!( + "Initial ClientStateSync delivery failed for user {:?}, session {:?}: {}", + user_id, + session_id, + error + ); + } else { + log!( + "ClientStateSync user={:?} session={:?} stage=sent_to_omikron", + user_id, + session_id + ); + } + return; + } + + if let Err(error) = self.send_message(&response).await { + log!("ClientConnected response delivery failed: {}", error); + } } async fn handle_client_state_ack(self: Arc, cv: &CommunicationValue) { - let _ = self - .send_message(&message_handlers::handle_client_state_ack(cv)) - .await; + let response = message_handlers::handle_client_state_ack(cv); + let user_id = cv + .require_sender() + .ok() + .and_then(|id| i64::try_from(id).ok()); + if self.send_message(&response).await.is_ok() + && response.is_type(CommunicationType::Success) + && let Some(user_id) = user_id + { + if let Err(error) = relay_queue::resume_client_deliveries(user_id) { + log!("Pending Relay client resume failed: {}", error); + } else { + self.flush_pending_relays_for_user(Some(user_id)).await; + } + } } fn mutation_live_message( @@ -3075,6 +3456,14 @@ impl OmikronConnection { } } +fn recipient_block_policy_applies( + recipient_is_local: bool, + recipient_id: i64, + signer_id: i64, +) -> bool { + recipient_is_local && recipient_id != signer_id +} + // ============================================================================ // Global Instance // ============================================================================ @@ -3287,4 +3676,11 @@ mod tests { OmikronError::Rejected(CommunicationType::ErrorNotFound, _) )); } + + #[test] + fn recipient_block_policy_skips_self_delivery() { + assert!(!recipient_block_policy_applies(true, 42, 42)); + assert!(recipient_block_policy_applies(true, 42, 43)); + assert!(!recipient_block_policy_applies(false, 42, 43)); + } }