From c677ac678b70dadf9fa1f58685b872a818199f93 Mon Sep 17 00:00:00 2001 From: Harris Hancock Date: Fri, 7 Aug 2026 17:29:53 +0100 Subject: [PATCH] Re-land the fix and regression test for vuln-187 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A draining read delivers its result to pumpToImpl through a kj event, and runs the stream's pull() callback along the way. A pull() that aborts the request's signal makes the canceler destroy the pump's coroutine frame from inside that event, so the event is destroyed while still firing and kj aborts the process with "Promise callback destroyed itself". Neither isolate-lock round trip in the pump is awaited directly any more. Each is handed to the IoContext's task set, and the pump awaits an unrelated fulfiller, so dropping the pump destroys only that await and leaves the firing event to be unwound by its new owner. Those tasks outlive the coroutine frame, so they reach the DrainingReader through a kj::Weak rather than the frame-owned kj::Own. Without that, the read lambda would run against a freed reader whenever the pump is dropped before the isolate lock is acquired. The fix originally landed as 00e86e120 and was swept away by 4447562d6, a wholesale revert of the streams-cleanup-investigation merge. That first version captured the reader by reference, which is what the kj::Weak above replaces; this follows the later 2de66f75e, which had corrected it. Both are James's, as is the third attempt, f3f408768, which drops the code change and keeps only the test — not an option here, since the abort still reproduces without it. The surrounding tree has moved on twice over: the draining read is no longer autogated, so the test needs no gate, and DrainingReader is a kj::PtrTarget, so the weak pointer comes from addWeakToThis() rather than a hand-rolled cell. Co-authored-by: James M Snell Assisted-by: OpenCode:claude-opus-5 --- src/workerd/api/streams/readable.h | 3 + src/workerd/api/streams/standard.c++ | 67 ++++++++++++++++--- src/workerd/api/tests/BUILD.bazel | 9 +++ src/workerd/api/tests/autovuln-187-echo.js | 9 +++ src/workerd/api/tests/autovuln-187-test.js | 36 ++++++++++ .../api/tests/autovuln-187-test.wd-test | 24 +++++++ 6 files changed, 139 insertions(+), 9 deletions(-) create mode 100644 src/workerd/api/tests/autovuln-187-echo.js create mode 100644 src/workerd/api/tests/autovuln-187-test.js create mode 100644 src/workerd/api/tests/autovuln-187-test.wd-test diff --git a/src/workerd/api/streams/readable.h b/src/workerd/api/streams/readable.h index ef6b68d4dc0..c0e95addcd0 100644 --- a/src/workerd/api/streams/readable.h +++ b/src/workerd/api/streams/readable.h @@ -255,6 +255,9 @@ class DrainingReader: public ReadableStreamController::Reader { kj::Ptr getPtr() { return addPtrToThis(); } + // A pointer for holders that may outlive the reader and must notice when they have. + kj::Weak getWeak() { return addWeakToThis(); } + private: struct Initial {}; using Attached = jsg::Ref; diff --git a/src/workerd/api/streams/standard.c++ b/src/workerd/api/streams/standard.c++ index dfc8d329191..a3718f9d36f 100644 --- a/src/workerd/api/streams/standard.c++ +++ b/src/workerd/api/streams/standard.c++ @@ -3479,6 +3479,25 @@ class AllReader { } }; +// Hands `promise` off to a holder that outlives the caller, reporting its outcome +// through `fulfiller`. Fulfilling or rejecting a fulfiller whose promise is already +// gone is a no-op, so the caller is free to stop listening at any point. +template +kj::Promise forwardToFulfiller( + kj::Promise promise, kj::Own> fulfiller) { + KJ_TRY { + if constexpr (jsg::isVoid()) { + co_await promise; + fulfiller->fulfill(); + } else { + fulfiller->fulfill(co_await promise); + } + } + KJ_CATCH(exception) { + fulfiller->reject(kj::mv(exception)); + } +} + // pumpToImpl uses a DrainingReader to efficiently pull all synchronously available // data from the stream in each iteration, then writes it to the sink using vectored // I/O. This minimizes isolate lock acquisitions by batching: each time the lock is @@ -3487,11 +3506,23 @@ class AllReader { // // The pump loop is a kj coroutine. Dropping the returned kj::Promise drops the // coroutine frame, which destroys the DrainingReader (releasing the stream lock) -// and the sink. No WeakRef/IoOwn dance is needed because ownership is clear. +// and the sink. // The coroutine that implements the pump loop takes ownership of the DrainingReader // and sink. The jsg::Ref is not passed into the coroutine because // jsg::Ref is disallowed in coroutine parameters; instead, the DrainingReader holds // a reference to the stream internally. +// +// Neither isolate-lock round trip below is awaited directly. A draining read runs +// the stream's pull() callback, and pull() can abort the request, which drops the +// pump promise from inside the very event that is delivering the read's result. +// Destroying an event while it is firing trips "Promise callback destroyed itself". +// Each run() is therefore handed to the IoContext's task set, and the pump awaits an +// unrelated fulfiller. Dropping the pump then destroys only that await; the firing +// event stays owned by the task set, which unwinds it once it is safe to. +// +// The consequence is that those tasks outlive the coroutine frame, so they must not +// name anything the frame owns. The DrainingReader is reached through a kj::Weak, +// which reports the frame's destruction rather than dangling into it. kj::Promise pumpToImpl(IoContext& ioContext, kj::Own reader, kj::Own sink, @@ -3503,13 +3534,22 @@ kj::Promise pumpToImpl(IoContext& ioContext, while (true) { // Perform a draining read to get all synchronously available data if possible // or fall back to a regular read if not. - DrainingReadResult result = co_await ioContext.run([&reader](jsg::Lock& js) mutable { + auto prp = kj::newPromiseAndFulfiller(); + auto promise = ioContext.run([weakReader = reader->getWeak()]( + jsg::Lock& js) mutable -> kj::Promise { auto& ioContext = IoContext::current(); - // Use a 256KB limit to allow periodic yielding to the event loop, - // preventing a fast producer from monopolizing the thread. - constexpr size_t kMaxReadPerCycle = 256 * 1024; - return ioContext.awaitJs(js, reader->read(js, kMaxReadPerCycle)); + KJ_IF_SOME(reader, weakReader.tryGet()) { + // Use a 256KB limit to allow periodic yielding to the event loop, + // preventing a fast producer from monopolizing the thread. + constexpr size_t kMaxReadPerCycle = 256 * 1024; + return ioContext.awaitJs(js, reader.read(js, kMaxReadPerCycle)); + } else { + return KJ_EXCEPTION(DISCONNECTED, "The pump was canceled."); + } }); + ioContext.addTask(forwardToFulfiller(kj::mv(promise), kj::mv(prp.fulfiller))); + + DrainingReadResult result = co_await prp.promise; // Write all the chunks we received using vectored write for efficiency. if (result.chunks.size() > 0) { @@ -3534,11 +3574,20 @@ kj::Promise pumpToImpl(IoContext& ioContext, sink->abort(exception.clone()); } - co_await ioContext.run([&reader, ex = exception.clone()](jsg::Lock& js) mutable { + auto prp = kj::newPromiseAndFulfiller(); + auto promise = ioContext.run([weakReader = reader->getWeak(), ex = exception.clone()]( + jsg::Lock& js) mutable -> kj::Promise { auto& ioContext = IoContext::current(); - auto error = js.exceptionToJsValue(kj::mv(ex)); - return ioContext.awaitJs(js, reader->cancel(js, error.getHandle(js))); + KJ_IF_SOME(reader, weakReader.tryGet()) { + auto error = js.exceptionToJsValue(kj::mv(ex)); + return ioContext.awaitJs(js, reader.cancel(js, error.getHandle(js))); + } else { + return KJ_EXCEPTION(DISCONNECTED, "The pump was canceled."); + } }); + ioContext.addTask(forwardToFulfiller(kj::mv(promise), kj::mv(prp.fulfiller))); + + co_await prp.promise; kj::throwFatalException(kj::mv(exception)); } } diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index 35393b36947..2e70ba39825 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -804,6 +804,15 @@ wd_test( data = ["autovuln-319-test.js"], ) +wd_test( + src = "autovuln-187-test.wd-test", + args = ["--experimental"], + data = [ + "autovuln-187-echo.js", + "autovuln-187-test.js", + ], +) + wd_test( src = "unsafe-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/autovuln-187-echo.js b/src/workerd/api/tests/autovuln-187-echo.js new file mode 100644 index 00000000000..169b8e9970e --- /dev/null +++ b/src/workerd/api/tests/autovuln-187-echo.js @@ -0,0 +1,9 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +export default { + fetch() { + return new Response('ok'); + }, +}; diff --git a/src/workerd/api/tests/autovuln-187-test.js b/src/workerd/api/tests/autovuln-187-test.js new file mode 100644 index 00000000000..e9a8085dee2 --- /dev/null +++ b/src/workerd/api/tests/autovuln-187-test.js @@ -0,0 +1,36 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +import { rejects } from 'node:assert'; + +// Regression test for AUTOVULN-CLOUDFLARE-WORKERD-187. +// +// A draining read delivers its result to pumpToImpl through a kj event, and it +// runs the stream's pull() callback along the way. A pull() that aborts the +// request's signal makes the canceler destroy the pump's coroutine frame from +// inside that event, so the event is destroyed while still firing and kj aborts +// the process with "Promise callback destroyed itself". +export default { + async test(_ctrl, env) { + const ac = new AbortController(); + let n = 0; + const rs = new ReadableStream({ + pull(c) { + if (++n === 2) ac.abort(); + c.enqueue(new Uint8Array([65])); + }, + }); + await rejects( + env.ECHO.fetch('http://x/', { + method: 'POST', + body: rs, + signal: ac.signal, + duplex: 'half', + }), + { + message: 'The operation was aborted', + } + ); + }, +}; diff --git a/src/workerd/api/tests/autovuln-187-test.wd-test b/src/workerd/api/tests/autovuln-187-test.wd-test new file mode 100644 index 00000000000..826e4b57c5f --- /dev/null +++ b/src/workerd/api/tests/autovuln-187-test.wd-test @@ -0,0 +1,24 @@ +using Workerd = import "/workerd/workerd.capnp"; + +const unitTests :Workerd.Config = ( + services = [ + ( name = "autovuln-187-test", + worker = ( + modules = [ + (name = "worker", esModule = embed "autovuln-187-test.js") + ], + compatibilityFlags = ["nodejs_compat", "streams_enable_constructors"], + bindings = [ + (name = "ECHO", service = "echo"), + ], + ) + ), + ( name = "echo", + worker = ( + modules = [ + (name = "worker", esModule = embed "autovuln-187-echo.js") + ], + ) + ), + ], +);