1164 lines
45 KiB
Rust
1164 lines
45 KiB
Rust
use std::sync::Arc;
|
|
|
|
use mongodb::bson::doc;
|
|
use tracing::Instrument;
|
|
|
|
use compliance_core::models::*;
|
|
use compliance_core::traits::Scanner;
|
|
use compliance_core::AgentConfig;
|
|
|
|
use crate::database::Database;
|
|
use crate::error::AgentError;
|
|
use crate::llm::LlmClient;
|
|
use crate::pipeline::cve::CveScanner;
|
|
use crate::pipeline::git::GitOps;
|
|
use crate::pipeline::gitleaks::GitleaksScanner;
|
|
use crate::pipeline::lint::LintScanner;
|
|
use crate::pipeline::patterns::{GdprPatternScanner, OAuthPatternScanner};
|
|
use crate::pipeline::plan::build_scan_plan;
|
|
use crate::pipeline::repo_view::RepoView;
|
|
use crate::pipeline::sbom::SbomScanner;
|
|
use crate::pipeline::semgrep::SemgrepScanner;
|
|
|
|
/// Context from graph analysis passed to LLM triage for enhanced filtering
|
|
#[derive(Debug)]
|
|
#[allow(dead_code)]
|
|
pub struct GraphContext {
|
|
pub node_count: u32,
|
|
pub edge_count: u32,
|
|
pub community_count: u32,
|
|
pub impacts: Vec<compliance_core::models::graph::ImpactAnalysis>,
|
|
}
|
|
|
|
pub struct PipelineOrchestrator {
|
|
pub(super) config: AgentConfig,
|
|
pub(super) db: Database,
|
|
pub(super) llm: Arc<LlmClient>,
|
|
pub(super) http: reqwest::Client,
|
|
}
|
|
|
|
impl PipelineOrchestrator {
|
|
pub fn new(
|
|
config: AgentConfig,
|
|
db: Database,
|
|
llm: Arc<LlmClient>,
|
|
http: reqwest::Client,
|
|
) -> Self {
|
|
Self {
|
|
config,
|
|
db,
|
|
llm,
|
|
http,
|
|
}
|
|
}
|
|
|
|
#[tracing::instrument(skip_all, fields(repo_id = repo.name.as_str()))]
|
|
async fn run_pipeline(&self, repo: &RepoView, scan_run_id: &str) -> Result<u32, AgentError> {
|
|
let repo_id = repo.id.as_ref().map(|id| id.to_hex()).unwrap_or_default();
|
|
|
|
// Stage 0: Change detection
|
|
tracing::info!("[{repo_id}] Stage 0: Change detection");
|
|
let creds = GitOps::make_repo_credentials(&self.config, repo);
|
|
let git_ops = GitOps::new(&self.config.git_clone_base_path, creds);
|
|
let repo_path = git_ops.clone_or_fetch(&repo.git_url, &repo.name)?;
|
|
|
|
if !GitOps::has_new_commits(&repo_path, repo.last_scanned_commit.as_deref())? {
|
|
tracing::info!("[{repo_id}] No new commits, skipping scan");
|
|
return Ok(0);
|
|
}
|
|
|
|
let mut all_findings: Vec<Finding> = Vec::new();
|
|
|
|
// Stage 1: Semgrep SAST
|
|
tracing::info!("[{repo_id}] Stage 1: Semgrep SAST");
|
|
self.update_phase(scan_run_id, "sast").await;
|
|
match async {
|
|
let semgrep = SemgrepScanner;
|
|
semgrep.scan(&repo_path, &repo_id).await
|
|
}
|
|
.instrument(tracing::info_span!("stage_sast"))
|
|
.await
|
|
{
|
|
Ok(output) => all_findings.extend(output.findings),
|
|
Err(e) => tracing::warn!("[{repo_id}] Semgrep failed: {e}"),
|
|
}
|
|
|
|
// Stage 2: SBOM Generation
|
|
tracing::info!("[{repo_id}] Stage 2: SBOM Generation");
|
|
self.update_phase(scan_run_id, "sbom_generation").await;
|
|
let mut sbom_entries = match async {
|
|
let sbom_scanner = SbomScanner;
|
|
sbom_scanner.scan(&repo_path, &repo_id).await
|
|
}
|
|
.instrument(tracing::info_span!("stage_sbom_generation"))
|
|
.await
|
|
{
|
|
Ok(output) => output.sbom_entries,
|
|
Err(e) => {
|
|
tracing::warn!("[{repo_id}] SBOM generation failed: {e}");
|
|
Vec::new()
|
|
}
|
|
};
|
|
|
|
// Stage 3: CVE Scanning
|
|
tracing::info!("[{repo_id}] Stage 3: CVE Scanning");
|
|
self.update_phase(scan_run_id, "cve_scanning").await;
|
|
let cve_scanner = CveScanner::new(
|
|
self.http.clone(),
|
|
self.config.searxng_url.clone(),
|
|
self.config.nvd_api_key.as_ref().map(|k| {
|
|
use secrecy::ExposeSecret;
|
|
k.expose_secret().to_string()
|
|
}),
|
|
);
|
|
let cve_alerts = match tokio::time::timeout(
|
|
std::time::Duration::from_secs(600),
|
|
async {
|
|
cve_scanner
|
|
.scan_dependencies(&repo_id, &mut sbom_entries)
|
|
.await
|
|
}
|
|
.instrument(tracing::info_span!("stage_cve_scanning")),
|
|
)
|
|
.await
|
|
{
|
|
Ok(Ok(alerts)) => alerts,
|
|
Ok(Err(e)) => {
|
|
tracing::warn!("[{repo_id}] CVE scanning failed: {e}");
|
|
Vec::new()
|
|
}
|
|
Err(_) => {
|
|
tracing::warn!("[{repo_id}] CVE scanning timed out after 10 minutes");
|
|
Vec::new()
|
|
}
|
|
};
|
|
|
|
// Stage 4: Pattern Scanning (GDPR + OAuth)
|
|
tracing::info!("[{repo_id}] Stage 4: Pattern Scanning");
|
|
self.update_phase(scan_run_id, "pattern_scanning").await;
|
|
{
|
|
let pattern_findings = async {
|
|
let mut findings = Vec::new();
|
|
let gdpr = GdprPatternScanner::new();
|
|
match gdpr.scan(&repo_path, &repo_id).await {
|
|
Ok(output) => findings.extend(output.findings),
|
|
Err(e) => tracing::warn!("[{repo_id}] GDPR pattern scan failed: {e}"),
|
|
}
|
|
let oauth = OAuthPatternScanner::new();
|
|
match oauth.scan(&repo_path, &repo_id).await {
|
|
Ok(output) => findings.extend(output.findings),
|
|
Err(e) => tracing::warn!("[{repo_id}] OAuth pattern scan failed: {e}"),
|
|
}
|
|
findings
|
|
}
|
|
.instrument(tracing::info_span!("stage_pattern_scanning"))
|
|
.await;
|
|
all_findings.extend(pattern_findings);
|
|
}
|
|
|
|
// Stage 4a: Secret Detection (Gitleaks)
|
|
tracing::info!("[{repo_id}] Stage 4a: Secret Detection");
|
|
self.update_phase(scan_run_id, "secret_detection").await;
|
|
match async {
|
|
let gitleaks = GitleaksScanner;
|
|
gitleaks.scan(&repo_path, &repo_id).await
|
|
}
|
|
.instrument(tracing::info_span!("stage_secret_detection"))
|
|
.await
|
|
{
|
|
Ok(output) => all_findings.extend(output.findings),
|
|
Err(e) => tracing::warn!("[{repo_id}] Gitleaks failed: {e}"),
|
|
}
|
|
|
|
// Stage 4b: Lint Scanning
|
|
tracing::info!("[{repo_id}] Stage 4b: Lint Scanning");
|
|
self.update_phase(scan_run_id, "lint_scanning").await;
|
|
match async {
|
|
let lint = LintScanner;
|
|
lint.scan(&repo_path, &repo_id).await
|
|
}
|
|
.instrument(tracing::info_span!("stage_lint_scanning"))
|
|
.await
|
|
{
|
|
Ok(output) => all_findings.extend(output.findings),
|
|
Err(e) => tracing::warn!("[{repo_id}] Lint scanning failed: {e}"),
|
|
}
|
|
|
|
// Stage 4.5: Graph Building
|
|
tracing::info!("[{repo_id}] Stage 4.5: Graph Building");
|
|
self.update_phase(scan_run_id, "graph_building").await;
|
|
let graph_context = match async {
|
|
self.build_code_graph(&repo_path, &repo_id, &all_findings)
|
|
.await
|
|
}
|
|
.instrument(tracing::info_span!("stage_graph_building"))
|
|
.await
|
|
{
|
|
Ok(ctx) => Some(ctx),
|
|
Err(e) => {
|
|
tracing::warn!("[{repo_id}] Graph building failed: {e}");
|
|
None
|
|
}
|
|
};
|
|
|
|
// Stage 5: LLM Triage (enhanced with graph context)
|
|
tracing::info!(
|
|
"[{repo_id}] Stage 5: LLM Triage ({} findings)",
|
|
all_findings.len()
|
|
);
|
|
self.update_phase(scan_run_id, "llm_triage").await;
|
|
let triaged = crate::llm::triage::triage_findings(
|
|
&self.llm,
|
|
&mut all_findings,
|
|
graph_context.as_ref(),
|
|
)
|
|
.await;
|
|
tracing::info!("[{repo_id}] Triaged: {triaged} findings passed confidence threshold");
|
|
|
|
// Stage 5b: control triage — stamp findings with the compliance control(s)
|
|
// they're evidence for and flag control false positives (grounded LLM over
|
|
// deterministic tool output). No-op unless breakpilot is configured.
|
|
self.update_phase(scan_run_id, "control_triage").await;
|
|
let tagged = crate::controls::triage_repo_findings(
|
|
&self.config,
|
|
self.llm.clone(),
|
|
&repo_path,
|
|
&mut all_findings,
|
|
)
|
|
.await;
|
|
if tagged > 0 {
|
|
tracing::info!("[{repo_id}] Control triage tagged {tagged} findings with control refs");
|
|
}
|
|
|
|
// Stage 5c: semantic control mapping — scale path for the master-controls
|
|
// corpus (no CWE to LUT on): embed each finding's region, retrieve the
|
|
// nearest master controls, grounded-judge, and stamp confirmed refs. On by
|
|
// default (validated live); the corpus embedding is cached so only the
|
|
// first scan after a catalog change pays it.
|
|
if self.config.breakpilot.semantic_mapping {
|
|
self.update_phase(scan_run_id, "semantic_control_mapping")
|
|
.await;
|
|
let sem = crate::controls::semantic_stamp_findings(
|
|
&self.config,
|
|
self.llm.clone(),
|
|
&repo_path,
|
|
&mut all_findings,
|
|
)
|
|
.await;
|
|
if sem > 0 {
|
|
tracing::info!(
|
|
"[{repo_id}] Semantic mapping tagged {sem} findings with master-control refs"
|
|
);
|
|
}
|
|
}
|
|
|
|
// Stage 5d: grounded surface checks — the absence-based controls (no
|
|
// rate limiting, no security logging, no update-signature check) have no
|
|
// syntactic pattern to match, so we retrieve the code surface each governs
|
|
// and let the grounded judge decide whether it holds, producing net-new
|
|
// findings already tagged + grounded. On by default (validated live); it
|
|
// covers the 8 absence-based CRA controls.
|
|
if self.config.breakpilot.grounded_control_checks {
|
|
self.update_phase(scan_run_id, "grounded_control_checks")
|
|
.await;
|
|
let grounded = crate::controls::grounded_surface_findings(
|
|
&self.config,
|
|
self.llm.clone(),
|
|
&repo_path,
|
|
&repo_id,
|
|
)
|
|
.await;
|
|
if !grounded.is_empty() {
|
|
tracing::info!(
|
|
"[{repo_id}] Grounded surface checks raised {} control findings",
|
|
grounded.len()
|
|
);
|
|
all_findings.extend(grounded);
|
|
}
|
|
}
|
|
|
|
// Dedup against existing findings and insert new ones
|
|
let mut new_count = 0u32;
|
|
let mut new_findings: Vec<Finding> = Vec::new();
|
|
for mut finding in all_findings {
|
|
finding.scan_run_id = Some(scan_run_id.to_string());
|
|
// Check if fingerprint already exists
|
|
let existing = self
|
|
.db
|
|
.findings()
|
|
.find_one(doc! { "fingerprint": &finding.fingerprint })
|
|
.await?;
|
|
if existing.is_none() {
|
|
let result = self.db.findings().insert_one(&finding).await?;
|
|
finding.id = result.inserted_id.as_object_id();
|
|
new_findings.push(finding);
|
|
new_count += 1;
|
|
}
|
|
}
|
|
|
|
// Remove stale SBOM entries for this repo before reinserting
|
|
if !sbom_entries.is_empty() {
|
|
self.db
|
|
.sbom_entries()
|
|
.delete_many(doc! { "repo_id": &repo.id })
|
|
.await?;
|
|
}
|
|
|
|
// Persist SBOM entries
|
|
for entry in &sbom_entries {
|
|
let filter = doc! {
|
|
"repo_id": &entry.repo_id,
|
|
"name": &entry.name,
|
|
"version": &entry.version,
|
|
};
|
|
let update = mongodb::bson::to_document(entry)
|
|
.map(|d| doc! { "$set": d })
|
|
.unwrap_or_else(|_| doc! {});
|
|
self.db
|
|
.sbom_entries()
|
|
.update_one(filter, update)
|
|
.upsert(true)
|
|
.await?;
|
|
}
|
|
|
|
// Persist CVE alerts and create notifications (shared with the PLC path).
|
|
let new_notif_count = self
|
|
.persist_cve_alerts(&repo_id, &repo.name, &cve_alerts)
|
|
.await?;
|
|
if new_notif_count > 0 {
|
|
tracing::info!("[{repo_id}] Created {new_notif_count} CVE notification(s)");
|
|
}
|
|
|
|
// Stage 6: Issue Creation
|
|
tracing::info!("[{repo_id}] Stage 6: Issue Creation");
|
|
self.update_phase(scan_run_id, "issue_creation").await;
|
|
if let Err(e) = self
|
|
.create_tracker_issues(repo, &repo_id, &new_findings)
|
|
.await
|
|
{
|
|
tracing::warn!("[{repo_id}] Issue creation failed: {e}");
|
|
}
|
|
|
|
// The onboarded target's findings_count and the git artifact's
|
|
// last_scanned_commit watermark are persisted by `finalize_target` after
|
|
// `run_pipeline` returns.
|
|
|
|
// Stage 8: DAST (async, optional — only if a DastTarget is configured)
|
|
tracing::info!("[{repo_id}] Stage 8: Checking for DAST targets");
|
|
self.update_phase(scan_run_id, "dast_scanning").await;
|
|
self.maybe_trigger_dast(&repo_id, scan_run_id).await;
|
|
|
|
tracing::info!("[{repo_id}] Scan complete: {new_count} new findings");
|
|
Ok(new_count)
|
|
}
|
|
|
|
/// Unified entry point (behind `UNIFIED_PIPELINE`): run a scan for an
|
|
/// `OnboardedTarget`. Mirrors [`Self::run`] but sources the target from
|
|
/// `onboarded_targets` and dispatches by the scan plan.
|
|
#[tracing::instrument(skip_all, fields(target_id = %target_id, trigger = ?trigger))]
|
|
pub async fn run_target(
|
|
&self,
|
|
target_id: &str,
|
|
trigger: ScanTrigger,
|
|
) -> Result<(), AgentError> {
|
|
let oid = mongodb::bson::oid::ObjectId::parse_str(target_id)
|
|
.map_err(|e| AgentError::Other(e.to_string()))?;
|
|
let target = self
|
|
.db
|
|
.onboarded_targets()
|
|
.find_one(doc! { "_id": oid })
|
|
.await?
|
|
.ok_or_else(|| AgentError::Other(format!("Onboarded target {target_id} not found")))?;
|
|
|
|
let scan_run = ScanRun::new(target_id.to_string(), trigger);
|
|
let insert = self.db.scan_runs().insert_one(&scan_run).await?;
|
|
let scan_run_id = insert
|
|
.inserted_id
|
|
.as_object_id()
|
|
.map(|id| id.to_hex())
|
|
.unwrap_or_default();
|
|
|
|
let result = self.run_target_pipeline(&target, &scan_run_id).await;
|
|
match &result {
|
|
Ok(count) => {
|
|
self.db
|
|
.scan_runs()
|
|
.update_one(
|
|
doc! { "_id": &insert.inserted_id },
|
|
doc! { "$set": {
|
|
"status": "completed",
|
|
"current_phase": "completed",
|
|
"new_findings_count": *count as i64,
|
|
"completed_at": mongodb::bson::DateTime::now(),
|
|
} },
|
|
)
|
|
.await?;
|
|
// Refresh the target's cached findings count. The shared pipeline
|
|
// (Stage 7) increments `repositories`, which the unified path does
|
|
// not use, so set the accurate total on the target itself.
|
|
let total = self
|
|
.db
|
|
.findings()
|
|
.count_documents(doc! { "repo_id": target_id })
|
|
.await
|
|
.unwrap_or(*count as u64);
|
|
self.db
|
|
.onboarded_targets()
|
|
.update_one(
|
|
doc! { "_id": oid },
|
|
doc! { "$set": {
|
|
"findings_count": total as i64,
|
|
"updated_at": mongodb::bson::DateTime::now(),
|
|
} },
|
|
)
|
|
.await?;
|
|
}
|
|
Err(e) => {
|
|
tracing::error!(target_id, error = %e, "Unified scan pipeline failed");
|
|
self.db
|
|
.scan_runs()
|
|
.update_one(
|
|
doc! { "_id": &insert.inserted_id },
|
|
doc! { "$set": {
|
|
"status": "failed",
|
|
"error_message": e.to_string(),
|
|
"completed_at": mongodb::bson::DateTime::now(),
|
|
} },
|
|
)
|
|
.await?;
|
|
}
|
|
}
|
|
result.map(|_| ())
|
|
}
|
|
|
|
/// Run the applicable scans for a target. For a code target this reuses the
|
|
/// full legacy pipeline over the code artifact (clone → SAST umbrella →
|
|
/// triage → persist → issues → DAST); firmware/PLC/mobile scanners are
|
|
/// follow-ups (#128/#129/#130). Returns the number of new findings.
|
|
async fn run_target_pipeline(
|
|
&self,
|
|
target: &OnboardedTarget,
|
|
scan_run_id: &str,
|
|
) -> Result<u32, AgentError> {
|
|
let target_id = target.id.map(|id| id.to_hex()).unwrap_or_default();
|
|
let plan = build_scan_plan(target);
|
|
tracing::info!(
|
|
target_id = %target_id,
|
|
target_type = %target.target_type,
|
|
planned_steps = plan.steps.len(),
|
|
"Unified pipeline: scan plan built"
|
|
);
|
|
|
|
// Ingest + classify (tramiton for firmware) and store the detected type.
|
|
self.classify_and_store(target, &target_id, scan_run_id)
|
|
.await;
|
|
// Provision a DAST target from a LiveUrl artifact so DAST fires for
|
|
// wizard-created targets, not just migrated ones.
|
|
self.ensure_dast_target(target, &plan).await;
|
|
|
|
// PLC/SPS targets: the control-logic scan consumes the PLC source (an
|
|
// uploaded PlcProject *or* a git repo / source archive of PLCopen XML / ST
|
|
// exports), so it takes over the code artifact — we don't also run the
|
|
// SAST pipeline over it. A PLC device is reachable, so DAST still runs
|
|
// against a WebVisu / exposed endpoint when one is provisioned.
|
|
let mut new_count = 0u32;
|
|
let plc = plan.has(ScanType::PlcControlLogic);
|
|
let ics = plan.has(ScanType::IcsProbe);
|
|
if plc {
|
|
new_count += self.run_plc_scan(target, &target_id, scan_run_id).await?;
|
|
// Provision-and-test (#183): with the control logic but no reachable
|
|
// device, instantiate it on an ephemeral soft-PLC and probe that
|
|
// instead of the customer's OT network. Opt-in (needs Docker) and only
|
|
// when there is no live URL to probe directly. Never fails the scan.
|
|
if self.config.plc_runtime.enabled && target.live_url().is_none() {
|
|
match self
|
|
.run_provisioned_plc_test(target, &target_id, scan_run_id)
|
|
.await
|
|
{
|
|
Ok(n) => new_count += n,
|
|
Err(e) => {
|
|
tracing::warn!(target_id = %target_id, error = %e, "provision-and-test failed")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if ics {
|
|
new_count += self.run_ics_probe(target, &target_id, scan_run_id).await?;
|
|
}
|
|
if plc || ics {
|
|
// PLC/SPS device: also DAST against a WebVisu / exposed endpoint, but
|
|
// only when DAST is actually planned — a device reachable only over an
|
|
// industrial protocol (e.g. modbus://) has no web surface to crawl, and
|
|
// running DAST there just fails at reconnaissance. Gating here (not only
|
|
// at provisioning) also stops a DAST target left over from an earlier
|
|
// run from re-triggering. The control-logic scan already consumed the
|
|
// code artifact, so the SAST pipeline is not re-run.
|
|
if plan.has(ScanType::Dast) {
|
|
self.update_phase(scan_run_id, "dast_scanning").await;
|
|
self.maybe_trigger_dast(&target_id, scan_run_id).await;
|
|
}
|
|
return Ok(new_count);
|
|
}
|
|
|
|
match target.code_artifact() {
|
|
Some(code) if code.kind == ArtifactKind::GitRepo => {
|
|
let repo = RepoView::from_target(target, code);
|
|
let n = self.run_pipeline(&repo, scan_run_id).await?;
|
|
self.finalize_target(target, &repo, n).await?;
|
|
new_count += n;
|
|
}
|
|
Some(_) => {
|
|
tracing::warn!(
|
|
target_id = %target_id,
|
|
"Unified pipeline: source-archive scanning not yet wired; skipping"
|
|
);
|
|
}
|
|
None => {
|
|
// No code to scan (a migrated DAST target). Firmware/mobile static
|
|
// scanners land in #128/#129; DAST for a running URL works when a
|
|
// DastTarget row exists (provisioned above from a LiveUrl, or from
|
|
// a migrated target).
|
|
tracing::info!(
|
|
target_id = %target_id,
|
|
"Unified pipeline: no code artifact; attempting DAST"
|
|
);
|
|
self.update_phase(scan_run_id, "dast_scanning").await;
|
|
self.maybe_trigger_dast(&target_id, scan_run_id).await;
|
|
}
|
|
}
|
|
Ok(new_count)
|
|
}
|
|
|
|
/// Analyze a PLC/SPS project (Structured Text / PLCopen XML) for
|
|
/// control-logic security issues and persist the new findings.
|
|
async fn run_plc_scan(
|
|
&self,
|
|
target: &OnboardedTarget,
|
|
target_id: &str,
|
|
scan_run_id: &str,
|
|
) -> Result<u32, AgentError> {
|
|
tracing::info!(target_id, "[{target_id}] PLC control-logic analysis");
|
|
self.update_phase(scan_run_id, "plc_analysis").await;
|
|
|
|
let ctx = crate::ingest::IngestContext::from_config(&self.config, target_id);
|
|
let ingest_set = crate::ingest::ingest_all(target, &ctx)?;
|
|
// Every PLC-source artifact on the target: dedicated PLC projects plus any
|
|
// code artifacts (git repo / source archive) holding PLCopen XML / ST
|
|
// exports. A target can carry several (e.g. one POU export per file).
|
|
let sources: Vec<&Artifact> = target
|
|
.artifacts
|
|
.iter()
|
|
.filter(|a| {
|
|
matches!(
|
|
a.kind,
|
|
ArtifactKind::PlcProject | ArtifactKind::GitRepo | ArtifactKind::SourceArchive
|
|
)
|
|
})
|
|
.collect();
|
|
if sources.is_empty() {
|
|
tracing::warn!(target_id, "PLC scan: no PLC source artifact");
|
|
return Ok(0);
|
|
}
|
|
|
|
let mut all_findings = Vec::new();
|
|
let mut all_sbom: Vec<SbomEntry> = Vec::new();
|
|
let mut sbom_seen = std::collections::BTreeSet::new();
|
|
for a in &sources {
|
|
let Some(path) = ingest_set.get(&a.id).and_then(|ia| ia.working_path.clone()) else {
|
|
continue;
|
|
};
|
|
all_findings.extend(crate::pipeline::plc::analyze_tree(&path, target_id));
|
|
// Control-application SBOM: CODESYS libraries + runtime from a
|
|
// `.projectarchive` (uploaded, or committed in the working tree).
|
|
let archive = a
|
|
.stored_path
|
|
.clone()
|
|
.unwrap_or_else(|| a.source_ref.clone());
|
|
for e in crate::pipeline::plc::sbom::collect_sbom(
|
|
std::path::Path::new(&archive),
|
|
&path,
|
|
target_id,
|
|
) {
|
|
if sbom_seen.insert((e.name.clone(), e.version.clone())) {
|
|
all_sbom.push(e);
|
|
}
|
|
}
|
|
}
|
|
tracing::info!(
|
|
target_id,
|
|
artifacts = sources.len(),
|
|
found = all_findings.len(),
|
|
"PLC control-logic analysis complete"
|
|
);
|
|
|
|
let mut new_count = 0u32;
|
|
for mut finding in all_findings {
|
|
finding.scan_run_id = Some(scan_run_id.to_string());
|
|
if self
|
|
.db
|
|
.findings()
|
|
.find_one(doc! { "fingerprint": &finding.fingerprint })
|
|
.await?
|
|
.is_none()
|
|
{
|
|
self.db.findings().insert_one(&finding).await?;
|
|
new_count += 1;
|
|
}
|
|
}
|
|
|
|
if !all_sbom.is_empty() {
|
|
if let Err(e) = self
|
|
.persist_control_app_sbom(target_id, &target.name, all_sbom)
|
|
.await
|
|
{
|
|
tracing::warn!(target_id, error = %e, "control-app SBOM persist failed");
|
|
}
|
|
}
|
|
Ok(new_count)
|
|
}
|
|
|
|
/// Provision-and-test (#183): instantiate the target's control logic on an
|
|
/// ephemeral soft-PLC (OpenPLC), start it, probe the provisioned Modbus
|
|
/// endpoint, and tear the instance down. Used when a PLC/SPS target has the
|
|
/// control logic but no reachable live device to probe directly. Guarded by
|
|
/// `plc_runtime.enabled` (needs Docker); persists the same [`ScanType::IcsProbe`]
|
|
/// findings as a live probe.
|
|
async fn run_provisioned_plc_test(
|
|
&self,
|
|
target: &OnboardedTarget,
|
|
target_id: &str,
|
|
scan_run_id: &str,
|
|
) -> Result<u32, AgentError> {
|
|
self.update_phase(scan_run_id, "plc_provision").await;
|
|
|
|
// Locate a loadable control-logic program among the PLC-source artifacts
|
|
// (same selection as the static PLC scan: dedicated PLC projects plus code
|
|
// artifacts holding PLCopen XML / ST exports).
|
|
let ctx = crate::ingest::IngestContext::from_config(&self.config, target_id);
|
|
let ingest_set = crate::ingest::ingest_all(target, &ctx)?;
|
|
let program = target
|
|
.artifacts
|
|
.iter()
|
|
.filter(|a| {
|
|
matches!(
|
|
a.kind,
|
|
ArtifactKind::PlcProject | ArtifactKind::GitRepo | ArtifactKind::SourceArchive
|
|
)
|
|
})
|
|
.find_map(|a| {
|
|
let path = ingest_set
|
|
.get(&a.id)
|
|
.and_then(|ia| ia.working_path.clone())?;
|
|
werkbank_exec::plc::extract_program(&path)
|
|
});
|
|
let Some(program) = program else {
|
|
tracing::info!(
|
|
target_id,
|
|
"provision-and-test: no loadable control-logic program"
|
|
);
|
|
return Ok(0);
|
|
};
|
|
|
|
let http = werkbank_exec::plc::http_client()?;
|
|
let provisioner = werkbank_exec::plc::DockerSoftPlc::new(self.config.plc_runtime.clone());
|
|
let outcome = werkbank_exec::plc::provision_and_test(
|
|
&provisioner,
|
|
&http,
|
|
&self.config.plc_runtime,
|
|
&program,
|
|
target_id,
|
|
)
|
|
.await?;
|
|
tracing::info!(
|
|
target_id,
|
|
found = outcome.findings.len(),
|
|
dast = outcome.dast.is_some(),
|
|
"provision-and-test complete"
|
|
);
|
|
|
|
let mut new_count = 0u32;
|
|
for mut finding in outcome.findings {
|
|
finding.scan_run_id = Some(scan_run_id.to_string());
|
|
if self
|
|
.db
|
|
.findings()
|
|
.find_one(doc! { "fingerprint": &finding.fingerprint })
|
|
.await?
|
|
.is_none()
|
|
{
|
|
self.db.findings().insert_one(&finding).await?;
|
|
new_count += 1;
|
|
}
|
|
}
|
|
|
|
// Persist the DAST scan of the provisioned web endpoint, linked to this
|
|
// scan run (mirrors `maybe_trigger_dast`).
|
|
if let Some(dast) = outcome.dast {
|
|
let mut scan_run = dast.scan_run;
|
|
scan_run.sast_scan_run_id = Some(scan_run_id.to_string());
|
|
if let Err(e) = self.db.dast_scan_runs().insert_one(&scan_run).await {
|
|
tracing::warn!(target_id, error = %e, "failed to store provisioned DAST scan run");
|
|
}
|
|
for finding in &dast.findings {
|
|
if let Err(e) = self.db.dast_findings().insert_one(finding).await {
|
|
tracing::warn!(target_id, error = %e, "failed to store provisioned DAST finding");
|
|
}
|
|
}
|
|
}
|
|
Ok(new_count)
|
|
}
|
|
|
|
/// Probe a running PLC/SPS device over industrial protocols (Modbus/TCP, …)
|
|
/// and persist findings for exposed / unauthenticated control access. The
|
|
/// probe is read-only; it targets the Modbus port of the target's live URL.
|
|
async fn run_ics_probe(
|
|
&self,
|
|
target: &OnboardedTarget,
|
|
target_id: &str,
|
|
scan_run_id: &str,
|
|
) -> Result<u32, AgentError> {
|
|
self.update_phase(scan_run_id, "ics_probe").await;
|
|
let Some(endpoint) = target.live_url().map(|a| a.source_ref.clone()) else {
|
|
tracing::warn!(target_id, "ICS probe: no live URL");
|
|
return Ok(0);
|
|
};
|
|
// Short per-request budget so an unreachable device doesn't stall the scan.
|
|
let budget = std::time::Duration::from_secs(5);
|
|
let findings = werkbank_exec::ics::probe_target(&endpoint, target_id, budget).await;
|
|
tracing::info!(
|
|
target_id,
|
|
endpoint = %endpoint,
|
|
found = findings.len(),
|
|
"ICS probe complete"
|
|
);
|
|
let mut new_count = 0u32;
|
|
for mut finding in findings {
|
|
finding.scan_run_id = Some(scan_run_id.to_string());
|
|
if self
|
|
.db
|
|
.findings()
|
|
.find_one(doc! { "fingerprint": &finding.fingerprint })
|
|
.await?
|
|
.is_none()
|
|
{
|
|
self.db.findings().insert_one(&finding).await?;
|
|
new_count += 1;
|
|
}
|
|
}
|
|
Ok(new_count)
|
|
}
|
|
|
|
/// Store a control-application SBOM (CODESYS libraries + runtime) for a target
|
|
/// and match it against known CVEs. Scoped to `package_manager = "codesys"` so
|
|
/// it refreshes on re-scan and coexists with any firmware/source SBOM. The
|
|
/// runtime `Cmp*` / `3SLicense` components carry real CODESYS advisories, so
|
|
/// this is where PLC-device CVE coverage comes from.
|
|
async fn persist_control_app_sbom(
|
|
&self,
|
|
target_id: &str,
|
|
target_name: &str,
|
|
mut entries: Vec<SbomEntry>,
|
|
) -> Result<(), AgentError> {
|
|
if entries.is_empty() {
|
|
return Ok(());
|
|
}
|
|
self.db
|
|
.sbom_entries()
|
|
.delete_many(doc! { "repo_id": target_id, "package_manager": "codesys" })
|
|
.await?;
|
|
|
|
let cve_scanner = CveScanner::new(
|
|
self.http.clone(),
|
|
self.config.searxng_url.clone(),
|
|
self.config.nvd_api_key.as_ref().map(|k| {
|
|
use secrecy::ExposeSecret;
|
|
k.expose_secret().to_string()
|
|
}),
|
|
);
|
|
let mut alerts = match tokio::time::timeout(
|
|
std::time::Duration::from_secs(600),
|
|
cve_scanner.scan_dependencies(target_id, &mut entries),
|
|
)
|
|
.await
|
|
{
|
|
Ok(Ok(a)) => a,
|
|
Ok(Err(e)) => {
|
|
tracing::warn!(target_id, error = %e, "control-app CVE scan failed");
|
|
Vec::new()
|
|
}
|
|
Err(_) => {
|
|
tracing::warn!(target_id, "control-app CVE scan timed out");
|
|
Vec::new()
|
|
}
|
|
};
|
|
// OSV can't match `pkg:codesys/*` (no such ecosystem); CODESYS advisories
|
|
// live in NVD keyed by CPE + runtime version. Add those (best-effort).
|
|
if let Ok(codesys) = tokio::time::timeout(
|
|
std::time::Duration::from_secs(120),
|
|
cve_scanner.scan_codesys(target_id, &mut entries),
|
|
)
|
|
.await
|
|
{
|
|
alerts.extend(codesys);
|
|
} else {
|
|
tracing::warn!(target_id, "CODESYS CVE match timed out");
|
|
}
|
|
|
|
for entry in &entries {
|
|
let filter = doc! {
|
|
"repo_id": &entry.repo_id,
|
|
"name": &entry.name,
|
|
"version": &entry.version,
|
|
};
|
|
if let Ok(d) = mongodb::bson::to_document(entry) {
|
|
self.db
|
|
.sbom_entries()
|
|
.update_one(filter, doc! { "$set": d })
|
|
.upsert(true)
|
|
.await?;
|
|
}
|
|
}
|
|
let new_notifs = self
|
|
.persist_cve_alerts(target_id, target_name, &alerts)
|
|
.await?;
|
|
tracing::info!(
|
|
target_id,
|
|
components = entries.len(),
|
|
alerts = alerts.len(),
|
|
notifications = new_notifs,
|
|
"control-app SBOM stored"
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
/// Upsert CVE alerts for a target and create dedup'd CVE notifications;
|
|
/// returns the number of newly-created notifications. Shared by the SAST
|
|
/// pipeline and the PLC control-app SBOM path, so every SBOM source (source,
|
|
/// firmware, CODESYS libraries/runtime) raises the same notifications.
|
|
async fn persist_cve_alerts(
|
|
&self,
|
|
repo_id: &str,
|
|
repo_name: &str,
|
|
alerts: &[CveAlert],
|
|
) -> Result<u32, AgentError> {
|
|
use compliance_core::models::notification::{parse_severity, CveNotification};
|
|
|
|
let mut new_notif = 0u32;
|
|
for alert in alerts {
|
|
let filter = doc! { "cve_id": &alert.cve_id, "repo_id": &alert.repo_id };
|
|
let update = mongodb::bson::to_document(alert)
|
|
.map(|d| doc! { "$set": d })
|
|
.unwrap_or_else(|_| doc! {});
|
|
self.db
|
|
.cve_alerts()
|
|
.update_one(filter, update)
|
|
.upsert(true)
|
|
.await?;
|
|
|
|
// Dedup notifications by cve + repo + package + version.
|
|
let notif_filter = doc! {
|
|
"cve_id": &alert.cve_id,
|
|
"repo_id": &alert.repo_id,
|
|
"package_name": &alert.affected_package,
|
|
"package_version": &alert.affected_version,
|
|
};
|
|
let severity = parse_severity(alert.severity.as_deref(), alert.cvss_score);
|
|
let mut notification = CveNotification::new(
|
|
alert.cve_id.clone(),
|
|
repo_id.to_string(),
|
|
repo_name.to_string(),
|
|
alert.affected_package.clone(),
|
|
alert.affected_version.clone(),
|
|
severity,
|
|
);
|
|
notification.cvss_score = alert.cvss_score;
|
|
notification.summary = alert.summary.clone();
|
|
notification.url = Some(format!("https://osv.dev/vulnerability/{}", alert.cve_id));
|
|
let notif_update = doc! {
|
|
"$setOnInsert": mongodb::bson::to_bson(¬ification).unwrap_or_default()
|
|
};
|
|
if let Ok(result) = self
|
|
.db
|
|
.cve_notifications()
|
|
.update_one(notif_filter, notif_update)
|
|
.upsert(true)
|
|
.await
|
|
{
|
|
if result.upserted_id.is_some() {
|
|
new_notif += 1;
|
|
}
|
|
}
|
|
}
|
|
Ok(new_notif)
|
|
}
|
|
|
|
/// Ingest the target's artifacts, classify (tramiton for firmware/RTOS/Yocto,
|
|
/// heuristics otherwise), and store the detected classification on the target.
|
|
/// Best-effort — never fails the scan.
|
|
async fn classify_and_store(
|
|
&self,
|
|
target: &OnboardedTarget,
|
|
target_id: &str,
|
|
scan_run_id: &str,
|
|
) {
|
|
self.update_phase(scan_run_id, "classification").await;
|
|
let ctx = crate::ingest::IngestContext::from_config(&self.config, target_id);
|
|
let ingest_set = match crate::ingest::ingest_all(target, &ctx) {
|
|
Ok(set) => set,
|
|
Err(e) => {
|
|
tracing::warn!(target_id, error = %e, "Unified pipeline: ingest for classification failed");
|
|
return;
|
|
}
|
|
};
|
|
let working_paths = ingest_set.working_paths();
|
|
match crate::classify::classify_target(
|
|
target,
|
|
&working_paths,
|
|
&crate::classify::TramitonNative,
|
|
)
|
|
.await
|
|
{
|
|
Ok(classification) => {
|
|
tracing::info!(
|
|
target_id,
|
|
suggested = %classification.suggested,
|
|
"Unified pipeline: classified target"
|
|
);
|
|
if let (Some(oid), Ok(bson)) = (target.id, mongodb::bson::to_bson(&classification))
|
|
{
|
|
let _ = self
|
|
.db
|
|
.onboarded_targets()
|
|
.update_one(
|
|
doc! { "_id": oid },
|
|
doc! { "$set": { "classification": bson } },
|
|
)
|
|
.await;
|
|
}
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(target_id, error = %e, "Unified pipeline: classification failed")
|
|
}
|
|
}
|
|
|
|
// Analysis-based firmware SBOM: for embedded targets, derive components
|
|
// (resolved libraries + cross-toolchain) from tramiton's build-plan
|
|
// analysis over the already-ingested source — no build, no binary
|
|
// upload. Best-effort; empty when no build plan forms.
|
|
if crate::pipeline::firmware_sbom::is_firmware_target(target.target_type) {
|
|
if let Some(code) = target.code_artifact() {
|
|
if let Some(path) = working_paths.get(&code.id) {
|
|
let entries =
|
|
crate::pipeline::firmware_sbom::firmware_sbom_entries(path, target_id)
|
|
.await;
|
|
if !entries.is_empty() {
|
|
let _ = self
|
|
.db
|
|
.sbom_entries()
|
|
.delete_many(doc! { "repo_id": target_id })
|
|
.await;
|
|
for entry in &entries {
|
|
let filter = doc! {
|
|
"repo_id": &entry.repo_id,
|
|
"name": &entry.name,
|
|
"version": &entry.version,
|
|
};
|
|
if let Ok(d) = mongodb::bson::to_document(entry) {
|
|
let _ = self
|
|
.db
|
|
.sbom_entries()
|
|
.update_one(filter, doc! { "$set": d })
|
|
.upsert(true)
|
|
.await;
|
|
}
|
|
}
|
|
tracing::info!(
|
|
target_id,
|
|
count = entries.len(),
|
|
"Firmware SBOM: stored components from tramiton analysis"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// If the target has a `LiveUrl` artifact and DAST is planned, provision a
|
|
/// `DastTarget` (keyed by `repo_id` = target id) so the existing DAST trigger
|
|
/// fires for wizard-created targets. Idempotent.
|
|
async fn ensure_dast_target(
|
|
&self,
|
|
target: &OnboardedTarget,
|
|
plan: &crate::pipeline::plan::ScanPlan,
|
|
) {
|
|
if !plan.has(ScanType::Dast) {
|
|
return;
|
|
}
|
|
let (Some(url), Some(oid)) = (target.live_url(), target.id) else {
|
|
return;
|
|
};
|
|
let target_id = oid.to_hex();
|
|
if self
|
|
.db
|
|
.dast_targets()
|
|
.find_one(doc! { "repo_id": &target_id })
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.is_some()
|
|
{
|
|
return; // already provisioned
|
|
}
|
|
let kind = url
|
|
.web
|
|
.as_ref()
|
|
.map(|w| w.target_kind.clone())
|
|
.unwrap_or(DastTargetType::WebApp);
|
|
let mut dast = DastTarget::new(target.name.clone(), url.source_ref.clone(), kind);
|
|
dast.repo_id = Some(target_id);
|
|
if let Some(web) = &url.web {
|
|
dast.excluded_paths = web.excluded_paths.clone();
|
|
dast.max_crawl_depth = web.max_crawl_depth;
|
|
dast.rate_limit = web.rate_limit;
|
|
dast.allow_destructive = web.allow_destructive;
|
|
}
|
|
if let Some(auth) = &url.auth {
|
|
dast.auth_config = Some(DastAuthConfig {
|
|
method: auth.method.clone(),
|
|
login_url: auth.login_url.clone(),
|
|
username: auth.username.clone(),
|
|
password: None,
|
|
token: auth.secret.clone(),
|
|
headers: auth.headers.clone(),
|
|
});
|
|
}
|
|
if let Err(e) = self.db.dast_targets().insert_one(&dast).await {
|
|
tracing::warn!(error = %e, "Unified pipeline: failed to provision DAST target");
|
|
}
|
|
}
|
|
|
|
/// Sync the onboarded-target document after a scan: bump `findings_count`
|
|
/// and advance the git artifact's `last_scanned_commit` watermark.
|
|
async fn finalize_target(
|
|
&self,
|
|
target: &OnboardedTarget,
|
|
repo: &RepoView,
|
|
new_count: u32,
|
|
) -> Result<(), AgentError> {
|
|
let oid = match target.id {
|
|
Some(id) => id,
|
|
None => return Ok(()),
|
|
};
|
|
self.db
|
|
.onboarded_targets()
|
|
.update_one(
|
|
doc! { "_id": oid },
|
|
doc! {
|
|
"$inc": { "findings_count": new_count as i64 },
|
|
"$set": { "updated_at": mongodb::bson::DateTime::now() },
|
|
},
|
|
)
|
|
.await?;
|
|
|
|
let repo_path = std::path::Path::new(&self.config.git_clone_base_path).join(&repo.name);
|
|
if let (Ok(sha), Some(code)) = (GitOps::get_head_sha(&repo_path), target.code_artifact()) {
|
|
self.db
|
|
.onboarded_targets()
|
|
.update_one(
|
|
doc! { "_id": oid, "artifacts.id": &code.id },
|
|
doc! { "$set": { "artifacts.$.git.last_scanned_commit": sha } },
|
|
)
|
|
.await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub(super) async fn update_phase(&self, scan_run_id: &str, phase: &str) {
|
|
if let Ok(oid) = mongodb::bson::oid::ObjectId::parse_str(scan_run_id) {
|
|
let _ = self
|
|
.db
|
|
.scan_runs()
|
|
.update_one(
|
|
doc! { "_id": oid },
|
|
doc! {
|
|
"$set": { "current_phase": phase },
|
|
"$push": { "phases_completed": phase },
|
|
},
|
|
)
|
|
.await;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Extract the scheme + host from a git URL.
|
|
/// e.g. "https://gitea.example.com/owner/repo.git" -> "https://gitea.example.com"
|
|
/// e.g. "ssh://git@gitea.example.com:22/owner/repo.git" -> "https://gitea.example.com"
|
|
pub(super) fn extract_base_url(git_url: &str) -> Option<String> {
|
|
if let Some(rest) = git_url.strip_prefix("https://") {
|
|
let host = rest.split('/').next()?;
|
|
Some(format!("https://{host}"))
|
|
} else if let Some(rest) = git_url.strip_prefix("http://") {
|
|
let host = rest.split('/').next()?;
|
|
Some(format!("http://{host}"))
|
|
} else if let Some(rest) = git_url.strip_prefix("ssh://") {
|
|
// ssh://git@host:port/path -> extract host
|
|
let after_at = rest.find('@').map(|i| &rest[i + 1..]).unwrap_or(rest);
|
|
let host = after_at.split(&[':', '/'][..]).next()?;
|
|
Some(format!("https://{host}"))
|
|
} else if let Some(at_pos) = git_url.find('@') {
|
|
// SCP-style: git@host:owner/repo.git
|
|
let after_at = &git_url[at_pos + 1..];
|
|
let host = after_at.split(':').next()?;
|
|
Some(format!("https://{host}"))
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[allow(clippy::expect_used, clippy::unwrap_used)]
|
|
mod tests {
|
|
use super::*;
|
|
use compliance_core::models::{
|
|
Artifact, ArtifactAuth, IssueTrackerConfig, TargetType, TrackerType,
|
|
};
|
|
|
|
#[test]
|
|
fn repo_view_preserves_id_git_auth_and_tracker() {
|
|
let mut target = OnboardedTarget::new("acme".to_string(), TargetType::WebApp);
|
|
target.id = Some(mongodb::bson::oid::ObjectId::new());
|
|
target.findings_count = 3;
|
|
target.scan_config.issue_tracker = Some(IssueTrackerConfig {
|
|
tracker_type: Some(TrackerType::Gitea),
|
|
owner: Some("acme".to_string()),
|
|
repo: Some("web".to_string()),
|
|
token: Some("tt".to_string()),
|
|
});
|
|
|
|
let mut artifact = Artifact::git_repo("https://git/acme.git", "develop");
|
|
if let Some(git) = artifact.git.as_mut() {
|
|
git.last_scanned_commit = Some("abc123".to_string());
|
|
}
|
|
artifact.auth = Some(ArtifactAuth {
|
|
method: "token".to_string(),
|
|
username: Some("bob".to_string()),
|
|
secret: Some("pat".to_string()),
|
|
..Default::default()
|
|
});
|
|
target.artifacts.push(artifact);
|
|
|
|
let code = target.code_artifact().expect("code artifact");
|
|
let repo = RepoView::from_target(&target, code);
|
|
|
|
assert_eq!(repo.id, target.id); // preserved
|
|
assert_eq!(repo.git_url, "https://git/acme.git");
|
|
assert_eq!(repo.default_branch, "develop");
|
|
assert_eq!(repo.last_scanned_commit.as_deref(), Some("abc123"));
|
|
assert_eq!(repo.auth_token.as_deref(), Some("pat"));
|
|
assert_eq!(repo.auth_username.as_deref(), Some("bob"));
|
|
assert_eq!(repo.tracker_type, Some(TrackerType::Gitea));
|
|
assert_eq!(repo.tracker_owner.as_deref(), Some("acme"));
|
|
assert_eq!(repo.findings_count, 3);
|
|
}
|
|
}
|