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::AgentConfig;
|
||||
|
||||
use crate::database::{Database, DatabasePool};
|
||||
use crate::database::DatabasePool;
|
||||
use crate::llm::LlmClient;
|
||||
use crate::pipeline::orchestrator::PipelineOrchestrator;
|
||||
|
||||
@@ -16,12 +16,9 @@ const DEFAULT_MAX_CONCURRENT_SESSIONS: usize = 5;
|
||||
#[derive(Clone)]
|
||||
pub struct ComplianceAgent {
|
||||
pub config: AgentConfig,
|
||||
/// Transitional single-database handle. Used by handlers that have
|
||||
/// not yet been migrated to `db_pool.for_tenant(&ctx)` (M7.2-B/C).
|
||||
/// Will be removed once every call site is tenant-scoped (M7.2-D).
|
||||
pub db: Database,
|
||||
/// Per-tenant Mongo broker introduced in M7.2-A. Handlers should
|
||||
/// prefer this and obtain a tenant-scoped [`Database`] from it.
|
||||
/// Per-tenant Mongo broker. Every code path must obtain a
|
||||
/// tenant-scoped [`crate::database::Database`] from this pool —
|
||||
/// there is no single shared database any more.
|
||||
pub db_pool: DatabasePool,
|
||||
pub llm: Arc<LlmClient>,
|
||||
pub http: reqwest::Client,
|
||||
@@ -34,7 +31,7 @@ pub struct 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(
|
||||
config.litellm_url.clone(),
|
||||
config.litellm_api_key.clone(),
|
||||
@@ -48,7 +45,6 @@ impl ComplianceAgent {
|
||||
.unwrap_or_default();
|
||||
Self {
|
||||
config,
|
||||
db,
|
||||
db_pool,
|
||||
llm,
|
||||
http,
|
||||
@@ -60,28 +56,27 @@ impl ComplianceAgent {
|
||||
|
||||
pub async fn run_scan(
|
||||
&self,
|
||||
tenant_id: &str,
|
||||
repo_id: &str,
|
||||
trigger: compliance_core::models::ScanTrigger,
|
||||
) -> Result<(), crate::error::AgentError> {
|
||||
let orchestrator = PipelineOrchestrator::new(
|
||||
self.config.clone(),
|
||||
self.db.clone(),
|
||||
self.llm.clone(),
|
||||
self.http.clone(),
|
||||
);
|
||||
let db = self.db_pool.for_tenant_id(tenant_id).await?;
|
||||
let orchestrator =
|
||||
PipelineOrchestrator::new(self.config.clone(), db, self.llm.clone(), self.http.clone());
|
||||
orchestrator.run(repo_id, trigger).await
|
||||
}
|
||||
|
||||
/// Run a PR review: scan the diff and post review comments.
|
||||
pub async fn run_pr_review(
|
||||
&self,
|
||||
tenant_id: &str,
|
||||
repo_id: &str,
|
||||
pr_number: u64,
|
||||
base_sha: &str,
|
||||
head_sha: &str,
|
||||
) -> Result<(), crate::error::AgentError> {
|
||||
let repo = self
|
||||
.db
|
||||
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)
|
||||
@@ -92,12 +87,8 @@ impl ComplianceAgent {
|
||||
crate::error::AgentError::Other(format!("Repository {repo_id} not found"))
|
||||
})?;
|
||||
|
||||
let orchestrator = PipelineOrchestrator::new(
|
||||
self.config.clone(),
|
||||
self.db.clone(),
|
||||
self.llm.clone(),
|
||||
self.http.clone(),
|
||||
);
|
||||
let orchestrator =
|
||||
PipelineOrchestrator::new(self.config.clone(), db, self.llm.clone(), self.http.clone());
|
||||
orchestrator
|
||||
.run_pr_review(&repo, repo_id, pr_number, base_sha, head_sha)
|
||||
.await
|
||||
|
||||
@@ -158,11 +158,16 @@ pub async fn get_ssh_public_key(
|
||||
#[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(&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}");
|
||||
}
|
||||
});
|
||||
|
||||
@@ -78,19 +78,28 @@ impl DatabasePool {
|
||||
/// first call per tenant (per process). Cheap on the hot path —
|
||||
/// subsequent calls skip the round-trip.
|
||||
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));
|
||||
// `DashMap::insert` returns the previous value; `None` means we
|
||||
// were the first writer for this tenant_id and own the
|
||||
// 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 {
|
||||
// Roll the marker back so the next request retries.
|
||||
self.ensured.remove(&ctx.tenant_id);
|
||||
self.ensured.remove(tenant_id);
|
||||
return Err(e);
|
||||
}
|
||||
tracing::debug!(
|
||||
tenant_id = %ctx.tenant_id,
|
||||
tenant_id = %tenant_id,
|
||||
db_name = %db_name,
|
||||
"Indexes ensured for tenant database"
|
||||
);
|
||||
@@ -131,6 +140,43 @@ impl DatabasePool {
|
||||
pub fn 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.
|
||||
|
||||
@@ -25,16 +25,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
}
|
||||
|
||||
tracing::info!("Connecting to MongoDB...");
|
||||
let db = database::Database::connect(&config.mongodb_uri, &config.mongodb_database).await?;
|
||||
db.ensure_indexes().await?;
|
||||
|
||||
// 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.
|
||||
// Per-tenant pool only — the agent has no shared "default" database
|
||||
// after M7.2-D. `mongodb_database` is now the db-name prefix used
|
||||
// for tenant databases (`<prefix>_<tenant_id>`).
|
||||
let db_pool =
|
||||
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...");
|
||||
let scheduler_agent = agent.clone();
|
||||
|
||||
@@ -4,8 +4,14 @@ use tokio_cron_scheduler::{Job, JobScheduler};
|
||||
use compliance_core::models::ScanTrigger;
|
||||
|
||||
use crate::agent::ComplianceAgent;
|
||||
use crate::database::Database;
|
||||
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> {
|
||||
let sched = JobScheduler::new()
|
||||
.await
|
||||
@@ -18,7 +24,9 @@ pub async fn start_scheduler(agent: &ComplianceAgent) -> Result<(), AgentError>
|
||||
let agent = scan_agent.clone();
|
||||
Box::pin(async move {
|
||||
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}")))?;
|
||||
@@ -34,7 +42,9 @@ pub async fn start_scheduler(agent: &ComplianceAgent) -> Result<(), AgentError>
|
||||
let agent = cve_agent.clone();
|
||||
Box::pin(async move {
|
||||
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}")))?;
|
||||
@@ -48,8 +58,9 @@ pub async fn start_scheduler(agent: &ComplianceAgent) -> Result<(), AgentError>
|
||||
.await
|
||||
.map_err(|e| AgentError::Scheduler(format!("Failed to start scheduler: {e}")))?;
|
||||
|
||||
let tenants = scheduler_tenants();
|
||||
tracing::info!(
|
||||
"Scheduler started: scans='{}', CVE monitor='{}'",
|
||||
"Scheduler started: scans='{}', CVE monitor='{}', tenants={tenants:?}",
|
||||
agent.config.scan_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;
|
||||
|
||||
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,
|
||||
Err(e) => {
|
||||
tracing::error!("Failed to list repos for scheduled scan: {e}");
|
||||
tracing::error!("Failed to list repos for tenant '{tenant_id}': {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
@@ -75,33 +120,44 @@ async fn scan_all_repos(agent: &ComplianceAgent) {
|
||||
|
||||
for repo in repos {
|
||||
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 {
|
||||
tracing::error!("Scheduled scan failed for {}: {e}", repo.name);
|
||||
if let Err(e) = agent
|
||||
.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::SbomEntry;
|
||||
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
|
||||
let cursor = match agent.db.sbom_entries().find(doc! {}).await {
|
||||
let cursor = match db.sbom_entries().find(doc! {}).await {
|
||||
Ok(c) => c,
|
||||
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;
|
||||
}
|
||||
};
|
||||
let entries: Vec<SbomEntry> = cursor.filter_map(|r| async { r.ok() }).collect().await;
|
||||
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;
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
"CVE monitor: checking {} dependencies for new CVEs",
|
||||
"CVE monitor: checking {} dependencies for new CVEs (tenant '{tenant_id}')",
|
||||
entries.len()
|
||||
);
|
||||
|
||||
@@ -112,7 +168,7 @@ async fn monitor_cves(agent: &ComplianceAgent) {
|
||||
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)) = 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());
|
||||
}
|
||||
}
|
||||
@@ -160,8 +216,7 @@ async fn monitor_cves(agent: &ComplianceAgent) {
|
||||
for alert in &alerts {
|
||||
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 _ = agent
|
||||
.db
|
||||
let _ = db
|
||||
.cve_alerts()
|
||||
.update_one(filter, update)
|
||||
.upsert(true)
|
||||
@@ -174,8 +229,7 @@ async fn monitor_cves(agent: &ComplianceAgent) {
|
||||
continue;
|
||||
}
|
||||
if let Some(entry_id) = &entry.id {
|
||||
let _ = agent
|
||||
.db
|
||||
let _ = db
|
||||
.sbom_entries()
|
||||
.update_one(
|
||||
doc! { "_id": entry_id },
|
||||
@@ -213,8 +267,7 @@ async fn monitor_cves(agent: &ComplianceAgent) {
|
||||
let update = doc! {
|
||||
"$setOnInsert": mongodb::bson::to_bson(¬ification).unwrap_or_default()
|
||||
};
|
||||
match agent
|
||||
.db
|
||||
match db
|
||||
.cve_notifications()
|
||||
.update_one(filter, update)
|
||||
.upsert(true)
|
||||
@@ -232,8 +285,10 @@ async fn monitor_cves(agent: &ComplianceAgent) {
|
||||
}
|
||||
|
||||
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 {
|
||||
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(
|
||||
Extension(agent): Extension<Arc<ComplianceAgent>>,
|
||||
Path(repo_id): Path<String>,
|
||||
Path((tenant_id, repo_id)): Path<(String, String)>,
|
||||
headers: HeaderMap,
|
||||
body: Bytes,
|
||||
) -> 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) {
|
||||
Ok(oid) => oid,
|
||||
Err(_) => return StatusCode::NOT_FOUND,
|
||||
};
|
||||
let repo = match agent
|
||||
.db
|
||||
let db = match agent.db_pool.for_tenant_id(&tenant_id).await {
|
||||
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()
|
||||
.find_one(mongodb::bson::doc! { "_id": oid })
|
||||
.await
|
||||
{
|
||||
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;
|
||||
}
|
||||
};
|
||||
@@ -66,15 +72,21 @@ pub async fn handle_gitea_webhook(
|
||||
"push" => {
|
||||
let agent_clone = (*agent).clone();
|
||||
let repo_id = repo_id.clone();
|
||||
let tenant_id = tenant_id.clone();
|
||||
tokio::spawn(async move {
|
||||
tracing::info!("Gitea push webhook: triggering scan for {repo_id}");
|
||||
if let Err(e) = agent_clone.run_scan(&repo_id, ScanTrigger::Webhook).await {
|
||||
tracing::info!(
|
||||
"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}");
|
||||
}
|
||||
});
|
||||
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}'");
|
||||
StatusCode::OK
|
||||
@@ -84,6 +96,7 @@ pub async fn handle_gitea_webhook(
|
||||
|
||||
async fn handle_pull_request(
|
||||
agent: Arc<ComplianceAgent>,
|
||||
tenant_id: &str,
|
||||
repo_id: &str,
|
||||
payload: &serde_json::Value,
|
||||
) -> StatusCode {
|
||||
@@ -106,13 +119,14 @@ async fn handle_pull_request(
|
||||
}
|
||||
|
||||
let repo_id = repo_id.to_string();
|
||||
let tenant_id = tenant_id.to_string();
|
||||
let head_sha = head_sha.to_string();
|
||||
let base_sha = base_sha.to_string();
|
||||
let agent_clone = (*agent).clone();
|
||||
tokio::spawn(async move {
|
||||
tracing::info!("Gitea PR webhook: reviewing PR #{pr_number} on {repo_id}");
|
||||
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
|
||||
{
|
||||
tracing::error!("PR review failed for #{pr_number}: {e}");
|
||||
|
||||
@@ -14,24 +14,30 @@ type HmacSha256 = Hmac<Sha256>;
|
||||
|
||||
pub async fn handle_github_webhook(
|
||||
Extension(agent): Extension<Arc<ComplianceAgent>>,
|
||||
Path(repo_id): Path<String>,
|
||||
Path((tenant_id, repo_id)): Path<(String, String)>,
|
||||
headers: HeaderMap,
|
||||
body: Bytes,
|
||||
) -> 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) {
|
||||
Ok(oid) => oid,
|
||||
Err(_) => return StatusCode::NOT_FOUND,
|
||||
};
|
||||
let repo = match agent
|
||||
.db
|
||||
let db = match agent.db_pool.for_tenant_id(&tenant_id).await {
|
||||
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()
|
||||
.find_one(mongodb::bson::doc! { "_id": oid })
|
||||
.await
|
||||
{
|
||||
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;
|
||||
}
|
||||
};
|
||||
@@ -66,15 +72,21 @@ pub async fn handle_github_webhook(
|
||||
"push" => {
|
||||
let agent_clone = (*agent).clone();
|
||||
let repo_id = repo_id.clone();
|
||||
let tenant_id = tenant_id.clone();
|
||||
tokio::spawn(async move {
|
||||
tracing::info!("GitHub push webhook: triggering scan for {repo_id}");
|
||||
if let Err(e) = agent_clone.run_scan(&repo_id, ScanTrigger::Webhook).await {
|
||||
tracing::info!(
|
||||
"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}");
|
||||
}
|
||||
});
|
||||
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}'");
|
||||
StatusCode::OK
|
||||
@@ -84,6 +96,7 @@ pub async fn handle_github_webhook(
|
||||
|
||||
async fn handle_pull_request(
|
||||
agent: Arc<ComplianceAgent>,
|
||||
tenant_id: &str,
|
||||
repo_id: &str,
|
||||
payload: &serde_json::Value,
|
||||
) -> StatusCode {
|
||||
@@ -105,13 +118,14 @@ async fn handle_pull_request(
|
||||
}
|
||||
|
||||
let repo_id = repo_id.to_string();
|
||||
let tenant_id = tenant_id.to_string();
|
||||
let head_sha = head_sha.to_string();
|
||||
let base_sha = base_sha.to_string();
|
||||
let agent_clone = (*agent).clone();
|
||||
tokio::spawn(async move {
|
||||
tracing::info!("GitHub PR webhook: reviewing PR #{pr_number} on {repo_id}");
|
||||
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
|
||||
{
|
||||
tracing::error!("PR review failed for #{pr_number}: {e}");
|
||||
|
||||
@@ -10,24 +10,30 @@ use crate::agent::ComplianceAgent;
|
||||
|
||||
pub async fn handle_gitlab_webhook(
|
||||
Extension(agent): Extension<Arc<ComplianceAgent>>,
|
||||
Path(repo_id): Path<String>,
|
||||
Path((tenant_id, repo_id)): Path<(String, String)>,
|
||||
headers: HeaderMap,
|
||||
body: Bytes,
|
||||
) -> 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) {
|
||||
Ok(oid) => oid,
|
||||
Err(_) => return StatusCode::NOT_FOUND,
|
||||
};
|
||||
let repo = match agent
|
||||
.db
|
||||
let db = match agent.db_pool.for_tenant_id(&tenant_id).await {
|
||||
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()
|
||||
.find_one(mongodb::bson::doc! { "_id": oid })
|
||||
.await
|
||||
{
|
||||
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;
|
||||
}
|
||||
};
|
||||
@@ -59,15 +65,21 @@ pub async fn handle_gitlab_webhook(
|
||||
"push" => {
|
||||
let agent_clone = (*agent).clone();
|
||||
let repo_id = repo_id.clone();
|
||||
let tenant_id = tenant_id.clone();
|
||||
tokio::spawn(async move {
|
||||
tracing::info!("GitLab push webhook: triggering scan for {repo_id}");
|
||||
if let Err(e) = agent_clone.run_scan(&repo_id, ScanTrigger::Webhook).await {
|
||||
tracing::info!(
|
||||
"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}");
|
||||
}
|
||||
});
|
||||
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}'");
|
||||
StatusCode::OK
|
||||
@@ -77,6 +89,7 @@ pub async fn handle_gitlab_webhook(
|
||||
|
||||
async fn handle_merge_request(
|
||||
agent: Arc<ComplianceAgent>,
|
||||
tenant_id: &str,
|
||||
repo_id: &str,
|
||||
payload: &serde_json::Value,
|
||||
) -> StatusCode {
|
||||
@@ -101,13 +114,14 @@ async fn handle_merge_request(
|
||||
}
|
||||
|
||||
let repo_id = repo_id.to_string();
|
||||
let tenant_id = tenant_id.to_string();
|
||||
let head_sha = head_sha.to_string();
|
||||
let base_sha = base_sha.to_string();
|
||||
let agent_clone = (*agent).clone();
|
||||
tokio::spawn(async move {
|
||||
tracing::info!("GitLab MR webhook: reviewing MR !{mr_iid} on {repo_id}");
|
||||
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
|
||||
{
|
||||
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> {
|
||||
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(
|
||||
"/webhook/github/{repo_id}",
|
||||
"/webhook/{tenant_id}/github/{repo_id}",
|
||||
post(github::handle_github_webhook),
|
||||
)
|
||||
.route(
|
||||
"/webhook/gitlab/{repo_id}",
|
||||
"/webhook/{tenant_id}/gitlab/{repo_id}",
|
||||
post(gitlab::handle_gitlab_webhook),
|
||||
)
|
||||
.route(
|
||||
"/webhook/gitea/{repo_id}",
|
||||
"/webhook/{tenant_id}/gitea/{repo_id}",
|
||||
post(gitea::handle_gitea_webhook),
|
||||
)
|
||||
.layer(Extension(Arc::new(agent.clone())));
|
||||
|
||||
@@ -7,7 +7,7 @@ use std::sync::Arc;
|
||||
|
||||
use compliance_agent::agent::ComplianceAgent;
|
||||
use compliance_agent::api;
|
||||
use compliance_agent::database::{Database, DatabasePool};
|
||||
use compliance_agent::database::DatabasePool;
|
||||
use compliance_core::AgentConfig;
|
||||
use secrecy::SecretString;
|
||||
|
||||
@@ -28,11 +28,6 @@ impl TestServer {
|
||||
// Unique database name per test run to avoid collisions
|
||||
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)
|
||||
.await
|
||||
.expect("Failed to build DatabasePool");
|
||||
@@ -73,7 +68,7 @@ impl TestServer {
|
||||
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
|
||||
// handler takes a TenantCtx extractor; without KC in the test
|
||||
@@ -164,12 +159,11 @@ impl TestServer {
|
||||
&self.db_name
|
||||
}
|
||||
|
||||
/// Drop the test database on cleanup. Post-M7.2-B the actual data
|
||||
/// lives in `<db_name>_<tenant>` per-tenant databases; list those
|
||||
/// off the cluster and drop them too.
|
||||
/// Drop every per-tenant database belonging to this test run.
|
||||
/// Post-M7.2-D the agent never opens a `db_name` directly —
|
||||
/// data lives only in `<db_name>_<tenant>` per-tenant databases.
|
||||
pub async fn cleanup(&self) {
|
||||
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 {
|
||||
let prefix = format!("{}_", self.db_name);
|
||||
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]
|
||||
async fn tenant_db_name_falls_back_to_hash_when_too_long() {
|
||||
let uri = std::env::var("TEST_MONGODB_URI")
|
||||
|
||||
Reference in New Issue
Block a user