[Fix] Connectivity

src/rho/client_connection.rs

@@ -17,6 +17,13 @@ use tokio::sync::RwLock;
 use trust_dns_resolver::TokioAsyncResolver;
 use uuid::Uuid;
 
+fn authenticated_peer_control_request(
+    cv: &CommunicationValue,
+    user_id: u64,
+) -> CommunicationValue {
+    cv.clone().with_sender(user_id)
+}
+
 pub struct ClientConnection {
     pub state: Arc<AppState>,
     pub user_id: u64,
@@ -210,7 +217,10 @@ impl ClientConnection {
                         .await;
                     return;
                 };
-                match rho.await_peer_control(&cv.with_sender(self.user_id)).await {
+                match rho
+                    .await_peer_control(&authenticated_peer_control_request(&cv, self.user_id))
+                    .await
+                {
                     Ok(response) => self.send_message(&response).await,
                     Err(_) => {
                         self.send_error_response(message_id, CommunicationType::ErrorInternal)
@@ -968,6 +978,29 @@ impl ClientConnection {
     }
 }
 
+#[cfg(test)]
+mod tests {
+    use super::authenticated_peer_control_request;
+    use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
+
+    #[test]
+    fn peer_control_requests_replace_a_client_supplied_sender() {
+        let request = CommunicationValue::new(CommunicationType::SyncedSettingSet)
+            .with_sender(99)
+            .add_typed_default(DataType::UserId, DataValue::SignedNumber(99));
+
+        let authenticated = authenticated_peer_control_request(&request, 7);
+
+        assert_eq!(authenticated.sender(), Some(7));
+        assert_eq!(
+            authenticated
+                .get_data(DataType::UserId)
+                .and_then(|value| value.as_number()),
+            Some(99)
+        );
+    }
+}
+
 // Implement Clone to make it easier to work with Arc<ClientConnection>
 impl Clone for ClientConnection {
     fn clone(&self) -> Self {

src/rho/iota_connection.rs

@@ -302,6 +302,13 @@ impl IotaConnection {
             return;
         }
 
+        log_cv_in!(PrintType::Iota, cv);
+
+        if cv.is_type(CommunicationType::SyncedSettingChanged) {
+            self.forward_to_client(cv).await;
+            return;
+        }
+
         let msg_id = message_id;
         if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) {
             if (task)(self.clone(), cv.clone()) {
@@ -309,8 +316,6 @@ impl IotaConnection {
             }
         }
 
-        log_cv_in!(PrintType::Iota, cv);
-
         let cv = if cv.is_type(CommunicationType::ClientStateSync) {
             self.add_call_state(cv).await
         } else {

src/rho/relay_router.rs

@@ -95,6 +95,10 @@ pub fn message_security_class(frame: &CommunicationValue) -> MessageSecurityClas
     } else if frame.is_type(CommunicationType::GetChatSecret)
         || frame.is_type(CommunicationType::MessageGet)
         || frame.is_type(CommunicationType::MessagesGet)
+        || frame.is_type(CommunicationType::SyncedSettingSet)
+        || frame.is_type(CommunicationType::SyncedSettingGet)
+        || frame.is_type(CommunicationType::SyncedSettingDelete)
+        || frame.is_type(CommunicationType::SyncedSettingsList)
     {
         MessageSecurityClass::AuthenticatedPeerControl
     } else {
@@ -296,7 +300,8 @@ mod tests {
     use mtp::codec::{CommunicationType, CommunicationValue, DataValue};
 
     use super::{
-        RelayRouteError, RelaySource, RouteTarget, prepare_frame, validate_source_next_hop,
+        MessageSecurityClass, RelayRouteError, RelaySource, RouteTarget, message_security_class,
+        prepare_frame, validate_source_next_hop,
     };
 
     fn wire(target: RouteTarget) -> u64 {
@@ -418,4 +423,37 @@ mod tests {
         };
         assert_eq!(routed.payload(), response.payload());
     }
+
+    #[test]
+    fn synchronized_setting_requests_use_authenticated_peer_control() {
+        for setting_type in [
+            CommunicationType::SyncedSettingSet,
+            CommunicationType::SyncedSettingGet,
+            CommunicationType::SyncedSettingDelete,
+            CommunicationType::SyncedSettingsList,
+        ] {
+            let frame = CommunicationValue::new(setting_type);
+            assert_eq!(
+                message_security_class(&frame),
+                MessageSecurityClass::AuthenticatedPeerControl
+            );
+        }
+    }
+
+    #[test]
+    fn synchronized_setting_requests_are_not_relay_only() {
+        for setting_type in [
+            CommunicationType::SyncedSettingSet,
+            CommunicationType::SyncedSettingGet,
+            CommunicationType::SyncedSettingDelete,
+            CommunicationType::SyncedSettingsList,
+            CommunicationType::SyncedSettingChanged,
+        ] {
+            let frame = CommunicationValue::new(setting_type);
+            assert_ne!(
+                message_security_class(&frame),
+                MessageSecurityClass::RelayOnly
+            );
+        }
+    }
 }

src/rho/rho_connection.rs

@@ -18,6 +18,16 @@ pub struct RhoConnection {
     app_connections: DashMap<(u64, String, Uuid), Arc<AppConnection>>,
 }
 
+fn message_targets_client(cv: &CommunicationValue, user_id: u64, session_id: u64) -> bool {
+    if cv.receiver() != Some(user_id) {
+        return false;
+    }
+    let Some(target_session_id) = cv.get_data(DataType::SessionId).as_number() else {
+        return true;
+    };
+    target_session_id == i128::from(session_id)
+}
+
 impl RhoConnection {
     /// Create a new RhoConnection
     pub async fn new(iota_connection: Arc<IotaConnection>, user_ids: Vec<i64>) -> Self {
@@ -254,25 +264,19 @@ 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 {
+        if cv.receiver().is_none() {
             log_err!(
                 0,
                 crate::util::logger::PrintType::General,
                 "Discarded message without an MTP receiver"
             );
             return;
-        };
-        let session_id = cv.get_data(DataType::SessionId).as_number();
+        }
 
         for connection in connections.iter() {
-            if connection.user_id != receiver_id {
+            if !message_targets_client(&cv, connection.user_id, connection.session_id) {
                 continue;
             }
-            if let Some(session_id) = session_id {
-                if connection.session_id as i128 != session_id {
-                    continue;
-                }
-            }
             connection.clone().send_message(&cv).await;
         }
     }
@@ -357,3 +361,20 @@ impl RhoConnection {
         self.client_connections.len()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::message_targets_client;
+    use mtp::codec::{CommunicationType, CommunicationValue, DataType};
+
+    #[test]
+    fn sessionless_setting_change_targets_all_devices_for_one_user() {
+        let event =
+            CommunicationValue::new(CommunicationType::SyncedSettingChanged).with_receiver(7);
+
+        assert!(message_targets_client(&event, 7, 1));
+        assert!(message_targets_client(&event, 7, 2));
+        assert!(!message_targets_client(&event, 8, 1));
+        assert!(event.get_data(DataType::SessionId).is_none());
+    }
+}