Skip to main content

max / makenotwork

41.0 KB · 919 lines History Blame Raw
1 //! Async scan worker.
2 //!
3 //! Spawned at startup from `main.rs`. Drains `scan_jobs` via
4 //! `db::scan_jobs::claim_next` and runs each job through the pipeline. On
5 //! completion, updates the target entity's `scan_status` (for entities that
6 //! have one) and creates a WAM ticket on `Quarantined`.
7 //!
8 //! The queue itself is `crate::db::scan_jobs`; the layers are `crate::scanning`.
9
10 use std::sync::Arc;
11 use std::time::Duration;
12
13 use sqlx::PgPool;
14 use tokio::sync::Semaphore;
15 use uuid::Uuid;
16
17 use crate::constants;
18 use crate::db::{
19 self, FileScanStatus, ItemId, VersionId,
20 scan_jobs::{ScanJob, ScanTargetKind},
21 };
22 use crate::storage::{FileType, StorageBackend};
23 use crate::wam_client::WamClient;
24
25 use super::{LayerResult, LayerVerdict, ScanPipeline, ScanResult};
26
27 /// Worker poll interval when the queue is empty.
28 const IDLE_POLL_INTERVAL: Duration = Duration::from_millis(500);
29
30 /// How long a `running` job can go without a heartbeat before the reaper resets
31 /// it. This is measured from the worker's last liveness beat (see
32 /// [`HEARTBEAT_INTERVAL`]), not from claim time, so a genuinely slow-but-
33 /// progressing scan is never reaped, only a crashed or hung worker crosses it.
34 const STUCK_JOB_SECS: i64 = 300;
35
36 /// Cadence at which the worker running a job refreshes its `heartbeat_at`. Must
37 /// be comfortably smaller than [`STUCK_JOB_SECS`] (here ~10x) so a live job that
38 /// is merely slow keeps beating well inside the reaper's window.
39 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30);
40
41 /// Cadence at which any worker tries to reap stuck jobs.
42 const REAPER_INTERVAL: Duration = Duration::from_mins(1);
43
44 /// Aborts a spawned task when dropped, so a job's heartbeat companion never
45 /// outlives the job, including when `process_job` returns early with an error.
46 struct AbortOnDrop(tokio::task::JoinHandle<()>);
47
48 impl Drop for AbortOnDrop {
49 fn drop(&mut self) {
50 self.0.abort();
51 }
52 }
53
54 /// Run `process_job` while a companion task keeps the job's `heartbeat_at`
55 /// fresh. The reaper distinguishes a slow-but-alive scan (recent beat) from a
56 /// crashed worker (stale beat) purely by this heartbeat, so a large scan that
57 /// runs past `STUCK_JOB_SECS` is no longer reclaimed and double-processed. The
58 /// companion is aborted the moment the job finishes (via `AbortOnDrop`),
59 /// error or not, so it cannot bump a job that has already left `running`.
60 async fn process_job_with_heartbeat(
61 ctx: &WorkerContext,
62 job: ScanJob,
63 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
64 let job_id = job.id;
65 let hb_db = ctx.db.clone();
66 let heartbeat = tokio::spawn(async move {
67 let mut ticker = tokio::time::interval(HEARTBEAT_INTERVAL);
68 // The claim already stamped heartbeat_at, so skip the immediate first
69 // tick and beat one interval from now.
70 ticker.tick().await;
71 loop {
72 ticker.tick().await;
73 if let Err(e) = db::scan_jobs::bump_heartbeat(&hb_db, job_id).await {
74 tracing::warn!(%job_id, error = %e, "scan heartbeat bump failed");
75 }
76 }
77 });
78 let _guard = AbortOnDrop(heartbeat);
79 process_job(ctx, job).await
80 }
81
82 /// Shared dependencies the worker pool needs.
83 pub struct WorkerContext {
84 pub db: PgPool,
85 pub s3: Arc<dyn StorageBackend>,
86 pub pipeline: Arc<ScanPipeline>,
87 pub scan_semaphore: Arc<Semaphore>,
88 pub wam: Option<WamClient>,
89 /// Bounded, shutdown-drained background pool. Quarantine side-effects (WAM
90 /// ticket, CF purge) go through this rather than a raw `tokio::spawn` so they
91 /// can't accumulate unbounded or be dropped on shutdown (Run 9).
92 pub bg: crate::background::BackgroundTx,
93 /// Cloudflare edge-cache purger, `None` when `CF_API_TOKEN`/`CF_ZONE_ID`
94 /// aren't configured. On quarantine we delete the origin object and, if a
95 /// purger is present, evict its `cdn_base_url`-prefixed URL from the edge so
96 /// an already-cached malicious copy stops serving before its TTL lapses.
97 pub cloudflare: Option<crate::cloudflare::CloudflarePurger>,
98 /// CDN base URL (e.g. "https://cdn.makenot.work"), used to build the purge
99 /// target for a quarantined object's public URL.
100 pub cdn_base_url: Arc<str>,
101 /// SyncKit-bucket backend. OTA artifacts live here (not the main `s3`), so a
102 /// `ScanTargetKind::OtaArtifact` job downloads from this backend instead.
103 /// `None` when SyncKit storage isn't configured (OTA scans then fail closed).
104 pub synckit_s3: Option<Arc<dyn StorageBackend>>,
105 /// Public, CDN-served bucket backend. Staging + scanning always read from
106 /// `s3`/`synckit_s3` (private); this backend is used ONLY to promote a Clean
107 /// object of an `is_cdn_served_without_gate()` image kind cross-bucket into
108 /// the public bucket. `None` when the public bucket isn't configured (image
109 /// promotes then fail closed, leaving the entity on its unserved staging key).
110 pub public_s3: Option<Arc<dyn StorageBackend>>,
111 /// Carried for one reason: an OTA artifact's verdict is mirrored into
112 /// `refs/notes/mnw/scan` on the commit its release was built from, and
113 /// finding the repository on disk needs `build.git_repos_path`.
114 pub config: crate::config::Config,
115 }
116
117 /// Decide the final status for a non-quarantined scan.
118 ///
119 /// First, the pipeline's own status is authoritative when it already says
120 /// `HeldForReview`: that means a `FailClosed` layer errored (a structural/YARA
121 /// panic, an oversize-to-spool file, or a reachable-but-incomplete clamav scan),
122 /// i.e. the file was *not fully scanned*. `ErrorPolicy::FailClosed` means "hold
123 /// for admin review" unconditionally, including for trusted uploaders, so we
124 /// never downgrade it here. (CHRONIC S1's fix for trusted uploaders depends on
125 /// this: the `clamav_incomplete` hold would otherwise be silently turned back
126 /// into `Clean` below.)
127 ///
128 /// Otherwise the pipeline returned `Clean`: a trusted uploader normally passes;
129 /// an untrusted one always routes to admin review. The ClamAV degraded-mode
130 /// overlay adds one exception, the clamav layer is `FailOpen` (a transport
131 /// error makes `final_status` skip it → Clean), but accepting a trusted upload
132 /// on zero AV coverage is the one fail-open we refuse, so ANY clamav error holds
133 /// the file for admin review rather than passing it (Run 9 Sec-S1: fail closed on
134 /// the first error instead of waiting for a runtime probe to observe a sustained
135 /// outage, the probe-lag window is gone).
136 fn resolve_pass_status(
137 pipeline_status: FileScanStatus,
138 is_trusted: bool,
139 layers: &[LayerResult],
140 ) -> FileScanStatus {
141 if pipeline_status == FileScanStatus::HeldForReview {
142 return FileScanStatus::HeldForReview;
143 }
144 if is_trusted && !clamav_layer_errored(layers) {
145 FileScanStatus::Clean
146 } else {
147 FileScanStatus::HeldForReview
148 }
149 }
150
151 /// True if the clamav layer reported a transport/scan error on this file. Under
152 /// the layer's `FailOpen` policy `final_status` skips such an error, so the trust
153 /// overlay re-reads it to fail closed for trusted uploads.
154 fn clamav_layer_errored(layers: &[LayerResult]) -> bool {
155 layers
156 .iter()
157 .any(|l| l.layer == "clamav" && l.verdict == LayerVerdict::Error)
158 }
159
160 /// Spawn `n` scan workers on the current tokio runtime. Each worker drains
161 /// `scan_jobs` independently with FOR UPDATE SKIP LOCKED. A single reaper
162 /// task per pool resets jobs that get stuck in `running`.
163 ///
164 /// All tasks observe `shutdown_rx`: when the sender is dropped (or the value
165 /// changes), they exit on their next idle cycle.
166 pub fn spawn_pool(
167 n: usize,
168 ctx: &Arc<WorkerContext>,
169 shutdown_rx: tokio::sync::watch::Receiver<()>,
170 ) {
171 for worker_id in 0..n {
172 let ctx = Arc::clone(ctx);
173 let mut shutdown_rx = shutdown_rx.clone();
174 tokio::spawn(async move {
175 tracing::info!(worker_id, "scan worker started");
176 loop {
177 match db::scan_jobs::claim_next(&ctx.db).await {
178 Ok(Some(job)) => {
179 let job_id = job.id;
180 if let Err(e) = process_job_with_heartbeat(&ctx, job).await {
181 tracing::error!(worker_id, %job_id, error = %e, "scan job failed");
182 if let Err(e2) =
183 db::scan_jobs::mark_failed(&ctx.db, job_id, &e.to_string()).await
184 {
185 tracing::error!(worker_id, %job_id, error = %e2, "failed to mark job failed");
186 }
187 }
188 }
189 Ok(None) => {
190 tokio::select! {
191 () = tokio::time::sleep(IDLE_POLL_INTERVAL) => {}
192 res = shutdown_rx.changed() => {
193 if res.is_err() {
194 tracing::info!(worker_id, "scan worker shutting down");
195 break;
196 }
197 }
198 }
199 }
200 Err(e) => {
201 tracing::error!(worker_id, error = %e, "claim_next failed; backing off");
202 tokio::select! {
203 () = tokio::time::sleep(Duration::from_secs(5)) => {}
204 res = shutdown_rx.changed() => {
205 if res.is_err() {
206 break;
207 }
208 }
209 }
210 }
211 }
212 }
213 });
214 }
215
216 let ctx_reaper = Arc::clone(ctx);
217 let mut shutdown_rx = shutdown_rx;
218 tokio::spawn(async move {
219 loop {
220 match db::scan_jobs::reap_stuck(&ctx_reaper.db, STUCK_JOB_SECS).await {
221 Ok(n) if n > 0 => {
222 tracing::warn!(
223 reset = n,
224 max_age_secs = STUCK_JOB_SECS,
225 "reset stuck scan jobs"
226 );
227 }
228 Ok(_) => {}
229 Err(e) => tracing::error!(error = %e, "scan job reaper failed"),
230 }
231 tokio::select! {
232 () = tokio::time::sleep(REAPER_INTERVAL) => {}
233 res = shutdown_rx.changed() => {
234 if res.is_err() {
235 break;
236 }
237 }
238 }
239 }
240 });
241 }
242
243 /// Test/dev helper: claim and process at most one queued scan job synchronously.
244 /// Returns `Ok(true)` when a job ran, `Ok(false)` when the queue was empty.
245 /// Mirrors `spawn_pool`'s per-iteration logic without spawning a background
246 /// task, so integration tests can deterministically drain the queue between
247 /// upload-confirm and assertion.
248 pub async fn process_next_for_test(
249 ctx: &WorkerContext,
250 ) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
251 match db::scan_jobs::claim_next(&ctx.db).await? {
252 Some(job) => {
253 let job_id = job.id;
254 if let Err(e) = process_job(ctx, job).await {
255 db::scan_jobs::mark_failed(&ctx.db, job_id, &e.to_string()).await?;
256 return Err(e);
257 }
258 Ok(true)
259 }
260 None => Ok(false),
261 }
262 }
263
264 /// Run a single scan job end-to-end. On success the job is marked done; the
265 /// caller marks failed if this returns an error.
266 ///
267 /// On pipeline error (e.g. S3 download failure), reset the entity from
268 /// Scanning back to HeldForReview before bubbling the error up. Otherwise
269 /// the entity stays stuck at Scanning forever, a real regression we hit
270 /// in production with stale s3_keys.
271 #[tracing::instrument(skip_all, fields(%job_id = job.id, target_kind = %job.target_kind, %target_id = job.target_id, attempts = job.attempts))]
272 async fn process_job(
273 ctx: &WorkerContext,
274 job: ScanJob,
275 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
276 let job_id = job.id;
277 let kind = job
278 .typed_kind()
279 .ok_or_else(|| format!("unknown target_kind: {}", job.target_kind))?;
280 let file_type = job
281 .typed_file_type()
282 .ok_or_else(|| format!("unknown file_type: {}", job.file_type))?;
283 let target_id = job.target_id;
284 let started = std::time::Instant::now();
285
286 // Mark target as Scanning while the worker is running (only entities with
287 // a scan_status column). This is a visible signal in the admin dashboard
288 // queue panel.
289 update_entity_status(&ctx.db, kind, target_id, FileScanStatus::Scanning)
290 .await
291 .ok();
292
293 let entity_status = match run_pipeline_and_decide(ctx, &job, kind, file_type).await {
294 Ok(s) => s,
295 Err(e) => {
296 // Pipeline blew up, most often a stale s3_key. Reset entity to
297 // HeldForReview so admins see it on the dashboard and decide
298 // whether to delete the orphan record.
299 update_entity_status(&ctx.db, kind, target_id, FileScanStatus::HeldForReview)
300 .await
301 .ok();
302 crate::metrics::record_scan_verdict("error");
303 crate::metrics::record_scan_duration(started.elapsed().as_secs_f64());
304 return Err(e);
305 }
306 };
307 // Stamp the entity's terminal scan_status. For a Clean staging upload the
308 // promote inside `run_pipeline_and_decide` already set the key column AND
309 // `scan_status = 'clean'` in one transaction; this re-stamp is then a
310 // harmless idempotent write. It is load-bearing, though, for a Clean file
311 // that was uploaded server-side to a NON-staging key (the build runner's OTA
312 // artifacts), there is nothing to promote, so this is the only place its
313 // status is cleared. (For the gate-less image kinds this is a no-op; their
314 // row was stamped by the promote or the held-image branch above.)
315 update_entity_status(&ctx.db, kind, target_id, entity_status).await?;
316
317 // An OTA artifact's verdict belongs in the repository it was built from, not
318 // only in a table. Runs after the stamp above so the release's artifacts are
319 // read back with this one's verdict already in them.
320 if kind == ScanTargetKind::OtaArtifact {
321 annotate_release_scan(ctx, db::OtaArtifactId::from(target_id)).await;
322 }
323
324 // Verdict + duration metrics (Run 20 Observability): quarantine/hold/error
325 // rates and scan latency are now graphable at /metrics.
326 let verdict_label = match entity_status {
327 FileScanStatus::Clean => "clean",
328 FileScanStatus::Quarantined => "quarantined",
329 FileScanStatus::HeldForReview => "held_for_review",
330 FileScanStatus::Pending => "pending",
331 FileScanStatus::Scanning => "scanning",
332 FileScanStatus::Error => "error",
333 };
334 crate::metrics::record_scan_verdict(verdict_label);
335 crate::metrics::record_scan_duration(started.elapsed().as_secs_f64());
336
337 db::scan_jobs::mark_done(&ctx.db, job_id).await?;
338 Ok(())
339 }
340
341 /// Run the pipeline against the S3 object and return the entity status to
342 /// apply, honoring the size guard, trust gate, and WAM ticketing.
343 async fn run_pipeline_and_decide(
344 ctx: &WorkerContext,
345 job: &ScanJob,
346 kind: ScanTargetKind,
347 file_type: FileType,
348 ) -> Result<FileScanStatus, Box<dyn std::error::Error + Send + Sync>> {
349 // Two paths, gated on file size. Small files go through the original
350 // buffered `Pipeline::scan(Vec<u8>)`: a single S3 GET into a heap
351 // buffer, then layers walk the slice. Big files (>= SCAN_MAX_MEMORY_BYTES)
352 // stream from S3 into a tempfile under SCAN_SPOOL_DIR, then layers
353 // run against the spooled path (mmap or streamed). The buffered path
354 // stays alive: it's the hot path for tip-jar avatars / small audio /
355 // download files, and avoiding the tempfile syscall + write matters at
356 // that scale. Both branches run S3 IO *outside* the scan_semaphore:
357 // the permit bounds the CPU/clamd-heavy scan phase, not network IO.
358 // Holding it across the GET serializes downloads at SCAN_MAX_CONCURRENT
359 // and lets a scan backlog starve the DB pool.
360 // OTA artifacts live in the SyncKit bucket; everything else in the main
361 // bucket. `kind.storage_bucket()` is the single source of truth: the same
362 // value drives the download client here, the quarantine delete, and the
363 // durable-delete enqueue below, so they can't diverge (ultra-fuzz Run #24
364 // Storage CRITICAL was a wrong-bucket quarantine delete from exactly that
365 // divergence).
366 let bucket = kind.storage_bucket();
367 let backend: &Arc<dyn StorageBackend> = match bucket {
368 crate::storage::S3Bucket::Synckit => ctx.synckit_s3.as_ref().ok_or_else(|| {
369 Box::<dyn std::error::Error + Send + Sync>::from(
370 "SyncKit storage not configured; cannot scan OTA artifact",
371 )
372 })?,
373 // `storage_bucket()` is the STAGING/scan bucket and is never `Public`
374 // (unscanned bytes stay private; only the promoted content object moves
375 // to the public bucket, see `content_served_from_public_bucket`). Guard
376 // it so a future routing change fails loudly rather than reading a
377 // scan object from the wrong bucket.
378 crate::storage::S3Bucket::Public => {
379 return Err(Box::<dyn std::error::Error + Send + Sync>::from(
380 "invariant: a staging/scan object must never live in the public bucket",
381 ));
382 }
383 crate::storage::S3Bucket::Main => &ctx.s3,
384 };
385
386 let result: ScanResult = if job.file_size_bytes as u64 > constants::SCAN_SPOOL_MAX_BYTES {
387 // Too large to spool for the CPU layers. Hold for admin review under an
388 // explicit policy rather than letting `download_into_tempfile` return an
389 // Err that marks the job failed, that left the file stuck `Pending`
390 // (download-blocked but never resolved) and retried forever. See
391 // `super::too_large_to_scan`.
392 tracing::warn!(
393 job_id = %job.id, size = job.file_size_bytes,
394 cap = constants::SCAN_SPOOL_MAX_BYTES,
395 "upload exceeds scan spool ceiling; holding for review (not auto-scanned)"
396 );
397 super::too_large_to_scan(job.file_size_bytes as u64)
398 } else if (job.file_size_bytes as usize) < constants::SCAN_MAX_MEMORY_BYTES {
399 // `download_object_buf_capped` returns the aggregated body as `Bytes` (no
400 // `to_vec` copy); `scan` takes it directly (Run #2 Performance SERIOUS).
401 // Bound the aggregation by the recorded size + slack, never above the
402 // in-memory threshold: `file_size_bytes` is asserted at upload and could
403 // under-report the real object, so cap the buffered read like the spool
404 // path does rather than pulling an unbounded body into RAM (Run 22 Perf).
405 let cap = (job.file_size_bytes as u64)
406 .saturating_add(constants::SCAN_SPOOL_SLACK_BYTES)
407 .min(constants::SCAN_MAX_MEMORY_BYTES as u64);
408 let data = backend.download_object_buf_capped(&job.s3_key, cap).await?;
409 let _permit = ctx.scan_semaphore.acquire().await?;
410 Arc::clone(&ctx.pipeline).scan(data, file_type).await
411 } else {
412 let stream = backend.download_stream(&job.s3_key).await?;
413 let spool = super::spool::download_into_tempfile(
414 std::path::Path::new(constants::SCAN_SPOOL_DIR),
415 &job.id.to_string(),
416 &job.s3_key,
417 job.file_size_bytes as u64,
418 stream,
419 )
420 .await?;
421 let _permit = ctx.scan_semaphore.acquire().await?;
422 Arc::clone(&ctx.pipeline)
423 .scan_stream(spool, file_type)
424 .await
425 };
426
427 db::scanning::insert_scan_result(&ctx.db, &job.s3_key, &result).await?;
428
429 if result.status == FileScanStatus::Quarantined {
430 let failed_layers: Vec<&str> = result
431 .layers
432 .iter()
433 .filter(|l| l.verdict == LayerVerdict::Fail)
434 .map(|l| l.layer)
435 .collect();
436 if let Some(wam) = ctx.wam.clone() {
437 // Fire-and-forget: the WAM call has a multi-second timeout, and the
438 // verdict enforcement below (row purge + object delete) must not wait
439 // on it, otherwise a WAM outage stalls every quarantine ~5s before
440 // the malicious object is removed (Run #21 Performance MODERATE).
441 // Matches the spawned-ticket pattern the checkout path already uses.
442 let title = format!("File quarantined: {}", job.s3_key);
443 let body = format!(
444 "Upload by user {} flagged as malicious.\n\
445 Failed layers: {}\nFile type: {file_type:?}\nSize: {}",
446 job.user_id,
447 failed_layers.join(", "),
448 job.file_size_bytes,
449 );
450 let s3_key = job.s3_key.clone();
451 ctx.bg.spawn("malware quarantine ticket", async move {
452 wam.create_ticket(
453 &title,
454 Some(&body),
455 "high",
456 "malware-quarantine",
457 Some(&s3_key),
458 )
459 .await;
460 });
461 }
462
463 // Enforce the verdict by removing the malicious content. Two parts:
464 //
465 // 1. DB row (gate-less image kinds only). Project/gallery/content-insertion
466 // images carry no `scan_status` column and no app-proxied download
467 // route, so the only way to stop the (Cloudflare-served) URL from
468 // rendering is to delete the row. Doing it first also makes the
469 // s3_key non-live, so the durable-deletion queue won't park the
470 // object behind the `is_s3_key_live` guard.
471 // 2. S3 object (every kind). The per-request `scan_status` gate on
472 // Item/Version/Media stops the *proxied* download, but free
473 // downloadable content is served as a PERMANENT cdn.makenot.work/{key}
474 // URL with no per-request gate, so a leaked or edge-cached URL keeps
475 // serving the malware from origin until the object is gone. Deleting
476 // the object closes that hole for downloadable content the same way
477 // it already does for images.
478 //
479 // Edge cache: Cloudflare caches objects immutably for a year, so an
480 // already-edge-cached copy would survive origin deletion until the cache
481 // TTL lapses. After the origin delete below we fire a Cloudflare
482 // cache-purge for the object's URL (see the `ctx.cloudflare` block) when
483 // `CF_API_TOKEN`/`CF_ZONE_ID` are configured; when they aren't, the purge
484 // is a logged no-op and the WAM ticket above remains the manual trigger.
485 // The verdict + failed layers stay in file_scan_results for admin review.
486 let row_deleted = kind.is_cdn_served_without_gate();
487 if row_deleted {
488 match db::scanning::purge_cdn_image_rows_by_key(&ctx.db, &job.s3_key).await {
489 Ok(n) => tracing::warn!(
490 s3_key = %job.s3_key, target_kind = %kind.as_str(), rows_removed = n,
491 "removed quarantined CDN-served image row(s); URL is no longer rendered"
492 ),
493 Err(e) => tracing::error!(
494 s3_key = %job.s3_key, target_kind = %kind.as_str(), error = %e,
495 "FAILED to remove quarantined image row(s); the URL may still render until manual removal"
496 ),
497 }
498 }
499
500 // Track whether the malicious object was actually removed (deleted now,
501 // or durably enqueued for deletion). A kind that doesn't purge objects
502 // has nothing to remove here, so it counts as removed. If removal fails
503 // we must NOT report a successful quarantine (see the check below).
504 let mut object_removed = !kind.quarantine_purges_object();
505 if kind.quarantine_purges_object() {
506 // Sanctioned direct delete: the quarantine worker must immediately
507 // purge a confirmed-malicious object (the durable queue would park a
508 // still-referenced gated key). Mint the authority the sealed delete
509 // API requires.
510 let auth = crate::storage::S3DeleteAuthority::new();
511 match backend
512 .delete_object(&auth, &crate::storage::S3Key::from_stored(&job.s3_key))
513 .await
514 {
515 Ok(()) => {
516 tracing::warn!(
517 s3_key = %job.s3_key, target_kind = %kind.as_str(),
518 "purged quarantined object from storage"
519 );
520 object_removed = true;
521 }
522 Err(e) => {
523 tracing::error!(
524 s3_key = %job.s3_key, target_kind = %kind.as_str(), error = %e,
525 "immediate purge of quarantined object failed"
526 );
527 // The durable-deletion queue only deletes keys the
528 // `is_s3_key_live` guard considers dead. For gate-less image
529 // kinds we just deleted the row, so the key is dead and the
530 // queue will finish the job. For gated kinds (Item/Version/
531 // Media) the entity row still references the key, the queue
532 // would park it indefinitely, so enqueuing is futile and the
533 // job must be retried (below) to re-attempt the direct delete.
534 if row_deleted {
535 match db::pending_s3_deletions::enqueue_deletions(
536 &ctx.db,
537 &[(job.s3_key.clone(), bucket.as_str().to_string())],
538 "malware_quarantine",
539 )
540 .await
541 {
542 Ok(()) => {
543 tracing::warn!(
544 s3_key = %job.s3_key,
545 "quarantined object enqueued for durable deletion after direct purge failed"
546 );
547 object_removed = true;
548 }
549 Err(enqueue_err) => tracing::error!(
550 s3_key = %job.s3_key, error = %enqueue_err,
551 "FAILED to enqueue quarantined object for durable deletion; will retry the scan job"
552 ),
553 }
554 } else {
555 tracing::error!(
556 s3_key = %job.s3_key, target_kind = %kind.as_str(),
557 "quarantined object still referenced by its entity row; durable queue cannot delete it, will retry the scan job"
558 );
559 }
560 }
561 }
562 }
563
564 // Never report a successful quarantine while the malware is still live at
565 // origin. If we could neither delete nor durably enqueue the object,
566 // return an error: `process_job` leaves the job un-done (the scan-job
567 // retry budget re-runs the quarantine, a transient S3 delete resolves on
568 // retry) and holds the entity for review meanwhile; the WAM ticket above
569 // is the manual-removal escalation if retries are exhausted (ultra-fuzz
570 // Run 13 Storage: quarantine must not fail-to-success).
571 if !object_removed {
572 return Err(format!(
573 "quarantine incomplete: malicious object {} could neither be purged nor enqueued for deletion",
574 job.s3_key
575 )
576 .into());
577 }
578
579 // Evict any edge-cached copy of the now-deleted object. Origin deletion
580 // above stops origin serving, but Cloudflare caches cdn.makenot.work/{key}
581 // immutably for up to a year, so a previously-fetched malicious URL keeps
582 // serving from the edge until its TTL lapses. Fire-and-forget (the CF API
583 // has its own latency and must not stall enforcement); a no-op when the
584 // purger isn't configured (the WAM ticket stays the fallback).
585 // Purging a never-cached URL is harmless, so we don't gate on kind.
586 if let Some(cf) = ctx.cloudflare.clone() {
587 let url = format!("{}/{}", ctx.cdn_base_url.trim_end_matches('/'), job.s3_key);
588 ctx.bg.spawn("malware quarantine cdn purge", async move {
589 cf.purge_urls(vec![url]).await;
590 });
591 }
592
593 return Ok(FileScanStatus::Quarantined);
594 }
595
596 // Pipeline returned Clean or HeldForReview. Apply the uploader-trust
597 // overlay (untrusted users always route to admin review) plus the ClamAV
598 // degraded-mode overlay (a trusted upload whose clamav layer errored is held
599 // rather than passed on reduced AV coverage). See `resolve_pass_status`.
600 let is_trusted = db::users::is_upload_trusted(&ctx.db, job.user_id).await?;
601 let status = resolve_pass_status(result.status, is_trusted, &result.layers);
602
603 // Visibility: an otherwise-clean trusted upload that we held *because* its
604 // clamav layer errored. We now fail closed on the first such error (Run 9
605 // Sec-S1), so this is no longer a silent acceptance, but a clamd outage that
606 // starts holding trusted uploads is still operationally important, so surface
607 // it via a metric (scraped) and a WAM ticket (active alert).
608 if clamav_degraded_hold_occurred(result.status, is_trusted, &result.layers) {
609 tracing::warn!(
610 s3_key = %job.s3_key,
611 user_id = %job.user_id,
612 "clamav layer errored on a trusted upload; held for review on reduced AV coverage"
613 );
614 crate::metrics::record_clamav_degraded_hold();
615 if let Some(wam) = ctx.wam.clone() {
616 let body = format!(
617 "A trusted upload was held for review because its clamav layer errored \
618 (reduced AV coverage, clamd may be unreachable).\n\n\
619 s3_key: {}\nuser_id: {}",
620 job.s3_key, job.user_id
621 );
622 wam.create_ticket(
623 "ClamAV degraded: trusted upload held on reduced AV coverage",
624 Some(&body),
625 "medium",
626 "clamav-degraded-hold",
627 Some(&job.s3_key),
628 )
629 .await;
630 }
631 }
632
633 // Gate-less CDN-served image kinds (item/project covers, gallery carousels,
634 // content-insertion clips) carry no entity `scan_status` column for
635 // `update_entity_status` to flip, and they render straight from
636 // cdn.makenot.work/{key} with no per-request gate. Stamp their per-row
637 // scan_status here, keyed on s3_key (symmetric with the quarantine purge
638 // above), so the fail-closed render gate can distinguish an unscanned
639 // `pending` row from a cleared `clean` one. We stamp the trust-overlaid
640 // `status`, not the raw pipeline `result.status`: an untrusted (or held)
641 // upload's image stays hidden until an admin clears it, the same
642 // fail-closed posture the gated Item/Version/Media kinds already get.
643 // Quarantine never reaches here (it returned above with the row purged).
644 // C1 scan-then-promote: a Clean verdict copies the object from its unserved
645 // staging key to the immutable content key and repoints the entity (gated
646 // kinds by id, CDN-image kinds by staging key, incl. rebuilding the public
647 // URL) in one shared step. This is the ONLY place a served key comes into
648 // existence, the presign handlers can only mint staging keys (the sealed
649 // generators are `pub(crate)`, callable only from here via `content_key`).
650 // Fail-safe: a copy/DB error bubbles up so `process_job` leaves the job
651 // un-done and the entity keeps pointing at the (unserved) staging key; the
652 // retry re-promotes.
653 if status == FileScanStatus::Clean && job.s3_key.starts_with("staging/") {
654 // Client-presigned upload: copy staging -> content key and repoint the
655 // entity. A Clean file at a non-staging key was uploaded server-side to
656 // its final key (build-runner OTA) and needs no promote, its status is
657 // stamped by `update_entity_status` in `process_job`.
658 super::promote_staging_to_content(
659 &ctx.db,
660 backend.as_ref(),
661 ctx.public_s3.as_deref(),
662 &ctx.cdn_base_url,
663 kind,
664 file_type,
665 job.target_id,
666 job.user_id,
667 &job.s3_key,
668 &result.sha256,
669 bucket,
670 )
671 .await?;
672 } else if status != FileScanStatus::Clean && kind.is_cdn_served_without_gate() {
673 // Held/pending image (never promoted): stamp the row keyed on its staging
674 // key so the fail-closed render gate hides it until an admin clears it.
675 match db::scanning::set_cdn_image_scan_status_by_key(&ctx.db, &job.s3_key, status).await {
676 Ok(n) => tracing::info!(
677 s3_key = %job.s3_key, target_kind = %kind.as_str(), scan_status = %status, rows = n,
678 "stamped CDN-served image scan_status (renders only when clean)"
679 ),
680 Err(e) => tracing::warn!(
681 s3_key = %job.s3_key, target_kind = %kind.as_str(), error = %e,
682 "failed to stamp CDN-served image scan_status; image stays hidden until re-scan"
683 ),
684 }
685 }
686
687 Ok(status)
688 }
689
690 /// True when an otherwise-clean trusted upload was held *because* its clamav layer
691 /// errored (the file would have passed but for the AV-coverage gap). Extracted as
692 /// a pure predicate so the condition is unit-tested rather than living only inline.
693 fn clamav_degraded_hold_occurred(
694 pipeline_status: FileScanStatus,
695 is_trusted: bool,
696 layers: &[crate::scanning::LayerResult],
697 ) -> bool {
698 pipeline_status == FileScanStatus::Clean && is_trusted && clamav_layer_errored(layers)
699 }
700
701 /// Update the per-entity `scan_status` column for the kinds that have one.
702 /// `ItemImage` / `ProjectImage` / `GalleryImage` / `ContentInsertion` don't
703 /// carry their own column, the worker still scanned the file and recorded
704 /// results, but there's no status to flip on those entities (the `ItemImage`
705 /// cover shares the `items` row but must NOT flip `items.scan_status`, which
706 /// gates the audio/video, not the cover).
707 /// Mirror a release's scan verdicts into `refs/notes/mnw/scan`, once every
708 /// artifact in it has one.
709 ///
710 /// Waiting for the last artifact is the point. A note saying one of three is
711 /// clean is not a verdict, and writing it per artifact would leave two
712 /// misleading intermediate states in the notes ref's history for anyone reading
713 /// it later. Every early return here is an ordinary case: a release nobody
714 /// built from a tag has no commit to annotate.
715 async fn annotate_release_scan(ctx: &WorkerContext, artifact_id: db::OtaArtifactId) {
716 let Ok(Some(artifact)) = db::ota::get_artifact_by_id(&ctx.db, artifact_id).await else {
717 return;
718 };
719 let Ok(artifacts) = db::ota::list_artifacts(&ctx.db, artifact.release_id).await else {
720 return;
721 };
722 if artifacts.is_empty()
723 || artifacts.iter().any(|a| {
724 matches!(
725 a.scan_status,
726 FileScanStatus::Pending | FileScanStatus::Scanning
727 )
728 })
729 {
730 return;
731 }
732
733 // The build is what ties a release to a tag, and so to a commit. A release
734 // uploaded by hand has none.
735 let Ok(Some(build)) = db::builds::get_build_by_release(&ctx.db, artifact.release_id).await
736 else {
737 return;
738 };
739 let Ok(Some(config)) = db::builds::get_build_config_by_app(&ctx.db, build.app_id).await else {
740 return;
741 };
742 let Ok(Some(repo)) = db::git_repos::get_repo_by_id(&ctx.db, config.repo_id).await else {
743 return;
744 };
745 let Ok(Some(owner)) = db::users::get_user_by_id(&ctx.db, repo.user_id).await else {
746 return;
747 };
748 let owner = owner.username.to_string();
749
750 let Some(target) = crate::routes::git::notes_server::resolve_tag_commit(
751 &ctx.config,
752 &owner,
753 &repo.name,
754 &build.tag,
755 )
756 .await
757 else {
758 return;
759 };
760
761 let verdicts: Vec<(String, FileScanStatus)> = artifacts
762 .iter()
763 .map(|a| (format!("{}/{}", a.target, a.arch), a.scan_status))
764 .collect();
765
766 crate::routes::git::notes_server::note_scan(
767 &crate::routes::git::notes_server::OwnedRepo {
768 db: &ctx.db,
769 config: &ctx.config,
770 id: repo.id,
771 owner: &owner,
772 name: &repo.name,
773 },
774 target,
775 &build.version,
776 &verdicts,
777 )
778 .await;
779 }
780
781 async fn update_entity_status(
782 db: &PgPool,
783 kind: ScanTargetKind,
784 target_id: Uuid,
785 status: FileScanStatus,
786 ) -> Result<(), sqlx::Error> {
787 match kind {
788 ScanTargetKind::Version => {
789 db::scanning::update_version_scan_status(db, VersionId::from(target_id), status).await
790 }
791 ScanTargetKind::Item => {
792 db::scanning::update_item_scan_status(db, ItemId::from(target_id), status).await
793 }
794 ScanTargetKind::Media => {
795 db::scanning::update_media_file_scan_status(
796 db,
797 db::MediaFileId::from(target_id),
798 status,
799 )
800 .await
801 }
802 ScanTargetKind::OtaArtifact => {
803 db::ota::update_artifact_scan_status(db, db::OtaArtifactId::from(target_id), status)
804 .await
805 }
806 ScanTargetKind::ItemImage
807 | ScanTargetKind::ProjectImage
808 | ScanTargetKind::GalleryImage
809 | ScanTargetKind::ContentInsertion => Ok(()),
810 }
811 }
812
813 #[cfg(test)]
814 mod tests {
815 use super::*;
816
817 fn layer(name: &'static str, verdict: LayerVerdict) -> LayerResult {
818 LayerResult {
819 layer: name,
820 verdict,
821 detail: None,
822 }
823 }
824
825 // The first arg is the pipeline's own status (Clean unless a FailClosed
826 // layer errored). The cases below exercise a Clean pipeline; the
827 // *_not_downgraded tests cover a HeldForReview pipeline.
828 const CLEAN: FileScanStatus = FileScanStatus::Clean;
829
830 #[test]
831 fn trusted_passes_when_clamav_clean() {
832 let layers = [layer("clamav", LayerVerdict::Pass)];
833 assert_eq!(
834 resolve_pass_status(CLEAN, true, &layers),
835 FileScanStatus::Clean
836 );
837 }
838
839 #[test]
840 fn trusted_held_on_first_clamav_error() {
841 // Sec-S1 (Run 9): fail closed on the FIRST clamav error. The clamav layer
842 // is FailOpen (final_status skipped the error → Clean), but a trusted
843 // upload on zero AV coverage is held for review immediately, no waiting
844 // for a runtime probe to observe a sustained outage.
845 let layers = [layer("clamav", LayerVerdict::Error)];
846 assert_eq!(
847 resolve_pass_status(CLEAN, true, &layers),
848 FileScanStatus::HeldForReview
849 );
850 }
851
852 #[test]
853 fn untrusted_always_held() {
854 let layers = [layer("clamav", LayerVerdict::Pass)];
855 assert_eq!(
856 resolve_pass_status(CLEAN, false, &layers),
857 FileScanStatus::HeldForReview
858 );
859 }
860
861 #[test]
862 fn trusted_held_when_pipeline_already_held_not_downgraded() {
863 // CHRONIC S1: a reachable-but-incomplete clamav scan (or any FailClosed
864 // layer error) makes final_status = HeldForReview. A trusted uploader must
865 // NOT have that downgraded to Clean.
866 let layers = [layer("clamav_incomplete", LayerVerdict::Error)];
867 assert_eq!(
868 resolve_pass_status(FileScanStatus::HeldForReview, true, &layers),
869 FileScanStatus::HeldForReview
870 );
871 }
872
873 #[test]
874 fn pipeline_hold_survives_even_with_no_clamav_layer() {
875 // The no-downgrade guard is layer-agnostic: any FailClosed hold (e.g. an
876 // oversize-to-spool file or a structural panic) is honored for trusted
877 // uploaders too.
878 let layers = [layer("scan_size_limit", LayerVerdict::Error)];
879 assert_eq!(
880 resolve_pass_status(FileScanStatus::HeldForReview, true, &layers),
881 FileScanStatus::HeldForReview
882 );
883 }
884
885 // ── clamav_degraded_hold_occurred (observability predicate) ──
886
887 #[test]
888 fn degraded_hold_detected_for_trusted_clamav_error() {
889 let layers = [layer("clamav", LayerVerdict::Error)];
890 assert!(clamav_degraded_hold_occurred(CLEAN, true, &layers));
891 }
892
893 #[test]
894 fn degraded_hold_not_for_untrusted_upload() {
895 // Untrusted uploads are always held regardless of clamav; the degraded
896 // metric tracks only the trusted-upload coverage-gap case.
897 let layers = [layer("clamav", LayerVerdict::Error)];
898 assert!(!clamav_degraded_hold_occurred(CLEAN, false, &layers));
899 }
900
901 #[test]
902 fn degraded_hold_not_when_clamav_passed() {
903 let layers = [layer("clamav", LayerVerdict::Pass)];
904 assert!(!clamav_degraded_hold_occurred(CLEAN, true, &layers));
905 }
906
907 #[test]
908 fn degraded_hold_not_when_pipeline_not_clean() {
909 // A file the pipeline already held (FailClosed) wasn't held *because* of
910 // the clamav coverage gap, so it is not a degraded-hold event.
911 let layers = [layer("clamav", LayerVerdict::Error)];
912 assert!(!clamav_degraded_hold_occurred(
913 FileScanStatus::HeldForReview,
914 true,
915 &layers
916 ));
917 }
918 }
919