From 47fb6f320ae9146ff1b3c64b33b44b050835de68 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Tue, 1 Sep 2026 22:17:57 +0200 Subject: [PATCH] [Add] Client settings --- mtp-type-maps | 2 +- src/rho/client_connection.rs | 91 ++++++++++++++++++++---------------- src/rho/iota_connection.rs | 7 ++- src/rho/relay_router.rs | 11 +++++ 4 files changed, 69 insertions(+), 42 deletions(-) diff --git a/mtp-type-maps b/mtp-type-maps index f4e45aa..4b82f4f 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit f4e45aa3a3ad0e3c3a257f66857b904a1af7901c +Subproject commit 4b82f4f8139ed9aa74fa86f73ba8f0d565703c86 diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 5625793..f1f4476 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -12,11 +12,20 @@ use crate::{log_cv_in, log_cv_out, log_err, log_out}; use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use std::str::FromStr; use std::sync::Arc; +use std::sync::atomic::{AtomicU32, Ordering}; use std::time::Duration; use tokio::sync::RwLock; use trust_dns_resolver::TokioAsyncResolver; use uuid::Uuid; +static NEXT_POLICY_REQUEST_ID: AtomicU32 = AtomicU32::new(1); + +fn next_policy_request_id() -> u32 { + NEXT_POLICY_REQUEST_ID + .fetch_add(1, Ordering::Relaxed) + .max(1) +} + fn authenticated_peer_control_request(cv: &CommunicationValue, user_id: u64) -> CommunicationValue { cv.clone().with_sender(user_id) } @@ -600,6 +609,48 @@ impl ClientConnection { return; } }; + // Resolve the recipient before creating call state. Without the + // recipient Iota, policy cannot be evaluated and delivery must fail. + let target_rho = match self.state.rho.get_for_user(receiver_id as i64).await { + Some(rho) => rho, + None => { + self.send_error_response(message_id, CommunicationType::ErrorNoIota) + .await; + return; + } + }; + let sender_id = self.get_user_id().await as i64; + let policy_request = CommunicationValue::new(CommunicationType::UserBlockCheck) + .with_id(next_policy_request_id()) + .add_typed_default( + DataType::SenderId, + DataValue::SignedNumber(sender_id.into()), + ) + .add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id)); + let policy_response = target_rho + .get_iota_connection() + .clone() + .await_response(&policy_request, Some(Duration::from_secs(8))) + .await; + let blocked = match policy_response { + Ok(response) if response.is_type(CommunicationType::UserBlockCheck) => { + response.get_data(DataType::IsBlocked).as_bool() + } + _ => None, + }; + match blocked { + Some(true) => { + self.send_error_response(message_id, CommunicationType::ErrorNotAuthenticated) + .await; + return; + } + Some(false) => {} + None => { + self.send_error_response(message_id, CommunicationType::ErrorInternal) + .await; + return; + } + } let invited = self .state .call_manager @@ -621,46 +672,6 @@ impl ClientConnection { return; } - // Find target RhoConnection - let target_rho = match self.state.rho.get_for_user(receiver_id as i64).await { - Some(rho) => rho, - _ => { - // Get sender user ID - let sender_id = self.get_user_id().await; - - // 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::CallId, DataValue::Str(call_id.to_string())) - .add_typed_default( - DataType::Notifications, - DataValue::Str("call_invite".to_string()), - ); - - let omega_conn = self.state.omega.clone(); - // Send fire-and-forget, don't await to avoid blocking - tokio::spawn(async move { - let _ = omega_conn.send_message(&push_cv).await; - }); - - let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound) - .with_id(message_id) - .add_typed_default( - DataType::ReceiverId, - DataValue::SignedNumber(receiver_id.into()), - ); - self.send_message(&error_cv).await; - return; - } - }; - - // Get sender user ID - let sender_id = self.get_user_id().await as i64; - // Create and send call distribution message let forward = CommunicationValue::new(CommunicationType::CallInvite) .with_receiver(receiver_id as u64) diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 5b9e08d..0130218 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -310,7 +310,12 @@ impl IotaConnection { log_cv_in!(PrintType::Iota, cv); - if cv.is_type(CommunicationType::SyncedSettingChanged) { + if cv.is_type(CommunicationType::SyncedSettingChanged) + || cv.is_type(CommunicationType::UserBlobChanged) + || cv.is_type(CommunicationType::BlockedUsersChanged) + || cv.is_type(CommunicationType::ReceiptPolicyChanged) + || cv.is_type(CommunicationType::MessageStoragePolicyChanged) + { self.forward_to_client(cv).await; return; } diff --git a/src/rho/relay_router.rs b/src/rho/relay_router.rs index a903256..5117db9 100644 --- a/src/rho/relay_router.rs +++ b/src/rho/relay_router.rs @@ -99,6 +99,17 @@ pub fn message_security_class(frame: &CommunicationValue) -> MessageSecurityClas || frame.is_type(CommunicationType::SyncedSettingGet) || frame.is_type(CommunicationType::SyncedSettingDelete) || frame.is_type(CommunicationType::SyncedSettingsList) + || frame.is_type(CommunicationType::UserBlobPut) + || frame.is_type(CommunicationType::UserBlobGet) + || frame.is_type(CommunicationType::UserBlobDelete) + || frame.is_type(CommunicationType::UserBlobList) + || frame.is_type(CommunicationType::UserBlock) + || frame.is_type(CommunicationType::UserUnblock) + || frame.is_type(CommunicationType::BlockedUsersGet) + || frame.is_type(CommunicationType::ReceiptPolicyGet) + || frame.is_type(CommunicationType::ReceiptPolicySet) + || frame.is_type(CommunicationType::MessageStoragePolicyGet) + || frame.is_type(CommunicationType::MessageStoragePolicySet) { MessageSecurityClass::AuthenticatedPeerControl } else {