fix: validate grpc-accept-encoding header when client sends gzip request

- Add validation in Http2ClientStreamTransportState to check if server response
  includes grpc-accept-encoding: gzip when client sent gzip-encoded request
- Log warning at FINE level when server misbehavior detected (missing or invalid header)
- Add setMessageCompression(boolean, String) method to ClientStream interface and
  implementations to track when client sends gzip-compressed requests
- Add unit tests for grpc-accept-encoding validation

Fixes #1804

Author: jlaportebot

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.

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.

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

@@ -261,6 +261,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");

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

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.
    */

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;

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);

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

@@ -24,6 +24,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 +67,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 +83,20 @@ 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")
+   */
+  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 +133,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 +284,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.
+   *
+   * <p>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)
+    String[] encodings = GrpcUtil.ACCEPT_ENCODING_SPLITTER.split(acceptEncoding);
+    boolean gzipAccepted = false;
+    for (String encoding : encodings) {
+      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);
+    }
+  }
+}

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() {}
 

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 {