Improve import process

backend/Cargo.toml

@@ -11,9 +11,9 @@ members = [
 ]
 
 [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 = { git = "https://github.com/ctron/csaf-walker", rev = "95d1d1e25a0def07e563f6a92722204cee861ea4" }
+sbom-walker = { git = "https://github.com/ctron/csaf-walker", rev = "95d1d1e25a0def07e563f6a92722204cee861ea4" }
+walker-common = { git = "https://github.com/ctron/csaf-walker", rev = "95d1d1e25a0def07e563f6a92722204cee861ea4" }
 
 #csaf-walker = { path = "../../csaf-walker/csaf" }
 #sbom-walker = { path = "../../csaf-walker/sbom" }

backend/api/src/system/advisory/mod.rs

@@ -55,19 +55,23 @@ impl InnerSystem {
 
     pub async fn ingest_advisory(
         &self,
-        identifer: &str,
-        location: &str,
-        sha256: &str,
+        identifier: impl Into<String>,
+        location: impl Into<String>,
+        sha256: impl Into<String>,
         tx: Transactional<'_>,
     ) -> Result<AdvisoryContext, Error> {
-        if let Some(found) = self.get_advisory(identifer, location, sha256).await? {
+        let identifier = identifier.into();
+        let location = location.into();
+        let sha256 = sha256.into();
+
+        if let Some(found) = self.get_advisory(&identifier, &location, &sha256).await? {
             return Ok(found);
         }
 
         let model = entity::advisory::ActiveModel {
-            identifier: Set(identifer.to_string()),
-            location: Set(location.to_string()),
-            sha256: Set(sha256.to_string()),
+            identifier: Set(identifier),
+            location: Set(location),
+            sha256: Set(sha256),
             ..Default::default()
         };
 

backend/importer/.gitignore

@@ -0,0 +1 @@
+/data

backend/importer/Cargo.toml

@@ -29,4 +29,5 @@ walker-common = "=0.6.0-alpha.8"
 url = "2.4.0"
 sha2 = "0.10.6"
 async-trait = "0.1"
-
+indicatif = { version = "0.17.8", features = [] }
+indicatif-log-bridge = "0.2"

backend/importer/README.md

@@ -0,0 +1,28 @@
+# Importing data
+
+The following commands require some env-var for connecting to the database. You can supply them e.g., using `env`:
+
+```bash
+env DB_USER=postgres DB_PASSWORD=eggs <command to run>
+```
+
+## Importing advisories
+
+```bash
+cargo run -p trustify-cli -- importer csaf https://www.redhat.com --only-prefix cve-2023-
+```
+
+Or, using a locally cached version:
+
+```bash
+mkdir data/csaf
+csaf download https://www.redhat.com --only-prefix cve-2023- -d data/csaf
+```
+
+And then:
+
+```bash
+cargo run -p trustify-cli -- importer csaf data/csaf
+```
+
+## Importing SBOMs

backend/importer/src/csaf/mod.rs

@@ -1,23 +1,21 @@
-use ::csaf::document::Category;
-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 crate::progress::init_log_and_progress;
+use ::csaf::{document::Category, Csaf};
+use csaf_walker::{
+    retrieve::RetrievingVisitor,
+    source::{DispatchSource, FileSource, HttpSource},
+    validation::{ValidatedAdvisory, ValidationError, ValidationVisitor},
+    visitors::filter::{FilterConfig, FilteringVisitor},
+    walker::Walker,
+};
 use sha2::{Digest, Sha256};
 use std::collections::HashSet;
 use std::process::ExitCode;
 use std::time::SystemTime;
 use time::{Date, Month, UtcOffset};
-use trustify_api::db::Transactional;
-use trustify_api::system::InnerSystem;
+use trustify_api::{db::Transactional, system::InnerSystem};
 use trustify_common::config::Database;
 use url::Url;
-use walker_common::fetcher::Fetcher;
-use walker_common::utils::hex::Hex;
-use walker_common::validate::ValidationOptions;
+use walker_common::{fetcher::Fetcher, utils::hex::Hex, validate::ValidationOptions};
 
 /// Run the importer
 #[derive(clap::Args, Debug)]
@@ -29,21 +27,24 @@ pub struct ImportCsafCommand {
     pub source: String,
 
     /// If the source is a full source URL
-    #[arg(long)]
+    #[arg(long, env)]
     pub full_source_url: bool,
 
     /// Distribution URLs or ROLIE feed URLs to skip
-    #[arg(long)]
+    #[arg(long, env)]
     pub skip_url: Vec<String>,
 
     /// Only consider files having any of those prefixes. An empty list will accept all files.
-    #[arg(long)]
+    #[arg(long, env)]
     pub only_prefix: Vec<String>,
+
+    #[arg(long, env, default_value_t = 1)]
+    pub workers: usize,
 }
 
 impl ImportCsafCommand {
     pub async fn run(self) -> anyhow::Result<ExitCode> {
-        env_logger::init();
+        let progress = init_log_and_progress()?;
 
         let system = InnerSystem::with_config(&self.database).await?;
 
@@ -85,7 +86,7 @@ impl ImportCsafCommand {
                     };
 
                     let url = doc.url.clone();
-                    log::info!("processing: {url}");
+                    log::debug!("processing: {url}");
 
                     if let Err(err) = process(&system, doc).await {
                         log::warn!("Failed to process {url}: {err}");
@@ -107,7 +108,7 @@ impl ImportCsafCommand {
 
         // walker
 
-        let mut walker = Walker::new(source);
+        let mut walker = Walker::new(source).with_progress(progress);
 
         if !self.skip_url.is_empty() {
             // set up a distribution filter by URL
@@ -117,7 +118,7 @@ impl ImportCsafCommand {
             });
         }
 
-        walker.walk(visitor).await?;
+        walker.walk_parallel(self.workers, visitor).await?;
 
         Ok(ExitCode::SUCCESS)
     }
@@ -133,21 +134,19 @@ async fn process(system: &InnerSystem, doc: ValidatedAdvisory) -> anyhow::Result
     }
 
     log::info!("Ingesting: {}", doc.url);
-    let sha256: String = match doc.sha256.clone() {
+    let sha256 = match doc.sha256.clone() {
         Some(sha) => sha.expected.clone(),
         None => {
-            let mut actual = Sha256::new();
-            actual.update(&doc.data);
-            let digest: Output<Sha256> = actual.finalize();
+            let digest = Sha256::digest(&doc.data);
             Hex(&digest).to_lower()
         }
     };
 
     let advisory = system
         .ingest_advisory(
             &csaf.document.tracking.id,
-            doc.url.as_ref(),
-            &sha256,
+            doc.url.to_string(),
+            sha256,
             Transactional::None,
         )
         .await?;

backend/importer/src/lib.rs

@@ -4,6 +4,7 @@ use clap::Subcommand;
 use trustify_common::config::Database;
 
 mod csaf;
+mod progress;
 mod sbom;
 
 #[derive(Subcommand, Debug)]

backend/importer/src/progress.rs

@@ -0,0 +1,26 @@
+use anyhow::Context;
+use indicatif::MultiProgress;
+use indicatif_log_bridge::LogWrapper;
+use std::io::IsTerminal;
+use walker_common::progress::Progress;
+
+/// Set up the env_logger and attach a progress interface if we are running on a terminal.
+pub(crate) fn init_log_and_progress() -> anyhow::Result<Progress> {
+    let mut builder = env_logger::builder();
+    let logger = builder.build();
+
+    match std::io::stdin().is_terminal() {
+        true => {
+            let max_level = logger.filter();
+            let multi = MultiProgress::new();
+
+            let log = LogWrapper::new(multi.clone(), logger);
+            // NOTE: LogWrapper::try_init is buggy and messes up the log levels
+            log::set_boxed_logger(Box::new(log)).context("failed to initialize logger")?;
+            log::set_max_level(max_level);
+
+            Ok(multi.into())
+        }
+        false => Ok(Progress::default()),
+    }
+}

backend/importer/src/sbom/mod.rs

@@ -1,3 +1,4 @@
+use crate::progress::init_log_and_progress;
 use sbom_walker::{
     retrieve::RetrievingVisitor,
     source::{DispatchSource, FileSource, HttpSource},
@@ -26,7 +27,7 @@ pub struct ImportSbomCommand {
 
 impl ImportSbomCommand {
     pub async fn run(self) -> anyhow::Result<ExitCode> {
-        env_logger::init();
+        let progress = init_log_and_progress()?;
 
         log::info!("Ingesting SBOMs");
 
@@ -63,7 +64,10 @@ impl ImportSbomCommand {
 
         // walker
 
-        Walker::new(source).walk(visitor).await?;
+        Walker::new(source)
+            .with_progress(progress)
+            .walk(visitor)
+            .await?;
 
         Ok(ExitCode::SUCCESS)
     }