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, } } 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"); + } +}