feat: allow filtering from the command line

backend/Cargo.toml

@@ -10,3 +10,11 @@ members = [
   "server",
 ]
 
+[patch.crates-io]
+csaf-walker = { git = "https://github.com/ctron/csaf-walker", rev = "c96c96f2f2dff240c394c065354740e7208f4ee1" }
+sbom-walker = { git = "https://github.com/ctron/csaf-walker", rev = "c96c96f2f2dff240c394c065354740e7208f4ee1" }
+walker-common = { git = "https://github.com/ctron/csaf-walker", rev = "c96c96f2f2dff240c394c065354740e7208f4ee1" }
+
+#csaf-walker = { path = "../../csaf-walker/csaf" }
+#sbom-walker = { path = "../../csaf-walker/sbom" }
+#walker-common = { path = "../../csaf-walker/common" }

backend/importer/Cargo.toml

@@ -4,16 +4,16 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-trustify-api = { path = "../api"}
-trustify-entity = { path = "../entity"}
+trustify-api = { path = "../api" }
+trustify-entity = { path = "../entity" }
 trustify-migration = { path = "../migration" }
-trustify-common = { path = "../common"}
+trustify-common = { path = "../common" }
 
 anyhow = "1.0.72"
 clap = { version = "4", features = ["derive"] }
 csaf = "0.5"
-csaf-walker = { version = "=0.6.0-alpha.8" }
-sbom-walker = { version = "=0.6.0-alpha.8" }
+csaf-walker = { version = "=0.6.0-alpha.8", default-features = false, features = ["crypto-openssl", "csaf"] }
+sbom-walker = { version = "=0.6.0-alpha.8", default-features = false, features = ["crypto-openssl", "cyclonedx-bom", "spdx-rs"] }
 env_logger = "0.11.0"
 log = "0.4.19"
 packageurl = "0.3.0"

backend/importer/src/csaf/mod.rs

@@ -3,9 +3,11 @@ use ::csaf::Csaf;
 use csaf_walker::retrieve::RetrievingVisitor;
 use csaf_walker::source::{DispatchSource, FileSource, HttpSource};
 use csaf_walker::validation::{ValidatedAdvisory, ValidationError, ValidationVisitor};
+use csaf_walker::visitors::filter::{FilterConfig, FilteringVisitor};
 use csaf_walker::walker::Walker;
 use sha2::digest::Output;
 use sha2::{Digest, Sha256};
+use std::collections::HashSet;
 use std::process::ExitCode;
 use std::time::SystemTime;
 use time::{Date, Month, UtcOffset};
@@ -24,8 +26,19 @@ pub struct ImportCsafCommand {
     pub database: Database,
 
     /// Source URL or path
-    #[arg(short, long)]
-    pub(crate) source: String,
+    pub source: String,
+
+    /// If the source is a full source URL
+    #[arg(long)]
+    pub full_source_url: bool,
+
+    /// Distribution URLs or ROLIE feed URLs to skip
+    #[arg(long)]
+    pub skip_url: Vec<String>,
+
+    /// Only consider files having any of those prefixes. An empty list will accept all files.
+    #[arg(long)]
+    pub only_prefix: Vec<String>,
 }
 
 impl ImportCsafCommand {
@@ -34,20 +47,6 @@ impl ImportCsafCommand {
 
         let system = InnerSystem::with_config(&self.database).await?;
 
-        let filter = |name: &str| {
-            // RHAT: we have advisories marked as "vex"
-            if !name.starts_with("cve-") {
-                return false;
-            }
-
-            // only work with 2023 data for now
-            if !name.starts_with("cve-2023-") {
-                return false;
-            }
-
-            true
-        };
-
         //  because we still have GPG v3 signatures
         let options = ValidationOptions::new().validation_date(SystemTime::from(
             Date::from_calendar_date(2007, Month::January, 1)?
@@ -56,15 +55,23 @@ impl ImportCsafCommand {
         ));
 
         let source: DispatchSource = match Url::parse(&self.source) {
-            Ok(url) => HttpSource::new(
-                url,
-                Fetcher::new(Default::default()).await?,
-                Default::default(),
-            )
-            .into(),
+            Ok(mut url) => {
+                if !self.full_source_url {
+                    url = url.join("/.well-known/csaf/provider-metadata.json")?;
+                }
+                log::info!("Provider metadata: {url}");
+                HttpSource::new(
+                    url,
+                    Fetcher::new(Default::default()).await?,
+                    Default::default(),
+                )
+                .into()
+            }
             Err(_) => FileSource::new(&self.source, None)?.into(),
         };
 
+        // validate (called by retriever)
+
         let visitor =
             ValidationVisitor::new(move |doc: Result<ValidatedAdvisory, ValidationError>| {
                 let system = system.clone();
@@ -78,15 +85,7 @@ impl ImportCsafCommand {
                     };
 
                     let url = doc.url.clone();
-
-                    match url.path_segments().and_then(|path| path.last()) {
-                        Some(name) => {
-                            if !filter(name) {
-                                return Ok(());
-                            }
-                        }
-                        None => return Ok(()),
-                    }
+                    log::info!("processing: {url}");
 
                     if let Err(err) = process(&system, doc).await {
                         log::warn!("Failed to process {url}: {err}");
@@ -97,14 +96,34 @@ impl ImportCsafCommand {
             })
             .with_options(options);
 
-        Walker::new(source.clone())
-            .walk(RetrievingVisitor::new(source, visitor))
-            .await?;
+        // retrieve (called by filter)
+
+        let visitor = RetrievingVisitor::new(source.clone(), visitor);
+
+        //  filter (called by walker)
+
+        let config = FilterConfig::new().extend_only_prefixes(self.only_prefix);
+        let visitor = FilteringVisitor { config, visitor };
+
+        // walker
+
+        let mut walker = Walker::new(source);
+
+        if !self.skip_url.is_empty() {
+            // set up a distribution filter by URL
+            let skip_urls = HashSet::<String>::from_iter(self.skip_url);
+            walker = walker.with_distribution_filter(move |distribution| {
+                skip_urls.contains(distribution.url().as_str())
+            });
+        }
+
+        walker.walk(visitor).await?;
 
         Ok(ExitCode::SUCCESS)
     }
 }
 
+/// Process a single, validated advisory
 async fn process(system: &InnerSystem, doc: ValidatedAdvisory) -> anyhow::Result<()> {
     let csaf = serde_json::from_slice::<Csaf>(&doc.data)?;
 

backend/importer/src/sbom/mod.rs

@@ -21,14 +21,14 @@ pub struct ImportSbomCommand {
 
     /// Source URL or path
     #[arg(short, long)]
-    pub(crate) source: String,
+    pub source: String,
 }
 
 impl ImportSbomCommand {
     pub async fn run(self) -> anyhow::Result<ExitCode> {
         env_logger::init();
 
-        println!("Ingesting SBOMs");
+        log::info!("Ingesting SBOMs");
 
         let system = InnerSystem::with_config(&self.database).await?;
 
@@ -42,8 +42,12 @@ impl ImportSbomCommand {
             Err(_) => FileSource::new(&self.source, None)?.into(),
         };
 
+        // process (called by validator)
+
         let process = process::ProcessVisitor { system };
 
+        // validate (called by retriever)
+
         //  because we still have GPG v3 signatures
         let options = ValidationOptions::new().validation_date(SystemTime::from(
             Date::from_calendar_date(2007, Month::January, 1)?
@@ -53,9 +57,13 @@ impl ImportSbomCommand {
 
         let validation = ValidationVisitor::new(process).with_options(options);
 
-        Walker::new(source.clone())
-            .walk(RetrievingVisitor::new(source, validation))
-            .await?;
+        // retriever (called by filter)
+
+        let visitor = RetrievingVisitor::new(source.clone(), validation);
+
+        // walker
+
+        Walker::new(source).walk(visitor).await?;
 
         Ok(ExitCode::SUCCESS)
     }