Repository Public
1// Background jobs: antivirus scans, previews, portal submissions (quarantine → inspection → atomic placement),2// garbage collection and housekeeping.3import { createReadStream } from 'node:fs';4import { open, readdir, rm, stat, writeFile } from 'node:fs/promises';5import { join } from 'node:path';6import { extOf, previewKind } from '@spd/shared';7import { fileTypeFromFile } from 'file-type';8import { driver, ensureObject, materialize, type ObjectRow, readSealed, sealFile } from './blob.ts';9import { config } from './config.ts';10import { audit, type Db, lockOwner, notify, sql, SYSTEM, tx } from './db.ts';11import { enqueue, type Handler, requeueStuck } from './jobs.ts';12import { BLOCKED_EXT, type RequestRow } from './requests.ts';13import { clamdScan, listTar, listZip, officeToPdf, assertNotBomb, ScannerUnavailable, sheetPreview, sipsJpeg, ToolError, videoFrame, withWorkDir } from './tools.ts';14import { addFile, createFolder } from './tree.ts';15import { assertQuota } from './uploads.ts';1617// --- antivirus for content already in the drive (the owner's own uploads) ----------------------------------18const scanObject: Handler = async (p) => {19 const [o] = await sql<ObjectRow[]>`select *, size::text from storage_objects where id = ${String(p.objectId)}`;20 if (!o || o.scan_status !== 'pending') return;21 let r;22 try { r = await clamdScan(readSealed(o), Number(o.size)); } catch (e) {23 if (e instanceof ScannerUnavailable && config.avMode === 'optional') { await sql`update storage_objects set scan_status = 'error', scan_detail = 'scanner_unavailable', scanned_at = now() where id = ${o.id}`; return; }24 throw e; // retried with back-off; the object stays "pending"25 }26 await tx(async (t) => {27 await t`update storage_objects set scan_status = ${r.status}, scan_detail = ${r.status === 'infected' ? r.signature : r.status === 'error' ? r.detail : null}, scanned_at = now() where id = ${o.id}`;28 if (r.status === 'infected') {29 const files = await t<{ id: string; name: string }[]>`select distinct f.id, f.name from files f join file_versions v on v.file_id = f.id where v.object_id = ${o.id} limit 20`;30 await audit(t, SYSTEM, 'scan.infected', { ownerId: o.owner_id, targetType: 'object', targetId: o.id, meta: { signature: r.signature, files: files.map((f) => f.id) } });31 await notify(t, o.owner_id, 'scan.infected', { signature: r.signature, files });32 }33 });34};3536// --- previews ----------------------------------------------------------------------------------------------37type Kind = 'thumb' | 'image' | 'pdf' | 'sheet' | 'archive';38async function savePreview(objectId: string, kind: Kind, path: string): Promise<void> {39 const size = (await stat(path)).size;40 const sealed = await sealFile(path, size);41 const old = await sql<{ storage_key: string | null }[]>`select storage_key from previews where object_id = ${objectId} and kind = ${kind}`;42 try {43 await sql`insert into previews (object_id, kind, status, storage_key, wrapped_key, size) values (${objectId}, ${kind}, 'ready', ${sealed.storage_key}, ${sealed.wrapped_key}, ${size})44 on conflict (object_id, kind) do update set status = 'ready', storage_key = excluded.storage_key, wrapped_key = excluded.wrapped_key, size = excluded.size, error = null, updated_at = now()`;45 } catch (e) { await driver.remove(sealed.storage_key); throw e; } // the object was deleted meanwhile46 if (old[0]?.storage_key) await driver.remove(old[0].storage_key);47}48const markPreview = (objectId: string, kind: Kind, status: 'failed' | 'unsupported' | 'pending', error?: string) =>49 sql`insert into previews (object_id, kind, status, error) values (${objectId}, ${kind}, ${status}, ${error ?? null})50 on conflict (object_id, kind) do update set status = excluded.status, error = excluded.error, updated_at = now() where previews.status <> 'ready'`.catch(() => {});5152/** Messages safe to show to users for a failed preview. */53const reason = (e: unknown) => (e instanceof ToolError ? e.message : 'the preview could not be produced');54const preview: Handler = async (p) => {55 const objectId = String(p.objectId);56 const [o] = await sql<(ObjectRow & { name: string | null })[]>`select o.*, o.size::text, (select f.name from files f join file_versions v on v.file_id = f.id where v.object_id = o.id order by v.created_at limit 1) as name57 from storage_objects o where o.id = ${objectId}`;58 if (!o?.name) return;59 const kind = previewKind(o.name), size = Number(o.size), ext = extOf(o.name);60 const plan: Kind[] = kind === 'image' || kind === 'pdf' || kind === 'video' ? ['thumb'] : kind === 'heic' ? ['thumb', 'image'] : kind === 'office' ? ['pdf', 'thumb'] : kind === 'sheet' ? ['sheet'] : kind === 'archive' ? ['archive'] : [];61 if (!plan.length || size === 0) return;62 for (const k of plan) await markPreview(objectId, k, 'pending');63 const tooBig = (limit: number) => { for (const k of plan) void markPreview(objectId, k, 'failed', `file too large for a preview (limit ${Math.round(limit / 1024 ** 2)} MB)`); };64 if ((kind === 'office' && size > config.officeMaxBytes) || (kind === 'sheet' && size > config.sheetMaxBytes) || ((kind === 'image' || kind === 'heic') && size > 200 * 1024 ** 2)) return tooBig(kind === 'office' ? config.officeMaxBytes : kind === 'sheet' ? config.sheetMaxBytes : 200 * 1024 ** 2);65 await withWorkDir(async (work) => {66 const input = join(work, `input.${ext || 'bin'}`);67 await materialize(o, input);68 const each = async (k: Kind, fn: () => Promise<void>) => { try { await fn(); } catch (e) { if (!(e instanceof ToolError)) console.error(JSON.stringify({ level: 'error', msg: 'preview failed', object: objectId, kind: k, error: String((e as Error).message) })); await markPreview(objectId, k, 'failed', reason(e)); } };69 if (kind === 'image' || kind === 'heic' || kind === 'pdf') await each('thumb', async () => { const out = join(work, 'thumb.jpg'); await sipsJpeg(work, input, out, 480); await savePreview(objectId, 'thumb', out); });70 if (kind === 'heic') await each('image', async () => { const out = join(work, 'full.jpg'); await sipsJpeg(work, input, out, 2560); await savePreview(objectId, 'image', out); });71 if (kind === 'video') await each('thumb', async () => { const out = join(work, 'thumb.jpg'); if (await videoFrame(work, input, out)) await savePreview(objectId, 'thumb', out); else await markPreview(objectId, 'thumb', 'unsupported'); });72 if (kind === 'office') {73 let pdf: string | undefined;74 await each('pdf', async () => { if (['docx', 'pptx', 'odt', 'odp'].includes(ext)) await assertNotBomb(input); pdf = await officeToPdf(work, input); await savePreview(objectId, 'pdf', pdf); });75 if (pdf) await each('thumb', async () => { const out = join(work, 'thumb.jpg'); await sipsJpeg(work, pdf!, out, 480); await savePreview(objectId, 'thumb', out); });76 else await markPreview(objectId, 'thumb', 'unsupported');77 }78 if (kind === 'sheet') await each('sheet', async () => { const out = join(work, 'sheet-out.json'); await writeFile(out, JSON.stringify(await sheetPreview(work, input))); await savePreview(objectId, 'sheet', out); });79 if (kind === 'archive') await each('archive', async () => {80 const listing = ext === 'zip' ? await listZip(input) : await listTar(input, ext !== 'tar');81 const out = join(work, 'archive.json'); await writeFile(out, JSON.stringify(listing)); await savePreview(objectId, 'archive', out);82 });83 });84};8586// --- portal submissions ------------------------------------------------------------------------------------87const EXECUTABLE = new Set(['application/x-msdownload', 'application/x-mach-binary', 'application/x-elf', 'application/vnd.microsoft.portable-executable', 'application/x-apple-diskimage', 'application/java-archive', 'application/x-sh',88 'application/vnd.android.package-archive', 'application/x-msi', 'application/x-deb', 'application/x-rpm']);89/** Extensions that legitimately share one signature family. */90const FAMILY: Record<string, string> = { jpeg: 'jpg', jpe: 'jpg', tif: 'tiff', heif: 'heic', htm: 'html', mpg: 'mpeg', m4v: 'mp4', mov: 'mp4', m4a: 'mp4', qt: 'mp4', '3gp': 'mp4', oga: 'ogg', ogv: 'ogg', tgz: 'gz', docm: 'docx', xlsm: 'xlsx', pptm: 'pptx', cfb: 'doc', xls: 'doc', ppt: 'doc', msg: 'doc', zip: 'zip', jar: 'zip', epub: 'zip', odt: 'zip', ods: 'zip', odp: 'zip', docx: 'zip', xlsx: 'zip', pptx: 'zip', key: 'zip', pages: 'zip', numbers: 'zip' };91const fam = (e: string) => FAMILY[e] ?? e;92/** Does the content's signature agree with the name it was given? Returns a reject code, or null when acceptable. */93export async function inspectSignature(path: string, name: string): Promise<{ mime: string; reject: string | null }> {94 const ext = extOf(name);95 const head = Buffer.alloc(4);96 const fh = await open(path, 'r'); try { await fh.read(head, 0, 4, 0); } finally { await fh.close(); }97 if (head.subarray(0, 2).toString('latin1') === 'MZ' || head.subarray(0, 2).toString('latin1') === '#!' || head.equals(Buffer.from([0x7f, 0x45, 0x4c, 0x46]))98 || [0xfeedface, 0xfeedfacf, 0xcefaedfe, 0xcffaedfe, 0xcafebabe].includes(head.readUInt32BE(0))) return { mime: 'application/x-executable', reject: 'executable' };99 const t = await fileTypeFromFile(path).catch(() => undefined);100 if (!t) return { mime: 'application/octet-stream', reject: null }; // no known signature (text, CSV, unknown): the extension policy already applied101 if (EXECUTABLE.has(t.mime)) return { mime: t.mime, reject: 'executable' };102 // a recognised signature must belong to the family the extension announces (a .pdf that is really a ZIP is refused)103 const ok = fam(t.ext) === fam(ext) || (fam(ext) === 'zip' && fam(t.ext) === 'zip') || (t.mime.startsWith('text/') || t.mime === 'application/xml');104 return { mime: t.mime, reject: ok ? null : 'type_mismatch' };105}106107interface ItemRow { id: string; safe_name: string; size: string; sha256: string; quarantine_path: string; status: string; detected_mime: string | null; reject_code: string | null }108interface SubRow { id: string; request_id: string; public_ref: string; status: string; submitter_name: string; reference: string; created_at: Date }109const processSubmission: Handler = async (p) => {110 const subId = String(p.submissionId);111 const [sub] = await sql<SubRow[]>`select * from upload_submissions where id = ${subId}`;112 if (!sub || sub.status !== 'processing') return;113 const [portal] = await sql<RequestRow[]>`select * from upload_requests where id = ${sub.request_id}`;114 if (!portal) return;115 const items = await sql<ItemRow[]>`select *, size::text from upload_items where submission_id = ${subId} and status = 'quarantined' order by position, id`;116 const reject = async (i: ItemRow, code: string, meta: Record<string, unknown> = {}) => {117 await rm(i.quarantine_path, { force: true });118 await sql`update upload_items set status = 'rejected', reject_code = ${code}, quarantine_path = null, updated_at = now() where id = ${i.id}`;119 await audit(sql, SYSTEM, 'submission.item_rejected', { ownerId: portal.owner_id, targetType: 'submission', targetId: subId, meta: { item: i.id, name: i.safe_name, code, ...meta } });120 i.status = 'rejected';121 };122 // 1. inspection, file by file; nothing has touched the target folder yet123 for (const i of items) {124 const ext = extOf(i.safe_name);125 if (BLOCKED_EXT.has(ext) || (portal.allowed_extensions.length && !portal.allowed_extensions.includes(ext))) { await reject(i, 'type_not_allowed'); continue; }126 if (!(await stat(i.quarantine_path).then((s) => s.size === Number(i.size), () => false))) { await reject(i, 'incomplete'); continue; }127 const sig = await inspectSignature(i.quarantine_path, i.safe_name);128 await sql`update upload_items set detected_mime = ${sig.mime} where id = ${i.id}`;129 if (sig.reject) { await reject(i, sig.reject, { mime: sig.mime }); continue; }130 let scan;131 try { scan = await clamdScan(createReadStream(i.quarantine_path), Number(i.size)); } catch (e) {132 if (!(e instanceof ScannerUnavailable) || config.avMode === 'required') throw e; // stays in quarantine; the job is retried133 scan = { status: 'error' as const, detail: 'scanner_unavailable' };134 }135 if (scan.status === 'infected') { await reject(i, 'infected', { signature: scan.signature }); continue; }136 if (scan.status === 'error' && config.avMode === 'required') { await reject(i, 'unscannable', { detail: scan.detail }); continue; }137 (i as ItemRow & { scan?: string }).reject_code = null;138 (i as ItemRow & { clean?: boolean }).clean = scan.status === 'clean';139 }140 // 2. placement of everything that passed, in one transaction: the folder sees all accepted files or none141 const accepted = items.filter((i) => i.status === 'quarantined');142 const createdBlobs: string[] = [];143 try {144 await tx(async (t) => {145 await lockOwner(t, portal.owner_id);146 const [target] = await t<{ id: string }[]>`select id from folders where id = ${portal.folder_id} and trashed_at is null`;147 let outcome: 'completed' | 'rejected' = accepted.length ? 'completed' : 'rejected';148 let folderId: string | null = null, placedBytes = 0, failCode: string | null = null;149 if (accepted.length && !target) failCode = 'target_unavailable';150 if (accepted.length && target) {151 const total = accepted.reduce((n, i) => n + Number(i.size), 0);152 await assertQuota(t, portal.owner_id, total).catch(() => { failCode = 'quota_exceeded'; });153 }154 if (failCode) outcome = 'rejected';155 else if (accepted.length) {156 folderId = target!.id;157 if (portal.subfolder_per_submission) {158 const label = [sub.created_at.toISOString().slice(0, 10), sub.submitter_name || sub.reference, sub.public_ref].filter(Boolean).join(' ');159 folderId = (await createFolder(t, portal.owner_id, target!.id, label, 'rename')).id;160 }161 for (const i of accepted) {162 const [object, created] = await ensureObject(t, portal.owner_id, i.quarantine_path, Number(i.size), i.sha256);163 if (created) createdBlobs.push(object.storage_key);164 if ((i as ItemRow & { clean?: boolean }).clean && object.scan_status !== 'clean') { await t`update storage_objects set scan_status = 'clean', scanned_at = now() where id = ${object.id}`; object.scan_status = 'clean'; }165 const { file } = await addFile(t, { ownerId: portal.owner_id, folderId, name: i.safe_name, object, onConflict: 'rename', origin: 'request', actorId: null });166 await t`update upload_items set status = 'placed', file_id = ${file.id}, quarantine_path = null, updated_at = now() where id = ${i.id}`;167 placedBytes += Number(i.size);168 }169 await t`update upload_requests set bytes_received = bytes_received + ${placedBytes}, updated_at = now() where id = ${portal.id}`;170 }171 if (failCode) for (const i of accepted) await t`update upload_items set status = 'rejected', reject_code = ${failCode}, quarantine_path = null, updated_at = now() where id = ${i.id}`;172 await t`update upload_submissions set status = ${outcome}, completed_at = now(), folder_id = ${folderId} where id = ${subId}`;173 const rejected = items.length - (failCode ? 0 : accepted.length);174 await audit(t, SYSTEM, `submission.${outcome}`, { ownerId: portal.owner_id, targetType: 'submission', targetId: subId, meta: { placed: failCode ? 0 : accepted.length, rejected: failCode ? items.length : rejected, bytes: placedBytes, reason: failCode } });175 if (portal.notify_owner) await notify(t, portal.owner_id, 'request.submission', { requestId: portal.id, title: portal.title, submissionId: subId, ref: sub.public_ref, name: sub.submitter_name, placed: failCode ? 0 : accepted.length, rejected: failCode ? items.length : rejected, folderId, reason: failCode });176 });177 } catch (e) { for (const k of createdBlobs) await driver.remove(k); throw e; }178 for (const i of accepted) await rm(i.quarantine_path, { force: true });179};180181// --- garbage collection and housekeeping -------------------------------------------------------------------182/** Remove an owner's objects no version references any more, with their previews. */183const gcObjects: Handler = async (p) => {184 const ownerId = String(p.ownerId);185 const keys = await tx(async (t) => {186 await lockOwner(t, ownerId);187 const dead = await t<{ id: string; storage_key: string }[]>`select o.id, o.storage_key from storage_objects o where o.owner_id = ${ownerId} and not exists (select 1 from file_versions v where v.object_id = o.id)`;188 if (!dead.length) return [];189 const ids = dead.map((d) => d.id);190 const prev = await t<{ storage_key: string }[]>`select storage_key from previews where object_id = any(${ids}) and storage_key is not null`;191 await t`delete from storage_objects where id = any(${ids})`;192 return [...dead.map((d) => d.storage_key), ...prev.map((x) => x.storage_key)];193 });194 for (const k of keys) await driver.remove(k);195};196197/** Periodic clean-up: abandoned uploads, expired submissions, old sessions and challenges, stray temp files. */198export async function housekeeping(): Promise<void> {199 await requeueStuck();200 const stale = await sql<{ id: string; item_id: string | null }[]>`update uploads set status = 'aborted', updated_at = now() where status in ('active', 'completing') and (expires_at < now() or (status = 'completing' and updated_at < now() - interval '1 hour')) returning id, item_id`;201 for (const u of stale) await rm(join(u.item_id ? config.quarantineDir : config.tmpDir, `up-${u.id}`), { force: true });202 // submissions never finalised: their items are dropped, their reserved bytes released203 const dead = await sql<{ id: string }[]>`update upload_submissions set status = 'expired', completed_at = now() where status = 'open' and created_at < now() - interval '24 hours' returning id`;204 for (const s of dead) {205 const ups = await sql<{ id: string }[]>`select u.id from uploads u join upload_items i on i.id = u.item_id where i.submission_id = ${s.id}`;206 for (const u of ups) await rm(join(config.quarantineDir, `up-${u.id}`), { force: true });207 await sql`update upload_items set status = 'rejected', reject_code = 'expired', quarantine_path = null where submission_id = ${s.id} and status in ('uploading', 'quarantined')`;208 await sql`update uploads set status = 'aborted' where item_id in (select id from upload_items where submission_id = ${s.id}) and status <> 'done'`;209 }210 await sql`delete from uploads where status in ('done', 'aborted') and updated_at < now() - interval '7 days'`;211 await sql`delete from sessions where expires_at < now() - interval '30 days' or revoked_at < now() - interval '30 days'`;212 await sql`delete from auth_challenges where expires_at < now()`;213 await sql`delete from jobs where status = 'done' and finished_at < now() - interval '7 days'`;214 await sql`delete from invites where expires_at < now() - interval '30 days'`;215 // files in the temp areas that no live upload accounts for (crash leftovers)216 for (const dir of [config.tmpDir, config.quarantineDir, config.workDir]) {217 for (const name of await readdir(dir).catch(() => [])) {218 const path = join(dir, name);219 const st = await stat(path).catch(() => null);220 if (!st || Date.now() - st.mtimeMs < 48 * 3600_000) continue;221 const m = /^up-([0-9a-f-]{36})$/.exec(name);222 if (m) { const [live] = await sql`select 1 from uploads where id = ${m[1]!} and status in ('active', 'completing')`; const [q] = await sql`select 1 from upload_items where quarantine_path = ${path} and status = 'quarantined'`; if (live || q) continue; }223 await rm(path, { recursive: true, force: true });224 }225 }226 // submissions waiting for inspection whose job gave up (antivirus down for hours): try again227 const waiting = await sql<{ id: string }[]>`select id from upload_submissions where status = 'processing' and finalized_at < now() - interval '10 minutes'`;228 for (const s of waiting) await enqueue(sql, 'process_submission', { submissionId: s.id }, { dedupeKey: `submission:${s.id}`, maxAttempts: 30 });229 const owners = await sql<{ owner_id: string }[]>`select distinct owner_id from storage_objects o where not exists (select 1 from file_versions v where v.object_id = o.id)`;230 for (const o of owners) await enqueue(sql, 'gc_objects', { ownerId: o.owner_id }, { dedupeKey: `gc:${o.owner_id}` });231}232233export const handlers: Record<string, Handler> = { scan_object: scanObject, preview, process_submission: processSubmission, gc_objects: gcObjects };234export type { Db };235