node:fs: keep ArrayBuffer storage alive across worker.terminate() during async write

src/jsc/bindings/bindings.cpp

@@ -3244,6 +3244,10 @@ CPP_DECL bool JSC__JSValue__pinArrayBuffer(JSC::EncodedJSValue v)
     if (auto* buf = arrayBufferImpl(JSC::JSValue::decode(v))) {
         if (!buf->isShared())
             buf->pin();
+        // pin() only blocks transfer(); the native ref is what keeps the
+        // contents alive past Heap::lastChanceToFinalize. Balanced by
+        // unpinArrayBuffer; JS thread only (DeferrableRefCounted is not atomic).
+        buf->ref();
         return true;
     }
     return false;
@@ -3253,6 +3257,7 @@ CPP_DECL void JSC__JSValue__unpinArrayBuffer(JSC::EncodedJSValue v)
     if (auto* buf = arrayBufferImpl(JSC::JSValue::decode(v))) {
         if (!buf->isShared())
             buf->unpin();
+        buf->deref();
     }
 }
 
@@ -3295,6 +3300,7 @@ CPP_DECL int32_t JSC__JSValue__borrowBytesForOffThread(JSC::EncodedJSValue v, co
         if (!buf) return 0;
         if (!buf->isShared())
             buf->pin();
+        buf->ref();
         *out_ptr = static_cast<const uint8_t*>(view->vector());
         *out_len = view->byteLength();
         return 2;
@@ -3304,6 +3310,7 @@ CPP_DECL int32_t JSC__JSValue__borrowBytesForOffThread(JSC::EncodedJSValue v, co
         if (!buf || buf->isDetached()) return 0;
         if (!buf->isShared())
             buf->pin();
+        buf->ref();
         *out_ptr = static_cast<const uint8_t*>(buf->data());
         *out_len = buf->byteLength();
         return 2;
@@ -6593,6 +6600,7 @@ extern "C" int32_t Bun__JSArray__collectBufferSpans(
                 return 2;
             if (!buf->isShared())
                 buf->pin();
+            buf->ref();
         }
         append(ctx, JSC::JSValue::encode(view), view->vector(), view->byteLength());
     }

test/js/node/fs/fs-read-worker-terminate.test.ts

@@ -0,0 +1,91 @@
+import { describe, expect, test } from "bun:test";
+import { bunEnv, bunExe, isASAN, isLinux, tempDir } from "harness";
+
+// worker.terminate() while an async node:fs read into a user Buffer is parked
+// in read(2) on an empty FIFO: Heap::lastChanceToFinalize would free the
+// ArrayBufferContents while the kernel still holds the destination address,
+// so a later write to the FIFO copies the peer's bytes into freed (and
+// possibly reused) memory. The kernel's copy_to_user is invisible to ASAN,
+// so the oracle is a direct address probe: the worker reports ptr(buf) before
+// parking, the parent terminates it, and bun:ffi's read.u8 at that address
+// either observes the worker's fill byte (storage alive) or trips ASAN
+// heap-use-after-free (storage freed mid-read). The FIFO is never written,
+// so read(2) stays parked and the separate completion-into-dead-VM crash
+// (#34154) is not reached.
+describe.concurrent.skipIf(!isLinux || !isASAN)(
+  "worker.terminate() during parked async node:fs read keeps the destination buffer alive",
+  () => {
+    for (const api of ["read", "readv"] as const) {
+      test(
+        `fs.${api}`,
+        async () => {
+          const fixture = `
+import { Worker } from "node:worker_threads";
+import fs from "node:fs";
+import path from "node:path";
+import { execFileSync } from "node:child_process";
+import { read as ffiRead } from "bun:ffi";
+
+const dir = process.argv[2];
+const API = process.argv[3];
+const fifo = path.join(dir, "fifo");
+execFileSync("mkfifo", [fifo]);
+// O_RDWR on a FIFO (Linux) never blocks on open and never delivers EOF, so
+// the worker's read(2) parks on an empty pipe for as long as we need.
+const fd = fs.openSync(fifo, fs.constants.O_RDWR);
+
+const src =
+  "import { parentPort, workerData as d } from 'node:worker_threads';\\n" +
+  "import fs from 'node:fs';\\n" +
+  "import { ptr } from 'bun:ffi';\\n" +
+  "const buf = Buffer.allocUnsafeSlow(8 << 20).fill(0x42);\\n" +
+  "if (d.api === 'read') fs.read(d.fd, buf, 0, buf.length, null, () => {});\\n" +
+  "else fs.readv(d.fd, [buf.subarray(0, 4 << 20), buf.subarray(4 << 20)], null, () => {});\\n" +
+  "parentPort.postMessage({ addr: ptr(buf), len: buf.length });\\n";
+const wfile = path.join(dir, "w.mjs");
+fs.writeFileSync(wfile, src);
+
+const w = new Worker(wfile, { workerData: { fd, api: API } });
+w.on("error", e => { console.log("WORKER_ERR", String(e)); process.exit(1); });
+const { addr, len } = await new Promise(r => w.once("message", r));
+await w.terminate();
+// Probe both ends of the range the pool thread handed to read(2). Without
+// the native ref this is the first access after lastChanceToFinalize freed
+// the storage and ASAN aborts with heap-use-after-free.
+const head = ffiRead.u8(addr, 0);
+const tail = ffiRead.u8(addr, len - 1);
+if (head !== 0x42 || tail !== 0x42) {
+  console.log("FAIL saw 0x" + head.toString(16) + "/0x" + tail.toString(16) + " at freed destination");
+  process.exit(1);
+}
+console.log("PASS addr=0x" + addr.toString(16) + " len=" + len);
+process.exit(0);
+`;
+
+          using dir = tempDir("fs-read-term", { "fixture.mjs": fixture });
+          await using proc = Bun.spawn({
+            cmd: [bunExe(), "--smol", "fixture.mjs", String(dir), api],
+            env: {
+              ...bunEnv,
+              // Route JSC ArrayBuffer storage through system malloc so ASAN
+              // owns the allocation and the ffi probe can observe the free.
+              Malloc: "1",
+              // The fail-before ASAN report is the signal; symbolization adds
+              // several seconds for no information the assertion uses.
+              ASAN_OPTIONS: (bunEnv.ASAN_OPTIONS ? bunEnv.ASAN_OPTIONS + ":" : "") + "symbolize=0",
+            },
+            cwd: String(dir),
+            stdout: "pipe",
+            stderr: "pipe",
+          });
+          const [stdout, stderr] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+          expect(stderr).not.toContain("AddressSanitizer");
+          expect(stdout + stderr).not.toContain("FAIL");
+          expect(stdout).toMatch(/^PASS addr=0x[0-9a-f]+ len=8388608$/m);
+        },
+        20_000,
+      );
+    }
+  },
+);

test/js/node/fs/fs-write-worker-terminate.test.ts

@@ -0,0 +1,101 @@
+import { describe, expect, test } from "bun:test";
+import { bunEnv, bunExe, isASAN, isWindows, tempDir } from "harness";
+
+// worker.terminate() while an async node:fs write of a user Buffer is still
+// inside write(2) on a pool thread: Heap::lastChanceToFinalize would free the
+// ArrayBufferContents while the kernel is still copying from it, so the fd
+// receives freed-heap bytes instead of the bytes the writer held. The reader
+// is the kernel, so the oracle is content: the worker fills its buffer with
+// one known byte and writes only to one FIFO; the parent flags any byte that
+// writer never held.
+describe.concurrent.skipIf(isWindows || !isASAN)(
+  "worker.terminate() during in-flight async node:fs write does not write freed memory",
+  () => {
+    for (const api of ["write", "writev", "writeFile"] as const) {
+      test(`fs.${api}`, async () => {
+        const fixture = `
+import { Worker } from "node:worker_threads";
+import fs from "node:fs";
+import path from "node:path";
+import { execFileSync } from "node:child_process";
+
+const dir = process.argv[2];
+const API = process.argv[3];
+const fifo = path.join(dir, "fifo");
+execFileSync("mkfifo", [fifo]);
+const rfd = fs.openSync(fifo, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK);
+
+const src =
+  "import { parentPort, workerData as d } from 'node:worker_threads';\\n" +
+  "import fs from 'node:fs';\\n" +
+  // 8 MiB malloc-backed Buffer filled with this worker's unique byte
+  "const buf = Buffer.allocUnsafe(8 << 20).fill(d.gen);\\n" +
+  "let fd;\\n" +
+  "async function lane() { for (;;) {\\n" +
+  "  if (d.api === 'write') await new Promise(r => fs.write(fd ??= fs.openSync(d.fifo, 'w'), buf, 0, buf.length, null, r));\\n" +
+  "  else if (d.api === 'writev') await new Promise(r => fs.writev(fd ??= fs.openSync(d.fifo, 'w'), [buf.subarray(0, 4 << 20), buf.subarray(4 << 20)], null, r));\\n" +
+  "  else if (d.api === 'writeFile') await new Promise(r => fs.writeFile(d.fifo, buf, r));\\n" +
+  "} }\\n" +
+  "parentPort.postMessage('up'); lane().catch(() => {});\\n";
+const wfile = path.join(dir, "w.mjs");
+fs.writeFileSync(wfile, src);
+
+const GEN = 0x42;
+const chunk = Buffer.alloc(1 << 16);
+let foreign = 0;
+function drain() {
+  let n;
+  try { n = fs.readSync(rfd, chunk, 0, chunk.length, null); } catch { return false; }
+  if (n <= 0) return false;
+  for (let j = 0; j < n; j++) {
+    if (chunk[j] !== GEN) {
+      let k = j; while (k < n && chunk[k] === chunk[j]) k++;
+      if (++foreign <= 4)
+        console.log("FOREIGN fifo byte 0x" + chunk[j].toString(16) + " run=" + (k - j) + " writer-held=0x42");
+      j = k - 1;
+    }
+  }
+  return true;
+}
+
+const w = new Worker(wfile, { workerData: { fifo, gen: GEN, api: API } });
+w.on("error", () => {});
+await Promise.race([new Promise(res => w.once("message", res)), Bun.sleep(5000)]);
+// Let write(2) park on the full pipe, then tear the VM down while the kernel
+// is still copying from the buffer.
+await Bun.sleep(30);
+await w.terminate();
+// Drain enough of what the in-flight write(2) streamed to observe the bytes
+// it produced after the VM died. Stop well short of the full 8 MiB so the
+// pool thread stays parked inside write(2) and never reaches the dead-VM
+// completion path (that crash is tracked separately).
+let idle = 0, reads = 0;
+while (idle < 20 && reads < 48) { if (drain()) { idle = 0; reads++; } else { idle++; await Bun.sleep(5); } }
+console.log(foreign ? "FAIL foreign-chunks=" + foreign : "PASS reads=" + reads);
+process.exit(foreign ? 1 : 0);
+`;
+
+        using dir = tempDir("fs-write-term", { "fixture.mjs": fixture });
+        await using proc = Bun.spawn({
+          cmd: [bunExe(), "--smol", "fixture.mjs", String(dir), api],
+          env: {
+            ...bunEnv,
+            // Route JSC ArrayBuffer storage through system malloc so ASAN
+            // free-fills it on release (bmalloc/Gigacage hides the UAF).
+            Malloc: "1",
+            ASAN_OPTIONS: (bunEnv.ASAN_OPTIONS ? bunEnv.ASAN_OPTIONS + ":" : "") + "max_free_fill_size=268435456",
+          },
+          cwd: String(dir),
+          stdout: "pipe",
+          stderr: "pipe",
+        });
+        const [stdout, stderr] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+        expect(stdout + stderr).not.toContain("FOREIGN");
+        // Prove the oracle reached its verdict and drained at least one chunk;
+        // a crash or worker-spawn failure before this point must fail the test.
+        expect(stdout).toMatch(/^PASS reads=[1-9]\d*$/m);
+      }, 30_000);
+    }
+  },
+);