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

[FIX] Migrated to Tensamin Transport Protocol

Browse source
This commit is contained in:
Alex Emmet 2026-02-27 23:44:47 +01:00
commit c9d21fd3c6
24 changed files with 1451 additions and 2252 deletions
Download patch file Download diff file Expand all files Collapse all files

618
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

13
Cargo.toml
View file

@ -4,6 +4,9 @@ version = "0.1.0"
edition = "2024" edition = "2024"
[dependencies] [dependencies]
epsilon-core = { git = "https://github.com/Tensamin/Epsilon.git", package = "epsilon-core" }
epsilon-native = { git = "https://github.com/Tensamin/Epsilon.git", package = "epsilon-native" }
ansi_term = "*" ansi_term = "*"
async-tungstenite = { version = "0.32.1", features = [ async-tungstenite = { version = "0.32.1", features = [
"futures-03-sink", "futures-03-sink",
@ -56,7 +59,7 @@ pkcs8 = { version = "*", features = ["alloc"] }
rand = "0.8" rand = "0.8"
rand_core = { version = "0.6", features = ["getrandom", "std"] } rand_core = { version = "0.6", features = ["getrandom", "std"] }
reqwest = "*" reqwest = "*"
rustls = { version = "*", default-features = false, features = ["ring"] } rustls = { version = "0.23.37", default-features = false, features = ["ring"] }
serde = { version = "1.0.219", features = ["derive"] } serde = { version = "1.0.219", features = ["derive"] }
sha2 = "*" sha2 = "*"
sys-info = "*" sys-info = "*"
@ -85,7 +88,9 @@ cbc = "0.1.2"
aes = "0.8.4" aes = "0.8.4"
strum = "0.27.2" strum = "0.27.2"
strum_macros = "0.27.2" strum_macros = "0.27.2"
livekit = "0.7.25" livekit = "0.7.32"
livekit-api = { version = "0.4.10", features = ["native-tls"] } livekit-api = { version = "0.4.14", features = ["native-tls"] }
hyper = { version = "1.8.1", features = ["http1", "http2"] } hyper = { version = "1.8.1", features = ["http1", "http2"] }
livekit-protocol = "0.6.0" livekit-protocol = "0.7.1"
rustls-pemfile = "2.2.0"
quinn = "0.11.9"

316
src/anonymous_clients/anonymous_client_connection.rs
View file

@ -1,27 +1,24 @@
use async_tungstenite::tungstenite::Message; use epsilon_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use async_tungstenite::{WebSocketReceiver, WebSocketSender}; use epsilon_native::{Receiver, Sender};
use json::JsonValue;
use json::number::Number;
use std::str::FromStr; use std::str::FromStr;
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH}; use std::time::Duration;
use tokio::sync::RwLock; use tokio::sync::RwLock;
use tokio_util::compat::Compat;
use tungstenite::Utf8Bytes;
use uuid::Uuid; use uuid::Uuid;
use crate::anonymous_clients::anonymous_manager::{self, generate_username}; use crate::anonymous_clients::anonymous_manager::{self, generate_username};
use crate::calls::call_manager; use crate::calls::call_manager;
use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes};
use crate::omega::omega_connection::get_omega_connection; use crate::omega::omega_connection::get_omega_connection;
use crate::rho::connection::GeneralConnection;
use crate::rho::rho_manager; use crate::rho::rho_manager;
use crate::util::logger::PrintType; use crate::util::logger::PrintType;
use crate::{log_in, log_out}; use crate::{log_cv_in, log_cv_out, log_out};
pub struct AnonymousClientConnection { pub struct AnonymousClientConnection {
pub sender: Arc<RwLock<WebSocketSender<Compat<tokio::net::TcpStream>>>>, user_id: u64,
pub receiver: Arc<RwLock<WebSocketReceiver<Compat<tokio::net::TcpStream>>>>,
pub user_id: Arc<RwLock<i64>>, pub sender: Arc<Sender>,
pub receiver: Arc<Receiver>,
pub ping: Arc<RwLock<i64>>, pub ping: Arc<RwLock<i64>>,
pub interested_users: Arc<RwLock<Vec<i64>>>, pub interested_users: Arc<RwLock<Vec<i64>>>,
is_open: Arc<RwLock<bool>>, is_open: Arc<RwLock<bool>>,
@ -31,33 +28,33 @@ pub struct AnonymousClientConnection {
} }
impl AnonymousClientConnection { impl AnonymousClientConnection {
/// Create a new AnonymousClientConnection pub async fn from_general(general: Arc<GeneralConnection>, user_id: u64) -> Arc<Self> {
pub fn new(
sender: WebSocketSender<Compat<tokio::net::TcpStream>>,
receiver: WebSocketReceiver<Compat<tokio::net::TcpStream>>,
) -> Arc<Self> {
let username: String = generate_username(); let username: String = generate_username();
Arc::new(Self { Arc::new(Self {
sender: Arc::new(RwLock::new(sender)), user_id: user_id,
receiver: Arc::new(RwLock::new(receiver)),
user_id: Arc::new(RwLock::new( ping: Arc::new(RwLock::new(0)),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis() as i64,
)),
ping: Arc::new(RwLock::new(-1)),
interested_users: Arc::new(RwLock::new(Vec::new())), interested_users: Arc::new(RwLock::new(Vec::new())),
is_open: Arc::new(RwLock::new(true)), is_open: Arc::new(RwLock::new(true)),
sender: general.sender.clone(),
receiver: general.receiver.clone(),
user_name: Arc::new(RwLock::new(username.to_lowercase())), user_name: Arc::new(RwLock::new(username.to_lowercase())),
display_name: Arc::new(RwLock::new(username)), display_name: Arc::new(RwLock::new(username)),
avatar: Arc::new(RwLock::new(String::new())), avatar: Arc::new(RwLock::new(String::new())),
}) })
} }
pub fn start(self: Arc<Self>) {
let self_clone = self.clone();
tokio::spawn(async move {
while let Ok(cv) = self_clone.receiver.receive().await {
self_clone.clone().handle_message(cv).await;
}
});
}
/// Get the user ID /// Get the user ID
pub async fn get_user_id(&self) -> i64 { pub fn get_user_id(&self) -> u64 {
*self.user_id.read().await self.user_id
} }
/// Get the user name /// Get the user name
@ -79,66 +76,35 @@ impl AnonymousClientConnection {
self.avatar.read().await.clone() self.avatar.read().await.clone()
} }
/// Send a string message to the client
pub async fn send_message_str(self: Arc<Self>, message: &str) {
let mut session = self.sender.write().await;
if let Err(e) = session
.send(Message::Text(Utf8Bytes::from(message.to_string())))
.await
{
log_out!(
self.get_user_id().await,
PrintType::Client,
"Failed to send message to anonymous client: {}",
e,
);
}
}
/// Send a CommunicationValue to the client /// Send a CommunicationValue to the client
pub async fn send_message(self: Arc<Self>, cv: &CommunicationValue) { pub async fn send_message(self: Arc<Self>, cv: &CommunicationValue) {
if !*self.is_open.read().await { if !*self.is_open.read().await {
log_out!( log_out!(
self.get_user_id().await, self.user_id as i64,
PrintType::Client, PrintType::Client,
"Attempted to send message to a closed connection." "Attempted to send message to a closed connection."
); );
return; return;
} }
if !cv.is_type(CommunicationType::pong) { if !cv.is_type(CommunicationType::pong) {
log_out!( log_cv_out!(PrintType::Client, &cv);
self.get_user_id().await,
PrintType::Client,
"{}",
&cv.to_json().to_string()
);
} }
self.send_message_str(&cv.to_json().to_string()).await; self.sender.send(&cv).await;
} }
/// Handle incoming message from client /// Handle incoming message from client
pub async fn handle_message(self: Arc<Self>, message: Utf8Bytes) { pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
tokio::spawn(async move { tokio::spawn(async move {
let cv = CommunicationValue::from_json(&message);
if cv.is_type(CommunicationType::ping) { if cv.is_type(CommunicationType::ping) {
self.handle_ping(cv).await; self.handle_ping(cv).await;
return; return;
} }
log_in!( log_cv_in!(PrintType::Client, &cv);
self.get_user_id().await,
PrintType::Client,
"Anonymous: {}",
&cv.to_json().to_string()
);
if cv.is_type(CommunicationType::identification) { if cv.is_type(CommunicationType::identification) {
let call_id = Uuid::parse_str( let call_id =
cv.get_data(DataTypes::call_id) Uuid::parse_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or(""))
.unwrap_or(&JsonValue::Null) .unwrap_or(Uuid::new_v4());
.as_str()
.unwrap_or(""),
)
.unwrap_or(Uuid::new_v4());
let call = if let Some(call) = call_manager::get_call(call_id).await { let call = if let Some(call) = call_manager::get_call(call_id).await {
if call.is_anonymous().await { if call.is_anonymous().await {
@ -160,79 +126,62 @@ impl AnonymousClientConnection {
return; return;
}; };
let mut invited = JsonValue::new_array(); let mut invited = Vec::new();
for call_invitee in call.members.read().await.clone() { for call_invitee in call.members.read().await.clone() {
let call_invitee_cv = get_omega_connection() let call_invitee_cv = get_omega_connection()
.await_response( .await_response(
&CommunicationValue::new(CommunicationType::get_user_data).add_data( &CommunicationValue::new(CommunicationType::get_user_data).add_data(
DataTypes::user_id, DataTypes::user_id,
JsonValue::from(call_invitee.user_id), DataValue::Number(call_invitee.user_id as i64),
), ),
Some(Duration::from_secs(2)), Some(Duration::from_secs(2)),
) )
.await .await
.unwrap(); .unwrap();
let mut json_invitee = JsonValue::new_object(); let mut json_invitee = Vec::new();
let _ = json_invitee.insert( let _ = json_invitee.push((
"user_id", DataTypes::user_id,
call_invitee_cv call_invitee_cv.get_data(DataTypes::user_id).clone(),
.get_data(DataTypes::user_id) ));
.unwrap_or(&JsonValue::Null) let _ = json_invitee.push((
.clone(), DataTypes::username,
); call_invitee_cv.get_data(DataTypes::username).clone(),
let _ = json_invitee.insert( ));
"username", let _ = json_invitee.push((
call_invitee_cv DataTypes::display,
.get_data(DataTypes::username) call_invitee_cv.get_data(DataTypes::display).clone(),
.unwrap_or(&JsonValue::Null) ));
.clone(), let _ = json_invitee.push((
); DataTypes::avatar,
let _ = json_invitee.insert( call_invitee_cv.get_data(DataTypes::avatar).clone(),
"display", ));
call_invitee_cv
.get_data(DataTypes::display)
.unwrap_or(&JsonValue::Null)
.clone(),
);
let _ = json_invitee.insert(
"avatar",
call_invitee_cv
.get_data(DataTypes::avatar)
.unwrap_or(&JsonValue::Null)
.clone(),
);
let _ = invited.push(json_invitee); let _ = invited.push(DataValue::Container(json_invitee));
} }
let token = call.create_anonymous_token(self.get_user_id().await).await; let token = call.create_anonymous_token(self.get_user_id()).await;
let mut serialized = JsonValue::new_object(); let mut serialized = Vec::new();
let _ = serialized.insert("call_id", JsonValue::String(call_id.to_string())); let _ = serialized.push((DataTypes::call_id, DataValue::Str(call_id.to_string())));
let _ = serialized.insert("call_invited", invited.clone()); let _ =
let _ = serialized.insert("call_members", invited); serialized.push((DataTypes::call_invited, DataValue::Array(invited.clone())));
let _ = serialized.insert("call_token", JsonValue::String(token.unwrap())); let _ = serialized.push((DataTypes::call_members, DataValue::Array(invited)));
let _ = serialized.push((DataTypes::call_token, DataValue::Str(token.unwrap())));
self.clone() self.clone()
.send_message( .send_message(
&&CommunicationValue::new(CommunicationType::identification_response) &&CommunicationValue::new(CommunicationType::identification_response)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_data( .add_data(DataTypes::user_id, DataValue::Number(self.user_id as i64))
DataTypes::user_id,
JsonValue::from(self.get_user_id().await),
)
.add_data( .add_data(
DataTypes::username, DataTypes::username,
JsonValue::String(self.clone().get_user_name().await), DataValue::Str(self.clone().get_user_name().await),
) )
.add_data( .add_data(
DataTypes::display, DataTypes::display,
JsonValue::String(self.get_display_name().await), DataValue::Str(self.get_display_name().await),
) )
.add_data( .add_data(DataTypes::avatar, DataValue::Str(self.get_avatar().await))
DataTypes::avatar, .add_data(DataTypes::call_state, DataValue::Container(serialized)),
JsonValue::String(self.get_avatar().await),
)
.add_data(DataTypes::call_state, serialized),
) )
.await; .await;
} }
@ -271,11 +220,7 @@ impl AnonymousClientConnection {
} }
if cv.is_type(CommunicationType::change_user_data) { if cv.is_type(CommunicationType::change_user_data) {
if let Some(display_name) = cv if let Some(display_name) = cv.get_data(DataTypes::display).as_str() {
.get_data(DataTypes::display)
.unwrap_or(&JsonValue::Null)
.as_str()
{
let _ = self.set_display_name(display_name.to_string()).await; let _ = self.set_display_name(display_name.to_string()).await;
} }
@ -284,17 +229,9 @@ impl AnonymousClientConnection {
if cv.is_type(CommunicationType::get_user_data) { if cv.is_type(CommunicationType::get_user_data) {
if let Some(anonymous) = { if let Some(anonymous) = {
if let Some(user_id) = cv if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
.get_data(DataTypes::user_id) anonymous_manager::get_anonymous_user(user_id as u64).await
.unwrap_or(&JsonValue::Null) } else if let Some(username) = cv.get_data(DataTypes::username).as_str() {
.as_i64()
{
anonymous_manager::get_anonymous_user(user_id).await
} else if let Some(username) = cv
.get_data(DataTypes::username)
.unwrap_or(&JsonValue::Null)
.as_str()
{
anonymous_manager::get_anonymous_user_by_name(username.to_string()).await anonymous_manager::get_anonymous_user_by_name(username.to_string()).await
} else { } else {
None None
@ -302,14 +239,23 @@ impl AnonymousClientConnection {
} { } {
let response = CommunicationValue::new(CommunicationType::get_user_data) let response = CommunicationValue::new(CommunicationType::get_user_data)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_data_str(DataTypes::username, anonymous.get_user_name().await) .add_data(
DataTypes::username,
DataValue::Str(anonymous.get_user_name().await),
)
.add_data( .add_data(
DataTypes::user_id, DataTypes::user_id,
JsonValue::Number(Number::from(anonymous.get_user_id().await)), DataValue::Number(anonymous.user_id as i64),
) )
.add_data_str(DataTypes::display, anonymous.get_display_name().await) .add_data(
.add_data_str(DataTypes::user_state, "online".to_string()) DataTypes::display,
.add_data_str(DataTypes::avatar, anonymous.get_avatar().await); DataValue::Str(anonymous.get_display_name().await),
)
.add_data(DataTypes::user_state, DataValue::Str("online".to_string()))
.add_data(
DataTypes::avatar,
DataValue::Str(anonymous.get_avatar().await),
);
self.send_message(&response).await; self.send_message(&response).await;
@ -330,10 +276,7 @@ impl AnonymousClientConnection {
let client_for_closure = self.clone(); let client_for_closure = self.clone();
tokio::spawn(async move { tokio::spawn(async move {
let response_cv = get_omega_connection() let response_cv = get_omega_connection()
.await_response( .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20)))
&cv.with_sender(*self.user_id.read().await),
Some(Duration::from_secs(20)),
)
.await; .await;
if let Ok(response_cv) = response_cv { if let Ok(response_cv) = response_cv {
client_for_closure.send_message(&response_cv).await; client_for_closure.send_message(&response_cv).await;
@ -344,7 +287,7 @@ impl AnonymousClientConnection {
/// Handle ping message /// Handle ping message
async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) { async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) {
// Update our ping if provided // Update our ping if provided
if let Some(last_ping) = cv.get_data(DataTypes::last_ping) { if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
if let Ok(ping_val) = last_ping.to_string().parse::<i64>() { if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
let mut ping_guard = self.ping.write().await; let mut ping_guard = self.ping.write().await;
*ping_guard = ping_val; *ping_guard = ping_val;
@ -367,11 +310,7 @@ impl AnonymousClientConnection {
/// Handle call invite /// Handle call invite
async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
let receiver_id: i64 = cv let receiver_id: i64 = cv.get_data(DataTypes::receiver_id).as_number().unwrap_or(0);
.get_data(DataTypes::receiver_id)
.unwrap_or(&json::JsonValue::Number(Number::from(0)))
.as_i64()
.unwrap_or(0);
if receiver_id == 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::error_no_user_id)
.await; .await;
@ -379,7 +318,7 @@ impl AnonymousClientConnection {
} }
let call_id = match cv.get_data(DataTypes::call_id) { let call_id = match cv.get_data(DataTypes::call_id) {
Some(id_str) => match Uuid::parse_str(&id_str.to_string()) { DataValue::Str(id_str) => match Uuid::parse_str(&id_str.to_string()) {
Ok(id) => id, Ok(id) => id,
Err(_) => { Err(_) => {
self.send_error_response( self.send_error_response(
@ -397,8 +336,7 @@ impl AnonymousClientConnection {
} }
}; };
let invited = let invited = call_manager::add_invite(call_id, self.user_id, receiver_id as u64).await;
call_manager::add_invite(call_id, *self.user_id.read().await, receiver_id).await;
if !invited { if !invited {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_call_id) self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_call_id)
.await; .await;
@ -416,15 +354,18 @@ impl AnonymousClientConnection {
}; };
// Get sender user ID // Get sender user ID
let sender_id = self.get_user_id().await; let sender_id = self.get_user_id();
// Create and send call distribution message // Create and send call distribution message
let forward = CommunicationValue::new(CommunicationType::call_invite) let forward = CommunicationValue::new(CommunicationType::call_invite)
.with_receiver(receiver_id) .with_receiver(receiver_id as u64)
.with_sender(sender_id) .with_sender(sender_id)
.add_data_str(DataTypes::call_id, call_id.to_string()) .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
.add_data_str(DataTypes::receiver_id, receiver_id.to_string()) .add_data(
.add_data_str(DataTypes::sender_id, sender_id.to_string()); DataTypes::receiver_id,
DataValue::Str(receiver_id.to_string()),
)
.add_data(DataTypes::sender_id, DataValue::Str(sender_id.to_string()));
target_rho.message_to_client(forward).await; target_rho.message_to_client(forward).await;
@ -434,10 +375,10 @@ impl AnonymousClientConnection {
/// Handle get call request /// Handle get call request
async fn handle_get_call(self: Arc<Self>, cv: CommunicationValue) { async fn handle_get_call(self: Arc<Self>, cv: CommunicationValue) {
let user_id = self.get_user_id().await; let user_id = self.get_user_id();
let call_id = match cv.get_data(DataTypes::call_id) { let call_id = match cv.get_data(DataTypes::call_id) {
Some(id_str) => match Uuid::parse_str(&id_str.to_string()) { DataValue::Str(id_str) => match Uuid::parse_str(&id_str.to_string()) {
Ok(id) => id, Ok(id) => id,
Err(_) => { Err(_) => {
self.send_error_response(&cv.get_id(), CommunicationType::error) self.send_error_response(&cv.get_id(), CommunicationType::error)
@ -456,7 +397,7 @@ impl AnonymousClientConnection {
let response = CommunicationValue::new(CommunicationType::call_token) let response = CommunicationValue::new(CommunicationType::call_token)
.with_id(cv.get_id()) .with_id(cv.get_id())
.with_receiver(user_id) .with_receiver(user_id)
.add_data_str(DataTypes::call_token, token); .add_data(DataTypes::call_token, DataValue::Str(token.to_string()));
self.send_message(&response).await; self.send_message(&response).await;
} else { } else {
self.send_error_response(&cv.get_id(), CommunicationType::error) self.send_error_response(&cv.get_id(), CommunicationType::error)
@ -465,33 +406,20 @@ impl AnonymousClientConnection {
} }
} }
async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) {
let call_id = Uuid::from_str( let call_id =
cv.get_data(DataTypes::call_id) Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap();
.unwrap_or(&JsonValue::Null) let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
.as_str() let untill = cv.get_data(DataTypes::untill).as_number().unwrap_or(0);
.unwrap_or(""),
)
.unwrap();
let user_id = cv
.get_data(DataTypes::user_id)
.unwrap_or(&JsonValue::Null)
.as_i64()
.unwrap_or(0);
let untill = cv
.get_data(DataTypes::untill)
.unwrap_or(&JsonValue::Null)
.as_i64()
.unwrap_or(0);
let call = call_manager::get_call(call_id).await; let call = call_manager::get_call(call_id).await;
if let Some(call) = call { if let Some(call) = call {
if call if call
.get_caller(self.get_user_id().await) .get_caller(self.get_user_id())
.await .await
.unwrap() .unwrap()
.has_admin() .has_admin()
{ {
call.get_caller(user_id) call.get_caller(user_id as u64)
.await .await
.unwrap() .unwrap()
.set_timeout(untill) .set_timeout(untill)
@ -500,38 +428,25 @@ impl AnonymousClientConnection {
} }
} }
async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) {
let call_id = Uuid::from_str( let call_id =
cv.get_data(DataTypes::call_id) Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap();
.unwrap_or(&JsonValue::Null) let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
.as_str()
.unwrap_or(""),
)
.unwrap();
let user_id = cv
.get_data(DataTypes::user_id)
.unwrap_or(&JsonValue::Null)
.as_i64()
.unwrap_or(0);
let call = call_manager::get_call(call_id).await; let call = call_manager::get_call(call_id).await;
if let Some(call) = call { if let Some(call) = call {
if call if call
.get_caller(self.get_user_id().await) .get_caller(self.get_user_id())
.await .await
.unwrap() .unwrap()
.has_admin() .has_admin()
{ {
call.remove_caller(user_id).await; call.remove_caller(user_id as u64).await;
} }
} }
} }
/// Send error response /// Send error response
async fn send_error_response( async fn send_error_response(self: Arc<Self>, message_id: &u32, error_type: CommunicationType) {
self: Arc<Self>,
message_id: &Uuid,
error_type: CommunicationType,
) {
let error = CommunicationValue::new(error_type).with_id(*message_id); let error = CommunicationValue::new(error_type).with_id(*message_id);
self.send_message(&error).await; self.send_message(&error).await;
} }
@ -544,8 +459,7 @@ impl AnonymousClientConnection {
} }
*is_open_guard = false; *is_open_guard = false;
let mut session = self.sender.write().await; let _ = self.sender.close();
let _ = session.close(None).await;
} }
/// Set interested users list /// Set interested users list
@ -563,8 +477,8 @@ impl AnonymousClientConnection {
let interested_guard = self.clone().get_interested_users().await; let interested_guard = self.clone().get_interested_users().await;
if interested_guard.contains(&user_id) { if interested_guard.contains(&user_id) {
let notification = CommunicationValue::new(CommunicationType::client_changed) let notification = CommunicationValue::new(CommunicationType::client_changed)
.add_data_str(DataTypes::user_id, user_id.to_string()) .add_data(DataTypes::user_id, DataValue::Str(user_id.to_string()))
.add_data_str(DataTypes::user_state, format!("online")); .add_data(DataTypes::user_state, DataValue::Str("online".to_string()));
self.send_message(&notification).await; self.send_message(&notification).await;
} }
@ -583,7 +497,7 @@ impl Clone for AnonymousClientConnection {
Self { Self {
sender: Arc::clone(&self.sender), sender: Arc::clone(&self.sender),
receiver: Arc::clone(&self.receiver), receiver: Arc::clone(&self.receiver),
user_id: Arc::clone(&self.user_id), user_id: self.user_id,
ping: Arc::clone(&self.ping), ping: Arc::clone(&self.ping),
interested_users: Arc::clone(&self.interested_users), interested_users: Arc::clone(&self.interested_users),
is_open: Arc::clone(&self.is_open), is_open: Arc::clone(&self.is_open),

8
src/anonymous_clients/anonymous_manager.rs
View file

@ -6,18 +6,18 @@ use std::sync::Arc;
use crate::anonymous_clients::anonymous_client_connection::AnonymousClientConnection; use crate::anonymous_clients::anonymous_client_connection::AnonymousClientConnection;
static ANONYMOUS_USERS: Lazy<DashMap<i64, Arc<AnonymousClientConnection>>> = static ANONYMOUS_USERS: Lazy<DashMap<u64, Arc<AnonymousClientConnection>>> =
Lazy::new(|| DashMap::new()); Lazy::new(|| DashMap::new());
pub async fn add_anonymous_user(connection: Arc<AnonymousClientConnection>) { pub async fn add_anonymous_user(connection: Arc<AnonymousClientConnection>) {
ANONYMOUS_USERS.insert(connection.get_user_id().await, connection); ANONYMOUS_USERS.insert(connection.get_user_id(), connection);
} }
pub async fn remove_anonymous_user(user_id: i64) { pub async fn remove_anonymous_user(user_id: u64) {
ANONYMOUS_USERS.remove(&user_id); ANONYMOUS_USERS.remove(&user_id);
} }
pub async fn get_anonymous_user(user_id: i64) -> Option<Arc<AnonymousClientConnection>> { pub async fn get_anonymous_user(user_id: u64) -> Option<Arc<AnonymousClientConnection>> {
ANONYMOUS_USERS.get(&user_id).map(|c| c.clone()) ANONYMOUS_USERS.get(&user_id).map(|c| c.clone())
} }

13
src/calls/call_group.rs
View file

@ -1,11 +1,11 @@
use json::JsonValue; use epsilon_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use std::{env, sync::Arc, time::Duration}; use std::{env, sync::Arc, time::Duration};
use tokio::sync::RwLock; use tokio::sync::RwLock;
use uuid::Uuid; use uuid::Uuid;
use crate::{ use crate::{
calls::{call_util, caller::Caller}, calls::{call_util, caller::Caller},
data::communication::{CommunicationType, CommunicationValue, DataTypes},
omega::omega_connection::get_omega_connection, omega::omega_connection::get_omega_connection,
}; };
@ -28,7 +28,7 @@ impl CallGroup {
} }
} }
pub async fn get_caller(&self, user_id: i64) -> Option<Arc<Caller>> { pub async fn get_caller(&self, user_id: u64) -> Option<Arc<Caller>> {
self.members self.members
.read() .read()
.await .await
@ -62,7 +62,7 @@ impl CallGroup {
let response_cv = get_omega_connection() let response_cv = get_omega_connection()
.await_response( .await_response(
&CommunicationValue::new(CommunicationType::shorten_link) &CommunicationValue::new(CommunicationType::shorten_link)
.add_data(DataTypes::link, JsonValue::from(long_link)), .add_data(DataTypes::link, DataValue::Str(long_link)),
Some(Duration::from_secs(20)), Some(Duration::from_secs(20)),
) )
.await; .await;
@ -70,7 +70,6 @@ impl CallGroup {
*self.short_link.write().await = Some( *self.short_link.write().await = Some(
response response
.get_data(DataTypes::link) .get_data(DataTypes::link)
.unwrap()
.as_str() .as_str()
.unwrap() .unwrap()
.to_string(), .to_string(),
@ -84,7 +83,7 @@ impl CallGroup {
} }
} }
pub async fn create_anonymous_token(&self, user_id: i64) -> Option<String> { pub async fn create_anonymous_token(&self, user_id: u64) -> Option<String> {
if self.is_anonymous().await { if self.is_anonymous().await {
if let Ok(token) = call_util::create_token(user_id, self.call_id, false) { if let Ok(token) = call_util::create_token(user_id, self.call_id, false) {
return Some(token); return Some(token);
@ -93,7 +92,7 @@ impl CallGroup {
None None
} }
pub async fn remove_caller(&self, user_id: i64) { pub async fn remove_caller(&self, user_id: u64) {
let _ = call_util::remove_participant(self.call_id, user_id).await; let _ = call_util::remove_participant(self.call_id, user_id).await;
self.members self.members
.write() .write()

8
src/calls/call_manager.rs
View file

@ -7,7 +7,7 @@ use crate::calls::{call_group::CallGroup, caller::Caller};
pub static CALL_GROUPS: Lazy<DashMap<Uuid, Arc<CallGroup>>> = Lazy::new(|| DashMap::new()); pub static CALL_GROUPS: Lazy<DashMap<Uuid, Arc<CallGroup>>> = Lazy::new(|| DashMap::new());
pub async fn get_call_invites(user_id: i64) -> Vec<Arc<Caller>> { pub async fn get_call_invites(user_id: u64) -> Vec<Arc<Caller>> {
let mut callers = Vec::new(); let mut callers = Vec::new();
for (_, cg) in CALL_GROUPS.clone().into_iter() { for (_, cg) in CALL_GROUPS.clone().into_iter() {
let members = cg.members.read().await; let members = cg.members.read().await;
@ -28,7 +28,7 @@ pub async fn get_call(call_id: Uuid) -> Option<Arc<CallGroup>> {
} }
} }
pub async fn get_call_groups(user_id: i64) -> Vec<Arc<CallGroup>> { pub async fn get_call_groups(user_id: u64) -> Vec<Arc<CallGroup>> {
let mut call_groups = Vec::new(); let mut call_groups = Vec::new();
for (_, cg) in CALL_GROUPS.clone().into_iter() { for (_, cg) in CALL_GROUPS.clone().into_iter() {
let is_member = { let is_member = {
@ -43,7 +43,7 @@ pub async fn get_call_groups(user_id: i64) -> Vec<Arc<CallGroup>> {
call_groups call_groups
} }
pub async fn get_call_token(user_id: i64, call_id: Uuid) -> Option<String> { pub async fn get_call_token(user_id: u64, call_id: Uuid) -> Option<String> {
if let Some(cg) = CALL_GROUPS.get(&call_id) { if let Some(cg) = CALL_GROUPS.get(&call_id) {
let mut members = cg.members.write().await; let mut members = cg.members.write().await;
@ -80,7 +80,7 @@ pub async fn get_call_token(user_id: i64, call_id: Uuid) -> Option<String> {
Some(caller.create_token()) Some(caller.create_token())
} }
pub async fn add_invite(call_id: Uuid, inviter_id: i64, invitee_id: i64) -> bool { pub async fn add_invite(call_id: Uuid, inviter_id: u64, invitee_id: u64) -> bool {
if let Some(cg) = CALL_GROUPS.get(&call_id) { if let Some(cg) = CALL_GROUPS.get(&call_id) {
let mut members = cg.members.write().await; let mut members = cg.members.write().await;

4
src/calls/call_util.rs
View file

@ -35,7 +35,7 @@ pub fn get_livekit() -> Result<(String, String, String), ()> {
Ok((hostname, api_key, api_secret)) Ok((hostname, api_key, api_secret))
} }
pub fn create_token(user_id: i64, call_id: Uuid, has_admin: bool) -> Result<String, ()> { pub fn create_token(user_id: u64, call_id: Uuid, has_admin: bool) -> Result<String, ()> {
let (_, api_key, api_secret) = get_livekit()?; let (_, api_key, api_secret) = get_livekit()?;
let token = access_token::AccessToken::with_api_key(&api_key, &api_secret) let token = access_token::AccessToken::with_api_key(&api_key, &api_secret)
@ -69,7 +69,7 @@ pub async fn get_room(call_id: Uuid) -> Result<(RoomClient, Room), ()> {
return Err(()); return Err(());
} }
pub async fn remove_participant(call_id: Uuid, user_id: i64) -> Result<(), ()> { pub async fn remove_participant(call_id: Uuid, user_id: u64) -> Result<(), ()> {
if let Ok((hostname, api_key, api_secret)) = get_livekit() { if let Ok((hostname, api_key, api_secret)) = get_livekit() {
let room_service = RoomClient::with_api_key(&hostname, &api_key, &api_secret); let room_service = RoomClient::with_api_key(&hostname, &api_key, &api_secret);
if let Ok(_) = room_service if let Ok(_) = room_service

4
src/calls/caller.rs
View file

@ -6,14 +6,14 @@ use uuid::Uuid;
use crate::calls::call_util; use crate::calls::call_util;
pub struct Caller { pub struct Caller {
pub user_id: i64, pub user_id: u64,
pub call_id: Uuid, pub call_id: Uuid,
pub has_admin: bool, pub has_admin: bool,
pub timeout: RwLock<i64>, pub timeout: RwLock<i64>,
} }
impl Caller { impl Caller {
pub fn new(user_id: i64, call_id: Uuid, has_admin: bool) -> Self { pub fn new(user_id: u64, call_id: Uuid, has_admin: bool) -> Self {
Caller { Caller {
user_id, user_id,
call_id, call_id,

403
src/data/communication.rs
View file

@ -1,403 +0,0 @@
use json::number::Number;
use json::{Array, JsonValue, object, parse};
use std::collections::HashMap;
use std::time::{SystemTime, UNIX_EPOCH};
use strum::IntoEnumIterator;
use strum_macros::EnumIter;
use uuid::Uuid;
#[derive(Eq, Hash, PartialEq, EnumIter, Clone, Debug)]
#[allow(non_camel_case_types, dead_code)]
pub enum DataTypes {
error_type,
accepted_ids,
uuid,
register_id,
link,
settings,
settings_name,
chat_partner_id,
chat_partner_name,
iota_id,
user_id,
user_ids,
iota_ids,
user_state,
user_states,
user_pings,
call_state,
screen_share,
private_key_hash,
accepted,
accepted_profiles,
denied_profiles,
content,
messages,
notifications,
send_time,
get_time,
get_variant,
shared_secret_own,
shared_secret_other,
shared_secret_sign,
shared_secret,
call_id,
call_token,
untill,
enabled,
start_date,
end_date,
receiver_id,
sender_id,
signature,
signed,
message,
message_state,
last_ping,
ping_iota,
ping_clients,
matches,
omikron,
offset,
amount,
position,
name,
path,
codec,
function,
payload,
result,
interactables,
want_to_watch,
watcher,
created_at,
username,
display,
avatar,
about,
status,
public_key,
sub_level,
sub_end,
community_address,
challenge,
community_title,
communities,
rho_connections,
user,
online_status,
omikron_id,
omikron_connections,
reset_token,
new_token,
}
impl DataTypes {
pub fn parse(p0: String) -> DataTypes {
for datatype in DataTypes::iter() {
if datatype.to_string().to_lowercase().replace('_', "")
== p0.to_lowercase().replace('_', "")
{
return datatype;
}
}
DataTypes::error_type
}
pub fn to_string(&self) -> String {
return format!("{:?}", self);
}
}
#[derive(PartialEq, Clone, EnumIter, Debug)]
#[allow(non_camel_case_types, dead_code)]
pub enum CommunicationType {
error,
error_anonymous,
error_internal,
error_invalid_data,
error_invalid_user_id,
error_invalid_omikron_id,
error_not_found,
error_not_authenticated,
error_no_iota,
error_invalid_challenge,
error_invalid_secret,
error_invalid_private_key,
error_invalid_public_key,
error_no_user_id,
error_no_call_id,
error_invalid_call_id,
success,
shorten_link,
settings_save,
settings_load,
settings_list,
message,
message_state,
message_send,
message_live,
message_other_iota,
message_chunk,
messages_get,
push_notification,
read_notification,
get_notifications,
change_confirm,
confirm_receive,
confirm_read,
get_chats,
get_states,
add_community,
remove_community,
get_communities,
challenge,
challenge_response,
register,
register_response,
identification,
identification_response,
register_iota,
register_iota_success,
ping,
pong,
add_conversation,
send_chat,
client_changed,
client_connected,
client_disconnected,
client_closed,
public_key,
private_key,
webrtc_sdp,
webrtc_ice,
start_stream,
end_stream,
watch_stream,
call_token,
call_invite,
call_disconnect_user,
call_timeout_user,
call_set_anonymous_joining,
end_call,
function,
update,
create_user,
rho_update,
user_connected,
user_disconnected,
iota_connected,
iota_disconnected,
sync_client_iota_status,
get_user_data,
get_iota_data,
iota_user_data,
change_user_data,
change_iota_data,
get_register,
complete_register_user,
complete_register_iota,
delete_user,
delete_iota,
start_register,
complete_register,
}
impl CommunicationType {
pub fn parse(p0: String) -> CommunicationType {
for datatype in CommunicationType::iter() {
if datatype.to_string().to_lowercase().replace('_', "")
== p0.to_lowercase().replace('_', "")
{
return datatype;
}
}
CommunicationType::error
}
pub fn to_string(&self) -> String {
return format!("{:?}", self);
}
}
#[derive(Debug, Clone)]
pub struct CommunicationValue {
id: Uuid,
comm_type: CommunicationType,
sender: i64,
receiver: i64,
data: HashMap<DataTypes, JsonValue>,
}
#[allow(dead_code)]
impl CommunicationValue {
pub fn new(comm_type: CommunicationType) -> Self {
Self {
id: Uuid::new_v4(),
comm_type,
sender: 0,
receiver: 0,
data: HashMap::new(),
}
}
pub fn with_id(mut self, p0: Uuid) -> Self {
self.id = p0;
self
}
pub fn get_id(&self) -> Uuid {
self.id.clone()
}
pub fn with_sender(mut self, sender: i64) -> Self {
self.sender = sender;
self
}
pub fn get_sender(&self) -> i64 {
self.sender.clone()
}
pub fn with_receiver(mut self, receiver: i64) -> Self {
self.receiver = receiver;
self
}
pub fn get_receiver(&self) -> i64 {
self.receiver.clone()
}
pub fn add_data_num(mut self, key: DataTypes, value: Number) -> Self {
self.data.insert(key, JsonValue::Number(value));
self
}
pub fn add_data_str(mut self, key: DataTypes, value: String) -> Self {
self.data.insert(key, JsonValue::String(value));
self
}
pub fn add_data(mut self, key: DataTypes, value: JsonValue) -> Self {
self.data.insert(key, value);
self
}
pub fn add_array(mut self, key: DataTypes, value: Array) -> Self {
self.data.insert(key, JsonValue::Array(value));
self
}
pub fn get_data(&self, key: DataTypes) -> Option<&JsonValue> {
self.data.get(&key)
}
pub fn get_type(&self) -> CommunicationType {
self.comm_type.clone()
}
pub fn is_type(&self, p0: CommunicationType) -> bool {
self.comm_type == p0
}
pub fn to_json(&self) -> JsonValue {
let mut jdata = object! {};
for (k, v) in &self.data {
jdata[&format!("{:?}", k)] = JsonValue::from(v.clone());
}
if self.sender > 0 && self.receiver > 0 {
object! {
id: self.id.to_string(),
type: format!("{:?}", self.comm_type),
sender: self.sender,
receiver: self.receiver,
data: jdata
}
} else if self.sender > 0 {
object! {
id: self.id.to_string(),
type: format!("{:?}", self.comm_type),
sender: self.sender,
data: jdata
}
} else if self.receiver > 0 {
object! {
id: self.id.to_string(),
type: format!("{:?}", self.comm_type),
receiver: self.receiver,
data: jdata
}
} else {
object! {
id: self.id.to_string(),
type: format!("{:?}", self.comm_type),
data: jdata
}
}
}
pub fn from_json(json_str: &str) -> Self {
if let Ok(parsed) = parse(json_str) {
let comm_type = CommunicationType::parse(parsed["type"].to_string());
let mut sender: i64 = 0;
if parsed.has_key("sender") {
sender = parsed["sender"].as_i64().unwrap_or(0);
}
let mut receiver: i64 = 0;
if parsed.has_key("receiver") {
receiver = parsed["receiver"].as_i64().unwrap_or(0);
}
let uuid =
Uuid::parse_str(parsed["id"].as_str().unwrap_or("")).unwrap_or(Uuid::new_v4());
let mut data = HashMap::new();
if parsed["data"].is_object() {
for (k, v) in parsed["data"].entries() {
data.insert(DataTypes::parse(k.to_string()), v.clone());
}
}
Self {
id: uuid,
comm_type,
sender,
receiver,
data,
}
} else {
Self {
id: Uuid::new_v4(),
comm_type: CommunicationType::error,
sender: 0,
receiver: 0,
data: HashMap::new(),
}
}
}
pub fn forward_to_other_iota(original: &mut CommunicationValue) -> CommunicationValue {
let receiver = original
.get_data(DataTypes::receiver_id)
.unwrap_or(&JsonValue::Number(Number::from(0)))
.as_i64()
.unwrap_or(0);
let now_ms = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let sender = original.get_sender();
CommunicationValue::new(CommunicationType::message_other_iota)
.with_id(original.get_id())
.with_receiver(receiver)
.add_data(
DataTypes::receiver_id,
JsonValue::Number(Number::from(receiver)),
)
.with_sender(sender)
.add_data(DataTypes::send_time, JsonValue::String(now_ms.to_string()))
.add_data(
DataTypes::sender_id,
JsonValue::Number(Number::from(sender)),
)
.add_data(
DataTypes::content,
JsonValue::String(original.get_data(DataTypes::content).unwrap().to_string()),
)
}
}

1
src/data/mod.rs
View file

@ -1,2 +1 @@
pub mod communication;
pub mod user; pub mod user;

200
src/main.rs
View file

@ -5,24 +5,17 @@ mod omega;
mod rho; mod rho;
mod util; mod util;
use async_tungstenite::accept_hdr_async; use std::env;
use dotenv::dotenv; use dotenv::dotenv;
use futures::StreamExt;
use once_cell::sync::Lazy; use once_cell::sync::Lazy;
use std::{env, sync::Arc};
use tokio::net::TcpListener;
use tokio_util::compat::TokioAsyncReadCompatExt;
use tungstenite::handshake::server::{Request, Response};
use crate::{ use crate::{
anonymous_clients::{
anonymous_client_connection::AnonymousClientConnection, anonymous_manager,
},
calls::call_util::garbage_collect_calls, calls::call_util::garbage_collect_calls,
rho::{client_connection::ClientConnection, iota_connection::IotaConnection}, rho::server::start,
util::{ util::{
crypto_helper::{load_public_key, load_secret_key}, crypto_helper::{load_public_key, load_secret_key},
logger::{PrintType, startup}, logger::startup,
}, },
}; };
@ -39,191 +32,8 @@ pub fn get_public_key() -> x448::PublicKey {
async fn main() { async fn main() {
dotenv().ok(); dotenv().ok();
startup(); startup();
let address = format!(
"{}:{}",
env::var("IP").unwrap_or("0.0.0.0".to_string()),
env::var("PORT").unwrap_or("959".to_string())
);
let listener = TcpListener::bind(&address).await.unwrap();
log!( start(959).await;
0,
PrintType::General,
"WebSocket server listening on {}",
address,
);
garbage_collect_calls(); garbage_collect_calls();
while let Ok((stream, _)) = listener.accept().await {
tokio::spawn(async move {
let mut path: String = "/".to_string();
let callback = |req: &Request, response: Response| {
path = req.uri().path().to_string();
Ok(response)
};
let ws_stream = match accept_hdr_async(stream.compat(), callback).await {
Ok(ws) => ws,
Err(e) => {
log!(0, PrintType::General, "WebSocket upgrade failed: {}", e,);
return;
}
};
let (sender, receiver) = ws_stream.split();
if path.starts_with("/ws/client") {
log_in!(0, PrintType::Client, "New Client connection");
let client_conn: Arc<ClientConnection> =
Arc::from(ClientConnection::new(sender, receiver));
loop {
let msg_result = {
let mut session_lock = client_conn.receiver.write().await;
session_lock.next().await
};
match msg_result {
Some(Ok(msg)) => {
if msg.is_text() {
let text = msg.into_text().unwrap();
client_conn.clone().handle_message(text).await;
} else if msg.is_close() {
log_in!(
client_conn.get_user_id().await,
PrintType::Client,
"Client disconnected"
);
client_conn.handle_close().await;
return;
}
}
Some(Err(e)) => {
log_err!(
client_conn.get_user_id().await,
PrintType::Client,
"WebSocket error: {}",
e
);
client_conn.handle_close().await;
return;
}
_ => {
log_in!(
client_conn.get_user_id().await,
PrintType::Client,
"Client stream ended"
);
client_conn.handle_close().await;
return;
}
}
}
} else if path.starts_with("/ws/anonymous_client") {
log_in!(0, PrintType::Client, "New Anonymous Client connection");
let client_conn: Arc<AnonymousClientConnection> =
Arc::from(AnonymousClientConnection::new(sender, receiver));
anonymous_manager::add_anonymous_user(client_conn.clone()).await;
loop {
let msg_result = {
let mut session_lock = client_conn.receiver.write().await;
session_lock.next().await
};
match msg_result {
Some(Ok(msg)) => {
if msg.is_text() {
let text = msg.into_text().unwrap();
client_conn.clone().handle_message(text).await;
} else if msg.is_close() {
log_in!(
client_conn.get_user_id().await,
PrintType::Client,
"Anonymous Client disconnected"
);
anonymous_manager::remove_anonymous_user(
client_conn.get_user_id().await,
)
.await;
client_conn.handle_close().await;
return;
}
}
Some(Err(e)) => {
log_err!(
client_conn.get_user_id().await,
PrintType::Client,
"WebSocket error: {}",
e
);
anonymous_manager::remove_anonymous_user(
client_conn.get_user_id().await,
)
.await;
client_conn.handle_close().await;
return;
}
_ => {
log_in!(
client_conn.get_user_id().await,
PrintType::Client,
"Anonymous Client stream ended"
);
anonymous_manager::remove_anonymous_user(
client_conn.get_user_id().await,
)
.await;
client_conn.handle_close().await;
return;
}
}
}
} else if path.starts_with("/ws/iota") {
log_in!(0, PrintType::Iota, "New Iota connection");
let iota_conn: Arc<IotaConnection> =
Arc::from(IotaConnection::new(sender, receiver));
loop {
let msg_result = {
let mut session_lock = iota_conn.receiver.write().await;
session_lock.next().await
};
match msg_result {
Some(Ok(msg)) => {
if msg.is_text() {
let text = msg.into_text().unwrap();
iota_conn.clone().handle_message(text).await;
} else if msg.is_close() {
log_in!(
iota_conn.get_iota_id().await,
PrintType::Iota,
"Iota disconnected"
);
iota_conn.handle_close().await;
return;
}
}
Some(Err(e)) => {
log_err!(
iota_conn.get_iota_id().await,
PrintType::Iota,
"WebSocket error: {}",
e
);
iota_conn.handle_close().await;
return;
}
_ => {
// Stream ended
log_in!(
iota_conn.get_iota_id().await,
PrintType::Iota,
"Iota stream ended"
);
iota_conn.handle_close().await;
return;
}
}
}
}
});
}
} }

1
src/omega/mod.rs
View file

@ -1,2 +1 @@
pub mod omega_connection; pub mod omega_connection;
pub mod ping_pong_task;

557
src/omega/omega_connection.rs
View file

@ -1,392 +1,165 @@
use async_tungstenite::{ use std::{
WebSocketReceiver, WebSocketSender, sync::Arc,
stream::Stream, time::{Duration, Instant},
tokio::{TokioAdapter, connect_async},
tungstenite::protocol::Message,
}; };
use crossterm::style::Print;
use dashmap::DashMap; use dashmap::DashMap;
use futures::prelude::*;
use json::{JsonValue, number::Number};
use once_cell::sync::Lazy; use once_cell::sync::Lazy;
use std::{collections::HashMap, env, sync::Arc, time::Duration}; use tokio::sync::{Mutex, RwLock, mpsc};
use tokio::{
net::TcpStream,
sync::{Mutex, RwLock, mpsc},
time::{Instant, sleep},
};
use tokio_native_tls::TlsStream;
use uuid::Uuid;
use crate::{ use epsilon_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
data::{ use epsilon_native::{Receiver, Sender, connect};
communication::{CommunicationType, CommunicationValue, DataTypes},
user::UserStatus,
},
get_private_key, log, log_cv_out, log_in, log_out,
rho::rho_manager::{self, RHO_CONNECTIONS, connection_count},
util::{
crypto_helper::{decrypt_b64, secret_key_to_base64},
logger::PrintType,
},
};
use crate::{log_cv_in, log_err};
pub struct WaitingTask { use crate::{data::user::UserStatus, rho::rho_manager};
pub task: Box<dyn Fn(Arc<OmegaConnection>, CommunicationValue) -> bool + Send + Sync>,
pub inserted_at: Instant,
}
pub static WAITING_TASKS: Lazy<DashMap<Uuid, WaitingTask>> = Lazy::new(DashMap::new); static WAITING: Lazy<
DashMap<
u32,
(
Instant,
Box<dyn Fn(Arc<OmegaConnection>, CommunicationValue) -> bool + Send + Sync>,
),
>,
> = Lazy::new(DashMap::new);
static OMEGA_CONNECTION: Lazy<Arc<OmegaConnection>> = Lazy::new(|| { static OMEGA_CONNECTION: Lazy<Arc<OmegaConnection>> = Lazy::new(|| OmegaConnection::new());
let conn = Arc::new(OmegaConnection::new());
let conn_clone = conn.clone();
tokio::spawn(async move {
conn_clone.connect_internal(0).await;
});
tokio::spawn(async {
loop {
sleep(Duration::from_secs(60)).await;
WAITING_TASKS.retain(|_, v| v.inserted_at.elapsed() < Duration::from_secs(60));
}
});
conn
});
#[allow(dead_code)]
pub fn get_omega_connection() -> Arc<OmegaConnection> { pub fn get_omega_connection() -> Arc<OmegaConnection> {
OMEGA_CONNECTION.clone() OMEGA_CONNECTION.clone()
} }
#[derive(Clone)] #[derive(Clone, PartialEq)]
pub struct OmegaConnection { enum State {
write: Arc<
Mutex<
Option<
WebSocketSender<
Stream<TokioAdapter<TcpStream>, TokioAdapter<TlsStream<TcpStream>>>,
>,
>,
>,
>,
read: Arc<
Mutex<
Option<
WebSocketReceiver<
Stream<TokioAdapter<TcpStream>, TokioAdapter<TlsStream<TcpStream>>>,
>,
>,
>,
>,
pingpong: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
pub last_ping: Arc<Mutex<i64>>,
pub message_send_times: Arc<Mutex<HashMap<Uuid, Instant>>>,
state: Arc<RwLock<ConnectionState>>,
}
#[derive(Clone, PartialEq, Eq)]
enum ConnectionState {
Disconnected, Disconnected,
Connecting, Connecting,
Connected, Connected,
} }
pub struct OmegaConnection {
sender: Arc<Mutex<Option<Sender>>>,
receiver: Arc<Mutex<Option<Receiver>>>,
state: Arc<RwLock<State>>,
}
impl OmegaConnection { impl OmegaConnection {
pub fn new() -> Self { pub fn new() -> Arc<Self> {
OmegaConnection { Arc::new(Self {
read: Arc::new(Mutex::new(None)), sender: Arc::new(Mutex::new(None)),
write: Arc::new(Mutex::new(None)), receiver: Arc::new(Mutex::new(None)),
pingpong: Arc::new(Mutex::new(None)), state: Arc::new(RwLock::new(State::Disconnected)),
last_ping: Arc::new(Mutex::new(-1)), })
message_send_times: Arc::new(Mutex::new(HashMap::new())),
state: Arc::new(RwLock::new(ConnectionState::Disconnected)),
}
} }
async fn connect_internal(self: Arc<OmegaConnection>, mut retry: usize) {
if self.state.read().await.eq(&ConnectionState::Connected) {
return;
}
if self.state.read().await.eq(&ConnectionState::Connecting) {
let start = Instant::now();
let timeout = Duration::from_secs(10);
while self.state.read().await.eq(&ConnectionState::Connecting)
&& start.elapsed() < timeout
{
sleep(Duration::from_millis(100)).await;
}
return;
}
*self.state.write().await = ConnectionState::Connecting; pub async fn connect(self: Arc<Self>, addr: &str) -> Result<(), String> {
*self.state.write().await = State::Connecting;
if let Some(handle) = self.pingpong.lock().await.take() { let (sender, receiver) = connect(addr)
handle.abort(); .await
} .map_err(|e| format!("Connect error: {e:?}"))?;
loop { *self.sender.lock().await = Some(sender);
if retry > 500 { *self.receiver.lock().await = Some(receiver);
log_err!(
0,
PrintType::Omega,
"Max retry attempts reached, giving up."
);
*self.state.write().await = ConnectionState::Disconnected;
return;
}
let url_str = *self.state.write().await = State::Connected;
env::var("OMEGA_HOST").unwrap_or("wss://omega.tensamin.net/ws/omikron".to_string());
match connect_async(&url_str).await {
Ok((ws_stream, _)) => {
log_in!(0, PrintType::Omega, "WebSocket connected to {}", url_str);
retry = 0;
let (write, read) = ws_stream.split();
*self.read.lock().await = Some(read);
*self.write.lock().await = Some(write);
*self.state.write().await = ConnectionState::Connected; let read_self = self.clone();
tokio::spawn(async move {
read_self.read_loop().await;
});
let read_loop_self = self.clone(); self.identify().await?;
let read_loop_handle = tokio::spawn(async move {
read_loop_self.read_loop().await;
});
let cloned_self = self.clone(); Ok(())
tokio::spawn(async move { }
let id = Uuid::new_v4();
let identify_msg = async fn identify(&self) -> Result<(), String> {
CommunicationValue::new(CommunicationType::identification) let msg = CommunicationValue::new(CommunicationType::identification)
.with_id(id) .add_data(DataTypes::omikron, DataValue::Number(1));
.add_data(
DataTypes::omikron,
JsonValue::Number(Number::from(
env::var("ID")
.unwrap_or("0".to_string())
.parse::<i64>()
.unwrap_or(0),
)),
);
WAITING_TASKS.insert(
id,
WaitingTask {
task: Box::new(|selfc, cv| {
if cv.is_type(CommunicationType::error_not_found) {
log_err!(0,
PrintType::Omega,
"Identification failed: Omikron ID not found on Omega.",
);
return false;
}
if !cv.is_type(CommunicationType::challenge) {
return false;
}
tokio::spawn(async move {
let task = async move {
let challenge = cv
.get_data(DataTypes::challenge)
.and_then(|v| v.as_str())
.ok_or_else(|| {
"Challenge not found or not a string".to_string()
})?;
let server_pub_key = cv self.await_response(&msg, Some(Duration::from_secs(10)))
.get_data(DataTypes::public_key) .await?;
.and_then(|v| v.as_str())
.ok_or_else(|| {
"Public key from server not found or not a string"
.to_string()
})?;
let decrypted_challenge = decrypt_b64( Ok(())
&secret_key_to_base64(&get_private_key()),
server_pub_key,
challenge,
)
.map_err(|e| {
format!("Failed to decrypt challenge: {:?}", e)
})?;
let response_msg = CommunicationValue::new(
CommunicationType::challenge_response,
)
.with_id(cv.get_id())
.add_data(
DataTypes::challenge,
JsonValue::String(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::identification_response)
{
log_err!(0,
PrintType::Omega,
"Expected identification_response, got something else.",
);
return false;
}
if let Some(accepted) = final_cv.get_data(DataTypes::accepted).and_then(|v| v.as_bool()) {
if !accepted {
log_err!(0, PrintType::Omega, "Omega did not accept identification.");
return false;
}
} else {
log_err!(0, PrintType::Omega, "Omega response did not contain 'accepted' field.");
return false;
}
tokio::spawn(async move {
let mut connected_iota_ids: Vec<JsonValue> = Vec::new();
let mut connected_user_ids: Vec<JsonValue> = Vec::new();
let rho_connections_reader = RHO_CONNECTIONS.read().await;
for iota_id in rho_connections_reader.keys() {
connected_iota_ids.push(JsonValue::from(*iota_id));
}
for rho in rho_connections_reader.values() {
for client_conn in rho.get_client_connections().await {
connected_user_ids.push(JsonValue::from(client_conn.get_user_id().await));
}
}
drop(rho_connections_reader);
let sync_msg = CommunicationValue::new(CommunicationType::sync_client_iota_status)
.add_data(DataTypes::iota_ids, JsonValue::Array(connected_iota_ids))
.add_data(DataTypes::user_ids, JsonValue::Array(connected_user_ids))
.add_data(DataTypes::rho_connections, JsonValue::from(connection_count().await));
selfc.send_message(&sync_msg).await;
});
log!(0,
PrintType::Omega,
"Successfully identified with Omega.",
);
true
}),
inserted_at: Instant::now(),
}
);
selfc.send_message(&response_msg).await;
Ok::<(), String>(())
};
if let Err(e) = task.await {
log_err!(0, PrintType::Omega, "{}", &e);
}
});
true
}),
inserted_at: Instant::now(),
}
);
cloned_self.send_message(&identify_msg).await
});
let ping_pong_self = self.clone();
let handle = tokio::spawn(async move {
loop {
if *ping_pong_self.state.read().await != ConnectionState::Connected {
break;
}
ping_pong_self.send_ping().await;
sleep(Duration::from_secs(5)).await;
}
});
*self.pingpong.lock().await = Some(handle);
read_loop_handle.await.unwrap_or_else(|e| {
log_err!(0, PrintType::Omega, "Read loop task failed: {}", e)
});
*self.read.lock().await = None;
*self.write.lock().await = None;
*self.state.write().await = ConnectionState::Disconnected;
log_err!(0, PrintType::Omega, "Connection lost. Retrying...");
retry += 1;
sleep(Duration::from_secs(2)).await;
}
Err(e) => {
*self.state.write().await = ConnectionState::Disconnected;
log_err!(
0,
PrintType::Omega,
"WebSocket connection failed (attempt {}): {}",
retry + 1,
e,
);
retry += 1;
sleep(Duration::from_secs(2)).await;
continue;
}
}
}
} }
async fn read_loop(self: Arc<Self>) { async fn read_loop(self: Arc<Self>) {
let mut reader = match self.read.lock().await.take() {
Some(reader) => reader,
None => return,
};
loop { loop {
let msg = reader.next().await; let result = {
let mut guard = self.receiver.lock().await;
match msg { match guard.as_mut() {
Some(Ok(Message::Text(msg))) => { Some(receiver) => receiver.receive().await,
let cv = CommunicationValue::from_json(&msg); None => return,
log_cv_in!(PrintType::Omega, cv);
if cv.is_type(CommunicationType::pong) || cv.is_type(CommunicationType::ping) {
self.handle_pong(&cv, true).await;
continue;
}
let msg_id = cv.get_id();
if let Some(task) = WAITING_TASKS.remove(&msg_id) {
if (task.1.task)(self.clone(), cv.clone()) {
continue;
}
} else {
}
} }
#[allow(non_snake_case)] };
Some(Ok(Message::Close(_))) | None => break,
Some(Err(_)) => break, let cv = match result {
_ => {} Ok(v) => v,
Err(_) => {
*self.state.write().await = State::Disconnected;
return;
}
};
let id = cv.get_id();
if let Some((_, task)) = WAITING.remove(&id) {
(task.1)(self.clone(), cv);
continue;
}
if cv.is_type(CommunicationType::ping) {
let pong = CommunicationValue::new(CommunicationType::pong).with_id(id);
let _ = self.send(&pong).await;
} }
} }
} }
pub async fn send_message(&self, cv: &CommunicationValue) { pub async fn send(&self, cv: &CommunicationValue) -> Result<(), String> {
log_cv_out!(PrintType::Omega, cv); let guard = self.sender.lock().await;
let msg = cv.to_json().to_string(); if let Some(sender) = guard.as_ref() {
sender
let mut guard = self.write.lock().await; .send(cv)
if let Some(ws) = guard.as_mut() { .await
let _ = ws.send(Message::Text(msg.into())).await; .map_err(|e| format!("Send error: {e:?}"))
} else {
Err("Not connected".into())
} }
} }
pub async fn await_response(
&self,
cv: &CommunicationValue,
timeout: Option<Duration>,
) -> Result<CommunicationValue, String> {
let (tx, mut rx) = mpsc::channel(1);
let id = cv.get_id();
WAITING.insert(
id.into(),
(
Instant::now(),
Box::new(move |_, response| {
let _ = tx.try_send(response);
true
}),
),
);
self.send(cv).await?;
match tokio::time::timeout(timeout.unwrap_or(Duration::from_secs(10)), rx.recv()).await {
Ok(Some(v)) => Ok(v),
_ => {
WAITING.remove(&id.into());
Err("Timeout waiting for response".into())
}
}
}
pub async fn close_iota(iota_id: i64) { pub async fn close_iota(iota_id: i64) {
let cv = CommunicationValue::new(CommunicationType::iota_disconnected) let cv = CommunicationValue::new(CommunicationType::iota_disconnected)
.add_data(DataTypes::iota_id, JsonValue::from(iota_id)); .add_data(DataTypes::iota_id, DataValue::Number(iota_id));
OmegaConnection::send_global(cv).await; OmegaConnection::send_global(cv).await;
} }
pub async fn client_changed(_iota_id: i64, user_id: i64, state: UserStatus) { pub async fn client_changed(_iota_id: i64, user_id: i64, state: UserStatus) {
let msg_type = match state { let msg_type = match state {
UserStatus::iota_offline => Some(CommunicationType::user_disconnected), UserStatus::iota_offline => Some(CommunicationType::user_disconnected),
@ -396,25 +169,28 @@ impl OmegaConnection {
if let Some(t) = msg_type { if let Some(t) = msg_type {
let cv = let cv =
CommunicationValue::new(t).add_data(DataTypes::user_id, JsonValue::from(user_id)); CommunicationValue::new(t).add_data(DataTypes::user_id, DataValue::Number(user_id));
OmegaConnection::send_global(cv).await; OmegaConnection::send_global(cv).await;
} }
} }
async fn send_global(cv: CommunicationValue) {
OMEGA_CONNECTION.send(&cv).await;
}
pub async fn user_states(user_id: i64, user_ids: Vec<i64>) { pub async fn user_states(user_id: i64, user_ids: Vec<i64>) {
let user_ids_str = user_ids let user_ids_str = user_ids
.iter() .iter()
.map(|id| id.to_string()) .map(|id| DataValue::Number(*id))
.collect::<Vec<_>>() .collect::<Vec<_>>();
.join(",");
let cv = CommunicationValue::new(CommunicationType::get_states) let cv = CommunicationValue::new(CommunicationType::get_states)
.add_data(DataTypes::user_ids, JsonValue::from(user_ids_str)); .add_data(DataTypes::user_ids, DataValue::Array(user_ids_str));
let msg_id = cv.get_id(); let msg_id = cv.get_id();
WAITING_TASKS.insert( WAITING.insert(
msg_id, msg_id,
WaitingTask { (
task: Box::new( Instant::now(),
Box::new(
move |_: Arc<OmegaConnection>, response: CommunicationValue| { move |_: Arc<OmegaConnection>, response: CommunicationValue| {
tokio::spawn(async move { tokio::spawn(async move {
let rho = rho_manager::get_rho_con_for_user(user_id).await; let rho = rho_manager::get_rho_con_for_user(user_id).await;
@ -427,84 +203,9 @@ impl OmegaConnection {
true true
}, },
), ),
inserted_at: Instant::now(), ),
},
); );
OmegaConnection::send_global(cv).await; OmegaConnection::send_global(cv).await;
} }
async fn send_global(cv: CommunicationValue) {
OMEGA_CONNECTION.send_message(&cv).await;
}
pub async fn await_connection(&self, timeout_duration: Option<Duration>) -> Result<(), String> {
if *self.state.read().await == ConnectionState::Connected {
return Ok(());
}
let timeout = timeout_duration.unwrap_or(Duration::from_secs(10));
let start = Instant::now();
loop {
if *self.state.read().await == ConnectionState::Connected {
return Ok(());
}
if start.elapsed() >= timeout {
return Err(format!(
"Connection not established within {} seconds",
timeout.as_secs()
));
}
sleep(Duration::from_millis(100)).await; // short interval polling
}
}
pub async fn await_response(
&self,
cv: &CommunicationValue,
timeout_duration: Option<Duration>,
) -> Result<CommunicationValue, String> {
self.await_connection(timeout_duration).await?;
let (tx, mut rx) = mpsc::channel(1);
let msg_id = cv.get_id();
let task_tx = tx.clone();
WAITING_TASKS.insert(
msg_id,
WaitingTask {
task: Box::new(move |_, response_cv| {
let inner_tx = task_tx.clone();
tokio::spawn(async move {
if inner_tx.send(response_cv).await.is_err() {
log_err!(
0,
PrintType::Omega,
"Failed to send response back to awaiter",
);
}
});
true
}),
inserted_at: Instant::now(),
},
);
self.send_message(cv).await;
let timeout = timeout_duration.unwrap_or(Duration::from_secs(10));
match tokio::time::timeout(timeout, rx.recv()).await {
Ok(Some(response_cv)) => Ok(response_cv),
Ok(_) => Err("Failed to receive response, channel was closed.".to_string()),
Err(_) => {
WAITING_TASKS.remove(&msg_id);
Err(format!(
"Request timed out after {} seconds.",
timeout.as_secs()
))
}
}
}
} }

23
src/omega/ping_pong_task.rs
View file

@ -1,30 +1,29 @@
use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes}; use epsilon_core::{CommunicationType, CommunicationValue, DataTypes, DataValue, rand_u32};
use crate::omega::omega_connection::OmegaConnection;
use json::number::Number;
use std::time::Duration; use std::time::Duration;
use tokio::time::Instant; use tokio::time::Instant;
use uuid::Uuid;
use crate::omega::omega_connection::OmegaConnection;
const PING_TIMEOUT: Duration = Duration::from_secs(30); const PING_TIMEOUT: Duration = Duration::from_secs(30);
impl OmegaConnection { impl OmegaConnection {
pub async fn send_ping(&self) { pub async fn send_ping(&self) {
let uuid = Uuid::new_v4(); let id = rand_u32();
let send_time = Instant::now(); let send_time = Instant::now();
let mut message_send_times = self.message_send_times.lock().await; let mut message_send_times = self.message_send_times.lock().await;
message_send_times.retain(|_uuid, time| time.elapsed() < PING_TIMEOUT); message_send_times.retain(|_uuid, time| time.elapsed() < PING_TIMEOUT);
message_send_times.insert(uuid, send_time); message_send_times.insert(id as i64, send_time);
self.send_ping_message(uuid).await; self.send_ping_message(id).await;
} }
pub async fn send_ping_message(&self, uuid: Uuid) { pub async fn send_ping_message(&self, id: u32) {
let ping_message = CommunicationValue::new(CommunicationType::ping) let ping_message = CommunicationValue::new(CommunicationType::ping)
.with_id(uuid) .with_id(id)
.add_data_num( .add_data(
DataTypes::last_ping, DataTypes::last_ping,
Number::from(*self.last_ping.lock().await), DataValue::Number(self.last_ping.lock().await.unwrap()),
); );
self.send_message(&ping_message).await; self.send_message(&ping_message).await;
@ -34,7 +33,7 @@ impl OmegaConnection {
pub async fn handle_pong(&self, cv: &CommunicationValue, _log: bool) { pub async fn handle_pong(&self, cv: &CommunicationValue, _log: bool) {
let id = cv.get_id(); let id = cv.get_id();
let mut message_send_times = self.message_send_times.lock().await; let mut message_send_times = self.message_send_times.lock().await;
if let Some(send_time) = message_send_times.remove(&id) { if let Some(send_time) = message_send_times.remove(&(id as i64)) {
let ping = Instant::now().duration_since(send_time).as_millis() as i64; let ping = Instant::now().duration_since(send_time).as_millis() as i64;
*self.last_ping.lock().await = ping; *self.last_ping.lock().await = ping;
} }

496
src/rho/client_connection.rs
View file

@ -1,43 +1,24 @@
use async_tungstenite::tungstenite::Message;
use async_tungstenite::{WebSocketReceiver, WebSocketSender};
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use json::JsonValue;
use json::number::Number;
use rand::Rng;
use rand::distributions::Alphanumeric;
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use sysinfo::System;
use tokio::sync::RwLock;
use tokio_util::compat::Compat;
use tungstenite::Utf8Bytes;
use uuid::Uuid;
use super::{rho_connection::RhoConnection, rho_manager};
use crate::anonymous_clients::anonymous_manager; use crate::anonymous_clients::anonymous_manager;
use crate::calls::{call_manager, call_util}; use crate::calls::{call_manager, call_util};
use crate::omega::omega_connection::get_omega_connection; use crate::omega::omega_connection::get_omega_connection;
use crate::util::crypto_helper::{load_public_key, public_key_to_base64}; use crate::rho::connection::GeneralConnection;
use crate::util::crypto_util::{DataFormat, SecurePayload}; use crate::rho::{rho_connection::RhoConnection, rho_manager};
use crate::util::logger::PrintType; use crate::util::logger::PrintType;
use crate::{ use crate::{data::user::UserStatus, omega::omega_connection::OmegaConnection};
data::{ use crate::{log_cv_in, log_cv_out, log_out};
communication::{CommunicationType, CommunicationValue, DataTypes}, use epsilon_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
user::UserStatus, use epsilon_native::{Receiver, Sender};
}, use std::str::FromStr;
omega::omega_connection::OmegaConnection, use std::sync::Arc;
}; use std::time::{Duration, SystemTime, UNIX_EPOCH};
use crate::{get_private_key, get_public_key, log_cv_in, log_in, log_out}; use tokio::sync::RwLock;
use uuid::Uuid;
pub struct ClientConnection { pub struct ClientConnection {
pub sender: Arc<RwLock<WebSocketSender<Compat<tokio::net::TcpStream>>>>, pub user_id: u64,
pub receiver: Arc<RwLock<WebSocketReceiver<Compat<tokio::net::TcpStream>>>>,
pub user_id: Arc<RwLock<i64>>, pub sender: Arc<Sender>,
identified: Arc<RwLock<bool>>, pub receiver: Arc<Receiver>,
challenged: Arc<RwLock<bool>>,
challenge: Arc<RwLock<String>>,
pub ping: Arc<RwLock<i64>>, pub ping: Arc<RwLock<i64>>,
pub_key: Arc<RwLock<Option<Vec<u8>>>>, pub_key: Arc<RwLock<Option<Vec<u8>>>>,
pub rho_connection: Arc<RwLock<Option<Arc<RhoConnection>>>>, pub rho_connection: Arc<RwLock<Option<Arc<RhoConnection>>>>,
@ -46,34 +27,30 @@ pub struct ClientConnection {
} }
impl ClientConnection { impl ClientConnection {
/// Create a new ClientConnection pub async fn from_general(general: Arc<GeneralConnection>, user_id: u64) -> Arc<Self> {
pub fn new(
sender: WebSocketSender<Compat<tokio::net::TcpStream>>,
receiver: WebSocketReceiver<Compat<tokio::net::TcpStream>>,
) -> Arc<Self> {
Arc::new(Self { Arc::new(Self {
sender: Arc::new(RwLock::new(sender)), ping: Arc::new(RwLock::new(0)),
receiver: Arc::new(RwLock::new(receiver)),
user_id: Arc::new(RwLock::new(0)),
identified: Arc::new(RwLock::new(false)),
challenged: Arc::new(RwLock::new(false)),
challenge: Arc::new(RwLock::new(String::new())),
ping: Arc::new(RwLock::new(-1)),
pub_key: Arc::new(RwLock::new(None)), pub_key: Arc::new(RwLock::new(None)),
rho_connection: Arc::new(RwLock::new(None)), rho_connection: Arc::new(RwLock::new(None)),
interested_users: Arc::new(RwLock::new(Vec::new())), interested_users: Arc::new(RwLock::new(Vec::new())),
is_open: Arc::new(RwLock::new(true)), is_open: Arc::new(RwLock::new(true)),
sender: general.sender.clone(),
receiver: general.receiver.clone(),
user_id: user_id,
}) })
} }
pub fn start(self: Arc<Self>) {
let self_clone = self.clone();
tokio::spawn(async move {
while let Ok(cv) = self_clone.receiver.receive().await {
self_clone.clone().handle_message(cv).await;
}
});
}
/// Get the user ID /// Get the user ID
pub async fn get_user_id(&self) -> i64 { pub async fn get_user_id(&self) -> u64 {
*self.user_id.read().await self.user_id
}
/// Check if connection is identified
pub async fn is_identified(&self) -> bool {
*self.identified.read().await
} }
/// Get current ping /// Get current ping
@ -86,211 +63,30 @@ impl ClientConnection {
self.rho_connection.read().await.clone() self.rho_connection.read().await.clone()
} }
/// Send a string message to the client
pub async fn send_message_str(self: Arc<Self>, message: &str) {
let mut session = self.sender.write().await;
if let Err(e) = session
.send(Message::Text(Utf8Bytes::from(message.to_string())))
.await
{
log_out!(
self.get_user_id().await,
PrintType::Client,
"Failed to send message to client: {}",
e,
);
}
}
/// Send a CommunicationValue to the client /// Send a CommunicationValue to the client
pub async fn send_message(self: Arc<Self>, cv: &CommunicationValue) { pub async fn send_message(self: Arc<Self>, cv: &CommunicationValue) {
if !*self.is_open.read().await { if !*self.is_open.read().await {
log_out!( log_out!(
self.get_user_id().await, self.user_id as i64,
PrintType::Client, PrintType::Client,
"Attempted to send message to a closed connection." "Attempted to send message to a closed connection."
); );
return; return;
} }
if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) { if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) {
log_out!( log_cv_out!(PrintType::Client, &cv);
self.get_user_id().await,
PrintType::Client,
"{}",
&cv.to_json().to_string()
);
} }
self.send_message_str(&cv.to_json().to_string()).await; self.sender.send(&cv).await;
} }
/// Handle incoming message from client /// Handle incoming message from client
pub async fn handle_message(self: Arc<Self>, message: Utf8Bytes) { pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
tokio::spawn(async move { tokio::spawn(async move {
let cv = CommunicationValue::from_json(&message);
if cv.is_type(CommunicationType::ping) { if cv.is_type(CommunicationType::ping) {
self.handle_ping(cv).await; self.handle_ping(cv).await;
return; return;
} }
log_cv_in!(PrintType::Client, cv); log_cv_in!(PrintType::Client, cv);
let identified = *self.identified.read().await;
let challenged = *self.challenged.read().await;
// Handle identification
if !identified && cv.is_type(CommunicationType::identification) {
let user_id = cv
.get_data(DataTypes::user_id)
.and_then(|v| v.as_i64())
.unwrap_or(0);
if user_id == 0 {
log_out!(
self.get_user_id().await,
PrintType::Client,
"Invalid USER ID"
);
self.clone()
.send_error_response(&cv.get_id(), CommunicationType::error_invalid_data)
.await;
self.close().await;
return;
}
*self.user_id.write().await = user_id;
let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_user_data)
.with_id(cv.get_id())
.add_data(DataTypes::user_id, JsonValue::from(user_id));
let response_cv = get_omega_connection()
.await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
.await;
if let Ok(response_cv) = response_cv {
if !response_cv.is_type(CommunicationType::get_user_data) {
self.clone()
.send_error_response(&cv.get_id(), CommunicationType::error_internal)
.await;
self.close().await;
return;
}
let base64_pub = response_cv
.get_data(DataTypes::public_key)
.and_then(|v| v.as_str())
.unwrap_or("");
let pub_key = match load_public_key(base64_pub) {
Some(pk) => pk,
_ => {
self.clone()
.send_error_response(
&cv.get_id(),
CommunicationType::error_invalid_public_key,
)
.await;
self.close().await;
return;
}
};
*self.pub_key.write().await = Some(pub_key.as_bytes().to_vec());
let challenge: String = rand::thread_rng()
.sample_iter(&Alphanumeric)
.take(32)
.map(char::from)
.collect();
*self.challenge.write().await = challenge.clone();
let encrypted_challenge =
SecurePayload::new(&challenge, DataFormat::Raw, get_private_key())
.unwrap()
.encrypt_x448(pub_key)
.unwrap()
.export(DataFormat::Base64);
*self.identified.write().await = true;
let challenge_msg = CommunicationValue::new(CommunicationType::challenge)
.with_id(cv.get_id())
.add_data_str(
DataTypes::public_key,
public_key_to_base64(&get_public_key()),
)
.add_data_str(DataTypes::challenge, encrypted_challenge);
self.send_message(&challenge_msg).await;
} else {
self.clone()
.send_error_response(&cv.get_id(), CommunicationType::error_internal)
.await;
self.close().await;
return;
}
return;
}
if identified && !challenged && cv.is_type(CommunicationType::challenge_response) {
let client_response = cv
.get_data(DataTypes::challenge)
.and_then(|v| v.as_str())
.unwrap_or("");
let debase64d = STANDARD.decode(&client_response).unwrap();
if String::from_utf8(debase64d.clone()).unwrap() == *self.challenge.read().await {
*self.challenged.write().await = true;
let user_id = self.get_user_id().await;
let rho_connection = match rho_manager::get_rho_con_for_user(user_id).await {
Some(rho) => rho,
_ => {
self.send_error_response(
&cv.get_id(),
CommunicationType::error_no_iota,
)
.await;
return;
}
};
rho_connection.add_client_connection(self.clone()).await;
// Set identification data
{
let mut user_id_guard = self.user_id.write().await;
*user_id_guard = user_id;
}
{
let mut identified_guard = self.identified.write().await;
*identified_guard = true;
}
*self.rho_connection.write().await = Some(Arc::clone(&rho_connection));
let response =
CommunicationValue::new(CommunicationType::identification_response)
.with_id(cv.get_id());
self.send_message(&response).await;
} else {
self.clone()
.send_error_response(
&cv.get_id(),
CommunicationType::error_not_authenticated,
)
.await;
self.close().await;
return;
}
return;
}
if !self.is_identified().await {
self.clone()
.send_error_response(&cv.get_id(), CommunicationType::error_not_authenticated)
.await;
self.close().await;
return;
}
// Handle client status changes // Handle client status changes
if cv.is_type(CommunicationType::client_changed) { if cv.is_type(CommunicationType::client_changed) {
@ -326,17 +122,9 @@ impl ClientConnection {
} }
if cv.is_type(CommunicationType::get_user_data) { if cv.is_type(CommunicationType::get_user_data) {
if let Some(anonymous) = { if let Some(anonymous) = {
if let Some(user_id) = cv if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
.get_data(DataTypes::user_id) anonymous_manager::get_anonymous_user(user_id as u64).await
.unwrap_or(&JsonValue::Null) } else if let Some(username) = cv.get_data(DataTypes::username).as_str() {
.as_i64()
{
anonymous_manager::get_anonymous_user(user_id).await
} else if let Some(username) = cv
.get_data(DataTypes::username)
.unwrap_or(&JsonValue::Null)
.as_str()
{
anonymous_manager::get_anonymous_user_by_name(username.to_string()).await anonymous_manager::get_anonymous_user_by_name(username.to_string()).await
} else { } else {
None None
@ -344,14 +132,23 @@ impl ClientConnection {
} { } {
let response = CommunicationValue::new(CommunicationType::get_user_data) let response = CommunicationValue::new(CommunicationType::get_user_data)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_data_str(DataTypes::username, anonymous.get_user_name().await) .add_data(
DataTypes::username,
DataValue::Str(anonymous.get_user_name().await),
)
.add_data( .add_data(
DataTypes::user_id, DataTypes::user_id,
JsonValue::Number(Number::from(anonymous.get_user_id().await)), DataValue::Number(anonymous.get_user_id() as i64),
) )
.add_data_str(DataTypes::display, anonymous.get_display_name().await) .add_data(
.add_data_str(DataTypes::user_state, "online".to_string()) DataTypes::display,
.add_data_str(DataTypes::avatar, anonymous.get_avatar().await); 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()));
self.send_message(&response).await; self.send_message(&response).await;
@ -367,7 +164,8 @@ impl ClientConnection {
|| cv.is_type(CommunicationType::delete_user) || cv.is_type(CommunicationType::delete_user)
{ {
let sender = self.get_user_id().await; let sender = self.get_user_id().await;
self.handle_omega_forward(cv.with_sender(sender)).await; self.handle_omega_forward(cv.with_sender(sender as u64))
.await;
return; return;
} }
// Forward other messages to Iota // Forward other messages to Iota
@ -378,10 +176,7 @@ impl ClientConnection {
let client_for_closure = self.clone(); let client_for_closure = self.clone();
tokio::spawn(async move { tokio::spawn(async move {
let response_cv = get_omega_connection() let response_cv = get_omega_connection()
.await_response( .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20)))
&cv.with_sender(*self.user_id.read().await),
Some(Duration::from_secs(20)),
)
.await; .await;
if let Ok(response_cv) = response_cv { if let Ok(response_cv) = response_cv {
client_for_closure.send_message(&response_cv).await; client_for_closure.send_message(&response_cv).await;
@ -392,13 +187,13 @@ impl ClientConnection {
/// Handle ping message /// Handle ping message
async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) { async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) {
// Update our ping if provided // Update our ping if provided
if let Some(last_ping) = cv.get_data(DataTypes::last_ping) { if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
let current = SystemTime::now() let current = SystemTime::now()
.duration_since(UNIX_EPOCH) .duration_since(UNIX_EPOCH)
.unwrap() .unwrap()
.as_millis(); .as_millis();
let mut ping_guard = self.ping.write().await; let mut ping_guard = self.ping.write().await;
*ping_guard = current as i64 - last_ping.as_i64().unwrap(); *ping_guard = current as i64 - last_ping;
} }
// Get Iota ping from RhoConnection // Get Iota ping from RhoConnection
@ -411,7 +206,7 @@ impl ClientConnection {
// Send pong response // Send pong response
let response = CommunicationValue::new(CommunicationType::pong) let response = CommunicationValue::new(CommunicationType::pong)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_data(DataTypes::ping_iota, JsonValue::from(iota_ping)); .add_data(DataTypes::ping_iota, DataValue::Number(iota_ping));
self.send_message(&response).await; self.send_message(&response).await;
} }
@ -419,51 +214,47 @@ impl ClientConnection {
/// Handle client status change /// Handle client status change
async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) { async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) {
let user_id = self.get_user_id().await; let user_id = self.get_user_id().await;
if let Some(_status_str) = cv.get_data(DataTypes::user_state) { if let DataValue::Str(_status_str) = cv.get_data(DataTypes::user_state) {
let user_status = UserStatus::user_online; let user_status = UserStatus::user_online;
if let Some(rho_conn) = self.get_rho_connection().await { if let Some(rho_conn) = self.get_rho_connection().await {
OmegaConnection::client_changed(rho_conn.get_iota_id().await, user_id, user_status) OmegaConnection::client_changed(
.await; rho_conn.get_iota_id().await as i64,
user_id as i64,
user_status,
)
.await;
} }
} }
} }
/// Handle call invite /// Handle call invite
async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
let receiver_id: i64 = cv let receiver_id: i64 = cv.get_data(DataTypes::receiver_id).as_number().unwrap_or(0);
.get_data(DataTypes::receiver_id)
.unwrap_or(&json::JsonValue::Number(Number::from(0)))
.as_i64()
.unwrap_or(0);
if receiver_id == 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::error_no_user_id)
.await; .await;
return; return;
} }
let call_id = match cv.get_data(DataTypes::call_id) { let call_id = match cv.get_data(DataTypes::call_id) {
Some(id_str) => match Uuid::parse_str(&id_str.to_string()) { DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) {
Ok(id) => id, Ok(id) => id,
Err(_) => { Err(_) => {
self.send_error_response( self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
&cv.get_id(), .await;
CommunicationType::error_invalid_call_id,
)
.await;
return; return;
} }
}, },
_ => { _ => {
self.send_error_response(&cv.get_id(), CommunicationType::error_no_call_id) self.send_error_response(cv.get_id(), CommunicationType::error_no_call_id)
.await; .await;
return; return;
} }
}; };
let invited = let invited = call_manager::add_invite(call_id, self.user_id, receiver_id as u64).await;
call_manager::add_invite(call_id, *self.user_id.read().await, receiver_id).await;
if !invited { if !invited {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_call_id) self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
.await; .await;
return; return;
} }
@ -472,7 +263,7 @@ impl ClientConnection {
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).await {
Some(rho) => rho, Some(rho) => rho,
_ => { _ => {
self.send_error_response(&cv.get_id(), CommunicationType::error) self.send_error_response(cv.get_id(), CommunicationType::error)
.await; .await;
return; return;
} }
@ -483,11 +274,14 @@ impl ClientConnection {
// Create and send call distribution message // Create and send call distribution message
let forward = CommunicationValue::new(CommunicationType::call_invite) let forward = CommunicationValue::new(CommunicationType::call_invite)
.with_receiver(receiver_id) .with_receiver(receiver_id as u64)
.with_sender(sender_id) .with_sender(sender_id as u64)
.add_data_str(DataTypes::call_id, call_id.to_string()) .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
.add_data_str(DataTypes::receiver_id, receiver_id.to_string()) .add_data(
.add_data_str(DataTypes::sender_id, sender_id.to_string()); DataTypes::receiver_id,
DataValue::Str(receiver_id.to_string()),
)
.add_data(DataTypes::sender_id, DataValue::Str(sender_id.to_string()));
target_rho.message_to_client(forward).await; target_rho.message_to_client(forward).await;
@ -500,16 +294,16 @@ impl ClientConnection {
let user_id = self.get_user_id().await; 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(DataTypes::call_id) {
Some(id_str) => match Uuid::parse_str(&id_str.to_string()) { DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) {
Ok(id) => id, Ok(id) => id,
Err(_) => { Err(_) => {
self.send_error_response(&cv.get_id(), CommunicationType::error) self.send_error_response(cv.get_id(), CommunicationType::error)
.await; .await;
return; return;
} }
}, },
_ => { _ => {
self.send_error_response(&cv.get_id(), CommunicationType::error) self.send_error_response(cv.get_id(), CommunicationType::error)
.await; .await;
return; return;
} }
@ -518,33 +312,20 @@ impl ClientConnection {
if let Some(token) = call_manager::get_call_token(user_id, call_id).await { 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::call_token)
.with_id(cv.get_id()) .with_id(cv.get_id())
.with_receiver(user_id) .with_receiver(user_id as u64)
.add_data_str(DataTypes::call_token, token); .add_data(DataTypes::call_token, DataValue::Str(token));
self.send_message(&response).await; self.send_message(&response).await;
} else { } else {
self.send_error_response(&cv.get_id(), CommunicationType::error) self.send_error_response(cv.get_id(), CommunicationType::error)
.await; .await;
return; return;
} }
} }
async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) {
let call_id = Uuid::from_str( let call_id =
cv.get_data(DataTypes::call_id) Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap();
.unwrap_or(&JsonValue::Null) let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
.as_str() let untill = cv.get_data(DataTypes::untill).as_number().unwrap_or(0);
.unwrap_or(""),
)
.unwrap();
let user_id = cv
.get_data(DataTypes::user_id)
.unwrap_or(&JsonValue::Null)
.as_i64()
.unwrap_or(0);
let untill = cv
.get_data(DataTypes::untill)
.unwrap_or(&JsonValue::Null)
.as_i64()
.unwrap_or(0);
let call = call_manager::get_call(call_id).await; let call = call_manager::get_call(call_id).await;
if let Some(call) = call { if let Some(call) = call {
@ -554,8 +335,8 @@ impl ClientConnection {
.unwrap() .unwrap()
.has_admin() .has_admin()
{ {
let _ = call_util::remove_participant(call_id, user_id).await; let _ = call_util::remove_participant(call_id, user_id as u64).await;
call.get_caller(user_id) call.get_caller(user_id as u64)
.await .await
.unwrap() .unwrap()
.set_timeout(untill) .set_timeout(untill)
@ -564,18 +345,9 @@ impl ClientConnection {
} }
} }
async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) {
let call_id = Uuid::from_str( let call_id =
cv.get_data(DataTypes::call_id) Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap();
.unwrap_or(&JsonValue::Null) let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
.as_str()
.unwrap_or(""),
)
.unwrap();
let user_id = cv
.get_data(DataTypes::user_id)
.unwrap_or(&JsonValue::Null)
.as_i64()
.unwrap_or(0);
let call = call_manager::get_call(call_id).await; let call = call_manager::get_call(call_id).await;
if let Some(call) = call { if let Some(call) = call {
@ -585,23 +357,14 @@ impl ClientConnection {
.unwrap() .unwrap()
.has_admin() .has_admin()
{ {
call.remove_caller(user_id).await; call.remove_caller(user_id as u64).await;
} }
} }
} }
async fn handle_call_set_anonymous_joining(self: Arc<Self>, cv: CommunicationValue) { async fn handle_call_set_anonymous_joining(self: Arc<Self>, cv: CommunicationValue) {
let call_id = Uuid::from_str( let call_id =
cv.get_data(DataTypes::call_id) Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap();
.unwrap_or(&JsonValue::Null) let enable = cv.get_data(DataTypes::enabled).as_bool().unwrap_or(true);
.as_str()
.unwrap_or(""),
)
.unwrap();
let enable = cv
.get_data(DataTypes::enabled)
.unwrap_or(&JsonValue::Null)
.as_bool()
.unwrap_or(true);
let call = call_manager::get_call(call_id).await; let call = call_manager::get_call(call_id).await;
@ -620,10 +383,10 @@ impl ClientConnection {
let mut response_cv = let mut response_cv =
CommunicationValue::new(CommunicationType::call_set_anonymous_joining) CommunicationValue::new(CommunicationType::call_set_anonymous_joining)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_data(DataTypes::call_id, JsonValue::String(call_id.to_string())) .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
.add_data(DataTypes::enabled, JsonValue::Boolean(enable)); .add_data(DataTypes::enabled, DataValue::Bool(enable));
if let Some(short_link) = short_link { if let Some(short_link) = short_link {
response_cv = response_cv.add_data(DataTypes::link, JsonValue::String(short_link)); response_cv = response_cv.add_data(DataTypes::link, DataValue::Str(short_link));
} }
self.send_message(&response_cv).await; self.send_message(&response_cv).await;
} }
@ -631,11 +394,13 @@ impl ClientConnection {
/// Forward message to Iota /// Forward message to Iota
async fn forward_to_iota(self: Arc<Self>, cv: CommunicationValue) { async fn forward_to_iota(self: Arc<Self>, cv: CommunicationValue) {
if cv.is_type(CommunicationType::add_conversation) if cv.is_type(CommunicationType::add_conversation)
&& cv.get_data(DataTypes::chat_partner_id).is_none() && cv
.get_data(DataTypes::chat_partner_id)
.as_number()
.is_some()
{ {
let chat_partner_name = cv let chat_partner_name = cv
.get_data(DataTypes::chat_partner_name) .get_data(DataTypes::chat_partner_name)
.unwrap_or(&JsonValue::Null)
.as_str() .as_str()
.unwrap_or("") .unwrap_or("")
.to_string(); .to_string();
@ -644,7 +409,7 @@ impl ClientConnection {
.await .await
.is_some() .is_some()
{ {
self.send_error_response(&cv.get_id(), CommunicationType::error_anonymous) self.send_error_response(cv.get_id(), CommunicationType::error_anonymous)
.await; .await;
return; return;
} }
@ -655,25 +420,22 @@ impl ClientConnection {
.with_id(cv.clone().get_id()) .with_id(cv.clone().get_id())
.add_data( .add_data(
DataTypes::username, DataTypes::username,
JsonValue::from(chat_partner_name.clone()), DataValue::Str(chat_partner_name.clone()),
), ),
Some(Duration::from_secs(20)), Some(Duration::from_secs(20)),
) )
.await; .await;
let chat_partner_id = { let chat_partner_id = {
if let Ok(load_uuid_response) = load_uuid_response { if let Ok(load_uuid_response) = load_uuid_response {
load_uuid_response load_uuid_response.get_data(DataTypes::user_id).clone()
.get_data(DataTypes::user_id)
.unwrap_or(&JsonValue::Null)
.clone()
} else { } else {
JsonValue::Null DataValue::Null
} }
}; };
if let Some(rho_conn) = self.get_rho_connection().await { if let Some(rho_conn) = self.get_rho_connection().await {
let updated_cv = cv let updated_cv = cv
.with_sender(self.get_user_id().await) .with_sender(self.get_user_id().await as u64)
.add_data(DataTypes::chat_partner_id, chat_partner_id); .add_data(DataTypes::chat_partner_id, chat_partner_id);
rho_conn.message_to_iota(updated_cv).await; rho_conn.message_to_iota(updated_cv).await;
} }
@ -681,18 +443,14 @@ impl ClientConnection {
} }
if let Some(rho_conn) = self.get_rho_connection().await { if let Some(rho_conn) = self.get_rho_connection().await {
let updated_cv = cv.with_sender(self.get_user_id().await); let updated_cv = cv.with_sender(self.get_user_id().await as u64);
rho_conn.message_to_iota(updated_cv).await; rho_conn.message_to_iota(updated_cv).await;
} }
} }
/// Send error response /// Send error response
async fn send_error_response( async fn send_error_response(self: Arc<Self>, message_id: u32, error_type: CommunicationType) {
self: Arc<Self>, let error = CommunicationValue::new(error_type).with_id(message_id);
message_id: &Uuid,
error_type: CommunicationType,
) {
let error = CommunicationValue::new(error_type).with_id(*message_id);
self.send_message(&error).await; self.send_message(&error).await;
} }
@ -704,8 +462,7 @@ impl ClientConnection {
} }
*is_open_guard = false; *is_open_guard = false;
let mut session = self.sender.write().await; let _ = self.sender.close();
let _ = session.close(None).await;
} }
/// Set interested users list /// Set interested users list
@ -723,8 +480,8 @@ impl ClientConnection {
let interested_guard = self.clone().get_interested_users().await; let interested_guard = self.clone().get_interested_users().await;
if interested_guard.contains(&user_id) { if interested_guard.contains(&user_id) {
let notification = CommunicationValue::new(CommunicationType::client_changed) let notification = CommunicationValue::new(CommunicationType::client_changed)
.add_data_str(DataTypes::user_id, user_id.to_string()) .add_data(DataTypes::user_id, DataValue::Str(user_id.to_string()))
.add_data_str(DataTypes::user_state, format!("online")); .add_data(DataTypes::user_state, DataValue::Str("online".to_string()));
self.send_message(&notification).await; self.send_message(&notification).await;
} }
@ -732,13 +489,11 @@ impl ClientConnection {
/// Handle connection close /// Handle connection close
pub async fn handle_close(&self) { pub async fn handle_close(&self) {
if self.is_identified().await { let user_id = self.get_user_id().await;
let user_id = self.get_user_id().await; if let Some(rho_conn) = rho_manager::get_rho_con_for_user(user_id as i64).await {
if let Some(rho_conn) = rho_manager::get_rho_con_for_user(user_id).await { rho_conn
rho_conn .close_client_connection(Arc::new(self.clone()))
.close_client_connection(Arc::new(self.clone())) .await;
.await;
}
} }
} }
} }
@ -749,10 +504,7 @@ impl Clone for ClientConnection {
Self { Self {
sender: Arc::clone(&self.sender), sender: Arc::clone(&self.sender),
receiver: Arc::clone(&self.receiver), receiver: Arc::clone(&self.receiver),
user_id: Arc::clone(&self.user_id), user_id: self.user_id,
identified: Arc::clone(&self.identified),
challenged: Arc::clone(&self.challenged),
challenge: Arc::clone(&self.challenge),
ping: Arc::clone(&self.ping), ping: Arc::clone(&self.ping),
pub_key: Arc::clone(&self.pub_key), pub_key: Arc::clone(&self.pub_key),
rho_connection: Arc::clone(&self.rho_connection), rho_connection: Arc::clone(&self.rho_connection),

174
src/rho/connection.rs Normal file
View file

@ -0,0 +1,174 @@
use epsilon_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use epsilon_native::{Receiver, Sender};
use rand::{Rng, distributions::Alphanumeric};
use std::{sync::Arc, time::Duration};
use tokio::sync::RwLock;
use crate::{
anonymous_clients::anonymous_client_connection::AnonymousClientConnection,
get_private_key, get_public_key,
omega::omega_connection::get_omega_connection,
rho::{client_connection::ClientConnection, iota_connection::IotaConnection},
util::{
crypto_helper::{load_public_key, public_key_to_base64},
crypto_util::{DataFormat, SecurePayload},
},
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionKind {
Client,
Iota,
AnonymousClient,
Phi,
}
pub struct GeneralConnection {
pub sender: Arc<Sender>,
pub receiver: Arc<Receiver>,
identified: Arc<RwLock<bool>>,
challenged: Arc<RwLock<bool>>,
challenge: Arc<RwLock<String>>,
connection_kind: Arc<RwLock<Option<ConnectionKind>>>,
id: Arc<RwLock<u64>>,
pub_key: Arc<RwLock<Option<Vec<u8>>>>,
}
impl GeneralConnection {
pub fn new(sender: Sender, receiver: Receiver) -> Arc<Self> {
Arc::new(Self {
sender: Arc::new(sender),
receiver: Arc::new(receiver),
identified: Arc::new(RwLock::new(false)),
challenged: Arc::new(RwLock::new(false)),
challenge: Arc::new(RwLock::new(String::new())),
connection_kind: Arc::new(RwLock::new(None)),
id: Arc::new(RwLock::new(0)),
pub_key: Arc::new(RwLock::new(None)),
})
}
}
impl GeneralConnection {
pub async fn handle(self: Arc<Self>) {
loop {
let cv = match self.receiver.receive().await {
Ok(v) => v,
Err(_) => break,
};
if !*self.identified.read().await {
self.handle_identification(cv).await;
continue;
}
if !*self.challenged.read().await {
self.handle_challenge_response(cv).await;
continue;
}
if self.migrate().await {
break;
}
}
}
async fn handle_identification(self: &Arc<Self>, cv: CommunicationValue) {
if !cv.is_type(CommunicationType::identification) {
return;
}
if let DataValue::Number(iota_id) = cv.get_data(DataTypes::iota_id) {
*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 response_cv = get_omega_connection()
.await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
.await;
let response_cv = match response_cv {
Ok(r) => r,
Err(_) => return,
};
let base64_pub = response_cv
.get_data(DataTypes::public_key)
.as_str()
.unwrap_or("");
let pub_key = match load_public_key(base64_pub) {
Some(pk) => pk,
None => return,
};
*self.pub_key.write().await = Some(pub_key.as_bytes().to_vec());
let challenge: String = rand::thread_rng()
.sample_iter(&Alphanumeric)
.take(32)
.map(char::from)
.collect();
*self.challenge.write().await = challenge.clone();
*self.identified.write().await = true;
let encrypted_challenge =
SecurePayload::new(&challenge, DataFormat::Base64, get_private_key())
.unwrap()
.encrypt_x448(pub_key)
.unwrap()
.export(DataFormat::Base64);
let response = CommunicationValue::new(CommunicationType::challenge)
.add_data(
DataTypes::public_key,
DataValue::Str(public_key_to_base64(&get_public_key())),
)
.add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
let _ = self.sender.send(&response).await;
}
}
async fn handle_challenge_response(self: &Arc<Self>, cv: CommunicationValue) {
if !cv.is_type(CommunicationType::challenge_response) {
return;
}
if let DataValue::Str(response) = cv.get_data(DataTypes::challenge) {
if *response == *self.challenge.read().await {
*self.challenged.write().await = true;
}
}
}
async fn migrate(self: &Arc<Self>) -> bool {
let kind = match *self.connection_kind.read().await {
Some(kind) => kind,
None => return false,
};
let id = *self.id.read().await;
match kind {
ConnectionKind::Client => {
let client = ClientConnection::from_general(self.clone(), id).await;
client.start();
}
ConnectionKind::Iota => {
let iota = IotaConnection::from_general(self.clone(), id).await;
iota.start();
}
ConnectionKind::AnonymousClient => {
let client = AnonymousClientConnection::from_general(self.clone(), id).await;
client.start();
}
ConnectionKind::Phi => {
let iota = ClientConnection::from_general(self.clone(), id).await;
iota.start();
}
}
true
}
}

514
src/rho/iota_connection.rs
View file

@ -1,24 +1,18 @@
use crate::calls::call_group::CallGroup; use crate::calls::call_group::CallGroup;
use crate::calls::call_manager; use crate::calls::call_manager;
use crate::get_private_key;
use crate::get_public_key;
use crate::log_cv_in; use crate::log_cv_in;
use crate::log_cv_out;
use crate::log_err; use crate::log_err;
use crate::log_in;
use crate::log_out;
use crate::omega::omega_connection::get_omega_connection; use crate::omega::omega_connection::get_omega_connection;
use crate::util::crypto_helper::load_public_key; use crate::rho::connection::GeneralConnection;
use crate::util::crypto_helper::public_key_to_base64;
use crate::util::crypto_util::DataFormat;
use crate::util::crypto_util::SecurePayload;
use crate::util::logger::PrintType; use crate::util::logger::PrintType;
use async_tungstenite::WebSocketReceiver;
use async_tungstenite::WebSocketSender;
use async_tungstenite::tungstenite::Message;
use dashmap::DashMap; use dashmap::DashMap;
use json::JsonValue; use epsilon_core::CommunicationType;
use rand::Rng; use epsilon_core::CommunicationValue;
use rand::distributions::Alphanumeric; use epsilon_core::DataTypes;
use epsilon_core::DataValue;
use epsilon_native::Receiver;
use epsilon_native::Sender;
use std::{ use std::{
collections::HashMap, collections::HashMap,
sync::{Arc, Weak}, sync::{Arc, Weak},
@ -26,56 +20,48 @@ use std::{
}; };
use tokio::sync::RwLock; use tokio::sync::RwLock;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio_util::compat::Compat;
use tungstenite::Utf8Bytes;
use uuid::Uuid;
use x448::PublicKey; use x448::PublicKey;
use super::{rho_connection::RhoConnection, rho_manager}; use super::{rho_connection::RhoConnection, rho_manager};
use crate::{ use crate::omega::omega_connection::OmegaConnection;
data::communication::{CommunicationType, CommunicationValue, DataTypes},
omega::omega_connection::OmegaConnection,
};
pub struct IotaConnection { pub struct IotaConnection {
pub sender: Arc<RwLock<WebSocketSender<Compat<tokio::net::TcpStream>>>>, pub iota_id: u64,
pub receiver: Arc<RwLock<WebSocketReceiver<Compat<tokio::net::TcpStream>>>>, pub sender: Arc<Sender>,
pub iota_id: Arc<RwLock<i64>>, pub receiver: Arc<Receiver>,
pub user_ids: Arc<RwLock<Vec<i64>>>, pub user_ids: Arc<RwLock<Vec<u64>>>,
identified: Arc<RwLock<bool>>,
challenged: Arc<RwLock<bool>>,
challenge: Arc<RwLock<String>>,
pub ping: Arc<RwLock<i64>>, pub ping: Arc<RwLock<i64>>,
pub_key: Arc<RwLock<Option<Vec<u8>>>>, pub_key: Arc<RwLock<Option<Vec<u8>>>>,
pub waiting_tasks: pub waiting_tasks:
DashMap<Uuid, Box<dyn Fn(Arc<IotaConnection>, CommunicationValue) -> bool + Send + Sync>>, DashMap<u32, Box<dyn Fn(Arc<IotaConnection>, CommunicationValue) -> bool + Send + Sync>>,
pub rho_connection: Arc<RwLock<Option<Weak<RhoConnection>>>>, pub rho_connection: Arc<RwLock<Option<Weak<RhoConnection>>>>,
} }
impl IotaConnection { impl IotaConnection {
/// Create a new IotaConnection pub async fn from_general(general: Arc<GeneralConnection>, iota_id: u64) -> Arc<Self> {
pub fn new(
sender: WebSocketSender<Compat<tokio::net::TcpStream>>,
receiver: WebSocketReceiver<Compat<tokio::net::TcpStream>>,
) -> Arc<Self> {
Arc::new(Self { Arc::new(Self {
sender: Arc::new(RwLock::new(sender)),
receiver: Arc::new(RwLock::new(receiver)),
iota_id: Arc::new(RwLock::new(0)),
user_ids: Arc::new(RwLock::new(Vec::new())),
identified: Arc::new(RwLock::new(false)),
challenged: Arc::new(RwLock::new(false)),
challenge: Arc::new(RwLock::new(String::new())),
ping: Arc::new(RwLock::new(0)), ping: Arc::new(RwLock::new(0)),
pub_key: Arc::new(RwLock::new(None)), pub_key: Arc::new(RwLock::new(None)),
waiting_tasks: DashMap::new(),
rho_connection: Arc::new(RwLock::new(None)), rho_connection: Arc::new(RwLock::new(None)),
user_ids: Arc::new(RwLock::new(Vec::new())),
sender: general.sender.clone(),
receiver: general.receiver.clone(),
iota_id: iota_id,
waiting_tasks: DashMap::new(),
}) })
} }
pub fn start(self: Arc<Self>) {
let self_clone = self.clone();
tokio::spawn(async move {
while let Ok(cv) = self_clone.receiver.receive().await {
self_clone.clone().handle_message(cv).await;
}
});
}
/// Get the Iota ID /// Get the Iota ID
pub async fn get_iota_id(&self) -> i64 { pub async fn get_iota_id(&self) -> u64 {
*self.iota_id.read().await self.iota_id
} }
pub async fn get_public_key(&self) -> Option<PublicKey> { pub async fn get_public_key(&self) -> Option<PublicKey> {
@ -87,15 +73,10 @@ impl IotaConnection {
} }
/// Get the user IDs /// Get the user IDs
pub async fn get_user_ids(&self) -> Vec<i64> { pub async fn get_user_ids(&self) -> Vec<u64> {
self.user_ids.read().await.clone() self.user_ids.read().await.clone()
} }
/// Check if connection is identified
pub async fn is_identified(&self) -> bool {
*self.identified.read().await
}
/// Get current ping /// Get current ping
pub async fn get_ping(&self) -> i64 { pub async fn get_ping(&self) -> i64 {
*self.ping.read().await *self.ping.read().await
@ -117,38 +98,16 @@ impl IotaConnection {
} }
} }
/// Send a message to the Iota
pub async fn send_message_str(&self, message: &str) {
let mut session = self.sender.write().await;
if let Err(e) = session
.send(Message::Text(Utf8Bytes::from(message.to_string())))
.await
{
log_err!(
self.get_iota_id().await,
PrintType::Iota,
"Failed to send WebSocket message: {:?}",
e,
);
}
}
/// Send a CommunicationValue to the Iota /// Send a CommunicationValue to the Iota
pub async fn send_message(&self, cv: &CommunicationValue) { pub async fn send_message(&self, cv: &CommunicationValue) {
if !cv.is_type(CommunicationType::pong) { if !cv.is_type(CommunicationType::pong) {
log_out!( log_cv_out!(PrintType::Iota, cv);
self.get_iota_id().await,
PrintType::Iota,
"{}",
cv.to_json().to_string()
);
} }
self.send_message_str(&cv.to_json().to_string()).await; self.sender.send(&cv).await;
} }
/// Handle incoming message from Iota /// Handle incoming message from Iota
pub async fn handle_message(self: Arc<Self>, message: Utf8Bytes) { pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
let cv = CommunicationValue::from_json(&message);
// Handle ping // 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; self.handle_ping(cv).await;
@ -157,261 +116,6 @@ impl IotaConnection {
log_cv_in!(PrintType::Iota, cv); log_cv_in!(PrintType::Iota, cv);
let identified = *self.identified.read().await;
let challenged = *self.challenged.read().await;
if !identified && cv.is_type(CommunicationType::identification) {
let iota_id = cv
.get_data(DataTypes::iota_id)
.and_then(|v| v.as_i64())
.unwrap_or(0);
if iota_id == 0 {
log_out!(self.get_iota_id().await, PrintType::Iota, "Invalid IOTA ID");
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_data)
.await;
self.close().await;
return;
}
*self.iota_id.write().await = iota_id;
let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_iota_data)
.with_id(cv.get_id())
.add_data(DataTypes::iota_id, JsonValue::from(iota_id));
let response_cv = get_omega_connection()
.await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
.await;
if let Ok(response_cv) = response_cv {
if !response_cv.is_type(CommunicationType::get_iota_data) {
self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
.await;
self.close().await;
return;
}
let base64_pub = response_cv
.get_data(DataTypes::public_key)
.and_then(|v| v.as_str())
.unwrap_or("");
let pub_key = match load_public_key(base64_pub) {
Some(pk) => pk,
_ => {
self.send_error_response(
&cv.get_id(),
CommunicationType::error_invalid_public_key,
)
.await;
self.close().await;
return;
}
};
*self.pub_key.write().await = Some(pub_key.as_bytes().to_vec());
let challenge: String = rand::thread_rng()
.sample_iter(&Alphanumeric)
.take(32)
.map(char::from)
.collect();
*self.challenge.write().await = challenge.clone();
let encrypted_challenge =
SecurePayload::new(&challenge, DataFormat::Base64, get_private_key())
.unwrap()
.encrypt_x448(pub_key)
.unwrap()
.export(DataFormat::Base64);
*self.identified.write().await = true;
let challenge_msg = CommunicationValue::new(CommunicationType::challenge)
.with_id(cv.get_id())
.add_data_str(
DataTypes::public_key,
public_key_to_base64(&get_public_key()),
)
.add_data_str(DataTypes::challenge, encrypted_challenge);
self.send_message(&challenge_msg).await;
} else {
self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
.await;
self.close().await;
return;
}
return;
} else if !identified && cv.is_type(CommunicationType::register_iota) {
let base64_pub = cv
.get_data(DataTypes::public_key)
.and_then(|v| v.as_str())
.unwrap_or("");
if base64_pub.is_empty() {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_public_key)
.await;
self.close().await;
return;
}
let register_msg = CommunicationValue::new(CommunicationType::complete_register_iota)
.with_id(cv.get_id())
.add_data(
DataTypes::public_key,
JsonValue::String(base64_pub.to_string()),
);
let iota_conn_clone = self.clone();
let register_response: Result<CommunicationValue, _> = get_omega_connection()
.await_response(&register_msg, Some(Duration::from_secs(20)))
.await;
let iota_conn_for_task = iota_conn_clone.clone();
if let Ok(register_response) = register_response {
if !register_response.is_type(CommunicationType::complete_register_iota) {
iota_conn_for_task
.send_error_response(&cv.get_id(), CommunicationType::error_internal)
.await;
iota_conn_for_task.close().await;
return;
}
let new_iota_id = register_response
.get_data(DataTypes::iota_id)
.and_then(|v| v.as_i64())
.unwrap_or(0);
if new_iota_id == 0 {
iota_conn_for_task
.send_error_response(&cv.get_id(), CommunicationType::error_internal)
.await;
iota_conn_for_task.close().await;
return;
}
*iota_conn_for_task.iota_id.write().await = new_iota_id;
*iota_conn_for_task.identified.write().await = true;
let success_msg = CommunicationValue::new(CommunicationType::success)
.with_id(cv.get_id())
.add_data(DataTypes::iota_id, JsonValue::from(new_iota_id));
iota_conn_for_task.send_message(&success_msg).await;
} else {
iota_conn_for_task
.send_error_response(&cv.get_id(), CommunicationType::error_internal)
.await;
iota_conn_for_task.close().await;
}
return;
}
if identified && !challenged && cv.is_type(CommunicationType::challenge_response) {
let client_response = cv
.get_data(DataTypes::challenge)
.and_then(|v| v.as_str())
.unwrap_or("");
if client_response == *self.challenge.read().await {
*self.challenged.write().await = true;
let iota_id = self.get_iota_id().await;
if rho_manager::contains_iota(iota_id).await {
if let Some(existing_rho) = rho_manager::get_rho_by_iota(iota_id).await {
existing_rho.close_iota_connection().await;
}
}
// Inform Omega & Verify Users
let iota_users_cv = get_omega_connection()
.await_response(
&CommunicationValue::new(CommunicationType::iota_connected).add_data(
DataTypes::iota_id,
JsonValue::from(self.get_iota_id().await),
),
Some(Duration::from_secs(20)),
)
.await;
let mut user_ids: Vec<i64> = Vec::new();
if let Ok(iota_users_cv) = iota_users_cv {
if !iota_users_cv.is_type(CommunicationType::iota_user_data) {
log_err!(
self.get_iota_id().await,
PrintType::Omikron,
"Invalid communication type {:?}",
iota_users_cv.get_type()
);
return;
}
let val_user_ids = iota_users_cv.get_data(DataTypes::user_ids).unwrap().clone();
match val_user_ids {
JsonValue::Array(arr) => {
for item in arr {
if let JsonValue::Number(_) = item {
user_ids.push(item.as_i64().unwrap_or(0));
}
}
}
_ => {}
}
} else {
log_err!(
self.get_iota_id().await,
PrintType::Omikron,
"Failed to retrieve user IDs"
);
}
log_in!(
self.get_iota_id().await,
PrintType::General,
"User IDs: {:?}",
user_ids.clone()
);
*self.user_ids.write().await = user_ids.clone();
let rho_connection =
Arc::new(RhoConnection::new(self.clone(), user_ids.clone()).await);
self.set_rho_connection(Arc::downgrade(&rho_connection))
.await;
rho_manager::add_rho(rho_connection).await;
let mut str = String::new();
for id in &user_ids {
str.push_str(&format!(",{}", id));
}
if !str.is_empty() {
str.remove(0);
}
self.send_message(
&CommunicationValue::new(CommunicationType::identification_response)
.with_id(cv.get_id())
.add_data_str(DataTypes::accepted_ids, str)
.add_data_str(DataTypes::accepted, user_ids.len().to_string()),
)
.await;
} else {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_challenge)
.await;
self.close().await;
}
return;
}
if !self.is_identified().await {
self.send_error_response(&cv.get_id(), CommunicationType::error_not_authenticated)
.await;
self.close().await;
return;
}
// Handle GET_CHATS // Handle GET_CHATS
if cv.is_type(CommunicationType::get_chats) { if cv.is_type(CommunicationType::get_chats) {
self.handle_get_chats(cv).await; self.handle_get_chats(cv).await;
@ -420,7 +124,7 @@ impl IotaConnection {
// Handle forwarding to other Iotas or clients // Handle forwarding to other Iotas or clients
let receiver_id = cv.get_receiver(); let receiver_id = cv.get_receiver();
if (receiver_id != 0 && !self.get_user_ids().await.contains(&receiver_id)) 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::message_other_iota)
|| cv.is_type(CommunicationType::send_chat) || cv.is_type(CommunicationType::send_chat)
{ {
@ -437,31 +141,28 @@ impl IotaConnection {
|| cv.is_type(CommunicationType::delete_iota) || cv.is_type(CommunicationType::delete_iota)
{ {
let sender = self.get_iota_id().await; let sender = self.get_iota_id().await;
self.handle_omega_forward(cv.with_sender(sender)).await; self.handle_omega_forward(cv.with_sender(sender as u64))
.await;
return; return;
} }
// Forward to client // Forward to client
self.forward_to_client(cv).await; self.forward_to_client(cv).await;
} }
async fn send_error_response(&self, message_id: &Uuid, error_type: CommunicationType) { async fn send_error_response(&self, message_id: u32, error_type: CommunicationType) {
let error = CommunicationValue::new(error_type).with_id(*message_id); let error = CommunicationValue::new(error_type).with_id(message_id);
self.send_message(&error).await; self.send_message(&error).await;
} }
async fn close(&self) { async fn close(&self) {
let mut sender = self.sender.write().await; let _ = self.sender.close();
let _ = sender.close(None).await;
} }
async fn handle_omega_forward(self: Arc<Self>, cv: CommunicationValue) { async fn handle_omega_forward(self: Arc<Self>, cv: CommunicationValue) {
let iota_for_closure = self.clone(); let iota_for_closure = self.clone();
tokio::spawn(async move { tokio::spawn(async move {
let response_cv = get_omega_connection() let response_cv = get_omega_connection()
.await_response( .await_response(&cv.with_sender(self.iota_id), Some(Duration::from_secs(20)))
&cv.with_sender(*self.iota_id.read().await),
Some(Duration::from_secs(20)),
)
.await; .await;
if let Ok(response_cv) = response_cv { if let Ok(response_cv) = response_cv {
iota_for_closure.send_message(&response_cv).await; iota_for_closure.send_message(&response_cv).await;
@ -470,7 +171,7 @@ impl IotaConnection {
} }
/// Handle ping message /// Handle ping message
async fn handle_ping(&self, cv: CommunicationValue) { async fn handle_ping(&self, cv: CommunicationValue) {
if let Some(last_ping) = cv.get_data(DataTypes::last_ping) { if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
if let Ok(ping_val) = last_ping.to_string().parse::<i64>() { if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
let mut ping_guard = self.ping.write().await; let mut ping_guard = self.ping.write().await;
*ping_guard = ping_val; *ping_guard = ping_val;
@ -483,13 +184,13 @@ impl IotaConnection {
HashMap::new() HashMap::new()
}; };
let pings = client_pings let pings: Vec<(DataTypes, DataValue)> = client_pings
.into_iter() .into_iter()
.map(|(k, v)| (k, JsonValue::String(v.to_string()))) .map(|(k, v)| (DataTypes::parse(k), DataValue::Number(v)))
.collect(); .collect();
let response = CommunicationValue::new(CommunicationType::pong) let response = CommunicationValue::new(CommunicationType::pong)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_data(DataTypes::ping_clients, JsonValue::Object(pings)); .add_data(DataTypes::ping_clients, DataValue::Container(pings));
self.send_message(&response).await; self.send_message(&response).await;
} }
@ -498,8 +199,8 @@ impl IotaConnection {
let receiver_id = cv.get_receiver(); let receiver_id = cv.get_receiver();
let sender_id = cv.get_sender(); let sender_id = cv.get_sender();
if self.get_user_ids().await.contains(&sender_id) { if self.get_user_ids().await.contains(&(sender_id as u64)) {
if let Some(target_rho) = rho_manager::get_rho_con_for_user(receiver_id).await { if let Some(target_rho) = rho_manager::get_rho_con_for_user(receiver_id as i64).await {
target_rho.message_to_iota(cv).await; target_rho.message_to_iota(cv).await;
} else { } else {
let error = CommunicationValue::new(CommunicationType::error_no_iota) let error = CommunicationValue::new(CommunicationType::error_no_iota)
@ -511,7 +212,7 @@ impl IotaConnection {
self.send_message( self.send_message(
&CommunicationValue::new(CommunicationType::error_invalid_user_id).add_data( &CommunicationValue::new(CommunicationType::error_invalid_user_id).add_data(
DataTypes::error_type, DataTypes::error_type,
JsonValue::String( DataValue::Str(
"You are sending to another User without authority.".to_string(), "You are sending to another User without authority.".to_string(),
), ),
), ),
@ -525,77 +226,98 @@ impl IotaConnection {
let receiver_id = cv.get_receiver(); let receiver_id = cv.get_receiver();
let mut interested_ids: Vec<i64> = Vec::new(); let mut interested_ids: Vec<i64> = Vec::new();
// loading Calls // ============================
// Load Calls
// ============================
let calls: Vec<Arc<CallGroup>> = call_manager::get_call_groups(receiver_id).await; let calls: Vec<Arc<CallGroup>> = call_manager::get_call_groups(receiver_id).await;
let mut invites: HashMap<i64, Vec<JsonValue>> = HashMap::new();
let empty = &calls.is_empty(); let mut invites: HashMap<i64, Vec<DataValue>> = HashMap::new();
let empty = calls.is_empty();
for call in calls { for call in calls {
for inviter in call.members.read().await.iter() { for inviter in call.members.read().await.iter() {
let call_self = call.get_caller(receiver_id).await.unwrap(); let call_self = call.get_caller(receiver_id).await.unwrap();
let admin = call_self.has_admin();
let inviter_id = inviter.user_id; let inviter_id = inviter.user_id;
let timeout = *call_self.timeout.read().await; let timeout = *call_self.timeout.read().await;
let admin = call_self.has_admin();
// Build call container
let mut call_map: HashMap<DataTypes, DataValue> = HashMap::new();
call_map.insert(DataTypes::call_id, DataValue::Str(call.call_id.to_string()));
let mut call_obj = JsonValue::new_object();
let _ = call_obj.insert("call_id", JsonValue::String(call.call_id.to_string()));
if timeout > 0 { if timeout > 0 {
let _ = call_obj.insert("timeout", JsonValue::from(timeout)); call_map.insert(DataTypes::timeout, DataValue::Number(timeout as i64));
}
if admin {
let _ = call_obj.insert("admin", JsonValue::Boolean(admin));
} }
if let Some(call_ids) = invites.get_mut(&inviter_id) { if admin {
call_ids.push(call_obj); call_map.insert(DataTypes::has_admin, DataValue::Bool(true));
} else {
invites.insert(inviter_id, vec![call_obj]);
} }
let call_container = DataValue::container_from_map(&call_map);
invites
.entry(inviter_id as i64)
.or_insert_with(Vec::new)
.push(call_container);
} }
} }
// Process contacts and add call information // ============================
let enriched_contacts = if *empty { // Enrich Contacts
if let Some(contacts_data) = cv.get_data(DataTypes::user_ids) { // ============================
log_in!(self.get_iota_id().await, PrintType::Call, "Call empty"); let enriched_contacts = if empty {
contacts_data.clone() match cv.get_data(DataTypes::user_ids) {
} else { DataValue::Array(arr) => DataValue::Array(arr.clone()),
log_in!( _ => DataValue::Array(vec![]),
self.get_iota_id().await,
PrintType::Call,
"Call empty No Data"
);
JsonValue::new_array()
} }
} else { } else {
let mut enrc_contacts = JsonValue::new_array(); let mut enriched: Vec<DataValue> = Vec::new();
if let Some(contacts_data) = cv.get_data(DataTypes::user_ids) {
if let JsonValue::Array(user_ids) = contacts_data { if let DataValue::Array(users) = cv.get_data(DataTypes::user_ids) {
for user_json in user_ids { for user_val in users {
let user_id = user_json["user_id"].as_i64().unwrap_or(0); if let DataValue::Container(entries) = user_val {
interested_ids.push(user_id); let mut user_map: HashMap<DataTypes, DataValue> =
let mut enriched_contact = user_json.clone(); entries.iter().cloned().collect();
if let Some(calls) = invites.get(&user_id) {
let _ = // extract user_id
enriched_contact.insert("calls", JsonValue::Array(calls.clone())); if let Some(DataValue::Number(user_id)) = user_map.get(&DataTypes::user_id)
{
interested_ids.push(*user_id);
// attach calls if exists
if let Some(call_list) = invites.get(user_id) {
user_map
.insert(DataTypes::calls, DataValue::Array(call_list.clone()));
}
} }
let _ = enrc_contacts.push(enriched_contact);
enriched.push(DataValue::container_from_map(&user_map));
} }
} else {
enrc_contacts = contacts_data.clone();
} }
} }
enrc_contacts
DataValue::Array(enriched)
}; };
// Notify OmegaConnection about user states // ============================
OmegaConnection::user_states(receiver_id, interested_ids.clone()).await; // Notify Omega
// ============================
OmegaConnection::user_states(receiver_id as i64, interested_ids.clone()).await;
// Set interested users in RhoConnection // ============================
// Notify Rho
// ============================
if let Some(rho_conn) = self.get_rho_connection().await { if let Some(rho_conn) = self.get_rho_connection().await {
rho_conn.set_interested(receiver_id, interested_ids).await; rho_conn
.set_interested(receiver_id as i64, interested_ids)
.await;
} }
// ============================
// Forward to client // Forward to client
// ============================
self.forward_to_client(cv.add_data(DataTypes::user_ids, enriched_contacts)) self.forward_to_client(cv.add_data(DataTypes::user_ids, enriched_contacts))
.await; .await;
} }
@ -607,7 +329,7 @@ impl IotaConnection {
rho_conn.message_to_client(updated_cv).await; rho_conn.message_to_client(updated_cv).await;
} else { } else {
log_err!( log_err!(
self.get_iota_id().await, self.get_iota_id().await as i64,
PrintType::General, PrintType::General,
"Failed to forward message to client" "Failed to forward message to client"
); );
@ -615,10 +337,8 @@ impl IotaConnection {
} }
pub async fn handle_close(&self) { pub async fn handle_close(&self) {
if self.is_identified().await { if let Some(rho_conn) = self.get_rho_connection().await {
if let Some(rho_conn) = self.get_rho_connection().await { rho_conn.close_iota_connection().await;
rho_conn.close_iota_connection().await;
}
} }
} }
@ -638,7 +358,7 @@ impl IotaConnection {
tokio::spawn(async move { tokio::spawn(async move {
if let Err(e) = inner_tx.send(response_cv).await { if let Err(e) = inner_tx.send(response_cv).await {
log_err!( log_err!(
io.get_iota_id().await, io.get_iota_id().await as i64,
PrintType::Iota, PrintType::Iota,
"Failed to send response back to awaiter: {}", "Failed to send response back to awaiter: {}",
e e

2
src/rho/mod.rs
View file

@ -1,4 +1,6 @@
pub mod client_connection; pub mod client_connection;
pub mod connection;
pub mod iota_connection; pub mod iota_connection;
pub mod rho_connection; pub mod rho_connection;
pub mod rho_manager; pub mod rho_manager;
pub mod server;

30
src/rho/rho_connection.rs
View file

@ -1,10 +1,8 @@
use super::{client_connection::ClientConnection, iota_connection::IotaConnection, rho_manager}; use super::{client_connection::ClientConnection, iota_connection::IotaConnection, rho_manager};
use crate::data::{
communication::{CommunicationType, CommunicationValue, DataTypes}, use crate::data::user::UserStatus;
user::UserStatus,
};
use crate::omega::omega_connection::OmegaConnection; use crate::omega::omega_connection::OmegaConnection;
use json::{JsonValue, number::Number}; use epsilon_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::Arc; use std::sync::Arc;
use tokio::sync::RwLock; use tokio::sync::RwLock;
@ -27,8 +25,8 @@ impl RhoConnection {
rho_connection rho_connection
} }
pub async fn get_iota_id(&self) -> i64 { pub async fn get_iota_id(&self) -> u64 {
self.iota_connection.get_iota_id().await self.iota_connection.iota_id
} }
pub fn get_user_ids(&self) -> &Vec<i64> { pub fn get_user_ids(&self) -> &Vec<i64> {
@ -52,7 +50,7 @@ impl RhoConnection {
let connections = self.client_connections.read().await; let connections = self.client_connections.read().await;
let mut collections = Vec::new(); let mut collections = Vec::new();
for con in connections.iter() { for con in connections.iter() {
if con.get_user_id().await == user_id { if con.get_user_id().await == user_id as u64 {
collections.push(con.clone()); collections.push(con.clone());
} }
} }
@ -63,7 +61,7 @@ impl RhoConnection {
pub async fn add_client_connection(&self, connection: Arc<ClientConnection>) { pub async fn add_client_connection(&self, connection: Arc<ClientConnection>) {
let notification = CommunicationValue::new(CommunicationType::client_connected).add_data( let notification = CommunicationValue::new(CommunicationType::client_connected).add_data(
DataTypes::user_id, DataTypes::user_id,
JsonValue::Number(Number::from(connection.get_user_id().await)), DataValue::Number(connection.get_user_id().await as i64),
); );
self.iota_connection.send_message(&notification).await; self.iota_connection.send_message(&notification).await;
@ -74,8 +72,8 @@ impl RhoConnection {
} }
OmegaConnection::client_changed( OmegaConnection::client_changed(
self.get_iota_id().await, self.get_iota_id().await as i64,
connection.get_user_id().await, connection.get_user_id().await as i64,
UserStatus::user_online, UserStatus::user_online,
) )
.await; .await;
@ -93,8 +91,8 @@ impl RhoConnection {
// Notify OmegaConnection // Notify OmegaConnection
OmegaConnection::client_changed( OmegaConnection::client_changed(
self.get_iota_id().await, self.get_iota_id().await as i64,
connection.get_user_id().await, connection.get_user_id().await as i64,
UserStatus::user_offline, UserStatus::user_offline,
) )
.await; .await;
@ -109,10 +107,10 @@ impl RhoConnection {
} }
// Remove from manager // Remove from manager
rho_manager::remove_rho(self.get_iota_id().await).await; rho_manager::remove_rho(self.get_iota_id().await as i64).await;
// Notify OmegaConnection // Notify OmegaConnection
OmegaConnection::close_iota(self.get_iota_id().await).await; OmegaConnection::close_iota(self.get_iota_id().await as i64).await;
} }
/// Send message from Iota to specific client /// Send message from Iota to specific client
@ -136,7 +134,7 @@ impl RhoConnection {
let connections = self.client_connections.read().await; let connections = self.client_connections.read().await;
for connection in connections.iter() { for connection in connections.iter() {
let conn_user_id = connection.get_user_id().await; let conn_user_id = connection.get_user_id().await;
if conn_user_id == user_id { if conn_user_id == user_id as u64 {
connection connection
.clone() .clone()
.set_interested_users(interested_ids.clone()) .set_interested_users(interested_ids.clone())

2
src/rho/rho_manager.rs
View file

@ -41,7 +41,7 @@ pub async fn remove_rho(iota_id: i64) -> Option<Arc<RhoConnection>> {
pub async fn add_rho(rho_connection: Arc<RhoConnection>) { pub async fn add_rho(rho_connection: Arc<RhoConnection>) {
let mut connections = RHO_CONNECTIONS.write().await; let mut connections = RHO_CONNECTIONS.write().await;
let iota_id = rho_connection.get_iota_id().await; let iota_id = rho_connection.get_iota_id().await;
connections.insert(iota_id, rho_connection); connections.insert(iota_id as i64, rho_connection);
} }
/// Get a RhoConnection by Iota ID directly /// Get a RhoConnection by Iota ID directly

57
src/rho/server.rs Normal file
View file

@ -0,0 +1,57 @@
use crate::{rho::connection::GeneralConnection, util::file_util::load_file_buf};
use epsilon_native::Host;
use quinn::ServerConfig;
use rustls::pki_types::PrivateKeyDer;
use std::sync::Arc;
use tokio::io::unix::AsyncFd;
pub async fn start(port: u16) {
let tls_cfg = load_tls().expect("TLS config failed");
let server_crypto = quinn::crypto::rustls::QuicServerConfig::try_from(tls_cfg)
.expect("Failed to convert to QuicServerConfig");
let server_cfg = ServerConfig::with_crypto(Arc::new(server_crypto));
let mut host: Host = epsilon_native::host(port, server_cfg).await.unwrap();
tokio::spawn(async move {
while let Some((sender, receiver)) = host.next().await {
tokio::spawn(async move {
GeneralConnection::new(sender, receiver).handle().await;
});
}
});
}
fn load_tls() -> Option<rustls::ServerConfig> {
let mut cert_file_buf = load_file_buf("certs", "cert.pem").ok()?;
let mut key_file_buf = load_file_buf("certs", "cert.key").ok()?;
let cert_chain = rustls_pemfile::certs(&mut cert_file_buf)
.collect::<Result<Vec<_>, _>>()
.ok()?;
let mut keys: Vec<PrivateKeyDer> = rustls_pemfile::pkcs8_private_keys(&mut key_file_buf)
.map(|k| k.map(Into::into))
.collect::<Result<Vec<_>, _>>()
.ok()?;
if keys.is_empty() {
let mut key_file_buf = load_file_buf("certs", "cert.key").ok()?;
keys = rustls_pemfile::rsa_private_keys(&mut key_file_buf)
.map(|k| k.map(Into::into))
.collect::<Result<Vec<_>, _>>()
.ok()?;
}
if keys.is_empty() {
return None;
}
let cfg = rustls::ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(cert_chain, keys.remove(0))
.ok()?;
Some(cfg)
}

179
src/util/file_util.rs Normal file
View file

@ -0,0 +1,179 @@
use std::fs::{self, File};
use std::io::{self, BufReader, Read};
use std::path::{Path, PathBuf};
use crate::log;
use crate::util::logger::PrintType;
#[allow(dead_code)]
pub fn delete_directory(path: &str) -> bool {
let dir = Path::new(&get_directory()).join(path);
delete_dir_recursive(&dir)
}
#[allow(dead_code)]
fn delete_dir_recursive(directory: &Path) -> bool {
if !directory.exists() {
return false;
}
if let Err(e) = fs::remove_dir_all(directory) {
log!(
0,
PrintType::General,
"[IMPORTANT] Couldn't delete directory {}: {}",
directory.display(),
e,
);
return false;
}
true
}
#[allow(dead_code)]
pub fn delete_user_directory(user_id: i64) {
let user_dir = Path::new(&get_directory())
.join("users")
.join(user_id.to_string());
let _ = delete_dir_recursive(&user_dir);
}
pub fn load_file_buf(path: &str, name: &str) -> io::Result<BufReader<File>> {
let dir = Path::new(&get_directory()).join(path);
let file_path = dir.join(name);
// Ensure the directory exists, create if necessary
if !dir.exists() {
if let Err(_) = fs::create_dir_all(&dir) {
return Err(io::Error::new(
io::ErrorKind::NotFound,
"Directory creation failed",
));
}
}
// Create the file if it doesn't exist
if !file_path.exists() {
return Err(io::Error::new(
io::ErrorKind::NotFound,
"File creation failed",
));
}
// Open the file and return a BufReader for efficient reading
let file = File::open(&file_path)?;
Ok(BufReader::new(file))
}
pub fn has_file(path: &str, name: &str) -> bool {
let dir = Path::new(&get_directory()).join(path);
let file_path = dir.join(name);
if !dir.exists() {
return false;
}
if !file_path.exists() {
return false;
}
true
}
pub fn has_dir(path: &str) -> bool {
let dir = Path::new(&get_directory()).join(path);
if !dir.exists() {
return false;
}
true
}
pub fn load_file(path: &str, name: &str) -> String {
let dir = Path::new(&get_directory()).join(path);
let file_path = dir.join(name);
if !dir.exists() {
if let Err(e) = fs::create_dir_all(&dir) {
log!(
0,
PrintType::General,
"[IMPORTANT] Couldn't create directories: {}",
e
);
return String::new();
}
return String::new();
}
if !file_path.exists() {
if let Err(e) = File::create(&file_path) {
log!(
0,
PrintType::General,
"[IMPORTANT] Couldn't create file: {}",
e
);
}
return String::new();
}
let mut content = String::new();
if let Ok(mut f) = File::open(&file_path) {
let _ = f.read_to_string(&mut content);
}
content
}
pub fn load_file_vec(path: &str, name: &str) -> Result<Vec<u8>, std::io::Error> {
let dir = Path::new(&get_directory()).join(path);
let file_path = dir.join(name);
std::fs::read(file_path)
}
pub fn save_file(path: &str, name: &str, value: &str) {
let dir = Path::new(&get_directory()).join(path);
let file_path = dir.join(name);
if !dir.exists() {
if let Err(e) = fs::create_dir_all(&dir) {
log!(
0,
PrintType::General,
"[IMPORTANT] Couldn't create directories: {}",
e
);
return;
}
}
if let Err(e) = fs::write(&file_path, value) {
log!(
0,
PrintType::General,
"[IMPORTANT] Couldn't write file {}: {}",
file_path.display(),
e
);
}
}
pub fn get_children(path: &str) -> Vec<String> {
let dir = Path::new(&get_directory()).join(path);
let mut children = Vec::new();
if let Ok(entries) = fs::read_dir(&dir) {
for entry in entries {
if let Ok(entry) = entry {
children.push(entry.file_name().to_string_lossy().to_string());
}
}
}
children
}
pub fn get_directory() -> String {
let exe = std::env::current_exe().unwrap_or_else(|_| PathBuf::from("."));
exe.parent()
.unwrap_or(Path::new("."))
.to_string_lossy()
.to_string()
}

89
src/util/logger.rs
View file

@ -1,4 +1,5 @@
use std::{ use std::{
collections::HashMap,
fs::{self, OpenOptions}, fs::{self, OpenOptions},
io::Write, io::Write,
path::Path, path::Path,
@ -8,9 +9,7 @@ use std::{
}; };
use ansi_term::Color; use ansi_term::Color;
use json::JsonValue; use epsilon_core::{CommunicationValue, DataTypes, DataValue};
use crate::data::communication::CommunicationValue;
static LOGGER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new(); static LOGGER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new();
@ -151,7 +150,7 @@ pub fn log_cv_internal(
let formatted = format_cv(cv); let formatted = format_cv(cv);
log_internal( log_internal(
cv.get_sender(), cv.get_sender() as i64,
print_type.unwrap_or(PrintType::General), print_type.unwrap_or(PrintType::General),
prefix, prefix,
false, false,
@ -176,24 +175,80 @@ pub fn format_cv(cv: &CommunicationValue) -> String {
let comm_type = cv.get_type().to_string(); let comm_type = cv.get_type().to_string();
parts.push(format!("{}", comm_type)); parts.push(format!("{}", comm_type));
let mut data_parts = Vec::new(); let data: &HashMap<DataTypes, DataValue> = cv.get_data_container();
if let JsonValue::Object(data) = &cv.clone().to_json()["data"] {
for (key, value) in data.iter() {
let val_string = match value {
JsonValue::String(s) => s.clone(),
_ => value.dump(),
};
data_parts.push(format!("{} {}", key, val_string)); let formated_data =
} format_data_container(data.iter().map(|(k, v)| (k.clone(), v.clone())).collect());
}
if !data_parts.is_empty() { parts.push(format!("{}", formated_data));
parts.push(format!("{}", data_parts.join(", ")));
}
parts.join(": ") parts.join(": ")
} }
fn format_data_container(data: Vec<(DataTypes, DataValue)>) -> String {
let parts: Vec<String> = data
.into_iter()
.map(|(key, value)| {
let key_str = key.to_string();
match value {
DataValue::Str(s) => format!("{}=\"{}\"", key_str, s),
DataValue::Container(inner) => {
let inner_formatted = format_data_container(inner);
format!("{}={{ {} }}", key_str, inner_formatted)
}
DataValue::Array(arr) => {
let arr_formatted = format_array(arr);
format!("{}=[{}]", key_str, arr_formatted)
}
DataValue::Bool(b) => format!("{}={}", key_str, b),
DataValue::BoolTrue => format!("{}=true", key_str),
DataValue::BoolFalse => format!("{}=false", key_str),
DataValue::Number(num) => format!("{}={}", key_str, num),
_ => "".to_string(),
}
})
.collect();
parts.join(", ")
}
fn format_array(arr: Vec<DataValue>) -> String {
let parts: Vec<String> = arr
.into_iter()
.map(|value| match value {
DataValue::Str(s) => format!("\"{}\"", s),
DataValue::Container(inner) => {
let inner_formatted = format_data_container(inner);
format!("{{ {} }}", inner_formatted)
}
DataValue::Array(inner_arr) => {
let formatted = format_array(inner_arr);
format!("[{}]", formatted)
}
DataValue::Bool(b) => b.to_string(),
DataValue::BoolTrue => "true".to_string(),
DataValue::BoolFalse => "false".to_string(),
DataValue::Number(num) => num.to_string(),
_ => String::new(),
})
.collect();
parts.join(", ")
}
#[macro_export] #[macro_export]
macro_rules! log_cv { macro_rules! log_cv {
($kind:expr, $cv:expr) => { ($kind:expr, $cv:expr) => {

1
src/util/mod.rs
View file

@ -1,3 +1,4 @@
pub mod crypto_helper; pub mod crypto_helper;
pub mod crypto_util; pub mod crypto_util;
pub mod file_util;
pub mod logger; pub mod logger;