//! Real-time interception. //! //! A background thread owns an inotify instance watching the configured //! directories (recursively — we walk each dir and add a watch per //! subdirectory, and add a watch when a new directory appears). For every //! file-appearing / file-written event it: //! //! 1. increments the "files seen" counter, //! 2. skips excluded paths, //! 3. runs the engine on that one file, //! 4. on a hit, either quarantines it (default) or just raises an alert, //! 5. feeds a rolling 60-second window of write events into the //! **ransomware heuristic** — a burst of writes past the configured //! per-minute threshold raises a ransomware alarm. //! //! The engine, quarantine store, event log, and settings are all //! `Arc`-shared, so the monitor and the RPC threads cooperate without //! locking the world. use hound_api::{QuarantineEntry, RealtimeStatus, Settings}; use std::collections::{HashMap, VecDeque}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::thread; use std::time::{Duration, SystemTime}; use crate::engine::engine; use crate::events::EventLog; use crate::quarantine::Quarantine; use crate::settings::SettingsStore; use inotify::{EventMask, Inotify, WatchDescriptor, WatchMask}; const WINDOW: Duration = Duration::from_secs(60); /// Mutable state the monitor writes and `status()` reads. #[derive(Default)] struct Counters { files_seen: u64, files_quarantined: u64, last_event_at: Option, ransomware: String, } impl Counters { fn new() -> Self { Self { ransomware: "calm".into(), ..Default::default() } } } /// The real-time monitor. Cheaply cloned (all state is shared). pub struct RealtimeMonitor { counters: Arc>, started_at: Arc>>, running: Arc, active: Arc, handle: Arc>>>, watch_dirs: Arc>>, settings: SettingsStore, quarantine: Quarantine, events: EventLog, } impl Clone for RealtimeMonitor { fn clone(&self) -> Self { Self { counters: Arc::clone(&self.counters), started_at: Arc::clone(&self.started_at), running: Arc::clone(&self.running), active: Arc::clone(&self.active), handle: Arc::clone(&self.handle), watch_dirs: Arc::clone(&self.watch_dirs), settings: self.settings.clone(), quarantine: self.quarantine.clone(), events: self.events.clone(), } } } impl RealtimeMonitor { pub fn new(settings: SettingsStore, quarantine: Quarantine, events: EventLog) -> Self { Self { counters: Arc::new(Mutex::new(Counters::new())), started_at: Arc::new(Mutex::new(None)), running: Arc::new(AtomicBool::new(false)), active: Arc::new(AtomicBool::new(false)), handle: Arc::new(Mutex::new(None)), watch_dirs: Arc::new(Mutex::new(Vec::new())), settings, quarantine, events, } } pub fn is_running(&self) -> bool { self.running.load(Ordering::Relaxed) } pub fn is_active(&self) -> bool { self.active.load(Ordering::Relaxed) } pub fn watch_dirs(&self) -> Vec { self.watch_dirs.lock().unwrap().clone() } pub fn status(&self) -> RealtimeStatus { let s = self.settings.get(); let c = self.counters.lock().unwrap(); let started = *self.started_at.lock().unwrap(); let uptime = started .and_then(|t| SystemTime::now().duration_since(t).ok()) .map(|d| d.as_secs()) .unwrap_or(0); RealtimeStatus { enabled: s.realtime_enabled && !s.paused, watching: self .watch_dirs() .iter() .map(|p| p.display().to_string()) .collect(), files_seen: c.files_seen, files_quarantined: c.files_quarantined, last_event_at: c.last_event_at.clone(), uptime_secs: uptime, active: self.is_active(), ransomware: c.ransomware.clone(), } } /// Start the monitor thread if it isn't already running. Idempotent. pub fn start(&self) -> Result<(), String> { if self.is_running() { return Ok(()); } // Expand configured watch dirs (tildes) to real paths. let s = self.settings.get(); let dirs: Vec = s .realtime_watch .iter() .filter_map(|d| resolve_dir(d)) .collect(); *self.watch_dirs.lock().unwrap() = dirs.clone(); if !self.is_active() { *self.started_at.lock().unwrap() = Some(SystemTime::now()); } self.running.store(true, Ordering::Relaxed); let counters = Arc::clone(&self.counters); let running = Arc::clone(&self.running); let active = Arc::clone(&self.active); let settings = self.settings.clone(); let quarantine = self.quarantine.clone(); let events = self.events.clone(); let watch_dirs = Arc::clone(&self.watch_dirs); let handle = thread::Builder::new() .name("houndd-realtime".into()) .spawn(move || { run_monitor( dirs, &counters, &running, &active, &settings, &quarantine, &events, &watch_dirs, ) }) .map_err(|e| e.to_string())?; *self.handle.lock().unwrap() = Some(handle); Ok(()) } /// Stop the monitor thread (sets the running flag; the loop exits at /// the next 1s tick). pub fn stop(&self) { self.running.store(false, Ordering::Relaxed); let h = self.handle.lock().unwrap().take(); if let Some(h) = h { let _ = h.join(); } self.active.store(false, Ordering::Relaxed); } } fn run_monitor( initial_dirs: Vec, counters: &Arc>, running: &Arc, active: &Arc, settings: &SettingsStore, quarantine: &Quarantine, events: &EventLog, watch_dirs: &Arc>>, ) { let Ok(mut inotify) = Inotify::init() else { eprintln!("realtime: inotify init failed — monitor idle"); return; }; // inotify 0.10 has no wd→path lookup, so we keep our own map. let mut wd_map: HashMap = HashMap::new(); let mask = WatchMask::CREATE | WatchMask::CLOSE_WRITE | WatchMask::MOVED_TO; for d in &initial_dirs { add_recursive_watches(&mut inotify, d, mask, &mut wd_map); } active.store(true, Ordering::Relaxed); eprintln!( "realtime: watching {} dir(s), {} watch(es)", initial_dirs.len(), wd_map.len() ); let mut buf = [0u8; 16_384]; let mut write_window: VecDeque = VecDeque::new(); let engine = engine(); while running.load(Ordering::Relaxed) { let evts = match inotify.read_events(&mut buf) { Ok(iter) => iter.collect::>(), // Non-blocking fd: nothing queued right now — sleep and retry. Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { thread::sleep(Duration::from_millis(100)); continue; } Err(e) => { eprintln!("realtime: inotify read error: {e} — backing off"); thread::sleep(Duration::from_millis(500)); continue; } }; if evts.is_empty() { continue; } let now = SystemTime::now(); for ev in &evts { let (is_write, is_new_dir, path) = match classify(ev, &wd_map) { Some(v) => v, None => continue, }; let mut c = counters.lock().unwrap(); if is_new_dir { // A new directory inside a watched dir → watch it too. if let Ok(wd) = inotify.watches().add(&path, mask) { wd_map.insert(wd, path.clone()); watch_dirs.lock().unwrap().push(path.clone()); } continue; } if !is_write { continue; } // File write/move event. c.files_seen += 1; c.last_event_at = Some(crate::engine::to_rfc3339(now)); drop(c); let s = settings.get(); if s.paused || !s.realtime_enabled { continue; } // Ransomware write-burst heuristic. write_window.push_back(now); while let Some(front) = write_window.front() { if now.duration_since(*front).unwrap_or_default() > WINDOW { write_window.pop_front(); } else { break; } } if s.ransomware_guard && (write_window.len() as u32) >= s.ransomware_threshold_per_min { let mut c = counters.lock().unwrap(); if c.ransomware != "alarm" { c.ransomware = "alarm".into(); } drop(c); events.push( "ransomware", "critical", format!( "write burst: {} file writes in 60s (threshold {}) — possible ransomware", write_window.len(), s.ransomware_threshold_per_min ), ); } if excluded(&path, &s.exclude_paths) { continue; } // Scan this single file with the engine. match engine.scan(path.to_str().unwrap_or(""), false) { Ok(r) if !r.is_clean() => { let found = r.found.first().cloned(); let virus = found .as_ref() .map(|f| f.virus.clone()) .unwrap_or_else(|| "unknown".into()); if s.on_detect == "quarantine" { match quarantine.add(path.to_str().unwrap_or(""), &virus) { Ok(entry) => { let mut c = counters.lock().unwrap(); c.files_quarantined += 1; drop(c); events.push( "quarantine", "critical", format!( "real-time: quarantined {} ({}) as {:?}", entry.original_path, virus, entry.id ), ); } Err(e) => { events.push( "realtime", "warn", format!("real-time: failed to quarantine {path:?}: {e}"), ); } } } else { events.push( "threat", "critical", format!("real-time: {} found {} (alert-only)", virus, path.display()), ); } } _ => {} } } } active.store(false, Ordering::Relaxed); } /// Walk `root` and add an inotify watch to it and every subdirectory, /// recording each watch descriptor for later path resolution. fn add_recursive_watches( inotify: &mut Inotify, root: &Path, mask: WatchMask, wd_map: &mut HashMap, ) { let mut stack = vec![root.to_path_buf()]; while let Some(dir) = stack.pop() { if let Ok(wd) = inotify.watches().add(&dir, mask) { wd_map.insert(wd, dir.clone()); } if let Ok(rd) = std::fs::read_dir(&dir) { for entry in rd.flatten() { let p = entry.path(); if p.is_dir() { stack.push(p); } } } } } /// Decide what an inotify event means for us. /// Returns (is_file_write_event, is_new_dir, resolved_path). fn classify>( ev: &inotify::Event, wd_map: &HashMap, ) -> Option<(bool, bool, PathBuf)> { let base = wd_map.get(&ev.wd)?.clone(); let mut path = base.clone(); if let Some(name) = &ev.name { let name = name.as_ref().to_string_lossy(); if !name.is_empty() { path = path.join(name.as_ref()); } } let write = ev .mask .intersects(EventMask::MOVED_TO | EventMask::CREATE | EventMask::CLOSE_WRITE); // A child path that resolves to a directory is a new dir to watch. let is_new_dir = path != base && path.is_dir(); Some((write, is_new_dir, path)) } /// Expand a possibly-tilde path to a real directory, or None if it doesn't /// resolve to an existing dir. pub fn resolve_dir(spec: &str) -> Option { let spec = spec.trim(); if spec.is_empty() { return None; } let path = if let Some(rest) = spec.strip_prefix("~/") { let home = std::env::var("HOME").ok()?; PathBuf::from(home).join(rest) } else { PathBuf::from(spec) }; let canonical = path.canonicalize().ok()?; if canonical.is_dir() { Some(canonical) } else { None } } /// True when `path` falls under any exclude entry (prefix match). pub fn excluded(path: &Path, excludes: &[String]) -> bool { let s = path.to_string_lossy(); excludes.iter().any(|e| { let e = e.trim(); if e.is_empty() { return false; } if e.ends_with('/') { s.starts_with(e) } else { s == e || s.starts_with(&format!("{e}/")) } }) } /// Convenience for the RPC layer: quarantine one file and log it. pub fn quarantine_and_log( quarantine: &Quarantine, events: &EventLog, path: &str, virus: &str, ) -> anyhow::Result { let entry = quarantine.add(path, virus)?; events.push( "quarantine", "critical", format!( "quarantined {} ({}) as {:?}", entry.original_path, virus, entry.id ), ); Ok(entry) } /// Re-exported so callers don't reach into `settings` directly for the type. #[allow(dead_code)] pub(crate) fn settings_type() -> Settings { Settings::default() } #[cfg(test)] mod tests { use super::*; #[test] fn resolve_dir_expands_tilde() { let home = std::env::var("HOME").unwrap(); let p = resolve_dir("~/").unwrap(); assert!(p.starts_with(PathBuf::from(&home))); } #[test] fn resolve_dir_missing_is_none() { assert!(resolve_dir("/no/such/dir/here").is_none()); } #[test] fn excluded_prefix_match() { let p = PathBuf::from("/proc/self/1"); assert!(excluded(&p, &["/proc".into()])); assert!(excluded(&p, &["/proc/".into()])); assert!(!excluded(&p, &["/home".into()])); } #[test] fn monitor_start_stop_is_clean() { let dir = std::env::temp_dir().join(format!("hound-rt-{}", std::process::id())); let _ = std::fs::create_dir_all(&dir); // Isolate settings + quarantine data. let cfg = std::env::temp_dir().join(format!("hound-rt-cfg-{}", std::process::id())); let data = std::env::temp_dir().join(format!("hound-rt-data-{}", std::process::id())); let _env_guard = crate::test_util::locked(); std::env::set_var("XDG_CONFIG_HOME", &cfg); std::env::set_var("XDG_DATA_HOME", &data); let settings = SettingsStore::load(); let events = EventLog::new(); let quarantine = Quarantine::new(); let mon = RealtimeMonitor::new(settings, quarantine, events); mon.start().unwrap(); assert!(mon.is_running()); std::thread::sleep(Duration::from_millis(150)); let st = mon.status(); assert!(st.uptime_secs < 5); mon.stop(); assert!(!mon.is_running()); std::env::remove_var("XDG_CONFIG_HOME"); std::env::remove_var("XDG_DATA_HOME"); for d in [&dir, &cfg, &data] { let _ = std::fs::remove_dir_all(d); } } }