Skip to main content

max / makenotwork

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