Local Auth (start of dezentralization)
Prep for WebUI Upgrade to SSL optional
This commit is contained in:
Alex Emmet
parent
258404b7fa
commit
5f40158b00
11 changed files with 456 additions and 73 deletions
|
|
@ -1,3 +1,4 @@
|
|||
use crate::auth::local_auth;
|
||||
use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes};
|
||||
use crate::gui::log_panel::{log_cv, log_message, log_message_trans};
|
||||
use crate::langu::language_manager::format;
|
||||
|
|
@ -5,33 +6,39 @@ use crate::users::contact::Contact;
|
|||
use crate::users::user_community_util::UserCommunityUtil;
|
||||
use crate::util::chat_files;
|
||||
use crate::util::chats_util::{get_user, get_users, mod_user};
|
||||
use crate::util::file_util::{get_children, load_file, save_file};
|
||||
use futures::Stream;
|
||||
use futures::stream::{SplitSink, SplitStream};
|
||||
use futures_util::sink::Sink;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use hyper::upgrade::Upgraded;
|
||||
use hyper_util::rt::TokioIo;
|
||||
use json::JsonValue;
|
||||
use std::collections::HashMap;
|
||||
use std::str::FromStr;
|
||||
use std::sync::Arc;
|
||||
use std::sync::{Arc, LazyLock};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use tokio::net::TcpStream;
|
||||
use tokio::sync::Mutex;
|
||||
use tokio::sync::{Mutex, RwLock};
|
||||
use tokio::time::{Duration, Instant, sleep};
|
||||
use tokio_tungstenite::{
|
||||
MaybeTlsStream, WebSocketStream, connect_async, tungstenite::protocol::Message,
|
||||
};
|
||||
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message};
|
||||
use tungstenite::Utf8Bytes;
|
||||
use uuid::Uuid;
|
||||
pub static OMIKRON_CONNECTION: LazyLock<Arc<RwLock<Option<Arc<OmikronConnection>>>>> =
|
||||
LazyLock::new(|| Arc::new(RwLock::new(None)));
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum ConnectionVariant {
|
||||
Omikron,
|
||||
ClientUnauthenticated,
|
||||
ClientAuthenticated,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct OmikronConnection {
|
||||
pub(crate) writer: Arc<
|
||||
Mutex<
|
||||
Option<
|
||||
futures_util::stream::SplitSink<
|
||||
WebSocketStream<MaybeTlsStream<TcpStream>>,
|
||||
Message,
|
||||
>,
|
||||
>,
|
||||
>,
|
||||
>,
|
||||
pub variant: Arc<RwLock<ConnectionVariant>>,
|
||||
pub user_id: Arc<RwLock<Option<Uuid>>>,
|
||||
pub(crate) writer:
|
||||
Arc<Mutex<Option<Box<dyn Sink<Message, Error = tungstenite::Error> + Send + Unpin>>>>,
|
||||
waiting: Arc<Mutex<HashMap<Uuid, Box<dyn Fn(CommunicationValue) + Send + Sync>>>>, // waiting for responses
|
||||
pingpong: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>, // ping-pong handler
|
||||
pub last_ping: Arc<Mutex<i64>>,
|
||||
|
|
@ -42,6 +49,8 @@ pub struct OmikronConnection {
|
|||
impl OmikronConnection {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
variant: Arc::new(RwLock::new(ConnectionVariant::Omikron)),
|
||||
user_id: Arc::new(RwLock::new(None)),
|
||||
writer: Arc::new(Mutex::new(None)),
|
||||
waiting: Arc::new(Mutex::new(HashMap::new())),
|
||||
pingpong: Arc::new(Mutex::new(None)),
|
||||
|
|
@ -50,17 +59,42 @@ impl OmikronConnection {
|
|||
is_connected: Arc::new(Mutex::new(false)),
|
||||
}
|
||||
}
|
||||
pub async fn client(
|
||||
writer: SplitSink<tokio_tungstenite::WebSocketStream<TokioIo<Upgraded>>, Message>,
|
||||
reader: SplitStream<tokio_tungstenite::WebSocketStream<TokioIo<Upgraded>>>,
|
||||
) -> Arc<Self> {
|
||||
let connection = Arc::new(Self {
|
||||
variant: Arc::new(RwLock::new(ConnectionVariant::ClientUnauthenticated)),
|
||||
user_id: Arc::new(RwLock::new(None)),
|
||||
writer: Arc::new(Mutex::new(Some(Box::new(writer)
|
||||
as Box<dyn Sink<Message, Error = tungstenite::Error> + Send + Unpin>))),
|
||||
waiting: Arc::new(Mutex::new(HashMap::new())),
|
||||
pingpong: Arc::new(Mutex::new(None)),
|
||||
last_ping: Arc::new(Mutex::new(-1)),
|
||||
message_send_times: Arc::new(Mutex::new(HashMap::new())),
|
||||
is_connected: Arc::new(Mutex::new(false)),
|
||||
});
|
||||
let boxed_reader: Box<
|
||||
dyn Stream<Item = Result<Message, tungstenite::Error>> + Send + Unpin,
|
||||
> = Box::new(reader);
|
||||
|
||||
connection.spawn_listener(boxed_reader).await;
|
||||
connection
|
||||
}
|
||||
pub async fn is_connected(&self) -> bool {
|
||||
*self.is_connected.lock().await
|
||||
}
|
||||
/// Connect loop with retry
|
||||
pub async fn connect(&self) {
|
||||
pub async fn connect(self: &Arc<Self>) {
|
||||
loop {
|
||||
match connect_async("wss://app.tensamin.net/ws/iota/").await {
|
||||
Ok((ws_stream, _)) => {
|
||||
let (write_half, read_half) = ws_stream.split();
|
||||
*self.writer.lock().await = Some(write_half);
|
||||
self.spawn_listener(read_half).await;
|
||||
*self.writer.lock().await = Some(Box::new(write_half));
|
||||
let boxed_reader: Box<
|
||||
dyn Stream<Item = Result<Message, tungstenite::Error>> + Send + Unpin,
|
||||
> = Box::new(read_half);
|
||||
self.clone().spawn_listener(boxed_reader).await;
|
||||
let cloned_self = self.clone();
|
||||
let handle = tokio::spawn(async move {
|
||||
loop {
|
||||
|
|
@ -85,23 +119,32 @@ impl OmikronConnection {
|
|||
Self::send_message_static(&self.writer, msg).await
|
||||
}
|
||||
|
||||
pub async fn set_variant(self: &Arc<Self>, variant: ConnectionVariant) {
|
||||
*self.variant.write().await = variant;
|
||||
}
|
||||
pub async fn set_user_id(self: &Arc<Self>, user_id: Uuid) {
|
||||
*self.user_id.write().await = Some(user_id);
|
||||
}
|
||||
|
||||
/// Listener for all incoming messages
|
||||
async fn spawn_listener(
|
||||
&self,
|
||||
mut read_half: futures_util::stream::SplitStream<
|
||||
WebSocketStream<MaybeTlsStream<TcpStream>>,
|
||||
>,
|
||||
self: &Arc<Self>,
|
||||
mut read_half: Box<dyn Stream<Item = Result<Message, tungstenite::Error>> + Send + Unpin>,
|
||||
) {
|
||||
let waiting_out = self.waiting.clone();
|
||||
let writer_out = self.writer.clone();
|
||||
let is_connected_out = self.is_connected.clone();
|
||||
let sel_out = self.clone();
|
||||
let variant = self.variant.clone();
|
||||
let sel_arc_out = self.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(msg) = read_half.next().await {
|
||||
let waiting = waiting_out.clone();
|
||||
let writer = writer_out.clone();
|
||||
let is_connected = is_connected_out.clone();
|
||||
let sel = sel_out.clone();
|
||||
let variant = variant.clone();
|
||||
let sel_arc = sel_arc_out.clone();
|
||||
tokio::spawn(async move {
|
||||
match msg {
|
||||
Ok(Message::Close(Some(frame))) => {
|
||||
|
|
@ -115,6 +158,96 @@ impl OmikronConnection {
|
|||
sel.handle_pong(&cv, true).await;
|
||||
return;
|
||||
}
|
||||
let com = variant.read().await.clone();
|
||||
if com == ConnectionVariant::ClientUnauthenticated {
|
||||
if cv.is_type(CommunicationType::identification) {
|
||||
// Extract user ID
|
||||
let user_id = match cv.get_data(DataTypes::user_id) {
|
||||
Some(id_str) => {
|
||||
match Uuid::parse_str(&id_str.to_string()) {
|
||||
Ok(id) => id,
|
||||
Err(_) => {
|
||||
sel_arc.send_message(
|
||||
CommunicationValue::new(CommunicationType::error_invalid_user_id)
|
||||
.with_id(cv.get_id())
|
||||
.to_json()
|
||||
.to_string()
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
sel_arc
|
||||
.send_message(
|
||||
CommunicationValue::new(
|
||||
CommunicationType::error_invalid_user_id,
|
||||
)
|
||||
.with_id(cv.get_id())
|
||||
.to_json()
|
||||
.to_string(),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
// Validate private key
|
||||
if let Some(private_key_hash) =
|
||||
cv.get_data(DataTypes::private_key_hash)
|
||||
{
|
||||
log_message(format!(
|
||||
"private_key_hash: {}",
|
||||
private_key_hash
|
||||
));
|
||||
let is_valid = local_auth::is_private_key_valid(
|
||||
&user_id,
|
||||
&private_key_hash.to_string(),
|
||||
);
|
||||
|
||||
if !is_valid {
|
||||
log_message("Invalid private key");
|
||||
sel_arc.send_message(
|
||||
CommunicationValue::new(
|
||||
CommunicationType::error_invalid_private_key,
|
||||
)
|
||||
.with_id(cv.get_id())
|
||||
.to_json()
|
||||
.to_string(),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
log_message("Missing private key");
|
||||
sel_arc
|
||||
.send_message(
|
||||
CommunicationValue::new(
|
||||
CommunicationType::error_invalid_private_key,
|
||||
)
|
||||
.with_id(cv.get_id())
|
||||
.to_json()
|
||||
.to_string(),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
|
||||
// Set identification data
|
||||
|
||||
sel_arc.set_user_id(user_id).await;
|
||||
sel_arc
|
||||
.set_variant(ConnectionVariant::ClientAuthenticated)
|
||||
.await;
|
||||
|
||||
let response = CommunicationValue::new(
|
||||
CommunicationType::identification_response,
|
||||
)
|
||||
.with_id(cv.get_id());
|
||||
sel_arc.send_message(response.to_json().to_string()).await;
|
||||
}
|
||||
}
|
||||
// ************************************************ //
|
||||
// Direct messages //
|
||||
// ************************************************ //
|
||||
|
|
@ -326,6 +459,79 @@ impl OmikronConnection {
|
|||
.await;
|
||||
return;
|
||||
}
|
||||
|
||||
if cv.is_type(CommunicationType::settings_save) {
|
||||
let my_id = cv.get_sender().unwrap();
|
||||
let settings_name =
|
||||
cv.get_data(DataTypes::settings_name).unwrap().to_string();
|
||||
let settings_value =
|
||||
cv.get_data(DataTypes::payload).unwrap().to_string();
|
||||
|
||||
save_file(
|
||||
&format!("users/{}/settings/", my_id),
|
||||
&format!("{}.settings", settings_name),
|
||||
&settings_value,
|
||||
);
|
||||
|
||||
let response =
|
||||
CommunicationValue::new(CommunicationType::settings_save)
|
||||
.with_receiver(my_id)
|
||||
.with_id(cv.get_id());
|
||||
|
||||
Self::send_message_static(
|
||||
&writer.clone(),
|
||||
response.to_json().to_string(),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
if cv.is_type(CommunicationType::settings_load) {
|
||||
let my_id = cv.get_sender().unwrap();
|
||||
let settings_name =
|
||||
cv.get_data(DataTypes::settings_name).unwrap().to_string();
|
||||
let settings_value_str = load_file(
|
||||
&format!("users/{}/settings/", my_id),
|
||||
&format!("{}.settings", settings_name),
|
||||
);
|
||||
let settings_value_json = JsonValue::from(settings_value_str);
|
||||
let response =
|
||||
CommunicationValue::new(CommunicationType::settings_load)
|
||||
.with_id(cv.get_id())
|
||||
.with_receiver(my_id)
|
||||
.add_data(DataTypes::payload, settings_value_json)
|
||||
.add_data_str(DataTypes::settings_name, settings_name);
|
||||
|
||||
Self::send_message_static(
|
||||
&writer.clone(),
|
||||
response.to_json().to_string(),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
if cv.is_type(CommunicationType::settings_list) {
|
||||
let my_id = cv.get_sender().unwrap();
|
||||
let settings = get_children(&format!("users/{}/settings/", my_id));
|
||||
let mut settings_json = JsonValue::new_array();
|
||||
for s in settings {
|
||||
let s = s.replace(".settings", "");
|
||||
if s.is_empty() {
|
||||
continue;
|
||||
}
|
||||
let _ = settings_json.push(JsonValue::String(s));
|
||||
}
|
||||
let response =
|
||||
CommunicationValue::new(CommunicationType::settings_list)
|
||||
.with_id(cv.get_id())
|
||||
.with_receiver(my_id)
|
||||
.add_data(DataTypes::settings, settings_json);
|
||||
|
||||
Self::send_message_static(
|
||||
&writer.clone(),
|
||||
response.to_json().to_string(),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
log_message(format!("[Omikron] Error: {}", e));
|
||||
|
|
@ -340,14 +546,7 @@ impl OmikronConnection {
|
|||
}
|
||||
pub async fn send_message_static(
|
||||
writer: &Arc<
|
||||
Mutex<
|
||||
Option<
|
||||
futures_util::stream::SplitSink<
|
||||
WebSocketStream<MaybeTlsStream<TcpStream>>,
|
||||
Message,
|
||||
>,
|
||||
>,
|
||||
>,
|
||||
Mutex<Option<Box<dyn Sink<Message, Error = tungstenite::Error> + Send + Unpin>>>,
|
||||
>,
|
||||
msg: String,
|
||||
) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue