Add notifications with ntfy

bin/Cargo.toml

@@ -23,8 +23,17 @@ path = "src/check.rs"
 rayhunter = { path = "../lib" }
 toml = "0.8.8"
 serde = { version = "1.0.193", features = ["derive"] }
-tokio = { version = "1.44.2", default-features = false, features = ["fs", "signal", "process", "rt-multi-thread"] }
-axum = { version = "0.8", default-features = false, features = ["http1", "tokio", "json"] }
+tokio = { version = "1.44.2", default-features = false, features = [
+    "fs",
+    "signal",
+    "process",
+    "rt-multi-thread",
+] }
+axum = { version = "0.8", default-features = false, features = [
+    "http1",
+    "tokio",
+    "json",
+] }
 thiserror = "1.0.52"
 libc = "0.2.150"
 log = "0.4.20"
@@ -38,8 +47,14 @@ tokio-stream = { version = "0.1.14", default-features = false }
 futures = { version = "0.3.30", default-features = false }
 clap = { version = "4.5.2", features = ["derive"] }
 serde_json = "1.0.114"
-image = { version =  "0.25.1", default-features = false, features = ["png", "gif"] }
+image = { version = "0.25.1", default-features = false, features = [
+    "png",
+    "gif",
+] }
 tempfile = "3.10.1"
 simple_logger = "5.0.0"
 async_zip = { version = "0.0.17", features = ["tokio"] }
 anyhow = "1.0.98"
+reqwest = { version = "0.12.20", default-features = false, features = [
+    "rustls-tls",
+] }

bin/src/config.rs

@@ -14,6 +14,7 @@ pub struct Config {
     pub enable_dummy_analyzer: bool,
     pub colorblind_mode: bool,
     pub key_input_mode: u8,
+    pub ntfy_topic: Option<String>,
     pub analyzers: AnalyzerConfig,
 }
 
@@ -28,6 +29,7 @@ impl Default for Config {
             colorblind_mode: false,
             key_input_mode: 0,
             analyzers: AnalyzerConfig::default(),
+            ntfy_topic: None,
         }
     }
 }

bin/src/daemon.rs

@@ -5,6 +5,7 @@ mod display;
 mod dummy_analyzer;
 mod error;
 mod key_input;
+mod notifications;
 mod pcap;
 mod qmdl_store;
 mod server;
@@ -17,6 +18,7 @@ use std::sync::Arc;
 use crate::config::{parse_args, parse_config};
 use crate::diag::run_diag_read_thread;
 use crate::error::RayhunterError;
+use crate::notifications::{run_notification_worker, NotificationService};
 use crate::pcap::get_pcap;
 use crate::qmdl_store::RecordingStore;
 use crate::server::{get_config, get_qmdl, get_zip, serve_static, set_config, ServerState};
@@ -212,6 +214,9 @@ async fn run_with_config(
     let (analysis_tx, analysis_rx) = mpsc::channel::<AnalysisCtrlMessage>(5);
     let mut maybe_ui_shutdown_tx = None;
     let mut maybe_key_input_shutdown_tx = None;
+
+    let notification_service = NotificationService::new(config.ntfy_topic.clone());
+
     if !config.debug_mode {
         let (ui_shutdown_tx, ui_shutdown_rx) = oneshot::channel();
         maybe_ui_shutdown_tx = Some(ui_shutdown_tx);
@@ -232,6 +237,7 @@ async fn run_with_config(
             analysis_tx.clone(),
             config.enable_dummy_analyzer,
             config.analyzers.clone(),
+            notification_service.new_handler(),
         );
         info!("Starting UI");
         display::update_ui(&task_tracker, &config, ui_shutdown_rx, ui_update_rx);
@@ -271,6 +277,7 @@ async fn run_with_config(
         qmdl_store_lock.clone(),
         analysis_tx.clone(),
     );
+    run_notification_worker(&task_tracker, notification_service);
     let state = Arc::new(ServerState {
         config_path: args.config_path.clone(),
         config,

bin/src/diag.rs

@@ -1,5 +1,6 @@
 use std::pin::pin;
 use std::sync::Arc;
+use std::time::Duration;
 
 use axum::body::Body;
 use axum::extract::{Path, State};
@@ -20,6 +21,7 @@ use tokio_util::task::TaskTracker;
 
 use crate::analysis::{AnalysisCtrlMessage, AnalysisWriter};
 use crate::display;
+use crate::notifications::Notification;
 use crate::qmdl_store::{RecordingStore, RecordingStoreError};
 use crate::server::ServerState;
 
@@ -38,6 +40,7 @@ pub fn run_diag_read_thread(
     analysis_sender: Sender<AnalysisCtrlMessage>,
     enable_dummy_analyzer: bool,
     analyzer_config: AnalyzerConfig,
+    notification_channel: tokio::sync::mpsc::Sender<Notification>,
 ) {
     task_tracker.spawn(async move {
         let (initial_qmdl_file, initial_analysis_file) = qmdl_store_lock.write().await.new_entry().await.expect("failed creating QMDL file entry");
@@ -45,6 +48,7 @@ pub fn run_diag_read_thread(
         let mut diag_stream = pin!(dev.as_stream().into_stream());
         let mut maybe_analysis_writer = Some(AnalysisWriter::new(initial_analysis_file, enable_dummy_analyzer, &analyzer_config).await
             .expect("failed to create analysis writer"));
+
         loop {
             tokio::select! {
                 msg = qmdl_file_rx.recv() => {
@@ -130,13 +134,18 @@ pub fn run_diag_read_thread(
                             }
 
                             if let Some(analysis_writer) = maybe_analysis_writer.as_mut() {
-                                let analysis_output = analysis_writer.analyze(container).await
+                                let (analysis_file_len, heuristic_warning) = analysis_writer.analyze(container).await
                                     .expect("failed to analyze container");
-                                let (analysis_file_len, heuristic_warning) = analysis_output;
                                 if heuristic_warning {
                                     info!("a heuristic triggered on this run!");
                                     ui_update_sender.send(display::DisplayState::WarningDetected).await
                                         .expect("couldn't send ui update message: {}");
+                                    notification_channel.send(
+                                        Notification::new(
+                                            "heuristic-warning".to_string(),
+                                            "New warning triggered!".to_string(),
+                                            Some(Duration::from_secs(60*5)))
+                                        ).await.expect("Failed to send to notification channel");
                                 }
                                 let mut qmdl_store = qmdl_store_lock.write().await;
                                 let index = qmdl_store.current_entry.expect("DiagDevice had qmdl_writer, but QmdlStore didn't have current entry???");

bin/src/notifications.rs

@@ -0,0 +1,152 @@
+use std::{
+    cmp::min,
+    collections::HashMap,
+    time::{Duration, Instant},
+};
+
+use log::error;
+use tokio::sync::mpsc::{self, error::TryRecvError};
+use tokio_util::task::TaskTracker;
+
+static NTFY_BASE_URL: &str = "https://ntfy.sh/";
+
+pub struct Notification {
+    message_type: String,
+    message: String,
+    debounce: Option<Duration>,
+}
+
+impl Notification {
+    pub fn new(message_type: String, message: String, debounce: Option<Duration>) -> Self {
+        Notification {
+            message_type,
+            message,
+            debounce,
+        }
+    }
+}
+
+struct NotificationStatus {
+    message: String,
+    needs_sending: bool,
+    last_sent: Option<Instant>,
+    last_attempt: Option<Instant>,
+    failed_since_last_success: u32,
+}
+
+pub struct NotificationService {
+    channel_name: Option<String>,
+    tx: mpsc::Sender<Notification>,
+    rx: mpsc::Receiver<Notification>,
+}
+
+impl NotificationService {
+    pub fn new(channel_name: Option<String>) -> Self {
+        let (tx, rx) = mpsc::channel(10);
+        Self {
+            channel_name,
+            tx,
+            rx,
+        }
+    }
+
+    pub fn new_handler(&self) -> mpsc::Sender<Notification> {
+        return self.tx.clone();
+    }
+}
+
+pub fn run_notification_worker(
+    task_tracker: &TaskTracker,
+    mut notification_service: NotificationService,
+) {
+    task_tracker.spawn(async move {
+        let channel_name = notification_service.channel_name.unwrap_or("".into());
+
+        if channel_name != String::from("") {
+            let mut notification_statuses = HashMap::new();
+            let http_client = reqwest::Client::new();
+
+            loop {
+                // Get any notifications since the last time we checked
+                loop {
+                    match notification_service.rx.try_recv() {
+                        Ok(notification) => {
+                            let status = notification_statuses
+                                .entry(notification.message_type)
+                                .or_insert_with(|| NotificationStatus {
+                                    message: "".to_string(),
+                                    needs_sending: true,
+                                    last_sent: None,
+                                    last_attempt: None,
+                                    failed_since_last_success: 0,
+                                });
+                            // Ignore if we're in the debounce period
+                            if let Some(debounce) = notification.debounce {
+                                if let Some(last_sent) = status.last_sent {
+                                    if last_sent.elapsed() < debounce {
+                                        continue;
+                                    }
+                                }
+                            }
+                            status.message = notification.message;
+                            status.needs_sending = true;
+                        }
+                        Err(TryRecvError::Empty) => {
+                            break;
+                        }
+                        Err(TryRecvError::Disconnected) => {
+                            return;
+                        }
+                    }
+                }
+
+                // Attempt to send pending notifications
+                for notification in notification_statuses.values_mut() {
+                    if !notification.needs_sending {
+                        continue;
+                    }
+
+                    // Backoff retries, up to a maximum of 256 seconds.
+                    if let Some(last_attempt) = notification.last_attempt {
+                        let min_wait_time = Duration::from_secs(
+                            2u64.pow(min(notification.failed_since_last_success, 8)),
+                        );
+                        if last_attempt.elapsed() < min_wait_time {
+                            continue;
+                        }
+                    }
+
+                    match http_client
+                        .post(format!("{}{}", NTFY_BASE_URL, channel_name))
+                        .body(notification.message.clone())
+                        .send()
+                        .await
+                    {
+                        Ok(response) => {
+                            if response.status().is_success() {
+                                notification.last_sent = Some(Instant::now());
+                                notification.failed_since_last_success = 0;
+                                notification.needs_sending = false;
+                            } else {
+                                notification.failed_since_last_success += 1;
+                                notification.last_attempt = Some(Instant::now());
+                            }
+                        }
+                        Err(e) => {
+                            error!("Failed to send notification to ntfy: {}", e);
+                            notification.failed_since_last_success += 1;
+                        }
+                    }
+                }
+
+                tokio::time::sleep(Duration::from_secs(2)).await;
+            }
+        }
+        // If there's no channel name we'll just discard the notifications
+        else {
+            loop {
+                notification_service.rx.recv().await;
+            }
+        }
+    });
+}

bin/web/src/lib/components/ConfigForm.svelte

@@ -94,6 +94,17 @@
                     </select>
                 </div>
 
+                <div>
+                    <label for="ntfy_topic" class="block text-sm font-medium text-gray-700 mb-1">
+                        ntfy Topic
+					</label>
+					<input
+                        id="ntfy_topic"
+                        bind:value={config.ntfy_topic}
+                        class="w-full px-3 py-2 border border-gray-300 rounded-md focus:outline-none focus:ring-2 focus:ring-rayhunter-blue"
+                    >
+                </div>
+                
                 <div class="space-y-3">
                     <div class="flex items-center">
                         <input

bin/web/src/lib/utils.svelte.ts

@@ -12,6 +12,7 @@ export interface Config {
     ui_level: number;
     colorblind_mode: boolean;
     key_input_mode: number;
+    ntfy_topic: string;
     analyzers: AnalyzerConfig;
 }