From 0a5023ce6b81bf5e736a91117421c3ac164bbd17 Mon Sep 17 00:00:00 2001 From: Chris Beck Date: Fri, 5 Dec 2025 14:31:02 -0700 Subject: [PATCH] fixup to syslog parsing around EOF handling, add tests --- signal-gateway-bin/src/syslog/mod.rs | 262 ++++++++++++++++++++++----- 1 file changed, 216 insertions(+), 46 deletions(-) diff --git a/signal-gateway-bin/src/syslog/mod.rs b/signal-gateway-bin/src/syslog/mod.rs index f423fcd..3c5ef69 100644 --- a/signal-gateway-bin/src/syslog/mod.rs +++ b/signal-gateway-bin/src/syslog/mod.rs @@ -8,10 +8,10 @@ use signal_gateway::{Gateway, Level, LogMessage}; use std::{net::SocketAddr, str::FromStr, sync::Arc}; use syslog_rfc5424::{SyslogMessage, SyslogSeverity}; use tokio::{ - io::{AsyncReadExt, BufReader}, + io::{AsyncRead, AsyncReadExt, BufReader}, net::{TcpListener, TcpStream, UdpSocket}, }; -use tracing::{error, info}; +use tracing::{error, info, trace}; /// Configuration for the syslog listener (supports both UDP and TCP). #[derive(Clone, Conf, Debug)] @@ -101,7 +101,7 @@ impl SyslogConfig { continue; }; - info!("Accepted syslog TCP connection from {addr}"); + trace!("Accepted syslog TCP connection from {addr}"); let gateway = gateway.clone(); let config = config.clone(); @@ -110,8 +110,9 @@ impl SyslogConfig { tokio::spawn(async move { if let Err(err) = handle_tcp_connection(stream, &gateway, &config).await { error!("Syslog TCP connection from {addr} error: {err}"); + } else { + trace!("Syslog TCP connection from {addr} closed"); } - info!("Syslog TCP connection from {addr} closed"); }); } })) @@ -127,32 +128,8 @@ async fn handle_tcp_connection( let mut reader = BufReader::new(stream); loop { - // Read first byte to determine framing method - let first_byte = match reader.read_u8().await { - Ok(b) => b, - Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(()), // Clean EOF - Err(e) => return Err(e), - }; - - let msg_bytes = match first_byte { - // Whitespace: skip and continue - b' ' | b'\t' | b'\n' | b'\r' => continue, - - // Octet counting: starts with non-zero digit - b'1'..=b'9' => read_octet_counted_message(&mut reader, first_byte).await?, - - // Non-transparent framing: starts with '<' (beginning of syslog PRI) - b'<' => read_non_transparent_message(&mut reader, first_byte, config).await?, - - // Invalid first byte - _ => { - error!( - "Syslog TCP: invalid first byte 0x{:02x} ({:?}), closing connection", - first_byte, - char::from(first_byte) - ); - return Ok(()); - } + let Some(msg_bytes) = read_framed_syslog_bytes(&mut reader, config.tcp_cr_is_delimiter).await? else { + return Ok(()); // Clean EOF }; // Parse and process the message @@ -173,12 +150,68 @@ async fn handle_tcp_connection( } } +/// Read a single framed syslog message using RFC 6587 framing detection. +/// +/// Returns: +/// - `Ok(Some(bytes))` - A complete message was read +/// - `Ok(None)` - EOF reached (no more messages) +/// - `Err(InvalidData)` - Invalid framing (e.g., unexpected first byte) +/// - `Err(other)` - I/O error +async fn read_framed_syslog_bytes( + reader: &mut BufReader, + cr_is_delimiter: bool, +) -> std::io::Result>> +where + R: AsyncRead + Unpin, +{ + loop { + // Read first byte to determine framing method + let first_byte = match reader.read_u8().await { + Ok(b) => b, + Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None), + Err(e) => return Err(e), + }; + + match first_byte { + // Whitespace: skip and continue + b' ' | b'\t' | b'\n' | b'\r' => continue, + + // Octet counting: starts with non-zero digit + b'1'..=b'9' => { + return read_octet_counted_message(reader, first_byte).await.map(Some) + } + + // Non-transparent framing: starts with '<' (beginning of syslog PRI) + b'<' => { + return read_non_transparent_message(reader, first_byte, cr_is_delimiter) + .await + .map(Some) + } + + // Invalid first byte + _ => { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!( + "invalid first byte 0x{:02x} ({:?})", + first_byte, + char::from(first_byte) + ), + )) + } + } + } +} + /// Read an octet-counted message (RFC 6587 section 3.4.1). /// Format: MSG-LEN SP SYSLOG-MSG -async fn read_octet_counted_message( - reader: &mut BufReader, +async fn read_octet_counted_message( + reader: &mut BufReader, first_digit: u8, -) -> std::io::Result> { +) -> std::io::Result> +where + R: AsyncRead + Unpin, +{ let mut len: usize = (first_digit - b'0') as usize; // Read remaining digits until space @@ -211,28 +244,34 @@ async fn read_octet_counted_message( /// Read a non-transparent framed message (RFC 6587 section 3.4.2). /// Delimiter is NUL or LF by default, or NUL or CR if `tcp_cr_is_delimiter` is set. -async fn read_non_transparent_message( - reader: &mut BufReader, +/// Returns the message buffer on EOF (connection closed) as well as on delimiter. +async fn read_non_transparent_message( + reader: &mut BufReader, first_byte: u8, - config: &SyslogConfig, -) -> std::io::Result> { - let mut buf = vec![first_byte]; + cr_is_delimiter: bool, +) -> std::io::Result> +where + R: AsyncRead + Unpin, +{ + let mut msg = vec![first_byte]; - let line_delimiter = if config.tcp_cr_is_delimiter { - b'\r' - } else { - b'\n' - }; + let line_delimiter = if cr_is_delimiter { b'\r' } else { b'\n' }; loop { - let b = reader.read_u8().await?; + let b = match reader.read_u8().await { + Ok(b) => b, + Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break, // EOF + Err(e) => return Err(e), + }; + if b == b'\0' || b == line_delimiter { break; } - buf.push(b); + + msg.push(b); } - Ok(buf) + Ok(msg) } /// Convert SyslogSeverity to our Level enum @@ -302,4 +341,135 @@ mod tests { ) .unwrap(); } + + #[tokio::test] + async fn test_framing_eof_returns_none() { + let data: &[u8] = b""; + let mut reader = BufReader::new(data); + let result = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(result, None); + } + + #[tokio::test] + async fn test_framing_octet_counting() { + let data: &[u8] = b"5 hello"; + let mut reader = BufReader::new(data); + let result = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(result, Some(b"hello".to_vec())); + } + + #[tokio::test] + async fn test_framing_octet_counting_multidigit_length() { + let msg = "a]".repeat(50); // 100 bytes + let data = format!("100 {msg}"); + let mut reader = BufReader::new(data.as_bytes()); + let result = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(result, Some(msg.as_bytes().to_vec())); + } + + #[tokio::test] + async fn test_framing_non_transparent_lf_delimiter() { + let data: &[u8] = b"<14>1 test message\n"; + let mut reader = BufReader::new(data); + let result = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(result, Some(b"<14>1 test message".to_vec())); + } + + #[tokio::test] + async fn test_framing_non_transparent_nul_delimiter() { + let data: &[u8] = b"<14>1 test message\0"; + let mut reader = BufReader::new(data); + let result = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(result, Some(b"<14>1 test message".to_vec())); + } + + #[tokio::test] + async fn test_framing_non_transparent_cr_delimiter() { + let data: &[u8] = b"<14>1 test message\r"; + let mut reader = BufReader::new(data); + // With cr_is_delimiter=true, CR is a delimiter + let result = read_framed_syslog_bytes(&mut reader, true).await.unwrap(); + assert_eq!(result, Some(b"<14>1 test message".to_vec())); + } + + #[tokio::test] + async fn test_framing_non_transparent_cr_not_delimiter_by_default() { + let data: &[u8] = b"<14>1 test\rmessage\n"; + let mut reader = BufReader::new(data); + // With cr_is_delimiter=false, CR is part of the message + let result = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(result, Some(b"<14>1 test\rmessage".to_vec())); + } + + #[tokio::test] + async fn test_framing_non_transparent_eof_returns_buffer() { + // No delimiter, just EOF - should return the accumulated buffer + let data: &[u8] = b"<14>1 test message"; + let mut reader = BufReader::new(data); + let result = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(result, Some(b"<14>1 test message".to_vec())); + } + + #[tokio::test] + async fn test_framing_whitespace_skipping() { + let data: &[u8] = b" \n\t\r5 hello"; + let mut reader = BufReader::new(data); + let result = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(result, Some(b"hello".to_vec())); + } + + #[tokio::test] + async fn test_framing_whitespace_only_then_eof() { + let data: &[u8] = b" \n\t\r"; + let mut reader = BufReader::new(data); + let result = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(result, None); + } + + #[tokio::test] + async fn test_framing_invalid_first_byte() { + let data: &[u8] = b"xyz"; + let mut reader = BufReader::new(data); + let result = read_framed_syslog_bytes(&mut reader, false).await; + assert!(result.is_err()); + assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::InvalidData); + } + + #[tokio::test] + async fn test_framing_zero_not_valid_start() { + // '0' is not a valid start for octet counting (must be 1-9) + let data: &[u8] = b"0 hello"; + let mut reader = BufReader::new(data); + let result = read_framed_syslog_bytes(&mut reader, false).await; + assert!(result.is_err()); + assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::InvalidData); + } + + #[tokio::test] + async fn test_framing_multiple_messages() { + let data: &[u8] = b"5 hello<14>1 world\n3 foo"; + let mut reader = BufReader::new(data); + + let r1 = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(r1, Some(b"hello".to_vec())); + + let r2 = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(r2, Some(b"<14>1 world".to_vec())); + + let r3 = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(r3, Some(b"foo".to_vec())); + + let r4 = read_framed_syslog_bytes(&mut reader, false).await.unwrap(); + assert_eq!(r4, None); + } + + #[tokio::test] + async fn test_framing_octet_counting_invalid_separator() { + // After digits, we expect a space, not another character + let data: &[u8] = b"5xhello"; + let mut reader = BufReader::new(data); + let result = read_framed_syslog_bytes(&mut reader, false).await; + assert!(result.is_err()); + assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::InvalidData); + } }