diff --git a/Cargo.lock b/Cargo.lock index 2469eb9..cb3872a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1615,7 +1615,7 @@ dependencies = [ [[package]] name = "mtp" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-client", "mtp-codec", @@ -1631,7 +1631,7 @@ dependencies = [ [[package]] name = "mtp-client" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-codec", "mtp-common", @@ -1644,7 +1644,7 @@ dependencies = [ [[package]] name = "mtp-codec" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "base64 0.23.1", "byteorder", @@ -1658,7 +1658,7 @@ dependencies = [ [[package]] name = "mtp-common" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "quinn", "rustls", @@ -1669,7 +1669,7 @@ dependencies = [ [[package]] name = "mtp-crypto" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "argon2", "base64 0.22.1", @@ -1692,7 +1692,7 @@ dependencies = [ [[package]] name = "mtp-files" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-crypto", "rand 0.10.2", @@ -1703,7 +1703,7 @@ dependencies = [ [[package]] name = "mtp-host" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "mtp-codec", "mtp-common", @@ -1719,7 +1719,7 @@ dependencies = [ [[package]] name = "mtp-transport" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "async-trait", "mtp-codec", @@ -1739,7 +1739,7 @@ dependencies = [ [[package]] name = "mtp-type-map" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "serde", "serde_yaml", @@ -1748,7 +1748,7 @@ dependencies = [ [[package]] name = "mtp-webserver" version = "0.3.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a5c8d4f0c898c78351e9d54124886c86e789a22a" +source = "git+https://git.methanium.net/Methanium/mtp.git#b331b9f6a3943d0331d8fdcb7c696d6c2bc5e8b8" dependencies = [ "async-trait", "bytes", @@ -2026,12 +2026,12 @@ dependencies = [ "rand 0.10.2", "rustls", "serde_json", + "strum", "strum_macros", "thiserror 2.0.20", "tokio", "trust-dns-resolver", "uuid", - "zeroize", ] [[package]] @@ -2988,6 +2988,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "strum" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" + [[package]] name = "strum_macros" version = "0.28.0" diff --git a/Cargo.toml b/Cargo.toml index 101bb0d..8b29617 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,7 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = [ "client", "crypto", "files", + "raw", ] } mtp-transport = { git = "https://git.methanium.net/Methanium/mtp.git" } @@ -27,18 +28,10 @@ rustls = { version = "0.23.42", default-features = false, features = [ tokio = { version = "1.53.0", features = ["full"] } log = "0.4" dotenv = "0.15.0" +strum = "0.28.0" strum_macros = "0.28.0" livekit-api = { version = "0.6.0", features = ["rustls-tls-native-roots"] } livekit-protocol = "=0.7.10" thiserror = "2.0.19" trust-dns-resolver = "0.23.2" serde_json = "1.0.151" -zeroize = "1.9" - -[features] -raw-migration = ["mtp/raw"] - -[[bin]] -name = "migrate_raw_keyring" -path = "src/bin/migrate_raw_keyring.rs" -required-features = ["raw-migration"] diff --git a/README.md b/README.md index 1ca8ee3..401bec9 100644 --- a/README.md +++ b/README.md @@ -4,26 +4,3 @@ It's primary purpose is to connect you're client to your Iota & hide your IP and The Omikron also host Voice-Calls. The Omikron is only used when the Iota is in Centralized and Hybrid mode, or when the Client uses the Tensamin Client with default configuration. - -## Configuration - -`OMIKRON_IDENTITY_SECRET` is required. Provision it through the deployment's secret environment before starting Omikron. Omikron uses it to load `omikron.mk` as a protected keyring and fails startup if the secret or existing identity cannot be loaded. - -Rho connection budgets can be configured with: - -- `RHO_MAX_CONNECTIONS`, default `256` -- `RHO_MAX_ANONYMOUS_CONNECTIONS`, default `128` -- `RHO_MAX_ANONYMOUS_CONNECTIONS_PER_IP`, default `16` - -`RHO_MAX_CONNECTIONS` is applied both to Omikron's application session -semaphore and to the MTP WebServer admission semaphore. This means the same -budget limits transport handshakes and authenticated Rho sessions instead of -only limiting connections after authentication. - -To migrate an existing raw `omikron.mk`, provision `OMIKRON_IDENTITY_SECRET` and run: - -```sh -cargo run --features raw-migration --bin migrate_raw_keyring -``` - -The normal Omikron binary does not enable raw keyring support. diff --git a/mtp-type-maps b/mtp-type-maps index a297dcc..486541b 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit a297dcce60bc6e84696c6a16f5fd510beb2ca643 +Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index 04b9da1..be04ab5 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -9,7 +9,7 @@ use crate::anonymous_clients::anonymous_manager::{self, generate_username}; use crate::app_state::AppState; use crate::calls::call_group::call_invite_secret_from_cv; use crate::rho::connection::{ - GeneralConnection, MtpReceiver, MtpSender, OptionalDataValueCompat, RequiredMtpFields, + GeneralConnection, MtpReceiver, MtpSender, MtpValueCompat, OptionalDataValueCompat, }; use crate::util::data_type_id; use crate::util::logger::PrintType; @@ -108,25 +108,10 @@ impl AnonymousClientConnection { }; tokio::spawn(async move { let _permit = permit; - let message_id = match cv.require_id() { - Ok(message_id) => message_id, - Err(error) => { - let response = - CommunicationValue::new(CommunicationType::ErrorInvalidData).without_id(); - log_out!( - self.user_id as i64, - PrintType::Client, - "Rejected malformed message: {}", - error - ); - self.send_message(&response).await; - return; - } - }; log_cv_in!(PrintType::Client, &cv); if cv.is_type(CommunicationType::Relay) { - self.send_error_response(message_id, CommunicationType::ErrorNotAuthenticated) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorNotAuthenticated) .await; return; } @@ -140,15 +125,18 @@ impl AnonymousClientConnection { call } else { self.send_error_response( - message_id, + &cv.get_id(), CommunicationType::ErrorNotAuthenticated, ) .await; return; } } else { - self.send_error_response(message_id, CommunicationType::ErrorNotAuthenticated) - .await; + self.send_error_response( + &cv.get_id(), + CommunicationType::ErrorNotAuthenticated, + ) + .await; return; }; @@ -221,7 +209,7 @@ impl AnonymousClientConnection { self.clone() .send_message( &&CommunicationValue::new(CommunicationType::IdentificationResponse) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default( DataType::UserId, DataValue::SignedNumber(self.user_id.into()), @@ -256,7 +244,7 @@ impl AnonymousClientConnection { // Presence is account-scoped and anonymous sessions have no // persisted account preference to change. if cv.is_type(CommunicationType::ClientChanged) { - self.send_error_response(message_id, CommunicationType::ErrorNoUserId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId) .await; return; } @@ -302,7 +290,7 @@ impl AnonymousClientConnection { } } { let response = CommunicationValue::new(CommunicationType::GetUserData) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default( DataType::Username, DataValue::Str(anonymous.get_user_name().await), @@ -355,12 +343,9 @@ impl AnonymousClientConnection { /// Handle call invite async fn handle_call_invite(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let receiver_id: i64 = cv.get_data(DataType::ReceiverId).as_number().unwrap_or(0) as i64; if receiver_id == 0 { - self.send_error_response(message_id, CommunicationType::ErrorNoUserId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId) .await; return; } @@ -369,13 +354,13 @@ impl AnonymousClientConnection { Some(DataValue::Str(id_str)) => match Uuid::parse_str(id_str) { Ok(id) => id, Err(_) => { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; } }, _ => { - self.send_error_response(message_id, CommunicationType::ErrorNoCallId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoCallId) .await; return; } @@ -384,7 +369,7 @@ impl AnonymousClientConnection { let secret = match call_invite_secret_from_cv(&cv) { Some(secret) => secret, None => { - self.send_error_response(message_id, CommunicationType::BadRequest) + self.send_error_response(&cv.get_id(), CommunicationType::BadRequest) .await; return; } @@ -395,7 +380,7 @@ impl AnonymousClientConnection { .add_invite(call_id, self.user_id, receiver_id as u64, secret.clone()) .await; if !invited { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; } @@ -405,7 +390,7 @@ impl AnonymousClientConnection { .call_manager .should_forward_invite(self.user_id, receiver_id as u64) { - let response = CommunicationValue::new(CommunicationType::Success).with_id(message_id); + let response = CommunicationValue::new(CommunicationType::Success).with_id(cv.get_id()); self.send_message(&response).await; return; } @@ -437,7 +422,7 @@ impl AnonymousClientConnection { }); let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default( DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into()), @@ -467,28 +452,25 @@ impl AnonymousClientConnection { target_rho.message_to_client(forward).await; - let response = CommunicationValue::new(CommunicationType::Success).with_id(message_id); + let response = CommunicationValue::new(CommunicationType::Success).with_id(cv.get_id()); self.send_message(&response).await; } /// Handle get call request async fn handle_get_call(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let user_id = self.get_user_id(); let call_id = match cv.get_data(DataType::CallId) { Some(DataValue::Str(id_str)) => match Uuid::parse_str(id_str) { Ok(id) => id, Err(_) => { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; } }, _ => { - self.send_error_response(message_id, CommunicationType::ErrorNoCallId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoCallId) .await; return; } @@ -502,7 +484,7 @@ impl AnonymousClientConnection { { Ok(token) => { let response = CommunicationValue::new(CommunicationType::CallToken) - .with_id(message_id) + .with_id(cv.get_id()) .with_receiver(user_id) .add_typed_default(DataType::CallToken, DataValue::Str(token)); self.send_message(&response).await; @@ -514,19 +496,16 @@ impl AnonymousClientConnection { error ); let error_cv = CommunicationValue::new(CommunicationType::ErrorNoCallId) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())); self.send_message(&error_cv).await; } } } async fn handle_call_timeout_user(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")) else { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; }; @@ -540,12 +519,12 @@ impl AnonymousClientConnection { .unwrap_or(0); let Some(call) = self.state.call_manager.get_call(call_id).await else { - self.send_error_response(message_id, CommunicationType::ErrorNotFound) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorNotFound) .await; return; }; let Some(caller) = call.get_caller(self.get_user_id()).await else { - self.send_error_response(message_id, CommunicationType::ErrorInvalidUserId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidUserId) .await; return; }; @@ -556,12 +535,9 @@ impl AnonymousClientConnection { } } async fn handle_call_disconnect_user(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")) else { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; }; @@ -571,12 +547,12 @@ impl AnonymousClientConnection { .unwrap_or(0); let Some(call) = self.state.call_manager.get_call(call_id).await else { - self.send_error_response(message_id, CommunicationType::ErrorNotFound) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorNotFound) .await; return; }; let Some(caller) = call.get_caller(self.get_user_id()).await else { - self.send_error_response(message_id, CommunicationType::ErrorInvalidUserId) + self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidUserId) .await; return; }; @@ -586,8 +562,8 @@ impl AnonymousClientConnection { } /// Send error response - async fn send_error_response(self: Arc, message_id: u32, error_type: CommunicationType) { - let error = CommunicationValue::new(error_type).with_id(message_id); + async fn send_error_response(self: Arc, message_id: &u32, error_type: CommunicationType) { + let error = CommunicationValue::new(error_type).with_id(*message_id); self.send_message(&error).await; } /// Close the connection diff --git a/src/app_state.rs b/src/app_state.rs index 525f469..0b8fb5a 100644 --- a/src/app_state.rs +++ b/src/app_state.rs @@ -1,8 +1,6 @@ -use std::{net::IpAddr, sync::Arc}; +use std::sync::Arc; -use dashmap::DashMap; use mtp::crypto::Keyring; -use tokio::sync::Semaphore; use crate::{ calls::{call_manager::CallManager, call_util::LiveKitService}, @@ -12,58 +10,6 @@ use crate::{ services::call_state::CallStateAggregator, }; -pub struct RhoConnectionLimits { - pub all: Arc, - pub anonymous: Arc, - anonymous_by_ip: Arc>, - max_anonymous_per_ip: usize, -} - -impl RhoConnectionLimits { - pub fn new(max_all: usize, max_anonymous: usize, max_anonymous_per_ip: usize) -> Self { - Self { - all: Arc::new(Semaphore::new(max_all)), - anonymous: Arc::new(Semaphore::new(max_anonymous)), - anonymous_by_ip: Arc::new(DashMap::new()), - max_anonymous_per_ip, - } - } - - pub fn try_acquire_anonymous_per_ip(&self, ip: IpAddr) -> Option { - let mut count = self.anonymous_by_ip.entry(ip).or_insert(0); - if *count >= self.max_anonymous_per_ip { - return None; - } - *count += 1; - drop(count); - - Some(RhoIpConnectionPermit { - ip, - active_by_ip: self.anonymous_by_ip.clone(), - }) - } - - #[cfg(test)] - fn anonymous_count_for_ip(&self, ip: IpAddr) -> usize { - self.anonymous_by_ip.get(&ip).map_or(0, |count| *count) - } -} - -pub struct RhoIpConnectionPermit { - ip: IpAddr, - active_by_ip: Arc>, -} - -impl Drop for RhoIpConnectionPermit { - fn drop(&mut self) { - if let Some(mut count) = self.active_by_ip.get_mut(&self.ip) { - *count = count.saturating_sub(1); - } - self.active_by_ip - .remove_if(&self.ip, |_, count| *count == 0); - } -} - /* * Holds startup dependencies so listener and connection constructors can take * one explicit handle while the remaining manager migrations are completed. @@ -76,7 +22,6 @@ pub struct AppState { pub call_manager: Arc, pub call_state_aggregator: Arc, pub livekit: Arc, - pub rho_connection_limits: Arc, } impl AppState { @@ -88,11 +33,6 @@ impl AppState { call_manager: Arc, livekit: Arc, ) -> Arc { - let rho_connection_limits = Arc::new(RhoConnectionLimits::new( - config.rho_max_connections, - config.rho_max_anonymous_connections, - config.rho_max_anonymous_connections_per_ip, - )); Arc::new(Self { config, keyring, @@ -101,50 +41,10 @@ impl AppState { call_state_aggregator: Arc::new(CallStateAggregator::new(call_manager.clone())), call_manager, livekit, - rho_connection_limits, }) } pub fn keyring_for_host(&self) -> Result { - let bytes = self - .keyring - .try_to_bytes() - .map_err(|error| error.to_string())?; - Keyring::from_bytes(&bytes).map_err(|error| error.to_string()) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::net::{IpAddr, Ipv4Addr}; - - #[tokio::test] - async fn permits_are_released_after_a_session_ends() { - let limits = RhoConnectionLimits::new(1, 1, 1); - let permit = limits.all.clone().try_acquire_owned().unwrap(); - assert!(limits.all.clone().try_acquire_owned().is_err()); - drop(permit); - assert!(limits.all.clone().try_acquire_owned().is_ok()); - } - - #[tokio::test] - async fn anonymous_pool_does_not_consume_global_authenticated_capacity() { - let limits = RhoConnectionLimits::new(2, 1, 1); - let _anonymous = limits.anonymous.clone().try_acquire_owned().unwrap(); - assert!(limits.anonymous.clone().try_acquire_owned().is_err()); - let _authenticated = limits.all.clone().try_acquire_owned().unwrap(); - assert!(limits.all.clone().try_acquire_owned().is_ok()); - } - - #[test] - fn per_ip_permits_are_released() { - let limits = RhoConnectionLimits::new(2, 2, 1); - let ip = IpAddr::V4(Ipv4Addr::LOCALHOST); - let permit = limits.try_acquire_anonymous_per_ip(ip).unwrap(); - assert!(limits.try_acquire_anonymous_per_ip(ip).is_none()); - drop(permit); - assert_eq!(limits.anonymous_count_for_ip(ip), 0); - assert!(limits.try_acquire_anonymous_per_ip(ip).is_some()); + Keyring::from_bytes(&self.keyring.to_bytes()).map_err(|error| error.to_string()) } } diff --git a/src/bin/migrate_raw_keyring.rs b/src/bin/migrate_raw_keyring.rs deleted file mode 100644 index bdd82d4..0000000 --- a/src/bin/migrate_raw_keyring.rs +++ /dev/null @@ -1,23 +0,0 @@ -#[path = "../identity.rs"] -mod identity; - -use identity::{ - KEYRING_PATH, PUBLIC_KEY_PATH, identity_secret_from_environment, migrate_raw_keyring, -}; - -fn main() { - let secret = match identity_secret_from_environment() { - Ok(secret) => secret, - Err(error) => { - eprintln!("Unable to load Omikron identity secret: {error}"); - std::process::exit(1); - } - }; - - if let Err(error) = migrate_raw_keyring(secret.as_slice(), KEYRING_PATH, PUBLIC_KEY_PATH) { - eprintln!("Unable to migrate Omikron identity: {error}"); - std::process::exit(1); - } - - println!("Migrated {KEYRING_PATH} to protected keyring storage"); -} diff --git a/src/calls/call_group.rs b/src/calls/call_group.rs index 54da303..6e29763 100755 --- a/src/calls/call_group.rs +++ b/src/calls/call_group.rs @@ -15,7 +15,6 @@ use crate::{ pub struct CallGroup { pub call_id: Uuid, pub members: RwLock>>, - #[allow(dead_code)] pub show: RwLock, pub anonymous_joining: RwLock, pub short_link: RwLock>, @@ -101,7 +100,6 @@ pub fn call_invite_secret_from_cv(cv: &CommunicationValue) -> Option) -> Self { Self::new_with_service(call_id, user, Arc::new(LiveKitService::new(None))) } diff --git a/src/calls/call_util.rs b/src/calls/call_util.rs index edd2b9b..8476c8c 100644 --- a/src/calls/call_util.rs +++ b/src/calls/call_util.rs @@ -134,7 +134,6 @@ impl LiveKitService { Ok(()) } - #[allow(dead_code)] pub fn garbage_collect_calls(self: Arc, manager: Arc) { tokio::spawn(async move { loop { @@ -149,7 +148,6 @@ impl LiveKitService { } } -#[allow(dead_code)] pub async fn clean_calls(manager: &CallManager, room_service: RoomClient) { let rooms = match tokio::time::timeout(LIVEKIT_REQUEST_TIMEOUT, room_service.list_rooms(Vec::new())) diff --git a/src/calls/caller.rs b/src/calls/caller.rs index ee5c480..6a69566 100644 --- a/src/calls/caller.rs +++ b/src/calls/caller.rs @@ -39,7 +39,6 @@ impl Caller { pub async fn set_timeout(&self, timeout: i64) { *self.timeout.write().await = timeout; } - #[allow(dead_code)] pub fn create_token(&self, livekit: &LiveKitService) -> Result { livekit.create_token(self.user_id, self.call_id, self.has_admin()) } diff --git a/src/config.rs b/src/config.rs index 8137cde..484da67 100644 --- a/src/config.rs +++ b/src/config.rs @@ -7,9 +7,6 @@ const DEFAULT_OMEGA_HOST: &str = "tensamin.net"; const DEFAULT_OMEGA_PORT: u16 = 9187; const DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS: u64 = 20; const DEFAULT_OMEGA_SYNC_RETRIES: u32 = 3; -const DEFAULT_RHO_MAX_CONNECTIONS: usize = 256; -const DEFAULT_RHO_MAX_ANONYMOUS_CONNECTIONS: usize = 128; -const DEFAULT_RHO_MAX_ANONYMOUS_CONNECTIONS_PER_IP: usize = 16; #[derive(Clone, Debug, Eq, PartialEq)] pub struct LiveKitConfig { @@ -29,12 +26,6 @@ pub struct Config { /// Number of synchronization requests before the transport is closed and /// the normal reconnect loop starts. pub omega_sync_retries: u32, - /// Maximum number of application sessions accepted by the Rho listener. - pub rho_max_connections: usize, - /// Maximum number of anonymous application sessions. - pub rho_max_anonymous_connections: usize, - /// Maximum number of anonymous sessions from one peer IP address. - pub rho_max_anonymous_connections_per_ip: usize, pub livekit: Option, } @@ -63,16 +54,6 @@ impl Config { )?); let omega_sync_retries = parse_or_default("OMEGA_SYNC_RETRIES", DEFAULT_OMEGA_SYNC_RETRIES)?.max(1); - let rho_max_connections = - parse_positive_or_default("RHO_MAX_CONNECTIONS", DEFAULT_RHO_MAX_CONNECTIONS)?; - let rho_max_anonymous_connections = parse_positive_or_default( - "RHO_MAX_ANONYMOUS_CONNECTIONS", - DEFAULT_RHO_MAX_ANONYMOUS_CONNECTIONS, - )?; - let rho_max_anonymous_connections_per_ip = parse_positive_or_default( - "RHO_MAX_ANONYMOUS_CONNECTIONS_PER_IP", - DEFAULT_RHO_MAX_ANONYMOUS_CONNECTIONS_PER_IP, - )?; let omega_host = env::var("OMEGA_HOST") .unwrap_or_else(|_| DEFAULT_OMEGA_HOST.to_string()) .trim() @@ -92,25 +73,11 @@ impl Config { omikron_id, omega_sync_timeout, omega_sync_retries, - rho_max_connections, - rho_max_anonymous_connections, - rho_max_anonymous_connections_per_ip, livekit: livekit_from_environment()?, }) } } -fn parse_positive_or_default(name: &'static str, default: usize) -> Result { - let value = parse_or_default(name, default)?; - if value == 0 { - return Err(ConfigError::InvalidValue { - name, - kind: "positive number", - }); - } - Ok(value) -} - fn parse_or_default(name: &'static str, default: T) -> Result where T: std::str::FromStr, diff --git a/src/identity.rs b/src/identity.rs deleted file mode 100644 index c0459c2..0000000 --- a/src/identity.rs +++ /dev/null @@ -1,181 +0,0 @@ -use std::{env, io::ErrorKind, path::Path}; - -use mtp::crypto::Keyring; -use mtp::files::{FileError, load_keyring, save_keyring, save_public_key_bundle}; -use zeroize::Zeroizing; - -pub const KEYRING_PATH: &str = "./omikron.mk"; -pub const PUBLIC_KEY_PATH: &str = "./omikron.mpkb"; -const IDENTITY_SECRET_ENV: &str = "OMIKRON_IDENTITY_SECRET"; - -pub fn identity_secret_from_environment() -> Result>, String> { - let secret = env::var(IDENTITY_SECRET_ENV).map_err(|_| { - format!("{IDENTITY_SECRET_ENV} must be set before Omikron networking starts") - })?; - if secret.trim().is_empty() { - return Err(format!("{IDENTITY_SECRET_ENV} must not be empty")); - } - Ok(Zeroizing::new(secret.into_bytes())) -} - -#[allow(dead_code)] -pub fn load_or_create_keyring( - passphrase: &[u8], - keyring_path: impl AsRef, - public_key_path: impl AsRef, -) -> Result { - let keyring_path = keyring_path.as_ref(); - let public_key_path = public_key_path.as_ref(); - - match load_keyring(keyring_path, passphrase) { - Ok(keyring) => Ok(keyring), - Err(FileError::Io(error)) if error.kind() == ErrorKind::NotFound => { - let keyring = Keyring::generate(); - save_keyring(&keyring, keyring_path, passphrase).map_err(|error| { - format!("unable to persist Omikron keyring at {keyring_path:?}: {error}") - })?; - save_public_key_bundle(&keyring.public_key_bundle(), public_key_path).map_err( - |error| { - format!( - "unable to persist Omikron public key bundle at {public_key_path:?}: {error}" - ) - }, - )?; - eprintln!( - "Generated new protected keyring at {}", - keyring_path.display() - ); - Ok(keyring) - } - Err(error) => Err(format!( - "unable to load existing Omikron identity from {}: {error}", - keyring_path.display() - )), - } -} - -#[cfg(feature = "raw-migration")] -#[allow(dead_code)] -pub fn migrate_raw_keyring( - passphrase: &[u8], - keyring_path: impl AsRef, - public_key_path: impl AsRef, -) -> Result { - use mtp::files::load_keyring_raw; - - let keyring = load_keyring_raw(keyring_path.as_ref()) - .map_err(|error| format!("raw keyring migration failed: {error}"))?; - save_keyring(&keyring, keyring_path.as_ref(), passphrase) - .map_err(|error| format!("protected keyring write failed: {error}"))?; - save_public_key_bundle(&keyring.public_key_bundle(), public_key_path.as_ref()) - .map_err(|error| format!("public key bundle write failed: {error}"))?; - Ok(keyring) -} - -#[cfg(test)] -mod tests { - use super::*; - use std::{ - fs, - time::{SystemTime, UNIX_EPOCH}, - }; - - struct TestPaths { - directory: std::path::PathBuf, - keyring: std::path::PathBuf, - public_key: std::path::PathBuf, - } - - impl TestPaths { - fn new() -> Self { - let suffix = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_nanos(); - let directory = env::temp_dir().join(format!("omikron-identity-{suffix}")); - fs::create_dir_all(&directory).unwrap(); - Self { - keyring: directory.join("omikron.mk"), - public_key: directory.join("omikron.mpkb"), - directory, - } - } - } - - impl Drop for TestPaths { - fn drop(&mut self) { - let _ = fs::remove_dir_all(&self.directory); - } - } - - #[test] - fn missing_identity_is_created_and_can_be_loaded_again() { - let paths = TestPaths::new(); - let passphrase = b"test identity secret"; - let first = load_or_create_keyring(passphrase, &paths.keyring, &paths.public_key).unwrap(); - let first_public = first.public_key_bundle().try_as_bytes().unwrap(); - let second = load_or_create_keyring(passphrase, &paths.keyring, &paths.public_key).unwrap(); - - assert_eq!( - second.public_key_bundle().try_as_bytes().unwrap(), - first_public - ); - } - - #[test] - fn corrupt_identity_fails_closed() { - let paths = TestPaths::new(); - fs::write(&paths.keyring, b"not a keyring").unwrap(); - - let error = - load_or_create_keyring(b"test identity secret", &paths.keyring, &paths.public_key) - .unwrap_err(); - assert!(error.contains("unable to load existing Omikron identity")); - } - - #[test] - fn unreadable_identity_fails_closed() { - let paths = TestPaths::new(); - fs::create_dir(&paths.keyring).unwrap(); - - let error = - load_or_create_keyring(b"test identity secret", &paths.keyring, &paths.public_key) - .unwrap_err(); - assert!(error.contains("unable to load existing Omikron identity")); - } - - #[test] - fn public_key_persistence_failure_fails_closed() { - let paths = TestPaths::new(); - fs::create_dir(&paths.public_key).unwrap(); - - let error = - load_or_create_keyring(b"test identity secret", &paths.keyring, &paths.public_key) - .unwrap_err(); - assert!(error.contains("unable to persist Omikron public key bundle")); - } - - #[cfg(feature = "raw-migration")] - #[test] - fn raw_identity_migration_rewrites_protected_storage() { - use mtp::files::{load_keyring, save_keyring_raw}; - - let paths = TestPaths::new(); - let original = Keyring::generate(); - save_keyring_raw(&original, &paths.keyring).unwrap(); - - let migrated = - migrate_raw_keyring(b"test identity secret", &paths.keyring, &paths.public_key) - .unwrap(); - let loaded = load_keyring(&paths.keyring, b"test identity secret").unwrap(); - - assert_eq!( - migrated.public_key_bundle().try_as_bytes().unwrap(), - original.public_key_bundle().try_as_bytes().unwrap() - ); - assert_eq!( - loaded.public_key_bundle().try_as_bytes().unwrap(), - original.public_key_bundle().try_as_bytes().unwrap() - ); - } -} diff --git a/src/main.rs b/src/main.rs index 34f54eb..760e37c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,7 +3,6 @@ mod app_state; mod calls; mod config; mod data; -mod identity; mod omega; mod rho; mod services; @@ -19,20 +18,35 @@ pub static WORKING_DIR: Lazy = use rustls::crypto::aws_lc_rs::default_provider; use mtp::crypto::Keyring; +use mtp::files::{load_keyring_raw, save_keyring_raw, save_public_key_bundle}; use crate::{ app_state::AppState, calls::{call_manager::CallManager, call_util::LiveKitService}, config::Config, - identity::{ - KEYRING_PATH, PUBLIC_KEY_PATH, identity_secret_from_environment, load_or_create_keyring, - }, omega::omega_connection::{OmegaConnection, start_task_cleanup_loop}, rho::rho_manager::RhoManager, rho::server::start, util::logger::startup, }; +const KEYRING_PATH: &str = "./omikron.mk"; +const PUBLIC_KEY_PATH: &str = "./omikron.mpkb"; + +fn load_keyring() -> Result { + match load_keyring_raw(KEYRING_PATH) { + Ok(keyring) => Ok(keyring), + Err(_) => { + let kr = Keyring::generate(); + save_keyring_raw(&kr, KEYRING_PATH).map_err(|error| error.to_string())?; + save_public_key_bundle(&kr.public_key_bundle(), PUBLIC_KEY_PATH) + .map_err(|error| error.to_string())?; + eprintln!("Generated new keyring at {}", KEYRING_PATH); + Ok(kr) + } + } +} + #[tokio::main] async fn main() { if let Err(_) = default_provider().install_default() { @@ -50,22 +64,13 @@ async fn main() { } }; - let identity_secret = match identity_secret_from_environment() { - Ok(secret) => secret, + let keyring = match load_keyring() { + Ok(keyring) => keyring, Err(error) => { - eprintln!("Unable to load Omikron identity secret: {error}"); + eprintln!("Unable to load keyring: {error}"); return; } }; - let keyring = - match load_or_create_keyring(identity_secret.as_slice(), KEYRING_PATH, PUBLIC_KEY_PATH) { - Ok(keyring) => keyring, - Err(error) => { - eprintln!("Unable to load Omikron identity: {error}"); - return; - } - }; - drop(identity_secret); let omega_keyring = match keyring_for_omega(&keyring) { Ok(keyring) => keyring, @@ -97,6 +102,5 @@ async fn main() { } fn keyring_for_omega(keyring: &Keyring) -> Result { - let bytes = keyring.try_to_bytes().map_err(|error| error.to_string())?; - Keyring::from_bytes(&bytes).map_err(|error| error.to_string()) + Keyring::from_bytes(&keyring.to_bytes()).map_err(|error| error.to_string()) } diff --git a/src/omega/capabilities.rs b/src/omega/capabilities.rs index cff09a8..b658068 100644 --- a/src/omega/capabilities.rs +++ b/src/omega/capabilities.rs @@ -42,7 +42,6 @@ impl PeerCapabilities { format!("{OMIKRON_PREFIX}{}", names.join(",")) } - #[allow(dead_code)] pub fn from_identification_description(description: Option<&str>) -> Result { parse_capabilities(description, OMIKRON_PREFIX) } diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index 3e7755c..f8d505f 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -2,7 +2,7 @@ use super::capabilities::PeerCapabilities; use crate::{ config::Config, log_cv_in, log_cv_out, log_err, log_in, log_out, - rho::connection::{OptionalDataValueCompat, RequiredMtpFields}, + rho::connection::{MtpValueCompat, OptionalDataValueCompat}, rho::relay_router, rho::rho_manager::RhoManager, util::{data_type_id, logger::PrintType}, @@ -16,7 +16,6 @@ use mtp::{ host::{Policy, SendMode}, }; use mtp_transport::ConnectionHandle; -use rand::RngExt; use std::{sync::Arc, time::Duration}; use tokio::{ sync::{Mutex, RwLock, mpsc, oneshot, watch}, @@ -28,7 +27,6 @@ use uuid::Uuid; const RECONNECT_DELAY: Duration = Duration::from_secs(5); const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(300); -const RECONNECT_JITTER: Duration = Duration::from_secs(1); const CONNECTION_TIMEOUT: Duration = Duration::from_secs(10); const CAPABILITY_NEGOTIATION_TIMEOUT: Duration = Duration::from_secs(1); const PING_INTERVAL: Duration = Duration::from_secs(5); @@ -38,29 +36,13 @@ const MAX_CONCURRENT_REQUESTS: usize = 128; const CIRCUIT_BREAKER_FAILURE_THRESHOLD: u32 = 3; const CIRCUIT_BREAKER_COOLDOWN: Duration = Duration::from_secs(30); -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum ConnectionOutcome { - HealthySessionEnded, - FailedBeforeHealthy, -} - -fn reconnect_base_after_outcome(current: Duration, outcome: ConnectionOutcome) -> Duration { - match outcome { - ConnectionOutcome::HealthySessionEnded => RECONNECT_DELAY, - ConnectionOutcome::FailedBeforeHealthy => current, - } -} - -fn reconnect_delay_with_jitter(delay: Duration) -> Duration { - let max_jitter_ms = std::cmp::min(delay, RECONNECT_JITTER).as_millis() as u64; - delay + Duration::from_millis(rand::rng().random_range(0..=max_jitter_ms)) -} - fn client_changed_target(value: &CommunicationValue) -> Option<(i64, i64)> { if !value.is_type(CommunicationType::ClientChanged) { return None; } - let receiver = i64::try_from(value.receiver()?).ok().filter(|id| *id > 0)?; + let receiver = i64::try_from(value.get_receiver()) + .ok() + .filter(|id| *id > 0)?; let session_id = value .get_data(DataType::SessionId) .as_signed_number() @@ -302,45 +284,33 @@ impl OmegaConnection { } match self.clone().connect_once().await { - Ok(outcome) => { - reconnect_delay = reconnect_base_after_outcome(reconnect_delay, outcome); - match outcome { - ConnectionOutcome::HealthySessionEnded => { - log_err!( - 0, - PrintType::Omega, - "Healthy Omega connection ended; reconnecting" - ); - } - ConnectionOutcome::FailedBeforeHealthy => { - log_err!( - 0, - PrintType::Omega, - "Omega connection failed before synchronization" - ); - } + Ok(()) => { + // Connection closed gracefully, check if we should reconnect + if *self.reconnect_on_close.read().await { + log_err!( + 0, + PrintType::Omega, + "Connection lost, reconnecting in {:?}...", + reconnect_delay + ); + } else { + log_in!(0, PrintType::Omega, "Connection closed, not reconnecting"); + break; } } - Err(error) => { - log_err!(0, PrintType::Omega, "Connection failed: {}", error); + Err(e) => { + log_err!( + 0, + PrintType::Omega, + "Connection failed: {}, retrying in {:?}...", + e, + reconnect_delay + ); } } - if !*self.reconnect_on_close.read().await { - log_in!(0, PrintType::Omega, "Connection closed, not reconnecting"); - break; - } - - let retry_delay = reconnect_delay_with_jitter(reconnect_delay); - log_err!( - 0, - PrintType::Omega, - "Retrying Omega connection in {:?}...", - retry_delay - ); - tokio::select! { - _ = sleep(retry_delay) => {} + _ = sleep(reconnect_delay) => {} _ = shutdown_rx.changed() => { if *shutdown_rx.borrow() { break; @@ -352,7 +322,7 @@ impl OmegaConnection { } } - async fn connect_once(self: Arc) -> Result { + async fn connect_once(self: Arc) -> Result<(), String> { *self.state.write().await = ConnectionState::Connecting; let client_config = ClientConfig::new(format!("https://{}:{}", self.host, self.port)) @@ -361,6 +331,7 @@ impl OmegaConnection { .with_policy( Policy::default() .with_send_mode(SendMode::SingleStreamPerMessage) + .with_max_message_size(1_000_000_000) .with_timeouts( Duration::from_millis(5_000), Duration::from_millis(5_000), @@ -372,7 +343,8 @@ impl OmegaConnection { .with_max_concurrent_stream_tasks(64) .with_persistent_stream_retries(5, Duration::from_secs(5)), ) - .with_ping_interval(PING_INTERVAL); + .with_ping_interval(PING_INTERVAL) + .with_max_missed_pings(0); let host_public_key = load_public_key_bundle("./omega.mpkb") .map_err(|e| format!("Failed to load omega.mpkb: {}", e))?; @@ -417,7 +389,7 @@ impl OmegaConnection { self.fail_synchronization("capability negotiation", error) .await; let _ = read_handle.await; - return Ok(ConnectionOutcome::FailedBeforeHealthy); + return Err("Omega capability negotiation failed".to_string()); } Ok(Err(_)) | Err(_) => { // An Omega from before capability negotiation sends no second @@ -431,7 +403,7 @@ impl OmegaConnection { // Tell omega our current state now that we're actually connected - // doing this after teardown (as before) sent into a sender that had // already been cleared, silently dropping the sync every time. - let reached_healthy_state = self.clone().sync_client_iota_status().await; + self.clone().sync_client_iota_status().await; // Wait for read loop to complete (connection closed) let result = read_handle.await; @@ -456,13 +428,19 @@ impl OmegaConnection { *self.state.write().await = ConnectionState::Disconnected; match result { - Ok(()) if reached_healthy_state => Ok(ConnectionOutcome::HealthySessionEnded), - Ok(()) => Ok(ConnectionOutcome::FailedBeforeHealthy), + Ok(()) => { + // Check if we should reconnect + if *self.reconnect_on_close.read().await { + Err("Connection closed, will reconnect".to_string()) + } else { + Ok(()) + } + } Err(e) => Err(format!("Read loop error: {}", e)), } } - async fn sync_client_iota_status(&self) -> bool { + async fn sync_client_iota_status(&self) { *self.state.write().await = ConnectionState::SynchronizingRoutes; let mut connected_iota_ids: Vec = Vec::new(); let mut connected_sessions: Vec = Vec::new(); @@ -519,16 +497,16 @@ impl OmegaConnection { if let Err(error) = self.send_message_result(&sync_msg).await { self.fail_synchronization("legacy route synchronization", error) .await; - return false; + return; } *self.state.write().await = ConnectionState::SynchronizingSubscriptions; if let Err(error) = self.restore_presence_subscriptions().await { self.fail_synchronization("subscription restoration", error) .await; - return false; + return; } *self.state.write().await = ConnectionState::Ready; - return true; + return; } let mut response = Err("route synchronization did not start".to_string()); @@ -557,7 +535,7 @@ impl OmegaConnection { ); self.fail_synchronization("subscription restoration", error) .await; - return false; + return; } *self.state.write().await = ConnectionState::Ready; } @@ -568,15 +546,14 @@ impl OmegaConnection { ); self.fail_synchronization("route synchronization", error) .await; - return false; + return; } Err(error) => { self.fail_synchronization("route synchronization", error) .await; - return false; + return; } } - true } async fn fail_synchronization(&self, phase: &str, error: String) { @@ -623,39 +600,8 @@ impl OmegaConnection { } if cv.is_type(CommunicationType::Relay) { - let request_id = match cv.require_id() { - Ok(request_id) => request_id, - Err(error) => { - log_err!( - self.omikron_id as i64, - PrintType::Omega, - "Rejected malformed relay: {}", - error - ); - let response = - CommunicationValue::new(CommunicationType::ErrorInvalidData) - .without_id(); - let _ = self.send_message_result(&response).await; - continue; - } - }; - let destination_iota = match cv.require_receiver() { - Ok(destination_iota) => destination_iota, - Err(error) => { - log_err!( - self.omikron_id as i64, - PrintType::Omega, - "Rejected malformed relay: {}", - error - ); - let response = CommunicationValue::new( - CommunicationType::ErrorInvalidData, - ) - .with_id(request_id); - let _ = self.send_message_result(&response).await; - continue; - } - }; + let destination_iota = cv.receiver().unwrap_or_default(); + let request_id = cv.get_id(); let response = match relay_router::route_from_omega(&self.rho, cv).await { Ok(()) => CommunicationValue::new(CommunicationType::Success) .with_id(request_id), @@ -684,18 +630,7 @@ impl OmegaConnection { continue; } - let msg_id = match cv.require_id() { - Ok(msg_id) => msg_id, - Err(error) => { - log_err!( - self.omikron_id as i64, - PrintType::Omega, - "Discarded response without an MTP id: {}", - error - ); - continue; - } - }; + let msg_id = cv.get_id(); if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) { if (task.task)(self.clone(), cv.clone()) { continue; @@ -814,7 +749,6 @@ impl OmegaConnection { self.try_send_message(cv).await } - #[allow(dead_code)] pub async fn supports_set_user_state(&self) -> bool { self.peer_capabilities.read().await.set_user_state_v1 } @@ -873,12 +807,10 @@ impl OmegaConnection { cv: &CommunicationValue, timeout_duration: Option, ) -> Result { - let msg_id = cv - .require_id() - .map_err(|error| format!("request is missing correlation id: {error}"))?; self.await_connection(timeout_duration).await?; let (tx, mut rx) = mpsc::channel(1); + let msg_id = cv.get_id(); self.waiting_tasks.insert( msg_id, @@ -887,9 +819,9 @@ impl OmegaConnection { log_in!( 0, PrintType::Omega, - "Matched Omega response (request_id={}, response_id={:?}, type={})", + "Matched Omega response (request_id={}, response_id={}, type={})", msg_id, - response_cv.id(), + response_cv.get_id(), response_cv .get_comm_type_enum() .map(|kind| kind.to_string()) @@ -948,7 +880,7 @@ impl OmegaConnection { pub async fn close_iota(&self, iota_id: i64) -> Result<(), String> { let cv = CommunicationValue::new(CommunicationType::IotaDisconnected) .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); - let request_id = cv.require_id().map_err(|error| error.to_string())?; + let request_id = cv.get_id(); let result = self.lifecycle_request(&cv).await; if let Err(error) = &result { self.log_lifecycle_failure("IotaDisconnected", iota_id, None, request_id, error); @@ -965,7 +897,7 @@ impl OmegaConnection { DataType::SessionId, DataValue::SignedNumber(session_id.into()), ); - let request_id = cv.require_id().map_err(|error| error.to_string())?; + let request_id = cv.get_id(); let result = self.lifecycle_request(&cv).await; if let Err(error) = &result { self.log_lifecycle_failure( @@ -1006,7 +938,7 @@ impl OmegaConnection { .as_millis() as i128, ), ); - let request_id = request.require_id().map_err(|error| error.to_string())?; + let request_id = request.get_id(); let result = self.lifecycle_request(&request).await; if let Err(error) = &result { self.log_lifecycle_failure( @@ -1024,7 +956,7 @@ impl OmegaConnection { pub async fn iota_connected(&self, iota_id: i64) -> Result<(), String> { let request = CommunicationValue::new(CommunicationType::IotaConnected) .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); - let request_id = request.require_id().map_err(|error| error.to_string())?; + let request_id = request.get_id(); let result = self.lifecycle_request(&request).await; if let Err(error) = &result { self.log_lifecycle_failure("IotaConnected", iota_id, None, request_id, error); @@ -1033,7 +965,6 @@ impl OmegaConnection { result } - #[allow(dead_code)] pub async fn reconcile_routes(&self) { self.sync_client_iota_status().await; } @@ -1243,11 +1174,7 @@ impl OmegaConnection { #[cfg(test)] mod tests { - use super::{ - ConnectionOutcome, MAX_RECONNECT_DELAY, PeerCapabilities, RECONNECT_DELAY, - client_changed_target, parse_omega_capability_response, reconnect_base_after_outcome, - reconnect_delay_with_jitter, - }; + use super::{PeerCapabilities, client_changed_target, parse_omega_capability_response}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; fn notification() -> CommunicationValue { @@ -1269,9 +1196,6 @@ mod tests { let mut missing_state = notification(); missing_state.remove_data(DataType::UserState); assert_eq!(client_changed_target(&missing_state), None); - - let missing_receiver = notification().without_receiver(); - assert_eq!(client_changed_target(&missing_receiver), None); } #[test] @@ -1284,35 +1208,6 @@ mod tests { assert_eq!(client_changed_target(&invalid), None); } - #[test] - fn reconnect_jitter_is_bounded_without_changing_the_base_delay() { - for _ in 0..32 { - let delay = reconnect_delay_with_jitter(RECONNECT_DELAY); - assert!(delay >= RECONNECT_DELAY); - assert!(delay <= RECONNECT_DELAY + std::time::Duration::from_secs(1)); - } - - assert!(MAX_RECONNECT_DELAY > RECONNECT_DELAY); - } - - #[test] - fn healthy_session_resets_accumulated_reconnect_backoff() { - let mut delay = RECONNECT_DELAY; - for _ in 0..4 { - delay = std::cmp::min(delay * 2, MAX_RECONNECT_DELAY); - } - assert!(delay > RECONNECT_DELAY); - - assert_eq!( - reconnect_base_after_outcome(delay, ConnectionOutcome::FailedBeforeHealthy), - delay - ); - assert_eq!( - reconnect_base_after_outcome(delay, ConnectionOutcome::HealthySessionEnded), - RECONNECT_DELAY - ); - } - #[test] fn capability_response_negotiates_new_omega() { let response = CommunicationValue::new(CommunicationType::IdentificationResponse) diff --git a/src/rho/app_connection.rs b/src/rho/app_connection.rs index c6af5e9..7e8778a 100644 --- a/src/rho/app_connection.rs +++ b/src/rho/app_connection.rs @@ -1,7 +1,7 @@ use crate::anonymous_clients::anonymous_manager; use crate::app_state::AppState; use crate::rho::connection::{ - GeneralConnection, MtpReceiver, MtpSender, OptionalDataValueCompat, RequiredMtpFields, + GeneralConnection, MtpReceiver, MtpSender, MtpValueCompat, OptionalDataValueCompat, }; use crate::rho::relay_router::{self, RelaySource}; use crate::rho::rho_connection::RhoConnection; @@ -89,38 +89,12 @@ impl AppConnection { /// Handle incoming message from app pub async fn handle_message(self: Arc, cv: CommunicationValue) { tokio::spawn(async move { - let message_id = match cv.require_id() { - Ok(message_id) => message_id, - Err(error) => { - log_err!( - self.user_id as i64, - PrintType::App, - "Rejected malformed message: {}", - error - ); - let response = - CommunicationValue::new(CommunicationType::ErrorInvalidData).without_id(); - self.send_message(&response).await; - return; - } - }; log_cv_in!(PrintType::App, cv); if cv.is_type(CommunicationType::Relay) { - let next_hop = match cv.require_receiver() { - Ok(next_hop) => next_hop, - Err(error) => { - log_err!( - self.user_id as i64, - PrintType::App, - "Rejected malformed relay: {}", - error - ); - self.send_error_response(message_id, CommunicationType::ErrorInvalidData) - .await; - return; - } - }; + let cv = relay_router::ensure_relay_frame_id(cv); + let request_id = cv.get_id(); + let next_hop = cv.receiver().unwrap_or_default(); let result = match self.get_rho_connection().await { Some(rho) => { relay_router::route_relay( @@ -128,7 +102,7 @@ impl AppConnection { RelaySource::Client { iota_id: rho.get_iota_id().await, }, - relay_router::ensure_relay_frame_id(cv), + cv, ) .await } @@ -136,7 +110,7 @@ impl AppConnection { }; let response = match result { Ok(()) => { - CommunicationValue::new(CommunicationType::Success).with_id(message_id) + CommunicationValue::new(CommunicationType::Success).with_id(request_id) } Err(error) => { log_err!( @@ -147,7 +121,7 @@ impl AppConnection { error ); CommunicationValue::new(relay_router::error_response_type(&error)) - .with_id(message_id) + .with_id(request_id) } }; self.send_message(&response).await; @@ -156,7 +130,7 @@ impl AppConnection { if cv.is_type(CommunicationType::Success) { if let Some(rho) = self.get_rho_connection().await { - rho.forward_relay_ack(self.user_id, message_id).await; + rho.forward_relay_ack(self.user_id, cv.get_id()).await; } return; } @@ -165,7 +139,7 @@ impl AppConnection { relay_router::message_security_class(&cv), relay_router::MessageSecurityClass::RelayOnly ) { - self.send_error_response(message_id, CommunicationType::ErrorInvalidData) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData) .await; return; } @@ -181,7 +155,7 @@ impl AppConnection { } } { let response = CommunicationValue::new(CommunicationType::GetUserData) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default( DataType::Username, DataValue::Str(anonymous.get_user_name().await), @@ -223,7 +197,7 @@ impl AppConnection { "Rejected unsupported communication type {}", cv.get_type() ); - self.send_error_response(message_id, CommunicationType::ErrorInvalidData) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData) .await; }); } diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 5a25ec0..ba83728 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -3,7 +3,7 @@ use crate::app_state::AppState; use crate::calls::call_group::call_invite_secret_from_cv; use crate::data::user::UserStatus; use crate::rho::connection::{ - GeneralConnection, MtpReceiver, MtpSender, OptionalDataValueCompat, RequiredMtpFields, + GeneralConnection, MtpReceiver, MtpSender, MtpValueCompat, OptionalDataValueCompat, }; use crate::rho::relay_router::{self, RelaySource}; use crate::rho::rho_connection::RhoConnection; @@ -117,41 +117,14 @@ impl ClientConnection { }; tokio::spawn(async move { let _permit = permit; - let message_id = match cv.require_id() { - Ok(message_id) => message_id, - Err(error) => { - log_err!( - self.user_id as i64, - PrintType::Client, - "Rejected malformed message: {}", - error - ); - let response = - CommunicationValue::new(CommunicationType::ErrorInvalidData).without_id(); - self.send_message(&response).await; - return; - } - }; log_cv_in!(PrintType::Client, cv); let mut cv = cv; if cv.is_type(CommunicationType::Relay) { - let next_hop = match cv.require_receiver() { - Ok(next_hop) => next_hop, - Err(error) => { - log_err!( - self.user_id as i64, - PrintType::Client, - "Rejected malformed relay: {}", - error - ); - self.send_error_response(message_id, CommunicationType::ErrorInvalidData) - .await; - return; - } - }; cv = relay_router::ensure_relay_frame_id(cv); + let request_id = cv.get_id(); + let next_hop = cv.receiver().unwrap_or_default(); let result = match self.get_rho_connection().await { Some(rho) => { relay_router::route_relay( @@ -167,7 +140,7 @@ impl ClientConnection { }; let response = match result { Ok(()) => { - CommunicationValue::new(CommunicationType::Success).with_id(message_id) + CommunicationValue::new(CommunicationType::Success).with_id(request_id) } Err(error) => { log_err!( @@ -178,7 +151,7 @@ impl ClientConnection { error ); CommunicationValue::new(relay_router::error_response_type(&error)) - .with_id(message_id) + .with_id(request_id) } }; self.send_message(&response).await; @@ -187,7 +160,7 @@ impl ClientConnection { if cv.is_type(CommunicationType::Success) { if let Some(rho) = self.get_rho_connection().await { - rho.forward_relay_ack(self.user_id, message_id).await; + rho.forward_relay_ack(self.user_id, cv.get_id()).await; } return; } @@ -196,7 +169,7 @@ impl ClientConnection { relay_router::message_security_class(&cv), relay_router::MessageSecurityClass::RelayOnly ) { - self.send_error_response(message_id, CommunicationType::ErrorInvalidData) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData) .await; return; } @@ -205,11 +178,11 @@ impl ClientConnection { // user fields, if present, are deliberately ignored: an // authenticated connection may only change its own state. if cv.is_type(CommunicationType::ClientChanged) - && cv.get_data(DataType::UserState).is_some() + && cv.get_data_opt(DataType::UserState).is_some() { self.handle_set_user_state( CommunicationValue::new(CommunicationType::ClientChanged) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default( DataType::UserState, cv.get_data(DataType::UserState) @@ -267,7 +240,7 @@ impl ClientConnection { } } { let response = CommunicationValue::new(CommunicationType::GetUserData) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default( DataType::Username, DataValue::Str(anonymous.get_user_name().await), @@ -303,19 +276,19 @@ impl ClientConnection { || cv.is_type(CommunicationType::DeleteUser) { if cv.is_type(CommunicationType::ChangeUserData) - && cv.get_data(DataType::OnlineStatus).is_some() + && cv.get_data_opt(DataType::OnlineStatus).is_some() { let mut profile_request = cv.clone(); let preference = profile_request .remove_data(DataType::OnlineStatus) .unwrap_or(DataValue::Null); let state_request = CommunicationValue::new(CommunicationType::ClientChanged) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::UserState, preference); let state_response = match self.request_set_user_state(state_request).await { Ok(response) => response, Err(error_type) => { - self.send_error_response(message_id, error_type).await; + self.send_error_response(cv.get_id(), error_type).await; return; } }; @@ -338,7 +311,7 @@ impl ClientConnection { } Ok(response) => self.send_message(&response).await, Err(_) => { - self.send_error_response(message_id, CommunicationType::ErrorInternal) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInternal) .await; } } @@ -363,7 +336,7 @@ impl ClientConnection { if is_per_device_settings { let Some(session_id) = session_id else { let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(message_id) + .with_id(cv.get_id()) .with_receiver(self.user_id) .add_typed_default( DataType::Message, @@ -379,7 +352,7 @@ impl ClientConnection { if session_id != expected_session_id { let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(message_id) + .with_id(cv.get_id()) .with_receiver(self.user_id) .add_typed_default( DataType::Message, @@ -395,7 +368,7 @@ impl ClientConnection { } else if let Some(session_id) = session_id { if session_id != expected_session_id { let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(message_id) + .with_id(cv.get_id()) .with_receiver(self.user_id) .add_typed_default( DataType::Message, @@ -423,7 +396,7 @@ impl ClientConnection { if let Some(session_id) = cv.get_data(DataType::SessionId).as_signed_number() { if session_id != expected_session_id { let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default( DataType::SessionId, DataValue::SignedNumber(expected_session_id), @@ -444,14 +417,12 @@ impl ClientConnection { "Rejected unsupported communication type {}", cv.get_type() ); - self.send_error_response(message_id, CommunicationType::ErrorInvalidData) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData) .await; }); } async fn handle_omega_forward(self: Arc, cv: CommunicationValue) { - let Ok(request_id) = cv.require_id() else { - return; - }; + let request_id = cv.get_id(); match self.await_omega_response(cv).await { Ok(response_cv) => self.send_message(&response_cv).await, Err(_) => { @@ -474,9 +445,6 @@ impl ClientConnection { &self, cv: CommunicationValue, ) -> Result { - let message_id = cv - .require_id() - .map_err(|_| CommunicationType::ErrorInvalidData)?; if !self.state.omega.is_ready().await { return Err(CommunicationType::ErrorInternal); } @@ -491,7 +459,7 @@ impl ClientConnection { return Err(CommunicationType::ErrorNoIota); }; let request = CommunicationValue::new(CommunicationType::ClientChanged) - .with_id(message_id) + .with_id(cv.get_id()) .with_sender(self.user_id) .add_typed_default( DataType::UserId, @@ -504,28 +472,22 @@ impl ClientConnection { .await .map_err(|_| CommunicationType::ErrorInternal)?; return Ok(CommunicationValue::new(CommunicationType::Success) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::UserState, DataValue::Str(state.to_string()))); } async fn handle_set_user_state(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; match self.request_set_user_state(cv.clone()).await { Ok(response) => self.send_message(&response).await, - Err(error_type) => self.send_error_response(message_id, error_type).await, + Err(error_type) => self.send_error_response(cv.get_id(), error_type).await, } } /// Handle call invite async fn handle_call_invite(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let receiver_id: i128 = cv.get_data(DataType::ReceiverId).as_number().unwrap_or(0); if receiver_id == 0 { - self.send_error_response(message_id, CommunicationType::ErrorNoUserId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorNoUserId) .await; return; } @@ -534,13 +496,13 @@ impl ClientConnection { Some(DataValue::Str(id_str)) => match Uuid::parse_str(id_str) { Ok(id) => id, Err(_) => { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; } }, _ => { - self.send_error_response(message_id, CommunicationType::ErrorNoCallId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorNoCallId) .await; return; } @@ -549,7 +511,7 @@ impl ClientConnection { let secret = match call_invite_secret_from_cv(&cv) { Some(secret) => secret, None => { - self.send_error_response(message_id, CommunicationType::BadRequest) + self.send_error_response(cv.get_id(), CommunicationType::BadRequest) .await; return; } @@ -560,7 +522,7 @@ impl ClientConnection { .add_invite(call_id, self.user_id, receiver_id as u64, secret.clone()) .await; if !invited { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; } @@ -570,7 +532,7 @@ impl ClientConnection { .call_manager .should_forward_invite(self.user_id, receiver_id as u64) { - let response = CommunicationValue::new(CommunicationType::Success).with_id(message_id); + let response = CommunicationValue::new(CommunicationType::Success).with_id(cv.get_id()); self.send_message(&response).await; return; } @@ -602,7 +564,7 @@ impl ClientConnection { }); let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default( DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into()), @@ -632,28 +594,25 @@ impl ClientConnection { target_rho.message_to_client(forward).await; - let response = CommunicationValue::new(CommunicationType::Success).with_id(message_id); + let response = CommunicationValue::new(CommunicationType::Success).with_id(cv.get_id()); self.send_message(&response).await; } /// Handle get call request async fn handle_get_call(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let user_id = self.get_user_id().await; let call_id = match cv.get_data(DataType::CallId) { Some(DataValue::Str(id_str)) => match Uuid::parse_str(id_str) { Ok(id) => id, Err(_) => { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; } }, _ => { - self.send_error_response(message_id, CommunicationType::ErrorNoCallId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorNoCallId) .await; return; } @@ -667,7 +626,7 @@ impl ClientConnection { { Ok(token) => { let response = CommunicationValue::new(CommunicationType::CallToken) - .with_id(message_id) + .with_id(cv.get_id()) .with_receiver(user_id as u64) .add_typed_default(DataType::CallToken, DataValue::Str(token)); self.send_message(&response).await; @@ -675,29 +634,26 @@ impl ClientConnection { Err(error) => { log::warn!("Unable to create call token for {}: {}", call_id, error); let error_cv = CommunicationValue::new(CommunicationType::ErrorNoCallId) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())); self.send_message(&error_cv).await; } } } async fn handle_get_call_data(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let user_id = self.get_user_id().await; let call_id = match cv.get_data(DataType::CallId) { Some(DataValue::Str(id_str)) => match Uuid::parse_str(id_str) { Ok(id) => id, Err(_) => { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; } }, _ => { - self.send_error_response(message_id, CommunicationType::ErrorNoCallId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorNoCallId) .await; return; } @@ -714,20 +670,20 @@ impl ClientConnection { } let response = CommunicationValue::new(CommunicationType::CallData) - .with_id(message_id) + .with_id(cv.get_id()) .with_receiver(user_id as u64) .add_typed_default(DataType::UserIds, DataValue::Array(user_ids)); self.send_message(&response).await; } else { let error_cv = CommunicationValue::new(CommunicationType::ErrorInvalidUserId) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())); self.send_message(&error_cv).await; return; } } else { let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())); self.send_message(&error_cv).await; return; @@ -735,12 +691,9 @@ impl ClientConnection { } async fn handle_call_timeout_user(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")) else { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; }; @@ -754,13 +707,13 @@ impl ClientConnection { .unwrap_or(0); let Some(call) = self.state.call_manager.get_call(call_id).await else { - self.send_error_response(message_id, CommunicationType::ErrorNotFound) + self.send_error_response(cv.get_id(), CommunicationType::ErrorNotFound) .await; return; }; let Some(caller) = call.get_caller(self.get_user_id().await).await else { - self.send_error_response(message_id, CommunicationType::ErrorInvalidUserId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidUserId) .await; return; }; @@ -776,12 +729,9 @@ impl ClientConnection { } } async fn handle_call_disconnect_user(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")) else { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; }; @@ -791,12 +741,12 @@ impl ClientConnection { .unwrap_or(0); let Some(call) = self.state.call_manager.get_call(call_id).await else { - self.send_error_response(message_id, CommunicationType::ErrorNotFound) + self.send_error_response(cv.get_id(), CommunicationType::ErrorNotFound) .await; return; }; let Some(caller) = call.get_caller(self.get_user_id().await).await else { - self.send_error_response(message_id, CommunicationType::ErrorInvalidUserId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidUserId) .await; return; }; @@ -805,12 +755,9 @@ impl ClientConnection { } } async fn handle_call_set_anonymous_joining(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")) else { - self.send_error_response(message_id, CommunicationType::ErrorInvalidCallId) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId) .await; return; }; @@ -833,7 +780,7 @@ impl ClientConnection { short_link = call.get_short_link().await; } let mut response_cv = CommunicationValue::new(CommunicationType::CallSetAnonymousJoining) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string())) .add_typed_default(DataType::Enabled, DataValue::Bool(enable)); if let Some(short_link) = short_link { @@ -843,9 +790,6 @@ impl ClientConnection { } async fn handle_load_txt_record(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; if let Some(path) = cv.get_data(DataType::Path).as_str() { let resolver = match TokioAsyncResolver::tokio_from_system_conf() { Ok(r) => r, @@ -855,7 +799,7 @@ impl ClientConnection { .cloned() .unwrap_or(DataValue::Null); let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::Path, path_data); self.send_message(&error_cv).await; return; @@ -874,7 +818,7 @@ impl ClientConnection { Ok(text) => text, Err(_) => { self.send_error_response( - message_id, + cv.get_id(), CommunicationType::ErrorInvalidData, ) .await; @@ -883,8 +827,8 @@ impl ClientConnection { }; let response = CommunicationValue::new(CommunicationType::LoadTxtRecord) - .with_id(message_id) - .add_typed_default(DataType::AppContent, DataValue::Str(record_text)); + .with_id(cv.get_id()) + .add_typed_default(DataType::Content, DataValue::Str(record_text)); self.send_message(&response).await; return; } @@ -894,7 +838,7 @@ impl ClientConnection { .cloned() .unwrap_or(DataValue::Null); let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::Path, path_data); self.send_message(&error_cv).await; } @@ -904,7 +848,7 @@ impl ClientConnection { .cloned() .unwrap_or(DataValue::Null); let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::Path, path_data); self.send_message(&error_cv).await; } @@ -918,7 +862,7 @@ impl ClientConnection { .cloned() .unwrap_or(DataValue::Null); let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) - .with_id(message_id) + .with_id(cv.get_id()) .add_typed_default(DataType::Path, path_data); self.send_message(&error_cv).await; } diff --git a/src/rho/connection.rs b/src/rho/connection.rs index fed5169..8109f36 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -16,38 +16,37 @@ use crate::{ use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use mtp::host::AuthState; use mtp::webserver::{WebMTPConnection, WebMtpReceiver, WebMtpSender}; -use thiserror::Error; pub type MtpSender = WebMtpSender; pub type MtpReceiver = WebMtpReceiver; -#[derive(Debug, Clone, Copy, Error, PartialEq, Eq)] -pub enum FrameValidationError { - #[error("message is missing an MTP id")] - MissingId, - #[error("message is missing an MTP sender")] - MissingSender, - #[error("message is missing an MTP receiver")] - MissingReceiver, +/* + * MTP 0.3 exposes absent frame fields and data entries as Options. These + * adapters keep legacy control handlers explicit while Relay code uses the + * native optional accessors directly. + */ +pub(crate) trait MtpValueCompat { + fn get_id(&self) -> u32; + fn get_sender(&self) -> u64; + fn get_receiver(&self) -> u64; + fn get_data_opt(&self, data_type: DataType) -> Option<&DataValue>; } -pub trait RequiredMtpFields { - fn require_id(&self) -> Result; - fn require_sender(&self) -> Result; - fn require_receiver(&self) -> Result; -} - -impl RequiredMtpFields for CommunicationValue { - fn require_id(&self) -> Result { - self.id().ok_or(FrameValidationError::MissingId) +impl MtpValueCompat for CommunicationValue { + fn get_id(&self) -> u32 { + self.id().unwrap_or_default() } - fn require_sender(&self) -> Result { - self.sender().ok_or(FrameValidationError::MissingSender) + fn get_sender(&self) -> u64 { + self.sender().unwrap_or_default() } - fn require_receiver(&self) -> Result { - self.receiver().ok_or(FrameValidationError::MissingReceiver) + fn get_receiver(&self) -> u64 { + self.receiver().unwrap_or_default() + } + + fn get_data_opt(&self, data_type: DataType) -> Option<&DataValue> { + self.get_data(data_type) } } @@ -56,7 +55,6 @@ pub(crate) trait OptionalDataValueCompat { fn as_number(&self) -> Option; fn as_signed_number(&self) -> Option; fn as_str(&self) -> Option<&str>; - #[allow(dead_code)] fn as_bytes(&self) -> Option>; } @@ -82,42 +80,6 @@ impl OptionalDataValueCompat for Option<&DataValue> { } } -#[cfg(test)] -mod tests { - use super::{FrameValidationError, RequiredMtpFields}; - use mtp::codec::{CommunicationType, CommunicationValue}; - - #[test] - fn required_fields_preserve_missing_field_errors() { - let frame = CommunicationValue::new(CommunicationType::Success) - .without_id() - .without_sender() - .without_receiver(); - - assert_eq!(frame.require_id(), Err(FrameValidationError::MissingId)); - assert_eq!( - frame.require_sender(), - Err(FrameValidationError::MissingSender) - ); - assert_eq!( - frame.require_receiver(), - Err(FrameValidationError::MissingReceiver) - ); - } - - #[test] - fn zero_is_a_present_routing_value() { - let frame = CommunicationValue::new(CommunicationType::Success) - .with_id(0) - .with_sender(0) - .with_receiver(0); - - assert_eq!(frame.require_id(), Ok(0)); - assert_eq!(frame.require_sender(), Ok(0)); - assert_eq!(frame.require_receiver(), Ok(0)); - } -} - /* * How a connection identified itself during the mtp handshake driven by * `server.rs` ("iota" / "client" authenticated logins, "anonymous" @@ -187,10 +149,6 @@ impl GeneralConnection { })) } - pub fn connection_kind(&self) -> ConnectionKind { - self.connection_kind - } - pub async fn handle(self: Arc) { log_in!(0, PrintType::General, "General connection handler started"); if self.migrate().await { diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index bd822cb..de4f02a 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -6,7 +6,7 @@ use crate::log_err; use crate::log_in; use crate::log_out; use crate::rho::connection::{ - GeneralConnection, MtpReceiver, MtpSender, OptionalDataValueCompat, RequiredMtpFields, + GeneralConnection, MtpReceiver, MtpSender, MtpValueCompat, OptionalDataValueCompat, }; use crate::rho::relay_router::{self, RelaySource}; use crate::util::data_type_id; @@ -33,9 +33,8 @@ fn contact_snapshot(value: &CommunicationValue) -> Option<(i64, i64, Vec)> return None; } - let user_id = value - .receiver() - .and_then(|id| i64::try_from(id).ok()) + let user_id = i64::try_from(value.get_receiver()) + .ok() .filter(|id| *id > 0)?; let session_id = value .get_data(DataType::SessionId) @@ -223,7 +222,6 @@ impl IotaConnection { } } - #[allow(dead_code)] pub async fn send_relay(&self, cv: &CommunicationValue) -> Result<(), String> { self.sender .send(cv) @@ -237,46 +235,20 @@ impl IotaConnection { return; }; let _permit = permit; - let message_id = match cv.require_id() { - Ok(message_id) => message_id, - Err(error) => { - log_err!( - self.iota_id as i64, - PrintType::Iota, - "Rejected malformed message: {}", - error - ); - let response = - CommunicationValue::new(CommunicationType::ErrorInvalidData).without_id(); - self.send_message(&response).await; - return; - } - }; if cv.is_type(CommunicationType::Relay) { - let next_hop = match cv.require_receiver() { - Ok(next_hop) => next_hop, - Err(error) => { - log_err!( - self.iota_id as i64, - PrintType::Iota, - "Rejected malformed relay: {}", - error - ); - self.send_error_response(message_id, CommunicationType::ErrorInvalidData, None) - .await; - return; - } - }; + let cv = relay_router::ensure_relay_frame_id(cv); + let request_id = cv.get_id(); + let next_hop = cv.receiver().unwrap_or_default(); let response = match relay_router::route_relay( &self.state, RelaySource::Iota { iota_id: self.iota_id, }, - relay_router::ensure_relay_frame_id(cv), + cv, ) .await { - Ok(()) => CommunicationValue::new(CommunicationType::Success).with_id(message_id), + Ok(()) => CommunicationValue::new(CommunicationType::Success).with_id(request_id), Err(error) => { log_err!( self.iota_id as i64, @@ -286,7 +258,7 @@ impl IotaConnection { error ); CommunicationValue::new(relay_router::error_response_type(&error)) - .with_id(message_id) + .with_id(request_id) } }; self.send_message(&response).await; @@ -297,12 +269,12 @@ impl IotaConnection { crate::rho::relay_router::message_security_class(&cv), crate::rho::relay_router::MessageSecurityClass::RelayOnly ) { - self.send_error_response(message_id, CommunicationType::ErrorInvalidData, None) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData, None) .await; return; } - let msg_id = message_id; + let msg_id = cv.get_id(); if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) { if (task)(self.clone(), cv.clone()) { return; @@ -337,7 +309,7 @@ impl IotaConnection { if cv.is_type(CommunicationType::StateSubscribe) { self.send_error_response( - message_id, + cv.get_id(), CommunicationType::ErrorInvalidData, Some("StateSubscribe must come from an authoritative contact snapshot"), ) @@ -363,7 +335,7 @@ impl IotaConnection { self.iota_id as i64, PrintType::Omega, "Forwarding CompleteRegisterUser to Omega (request_id={})", - message_id + request.get_id() ); let mut response_cv = self .state @@ -376,7 +348,7 @@ impl IotaConnection { self.iota_id as i64, PrintType::Omega, "CompleteRegisterUser request_id={} failed: {}; retrying once", - message_id, + request.get_id(), error ); response_cv = self @@ -391,9 +363,9 @@ impl IotaConnection { log_in!( self.iota_id as i64, PrintType::Omega, - "Omega completed registration (request_id={}, response_id={:?}, type={})", - message_id, - response_cv.id(), + "Omega completed registration (request_id={}, response_id={}, type={})", + request.get_id(), + response_cv.get_id(), response_cv .get_comm_type_enum() .map(|kind| kind.to_string()) @@ -436,7 +408,7 @@ impl IotaConnection { self.add_user_id(user_id as u64).await; self.send_message( &CommunicationValue::new(CommunicationType::Success) - .with_id(message_id), + .with_id(cv.get_id()), ) .await; return; @@ -444,15 +416,15 @@ impl IotaConnection { Ok(verified) => log_err!( self.iota_id as i64, PrintType::Omega, - "Registration verification returned an unexpected user (request_id={:?}, response_id={:?})", - verification.id(), - verified.id() + "Registration verification returned an unexpected user (request_id={}, response_id={})", + verification.get_id(), + verified.get_id() ), Err(verify_error) => log_err!( self.iota_id as i64, PrintType::Omega, - "Registration verification failed after request_id={:?}: {}", - verification.id(), + "Registration verification failed after request_id={}: {}", + verification.get_id(), verify_error ), } @@ -464,7 +436,7 @@ impl IotaConnection { error ); self.send_error_response( - message_id, + cv.get_id(), CommunicationType::ErrorInternal, Some(&format!("Omega forwarding failed: {error}")), ) @@ -505,7 +477,7 @@ impl IotaConnection { "Rejected unsupported communication type {}", cv.get_type() ); - self.send_error_response(message_id, CommunicationType::ErrorInvalidData, None) + self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidData, None) .await; } @@ -542,9 +514,6 @@ impl IotaConnection { } async fn handle_omega_forward_without_sender(self: Arc, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; let iota_for_closure = self.clone(); let request = cv.clone().add_typed_default( DataType::IotaId, @@ -561,7 +530,7 @@ impl IotaConnection { self.iota_id as i64, PrintType::Omega, "GetRegister request_id={} failed: {}; retrying once", - message_id, + request.get_id(), error ); response_cv = self @@ -581,7 +550,7 @@ impl IotaConnection { error ); self.send_error_response( - message_id, + cv.get_id(), CommunicationType::ErrorInternal, Some(&format!("Omega forwarding failed: {error}")), ) @@ -591,17 +560,7 @@ impl IotaConnection { } /// Handle GET_CHATS message async fn handle_get_chats(&self, cv: CommunicationValue) { - let Ok(message_id) = cv.require_id() else { - return; - }; - let Ok(user_id) = cv.require_sender() else { - log_err!( - self.iota_id as i64, - PrintType::Iota, - "Rejected get_chats without an MTP sender" - ); - return; - }; + let user_id = cv.get_sender(); // Authority check: user must be linked to this Iota if !self.get_user_ids().await.contains(&user_id) { @@ -623,7 +582,7 @@ impl IotaConnection { else { self.forward_to_client( CommunicationValue::new(CommunicationType::ErrorInvalidData) - .with_id(message_id) + .with_id(cv.get_id()) .with_receiver(user_id), ) .await; @@ -804,14 +763,7 @@ impl IotaConnection { async fn add_call_state(&self, response: CommunicationValue) -> CommunicationValue { let mut output = response.clone(); - let Some(user_id) = response.receiver() else { - log_err!( - self.iota_id as i64, - PrintType::Iota, - "Discarded response without an MTP receiver" - ); - return output; - }; + let user_id = response.get_receiver(); let typed_data: Vec<_> = response.iter_typed_data().collect(); for (key, value) in typed_data { @@ -855,9 +807,7 @@ impl IotaConnection { timeout_duration: Option, ) -> Result { let (tx, mut rx) = mpsc::channel(1); - let msg_id = cv - .require_id() - .map_err(|error| format!("request is missing correlation id: {error}"))?; + let msg_id = cv.get_id(); let task_tx = tx.clone(); self.waiting_tasks.insert( diff --git a/src/rho/relay_router.rs b/src/rho/relay_router.rs index bd0f0ab..0e0ffaa 100644 --- a/src/rho/relay_router.rs +++ b/src/rho/relay_router.rs @@ -26,7 +26,6 @@ pub enum RouteTarget { } impl RouteTarget { - #[allow(dead_code)] pub fn wire_id(self) -> Option { let (kind, id) = match self { Self::User(id) => (USER_TARGET_KIND, id), @@ -252,7 +251,7 @@ fn route_response(response: CommunicationValue) -> Result<(), RelayRouteError> { } pub fn ensure_relay_frame_id(frame: CommunicationValue) -> CommunicationValue { - if frame.id().is_some() { + if frame.id().is_some_and(|id| id != 0) { return frame; } let id = NEXT_RELAY_FRAME_ID.fetch_add(1, Ordering::Relaxed).max(1); diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index f4aa658..094bb1b 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -3,7 +3,10 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection use super::relay_router::RouteTarget; use crate::{ log_err, - rho::{app_connection::AppConnection, connection::OptionalDataValueCompat}, + rho::{ + app_connection::AppConnection, + connection::{MtpValueCompat, OptionalDataValueCompat}, + }, }; use dashmap::DashMap; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; @@ -254,14 +257,7 @@ impl RhoConnection { /// Send message from Iota to specific client pub async fn message_to_client(&self, cv: CommunicationValue) { let connections = self.get_client_connections().await; - let Some(receiver_id) = cv.receiver() else { - log_err!( - 0, - crate::util::logger::PrintType::General, - "Discarded message without an MTP receiver" - ); - return; - }; + let receiver_id = cv.get_receiver(); let session_id = cv.get_data(DataType::SessionId).as_number(); for connection in connections.iter() { @@ -277,7 +273,6 @@ impl RhoConnection { } } - #[allow(dead_code)] pub async fn message_to_iota(&self, cv: CommunicationValue) { self.iota_connection.send_message(&cv).await; } @@ -313,7 +308,6 @@ impl RhoConnection { send_error.map_or(Ok(()), Err) } - #[allow(dead_code)] pub async fn send_relay_to_iota(&self, cv: &CommunicationValue) -> Result<(), String> { self.iota_connection.send_relay(cv).await } diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index 18003bf..2eb4632 100644 --- a/src/rho/rho_manager.rs +++ b/src/rho/rho_manager.rs @@ -24,7 +24,6 @@ impl RhoManager { self.users.get(&user_id).map(|entry| entry.value().clone()) } - #[allow(dead_code)] pub async fn contains_iota(&self, iota_id: i64) -> bool { self.connections.contains_key(&iota_id) } diff --git a/src/rho/server.rs b/src/rho/server.rs index 445e27c..5f158e9 100644 --- a/src/rho/server.rs +++ b/src/rho/server.rs @@ -9,7 +9,7 @@ use crate::{ app_state::AppState, log, log_err, omega::omega_connection::OmegaConnection, - rho::connection::{ConnectionKind, GeneralConnection, OptionalDataValueCompat}, + rho::connection::{GeneralConnection, OptionalDataValueCompat}, util::{file_util::load_file_vec, logger::PrintType}, }; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; @@ -17,27 +17,6 @@ use mtp::crypto::PublicKeyBundle; use mtp::host::{AuthenticationPolicy, HostConfig, Policy, SendMode}; use mtp::webserver::{MTPWebServer, WebServerConfig}; -fn web_config(max_connections: usize) -> Result { - WebServerConfig::new() - .max_connections(max_connections) - .route("/", |_request, response| async move { response.body("OK") }) -} - -fn rho_policy() -> Policy { - Policy::default() - .with_send_mode(SendMode::SingleStreamPerMessage) - .with_timeouts( - Duration::from_millis(2_000), - Duration::from_millis(2_000), - Duration::from_millis(30_000), - ) - .with_keep_alive(Some(Duration::from_secs(6))) - .with_max_idle_timeout(Some(Duration::from_secs(30))) - .with_receiver_queue_capacity(1000) - .with_max_concurrent_stream_tasks(10) - .with_persistent_stream_retries(5, Duration::from_secs(5)) -} - /* * Resolves the PublicKeyBundle mtp needs to verify a login's signed * challenge response. "iota"/"client" ids are looked up through Omega, the @@ -96,23 +75,10 @@ pub async fn complete_register( } println!("Iota connection request"); - let pub_key_bytes = match pub_key.try_as_bytes() { - Ok(bytes) => bytes, - Err(error) => { - log_err!( - 0, - PrintType::General, - "Failed to serialize Iota public key: {}", - error - ); - return 0; - } - }; - let request = CommunicationValue::new(CommunicationType::CompleteRegisterIota) .add_typed_default( DataType::PublicKey, - DataValue::Str(BASE64_STD.encode(pub_key_bytes)), + DataValue::Str(BASE64_STD.encode(pub_key.as_bytes())), ); let response = match omega @@ -148,7 +114,21 @@ pub async fn start(state: Arc) -> Result<(), Box) -> Result<(), Box) -> Result<(), Box permit, - Err(_) => { - log_err!( - 0, - PrintType::General, - "Rejected connection: global Rho connection limit reached" - ); - continue; - } - }; - let peer_ip = conn.remote_addr.map(|address| address.ip()); let state = state.clone(); tokio::spawn(async move { - let Some(conn) = GeneralConnection::new(conn, state.clone()) else { + let Some(conn) = GeneralConnection::new(conn, state) else { log_err!( 0, PrintType::General, @@ -217,58 +186,6 @@ pub async fn start(state: Arc) -> Result<(), Box Some(permit), - Err(_) => { - log_err!( - 0, - PrintType::General, - "Rejected anonymous connection: anonymous limit reached" - ); - return; - } - } - } else { - None - }; - - let anonymous_ip_permit = if conn.connection_kind() == ConnectionKind::AnonymousClient { - let Some(peer_ip) = peer_ip else { - log_err!( - 0, - PrintType::General, - "Rejected anonymous connection: peer address unavailable" - ); - return; - }; - match state - .rho_connection_limits - .try_acquire_anonymous_per_ip(peer_ip) - { - Some(permit) => Some(permit), - None => { - log_err!( - 0, - PrintType::General, - "Rejected anonymous connection: per-IP limit reached" - ); - return; - } - } - } else { - None - }; - - let _global_permit = global_permit; - let _anonymous_permit = anonymous_permit; - let _anonymous_ip_permit = anonymous_ip_permit; conn.handle().await; }); } @@ -276,22 +193,3 @@ pub async fn start(state: Arc) -> Result<(), Box, } -#[allow(dead_code)] impl CallService { pub fn new(state: Arc) -> Self { Self { state } diff --git a/src/services/routing_service.rs b/src/services/routing_service.rs index 5410644..86f034f 100644 --- a/src/services/routing_service.rs +++ b/src/services/routing_service.rs @@ -4,12 +4,10 @@ use mtp::codec::CommunicationValue; use crate::{app_state::AppState, rho::rho_connection::RhoConnection}; -#[allow(dead_code)] pub struct RoutingService { state: Arc, } -#[allow(dead_code)] impl RoutingService { pub fn new(state: Arc) -> Self { Self { state } diff --git a/src/services/user_service.rs b/src/services/user_service.rs index 12e6e59..efc72a5 100644 --- a/src/services/user_service.rs +++ b/src/services/user_service.rs @@ -2,12 +2,10 @@ use std::sync::Arc; use crate::{app_state::AppState, rho::rho_connection::RhoConnection}; -#[allow(dead_code)] pub struct UserService { state: Arc, } -#[allow(dead_code)] impl UserService { pub fn new(state: Arc) -> Self { Self { state } diff --git a/src/util/logger.rs b/src/util/logger.rs index 7968ef6..6e688cb 100644 --- a/src/util/logger.rs +++ b/src/util/logger.rs @@ -9,6 +9,8 @@ use std::{ use ansi_term::Color; use mtp::codec::{CommunicationType, CommunicationValue, DataTypeId, DataValue, Version}; +use crate::rho::connection::MtpValueCompat; + static LOGGER: OnceLock> = OnceLock::new(); #[allow(dead_code)] @@ -151,9 +153,7 @@ pub fn log_cv_internal( let formatted = format_cv(cv); log_internal( - cv.sender() - .and_then(|sender| i64::try_from(sender).ok()) - .unwrap_or(0), + cv.get_sender() as i64, print_type.unwrap_or(PrintType::General), prefix, false, @@ -164,14 +164,14 @@ pub fn log_cv_internal( pub fn format_cv(cv: &CommunicationValue) -> String { let mut parts = Vec::new(); - let sender = cv.sender(); - let receiver = cv.receiver(); + let sender = cv.get_sender(); + let receiver = cv.get_receiver(); - if let (Some(sender), Some(receiver)) = (sender, receiver) { + if sender > 0 && receiver > 0 { parts.push(format!("{} > {}", sender, receiver)); - } else if let Some(sender) = sender { + } else if sender > 0 { parts.push(format!("{}", sender)); - } else if let Some(receiver) = receiver { + } else if receiver > 0 { parts.push(format!("> {}", receiver)); } @@ -179,7 +179,7 @@ pub fn format_cv(cv: &CommunicationValue) -> String { .get_comm_type_enum() .map(|kind| kind.to_string()) .unwrap_or_else(|| cv.get_type().to_string()); - parts.push(format!("{} (id={:?})", comm_type, cv.id())); + parts.push(format!("{} (id={})", comm_type, cv.get_id())); if cv.is_type(CommunicationType::Relay) { parts.push("".to_string());