From 1883e0fcdb82963669167092cacce518fa36f840 Mon Sep 17 00:00:00 2001 From: Guy Bedford Date: Fri, 21 Aug 2026 17:13:23 -0700 Subject: [PATCH] Fix EventSource use-after-free EventSourceSink queued a gated callback capturing a raw EventSource pointer. If the EventSource was collected before callback execution, the callback dereferenced freed native memory. Carry a weak EventSource reference through the sink and promote it under the isolate lock before dispatch. --- src/workerd/api/eventsource.c++ | 25 +++++++---- src/workerd/api/eventsource.h | 3 +- src/workerd/api/tests/BUILD.bazel | 6 +++ src/workerd/api/tests/eventsource-gc-test.js | 41 +++++++++++++++++++ .../api/tests/eventsource-gc-test.wd-test | 28 +++++++++++++ 5 files changed, 95 insertions(+), 8 deletions(-) create mode 100644 src/workerd/api/tests/eventsource-gc-test.js create mode 100644 src/workerd/api/tests/eventsource-gc-test.wd-test diff --git a/src/workerd/api/eventsource.c++ b/src/workerd/api/eventsource.c++ index 27d1b023189..4fde315493d 100644 --- a/src/workerd/api/eventsource.c++ +++ b/src/workerd/api/eventsource.c++ @@ -17,7 +17,9 @@ namespace workerd::api { namespace { class EventSourceSink final: public WritableStreamSink { public: - EventSourceSink(EventSource& eventSource): eventSource(eventSource) {} + EventSourceSink(EventSource& eventSource, kj::Rc> weakEventSource) + : eventSource(eventSource), + weakEventSource(kj::mv(weakEventSource)) {} kj::Promise write(kj::ArrayPtr buffer) override { // The event stream is a new-line delimited format where each line represents an event. @@ -86,6 +88,7 @@ class EventSourceSink final: public WritableStreamSink { private: kj::Maybe eventSource; + kj::Rc> weakEventSource; // Retained bytes to be processed in the next write. kj::Vector kept; @@ -181,7 +184,7 @@ class EventSourceSink final: public WritableStreamSink { auto pending = pendingMessages.releaseAsArray(); // If the event source is gone, just drop the messages on the floor. KJ_IF_SOME(es, eventSource) { - es.enqueueMessages(kj::mv(pending)); + es.enqueueMessages(kj::mv(pending), weakEventSource.addRef()); } } @@ -497,8 +500,11 @@ void EventSource::run(jsg::Lock& js, // or errored. context .awaitIo(js, - processBody( - context, readable.pumpTo(js, kj::heap(*this), EndStream::YES))) + processBody(context, + readable.pumpTo(js, + kj::heap( + *this, kj::rc>(JSG_THIS_WEAK(js))), + EndStream::YES))) .then(js, kj::mv(onSuccess), kj::mv(onFailed)); } @@ -509,9 +515,14 @@ void EventSource::close(jsg::Lock& js) { readyState = State::CLOSED; } -void EventSource::enqueueMessages(kj::Array messages) { - context.addTask(context.run([this, messages = kj::mv(messages)]( - auto& lock) mutable { notifyMessages(lock, kj::mv(messages)); })); +void EventSource::enqueueMessages( + kj::Array messages, kj::Rc> weakSelf) { + context.addTask( + context.run([weakSelf = kj::mv(weakSelf), messages = kj::mv(messages)](auto& lock) mutable { + KJ_IF_SOME(self, weakSelf->tryAddRef(lock)) { + self->notifyMessages(lock, kj::mv(messages)); + } + })); } void EventSource::setReconnectionTime(uint32_t time) { diff --git a/src/workerd/api/eventsource.h b/src/workerd/api/eventsource.h index c0c84498122..ea1bfce96a8 100644 --- a/src/workerd/api/eventsource.h +++ b/src/workerd/api/eventsource.h @@ -122,7 +122,8 @@ class EventSource: public EventTarget { // Called by the internal implementation to notify the EventSource about messages // received from the server. - void enqueueMessages(kj::Array messages); + void enqueueMessages( + kj::Array messages, kj::Rc> weakSelf); // Called by the internal implementation to notify the EventSource that the server // has provided a new reconnection time. diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index 65cbf1f8ac9..9509ebcb6f7 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -555,6 +555,12 @@ wd_test( data = ["events-test.js"], ) +wd_test( + src = "eventsource-gc-test.wd-test", + args = ["--experimental"], + data = ["eventsource-gc-test.js"], +) + wd_test( src = "eventsource-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/eventsource-gc-test.js b/src/workerd/api/tests/eventsource-gc-test.js new file mode 100644 index 00000000000..4c28817729f --- /dev/null +++ b/src/workerd/api/tests/eventsource-gc-test.js @@ -0,0 +1,41 @@ +// 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 assert from 'node:assert'; +import { DurableObject } from 'cloudflare:workers'; + +export class EventSourceGcDo extends DurableObject { + async trigger() { + await this.ctx.blockConcurrencyWhile(async () => { + // EventSource delivery acquires the actor input gate, so messages stay queued until this + // callback returns. + const response = await this.env.backend.fetch('https://example.com'); + const ended = Promise.withResolvers(); + let source = EventSource.from(response.body); + source.onerror = () => ended.resolve(); + + await ended.promise; + + const sourceRef = new WeakRef(source); + source = null; + await scheduler.wait(0); + gc(); + await scheduler.wait(0); + gc(); + await scheduler.wait(0); + assert.strictEqual(sourceRef.deref(), undefined); + }); + + await scheduler.wait(0); + } +} + +export default { + async test(_controller, env) { + await env.ns.getByName('test').trigger(); + }, + fetch() { + return new Response('data: message\n\n'); + }, +}; diff --git a/src/workerd/api/tests/eventsource-gc-test.wd-test b/src/workerd/api/tests/eventsource-gc-test.wd-test new file mode 100644 index 00000000000..41f66d7c5aa --- /dev/null +++ b/src/workerd/api/tests/eventsource-gc-test.wd-test @@ -0,0 +1,28 @@ +using Workerd = import "/workerd/workerd.capnp"; + +const config :Workerd.Config = ( + v8Flags = ["--expose-gc"], + services = [ + (name = "eventsource-gc-test", worker = .testWorker), + ], +); + +const testWorker :Workerd.Worker = ( + compatibilityFlags = [ + "enable_weak_ref", + "experimental", + "nodejs_compat", + "streams_enable_constructors", + ], + modules = [ + (name = "worker", esModule = embed "eventsource-gc-test.js"), + ], + durableObjectNamespaces = [ + (className = "EventSourceGcDo", uniqueKey = "c9f59b7c629443cd973bd29cb3a5d5e9"), + ], + durableObjectStorage = (inMemory = void), + bindings = [ + (name = "backend", service = "eventsource-gc-test"), + (name = "ns", durableObjectNamespace = "EventSourceGcDo"), + ], +);