diff --git a/README.md b/README.md index 6d9ef20..2e06efb 100644 --- a/README.md +++ b/README.md @@ -170,6 +170,7 @@ duplicate comments. | Worker container isolation | Implemented | `worker.backend = "container"` runs each attempt in a fresh hardened container (read-only rootfs, cap-drop ALL, cpu/memory/pids/tmpfs/network limits, env-only token injection, kill-by-name on timeout); hosted posture refuses host execution without explicit opt-in — see [docs/container-isolation.md](docs/container-isolation.md). | | Pull request creation | Partial | Opens draft PRs from session results against the repository's resolved default/base branch. | | CovenCave task polling | Implemented | Task API served from the durable store, survives restarts, and is gated by the tenant boundary — `token` mode fails closed, tenant tokens are installation-scoped, and every read is audited (see [docs/security.md](docs/security.md)). | +| Cave oversight dashboard | Implemented | Tenant-scoped data behind the four Cave views: task history (`/api/github/tasks`), usage (`/api/github/usage`), familiar routing (`/api/github/routing`), and a task-lifecycle audit stream (`/api/github/audit`) — acceptance, execution/retries/timeout, and PR creation. | | Durable queue / task store | Implemented | Deliveries deduplicated by `X-GitHub-Delivery` before GitHub hears success; the SQLite `tasks` table is the queue (atomic claims, no drop path) and interrupted work is requeued at startup ([design](docs/durable-task-store.md)). | | Hosted tier | Planned | See [Hosted vs self-hosted](docs/hosted-vs-self-hosted.md). | | Familiar trust contract | Planned | See [Familiar Contract](FAMILIAR-CONTRACT.md). | diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index da39cc0..2847201 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -100,6 +100,25 @@ pub struct RepoRoutingOverride { pub reviews: Option, } +/// A serializable installation routing policy for the Cave dashboard (#18). +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct RoutingView { + pub installation_id: u64, + pub account: Option, + /// Familiar ids this installation may route to (resolved allow-list, or all + /// configured familiars). + pub familiars: Vec, + /// Installation-level trigger lanes. + pub triggers: TriggerPolicy, + /// Per-repo trigger overrides. + pub repos: std::collections::HashMap, + /// `true` when the installation is explicitly configured; `false` means + /// open routing (nothing configured) or fail-closed (unlisted) — read + /// `familiars` to tell which. + pub listed: bool, +} + /// The effective routing view for one delivery: which familiars are visible /// and which trigger lanes are open (issue #7). pub struct RoutingScope<'a> { @@ -542,6 +561,60 @@ impl Config { RoutingScope { familiars, triggers } } + /// The routing policy view for one installation, for the Cave dashboard + /// (issue #18). Resolves familiars and triggers the same way [`scope_for`] + /// does, but at the installation level (not one delivery), and includes the + /// per-repo overrides so the dashboard can render them. + /// + /// [`scope_for`]: Config::scope_for + pub fn routing_view(&self, installation_id: u64) -> RoutingView { + let all_familiars = || self.familiars.iter().map(|f| f.id.clone()).collect::>(); + // Open routing (no installations configured): every familiar, all lanes. + if self.installations.is_empty() { + return RoutingView { + installation_id, + account: None, + familiars: all_familiars(), + triggers: TriggerPolicy::default(), + repos: std::collections::HashMap::new(), + listed: false, + }; + } + // Configured but unlisted: fail closed — routes nothing. + let Some(installation) = self.installations.iter().find(|i| i.id == installation_id) else { + return RoutingView { + installation_id, + account: None, + familiars: Vec::new(), + triggers: TriggerPolicy { + assignment: false, + labels: false, + commands: false, + reviews: false, + }, + repos: std::collections::HashMap::new(), + listed: false, + }; + }; + let familiars = if installation.familiars.is_empty() { + all_familiars() + } else { + self.familiars + .iter() + .map(|f| f.id.clone()) + .filter(|id| installation.familiars.contains(id)) + .collect() + }; + RoutingView { + installation_id, + account: installation.account.clone(), + familiars, + triggers: installation.triggers, + repos: installation.repos.clone(), + listed: true, + } + } + /// Usage limits for one installation (issue #15). Unlimited when the /// installation is not configured. pub fn limits_for(&self, installation_id: u64) -> InstallationLimits { @@ -1174,6 +1247,75 @@ mod tests { .collect() } + fn routing_config(familiars: Vec) -> Config { + let dir = tmpdir(); + config_with( + GitHubAppConfig { + app_id: 1, + private_key_path: write_pem(&dir), + webhook_secret: "a-long-random-webhook-secret".into(), + api_base_url: None, + }, + WorkerConfig { + concurrency: 1, + coven_code_bin: write_bin(&dir), + workspace_root: dir.clone(), + timeout_secs: 600, + max_retries: 2, + backend: WorkerBackendKind::Host, + container: ContainerConfig::default(), + allow_host_backend: true, + }, + familiars, + ) + } + + fn fam(id: &str) -> FamiliarConfig { + FamiliarConfig { + id: id.into(), + display_name: id.into(), + bot_username: format!("coven-{id}[bot]"), + model: None, + skills: vec![], + trigger_labels: vec![], + } + } + + #[test] + fn routing_view_reflects_open_listed_and_fail_closed() { + let mut cfg = routing_config(vec![fam("cody"), fam("nova")]); + + // Open routing (no installations): every familiar, all lanes, unlisted. + let open = cfg.routing_view(123); + assert_eq!(open.familiars, vec!["cody", "nova"]); + assert!(open.triggers.reviews); + assert!(!open.listed); + + // Listed with an allow-list and a disabled lane. + cfg.installations = vec![InstallationConfig { + id: 123, + account: Some("acme".into()), + familiars: vec!["cody".into()], + triggers: TriggerPolicy { + reviews: false, + ..Default::default() + }, + limits: InstallationLimits::default(), + repos: std::collections::HashMap::new(), + }]; + let listed = cfg.routing_view(123); + assert_eq!(listed.familiars, vec!["cody"]); + assert_eq!(listed.account.as_deref(), Some("acme")); + assert!(!listed.triggers.reviews); + assert!(listed.listed); + + // Configured but unlisted → fail closed. + let closed = cfg.routing_view(999); + assert!(closed.familiars.is_empty()); + assert!(!closed.triggers.assignment); + assert!(!closed.listed); + } + #[test] fn review_policy_resolves_repo_overrides() { let policy = ReviewConfig { diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 348be73..edfd433 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -12,7 +12,7 @@ use tracing_subscriber::EnvFilter; use coven_github_config::Config; use coven_github_webhook::routes::{ - handle_webhook, list_memory, list_tasks, revoke_memory, usage, AppState, + audit, handle_webhook, list_memory, list_tasks, revoke_memory, routing, usage, AppState, }; use coven_github_worker as worker; @@ -165,6 +165,8 @@ async fn main() -> Result<()> { .route("/api/github/memory", get(list_memory)) .route("/api/github/memory/revoke", post(revoke_memory)) .route("/api/github/usage", get(usage)) + .route("/api/github/audit", get(audit)) + .route("/api/github/routing", get(routing)) .with_state(state) .layer(TraceLayer::new_for_http()); diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 3878191..ab7c85f 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -480,6 +480,19 @@ pub struct UsageRow { pub runtime_secs: u64, } +/// One task-lifecycle audit event for the Cave dashboard (issue #18). +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AuditEvent { + /// RFC 3339 timestamp. + pub at: String, + pub task_id: Option, + pub repo: Option, + /// `ignored:` | `attempt:` | `pr_opened`. + pub kind: String, + pub detail: Option, +} + impl Store { /// Appends a task-API read to the audit trail (issue #3). pub async fn record_api_read( @@ -509,6 +522,92 @@ impl Store { .expect("store task panicked") } + /// Assembles a task-lifecycle audit stream for the Cave dashboard (issue + /// #18), scoped to the caller's installation and capped at `limit`, newest + /// first. Draws from three durable sources: ignored deliveries (acceptance + /// rejections), task attempts (execution / retries / timeout / failure), + /// and opened PRs (publication). + pub async fn audit_events(&self, scope: Option, limit: u32) -> Result> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + let inst = scope.as_ref().map(|s| s.installation_id); + let mut events: Vec = Vec::new(); + + // Ignored deliveries — the adapter received but did not act. + let mut stmt = conn.prepare( + "SELECT received_at, repo, routing FROM webhook_deliveries + WHERE routing LIKE 'ignored:%' AND (?1 IS NULL OR installation_id = ?1)", + )?; + for row in stmt.query_map(params![inst], |r| { + Ok(AuditEvent { + at: r.get(0)?, + task_id: None, + repo: r.get::<_, Option>(1)?, + kind: r.get::<_, String>(2)?, + detail: None, + }) + })? { + events.push(row?); + } + + // Task attempts — one per execution, carrying retries and outcomes. + let mut stmt = conn.prepare( + "SELECT COALESCE(a.ended_at, a.started_at), a.task_id, t.repo, a.attempt, + a.outcome, a.detail + FROM task_attempts a JOIN tasks t ON a.task_id = t.id + WHERE (?1 IS NULL OR t.installation_id = ?1)", + )?; + for row in stmt.query_map(params![inst], |r| { + let attempt: i64 = r.get(3)?; + let outcome: Option = r.get(4)?; + let detail: Option = r.get(5)?; + Ok(AuditEvent { + at: r.get(0)?, + task_id: Some(r.get::<_, String>(1)?), + repo: Some(r.get::<_, String>(2)?), + kind: format!("attempt:{}", outcome.as_deref().unwrap_or("running")), + detail: detail.map(|d| format!("attempt {attempt}: {d}")), + }) + })? { + events.push(row?); + } + + // Opened pull requests — the publication events. + let mut stmt = conn.prepare( + "SELECT updated_at, id, repo, pr_number FROM tasks + WHERE pr_number IS NOT NULL AND (?1 IS NULL OR installation_id = ?1)", + )?; + for row in stmt.query_map(params![inst], |r| { + Ok(AuditEvent { + at: r.get(0)?, + task_id: Some(r.get::<_, String>(1)?), + repo: Some(r.get::<_, String>(2)?), + kind: "pr_opened".to_string(), + detail: Some(format!("PR #{}", r.get::<_, u64>(3)?)), + }) + })? { + events.push(row?); + } + + // Newest first, honor an optional repo narrowing, then cap. + events.sort_by(|a, b| b.at.cmp(&a.at)); + let repos = scope.as_ref().and_then(|s| s.repos.as_ref()); + let filtered = events + .into_iter() + .filter(|e| match (repos, &e.repo) { + (Some(allowed), Some(repo)) => allowed.contains(repo), + (Some(_), None) => false, + (None, _) => true, + }) + .take(limit as usize) + .collect(); + Ok(filtered) + }) + .await + .expect("store task panicked") + } + /// `(caller, scope, action, result)` audit rows, oldest first. pub async fn api_audit_entries(&self) -> Result> { let conn = self.conn.clone(); @@ -1613,6 +1712,49 @@ mod queue_tests { assert!(text.iter().any(|t| t.starts_with("task:"))); } + #[tokio::test] + async fn audit_events_assemble_lifecycle_scoped_to_installation() { + let store = Store::open_in_memory().expect("open"); + // A finished task that opened a PR (attempt + pr_opened events). + enqueue(&store, "d1", &fix_task("t1")).await; + store.claim_next(&Default::default()).await.unwrap(); + store + .finish( + "t1", + Terminal { + state: TerminalState::Completed, + result_status: Some("success".to_string()), + branch: Some("cody/fix".to_string()), + pr_number: Some(9), + summary: Some("done".to_string()), + detail: Some("attempt ok".to_string()), + }, + ) + .await + .expect("finish"); + // An ignored delivery (acceptance rejection). + store + .record_delivery(delivery("d-ig"), Routing::Ignored("unsupported")) + .await + .unwrap(); + + let events = store + .audit_events(Some(ApiScope { installation_id: 1, repos: None }), 100) + .await + .unwrap(); + let kinds: Vec<&str> = events.iter().map(|e| e.kind.as_str()).collect(); + assert!(kinds.contains(&"pr_opened"), "kinds: {kinds:?}"); + assert!(kinds.iter().any(|k| k.starts_with("attempt:")), "kinds: {kinds:?}"); + assert!(kinds.contains(&"ignored:unsupported"), "kinds: {kinds:?}"); + + // A different installation sees none of it. + let other = store + .audit_events(Some(ApiScope { installation_id: 999, repos: None }), 100) + .await + .unwrap(); + assert!(other.is_empty()); + } + #[tokio::test] async fn finish_reaches_terminal_state_and_closes_the_attempt() { let store = Store::open_in_memory().expect("open"); diff --git a/crates/webhook/src/routes.rs b/crates/webhook/src/routes.rs index f15d203..94c4e77 100644 --- a/crates/webhook/src/routes.rs +++ b/crates/webhook/src/routes.rs @@ -2,7 +2,7 @@ use axum::{ body::Bytes, - extract::State, + extract::{Query, State}, http::{HeaderMap, StatusCode}, response::IntoResponse, Json, @@ -349,6 +349,120 @@ pub async fn usage(State(state): State, headers: HeaderMap) -> impl In } } +/// Number of audit events one dashboard request returns. +const AUDIT_LIMIT: u32 = 200; + +/// GET /api/github/audit — task-lifecycle audit stream for the Cave dashboard +/// (issue #18), tenant-scoped and audited, same contract as `list_tasks`. +pub async fn audit(State(state): State, headers: HeaderMap) -> impl IntoResponse { + let action = "audit"; + let (caller, scope) = match authorize_api(&state.config.api, &headers) { + ApiCaller::Denied => { + if let Err(e) = state + .store + .record_api_read("anonymous", "none", action, "denied") + .await + { + error!("api audit write failed: {e:#}"); + } + return ( + StatusCode::UNAUTHORIZED, + Json(json!({"ok": false, "error": "unauthorized"})), + ) + .into_response(); + } + ApiCaller::Open => ("open".to_string(), None), + ApiCaller::Service => ("service".to_string(), None), + ApiCaller::Tenant(tenant) => ( + format!("tenant:{}", tenant.installation_id), + Some(ApiScope { + installation_id: tenant.installation_id, + repos: if tenant.repos.is_empty() { + None + } else { + Some(tenant.repos.clone()) + }, + }), + ), + }; + let scope_label = scope + .as_ref() + .map(|s| format!("installation:{}", s.installation_id)) + .unwrap_or_else(|| "all".to_string()); + + match state.store.audit_events(scope, AUDIT_LIMIT).await { + Ok(events) => { + if let Err(e) = state + .store + .record_api_read(&caller, &scope_label, action, &format!("ok:{}", events.len())) + .await + { + error!("api audit write failed: {e:#}"); + } + Json(json!({ "ok": true, "events": events })).into_response() + } + Err(e) => { + error!("audit stream unavailable: {e:#}"); + let _ = state + .store + .record_api_read(&caller, &scope_label, action, "error") + .await; + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"ok": false, "error": "audit unavailable"})), + ) + .into_response() + } + } +} + +/// GET /api/github/routing[?installation_id=N] — the effective routing policy +/// for the Cave dashboard (issue #18). A tenant sees its own installation; the +/// service/open caller names the installation via the query string. +pub async fn routing( + State(state): State, + headers: HeaderMap, + Query(params): Query>, +) -> impl IntoResponse { + let action = "routing"; + let (caller, installation_id) = match authorize_api(&state.config.api, &headers) { + ApiCaller::Denied => { + let _ = state + .store + .record_api_read("anonymous", "none", action, "denied") + .await; + return ( + StatusCode::UNAUTHORIZED, + Json(json!({"ok": false, "error": "unauthorized"})), + ) + .into_response(); + } + ApiCaller::Tenant(tenant) => ( + format!("tenant:{}", tenant.installation_id), + tenant.installation_id, + ), + ApiCaller::Service | ApiCaller::Open => { + match params.get("installation_id").and_then(|s| s.parse::().ok()) { + Some(id) => ("service".to_string(), id), + None => { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"ok": false, "error": "installation_id required"})), + ) + .into_response() + } + } + } + }; + + let view = state.config.routing_view(installation_id); + let _ = state + .store + .record_api_read(&caller, &installation_id.to_string(), action, "ok") + .await; + Json(json!({ "ok": true, "routing": view })).into_response() +} + /// Resolved identity of a task-API caller. enum ApiCaller<'a> { /// `open` mode: unauthenticated local development. @@ -2011,6 +2125,105 @@ mod tenancy_tests { (status, serde_json::from_slice(&bytes).expect("json")) } + async fn call_audit(state: &AppState, bearer: Option<&str>) -> (StatusCode, serde_json::Value) { + let mut headers = HeaderMap::new(); + if let Some(token) = bearer { + headers.insert("authorization", format!("Bearer {token}").parse().unwrap()); + } + let response = audit(State(state.clone()), headers).await.into_response(); + let status = response.status(); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + (status, serde_json::from_slice(&bytes).unwrap()) + } + + async fn call_routing( + state: &AppState, + bearer: Option<&str>, + installation_id: Option<&str>, + ) -> (StatusCode, serde_json::Value) { + let mut headers = HeaderMap::new(); + if let Some(token) = bearer { + headers.insert("authorization", format!("Bearer {token}").parse().unwrap()); + } + let mut params = std::collections::HashMap::new(); + if let Some(id) = installation_id { + params.insert("installation_id".to_string(), id.to_string()); + } + let response = routing(State(state.clone()), headers, Query(params)) + .await + .into_response(); + let status = response.status(); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + (status, serde_json::from_slice(&bytes).unwrap()) + } + + async fn seed_ignored(state: &AppState, installation_id: u64) { + state + .store + .record_delivery( + coven_github_store::Delivery { + delivery_id: format!("ig-{installation_id}"), + event: "issues".to_string(), + action: Some("closed".to_string()), + installation_id: Some(installation_id), + repo: Some("OpenCoven/demo".to_string()), + payload_hash: "h".to_string(), + }, + coven_github_store::Routing::Ignored("unsupported"), + ) + .await + .expect("seed ignored delivery"); + } + + #[tokio::test] + async fn audit_is_tenant_scoped_and_fails_closed() { + let state = token_state(two_tenant_api()); + seed_ignored(&state, 1).await; + seed_ignored(&state, 2).await; + + // No token in token mode → fail closed. + let (status, json) = call_audit(&state, None).await; + assert_eq!(status, StatusCode::UNAUTHORIZED); + assert!(json.get("events").is_none()); + + // Installation 1's token sees only its own event. + let (status, json) = call_audit(&state, Some("tenant-one-0123456789abcdef")).await; + assert_eq!(status, StatusCode::OK); + let events = json["events"].as_array().unwrap(); + assert_eq!(events.len(), 1); + assert_eq!(events[0]["kind"], "ignored:unsupported"); + + // Service sees both installations. + let (_, json) = call_audit(&state, Some("service-token-0123456789abcdef")).await; + assert_eq!(json["events"].as_array().unwrap().len(), 2); + } + + #[tokio::test] + async fn routing_view_is_tenant_scoped_and_fails_closed() { + let state = token_state(two_tenant_api()); + + // Denied without a token. + let (status, _) = call_routing(&state, None, None).await; + assert_eq!(status, StatusCode::UNAUTHORIZED); + + // A tenant sees its own installation's routing view. + let (status, json) = call_routing(&state, Some("tenant-one-0123456789abcdef"), None).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(json["routing"]["installationId"], 1); + + // Service must name the installation. + let (status, _) = call_routing(&state, Some("service-token-0123456789abcdef"), None).await; + assert_eq!(status, StatusCode::BAD_REQUEST); + let (status, json) = + call_routing(&state, Some("service-token-0123456789abcdef"), Some("42")).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(json["routing"]["installationId"], 42); + } + async fn seed_memory(state: &AppState, installation_id: u64, repo: &str) { state .store