(feat): call crypto migration

src/anonymous_clients/anonymous_client_connection.rs

@@ -7,7 +7,7 @@ use tokio::sync::RwLock;
 use uuid::Uuid;
 
 use crate::anonymous_clients::anonymous_manager::{self, generate_username};
-use crate::calls::call_manager;
+use crate::calls::{call_group::call_invite_secret_from_cv, call_manager};
 use crate::data::user::UserStatus;
 use crate::omega::omega_connection::{OmegaConnection, get_omega_connection};
 use crate::rho::connection::GeneralConnection;
@@ -367,18 +367,29 @@ impl AnonymousClientConnection {
             }
         };
 
-        let secret = cv
+        let secret = match call_invite_secret_from_cv(&cv) {
-            .get_data(DataType::CallSecret)
+            Some(secret) => secret,
-            .as_str()
+            None => {
-            .map(|s| s.to_string());
+                self.send_error_response(&cv.get_id(), CommunicationType::BadRequest)
+                    .await;
+                return;
+            }
+        };
         let invited =
-            call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await;
+            call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret.clone())
+                .await;
         if !invited {
             self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidCallId)
                 .await;
             return;
         }
 
+        if !call_manager::should_forward_invite(self.user_id, receiver_id as u64) {
+            let response = CommunicationValue::new(CommunicationType::Success).with_id(cv.get_id());
+            self.send_message(&response).await;
+            return;
+        }
+
         // Find target RhoConnection
         let target_rho = match rho_manager::get_rho_con_for_user(receiver_id).await {
             Some(rho) => rho,
@@ -423,6 +434,7 @@ impl AnonymousClientConnection {
         let forward = CommunicationValue::new(CommunicationType::CallInvite)
             .with_receiver(receiver_id as u64)
             .with_sender(sender_id)
+            .add_typed_default(DataType::CallSecret, secret.to_data_value())
             .add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string()))
             .add_typed_default(
                 DataType::ReceiverId,

src/calls/call_group.rs

@@ -1,4 +1,4 @@
-use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
+use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataTypeId, DataValue, TypeMap};
 use serde_json::Map;
 
 use std::{collections::BTreeMap, env, sync::Arc, time::Duration};
@@ -16,7 +16,84 @@ pub struct CallGroup {
     pub show: RwLock<bool>,
     pub anonymous_joining: RwLock<bool>,
     pub short_link: RwLock<Option<String>>,
-    pub secrets: RwLock<BTreeMap<(u64, u64), String>>,
+    pub secrets: RwLock<BTreeMap<u64, CallSecretEnvelope>>,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct CallSecretEnvelope {
+    pub secret_id: String,
+    pub version_number: i64,
+    pub encrypted_secret: Vec<u8>,
+    pub kem_ciphertext: Vec<u8>,
+    pub wrapping_scheme: String,
+}
+
+impl CallSecretEnvelope {
+    pub fn from_data_value(value: &DataValue) -> Option<Self> {
+        let tm = TypeMap::latest();
+        let secret_id = value
+            .get_field(DataType::SecretId.to_id(&tm))?
+            .as_str()?
+            .to_string();
+        let version_number = value
+            .get_field(DataType::VersionNumber.to_id(&tm))?
+            .as_signed_number()
+            .and_then(|n| i64::try_from(n).ok())
+            .or_else(|| {
+                value
+                    .get_field(DataType::VersionNumber.to_id(&tm))?
+                    .as_number()
+                    .and_then(|n| i64::try_from(n).ok())
+            })?;
+        let encrypted_secret = value
+            .get_field(DataType::EncryptedSecret.to_id(&tm))?
+            .as_bytes()?;
+        let kem_ciphertext = value
+            .get_field(DataType::KemCiphertext.to_id(&tm))?
+            .as_bytes()?;
+        let wrapping_scheme = value
+            .get_field(DataType::WrappingScheme.to_id(&tm))?
+            .as_str()?
+            .to_string();
+
+        Some(Self {
+            secret_id,
+            version_number,
+            encrypted_secret,
+            kem_ciphertext,
+            wrapping_scheme,
+        })
+    }
+
+    pub fn to_data_value(&self) -> DataValue {
+        let tm = TypeMap::latest();
+        let mut map: BTreeMap<DataTypeId, DataValue> = BTreeMap::new();
+        map.insert(
+            DataType::SecretId.to_id(&tm),
+            DataValue::Str(self.secret_id.clone()),
+        );
+        map.insert(
+            DataType::VersionNumber.to_id(&tm),
+            DataValue::SignedNumber(self.version_number.into()),
+        );
+        map.insert(
+            DataType::EncryptedSecret.to_id(&tm),
+            DataValue::Bytes(self.encrypted_secret.clone()),
+        );
+        map.insert(
+            DataType::KemCiphertext.to_id(&tm),
+            DataValue::Bytes(self.kem_ciphertext.clone()),
+        );
+        map.insert(
+            DataType::WrappingScheme.to_id(&tm),
+            DataValue::Str(self.wrapping_scheme.clone()),
+        );
+        DataValue::container_from_map(&map)
+    }
+}
+
+pub fn call_invite_secret_from_cv(cv: &CommunicationValue) -> Option<CallSecretEnvelope> {
+    CallSecretEnvelope::from_data_value(cv.get_data(DataType::CallSecret))
 }
 
 impl CallGroup {
@@ -40,6 +117,10 @@ impl CallGroup {
             .cloned()
     }
 
+    pub async fn get_secret_for_user(&self, user_id: u64) -> Option<CallSecretEnvelope> {
+        self.secrets.read().await.get(&user_id).cloned()
+    }
+
     pub async fn is_anonymous(&self) -> bool {
         *self.anonymous_joining.read().await
     }
@@ -154,3 +235,110 @@ impl CallGroup {
         self.short_link.read().await.clone()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn envelope(label: &str) -> CallSecretEnvelope {
+        CallSecretEnvelope {
+            secret_id: format!("call:test:{label}"),
+            version_number: 1,
+            encrypted_secret: format!("encrypted:{label}").into_bytes(),
+            kem_ciphertext: format!("kem:{label}").into_bytes(),
+            wrapping_scheme: "mtp-call-secret-kem-chacha20poly1305-hkdf-sha256-v1".to_string(),
+        }
+    }
+
+    #[test]
+    fn call_secret_plain_string_is_rejected() {
+        assert!(
+            CallSecretEnvelope::from_data_value(&DataValue::Str("old-secret".to_string()))
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn call_secret_missing_required_field_is_rejected() {
+        let tm = TypeMap::latest();
+        let mut map: BTreeMap<DataTypeId, DataValue> = BTreeMap::new();
+        map.insert(
+            DataType::SecretId.to_id(&tm),
+            DataValue::Str("call:test:main".to_string()),
+        );
+        map.insert(
+            DataType::VersionNumber.to_id(&tm),
+            DataValue::SignedNumber(1),
+        );
+        map.insert(
+            DataType::EncryptedSecret.to_id(&tm),
+            DataValue::Bytes(vec![1, 2, 3]),
+        );
+        map.insert(
+            DataType::WrappingScheme.to_id(&tm),
+            DataValue::Str("mtp-call-secret-kem-chacha20poly1305-hkdf-sha256-v1".to_string()),
+        );
+
+        assert!(
+            CallSecretEnvelope::from_data_value(&DataValue::container_from_map(&map)).is_none()
+        );
+    }
+
+    #[test]
+    fn call_secret_envelope_round_trips_opaque_bytes() {
+        let secret = envelope("receiver");
+
+        assert_eq!(
+            CallSecretEnvelope::from_data_value(&secret.to_data_value()),
+            Some(secret)
+        );
+    }
+
+    #[test]
+    fn call_invite_missing_call_secret_is_rejected() {
+        let cv = CommunicationValue::new(CommunicationType::CallInvite);
+
+        assert!(call_invite_secret_from_cv(&cv).is_none());
+    }
+
+    #[test]
+    fn call_invite_parses_call_secret_envelope() {
+        let secret = envelope("receiver");
+        let cv = CommunicationValue::new(CommunicationType::CallInvite)
+            .add_typed_default(DataType::CallSecret, secret.clone().to_data_value());
+
+        assert_eq!(call_invite_secret_from_cv(&cv), Some(secret));
+    }
+
+    #[test]
+    fn receiver_call_invite_contains_receiver_envelope_unchanged() {
+        let receiver_secret = envelope("receiver");
+        let cv = CommunicationValue::new(CommunicationType::CallInvite).add_typed_default(
+            DataType::CallSecret,
+            receiver_secret.clone().to_data_value(),
+        );
+
+        assert_eq!(
+            CallSecretEnvelope::from_data_value(cv.get_data(DataType::CallSecret)),
+            Some(receiver_secret)
+        );
+    }
+
+    #[tokio::test]
+    async fn call_group_returns_only_current_users_secret() {
+        let call_id = Uuid::new_v4();
+        let group = CallGroup::new(call_id, Arc::new(Caller::new(1, call_id, true)));
+        let own_secret = envelope("own");
+        let receiver_secret = envelope("receiver");
+
+        {
+            let mut secrets = group.secrets.write().await;
+            secrets.insert(1, own_secret.clone());
+            secrets.insert(2, receiver_secret.clone());
+        }
+
+        assert_eq!(group.get_secret_for_user(1).await, Some(own_secret));
+        assert_eq!(group.get_secret_for_user(2).await, Some(receiver_secret));
+        assert_eq!(group.get_secret_for_user(3).await, None);
+    }
+}

src/calls/call_manager.rs

@@ -3,7 +3,11 @@ use once_cell::sync::Lazy;
 use std::sync::Arc;
 use uuid::Uuid;
 
-use crate::calls::{call_group::CallGroup, call_util, caller::Caller};
+use crate::calls::{
+    call_group::{CallGroup, CallSecretEnvelope},
+    call_util,
+    caller::Caller,
+};
 
 pub static CALL_GROUPS: Lazy<DashMap<Uuid, Arc<CallGroup>>> = Lazy::new(|| DashMap::new());
 #[allow(dead_code)]
@@ -74,7 +78,7 @@ pub async fn add_invite(
     call_id: Uuid,
     inviter_id: u64,
     invitee_id: u64,
-    secret: Option<String>,
+    secret: CallSecretEnvelope,
 ) -> bool {
     if let Some(cg) = CALL_GROUPS.get(&call_id) {
         let mut members = cg.members.write().await;
@@ -86,14 +90,89 @@ pub async fn add_invite(
                 members.push(Arc::new(Caller::new(invitee_id, call_id, false)));
             }
 
-            if let Some(secret) = secret {
             let mut secrets = cg.secrets.write().await;
-                secrets.insert((inviter_id, invitee_id), secret.clone());
+            secrets.insert(invitee_id, secret);
-                secrets.insert((invitee_id, inviter_id), secret);
-            }
 
             return true;
         }
     }
     false
 }
+
+pub fn should_forward_invite(inviter_id: u64, invitee_id: u64) -> bool {
+    inviter_id != invitee_id
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn envelope(label: &str) -> CallSecretEnvelope {
+        CallSecretEnvelope {
+            secret_id: format!("call:test:{label}"),
+            version_number: 1,
+            encrypted_secret: format!("encrypted:{label}").into_bytes(),
+            kem_ciphertext: format!("kem:{label}").into_bytes(),
+            wrapping_scheme: "mtp-call-secret-kem-chacha20poly1305-hkdf-sha256-v1".to_string(),
+        }
+    }
+
+    #[tokio::test]
+    async fn add_invite_stores_receiver_envelope_only_under_receiver_id() {
+        let call_id = Uuid::new_v4();
+        let sender_id = 11;
+        let receiver_id = 22;
+        let group = Arc::new(CallGroup::new(
+            call_id,
+            Arc::new(Caller::new(sender_id, call_id, true)),
+        ));
+        CALL_GROUPS.insert(call_id, group.clone());
+
+        let receiver_secret = envelope("receiver");
+
+        assert!(add_invite(call_id, sender_id, receiver_id, receiver_secret.clone(),).await);
+
+        assert_eq!(
+            group.get_secret_for_user(receiver_id).await,
+            Some(receiver_secret.clone())
+        );
+        assert_eq!(group.get_secret_for_user(sender_id).await, None);
+
+        CALL_GROUPS.remove(&call_id);
+    }
+
+    #[tokio::test]
+    async fn self_invite_stores_secret_without_duplicating_member() {
+        let call_id = Uuid::new_v4();
+        let sender_id = 33;
+        let group = Arc::new(CallGroup::new(
+            call_id,
+            Arc::new(Caller::new(sender_id, call_id, true)),
+        ));
+        CALL_GROUPS.insert(call_id, group.clone());
+
+        let secret = envelope("self");
+
+        assert!(add_invite(call_id, sender_id, sender_id, secret.clone()).await);
+
+        assert_eq!(group.get_secret_for_user(sender_id).await, Some(secret));
+        assert_eq!(
+            group
+                .members
+                .read()
+                .await
+                .iter()
+                .filter(|member| member.user_id == sender_id)
+                .count(),
+            1
+        );
+
+        CALL_GROUPS.remove(&call_id);
+    }
+
+    #[test]
+    fn self_invites_are_not_forwarded() {
+        assert!(!should_forward_invite(44, 44));
+        assert!(should_forward_invite(44, 55));
+    }
+}

src/rho/client_connection.rs

@@ -1,5 +1,5 @@
 use crate::anonymous_clients::anonymous_manager;
-use crate::calls::{call_manager, call_util};
+use crate::calls::{call_group::call_invite_secret_from_cv, call_manager, call_util};
 use crate::omega::omega_connection::get_omega_connection;
 use crate::rho::connection::GeneralConnection;
 use crate::rho::{rho_connection::RhoConnection, rho_manager};
@@ -354,18 +354,29 @@ impl ClientConnection {
             }
         };
 
-        let secret = cv
+        let secret = match call_invite_secret_from_cv(&cv) {
-            .get_data(DataType::CallSecret)
+            Some(secret) => secret,
-            .as_str()
+            None => {
-            .map(|s| s.to_string());
+                self.send_error_response(cv.get_id(), CommunicationType::BadRequest)
+                    .await;
+                return;
+            }
+        };
         let invited =
-            call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await;
+            call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret.clone())
+                .await;
         if !invited {
             self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId)
                 .await;
             return;
         }
 
+        if !call_manager::should_forward_invite(self.user_id, receiver_id as u64) {
+            let response = CommunicationValue::new(CommunicationType::Success).with_id(cv.get_id());
+            self.send_message(&response).await;
+            return;
+        }
+
         // Find target RhoConnection
         let target_rho = match rho_manager::get_rho_con_for_user(receiver_id as i64).await {
             Some(rho) => rho,
@@ -410,10 +421,7 @@ impl ClientConnection {
         let forward = CommunicationValue::new(CommunicationType::CallInvite)
             .with_receiver(receiver_id as u64)
             .with_sender(sender_id as u64)
-            .add_typed_default(
+            .add_typed_default(DataType::CallSecret, secret.to_data_value())
-                DataType::CallSecret,
-                cv.get_data(DataType::CallSecret).clone(),
-            )
             .add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string()))
             .add_typed_default(
                 DataType::ReceiverId,

src/rho/connection.rs

@@ -207,7 +207,16 @@ impl GeneralConnection {
                                                 );
                                             }
 
-                                            // Add to global calls (without contact-specific secret)
+                                            if let Some(secret) =
+                                                call.get_secret_for_user(user_id as u64).await
+                                            {
+                                                base_call_map.insert(
+                                                    DataType::CallSecret.to_id(&tm),
+                                                    secret.to_data_value(),
+                                                );
+                                            }
+
+                                            // Add to global calls with only this user's recipient-specific secret.
                                             global_calls.push(DataValue::container_from_map(
                                                 &base_call_map,
                                             ));
@@ -218,19 +227,7 @@ impl GeneralConnection {
                                                     continue;
                                                 }
 
-                                                let mut contact_call_map = base_call_map.clone();
+                                                let contact_call_map = base_call_map.clone();
-
-                                                if let Some(secret) = call
-                                                    .secrets
-                                                    .read()
-                                                    .await
-                                                    .get(&(member_id, user_id as u64))
-                                                {
-                                                    contact_call_map.insert(
-                                                        DataType::CallSecret.to_id(&tm),
-                                                        DataValue::Str(secret.clone()),
-                                                    );
-                                                }
 
                                                 invites
                                                     .entry(member_id as i64)

src/rho/iota_connection.rs

@@ -433,7 +433,11 @@ impl IotaConnection {
                     base_call_map.insert(DataType::HasAdmin.to_id(&tm), DataValue::Bool(true));
                 }
 
-                // Add to global calls (without contact-specific secret)
+                if let Some(secret) = call.get_secret_for_user(user_id).await {
+                    base_call_map.insert(DataType::CallSecret.to_id(&tm), secret.to_data_value());
+                }
+
+                // Add to global calls with only this user's recipient-specific secret.
                 global_calls.push(DataValue::container_from_map(&base_call_map));
 
                 // Attach this call to EVERY member of the call (other than ourselves)
@@ -443,15 +447,7 @@ impl IotaConnection {
                         continue;
                     }
 
-                    let mut contact_call_map = base_call_map.clone();
+                    let contact_call_map = base_call_map.clone();
-
-                    // Add secret if it exists for this pairing
-                    if let Some(secret) = call.secrets.read().await.get(&(member_id, user_id)) {
-                        contact_call_map.insert(
-                            DataType::CallSecret.to_id(&tm),
-                            DataValue::Str(secret.clone()),
-                        );
-                    }
 
                     invites
                         .entry(member_id as i64)