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/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/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)]