Skip to main content

max / makenotwork

20.2 KB · 552 lines History Blame Raw
1 //! Database operations for file scan results.
2
3 use chrono::{DateTime, Utc};
4 use sqlx::{FromRow, PgPool};
5 use uuid::Uuid;
6
7 use super::FileScanStatus;
8 use super::ItemId;
9 use super::UserId;
10 use super::VersionId;
11 use crate::scanning::ScanResult;
12
13 /// An item held for review, joined with creator info and latest scan layers.
14 #[derive(Debug, Clone, FromRow)]
15 pub struct HeldItemRow {
16 pub item_id: ItemId,
17 pub item_title: String,
18 pub s3_key: Option<String>,
19 pub creator_username: String,
20 pub creator_id: UserId,
21 pub upload_trusted: bool,
22 pub held_at: DateTime<Utc>,
23 /// Latest `file_scan_results.scan_layers` JSON for this entity's s3_key,
24 /// or `null` if no scan has run yet. The dashboard renders this as chips.
25 pub scan_layers: Option<serde_json::Value>,
26 }
27
28 /// A version held for review, joined with creator info and latest scan layers.
29 #[derive(Debug, Clone, FromRow)]
30 pub struct HeldVersionRow {
31 pub item_id: ItemId,
32 pub item_title: String,
33 pub version_id: VersionId,
34 pub version_number: String,
35 pub s3_key: Option<String>,
36 pub creator_username: String,
37 pub creator_id: UserId,
38 pub upload_trusted: bool,
39 pub held_at: DateTime<Utc>,
40 pub scan_layers: Option<serde_json::Value>,
41 }
42
43 /// Remove every CDN-served image reference to `s3_key`, across the image
44 /// surfaces that have no per-row scan gate: the gallery tables
45 /// (`item_images.s3_key`, `project_images.s3_key`, `content_insertions.storage_key`)
46 /// and the item *cover* columns on `items` (`cover_s3_key`/`cover_image_url`).
47 /// Returns the number of rows affected.
48 ///
49 /// On quarantine of these kinds, removing the reference IS the primary
50 /// enforcement: it stops the app from ever rendering the (Cloudflare-served) URL
51 /// again, and, critically, makes the key non-live so the durable S3-deletion
52 /// queue will actually purge the object instead of parking it behind the
53 /// `is_s3_key_live` guard. `storage_used` counters self-heal on the weekly
54 /// `recalculate_all_storage_used` pass; we accept a transient over-count for a
55 /// malicious upload rather than join through three ownership paths here.
56 ///
57 /// The item cover is a special case: the bad image lives in columns ON the
58 /// `items` row, so we NULL the three cover columns rather than DELETE the row,
59 /// deleting it would take the legitimate audio/video track down with the
60 /// thumbnail. Both cover references (`cover_s3_key` exact + `cover_image_url`
61 /// suffix) are registered in `S3_KEY_REFS`, so NULLing both makes the key dead.
62 #[tracing::instrument(skip_all)]
63 pub async fn purge_cdn_image_rows_by_key(db: &PgPool, s3_key: &str) -> Result<u64, sqlx::Error> {
64 // Wrap all reference-clearing statements in one transaction: a mid-loop
65 // failure must not leave a partially-purged malicious reference set (some
66 // tables still pointing at a quarantined object while others were cleared).
67 let mut tx = db.begin().await?;
68 let mut removed = 0u64;
69 for sql in [
70 "DELETE FROM item_images WHERE s3_key = $1",
71 "DELETE FROM project_images WHERE s3_key = $1",
72 "DELETE FROM content_insertions WHERE storage_key = $1",
73 ] {
74 removed += sqlx::query(sql)
75 .bind(s3_key)
76 .execute(&mut *tx)
77 .await?
78 .rows_affected();
79 }
80 // Item + project covers: NULL the columns in place, keeping the row (a track
81 // or a project must not be deleted along with its quarantined thumbnail).
82 // Matches on the exact `cover_s3_key`; the cover_image_url suffix is cleared
83 // in the same statement so neither reference keeps the object live. BOTH the
84 // items and projects cover surfaces must be cleared, the read-side
85 // counterpart `set_cdn_image_scan_status_by_key` stamps both, and
86 // `promote_cdn_image_by_key` promotes both, so a purge that skipped projects
87 // would leave a quarantined project cover rendered AND its (public-bucket)
88 // object un-reapable behind the `is_s3_key_live` guard.
89 for sql in [
90 "UPDATE items SET cover_s3_key = NULL, cover_image_url = NULL, \
91 cover_file_size_bytes = NULL, updated_at = NOW() WHERE cover_s3_key = $1",
92 "UPDATE projects SET cover_s3_key = NULL, cover_image_url = NULL, \
93 updated_at = NOW() WHERE cover_s3_key = $1",
94 ] {
95 removed += sqlx::query(sql)
96 .bind(s3_key)
97 .execute(&mut *tx)
98 .await?
99 .rows_affected();
100 }
101 tx.commit().await?;
102 Ok(removed)
103 }
104
105 /// Of the given S3 keys, the subset whose authoritative scan verdict is
106 /// `Quarantined` (confirmed malicious).
107 ///
108 /// `file_scan_results` is the per-object scan record the quarantine worker
109 /// writes; keying on it covers every content surface (audio/cover/video,
110 /// versions, insertions) uniformly. Used to exclude quarantined objects from
111 /// bulk key reads that ultimately serve content to a user, notably content
112 /// export, which collects raw keys and would otherwise hand a creator back their
113 /// own quarantined object that every download path already refuses (Run 20
114 /// Security). Absent rows (never scanned) are not quarantined and are not
115 /// returned.
116 #[tracing::instrument(skip_all)]
117 pub async fn quarantined_s3_keys(
118 pool: &PgPool,
119 keys: &[String],
120 ) -> Result<std::collections::HashSet<String>, sqlx::Error> {
121 if keys.is_empty() {
122 return Ok(std::collections::HashSet::new());
123 }
124 let rows: Vec<(String,)> = sqlx::query_as(
125 "SELECT DISTINCT s3_key FROM file_scan_results \
126 WHERE s3_key = ANY($1) AND scan_status = 'quarantined'",
127 )
128 .bind(keys)
129 .fetch_all(pool)
130 .await?;
131 Ok(rows.into_iter().map(|(k,)| k).collect())
132 }
133
134 /// Set the per-row `scan_status` on every CDN-served image surface keyed by
135 /// `s3_key`, in one transaction. The symmetric read-side counterpart of
136 /// [`purge_cdn_image_rows_by_key`]: where the purge DELETEs (or NULLs) a
137 /// quarantined reference, this stamps a non-quarantine terminal verdict,
138 /// `clean` (renders) or `held` (stays hidden), so the fail-closed gate can
139 /// distinguish "not yet scanned" (`pending`) from "scanned and cleared".
140 ///
141 /// Touches the same surfaces as the purge: the gallery tables
142 /// (`item_images.s3_key`, `project_images.s3_key`,
143 /// `content_insertions.storage_key`) and the cover columns on `items` /
144 /// `projects` (`cover_s3_key`). Wrapped in a single transaction so a mid-loop
145 /// failure can't leave the reference set half-stamped (some surfaces cleared to
146 /// `clean` while others stay `pending`). Returns total rows affected.
147 #[tracing::instrument(skip_all)]
148 pub async fn set_cdn_image_scan_status_by_key(
149 db: &PgPool,
150 s3_key: &str,
151 status: FileScanStatus,
152 ) -> Result<u64, sqlx::Error> {
153 let mut tx = db.begin().await?;
154 let mut affected = 0u64;
155 for sql in [
156 "UPDATE item_images SET scan_status = $1 WHERE s3_key = $2",
157 "UPDATE project_images SET scan_status = $1 WHERE s3_key = $2",
158 "UPDATE content_insertions SET scan_status = $1 WHERE storage_key = $2",
159 "UPDATE items SET cover_scan_status = $1 WHERE cover_s3_key = $2",
160 "UPDATE projects SET cover_scan_status = $1 WHERE cover_s3_key = $2",
161 ] {
162 affected += sqlx::query(sql)
163 .bind(status)
164 .bind(s3_key)
165 .execute(&mut *tx)
166 .await?
167 .rows_affected();
168 }
169 tx.commit().await?;
170 Ok(affected)
171 }
172
173 /// Insert a scan result record for audit trail.
174 #[tracing::instrument(skip_all)]
175 pub async fn insert_scan_result(
176 db: &PgPool,
177 s3_key: &str,
178 result: &ScanResult,
179 ) -> Result<Uuid, sqlx::Error> {
180 let layers_json =
181 serde_json::to_value(&result.layers).unwrap_or_else(|_| serde_json::Value::Array(vec![]));
182
183 let id = sqlx::query_scalar::<_, Uuid>(
184 r"
185 INSERT INTO file_scan_results (s3_key, scan_status, scan_layers, sha256, file_size_bytes)
186 VALUES ($1, $2, $3, $4, $5)
187 RETURNING id
188 ",
189 )
190 .bind(s3_key)
191 .bind(result.status)
192 .bind(&layers_json)
193 .bind(&result.sha256)
194 .bind(result.file_size as i64)
195 .fetch_one(db)
196 .await?;
197
198 Ok(id)
199 }
200
201 /// Update the scan_status column on an item.
202 #[tracing::instrument(skip_all)]
203 pub async fn update_item_scan_status(
204 db: &PgPool,
205 item_id: ItemId,
206 status: FileScanStatus,
207 ) -> Result<(), sqlx::Error> {
208 sqlx::query(
209 r"
210 UPDATE items SET scan_status = $1, updated_at = NOW() WHERE id = $2
211 ",
212 )
213 .bind(status)
214 .bind(item_id)
215 .execute(db)
216 .await?;
217
218 Ok(())
219 }
220
221 /// Update the scan_status column on a version.
222 #[tracing::instrument(skip_all)]
223 pub async fn update_version_scan_status(
224 db: &PgPool,
225 version_id: VersionId,
226 status: FileScanStatus,
227 ) -> Result<(), sqlx::Error> {
228 sqlx::query(
229 r"
230 UPDATE versions SET scan_status = $1 WHERE id = $2
231 ",
232 )
233 .bind(status)
234 .bind(version_id)
235 .execute(db)
236 .await?;
237
238 Ok(())
239 }
240
241 /// Update the scan_status column on a media file.
242 #[tracing::instrument(skip_all)]
243 pub async fn update_media_file_scan_status(
244 db: &PgPool,
245 media_file_id: crate::db::MediaFileId,
246 status: FileScanStatus,
247 ) -> Result<(), sqlx::Error> {
248 let id: uuid::Uuid = media_file_id.into();
249 sqlx::query("UPDATE media_files SET scan_status = $1 WHERE id = $2")
250 .bind(status)
251 .bind(id)
252 .execute(db)
253 .await?;
254 Ok(())
255 }
256
257 /// Most recent content hash recorded for a key, from the scan-result audit
258 /// trail. The scan worker computes and stores the sha256 in `file_scan_results`
259 /// at scan time; the admin-approve promote path (which flips a held file to
260 /// Clean and must copy it to its content-addressed key) has only the staging
261 /// key in scope, so it reads the hash back here. Empty hashes, recorded by a
262 /// degraded/held scan that never fully hashed the object, are skipped, so a
263 /// content key is never derived from a blank digest.
264 #[tracing::instrument(skip_all)]
265 pub async fn latest_sha256_by_key(
266 db: &PgPool,
267 s3_key: &str,
268 ) -> Result<Option<String>, sqlx::Error> {
269 sqlx::query_scalar::<_, String>(
270 "SELECT sha256 FROM file_scan_results \
271 WHERE s3_key = $1 AND sha256 <> '' \
272 ORDER BY scanned_at DESC LIMIT 1",
273 )
274 .bind(s3_key)
275 .fetch_optional(db)
276 .await
277 }
278
279 /// Repoint a GATED entity (Item audio/video, Version, Media, OTA artifact) from
280 /// its staging key to the immutable content key and mark it Clean in one write.
281 ///
282 /// This is the scan-then-promote closing move for the kinds that carry their own
283 /// `scan_status` gate. The `(table, key column)` pair is a compile-time constant
284 /// selected by `(kind, file_type)`, never user input, so formatting it into the
285 /// statement is safe. Uses the runtime `query` (not the `!` macro) so extending
286 /// the promote set needs no offline-cache regeneration. Accepts any executor so
287 /// the caller can run it inside the same transaction as the staging-key delete
288 /// enqueue.
289 pub async fn promote_gated<'e>(
290 executor: impl sqlx::PgExecutor<'e>,
291 kind: crate::db::scan_jobs::ScanTargetKind,
292 file_type: crate::storage::FileType,
293 target_id: Uuid,
294 content_key: &str,
295 ) -> Result<(), sqlx::Error> {
296 use crate::db::scan_jobs::ScanTargetKind as K;
297 use crate::storage::FileType as F;
298 let sql: &'static str = match (kind, file_type) {
299 (K::Item, F::Audio) => {
300 "UPDATE items SET audio_s3_key = $1, scan_status = 'clean', updated_at = NOW() WHERE id = $2"
301 }
302 (K::Item, F::Video) => {
303 "UPDATE items SET video_s3_key = $1, scan_status = 'clean', updated_at = NOW() WHERE id = $2"
304 }
305 (K::Version, _) => "UPDATE versions SET s3_key = $1, scan_status = 'clean' WHERE id = $2",
306 (K::Media, _) => "UPDATE media_files SET s3_key = $1, scan_status = 'clean' WHERE id = $2",
307 (K::OtaArtifact, _) => {
308 "UPDATE ota_artifacts SET s3_key = $1, scan_status = 'clean' WHERE id = $2"
309 }
310 (other_kind, other_ft) => {
311 // A CDN-image kind (or an Item file role with no key column) must go
312 // through `promote_cdn_image_by_key` instead; reaching here is a wiring
313 // bug, not a data condition.
314 return Err(sqlx::Error::Protocol(format!(
315 "promote_gated: {other_kind:?}/{other_ft:?} is not a gated promote target"
316 )));
317 }
318 };
319 sqlx::query(sql)
320 .bind(content_key)
321 .bind(target_id)
322 .execute(executor)
323 .await?;
324 Ok(())
325 }
326
327 /// Repoint every CDN-served image surface that currently references `staging_key`
328 /// to the immutable `content_key` (and its rebuilt public `content_url`), marking
329 /// the row Clean. Mirrors [`set_cdn_image_scan_status_by_key`]'s surface list so
330 /// the promote and the status-stamp can't drift apart. Only the one surface whose
331 /// key matches is touched; the others no-op. `content_insertions` carries no URL
332 /// column (it is served presigned, not straight from the CDN) so only its key and
333 /// status move. Returns the number of rows repointed (expected: exactly 1).
334 pub async fn promote_cdn_image_by_key(
335 conn: &mut sqlx::PgConnection,
336 staging_key: &str,
337 content_key: &str,
338 content_url: &str,
339 ) -> Result<u64, sqlx::Error> {
340 let mut affected = 0u64;
341 // (SQL, binds_content_url), the surfaces with a materialized public URL take
342 // three binds (key, url, where-key); content_insertions takes two.
343 for (sql, has_url) in [
344 (
345 "UPDATE item_images SET s3_key = $1, image_url = $2, scan_status = 'clean' WHERE s3_key = $3",
346 true,
347 ),
348 (
349 "UPDATE project_images SET s3_key = $1, image_url = $2, scan_status = 'clean' WHERE s3_key = $3",
350 true,
351 ),
352 (
353 "UPDATE items SET cover_s3_key = $1, cover_image_url = $2, cover_scan_status = 'clean' WHERE cover_s3_key = $3",
354 true,
355 ),
356 (
357 "UPDATE projects SET cover_s3_key = $1, cover_image_url = $2, cover_scan_status = 'clean' WHERE cover_s3_key = $3",
358 true,
359 ),
360 (
361 "UPDATE content_insertions SET storage_key = $1, scan_status = 'clean' WHERE storage_key = $2",
362 false,
363 ),
364 ] {
365 let q = if has_url {
366 sqlx::query(sql)
367 .bind(content_key)
368 .bind(content_url)
369 .bind(staging_key)
370 } else {
371 sqlx::query(sql).bind(content_key).bind(staging_key)
372 };
373 affected += q.execute(&mut *conn).await?.rows_affected();
374 }
375 Ok(affected)
376 }
377
378 /// Get items held for review, joined with creator info + latest scan layers.
379 /// Oldest first.
380 #[tracing::instrument(skip_all)]
381 pub async fn get_held_items(db: &PgPool) -> Result<Vec<HeldItemRow>, sqlx::Error> {
382 let rows = sqlx::query_as::<_, HeldItemRow>(
383 r"
384 SELECT i.id AS item_id, i.title AS item_title,
385 COALESCE(i.audio_s3_key, i.cover_s3_key) AS s3_key,
386 u.username AS creator_username, u.id AS creator_id,
387 u.upload_trusted, i.updated_at AS held_at,
388 (
389 SELECT fsr.scan_layers FROM file_scan_results fsr
390 WHERE fsr.s3_key = COALESCE(i.audio_s3_key, i.cover_s3_key)
391 ORDER BY fsr.scanned_at DESC LIMIT 1
392 ) AS scan_layers
393 FROM items i
394 JOIN projects p ON p.id = i.project_id
395 JOIN users u ON u.id = p.user_id
396 WHERE i.scan_status = 'held_for_review'
397 ORDER BY i.updated_at ASC
398 LIMIT 1000
399 ",
400 )
401 .fetch_all(db)
402 .await?;
403
404 Ok(rows)
405 }
406
407 /// Get versions held for review, joined with creator info + latest scan layers.
408 /// Oldest first.
409 #[tracing::instrument(skip_all)]
410 pub async fn get_held_versions(db: &PgPool) -> Result<Vec<HeldVersionRow>, sqlx::Error> {
411 let rows = sqlx::query_as::<_, HeldVersionRow>(
412 r"
413 SELECT i.id AS item_id, i.title AS item_title,
414 v.id AS version_id, v.version_number,
415 v.s3_key,
416 u.username AS creator_username, u.id AS creator_id,
417 u.upload_trusted, v.created_at AS held_at,
418 (
419 SELECT fsr.scan_layers FROM file_scan_results fsr
420 WHERE fsr.s3_key = v.s3_key
421 ORDER BY fsr.scanned_at DESC LIMIT 1
422 ) AS scan_layers
423 FROM versions v
424 JOIN items i ON i.id = v.item_id
425 JOIN projects p ON p.id = i.project_id
426 JOIN users u ON u.id = p.user_id
427 WHERE v.scan_status = 'held_for_review'
428 ORDER BY v.created_at ASC
429 LIMIT 1000
430 ",
431 )
432 .fetch_all(db)
433 .await?;
434
435 Ok(rows)
436 }
437
438 /// Per-layer aggregate stats over a window for the admin dashboard.
439 #[derive(Debug, Clone, FromRow)]
440 pub struct LayerHealthRow {
441 pub layer: String,
442 pub pass_count: i64,
443 pub skip_count: i64,
444 pub fail_count: i64,
445 pub error_count: i64,
446 pub last_pass_or_skip: Option<DateTime<Utc>>,
447 }
448
449 /// Compute per-layer health stats over the last N hours.
450 ///
451 /// Reads `file_scan_results.scan_layers` JSONB and rolls up verdict counts
452 /// per layer. `last_pass_or_skip` is the most recent timestamp at which the
453 /// layer returned a non-error, non-fail verdict, the indicator the admin
454 /// panel uses to flag a layer as down.
455 #[tracing::instrument(skip_all)]
456 pub async fn layer_health_window(
457 db: &PgPool,
458 hours: i64,
459 ) -> Result<Vec<LayerHealthRow>, sqlx::Error> {
460 sqlx::query_as::<_, LayerHealthRow>(
461 r"
462 WITH expanded AS (
463 SELECT fsr.scanned_at,
464 (l ->> 'layer') AS layer,
465 (l ->> 'verdict') AS verdict
466 FROM file_scan_results fsr,
467 jsonb_array_elements(fsr.scan_layers) AS l
468 WHERE fsr.scanned_at > NOW() - ($1 || ' hours')::interval
469 )
470 SELECT layer,
471 COUNT(*) FILTER (WHERE verdict = 'pass') AS pass_count,
472 COUNT(*) FILTER (WHERE verdict = 'skip') AS skip_count,
473 COUNT(*) FILTER (WHERE verdict = 'fail') AS fail_count,
474 COUNT(*) FILTER (WHERE verdict = 'error') AS error_count,
475 MAX(scanned_at) FILTER (WHERE verdict IN ('pass', 'skip')) AS last_pass_or_skip
476 FROM expanded
477 GROUP BY layer
478 ORDER BY layer
479 ",
480 )
481 .bind(hours.to_string())
482 .fetch_all(db)
483 .await
484 }
485
486 /// A scan-history row for the dashboard's "Recent" grid.
487 #[derive(Debug, Clone, FromRow)]
488 pub struct ScanHistoryRow {
489 pub scanned_at: DateTime<Utc>,
490 pub s3_key: String,
491 pub scan_status: String,
492 pub sha256: Option<String>,
493 pub file_size_bytes: Option<i64>,
494 pub scan_layers: serde_json::Value,
495 }
496
497 /// Aggregate counts of entities currently in non-clean states. Used by the
498 /// PoM health endpoint to alert on growing review backlogs.
499 #[derive(Debug, Clone)]
500 pub struct HeldCounts {
501 pub held_versions: i64,
502 pub held_items: i64,
503 pub held_media: i64,
504 }
505
506 #[tracing::instrument(skip_all)]
507 pub async fn held_counts(db: &PgPool) -> Result<HeldCounts, sqlx::Error> {
508 let held_versions: i64 =
509 sqlx::query_scalar("SELECT COUNT(*) FROM versions WHERE scan_status = 'held_for_review'")
510 .fetch_one(db)
511 .await?;
512 let held_items: i64 =
513 sqlx::query_scalar("SELECT COUNT(*) FROM items WHERE scan_status = 'held_for_review'")
514 .fetch_one(db)
515 .await?;
516 let held_media: i64 = sqlx::query_scalar(
517 "SELECT COUNT(*) FROM media_files WHERE scan_status = 'held_for_review'",
518 )
519 .fetch_one(db)
520 .await?;
521 Ok(HeldCounts {
522 held_versions,
523 held_items,
524 held_media,
525 })
526 }
527
528 /// Recent scan results across all entities. Newest first, capped at `limit`.
529 /// Used by the Recent History collapsible section. `since_hours` bounds the
530 /// window so the grid renders fast.
531 #[tracing::instrument(skip_all)]
532 pub async fn recent_history(
533 db: &PgPool,
534 since_hours: i64,
535 limit: i64,
536 ) -> Result<Vec<ScanHistoryRow>, sqlx::Error> {
537 sqlx::query_as::<_, ScanHistoryRow>(
538 r"
539 SELECT fsr.scanned_at, fsr.s3_key, fsr.scan_status,
540 fsr.sha256, fsr.file_size_bytes, fsr.scan_layers
541 FROM file_scan_results fsr
542 WHERE fsr.scanned_at > NOW() - ($1 || ' hours')::interval
543 ORDER BY fsr.scanned_at DESC
544 LIMIT $2
545 ",
546 )
547 .bind(since_hours.to_string())
548 .bind(limit)
549 .fetch_all(db)
550 .await
551 }
552