node:http: emit 'data' on the server 'connection' socket with the raw request bytes

packages/bun-uws/src/App.h

@@ -750,6 +750,9 @@ struct TemplatedApp {
     void setOnSocketData(HttpContextData<SSL>::OnSocketDataCallback onData) {
         httpContext->getSocketContextData()->onSocketData = onData;
     }
+    void setOnSocketRawData(HttpContextData<SSL>::OnSocketRawDataCallback onRawData) {
+        httpContext->getSocketContextData()->onSocketRawData = onRawData;
+    }
 
     void setOnClientError(HttpContextData<SSL>::OnClientErrorCallback onClientError) {
         httpContext->getSocketContextData()->onClientError = std::move(onClientError);

packages/bun-uws/src/HttpContext.h

@@ -295,6 +295,21 @@ struct HttpContext {
 
         HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) us_socket_ext(s);
 
+        /* node:http compat: forward the raw TCP payload to the JS socket
+         * wrapper so a user 'data' listener on the 'connection' socket sees the
+         * raw request bytes (Node feeds its parser from that event). Tunnel
+         * mode routes through onSocketData instead. */
+        bool rawTapCoveredThisRead = false;
+        if constexpr (IsNodeHttp) {
+            if (!httpResponseData->isConnectRequest && httpResponseData->socketData && httpContextData->onSocketRawData) {
+                rawTapCoveredThisRead = httpContextData->onSocketRawData(httpResponseData->socketData, SSL, s, data, length);
+                if (us_socket_is_closed(s)) {
+                    us_socket_unref(s);
+                    return s;
+                }
+            }
+        }
+
         /* node:http compat: HTTP parsing stopped on this connection (a parse error
          * was already delivered to 'clientError', or the JS layer freed the
          * parser); ignore further request bytes. CONNECT/Upgrade tunnels are not
@@ -475,7 +490,7 @@ struct HttpContext {
             /* Continue parsing */
             return s;
 
-        }, [httpResponseData, httpContextData](void *user, std::string_view data, bool fin) -> void * {
+        }, [httpResponseData, httpContextData, rawTapCoveredThisRead](void *user, std::string_view data, bool fin) -> void * {
 
             /* node:http compat: an accepted Upgrade request's body just completed -
              * after this fin chunk has been delivered to the request body stream
@@ -499,7 +514,7 @@ struct HttpContext {
                 }
             }
 
-            if (httpResponseData->isConnectRequest && httpResponseData->socketData && httpContextData->onSocketData) {
+            if (httpResponseData->isConnectRequest && !rawTapCoveredThisRead && httpResponseData->socketData && httpContextData->onSocketData) {
                 httpContextData->onSocketData(httpResponseData->socketData, SSL, (struct us_socket_t *) user, data.data(), data.length(), fin);
             }
 

packages/bun-uws/src/HttpContextData.h

@@ -54,6 +54,7 @@ struct alignas(16) HttpContextData {
 private:
     std::vector<MoveOnlyFunction<void(HttpResponse<SSL> *, int)>> filterHandlers;
     using OnSocketDataCallback = void (*)(void* userData, int is_ssl, struct us_socket_t *rawSocket, const char *data, int length, bool last);
+    using OnSocketRawDataCallback = bool (*)(void* userData, int is_ssl, struct us_socket_t *rawSocket, const char *data, int length);
     using OnSocketDrainCallback = void (*)(void* userData, int is_ssl, struct us_socket_t *rawSocket);
     using OnSocketUpgradedCallback = void (*)(void* userData, int is_ssl, struct us_socket_t *rawSocket);
     using OnClientErrorCallback = MoveOnlyFunction<void(int is_ssl, struct us_socket_t *rawSocket, uWS::HttpParserError errorCode, char *rawPacket, int rawPacketLength)>;
@@ -76,6 +77,10 @@ struct alignas(16) HttpContextData {
     OnSocketClosedCallback onSocketClosed = nullptr;
     OnSocketDrainCallback onSocketDrain = nullptr;
     OnSocketDataCallback onSocketData = nullptr;
+    /* node:http: tap of every TCP payload before HTTP parsing, so a user
+     * 'data' listener on the 'connection' socket sees the raw request bytes.
+     * Returns true when a JS callback was armed (the bytes will reach JS). */
+    OnSocketRawDataCallback onSocketRawData = nullptr;
     OnSocketUpgradedCallback onSocketUpgraded = nullptr;
     OnClientErrorCallback onClientError = nullptr;
 

src/js/node/_http_server.ts

@@ -898,6 +898,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
 
         if (isSocketNew && !reachedRequestsLimit) {
           server.emit("connection", socket);
+          maybeEnableRawSocketData(socket);
         }
 
         if (!isPipelined) {
@@ -1226,6 +1227,7 @@ function onServerConnection(this: Server, socketHandle) {
   if (isTLS && socketHandle.secureEstablished) {
     this.emit("secureConnection", socket);
   }
+  maybeEnableRawSocketData(socket);
 }
 
 // Like Node.js: server.setTimeout only records the per-socket inactivity
@@ -1393,6 +1395,13 @@ function detachSocketListenersForHandoff(socket) {
 }
 const kSocketTimeoutTimer = Symbol("socketTimeoutTimer");
 const kStreamingEnabled = Symbol("kStreamingEnabled");
+const kEnableRawData = Symbol("kEnableRawData");
+// Arm the native raw-bytes tap so 'data'/'readable' see the raw request bytes.
+function maybeEnableRawSocketData(socket) {
+  if (!socket[kStreamingEnabled] && (socket.listenerCount("data") > 0 || socket.listenerCount("readable") > 0)) {
+    socket[kEnableRawData]();
+  }
+}
 // Scratch options object for the builtin ServerResponse (see the dispatcher).
 const scratchResponseOptions = {
   [kHandle]: undefined,
@@ -1625,9 +1634,21 @@ const NodeHTTPServerSocket = class Socket extends NetSocket {
     }
     this.emit("drain");
   }
+  #onRawData(chunk) {
+    this._unrefTimer();
+    this.bytesRead += chunk.length;
+    this.push(chunk);
+  }
+  [kEnableRawData]() {
+    const handle = this[kHandle];
+    if (handle && !handle.onrawdata) {
+      handle.onrawdata = this.#onRawData.bind(this);
+    }
+  }
   #onData(chunk, last) {
     this._unrefTimer();
     if (chunk) {
+      this.bytesRead += chunk.length;
       this.push(chunk);
     }
     if (last) {
@@ -1866,19 +1887,23 @@ const NodeHTTPServerSocket = class Socket extends NetSocket {
         const message = this._httpMessage;
         const req = message?.req;
 
-        if ((bodyReadState & NodeHTTPBodyReadState.done) !== 0) {
-          emitServerSocketEOFNT(this, req);
+        // The raw tap owns the socket readable; parsed body goes to req only.
+        if (!handle.onrawdata) {
+          if ((bodyReadState & NodeHTTPBodyReadState.done) !== 0) {
+            emitServerSocketEOFNT(this, req);
+          }
+          this.push(resumed);
         }
         if (req) {
           req.push(resumed);
         }
-        this.push(resumed);
       }
     }
   }
 
   _read(_size) {
     // https://github.com/nodejs/node/blob/13e3aef053776be9be262f210dc438ecec4a3c8d/lib/net.js#L725-L737
+    maybeEnableRawSocketData(this);
     this.#resumeSocket();
   }
 

src/jsc/bindings/NodeHTTP.cpp

@@ -613,6 +613,11 @@ static void assignOnNodeJSCompat(uWS::TemplatedApp<isSSL>* app)
         ASSERT(rawSocket == socket->socket || socket->socket == nullptr);
         socket->onData(data, length, last);
     });
+    app->setOnSocketRawData([](void* socketData, int is_ssl, struct us_socket_t* rawSocket, const char* data, int length) -> bool {
+        auto* socket = reinterpret_cast<JSNodeHTTPServerSocket*>(socketData);
+        ASSERT(rawSocket == socket->socket || socket->socket == nullptr);
+        return socket->onRawData(data, length);
+    });
     app->setOnSocketUpgraded([](void* socketData, int is_ssl, struct us_socket_t* rawSocket) -> void {
         auto* socket = reinterpret_cast<JSNodeHTTPServerSocket*>(socketData);
         // the socket is adopted and might not be the same as the rawSocket

src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp

@@ -649,6 +649,54 @@ void JSNodeHTTPServerSocket::onData(const char* data, int length, bool last)
     }
 }
 
+bool JSNodeHTTPServerSocket::onRawData(const char* data, int length)
+{
+    if (!functionToCallOnRawData) {
+        return false;
+    }
+    Zig::GlobalObject* globalObject = static_cast<Zig::GlobalObject*>(this->globalObject());
+    WebCore::ScriptExecutionContext* scriptExecutionContext = globalObject->scriptExecutionContext();
+    if (!scriptExecutionContext) {
+        return false;
+    }
+
+    auto scope = DECLARE_TOP_EXCEPTION_SCOPE(globalObject->vm());
+    JSC::JSUint8Array* buffer = WebCore::createBuffer(globalObject, std::span<const uint8_t>(reinterpret_cast<const uint8_t*>(data), length));
+    auto chunk = JSC::JSValue(buffer);
+    if (auto* exception = scope.exception()) {
+        (void)scope.tryClearException();
+        globalObject->reportUncaughtExceptionAtEventLoop(globalObject, exception);
+        return false;
+    }
+    gcProtect(chunk);
+    scriptExecutionContext->postTask([self = this, chunk = chunk](ScriptExecutionContext& context) {
+        WTF::NakedPtr<JSC::Exception> exception;
+        auto* globalObject = defaultGlobalObject(context.globalObject());
+        auto* thisObject = self;
+        auto* callbackObject = thisObject->functionToCallOnRawData.get();
+        EnsureStillAliveScope ensureChunkStillAlive(chunk);
+        gcUnprotect(chunk);
+        if (!callbackObject) {
+            return;
+        }
+
+        auto callData = JSC::getCallData(callbackObject);
+        MarkedArgumentBuffer args;
+        args.append(chunk);
+        EnsureStillAliveScope ensureStillAlive(self);
+
+        if (globalObject->scriptExecutionStatus(globalObject, thisObject) == ScriptExecutionStatus::Running) {
+            profiledCall(globalObject, JSC::ProfilingReason::API, callbackObject, callData, thisObject, args, exception);
+
+            if (auto* ptr = exception.get()) {
+                exception.clear();
+                globalObject->reportUncaughtExceptionAtEventLoop(globalObject, ptr);
+            }
+        }
+    });
+    return true;
+}
+
 JSC::Structure* JSNodeHTTPServerSocket::createStructure(JSC::VM& vm, JSC::JSGlobalObject* globalObject)
 {
     auto* structure = JSC::Structure::create(vm, globalObject, globalObject->objectPrototype(), JSC::TypeInfo(JSC::ObjectType, StructureFlags), JSNodeHTTPServerSocketPrototype::info());
@@ -672,6 +720,7 @@ void JSNodeHTTPServerSocket::visitChildrenImpl(JSCell* cell, Visitor& visitor)
     visitor.append(fn->functionToCallOnClose);
     visitor.append(fn->functionToCallOnDrain);
     visitor.append(fn->functionToCallOnData);
+    visitor.append(fn->functionToCallOnRawData);
     visitor.append(fn->m_remoteAddress);
     visitor.append(fn->m_localAddress);
     visitor.append(fn->m_duplex);

src/jsc/bindings/node/JSNodeHTTPServerSocket.h

@@ -128,6 +128,7 @@ class JSNodeHTTPServerSocket : public JSC::JSDestructibleObject {
     mutable JSC::WriteBarrier<JSC::JSObject> functionToCallOnClose;
     mutable JSC::WriteBarrier<JSC::JSObject> functionToCallOnDrain;
     mutable JSC::WriteBarrier<JSC::JSObject> functionToCallOnData;
+    mutable JSC::WriteBarrier<JSC::JSObject> functionToCallOnRawData;
     mutable JSC::WriteBarrier<WebCore::JSNodeHTTPResponse> currentResponseObject;
     mutable JSC::WriteBarrier<JSC::JSObject> m_remoteAddress;
     mutable JSC::WriteBarrier<JSC::JSObject> m_localAddress;
@@ -160,6 +161,7 @@ class JSNodeHTTPServerSocket : public JSC::JSDestructibleObject {
     void onClose();
     void onDrain();
     void onData(const char* data, int length, bool last);
+    bool onRawData(const char* data, int length);
 
     static JSC::Structure* createStructure(JSC::VM& vm, JSC::JSGlobalObject* globalObject);
     void finishCreation(JSC::VM& vm);

src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp

@@ -27,6 +27,8 @@ JSC_DECLARE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnClose);
 JSC_DECLARE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnDrain);
 JSC_DECLARE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnData);
 JSC_DECLARE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterOnData);
+JSC_DECLARE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnRawData);
+JSC_DECLARE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterOnRawData);
 JSC_DECLARE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterBytesWritten);
 JSC_DECLARE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketClose);
 JSC_DECLARE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketWrite);
@@ -57,6 +59,7 @@ static const JSC::HashTableValue JSNodeHTTPServerSocketPrototypeTableValues[] =
     { "onclose"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterOnClose, jsNodeHttpServerSocketSetterOnClose } },
     { "ondrain"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterOnDrain, jsNodeHttpServerSocketSetterOnDrain } },
     { "ondata"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterOnData, jsNodeHttpServerSocketSetterOnData } },
+    { "onrawdata"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterOnRawData, jsNodeHttpServerSocketSetterOnRawData } },
     { "bytesWritten"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterBytesWritten, noOpSetter } },
     { "closed"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor | JSC::PropertyAttribute::ReadOnly), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterClosed, noOpSetter } },
     { "response"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor | JSC::PropertyAttribute::ReadOnly), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterResponse, noOpSetter } },
@@ -469,6 +472,44 @@ JSC_DEFINE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnData, (JSC::JSGlobalObjec
     return true;
 }
 
+JSC_DEFINE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterOnRawData, (JSC::JSGlobalObject * globalObject, JSC::EncodedJSValue thisValue, JSC::PropertyName))
+{
+    auto* thisObject = dynamicDowncast<JSNodeHTTPServerSocket>(JSC::JSValue::decode(thisValue));
+    if (!thisObject) [[unlikely]] {
+        return JSValue::encode(JSC::jsUndefined());
+    }
+
+    if (thisObject->functionToCallOnRawData) {
+        return JSValue::encode(thisObject->functionToCallOnRawData.get());
+    }
+
+    return JSValue::encode(JSC::jsUndefined());
+}
+
+JSC_DEFINE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnRawData, (JSC::JSGlobalObject * globalObject, JSC::EncodedJSValue thisValue, JSC::EncodedJSValue encodedValue, JSC::PropertyName propertyName))
+{
+    auto& vm = globalObject->vm();
+    auto scope = DECLARE_THROW_SCOPE(vm);
+
+    auto* thisObject = dynamicDowncast<JSNodeHTTPServerSocket>(JSC::JSValue::decode(thisValue));
+    if (!thisObject) [[unlikely]] {
+        return false;
+    }
+    JSValue value = JSC::JSValue::decode(encodedValue);
+
+    if (value.isUndefined() || value.isNull()) {
+        thisObject->functionToCallOnRawData.clear();
+        return true;
+    }
+
+    if (!value.isCallable()) {
+        return false;
+    }
+
+    thisObject->functionToCallOnRawData.set(vm, thisObject, value.getObject());
+    return true;
+}
+
 JSC_DEFINE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnClose, (JSC::JSGlobalObject * globalObject, JSC::EncodedJSValue thisValue, JSC::EncodedJSValue encodedValue, JSC::PropertyName propertyName))
 {
     auto& vm = globalObject->vm();