nexus_shield/endpoint/
mod.rs

1// ============================================================================
2// File: endpoint/mod.rs
3// Description: Real-time endpoint protection engine — core types and orchestrator
4// Author: Andrew Jewell Sr. - AutomataNexus
5// Updated: March 24, 2026
6// ============================================================================
7//! # Endpoint Protection Engine
8//!
9//! Real-time file, process, network, and memory monitoring with multi-engine
10//! malware detection. Developer-aware allowlisting eliminates false positives
11//! on dev machines.
12
13pub mod allowlist;
14pub mod container_scanner;
15pub mod dns_filter;
16pub mod file_quarantine;
17pub mod fim;
18pub mod heuristics;
19pub mod memory_scanner;
20pub mod network_monitor;
21pub mod process_monitor;
22pub mod rootkit_detector;
23pub mod signatures;
24pub mod supply_chain;
25pub mod threat_intel;
26pub mod usb_monitor;
27pub mod watcher;
28pub mod yara_engine;
29
30use chrono::{DateTime, Utc};
31use parking_lot::RwLock;
32use serde::{Deserialize, Serialize};
33use std::collections::VecDeque;
34use std::path::{Path, PathBuf};
35use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
36use std::sync::Arc;
37
38use crate::audit_chain::{AuditChain, SecurityEventType};
39
40// =============================================================================
41// Severity
42// =============================================================================
43
44/// Detection severity levels, ordered from lowest to highest.
45#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
46pub enum Severity {
47    Info,
48    Low,
49    Medium,
50    High,
51    Critical,
52}
53
54impl std::fmt::Display for Severity {
55    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
56        match self {
57            Self::Info => write!(f, "info"),
58            Self::Low => write!(f, "low"),
59            Self::Medium => write!(f, "medium"),
60            Self::High => write!(f, "high"),
61            Self::Critical => write!(f, "critical"),
62        }
63    }
64}
65
66// =============================================================================
67// Detection Category
68// =============================================================================
69
70/// Category of a detection, carrying module-specific metadata.
71#[derive(Debug, Clone, Serialize, Deserialize)]
72pub enum DetectionCategory {
73    MalwareSignature { name: String, family: String },
74    HeuristicAnomaly { rule: String },
75    SuspiciousProcess { pid: u32, name: String },
76    NetworkAnomaly { connection: String },
77    MemoryAnomaly { pid: u32, region: String },
78    RootkitIndicator { technique: String },
79    YaraMatch { rule_name: String, tags: Vec<String> },
80    FilelessMalware { technique: String },
81}
82
83// =============================================================================
84// Recommended Action
85// =============================================================================
86
87/// Action the engine recommends after a detection.
88#[derive(Debug, Clone, Serialize, Deserialize)]
89pub enum RecommendedAction {
90    LogOnly,
91    Alert,
92    Quarantine { source_path: PathBuf },
93    KillProcess { pid: u32 },
94    BlockConnection { addr: String },
95    Multi(Vec<RecommendedAction>),
96}
97
98impl std::fmt::Display for RecommendedAction {
99    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
100        match self {
101            Self::LogOnly => write!(f, "log_only"),
102            Self::Alert => write!(f, "alert"),
103            Self::Quarantine { source_path } => {
104                write!(f, "quarantine({})", source_path.display())
105            }
106            Self::KillProcess { pid } => write!(f, "kill({})", pid),
107            Self::BlockConnection { addr } => write!(f, "block({})", addr),
108            Self::Multi(actions) => {
109                let names: Vec<String> = actions.iter().map(|a| a.to_string()).collect();
110                write!(f, "multi[{}]", names.join(", "))
111            }
112        }
113    }
114}
115
116// =============================================================================
117// Scan Result
118// =============================================================================
119
120/// Unified result returned by every scanner engine.
121#[derive(Debug, Clone, Serialize, Deserialize)]
122pub struct ScanResult {
123    /// Unique detection ID.
124    pub id: String,
125    /// When the detection occurred.
126    pub timestamp: DateTime<Utc>,
127    /// Which scanner produced this result.
128    pub scanner: String,
129    /// What was scanned (file path, PID, connection, etc.).
130    pub target: String,
131    /// Severity of the detection.
132    pub severity: Severity,
133    /// Category with detection-specific metadata.
134    pub category: DetectionCategory,
135    /// Human-readable description.
136    pub description: String,
137    /// Confidence score 0.0–1.0.
138    pub confidence: f64,
139    /// Recommended action.
140    pub action: RecommendedAction,
141    /// SHA-256 of the scanned artifact (if applicable).
142    pub artifact_hash: Option<String>,
143}
144
145impl ScanResult {
146    /// Create a new ScanResult with auto-generated ID and timestamp.
147    pub fn new(
148        scanner: impl Into<String>,
149        target: impl Into<String>,
150        severity: Severity,
151        category: DetectionCategory,
152        description: impl Into<String>,
153        confidence: f64,
154        action: RecommendedAction,
155    ) -> Self {
156        Self {
157            id: uuid::Uuid::new_v4().to_string(),
158            timestamp: Utc::now(),
159            scanner: scanner.into(),
160            target: target.into(),
161            severity,
162            category,
163            description: description.into(),
164            confidence: confidence.clamp(0.0, 1.0),
165            action,
166            artifact_hash: None,
167        }
168    }
169
170    /// Attach an artifact hash to this result.
171    pub fn with_hash(mut self, hash: String) -> Self {
172        self.artifact_hash = Some(hash);
173        self
174    }
175}
176
177// =============================================================================
178// Scanner Trait
179// =============================================================================
180
181/// Trait that all scanning engines implement.
182#[async_trait::async_trait]
183pub trait Scanner: Send + Sync {
184    /// Human-readable name of this scanner.
185    fn name(&self) -> &str;
186
187    /// Whether this scanner is currently enabled and operational.
188    fn is_active(&self) -> bool;
189
190    /// Scan a file on disk. Returns empty vec if clean.
191    async fn scan_file(&self, path: &Path) -> Vec<ScanResult>;
192
193    /// Scan raw bytes (for in-memory content). Default: no-op.
194    async fn scan_bytes(&self, _data: &[u8], _label: &str) -> Vec<ScanResult> {
195        Vec::new()
196    }
197
198    /// Scan a running process by PID. Default: no-op.
199    async fn scan_process(&self, _pid: u32) -> Vec<ScanResult> {
200        Vec::new()
201    }
202}
203
204// =============================================================================
205// Endpoint Configuration
206// =============================================================================
207
208/// Top-level configuration for the endpoint protection engine.
209#[derive(Debug, Clone, Serialize, Deserialize)]
210pub struct EndpointConfig {
211    pub enabled: bool,
212    pub enable_watcher: bool,
213    pub enable_process_monitor: bool,
214    pub enable_network_monitor: bool,
215    pub enable_memory_scanner: bool,
216    pub enable_rootkit_detector: bool,
217    pub enable_dns_filter: bool,
218    pub enable_usb_monitor: bool,
219    pub enable_fim: bool,
220    pub data_dir: PathBuf,
221    pub watcher: watcher::WatcherConfig,
222    pub process_monitor: process_monitor::ProcessMonitorConfig,
223    pub network_monitor: network_monitor::NetworkMonitorConfig,
224    pub memory_scanner: memory_scanner::MemoryScanConfig,
225    pub rootkit_detector: rootkit_detector::RootkitConfig,
226    pub heuristics: heuristics::HeuristicConfig,
227    pub quarantine: file_quarantine::QuarantineVaultConfig,
228    pub allowlist: allowlist::AllowlistConfig,
229    pub threat_intel: threat_intel::ThreatIntelConfig,
230    pub signatures: signatures::SignatureConfig,
231    pub dns_filter: dns_filter::DnsFilterConfig,
232    pub usb_monitor: usb_monitor::UsbMonitorConfig,
233    pub fim: fim::FimConfig,
234}
235
236impl Default for EndpointConfig {
237    fn default() -> Self {
238        let data_dir = dirs_or_default();
239        Self {
240            enabled: true,
241            enable_watcher: true,
242            enable_process_monitor: true,
243            enable_network_monitor: true,
244            enable_memory_scanner: false, // requires elevated privileges
245            enable_rootkit_detector: false, // requires root
246            enable_dns_filter: false, // opt-in: requires configuring system DNS
247            enable_usb_monitor: true, // on by default: monitors for USB insertions
248            enable_fim: false, // opt-in: baselines system files, alerts on changes
249            data_dir: data_dir.clone(),
250            watcher: watcher::WatcherConfig::default(),
251            process_monitor: process_monitor::ProcessMonitorConfig::default(),
252            network_monitor: network_monitor::NetworkMonitorConfig::default(),
253            memory_scanner: memory_scanner::MemoryScanConfig::default(),
254            rootkit_detector: rootkit_detector::RootkitConfig::new(data_dir.clone()),
255            heuristics: heuristics::HeuristicConfig::default(),
256            quarantine: file_quarantine::QuarantineVaultConfig::new(data_dir.join("quarantine")),
257            allowlist: allowlist::AllowlistConfig::default(),
258            threat_intel: threat_intel::ThreatIntelConfig::new(data_dir.join("threat-intel")),
259            signatures: signatures::SignatureConfig::new(data_dir.join("signatures.ndjson")),
260            dns_filter: dns_filter::DnsFilterConfig::default(),
261            usb_monitor: usb_monitor::UsbMonitorConfig::default(),
262            fim: fim::FimConfig::default(),
263        }
264    }
265}
266
267fn dirs_or_default() -> PathBuf {
268    std::env::var("HOME")
269        .map(|h| PathBuf::from(h).join(".nexus-shield"))
270        .unwrap_or_else(|_| PathBuf::from("/tmp/nexus-shield"))
271}
272
273// =============================================================================
274// Endpoint Stats
275// =============================================================================
276
277/// Runtime statistics for the endpoint protection engine.
278#[derive(Debug, Clone, Serialize, Deserialize)]
279pub struct EndpointStats {
280    pub total_files_scanned: u64,
281    pub total_threats_detected: u64,
282    pub active_monitors: Vec<String>,
283    pub quarantined_files: usize,
284    pub last_scan_time: Option<DateTime<Utc>>,
285    pub scanners_active: Vec<String>,
286}
287
288// =============================================================================
289// Endpoint Engine
290// =============================================================================
291
292/// Orchestrates all endpoint protection subsystems.
293pub struct EndpointEngine {
294    /// All registered scanning engines.
295    scanners: Vec<Arc<dyn Scanner>>,
296    /// Developer-aware allowlist.
297    pub allowlist: Arc<allowlist::DeveloperAllowlist>,
298    /// Threat intelligence database.
299    pub threat_intel: Arc<threat_intel::ThreatIntelDB>,
300    /// File quarantine vault.
301    pub quarantine: Arc<file_quarantine::QuarantineVault>,
302    /// DNS filtering proxy.
303    pub dns_filter: Option<Arc<dns_filter::DnsFilter>>,
304    /// Container image scanner (on-demand).
305    pub container_scanner: container_scanner::ContainerScanner,
306    /// Supply chain dependency scanner (on-demand).
307    pub supply_chain_scanner: supply_chain::SupplyChainScanner,
308    /// Broadcast channel for real-time scan results.
309    result_tx: tokio::sync::broadcast::Sender<ScanResult>,
310    /// Detection history (ring buffer).
311    history: Arc<RwLock<VecDeque<ScanResult>>>,
312    /// Configuration.
313    config: EndpointConfig,
314    /// Counters (Arc-wrapped for safe sharing across spawned tasks).
315    files_scanned: Arc<AtomicU64>,
316    threats_detected: Arc<AtomicU64>,
317    /// Whether the engine is running.
318    running: AtomicBool,
319}
320
321impl EndpointEngine {
322    /// Create a new endpoint engine with the given configuration.
323    pub fn new(config: EndpointConfig) -> Self {
324        let (result_tx, _) = tokio::sync::broadcast::channel(1024);
325
326        // Ensure data directory exists
327        let _ = std::fs::create_dir_all(&config.data_dir);
328        let _ = std::fs::create_dir_all(config.data_dir.join("quarantine"));
329        let _ = std::fs::create_dir_all(config.data_dir.join("threat-intel"));
330        tracing::info!(data_dir = %config.data_dir.display(), "Endpoint data directory initialized");
331
332        // Initialize subsystems
333        let allowlist = Arc::new(allowlist::DeveloperAllowlist::new(config.allowlist.clone()));
334        let threat_intel = Arc::new(threat_intel::ThreatIntelDB::new(config.threat_intel.clone()));
335        let quarantine = Arc::new(file_quarantine::QuarantineVault::new(config.quarantine.clone()));
336
337        // DNS filter (if enabled)
338        let dns_filter = if config.enable_dns_filter {
339            Some(Arc::new(dns_filter::DnsFilter::new(
340                config.dns_filter.clone(),
341                Arc::clone(&threat_intel),
342            )))
343        } else {
344            None
345        };
346
347        // Build scanner list
348        let mut scanners: Vec<Arc<dyn Scanner>> = Vec::new();
349
350        // Signature engine
351        let sig_engine = signatures::SignatureEngine::new(config.signatures.clone());
352        scanners.push(Arc::new(sig_engine));
353
354        // Heuristic engine
355        let heur_engine = heuristics::HeuristicEngine::new(config.heuristics.clone());
356        scanners.push(Arc::new(heur_engine));
357
358        // YARA engine
359        let yara = yara_engine::YaraEngine::new(None);
360        scanners.push(Arc::new(yara));
361
362        // On-demand scanners
363        let container_scanner = container_scanner::ContainerScanner::new(
364            container_scanner::ContainerScanConfig::default(),
365        );
366        let supply_chain_scanner = supply_chain::SupplyChainScanner::new(
367            supply_chain::SupplyChainConfig::default(),
368        );
369
370        Self {
371            scanners,
372            allowlist,
373            threat_intel,
374            quarantine,
375            dns_filter,
376            container_scanner,
377            supply_chain_scanner,
378            result_tx,
379            history: Arc::new(RwLock::new(VecDeque::with_capacity(10000))),
380            config,
381            files_scanned: Arc::new(AtomicU64::new(0)),
382            threats_detected: Arc::new(AtomicU64::new(0)),
383            running: AtomicBool::new(false),
384        }
385    }
386
387    /// Start all background monitors. Returns JoinHandles for spawned tasks.
388    pub async fn start(&self, audit: Arc<AuditChain>) -> Vec<tokio::task::JoinHandle<()>> {
389        self.running.store(true, Ordering::SeqCst);
390        let mut handles = Vec::new();
391
392        // Record startup event
393        audit.record(
394            SecurityEventType::EndpointScanStarted,
395            "system",
396            "Endpoint protection engine started",
397            0.0,
398        );
399
400        // Start file watcher
401        if self.config.enable_watcher {
402            let (scan_tx, mut scan_rx) = tokio::sync::mpsc::unbounded_channel::<PathBuf>();
403            let watcher_handle = watcher::FileWatcher::new(
404                self.config.watcher.clone(),
405                scan_tx,
406            );
407
408            let allowlist = Arc::clone(&self.allowlist);
409            let _watcher_task = watcher_handle.start(allowlist);
410
411            // File scan consumer task
412            let scanners = self.scanners.clone();
413            let result_tx = self.result_tx.clone();
414            let history = Arc::clone(&self.history);
415            let quarantine = Arc::clone(&self.quarantine);
416            let audit2 = Arc::clone(&audit);
417            let files_scanned = Arc::clone(&self.files_scanned);
418            let threats_detected = Arc::clone(&self.threats_detected);
419
420            let handle = tokio::spawn(async move {
421                while let Some(path) = scan_rx.recv().await {
422                    tracing::debug!(file = %path.display(), "Scanning file");
423
424                    // Run all scanners on the file
425                    let mut all_results = Vec::new();
426                    for scanner in &scanners {
427                        if scanner.is_active() {
428                            let results = scanner.scan_file(&path).await;
429                            all_results.extend(results);
430                        }
431                    }
432
433                    files_scanned.fetch_add(1, Ordering::Relaxed);
434
435                    if all_results.is_empty() {
436                        tracing::trace!(file = %path.display(), "File clean");
437                    } else {
438                        tracing::warn!(
439                            file = %path.display(),
440                            threats = all_results.len(),
441                            "THREAT DETECTED"
442                        );
443                    }
444
445                    // Process results
446                    for result in all_results {
447                        threats_detected.fetch_add(1, Ordering::Relaxed);
448
449                        // Quarantine if needed
450                        if let RecommendedAction::Quarantine { ref source_path } = result.action {
451                            let _ = quarantine.quarantine_file(
452                                source_path,
453                                &result.description,
454                                &result.scanner,
455                                result.severity,
456                            );
457                        }
458
459                        // Record to audit chain
460                        audit2.record(
461                            SecurityEventType::MalwareDetected,
462                            &result.target,
463                            &result.description,
464                            result.confidence,
465                        );
466
467                        // Broadcast and save to history
468                        let _ = result_tx.send(result.clone());
469                        let mut hist = history.write();
470                        if hist.len() >= 10000 {
471                            hist.pop_front();
472                        }
473                        hist.push_back(result);
474                    }
475                }
476            });
477            handles.push(handle);
478        }
479
480        // Start DNS filter proxy
481        if self.config.enable_dns_filter {
482            if let Some(ref dns) = self.dns_filter {
483                let (dns_tx, mut dns_rx) = tokio::sync::mpsc::unbounded_channel::<ScanResult>();
484                let dns_handle = Arc::clone(dns).start(dns_tx);
485                handles.push(dns_handle);
486
487                // DNS detection consumer
488                let history = Arc::clone(&self.history);
489                let audit3 = Arc::clone(&audit);
490                let result_tx = self.result_tx.clone();
491                let threats_detected = Arc::clone(&self.threats_detected);
492                let dns_consumer = tokio::spawn(async move {
493                    while let Some(result) = dns_rx.recv().await {
494                        threats_detected.fetch_add(1, Ordering::Relaxed);
495                        audit3.record(
496                            SecurityEventType::MalwareDetected,
497                            &result.target,
498                            &result.description,
499                            result.confidence,
500                        );
501                        let _ = result_tx.send(result.clone());
502                        let mut hist = history.write();
503                        if hist.len() >= 10000 {
504                            hist.pop_front();
505                        }
506                        hist.push_back(result);
507                    }
508                });
509                handles.push(dns_consumer);
510            }
511        }
512
513        // Start USB monitor
514        if self.config.enable_usb_monitor {
515            let (usb_tx, mut usb_rx) = tokio::sync::mpsc::unbounded_channel::<ScanResult>();
516            let usb_mon = Arc::new(usb_monitor::UsbMonitor::new(self.config.usb_monitor.clone()));
517            let usb_handle = Arc::clone(&usb_mon).start(usb_tx);
518            handles.push(usb_handle);
519
520            // USB detection consumer
521            let history = Arc::clone(&self.history);
522            let audit4 = Arc::clone(&audit);
523            let result_tx = self.result_tx.clone();
524            let threats_detected = Arc::clone(&self.threats_detected);
525            let usb_consumer = tokio::spawn(async move {
526                while let Some(result) = usb_rx.recv().await {
527                    threats_detected.fetch_add(1, Ordering::Relaxed);
528                    audit4.record(
529                        SecurityEventType::MalwareDetected,
530                        &result.target,
531                        &result.description,
532                        result.confidence,
533                    );
534                    let _ = result_tx.send(result.clone());
535                    let mut hist = history.write();
536                    if hist.len() >= 10000 {
537                        hist.pop_front();
538                    }
539                    hist.push_back(result);
540                }
541            });
542            handles.push(usb_consumer);
543        }
544
545        // Start process monitor
546        if self.config.enable_process_monitor {
547            let (pm_tx, mut pm_rx) = tokio::sync::mpsc::unbounded_channel::<ScanResult>();
548            let proc_mon = Arc::new(process_monitor::ProcessMonitor::new(self.config.process_monitor.clone()));
549            let pm_handle = Arc::clone(&proc_mon).start(pm_tx);
550            handles.push(pm_handle);
551
552            let history = Arc::clone(&self.history);
553            let audit_pm = Arc::clone(&audit);
554            let result_tx = self.result_tx.clone();
555            let threats_detected = Arc::clone(&self.threats_detected);
556            let pm_consumer = tokio::spawn(async move {
557                while let Some(result) = pm_rx.recv().await {
558                    threats_detected.fetch_add(1, Ordering::Relaxed);
559                    audit_pm.record(SecurityEventType::SuspiciousProcess, &result.target, &result.description, result.confidence);
560                    let _ = result_tx.send(result.clone());
561                    let mut hist = history.write();
562                    if hist.len() >= 10000 { hist.pop_front(); }
563                    hist.push_back(result);
564                }
565            });
566            handles.push(pm_consumer);
567        }
568
569        // Start network monitor
570        if self.config.enable_network_monitor {
571            let (nm_tx, mut nm_rx) = tokio::sync::mpsc::unbounded_channel::<ScanResult>();
572            let net_mon = Arc::new(network_monitor::NetworkMonitor::new(
573                self.config.network_monitor.clone(),
574                Arc::clone(&self.threat_intel),
575            ));
576            let nm_handle = Arc::clone(&net_mon).start(nm_tx);
577            handles.push(nm_handle);
578
579            let history = Arc::clone(&self.history);
580            let audit_nm = Arc::clone(&audit);
581            let result_tx = self.result_tx.clone();
582            let threats_detected = Arc::clone(&self.threats_detected);
583            let nm_consumer = tokio::spawn(async move {
584                while let Some(result) = nm_rx.recv().await {
585                    threats_detected.fetch_add(1, Ordering::Relaxed);
586                    audit_nm.record(SecurityEventType::SuspiciousNetwork, &result.target, &result.description, result.confidence);
587                    let _ = result_tx.send(result.clone());
588                    let mut hist = history.write();
589                    if hist.len() >= 10000 { hist.pop_front(); }
590                    hist.push_back(result);
591                }
592            });
593            handles.push(nm_consumer);
594        }
595
596        // Start memory scanner
597        if self.config.enable_memory_scanner {
598            let (ms_tx, mut ms_rx) = tokio::sync::mpsc::unbounded_channel::<ScanResult>();
599            let mem_scan = Arc::new(memory_scanner::MemoryScanner::new(self.config.memory_scanner.clone()));
600            let ms_handle = Arc::clone(&mem_scan).start(ms_tx);
601            handles.push(ms_handle);
602
603            let history = Arc::clone(&self.history);
604            let audit_ms = Arc::clone(&audit);
605            let result_tx = self.result_tx.clone();
606            let threats_detected = Arc::clone(&self.threats_detected);
607            let ms_consumer = tokio::spawn(async move {
608                while let Some(result) = ms_rx.recv().await {
609                    threats_detected.fetch_add(1, Ordering::Relaxed);
610                    audit_ms.record(SecurityEventType::MemoryAnomaly, &result.target, &result.description, result.confidence);
611                    let _ = result_tx.send(result.clone());
612                    let mut hist = history.write();
613                    if hist.len() >= 10000 { hist.pop_front(); }
614                    hist.push_back(result);
615                }
616            });
617            handles.push(ms_consumer);
618        }
619
620        // Start rootkit detector
621        if self.config.enable_rootkit_detector {
622            let (rk_tx, mut rk_rx) = tokio::sync::mpsc::unbounded_channel::<ScanResult>();
623            let rk_det = Arc::new(rootkit_detector::RootkitDetector::new(self.config.rootkit_detector.clone()));
624            let rk_handle = Arc::clone(&rk_det).start(rk_tx);
625            handles.push(rk_handle);
626
627            let history = Arc::clone(&self.history);
628            let audit_rk = Arc::clone(&audit);
629            let result_tx = self.result_tx.clone();
630            let threats_detected = Arc::clone(&self.threats_detected);
631            let rk_consumer = tokio::spawn(async move {
632                while let Some(result) = rk_rx.recv().await {
633                    threats_detected.fetch_add(1, Ordering::Relaxed);
634                    audit_rk.record(SecurityEventType::RootkitIndicator, &result.target, &result.description, result.confidence);
635                    let _ = result_tx.send(result.clone());
636                    let mut hist = history.write();
637                    if hist.len() >= 10000 { hist.pop_front(); }
638                    hist.push_back(result);
639                }
640            });
641            handles.push(rk_consumer);
642        }
643
644        // Start File Integrity Monitor
645        if self.config.enable_fim {
646            let (fim_tx, mut fim_rx) = tokio::sync::mpsc::unbounded_channel::<ScanResult>();
647            let fim_mon = Arc::new(fim::FimMonitor::new(self.config.fim.clone()));
648            let fim_handle = Arc::clone(&fim_mon).start(fim_tx);
649            handles.push(fim_handle);
650
651            // FIM detection consumer
652            let history = Arc::clone(&self.history);
653            let audit5 = Arc::clone(&audit);
654            let result_tx = self.result_tx.clone();
655            let threats_detected = Arc::clone(&self.threats_detected);
656            let fim_consumer = tokio::spawn(async move {
657                while let Some(result) = fim_rx.recv().await {
658                    threats_detected.fetch_add(1, Ordering::Relaxed);
659                    audit5.record(
660                        SecurityEventType::MalwareDetected,
661                        &result.target,
662                        &result.description,
663                        result.confidence,
664                    );
665                    let _ = result_tx.send(result.clone());
666                    let mut hist = history.write();
667                    if hist.len() >= 10000 {
668                        hist.pop_front();
669                    }
670                    hist.push_back(result);
671                }
672            });
673            handles.push(fim_consumer);
674        }
675
676        handles
677    }
678
679    /// Scan a single file with all engines.
680    pub async fn scan_file(&self, path: &Path) -> Vec<ScanResult> {
681        if self.allowlist.should_skip_path(path) {
682            return Vec::new();
683        }
684
685        self.files_scanned.fetch_add(1, Ordering::Relaxed);
686        let mut results = Vec::new();
687
688        // Check if it's a dependency lock file (supply chain scan)
689        if supply_chain::SupplyChainScanner::detect_ecosystem(path).is_some() {
690            let mut sc_results = self.supply_chain_scanner.scan_file(path);
691            results.append(&mut sc_results);
692        }
693
694        for scanner in &self.scanners {
695            if scanner.is_active() {
696                let mut r = scanner.scan_file(path).await;
697                results.append(&mut r);
698            }
699        }
700
701        if !results.is_empty() {
702            self.threats_detected
703                .fetch_add(results.len() as u64, Ordering::Relaxed);
704            let mut hist = self.history.write();
705            for r in &results {
706                if hist.len() >= 10000 {
707                    hist.pop_front();
708                }
709                hist.push_back(r.clone());
710            }
711        }
712
713        results
714    }
715
716    /// Scan a directory recursively.
717    pub async fn scan_dir(&self, dir: &Path) -> Vec<ScanResult> {
718        let mut results = Vec::new();
719        if let Ok(entries) = std::fs::read_dir(dir) {
720            for entry in entries.flatten() {
721                let path = entry.path();
722                if path.is_dir() {
723                    if !self.allowlist.should_skip_path(&path) {
724                        let mut r = Box::pin(self.scan_dir(&path)).await;
725                        results.append(&mut r);
726                    }
727                } else if path.is_file() {
728                    let mut r = self.scan_file(&path).await;
729                    results.append(&mut r);
730                }
731            }
732        }
733        results
734    }
735
736    /// Scan a Docker image for security issues.
737    pub fn scan_container_image(&self, image: &str) -> Vec<ScanResult> {
738        self.container_scanner.scan_image(image)
739    }
740
741    /// Scan a dependency lock file for supply chain risks.
742    pub fn scan_dependencies(&self, path: &Path) -> Vec<ScanResult> {
743        self.supply_chain_scanner.scan_file(path)
744    }
745
746    /// Subscribe to real-time scan results.
747    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<ScanResult> {
748        self.result_tx.subscribe()
749    }
750
751    /// Get recent detection history.
752    pub fn recent_detections(&self, count: usize) -> Vec<ScanResult> {
753        let hist = self.history.read();
754        hist.iter().rev().take(count).cloned().collect()
755    }
756
757    /// Get runtime statistics.
758    pub fn stats(&self) -> EndpointStats {
759        let mut active = Vec::new();
760        if self.config.enable_watcher {
761            active.push("file_watcher".to_string());
762        }
763        if self.config.enable_process_monitor {
764            active.push("process_monitor".to_string());
765        }
766        if self.config.enable_network_monitor {
767            active.push("network_monitor".to_string());
768        }
769        if self.config.enable_memory_scanner {
770            active.push("memory_scanner".to_string());
771        }
772        if self.config.enable_rootkit_detector {
773            active.push("rootkit_detector".to_string());
774        }
775        if self.config.enable_dns_filter {
776            active.push("dns_filter".to_string());
777        }
778        if self.config.enable_usb_monitor {
779            active.push("usb_monitor".to_string());
780        }
781        if self.config.enable_fim {
782            active.push("fim".to_string());
783        }
784
785        let scanner_names: Vec<String> = self
786            .scanners
787            .iter()
788            .filter(|s| s.is_active())
789            .map(|s| s.name().to_string())
790            .collect();
791
792        EndpointStats {
793            total_files_scanned: self.files_scanned.load(Ordering::Relaxed),
794            total_threats_detected: self.threats_detected.load(Ordering::Relaxed),
795            active_monitors: active,
796            quarantined_files: self.quarantine.list_entries().len(),
797            last_scan_time: self.history.read().back().map(|r| r.timestamp),
798            scanners_active: scanner_names,
799        }
800    }
801
802    /// Check if the engine is running.
803    pub fn is_running(&self) -> bool {
804        self.running.load(Ordering::SeqCst)
805    }
806}
807
808// =============================================================================
809// Tests
810// =============================================================================
811
812#[cfg(test)]
813mod tests {
814    use super::*;
815
816    #[test]
817    fn test_severity_ordering() {
818        assert!(Severity::Info < Severity::Low);
819        assert!(Severity::Low < Severity::Medium);
820        assert!(Severity::Medium < Severity::High);
821        assert!(Severity::High < Severity::Critical);
822    }
823
824    #[test]
825    fn test_severity_display() {
826        assert_eq!(Severity::Critical.to_string(), "critical");
827        assert_eq!(Severity::Info.to_string(), "info");
828    }
829
830    #[test]
831    fn test_scan_result_creation() {
832        let result = ScanResult::new(
833            "test_scanner",
834            "/tmp/malware.exe",
835            Severity::High,
836            DetectionCategory::MalwareSignature {
837                name: "EICAR".to_string(),
838                family: "Test".to_string(),
839            },
840            "EICAR test file detected",
841            0.99,
842            RecommendedAction::Quarantine {
843                source_path: PathBuf::from("/tmp/malware.exe"),
844            },
845        );
846        assert!(!result.id.is_empty());
847        assert_eq!(result.scanner, "test_scanner");
848        assert_eq!(result.severity, Severity::High);
849        assert_eq!(result.confidence, 0.99);
850    }
851
852    #[test]
853    fn test_scan_result_with_hash() {
854        let result = ScanResult::new(
855            "sig",
856            "/tmp/test",
857            Severity::Low,
858            DetectionCategory::HeuristicAnomaly {
859                rule: "test".to_string(),
860            },
861            "test",
862            0.5,
863            RecommendedAction::LogOnly,
864        )
865        .with_hash("abc123".to_string());
866        assert_eq!(result.artifact_hash, Some("abc123".to_string()));
867    }
868
869    #[test]
870    fn test_confidence_clamping() {
871        let r1 = ScanResult::new(
872            "s", "t", Severity::Low,
873            DetectionCategory::HeuristicAnomaly { rule: "x".into() },
874            "d", 1.5, RecommendedAction::LogOnly,
875        );
876        assert_eq!(r1.confidence, 1.0);
877
878        let r2 = ScanResult::new(
879            "s", "t", Severity::Low,
880            DetectionCategory::HeuristicAnomaly { rule: "x".into() },
881            "d", -0.5, RecommendedAction::LogOnly,
882        );
883        assert_eq!(r2.confidence, 0.0);
884    }
885
886    #[test]
887    fn test_recommended_action_display() {
888        assert_eq!(RecommendedAction::LogOnly.to_string(), "log_only");
889        assert_eq!(RecommendedAction::Alert.to_string(), "alert");
890        assert_eq!(
891            RecommendedAction::KillProcess { pid: 1234 }.to_string(),
892            "kill(1234)"
893        );
894    }
895
896    #[test]
897    fn test_endpoint_config_default() {
898        let config = EndpointConfig::default();
899        assert!(config.enabled);
900        assert!(config.enable_watcher);
901        assert!(config.enable_process_monitor);
902        assert!(!config.enable_memory_scanner); // requires elevated
903        assert!(!config.enable_rootkit_detector); // requires root
904        assert!(!config.enable_dns_filter); // opt-in
905        assert!(config.enable_usb_monitor); // on by default
906        assert!(!config.enable_fim); // opt-in
907    }
908}
nexus_shield/endpoint/mod.rs

nexus_shield/endpoint/
mod.rs
