From 6ae27ab2cbc3d610573120cd5d562145523c4f7e Mon Sep 17 00:00:00 2001 From: rh-jfuller Date: Fri, 9 Oct 2026 14:10:53 +0200 Subject: [PATCH 1/4] feat(ingestor): gate semantic validators behind Cargo feature The validator backends (scheck, CSAF spec, Conforma) were always compiled in eg. pulling scheck dependency tree into every binary. Add a 'semantic-validation' Cargo feature, on by default, gating the three backend modules and making an optional dependency. --- .github/workflows/ci.yaml | 7 +++ modules/ingestor/Cargo.toml | 10 +++- .../ingestor/src/service/validation/config.rs | 52 ++++++++++++------- .../ingestor/src/service/validation/mod.rs | 4 ++ 4 files changed, 53 insertions(+), 20 deletions(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 6978d0266f..c485d1ecdb 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -74,6 +74,13 @@ jobs: GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}" # for embedded postgresql run: cargo clippy --all-targets --all-features -- -D warnings -D clippy::unwrap_used -D clippy::expect_used -D clippy::panic -D clippy::todo -D clippy::unimplemented + # ADR 00022: every Cargo feature must also be built with the feature off. + # Lib only: dev-dependencies pull in the default features again. + - name: Clippy (semantic-validation off) + env: + GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}" # for embedded postgresql + run: cargo clippy -p trustify-module-ingestor --no-default-features -- -D warnings -D clippy::unwrap_used -D clippy::expect_used -D clippy::panic -D clippy::todo -D clippy::unimplemented + test: runs-on: ubuntu-24.04 diff --git a/modules/ingestor/Cargo.toml b/modules/ingestor/Cargo.toml index 18255f1cde..86251030b5 100644 --- a/modules/ingestor/Cargo.toml +++ b/modules/ingestor/Cargo.toml @@ -6,6 +6,14 @@ publish.workspace = true license.workspace = true rust-version.workspace = true +[features] +default = ["semantic-validation"] + +# The semantic validator subsystem (ADR 00020): scheck, CSAF spec and Conforma +# backends. Disabling it drops the `scheck` dependency tree; the validator +# config still parses, but building a configured validator fails. See ADR 00022. +semantic-validation = ["dep:scheck"] + [dependencies] trustify-auth = { workspace = true } trustify-common = { workspace = true } @@ -33,7 +41,7 @@ quick-xml = { workspace = true } reqwest = { workspace = true, features = ["stream"] } roxmltree = { workspace = true } sbom-walker = { workspace = true } -scheck = { workspace = true } +scheck = { workspace = true, optional = true } sea-orm = { workspace = true } sea-query = { workspace = true } semver = { workspace = true } diff --git a/modules/ingestor/src/service/validation/config.rs b/modules/ingestor/src/service/validation/config.rs index f60cde22ab..d64f225f43 100644 --- a/modules/ingestor/src/service/validation/config.rs +++ b/modules/ingestor/src/service/validation/config.rs @@ -3,14 +3,15 @@ //! The default configuration is empty, which disables validation entirely and //! preserves the pre-existing ingestion behaviour. See ADR 00020. +#[cfg(feature = "semantic-validation")] +use crate::service::validation::{ScheckValidator, conforma, csaf, scheck}; use crate::service::{ Format, - validation::{ - OnError, ScheckValidator, Severity, ValidationMode, Validator, conforma, csaf, scheck, - }, + validation::{OnError, Severity, ValidationMode, Validator}, }; +#[cfg(feature = "semantic-validation")] use anyhow::Context; -use std::{collections::HashSet, fs, path::PathBuf, sync::Arc}; +use std::{collections::HashSet, path::PathBuf, sync::Arc}; /// Configuration for the complete set of validators. #[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)] @@ -113,24 +114,35 @@ pub fn build(config: &ValidatorsConfig) -> Result>, anyho "duplicate validator name: {}", validator.name ); - match &validator.backend { - Backend::Scheck { rules, phase } => { - validators.push(Arc::new(build_scheck(validator, rules, phase.as_deref())?)); - } - Backend::Csaf { profile } => { - validators.push(Arc::new(csaf::Validator::new( - validator, - profile.as_deref(), - ))); - } - Backend::Conforma(conforma) => { - validators.push(Arc::new(conforma::build(validator, conforma)?)); - } - } + validators.push(build_one(validator)?); } Ok(validators) } +/// Build a single validator from its configuration. +fn build_one(validator: &ValidatorConfig) -> Result, anyhow::Error> { + match &validator.backend { + #[cfg(feature = "semantic-validation")] + Backend::Scheck { rules, phase } => { + Ok(Arc::new(build_scheck(validator, rules, phase.as_deref())?)) + } + #[cfg(feature = "semantic-validation")] + Backend::Csaf { profile } => Ok(Arc::new(csaf::Validator::new( + validator, + profile.as_deref(), + ))), + #[cfg(feature = "semantic-validation")] + Backend::Conforma(conforma) => Ok(Arc::new(conforma::build(validator, conforma)?)), + #[allow(unreachable_patterns)] + backend => anyhow::bail!( + "validator '{}' uses backend {backend:?}, which is not compiled into this build \ + (the 'semantic-validation' feature is disabled)", + validator.name, + ), + } +} + +#[cfg(feature = "semantic-validation")] fn build_scheck( config: &ValidatorConfig, rules: &[PathBuf], @@ -138,7 +150,7 @@ fn build_scheck( ) -> Result { let mut schemas = Vec::with_capacity(rules.len()); for path in rules { - let contents = fs::read_to_string(path) + let contents = std::fs::read_to_string(path) .with_context(|| format!("reading scheck ruleset {}", path.display()))?; schemas.push(scheck::parse_ruleset(path, &contents)?); } @@ -264,6 +276,7 @@ validators: ); } + #[cfg(feature = "semantic-validation")] #[test] fn builds_csaf_validator_from_config() { let config = ValidatorsConfig { @@ -284,6 +297,7 @@ validators: assert_eq!(validators[0].name(), "csaf-spec"); } + #[cfg(feature = "semantic-validation")] #[test] fn builds_scheck_validator_from_ruleset_file() { let dir = tempfile::tempdir().expect("tempdir"); diff --git a/modules/ingestor/src/service/validation/mod.rs b/modules/ingestor/src/service/validation/mod.rs index 797af9545b..a8b2d2841b 100644 --- a/modules/ingestor/src/service/validation/mod.rs +++ b/modules/ingestor/src/service/validation/mod.rs @@ -8,11 +8,15 @@ //! See ADR 00020 for the design and rationale. pub mod config; +#[cfg(feature = "semantic-validation")] pub mod conforma; +#[cfg(feature = "semantic-validation")] pub mod csaf; +#[cfg(feature = "semantic-validation")] pub mod scheck; pub use config::{Backend, ConformaConfig, ValidatorConfig, ValidatorsConfig, build}; +#[cfg(feature = "semantic-validation")] pub use scheck::ScheckValidator; use crate::service::Format; From 3c91209b3777f021fee46d31b9f9605d20d2b3b3 Mon Sep 17 00:00:00 2001 From: rh-jfuller Date: Fri, 9 Oct 2026 17:21:41 +0200 Subject: [PATCH 2/4] feat(ingestor): persist semantic validation reports --- Cargo.lock | 1 + common/auth/src/authenticator/default.rs | 1 + common/auth/src/permission.rs | 3 + docs/oidc.md | 1 + entity/src/lib.rs | 1 + entity/src/validation_report.rs | 216 +++++++ etc/validators/README.md | 57 +- etc/validators/validators.yaml | 16 +- migration/src/lib.rs | 2 + .../src/m0002390_create_validation_report.rs | 239 +++++++ modules/fundamental/src/endpoints.rs | 14 +- modules/fundamental/src/lib.rs | 1 + .../src/validation/endpoints/mod.rs | 89 +++ .../src/validation/endpoints/test.rs | 141 ++++ modules/fundamental/src/validation/mod.rs | 5 + .../fundamental/src/validation/model/mod.rs | 60 ++ .../fundamental/src/validation/service/mod.rs | 91 +++ modules/ingestor/Cargo.toml | 3 +- modules/ingestor/src/endpoints.rs | 8 +- modules/ingestor/src/model.rs | 2 +- modules/ingestor/src/service/mod.rs | 450 +++++++++++-- .../ingestor/src/service/validation/config.rs | 217 ++++++- .../src/service/validation/conforma.rs | 24 +- .../ingestor/src/service/validation/csaf.rs | 22 +- .../ingestor/src/service/validation/mod.rs | 10 +- .../ingestor/src/service/validation/scheck.rs | 25 +- .../ingestor/src/service/validation/store.rs | 602 ++++++++++++++++++ openapi.yaml | 283 +++++++- server/src/profile/api.rs | 4 +- 29 files changed, 2488 insertions(+), 100 deletions(-) create mode 100644 entity/src/validation_report.rs create mode 100644 migration/src/m0002390_create_validation_report.rs create mode 100644 modules/fundamental/src/validation/endpoints/mod.rs create mode 100644 modules/fundamental/src/validation/endpoints/test.rs create mode 100644 modules/fundamental/src/validation/mod.rs create mode 100644 modules/fundamental/src/validation/model/mod.rs create mode 100644 modules/fundamental/src/validation/service/mod.rs create mode 100644 modules/ingestor/src/service/validation/store.rs diff --git a/Cargo.lock b/Cargo.lock index 9e4e3ba81a..66957a94bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9600,6 +9600,7 @@ dependencies = [ "serde-cyclonedx", "serde_json", "serde_yaml_ng", + "sha2 0.11.0", "spdx-expression", "spdx-rs", "strum 0.28.0", diff --git a/common/auth/src/authenticator/default.rs b/common/auth/src/authenticator/default.rs index 54e2aa6d28..5474177a31 100644 --- a/common/auth/src/authenticator/default.rs +++ b/common/auth/src/authenticator/default.rs @@ -32,6 +32,7 @@ pub const DEFAULT_SCOPE_MAPPINGS: &[(&str, &[&str])] = &[ "read.metadata", "read.sbom", "read.sbomGroup", + "read.validation", "read.weakness", "read.systemInformation", ], diff --git a/common/auth/src/permission.rs b/common/auth/src/permission.rs index 53c74c929e..f125ef6e0b 100644 --- a/common/auth/src/permission.rs +++ b/common/auth/src/permission.rs @@ -109,6 +109,9 @@ permission! { #[strum(serialize = "delete.sbomGroup")] DeleteSbomGroup, + #[strum(serialize = "read.validation")] + ReadValidation, + #[strum(serialize = "upload.dataset")] UploadDataset, diff --git a/docs/oidc.md b/docs/oidc.md index 17e494aa6e..2081b159d8 100644 --- a/docs/oidc.md +++ b/docs/oidc.md @@ -95,6 +95,7 @@ The CLI alternative provides predefined scope mappings that cannot be customized "read.importer", "read.metadata", "read.sbom", + "read.validation", "read.weakness", "read.systemInformation", ], diff --git a/entity/src/lib.rs b/entity/src/lib.rs index d9af86315c..e6c81d0fc2 100644 --- a/entity/src/lib.rs +++ b/entity/src/lib.rs @@ -45,6 +45,7 @@ pub mod sbom_package_license; pub mod source_document; pub mod status; pub mod user_preferences; +pub mod validation_report; pub mod version_range; pub mod version_scheme; pub mod versioned_purl; diff --git a/entity/src/validation_report.rs b/entity/src/validation_report.rs new file mode 100644 index 0000000000..7f1f10d830 --- /dev/null +++ b/entity/src/validation_report.rs @@ -0,0 +1,216 @@ +use sea_orm::entity::prelude::*; + +/// The result of one semantic validator run against one document. +/// +/// Rows are append-only: a run inserts a row only when its result differs from +/// the latest stored one for the same `(document_sha256, validator)`. The +/// current result is therefore the row with the greatest `created_at`, and +/// older rows are the history of how that verdict changed. +/// +/// Documents rejected by a `verify` validator are recorded with +/// `blocked = true`. Those never reach storage, so there is no `source_document` +/// row to reference — the digest is the only link to the document. +#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] +#[sea_orm(table_name = "validation_report")] +pub struct Model { + #[sea_orm(primary_key)] + pub id: Uuid, + /// SHA-256 of the raw document, as lowercase hex. + pub document_sha256: String, + /// The [`crate::validation_report::Model::validator`] name from configuration. + pub validator: String, + /// Whether the validator gated ingestion or only reported. + pub mode: Mode, + /// Whether the document met the validator's threshold. + pub outcome: Outcome, + /// True when this run rejected the document, so nothing was ingested. + pub blocked: bool, + /// Highest severity across the findings; `None` when there are none. + pub max_severity: Option, + /// Number of findings produced, before truncation. + pub finding_count: i32, + /// The findings, as a JSON array; truncated to the configured caps. + pub findings: serde_json::Value, + /// True when `findings` holds fewer entries than `finding_count`. + pub truncated: bool, + /// Digest of the result, used to detect that a re-run changed nothing. + pub content_hash: String, + /// Digest of the validator configuration that produced this result. + pub config_fingerprint: String, + /// How the document entered the system. + pub ingest_source: IngestSource, + /// Name of the importer, when `ingest_source` is [`IngestSource::Importer`]. + pub importer_name: Option, + pub created_at: time::OffsetDateTime, +} + +/// Whether a validator gates ingestion or only records findings. +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Eq, + EnumIter, + DeriveActiveEnum, + serde::Serialize, + serde::Deserialize, + strum::Display, +)] +#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "validation_mode")] +#[serde(rename_all = "lowercase")] +#[strum(serialize_all = "lowercase", ascii_case_insensitive)] +pub enum Mode { + /// Findings are recorded but never block ingestion. + #[sea_orm(string_value = "report")] + Report, + /// A failing outcome blocks ingestion. + #[sea_orm(string_value = "verify")] + Verify, +} + +/// Whether a validator considered the document acceptable. +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Eq, + EnumIter, + DeriveActiveEnum, + serde::Serialize, + serde::Deserialize, + strum::Display, +)] +#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "validation_outcome")] +#[serde(rename_all = "lowercase")] +#[strum(serialize_all = "lowercase", ascii_case_insensitive)] +pub enum Outcome { + /// No finding met the validator's blocking threshold. + #[sea_orm(string_value = "passed")] + Passed, + /// At least one finding met the validator's blocking threshold. + #[sea_orm(string_value = "failed")] + Failed, +} + +/// Severity of a validation finding. +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Eq, + PartialOrd, + Ord, + EnumIter, + DeriveActiveEnum, + serde::Serialize, + serde::Deserialize, + strum::Display, +)] +#[sea_orm( + rs_type = "String", + db_type = "Enum", + enum_name = "validation_severity" +)] +#[serde(rename_all = "lowercase")] +#[strum(serialize_all = "lowercase", ascii_case_insensitive)] +pub enum Severity { + #[sea_orm(string_value = "info")] + Info, + #[sea_orm(string_value = "warning")] + Warning, + #[sea_orm(string_value = "error")] + Error, + #[sea_orm(string_value = "fatal")] + Fatal, +} + +/// How a document entered the system. +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Eq, + EnumIter, + DeriveActiveEnum, + serde::Serialize, + serde::Deserialize, + strum::Display, +)] +#[sea_orm( + rs_type = "String", + db_type = "Enum", + enum_name = "validation_ingest_source" +)] +#[serde(rename_all = "lowercase")] +#[strum(serialize_all = "lowercase", ascii_case_insensitive)] +pub enum IngestSource { + /// Uploaded through the API. + #[sea_orm(string_value = "api")] + Api, + /// Fetched by an importer run. + #[sea_orm(string_value = "importer")] + Importer, + /// Loaded as part of a dataset archive. + #[sea_orm(string_value = "dataset")] + Dataset, + /// Ingested by internal code, including tests. + #[sea_orm(string_value = "internal")] + Internal, +} + +/// Relation to [`super::source_document`] by digest. +/// +/// This is a query-time join only — there is deliberately no foreign key in the +/// schema, because reports are also written for rejected documents, which never +/// get a `source_document` row. +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm( + belongs_to = "super::source_document::Entity" + from = "Column::DocumentSha256" + to = "super::source_document::Column::Sha256")] + SourceDocument, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::SourceDocument.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} + +#[cfg(test)] +mod test { + use super::*; + use serde_json::json; + use test_log::test; + + #[test] + fn severity_ordering_and_strings() { + assert!(Severity::Fatal > Severity::Error); + assert!(Severity::Error > Severity::Warning); + assert!(Severity::Warning > Severity::Info); + + for (s, severity) in [ + ("info", Severity::Info), + ("warning", Severity::Warning), + ("error", Severity::Error), + ("fatal", Severity::Fatal), + ] { + assert_eq!(severity.to_string(), s); + assert_eq!(json!(severity), json!(s)); + } + } + + #[test] + fn enum_strings() { + assert_eq!(json!(Mode::Verify), json!("verify")); + assert_eq!(json!(Outcome::Failed), json!("failed")); + assert_eq!(json!(IngestSource::Importer), json!("importer")); + } +} diff --git a/etc/validators/README.md b/etc/validators/README.md index f512533fd8..e89c42f0f7 100644 --- a/etc/validators/README.md +++ b/etc/validators/README.md @@ -1,14 +1,14 @@ # Example semantic validators Ready-to-use configuration and rulesets for the ingestion-time semantic -validators described in ADR 00020. Validation is **disabled by default**; +validators described in ADR 00021. Validation is **disabled by default**; enabling it is opt-in via a config file. ## Layout | File | Purpose | |------|---------| -| `validators.yaml` | The validator set. Referenced by `--validators-config` / `TRUSTD_VALIDATORS_CONFIG`. | +| `validators.yaml` | The validator set and the storage settings. Referenced by `--validators-config` / `TRUSTD_VALIDATORS_CONFIG`. | | `rules/csaf-mandatory.json` | CSAF 2.0 required-field checks (JSON ruleset). | | `rules/spdx-min.json` | SPDX 2.2/2.3 minimum-element checks (JSON ruleset). | | `rules/cyclonedx-min.json` | CycloneDX minimum checks (JSON ruleset). | @@ -27,7 +27,45 @@ TRUSTD_VALIDATORS_CONFIG=etc/validators/validators.yaml trustd api `rules` paths in `validators.yaml` are resolved relative to the process working directory. Use absolute paths in production deployments. -To view associated logs set: +## Top-level settings + +Alongside `validators`, the file carries two settings that apply to all of them: + +- `persist_reports` — defaults to `true`. When `false`, nothing is written to + the `validation_report` table and every result is logged at `debug` instead. + This overrides the per-validator `persist` setting. +- `caps.max_findings` / `caps.max_findings_bytes` — bound how much of a report + is stored, defaulting to 200 findings and 64 KiB. Validator output is + untrusted and unbounded, and a loose ruleset against a large document can + otherwise produce megabytes per report. A report that hits either limit is + stored with the surplus findings dropped and `truncated` set, while + `finding_count` still reports everything the validator found. + +## Stored results + +Every validator result for a document that is ingested is recorded, in both +`report` and `verify` mode. A document that `verify` **rejects** is recorded +too, with `blocked` set — it never reaches storage, so this row is the only +trace that the instance refused it. + +The table is append-only. A validator that runs again against an unchanged +document with an unchanged configuration writes nothing; a row appears only +when the result or the validator configuration differs from the last one +stored. Re-running an importer over documents that have not changed therefore +does not grow the table, and when a ruleset does change, the previous verdict +is still there next to the new one. + +Reports are read back through `GET /api/v3/validation`, which is filterable and +paginated, and `GET /api/v3/validation/{key}`, where the key is a digest +(`sha256:`) or the `urn:uuid:` of an ingested SBOM or advisory. +Rejected documents are only reachable by digest, or by filtering the list on +`blocked`. Both endpoints require the `read.validation` permission. + +Reports are *not* stored for `validate_named`, the explicit by-name invocation +used by internal callers: that is an ad-hoc query, not an ingestion. + +To view the logs instead — which is all you get with `persist_reports: false` +— set: ```bash RUST_LOG=trustify_module_ingestor=debug @@ -57,6 +95,15 @@ Each validator declares: `basic`. - `run_on_ingest` — defaults to `true`; set it to `false` to register a validator for explicit invocation from internal code without running it on every ingest. +- `persist` — defaults to `true`; set it to `false` to log this validator's + results instead of storing them. +- `revalidate` — defaults to `always`: a document is validated again even when + it has been seen before. Set it to `on_change` to skip the run when neither + the document nor the validator configuration has changed since the stored + result. Only safe for backends whose verdict depends entirely on inputs we + can fingerprint, which excludes `conforma`: its policy lives on the remote + server and can change without any change here, so `on_change` would serve a + stale verdict. - `backend.url` — for the `conforma` backend, the base URL of the remote `ec validate input --server` instance. The client posts to `/v1/validate/input`; `backend.timeout_seconds` defaults to `120`. The server's @@ -97,6 +144,10 @@ ec validate input --server --server-address 0.0.0.0 --policy policy.yaml Conforma loads the policy at server startup; restart that instance to apply policy changes. +Changing a ruleset file counts as a configuration change: the fingerprint +recorded with each result covers the ruleset contents, not just the path in +`validators.yaml`. + In the provided config, `csaf-spec` runs official CSAF specification tests in `report` mode. `csaf-mandatory` and `cyclonedx-min` run scheck-based checks in `report` mode (observability only) while `spdx-min` runs in diff --git a/etc/validators/validators.yaml b/etc/validators/validators.yaml index fec5545b2d..74bd88afaf 100644 --- a/etc/validators/validators.yaml +++ b/etc/validators/validators.yaml @@ -1,4 +1,4 @@ -# Semantic validators configuration (ADR 00020). +# Semantic validators configuration (ADR 00021). # # Point the server at this file to enable ingestion-time validation: # @@ -10,6 +10,16 @@ # (use absolute paths in production). When no validators are configured, # validation is disabled and ingestion behaves as before. +# Store validation results in the `validation_report` table. Set to false to +# fall back to logging them at debug level, leaving the table untouched. +persist_reports: true + +# Bounds on how much of a report is stored. Findings beyond either limit are +# dropped and the report is flagged as truncated. +caps: + max_findings: 200 + max_findings_bytes: 65536 + validators: # CSAF advisories: official spec validation via csaf-rs. # Profiles: basic (schema + mandatory), extended (+ recommended), @@ -24,6 +34,10 @@ validators: mode: report threshold: error on_error: continue + # Defaults, shown here once: store every result, and never reuse a stored + # verdict for a document we have seen before. + persist: true + revalidate: always # CSAF advisories: additional scheck-based structural checks. - name: csaf-mandatory diff --git a/migration/src/lib.rs b/migration/src/lib.rs index 444dcd0008..d8cfd85a52 100644 --- a/migration/src/lib.rs +++ b/migration/src/lib.rs @@ -82,6 +82,7 @@ mod m0002350_drop_purl_status_gist_index; mod m0002360_pythonver_cmp_parallel_restricted; mod m0002370_importer_quay_auth; mod m0002380_version_function_parallelism; +mod m0002390_create_validation_report; pub trait MigratorExt: Send { fn build_migrations() -> Migrations; @@ -177,6 +178,7 @@ impl MigratorExt for Migrator { .normal(m0002360_pythonver_cmp_parallel_restricted::Migration) .normal(m0002370_importer_quay_auth::Migration) .normal(m0002380_version_function_parallelism::Migration) + .normal(m0002390_create_validation_report::Migration) } } diff --git a/migration/src/m0002390_create_validation_report.rs b/migration/src/m0002390_create_validation_report.rs new file mode 100644 index 0000000000..4a754fd559 --- /dev/null +++ b/migration/src/m0002390_create_validation_report.rs @@ -0,0 +1,239 @@ +use sea_orm_migration::prelude::*; +use sea_query::extension::postgres::Type; +use strum::VariantNames; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let builder = manager.get_connection().get_database_backend(); + + // The `Table` variant of each enum names the PostgreSQL type, so it is + // skipped when emitting the value list. + for stmt in [ + Type::create() + .as_enum(ValidationMode::Table) + .values(ValidationMode::VARIANTS.iter().skip(1).copied()) + .to_owned(), + Type::create() + .as_enum(ValidationOutcome::Table) + .values(ValidationOutcome::VARIANTS.iter().skip(1).copied()) + .to_owned(), + Type::create() + .as_enum(ValidationSeverity::Table) + .values(ValidationSeverity::VARIANTS.iter().skip(1).copied()) + .to_owned(), + Type::create() + .as_enum(ValidationIngestSource::Table) + .values(ValidationIngestSource::VARIANTS.iter().skip(1).copied()) + .to_owned(), + ] { + manager + .get_connection() + .execute_unprepared(&builder.build(&stmt).to_string()) + .await?; + } + + manager + .create_table( + Table::create() + .table(ValidationReport::Table) + .if_not_exists() + .col( + ColumnDef::new(ValidationReport::Id) + .uuid() + .not_null() + .primary_key(), + ) + .col( + ColumnDef::new(ValidationReport::DocumentSha256) + .text() + .not_null(), + ) + .col( + ColumnDef::new(ValidationReport::Validator) + .text() + .not_null(), + ) + .col( + ColumnDef::new(ValidationReport::Mode) + .custom(ValidationMode::Table) + .not_null(), + ) + .col( + ColumnDef::new(ValidationReport::Outcome) + .custom(ValidationOutcome::Table) + .not_null(), + ) + .col( + ColumnDef::new(ValidationReport::Blocked) + .boolean() + .not_null() + .default(false), + ) + .col( + ColumnDef::new(ValidationReport::MaxSeverity) + .custom(ValidationSeverity::Table) + .null(), + ) + .col( + ColumnDef::new(ValidationReport::FindingCount) + .integer() + .not_null(), + ) + .col( + ColumnDef::new(ValidationReport::Findings) + .json_binary() + .not_null(), + ) + .col( + ColumnDef::new(ValidationReport::Truncated) + .boolean() + .not_null() + .default(false), + ) + .col( + ColumnDef::new(ValidationReport::ContentHash) + .text() + .not_null(), + ) + .col( + ColumnDef::new(ValidationReport::ConfigFingerprint) + .text() + .not_null(), + ) + .col( + ColumnDef::new(ValidationReport::IngestSource) + .custom(ValidationIngestSource::Table) + .not_null(), + ) + .col(ColumnDef::new(ValidationReport::ImporterName).text().null()) + .col( + ColumnDef::new(ValidationReport::CreatedAt) + .timestamp_with_time_zone() + .not_null() + .default(Expr::current_timestamp()), + ) + .to_owned(), + ) + .await?; + + // Rows are append-only: a validation run inserts only when its result + // differs from the latest stored one. Reads resolve the current result + // with `DISTINCT ON (document_sha256, validator) ORDER BY created_at + // DESC`, which this index serves; its leading column also serves + // lookups by digest alone. Deliberately no index on `created_at`: the + // table is write-heavy and read paths are nearly always filtered by + // document or by `blocked`. + manager + .get_connection() + .execute_unprepared( + r#" + CREATE INDEX IF NOT EXISTS idx_validation_report_document_validator + ON validation_report (document_sha256, validator, created_at DESC) + "#, + ) + .await?; + + // Rejections are rare, so a partial index stays small and costs a write + // only for the rows it covers. It backs "what did this instance refuse", + // which has no document to filter on. + manager + .get_connection() + .execute_unprepared( + r#" + CREATE INDEX IF NOT EXISTS idx_validation_report_blocked + ON validation_report (created_at DESC) WHERE blocked + "#, + ) + .await?; + + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table( + Table::drop() + .table(ValidationReport::Table) + .if_exists() + .to_owned(), + ) + .await?; + + for name in [ + ValidationIngestSource::Table.into_iden(), + ValidationSeverity::Table.into_iden(), + ValidationOutcome::Table.into_iden(), + ValidationMode::Table.into_iden(), + ] { + manager + .drop_type(Type::drop().if_exists().name(name).to_owned()) + .await?; + } + + Ok(()) + } +} + +#[derive(DeriveIden)] +pub enum ValidationReport { + Table, + Id, + DocumentSha256, + Validator, + Mode, + Outcome, + Blocked, + MaxSeverity, + FindingCount, + Findings, + Truncated, + ContentHash, + ConfigFingerprint, + IngestSource, + ImporterName, + CreatedAt, +} + +#[derive(DeriveIden, strum::VariantNames)] +#[strum(serialize_all = "lowercase")] +#[allow(unused)] +pub enum ValidationMode { + Table, + Report, + Verify, +} + +#[derive(DeriveIden, strum::VariantNames)] +#[strum(serialize_all = "lowercase")] +#[allow(unused)] +pub enum ValidationOutcome { + Table, + Passed, + Failed, +} + +#[derive(DeriveIden, strum::VariantNames)] +#[strum(serialize_all = "lowercase")] +#[allow(unused)] +pub enum ValidationSeverity { + Table, + Info, + Warning, + Error, + Fatal, +} + +#[derive(DeriveIden, strum::VariantNames)] +#[strum(serialize_all = "lowercase")] +#[allow(unused)] +pub enum ValidationIngestSource { + Table, + Api, + Importer, + Dataset, + Internal, +} diff --git a/modules/fundamental/src/endpoints.rs b/modules/fundamental/src/endpoints.rs index cadd25fb54..38f1550bfe 100644 --- a/modules/fundamental/src/endpoints.rs +++ b/modules/fundamental/src/endpoints.rs @@ -5,12 +5,15 @@ use trustify_common::db::{self, pagination_cache::PaginationCache}; use trustify_module_analysis::service::AnalysisService; use trustify_module_ingestor::common; use trustify_module_ingestor::graph::Graph; -use trustify_module_ingestor::service::{IngestorService, validation::Validator}; +use trustify_module_ingestor::service::{ + IngestorService, + validation::{Validator, store::Provenance}, +}; use trustify_module_storage::service::dispatch::DispatchBackend; use utoipa::{IntoParams, ToSchema}; use crate::{ - advisory, crypto, exploit, license, organization, product, purl, sbom, sbom_group, + advisory, crypto, exploit, license, organization, product, purl, sbom, sbom_group, validation, vulnerability, weakness, }; @@ -56,8 +59,10 @@ pub fn configure( graph: Graph, validators: Vec>, ) { - let ingestor_service = - IngestorService::new(graph, storage, Some(analysis)).with_validators(validators); + let ingestor_service = IngestorService::new(graph, storage, Some(analysis)) + .with_validators(validators) + .with_provenance(Provenance::Api) + .with_report_store(db_rw.clone()); svc.app_data(web::Data::new(ingestor_service.clone())); advisory::endpoints::configure( @@ -86,6 +91,7 @@ pub fn configure( config.sbom_upload_limit, cache.clone(), ); + validation::endpoints::configure(svc, db_ro.clone(), cache.clone()); vulnerability::endpoints::configure(svc, db_ro.clone(), cache.clone()); weakness::endpoints::configure(svc, db_ro.clone(), cache.clone()); sbom_group::endpoints::configure(svc, db_rw, db_ro, config.max_group_name_length, cache); diff --git a/modules/fundamental/src/lib.rs b/modules/fundamental/src/lib.rs index b4eadf0f53..07c790b18b 100644 --- a/modules/fundamental/src/lib.rs +++ b/modules/fundamental/src/lib.rs @@ -12,6 +12,7 @@ pub mod product; pub mod purl; pub mod sbom; pub mod source_document; +pub mod validation; #[allow(deprecated)] pub mod vulnerability; pub mod weakness; diff --git a/modules/fundamental/src/validation/endpoints/mod.rs b/modules/fundamental/src/validation/endpoints/mod.rs new file mode 100644 index 0000000000..a6b2986747 --- /dev/null +++ b/modules/fundamental/src/validation/endpoints/mod.rs @@ -0,0 +1,89 @@ +use crate::validation::{model::ValidationReportSummary, service::ValidationService}; +use actix_web::{HttpResponse, Responder, get, web}; +use trustify_auth::{ReadValidation, authorizer::Require}; +use trustify_common::{ + db::{self, pagination_cache::PaginationCache, query::Query}, + id::Id, + model::{Paginated, PaginatedResults}, +}; +use utoipa::IntoParams; + +pub fn configure( + config: &mut utoipa_actix_web::service_config::ServiceConfig, + db: db::ReadOnly, + cache: PaginationCache, +) { + let service = ValidationService::new(cache); + + config + .app_data(web::Data::new(db)) + .app_data(web::Data::new(service)) + .service(list_validation_reports) + .service(get_validation_reports); +} + +/// Filters that cannot be expressed as columns of the report itself. +#[derive(Clone, Debug, Default, IntoParams, serde::Deserialize)] +pub struct DocumentFilter { + /// Name of the document: an SBOM's describing node, or an advisory + /// identifier. Neither is unique, so this can match several documents. + pub name: Option, +} + +#[utoipa::path( + tag = "validation", + operation_id = "listValidationReports", + params( + DocumentFilter, + Query, + Paginated, + ), + responses( + (status = 200, description = "Matching validation reports", body = PaginatedResults), + ), +)] +#[get("/v3/validation")] +/// List validation reports +pub async fn list_validation_reports( + state: web::Data, + db: web::Data, + web::Query(filter): web::Query, + web::Query(search): web::Query, + web::Query(paginated): web::Query, + _: Require, +) -> actix_web::Result { + let tx = db.begin().await?; + + Ok(HttpResponse::Ok().json(match &filter.name { + Some(name) => state.get_by_name(name, paginated, &tx).await?, + None => state.list(search, paginated, &tx).await?, + })) +} + +#[utoipa::path( + tag = "validation", + operation_id = "getValidationReports", + params( + ("key" = Id, Path, description = "Identifier of the document, either a digest e.g. `sha256:` or `urn:uuid:` of an ingested SBOM or advisory"), + ), + responses( + (status = 200, description = "Validation reports for the document", body = Vec), + (status = 400, description = "The key could not be interpreted as a document identifier"), + ), +)] +#[get("/v3/validation/{key}")] +/// Retrieve the validation reports of one document +pub async fn get_validation_reports( + state: web::Data, + db: web::Data, + key: web::Path, + _: Require, +) -> actix_web::Result { + let tx = db.begin().await?; + let id = key.parse::().map_err(crate::Error::from)?; + + Ok(HttpResponse::Ok().json(state.get(id, &tx).await?)) +} + +#[cfg(test)] +mod test; diff --git a/modules/fundamental/src/validation/endpoints/test.rs b/modules/fundamental/src/validation/endpoints/test.rs new file mode 100644 index 0000000000..5f4a1978c3 --- /dev/null +++ b/modules/fundamental/src/validation/endpoints/test.rs @@ -0,0 +1,141 @@ +use crate::{test::caller, validation::model::ValidationReportSummary}; +use actix_web::test::TestRequest; +use sea_orm::{ActiveValue::Set, EntityTrait}; +use test_context::test_context; +use test_log::test; +use time::OffsetDateTime; +use trustify_common::id::Id; +use trustify_common::model::PaginatedResults; +use trustify_entity::validation_report; +use trustify_module_ingestor::service::validation::store::digest_for_document; +use trustify_test_context::{TrustifyContext, call::CallService}; +use uuid::Uuid; + +/// Insert a report directly: the endpoints read what ingestion recorded, and +/// the recording itself is covered by the ingestor's own tests. +async fn record( + ctx: &TrustifyContext, + document_sha256: &str, + validator: &str, + blocked: bool, +) -> Result<(), anyhow::Error> { + validation_report::Entity::insert(validation_report::ActiveModel { + id: Set(Uuid::now_v7()), + document_sha256: Set(document_sha256.to_string()), + validator: Set(validator.to_string()), + mode: Set(validation_report::Mode::Verify), + outcome: Set(validation_report::Outcome::Failed), + blocked: Set(blocked), + max_severity: Set(Some(validation_report::Severity::Error)), + finding_count: Set(1), + findings: Set(serde_json::json!([{ "severity": "error", "message": "nope" }])), + truncated: Set(false), + content_hash: Set(format!("hash-{validator}")), + config_fingerprint: Set("fingerprint".to_string()), + ingest_source: Set(validation_report::IngestSource::Api), + importer_name: Set(None), + created_at: Set(OffsetDateTime::now_utc()), + }) + .exec(&ctx.db) + .await?; + + Ok(()) +} + +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn list_validation_reports(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + record(ctx, &"a".repeat(64), "scheck", false).await?; + record(ctx, &"b".repeat(64), "csaf-spec", true).await?; + + let app = caller(ctx).await?; + let request = TestRequest::get() + .uri("/api/v3/validation?total=true") + .to_request(); + let response: PaginatedResults = + app.call_and_read_body_json(request).await; + + assert_eq!(response.total, Some(2)); + + // A rejected document is only visible here: it has no SBOM or advisory to + // be listed under. + let request = TestRequest::get() + .uri("/api/v3/validation?q=blocked%3Dtrue") + .to_request(); + let response: PaginatedResults = + app.call_and_read_body_json(request).await; + + assert_eq!(response.items.len(), 1); + assert_eq!(response.items[0].validator, "csaf-spec"); + + Ok(()) +} + +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn get_validation_reports_by_digest(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + let digest = "c".repeat(64); + record(ctx, &digest, "scheck", false).await?; + + let app = caller(ctx).await?; + let request = TestRequest::get() + .uri(&format!("/api/v3/validation/sha256%3A{digest}")) + .to_request(); + let response: Vec = app.call_and_read_body_json(request).await; + + assert_eq!(response.len(), 1); + assert_eq!(response[0].document_sha256, digest); + assert!(!response[0].blocked); + + Ok(()) +} + +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn get_validation_reports_by_sbom_id(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + let result = ctx + .ingest_document("zookeeper-3.9.2-cyclonedx.json") + .await?; + let digest = digest_for_document(&ctx.db, Id::Uuid(Uuid::parse_str(&result.id)?)) + .await? + .expect("ingested document has a source document"); + + record(ctx, &digest, "scheck", false).await?; + + let app = caller(ctx).await?; + let request = TestRequest::get() + .uri(&format!("/api/v3/validation/urn:uuid:{}", result.id)) + .to_request(); + let response: Vec = app.call_and_read_body_json(request).await; + + assert_eq!(response.len(), 1); + assert_eq!(response[0].document_sha256, digest); + + Ok(()) +} + +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn list_validation_reports_by_name(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + let result = ctx + .ingest_document("zookeeper-3.9.2-cyclonedx.json") + .await?; + let digest = digest_for_document(&ctx.db, Id::Uuid(Uuid::parse_str(&result.id)?)) + .await? + .expect("ingested document has a source document"); + + record(ctx, &digest, "scheck", false).await?; + record(ctx, &"d".repeat(64), "scheck", false).await?; + + let app = caller(ctx).await?; + let request = TestRequest::get() + .uri("/api/v3/validation?name=zookeeper") + .to_request(); + let response: PaginatedResults = + app.call_and_read_body_json(request).await; + + assert_eq!(response.items.len(), 1); + assert_eq!(response.items[0].document_sha256, digest); + + Ok(()) +} diff --git a/modules/fundamental/src/validation/mod.rs b/modules/fundamental/src/validation/mod.rs new file mode 100644 index 0000000000..cfd03e81e1 --- /dev/null +++ b/modules/fundamental/src/validation/mod.rs @@ -0,0 +1,5 @@ +//! Read access to the semantic validation reports recorded during ingestion. + +pub mod endpoints; +pub mod model; +pub mod service; diff --git a/modules/fundamental/src/validation/model/mod.rs b/modules/fundamental/src/validation/model/mod.rs new file mode 100644 index 0000000000..2e44e7b889 --- /dev/null +++ b/modules/fundamental/src/validation/model/mod.rs @@ -0,0 +1,60 @@ +use trustify_entity::validation_report; +use utoipa::ToSchema; + +/// A stored result of one semantic validator run against one document. +/// +/// Reports are append-only, so several may exist for the same document and +/// validator: each one records a result that differed from the one before it. +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, ToSchema)] +pub struct ValidationReportSummary { + /// SHA-256 of the validated document, as lowercase hex. + pub document_sha256: String, + /// Name of the validator that produced the result. + pub validator: String, + /// Whether the validator gated ingestion or only reported. + #[schema(value_type = String)] + pub mode: validation_report::Mode, + /// Whether the document met the validator's threshold. + #[schema(value_type = String)] + pub outcome: validation_report::Outcome, + /// True when this result rejected the document, so nothing was ingested. + pub blocked: bool, + /// Highest severity across the findings; absent when there are none. + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schema(value_type = Option)] + pub max_severity: Option, + /// Number of findings produced, before truncation. + pub finding_count: i32, + /// The findings, as stored. Fewer than `finding_count` when `truncated`. + pub findings: serde_json::Value, + /// True when the stored findings were capped. + pub truncated: bool, + /// How the document entered the system. + #[schema(value_type = String)] + pub ingest_source: validation_report::IngestSource, + /// Name of the importer, when ingested by one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub importer_name: Option, + #[serde(with = "time::serde::rfc3339")] + #[schema(value_type = String, format = DateTime)] + pub created_at: time::OffsetDateTime, +} + +impl From for ValidationReportSummary { + fn from(value: validation_report::Model) -> Self { + Self { + document_sha256: value.document_sha256, + validator: value.validator, + mode: value.mode, + outcome: value.outcome, + blocked: value.blocked, + max_severity: value.max_severity, + finding_count: value.finding_count, + findings: value.findings, + truncated: value.truncated, + ingest_source: value.ingest_source, + importer_name: value.importer_name, + created_at: value.created_at, + } + } +} diff --git a/modules/fundamental/src/validation/service/mod.rs b/modules/fundamental/src/validation/service/mod.rs new file mode 100644 index 0000000000..0248ea1cbd --- /dev/null +++ b/modules/fundamental/src/validation/service/mod.rs @@ -0,0 +1,91 @@ +use crate::{Error, validation::model::ValidationReportSummary}; +use sea_orm::{ConnectionTrait, EntityTrait, QueryOrder}; +use trustify_common::{ + db::{ + limiter::{LimitedResult, LimiterTrait}, + pagination_cache::PaginationCache, + query::{Filtering, Query}, + }, + id::Id, + model::{PaginatedResults, Pagination}, +}; +use trustify_entity::validation_report; +use trustify_module_ingestor::service::validation::store; + +/// Read access to stored validation reports. +pub struct ValidationService { + cache: PaginationCache, +} + +impl ValidationService { + /// Creates a new validation report service. + pub fn new(cache: PaginationCache) -> Self { + Self { cache } + } + + /// Lists reports matching the given query, newest first. + /// + /// Includes reports for documents that were rejected, which exist nowhere + /// else: they have no document resource to be listed under. + pub async fn list( + &self, + query: Query, + paginated: impl Pagination, + connection: &C, + ) -> Result, Error> { + let limiter = validation_report::Entity::find() + .filtering(query)? + .order_by_desc(validation_report::Column::CreatedAt) + .limiting(connection, paginated, &self.cache)?; + + let LimitedResult { items, total } = limiter.fetch().await?; + let total = total.requested(paginated.total()).await?; + + Ok(PaginatedResults { + items: items.into_iter().map(Into::into).collect(), + total, + }) + } + + /// Every report for one document, newest first. + /// + /// Accepts an internal UUID or any supported digest. Only a SHA-256 + /// matches a rejected document: everything else resolves through the + /// ingested document it belongs to. + pub async fn get( + &self, + id: Id, + connection: &C, + ) -> Result, Error> { + let reports = store::by_document_id(id) + .map_err(|err| Error::bad_request(err.to_string(), None::))? + .order_by_desc(validation_report::Column::CreatedAt) + .all(connection) + .await?; + + Ok(reports.into_iter().map(Into::into).collect()) + } + + /// Every report for documents with the given name, newest first. + /// + /// Matches the name of an SBOM's describing node and the advisory + /// identifier, neither of which is unique. + pub async fn get_by_name( + &self, + name: &str, + paginated: impl Pagination, + connection: &C, + ) -> Result, Error> { + let limiter = store::by_document_name(name) + .order_by_desc(validation_report::Column::CreatedAt) + .limiting(connection, paginated, &self.cache)?; + + let LimitedResult { items, total } = limiter.fetch().await?; + let total = total.requested(paginated.total()).await?; + + Ok(PaginatedResults { + items: items.into_iter().map(Into::into).collect(), + total, + }) + } +} diff --git a/modules/ingestor/Cargo.toml b/modules/ingestor/Cargo.toml index 86251030b5..652f7e02b3 100644 --- a/modules/ingestor/Cargo.toml +++ b/modules/ingestor/Cargo.toml @@ -9,7 +9,7 @@ rust-version.workspace = true [features] default = ["semantic-validation"] -# The semantic validator subsystem (ADR 00020): scheck, CSAF spec and Conforma +# The semantic validator subsystem (ADR 00021): scheck, CSAF spec and Conforma # backends. Disabling it drops the `scheck` dependency tree; the validator # config still parses, but building a configured validator fails. See ADR 00022. semantic-validation = ["dep:scheck"] @@ -43,6 +43,7 @@ roxmltree = { workspace = true } sbom-walker = { workspace = true } scheck = { workspace = true, optional = true } sea-orm = { workspace = true } +sha2 = { workspace = true } sea-query = { workspace = true } semver = { workspace = true } serde = { workspace = true, features = ["derive"] } diff --git a/modules/ingestor/src/endpoints.rs b/modules/ingestor/src/endpoints.rs index f5cb339399..92187fc0d6 100644 --- a/modules/ingestor/src/endpoints.rs +++ b/modules/ingestor/src/endpoints.rs @@ -1,6 +1,6 @@ use crate::{ graph::Graph, - service::{Error, IngestorService, validation::Validator}, + service::{Error, IngestorService, validation::Validator, validation::store::Provenance}, }; use actix_web::{HttpResponse, Responder, post, web}; use sea_orm::TransactionTrait; @@ -21,8 +21,10 @@ pub fn configure( analysis: Option, validators: Vec>, ) { - let ingestor_service = - IngestorService::new(Graph::new(), storage, analysis).with_validators(validators); + let ingestor_service = IngestorService::new(Graph::new(), storage, analysis) + .with_validators(validators) + .with_provenance(Provenance::Dataset) + .with_report_store(db.clone()); svc.app_data(web::Data::new(ingestor_service)) .app_data(web::Data::new(config)) diff --git a/modules/ingestor/src/model.rs b/modules/ingestor/src/model.rs index 9c49d57b85..4f86446339 100644 --- a/modules/ingestor/src/model.rs +++ b/modules/ingestor/src/model.rs @@ -17,7 +17,7 @@ pub struct IngestResult { pub warnings: Vec, /// Structured reports from semantic validators run during ingestion. /// - /// See ADR 00020. Empty when no validators are configured or none apply + /// See ADR 00021. Empty when no validators are configured or none apply /// to the document's format. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub validation: Vec, diff --git a/modules/ingestor/src/service/mod.rs b/modules/ingestor/src/service/mod.rs index cb1f316dd5..a5c5262e4b 100644 --- a/modules/ingestor/src/service/mod.rs +++ b/modules/ingestor/src/service/mod.rs @@ -20,12 +20,13 @@ use crate::{ dataset::{DatasetIngestResult, DatasetLoader}, validation::{ Finding, InvocationError, OnError, Severity, ValidationMode, ValidationOutcome, - ValidationReport, Validator, ValidatorInput, + ValidationReport, Validator, ValidatorInput, store, store::Provenance, }, }, }; use actix_web::{HttpResponse, ResponseError, body::BoxBody}; use anyhow::anyhow; +use hex::ToHex; use jsonpath_rust::parser::errors::JsonPathError; use parking_lot::Mutex; use sbom_walker::report::ReportSink; @@ -34,8 +35,15 @@ use sea_orm::{ConnectionTrait, TransactionTrait}; use std::{fmt::Debug, sync::Arc, time::Instant}; use tokio::task::JoinError; use tracing::instrument; -use trustify_common::db::change::{ChangeEntity, ChangeOperation, record_change}; -use trustify_common::{db::DatabaseErrors, error::ErrorInformation, id::IdError}; +use trustify_common::{ + db::{ + DatabaseErrors, ReadWrite, + change::{ChangeEntity, ChangeOperation, record_change}, + }, + error::ErrorInformation, + hashing::Contexts, + id::IdError, +}; use trustify_entity::labels::Labels; use trustify_module_analysis::service::AnalysisService; use trustify_module_storage::service::{StorageBackend, dispatch::DispatchBackend}; @@ -220,6 +228,10 @@ pub struct IngestorService { storage: DispatchBackend, analysis: Option, validators: Arc<[Arc]>, + provenance: Arc, + /// Connection used to record rejections, which must outlive the caller's + /// rolled-back transaction. + reports: Option, } impl IngestorService { @@ -233,18 +245,36 @@ impl IngestorService { storage: storage.into(), analysis, validators: Vec::new().into(), + provenance: Arc::new(Provenance::default()), + reports: None, } } /// Attach the registered validators available to ingestion and internal callers. /// /// With an empty set (the default), ingestion behaves as if validation did - /// not exist. See ADR 00020. + /// not exist. See ADR 00021. pub fn with_validators(mut self, validators: Vec>) -> Self { self.validators = validators.into(); self } + /// Declare how documents reach this instance, recorded with every report. + pub fn with_provenance(mut self, provenance: Provenance) -> Self { + self.provenance = Arc::new(provenance); + self + } + + /// Attach the connection used to record documents rejected by validation. + /// + /// A rejection is refused ingestion, so the caller's transaction is rolled + /// back and anything written on it is lost. Without this connection, a + /// rejection is only logged. + pub fn with_report_store(mut self, db: ReadWrite) -> Self { + self.reports = Some(db); + self + } + pub fn storage(&self) -> &DispatchBackend { &self.storage } @@ -261,19 +291,91 @@ impl IngestorService { } /// Run all applicable validators against a document. + #[instrument(skip(self, bytes), fields(bytes = bytes.len()))] + async fn run_validators(&self, bytes: &[u8], fmt: Format) -> Vec { + run_validators(&self.validators, bytes, fmt).await + } + + /// Record the results that asked to be stored. /// - /// Returns the collected reports on success. Returns - /// [`Error::ValidationRejected`] if any [`ValidationMode::Verify`] validator - /// blocks the document — either because its outcome is - /// [`ValidationOutcome::Failed`], or because it errored while configured to - /// [`OnError::Block`]. - #[instrument(skip(self, bytes), fields(bytes = bytes.len()), err(level = tracing::Level::INFO))] - async fn run_validators( + /// Never fails the caller: the document is already ingested, so losing the + /// audit trail is worth a warning, not a rolled-back ingest. + async fn store_reports( &self, - bytes: &[u8], - fmt: Format, - ) -> Result, Error> { - run_validators(&self.validators, bytes, fmt).await + conn: &C, + document_sha256: &str, + validated: &[Validated], + ) { + let pending = validated + .iter() + .filter(|entry| entry.persist) + .map(|entry| store::PendingReport { + report: &entry.report, + mode: entry.mode, + blocked: entry.blocked, + config_fingerprint: &entry.fingerprint, + }) + .collect::>(); + + for entry in validated.iter().filter(|entry| !entry.persist) { + tracing::debug!( + validator = %entry.report.validator, + outcome = ?entry.report.outcome, + "validation report not persisted" + ); + } + + if pending.is_empty() { + return; + } + + let context = store::PersistContext::new( + document_sha256, + &self.provenance, + self.caps().unwrap_or_default(), + ); + match store::persist_if_changed(conn, &context, pending).await { + Ok(0) => tracing::debug!("validation reports unchanged"), + Ok(count) => tracing::debug!("recorded {count} validation report(s)"), + Err(err) => tracing::warn!("failed to record validation reports: {err}"), + } + } + + /// Record a document that validation refused. + /// + /// Runs on its own connection: the caller rolls back on the error returned + /// here, which would take the record with it. Failing to record a rejection + /// must not change the rejection itself, so errors are logged only. + async fn record_rejection(&self, bytes: &[u8], fmt: Format, validated: &[Validated]) { + let Some(db) = &self.reports else { + tracing::debug!( + format = %fmt, + "document rejected by validation; not recorded (no report store configured)" + ); + return; + }; + + // The document never reaches storage, so nothing has hashed it yet. + let mut contexts = Contexts::new(); + contexts.update(bytes); + let document_sha256 = contexts.finish().sha256.encode_hex::(); + + let result = db + .transaction(async |tx| { + self.store_reports(tx, &document_sha256, validated).await; + Ok::<_, DbErr>(()) + }) + .await; + if let Err(err) = result { + tracing::warn!("failed to record rejected document: {err}"); + } + } + + /// The caps configured for the validators, which are global. + fn caps(&self) -> Option { + self.validators + .first() + .map(|validator| validator.persistence().caps) } #[instrument(skip_all, err(level=tracing::Level::INFO))] @@ -293,20 +395,34 @@ impl IngestorService { // Run semantic validators before persisting anything. A blocking // (verify) failure returns an error here, so no bytes are stored and no - // graph rows are created. See ADR 00020. - let reports = self.run_validators(bytes, fmt).await?; + // graph rows are created. See ADR 00021. + let validated = self.run_validators(bytes, fmt).await; + + let blocked = blocking_reports(&validated); + if !blocked.is_empty() { + self.record_rejection(bytes, fmt, &validated).await; + return Err(Error::ValidationRejected(blocked)); + } - let result = self + let stored = self .storage .store(bytes) .await .map_err(|err| Error::Storage(anyhow!("{err}")))?; + let document_sha256 = stored.digests.sha256.encode_hex::(); let mut result = detector - .load(&self.graph, labels.into(), issuer, &result.digests, tx) + .load(&self.graph, labels.into(), issuer, &stored.digests, tx) .await?; - attach_validation(&mut result, reports); + // On the caller's transaction: the reports describe a document that is + // only ingested if that transaction commits. + self.store_reports(tx, &document_sha256, &validated).await; + + attach_validation( + &mut result, + validated.into_iter().map(|entry| entry.report).collect(), + ); let change_entity = match fmt { Format::CSAF | Format::CVE | Format::OSV => Some(ChangeEntity::Advisory), @@ -383,37 +499,54 @@ impl IngestorService { } } +/// The outcome of one validator, with what the store needs to record it. +#[derive(Debug)] +struct Validated { + report: ValidationReport, + mode: ValidationMode, + /// True when this result rejects the document. + blocked: bool, + /// Digest of the configuration that produced the result. + fingerprint: String, + /// Whether the result is stored at all. + persist: bool, +} + /// Run all applicable validators against a document. /// -/// Returns the collected reports, or [`Error::ValidationRejected`] if a -/// [`ValidationMode::Verify`] validator blocks the document. See ADR 00020. +/// Returns one entry per validator that ran. Blocking is reported per entry +/// rather than as an error, because a rejected document still has to be +/// recorded before ingestion is refused. See ADR 00021. async fn run_validators( validators: &[Arc], bytes: &[u8], fmt: Format, -) -> Result, Error> { +) -> Vec { if validators.is_empty() { - return Ok(Vec::new()); + return Vec::new(); } let input = ValidatorInput { bytes, format: fmt }; - let mut reports = Vec::new(); - let mut blocked = Vec::new(); + let mut results = Vec::new(); for validator in validators { if !validator.run_on_ingest() || !validator.applies_to(fmt) { continue; } + let persistence = validator.persistence(); match validator.validate(&input).await { Ok(report) => { log_report(fmt, &report); - if validator.mode() == ValidationMode::Verify - && report.outcome == ValidationOutcome::Failed - { - blocked.push(report.clone()); - } - reports.push(report); + let blocked = validator.mode() == ValidationMode::Verify + && report.outcome == ValidationOutcome::Failed; + results.push(Validated { + report, + mode: validator.mode(), + blocked, + fingerprint: persistence.fingerprint.clone(), + persist: persistence.persist, + }); } Err(err) => match (validator.mode(), validator.on_error()) { (ValidationMode::Verify, OnError::Block) => { @@ -421,7 +554,13 @@ async fn run_validators( validator = validator.name(), "verify validator errored, blocking ingestion: {err}" ); - blocked.push(errored_report(validator.name(), &err)); + results.push(Validated { + report: errored_report(validator.name(), &err), + mode: ValidationMode::Verify, + blocked: true, + fingerprint: persistence.fingerprint.clone(), + persist: persistence.persist, + }); } _ => { tracing::warn!( @@ -433,22 +572,16 @@ async fn run_validators( } } - if blocked.is_empty() { - Ok(reports) - } else { - let validators = blocked - .iter() - .map(|report| report.validator.as_str()) - .collect::>() - .join(", "); - tracing::warn!( - format = %fmt, - validators = %validators, - "document rejected by validation ({} report(s))", - blocked.len(), - ); - Err(Error::ValidationRejected(blocked)) - } + results +} + +/// The reports that rejected the document, if any. +fn blocking_reports(validated: &[Validated]) -> Vec { + validated + .iter() + .filter(|entry| entry.blocked) + .map(|entry| entry.report.clone()) + .collect() } /// Log a validation report and its findings. @@ -560,8 +693,8 @@ impl ReportSink for Discard { mod validation_tests { use super::*; use crate::service::validation::{ - Finding, OnError, Severity, ValidationMode, ValidationOutcome, ValidationReport, Validator, - ValidatorError, ValidatorInput, + Finding, OnError, Persistence, Severity, ValidationMode, ValidationOutcome, + ValidationReport, Validator, ValidatorError, ValidatorInput, }; use sea_orm::prelude::async_trait; @@ -574,6 +707,7 @@ mod validation_tests { applies: bool, run_on_ingest: bool, result: MockResult, + persistence: Persistence, } #[derive(Debug, Clone, Copy)] @@ -592,6 +726,7 @@ mod validation_tests { applies: true, run_on_ingest: true, result, + persistence: Persistence::default(), } } @@ -631,6 +766,9 @@ mod validation_tests { fn applies_to(&self, _format: Format) -> bool { self.applies } + fn persistence(&self) -> &Persistence { + &self.persistence + } async fn validate( &self, _input: &ValidatorInput<'_>, @@ -654,10 +792,17 @@ mod validation_tests { } } + /// Mirror how `ingest` turns validator results into an ingest outcome. fn run(validators: Vec>) -> Result, Error> { - tokio::runtime::Runtime::new() + let validated = tokio::runtime::Runtime::new() .expect("runtime") - .block_on(run_validators(&validators, b"{}", Format::CSAF)) + .block_on(run_validators(&validators, b"{}", Format::CSAF)); + + let blocked = blocking_reports(&validated); + match blocked.is_empty() { + true => Ok(validated.into_iter().map(|entry| entry.report).collect()), + false => Err(Error::ValidationRejected(blocked)), + } } #[test] @@ -748,8 +893,8 @@ mod validation_tests { assert!( run_validators(&validators, b"{}", Format::CSAF) .await - .expect("skipped during ingestion") - .is_empty() + .is_empty(), + "validator must be skipped during ingestion" ); let input = ValidatorInput { @@ -762,3 +907,198 @@ mod validation_tests { assert_eq!(report.outcome, ValidationOutcome::Failed); } } + +#[cfg(test)] +mod persistence_tests { + use super::*; + use crate::service::validation::{ + Persistence, Severity, ValidationMode, ValidationOutcome, ValidationReport, Validator, + ValidatorError, ValidatorInput, + store::{Provenance, StoredReport}, + }; + use sea_orm::{EntityTrait, prelude::async_trait}; + use test_context::test_context; + use test_log::test; + use trustify_entity::validation_report; + use trustify_test_context::{TrustifyContext, document_bytes}; + + /// A validator with a fixed verdict, used to exercise the ingest paths. + #[derive(Debug)] + struct FixedValidator { + mode: ValidationMode, + outcome: ValidationOutcome, + persistence: Persistence, + } + + impl FixedValidator { + fn new(mode: ValidationMode, outcome: ValidationOutcome) -> Self { + Self { + mode, + outcome, + persistence: Persistence::default(), + } + } + + fn persist(mut self, persist: bool) -> Self { + self.persistence.persist = persist; + self + } + } + + #[async_trait::async_trait] + impl Validator for FixedValidator { + fn name(&self) -> &str { + "fixed" + } + fn mode(&self) -> ValidationMode { + self.mode + } + fn threshold(&self) -> Severity { + Severity::Error + } + fn on_error(&self) -> OnError { + OnError::Block + } + fn applies_to(&self, _format: Format) -> bool { + true + } + fn persistence(&self) -> &Persistence { + &self.persistence + } + async fn validate( + &self, + _input: &ValidatorInput<'_>, + ) -> Result { + Ok(ValidationReport { + validator: self.name().to_string(), + findings: vec![], + outcome: self.outcome, + }) + } + } + + async fn ingest( + ctx: &TrustifyContext, + validator: FixedValidator, + ) -> Result, anyhow::Error> { + let ingestor = IngestorService::new(Graph::new(), ctx.storage.clone(), None) + .with_validators(vec![Arc::new(validator)]) + .with_provenance(Provenance::Api) + .with_report_store(trustify_common::db::ReadWrite::new(ctx.db.clone())); + + let bytes = document_bytes("zookeeper-3.9.2-cyclonedx.json").await?; + Ok(ctx + .db + .transaction(async |tx| { + ingestor + .ingest( + &bytes, + Format::Unknown, + ("source", "test"), + None, + Cache::Skip, + tx, + ) + .await + }) + .await) + } + + async fn stored(ctx: &TrustifyContext) -> Result, anyhow::Error> { + Ok(validation_report::Entity::find().all(&ctx.db).await?) + } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn report_mode_is_recorded(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + ingest( + ctx, + FixedValidator::new(ValidationMode::Report, ValidationOutcome::Failed), + ) + .await? + .expect("report mode must not block"); + + let reports = stored(ctx).await?; + assert_eq!(reports.len(), 1); + assert_eq!(reports[0].mode, validation_report::Mode::Report); + assert!(!reports[0].blocked); + assert_eq!( + reports[0].ingest_source, + validation_report::IngestSource::Api + ); + + Ok(()) + } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn verify_pass_is_recorded(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + ingest( + ctx, + FixedValidator::new(ValidationMode::Verify, ValidationOutcome::Passed), + ) + .await? + .expect("verify + passed must not block"); + + let reports = stored(ctx).await?; + assert_eq!(reports.len(), 1); + assert_eq!(reports[0].mode, validation_report::Mode::Verify); + assert!(!reports[0].blocked); + + Ok(()) + } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn rejection_survives_rollback(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + let result = ingest( + ctx, + FixedValidator::new(ValidationMode::Verify, ValidationOutcome::Failed), + ) + .await?; + assert!(matches!(result, Err(Error::ValidationRejected(_)))); + + // The ingest transaction rolled back with the rejection, so the record + // can only be here if it was written on its own connection. + let reports = stored(ctx).await?; + assert_eq!(reports.len(), 1); + assert!(reports[0].blocked); + assert_eq!(reports[0].outcome, validation_report::Outcome::Failed); + + Ok(()) + } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn persist_disabled_writes_nothing(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + ingest( + ctx, + FixedValidator::new(ValidationMode::Report, ValidationOutcome::Passed).persist(false), + ) + .await? + .expect("ingest"); + + assert!(stored(ctx).await?.is_empty()); + + Ok(()) + } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn repeated_ingest_does_not_append(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + for _ in 0..3 { + ingest( + ctx, + FixedValidator::new(ValidationMode::Report, ValidationOutcome::Passed), + ) + .await? + .expect("ingest"); + } + + // Re-ingesting an unchanged document is the dominant importer case: it + // must not grow the table. + assert_eq!(stored(ctx).await?.len(), 1); + + Ok(()) + } +} diff --git a/modules/ingestor/src/service/validation/config.rs b/modules/ingestor/src/service/validation/config.rs index d64f225f43..330a0db006 100644 --- a/modules/ingestor/src/service/validation/config.rs +++ b/modules/ingestor/src/service/validation/config.rs @@ -1,24 +1,129 @@ //! Configuration for semantic validators and construction of the validator set. //! //! The default configuration is empty, which disables validation entirely and -//! preserves the pre-existing ingestion behaviour. See ADR 00020. +//! preserves the pre-existing ingestion behaviour. See ADR 00021. #[cfg(feature = "semantic-validation")] use crate::service::validation::{ScheckValidator, conforma, csaf, scheck}; use crate::service::{ Format, - validation::{OnError, Severity, ValidationMode, Validator}, + validation::{OnError, Severity, ValidationMode, Validator, store::Caps}, }; #[cfg(feature = "semantic-validation")] use anyhow::Context; use std::{collections::HashSet, path::PathBuf, sync::Arc}; /// Configuration for the complete set of validators. -#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct ValidatorsConfig { /// The validators to run, in order. #[serde(default)] pub validators: Vec, + /// Whether results are stored. When false, every validator falls back to + /// logging its result at `debug` regardless of its own `persist` setting. + #[serde(default = "default_persist")] + pub persist_reports: bool, + /// Bounds on how much of each report is stored. + #[serde(default)] + pub caps: Caps, +} + +impl Default for ValidatorsConfig { + fn default() -> Self { + Self { + validators: Vec::new(), + persist_reports: default_persist(), + caps: Caps::default(), + } + } +} + +/// When a validator re-runs against a document it has already seen. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "snake_case")] +pub enum Revalidate { + /// Always run the validator, even for an unchanged document. + #[default] + Always, + /// Skip the run when the document and the validator configuration are + /// unchanged since the stored result. + /// + /// Only safe for backends whose verdict depends entirely on inputs we can + /// fingerprint. A remote backend's policy can change without any change + /// here, so this would serve a stale verdict. + OnChange, +} + +/// What a validator contributes to the stored report, and when. +/// +/// Derived from configuration once, at build time, so the ingest path does not +/// have to consult the configuration again. +#[derive(Clone, Debug)] +pub struct Persistence { + /// Whether results from this validator are stored. + pub persist: bool, + /// Whether an unchanged document is validated again. + pub revalidate: Revalidate, + /// Bounds on how much of the report is stored. + pub caps: Caps, + /// Digest of the configuration that produces this validator's verdicts. + pub fingerprint: String, +} + +impl Default for Persistence { + /// Store with the default caps, under an unknown configuration. + fn default() -> Self { + Self { + persist: default_persist(), + revalidate: Revalidate::default(), + caps: Caps::default(), + fingerprint: String::new(), + } + } +} + +impl Persistence { + /// Derive the persistence settings for one validator. + /// + /// `backend_inputs` carries the bytes that identify the backend's own + /// inputs — ruleset contents for scheck, for instance — which the + /// serialized configuration does not capture. Changing any of them yields + /// a new fingerprint, so the next run records a new row rather than + /// claiming the old verdict came from the new configuration. + pub fn new( + global: &ValidatorsConfig, + config: &ValidatorConfig, + backend_inputs: &[&[u8]], + ) -> Result { + use sha2::{Digest, Sha256}; + + let mut hasher = Sha256::new(); + // Only what can change a verdict: the persistence settings themselves + // are deliberately excluded, so toggling them does not look like a new + // result. + for part in [ + config.name.as_bytes(), + serde_json::to_string(&config.backend)?.as_bytes(), + serde_json::to_string(&config.formats)?.as_bytes(), + serde_json::to_string(&config.mode)?.as_bytes(), + serde_json::to_string(&config.threshold)?.as_bytes(), + serde_json::to_string(&config.on_error)?.as_bytes(), + ] { + hasher.update(part); + hasher.update(b"\0"); + } + for input in backend_inputs { + hasher.update(input); + hasher.update(b"\0"); + } + + Ok(Self { + persist: global.persist_reports && config.persist, + revalidate: config.revalidate, + caps: global.caps, + fingerprint: hex::encode(hasher.finalize()), + }) + } } /// Which backend implements a validator and its backend-specific settings. @@ -77,6 +182,12 @@ pub struct ValidatorConfig { /// For verify mode, the behaviour when the validator itself errors. #[serde(default)] pub on_error: OnError, + /// Whether results from this validator are stored. + #[serde(default = "default_persist")] + pub persist: bool, + /// Whether an unchanged document is validated again. + #[serde(default)] + pub revalidate: Revalidate, } fn default_threshold() -> Severity { @@ -87,6 +198,10 @@ fn default_run_on_ingest() -> bool { true } +fn default_persist() -> bool { + true +} + /// URL and request timeout for a long-lived `ec validate input --server` instance. #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct ConformaConfig { @@ -114,25 +229,46 @@ pub fn build(config: &ValidatorsConfig) -> Result>, anyho "duplicate validator name: {}", validator.name ); - validators.push(build_one(validator)?); + validators.push(build_one(config, validator)?); } Ok(validators) } /// Build a single validator from its configuration. -fn build_one(validator: &ValidatorConfig) -> Result, anyhow::Error> { +// Without any backend compiled in, only the fallback arm below remains, which +// does not look at the global settings. +#[cfg_attr(not(feature = "semantic-validation"), allow(unused_variables))] +fn build_one( + global: &ValidatorsConfig, + validator: &ValidatorConfig, +) -> Result, anyhow::Error> { match &validator.backend { #[cfg(feature = "semantic-validation")] - Backend::Scheck { rules, phase } => { - Ok(Arc::new(build_scheck(validator, rules, phase.as_deref())?)) - } - #[cfg(feature = "semantic-validation")] - Backend::Csaf { profile } => Ok(Arc::new(csaf::Validator::new( + Backend::Scheck { rules, phase } => Ok(Arc::new(build_scheck( + global, validator, - profile.as_deref(), - ))), + rules, + phase.as_deref(), + )?)), #[cfg(feature = "semantic-validation")] - Backend::Conforma(conforma) => Ok(Arc::new(conforma::build(validator, conforma)?)), + Backend::Csaf { profile } => { + // The profile selects which spec tests run, so it belongs to the + // fingerprint; the serialized backend already carries it. + let persistence = Persistence::new(global, validator, &[])?; + Ok(Arc::new(csaf::Validator::new( + validator, + profile.as_deref(), + persistence, + ))) + } + #[cfg(feature = "semantic-validation")] + Backend::Conforma(conforma) => { + // Nothing here fingerprints the remote policy -- it is configured + // on the Conforma server and can change without us knowing, which + // is why `Revalidate::OnChange` is unsafe for this backend. + let persistence = Persistence::new(global, validator, &[])?; + Ok(Arc::new(conforma::build(validator, conforma, persistence)?)) + } #[allow(unreachable_patterns)] backend => anyhow::bail!( "validator '{}' uses backend {backend:?}, which is not compiled into this build \ @@ -144,23 +280,57 @@ fn build_one(validator: &ValidatorConfig) -> Result, anyhow:: #[cfg(feature = "semantic-validation")] fn build_scheck( + global: &ValidatorsConfig, config: &ValidatorConfig, rules: &[PathBuf], phase: Option<&str>, ) -> Result { let mut schemas = Vec::with_capacity(rules.len()); + let mut sources = Vec::with_capacity(rules.len()); for path in rules { let contents = std::fs::read_to_string(path) .with_context(|| format!("reading scheck ruleset {}", path.display()))?; schemas.push(scheck::parse_ruleset(path, &contents)?); + sources.push(contents); } - Ok(ScheckValidator::new(config, schemas, phase)) + + // The configuration records ruleset paths, not their contents: editing a + // ruleset in place has to change the fingerprint. + let inputs = sources.iter().map(|s| s.as_bytes()).collect::>(); + let persistence = Persistence::new(global, config, &inputs)?; + + Ok(ScheckValidator::new(config, schemas, phase, persistence)) } #[cfg(test)] mod tests { use super::*; + #[test] + fn persistence_defaults_to_storing_everything() { + let config = ValidatorsConfig::default(); + assert!(config.persist_reports); + + let config: ValidatorsConfig = + serde_yml::from_str("validators:\n - name: v\n").expect("parses"); + let validator = &config.validators[0]; + assert!(validator.persist); + // Never skip a run by default -- a cached verdict is only safe when we + // can prove the inputs are unchanged. + assert_eq!(validator.revalidate, Revalidate::Always); + } + + #[test] + fn global_persist_reports_overrides_validators() { + let config: ValidatorsConfig = serde_yml::from_str( + "persist_reports: false\nvalidators:\n - name: v\n persist: true\n", + ) + .expect("parses"); + + let validators = build(&config).expect("builds"); + assert!(!validators[0].persistence().persist); + } + #[test] fn default_config_builds_empty_set() { let config = ValidatorsConfig::default(); @@ -280,6 +450,8 @@ validators: #[test] fn builds_csaf_validator_from_config() { let config = ValidatorsConfig { + persist_reports: true, + caps: Default::default(), validators: vec![ValidatorConfig { name: "csaf-spec".into(), backend: Backend::Csaf { @@ -290,6 +462,8 @@ validators: mode: ValidationMode::Verify, threshold: Severity::Error, on_error: OnError::Block, + persist: true, + revalidate: Revalidate::Always, }], }; let validators = build(&config).expect("builds"); @@ -309,6 +483,8 @@ validators: .expect("write ruleset"); let config = ValidatorsConfig { + persist_reports: true, + caps: Default::default(), validators: vec![ValidatorConfig { name: "scheck".into(), backend: Backend::Scheck { @@ -320,12 +496,25 @@ validators: mode: ValidationMode::Report, threshold: Severity::Error, on_error: OnError::Block, + persist: true, + revalidate: Revalidate::Always, }], }; let validators = build(&config).expect("builds"); assert_eq!(validators.len(), 1); assert_eq!(validators[0].name(), "scheck"); + + // Editing the ruleset in place has to change the fingerprint: the + // configuration only records the path. + let before = validators[0].persistence().fingerprint.clone(); + std::fs::write( + dir.path().join("rules.json"), + r#"{"title":"t","patterns":[{"name":"p","title":"p","rules":[{"context":"$","checks":[{"kind":"assert","test":{"type":"exists","path":"$.y"},"message":"needs y"}]}]}]}"#, + ) + .expect("rewrite ruleset"); + let validators = build(&config).expect("builds"); + assert_ne!(validators[0].persistence().fingerprint, before); } #[test] diff --git a/modules/ingestor/src/service/validation/conforma.rs b/modules/ingestor/src/service/validation/conforma.rs index 8d0ee6c855..78eb92722c 100644 --- a/modules/ingestor/src/service/validation/conforma.rs +++ b/modules/ingestor/src/service/validation/conforma.rs @@ -3,8 +3,8 @@ use crate::service::{ Format, validation::{ - Finding, OnError, Severity, ValidationMode, ValidationOutcome, ValidationReport, Validator, - ValidatorError, ValidatorInput, + Finding, OnError, Persistence, Severity, ValidationMode, ValidationOutcome, + ValidationReport, Validator, ValidatorError, ValidatorInput, config::{ConformaConfig, ValidatorConfig}, }, }; @@ -28,10 +28,15 @@ pub struct ConformaValidator { timeout: Duration, endpoint: Url, client: reqwest::Client, + persistence: Persistence, } impl ConformaValidator { - fn new(config: &ValidatorConfig, conforma: &ConformaConfig) -> anyhow::Result { + fn new( + config: &ValidatorConfig, + conforma: &ConformaConfig, + persistence: Persistence, + ) -> anyhow::Result { ensure!( conforma.timeout_seconds > 0 && conforma.timeout_seconds <= 86_400, "Conforma timeout_seconds must be between 1 and 86400" @@ -55,6 +60,7 @@ impl ConformaValidator { timeout: Duration::from_secs(conforma.timeout_seconds), endpoint, client, + persistence, }) } @@ -128,6 +134,10 @@ impl Validator for ConformaValidator { .any(|configured| format.matches_hint(*configured)) } + fn persistence(&self) -> &Persistence { + &self.persistence + } + async fn validate( &self, input: &ValidatorInput<'_>, @@ -143,8 +153,9 @@ impl Validator for ConformaValidator { pub(crate) fn build( config: &ValidatorConfig, conforma: &ConformaConfig, + persistence: Persistence, ) -> anyhow::Result { - ConformaValidator::new(config, conforma) + ConformaValidator::new(config, conforma, persistence) } fn endpoint_url(url: &str) -> anyhow::Result { @@ -297,6 +308,7 @@ fn map_report( #[cfg(test)] mod tests { use super::*; + use crate::service::validation::Revalidate; use crate::service::validation::{ Backend, ConformaConfig, OnError, ValidatorConfig, ValidatorsConfig, validate_named, }; @@ -318,6 +330,8 @@ mod tests { fn validators_config(url: String) -> ValidatorsConfig { ValidatorsConfig { + persist_reports: true, + caps: Default::default(), validators: vec![ValidatorConfig { name: "conforma-test".into(), backend: Backend::Conforma(ConformaConfig { @@ -329,6 +343,8 @@ mod tests { mode: ValidationMode::Report, threshold: Severity::Error, on_error: OnError::Continue, + persist: true, + revalidate: Revalidate::Always, }], } } diff --git a/modules/ingestor/src/service/validation/csaf.rs b/modules/ingestor/src/service/validation/csaf.rs index 7fc046afb4..fbdc8f9809 100644 --- a/modules/ingestor/src/service/validation/csaf.rs +++ b/modules/ingestor/src/service/validation/csaf.rs @@ -7,8 +7,8 @@ use crate::service::{ Format, validation::{ - Finding, OnError, Severity, ValidationMode, ValidationOutcome, ValidationReport, - ValidatorError, ValidatorInput, config::ValidatorConfig, + Finding, OnError, Persistence, Severity, ValidationMode, ValidationOutcome, + ValidationReport, ValidatorError, ValidatorInput, config::ValidatorConfig, }, }; use anyhow::anyhow; @@ -28,10 +28,11 @@ pub struct Validator { threshold: Severity, on_error: OnError, run_on_ingest: bool, + persistence: Persistence, } impl Validator { - pub fn new(config: &ValidatorConfig, profile: Option<&str>) -> Self { + pub fn new(config: &ValidatorConfig, profile: Option<&str>, persistence: Persistence) -> Self { Self { name: config.name.clone(), profile: profile.unwrap_or("basic").to_owned(), @@ -39,6 +40,7 @@ impl Validator { threshold: config.threshold, on_error: config.on_error, run_on_ingest: config.run_on_ingest, + persistence, } } } @@ -52,6 +54,7 @@ impl fmt::Debug for Validator { .field("threshold", &self.threshold) .field("on_error", &self.on_error) .field("run_on_ingest", &self.run_on_ingest) + .field("persistence", &self.persistence) .finish() } } @@ -124,6 +127,10 @@ impl super::Validator for Validator { format == Format::CSAF } + fn persistence(&self) -> &Persistence { + &self.persistence + } + async fn validate( &self, input: &ValidatorInput<'_>, @@ -170,6 +177,7 @@ impl super::Validator for Validator { #[cfg(test)] mod tests { use super::*; + use crate::service::validation::Revalidate; use crate::service::validation::{Backend, Validator as _, ValidatorConfig}; fn config(profile: Option<&str>, mode: ValidationMode) -> ValidatorConfig { @@ -183,11 +191,13 @@ mod tests { mode, threshold: Severity::Error, on_error: OnError::Block, + persist: true, + revalidate: Revalidate::Always, } } fn validator(profile: Option<&str>, mode: ValidationMode) -> Validator { - Validator::new(&config(profile, mode), profile) + Validator::new(&config(profile, mode), profile, Persistence::default()) } #[tokio::test] @@ -289,8 +299,10 @@ mod tests { mode: ValidationMode::Report, threshold: Severity::Error, on_error: OnError::Block, + persist: true, + revalidate: Revalidate::Always, }; - let v = Validator::new(&config, None); + let v = Validator::new(&config, None, Persistence::default()); assert!(v.applies_to(Format::CSAF)); assert!(!v.applies_to(Format::CVE)); assert!(!v.applies_to(Format::OSV)); diff --git a/modules/ingestor/src/service/validation/mod.rs b/modules/ingestor/src/service/validation/mod.rs index a8b2d2841b..ab07228283 100644 --- a/modules/ingestor/src/service/validation/mod.rs +++ b/modules/ingestor/src/service/validation/mod.rs @@ -5,7 +5,7 @@ //! modes: [`ValidationMode::Report`] (findings are recorded but never block) or //! [`ValidationMode::Verify`] (a failing outcome blocks ingestion). //! -//! See ADR 00020 for the design and rationale. +//! See ADR 00021 for the design and rationale. pub mod config; #[cfg(feature = "semantic-validation")] @@ -14,8 +14,11 @@ pub mod conforma; pub mod csaf; #[cfg(feature = "semantic-validation")] pub mod scheck; +pub mod store; -pub use config::{Backend, ConformaConfig, ValidatorConfig, ValidatorsConfig, build}; +pub use config::{ + Backend, ConformaConfig, Persistence, Revalidate, ValidatorConfig, ValidatorsConfig, build, +}; #[cfg(feature = "semantic-validation")] pub use scheck::ScheckValidator; @@ -158,6 +161,9 @@ pub trait Validator: Send + Sync + Debug { /// Whether this validator applies to the given document format. fn applies_to(&self, format: Format) -> bool; + /// How this validator's results are stored, and when it re-runs. + fn persistence(&self) -> &Persistence; + /// Run the validator against a document. /// /// Returns a [`ValidationReport`] on success. Returning `Err` means the diff --git a/modules/ingestor/src/service/validation/scheck.rs b/modules/ingestor/src/service/validation/scheck.rs index 203768edc7..93b9d3b783 100644 --- a/modules/ingestor/src/service/validation/scheck.rs +++ b/modules/ingestor/src/service/validation/scheck.rs @@ -1,13 +1,13 @@ //! In-process [`Validator`] backend using the [`scheck`] semantic validator. //! //! Rulesets are parsed once at construction; each document is validated against -//! every configured ruleset. See ADR 00020. +//! every configured ruleset. See ADR 00021. use crate::service::{ Format, validation::{ - Finding, OnError, Severity, ValidationMode, ValidationOutcome, ValidationReport, Validator, - ValidatorError, ValidatorInput, config::ValidatorConfig, + Finding, OnError, Persistence, Severity, ValidationMode, ValidationOutcome, + ValidationReport, Validator, ValidatorError, ValidatorInput, config::ValidatorConfig, }, }; use anyhow::{Context, anyhow}; @@ -25,6 +25,7 @@ pub struct ScheckValidator { threshold: Severity, on_error: OnError, run_on_ingest: bool, + persistence: Persistence, } impl ScheckValidator { @@ -33,6 +34,7 @@ impl ScheckValidator { config: &ValidatorConfig, schemas: Vec, phase: Option<&str>, + persistence: Persistence, ) -> Self { Self { name: config.name.clone(), @@ -43,6 +45,7 @@ impl ScheckValidator { threshold: config.threshold, on_error: config.on_error, run_on_ingest: config.run_on_ingest, + persistence, } } } @@ -58,6 +61,7 @@ impl fmt::Debug for ScheckValidator { .field("threshold", &self.threshold) .field("on_error", &self.on_error) .field("run_on_ingest", &self.run_on_ingest) + .field("persistence", &self.persistence) .finish() } } @@ -132,6 +136,10 @@ impl Validator for ScheckValidator { .any(|configured| format.matches_hint(*configured)) } + fn persistence(&self) -> &Persistence { + &self.persistence + } + async fn validate( &self, input: &ValidatorInput<'_>, @@ -178,6 +186,7 @@ impl Validator for ScheckValidator { #[cfg(test)] mod tests { use super::*; + use crate::service::validation::Revalidate; use crate::service::validation::ValidatorConfig; /// A ruleset asserting the document has a `document` object at the root. @@ -212,12 +221,19 @@ mod tests { mode, threshold: Severity::Error, on_error: OnError::Block, + persist: true, + revalidate: Revalidate::Always, } } fn validator(mode: ValidationMode) -> ScheckValidator { let schema = serde_json::from_str(RULESET).expect("valid ruleset"); - ScheckValidator::new(&config(vec![Format::CSAF], mode), vec![schema], None) + ScheckValidator::new( + &config(vec![Format::CSAF], mode), + vec![schema], + None, + Persistence::default(), + ) } #[tokio::test] @@ -260,6 +276,7 @@ mod tests { &config(vec![Format::SBOM], ValidationMode::Report), vec![schema], None, + Persistence::default(), ); assert!(validator.applies_to(Format::SPDX)); assert!(validator.applies_to(Format::CycloneDX)); diff --git a/modules/ingestor/src/service/validation/store.rs b/modules/ingestor/src/service/validation/store.rs new file mode 100644 index 0000000000..ee633530ac --- /dev/null +++ b/modules/ingestor/src/service/validation/store.rs @@ -0,0 +1,602 @@ +//! Persistence and retrieval of [`ValidationReport`]s. +//! +//! The store is **append-only**. A validator run inserts a row only when its +//! result differs from the latest stored result for the same document and +//! validator; an unchanged re-run writes nothing at all. That keeps the +//! repeated-ingest path — the dominant importer case — free of row updates, +//! and preserves the history of how a verdict changed when a ruleset changes. +//! +//! Rows are keyed by the document's SHA-256 rather than by a `source_document` +//! foreign key, because documents rejected by a `verify` validator are recorded +//! too, and those never reach storage. + +use crate::service::validation::{ + Finding, Severity, ValidationMode, ValidationOutcome, ValidationReport, +}; +use sea_orm::{ + ActiveValue::Set, ColumnTrait, ConnectionTrait, DbErr, EntityTrait, JoinType, QueryFilter, + QueryOrder, QuerySelect, RelationTrait, Select, +}; +use sea_query::Condition; +use sha2::{Digest as _, Sha256}; +use std::collections::HashMap; +use time::OffsetDateTime; +use tracing::instrument; +use trustify_common::id::{Id, IdError}; +use trustify_entity::{advisory, sbom, sbom_node, source_document, validation_report}; +use uuid::Uuid; + +pub use trustify_entity::validation_report::{IngestSource, Model as StoredReport}; + +/// An error raised while reading or writing validation reports. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// The database rejected the query. + #[error(transparent)] + Database(#[from] DbErr), + /// The document identifier could not be turned into a filter. + #[error(transparent)] + Id(#[from] IdError), + /// The findings could not be serialized for storage. + #[error("failed to serialize findings: {0}")] + Serialize(#[source] serde_json::Error), +} + +/// Bounds on how much of a report is written. +/// +/// Validator output is untrusted and unbounded: a loose ruleset against a large +/// document can produce megabytes of findings. Capping them bounds both the +/// serialization work on the ingest path and the row size. +#[derive(Clone, Copy, Debug, serde::Deserialize, serde::Serialize)] +pub struct Caps { + #[serde(default = "default_max_findings")] + /// Maximum number of findings stored per report. + pub max_findings: usize, + /// Maximum serialized size of the stored findings, in bytes. + #[serde(default = "default_max_findings_bytes")] + pub max_findings_bytes: usize, +} + +fn default_max_findings() -> usize { + 200 +} + +fn default_max_findings_bytes() -> usize { + 64 * 1024 +} + +impl Default for Caps { + fn default() -> Self { + Self { + max_findings: default_max_findings(), + max_findings_bytes: default_max_findings_bytes(), + } + } +} + +/// How a document reached the ingestor. +/// +/// Set on the service rather than passed per call: every construction site +/// already knows what it is, and the importers build their own instance per +/// run. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub enum Provenance { + /// Uploaded through the API. + Api, + /// Fetched by the named importer. + Importer(String), + /// Loaded as part of a dataset archive. + Dataset, + /// Ingested by internal code, including tests. + #[default] + Internal, +} + +impl Provenance { + fn source(&self) -> IngestSource { + match self { + Self::Api => IngestSource::Api, + Self::Importer(_) => IngestSource::Importer, + Self::Dataset => IngestSource::Dataset, + Self::Internal => IngestSource::Internal, + } + } + + fn importer(&self) -> Option<&str> { + match self { + Self::Importer(name) => Some(name), + _ => None, + } + } +} + +/// Where a document came from, and under which configuration it was validated. +#[derive(Clone, Copy, Debug)] +pub struct PersistContext<'a> { + /// SHA-256 of the raw document, as lowercase hex. + pub document_sha256: &'a str, + /// How the document entered the system. + pub ingest_source: IngestSource, + /// Name of the importer, when `ingest_source` is [`IngestSource::Importer`]. + pub importer_name: Option<&'a str>, + /// Bounds on how much of each report is stored. + pub caps: Caps, +} + +impl<'a> PersistContext<'a> { + /// Build the context for a document, taking provenance from the service. + pub fn new(document_sha256: &'a str, provenance: &'a Provenance, caps: Caps) -> Self { + Self { + document_sha256, + ingest_source: provenance.source(), + importer_name: provenance.importer(), + caps, + } + } +} + +/// One validator result awaiting persistence. +#[derive(Clone, Copy, Debug)] +pub struct PendingReport<'a> { + /// The report produced by the validator. + pub report: &'a ValidationReport, + /// The mode the validator ran in. + pub mode: ValidationMode, + /// True when this result rejected the document. + pub blocked: bool, + /// Digest of the configuration that produced the result. + pub config_fingerprint: &'a str, +} + +/// Store the reports whose result differs from the latest stored one. +/// +/// Returns the number of rows inserted; zero means every result was already +/// recorded. Reads the latest content hashes in one query and writes the new +/// rows in one statement, so the cost is at most two round trips per document +/// regardless of how many validators ran. +#[instrument(skip_all, fields(document = ctx.document_sha256), err(level = tracing::Level::INFO))] +pub async fn persist_if_changed<'a, C: ConnectionTrait>( + conn: &C, + ctx: &PersistContext<'_>, + pending: impl IntoIterator>, +) -> Result { + let pending = pending.into_iter().collect::>(); + if pending.is_empty() { + return Ok(0); + } + + let names = pending + .iter() + .map(|entry| entry.report.validator.as_str()) + .collect::>(); + let known = latest_content_hashes(conn, ctx.document_sha256, names).await?; + + let mut models = Vec::with_capacity(pending.len()); + for entry in pending { + let (model, content_hash) = to_model(ctx, &entry)?; + if known.get(&entry.report.validator) == Some(&content_hash) { + continue; + } + models.push(model); + } + + let inserted = models.len(); + if inserted > 0 { + validation_report::Entity::insert_many(models) + .exec(conn) + .await?; + } + + Ok(inserted) +} + +/// The latest content hash per validator for one document. +async fn latest_content_hashes<'a, C: ConnectionTrait>( + conn: &C, + document_sha256: &str, + validators: impl IntoIterator, +) -> Result, Error> { + /// Only the columns needed to decide whether a result changed — notably + /// not `findings`, which can be tens of kilobytes per row. + #[derive(Debug, sea_orm::FromQueryResult)] + struct Row { + validator: String, + content_hash: String, + } + + let rows = validation_report::Entity::find() + .select_only() + .column(validation_report::Column::Validator) + .column(validation_report::Column::ContentHash) + .filter(validation_report::Column::DocumentSha256.eq(document_sha256)) + .filter(validation_report::Column::Validator.is_in(validators)) + // Newest last, so later rows overwrite earlier ones in the map below. + .order_by_asc(validation_report::Column::CreatedAt) + .into_model::() + .all(conn) + .await?; + + Ok(rows + .into_iter() + .map(|row| (row.validator, row.content_hash)) + .collect()) +} + +/// The current result of every validator that has run against one document. +/// +/// Bounded by the number of validators, not by history: older rows for the same +/// validator are discarded. +#[instrument(skip(conn), err(level = tracing::Level::INFO))] +pub async fn latest_for_digest( + conn: &C, + document_sha256: &str, +) -> Result, Error> { + let mut latest = HashMap::new(); + for report in by_digest(document_sha256) + .order_by_asc(validation_report::Column::CreatedAt) + .all(conn) + .await? + { + latest.insert(report.validator.clone(), report); + } + + let mut reports = latest.into_values().collect::>(); + reports.sort_by(|left, right| left.validator.cmp(&right.validator)); + Ok(reports) +} + +/// Every report recorded for a document digest, newest first. +/// +/// Composable: callers add their own ordering, pagination or filters. +pub fn by_digest(document_sha256: &str) -> Select { + validation_report::Entity::find() + .filter(validation_report::Column::DocumentSha256.eq(document_sha256)) +} + +/// Every report recorded for the document identified by `id`. +/// +/// Accepts an internal UUID (of an SBOM or an advisory) or any supported +/// digest. A UUID or a non-SHA-256 digest resolves through `source_document`, +/// so it only matches documents that were actually ingested — a rejected +/// document can only be found by its SHA-256. +pub fn by_document_id(id: Id) -> Result, Error> { + // The common case needs no join: the digest is the stored key. + if let Id::Sha256(sha256) = &id { + return Ok(by_digest(sha256)); + } + + let query = validation_report::Entity::find().join( + JoinType::InnerJoin, + validation_report::Relation::SourceDocument.def(), + ); + + Ok(match id { + Id::Uuid(uuid) => query + .join_rev(JoinType::LeftJoin, sbom::Relation::SourceDocument.def()) + .join_rev(JoinType::LeftJoin, advisory::Relation::SourceDocument.def()) + .filter( + Condition::any() + .add(sbom::Column::SbomId.eq(uuid)) + .add(advisory::Column::Id.eq(uuid)), + ), + Id::Sha384(digest) => query.filter(source_document::Column::Sha384.eq(digest)), + Id::Sha512(digest) => query.filter(source_document::Column::Sha512.eq(digest)), + id => return Err(Error::Id(IdError::UnsupportedAlgorithm(id.to_string()))), + }) +} + +/// Every report recorded for documents with the given name. +/// +/// Matches the name of the SBOM's describing node and the advisory identifier. +/// Neither is unique, so this can span several documents; callers are expected +/// to paginate. +pub fn by_document_name(name: &str) -> Select { + validation_report::Entity::find() + .join( + JoinType::InnerJoin, + validation_report::Relation::SourceDocument.def(), + ) + .join_rev(JoinType::LeftJoin, sbom::Relation::SourceDocument.def()) + .join(JoinType::LeftJoin, sbom::Relation::SbomNode.def()) + .join_rev(JoinType::LeftJoin, advisory::Relation::SourceDocument.def()) + .filter( + Condition::any() + .add(sbom_node::Column::Name.eq(name)) + .add(advisory::Column::Identifier.eq(name)), + ) +} + +/// Build the row for one pending report, returning it with its content hash. +fn to_model( + ctx: &PersistContext<'_>, + entry: &PendingReport<'_>, +) -> Result<(validation_report::ActiveModel, String), Error> { + let finding_count = entry.report.findings.len(); + let max_severity = entry + .report + .findings + .iter() + .map(|finding| finding.severity) + .max() + .map(map_severity); + + let (findings, truncated) = truncate(&entry.report.findings, ctx.caps)?; + let mode = map_mode(entry.mode); + let outcome = map_outcome(entry.report.outcome); + + // Everything that distinguishes one result from another, including the + // configuration that produced it, so that a ruleset change records a new + // row even when the findings are identical. + let mut hasher = Sha256::new(); + for part in [ + entry.report.validator.as_str(), + &mode.to_string(), + &outcome.to_string(), + &entry.blocked.to_string(), + &max_severity.map(|s| s.to_string()).unwrap_or_default(), + &finding_count.to_string(), + &truncated.to_string(), + entry.config_fingerprint, + &findings.to_string(), + ] { + hasher.update(part.as_bytes()); + hasher.update(b"\0"); + } + let content_hash = hex::encode(hasher.finalize()); + + let model = validation_report::ActiveModel { + id: Set(Uuid::now_v7()), + document_sha256: Set(ctx.document_sha256.to_owned()), + validator: Set(entry.report.validator.clone()), + mode: Set(mode), + outcome: Set(outcome), + blocked: Set(entry.blocked), + max_severity: Set(max_severity), + // Saturates rather than wrapping; a report with more than 2^31 findings + // is not a number anyone needs to be exact about. + finding_count: Set(i32::try_from(finding_count).unwrap_or(i32::MAX)), + findings: Set(findings), + truncated: Set(truncated), + content_hash: Set(content_hash.clone()), + config_fingerprint: Set(entry.config_fingerprint.to_owned()), + ingest_source: Set(ctx.ingest_source), + importer_name: Set(ctx.importer_name.map(str::to_owned)), + created_at: Set(OffsetDateTime::now_utc()), + }; + + Ok((model, content_hash)) +} + +/// Serialize findings, dropping entries until they fit within `caps`. +/// +/// Bounds both the work done on the ingest path and the row size: validator +/// output is untrusted, and a loose ruleset against a large document can +/// otherwise produce megabytes of JSON per report. +fn truncate(findings: &[Finding], caps: Caps) -> Result<(serde_json::Value, bool), Error> { + let mut kept = findings.len().min(caps.max_findings); + loop { + let value = serde_json::to_value(&findings[..kept]).map_err(Error::Serialize)?; + if kept == 0 || value.to_string().len() <= caps.max_findings_bytes { + return Ok((value, kept < findings.len())); + } + kept /= 2; + } +} + +fn map_mode(mode: ValidationMode) -> validation_report::Mode { + match mode { + ValidationMode::Report => validation_report::Mode::Report, + ValidationMode::Verify => validation_report::Mode::Verify, + } +} + +fn map_outcome(outcome: ValidationOutcome) -> validation_report::Outcome { + match outcome { + ValidationOutcome::Passed => validation_report::Outcome::Passed, + ValidationOutcome::Failed => validation_report::Outcome::Failed, + } +} + +fn map_severity(severity: Severity) -> validation_report::Severity { + match severity { + Severity::Info => validation_report::Severity::Info, + Severity::Warning => validation_report::Severity::Warning, + Severity::Error => validation_report::Severity::Error, + Severity::Fatal => validation_report::Severity::Fatal, + } +} + +/// Resolve the `source_document` digest for an ingested document, if it exists. +/// +/// Used by callers that hold a document's internal ID and need the key this +/// store is organised by. +#[instrument(skip(conn), err(level = tracing::Level::INFO))] +pub async fn digest_for_document( + conn: &C, + id: Id, +) -> Result, Error> { + if let Id::Sha256(sha256) = id { + return Ok(Some(sha256)); + } + + let digest = source_document::Entity::find() + .join_rev(JoinType::LeftJoin, sbom::Relation::SourceDocument.def()) + .join_rev(JoinType::LeftJoin, advisory::Relation::SourceDocument.def()) + .filter(match id { + Id::Uuid(uuid) => Condition::any() + .add(sbom::Column::SbomId.eq(uuid)) + .add(advisory::Column::Id.eq(uuid)), + Id::Sha384(digest) => Condition::all().add(source_document::Column::Sha384.eq(digest)), + Id::Sha512(digest) => Condition::all().add(source_document::Column::Sha512.eq(digest)), + Id::Sha256(digest) => Condition::all().add(source_document::Column::Sha256.eq(digest)), + id => return Err(Error::Id(IdError::UnsupportedAlgorithm(id.to_string()))), + }) + .one(conn) + .await?; + + Ok(digest.map(|document| document.sha256)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::service::validation::Finding; + use test_context::test_context; + use test_log::test; + use trustify_test_context::TrustifyContext; + + const DIGEST: &str = "0000000000000000000000000000000000000000000000000000000000000001"; + + fn report(findings: Vec, outcome: ValidationOutcome) -> ValidationReport { + ValidationReport { + validator: "test-validator".to_string(), + findings, + outcome, + } + } + + fn finding(severity: Severity, message: &str) -> Finding { + Finding { + severity, + message: message.to_string(), + path: None, + rule: None, + } + } + + fn context(document_sha256: &str) -> PersistContext<'_> { + PersistContext { + document_sha256, + ingest_source: IngestSource::Internal, + importer_name: None, + caps: Caps::default(), + } + } + + fn pending<'a>(report: &'a ValidationReport, fingerprint: &'a str) -> PendingReport<'a> { + PendingReport { + report, + mode: ValidationMode::Report, + blocked: false, + config_fingerprint: fingerprint, + } + } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn unchanged_result_writes_nothing(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + let report = report(vec![], ValidationOutcome::Passed); + + let inserted = + persist_if_changed(&ctx.db, &context(DIGEST), [pending(&report, "fp-1")]).await?; + assert_eq!(inserted, 1); + + // The same result from a later run must not produce a second row: this + // is what keeps repeated importer ingests off the write path. + let inserted = + persist_if_changed(&ctx.db, &context(DIGEST), [pending(&report, "fp-1")]).await?; + assert_eq!(inserted, 0); + + assert_eq!(by_digest(DIGEST).all(&ctx.db).await?.len(), 1); + + Ok(()) + } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn changed_result_appends(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + let digest = &DIGEST.replace("1", "2"); + + let passed = report(vec![], ValidationOutcome::Passed); + persist_if_changed(&ctx.db, &context(digest), [pending(&passed, "fp-1")]).await?; + + let failed = report( + vec![finding(Severity::Error, "missing field")], + ValidationOutcome::Failed, + ); + let inserted = + persist_if_changed(&ctx.db, &context(digest), [pending(&failed, "fp-1")]).await?; + assert_eq!(inserted, 1); + + // History is kept, and the newer verdict is the current one. + assert_eq!(by_digest(digest).all(&ctx.db).await?.len(), 2); + + let latest = latest_for_digest(&ctx.db, digest).await?; + assert_eq!(latest.len(), 1); + assert_eq!(latest[0].outcome, validation_report::Outcome::Failed); + assert_eq!(latest[0].finding_count, 1); + assert_eq!( + latest[0].max_severity, + Some(validation_report::Severity::Error) + ); + + Ok(()) + } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn changed_configuration_appends(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + let digest = &DIGEST.replace("1", "3"); + let report = report(vec![], ValidationOutcome::Passed); + + persist_if_changed(&ctx.db, &context(digest), [pending(&report, "fp-1")]).await?; + + // Same verdict, different ruleset: the row records which configuration + // produced it, so a new row is required to stay accurate. + let inserted = + persist_if_changed(&ctx.db, &context(digest), [pending(&report, "fp-2")]).await?; + assert_eq!(inserted, 1); + + Ok(()) + } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn findings_are_capped(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + let digest = &DIGEST.replace("1", "4"); + let findings = (0..10) + .map(|i| finding(Severity::Warning, &format!("finding {i}"))) + .collect(); + let report = report(findings, ValidationOutcome::Passed); + + let mut ctx_with_caps = context(digest); + ctx_with_caps.caps = Caps { + max_findings: 3, + max_findings_bytes: 64 * 1024, + }; + persist_if_changed(&ctx.db, &ctx_with_caps, [pending(&report, "fp-1")]).await?; + + let stored = latest_for_digest(&ctx.db, digest).await?; + assert!(stored[0].truncated); + // The count reports everything the validator found, even though only + // the capped subset is stored. + assert_eq!(stored[0].finding_count, 10); + assert_eq!(stored[0].findings.as_array().map(Vec::len), Some(3)); + + Ok(()) + } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn resolves_by_document_id_and_name(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + let result = ctx + .ingest_document("zookeeper-3.9.2-cyclonedx.json") + .await?; + let digest = digest_for_document(&ctx.db, Id::Uuid(Uuid::parse_str(&result.id)?)) + .await? + .expect("ingested document has a source document"); + + let report = report(vec![], ValidationOutcome::Passed); + persist_if_changed(&ctx.db, &context(&digest), [pending(&report, "fp-1")]).await?; + + let by_id = by_document_id(Id::Uuid(Uuid::parse_str(&result.id)?))? + .all(&ctx.db) + .await?; + assert_eq!(by_id.len(), 1); + + let by_name = by_document_name("zookeeper").all(&ctx.db).await?; + assert_eq!(by_name.len(), 1); + + Ok(()) + } +} diff --git a/openapi.yaml b/openapi.yaml index 423e76feb7..3371016a99 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -4454,6 +4454,156 @@ paths: description: User preferences are deleted '412': description: The provided If-Match revision did not match the actual revision + /api/v3/validation: + get: + tags: + - validation + summary: List validation reports + operationId: listValidationReports + parameters: + - name: name + in: query + description: |- + Name of the document: an SBOM's describing node, or an advisory + identifier. Neither is unique, so this can match several documents. + required: false + schema: + type: + - string + - 'null' + - name: q + in: query + description: | + EBNF grammar for the _q_ parameter: + ```text + q = ( values | filter ) { '&' q } + values = value { '|', values } + filter = field, operator, values + operator = "=" | "!=" | "~" | "!~" | ">=" | ">" | "<=" | "<" + value = (* any text but escape special characters with '\' *) + field = (* must match an entity attribute name *) + ``` + Any values in a _q_ will result in a case-insensitive "full + text search", effectively producing an OR clause of LIKE + clauses for every string-ish field in the resource being + queried. + + Examples: + - `foo` - any field containing 'foo' + - `foo|bar` - any field containing either 'foo' OR 'bar' + - `foo&bar` - some field contains 'foo' AND some field contains 'bar' + + A _filter_ may also be used to constrain the results. The + filter's field name must correspond to one of the resource's + attributes. If it doesn't, an error will be returned + containing a list of the valid fields for that resource. + + An ASCII value of `NUL`, percent-encoded as `%00`, may be used + to find resources on which a particular field isn't set. For + example, `name=%00` and `name!=%00` yield the WHERE clauses, + 'NAME IS NULL' and 'NAME IS NOT NULL', respectively. + + Examples: + - `name=foo` - entity's _name_ matches 'foo' exactly + - `name~foo` - entity's _name_ contains 'foo', case-insensitive + - `name~foo|bar` - entity's _name_ contains either 'foo' OR 'bar', case-insensitive + - `name=` - entity's _name_ is the empty string, '' + - `name=%00` - entity's _name_ isn't set + - `published>3 days ago` - date values can be "human time" + + Multiple full text searches and/or filters should be + '&'-delimited -- they are logically AND'd together. + + - `red hat|fedora&labels:type=cve|osv&published>last wednesday 17:00` + + Fields corresponding to JSON objects in the database may use a + ':' to delimit the column name and the object key, + e.g. `purl:qualifiers:type=pom` + + Any operator or special character, e.g. '|', '&', within a + value should be escaped by prefixing it with a backslash. + required: false + schema: + type: string + - name: sort + in: query + description: | + EBNF grammar for the _sort_ parameter: + ```text + sort = field [ ':', order ] { ',' sort } + order = ( "asc" | "desc" ) + field = (* must match the name of entity's attributes *) + ``` + The optional _order_ should be one of "asc" or "desc". If + omitted, the order defaults to "asc". + + Each _field_ name must correspond to one of the columns of the + table holding the entities being queried. Those corresponding + to JSON objects in the database may use a ':' to delimit the + column name and the object key, + e.g. `purl:qualifiers:type:desc` + required: false + schema: + type: string + - name: offset + in: query + description: |- + The first item to return, skipping all that come before it. + + NOTE: The order of items is defined by the API being called. + required: false + schema: + type: integer + format: int64 + minimum: 0 + - name: limit + in: query + description: |- + The maximum number of entries to return. + + Zero means: return no items (the total count is still computed if requested). + required: false + schema: + type: integer + format: int64 + minimum: 0 + - name: total + in: query + description: Whether to compute and return the total count of matching items. + required: false + schema: + type: boolean + responses: + '200': + description: Matching validation reports + content: + application/json: + schema: + $ref: '#/components/schemas/PaginatedResults_ValidationReportSummary' + /api/v3/validation/{key}: + get: + tags: + - validation + summary: Retrieve the validation reports of one document + operationId: getValidationReports + parameters: + - name: key + in: path + description: Identifier of the document, either a digest e.g. `sha256:` or `urn:uuid:` of an ingested SBOM or advisory + required: true + schema: + $ref: '#/components/schemas/Id' + responses: + '200': + description: Validation reports for the document + content: + application/json: + schema: + type: array + items: + $ref: '#/components/schemas/ValidationReportSummary' + '400': + description: The key could not be interpreted as a document identifier /api/v3/vulnerability: get: tags: @@ -6174,7 +6324,7 @@ components: description: |- Structured reports from semantic validators run during ingestion. - See ADR 00020. Empty when no validators are configured or none apply + See ADR 00021. Empty when no validators are configured or none apply to the document's format. warnings: type: array @@ -7268,6 +7418,78 @@ components: - 'null' format: int64 minimum: 0 + PaginatedResults_ValidationReportSummary: + type: object + required: + - items + properties: + items: + type: array + items: + type: object + description: |- + A stored result of one semantic validator run against one document. + + Reports are append-only, so several may exist for the same document and + validator: each one records a result that differed from the one before it. + required: + - document_sha256 + - validator + - mode + - outcome + - blocked + - finding_count + - findings + - truncated + - ingest_source + - created_at + properties: + blocked: + type: boolean + description: True when this result rejected the document, so nothing was ingested. + created_at: + type: string + format: date-time + document_sha256: + type: string + description: SHA-256 of the validated document, as lowercase hex. + finding_count: + type: integer + format: int32 + description: Number of findings produced, before truncation. + findings: + description: The findings, as stored. Fewer than `finding_count` when `truncated`. + importer_name: + type: + - string + - 'null' + description: Name of the importer, when ingested by one. + ingest_source: + type: string + description: How the document entered the system. + max_severity: + type: + - string + - 'null' + description: Highest severity across the findings; absent when there are none. + mode: + type: string + description: Whether the validator gated ingestion or only reported. + outcome: + type: string + description: Whether the document met the validator's threshold. + truncated: + type: boolean + description: True when the stored findings were capped. + validator: + type: string + description: Name of the validator that produced the result. + total: + type: + - integer + - 'null' + format: int64 + minimum: 0 PaginatedResults_VulnerabilitySummary: type: object required: @@ -8311,6 +8533,65 @@ components: validator: type: string description: The [`Validator::name`] that produced this report. + ValidationReportSummary: + type: object + description: |- + A stored result of one semantic validator run against one document. + + Reports are append-only, so several may exist for the same document and + validator: each one records a result that differed from the one before it. + required: + - document_sha256 + - validator + - mode + - outcome + - blocked + - finding_count + - findings + - truncated + - ingest_source + - created_at + properties: + blocked: + type: boolean + description: True when this result rejected the document, so nothing was ingested. + created_at: + type: string + format: date-time + document_sha256: + type: string + description: SHA-256 of the validated document, as lowercase hex. + finding_count: + type: integer + format: int32 + description: Number of findings produced, before truncation. + findings: + description: The findings, as stored. Fewer than `finding_count` when `truncated`. + importer_name: + type: + - string + - 'null' + description: Name of the importer, when ingested by one. + ingest_source: + type: string + description: How the document entered the system. + max_severity: + type: + - string + - 'null' + description: Highest severity across the findings; absent when there are none. + mode: + type: string + description: Whether the validator gated ingestion or only reported. + outcome: + type: string + description: Whether the document met the validator's threshold. + truncated: + type: boolean + description: True when the stored findings were capped. + validator: + type: string + description: Name of the validator that produced the result. ValidationSeverity: type: string description: Severity of a single validation finding. diff --git a/server/src/profile/api.rs b/server/src/profile/api.rs index 4b28ba5841..c0dca06af6 100644 --- a/server/src/profile/api.rs +++ b/server/src/profile/api.rs @@ -132,7 +132,7 @@ pub struct Run { /// Path to a semantic validators configuration file (YAML). /// /// When unset (the default), no validators run and ingestion behaves as - /// before. See ADR 00020. + /// before. See ADR 00021. #[arg(long, env = "TRUSTD_VALIDATORS_CONFIG")] pub validators_config: Option, @@ -508,7 +508,7 @@ impl InitData { let ei_config = run.exploit_intelligence.into_config().await?; - // Build the semantic validator set (ADR 00020). With no config file, + // Build the semantic validator set (ADR 00021). With no config file, // this is empty and ingestion behaves exactly as before. let validators = match &run.validators_config { Some(path) => { From 2058fdeb42c752209253e7b54f14ea45c2c0a991 Mon Sep 17 00:00:00 2001 From: rh-jfuller Date: Fri, 9 Oct 2026 17:52:15 +0200 Subject: [PATCH 3/4] fix(ingestor): check validator names before building backends --- modules/ingestor/src/service/validation/config.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/modules/ingestor/src/service/validation/config.rs b/modules/ingestor/src/service/validation/config.rs index 330a0db006..4089e005df 100644 --- a/modules/ingestor/src/service/validation/config.rs +++ b/modules/ingestor/src/service/validation/config.rs @@ -221,7 +221,9 @@ fn default_conforma_timeout_seconds() -> u64 { /// Returns an empty set when no validators are configured, preserving the /// default validation-disabled behaviour. pub fn build(config: &ValidatorsConfig) -> Result>, anyhow::Error> { - let mut validators: Vec> = Vec::with_capacity(config.validators.len()); + // Check the configuration as a whole before building anything: a backend + // can read ruleset files or reject itself as not compiled in, and reporting + // that instead of a plain duplicate name would be misleading. let mut names = HashSet::with_capacity(config.validators.len()); for validator in &config.validators { anyhow::ensure!( @@ -229,6 +231,10 @@ pub fn build(config: &ValidatorsConfig) -> Result>, anyho "duplicate validator name: {}", validator.name ); + } + + let mut validators: Vec> = Vec::with_capacity(config.validators.len()); + for validator in &config.validators { validators.push(build_one(config, validator)?); } Ok(validators) From 89113b98038cf198e947225a14ec0d420b097992 Mon Sep 17 00:00:00 2001 From: rh-jfuller Date: Fri, 9 Oct 2026 18:48:24 +0200 Subject: [PATCH 4/4] fix(validation): honor cached reports and harden report writes Reuse unchanged verdicts, serialize report deduplication, preserve combined filters, and test the ingestor without semantic-validation. --- .github/workflows/ci.yaml | 20 +++ etc/validators/README.md | 9 +- migration/Cargo.toml | 2 +- .../src/validation/endpoints/mod.rs | 2 +- .../src/validation/endpoints/test.rs | 7 + .../fundamental/src/validation/service/mod.rs | 2 + modules/ingestor/src/service/mod.rs | 128 ++++++++++++++++-- .../ingestor/src/service/validation/config.rs | 27 +++- .../ingestor/src/service/validation/store.rs | 123 +++++++++-------- test-context/Cargo.toml | 2 +- 10 files changed, 251 insertions(+), 71 deletions(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index c485d1ecdb..268f19c32c 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -81,6 +81,25 @@ jobs: GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}" # for embedded postgresql run: cargo clippy -p trustify-module-ingestor --no-default-features -- -D warnings -D clippy::unwrap_used -D clippy::expect_used -D clippy::panic -D clippy::todo -D clippy::unimplemented + test-semantic-validation-off: + name: Test (semantic-validation off) + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v7 + - uses: ./.github/actions/setup-rust + - uses: Swatinem/rust-cache@v2 + with: + shared-key: ci + - name: Cache Theseus Postgresql Installation + uses: actions/cache@v6 + with: + path: ~/.theseus/postgresql + key: ${{ runner.os }}-theseus-postgresql-${{ hashFiles('**/Cargo.lock') }} + - name: Test ingestor library without semantic validation + run: cargo test -p trustify-module-ingestor --no-default-features --lib + env: + GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}" # for embedded postgresql + test: runs-on: ubuntu-24.04 @@ -168,6 +187,7 @@ jobs: - supply-chain - check - test + - test-semantic-validation-off if: always() steps: - name: Success diff --git a/etc/validators/README.md b/etc/validators/README.md index e89c42f0f7..d1d86f7a0d 100644 --- a/etc/validators/README.md +++ b/etc/validators/README.md @@ -100,10 +100,11 @@ Each validator declares: - `revalidate` — defaults to `always`: a document is validated again even when it has been seen before. Set it to `on_change` to skip the run when neither the document nor the validator configuration has changed since the stored - result. Only safe for backends whose verdict depends entirely on inputs we - can fingerprint, which excludes `conforma`: its policy lives on the remote - server and can change without any change here, so `on_change` would serve a - stale verdict. + result. Only safe for backends whose verdict depends entirely on inputs we + can fingerprint, which excludes `conforma`: its policy lives on the remote + server and can change without any change here, so `on_change` would serve a + stale verdict. Reports whose stored findings were truncated are revalidated + to preserve the complete findings in the ingest response. - `backend.url` — for the `conforma` backend, the base URL of the remote `ec validate input --server` instance. The client posts to `/v1/validate/input`; `backend.timeout_seconds` defaults to `120`. The server's diff --git a/migration/Cargo.toml b/migration/Cargo.toml index 9b961e4d5e..f8e7121ddf 100644 --- a/migration/Cargo.toml +++ b/migration/Cargo.toml @@ -16,7 +16,7 @@ long_running = [] # enable long-running tests [dependencies] trustify-common = { workspace = true } trustify-entity = { workspace = true } -trustify-module-ingestor = { workspace = true } +trustify-module-ingestor = { path = "../modules/ingestor", default-features = false } trustify-module-storage = { workspace = true } anyhow = { workspace = true } diff --git a/modules/fundamental/src/validation/endpoints/mod.rs b/modules/fundamental/src/validation/endpoints/mod.rs index a6b2986747..6a68609a20 100644 --- a/modules/fundamental/src/validation/endpoints/mod.rs +++ b/modules/fundamental/src/validation/endpoints/mod.rs @@ -55,7 +55,7 @@ pub async fn list_validation_reports( let tx = db.begin().await?; Ok(HttpResponse::Ok().json(match &filter.name { - Some(name) => state.get_by_name(name, paginated, &tx).await?, + Some(name) => state.get_by_name(name, search, paginated, &tx).await?, None => state.list(search, paginated, &tx).await?, })) } diff --git a/modules/fundamental/src/validation/endpoints/test.rs b/modules/fundamental/src/validation/endpoints/test.rs index 5f4a1978c3..2a05d803bb 100644 --- a/modules/fundamental/src/validation/endpoints/test.rs +++ b/modules/fundamental/src/validation/endpoints/test.rs @@ -137,5 +137,12 @@ async fn list_validation_reports_by_name(ctx: &TrustifyContext) -> Result<(), an assert_eq!(response.items.len(), 1); assert_eq!(response.items[0].document_sha256, digest); + let request = TestRequest::get() + .uri("/api/v3/validation?name=zookeeper&q=blocked%3Dtrue") + .to_request(); + let response: PaginatedResults = + app.call_and_read_body_json(request).await; + assert!(response.items.is_empty()); + Ok(()) } diff --git a/modules/fundamental/src/validation/service/mod.rs b/modules/fundamental/src/validation/service/mod.rs index 0248ea1cbd..6a3324609f 100644 --- a/modules/fundamental/src/validation/service/mod.rs +++ b/modules/fundamental/src/validation/service/mod.rs @@ -73,10 +73,12 @@ impl ValidationService { pub async fn get_by_name( &self, name: &str, + query: Query, paginated: impl Pagination, connection: &C, ) -> Result, Error> { let limiter = store::by_document_name(name) + .filtering(query)? .order_by_desc(validation_report::Column::CreatedAt) .limiting(connection, paginated, &self.cache)?; diff --git a/modules/ingestor/src/service/mod.rs b/modules/ingestor/src/service/mod.rs index a5c5262e4b..9786a4f058 100644 --- a/modules/ingestor/src/service/mod.rs +++ b/modules/ingestor/src/service/mod.rs @@ -19,8 +19,9 @@ use crate::{ service::{ dataset::{DatasetIngestResult, DatasetLoader}, validation::{ - Finding, InvocationError, OnError, Severity, ValidationMode, ValidationOutcome, - ValidationReport, Validator, ValidatorInput, store, store::Provenance, + Finding, InvocationError, OnError, Persistence, Revalidate, Severity, ValidationMode, + ValidationOutcome, ValidationReport, Validator, ValidatorInput, store, + store::Provenance, }, }, }; @@ -45,6 +46,7 @@ use trustify_common::{ id::IdError, }; use trustify_entity::labels::Labels; +use trustify_entity::validation_report; use trustify_module_analysis::service::AnalysisService; use trustify_module_storage::service::{StorageBackend, dispatch::DispatchBackend}; @@ -291,9 +293,36 @@ impl IngestorService { } /// Run all applicable validators against a document. - #[instrument(skip(self, bytes), fields(bytes = bytes.len()))] - async fn run_validators(&self, bytes: &[u8], fmt: Format) -> Vec { - run_validators(&self.validators, bytes, fmt).await + #[instrument(skip(self, bytes, conn), fields(bytes = bytes.len()))] + async fn run_validators( + &self, + bytes: &[u8], + fmt: Format, + conn: &C, + ) -> Vec { + let needs_cached_reports = self.validators.iter().any(|validator| { + validator.run_on_ingest() + && validator.applies_to(fmt) + && validator.persistence().persist + && validator.persistence().revalidate == Revalidate::OnChange + }); + + let previous = if needs_cached_reports { + let mut contexts = Contexts::new(); + contexts.update(bytes); + let digest = contexts.finish().sha256.encode_hex::(); + match store::latest_for_digest(conn, &digest).await { + Ok(reports) => reports, + Err(err) => { + tracing::warn!("failed to load cached validation reports: {err}"); + Vec::new() + } + } + } else { + Vec::new() + }; + + run_validators(&self.validators, bytes, fmt, &previous).await } /// Record the results that asked to be stored. @@ -396,7 +425,7 @@ impl IngestorService { // Run semantic validators before persisting anything. A blocking // (verify) failure returns an error here, so no bytes are stored and no // graph rows are created. See ADR 00021. - let validated = self.run_validators(bytes, fmt).await; + let validated = self.run_validators(bytes, fmt, tx).await; let blocked = blocking_reports(&validated); if !blocked.is_empty() { @@ -521,6 +550,7 @@ async fn run_validators( validators: &[Arc], bytes: &[u8], fmt: Format, + previous: &[store::StoredReport], ) -> Vec { if validators.is_empty() { return Vec::new(); @@ -535,6 +565,11 @@ async fn run_validators( } let persistence = validator.persistence(); + if let Some(cached) = cached_result(validator.as_ref(), persistence, previous) { + results.push(cached); + continue; + } + match validator.validate(&input).await { Ok(report) => { log_report(fmt, &report); @@ -575,6 +610,44 @@ async fn run_validators( results } +fn cached_result( + validator: &dyn Validator, + persistence: &Persistence, + previous: &[store::StoredReport], +) -> Option { + if !persistence.persist || persistence.revalidate != Revalidate::OnChange { + return None; + } + + let mode = match validator.mode() { + ValidationMode::Report => validation_report::Mode::Report, + ValidationMode::Verify => validation_report::Mode::Verify, + }; + let stored = previous.iter().find(|stored| { + stored.validator == validator.name() + && stored.mode == mode + && stored.config_fingerprint == persistence.fingerprint + && !stored.truncated + })?; + let findings = serde_json::from_value(stored.findings.clone()).ok()?; + let outcome = match stored.outcome { + validation_report::Outcome::Passed => ValidationOutcome::Passed, + validation_report::Outcome::Failed => ValidationOutcome::Failed, + }; + + Some(Validated { + report: ValidationReport { + validator: stored.validator.clone(), + findings, + outcome, + }, + mode: validator.mode(), + blocked: stored.blocked, + fingerprint: persistence.fingerprint.clone(), + persist: true, + }) +} + /// The reports that rejected the document, if any. fn blocking_reports(validated: &[Validated]) -> Vec { validated @@ -796,7 +869,7 @@ mod validation_tests { fn run(validators: Vec>) -> Result, Error> { let validated = tokio::runtime::Runtime::new() .expect("runtime") - .block_on(run_validators(&validators, b"{}", Format::CSAF)); + .block_on(run_validators(&validators, b"{}", Format::CSAF, &[])); let blocked = blocking_reports(&validated); match blocked.is_empty() { @@ -891,7 +964,7 @@ mod validation_tests { .run_on_ingest(false), )]; assert!( - run_validators(&validators, b"{}", Format::CSAF) + run_validators(&validators, b"{}", Format::CSAF, &[]) .await .is_empty(), "validator must be skipped during ingestion" @@ -917,6 +990,7 @@ mod persistence_tests { store::{Provenance, StoredReport}, }; use sea_orm::{EntityTrait, prelude::async_trait}; + use std::sync::atomic::{AtomicUsize, Ordering}; use test_context::test_context; use test_log::test; use trustify_entity::validation_report; @@ -928,6 +1002,7 @@ mod persistence_tests { mode: ValidationMode, outcome: ValidationOutcome, persistence: Persistence, + calls: Option>, } impl FixedValidator { @@ -936,6 +1011,7 @@ mod persistence_tests { mode, outcome, persistence: Persistence::default(), + calls: None, } } @@ -943,6 +1019,17 @@ mod persistence_tests { self.persistence.persist = persist; self } + + fn on_change(mut self) -> Self { + self.persistence.revalidate = Revalidate::OnChange; + self.persistence.fingerprint = "test-fingerprint".into(); + self + } + + fn counted(mut self, calls: Arc) -> Self { + self.calls = Some(calls); + self + } } #[async_trait::async_trait] @@ -969,6 +1056,9 @@ mod persistence_tests { &self, _input: &ValidatorInput<'_>, ) -> Result { + if let Some(calls) = &self.calls { + calls.fetch_add(1, Ordering::Relaxed); + } Ok(ValidationReport { validator: self.name().to_string(), findings: vec![], @@ -1101,4 +1191,26 @@ mod persistence_tests { Ok(()) } + + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn on_change_reuses_an_unchanged_report( + ctx: &TrustifyContext, + ) -> Result<(), anyhow::Error> { + let calls = Arc::new(AtomicUsize::new(0)); + for _ in 0..2 { + ingest( + ctx, + FixedValidator::new(ValidationMode::Report, ValidationOutcome::Passed) + .on_change() + .counted(Arc::clone(&calls)), + ) + .await? + .expect("ingest"); + } + + assert_eq!(calls.load(Ordering::Relaxed), 1); + assert_eq!(stored(ctx).await?.len(), 1); + Ok(()) + } } diff --git a/modules/ingestor/src/service/validation/config.rs b/modules/ingestor/src/service/validation/config.rs index 4089e005df..56fb7d9988 100644 --- a/modules/ingestor/src/service/validation/config.rs +++ b/modules/ingestor/src/service/validation/config.rs @@ -47,6 +47,8 @@ pub enum Revalidate { Always, /// Skip the run when the document and the validator configuration are /// unchanged since the stored result. + /// Truncated stored findings are always revalidated so the ingest response + /// can include the complete report. /// /// Only safe for backends whose verdict depends entirely on inputs we can /// fingerprint. A remote backend's policy can change without any change @@ -272,6 +274,11 @@ fn build_one( // Nothing here fingerprints the remote policy -- it is configured // on the Conforma server and can change without us knowing, which // is why `Revalidate::OnChange` is unsafe for this backend. + anyhow::ensure!( + validator.revalidate != Revalidate::OnChange, + "validator '{}' uses on_change with Conforma, whose policy cannot be fingerprinted", + validator.name, + ); let persistence = Persistence::new(global, validator, &[])?; Ok(Arc::new(conforma::build(validator, conforma, persistence)?)) } @@ -333,8 +340,8 @@ mod tests { ) .expect("parses"); - let validators = build(&config).expect("builds"); - assert!(!validators[0].persistence().persist); + let persistence = Persistence::new(&config, &config.validators[0], &[]).expect("builds"); + assert!(!persistence.persist); } #[test] @@ -429,6 +436,22 @@ validators: assert_eq!(conforma.timeout_seconds, 120); } + #[cfg(feature = "semantic-validation")] + #[test] + fn conforma_rejects_on_change_revalidation() { + let config: ValidatorsConfig = serde_json::from_value(serde_json::json!({ + "validators": [{ + "name": "policy-a", + "backend": { "type": "conforma", "url": "https://ec.example.test" }, + "revalidate": "on_change" + }] + })) + .expect("parses"); + + let error = build(&config).expect_err("Conforma cannot be cached safely"); + assert!(error.to_string().contains("on_change with Conforma")); + } + #[test] fn conforma_backend_requires_settings() { assert!( diff --git a/modules/ingestor/src/service/validation/store.rs b/modules/ingestor/src/service/validation/store.rs index ee633530ac..ade7c8117f 100644 --- a/modules/ingestor/src/service/validation/store.rs +++ b/modules/ingestor/src/service/validation/store.rs @@ -14,10 +14,10 @@ use crate::service::validation::{ Finding, Severity, ValidationMode, ValidationOutcome, ValidationReport, }; use sea_orm::{ - ActiveValue::Set, ColumnTrait, ConnectionTrait, DbErr, EntityTrait, JoinType, QueryFilter, - QueryOrder, QuerySelect, RelationTrait, Select, + ActiveValue::Set, ColumnTrait, ConnectionTrait, DatabaseBackend, DbErr, EntityTrait, JoinType, + QueryFilter, QuerySelect, RelationTrait, Select, Statement, }; -use sea_query::Condition; +use sea_query::{Condition, Expr, Query as SeaQuery}; use sha2::{Digest as _, Sha256}; use std::collections::HashMap; use time::OffsetDateTime; @@ -151,9 +151,9 @@ pub struct PendingReport<'a> { /// Store the reports whose result differs from the latest stored one. /// /// Returns the number of rows inserted; zero means every result was already -/// recorded. Reads the latest content hashes in one query and writes the new -/// rows in one statement, so the cost is at most two round trips per document -/// regardless of how many validators ran. +/// recorded. Uses a transaction-scoped advisory lock, reads the latest reports, +/// then writes any changed rows in one statement. Must be called inside a +/// transaction so concurrent ingests of the same document cannot duplicate rows. #[instrument(skip_all, fields(document = ctx.document_sha256), err(level = tracing::Level::INFO))] pub async fn persist_if_changed<'a, C: ConnectionTrait>( conn: &C, @@ -165,11 +165,15 @@ pub async fn persist_if_changed<'a, C: ConnectionTrait>( return Ok(0); } - let names = pending - .iter() - .map(|entry| entry.report.validator.as_str()) - .collect::>(); - let known = latest_content_hashes(conn, ctx.document_sha256, names).await?; + let lock = SeaQuery::select() + .expr(Expr::cust_with_values( + "pg_advisory_xact_lock(hashtextextended($1, 0))", + [ctx.document_sha256], + )) + .to_owned(); + conn.query_one(&lock).await?; + + let known = latest_content_hashes(conn, ctx.document_sha256).await?; let mut models = Vec::with_capacity(pending.len()); for entry in pending { @@ -191,32 +195,12 @@ pub async fn persist_if_changed<'a, C: ConnectionTrait>( } /// The latest content hash per validator for one document. -async fn latest_content_hashes<'a, C: ConnectionTrait>( +async fn latest_content_hashes( conn: &C, document_sha256: &str, - validators: impl IntoIterator, ) -> Result, Error> { - /// Only the columns needed to decide whether a result changed — notably - /// not `findings`, which can be tens of kilobytes per row. - #[derive(Debug, sea_orm::FromQueryResult)] - struct Row { - validator: String, - content_hash: String, - } - - let rows = validation_report::Entity::find() - .select_only() - .column(validation_report::Column::Validator) - .column(validation_report::Column::ContentHash) - .filter(validation_report::Column::DocumentSha256.eq(document_sha256)) - .filter(validation_report::Column::Validator.is_in(validators)) - // Newest last, so later rows overwrite earlier ones in the map below. - .order_by_asc(validation_report::Column::CreatedAt) - .into_model::() - .all(conn) - .await?; - - Ok(rows + Ok(latest_for_digest(conn, document_sha256) + .await? .into_iter() .map(|row| (row.validator, row.content_hash)) .collect()) @@ -231,16 +215,17 @@ pub async fn latest_for_digest( conn: &C, document_sha256: &str, ) -> Result, Error> { - let mut latest = HashMap::new(); - for report in by_digest(document_sha256) - .order_by_asc(validation_report::Column::CreatedAt) + let rows = validation_report::Entity::find() + .from_raw_sql(Statement::from_sql_and_values( + DatabaseBackend::Postgres, + "SELECT DISTINCT ON (validator) * FROM validation_report \ + WHERE document_sha256 = $1 ORDER BY validator, created_at DESC, id DESC", + [document_sha256.to_owned().into()], + )) .all(conn) - .await? - { - latest.insert(report.validator.clone(), report); - } + .await?; - let mut reports = latest.into_values().collect::>(); + let mut reports = rows; reports.sort_by(|left, right| left.validator.cmp(&right.validator)); Ok(reports) } @@ -482,19 +467,28 @@ mod tests { } } + async fn persist( + ctx: &TrustifyContext, + persist_context: PersistContext<'_>, + pending: PendingReport<'_>, + ) -> Result { + Ok(ctx + .db + .transaction(async |tx| persist_if_changed(tx, &persist_context, [pending]).await) + .await?) + } + #[test_context(TrustifyContext)] #[test(tokio::test)] async fn unchanged_result_writes_nothing(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { let report = report(vec![], ValidationOutcome::Passed); - let inserted = - persist_if_changed(&ctx.db, &context(DIGEST), [pending(&report, "fp-1")]).await?; + let inserted = persist(ctx, context(DIGEST), pending(&report, "fp-1")).await?; assert_eq!(inserted, 1); // The same result from a later run must not produce a second row: this // is what keeps repeated importer ingests off the write path. - let inserted = - persist_if_changed(&ctx.db, &context(DIGEST), [pending(&report, "fp-1")]).await?; + let inserted = persist(ctx, context(DIGEST), pending(&report, "fp-1")).await?; assert_eq!(inserted, 0); assert_eq!(by_digest(DIGEST).all(&ctx.db).await?.len(), 1); @@ -502,20 +496,42 @@ mod tests { Ok(()) } + #[test_context(TrustifyContext)] + #[test(tokio::test)] + async fn concurrent_unchanged_results_are_inserted_once( + ctx: &TrustifyContext, + ) -> Result<(), anyhow::Error> { + let digest = DIGEST.replace('1', "9"); + let persist_context = context(&digest); + let report = report(vec![], ValidationOutcome::Passed); + let pending = pending(&report, "fp-1"); + + let first = ctx + .db + .transaction(async |tx| persist_if_changed(tx, &persist_context, [pending]).await); + let second = ctx + .db + .transaction(async |tx| persist_if_changed(tx, &persist_context, [pending]).await); + let (first, second) = tokio::join!(first, second); + + assert_eq!(first? + second?, 1); + assert_eq!(by_digest(&digest).all(&ctx.db).await?.len(), 1); + Ok(()) + } + #[test_context(TrustifyContext)] #[test(tokio::test)] async fn changed_result_appends(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { let digest = &DIGEST.replace("1", "2"); let passed = report(vec![], ValidationOutcome::Passed); - persist_if_changed(&ctx.db, &context(digest), [pending(&passed, "fp-1")]).await?; + persist(ctx, context(digest), pending(&passed, "fp-1")).await?; let failed = report( vec![finding(Severity::Error, "missing field")], ValidationOutcome::Failed, ); - let inserted = - persist_if_changed(&ctx.db, &context(digest), [pending(&failed, "fp-1")]).await?; + let inserted = persist(ctx, context(digest), pending(&failed, "fp-1")).await?; assert_eq!(inserted, 1); // History is kept, and the newer verdict is the current one. @@ -539,12 +555,11 @@ mod tests { let digest = &DIGEST.replace("1", "3"); let report = report(vec![], ValidationOutcome::Passed); - persist_if_changed(&ctx.db, &context(digest), [pending(&report, "fp-1")]).await?; + persist(ctx, context(digest), pending(&report, "fp-1")).await?; // Same verdict, different ruleset: the row records which configuration // produced it, so a new row is required to stay accurate. - let inserted = - persist_if_changed(&ctx.db, &context(digest), [pending(&report, "fp-2")]).await?; + let inserted = persist(ctx, context(digest), pending(&report, "fp-2")).await?; assert_eq!(inserted, 1); Ok(()) @@ -564,7 +579,7 @@ mod tests { max_findings: 3, max_findings_bytes: 64 * 1024, }; - persist_if_changed(&ctx.db, &ctx_with_caps, [pending(&report, "fp-1")]).await?; + persist(ctx, ctx_with_caps, pending(&report, "fp-1")).await?; let stored = latest_for_digest(&ctx.db, digest).await?; assert!(stored[0].truncated); @@ -587,7 +602,7 @@ mod tests { .expect("ingested document has a source document"); let report = report(vec![], ValidationOutcome::Passed); - persist_if_changed(&ctx.db, &context(&digest), [pending(&report, "fp-1")]).await?; + persist(ctx, context(&digest), pending(&report, "fp-1")).await?; let by_id = by_document_id(Id::Uuid(Uuid::parse_str(&result.id)?))? .all(&ctx.db) diff --git a/test-context/Cargo.toml b/test-context/Cargo.toml index 88efa8ea7d..4087849b1e 100644 --- a/test-context/Cargo.toml +++ b/test-context/Cargo.toml @@ -12,7 +12,7 @@ trustify-common = { workspace = true } trustify-db = { workspace = true } trustify-entity = { workspace = true } trustify-migration = { workspace = true } -trustify-module-ingestor = { workspace = true } +trustify-module-ingestor = { path = "../modules/ingestor", default-features = false } trustify-module-storage = { workspace = true } actix-http = { workspace = true }