rsigma 0.12.0

CLI for parsing, validating, linting and evaluating Sigma detection rules
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

use std::path::Path;

use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use tokio::sync::mpsc;

/// Watches a path for file changes and sends reload signals via a channel.
///
/// In addition to the rules path, optionally watches pipeline file paths.
/// Any YAML change in watched paths triggers a reload signal.
pub fn spawn_file_watcher(
    rules_path: &Path,
    pipeline_paths: &[&Path],
    reload_tx: mpsc::Sender<()>,
) -> Option<RecommendedWatcher> {
    let tx = reload_tx.clone();
    let mut watcher = match RecommendedWatcher::new(
        move |res: Result<Event, notify::Error>| match res {
            Ok(event) => {
                if matches!(
                    event.kind,
                    EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_)
                ) {
                    let is_yaml = event.paths.iter().any(|p| {
                        matches!(p.extension().and_then(|e| e.to_str()), Some("yml" | "yaml"))
                    });
                    if is_yaml {
                        match tx.try_send(()) {
                            Ok(()) => {}
                            Err(mpsc::error::TrySendError::Full(_)) => {
                                tracing::debug!(
                                    "Reload channel full, event coalesced into pending reload"
                                );
                            }
                            Err(mpsc::error::TrySendError::Closed(_)) => {
                                tracing::warn!("Reload channel closed, watcher event dropped");
                            }
                        }
                    }
                }
            }
            Err(e) => {
                tracing::warn!(error = %e, "File watcher error");
            }
        },
        notify::Config::default(),
    ) {
        Ok(w) => w,
        Err(e) => {
            tracing::warn!(error = %e, "Could not create file watcher, hot-reload disabled");
            return None;
        }
    };

    if let Err(e) = watcher.watch(rules_path, RecursiveMode::Recursive) {
        tracing::warn!(error = %e, path = %rules_path.display(), "Could not watch rules path");
        return None;
    }

    tracing::info!(path = %rules_path.display(), "Watching rules directory for changes");

    for path in pipeline_paths {
        if let Err(e) = watcher.watch(path, RecursiveMode::NonRecursive) {
            tracing::warn!(
                error = %e,
                path = %path.display(),
                "Could not watch pipeline file"
            );
        } else {
            tracing::info!(path = %path.display(), "Watching pipeline file for changes");
        }
    }

    Some(watcher)
}

/// Set up a SIGHUP handler that sends reload signals and source re-resolution triggers.
#[cfg(unix)]
pub async fn sighup_listener(
    reload_tx: mpsc::Sender<()>,
    sources_trigger_tx: Option<mpsc::Sender<rsigma_runtime::sources::refresh::RefreshTrigger>>,
) {
    use tokio::signal::unix::{SignalKind, signal};

    let mut sig = match signal(SignalKind::hangup()) {
        Ok(s) => s,
        Err(e) => {
            tracing::warn!(error = %e, "Could not register SIGHUP handler");
            return;
        }
    };

    loop {
        sig.recv().await;
        tracing::info!("SIGHUP received, triggering reload and source re-resolution");
        let _ = reload_tx.try_send(());
        if let Some(tx) = &sources_trigger_tx {
            let _ = tx.try_send(rsigma_runtime::sources::refresh::RefreshTrigger::All);
        }
    }
}

#[cfg(not(unix))]
pub async fn sighup_listener(
    _reload_tx: mpsc::Sender<()>,
    _sources_trigger_tx: Option<mpsc::Sender<rsigma_runtime::sources::refresh::RefreshTrigger>>,
) {
    std::future::pending::<()>().await;
}