From aabb6507d2e729dc11be094c577930a421c4b95b Mon Sep 17 00:00:00 2001 From: Anton Liashkevich Date: Tue, 29 Sep 2026 20:46:16 -0400 Subject: [PATCH 1/2] Log peer address and received bytes on tunnel handshake failures The "early eof" / "Failed to read token hash" errors gave no hint about what was connecting. Handshake failures now log the peer address, why the read failed (closed, timed out, I/O error), and how many bytes arrived, with a hex and text preview so HTTP or TLS probes are easy to spot. Connections that close without sending anything (port scans, TCP health checks) are logged at info instead of error. read_from_stream now owns the handshake timeout and no longer logs on its own, which removes the duplicate tunnel_token.rs:49 error line. Co-Authored-By: Claude Opus 5.5 --- wallguard-server/src/reverse_tunnel/mod.rs | 38 +++-- .../src/reverse_tunnel/tunnel_token.rs | 135 +++++++++++++++++- 2 files changed, 154 insertions(+), 19 deletions(-) diff --git a/wallguard-server/src/reverse_tunnel/mod.rs b/wallguard-server/src/reverse_tunnel/mod.rs index 7c775117..44bf4584 100644 --- a/wallguard-server/src/reverse_tunnel/mod.rs +++ b/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 ); diff --git a/wallguard-server/src/reverse_tunnel/tunnel_token.rs b/wallguard-server/src/reverse_tunnel/tunnel_token.rs index f0e15559..24c14fc9 100644 --- a/wallguard-server/src/reverse_tunnel/tunnel_token.rs +++ b/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 { + /// 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 { 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, +} + +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 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 { + 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"); + } +} From 0a2dd61022c7b883dd2c24d1c494bbaffae711dd Mon Sep 17 00:00:00 2001 From: Anton Liashkevich Date: Tue, 29 Sep 2026 20:49:32 -0400 Subject: [PATCH 2/2] Add STORE_TELEMETRY_IN_DATASTORE switch for connections and resources Writing connections and system resources to the datastore puts heavy load on the machine. Setting STORE_TELEMETRY_IN_DATASTORE=false makes both RPCs log that the data was accepted and return immediately, skipping all datastore calls (device authorization check, IP info lookups and the inserts). Defaults to true, so existing deployments are unaffected. Co-Authored-By: Claude Opus 5.5 --- wallguard-server/README.md | 1 + .../src/control_service/config.rs | 20 +++++++++++++++++++ .../rpc/handle_connections_data.rs | 9 +++++++++ .../rpc/handle_system_resources_data.rs | 9 +++++++++ .../src/control_service/service.rs | 11 ++++++++++ 5 files changed, 50 insertions(+) diff --git a/wallguard-server/README.md b/wallguard-server/README.md index cc9b4d1f..28fa12d5 100644 --- a/wallguard-server/README.md +++ b/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 | diff --git a/wallguard-server/src/control_service/config.rs b/wallguard-server/src/control_service/config.rs index 103af6e0..79a8bfc8 100644 --- a/wallguard-server/src/control_service/config.rs +++ b/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 + } + } +} diff --git a/wallguard-server/src/control_service/rpc/handle_connections_data.rs b/wallguard-server/src/control_service/rpc/handle_connections_data.rs index 4785a233..92e51de0 100644 --- a/wallguard-server/src/control_service/rpc/handle_connections_data.rs +++ b/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 diff --git a/wallguard-server/src/control_service/rpc/handle_system_resources_data.rs b/wallguard-server/src/control_service/rpc/handle_system_resources_data.rs index 215ca4a6..0ecab036 100644 --- a/wallguard-server/src/control_service/rpc/handle_system_resources_data.rs +++ b/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 diff --git a/wallguard-server/src/control_service/service.rs b/wallguard-server/src/control_service/service.rs index c828565a..a50b91c1 100644 --- a/wallguard-server/src/control_service/service.rs +++ b/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>, + 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, } }