Merge pull request #82 from sebadob/audit-part-3

Audit part 3
sebadob · Oct 18, 2023 · 7b95acc · 7b95acc
2 parents 488f9de + e32b4ec
commit 7b95acc
Show file tree

Hide file tree

Showing 13 changed files with 299 additions and 59 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -9,7 +9,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.16.1"
+version = "0.16.2-20231017"
 edition = "2021"
 authors = ["Sebastian Dobe <[email protected]>"]
 license = "AGPLv3"

diff --git a/migrations/postgres/10_events.sql b/migrations/postgres/10_events.sql
@@ -1,9 +1,9 @@
 create table events
 (
-    id        varchar     not null,
-    timestamp timestamptz not null,
-    typ       varchar     not null,
-    level     smallint    not null,
+    id        varchar  not null,
+    timestamp bigint   not null,
+    typ       varchar  not null,
+    level     smallint not null,
     ip        varchar,
     constraint events_pk
         primary key (id, timestamp)

diff --git a/migrations/sqlite/10_events.sql b/migrations/sqlite/10_events.sql
@@ -1,9 +1,9 @@
 create table events
 (
-    id        varchar     not null,
-    timestamp timestamptz not null,
-    typ       varchar     not null,
-    level     smallint    not null,
+    id        varchar  not null,
+    timestamp bigint   not null,
+    typ       varchar  not null,
+    level     smallint not null,
     ip        varchar,
     constraint events_pk
         primary key (id, timestamp)

diff --git a/rauthy-book/src/config/config.md b/rauthy-book/src/config/config.md
@@ -260,6 +260,11 @@ MAX_HASH_THREADS=1
 # setting the initial password. (default: 86400)
 #ML_LT_PWD_FIRST=86400
 
+# The interval in seconds in which keep-alives should be sent to SSE clients.
+# Depending on your network setup, proxy timeouts, ..., you may adjust this value to fit your needs.
+# default: 30
+#SSE_KEEP_ALIVE=30
+
 #####################################
 ############# LOGGING ###############
 #####################################

diff --git a/rauthy-common/src/constants.rs b/rauthy-common/src/constants.rs
@@ -54,6 +54,8 @@ lazy_static! {
         .unwrap_or_else(|_| String::from("false"))
         .parse::<bool>()
         .expect("DEV_MODE cannot be parsed to bool - bad format");
+    pub static ref HA_MODE: bool =
+        env::var("HA_MODE").map(|s| s.to_lowercase() == "true").unwrap_or(false);
 
     pub static ref RE_ALG: Regex = Regex::new(r"^(RS256|RS384|RS512|EdDSA)$").unwrap();
     pub static ref RE_ATTR: Regex = Regex::new(r"^[a-zA-Z0-9-_/]{2,32}$").unwrap();
@@ -136,12 +138,16 @@ lazy_static! {
         .unwrap_or_else(|_| String::from("5400"))
         .parse::<u32>()
         .expect("SESSION_TIMEOUT cannot be parsed to u32 - bad format");
-
     pub static ref SESSION_VALIDATE_IP: bool = env::var("SESSION_VALIDATE_IP")
         .unwrap_or_else(|_| String::from("true"))
         .parse::<bool>()
         .expect("SESSION_VALIDATE_IP cannot be parsed to bool - bad format");
 
+    pub static ref SSE_KEEP_ALIVE: u16 = env::var("SSE_KEEP_ALIVE")
+        .unwrap_or_else(|_| String::from("30"))
+        .parse::<u16>()
+        .expect("SSE_KEEP_ALIVE cannot be parsed to u16 - bad format");
+
     pub static ref EMAIL_SUB_PREFIX: String = env::var("EMAIL_SUB_PREFIX")
         .unwrap_or_else(|_| String::from("Rauthy IAM"));
     pub static ref SMTP_USERNAME: String = env::var("SMTP_USERNAME")

diff --git a/rauthy-handlers/src/events.rs b/rauthy-handlers/src/events.rs
@@ -0,0 +1,30 @@
+use actix_web::{get, web, Responder};
+use actix_web_grants::proc_macro::has_roles;
+use actix_web_lab::sse;
+use rauthy_common::constants::SSE_KEEP_ALIVE;
+use rauthy_common::error_response::{ErrorResponse, ErrorResponseType};
+use rauthy_models::app_state::AppState;
+use rauthy_models::events::listener::EventRouterMsg;
+use std::time::Duration;
+
+#[get("/events")]
+// #[has_roles("rauthy_admin")] // TODO ADD BACK IN AFTER LOCAL TESTING!!!
+pub async fn sse_events(data: web::Data<AppState>) -> Result<impl Responder, ErrorResponse> {
+    let (tx, sse) = sse::channel(5);
+
+    if let Err(err) = data
+        .tx_events_router
+        .send_async(EventRouterMsg::ClientReg {
+            ip: "".to_string(),
+            tx,
+        })
+        .await
+    {
+        Err(ErrorResponse::new(
+            ErrorResponseType::Internal,
+            format!("Cannot register SSE client: {:?}", err),
+        ))
+    } else {
+        Ok(sse.with_keep_alive(Duration::from_secs(*SSE_KEEP_ALIVE as u64)))
+    }
+}
diff --git a/rauthy-handlers/src/lib.rs b/rauthy-handlers/src/lib.rs
@@ -26,6 +26,7 @@ use rust_embed::RustEmbed;
 use tracing::error;
 
 pub mod clients;
+pub mod events;
 pub mod generic;
 pub mod groups;
 pub mod middleware;

diff --git a/rauthy-main/src/main.rs b/rauthy-main/src/main.rs
@@ -41,7 +41,7 @@ use rauthy_common::password_hasher;
 use rauthy_handlers::middleware::logging::RauthyLoggingMiddleware;
 use rauthy_handlers::middleware::session::RauthySessionMiddleware;
 use rauthy_handlers::openapi::ApiDoc;
-use rauthy_handlers::{clients, generic, groups, oidc, roles, scopes, sessions, users};
+use rauthy_handlers::{clients, events, generic, groups, oidc, roles, scopes, sessions, users};
 use rauthy_models::app_state::{AppState, Caches, DbPool};
 use rauthy_models::email::EMail;
 use rauthy_models::events::event::Event;
@@ -167,45 +167,75 @@ async fn main() -> Result<(), Box<dyn Error>> {
         ha_cache_config: cache_config.clone(),
     };
     let (tx_events, rx_events) = flume::unbounded();
-    let app_state = web::Data::new(AppState::new(tx_email, tx_events.clone(), caches).await?);
+    let (tx_events_router, rx_events_router) = flume::unbounded();
+    let app_state = web::Data::new(
+        AppState::new(
+            tx_email,
+            tx_events.clone(),
+            tx_events_router.clone(),
+            caches,
+        )
+        .await?,
+    );
 
     // TODO remove with v0.17
     TEMP_migrate_passkeys_uv(&app_state.db)
         .await
         .expect("Passkey UV migration to not fail");
 
     // events listener
-    tokio::spawn(EventListener::listen(rx_events, app_state.db.clone()));
+    tokio::spawn(EventListener::listen(
+        tx_events_router,
+        rx_events_router,
+        rx_events,
+        app_state.db.clone(),
+    ));
 
     // TODO REMOVE AFTER TESTING
-    Event::brute_force("192.15.15.1".to_string())
-        .send(&tx_events)
-        .await
-        .unwrap();
-    Event::dos("192.15.15.1".to_string())
-        .send(&tx_events)
-        .await
-        .unwrap();
-    Event::invalid_login(2, "192.15.15.1".to_string())
-        .send(&tx_events)
-        .await
-        .unwrap();
-    Event::invalid_login(5, "192.15.15.1".to_string())
-        .send(&tx_events)
-        .await
-        .unwrap();
-    Event::invalid_login(7, "192.15.15.1".to_string())
-        .send(&tx_events)
-        .await
-        .unwrap();
-    Event::ip_blacklisted("192.15.15.1".to_string())
-        .send(&tx_events)
-        .await
-        .unwrap();
-    Event::rauthy_admin("[email protected]".to_string())
-        .send(&tx_events)
-        .await
-        .unwrap();
+    let txe = tx_events.clone();
+    tokio::spawn(async move {
+        time::sleep(Duration::from_secs(5)).await;
+        Event::brute_force("192.15.15.1".to_string())
+            .send(&txe)
+            .await
+            .unwrap();
+
+        time::sleep(Duration::from_secs(5)).await;
+        Event::dos("192.15.15.1".to_string())
+            .send(&txe)
+            .await
+            .unwrap();
+
+        time::sleep(Duration::from_secs(5)).await;
+        Event::invalid_login(2, "192.15.15.1".to_string())
+            .send(&txe)
+            .await
+            .unwrap();
+
+        time::sleep(Duration::from_secs(2)).await;
+        Event::invalid_login(5, "192.15.15.1".to_string())
+            .send(&txe)
+            .await
+            .unwrap();
+
+        time::sleep(Duration::from_secs(1)).await;
+        Event::invalid_login(7, "192.15.15.1".to_string())
+            .send(&txe)
+            .await
+            .unwrap();
+
+        time::sleep(Duration::from_secs(5)).await;
+        Event::ip_blacklisted("192.15.15.1".to_string())
+            .send(&txe)
+            .await
+            .unwrap();
+
+        time::sleep(Duration::from_secs(10)).await;
+        Event::rauthy_admin("[email protected]".to_string())
+            .send(&txe)
+            .await
+            .unwrap();
+    });
 
     // spawn password hash limiter
     tokio::spawn(password_hasher::run());
@@ -374,6 +404,7 @@ async fn actix_main(app_state: web::Data<AppState>) -> std::io::Result<()> {
                     .service(generic::redirect_v1)
                     .service(
                     web::scope("/v1")
+                        .service(events::sse_events)
                         .service(generic::get_index)
                         .service(generic::get_account_html)
                         .service(generic::get_admin_html)

diff --git a/rauthy-models/src/app_state.rs b/rauthy-models/src/app_state.rs
@@ -1,13 +1,14 @@
 use crate::email::EMail;
 use crate::entity::db_version::DbVersion;
 use crate::events::event::Event;
+use crate::events::listener::EventRouterMsg;
 use crate::migration::db_migrate;
 use crate::migration::db_migrate::migrate_init_prod;
 use crate::migration::db_migrate_dev::migrate_dev_data;
 use crate::ListenScheme;
 use anyhow::Context;
 use argon2::Params;
-use rauthy_common::constants::{DATABASE_URL, DB_TYPE, DEV_MODE, PROXY_MODE};
+use rauthy_common::constants::{DATABASE_URL, DB_TYPE, DEV_MODE, HA_MODE, PROXY_MODE};
 use rauthy_common::DbType;
 use regex::Regex;
 use serde::{Deserialize, Serialize};
@@ -54,6 +55,7 @@ pub struct AppState {
     pub ml_lt_pwd_reset: u32,
     pub tx_email: mpsc::Sender<EMail>,
     pub tx_events: flume::Sender<Event>,
+    pub tx_events_router: flume::Sender<EventRouterMsg>,
     pub caches: Caches,
     pub webauthn: Arc<Webauthn>,
 }
@@ -62,6 +64,7 @@ impl AppState {
     pub async fn new(
         tx_email: mpsc::Sender<EMail>,
         tx_events: flume::Sender<Event>,
+        tx_events_router: flume::Sender<EventRouterMsg>,
         caches: Caches,
     ) -> anyhow::Result<Self> {
         dotenvy::dotenv().ok();
@@ -196,6 +199,7 @@ impl AppState {
             ml_lt_pwd_reset,
             tx_email,
             tx_events,
+            tx_events_router,
             caches,
             webauthn,
         })
@@ -311,7 +315,7 @@ impl AppState {
         }
 
         if let Ok(from) = env::var("MIGRATE_DB_FROM") {
-            if is_ha_mode() {
+            if *HA_MODE {
                 error!(
                     r#"
     You cannot use 'MIGRATE_DB_FROM' with an active 'HA_MODE'.
@@ -373,7 +377,7 @@ impl AppState {
         migration_only: bool,
     ) -> anyhow::Result<sqlx::SqlitePool> {
         // HA_MODE must not be enabled while using SQLite
-        if !migration_only && is_ha_mode() {
+        if !migration_only && *HA_MODE {
             let msg = "HA_MODE must not be enabled while using SQLite";
             error!("{msg}");
             panic!("{msg}");
@@ -517,7 +521,3 @@ impl WellKnown {
         }
     }
 }
-
-fn is_ha_mode() -> bool {
-    env::var("HA_MODE").map(|s| s.to_lowercase()) == Ok("true".to_string())
-}
diff --git a/rauthy-models/src/events/event.rs b/rauthy-models/src/events/event.rs
@@ -152,6 +152,10 @@ impl Event {
         .await?;
         Ok(())
     }
+
+    pub fn as_json(&self) -> String {
+        serde_json::to_string(self).unwrap()
+    }
 }
 
 impl Event {