diff --git a/backend/crates/api/src/lib.rs b/backend/crates/api/src/lib.rs index 700cf01..ac8d0a5 100644 --- a/backend/crates/api/src/lib.rs +++ b/backend/crates/api/src/lib.rs @@ -161,6 +161,10 @@ impl AppState { } pub async fn bootstrap(&self) -> anyhow::Result<()> { + let stale = self.jobs.recover().await?; + if stale > 0 { + tracing::warn!("marked {stale} interrupted job run(s) as failed"); + } if let Some((email, password)) = &self.cfg.bootstrap_admin { if self.users.bootstrap_admin(email, password).await? { tracing::info!("created bootstrap admin {email}"); diff --git a/backend/crates/application/src/jobs.rs b/backend/crates/application/src/jobs.rs index e5a0bc1..3da736b 100644 --- a/backend/crates/application/src/jobs.rs +++ b/backend/crates/application/src/jobs.rs @@ -94,22 +94,41 @@ impl JobRunner { } async fn execute(runs: Arc, handler: Arc, run: &JobRun) { - let log = RepoLog { + let log = Arc::new(RepoLog { runs: runs.clone(), id: run.id, - }; - let status = match handler.run(run.params.clone(), &log).await { - Ok(()) => JobStatus::Success, - Err(e) => { + }); + // Run the handler in its own task so a panic is contained and reported. + let (params, task_log) = (run.params.clone(), log.clone()); + let joined = tokio::spawn(async move { handler.run(params, &*task_log).await }).await; + let status = match joined { + Ok(Ok(())) => JobStatus::Success, + Ok(Err(e)) => { log.line(&format!("ERROR: {e}")).await; JobStatus::Failed } + Err(e) => { + log.line(&format!("ERROR: job panicked: {e}")).await; + JobStatus::Failed + } }; if let Err(e) = runs.finish(run.id, status).await { tracing_line(&format!("failed to finish job: {e}")); } } + /// Mark runs left in `running` state by a previous process as failed. Returns the count. + pub async fn recover(&self) -> Result { + let stale = self.runs.running().await?; + for r in &stale { + self.runs + .append_log(r.id, "ERROR: interrupted by a restart of the service") + .await?; + self.runs.finish(r.id, JobStatus::Failed).await?; + } + Ok(stale.len()) + } + /// Start a job in the background. Fails with `Conflict` if the kind is already running. pub async fn start( &self, diff --git a/backend/crates/application/src/test_fakes.rs b/backend/crates/application/src/test_fakes.rs index f7193fa..d8e2c4e 100644 --- a/backend/crates/application/src/test_fakes.rs +++ b/backend/crates/application/src/test_fakes.rs @@ -292,6 +292,16 @@ impl JobRunRepository for MemJobRuns { .find(|r| r.kind == kind && r.status != JobStatus::Running) .cloned()) } + async fn running(&self) -> Result, DomainError> { + Ok(self + .0 + .lock() + .unwrap() + .iter() + .filter(|r| r.status == JobStatus::Running) + .cloned() + .collect()) + } } pub fn smtp() -> SmtpSettings { diff --git a/backend/crates/application/src/tests/jobs_tests.rs b/backend/crates/application/src/tests/jobs_tests.rs index 8015b62..e0b1e24 100644 --- a/backend/crates/application/src/tests/jobs_tests.rs +++ b/backend/crates/application/src/tests/jobs_tests.rs @@ -7,6 +7,7 @@ use uuid::Uuid; use crate::jobs::{JobHandler, JobLog, JobRunner}; use crate::test_fakes::MemJobRuns; +use domain::ports::JobRunRepository; struct Echo; #[async_trait] @@ -102,3 +103,48 @@ async fn list_returns_newest_first_with_limit() { assert_eq!(list.len(), 2); assert!(list[0].started_at >= list[1].started_at); } + +struct Panics; +#[async_trait] +impl JobHandler for Panics { + async fn run(&self, _: Option, log: &dyn JobLog) -> Result<(), String> { + log.line("about to panic").await; + panic!("handler bug"); + } +} + +#[tokio::test] +async fn panicking_handler_marks_run_failed() { + let runs = Arc::new(MemJobRuns::default()); + let r = JobRunner::new(runs.clone()).register(JobKind::Backup, Arc::new(Panics)); + let run = r.start(JobKind::Backup, None, "test").await.unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let run = r.get(run.id).await.unwrap(); + assert_eq!(run.status, JobStatus::Failed); + assert!(run.log.contains("panicked"), "{}", run.log); + // the kind is free again + assert!(r.start(JobKind::Backup, None, "test").await.is_ok()); +} + +#[tokio::test] +async fn stale_running_runs_are_failed_on_recovery() { + let runs = Arc::new(MemJobRuns::default()); + runs.insert(&domain::jobs::JobRun { + id: Uuid::new_v4(), + kind: JobKind::PackageRefresh, + params: None, + status: JobStatus::Running, + started_at: chrono::Utc::now(), + finished_at: None, + log: String::new(), + triggered_by: "old process".into(), + }) + .await + .unwrap(); + let r = JobRunner::new(runs.clone()).register(JobKind::PackageRefresh, Arc::new(Echo)); + assert_eq!(r.recover().await.unwrap(), 1); + let list = r.list(10).await.unwrap(); + assert_eq!(list[0].status, JobStatus::Failed); + assert!(list[0].log.contains("interrupted")); + assert!(r.start(JobKind::PackageRefresh, None, "test").await.is_ok()); +} diff --git a/backend/crates/domain/src/ports.rs b/backend/crates/domain/src/ports.rs index 9ed16b1..842984a 100644 --- a/backend/crates/domain/src/ports.rs +++ b/backend/crates/domain/src/ports.rs @@ -78,6 +78,8 @@ pub trait JobRunRepository: Send + Sync { async fn list(&self, limit: u32) -> Result, DomainError>; async fn find_running(&self, kind: JobKind) -> Result, DomainError>; async fn last_finished(&self, kind: JobKind) -> Result, DomainError>; + /// All runs still marked running (e.g. after a crash). + async fn running(&self) -> Result, DomainError>; } /// Reads the state of the Debian host. diff --git a/backend/crates/infrastructure/src/sqlite.rs b/backend/crates/infrastructure/src/sqlite.rs index 27ec044..a867f91 100644 --- a/backend/crates/infrastructure/src/sqlite.rs +++ b/backend/crates/infrastructure/src/sqlite.rs @@ -393,6 +393,15 @@ impl domain::ports::JobRunRepository for SqliteJobRuns { .map(|r| r.as_ref().map(job_from_row)) .map_err(storage) } + async fn running(&self) -> Result, DomainError> { + sqlx::query(&format!( + "SELECT {JOB_COLS} FROM job_runs WHERE status = 'running'" + )) + .fetch_all(&self.0) + .await + .map(|rows| rows.iter().map(job_from_row).collect()) + .map_err(storage) + } async fn last_finished(&self, kind: JobKind) -> Result, DomainError> { sqlx::query(&format!("SELECT {JOB_COLS} FROM job_runs WHERE kind = ? AND status != 'running' ORDER BY started_at DESC LIMIT 1")) .bind(kind.as_str()) diff --git a/backend/crates/infrastructure/src/trivy.rs b/backend/crates/infrastructure/src/trivy.rs index c40ac4b..08eba6a 100644 --- a/backend/crates/infrastructure/src/trivy.rs +++ b/backend/crates/infrastructure/src/trivy.rs @@ -18,26 +18,133 @@ impl TrivyScanner { } } -/// Parse `trivy ... --format json` output into findings (all results merged, deduplicated). -pub fn parse_trivy_json(_json: &str) -> Result, DomainError> { - todo!() +#[derive(serde::Deserialize)] +struct Report { + #[serde(rename = "Results", default)] + results: Option>, } +#[derive(serde::Deserialize)] +struct ResultEntry { + #[serde(rename = "Vulnerabilities", default)] + vulnerabilities: Option>, +} + +#[derive(serde::Deserialize)] +struct Vuln { + #[serde(rename = "VulnerabilityID")] + id: String, + #[serde(rename = "PkgName", default)] + pkg: String, + #[serde(rename = "InstalledVersion", default)] + installed: String, + #[serde(rename = "FixedVersion", default)] + fixed: Option, + #[serde(rename = "Severity", default)] + severity: String, + #[serde(rename = "Title", default)] + title: Option, + #[serde(rename = "PrimaryURL", default)] + url: Option, +} + +/// Parse `trivy ... --format json` output into findings (all results merged, deduplicated). +pub fn parse_trivy_json(json: &str) -> Result, DomainError> { + let report: Report = serde_json::from_str(json) + .map_err(|e| DomainError::Unavailable(format!("trivy output: {e}")))?; + let mut seen = std::collections::HashSet::new(); + let mut out = Vec::new(); + let vulns = report + .results + .unwrap_or_default() + .into_iter() + .flat_map(|r| r.vulnerabilities.unwrap_or_default()); + for v in vulns { + let f = RawFinding { + title: v + .title + .filter(|t| !t.is_empty()) + .unwrap_or_else(|| format!("{} in {}", v.id, v.pkg)), + url: v + .url + .unwrap_or_else(|| format!("https://nvd.nist.gov/vuln/detail/{}", v.id)), + cve_id: v.id, + severity: Severity::parse(&v.severity), + package: v.pkg, + installed_version: v.installed, + fixed_version: v.fixed.filter(|f| !f.is_empty()), + }; + if seen.insert(f.key()) { + out.push(f); + } + } + Ok(out) +} + +impl TrivyScanner { + async fn scan( + &self, + args: &[&str], + out: &dyn LineSink, + ) -> Result, DomainError> { + out.line(&format!("$ trivy {}", args.join(" "))); + let res = self.runner.run("trivy", args).await?; + if !res.success { + let tail: Vec<&str> = res.stderr.lines().rev().take(5).collect(); + for l in tail.into_iter().rev() { + out.line(l); + } + return Err(DomainError::Unavailable( + "trivy exited with a non-zero status".into(), + )); + } + let findings = parse_trivy_json(&res.stdout)?; + out.line(&format!("{} finding(s)", findings.len())); + Ok(findings) + } +} + +const COMMON: [&str; 7] = [ + "--format", + "json", + "--quiet", + "--scanners", + "vuln", + "--timeout", + "15m", +]; + #[async_trait] impl VulnerabilityScanner for TrivyScanner { async fn version(&self) -> Result { - let _ = &self.runner; - todo!() + let res = self.runner.run("trivy", &["--version"]).await?; + if !res.success { + return Err(DomainError::Unavailable("trivy is not installed".into())); + } + Ok(res + .stdout + .lines() + .next() + .unwrap_or_default() + .trim_start_matches("Version:") + .trim() + .to_string()) } - async fn scan_os(&self, _out: &dyn LineSink) -> Result, DomainError> { - todo!() + async fn scan_os(&self, out: &dyn LineSink) -> Result, DomainError> { + let mut args = vec!["rootfs"]; + args.extend(COMMON); + args.push("/"); + self.scan(&args, out).await } async fn scan_image( &self, - _image: &str, - _out: &dyn LineSink, + image: &str, + out: &dyn LineSink, ) -> Result, DomainError> { - todo!() + let mut args = vec!["image"]; + args.extend(COMMON); + args.push(image); + self.scan(&args, out).await } }