Gateway events

crates/defguard/src/main.rs

@@ -41,7 +41,7 @@ use defguard_core::{
     version::IncompatibleComponents,
 };
 use defguard_event_logger::run_event_logger;
-use defguard_gateway_manager::{GatewayManager, GatewayTxSet};
+use defguard_gateway_manager::{GatewayTxSet, manager::GatewayManager};
 use defguard_proxy_manager::{ProxyManager, ProxyTxSet};
 use defguard_session_manager::{events::SessionManagerEvent, run_session_manager};
 use defguard_setup::{
@@ -300,8 +300,7 @@ async fn main() -> Result<(), anyhow::Error> {
 
     // setup communication channels for services
     let (webhook_tx, webhook_rx) = unbounded_channel::<AppEvent>();
-    // RX is discarded here since it can be derived from TX later on
-    let (gateway_tx, _gateway_rx) = broadcast::channel::<GatewayCommand>(256);
+    let (gateway_tx, gateway_rx) = unbounded_channel::<GatewayCommand>();
     let (peer_stats_tx, peer_stats_rx) = unbounded_channel::<PeerStatsUpdate>();
 
     let worker_state = Arc::new(Mutex::new(WorkerState::new(webhook_tx.clone())));
@@ -345,8 +344,8 @@ async fn main() -> Result<(), anyhow::Error> {
 
     let mut gateway_manager = GatewayManager::new(
         pool.clone(),
-        GatewayTxSet::new(gateway_tx.clone(), peer_stats_tx)
-            .with_connection_events(gateway_connection_event_tx),
+        GatewayTxSet::new(peer_stats_tx).with_connection_events(gateway_connection_event_tx),
+        gateway_rx,
     );
 
     debug!("Resetting proxy connection state on startup");

crates/defguard_common/src/db/models/group.rs

@@ -232,7 +232,7 @@ mod test {
         assert!(!user.remove_from_group(&pool, &group).await.unwrap());
 
         let members = group.member_usernames(&pool).await.unwrap();
-        assert!(members.is_empty());
+        assert_eq!(members, [] as [String; 0]);
     }
 
     #[sqlx::test]

crates/defguard_common/src/db/models/settings/tests.rs

@@ -1,5 +1,3 @@
-use std::str::FromStr;
-
 use humantime::Duration;
 use reqwest::Url;
 use rsa::{RsaPrivateKey, pkcs8::EncodePrivateKey};
@@ -29,7 +27,7 @@ fn test_smtp_config() {
 
     // add auth info
     settings.smtp.user = Some("smtp_user".into());
-    settings.smtp.password = Some(SecretStringWrapper::from_str("hunter2").unwrap());
+    settings.smtp.password = Some(SecretStringWrapper::from("hunter2"));
     assert!(settings.smtp_configured());
 }
 

crates/defguard_common/src/db/models/user.rs

@@ -1349,8 +1349,6 @@ impl Distribution<User<NoId>> for Standard {
 
 #[cfg(test)]
 mod test {
-    use std::str::FromStr;
-
     use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
 
     use super::*;
@@ -1829,7 +1827,7 @@ mod test {
         settings.smtp.sender = Some("noreply@example.com".into());
         settings.ldap_url = Some("ldap://localhost".into());
         settings.ldap_bind_username = Some("cn=admin,dc=example,dc=com".into());
-        settings.ldap_bind_password = Some(SecretStringWrapper::from_str("secret").unwrap());
+        settings.ldap_bind_password = Some(SecretStringWrapper::from("secret"));
         settings.ldap_username_attr = Some("uid".into());
         settings.ldap_user_search_base = Some("ou=users,dc=example,dc=com".into());
         settings.ldap_user_obj_class = Some("inetOrgPerson".into());

crates/defguard_common/src/db/models/wireguard/tests.rs

@@ -3,7 +3,7 @@ use std::{assert_matches, net::Ipv6Addr, str::FromStr};
 use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
 
 use super::*;
-use crate::db::setup_pool;
+use crate::db::{models::device, setup_pool};
 
 #[test]
 fn test_set_address() {
@@ -135,7 +135,7 @@ async fn test_get_allowed_devices_for_user(_: PgPoolOptions, options: PgConnectO
         .get_allowed_devices_for_user(&mut pool.acquire().await.unwrap(), Id::from(999))
         .await
         .unwrap();
-    assert!(devices.is_empty());
+    assert_eq!(devices, [] as [device::Device<i64>; 0]);
 }
 
 #[sqlx::test]
@@ -223,7 +223,7 @@ async fn test_get_allowed_devices_for_user_with_groups(
         .get_allowed_devices_for_user(&mut transaction, user2.id)
         .await
         .unwrap();
-    assert!(devices.is_empty());
+    assert_eq!(devices, [] as [device::Device<i64>; 0]);
 }
 
 #[sqlx::test]

crates/defguard_common/src/gateway_event.rs

@@ -1,10 +1,10 @@
 //! Gateway command types and helpers for communicating with the gateway manager service.
 //!
 //! [`GatewayCommand`] is the primary type sent from core to the gateway manager over
-//! an in-process broadcast channel. The gateway manager converts each command to the
+//! an in-process MPSC channel. The gateway manager converts each command to the
 //! appropriate protobuf wire message before forwarding it to the gateway daemon.
 
-use tokio::sync::broadcast::Sender;
+use tokio::sync::mpsc::UnboundedSender;
 use tracing::{debug, error};
 
 use crate::{
@@ -25,9 +25,8 @@ use crate::{
 /// protobuf wire types happens at the serialization boundary in the gateway manager.
 #[derive(Clone, Debug)]
 pub enum GatewayCommand {
-    NetworkCreated(Id, WireguardNetwork<Id>),
+    NetworkCreated(WireguardNetwork<Id>),
     NetworkModified(
-        Id,
         WireguardNetwork<Id>,
         Vec<WireguardPeer>,
         Option<FirewallConfig>,
@@ -45,7 +44,7 @@ pub enum GatewayCommand {
 /// Sends a [`GatewayCommand`] to the gateway manager service.
 ///
 /// In API handler context prefer `AppState::send_gateway_command`.
-pub fn send_gateway_command(command: GatewayCommand, gateway_tx: &Sender<GatewayCommand>) {
+pub fn send_gateway_command(command: GatewayCommand, gateway_tx: &UnboundedSender<GatewayCommand>) {
     debug!("Sending the following command to Gateway Manager: {command:?}");
     if let Err(err) = gateway_tx.send(command) {
         error!("Error sending gateway command: {err}");
@@ -57,7 +56,7 @@ pub fn send_gateway_command(command: GatewayCommand, gateway_tx: &Sender<Gateway
 /// In API handler context prefer `AppState::send_multiple_gateway_commands`.
 pub fn send_multiple_gateway_commands(
     commands: Vec<GatewayCommand>,
-    gateway_tx: &Sender<GatewayCommand>,
+    gateway_tx: &UnboundedSender<GatewayCommand>,
 ) {
     debug!("Sending {} gateway commands", commands.len());
     for command in commands {

crates/defguard_common/src/secret.rs

@@ -1,4 +1,4 @@
-use std::{convert::Infallible, error::Error, str::FromStr};
+use std::error::Error;
 
 use secrecy::{ExposeSecret, SecretString};
 use serde::Deserialize;
@@ -31,11 +31,9 @@ impl From<String> for SecretStringWrapper {
     }
 }
 
-impl FromStr for SecretStringWrapper {
-    type Err = Infallible;
-
-    fn from_str(src: &str) -> Result<Self, Self::Err> {
-        Ok(Self(SecretString::from(src)))
+impl From<&str> for SecretStringWrapper {
+    fn from(value: &str) -> Self {
+        Self(SecretString::from(value))
     }
 }
 

crates/defguard_common/src/types/user_info.rs

@@ -311,8 +311,14 @@ mod test {
 
         assert_eq!(group1.member_usernames(&pool).await.unwrap(), ["hpotter"]);
         assert_eq!(group3.member_usernames(&pool).await.unwrap(), ["hpotter"]);
-        assert!(group2.member_usernames(&pool).await.unwrap().is_empty());
-        assert!(group4.member_usernames(&pool).await.unwrap().is_empty());
+        assert_eq!(
+            group2.member_usernames(&pool).await.unwrap(),
+            [] as [std::string::String; 0]
+        );
+        assert_eq!(
+            group4.member_usernames(&pool).await.unwrap(),
+            [] as [std::string::String; 0]
+        );
     }
 
     // Admin updating another user must be able to change all profile

crates/defguard_core/Cargo.toml

@@ -94,7 +94,8 @@ webauthn-rs-proto.workspace = true
 x25519-dalek.workspace = true
 ammonia = "4.1"
 regex = "1.10"
-uaparser = "0.6"
+serde-saphyr = "1.3"
+ua-parser = "0.2"
 async-stream = "0.3"
 
 [features]
@@ -116,7 +117,7 @@ defguard_common = { workspace = true, features = ["test-support"] }
 claims.workspace = true
 ed25519-dalek.workspace = true
 getrandom.workspace = true
-hyper-util = "0.1"
+hyper-util.workspace = true
 reqwest = { version = "0.12", features = [
     "cookies",
     "json",

crates/defguard_core/src/appstate.rs

@@ -7,10 +7,7 @@ use reqwest::Client;
 use serde_json::json;
 use sqlx::PgPool;
 use tokio::{
-    sync::{
-        broadcast::Sender,
-        mpsc::{UnboundedReceiver, UnboundedSender},
-    },
+    sync::{broadcast, mpsc},
     task::spawn,
 };
 
@@ -27,15 +24,15 @@ const X_DEFGUARD_EVENT: &str = "x-defguard-event";
 #[derive(Clone)]
 pub struct AppState {
     pub pool: PgPool,
-    tx: UnboundedSender<AppEvent>,
-    pub gateway_tx: Sender<GatewayCommand>,
-    pub web_reload_tx: tokio::sync::broadcast::Sender<()>,
+    tx: mpsc::UnboundedSender<AppEvent>,
+    pub gateway_tx: mpsc::UnboundedSender<GatewayCommand>,
+    pub web_reload_tx: broadcast::Sender<()>,
     key: Key,
-    pub event_tx: UnboundedSender<ApiEvent>,
-    pub ldap_tx: UnboundedSender<LdapSyncEventType>,
-    pub dirsync_tx: UnboundedSender<DirectorySyncEvent>,
+    pub event_tx: mpsc::UnboundedSender<ApiEvent>,
+    pub ldap_tx: mpsc::UnboundedSender<LdapSyncEventType>,
+    pub dirsync_tx: mpsc::UnboundedSender<DirectorySyncEvent>,
     pub incompatible_components: Arc<RwLock<IncompatibleComponents>>,
-    pub proxy_control_tx: tokio::sync::mpsc::Sender<ProxyControlMessage>,
+    pub proxy_control_tx: mpsc::Sender<ProxyControlMessage>,
     /// Reflects whether the HTTP server is currently running with TLS
     pub tls_active: Arc<AtomicBool>,
 }
@@ -50,7 +47,7 @@ impl AppState {
     }
 
     /// Handle webhook events
-    async fn handle_triggers(pool: PgPool, mut rx: UnboundedReceiver<AppEvent>) {
+    async fn handle_triggers(pool: PgPool, mut rx: mpsc::UnboundedReceiver<AppEvent>) {
         let reqwest_client = Client::builder().user_agent("reqwest").build().unwrap();
         while let Some(msg) = rx.recv().await {
             debug!("WebHook triggered");
@@ -116,16 +113,16 @@ impl AppState {
     /// Create application state
     pub fn new(
         pool: PgPool,
-        tx: UnboundedSender<AppEvent>,
-        rx: UnboundedReceiver<AppEvent>,
-        gateway_tx: Sender<GatewayCommand>,
-        web_reload_tx: tokio::sync::broadcast::Sender<()>,
+        tx: mpsc::UnboundedSender<AppEvent>,
+        rx: mpsc::UnboundedReceiver<AppEvent>,
+        gateway_tx: mpsc::UnboundedSender<GatewayCommand>,
+        web_reload_tx: broadcast::Sender<()>,
         key: Key,
-        event_tx: UnboundedSender<ApiEvent>,
-        ldap_tx: UnboundedSender<LdapSyncEventType>,
-        dirsync_tx: UnboundedSender<DirectorySyncEvent>,
+        event_tx: mpsc::UnboundedSender<ApiEvent>,
+        ldap_tx: mpsc::UnboundedSender<LdapSyncEventType>,
+        dirsync_tx: mpsc::UnboundedSender<DirectorySyncEvent>,
         incompatible_components: Arc<RwLock<IncompatibleComponents>>,
-        proxy_control_tx: tokio::sync::mpsc::Sender<ProxyControlMessage>,
+        proxy_control_tx: mpsc::Sender<ProxyControlMessage>,
         tls_active: Arc<AtomicBool>,
     ) -> Self {
         spawn(Self::handle_triggers(pool.clone(), rx));

crates/defguard_core/src/enterprise/allowed_ips/tests.rs

@@ -191,7 +191,7 @@ async fn test_explicit_user_allow(_: PgPoolOptions, options: PgConnectOptions) {
     let result = get_allowed_ips_from_acl_rules(&mut conn, &location, &other_user)
         .await
         .unwrap();
-    assert!(result.is_empty());
+    assert_eq!(result, [] as [ipnetwork::IpNetwork; 0]);
 }
 
 #[sqlx::test]
@@ -253,7 +253,7 @@ async fn test_group_membership_allow(_: PgPoolOptions, options: PgConnectOptions
     let result = get_allowed_ips_from_acl_rules(&mut conn, &location, &non_member)
         .await
         .unwrap();
-    assert!(result.is_empty());
+    assert_eq!(result, [] as [ipnetwork::IpNetwork; 0]);
 }
 
 #[sqlx::test]
@@ -546,7 +546,7 @@ async fn test_non_matching_rule_excluded(_: PgPoolOptions, options: PgConnectOpt
     let result = get_allowed_ips_from_acl_rules(&mut conn, &location, &user)
         .await
         .unwrap();
-    assert!(result.is_empty());
+    assert_eq!(result, [] as [ipnetwork::IpNetwork; 0]);
 }
 
 #[sqlx::test]
@@ -699,7 +699,7 @@ async fn test_predefined_destination_included(_: PgPoolOptions, options: PgConne
         .await
         .unwrap();
 
-    assert_eq!(result, vec!["172.16.0.0/12".parse().unwrap()]);
+    assert_eq!(result, ["172.16.0.0/12".parse().unwrap()]);
 }
 
 #[sqlx::test]

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

@@ -17,7 +17,7 @@ use sqlx::{
     query, query_as, query_scalar,
 };
 use thiserror::Error;
-use tokio::sync::broadcast::Sender;
+use tokio::sync::mpsc::UnboundedSender;
 
 use crate::{
     enterprise::{
@@ -555,7 +555,7 @@ impl AclRule {
         rules: &[Id],
         actor: &str,
         pool: &PgPool,
-        gateway_tx: &Sender<GatewayCommand>,
+        gateway_tx: &UnboundedSender<GatewayCommand>,
     ) -> Result<(), AclError> {
         debug!("Applying {} ACL rules: {rules:?}", rules.len());
         let mut transaction = pool.begin().await?;
@@ -575,8 +575,7 @@ impl AclRule {
         }
         info!("Applied {} ACL rules: {rules:?}", rules.len());
 
-        let affected_locations: Vec<WireguardNetwork<Id>> =
-            affected_locations.into_iter().collect();
+        let affected_locations = affected_locations.into_iter().collect::<Vec<_>>();
         debug!(
             "{} locations affected by applied ACL rules. Sending gateway firewall update events \
             for each location",
@@ -1830,7 +1829,7 @@ impl AclAlias {
         kind: AliasKind,
         actor: &str,
         pool: &PgPool,
-        gateway_tx: &Sender<GatewayCommand>,
+        gateway_tx: &UnboundedSender<GatewayCommand>,
     ) -> Result<(), AclError> {
         debug!(
             "Applying {} ACL aliases of kind {kind:?}: {aliases:?}",

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

@@ -12,7 +12,7 @@ use paste::paste;
 use reqwest::header::AUTHORIZATION;
 use sqlx::{PgConnection, PgPool};
 use thiserror::Error;
-use tokio::sync::{broadcast::Sender, mpsc::UnboundedSender};
+use tokio::sync::mpsc::UnboundedSender;
 
 use super::{
     db::models::{
@@ -351,7 +351,7 @@ async fn sync_user_groups<T: DirectorySync>(
     directory_sync: &T,
     user: &User<Id>,
     pool: &PgPool,
-    gateway_tx: &Sender<GatewayCommand>,
+    gateway_tx: &UnboundedSender<GatewayCommand>,
     ldap_tx: &UnboundedSender<LdapSyncEventType>,
     dirsync_tx: &UnboundedSender<DirectorySyncEvent>,
     provider_name: &str,
@@ -467,7 +467,7 @@ pub(crate) async fn test_directory_sync_connection(
 pub async fn sync_user_groups_if_configured(
     user: &User<Id>,
     pool: &PgPool,
-    gateway_tx: &Sender<GatewayCommand>,
+    gateway_tx: &UnboundedSender<GatewayCommand>,
     ldap_tx: &UnboundedSender<LdapSyncEventType>,
     dirsync_tx: &UnboundedSender<DirectorySyncEvent>,
 ) -> Result<(), DirectorySyncError> {
@@ -567,7 +567,7 @@ async fn create_and_add_to_group(
 async fn sync_all_users_groups<T: DirectorySync>(
     directory_sync: &T,
     pool: &PgPool,
-    gateway_tx: &Sender<GatewayCommand>,
+    gateway_tx: &UnboundedSender<GatewayCommand>,
     ldap_tx: &UnboundedSender<LdapSyncEventType>,
     dirsync_tx: &UnboundedSender<DirectorySyncEvent>,
     provider_name: &str,
@@ -739,7 +739,7 @@ fn is_directory_sync_enabled(provider: Option<&OpenIdProvider<Id>>) -> bool {
 
 async fn sync_all_users_state(
     pool: &PgPool,
-    gateway_tx: &Sender<GatewayCommand>,
+    gateway_tx: &UnboundedSender<GatewayCommand>,
     ldap_tx: &UnboundedSender<LdapSyncEventType>,
     dirsync_tx: &UnboundedSender<DirectorySyncEvent>,
     all_users: &[DirectoryUser],
@@ -1292,7 +1292,7 @@ pub(crate) async fn get_directory_sync_interval(pool: &PgPool) -> u64 {
 // Performs the directory sync job. This function is called by the utility thread.
 pub(crate) async fn do_directory_sync(
     pool: &PgPool,
-    gateway_tx: &Sender<GatewayCommand>,
+    gateway_tx: &UnboundedSender<GatewayCommand>,
     ldap_tx: &UnboundedSender<LdapSyncEventType>,
     dirsync_tx: &UnboundedSender<DirectorySyncEvent>,
 ) -> Result<(), DirectorySyncError> {