From a7cd3bdcdefefbd13dc1a6276c768962bb37e166 Mon Sep 17 00:00:00 2001 From: Chris Beck Date: Fri, 5 Dec 2025 11:20:58 -0700 Subject: [PATCH] move Logfilter to the Logmessage module, change MessageHandler to a proper trait --- Cargo.lock | 2 + signal-gateway-bin/Cargo.toml | 1 + signal-gateway-bin/src/admin_http/mod.rs | 73 +++++++++------ signal-gateway-bin/src/admin_netcat/mod.rs | 86 +++++++++-------- signal-gateway-bin/src/main.rs | 4 +- signal-gateway/Cargo.toml | 1 + signal-gateway/src/gateway/log_handler.rs | 54 ++--------- signal-gateway/src/gateway/mod.rs | 51 +++------- signal-gateway/src/lib.rs | 6 +- signal-gateway/src/log_message.rs | 104 +++++++++++++++++++++ signal-gateway/src/message_handler.rs | 50 +++++++++- 11 files changed, 270 insertions(+), 162 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 49fda46..203bd09 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1645,6 +1645,7 @@ checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" name = "signal-gateway" version = "0.1.0" dependencies = [ + "async-trait", "chrono", "chrono-tz", "conf", @@ -1672,6 +1673,7 @@ dependencies = [ name = "signal-gateway-bin" version = "0.1.0" dependencies = [ + "async-trait", "chrono", "conf", "conf-extra", diff --git a/signal-gateway-bin/Cargo.toml b/signal-gateway-bin/Cargo.toml index f1f4117..399b9cf 100644 --- a/signal-gateway-bin/Cargo.toml +++ b/signal-gateway-bin/Cargo.toml @@ -18,6 +18,7 @@ rustls-tls = ["signal-gateway/rustls-tls"] [dependencies] signal-gateway = { path = "../signal-gateway" } +async-trait = { workspace = true } chrono = { workspace = true } conf = { workspace = true } conf-extra = { workspace = true } diff --git a/signal-gateway-bin/src/admin_http/mod.rs b/signal-gateway-bin/src/admin_http/mod.rs index 0bbb1ed..2a8409b 100644 --- a/signal-gateway-bin/src/admin_http/mod.rs +++ b/signal-gateway-bin/src/admin_http/mod.rs @@ -3,8 +3,11 @@ //! This module handles admin messages not handled by the gateway by making an HTTP POST request //! with the message as the body, and returning the response body as the reply. +use async_trait::async_trait; use conf::Conf; -use signal_gateway::{AdminMessageResponse, MessageHandler, MessageHandlerResult}; +use signal_gateway::{ + AdminMessageResponse, Context, MessageHandler, MessageHandlerResult, VerifiedSignalMessage, +}; use std::time::Duration; /// Configuration for the admin HTTP client @@ -19,46 +22,54 @@ pub struct AdminHttpConfig { } impl AdminHttpConfig { - /// Create a message handler function from this config. + /// Create a message handler from this config. /// /// The returned handler makes an HTTP POST request to the configured URL /// with the message as the body, and returns the response body. - pub fn into_handler(self) -> MessageHandler { + pub fn into_handler(self) -> Box { let client = reqwest::Client::builder() .timeout(self.timeout) .build() .expect("Failed to build HTTP client"); - Box::new(move |message: String| { - let client = client.clone(); - let url = self.url.clone(); - Box::pin(async move { handle_message(&client, &url, message).await }) + Box::new(AdminHttpHandler { + client, + url: self.url, }) } } -/// Handle a message by POSTing it to the configured HTTP server -async fn handle_message( - client: &reqwest::Client, - url: &str, - message: String, -) -> MessageHandlerResult { - let response = client - .post(url) - .body(message) - .send() - .await - .map_err(|err| (502u16, format!("HTTP request failed: {err}").into()))?; - - let status = response.status(); - let body = response - .text() - .await - .map_err(|err| (502u16, format!("Failed to read response body: {err}").into()))?; - - if !status.is_success() { - return Err((status.as_u16(), body.into())); - } - - Ok(AdminMessageResponse::new(body)) +/// Message handler that forwards messages to an HTTP server. +struct AdminHttpHandler { + client: reqwest::Client, + url: String, +} + +#[async_trait] +impl MessageHandler for AdminHttpHandler { + async fn handle_verified_signal_message( + &self, + msg: VerifiedSignalMessage, + _context: &dyn Context, + ) -> MessageHandlerResult { + let response = self + .client + .post(&self.url) + .body(msg.message) + .send() + .await + .map_err(|err| (502u16, format!("HTTP request failed: {err}").into()))?; + + let status = response.status(); + let body = response + .text() + .await + .map_err(|err| (502u16, format!("Failed to read response body: {err}").into()))?; + + if !status.is_success() { + return Err((status.as_u16(), body.into())); + } + + Ok(AdminMessageResponse::new(body)) + } } diff --git a/signal-gateway-bin/src/admin_netcat/mod.rs b/signal-gateway-bin/src/admin_netcat/mod.rs index c6168e0..a7cd835 100644 --- a/signal-gateway-bin/src/admin_netcat/mod.rs +++ b/signal-gateway-bin/src/admin_netcat/mod.rs @@ -3,8 +3,11 @@ //! This module handles admin messages not handled by the gateway by opening a TCP connection, //! writing the message terminated with CRLF, and reading the response until CRLF. +use async_trait::async_trait; use conf::Conf; -use signal_gateway::{AdminMessageResponse, MessageHandler, MessageHandlerResult}; +use signal_gateway::{ + AdminMessageResponse, Context, MessageHandler, MessageHandlerResult, VerifiedSignalMessage, +}; use std::time::Duration; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, @@ -24,46 +27,55 @@ pub struct AdminNetcatConfig { } impl AdminNetcatConfig { - /// Create a message handler function from this config. + /// Create a message handler from this config. /// /// The returned handler opens a TCP connection to the configured address, /// writes the message terminated with CRLF, and reads the response until CRLF. - pub fn into_handler(self) -> MessageHandler { - Box::new(move |message: String| { - let config = self.clone(); - Box::pin(async move { handle_message(&config, message).await }) - }) + pub fn into_handler(self) -> Box { + Box::new(AdminNetcatHandler { config: self }) } } -/// Handle a message by forwarding it to the configured TCP server -async fn handle_message(config: &AdminNetcatConfig, message: String) -> MessageHandlerResult { - // Connect to server - let mut stream = timeout(config.timeout, TcpStream::connect(&config.tcp_addr)) - .await - .map_err(|_| (504u16, "connecting: timeout".into()))? - .map_err(|err| (502u16, format!("connecting: {err}").into()))?; - - // Write message with CRLF terminator - let message = format!("{message}\r\n"); - timeout(config.timeout, stream.write_all(message.as_bytes())) - .await - .map_err(|_| (504u16, "writing: timeout".into()))? - .map_err(|err| (502u16, format!("writing: {err}").into()))?; - - // Read response until CR - let mut reader = BufReader::new(stream); - let mut buf = Vec::new(); - timeout(config.timeout, reader.read_until(b'\r', &mut buf)) - .await - .map_err(|_| (504u16, "reading: timeout".into()))? - .map_err(|err| (502u16, format!("reading: {err}").into()))?; - - // Convert to string and trim the trailing CR - let text = std::str::from_utf8(&buf) - .map_err(|err| (502u16, format!("utf8: {err}").into()))? - .trim_end_matches(['\r', '\n']) - .to_owned(); - - Ok(AdminMessageResponse::new(text)) +/// Message handler that forwards messages to a TCP server. +struct AdminNetcatHandler { + config: AdminNetcatConfig, +} + +#[async_trait] +impl MessageHandler for AdminNetcatHandler { + async fn handle_verified_signal_message( + &self, + msg: VerifiedSignalMessage, + _context: &dyn Context, + ) -> MessageHandlerResult { + // Connect to server + let mut stream = + timeout(self.config.timeout, TcpStream::connect(&self.config.tcp_addr)) + .await + .map_err(|_| (504u16, "connecting: timeout".into()))? + .map_err(|err| (502u16, format!("connecting: {err}").into()))?; + + // Write message with CRLF terminator + let message = format!("{}\r\n", msg.message); + timeout(self.config.timeout, stream.write_all(message.as_bytes())) + .await + .map_err(|_| (504u16, "writing: timeout".into()))? + .map_err(|err| (502u16, format!("writing: {err}").into()))?; + + // Read response until CR + let mut reader = BufReader::new(stream); + let mut buf = Vec::new(); + timeout(self.config.timeout, reader.read_until(b'\r', &mut buf)) + .await + .map_err(|_| (504u16, "reading: timeout".into()))? + .map_err(|err| (502u16, format!("reading: {err}").into()))?; + + // Convert to string and trim the trailing CR + let text = std::str::from_utf8(&buf) + .map_err(|err| (502u16, format!("utf8: {err}").into()))? + .trim_end_matches(['\r', '\n']) + .to_owned(); + + Ok(AdminMessageResponse::new(text)) + } } diff --git a/signal-gateway-bin/src/main.rs b/signal-gateway-bin/src/main.rs index 4c3c243..85e0d27 100644 --- a/signal-gateway-bin/src/main.rs +++ b/signal-gateway-bin/src/main.rs @@ -2,7 +2,7 @@ use conf::{Conf, Subcommands}; use hyper::service::service_fn; use hyper_util::rt::TokioIo; use hyper_util::server::conn::auto; -use signal_gateway::{Gateway, GatewayConfig, MessageHandler}; +use signal_gateway::{Gateway, GatewayConfig}; use std::{net::SocketAddr, sync::Arc, time::Duration}; use tokio::net::TcpListener; use tokio_util::sync::CancellationToken; @@ -34,7 +34,7 @@ enum AdminHandlerCommand { } impl AdminHandlerCommand { - fn into_handler(self) -> MessageHandler { + fn into_handler(self) -> Box { match self { AdminHandlerCommand::Netcat(config) => config.into_handler(), AdminHandlerCommand::Http(config) => config.into_handler(), diff --git a/signal-gateway/Cargo.toml b/signal-gateway/Cargo.toml index 0a3ae67..296e66f 100644 --- a/signal-gateway/Cargo.toml +++ b/signal-gateway/Cargo.toml @@ -14,6 +14,7 @@ rustls-tls = ["prometheus-http-client/rustls-tls"] [dependencies] prometheus-http-client = { workspace = true, default-features = false } +async-trait = { workspace = true } chrono = { workspace = true } conf = { workspace = true } conf-extra = { workspace = true } diff --git a/signal-gateway/src/gateway/log_handler.rs b/signal-gateway/src/gateway/log_handler.rs index c8148a2..33fc1f5 100644 --- a/signal-gateway/src/gateway/log_handler.rs +++ b/signal-gateway/src/gateway/log_handler.rs @@ -2,7 +2,7 @@ use super::circular_buffer::CircularBuffer; use super::{AdminMessage, MultiRateLimiter, RateThreshold, SourceLocationRateLimiter}; use crate::{ human_duration::HumanTMinus, - log_message::{Level, LogMessage, Origin}, + log_message::{Level, LogFilter, LogMessage, Origin}, }; use chrono::{TimeDelta, Utc}; use conf::Conf; @@ -50,49 +50,11 @@ pub struct LogHandlerConfig { #[derive(Clone, Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct AlertRule { - #[serde(default)] - pub msg_contains: String, - #[serde(default)] - pub module_equals: String, - #[serde(default)] - pub file_equals: String, - #[serde(default)] - pub line_equals: String, + #[serde(flatten)] + pub filter: LogFilter, pub threshold: RateThreshold, } -impl AlertRule { - /// Check if a log message passes the filter defined by this rule - fn eval_filter(&self, log_msg: &LogMessage) -> bool { - if !self.msg_contains.is_empty() && !log_msg.msg.contains(&self.msg_contains) { - return false; - } - - if !self.module_equals.is_empty() { - match log_msg.module_path.as_deref() { - Some(module) if module == self.module_equals.as_str() => {} - _ => return false, - } - } - - if !self.file_equals.is_empty() { - match log_msg.file.as_deref() { - Some(file) if file == self.file_equals.as_str() => {} - _ => return false, - } - } - - if !self.line_equals.is_empty() { - match log_msg.line.as_deref() { - Some(line) if line == self.line_equals.as_str() => {} - _ => return false, - } - } - - true - } -} - /// The log handler takes log messages from a single origin and decides what /// to do with them. /// @@ -131,15 +93,15 @@ impl LogHandler { let any_rule_uses_module = config .alert_rate_limits .iter() - .any(|r| !r.module_equals.is_empty()); + .any(|r| r.filter.uses_module()); let any_rule_uses_file = config .alert_rate_limits .iter() - .any(|r| !r.file_equals.is_empty()); + .any(|r| r.filter.uses_file()); let any_rule_uses_line = config .alert_rate_limits .iter() - .any(|r| !r.line_equals.is_empty()); + .any(|r| r.filter.uses_line()); let rate_limiters = config .alert_rate_limits .iter() @@ -320,8 +282,8 @@ impl LogHandler { // Check each configured rule - track which rule suppressed the alert // Note: we check all rules even if one already suppressed, to update all rate limiters let mut suppressed_by_rule: Option = None; - for (idx, (filter, limiter)) in self.rate_limiters.iter().enumerate() { - if filter.eval_filter(log_msg) && !limiter.lock().await.evaluate(ts_sec) { + for (idx, (rule, limiter)) in self.rate_limiters.iter().enumerate() { + if rule.filter.matches(log_msg) && !limiter.lock().await.evaluate(ts_sec) { suppressed_by_rule.get_or_insert(idx); } } diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index 6df8c4a..33bfbd1 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/signal-gateway/src/gateway/mod.rs @@ -6,7 +6,9 @@ use crate::{ Envelope, Identity, MessageTarget, RpcClient, RpcClientError, SignalMessage, connect_tcp, }, log_message::{LogMessage, Origin}, - message_handler::{AdminMessageResponse, MessageHandler, MessageHandlerResult}, + message_handler::{ + AdminMessageResponse, Context, MessageHandler, MessageHandlerResult, VerifiedSignalMessage, + }, prometheus::{Prometheus, PrometheusConfig}, }; use chrono::Utc; @@ -181,14 +183,14 @@ pub struct Gateway { /// Log handlers keyed by origin (app + host). Lazily created when first message from an origin arrives. log_handlers: RwLock>, /// Handler for admin messages that don't start with `/` - message_handler: Option, + message_handler: Option>, } impl Gateway { pub async fn new( config: GatewayConfig, token: CancellationToken, - message_handler: Option, + message_handler: Option>, ) -> Self { let (admin_mq_tx, admin_mq_rx) = unbounded_channel(); @@ -469,7 +471,8 @@ impl Gateway { self.handle_gateway_command(cmd).await } else if let Some(handler) = &self.message_handler { - handler(data.message.clone()).await + let msg = VerifiedSignalMessage::new(data.message.clone(), data.timestamp); + handler.handle_verified_signal_message(msg, &GatewayContext).await } else { Err((501u16, "No message handler configured".into())) } @@ -828,6 +831,11 @@ impl Gateway { } } +/// Placeholder context for message handlers. +struct GatewayContext; + +impl Context for GatewayContext {} + impl Drop for Gateway { fn drop(&mut self) { self.token.cancel(); @@ -922,39 +930,4 @@ mod tests { assert!(parse_gateway_command("/").is_err()); assert!(parse_gateway_command("").is_err()); } - - #[test] - fn test_origin_matches_filter() { - let origin = Origin { - app: "muad-dib".into(), - host: "tokyo-server".into(), - }; - - // Without @: matches if app OR host contains the string - assert!(origin.matches_filter("muad")); - assert!(origin.matches_filter("dib")); - assert!(origin.matches_filter("tokyo")); - assert!(origin.matches_filter("server")); - assert!(!origin.matches_filter("paris")); - - // With @: app must contain first part AND host must contain second part - assert!(origin.matches_filter("muad@tokyo")); - assert!(origin.matches_filter("dib@server")); - assert!(origin.matches_filter("muad-dib@tokyo-server")); - assert!(!origin.matches_filter("muad@paris")); - assert!(!origin.matches_filter("other@tokyo")); - - // Empty parts with @ - assert!(origin.matches_filter("@tokyo")); // empty app filter matches any app - assert!(origin.matches_filter("muad@")); // empty host filter matches any host - assert!(origin.matches_filter("@")); // both empty, matches everything - - // Edge case: filter matches the @ in the format but origin has no @ - let origin2 = Origin { - app: "app".into(), - host: "host".into(), - }; - assert!(origin2.matches_filter("app@host")); - assert!(!origin2.matches_filter("app@other")); - } } diff --git a/signal-gateway/src/lib.rs b/signal-gateway/src/lib.rs index 30cd186..92d83a3 100644 --- a/signal-gateway/src/lib.rs +++ b/signal-gateway/src/lib.rs @@ -9,5 +9,7 @@ pub(crate) mod prometheus; pub(crate) mod transports; pub use gateway::{Gateway, GatewayConfig}; -pub use log_message::{Level, LogMessage, LogMessageBuilder}; -pub use message_handler::{AdminMessageResponse, MessageHandler, MessageHandlerResult}; +pub use log_message::{Level, LogFilter, LogMessage, LogMessageBuilder}; +pub use message_handler::{ + AdminMessageResponse, Context, MessageHandler, MessageHandlerResult, VerifiedSignalMessage, +}; diff --git a/signal-gateway/src/log_message.rs b/signal-gateway/src/log_message.rs index 5d98a9c..97667fc 100644 --- a/signal-gateway/src/log_message.rs +++ b/signal-gateway/src/log_message.rs @@ -1,5 +1,7 @@ //! Log message schema used by this crate +use serde::Deserialize; + #[non_exhaustive] #[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd)] pub enum Level { @@ -171,3 +173,105 @@ impl Origin { } } } + +/// Filter criteria for matching log messages +#[derive(Clone, Debug, Default, Deserialize)] +pub struct LogFilter { + #[serde(default)] + pub msg_contains: String, + #[serde(default)] + pub module_equals: String, + #[serde(default)] + pub file_equals: String, + #[serde(default)] + pub line_equals: String, +} + +impl LogFilter { + /// Check if a log message matches this filter. + /// + /// Returns true if all non-empty filter fields match the log message. + pub fn matches(&self, log_msg: &LogMessage) -> bool { + if !self.msg_contains.is_empty() && !log_msg.msg.contains(&self.msg_contains) { + return false; + } + + if !self.module_equals.is_empty() { + match log_msg.module_path.as_deref() { + Some(module) if module == self.module_equals.as_str() => {} + _ => return false, + } + } + + if !self.file_equals.is_empty() { + match log_msg.file.as_deref() { + Some(file) if file == self.file_equals.as_str() => {} + _ => return false, + } + } + + if !self.line_equals.is_empty() { + match log_msg.line.as_deref() { + Some(line) if line == self.line_equals.as_str() => {} + _ => return false, + } + } + + true + } + + /// Returns true if this filter uses the module field + pub fn uses_module(&self) -> bool { + !self.module_equals.is_empty() + } + + /// Returns true if this filter uses the file field + pub fn uses_file(&self) -> bool { + !self.file_equals.is_empty() + } + + /// Returns true if this filter uses the line field + pub fn uses_line(&self) -> bool { + !self.line_equals.is_empty() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_origin_matches_filter() { + let origin = Origin { + app: "muad-dib".into(), + host: "tokyo-server".into(), + }; + + // Without @: matches if app OR host contains the string + assert!(origin.matches_filter("muad")); + assert!(origin.matches_filter("dib")); + assert!(origin.matches_filter("tokyo")); + assert!(origin.matches_filter("server")); + assert!(!origin.matches_filter("paris")); + + // With @: app must contain first part AND host must contain second part + assert!(origin.matches_filter("muad@tokyo")); + assert!(origin.matches_filter("dib@server")); + assert!(origin.matches_filter("muad-dib@tokyo-server")); + assert!(!origin.matches_filter("muad@paris")); + assert!(!origin.matches_filter("other@tokyo")); + + // Empty parts with @ + assert!(origin.matches_filter("@tokyo")); // empty app filter matches any app + assert!(origin.matches_filter("muad@")); // empty host filter matches any host + assert!(origin.matches_filter("@")); // both empty, matches everything + + // Edge case: filter matches the @ in the format but origin has no @ + let origin2 = Origin { + app: "app".into(), + host: "host".into(), + }; + assert!(origin2.matches_filter("app@host")); + assert!(!origin2.matches_filter("app@other")); + } +} diff --git a/signal-gateway/src/message_handler.rs b/signal-gateway/src/message_handler.rs index c09efcc..acb50db 100644 --- a/signal-gateway/src/message_handler.rs +++ b/signal-gateway/src/message_handler.rs @@ -1,6 +1,37 @@ //! Message handler types for admin messages not handled by the gateway. -use std::{error::Error, future::Future, path::PathBuf, pin::Pin}; +use async_trait::async_trait; +use std::{error::Error, path::PathBuf}; + +/// A verified Signal message from an admin. +/// +/// This struct contains the message content and metadata for a message +/// that has been verified as coming from a trusted admin. +#[non_exhaustive] +#[derive(Clone, Debug)] +pub struct VerifiedSignalMessage { + /// The text content of the message. + pub message: String, + /// The timestamp of the message (milliseconds since Unix epoch). + pub timestamp: u64, +} + +impl VerifiedSignalMessage { + /// Create a new verified signal message. + pub fn new(message: impl Into, timestamp: u64) -> Self { + Self { + message: message.into(), + timestamp, + } + } +} + +/// Context for message handler operations. +/// +/// This trait provides access to gateway functionality that message handlers +/// may need. Currently empty, but reserved for future expansion. +#[async_trait] +pub trait Context: Send + Sync {} /// Response to an admin message. #[non_exhaustive] @@ -77,7 +108,16 @@ impl AdminMessageResponseBuilder { /// Result type for message handler responses. pub type MessageHandlerResult = Result)>; -/// Handler function for admin messages that don't start with `/`. -/// Takes the message text and returns a response. -pub type MessageHandler = - Box Pin + Send>> + Send + Sync>; +/// Handler for admin messages that don't start with `/`. +#[async_trait] +pub trait MessageHandler: Send + Sync { + /// Handle a verified Signal message from an admin. + /// + /// This is called for admin messages that don't start with `/` (which are + /// handled as gateway commands). + async fn handle_verified_signal_message( + &self, + msg: VerifiedSignalMessage, + context: &dyn Context, + ) -> MessageHandlerResult; +}