diff --git a/Cargo.lock b/Cargo.lock index ef01e01..db97f40 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2179,6 +2179,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "uuid", ] [[package]] diff --git a/iota-connection/Cargo.toml b/iota-connection/Cargo.toml index bdcd27c..d72230f 100644 --- a/iota-connection/Cargo.toml +++ b/iota-connection/Cargo.toml @@ -15,6 +15,7 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git", rev = "bb0f682b735d serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1.50.0", features = ["rt"] } +uuid = "1" [dev-dependencies] tokio = { version = "1.50.0", features = ["macros", "rt"] } diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index 8b5ff37..410f6b1 100644 --- a/iota-connection/src/message_handlers.rs +++ b/iota-connection/src/message_handlers.rs @@ -1,5 +1,6 @@ use crate::message_common::*; use iota_logger::{LogLevel, PrintType, log_event}; +use iota_storage::util::call_envelopes::{self, CallEnvelope}; use iota_storage::util::chat_files::{self, MessageState}; use iota_storage::util::chats_util::{self, get_user, has_user, mod_user}; use iota_storage::util::communities_util::CommunitiesUtil; @@ -3001,15 +3002,7 @@ pub fn handle_user_block_check(cv: &CommunicationValue) -> CommunicationValue { let Some(receiver_id) = data_i64(cv, DataType::ReceiverId).filter(|id| *id > 0) else { return error_response(cv, CommunicationType::ErrorInvalidData); }; - let sender_principal = match chats_util::get_user(receiver_id, sender_id) { - Ok(Some(contact)) => match contact.principal { - Some(principal) => principal, - None => return error_response(cv, CommunicationType::ErrorNotFound), - }, - Ok(None) => return error_response(cv, CommunicationType::ErrorNotFound), - Err(error) => return internal_handler_error(cv, "blocked_users.check_contact", error), - }; - match blocked_users::is_principal_blocked(receiver_id, sender_principal) { + match blocked_users::is_blocked(receiver_id, sender_id) { Ok(blocked) => CommunicationValue::new(CommunicationType::UserBlockCheck) .with_request_id(cv) .add_typed_default(DataType::IsBlocked, DataValue::Bool(blocked)), @@ -3017,6 +3010,155 @@ pub fn handle_user_block_check(cv: &CommunicationValue) -> CommunicationValue { } } +pub fn handle_call_contact_check(cv: &CommunicationValue) -> CommunicationValue { + let (Some(sender), Some(recipient)) = ( + data_i64(cv, DataType::SenderId).filter(|id| *id > 0), + data_i64(cv, DataType::ReceiverId).filter(|id| *id > 0), + ) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + match chats_util::get_user(sender, recipient) { + Ok(contact) => CommunicationValue::new(CommunicationType::CallContactCheck) + .with_request_id(cv) + .add_typed_default(DataType::IsBlocked, DataValue::Bool(contact.is_some())), + Err(error) => internal_handler_error(cv, "call.contact_check", error), + } +} + +pub fn handle_call_envelope_store( + cv: &CommunicationValue, + owner_omikron_id: i64, +) -> CommunicationValue { + let Some(recipient_id) = data_i64(cv, DataType::ReceiverId).filter(|id| *id > 0) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let Some(inviter_id) = data_i64(cv, DataType::CallInviterId).filter(|id| *id > 0) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let Some(call_id) = + data_string(cv, DataType::CallId).filter(|id| uuid::Uuid::parse_str(id).is_ok()) + else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let Some(created_at) = data_i64(cv, DataType::CreatedAt) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let secret_value = cv.get_data(DataType::CallSecret); + let Some(secret) = secret_value.as_ref() else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let tm = TypeMap::latest(); + let field = |dt: DataType| secret.get_field(dt.try_to_id(&tm)?); + let ( + Some(secret_id), + Some(version_number), + Some(encrypted_secret), + Some(kem_ciphertext), + Some(wrapping_scheme), + ) = ( + field(DataType::SecretId).and_then(DataValue::as_str), + field(DataType::VersionNumber) + .and_then(DataValue::as_number) + .and_then(|n| i64::try_from(n).ok()), + field(DataType::EncryptedSecret).and_then(DataValue::as_bytes), + field(DataType::KemCiphertext).and_then(DataValue::as_bytes), + field(DataType::WrappingScheme).and_then(DataValue::as_str), + ) + else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let now = now_millis_i64(); + let expires_at = created_at.saturating_add(3 * 24 * 60 * 60 * 1000); + if created_at > now + 60_000 + || expires_at <= now + || secret_id.len() > 256 + || encrypted_secret.len() > 1024 + || kem_ciphertext.len() > 4096 + || wrapping_scheme.len() > 128 + { + return error_response(cv, CommunicationType::ErrorInvalidData); + } + let envelope = CallEnvelope { + recipient_id, + call_id, + owner_omikron_id, + inviter_id, + created_at, + expires_at, + secret_id: secret_id.to_string(), + version_number, + encrypted_secret, + kem_ciphertext, + wrapping_scheme: wrapping_scheme.to_string(), + }; + match call_envelopes::store(&envelope, now) { + Ok(true) => success_response(cv), + Ok(false) => error_response(cv, CommunicationType::ErrorNotFound), + Err(error) => internal_handler_error(cv, "call.envelope_store", error), + } +} + +pub fn handle_call_envelope_get( + cv: &CommunicationValue, + owner_omikron_id: i64, +) -> CommunicationValue { + let (Some(recipient_id), Some(call_id)) = ( + data_i64(cv, DataType::ReceiverId).filter(|id| *id > 0), + data_string(cv, DataType::CallId).filter(|id| uuid::Uuid::parse_str(id).is_ok()), + ) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + match call_envelopes::get(recipient_id, &call_id, owner_omikron_id, now_millis_i64()) { + Ok(Some(envelope)) => { + let secret = typed_container(vec![ + (DataType::SecretId, DataValue::Str(envelope.secret_id)), + ( + DataType::VersionNumber, + DataValue::SignedNumber(envelope.version_number.into()), + ), + ( + DataType::EncryptedSecret, + DataValue::Bytes(envelope.encrypted_secret), + ), + ( + DataType::KemCiphertext, + DataValue::Bytes(envelope.kem_ciphertext), + ), + ( + DataType::WrappingScheme, + DataValue::Str(envelope.wrapping_scheme), + ), + ]); + CommunicationValue::new(CommunicationType::CallEnvelopeGet) + .with_request_id(cv) + .add_typed_default(DataType::CallSecret, secret) + .add_typed_default( + DataType::CallInviterId, + DataValue::SignedNumber(envelope.inviter_id.into()), + ) + .add_typed_default( + DataType::CreatedAt, + DataValue::SignedNumber(envelope.created_at.into()), + ) + } + Ok(None) => error_response(cv, CommunicationType::ErrorNotFound), + Err(error) => internal_handler_error(cv, "call.envelope_get", error), + } +} + +pub fn handle_call_contact_accept(cv: &CommunicationValue) -> CommunicationValue { + let (Some(recipient), Some(inviter)) = ( + data_i64(cv, DataType::ReceiverId).filter(|id| *id > 0), + data_i64(cv, DataType::CallInviterId).filter(|id| *id > 0), + ) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + match add_conversation_for_user(recipient, inviter, None) { + Ok(()) => success_response(cv), + Err(error) => internal_handler_error(cv, "call.contact_accept", error), + } +} + #[cfg(test)] mod stored_message_tests { use super::stored_message_value; diff --git a/iota-core/src/main.rs b/iota-core/src/main.rs index ff6ce86..e535bb6 100644 --- a/iota-core/src/main.rs +++ b/iota-core/src/main.rs @@ -106,6 +106,7 @@ async fn main() { log!("Failed to initialize user storage: {}", error); return; } + iota_storage::util::call_envelopes::start_expiry_cleanup(); if let Err(_) = user_manager::load_users_sync() { log_t!("user_load_failed"); } diff --git a/iota-daemon/src/main.rs b/iota-daemon/src/main.rs index 6b0fa6a..7b5ffe4 100644 --- a/iota-daemon/src/main.rs +++ b/iota-daemon/src/main.rs @@ -133,6 +133,7 @@ async fn main() -> ExitCode { ); } else { runtime.set_component_healthy(iota_ipc::ComponentId::Storage, None); + iota_storage::util::call_envelopes::start_expiry_cleanup(); } // Bind before migration and service startup: a successful bind is the diff --git a/iota-storage/src/util/call_envelopes.rs b/iota-storage/src/util/call_envelopes.rs new file mode 100644 index 0000000..497463f --- /dev/null +++ b/iota-storage/src/util/call_envelopes.rs @@ -0,0 +1,81 @@ +use crate::{storage_error::StorageError, util::db}; +use rusqlite::{OptionalExtension, params}; + +pub fn remove_expired(now: i64) -> Result { + db::with_db( + |conn| Ok(conn.execute("DELETE FROM call_envelopes WHERE expires_at <= ?1", [now])?), + ) +} + +pub fn start_expiry_cleanup() { + tokio::spawn(async { + loop { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_millis() as i64) + .unwrap_or_default(); + if let Err(error) = remove_expired(now) { + eprintln!("Call envelope cleanup failed: {error}"); + } + tokio::time::sleep(std::time::Duration::from_secs(60)).await; + } + }); +} + +#[derive(Clone)] +pub struct CallEnvelope { + pub recipient_id: i64, + pub call_id: String, + pub owner_omikron_id: i64, + pub inviter_id: i64, + pub created_at: i64, + pub expires_at: i64, + pub secret_id: String, + pub version_number: i64, + pub encrypted_secret: Vec, + pub kem_ciphertext: Vec, + pub wrapping_scheme: String, +} + +pub fn store(envelope: &CallEnvelope, now: i64) -> Result { + db::with_immediate_transaction(|tx| { + tx.execute("DELETE FROM call_envelopes WHERE expires_at <= ?1", [now])?; + let existing: Option<(i64, i64, i64, String, i64, Vec, Vec, String)> = tx.query_row( + "SELECT owner_omikron_id, inviter_id, created_at, secret_id, version_number, encrypted_secret, kem_ciphertext, wrapping_scheme FROM call_envelopes WHERE recipient_id = ?1 AND call_id = ?2", + params![envelope.recipient_id, envelope.call_id], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?, row.get(5)?, row.get(6)?, row.get(7)?)), + ).optional()?; + if let Some((owner, inviter, created, id, version, encrypted, kem, scheme)) = existing { + return Ok(owner == envelope.owner_omikron_id + && inviter == envelope.inviter_id + && created == envelope.created_at + && id == envelope.secret_id + && version == envelope.version_number + && encrypted == envelope.encrypted_secret + && kem == envelope.kem_ciphertext + && scheme == envelope.wrapping_scheme); + } + tx.execute( + "INSERT INTO call_envelopes (recipient_id, call_id, owner_omikron_id, inviter_id, created_at, expires_at, secret_id, version_number, encrypted_secret, kem_ciphertext, wrapping_scheme) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", + params![envelope.recipient_id, envelope.call_id, envelope.owner_omikron_id, envelope.inviter_id, envelope.created_at, envelope.expires_at, envelope.secret_id, envelope.version_number, envelope.encrypted_secret, envelope.kem_ciphertext, envelope.wrapping_scheme], + )?; + Ok(true) + }) +} + +pub fn get( + recipient_id: i64, + call_id: &str, + owner: i64, + now: i64, +) -> Result, StorageError> { + db::with_db(|conn| { + conn.query_row( + "SELECT recipient_id, call_id, owner_omikron_id, inviter_id, created_at, expires_at, secret_id, version_number, encrypted_secret, kem_ciphertext, wrapping_scheme FROM call_envelopes WHERE recipient_id = ?1 AND call_id = ?2 AND owner_omikron_id = ?3 AND expires_at > ?4", + params![recipient_id, call_id, owner, now], + |row| Ok(CallEnvelope { + recipient_id: row.get(0)?, call_id: row.get(1)?, owner_omikron_id: row.get(2)?, inviter_id: row.get(3)?, created_at: row.get(4)?, expires_at: row.get(5)?, secret_id: row.get(6)?, version_number: row.get(7)?, encrypted_secret: row.get(8)?, kem_ciphertext: row.get(9)?, wrapping_scheme: row.get(10)?, + }), + ).optional().map_err(Into::into) + }) +} diff --git a/iota-storage/src/util/db.rs b/iota-storage/src/util/db.rs index b0f1404..d9e9b86 100644 --- a/iota-storage/src/util/db.rs +++ b/iota-storage/src/util/db.rs @@ -1999,6 +1999,29 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { )?; } + if current_version < 46 { + conn.execute_batch( + r#" + CREATE TABLE IF NOT EXISTS call_envelopes ( + recipient_id INTEGER NOT NULL, + call_id TEXT NOT NULL, + owner_omikron_id INTEGER NOT NULL, + inviter_id INTEGER NOT NULL, + created_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL, + secret_id TEXT NOT NULL, + version_number INTEGER NOT NULL, + encrypted_secret BLOB NOT NULL, + kem_ciphertext BLOB NOT NULL, + wrapping_scheme TEXT NOT NULL, + PRIMARY KEY (recipient_id, call_id) + ); + CREATE INDEX IF NOT EXISTS call_envelopes_expiry ON call_envelopes (expires_at); + PRAGMA user_version = 46; + "#, + )?; + } + Ok(()) } @@ -2073,7 +2096,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, 45); + assert_eq!(version, 46); 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")?; @@ -2092,7 +2115,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, 45); + assert_eq!(version, 46); for table in [ "sync_heads", "sync_events", @@ -2173,7 +2196,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, 45); + assert_eq!(version, 46); for column in [ "id", "user_id", @@ -2268,7 +2291,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, 45); + assert_eq!(version, 46); Ok(()) } @@ -2306,7 +2329,7 @@ mod tests { })?; assert_eq!(count, 0); let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; - assert_eq!(version, 45); + assert_eq!(version, 46); Ok(()) } diff --git a/iota-storage/src/util/mod.rs b/iota-storage/src/util/mod.rs index c9e15b5..3d11d2c 100644 --- a/iota-storage/src/util/mod.rs +++ b/iota-storage/src/util/mod.rs @@ -1,4 +1,5 @@ pub mod blocked_users; +pub mod call_envelopes; pub mod chat_files; pub mod chats_util; pub mod client_message_delivery; diff --git a/mtp-type-maps b/mtp-type-maps index 87f9efe..4eccf97 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit 87f9efe009cd7fea7823907825d8921d85ad4413 +Subproject commit 4eccf976c224a2a0d95c8f3ead596265a74a4fab diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 6a577f8..e5ea992 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -1770,6 +1770,10 @@ impl OmikronConnection { dispatch!(MessageStoragePolicyGet, handle_message_storage_policy_get); dispatch!(MessageStoragePolicySet, handle_message_storage_policy_set); dispatch!(UserBlockCheck, handle_user_block_check); + dispatch!(CallContactCheck, handle_call_contact_check); + dispatch!(CallEnvelopeStore, handle_call_envelope_store); + dispatch!(CallEnvelopeGet, handle_call_envelope_get); + dispatch!(CallContactAccept, handle_call_contact_accept); dispatch!(EraseHostedUserData, handle_erase_hosted_user_data); dispatch!(ProvisionIotaUser, handle_iota_user_provisioning); } @@ -3095,6 +3099,42 @@ impl OmikronConnection { .await; } + async fn handle_call_contact_check(self: Arc, cv: &CommunicationValue) { + let _ = self + .send_message(&message_handlers::handle_call_contact_check(cv)) + .await; + } + + async fn handle_call_envelope_store(self: Arc, cv: &CommunicationValue) { + let response = match CONFIG + .load() + .omikron_id + .and_then(|id| i64::try_from(id).ok()) + { + Some(owner) => message_handlers::handle_call_envelope_store(cv, owner), + None => error_response(cv, CommunicationType::ErrorInternal), + }; + let _ = self.send_message(&response).await; + } + + async fn handle_call_envelope_get(self: Arc, cv: &CommunicationValue) { + let response = match CONFIG + .load() + .omikron_id + .and_then(|id| i64::try_from(id).ok()) + { + Some(owner) => message_handlers::handle_call_envelope_get(cv, owner), + None => error_response(cv, CommunicationType::ErrorInternal), + }; + let _ = self.send_message(&response).await; + } + + async fn handle_call_contact_accept(self: Arc, cv: &CommunicationValue) { + let _ = self + .send_message(&message_handlers::handle_call_contact_accept(cv)) + .await; + } + // ------------------------------------------------------------------------- // Public API // -------------------------------------------------------------------------