Gateway wizard

Cargo.toml

@@ -7,7 +7,7 @@ edition = "2024"
 axum = "0.8"
 base64 = "0.22"
 clap = { version = "4.5", features = ["derive", "env"] }
-defguard_version = { git = "https://github.com/DefGuard/defguard.git", rev = "640bae9a0aea1e11395f0a29fb8c84eeefd7f115" }
+defguard_version = { git = "https://github.com/DefGuard/defguard.git", rev = "5be16525f5208739fd79384b30d8ac5056ffdb2f" }
 defguard_wireguard_rs = { git = "https://github.com/DefGuard/wireguard-rs", rev = "d0b01eabca015ea6c7ddf4e255a0228074684e96" }
 defguard_certs = { git = "https://github.com/DefGuard/defguard.git", rev = "290bdee718f51179c71e07f3bce3f8a0cbfb9379" }
 env_logger = "0.11"
@@ -35,6 +35,8 @@ tonic = { version = "0.14", default-features = false, features = [
 tracing = "0.1"
 tonic-prost = "0.14"
 tower = "0.5"
+chrono = "0.4.43"
+tracing-subscriber = "0.3.22"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 nftnl = { git = "https://github.com/DefGuard/nftnl-rs.git", rev = "1a1147271f43b9d7182a114bb056a5224c35d38f" }

Dockerfile

@@ -7,7 +7,7 @@ RUN cargo build --release
 
 FROM public.ecr.aws/docker/library/debian:13-slim
 RUN apt-get update && apt-get -y --no-install-recommends install \
-    iproute2 wireguard-tools sudo ca-certificates iptables ebtables nftables && \
+    iproute2 wireguard-tools sudo ca-certificates iptables ebtables nftables lsb-release && \
     apt-get clean && rm -rf /var/lib/apt/lists/*
 WORKDIR /app
 COPY --from=builder /app/target/release/defguard-gateway /usr/local/bin

src/lib.rs

@@ -27,6 +27,7 @@ use syslog::{BasicLogger, Facility, Formatter3164};
 use tokio::sync::oneshot;
 
 pub mod enterprise;
+pub mod logging;
 pub mod setup;
 
 pub const VERSION: &str = concat!(env!("CARGO_PKG_VERSION"), "+", env!("VERGEN_GIT_SHA"));

src/logging.rs

@@ -0,0 +1,75 @@
+use defguard_version::Version;
+use tokio::sync::mpsc::Sender;
+use tracing::{Event, Subscriber};
+use tracing_subscriber::{Layer, layer::SubscriberExt, util::SubscriberInitExt};
+
+use crate::proto::gateway::LogEntry;
+
+pub fn init_tracing(own_version: &Version, level: &str, logs_tx: Option<Sender<LogEntry>>) {
+    let subscriber = tracing_subscriber::registry();
+    let subscriber =
+        defguard_version::tracing::with_version_formatters(own_version, level, subscriber);
+
+    if let Some(tx) = logs_tx {
+        let sender_layer = LogSenderLayer::new(tx);
+        subscriber.with(sender_layer).init();
+    } else {
+        subscriber.init();
+    }
+
+    info!("Tracing initialized");
+}
+
+/// A tracing layer that sends log entries to a gRPC logs channel.
+pub struct LogSenderLayer {
+    logs_tx: Sender<LogEntry>,
+}
+
+impl LogSenderLayer {
+    #[must_use]
+    pub const fn new(logs_tx: Sender<LogEntry>) -> Self {
+        Self { logs_tx }
+    }
+}
+
+impl<S> Layer<S> for LogSenderLayer
+where
+    S: Subscriber,
+{
+    fn on_event(&self, event: &Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) {
+        if self.logs_tx.is_closed() {
+            return;
+        }
+
+        let mut visitor = LogVisitor::default();
+        event.record(&mut visitor);
+
+        let entry = LogEntry {
+            level: format!("{:?}", event.metadata().level()),
+            target: event.metadata().target().to_string(),
+            message: visitor.message,
+            timestamp: chrono::Utc::now().to_rfc3339(),
+            fields: visitor.fields,
+        };
+
+        // Drop the buffer overflow error for now
+        let _ = self.logs_tx.try_send(entry);
+    }
+}
+
+#[derive(Default)]
+struct LogVisitor {
+    message: String,
+    fields: std::collections::HashMap<String, String>,
+}
+
+impl tracing::field::Visit for LogVisitor {
+    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
+        if field.name() == "message" {
+            self.message = format!("{value:?}");
+        } else {
+            self.fields
+                .insert(field.name().to_string(), format!("{value:?}"));
+        }
+    }
+}

src/main.rs

@@ -13,14 +13,15 @@ use defguard_gateway::{
     execute_command,
     gateway::{Gateway, GatewayServer, TlsConfig, run_stats},
     init_syslog,
+    logging::init_tracing,
     server::run_server,
     setup::GatewaySetupServer,
 };
 use defguard_version::Version;
 #[cfg(not(any(target_os = "macos", target_os = "netbsd")))]
 use defguard_wireguard_rs::Kernel;
 use defguard_wireguard_rs::{Userspace, WGApi};
-use tokio::task::JoinSet;
+use tokio::{sync::mpsc, task::JoinSet};
 
 #[tokio::main]
 async fn main() -> Result<(), GatewayError> {
@@ -35,15 +36,35 @@ async fn main() -> Result<(), GatewayError> {
         file.write_all(pid.to_string().as_bytes())?;
     }
 
+    let cert_dir = &config.cert_dir;
+    if !cert_dir.exists() {
+        tokio::fs::create_dir_all(cert_dir).await?;
+    }
+
+    let (grpc_cert, grpc_key) = (
+        read_to_string(cert_dir.join(GRPC_CERT_NAME)).ok(),
+        read_to_string(cert_dir.join(GRPC_KEY_NAME)).ok(),
+    );
+
+    let needs_setup = grpc_cert.is_none() || grpc_key.is_none();
+
+    // TODO: The channel size may need to be adjusted or some other approach should be used
+    // to avoid dropping log messages.
+    let (logs_tx, logs_rx) = if needs_setup {
+        let (logs_tx, logs_rx) = mpsc::channel(200);
+        (Some(logs_tx), Some(logs_rx))
+    } else {
+        (None, None)
+    };
+
     // setup logging
     if config.use_syslog {
         if let Err(error) = init_syslog(&config, pid) {
             log::error!("Unable to initialize syslog. Is the syslog daemon running?");
             return Err(error);
         }
     } else {
-        let version = Version::parse(VERSION)?;
-        defguard_version::tracing::init(version, &config.log_level)?;
+        init_tracing(&Version::parse(VERSION)?, &config.log_level, logs_tx);
     }
 
     if let Some(pre_up) = &config.pre_up {
@@ -86,28 +107,19 @@ async fn main() -> Result<(), GatewayError> {
     let gateway = Arc::new(Mutex::new(gateway));
     tasks.spawn(run_stats(Arc::clone(&gateway), config.stats_period()));
 
-    let cert_dir = &config.cert_dir;
-    if !cert_dir.exists() {
-        tokio::fs::create_dir_all(cert_dir).await?;
-    }
-    let tls_config = if let (Some(cert), Some(key)) = (
-        read_to_string(cert_dir.join(GRPC_CERT_NAME)).ok(),
-        read_to_string(cert_dir.join(GRPC_KEY_NAME)).ok(),
-    ) {
-        log::info!(
-            "Using existing gRPC TLS certificates from {}",
-            cert_dir.display()
-        );
-        TlsConfig {
-            grpc_cert_pem: cert,
-            grpc_key_pem: key,
-        }
-    } else {
+    let tls_config = if needs_setup {
         log::info!(
             "gRPC TLS certificates not found in {}. They will be generated during setup.",
             cert_dir.display()
         );
-        let setup_server = GatewaySetupServer::default();
+
+        let Some(logs_rx) = logs_rx else {
+            return Err(GatewayError::SetupError(
+                "Logs receiver channel is missing during gateway setup".to_string(),
+            ));
+        };
+
+        let setup_server = GatewaySetupServer::new(Arc::new(tokio::sync::Mutex::new(logs_rx)));
         let tls_config = setup_server.await_setup(config.clone()).await?;
 
         let cert_path = cert_dir.join(GRPC_CERT_NAME);
@@ -120,6 +132,19 @@ async fn main() -> Result<(), GatewayError> {
         );
 
         tls_config
+    } else if let (Some(cert), Some(key)) = (grpc_cert, grpc_key) {
+        log::info!(
+            "Using existing gRPC TLS certificates from {}",
+            cert_dir.display()
+        );
+        TlsConfig {
+            grpc_cert_pem: cert,
+            grpc_key_pem: key,
+        }
+    } else {
+        return Err(GatewayError::SetupError(
+            "gRPC TLS certificates are missing after setup".to_string(),
+        ));
     };
 
     // Launch gRPC server.