Implement multiple proxy handling

Cargo.toml

@@ -17,6 +17,7 @@ defguard_event_logger = { path = "./crates/defguard_event_logger", version = "0.
 defguard_event_router = { path = "./crates/defguard_event_router", version = "0.0.0" }
 defguard_mail = { path = "./crates/defguard_mail", version = "0.0.0" }
 defguard_proto = { path = "./crates/defguard_proto", version = "0.0.0" }
+defguard_proxy_manager = { path = "./crates/defguard_proxy_manager", version = "0.0.0" }
 defguard_session_manager = { path = "./crates/defguard_session_manager", version = "0.0.0" }
 defguard_version = { path = "./crates/defguard_version", version = "0.0.0" }
 defguard_web_ui = { path = "./crates/defguard_web_ui", version = "0.0.0" }

crates/defguard/Cargo.toml

@@ -14,6 +14,7 @@ defguard_core = { workspace = true }
 defguard_event_router = { workspace = true }
 defguard_event_logger = { workspace = true }
 defguard_mail = { workspace = true }
+defguard_proxy_manager = { workspace = true }
 defguard_session_manager = { workspace = true }
 defguard_version = { workspace = true }
 

crates/defguard/src/main.rs

@@ -29,7 +29,7 @@ use defguard_core::{
     grpc::{
         WorkerState,
         gateway::{client_state::ClientMap, events::GatewayEvent, map::GatewayMap},
-        run_grpc_bidi_stream, run_grpc_server,
+        run_grpc_server,
     },
     init_dev_env, init_vpn_location, run_web_server,
     utility_thread::run_utility_thread,
@@ -40,6 +40,7 @@ use defguard_core::{
 use defguard_event_logger::{message::EventLoggerMessage, run_event_logger};
 use defguard_event_router::{RouterReceiverSet, run_event_router};
 use defguard_mail::{Mail, run_mail_handler};
+use defguard_proxy_manager::{ProxyOrchestrator, ProxyTxSet};
 // use defguard_session_manager::run_session_manager;
 use secrecy::ExposeSecret;
 use tokio::sync::{broadcast, mpsc::unbounded_channel};
@@ -158,15 +159,13 @@ async fn main() -> Result<(), anyhow::Error> {
         }
     }
 
+    let proxy_tx = ProxyTxSet::new(wireguard_tx.clone(), mail_tx.clone(), bidi_event_tx.clone());
+    let proxy_orchestrator =
+        ProxyOrchestrator::new(pool.clone(), proxy_tx, Arc::clone(&incompatible_components));
+
     // run services
     tokio::select! {
-        res = run_grpc_bidi_stream(
-            pool.clone(),
-            wireguard_tx.clone(),
-            mail_tx.clone(),
-            bidi_event_tx,
-            Arc::clone(&incompatible_components),
-        ), if config.proxy_url.is_some() => error!("Proxy gRPC stream returned early: {res:?}"),
+        res = proxy_orchestrator.run(&config.proxy_url) => error!("ProxyOrchestrator returned early: {res:?}"),
         res = run_grpc_server(
             Arc::clone(&worker_state),
             pool.clone(),

crates/defguard_core/src/db/models/enrollment.rs

@@ -4,14 +4,18 @@ use defguard_common::{
     config::server_config,
     db::{
         Id,
-        models::{Settings, user::User},
+        models::{Settings, settings::defaults::WELCOME_EMAIL_SUBJECT, user::User},
     },
     random::gen_alphanumeric,
 };
-use defguard_mail::templates::{self, TemplateError, safe_tera};
-use sqlx::{Error as SqlxError, PgConnection, PgExecutor, PgPool, query, query_as};
+use defguard_mail::{
+    Mail,
+    templates::{self, TemplateError, safe_tera},
+};
+use sqlx::{Error as SqlxError, PgConnection, PgExecutor, PgPool, Transaction, query, query_as};
 use tera::Context;
 use thiserror::Error;
+use tokio::sync::mpsc::UnboundedSender;
 use tonic::{Code, Status};
 
 pub static ENROLLMENT_TOKEN_TYPE: &str = "ENROLLMENT";
@@ -380,6 +384,80 @@ impl Token {
             device_info,
         )?)
     }
+
+    // Send configured welcome email to user after finishing enrollment
+    pub async fn send_welcome_email(
+        &self,
+        transaction: &mut Transaction<'_, sqlx::Postgres>,
+        mail_tx: &UnboundedSender<Mail>,
+        user: &User<Id>,
+        settings: &Settings,
+        ip_address: &str,
+        device_info: Option<&str>,
+    ) -> Result<(), TokenError> {
+        debug!("Sending welcome mail to {}", user.username);
+        let mail = Mail {
+            to: user.email.clone(),
+            subject: settings
+                .enrollment_welcome_email_subject
+                .clone()
+                .unwrap_or_else(|| WELCOME_EMAIL_SUBJECT.to_string()),
+            content: self
+                .get_welcome_email_content(&mut *transaction, ip_address, device_info)
+                .await?,
+            attachments: Vec::new(),
+            result_tx: None,
+        };
+        match mail_tx.send(mail) {
+            Ok(()) => {
+                info!("Sent enrollment welcome mail to {}", user.username);
+                Ok(())
+            }
+            Err(err) => {
+                error!("Error sending welcome mail: {err}");
+                Err(TokenError::NotificationError(err.to_string()))
+            }
+        }
+    }
+
+    // Notify admin that a user has completed enrollment
+    pub fn send_admin_notification(
+        mail_tx: &UnboundedSender<Mail>,
+        admin: &User<Id>,
+        user: &User<Id>,
+        ip_address: &str,
+        device_info: Option<&str>,
+    ) -> Result<(), TokenError> {
+        debug!(
+            "Sending enrollment success notification for user {} to {}",
+            user.username, admin.username
+        );
+        let mail = Mail {
+            to: admin.email.clone(),
+            subject: "[defguard] User enrollment completed".into(),
+            content: templates::enrollment_admin_notification(
+                &user.clone().into(),
+                &admin.clone().into(),
+                ip_address,
+                device_info,
+            )?,
+            attachments: Vec::new(),
+            result_tx: None,
+        };
+        match mail_tx.send(mail) {
+            Ok(()) => {
+                info!(
+                    "Sent enrollment success notification for user {} to {}",
+                    user.username, admin.username
+                );
+                Ok(())
+            }
+            Err(err) => {
+                error!("Error sending welcome mail: {err}");
+                Err(TokenError::NotificationError(err.to_string()))
+            }
+        }
+    }
 }
 
 pub fn enrollment_welcome_message(settings: &Settings) -> Result<String, TokenError> {

crates/defguard_core/src/enterprise/db/models/openid_provider.rs

@@ -208,10 +208,7 @@ impl OpenIdProvider {
 }
 
 impl OpenIdProvider<Id> {
-    pub(crate) async fn find_by_name<'e, E>(
-        executor: E,
-        name: &str,
-    ) -> Result<Option<Self>, SqlxError>
+    pub async fn find_by_name<'e, E>(executor: E, name: &str) -> Result<Option<Self>, SqlxError>
     where
         E: PgExecutor<'e>,
     {
@@ -230,7 +227,7 @@ impl OpenIdProvider<Id> {
         .await
     }
 
-    pub(crate) async fn get_current<'e, E>(executor: E) -> Result<Option<Self>, SqlxError>
+    pub async fn get_current<'e, E>(executor: E) -> Result<Option<Self>, SqlxError>
     where
         E: PgExecutor<'e>,
     {

crates/defguard_core/src/enterprise/directory_sync/mod.rs

@@ -409,7 +409,7 @@ pub(crate) async fn test_directory_sync_connection(
 }
 
 /// Sync user groups with the directory if directory sync is enabled and configured, skip otherwise
-pub(crate) async fn sync_user_groups_if_configured(
+pub async fn sync_user_groups_if_configured(
     user: &User<Id>,
     pool: &PgPool,
     wg_tx: &Sender<GatewayEvent>,

crates/defguard_core/src/enterprise/grpc/desktop_client_mfa.rs

@@ -10,7 +10,7 @@ use crate::{
     },
     events::{BidiRequestContext, BidiStreamEvent, BidiStreamEventType, DesktopClientMfaEvent},
     grpc::{
-        client_mfa::{ClientLoginSession, ClientMfaServer},
+        proxy::client_mfa::{ClientLoginSession, ClientMfaServer},
         utils::parse_client_ip_agent,
     },
 };

crates/defguard_core/src/enterprise/handlers/openid_login.rs

@@ -31,7 +31,7 @@ static CSRF_COOKIE_NAME: &str = "csrf";
 static NONCE_COOKIE_NAME: &str = "nonce";
 // The select_account prompt is not supported by all providers, most notably not by JumpCloud.
 // Currently it's only enabled for Google, as it was tested to work there.
-pub(crate) const SELECT_ACCOUNT_SUPPORTED_PROVIDERS: &[&str] = &["Google"];
+pub const SELECT_ACCOUNT_SUPPORTED_PROVIDERS: &[&str] = &["Google"];
 
 use super::LicenseInfo;
 use crate::{
@@ -126,7 +126,7 @@ async fn get_provider_metadata(url: &str) -> Result<CoreProviderMetadata, WebErr
 
 /// Build a state with optional embedded data. Useful for passing additional information around the
 /// authentication flow.
-pub(crate) fn build_state(state_data: Option<String>) -> CsrfToken {
+pub fn build_state(state_data: Option<String>) -> CsrfToken {
     let csrf_token = CsrfToken::new_random();
     if let Some(data) = state_data {
         let combined = format!("{}.{data}", csrf_token.secret());
@@ -155,7 +155,7 @@ pub(crate) fn extract_state_data(state: &str) -> Option<String> {
 
 /// Build OpenID Connect client.
 /// `url`: redirect/callback URL
-pub(crate) async fn make_oidc_client(
+pub async fn make_oidc_client(
     url: Url,
     provider: &OpenIdProvider<Id>,
 ) -> Result<
@@ -186,7 +186,7 @@ pub(crate) async fn make_oidc_client(
 }
 
 /// Get or create `User` from OpenID claims.
-pub(crate) async fn user_from_claims(
+pub async fn user_from_claims(
     pool: &PgPool,
     nonce: Nonce,
     code: AuthorizationCode,

crates/defguard_core/src/enterprise/ldap/utils.rs

@@ -57,7 +57,7 @@ pub(crate) async fn login_through_ldap(
 }
 
 /// Convenience wrapper around [`ldap_update_users_state`] to update a single user.
-pub(crate) async fn ldap_update_user_state(user: &mut User<Id>, pool: &PgPool) {
+pub async fn ldap_update_user_state(user: &mut User<Id>, pool: &PgPool) {
     let vec = vec![user];
     Box::pin(ldap_update_users_state(vec, pool)).await;
 }
@@ -77,7 +77,7 @@ pub(crate) async fn ldap_update_users_state(users: Vec<&mut User<Id>>, pool: &Pg
 /// This will set the `ldap_pass_randomized` field to `true` in the user.
 ///
 /// If the user already exists, the creation will be skipped.
-pub(crate) async fn ldap_add_user(user: &mut User<Id>, password: Option<&str>, pool: &PgPool) {
+pub async fn ldap_add_user(user: &mut User<Id>, password: Option<&str>, pool: &PgPool) {
     let _: Result<(), LdapError> = with_ldap_status(pool, async {
         debug!("Creating user {user} in LDAP");
         if !ldap_sync_allowed_for_user(user, pool).await? {
@@ -273,7 +273,7 @@ pub(crate) async fn ldap_remove_users_from_groups(
     .await;
 }
 
-pub(crate) async fn ldap_change_password(user: &mut User<Id>, password: &str, pool: &PgPool) {
+pub async fn ldap_change_password(user: &mut User<Id>, password: &str, pool: &PgPool) {
     let _: Result<(), LdapError> = with_ldap_status(pool, async {
         debug!("Changing password for user {user} in LDAP");
         if !ldap_sync_allowed_for_user(user, pool).await? {

crates/defguard_core/src/enterprise/mod.rs

@@ -16,12 +16,12 @@ use limits::get_counts;
 use crate::enterprise::license::LicenseTier;
 
 /// Helper function to gate features which require a base license (Team or Business tier)
-pub(crate) fn is_business_license_active() -> bool {
+pub fn is_business_license_active() -> bool {
     is_license_tier_active(LicenseTier::Business)
 }
 
 /// Helper function to gate features which require an Enterprise tier license
-pub(crate) fn is_enterprise_license_active() -> bool {
+pub fn is_enterprise_license_active() -> bool {
     is_license_tier_active(LicenseTier::Enterprise)
 }
 

crates/defguard_core/src/grpc/client_version.rs

@@ -46,7 +46,7 @@ pub(crate) fn parse_client_version_platform(
 
 /// Represents a client feature that may have minimum version and OS family requirements.
 #[derive(Debug)]
-pub(crate) enum ClientFeature {
+pub enum ClientFeature {
     ServiceLocations,
 }
 
@@ -63,7 +63,7 @@ impl ClientFeature {
         }
     }
 
-    pub(crate) fn is_supported_by_device(&self, info: Option<&DeviceInfo>) -> bool {
+    pub fn is_supported_by_device(&self, info: Option<&DeviceInfo>) -> bool {
         let (version, platform) = parse_client_version_platform(info);
 
         // No minimum version = matches all