From 0d24154af094f4420581064b4891033cfbfe7a1b Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Fri, 3 Jul 2026 20:17:20 +0200 Subject: [PATCH] [WIP] MTP migration --- .gitignore | 3 + Cargo.lock | 36 ++-- Cargo.toml | 4 + .../anonymous_client_connection.rs | 63 ++++-- src/main.rs | 29 ++- src/omega/omega_connection.rs | 183 ++++-------------- src/rho/app_connection.rs | 12 +- src/rho/client_connection.rs | 137 ++++++++++--- src/rho/connection.rs | 76 ++++---- src/rho/iota_connection.rs | 43 ++-- src/rho/server.rs | 79 +++++--- 11 files changed, 379 insertions(+), 286 deletions(-) diff --git a/.gitignore b/.gitignore index 595e4ab..11e3519 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,6 @@ target # Added by cargo **/target **/logs + +*.mk +*.mpkb diff --git a/Cargo.lock b/Cargo.lock index eaed133..1547438 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1722,12 +1722,13 @@ dependencies = [ [[package]] name = "mtp" version = "0.1.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4" +source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d" dependencies = [ "mtp-client", "mtp-codec", "mtp-common", "mtp-crypto", + "mtp-files", "mtp-host", "mtp-transport", "mtp-type-map", @@ -1736,7 +1737,7 @@ dependencies = [ [[package]] name = "mtp-client" version = "0.1.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4" +source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d" dependencies = [ "mtp-codec", "mtp-common", @@ -1749,7 +1750,7 @@ dependencies = [ [[package]] name = "mtp-codec" version = "0.1.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4" +source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d" dependencies = [ "base64 0.22.1", "byteorder", @@ -1762,7 +1763,7 @@ dependencies = [ [[package]] name = "mtp-common" version = "0.1.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4" +source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d" dependencies = [ "quinn", "rustls", @@ -1773,8 +1774,9 @@ dependencies = [ [[package]] name = "mtp-crypto" version = "0.1.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4" +source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d" dependencies = [ + "base64 0.22.1", "chacha20poly1305", "ed25519-dalek", "getrandom 0.4.3", @@ -1788,10 +1790,19 @@ dependencies = [ "zeroize", ] +[[package]] +name = "mtp-files" +version = "0.1.0" +source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d" +dependencies = [ + "mtp-crypto", + "thiserror 1.0.69", +] + [[package]] name = "mtp-host" version = "0.1.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4" +source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d" dependencies = [ "mtp-codec", "mtp-common", @@ -1804,7 +1815,7 @@ dependencies = [ [[package]] name = "mtp-transport" version = "0.1.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4" +source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d" dependencies = [ "log", "mtp-codec", @@ -1818,7 +1829,7 @@ dependencies = [ [[package]] name = "mtp-type-map" version = "0.1.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4" +source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d" dependencies = [ "serde", "serde_yaml", @@ -1871,9 +1882,9 @@ dependencies = [ [[package]] name = "num-bigint" -version = "0.4.6" +version = "0.4.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9" +checksum = "c863e9ab5e7bf9c99ba75e1050f1e4d624ae87ed3532d6238ffbdc7b585dbbe6" dependencies = [ "num-integer", "num-traits", @@ -2092,6 +2103,7 @@ dependencies = [ "livekit-protocol", "log", "mtp", + "mtp-transport", "once_cell", "rand 0.8.6", "rustls", @@ -2707,9 +2719,9 @@ dependencies = [ [[package]] name = "rustc-hash" -version = "2.1.2" +version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" [[package]] name = "rustc_version" diff --git a/Cargo.toml b/Cargo.toml index d633517..02636ec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,11 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = [ "host", "client", "crypto", + "files", ] } +# Needed directly for `ConnectionHandle`, which `mtp-client` does not re-export. +# Pulled in transitively already via `mtp`, so this just names the same crate/commit. +mtp-transport = { git = "https://git.methanium.net/Methanium/mtp.git" } ansi_term = "*" uuid = { version = "*", features = ["v4"] } diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index 39776a4..b2c90e7 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -1,5 +1,5 @@ use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue, TypeMap}; -use mtp::transport::{Receiver, Sender}; +use mtp::host::{Receiver, Sender}; use std::str::FromStr; use std::sync::Arc; use std::time::Duration; @@ -138,10 +138,11 @@ impl AnonymousClientConnection { for call_invitee in call.members.read().await.clone() { let call_invitee_cv = get_omega_connection() .await_response( - &CommunicationValue::new(CommunicationType::GetUserData).add_typed_default( - DataType::UserId, - DataValue::SignedNumber(call_invitee.user_id.into()), - ), + &CommunicationValue::new(CommunicationType::GetUserData) + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(call_invitee.user_id.into()), + ), Some(Duration::from_secs(2)), ) .await @@ -164,7 +165,12 @@ impl AnonymousClientConnection { call_invitee_cv.get_data(DataType::Avatar).clone(), )); - let _ = invited.push(DataValue::Container(json_invitee.iter().map(|(k, v)| (k.to_id(&TypeMap::latest()), v.clone())).collect())); + let _ = invited.push(DataValue::Container( + json_invitee + .iter() + .map(|(k, v)| (k.to_id(&TypeMap::latest()), v.clone())) + .collect(), + )); } let token = call.create_anonymous_token(self.get_user_id()).await; @@ -190,8 +196,19 @@ impl AnonymousClientConnection { DataType::Display, DataValue::Str(self.get_display_name().await), ) - .add_typed_default(DataType::Avatar, DataValue::Str(self.get_avatar().await)) - .add_typed_default(DataType::CallState, DataValue::Container(serialized.iter().map(|(k, v)| (k.to_id(&TypeMap::latest()), v.clone())).collect())), + .add_typed_default( + DataType::Avatar, + DataValue::Str(self.get_avatar().await), + ) + .add_typed_default( + DataType::CallState, + DataValue::Container( + serialized + .iter() + .map(|(k, v)| (k.to_id(&TypeMap::latest()), v.clone())) + .collect(), + ), + ), ) .await; } @@ -261,7 +278,10 @@ impl AnonymousClientConnection { DataType::Display, DataValue::Str(anonymous.get_display_name().await), ) - .add_typed_default(DataType::UserState, DataValue::Str("online".to_string())) + .add_typed_default( + DataType::UserState, + DataValue::Str("online".to_string()), + ) .add_typed_default( DataType::Avatar, DataValue::Str(anonymous.get_avatar().await), @@ -321,7 +341,10 @@ impl AnonymousClientConnection { /// Handle call invite async fn handle_call_invite(self: Arc, cv: CommunicationValue) { - let receiver_id: i64 = cv.get_data(DataType::ReceiverId).as_signed_number().unwrap_or(0) as i64; + let receiver_id: i64 = cv + .get_data(DataType::ReceiverId) + .as_signed_number() + .unwrap_or(0) as i64; if receiver_id == 0 { self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId) .await; @@ -384,7 +407,10 @@ impl AnonymousClientConnection { let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) .with_id(cv.get_id()) - .add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into())); + .add_typed_default( + DataType::ReceiverId, + DataValue::SignedNumber(receiver_id.into()), + ); self.send_message(&error_cv).await; return; } @@ -446,8 +472,14 @@ impl AnonymousClientConnection { } async fn handle_call_timeout_user(self: Arc, cv: CommunicationValue) { let call_id = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")).unwrap(); - let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0); - let untill = cv.get_data(DataType::Untill).as_signed_number().unwrap_or(0); + let user_id = cv + .get_data(DataType::UserId) + .as_signed_number() + .unwrap_or(0); + let untill = cv + .get_data(DataType::Untill) + .as_signed_number() + .unwrap_or(0); let call = call_manager::get_call(call_id).await; if let Some(call) = call { @@ -467,7 +499,10 @@ impl AnonymousClientConnection { } async fn handle_call_disconnect_user(self: Arc, cv: CommunicationValue) { let call_id = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")).unwrap(); - let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0); + let user_id = cv + .get_data(DataType::UserId) + .as_signed_number() + .unwrap_or(0); let call = call_manager::get_call(call_id).await; if let Some(call) = call { diff --git a/src/main.rs b/src/main.rs index 8476240..e62809c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,34 +15,33 @@ pub static WORKING_DIR: Lazy = Lazy::new(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))); use rustls::crypto::aws_lc_rs::default_provider; -use base64::engine::general_purpose::STANDARD as BASE64_STD; -use base64::Engine as _; use mtp::crypto::Keyring; +use mtp::files::{load_keyring as load_keyring_file, save_keyring, save_public_key_bundle}; use crate::{ - calls::call_util::garbage_collect_calls, - omega::omega_connection::get_omega_connection, - rho::server::start, - util::logger::startup, + calls::call_util::garbage_collect_calls, omega::omega_connection::get_omega_connection, + rho::server::start, util::logger::startup, }; +const KEYRING_PATH: &str = "./omikron.mk"; + static KEYRING: Lazy = Lazy::new(|| { - if let Ok(encoded) = env::var("KEYRING") { - let bytes = BASE64_STD.decode(&encoded).expect("Invalid KEYRING base64"); - Keyring::from_bytes(&bytes).expect("Invalid KEYRING data") - } else { + load_keyring_file(KEYRING_PATH).unwrap_or_else(|_| { let kr = Keyring::generate(); - eprintln!( - "Generated KEYRING (save to env): {}", - BASE64_STD.encode(&kr.to_bytes()) - ); + save_keyring(&kr, KEYRING_PATH).expect("Failed to save generated keyring"); + save_public_key_bundle(&kr.public_key_bundle(), KEYRING_PATH) + .expect("Failed to save generated public key bundle"); + eprintln!("Generated new keyring at {}", KEYRING_PATH); kr - } + }) }); pub fn get_keyring() -> &'static Keyring { &KEYRING } +pub fn load_keyring() -> Keyring { + Keyring::from_bytes(&KEYRING.to_bytes()).unwrap() +} #[tokio::main] async fn main() { diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index 2cca4f7..6e52945 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -1,19 +1,18 @@ -use base64::engine::general_purpose::STANDARD as BASE64_STD; -use base64::Engine as _; -use mtp::crypto::decrypt_with; - use crate::{ data::user::UserStatus, - get_keyring, log, log_cv_in, log_cv_out, log_err, log_in, + load_keyring, log_cv_in, log_cv_out, log_err, log_in, rho::rho_manager::{self, RHO_CONNECTIONS, connection_count}, - util::{ - file_util::load_file_vec, - logger::PrintType, - }, + util::logger::PrintType, }; use dashmap::DashMap; +use mtp::client::{Client, Receiver, Sender}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; -use mtp::transport::{Policy, Receiver, SendMode, Sender}; +use mtp::{ + client::ClientConfig, + files::load_public_key_bundle, + host::{Policy, SendMode}, +}; +use mtp_transport::ConnectionHandle; use once_cell::sync::Lazy; use std::{collections::HashMap, env, sync::Arc, time::Duration}; use tokio::{ @@ -21,6 +20,7 @@ use tokio::{ task::JoinHandle, time::{Instant, sleep}, }; + use uuid::Uuid; // ============================================================================ @@ -109,7 +109,6 @@ pub struct OmegaConnection { connection_loop_handle: Arc>>>, host: String, port: u16, - server_cert: Vec, last_ping: Arc>, heartbeat_handle: Arc>>>, message_send_times: Arc>>, @@ -125,16 +124,9 @@ impl OmegaConnection { Self::with_host(&host, port) } + // omikron and omega are both served over publicly trusted (fullchain) certs on + // the same domain, so the client uses system root trust - no pinning needed. pub fn with_host(host: &str, port: u16) -> Self { - // Load server certificate from default location - let server_cert = - load_file_vec("certs", "cert.pem").expect("Failed to load server certificate"); - - Self::with_host_and_cert(host, port, server_cert) - } - - // New constructor that accepts certificate directly - pub fn with_host_and_cert(host: &str, port: u16, server_cert: Vec) -> Self { let (shutdown_tx, _) = watch::channel(false); OmegaConnection { @@ -143,7 +135,6 @@ impl OmegaConnection { connection_loop_handle: Arc::new(Mutex::new(None)), host: host.to_string(), port, - server_cert, last_ping: Arc::new(Mutex::new(-1)), heartbeat_handle: Arc::new(Mutex::new(None)), message_send_times: Arc::new(Mutex::new(HashMap::new())), @@ -265,12 +256,14 @@ impl OmegaConnection { async fn connect_once(self: Arc) -> Result<(), String> { *self.state.write().await = ConnectionState::Connecting; - let addr_str = format!("https://{}:{}", self.host, self.port); - - let (sender, mut receiver) = mtp::transport::client::connect( - &addr_str, - None, - Policy { + let client_config = ClientConfig::new(format!("https://{}:{}", self.host, self.port)) + .with_client_id( + env::var("ID") + .unwrap_or("0".to_string()) + .parse::() + .unwrap_or(0), + ) + .with_policy(Policy { send_mode: SendMode::SingleStreamPerMessage, max_message_size: 1_000_000_000, close_frame_len: u32::MAX, @@ -285,20 +278,25 @@ impl OmegaConnection { max_transient_recv_errors: 20, transient_recv_backoff: Duration::from_millis(100), receiver_queue_capacity: 1000, - }, - ) - .await - .map_err(|e| format!("Connection failed: {}", e))?; + }); + + let host_public_key = load_public_key_bundle("./omega.mpkb") + .map_err(|e| format!("Failed to load omega.mpkb: {}", e))?; + + let mut connection = + Client::auth_connect(client_config, &load_keyring(), &host_public_key) + .await + .map_err(|e| format!("Connection failed: {}", e))?; log_in!( 0, PrintType::Omega, "QUIC connection established to {}", - addr_str + format!("https://{}:{}", self.host, self.port) ); // Store sender - let sender_arc = Arc::new(sender); + let sender_arc = Arc::new(connection.sender); *self.sender.write().await = Some(sender_arc.clone()); *self.state.write().await = ConnectionState::Connected { identified: false }; @@ -308,12 +306,11 @@ impl OmegaConnection { // Start read loop let read_self = self.clone(); let read_handle = tokio::spawn(async move { - read_self.read_loop(&mut receiver, sender_handle).await; + read_self + .read_loop(&mut connection.receiver, sender_handle) + .await; }); - // Send identification - self.send_identification().await; - // Start heartbeat let heartbeat_self = self.clone(); let heartbeat_handle = tokio::spawn(async move { @@ -332,6 +329,8 @@ impl OmegaConnection { handle.abort(); } + self.clone().sync_client_iota_status().await; + match result { Ok(()) => { // Check if we should reconnect @@ -345,112 +344,6 @@ impl OmegaConnection { } } - // ------------------------------------------------------------------------- - // Identification Handshake - // ------------------------------------------------------------------------- - - async fn send_identification(&self) { - let id = rand::random::(); - - let omikron_id = env::var("ID") - .unwrap_or("0".to_string()) - .parse::() - .unwrap_or(0); - - let identify_msg = CommunicationValue::new(CommunicationType::Identification) - .with_id(id) - .add_typed_default(DataType::OmikronId, DataValue::SignedNumber(omikron_id.into())); - - WAITING_TASKS.insert( - id, - WaitingTask { - task: Box::new(|selfc, cv| { - if cv.is_type(CommunicationType::ErrorNotFound) { - log_err!( - 0, - PrintType::Omega, - "Identification failed: Omikron ID not found" - ); - return false; - } - if !cv.is_type(CommunicationType::Challenge) { - return false; - } - - tokio::spawn(async move { - if let Err(e) = selfc.handle_challenge(cv).await { - log_err!(0, PrintType::Omega, "Challenge handling failed: {}", e); - } - }); - true - }), - inserted_at: Instant::now(), - }, - ); - - self.send_message(&identify_msg).await; - } - - async fn handle_challenge(&self, cv: CommunicationValue) -> Result<(), String> { - let challenge_b64 = cv - .get_data(DataType::Challenge) - .as_str() - .ok_or("Challenge not found")?; - - let blob = BASE64_STD - .decode(challenge_b64) - .map_err(|e| format!("Base64 decode failed: {}", e))?; - let decrypted = decrypt_with(&blob, get_keyring(), b"challenge") - .map_err(|e| format!("Decryption failed: {:?}", e))?; - let decrypted_challenge = - String::from_utf8(decrypted).map_err(|_| "Decrypted challenge not valid UTF-8")?; - - let response_msg = CommunicationValue::new(CommunicationType::ChallengeResponse) - .with_id(cv.get_id()) - .add_typed_default(DataType::Challenge, DataValue::Str(decrypted_challenge)); - - let response_id = response_msg.get_id(); - - WAITING_TASKS.insert( - response_id, - WaitingTask { - task: Box::new(|selfc, final_cv| { - if !final_cv.is_type(CommunicationType::IdentificationResponse) { - log_err!(0, PrintType::Omega, "Expected identification_response"); - return false; - } - - let accepted = final_cv - .get_data(DataType::Accepted) - .as_bool() - .unwrap_or(false); - - if !accepted { - log_err!(0, PrintType::Omega, "Omega did not accept identification"); - return false; - } - - tokio::spawn(async move { - let mut state = selfc.state.write().await; - if let ConnectionState::Connected { identified: _ } = *state { - *state = ConnectionState::Connected { identified: true }; - } - drop(state); - - selfc.sync_client_iota_status().await; - }); - - log!(0, PrintType::Omega, "Successfully identified with Omega"); - true - }), - inserted_at: Instant::now(), - }, - ); - - self.send_message(&response_msg).await; - Ok(()) - } - async fn sync_client_iota_status(self: Arc) { let mut connected_iota_ids: Vec = Vec::new(); let mut connected_user_ids: Vec = Vec::new(); @@ -489,7 +382,7 @@ impl OmegaConnection { async fn read_loop( self: Arc, receiver: &mut Receiver, - sender_handle: Arc, + sender_handle: Arc, ) { // Monitor both receiver and sender handle for close let mut close_rx = sender_handle.subscribe_close(); @@ -520,7 +413,7 @@ impl OmegaConnection { let mut user_ids: Vec = Vec::new(); for value in users { if let DataValue::SignedNumber(user_id) = value { - user_ids.push(*user_id as u64); + user_ids.push(user_id.clone() as u64); } } let connections = crate::rho::rho_manager::RHO_CONNECTIONS.read().await; diff --git a/src/rho/app_connection.rs b/src/rho/app_connection.rs index af13453..5dc292f 100644 --- a/src/rho/app_connection.rs +++ b/src/rho/app_connection.rs @@ -5,7 +5,7 @@ use crate::rho::{rho_connection::RhoConnection, rho_manager}; use crate::util::logger::PrintType; use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; -use mtp::transport::{Receiver, Sender}; +use mtp::host::{Receiver, Sender}; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::RwLock; @@ -129,7 +129,10 @@ impl AppConnection { DataType::Avatar, DataValue::Str(anonymous.get_avatar().await), ) - .add_typed_default(DataType::UserState, DataValue::Str("online".to_string())); + .add_typed_default( + DataType::UserState, + DataValue::Str("online".to_string()), + ); self.send_message(&response).await; @@ -183,7 +186,10 @@ impl AppConnection { // Send pong response let response = CommunicationValue::new(CommunicationType::Pong) .with_id(cv.get_id()) - .add_typed_default(DataType::PingIota, DataValue::SignedNumber(iota_ping.into())); + .add_typed_default( + DataType::PingIota, + DataValue::SignedNumber(iota_ping.into()), + ); self.send_message(&response).await; } diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index d7fe244..2f0ca96 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -7,7 +7,7 @@ use crate::util::logger::PrintType; use crate::{data::user::UserStatus, omega::omega_connection::OmegaConnection}; use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; -use mtp::transport::{Receiver, Sender}; +use mtp::host::{Receiver, Sender}; use std::str::FromStr; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; @@ -158,11 +158,26 @@ impl ClientConnection { } { let response = CommunicationValue::new(CommunicationType::GetUserData) .with_id(cv.get_id()) - .add_typed_default(DataType::Username, DataValue::Str(anonymous.get_user_name().await)) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(anonymous.get_user_id().into())) - .add_typed_default(DataType::Display, DataValue::Str(anonymous.get_display_name().await)) - .add_typed_default(DataType::Avatar, DataValue::Str(anonymous.get_avatar().await)) - .add_typed_default(DataType::UserState, DataValue::Str("online".to_string())); + .add_typed_default( + DataType::Username, + DataValue::Str(anonymous.get_user_name().await), + ) + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(anonymous.get_user_id().into()), + ) + .add_typed_default( + DataType::Display, + DataValue::Str(anonymous.get_display_name().await), + ) + .add_typed_default( + DataType::Avatar, + DataValue::Str(anonymous.get_avatar().await), + ) + .add_typed_default( + DataType::UserState, + DataValue::Str("online".to_string()), + ); self.send_message(&response).await; @@ -198,8 +213,14 @@ impl ClientConnection { let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) .with_id(cv.get_id()) .with_receiver(self.user_id) - .add_typed_default(DataType::Message, DataValue::Str("Missing session_id".to_string())) - .add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id)); + .add_typed_default( + DataType::Message, + DataValue::Str("Missing session_id".to_string()), + ) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(expected_session_id), + ); self.send_message(&response).await; return; }; @@ -208,8 +229,14 @@ impl ClientConnection { let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) .with_id(cv.get_id()) .with_receiver(self.user_id) - .add_typed_default(DataType::Message, DataValue::Str("session_id mismatch".to_string())) - .add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id)); + .add_typed_default( + DataType::Message, + DataValue::Str("session_id mismatch".to_string()), + ) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(expected_session_id), + ); self.send_message(&response).await; return; } @@ -218,13 +245,22 @@ impl ClientConnection { let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) .with_id(cv.get_id()) .with_receiver(self.user_id) - .add_typed_default(DataType::Message, DataValue::Str("session_id mismatch".to_string())) - .add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id)); + .add_typed_default( + DataType::Message, + DataValue::Str("session_id mismatch".to_string()), + ) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(expected_session_id), + ); self.send_message(&response).await; return; } } else { - cv = cv.add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id)); + cv = cv.add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(expected_session_id), + ); } } @@ -266,7 +302,10 @@ impl ClientConnection { // Send pong response let response = CommunicationValue::new(CommunicationType::Pong) .with_id(cv.get_id()) - .add_typed_default(DataType::PingIota, DataValue::SignedNumber(iota_ping.into())); + .add_typed_default( + DataType::PingIota, + DataValue::SignedNumber(iota_ping.into()), + ); self.send_message(&response).await; } @@ -289,7 +328,10 @@ impl ClientConnection { /// Handle call invite async fn handle_call_invite(self: Arc, cv: CommunicationValue) { - let receiver_id: i128 = cv.get_data(DataType::ReceiverId).as_signed_number().unwrap_or(0); + let receiver_id: i128 = cv + .get_data(DataType::ReceiverId) + .as_signed_number() + .unwrap_or(0); if receiver_id == 0 { self.send_error_response(cv.get_id(), CommunicationType::ErrorNoUserId) .await; @@ -334,9 +376,15 @@ impl ClientConnection { // User is offline - send push notification for call invite let push_cv = CommunicationValue::new(CommunicationType::PushNotification) .with_receiver(receiver_id as u64) - .add_typed_default(DataType::SenderId, DataValue::SignedNumber(sender_id.into())) + .add_typed_default( + DataType::SenderId, + DataValue::SignedNumber(sender_id.into()), + ) .add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())) - .add_typed_default(DataType::Notifications, DataValue::Str("call_invite".to_string())); + .add_typed_default( + DataType::Notifications, + DataValue::Str("call_invite".to_string()), + ); let omega_conn = get_omega_connection(); // Send fire-and-forget, don't await to avoid blocking @@ -346,7 +394,10 @@ impl ClientConnection { let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) .with_id(cv.get_id()) - .add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into())); + .add_typed_default( + DataType::ReceiverId, + DataValue::SignedNumber(receiver_id.into()), + ); self.send_message(&error_cv).await; return; } @@ -359,10 +410,19 @@ impl ClientConnection { let forward = CommunicationValue::new(CommunicationType::CallInvite) .with_receiver(receiver_id as u64) .with_sender(sender_id as u64) - .add_typed_default(DataType::CallSecret, cv.get_data(DataType::CallSecret).clone()) + .add_typed_default( + DataType::CallSecret, + cv.get_data(DataType::CallSecret).clone(), + ) .add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())) - .add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into())) - .add_typed_default(DataType::SenderId, DataValue::SignedNumber(sender_id.into())); + .add_typed_default( + DataType::ReceiverId, + DataValue::SignedNumber(receiver_id.into()), + ) + .add_typed_default( + DataType::SenderId, + DataValue::SignedNumber(sender_id.into()), + ); target_rho.message_to_client(forward).await; @@ -461,8 +521,14 @@ impl ClientConnection { .await; return; }; - let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0); - let untill = cv.get_data(DataType::Untill).as_signed_number().unwrap_or(0); + let user_id = cv + .get_data(DataType::UserId) + .as_signed_number() + .unwrap_or(0); + let untill = cv + .get_data(DataType::Untill) + .as_signed_number() + .unwrap_or(0); let Some(call) = call_manager::get_call(call_id).await else { self.send_error_response(cv.get_id(), CommunicationType::ErrorNotFound) @@ -489,7 +555,10 @@ impl ClientConnection { .await; return; }; - let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0); + let user_id = cv + .get_data(DataType::UserId) + .as_signed_number() + .unwrap_or(0); let Some(call) = call_manager::get_call(call_id).await else { self.send_error_response(cv.get_id(), CommunicationType::ErrorNotFound) @@ -546,7 +615,10 @@ impl ClientConnection { let response = CommunicationValue::new(CommunicationType::LoadTxtRecord) .with_id(cv.get_id()) - .add_typed_default(DataType::Content, DataValue::Str(s.to_string())); + .add_typed_default( + DataType::Content, + DataValue::Str(s.to_string()), + ); self.send_message(&response).await; return; } @@ -579,7 +651,10 @@ impl ClientConnection { ); if cv.is_type(CommunicationType::AddConversation) - && cv.get_data(DataType::ChatPartnerId).as_signed_number().is_none() + && cv + .get_data(DataType::ChatPartnerId) + .as_signed_number() + .is_none() { let chat_partner_name = cv .get_data(DataType::ChatPartnerName) @@ -600,7 +675,10 @@ impl ClientConnection { .await_response( &CommunicationValue::new(CommunicationType::GetUserData) .with_id(cv.clone().get_id()) - .add_typed_default(DataType::Username, DataValue::Str(chat_partner_name.clone())), + .add_typed_default( + DataType::Username, + DataValue::Str(chat_partner_name.clone()), + ), Some(Duration::from_secs(20)), ) .await; @@ -665,7 +743,10 @@ impl ClientConnection { ); let error_cv = CommunicationValue::new(CommunicationType::ErrorNoIota) .with_id(msg_id) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(sender_user_id.into())); + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(sender_user_id.into()), + ); self.send_message(&error_cv).await; } } diff --git a/src/rho/connection.rs b/src/rho/connection.rs index 7da7cc1..b875228 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -1,15 +1,15 @@ use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataTypeId, DataValue}; -use mtp::transport::{Receiver, Sender}; +use mtp::host::{Receiver, Sender}; use rand::{Rng, distributions::Alphanumeric}; use std::{collections::BTreeMap, collections::HashMap, sync::Arc, time::Duration}; use tokio::sync::RwLock; use uuid::Uuid; -use base64::engine::general_purpose::STANDARD as BASE64_STD; use base64::Engine as _; +use base64::engine::general_purpose::STANDARD as BASE64_STD; use mtp::crypto::{ - encrypt_for, EncryptionType, KemPublicKey, PublicKeyBundle, SignaturePqPublicKey, - SignaturePublicKey, + EncryptionType, KemPublicKey, PublicKeyBundle, SignaturePqPublicKey, SignaturePublicKey, + encrypt_for, }; use crate::{ @@ -179,15 +179,11 @@ impl GeneralConnection { .unwrap(), ); - let our_pk = BASE64_STD.encode( - get_keyring().public_key_bundle().kem_public_key.as_bytes(), - ); + let our_pk = + BASE64_STD.encode(get_keyring().public_key_bundle().kem_public_key.as_bytes()); let response = CommunicationValue::new(CommunicationType::AppChallenge) .with_id(cv.get_id()) - .add_typed_default( - DataType::PublicKey, - DataValue::Str(our_pk), - ) + .add_typed_default(DataType::PublicKey, DataValue::Str(our_pk)) .add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge)); log_cv_out!(response); @@ -267,15 +263,11 @@ impl GeneralConnection { .unwrap(), ); - let our_pk = BASE64_STD.encode( - get_keyring().public_key_bundle().kem_public_key.as_bytes(), - ); + let our_pk = + BASE64_STD.encode(get_keyring().public_key_bundle().kem_public_key.as_bytes()); let response = CommunicationValue::new(CommunicationType::Challenge) .with_id(cv.get_id()) - .add_typed_default( - DataType::PublicKey, - DataValue::Str(our_pk), - ) + .add_typed_default(DataType::PublicKey, DataValue::Str(our_pk)) .add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge)); log_cv_out!(response); @@ -370,16 +362,12 @@ impl GeneralConnection { CommunicationType::Challenge }; - let our_pk = BASE64_STD.encode( - get_keyring().public_key_bundle().kem_public_key.as_bytes(), - ); + let our_pk = + BASE64_STD.encode(get_keyring().public_key_bundle().kem_public_key.as_bytes()); let response = CommunicationValue::new(challenge_type) .with_id(cv.get_id()) .with_receiver(*self.session_id.read().await) - .add_typed_default( - DataType::PublicKey, - DataValue::Str(our_pk), - ) + .add_typed_default(DataType::PublicKey, DataValue::Str(our_pk)) .add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge)); log_cv_out!(response); @@ -440,7 +428,10 @@ impl GeneralConnection { match kind { ConnectionKind::Client => { let notify = CommunicationValue::new(CommunicationType::UserConnected) - .add_typed_default(DataType::UserId, DataValue::SignedNumber((id as i64).into())); + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber((id as i64).into()), + ); get_omega_connection().send_message(¬ify).await; let user_id = id as i64; @@ -451,7 +442,10 @@ impl GeneralConnection { if rho.is_none() { let get_user_msg = CommunicationValue::new(CommunicationType::GetUserData) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())); + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(user_id.into()), + ); if let Ok(user_data_cv) = get_omega_connection() .await_response(&get_user_msg, Some(Duration::from_secs(20))) @@ -475,8 +469,14 @@ impl GeneralConnection { if let Some(rho_conn) = rho { let session_id = *self.session_id.read().await as i64; let iota_msg = CommunicationValue::new(CommunicationType::ClientConnected) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) - .add_typed_default(DataType::SessionId, DataValue::SignedNumber(session_id.into())); + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(user_id.into()), + ) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ); if let Ok(resp) = rho_conn .get_iota_connection() @@ -526,7 +526,9 @@ impl GeneralConnection { if timeout > 0 { base_call_map.insert( DataType::Timeout.to_id(&tm), - DataValue::SignedNumber((timeout as i64).into()), + DataValue::SignedNumber( + (timeout as i64).into(), + ), ); } @@ -578,7 +580,8 @@ impl GeneralConnection { if let Some(DataValue::SignedNumber(id)) = contact_map.get(&DataType::UserId.to_id(&tm)) { - if let Some(call_list) = invites.get(&(*id as i64)) { + if let Some(call_list) = invites.get(&(*id as i64)) + { contact_map.insert( DataType::Calls.to_id(&tm), DataValue::Array(call_list.clone()), @@ -592,8 +595,10 @@ impl GeneralConnection { } } - ident_resp = ident_resp - .add_typed_default(DataType::Calls, DataValue::Array(global_calls)); + ident_resp = ident_resp.add_typed_default( + DataType::Calls, + DataValue::Array(global_calls), + ); DataValue::Array(new_contacts) } else { v.clone() @@ -672,7 +677,10 @@ impl GeneralConnection { if rho.is_none() { let get_user_msg = CommunicationValue::new(CommunicationType::GetUserData) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())); + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(user_id.into()), + ); if let Ok(user_data_cv) = get_omega_connection() .await_response(&get_user_msg, Some(Duration::from_secs(20))) diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 35c2b2d..c872ccb 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -14,13 +14,13 @@ use mtp::codec::DataType; use mtp::codec::DataTypeId; use mtp::codec::DataValue; use mtp::codec::TypeMap; -use mtp::transport::Receiver; -use mtp::transport::Sender; +use mtp::crypto::KemPublicKey; +use mtp::host::Receiver; +use mtp::host::Sender; use std::collections::BTreeMap; use std::{collections::HashMap, sync::Arc, time::Duration}; use tokio::sync::RwLock; use tokio::sync::mpsc; -use mtp::crypto::KemPublicKey; use super::{rho_connection::RhoConnection, rho_manager}; use crate::omega::omega_connection::OmegaConnection; @@ -265,7 +265,10 @@ impl IotaConnection { if let Ok(uid) = k.parse::() { map.insert(DataType::UserId.to_id(&tm), DataValue::SignedNumber(uid)); } - map.insert(DataType::LastPing.to_id(&tm), DataValue::SignedNumber(v.into())); + map.insert( + DataType::LastPing.to_id(&tm), + DataValue::SignedNumber(v.into()), + ); DataValue::container_from_map(&map) }) .collect(); @@ -368,12 +371,20 @@ impl IotaConnection { // Build base call container let mut base_call_map: BTreeMap = BTreeMap::new(); - base_call_map.insert(DataType::CallId.to_id(&tm), DataValue::Str(call.call_id.to_string())); - base_call_map.insert(DataType::CallMembers.to_id(&tm), DataValue::Array(member_ids)); + base_call_map.insert( + DataType::CallId.to_id(&tm), + DataValue::Str(call.call_id.to_string()), + ); + base_call_map.insert( + DataType::CallMembers.to_id(&tm), + DataValue::Array(member_ids), + ); if timeout > 0 { - base_call_map - .insert(DataType::Timeout.to_id(&tm), DataValue::SignedNumber(timeout.into())); + base_call_map.insert( + DataType::Timeout.to_id(&tm), + DataValue::SignedNumber(timeout.into()), + ); } if admin { @@ -394,8 +405,10 @@ impl IotaConnection { // 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(DataType::CallSecret.to_id(&tm), DataValue::Str(secret.clone())); + contact_call_map.insert( + DataType::CallSecret.to_id(&tm), + DataValue::Str(secret.clone()), + ); } invites @@ -423,14 +436,18 @@ impl IotaConnection { let mut user_map: BTreeMap = entries.iter().cloned().collect(); - if let Some(DataValue::SignedNumber(id)) = user_map.get(&DataType::UserId.to_id(&tm)) { + if let Some(DataValue::SignedNumber(id)) = + user_map.get(&DataType::UserId.to_id(&tm)) + { interested_ids.push(*id as i64); if let Some(call_list) = invites.get(&(*id as i64)) && !call_list.is_empty() { - user_map - .insert(DataType::Calls.to_id(&tm), DataValue::Array(call_list.clone())); + user_map.insert( + DataType::Calls.to_id(&tm), + DataValue::Array(call_list.clone()), + ); } } diff --git a/src/rho/server.rs b/src/rho/server.rs index ee3a837..7f14ae6 100644 --- a/src/rho/server.rs +++ b/src/rho/server.rs @@ -1,46 +1,81 @@ use std::net::{IpAddr, Ipv4Addr}; use std::time::Duration; +use crate::load_keyring; use crate::{ log, rho::connection::GeneralConnection, util::{file_util::load_file_vec, logger::PrintType}, }; -use mtp::transport::{Host, Policy, SendMode, host}; +use mtp::crypto::PublicKeyBundle; +use mtp::host::{AuthenticationPolicy, Host, HostConfig, Policy, SendMode}; + +pub async fn get_by_connector_id( + _client_id: u64, + description: Option, +) -> Option { + if let Some(description) = description { + if description == "iota" { + todo!() + } else if description == "client" { + todo!() + } else if description == "anonymous" { + todo!() + } else if description == "app" { + todo!() + } + } + None +} +pub async fn complete_register(_pub_key: PublicKeyBundle, description: Option) -> u64 { + if let Some(description) = description { + if description == "iota" { + todo!() + } + } + 0 +} pub async fn start(port: u16) -> Result<(), Box> { let cert_pem = load_file_vec("certs", "cert.pem").expect("Error loading Pemfile"); let key_pem = load_file_vec("certs", "key.pem").expect("Error loading Keyfile"); - let mut host: Host = host( - IpAddr::V4(Ipv4Addr::UNSPECIFIED), + let host_config = HostConfig::new( + IpAddr::from(Ipv4Addr::new(0, 0, 0, 0)), port, cert_pem, key_pem, - Policy { - send_mode: SendMode::SingleStreamPerMessage, - max_message_size: 1_000_000_000, - close_frame_len: u32::MAX, - application_close_code: 0, - open_stream_timeout: Duration::from_millis(2_000), - write_timeout: Duration::from_millis(2_000), - accept_stream_timeout: Duration::from_millis(10_000), - read_timeout: Duration::from_millis(30_000), - keep_alive_interval: Some(Duration::from_secs(6)), - max_idle_timeout: Some(Duration::from_secs(30)), - force_close_delay: Duration::from_millis(300), - max_transient_recv_errors: 20, - transient_recv_backoff: Duration::from_millis(100), - receiver_queue_capacity: 1000, - }, ) - .await?; + .with_policy(Policy { + send_mode: SendMode::SingleStreamPerMessage, + max_message_size: 1_000_000_000, + close_frame_len: u32::MAX, + application_close_code: 0, + open_stream_timeout: Duration::from_millis(2_000), + write_timeout: Duration::from_millis(2_000), + accept_stream_timeout: Duration::from_millis(10_000), + read_timeout: Duration::from_millis(30_000), + keep_alive_interval: Some(Duration::from_secs(6)), + max_idle_timeout: Some(Duration::from_secs(30)), + force_close_delay: Duration::from_millis(300), + max_transient_recv_errors: 20, + transient_recv_backoff: Duration::from_millis(100), + receiver_queue_capacity: 1000, + }) + .with_authentication( + load_keyring(), + Box::new(|user_id, description| Box::pin(get_by_connector_id(user_id, description))), + Box::new(|pub_key, description| Box::pin(complete_register(pub_key, description))), + ) + .with_authentication_policy(AuthenticationPolicy::AllowAuthentication); + + let mut host: Host = Host::new(host_config).await?; log!(0, PrintType::General, "Server listening on port {}", port); - while let Some((sender, receiver)) = host.next().await { + while let Ok(Some(conn)) = host.accept().await { tokio::spawn(async move { - let conn = GeneralConnection::new(sender, receiver); + let conn = GeneralConnection::new(conn.sender, conn.receiver); conn.handle().await; }); }