diff --git a/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java b/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java index 9873adcb447..c9d9d5f16e2 100644 --- a/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java +++ b/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java @@ -159,6 +159,11 @@ public final void setMessageCompression(boolean enable) { // Ignore. } + @Override + public void setMessageCompression(boolean enabled, String compressorName) { + // Ignore. + } + @Override public void setAuthority(String authority) { // Ignore. diff --git a/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java b/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java index 8a899d621a2..2b1f5ddf163 100644 --- a/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java +++ b/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java @@ -151,6 +151,11 @@ public final void setMessageCompression(boolean enable) { // Ignore. } + @Override + public void setMessageCompression(boolean enabled, String compressorName) { + // Ignore. + } + @Override public void setAuthority(String authority) { // Ignore. diff --git a/core/src/main/java/io/grpc/internal/AbstractClientStream.java b/core/src/main/java/io/grpc/internal/AbstractClientStream.java index bce1820b482..512fc59d57b 100644 --- a/core/src/main/java/io/grpc/internal/AbstractClientStream.java +++ b/core/src/main/java/io/grpc/internal/AbstractClientStream.java @@ -147,6 +147,17 @@ public final void setDecompressorRegistry(DecompressorRegistry decompressorRegis transportState().setDecompressorRegistry(decompressorRegistry); } + /** + * {@inheritDoc} + * + *

Delegates to the transport state so that subclasses (e.g., NettyClientStream, + * InProcessClientStream) inherit the new implementation via their TransportState. + */ + @Override + public void setMessageCompression(boolean enabled, String compressorName) { + transportState().setMessageCompression(enabled, compressorName); + } + /** {@inheritDoc} */ @Override protected abstract TransportState transportState(); @@ -261,6 +272,18 @@ private void setDecompressorRegistry(DecompressorRegistry decompressorRegistry) checkNotNull(decompressorRegistry, "decompressorRegistry"); } + /** + * Sets whether the client is sending a gzip-compressed request. This is called by + * {@link ClientCallImpl#setMessageCompression(boolean)} when the compressor is gzip. + * This information is used to validate the server's {@code grpc-accept-encoding} response header. + * + * @param enabled whether message compression is enabled + * @param compressorName the name of the compressor being used (e.g., "gzip") + */ + public void setMessageCompression(boolean enabled, String compressorName) { + // Default implementation does nothing. Override in Http2ClientStreamTransportState. + } + @VisibleForTesting public final void setListener(ClientStreamListener listener) { checkState(this.listener == null, "Already called setListener"); diff --git a/core/src/main/java/io/grpc/internal/ClientCallImpl.java b/core/src/main/java/io/grpc/internal/ClientCallImpl.java index 4b24b1eae3d..3833deaa8db 100644 --- a/core/src/main/java/io/grpc/internal/ClientCallImpl.java +++ b/core/src/main/java/io/grpc/internal/ClientCallImpl.java @@ -542,7 +542,8 @@ private void sendMessageInternal(ReqT message) { @Override public void setMessageCompression(boolean enabled) { checkState(stream != null, "Not started"); - stream.setMessageCompression(enabled); + String compressorName = callOptions.getCompressor(); + stream.setMessageCompression(enabled, compressorName); } @Override diff --git a/core/src/main/java/io/grpc/internal/ClientStream.java b/core/src/main/java/io/grpc/internal/ClientStream.java index 7dba4d70871..fb7383a8780 100644 --- a/core/src/main/java/io/grpc/internal/ClientStream.java +++ b/core/src/main/java/io/grpc/internal/ClientStream.java @@ -88,6 +88,16 @@ public interface ClientStream extends Stream { */ void setMaxOutboundMessageSize(int maxSize); + /** + * Sets whether the client is sending a gzip-compressed request. This is called by + * {@link ClientCallImpl#setMessageCompression(boolean)} when the compressor is gzip. + * This information is used to validate the server's {@code grpc-accept-encoding} response header. + * + * @param enabled whether message compression is enabled + * @param compressorName the name of the compressor being used (e.g., "gzip") + */ + void setMessageCompression(boolean enabled, String compressorName); + /** * Sets the effective deadline of the RPC. */ diff --git a/core/src/main/java/io/grpc/internal/DelayedStream.java b/core/src/main/java/io/grpc/internal/DelayedStream.java index a2b1e963ac5..1980458b113 100644 --- a/core/src/main/java/io/grpc/internal/DelayedStream.java +++ b/core/src/main/java/io/grpc/internal/DelayedStream.java @@ -447,6 +447,21 @@ public void run() { } } + @Override + public void setMessageCompression(boolean enabled, String compressorName) { + checkState(listener != null, "May only be called after start"); + if (passThrough) { + realStream.setMessageCompression(enabled, compressorName); + } else { + delayOrExecute(new Runnable() { + @Override + public void run() { + realStream.setMessageCompression(enabled, compressorName); + } + }); + } + } + @VisibleForTesting ClientStream getRealStream() { return realStream; diff --git a/core/src/main/java/io/grpc/internal/ForwardingClientStream.java b/core/src/main/java/io/grpc/internal/ForwardingClientStream.java index 22719c9b341..da7428751f5 100644 --- a/core/src/main/java/io/grpc/internal/ForwardingClientStream.java +++ b/core/src/main/java/io/grpc/internal/ForwardingClientStream.java @@ -62,6 +62,11 @@ public void setMessageCompression(boolean enable) { delegate().setMessageCompression(enable); } + @Override + public void setMessageCompression(boolean enabled, String compressorName) { + delegate().setMessageCompression(enabled, compressorName); + } + @Override public void cancel(Status reason) { delegate().cancel(reason); diff --git a/core/src/main/java/io/grpc/internal/Http2ClientStreamTransportState.java b/core/src/main/java/io/grpc/internal/Http2ClientStreamTransportState.java index 7124f2fc88a..b9b4c79c814 100644 --- a/core/src/main/java/io/grpc/internal/Http2ClientStreamTransportState.java +++ b/core/src/main/java/io/grpc/internal/Http2ClientStreamTransportState.java @@ -16,6 +16,9 @@ package io.grpc.internal; +import static io.grpc.internal.GrpcUtil.ACCEPT_ENCODING_SPLITTER; +import static io.grpc.internal.GrpcUtil.MESSAGE_ACCEPT_ENCODING_KEY; + import com.google.common.base.Preconditions; import io.grpc.CallOptions; import io.grpc.InternalMetadata; @@ -24,6 +27,8 @@ import io.grpc.Status; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; +import java.util.logging.Level; +import java.util.logging.Logger; import javax.annotation.Nullable; /** @@ -65,6 +70,14 @@ public Integer parseAsciiString(byte[] serialized) { private Charset errorCharset = StandardCharsets.UTF_8; private boolean headersReceived; + /** + * Tracks whether the client sent a gzip-encoded request. This is set by {@link + * #setMessageCompression(boolean, String)} when the compressor is gzip. + */ + private boolean clientSentGzipRequest = false; + + private static final Logger log = Logger.getLogger(Http2ClientStreamTransportState.class.getName()); + protected Http2ClientStreamTransportState( int maxMessageSize, StatsTraceContext statsTraceCtx, @@ -73,6 +86,21 @@ protected Http2ClientStreamTransportState( super(maxMessageSize, statsTraceCtx, transportTracer, options); } + /** + * Sets whether the client is sending a gzip-compressed request. This is called by + * {@link ClientCallImpl#setMessageCompression(boolean)} when the compressor is gzip. + * This information is used to validate the server's {@code grpc-accept-encoding} response header. + * + * @param enabled whether message compression is enabled + * @param compressorName the name of the compressor being used (e.g., "gzip") + */ + @Override + public final void setMessageCompression(boolean enabled, String compressorName) { + if (enabled && "gzip".equals(compressorName)) { + clientSentGzipRequest = true; + } + } + /** * Called to process a failure in HTTP/2 processing. It should notify the transport to cancel the * stream and call {@code transportReportStatus()}. @@ -109,6 +137,9 @@ protected void transportHeadersReceived(Metadata headers) { return; } + // Validate grpc-accept-encoding header if client sent gzip request + validateGrpcAcceptEncoding(headers); + stripTransportDetails(headers); inboundHeadersReceived(headers); } finally { @@ -257,4 +288,47 @@ private static void stripTransportDetails(Metadata metadata) { metadata.discardAll(InternalStatus.CODE_KEY); metadata.discardAll(InternalStatus.MESSAGE_KEY); } -} + + /** + * Validates that the server's response includes a {@code grpc-accept-encoding} header that + * includes {@code gzip} when the client sent a gzip-encoded request. + * + *

According to the gRPC spec, when a client sends a gzip-encoded request, the server must + * respond with {@code grpc-accept-encoding: gzip} in the response headers to indicate it can + * accept gzip-encoded responses. If this header is missing or doesn't include gzip, it's a + * server misbehavior that we log at FINE level. + * + * @param headers the response headers from the server + */ + private void validateGrpcAcceptEncoding(Metadata headers) { + if (!clientSentGzipRequest) { + // No validation needed if client didn't send gzip + return; + } + + byte[] acceptEncodingBytes = headers.get(MESSAGE_ACCEPT_ENCODING_KEY); + if (acceptEncodingBytes == null) { + log.log(Level.FINE, + "Server sent gzip-encoded request but response missing grpc-accept-encoding header. " + + "This is server misbehavior."); + return; + } + + String acceptEncoding = new String(acceptEncodingBytes, StandardCharsets.US_ASCII); + // Check if gzip is in the accepted encodings (comma-separated list) + Iterable encodingsIterable = GrpcUtil.ACCEPT_ENCODING_SPLITTER.split(acceptEncoding); + boolean gzipAccepted = false; + for (String encoding : encodingsIterable) { + if ("gzip".equalsIgnoreCase(encoding.trim())) { + gzipAccepted = true; + break; + } + } + + if (!gzipAccepted) { + log.log(Level.FINE, + "Server sent gzip-encoded request but grpc-accept-encoding ({0}) does not include gzip. " + + "This is server misbehavior.", acceptEncoding); + } + } +} \ No newline at end of file diff --git a/core/src/main/java/io/grpc/internal/NoopClientStream.java b/core/src/main/java/io/grpc/internal/NoopClientStream.java index d77d72a5412..823d3e2f343 100644 --- a/core/src/main/java/io/grpc/internal/NoopClientStream.java +++ b/core/src/main/java/io/grpc/internal/NoopClientStream.java @@ -68,6 +68,11 @@ public void setMessageCompression(boolean enable) { // noop } + @Override + public void setMessageCompression(boolean enabled, String compressorName) { + // noop + } + @Override public void optimizeForDirectExecutor() {} diff --git a/core/src/main/java/io/grpc/internal/RetriableStream.java b/core/src/main/java/io/grpc/internal/RetriableStream.java index 0c37a0beaca..d422282df26 100644 --- a/core/src/main/java/io/grpc/internal/RetriableStream.java +++ b/core/src/main/java/io/grpc/internal/RetriableStream.java @@ -689,6 +689,18 @@ public void runWith(Substream substream) { delayOrExecute(new MessageCompressionEntry()); } + @Override + public final void setMessageCompression(boolean enabled, String compressorName) { + class MessageCompressionEntry implements BufferEntry { + @Override + public void runWith(Substream substream) { + substream.stream.setMessageCompression(enabled, compressorName); + } + } + + delayOrExecute(new MessageCompressionEntry()); + } + @Override public final void halfClose() { class HalfCloseEntry implements BufferEntry { diff --git a/core/src/test/java/io/grpc/internal/Http2ClientStreamTransportStateGrpcAcceptEncodingTest.java b/core/src/test/java/io/grpc/internal/Http2ClientStreamTransportStateGrpcAcceptEncodingTest.java new file mode 100644 index 00000000000..4601d347cc0 --- /dev/null +++ b/core/src/test/java/io/grpc/internal/Http2ClientStreamTransportStateGrpcAcceptEncodingTest.java @@ -0,0 +1,242 @@ +/* + * Copyright 2026 The gRPC Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.grpc.internal; + +import static io.grpc.internal.ClientStreamListener.RpcProgress.PROCESSED; +import static io.grpc.internal.GrpcUtil.DEFAULT_MAX_MESSAGE_SIZE; +import static io.grpc.internal.GrpcUtil.MESSAGE_ACCEPT_ENCODING_KEY; +import static io.grpc.internal.GrpcUtil.MESSAGE_ENCODING_KEY; +import static java.nio.charset.StandardCharsets.US_ASCII; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.same; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +import io.grpc.CallOptions; +import io.grpc.InternalMetadata; +import io.grpc.Metadata; +import io.grpc.Status; +import io.grpc.Status.Code; +import io.grpc.internal.Http2ClientStreamTransportState; +import io.grpc.internal.TransportTracer; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatchers; +import org.mockito.Captor; +import org.mockito.Mock; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; +import org.mockito.stubbing.Answer; + +/** Unit tests for grpc-accept-encoding validation in {@link Http2ClientStreamTransportState}. */ +@RunWith(JUnit4.class) +public class Http2ClientStreamTransportStateGrpcAcceptEncodingTest { + + @Rule + public final MockitoRule mocks = MockitoJUnit.rule(); + + private final Metadata.Key testStatusMashaller = + InternalMetadata.keyOf(":status", Metadata.ASCII_STRING_MARSHALLER); + + private TransportTracer transportTracer; + @Mock private ClientStreamListener mockListener; + @Captor private ArgumentCaptor statusCaptor; + + @Before + public void setUp() { + transportTracer = new TransportTracer(); + + doAnswer(new Answer() { + @Override + public Void answer(InvocationOnMock invocation) throws Throwable { + StreamListener.MessageProducer producer = + (StreamListener.MessageProducer) invocation.getArguments()[0]; + while (producer.next() != null) {} + return null; + } + }).when(mockListener).messagesAvailable(ArgumentMatchers.any()); + } + + @Test + public void transportHeadersReceived_validGrpcAcceptEncoding_gzip() { + BaseTransportState state = new BaseTransportState(transportTracer); + state.setListener(mockListener); + // Client sent gzip-encoded request + state.setMessageCompression(true, "gzip"); + + Metadata headers = new Metadata(); + headers.put(testStatusMashaller, "200"); + headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), + "application/grpc"); + headers.put(MESSAGE_ACCEPT_ENCODING_KEY, "gzip".getBytes(US_ASCII)); + state.transportHeadersReceived(headers); + + verify(mockListener, never()).closed(any(Status.class), same(PROCESSED), any(Metadata.class)); + verify(mockListener).headersRead(headers); + } + + @Test + public void transportHeadersReceived_missingGrpcAcceptEncoding_whenGzipSent_logsWarning() { + BaseTransportState state = new BaseTransportState(transportTracer); + state.setListener(mockListener); + // Client sent gzip-encoded request + state.setMessageCompression(true, "gzip"); + + Metadata headers = new Metadata(); + headers.put(testStatusMashaller, "200"); + headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), + "application/grpc"); + // No grpc-accept-encoding header when client sent gzip + state.transportHeadersReceived(headers); + + // Should still notify listener but log warning + verify(mockListener).headersRead(headers); + } + + @Test + public void transportHeadersReceived_grpcAcceptEncodingIdentity_whenGzipSent_logsWarning() { + BaseTransportState state = new BaseTransportState(transportTracer); + state.setListener(mockListener); + // Client sent gzip-encoded request + state.setMessageCompression(true, "gzip"); + + Metadata headers = new Metadata(); + headers.put(testStatusMashaller, "200"); + headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), + "application/grpc"); + // Server only accepts identity when client sent gzip + headers.put(MESSAGE_ACCEPT_ENCODING_KEY, "identity".getBytes(US_ASCII)); + state.transportHeadersReceived(headers); + + // Should still notify listener but log warning + verify(mockListener).headersRead(headers); + } + + @Test + public void transportHeadersReceived_grpcAcceptEncodingGzipAndDeflate_whenGzipSent_ok() { + BaseTransportState state = new BaseTransportState(transportTracer); + state.setListener(mockListener); + // Client sent gzip-encoded request + state.setMessageCompression(true, "gzip"); + + Metadata headers = new Metadata(); + headers.put(testStatusMashaller, "200"); + headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), + "application/grpc"); + // Server accepts gzip and deflate + headers.put(MESSAGE_ACCEPT_ENCODING_KEY, "gzip,deflate".getBytes(US_ASCII)); + state.transportHeadersReceived(headers); + + verify(mockListener).headersRead(headers); + } + + @Test + public void transportHeadersReceived_noClientCompression_noWarning() { + BaseTransportState state = new BaseTransportState(transportTracer); + state.setListener(mockListener); + // Client did NOT send compressed request (no setMessageCompression call) + + Metadata headers = new Metadata(); + headers.put(testStatusMashaller, "200"); + headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), + "application/grpc"); + // Client didn't send compressed request, so no validation needed + state.transportHeadersReceived(headers); + + verify(mockListener).headersRead(headers); + } + + @Test + public void transportHeadersReceived_grpcAcceptEncodingMissingOnTrailers() { + BaseTransportState state = new BaseTransportState(transportTracer); + state.setListener(mockListener); + // Client sent gzip-encoded request + state.setMessageCompression(true, "gzip"); + + Metadata headers = new Metadata(); + headers.put(testStatusMashaller, "200"); + headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), + "application/grpc"); + state.transportHeadersReceived(headers); + + Metadata trailers = new Metadata(); + trailers.put(Metadata.Key.of("grpc-status", Metadata.ASCII_STRING_MARSHALLER), "0"); + state.transportTrailersReceived(trailers); + + verify(mockListener).closed(Status.OK, PROCESSED, trailers); + } + + @Test + public void transportHeadersReceived_grpcAcceptEncodingCaseInsensitive() { + BaseTransportState state = new BaseTransportState(transportTracer); + state.setListener(mockListener); + // Client sent gzip-encoded request + state.setMessageCompression(true, "gzip"); + + Metadata headers = new Metadata(); + headers.put(testStatusMashaller, "200"); + headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), + "application/grpc"); + // Server accepts GZIP (uppercase) + headers.put(MESSAGE_ACCEPT_ENCODING_KEY, "GZIP".getBytes(US_ASCII)); + state.transportHeadersReceived(headers); + + verify(mockListener).headersRead(headers); + } + + private static class BaseTransportState extends Http2ClientStreamTransportState { + private int onReadyThreshold; + + public BaseTransportState(TransportTracer transportTracer, CallOptions options) { + super(DEFAULT_MAX_MESSAGE_SIZE, StatsTraceContext.NOOP, transportTracer, options); + } + + public BaseTransportState(TransportTracer transportTracer) { + this(transportTracer, CallOptions.DEFAULT); + } + + @Override + protected void http2ProcessingFailed(Status status, boolean stopDelivery, Metadata trailers) { + transportReportStatus(status, stopDelivery, trailers); + } + + @Override + public void deframeFailed(Throwable cause) {} + + @Override + public void bytesRead(int processedBytes) {} + + @Override + public void runOnTransportThread(Runnable r) { + r.run(); + } + + @Override + void setOnReadyThreshold(int numBytes) { + onReadyThreshold = numBytes; + super.setOnReadyThreshold(numBytes); + } + } +} \ No newline at end of file diff --git a/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java b/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java index a92f10fd5c5..9d8ade99b1a 100644 --- a/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java +++ b/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java @@ -668,6 +668,11 @@ public void setMessageCompression(boolean enable) { // noop } + @Override + public void setMessageCompression(boolean enabled, String compressorName) { + // noop + } + @Override public void optimizeForDirectExecutor() {} @@ -891,9 +896,6 @@ public void halfClose() { syncContext.drain(); } - @Override - public void setMessageCompression(boolean enable) {} - @Override public void setAuthority(String string) { InProcessStream.this.authority = string; @@ -945,6 +947,16 @@ public void setDeadline(Deadline deadline) { @Override public void appendTimeoutInsight(InsightBuilder insight) { } + + @Override + public void setMessageCompression(boolean enable) { + // noop + } + + @Override + public void setMessageCompression(boolean enabled, String compressorName) { + // noop + } } }