Chore/tunnel handshake logging

wallguard-server/README.md

@@ -9,6 +9,7 @@
 | `DATASTORE_HOST` | Hostname or IP of the datastore |
 | `DATASTORE_PORT` | Port of the datastore |
 | `DATASTORE_TLS` | Whether to use TLS / secure connection to datastore |
+| `STORE_TELEMETRY_IN_DATASTORE` | Set to `false` to accept connections and system resources data without saving it to the datastore (default `true`) |
 | `ROOT_ACCOUNT_ID` | Datastore root account ID |
 | `ROOT_ACCOUNT_SECRET` | Datastore root account secret |
 | `SYSTEM_ACCOUNT_ID` | Datastore system account ID |

wallguard-server/src/control_service/config.rs

@@ -28,3 +28,23 @@ impl ControlServiceConfig {
         Self::default()
     }
 }
+
+/// Reads `STORE_TELEMETRY_IN_DATASTORE`. When `false`, connections and system
+/// resources reported by agents are accepted but not written to the datastore.
+/// Defaults to `true`.
+pub fn store_telemetry_in_datastore() -> bool {
+    const NAME: &str = "STORE_TELEMETRY_IN_DATASTORE";
+
+    let Ok(raw) = std::env::var(NAME) else {
+        return true;
+    };
+
+    match raw.trim().to_lowercase().as_str() {
+        "true" | "1" | "yes" => true,
+        "false" | "0" | "no" => false,
+        _ => {
+            log::warn!("{NAME} is set to {raw:?}, which is not a valid boolean; using true");
+            true
+        }
+    }
+}

wallguard-server/src/control_service/rpc/handle_connections_data.rs

@@ -15,6 +15,15 @@ impl WallGuardService {
         let token =
             Token::from_jwt(&data.token).map_err(|_| Status::internal("Malformed JWT token"))?;
 
+        if !self.store_telemetry {
+            log::info!(
+                "Accepted {} connections from device {} (not stored: STORE_TELEMETRY_IN_DATASTORE=false)",
+                data.connections.len(),
+                token.account.device_id().unwrap_or("unknown")
+            );
+            return Ok(Response::new(()));
+        }
+
         let _ = self
             .ensure_device_exists_and_authrorized(&token)
             .await

wallguard-server/src/control_service/rpc/handle_system_resources_data.rs

@@ -13,6 +13,15 @@ impl WallGuardService {
         let token =
             Token::from_jwt(&data.token).map_err(|_| Status::internal("Malformed JWT token"))?;
 
+        if !self.store_telemetry {
+            log::info!(
+                "Accepted {} system resources from device {} (not stored: STORE_TELEMETRY_IN_DATASTORE=false)",
+                data.resources.len(),
+                token.account.device_id().unwrap_or("unknown")
+            );
+            return Ok(Response::new(()));
+        }
+
         let device = self
             .ensure_device_exists_and_authrorized(&token)
             .await

wallguard-server/src/control_service/service.rs

@@ -1,4 +1,5 @@
 use crate::app_context::AppContext;
+use crate::control_service::config::store_telemetry_in_datastore;
 use crate::traffic_handler::ip_info::ip_info_handler;
 use nullnet_liberror::{Error, ErrorHandler, Location, location};
 use std::net::{IpAddr, SocketAddr};
@@ -22,6 +23,7 @@ const IP_INFO_CACHE_SIZE: usize = 10_000;
 pub struct WallGuardService {
     pub(crate) context: AppContext,
     pub(crate) ip_info_tx: mpsc::Sender<Option<IpAddr>>,
+    pub(crate) store_telemetry: bool,
 }
 
 impl WallGuardService {
@@ -34,9 +36,18 @@ impl WallGuardService {
             ip_info_handler(&ip_info_rx, IP_INFO_CACHE_SIZE, &handle, ctx);
         });
 
+        let store_telemetry = store_telemetry_in_datastore();
+        if !store_telemetry {
+            log::warn!(
+                "STORE_TELEMETRY_IN_DATASTORE=false: connections and system resources \
+                 will be accepted but not saved to the datastore"
+            );
+        }
+
         Self {
             context,
             ip_info_tx,
+            store_telemetry,
         }
     }
 

wallguard-server/src/reverse_tunnel/mod.rs

@@ -6,8 +6,8 @@ use tokio::io::AsyncWriteExt;
 
 use tokio::sync::Mutex;
 use tokio::sync::oneshot;
-use tunnel_token::TokenHash;
 use tunnel_token::TunnelToken;
+use tunnel_token::{HandshakeFailure, TokenHash};
 
 mod config;
 mod tunnel_instance;
@@ -73,8 +73,8 @@ pub async fn run_tunnel_acceptor(context: AppContext) -> Result<(), Error> {
         .handle_err(location!())?;
 
     loop {
-        let mut stream = match listener.accept().await {
-            Ok((stream, _)) => stream,
+        let (mut stream, peer) = match listener.accept().await {
+            Ok(accepted) => accepted,
             Err(err) => {
                 log::error!("Tunnel acceptor: failed to accept connection: {err}");
                 tokio::time::sleep(ACCEPT_ERROR_BACKOFF).await;
@@ -90,13 +90,23 @@ pub async fn run_tunnel_acceptor(context: AppContext) -> Result<(), Error> {
              * Send Confirmation or Rejection message to the client
              */
 
-            let Ok(Ok(hash)) =
-                tokio::time::timeout(HANDSHAKE_TIMEOUT, TokenHash::read_from_stream(&mut stream))
-                    .await
-            else {
-                log::error!("Failed to read token hash from newely accepted TCP stream");
-                let _ = stream.shutdown().await;
-                return;
+            log::debug!("Tunnel acceptor: accepted connection from {peer}");
+
+            let hash = match TokenHash::read_from_stream(&mut stream, HANDSHAKE_TIMEOUT).await {
+                Ok(hash) => hash,
+                Err(err) => {
+                    // A connection that closes without sending anything is
+                    // usually a port probe or TCP health check, not an agent.
+                    if matches!(err.failure, HandshakeFailure::Closed) && err.received.is_empty() {
+                        log::info!(
+                            "Tunnel handshake from {peer}: connection closed without sending data"
+                        );
+                    } else {
+                        log::error!("Tunnel handshake from {peer} failed: {err}");
+                    }
+                    let _ = stream.shutdown().await;
+                    return;
+                }
             };
 
             let mut tunnel = TunnelInstance::from(stream);
@@ -105,12 +115,16 @@ pub async fn run_tunnel_acceptor(context: AppContext) -> Result<(), Error> {
                 Some(channel) => {
                     if let Err(mut tunnel) = channel.send(tunnel) {
                         let _ = tunnel.shutdown().await;
-                        log::error!("Failed to send tunnel instance");
+                        log::error!(
+                            "Failed to hand over tunnel from {peer}: the requester is no longer waiting"
+                        );
+                    } else {
+                        log::debug!("Tunnel acceptor: tunnel from {peer} established");
                     }
                 }
                 None => {
                     log::warn!(
-                        "Received tunnel connection with unknown token hash: {:?}",
+                        "Received tunnel connection from {peer} with unknown token hash: {:?}",
                         hash
                     );
 

wallguard-server/src/reverse_tunnel/tunnel_token.rs

@@ -1,4 +1,6 @@
 use nullnet_liberror::{Error, ErrorHandler, Location, location};
+use std::fmt::{Display, Formatter};
+use std::time::Duration;
 use tokio::io::AsyncReadExt;
 use tokio::net::TcpStream;
 
@@ -39,16 +41,84 @@ impl TokenHash {
     /// is a fixed-size SHA-256 hash that can be used to identify the reverse tunnel.
     ///
     /// # Errors
-    /// Returns an error if reading from the stream fails or fewer than 32 bytes are received.
-    pub async fn read_from_stream(stream: &mut TcpStream) -> Result<Self, Error> {
+    /// Returns a [`HandshakeError`] if the stream fails, closes before 32 bytes
+    /// are received, or `timeout` elapses. The error carries whatever bytes did
+    /// arrive, which helps identify what is connecting to the port.
+    pub async fn read_from_stream(
+        stream: &mut TcpStream,
+        timeout: Duration,
+    ) -> Result<Self, HandshakeError> {
         let mut hash = TokenHash::default();
+        let mut received = 0;
 
-        stream
-            .read_exact(&mut hash.digest)
-            .await
-            .handle_err(location!())?;
+        let result = tokio::time::timeout(timeout, async {
+            while received < TOKEN_HASH_SIZE {
+                match stream.read(&mut hash.digest[received..]).await {
+                    Ok(0) => return Err(HandshakeFailure::Closed),
+                    Ok(n) => received += n,
+                    Err(err) => return Err(HandshakeFailure::Io(err)),
+                }
+            }
+            Ok(())
+        })
+        .await;
+
+        let failure = match result {
+            Ok(Ok(())) => return Ok(hash),
+            Ok(Err(failure)) => failure,
+            Err(_) => HandshakeFailure::Timeout(timeout),
+        };
+
+        Err(HandshakeError {
+            failure,
+            received: hash.digest[..received].to_vec(),
+        })
+    }
+}
+
+#[derive(Debug)]
+pub enum HandshakeFailure {
+    Closed,
+    Timeout(Duration),
+    Io(std::io::Error),
+}
+
+/// Failure to read a token hash, along with the bytes received before it.
+#[derive(Debug)]
+pub struct HandshakeError {
+    pub failure: HandshakeFailure,
+    pub received: Vec<u8>,
+}
+
+impl Display for HandshakeError {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match &self.failure {
+            HandshakeFailure::Closed => write!(f, "peer closed the connection")?,
+            HandshakeFailure::Timeout(timeout) => write!(f, "timed out after {timeout:?}")?,
+            HandshakeFailure::Io(err) => write!(f, "I/O error: {err}")?,
+        }
+
+        write!(f, " after {}/{TOKEN_HASH_SIZE} bytes", self.received.len())?;
+
+        if !self.received.is_empty() {
+            // Printable preview makes common probes obvious, e.g. "GET / HTTP"
+            // from an HTTP health check or 0x16 0x03 from a TLS ClientHello.
+            let hex: String = self.received.iter().map(|b| format!("{b:02x}")).collect();
+            let text: String = self
+                .received
+                .iter()
+                .map(|&b| {
+                    if b.is_ascii_graphic() || b == b' ' {
+                        b as char
+                    } else {
+                        '.'
+                    }
+                })
+                .collect();
+            write!(f, " (hex: {hex}, text: {text:?})")?;
+        }
 
-        Ok(hash)
+        Ok(())
     }
 }
 
@@ -82,3 +152,54 @@ impl From<TunnelToken> for TokenHash {
         sha256_digest_bytes(&value.token).into()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::AsyncWriteExt;
+    use tokio::net::TcpListener;
+
+    async fn handshake(
+        payload: &'static [u8],
+        keep_open: bool,
+    ) -> Result<TokenHash, HandshakeError> {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+
+        let client = tokio::spawn(async move {
+            let mut stream = TcpStream::connect(addr).await.unwrap();
+            stream.write_all(payload).await.unwrap();
+            if keep_open {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            }
+        });
+
+        let (mut stream, _) = listener.accept().await.unwrap();
+        let result = TokenHash::read_from_stream(&mut stream, Duration::from_millis(200)).await;
+        client.await.unwrap();
+        result
+    }
+
+    #[tokio::test]
+    async fn reads_full_hash() {
+        let hash = handshake(&[7u8; TOKEN_HASH_SIZE], false).await.unwrap();
+        assert_eq!(hash, TokenHash::from([7u8; TOKEN_HASH_SIZE]));
+    }
+
+    #[tokio::test]
+    async fn reports_bytes_received_before_close() {
+        let err = handshake(b"GET / HTTP/1.1\r\n", false).await.unwrap_err();
+        assert!(matches!(err.failure, HandshakeFailure::Closed));
+        assert_eq!(err.received, b"GET / HTTP/1.1\r\n");
+        let msg = err.to_string();
+        assert!(msg.contains("16/32 bytes"), "{msg}");
+        assert!(msg.contains(r#"text: "GET / HTTP/1.1..""#), "{msg}");
+    }
+
+    #[tokio::test]
+    async fn reports_timeout() {
+        let err = handshake(b"ab", true).await.unwrap_err();
+        assert!(matches!(err.failure, HandshakeFailure::Timeout(_)));
+        assert_eq!(err.received, b"ab");
+    }
+}