Fix remote OOM via delayed deserialization for unary/server-streaming calls

Protobuf unknown-field or repeated field amplification can lead to remote OOM if an attacker
sends a unary request but holds the stream open without half-closing.
This change delays the deserialization of incoming messages for calls where
the client sends at most one message (Unary and Server Streaming) until
the client actually half-closes the stream (sends END_STREAM).

If the call is cancelled before half-close, the buffered raw message is
discarded without being deserialized, preventing the memory explosion.

Author: kannanjgithub

core/src/main/java/io/grpc/internal/ServerCallImpl.java

@@ -34,6 +34,7 @@
 import io.grpc.CompressorRegistry;
 import io.grpc.Context;
 import io.grpc.DecompressorRegistry;
+import io.grpc.Detachable;
 import io.grpc.InternalDecompressorRegistry;
 import io.grpc.InternalStatus;
 import io.grpc.Metadata;
@@ -45,6 +46,9 @@
 import io.perfmark.PerfMark;
 import io.perfmark.Tag;
 import io.perfmark.TaskCloseable;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.io.InputStream;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -288,6 +292,7 @@ static final class ServerStreamListenerImpl<ReqT> implements ServerStreamListene
     private final ServerCallImpl<ReqT, ?> call;
     private final ServerCall.Listener<ReqT> listener;
     private final Context.CancellableContext context;
+    private InputStream delayedMessage;
 
     public ServerStreamListenerImpl(
         ServerCallImpl<ReqT, ?> call, ServerCall.Listener<ReqT> listener,
@@ -320,6 +325,20 @@ public void messagesAvailable(MessageProducer producer) {
       }
     }
 
+    private static InputStream bufferMessage(InputStream is) throws IOException {
+      if (is instanceof Detachable) {
+        return ((Detachable) is).detach();
+      }
+      // Fallback: copy to byte array
+      ByteArrayOutputStream baos = new ByteArrayOutputStream();
+      byte[] buffer = new byte[4096];
+      int bytesRead;
+      while ((bytesRead = is.read(buffer)) != -1) {
+        baos.write(buffer, 0, bytesRead);
+      }
+      return new ByteArrayInputStream(baos.toByteArray());
+    }
+
     @SuppressWarnings("Finally") // The code avoids suppressing the exception thrown from try
     private void messagesAvailableInternal(final MessageProducer producer) {
       if (call.cancelled) {
@@ -330,13 +349,32 @@ private void messagesAvailableInternal(final MessageProducer producer) {
       InputStream message;
       try {
         while ((message = producer.next()) != null) {
-          try {
-            listener.onMessage(call.method.parseRequest(message));
-          } catch (Throwable t) {
-            GrpcUtil.closeQuietly(message);
-            throw t;
+          if (call.method.getType().clientSendsOneMessage()) {
+            if (delayedMessage != null) {
+              GrpcUtil.closeQuietly(message);
+              call.close(
+                  Status.INTERNAL.withDescription("Too many requests"),
+                  new Metadata());
+              GrpcUtil.closeQuietly(delayedMessage);
+              delayedMessage = null;
+              return;
+            }
+            try {
+              delayedMessage = bufferMessage(message);
+            } catch (Throwable t) {
+              GrpcUtil.closeQuietly(message);
+              throw t;
+            }
+            message.close();
+          } else {
+            try {
+              listener.onMessage(call.method.parseRequest(message));
+            } catch (Throwable t) {
+              GrpcUtil.closeQuietly(message);
+              throw t;
+            }
+            message.close();
           }
-          message.close();
         }
       } catch (Throwable t) {
         GrpcUtil.closeQuietly(producer);
@@ -353,6 +391,23 @@ public void halfClosed() {
           return;
         }
 
+        if (delayedMessage != null) {
+          InputStream message = delayedMessage;
+          delayedMessage = null;
+          try {
+            listener.onMessage(call.method.parseRequest(message));
+          } catch (Throwable t) {
+            GrpcUtil.closeQuietly(message);
+            Throwables.throwIfUnchecked(t);
+            throw new RuntimeException(t);
+          }
+          try {
+            message.close();
+          } catch (IOException e) {
+            throw new RuntimeException(e);
+          }
+        }
+
         listener.onHalfClose();
       }
     }
@@ -366,6 +421,10 @@ public void closed(Status status) {
     }
 
     private void closedInternal(Status status) {
+      if (delayedMessage != null) {
+        GrpcUtil.closeQuietly(delayedMessage);
+        delayedMessage = null;
+      }
       Throwable cancelCause = null;
       try {
         if (status.isOk()) {