tensamin/omikron
Watch
6
0
Fork
You've already forked omikron
 0
Code Issues Pull requests 3 Projects Releases Packages Wiki Activity Actions

Initial migration to MTP [Broken]

Browse source
This commit is contained in:
Alex Emmet 2026-06-28 13:52:58 +02:00
commit 7ea84cc54b
15 changed files with 965 additions and 1190 deletions
Download patch file Download diff file Expand all files Collapse all files

81
src/rho/app_connection.rs
View file

@ -4,17 +4,18 @@ use crate::rho::connection::GeneralConnection;
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 std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::RwLock;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use ttp_native::{Receiver, Sender};
use uuid::Uuid;
pub struct AppConnection {
pub user_id: u64,
pub app_identifier: String,
pub app_session: Uuid,
pub client_version: String,
pub sender: Arc<Sender>,
pub receiver: Arc<Receiver>,
@ -36,6 +37,7 @@ impl AppConnection {
user_id: user_id,
app_identifier: general.app_identifier.read().await.clone().unwrap(),
app_session: general.app_session.read().await.clone().unwrap(),
client_version: general.client_version.read().await.clone(),
})
}
pub fn start(self: Arc<Self>) {
@ -52,7 +54,7 @@ impl AppConnection {
tokio::time::sleep(Duration::from_millis(50)).await;
if self_clone2.get_rho_connection().await.is_none() {
self_clone2
.send_error_response(0, CommunicationType::error_no_iota)
.send_error_response(0, CommunicationType::ErrorNoIota)
.await;
}
});
@ -84,7 +86,7 @@ impl AppConnection {
);
return;
}
if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) {
if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) {
log_cv_out!(PrintType::App, &cv);
}
let _ = self.sender.send(&cv).await;
@ -93,41 +95,41 @@ impl AppConnection {
/// Handle incoming message from app
pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
tokio::spawn(async move {
if cv.is_type(CommunicationType::ping) {
if cv.is_type(CommunicationType::Ping) {
self.handle_ping(cv).await;
return;
}
log_cv_in!(PrintType::App, cv);
if cv.is_type(CommunicationType::get_user_data) {
if cv.is_type(CommunicationType::GetUserData) {
if let Some(anonymous) = {
if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
if let Some(user_id) = cv.get_data(DataType::UserId).as_number() {
anonymous_manager::get_anonymous_user(user_id as u64).await
} else if let Some(username) = cv.get_data(DataTypes::username).as_str() {
} else if let Some(username) = cv.get_data(DataType::Username).as_str() {
anonymous_manager::get_anonymous_user_by_name(username.to_string()).await
} else {
None
}
} {
let response = CommunicationValue::new(CommunicationType::get_user_data)
let response = CommunicationValue::new(CommunicationType::GetUserData)
.with_id(cv.get_id())
.add_data(
DataTypes::username,
.add_typed_default(
DataType::Username,
DataValue::Str(anonymous.get_user_name().await),
)
.add_data(
DataTypes::user_id,
DataValue::Number(anonymous.get_user_id() as i64),
.add_typed_default(
DataType::UserId,
DataValue::SignedNumber(anonymous.get_user_id().into()),
)
.add_data(
DataTypes::display,
.add_typed_default(
DataType::Display,
DataValue::Str(anonymous.get_display_name().await),
)
.add_data(
DataTypes::avatar,
.add_typed_default(
DataType::Avatar,
DataValue::Str(anonymous.get_avatar().await),
)
.add_data(DataTypes::user_state, DataValue::Str("online".to_string()));
.add_typed_default(DataType::UserState, DataValue::Str("online".to_string()));
self.send_message(&response).await;
@ -135,8 +137,8 @@ impl AppConnection {
}
}
if cv.is_type(CommunicationType::get_user_data)
|| cv.is_type(CommunicationType::get_iota_data)
if cv.is_type(CommunicationType::GetUserData)
|| cv.is_type(CommunicationType::GetIotaData)
{
let sender = self.get_user_id().await;
self.handle_omega_forward(cv.with_sender(sender as u64))
@ -162,13 +164,13 @@ impl AppConnection {
/// Handle ping message
async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) {
// Update our ping if provided
if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) {
let current = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis();
let mut ping_guard = self.ping.write().await;
*ping_guard = current as i64 - last_ping;
*ping_guard = (current as i128 - *last_ping) as i64;
}
// Get Iota ping from RhoConnection
@ -179,9 +181,9 @@ impl AppConnection {
};
// Send pong response
let response = CommunicationValue::new(CommunicationType::pong)
let response = CommunicationValue::new(CommunicationType::Pong)
.with_id(cv.get_id())
.add_data(DataTypes::ping_iota, DataValue::Number(iota_ping));
.add_typed_default(DataType::PingIota, DataValue::SignedNumber(iota_ping.into()));
self.send_message(&response).await;
}
@ -202,14 +204,11 @@ impl AppConnection {
cv.get_receiver()
);
if cv.is_type(CommunicationType::add_conversation)
&& cv
.get_data(DataTypes::chat_partner_id)
.as_number()
.is_none()
if cv.is_type(CommunicationType::AddConversation)
&& cv.get_data(DataType::ChatPartnerId).as_number().is_none()
{
let chat_partner_name = cv
.get_data(DataTypes::chat_partner_name)
.get_data(DataType::ChatPartnerName)
.as_str()
.unwrap_or("")
.to_string();
@ -218,17 +217,17 @@ impl AppConnection {
.await
.is_some()
{
self.send_error_response(cv.get_id(), CommunicationType::error_anonymous)
self.send_error_response(cv.get_id(), CommunicationType::ErrorAnonymous)
.await;
return;
}
let load_uuid_response = get_omega_connection()
.await_response(
&CommunicationValue::new(CommunicationType::get_user_data)
&CommunicationValue::new(CommunicationType::GetUserData)
.with_id(cv.clone().get_id())
.add_data(
DataTypes::username,
.add_typed_default(
DataType::Username,
DataValue::Str(chat_partner_name.clone()),
),
Some(Duration::from_secs(20)),
@ -236,7 +235,7 @@ impl AppConnection {
.await;
let chat_partner_id = {
if let Ok(load_uuid_response) = load_uuid_response {
load_uuid_response.get_data(DataTypes::user_id).clone()
load_uuid_response.get_data(DataType::UserId).clone()
} else {
DataValue::Null
}
@ -255,7 +254,7 @@ impl AppConnection {
let updated_cv = cv
.with_sender(sender_user_id as u64)
.add_data(DataTypes::chat_partner_id, chat_partner_id);
.add_typed_default(DataType::ChatPartnerId, chat_partner_id);
rho_conn.message_to_iota(updated_cv).await;
} else {
log_err!(
@ -293,9 +292,12 @@ impl AppConnection {
msg_type,
msg_id
);
let error_cv = CommunicationValue::new(CommunicationType::error_no_iota)
let error_cv = CommunicationValue::new(CommunicationType::ErrorNoIota)
.with_id(msg_id)
.add_data(DataTypes::user_id, DataValue::Number(sender_user_id as i64));
.add_typed_default(
DataType::UserId,
DataValue::SignedNumber(sender_user_id.into()),
);
self.send_message(&error_cv).await;
}
}
@ -336,6 +338,7 @@ impl Clone for AppConnection {
user_id: self.user_id,
app_identifier: self.app_identifier.clone(),
app_session: self.app_session,
client_version: self.client_version.clone(),
ping: Arc::clone(&self.ping),
pub_key: Arc::clone(&self.pub_key),
rho_connection: Arc::clone(&self.rho_connection),

319
src/rho/client_connection.rs
View file

@ -6,17 +6,18 @@ use crate::rho::{rho_connection::RhoConnection, rho_manager};
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 std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::RwLock;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use ttp_native::{Receiver, Sender};
use uuid::Uuid;
pub struct ClientConnection {
pub user_id: u64,
pub session_id: u64,
pub client_version: String,
pub sender: Arc<Sender>,
pub receiver: Arc<Receiver>,
@ -39,6 +40,7 @@ impl ClientConnection {
receiver: general.receiver.clone(),
user_id: user_id,
session_id: general.session_id.read().await.clone(),
client_version: general.client_version.read().await.clone(),
})
}
pub fn start(self: Arc<Self>) {
@ -55,7 +57,7 @@ impl ClientConnection {
tokio::time::sleep(Duration::from_millis(50)).await;
if self_clone2.get_rho_connection().await.is_none() {
self_clone2
.send_error_response(0, CommunicationType::error_no_iota)
.send_error_response(0, CommunicationType::ErrorNoIota)
.await;
}
});
@ -86,7 +88,7 @@ impl ClientConnection {
);
return;
}
if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) {
if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) {
log_cv_out!(PrintType::Client, &cv);
}
let _ = self.sender.send(&cv).await;
@ -95,7 +97,7 @@ impl ClientConnection {
/// Handle incoming message from client
pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
tokio::spawn(async move {
if cv.is_type(CommunicationType::ping) {
if cv.is_type(CommunicationType::Ping) {
self.handle_ping(cv).await;
return;
}
@ -104,75 +106,63 @@ impl ClientConnection {
let mut cv = cv;
// Handle client status changes
if cv.is_type(CommunicationType::client_changed) {
if cv.is_type(CommunicationType::ClientChanged) {
self.handle_client_changed(cv).await;
return;
}
// Handle call invites
if cv.is_type(CommunicationType::call_invite) {
if cv.is_type(CommunicationType::CallInvite) {
self.handle_call_invite(cv).await;
return;
}
// Handle get call requests
if cv.is_type(CommunicationType::call_token) {
if cv.is_type(CommunicationType::CallToken) {
self.handle_get_call(cv).await;
return;
}
if cv.is_type(CommunicationType::call_data) {
if cv.is_type(CommunicationType::CallData) {
self.handle_get_call_data(cv).await;
return;
}
if cv.is_type(CommunicationType::call_disconnect_user) {
if cv.is_type(CommunicationType::CallDisconnectUser) {
self.handle_call_disconnect_user(cv).await;
return;
}
if cv.is_type(CommunicationType::call_timeout_user) {
if cv.is_type(CommunicationType::CallTimeoutUser) {
self.handle_call_timeout_user(cv).await;
return;
}
if cv.is_type(CommunicationType::call_set_anonymous_joining) {
if cv.is_type(CommunicationType::CallSetAnonymousJoining) {
self.handle_call_set_anonymous_joining(cv).await;
return;
}
if cv.is_type(CommunicationType::load_txt_record) {
if cv.is_type(CommunicationType::LoadTxtRecord) {
self.handle_load_txt_record(cv).await;
return;
}
if cv.is_type(CommunicationType::get_user_data) {
if cv.is_type(CommunicationType::GetUserData) {
if let Some(anonymous) = {
if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
if let Some(user_id) = cv.get_data(DataType::UserId).as_signed_number() {
anonymous_manager::get_anonymous_user(user_id as u64).await
} else if let Some(username) = cv.get_data(DataTypes::username).as_str() {
} else if let Some(username) = cv.get_data(DataType::Username).as_str() {
anonymous_manager::get_anonymous_user_by_name(username.to_string()).await
} else {
None
}
} {
let response = CommunicationValue::new(CommunicationType::get_user_data)
let response = CommunicationValue::new(CommunicationType::GetUserData)
.with_id(cv.get_id())
.add_data(
DataTypes::username,
DataValue::Str(anonymous.get_user_name().await),
)
.add_data(
DataTypes::user_id,
DataValue::Number(anonymous.get_user_id() as i64),
)
.add_data(
DataTypes::display,
DataValue::Str(anonymous.get_display_name().await),
)
.add_data(
DataTypes::avatar,
DataValue::Str(anonymous.get_avatar().await),
)
.add_data(DataTypes::user_state, DataValue::Str("online".to_string()));
.add_typed_default(DataType::Username, DataValue::Str(anonymous.get_user_name().await))
.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;
@ -180,12 +170,12 @@ impl ClientConnection {
}
}
if cv.is_type(CommunicationType::change_user_data)
|| cv.is_type(CommunicationType::read_notification)
|| cv.is_type(CommunicationType::get_notifications)
|| cv.is_type(CommunicationType::get_user_data)
|| cv.is_type(CommunicationType::get_iota_data)
|| cv.is_type(CommunicationType::delete_user)
if cv.is_type(CommunicationType::ChangeUserData)
|| cv.is_type(CommunicationType::ReadNotification)
|| cv.is_type(CommunicationType::GetNotifications)
|| cv.is_type(CommunicationType::GetUserData)
|| cv.is_type(CommunicationType::GetIotaData)
|| cv.is_type(CommunicationType::DeleteUser)
{
let sender = self.get_user_id().await;
self.handle_omega_forward(cv.with_sender(sender as u64))
@ -193,72 +183,48 @@ impl ClientConnection {
return;
}
let is_per_device_settings = cv.is_type(CommunicationType::settings_save)
|| cv.is_type(CommunicationType::settings_load)
|| cv.is_type(CommunicationType::settings_list);
let is_global_settings = cv.is_type(CommunicationType::global_settings_save)
|| cv.is_type(CommunicationType::global_settings_load);
let is_per_device_settings = cv.is_type(CommunicationType::SettingsSave)
|| cv.is_type(CommunicationType::SettingsLoad)
|| cv.is_type(CommunicationType::SettingsList);
let is_global_settings = cv.is_type(CommunicationType::GlobalSettingsSave)
|| cv.is_type(CommunicationType::GlobalSettingsLoad);
if is_per_device_settings || is_global_settings {
let expected_session_id = self.session_id as i64;
let session_id = cv.get_data(DataTypes::session_id).as_number();
let expected_session_id: i128 = self.session_id.into();
let session_id = cv.get_data(DataType::SessionId).as_signed_number();
if is_per_device_settings {
let Some(session_id) = session_id else {
let response =
CommunicationValue::new(CommunicationType::error_invalid_data)
.with_id(cv.get_id())
.with_receiver(self.user_id)
.add_data(
DataTypes::message,
DataValue::Str("Missing session_id".to_string()),
)
.add_data(
DataTypes::session_id,
DataValue::Number(expected_session_id),
);
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(DataType::SessionId, DataValue::SignedNumber(expected_session_id));
self.send_message(&response).await;
return;
};
if session_id != expected_session_id {
let response =
CommunicationValue::new(CommunicationType::error_invalid_data)
.with_id(cv.get_id())
.with_receiver(self.user_id)
.add_data(
DataTypes::message,
DataValue::Str("session_id mismatch".to_string()),
)
.add_data(
DataTypes::session_id,
DataValue::Number(expected_session_id),
);
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(DataType::SessionId, DataValue::SignedNumber(expected_session_id));
self.send_message(&response).await;
return;
}
} else if let Some(session_id) = session_id {
if session_id != expected_session_id {
let response =
CommunicationValue::new(CommunicationType::error_invalid_data)
.with_id(cv.get_id())
.with_receiver(self.user_id)
.add_data(
DataTypes::message,
DataValue::Str("session_id mismatch".to_string()),
)
.add_data(
DataTypes::session_id,
DataValue::Number(expected_session_id),
);
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(DataType::SessionId, DataValue::SignedNumber(expected_session_id));
self.send_message(&response).await;
return;
}
} else {
cv = cv.add_data(
DataTypes::session_id,
DataValue::Number(expected_session_id),
);
cv = cv.add_typed_default(DataType::SessionId, DataValue::SignedNumber(expected_session_id));
}
}
@ -281,13 +247,13 @@ impl ClientConnection {
/// Handle ping message
async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) {
// Update our ping if provided
if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) {
let current = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis();
let mut ping_guard = self.ping.write().await;
*ping_guard = current as i64 - last_ping;
*ping_guard = (current as i128 - *last_ping) as i64;
}
// Get Iota ping from RhoConnection
@ -298,9 +264,9 @@ impl ClientConnection {
};
// Send pong response
let response = CommunicationValue::new(CommunicationType::pong)
let response = CommunicationValue::new(CommunicationType::Pong)
.with_id(cv.get_id())
.add_data(DataTypes::ping_iota, DataValue::Number(iota_ping));
.add_typed_default(DataType::PingIota, DataValue::SignedNumber(iota_ping.into()));
self.send_message(&response).await;
}
@ -308,7 +274,7 @@ impl ClientConnection {
/// Handle client status change
async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) {
let user_id = self.get_user_id().await;
if let DataValue::Str(status_str) = cv.get_data(DataTypes::user_state) {
if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) {
let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online);
if let Some(rho_conn) = self.get_rho_connection().await {
OmegaConnection::client_changed(
@ -323,57 +289,54 @@ impl ClientConnection {
/// Handle call invite
async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
let receiver_id: i64 = cv.get_data(DataTypes::receiver_id).as_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::error_no_user_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorNoUserId)
.await;
return;
}
let call_id = match cv.get_data(DataTypes::call_id) {
let call_id = match cv.get_data(DataType::CallId) {
DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) {
Ok(id) => id,
Err(_) => {
self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId)
.await;
return;
}
},
_ => {
self.send_error_response(cv.get_id(), CommunicationType::error_no_call_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorNoCallId)
.await;
return;
}
};
let secret = cv
.get_data(DataTypes::call_secret)
.get_data(DataType::CallSecret)
.as_str()
.map(|s| s.to_string());
let invited =
call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await;
if !invited {
self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId)
.await;
return;
}
// Find target RhoConnection
let target_rho = match rho_manager::get_rho_con_for_user(receiver_id).await {
let target_rho = match rho_manager::get_rho_con_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::push_notification)
let push_cv = CommunicationValue::new(CommunicationType::PushNotification)
.with_receiver(receiver_id as u64)
.add_data(DataTypes::sender_id, DataValue::Number(sender_id as i64))
.add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
.add_data(
DataTypes::notifications,
DataValue::Str("call_invite".to_string()),
);
.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 = get_omega_connection();
// Send fire-and-forget, don't await to avoid blocking
@ -381,9 +344,9 @@ impl ClientConnection {
let _ = omega_conn.send_message(&push_cv).await;
});
let error_cv = CommunicationValue::new(CommunicationType::error_not_found)
let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound)
.with_id(cv.get_id())
.add_data(DataTypes::receiver_id, DataValue::Number(receiver_id));
.add_typed_default(DataType::ReceiverId, DataValue::SignedNumber(receiver_id.into()));
self.send_message(&error_cv).await;
return;
}
@ -393,20 +356,17 @@ impl ClientConnection {
let sender_id = self.get_user_id().await as i64;
// Create and send call distribution message
let forward = CommunicationValue::new(CommunicationType::call_invite)
let forward = CommunicationValue::new(CommunicationType::CallInvite)
.with_receiver(receiver_id as u64)
.with_sender(sender_id as u64)
.add_data(
DataTypes::call_secret,
cv.get_data(DataTypes::call_secret).clone(),
)
.add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
.add_data(DataTypes::receiver_id, DataValue::Number(receiver_id))
.add_data(DataTypes::sender_id, DataValue::Number(sender_id));
.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(DataType::SenderId, DataValue::SignedNumber(sender_id.into()));
target_rho.message_to_client(forward).await;
let response = CommunicationValue::new(CommunicationType::success).with_id(cv.get_id());
let response = CommunicationValue::new(CommunicationType::Success).with_id(cv.get_id());
self.send_message(&response).await;
}
@ -414,32 +374,32 @@ impl ClientConnection {
async fn handle_get_call(self: Arc<Self>, cv: CommunicationValue) {
let user_id = self.get_user_id().await;
let call_id = match cv.get_data(DataTypes::call_id) {
let call_id = match cv.get_data(DataType::CallId) {
DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) {
Ok(id) => id,
Err(_) => {
self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId)
.await;
return;
}
},
_ => {
self.send_error_response(cv.get_id(), CommunicationType::error_no_call_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorNoCallId)
.await;
return;
}
};
if let Some(token) = call_manager::get_call_token(user_id, call_id).await {
let response = CommunicationValue::new(CommunicationType::call_token)
let response = CommunicationValue::new(CommunicationType::CallToken)
.with_id(cv.get_id())
.with_receiver(user_id as u64)
.add_data(DataTypes::call_token, DataValue::Str(token));
.add_typed_default(DataType::CallToken, DataValue::Str(token));
self.send_message(&response).await;
} else {
let error_cv = CommunicationValue::new(CommunicationType::error_no_call_id)
let error_cv = CommunicationValue::new(CommunicationType::ErrorNoCallId)
.with_id(cv.get_id())
.add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()));
.add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string()));
self.send_message(&error_cv).await;
return;
}
@ -447,17 +407,17 @@ impl ClientConnection {
async fn handle_get_call_data(self: Arc<Self>, cv: CommunicationValue) {
let user_id = self.get_user_id().await;
let call_id = match cv.get_data(DataTypes::call_id) {
let call_id = match cv.get_data(DataType::CallId) {
DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) {
Ok(id) => id,
Err(_) => {
self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId)
.await;
return;
}
},
_ => {
self.send_error_response(cv.get_id(), CommunicationType::error_no_call_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorNoCallId)
.await;
return;
}
@ -469,75 +429,75 @@ impl ClientConnection {
let members = call.members.read().await.clone();
for member in members {
if member.user_id == user_id {
user_ids.push(DataValue::Number(member.user_id as i64));
user_ids.push(DataValue::SignedNumber(member.user_id.into()));
}
}
let response = CommunicationValue::new(CommunicationType::call_data)
let response = CommunicationValue::new(CommunicationType::CallData)
.with_id(cv.get_id())
.with_receiver(user_id as u64)
.add_data(DataTypes::user_ids, DataValue::Array(user_ids));
.add_typed_default(DataType::UserIds, DataValue::Array(user_ids));
self.send_message(&response).await;
} else {
let error_cv = CommunicationValue::new(CommunicationType::error_invalid_user_id)
let error_cv = CommunicationValue::new(CommunicationType::ErrorInvalidUserId)
.with_id(cv.get_id())
.add_data(DataTypes::user_id, DataValue::Number(user_id as i64));
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()));
self.send_message(&error_cv).await;
return;
}
} else {
let error_cv = CommunicationValue::new(CommunicationType::error_not_found)
let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound)
.with_id(cv.get_id())
.add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()));
.add_typed_default(DataType::CallId, DataValue::Str(call_id.to_string()));
self.send_message(&error_cv).await;
return;
}
}
async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) {
let Ok(call_id) = Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or(""))
let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or(""))
else {
self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId)
.await;
return;
};
let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
let untill = cv.get_data(DataTypes::untill).as_number().unwrap_or(0);
let user_id = cv.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::error_not_found)
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(cv.get_id(), CommunicationType::error_invalid_user_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
return;
};
if caller.has_admin() {
let _ = call_util::remove_participant(call_id, user_id as u64).await;
if let Some(target) = call.get_caller(user_id as u64).await {
target.set_timeout(untill).await;
target.set_timeout(untill as i64).await;
}
}
}
async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) {
let Ok(call_id) = Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or(""))
let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or(""))
else {
self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId)
.await;
return;
};
let user_id = cv.get_data(DataTypes::user_id).as_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::error_not_found)
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(cv.get_id(), CommunicationType::error_invalid_user_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
return;
};
@ -546,13 +506,13 @@ impl ClientConnection {
}
}
async fn handle_call_set_anonymous_joining(self: Arc<Self>, cv: CommunicationValue) {
let Ok(call_id) = Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or(""))
let Ok(call_id) = Uuid::from_str(cv.get_data(DataType::CallId).as_str().unwrap_or(""))
else {
self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
self.send_error_response(cv.get_id(), CommunicationType::ErrorInvalidCallId)
.await;
return;
};
let enable = cv.get_data(DataTypes::enabled).as_bool().unwrap_or(true);
let enable = cv.get_data(DataType::Enabled).as_bool().unwrap_or(true);
let call = call_manager::get_call(call_id).await;
@ -565,19 +525,18 @@ impl ClientConnection {
}
short_link = call.get_short_link().await;
}
let mut response_cv =
CommunicationValue::new(CommunicationType::call_set_anonymous_joining)
.with_id(cv.get_id())
.add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
.add_data(DataTypes::enabled, DataValue::Bool(enable));
let mut response_cv = CommunicationValue::new(CommunicationType::CallSetAnonymousJoining)
.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 {
response_cv = response_cv.add_data(DataTypes::link, DataValue::Str(short_link));
response_cv = response_cv.add_typed_default(DataType::Link, DataValue::Str(short_link));
}
self.send_message(&response_cv).await;
}
async fn handle_load_txt_record(self: Arc<Self>, cv: CommunicationValue) {
if let Some(path) = cv.get_data(DataTypes::path).as_str() {
if let Some(path) = cv.get_data(DataType::Path).as_str() {
if let Ok(builder) = hickory_resolver::Resolver::builder_tokio() {
let resolver = builder.build();
if let Ok(lookup) = resolver.txt_lookup(path).await {
@ -585,12 +544,9 @@ impl ClientConnection {
for txt_data in record.txt_data() {
if let Ok(s) = std::str::from_utf8(txt_data) {
let response =
CommunicationValue::new(CommunicationType::load_txt_record)
CommunicationValue::new(CommunicationType::LoadTxtRecord)
.with_id(cv.get_id())
.add_data(
DataTypes::content,
DataValue::Str(s.to_string()),
);
.add_typed_default(DataType::Content, DataValue::Str(s.to_string()));
self.send_message(&response).await;
return;
}
@ -599,10 +555,10 @@ impl ClientConnection {
}
}
}
let path_data = cv.get_data(DataTypes::path).clone();
let error_cv = CommunicationValue::new(CommunicationType::error_not_found)
let path_data = cv.get_data(DataType::Path).clone();
let error_cv = CommunicationValue::new(CommunicationType::ErrorNotFound)
.with_id(cv.get_id())
.add_data(DataTypes::path, path_data);
.add_typed_default(DataType::Path, path_data);
self.send_message(&error_cv).await;
}
@ -622,14 +578,11 @@ impl ClientConnection {
cv.get_receiver()
);
if cv.is_type(CommunicationType::add_conversation)
&& cv
.get_data(DataTypes::chat_partner_id)
.as_number()
.is_none()
if cv.is_type(CommunicationType::AddConversation)
&& cv.get_data(DataType::ChatPartnerId).as_signed_number().is_none()
{
let chat_partner_name = cv
.get_data(DataTypes::chat_partner_name)
.get_data(DataType::ChatPartnerName)
.as_str()
.unwrap_or("")
.to_string();
@ -638,25 +591,22 @@ impl ClientConnection {
.await
.is_some()
{
self.send_error_response(cv.get_id(), CommunicationType::error_anonymous)
self.send_error_response(cv.get_id(), CommunicationType::ErrorAnonymous)
.await;
return;
}
let load_uuid_response = get_omega_connection()
.await_response(
&CommunicationValue::new(CommunicationType::get_user_data)
&CommunicationValue::new(CommunicationType::GetUserData)
.with_id(cv.clone().get_id())
.add_data(
DataTypes::username,
DataValue::Str(chat_partner_name.clone()),
),
.add_typed_default(DataType::Username, DataValue::Str(chat_partner_name.clone())),
Some(Duration::from_secs(20)),
)
.await;
let chat_partner_id = {
if let Ok(load_uuid_response) = load_uuid_response {
load_uuid_response.get_data(DataTypes::user_id).clone()
load_uuid_response.get_data(DataType::UserId).clone()
} else {
DataValue::Null
}
@ -675,7 +625,7 @@ impl ClientConnection {
let updated_cv = cv
.with_sender(sender_user_id as u64)
.add_data(DataTypes::chat_partner_id, chat_partner_id);
.add_typed_default(DataType::ChatPartnerId, chat_partner_id);
rho_conn.message_to_iota(updated_cv).await;
} else {
log_err!(
@ -713,9 +663,9 @@ impl ClientConnection {
msg_type,
msg_id
);
let error_cv = CommunicationValue::new(CommunicationType::error_no_iota)
let error_cv = CommunicationValue::new(CommunicationType::ErrorNoIota)
.with_id(msg_id)
.add_data(DataTypes::user_id, DataValue::Number(sender_user_id as i64));
.add_typed_default(DataType::UserId, DataValue::SignedNumber(sender_user_id.into()));
self.send_message(&error_cv).await;
}
}
@ -758,9 +708,9 @@ impl ClientConnection {
} else {
user_status
};
let notification = CommunicationValue::new(CommunicationType::client_changed)
.add_data(DataTypes::user_id, DataValue::Str(user_id.to_string()))
.add_data(DataTypes::user_state, DataValue::Str(status.to_string()));
let notification = CommunicationValue::new(CommunicationType::ClientChanged)
.add_typed_default(DataType::UserId, DataValue::Str(user_id.to_string()))
.add_typed_default(DataType::UserState, DataValue::Str(status.to_string()));
self.send_message(&notification).await;
}
@ -785,6 +735,7 @@ impl Clone for ClientConnection {
receiver: Arc::clone(&self.receiver),
user_id: self.user_id,
session_id: self.session_id,
client_version: self.client_version.clone(),
ping: Arc::clone(&self.ping),
pub_key: Arc::clone(&self.pub_key),
rho_connection: Arc::clone(&self.rho_connection),

202
src/rho/connection.rs
View file

@ -1,8 +1,8 @@
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataTypeId, DataValue};
use mtp::transport::{Receiver, Sender};
use rand::{Rng, distributions::Alphanumeric};
use std::{collections::BTreeMap, collections::HashMap, sync::Arc, time::Duration};
use tokio::sync::RwLock;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use ttp_native::{Receiver, Sender};
use uuid::Uuid;
use crate::{
@ -45,6 +45,7 @@ pub struct GeneralConnection {
pub session_id: Arc<RwLock<u64>>,
pub app_identifier: Arc<RwLock<Option<String>>>,
pub app_session: Arc<RwLock<Option<Uuid>>>,
pub client_version: Arc<RwLock<String>>,
pub_key: Arc<RwLock<Option<Vec<u8>>>>,
}
@ -63,6 +64,7 @@ impl GeneralConnection {
session_id: Arc::new(RwLock::new(0)),
app_identifier: Arc::new(RwLock::new(None)),
app_session: Arc::new(RwLock::new(None)),
client_version: Arc::new(RwLock::new(mtp::codec::PROTOCOL_VERSION.to_string())),
pub_key: Arc::new(RwLock::new(None)),
})
}
@ -95,26 +97,31 @@ impl GeneralConnection {
log_out!(0, PrintType::General, "General connection handler stopped");
}
async fn handle_identification(self: &Arc<Self>, cv: CommunicationValue) {
if cv.is_type(CommunicationType::register_iota) {
if let DataValue::Str(pub_key) = cv.get_data(DataTypes::public_key) {
let msg = CommunicationValue::new(CommunicationType::complete_register_iota)
.add_data(DataTypes::public_key, DataValue::Str(pub_key.clone()));
if let Some(version) = cv.get_data(DataType::Version).as_str() {
*self.client_version.write().await = version.to_string();
}
if cv.is_type(CommunicationType::RegisterIota) {
if let DataValue::Str(pub_key) = cv.get_data(DataType::PublicKey) {
let msg = CommunicationValue::new(CommunicationType::CompleteRegisterIota)
.add_typed_default(DataType::PublicKey, DataValue::Str(pub_key.clone()));
let response = get_omega_connection()
.await_response(&msg, Some(Duration::from_secs(20)))
.await;
if let Ok(response_cv) = response {
if let DataValue::Number(iota_id) = response_cv.get_data(DataTypes::iota_id) {
let success_msg = CommunicationValue::new(CommunicationType::success)
if let DataValue::SignedNumber(iota_id) = response_cv.get_data(DataType::IotaId)
{
let success_msg = CommunicationValue::new(CommunicationType::Success)
.with_id(cv.get_id())
.add_data(DataTypes::iota_id, DataValue::Number(*iota_id));
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(*iota_id));
log_cv_out!(success_msg);
let _ = self.sender.send(&success_msg).await;
} else {
let err_msg =
CommunicationValue::new(CommunicationType::error_invalid_omikron_id)
CommunicationValue::new(CommunicationType::ErrorInvalidOmikronId)
.with_id(cv.get_id());
let _ = self.sender.send(&err_msg).await;
}
@ -123,19 +130,16 @@ impl GeneralConnection {
return;
}
if cv.is_type(CommunicationType::app_identification) {
if cv.is_type(CommunicationType::AppIdentification) {
let app_identifier = cv
.get_data(DataTypes::app_identifier)
.get_data(DataType::AppIdentifier)
.as_str()
.unwrap_or("")
.to_string();
let app_session_id_str = cv.get_data(DataTypes::app_session).as_str().unwrap_or("");
let app_session_id_str = cv.get_data(DataType::AppSession).as_str().unwrap_or("");
let app_session_id = Uuid::parse_str(app_session_id_str).unwrap_or(Uuid::new_v4());
let pub_key_str = cv
.get_data(DataTypes::app_public_key)
.as_str()
.unwrap_or("");
let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
let pub_key_str = cv.get_data(DataType::AppPublicKey).as_str().unwrap_or("");
let user_id = cv.get_data(DataType::UserId).as_number().unwrap_or(0);
*self.id.write().await = user_id as u64;
*self.app_identifier.write().await = Some(app_identifier);
@ -164,29 +168,29 @@ impl GeneralConnection {
.unwrap()
.export(DataFormat::Base64);
let response = CommunicationValue::new(CommunicationType::app_challenge)
let response = CommunicationValue::new(CommunicationType::AppChallenge)
.with_id(cv.get_id())
.add_data(
DataTypes::public_key,
.add_typed_default(
DataType::PublicKey,
DataValue::Str(public_key_to_base64(&get_public_key())),
)
.add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
.add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge));
log_cv_out!(response);
let _ = self.sender.send(&response).await;
return;
}
if !cv.is_type(CommunicationType::identification) {
if !cv.is_type(CommunicationType::Identification) {
return;
}
if let DataValue::Number(iota_id) = cv.get_data(DataTypes::iota_id) {
if let DataValue::SignedNumber(iota_id) = cv.get_data(DataType::IotaId) {
*self.id.write().await = *iota_id as u64;
*self.connection_kind.write().await = Some(ConnectionKind::Iota);
let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_iota_data)
.add_data(DataTypes::iota_id, DataValue::Number(*iota_id));
let get_pub_key_msg = CommunicationValue::new(CommunicationType::GetIotaData)
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(*iota_id));
let response_cv = get_omega_connection()
.await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
@ -196,7 +200,7 @@ impl GeneralConnection {
Ok(r) => r,
Err(_) => {
log_err!(
*iota_id,
*iota_id as i64,
PrintType::Iota,
"Failed to get Iota data from Omega for iota_id={}",
iota_id
@ -206,7 +210,7 @@ impl GeneralConnection {
};
let base64_pub = response_cv
.get_data(DataTypes::public_key)
.get_data(DataType::PublicKey)
.as_str()
.unwrap_or("");
@ -214,7 +218,7 @@ impl GeneralConnection {
Some(pk) => pk,
None => {
log_err!(
*iota_id,
*iota_id as i64,
PrintType::Iota,
"Failed to load public key for iota_id={}",
iota_id
@ -241,48 +245,48 @@ impl GeneralConnection {
.unwrap()
.export(DataFormat::Base64);
let response = CommunicationValue::new(CommunicationType::challenge)
let response = CommunicationValue::new(CommunicationType::Challenge)
.with_id(cv.get_id())
.add_data(
DataTypes::public_key,
.add_typed_default(
DataType::PublicKey,
DataValue::Str(public_key_to_base64(&get_public_key())),
)
.add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
.add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge));
log_cv_out!(response);
let _ = self.sender.send(&response).await;
} else if let DataValue::Number(user_id) = cv.get_data(DataTypes::user_id) {
} else if let DataValue::SignedNumber(user_id) = cv.get_data(DataType::UserId) {
*self.id.write().await = *user_id as u64;
*self.session_id.write().await = match cv.get_data(DataTypes::session_id) {
DataValue::Number(s) => *s as u64,
*self.session_id.write().await = match cv.get_data(DataType::SessionId) {
DataValue::SignedNumber(s) => *s as u64,
_ => cv.get_sender(),
};
#[allow(unused_assignments)]
let mut base64_pub = String::new();
if cv.is_type(CommunicationType::app_identification) {
if cv.is_type(CommunicationType::AppIdentification) {
*self.connection_kind.write().await = Some(ConnectionKind::Phi);
if let DataValue::Str(app_id) = cv.get_data(DataTypes::app_identifier) {
if let DataValue::Str(app_id) = cv.get_data(DataType::AppIdentifier) {
*self.app_identifier.write().await = Some(app_id.clone());
}
if let DataValue::Str(app_sess) = cv.get_data(DataTypes::app_session) {
if let DataValue::Str(app_sess) = cv.get_data(DataType::AppSession) {
if let Ok(uuid) = uuid::Uuid::parse_str(&app_sess) {
*self.app_session.write().await = Some(uuid);
}
}
base64_pub = cv
.get_data(DataTypes::app_public_key)
.get_data(DataType::AppPublicKey)
.as_str()
.unwrap_or("")
.to_string();
} else {
*self.connection_kind.write().await = Some(ConnectionKind::Client);
let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_user_data)
.add_data(DataTypes::user_id, DataValue::Number(*user_id));
let get_pub_key_msg = CommunicationValue::new(CommunicationType::GetUserData)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(*user_id));
let response_cv = get_omega_connection()
.await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
@ -296,7 +300,7 @@ impl GeneralConnection {
};
base64_pub = response_cv
.get_data(DataTypes::public_key)
.get_data(DataType::PublicKey)
.as_str()
.unwrap_or("")
.to_string();
@ -327,20 +331,20 @@ impl GeneralConnection {
.unwrap()
.export(DataFormat::Base64);
let challenge_type = if cv.is_type(CommunicationType::app_identification) {
CommunicationType::app_challenge
let challenge_type = if cv.is_type(CommunicationType::AppIdentification) {
CommunicationType::AppChallenge
} else {
CommunicationType::challenge
CommunicationType::Challenge
};
let response = CommunicationValue::new(challenge_type)
.with_id(cv.get_id())
.with_receiver(*self.session_id.read().await)
.add_data(
DataTypes::public_key,
.add_typed_default(
DataType::PublicKey,
DataValue::Str(public_key_to_base64(&get_public_key())),
)
.add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
.add_typed_default(DataType::Challenge, DataValue::Str(encrypted_challenge));
log_cv_out!(response);
let _ = self.sender.send(&response).await;
@ -349,13 +353,13 @@ impl GeneralConnection {
async fn handle_challenge_response(self: &Arc<Self>, cv: CommunicationValue) {
let id = *self.id.read().await as i64;
if !cv.is_type(CommunicationType::challenge_response)
&& !cv.is_type(CommunicationType::app_challenge_response)
if !cv.is_type(CommunicationType::ChallengeResponse)
&& !cv.is_type(CommunicationType::AppChallengeResponse)
{
return;
}
if let DataValue::Str(response) = cv.get_data(DataTypes::challenge) {
if let DataValue::Str(response) = cv.get_data(DataType::Challenge) {
let expected = self.challenge.read().await.clone();
if *response == expected {
@ -369,10 +373,10 @@ impl GeneralConnection {
expected,
response
);
let err_msg = CommunicationValue::new(CommunicationType::error_invalid_challenge)
let err_msg = CommunicationValue::new(CommunicationType::ErrorInvalidChallenge)
.with_id(cv.get_id())
.add_data(
DataTypes::message,
.add_typed_default(
DataType::Message,
DataValue::Str(
"Challenge response mismatch; authentication rejected".to_string(),
),
@ -399,8 +403,8 @@ impl GeneralConnection {
match kind {
ConnectionKind::Client => {
let notify = CommunicationValue::new(CommunicationType::user_connected)
.add_data(DataTypes::user_id, DataValue::Number(id as i64));
let notify = CommunicationValue::new(CommunicationType::UserConnected)
.add_typed_default(DataType::UserId, DataValue::SignedNumber((id as i64).into()));
get_omega_connection().send_message(&notify).await;
let user_id = id as i64;
@ -410,18 +414,18 @@ impl GeneralConnection {
let mut rho = rho_manager::get_rho_con_for_user(user_id).await;
if rho.is_none() {
let get_user_msg = CommunicationValue::new(CommunicationType::get_user_data)
.add_data(DataTypes::user_id, DataValue::Number(user_id));
let get_user_msg = CommunicationValue::new(CommunicationType::GetUserData)
.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)))
.await
{
if let DataValue::Number(iota_id) =
user_data_cv.get_data(DataTypes::iota_id)
if let DataValue::SignedNumber(iota_id) =
user_data_cv.get_data(DataType::IotaId)
{
if let Some(bound_rho) =
rho_manager::bind_user_to_iota(user_id, *iota_id).await
rho_manager::bind_user_to_iota(user_id, *iota_id as i64).await
{
bound_rho.bind_user_id(user_id).await;
rho = Some(bound_rho);
@ -434,9 +438,9 @@ impl GeneralConnection {
if let Some(rho_conn) = rho {
let session_id = *self.session_id.read().await as i64;
let iota_msg = CommunicationValue::new(CommunicationType::client_connected)
.add_data(DataTypes::user_id, DataValue::Number(user_id))
.add_data(DataTypes::session_id, DataValue::Number(session_id));
let iota_msg = CommunicationValue::new(CommunicationType::ClientConnected)
.add_typed_default(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()
@ -445,10 +449,11 @@ impl GeneralConnection {
.await
{
let mut ident_resp =
CommunicationValue::new(CommunicationType::identification_response)
CommunicationValue::new(CommunicationType::IdentificationResponse)
.with_id(*self.challenge_cv_id.read().await);
for (k, v) in resp.get_data_container() {
let value_to_add = if k == &DataTypes::contacts {
let tm = mtp::codec::TypeMap::latest();
for (k, v) in resp.iter_typed_data() {
let value_to_add = if k == Some(DataType::Contacts) {
if let Some(contacts) = v.as_array() {
let call_groups =
call_manager::get_call_groups(user_id as u64).await;
@ -468,30 +473,30 @@ impl GeneralConnection {
// List of all members in the call
let member_ids: Vec<DataValue> = members
.iter()
.map(|m| DataValue::Number(m.user_id as i64))
.map(|m| DataValue::SignedNumber(m.user_id.into()))
.collect();
let mut base_call_map: BTreeMap<DataTypes, DataValue> =
let mut base_call_map: BTreeMap<DataTypeId, DataValue> =
BTreeMap::new();
base_call_map.insert(
DataTypes::call_id,
DataType::CallId.to_id(&tm),
DataValue::Str(call.call_id.to_string()),
);
base_call_map.insert(
DataTypes::call_members,
DataType::CallMembers.to_id(&tm),
DataValue::Array(member_ids),
);
if timeout > 0 {
base_call_map.insert(
DataTypes::timeout,
DataValue::Number(timeout as i64),
DataType::Timeout.to_id(&tm),
DataValue::SignedNumber((timeout as i64).into()),
);
}
if admin {
base_call_map.insert(
DataTypes::has_admin,
DataType::HasAdmin.to_id(&tm),
DataValue::Bool(true),
);
}
@ -516,7 +521,7 @@ impl GeneralConnection {
.get(&(member_id, user_id as u64))
{
contact_call_map.insert(
DataTypes::call_secret,
DataType::CallSecret.to_id(&tm),
DataValue::Str(secret.clone()),
);
}
@ -534,12 +539,12 @@ impl GeneralConnection {
let mut new_contacts: Vec<DataValue> = Vec::new();
for contact in contacts {
if let Some(mut contact_map) = contact.as_map() {
if let Some(DataValue::Number(id)) =
contact_map.get(&DataTypes::user_id)
if let Some(DataValue::SignedNumber(id)) =
contact_map.get(&DataType::UserId.to_id(&tm))
{
if let Some(call_list) = invites.get(id) {
if let Some(call_list) = invites.get(&(*id as i64)) {
contact_map.insert(
DataTypes::calls,
DataType::Calls.to_id(&tm),
DataValue::Array(call_list.clone()),
);
}
@ -552,7 +557,7 @@ impl GeneralConnection {
}
ident_resp = ident_resp
.add_data(DataTypes::calls, DataValue::Array(global_calls));
.add_typed_default(DataType::Calls, DataValue::Array(global_calls));
DataValue::Array(new_contacts)
} else {
v.clone()
@ -561,7 +566,9 @@ impl GeneralConnection {
v.clone()
};
ident_resp = ident_resp.add_data(k.clone(), value_to_add);
if let Some(dt) = k {
ident_resp = ident_resp.add_typed_default(dt, value_to_add);
}
}
log_cv_out!(ident_resp);
let _ = self.sender.send(&ident_resp).await;
@ -580,8 +587,8 @@ impl GeneralConnection {
client.start();
}
ConnectionKind::Iota => {
let notify = CommunicationValue::new(CommunicationType::iota_connected)
.add_data(DataTypes::iota_id, DataValue::Number(id as i64));
let notify = CommunicationValue::new(CommunicationType::IotaConnected)
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.into()));
get_omega_connection().send_message(&notify).await;
let iota = IotaConnection::from_general(self.clone(), id).await;
@ -592,17 +599,17 @@ impl GeneralConnection {
rho_manager::add_rho(rho).await;
let get_iota_msg = CommunicationValue::new(CommunicationType::get_iota_data)
.add_data(DataTypes::iota_id, DataValue::Number(id as i64));
let get_iota_msg = CommunicationValue::new(CommunicationType::GetIotaData)
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.into()));
if let Ok(iota_data_cv) = get_omega_connection()
.await_response(&get_iota_msg, Some(Duration::from_secs(20)))
.await
{
if let DataValue::Array(users) = iota_data_cv.get_data(DataTypes::user_ids) {
if let DataValue::Array(users) = iota_data_cv.get_data(DataType::UserIds) {
let mut user_ids: Vec<u64> = Vec::new();
for value in users {
if let DataValue::Number(user_id) = value {
if let DataValue::SignedNumber(user_id) = value {
user_ids.push(*user_id as u64);
}
}
@ -610,10 +617,9 @@ impl GeneralConnection {
}
}
let ident_resp =
CommunicationValue::new(CommunicationType::identification_response)
.with_id(*self.challenge_cv_id.read().await)
.add_data(DataTypes::accepted, DataValue::Bool(true));
let ident_resp = CommunicationValue::new(CommunicationType::IdentificationResponse)
.with_id(*self.challenge_cv_id.read().await)
.add_typed_default(DataType::Accepted, DataValue::Bool(true));
log_cv_out!(ident_resp);
let _ = self.sender.send(&ident_resp).await;
@ -629,18 +635,18 @@ impl GeneralConnection {
let mut rho = rho_manager::get_rho_con_for_user(user_id).await;
if rho.is_none() {
let get_user_msg = CommunicationValue::new(CommunicationType::get_user_data)
.add_data(DataTypes::user_id, DataValue::Number(user_id));
let get_user_msg = CommunicationValue::new(CommunicationType::GetUserData)
.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)))
.await
{
if let DataValue::Number(iota_id) =
user_data_cv.get_data(DataTypes::iota_id)
if let DataValue::SignedNumber(iota_id) =
user_data_cv.get_data(DataType::IotaId)
{
if let Some(bound_rho) =
rho_manager::bind_user_to_iota(user_id, *iota_id).await
rho_manager::bind_user_to_iota(user_id, *iota_id as i64).await
{
bound_rho.bind_user_id(user_id).await;
rho = Some(bound_rho);
@ -659,7 +665,7 @@ impl GeneralConnection {
}
let response =
CommunicationValue::new(CommunicationType::app_identification_response)
CommunicationValue::new(CommunicationType::AppIdentificationResponse)
.with_id(*self.challenge_cv_id.read().await)
.with_receiver(*self.session_id.read().await);

102
src/rho/iota_connection.rs
View file

@ -8,16 +8,18 @@ use crate::omega::omega_connection::get_omega_connection;
use crate::rho::connection::GeneralConnection;
use crate::util::logger::PrintType;
use dashmap::DashMap;
use mtp::codec::CommunicationType;
use mtp::codec::CommunicationValue;
use mtp::codec::DataType;
use mtp::codec::DataTypeId;
use mtp::codec::DataValue;
use mtp::codec::TypeMap;
use mtp::transport::Receiver;
use mtp::transport::Sender;
use std::collections::BTreeMap;
use std::{collections::HashMap, sync::Arc, time::Duration};
use tokio::sync::RwLock;
use tokio::sync::mpsc;
use ttp_core::CommunicationType;
use ttp_core::CommunicationValue;
use ttp_core::DataTypes;
use ttp_core::DataValue;
use ttp_native::Receiver;
use ttp_native::Sender;
use x448::PublicKey;
use super::{rho_connection::RhoConnection, rho_manager};
@ -26,6 +28,7 @@ use crate::omega::omega_connection::OmegaConnection;
#[allow(dead_code)]
pub struct IotaConnection {
pub iota_id: u64,
pub client_version: String,
pub sender: Arc<Sender>,
pub receiver: Arc<Receiver>,
pub user_ids: Arc<RwLock<Vec<u64>>>,
@ -46,6 +49,7 @@ impl IotaConnection {
sender: general.sender.clone(),
receiver: general.receiver.clone(),
iota_id: iota_id,
client_version: general.client_version.read().await.clone(),
waiting_tasks: DashMap::new(),
})
}
@ -141,7 +145,7 @@ impl IotaConnection {
/// Send a CommunicationValue to the Iota
pub async fn send_message(&self, cv: &CommunicationValue) {
if !cv.is_type(CommunicationType::pong) {
if !cv.is_type(CommunicationType::Pong) {
log_cv_out!(PrintType::Iota, cv);
}
if let Err(e) = self.sender.send(&cv).await {
@ -164,7 +168,7 @@ impl IotaConnection {
}
// Handle ping
if cv.is_type(CommunicationType::ping) || cv.is_type(CommunicationType::pong) {
if cv.is_type(CommunicationType::Ping) || cv.is_type(CommunicationType::Pong) {
self.handle_ping(cv).await;
return;
}
@ -172,7 +176,7 @@ impl IotaConnection {
log_cv_in!(PrintType::Iota, cv);
// Handle GET_CHATS
if cv.is_type(CommunicationType::get_chats) {
if cv.is_type(CommunicationType::GetChats) {
self.handle_get_chats(cv).await;
return;
}
@ -180,14 +184,14 @@ impl IotaConnection {
// Handle forwarding to other Iotas or clients
let receiver_id = cv.get_receiver();
if (receiver_id != 0 && !self.get_user_ids().await.contains(&(receiver_id as u64)))
|| cv.is_type(CommunicationType::message_other_iota)
|| cv.is_type(CommunicationType::send_chat)
|| cv.is_type(CommunicationType::MessageOtherIota)
|| cv.is_type(CommunicationType::SendChat)
{
self.handle_forward_message(cv).await;
return;
}
if cv.is_type(CommunicationType::complete_register_user) {
if cv.is_type(CommunicationType::CompleteRegisterUser) {
let response_cv = get_omega_connection()
.await_response(
&cv.clone().with_sender(self.iota_id),
@ -195,8 +199,8 @@ impl IotaConnection {
)
.await;
if let Ok(response_cv) = response_cv {
if response_cv.is_type(CommunicationType::success) {
if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
if response_cv.is_type(CommunicationType::Success) {
if let Some(user_id) = cv.get_data(DataType::UserId).as_number() {
self.add_user_id(user_id as u64).await;
}
}
@ -205,12 +209,12 @@ impl IotaConnection {
return;
}
if cv.is_type(CommunicationType::change_iota_data)
|| cv.is_type(CommunicationType::push_notification)
|| cv.is_type(CommunicationType::get_user_data)
|| cv.is_type(CommunicationType::get_iota_data)
|| cv.is_type(CommunicationType::get_register)
|| cv.is_type(CommunicationType::delete_iota)
if cv.is_type(CommunicationType::ChangeIotaData)
|| cv.is_type(CommunicationType::PushNotification)
|| cv.is_type(CommunicationType::GetUserData)
|| cv.is_type(CommunicationType::GetIotaData)
|| cv.is_type(CommunicationType::GetRegister)
|| cv.is_type(CommunicationType::DeleteIota)
{
let sender = self.get_iota_id().await;
@ -243,7 +247,7 @@ impl IotaConnection {
}
/// Handle ping message
async fn handle_ping(&self, cv: CommunicationValue) {
if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) {
if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
let mut ping_guard = self.ping.write().await;
*ping_guard = ping_val;
@ -256,13 +260,21 @@ impl IotaConnection {
HashMap::new()
};
let pings: Vec<(DataTypes, DataValue)> = client_pings
let tm = TypeMap::latest();
let pings: Vec<DataValue> = client_pings
.into_iter()
.map(|(k, v)| (DataTypes::parse(k), DataValue::Number(v)))
.map(|(k, v)| {
let mut map = BTreeMap::new();
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()));
DataValue::container_from_map(&map)
})
.collect();
let response = CommunicationValue::new(CommunicationType::pong)
let response = CommunicationValue::new(CommunicationType::Pong)
.with_id(cv.get_id())
.add_data(DataTypes::ping_clients, DataValue::Container(pings));
.add_typed_default(DataType::PingClients, DataValue::Array(pings));
self.send_message(&response).await;
}
@ -288,7 +300,7 @@ impl IotaConnection {
if let Some(target_rho) = rho_manager::get_rho_con_for_user(receiver_id as i64).await {
target_rho.message_to_iota(cv).await;
} else {
let error = CommunicationValue::new(CommunicationType::error_no_iota)
let error = CommunicationValue::new(CommunicationType::ErrorNoIota)
.with_id(cv.get_id())
.with_sender(cv.get_sender());
self.send_message(&error).await;
@ -303,8 +315,8 @@ impl IotaConnection {
);
self.send_message(
&CommunicationValue::new(CommunicationType::error_invalid_user_id).add_data(
DataTypes::error_type,
&CommunicationValue::new(CommunicationType::ErrorInvalidUserId).add_typed_default(
DataType::ErrorType,
DataValue::Str(
"You are sending to another User without authority.".to_string(),
),
@ -330,6 +342,7 @@ impl IotaConnection {
}
let mut interested_ids: Vec<i64> = Vec::new();
let tm = TypeMap::latest();
// ============================
// Load Calls
@ -353,20 +366,21 @@ impl IotaConnection {
// List of all members in the call
let member_ids: Vec<DataValue> = members
.iter()
.map(|m| DataValue::Number(m.user_id as i64))
.map(|m| DataValue::SignedNumber(m.user_id.into()))
.collect();
// Build base call container
let mut base_call_map: BTreeMap<DataTypes, DataValue> = BTreeMap::new();
base_call_map.insert(DataTypes::call_id, DataValue::Str(call.call_id.to_string()));
base_call_map.insert(DataTypes::call_members, DataValue::Array(member_ids));
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(DataType::CallMembers.to_id(&tm), DataValue::Array(member_ids));
if timeout > 0 {
base_call_map.insert(DataTypes::timeout, DataValue::Number(timeout as i64));
base_call_map
.insert(DataType::Timeout.to_id(&tm), DataValue::SignedNumber(timeout.into()));
}
if admin {
base_call_map.insert(DataTypes::has_admin, DataValue::Bool(true));
base_call_map.insert(DataType::HasAdmin.to_id(&tm), DataValue::Bool(true));
}
// Add to global calls (without contact-specific secret)
@ -384,7 +398,7 @@ 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
.insert(DataTypes::call_secret, DataValue::Str(secret.clone()));
.insert(DataType::CallSecret.to_id(&tm), DataValue::Str(secret.clone()));
}
invites
@ -399,27 +413,27 @@ impl IotaConnection {
// Enrich Contacts
// ============================
let enriched_contacts = if empty {
match cv.get_data(DataTypes::user_ids) {
match cv.get_data(DataType::UserIds) {
DataValue::Array(arr) => DataValue::Array(arr.clone()),
_ => DataValue::Array(vec![]),
}
} else {
let mut enriched: Vec<DataValue> = Vec::new();
if let DataValue::Array(users) = cv.get_data(DataTypes::user_ids) {
if let DataValue::Array(users) = cv.get_data(DataType::UserIds) {
for user_val in users {
if let DataValue::Container(entries) = user_val {
let mut user_map: BTreeMap<DataTypes, DataValue> =
let mut user_map: BTreeMap<DataTypeId, DataValue> =
entries.iter().cloned().collect();
if let Some(DataValue::Number(id)) = user_map.get(&DataTypes::user_id) {
interested_ids.push(*id);
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)
if let Some(call_list) = invites.get(&(*id as i64))
&& !call_list.is_empty()
{
user_map
.insert(DataTypes::calls, DataValue::Array(call_list.clone()));
.insert(DataType::Calls.to_id(&tm), DataValue::Array(call_list.clone()));
}
}
@ -449,8 +463,8 @@ impl IotaConnection {
// Forward to client
// ============================
self.forward_to_client(
cv.add_data(DataTypes::user_ids, enriched_contacts)
.add_data(DataTypes::calls, DataValue::Array(global_calls)),
cv.add_typed_default(DataType::UserIds, enriched_contacts)
.add_typed_default(DataType::Calls, DataValue::Array(global_calls)),
)
.await;
}

10
src/rho/rho_connection.rs
View file

@ -2,10 +2,10 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection
use crate::omega::omega_connection::OmegaConnection;
use crate::{data::user::UserStatus, rho::app_connection::AppConnection};
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use uuid::Uuid;
pub struct RhoConnection {
@ -125,8 +125,8 @@ impl RhoConnection {
let user_id = connection.user_id as i64;
let session_id = connection.session_id as i64;
let notification = CommunicationValue::new(CommunicationType::client_connected)
.add_data(DataTypes::user_id, DataValue::Number(user_id));
let notification = CommunicationValue::new(CommunicationType::ClientConnected)
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()));
self.iota_connection.send_message(&notification).await;
@ -216,14 +216,14 @@ impl RhoConnection {
pub async fn message_to_client(&self, cv: CommunicationValue) {
let connections = self.client_connections.read().await;
let receiver_id = cv.get_receiver();
let session_id = cv.get_data(DataTypes::session_id).as_number();
let session_id = cv.get_data(DataType::SessionId).as_number();
for connection in connections.iter() {
if connection.user_id != receiver_id {
continue;
}
if let Some(session_id) = session_id {
if connection.session_id as i64 != session_id {
if connection.session_id as i128 != session_id {
continue;
}
}

4
src/rho/server.rs
View file

@ -1,3 +1,4 @@
use std::net::{IpAddr, Ipv4Addr};
use std::time::Duration;
use crate::{
@ -5,7 +6,7 @@ use crate::{
rho::connection::GeneralConnection,
util::{file_util::load_file_vec, logger::PrintType},
};
use ttp_native::{Host, Policy, SendMode, host};
use mtp::transport::{Host, Policy, SendMode, host};
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");
@ -13,6 +14,7 @@ pub async fn start(port: u16) -> Result<(), Box<dyn std::error::Error>> {
let key_pem = load_file_vec("certs", "key.pem").expect("Error loading Keyfile");
let mut host: Host = host(
IpAddr::V4(Ipv4Addr::UNSPECIFIED),
port,
cert_pem,
key_pem,