Implemented basic minimal HTTP POST web hook for new mails

Author: cry-inc

src/background.rs

@@ -5,6 +5,7 @@ use crate::state::{
     AppState, DmarcReportWithMailId, FileType, ReportParsingError, TlsReportWithMailId,
 };
 use crate::unpack::extract_report_files;
+use crate::web_hook::mail_web_hook;
 use crate::{dmarc, tls};
 use anyhow::{Context, Result};
 use chrono::Local;
@@ -14,7 +15,7 @@ use std::time::{Duration, Instant, SystemTime};
 use tokio::sync::Mutex;
 use tokio::sync::mpsc::Receiver;
 use tokio::task::JoinHandle;
-use tracing::{error, info, trace, warn};
+use tracing::{debug, error, info, trace, warn};
 
 pub fn start_bg_task(
     config: Configuration,
@@ -31,13 +32,20 @@ pub fn start_bg_task(
             info!("Starting background update...");
             match bg_update(&config, &state).await {
                 Ok(new_mails) => {
-                    if !new_mails.is_empty() {
-                        info!("Detected {} new mails", new_mails.len());
-                    }
+                    info!("Detected {} new mails", new_mails.len());
                     info!(
                         "Finished background update after {:.3}s",
                         start.elapsed().as_secs_f64()
                     );
+                    if !new_mails.is_empty() && config.mail_web_hook_url.is_some() {
+                        debug!("Calling web hook for new mails...");
+                        for mail_id in &new_mails {
+                            if let Err(err) = mail_web_hook(&config, mail_id).await {
+                                warn!("Failed to call web hook for mail {mail_id}: {err:#}");
+                            }
+                        }
+                        debug!("Finished calling web hook for new mails");
+                    }
                 }
                 Err(err) => error!("Failed background update: {err:#}"),
             };

src/config.rs

@@ -138,6 +138,15 @@ pub struct Configuration {
     /// Maximum mail size in bytes, anything bigger will be ignored and not parsed
     #[arg(long, env, default_value_t = 1024 * 1024 * 1)]
     pub max_mail_size: u32,
+
+    /// URL for optional web hook that is called via HTTP when a new mail is detected.
+    /// Please note that this app does not have a persistent store for already known mails.
+    /// When the application starts, all existing mails in the IMAP account are considered known.
+    /// Only the subsequent updates that occur while the app is running will be able to detect new mails.
+    /// The URL only supports plain HTTP and will reveive a HTTP POST request.
+    /// Example value: http://myserver.org/api/new_mails_endpoint
+    #[arg(long, env)]
+    pub mail_web_hook_url: Option<String>,
 }
 
 impl Configuration {
@@ -179,6 +188,8 @@ impl Configuration {
         info!("HTTPS Cache Dir: {:?}", self.https_auto_cert_cache);
 
         info!("Maximum Mail Body Size: {} bytes", self.max_mail_size);
+
+        info!("Mail Web Hook URL: {:?}", self.mail_web_hook_url);
     }
 }
 

src/main.rs

@@ -12,6 +12,7 @@ mod mail;
 mod state;
 mod tls;
 mod unpack;
+mod web_hook;
 mod whois;
 
 use crate::background::start_bg_task;

src/web_hook.rs

@@ -0,0 +1,80 @@
+use crate::config::Configuration;
+use anyhow::{Context, Result, bail, ensure};
+use axum::http::uri::Scheme;
+use http_body_util::{BodyExt, Empty};
+use hyper::body::Bytes;
+use hyper::client::conn::http1;
+use hyper::{Method, Request, Uri};
+use hyper_util::rt::TokioIo;
+use tokio::net::TcpStream;
+use tracing::{debug, error};
+
+pub async fn mail_web_hook(config: &Configuration, mail_id: &str) -> Result<()> {
+    let url = config
+        .mail_web_hook_url
+        .as_deref()
+        .context("Failed to get web hook URL for new mails")?;
+
+    // Create and parse URI
+    let uri = url.parse::<Uri>().context("Failed to parse URL")?;
+    ensure!(
+        uri.scheme().context("URL has no scheme")? == &Scheme::HTTP,
+        "Only plain HTTP is supported"
+    );
+
+    // Get the host and the port
+    let host = uri.host().context("URL has no host")?.to_string();
+    let port = uri.port_u16().unwrap_or(80);
+
+    // Open a TCP connection to the remote host
+    let address = format!("{host}:{port}");
+    let stream = TcpStream::connect(&address)
+        .await
+        .context(format!("Failed to connect TCP stream at {address}"))?;
+
+    // Create the Hyper client
+    let io = TokioIo::new(stream);
+    let (mut sender, conn) = http1::handshake(io)
+        .await
+        .context("Failed to create HTTP handshake")?;
+
+    // Spawn a task to drive the HTTP state
+    tokio::task::spawn(async move {
+        if let Err(err) = conn.await {
+            error!("Connection failed: {err:?}");
+        }
+    });
+
+    // Create and send HTTP request
+    let req = Request::builder()
+        .uri(uri)
+        .method(Method::POST)
+        .header(hyper::header::HOST, host)
+        .body(Empty::<Bytes>::new())
+        .context("Failed to create HTTP request")?;
+    let mut res = sender
+        .send_request(req)
+        .await
+        .context("Failed to send HTTP request")?;
+
+    let status_code = res.status().as_u16();
+    debug!("Web hook for new mail {mail_id} responded with status code {status_code}");
+
+    // Get response body piece by piece
+    let mut body = Vec::new();
+    while let Some(next) = res.frame().await {
+        let frame = next.context("Failed to receive next HTTP response chunk")?;
+        if let Some(chunk) = frame.data_ref() {
+            body.extend_from_slice(chunk);
+        }
+        if body.len() > 1024 * 1024 {
+            bail!("HTTP response too big");
+        }
+    }
+
+    // Parse and log response body
+    let body = String::from_utf8_lossy(&body);
+    debug!("Web hook for new mail {mail_id} responded with body: {body}");
+
+    Ok(())
+}