diff --git a/Cargo.lock b/Cargo.lock index c110546e6..f05f8af59 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6006,6 +6006,7 @@ dependencies = [ "js-sys", "log", "mime", + "mime_guess", "native-tls", "percent-encoding", "pin-project-lite", @@ -8510,6 +8511,7 @@ dependencies = [ "actix-web", "anyhow", "async-trait", + "backon", "bytes", "bytesize", "chrono", @@ -8534,6 +8536,7 @@ dependencies = [ "osv", "packageurl", "regex", + "reqwest 0.13.4", "roxmltree", "rstest", "sanitize-filename", @@ -8791,6 +8794,7 @@ dependencies = [ "clap", "futures", "garage-door", + "humantime", "log", "rand 0.10.2", "rstest", diff --git a/Cargo.toml b/Cargo.toml index 47c750262..3491c796f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,6 +48,7 @@ aws-config = { version = "1.8.14", features = ["behavior-version-latest"] } aws-sdk-s3 = { version = "1.124.0", default-features = false, features = ["behavior-version-latest", "default-https-client", "http-1x", "rt-tokio", "sigv4a"] } aws-smithy-http-client = { version = "1.2.0", features = ["rustls-aws-lc"] } aws-smithy-types = { version = "1.4.5" } +backon = "1.6" base16ct = "1" base64 = "0.22" build-info = "0.0.44" diff --git a/common/auth/src/permission.rs b/common/auth/src/permission.rs index 57e8e1740..f0d8c6375 100644 --- a/common/auth/src/permission.rs +++ b/common/auth/src/permission.rs @@ -117,6 +117,11 @@ permission! { #[strum(serialize = "ai")] Ai, + + #[strum(serialize = "create.exploitIntelligence")] + CreateExploitIntelligence, + #[strum(serialize = "read.exploitIntelligence")] + ReadExploitIntelligence, } } diff --git a/docs/env-vars.md b/docs/env-vars.md index f5dd5489c..498fc0fb7 100644 --- a/docs/env-vars.md +++ b/docs/env-vars.md @@ -62,6 +62,19 @@ | `UI_ISSUER_URL` | Issuer URL used by the UI | `http://localhost:8090/realms/trustify` | | `UI_LOAD_USER` | Whether to load user info | `true` | | `UI_SCOPE` | Scopes to request | `openid` | +| `EXPLOIT_INTELLIGENCE_URL` | Base URL of the Exploit Intelligence client service | | +| `EXPLOIT_INTELLIGENCE_UI_URL` | Base URL of the EI web UI for deep-linking to reports | | +| `EXPLOIT_INTELLIGENCE_POLL_INTERVAL`| Polling interval for EI analysis completion (humantime) | `30s` | +| `EXPLOIT_INTELLIGENCE_MAX_POLL_DURATION`| Maximum duration before EI polling is considered timed out (humantime) | `30m` | +| `EXPLOIT_INTELLIGENCE_UPLOAD_MAX_RETRIES`| Maximum number of upload retry attempts | `3` | +| `EXPLOIT_INTELLIGENCE_UPLOAD_RETRY_DELAY`| Initial delay between upload retries (humantime, exponential backoff) | `1s` | +| `EXPLOIT_INTELLIGENCE_MAX_CONSECUTIVE_POLL_FAILURES`| Maximum consecutive poll failures before giving up | `5` | +| `EXPLOIT_INTELLIGENCE_AUTH_TOKEN` | Static authentication token for the Exploit Intelligence service (used when OIDC is not configured) | | +| `EXPLOIT_INTELLIGENCE_OIDC_ISSUER_URL` | OIDC issuer URL for EI service authentication (client credentials flow with discovery) | | +| `EXPLOIT_INTELLIGENCE_OIDC_CLIENT_ID` | OIDC client ID for EI service authentication | | +| `EXPLOIT_INTELLIGENCE_OIDC_CLIENT_SECRET`| OIDC client secret for EI service authentication | | +| `EXPLOIT_INTELLIGENCE_OIDC_REFRESH_BEFORE`| Duration an EI access token must still be valid before requesting a new one | `30s` | +| `EXPLOIT_INTELLIGENCE_OIDC_TLS_INSECURE`| Allow insecure TLS connections with the EI OIDC issuer | `false` | ## Data Migration diff --git a/entity/src/exploit_intelligence_job.rs b/entity/src/exploit_intelligence_job.rs index 1384014ac..bee025ae4 100644 --- a/entity/src/exploit_intelligence_job.rs +++ b/entity/src/exploit_intelligence_job.rs @@ -72,6 +72,13 @@ pub struct Model { /// Expected number of components in the product. pub total_components: Option, + /// Visibility timeout for the job queue pattern. When set, the job is + /// invisible to other workers until this timestamp is reached. + pub visible_at: Option, + + /// Number of times this job has been dequeued for processing. + pub retry_count: i32, + /// Timestamp when the job was created. pub created: OffsetDateTime, diff --git a/entity/src/exploit_intelligence_job_component.rs b/entity/src/exploit_intelligence_job_component.rs index 0dfcaf9e9..ccc3406f3 100644 --- a/entity/src/exploit_intelligence_job_component.rs +++ b/entity/src/exploit_intelligence_job_component.rs @@ -35,6 +35,10 @@ pub struct Model { /// FK to advisory — set when a VEX document is ingested from the result. pub advisory_id: Option, + /// Whether this component was excluded by the EI service (e.g., + /// unsupported ecosystem) rather than genuinely failing. + pub excluded: bool, + /// Error details for failed analyses. pub error_message: Option, diff --git a/migration/src/lib.rs b/migration/src/lib.rs index acc6bd056..da5917bf4 100644 --- a/migration/src/lib.rs +++ b/migration/src/lib.rs @@ -66,7 +66,7 @@ mod m0002210_sbom_node_name_index; mod m0002220_drop_qualified_purl_gist_indexes; mod m0002230_sle_license_id_index; mod m0002240_product_version_sbom_index; -mod m0002230_create_exploit_intelligence_job; +mod m0002250_create_exploit_intelligence_job; pub trait MigratorExt: Send { fn build_migrations() -> Migrations; @@ -148,7 +148,7 @@ impl MigratorExt for Migrator { .normal(m0002220_drop_qualified_purl_gist_indexes::Migration) .normal(m0002230_sle_license_id_index::Migration) .normal(m0002240_product_version_sbom_index::Migration) - .normal(m0002230_create_exploit_intelligence_job::Migration) + .normal(m0002250_create_exploit_intelligence_job::Migration) } } diff --git a/migration/src/m0002230_create_exploit_intelligence_job.rs b/migration/src/m0002250_create_exploit_intelligence_job.rs similarity index 88% rename from migration/src/m0002230_create_exploit_intelligence_job.rs rename to migration/src/m0002250_create_exploit_intelligence_job.rs index 0f5bfbd7b..2660834d4 100644 --- a/migration/src/m0002230_create_exploit_intelligence_job.rs +++ b/migration/src/m0002250_create_exploit_intelligence_job.rs @@ -57,6 +57,17 @@ impl MigrationTrait for Migration { .col(ColumnDef::new(ExploitIntelligenceJob::ErrorMessage).string()) .col(ColumnDef::new(ExploitIntelligenceJob::ProductId).string()) .col(ColumnDef::new(ExploitIntelligenceJob::TotalComponents).integer()) + .col( + ColumnDef::new(ExploitIntelligenceJob::VisibleAt) + .timestamp_with_time_zone() + .null(), + ) + .col( + ColumnDef::new(ExploitIntelligenceJob::RetryCount) + .integer() + .not_null() + .default(0), + ) .col( ColumnDef::new(ExploitIntelligenceJob::Created) .timestamp_with_time_zone() @@ -102,6 +113,18 @@ impl MigrationTrait for Migration { ) .await?; + // Partial unique index to prevent duplicate active jobs for the same + // (sbom_id, vulnerability_id) pair. SeaORM's Index builder does not + // support partial indexes, so we use raw SQL. + manager + .get_connection() + .execute_unprepared( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_ei_job_active_dedup \ + ON exploit_intelligence_job (sbom_id, vulnerability_id) \ + WHERE status IN ('pending', 'running')", + ) + .await?; + manager .create_table( Table::create() @@ -138,6 +161,12 @@ impl MigrationTrait for Migration { .custom(EiFinding::Table), ) .col(ColumnDef::new(ExploitIntelligenceJobComponent::AdvisoryId).uuid()) + .col( + ColumnDef::new(ExploitIntelligenceJobComponent::Excluded) + .boolean() + .not_null() + .default(false), + ) .col(ColumnDef::new(ExploitIntelligenceJobComponent::ErrorMessage).string()) .col( ColumnDef::new(ExploitIntelligenceJobComponent::Created) @@ -222,6 +251,11 @@ impl MigrationTrait for Migration { ) .await?; + manager + .get_connection() + .execute_unprepared("DROP INDEX IF EXISTS idx_ei_job_active_dedup") + .await?; + manager .drop_index( Index::drop() @@ -287,6 +321,8 @@ enum ExploitIntelligenceJob { ErrorMessage, ProductId, TotalComponents, + VisibleAt, + RetryCount, Created, Updated, } @@ -301,6 +337,7 @@ enum ExploitIntelligenceJobComponent { Status, Finding, AdvisoryId, + Excluded, ErrorMessage, Created, Updated, diff --git a/modules/fundamental/Cargo.toml b/modules/fundamental/Cargo.toml index 71823b7f2..7278d96c9 100644 --- a/modules/fundamental/Cargo.toml +++ b/modules/fundamental/Cargo.toml @@ -21,6 +21,8 @@ trustify-query-derive = { workspace = true } actix-http = { workspace = true } actix-web = { workspace = true } +backon = { workspace = true } +bytes = { workspace = true } anyhow = { workspace = true } async-trait = { workspace = true } cpe = { workspace = true } @@ -30,6 +32,7 @@ futures-util = { workspace = true } isx = { workspace = true } itertools = { workspace = true } log = { workspace = true } +reqwest = { workspace = true, features = ["form", "json", "multipart"] } sanitize-filename = { workspace = true } sea-orm = { workspace = true } sea-query = { workspace = true } @@ -43,8 +46,10 @@ tar = { workspace = true } thiserror = { workspace = true } time = { workspace = true } tokio = { workspace = true, features = ["full"] } +tokio-util = { workspace = true, features = ["full"] } tracing = { workspace = true } tracing-futures = { workspace = true, features = ["futures-03"] } +urlencoding = { workspace = true } utoipa = { workspace = true, features = ["actix_extras", "uuid", "time"] } utoipa-actix-web = { workspace = true } uuid = { workspace = true } @@ -86,7 +91,6 @@ strum = { workspace = true } tar = { workspace = true } test-context = { workspace = true } test-log = { workspace = true, features = ["log", "trace"] } -tokio-util = { workspace = true } tracing = { workspace = true, features = ["std"]} tracing-core = { workspace = true } tracing-opentelemetry = { workspace = true } diff --git a/modules/fundamental/src/endpoints.rs b/modules/fundamental/src/endpoints.rs index 960436618..1e525e468 100644 --- a/modules/fundamental/src/endpoints.rs +++ b/modules/fundamental/src/endpoints.rs @@ -6,6 +6,8 @@ use trustify_module_ingestor::service::IngestorService; use trustify_module_storage::service::dispatch::DispatchBackend; use utoipa::{IntoParams, ToSchema}; +use crate::exploit_intelligence::service::{ExploitIntelligenceConfig, ExploitIntelligenceService}; + #[derive(Clone, Debug, Eq, PartialEq, Default)] pub struct Config { pub sbom_upload_limit: usize, @@ -13,6 +15,7 @@ pub struct Config { pub max_group_name_length: usize, } +#[allow(clippy::too_many_arguments)] pub fn configure( svc: &mut utoipa_actix_web::service_config::ServiceConfig, config: Config, @@ -21,9 +24,10 @@ pub fn configure( storage: impl Into, analysis: AnalysisService, cache: PaginationCache, + ei_config: Option, ) { let ingestor_service = IngestorService::new(Graph::new(), storage, Some(analysis)); - svc.app_data(web::Data::new(ingestor_service)); + svc.app_data(web::Data::new(ingestor_service.clone())); crate::advisory::endpoints::configure( svc, @@ -45,7 +49,17 @@ pub fn configure( ); crate::vulnerability::endpoints::configure(svc, db_ro.clone(), cache.clone()); crate::weakness::endpoints::configure(svc, db_ro.clone(), cache.clone()); - crate::sbom_group::endpoints::configure(svc, db_rw, db_ro, config.max_group_name_length, cache); + crate::sbom_group::endpoints::configure( + svc, + db_rw.clone(), + db_ro.clone(), + config.max_group_name_length, + cache, + ); + + let ei_service = ExploitIntelligenceService::new(ei_config) + .unwrap_or_else(|e| panic!("failed to create ExploitIntelligenceService: {e}")); + crate::exploit_intelligence::endpoints::configure(svc, db_rw, db_ro, ei_service); } #[derive(Clone, Debug, PartialEq, Eq, Default, ToSchema, serde::Deserialize, IntoParams)] diff --git a/modules/fundamental/src/exploit_intelligence/auth.rs b/modules/fundamental/src/exploit_intelligence/auth.rs new file mode 100644 index 000000000..3d5342e54 --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/auth.rs @@ -0,0 +1,19 @@ +use std::sync::Arc; +use trustify_auth::client::{BearerTokenProvider, OpenIdTokenProviderConfig, TokenProvider}; + +/// Build an EI [`TokenProvider`] from OIDC configuration, falling back to a +/// static token if OIDC is not configured. +/// +/// Returns `None` when neither OIDC nor a static token is available. +pub async fn build_provider( + oidc_config: Option, + static_token: Option, +) -> anyhow::Result>> { + match oidc_config { + Some(config) => Ok(Some( + OpenIdTokenProviderConfig::new_provider(Some(config)).await?, + )), + None => Ok(static_token + .map(|t| -> Arc { Arc::new(BearerTokenProvider { token: t }) })), + } +} diff --git a/modules/fundamental/src/exploit_intelligence/ei_client.rs b/modules/fundamental/src/exploit_intelligence/ei_client.rs new file mode 100644 index 000000000..1bd20afc7 --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/ei_client.rs @@ -0,0 +1,132 @@ +use serde::Deserialize; + +/// Response from POST /api/v1/products/upload-cyclonedx +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct UploadResponse { + pub report_request_id: ReportRequestId, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ReportRequestId { + pub report_id: String, +} + +/// Response from GET /api/v1/reports/by-scan-id/{scan_id} +#[derive(Debug, Deserialize)] +pub struct PollResponse { + pub status: String, + pub report: Option, +} + +#[derive(Debug, Deserialize)] +pub struct Report { + pub output: Option, + pub error: Option, +} + +#[derive(Debug, Deserialize)] +pub struct ReportError { + #[serde(rename = "type")] + pub error_type: Option, + pub message: Option, +} + +#[derive(Debug, Deserialize)] +pub struct ReportOutput { + pub analysis: Option>, + /// VEX kept as `serde_json::Value` since it is forwarded directly to the ingestor. + pub vex: Option, +} + +#[derive(Debug, Deserialize)] +pub struct AnalysisEntry { + pub justification: Option, +} + +#[derive(Debug, Deserialize)] +pub struct Justification { + pub status: Option, +} + +// ── SPDX multi-component product flow ──────────────────────────────── + +/// Response from POST /api/v1/products/upload-spdx +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SpdxUploadResponse { + pub product_id: String, +} + +/// Response from GET /api/v1/products/{productId} +#[derive(Debug, Deserialize)] +pub struct ProductStatusResponse { + pub data: ProductData, + pub summary: ProductSummary, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ProductData { + #[serde(alias = "id")] + pub product_id: String, + pub name: Option, + pub version: Option, + #[serde(default)] + pub components: Vec, + #[serde(default)] + pub submission_failures: Vec, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ProductComponent { + pub component_id: Option, + pub name: String, + pub version: Option, + pub purl: Option, + pub scan_id: Option, + pub state: Option, +} + +#[derive(Debug, Deserialize)] +pub struct SubmissionFailure { + pub name: Option, + pub version: Option, + pub image: Option, + pub error: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ProductSummary { + pub product_state: String, + #[serde(default)] + pub status_counts: std::collections::HashMap, + #[serde(default)] + pub justification_status_counts: std::collections::HashMap, +} + +/// Entry from GET /api/v1/reports?productId={id}. +/// +/// Each entry represents a single component's analysis report within a product. +/// The `vulns` array contains the per-CVE justification from the EI engine. +/// Note: this endpoint does NOT include the full report output (no VEX data). +/// For VEX ingestion, use `GET /api/v1/reports/by-scan-id/{scanId}` per component. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ProductReportEntry { + pub scan_id: String, + pub state: String, + pub image_name: String, + #[serde(default)] + pub vulns: Vec, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ReportVuln { + pub vuln_id: Option, + pub justification: Option, +} diff --git a/modules/fundamental/src/exploit_intelligence/endpoints/mod.rs b/modules/fundamental/src/exploit_intelligence/endpoints/mod.rs new file mode 100644 index 000000000..2b4f67341 --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/endpoints/mod.rs @@ -0,0 +1,167 @@ +#[cfg(test)] +mod test; + +use crate::{ + Error, + exploit_intelligence::{ + model::{ + AnalyzeRequest, AnalyzeResponse, ExploitIntelligenceJobDetails, + ExploitIntelligenceJobSummary, JobStatus, + }, + service::ExploitIntelligenceService, + }, +}; +use actix_web::{HttpResponse, Responder, get, post, web}; +use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, TransactionTrait}; +use trustify_auth::{CreateExploitIntelligence, ReadExploitIntelligence, authorizer::Require}; +use trustify_common::{ + db, + model::{Paginated, PaginatedResults}, +}; +use uuid::Uuid; + +/// Register exploit intelligence endpoints. +pub fn configure( + config: &mut utoipa_actix_web::service_config::ServiceConfig, + db_rw: db::ReadWrite, + db_ro: db::ReadOnly, + ei_service: ExploitIntelligenceService, +) { + config + .app_data(web::Data::new(db_rw)) + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(analyze) + .service(list_jobs) + .service(get_job); +} + +/// Query parameters for filtering the job list. +#[derive(Debug, Clone, serde::Deserialize, utoipa::IntoParams)] +pub struct JobFilter { + /// Filter by SBOM ID. + pub sbom_id: Option, + /// Filter by job status. + pub status: Option, +} + +/// Trigger an Exploit Intelligence analysis for a CVE against an SBOM. +/// +/// Creates a job record that the background worker will pick up and +/// process. Returns immediately after persisting the job. +#[utoipa::path( + tag = "exploit-intelligence", + operation_id = "analyzeExploitIntelligence", + request_body = AnalyzeRequest, + responses( + (status = 200, description = "Active analysis job already exists", body = AnalyzeResponse), + (status = 201, description = "Analysis job created", body = AnalyzeResponse), + (status = 404, description = "SBOM not found"), + (status = 503, description = "Exploit Intelligence service not configured"), + ), +)] +#[post("/v3/exploit-intelligence/analyze")] +pub async fn analyze( + ei_service: web::Data, + db_rw: web::Data, + body: web::Json, + _: Require, +) -> actix_web::Result { + // Verify EI is configured before doing any work. + let _ = ei_service.runtime().ok_or(Error::Unavailable)?; + + // Use a single read-write transaction for the existence check, active-job + // deduplication, and job creation to prevent TOCTOU race conditions where + // two concurrent requests could both pass the check and create duplicate jobs. + let tx = db_rw.begin().await.map_err(Error::from)?; + + // Verify the SBOM exists. + let sbom = trustify_entity::sbom::Entity::find() + .filter(trustify_entity::sbom::Column::SbomId.eq(body.sbom_id)) + .one(&tx) + .await + .map_err(Error::from)?; + + if sbom.is_none() { + return Ok(HttpResponse::NotFound().finish()); + } + + // Check for an existing active job (deduplication). + if let Some(active_job) = ei_service + .find_active_job(body.sbom_id, &body.vulnerability_id, &tx) + .await? + { + // No need to commit -- no mutations were made. + return Ok(HttpResponse::Ok().json(AnalyzeResponse { + job_id: active_job.id, + status: active_job.status, + })); + } + + // Create the job record within the same transaction. + let job = ei_service + .create_job(body.sbom_id, &body.vulnerability_id, &tx) + .await?; + + tx.commit().await.map_err(Error::from)?; + + Ok(HttpResponse::Created().json(AnalyzeResponse { + job_id: job.id, + status: JobStatus::Pending, + })) +} + +/// List Exploit Intelligence analysis jobs. +#[utoipa::path( + tag = "exploit-intelligence", + operation_id = "listExploitIntelligenceJobs", + params( + Paginated, + JobFilter, + ), + responses( + (status = 200, description = "List of analysis jobs", body = PaginatedResults), + ), +)] +#[get("/v3/exploit-intelligence/jobs")] +pub async fn list_jobs( + ei_service: web::Data, + db: web::Data, + web::Query(paginated): web::Query, + web::Query(filter): web::Query, + _: Require, +) -> actix_web::Result { + let tx = db.begin().await?; + let result = ei_service + .fetch_jobs(paginated, filter.sbom_id, filter.status, &tx) + .await?; + Ok(HttpResponse::Ok().json(result)) +} + +/// Get details of a specific Exploit Intelligence analysis job. +#[utoipa::path( + tag = "exploit-intelligence", + operation_id = "getExploitIntelligenceJob", + params( + ("id" = Uuid, Path, description = "Job identifier"), + ), + responses( + (status = 200, description = "Job details", body = ExploitIntelligenceJobDetails), + (status = 404, description = "Job not found"), + ), +)] +#[get("/v3/exploit-intelligence/jobs/{id}")] +pub async fn get_job( + ei_service: web::Data, + db: web::Data, + id: web::Path, + _: Require, +) -> actix_web::Result { + let tx = db.begin().await?; + let job = ei_service.fetch_job(*id, &tx).await?; + + match job { + Some(details) => Ok(HttpResponse::Ok().json(details)), + None => Ok(HttpResponse::NotFound().finish()), + } +} diff --git a/modules/fundamental/src/exploit_intelligence/endpoints/test.rs b/modules/fundamental/src/exploit_intelligence/endpoints/test.rs new file mode 100644 index 000000000..6dc1b8ed9 --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/endpoints/test.rs @@ -0,0 +1,789 @@ +use crate::exploit_intelligence::{ + model::{ + AnalyzeRequest, AnalyzeResponse, ExploitIntelligenceJobDetails, + ExploitIntelligenceJobSummary, JobFinding, JobStatus, + }, + service::{ExploitIntelligenceConfig, ExploitIntelligenceService}, +}; +use actix_web::{ + App, + test::{self, TestRequest}, + web, +}; +use sea_orm::{ActiveModelTrait, Set, TransactionTrait}; +use test_context::test_context; +use test_log::test; +use time::OffsetDateTime; +use trustify_common::{db, model::PaginatedResults}; +use trustify_entity::exploit_intelligence_job::{self, ExploitIntelligenceJobStatus}; +use trustify_test_context::{TrustifyContext, app::TestApp}; +use uuid::Uuid; + +/// Ingest an SBOM document and return its UUID. +async fn ingest_and_parse_sbom_id(ctx: &TrustifyContext, path: &str) -> Uuid { + let result = ctx.ingest_document(path).await.expect("ingest test SBOM"); + let id = result + .id + .strip_prefix("urn:uuid:") + .unwrap_or(&result.id); + Uuid::parse_str(id).expect("parse sbom UUID") +} + +/// Helper to create a test service with a fake EI backend URL. +fn test_service() -> ExploitIntelligenceService { + ExploitIntelligenceService::new(Some(ExploitIntelligenceConfig { + url: "http://localhost:9999".to_string(), + ui_url: Some("http://localhost:9999".to_string()), + poll_interval: std::time::Duration::from_secs(1), + max_poll_duration: std::time::Duration::from_secs(1800), + upload_max_retries: 3, + upload_retry_delay: std::time::Duration::from_secs(1), + max_consecutive_poll_failures: 5, + token_provider: None, + })) + .unwrap() +} + +/// Helper to insert a job record directly for testing query endpoints. +async fn insert_test_job( + db: &trustify_common::db::Database, + id: Uuid, + sbom_id: Uuid, + vulnerability_id: &str, + status: ExploitIntelligenceJobStatus, + error_message: Option<&str>, +) -> exploit_intelligence_job::Model { + let now = OffsetDateTime::now_utc(); + let model = exploit_intelligence_job::ActiveModel { + id: Set(id), + sbom_id: Set(sbom_id), + vulnerability_id: Set(vulnerability_id.to_string()), + status: Set(status), + error_message: Set(error_message.map(|s| s.to_string())), + product_id: Set(None), + total_components: Set(None), + visible_at: Set(None), + retry_count: Set(0), + created: Set(now), + updated: Set(now), + }; + + let tx = db.begin().await.expect("begin tx"); + let result = model.insert(&tx).await.expect("insert job"); + tx.commit().await.expect("commit tx"); + result +} + +/// Helper to insert a component record for a job, with optional finding. +async fn insert_test_component( + db: &trustify_common::db::Database, + job_id: Uuid, + component_ref: &str, + status: ExploitIntelligenceJobStatus, + finding: Option, +) -> trustify_entity::exploit_intelligence_job_component::Model { + use trustify_entity::exploit_intelligence_job_component; + + let now = OffsetDateTime::now_utc(); + let model = exploit_intelligence_job_component::ActiveModel { + id: Set(Uuid::now_v7()), + job_id: Set(job_id), + component_ref: Set(component_ref.to_string()), + scan_id: Set(Some(format!("scan-{job_id}"))), + status: Set(status), + finding: Set(finding), + advisory_id: Set(None), + excluded: Set(false), + error_message: Set(None), + created: Set(now), + updated: Set(now), + }; + + let tx = db.begin().await.expect("begin tx"); + let result = model.insert(&tx).await.expect("insert component"); + tx.commit().await.expect("commit tx"); + result +} + +/// Verifies that the GET job details endpoint returns the correct state +/// for a completed job. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn get_job_completed(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given a completed job with a component record in the database + let sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + let job_id = Uuid::now_v7(); + insert_test_job( + &ctx.db, + job_id, + sbom_id, + "CVE-2024-1234", + ExploitIntelligenceJobStatus::Completed, + None, + ) + .await; + insert_test_component( + &ctx.db, + job_id, + "test-component", + ExploitIntelligenceJobStatus::Completed, + Some(exploit_intelligence_job::Finding::NotVulnerable), + ) + .await; + + let ei_service = test_service(); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::get_job), + ) + .await; + + // When requesting the job details + let req = TestRequest::get() + .uri(&format!("/v3/exploit-intelligence/jobs/{job_id}")) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then the response contains the completed job details + assert_eq!(resp.status(), 200); + let body: ExploitIntelligenceJobDetails = test::read_body_json(resp).await; + assert_eq!(body.id, job_id); + assert_eq!(body.status, JobStatus::Completed); + assert_eq!(body.vulnerability_id, "CVE-2024-1234"); + assert_eq!(body.finding, Some(JobFinding::NotVulnerable)); + + Ok(()) +} + +/// Verifies that the GET job details endpoint returns the correct state +/// for a pending job. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn get_job_pending(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given a pending job record in the database + let sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + let job_id = Uuid::now_v7(); + insert_test_job( + &ctx.db, + job_id, + sbom_id, + "CVE-2024-5678", + ExploitIntelligenceJobStatus::Pending, + None, + ) + .await; + + let ei_service = test_service(); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::get_job), + ) + .await; + + // When requesting the job details + let req = TestRequest::get() + .uri(&format!("/v3/exploit-intelligence/jobs/{job_id}")) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then the response shows the pending status + assert_eq!(resp.status(), 200); + let body: ExploitIntelligenceJobDetails = test::read_body_json(resp).await; + assert_eq!(body.id, job_id); + assert_eq!(body.status, JobStatus::Pending); + assert_eq!(body.finding, None); + + Ok(()) +} + +/// Verifies that the GET job details endpoint returns the correct state +/// for a failed job. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn get_job_failed(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given a failed job record in the database + let sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + let job_id = Uuid::now_v7(); + insert_test_job( + &ctx.db, + job_id, + sbom_id, + "CVE-2024-9999", + ExploitIntelligenceJobStatus::Failed, + Some("analysis timed out"), + ) + .await; + + let ei_service = test_service(); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::get_job), + ) + .await; + + // When requesting the job details + let req = TestRequest::get() + .uri(&format!("/v3/exploit-intelligence/jobs/{job_id}")) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then the response shows the failed status with error details + assert_eq!(resp.status(), 200); + let body: ExploitIntelligenceJobDetails = test::read_body_json(resp).await; + assert_eq!(body.id, job_id); + assert_eq!(body.status, JobStatus::Failed); + assert_eq!(body.error_message.as_deref(), Some("analysis timed out")); + + Ok(()) +} + +/// Verifies that requesting a non-existent job returns 404. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn get_job_not_found(ctx: &TrustifyContext) -> anyhow::Result<()> { + let ei_service = test_service(); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::get_job), + ) + .await; + + // When requesting a non-existent job + let req = TestRequest::get() + .uri(&format!("/v3/exploit-intelligence/jobs/{}", Uuid::now_v7())) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then a 404 is returned + assert_eq!(resp.status(), 404); + + Ok(()) +} + +/// Verifies that the list endpoint supports pagination. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn list_jobs_paginated(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given three job records in the database + let sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + for i in 0..3 { + insert_test_job( + &ctx.db, + Uuid::now_v7(), + sbom_id, + &format!("CVE-2024-{i:04}"), + ExploitIntelligenceJobStatus::Pending, + None, + ) + .await; + } + + let ei_service = test_service(); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::list_jobs), + ) + .await; + + // When requesting with limit=2 + let req = TestRequest::get() + .uri("/v3/exploit-intelligence/jobs?limit=2") + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then only 2 items are returned but total reflects all 3 + assert_eq!(resp.status(), 200); + let body: PaginatedResults = test::read_body_json(resp).await; + assert_eq!(body.items.len(), 2); + assert_eq!(body.total, Some(3)); + + Ok(()) +} + +/// Verifies that the list endpoint supports filtering by sbom_id. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn list_jobs_filter_by_sbom_id(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given jobs for two different SBOMs + let target_sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + let other_sbom_id = ingest_and_parse_sbom_id(ctx, "zookeeper-3.9.2-cyclonedx.json").await; + + insert_test_job( + &ctx.db, + Uuid::now_v7(), + target_sbom_id, + "CVE-2024-0001", + ExploitIntelligenceJobStatus::Pending, + None, + ) + .await; + + insert_test_job( + &ctx.db, + Uuid::now_v7(), + other_sbom_id, + "CVE-2024-0002", + ExploitIntelligenceJobStatus::Pending, + None, + ) + .await; + + let ei_service = test_service(); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::list_jobs), + ) + .await; + + // When filtering by the target SBOM ID + let req = TestRequest::get() + .uri(&format!( + "/v3/exploit-intelligence/jobs?sbom_id={target_sbom_id}" + )) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then only the matching job is returned + assert_eq!(resp.status(), 200); + let body: PaginatedResults = test::read_body_json(resp).await; + assert_eq!(body.items.len(), 1); + assert_eq!(body.total, Some(1)); + + Ok(()) +} + +/// Verifies job deduplication: submitting an analyze request for an SBOM+CVE +/// that already has an active (Pending or Running) job returns the existing +/// job instead of creating a duplicate. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn analyze_deduplicates_active_jobs(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given a real SBOM in the database + let sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + + // And an existing pending job for that SBOM and CVE + let existing_job_id = Uuid::now_v7(); + insert_test_job( + &ctx.db, + existing_job_id, + sbom_id, + "CVE-2024-DEDUP", + ExploitIntelligenceJobStatus::Pending, + None, + ) + .await; + + let ei_service = test_service(); + let db_rw = db::ReadWrite::new(ctx.db.clone()); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_rw)) + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::analyze), + ) + .await; + + // When submitting an analyze request for the same SBOM + CVE + let req = TestRequest::post() + .uri("/v3/exploit-intelligence/analyze") + .set_json(AnalyzeRequest { + sbom_id, + vulnerability_id: "CVE-2024-DEDUP".to_string(), + }) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then the response returns the existing job (not a new one) with 200 + assert_eq!(resp.status(), 200); + let body: AnalyzeResponse = test::read_body_json(resp).await; + assert_eq!(body.job_id, existing_job_id); + assert_eq!(body.status, JobStatus::Pending); + + Ok(()) +} + +/// Verifies that a job updated with a timeout error message is correctly +/// reflected when fetched via the GET endpoint. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn get_job_timeout_failure(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given a running job record in the database + let sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + let job_id = Uuid::now_v7(); + insert_test_job( + &ctx.db, + job_id, + sbom_id, + "CVE-2024-TIMEOUT", + ExploitIntelligenceJobStatus::Running, + None, + ) + .await; + + // When the service marks it as failed with a timeout error + let ei_service = test_service(); + let db_rw = db::ReadWrite::new(ctx.db.clone()); + ei_service + .update_job_failed(job_id, "analysis timed out after 1800s", &db_rw) + .await?; + + // Then fetching the job via GET shows the failed status and timeout message + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::get_job), + ) + .await; + + let req = TestRequest::get() + .uri(&format!("/v3/exploit-intelligence/jobs/{job_id}")) + .to_request(); + + let resp = test::call_service(&app, req).await; + + assert_eq!(resp.status(), 200); + let body: ExploitIntelligenceJobDetails = test::read_body_json(resp).await; + assert_eq!(body.id, job_id); + assert_eq!(body.status, JobStatus::Failed); + assert_eq!( + body.error_message.as_deref(), + Some("analysis timed out after 1800s") + ); + assert_eq!(body.finding, None); + + Ok(()) +} + +/// Verifies that a completed job with a vulnerable component finding is +/// returned correctly -- the aggregate finding is computed from components. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn get_job_finding_without_advisory(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given a completed job with a vulnerable component + let sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + let job_id = Uuid::now_v7(); + insert_test_job( + &ctx.db, + job_id, + sbom_id, + "CVE-2024-VULN", + ExploitIntelligenceJobStatus::Completed, + None, + ) + .await; + insert_test_component( + &ctx.db, + job_id, + "vulnerable-component", + ExploitIntelligenceJobStatus::Completed, + Some(exploit_intelligence_job::Finding::Vulnerable), + ) + .await; + + let ei_service = test_service(); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::get_job), + ) + .await; + + // When fetching the job details + let req = TestRequest::get() + .uri(&format!("/v3/exploit-intelligence/jobs/{job_id}")) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then the response shows the aggregate finding computed from the component + assert_eq!(resp.status(), 200); + let body: ExploitIntelligenceJobDetails = test::read_body_json(resp).await; + assert_eq!(body.id, job_id); + assert_eq!(body.status, JobStatus::Completed); + assert_eq!(body.finding, Some(JobFinding::Vulnerable)); + assert_eq!(body.error_message, None); + + Ok(()) +} + +/// Verifies that POSTing an analyze request for a valid SBOM creates a new job +/// and returns 201 with the job_id and a "pending" status. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn analyze_creates_new_job(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given a real SBOM in the database + let sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + + let ei_service = test_service(); + let db_rw = db::ReadWrite::new(ctx.db.clone()); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_rw)) + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::analyze), + ) + .await; + + // When submitting an analyze request (no pre-existing job) + let req = TestRequest::post() + .uri("/v3/exploit-intelligence/analyze") + .set_json(AnalyzeRequest { + sbom_id, + vulnerability_id: "CVE-2024-NEW".to_string(), + }) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then a new job is created and returned with 201 status + assert_eq!(resp.status(), 201); + let body: AnalyzeResponse = test::read_body_json(resp).await; + assert_eq!(body.status, JobStatus::Pending); + // The job_id should be a valid UUID (non-nil) + assert_ne!(body.job_id, Uuid::nil()); + + Ok(()) +} + +/// Verifies that the GET job details endpoint returns the correct state +/// for a running job. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn get_job_running(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given a running job record in the database + let sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + let job_id = Uuid::now_v7(); + insert_test_job( + &ctx.db, + job_id, + sbom_id, + "CVE-2024-RUN", + ExploitIntelligenceJobStatus::Running, + None, + ) + .await; + + let ei_service = test_service(); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::get_job), + ) + .await; + + // When requesting the job details + let req = TestRequest::get() + .uri(&format!("/v3/exploit-intelligence/jobs/{job_id}")) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then the response shows the running status + assert_eq!(resp.status(), 200); + let body: ExploitIntelligenceJobDetails = test::read_body_json(resp).await; + assert_eq!(body.id, job_id); + assert_eq!(body.status, JobStatus::Running); + assert_eq!(body.vulnerability_id, "CVE-2024-RUN"); + assert_eq!(body.finding, None); + + Ok(()) +} + +/// Verifies that the list endpoint supports filtering by job status, +/// returning only jobs that match the requested status. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn list_jobs_filter_by_status(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given jobs with different statuses + let sbom_id = ingest_and_parse_sbom_id(ctx, "cyclonedx/application.cdx.json").await; + insert_test_job( + &ctx.db, + Uuid::now_v7(), + sbom_id, + "CVE-2024-S001", + ExploitIntelligenceJobStatus::Completed, + None, + ) + .await; + + insert_test_job( + &ctx.db, + Uuid::now_v7(), + sbom_id, + "CVE-2024-S002", + ExploitIntelligenceJobStatus::Pending, + None, + ) + .await; + + insert_test_job( + &ctx.db, + Uuid::now_v7(), + sbom_id, + "CVE-2024-S003", + ExploitIntelligenceJobStatus::Completed, + None, + ) + .await; + + let ei_service = test_service(); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::list_jobs), + ) + .await; + + // When filtering by completed status + let req = TestRequest::get() + .uri("/v3/exploit-intelligence/jobs?status=completed") + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then only the two completed jobs are returned + assert_eq!(resp.status(), 200); + let body: PaginatedResults = test::read_body_json(resp).await; + assert_eq!(body.items.len(), 2); + assert_eq!(body.total, Some(2)); + for item in &body.items { + assert_eq!(item.status, JobStatus::Completed); + } + + Ok(()) +} + +/// Verifies that POSTing an analyze request with a non-existent SBOM ID +/// returns 404. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn analyze_sbom_not_found(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given an app configured with EI support + let ei_service = test_service(); + let db_rw = db::ReadWrite::new(ctx.db.clone()); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_rw)) + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::analyze), + ) + .await; + + // When submitting an analyze request with a non-existent SBOM ID + let req = TestRequest::post() + .uri("/v3/exploit-intelligence/analyze") + .set_json(AnalyzeRequest { + sbom_id: Uuid::now_v7(), + vulnerability_id: "CVE-2024-GHOST".to_string(), + }) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then a 404 is returned + assert_eq!(resp.status(), 404); + + Ok(()) +} + +/// Verifies that POSTing an analyze request when Exploit Intelligence is not +/// configured returns 503 Service Unavailable. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn analyze_ei_not_configured(ctx: &TrustifyContext) -> anyhow::Result<()> { + // Given an app with no EI configuration + let ei_service = ExploitIntelligenceService::new(None).unwrap(); + let db_rw = db::ReadWrite::new(ctx.db.clone()); + let db_ro = db::ReadOnly::new(ctx.db.clone()); + + let app = test::init_service( + App::new() + .add_test_authorizer() + .app_data(web::Data::new(db_rw)) + .app_data(web::Data::new(db_ro)) + .app_data(web::Data::new(ei_service)) + .service(super::analyze), + ) + .await; + + // When submitting an analyze request + let req = TestRequest::post() + .uri("/v3/exploit-intelligence/analyze") + .set_json(AnalyzeRequest { + sbom_id: Uuid::now_v7(), + vulnerability_id: "CVE-2024-NOCONFIG".to_string(), + }) + .to_request(); + + let resp = test::call_service(&app, req).await; + + // Then a 503 Service Unavailable is returned + assert_eq!(resp.status(), 503); + + Ok(()) +} diff --git a/modules/fundamental/src/exploit_intelligence/mod.rs b/modules/fundamental/src/exploit_intelligence/mod.rs new file mode 100644 index 000000000..4f768a58d --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/mod.rs @@ -0,0 +1,7 @@ +pub mod auth; +pub mod ei_client; +pub(crate) mod endpoints; + +pub mod model; +pub mod runner; +pub mod service; diff --git a/modules/fundamental/src/exploit_intelligence/model/mod.rs b/modules/fundamental/src/exploit_intelligence/model/mod.rs new file mode 100644 index 000000000..a73de2a2e --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/model/mod.rs @@ -0,0 +1,440 @@ +use serde::{Deserialize, Serialize}; +use time::OffsetDateTime; +use trustify_entity::exploit_intelligence_job::{ExploitIntelligenceJobStatus, Finding}; +use utoipa::ToSchema; +use uuid::Uuid; + +/// Request body for triggering an Exploit Intelligence analysis. +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct AnalyzeRequest { + /// The SBOM identifier to analyze. + pub sbom_id: Uuid, + /// The vulnerability identifier (e.g., "CVE-2024-9680"). + pub vulnerability_id: String, +} + +/// Response returned when an analysis job is successfully created. +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct AnalyzeResponse { + /// The unique job identifier. + pub job_id: Uuid, + /// Current lifecycle status of the job. + pub status: JobStatus, +} + +/// Summary of an Exploit Intelligence analysis job. +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct ExploitIntelligenceJobSummary { + /// The unique job identifier. + pub id: Uuid, + /// FK to the SBOM being analysed. + pub sbom_id: Uuid, + /// CVE identifier. + pub vulnerability_id: String, + /// Current lifecycle status of the job. + pub status: JobStatus, + /// Analysis result from Exploit Intelligence. + pub finding: Option, + /// Error details when the job has failed. + pub error_message: Option, + /// Link to the EI human-readable justification report. + pub report_url: Option, + /// EI product ID for multi-component SPDX flow. + pub product_id: Option, + /// Total components in the product. + pub total_components: Option, + /// Number of completed components (computed at query time, not stored). + pub completed_components: Option, + /// Number of genuinely failed components (computed at query time, not stored). + pub failed_components: Option, + /// Number of components excluded by the EI service (computed at query time, not stored). + pub excluded_components: Option, + /// Number of completed components with Vulnerable finding (computed at query time). + pub vulnerable_components: Option, + /// Number of completed components with NotVulnerable finding (computed at query time). + pub not_vulnerable_components: Option, + /// Number of completed components with Uncertain finding (computed at query time). + pub uncertain_components: Option, + /// Timestamp when the job was created. + #[schema(value_type = String)] + pub created: OffsetDateTime, + /// Timestamp when the job was last updated. + #[schema(value_type = String)] + pub updated: OffsetDateTime, +} + +/// Detailed view of an Exploit Intelligence analysis job. +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct ExploitIntelligenceJobDetails { + /// The unique job identifier. + pub id: Uuid, + /// FK to the SBOM being analysed. + pub sbom_id: Uuid, + /// CVE identifier. + pub vulnerability_id: String, + /// Current lifecycle status of the job. + pub status: JobStatus, + /// Analysis result computed from component findings. + pub finding: Option, + /// Link to the EI human-readable justification report. + pub report_url: Option, + /// Error details for failed analyses. + pub error_message: Option, + /// EI product ID for multi-component SPDX flow. + pub product_id: Option, + /// Total components in the product. + pub total_components: Option, + /// Number of completed components (computed at query time, not stored). + pub completed_components: Option, + /// Number of genuinely failed components (computed at query time, not stored). + pub failed_components: Option, + /// Number of components excluded by the EI service (computed at query time, not stored). + pub excluded_components: Option, + /// Number of completed components with Vulnerable finding (computed at query time). + pub vulnerable_components: Option, + /// Number of completed components with NotVulnerable finding (computed at query time). + pub not_vulnerable_components: Option, + /// Number of completed components with Uncertain finding (computed at query time). + pub uncertain_components: Option, + /// Per-component analysis results (populated for multi-component jobs). + pub components: Vec, + /// Timestamp when the job was created. + #[schema(value_type = String)] + pub created: OffsetDateTime, + /// Timestamp when the job was last updated. + #[schema(value_type = String)] + pub updated: OffsetDateTime, +} + +/// Aggregated component counts for a job, used to populate summary fields +/// and compute the overall finding. +#[derive(Debug, Clone, Default)] +pub struct ComponentCounts { + pub completed: i32, + pub failed: i32, + pub excluded: i32, + pub total: i32, + pub vulnerable: i32, + pub not_vulnerable: i32, + pub uncertain: i32, +} + +/// Per-component analysis result within an analysis job. +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct ComponentResult { + /// The unique component record identifier. + pub id: Uuid, + /// Identifier of the component (purl or container image reference). + pub component_ref: String, + /// Current lifecycle status of this component's analysis. + pub status: JobStatus, + /// Analysis result from Exploit Intelligence, if completed. + pub finding: Option, + /// Link to the EI human-readable justification report for this component. + pub report_url: Option, + /// FK to advisory — set when a VEX document is ingested from the result. + pub advisory_id: Option, + /// Whether this component was excluded by the EI service (e.g., + /// unsupported ecosystem) rather than genuinely failing. + pub excluded: bool, + /// Error details for failed component analyses. + pub error_message: Option, + /// Timestamp when the component record was created. + #[schema(value_type = String)] + pub created: OffsetDateTime, + /// Timestamp when the component record was last updated. + #[schema(value_type = String)] + pub updated: OffsetDateTime, +} + +/// Lifecycle status of an Exploit Intelligence analysis job (API representation). +#[derive(Debug, Copy, Clone, Serialize, Deserialize, PartialEq, Eq, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum JobStatus { + /// Job has been created but not yet started. + Pending, + /// Analysis is in progress. + Running, + /// Analysis completed successfully. + Completed, + /// Analysis failed with an error. + Failed, +} + +impl From for JobStatus { + fn from(status: ExploitIntelligenceJobStatus) -> Self { + match status { + ExploitIntelligenceJobStatus::Pending => Self::Pending, + ExploitIntelligenceJobStatus::Running => Self::Running, + ExploitIntelligenceJobStatus::Completed => Self::Completed, + ExploitIntelligenceJobStatus::Failed => Self::Failed, + } + } +} + +/// Analysis finding from Exploit Intelligence (API representation). +#[derive(Debug, Copy, Clone, Serialize, Deserialize, PartialEq, Eq, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum JobFinding { + /// The analysed component is vulnerable to the CVE. + Vulnerable, + /// The analysed component is not vulnerable to the CVE. + NotVulnerable, + /// The analysis was inconclusive. + Uncertain, +} + +impl From for JobFinding { + fn from(finding: Finding) -> Self { + match finding { + Finding::Vulnerable => Self::Vulnerable, + Finding::NotVulnerable => Self::NotVulnerable, + Finding::Uncertain => Self::Uncertain, + } + } +} + +impl From for ExploitIntelligenceJobStatus { + fn from(status: JobStatus) -> Self { + match status { + JobStatus::Pending => Self::Pending, + JobStatus::Running => Self::Running, + JobStatus::Completed => Self::Completed, + JobStatus::Failed => Self::Failed, + } + } +} + +impl ComponentResult { + /// Create a component result from an entity model. + /// + /// `ui_url` and `vulnerability_id` are used to compute the report URL + /// for completed components with a scan ID. + pub fn from_entity( + model: &trustify_entity::exploit_intelligence_job_component::Model, + ui_url: Option<&str>, + vulnerability_id: &str, + ) -> Self { + let report_url = ui_url.and_then(|base| { + model.scan_id.as_ref().map(|scan_id| { + format!( + "{}/reports/component/{}/{}", + base, + urlencoding::encode(vulnerability_id), + urlencoding::encode(scan_id), + ) + }) + }); + Self { + id: model.id, + component_ref: model.component_ref.clone(), + status: model.status.into(), + finding: model.finding.map(JobFinding::from), + report_url, + advisory_id: model.advisory_id, + excluded: model.excluded, + error_message: model.error_message.clone(), + created: model.created, + updated: model.updated, + } + } +} + +/// Compute the EI report URL from a job's identifiers and the UI base URL. +/// +/// For multi-component jobs (with `product_id`), returns the product-level URL. +/// For single-component jobs, returns `None` — the component's own report URL +/// (in [`ComponentResult`]) provides the link instead. +/// Compute the EI report URL from a job's identifiers and the UI base URL. +/// +/// For multi-component (SPDX) jobs with a `product_id`, builds a product URL. +/// For single-component (CycloneDX) jobs, uses the component's `scan_id` +/// passed via `first_scan_id` to build a component URL. +fn compute_job_report_url( + model: &trustify_entity::exploit_intelligence_job::Model, + ui_url: Option<&str>, + first_scan_id: Option<&str>, +) -> Option { + let base = ui_url?; + if let Some(ref product_id) = model.product_id { + Some(format!( + "{}/reports/product/{}/{}", + base, + urlencoding::encode(product_id), + urlencoding::encode(&model.vulnerability_id), + )) + } else { + first_scan_id.map(|scan_id| { + format!( + "{}/reports/component/{}/{}", + base, + urlencoding::encode(&model.vulnerability_id), + urlencoding::encode(scan_id), + ) + }) + } +} + +impl ExploitIntelligenceJobSummary { + /// Create a summary from an entity model. + /// + /// `ui_url` is used to compute the report URL at response time. + /// Finding is not available without component data and will be `None`. + pub fn from_entity( + model: &trustify_entity::exploit_intelligence_job::Model, + ui_url: Option<&str>, + ) -> Self { + Self { + id: model.id, + sbom_id: model.sbom_id, + vulnerability_id: model.vulnerability_id.clone(), + status: model.status.into(), + finding: None, + error_message: model.error_message.clone(), + report_url: compute_job_report_url(model, ui_url, None), + product_id: model.product_id.clone(), + total_components: model.total_components, + completed_components: None, + failed_components: None, + excluded_components: None, + vulnerable_components: None, + not_vulnerable_components: None, + uncertain_components: None, + created: model.created, + updated: model.updated, + } + } + + /// Create a summary from an entity model with pre-computed component counts. + /// + /// The aggregate finding is computed from component count data: + /// - If any components are vulnerable, the finding is `Vulnerable`. + /// - If all completed components are not vulnerable, the finding is `NotVulnerable`. + /// - Otherwise, the finding is `Uncertain`. + pub fn from_entity_with_counts( + model: &trustify_entity::exploit_intelligence_job::Model, + ui_url: Option<&str>, + counts: &ComponentCounts, + ) -> Self { + let mut summary = Self::from_entity(model, ui_url); + summary.completed_components = Some(counts.completed); + summary.failed_components = Some(counts.failed); + summary.excluded_components = Some(counts.excluded); + summary.vulnerable_components = Some(counts.vulnerable); + summary.not_vulnerable_components = Some(counts.not_vulnerable); + summary.uncertain_components = Some(counts.uncertain); + + // Compute aggregate finding from component counts. + // If any component is vulnerable, the whole job is vulnerable. + // If some components genuinely failed (not excluded), we can't be + // sure what they would have found — the aggregate is uncertain. + // Only if all analysed components are not_vulnerable is the job + // confidently not_vulnerable. + if counts.vulnerable > 0 { + summary.finding = Some(JobFinding::Vulnerable); + } else if counts.failed > 0 { + // Some components genuinely failed — we can't confidently + // aggregate, so leave finding as None. The UI can use + // failed_components > 0 to explain why. + } else if counts.completed > 0 && counts.not_vulnerable == counts.completed { + summary.finding = Some(JobFinding::NotVulnerable); + } else if counts.completed > 0 { + summary.finding = Some(JobFinding::Uncertain); + } + // else: no completed components yet, finding stays None + + summary + } +} + +impl ExploitIntelligenceJobDetails { + /// Create a detailed view from an entity model. + /// + /// `ui_url` is used to compute the report URL at response time. + /// Finding is not available without component data and will be `None`. + pub fn from_entity( + model: &trustify_entity::exploit_intelligence_job::Model, + ui_url: Option<&str>, + ) -> Self { + Self { + id: model.id, + sbom_id: model.sbom_id, + vulnerability_id: model.vulnerability_id.clone(), + status: model.status.into(), + finding: None, + report_url: compute_job_report_url(model, ui_url, None), + error_message: model.error_message.clone(), + product_id: model.product_id.clone(), + total_components: model.total_components, + completed_components: None, + failed_components: None, + excluded_components: None, + vulnerable_components: None, + not_vulnerable_components: None, + uncertain_components: None, + components: Vec::new(), + created: model.created, + updated: model.updated, + } + } + + /// Create a detailed view from an entity model with component results. + /// + /// The aggregate finding is computed from the component models. + pub fn from_entity_with_components( + model: &trustify_entity::exploit_intelligence_job::Model, + component_models: &[trustify_entity::exploit_intelligence_job_component::Model], + ui_url: Option<&str>, + ) -> Self { + let first_scan_id = component_models.first().and_then(|c| c.scan_id.as_deref()); + let mut details = Self { + report_url: compute_job_report_url(model, ui_url, first_scan_id), + ..Self::from_entity(model, ui_url) + }; + let vuln_id = &model.vulnerability_id; + details.components = component_models + .iter() + .map(|c| ComponentResult::from_entity(c, ui_url, vuln_id)) + .collect(); + let completed = component_models + .iter() + .filter(|c| c.status == ExploitIntelligenceJobStatus::Completed && !c.excluded) + .count() as i32; + let failed = component_models + .iter() + .filter(|c| c.status == ExploitIntelligenceJobStatus::Failed && !c.excluded) + .count() as i32; + let excluded = component_models.iter().filter(|c| c.excluded).count() as i32; + let vulnerable = component_models + .iter() + .filter(|c| c.finding == Some(Finding::Vulnerable)) + .count() as i32; + let not_vulnerable = component_models + .iter() + .filter(|c| c.finding == Some(Finding::NotVulnerable)) + .count() as i32; + let uncertain = component_models + .iter() + .filter(|c| c.finding == Some(Finding::Uncertain)) + .count() as i32; + details.completed_components = Some(completed); + details.failed_components = Some(failed); + details.excluded_components = Some(excluded); + details.vulnerable_components = Some(vulnerable); + details.not_vulnerable_components = Some(not_vulnerable); + details.uncertain_components = Some(uncertain); + + // Compute aggregate finding from components (same logic as summary). + if vulnerable > 0 { + details.finding = Some(JobFinding::Vulnerable); + } else if failed > 0 { + // Some components genuinely failed — leave finding as None. + } else if completed > 0 && not_vulnerable == completed { + details.finding = Some(JobFinding::NotVulnerable); + } else if completed > 0 { + details.finding = Some(JobFinding::Uncertain); + } + + details + } +} diff --git a/modules/fundamental/src/exploit_intelligence/runner/analysis.rs b/modules/fundamental/src/exploit_intelligence/runner/analysis.rs new file mode 100644 index 000000000..6261d6c07 --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/runner/analysis.rs @@ -0,0 +1,233 @@ +use crate::exploit_intelligence::{ + ei_client::{SpdxUploadResponse, UploadResponse}, + runner::{EiRequestError, classify_response_error}, + service::{ExploitIntelligenceConfig, ExploitIntelligenceService}, +}; +use crate::source_document::model::SourceDocument; +use backon::{ExponentialBuilder, Retryable}; +use futures_util::TryStreamExt; +use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; +use trustify_auth::client::TokenInjector; +use trustify_common::db; +use trustify_entity::exploit_intelligence_job; +use trustify_module_ingestor::service::{DocumentDetector, Format, IngestorService}; +use trustify_module_storage::service::{StorageBackend, StorageKey}; + +use super::polling::{poll_for_product_result, poll_for_result}; + +/// Background task: submit SBOM to EI client, poll for results, ingest VEX. +/// +/// Accepts the full job model so the caller does not need to pass individual +/// fields. Uses `db_ro` for read-only SBOM retrieval and `db_rw` for +/// mutations. +pub(crate) async fn run_analysis( + ei_service: &ExploitIntelligenceService, + ingestor: &IngestorService, + db_rw: &db::ReadWrite, + db_ro: &db::ReadOnly, + config: &ExploitIntelligenceConfig, + client: &reqwest::Client, + job: &exploit_intelligence_job::Model, +) -> Result<(), anyhow::Error> { + let job_id = job.id; + let sbom_id = job.sbom_id; + let vulnerability_id = &job.vulnerability_id; + + // Retrieve the raw SBOM document from storage (read-only -- no mutations needed). + let tx = db_ro.begin().await?; + let sbom_model = trustify_entity::sbom::Entity::find() + .filter(trustify_entity::sbom::Column::SbomId.eq(sbom_id)) + .one(&tx) + .await? + .ok_or_else(|| anyhow::anyhow!("SBOM not found: {sbom_id}"))?; + + let source_doc = + trustify_entity::source_document::Entity::find_by_id(sbom_model.source_document_id) + .one(&tx) + .await? + .ok_or_else(|| anyhow::anyhow!("source document not found for SBOM: {sbom_id}"))?; + + // Reuse the SourceDocument model's TryInto implementation, + // which handles the "sha256:" prefix and fallback to sha384/sha512. + let source_doc_model = SourceDocument::from_entity(&source_doc); + let storage_key: StorageKey = source_doc_model + .try_into() + .map_err(|e| anyhow::anyhow!("storage key error: {e}"))?; + + let stream = ingestor + .storage() + .retrieve(storage_key) + .await + .map_err(|e| anyhow::anyhow!("storage retrieve error: {e}"))? + .ok_or_else(|| anyhow::anyhow!("SBOM document not found in storage"))?; + + let sbom_bytes: Vec = stream + .try_fold( + Vec::with_capacity(source_doc.size as usize), + |mut acc, chunk| async move { + acc.extend_from_slice(&chunk); + Ok(acc) + }, + ) + .await + .map_err(|e| anyhow::anyhow!("failed to read SBOM stream: {e}"))?; + + drop(tx); + + // Detect SBOM format from content rather than labels (which users can modify). + let format = DocumentDetector::detect(&sbom_bytes) + .map_err(|e| anyhow::anyhow!("failed to detect SBOM format: {e}"))? + .format(); + + // Wrap in Bytes for O(1) cloning inside the retry closures. + let sbom_data = bytes::Bytes::from(sbom_bytes); + + if format == Format::SPDX { + // SPDX multi-component flow: upload to /api/v1/products/upload-spdx + let upload_url = format!("{}/api/v1/products/upload-spdx", config.url); + + let response = (|| { + let url = upload_url.clone(); + let sbom = sbom_data.clone(); + let vuln_id = vulnerability_id.to_string(); + let cli = client.clone(); + let tp = config.token_provider.clone(); + async move { + let form = reqwest::multipart::Form::new().text("cveId", vuln_id).part( + "file", + reqwest::multipart::Part::bytes(sbom.to_vec()) + .file_name("sbom.json") + .mime_str("application/json") + .map_err(|e| { + EiRequestError::Permanent(anyhow::anyhow!("mime error: {e}")) + })?, + ); + + let request = cli + .post(&url) + .multipart(form) + .inject_token(&tp) + .await + .map_err(|e| { + EiRequestError::Permanent(anyhow::anyhow!( + "failed to obtain auth token: {e}" + )) + })?; + + classify_response_error("SPDX upload", request.send().await).await + } + }) + .retry( + ExponentialBuilder::default() + .with_min_delay(config.upload_retry_delay) + .with_max_times(config.upload_max_retries), + ) + .when(|e| e.is_transient()) + .notify(|e, dur| tracing::warn!(?dur, %e, "retrying EI upload")) + .await + .map_err(EiRequestError::into_anyhow)?; + + let spdx_upload: SpdxUploadResponse = response.json().await?; + let product_id = spdx_upload.product_id; + + // Update job to running with the product ID. + ei_service + .update_job_running_with_product(job_id, &product_id, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to update job: {e}"))?; + + tracing::info!( + job_id = %job_id, + product_id = %product_id, + "SPDX product uploaded to EI, starting product-level polling" + ); + + // Hand off to the product-level poll loop. + poll_for_product_result( + ei_service, + ingestor, + db_rw, + config, + client, + job_id, + &product_id, + ) + .await + } else { + // CycloneDX single-component flow. + // Create the component record before uploading so it is tracked. + let component_ref = sbom_model + .document_id + .clone() + .unwrap_or_else(|| sbom_model.node_id.clone()); + let component_id = ei_service + .create_component(job_id, &component_ref, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to create component: {e}"))?; + + let upload_url = format!("{}/api/v1/products/upload-cyclonedx", config.url); + + let response = (|| { + let url = upload_url.clone(); + let sbom = sbom_data.clone(); + let vuln_id = vulnerability_id.to_string(); + let cli = client.clone(); + let tp = config.token_provider.clone(); + async move { + let form = reqwest::multipart::Form::new().text("cveId", vuln_id).part( + "file", + reqwest::multipart::Part::bytes(sbom.to_vec()) + .file_name("sbom.json") + .mime_str("application/json") + .map_err(|e| { + EiRequestError::Permanent(anyhow::anyhow!("mime error: {e}")) + })?, + ); + + let request = cli + .post(&url) + .multipart(form) + .inject_token(&tp) + .await + .map_err(|e| { + EiRequestError::Permanent(anyhow::anyhow!( + "failed to obtain auth token: {e}" + )) + })?; + + classify_response_error("CycloneDX upload", request.send().await).await + } + }) + .retry( + ExponentialBuilder::default() + .with_min_delay(config.upload_retry_delay) + .with_max_times(config.upload_max_retries), + ) + .when(|e| e.is_transient()) + .notify(|e, dur| tracing::warn!(?dur, %e, "retrying EI upload")) + .await + .map_err(EiRequestError::into_anyhow)?; + + let upload: UploadResponse = response.json().await?; + let scan_id = upload.report_request_id.report_id; + + // Update component to running with scan ID. + ei_service + .update_component_running(component_id, &scan_id, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to update component: {e}"))?; + + // Hand off to the poll loop. + poll_for_result( + ei_service, + ingestor, + db_rw, + config, + client, + job_id, + &scan_id, + component_id, + ) + .await + } +} diff --git a/modules/fundamental/src/exploit_intelligence/runner/mod.rs b/modules/fundamental/src/exploit_intelligence/runner/mod.rs new file mode 100644 index 000000000..7997c3f5c --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/runner/mod.rs @@ -0,0 +1,72 @@ +pub(crate) mod analysis; +pub(crate) mod polling; +pub mod worker; + +/// Error classification for EI HTTP requests. +#[derive(Debug, thiserror::Error)] +pub(crate) enum EiRequestError { + /// Retryable failure (network timeout, 5xx, etc.) + #[error("transient: {0}")] + Transient(String), + /// Fatal failure (auth error, 4xx, parse error) + #[error("permanent: {0}")] + Permanent(anyhow::Error), +} + +impl EiRequestError { + pub(crate) fn is_transient(&self) -> bool { + matches!(self, Self::Transient(_)) + } + + pub(crate) fn into_anyhow(self) -> anyhow::Error { + match self { + Self::Transient(msg) => anyhow::anyhow!(msg), + Self::Permanent(err) => err, + } + } +} + +/// Classify an HTTP response or reqwest error into an EiRequestError. +pub(crate) async fn classify_response_error( + operation: &str, + result: Result, +) -> Result { + match result { + Ok(resp) => { + let status = resp.status(); + if status.is_success() { + Ok(resp) + } else { + let body = resp.text().await.unwrap_or_default(); + let body = body.trim(); + let detail = if body.is_empty() || body.starts_with('<') { + format!("{operation}: HTTP {status}") + } else if body.len() <= 512 { + format!("{operation}: HTTP {status}: {body}") + } else { + let end = body + .char_indices() + .map(|(i, _)| i) + .find(|&i| i >= 512) + .unwrap_or(body.len()); + format!("{operation}: HTTP {status}: {}", &body[..end]) + }; + match status.as_u16() { + 408 | 429 | 502 | 503 | 504 => Err(EiRequestError::Transient(detail)), + _ => Err(EiRequestError::Permanent(anyhow::anyhow!("{detail}"))), + } + } + } + Err(err) => { + if err.is_connect() || err.is_timeout() || err.is_request() { + Err(EiRequestError::Transient(format!( + "{operation}: network error: {err}" + ))) + } else { + Err(EiRequestError::Permanent(anyhow::anyhow!( + "{operation}: network error: {err}" + ))) + } + } + } +} diff --git a/modules/fundamental/src/exploit_intelligence/runner/polling.rs b/modules/fundamental/src/exploit_intelligence/runner/polling.rs new file mode 100644 index 000000000..f51ee541c --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/runner/polling.rs @@ -0,0 +1,633 @@ +use crate::exploit_intelligence::{ + runner::{EiRequestError, classify_response_error}, + service::ExploitIntelligenceService, +}; +use backon::{ConstantBuilder, Retryable}; +use sea_orm::TransactionTrait; +use std::cell::Cell; +use std::str::FromStr; +use std::time::Duration; +use trustify_auth::client::TokenInjector; +use trustify_common::db; +use trustify_entity::exploit_intelligence_job::Finding; +use trustify_entity::labels::Labels; +use trustify_module_ingestor::service::{Cache, Format, IngestorService}; +use uuid::Uuid; + +/// Poll the EI service for an in-flight single-component analysis and ingest the result. +/// +/// This is the second half of the analysis lifecycle: it assumes the SBOM has +/// already been uploaded and a `scan_id` obtained. +/// +/// Each successful poll tick extends the job's visibility window so that +/// other instances know this job is still being actively processed. EI +/// analysis can take hours; without extension the short initial window +/// would expire and another instance would re-claim the job. +/// +/// Updates both the component record (with finding/advisory) and the parent +/// job record (status only) upon completion. +#[allow(clippy::too_many_arguments)] +pub(crate) async fn poll_for_result( + ei_service: &ExploitIntelligenceService, + ingestor: &IngestorService, + db_rw: &db::ReadWrite, + config: &crate::exploit_intelligence::service::ExploitIntelligenceConfig, + client: &reqwest::Client, + job_id: Uuid, + scan_id: &str, + component_id: Uuid, +) -> Result<(), anyhow::Error> { + let poll_url = format!( + "{}/api/v1/reports/by-scan-id/{}", + config.url, + urlencoding::encode(scan_id) + ); + let poll_interval = config.poll_interval.max(Duration::from_secs(5)); + let max_consecutive = config.max_consecutive_poll_failures; + let consecutive_failures = Cell::new(0u32); + + let result = tokio::time::timeout( + config.max_poll_duration, + (|| { + let url = poll_url.clone(); + let cli = client.clone(); + let tp = config.token_provider.clone(); + let ei_svc = ei_service.clone(); + let db = db_rw.clone(); + async move { + let poll_request = cli.get(&url).inject_token(&tp).await.map_err(|e| { + EiRequestError::Permanent(anyhow::anyhow!("failed to obtain auth token: {e}")) + })?; + + let resp = + classify_response_error("poll scan result", poll_request.send().await).await?; + + let poll: crate::exploit_intelligence::ei_client::PollResponse = + resp.json().await.map_err(|e| { + EiRequestError::Permanent(anyhow::anyhow!( + "failed to parse EI poll response: {e}" + )) + })?; + + match poll.status.as_str() { + "completed" | "failed" | "expired" => Ok(poll), + _ => { + tracing::debug!( + status = poll.status, + "exploit intelligence analysis still in progress" + ); + // Extend visibility so other instances know we're alive. + if let Err(e) = ei_svc + .extend_visibility(job_id, std::time::Duration::from_secs(5 * 60), &*db) + .await + { + tracing::warn!(job_id = %job_id, error = %e, "failed to extend job visibility"); + } + Err(EiRequestError::Transient("pending".to_string())) + } + } + } + }) + .retry( + ConstantBuilder::default() + .with_delay(poll_interval) + .without_max_times(), + ) + .when(|e| match e { + EiRequestError::Transient(reason) if reason == "pending" => { + consecutive_failures.set(0); + true + } + EiRequestError::Transient(_) => { + let n = consecutive_failures.get() + 1; + consecutive_failures.set(n); + n < max_consecutive + } + EiRequestError::Permanent(_) => false, + }) + .notify(|e, dur| { + if !matches!(e, EiRequestError::Transient(r) if r == "pending") { + tracing::warn!(?dur, %e, "transient poll failure"); + } + }), + ) + .await; + + // Convert timeout/retry result to our expected shape. + let poll = match result { + Ok(Ok(poll)) => poll, + Ok(Err(e)) => { + let err = e.into_anyhow(); + let msg = err.to_string(); + ei_service.update_job_failed(job_id, &msg, db_rw).await?; + return Err(err); + } + Err(_elapsed) => { + let msg = format!("polling timed out after {:?}", config.max_poll_duration); + ei_service.update_job_failed(job_id, &msg, db_rw).await?; + return Err(anyhow::anyhow!("{msg}")); + } + }; + + // Process the terminal poll response. + match poll.status.as_str() { + "completed" => { + // Extract finding from the analysis result. + let finding = poll + .report + .as_ref() + .and_then(|r| r.output.as_ref()) + .and_then(|o| o.analysis.as_ref()) + .and_then(|a| a.first()) + .and_then(|e| e.justification.as_ref()) + .and_then(|j| j.status.as_deref()) + .map(|s| match s.to_ascii_lowercase().as_str() { + "false" => Finding::NotVulnerable, + "true" => Finding::Vulnerable, + _ => Finding::Uncertain, + }) + .unwrap_or(Finding::Uncertain); + + // Check if VEX document is available. + let vex_json = poll + .report + .as_ref() + .and_then(|r| r.output.as_ref()) + .and_then(|o| o.vex.as_ref()); + + let advisory_id = if let Some(vex) = vex_json { + if !vex.is_null() { + // Ingest the VEX document as an advisory with source label. + let vex_bytes = serde_json::to_vec(vex)?; + let mut labels = Labels::default(); + labels + .0 + .insert("source".to_string(), "exploit-intelligence".to_string()); + + let tx = db_rw.begin().await?; + let result = ingestor + .ingest(&vex_bytes, Format::Advisory, labels, None, Cache::Skip, &tx) + .await + .map_err(|e| anyhow::anyhow!("VEX ingestion failed: {e}"))?; + tx.commit().await?; + + tracing::info!( + job_id = %job_id, + advisory_id = %result.id, + "ingested VEX advisory from exploit intelligence" + ); + + result + .id + .strip_prefix("urn:uuid:") + .and_then(|id| Uuid::from_str(id).ok()) + } else { + None + } + } else { + None + }; + + // Update the component with the per-analysis results. + ei_service + .update_component_completed(component_id, finding, advisory_id, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to update component: {e}"))?; + + // Mark the parent job as completed (finding lives on the component). + ei_service + .update_job_completed(job_id, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to update job: {e}"))?; + + tracing::info!( + job_id = %job_id, + finding = ?finding, + "exploit intelligence analysis completed" + ); + + Ok(()) + } + "failed" | "expired" => { + // EI completed its lifecycle but the analysis itself failed. + // Mark the component as failed (not the job — the job's + // orchestration succeeded, only the analysis outcome is negative). + let error_msg = poll + .report + .as_ref() + .and_then(|r| r.error.as_ref()) + .and_then(|e| e.message.as_deref()) + .unwrap_or("analysis failed") + .to_string(); + + ei_service + .update_component_failed(component_id, &error_msg, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to update component: {e}"))?; + + ei_service + .update_job_completed(job_id, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to update job: {e}"))?; + + Ok(()) + } + _ => unreachable!("backon retry only returns terminal states"), + } +} + +/// Poll the EI service for a multi-component product and process per-component results. +/// +/// This is the SPDX counterpart of `poll_for_result`. It polls the product-level +/// endpoint until all components are complete, then fetches each component's +/// report individually. Like `poll_for_result`, each tick extends the job's +/// visibility window. +#[allow(clippy::too_many_arguments)] +pub(crate) async fn poll_for_product_result( + ei_service: &ExploitIntelligenceService, + ingestor: &IngestorService, + db_rw: &db::ReadWrite, + config: &crate::exploit_intelligence::service::ExploitIntelligenceConfig, + client: &reqwest::Client, + job_id: Uuid, + product_id: &str, +) -> Result<(), anyhow::Error> { + let product_url = format!( + "{}/api/v1/products/{}", + config.url, + urlencoding::encode(product_id) + ); + let poll_interval = config.poll_interval.max(Duration::from_secs(5)); + let max_consecutive = config.max_consecutive_poll_failures; + let consecutive_failures = Cell::new(0u32); + + let mut components_initialized = false; + + // Maps EI scan_id -> our internal component_id (populated on first successful poll). + let mut scan_to_component: std::collections::HashMap = + std::collections::HashMap::new(); + + // Check whether component records already exist for this job (e.g. after a + // crash/restart that orphaned a running job). If they do, skip re-initialization + // to avoid duplicate component rows. + let existing_components = ei_service + .fetch_components(job_id, &**db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to fetch existing components: {e}"))?; + if !existing_components.is_empty() { + for c in &existing_components { + if let Some(ref scan_id) = c.scan_id { + scan_to_component.insert(scan_id.clone(), c.id); + } + } + components_initialized = true; + tracing::info!( + job_id = %job_id, + existing_components = existing_components.len(), + "resumed with existing component records, skipping re-initialization" + ); + } + + let result = tokio::time::timeout( + config.max_poll_duration, + (|| { + let url = product_url.clone(); + let cli = client.clone(); + let tp = config.token_provider.clone(); + let ei_svc = ei_service.clone(); + let db = db_rw.clone(); + async move { + let poll_request = cli.get(&url).inject_token(&tp).await.map_err(|e| { + EiRequestError::Permanent(anyhow::anyhow!("failed to obtain auth token: {e}")) + })?; + + let resp = + classify_response_error("poll product status", poll_request.send().await) + .await?; + + let product: crate::exploit_intelligence::ei_client::ProductStatusResponse = + resp.json().await.map_err(|e| { + EiRequestError::Permanent(anyhow::anyhow!( + "failed to parse EI product response: {e}" + )) + })?; + + match product.summary.product_state.as_str() { + "completed" | "failed" => Ok(product), + _ => { + tracing::debug!( + state = product.summary.product_state, + "SPDX product analysis still in progress" + ); + if let Err(e) = ei_svc + .extend_visibility(job_id, std::time::Duration::from_secs(5 * 60), &*db) + .await + { + tracing::warn!(job_id = %job_id, error = %e, "failed to extend job visibility"); + } + Err(EiRequestError::Transient("pending".to_string())) + } + } + } + }) + .retry( + ConstantBuilder::default() + .with_delay(poll_interval) + .without_max_times(), + ) + .when(|e| match e { + EiRequestError::Transient(reason) if reason == "pending" => { + consecutive_failures.set(0); + true + } + EiRequestError::Transient(_) => { + let n = consecutive_failures.get() + 1; + consecutive_failures.set(n); + n < max_consecutive + } + EiRequestError::Permanent(_) => false, + }) + .notify(|e, dur| { + if !matches!(e, EiRequestError::Transient(r) if r == "pending") { + tracing::warn!(?dur, %e, "transient poll failure"); + } + }), + ) + .await; + + // Convert timeout/retry result to our expected shape. + let product = match result { + Ok(Ok(product)) => product, + Ok(Err(e)) => { + let err = e.into_anyhow(); + let msg = err.to_string(); + ei_service.update_job_failed(job_id, &msg, db_rw).await?; + return Err(err); + } + Err(_elapsed) => { + let msg = format!("polling timed out after {:?}", config.max_poll_duration); + ei_service.update_job_failed(job_id, &msg, db_rw).await?; + return Err(anyhow::anyhow!("{msg}")); + } + }; + + // Process the terminal product response. + match product.summary.product_state.as_str() { + "completed" => { + tracing::info!( + job_id = %job_id, + product_id = %product_id, + "SPDX product analysis completed, fetching per-component reports" + ); + + // The EI product endpoint does not return per-component + // data (its `components` array is always empty). Instead, + // fetch the reports list filtered by product ID to discover + // each component's scan ID and justification. + let reports_url = format!( + "{}/api/v1/reports?productId={}", + config.url, + urlencoding::encode(product_id) + ); + + let reports_request = client + .get(&reports_url) + .inject_token(&config.token_provider) + .await + .map_err(|e| anyhow::anyhow!("failed to obtain auth token: {e}"))?; + + let reports_response = reports_request + .send() + .await + .map_err(|e| anyhow::anyhow!("failed to fetch product reports: {e}"))?; + + if !reports_response.status().is_success() { + let status = reports_response.status(); + let body = reports_response.text().await.unwrap_or_default(); + return Err(anyhow::anyhow!( + "EI product reports request failed ({status}): {body}" + )); + } + + let component_reports: Vec = + reports_response.json().await?; + + // Record submission failures as excluded components. + if !components_initialized { + for failure in &product.data.submission_failures { + let image = failure.image.as_deref().unwrap_or("unknown"); + let error = failure.error.as_deref().unwrap_or("excluded by EI"); + + ei_service + .create_excluded_component(job_id, image, error, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to create excluded component: {e}"))?; + } + } + + tracing::info!( + job_id = %job_id, + product_id = %product_id, + reports = component_reports.len(), + excluded = product.data.submission_failures.len(), + "processing per-component reports for SPDX product" + ); + + for entry in &component_reports { + let scan_id = &entry.scan_id; + let image_name = &entry.image_name; + + // Create a component record for this report. + let component_id = if !components_initialized { + let id = ei_service + .create_component(job_id, image_name, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to create component: {e}"))?; + ei_service + .update_component_running(id, scan_id, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to update component: {e}"))?; + id + } else { + match scan_to_component.get(scan_id) { + Some(id) => *id, + None => { + // Component was not tracked before the crash — create it now + // so it is not silently dropped. + let id = ei_service + .create_component(job_id, image_name, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to create component: {e}"))?; + ei_service + .update_component_running(id, scan_id, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to update component: {e}"))?; + id + } + } + }; + + // Handle non-completed reports (expired, failed on EI side). + if entry.state != "completed" { + ei_service + .update_component_failed( + component_id, + &format!("EI report state: {}", entry.state), + db_rw, + ) + .await + .map_err(|e| anyhow::anyhow!("failed to update component: {e}"))?; + continue; + } + + // Extract finding from the justification in the reports list. + let finding = entry + .vulns + .first() + .and_then(|v| v.justification.as_ref()) + .and_then(|j| j.status.as_deref()) + .map(|s| match s.to_ascii_lowercase().as_str() { + "false" => Finding::NotVulnerable, + "true" => Finding::Vulnerable, + _ => Finding::Uncertain, + }) + .unwrap_or(Finding::Uncertain); + + // Fetch the full by-scan-id report for VEX ingestion. + let full_report_url = format!( + "{}/api/v1/reports/by-scan-id/{}", + config.url, + urlencoding::encode(scan_id) + ); + + let full_request = client + .get(&full_report_url) + .inject_token(&config.token_provider) + .await + .map_err(|e| anyhow::anyhow!("failed to obtain auth token: {e}"))?; + + let advisory_id = match full_request.send().await { + Ok(resp) if resp.status().is_success() => { + let poll_resp: crate::exploit_intelligence::ei_client::PollResponse = + resp.json().await?; + + let vex_json = poll_resp + .report + .as_ref() + .and_then(|r| r.output.as_ref()) + .and_then(|o| o.vex.as_ref()); + + if let Some(vex) = vex_json { + if !vex.is_null() { + let vex_bytes = serde_json::to_vec(vex)?; + let mut labels = Labels::default(); + labels.0.insert( + "source".to_string(), + "exploit-intelligence".to_string(), + ); + + let tx = db_rw.begin().await?; + let result = ingestor + .ingest( + &vex_bytes, + Format::Advisory, + labels, + None, + Cache::Skip, + &tx, + ) + .await + .map_err(|e| anyhow::anyhow!("VEX ingestion failed: {e}"))?; + tx.commit().await?; + + tracing::info!( + job_id = %job_id, + component = image_name, + advisory_id = %result.id, + "ingested VEX advisory for SPDX component" + ); + + result + .id + .strip_prefix("urn:uuid:") + .and_then(|id| Uuid::from_str(id).ok()) + } else { + None + } + } else { + None + } + } + _ => { + tracing::warn!( + job_id = %job_id, + scan_id = %scan_id, + "failed to fetch full report for VEX ingestion, skipping" + ); + None + } + }; + + ei_service + .update_component_completed(component_id, finding, advisory_id, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to update component: {e}"))?; + } + + // Check whether any component actually has a finding. + // If every component was excluded or failed, mark the job + // as failed rather than misleadingly "completed". + let components = ei_service + .fetch_components(job_id, &**db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to fetch components: {e}"))?; + + let has_finding = components.iter().any(|c| { + c.status == trustify_entity::exploit_intelligence_job::ExploitIntelligenceJobStatus::Completed + && c.finding.is_some() + }); + let all_excluded = !components.is_empty() && components.iter().all(|c| c.excluded); + + if has_finding || all_excluded { + ei_service + .update_job_completed(job_id, db_rw) + .await + .map_err(|e| anyhow::anyhow!("failed to update job: {e}"))?; + + tracing::info!( + job_id = %job_id, + product_id = %product_id, + all_excluded, + "SPDX product analysis completed" + ); + } else { + let failed_not_excluded = components.iter().filter(|c| !c.excluded).count(); + ei_service + .update_job_failed( + job_id, + &format!( + "no components produced findings ({} total, {failed_not_excluded} failed)", + components.len() + ), + db_rw, + ) + .await + .map_err(|e| anyhow::anyhow!("failed to update job: {e}"))?; + + tracing::warn!( + job_id = %job_id, + product_id = %product_id, + "SPDX product analysis failed — no components have findings" + ); + } + + Ok(()) + } + "failed" => { + let msg = format!("EI product analysis failed for product {product_id}"); + ei_service.update_job_failed(job_id, &msg, db_rw).await?; + Ok(()) + } + _ => unreachable!("backon retry only returns terminal states"), + } +} diff --git a/modules/fundamental/src/exploit_intelligence/runner/worker.rs b/modules/fundamental/src/exploit_intelligence/runner/worker.rs new file mode 100644 index 000000000..c1e393977 --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/runner/worker.rs @@ -0,0 +1,116 @@ +//! Background worker loop for processing Exploit Intelligence jobs. +//! +//! Uses a PostgreSQL job queue pattern: each tick dequeues a bounded +//! batch of eligible jobs via `FOR UPDATE SKIP LOCKED` and sets a short +//! visibility window on each. While processing, the worker periodically +//! extends the window so other instances know the job is still alive. +//! If the worker crashes, the window expires and another instance +//! reclaims the job on its next tick. +//! +//! No semaphore or concurrency limit is needed because each job's work +//! is a lightweight async poll loop (one HTTP GET every ~30s). The batch +//! LIMIT on dequeue ensures work is shared across instances. + +use crate::exploit_intelligence::{ + runner::analysis::run_analysis, service::ExploitIntelligenceService, +}; +use std::time::Duration; +use tokio::task::LocalSet; +use tokio::time::MissedTickBehavior; +use trustify_common::db; +use trustify_module_ingestor::service::IngestorService; + +/// Maximum jobs to claim per tick. Keeps one instance from starving +/// others when multiple instances share the same database. +const DEQUEUE_BATCH_SIZE: u64 = 50; + +/// Run the EI worker loop, polling for and processing jobs. +/// +/// The loop runs indefinitely, dequeuing eligible jobs every +/// `poll_interval`. Each claimed job is spawned as an independent +/// async task — no concurrency cap is needed because the work is +/// I/O-bound polling (not CPU/memory intensive like importer jobs). +/// +/// Wraps the inner loop in a `LocalSet` because spawned tasks use +/// `actix_web::rt::spawn` (which requires a local task context). +pub async fn run_worker( + ei_service: ExploitIntelligenceService, + ingestor: IngestorService, + db_rw: db::ReadWrite, + db_ro: db::ReadOnly, + poll_interval: Duration, +) -> anyhow::Result<()> { + LocalSet::new() + .run_until(run_worker_inner( + ei_service, + ingestor, + db_rw, + db_ro, + poll_interval, + )) + .await +} + +async fn run_worker_inner( + ei_service: ExploitIntelligenceService, + ingestor: IngestorService, + db_rw: db::ReadWrite, + db_ro: db::ReadOnly, + poll_interval: Duration, +) -> anyhow::Result<()> { + let mut interval = tokio::time::interval(poll_interval); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + + loop { + interval.tick().await; + + let jobs = match ei_service.dequeue_jobs(DEQUEUE_BATCH_SIZE, &db_rw).await { + Ok(j) => j, + Err(e) => { + tracing::error!(error = %e, "failed to dequeue EI jobs"); + continue; + } + }; + + // Mark retry-exhausted jobs as failed so they don't linger + // as invisible zombies (excluded from dequeue but never terminal). + if let Err(e) = ei_service.fail_exhausted_jobs(&db_rw).await { + tracing::error!(error = %e, "failed to mark exhausted EI jobs as failed"); + } + + for job in jobs { + let ei_svc = ei_service.clone(); + let ingestor_svc = ingestor.clone(); + let db_rw_clone = db_rw.clone(); + let db_ro_clone = db_ro.clone(); + let job_id = job.id; + + let runtime = match ei_svc.runtime().cloned() { + Some(rt) => rt, + None => continue, + }; + + actix_web::rt::spawn(async move { + let config = &runtime.config; + let client = &runtime.client; + + if let Err(e) = run_analysis( + &ei_svc, + &ingestor_svc, + &db_rw_clone, + &db_ro_clone, + config, + client, + &job, + ) + .await + { + tracing::error!(job_id = %job_id, error = %e, "EI analysis failed"); + let _ = ei_svc + .update_job_failed(job_id, &e.to_string(), &db_rw_clone) + .await; + } + }); + } + } +} diff --git a/modules/fundamental/src/exploit_intelligence/service/mod.rs b/modules/fundamental/src/exploit_intelligence/service/mod.rs new file mode 100644 index 000000000..9ffc6d02d --- /dev/null +++ b/modules/fundamental/src/exploit_intelligence/service/mod.rs @@ -0,0 +1,928 @@ +use crate::{ + Error, + exploit_intelligence::model::{ + ComponentCounts, ExploitIntelligenceJobDetails, ExploitIntelligenceJobSummary, JobStatus, + }, +}; +use sea_orm::sea_query::SimpleExpr; +use sea_orm::{ + ActiveModelTrait, ColumnTrait, Condition, ConnectionTrait, DbErr, EntityTrait, PaginatorTrait, + QueryFilter, QueryOrder, QuerySelect, Set, TransactionTrait, + sea_query::{Expr, LockBehavior, LockType}, +}; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; +use time::OffsetDateTime; +use tracing::{Instrument, info_span, instrument}; +use trustify_common::{ + db, + model::{Paginated, PaginatedResults}, +}; +use trustify_entity::exploit_intelligence_job::{self, ExploitIntelligenceJobStatus, Finding}; +use trustify_entity::exploit_intelligence_job_component; +use uuid::Uuid; + +/// Cast a job status string to the PostgreSQL enum type. +/// `update_many` with `Expr::value()` sends a plain text value, which +/// PostgreSQL rejects because the column is a custom enum. This helper +/// wraps the value in an explicit CAST. +fn status_expr(status: ExploitIntelligenceJobStatus) -> SimpleExpr { + let s = match status { + ExploitIntelligenceJobStatus::Pending => "pending", + ExploitIntelligenceJobStatus::Running => "running", + ExploitIntelligenceJobStatus::Completed => "completed", + ExploitIntelligenceJobStatus::Failed => "failed", + }; + Expr::cust_with_values("CAST($1 AS exploit_intelligence_job_status)", [s]) +} + +/// Extension for guarded updates where 0 rows affected (status guard +/// mismatch) is expected — another worker may have already transitioned +/// the record. +trait AllowNotUpdated { + fn ok_if_not_updated(self) -> Result<(), Error>; +} + +impl AllowNotUpdated for Result { + fn ok_if_not_updated(self) -> Result<(), Error> { + match self { + Ok(_) | Err(DbErr::RecordNotUpdated) => Ok(()), + Err(e) => Err(Error::from(e)), + } + } +} + +/// Configuration for the Exploit Intelligence client connection. +#[derive(Clone)] +pub struct ExploitIntelligenceConfig { + /// Base URL of the Exploit Intelligence client service. + pub url: String, + /// Base URL of the Exploit Intelligence web UI for deep-linking to reports. + /// When set, completed jobs include a clickable report URL. + pub ui_url: Option, + /// Polling interval for checking analysis completion. + /// Minimum enforced: 5 seconds (values below are clamped at the polling site). + pub poll_interval: Duration, + /// Maximum duration before a polling loop is considered timed out. + pub max_poll_duration: Duration, + /// Maximum number of upload retry attempts (exponential backoff). + pub upload_max_retries: usize, + /// Initial delay between upload retries (exponential backoff). + pub upload_retry_delay: Duration, + /// Maximum consecutive transient poll failures before giving up. + pub max_consecutive_poll_failures: u32, + /// Optional token provider for EI service authentication. + /// Supports static tokens and OIDC client credentials flow. + pub token_provider: Option>, +} + +impl std::fmt::Debug for ExploitIntelligenceConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ExploitIntelligenceConfig") + .field("url", &self.url) + .field("ui_url", &self.ui_url) + .field("poll_interval", &self.poll_interval) + .field("max_poll_duration", &self.max_poll_duration) + .field("upload_max_retries", &self.upload_max_retries) + .field("upload_retry_delay", &self.upload_retry_delay) + .field( + "max_consecutive_poll_failures", + &self.max_consecutive_poll_failures, + ) + .field( + "token_provider", + &self.token_provider.as_ref().map(|_| "[configured]"), + ) + .finish() + } +} + +/// Holds the EI configuration together with a shared HTTP client. +/// +/// These two values are always present or absent together: when EI is +/// enabled both exist; when it is disabled neither does. +#[derive(Clone, Debug)] +pub struct EiRuntime { + /// Configuration for the Exploit Intelligence client connection. + pub config: ExploitIntelligenceConfig, + /// Shared HTTP client for EI background tasks. Built once and reused + /// across all analysis invocations to benefit from connection pooling. + pub client: reqwest::Client, +} + +/// Service for orchestrating Exploit Intelligence analysis jobs. +#[derive(Clone)] +pub struct ExploitIntelligenceService { + runtime: Option, +} + +impl ExploitIntelligenceService { + /// Create a new service instance with optional configuration. + /// + /// When `config` is `None`, the service will return `Error::Unavailable` + /// for all analysis operations and no HTTP client is created. + pub fn new(config: Option) -> Result { + let runtime = config + .map(|config| -> Result { + let client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_secs(60)) + .pool_max_idle_per_host(4) + .pool_idle_timeout(Duration::from_secs(90)) + .build() + .map_err(|e| anyhow::anyhow!("failed to build reqwest client: {e}"))?; + Ok(EiRuntime { config, client }) + }) + .transpose()?; + Ok(Self { runtime }) + } + + /// Returns a reference to the shared HTTP client, if EI is enabled. + pub fn http_client(&self) -> Option<&reqwest::Client> { + self.runtime.as_ref().map(|rt| &rt.client) + } + + /// Returns the runtime (config + client) bundle, if EI is enabled. + pub fn runtime(&self) -> Option<&EiRuntime> { + self.runtime.as_ref() + } + + /// Returns the EI web UI base URL, if configured. + pub fn ui_url(&self) -> Option<&str> { + self.runtime + .as_ref() + .and_then(|r| r.config.ui_url.as_deref()) + } + + /// Create an analysis job record in the database. + /// + /// The job is created with `visible_at = NULL` so the background worker + /// picks it up on its next poll cycle. + /// + /// Returns the created job details. The caller must provide a connection + /// (typically a transaction) and is responsible for committing it. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn create_job( + &self, + sbom_id: Uuid, + vulnerability_id: &str, + connection: &C, + ) -> Result { + let _config = self.runtime.as_ref().ok_or(Error::Unavailable)?; + + let now = OffsetDateTime::now_utc(); + let job_id = Uuid::now_v7(); + + let model = exploit_intelligence_job::ActiveModel { + id: Set(job_id), + sbom_id: Set(sbom_id), + vulnerability_id: Set(vulnerability_id.to_string()), + status: Set(ExploitIntelligenceJobStatus::Pending), + error_message: Set(None), + product_id: Set(None), + total_components: Set(None), + visible_at: Set(None), + retry_count: Set(0), + created: Set(now), + updated: Set(now), + }; + + let result = model.insert(connection).await.map_err(Error::from)?; + + Ok(ExploitIntelligenceJobDetails::from_entity( + &result, + self.ui_url(), + )) + } + + /// Atomically dequeue eligible jobs using `FOR UPDATE SKIP LOCKED`. + /// + /// At most `limit` jobs are claimed per call to avoid one instance + /// starving others. Each claimed job gets a short visibility window; + /// the worker must periodically call `extend_visibility` to prove it + /// is still alive. If it crashes, the window expires and another + /// instance reclaims the job on its next tick. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn dequeue_jobs( + &self, + limit: u64, + db: &db::ReadWrite, + ) -> Result, Error> { + let runtime = self.runtime.as_ref().ok_or(Error::Unavailable)?; + // Short initial window — the worker extends it while processing. + let visibility_timeout = Duration::from_secs(5 * 60); + let max_retries = runtime.config.upload_max_retries as i32; + + let tx = db.begin().await?; + let now = OffsetDateTime::now_utc(); + + let jobs = exploit_intelligence_job::Entity::find() + .filter(exploit_intelligence_job::Column::Status.is_in([ + ExploitIntelligenceJobStatus::Pending, + ExploitIntelligenceJobStatus::Running, + ])) + .filter( + Condition::any() + .add(exploit_intelligence_job::Column::VisibleAt.is_null()) + .add(exploit_intelligence_job::Column::VisibleAt.lte(now)), + ) + .filter(exploit_intelligence_job::Column::RetryCount.lt(max_retries)) + .lock_with_behavior(LockType::Update, LockBehavior::SkipLocked) + .limit(limit) + .all(&tx) + .await + .map_err(Error::from)?; + + if !jobs.is_empty() { + let job_ids: Vec = jobs.iter().map(|j| j.id).collect(); + exploit_intelligence_job::Entity::update_many() + .col_expr( + exploit_intelligence_job::Column::Status, + status_expr(ExploitIntelligenceJobStatus::Running), + ) + .col_expr( + exploit_intelligence_job::Column::VisibleAt, + Expr::value(Some(now + visibility_timeout)), + ) + .col_expr( + exploit_intelligence_job::Column::RetryCount, + Expr::col(exploit_intelligence_job::Column::RetryCount).add(1), + ) + .col_expr(exploit_intelligence_job::Column::Updated, Expr::value(now)) + .filter(exploit_intelligence_job::Column::Id.is_in(job_ids)) + .exec(&tx) + .await + .map_err(Error::from)?; + } + + tx.commit().await?; + Ok(jobs) + } + + /// Push the visibility window forward so other instances know this job + /// is still being actively processed. Without periodic extension, a + /// long-running analysis (hours) would exceed the initial short window + /// and be re-claimed by another instance. + #[instrument(skip(self, db), err(level = tracing::Level::INFO))] + pub async fn extend_visibility( + &self, + job_id: Uuid, + extension: Duration, + db: &impl ConnectionTrait, + ) -> Result<(), Error> { + let now = OffsetDateTime::now_utc(); + exploit_intelligence_job::Entity::update_many() + .col_expr( + exploit_intelligence_job::Column::VisibleAt, + Expr::value(Some(now + extension)), + ) + .col_expr(exploit_intelligence_job::Column::Updated, Expr::value(now)) + .filter(exploit_intelligence_job::Column::Id.eq(job_id)) + .filter( + exploit_intelligence_job::Column::Status.eq(ExploitIntelligenceJobStatus::Running), + ) + .exec(db) + .await + .map_err(Error::from)?; + Ok(()) + } + + /// Mark jobs that have exhausted their retry budget as Failed. + /// + /// Without this, retry-exhausted jobs would remain in Running/Pending + /// status forever — excluded from dequeue (retry_count >= max) but + /// never transitioned to a terminal state. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn fail_exhausted_jobs(&self, db: &db::ReadWrite) -> Result<(), Error> { + let runtime = self.runtime.as_ref().ok_or(Error::Unavailable)?; + let max_retries = runtime.config.upload_max_retries as i32; + let now = OffsetDateTime::now_utc(); + + exploit_intelligence_job::Entity::update_many() + .col_expr( + exploit_intelligence_job::Column::Status, + status_expr(ExploitIntelligenceJobStatus::Failed), + ) + .col_expr( + exploit_intelligence_job::Column::ErrorMessage, + Expr::value(Some("max retries exhausted".to_string())), + ) + .col_expr( + exploit_intelligence_job::Column::VisibleAt, + Expr::value(Option::::None), + ) + .col_expr(exploit_intelligence_job::Column::Updated, Expr::value(now)) + .filter(exploit_intelligence_job::Column::Status.is_in([ + ExploitIntelligenceJobStatus::Pending, + ExploitIntelligenceJobStatus::Running, + ])) + .filter(exploit_intelligence_job::Column::RetryCount.gte(max_retries)) + .filter( + Condition::any() + .add(exploit_intelligence_job::Column::VisibleAt.is_null()) + .add(exploit_intelligence_job::Column::VisibleAt.lte(now)), + ) + .exec(&**db) + .await + .map_err(Error::from)?; + + Ok(()) + } + + /// Fetch details of a single job by its ID. + /// + /// Component records are always loaded so that the aggregate finding and + /// component counts can be computed. Every job (including single-component + /// CycloneDX jobs) now has at least one component row. + #[instrument(skip(self, connection), err(level = tracing::Level::INFO))] + pub async fn fetch_job( + &self, + id: Uuid, + connection: &C, + ) -> Result, Error> { + let result = exploit_intelligence_job::Entity::find_by_id(id) + .one(connection) + .instrument(info_span!("fetch exploit intelligence job")) + .await + .map_err(Error::from)?; + + match result { + Some(model) => { + let components = self.fetch_components(model.id, connection).await?; + if components.is_empty() { + Ok(Some(ExploitIntelligenceJobDetails::from_entity( + &model, + self.ui_url(), + ))) + } else { + Ok(Some( + ExploitIntelligenceJobDetails::from_entity_with_components( + &model, + &components, + self.ui_url(), + ), + )) + } + } + None => Ok(None), + } + } + + /// Fetch a paginated list of jobs, optionally filtered by SBOM ID and status. + #[instrument(skip(self, connection), err(level = tracing::Level::INFO))] + pub async fn fetch_jobs( + &self, + paginated: Paginated, + sbom_id: Option, + status: Option, + connection: &C, + ) -> Result, Error> { + let mut query = exploit_intelligence_job::Entity::find(); + + if let Some(sbom_id) = sbom_id { + query = query.filter(exploit_intelligence_job::Column::SbomId.eq(sbom_id)); + } + + if let Some(status) = status { + let db_status: ExploitIntelligenceJobStatus = status.into(); + query = query.filter(exploit_intelligence_job::Column::Status.eq(db_status)); + } + + let total = if paginated.total { + Some( + query + .clone() + .count(connection) + .instrument(info_span!("count exploit intelligence jobs")) + .await + .map_err(Error::from)?, + ) + } else { + None + }; + + let results = query + .order_by_desc(exploit_intelligence_job::Column::Created) + .offset(paginated.offset) + .limit(paginated.limit) + .all(connection) + .instrument(info_span!("fetch exploit intelligence jobs")) + .await + .map_err(Error::from)?; + + // Compute component counts per job via aggregate query, grouped by + // status, finding, and excluded flag. + let job_ids: Vec = results.iter().map(|j| j.id).collect(); + let mut counts_map: HashMap = HashMap::new(); + + if !job_ids.is_empty() { + let rows: Vec<( + Uuid, + ExploitIntelligenceJobStatus, + Option, + bool, + i64, + )> = exploit_intelligence_job_component::Entity::find() + .select_only() + .column(exploit_intelligence_job_component::Column::JobId) + .column(exploit_intelligence_job_component::Column::Status) + .column(exploit_intelligence_job_component::Column::Finding) + .column(exploit_intelligence_job_component::Column::Excluded) + .expr(Expr::col(exploit_intelligence_job_component::Column::Id).count()) + .filter(exploit_intelligence_job_component::Column::JobId.is_in(job_ids)) + .group_by(exploit_intelligence_job_component::Column::JobId) + .group_by(exploit_intelligence_job_component::Column::Status) + .group_by(exploit_intelligence_job_component::Column::Finding) + .group_by(exploit_intelligence_job_component::Column::Excluded) + .into_tuple() + .all(connection) + .instrument(info_span!("count exploit intelligence job components")) + .await + .map_err(Error::from)?; + + for (job_id, status, finding, excluded, count) in rows { + let entry = counts_map.entry(job_id).or_default(); + let count = count as i32; + entry.total += count; + if excluded { + entry.excluded += count; + } else if status == ExploitIntelligenceJobStatus::Completed { + entry.completed += count; + match finding { + Some(Finding::Vulnerable) => entry.vulnerable += count, + Some(Finding::NotVulnerable) => entry.not_vulnerable += count, + Some(Finding::Uncertain) => entry.uncertain += count, + _ => {} + } + } else if status == ExploitIntelligenceJobStatus::Failed { + entry.failed += count; + } + } + } + + // Fetch the first scan_id per job for report URL computation. + // Single-component jobs need this since scan_id lives on the component. + let job_ids_for_scan: Vec = results + .iter() + .filter(|j| j.product_id.is_none()) + .map(|j| j.id) + .collect(); + let mut scan_id_map: HashMap = HashMap::new(); + if !job_ids_for_scan.is_empty() { + let scan_rows: Vec<(Uuid, Option)> = + exploit_intelligence_job_component::Entity::find() + .select_only() + .column(exploit_intelligence_job_component::Column::JobId) + .column(exploit_intelligence_job_component::Column::ScanId) + .filter( + exploit_intelligence_job_component::Column::JobId.is_in(job_ids_for_scan), + ) + .filter(exploit_intelligence_job_component::Column::ScanId.is_not_null()) + .into_tuple() + .all(connection) + .instrument(info_span!("fetch component scan_ids")) + .await + .map_err(Error::from)?; + for (job_id, scan_id) in scan_rows { + if let Some(sid) = scan_id { + scan_id_map.entry(job_id).or_insert(sid); + } + } + } + + let items = results + .iter() + .map(|model| { + let mut summary = if let Some(c) = counts_map.get(&model.id) { + let mut s = ExploitIntelligenceJobSummary::from_entity_with_counts( + model, + self.ui_url(), + c, + ); + if s.total_components.is_none() { + s.total_components = Some(c.total); + } + s + } else { + ExploitIntelligenceJobSummary::from_entity(model, self.ui_url()) + }; + + // Set report_url from component scan_id for single-component jobs. + if model.product_id.is_none() + && let Some(scan_id) = scan_id_map.get(&model.id) + && let Some(base) = self.ui_url() + { + summary.report_url = Some(format!( + "{}/reports/component/{}/{}", + base, + urlencoding::encode(&model.vulnerability_id), + urlencoding::encode(scan_id), + )); + } + + summary + }) + .collect(); + + Ok(PaginatedResults { items, total }) + } + + /// Update a job to completed. + /// + /// Only transitions the job if its current status is `Running`, + /// which prevents double-completion. The visibility timeout is + /// cleared (set to NULL) on terminal state. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn update_job_completed( + &self, + job_id: Uuid, + db: &db::ReadWrite, + ) -> Result<(), Error> { + let model = exploit_intelligence_job::ActiveModel { + id: Set(job_id), + status: Set(ExploitIntelligenceJobStatus::Completed), + visible_at: Set(None), + updated: Set(OffsetDateTime::now_utc()), + ..Default::default() + }; + + exploit_intelligence_job::Entity::update(model) + .filter( + exploit_intelligence_job::Column::Status.eq(ExploitIntelligenceJobStatus::Running), + ) + .exec(&**db) + .await + .ok_if_not_updated()?; + + Ok(()) + } + + /// Update a job to failed with an error message. + /// + /// Only transitions the job if its current status is `Running`, + /// which prevents double-completion. The visibility timeout is + /// cleared (set to NULL) on terminal state. + /// + /// Also cascades failure to any non-terminal components so the API + /// never shows a Failed job with Running/Pending components. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn update_job_failed( + &self, + job_id: Uuid, + error_message: &str, + db: &db::ReadWrite, + ) -> Result<(), Error> { + let now = OffsetDateTime::now_utc(); + let tx = db.begin().await?; + + let model = exploit_intelligence_job::ActiveModel { + id: Set(job_id), + status: Set(ExploitIntelligenceJobStatus::Failed), + error_message: Set(Some(error_message.to_string())), + visible_at: Set(None), + updated: Set(now), + ..Default::default() + }; + + exploit_intelligence_job::Entity::update(model) + .filter( + exploit_intelligence_job::Column::Status.eq(ExploitIntelligenceJobStatus::Running), + ) + .exec(&tx) + .await + .ok_if_not_updated()?; + + // Cascade: fail any non-terminal components belonging to this job. + // Runs in the same transaction so a crash between the two UPDATEs + // cannot leave components in Running while the job is Failed. + exploit_intelligence_job_component::Entity::update_many() + .col_expr( + exploit_intelligence_job_component::Column::Status, + status_expr(ExploitIntelligenceJobStatus::Failed), + ) + .col_expr( + exploit_intelligence_job_component::Column::ErrorMessage, + Expr::value(Some("parent job failed".to_string())), + ) + .col_expr( + exploit_intelligence_job_component::Column::Updated, + Expr::value(now), + ) + .filter(exploit_intelligence_job_component::Column::JobId.eq(job_id)) + .filter( + exploit_intelligence_job_component::Column::Status + .ne(ExploitIntelligenceJobStatus::Completed), + ) + .filter( + exploit_intelligence_job_component::Column::Status + .ne(ExploitIntelligenceJobStatus::Failed), + ) + .exec(&tx) + .await + .map_err(Error::from)?; + + tx.commit().await?; + + Ok(()) + } + + /// Find an existing job in Pending or Running state for the given SBOM and vulnerability. + /// + /// Used for deduplication: if an active job already exists for the same + /// (sbom_id, vulnerability_id) pair, the caller can return it instead of + /// creating a duplicate. + #[instrument(skip(self, connection), err(level = tracing::Level::INFO))] + pub async fn find_active_job( + &self, + sbom_id: Uuid, + vulnerability_id: &str, + connection: &C, + ) -> Result, Error> { + let result = exploit_intelligence_job::Entity::find() + .filter(exploit_intelligence_job::Column::SbomId.eq(sbom_id)) + .filter(exploit_intelligence_job::Column::VulnerabilityId.eq(vulnerability_id)) + .filter(exploit_intelligence_job::Column::Status.is_in([ + ExploitIntelligenceJobStatus::Pending, + ExploitIntelligenceJobStatus::Running, + ])) + .one(connection) + .instrument(info_span!("find active exploit intelligence job")) + .await + .map_err(Error::from)?; + + Ok(result.map(|m| ExploitIntelligenceJobDetails::from_entity(&m, self.ui_url()))) + } + + /// Create an analysis job with product_id and total_components set (multi-component SPDX flow). + /// + /// The job is created with `visible_at = NULL` so the background worker + /// picks it up on its next poll cycle. + /// + /// Returns the created job details. The caller is responsible for creating + /// component records and spawning background tasks. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn create_job_with_product( + &self, + sbom_id: Uuid, + vulnerability_id: &str, + product_id: &str, + total_components: i32, + db: &db::ReadWrite, + ) -> Result { + let _config = self.runtime.as_ref().ok_or(Error::Unavailable)?; + let now = OffsetDateTime::now_utc(); + let job_id = Uuid::now_v7(); + + let model = exploit_intelligence_job::ActiveModel { + id: Set(job_id), + sbom_id: Set(sbom_id), + vulnerability_id: Set(vulnerability_id.to_string()), + status: Set(ExploitIntelligenceJobStatus::Pending), + error_message: Set(None), + product_id: Set(Some(product_id.to_string())), + total_components: Set(Some(total_components)), + visible_at: Set(None), + retry_count: Set(0), + created: Set(now), + updated: Set(now), + }; + + let tx = db.begin().await?; + let result = model.insert(&tx).await.map_err(Error::from)?; + tx.commit().await?; + + Ok(ExploitIntelligenceJobDetails::from_entity( + &result, + self.ui_url(), + )) + } + + /// Create a component record for an analysis job. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn create_component( + &self, + job_id: Uuid, + component_ref: &str, + db: &db::ReadWrite, + ) -> Result { + use trustify_entity::exploit_intelligence_job_component; + + let now = OffsetDateTime::now_utc(); + let component_id = Uuid::now_v7(); + + let model = exploit_intelligence_job_component::ActiveModel { + id: Set(component_id), + job_id: Set(job_id), + component_ref: Set(component_ref.to_string()), + scan_id: Set(None), + status: Set(ExploitIntelligenceJobStatus::Pending), + finding: Set(None), + advisory_id: Set(None), + excluded: Set(false), + error_message: Set(None), + created: Set(now), + updated: Set(now), + }; + + let tx = db.begin().await?; + model.insert(&tx).await.map_err(Error::from)?; + tx.commit().await?; + + Ok(component_id) + } + + /// Create a component record marked as excluded by the EI service. + /// + /// Used for components that cannot be analysed (e.g., unsupported ecosystem). + /// These are distinct from genuinely failed components. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn create_excluded_component( + &self, + job_id: Uuid, + component_ref: &str, + error_message: &str, + db: &db::ReadWrite, + ) -> Result { + use trustify_entity::exploit_intelligence_job_component; + + let now = OffsetDateTime::now_utc(); + let component_id = Uuid::now_v7(); + + let model = exploit_intelligence_job_component::ActiveModel { + id: Set(component_id), + job_id: Set(job_id), + component_ref: Set(component_ref.to_string()), + scan_id: Set(None), + status: Set(ExploitIntelligenceJobStatus::Failed), + finding: Set(None), + advisory_id: Set(None), + excluded: Set(true), + error_message: Set(Some(error_message.to_string())), + created: Set(now), + updated: Set(now), + }; + + let tx = db.begin().await?; + model.insert(&tx).await.map_err(Error::from)?; + tx.commit().await?; + + Ok(component_id) + } + + /// Update a component's status to running and store the scan ID from the EI service. + /// + /// Includes a guard: only updates if the component is not already in a terminal + /// state (completed), preventing silent overwrites of finished components. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn update_component_running( + &self, + component_id: Uuid, + scan_id: &str, + db: &db::ReadWrite, + ) -> Result<(), Error> { + use trustify_entity::exploit_intelligence_job_component; + + let model = exploit_intelligence_job_component::ActiveModel { + id: Set(component_id), + status: Set(ExploitIntelligenceJobStatus::Running), + scan_id: Set(Some(scan_id.to_string())), + updated: Set(OffsetDateTime::now_utc()), + ..Default::default() + }; + + exploit_intelligence_job_component::Entity::update(model) + .filter( + exploit_intelligence_job_component::Column::Status + .ne(ExploitIntelligenceJobStatus::Completed), + ) + .exec(&**db) + .await + .ok_if_not_updated()?; + + Ok(()) + } + + /// Update a component to completed with the analysis finding. + /// + /// Includes a guard: only updates if the component is not already completed, + /// preventing silent re-writes of terminal state. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn update_component_completed( + &self, + component_id: Uuid, + finding: exploit_intelligence_job::Finding, + advisory_id: Option, + db: &db::ReadWrite, + ) -> Result<(), Error> { + use trustify_entity::exploit_intelligence_job_component; + + let model = exploit_intelligence_job_component::ActiveModel { + id: Set(component_id), + status: Set(ExploitIntelligenceJobStatus::Completed), + finding: Set(Some(finding)), + advisory_id: Set(advisory_id), + updated: Set(OffsetDateTime::now_utc()), + ..Default::default() + }; + + exploit_intelligence_job_component::Entity::update(model) + .filter( + exploit_intelligence_job_component::Column::Status + .ne(ExploitIntelligenceJobStatus::Completed), + ) + .exec(&**db) + .await + .ok_if_not_updated()?; + + Ok(()) + } + + /// Update a component to failed with an error message. + /// + /// Includes a guard: only updates if the component is not already completed, + /// preventing silent re-writes of terminal state. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn update_component_failed( + &self, + component_id: Uuid, + error_message: &str, + db: &db::ReadWrite, + ) -> Result<(), Error> { + use trustify_entity::exploit_intelligence_job_component; + + let model = exploit_intelligence_job_component::ActiveModel { + id: Set(component_id), + status: Set(ExploitIntelligenceJobStatus::Failed), + error_message: Set(Some(error_message.to_string())), + updated: Set(OffsetDateTime::now_utc()), + ..Default::default() + }; + + exploit_intelligence_job_component::Entity::update(model) + .filter( + exploit_intelligence_job_component::Column::Status + .ne(ExploitIntelligenceJobStatus::Completed), + ) + .exec(&**db) + .await + .ok_if_not_updated()?; + + Ok(()) + } + + /// Update a job's status to running and store the product ID from the EI service. + /// + /// Uses a simple `WHERE id = ?` filter. The visibility timeout ensures + /// only one worker processes a job at a time. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn update_job_running_with_product( + &self, + job_id: Uuid, + product_id: &str, + db: &db::ReadWrite, + ) -> Result<(), Error> { + let model = exploit_intelligence_job::ActiveModel { + id: Set(job_id), + status: Set(ExploitIntelligenceJobStatus::Running), + product_id: Set(Some(product_id.to_string())), + updated: Set(OffsetDateTime::now_utc()), + ..Default::default() + }; + + exploit_intelligence_job::Entity::update(model) + .filter( + exploit_intelligence_job::Column::Status.eq(ExploitIntelligenceJobStatus::Running), + ) + .exec(&**db) + .await + .ok_if_not_updated()?; + + Ok(()) + } + + /// Fetch all component records for a given job, ordered by creation time. + #[instrument(skip_all, err(level = tracing::Level::INFO))] + pub async fn fetch_components( + &self, + job_id: Uuid, + connection: &C, + ) -> Result, Error> { + use trustify_entity::exploit_intelligence_job_component; + + let results = exploit_intelligence_job_component::Entity::find() + .filter(exploit_intelligence_job_component::Column::JobId.eq(job_id)) + .order_by_asc(exploit_intelligence_job_component::Column::Created) + .all(connection) + .instrument(info_span!("fetch exploit intelligence job components")) + .await + .map_err(Error::from)?; + + Ok(results) + } + + /// Returns the EI configuration, if available. + pub fn config(&self) -> Option<&ExploitIntelligenceConfig> { + self.runtime.as_ref().map(|rt| &rt.config) + } +} diff --git a/modules/fundamental/src/lib.rs b/modules/fundamental/src/lib.rs index c8b3c183e..e6bd84cc3 100644 --- a/modules/fundamental/src/lib.rs +++ b/modules/fundamental/src/lib.rs @@ -4,6 +4,7 @@ pub mod advisory; pub mod common; pub mod endpoints; pub mod error; +pub mod exploit_intelligence; pub mod license; pub mod organization; pub mod product; diff --git a/modules/fundamental/src/test/common.rs b/modules/fundamental/src/test/common.rs index 9b56aae55..7ac67044a 100644 --- a/modules/fundamental/src/test/common.rs +++ b/modules/fundamental/src/test/common.rs @@ -27,6 +27,7 @@ pub async fn caller_with( ctx.storage.clone(), analysis.clone(), cache, + None, ); trustify_module_analysis::endpoints::configure(svc, db_ro, analysis); }) diff --git a/openapi.yaml b/openapi.yaml index 653e46cfe..6918acd6c 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -1033,6 +1033,121 @@ paths: description: Uploaded the dataset '400': description: The file could not be parsed as an dataset + /api/v3/exploit-intelligence/analyze: + post: + tags: + - exploit-intelligence + summary: Trigger an Exploit Intelligence analysis for a CVE against an SBOM. + description: |- + Creates a job record and spawns a background task that submits the SBOM + to the EI client service, polls for completion, and ingests any returned + VEX document. + operationId: analyzeExploitIntelligence + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/AnalyzeRequest' + required: true + responses: + '200': + description: Active analysis job already exists + content: + application/json: + schema: + $ref: '#/components/schemas/AnalyzeResponse' + '201': + description: Analysis job created + content: + application/json: + schema: + $ref: '#/components/schemas/AnalyzeResponse' + '404': + description: SBOM not found + '503': + description: Exploit Intelligence service not configured + /api/v3/exploit-intelligence/jobs: + get: + tags: + - exploit-intelligence + summary: List Exploit Intelligence analysis jobs. + operationId: listExploitIntelligenceJobs + parameters: + - 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 + - name: sbom_id + in: query + description: Filter by SBOM ID. + required: false + schema: + type: + - string + - 'null' + format: uuid + - name: status + in: query + description: Filter by job status. + required: false + schema: + oneOf: + - type: 'null' + - $ref: '#/components/schemas/JobStatus' + responses: + '200': + description: List of analysis jobs + content: + application/json: + schema: + $ref: '#/components/schemas/PaginatedResults_ExploitIntelligenceJobSummary' + /api/v3/exploit-intelligence/jobs/{id}: + get: + tags: + - exploit-intelligence + summary: Get details of a specific Exploit Intelligence analysis job. + operationId: getExploitIntelligenceJob + parameters: + - name: id + in: path + description: Job identifier + required: true + schema: + type: string + format: uuid + responses: + '200': + description: Job details + content: + application/json: + schema: + $ref: '#/components/schemas/ExploitIntelligenceJobDetails' + '404': + description: Job not found /api/v3/group/sbom: get: tags: @@ -4340,6 +4455,34 @@ components: cache: $ref: '#/components/schemas/CacheStatusDetails' description: Details about the cache + AnalyzeRequest: + type: object + description: Request body for triggering an Exploit Intelligence analysis. + required: + - sbom_id + - vulnerability_id + properties: + sbom_id: + type: string + format: uuid + description: The SBOM identifier to analyze. + vulnerability_id: + type: string + description: The vulnerability identifier (e.g., "CVE-2024-9680"). + AnalyzeResponse: + type: object + description: Response returned when an analysis job is successfully created. + required: + - job_id + - status + properties: + job_id: + type: string + format: uuid + description: The unique job identifier. + status: + $ref: '#/components/schemas/JobStatus' + description: Current lifecycle status of the job. BasePurlDetails: allOf: - $ref: '#/components/schemas/BasePurlHead' @@ -4560,6 +4703,57 @@ components: period: type: string description: The period the importer should be run. + ComponentResult: + type: object + description: Per-component analysis result within a multi-component product job. + required: + - id + - component_purl + - component_name + - status + - created + - updated + properties: + advisory_id: + type: + - string + - 'null' + format: uuid + description: FK to advisory — set when a VEX document is ingested from the result. + component_name: + type: string + description: Human-readable name of the component. + component_purl: + type: string + description: Package URL (purl) of the component being analysed. + created: + type: string + description: Timestamp when the component record was created. + error_message: + type: + - string + - 'null' + description: Error details for failed component analyses. + finding: + oneOf: + - type: 'null' + - $ref: '#/components/schemas/JobFinding' + description: Analysis result from Exploit Intelligence, if completed. + id: + type: string + format: uuid + description: The unique component record identifier. + report_url: + type: + - string + - 'null' + description: Link to the EI human-readable justification report for this component. + status: + $ref: '#/components/schemas/JobStatus' + description: Current lifecycle status of this component's analysis. + updated: + type: string + description: Timestamp when the component record was last updated. Cpe: type: string format: uri @@ -4636,6 +4830,189 @@ components: message: type: string description: A human-readable error message + ExploitIntelligenceJobDetails: + type: object + description: Detailed view of an Exploit Intelligence analysis job. + required: + - id + - vulnerability_id + - status + - components + - created + - updated + properties: + advisory_id: + type: + - string + - 'null' + format: uuid + description: FK to advisory — set when a VEX document is ingested from the result. + completed_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components (computed at query time, not stored). + components: + type: array + items: + $ref: '#/components/schemas/ComponentResult' + description: Per-component analysis results (populated for multi-component jobs). + created: + type: string + description: Timestamp when the job was created. + error_message: + type: + - string + - 'null' + description: Error details for failed analyses. + failed_components: + type: + - integer + - 'null' + format: int32 + description: Number of failed/excluded components (computed at query time, not stored). + finding: + oneOf: + - type: 'null' + - $ref: '#/components/schemas/JobFinding' + description: Analysis result from Exploit Intelligence. + id: + type: string + format: uuid + description: The unique job identifier. + not_vulnerable_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components with NotVulnerable finding (computed at query time). + product_id: + type: + - string + - 'null' + description: EI product ID for multi-component SPDX flow. + report_url: + type: + - string + - 'null' + description: Link to the EI human-readable justification report. + sbom_id: + type: + - string + - 'null' + format: uuid + description: FK to the SBOM being analysed. + status: + $ref: '#/components/schemas/JobStatus' + description: Current lifecycle status of the job. + total_components: + type: + - integer + - 'null' + format: int32 + description: Total components in the product. + uncertain_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components with Uncertain finding (computed at query time). + updated: + type: string + description: Timestamp when the job was last updated. + vulnerability_id: + type: string + description: CVE identifier. + vulnerable_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components with Vulnerable finding (computed at query time). + ExploitIntelligenceJobSummary: + type: object + description: Summary of an Exploit Intelligence analysis job. + required: + - id + - vulnerability_id + - status + - created + - updated + properties: + completed_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components (computed at query time, not stored). + created: + type: string + description: Timestamp when the job was created. + failed_components: + type: + - integer + - 'null' + format: int32 + description: Number of failed/excluded components (computed at query time, not stored). + finding: + oneOf: + - type: 'null' + - $ref: '#/components/schemas/JobFinding' + description: Analysis result from Exploit Intelligence. + id: + type: string + format: uuid + description: The unique job identifier. + not_vulnerable_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components with NotVulnerable finding (computed at query time). + product_id: + type: + - string + - 'null' + description: EI product ID for multi-component SPDX flow. + report_url: + type: + - string + - 'null' + description: Link to the EI human-readable justification report. + sbom_id: + type: + - string + - 'null' + format: uuid + description: FK to the SBOM being analysed. + status: + $ref: '#/components/schemas/JobStatus' + description: Current lifecycle status of the job. + total_components: + type: + - integer + - 'null' + format: int32 + description: Total components in the product. + uncertain_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components with Uncertain finding (computed at query time). + updated: + type: string + description: Timestamp when the job was last updated. + vulnerability_id: + type: string + description: CVE identifier. + vulnerable_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components with Vulnerable finding (computed at query time). ExternalReferenceQuery: type: object properties: @@ -4911,6 +5288,21 @@ components: items: type: string description: Warnings that occurred during the import process + JobFinding: + type: string + description: Analysis finding from Exploit Intelligence (API representation). + enum: + - vulnerable + - not_vulnerable + - uncertain + JobStatus: + type: string + description: Lifecycle status of an Exploit Intelligence analysis job (API representation). + enum: + - pending + - running + - completed + - failed Labels: type: object additionalProperties: @@ -5132,6 +5524,102 @@ components: - 'null' format: int64 minimum: 0 + PaginatedResults_ExploitIntelligenceJobSummary: + type: object + required: + - items + properties: + items: + type: array + items: + type: object + description: Summary of an Exploit Intelligence analysis job. + required: + - id + - vulnerability_id + - status + - created + - updated + properties: + completed_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components (computed at query time, not stored). + created: + type: string + description: Timestamp when the job was created. + failed_components: + type: + - integer + - 'null' + format: int32 + description: Number of failed/excluded components (computed at query time, not stored). + finding: + oneOf: + - type: 'null' + - $ref: '#/components/schemas/JobFinding' + description: Analysis result from Exploit Intelligence. + id: + type: string + format: uuid + description: The unique job identifier. + not_vulnerable_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components with NotVulnerable finding (computed at query time). + product_id: + type: + - string + - 'null' + description: EI product ID for multi-component SPDX flow. + report_url: + type: + - string + - 'null' + description: Link to the EI human-readable justification report. + sbom_id: + type: + - string + - 'null' + format: uuid + description: FK to the SBOM being analysed. + status: + $ref: '#/components/schemas/JobStatus' + description: Current lifecycle status of the job. + total_components: + type: + - integer + - 'null' + format: int32 + description: Total components in the product. + uncertain_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components with Uncertain finding (computed at query time). + updated: + type: string + description: Timestamp when the job was last updated. + vulnerability_id: + type: string + description: CVE identifier. + vulnerable_components: + type: + - integer + - 'null' + format: int32 + description: Number of completed components with Vulnerable finding (computed at query time). + total: + type: + - integer + - 'null' + format: int64 + minimum: 0 PaginatedResults_GroupDetails: type: object required: diff --git a/server/Cargo.toml b/server/Cargo.toml index 3db494c9d..47eb5388b 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -26,6 +26,7 @@ bytesize = { workspace = true } clap = { workspace = true, features = ["derive"] } futures = { workspace = true } garage-door = { workspace = true, optional = true } +humantime = { workspace = true } log = { workspace = true } rand = { workspace = true } serde = { workspace = true } diff --git a/server/src/openapi.rs b/server/src/openapi.rs index 4e8bf4348..43b292182 100644 --- a/server/src/openapi.rs +++ b/server/src/openapi.rs @@ -26,6 +26,7 @@ pub async fn create_openapi() -> anyhow::Result { auth: None, analysis, read_only: false, + ei_config: None, }, ); }) diff --git a/server/src/profile/api.rs b/server/src/profile/api.rs index 7758159c9..bfac3326b 100644 --- a/server/src/profile/api.rs +++ b/server/src/profile/api.rs @@ -5,7 +5,7 @@ use crate::{endpoints, profile::spawn_db_check, sample_data}; use actix_web::web; use bytesize::ByteSize; use futures::FutureExt; -use std::{env, process::ExitCode, sync::Arc}; +use std::{env, process::ExitCode, sync::Arc, time::Duration}; use trustify_auth::{ auth::AuthConfigArguments, authenticator::Authenticator, @@ -32,6 +32,11 @@ use trustify_infrastructure::{ otel::{Metrics as OtelMetrics, Tracing}, }; use trustify_module_analysis::{config::AnalysisConfig, service::AnalysisService}; +use trustify_module_fundamental::exploit_intelligence::{ + auth::build_provider, + runner::worker::run_worker, + service::{ExploitIntelligenceConfig, ExploitIntelligenceService}, +}; use trustify_module_ingestor::graph::Graph; use trustify_module_storage::{config::StorageConfig, service::dispatch::DispatchBackend}; use trustify_module_ui::{UI, endpoints::UiResources}; @@ -92,6 +97,10 @@ pub struct Run { )] pub scan_limit: BinaryByteSize, + /// Exploit Intelligence configuration + #[command(flatten)] + pub exploit_intelligence: ExploitIntelligenceArgs, + // flattened commands must go last // /// Analysis configuration @@ -129,6 +138,166 @@ pub struct Run { pub ui: UiConfig, } +/// All Exploit Intelligence CLI arguments. +#[derive(clap::Args, Debug)] +#[command(next_help_heading = "Exploit Intelligence")] +pub struct ExploitIntelligenceArgs { + /// Base URL of the Exploit Intelligence client service. + #[arg(long, env = "EXPLOIT_INTELLIGENCE_URL")] + pub exploit_intelligence_url: Option, + + /// Base URL of the Exploit Intelligence web UI for deep-linking to reports. + #[arg(long, env = "EXPLOIT_INTELLIGENCE_UI_URL")] + pub exploit_intelligence_ui_url: Option, + + /// Polling interval for EI analysis completion (humantime, e.g. "30s"). + #[arg( + long, + env = "EXPLOIT_INTELLIGENCE_POLL_INTERVAL", + default_value = "30s" + )] + pub exploit_intelligence_poll_interval: humantime::Duration, + + /// Maximum duration before EI polling is considered timed out (humantime, e.g. "30m"). + #[arg( + long, + env = "EXPLOIT_INTELLIGENCE_MAX_POLL_DURATION", + default_value = "30m" + )] + pub exploit_intelligence_max_poll_duration: humantime::Duration, + + /// Maximum number of upload retry attempts for EI analysis. + #[arg( + long, + env = "EXPLOIT_INTELLIGENCE_UPLOAD_MAX_RETRIES", + default_value_t = 3 + )] + pub exploit_intelligence_upload_max_retries: usize, + + /// Initial delay between upload retries (humantime, exponential backoff). + #[arg( + long, + env = "EXPLOIT_INTELLIGENCE_UPLOAD_RETRY_DELAY", + default_value = "1s" + )] + pub exploit_intelligence_upload_retry_delay: humantime::Duration, + + /// Maximum consecutive poll failures before giving up. + #[arg( + long, + env = "EXPLOIT_INTELLIGENCE_MAX_CONSECUTIVE_POLL_FAILURES", + default_value_t = 5 + )] + pub exploit_intelligence_max_consecutive_poll_failures: u32, + + /// Authentication token for the Exploit Intelligence service (static token, for backward compatibility). + #[arg(long, env = "EXPLOIT_INTELLIGENCE_AUTH_TOKEN")] + pub exploit_intelligence_auth_token: Option, + + #[command(flatten)] + pub oidc: EiOidcArguments, +} + +impl ExploitIntelligenceArgs { + /// Convert CLI arguments into an optional `ExploitIntelligenceConfig`. + /// + /// Returns `None` when no EI URL is configured (feature disabled). + pub async fn into_config(self) -> Result, anyhow::Error> { + let Some(url) = self.exploit_intelligence_url else { + return Ok(None); + }; + + let token_provider = build_provider( + self.oidc.into_config(), + self.exploit_intelligence_auth_token, + ) + .await?; + + Ok(Some(ExploitIntelligenceConfig { + url, + ui_url: self.exploit_intelligence_ui_url, + poll_interval: self.exploit_intelligence_poll_interval.into(), + max_poll_duration: self.exploit_intelligence_max_poll_duration.into(), + upload_max_retries: self.exploit_intelligence_upload_max_retries, + upload_retry_delay: self.exploit_intelligence_upload_retry_delay.into(), + max_consecutive_poll_failures: self.exploit_intelligence_max_consecutive_poll_failures, + token_provider, + })) + } +} + +/// Clap arguments for EI-specific OIDC credentials. +/// +/// Separate from the main Trustify OIDC config because the EI service may +/// use a different IdP or client registration. +#[derive(clap::Args, Debug, Clone)] +#[command(next_help_heading = "Exploit Intelligence OIDC")] +pub struct EiOidcArguments { + #[arg( + id = "ei_oidc_client_id", + long = "ei-oidc-client-id", + env = "EXPLOIT_INTELLIGENCE_OIDC_CLIENT_ID" + )] + pub client_id: Option, + + #[arg( + id = "ei_oidc_client_secret", + long = "ei-oidc-client-secret", + env = "EXPLOIT_INTELLIGENCE_OIDC_CLIENT_SECRET" + )] + pub client_secret: Option, + + #[arg( + id = "ei_oidc_issuer_url", + long = "ei-oidc-issuer-url", + env = "EXPLOIT_INTELLIGENCE_OIDC_ISSUER_URL" + )] + pub issuer_url: Option, + + #[arg( + id = "ei_oidc_refresh_before", + long = "ei-oidc-refresh-before", + env = "EXPLOIT_INTELLIGENCE_OIDC_REFRESH_BEFORE", + default_value = "30s" + )] + pub refresh_before: humantime::Duration, + + #[arg( + id = "ei_oidc_tls_insecure", + long = "ei-oidc-tls-insecure", + env = "EXPLOIT_INTELLIGENCE_OIDC_TLS_INSECURE", + default_value = "false" + )] + pub tls_insecure: bool, +} + +impl EiOidcArguments { + fn into_config(self) -> Option { + match (self.client_id, self.client_secret, self.issuer_url) { + (Some(client_id), Some(client_secret), Some(issuer_url)) => { + Some(trustify_auth::client::OpenIdTokenProviderConfig { + client_id, + client_secret, + issuer_url, + refresh_before: self.refresh_before, + tls_insecure: self.tls_insecure, + }) + } + (None, None, None) => None, + _ => { + log::warn!( + "Incomplete EI OIDC configuration: all three of \ + EXPLOIT_INTELLIGENCE_OIDC_CLIENT_ID, \ + EXPLOIT_INTELLIGENCE_OIDC_CLIENT_SECRET, and \ + EXPLOIT_INTELLIGENCE_OIDC_ISSUER_URL must be set together. \ + Falling back to static token or no auth." + ); + None + } + } + } +} + mod default { use bytesize::ByteSize; use trustify_common::model::BinaryByteSize; @@ -194,6 +363,7 @@ struct InitData { config: ModuleConfig, analysis: AnalysisService, read_only: bool, + ei_config: Option, } /// Groups all module configurations. @@ -298,6 +468,8 @@ impl InitData { }, }; + let ei_config = run.exploit_intelligence.into_config().await?; + Ok(InitData { analysis: AnalysisService::new(run.analysis, db_ro.clone()), authenticator, @@ -315,6 +487,7 @@ impl InitData { embedded_oidc, ui, read_only: run.read_only, + ei_config, }) } @@ -322,6 +495,40 @@ impl InitData { async fn run(mut self) -> anyhow::Result<()> { let ui = Arc::new(UiResources::new(&self.ui)?); + // Build the EI worker task before the HTTP server captures `self`. + // The worker needs an IngestorService, which is also created inside + // the module configuration — build one here and clone it for the worker. + let ei_worker_task = if let Some(ref ei_config) = self.ei_config { + if !self.read_only { + let ei_service = ExploitIntelligenceService::new(Some(ei_config.clone()))?; + let ingestor_for_worker = trustify_module_ingestor::service::IngestorService::new( + Graph::new(), + self.storage.clone(), + Some(self.analysis.clone()), + ); + let db_rw = self.db_rw.clone(); + let db_ro = self.db_ro.clone(); + + Some( + async move { + run_worker( + ei_service, + ingestor_for_worker, + db_rw, + db_ro, + Duration::from_secs(5), // worker poll interval + ) + .await + } + .boxed_local(), + ) + } else { + None + } + } else { + None + }; + let http = { HttpServerBuilder::try_from(self.http)? .tracing(self.tracing) @@ -341,6 +548,7 @@ impl InitData { auth: self.authenticator.clone(), analysis: self.analysis.clone(), read_only: self.read_only, + ei_config: self.ei_config.clone(), }, ); }) @@ -351,6 +559,10 @@ impl InitData { #[allow(unused_mut)] let mut tasks = vec![http]; + if let Some(worker) = ei_worker_task { + tasks.push(worker); + } + // track the embedded OIDC server task #[cfg(feature = "garage-door")] if let Some(embedded_oidc) = self.embedded_oidc.take() { @@ -391,6 +603,7 @@ pub(crate) struct Config { pub(crate) analysis: AnalysisService, pub(crate) auth: Option>, pub(crate) read_only: bool, + pub(crate) ei_config: Option, } pub(crate) fn configure(svc: &mut utoipa_actix_web::service_config::ServiceConfig, config: Config) { @@ -408,6 +621,7 @@ pub(crate) fn configure(svc: &mut utoipa_actix_web::service_config::ServiceConfi auth, analysis, read_only, + ei_config, } = config; let graph = Graph::new(); @@ -441,6 +655,7 @@ pub(crate) fn configure(svc: &mut utoipa_actix_web::service_config::ServiceConfi storage, analysis.clone(), cache, + ei_config.clone(), ); trustify_module_analysis::endpoints::configure(svc, db_ro.clone(), analysis); trustify_module_user::endpoints::configure(svc); @@ -521,6 +736,7 @@ mod test { auth: None, analysis, read_only: false, + ei_config: None, }, ); }) @@ -594,6 +810,7 @@ mod test { auth: None, analysis, read_only, + ei_config: None, }, ); })