feat: implement csaf importer for server

importer/README.md

@@ -34,7 +34,7 @@ cargo run -p trustify-trustd -- importer csaf data/csaf
 ## Importing SBOMs
 
 ```bash
-cargo run -p trustify-trustd -- importer sbom https://access.redhat.com/security/data/sbom/beta/
+cargo run -p trustify-trustd -- importer sbom https://access.redhat.com/security/data/sbom/beta/  --key https://access.redhat.com/security/data/97f5eac4.txt#77E79ABE93673533ED09EBE2DCE3823597F5EAC4
 ```
 
 Or, using a locally cached version:

importer/src/csaf/mod.rs

@@ -1,24 +1,24 @@
 use crate::progress::init_log_and_progress;
-use csaf::Csaf;
 use csaf_walker::{
     retrieve::RetrievingVisitor,
     source::{DispatchSource, FileSource, HttpSource},
-    validation::{ValidatedAdvisory, ValidationError, ValidationVisitor},
+    validation::ValidationVisitor,
     visitors::filter::{FilterConfig, FilteringVisitor},
     walker::Walker,
 };
-use sha2::{Digest, Sha256};
+use parking_lot::Mutex;
 use std::collections::HashSet;
 use std::process::ExitCode;
-use std::time::SystemTime;
-use time::{Date, Month, UtcOffset};
-use trustify_common::config::Database;
-use trustify_common::db;
+use std::sync::Arc;
+use trustify_common::{config::Database, db};
 use trustify_graph::graph::Graph;
-use trustify_ingestors as ingestors;
+use trustify_module_importer::server::{
+    common::validation,
+    csaf::storage,
+    report::{Report, ReportBuilder, ScannerError, SplitScannerError},
+};
 use url::Url;
-use walker_common::utils::hex::Hex;
-use walker_common::{fetcher::Fetcher, validate::ValidationOptions};
+use walker_common::{fetcher::Fetcher, progress::Progress};
 
 /// Import from a CSAF source
 #[derive(clap::Args, Debug)]
@@ -49,20 +49,27 @@ impl ImportCsafCommand {
     pub async fn run(self) -> anyhow::Result<ExitCode> {
         let progress = init_log_and_progress()?;
 
+        log::info!("Ingesting CSAF");
+
+        let (report, result) = self.run_once(progress).await.split()?;
+
+        log::info!("Import report: {report:#?}");
+
+        result.map(|()| ExitCode::SUCCESS)
+    }
+
+    pub async fn run_once(self, progress: Progress) -> Result<Report, ScannerError> {
+        let report = Arc::new(Mutex::new(ReportBuilder::new()));
+
         let db = db::Database::with_external_config(&self.database, false).await?;
         let system = Graph::new(db);
 
-        //  because we still have GPG v3 signatures
-        let options = ValidationOptions::new().validation_date(SystemTime::from(
-            Date::from_calendar_date(2007, Month::January, 1)?
-                .midnight()
-                .assume_offset(UtcOffset::UTC),
-        ));
-
         let source: DispatchSource = match Url::parse(&self.source) {
             Ok(mut url) => {
                 if !self.full_source_url {
-                    url = url.join("/.well-known/csaf/provider-metadata.json")?;
+                    url = url
+                        .join("/.well-known/csaf/provider-metadata.json")
+                        .map_err(|err| ScannerError::Critical(err.into()))?;
                 }
                 log::info!("Provider metadata: {url}");
                 HttpSource::new(
@@ -75,31 +82,18 @@ impl ImportCsafCommand {
             Err(_) => FileSource::new(&self.source, None)?.into(),
         };
 
+        // storage (called by validator)
+
+        let visitor = storage::StorageVisitor {
+            system,
+            report: report.clone(),
+        };
+
         // validate (called by retriever)
 
-        let visitor =
-            ValidationVisitor::new(move |doc: Result<ValidatedAdvisory, ValidationError>| {
-                let system = system.clone();
-                async move {
-                    let doc = match doc {
-                        Ok(doc) => doc,
-                        Err(err) => {
-                            log::warn!("Ignore error: {err}");
-                            return Ok::<(), anyhow::Error>(());
-                        }
-                    };
-
-                    let url = doc.url.clone();
-                    log::debug!("processing: {url}");
-
-                    if let Err(err) = process(&system, doc).await {
-                        log::warn!("Failed to process {url}: {err}");
-                    }
-
-                    Ok(())
-                }
-            })
-            .with_options(options);
+        //  because we still have GPG v3 signatures
+        let options = validation::options(true)?;
+        let visitor = ValidationVisitor::new(visitor).with_options(options);
 
         // retrieve (called by filter)
 
@@ -122,24 +116,19 @@ impl ImportCsafCommand {
             });
         }
 
-        walker.walk_parallel(self.workers, visitor).await?;
-
-        Ok(ExitCode::SUCCESS)
-    }
-}
-
-/// Process a single, validated advisory
-async fn process(system: &Graph, doc: ValidatedAdvisory) -> anyhow::Result<()> {
-    let csaf = serde_json::from_slice::<Csaf>(&doc.data)?;
-
-    let sha256 = match doc.sha256.clone() {
-        Some(sha) => sha.expected.clone(),
-        None => {
-            let digest = Sha256::digest(&doc.data);
-            Hex(&digest).to_lower()
+        walker
+            .walk_parallel(self.workers, visitor)
+            .await // if the walker fails, we record the outcome as part of the report, but skip any
+            // further processing, like storing the marker
+            .map_err(|err| ScannerError::Normal {
+                err: err.into(),
+                report: report.lock().clone().build(),
+            })?;
+
+        Ok(match Arc::try_unwrap(report) {
+            Ok(report) => report.into_inner(),
+            Err(report) => report.lock().clone(),
         }
-    };
-
-    ingestors::advisory::csaf::ingest(system, csaf, &sha256, doc.url.as_str()).await?;
-    Ok(())
+        .build())
+    }
 }

importer/src/sbom/mod.rs

@@ -8,16 +8,15 @@ use sbom_walker::{
 };
 use std::process::ExitCode;
 use std::sync::Arc;
-use std::time::SystemTime;
-use time::{Date, Month, UtcOffset};
 use trustify_common::{config::Database, db};
 use trustify_graph::graph::Graph;
 use trustify_module_importer::server::{
+    common::validation,
     report::{Report, ReportBuilder, ScannerError, SplitScannerError},
     sbom::storage,
 };
 use url::Url;
-use walker_common::{fetcher::Fetcher, progress::Progress, validate::ValidationOptions};
+use walker_common::{fetcher::Fetcher, progress::Progress};
 
 /// Import SBOMs
 #[derive(clap::Args, Debug)]
@@ -69,24 +68,18 @@ impl ImportSbomCommand {
             Err(_) => FileSource::new(&self.source, None)?.into(),
         };
 
-        // process (called by validator)
+        // storage (called by validator)
 
-        let process = storage::StorageVisitor {
+        let storage = storage::StorageVisitor {
             system,
             report: report.clone(),
         };
 
         // 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)
-                .map_err(|err| ScannerError::Critical(err.into()))?
-                .midnight()
-                .assume_offset(UtcOffset::UTC),
-        ));
-
-        let validation = ValidationVisitor::new(process).with_options(options);
+        let options = validation::options(true)?;
+        let validation = ValidationVisitor::new(storage).with_options(options);
 
         // retriever (called by filter)
 

ingestors/Cargo.toml

@@ -9,20 +9,21 @@ trustify-common = { path = "../common"}
 trustify-graph = { path = "../graph"}
 trustify-entity = { path = "../entity" }
 
-serde = { version = "1.0.183", features = ["derive"] }
-serde_json = "1.0.114"
-chrono = { version = "0.4.35", features = ["serde"] }
-tokio = { version = "1.30.0", features = ["full"] }
-thiserror = "1.0.58"
 anyhow = "1.0.72"
+chrono = { version = "0.4.35", features = ["serde"] }
+csaf = "0.5.0"
+env_logger = "0.11.0"
+hex = "0.4.3"
+humantime = "2"
 log = "0.4.19"
-env_logger = "0.10.0"
 reqwest = "0.11"
 ring = "0.17.8"
-hex = "0.4.3"
-csaf = "0.5.0"
-sha2 = "0.10.8"
 sea-orm = "*"
+serde = { version = "1.0.183", features = ["derive"] }
+serde_json = "1.0.114"
+sha2 = "0.10.8"
+thiserror = "1.0.58"
+tokio = { version = "1.30.0", features = ["full"] }
 
 [dev-dependencies]
 test-log = { version = "0.2.15", features = ["env_logger", "trace"] }

ingestors/src/advisory/csaf.rs

@@ -1,4 +1,5 @@
 use csaf::Csaf;
+use std::time::Instant;
 
 use trustify_common::db::Transactional;
 use trustify_graph::graph::Graph;
@@ -9,15 +10,25 @@ pub async fn ingest(
     sha256: &str,
     location: &str,
 ) -> anyhow::Result<i32> {
-    let identifier = &csaf.document.tracking.id;
+    let identifier = csaf.document.tracking.id.clone();
 
-    log::info!("Ingesting: {} from {}", identifier, location);
+    log::debug!("Ingesting: {} from {}", identifier, location);
+
+    let start = Instant::now();
 
     let advisory = system
-        .ingest_advisory(identifier, location, sha256, Transactional::None)
+        .ingest_advisory(&identifier, location, sha256, Transactional::None)
         .await?;
 
     advisory.ingest_csaf(csaf).await?;
 
+    let duration = Instant::now() - start;
+    log::info!(
+        "Ingested: {} from {}: took {}",
+        identifier,
+        location,
+        humantime::Duration::from(duration),
+    );
+
     Ok(advisory.advisory.id)
 }

modules/importer/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 trustify-common = { path = "../../common" }
 trustify-entity = { path = "../../entity" }
 trustify-graph = { path = "../../graph" }
+trustify-ingestors = { path = "../../ingestors" }
 
 actix-web = "4"
 anyhow = "1.0.72"

modules/importer/README.md

@@ -1,10 +1,16 @@
+Create a new CSAF importer:
+
+```bash
+http POST localhost:8080/api/v1/importer/redhat-csaf csaf[source]=https://redhat.com/.well-known/csaf/provider-metadata.json csaf[disabled]:=false csaf[onlyPatterns][]="^cve-2023-" csaf[period]=30s csaf[v3Signatures]:=true
+```
+
 Create a new SBOM importer:
 
 ```bash
-http POST localhost:8080/api/v1/importer/test sbom[source]=https://access.redhat.com/security/data/sbom/beta/ sbom[keys][]=https://access.redhat.com/security/data/97f5eac4.txt#77E79ABE93673533ED09EBE2DCE3823597F5EAC4 sbom[disabled]:=false sbom[onlyPatterns][]=quarkus sbom[period]=30s
+http POST localhost:8080/api/v1/importer/redhat-sbom sbom[source]=https://access.redhat.com/security/data/sbom/beta/ sbom[keys][]=https://access.redhat.com/security/data/97f5eac4.txt#77E79ABE93673533ED09EBE2DCE3823597F5EAC4 sbom[disabled]:=false sbom[onlyPatterns][]=quarkus sbom[period]=30s sbom[v3Signatures]:=true
 ```
 
-Get all importer:
+Get all importers:
 
 ```bash
 http GET localhost:8080/api/v1/importer
@@ -13,11 +19,13 @@ http GET localhost:8080/api/v1/importer
 Get a specific importer:
 
 ```bash
-http GET localhost:8080/api/v1/importer/test
+http GET localhost:8080/api/v1/importer/redhat-csaf
+http GET localhost:8080/api/v1/importer/redhat-sbom
 ```
 
 Get reports:
 
 ```bash
-http GET localhost:8080/api/v1/importer/test/report
+http GET localhost:8080/api/v1/importer/redhat-csaf/report
+http GET localhost:8080/api/v1/importer/redhat-sbom/report
 ```

modules/importer/src/model/csaf.rs

@@ -0,0 +1,29 @@
+use super::*;
+
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct CsafImporter {
+    #[serde(flatten)]
+    pub common: CommonImporter,
+
+    pub source: String,
+
+    #[serde(default)]
+    pub v3_signatures: bool,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub only_patterns: Vec<String>,
+}
+
+impl Deref for CsafImporter {
+    type Target = CommonImporter;
+
+    fn deref(&self) -> &Self::Target {
+        &self.common
+    }
+}
+
+impl DerefMut for CsafImporter {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.common
+    }
+}