move Logfilter to the Logmessage module, change MessageHandler to a proper trait

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",

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 }

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,33 +22,40 @@ 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<dyn MessageHandler> {
         let client = reqwest::Client::builder()
             .timeout(self.timeout)
             .build()
             .expect("Failed to build HTTP client");
 
-        Box::new(move |message: String| {
+        Box::new(AdminHttpHandler {
-            let client = client.clone();
+            client,
-            let url = self.url.clone();
+            url: self.url,
-            Box::pin(async move { handle_message(&client, &url, message).await })
         })
     }
 }
 
-/// Handle a message by POSTing it to the configured HTTP server
+/// Message handler that forwards messages to an HTTP server.
-async fn handle_message(
+struct AdminHttpHandler {
-    client: &reqwest::Client,
+    client: reqwest::Client,
-    url: &str,
+    url: String,
-    message: String,
+}
+
+#[async_trait]
+impl MessageHandler for AdminHttpHandler {
+    async fn handle_verified_signal_message(
+        &self,
+        msg: VerifiedSignalMessage,
+        _context: &dyn Context,
     ) -> MessageHandlerResult {
-    let response = client
+        let response = self
-        .post(url)
+            .client
-        .body(message)
+            .post(&self.url)
+            .body(msg.message)
             .send()
             .await
             .map_err(|err| (502u16, format!("HTTP request failed: {err}").into()))?;
@@ -62,3 +72,4 @@ async fn handle_message(
 
         Ok(AdminMessageResponse::new(body))
     }
+}

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,29 +27,37 @@ 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 {
+    pub fn into_handler(self) -> Box<dyn MessageHandler> {
-        Box::new(move |message: String| {
+        Box::new(AdminNetcatHandler { config: self })
-            let config = self.clone();
-            Box::pin(async move { handle_message(&config, message).await })
-        })
     }
 }
 
-/// Handle a message by forwarding it to the configured TCP server
+/// Message handler that forwards messages to a TCP server.
-async fn handle_message(config: &AdminNetcatConfig, message: String) -> MessageHandlerResult {
+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(config.timeout, TcpStream::connect(&config.tcp_addr))
+        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!("{message}\r\n");
+        let message = format!("{}\r\n", msg.message);
-    timeout(config.timeout, stream.write_all(message.as_bytes()))
+        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()))?;
@@ -54,7 +65,7 @@ async fn handle_message(config: &AdminNetcatConfig, message: String) -> MessageH
         // 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))
+        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()))?;
@@ -67,3 +78,4 @@ async fn handle_message(config: &AdminNetcatConfig, message: String) -> MessageH
 
         Ok(AdminMessageResponse::new(text))
     }
+}

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<dyn signal_gateway::MessageHandler> {
         match self {
             AdminHandlerCommand::Netcat(config) => config.into_handler(),
             AdminHandlerCommand::Http(config) => config.into_handler(),

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 }

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)]
+    #[serde(flatten)]
-    pub msg_contains: String,
+    pub filter: LogFilter,
-    #[serde(default)]
-    pub module_equals: String,
-    #[serde(default)]
-    pub file_equals: String,
-    #[serde(default)]
-    pub line_equals: String,
     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<usize> = None;
-        for (idx, (filter, limiter)) in self.rate_limiters.iter().enumerate() {
+        for (idx, (rule, limiter)) in self.rate_limiters.iter().enumerate() {
-            if filter.eval_filter(log_msg) && !limiter.lock().await.evaluate(ts_sec) {
+            if rule.filter.matches(log_msg) && !limiter.lock().await.evaluate(ts_sec) {
                 suppressed_by_rule.get_or_insert(idx);
             }
         }

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<HashMap<Origin, LogHandler>>,
     /// Handler for admin messages that don't start with `/`
-    message_handler: Option<MessageHandler>,
+    message_handler: Option<Box<dyn MessageHandler>>,
 }
 
 impl Gateway {
     pub async fn new(
         config: GatewayConfig,
         token: CancellationToken,
-        message_handler: Option<MessageHandler>,
+        message_handler: Option<Box<dyn MessageHandler>>,
     ) -> 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"));
-    }
 }

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 log_message::{Level, LogFilter, LogMessage, LogMessageBuilder};
-pub use message_handler::{AdminMessageResponse, MessageHandler, MessageHandlerResult};
+pub use message_handler::{
+    AdminMessageResponse, Context, MessageHandler, MessageHandlerResult, VerifiedSignalMessage,
+};

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

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<String>, 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<AdminMessageResponse, (u16, Box<dyn Error + Send + Sync>)>;
 
-/// Handler function for admin messages that don't start with `/`.
+/// Handler for admin messages that don't start with `/`.
-/// Takes the message text and returns a response.
+#[async_trait]
-pub type MessageHandler =
+pub trait MessageHandler: Send + Sync {
-    Box<dyn Fn(String) -> Pin<Box<dyn Future<Output = MessageHandlerResult> + Send>> + 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;
+}