diff --git a/Cargo.lock b/Cargo.lock index 786111f..de65a25 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -204,6 +204,20 @@ dependencies = [ "syn 2.0.96", ] +[[package]] +name = "actix-ws" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3a1fb4f9f2794b0aadaf2ba5f14a6f034c7e86957b458c506a8cb75953f2d99" +dependencies = [ + "actix-codec", + "actix-http", + "actix-web", + "bytestring", + "futures-core", + "tokio", +] + [[package]] name = "addr2line" version = "0.24.2" @@ -428,6 +442,22 @@ dependencies = [ "syn 2.0.96", ] +[[package]] +name = "async-tungstenite" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee88b4c88ac8c9ea446ad43498955750a4bbe64c4392f21ccfe5d952865e318f" +dependencies = [ + "atomic-waker", + "futures-core", + "futures-io", + "futures-task", + "futures-util", + "log", + "pin-project-lite", + "tungstenite", +] + [[package]] name = "atoi" version = "2.0.0" @@ -452,6 +482,12 @@ dependencies = [ "bytemuck", ] +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + [[package]] name = "atomic_float" version = "1.1.0" @@ -951,6 +987,12 @@ dependencies = [ "syn 2.0.96", ] +[[package]] +name = "data-encoding" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476" + [[package]] name = "deflate64" version = "0.1.9" @@ -1247,6 +1289,7 @@ checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876" dependencies = [ "futures-channel", "futures-core", + "futures-executor", "futures-io", "futures-sink", "futures-task", @@ -1297,6 +1340,17 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" +[[package]] +name = "futures-macro" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.96", +] + [[package]] name = "futures-sink" version = "0.3.31" @@ -1318,6 +1372,7 @@ dependencies = [ "futures-channel", "futures-core", "futures-io", + "futures-macro", "futures-sink", "futures-task", "memchr", @@ -2495,8 +2550,10 @@ version = "0.5.0" dependencies = [ "actix-files", "actix-web", + "actix-ws", "clap", "env_logger", + "futures", "git-version", "handlebars", "hex", @@ -2548,6 +2605,7 @@ dependencies = [ "async-trait", "base64 0.22.1", "chrono", + "futures", "hex", "jsonwebtoken", "libfj", @@ -2559,6 +2617,7 @@ dependencies = [ "polariton_server", "rand 0.9.0", "reqwest", + "reqwest-websocket", "serde", "serde_json", "sha2", @@ -2670,6 +2729,7 @@ dependencies = [ "chrono", "clap", "env_logger", + "futures", "hex", "libfj", "log", @@ -3396,6 +3456,26 @@ dependencies = [ "windows-registry", ] +[[package]] +name = "reqwest-websocket" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd5f79b25f7f17a62cc9337108974431a66ae5a723ac0d9fe78ac1cce2027720" +dependencies = [ + "async-tungstenite", + "bytes", + "futures-util", + "reqwest", + "serde", + "serde_json", + "thiserror 2.0.11", + "tokio", + "tokio-util", + "tracing", + "tungstenite", + "web-sys", +] + [[package]] name = "ring" version = "0.17.8" @@ -4565,6 +4645,7 @@ checksum = "d7fcaa8d55a2bdd6b83ace262b016eca0d79ee02818c5c1bcdf0305114081078" dependencies = [ "bytes", "futures-core", + "futures-io", "futures-sink", "pin-project-lite", "tokio", @@ -4710,6 +4791,23 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eadc29d668c91fcc564941132e17b28a7ceb2f3ebf0b9dae3e03fd7a6748eb0d" +dependencies = [ + "bytes", + "data-encoding", + "http 1.2.0", + "httparse", + "log", + "rand 0.9.0", + "sha1", + "thiserror 2.0.11", + "utf-8", +] + [[package]] name = "typenum" version = "1.18.0" @@ -4815,6 +4913,12 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf16_iter" version = "1.0.5" diff --git a/Cargo.toml b/Cargo.toml index a6346e8..bc9e7f9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,6 +42,7 @@ polariton_server = { version = "0.4", features = [ "tokio-async" ] } serde = { version = "1.0", features = [ "derive" ] } serde_json = "1.0" async-trait = "0.1" +futures = "0.3" chrono = "0.4" git-version = "0.3" rand = { version = "0.9", features = [ "thread_rng" ] } diff --git a/assets/robocraft/config.json b/assets/robocraft/config.json index 3ae87fc..707d3f3 100644 --- a/assets/robocraft/config.json +++ b/assets/robocraft/config.json @@ -14470,6 +14470,14 @@ "built_in": "Version" } }, + { + "regex": "\\?banner", + "op": { + "type": "BuiltIn", + "built_in": "Intercom", + "intercom": "DevMessage" + } + }, { "regex": "\\?help", "op": { diff --git a/cdn/src/robocraft/internal_auth.rs b/cdn/src/robocraft/internal_auth.rs index f0e6518..3cc5d4e 100644 --- a/cdn/src/robocraft/internal_auth.rs +++ b/cdn/src/robocraft/internal_auth.rs @@ -26,16 +26,16 @@ impl IntercomAuth { if bearer.to_lowercase() == "internal" || bearer.to_lowercase() == "bearer" { self.validate_token(token, salt)?; } else { - return Err(super::IntercomOpError::BadHeader); + return Err(IntercomOpError::Unauthorized); } } else { - return Err(super::IntercomOpError::BadHeader); + return Err(IntercomOpError::BadHeader); } } else { - return Err(super::IntercomOpError::BadHeader); + return Err(IntercomOpError::BadHeader); } } else { - return Err(super::IntercomOpError::Unauthorized); + return Err(IntercomOpError::Unauthorized); } Ok(()) } diff --git a/rc_auth/Cargo.toml b/rc_auth/Cargo.toml index f808054..f825f3f 100644 --- a/rc_auth/Cargo.toml +++ b/rc_auth/Cargo.toml @@ -10,9 +10,11 @@ readme.workspace = true [dependencies] actix-web.workspace = true actix-files.workspace = true +actix-ws = "0.3" log.workspace = true env_logger.workspace = true tokio = { version = "1.43", features = [ "rt-multi-thread" ] } +futures.workspace = true clap.workspace = true oj_rc_core = { version = "*", path = "../rc_core" } libfj.workspace = true diff --git a/rc_auth/src/main.rs b/rc_auth/src/main.rs index c0c7b05..801e977 100644 --- a/rc_auth/src/main.rs +++ b/rc_auth/src/main.rs @@ -20,6 +20,8 @@ async fn main() -> std::io::Result<()> { let cli_args = cli::CliArgs::get(); let cli_args2 = actix_web::web::Data::new(cli_args.clone()); let rc_preloaded = actix_web::web::Data::new(cli_args.clone().preloaded().await); + let internal_auth = actix_web::web::Data::new(crate::robocraft::intercom::IntercomAuth::new(&cli_args.data_robocraft)?); + let user_registry = actix_web::web::Data::new(crate::robocraft::intercom::Users::new()); let mut handlebars = handlebars::Handlebars::new(); handlebars @@ -38,6 +40,8 @@ async fn main() -> std::io::Result<()> { App::new() .app_data(cli_args2.clone()) .app_data(rc_preloaded.clone()) + .app_data(internal_auth.clone()) + .app_data(user_registry.clone()) .app_data(handlebars_ref.clone()) .service(index) .service(robocraft::registration::form_submit) @@ -46,13 +50,8 @@ async fn main() -> std::io::Result<()> { .service(robocraft::email::email_password_auth) .service(robocraft::steam::steam_auth) .service(robocraft::username::user_password_auth) - /*.service(robocraft::live_data::live_data_json) - .service(robocraft::user_avatar::get) - .service(robocraft::clan_avatar::get) - .service(robocraft::brawl_data::get) - .service(robocraft::campaign_data::get) - .service(robocraft::factory::arc::get) - .service(robocraft::favicon::get)*/ + .service(robocraft::intercom::services_ws) + .service(robocraft::intercom::service_msg) }) .bind((cli_args.ip, cli_args.port))? .run() diff --git a/rc_auth/src/robocraft/intercom/internal_auth.rs b/rc_auth/src/robocraft/intercom/internal_auth.rs new file mode 100644 index 0000000..37a7682 --- /dev/null +++ b/rc_auth/src/robocraft/intercom/internal_auth.rs @@ -0,0 +1,83 @@ +pub struct IntercomAuth { + key: Vec, +} + +impl IntercomAuth { + pub fn new(data: impl AsRef) -> std::io::Result { + let key = std::fs::read(data.as_ref().join(oj_rc_core::persist::user::TOKEN_SECRET_FILENAME))?; + Ok(Self { + key, + }) + } + + fn validate_token(&self, received_token: &str, salt: &str) -> Result<(), IntercomOpError> { + let expected_token = oj_rc_core::persist::user::generate_intercom_token(salt.as_bytes(), &self.key); + if received_token.to_lowercase() == expected_token.to_lowercase() { + Ok(()) + } else { + Err(IntercomOpError::Unauthorized) + } + } + + pub fn validate(&self, req: &actix_web::HttpRequest, salt: &str) -> Result<(), IntercomOpError> { + if let Some(auth_header) = req.headers().get("Authorization") { + if let Ok(header_val) = auth_header.to_str() { + if let Some((bearer, token)) = header_val.split_once(" ") { + if bearer.to_lowercase() == "internal" || bearer.to_lowercase() == "bearer" { + self.validate_token(token, salt)?; + } else { + return Err(IntercomOpError::Unauthorized); + } + } else { + return Err(IntercomOpError::BadHeader); + } + } else { + return Err(IntercomOpError::BadHeader); + } + } else { + return Err(IntercomOpError::Unauthorized); + } + Ok(()) + } +} + +#[derive(Debug)] +pub enum IntercomOpError { + BadHeader, + Unauthorized, + #[allow(dead_code)] + Io(std::io::Error), + #[allow(dead_code)] + Unknown, +} + +impl core::fmt::Display for IntercomOpError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "Intercom error variant {:?}", self) + } +} + +impl actix_web::error::ResponseError for IntercomOpError { + fn status_code(&self) -> actix_web::http::StatusCode { + match self { + Self::BadHeader => actix_web::http::StatusCode::BAD_REQUEST, + Self::Unauthorized => actix_web::http::StatusCode::FORBIDDEN, + Self::Io(_) => actix_web::http::StatusCode::INSUFFICIENT_STORAGE, + Self::Unknown => actix_web::http::StatusCode::INTERNAL_SERVER_ERROR, + } + } + + fn error_response(&self) -> actix_web::HttpResponse { + match self { + Self::Io(io_e) => { + actix_web::HttpResponse::new(self.status_code()) + .set_body(format!("Intercom IO error: {}", io_e)) + .map_into_boxed_body() + }, + _ => { + actix_web::HttpResponse::new(self.status_code()).set_body(self.to_string()).map_into_boxed_body() + } + } + + } +} diff --git a/rc_auth/src/robocraft/intercom/mod.rs b/rc_auth/src/robocraft/intercom/mod.rs new file mode 100644 index 0000000..0603417 --- /dev/null +++ b/rc_auth/src/robocraft/intercom/mod.rs @@ -0,0 +1,8 @@ +mod internal_auth; +pub use internal_auth::{IntercomAuth, IntercomOpError}; + +mod services; +pub use services::{services_ws, service_msg}; + +mod user_registry; +pub use user_registry::Users; diff --git a/rc_auth/src/robocraft/intercom/services.rs b/rc_auth/src/robocraft/intercom/services.rs new file mode 100644 index 0000000..bd5f515 --- /dev/null +++ b/rc_auth/src/robocraft/intercom/services.rs @@ -0,0 +1,47 @@ +use actix_web::{rt, web::{Payload, Data, Path, Json}, Error, HttpRequest, HttpResponse, get, post}; +//use actix_ws::AggregatedMessage; +//use futures::StreamExt as _; + +#[get("/intercom/.oj_services/{name}")] +pub async fn services_ws(req: HttpRequest, stream: Payload, auth: Data, reg: Data, name: Path) -> Result { + auth.validate(&req, &format!(".oj_services/{}", name))?; + let (res, mut session, _stream) = actix_ws::handle(&req, stream)?; + + /*let mut stream = stream + .aggregate_continuations() + .max_continuation_size(2_usize.pow(20)); // aggregate continuation frames up to 1MiB + */ + + let (tx, mut rx) = tokio::sync::mpsc::channel(16); + reg.register_service(name.clone(), tx).await; + log::debug!("Registered web services intercom websocket for user {}", name); + + // start task but don't wait for it + rt::spawn(async move { + while let Some(msg) = rx.recv().await { + if let Err(e) = session.text(serde_json::to_string(&msg).unwrap()).await { + log::warn!("Failed to send services intercom to user {}: {}", name, e); + break; + } + } + reg.remove_service(name.clone()).await; + rx.close(); + session.close(Some(actix_ws::CloseReason { + code: actix_ws::CloseCode::Normal, + description: Some("End of channel".to_owned()), + })).await.expect("Failed to close a services intercom websocket session"); + log::debug!("Web services intercom websocket closed for {}", name); + }); + + // respond immediately with response connected to WS session + Ok(res) +} + +#[post("/intercom/.oj_services/{name}/messages")] +pub async fn service_msg(req: HttpRequest, body: Json, auth: Data, reg: Data, name: Path) -> Result { + log::debug!("Got intercom message from {} to {:?}", name, body.public_ids.as_slice()); + auth.validate(&req, &format!(".oj_services/{}/messages", name))?; + log::debug!("Authenticated intercom message from {} to {:?}", name, body.public_ids.as_slice()); + reg.broadcast_service_message(body.0).await; + Ok(HttpResponse::NoContent().finish()) +} diff --git a/rc_auth/src/robocraft/intercom/user_registry.rs b/rc_auth/src/robocraft/intercom/user_registry.rs new file mode 100644 index 0000000..0bbb14d --- /dev/null +++ b/rc_auth/src/robocraft/intercom/user_registry.rs @@ -0,0 +1,42 @@ +use oj_rc_core::persist::user::intercom::IntercomWebServiceUserMessage; + +pub struct Users { + service_listeners: tokio::sync::RwLock>>, +} + +impl Users { + pub fn new() -> Self { + Self { + service_listeners: tokio::sync::RwLock::new(std::collections::HashMap::with_capacity(16)), + } + } + + pub async fn register_service(&self, public_id: String, sender: tokio::sync::mpsc::Sender) { + let mut write_lock = self.service_listeners.write().await; + if let Some(old_sender) = write_lock.insert(public_id.clone(), sender) { + if !old_sender.is_closed() { + log::warn!("Replaced web services intercom channel for user {} (why duplicate!?)", public_id); + } + } + } + + pub async fn remove_service(&self, public_id: String) { + let mut write_lock = self.service_listeners.write().await; + if write_lock.remove(&public_id).is_none() { + log::warn!("Tried to remove web services intercom channel for user {} without listener", public_id); + } + } + + pub async fn broadcast_service_message(&self, msg: oj_rc_core::persist::user::intercom::IntercomWebServiceMessage) { + let read_lock = self.service_listeners.read().await; + for public_id in msg.public_ids { + if let Some(tx) = read_lock.get(&public_id) { + if let Err(e) = tx.send(msg.data.clone()).await { + log::error!("Failed to send web service intercom message to {}: {}", public_id, e); + } + } else { + log::warn!("Not sending web service intercom message to user {}; no listener found", public_id); + } + } + } +} diff --git a/rc_auth/src/robocraft/mod.rs b/rc_auth/src/robocraft/mod.rs index fe6d800..1ae78cb 100644 --- a/rc_auth/src/robocraft/mod.rs +++ b/rc_auth/src/robocraft/mod.rs @@ -2,6 +2,7 @@ pub mod email; pub mod registration; pub mod steam; pub mod username; +pub mod intercom; pub struct RcConfig { //pub data: std::path::PathBuf, diff --git a/rc_chat_room/src/state/chat/config.rs b/rc_chat_room/src/state/chat/config.rs index 58b4df1..5717ba7 100644 --- a/rc_chat_room/src/state/chat/config.rs +++ b/rc_chat_room/src/state/chat/config.rs @@ -67,7 +67,7 @@ impl ChatCommand { async fn perform_if_match<'b, 'c>(&self, text: &str, ctx: CommandContext<'b, 'c>) -> Option { if let Some(cap) = self.regex.captures(text) { - Some(self.op.perform_command(cap, ctx).await) + Some(self.op.perform_command(text, cap, ctx).await) } else { None } @@ -89,9 +89,9 @@ impl ChatOperation { } } - async fn perform_command<'a, 'b, 'c>(&self, _captures: regex::Captures<'a>, ctx: CommandContext<'b, 'c>) -> String { + async fn perform_command<'a, 'b, 'c>(&self, text: &str, _captures: regex::Captures<'a>, ctx: CommandContext<'b, 'c>) -> String { match self { - Self::BuiltIn(b_in) => b_in.do_command(ctx).await, + Self::BuiltIn(b_in) => b_in.do_command(text, ctx).await, Self::Custom => "{not implemented}".to_owned(), Self::Nop => "{no op}".to_owned(), } @@ -107,6 +107,7 @@ impl ChatOperation { } enum BuiltIn { + Intercom(Intercom), OnlineUsers, TotalUsers, Version, @@ -116,6 +117,7 @@ enum BuiltIn { impl BuiltIn { fn from_persist(b_in: oj_rc_core::persist::BuiltInChatOperation) -> Self { match b_in { + oj_rc_core::persist::BuiltInChatOperation::Intercom(com) => Self::Intercom(Intercom::from_persist(com)), oj_rc_core::persist::BuiltInChatOperation::OnlineUsers => Self::OnlineUsers, oj_rc_core::persist::BuiltInChatOperation::TotalUsers => Self::TotalUsers, oj_rc_core::persist::BuiltInChatOperation::Version => Self::Version, @@ -127,8 +129,9 @@ impl BuiltIn { regex.trim_start_matches("\\") } - async fn do_command<'b, 'c>(&self, ctx: CommandContext<'b, 'c>) -> String { + async fn do_command<'b, 'c>(&self, text: &str, ctx: CommandContext<'b, 'c>) -> String { match self { + Self::Intercom(intercom) => intercom.do_command(text, ctx).await, Self::OnlineUsers => { let online_count = ctx.chat_system.user_count(); if online_count == 1 { @@ -173,6 +176,7 @@ impl BuiltIn { fn do_help(&self) -> String { match self { + Self::Intercom(i) => i.do_help(), Self::OnlineUsers => "Show total users online".to_owned(), Self::TotalUsers => "Show total users registered".to_owned(), Self::Version => "Show chat server version information".to_owned(), @@ -180,3 +184,38 @@ impl BuiltIn { } } } + +enum Intercom { + DevMessage, +} + +impl Intercom { + fn from_persist(intercom: oj_rc_core::persist::IntercomChatOperation) -> Self { + match intercom { + oj_rc_core::persist::IntercomChatOperation::DevMessage => Self::DevMessage, + } + } + + async fn do_command<'b, 'c>(&self, text: &str, ctx: CommandContext<'b, 'c>) -> String { + match self { + Self::DevMessage => { + let pub_id = ctx.user.public_id(); + ctx.user.show_dev_message( + oj_rc_core::persist::user::intercom::IntercomDevMessage { + message: text.trim().split_once(' ').map(|x| x.1.to_owned()).unwrap_or_else(|| "???".to_owned()), + duration: 10, + }, + vec![pub_id.to_owned()], + ).await; + format!("Sent dev message to {}", pub_id) + } + } + + } + + fn do_help(&self) -> String { + match self { + Self::DevMessage => "Show dev message to yourself".to_owned(), + } + } +} diff --git a/rc_core/Cargo.toml b/rc_core/Cargo.toml index 9b777e6..f9303d7 100644 --- a/rc_core/Cargo.toml +++ b/rc_core/Cargo.toml @@ -18,6 +18,7 @@ chrono.workspace = true polariton_server.workspace = true tokio = { version = "1.43", features = [ "net", "macros", "rt-multi-thread", "io-util", "time" ] } async-trait.workspace = true +futures.workspace = true rand.workspace = true num-quaternion.workspace = true @@ -29,6 +30,7 @@ argon2 = { version = "0.5", features = [ "std" ] } # intercom sha2 = "0.10" reqwest = { version = "0.12", default-features = false, features = [ "rustls-tls", "charset" ] } +reqwest-websocket = { version = "0.5", default-features = false, features = [ "json" ] } oj_rc_database = { version = "*", path = "../rc_database" } oj_rc_factory = { version = "*", path = "../rc_factory" } diff --git a/rc_core/src/persist/chat.rs b/rc_core/src/persist/chat.rs index f3dc1d6..b3aec2d 100644 --- a/rc_core/src/persist/chat.rs +++ b/rc_core/src/persist/chat.rs @@ -31,12 +31,19 @@ pub enum ChatOperation { #[derive(Serialize, Deserialize, Clone, Debug)] #[serde(tag = "built_in")] pub enum BuiltInChatOperation { + Intercom(IntercomChatOperation), OnlineUsers, TotalUsers, Version, Help, } +#[derive(Serialize, Deserialize, Clone, Debug)] +#[serde(tag = "intercom")] +pub enum IntercomChatOperation { + DevMessage, +} + fn default_pub_channs() -> Vec { vec![ diff --git a/rc_core/src/persist/config/cubes_json.rs b/rc_core/src/persist/config/cubes_json.rs index 7ed9de5..2122a00 100644 --- a/rc_core/src/persist/config/cubes_json.rs +++ b/rc_core/src/persist/config/cubes_json.rs @@ -271,6 +271,8 @@ impl super::ConfigProvider for CubeConfig { auto_signup: self.settings.server.auto_signup, queue_mode: super::QueueChangeMode::from_persist(self.settings.server.queue_mode.clone()), cdn_url: self.settings.server.cdn_url.trim_end_matches('/').to_owned(), + auth_url: self.settings.server.auth_url.trim_matches('/').to_owned(), + intercom_url: self.settings.server.intercom_url.trim_matches('/').to_owned(), } } diff --git a/rc_core/src/persist/config/traits.rs b/rc_core/src/persist/config/traits.rs index 22d6b19..e764ddc 100644 --- a/rc_core/src/persist/config/traits.rs +++ b/rc_core/src/persist/config/traits.rs @@ -113,6 +113,8 @@ pub struct ServerConfig { pub auto_signup: bool, pub queue_mode: QueueChangeMode, pub cdn_url: String, + pub auth_url: String, + pub intercom_url: String, } pub enum QueueChangeMode { diff --git a/rc_core/src/persist/mod.rs b/rc_core/src/persist/mod.rs index af7b0c9..a6ba94e 100644 --- a/rc_core/src/persist/mod.rs +++ b/rc_core/src/persist/mod.rs @@ -30,7 +30,7 @@ mod settings; pub use settings::{Settings, QueueMode}; mod chat; -pub use chat::{ChatConfig, ChatCommand, ChatOperation, BuiltInChatOperation}; +pub use chat::{ChatConfig, ChatCommand, ChatOperation, BuiltInChatOperation, IntercomChatOperation}; mod vehicle_factory; pub use vehicle_factory::{FactoryConfig, AdapterSettings, ArcFactorySettings}; diff --git a/rc_core/src/persist/settings.rs b/rc_core/src/persist/settings.rs index 52b104b..e8800b0 100644 --- a/rc_core/src/persist/settings.rs +++ b/rc_core/src/persist/settings.rs @@ -78,6 +78,10 @@ pub struct ServerSettings { pub queue_mode: QueueMode, #[serde(default = "default_cdn_root_url")] pub cdn_url: String, + #[serde(default = "default_auth_root_url")] + pub auth_url: String, + #[serde(default = "default_intercom_root_url")] + pub intercom_url: String, #[serde(default = "default_feedback_url")] pub feedback_url: String, #[serde(default = "default_support_url")] @@ -104,6 +108,8 @@ fn default_server_conf() -> ServerSettings { auto_signup: false, queue_mode: QueueMode::Notify, cdn_url: default_cdn_root_url(), + auth_url: default_auth_root_url(), + intercom_url: default_intercom_root_url(), feedback_url: default_feedback_url(), support_url: default_support_url(), wiki_url: default_wiki_url(), @@ -114,6 +120,14 @@ fn default_cdn_root_url() -> String { "http://127.0.0.1:8010".to_owned() } +fn default_auth_root_url() -> String { + "http://127.0.0.1:8001".to_owned() // mostly used for intercom +} + +fn default_intercom_root_url() -> String { + "ws://127.0.0.1:8001".to_owned() +} + fn default_feedback_url() -> String { "https://mstdn.ca/@ngram".to_owned() } diff --git a/rc_core/src/persist/user/account_json.rs b/rc_core/src/persist/user/account_json.rs index ae65698..fe51af3 100644 --- a/rc_core/src/persist/user/account_json.rs +++ b/rc_core/src/persist/user/account_json.rs @@ -9,6 +9,8 @@ pub struct AccountProvider { fake_players: std::sync::Arc>, auto_signups: bool, cdn: std::sync::Arc, + auth: std::sync::Arc, + intercom: std::sync::Arc, secret: std::sync::Arc>, db: std::sync::Arc, } @@ -29,6 +31,8 @@ impl AccountProvider { fake_players: std::sync::Arc::new(>::fake_players(conf)), auto_signups: server_settings.auto_signup, cdn: std::sync::Arc::new(server_settings.cdn_url), + auth: std::sync::Arc::new(server_settings.auth_url), + intercom: std::sync::Arc::new(server_settings.intercom_url), secret: std::sync::Arc::new(secret), db: std::sync::Arc::new(db), }) @@ -109,10 +113,12 @@ impl super::UserProvider for AccountProvider { garage_upgrades: self.garage_upgrades.clone(), fake_players: self.fake_players.clone(), cdn: self.cdn.clone(), + auth: self.auth.clone(), + intercom: self.intercom.clone(), + http_client: std::sync::Arc::new(reqwest::Client::new()), db: self.db.clone(), secret: self.secret.clone(), })) - //Err("Unable to authenticate".to_string()) } async fn multiplayer_authenticate(&self, user: String) -> Result + Send + Sync>, super::AuthError> { @@ -145,6 +151,9 @@ impl super::UserProvider for AccountProvider { garage_upgrades: self.garage_upgrades.clone(), fake_players: self.fake_players.clone(), cdn: self.cdn.clone(), + auth: self.auth.clone(), + intercom: self.intercom.clone(), + http_client: std::sync::Arc::new(reqwest::Client::new()), db: self.db.clone(), secret: self.secret.clone(), })) @@ -307,6 +316,9 @@ pub(super) struct UserData { pub(super) garage_upgrades: std::sync::Arc, pub(super) fake_players: std::sync::Arc>, pub(super) cdn: std::sync::Arc, + pub(super) auth: std::sync::Arc, + pub(super) intercom: std::sync::Arc, + pub(super) http_client: std::sync::Arc, pub(super) db: std::sync::Arc, pub(super) secret: std::sync::Arc>, } @@ -620,26 +632,6 @@ const UNEXPECTED_ERR: i16 = crate::data::error_codes::WebServicesError::Unexpect #[async_trait::async_trait] impl super::User for UserData { - fn public_id(&self) -> &'_ str { - &self.account.public_id - } - - fn is_mod(&self) -> bool { - self.perms.moderator - } - - fn is_admin(&self) -> bool { - self.perms.administrator - } - - fn is_dev(&self) -> bool { - self.perms.developer - } - - fn is_banned(&self) -> bool { - self.perms.banned - } - async fn unlocked_parts(&self) -> Vec { match self.db.user_aux_by_user_id_and_descriptor(self.account.id, oj_rc_database::schema::user_aux::Descriptor::UnlockedParts).await { Ok(Some(parts)) => { diff --git a/rc_core/src/persist/user/common.rs b/rc_core/src/persist/user/common.rs index da0d71f..c1d296e 100644 --- a/rc_core/src/persist/user/common.rs +++ b/rc_core/src/persist/user/common.rs @@ -2,6 +2,26 @@ use super::account_json::UserData; #[async_trait::async_trait] impl super::CommonUser for UserData { + fn public_id(&self) -> &'_ str { + &self.account.public_id + } + + fn is_mod(&self) -> bool { + self.perms.moderator + } + + fn is_admin(&self) -> bool { + self.perms.administrator + } + + fn is_dev(&self) -> bool { + self.perms.developer + } + + fn is_banned(&self) -> bool { + self.perms.banned + } + async fn resolve_config_vehicle(&self, vehicle: &crate::persist::config::VehicleInfo, factory: &dyn oj_rc_factory::VehicleFactoryAdapter, weapon_order: &crate::cubes::WeaponListParser, cpu_counter: &crate::cubes::CpuListParser) -> Result { self.resolve_vehicle(vehicle, factory, weapon_order, cpu_counter).await } diff --git a/rc_core/src/persist/user/intercom.rs b/rc_core/src/persist/user/intercom.rs index 4dee647..2477692 100644 --- a/rc_core/src/persist/user/intercom.rs +++ b/rc_core/src/persist/user/intercom.rs @@ -1,3 +1,40 @@ +use serde::{Serialize, Deserialize}; + +impl super::account_json::UserData { + async fn listen_on_websocket(&self, server_name: &str) -> Result, reqwest_websocket::Error> { + use reqwest_websocket::RequestBuilderExt; + let token = generate_token(format!("{}/{}", server_name, self.account.public_id).as_bytes(), &self.secret); + let auth_header_val = format!("Internal {}", token); + let url = format!("{}/intercom/{}/{}", self.intercom, server_name, self.account.public_id); + log::debug!("Listening on websocket {}", url); + let websocket = self.http_client.get(url) + .header("Authorization", auth_header_val) + .upgrade() + .send() + .await? + .into_websocket() + .await?; + Ok(super::IntercomListener { + websocket, + _d: Default::default(), + }) + } + + async fn post_to_intercom(&self, data: &D, server_name: &str, operation: &str) -> Result<(), reqwest::Error> { + let path = format!("{}/{}/{}", server_name, self.account.public_id, operation); + let token = generate_token(path.as_bytes(), &self.secret); + let auth_header_val = format!("Internal {}", token); + let url = format!("{}/intercom/{}", self.auth, path); + log::debug!("Posting intercom message to {}", url); + self.http_client.post(url) + .header("Authorization", auth_header_val) + .json(data) + .send() + .await?; + Ok(()) + } +} + #[async_trait::async_trait] impl super::IntercomUser for super::account_json::UserData { async fn save_custom_avatar(&self, image: Vec) -> Result<(), polariton_server::operations::SimpleOpError> { @@ -5,7 +42,7 @@ impl super::IntercomUser for super::account_json::UserData { let token = generate_token(self.account.public_id.as_bytes(), &self.secret); let auth_header_val = format!("Internal {}", token); let url = format!("{}/customavatar/Live/{}", self.cdn, self.account.public_id); - if let Err(e) = reqwest::Client::new().post(url) + if let Err(e) = self.http_client.post(url) .header("Authorization", auth_header_val) .body(image) .send() @@ -15,6 +52,42 @@ impl super::IntercomUser for super::account_json::UserData { } Ok(()) } + + async fn webservice_listener(&self) -> Result, polariton_server::operations::SimpleOpError> { + self.listen_on_websocket(".oj_services").await + .map_err(|e| polariton_server::operations::SimpleOpError::with_message( + crate::data::error_codes::WebServicesError::PlatformFeatureNotAvailable as i16, + e.to_string() + )) + } + + async fn show_dev_message(&self, msg: IntercomDevMessage, to: Vec) { + let data = IntercomWebServiceMessage { + public_ids: to, + data: IntercomWebServiceUserMessage::DevMessage(msg), + }; + if let Err(e) = self.post_to_intercom(&data, ".oj_services", "messages").await { + log::error!("Failed to send intercom dev message: {}", e); + } + } +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct IntercomWebServiceMessage { + pub public_ids: Vec, + pub data: IntercomWebServiceUserMessage, +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +#[serde(tag = "type")] +pub enum IntercomWebServiceUserMessage { + DevMessage(IntercomDevMessage), +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct IntercomDevMessage { + pub message: String, + pub duration: u32, } pub fn generate_token(salt: &[u8], key: &[u8]) -> String { diff --git a/rc_core/src/persist/user/mod.rs b/rc_core/src/persist/user/mod.rs index 90eadde..5631846 100644 --- a/rc_core/src/persist/user/mod.rs +++ b/rc_core/src/persist/user/mod.rs @@ -11,9 +11,9 @@ mod inventory; pub use inventory::UnlockedParts; mod traits; -pub use traits::{UserProvider, User, UserToken, UserSlots, UserSlotData, VehicleData, UserInfo, UserLoginInfo, ExtraUserInfo, UserAuthenticator, NewSlotData, UserId, RegistrationInfo, VehicleUploadData, ChatUser, AvatarInfo, GetAvatarInfo, ControlData, ControlType, CustomisationData, GetCustomisationData, SetSanction, SanctionType, LobbyUser, GameDescriptor, PlayerLobbyDescriptor, MultiplayerUser, MultiplayerError, MultiplayerErrorCode, PlayerDescriptor, GameEventSetter, CurrentGameEvent, AuthError, IntercomUser, FakePlayers, ResolvedVehicle, CommonUser}; +pub use traits::{UserProvider, User, UserToken, UserSlots, UserSlotData, VehicleData, UserInfo, UserLoginInfo, ExtraUserInfo, UserAuthenticator, NewSlotData, UserId, RegistrationInfo, VehicleUploadData, ChatUser, AvatarInfo, GetAvatarInfo, ControlData, ControlType, CustomisationData, GetCustomisationData, SetSanction, SanctionType, LobbyUser, GameDescriptor, PlayerLobbyDescriptor, MultiplayerUser, MultiplayerError, MultiplayerErrorCode, PlayerDescriptor, GameEventSetter, CurrentGameEvent, AuthError, IntercomUser, FakePlayers, ResolvedVehicle, CommonUser, IntercomListener}; -mod intercom; +pub mod intercom; pub use intercom::generate_token as generate_intercom_token; mod multiplayer; diff --git a/rc_core/src/persist/user/traits.rs b/rc_core/src/persist/user/traits.rs index ea575f1..8b244dc 100644 --- a/rc_core/src/persist/user/traits.rs +++ b/rc_core/src/persist/user/traits.rs @@ -62,11 +62,6 @@ pub trait UserAuthenticator { #[async_trait::async_trait] pub trait User: ChatUser + LobbyUser + MultiplayerUser + IntercomUser + CommonUser { - fn public_id(&self) -> &'_ str; - fn is_mod(&self) -> bool; - fn is_admin(&self) -> bool; - fn is_dev(&self) -> bool; - fn is_banned(&self) -> bool; async fn unlocked_parts(&self) -> Vec; async fn selected_garage(&self) -> (String, u32); async fn select_garage(&self, slot: i32) -> Result<(), i16>; @@ -203,7 +198,7 @@ pub struct AvatarInfo { } #[async_trait::async_trait] -pub trait ChatUser: CommonUser { +pub trait ChatUser: CommonUser + IntercomUser { async fn subscribed_channels(&self) -> Result, i16>; async fn subscribed_channels_strings(&self) -> Result, i16>; async fn add_subscribed_channel(&self, channel: String, channel_ty: crate::data::channel::ChatChannelType) -> Result, i16>; @@ -340,6 +335,20 @@ pub trait MultiplayerUser: CommonUser { #[async_trait::async_trait] pub trait IntercomUser { async fn save_custom_avatar(&self, image: Vec) -> Result<(), polariton_server::operations::SimpleOpError>; + async fn webservice_listener(&self) -> Result, polariton_server::operations::SimpleOpError>; + async fn show_dev_message(&self, msg: super::intercom::IntercomDevMessage, to: Vec); +} + +pub struct IntercomListener { + pub(super) websocket: reqwest_websocket::WebSocket, + pub(super) _d: std::marker::PhantomData, +} + +impl IntercomListener { + pub async fn listen(self) -> impl futures::Stream> + Unpin { + use futures::StreamExt; + self.websocket.map(|msg| msg.and_then(|msg| msg.json())) + } } pub struct ResolvedVehicle { @@ -359,4 +368,9 @@ pub struct ResolvedVehicle { #[async_trait::async_trait] pub trait CommonUser: Send + Sync { async fn resolve_config_vehicle(&self, vehicle: &crate::persist::config::VehicleInfo, factory: &dyn oj_rc_factory::VehicleFactoryAdapter, weapon_order: &crate::cubes::WeaponListParser, cpu_counter: &crate::cubes::CpuListParser) -> Result; + fn public_id(&self) -> &'_ str; + fn is_mod(&self) -> bool; + fn is_admin(&self) -> bool; + fn is_dev(&self) -> bool; + fn is_banned(&self) -> bool; } diff --git a/rc_services_room/Cargo.toml b/rc_services_room/Cargo.toml index 6eb6dc7..4050050 100644 --- a/rc_services_room/Cargo.toml +++ b/rc_services_room/Cargo.toml @@ -11,6 +11,7 @@ authors.workspace = true log.workspace = true env_logger.workspace = true tokio = { version = "1.43", features = [ "net", "macros", "rt-multi-thread", "io-util" ] } +futures.workspace = true clap.workspace = true polariton.workspace = true oj_polariton_auth = { version = "*", path = "../polariton_auth" } diff --git a/rc_services_room/src/events/dev_message.rs b/rc_services_room/src/events/dev_message.rs new file mode 100644 index 0000000..1ba49b2 --- /dev/null +++ b/rc_services_room/src/events/dev_message.rs @@ -0,0 +1,20 @@ +pub struct DevMessage { + pub message: String, + pub duration: i32, +} + +impl polariton_server::events::IntoEvent for DevMessage { + const CHANNEL: u8 = 0; + const ENCRYPT: bool = true; + const RELIABLE: bool = true; + + fn into_event(self) -> polariton::operation::Event { + let mut params = polariton::operation::ParameterTable::with_capacity(2); + params.insert(2, polariton::operation::Typed::Bytes(self.message.as_bytes().to_vec().into())); + params.insert(15, polariton::operation::Typed::Int(self.duration)); + polariton::operation::Event { + code: 1, + params, + } + } +} diff --git a/rc_services_room/src/events/handler.rs b/rc_services_room/src/events/handler.rs new file mode 100644 index 0000000..ecb6d9d --- /dev/null +++ b/rc_services_room/src/events/handler.rs @@ -0,0 +1,58 @@ +use oj_rc_core::persist::user::IntercomListener; +use oj_rc_core::persist::user::intercom::IntercomWebServiceUserMessage; + +pub struct IntercomHandler { + listener: IntercomListener, + user: std::sync::Weak + Send + Sync>>, + emitter: polariton_server::events::EventEmitter<()>, +} + +impl IntercomHandler { + pub fn new( + listener: IntercomListener, + user: &std::sync::Arc + Send + Sync>>, + emitter: &polariton_server::events::EventEmitter<()>, + ) -> Self { + Self { + listener, + user: std::sync::Arc::downgrade(user), + emitter: emitter.to_owned(), + } + } + + async fn run_loop( + listener: IntercomListener, + user: std::sync::Weak + Send + Sync>>, + emitter: polariton_server::events::EventEmitter<()> + ) { + use futures::StreamExt; + let mut listener = listener.listen().await; + while let Some(msg) = listener.next().await { + match msg { + Ok(msg) => { + if let Some(_user) = user.upgrade() { + match msg { + IntercomWebServiceUserMessage::DevMessage(msg) => { + let event = super::DevMessage { + message: msg.message, + duration: msg.duration as i32, + }; + emitter.emit(event); + }, + } + } else { + break; + } + }, + Err(e) => { + log::error!("Bad intercom message received: {}", e); + } + } + + } + } + + pub fn run(self) -> tokio::task::JoinHandle<()> { + tokio::spawn(Self::run_loop(self.listener, self.user, self.emitter)) + } +} diff --git a/rc_services_room/src/events/mod.rs b/rc_services_room/src/events/mod.rs index e69de29..766f6da 100644 --- a/rc_services_room/src/events/mod.rs +++ b/rc_services_room/src/events/mod.rs @@ -0,0 +1,5 @@ +mod handler; +pub use handler::IntercomHandler; + +mod dev_message; +pub use dev_message::DevMessage; diff --git a/rc_services_room/src/operations/more_auth.rs b/rc_services_room/src/operations/more_auth.rs index a8c6081..27f3389 100644 --- a/rc_services_room/src/operations/more_auth.rs +++ b/rc_services_room/src/operations/more_auth.rs @@ -16,7 +16,8 @@ impl Operation for MoreLobbyAuth { if let Some(Typed::Str(auth_payload)) = params_dict.get(&Self::AUTH_PAYLOAD_KEY) { //let mut write_lock = user.write().unwrap(); if user.update_with_auth(&auth_payload.string).await { - if user.user().unwrap().is_banned() { + let user_info = user.user().unwrap(); + if user_info.is_banned() { return polariton::operation::OperationResponse { code: Self::op_code(), return_code: oj_rc_core::data::error_codes::WebServicesError::Banned as i16, @@ -24,13 +25,27 @@ impl Operation for MoreLobbyAuth { params: polariton::operation::ParameterTable::with_capacity(0), } } else { - let mut resp_params = std::collections::HashMap::with_capacity(1); - resp_params.insert(Self::AUTH_PAYLOAD_KEY, polariton::operation::Typed::Byte(0)); - return polariton::operation::OperationResponse { - code: Self::op_code(), - return_code: 0, - message: polariton::operation::Typed::Null, - params: resp_params.into(), + match user_info.webservice_listener().await { + Ok(listener) => { + let mut resp_params = std::collections::HashMap::with_capacity(1); + resp_params.insert(Self::AUTH_PAYLOAD_KEY, polariton::operation::Typed::Byte(0)); + crate::events::IntercomHandler::new(listener, &user_info, user.event_sender()).run(); + return polariton::operation::OperationResponse { + code: Self::op_code(), + return_code: 0, + message: polariton::operation::Typed::Null, + params: resp_params.into(), + } + }, + Err(e) => { + log::error!("Failed to start web service intercom listener for user {}: {}", user_info.public_id(), e.error_msg().map(|x| x.to_owned()).unwrap_or("".to_string())); + return polariton::operation::OperationResponse { + code: Self::op_code(), + return_code: oj_rc_core::data::error_codes::WebServicesError::PlatformFeatureNotAvailable as i16, + message: polariton::operation::Typed::Null, + params: polariton::operation::ParameterTable::with_capacity(0), + } + }, } }