diff --git a/Cargo.lock b/Cargo.lock index 63b4a96..5245b66 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -91,6 +91,12 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "anyhow" +version = "1.0.97" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dcfed56ad506cb2c684a14971b8861fdc3baaaae314b9e5f9bb532cbe3ba7a4f" + [[package]] name = "api_key" version = "0.1.0" @@ -301,6 +307,26 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "fastwebsockets" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "305d3ba574508e27190906d11707dad683e0494e6b85eae9b044cb2734a5e422" +dependencies = [ + "base64", + "bytes", + "http-body-util", + "hyper 1.6.0", + "hyper-util", + "pin-project", + "rand", + "sha1", + "simdutf8", + "thiserror 1.0.69", + "tokio", + "utf-8", +] + [[package]] name = "fnv" version = "1.0.7" @@ -526,6 +552,29 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "http-body" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" +dependencies = [ + "bytes", + "http 1.3.1", +] + +[[package]] +name = "http-body-util" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" +dependencies = [ + "bytes", + "futures-core", + "http 1.3.1", + "http-body 1.0.1", + "pin-project-lite", +] + [[package]] name = "httparse" version = "1.10.1" @@ -550,7 +599,7 @@ dependencies = [ "futures-util", "h2", "http 0.2.12", - "http-body", + "http-body 0.4.6", "httparse", "httpdate", "itoa", @@ -562,6 +611,41 @@ dependencies = [ "want", ] +[[package]] +name = "hyper" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc2b571658e38e0c01b1fdca3bbbe93c00d3d71693ff2770043f8c29bc7d6f80" +dependencies = [ + "bytes", + "futures-channel", + "futures-util", + "http 1.3.1", + "http-body 1.0.1", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-util" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df2dcfbe0677734ab2f3ffa7fa7bfd4706bfdc1ef393f2ee30184aed67e631b4" +dependencies = [ + "bytes", + "futures-util", + "http 1.3.1", + "http-body 1.0.1", + "hyper 1.6.0", + "pin-project-lite", + "tokio", +] + [[package]] name = "iana-time-zone" version = "0.1.61" @@ -862,10 +946,15 @@ dependencies = [ name = "nexus" version = "0.1.0" dependencies = [ + "anyhow", "api_key", "chrono", "env_logger", + "fastwebsockets", "futures", + "http-body-util", + "hyper 1.6.0", + "hyper-util", "log", "os_path", "regex", @@ -1159,6 +1248,12 @@ dependencies = [ "libc", ] +[[package]] +name = "simdutf8" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" + [[package]] name = "simple-error" version = "0.3.1" @@ -1466,7 +1561,7 @@ dependencies = [ "futures-util", "headers", "http 0.2.12", - "hyper", + "hyper 0.14.32", "log", "mime", "mime_guess", diff --git a/Cargo.toml b/Cargo.toml index 80fac23..d644777 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,3 +21,8 @@ tokio-util = "0.7.13" url = "2.5.4" uuid = { version = "1.15.1", features = ["v4"] } warp = "0.3.7" +anyhow = "1.0.97" +fastwebsockets = { version = "0.10.0", features = ["hyper", "unstable-split", "upgrade"] } +http-body-util = "0.1.3" +hyper-util = "0.1.10" +hyper = "1.6.0" diff --git a/src/arbiter/mod.rs b/src/arbiter/mod.rs index 4d9a087..5fe00dd 100644 --- a/src/arbiter/mod.rs +++ b/src/arbiter/mod.rs @@ -17,13 +17,13 @@ use url::Url; use crate::server::models::{ CoreUserConfig, IPCMessageWithId, - Store, + NexusStore, }; pub async fn arbiter_main( ipc_tx: UnboundedSender, mut ipc_rx: UnboundedReceiver, - store: Arc, + store: Arc, user_config: Arc, cancellation_tokens: (CancellationToken, CancellationToken), ) { diff --git a/src/arbiter/models.rs b/src/arbiter/models.rs index ec5a61c..15a3387 100644 --- a/src/arbiter/models.rs +++ b/src/arbiter/models.rs @@ -58,6 +58,7 @@ pub struct ApiKey { pub allowed_events_from: Vec, pub user_id: String, pub echo: bool, + pub proxy: bool } impl ApiKey { @@ -68,6 +69,7 @@ impl ApiKey { allowed_events_from: self.allowed_events_from, user_id: self.user_id, echo: self.echo, + proxy: self.proxy } } } @@ -79,6 +81,7 @@ pub struct ApiKeyWithKey { pub allowed_events_from: Vec, pub user_id: String, pub echo: bool, + pub proxy: bool } impl Into for ApiKeyWithKey { @@ -88,6 +91,7 @@ impl Into for ApiKeyWithKey { allowed_events_from: self.allowed_events_from, user_id: self.user_id, echo: self.echo, + proxy: self.proxy } } } @@ -99,6 +103,7 @@ impl Into for ApiKeyWithKey { allowed_events_to: self.allowed_events_to, allowed_events_from: self.allowed_events_from, echo: self.echo, + proxy: self.proxy } } } @@ -109,4 +114,5 @@ pub struct ApiKeyWithKeyWithoutUID { pub allowed_events_to: Vec, pub allowed_events_from: Vec, pub echo: bool, + pub proxy: bool } diff --git a/src/client/mod.rs b/src/client/mod.rs new file mode 100644 index 0000000..a47d472 --- /dev/null +++ b/src/client/mod.rs @@ -0,0 +1,286 @@ +use crate::server::models::{ + IPCMessageWithId, + UserConfig, +}; +use fastwebsockets::{ + FragmentCollector, + Frame, + OpCode, + handshake, +}; +use http_body_util::Empty; +use hyper::{ + Request, + body::Bytes, + header::{ + CONNECTION, + UPGRADE, + }, +}; +use log::info; +use std::future::Future; +use std::sync::Arc; +use std::time::Duration; +use tokio::net::TcpStream; +use tokio::{ + sync::{ + Mutex, + mpsc::{ + UnboundedReceiver, + UnboundedSender, + unbounded_channel, + }, + }, + task::{ + JoinHandle, + spawn, + }, +}; +use tokio_util::sync::CancellationToken; + +/// Use when connecting a program to nexus *remotely*. Use the nexus::server::listener's +/// nexus_listener() function to run an embedded server. +pub struct NexusClient { + secure: bool, + url: String, + // TODO: Make optional as we can connect via OS socket as well. + port: u16, + config: UserConfig, + connected: bool, + keep_trying: bool, + to_server: Option>, + from_server: Option>, + // TODO: Add user registry to see if a user is connected via this client to route back instead of sending to server. +} + +struct AddSend(T); + +unsafe impl Send for AddSend {} +unsafe impl Sync for AddSend {} + +struct SpawnExecutor; + +impl hyper::rt::Executor for SpawnExecutor +where + Fut: Future + Send + 'static, + Fut::Output: Send + 'static, +{ + fn execute(&self, fut: Fut) { + tokio::task::spawn(fut); + } +} + +impl NexusClient { + pub fn new( + secure: bool, + url: &String, + port: &u16, + config: &UserConfig, + keep_trying: bool, + ) -> Self { + NexusClient { + secure, + url: url.clone(), + port: port.clone(), + config: config.clone(), + connected: false, + keep_trying, + to_server: None, + from_server: None + } + } + + // TODO: Send as connected API key. (ensure routing is supported) + // TODO: Send as proxied API key. (ensure routing is supported) + + pub async fn connect_to_ws( + &mut self, + ) -> Result< + ( + JoinHandle<()>, + (CancellationToken, CancellationToken) + ), + anyhow::Error, + > { + let mut keep_trying = true; + let cancellation_tokens = (CancellationToken::new(), CancellationToken::new()); + let mut ws_opt = None; + let mut error: Option = None; + + while keep_trying { + match TcpStream::connect(format!( + "{}:{}", + self.url.clone(), + self.port.clone().to_string() + )).await { + Ok(stream) => { + match Request::builder() + .method("GET") + .uri(format!( + "{}://{}:{}/", + (if self.secure { "https" } else { "http" }), + self.url.clone(), + self.port.clone().to_string() + )) + .header( + "Host", + format!("{}:{}", self.url.clone(), self.port.clone().to_string()), + ) + .header(UPGRADE, "websocket") + .header(CONNECTION, "upgrade") + .header( + "Sec-WebSocket-Key", + fastwebsockets::handshake::generate_key(), + ) + // TODO: Add API key! + .header("Sec-WebSocket-Version", "13") + .body(Empty::::new()) + { + Ok(req) => match handshake::client(&SpawnExecutor, req, stream).await { + Ok((the_socket, _)) => { + ws_opt = Some(the_socket); + } + Err(e) => { + if self.keep_trying { + tokio::time::sleep(Duration::from_secs(1)).await; + } else { + error = Some(e.into()); + keep_trying = false; + } + } + }, + Err(e) => { + if self.keep_trying { + tokio::time::sleep(Duration::from_secs(1)).await; + } else { + error = Some(e.into()); + keep_trying = false; + } + } + } + } + Err(e) => { + if self.keep_trying { + tokio::time::sleep(Duration::from_secs(1)).await; + } else { + error = Some(e.into()); + keep_trying = false; + } + } + } + } + + match ws_opt { + Some(ws) => { + self.connected = true; + + let (mut reader, og_writer) = ws.split(tokio::io::split); + let writer = Arc::new(Mutex::new(AddSend(og_writer))); + let (handle_from_tx, handle_from_rx) = unbounded_channel::(); + let (handle_to_tx, mut handle_to_rx) = unbounded_channel::(); + + let senders_writer = writer.clone(); + let mut sender = move |frame| { + let senders_writer_two = senders_writer.clone(); + async move { senders_writer_two.lock().await.0.write_frame(frame).await } + }; + + let handle_tokens = cancellation_tokens.clone(); + let handle = spawn(async move { + let recv_tokens = Arc::new(handle_tokens.clone()); + let recv_handle = spawn(async move { + recv_tokens + .0 + .run_until_cancelled(async move { + // TODO: Use FragmentCollector! + while let Ok(mut message) = reader.read_frame(&mut sender).await { + match std::str::from_utf8(message.payload.to_mut()) { + Ok(str) => { + match serde_jsonc::from_str::(&str) { + Ok(msg) => { + match handle_from_tx.send(msg) { + Ok(_) => { + // TODO: + } + Err(e) => { + // TODO: + } + } + } + Err(e) => { + // TODO: + } + } + } + Err(e) => { + // TODO: + } + } + } + }) + .await; + }); + + let send_tokens = Arc::new(handle_tokens.clone()); + let send_handle = spawn(async move { + let sender_writer = writer.clone(); + send_tokens + .0 + .run_until_cancelled(async move { + while let Some(message) = handle_to_rx.recv().await { + match sender_writer + .lock() + .await + .0 + .write_frame(Frame::text( + serde_jsonc::to_string(&message).unwrap().as_bytes().into(), + )) + .await + { + Ok(_) => {} + Err(e) => { + // TODO: + } + } + } + }) + .await; + match writer + .lock() + .await + .0 + .write_frame(Frame::close( + OpCode::Close as u16, + "Client Shutdown.".as_bytes(), + )) + .await + { + Ok(_) => { + // TODO: + } + Err(e) => { + // TODO: + } + } + }); + + tokio::select! {_ = futures::future::join_all(vec![ + recv_handle, + send_handle + ]) => { + info!("WS threads have exited!"); + handle_tokens.1.cancel(); + }} + }); + + self.to_server = Some(handle_to_tx); + self.from_server = Some(handle_from_rx); + + Ok((handle, cancellation_tokens)) + } + None => { + return Err(error.unwrap()); + } + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 5b47e0c..b46c1dd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,7 +2,9 @@ #![feature(let_chains)] pub mod arbiter; +pub mod client; pub mod server; +pub mod user; mod utils; pub fn add(left: u64, right: u64) -> u64 { diff --git a/src/server/listener.rs b/src/server/listener.rs index b0d6c45..831c3e2 100644 --- a/src/server/listener.rs +++ b/src/server/listener.rs @@ -7,10 +7,11 @@ use crate::server::models::{ ClientWithId, CoreUserConfig, IPCMessageWithId, + NexusStore, Session, - Store, }; use crate::server::websockets::handle_ws_client; +use crate::server::{AUTH_HEADER, DEAUTH_EVENT}; use crate::utils::{ gen_cid_with_check, gen_ipc_message, @@ -50,6 +51,8 @@ use warp::{ http::StatusCode, }; +use super::models::UserConfig; + // example error response #[derive(Serialize, Debug)] struct ApiErrorResult { @@ -108,7 +111,7 @@ async fn handle_rejection( // middleware that looks for authorization header and validates it async fn ensure_authentication( path: String, - store: Arc>, + store: Arc>, auth_header: Option, ) -> Result<(UserWithId, ApiKeyWithKey, ClientWithId, Session), warp::reject::Rejection> { let client_id = gen_cid_with_check(&store).await; @@ -261,8 +264,8 @@ pub struct ServerHealth { async fn handle_ipc_send( sender: Arc>, msg: IPCMessageWithId, - user_config: &Arc, - store: &Store, + user_config: &Arc, + store: &NexusStore, ) { let users_mutex = &store.users.to_owned(); let users = users_mutex.lock().await; @@ -273,9 +276,9 @@ async fn handle_ipc_send( let keys_mutex = &store.api_keys.to_owned(); let keys = keys_mutex.lock().await; - let api_key_conf = keys.get(&user_config.api_key.clone()).expect(&format!( + let api_key_conf = keys.get(&user_config.api_keys[0].key.clone()).expect(&format!( "ERROR: Core user api_key not found: {}", - user_config.api_key.clone() + user_config.api_keys[0].key.clone() )); let mut event_sent = false; @@ -323,66 +326,25 @@ async fn handle_ipc_send( pub async fn nexus_listener( port: u16, - store: Arc, - mut arbiter_ipc: ( - &CoreUserConfig, - UnboundedReceiver, - UnboundedSender, - ), - mut renderer_ipc: ( - &CoreUserConfig, - UnboundedReceiver, - UnboundedSender, - ), - mut modman_ipc: ( - &CoreUserConfig, - UnboundedReceiver, - UnboundedSender, - ), - mut inference_engine_ipc: ( - &CoreUserConfig, - UnboundedReceiver, + store: Arc, + mut internal_client_senders: Vec<( + &UserConfig, UnboundedSender, - ), - mut appd_ipc: ( - &CoreUserConfig, + )>, + internal_client_recivers: Vec<( + &UserConfig, UnboundedReceiver, - UnboundedSender, - ), - mut warehouse_ipc: ( - &CoreUserConfig, - UnboundedReceiver, - UnboundedSender, - ), + )>, cancellation_tokens: (CancellationToken, CancellationToken), - nexus_user_config: Arc, ) { info!("Starting nexus on port: {}...", port); let clients_tx: Arc>>> = Arc::new(Mutex::new(HashMap::new())); - let arbiter_cfg = arbiter_ipc.0; - let arbiter_tx = arbiter_ipc.2; - let renderer_cfg = renderer_ipc.0; - let renderer_tx = renderer_ipc.2; - let modman_cfg = modman_ipc.0; - let modman_tx = modman_ipc.2; - let inference_engine_cfg = inference_engine_ipc.0; - let inference_engine_tx = inference_engine_ipc.2; - let appd_cfg = appd_ipc.0; - let appd_tx = appd_ipc.2; - let warehouse_cfg = warehouse_ipc.0; - let warehouse_tx = warehouse_ipc.2; - for client in vec![ - (arbiter_cfg, arbiter_tx), - (renderer_cfg, renderer_tx), - (modman_cfg, modman_tx), - (inference_engine_cfg, inference_engine_tx), - (appd_cfg, appd_tx), - (warehouse_cfg, warehouse_tx), - ] { + // For everything connected via thread IPC, we still need a client or else everything freaks out. + for client in internal_client_senders { let client_obj = Client { - api_key: client.0.api_key.clone(), + api_key: client.0.api_keys[0].key.clone(), user_id: client.0.id.clone(), active: true, }; @@ -424,7 +386,7 @@ pub async fn nexus_listener( let ws_path = warp::path("ws") .and(warp::any().map(|| "/ws".to_string())) .and(store_filter.clone()) - .and(warp::header::optional::("Authorization")) + .and(warp::header::optional::(AUTH_HEADER)) .and_then(ensure_authentication) .and(warp::ws()) .and(store_filter.clone()) @@ -454,6 +416,7 @@ pub async fn nexus_listener( } }, ) + // TODO: Handle bind errors! .expect(""); server.await; info!("Server stopped."); @@ -529,7 +492,7 @@ pub async fn nexus_listener( None => { error!("DANGER! Client: {}, had API key removed from store without closing connection on removal, THIS IS BAD; please report this! Closing connection...", client_id.clone()); - let kind = Url::parse("clover://nexus.clover.reboot-codes.com/clients/unauthorize") + let kind = Url::parse(DEAUTH_EVENT) .unwrap() .query_pairs_mut() .append_pair("id", &client_id.clone()) @@ -566,108 +529,32 @@ pub async fn nexus_listener( } }); - // Internal IPC Handles - let from_warehouse_cfg = Arc::new(warehouse_cfg.clone()); - let from_warehouse_store = Arc::new(store.clone()); - let from_warehouse_tx = Arc::new(from_client_tx.clone()); - let from_warehouse_token = cancellation_tokens.0.clone(); - let from_warehouse_handle = tokio::task::spawn(async move { - tokio::select! { - _ = from_warehouse_token.cancelled() => { - debug!("from_warehouse exited"); - }, - _ = async move { - while let Some(msg) = warehouse_ipc.1.recv().await { - handle_ipc_send(from_warehouse_tx.clone(), msg, &from_warehouse_cfg.clone(), &from_warehouse_store.clone()).await; - } - } => {} - } - }); - - let from_arbiter_cfg = Arc::new(arbiter_cfg.clone()); - let from_arbiter_store = Arc::new(store.clone()); - let from_arbiter_tx = Arc::new(from_client_tx.clone()); - let from_arbiter_token = cancellation_tokens.0.clone(); - let from_arbiter_handle = tokio::task::spawn(async move { - tokio::select! { - _ = from_arbiter_token.cancelled() => { - debug!("from_arbiter exited"); - }, - _ = async move { - while let Some(msg) = arbiter_ipc.1.recv().await { - handle_ipc_send(from_arbiter_tx.clone(), msg, &from_arbiter_cfg.clone(), &from_arbiter_store.clone()).await; - } - } => {} - } - }); - - let from_renderer_cfg = Arc::new(renderer_cfg.clone()); - let from_renderer_store = Arc::new(store.clone()); - let from_renderer_tx = Arc::new(from_client_tx.clone()); - let from_renderer_token = cancellation_tokens.0.clone(); - let from_renderer_handle = tokio::task::spawn(async move { - tokio::select! { - _ = from_renderer_token.cancelled() => { - debug!("from_renderer exited"); - }, - _ = async move { - while let Some(msg) = renderer_ipc.1.recv().await { - handle_ipc_send(from_renderer_tx.clone(), msg, &from_renderer_cfg.clone(), &from_renderer_store.clone()).await; - } - } => {} - } - }); - - let from_modman_cfg = Arc::new(modman_cfg.clone()); - let from_modman_store = Arc::new(store.clone()); - let from_modman_tx = Arc::new(from_client_tx.clone()); - let from_modman_token = cancellation_tokens.0.clone(); - let from_modman_handle = tokio::task::spawn(async move { - tokio::select! { - _ = from_modman_token.cancelled() => { - debug!("from_modman exited"); - }, - _ = async move { - while let Some(msg) = modman_ipc.1.recv().await { - handle_ipc_send(from_modman_tx.clone(), msg, &from_modman_cfg.clone(), &from_modman_store.clone()).await; - } - } => {} - } - }); - - let from_inference_engine_cfg = Arc::new(inference_engine_cfg.clone()); - let from_inference_engine_store = Arc::new(store.clone()); - let from_inference_engine_tx = Arc::new(from_client_tx.clone()); - let from_inference_engine_token = cancellation_tokens.0.clone(); - let from_inference_engine_handle = tokio::task::spawn(async move { - tokio::select! { - _ = from_inference_engine_token.cancelled() => { - debug!("from_reference_engine exited"); - }, - _ = async move { - while let Some(msg) = inference_engine_ipc.1.recv().await { - handle_ipc_send(from_inference_engine_tx.clone(), msg, &from_inference_engine_cfg.clone(), &from_inference_engine_store.clone()).await; - } - } => {} - } - }); + let mut internal_ipc_handles = vec![]; + + for client in internal_client_recivers { + // Internal IPC Handles + let client_cfg = Arc::new(client.0.clone()); + let internal_ipc_store = Arc::new(store.clone()); + let internal_ipc_tx = Arc::new(from_client_tx.clone()); + let internal_ipc_token = cancellation_tokens.0.clone(); + internal_ipc_handles.push(tokio::task::spawn(async move { + tokio::select! { + _ = internal_ipc_token.cancelled() => { + debug!("Internal IPC handle for client: \"{}\", exited!", client_cfg.id.clone()); + }, + _ = async move { + while let Some(msg) = client.1.recv().await { + handle_ipc_send(internal_ipc_tx.clone(), msg, &client_cfg.clone(), &internal_ipc_store.clone()).await; + } + } => {} + } + })); + } - let from_appd_cfg = Arc::new(appd_cfg.clone()); - let from_appd_store = Arc::new(store.clone()); - let from_appd_tx = Arc::new(from_client_tx.clone()); - let from_appd_token = cancellation_tokens.0.clone(); - let from_appd_handle = tokio::task::spawn(async move { - tokio::select! { - _ = from_appd_token.cancelled() => { - debug!("from_appd exited"); - }, - _ = async move { - while let Some(msg) = appd_ipc.1.recv().await { - handle_ipc_send(from_appd_tx.clone(), msg, &from_appd_cfg.clone(), &from_appd_store.clone()).await; - } - } => {} - } - }); + let mut handles = Vec::new(); + handles.push(http_handle); + handles.push(ipc_dispatch_handle); + handles.append(&mut internal_ipc_handles); let cleanup_token = cancellation_tokens.0.clone(); tokio::select! { @@ -681,16 +568,8 @@ pub async fn nexus_listener( } } - tokio::select! {_ = futures::future::join_all(vec![ - from_warehouse_handle, - http_handle, - ipc_dispatch_handle, - from_arbiter_handle, - from_renderer_handle, - from_modman_handle, - from_inference_engine_handle, - from_appd_handle - ]) => { - info!("nexus has stopped!"); + tokio::select! { + _ = futures::future::join_all(handles) => { + info!("Nexus server has stopped!"); }} } diff --git a/src/server/mod.rs b/src/server/mod.rs index b777955..fa464d5 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -3,3 +3,6 @@ pub mod models; pub mod websockets; // TODO: Move all events to ones with data that has a set serde model! + +pub const DEAUTH_EVENT: &str = "nexus://com.reboot-codes.nexus/websockets/deauthorize"; +pub const AUTH_HEADER: &str = "Authorization"; diff --git a/src/server/models.rs b/src/server/models.rs index 65fe4da..a57c3b4 100644 --- a/src/server/models.rs +++ b/src/server/models.rs @@ -96,45 +96,31 @@ pub struct UserConfig { // TODO: Add serialization/deserialization functions... // TODO: Add options for making certain models ephemeral or persistent. #[derive(Debug, Clone)] -pub struct Store { +pub struct NexusStore { pub users: Arc>>, pub api_keys: Arc>>, pub clients: Arc>>, pub messages: Arc>>, } -#[derive(Debug, Clone)] -pub struct CoreUserConfigs { - pub evtbuzz: CoreUserConfig, - pub arbiter: CoreUserConfig, - pub renderer: CoreUserConfig, - pub appd: CoreUserConfig, - pub modman: CoreUserConfig, - pub inference_engine: CoreUserConfig, - pub warehouse: CoreUserConfig, -} +impl NexusStore { + const MASTER_USER_TYPE: &str = "com.reboot-codes.nexus.master"; -impl Store { - pub fn new() -> Self { - Store { + /// Create a new store with a set master user. + pub async fn new(master_user_pretty_name: &String) -> (NexusStore, UserConfig) { + let mut ret = NexusStore { users: Arc::new(Mutex::new(HashMap::new())), api_keys: Arc::new(Mutex::new(HashMap::new())), clients: Arc::new(Mutex::new(HashMap::new())), messages: Arc::new(Mutex::new(HashMap::new())), - } - } + }; - // Create a new store with a set master user. - pub async fn new_configured_store() -> (Store, CoreUserConfig, CoreUserConfigs) { - let ret = Store::new(); + let master_user_config = ret.add_master_user(&master_user_pretty_name).await; - let master_user_config = ret.clone().create_master_user().await; - let core_users_config = ret.clone().add_all_core_users().await; - - (ret, master_user_config, core_users_config) + (ret, master_user_config) } - pub async fn add_user(self, user_config: UserConfig) { + pub async fn add_user(&mut self, user_config: UserConfig) { let mut key_ids: Vec = vec![]; for key_config in user_config.api_keys.iter() { key_ids.push(key_config.key.clone()); @@ -163,187 +149,21 @@ impl Store { } } - pub async fn create_master_user(self) -> CoreUserConfig { - let master_user_id = gen_uid_with_check(&self).await; - let master_api_key = gen_api_key_with_check(&self).await; - - self - .add_user(UserConfig { - id: master_user_id.clone(), - pretty_name: "Master User".to_string(), - user_type: "com.reboot-codes.clover.master".to_string(), - api_keys: vec![ApiKeyWithKeyWithoutUID { - allowed_events_to: vec![".*".to_string()], - allowed_events_from: vec![".*".to_string()], - key: master_api_key.clone(), - echo: true, - }], - }) - .await; - - CoreUserConfig { - id: master_user_id.clone(), - api_key: master_api_key.clone(), - } - } - - /// Adds all the core user accounts, returns their configurations. - pub async fn add_all_core_users(self) -> CoreUserConfigs { - // EvtBuzz - let evtbuzz_uid = gen_uid_with_check(&self).await; - let evtbuzz_key = gen_api_key_with_check(&self).await; - self - .clone() - .add_user(UserConfig { - user_type: "com.reboot-codes.clover.evtbuzz".to_string(), - pretty_name: "EvtBuzz".to_string(), - id: evtbuzz_uid.clone(), - api_keys: vec![ApiKeyWithKeyWithoutUID { - allowed_events_to: vec![".*".to_string()], - allowed_events_from: vec![".*".to_string()], - key: evtbuzz_key.clone(), - echo: true, - }], - }) - .await; - - // Arbiter - let arbiter_uid = gen_uid_with_check(&self).await; - let arbiter_key = gen_api_key_with_check(&self).await; - self - .clone() - .add_user(UserConfig { - user_type: "com.reboot-codes.clover.arbiter".to_string(), - pretty_name: "Arbiter".to_string(), - id: arbiter_uid.clone(), - api_keys: vec![ApiKeyWithKeyWithoutUID { - allowed_events_to: vec![".*".to_string()], - allowed_events_from: vec![".*".to_string()], - key: arbiter_key.clone(), - echo: true, - }], - }) - .await; - - // Renderer - let renderer_uid = gen_uid_with_check(&self).await; - let renderer_key = gen_api_key_with_check(&self).await; - self - .clone() - .add_user(UserConfig { - user_type: "com.reboot-codes.clover.renderer".to_string(), - pretty_name: "Renderer".to_string(), - id: renderer_uid.clone(), - api_keys: vec![ApiKeyWithKeyWithoutUID { - allowed_events_to: vec![".*".to_string()], - allowed_events_from: vec![".*".to_string()], - key: renderer_key.clone(), - echo: true, - }], - }) - .await; - - // AppD - let appd_uid = gen_uid_with_check(&self).await; - let appd_key = gen_api_key_with_check(&self).await; - self - .clone() - .add_user(UserConfig { - user_type: "com.reboot-codes.clover.appd".to_string(), - pretty_name: "appd".to_string(), - id: appd_uid.clone(), - api_keys: vec![ApiKeyWithKeyWithoutUID { - allowed_events_to: vec![".*".to_string()], - allowed_events_from: vec![".*".to_string()], - key: appd_key.clone(), - echo: true, - }], - }) - .await; - - // ModMan - let modman_uid = gen_uid_with_check(&self).await; - let modman_key = gen_api_key_with_check(&self).await; - self - .clone() - .add_user(UserConfig { - user_type: "com.reboot-codes.clover.modman".to_string(), - pretty_name: "ModMan".to_string(), - id: modman_uid.clone(), - api_keys: vec![ApiKeyWithKeyWithoutUID { - allowed_events_to: vec![".*".to_string()], - allowed_events_from: vec![".*".to_string()], - key: modman_key.clone(), - echo: true, - }], - }) - .await; - - // Inference Engine - let inference_engine_uid = gen_uid_with_check(&self).await; - let inference_engine_key = gen_api_key_with_check(&self).await; - self - .clone() - .add_user(UserConfig { - user_type: "com.reboot-codes.clover.inference-engine".to_string(), - pretty_name: "Inference Engine".to_string(), - id: inference_engine_uid.clone(), - api_keys: vec![ApiKeyWithKeyWithoutUID { - allowed_events_to: vec![".*".to_string()], - allowed_events_from: vec![".*".to_string()], - key: inference_engine_key.clone(), - echo: true, - }], - }) - .await; - - // Warehouse - let warehouse_uid = gen_uid_with_check(&self).await; - let warehouse_key = gen_api_key_with_check(&self).await; - self - .clone() - .add_user(UserConfig { - user_type: "com.reboot-codes.clover.warehouse".to_string(), - pretty_name: "Warehouse".to_string(), - id: warehouse_uid.clone(), - api_keys: vec![ApiKeyWithKeyWithoutUID { - allowed_events_to: vec![".*".to_string()], - allowed_events_from: vec![".*".to_string()], - key: warehouse_key.clone(), - echo: true, - }], - }) - .await; - - CoreUserConfigs { - evtbuzz: CoreUserConfig { - id: evtbuzz_uid.clone(), - api_key: evtbuzz_key.clone(), - }, - arbiter: CoreUserConfig { - id: arbiter_uid.clone(), - api_key: arbiter_key.clone(), - }, - renderer: CoreUserConfig { - id: renderer_uid.clone(), - api_key: renderer_key.clone(), - }, - appd: CoreUserConfig { - id: appd_uid.clone(), - api_key: appd_key.clone(), - }, - modman: CoreUserConfig { - id: modman_uid.clone(), - api_key: modman_key.clone(), - }, - inference_engine: CoreUserConfig { - id: inference_engine_uid.clone(), - api_key: inference_engine_key.clone(), - }, - warehouse: CoreUserConfig { - id: warehouse_uid.clone(), - api_key: warehouse_key.clone(), - }, - } + pub async fn add_master_user(&mut self, pretty_name: &String) -> UserConfig { + let ret = UserConfig { + id: gen_uid_with_check(self).await, + pretty_name: pretty_name.clone(), + user_type: NexusStore::MASTER_USER_TYPE.to_string(), + api_keys: vec![ApiKeyWithKeyWithoutUID { + allowed_events_to: vec![".*".to_string()], + allowed_events_from: vec![".*".to_string()], + key: gen_api_key_with_check(self).await, + echo: true, + }], + }; + + self.add_user(ret.clone()).await; + + ret } } diff --git a/src/server/websockets.rs b/src/server/websockets.rs index f2b1dbe..3076ee0 100644 --- a/src/server/websockets.rs +++ b/src/server/websockets.rs @@ -6,9 +6,10 @@ use crate::server::models::{ Client, ClientWithId, IPCMessageWithId, + NexusStore, Session, - Store, }; +use crate::server::DEAUTH_EVENT; use crate::utils::iso8601; use futures::{ SinkExt, @@ -44,12 +45,13 @@ use warp::filters::ws::{ pub struct WsIn { pub kind: String, pub message: String, + pub api_key: Option } pub async fn handle_ws_client( auth: (UserWithId, ApiKeyWithKey, ClientWithId, Session), ws: warp::ws::Ws, - store: Arc>, + store: Arc>, to_clients_tx: Arc>>>>, from_clients_tx: Arc>, ) -> Result { @@ -111,22 +113,66 @@ pub async fn handle_ws_client( debug!("Client: {}, send message: {{ \"id\": \"{}\", \"kind\": \"{}\", \"message\": \"{}\" }}...", ws_client.id.clone(), message_id.clone(), msg.kind.clone(), msg.message.clone()); let mut allowed_to_send = false; + + // Check if this message can be sent by this API key. for allowed_send_pattern in recv_api_key.allowed_events_from.clone() { match Regex::new(&allowed_send_pattern) { Ok(pattern) => { if pattern.is_match(&msg.kind.clone()) { - allowed_to_send = true; - break; + // Check if this message is being proxied via nexus::client... + match msg.api_key.clone() { + Some(proxied_key) => { + // If so, make sure to run through the API key verification process for this message independantly. + if recv_api_key.proxy { + match recv_store.api_keys.lock().await.get(&proxied_key.clone()) { + Some(proxied_api_key) => { + match recv_store.users.lock().await.get(&proxied_api_key.user_id.clone()) { + Some(_proxied_user) => { + for proxied_allowed_send_pattern in proxied_api_key.allowed_events_from.clone() { + match Regex::new(&proxied_allowed_send_pattern) { + Ok(proxied_pattern) => { + if proxied_pattern.is_match(&msg.kind.clone()) { + allowed_to_send = true; + break; + } + }, + Err(e) => { + warn!("Allowed send from pattern: \"{}\" (for user: \"{}\"), is not valid, due to:\n{}", proxied_allowed_send_pattern.clone(), proxied_api_key.user_id.clone(), e); + } + } + } + }, + None => { + // TODO: Proxied API key exists in store, but the associated User does not exist, refusing message. + // *This is a critical bug!* + // Theoretically, this shouldn't happen, but it could, and any cases should be refused and reported to the bug tracker. + } + } + }, + None => { + // TODO: Proxied API key does not exist in the store, discard message. + } + } + } else { + // TODO: The *session's* api key is NOT permitted to proxy messages, and the message will be discarded. + } + }, + None => { + // Or just send the message since no proxying is occuring. + allowed_to_send = true; + break; + } + } } }, Err(e) => { - warn!("Allowed send from pattern: \"{}\" (for user: \"{}\"), is not valid, due to:\n{}", allowed_send_pattern.clone(), recv_user.id.clone(), e); + warn!("Allowed \"send from\" pattern: \"{}\" (for user: \"{}\"), is not valid, due to:\n{}", allowed_send_pattern.clone(), recv_user.id.clone(), e); } } } if allowed_to_send { - let generated_message = IPCMessageWithId { id: message_id.clone(), author: format!("ws:{}?client={}", recv_api_key.user_id.clone(), ws_client.id.clone()), kind: msg.kind.clone(), message: msg.message.clone() }; + let generated_message = IPCMessageWithId { id: message_id.clone(), author: format!("ws://{}?client={}", recv_api_key.user_id.clone(), ws_client.id.clone()), kind: msg.kind.clone(), message: msg.message.clone() }; recv_store.messages.lock().await.insert(message_id.clone(), generated_message.clone().into()); @@ -144,13 +190,13 @@ pub async fn handle_ws_client( } }, Err(e) => { - warn!("Client: {}, error reading message: \"{}\", due to:\n {}", recv_client.id.clone(), message, e); + warn!("Client: {}, error reading message: \"{}\", due to:\n{}", recv_client.id.clone(), message, e); // TODO: Send err? } }; }, Err(e) => { - error!("Error reading message on client: \"{}\", {}", ws_client.id.clone(), e); + error!("Error reading message on client: \"{}\", due to:\n{}", ws_client.id.clone(), e); // TODO: send error? } }; @@ -165,7 +211,7 @@ pub async fn handle_ws_client( let send_client = Arc::new(client.clone()); let send_handle = tokio::task::spawn(async move { while let Some(msg) = to_client_rx.recv().await { - if msg.kind == Url::parse("clover://evtbuzz.clover.reboot-codes.com/clients/unauthorize") + if msg.kind == Url::parse(DEAUTH_EVENT) .unwrap() .query_pairs_mut() .append_pair("id", send_client.id.clone().as_str()) diff --git a/src/user/mod.rs b/src/user/mod.rs new file mode 100644 index 0000000..113e6c1 --- /dev/null +++ b/src/user/mod.rs @@ -0,0 +1,6 @@ +/// Use in a thread to connect to send messages to/recieve messages from nexus with. +/// Requires a [NexusClient](nexus::client::NexusClient) or a tokio IPC channel +/// connected to an embedded Nexus server to actually send messages. +/// +/// This struct basically just ensures that messages are formatted properly. +pub struct NexusUser {} diff --git a/src/utils.rs b/src/utils.rs index 5b980c7..d8fc81a 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,7 +1,7 @@ use crate::server::models::{ CoreUserConfig, IPCMessageWithId, - Store, + NexusStore, }; use api_key::types::{ ApiKeyResults, @@ -62,7 +62,7 @@ pub fn gen_api_key() -> String { } /// Generates a new API key after checking that it is not currently in the Store. -pub async fn gen_api_key_with_check(store: &Store) -> String { +pub async fn gen_api_key_with_check(store: &NexusStore) -> String { loop { let api_key = gen_api_key(); match store.api_keys.lock().await.get(&api_key.clone()) { @@ -75,7 +75,7 @@ pub async fn gen_api_key_with_check(store: &Store) -> String { } /// Generates a new UID after checking that it is currently not in the Store. -pub async fn gen_uid_with_check(store: &Store) -> String { +pub async fn gen_uid_with_check(store: &NexusStore) -> String { loop { let uid = Uuid::new_v4().to_string(); match store.users.lock().await.get(&uid.clone()) { @@ -87,7 +87,7 @@ pub async fn gen_uid_with_check(store: &Store) -> String { } } -pub async fn gen_message_id_with_check(store: &Store) -> String { +pub async fn gen_message_id_with_check(store: &NexusStore) -> String { loop { let message_id = Uuid::new_v4().to_string(); match store.messages.lock().await.get(&message_id.clone()) { @@ -100,7 +100,7 @@ pub async fn gen_message_id_with_check(store: &Store) -> String { } pub async fn gen_ipc_message( - store: &Store, + store: &NexusStore, user_config: &CoreUserConfig, kind: String, message: String, @@ -114,7 +114,7 @@ pub async fn gen_ipc_message( } } -pub async fn gen_cid_with_check(store: &Store) -> String { +pub async fn gen_cid_with_check(store: &NexusStore) -> String { loop { let client_id = Uuid::new_v4().to_string(); match store.clients.lock().await.get(&client_id.clone()) { @@ -127,7 +127,7 @@ pub async fn gen_cid_with_check(store: &Store) -> String { } pub async fn send_ipc_message( - store: &Store, + store: &NexusStore, user_config: &CoreUserConfig, ipc_tx: Arc>, kind: String,