vulnera_advisor/
manager.rs

1//! Vulnerability manager for orchestrating syncs and queries.
2//!
3//! The [`VulnerabilityManager`] is the main entry point for using this crate.
4//! Use [`VulnerabilityManagerBuilder`] for flexible configuration.
5
6use crate::config::{Config, OssIndexConfig, StoreConfig};
7use crate::error::{AdvisoryError, Result};
8use crate::models::{Advisory, Enrichment, Event, RangeType, Severity};
9use crate::purl::Purl;
10use crate::sources::epss::EpssSource;
11use crate::sources::kev::KevSource;
12use crate::sources::ossindex::OssIndexSource;
13use crate::sources::{AdvisorySource, ghsa::GHSASource, nvd::NVDSource, osv::OSVSource};
14use crate::store::{AdvisoryStore, DragonflyStore, EnrichmentData, HealthStatus, OssIndexCache};
15use std::collections::HashMap;
16use std::sync::Arc;
17use tracing::{debug, error, info, warn};
18
19/// Options for filtering vulnerability matches.
20#[derive(Debug, Clone, Default)]
21pub struct MatchOptions {
22    /// Minimum CVSS v3 score (0.0 - 10.0).
23    pub min_cvss: Option<f64>,
24    /// Minimum EPSS score (0.0 - 1.0).
25    pub min_epss: Option<f64>,
26    /// Only return KEV (actively exploited) vulnerabilities.
27    pub kev_only: bool,
28    /// Minimum severity level.
29    pub min_severity: Option<Severity>,
30    /// Include enrichment data (EPSS, KEV) in results.
31    pub include_enrichment: bool,
32}
33
34impl MatchOptions {
35    /// Create options that include all vulnerabilities with enrichment.
36    pub fn with_enrichment() -> Self {
37        Self {
38            include_enrichment: true,
39            ..Default::default()
40        }
41    }
42
43    /// Create options for high-severity vulnerabilities only.
44    pub fn high_severity() -> Self {
45        Self {
46            min_severity: Some(Severity::High),
47            include_enrichment: true,
48            ..Default::default()
49        }
50    }
51
52    /// Create options for actively exploited vulnerabilities only.
53    pub fn exploited_only() -> Self {
54        Self {
55            kev_only: true,
56            include_enrichment: true,
57            ..Default::default()
58        }
59    }
60}
61
62/// A key identifying a package for batch queries.
63#[derive(Debug, Clone, Hash, PartialEq, Eq)]
64pub struct PackageKey {
65    /// Package ecosystem (e.g., "npm", "PyPI").
66    pub ecosystem: String,
67    /// Package name.
68    pub name: String,
69    /// Optional version for matching.
70    pub version: Option<String>,
71}
72
73impl PackageKey {
74    /// Create a new package key.
75    pub fn new(ecosystem: impl Into<String>, name: impl Into<String>) -> Self {
76        Self {
77            ecosystem: ecosystem.into(),
78            name: name.into(),
79            version: None,
80        }
81    }
82
83    /// Create a package key with a version.
84    pub fn with_version(
85        ecosystem: impl Into<String>,
86        name: impl Into<String>,
87        version: impl Into<String>,
88    ) -> Self {
89        Self {
90            ecosystem: ecosystem.into(),
91            name: name.into(),
92            version: Some(version.into()),
93        }
94    }
95}
96
97/// Builder for configuring VulnerabilityManager.
98pub struct VulnerabilityManagerBuilder {
99    redis_url: Option<String>,
100    store_config: StoreConfig,
101    sources: Vec<Arc<dyn AdvisorySource + Send + Sync>>,
102    custom_store: Option<Arc<dyn AdvisoryStore + Send + Sync>>,
103    ossindex_source: Option<OssIndexSource>,
104}
105
106impl Default for VulnerabilityManagerBuilder {
107    fn default() -> Self {
108        Self::new()
109    }
110}
111
112impl VulnerabilityManagerBuilder {
113    /// Create a new builder with default settings.
114    pub fn new() -> Self {
115        Self {
116            redis_url: None,
117            store_config: StoreConfig::default(),
118            sources: Vec::new(),
119            custom_store: None,
120            ossindex_source: None,
121        }
122    }
123
124    /// Set the Redis connection URL.
125    pub fn redis_url(mut self, url: impl Into<String>) -> Self {
126        self.redis_url = Some(url.into());
127        self
128    }
129
130    /// Set the store configuration.
131    pub fn store_config(mut self, config: StoreConfig) -> Self {
132        self.store_config = config;
133        self
134    }
135
136    /// Use a custom store implementation.
137    pub fn store(mut self, store: Arc<dyn AdvisoryStore + Send + Sync>) -> Self {
138        self.custom_store = Some(store);
139        self
140    }
141
142    /// Add a vulnerability source.
143    pub fn add_source(mut self, source: Arc<dyn AdvisorySource + Send + Sync>) -> Self {
144        self.sources.push(source);
145        self
146    }
147
148    /// Add the GHSA source with the given token.
149    pub fn with_ghsa(mut self, token: impl Into<String>) -> Self {
150        self.sources.push(Arc::new(GHSASource::new(token.into())));
151        self
152    }
153
154    /// Add the NVD source with optional API key.
155    pub fn with_nvd(mut self, api_key: Option<String>) -> Self {
156        self.sources.push(Arc::new(NVDSource::new(api_key)));
157        self
158    }
159
160    /// Add the OSV source for specified ecosystems.
161    pub fn with_osv(mut self, ecosystems: Vec<String>) -> Self {
162        self.sources.push(Arc::new(OSVSource::new(ecosystems)));
163        self
164    }
165
166    /// Add default OSV ecosystems.
167    pub fn with_osv_defaults(self) -> Self {
168        self.with_osv(vec![
169            "npm".to_string(),
170            "PyPI".to_string(),
171            "Maven".to_string(),
172            "crates.io".to_string(),
173            "Go".to_string(),
174            "Packagist".to_string(),
175            "RubyGems".to_string(),
176            "NuGet".to_string(),
177        ])
178    }
179
180    /// Add the OSS Index source with optional configuration.
181    ///
182    /// OSS Index provides on-demand vulnerability queries by PURL.
183    /// If no config is provided, credentials are loaded from environment variables.
184    pub fn with_ossindex(mut self, config: Option<OssIndexConfig>) -> Self {
185        match OssIndexSource::new(config) {
186            Ok(source) => {
187                self.ossindex_source = Some(source);
188            }
189            Err(e) => {
190                warn!("Failed to configure OSS Index source: {}", e);
191            }
192        }
193        self
194    }
195
196    /// Build the VulnerabilityManager.
197    pub fn build(self) -> Result<VulnerabilityManager> {
198        let store: Arc<dyn AdvisoryStore + Send + Sync> = match self.custom_store {
199            Some(s) => s,
200            None => {
201                let url = self.redis_url.ok_or_else(|| {
202                    AdvisoryError::config("Redis URL is required. Use .redis_url() or .store()")
203                })?;
204                Arc::new(DragonflyStore::with_config(&url, self.store_config)?)
205            }
206        };
207
208        if self.sources.is_empty() {
209            warn!("No sources configured. Use .with_ghsa(), .with_nvd(), or .with_osv()");
210        }
211
212        Ok(VulnerabilityManager {
213            store,
214            sources: self.sources,
215            kev_source: KevSource::new(),
216            epss_source: EpssSource::new(),
217            ossindex_source: self.ossindex_source,
218        })
219    }
220}
221
222/// Main vulnerability manager for syncing and querying advisories.
223pub struct VulnerabilityManager {
224    store: Arc<dyn AdvisoryStore + Send + Sync>,
225    sources: Vec<Arc<dyn AdvisorySource + Send + Sync>>,
226    kev_source: KevSource,
227    epss_source: EpssSource,
228    ossindex_source: Option<OssIndexSource>,
229}
230
231impl VulnerabilityManager {
232    /// Create a new manager from a Config.
233    ///
234    /// This is a convenience method. For more control, use [`VulnerabilityManagerBuilder`].
235    pub async fn new(config: Config) -> Result<Self> {
236        let mut builder = VulnerabilityManagerBuilder::new()
237            .redis_url(&config.redis_url)
238            .store_config(config.store.clone());
239
240        // Add OSV source
241        builder = builder.with_osv_defaults();
242
243        // Add NVD source
244        builder = builder.with_nvd(config.nvd_api_key.clone());
245
246        // Add GHSA source if token is provided
247        if let Some(token) = &config.ghsa_token {
248            builder = builder.with_ghsa(token.clone());
249        }
250
251        // Add OSS Index source if configured
252        if config.ossindex.is_some() {
253            builder = builder.with_ossindex(config.ossindex.clone());
254        }
255
256        builder.build()
257    }
258
259    /// Create a builder for custom configuration.
260    pub fn builder() -> VulnerabilityManagerBuilder {
261        VulnerabilityManagerBuilder::new()
262    }
263
264    /// Get a reference to the underlying store.
265    pub fn store(&self) -> &Arc<dyn AdvisoryStore + Send + Sync> {
266        &self.store
267    }
268
269    /// Check the health of the store connection.
270    pub async fn health_check(&self) -> Result<HealthStatus> {
271        self.store.health_check().await
272    }
273
274    /// Sync advisories from all configured sources.
275    pub async fn sync_all(&self) -> Result<()> {
276        info!("Starting full vulnerability sync...");
277
278        let mut handles = Vec::new();
279
280        for source in &self.sources {
281            let source = source.clone();
282            let store = self.store.clone();
283
284            let handle = tokio::spawn(async move {
285                let last_sync = match store.last_sync(source.name()).await {
286                    Ok(Some(ts)) => match chrono::DateTime::parse_from_rfc3339(&ts) {
287                        Ok(dt) => Some(dt.with_timezone(&chrono::Utc)),
288                        Err(_) => None,
289                    },
290                    _ => None,
291                };
292
293                if let Some(since) = last_sync {
294                    info!("Syncing {} since {}", source.name(), since);
295                } else {
296                    info!("Syncing {} (full)", source.name());
297                }
298
299                match source.fetch(last_sync).await {
300                    Ok(advisories) => {
301                        if !advisories.is_empty() {
302                            if let Err(e) = store.upsert_batch(&advisories, source.name()).await {
303                                error!("Failed to store advisories for {}: {}", source.name(), e);
304                            }
305                        } else {
306                            info!("No new advisories for {}", source.name());
307                            // Update sync timestamp even if no new advisories
308                            if let Err(e) = store.update_sync_timestamp(source.name()).await {
309                                error!(
310                                    "Failed to update sync timestamp for {}: {}",
311                                    source.name(),
312                                    e
313                                );
314                            }
315                        }
316                    }
317                    Err(e) => {
318                        error!("Failed to fetch from {}: {}", source.name(), e);
319                    }
320                }
321            });
322            handles.push(handle);
323        }
324
325        // Wait for all tasks to complete
326        for handle in handles {
327            if let Err(e) = handle.await {
328                error!("Task join error: {}", e);
329            }
330        }
331
332        info!("Sync completed.");
333        Ok(())
334    }
335
336    /// Sync enrichment data (KEV and EPSS).
337    pub async fn sync_enrichment(&self) -> Result<()> {
338        info!("Syncing enrichment data (KEV, EPSS)...");
339
340        // Sync KEV data
341        match self.kev_source.fetch_catalog().await {
342            Ok(kev_entries) => {
343                info!("Processing {} KEV entries", kev_entries.len());
344                for (cve_id, entry) in kev_entries {
345                    let data = EnrichmentData {
346                        epss_score: None,
347                        epss_percentile: None,
348                        is_kev: true,
349                        kev_due_date: entry.due_date_utc().map(|d| d.to_rfc3339()),
350                        kev_date_added: entry.date_added_utc().map(|d| d.to_rfc3339()),
351                        kev_ransomware: Some(entry.is_ransomware_related()),
352                        updated_at: chrono::Utc::now().to_rfc3339(),
353                    };
354                    if let Err(e) = self.store.store_enrichment(&cve_id, &data).await {
355                        debug!("Failed to store KEV enrichment for {}: {}", cve_id, e);
356                    }
357                }
358            }
359            Err(e) => {
360                error!("Failed to fetch KEV catalog: {}", e);
361            }
362        }
363
364        Ok(())
365    }
366
367    /// Query advisories for a specific package.
368    pub async fn query(&self, ecosystem: &str, package: &str) -> Result<Vec<Advisory>> {
369        let advisories = self.store.get_by_package(ecosystem, package).await?;
370        Ok(crate::aggregator::ReportAggregator::aggregate(advisories))
371    }
372
373    /// Query advisories with enrichment data.
374    pub async fn query_enriched(&self, ecosystem: &str, package: &str) -> Result<Vec<Advisory>> {
375        let mut advisories = self.query(ecosystem, package).await?;
376        self.enrich_advisories(&mut advisories).await?;
377        Ok(advisories)
378    }
379
380    /// Query multiple packages in a batch.
381    pub async fn query_batch(
382        &self,
383        packages: &[PackageKey],
384    ) -> Result<HashMap<PackageKey, Vec<Advisory>>> {
385        let mut results = HashMap::new();
386
387        for pkg in packages {
388            let advisories = if let Some(version) = &pkg.version {
389                self.matches(&pkg.ecosystem, &pkg.name, version).await?
390            } else {
391                self.query(&pkg.ecosystem, &pkg.name).await?
392            };
393            results.insert(pkg.clone(), advisories);
394        }
395
396        Ok(results)
397    }
398
399    /// Check if a specific package version is affected by any vulnerabilities.
400    pub async fn matches(
401        &self,
402        ecosystem: &str,
403        package: &str,
404        version: &str,
405    ) -> Result<Vec<Advisory>> {
406        self.matches_with_options(ecosystem, package, version, &MatchOptions::default())
407            .await
408    }
409
410    /// Check if a package version is affected, with filtering options.
411    pub async fn matches_with_options(
412        &self,
413        ecosystem: &str,
414        package: &str,
415        version: &str,
416        options: &MatchOptions,
417    ) -> Result<Vec<Advisory>> {
418        let advisories = self.query(ecosystem, package).await?;
419        let mut matched = Vec::new();
420
421        for mut advisory in advisories {
422            let mut is_vulnerable = false;
423            for affected in &advisory.affected {
424                if affected.package.name != package || affected.package.ecosystem != ecosystem {
425                    continue;
426                }
427
428                // Check explicit versions
429                if affected.versions.contains(&version.to_string()) {
430                    is_vulnerable = true;
431                    break;
432                }
433
434                // Check ranges
435                for range in &affected.ranges {
436                    match range.range_type {
437                        RangeType::Semver => {
438                            if Self::matches_semver_range(version, &range.events) {
439                                is_vulnerable = true;
440                                break;
441                            }
442                        }
443                        RangeType::Ecosystem => {
444                            // For ecosystem ranges, try semver first as fallback
445                            if Self::matches_semver_range(version, &range.events) {
446                                is_vulnerable = true;
447                                break;
448                            }
449                        }
450                        RangeType::Git => {
451                            // Git ranges require commit hash comparison, skip for now
452                        }
453                    }
454                }
455                if is_vulnerable {
456                    break;
457                }
458            }
459
460            if is_vulnerable {
461                // Apply enrichment if requested
462                if options.include_enrichment {
463                    self.enrich_advisory(&mut advisory).await?;
464                }
465
466                // Apply filters
467                if self.advisory_passes_filters(&advisory, options) {
468                    matched.push(advisory);
469                }
470            }
471        }
472
473        Ok(matched)
474    }
475
476    /// Check if a version matches a semver range.
477    fn matches_semver_range(version: &str, events: &[Event]) -> bool {
478        let Ok(v) = semver::Version::parse(version) else {
479            return false;
480        };
481
482        let mut introduced: Option<semver::Version> = None;
483        let mut fixed: Option<semver::Version> = None;
484        let mut last_affected: Option<semver::Version> = None;
485
486        for event in events {
487            match event {
488                Event::Introduced(ver) => {
489                    if let Ok(parsed) = semver::Version::parse(ver) {
490                        introduced = Some(parsed);
491                    } else if ver == "0" {
492                        introduced = Some(semver::Version::new(0, 0, 0));
493                    }
494                }
495                Event::Fixed(ver) => {
496                    if let Ok(parsed) = semver::Version::parse(ver) {
497                        fixed = Some(parsed);
498                    }
499                }
500                Event::LastAffected(ver) => {
501                    if let Ok(parsed) = semver::Version::parse(ver) {
502                        last_affected = Some(parsed);
503                    }
504                }
505                Event::Limit(_) => {}
506            }
507        }
508
509        match (introduced, fixed, last_affected) {
510            (Some(start), Some(end), _) => v >= start && v < end,
511            (Some(start), None, Some(last)) => v >= start && v <= last,
512            (Some(start), None, None) => v >= start,
513            (None, Some(end), _) => v < end,
514            _ => false,
515        }
516    }
517
518    /// Enrich a single advisory with EPSS/KEV data.
519    async fn enrich_advisory(&self, advisory: &mut Advisory) -> Result<()> {
520        // Find CVE aliases
521        let cve_ids = Self::extract_cve_ids(advisory);
522
523        if cve_ids.is_empty() {
524            return Ok(());
525        }
526
527        // Look up enrichment data
528        for cve_id in &cve_ids {
529            if let Ok(Some(data)) = self.store.get_enrichment(cve_id).await {
530                let enrichment = advisory.enrichment.get_or_insert_with(Enrichment::default);
531                enrichment.epss_score = data.epss_score.or(enrichment.epss_score);
532                enrichment.epss_percentile = data.epss_percentile.or(enrichment.epss_percentile);
533                enrichment.is_kev = enrichment.is_kev || data.is_kev;
534                if data.kev_due_date.is_some() {
535                    enrichment.kev_due_date = data
536                        .kev_due_date
537                        .and_then(|s| chrono::DateTime::parse_from_rfc3339(&s).ok())
538                        .map(|d| d.with_timezone(&chrono::Utc));
539                }
540                if data.kev_ransomware.is_some() {
541                    enrichment.kev_ransomware = data.kev_ransomware;
542                }
543            }
544        }
545
546        Ok(())
547    }
548
549    /// Enrich multiple advisories with EPSS/KEV data.
550    async fn enrich_advisories(&self, advisories: &mut [Advisory]) -> Result<()> {
551        for advisory in advisories.iter_mut() {
552            self.enrich_advisory(advisory).await?;
553        }
554        Ok(())
555    }
556
557    /// Extract CVE IDs from an advisory (from ID or aliases).
558    fn extract_cve_ids(advisory: &Advisory) -> Vec<String> {
559        let mut cve_ids = Vec::new();
560
561        if advisory.id.starts_with("CVE-") {
562            cve_ids.push(advisory.id.clone());
563        }
564
565        if let Some(aliases) = &advisory.aliases {
566            for alias in aliases {
567                if alias.starts_with("CVE-") && !cve_ids.contains(alias) {
568                    cve_ids.push(alias.clone());
569                }
570            }
571        }
572
573        cve_ids
574    }
575
576    /// Check if an advisory passes the filter options.
577    fn advisory_passes_filters(&self, advisory: &Advisory, options: &MatchOptions) -> bool {
578        // Check KEV filter
579        if options.kev_only {
580            let is_kev = advisory
581                .enrichment
582                .as_ref()
583                .map(|e| e.is_kev)
584                .unwrap_or(false);
585            if !is_kev {
586                return false;
587            }
588        }
589
590        // Check CVSS filter
591        if let Some(min_cvss) = options.min_cvss {
592            let cvss = advisory
593                .enrichment
594                .as_ref()
595                .and_then(|e| e.cvss_v3_score)
596                .unwrap_or(0.0);
597            if cvss < min_cvss {
598                return false;
599            }
600        }
601
602        // Check EPSS filter
603        if let Some(min_epss) = options.min_epss {
604            let epss = advisory
605                .enrichment
606                .as_ref()
607                .and_then(|e| e.epss_score)
608                .unwrap_or(0.0);
609            if epss < min_epss {
610                return false;
611            }
612        }
613
614        // Check severity filter
615        if let Some(min_severity) = &options.min_severity {
616            let severity = advisory
617                .enrichment
618                .as_ref()
619                .and_then(|e| e.cvss_v3_severity)
620                .unwrap_or(Severity::None);
621            if severity < *min_severity {
622                return false;
623            }
624        }
625
626        true
627    }
628
629    /// Fetch live EPSS scores for CVEs (not from cache).
630    pub async fn fetch_epss_scores(&self, cve_ids: &[&str]) -> Result<HashMap<String, f64>> {
631        let scores = self.epss_source.fetch_scores(cve_ids).await?;
632        Ok(scores.into_iter().map(|(k, v)| (k, v.epss)).collect())
633    }
634
635    /// Check if a CVE is in the CISA KEV catalog.
636    pub async fn is_kev(&self, cve_id: &str) -> Result<bool> {
637        // Check cache first
638        if let Some(data) = self.store.get_enrichment(cve_id).await? {
639            return Ok(data.is_kev);
640        }
641
642        // Fetch from source
643        let entry = self.kev_source.is_kev(cve_id).await?;
644        Ok(entry.is_some())
645    }
646
647    // === OSS Index Methods ===
648
649    /// Query OSS Index for vulnerabilities affecting the given PURLs.
650    ///
651    /// This method provides automatic caching:
652    /// - First checks the cache for each PURL
653    /// - Only queries OSS Index for cache misses
654    /// - Caches results for future queries
655    ///
656    /// # Arguments
657    ///
658    /// * `purls` - Package URLs to query (e.g., "pkg:npm/lodash@4.17.20")
659    ///
660    /// # Returns
661    ///
662    /// Vector of advisories for all vulnerabilities found.
663    ///
664    /// # Errors
665    ///
666    /// Returns an error if OSS Index is not configured or if the query fails.
667    ///
668    /// # Example
669    ///
670    /// ```rust,ignore
671    /// use vulnera_advisors::{VulnerabilityManager, Purl};
672    ///
673    /// let manager = VulnerabilityManager::builder()
674    ///     .redis_url("redis://localhost:6379")
675    ///     .with_ossindex(None)
676    ///     .build()?;
677    ///
678    /// let purls = vec![
679    ///     Purl::new("npm", "lodash").with_version("4.17.20").to_string(),
680    /// ];
681    ///
682    /// let advisories = manager.query_ossindex(&purls).await?;
683    /// ```
684    pub async fn query_ossindex(&self, purls: &[String]) -> Result<Vec<Advisory>> {
685        let source = self.ossindex_source.as_ref().ok_or_else(|| {
686            AdvisoryError::config("OSS Index not configured. Use .with_ossindex() in builder.")
687        })?;
688
689        // Check cache for all PURLs
690        let mut cached_advisories = Vec::new();
691        let mut cache_misses = Vec::new();
692
693        for purl in purls {
694            let cache_key = Purl::cache_key_from_str(purl);
695            match self.store.get_ossindex_cache(&cache_key).await {
696                Ok(Some(cache)) if !cache.is_expired() => {
697                    debug!("OSS Index cache hit for {}", purl);
698                    cached_advisories.extend(cache.advisories);
699                }
700                _ => {
701                    debug!("OSS Index cache miss for {}", purl);
702                    cache_misses.push(purl.clone());
703                }
704            }
705        }
706
707        // Query OSS Index for cache misses
708        if !cache_misses.is_empty() {
709            debug!("Querying OSS Index for {} cache misses", cache_misses.len());
710            let fresh_advisories = source.query_advisories(&cache_misses).await.map_err(|e| {
711                AdvisoryError::SourceFetch {
712                    source_name: "ossindex".to_string(),
713                    message: e.to_string(),
714                }
715            })?;
716
717            // Group advisories by PURL for caching
718            let advisory_map = Self::group_advisories_by_purl(&cache_misses, &fresh_advisories);
719
720            // Cache results for each PURL
721            for (purl, advisories) in &advisory_map {
722                let cache_key = Purl::cache_key_from_str(purl);
723                let cache = OssIndexCache::new(advisories.clone());
724                if let Err(e) = self.store.store_ossindex_cache(&cache_key, &cache).await {
725                    debug!("Failed to cache OSS Index result for {}: {}", purl, e);
726                }
727            }
728
729            // Flatten and add to results
730            for advisories in advisory_map.into_values() {
731                cached_advisories.extend(advisories);
732            }
733        }
734
735        Ok(cached_advisories)
736    }
737
738    /// Query OSS Index for vulnerabilities with fallback to stored advisories.
739    ///
740    /// This method first queries OSS Index, then falls back to the local store
741    /// if the OSS Index query fails or returns no results.
742    ///
743    /// # Arguments
744    ///
745    /// * `packages` - List of packages to query (ecosystem, name, optional version)
746    ///
747    /// # Returns
748    ///
749    /// Map of package keys to their advisories.
750    pub async fn query_batch_with_ossindex(
751        &self,
752        packages: &[PackageKey],
753    ) -> Result<HashMap<PackageKey, Vec<Advisory>>> {
754        let mut results: HashMap<PackageKey, Vec<Advisory>> = HashMap::new();
755
756        // Build PURLs for packages that have versions
757        let (with_version, without_version): (Vec<_>, Vec<_>) =
758            packages.iter().partition(|p| p.version.is_some());
759
760        // Query OSS Index for packages with versions
761        if !with_version.is_empty() && self.ossindex_source.is_some() {
762            let purls: Vec<String> = with_version
763                .iter()
764                .map(|p| {
765                    Purl::new(&p.ecosystem, &p.name)
766                        .with_version(p.version.as_ref().unwrap())
767                        .to_string()
768                })
769                .collect();
770
771            match self.query_ossindex(&purls).await {
772                Ok(advisories) => {
773                    // Group advisories by package key
774                    for pkg in &with_version {
775                        let pkg_advisories: Vec<_> = advisories
776                            .iter()
777                            .filter(|a| {
778                                a.affected.iter().any(|aff| {
779                                    aff.package.ecosystem.eq_ignore_ascii_case(&pkg.ecosystem)
780                                        && aff.package.name == pkg.name
781                                })
782                            })
783                            .cloned()
784                            .collect();
785                        results.insert((*pkg).clone(), pkg_advisories);
786                    }
787                }
788                Err(e) => {
789                    warn!("OSS Index query failed, falling back to local store: {}", e);
790                    // Fallback to local store
791                    for pkg in &with_version {
792                        let advisories = if let Some(version) = &pkg.version {
793                            self.matches(&pkg.ecosystem, &pkg.name, version).await?
794                        } else {
795                            self.query(&pkg.ecosystem, &pkg.name).await?
796                        };
797                        results.insert((*pkg).clone(), advisories);
798                    }
799                }
800            }
801        }
802
803        // Query local store for packages without versions
804        for pkg in &without_version {
805            let advisories = self.query(&pkg.ecosystem, &pkg.name).await?;
806            results.insert((*pkg).clone(), advisories);
807        }
808
809        Ok(results)
810    }
811
812    /// Invalidate cached OSS Index results for specific PURLs.
813    ///
814    /// Use this to force a fresh query on the next call.
815    pub async fn invalidate_ossindex_cache(&self, purls: &[String]) -> Result<()> {
816        for purl in purls {
817            let cache_key = Purl::cache_key_from_str(purl);
818            self.store.invalidate_ossindex_cache(&cache_key).await?;
819        }
820        Ok(())
821    }
822
823    /// Invalidate all cached OSS Index results.
824    pub async fn invalidate_all_ossindex_cache(&self) -> Result<()> {
825        self.store.invalidate_all_ossindex_cache().await?;
826        Ok(())
827    }
828
829    /// Group advisories by their associated PURL.
830    fn group_advisories_by_purl(
831        purls: &[String],
832        advisories: &[Advisory],
833    ) -> HashMap<String, Vec<Advisory>> {
834        let mut map: HashMap<String, Vec<Advisory>> = HashMap::new();
835
836        // Initialize map with empty vectors for all PURLs
837        for purl in purls {
838            map.insert(purl.clone(), Vec::new());
839        }
840
841        // Group advisories
842        for advisory in advisories {
843            for affected in &advisory.affected {
844                // Find matching PURL
845                for purl in purls {
846                    if let Ok(parsed) = Purl::parse(purl) {
847                        if parsed.name == affected.package.name {
848                            map.entry(purl.clone()).or_default().push(advisory.clone());
849                            break;
850                        }
851                    }
852                }
853            }
854        }
855
856        map
857    }
858}
vulnera_advisor/manager.rs

vulnera_advisor/
manager.rs
