automatically use email address as 2fa provider (#4317)

* Fix #3624: fix manager permission within groups

* Query returns UUID only

* Fix issue when user is manager and in a group having access to all collections

* optimize condition check

* fix(groups): renaming and optimizations

* fix: wrong organization group membership detection

* Simplify group membership check

Co-authored-by: Stefan Melmuk <509385+stefan0xC@users.noreply.github.com>

* Remove unused statement

* improve check if the user has access via groups

instead of returning the two lists of member ids and later checking if
they contain the uuid of the current user, we really only care if
the current user has full access via a group or if they have
access to a given collection via a group

* improve comments for get_org_collections_details

* small refactor to make it easier to review

* fix(groups): query full access via group only when necessary

Co-authored-by: Mathijs van Veluw <black.dex@gmail.com>

* chore(fmt): apply rustfmt

---------

Co-authored-by: Stefan Melmuk <509385+stefan0xC@users.noreply.github.com>
Co-authored-by: Stefan Melmuk <stefan.melmuk@gmail.com>
Co-authored-by: Mathijs van Veluw <black.dex@gmail.com>

.env.template

@@ -84,12 +84,8 @@
 ### WebSocket ###
 #################
 
-## Enables websocket notifications
-# WEBSOCKET_ENABLED=false
-
-## Controls the WebSocket server address and port
-# WEBSOCKET_ADDRESS=0.0.0.0
-# WEBSOCKET_PORT=3012
+## Enable websocket notifications
+# ENABLE_WEBSOCKET=true
 
 ##########################
 ### Push notifications ###
@@ -448,6 +444,11 @@
 ##
 ## Maximum attempts before an email token is reset and a new email will need to be sent.
 # EMAIL_ATTEMPTS_LIMIT=3
+##
+## Setup email 2FA regardless of any organization policy
+# EMAIL_2FA_ENFORCE_ON_VERIFIED_INVITE=false
+## Automatically setup email 2FA as fallback provider when needed
+# EMAIL_2FA_AUTO_FALLBACK=false
 
 ## Other MFA/2FA settings
 ## Disable 2FA remember

Cargo.lock

@@ -3784,7 +3784,6 @@ dependencies = [
  "syslog",
  "time",
  "tokio",
- "tokio-tungstenite",
  "totp-lite",
  "tracing",
  "url",

Cargo.toml

@@ -60,7 +60,6 @@ rocket = { version = "0.5.0", features = ["tls", "json"], default-features = fal
 rocket_ws = { version ="0.1.0" }
 
 # WebSockets libraries
-tokio-tungstenite = "0.20.1"
 rmpv = "1.0.1" # MessagePack library
 
 # Concurrent HashMap used for WebSocket messaging and favicons

migrations/mysql/2024-02-14-135828_change_time_stamp_data_type/up.sql

@@ -0,0 +1 @@
+ALTER TABLE twofactor MODIFY last_used BIGINT NOT NULL;

migrations/postgresql/2024-02-14-135953_change_time_stamp_data_type/up.sql

@@ -0,0 +1,3 @@
+ALTER TABLE twofactor
+ALTER COLUMN last_used TYPE BIGINT,
+ALTER COLUMN last_used SET NOT NULL;

migrations/sqlite/2024-02-14-140000_change_time_stamp_data_type/up.sql

@@ -0,0 +1 @@
+-- Integer size in SQLite is already i64, so we don't need to do anything

src/api/admin.rs

@@ -510,7 +510,11 @@ async fn update_user_org_type(data: Json<UserOrgTypeData>, token: AdminToken, mu
         match OrgPolicy::is_user_allowed(&user_to_edit.user_uuid, &user_to_edit.org_uuid, true, &mut conn).await {
             Ok(_) => {}
             Err(OrgPolicyErr::TwoFactorMissing) => {
-                err!("You cannot modify this user to this type because it has no two-step login method activated");
+                if CONFIG.email_2fa_auto_fallback() {
+                    two_factor::email::find_and_activate_email_2fa(&user_to_edit.user_uuid, &mut conn).await?;
+                } else {
+                    err!("You cannot modify this user to this type because they have not setup 2FA");
+                }
             }
             Err(OrgPolicyErr::SingleOrgEnforced) => {
                 err!("You cannot modify this user to this type because it is a member of an organization which forbids it");

src/api/core/accounts.rs

@@ -5,8 +5,9 @@ use serde_json::Value;
 
 use crate::{
     api::{
-        core::log_user_event, register_push_device, unregister_push_device, AnonymousNotify, EmptyResult, JsonResult,
-        JsonUpcase, Notify, PasswordOrOtpData, UpdateType,
+        core::{log_user_event, two_factor::email},
+        register_push_device, unregister_push_device, AnonymousNotify, EmptyResult, JsonResult, JsonUpcase, Notify,
+        PasswordOrOtpData, UpdateType,
     },
     auth::{decode_delete, decode_invite, decode_verify_email, ClientHeaders, Headers},
     crypto,
@@ -104,6 +105,19 @@ fn enforce_password_hint_setting(password_hint: &Option<String>) -> EmptyResult
     }
     Ok(())
 }
+async fn is_email_2fa_required(org_user_uuid: Option<String>, conn: &mut DbConn) -> bool {
+    if !CONFIG._enable_email_2fa() {
+        return false;
+    }
+    if CONFIG.email_2fa_enforce_on_verified_invite() {
+        return true;
+    }
+    if org_user_uuid.is_some() {
+        return OrgPolicy::is_enabled_by_org(&org_user_uuid.unwrap(), OrgPolicyType::TwoFactorAuthentication, conn)
+            .await;
+    }
+    false
+}
 
 #[post("/accounts/register", data = "<data>")]
 async fn register(data: JsonUpcase<RegisterData>, conn: DbConn) -> JsonResult {
@@ -208,6 +222,10 @@ pub async fn _register(data: JsonUpcase<RegisterData>, mut conn: DbConn) -> Json
         } else if let Err(e) = mail::send_welcome(&user.email).await {
             error!("Error sending welcome email: {:#?}", e);
         }
+
+        if verified_by_invite && is_email_2fa_required(data.OrganizationUserId, &mut conn).await {
+            let _ = email::activate_email_2fa(&user, &mut conn).await;
+        }
     }
 
     user.save(&mut conn).await?;
@@ -559,6 +577,8 @@ async fn post_email_token(data: JsonUpcase<EmailTokenData>, headers: Headers, mu
         if let Err(e) = mail::send_change_email(&data.NewEmail, &token).await {
             error!("Error sending change-email email: {:#?}", e);
         }
+    } else {
+        debug!("Email change request for user ({}) to email ({}) with token ({})", user.uuid, data.NewEmail, token);
     }
 
     user.email_new = Some(data.NewEmail);

src/api/core/organizations.rs

@@ -320,30 +320,24 @@ async fn get_org_collections_details(org_id: &str, headers: ManagerHeadersLoose,
         None => err!("User is not part of organization"),
     };
 
+    // get all collection memberships for the current organization
     let coll_users = CollectionUser::find_by_organization(org_id, &mut conn).await;
 
+    // check if current user has full access to the organization (either directly or via any group)
+    let has_full_access_to_org = user_org.access_all
+        || (CONFIG.org_groups_enabled()
+            && GroupUser::has_full_access_by_member(org_id, &user_org.uuid, &mut conn).await);
+
     for col in Collection::find_by_organization(org_id, &mut conn).await {
-        let groups: Vec<Value> = if CONFIG.org_groups_enabled() {
-            CollectionGroup::find_by_collection(&col.uuid, &mut conn)
-                .await
-                .iter()
-                .map(|collection_group| {
-                    SelectionReadOnly::to_collection_group_details_read_only(collection_group).to_json()
-                })
-                .collect()
-        } else {
-            // The Bitwarden clients seem to call this API regardless of whether groups are enabled,
-            // so just act as if there are no groups.
-            Vec::with_capacity(0)
-        };
+        // assigned indicates whether the current user has access to the given collection
+        let mut assigned = has_full_access_to_org;
 
-        let mut assigned = false;
+        // get the users assigned directly to the given collection
         let users: Vec<Value> = coll_users
             .iter()
             .filter(|collection_user| collection_user.collection_uuid == col.uuid)
             .map(|collection_user| {
-                // Remember `user_uuid` is swapped here with the `user_org.uuid` with a join during the `CollectionUser::find_by_organization` call.
-                // We check here if the current user is assigned to this collection or not.
+                // check if the current user is assigned to this collection directly
                 if collection_user.user_uuid == user_org.uuid {
                     assigned = true;
                 }
@@ -351,10 +345,24 @@ async fn get_org_collections_details(org_id: &str, headers: ManagerHeadersLoose,
             })
             .collect();
 
-        if user_org.access_all {
-            assigned = true;
+        // check if the current user has access to the given collection via a group
+        if !assigned && CONFIG.org_groups_enabled() {
+            assigned = GroupUser::has_access_to_collection_by_member(&col.uuid, &user_org.uuid, &mut conn).await;
         }
 
+        // get the group details for the given collection
+        let groups: Vec<Value> = if CONFIG.org_groups_enabled() {
+            CollectionGroup::find_by_collection(&col.uuid, &mut conn)
+                .await
+                .iter()
+                .map(|collection_group| {
+                    SelectionReadOnly::to_collection_group_details_read_only(collection_group).to_json()
+                })
+                .collect()
+        } else {
+            Vec::with_capacity(0)
+        };
+
         let mut json_object = col.to_json();
         json_object["Assigned"] = json!(assigned);
         json_object["Users"] = json!(users);
@@ -1071,7 +1079,7 @@ async fn accept_invite(
     let claims = decode_invite(&data.Token)?;
 
     match User::find_by_mail(&claims.email, &mut conn).await {
-        Some(_) => {
+        Some(user) => {
             Invitation::take(&claims.email, &mut conn).await;
 
             if let (Some(user_org), Some(org)) = (&claims.user_org_id, &claims.org_id) {
@@ -1095,7 +1103,11 @@ async fn accept_invite(
                     match OrgPolicy::is_user_allowed(&user_org.user_uuid, org_id, false, &mut conn).await {
                         Ok(_) => {}
                         Err(OrgPolicyErr::TwoFactorMissing) => {
-                            err!("You cannot join this organization until you enable two-step login on your user account");
+                            if CONFIG.email_2fa_auto_fallback() {
+                                two_factor::email::activate_email_2fa(&user, &mut conn).await?;
+                            } else {
+                                err!("You cannot join this organization until you enable two-step login on your user account");
+                            }
                         }
                         Err(OrgPolicyErr::SingleOrgEnforced) => {
                             err!("You cannot join this organization because you are a member of an organization which forbids it");
@@ -1220,10 +1232,14 @@ async fn _confirm_invite(
         match OrgPolicy::is_user_allowed(&user_to_confirm.user_uuid, org_id, true, conn).await {
             Ok(_) => {}
             Err(OrgPolicyErr::TwoFactorMissing) => {
-                err!("You cannot confirm this user because it has no two-step login method activated");
+                if CONFIG.email_2fa_auto_fallback() {
+                    two_factor::email::find_and_activate_email_2fa(&user_to_confirm.user_uuid, conn).await?;
+                } else {
+                    err!("You cannot confirm this user because they have not setup 2FA");
+                }
             }
             Err(OrgPolicyErr::SingleOrgEnforced) => {
-                err!("You cannot confirm this user because it is a member of an organization which forbids it");
+                err!("You cannot confirm this user because they are a member of an organization which forbids it");
             }
         }
     }
@@ -1351,10 +1367,14 @@ async fn edit_user(
         match OrgPolicy::is_user_allowed(&user_to_edit.user_uuid, org_id, true, &mut conn).await {
             Ok(_) => {}
             Err(OrgPolicyErr::TwoFactorMissing) => {
-                err!("You cannot modify this user to this type because it has no two-step login method activated");
+                if CONFIG.email_2fa_auto_fallback() {
+                    two_factor::email::find_and_activate_email_2fa(&user_to_edit.user_uuid, &mut conn).await?;
+                } else {
+                    err!("You cannot modify this user to this type because they have not setup 2FA");
+                }
             }
             Err(OrgPolicyErr::SingleOrgEnforced) => {
-                err!("You cannot modify this user to this type because it is a member of an organization which forbids it");
+                err!("You cannot modify this user to this type because they are a member of an organization which forbids it");
             }
         }
     }
@@ -2151,10 +2171,14 @@ async fn _restore_organization_user(
                 match OrgPolicy::is_user_allowed(&user_org.user_uuid, org_id, false, conn).await {
                     Ok(_) => {}
                     Err(OrgPolicyErr::TwoFactorMissing) => {
-                        err!("You cannot restore this user because it has no two-step login method activated");
+                        if CONFIG.email_2fa_auto_fallback() {
+                            two_factor::email::find_and_activate_email_2fa(&user_org.user_uuid, conn).await?;
+                        } else {
+                            err!("You cannot restore this user because they have not setup 2FA");
+                        }
                     }
                     Err(OrgPolicyErr::SingleOrgEnforced) => {
-                        err!("You cannot restore this user because it is a member of an organization which forbids it");
+                        err!("You cannot restore this user because they are a member of an organization which forbids it");
                     }
                 }
             }

src/api/core/two_factor/authenticator.rs

@@ -157,7 +157,7 @@ pub async fn validate_totp_code(
         let generated = totp_custom::<Sha1>(30, 6, &decoded_secret, time);
 
         // Check the given code equals the generated and if the time_step is larger then the one last used.
-        if generated == totp_code && time_step > i64::from(twofactor.last_used) {
+        if generated == totp_code && time_step > twofactor.last_used {
             // If the step does not equals 0 the time is drifted either server or client side.
             if step != 0 {
                 warn!("TOTP Time drift detected. The step offset is {}", step);
@@ -165,10 +165,10 @@ pub async fn validate_totp_code(
 
             // Save the last used time step so only totp time steps higher then this one are allowed.
             // This will also save a newly created twofactor if the code is correct.
-            twofactor.last_used = time_step as i32;
+            twofactor.last_used = time_step;
             twofactor.save(conn).await?;
             return Ok(());
-        } else if generated == totp_code && time_step <= i64::from(twofactor.last_used) {
+        } else if generated == totp_code && time_step <= twofactor.last_used {
             warn!("This TOTP or a TOTP code within {} steps back or forward has already been used!", steps);
             err!(
                 format!("Invalid TOTP code! Server time: {} IP: {}", current_time.format("%F %T UTC"), ip.ip),

src/api/core/two_factor/email.rs

@@ -10,7 +10,7 @@ use crate::{
     auth::Headers,
     crypto,
     db::{
-        models::{EventType, TwoFactor, TwoFactorType},
+        models::{EventType, TwoFactor, TwoFactorType, User},
         DbConn,
     },
     error::{Error, MapResult},
@@ -297,6 +297,15 @@ impl EmailTokenData {
     }
 }
 
+pub async fn activate_email_2fa(user: &User, conn: &mut DbConn) -> EmptyResult {
+    if user.verified_at.is_none() {
+        err!("Auto-enabling of email 2FA failed because the users email address has not been verified!");
+    }
+    let twofactor_data = EmailTokenData::new(user.email.clone(), String::new());
+    let twofactor = TwoFactor::new(user.uuid.clone(), TwoFactorType::Email, twofactor_data.to_json());
+    twofactor.save(conn).await
+}
+
 /// Takes an email address and obscures it by replacing it with asterisks except two characters.
 pub fn obscure_email(email: &str) -> String {
     let split: Vec<&str> = email.rsplitn(2, '@').collect();
@@ -318,6 +327,14 @@ pub fn obscure_email(email: &str) -> String {
     format!("{}@{}", new_name, &domain)
 }
 
+pub async fn find_and_activate_email_2fa(user_uuid: &str, conn: &mut DbConn) -> EmptyResult {
+    if let Some(user) = User::find_by_uuid(user_uuid, conn).await {
+        activate_email_2fa(&user, conn).await
+    } else {
+        err!("User not found!");
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/api/mod.rs

@@ -23,7 +23,7 @@ pub use crate::api::{
     icons::routes as icons_routes,
     identity::routes as identity_routes,
     notifications::routes as notifications_routes,
-    notifications::{start_notification_server, AnonymousNotify, Notify, UpdateType, WS_ANONYMOUS_SUBSCRIPTIONS},
+    notifications::{AnonymousNotify, Notify, UpdateType, WS_ANONYMOUS_SUBSCRIPTIONS, WS_USERS},
     push::{
         push_cipher_update, push_folder_update, push_logout, push_send_update, push_user_update, register_push_device,
         unregister_push_device,

src/api/notifications.rs

@@ -1,23 +1,11 @@
-use std::{
-    net::{IpAddr, SocketAddr},
-    sync::Arc,
-    time::Duration,
-};
+use std::{net::IpAddr, sync::Arc, time::Duration};
 
 use chrono::{NaiveDateTime, Utc};
 use rmpv::Value;
-use rocket::{
-    futures::{SinkExt, StreamExt},
-    Route,
-};
-use tokio::{
-    net::{TcpListener, TcpStream},
-    sync::mpsc::Sender,
-};
-use tokio_tungstenite::{
-    accept_hdr_async,
-    tungstenite::{handshake, Message},
-};
+use rocket::{futures::StreamExt, Route};
+use tokio::sync::mpsc::Sender;
+
+use rocket_ws::{Message, WebSocket};
 
 use crate::{
     auth::{ClientIp, WsAccessTokenHeader},
@@ -30,7 +18,7 @@ use crate::{
 
 use once_cell::sync::Lazy;
 
-static WS_USERS: Lazy<Arc<WebSocketUsers>> = Lazy::new(|| {
+pub static WS_USERS: Lazy<Arc<WebSocketUsers>> = Lazy::new(|| {
     Arc::new(WebSocketUsers {
         map: Arc::new(dashmap::DashMap::new()),
     })
@@ -47,8 +35,15 @@ use super::{
     push_send_update, push_user_update,
 };
 
+static NOTIFICATIONS_DISABLED: Lazy<bool> = Lazy::new(|| !CONFIG.enable_websocket() && !CONFIG.push_enabled());
+
 pub fn routes() -> Vec<Route> {
-    routes![websockets_hub, anonymous_websockets_hub]
+    if CONFIG.enable_websocket() {
+        routes![websockets_hub, anonymous_websockets_hub]
+    } else {
+        info!("WebSocket are disabled, realtime sync functionality will not work!");
+        routes![]
+    }
 }
 
 #[derive(FromForm, Debug)]
@@ -108,7 +103,7 @@ impl Drop for WSAnonymousEntryMapGuard {
 
 #[get("/hub?<data..>")]
 fn websockets_hub<'r>(
-    ws: rocket_ws::WebSocket,
+    ws: WebSocket,
     data: WsAccessToken,
     ip: ClientIp,
     header_token: WsAccessTokenHeader,
@@ -192,11 +187,7 @@ fn websockets_hub<'r>(
 }
 
 #[get("/anonymous-hub?<token..>")]
-fn anonymous_websockets_hub<'r>(
-    ws: rocket_ws::WebSocket,
-    token: String,
-    ip: ClientIp,
-) -> Result<rocket_ws::Stream!['r], Error> {
+fn anonymous_websockets_hub<'r>(ws: WebSocket, token: String, ip: ClientIp) -> Result<rocket_ws::Stream!['r], Error> {
     let addr = ip.ip;
     info!("Accepting Anonymous Rocket WS connection from {addr}");
 
@@ -349,13 +340,19 @@ impl WebSocketUsers {
 
     // NOTE: The last modified date needs to be updated before calling these methods
     pub async fn send_user_update(&self, ut: UpdateType, user: &User) {
+        // Skip any processing if both WebSockets and Push are not active
+        if *NOTIFICATIONS_DISABLED {
+            return;
+        }
         let data = create_update(
             vec![("UserId".into(), user.uuid.clone().into()), ("Date".into(), serialize_date(user.updated_at))],
             ut,
             None,
         );
 
-        self.send_update(&user.uuid, &data).await;
+        if CONFIG.enable_websocket() {
+            self.send_update(&user.uuid, &data).await;
+        }
 
         if CONFIG.push_enabled() {
             push_user_update(ut, user);
@@ -363,13 +360,19 @@ impl WebSocketUsers {
     }
 
     pub async fn send_logout(&self, user: &User, acting_device_uuid: Option<String>) {
+        // Skip any processing if both WebSockets and Push are not active
+        if *NOTIFICATIONS_DISABLED {
+            return;
+        }
         let data = create_update(
             vec![("UserId".into(), user.uuid.clone().into()), ("Date".into(), serialize_date(user.updated_at))],
             UpdateType::LogOut,
             acting_device_uuid.clone(),
         );
 
-        self.send_update(&user.uuid, &data).await;
+        if CONFIG.enable_websocket() {
+            self.send_update(&user.uuid, &data).await;
+        }
 
         if CONFIG.push_enabled() {
             push_logout(user, acting_device_uuid);
@@ -383,6 +386,10 @@ impl WebSocketUsers {
         acting_device_uuid: &String,
         conn: &mut DbConn,
     ) {
+        // Skip any processing if both WebSockets and Push are not active
+        if *NOTIFICATIONS_DISABLED {
+            return;
+        }
         let data = create_update(
             vec![
                 ("Id".into(), folder.uuid.clone().into()),
@@ -393,7 +400,9 @@ impl WebSocketUsers {
             Some(acting_device_uuid.into()),
         );
 
-        self.send_update(&folder.user_uuid, &data).await;
+        if CONFIG.enable_websocket() {
+            self.send_update(&folder.user_uuid, &data).await;
+        }
 
         if CONFIG.push_enabled() {
             push_folder_update(ut, folder, acting_device_uuid, conn).await;
@@ -409,6 +418,10 @@ impl WebSocketUsers {
         collection_uuids: Option<Vec<String>>,
         conn: &mut DbConn,
     ) {
+        // Skip any processing if both WebSockets and Push are not active
+        if *NOTIFICATIONS_DISABLED {
+            return;
+        }
         let org_uuid = convert_option(cipher.organization_uuid.clone());
         // Depending if there are collections provided or not, we need to have different values for the following variables.
         // The user_uuid should be `null`, and the revision date should be set to now, else the clients won't sync the collection change.
@@ -434,8 +447,10 @@ impl WebSocketUsers {
             Some(acting_device_uuid.into()),
         );
 
-        for uuid in user_uuids {
-            self.send_update(uuid, &data).await;
+        if CONFIG.enable_websocket() {
+            for uuid in user_uuids {
+                self.send_update(uuid, &data).await;
+            }
         }
 
         if CONFIG.push_enabled() && user_uuids.len() == 1 {
@@ -451,6 +466,10 @@ impl WebSocketUsers {
         acting_device_uuid: &String,
         conn: &mut DbConn,
     ) {
+        // Skip any processing if both WebSockets and Push are not active
+        if *NOTIFICATIONS_DISABLED {
+            return;
+        }
         let user_uuid = convert_option(send.user_uuid.clone());
 
         let data = create_update(
@@ -463,8 +482,10 @@ impl WebSocketUsers {
             None,
         );
 
-        for uuid in user_uuids {
-            self.send_update(uuid, &data).await;
+        if CONFIG.enable_websocket() {
+            for uuid in user_uuids {
+                self.send_update(uuid, &data).await;
+            }
         }
         if CONFIG.push_enabled() && user_uuids.len() == 1 {
             push_send_update(ut, send, acting_device_uuid, conn).await;
@@ -478,12 +499,18 @@ impl WebSocketUsers {
         acting_device_uuid: &String,
         conn: &mut DbConn,
     ) {
+        // Skip any processing if both WebSockets and Push are not active
+        if *NOTIFICATIONS_DISABLED {
+            return;
+        }
         let data = create_update(
             vec![("Id".into(), auth_request_uuid.clone().into()), ("UserId".into(), user_uuid.clone().into())],
             UpdateType::AuthRequest,
             Some(acting_device_uuid.to_string()),
         );
-        self.send_update(user_uuid, &data).await;
+        if CONFIG.enable_websocket() {
+            self.send_update(user_uuid, &data).await;
+        }
 
         if CONFIG.push_enabled() {
             push_auth_request(user_uuid.to_string(), auth_request_uuid.to_string(), conn).await;
@@ -497,12 +524,18 @@ impl WebSocketUsers {
         approving_device_uuid: String,
         conn: &mut DbConn,
     ) {
+        // Skip any processing if both WebSockets and Push are not active
+        if *NOTIFICATIONS_DISABLED {
+            return;
+        }
         let data = create_update(
             vec![("Id".into(), auth_response_uuid.to_owned().into()), ("UserId".into(), user_uuid.clone().into())],
             UpdateType::AuthRequestResponse,
             approving_device_uuid.clone().into(),
         );
-        self.send_update(auth_response_uuid, &data).await;
+        if CONFIG.enable_websocket() {
+            self.send_update(auth_response_uuid, &data).await;
+        }
 
         if CONFIG.push_enabled() {
             push_auth_response(user_uuid.to_string(), auth_response_uuid.to_string(), approving_device_uuid, conn)
@@ -526,6 +559,9 @@ impl AnonymousWebSocketSubscriptions {
     }
 
     pub async fn send_auth_response(&self, user_uuid: &String, auth_response_uuid: &str) {
+        if !CONFIG.enable_websocket() {
+            return;
+        }
         let data = create_anonymous_update(
             vec![("Id".into(), auth_response_uuid.to_owned().into()), ("UserId".into(), user_uuid.clone().into())],
             UpdateType::AuthRequestResponse,
@@ -620,127 +656,3 @@ pub enum UpdateType {
 
 pub type Notify<'a> = &'a rocket::State<Arc<WebSocketUsers>>;
 pub type AnonymousNotify<'a> = &'a rocket::State<Arc<AnonymousWebSocketSubscriptions>>;
-
-pub fn start_notification_server() -> Arc<WebSocketUsers> {
-    let users = Arc::clone(&WS_USERS);
-    if CONFIG.websocket_enabled() {
-        let users2 = Arc::<WebSocketUsers>::clone(&users);
-        tokio::spawn(async move {
-            let addr = (CONFIG.websocket_address(), CONFIG.websocket_port());
-            info!("Starting WebSockets server on {}:{}", addr.0, addr.1);
-            let listener = TcpListener::bind(addr).await.expect("Can't listen on websocket port");
-
-            let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel::<()>();
-            CONFIG.set_ws_shutdown_handle(shutdown_tx);
-
-            loop {
-                tokio::select! {
-                    Ok((stream, addr)) = listener.accept() => {
-                        tokio::spawn(handle_connection(stream, Arc::<WebSocketUsers>::clone(&users2), addr));
-                    }
-
-                    _ = &mut shutdown_rx => {
-                        break;
-                    }
-                }
-            }
-
-            info!("Shutting down WebSockets server!")
-        });
-    }
-
-    users
-}
-
-async fn handle_connection(stream: TcpStream, users: Arc<WebSocketUsers>, addr: SocketAddr) -> Result<(), Error> {
-    let mut user_uuid: Option<String> = None;
-
-    info!("Accepting WS connection from {addr}");
-
-    // Accept connection, do initial handshake, validate auth token and get the user ID
-    use handshake::server::{Request, Response};
-    let mut stream = accept_hdr_async(stream, |req: &Request, res: Response| {
-        if let Some(token) = get_request_token(req) {
-            if let Ok(claims) = crate::auth::decode_login(&token) {
-                user_uuid = Some(claims.sub);
-                return Ok(res);
-            }
-        }
-        Err(Response::builder().status(401).body(None).unwrap())
-    })
-    .await?;
-
-    let user_uuid = user_uuid.expect("User UUID should be set after the handshake");
-
-    let (mut rx, guard) = {
-        // Add a channel to send messages to this client to the map
-        let entry_uuid = uuid::Uuid::new_v4();
-        let (tx, rx) = tokio::sync::mpsc::channel::<Message>(100);
-        users.map.entry(user_uuid.clone()).or_default().push((entry_uuid, tx));
-
-        // Once the guard goes out of scope, the connection will have been closed and the entry will be deleted from the map
-        (rx, WSEntryMapGuard::new(users, user_uuid, entry_uuid, addr.ip()))
-    };
-
-    let _guard = guard;
-    let mut interval = tokio::time::interval(Duration::from_secs(15));
-    loop {
-        tokio::select! {
-            res = stream.next() =>  {
-                match res {
-                    Some(Ok(message)) => {
-                        match message {
-                            // Respond to any pings
-                            Message::Ping(ping) => stream.send(Message::Pong(ping)).await?,
-                            Message::Pong(_) => {/* Ignored */},
-
-                            // We should receive an initial message with the protocol and version, and we will reply to it
-                            Message::Text(ref message) => {
-                                let msg = message.strip_suffix(RECORD_SEPARATOR as char).unwrap_or(message);
-
-                                if serde_json::from_str(msg).ok() == Some(INITIAL_MESSAGE) {
-                                    stream.send(Message::binary(INITIAL_RESPONSE)).await?;
-                                    continue;
-                                }
-                            }
-                            // Just echo anything else the client sends
-                            _ => stream.send(message).await?,
-                        }
-                    }
-                    _ => break,
-                }
-            }
-
-            res = rx.recv() => {
-                match res {
-                    Some(res) => stream.send(res).await?,
-                    None => break,
-                }
-            }
-
-            _ = interval.tick() => stream.send(Message::Ping(create_ping())).await?
-        }
-    }
-
-    Ok(())
-}
-
-fn get_request_token(req: &handshake::server::Request) -> Option<String> {
-    const ACCESS_TOKEN_KEY: &str = "access_token=";
-
-    if let Some(Ok(auth)) = req.headers().get("Authorization").map(|a| a.to_str()) {
-        if let Some(token_part) = auth.strip_prefix("Bearer ") {
-            return Some(token_part.to_owned());
-        }
-    }
-
-    if let Some(params) = req.uri().query() {
-        let params_iter = params.split('&').take(1);
-        for val in params_iter {
-            if let Some(stripped) = val.strip_prefix(ACCESS_TOKEN_KEY) {
-                return Some(stripped.to_owned());
-            }
-        }
-    }
-    None
-}

src/config.rs

@@ -39,7 +39,6 @@ macro_rules! make_config {
 
         struct Inner {
             rocket_shutdown_handle: Option<rocket::Shutdown>,
-            ws_shutdown_handle: Option<tokio::sync::oneshot::Sender<()>>,
 
             templates: Handlebars<'static>,
             config: ConfigItems,
@@ -361,7 +360,7 @@ make_config! {
         /// Sends folder
         sends_folder:           String, false,  auto,   |c| format!("{}/{}", c.data_folder, "sends");
         /// Temp folder |> Used for storing temporary file uploads
-        tmp_folder:           String, false,  auto,   |c| format!("{}/{}", c.data_folder, "tmp");
+        tmp_folder:             String, false,  auto,   |c| format!("{}/{}", c.data_folder, "tmp");
         /// Templates folder
         templates_folder:       String, false,  auto,   |c| format!("{}/{}", c.data_folder, "templates");
         /// Session JWT key
@@ -371,11 +370,7 @@ make_config! {
     },
     ws {
         /// Enable websocket notifications
-        websocket_enabled:      bool,   false,  def,    false;
-        /// Websocket address
-        websocket_address:      String, false,  def,    "0.0.0.0".to_string();
-        /// Websocket port
-        websocket_port:         u16,    false,  def,    3012;
+        enable_websocket:       bool,   false,  def,    true;
     },
     push {
         /// Enable push notifications
@@ -691,6 +686,10 @@ make_config! {
         email_expiration_time:  u64,    true,   def,      600;
         /// Maximum attempts |> Maximum attempts before an email token is reset and a new email will need to be sent
         email_attempts_limit:   u64,    true,   def,      3;
+        /// Automatically enforce at login |> Setup email 2FA provider regardless of any organization policy
+        email_2fa_enforce_on_verified_invite: bool,   true,   def,      false;
+        /// Auto-enable 2FA (Know the risks!) |> Automatically setup email 2FA as fallback provider when needed
+        email_2fa_auto_fallback: bool,  true,   def,      false;
     },
 }
 
@@ -893,6 +892,13 @@ fn validate_config(cfg: &ConfigItems) -> Result<(), Error> {
         err!("To enable email 2FA, a mail transport must be configured")
     }
 
+    if !cfg._enable_email_2fa && cfg.email_2fa_enforce_on_verified_invite {
+        err!("To enforce email 2FA on verified invitations, email 2fa has to be enabled!");
+    }
+    if !cfg._enable_email_2fa && cfg.email_2fa_auto_fallback {
+        err!("To use email 2FA as automatic fallback, email 2fa has to be enabled!");
+    }
+
     // Check if the icon blacklist regex is valid
     if let Some(ref r) = cfg.icon_blacklist_regex {
         let validate_regex = regex::Regex::new(r);
@@ -1071,7 +1077,6 @@ impl Config {
         Ok(Config {
             inner: RwLock::new(Inner {
                 rocket_shutdown_handle: None,
-                ws_shutdown_handle: None,
                 templates: load_templates(&config.templates_folder),
                 config,
                 _env,
@@ -1237,16 +1242,8 @@ impl Config {
         self.inner.write().unwrap().rocket_shutdown_handle = Some(handle);
     }
 
-    pub fn set_ws_shutdown_handle(&self, handle: tokio::sync::oneshot::Sender<()>) {
-        self.inner.write().unwrap().ws_shutdown_handle = Some(handle);
-    }
-
     pub fn shutdown(&self) {
         if let Ok(mut c) = self.inner.write() {
-            if let Some(handle) = c.ws_shutdown_handle.take() {
-                handle.send(()).ok();
-            }
-
             if let Some(handle) = c.rocket_shutdown_handle.take() {
                 handle.notify();
             }

src/db/models/group.rs

@@ -486,6 +486,39 @@ impl GroupUser {
         }}
     }
 
+    pub async fn has_access_to_collection_by_member(
+        collection_uuid: &str,
+        member_uuid: &str,
+        conn: &mut DbConn,
+    ) -> bool {
+        db_run! { conn: {
+            groups_users::table
+                .inner_join(collections_groups::table.on(
+                    collections_groups::groups_uuid.eq(groups_users::groups_uuid)
+                ))
+                .filter(collections_groups::collections_uuid.eq(collection_uuid))
+                .filter(groups_users::users_organizations_uuid.eq(member_uuid))
+                .count()
+                .first::<i64>(conn)
+                .unwrap_or(0) != 0
+        }}
+    }
+
+    pub async fn has_full_access_by_member(org_uuid: &str, member_uuid: &str, conn: &mut DbConn) -> bool {
+        db_run! { conn: {
+            groups_users::table
+                .inner_join(groups::table.on(
+                    groups::uuid.eq(groups_users::groups_uuid)
+                ))
+                .filter(groups::organizations_uuid.eq(org_uuid))
+                .filter(groups::access_all.eq(true))
+                .filter(groups_users::users_organizations_uuid.eq(member_uuid))
+                .count()
+                .first::<i64>(conn)
+                .unwrap_or(0) != 0
+        }}
+    }
+
     pub async fn update_user_revision(&self, conn: &mut DbConn) {
         match UserOrganization::find_by_uuid(&self.users_organizations_uuid, conn).await {
             Some(user) => User::update_uuid_revision(&user.user_uuid, conn).await,

src/db/models/org_policy.rs

@@ -340,4 +340,11 @@ impl OrgPolicy {
         }
         false
     }
+
+    pub async fn is_enabled_by_org(org_uuid: &str, policy_type: OrgPolicyType, conn: &mut DbConn) -> bool {
+        if let Some(policy) = OrgPolicy::find_by_org_and_type(org_uuid, policy_type, conn).await {
+            return policy.enabled;
+        }
+        false
+    }
 }

src/db/models/two_factor.rs

@@ -12,7 +12,7 @@ db_object! {
         pub atype: i32,
         pub enabled: bool,
         pub data: String,
-        pub last_used: i32,
+        pub last_used: i64,
     }
 }
 

src/db/schemas/mysql/schema.rs

@@ -160,7 +160,7 @@ table! {
         atype -> Integer,
         enabled -> Bool,
         data -> Text,
-        last_used -> Integer,
+        last_used -> BigInt,
     }
 }
 

src/db/schemas/postgresql/schema.rs

@@ -160,7 +160,7 @@ table! {
         atype -> Integer,
         enabled -> Bool,
         data -> Text,
-        last_used -> Integer,
+        last_used -> BigInt,
     }
 }
 

src/db/schemas/sqlite/schema.rs

@@ -160,7 +160,7 @@ table! {
         atype -> Integer,
         enabled -> Bool,
         data -> Text,
-        last_used -> Integer,
+        last_used -> BigInt,
     }
 }
 

src/error.rs

@@ -52,7 +52,6 @@ use rocket::error::Error as RocketErr;
 use serde_json::{Error as SerdeErr, Value};
 use std::io::Error as IoErr;
 use std::time::SystemTimeError as TimeErr;
-use tokio_tungstenite::tungstenite::Error as TungstError;
 use webauthn_rs::error::WebauthnError as WebauthnErr;
 use yubico::yubicoerror::YubicoError as YubiErr;
 
@@ -91,7 +90,6 @@ make_error! {
 
     DieselCon(DieselConErr): _has_source, _api_error,
     Webauthn(WebauthnErr):   _has_source, _api_error,
-    WebSocket(TungstError):  _has_source, _api_error,
 }
 
 impl std::fmt::Debug for Error {

src/main.rs

@@ -52,7 +52,7 @@ mod ratelimit;
 mod util;
 
 use crate::api::purge_auth_requests;
-use crate::api::WS_ANONYMOUS_SUBSCRIPTIONS;
+use crate::api::{WS_ANONYMOUS_SUBSCRIPTIONS, WS_USERS};
 pub use config::CONFIG;
 pub use error::{Error, MapResult};
 use rocket::data::{Limits, ToByteUnit};
@@ -65,7 +65,11 @@ async fn main() -> Result<(), Error> {
     launch_info();
 
     use log::LevelFilter as LF;
-    let level = LF::from_str(&CONFIG.log_level()).expect("Valid log level");
+    let level = LF::from_str(&CONFIG.log_level()).unwrap_or_else(|_| {
+        let valid_log_levels = LF::iter().map(|lvl| lvl.as_str().to_lowercase()).collect::<Vec<String>>().join(", ");
+        println!("Log level must be one of the following: {valid_log_levels}");
+        exit(1);
+    });
     init_logging(level).ok();
 
     let extra_debug = matches!(level, LF::Trace | LF::Debug);
@@ -497,7 +501,7 @@ async fn launch_rocket(pool: db::DbPool, extra_debug: bool) -> Result<(), Error>
         .register([basepath, "/api"].concat(), api::core_catchers())
         .register([basepath, "/admin"].concat(), api::admin_catchers())
         .manage(pool)
-        .manage(api::start_notification_server())
+        .manage(Arc::clone(&WS_USERS))
         .manage(Arc::clone(&WS_ANONYMOUS_SUBSCRIPTIONS))
         .attach(util::AppHeaders())
         .attach(util::Cors())

src/static/templates/email/change_email.hbs

@@ -2,5 +2,5 @@ Your Email Change
 <!---------------->
 To finalize changing your email address enter the following code in web vault: {{token}}
 
-If you did not try to change an email address, you can safely ignore this email.
+If you did not try to change your email address, contact your administrator.
 {{> email/email_footer_text }}

src/static/templates/email/change_email.html.hbs

@@ -9,7 +9,7 @@ Your Email Change
    </tr>
    <tr style="margin: 0; font-family: 'Helvetica Neue', Helvetica, Arial, sans-serif; box-sizing: border-box; font-size: 16px; color: #333; line-height: 25px; -webkit-font-smoothing: antialiased; -webkit-text-size-adjust: none;">
       <td class="content-block last" style="font-family: 'Helvetica Neue', Helvetica, Arial, sans-serif; box-sizing: border-box; font-size: 16px; color: #333; line-height: 25px; margin: 0; -webkit-font-smoothing: antialiased; padding: 0; -webkit-text-size-adjust: none; text-align: center;" valign="top" align="center">
-         If you did not try to change an email address, you can safely ignore this email.
+         If you did not try to change your email address, contact your administrator.
       </td>
    </tr>
 </table>