Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 08c4ec4cff | |||
| 0f6dd1135e |
@@ -6,7 +6,7 @@ use tokio::sync::{broadcast, watch, Semaphore};
|
|||||||
use compliance_core::models::pentest::PentestEvent;
|
use compliance_core::models::pentest::PentestEvent;
|
||||||
use compliance_core::AgentConfig;
|
use compliance_core::AgentConfig;
|
||||||
|
|
||||||
use crate::database::{Database, DatabasePool};
|
use crate::database::DatabasePool;
|
||||||
use crate::llm::LlmClient;
|
use crate::llm::LlmClient;
|
||||||
use crate::pipeline::orchestrator::PipelineOrchestrator;
|
use crate::pipeline::orchestrator::PipelineOrchestrator;
|
||||||
|
|
||||||
@@ -16,12 +16,9 @@ const DEFAULT_MAX_CONCURRENT_SESSIONS: usize = 5;
|
|||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub struct ComplianceAgent {
|
pub struct ComplianceAgent {
|
||||||
pub config: AgentConfig,
|
pub config: AgentConfig,
|
||||||
/// Transitional single-database handle. Used by handlers that have
|
/// Per-tenant Mongo broker. Every code path must obtain a
|
||||||
/// not yet been migrated to `db_pool.for_tenant(&ctx)` (M7.2-B/C).
|
/// tenant-scoped [`crate::database::Database`] from this pool —
|
||||||
/// Will be removed once every call site is tenant-scoped (M7.2-D).
|
/// there is no single shared database any more.
|
||||||
pub db: Database,
|
|
||||||
/// Per-tenant Mongo broker introduced in M7.2-A. Handlers should
|
|
||||||
/// prefer this and obtain a tenant-scoped [`Database`] from it.
|
|
||||||
pub db_pool: DatabasePool,
|
pub db_pool: DatabasePool,
|
||||||
pub llm: Arc<LlmClient>,
|
pub llm: Arc<LlmClient>,
|
||||||
pub http: reqwest::Client,
|
pub http: reqwest::Client,
|
||||||
@@ -34,7 +31,7 @@ pub struct ComplianceAgent {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl ComplianceAgent {
|
impl ComplianceAgent {
|
||||||
pub fn new(config: AgentConfig, db: Database, db_pool: DatabasePool) -> Self {
|
pub fn new(config: AgentConfig, db_pool: DatabasePool) -> Self {
|
||||||
let llm = Arc::new(LlmClient::new(
|
let llm = Arc::new(LlmClient::new(
|
||||||
config.litellm_url.clone(),
|
config.litellm_url.clone(),
|
||||||
config.litellm_api_key.clone(),
|
config.litellm_api_key.clone(),
|
||||||
@@ -48,7 +45,6 @@ impl ComplianceAgent {
|
|||||||
.unwrap_or_default();
|
.unwrap_or_default();
|
||||||
Self {
|
Self {
|
||||||
config,
|
config,
|
||||||
db,
|
|
||||||
db_pool,
|
db_pool,
|
||||||
llm,
|
llm,
|
||||||
http,
|
http,
|
||||||
@@ -60,28 +56,27 @@ impl ComplianceAgent {
|
|||||||
|
|
||||||
pub async fn run_scan(
|
pub async fn run_scan(
|
||||||
&self,
|
&self,
|
||||||
|
tenant_id: &str,
|
||||||
repo_id: &str,
|
repo_id: &str,
|
||||||
trigger: compliance_core::models::ScanTrigger,
|
trigger: compliance_core::models::ScanTrigger,
|
||||||
) -> Result<(), crate::error::AgentError> {
|
) -> Result<(), crate::error::AgentError> {
|
||||||
let orchestrator = PipelineOrchestrator::new(
|
let db = self.db_pool.for_tenant_id(tenant_id).await?;
|
||||||
self.config.clone(),
|
let orchestrator =
|
||||||
self.db.clone(),
|
PipelineOrchestrator::new(self.config.clone(), db, self.llm.clone(), self.http.clone());
|
||||||
self.llm.clone(),
|
|
||||||
self.http.clone(),
|
|
||||||
);
|
|
||||||
orchestrator.run(repo_id, trigger).await
|
orchestrator.run(repo_id, trigger).await
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Run a PR review: scan the diff and post review comments.
|
/// Run a PR review: scan the diff and post review comments.
|
||||||
pub async fn run_pr_review(
|
pub async fn run_pr_review(
|
||||||
&self,
|
&self,
|
||||||
|
tenant_id: &str,
|
||||||
repo_id: &str,
|
repo_id: &str,
|
||||||
pr_number: u64,
|
pr_number: u64,
|
||||||
base_sha: &str,
|
base_sha: &str,
|
||||||
head_sha: &str,
|
head_sha: &str,
|
||||||
) -> Result<(), crate::error::AgentError> {
|
) -> Result<(), crate::error::AgentError> {
|
||||||
let repo = self
|
let db = self.db_pool.for_tenant_id(tenant_id).await?;
|
||||||
.db
|
let repo = db
|
||||||
.repositories()
|
.repositories()
|
||||||
.find_one(mongodb::bson::doc! {
|
.find_one(mongodb::bson::doc! {
|
||||||
"_id": mongodb::bson::oid::ObjectId::parse_str(repo_id)
|
"_id": mongodb::bson::oid::ObjectId::parse_str(repo_id)
|
||||||
@@ -92,12 +87,8 @@ impl ComplianceAgent {
|
|||||||
crate::error::AgentError::Other(format!("Repository {repo_id} not found"))
|
crate::error::AgentError::Other(format!("Repository {repo_id} not found"))
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
let orchestrator = PipelineOrchestrator::new(
|
let orchestrator =
|
||||||
self.config.clone(),
|
PipelineOrchestrator::new(self.config.clone(), db, self.llm.clone(), self.http.clone());
|
||||||
self.db.clone(),
|
|
||||||
self.llm.clone(),
|
|
||||||
self.http.clone(),
|
|
||||||
);
|
|
||||||
orchestrator
|
orchestrator
|
||||||
.run_pr_review(&repo, repo_id, pr_number, base_sha, head_sha)
|
.run_pr_review(&repo, repo_id, pr_number, base_sha, head_sha)
|
||||||
.await
|
.await
|
||||||
|
|||||||
@@ -158,11 +158,16 @@ pub async fn get_ssh_public_key(
|
|||||||
#[tracing::instrument(skip_all, fields(repo_id = %id))]
|
#[tracing::instrument(skip_all, fields(repo_id = %id))]
|
||||||
pub async fn trigger_scan(
|
pub async fn trigger_scan(
|
||||||
Extension(agent): AgentExt,
|
Extension(agent): AgentExt,
|
||||||
|
tenant: TenantCtx,
|
||||||
Path(id): Path<String>,
|
Path(id): Path<String>,
|
||||||
) -> Result<Json<serde_json::Value>, StatusCode> {
|
) -> Result<Json<serde_json::Value>, StatusCode> {
|
||||||
let agent_clone = (*agent).clone();
|
let agent_clone = (*agent).clone();
|
||||||
|
let tenant_id = tenant.0.tenant_id.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
if let Err(e) = agent_clone.run_scan(&id, ScanTrigger::Manual).await {
|
if let Err(e) = agent_clone
|
||||||
|
.run_scan(&tenant_id, &id, ScanTrigger::Manual)
|
||||||
|
.await
|
||||||
|
{
|
||||||
tracing::error!("Manual scan failed for {id}: {e}");
|
tracing::error!("Manual scan failed for {id}: {e}");
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -78,19 +78,28 @@ impl DatabasePool {
|
|||||||
/// first call per tenant (per process). Cheap on the hot path —
|
/// first call per tenant (per process). Cheap on the hot path —
|
||||||
/// subsequent calls skip the round-trip.
|
/// subsequent calls skip the round-trip.
|
||||||
pub async fn for_tenant(&self, ctx: &TenantContext) -> Result<Database, AgentError> {
|
pub async fn for_tenant(&self, ctx: &TenantContext) -> Result<Database, AgentError> {
|
||||||
let db_name = self.tenant_db_name(&ctx.tenant_id);
|
self.for_tenant_id(&ctx.tenant_id).await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Like [`Self::for_tenant`] but accepts a bare tenant_id.
|
||||||
|
/// For background paths (scheduler, webhooks, pipeline orchestrators)
|
||||||
|
/// that don't have a full [`TenantContext`] but know which tenant
|
||||||
|
/// they're operating on (typically resolved from a URL path, a job
|
||||||
|
/// argument, or the registry).
|
||||||
|
pub async fn for_tenant_id(&self, tenant_id: &str) -> Result<Database, AgentError> {
|
||||||
|
let db_name = self.tenant_db_name(tenant_id);
|
||||||
let db = Database::from_database(self.client.database(&db_name));
|
let db = Database::from_database(self.client.database(&db_name));
|
||||||
// `DashMap::insert` returns the previous value; `None` means we
|
// `DashMap::insert` returns the previous value; `None` means we
|
||||||
// were the first writer for this tenant_id and own the
|
// were the first writer for this tenant_id and own the
|
||||||
// index-ensure work.
|
// index-ensure work.
|
||||||
if self.ensured.insert(ctx.tenant_id.clone(), ()).is_none() {
|
if self.ensured.insert(tenant_id.to_string(), ()).is_none() {
|
||||||
if let Err(e) = db.ensure_indexes().await {
|
if let Err(e) = db.ensure_indexes().await {
|
||||||
// Roll the marker back so the next request retries.
|
// Roll the marker back so the next request retries.
|
||||||
self.ensured.remove(&ctx.tenant_id);
|
self.ensured.remove(tenant_id);
|
||||||
return Err(e);
|
return Err(e);
|
||||||
}
|
}
|
||||||
tracing::debug!(
|
tracing::debug!(
|
||||||
tenant_id = %ctx.tenant_id,
|
tenant_id = %tenant_id,
|
||||||
db_name = %db_name,
|
db_name = %db_name,
|
||||||
"Indexes ensured for tenant database"
|
"Indexes ensured for tenant database"
|
||||||
);
|
);
|
||||||
@@ -131,6 +140,43 @@ impl DatabasePool {
|
|||||||
pub fn client(&self) -> &Client {
|
pub fn client(&self) -> &Client {
|
||||||
&self.client
|
&self.client
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// List every Mongo database currently belonging to this pool,
|
||||||
|
/// identified by the `<db_prefix>_` prefix. The result is the raw
|
||||||
|
/// database names — opening one for offboarding/cleanup goes
|
||||||
|
/// through [`Self::client`].
|
||||||
|
///
|
||||||
|
/// Note: hashed-fallback names (very long tenant_ids) lose the
|
||||||
|
/// original tenant_id at the cluster level — we know a database
|
||||||
|
/// exists for *some* tenant but not which one. In practice
|
||||||
|
/// tenant_ids are UUIDs (36 chars) and never hit the fallback,
|
||||||
|
/// so this is a theoretical concern, not an operational one.
|
||||||
|
pub async fn list_tenant_db_names(&self) -> Result<Vec<String>, AgentError> {
|
||||||
|
let prefix = format!("{}_", self.db_prefix);
|
||||||
|
let names = self.client.list_database_names().await?;
|
||||||
|
Ok(names
|
||||||
|
.into_iter()
|
||||||
|
.filter(|n| n.starts_with(&prefix))
|
||||||
|
.collect())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drop the database for a specific tenant. Used by GDPR delete
|
||||||
|
/// and tenant offboarding. Idempotent — dropping a non-existent
|
||||||
|
/// database is a no-op at the driver level.
|
||||||
|
///
|
||||||
|
/// Also evicts the tenant from the in-memory `ensured` set so a
|
||||||
|
/// later re-provision triggers fresh `ensure_indexes`.
|
||||||
|
pub async fn drop_tenant(&self, tenant_id: &str) -> Result<(), AgentError> {
|
||||||
|
let db_name = self.tenant_db_name(tenant_id);
|
||||||
|
self.client.database(&db_name).drop().await?;
|
||||||
|
self.ensured.remove(tenant_id);
|
||||||
|
tracing::info!(
|
||||||
|
tenant_id = %tenant_id,
|
||||||
|
db_name = %db_name,
|
||||||
|
"Dropped tenant database"
|
||||||
|
);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Mongo database names disallow `/`, `\`, `.`, `"`, `$`, ` `, and NUL.
|
/// Mongo database names disallow `/`, `\`, `.`, `"`, `$`, ` `, and NUL.
|
||||||
|
|||||||
@@ -25,16 +25,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
tracing::info!("Connecting to MongoDB...");
|
tracing::info!("Connecting to MongoDB...");
|
||||||
let db = database::Database::connect(&config.mongodb_uri, &config.mongodb_database).await?;
|
// Per-tenant pool only — the agent has no shared "default" database
|
||||||
db.ensure_indexes().await?;
|
// after M7.2-D. `mongodb_database` is now the db-name prefix used
|
||||||
|
// for tenant databases (`<prefix>_<tenant_id>`).
|
||||||
// M7.2-A: per-tenant pool. Uses `mongodb_database` as the db-name
|
|
||||||
// prefix so tenant databases land as `<prefix>_<tenant_id>` next to
|
|
||||||
// the legacy single-tenant database.
|
|
||||||
let db_pool =
|
let db_pool =
|
||||||
database::DatabasePool::connect(&config.mongodb_uri, &config.mongodb_database).await?;
|
database::DatabasePool::connect(&config.mongodb_uri, &config.mongodb_database).await?;
|
||||||
|
|
||||||
let agent = agent::ComplianceAgent::new(config.clone(), db.clone(), db_pool);
|
let agent = agent::ComplianceAgent::new(config.clone(), db_pool);
|
||||||
|
|
||||||
tracing::info!("Starting scheduler...");
|
tracing::info!("Starting scheduler...");
|
||||||
let scheduler_agent = agent.clone();
|
let scheduler_agent = agent.clone();
|
||||||
|
|||||||
@@ -4,8 +4,14 @@ use tokio_cron_scheduler::{Job, JobScheduler};
|
|||||||
use compliance_core::models::ScanTrigger;
|
use compliance_core::models::ScanTrigger;
|
||||||
|
|
||||||
use crate::agent::ComplianceAgent;
|
use crate::agent::ComplianceAgent;
|
||||||
|
use crate::database::Database;
|
||||||
use crate::error::AgentError;
|
use crate::error::AgentError;
|
||||||
|
|
||||||
|
/// Default tenant the scheduler runs against when `SCHEDULER_TENANT_IDS`
|
||||||
|
/// isn't set. Matches the dev-injector default so a bare `cargo run` has
|
||||||
|
/// the scheduler scanning whatever lives in `<prefix>_dev`.
|
||||||
|
const DEFAULT_SCHEDULER_TENANT_ID: &str = "dev";
|
||||||
|
|
||||||
pub async fn start_scheduler(agent: &ComplianceAgent) -> Result<(), AgentError> {
|
pub async fn start_scheduler(agent: &ComplianceAgent) -> Result<(), AgentError> {
|
||||||
let sched = JobScheduler::new()
|
let sched = JobScheduler::new()
|
||||||
.await
|
.await
|
||||||
@@ -18,7 +24,9 @@ pub async fn start_scheduler(agent: &ComplianceAgent) -> Result<(), AgentError>
|
|||||||
let agent = scan_agent.clone();
|
let agent = scan_agent.clone();
|
||||||
Box::pin(async move {
|
Box::pin(async move {
|
||||||
tracing::info!("Scheduled scan triggered");
|
tracing::info!("Scheduled scan triggered");
|
||||||
scan_all_repos(&agent).await;
|
for tenant_id in scheduler_tenants() {
|
||||||
|
scan_all_repos(&agent, &tenant_id).await;
|
||||||
|
}
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
.map_err(|e| AgentError::Scheduler(format!("Failed to create scan job: {e}")))?;
|
.map_err(|e| AgentError::Scheduler(format!("Failed to create scan job: {e}")))?;
|
||||||
@@ -34,7 +42,9 @@ pub async fn start_scheduler(agent: &ComplianceAgent) -> Result<(), AgentError>
|
|||||||
let agent = cve_agent.clone();
|
let agent = cve_agent.clone();
|
||||||
Box::pin(async move {
|
Box::pin(async move {
|
||||||
tracing::info!("CVE monitor triggered");
|
tracing::info!("CVE monitor triggered");
|
||||||
monitor_cves(&agent).await;
|
for tenant_id in scheduler_tenants() {
|
||||||
|
monitor_cves(&agent, &tenant_id).await;
|
||||||
|
}
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
.map_err(|e| AgentError::Scheduler(format!("Failed to create CVE monitor job: {e}")))?;
|
.map_err(|e| AgentError::Scheduler(format!("Failed to create CVE monitor job: {e}")))?;
|
||||||
@@ -48,8 +58,9 @@ pub async fn start_scheduler(agent: &ComplianceAgent) -> Result<(), AgentError>
|
|||||||
.await
|
.await
|
||||||
.map_err(|e| AgentError::Scheduler(format!("Failed to start scheduler: {e}")))?;
|
.map_err(|e| AgentError::Scheduler(format!("Failed to start scheduler: {e}")))?;
|
||||||
|
|
||||||
|
let tenants = scheduler_tenants();
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
"Scheduler started: scans='{}', CVE monitor='{}'",
|
"Scheduler started: scans='{}', CVE monitor='{}', tenants={tenants:?}",
|
||||||
agent.config.scan_schedule,
|
agent.config.scan_schedule,
|
||||||
agent.config.cve_monitor_schedule,
|
agent.config.cve_monitor_schedule,
|
||||||
);
|
);
|
||||||
@@ -60,13 +71,47 @@ pub async fn start_scheduler(agent: &ComplianceAgent) -> Result<(), AgentError>
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn scan_all_repos(agent: &ComplianceAgent) {
|
/// Tenants the scheduler iterates each tick. From `SCHEDULER_TENANT_IDS`
|
||||||
|
/// (comma-separated), or `DEFAULT_SCHEDULER_TENANT_ID` if unset. M7.2-D
|
||||||
|
/// will replace this with a pull from the tenant-registry.
|
||||||
|
fn scheduler_tenants() -> Vec<String> {
|
||||||
|
std::env::var("SCHEDULER_TENANT_IDS")
|
||||||
|
.ok()
|
||||||
|
.map(|s| {
|
||||||
|
s.split(',')
|
||||||
|
.map(str::trim)
|
||||||
|
.filter(|s| !s.is_empty())
|
||||||
|
.map(String::from)
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
})
|
||||||
|
.filter(|v| !v.is_empty())
|
||||||
|
.unwrap_or_else(|| vec![DEFAULT_SCHEDULER_TENANT_ID.to_string()])
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Resolve the per-tenant database. Logs and returns `None` on failure
|
||||||
|
/// so the loop in the caller can continue with other tenants.
|
||||||
|
async fn tenant_db(agent: &ComplianceAgent, tenant_id: &str) -> Option<Database> {
|
||||||
|
match agent.db_pool.for_tenant_id(tenant_id).await {
|
||||||
|
Ok(db) => Some(db),
|
||||||
|
Err(e) => {
|
||||||
|
tracing::error!("Scheduler: cannot open tenant database '{tenant_id}': {e}");
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn scan_all_repos(agent: &ComplianceAgent, tenant_id: &str) {
|
||||||
use futures_util::StreamExt;
|
use futures_util::StreamExt;
|
||||||
|
|
||||||
let cursor = match agent.db.repositories().find(doc! {}).await {
|
let db = match tenant_db(agent, tenant_id).await {
|
||||||
|
Some(db) => db,
|
||||||
|
None => return,
|
||||||
|
};
|
||||||
|
|
||||||
|
let cursor = match db.repositories().find(doc! {}).await {
|
||||||
Ok(c) => c,
|
Ok(c) => c,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::error!("Failed to list repos for scheduled scan: {e}");
|
tracing::error!("Failed to list repos for tenant '{tenant_id}': {e}");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
@@ -75,33 +120,44 @@ async fn scan_all_repos(agent: &ComplianceAgent) {
|
|||||||
|
|
||||||
for repo in repos {
|
for repo in repos {
|
||||||
let repo_id = repo.id.map(|id| id.to_hex()).unwrap_or_default();
|
let repo_id = repo.id.map(|id| id.to_hex()).unwrap_or_default();
|
||||||
if let Err(e) = agent.run_scan(&repo_id, ScanTrigger::Scheduled).await {
|
if let Err(e) = agent
|
||||||
tracing::error!("Scheduled scan failed for {}: {e}", repo.name);
|
.run_scan(tenant_id, &repo_id, ScanTrigger::Scheduled)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
tracing::error!(
|
||||||
|
"Scheduled scan failed for {} (tenant '{tenant_id}'): {e}",
|
||||||
|
repo.name
|
||||||
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn monitor_cves(agent: &ComplianceAgent) {
|
async fn monitor_cves(agent: &ComplianceAgent, tenant_id: &str) {
|
||||||
use compliance_core::models::notification::{parse_severity, CveNotification};
|
use compliance_core::models::notification::{parse_severity, CveNotification};
|
||||||
use compliance_core::models::SbomEntry;
|
use compliance_core::models::SbomEntry;
|
||||||
use futures_util::StreamExt;
|
use futures_util::StreamExt;
|
||||||
|
|
||||||
|
let db = match tenant_db(agent, tenant_id).await {
|
||||||
|
Some(db) => db,
|
||||||
|
None => return,
|
||||||
|
};
|
||||||
|
|
||||||
// Fetch all SBOM entries grouped by repo
|
// Fetch all SBOM entries grouped by repo
|
||||||
let cursor = match agent.db.sbom_entries().find(doc! {}).await {
|
let cursor = match db.sbom_entries().find(doc! {}).await {
|
||||||
Ok(c) => c,
|
Ok(c) => c,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::error!("CVE monitor: failed to list SBOM entries: {e}");
|
tracing::error!("CVE monitor: failed to list SBOM entries for '{tenant_id}': {e}");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
let entries: Vec<SbomEntry> = cursor.filter_map(|r| async { r.ok() }).collect().await;
|
let entries: Vec<SbomEntry> = cursor.filter_map(|r| async { r.ok() }).collect().await;
|
||||||
if entries.is_empty() {
|
if entries.is_empty() {
|
||||||
tracing::debug!("CVE monitor: no SBOM entries, skipping");
|
tracing::debug!("CVE monitor: no SBOM entries for tenant '{tenant_id}', skipping");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
"CVE monitor: checking {} dependencies for new CVEs",
|
"CVE monitor: checking {} dependencies for new CVEs (tenant '{tenant_id}')",
|
||||||
entries.len()
|
entries.len()
|
||||||
);
|
);
|
||||||
|
|
||||||
@@ -112,7 +168,7 @@ async fn monitor_cves(agent: &ComplianceAgent) {
|
|||||||
std::collections::HashMap::new();
|
std::collections::HashMap::new();
|
||||||
for rid in &repo_ids {
|
for rid in &repo_ids {
|
||||||
if let Ok(oid) = mongodb::bson::oid::ObjectId::parse_str(rid) {
|
if let Ok(oid) = mongodb::bson::oid::ObjectId::parse_str(rid) {
|
||||||
if let Ok(Some(repo)) = agent.db.repositories().find_one(doc! { "_id": oid }).await {
|
if let Ok(Some(repo)) = db.repositories().find_one(doc! { "_id": oid }).await {
|
||||||
repo_names.insert(rid.clone(), repo.name.clone());
|
repo_names.insert(rid.clone(), repo.name.clone());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -160,8 +216,7 @@ async fn monitor_cves(agent: &ComplianceAgent) {
|
|||||||
for alert in &alerts {
|
for alert in &alerts {
|
||||||
let filter = doc! { "cve_id": &alert.cve_id, "repo_id": &alert.repo_id };
|
let filter = doc! { "cve_id": &alert.cve_id, "repo_id": &alert.repo_id };
|
||||||
let update = doc! { "$setOnInsert": mongodb::bson::to_bson(alert).unwrap_or_default() };
|
let update = doc! { "$setOnInsert": mongodb::bson::to_bson(alert).unwrap_or_default() };
|
||||||
let _ = agent
|
let _ = db
|
||||||
.db
|
|
||||||
.cve_alerts()
|
.cve_alerts()
|
||||||
.update_one(filter, update)
|
.update_one(filter, update)
|
||||||
.upsert(true)
|
.upsert(true)
|
||||||
@@ -174,8 +229,7 @@ async fn monitor_cves(agent: &ComplianceAgent) {
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
if let Some(entry_id) = &entry.id {
|
if let Some(entry_id) = &entry.id {
|
||||||
let _ = agent
|
let _ = db
|
||||||
.db
|
|
||||||
.sbom_entries()
|
.sbom_entries()
|
||||||
.update_one(
|
.update_one(
|
||||||
doc! { "_id": entry_id },
|
doc! { "_id": entry_id },
|
||||||
@@ -213,8 +267,7 @@ async fn monitor_cves(agent: &ComplianceAgent) {
|
|||||||
let update = doc! {
|
let update = doc! {
|
||||||
"$setOnInsert": mongodb::bson::to_bson(¬ification).unwrap_or_default()
|
"$setOnInsert": mongodb::bson::to_bson(¬ification).unwrap_or_default()
|
||||||
};
|
};
|
||||||
match agent
|
match db
|
||||||
.db
|
|
||||||
.cve_notifications()
|
.cve_notifications()
|
||||||
.update_one(filter, update)
|
.update_one(filter, update)
|
||||||
.upsert(true)
|
.upsert(true)
|
||||||
@@ -232,8 +285,10 @@ async fn monitor_cves(agent: &ComplianceAgent) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if new_notifications > 0 {
|
if new_notifications > 0 {
|
||||||
tracing::info!("CVE monitor: created {new_notifications} new notification(s)");
|
tracing::info!(
|
||||||
|
"CVE monitor: created {new_notifications} new notification(s) for tenant '{tenant_id}'"
|
||||||
|
);
|
||||||
} else {
|
} else {
|
||||||
tracing::info!("CVE monitor: no new CVEs found");
|
tracing::info!("CVE monitor: no new CVEs found for tenant '{tenant_id}'");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -14,24 +14,30 @@ type HmacSha256 = Hmac<Sha256>;
|
|||||||
|
|
||||||
pub async fn handle_gitea_webhook(
|
pub async fn handle_gitea_webhook(
|
||||||
Extension(agent): Extension<Arc<ComplianceAgent>>,
|
Extension(agent): Extension<Arc<ComplianceAgent>>,
|
||||||
Path(repo_id): Path<String>,
|
Path((tenant_id, repo_id)): Path<(String, String)>,
|
||||||
headers: HeaderMap,
|
headers: HeaderMap,
|
||||||
body: Bytes,
|
body: Bytes,
|
||||||
) -> StatusCode {
|
) -> StatusCode {
|
||||||
// Look up the repo to get its webhook secret
|
// Look up the repo in the tenant's database to get its webhook secret
|
||||||
let oid = match mongodb::bson::oid::ObjectId::parse_str(&repo_id) {
|
let oid = match mongodb::bson::oid::ObjectId::parse_str(&repo_id) {
|
||||||
Ok(oid) => oid,
|
Ok(oid) => oid,
|
||||||
Err(_) => return StatusCode::NOT_FOUND,
|
Err(_) => return StatusCode::NOT_FOUND,
|
||||||
};
|
};
|
||||||
let repo = match agent
|
let db = match agent.db_pool.for_tenant_id(&tenant_id).await {
|
||||||
.db
|
Ok(db) => db,
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!("Gitea webhook: cannot open tenant database '{tenant_id}': {e}");
|
||||||
|
return StatusCode::NOT_FOUND;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let repo = match db
|
||||||
.repositories()
|
.repositories()
|
||||||
.find_one(mongodb::bson::doc! { "_id": oid })
|
.find_one(mongodb::bson::doc! { "_id": oid })
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(Some(repo)) => repo,
|
Ok(Some(repo)) => repo,
|
||||||
_ => {
|
_ => {
|
||||||
tracing::warn!("Gitea webhook: repo {repo_id} not found");
|
tracing::warn!("Gitea webhook: repo {repo_id} not found in tenant '{tenant_id}'");
|
||||||
return StatusCode::NOT_FOUND;
|
return StatusCode::NOT_FOUND;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
@@ -66,15 +72,21 @@ pub async fn handle_gitea_webhook(
|
|||||||
"push" => {
|
"push" => {
|
||||||
let agent_clone = (*agent).clone();
|
let agent_clone = (*agent).clone();
|
||||||
let repo_id = repo_id.clone();
|
let repo_id = repo_id.clone();
|
||||||
|
let tenant_id = tenant_id.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
tracing::info!("Gitea push webhook: triggering scan for {repo_id}");
|
tracing::info!(
|
||||||
if let Err(e) = agent_clone.run_scan(&repo_id, ScanTrigger::Webhook).await {
|
"Gitea push webhook: triggering scan for {repo_id} in tenant {tenant_id}"
|
||||||
|
);
|
||||||
|
if let Err(e) = agent_clone
|
||||||
|
.run_scan(&tenant_id, &repo_id, ScanTrigger::Webhook)
|
||||||
|
.await
|
||||||
|
{
|
||||||
tracing::error!("Webhook-triggered scan failed: {e}");
|
tracing::error!("Webhook-triggered scan failed: {e}");
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
StatusCode::OK
|
StatusCode::OK
|
||||||
}
|
}
|
||||||
"pull_request" => handle_pull_request(agent, &repo_id, &payload).await,
|
"pull_request" => handle_pull_request(agent, &tenant_id, &repo_id, &payload).await,
|
||||||
_ => {
|
_ => {
|
||||||
tracing::debug!("Gitea webhook: ignoring event '{event}'");
|
tracing::debug!("Gitea webhook: ignoring event '{event}'");
|
||||||
StatusCode::OK
|
StatusCode::OK
|
||||||
@@ -84,6 +96,7 @@ pub async fn handle_gitea_webhook(
|
|||||||
|
|
||||||
async fn handle_pull_request(
|
async fn handle_pull_request(
|
||||||
agent: Arc<ComplianceAgent>,
|
agent: Arc<ComplianceAgent>,
|
||||||
|
tenant_id: &str,
|
||||||
repo_id: &str,
|
repo_id: &str,
|
||||||
payload: &serde_json::Value,
|
payload: &serde_json::Value,
|
||||||
) -> StatusCode {
|
) -> StatusCode {
|
||||||
@@ -106,13 +119,14 @@ async fn handle_pull_request(
|
|||||||
}
|
}
|
||||||
|
|
||||||
let repo_id = repo_id.to_string();
|
let repo_id = repo_id.to_string();
|
||||||
|
let tenant_id = tenant_id.to_string();
|
||||||
let head_sha = head_sha.to_string();
|
let head_sha = head_sha.to_string();
|
||||||
let base_sha = base_sha.to_string();
|
let base_sha = base_sha.to_string();
|
||||||
let agent_clone = (*agent).clone();
|
let agent_clone = (*agent).clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
tracing::info!("Gitea PR webhook: reviewing PR #{pr_number} on {repo_id}");
|
tracing::info!("Gitea PR webhook: reviewing PR #{pr_number} on {repo_id}");
|
||||||
if let Err(e) = agent_clone
|
if let Err(e) = agent_clone
|
||||||
.run_pr_review(&repo_id, pr_number, &base_sha, &head_sha)
|
.run_pr_review(&tenant_id, &repo_id, pr_number, &base_sha, &head_sha)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
tracing::error!("PR review failed for #{pr_number}: {e}");
|
tracing::error!("PR review failed for #{pr_number}: {e}");
|
||||||
|
|||||||
@@ -14,24 +14,30 @@ type HmacSha256 = Hmac<Sha256>;
|
|||||||
|
|
||||||
pub async fn handle_github_webhook(
|
pub async fn handle_github_webhook(
|
||||||
Extension(agent): Extension<Arc<ComplianceAgent>>,
|
Extension(agent): Extension<Arc<ComplianceAgent>>,
|
||||||
Path(repo_id): Path<String>,
|
Path((tenant_id, repo_id)): Path<(String, String)>,
|
||||||
headers: HeaderMap,
|
headers: HeaderMap,
|
||||||
body: Bytes,
|
body: Bytes,
|
||||||
) -> StatusCode {
|
) -> StatusCode {
|
||||||
// Look up the repo to get its webhook secret
|
// Look up the repo in the tenant's database to get its webhook secret
|
||||||
let oid = match mongodb::bson::oid::ObjectId::parse_str(&repo_id) {
|
let oid = match mongodb::bson::oid::ObjectId::parse_str(&repo_id) {
|
||||||
Ok(oid) => oid,
|
Ok(oid) => oid,
|
||||||
Err(_) => return StatusCode::NOT_FOUND,
|
Err(_) => return StatusCode::NOT_FOUND,
|
||||||
};
|
};
|
||||||
let repo = match agent
|
let db = match agent.db_pool.for_tenant_id(&tenant_id).await {
|
||||||
.db
|
Ok(db) => db,
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!("GitHub webhook: cannot open tenant database '{tenant_id}': {e}");
|
||||||
|
return StatusCode::NOT_FOUND;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let repo = match db
|
||||||
.repositories()
|
.repositories()
|
||||||
.find_one(mongodb::bson::doc! { "_id": oid })
|
.find_one(mongodb::bson::doc! { "_id": oid })
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(Some(repo)) => repo,
|
Ok(Some(repo)) => repo,
|
||||||
_ => {
|
_ => {
|
||||||
tracing::warn!("GitHub webhook: repo {repo_id} not found");
|
tracing::warn!("GitHub webhook: repo {repo_id} not found in tenant '{tenant_id}'");
|
||||||
return StatusCode::NOT_FOUND;
|
return StatusCode::NOT_FOUND;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
@@ -66,15 +72,21 @@ pub async fn handle_github_webhook(
|
|||||||
"push" => {
|
"push" => {
|
||||||
let agent_clone = (*agent).clone();
|
let agent_clone = (*agent).clone();
|
||||||
let repo_id = repo_id.clone();
|
let repo_id = repo_id.clone();
|
||||||
|
let tenant_id = tenant_id.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
tracing::info!("GitHub push webhook: triggering scan for {repo_id}");
|
tracing::info!(
|
||||||
if let Err(e) = agent_clone.run_scan(&repo_id, ScanTrigger::Webhook).await {
|
"GitHub push webhook: triggering scan for {repo_id} in tenant {tenant_id}"
|
||||||
|
);
|
||||||
|
if let Err(e) = agent_clone
|
||||||
|
.run_scan(&tenant_id, &repo_id, ScanTrigger::Webhook)
|
||||||
|
.await
|
||||||
|
{
|
||||||
tracing::error!("Webhook-triggered scan failed: {e}");
|
tracing::error!("Webhook-triggered scan failed: {e}");
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
StatusCode::OK
|
StatusCode::OK
|
||||||
}
|
}
|
||||||
"pull_request" => handle_pull_request(agent, &repo_id, &payload).await,
|
"pull_request" => handle_pull_request(agent, &tenant_id, &repo_id, &payload).await,
|
||||||
_ => {
|
_ => {
|
||||||
tracing::debug!("GitHub webhook: ignoring event '{event}'");
|
tracing::debug!("GitHub webhook: ignoring event '{event}'");
|
||||||
StatusCode::OK
|
StatusCode::OK
|
||||||
@@ -84,6 +96,7 @@ pub async fn handle_github_webhook(
|
|||||||
|
|
||||||
async fn handle_pull_request(
|
async fn handle_pull_request(
|
||||||
agent: Arc<ComplianceAgent>,
|
agent: Arc<ComplianceAgent>,
|
||||||
|
tenant_id: &str,
|
||||||
repo_id: &str,
|
repo_id: &str,
|
||||||
payload: &serde_json::Value,
|
payload: &serde_json::Value,
|
||||||
) -> StatusCode {
|
) -> StatusCode {
|
||||||
@@ -105,13 +118,14 @@ async fn handle_pull_request(
|
|||||||
}
|
}
|
||||||
|
|
||||||
let repo_id = repo_id.to_string();
|
let repo_id = repo_id.to_string();
|
||||||
|
let tenant_id = tenant_id.to_string();
|
||||||
let head_sha = head_sha.to_string();
|
let head_sha = head_sha.to_string();
|
||||||
let base_sha = base_sha.to_string();
|
let base_sha = base_sha.to_string();
|
||||||
let agent_clone = (*agent).clone();
|
let agent_clone = (*agent).clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
tracing::info!("GitHub PR webhook: reviewing PR #{pr_number} on {repo_id}");
|
tracing::info!("GitHub PR webhook: reviewing PR #{pr_number} on {repo_id}");
|
||||||
if let Err(e) = agent_clone
|
if let Err(e) = agent_clone
|
||||||
.run_pr_review(&repo_id, pr_number, &base_sha, &head_sha)
|
.run_pr_review(&tenant_id, &repo_id, pr_number, &base_sha, &head_sha)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
tracing::error!("PR review failed for #{pr_number}: {e}");
|
tracing::error!("PR review failed for #{pr_number}: {e}");
|
||||||
|
|||||||
@@ -10,24 +10,30 @@ use crate::agent::ComplianceAgent;
|
|||||||
|
|
||||||
pub async fn handle_gitlab_webhook(
|
pub async fn handle_gitlab_webhook(
|
||||||
Extension(agent): Extension<Arc<ComplianceAgent>>,
|
Extension(agent): Extension<Arc<ComplianceAgent>>,
|
||||||
Path(repo_id): Path<String>,
|
Path((tenant_id, repo_id)): Path<(String, String)>,
|
||||||
headers: HeaderMap,
|
headers: HeaderMap,
|
||||||
body: Bytes,
|
body: Bytes,
|
||||||
) -> StatusCode {
|
) -> StatusCode {
|
||||||
// Look up the repo to get its webhook secret
|
// Look up the repo in the tenant's database to get its webhook secret
|
||||||
let oid = match mongodb::bson::oid::ObjectId::parse_str(&repo_id) {
|
let oid = match mongodb::bson::oid::ObjectId::parse_str(&repo_id) {
|
||||||
Ok(oid) => oid,
|
Ok(oid) => oid,
|
||||||
Err(_) => return StatusCode::NOT_FOUND,
|
Err(_) => return StatusCode::NOT_FOUND,
|
||||||
};
|
};
|
||||||
let repo = match agent
|
let db = match agent.db_pool.for_tenant_id(&tenant_id).await {
|
||||||
.db
|
Ok(db) => db,
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!("GitLab webhook: cannot open tenant database '{tenant_id}': {e}");
|
||||||
|
return StatusCode::NOT_FOUND;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let repo = match db
|
||||||
.repositories()
|
.repositories()
|
||||||
.find_one(mongodb::bson::doc! { "_id": oid })
|
.find_one(mongodb::bson::doc! { "_id": oid })
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(Some(repo)) => repo,
|
Ok(Some(repo)) => repo,
|
||||||
_ => {
|
_ => {
|
||||||
tracing::warn!("GitLab webhook: repo {repo_id} not found");
|
tracing::warn!("GitLab webhook: repo {repo_id} not found in tenant '{tenant_id}'");
|
||||||
return StatusCode::NOT_FOUND;
|
return StatusCode::NOT_FOUND;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
@@ -59,15 +65,21 @@ pub async fn handle_gitlab_webhook(
|
|||||||
"push" => {
|
"push" => {
|
||||||
let agent_clone = (*agent).clone();
|
let agent_clone = (*agent).clone();
|
||||||
let repo_id = repo_id.clone();
|
let repo_id = repo_id.clone();
|
||||||
|
let tenant_id = tenant_id.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
tracing::info!("GitLab push webhook: triggering scan for {repo_id}");
|
tracing::info!(
|
||||||
if let Err(e) = agent_clone.run_scan(&repo_id, ScanTrigger::Webhook).await {
|
"GitLab push webhook: triggering scan for {repo_id} in tenant {tenant_id}"
|
||||||
|
);
|
||||||
|
if let Err(e) = agent_clone
|
||||||
|
.run_scan(&tenant_id, &repo_id, ScanTrigger::Webhook)
|
||||||
|
.await
|
||||||
|
{
|
||||||
tracing::error!("Webhook-triggered scan failed: {e}");
|
tracing::error!("Webhook-triggered scan failed: {e}");
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
StatusCode::OK
|
StatusCode::OK
|
||||||
}
|
}
|
||||||
"merge_request" => handle_merge_request(agent, &repo_id, &payload).await,
|
"merge_request" => handle_merge_request(agent, &tenant_id, &repo_id, &payload).await,
|
||||||
_ => {
|
_ => {
|
||||||
tracing::debug!("GitLab webhook: ignoring event '{event_type}'");
|
tracing::debug!("GitLab webhook: ignoring event '{event_type}'");
|
||||||
StatusCode::OK
|
StatusCode::OK
|
||||||
@@ -77,6 +89,7 @@ pub async fn handle_gitlab_webhook(
|
|||||||
|
|
||||||
async fn handle_merge_request(
|
async fn handle_merge_request(
|
||||||
agent: Arc<ComplianceAgent>,
|
agent: Arc<ComplianceAgent>,
|
||||||
|
tenant_id: &str,
|
||||||
repo_id: &str,
|
repo_id: &str,
|
||||||
payload: &serde_json::Value,
|
payload: &serde_json::Value,
|
||||||
) -> StatusCode {
|
) -> StatusCode {
|
||||||
@@ -101,13 +114,14 @@ async fn handle_merge_request(
|
|||||||
}
|
}
|
||||||
|
|
||||||
let repo_id = repo_id.to_string();
|
let repo_id = repo_id.to_string();
|
||||||
|
let tenant_id = tenant_id.to_string();
|
||||||
let head_sha = head_sha.to_string();
|
let head_sha = head_sha.to_string();
|
||||||
let base_sha = base_sha.to_string();
|
let base_sha = base_sha.to_string();
|
||||||
let agent_clone = (*agent).clone();
|
let agent_clone = (*agent).clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
tracing::info!("GitLab MR webhook: reviewing MR !{mr_iid} on {repo_id}");
|
tracing::info!("GitLab MR webhook: reviewing MR !{mr_iid} on {repo_id}");
|
||||||
if let Err(e) = agent_clone
|
if let Err(e) = agent_clone
|
||||||
.run_pr_review(&repo_id, mr_iid, &base_sha, &head_sha)
|
.run_pr_review(&tenant_id, &repo_id, mr_iid, &base_sha, &head_sha)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
tracing::error!("MR review failed for !{mr_iid}: {e}");
|
tracing::error!("MR review failed for !{mr_iid}: {e}");
|
||||||
|
|||||||
@@ -9,17 +9,21 @@ use crate::webhooks::{gitea, github, gitlab};
|
|||||||
|
|
||||||
pub async fn start_webhook_server(agent: &ComplianceAgent) -> Result<(), AgentError> {
|
pub async fn start_webhook_server(agent: &ComplianceAgent) -> Result<(), AgentError> {
|
||||||
let app = Router::new()
|
let app = Router::new()
|
||||||
// Per-repo webhook URLs: /webhook/{platform}/{repo_id}
|
// Per-tenant per-repo webhook URLs: /webhook/{tenant_id}/{platform}/{repo_id}
|
||||||
|
// The tenant_id is resolved from the URL path because webhooks
|
||||||
|
// arrive without a JWT — they're authenticated via per-repo HMAC,
|
||||||
|
// not via the tenant gate. The dashboard surfaces the full URL
|
||||||
|
// including the tenant_id when the repo is registered.
|
||||||
.route(
|
.route(
|
||||||
"/webhook/github/{repo_id}",
|
"/webhook/{tenant_id}/github/{repo_id}",
|
||||||
post(github::handle_github_webhook),
|
post(github::handle_github_webhook),
|
||||||
)
|
)
|
||||||
.route(
|
.route(
|
||||||
"/webhook/gitlab/{repo_id}",
|
"/webhook/{tenant_id}/gitlab/{repo_id}",
|
||||||
post(gitlab::handle_gitlab_webhook),
|
post(gitlab::handle_gitlab_webhook),
|
||||||
)
|
)
|
||||||
.route(
|
.route(
|
||||||
"/webhook/gitea/{repo_id}",
|
"/webhook/{tenant_id}/gitea/{repo_id}",
|
||||||
post(gitea::handle_gitea_webhook),
|
post(gitea::handle_gitea_webhook),
|
||||||
)
|
)
|
||||||
.layer(Extension(Arc::new(agent.clone())));
|
.layer(Extension(Arc::new(agent.clone())));
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ use std::sync::Arc;
|
|||||||
|
|
||||||
use compliance_agent::agent::ComplianceAgent;
|
use compliance_agent::agent::ComplianceAgent;
|
||||||
use compliance_agent::api;
|
use compliance_agent::api;
|
||||||
use compliance_agent::database::{Database, DatabasePool};
|
use compliance_agent::database::DatabasePool;
|
||||||
use compliance_core::AgentConfig;
|
use compliance_core::AgentConfig;
|
||||||
use secrecy::SecretString;
|
use secrecy::SecretString;
|
||||||
|
|
||||||
@@ -28,11 +28,6 @@ impl TestServer {
|
|||||||
// Unique database name per test run to avoid collisions
|
// Unique database name per test run to avoid collisions
|
||||||
let db_name = format!("test_{}", uuid::Uuid::new_v4().simple());
|
let db_name = format!("test_{}", uuid::Uuid::new_v4().simple());
|
||||||
|
|
||||||
let db = Database::connect(&mongodb_uri, &db_name)
|
|
||||||
.await
|
|
||||||
.expect("Failed to connect to MongoDB — is it running?");
|
|
||||||
db.ensure_indexes().await.expect("Failed to create indexes");
|
|
||||||
|
|
||||||
let db_pool = DatabasePool::connect(&mongodb_uri, &db_name)
|
let db_pool = DatabasePool::connect(&mongodb_uri, &db_name)
|
||||||
.await
|
.await
|
||||||
.expect("Failed to build DatabasePool");
|
.expect("Failed to build DatabasePool");
|
||||||
@@ -73,7 +68,7 @@ impl TestServer {
|
|||||||
pentest_imap_password: None,
|
pentest_imap_password: None,
|
||||||
};
|
};
|
||||||
|
|
||||||
let agent = ComplianceAgent::new(config, db, db_pool);
|
let agent = ComplianceAgent::new(config, db_pool);
|
||||||
|
|
||||||
// Build the router with the agent extension. After M7.2-B every
|
// Build the router with the agent extension. After M7.2-B every
|
||||||
// handler takes a TenantCtx extractor; without KC in the test
|
// handler takes a TenantCtx extractor; without KC in the test
|
||||||
@@ -164,12 +159,11 @@ impl TestServer {
|
|||||||
&self.db_name
|
&self.db_name
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Drop the test database on cleanup. Post-M7.2-B the actual data
|
/// Drop every per-tenant database belonging to this test run.
|
||||||
/// lives in `<db_name>_<tenant>` per-tenant databases; list those
|
/// Post-M7.2-D the agent never opens a `db_name` directly —
|
||||||
/// off the cluster and drop them too.
|
/// data lives only in `<db_name>_<tenant>` per-tenant databases.
|
||||||
pub async fn cleanup(&self) {
|
pub async fn cleanup(&self) {
|
||||||
if let Ok(client) = mongodb::Client::with_uri_str(&self.mongodb_uri).await {
|
if let Ok(client) = mongodb::Client::with_uri_str(&self.mongodb_uri).await {
|
||||||
client.database(&self.db_name).drop().await.ok();
|
|
||||||
if let Ok(names) = client.list_database_names().await {
|
if let Ok(names) = client.list_database_names().await {
|
||||||
let prefix = format!("{}_", self.db_name);
|
let prefix = format!("{}_", self.db_name);
|
||||||
for name in names {
|
for name in names {
|
||||||
|
|||||||
@@ -158,6 +158,70 @@ async fn tenant_db_name_sanitizes_unsafe_characters() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn admin_helpers_list_and_drop_tenant_dbs() {
|
||||||
|
let uri = std::env::var("TEST_MONGODB_URI")
|
||||||
|
.unwrap_or_else(|_| "mongodb://root:example@localhost:27017/?authSource=admin".into());
|
||||||
|
let prefix = format!("m72d_{}", short_id());
|
||||||
|
let pool = DatabasePool::connect(&uri, &prefix).await.expect("connect");
|
||||||
|
|
||||||
|
let acme = ctx("00000000-0000-0000-0000-00000000acme", "acme");
|
||||||
|
let globex = ctx("00000000-0000-0000-0000-0000globex000", "globex");
|
||||||
|
|
||||||
|
// Provision two tenants and write a doc into each so the databases
|
||||||
|
// actually materialize on the cluster (Mongo lazily creates 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()
|
||||||
|
.insert_one(fixture_repo("acme-app", "git@example.com:acme/app.git"))
|
||||||
|
.await
|
||||||
|
.expect("insert acme");
|
||||||
|
globex_db
|
||||||
|
.repositories()
|
||||||
|
.insert_one(fixture_repo("globex-app", "git@example.com:globex/app.git"))
|
||||||
|
.await
|
||||||
|
.expect("insert globex");
|
||||||
|
|
||||||
|
// list_tenant_db_names sees both, filtered by prefix
|
||||||
|
let names = pool.list_tenant_db_names().await.expect("list tenants");
|
||||||
|
let acme_name = pool.tenant_db_name(&acme.tenant_id);
|
||||||
|
let globex_name = pool.tenant_db_name(&globex.tenant_id);
|
||||||
|
assert!(
|
||||||
|
names.contains(&acme_name),
|
||||||
|
"expected {acme_name} in {names:?}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
names.contains(&globex_name),
|
||||||
|
"expected {globex_name} in {names:?}"
|
||||||
|
);
|
||||||
|
for name in &names {
|
||||||
|
assert!(name.starts_with(&format!("{prefix}_")));
|
||||||
|
}
|
||||||
|
|
||||||
|
// drop_tenant removes acme's DB
|
||||||
|
pool.drop_tenant(&acme.tenant_id)
|
||||||
|
.await
|
||||||
|
.expect("drop acme tenant");
|
||||||
|
let after = pool
|
||||||
|
.list_tenant_db_names()
|
||||||
|
.await
|
||||||
|
.expect("list tenants after drop");
|
||||||
|
assert!(
|
||||||
|
!after.contains(&acme_name),
|
||||||
|
"acme should be gone after drop, got {after:?}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
after.contains(&globex_name),
|
||||||
|
"globex should still be present, got {after:?}"
|
||||||
|
);
|
||||||
|
|
||||||
|
// Cleanup remaining
|
||||||
|
pool.drop_tenant(&globex.tenant_id)
|
||||||
|
.await
|
||||||
|
.expect("drop globex tenant");
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn tenant_db_name_falls_back_to_hash_when_too_long() {
|
async fn tenant_db_name_falls_back_to_hash_when_too_long() {
|
||||||
let uri = std::env::var("TEST_MONGODB_URI")
|
let uri = std::env::var("TEST_MONGODB_URI")
|
||||||
|
|||||||
Reference in New Issue
Block a user