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

stable omega connection & iota auth

Browse source
This commit is contained in:
Alex Emmet 2026-01-14 16:22:26 +01:00
commit 1e33ad930a
3 changed files with 105 additions and 150 deletions
Download patch file Download diff file Expand all files Collapse all files

20
src/calls/call_manager.rs
View file

@ -94,23 +94,9 @@ pub fn garbage_collect_calls() {
}); });
} }
pub async fn clean_calls() { pub async fn clean_calls() {
let api_key = match env::var("LIVEKIT_API_KEY") { let room_service = RoomClient::new("https://call.tensamin.net").unwrap();
Ok(key) => key, let rooms = room_service.list_rooms(Vec::new()).await.unwrap();
Err(_) => { log!(PrintType::General, "{:?}", rooms);
log_err!(PrintType::General, "LIVEKIT_API_KEY not set!");
return;
}
};
let api_secret = match env::var("LIVEKIT_API_SECRET") {
Ok(secret) => secret,
Err(_) => {
log_err!(PrintType::General, "LIVEKIT_API_SECRET not set!");
return;
}
};
let room_service =
RoomClient::with_api_key("https://call.tensamin.net/", &api_key, &api_secret);
let rooms = match room_service.list_rooms(Vec::new()).await { let rooms = match room_service.list_rooms(Vec::new()).await {
Ok(rooms) => rooms, Ok(rooms) => rooms,
Err(e) => { Err(e) => {

17
src/omega/omega_connection.rs
View file

@ -33,10 +33,6 @@ pub static WAITING_TASKS: Lazy<
DashMap<Uuid, Box<dyn Fn(Arc<OmegaConnection>, CommunicationValue) -> bool + Send + Sync>>, DashMap<Uuid, Box<dyn Fn(Arc<OmegaConnection>, CommunicationValue) -> bool + Send + Sync>>,
> = Lazy::new(DashMap::new); > = Lazy::new(DashMap::new);
static GENERIC_TASK: Lazy<
Mutex<Option<Box<dyn Fn(Arc<OmegaConnection>, CommunicationValue) -> bool + Send + Sync>>>,
> = Lazy::new(|| Mutex::new(None));
static OMEGA_CONNECTION: Lazy<Arc<OmegaConnection>> = Lazy::new(|| { static OMEGA_CONNECTION: Lazy<Arc<OmegaConnection>> = Lazy::new(|| {
let conn = Arc::new(OmegaConnection::new()); let conn = Arc::new(OmegaConnection::new());
let conn_clone = conn.clone(); let conn_clone = conn.clone();
@ -311,24 +307,17 @@ impl OmegaConnection {
continue; continue;
} }
let msg_id = cv.get_id(); let msg_id = cv.get_id();
log_in!(PrintType::Omikron, "{}", &cv.to_json().to_string()); log_in!(PrintType::Omega, "{}", &cv.to_json().to_string());
// Handle waiting tasks // Handle waiting tasks
if let Some(task) = WAITING_TASKS.remove(&msg_id) { if let Some(task) = WAITING_TASKS.remove(&msg_id) {
if (task.1)(self.clone(), cv.clone()) { if (task.1)(self.clone(), cv.clone()) {
// continue in the read_loop continue;
} }
} else { } else {
// Handle generic task
let generic_task_option = GENERIC_TASK.lock().await;
if let Some(generic_task) = generic_task_option.as_ref() {
if generic_task(self.clone(), cv.clone()) {
// continue in the read_loop
}
}
} }
} }
#[allow(non_snake_case)] #[allow(non_snake_case)]
Some(Ok(Message::Close(_))) | None => break, Some(Ok(Message::Close(_))) | None => continue,
Some(Err(_)) => break, Some(Err(_)) => break,
_ => {} _ => {}
} }

196
src/rho/iota_connection.rs
View file

@ -21,6 +21,7 @@ use rand::distributions::Alphanumeric;
use std::{ use std::{
collections::HashMap, collections::HashMap,
sync::{Arc, Weak}, sync::{Arc, Weak},
time::Duration,
}; };
use tokio::sync::RwLock; use tokio::sync::RwLock;
use tokio_util::compat::Compat; use tokio_util::compat::Compat;
@ -137,6 +138,12 @@ impl IotaConnection {
/// 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>, message: Utf8Bytes) {
let cv = CommunicationValue::from_json(&message); let cv = CommunicationValue::from_json(&message);
// Handle ping
if cv.is_type(CommunicationType::ping) {
self.handle_ping(cv).await;
return;
}
let identified = *self.identified.read().await; let identified = *self.identified.read().await;
let challenged = *self.challenged.read().await; let challenged = *self.challenged.read().await;
@ -169,80 +176,70 @@ impl IotaConnection {
*self.user_ids.write().await = user_ids; *self.user_ids.write().await = user_ids;
let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_iota_data) 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)); .add_data(DataTypes::iota_id, JsonValue::from(iota_id));
let msg_id = get_pub_key_msg.get_id(); let response_cv = get_omega_connection()
let iota_conn_clone = self.clone(); .await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
let original_cv_id = cv.get_id(); .await;
WAITING_TASKS.insert( if let Ok(response_cv) = response_cv {
msg_id, if !response_cv.is_type(CommunicationType::get_iota_data) {
Box::new(move |_, response_cv: CommunicationValue| { self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
let iota_conn_for_task = iota_conn_clone.clone(); .await;
tokio::spawn(async move { self.close().await;
if !response_cv.is_type(CommunicationType::get_iota_data) { return;
iota_conn_for_task }
.send_error_response(
&original_cv_id,
CommunicationType::error_internal,
)
.await;
iota_conn_for_task.close().await;
return;
}
let base64_pub = response_cv let base64_pub = response_cv
.get_data(DataTypes::public_key) .get_data(DataTypes::public_key)
.and_then(|v| v.as_str()) .and_then(|v| v.as_str())
.unwrap_or(""); .unwrap_or("");
let pub_key = match load_public_key(base64_pub) { let pub_key = match load_public_key(base64_pub) {
Some(pk) => pk, Some(pk) => pk,
None => { None => {
iota_conn_for_task self.send_error_response(
.send_error_response( &cv.get_id(),
&original_cv_id, CommunicationType::error_invalid_public_key,
CommunicationType::error_invalid_public_key, )
) .await;
.await; self.close().await;
iota_conn_for_task.close().await; return;
return; }
} };
};
*iota_conn_for_task.pub_key.write().await = *self.pub_key.write().await = Some(pub_key.as_bytes().to_vec());
Some(pub_key.as_bytes().to_vec());
let challenge: String = rand::thread_rng() let challenge: String = rand::thread_rng()
.sample_iter(&Alphanumeric) .sample_iter(&Alphanumeric)
.take(32) .take(32)
.map(char::from) .map(char::from)
.collect(); .collect();
*iota_conn_for_task.challenge.write().await = challenge.clone(); *self.challenge.write().await = challenge.clone();
let encrypted_challenge = let encrypted_challenge =
encrypt(get_private_key(), pub_key, &challenge).unwrap_or_default(); encrypt(get_private_key(), pub_key, &challenge).unwrap_or_default();
*iota_conn_for_task.identified.write().await = true; *self.identified.write().await = true;
let challenge_msg = CommunicationValue::new(CommunicationType::challenge) let challenge_msg = CommunicationValue::new(CommunicationType::challenge)
.with_id(original_cv_id) .with_id(cv.get_id())
.add_data_str( .add_data_str(
DataTypes::public_key, DataTypes::public_key,
public_key_to_base64(&get_public_key()), public_key_to_base64(&get_public_key()),
) )
.add_data_str(DataTypes::challenge, encrypted_challenge); .add_data_str(DataTypes::challenge, encrypted_challenge);
iota_conn_for_task.send_message(&challenge_msg).await; self.send_message(&challenge_msg).await;
}); } else {
true self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
}), .await;
); }
get_omega_connection().send_message(&get_pub_key_msg).await;
return; return;
} else if !identified && cv.is_type(CommunicationType::complete_register_iota) { } else if !identified && cv.is_type(CommunicationType::register_iota) {
let base64_pub = cv let base64_pub = cv
.get_data(DataTypes::public_key) .get_data(DataTypes::public_key)
.and_then(|v| v.as_str()) .and_then(|v| v.as_str())
@ -256,60 +253,48 @@ impl IotaConnection {
} }
let register_msg = CommunicationValue::new(CommunicationType::complete_register_iota) let register_msg = CommunicationValue::new(CommunicationType::complete_register_iota)
.with_id(cv.get_id())
.add_data( .add_data(
DataTypes::public_key, DataTypes::public_key,
JsonValue::String(base64_pub.to_string()), JsonValue::String(base64_pub.to_string()),
); );
let msg_id = register_msg.get_id();
let iota_conn_clone = self.clone(); let iota_conn_clone = self.clone();
let original_cv_id = cv.get_id(); 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;
}
WAITING_TASKS.insert( let new_iota_id = register_response
msg_id, .get_data(DataTypes::iota_id)
Box::new(move |_, response_cv: CommunicationValue| { .and_then(|v| v.as_i64())
let iota_conn_for_task = iota_conn_clone.clone(); .unwrap_or(0);
tokio::spawn(async move {
if !response_cv.is_type(CommunicationType::complete_register_iota) {
iota_conn_for_task
.send_error_response(
&original_cv_id,
CommunicationType::error_internal,
)
.await;
iota_conn_for_task.close().await;
return;
}
let new_iota_id = response_cv if new_iota_id == 0 {
.get_data(DataTypes::iota_id) iota_conn_for_task
.and_then(|v| v.as_i64()) .send_error_response(&cv.get_id(), CommunicationType::error_internal)
.unwrap_or(0); .await;
iota_conn_for_task.close().await;
return;
}
if new_iota_id == 0 { *iota_conn_for_task.iota_id.write().await = new_iota_id;
iota_conn_for_task *iota_conn_for_task.identified.write().await = true;
.send_error_response(
&original_cv_id,
CommunicationType::error_internal,
)
.await;
iota_conn_for_task.close().await;
return;
}
*iota_conn_for_task.iota_id.write().await = new_iota_id; let success_msg = CommunicationValue::new(CommunicationType::success)
*iota_conn_for_task.identified.write().await = true; .with_id(cv.get_id())
.add_data(DataTypes::iota_id, JsonValue::from(new_iota_id));
let success_msg = CommunicationValue::new(CommunicationType::success) iota_conn_for_task.send_message(&success_msg).await;
.with_id(original_cv_id) }
.add_data(DataTypes::iota_id, JsonValue::from(new_iota_id));
iota_conn_for_task.send_message(&success_msg).await;
});
true
}),
);
get_omega_connection().send_message(&register_msg).await;
return; return;
} }
@ -374,11 +359,6 @@ impl IotaConnection {
return; return;
} }
// Handle ping
if cv.is_type(CommunicationType::ping) {
self.handle_ping(cv).await;
return;
}
log_in!(PrintType::Iota, "{}", &cv.to_json().to_string()); log_in!(PrintType::Iota, "{}", &cv.to_json().to_string());
// 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();