core: Delayed deserialization for unary/server-streaming calls - #13004
core: Delayed deserialization for unary/server-streaming calls#13004kannanjgithub wants to merge 3 commits into
Conversation
ca56fe0 to
5eb7065
Compare
… 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.
5eb7065 to
09d8c90
Compare
| if (call.method.getType().clientSendsOneMessage()) { | ||
| if (delayedMessage != null) { | ||
| GrpcUtil.closeQuietly(message); | ||
| call.close( |
There was a problem hiding this comment.
This notifies the application, but doesn't notify the transport, so this RPC could be leaked.
There was a problem hiding this comment.
Changed it to stream cancel now to abort stream and tear down transport without waiting for client half close .
| return; | ||
| } | ||
| try { | ||
| delayedMessage = bufferMessage(message); |
There was a problem hiding this comment.
Why are we making a copy here when we could just "not call close()" on the original message?
There was a problem hiding this comment.
That would work for the Detachable InputStream since the ref count on Netty ByteBuf is still going to be non-zero even when detaching, so we might as well have just held the reference pass on to us.
But for non Detachable InputStream such as for compressed streams, if we don't copy to heap and release the message InputStream passed it will continue to hold the native memory for zlib objects which are allocated per request message and can be larger than the request message size itself.
By detaching the InputStream for Detachables, bufferMessage allows the close handling for both detachable and non-detachable cases be uniform without having to check which case it is.
There was a problem hiding this comment.
No need for detachable, and I would agree we shouldn't use it here. This is the code that calls close(), so we can just "not call close" if we want to avoid it. The compression context is relevant/important, but if we decompress then that in itself can hold a lot more memory than the compression context. We could make MessageDeframer return lazy inputstreams that call decompressor.decompress() when the first bytes are read.
It would be good to consider if forcing the transport to deal with this might be better, at least long-term. If MessageDeframer was told there was only one message, then it could delay running processBody() until closeWhenComplete == true (it will still run readRequiredBytes() until it got the message, though). It might be ugly, especially detecting cardinality violation (but dealing with it would be trivial: throw an exception), but probably worth checking at some point. (Binder mostly wouldn't even need any changes, because unary is already handled specially, but I don't remember if you can use streaming mode for a unary call.)
| try { | ||
| message.close(); | ||
| } catch (IOException e) { | ||
| throw new RuntimeException(e); | ||
| } |
kannanjgithub
changed the title
- Use GrpcUtil.closeQuietly to close the buffered message in halfClosed() to prevent unnecessary exception propagation if close fails after successful message delivery. - Use stream.cancel instead of call.close when detecting too many requests for unary calls. This ensures the transport is notified to abort the stream (sending RST_STREAM) and immediately releases resources, preventing leaks from clients that withhold END_STREAM.
| GrpcUtil.closeQuietly(message); | ||
| call.stream.cancel(Status.INTERNAL.withDescription("Too many requests")); | ||
| GrpcUtil.closeQuietly(delayedMessage); | ||
| delayedMessage = null; |
There was a problem hiding this comment.
This puts the call back into a normal state, so if other events happen after this one (e.g., message, or half close), that could end up propagating to the application before the cancel is processed. I don't know the easiest way to handle that though; obviously we could set some more state/booleans. It is probably worth looking into the exception handling in the executor see what would happen if we throw here.
There was a problem hiding this comment.
If an exception is thrown from messagesAvailableInternal, the catch block in the wrapped code submitted to the call executor catches it and calls internalClose(t) eventually leading to an asynchronous callback from the transport. It still does not handle the race you mentioned. Instead I'm now invoking closedInternal synchronously when the error is detected. This synchronously sets call.cancelled = true and cancels the context before returning from the executor task.
There was a problem hiding this comment.
Now doesn't this code call the application's onCancel() twice? It calls it once here with the call to closedInternal(), and then later when the cancellation is actually processed.
…to avoid races with more messages or halfClose after the error scenario was observed.
2 participants
For calls where the client sends at most one message (Unary and Server Streaming), the incoming message is deserialized immediately but there can be a delay before the client halfcloses, only at which point in time the the deserialized message is needed . This change delays the deserialization of incoming messages for such calls 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.