Skip to main content

max / makenotwork

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