diff --git a/src/jsc/bindings/bindings.cpp b/src/jsc/bindings/bindings.cpp index d99045a75ad4..d58ceb53b30e 100644 --- a/src/jsc/bindings/bindings.cpp +++ b/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(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(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()); } diff --git a/test/js/node/fs/fs-read-worker-terminate.test.ts b/test/js/node/fs/fs-read-worker-terminate.test.ts new file mode 100644 index 000000000000..ccc9ce8b418b --- /dev/null +++ b/test/js/node/fs/fs-read-worker-terminate.test.ts @@ -0,0 +1,88 @@ +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", + // detect_leaks=0: Malloc=1 surfaces unrelated pre-existing worker + // teardown leaks to LSan; the oracle here is the ffi probe above, + // not leak accounting. symbolize=0: the fail-before report is the + // signal and symbolization adds seconds for nothing we assert on. + ASAN_OPTIONS: (bunEnv.ASAN_OPTIONS ? bunEnv.ASAN_OPTIONS + ":" : "") + "symbolize=0:detect_leaks=0", + }, + 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("FAIL"); + expect(stdout).toMatch(/^PASS addr=0x[0-9a-f]+ len=8388608$/m); + }, 20_000); + } + }, +); diff --git a/test/js/node/fs/fs-write-worker-terminate.test.ts b/test/js/node/fs/fs-write-worker-terminate.test.ts new file mode 100644 index 000000000000..c63a61bf7dae --- /dev/null +++ b/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); + } + }, +);