diff --git a/compliance-agent/src/agent.rs b/compliance-agent/src/agent.rs index 5327fec..4fa59fa 100644 --- a/compliance-agent/src/agent.rs +++ b/compliance-agent/src/agent.rs @@ -63,7 +63,11 @@ impl ComplianceAgent { let db = self.db_pool.for_tenant_id(tenant_id).await?; let orchestrator = PipelineOrchestrator::new(self.config.clone(), db, self.llm.clone(), self.http.clone()); - orchestrator.run(repo_id, trigger).await + if self.config.unified_pipeline { + orchestrator.run_target(repo_id, trigger).await + } else { + orchestrator.run(repo_id, trigger).await + } } /// Run a PR review: scan the diff and post review comments. diff --git a/compliance-agent/src/config.rs b/compliance-agent/src/config.rs index a754975..f05d74a 100644 --- a/compliance-agent/src/config.rs +++ b/compliance-agent/src/config.rs @@ -47,6 +47,9 @@ pub fn load_config() -> Result { .unwrap_or_else(|| "/tmp/compliance-scanner/repos".to_string()), artifact_store_base_path: env_var_opt("ARTIFACT_STORE_BASE_PATH") .unwrap_or_else(|| "/data/compliance-scanner/artifacts".to_string()), + unified_pipeline: env_var_opt("UNIFIED_PIPELINE") + .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) + .unwrap_or(false), ssh_key_path: env_var_opt("SSH_KEY_PATH") .unwrap_or_else(|| "/data/compliance-scanner/ssh/id_ed25519".to_string()), keycloak_url: env_var_opt("KEYCLOAK_URL"), diff --git a/compliance-agent/src/pentest/cleanup.rs b/compliance-agent/src/pentest/cleanup.rs index dce1c61..739b2f9 100644 --- a/compliance-agent/src/pentest/cleanup.rs +++ b/compliance-agent/src/pentest/cleanup.rs @@ -342,6 +342,7 @@ mod tests { pentest_imap_password: None, admin_api_token: None, tenant_registry_url: None, + unified_pipeline: false, } } diff --git a/compliance-agent/src/pipeline/mod.rs b/compliance-agent/src/pipeline/mod.rs index 2a75dfb..70d91b8 100644 --- a/compliance-agent/src/pipeline/mod.rs +++ b/compliance-agent/src/pipeline/mod.rs @@ -8,6 +8,7 @@ mod issue_creation; pub mod lint; pub mod orchestrator; pub mod patterns; +pub mod plan; mod pr_review; pub mod sbom; pub mod semgrep; diff --git a/compliance-agent/src/pipeline/orchestrator.rs b/compliance-agent/src/pipeline/orchestrator.rs index 9b8d8c5..d3c6da0 100644 --- a/compliance-agent/src/pipeline/orchestrator.rs +++ b/compliance-agent/src/pipeline/orchestrator.rs @@ -15,6 +15,7 @@ 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::sbom::SbomScanner; use crate::pipeline::semgrep::SemgrepScanner; @@ -419,6 +420,149 @@ impl PipelineOrchestrator { 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?; + } + 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 { + 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" + ); + + match target.code_artifact() { + Some(code) if code.kind == ArtifactKind::GitRepo => { + let repo = repo_view_from_target(target, code); + let new_count = self.run_pipeline(&repo, scan_run_id).await?; + self.finalize_target(target, &repo, new_count).await?; + Ok(new_count) + } + Some(_) => { + tracing::warn!( + target_id = %target_id, + "Unified pipeline: source-archive scanning not yet wired; skipping" + ); + Ok(0) + } + None => { + // No code to scan. Firmware/PLC/mobile static scanners land in + // #128/#129/#130; DAST for a running URL still works when a + // DastTarget row exists (migrated targets). + tracing::info!( + target_id = %target_id, + "Unified pipeline: no code artifact; attempting DAST only" + ); + self.update_phase(scan_run_id, "dast_scanning").await; + self.maybe_trigger_dast(&target_id, scan_run_id).await; + Ok(0) + } + } + } + + /// 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: &TrackedRepository, + 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 @@ -436,6 +580,37 @@ impl PipelineOrchestrator { } } +/// Build a legacy `TrackedRepository` view from an onboarded target's code +/// artifact, so the unified pipeline can reuse the existing repo pipeline. The +/// inverse of the migration's `repo_to_target`. `_id` is preserved so findings +/// and DAST lookups resolve against the same key. +fn repo_view_from_target(target: &OnboardedTarget, code: &Artifact) -> TrackedRepository { + let mut repo = TrackedRepository::new(target.name.clone(), code.source_ref.clone()); + repo.id = target.id; + if let Some(git) = &code.git { + repo.default_branch = git.default_branch.clone(); + repo.last_scanned_commit = git.last_scanned_commit.clone(); + repo.local_path = git.local_path.clone(); + } + if let Some(auth) = &code.auth { + repo.auth_token = auth.secret.clone(); + repo.auth_username = auth.username.clone(); + } + if let Some(it) = &target.scan_config.issue_tracker { + repo.tracker_type = it.tracker_type.clone(); + repo.tracker_owner = it.owner.clone(); + repo.tracker_repo = it.repo.clone(); + repo.tracker_token = it.token.clone(); + } + repo.scan_schedule = target.scan_schedule.clone(); + repo.webhook_enabled = target.webhook_enabled; + repo.webhook_secret = target.webhook_secret.clone(); + repo.findings_count = target.findings_count; + repo.created_at = target.created_at; + repo.updated_at = target.updated_at; + repo +} + /// 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" @@ -460,3 +635,50 @@ pub(super) fn extract_base_url(git_url: &str) -> Option { 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 = repo_view_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); + } +} diff --git a/compliance-agent/src/pipeline/plan.rs b/compliance-agent/src/pipeline/plan.rs new file mode 100644 index 0000000..6143324 --- /dev/null +++ b/compliance-agent/src/pipeline/plan.rs @@ -0,0 +1,195 @@ +//! The scan plan. +//! +//! [`build_scan_plan`] turns an [`OnboardedTarget`] into the concrete ordered +//! list of scans to run, each bound to the artifact it consumes. It intersects +//! the scan-applicability matrix ([`applicable_scans`]) with the target's +//! `scan_config` overrides: a scan runs when its required artifact is present and +//! it is either on by default or explicitly enabled, and is not explicitly +//! disabled. This is the decision engine the unified pipeline (`run_target`) +//! executes. + +use compliance_core::models::{Artifact, ArtifactKind, OnboardedTarget, ScanPhase, ScanType}; +use compliance_core::scan_matrix::applicable_scans; + +/// One scan to run, bound to the artifact it operates on. +#[derive(Debug, Clone, PartialEq)] +pub struct ScanStep { + /// The scan to run. + pub scan_type: ScanType, + /// The pipeline phase to report while it runs. + pub phase: ScanPhase, + /// The id of the artifact this scan consumes ([`Artifact::id`]). + pub artifact_id: String, +} + +/// The ordered set of scans to run for a target. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct ScanPlan { + /// The scans, in matrix order. + pub steps: Vec, +} + +impl ScanPlan { + /// Whether the plan contains a step for the given scan type. + pub fn has(&self, scan: ScanType) -> bool { + self.steps.iter().any(|s| s.scan_type == scan) + } + + /// Whether the plan is empty (nothing to run). + pub fn is_empty(&self) -> bool { + self.steps.is_empty() + } +} + +/// Build the scan plan for a target: matrix defaults ∩ `scan_config`, each scan +/// bound to the artifact it consumes. Scans whose required artifact is absent, or +/// that are disabled, or off-by-default and not explicitly enabled, are dropped. +pub fn build_scan_plan(target: &OnboardedTarget) -> ScanPlan { + let enabled = &target.scan_config.enabled_scans; + let disabled = &target.scan_config.disabled_scans; + + let mut steps = Vec::new(); + for option in applicable_scans(target) { + // Required artifact missing → not runnable. + if option.blocked_reason.is_some() { + continue; + } + // Explicit opt-out wins. + if disabled.contains(&option.scan) { + continue; + } + // Run if on by default, or explicitly enabled. + if !option.default_on && !enabled.contains(&option.scan) { + continue; + } + let Some(artifact) = resolve_artifact(target, option.required_artifact) else { + continue; + }; + steps.push(ScanStep { + scan_type: option.scan, + phase: phase_for(option.scan), + artifact_id: artifact.id.clone(), + }); + } + ScanPlan { steps } +} + +/// Resolve the artifact a scan consumes. A "code" requirement (represented by +/// `GitRepo`) is satisfied by a git repo *or* a source archive. +fn resolve_artifact(target: &OnboardedTarget, required: Option) -> Option<&Artifact> { + match required { + Some(ArtifactKind::GitRepo) => target.code_artifact(), + Some(kind) => target.first_of(kind), + None => target.code_artifact().or_else(|| target.artifacts.first()), + } +} + +/// The pipeline phase reported while a given scan runs. +fn phase_for(scan: ScanType) -> ScanPhase { + match scan { + ScanType::Sast => ScanPhase::Sast, + ScanType::Sbom => ScanPhase::SbomGeneration, + ScanType::Cve => ScanPhase::CveScanning, + ScanType::Gdpr | ScanType::OAuth => ScanPhase::PatternScanning, + ScanType::SecretDetection => ScanPhase::SecretDetection, + ScanType::Lint => ScanPhase::LintScanning, + ScanType::CodeReview => ScanPhase::CodeReview, + ScanType::Graph => ScanPhase::GraphBuilding, + ScanType::Dast => ScanPhase::DastScanning, + ScanType::FirmwareStatic => ScanPhase::FirmwareStatic, + ScanType::PlcControlLogic => ScanPhase::PlcAnalysis, + ScanType::MobileStatic => ScanPhase::MobileStatic, + ScanType::ContainerScan => ScanPhase::ContainerScan, + } +} + +#[cfg(test)] +#[allow(clippy::expect_used, clippy::unwrap_used)] +mod tests { + use super::*; + use compliance_core::models::{PlcFormat, TargetType}; + + fn target(target_type: TargetType, artifacts: Vec) -> OnboardedTarget { + let mut t = OnboardedTarget::new("t".to_string(), target_type); + t.artifacts = artifacts; + t + } + + fn step_for<'a>(plan: &'a ScanPlan, scan: ScanType) -> Option<&'a ScanStep> { + plan.steps.iter().find(|s| s.scan_type == scan) + } + + #[test] + fn webapp_with_code_and_url_runs_sast_and_dast_bound_to_the_right_artifacts() { + let code = Artifact::git_repo("https://git/x", "main"); + let url = Artifact::live_url("https://x"); + let (code_id, url_id) = (code.id.clone(), url.id.clone()); + let t = target(TargetType::WebApp, vec![code, url]); + + let plan = build_scan_plan(&t); + let sast = step_for(&plan, ScanType::Sast).expect("sast planned"); + assert_eq!(sast.artifact_id, code_id); + assert_eq!(sast.phase, ScanPhase::Sast); + let dast = step_for(&plan, ScanType::Dast).expect("dast planned"); + assert_eq!(dast.artifact_id, url_id); + } + + #[test] + fn webapp_without_url_omits_dast() { + let t = target(TargetType::WebApp, vec![Artifact::git_repo("u", "main")]); + let plan = build_scan_plan(&t); + assert!(plan.has(ScanType::Sast)); + assert!(!plan.has(ScanType::Dast)); + } + + #[test] + fn code_scan_binds_to_source_archive_when_no_git_repo() { + let arc = Artifact::source_archive("src.zip"); + let arc_id = arc.id.clone(); + let t = target(TargetType::BackendService, vec![arc]); + let plan = build_scan_plan(&t); + let sast = step_for(&plan, ScanType::Sast).expect("sast planned"); + assert_eq!(sast.artifact_id, arc_id); + } + + #[test] + fn firmware_sbom_and_cve_bind_to_the_firmware_image() { + let fw = Artifact::firmware_image("fw.bin"); + let fw_id = fw.id.clone(); + let t = target(TargetType::FirmwareBareMetal, vec![fw]); + let plan = build_scan_plan(&t); + let sbom = step_for(&plan, ScanType::Sbom).expect("sbom planned"); + assert_eq!(sbom.artifact_id, fw_id); + assert!(step_for(&plan, ScanType::FirmwareStatic).is_some()); + assert!(!plan.has(ScanType::Dast)); + } + + #[test] + fn plc_plans_only_control_logic() { + let plc = Artifact::plc_project("p.xml", PlcFormat::PlcopenXml); + let t = target(TargetType::PlcSps, vec![plc]); + let plan = build_scan_plan(&t); + assert_eq!(plan.steps.len(), 1); + assert_eq!(plan.steps[0].scan_type, ScanType::PlcControlLogic); + assert_eq!(plan.steps[0].phase, ScanPhase::PlcAnalysis); + } + + #[test] + fn disabled_scan_is_dropped_and_off_by_default_can_be_enabled() { + let mut t = target(TargetType::WebApp, vec![Artifact::git_repo("u", "main")]); + t.scan_config.disabled_scans = vec![ScanType::Lint]; + // CodeReview is off by default for web; enable it explicitly. + t.scan_config.enabled_scans = vec![ScanType::CodeReview]; + let plan = build_scan_plan(&t); + assert!(!plan.has(ScanType::Lint)); + assert!(plan.has(ScanType::CodeReview)); + assert!(plan.has(ScanType::Sast)); + } + + #[test] + fn no_code_artifact_yields_empty_plan_for_web() { + let t = target(TargetType::WebApp, vec![]); + let plan = build_scan_plan(&t); + assert!(plan.is_empty()); + } +} diff --git a/compliance-agent/tests/common/mod.rs b/compliance-agent/tests/common/mod.rs index 9d0934e..e85a475 100644 --- a/compliance-agent/tests/common/mod.rs +++ b/compliance-agent/tests/common/mod.rs @@ -70,6 +70,7 @@ impl TestServer { pentest_imap_password: None, admin_api_token: None, tenant_registry_url: None, + unified_pipeline: false, }; let agent = ComplianceAgent::new(config, db_pool); diff --git a/compliance-core/src/config.rs b/compliance-core/src/config.rs index db88fef..41e406d 100644 --- a/compliance-core/src/config.rs +++ b/compliance-core/src/config.rs @@ -49,6 +49,10 @@ pub struct AgentConfig { /// of tenants to iterate. When `None` or unreachable, scheduler /// falls back to `SCHEDULER_TENANT_IDS` env (M7.2-C). pub tenant_registry_url: Option, + /// When true, `run_scan` dispatches to the unified `run_target` pipeline + /// (reads `onboarded_targets`) instead of the legacy repository pipeline. + /// Env `UNIFIED_PIPELINE`. Defaults off during the transition. + pub unified_pipeline: bool, } #[derive(Clone, Debug, Serialize, Deserialize)]