refactor: rip out the legacy TrackedRepository / repositories path (#161)
CI / Check (push) Has been skipped
CI / Detect Changes (push) Successful in 3s
CI / Deploy Agent (push) Successful in 8m20s
CI / Deploy Dashboard (push) Successful in 3m13s
CI / Deploy Docs (push) Has been skipped
CI / Deploy MCP (push) Successful in 2m7s
CI / Check (push) Has been skipped
CI / Detect Changes (push) Successful in 3s
CI / Deploy Agent (push) Successful in 8m20s
CI / Deploy Dashboard (push) Successful in 3m13s
CI / Deploy Docs (push) Has been skipped
CI / Deploy MCP (push) Successful in 2m7s
This commit was merged in pull request #161.
This commit is contained in:
@@ -63,21 +63,12 @@ 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());
|
||||
if self.config.unified_pipeline {
|
||||
orchestrator.run_target(repo_id, trigger).await
|
||||
} else {
|
||||
orchestrator.run(repo_id, trigger).await
|
||||
}
|
||||
orchestrator.run_target(repo_id, trigger).await
|
||||
}
|
||||
|
||||
/// Run a scan for an onboarded target through the unified pipeline,
|
||||
/// unconditionally.
|
||||
///
|
||||
/// Unlike [`Self::run_scan`], this does *not* consult the
|
||||
/// `unified_pipeline` transition flag: the caller (the `/targets/{id}/scan`
|
||||
/// endpoint) operates on `onboarded_targets` by construction, so it must
|
||||
/// always dispatch to `run_target` regardless of how the legacy paths
|
||||
/// (scheduler, webhooks, `/repositories/{id}/scan`) are configured.
|
||||
/// Alias for [`Self::run_scan`] — every scan runs the unified onboarded-target
|
||||
/// pipeline. Kept as a distinct name for the `/targets/{id}/scan` endpoint's
|
||||
/// intent.
|
||||
pub async fn run_target_scan(
|
||||
&self,
|
||||
tenant_id: &str,
|
||||
@@ -100,16 +91,19 @@ impl ComplianceAgent {
|
||||
head_sha: &str,
|
||||
) -> Result<(), crate::error::AgentError> {
|
||||
let db = self.db_pool.for_tenant_id(tenant_id).await?;
|
||||
let repo = db
|
||||
.repositories()
|
||||
.find_one(mongodb::bson::doc! {
|
||||
"_id": mongodb::bson::oid::ObjectId::parse_str(repo_id)
|
||||
.map_err(|e| crate::error::AgentError::Other(e.to_string()))?
|
||||
})
|
||||
let oid = mongodb::bson::oid::ObjectId::parse_str(repo_id)
|
||||
.map_err(|e| crate::error::AgentError::Other(e.to_string()))?;
|
||||
let target = db
|
||||
.onboarded_targets()
|
||||
.find_one(mongodb::bson::doc! { "_id": oid })
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
crate::error::AgentError::Other(format!("Repository {repo_id} not found"))
|
||||
crate::error::AgentError::Other(format!("Target {repo_id} not found"))
|
||||
})?;
|
||||
let code = target.code_artifact().ok_or_else(|| {
|
||||
crate::error::AgentError::Other(format!("Target {repo_id} has no code artifact"))
|
||||
})?;
|
||||
let repo = crate::pipeline::repo_view::RepoView::from_target(&target, code);
|
||||
|
||||
let orchestrator =
|
||||
PipelineOrchestrator::new(self.config.clone(), db, self.llm.clone(), self.http.clone());
|
||||
|
||||
@@ -146,7 +146,7 @@ pub async fn build_embeddings(
|
||||
let agent_clone = (*agent).clone();
|
||||
tokio::spawn(async move {
|
||||
let repo = match db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.find_one(doc! { "_id": mongodb::bson::oid::ObjectId::parse_str(&repo_id).ok() })
|
||||
.await
|
||||
{
|
||||
@@ -194,14 +194,22 @@ pub async fn build_embeddings(
|
||||
}
|
||||
};
|
||||
|
||||
let code = match repo.code_artifact() {
|
||||
Some(c) => c,
|
||||
None => {
|
||||
tracing::error!("Target {repo_id} has no code artifact for embedding build");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let view = crate::pipeline::repo_view::RepoView::from_target(&repo, code);
|
||||
let creds = crate::pipeline::git::RepoCredentials {
|
||||
ssh_key_path: Some(agent_clone.config.ssh_key_path.clone()),
|
||||
auth_token: repo.auth_token.clone(),
|
||||
auth_username: repo.auth_username.clone(),
|
||||
auth_token: view.auth_token.clone(),
|
||||
auth_username: view.auth_username.clone(),
|
||||
};
|
||||
let git_ops =
|
||||
crate::pipeline::git::GitOps::new(&agent_clone.config.git_clone_base_path, creds);
|
||||
let repo_path = match git_ops.clone_or_fetch(&repo.git_url, &repo.name) {
|
||||
let repo_path = match git_ops.clone_or_fetch(&view.git_url, &view.name) {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
tracing::error!("Failed to clone repo for embedding build: {e}");
|
||||
|
||||
@@ -255,7 +255,7 @@ pub async fn get_file_content(
|
||||
|
||||
// Look up the repository to get repo name
|
||||
let repo = db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.find_one(doc! { "_id": mongodb::bson::oid::ObjectId::parse_str(&repo_id).ok() })
|
||||
.await
|
||||
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
|
||||
@@ -317,7 +317,7 @@ pub async fn trigger_build(
|
||||
let agent_clone = (*agent).clone();
|
||||
tokio::spawn(async move {
|
||||
let repo = match db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.find_one(doc! { "_id": mongodb::bson::oid::ObjectId::parse_str(&repo_id).ok() })
|
||||
.await
|
||||
{
|
||||
@@ -328,14 +328,22 @@ pub async fn trigger_build(
|
||||
}
|
||||
};
|
||||
|
||||
let code = match repo.code_artifact() {
|
||||
Some(c) => c,
|
||||
None => {
|
||||
tracing::error!("Target {repo_id} has no code artifact for graph build");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let view = crate::pipeline::repo_view::RepoView::from_target(&repo, code);
|
||||
let creds = crate::pipeline::git::RepoCredentials {
|
||||
ssh_key_path: Some(agent_clone.config.ssh_key_path.clone()),
|
||||
auth_token: repo.auth_token.clone(),
|
||||
auth_username: repo.auth_username.clone(),
|
||||
auth_token: view.auth_token.clone(),
|
||||
auth_username: view.auth_username.clone(),
|
||||
};
|
||||
let git_ops =
|
||||
crate::pipeline::git::GitOps::new(&agent_clone.config.git_clone_base_path, creds);
|
||||
let repo_path = match git_ops.clone_or_fetch(&repo.git_url, &repo.name) {
|
||||
let repo_path = match git_ops.clone_or_fetch(&view.git_url, &view.name) {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
tracing::error!("Failed to clone repo for graph build: {e}");
|
||||
|
||||
@@ -10,6 +10,18 @@ pub async fn health() -> Json<serde_json::Value> {
|
||||
Json(serde_json::json!({ "status": "ok" }))
|
||||
}
|
||||
|
||||
/// GET /api/v1/settings/ssh-public-key — the agent's SSH deploy public key,
|
||||
/// for adding as a read-only deploy key on private git targets.
|
||||
#[tracing::instrument(skip_all)]
|
||||
pub async fn get_ssh_public_key(
|
||||
axum::extract::Extension(agent): AgentExt,
|
||||
) -> Result<Json<serde_json::Value>, axum::http::StatusCode> {
|
||||
let public_path = format!("{}.pub", agent.config.ssh_key_path);
|
||||
let public_key =
|
||||
std::fs::read_to_string(&public_path).map_err(|_| axum::http::StatusCode::NOT_FOUND)?;
|
||||
Ok(Json(serde_json::json!({ "public_key": public_key.trim() })))
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all)]
|
||||
pub async fn stats_overview(
|
||||
axum::extract::Extension(agent): AgentExt,
|
||||
@@ -19,7 +31,7 @@ pub async fn stats_overview(
|
||||
let db = &db;
|
||||
|
||||
let total_repositories = db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.count_documents(doc! {})
|
||||
.await
|
||||
.unwrap_or(0);
|
||||
|
||||
@@ -12,7 +12,6 @@ pub mod notifications;
|
||||
pub mod onboarding;
|
||||
pub mod pentest_handlers;
|
||||
pub use pentest_handlers as pentest;
|
||||
pub mod repos;
|
||||
pub mod sbom;
|
||||
pub mod scans;
|
||||
|
||||
@@ -21,6 +20,5 @@ pub use dto::*;
|
||||
pub use findings::*;
|
||||
pub use health::*;
|
||||
pub use issues::*;
|
||||
pub use repos::*;
|
||||
pub use sbom::*;
|
||||
pub use scans::*;
|
||||
|
||||
@@ -246,15 +246,116 @@ pub async fn delete_target(
|
||||
.delete_one(doc! { "_id": oid })
|
||||
.await
|
||||
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
|
||||
// Cascade the collections keyed by repo_id == target id (best-effort).
|
||||
let by_repo = doc! { "repo_id": &id };
|
||||
let _ = db.findings().delete_many(by_repo.clone()).await;
|
||||
let _ = db.scan_runs().delete_many(by_repo.clone()).await;
|
||||
let _ = db.sbom_entries().delete_many(by_repo.clone()).await;
|
||||
let _ = db.cve_alerts().delete_many(by_repo).await;
|
||||
// Cascade all data keyed by repo_id == target id (best-effort).
|
||||
let db = &db;
|
||||
let _ = db.findings().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db.sbom_entries().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db.scan_runs().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db.cve_alerts().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db
|
||||
.tracker_issues()
|
||||
.delete_many(doc! { "repo_id": &id })
|
||||
.await;
|
||||
let _ = db.graph_nodes().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db.graph_edges().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db.graph_builds().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db
|
||||
.impact_analyses()
|
||||
.delete_many(doc! { "repo_id": &id })
|
||||
.await;
|
||||
let _ = db
|
||||
.code_embeddings()
|
||||
.delete_many(doc! { "repo_id": &id })
|
||||
.await;
|
||||
let _ = db
|
||||
.embedding_builds()
|
||||
.delete_many(doc! { "repo_id": &id })
|
||||
.await;
|
||||
|
||||
// DAST targets linked to this target, and all their downstream data.
|
||||
if let Ok(mut cursor) = db.dast_targets().find(doc! { "repo_id": &id }).await {
|
||||
use futures_util::StreamExt;
|
||||
while let Some(Ok(dt)) = cursor.next().await {
|
||||
let dast_target_id = dt.id.map(|oid| oid.to_hex()).unwrap_or_default();
|
||||
if !dast_target_id.is_empty() {
|
||||
cascade_delete_dast_target(db, &dast_target_id).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Pentest sessions linked directly to this target (not via a DAST target).
|
||||
if let Ok(mut cursor) = db.pentest_sessions().find(doc! { "repo_id": &id }).await {
|
||||
use futures_util::StreamExt;
|
||||
while let Some(Ok(session)) = cursor.next().await {
|
||||
let session_id = session.id.map(|oid| oid.to_hex()).unwrap_or_default();
|
||||
if !session_id.is_empty() {
|
||||
let _ = db
|
||||
.attack_chain_nodes()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
let _ = db
|
||||
.pentest_messages()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
let _ = db
|
||||
.dast_findings()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
let _ = db
|
||||
.pentest_sessions()
|
||||
.delete_many(doc! { "repo_id": &id })
|
||||
.await;
|
||||
|
||||
Ok(Json(serde_json::json!({ "status": "deleted" })))
|
||||
}
|
||||
|
||||
/// Delete a DAST target and everything downstream of it (pentest sessions +
|
||||
/// their attack chains / messages / findings, DAST scan runs + findings).
|
||||
async fn cascade_delete_dast_target(db: &crate::database::Database, target_id: &str) {
|
||||
use futures_util::StreamExt;
|
||||
if let Ok(mut cursor) = db
|
||||
.pentest_sessions()
|
||||
.find(doc! { "target_id": target_id })
|
||||
.await
|
||||
{
|
||||
while let Some(Ok(session)) = cursor.next().await {
|
||||
let session_id = session.id.map(|oid| oid.to_hex()).unwrap_or_default();
|
||||
if !session_id.is_empty() {
|
||||
let _ = db
|
||||
.attack_chain_nodes()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
let _ = db
|
||||
.pentest_messages()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
let _ = db
|
||||
.dast_findings()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
let _ = db
|
||||
.pentest_sessions()
|
||||
.delete_many(doc! { "target_id": target_id })
|
||||
.await;
|
||||
let _ = db
|
||||
.dast_findings()
|
||||
.delete_many(doc! { "target_id": target_id })
|
||||
.await;
|
||||
let _ = db
|
||||
.dast_scan_runs()
|
||||
.delete_many(doc! { "target_id": target_id })
|
||||
.await;
|
||||
if let Ok(oid) = mongodb::bson::oid::ObjectId::parse_str(target_id) {
|
||||
let _ = db.dast_targets().delete_one(doc! { "_id": oid }).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// POST /api/v1/targets/{id}/artifacts — attach an artifact (by reference).
|
||||
#[tracing::instrument(skip_all, fields(target_id = %id))]
|
||||
pub async fn add_artifact(
|
||||
|
||||
@@ -113,14 +113,14 @@ pub async fn create_session(
|
||||
session.config = Some(config.clone());
|
||||
session.repo_id = target.repo_id.clone();
|
||||
|
||||
// Resolve repo_id from git_repo_url if provided
|
||||
// Resolve repo_id (target id) from git_repo_url if provided
|
||||
if let Some(ref git_url) = config.git_repo_url {
|
||||
if let Ok(Some(repo)) = db
|
||||
.repositories()
|
||||
.find_one(doc! { "git_url": git_url })
|
||||
if let Ok(Some(target)) = db
|
||||
.onboarded_targets()
|
||||
.find_one(doc! { "artifacts.source_ref": git_url })
|
||||
.await
|
||||
{
|
||||
session.repo_id = repo.id.map(|oid| oid.to_hex());
|
||||
session.repo_id = target.id.map(|oid| oid.to_hex());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -380,17 +380,20 @@ pub async fn lookup_repo(
|
||||
) -> Result<Json<ApiResponse<serde_json::Value>>, StatusCode> {
|
||||
let db = tenant_db(&agent, &tenant).await?;
|
||||
let repo = db
|
||||
.repositories()
|
||||
.find_one(doc! { "git_url": ¶ms.url })
|
||||
.onboarded_targets()
|
||||
.find_one(doc! { "artifacts.source_ref": ¶ms.url })
|
||||
.await
|
||||
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
|
||||
|
||||
let data = match repo {
|
||||
Some(r) => serde_json::json!({
|
||||
"name": r.name,
|
||||
"default_branch": r.default_branch,
|
||||
"last_scanned_commit": r.last_scanned_commit,
|
||||
}),
|
||||
Some(r) => {
|
||||
let git = r.code_artifact().and_then(|c| c.git.as_ref());
|
||||
serde_json::json!({
|
||||
"name": r.name,
|
||||
"default_branch": git.map(|g| g.default_branch.clone()),
|
||||
"last_scanned_commit": git.and_then(|g| g.last_scanned_commit.clone()),
|
||||
})
|
||||
}
|
||||
None => serde_json::Value::Null,
|
||||
};
|
||||
|
||||
|
||||
@@ -1,339 +0,0 @@
|
||||
use axum::extract::{Extension, Path, Query};
|
||||
use axum::http::StatusCode;
|
||||
use axum::Json;
|
||||
use mongodb::bson::doc;
|
||||
|
||||
use super::dto::*;
|
||||
use compliance_core::models::*;
|
||||
use compliance_core::tenant_ctx::TenantCtx;
|
||||
|
||||
#[tracing::instrument(skip_all)]
|
||||
pub async fn list_repositories(
|
||||
Extension(agent): AgentExt,
|
||||
tenant: TenantCtx,
|
||||
Query(params): Query<PaginationParams>,
|
||||
) -> ApiResult<Vec<TrackedRepository>> {
|
||||
let db = tenant_db(&agent, &tenant).await?;
|
||||
let db = &db;
|
||||
let skip = (params.page.saturating_sub(1)) * params.limit as u64;
|
||||
let total = db
|
||||
.repositories()
|
||||
.count_documents(doc! {})
|
||||
.await
|
||||
.unwrap_or(0);
|
||||
|
||||
let repos = match db
|
||||
.repositories()
|
||||
.find(doc! {})
|
||||
.skip(skip)
|
||||
.limit(params.limit)
|
||||
.await
|
||||
{
|
||||
Ok(cursor) => collect_cursor_async(cursor).await,
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to fetch repositories: {e}");
|
||||
Vec::new()
|
||||
}
|
||||
};
|
||||
|
||||
Ok(Json(ApiResponse {
|
||||
data: repos,
|
||||
total: Some(total),
|
||||
page: Some(params.page),
|
||||
}))
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all)]
|
||||
pub async fn add_repository(
|
||||
Extension(agent): AgentExt,
|
||||
tenant: TenantCtx,
|
||||
Json(req): Json<AddRepositoryRequest>,
|
||||
) -> Result<Json<ApiResponse<TrackedRepository>>, (StatusCode, String)> {
|
||||
// Validate repository access before saving
|
||||
let creds = crate::pipeline::git::RepoCredentials {
|
||||
ssh_key_path: Some(agent.config.ssh_key_path.clone()),
|
||||
auth_token: req.auth_token.clone(),
|
||||
auth_username: req.auth_username.clone(),
|
||||
};
|
||||
|
||||
if let Err(e) = crate::pipeline::git::GitOps::test_access(&req.git_url, &creds) {
|
||||
return Err((
|
||||
StatusCode::BAD_REQUEST,
|
||||
format!("Cannot access repository: {e}"),
|
||||
));
|
||||
}
|
||||
|
||||
let mut repo = TrackedRepository::new(req.name, req.git_url);
|
||||
repo.default_branch = req.default_branch;
|
||||
repo.auth_token = req.auth_token;
|
||||
repo.auth_username = req.auth_username;
|
||||
repo.tracker_type = req.tracker_type;
|
||||
repo.tracker_owner = req.tracker_owner;
|
||||
repo.tracker_repo = req.tracker_repo;
|
||||
repo.tracker_token = req.tracker_token;
|
||||
repo.scan_schedule = req.scan_schedule;
|
||||
|
||||
let db = tenant_db(&agent, &tenant)
|
||||
.await
|
||||
.map_err(|s| (s, "failed to acquire tenant database".to_string()))?;
|
||||
db.repositories().insert_one(&repo).await.map_err(|_| {
|
||||
(
|
||||
StatusCode::CONFLICT,
|
||||
"Repository already exists".to_string(),
|
||||
)
|
||||
})?;
|
||||
|
||||
Ok(Json(ApiResponse {
|
||||
data: repo,
|
||||
total: None,
|
||||
page: None,
|
||||
}))
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all, fields(repo_id = %id))]
|
||||
pub async fn update_repository(
|
||||
Extension(agent): AgentExt,
|
||||
tenant: TenantCtx,
|
||||
Path(id): Path<String>,
|
||||
Json(req): Json<UpdateRepositoryRequest>,
|
||||
) -> Result<Json<serde_json::Value>, StatusCode> {
|
||||
let oid = mongodb::bson::oid::ObjectId::parse_str(&id).map_err(|_| StatusCode::BAD_REQUEST)?;
|
||||
let db = tenant_db(&agent, &tenant).await?;
|
||||
|
||||
let mut set_doc = doc! { "updated_at": mongodb::bson::DateTime::now() };
|
||||
|
||||
if let Some(name) = &req.name {
|
||||
set_doc.insert("name", name);
|
||||
}
|
||||
if let Some(branch) = &req.default_branch {
|
||||
set_doc.insert("default_branch", branch);
|
||||
}
|
||||
if let Some(token) = &req.auth_token {
|
||||
set_doc.insert("auth_token", token);
|
||||
}
|
||||
if let Some(username) = &req.auth_username {
|
||||
set_doc.insert("auth_username", username);
|
||||
}
|
||||
if let Some(tracker_type) = &req.tracker_type {
|
||||
set_doc.insert("tracker_type", tracker_type.to_string());
|
||||
}
|
||||
if let Some(owner) = &req.tracker_owner {
|
||||
set_doc.insert("tracker_owner", owner);
|
||||
}
|
||||
if let Some(repo) = &req.tracker_repo {
|
||||
set_doc.insert("tracker_repo", repo);
|
||||
}
|
||||
if let Some(token) = &req.tracker_token {
|
||||
set_doc.insert("tracker_token", token);
|
||||
}
|
||||
if let Some(schedule) = &req.scan_schedule {
|
||||
set_doc.insert("scan_schedule", schedule);
|
||||
}
|
||||
|
||||
let result = db
|
||||
.repositories()
|
||||
.update_one(doc! { "_id": oid }, doc! { "$set": set_doc })
|
||||
.await
|
||||
.map_err(|e| {
|
||||
tracing::warn!("Failed to update repository: {e}");
|
||||
StatusCode::INTERNAL_SERVER_ERROR
|
||||
})?;
|
||||
|
||||
if result.matched_count == 0 {
|
||||
return Err(StatusCode::NOT_FOUND);
|
||||
}
|
||||
|
||||
Ok(Json(serde_json::json!({ "status": "updated" })))
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all)]
|
||||
pub async fn get_ssh_public_key(
|
||||
Extension(agent): AgentExt,
|
||||
) -> Result<Json<serde_json::Value>, StatusCode> {
|
||||
let public_path = format!("{}.pub", agent.config.ssh_key_path);
|
||||
let public_key = std::fs::read_to_string(&public_path).map_err(|_| StatusCode::NOT_FOUND)?;
|
||||
Ok(Json(serde_json::json!({ "public_key": public_key.trim() })))
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all, fields(repo_id = %id))]
|
||||
pub async fn trigger_scan(
|
||||
Extension(agent): AgentExt,
|
||||
tenant: TenantCtx,
|
||||
Path(id): Path<String>,
|
||||
) -> Result<Json<serde_json::Value>, StatusCode> {
|
||||
let agent_clone = (*agent).clone();
|
||||
let tenant_id = tenant.0.tenant_id.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = agent_clone
|
||||
.run_scan(&tenant_id, &id, ScanTrigger::Manual)
|
||||
.await
|
||||
{
|
||||
tracing::error!("Manual scan failed for {id}: {e}");
|
||||
}
|
||||
});
|
||||
|
||||
Ok(Json(serde_json::json!({ "status": "scan_triggered" })))
|
||||
}
|
||||
|
||||
/// Return the webhook secret for a repository (used by dashboard to display it)
|
||||
pub async fn get_webhook_config(
|
||||
Extension(agent): AgentExt,
|
||||
tenant: TenantCtx,
|
||||
Path(id): Path<String>,
|
||||
) -> Result<Json<serde_json::Value>, StatusCode> {
|
||||
let oid = mongodb::bson::oid::ObjectId::parse_str(&id).map_err(|_| StatusCode::BAD_REQUEST)?;
|
||||
let db = tenant_db(&agent, &tenant).await?;
|
||||
let repo = db
|
||||
.repositories()
|
||||
.find_one(doc! { "_id": oid })
|
||||
.await
|
||||
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
|
||||
.ok_or(StatusCode::NOT_FOUND)?;
|
||||
|
||||
let tracker_type = repo
|
||||
.tracker_type
|
||||
.as_ref()
|
||||
.map(|t| t.to_string())
|
||||
.unwrap_or_else(|| "gitea".to_string());
|
||||
|
||||
Ok(Json(serde_json::json!({
|
||||
"webhook_secret": repo.webhook_secret,
|
||||
"tracker_type": tracker_type,
|
||||
})))
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all, fields(repo_id = %id))]
|
||||
pub async fn delete_repository(
|
||||
Extension(agent): AgentExt,
|
||||
tenant: TenantCtx,
|
||||
Path(id): Path<String>,
|
||||
) -> Result<Json<serde_json::Value>, StatusCode> {
|
||||
let oid = mongodb::bson::oid::ObjectId::parse_str(&id).map_err(|_| StatusCode::BAD_REQUEST)?;
|
||||
let db = tenant_db(&agent, &tenant).await?;
|
||||
let db = &db;
|
||||
|
||||
// Delete the repository
|
||||
let result = db
|
||||
.repositories()
|
||||
.delete_one(doc! { "_id": oid })
|
||||
.await
|
||||
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
|
||||
|
||||
if result.deleted_count == 0 {
|
||||
return Err(StatusCode::NOT_FOUND);
|
||||
}
|
||||
|
||||
// Cascade delete all related data
|
||||
let _ = db.findings().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db.sbom_entries().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db.scan_runs().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db.cve_alerts().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db
|
||||
.tracker_issues()
|
||||
.delete_many(doc! { "repo_id": &id })
|
||||
.await;
|
||||
let _ = db.graph_nodes().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db.graph_edges().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db.graph_builds().delete_many(doc! { "repo_id": &id }).await;
|
||||
let _ = db
|
||||
.impact_analyses()
|
||||
.delete_many(doc! { "repo_id": &id })
|
||||
.await;
|
||||
let _ = db
|
||||
.code_embeddings()
|
||||
.delete_many(doc! { "repo_id": &id })
|
||||
.await;
|
||||
let _ = db
|
||||
.embedding_builds()
|
||||
.delete_many(doc! { "repo_id": &id })
|
||||
.await;
|
||||
|
||||
// Cascade delete DAST targets linked to this repo, and all their downstream data
|
||||
// (scan runs, findings, pentest sessions, attack chains, messages)
|
||||
if let Ok(mut cursor) = db.dast_targets().find(doc! { "repo_id": &id }).await {
|
||||
use futures_util::StreamExt;
|
||||
while let Some(Ok(target)) = cursor.next().await {
|
||||
let target_id = target.id.map(|oid| oid.to_hex()).unwrap_or_default();
|
||||
if !target_id.is_empty() {
|
||||
cascade_delete_dast_target(db, &target_id).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Also delete pentest sessions linked directly to this repo (not via target)
|
||||
if let Ok(mut cursor) = db.pentest_sessions().find(doc! { "repo_id": &id }).await {
|
||||
use futures_util::StreamExt;
|
||||
while let Some(Ok(session)) = cursor.next().await {
|
||||
let session_id = session.id.map(|oid| oid.to_hex()).unwrap_or_default();
|
||||
if !session_id.is_empty() {
|
||||
let _ = db
|
||||
.attack_chain_nodes()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
let _ = db
|
||||
.pentest_messages()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
// Delete DAST findings produced by this session
|
||||
let _ = db
|
||||
.dast_findings()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
let _ = db
|
||||
.pentest_sessions()
|
||||
.delete_many(doc! { "repo_id": &id })
|
||||
.await;
|
||||
|
||||
Ok(Json(serde_json::json!({ "status": "deleted" })))
|
||||
}
|
||||
|
||||
/// Cascade-delete a DAST target and all its downstream data.
|
||||
async fn cascade_delete_dast_target(db: &crate::database::Database, target_id: &str) {
|
||||
// Delete pentest sessions for this target (and their attack chains + messages)
|
||||
if let Ok(mut cursor) = db
|
||||
.pentest_sessions()
|
||||
.find(doc! { "target_id": target_id })
|
||||
.await
|
||||
{
|
||||
use futures_util::StreamExt;
|
||||
while let Some(Ok(session)) = cursor.next().await {
|
||||
let session_id = session.id.map(|oid| oid.to_hex()).unwrap_or_default();
|
||||
if !session_id.is_empty() {
|
||||
let _ = db
|
||||
.attack_chain_nodes()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
let _ = db
|
||||
.pentest_messages()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
let _ = db
|
||||
.dast_findings()
|
||||
.delete_many(doc! { "session_id": &session_id })
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
let _ = db
|
||||
.pentest_sessions()
|
||||
.delete_many(doc! { "target_id": target_id })
|
||||
.await;
|
||||
|
||||
// Delete DAST scan runs and their findings
|
||||
let _ = db
|
||||
.dast_findings()
|
||||
.delete_many(doc! { "target_id": target_id })
|
||||
.await;
|
||||
let _ = db
|
||||
.dast_scan_runs()
|
||||
.delete_many(doc! { "target_id": target_id })
|
||||
.await;
|
||||
|
||||
// Delete the target itself
|
||||
if let Ok(oid) = mongodb::bson::oid::ObjectId::parse_str(target_id) {
|
||||
let _ = db.dast_targets().delete_one(doc! { "_id": oid }).await;
|
||||
}
|
||||
}
|
||||
@@ -11,20 +11,6 @@ pub fn build_router() -> Router {
|
||||
"/api/v1/settings/ssh-public-key",
|
||||
get(handlers::get_ssh_public_key),
|
||||
)
|
||||
.route("/api/v1/repositories", get(handlers::list_repositories))
|
||||
.route("/api/v1/repositories", post(handlers::add_repository))
|
||||
.route(
|
||||
"/api/v1/repositories/{id}/scan",
|
||||
post(handlers::trigger_scan),
|
||||
)
|
||||
.route(
|
||||
"/api/v1/repositories/{id}",
|
||||
delete(handlers::delete_repository).patch(handlers::update_repository),
|
||||
)
|
||||
.route(
|
||||
"/api/v1/repositories/{id}/webhook-config",
|
||||
get(handlers::get_webhook_config),
|
||||
)
|
||||
// Unified onboarding targets (#131).
|
||||
.route(
|
||||
"/api/v1/targets",
|
||||
|
||||
@@ -47,12 +47,6 @@ pub fn load_config() -> Result<AgentConfig, AgentError> {
|
||||
.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()),
|
||||
// Defaults ON: the unified onboarded-target pipeline is now the primary
|
||||
// path (no legacy `repositories` data in production). Set
|
||||
// `UNIFIED_PIPELINE=0` to fall back to the legacy repository pipeline.
|
||||
unified_pipeline: env_var_opt("UNIFIED_PIPELINE")
|
||||
.map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
|
||||
.unwrap_or(true),
|
||||
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"),
|
||||
|
||||
@@ -249,16 +249,6 @@ impl Database {
|
||||
}
|
||||
|
||||
pub async fn ensure_indexes(&self) -> Result<(), AgentError> {
|
||||
// repositories: unique git_url
|
||||
self.repositories()
|
||||
.create_index(
|
||||
IndexModel::builder()
|
||||
.keys(doc! { "git_url": 1 })
|
||||
.options(IndexOptions::builder().unique(true).build())
|
||||
.build(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
// findings: unique fingerprint
|
||||
self.findings()
|
||||
.create_index(
|
||||
@@ -479,10 +469,6 @@ impl Database {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn repositories(&self) -> Collection<TrackedRepository> {
|
||||
self.inner.collection("repositories")
|
||||
}
|
||||
|
||||
pub fn findings(&self) -> Collection<Finding> {
|
||||
self.inner.collection("findings")
|
||||
}
|
||||
|
||||
@@ -8,7 +8,6 @@ pub mod database;
|
||||
pub mod error;
|
||||
pub mod ingest;
|
||||
pub mod llm;
|
||||
pub mod migrate;
|
||||
pub mod pentest;
|
||||
pub mod pipeline;
|
||||
pub mod rag;
|
||||
|
||||
@@ -1,50 +1,4 @@
|
||||
use compliance_agent::{agent, api, config, database, migrate, scheduler, ssh, webhooks};
|
||||
|
||||
/// Run the `migrate onboarding` subcommand and exit. Backfills (or reverts) the
|
||||
/// unified `onboarded_targets` collection per tenant.
|
||||
///
|
||||
/// Usage: `compliance-agent migrate onboarding [--all | --tenant <id>] [--dry-run] [--revert]`
|
||||
async fn run_migration(
|
||||
args: &[String],
|
||||
pool: &database::DatabasePool,
|
||||
) -> Result<(), compliance_agent::error::AgentError> {
|
||||
if args.get(2).map(String::as_str) != Some("onboarding") {
|
||||
eprintln!(
|
||||
"usage: compliance-agent migrate onboarding [--all | --tenant <id>] [--dry-run] [--revert]"
|
||||
);
|
||||
std::process::exit(2);
|
||||
}
|
||||
let has = |flag: &str| args.iter().any(|a| a == flag);
|
||||
let dry_run = has("--dry-run");
|
||||
let revert = has("--revert");
|
||||
let tenant = args
|
||||
.iter()
|
||||
.position(|a| a == "--tenant")
|
||||
.and_then(|i| args.get(i + 1))
|
||||
.cloned();
|
||||
|
||||
let tenants: Vec<String> = if has("--all") {
|
||||
pool.list_tenant_ids().await?
|
||||
} else if let Some(t) = tenant {
|
||||
vec![t]
|
||||
} else {
|
||||
eprintln!("specify --all or --tenant <id>");
|
||||
std::process::exit(2);
|
||||
};
|
||||
|
||||
for tenant_id in tenants {
|
||||
let db = pool.for_tenant_id(&tenant_id).await?;
|
||||
if revert {
|
||||
migrate::onboarding::revert(&db).await?;
|
||||
println!("[{tenant_id}] reverted onboarding backfill");
|
||||
} else {
|
||||
let report = migrate::onboarding::backfill_onboarded_targets(&db, dry_run).await?;
|
||||
let prefix = if dry_run { "(dry-run) " } else { "" };
|
||||
println!("[{tenant_id}] {prefix}{report:?}");
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
use compliance_agent::{agent, api, config, database, scheduler, ssh, webhooks};
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
@@ -77,13 +31,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let db_pool =
|
||||
database::DatabasePool::connect(&config.mongodb_uri, &config.mongodb_database).await?;
|
||||
|
||||
// One-shot subcommands run and exit without starting the servers.
|
||||
let args: Vec<String> = std::env::args().collect();
|
||||
if args.get(1).map(String::as_str) == Some("migrate") {
|
||||
run_migration(&args, &db_pool).await?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let agent = agent::ComplianceAgent::new(config.clone(), db_pool);
|
||||
|
||||
tracing::info!("Starting scheduler...");
|
||||
|
||||
@@ -1,8 +0,0 @@
|
||||
//! One-time data migrations.
|
||||
//!
|
||||
//! Currently just the onboarding backfill ([`onboarding`]), which folds the
|
||||
//! legacy `repositories` and `dast_targets` collections into the unified
|
||||
//! `onboarded_targets` collection, preserving `_id` so every downstream record
|
||||
//! keyed by `repo_id` / `target_id` keeps resolving.
|
||||
|
||||
pub mod onboarding;
|
||||
@@ -1,406 +0,0 @@
|
||||
//! Backfill: legacy `repositories` + `dast_targets` → `onboarded_targets`.
|
||||
//!
|
||||
//! The transforms here are **id-preserving**: an [`OnboardedTarget`] keeps the
|
||||
//! same `_id` as the `TrackedRepository` / `DastTarget` it came from, so every
|
||||
//! downstream collection keyed by that hex id (findings, sbom, scan_runs,
|
||||
//! graph, dast_*, pentest_*) keeps resolving with zero row rewrites, and
|
||||
//! existing webhook URLs keep working. The mapping functions are pure and unit
|
||||
//! tested; the DB orchestration (idempotent per-tenant backfill + revert) is a
|
||||
//! thin driver over them.
|
||||
|
||||
use compliance_core::models::{
|
||||
Artifact, ArtifactKind, DastTarget, DastTargetType, GitArtifactConfig, IssueTrackerConfig,
|
||||
OnboardedTarget, TargetType, TrackedRepository, WebArtifactConfig,
|
||||
};
|
||||
use futures_util::TryStreamExt;
|
||||
use mongodb::bson::{doc, Document};
|
||||
|
||||
use crate::database::Database;
|
||||
use crate::error::AgentError;
|
||||
|
||||
/// Marker id in `schema_migrations` recording that the backfill has run.
|
||||
const MIGRATION_MARKER: &str = "onboarding_backfill_v1";
|
||||
|
||||
/// Summary of a backfill run.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
pub struct MigrationReport {
|
||||
/// Repositories turned into onboarded targets.
|
||||
pub repos_migrated: u64,
|
||||
/// DAST targets folded into an existing (repo-linked) target as a LiveUrl.
|
||||
pub dast_targets_folded: u64,
|
||||
/// DAST targets with no repo link, migrated as standalone targets.
|
||||
pub dast_targets_standalone: u64,
|
||||
/// Records skipped because a target with that `_id` already existed.
|
||||
pub skipped_existing: u64,
|
||||
}
|
||||
|
||||
/// Map a legacy `DastTargetType` to a unified [`TargetType`]. REST/GraphQL APIs
|
||||
/// are backend services; a browser app is a web app.
|
||||
fn target_type_for_dast(kind: &DastTargetType) -> TargetType {
|
||||
match kind {
|
||||
DastTargetType::WebApp => TargetType::WebApp,
|
||||
DastTargetType::RestApi | DastTargetType::GraphQl => TargetType::BackendService,
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the LiveUrl artifact for a DAST target (its base URL + crawl config +
|
||||
/// auth). Shared by fold-in and standalone migration.
|
||||
pub fn dast_to_artifact(dast: &DastTarget) -> Artifact {
|
||||
let mut artifact = Artifact::live_url(dast.base_url.clone());
|
||||
artifact.web = Some(WebArtifactConfig {
|
||||
target_kind: dast.target_type.clone(),
|
||||
excluded_paths: dast.excluded_paths.clone(),
|
||||
max_crawl_depth: dast.max_crawl_depth,
|
||||
rate_limit: dast.rate_limit,
|
||||
allow_destructive: dast.allow_destructive,
|
||||
});
|
||||
artifact.auth = dast.auth_config.clone().map(Into::into);
|
||||
artifact
|
||||
}
|
||||
|
||||
/// Map a `TrackedRepository` to an onboarded target, preserving `_id`. The git
|
||||
/// remote becomes a `GitRepo` artifact carrying the repo's branch, watermark,
|
||||
/// and auth; tracker config folds into `scan_config`.
|
||||
///
|
||||
/// `target_type` is a safe default (`BackendService`) — the classifier can
|
||||
/// refine it later; `classification` is left `None` (unconfirmed).
|
||||
pub fn repo_to_target(repo: &TrackedRepository) -> OnboardedTarget {
|
||||
let mut target = OnboardedTarget::new(repo.name.clone(), TargetType::BackendService);
|
||||
target.id = repo.id;
|
||||
|
||||
let mut artifact = Artifact::git_repo(repo.git_url.clone(), repo.default_branch.clone());
|
||||
artifact.git = Some(GitArtifactConfig {
|
||||
default_branch: repo.default_branch.clone(),
|
||||
last_scanned_commit: repo.last_scanned_commit.clone(),
|
||||
local_path: repo.local_path.clone(),
|
||||
});
|
||||
if repo.auth_token.is_some() || repo.auth_username.is_some() {
|
||||
artifact.auth = Some(compliance_core::models::ArtifactAuth {
|
||||
method: "token".to_string(),
|
||||
username: repo.auth_username.clone(),
|
||||
secret: repo.auth_token.clone(),
|
||||
..Default::default()
|
||||
});
|
||||
}
|
||||
target.artifacts.push(artifact);
|
||||
|
||||
if repo.tracker_type.is_some() {
|
||||
target.scan_config.issue_tracker = Some(IssueTrackerConfig {
|
||||
tracker_type: repo.tracker_type.clone(),
|
||||
owner: repo.tracker_owner.clone(),
|
||||
repo: repo.tracker_repo.clone(),
|
||||
token: repo.tracker_token.clone(),
|
||||
});
|
||||
}
|
||||
|
||||
target.scan_schedule = repo.scan_schedule.clone();
|
||||
target.webhook_enabled = repo.webhook_enabled;
|
||||
target.webhook_secret = repo.webhook_secret.clone();
|
||||
target.findings_count = repo.findings_count;
|
||||
target.created_at = repo.created_at;
|
||||
target.updated_at = repo.updated_at;
|
||||
target
|
||||
}
|
||||
|
||||
/// Append a DAST target's LiveUrl artifact onto an existing (repo-derived)
|
||||
/// target. If the repo default was `BackendService` but the DAST target is a
|
||||
/// browser web app, promote the type to `WebApp`.
|
||||
pub fn fold_dast_into_target(target: &mut OnboardedTarget, dast: &DastTarget) {
|
||||
if matches!(dast.target_type, DastTargetType::WebApp)
|
||||
&& target.target_type == TargetType::BackendService
|
||||
{
|
||||
target.target_type = TargetType::WebApp;
|
||||
}
|
||||
if !target.has(ArtifactKind::LiveUrl) {
|
||||
target.artifacts.push(dast_to_artifact(dast));
|
||||
}
|
||||
}
|
||||
|
||||
/// Map a repo-less DAST target to a standalone onboarded target, preserving `_id`.
|
||||
pub fn dast_to_standalone_target(dast: &DastTarget) -> OnboardedTarget {
|
||||
let mut target =
|
||||
OnboardedTarget::new(dast.name.clone(), target_type_for_dast(&dast.target_type));
|
||||
target.id = dast.id;
|
||||
target.artifacts.push(dast_to_artifact(dast));
|
||||
target.created_at = dast.created_at;
|
||||
target.updated_at = dast.updated_at;
|
||||
target
|
||||
}
|
||||
|
||||
/// Whether the onboarding backfill has already been applied to this database.
|
||||
pub async fn already_applied(db: &Database) -> Result<bool, AgentError> {
|
||||
let found = db
|
||||
.collection_named::<Document>("schema_migrations")
|
||||
.find_one(doc! { "_id": MIGRATION_MARKER })
|
||||
.await?;
|
||||
Ok(found.is_some())
|
||||
}
|
||||
|
||||
/// Backfill `onboarded_targets` from `repositories` + `dast_targets` for one
|
||||
/// tenant database.
|
||||
///
|
||||
/// Id-preserving and **idempotent**: targets that already exist (by `_id`) are
|
||||
/// skipped, so re-running is safe. With `dry_run`, computes the report without
|
||||
/// writing. The legacy collections are never deleted; the only mutation outside
|
||||
/// `onboarded_targets` is the history relink of folded DAST targets, which is
|
||||
/// logged so [`revert`] can undo it.
|
||||
pub async fn backfill_onboarded_targets(
|
||||
db: &Database,
|
||||
dry_run: bool,
|
||||
) -> Result<MigrationReport, AgentError> {
|
||||
let mut report = MigrationReport::default();
|
||||
|
||||
// 1. repositories -> onboarded_targets (preserve _id, skip existing).
|
||||
let mut repos = db.repositories().find(doc! {}).await?;
|
||||
while let Some(repo) = repos.try_next().await? {
|
||||
let Some(id) = repo.id else { continue };
|
||||
if db
|
||||
.onboarded_targets()
|
||||
.find_one(doc! { "_id": id })
|
||||
.await?
|
||||
.is_some()
|
||||
{
|
||||
report.skipped_existing += 1;
|
||||
continue;
|
||||
}
|
||||
if !dry_run {
|
||||
db.onboarded_targets()
|
||||
.insert_one(repo_to_target(&repo))
|
||||
.await?;
|
||||
}
|
||||
report.repos_migrated += 1;
|
||||
}
|
||||
|
||||
// 2. dast_targets -> fold into the linked repo target, or migrate standalone.
|
||||
let mut dasts = db.dast_targets().find(doc! {}).await?;
|
||||
while let Some(dast) = dasts.try_next().await? {
|
||||
let Some(dast_id) = dast.id else { continue };
|
||||
let repo_oid = dast
|
||||
.repo_id
|
||||
.as_deref()
|
||||
.and_then(|r| mongodb::bson::oid::ObjectId::parse_str(r).ok());
|
||||
let linked = match repo_oid {
|
||||
Some(oid) => db.onboarded_targets().find_one(doc! { "_id": oid }).await?,
|
||||
None => None,
|
||||
};
|
||||
|
||||
match (linked, repo_oid) {
|
||||
// Fold into an existing repo-derived target.
|
||||
(Some(mut target), Some(oid)) => {
|
||||
if target.has(ArtifactKind::LiveUrl) {
|
||||
report.skipped_existing += 1; // already folded on a prior run
|
||||
continue;
|
||||
}
|
||||
fold_dast_into_target(&mut target, &dast);
|
||||
if !dry_run {
|
||||
db.onboarded_targets()
|
||||
.replace_one(doc! { "_id": oid }, &target)
|
||||
.await?;
|
||||
relink_history(db, &dast_id.to_hex(), &oid.to_hex()).await?;
|
||||
}
|
||||
report.dast_targets_folded += 1;
|
||||
}
|
||||
// No linked repo target: migrate as a standalone target (keeps _id).
|
||||
_ => {
|
||||
if db
|
||||
.onboarded_targets()
|
||||
.find_one(doc! { "_id": dast_id })
|
||||
.await?
|
||||
.is_some()
|
||||
{
|
||||
report.skipped_existing += 1;
|
||||
continue;
|
||||
}
|
||||
if !dry_run {
|
||||
db.onboarded_targets()
|
||||
.insert_one(dast_to_standalone_target(&dast))
|
||||
.await?;
|
||||
}
|
||||
report.dast_targets_standalone += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !dry_run {
|
||||
db.collection_named::<Document>("schema_migrations")
|
||||
.update_one(
|
||||
doc! { "_id": MIGRATION_MARKER },
|
||||
doc! { "$set": { "applied_at": mongodb::bson::DateTime::now() } },
|
||||
)
|
||||
.upsert(true)
|
||||
.await?;
|
||||
}
|
||||
Ok(report)
|
||||
}
|
||||
|
||||
/// Relink DAST scan runs and pentest sessions from the old DAST target id to the
|
||||
/// unified target id, logging each move so [`revert`] can undo it.
|
||||
///
|
||||
/// Note: if multiple DAST targets fold into the same repo target, revert
|
||||
/// restores only the last-logged mapping — a rare edge. The source collections
|
||||
/// (`repositories`, `dast_targets`) are never deleted, so no data is lost.
|
||||
async fn relink_history(db: &Database, old_id: &str, new_id: &str) -> Result<(), AgentError> {
|
||||
db.dast_scan_runs()
|
||||
.update_many(
|
||||
doc! { "target_id": old_id },
|
||||
doc! { "$set": { "target_id": new_id } },
|
||||
)
|
||||
.await?;
|
||||
db.pentest_sessions()
|
||||
.update_many(
|
||||
doc! { "target_id": old_id },
|
||||
doc! { "$set": { "target_id": new_id } },
|
||||
)
|
||||
.await?;
|
||||
db.collection_named::<Document>("onboarding_migration_log")
|
||||
.insert_one(doc! { "old_target_id": old_id, "new_target_id": new_id })
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Undo the backfill: replay the relink log in reverse, drop `onboarded_targets`
|
||||
/// and the log, and clear the marker. The legacy collections are untouched, so
|
||||
/// this restores the pre-migration state.
|
||||
pub async fn revert(db: &Database) -> Result<(), AgentError> {
|
||||
let log = db.collection_named::<Document>("onboarding_migration_log");
|
||||
let mut cursor = log.find(doc! {}).await?;
|
||||
while let Some(entry) = cursor.try_next().await? {
|
||||
if let (Ok(old), Ok(new)) = (
|
||||
entry.get_str("old_target_id"),
|
||||
entry.get_str("new_target_id"),
|
||||
) {
|
||||
db.dast_scan_runs()
|
||||
.update_many(
|
||||
doc! { "target_id": new },
|
||||
doc! { "$set": { "target_id": old } },
|
||||
)
|
||||
.await?;
|
||||
db.pentest_sessions()
|
||||
.update_many(
|
||||
doc! { "target_id": new },
|
||||
doc! { "$set": { "target_id": old } },
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
db.onboarded_targets().drop().await?;
|
||||
log.drop().await?;
|
||||
db.collection_named::<Document>("schema_migrations")
|
||||
.delete_one(doc! { "_id": MIGRATION_MARKER })
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[allow(clippy::expect_used, clippy::unwrap_used)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use compliance_core::models::{DastAuthConfig, TrackerType};
|
||||
|
||||
fn repo() -> TrackedRepository {
|
||||
let mut r = TrackedRepository::new("acme".to_string(), "https://git/acme.git".to_string());
|
||||
r.id = Some(mongodb::bson::oid::ObjectId::new());
|
||||
r.default_branch = "develop".to_string();
|
||||
r.last_scanned_commit = Some("abc123".to_string());
|
||||
r.auth_token = Some("pat".to_string());
|
||||
r.auth_username = Some("bob".to_string());
|
||||
r.tracker_type = Some(TrackerType::Gitea);
|
||||
r.tracker_owner = Some("acme".to_string());
|
||||
r.findings_count = 7;
|
||||
r
|
||||
}
|
||||
|
||||
fn dast(repo_id: Option<String>, kind: DastTargetType) -> DastTarget {
|
||||
let mut d = DastTarget::new(
|
||||
"acme-web".to_string(),
|
||||
"https://acme.example.com".to_string(),
|
||||
kind,
|
||||
);
|
||||
d.id = Some(mongodb::bson::oid::ObjectId::new());
|
||||
d.repo_id = repo_id;
|
||||
d.max_crawl_depth = 5;
|
||||
d.auth_config = Some(DastAuthConfig {
|
||||
method: "bearer".to_string(),
|
||||
login_url: None,
|
||||
username: None,
|
||||
password: None,
|
||||
token: Some("tok".to_string()),
|
||||
headers: None,
|
||||
});
|
||||
d
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn repo_maps_preserving_id_and_git_artifact() {
|
||||
let r = repo();
|
||||
let t = repo_to_target(&r);
|
||||
assert_eq!(t.id, r.id); // id preserved
|
||||
assert_eq!(t.findings_count, 7);
|
||||
assert_eq!(t.scan_schedule, r.scan_schedule);
|
||||
let git = t.code_artifact().expect("git artifact");
|
||||
assert_eq!(git.kind, ArtifactKind::GitRepo);
|
||||
assert_eq!(git.source_ref, "https://git/acme.git");
|
||||
let gc = git.git.as_ref().expect("git config");
|
||||
assert_eq!(gc.default_branch, "develop");
|
||||
assert_eq!(gc.last_scanned_commit.as_deref(), Some("abc123"));
|
||||
let auth = git.auth.as_ref().expect("auth");
|
||||
assert_eq!(auth.secret.as_deref(), Some("pat"));
|
||||
assert_eq!(auth.username.as_deref(), Some("bob"));
|
||||
assert_eq!(
|
||||
t.scan_config
|
||||
.issue_tracker
|
||||
.as_ref()
|
||||
.and_then(|it| it.tracker_type.clone()),
|
||||
Some(TrackerType::Gitea)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn standalone_dast_maps_preserving_id_and_live_url() {
|
||||
let d = dast(None, DastTargetType::WebApp);
|
||||
let t = dast_to_standalone_target(&d);
|
||||
assert_eq!(t.id, d.id);
|
||||
assert_eq!(t.target_type, TargetType::WebApp);
|
||||
let url = t.live_url().expect("live url");
|
||||
assert_eq!(url.source_ref, "https://acme.example.com");
|
||||
let web = url.web.as_ref().expect("web config");
|
||||
assert_eq!(web.max_crawl_depth, 5);
|
||||
assert_eq!(
|
||||
url.auth.as_ref().and_then(|a| a.secret.clone()),
|
||||
Some("tok".to_string())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rest_api_dast_maps_to_backend_service() {
|
||||
let d = dast(None, DastTargetType::RestApi);
|
||||
assert_eq!(
|
||||
dast_to_standalone_target(&d).target_type,
|
||||
TargetType::BackendService
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fold_adds_live_url_and_promotes_webapp() {
|
||||
let mut t = repo_to_target(&repo());
|
||||
assert_eq!(t.target_type, TargetType::BackendService);
|
||||
fold_dast_into_target(&mut t, &dast(Some("x".to_string()), DastTargetType::WebApp));
|
||||
assert_eq!(t.target_type, TargetType::WebApp); // promoted
|
||||
assert!(t.has(ArtifactKind::LiveUrl));
|
||||
assert!(t.has(ArtifactKind::GitRepo));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fold_is_idempotent_on_live_url() {
|
||||
let mut t = repo_to_target(&repo());
|
||||
let d = dast(Some("x".to_string()), DastTargetType::WebApp);
|
||||
fold_dast_into_target(&mut t, &d);
|
||||
fold_dast_into_target(&mut t, &d);
|
||||
let live_urls = t
|
||||
.artifacts
|
||||
.iter()
|
||||
.filter(|a| a.kind == ArtifactKind::LiveUrl)
|
||||
.count();
|
||||
assert_eq!(live_urls, 1);
|
||||
}
|
||||
}
|
||||
@@ -342,7 +342,6 @@ mod tests {
|
||||
pentest_imap_password: None,
|
||||
admin_api_token: None,
|
||||
tenant_registry_url: None,
|
||||
unified_pipeline: false,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -138,7 +138,7 @@ impl GitOps {
|
||||
/// Build credentials from agent config + per-repo overrides
|
||||
pub fn make_repo_credentials(
|
||||
config: &compliance_core::AgentConfig,
|
||||
repo: &compliance_core::models::TrackedRepository,
|
||||
repo: &crate::pipeline::repo_view::RepoView,
|
||||
) -> RepoCredentials {
|
||||
RepoCredentials {
|
||||
ssh_key_path: Some(config.ssh_key_path.clone()),
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
use mongodb::bson::doc;
|
||||
|
||||
use crate::pipeline::repo_view::RepoView;
|
||||
use compliance_core::models::*;
|
||||
|
||||
use super::orchestrator::{extract_base_url, PipelineOrchestrator};
|
||||
@@ -10,7 +11,7 @@ use crate::trackers;
|
||||
impl PipelineOrchestrator {
|
||||
/// Build an issue tracker client from a repository's tracker configuration.
|
||||
/// Returns `None` if the repo has no tracker configured.
|
||||
pub(super) fn build_tracker(&self, repo: &TrackedRepository) -> Option<TrackerDispatch> {
|
||||
pub(super) fn build_tracker(&self, repo: &RepoView) -> Option<TrackerDispatch> {
|
||||
let tracker_type = repo.tracker_type.as_ref()?;
|
||||
// Per-repo token takes precedence, fall back to global config
|
||||
match tracker_type {
|
||||
@@ -81,7 +82,7 @@ impl PipelineOrchestrator {
|
||||
#[tracing::instrument(skip_all, fields(repo_id = %repo_id))]
|
||||
pub(super) async fn create_tracker_issues(
|
||||
&self,
|
||||
repo: &TrackedRepository,
|
||||
repo: &RepoView,
|
||||
repo_id: &str,
|
||||
new_findings: &[Finding],
|
||||
) -> Result<(), AgentError> {
|
||||
|
||||
@@ -11,6 +11,7 @@ pub mod orchestrator;
|
||||
pub mod patterns;
|
||||
pub mod plan;
|
||||
mod pr_review;
|
||||
pub mod repo_view;
|
||||
pub mod sbom;
|
||||
pub mod semgrep;
|
||||
mod tracker_dispatch;
|
||||
|
||||
@@ -16,6 +16,7 @@ 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;
|
||||
|
||||
@@ -51,72 +52,8 @@ impl PipelineOrchestrator {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all, fields(repo_id = %repo_id, trigger = ?trigger))]
|
||||
pub async fn run(&self, repo_id: &str, trigger: ScanTrigger) -> Result<(), AgentError> {
|
||||
// Look up the repository
|
||||
let repo = self
|
||||
.db
|
||||
.repositories()
|
||||
.find_one(doc! { "_id": mongodb::bson::oid::ObjectId::parse_str(repo_id).map_err(|e| AgentError::Other(e.to_string()))? })
|
||||
.await?
|
||||
.ok_or_else(|| AgentError::Other(format!("Repository {repo_id} not found")))?;
|
||||
|
||||
// Create scan run
|
||||
let scan_run = ScanRun::new(repo_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_pipeline(&repo, &scan_run_id).await;
|
||||
|
||||
// Update scan run status
|
||||
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!(repo_id, error = %e, "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(|_| ())
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip_all, fields(repo_id = repo.name.as_str()))]
|
||||
async fn run_pipeline(
|
||||
&self,
|
||||
repo: &TrackedRepository,
|
||||
scan_run_id: &str,
|
||||
) -> Result<u32, AgentError> {
|
||||
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
|
||||
@@ -130,7 +67,6 @@ impl PipelineOrchestrator {
|
||||
return Ok(0);
|
||||
}
|
||||
|
||||
let current_sha = GitOps::get_head_sha(&repo_path)?;
|
||||
let mut all_findings: Vec<Finding> = Vec::new();
|
||||
|
||||
// Stage 1: Semgrep SAST
|
||||
@@ -396,20 +332,9 @@ impl PipelineOrchestrator {
|
||||
tracing::warn!("[{repo_id}] Issue creation failed: {e}");
|
||||
}
|
||||
|
||||
// Stage 7: Update repository
|
||||
self.db
|
||||
.repositories()
|
||||
.update_one(
|
||||
doc! { "_id": repo.id },
|
||||
doc! {
|
||||
"$set": {
|
||||
"last_scanned_commit": ¤t_sha,
|
||||
"updated_at": mongodb::bson::DateTime::now(),
|
||||
},
|
||||
"$inc": { "findings_count": new_count as i64 },
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
// 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");
|
||||
@@ -526,7 +451,7 @@ impl PipelineOrchestrator {
|
||||
|
||||
match target.code_artifact() {
|
||||
Some(code) if code.kind == ArtifactKind::GitRepo => {
|
||||
let repo = repo_view_from_target(target, code);
|
||||
let repo = RepoView::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)
|
||||
@@ -703,7 +628,7 @@ impl PipelineOrchestrator {
|
||||
async fn finalize_target(
|
||||
&self,
|
||||
target: &OnboardedTarget,
|
||||
repo: &TrackedRepository,
|
||||
repo: &RepoView,
|
||||
new_count: u32,
|
||||
) -> Result<(), AgentError> {
|
||||
let oid = match target.id {
|
||||
@@ -751,37 +676,6 @@ 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"
|
||||
@@ -840,7 +734,7 @@ mod tests {
|
||||
target.artifacts.push(artifact);
|
||||
|
||||
let code = target.code_artifact().expect("code artifact");
|
||||
let repo = repo_view_from_target(&target, code);
|
||||
let repo = RepoView::from_target(&target, code);
|
||||
|
||||
assert_eq!(repo.id, target.id); // preserved
|
||||
assert_eq!(repo.git_url, "https://git/acme.git");
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
use crate::pipeline::repo_view::RepoView;
|
||||
use compliance_core::models::*;
|
||||
|
||||
use super::dedup::compute_fingerprint;
|
||||
@@ -14,7 +15,7 @@ impl PipelineOrchestrator {
|
||||
#[tracing::instrument(skip_all, fields(repo_id = %repo_id, pr_number))]
|
||||
pub async fn run_pr_review(
|
||||
&self,
|
||||
repo: &TrackedRepository,
|
||||
repo: &RepoView,
|
||||
repo_id: &str,
|
||||
pr_number: u64,
|
||||
base_sha: &str,
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
//! `RepoView` — an internal, non-persisted view of a code target for the scan
|
||||
//! pipeline.
|
||||
//!
|
||||
//! It replaces the old persisted `TrackedRepository` model. The pipeline
|
||||
//! (SAST → SBOM → CVE → triage → issues → DAST, and PR review) only ever needs a
|
||||
//! flat bundle of git + issue-tracker + auth fields; those are projected from an
|
||||
//! [`OnboardedTarget`] and its code [`Artifact`] by [`RepoView::from_target`].
|
||||
//! Nothing here is written to Mongo — onboarded targets are the sole persisted
|
||||
//! entity.
|
||||
|
||||
use compliance_core::models::{Artifact, OnboardedTarget, TrackerType};
|
||||
|
||||
/// A flat, pipeline-facing view of a code target. Built from an onboarded
|
||||
/// target; never persisted.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct RepoView {
|
||||
/// The onboarded target's id (used as `repo_id` across findings/sbom/etc.).
|
||||
pub id: Option<mongodb::bson::oid::ObjectId>,
|
||||
pub name: String,
|
||||
pub git_url: String,
|
||||
pub default_branch: String,
|
||||
pub local_path: Option<String>,
|
||||
pub scan_schedule: Option<String>,
|
||||
pub webhook_enabled: bool,
|
||||
pub webhook_secret: Option<String>,
|
||||
pub tracker_type: Option<TrackerType>,
|
||||
pub tracker_owner: Option<String>,
|
||||
pub tracker_repo: Option<String>,
|
||||
pub tracker_token: Option<String>,
|
||||
pub auth_token: Option<String>,
|
||||
pub auth_username: Option<String>,
|
||||
pub last_scanned_commit: Option<String>,
|
||||
pub findings_count: u32,
|
||||
}
|
||||
|
||||
impl RepoView {
|
||||
/// Project an onboarded target + its code artifact into a pipeline view.
|
||||
pub fn from_target(target: &OnboardedTarget, code: &Artifact) -> Self {
|
||||
let mut view = Self {
|
||||
id: target.id,
|
||||
name: target.name.clone(),
|
||||
git_url: code.source_ref.clone(),
|
||||
default_branch: "main".to_string(),
|
||||
local_path: None,
|
||||
scan_schedule: target.scan_schedule.clone(),
|
||||
webhook_enabled: target.webhook_enabled,
|
||||
webhook_secret: target.webhook_secret.clone(),
|
||||
tracker_type: None,
|
||||
tracker_owner: None,
|
||||
tracker_repo: None,
|
||||
tracker_token: None,
|
||||
auth_token: None,
|
||||
auth_username: None,
|
||||
last_scanned_commit: None,
|
||||
findings_count: target.findings_count,
|
||||
};
|
||||
if let Some(git) = &code.git {
|
||||
view.default_branch = git.default_branch.clone();
|
||||
view.last_scanned_commit = git.last_scanned_commit.clone();
|
||||
view.local_path = git.local_path.clone();
|
||||
}
|
||||
if let Some(auth) = &code.auth {
|
||||
view.auth_token = auth.secret.clone();
|
||||
view.auth_username = auth.username.clone();
|
||||
}
|
||||
if let Some(it) = &target.scan_config.issue_tracker {
|
||||
view.tracker_type = it.tracker_type.clone();
|
||||
view.tracker_owner = it.owner.clone();
|
||||
view.tracker_repo = it.repo.clone();
|
||||
view.tracker_token = it.token.clone();
|
||||
}
|
||||
view
|
||||
}
|
||||
}
|
||||
@@ -348,7 +348,7 @@ async fn monitor_cves(agent: &ComplianceAgent, tenant_id: &str) {
|
||||
std::collections::HashMap::new();
|
||||
for rid in &repo_ids {
|
||||
if let Ok(oid) = mongodb::bson::oid::ObjectId::parse_str(rid) {
|
||||
if let Ok(Some(repo)) = db.repositories().find_one(doc! { "_id": oid }).await {
|
||||
if let Ok(Some(repo)) = db.onboarded_targets().find_one(doc! { "_id": oid }).await {
|
||||
repo_names.insert(rid.clone(), repo.name.clone());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -31,7 +31,7 @@ pub async fn handle_gitea_webhook(
|
||||
}
|
||||
};
|
||||
let repo = match db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.find_one(mongodb::bson::doc! { "_id": oid })
|
||||
.await
|
||||
{
|
||||
|
||||
@@ -31,7 +31,7 @@ pub async fn handle_github_webhook(
|
||||
}
|
||||
};
|
||||
let repo = match db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.find_one(mongodb::bson::doc! { "_id": oid })
|
||||
.await
|
||||
{
|
||||
|
||||
@@ -27,7 +27,7 @@ pub async fn handle_gitlab_webhook(
|
||||
}
|
||||
};
|
||||
let repo = match db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.find_one(mongodb::bson::doc! { "_id": oid })
|
||||
.await
|
||||
{
|
||||
|
||||
@@ -70,7 +70,6 @@ impl TestServer {
|
||||
pentest_imap_password: None,
|
||||
admin_api_token: None,
|
||||
tenant_registry_url: None,
|
||||
unified_pipeline: false,
|
||||
};
|
||||
|
||||
let agent = ComplianceAgent::new(config, db_pool);
|
||||
|
||||
@@ -113,15 +113,16 @@ async fn delete_repo_cascades_to_dast_and_pentest_data() {
|
||||
// Create a repo
|
||||
let resp = server
|
||||
.post(
|
||||
"/api/v1/repositories",
|
||||
"/api/v1/targets",
|
||||
&json!({
|
||||
"name": "cascade-test",
|
||||
"git_url": "https://github.com/example/cascade-test.git",
|
||||
"target_type": "web_app",
|
||||
"artifacts": [{ "kind": "git_repo", "source_ref": "https://github.com/example/cascade-test.git", "branch": "main" }],
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
let body: serde_json::Value = resp.json().await.unwrap();
|
||||
let repo_id = body["data"]["id"].as_str().unwrap().to_string();
|
||||
let repo_id = body["data"]["_id"]["$oid"].as_str().unwrap().to_string();
|
||||
|
||||
// Insert DAST target linked to repo
|
||||
let target_id = insert_dast_target(&server, &repo_id, "cascade-target").await;
|
||||
@@ -140,9 +141,7 @@ async fn delete_repo_cascades_to_dast_and_pentest_data() {
|
||||
assert_eq!(count_docs(&server, "dast_findings").await, 1);
|
||||
|
||||
// Delete the repo
|
||||
let resp = server
|
||||
.delete(&format!("/api/v1/repositories/{repo_id}"))
|
||||
.await;
|
||||
let resp = server.delete(&format!("/api/v1/targets/{repo_id}")).await;
|
||||
assert_eq!(resp.status(), 200);
|
||||
|
||||
// All downstream data should be gone
|
||||
@@ -161,15 +160,16 @@ async fn delete_repo_cascades_sast_findings_and_sbom() {
|
||||
// Create a repo
|
||||
let resp = server
|
||||
.post(
|
||||
"/api/v1/repositories",
|
||||
"/api/v1/targets",
|
||||
&json!({
|
||||
"name": "sast-cascade",
|
||||
"git_url": "https://github.com/example/sast-cascade.git",
|
||||
"target_type": "web_app",
|
||||
"artifacts": [{ "kind": "git_repo", "source_ref": "https://github.com/example/sast-cascade.git", "branch": "main" }],
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
let body: serde_json::Value = resp.json().await.unwrap();
|
||||
let repo_id = body["data"]["id"].as_str().unwrap().to_string();
|
||||
let repo_id = body["data"]["_id"]["$oid"].as_str().unwrap().to_string();
|
||||
|
||||
// Insert SAST finding and SBOM entry
|
||||
let mongodb_uri = std::env::var("TEST_MONGODB_URI")
|
||||
@@ -209,9 +209,7 @@ async fn delete_repo_cascades_sast_findings_and_sbom() {
|
||||
assert_eq!(count_docs(&server, "sbom_entries").await, 1);
|
||||
|
||||
// Delete repo
|
||||
server
|
||||
.delete(&format!("/api/v1/repositories/{repo_id}"))
|
||||
.await;
|
||||
server.delete(&format!("/api/v1/targets/{repo_id}")).await;
|
||||
|
||||
// Both should be gone
|
||||
assert_eq!(count_docs(&server, "findings").await, 0);
|
||||
|
||||
@@ -3,5 +3,4 @@ mod dast;
|
||||
mod findings;
|
||||
mod health;
|
||||
mod onboarding;
|
||||
mod repositories;
|
||||
mod stats;
|
||||
|
||||
@@ -1,110 +0,0 @@
|
||||
use crate::common::TestServer;
|
||||
use serde_json::json;
|
||||
|
||||
#[tokio::test]
|
||||
async fn add_and_list_repository() {
|
||||
let server = TestServer::start().await;
|
||||
|
||||
// Initially empty
|
||||
let resp = server.get("/api/v1/repositories").await;
|
||||
assert_eq!(resp.status(), 200);
|
||||
let body: serde_json::Value = resp.json().await.unwrap();
|
||||
assert_eq!(body["data"].as_array().unwrap().len(), 0);
|
||||
|
||||
// Add a repository
|
||||
let resp = server
|
||||
.post(
|
||||
"/api/v1/repositories",
|
||||
&json!({
|
||||
"name": "test-repo",
|
||||
"git_url": "https://github.com/example/test-repo.git",
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(resp.status(), 200);
|
||||
let body: serde_json::Value = resp.json().await.unwrap();
|
||||
let repo_id = body["data"]["id"].as_str().unwrap().to_string();
|
||||
assert!(!repo_id.is_empty());
|
||||
|
||||
// List should now return 1
|
||||
let resp = server.get("/api/v1/repositories").await;
|
||||
let body: serde_json::Value = resp.json().await.unwrap();
|
||||
let repos = body["data"].as_array().unwrap();
|
||||
assert_eq!(repos.len(), 1);
|
||||
assert_eq!(repos[0]["name"], "test-repo");
|
||||
|
||||
server.cleanup().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn add_duplicate_repository_fails() {
|
||||
let server = TestServer::start().await;
|
||||
|
||||
let payload = json!({
|
||||
"name": "dup-repo",
|
||||
"git_url": "https://github.com/example/dup-repo.git",
|
||||
});
|
||||
|
||||
// First add succeeds
|
||||
let resp = server.post("/api/v1/repositories", &payload).await;
|
||||
assert_eq!(resp.status(), 200);
|
||||
|
||||
// Second add with same git_url should fail (unique index)
|
||||
let resp = server.post("/api/v1/repositories", &payload).await;
|
||||
assert_ne!(resp.status(), 200);
|
||||
|
||||
server.cleanup().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delete_repository() {
|
||||
let server = TestServer::start().await;
|
||||
|
||||
// Add a repo
|
||||
let resp = server
|
||||
.post(
|
||||
"/api/v1/repositories",
|
||||
&json!({
|
||||
"name": "to-delete",
|
||||
"git_url": "https://github.com/example/to-delete.git",
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
let body: serde_json::Value = resp.json().await.unwrap();
|
||||
let repo_id = body["data"]["id"].as_str().unwrap();
|
||||
|
||||
// Delete it
|
||||
let resp = server
|
||||
.delete(&format!("/api/v1/repositories/{repo_id}"))
|
||||
.await;
|
||||
assert_eq!(resp.status(), 200);
|
||||
|
||||
// List should be empty again
|
||||
let resp = server.get("/api/v1/repositories").await;
|
||||
let body: serde_json::Value = resp.json().await.unwrap();
|
||||
assert_eq!(body["data"].as_array().unwrap().len(), 0);
|
||||
|
||||
server.cleanup().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delete_nonexistent_repository_returns_404() {
|
||||
let server = TestServer::start().await;
|
||||
|
||||
let resp = server
|
||||
.delete("/api/v1/repositories/000000000000000000000000")
|
||||
.await;
|
||||
assert_eq!(resp.status(), 404);
|
||||
|
||||
server.cleanup().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delete_invalid_id_returns_400() {
|
||||
let server = TestServer::start().await;
|
||||
|
||||
let resp = server.delete("/api/v1/repositories/not-a-valid-id").await;
|
||||
assert_eq!(resp.status(), 400);
|
||||
|
||||
server.cleanup().await;
|
||||
}
|
||||
@@ -5,13 +5,14 @@ use serde_json::json;
|
||||
async fn stats_overview_reflects_inserted_data() {
|
||||
let server = TestServer::start().await;
|
||||
|
||||
// Add a repo
|
||||
// Add a target
|
||||
server
|
||||
.post(
|
||||
"/api/v1/repositories",
|
||||
"/api/v1/targets",
|
||||
&json!({
|
||||
"name": "stats-repo",
|
||||
"git_url": "https://github.com/example/stats-repo.git",
|
||||
"target_type": "web_app",
|
||||
"artifacts": [{ "kind": "git_repo", "source_ref": "https://github.com/example/stats-repo.git", "branch": "main" }],
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -1,156 +0,0 @@
|
||||
// Integration tests for the onboarding backfill migration.
|
||||
//
|
||||
// Requires MongoDB (set TEST_MONGODB_URI if not at the default).
|
||||
// Not run in CI (which is `--lib` only) — run locally:
|
||||
// cargo test -p compliance-agent --test e2e migration
|
||||
|
||||
use compliance_agent::database::{Database, DatabasePool};
|
||||
use compliance_agent::migrate::onboarding;
|
||||
use compliance_core::models::{
|
||||
ArtifactKind, DastTarget, DastTargetType, TargetType, TrackedRepository,
|
||||
};
|
||||
use mongodb::bson::{doc, Document};
|
||||
|
||||
async fn fresh_db() -> (DatabasePool, String, Database) {
|
||||
let uri = std::env::var("TEST_MONGODB_URI")
|
||||
.unwrap_or_else(|_| "mongodb://root:example@localhost:27017/?authSource=admin".into());
|
||||
// Prefix must fit the pool's 30-char cap (`<prefix>_<32 hex>` <= 63).
|
||||
let prefix = format!("t_{}", &uuid::Uuid::new_v4().simple().to_string()[..16]);
|
||||
let pool = DatabasePool::connect(&uri, &prefix)
|
||||
.await
|
||||
.expect("connect mongo");
|
||||
let db = pool.for_tenant_id("t1").await.expect("tenant db");
|
||||
(pool, prefix, db)
|
||||
}
|
||||
|
||||
async fn cleanup(pool: &DatabasePool, prefix: &str) {
|
||||
if let Ok(names) = pool.client().list_database_names().await {
|
||||
for n in names {
|
||||
if n.starts_with(prefix) {
|
||||
pool.client().database(&n).drop().await.ok();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn backfill_folds_relinks_is_idempotent_and_reversible() {
|
||||
let (pool, prefix, db) = fresh_db().await;
|
||||
|
||||
// Seed a repo.
|
||||
let repo = TrackedRepository::new("acme".into(), "https://git/acme.git".into());
|
||||
let repo_id = db
|
||||
.repositories()
|
||||
.insert_one(repo)
|
||||
.await
|
||||
.expect("insert repo")
|
||||
.inserted_id
|
||||
.as_object_id()
|
||||
.expect("repo oid");
|
||||
|
||||
// A DAST target linked to the repo (folds + promotes to WebApp + relinks).
|
||||
let mut linked = DastTarget::new(
|
||||
"acme-web".into(),
|
||||
"https://acme.example.com".into(),
|
||||
DastTargetType::WebApp,
|
||||
);
|
||||
linked.repo_id = Some(repo_id.to_hex());
|
||||
let linked_id = db
|
||||
.dast_targets()
|
||||
.insert_one(linked)
|
||||
.await
|
||||
.expect("insert linked dast")
|
||||
.inserted_id
|
||||
.as_object_id()
|
||||
.expect("linked oid");
|
||||
|
||||
// A repo-less DAST target (standalone).
|
||||
let standalone = DastTarget::new(
|
||||
"acme-api".into(),
|
||||
"https://api.acme.com".into(),
|
||||
DastTargetType::RestApi,
|
||||
);
|
||||
let standalone_id = db
|
||||
.dast_targets()
|
||||
.insert_one(standalone)
|
||||
.await
|
||||
.expect("insert standalone dast")
|
||||
.inserted_id
|
||||
.as_object_id()
|
||||
.expect("standalone oid");
|
||||
|
||||
// A DAST scan run pointing at the linked target — should be relinked to the repo.
|
||||
db.collection_named::<Document>("dast_scan_runs")
|
||||
.insert_one(doc! { "target_id": linked_id.to_hex(), "status": "completed" })
|
||||
.await
|
||||
.expect("insert dast run");
|
||||
|
||||
// --- Backfill ---
|
||||
assert!(!onboarding::already_applied(&db).await.unwrap());
|
||||
let report = onboarding::backfill_onboarded_targets(&db, false)
|
||||
.await
|
||||
.expect("backfill");
|
||||
assert_eq!(report.repos_migrated, 1);
|
||||
assert_eq!(report.dast_targets_folded, 1);
|
||||
assert_eq!(report.dast_targets_standalone, 1);
|
||||
assert!(onboarding::already_applied(&db).await.unwrap());
|
||||
|
||||
// Repo target: preserved _id, has git + folded live-url, promoted to WebApp.
|
||||
let repo_target = db
|
||||
.onboarded_targets()
|
||||
.find_one(doc! { "_id": repo_id })
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("repo target");
|
||||
assert!(repo_target.has(ArtifactKind::GitRepo));
|
||||
assert!(repo_target.has(ArtifactKind::LiveUrl));
|
||||
assert_eq!(repo_target.target_type, TargetType::WebApp);
|
||||
|
||||
// Standalone target: preserved _id, live-url, backend service.
|
||||
let standalone_target = db
|
||||
.onboarded_targets()
|
||||
.find_one(doc! { "_id": standalone_id })
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("standalone target");
|
||||
assert!(standalone_target.has(ArtifactKind::LiveUrl));
|
||||
assert_eq!(standalone_target.target_type, TargetType::BackendService);
|
||||
|
||||
// The DAST run was relinked from the old dast id to the repo (unified) id.
|
||||
let run = db
|
||||
.collection_named::<Document>("dast_scan_runs")
|
||||
.find_one(doc! {})
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("run");
|
||||
assert_eq!(run.get_str("target_id").unwrap(), repo_id.to_hex());
|
||||
|
||||
// --- Idempotent: re-run migrates nothing new ---
|
||||
let again = onboarding::backfill_onboarded_targets(&db, false)
|
||||
.await
|
||||
.expect("backfill again");
|
||||
assert_eq!(again.repos_migrated, 0);
|
||||
assert_eq!(again.dast_targets_folded, 0);
|
||||
assert_eq!(again.dast_targets_standalone, 0);
|
||||
assert!(again.skipped_existing >= 2);
|
||||
|
||||
// --- Revert: onboarded targets gone, relink undone, marker cleared ---
|
||||
onboarding::revert(&db).await.expect("revert");
|
||||
assert_eq!(
|
||||
db.onboarded_targets()
|
||||
.count_documents(doc! {})
|
||||
.await
|
||||
.unwrap(),
|
||||
0
|
||||
);
|
||||
let run_after = db
|
||||
.collection_named::<Document>("dast_scan_runs")
|
||||
.find_one(doc! {})
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("run");
|
||||
assert_eq!(run_after.get_str("target_id").unwrap(), linked_id.to_hex());
|
||||
assert!(!onboarding::already_applied(&db).await.unwrap());
|
||||
|
||||
cleanup(&pool, &prefix).await;
|
||||
}
|
||||
@@ -7,4 +7,3 @@
|
||||
// Or nightly: (via CI with MongoDB service container)
|
||||
|
||||
mod api;
|
||||
mod migration;
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
#![allow(clippy::expect_used, clippy::unwrap_used)]
|
||||
|
||||
use compliance_agent::database::DatabasePool;
|
||||
use compliance_core::models::TrackedRepository;
|
||||
use compliance_core::models::{Artifact, OnboardedTarget, TargetType};
|
||||
use compliance_core::{OrgRole, TenantContext, TenantStatus};
|
||||
use mongodb::bson::doc;
|
||||
|
||||
@@ -28,27 +28,12 @@ fn ctx(tenant_id: &str, slug: &str) -> TenantContext {
|
||||
}
|
||||
}
|
||||
|
||||
fn fixture_repo(name: &str, git_url: &str) -> TrackedRepository {
|
||||
TrackedRepository {
|
||||
id: None,
|
||||
name: name.to_string(),
|
||||
git_url: git_url.to_string(),
|
||||
default_branch: "main".to_string(),
|
||||
local_path: None,
|
||||
scan_schedule: None,
|
||||
webhook_enabled: false,
|
||||
webhook_secret: None,
|
||||
tracker_type: None,
|
||||
tracker_owner: None,
|
||||
tracker_repo: None,
|
||||
tracker_token: None,
|
||||
auth_token: None,
|
||||
auth_username: None,
|
||||
last_scanned_commit: None,
|
||||
findings_count: 0,
|
||||
created_at: chrono::Utc::now(),
|
||||
updated_at: chrono::Utc::now(),
|
||||
}
|
||||
fn fixture_repo(name: &str, git_url: &str) -> OnboardedTarget {
|
||||
let mut target = OnboardedTarget::new(name.to_string(), TargetType::WebApp);
|
||||
target
|
||||
.artifacts
|
||||
.push(Artifact::git_repo(git_url.to_string(), "main".to_string()));
|
||||
target
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -71,12 +56,12 @@ async fn pool_isolates_tenants_at_driver_level() {
|
||||
|
||||
// Write distinct repos into each tenant's database.
|
||||
acme_db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.insert_one(fixture_repo("acme-app", "git@example.com:acme/app.git"))
|
||||
.await
|
||||
.expect("insert acme");
|
||||
globex_db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.insert_one(fixture_repo(
|
||||
"globex-platform",
|
||||
"git@example.com:globex/platform.git",
|
||||
@@ -173,12 +158,12 @@ async fn admin_helpers_list_and_drop_tenant_dbs() {
|
||||
let acme_db = pool.for_tenant(&acme).await.expect("acme db");
|
||||
let globex_db = pool.for_tenant(&globex).await.expect("globex db");
|
||||
acme_db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.insert_one(fixture_repo("acme-app", "git@example.com:acme/app.git"))
|
||||
.await
|
||||
.expect("insert acme");
|
||||
globex_db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.insert_one(fixture_repo("globex-app", "git@example.com:globex/app.git"))
|
||||
.await
|
||||
.expect("insert globex");
|
||||
@@ -284,9 +269,9 @@ fn short_id() -> String {
|
||||
}
|
||||
|
||||
/// Drain a `repositories` find cursor on the given tenant database.
|
||||
async fn collect(db: &compliance_agent::database::Database) -> Vec<TrackedRepository> {
|
||||
async fn collect(db: &compliance_agent::database::Database) -> Vec<OnboardedTarget> {
|
||||
let mut cursor = db
|
||||
.repositories()
|
||||
.onboarded_targets()
|
||||
.find(doc! {})
|
||||
.await
|
||||
.expect("find repositories");
|
||||
|
||||
Reference in New Issue
Block a user