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

[Add] base of message states

Browse source
This commit is contained in:
Alex Emmet 2026-02-01 16:29:09 +01:00
commit 87d4ffab19
6 changed files with 153 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

3
src/data/communication.rs
View file

@ -53,6 +53,7 @@ pub enum DataTypes {
signature, signature,
signed, signed,
message, message,
message_state,
last_ping, last_ping,
ping_iota, ping_iota,
ping_clients, ping_clients,
@ -112,6 +113,7 @@ impl DataTypes {
#[allow(non_camel_case_types, dead_code)] #[allow(non_camel_case_types, dead_code)]
pub enum CommunicationType { pub enum CommunicationType {
error, error,
error_anonymous,
error_internal, error_internal,
error_invalid_data, error_invalid_data,
error_invalid_user_id, error_invalid_user_id,
@ -134,6 +136,7 @@ pub enum CommunicationType {
settings_load, settings_load,
settings_list, settings_list,
message, message,
message_state,
message_send, message_send,
message_live, message_live,
message_other_iota, message_other_iota,

146
src/omikron/omikron_connection.rs
View file

@ -2,6 +2,7 @@ use crate::auth::local_auth;
use crate::gui::log_panel::{log_cv, log_message_format}; use crate::gui::log_panel::{log_cv, log_message_format};
use crate::users::contact::Contact; use crate::users::contact::Contact;
use crate::users::user_community_util::UserCommunityUtil; use crate::users::user_community_util::UserCommunityUtil;
use crate::util::chat_files::MessageState;
use crate::util::chats_util::{get_user, mod_user}; use crate::util::chats_util::{get_user, mod_user};
use crate::util::crypto_util::{DataFormat, SecurePayload}; use crate::util::crypto_util::{DataFormat, SecurePayload};
use crate::util::file_util::{get_children, load_file, save_file}; use crate::util::file_util::{get_children, load_file, save_file};
@ -13,14 +14,15 @@ use crate::{
util::{config_util::CONFIG, crypto_helper}, util::{config_util::CONFIG, crypto_helper},
}; };
use dashmap::DashMap; use dashmap::DashMap;
use futures::Stream;
use futures::stream::{SplitSink, SplitStream}; use futures::stream::{SplitSink, SplitStream};
use futures::{FutureExt, Stream};
use futures_util::sink::Sink; use futures_util::sink::Sink;
use futures_util::{SinkExt, StreamExt}; use futures_util::{SinkExt, StreamExt};
use hyper::upgrade::Upgraded; use hyper::upgrade::Upgraded;
use hyper_util::rt::TokioIo; use hyper_util::rt::TokioIo;
use json::JsonValue; use json::JsonValue;
use json::number::Number; use json::number::Number;
use pkcs8::DecodePrivateKey;
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::{Arc, LazyLock}; use std::sync::{Arc, LazyLock};
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
@ -29,6 +31,7 @@ use tokio::time::{Duration, Instant, sleep};
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message}; use tokio_tungstenite::{connect_async, tungstenite::protocol::Message};
use tungstenite::Utf8Bytes; use tungstenite::Utf8Bytes;
use uuid::Uuid; use uuid::Uuid;
use warp::reply::Json;
pub static OMIKRON_CONNECTION: LazyLock<Arc<RwLock<Option<Arc<OmikronConnection>>>>> = pub static OMIKRON_CONNECTION: LazyLock<Arc<RwLock<Option<Arc<OmikronConnection>>>>> =
LazyLock::new(|| Arc::new(RwLock::new(None))); LazyLock::new(|| Arc::new(RwLock::new(None)));
@ -148,13 +151,10 @@ impl OmikronConnection {
// Login flow // Login flow
if self.connect_internal().await { if self.connect_internal().await {
self.send_message( self.send_message(
CommunicationValue::new(CommunicationType::identification) &CommunicationValue::new(CommunicationType::identification).add_data(
.add_data(
DataTypes::iota_id, DataTypes::iota_id,
JsonValue::Number(json::number::Number::from(iota_id)), JsonValue::Number(json::number::Number::from(iota_id)),
) ),
.to_json()
.to_string(),
) )
.await; .await;
} }
@ -207,8 +207,13 @@ impl OmikronConnection {
true true
} }
pub async fn send_message(&self, msg: String) { pub async fn send_message(&self, cv: &CommunicationValue) {
Self::send_message_static(&self.writer, Arc::clone(&self.is_connected), msg).await; Self::send_message_static(
&self.writer,
Arc::clone(&self.is_connected),
cv.to_json().to_string(),
)
.await;
} }
pub async fn set_variant(self: &Arc<Self>, variant: ConnectionVariant) { pub async fn set_variant(self: &Arc<Self>, variant: ConnectionVariant) {
@ -228,7 +233,6 @@ impl OmikronConnection {
let is_connected_out = self.is_connected.clone(); let is_connected_out = self.is_connected.clone();
let sel_out = self.clone(); let sel_out = self.clone();
let variant = self.variant.clone(); let variant = self.variant.clone();
let sel_arc_out = self.clone();
{ {
ACTIVE_TASKS.lock().unwrap().push("Listener".to_string()); ACTIVE_TASKS.lock().unwrap().push("Listener".to_string());
@ -238,14 +242,12 @@ impl OmikronConnection {
if *SHUTDOWN.read().await { if *SHUTDOWN.read().await {
break; break;
} }
Self::handle_message( sel_out.clone().handle_message(
msg, msg,
waiting_out.clone(), waiting_out.clone(),
writer_out.clone(), writer_out.clone(),
is_connected_out.clone(), is_connected_out.clone(),
sel_out.clone(),
variant.clone(), variant.clone(),
sel_arc_out.clone(),
); );
} }
*is_connected_out.lock().await = false; *is_connected_out.lock().await = false;
@ -259,6 +261,7 @@ impl OmikronConnection {
} }
} }
pub fn handle_message( pub fn handle_message(
self: Arc<Self>,
msg: Result<Message, tungstenite::Error>, msg: Result<Message, tungstenite::Error>,
waiting: Arc<DashMap<Uuid, Box<dyn Fn(CommunicationValue) + Send + Sync + 'static>>>, waiting: Arc<DashMap<Uuid, Box<dyn Fn(CommunicationValue) + Send + Sync + 'static>>>,
writer: Arc< writer: Arc<
@ -267,9 +270,7 @@ impl OmikronConnection {
>, >,
>, >,
is_connected: Arc<Mutex<bool>>, is_connected: Arc<Mutex<bool>>,
sel: Arc<OmikronConnection>,
variant: Arc<RwLock<ConnectionVariant>>, variant: Arc<RwLock<ConnectionVariant>>,
sel_arc: Arc<OmikronConnection>,
) { ) {
tokio::spawn(async move { tokio::spawn(async move {
match msg { match msg {
@ -281,7 +282,7 @@ impl OmikronConnection {
Ok(Message::Text(text)) => { Ok(Message::Text(text)) => {
let cv = CommunicationValue::from_json(&text); let cv = CommunicationValue::from_json(&text);
if cv.is_type(CommunicationType::pong) { if cv.is_type(CommunicationType::pong) {
sel.handle_pong(&cv, true).await; self.handle_pong(&cv, true).await;
return; return;
} }
if cv.is_type(CommunicationType::challenge) { if cv.is_type(CommunicationType::challenge) {
@ -324,7 +325,7 @@ impl OmikronConnection {
JsonValue::String(decrypted.export(DataFormat::Base64)), JsonValue::String(decrypted.export(DataFormat::Base64)),
); );
sel_arc.send_message(response.to_json().to_string()).await; self.send_message(&response).await;
} else { } else {
log_message("Failed to decrypt challenge"); log_message("Failed to decrypt challenge");
} }
@ -349,13 +350,11 @@ impl OmikronConnection {
.add_data( .add_data(
DataTypes::iota_id, DataTypes::iota_id,
JsonValue::Number(json::number::Number::from(iota_id)), JsonValue::Number(json::number::Number::from(iota_id)),
) );
.to_json()
.to_string();
let sel_arc_clone = sel_arc.clone(); let self_clone = self.clone();
tokio::spawn(async move { tokio::spawn(async move {
sel_arc_clone.send_message(login_message).await; self_clone.send_message(&login_message).await;
}); });
} else { } else {
log_message("Iota registration failed."); log_message("Iota registration failed.");
@ -379,14 +378,11 @@ impl OmikronConnection {
.as_i64() .as_i64()
.unwrap_or(0); .unwrap_or(0);
if user_id == 0 { if user_id == 0 {
sel_arc self.send_message(
.send_message( &CommunicationValue::new(
CommunicationValue::new(
CommunicationType::error_invalid_user_id, CommunicationType::error_invalid_user_id,
) )
.with_id(cv.get_id()) .with_id(cv.get_id()),
.to_json()
.to_string(),
) )
.await; .await;
return; return;
@ -403,28 +399,22 @@ impl OmikronConnection {
if !is_valid { if !is_valid {
log_message("Invalid private key"); log_message("Invalid private key");
sel_arc self.send_message(
.send_message( &CommunicationValue::new(
CommunicationValue::new(
CommunicationType::error_invalid_private_key, CommunicationType::error_invalid_private_key,
) )
.with_id(cv.get_id()) .with_id(cv.get_id()),
.to_json()
.to_string(),
) )
.await; .await;
return; return;
} }
} else { } else {
log_message("Missing private key"); log_message("Missing private key");
sel_arc self.send_message(
.send_message( &CommunicationValue::new(
CommunicationValue::new(
CommunicationType::error_invalid_private_key, CommunicationType::error_invalid_private_key,
) )
.with_id(cv.get_id()) .with_id(cv.get_id()),
.to_json()
.to_string(),
) )
.await; .await;
return; return;
@ -432,15 +422,14 @@ impl OmikronConnection {
// Set identification data // Set identification data
sel_arc.set_user_id(user_id).await; self.set_user_id(user_id).await;
sel_arc self.set_variant(ConnectionVariant::ClientAuthenticated)
.set_variant(ConnectionVariant::ClientAuthenticated)
.await; .await;
let response = let response =
CommunicationValue::new(CommunicationType::identification_response) CommunicationValue::new(CommunicationType::identification_response)
.with_id(cv.get_id()); .with_id(cv.get_id());
sel_arc.send_message(response.to_json().to_string()).await; self.send_message(&response).await;
} }
} }
// ************************************************ // // ************************************************ //
@ -451,6 +440,25 @@ impl OmikronConnection {
y(cv); y(cv);
return; return;
} }
if cv.is_type(CommunicationType::message_state) {
let sender_id = &cv.get_sender();
let receiver_id = &cv.get_receiver();
chat_files::change_message_state(
cv.get_data(DataTypes::send_time)
.unwrap_or(&JsonValue::new_object())
.as_i64()
.unwrap_or(0) as i64,
*receiver_id,
*sender_id,
MessageState::from_str(
cv.get_data(DataTypes::message_state)
.unwrap_or(&JsonValue::Null)
.as_str()
.unwrap_or(""),
),
);
}
if cv.is_type(CommunicationType::message_other_iota) { if cv.is_type(CommunicationType::message_other_iota) {
let sender_id = &cv.get_sender(); let sender_id = &cv.get_sender();
let receiver_id = &cv.get_receiver(); let receiver_id = &cv.get_receiver();
@ -485,12 +493,50 @@ impl OmikronConnection {
DataTypes::sender_id, DataTypes::sender_id,
JsonValue::Number(Number::from(cv.get_sender())), JsonValue::Number(Number::from(cv.get_sender())),
); );
Self::send_message_static( let user_resp = self
&writer.clone(), .clone()
is_connected, .await_response(&user_forward, Some(Duration::from_secs(10)))
user_forward.to_json().to_string(),
)
.await; .await;
if let Ok(user_resp) = user_resp {
let ms: MessageState = MessageState::from_str(
user_resp
.get_data(DataTypes::message_state)
.unwrap_or(&JsonValue::Null)
.as_str()
.unwrap_or(""),
)
.upgrade(MessageState::Send);
self.send_message(
&CommunicationValue::new(CommunicationType::message_state)
.with_id(cv.get_id())
.with_receiver(*sender_id)
.with_sender(*receiver_id)
.add_data(
DataTypes::send_time,
cv.get_data(DataTypes::send_time).unwrap().clone(),
)
.add_data(
DataTypes::message_state,
JsonValue::from(ms.as_str()),
),
);
} else {
self.send_message(
&CommunicationValue::new(CommunicationType::message_state)
.with_id(cv.get_id())
.with_receiver(*sender_id)
.with_sender(*receiver_id)
.add_data(
DataTypes::send_time,
cv.get_data(DataTypes::send_time).unwrap().clone(),
)
.add_data(
DataTypes::message_state,
JsonValue::from(MessageState::Send.as_str()),
),
);
}
return; return;
} }
@ -821,13 +867,13 @@ impl OmikronConnection {
}), }),
); );
self.send_message(cv.to_json().to_string()).await; self.send_message(&cv).await;
let timeout = timeout_duration.unwrap_or(Duration::from_secs(10)); let timeout = timeout_duration.unwrap_or(Duration::from_secs(10));
match tokio::time::timeout(timeout, rx.recv()).await { match tokio::time::timeout(timeout, rx.recv()).await {
Ok(Some(response_cv)) => Ok(response_cv), Ok(Some(response_cv)) => Ok(response_cv),
Ok(None) => Err("Failed to receive response, channel was closed.".to_string()), Ok(_) => Err("Failed to receive response, channel was closed.".to_string()),
Err(_) => { Err(_) => {
self.waiting.remove(&msg_id); self.waiting.remove(&msg_id);
Err(format!( Err(format!(

6
src/omikron/ping_pong_task.rs
View file

@ -20,11 +20,9 @@ impl OmikronConnection {
.add_data_num( .add_data_num(
DataTypes::last_ping, DataTypes::last_ping,
Number::from(*self.last_ping.lock().await), Number::from(*self.last_ping.lock().await),
) );
.to_json()
.to_string();
self.send_message(ping_message).await; self.send_message(&ping_message).await;
} }
/// Handles incoming pong and calculates latency /// Handles incoming pong and calculates latency

36
src/util/chat_files.rs
View file

@ -5,20 +5,40 @@ use std::path::Path;
use crate::gui::log_panel::log_message; use crate::gui::log_panel::log_message;
#[derive(PartialEq, Debug, Clone)]
pub enum MessageState { pub enum MessageState {
Read, Read,
Received, Received,
Send,
Sending, Sending,
Error,
} }
impl MessageState { impl MessageState {
fn as_str(&self) -> &'static str { pub fn as_str(&self) -> &'static str {
match self { match self {
MessageState::Read => "READ", MessageState::Read => "READ",
MessageState::Received => "RECEIVED", MessageState::Received => "RECEIVED",
MessageState::Send => "SEND",
MessageState::Sending => "SENDING", MessageState::Sending => "SENDING",
MessageState::Error => "ERROR", }
}
pub fn from_str(str: &str) -> Self {
match str.to_uppercase().as_str() {
"READ" => MessageState::Read,
"RECEIVED" => MessageState::Received,
"SEND" => MessageState::Send,
_ => MessageState::Sending,
}
}
pub fn upgrade(self, other: Self) -> Self {
if other == Self::Read || self == Self::Read {
Self::Read
} else if other == Self::Received || self == Self::Received {
Self::Received
} else if other == Self::Send || self == Self::Send {
Self::Send
} else {
Self::Sending
} }
} }
} }
@ -86,9 +106,9 @@ pub fn add_message(
save_file(&user_dir, &file_name, &message_chunk.dump()); save_file(&user_dir, &file_name, &message_chunk.dump());
} }
pub fn change_message_state( pub fn change_message_state(
timestamp: i64,
storage_owner: i64, storage_owner: i64,
external_user: i64, external_user: i64,
timestamp: i64,
new_state: MessageState, new_state: MessageState,
) -> std::io::Result<()> { ) -> std::io::Result<()> {
let user_dir = format!("users/{}/chats/{}", storage_owner, external_user); let user_dir = format!("users/{}/chats/{}", storage_owner, external_user);
@ -114,7 +134,13 @@ pub fn change_message_state(
let mut modified = false; let mut modified = false;
for i in 0..chunk.len() { for i in 0..chunk.len() {
if chunk[i]["message_time"].as_i64() == Some(timestamp) { if chunk[i]["message_time"].as_i64() == Some(timestamp) {
chunk[i]["message_state"] = JsonValue::from(new_state.as_str()); chunk[i]["message_state"] = JsonValue::from(
MessageState::from_str(
chunk[i]["message_state"].as_str().unwrap_or("SENDING"),
)
.upgrade(new_state.clone())
.as_str(),
);
modified = true; modified = true;
break; break;
} }

11
src/util/crypto_util.rs
View file

@ -109,11 +109,6 @@ impl SecurePayload {
let peer_pub = public_key.into(); let peer_pub = public_key.into();
let shared_secret = self.private_key.as_diffie_hellman(&peer_pub).unwrap(); let shared_secret = self.private_key.as_diffie_hellman(&peer_pub).unwrap();
println!(
"Encryption Shared Secret (Hex): {}",
hex::encode(shared_secret.as_bytes())
);
// 3. Key & Nonce Derivation (HKDF) // 3. Key & Nonce Derivation (HKDF)
// We derive 32 bytes for the key and 12 bytes for a deterministic nonce. // We derive 32 bytes for the key and 12 bytes for a deterministic nonce.
let hkdf = Hkdf::<Sha256>::new(None, shared_secret.as_bytes()); let hkdf = Hkdf::<Sha256>::new(None, shared_secret.as_bytes());
@ -168,12 +163,6 @@ impl SecurePayload {
let peer_pub = peer_public_key_bytes.into(); let peer_pub = peer_public_key_bytes.into();
let shared_secret = self.private_key.as_diffie_hellman(&peer_pub).unwrap(); let shared_secret = self.private_key.as_diffie_hellman(&peer_pub).unwrap();
// LOGGING: Shared Secret
println!(
"Decryption Shared Secret (Hex): {}",
hex::encode(shared_secret.as_bytes())
);
// 2. Key & Nonce Derivation (Must match encryption exactly) // 2. Key & Nonce Derivation (Must match encryption exactly)
let hkdf = Hkdf::<Sha256>::new(None, shared_secret.as_bytes()); let hkdf = Hkdf::<Sha256>::new(None, shared_secret.as_bytes());
let mut okm = [0u8; 44]; let mut okm = [0u8; 44];