From fc85e9377e928fed06adad2945da427945da3b5e Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Sun, 3 May 2026 15:02:15 +0200 Subject: [PATCH] [Fix] Calls, Spelling --- Cargo.lock | 72 +++---- Cargo.toml | 6 +- .../anonymous_client_connection.rs | 8 +- src/calls/call_manager.rs | 37 ++-- src/calls/call_util.rs | 15 ++ src/rho/app_connection.rs | 2 + src/rho/client_connection.rs | 7 +- src/rho/connection.rs | 197 +++++++++--------- src/rho/iota_connection.rs | 88 +++++--- src/rho/rho_connection.rs | 1 + 10 files changed, 251 insertions(+), 182 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 328135d..106aa7f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,40 +2,6 @@ # It is not intended for manual editing. version = 4 -[[package]] -name = "Omikron" -version = "0.1.0" -dependencies = [ - "aes-gcm", - "ansi_term", - "base64 0.22.1", - "dashmap", - "dotenv", - "futures", - "hex", - "hickory-resolver", - "hkdf 0.13.0", - "json", - "livekit-api", - "livekit-protocol", - "log", - "once_cell", - "rand 0.8.6", - "rand_core 0.6.4", - "rustls", - "serde", - "serde_json", - "sha2 0.11.0", - "strum", - "strum_macros", - "thiserror 2.0.18", - "tokio", - "ttp-core", - "ttp-native", - "uuid", - "x448", -] - [[package]] name = "aead" version = "0.5.2" @@ -2152,6 +2118,40 @@ dependencies = [ "asn1-rs", ] +[[package]] +name = "omikron" +version = "0.1.0" +dependencies = [ + "aes-gcm", + "ansi_term", + "base64 0.22.1", + "dashmap", + "dotenv", + "futures", + "hex", + "hickory-resolver", + "hkdf 0.13.0", + "json", + "livekit-api", + "livekit-protocol", + "log", + "once_cell", + "rand 0.8.6", + "rand_core 0.6.4", + "rustls", + "serde", + "serde_json", + "sha2 0.11.0", + "strum", + "strum_macros", + "thiserror 2.0.18", + "tokio", + "ttp-core", + "ttp-native", + "uuid", + "x448", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -3461,7 +3461,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "ttp-core" version = "0.1.0" -source = "git+https://git.methanium.net/Tensamin/TTP.git#9677eb6c6040e4e8d85ba8b6232aa954b5a9438d" +source = "git+https://git.methanium.net/Tensamin/TTP.git#0cc917d75562953b49a83804f9f37e89aca6fd88" dependencies = [ "base64 0.22.1", "byteorder", @@ -3474,7 +3474,7 @@ dependencies = [ [[package]] name = "ttp-native" version = "0.1.0" -source = "git+https://git.methanium.net/Tensamin/TTP.git#9677eb6c6040e4e8d85ba8b6232aa954b5a9438d" +source = "git+https://git.methanium.net/Tensamin/TTP.git#0cc917d75562953b49a83804f9f37e89aca6fd88" dependencies = [ "quinn", "rustls", diff --git a/Cargo.toml b/Cargo.toml index 1858c27..bc0734c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "Omikron" +name = "omikron" version = "0.1.0" edition = "2024" @@ -31,8 +31,8 @@ dotenv = "0.15.0" hkdf = "0.13" strum = "0.28.0" strum_macros = "0.28.0" -livekit-api = { version = "0.4.14", features = ["native-tls"] } -livekit-protocol = "0.7.1" +livekit-api = { version = "0.4.19", features = ["native-tls"] } +livekit-protocol = "0.7.5" thiserror = "2.0.18" hickory-resolver = "0.25.2" serde = "1.0.228" diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index 8c57490..6dff0c8 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -344,7 +344,12 @@ impl AnonymousClientConnection { } }; - let invited = call_manager::add_invite(call_id, self.user_id, receiver_id as u64).await; + let secret = cv + .get_data(DataTypes::call_secret) + .as_str() + .map(|s| s.to_string()); + let invited = + call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await; if !invited { self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_call_id) .await; @@ -459,6 +464,7 @@ impl AnonymousClientConnection { self.send_message(&error).await; } /// Close the connection + #[allow(unused)] pub async fn close(&self) { let mut is_open_guard = self.is_open.write().await; if !*is_open_guard { diff --git a/src/calls/call_manager.rs b/src/calls/call_manager.rs index 4595c77..6c33fd6 100644 --- a/src/calls/call_manager.rs +++ b/src/calls/call_manager.rs @@ -3,7 +3,7 @@ use once_cell::sync::Lazy; use std::sync::Arc; use uuid::Uuid; -use crate::calls::{call_group::CallGroup, caller::Caller}; +use crate::calls::{call_group::CallGroup, call_util, caller::Caller}; pub static CALL_GROUPS: Lazy>> = Lazy::new(|| DashMap::new()); #[allow(dead_code)] @@ -59,29 +59,23 @@ pub async fn get_call_token(user_id: u64, call_id: Uuid) -> Option { return Some(token); } - if let Some(cg) = CALL_GROUPS.get(&call_id) { - let cg_clone = cg.clone(); - - let mut members = cg_clone.members.write().await; - if let Some(member) = members.iter().find(|m| m.user_id == user_id) { - return Some(member.create_token()); - } - let new_caller = Arc::new(Caller::new(user_id, call_id, false)); - let token = new_caller.create_token(); - members.push(new_caller); - return Some(token); - } - let caller = Arc::new(Caller::new(user_id, call_id, true)); - let call_group = CallGroup::new(call_id, caller.clone()); - call_group.update_admins().await; + let call_group = Arc::new(CallGroup::new(call_id, caller.clone())); - CALL_GROUPS.insert(call_id, Arc::new(call_group)); + CALL_GROUPS.insert(call_id, call_group.clone()); + + let _ = call_util::create_room(call_id).await; + call_group.update_admins().await; Some(caller.create_token()) } -pub async fn add_invite(call_id: Uuid, inviter_id: u64, invitee_id: u64) -> bool { +pub async fn add_invite( + call_id: Uuid, + inviter_id: u64, + invitee_id: u64, + secret: Option, +) -> bool { if let Some(cg) = CALL_GROUPS.get(&call_id) { let mut members = cg.members.write().await; @@ -91,6 +85,13 @@ pub async fn add_invite(call_id: Uuid, inviter_id: u64, invitee_id: u64) -> bool if !members.iter().any(|m| m.user_id == invitee_id) { members.push(Arc::new(Caller::new(invitee_id, call_id, false))); } + + if let Some(secret) = secret { + let mut secrets = cg.secrets.write().await; + secrets.insert((inviter_id, invitee_id), secret.clone()); + secrets.insert((invitee_id, inviter_id), secret); + } + return true; } } diff --git a/src/calls/call_util.rs b/src/calls/call_util.rs index 17fefb2..eb976f0 100644 --- a/src/calls/call_util.rs +++ b/src/calls/call_util.rs @@ -1,3 +1,4 @@ +use livekit_api::services::room::CreateRoomOptions; use livekit_api::{ access_token::{self}, services::room::RoomClient, @@ -35,6 +36,20 @@ pub fn get_livekit() -> Result<(String, String, String), ()> { Ok((hostname, api_key, api_secret)) } +pub async fn create_room(call_id: Uuid) -> Result<(), ()> { + let (hostname, api_key, api_secret) = get_livekit()?; + let room_service = RoomClient::with_api_key(&hostname, &api_key, &api_secret); + + let options = CreateRoomOptions::default(); + + room_service + .create_room(&call_id.to_string(), options) + .await + .map_err(|_| ())?; + + Ok(()) +} + pub fn create_token(user_id: u64, call_id: Uuid, has_admin: bool) -> Result { let (_, api_key, api_secret) = get_livekit()?; diff --git a/src/rho/app_connection.rs b/src/rho/app_connection.rs index 0f59cbe..177d5b4 100644 --- a/src/rho/app_connection.rs +++ b/src/rho/app_connection.rs @@ -64,6 +64,7 @@ impl AppConnection { } /// Get current ping + #[allow(unused)] pub async fn get_ping(&self) -> i64 { *self.ping.read().await } @@ -304,6 +305,7 @@ impl AppConnection { } /// Close the connection + #[allow(unused)] pub async fn close(&self) { let mut is_open_guard = self.is_open.write().await; if !*is_open_guard { diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 89e3f29..ff8c215 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -274,7 +274,12 @@ impl ClientConnection { } }; - let invited = call_manager::add_invite(call_id, self.user_id, receiver_id as u64).await; + let secret = cv + .get_data(DataTypes::call_secret) + .as_str() + .map(|s| s.to_string()); + let invited = + call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await; if !invited { self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id) .await; diff --git a/src/rho/connection.rs b/src/rho/connection.rs index 604025b..f5866ad 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -1,5 +1,5 @@ use rand::{Rng, distributions::Alphanumeric}; -use std::{sync::Arc, time::Duration}; +use std::{collections::BTreeMap, collections::HashMap, sync::Arc, time::Duration}; use tokio::sync::RwLock; use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue}; use ttp_native::{Receiver, Sender}; @@ -7,7 +7,7 @@ use uuid::Uuid; use crate::{ anonymous_clients::anonymous_client_connection::AnonymousClientConnection, - calls::{call_group::CallGroup, call_manager, caller::Caller}, + calls::call_manager, get_private_key, get_public_key, log_cv_in, log_cv_out, log_err, log_in, log_out, omega::omega_connection::get_omega_connection, rho::{ @@ -163,7 +163,7 @@ impl GeneralConnection { .unwrap() .export(DataFormat::Base64); - let response = CommunicationValue::new(CommunicationType::app_challange) + let response = CommunicationValue::new(CommunicationType::app_challenge) .with_id(cv.get_id()) .add_data( DataTypes::public_key, @@ -315,7 +315,7 @@ impl GeneralConnection { .export(DataFormat::Base64); let challenge_type = if cv.is_type(CommunicationType::app_identification) { - CommunicationType::app_challange + CommunicationType::app_challenge } else { CommunicationType::challenge }; @@ -337,7 +337,7 @@ impl GeneralConnection { let id = *self.id.read().await as i64; if !cv.is_type(CommunicationType::challenge_response) - && !cv.is_type(CommunicationType::app_challange_response) + && !cv.is_type(CommunicationType::app_challenge_response) { return; } @@ -433,107 +433,108 @@ impl GeneralConnection { if let Some(contacts) = v.as_array() { let call_groups = call_manager::get_call_groups(user_id as u64).await; - let callers = - call_manager::get_call_invites(user_id as u64).await; - let mapped: Vec<(Arc, &Arc)> = { - callers - .iter() - .map(|caller| { - ( - Arc::clone(caller), - call_groups.iter().find(|call_group| { - call_group.call_id == caller.call_id - }), - ) - }) - .filter(|(_, call_group)| call_group.is_some()) - .map(|(caller, call_group)| { - (caller, call_group.unwrap()) - }) - .collect() - }; - if mapped.is_empty() { - DataValue::Array(contacts) - } else { - let mut new_contacts: Vec = Vec::new(); - for contact in contacts { - if let Some(mut contact_map) = contact.as_map() { - let filtered_mapped: Vec<( - Arc, - &Arc, - )> = mapped - .clone() - .into_iter() - .filter(|(caller, _)| { - caller.user_id - == contact_map - .get(&DataTypes::user_id) - .unwrap_or(&DataValue::Null) - .as_number() - .unwrap_or(0) - as u64 - }) - .collect(); - if filtered_mapped.is_empty() { - new_contacts.push( - DataValue::container_from_map(&contact_map), - ); + let mut invites: HashMap> = HashMap::new(); + let mut global_calls: Vec = Vec::new(); + + for call in call_groups { + let members = call.members.read().await; + + let call_self = + members.iter().find(|m| m.user_id == user_id as u64); + + if let Some(call_self) = call_self { + let timeout = *call_self.timeout.read().await; + let admin = call_self.has_admin(); + + // List of all members in the call + let member_ids: Vec = members + .iter() + .map(|m| DataValue::Number(m.user_id as i64)) + .collect(); + + let mut base_call_map: BTreeMap = + BTreeMap::new(); + base_call_map.insert( + DataTypes::call_id, + DataValue::Str(call.call_id.to_string()), + ); + base_call_map.insert( + DataTypes::call_members, + DataValue::Array(member_ids), + ); + + if timeout > 0 { + base_call_map.insert( + DataTypes::timeout, + DataValue::Number(timeout as i64), + ); + } + + if admin { + base_call_map.insert( + DataTypes::has_admin, + DataValue::Bool(true), + ); + } + + // Add to global calls (without contact-specific secret) + global_calls.push(DataValue::container_from_map( + &base_call_map, + )); + + for member in members.iter() { + let member_id = member.user_id; + if member_id == user_id as u64 { continue; } - let mut vec_of_filtered_calls: Vec = - Vec::new(); - for (caller, call_group) in filtered_mapped { - vec_of_filtered_calls.push( - DataValue::Container(vec![ - ( - DataTypes::call_id, - DataValue::Str( - caller.call_id.to_string(), - ), - ), - ( - DataTypes::call_secret, - DataValue::Str( - call_group - .secrets - .read() - .await - .get(&( - contact_map - .get( - &DataTypes::user_id, - ) - .unwrap_or( - &DataValue::Null, - ) - .as_number() - .unwrap_or(0) - as u64, - user_id as u64, - )) - .unwrap_or(&"".to_string()) - .clone(), - ), - ), - ]), + let mut contact_call_map = base_call_map.clone(); + + if let Some(secret) = call + .secrets + .read() + .await + .get(&(member_id, user_id as u64)) + { + contact_call_map.insert( + DataTypes::call_secret, + DataValue::Str(secret.clone()), ); } - if !vec_of_filtered_calls.is_empty() { - contact_map.insert( - DataTypes::calls, - DataValue::Array(vec_of_filtered_calls), - ); - } - new_contacts.push(DataValue::container_from_map( - &contact_map, - )); - } else { - new_contacts.push(contact.clone()); + + invites + .entry(member_id as i64) + .or_insert_with(Vec::new) + .push(DataValue::container_from_map( + &contact_call_map, + )); } } - DataValue::Array(new_contacts) } + + let mut new_contacts: Vec = Vec::new(); + for contact in contacts { + if let Some(mut contact_map) = contact.as_map() { + if let Some(DataValue::Number(id)) = + contact_map.get(&DataTypes::user_id) + { + if let Some(call_list) = invites.get(id) { + contact_map.insert( + DataTypes::calls, + DataValue::Array(call_list.clone()), + ); + } + } + new_contacts + .push(DataValue::container_from_map(&contact_map)); + } else { + new_contacts.push(contact.clone()); + } + } + + ident_resp = ident_resp + .add_data(DataTypes::calls, DataValue::Array(global_calls)); + DataValue::Array(new_contacts) } else { v.clone() } diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 383ec8b..187c584 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -316,44 +316,82 @@ impl IotaConnection { /// Handle GET_CHATS message async fn handle_get_chats(&self, cv: CommunicationValue) { - let receiver_id = cv.get_receiver(); + let user_id = cv.get_sender(); + + // Authority check: user must be linked to this Iota + if !self.get_user_ids().await.contains(&user_id) { + log_err!( + self.iota_id as i64, + PrintType::Iota, + "Rejected get_chats: sender_id={} is not authorized for this iota.", + user_id + ); + return; + } + let mut interested_ids: Vec = Vec::new(); // ============================ // Load Calls // ============================ - let calls: Vec> = call_manager::get_call_groups(receiver_id).await; + let calls: Vec> = call_manager::get_call_groups(user_id).await; let mut invites: HashMap> = HashMap::new(); + let mut global_calls: Vec = Vec::new(); let empty = calls.is_empty(); for call in calls { - for inviter in call.members.read().await.iter() { - let call_self = call.get_caller(receiver_id).await.unwrap(); + let members = call.members.read().await; - let inviter_id = inviter.user_id; + // Find ourselves in the call to get our specific status + let call_self = members.iter().find(|m| m.user_id == user_id).cloned(); + + if let Some(call_self) = call_self { let timeout = *call_self.timeout.read().await; let admin = call_self.has_admin(); - // Build call container - let mut call_map: BTreeMap = BTreeMap::new(); + // List of all members in the call + let member_ids: Vec = members + .iter() + .map(|m| DataValue::Number(m.user_id as i64)) + .collect(); - call_map.insert(DataTypes::call_id, DataValue::Str(call.call_id.to_string())); + // Build base call container + let mut base_call_map: BTreeMap = BTreeMap::new(); + base_call_map.insert(DataTypes::call_id, DataValue::Str(call.call_id.to_string())); + base_call_map.insert(DataTypes::call_members, DataValue::Array(member_ids)); if timeout > 0 { - call_map.insert(DataTypes::timeout, DataValue::Number(timeout as i64)); + base_call_map.insert(DataTypes::timeout, DataValue::Number(timeout as i64)); } if admin { - call_map.insert(DataTypes::has_admin, DataValue::Bool(true)); + base_call_map.insert(DataTypes::has_admin, DataValue::Bool(true)); } - let call_container = DataValue::container_from_map(&call_map); + // Add to global calls (without contact-specific secret) + global_calls.push(DataValue::container_from_map(&base_call_map)); - invites - .entry(inviter_id as i64) - .or_insert_with(Vec::new) - .push(call_container); + // Attach this call to EVERY member of the call (other than ourselves) + for member in members.iter() { + let member_id = member.user_id; + if member_id == user_id { + continue; + } + + let mut contact_call_map = base_call_map.clone(); + + // Add secret if it exists for this pairing + if let Some(secret) = call.secrets.read().await.get(&(member_id, user_id)) { + contact_call_map + .insert(DataTypes::call_secret, DataValue::Str(secret.clone())); + } + + invites + .entry(member_id as i64) + .or_insert_with(Vec::new) + .push(DataValue::container_from_map(&contact_call_map)); + } } } @@ -374,13 +412,10 @@ impl IotaConnection { let mut user_map: BTreeMap = entries.iter().cloned().collect(); - // extract user_id - if let Some(DataValue::Number(user_id)) = user_map.get(&DataTypes::user_id) - { - interested_ids.push(*user_id); + if let Some(DataValue::Number(id)) = user_map.get(&DataTypes::user_id) { + interested_ids.push(*id); - // attach calls if exists - if let Some(call_list) = invites.get(user_id) + if let Some(call_list) = invites.get(id) && !call_list.is_empty() { user_map @@ -399,22 +434,25 @@ impl IotaConnection { // ============================ // Notify Omega // ============================ - OmegaConnection::user_states(receiver_id as i64, interested_ids.clone()).await; + OmegaConnection::user_states(user_id as i64, interested_ids.clone()).await; // ============================ // Notify Rho // ============================ if let Some(rho_conn) = self.get_rho_connection().await { rho_conn - .set_interested(receiver_id as i64, interested_ids) + .set_interested(user_id as i64, interested_ids) .await; } // ============================ // Forward to client // ============================ - self.forward_to_client(cv.add_data(DataTypes::user_ids, enriched_contacts)) - .await; + self.forward_to_client( + cv.add_data(DataTypes::user_ids, enriched_contacts) + .add_data(DataTypes::calls, DataValue::Array(global_calls)), + ) + .await; } /// Forward message to client diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 534cf17..80daf20 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -77,6 +77,7 @@ impl RhoConnection { collections } + #[allow(unused)] pub async fn get_app_connections( &self, userid: Option,