WP-20/21: vulnerability management with Trivy and mail notifications

Trivy scanner adapter (rootfs + image JSON, parsed and deduplicated),
findings repository, scan diff that keeps first_seen, marks disappeared
findings fixed and skips failed targets, vulnerability_scan job (daily by
default), digest mail for new findings at or above a configurable severity,
/api/vulnerabilities routes, Vulnerabilities page with severity tiles,
filters, details and acknowledge, notification threshold in settings.
Deploy script installs Trivy from the Aqua apt repository.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

backend/crates/application/src/tests/vuln_tests.rs

@@ -196,12 +196,13 @@ async fn new_findings_at_or_above_threshold_trigger_one_mail() {
         ]),
     )]);
     svc.scan(&VecLog::default()).await.unwrap();
-    let sent = f.mailer.0.lock().unwrap();
-    assert_eq!(sent.len(), 1);
-    assert!(sent[0].1.contains("1 new"), "{}", sent[0].1);
-    assert!(sent[0].2.contains("CVE-3"));
-    assert!(!sent[0].2.contains("CVE-4"), "below threshold");
-    drop(sent);
+    {
+        let sent = f.mailer.0.lock().unwrap();
+        assert_eq!(sent.len(), 1);
+        assert!(sent[0].1.contains("1 new"), "{}", sent[0].1);
+        assert!(sent[0].2.contains("CVE-3"));
+        assert!(!sent[0].2.contains("CVE-4"), "below threshold");
+    }
 
     f.settings
         .set_notify_min_severity(Severity::Low)

backend/crates/application/src/vuln_service.rs

@@ -1,11 +1,14 @@
 //! Vulnerability scanning: diffs scanner results against stored findings,
 //! notifies about new ones by mail.
+use std::collections::HashSet;
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use domain::ports::{ClusterGateway, FindingRepository, VulnerabilityScanner};
+use chrono::{DateTime, Utc};
+use domain::ports::{ClusterGateway, FindingRepository, LineSink, VulnerabilityScanner};
 use domain::vuln::{
-    Finding, FindingFilter, FindingStatus, ScanReport, Severity, SeverityCounts, TargetKind,
+    Finding, FindingFilter, FindingStatus, RawFinding, ScanReport, Severity, SeverityCounts,
+    TargetKind,
 };
 use domain::DomainError;
 use uuid::Uuid;
@@ -14,10 +17,30 @@ use crate::jobs::{JobHandler, JobLog};
 use crate::SettingsService;
 
 pub struct VulnerabilityService {
-    pub(crate) scanner: Arc<dyn VulnerabilityScanner>,
-    pub(crate) findings: Arc<dyn FindingRepository>,
-    pub(crate) cluster: Arc<dyn ClusterGateway>,
-    pub(crate) settings: Arc<SettingsService>,
+    scanner: Arc<dyn VulnerabilityScanner>,
+    findings: Arc<dyn FindingRepository>,
+    cluster: Arc<dyn ClusterGateway>,
+    settings: Arc<SettingsService>,
+}
+
+/// Key of the notification threshold setting.
+pub const KEY_NOTIFY_MIN_SEVERITY: &str = "vuln.notify_min_severity";
+pub const DEFAULT_NOTIFY_MIN_SEVERITY: Severity = Severity::High;
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize)]
+pub struct Summary {
+    pub total: SeverityCounts,
+    pub os: SeverityCounts,
+    pub images: SeverityCounts,
+    pub last_scan: Option<DateTime<Utc>>,
+}
+
+struct ChannelSink(tokio::sync::mpsc::UnboundedSender<String>);
+
+impl LineSink for ChannelSink {
+    fn line(&self, text: &str) {
+        let _ = self.0.send(text.to_string());
+    }
 }
 
 impl VulnerabilityService {
@@ -35,50 +58,242 @@ impl VulnerabilityService {
         }
     }
 
-    /// Scan the OS and all cluster images, persist the diff, notify about new findings.
-    pub async fn scan(&self, _log: &dyn JobLog) -> Result<ScanReport, DomainError> {
-        todo!()
+    pub async fn scanner_version(&self) -> Result<String, DomainError> {
+        self.scanner.version().await
     }
 
-    pub async fn list(&self, _filter: FindingFilter) -> Result<Vec<Finding>, DomainError> {
-        todo!()
+    /// Scan the OS and all cluster images, persist the diff, notify about new findings.
+    pub async fn scan(&self, log: &dyn JobLog) -> Result<ScanReport, DomainError> {
+        let mut targets: Vec<(TargetKind, String)> = vec![(TargetKind::Os, "os".into())];
+        match self.cluster.overview().await {
+            Ok(o) => targets.extend(o.images().into_iter().map(|i| (TargetKind::Image, i))),
+            Err(e) => {
+                log.line(&format!("cluster unavailable, scanning OS only: {e}"))
+                    .await
+            }
+        }
+        let mut report = ScanReport {
+            targets: targets.iter().map(|(_, t)| t.clone()).collect(),
+            ..Default::default()
+        };
+        let now = Utc::now();
+        for (kind, target) in &targets {
+            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+            let sink = ChannelSink(tx);
+            let scan = async {
+                let r = match kind {
+                    TargetKind::Os => self.scanner.scan_os(&sink).await,
+                    TargetKind::Image => self.scanner.scan_image(target, &sink).await,
+                };
+                drop(sink);
+                r
+            };
+            let drain = async {
+                while let Some(l) = rx.recv().await {
+                    log.line(&l).await;
+                }
+            };
+            let (result, _) = tokio::join!(scan, drain);
+            match result {
+                Ok(raw) => {
+                    let (new, fixed) = self.apply(*kind, target, raw, now).await?;
+                    log.line(&format!("{target}: {} new, {fixed} fixed", new.len()))
+                        .await;
+                    report.new_findings.extend(new);
+                    report.fixed += fixed;
+                }
+                Err(e) => {
+                    log.line(&format!(
+                        "{target}: scan failed, keeping previous findings: {e}"
+                    ))
+                    .await;
+                    report.failed_targets.push(target.clone());
+                }
+            }
+        }
+        report.total_open = self.findings.counts(None).await?.total();
+        self.notify(&report, log).await;
+        Ok(report)
+    }
+
+    /// Diff scanner output against the stored active findings of one target.
+    async fn apply(
+        &self,
+        kind: TargetKind,
+        target: &str,
+        raw: Vec<RawFinding>,
+        now: DateTime<Utc>,
+    ) -> Result<(Vec<Finding>, usize), DomainError> {
+        let existing = self.findings.active_by_target(target).await?;
+        let mut seen: HashSet<String> = HashSet::new();
+        let mut still_present = Vec::new();
+        let mut new = Vec::new();
+        for r in raw {
+            if !seen.insert(r.key()) {
+                continue;
+            }
+            match existing.iter().find(|f| f.raw.key() == r.key()) {
+                Some(f) => still_present.push(f.id),
+                None => {
+                    let f = Finding {
+                        id: Uuid::new_v4(),
+                        target_kind: kind,
+                        target: target.into(),
+                        raw: r,
+                        status: FindingStatus::Open,
+                        first_seen: now,
+                        last_seen: now,
+                    };
+                    self.findings.insert(&f).await?;
+                    new.push(f);
+                }
+            }
+        }
+        self.findings.touch(&still_present, now).await?;
+        let mut fixed = 0;
+        for f in existing.iter().filter(|f| !still_present.contains(&f.id)) {
+            self.findings.set_status(f.id, FindingStatus::Fixed).await?;
+            fixed += 1;
+        }
+        Ok((new, fixed))
+    }
+
+    async fn notify(&self, report: &ScanReport, log: &dyn JobLog) {
+        let min = match self.settings.notify_min_severity().await {
+            Ok(m) => m,
+            Err(e) => {
+                log.line(&format!("notification settings unavailable: {e}"))
+                    .await;
+                return;
+            }
+        };
+        let relevant: Vec<&Finding> = report
+            .new_findings
+            .iter()
+            .filter(|f| f.raw.severity >= min)
+            .collect();
+        if relevant.is_empty() {
+            return;
+        }
+        let plural = if relevant.len() == 1 { "y" } else { "ies" };
+        let subject = format!(
+            "[SoftVisor Monitoring] {} new vulnerabilit{plural} (>= {})",
+            relevant.len(),
+            min.as_str()
+        );
+        let mut body = format!(
+            "The vulnerability scan found {} new finding(s) at or above severity {}:\n\n",
+            relevant.len(),
+            min.as_str()
+        );
+        for f in &relevant {
+            let fixed = f
+                .raw
+                .fixed_version
+                .as_ref()
+                .map(|v| format!(" (fixed in {v})"))
+                .unwrap_or_default();
+            body.push_str(&format!(
+                "- {} [{}] {} {} in {}{fixed}\n  {}\n",
+                f.raw.cve_id,
+                f.raw.severity.as_str(),
+                f.raw.package,
+                f.raw.installed_version,
+                f.target,
+                f.raw.url
+            ));
+        }
+        body.push_str(&format!("\nTotal open findings: {}\n", report.total_open));
+        match self.settings.send_mail(None, &subject, &body).await {
+            Ok(()) => {
+                log.line(&format!(
+                    "notification mail sent for {} finding(s)",
+                    relevant.len()
+                ))
+                .await
+            }
+            Err(DomainError::Validation(_)) => log.line("no mail sent: smtp not configured").await,
+            Err(e) => log.line(&format!("notification mail failed: {e}")).await,
+        }
+    }
+
+    pub async fn list(&self, filter: FindingFilter) -> Result<Vec<Finding>, DomainError> {
+        self.findings.list(&filter).await
+    }
+
+    pub async fn targets(&self) -> Result<Vec<String>, DomainError> {
+        let mut t: Vec<String> = self
+            .findings
+            .list(&FindingFilter::default())
+            .await?
+            .into_iter()
+            .map(|f| f.target)
+            .collect();
+        t.sort();
+        t.dedup();
+        Ok(t)
     }
 
     pub async fn summary(&self) -> Result<Summary, DomainError> {
-        todo!()
+        let all = self
+            .findings
+            .list(&FindingFilter {
+                include_fixed: true,
+                ..Default::default()
+            })
+            .await?;
+        Ok(Summary {
+            total: self.findings.counts(None).await?,
+            os: self.findings.counts(Some(TargetKind::Os)).await?,
+            images: self.findings.counts(Some(TargetKind::Image)).await?,
+            last_scan: all.iter().map(|f| f.last_seen).max(),
+        })
     }
 
     /// Only open <-> acknowledged transitions are allowed by users.
     pub async fn set_status(
         &self,
-        _id: Uuid,
-        _status: FindingStatus,
+        id: Uuid,
+        status: FindingStatus,
     ) -> Result<Finding, DomainError> {
-        todo!()
+        if status == FindingStatus::Fixed {
+            return Err(DomainError::Validation(
+                "findings are marked fixed by the scanner only".into(),
+            ));
+        }
+        let f = self.findings.get(id).await?.ok_or(DomainError::NotFound)?;
+        if f.status == FindingStatus::Fixed {
+            return Err(DomainError::Validation("finding is already fixed".into()));
+        }
+        self.findings.set_status(id, status).await?;
+        Ok(Finding { status, ..f })
     }
-
-    pub async fn scanner_version(&self) -> Result<String, DomainError> {
-        self.scanner.version().await
-    }
-}
-
-#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize)]
-pub struct Summary {
-    pub total: SeverityCounts,
-    pub os: SeverityCounts,
-    pub images: SeverityCounts,
-    pub last_scan: Option<chrono::DateTime<chrono::Utc>>,
 }
 
 pub struct VulnerabilityScanJob(pub Arc<VulnerabilityService>);
 
 #[async_trait]
 impl JobHandler for VulnerabilityScanJob {
-    async fn run(&self, _params: Option<String>, _log: &dyn JobLog) -> Result<(), String> {
-        todo!()
+    async fn run(&self, _params: Option<String>, log: &dyn JobLog) -> Result<(), String> {
+        let version = self
+            .0
+            .scanner_version()
+            .await
+            .map_err(|e| format!("scanner not available: {e}"))?;
+        log.line(&format!("scanner version {version}")).await;
+        let r = self.0.scan(log).await.map_err(|e| e.to_string())?;
+        log.line(&format!(
+            "scanned {} target(s), {} failed: {} new, {} fixed, {} open in total",
+            r.targets.len(),
+            r.failed_targets.len(),
+            r.new_findings.len(),
+            r.fixed,
+            r.total_open
+        ))
+        .await;
+        if !r.targets.is_empty() && r.failed_targets.len() == r.targets.len() {
+            return Err("all targets failed to scan".into());
+        }
+        Ok(())
     }
 }
-
-/// Key of the notification threshold setting.
-pub const KEY_NOTIFY_MIN_SEVERITY: &str = "vuln.notify_min_severity";
-pub const DEFAULT_NOTIFY_MIN_SEVERITY: Severity = Severity::High;