Add proof-of-concept server intercom service through auth server

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()

rc_auth/src/robocraft/intercom/internal_auth.rs

@@ -0,0 +1,83 @@
+pub struct IntercomAuth {
+    key: Vec<u8>,
+}
+
+impl IntercomAuth {
+    pub fn new(data: impl AsRef<std::path::Path>) -> std::io::Result<Self> {
+        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<actix_web::body::BoxBody> {
+        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()
+            }
+        }
+
+    }
+}

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;

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<super::IntercomAuth>, reg: Data<super::Users>, name: Path<String>) -> Result<HttpResponse, Error> {
+    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<oj_rc_core::persist::user::intercom::IntercomWebServiceMessage>, auth: Data<super::IntercomAuth>, reg: Data<super::Users>, name: Path<String>) -> Result<HttpResponse, super::IntercomOpError> {
+    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())
+}

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<std::collections::HashMap<String, tokio::sync::mpsc::Sender<IntercomWebServiceUserMessage>>>,
+}
+
+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<IntercomWebServiceUserMessage>) {
+        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);
+            }
+        }
+    }
+}

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,