feat: hourly CVE alerting with notification bell and API (#53)

compliance-agent/src/api/handlers/mod.rs

@@ -6,6 +6,7 @@ pub mod graph;
 pub mod health;
 pub mod help_chat;
 pub mod issues;
+pub mod notifications;
 pub mod pentest_handlers;
 pub use pentest_handlers as pentest;
 pub mod repos;

compliance-agent/src/api/handlers/notifications.rs

@@ -0,0 +1,178 @@
+use axum::extract::Extension;
+use axum::http::StatusCode;
+use axum::Json;
+use mongodb::bson::doc;
+use serde::Deserialize;
+
+use compliance_core::models::notification::CveNotification;
+
+use super::dto::{AgentExt, ApiResponse};
+
+/// GET /api/v1/notifications — List CVE notifications (newest first)
+#[tracing::instrument(skip_all)]
+pub async fn list_notifications(
+    Extension(agent): AgentExt,
+    axum::extract::Query(params): axum::extract::Query<NotificationFilter>,
+) -> Result<Json<ApiResponse<Vec<CveNotification>>>, StatusCode> {
+    let mut filter = doc! {};
+
+    // Filter by status (default: show new + read, exclude dismissed)
+    match params.status.as_deref() {
+        Some("all") => {}
+        Some(s) => {
+            filter.insert("status", s);
+        }
+        None => {
+            filter.insert("status", doc! { "$in": ["new", "read"] });
+        }
+    }
+
+    // Filter by severity
+    if let Some(ref sev) = params.severity {
+        filter.insert("severity", sev.as_str());
+    }
+
+    // Filter by repo
+    if let Some(ref repo_id) = params.repo_id {
+        filter.insert("repo_id", repo_id.as_str());
+    }
+
+    let page = params.page.unwrap_or(1).max(1);
+    let limit = params.limit.unwrap_or(50).min(200);
+    let skip = (page - 1) * limit as u64;
+
+    let total = agent
+        .db
+        .cve_notifications()
+        .count_documents(filter.clone())
+        .await
+        .unwrap_or(0);
+
+    let notifications: Vec<CveNotification> = match agent
+        .db
+        .cve_notifications()
+        .find(filter)
+        .sort(doc! { "created_at": -1 })
+        .skip(skip)
+        .limit(limit)
+        .await
+    {
+        Ok(cursor) => {
+            use futures_util::StreamExt;
+            let mut items = Vec::new();
+            let mut cursor = cursor;
+            while let Some(Ok(n)) = cursor.next().await {
+                items.push(n);
+            }
+            items
+        }
+        Err(e) => {
+            tracing::error!("Failed to list notifications: {e}");
+            return Err(StatusCode::INTERNAL_SERVER_ERROR);
+        }
+    };
+
+    Ok(Json(ApiResponse {
+        data: notifications,
+        total: Some(total),
+        page: Some(page),
+    }))
+}
+
+/// GET /api/v1/notifications/count — Count of unread notifications
+#[tracing::instrument(skip_all)]
+pub async fn notification_count(
+    Extension(agent): AgentExt,
+) -> Result<Json<serde_json::Value>, StatusCode> {
+    let count = agent
+        .db
+        .cve_notifications()
+        .count_documents(doc! { "status": "new" })
+        .await
+        .unwrap_or(0);
+
+    Ok(Json(serde_json::json!({ "count": count })))
+}
+
+/// PATCH /api/v1/notifications/:id/read — Mark a notification as read
+#[tracing::instrument(skip_all, fields(id = %id))]
+pub async fn mark_read(
+    Extension(agent): AgentExt,
+    axum::extract::Path(id): axum::extract::Path<String>,
+) -> Result<Json<serde_json::Value>, StatusCode> {
+    let oid = mongodb::bson::oid::ObjectId::parse_str(&id).map_err(|_| StatusCode::BAD_REQUEST)?;
+
+    let result = agent
+        .db
+        .cve_notifications()
+        .update_one(
+            doc! { "_id": oid },
+            doc! { "$set": {
+                "status": "read",
+                "read_at": mongodb::bson::DateTime::now(),
+            }},
+        )
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    if result.matched_count == 0 {
+        return Err(StatusCode::NOT_FOUND);
+    }
+    Ok(Json(serde_json::json!({ "status": "read" })))
+}
+
+/// PATCH /api/v1/notifications/:id/dismiss — Dismiss a notification
+#[tracing::instrument(skip_all, fields(id = %id))]
+pub async fn dismiss_notification(
+    Extension(agent): AgentExt,
+    axum::extract::Path(id): axum::extract::Path<String>,
+) -> Result<Json<serde_json::Value>, StatusCode> {
+    let oid = mongodb::bson::oid::ObjectId::parse_str(&id).map_err(|_| StatusCode::BAD_REQUEST)?;
+
+    let result = agent
+        .db
+        .cve_notifications()
+        .update_one(
+            doc! { "_id": oid },
+            doc! { "$set": { "status": "dismissed" } },
+        )
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    if result.matched_count == 0 {
+        return Err(StatusCode::NOT_FOUND);
+    }
+    Ok(Json(serde_json::json!({ "status": "dismissed" })))
+}
+
+/// POST /api/v1/notifications/read-all — Mark all new notifications as read
+#[tracing::instrument(skip_all)]
+pub async fn mark_all_read(
+    Extension(agent): AgentExt,
+) -> Result<Json<serde_json::Value>, StatusCode> {
+    let result = agent
+        .db
+        .cve_notifications()
+        .update_many(
+            doc! { "status": "new" },
+            doc! { "$set": {
+                "status": "read",
+                "read_at": mongodb::bson::DateTime::now(),
+            }},
+        )
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    Ok(Json(
+        serde_json::json!({ "updated": result.modified_count }),
+    ))
+}
+
+#[derive(Debug, Deserialize)]
+pub struct NotificationFilter {
+    pub status: Option<String>,
+    pub severity: Option<String>,
+    pub repo_id: Option<String>,
+    pub page: Option<u64>,
+    pub limit: Option<i64>,
+}

compliance-agent/src/api/routes.rs

@@ -101,6 +101,27 @@ pub fn build_router() -> Router {
         )
         // Help chat (documentation-grounded Q&A)
         .route("/api/v1/help/chat", post(handlers::help_chat::help_chat))
+        // CVE notification endpoints
+        .route(
+            "/api/v1/notifications",
+            get(handlers::notifications::list_notifications),
+        )
+        .route(
+            "/api/v1/notifications/count",
+            get(handlers::notifications::notification_count),
+        )
+        .route(
+            "/api/v1/notifications/read-all",
+            post(handlers::notifications::mark_all_read),
+        )
+        .route(
+            "/api/v1/notifications/{id}/read",
+            patch(handlers::notifications::mark_read),
+        )
+        .route(
+            "/api/v1/notifications/{id}/dismiss",
+            patch(handlers::notifications::dismiss_notification),
+        )
         // Pentest API endpoints
         .route(
             "/api/v1/pentest/lookup-repo",

compliance-agent/src/config.rs

@@ -42,7 +42,7 @@ pub fn load_config() -> Result<AgentConfig, AgentError> {
             .unwrap_or(3001),
         scan_schedule: env_var_opt("SCAN_SCHEDULE").unwrap_or_else(|| "0 0 */6 * * *".to_string()),
         cve_monitor_schedule: env_var_opt("CVE_MONITOR_SCHEDULE")
-            .unwrap_or_else(|| "0 0 0 * * *".to_string()),
+            .unwrap_or_else(|| "0 0 * * * *".to_string()),
         git_clone_base_path: env_var_opt("GIT_CLONE_BASE_PATH")
             .unwrap_or_else(|| "/tmp/compliance-scanner/repos".to_string()),
         ssh_key_path: env_var_opt("SSH_KEY_PATH")

compliance-agent/src/database.rs

@@ -78,6 +78,25 @@ impl Database {
             )
             .await?;
 
+        // cve_notifications: unique cve_id + repo_id + package, status filter
+        self.cve_notifications()
+            .create_index(
+                IndexModel::builder()
+                    .keys(
+                        doc! { "cve_id": 1, "repo_id": 1, "package_name": 1, "package_version": 1 },
+                    )
+                    .options(IndexOptions::builder().unique(true).build())
+                    .build(),
+            )
+            .await?;
+        self.cve_notifications()
+            .create_index(
+                IndexModel::builder()
+                    .keys(doc! { "status": 1, "created_at": -1 })
+                    .build(),
+            )
+            .await?;
+
         // tracker_issues: unique finding_id
         self.tracker_issues()
             .create_index(
@@ -222,6 +241,12 @@ impl Database {
         self.inner.collection("cve_alerts")
     }
 
+    pub fn cve_notifications(
+        &self,
+    ) -> Collection<compliance_core::models::notification::CveNotification> {
+        self.inner.collection("cve_notifications")
+    }
+
     pub fn tracker_issues(&self) -> Collection<TrackerIssue> {
         self.inner.collection("tracker_issues")
     }

compliance-agent/src/scheduler.rs

@@ -82,24 +82,158 @@ async fn scan_all_repos(agent: &ComplianceAgent) {
 }
 
 async fn monitor_cves(agent: &ComplianceAgent) {
+    use compliance_core::models::notification::{parse_severity, CveNotification};
+    use compliance_core::models::SbomEntry;
     use futures_util::StreamExt;
 
-    // Re-scan all SBOM entries for new CVEs
+    // Fetch all SBOM entries grouped by repo
     let cursor = match agent.db.sbom_entries().find(doc! {}).await {
         Ok(c) => c,
         Err(e) => {
-            tracing::error!("Failed to list SBOM entries for CVE monitoring: {e}");
+            tracing::error!("CVE monitor: failed to list SBOM entries: {e}");
             return;
         }
     };
-
-    let entries: Vec<_> = cursor.filter_map(|r| async { r.ok() }).collect().await;
-
+    let entries: Vec<SbomEntry> = cursor.filter_map(|r| async { r.ok() }).collect().await;
     if entries.is_empty() {
+        tracing::debug!("CVE monitor: no SBOM entries, skipping");
         return;
     }
 
-    tracing::info!("CVE monitor: checking {} dependencies", entries.len());
-    // The actual CVE checking is handled by the CveScanner in the pipeline
-    // This is a simplified version that just logs the activity
+    tracing::info!(
+        "CVE monitor: checking {} dependencies for new CVEs",
+        entries.len()
+    );
+
+    // Build a repo_id → repo_name lookup
+    let repo_ids: std::collections::HashSet<String> =
+        entries.iter().map(|e| e.repo_id.clone()).collect();
+    let mut repo_names: std::collections::HashMap<String, String> =
+        std::collections::HashMap::new();
+    for rid in &repo_ids {
+        if let Ok(oid) = mongodb::bson::oid::ObjectId::parse_str(rid) {
+            if let Ok(Some(repo)) = agent.db.repositories().find_one(doc! { "_id": oid }).await {
+                repo_names.insert(rid.clone(), repo.name.clone());
+            }
+        }
+    }
+
+    // Use the existing CveScanner to query OSV.dev
+    let nvd_key = agent.config.nvd_api_key.as_ref().map(|k| {
+        use secrecy::ExposeSecret;
+        k.expose_secret().to_string()
+    });
+    let scanner = crate::pipeline::cve::CveScanner::new(
+        agent.http.clone(),
+        agent.config.searxng_url.clone(),
+        nvd_key,
+    );
+
+    // Group entries by repo for scanning
+    let mut entries_by_repo: std::collections::HashMap<String, Vec<SbomEntry>> =
+        std::collections::HashMap::new();
+    for entry in entries {
+        entries_by_repo
+            .entry(entry.repo_id.clone())
+            .or_default()
+            .push(entry);
+    }
+
+    let mut new_notifications = 0u32;
+
+    for (repo_id, mut repo_entries) in entries_by_repo {
+        let repo_name = repo_names
+            .get(&repo_id)
+            .cloned()
+            .unwrap_or_else(|| repo_id.clone());
+
+        // Scan dependencies for CVEs
+        let alerts = match scanner.scan_dependencies(&repo_id, &mut repo_entries).await {
+            Ok(a) => a,
+            Err(e) => {
+                tracing::warn!("CVE monitor: scan failed for {repo_name}: {e}");
+                continue;
+            }
+        };
+
+        // Upsert CVE alerts (existing logic)
+        for alert in &alerts {
+            let filter = doc! { "cve_id": &alert.cve_id, "repo_id": &alert.repo_id };
+            let update = doc! { "$setOnInsert": mongodb::bson::to_bson(alert).unwrap_or_default() };
+            let _ = agent
+                .db
+                .cve_alerts()
+                .update_one(filter, update)
+                .upsert(true)
+                .await;
+        }
+
+        // Update SBOM entries with discovered vulnerabilities
+        for entry in &repo_entries {
+            if entry.known_vulnerabilities.is_empty() {
+                continue;
+            }
+            if let Some(entry_id) = &entry.id {
+                let _ = agent
+                    .db
+                    .sbom_entries()
+                    .update_one(
+                        doc! { "_id": entry_id },
+                        doc! { "$set": {
+                            "known_vulnerabilities": mongodb::bson::to_bson(&entry.known_vulnerabilities).unwrap_or_default(),
+                            "updated_at": mongodb::bson::DateTime::now(),
+                        }},
+                    )
+                    .await;
+            }
+        }
+
+        // Create notifications for NEW CVEs (dedup against existing notifications)
+        for alert in &alerts {
+            let filter = doc! {
+                "cve_id": &alert.cve_id,
+                "repo_id": &alert.repo_id,
+                "package_name": &alert.affected_package,
+                "package_version": &alert.affected_version,
+            };
+            // Only insert if not already exists (upsert with $setOnInsert)
+            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 update = doc! {
+                "$setOnInsert": mongodb::bson::to_bson(&notification).unwrap_or_default()
+            };
+            match agent
+                .db
+                .cve_notifications()
+                .update_one(filter, update)
+                .upsert(true)
+                .await
+            {
+                Ok(result) if result.upserted_id.is_some() => {
+                    new_notifications += 1;
+                }
+                Err(e) => {
+                    tracing::warn!("CVE monitor: failed to create notification: {e}");
+                }
+                _ => {} // Already exists
+            }
+        }
+    }
+
+    if new_notifications > 0 {
+        tracing::info!("CVE monitor: created {new_notifications} new notification(s)");
+    } else {
+        tracing::info!("CVE monitor: no new CVEs found");
+    }
 }