Link image findings to their workloads and suggest a fixing image update

Container findings now name the workloads that run the image and link
into the Kubernetes view, which highlights them. An update check asks the
registry for newer tags of the same variant, scans the newest one and
records which of the open findings are gone in it. The image row then
shows the candidate tag, how many findings it fixes and how many remain,
marks those CVEs in the expanded list, and offers to roll every workload
over to it. The check runs as a job, nightly for all running images or on
demand for one.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/crates/application/src/image_update_service.rs

@@ -0,0 +1,167 @@
+//! Suggests a newer tag for a running image and verifies that it fixes findings.
+use std::collections::HashSet;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use chrono::Utc;
+use domain::image::{newest_tag, ImageRef, ImageUpdate};
+use domain::ports::{
+    ClusterGateway, FindingRepository, ImageRegistry, ImageUpdateRepository, LineSink,
+    VulnerabilityScanner,
+};
+use domain::vuln::{FindingStatus, SeverityCounts};
+use domain::DomainError;
+
+use crate::jobs::{JobHandler, JobLog};
+
+pub struct ImageUpdateService {
+    registry: Arc<dyn ImageRegistry>,
+    scanner: Arc<dyn VulnerabilityScanner>,
+    findings: Arc<dyn FindingRepository>,
+    updates: Arc<dyn ImageUpdateRepository>,
+    cluster: Arc<dyn ClusterGateway>,
+}
+
+struct ChannelSink(tokio::sync::mpsc::UnboundedSender<String>);
+
+impl LineSink for ChannelSink {
+    fn line(&self, text: &str) {
+        let _ = self.0.send(text.to_string());
+    }
+}
+
+impl ImageUpdateService {
+    pub fn new(
+        registry: Arc<dyn ImageRegistry>,
+        scanner: Arc<dyn VulnerabilityScanner>,
+        findings: Arc<dyn FindingRepository>,
+        updates: Arc<dyn ImageUpdateRepository>,
+        cluster: Arc<dyn ClusterGateway>,
+    ) -> Self {
+        Self {
+            registry,
+            scanner,
+            findings,
+            updates,
+            cluster,
+        }
+    }
+
+    /// The newest tag of the image's repository that is newer than the running one.
+    pub async fn candidate(&self, image: &str) -> Result<Option<String>, DomainError> {
+        let parsed = ImageRef::parse(image)
+            .ok_or_else(|| DomainError::Validation(format!("cannot parse image '{image}'")))?;
+        let tags = self.registry.tags(image).await?;
+        Ok(newest_tag(&parsed.tag, &tags).map(|t| parsed.with_tag(&t)))
+    }
+
+    pub async fn stored(&self, image: &str) -> Result<Option<ImageUpdate>, DomainError> {
+        self.updates.get(image).await
+    }
+
+    pub async fn all(&self) -> Result<Vec<ImageUpdate>, DomainError> {
+        self.updates.list().await
+    }
+
+    /// Scan the newest available tag and record which of the open findings it fixes.
+    pub async fn check(
+        &self,
+        image: &str,
+        log: &dyn JobLog,
+    ) -> Result<Option<ImageUpdate>, DomainError> {
+        let Some(candidate) = self.candidate(image).await? else {
+            log.line(&format!("{image}: already on the newest tag"))
+                .await;
+            return Ok(None);
+        };
+        log.line(&format!("{image}: candidate {candidate}, scanning it"))
+            .await;
+
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<String>();
+        let sink = ChannelSink(tx);
+        let scan = async {
+            let r = self.scanner.scan_image(&candidate, &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);
+        let candidate_findings = result?;
+
+        let still_there: HashSet<String> = candidate_findings.iter().map(|f| f.key()).collect();
+        let current = self.findings.active_by_target(image).await?;
+        let mut fixed_counts = SeverityCounts::default();
+        let mut fixed: Vec<String> = Vec::new();
+        for f in current
+            .iter()
+            .filter(|f| f.status != FindingStatus::Fixed && !still_there.contains(&f.raw.key()))
+        {
+            fixed_counts.add(f.raw.severity);
+            fixed.push(f.raw.cve_id.clone());
+        }
+        fixed.sort();
+        fixed.dedup();
+
+        let update = ImageUpdate {
+            image: image.to_string(),
+            candidate,
+            checked_at: Utc::now(),
+            fixed_counts,
+            candidate_total: candidate_findings.len(),
+            fixed,
+        };
+        self.updates.upsert(&update).await?;
+        log.line(&format!(
+            "{image} → {}: fixes {} finding(s) ({} critical, {} high), {} remain",
+            update.candidate,
+            update.fixed.len(),
+            update.fixed_counts.critical,
+            update.fixed_counts.high,
+            update.candidate_total
+        ))
+        .await;
+        Ok(Some(update))
+    }
+
+    /// Check every image that currently runs on the cluster.
+    pub async fn check_running(&self, log: &dyn JobLog) -> Result<usize, DomainError> {
+        let images = self.cluster.overview().await?.images();
+        log.line(&format!(
+            "checking {} running image(s) for newer tags",
+            images.len()
+        ))
+        .await;
+        let mut found = 0;
+        for image in images {
+            match self.check(&image, log).await {
+                Ok(Some(_)) => found += 1,
+                Ok(None) => {}
+                Err(e) => log.line(&format!("{image}: check failed: {e}")).await,
+            }
+        }
+        Ok(found)
+    }
+}
+
+/// Job handler for `JobKind::ImageUpdateCheck`; params = one image, or empty for all running ones.
+pub struct ImageUpdateCheckJob(pub Arc<ImageUpdateService>);
+
+#[async_trait]
+impl JobHandler for ImageUpdateCheckJob {
+    async fn run(&self, params: Option<String>, log: &dyn JobLog) -> Result<(), String> {
+        match params.as_deref().map(str::trim).filter(|p| !p.is_empty()) {
+            Some(image) => {
+                self.0.check(image, log).await.map_err(|e| e.to_string())?;
+            }
+            None => {
+                let n = self.0.check_running(log).await.map_err(|e| e.to_string())?;
+                log.line(&format!("{n} image(s) have a newer tag")).await;
+            }
+        }
+        Ok(())
+    }
+}

backend/crates/application/src/lib.rs

@@ -2,6 +2,7 @@
 pub mod auth_service;
 pub mod backup_service;
 pub mod cluster_service;
+pub mod image_update_service;
 pub mod inventory_service;
 pub mod jobs;
 pub mod scheduler;
@@ -13,6 +14,7 @@ pub mod vuln_service;
 pub use auth_service::AuthService;
 pub use backup_service::{BackupDeps, BackupJob, BackupService};
 pub use cluster_service::ClusterService;
+pub use image_update_service::{ImageUpdateCheckJob, ImageUpdateService};
 pub use inventory_service::{InventoryService, PackageRefreshJob};
 pub use jobs::{JobHandler, JobLog, JobRunner};
 pub use settings_service::SettingsService;

backend/crates/application/src/test_fakes.rs

@@ -418,6 +418,7 @@ use domain::ports::ClusterGateway;
 #[derive(Default)]
 pub struct MemCluster {
     pub actions: Mutex<Vec<String>>,
+    pub fail: std::sync::atomic::AtomicBool,
 }
 
 pub fn sample_overview() -> ClusterOverview {
@@ -469,6 +470,9 @@ pub fn sample_overview() -> ClusterOverview {
 #[async_trait]
 impl ClusterGateway for MemCluster {
     async fn overview(&self) -> Result<ClusterOverview, DomainError> {
+        if self.fail.load(std::sync::atomic::Ordering::SeqCst) {
+            return Err(DomainError::Unavailable("cluster unreachable".into()));
+        }
         Ok(sample_overview())
     }
     async fn restart(&self, w: &WorkloadRef) -> Result<(), DomainError> {
@@ -964,3 +968,56 @@ pub fn strategy(name: &str, target_id: Uuid) -> BackupStrategy {
         enabled: true,
     }
 }
+
+use domain::image::ImageUpdate;
+use domain::ports::{ImageRegistry, ImageUpdateRepository};
+
+#[derive(Default)]
+pub struct MemRegistry {
+    pub tags: Mutex<Vec<String>>,
+    pub fail: std::sync::atomic::AtomicBool,
+}
+
+#[async_trait]
+impl ImageRegistry for MemRegistry {
+    async fn tags(&self, _image: &str) -> Result<Vec<String>, DomainError> {
+        if self.fail.load(std::sync::atomic::Ordering::SeqCst) {
+            return Err(DomainError::Unavailable("registry unreachable".into()));
+        }
+        Ok(self.tags.lock().unwrap().clone())
+    }
+}
+
+#[derive(Default)]
+pub struct MemImageUpdates(pub Mutex<Vec<ImageUpdate>>);
+
+#[async_trait]
+impl ImageUpdateRepository for MemImageUpdates {
+    async fn upsert(&self, update: &ImageUpdate) -> Result<(), DomainError> {
+        let mut v = self.0.lock().unwrap();
+        v.retain(|u| u.image != update.image);
+        v.push(update.clone());
+        Ok(())
+    }
+    async fn get(&self, image: &str) -> Result<Option<ImageUpdate>, DomainError> {
+        Ok(self
+            .0
+            .lock()
+            .unwrap()
+            .iter()
+            .find(|u| u.image == image)
+            .cloned())
+    }
+    async fn list(&self) -> Result<Vec<ImageUpdate>, DomainError> {
+        Ok(self.0.lock().unwrap().clone())
+    }
+}
+
+/// Settings service on in-memory stores, for services that only need it as a dependency.
+pub fn settings_service() -> Arc<crate::SettingsService> {
+    Arc::new(crate::SettingsService::new(
+        Arc::new(MemSettings::default()),
+        Arc::new(FakeCipher),
+        Arc::new(MemMailer::default()),
+    ))
+}

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

@@ -0,0 +1,165 @@
+use std::sync::{Arc, Mutex};
+
+use async_trait::async_trait;
+use domain::vuln::Severity;
+use domain::DomainError;
+
+use crate::jobs::{JobHandler, JobLog};
+use crate::test_fakes::{raw, FakeScanner, MemCluster, MemFindings, MemImageUpdates, MemRegistry};
+use crate::{ImageUpdateCheckJob, ImageUpdateService, VulnerabilityService};
+
+#[derive(Default)]
+struct VecLog(Mutex<Vec<String>>);
+#[async_trait]
+impl JobLog for VecLog {
+    async fn line(&self, text: &str) {
+        self.0.lock().unwrap().push(text.into());
+    }
+}
+
+const GITEA: &str = "gitea/gitea:1.22.3";
+
+struct F {
+    findings: Arc<MemFindings>,
+    updates: Arc<MemImageUpdates>,
+    registry: Arc<MemRegistry>,
+}
+
+fn fixture(scanner: FakeScanner, tags: &[&str]) -> (F, Arc<ImageUpdateService>) {
+    let findings = Arc::new(MemFindings::default());
+    let updates = Arc::new(MemImageUpdates::default());
+    let registry = Arc::new(MemRegistry::default());
+    *registry.tags.lock().unwrap() = tags.iter().map(|t| t.to_string()).collect();
+    let cluster = Arc::new(MemCluster::default());
+    let svc = ImageUpdateService::new(
+        registry.clone(),
+        Arc::new(scanner),
+        findings.clone(),
+        updates.clone(),
+        cluster,
+    );
+    (
+        F {
+            findings,
+            updates,
+            registry,
+        },
+        Arc::new(svc),
+    )
+}
+
+/// Scan the running image first so there are open findings to compare against.
+async fn seed(findings: Arc<MemFindings>, scanner: FakeScanner, cluster: Arc<MemCluster>) {
+    let settings = crate::test_fakes::settings_service();
+    let vulns = VulnerabilityService::new(Arc::new(scanner), findings, cluster, settings);
+    vulns.scan(&VecLog::default()).await.unwrap();
+}
+
+#[tokio::test]
+async fn suggests_the_newest_tag_of_the_same_variant() {
+    let (_f, svc) = fixture(
+        FakeScanner::default(),
+        &["1.22.3", "1.22.4", "1.23.0", "1.24.0-rc1", "latest"],
+    );
+    assert_eq!(
+        svc.candidate(GITEA).await.unwrap().as_deref(),
+        Some("gitea/gitea:1.23.0")
+    );
+    let (_f, svc) = fixture(FakeScanner::default(), &["1.22.3"]);
+    assert_eq!(
+        svc.candidate(GITEA).await.unwrap(),
+        None,
+        "already the newest"
+    );
+}
+
+#[tokio::test]
+async fn a_check_records_which_findings_the_candidate_fixes() {
+    // the running image has three flaws, the candidate only one of them
+    let running = FakeScanner::default().with(
+        GITEA,
+        Ok(vec![
+            raw("CVE-1", "git", "2.39", Severity::Critical, Some("2.40")),
+            raw("CVE-2", "curl", "7.8", Severity::High, None),
+            raw("CVE-3", "zlib", "1.2", Severity::Low, None),
+        ]),
+    );
+    let candidate = FakeScanner::default().with(
+        "gitea/gitea:1.23.0",
+        Ok(vec![raw("CVE-3", "zlib", "1.2", Severity::Low, None)]),
+    );
+    let (f, svc) = fixture(candidate, &["1.22.3", "1.23.0"]);
+    seed(f.findings.clone(), running, Arc::new(MemCluster::default())).await;
+
+    let log = VecLog::default();
+    let update = svc.check(GITEA, &log).await.unwrap().unwrap();
+    assert_eq!(update.candidate, "gitea/gitea:1.23.0");
+    assert_eq!(update.fixed, vec!["CVE-1", "CVE-2"]);
+    assert_eq!(update.fixed_counts.critical, 1);
+    assert_eq!(update.fixed_counts.high, 1);
+    assert_eq!(
+        update.candidate_total, 1,
+        "the candidate still has one finding"
+    );
+    assert_eq!(svc.stored(GITEA).await.unwrap().as_ref(), Some(&update));
+    let lines = log.0.lock().unwrap().join("\n");
+    assert!(lines.contains("fixes 2 finding(s)"), "{lines}");
+
+    // a repeated check overwrites the previous result
+    svc.check(GITEA, &VecLog::default()).await.unwrap();
+    assert_eq!(svc.all().await.unwrap().len(), 1);
+}
+
+#[tokio::test]
+async fn nothing_is_recorded_when_the_image_is_current() {
+    let (f, svc) = fixture(FakeScanner::default(), &["1.22.3"]);
+    let log = VecLog::default();
+    assert!(svc.check(GITEA, &log).await.unwrap().is_none());
+    assert!(f.updates.0.lock().unwrap().is_empty());
+    assert!(log
+        .0
+        .lock()
+        .unwrap()
+        .iter()
+        .any(|l| l.contains("newest tag")));
+}
+
+#[tokio::test]
+async fn a_registry_that_cannot_be_reached_fails_the_check_of_that_image_only() {
+    let (f, svc) = fixture(FakeScanner::default(), &["1.22.3", "1.23.0"]);
+    f.registry
+        .fail
+        .store(true, std::sync::atomic::Ordering::SeqCst);
+    assert!(matches!(
+        svc.check(GITEA, &VecLog::default()).await.unwrap_err(),
+        DomainError::Unavailable(_)
+    ));
+
+    // checking every running image keeps going and reports the failures in the log
+    let log = VecLog::default();
+    let found = svc.check_running(&log).await.unwrap();
+    assert_eq!(found, 0);
+    let lines = log.0.lock().unwrap().join("\n");
+    assert!(lines.contains("check failed"), "{lines}");
+}
+
+#[tokio::test]
+async fn the_job_checks_one_image_or_all_running_ones() {
+    let candidate = FakeScanner::default().with("gitea/gitea:1.23.0", Ok(vec![]));
+    let (f, svc) = fixture(candidate, &["1.22.3", "1.23.0", "16.4.0", "17.0.0"]);
+    let job = ImageUpdateCheckJob(svc.clone());
+    job.run(Some(GITEA.into()), &VecLog::default())
+        .await
+        .unwrap();
+    assert_eq!(f.updates.0.lock().unwrap().len(), 1);
+
+    let log = VecLog::default();
+    job.run(None, &log).await.unwrap();
+    assert!(log
+        .0
+        .lock()
+        .unwrap()
+        .iter()
+        .any(|l| l.contains("running image(s)")));
+    assert!(!f.updates.0.lock().unwrap().is_empty());
+}

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

@@ -1,6 +1,7 @@
 mod auth_service_tests;
 mod backup_tests;
 mod cluster_tests;
+mod image_update_tests;
 mod inventory_tests;
 mod jobs_tests;
 mod scheduler_tests;

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

@@ -25,6 +25,7 @@ struct F {
     mailer: Arc<MemMailer>,
     settings: Arc<SettingsService>,
     results: ScanResults,
+    cluster: Arc<MemCluster>,
 }
 
 fn fixture(scanner: FakeScanner) -> (F, VulnerabilityService) {
@@ -36,10 +37,11 @@ fn fixture(scanner: FakeScanner) -> (F, VulnerabilityService) {
         Arc::new(FakeCipher),
         mailer.clone(),
     ));
+    let cluster = Arc::new(MemCluster::default());
     let svc = VulnerabilityService::new(
         Arc::new(scanner),
         findings.clone(),
-        Arc::new(MemCluster::default()),
+        cluster.clone(),
         settings.clone(),
     );
     (
@@ -48,6 +50,7 @@ fn fixture(scanner: FakeScanner) -> (F, VulnerabilityService) {
             mailer,
             settings,
             results,
+            cluster,
         },
         svc,
     )
@@ -533,3 +536,66 @@ async fn findings_of_one_group_can_be_listed() {
         .unwrap();
     assert_eq!(image.len(), 1);
 }
+
+#[tokio::test]
+async fn container_groups_name_the_workloads_that_run_the_image() {
+    let scanner = FakeScanner::default()
+        .with(
+            "os",
+            Ok(vec![raw("CVE-1", "openssl", "3.0.1", Severity::High, None)]),
+        )
+        .with(
+            GITEA,
+            Ok(vec![raw("CVE-2", "git", "2.39", Severity::High, None)]),
+        );
+    let (_f, svc) = fixture(scanner);
+    svc.scan(&VecLog::default()).await.unwrap();
+
+    let images = svc
+        .image_groups(FindingFilter {
+            target_kind: Some(TargetKind::Image),
+            ..Default::default()
+        })
+        .await
+        .unwrap();
+    let gitea = images.iter().find(|g| g.group.key == GITEA).unwrap();
+    assert_eq!(gitea.workloads.len(), 1);
+    assert_eq!(gitea.workloads[0].namespace, "gitea");
+    assert_eq!(gitea.workloads[0].name, "gitea");
+    assert_eq!(
+        gitea.workloads[0].kind,
+        domain::cluster::WorkloadKind::Deployment
+    );
+    assert!(gitea.running, "the image is in use on the cluster");
+
+    // an image that no workload runs any more is reported as not running
+    let stale = images.iter().find(|g| g.group.key == PG);
+    assert!(stale.is_none_or(|g| g.workloads.is_empty()));
+}
+
+#[tokio::test]
+async fn image_groups_survive_an_unreachable_cluster() {
+    let scanner = FakeScanner::default().with(
+        GITEA,
+        Ok(vec![raw("CVE-2", "git", "2.39", Severity::High, None)]),
+    );
+    let (f, svc) = fixture(scanner);
+    svc.scan(&VecLog::default()).await.unwrap();
+    f.cluster
+        .fail
+        .store(true, std::sync::atomic::Ordering::SeqCst);
+
+    let images = svc
+        .image_groups(FindingFilter {
+            target_kind: Some(TargetKind::Image),
+            ..Default::default()
+        })
+        .await
+        .unwrap();
+    assert_eq!(images.len(), 1, "findings are still listed");
+    assert!(images[0].workloads.is_empty());
+    assert!(
+        !images[0].running,
+        "unknown while the cluster is unreachable"
+    );
+}

backend/crates/application/src/vuln_service.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
+use domain::cluster::WorkloadRef;
 use domain::ports::{ClusterGateway, FindingRepository, LineSink, VulnerabilityScanner};
 use domain::vuln::{
     Finding, FindingFilter, FindingGroup, FindingStatus, RawFinding, ScanReport, Severity,
@@ -27,6 +28,17 @@ pub struct VulnerabilityService {
 pub const KEY_NOTIFY_MIN_SEVERITY: &str = "vuln.notify_min_severity";
 pub const DEFAULT_NOTIFY_MIN_SEVERITY: Severity = Severity::High;
 
+/// An image group with the workloads that run the image.
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
+pub struct ImageGroup {
+    #[serde(flatten)]
+    pub group: FindingGroup,
+    /// Workloads on the cluster that currently use this image.
+    pub workloads: Vec<WorkloadRef>,
+    /// False when no workload uses it, or while the cluster is unreachable.
+    pub running: bool,
+}
+
 /// One scanned target with its number of open findings.
 #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
 pub struct TargetSummary {
@@ -242,6 +254,39 @@ impl VulnerabilityService {
         self.findings.groups(&filter).await
     }
 
+    /// Image groups together with the workloads that currently run them. An unreachable
+    /// cluster is not an error: the findings are still listed, just without their usage.
+    pub async fn image_groups(
+        &self,
+        filter: FindingFilter,
+    ) -> Result<Vec<ImageGroup>, DomainError> {
+        let groups = self.findings.groups(&filter).await?;
+        let overview = self.cluster.overview().await.ok();
+        Ok(groups
+            .into_iter()
+            .map(|group| {
+                let workloads: Vec<WorkloadRef> = overview
+                    .as_ref()
+                    .map(|o| {
+                        o.workloads_running(&group.key)
+                            .into_iter()
+                            .map(|w| WorkloadRef {
+                                namespace: w.namespace.clone(),
+                                kind: w.kind,
+                                name: w.name.clone(),
+                            })
+                            .collect()
+                    })
+                    .unwrap_or_default();
+                ImageGroup {
+                    running: !workloads.is_empty(),
+                    workloads,
+                    group,
+                }
+            })
+            .collect())
+    }
+
     /// Scanned targets that still have open findings, host first.
     pub async fn targets(&self) -> Result<Vec<TargetSummary>, DomainError> {
         let mut by_target: std::collections::HashMap<(TargetKind, String), usize> =