fix(uploads): prevent cross-photo contamination from filename collisions and non-atomic writes (#931) (stable) (#934)

* fix(uploads): prevent cross-photo contamination from filename collisions and non-atomic writes (#931) * test: pin the suffixed photo filename format in the NFD pipeline suite (#931) * chore(deps): promote p-limit to a direct dependency for the watermark limiter (#931) * test: make the suffix-uniqueness check deterministic-in-practice (#931) * fix(uploads): widen the anti-collision suffix to 48 bits (#931) * fix(uploads): hide staging files from list() + share one watermark limiter process-wide (#931) * fix(uploads): reclaim orphaned staging files + revalidate watermark settings in queued jobs (#931) --------- Co-authored-by: Paul Nothaft <paul@MacStudio-von-Paul.local>
2026-08-01 12:29:23 +02:00
parent 7974b9c6d7
commit fc99e2b233
10 changed files with 246 additions and 35 deletions
@@ -1,12 +1,12 @@
 {
  "name": "picpeak-backend",
-  "version": "3.45.6",
+  "version": "3.45.10",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "picpeak-backend",
-      "version": "3.45.6",
+      "version": "3.45.10",
      "dependencies": {
        "@aws-sdk/client-s3": "^3.850.0",
        "@aws-sdk/lib-storage": "^3.850.0",
@@ -42,6 +42,7 @@
        "node-stream-zip": "^1.15.0",
        "nodemailer": "^9.0.1",
        "otplib": "^12.0.1",
+        "p-limit": "^3.1.0",
        "pdf-lib": "^1.17.1",
        "pdfkit": "^0.17.2",
        "pg": "^8.16.3",
@@ -9558,7 +9559,6 @@
      "version": "3.1.0",
      "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-3.1.0.tgz",
      "integrity": "sha512-TYOanM3wGwNGsZN2cVTYPArw454xnXj5qmWF1bEoAc4+cU/ol7GVh7odevjp1FNHduHc3KZMcFduxU5Xc6uJRQ==",
-      "dev": true,
      "license": "MIT",
      "dependencies": {
        "yocto-queue": "^0.1.0"
@@ -12590,7 +12590,6 @@
      "version": "0.1.0",
      "resolved": "https://registry.npmjs.org/yocto-queue/-/yocto-queue-0.1.0.tgz",
      "integrity": "sha512-rVksvsnNCdJ/ohGc6xgPwyN8eheCxsiLM8mxuE/t/mOVqJewPuO1miLpTHQiRgTKCLexL4MeAFVagts7HmNZ2Q==",
-      "dev": true,
      "license": "MIT",
      "engines": {
        "node": ">=10"
@@ -51,6 +51,7 @@
    "node-stream-zip": "^1.15.0",
    "nodemailer": "^9.0.1",
    "otplib": "^12.0.1",
+    "p-limit": "^3.1.0",
    "pdf-lib": "^1.17.1",
    "pdfkit": "^0.17.2",
    "pg": "^8.16.3",
@@ -315,9 +315,10 @@ router.post('/:eventId/upload', adminAuth, requirePermission('photos.upload'), r
    const crypto = require('crypto');
    const uploadId = crypto.randomBytes(16).toString('hex');

-    // Counter base — same approximation as before. Strict uniqueness is
-    // already enforced by the filename template + DB unique index, so a
-    // small race here just retries a counter on conflict (rare).
+    // Counter base — a per-request approximation (concurrent upload
+    // requests can compute the same base; there is NO unique index on
+    // photos.filename). Uniqueness of the final path comes from the
+    // random suffix inside generatePhotoFilename (#931).
    const existingCount = await db('photos')
      .where({ event_id: eventId, type: photoType })
      .count('id as count')
@@ -326,9 +326,11 @@ async function queueFilesForProcessing(files, options = {}) {

  if (fileList.length === 0) return { uploadId, photos: queued, errors };

-  // Counter base — same approximation the upload route used pre-async.
-  // Strict uniqueness is still enforced by the filename template; on a
-  // collision the worker would just fail one photo.
+  // Counter base — a per-request approximation (concurrent calls can
+  // compute the same base). Uniqueness of the final path comes from the
+  // random suffix inside generatePhotoFilename (#931) — before that
+  // suffix, a counter collision silently overwrote the first photo's
+  // bytes at its already-recorded path.
  const existingCount = await db('photos')
    .where({ event_id: eventId, type: photoType })
    .count('id as count')
@@ -6,6 +6,11 @@ const crypto = require('crypto');

 const logger = require('../../utils/logger');

+// Staging files older than this are considered orphaned by a crash between
+// copy and rename, and are reclaimed during list() walks. Generous enough
+// that no legitimate in-flight copy (even multi-GB on slow NFS) hits it.
+const STAGING_RECLAIM_AGE_MS = 60 * 60 * 1000;
+
 /**
 * Filesystem-backed implementation of the StorageBackend interface.
 * All keys are relative to `root` (typically process.env.STORAGE_PATH).
@@ -67,8 +72,18 @@ class LocalFsStorage {
  async putFromFile(relPath, localPath, _options = {}) {
    const abs = this._resolve(relPath);
    await fsp.mkdir(path.dirname(abs), { recursive: true });
-    // copyFile is atomic from the destination's perspective on POSIX.
-    await fsp.copyFile(localPath, abs);
+    // copyFile truncates and rewrites the destination in place, so a
+    // concurrent reader (thumbnail/watermark generation, photo serving)
+    // can observe partial or foreign bytes mid-copy (#931). Copy to a
+    // sibling tmp file and rename, like put() above — rename IS atomic.
+    const tmp = `${abs}.tmp.${process.pid}.${crypto.randomBytes(4).toString('hex')}`;
+    try {
+      await fsp.copyFile(localPath, tmp);
+      await fsp.rename(tmp, abs);
+    } catch (err) {
+      await fsp.unlink(tmp).catch(() => {});
+      throw err;
+    }
  }

  async get(relPath) {
@@ -126,6 +141,23 @@ class LocalFsStorage {
        throw err;
      }
      for (const ent of dirents) {
+        // Hide in-flight staging files (put/putFromFile write `<key>.tmp.<pid>.<hex>`
+        // siblings before the atomic rename). Without this filter a
+        // concurrent archive/backup listing could stream a partial tmp
+        // entry or fail when the rename wins the race (#931). Stale ones
+        // (a crash between copy and rename orphans them) are reclaimed
+        // here — hiding without reclaiming would let interrupted uploads
+        // accumulate invisible files until the volume fills.
+        if (/\.tmp\.\d+\.[0-9a-f]+$/.test(ent.name)) {
+          const childAbs = path.join(dir, ent.name);
+          try {
+            const st = await fsp.stat(childAbs);
+            if (Date.now() - st.mtimeMs > STAGING_RECLAIM_AGE_MS) {
+              await fsp.unlink(childAbs).catch(() => {});
+            }
+          } catch { /* vanished (rename/cleanup won the race) — fine */ }
+          continue;
+        }
        const childAbs = path.join(dir, ent.name);
        const childRel = relBase ? `${relBase}/${ent.name}` : ent.name;
        if (ent.isDirectory()) {
@@ -0,0 +1,130 @@
+/**
+ * #931 — LocalFsStorage.putFromFile must be atomic. The old implementation
+ * used fs.copyFile straight onto the destination, which truncates and
+ * rewrites in place: a concurrent reader (thumbnail/watermark generation,
+ * photo serving) could observe partial or foreign bytes mid-copy. The fix
+ * copies to a sibling tmp file and renames, like put() always did.
+ */
+const fs = require('fs');
+const fsp = require('fs/promises');
+const os = require('os');
+const path = require('path');
+
+const LocalFsStorage = require('../LocalFsStorage');
+
+describe('LocalFsStorage.putFromFile', () => {
+  let root;
+  let srcDir;
+  let storage;
+
+  beforeEach(async () => {
+    root = await fsp.mkdtemp(path.join(os.tmpdir(), 'picpeak-lfs-root-'));
+    srcDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'picpeak-lfs-src-'));
+    storage = new LocalFsStorage({ root });
+  });
+
+  afterEach(async () => {
+    await fsp.rm(root, { recursive: true, force: true });
+    await fsp.rm(srcDir, { recursive: true, force: true });
+  });
+
+  it('writes the source bytes to the destination key', async () => {
+    const src = path.join(srcDir, 'a.jpg');
+    await fsp.writeFile(src, Buffer.from('photo-a-bytes'));
+
+    await storage.putFromFile('events/active/ev/a.jpg', src);
+
+    const out = await fsp.readFile(path.join(root, 'events/active/ev/a.jpg'));
+    expect(out.toString()).toBe('photo-a-bytes');
+  });
+
+  it('leaves no tmp files behind after a successful write', async () => {
+    const src = path.join(srcDir, 'a.jpg');
+    await fsp.writeFile(src, Buffer.from('photo-a-bytes'));
+
+    await storage.putFromFile('events/active/ev/a.jpg', src);
+
+    const entries = await fsp.readdir(path.join(root, 'events/active/ev'));
+    expect(entries).toEqual(['a.jpg']);
+  });
+
+  it('leaves no tmp files behind when the source is missing', async () => {
+    await expect(
+      storage.putFromFile('events/active/ev/missing.jpg', path.join(srcDir, 'nope.jpg'))
+    ).rejects.toThrow();
+
+    const entries = await fsp.readdir(path.join(root, 'events/active/ev')).catch(() => []);
+    expect(entries.filter((e) => e.includes('.tmp.'))).toEqual([]);
+  });
+
+  it('hides in-flight staging files from list()', async () => {
+    const src = path.join(srcDir, 'a.jpg');
+    await fsp.writeFile(src, Buffer.from('photo-a-bytes'));
+    await storage.putFromFile('events/active/ev/a.jpg', src);
+
+    // Simulate a concurrent writer's staging file: archiveEvent lists
+    // this exact prefix and must never see (stream/delete) it.
+    await fsp.writeFile(
+      path.join(root, 'events/active/ev/b.jpg.tmp.12345.deadbeef'),
+      Buffer.from('partial')
+    );
+
+    const keys = (await storage.list('events/active/ev')).map((e) => e.key ?? e);
+    expect(JSON.stringify(keys)).toContain('a.jpg');
+    expect(JSON.stringify(keys)).not.toContain('.tmp.');
+  });
+
+  it('reclaims stale orphaned staging files during list()', async () => {
+    const dir = path.join(root, 'events/active/ev');
+    await fsp.mkdir(dir, { recursive: true });
+    const fresh = path.join(dir, 'f.jpg.tmp.111.aaaaaaaa');
+    const stale = path.join(dir, 's.jpg.tmp.222.bbbbbbbb');
+    await fsp.writeFile(fresh, Buffer.from('in-flight'));
+    await fsp.writeFile(stale, Buffer.from('orphaned'));
+    // Age the "stale" one past the reclaim threshold (1h).
+    const old = new Date(Date.now() - 2 * 60 * 60 * 1000);
+    await fsp.utimes(stale, old, old);
+
+    await storage.list('events/active/ev');
+
+    // Fresh in-flight staging survives (a live copy may still rename it);
+    // the crash orphan is gone.
+    await expect(fsp.stat(fresh)).resolves.toBeDefined();
+    await expect(fsp.stat(stale)).rejects.toThrow();
+  });
+
+  it('never exposes a partially written destination (tmp+rename atomicity)', async () => {
+    // A large-ish payload so the copy is not a single instantaneous block.
+    const big = Buffer.alloc(8 * 1024 * 1024, 0xab);
+    const src = path.join(srcDir, 'big.bin');
+    await fsp.writeFile(src, big);
+
+    const key = 'events/active/ev/big.bin';
+    const dest = path.join(root, key);
+
+    // Poll the destination while the copy runs: it must either not exist
+    // yet or already have the full size — never an in-between truncated
+    // state (which is exactly what in-place copyFile produced).
+    const observed = [];
+    const poller = (async () => {
+      for (let i = 0; i < 200; i++) {
+        try {
+          const st = fs.statSync(dest);
+          observed.push(st.size);
+        } catch {
+          // not there yet — fine
+        }
+        await new Promise((r) => setImmediate(r));
+      }
+    })();
+
+    await storage.putFromFile(key, src);
+    await poller;
+
+    for (const size of observed) {
+      expect(size).toBe(big.length);
+    }
+    const out = await fsp.stat(dest);
+    expect(out.size).toBe(big.length);
+  });
+});
@@ -8,6 +8,7 @@
 * - Tracking regeneration progress
 */

+const pLimit = require('p-limit');
 const { db } = require('../database/db');
 const watermarkService = require('./watermarkService');
 const { resolvePhotoStorageKey, resolvePhotoFilePath } = require('./photoResolver');
@@ -22,6 +23,13 @@ class WatermarkGeneratorService {
    this.batchSize = 10;
    // Concurrent processing limit
    this.concurrentLimit = 2;
+    // ONE process-wide limiter for every sharp pipeline this service
+    // spawns (#931). Per-invocation limiters would stack: overlapping
+    // regenerateAll/generateForEvent calls each brought their own cap,
+    // and the fire-and-forget generateForPhoto side-effect (one per
+    // uploaded photo) had no cap at all — a 363-photo bulk upload could
+    // decode 363 full-resolution images concurrently (#628 OOM class).
+    this.limit = pLimit(this.concurrentLimit);
  }

  /**
@@ -62,11 +70,21 @@ class WatermarkGeneratorService {
      // (external reference mode). watermarkService needs a local file path.
      const event = { slug: photo.slug, source_mode: photo.source_mode, external_path: photo.external_path };
      const storageKey = resolvePhotoStorageKey(event, photo);
-      const result = storageKey
-        ? await withLocalCopy(storageKey, (lp) =>
-          watermarkService.generateAndSaveWatermark(photo, lp, settings)
-        )
-        : await watermarkService.generateAndSaveWatermark(photo, resolvePhotoFilePath(event, photo), settings);
+      const result = await this.limit(async () => {
+        // Revalidate inside the limited slot: a long queue (bulk upload)
+        // can hold this job for minutes, during which an admin may disable
+        // watermarking — running with the captured settings would recreate
+        // files AFTER clearAllWatermarks() wiped them (#931 round 3).
+        const fresh = await watermarkService.getWatermarkSettings();
+        if (!fresh || !fresh.enabled) {
+          return { success: false, watermarkPath: null, error: 'Watermarking is disabled' };
+        }
+        return storageKey
+          ? withLocalCopy(storageKey, (lp) =>
+            watermarkService.generateAndSaveWatermark(photo, lp, fresh)
+          )
+          : watermarkService.generateAndSaveWatermark(photo, resolvePhotoFilePath(event, photo), fresh);
+      });

      if (result.success) {
        // Update database with watermark path
@@ -122,7 +140,11 @@ class WatermarkGeneratorService {
        return { ...results, errors: ['Watermarking is disabled'] };
      }

-      // Process in batches
+      // Process in batches. processPhotoWatermark routes every sharp
+      // pipeline through the shared instance limiter — a bare Promise.all
+      // over the batch ran all 10 at once, decoding 10 full-resolution
+      // images simultaneously (#931; same OOM class as #628 in the
+      // thumbnail path).
      for (let i = 0; i < photos.length; i += this.batchSize) {
        const batch = photos.slice(i, i + this.batchSize);

@@ -168,11 +190,21 @@ class WatermarkGeneratorService {
    try {
      const event = { slug: photo.slug, source_mode: photo.source_mode, external_path: photo.external_path };
      const storageKey = resolvePhotoStorageKey(event, photo);
-      const result = storageKey
-        ? await withLocalCopy(storageKey, (lp) =>
-          watermarkService.generateAndSaveWatermark(photo, lp, settings)
-        )
-        : await watermarkService.generateAndSaveWatermark(photo, resolvePhotoFilePath(event, photo), settings);
+      const result = await this.limit(async () => {
+        // Same revalidation as generateForPhoto: batch jobs queue for a
+        // long time, and a disable mid-run must not recreate files after
+        // clearAllWatermarks(). The batch's `settings` snapshot is still
+        // used for rendering; only the enabled gate is rechecked.
+        const fresh = await watermarkService.getWatermarkSettings();
+        if (!fresh || !fresh.enabled) {
+          return { success: false, watermarkPath: null, error: 'Watermarking is disabled' };
+        }
+        return storageKey
+          ? withLocalCopy(storageKey, (lp) =>
+            watermarkService.generateAndSaveWatermark(photo, lp, settings)
+          )
+          : watermarkService.generateAndSaveWatermark(photo, resolvePhotoFilePath(event, photo), settings);
+      });

      if (result.success) {
        await db('photos')
@@ -235,7 +267,8 @@ class WatermarkGeneratorService {

      logger.info(`Starting watermark regeneration for ${photos.length} photos`);

-      // Process in batches
+      // Process in batches, capped at concurrentLimit parallel sharp
+      // pipelines via the shared instance limiter (see generateForEvent).
      for (let i = 0; i < photos.length; i += this.batchSize) {
        // Check if job was cancelled
        if (!this.activeJobs.has(jobId)) {
@@ -89,23 +89,25 @@ describe('sanitizeFilename — edge cases', () => {
 });

 describe('generatePhotoFilename — composed name uses the NFD pipeline', () => {
-  it('round-trips Ägypten + individual → Agypten_individual_0050.jpg (#607)', () => {
+  // The trailing _[0-9a-f]{12} is the anti-collision suffix (#931) that
+  // keeps concurrent uploads from assigning the same final storage path.
+  it('round-trips Ägypten + individual → Agypten_individual_0050 (#607)', () => {
    expect(generatePhotoFilename('Ägypten', 'individual', 50, '.jpg'))
-      .toBe('Agypten_individual_0050.jpg');
+      .toMatch(/^Agypten_individual_0050_[0-9a-f]{12}\.jpg$/);
  });

  it('handles missing category by defaulting to "uncategorized"', () => {
    expect(generatePhotoFilename('Wedding', null, 1, '.jpg'))
-      .toBe('Wedding_uncategorized_0001.jpg');
+      .toMatch(/^Wedding_uncategorized_0001_[0-9a-f]{12}\.jpg$/);
  });

  it('zero-pads the counter to 4 digits', () => {
-    expect(generatePhotoFilename('e', 'c', 7, '.png')).toBe('e_c_0007.png');
-    expect(generatePhotoFilename('e', 'c', 1234, '.png')).toBe('e_c_1234.png');
-    // 5+ digit counters intentionally overflow the pad — pinned because
-    // the unique index in the photos table doesn't care about pad width,
-    // only string uniqueness.
-    expect(generatePhotoFilename('e', 'c', 99999, '.png')).toBe('e_c_99999.png');
+    expect(generatePhotoFilename('e', 'c', 7, '.png')).toMatch(/^e_c_0007_[0-9a-f]{12}\.png$/);
+    expect(generatePhotoFilename('e', 'c', 1234, '.png')).toMatch(/^e_c_1234_[0-9a-f]{12}\.png$/);
+    // 5+ digit counters intentionally overflow the pad — pad width never
+    // mattered for uniqueness (there is no unique index on filenames);
+    // the random suffix is what guarantees it.
+    expect(generatePhotoFilename('e', 'c', 99999, '.png')).toMatch(/^e_c_99999_[0-9a-f]{12}\.png$/);
  });
 });

@@ -1,4 +1,5 @@
 const path = require('path');
+const crypto = require('crypto');

 /**
 * Sanitize a string to be used as a filename component
@@ -59,8 +60,18 @@ function generatePhotoFilename(eventName, categoryName, counter, extension) {
  const sanitizedEvent = sanitizeFilename(eventName, 30);
  const sanitizedCategory = sanitizeFilename(categoryName || 'uncategorized', 20);
  const paddedCounter = String(counter).padStart(4, '0');
-  
-  return `${sanitizedEvent}_${sanitizedCategory}_${paddedCounter}${extension}`;
+  // Random suffix (#931): the counter base is `count(*)+1` computed per
+  // upload request, so two concurrent bulk-upload requests can assign the
+  // same counter to different photos. Since files are written to their
+  // final path before any row exists (and photos has no unique index on
+  // filename — one can't be added without a dedupe migration on installs
+  // that already carry historical duplicates), a collision silently
+  // overwrites the first photo's bytes at its recorded path — cross-photo
+  // contamination. 48 bits keep the collision odds negligible even for
+  // pathological concurrency (two simultaneous 2000-photo uploads: ~7e-12).
+  const suffix = crypto.randomBytes(6).toString('hex');
+
+  return `${sanitizedEvent}_${sanitizedCategory}_${paddedCounter}_${suffix}${extension}`;
 }

 /**