okhttp: Add maxInboundMetadataSize

Fixes #4567

okhttp/src/main/java/io/grpc/okhttp/OkHttpChannelBuilder.java

@@ -161,6 +161,7 @@ public class OkHttpChannelBuilder extends
   private long keepAliveTimeoutNanos = DEFAULT_KEEPALIVE_TIMEOUT_NANOS;
   private int flowControlWindow = DEFAULT_FLOW_CONTROL_WINDOW;
   private boolean keepAliveWithoutCalls;
+  private int maxInboundMetadataSize = Integer.MAX_VALUE;
 
   protected OkHttpChannelBuilder(String host, int port) {
     this(GrpcUtil.authorityFromHostAndPort(host, port));
@@ -405,6 +406,25 @@ public class OkHttpChannelBuilder extends
     return this;
   }
 
+  /**
+   * Sets the maximum size of metadata allowed to be received. {@code Integer.MAX_VALUE} disables
+   * the enforcement. Defaults to no limit ({@code Integer.MAX_VALUE}).
+   *
+   * <p>The implementation does not currently limit memory usage; this value is checked only after
+   * the metadata is decoded from the wire. It does prevent large metadata from being passed to the
+   * application.
+   *
+   * @param bytes the maximum size of received metadata
+   * @return this
+   * @throws IllegalArgumentException if bytes is non-positive
+   * @since 1.17.0
+   */
+  public OkHttpChannelBuilder maxInboundMetadataSize(int bytes) {
+    Preconditions.checkArgument(bytes > 0, "maxInboundMetadataSize must be > 0");
+    this.maxInboundMetadataSize = bytes;
+    return this;
+  }
+
   @Override
   @Internal
   protected final ClientTransportFactory buildTransportFactory() {
@@ -412,7 +432,7 @@ public class OkHttpChannelBuilder extends
     return new OkHttpTransportFactory(transportExecutor, scheduledExecutorService,
         createSocketFactory(), hostnameVerifier, connectionSpec, maxInboundMessageSize(),
         enableKeepAlive, keepAliveTimeNanos, keepAliveTimeoutNanos, flowControlWindow,
-        keepAliveWithoutCalls, transportTracerFactory);
+        keepAliveWithoutCalls, maxInboundMetadataSize, transportTracerFactory);
   }
 
   @Override
@@ -491,6 +511,7 @@ public class OkHttpChannelBuilder extends
     private final long keepAliveTimeoutNanos;
     private final int flowControlWindow;
     private final boolean keepAliveWithoutCalls;
+    private final int maxInboundMetadataSize;
     private final ScheduledExecutorService timeoutService;
     private boolean closed;
 
@@ -505,6 +526,7 @@ public class OkHttpChannelBuilder extends
         long keepAliveTimeoutNanos,
         int flowControlWindow,
         boolean keepAliveWithoutCalls,
+        int maxInboundMetadataSize,
         TransportTracer.Factory transportTracerFactory) {
       usingSharedScheduler = timeoutService == null;
       this.timeoutService = usingSharedScheduler
@@ -518,6 +540,7 @@ public class OkHttpChannelBuilder extends
       this.keepAliveTimeoutNanos = keepAliveTimeoutNanos;
       this.flowControlWindow = flowControlWindow;
       this.keepAliveWithoutCalls = keepAliveWithoutCalls;
+      this.maxInboundMetadataSize = maxInboundMetadataSize;
 
       usingSharedExecutor = executor == null;
       this.transportTracerFactory =
@@ -556,6 +579,7 @@ public class OkHttpChannelBuilder extends
           flowControlWindow,
           options.getProxyParameters(),
           tooManyPingsRunnable,
+          maxInboundMetadataSize,
           transportTracerFactory.create());
       if (enableKeepAlive) {
         transport.enableKeepAlive(

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java

@@ -192,6 +192,7 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
   private long keepAliveTimeoutNanos;
   private boolean keepAliveWithoutCalls;
   private final Runnable tooManyPingsRunnable;
+  private final int maxInboundMetadataSize;
   @GuardedBy("lock")
   private final TransportTracer transportTracer;
   @GuardedBy("lock")
@@ -223,7 +224,7 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
       Executor executor, @Nullable SSLSocketFactory sslSocketFactory,
       @Nullable HostnameVerifier hostnameVerifier, ConnectionSpec connectionSpec,
       int maxMessageSize, int initialWindowSize, @Nullable ProxyParameters proxy,
-      Runnable tooManyPingsRunnable, TransportTracer transportTracer) {
+      Runnable tooManyPingsRunnable, int maxInboundMetadataSize, TransportTracer transportTracer) {
     this.address = Preconditions.checkNotNull(address, "address");
     this.defaultAuthority = authority;
     this.maxMessageSize = maxMessageSize;
@@ -241,6 +242,7 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
     this.proxy = proxy;
     this.tooManyPingsRunnable =
         Preconditions.checkNotNull(tooManyPingsRunnable, "tooManyPingsRunnable");
+    this.maxInboundMetadataSize = maxInboundMetadataSize;
     this.transportTracer = Preconditions.checkNotNull(transportTracer);
     initTransportTracer();
   }
@@ -281,6 +283,7 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
     this.proxy = null;
     this.tooManyPingsRunnable =
         Preconditions.checkNotNull(tooManyPingsRunnable, "tooManyPingsRunnable");
+    this.maxInboundMetadataSize = Integer.MAX_VALUE;
     this.transportTracer = Preconditions.checkNotNull(transportTracer, "transportTracer");
     initTransportTracer();
   }
@@ -1067,6 +1070,18 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
         List<Header> headerBlock,
         HeadersMode headersMode) {
       boolean unknownStream = false;
+      Status failedStatus = null;
+      if (maxInboundMetadataSize != Integer.MAX_VALUE) {
+        int metadataSize = headerBlockSize(headerBlock);
+        if (metadataSize > maxInboundMetadataSize) {
+          failedStatus = Status.RESOURCE_EXHAUSTED.withDescription(
+              String.format(
+                  "Response %s metadata larger than %d: %d",
+                  inFinished ? "trailer" : "header",
+                  maxInboundMetadataSize,
+                  metadataSize));
+        }
+      }
       synchronized (lock) {
         OkHttpClientStream stream = streams.get(streamId);
         if (stream == null) {
@@ -1076,7 +1091,14 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
             unknownStream = true;
           }
         } else {
-          stream.transportState().transportHeadersReceived(headerBlock, inFinished);
+          if (failedStatus == null) {
+            stream.transportState().transportHeadersReceived(headerBlock, inFinished);
+          } else {
+            if (!inFinished) {
+              frameWriter.rstStream(streamId, ErrorCode.CANCEL);
+            }
+            stream.transportState().transportReportStatus(failedStatus, false, new Metadata());
+          }
         }
       }
       if (unknownStream) {
@@ -1085,6 +1107,17 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
       }
     }
 
+    private int headerBlockSize(List<Header> headerBlock) {
+      // Calculate as defined for SETTINGS_MAX_HEADER_LIST_SIZE in RFC 7540 §6.5.2.
+      long size = 0;
+      for (int i = 0; i < headerBlock.size(); i++) {
+        Header header = headerBlock.get(i);
+        size += 32 + header.name.size() + header.value.size();
+      }
+      size = Math.min(size, Integer.MAX_VALUE);
+      return (int) size;
+    }
+
     @Override
     public void rstStream(int streamId, ErrorCode errorCode) {
       Status status = toGrpcStatus(errorCode).augmentDescription("Rst Stream");

okhttp/src/test/java/io/grpc/okhttp/OkHttpClientTransportTest.java

@@ -135,6 +135,7 @@ public class OkHttpClientTransportTest {
   private static final String NO_USER = null;
   private static final String NO_PW = null;
   private static final int DEFAULT_START_STREAM_ID = 3;
+  private static final int DEFAULT_MAX_INBOUND_METADATA_SIZE = Integer.MAX_VALUE;
 
   @Rule public final Timeout globalTimeout = Timeout.seconds(10);
 
@@ -245,6 +246,7 @@ public class OkHttpClientTransportTest {
         INITIAL_WINDOW_SIZE,
         NO_PROXY,
         tooManyPingsRunnable,
+        DEFAULT_MAX_INBOUND_METADATA_SIZE,
         transportTracer);
     String s = clientTransport.toString();
     assertTrue("Unexpected: " + s, s.contains("OkHttpClientTransport"));
@@ -1518,6 +1520,7 @@ public class OkHttpClientTransportTest {
         INITIAL_WINDOW_SIZE,
         NO_PROXY,
         tooManyPingsRunnable,
+        DEFAULT_MAX_INBOUND_METADATA_SIZE,
         transportTracer);
 
     String host = clientTransport.getOverridenHost();
@@ -1541,6 +1544,7 @@ public class OkHttpClientTransportTest {
         INITIAL_WINDOW_SIZE,
         NO_PROXY,
         tooManyPingsRunnable,
+        DEFAULT_MAX_INBOUND_METADATA_SIZE,
         new TransportTracer());
 
     ManagedClientTransport.Listener listener = mock(ManagedClientTransport.Listener.class);
@@ -1573,6 +1577,7 @@ public class OkHttpClientTransportTest {
         new ProxyParameters(
             (InetSocketAddress) serverSocket.getLocalSocketAddress(), NO_USER, NO_PW),
         tooManyPingsRunnable,
+        DEFAULT_MAX_INBOUND_METADATA_SIZE,
         transportTracer);
     clientTransport.start(transportListener);
 
@@ -1624,6 +1629,7 @@ public class OkHttpClientTransportTest {
         new ProxyParameters(
             (InetSocketAddress) serverSocket.getLocalSocketAddress(), NO_USER, NO_PW),
         tooManyPingsRunnable,
+        DEFAULT_MAX_INBOUND_METADATA_SIZE,
         transportTracer);
     clientTransport.start(transportListener);
 
@@ -1674,6 +1680,7 @@ public class OkHttpClientTransportTest {
         new ProxyParameters(
             (InetSocketAddress) serverSocket.getLocalSocketAddress(), NO_USER, NO_PW),
         tooManyPingsRunnable,
+        DEFAULT_MAX_INBOUND_METADATA_SIZE,
         transportTracer);
     clientTransport.start(transportListener);
 

okhttp/src/test/java/io/grpc/okhttp/OkHttpTransportTest.java

@@ -20,6 +20,7 @@ import io.grpc.ServerStreamTracer;
 import io.grpc.internal.AccessProtectedHack;
 import io.grpc.internal.ClientTransportFactory;
 import io.grpc.internal.FakeClock;
+import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.InternalServer;
 import io.grpc.internal.ManagedClientTransport;
 import io.grpc.internal.testing.AbstractTransportTest;
@@ -41,6 +42,7 @@ public class OkHttpTransportTest extends AbstractTransportTest {
           .forAddress("localhost", 0)
           .usePlaintext()
           .setTransportTracerFactory(fakeClockTransportTracer)
+          .maxInboundMetadataSize(GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE)
           .buildTransportFactory();
 
   @After
@@ -99,15 +101,10 @@ public class OkHttpTransportTest extends AbstractTransportTest {
     return true;
   }
 
-  // not yet implemented
   @Override
   @org.junit.Test
   @org.junit.Ignore
-  public void clientChecksInboundMetadataSize_header() {}
+  public void clientChecksInboundMetadataSize_trailer() {
-
+    // Server-side is flaky due to https://github.com/netty/netty/pull/8332
-  // not yet implemented
+  }
-  @Override
-  @org.junit.Test
-  @org.junit.Ignore
-  public void clientChecksInboundMetadataSize_trailer() {}
 }