diff --git a/compliance-agent/src/pipeline/orchestrator.rs b/compliance-agent/src/pipeline/orchestrator.rs index bb0ccca..23f5595 100644 --- a/compliance-agent/src/pipeline/orchestrator.rs +++ b/compliance-agent/src/pipeline/orchestrator.rs @@ -259,67 +259,12 @@ impl PipelineOrchestrator { .await?; } - // Persist CVE alerts and create notifications - { - use compliance_core::models::notification::{parse_severity, CveNotification}; - - let repo_name = repo.name.clone(); - let mut new_notif_count = 0u32; - - for alert in &cve_alerts { - // Upsert the alert - let filter = doc! { - "cve_id": &alert.cve_id, - "repo_id": &alert.repo_id, - }; - let update = mongodb::bson::to_document(alert) - .map(|d| doc! { "$set": d }) - .unwrap_or_else(|_| doc! {}); - self.db - .cve_alerts() - .update_one(filter, update) - .upsert(true) - .await?; - - // Create notification (dedup by cve_id + repo + package + version) - let notif_filter = doc! { - "cve_id": &alert.cve_id, - "repo_id": &alert.repo_id, - "package_name": &alert.affected_package, - "package_version": &alert.affected_version, - }; - let severity = parse_severity(alert.severity.as_deref(), alert.cvss_score); - let mut notification = CveNotification::new( - alert.cve_id.clone(), - repo_id.clone(), - repo_name.clone(), - alert.affected_package.clone(), - alert.affected_version.clone(), - severity, - ); - notification.cvss_score = alert.cvss_score; - notification.summary = alert.summary.clone(); - notification.url = Some(format!("https://osv.dev/vulnerability/{}", alert.cve_id)); - - let notif_update = doc! { - "$setOnInsert": mongodb::bson::to_bson(¬ification).unwrap_or_default() - }; - if let Ok(result) = self - .db - .cve_notifications() - .update_one(notif_filter, notif_update) - .upsert(true) - .await - { - if result.upserted_id.is_some() { - new_notif_count += 1; - } - } - } - - if new_notif_count > 0 { - tracing::info!("[{repo_id}] Created {new_notif_count} CVE notification(s)"); - } + // Persist CVE alerts and create notifications (shared with the PLC path). + let new_notif_count = self + .persist_cve_alerts(&repo_id, &repo.name, &cve_alerts) + .await?; + if new_notif_count > 0 { + tracing::info!("[{repo_id}] Created {new_notif_count} CVE notification(s)"); } // Stage 6: Issue Creation @@ -558,7 +503,10 @@ impl PipelineOrchestrator { target_id, ); if !sbom.is_empty() { - if let Err(e) = self.persist_control_app_sbom(target_id, sbom).await { + if let Err(e) = self + .persist_control_app_sbom(target_id, &target.name, sbom) + .await + { tracing::warn!(target_id, error = %e, "control-app SBOM persist failed"); } } @@ -573,6 +521,7 @@ impl PipelineOrchestrator { async fn persist_control_app_sbom( &self, target_id: &str, + target_name: &str, mut entries: Vec, ) -> Result<(), AgentError> { if entries.is_empty() { @@ -622,25 +571,80 @@ impl PipelineOrchestrator { .await?; } } - for alert in &alerts { - let filter = doc! { "cve_id": &alert.cve_id, "repo_id": &alert.repo_id }; - if let Ok(d) = mongodb::bson::to_document(alert) { - self.db - .cve_alerts() - .update_one(filter, doc! { "$set": d }) - .upsert(true) - .await?; - } - } + let new_notifs = self + .persist_cve_alerts(target_id, target_name, &alerts) + .await?; tracing::info!( target_id, components = entries.len(), alerts = alerts.len(), + notifications = new_notifs, "control-app SBOM stored" ); Ok(()) } + /// Upsert CVE alerts for a target and create dedup'd CVE notifications; + /// returns the number of newly-created notifications. Shared by the SAST + /// pipeline and the PLC control-app SBOM path, so every SBOM source (source, + /// firmware, CODESYS libraries/runtime) raises the same notifications. + async fn persist_cve_alerts( + &self, + repo_id: &str, + repo_name: &str, + alerts: &[CveAlert], + ) -> Result { + use compliance_core::models::notification::{parse_severity, CveNotification}; + + let mut new_notif = 0u32; + for alert in alerts { + let filter = doc! { "cve_id": &alert.cve_id, "repo_id": &alert.repo_id }; + let update = mongodb::bson::to_document(alert) + .map(|d| doc! { "$set": d }) + .unwrap_or_else(|_| doc! {}); + self.db + .cve_alerts() + .update_one(filter, update) + .upsert(true) + .await?; + + // Dedup notifications by cve + repo + package + version. + let notif_filter = doc! { + "cve_id": &alert.cve_id, + "repo_id": &alert.repo_id, + "package_name": &alert.affected_package, + "package_version": &alert.affected_version, + }; + let severity = parse_severity(alert.severity.as_deref(), alert.cvss_score); + let mut notification = CveNotification::new( + alert.cve_id.clone(), + repo_id.to_string(), + repo_name.to_string(), + alert.affected_package.clone(), + alert.affected_version.clone(), + severity, + ); + notification.cvss_score = alert.cvss_score; + notification.summary = alert.summary.clone(); + notification.url = Some(format!("https://osv.dev/vulnerability/{}", alert.cve_id)); + let notif_update = doc! { + "$setOnInsert": mongodb::bson::to_bson(¬ification).unwrap_or_default() + }; + if let Ok(result) = self + .db + .cve_notifications() + .update_one(notif_filter, notif_update) + .upsert(true) + .await + { + if result.upserted_id.is_some() { + new_notif += 1; + } + } + } + Ok(new_notif) + } + /// Ingest the target's artifacts, classify (tramiton for firmware/RTOS/Yocto, /// heuristics otherwise), and store the detected classification on the target. /// Best-effort — never fails the scan.