[WIP] MTP migration

.gitignore

@@ -17,3 +17,6 @@ target
 # Added by cargo
 **/target
 **/logs
+
+*.mk
+*.mpkb

Cargo.lock

@@ -1722,12 +1722,13 @@ dependencies = [
 [[package]]
 name = "mtp"
 version = "0.1.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d"
 dependencies = [
  "mtp-client",
  "mtp-codec",
  "mtp-common",
  "mtp-crypto",
+ "mtp-files",
  "mtp-host",
  "mtp-transport",
  "mtp-type-map",
@@ -1736,7 +1737,7 @@ dependencies = [
 [[package]]
 name = "mtp-client"
 version = "0.1.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d"
 dependencies = [
  "mtp-codec",
  "mtp-common",
@@ -1749,7 +1750,7 @@ dependencies = [
 [[package]]
 name = "mtp-codec"
 version = "0.1.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d"
 dependencies = [
  "base64 0.22.1",
  "byteorder",
@@ -1762,7 +1763,7 @@ dependencies = [
 [[package]]
 name = "mtp-common"
 version = "0.1.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d"
 dependencies = [
  "quinn",
  "rustls",
@@ -1773,8 +1774,9 @@ dependencies = [
 [[package]]
 name = "mtp-crypto"
 version = "0.1.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d"
 dependencies = [
+ "base64 0.22.1",
  "chacha20poly1305",
  "ed25519-dalek",
  "getrandom 0.4.3",
@@ -1788,10 +1790,19 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "mtp-files"
+version = "0.1.0"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d"
+dependencies = [
+ "mtp-crypto",
+ "thiserror 1.0.69",
+]
+
 [[package]]
 name = "mtp-host"
 version = "0.1.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d"
 dependencies = [
  "mtp-codec",
  "mtp-common",
@@ -1804,7 +1815,7 @@ dependencies = [
 [[package]]
 name = "mtp-transport"
 version = "0.1.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d"
 dependencies = [
  "log",
  "mtp-codec",
@@ -1818,7 +1829,7 @@ dependencies = [
 [[package]]
 name = "mtp-type-map"
 version = "0.1.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#56903049b66d67a5a34f8b6fb78820f19771f2f4"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b96c072a0f87de7828e45cf3dcd44aef4a9e459d"
 dependencies = [
  "serde",
  "serde_yaml",
@@ -1871,9 +1882,9 @@ dependencies = [
 
 [[package]]
 name = "num-bigint"
-version = "0.4.6"
+version = "0.4.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9"
+checksum = "c863e9ab5e7bf9c99ba75e1050f1e4d624ae87ed3532d6238ffbdc7b585dbbe6"
 dependencies = [
  "num-integer",
  "num-traits",
@@ -2092,6 +2103,7 @@ dependencies = [
  "livekit-protocol",
  "log",
  "mtp",
+ "mtp-transport",
  "once_cell",
  "rand 0.8.6",
  "rustls",
@@ -2707,9 +2719,9 @@ dependencies = [
 
 [[package]]
 name = "rustc-hash"
-version = "2.1.2"
+version = "2.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe"
+checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d"
 
 [[package]]
 name = "rustc_version"

Cargo.toml

@@ -8,7 +8,11 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = [
     "host",
     "client",
     "crypto",
+    "files",
 ] }
+# Needed directly for `ConnectionHandle`, which `mtp-client` does not re-export.
+# Pulled in transitively already via `mtp`, so this just names the same crate/commit.
+mtp-transport = { git = "https://git.methanium.net/Methanium/mtp.git" }
 
 ansi_term = "*"
 uuid = { version = "*", features = ["v4"] }

src/anonymous_clients/anonymous_client_connection.rs

@@ -1,5 +1,5 @@
 use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue, TypeMap};
-use mtp::transport::{Receiver, Sender};
+use mtp::host::{Receiver, Sender};
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
@@ -138,7 +138,8 @@ impl AnonymousClientConnection {
                 for call_invitee in call.members.read().await.clone() {
                     let call_invitee_cv = get_omega_connection()
                         .await_response(
-                            &CommunicationValue::new(CommunicationType::GetUserData).add_typed_default(
+                            &CommunicationValue::new(CommunicationType::GetUserData)
+                                .add_typed_default(
                                     DataType::UserId,
                                     DataValue::SignedNumber(call_invitee.user_id.into()),
                                 ),
@@ -164,7 +165,12 @@ impl AnonymousClientConnection {
                         call_invitee_cv.get_data(DataType::Avatar).clone(),
                     ));
 
-                    let _ = invited.push(DataValue::Container(json_invitee.iter().map(|(k, v)| (k.to_id(&TypeMap::latest()), v.clone())).collect()));
+                    let _ = invited.push(DataValue::Container(
+                        json_invitee
+                            .iter()
+                            .map(|(k, v)| (k.to_id(&TypeMap::latest()), v.clone()))
+                            .collect(),
+                    ));
                 }
 
                 let token = call.create_anonymous_token(self.get_user_id()).await;
@@ -190,8 +196,19 @@ impl AnonymousClientConnection {
                                 DataType::Display,
                                 DataValue::Str(self.get_display_name().await),
                             )
-                            .add_typed_default(DataType::Avatar, DataValue::Str(self.get_avatar().await))
+                            .add_typed_default(
-                            .add_typed_default(DataType::CallState, DataValue::Container(serialized.iter().map(|(k, v)| (k.to_id(&TypeMap::latest()), v.clone())).collect())),
+                                DataType::Avatar,
+                                DataValue::Str(self.get_avatar().await),
+                            )
+                            .add_typed_default(
+                                DataType::CallState,
+                                DataValue::Container(
+                                    serialized
+                                        .iter()
+                                        .map(|(k, v)| (k.to_id(&TypeMap::latest()), v.clone()))
+                                        .collect(),
+                                ),
+                            ),
                     )
                     .await;
             }
@@ -261,7 +278,10 @@ impl AnonymousClientConnection {
                             DataType::Display,
                             DataValue::Str(anonymous.get_display_name().await),
                         )
-                        .add_typed_default(DataType::UserState, DataValue::Str("online".to_string()))
+                        .add_typed_default(
+                            DataType::UserState,
+                            DataValue::Str("online".to_string()),
+                        )
                         .add_typed_default(
                             DataType::Avatar,
                             DataValue::Str(anonymous.get_avatar().await),
@@ -321,7 +341,10 @@ impl AnonymousClientConnection {
 
     /// Handle call invite
     async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
-        let receiver_id: i64 = cv.get_data(DataType::ReceiverId).as_signed_number().unwrap_or(0) as i64;
+        let receiver_id: i64 = cv
+            .get_data(DataType::ReceiverId)
+            .as_signed_number()
+            .unwrap_or(0) as i64;
         if receiver_id == 0 {
             self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId)
                 .await;
@@ -384,7 +407,10 @@ impl AnonymousClientConnection {
 
                 let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound)
                     .with_id(cv.get_id())
-                    .add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into()));
+                    .add_typed_default(
+                        DataType::ReceiverId,
+                        DataValue::SignedNumber(receiver_id.into()),
+                    );
                 self.send_message(&error_cv).await;
                 return;
             }
@@ -446,8 +472,14 @@ impl AnonymousClientConnection {
     }
     async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) {
         let call_id = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")).unwrap();
-        let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0);
+        let user_id = cv
-        let untill = cv.get_data(DataType::Untill).as_signed_number().unwrap_or(0);
+            .get_data(DataType::UserId)
+            .as_signed_number()
+            .unwrap_or(0);
+        let untill = cv
+            .get_data(DataType::Untill)
+            .as_signed_number()
+            .unwrap_or(0);
 
         let call = call_manager::get_call(call_id).await;
         if let Some(call) = call {
@@ -467,7 +499,10 @@ impl AnonymousClientConnection {
     }
     async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) {
         let call_id = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or("")).unwrap();
-        let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0);
+        let user_id = cv
+            .get_data(DataType::UserId)
+            .as_signed_number()
+            .unwrap_or(0);
 
         let call = call_manager::get_call(call_id).await;
         if let Some(call) = call {

src/main.rs

@@ -15,34 +15,33 @@ pub static WORKING_DIR: Lazy<PathBuf> =
     Lazy::new(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")));
 use rustls::crypto::aws_lc_rs::default_provider;
 
-use base64::engine::general_purpose::STANDARD as BASE64_STD;
-use base64::Engine as _;
 use mtp::crypto::Keyring;
+use mtp::files::{load_keyring as load_keyring_file, save_keyring, save_public_key_bundle};
 
 use crate::{
-    calls::call_util::garbage_collect_calls,
+    calls::call_util::garbage_collect_calls, omega::omega_connection::get_omega_connection,
-    omega::omega_connection::get_omega_connection,
+    rho::server::start, util::logger::startup,
-    rho::server::start,
-    util::logger::startup,
 };
 
+const KEYRING_PATH: &str = "./omikron.mk";
+
 static KEYRING: Lazy<Keyring> = Lazy::new(|| {
-    if let Ok(encoded) = env::var("KEYRING") {
+    load_keyring_file(KEYRING_PATH).unwrap_or_else(|_| {
-        let bytes = BASE64_STD.decode(&encoded).expect("Invalid KEYRING base64");
-        Keyring::from_bytes(&bytes).expect("Invalid KEYRING data")
-    } else {
         let kr = Keyring::generate();
-        eprintln!(
+        save_keyring(&kr, KEYRING_PATH).expect("Failed to save generated keyring");
-            "Generated KEYRING (save to env): {}",
+        save_public_key_bundle(&kr.public_key_bundle(), KEYRING_PATH)
-            BASE64_STD.encode(&kr.to_bytes())
+            .expect("Failed to save generated public key bundle");
-        );
+        eprintln!("Generated new keyring at {}", KEYRING_PATH);
         kr
-    }
+    })
 });
 
 pub fn get_keyring() -> &'static Keyring {
     &KEYRING
 }
+pub fn load_keyring() -> Keyring {
+    Keyring::from_bytes(&KEYRING.to_bytes()).unwrap()
+}
 
 #[tokio::main]
 async fn main() {

src/omega/omega_connection.rs

@@ -1,19 +1,18 @@
-use base64::engine::general_purpose::STANDARD as BASE64_STD;
-use base64::Engine as _;
-use mtp::crypto::decrypt_with;
-
 use crate::{
     data::user::UserStatus,
-    get_keyring, log, log_cv_in, log_cv_out, log_err, log_in,
+    load_keyring, log_cv_in, log_cv_out, log_err, log_in,
     rho::rho_manager::{self, RHO_CONNECTIONS, connection_count},
-    util::{
+    util::logger::PrintType,
-        file_util::load_file_vec,
-        logger::PrintType,
-    },
 };
 use dashmap::DashMap;
+use mtp::client::{Client, Receiver, Sender};
 use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
-use mtp::transport::{Policy, Receiver, SendMode, Sender};
+use mtp::{
+    client::ClientConfig,
+    files::load_public_key_bundle,
+    host::{Policy, SendMode},
+};
+use mtp_transport::ConnectionHandle;
 use once_cell::sync::Lazy;
 use std::{collections::HashMap, env, sync::Arc, time::Duration};
 use tokio::{
@@ -21,6 +20,7 @@ use tokio::{
     task::JoinHandle,
     time::{Instant, sleep},
 };
+
 use uuid::Uuid;
 
 // ============================================================================
@@ -109,7 +109,6 @@ pub struct OmegaConnection {
     connection_loop_handle: Arc<Mutex<Option<JoinHandle<()>>>>,
     host: String,
     port: u16,
-    server_cert: Vec<u8>,
     last_ping: Arc<Mutex<i64>>,
     heartbeat_handle: Arc<Mutex<Option<JoinHandle<()>>>>,
     message_send_times: Arc<Mutex<HashMap<Uuid, Instant>>>,
@@ -125,16 +124,9 @@ impl OmegaConnection {
         Self::with_host(&host, port)
     }
 
+    // omikron and omega are both served over publicly trusted (fullchain) certs on
+    // the same domain, so the client uses system root trust - no pinning needed.
     pub fn with_host(host: &str, port: u16) -> Self {
-        // Load server certificate from default location
-        let server_cert =
-            load_file_vec("certs", "cert.pem").expect("Failed to load server certificate");
-
-        Self::with_host_and_cert(host, port, server_cert)
-    }
-
-    // New constructor that accepts certificate directly
-    pub fn with_host_and_cert(host: &str, port: u16, server_cert: Vec<u8>) -> Self {
         let (shutdown_tx, _) = watch::channel(false);
 
         OmegaConnection {
@@ -143,7 +135,6 @@ impl OmegaConnection {
             connection_loop_handle: Arc::new(Mutex::new(None)),
             host: host.to_string(),
             port,
-            server_cert,
             last_ping: Arc::new(Mutex::new(-1)),
             heartbeat_handle: Arc::new(Mutex::new(None)),
             message_send_times: Arc::new(Mutex::new(HashMap::new())),
@@ -265,12 +256,14 @@ impl OmegaConnection {
     async fn connect_once(self: Arc<Self>) -> Result<(), String> {
         *self.state.write().await = ConnectionState::Connecting;
 
-        let addr_str = format!("https://{}:{}", self.host, self.port);
+        let client_config = ClientConfig::new(format!("https://{}:{}", self.host, self.port))
-
+            .with_client_id(
-        let (sender, mut receiver) = mtp::transport::client::connect(
+                env::var("ID")
-            &addr_str,
+                    .unwrap_or("0".to_string())
-            None,
+                    .parse::<u64>()
-            Policy {
+                    .unwrap_or(0),
+            )
+            .with_policy(Policy {
                 send_mode: SendMode::SingleStreamPerMessage,
                 max_message_size: 1_000_000_000,
                 close_frame_len: u32::MAX,
@@ -285,8 +278,13 @@ impl OmegaConnection {
                 max_transient_recv_errors: 20,
                 transient_recv_backoff: Duration::from_millis(100),
                 receiver_queue_capacity: 1000,
-            },
+            });
-        )
+
+        let host_public_key = load_public_key_bundle("./omega.mpkb")
+            .map_err(|e| format!("Failed to load omega.mpkb: {}", e))?;
+
+        let mut connection =
+            Client::auth_connect(client_config, &load_keyring(), &host_public_key)
                 .await
                 .map_err(|e| format!("Connection failed: {}", e))?;
 
@@ -294,11 +292,11 @@ impl OmegaConnection {
             0,
             PrintType::Omega,
             "QUIC connection established to {}",
-            addr_str
+            format!("https://{}:{}", self.host, self.port)
         );
 
         // Store sender
-        let sender_arc = Arc::new(sender);
+        let sender_arc = Arc::new(connection.sender);
         *self.sender.write().await = Some(sender_arc.clone());
         *self.state.write().await = ConnectionState::Connected { identified: false };
 
@@ -308,12 +306,11 @@ impl OmegaConnection {
         // Start read loop
         let read_self = self.clone();
         let read_handle = tokio::spawn(async move {
-            read_self.read_loop(&mut receiver, sender_handle).await;
+            read_self
+                .read_loop(&mut connection.receiver, sender_handle)
+                .await;
         });
 
-        // Send identification
-        self.send_identification().await;
-
         // Start heartbeat
         let heartbeat_self = self.clone();
         let heartbeat_handle = tokio::spawn(async move {
@@ -332,6 +329,8 @@ impl OmegaConnection {
             handle.abort();
         }
 
+        self.clone().sync_client_iota_status().await;
+
         match result {
             Ok(()) => {
                 // Check if we should reconnect
@@ -345,112 +344,6 @@ impl OmegaConnection {
         }
     }
 
-    // -------------------------------------------------------------------------
-    // Identification Handshake
-    // -------------------------------------------------------------------------
-
-    async fn send_identification(&self) {
-        let id = rand::random::<u32>();
-
-        let omikron_id = env::var("ID")
-            .unwrap_or("0".to_string())
-            .parse::<i64>()
-            .unwrap_or(0);
-
-        let identify_msg = CommunicationValue::new(CommunicationType::Identification)
-            .with_id(id)
-            .add_typed_default(DataType::OmikronId, DataValue::SignedNumber(omikron_id.into()));
-
-        WAITING_TASKS.insert(
-            id,
-            WaitingTask {
-                task: Box::new(|selfc, cv| {
-                    if cv.is_type(CommunicationType::ErrorNotFound) {
-                        log_err!(
-                            0,
-                            PrintType::Omega,
-                            "Identification failed: Omikron ID not found"
-                        );
-                        return false;
-                    }
-                    if !cv.is_type(CommunicationType::Challenge) {
-                        return false;
-                    }
-
-                    tokio::spawn(async move {
-                        if let Err(e) = selfc.handle_challenge(cv).await {
-                            log_err!(0, PrintType::Omega, "Challenge handling failed: {}", e);
-                        }
-                    });
-                    true
-                }),
-                inserted_at: Instant::now(),
-            },
-        );
-
-        self.send_message(&identify_msg).await;
-    }
-
-    async fn handle_challenge(&self, cv: CommunicationValue) -> Result<(), String> {
-        let challenge_b64 = cv
-            .get_data(DataType::Challenge)
-            .as_str()
-            .ok_or("Challenge not found")?;
-
-        let blob = BASE64_STD
-            .decode(challenge_b64)
-            .map_err(|e| format!("Base64 decode failed: {}", e))?;
-        let decrypted = decrypt_with(&blob, get_keyring(), b"challenge")
-            .map_err(|e| format!("Decryption failed: {:?}", e))?;
-        let decrypted_challenge =
-            String::from_utf8(decrypted).map_err(|_| "Decrypted challenge not valid UTF-8")?;
-
-        let response_msg = CommunicationValue::new(CommunicationType::ChallengeResponse)
-            .with_id(cv.get_id())
-            .add_typed_default(DataType::Challenge, DataValue::Str(decrypted_challenge));
-
-        let response_id = response_msg.get_id();
-
-        WAITING_TASKS.insert(
-            response_id,
-            WaitingTask {
-                task: Box::new(|selfc, final_cv| {
-                    if !final_cv.is_type(CommunicationType::IdentificationResponse) {
-                        log_err!(0, PrintType::Omega, "Expected identification_response");
-                        return false;
-                    }
-
-                    let accepted = final_cv
-                        .get_data(DataType::Accepted)
-                        .as_bool()
-                        .unwrap_or(false);
-
-                    if !accepted {
-                        log_err!(0, PrintType::Omega, "Omega did not accept identification");
-                        return false;
-                    }
-
-                    tokio::spawn(async move {
-                        let mut state = selfc.state.write().await;
-                        if let ConnectionState::Connected { identified: _ } = *state {
-                            *state = ConnectionState::Connected { identified: true };
-                        }
-                        drop(state);
-
-                        selfc.sync_client_iota_status().await;
-                    });
-
-                    log!(0, PrintType::Omega, "Successfully identified with Omega");
-                    true
-                }),
-                inserted_at: Instant::now(),
-            },
-        );
-
-        self.send_message(&response_msg).await;
-        Ok(())
-    }
-
     async fn sync_client_iota_status(self: Arc<Self>) {
         let mut connected_iota_ids: Vec<DataValue> = Vec::new();
         let mut connected_user_ids: Vec<DataValue> = Vec::new();
@@ -489,7 +382,7 @@ impl OmegaConnection {
     async fn read_loop(
         self: Arc<Self>,
         receiver: &mut Receiver,
-        sender_handle: Arc<mtp::transport::ConnectionHandle>,
+        sender_handle: Arc<ConnectionHandle>,
     ) {
         // Monitor both receiver and sender handle for close
         let mut close_rx = sender_handle.subscribe_close();
@@ -520,7 +413,7 @@ impl OmegaConnection {
                                     let mut user_ids: Vec<u64> = Vec::new();
                                     for value in users {
                                         if let DataValue::SignedNumber(user_id) = value {
-                                            user_ids.push(*user_id as u64);
+                                            user_ids.push(user_id.clone() as u64);
                                         }
                                     }
                                     let connections = crate::rho::rho_manager::RHO_CONNECTIONS.read().await;

src/rho/app_connection.rs

@@ -5,7 +5,7 @@ use crate::rho::{rho_connection::RhoConnection, rho_manager};
 use crate::util::logger::PrintType;
 use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out};
 use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
-use mtp::transport::{Receiver, Sender};
+use mtp::host::{Receiver, Sender};
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::RwLock;
@@ -129,7 +129,10 @@ impl AppConnection {
                             DataType::Avatar,
                             DataValue::Str(anonymous.get_avatar().await),
                         )
-                        .add_typed_default(DataType::UserState, DataValue::Str("online".to_string()));
+                        .add_typed_default(
+                            DataType::UserState,
+                            DataValue::Str("online".to_string()),
+                        );
 
                     self.send_message(&response).await;
 
@@ -183,7 +186,10 @@ impl AppConnection {
         // Send pong response
         let response = CommunicationValue::new(CommunicationType::Pong)
             .with_id(cv.get_id())
-            .add_typed_default(DataType::PingIota, DataValue::SignedNumber(iota_ping.into()));
+            .add_typed_default(
+                DataType::PingIota,
+                DataValue::SignedNumber(iota_ping.into()),
+            );
 
         self.send_message(&response).await;
     }

src/rho/client_connection.rs

@@ -7,7 +7,7 @@ use crate::util::logger::PrintType;
 use crate::{data::user::UserStatus, omega::omega_connection::OmegaConnection};
 use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out};
 use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
-use mtp::transport::{Receiver, Sender};
+use mtp::host::{Receiver, Sender};
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
@@ -158,11 +158,26 @@ impl ClientConnection {
                 } {
                     let response = CommunicationValue::new(CommunicationType::GetUserData)
                         .with_id(cv.get_id())
-                        .add_typed_default(DataType::Username, DataValue::Str(anonymous.get_user_name().await))
+                        .add_typed_default(
-                        .add_typed_default(DataType::UserId, DataValue::SignedNumber(anonymous.get_user_id().into()))
+                            DataType::Username,
-                        .add_typed_default(DataType::Display, DataValue::Str(anonymous.get_display_name().await))
+                            DataValue::Str(anonymous.get_user_name().await),
-                        .add_typed_default(DataType::Avatar, DataValue::Str(anonymous.get_avatar().await))
+                        )
-                        .add_typed_default(DataType::UserState, DataValue::Str("online".to_string()));
+                        .add_typed_default(
+                            DataType::UserId,
+                            DataValue::SignedNumber(anonymous.get_user_id().into()),
+                        )
+                        .add_typed_default(
+                            DataType::Display,
+                            DataValue::Str(anonymous.get_display_name().await),
+                        )
+                        .add_typed_default(
+                            DataType::Avatar,
+                            DataValue::Str(anonymous.get_avatar().await),
+                        )
+                        .add_typed_default(
+                            DataType::UserState,
+                            DataValue::Str("online".to_string()),
+                        );
 
                     self.send_message(&response).await;
 
@@ -198,8 +213,14 @@ impl ClientConnection {
                         let response = CommunicationValue::new(CommunicationType::ErrorInvalidData)
                             .with_id(cv.get_id())
                             .with_receiver(self.user_id)
-                            .add_typed_default(DataType::Message, DataValue::Str("Missing session_id".to_string()))
+                            .add_typed_default(
-                            .add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id));
+                                DataType::Message,
+                                DataValue::Str("Missing session_id".to_string()),
+                            )
+                            .add_typed_default(
+                                DataType::SessionId,
+                                DataValue::SignedNumber(expected_session_id),
+                            );
                         self.send_message(&response).await;
                         return;
                     };
@@ -208,8 +229,14 @@ impl ClientConnection {
                         let response = CommunicationValue::new(CommunicationType::ErrorInvalidData)
                             .with_id(cv.get_id())
                             .with_receiver(self.user_id)
-                            .add_typed_default(DataType::Message, DataValue::Str("session_id mismatch".to_string()))
+                            .add_typed_default(
-                            .add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id));
+                                DataType::Message,
+                                DataValue::Str("session_id mismatch".to_string()),
+                            )
+                            .add_typed_default(
+                                DataType::SessionId,
+                                DataValue::SignedNumber(expected_session_id),
+                            );
                         self.send_message(&response).await;
                         return;
                     }
@@ -218,13 +245,22 @@ impl ClientConnection {
                         let response = CommunicationValue::new(CommunicationType::ErrorInvalidData)
                             .with_id(cv.get_id())
                             .with_receiver(self.user_id)
-                            .add_typed_default(DataType::Message, DataValue::Str("session_id mismatch".to_string()))
+                            .add_typed_default(
-                            .add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id));
+                                DataType::Message,
+                                DataValue::Str("session_id mismatch".to_string()),
+                            )
+                            .add_typed_default(
+                                DataType::SessionId,
+                                DataValue::SignedNumber(expected_session_id),
+                            );
                         self.send_message(&response).await;
                         return;
                     }
                 } else {
-                    cv = cv.add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id));
+                    cv = cv.add_typed_default(
+                        DataType::SessionId,
+                        DataValue::SignedNumber(expected_session_id),
+                    );
                 }
             }
 
@@ -266,7 +302,10 @@ impl ClientConnection {
         // Send pong response
         let response = CommunicationValue::new(CommunicationType::Pong)
             .with_id(cv.get_id())
-            .add_typed_default(DataType::PingIota, DataValue::SignedNumber(iota_ping.into()));
+            .add_typed_default(
+                DataType::PingIota,
+                DataValue::SignedNumber(iota_ping.into()),
+            );
 
         self.send_message(&response).await;
     }
@@ -289,7 +328,10 @@ impl ClientConnection {
 
     /// Handle call invite
     async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
-        let receiver_id: i128 = cv.get_data(DataType::ReceiverId).as_signed_number().unwrap_or(0);
+        let receiver_id: i128 = cv
+            .get_data(DataType::ReceiverId)
+            .as_signed_number()
+            .unwrap_or(0);
         if receiver_id == 0 {
             self.send_error_response(cv.get_id(), CommunicationType::ErrorNoUserId)
                 .await;
@@ -334,9 +376,15 @@ impl ClientConnection {
                 // User is offline - send push notification for call invite
                 let push_cv = CommunicationValue::new(CommunicationType::PushNotification)
                     .with_receiver(receiver_id as u64)
-                    .add_typed_default(DataType::SenderId, DataValue::SignedNumber(sender_id.into()))
+                    .add_typed_default(
+                        DataType::SenderId,
+                        DataValue::SignedNumber(sender_id.into()),
+                    )
                     .add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string()))
-                    .add_typed_default(DataType::Notifications, DataValue::Str("call_invite".to_string()));
+                    .add_typed_default(
+                        DataType::Notifications,
+                        DataValue::Str("call_invite".to_string()),
+                    );
 
                 let omega_conn = get_omega_connection();
                 // Send fire-and-forget, don't await to avoid blocking
@@ -346,7 +394,10 @@ impl ClientConnection {
 
                 let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound)
                     .with_id(cv.get_id())
-                    .add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into()));
+                    .add_typed_default(
+                        DataType::ReceiverId,
+                        DataValue::SignedNumber(receiver_id.into()),
+                    );
                 self.send_message(&error_cv).await;
                 return;
             }
@@ -359,10 +410,19 @@ impl ClientConnection {
         let forward = CommunicationValue::new(CommunicationType::CallInvite)
             .with_receiver(receiver_id as u64)
             .with_sender(sender_id as u64)
-            .add_typed_default(DataType::CallSecret, cv.get_data(DataType::CallSecret).clone())
+            .add_typed_default(
+                DataType::CallSecret,
+                cv.get_data(DataType::CallSecret).clone(),
+            )
             .add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string()))
-            .add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into()))
+            .add_typed_default(
-            .add_typed_default(DataType::SenderId, DataValue::SignedNumber(sender_id.into()));
+                DataType::ReceiverId,
+                DataValue::SignedNumber(receiver_id.into()),
+            )
+            .add_typed_default(
+                DataType::SenderId,
+                DataValue::SignedNumber(sender_id.into()),
+            );
 
         target_rho.message_to_client(forward).await;
 
@@ -461,8 +521,14 @@ impl ClientConnection {
                 .await;
             return;
         };
-        let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0);
+        let user_id = cv
-        let untill = cv.get_data(DataType::Untill).as_signed_number().unwrap_or(0);
+            .get_data(DataType::UserId)
+            .as_signed_number()
+            .unwrap_or(0);
+        let untill = cv
+            .get_data(DataType::Untill)
+            .as_signed_number()
+            .unwrap_or(0);
 
         let Some(call) = call_manager::get_call(call_id).await else {
             self.send_error_response(cv.get_id(), CommunicationType::ErrorNotFound)
@@ -489,7 +555,10 @@ impl ClientConnection {
                 .await;
             return;
         };
-        let user_id = cv.get_data(DataType::UserId).as_signed_number().unwrap_or(0);
+        let user_id = cv
+            .get_data(DataType::UserId)
+            .as_signed_number()
+            .unwrap_or(0);
 
         let Some(call) = call_manager::get_call(call_id).await else {
             self.send_error_response(cv.get_id(), CommunicationType::ErrorNotFound)
@@ -546,7 +615,10 @@ impl ClientConnection {
                                 let response =
                                     CommunicationValue::new(CommunicationType::LoadTxtRecord)
                                         .with_id(cv.get_id())
-                                        .add_typed_default(DataType::Content, DataValue::Str(s.to_string()));
+                                        .add_typed_default(
+                                            DataType::Content,
+                                            DataValue::Str(s.to_string()),
+                                        );
                                 self.send_message(&response).await;
                                 return;
                             }
@@ -579,7 +651,10 @@ impl ClientConnection {
         );
 
         if cv.is_type(CommunicationType::AddConversation)
-            && cv.get_data(DataType::ChatPartnerId).as_signed_number().is_none()
+            && cv
+                .get_data(DataType::ChatPartnerId)
+                .as_signed_number()
+                .is_none()
         {
             let chat_partner_name = cv
                 .get_data(DataType::ChatPartnerName)
@@ -600,7 +675,10 @@ impl ClientConnection {
                 .await_response(
                     &CommunicationValue::new(CommunicationType::GetUserData)
                         .with_id(cv.clone().get_id())
-                        .add_typed_default(DataType::Username, DataValue::Str(chat_partner_name.clone())),
+                        .add_typed_default(
+                            DataType::Username,
+                            DataValue::Str(chat_partner_name.clone()),
+                        ),
                     Some(Duration::from_secs(20)),
                 )
                 .await;
@@ -665,7 +743,10 @@ impl ClientConnection {
             );
             let error_cv = CommunicationValue::new(CommunicationType::ErrorNoIota)
                 .with_id(msg_id)
-                .add_typed_default(DataType::UserId, DataValue::SignedNumber(sender_user_id.into()));
+                .add_typed_default(
+                    DataType::UserId,
+                    DataValue::SignedNumber(sender_user_id.into()),
+                );
             self.send_message(&error_cv).await;
         }
     }

src/rho/connection.rs

@@ -1,15 +1,15 @@
 use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataTypeId, DataValue};
-use mtp::transport::{Receiver, Sender};
+use mtp::host::{Receiver, Sender};
 use rand::{Rng, distributions::Alphanumeric};
 use std::{collections::BTreeMap, collections::HashMap, sync::Arc, time::Duration};
 use tokio::sync::RwLock;
 use uuid::Uuid;
 
-use base64::engine::general_purpose::STANDARD as BASE64_STD;
 use base64::Engine as _;
+use base64::engine::general_purpose::STANDARD as BASE64_STD;
 use mtp::crypto::{
-    encrypt_for, EncryptionType, KemPublicKey, PublicKeyBundle, SignaturePqPublicKey,
+    EncryptionType, KemPublicKey, PublicKeyBundle, SignaturePqPublicKey, SignaturePublicKey,
-    SignaturePublicKey,
+    encrypt_for,
 };
 
 use crate::{
@@ -179,15 +179,11 @@ impl GeneralConnection {
                 .unwrap(),
             );
 
-            let our_pk = BASE64_STD.encode(
+            let our_pk =
-                get_keyring().public_key_bundle().kem_public_key.as_bytes(),
+                BASE64_STD.encode(get_keyring().public_key_bundle().kem_public_key.as_bytes());
-            );
             let response = CommunicationValue::new(CommunicationType::AppChallenge)
                 .with_id(cv.get_id())
-                .add_typed_default(
+                .add_typed_default(DataType::PublicKey, DataValue::Str(our_pk))
-                    DataType::PublicKey,
-                    DataValue::Str(our_pk),
-                )
                 .add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge));
 
             log_cv_out!(response);
@@ -267,15 +263,11 @@ impl GeneralConnection {
                 .unwrap(),
             );
 
-            let our_pk = BASE64_STD.encode(
+            let our_pk =
-                get_keyring().public_key_bundle().kem_public_key.as_bytes(),
+                BASE64_STD.encode(get_keyring().public_key_bundle().kem_public_key.as_bytes());
-            );
             let response = CommunicationValue::new(CommunicationType::Challenge)
                 .with_id(cv.get_id())
-                .add_typed_default(
+                .add_typed_default(DataType::PublicKey, DataValue::Str(our_pk))
-                    DataType::PublicKey,
-                    DataValue::Str(our_pk),
-                )
                 .add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge));
 
             log_cv_out!(response);
@@ -370,16 +362,12 @@ impl GeneralConnection {
                 CommunicationType::Challenge
             };
 
-            let our_pk = BASE64_STD.encode(
+            let our_pk =
-                get_keyring().public_key_bundle().kem_public_key.as_bytes(),
+                BASE64_STD.encode(get_keyring().public_key_bundle().kem_public_key.as_bytes());
-            );
             let response = CommunicationValue::new(challenge_type)
                 .with_id(cv.get_id())
                 .with_receiver(*self.session_id.read().await)
-                .add_typed_default(
+                .add_typed_default(DataType::PublicKey, DataValue::Str(our_pk))
-                    DataType::PublicKey,
-                    DataValue::Str(our_pk),
-                )
                 .add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge));
 
             log_cv_out!(response);
@@ -440,7 +428,10 @@ impl GeneralConnection {
         match kind {
             ConnectionKind::Client => {
                 let notify = CommunicationValue::new(CommunicationType::UserConnected)
-                    .add_typed_default(DataType::UserId, DataValue::SignedNumber((id as i64).into()));
+                    .add_typed_default(
+                        DataType::UserId,
+                        DataValue::SignedNumber((id as i64).into()),
+                    );
                 get_omega_connection().send_message(&notify).await;
 
                 let user_id = id as i64;
@@ -451,7 +442,10 @@ impl GeneralConnection {
 
                 if rho.is_none() {
                     let get_user_msg = CommunicationValue::new(CommunicationType::GetUserData)
-                        .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()));
+                        .add_typed_default(
+                            DataType::UserId,
+                            DataValue::SignedNumber(user_id.into()),
+                        );
 
                     if let Ok(user_data_cv) = get_omega_connection()
                         .await_response(&get_user_msg, Some(Duration::from_secs(20)))
@@ -475,8 +469,14 @@ impl GeneralConnection {
                 if let Some(rho_conn) = rho {
                     let session_id = *self.session_id.read().await as i64;
                     let iota_msg = CommunicationValue::new(CommunicationType::ClientConnected)
-                        .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
+                        .add_typed_default(
-                        .add_typed_default(DataType::SessionId, DataValue::SignedNumber(session_id.into()));
+                            DataType::UserId,
+                            DataValue::SignedNumber(user_id.into()),
+                        )
+                        .add_typed_default(
+                            DataType::SessionId,
+                            DataValue::SignedNumber(session_id.into()),
+                        );
 
                     if let Ok(resp) = rho_conn
                         .get_iota_connection()
@@ -526,7 +526,9 @@ impl GeneralConnection {
                                             if timeout > 0 {
                                                 base_call_map.insert(
                                                     DataType::Timeout.to_id(&tm),
-                                                    DataValue::SignedNumber((timeout as i64).into()),
+                                                    DataValue::SignedNumber(
+                                                        (timeout as i64).into(),
+                                                    ),
                                                 );
                                             }
 
@@ -578,7 +580,8 @@ impl GeneralConnection {
                                             if let Some(DataValue::SignedNumber(id)) =
                                                 contact_map.get(&DataType::UserId.to_id(&tm))
                                             {
-                                                if let Some(call_list) = invites.get(&(*id as i64)) {
+                                                if let Some(call_list) = invites.get(&(*id as i64))
+                                                {
                                                     contact_map.insert(
                                                         DataType::Calls.to_id(&tm),
                                                         DataValue::Array(call_list.clone()),
@@ -592,8 +595,10 @@ impl GeneralConnection {
                                         }
                                     }
 
-                                    ident_resp = ident_resp
+                                    ident_resp = ident_resp.add_typed_default(
-                                        .add_typed_default(DataType::Calls, DataValue::Array(global_calls));
+                                        DataType::Calls,
+                                        DataValue::Array(global_calls),
+                                    );
                                     DataValue::Array(new_contacts)
                                 } else {
                                     v.clone()
@@ -672,7 +677,10 @@ impl GeneralConnection {
 
                 if rho.is_none() {
                     let get_user_msg = CommunicationValue::new(CommunicationType::GetUserData)
-                        .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()));
+                        .add_typed_default(
+                            DataType::UserId,
+                            DataValue::SignedNumber(user_id.into()),
+                        );
 
                     if let Ok(user_data_cv) = get_omega_connection()
                         .await_response(&get_user_msg, Some(Duration::from_secs(20)))

src/rho/iota_connection.rs

@@ -14,13 +14,13 @@ use mtp::codec::DataType;
 use mtp::codec::DataTypeId;
 use mtp::codec::DataValue;
 use mtp::codec::TypeMap;
-use mtp::transport::Receiver;
+use mtp::crypto::KemPublicKey;
-use mtp::transport::Sender;
+use mtp::host::Receiver;
+use mtp::host::Sender;
 use std::collections::BTreeMap;
 use std::{collections::HashMap, sync::Arc, time::Duration};
 use tokio::sync::RwLock;
 use tokio::sync::mpsc;
-use mtp::crypto::KemPublicKey;
 
 use super::{rho_connection::RhoConnection, rho_manager};
 use crate::omega::omega_connection::OmegaConnection;
@@ -265,7 +265,10 @@ impl IotaConnection {
                 if let Ok(uid) = k.parse::<i128>() {
                     map.insert(DataType::UserId.to_id(&tm), DataValue::SignedNumber(uid));
                 }
-                map.insert(DataType::LastPing.to_id(&tm), DataValue::SignedNumber(v.into()));
+                map.insert(
+                    DataType::LastPing.to_id(&tm),
+                    DataValue::SignedNumber(v.into()),
+                );
                 DataValue::container_from_map(&map)
             })
             .collect();
@@ -368,12 +371,20 @@ impl IotaConnection {
 
                 // Build base call container
                 let mut base_call_map: BTreeMap<DataTypeId, DataValue> = BTreeMap::new();
-                base_call_map.insert(DataType::CallId.to_id(&tm), DataValue::Str(call.call_id.to_string()));
+                base_call_map.insert(
-                base_call_map.insert(DataType::CallMembers.to_id(&tm), DataValue::Array(member_ids));
+                    DataType::CallId.to_id(&tm),
+                    DataValue::Str(call.call_id.to_string()),
+                );
+                base_call_map.insert(
+                    DataType::CallMembers.to_id(&tm),
+                    DataValue::Array(member_ids),
+                );
 
                 if timeout > 0 {
-                    base_call_map
+                    base_call_map.insert(
-                        .insert(DataType::Timeout.to_id(&tm), DataValue::SignedNumber(timeout.into()));
+                        DataType::Timeout.to_id(&tm),
+                        DataValue::SignedNumber(timeout.into()),
+                    );
                 }
 
                 if admin {
@@ -394,8 +405,10 @@ impl IotaConnection {
 
                     // Add secret if it exists for this pairing
                     if let Some(secret) = call.secrets.read().await.get(&(member_id, user_id)) {
-                        contact_call_map
+                        contact_call_map.insert(
-                            .insert(DataType::CallSecret.to_id(&tm), DataValue::Str(secret.clone()));
+                            DataType::CallSecret.to_id(&tm),
+                            DataValue::Str(secret.clone()),
+                        );
                     }
 
                     invites
@@ -423,14 +436,18 @@ impl IotaConnection {
                         let mut user_map: BTreeMap<DataTypeId, DataValue> =
                             entries.iter().cloned().collect();
 
-                        if let Some(DataValue::SignedNumber(id)) = user_map.get(&DataType::UserId.to_id(&tm)) {
+                        if let Some(DataValue::SignedNumber(id)) =
+                            user_map.get(&DataType::UserId.to_id(&tm))
+                        {
                             interested_ids.push(*id as i64);
 
                             if let Some(call_list) = invites.get(&(*id as i64))
                                 && !call_list.is_empty()
                             {
-                                user_map
+                                user_map.insert(
-                                    .insert(DataType::Calls.to_id(&tm), DataValue::Array(call_list.clone()));
+                                    DataType::Calls.to_id(&tm),
+                                    DataValue::Array(call_list.clone()),
+                                );
                             }
                         }
 

src/rho/server.rs

@@ -1,24 +1,53 @@
 use std::net::{IpAddr, Ipv4Addr};
 use std::time::Duration;
 
+use crate::load_keyring;
 use crate::{
     log,
     rho::connection::GeneralConnection,
     util::{file_util::load_file_vec, logger::PrintType},
 };
-use mtp::transport::{Host, Policy, SendMode, host};
+use mtp::crypto::PublicKeyBundle;
+use mtp::host::{AuthenticationPolicy, Host, HostConfig, Policy, SendMode};
+
+pub async fn get_by_connector_id(
+    _client_id: u64,
+    description: Option<String>,
+) -> Option<PublicKeyBundle> {
+    if let Some(description) = description {
+        if description == "iota" {
+            todo!()
+        } else if description == "client" {
+            todo!()
+        } else if description == "anonymous" {
+            todo!()
+        } else if description == "app" {
+            todo!()
+        }
+    }
+    None
+}
+pub async fn complete_register(_pub_key: PublicKeyBundle, description: Option<String>) -> u64 {
+    if let Some(description) = description {
+        if description == "iota" {
+            todo!()
+        }
+    }
+    0
+}
 
 pub async fn start(port: u16) -> Result<(), Box<dyn std::error::Error>> {
     let cert_pem = load_file_vec("certs", "cert.pem").expect("Error loading Pemfile");
 
     let key_pem = load_file_vec("certs", "key.pem").expect("Error loading Keyfile");
 
-    let mut host: Host = host(
+    let host_config = HostConfig::new(
-        IpAddr::V4(Ipv4Addr::UNSPECIFIED),
+        IpAddr::from(Ipv4Addr::new(0, 0, 0, 0)),
         port,
         cert_pem,
         key_pem,
-        Policy {
+    )
+    .with_policy(Policy {
         send_mode: SendMode::SingleStreamPerMessage,
         max_message_size: 1_000_000_000,
         close_frame_len: u32::MAX,
@@ -33,14 +62,20 @@ pub async fn start(port: u16) -> Result<(), Box<dyn std::error::Error>> {
         max_transient_recv_errors: 20,
         transient_recv_backoff: Duration::from_millis(100),
         receiver_queue_capacity: 1000,
-        },
+    })
+    .with_authentication(
+        load_keyring(),
+        Box::new(|user_id, description| Box::pin(get_by_connector_id(user_id, description))),
+        Box::new(|pub_key, description| Box::pin(complete_register(pub_key, description))),
     )
-    .await?;
+    .with_authentication_policy(AuthenticationPolicy::AllowAuthentication);
+
+    let mut host: Host = Host::new(host_config).await?;
     log!(0, PrintType::General, "Server listening on port {}", port);
 
-    while let Some((sender, receiver)) = host.next().await {
+    while let Ok(Some(conn)) = host.accept().await {
         tokio::spawn(async move {
-            let conn = GeneralConnection::new(sender, receiver);
+            let conn = GeneralConnection::new(conn.sender, conn.receiver);
             conn.handle().await;
         });
     }