In-progress changes for Authority verify in okhttp transport.

Author: kannanjgithub

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

@@ -360,6 +360,10 @@ public ManagedChannelImplBuilder(SocketAddress directServerAddress, String autho
     InternalConfiguratorRegistry.configureChannelBuilder(this);
   }
 
+  public ChannelCredentials getChannelCredentials() {
+    return channelCredentials;
+  }
+
   @Override
   public ManagedChannelImplBuilder directExecutor() {
     return executor(MoreExecutors.directExecutor());

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

@@ -47,6 +47,7 @@
 import io.grpc.internal.SharedResourceHolder.Resource;
 import io.grpc.internal.SharedResourcePool;
 import io.grpc.internal.TransportTracer;
+import io.grpc.internal.TransportTracer.Factory;
 import io.grpc.okhttp.internal.CipherSuite;
 import io.grpc.okhttp.internal.ConnectionSpec;
 import io.grpc.okhttp.internal.Platform;
@@ -536,7 +537,8 @@ OkHttpTransportFactory buildTransportFactory() {
         keepAliveWithoutCalls,
         maxInboundMetadataSize,
         transportTracerFactory,
-        useGetForSafeMethods);
+        useGetForSafeMethods,
+        managedChannelImplBuilder.getChannelCredentials());
   }
 
   OkHttpChannelBuilder disableCheckAuthority() {
@@ -799,6 +801,7 @@ static final class OkHttpTransportFactory implements ClientTransportFactory {
     private final boolean keepAliveWithoutCalls;
     final int maxInboundMetadataSize;
     final boolean useGetForSafeMethods;
+    private final ChannelCredentials channelCredentials;
     private boolean closed;
 
     private OkHttpTransportFactory(
@@ -815,8 +818,9 @@ private OkHttpTransportFactory(
         int flowControlWindow,
         boolean keepAliveWithoutCalls,
         int maxInboundMetadataSize,
-        TransportTracer.Factory transportTracerFactory,
-        boolean useGetForSafeMethods) {
+        Factory transportTracerFactory,
+        boolean useGetForSafeMethods,
+        ChannelCredentials channelCredentials) {
       this.executorPool = executorPool;
       this.executor = executorPool.getObject();
       this.scheduledExecutorServicePool = scheduledExecutorServicePool;
@@ -834,6 +838,7 @@ private OkHttpTransportFactory(
       this.keepAliveWithoutCalls = keepAliveWithoutCalls;
       this.maxInboundMetadataSize = maxInboundMetadataSize;
       this.useGetForSafeMethods = useGetForSafeMethods;
+      this.channelCredentials = channelCredentials;
 
       this.transportTracerFactory =
           Preconditions.checkNotNull(transportTracerFactory, "transportTracerFactory");
@@ -861,7 +866,8 @@ public void run() {
           options.getUserAgent(),
           options.getEagAttributes(),
           options.getHttpConnectProxiedSocketAddress(),
-          tooManyPingsRunnable);
+          tooManyPingsRunnable,
+          channelCredentials);
       if (enableKeepAlive) {
         transport.enableKeepAlive(
             true, keepAliveTimeNanosState.get(), keepAliveTimeoutNanos, keepAliveWithoutCalls);
@@ -897,7 +903,7 @@ public SwapChannelCredentialsResult swapChannelCredentials(ChannelCredentials ch
           keepAliveWithoutCalls,
           maxInboundMetadataSize,
           transportTracerFactory,
-          useGetForSafeMethods);
+          useGetForSafeMethods, managedChannelImplBuilder.getChannelCredentials());
       return new SwapChannelCredentialsResult(factory, result.callCredentials);
     }
 

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

@@ -29,6 +29,7 @@
 import com.google.common.util.concurrent.SettableFuture;
 import io.grpc.Attributes;
 import io.grpc.CallOptions;
+import io.grpc.ChannelCredentials;
 import io.grpc.ClientStreamTracer;
 import io.grpc.Grpc;
 import io.grpc.HttpConnectProxiedSocketAddress;
@@ -42,6 +43,8 @@
 import io.grpc.Status;
 import io.grpc.Status.Code;
 import io.grpc.StatusException;
+import io.grpc.TlsChannelCredentials;
+import io.grpc.internal.ClientStream;
 import io.grpc.internal.ClientStreamListener.RpcProgress;
 import io.grpc.internal.ConnectionClientTransport;
 import io.grpc.internal.GrpcAttributes;
@@ -54,6 +57,7 @@
 import io.grpc.internal.StatsTraceContext;
 import io.grpc.internal.TransportTracer;
 import io.grpc.okhttp.ExceptionHandlingFrameWriter.TransportExceptionHandler;
+import io.grpc.okhttp.OkHttpChannelBuilder.OkHttpTransportFactory;
 import io.grpc.okhttp.internal.ConnectionSpec;
 import io.grpc.okhttp.internal.Credentials;
 import io.grpc.okhttp.internal.StatusLine;
@@ -82,6 +86,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Random;
 import java.util.concurrent.BrokenBarrierException;
 import java.util.concurrent.CountDownLatch;
@@ -99,6 +104,7 @@
 import javax.net.ssl.SSLSession;
 import javax.net.ssl.SSLSocket;
 import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.X509ExtendedTrustManager;
 import okio.Buffer;
 import okio.BufferedSink;
 import okio.BufferedSource;
@@ -114,6 +120,7 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
       OutboundFlowController.Transport {
   private static final Map<ErrorCode, Status> ERROR_CODE_TO_STATUS = buildErrorCodeToStatusMap();
   private static final Logger log = Logger.getLogger(OkHttpClientTransport.class.getName());
+  private final ChannelCredentials channelCredentials;
 
   private static Map<ErrorCode, Status> buildErrorCodeToStatusMap() {
     Map<ErrorCode, Status> errorToStatus = new EnumMap<>(ErrorCode.class);
@@ -205,6 +212,8 @@ private static Map<ErrorCode, Status> buildErrorCodeToStatusMap() {
   private final boolean useGetForSafeMethods;
   @GuardedBy("lock")
   private final TransportTracer transportTracer;
+  private Optional<X509ExtendedTrustManager> x509ExtendedTrustManager;
+
   @GuardedBy("lock")
   private final InUseStateAggregator<OkHttpClientStream> inUseState =
       new InUseStateAggregator<OkHttpClientStream>() {
@@ -233,13 +242,14 @@ protected void handleNotInUse() {
   SettableFuture<Void> connectedFuture;
 
   public OkHttpClientTransport(
-      OkHttpChannelBuilder.OkHttpTransportFactory transportFactory,
+      OkHttpTransportFactory transportFactory,
       InetSocketAddress address,
       String authority,
       @Nullable String userAgent,
       Attributes eagAttrs,
       @Nullable HttpConnectProxiedSocketAddress proxiedAddr,
-      Runnable tooManyPingsRunnable) {
+      Runnable tooManyPingsRunnable,
+      ChannelCredentials channelCredentials) {
     this(
         transportFactory,
         address,
@@ -249,19 +259,21 @@ public OkHttpClientTransport(
         GrpcUtil.STOPWATCH_SUPPLIER,
         new Http2(),
         proxiedAddr,
-        tooManyPingsRunnable);
+        tooManyPingsRunnable,
+        channelCredentials);
   }
 
   private OkHttpClientTransport(
-      OkHttpChannelBuilder.OkHttpTransportFactory transportFactory,
+      OkHttpTransportFactory transportFactory,
       InetSocketAddress address,
       String authority,
       @Nullable String userAgent,
       Attributes eagAttrs,
       Supplier<Stopwatch> stopwatchFactory,
       Variant variant,
       @Nullable HttpConnectProxiedSocketAddress proxiedAddr,
-      Runnable tooManyPingsRunnable) {
+      Runnable tooManyPingsRunnable,
+      ChannelCredentials channelCredentials) {
     this.address = Preconditions.checkNotNull(address, "address");
     this.defaultAuthority = authority;
     this.maxMessageSize = transportFactory.maxMessageSize;
@@ -291,6 +303,7 @@ private OkHttpClientTransport(
     this.attributes = Attributes.newBuilder()
         .set(GrpcAttributes.ATTR_CLIENT_EAG_ATTRS, eagAttrs).build();
     this.useGetForSafeMethods = transportFactory.useGetForSafeMethods;
+    this.channelCredentials = channelCredentials;
     initTransportTracer();
   }
 
@@ -316,7 +329,8 @@ private OkHttpClientTransport(
         stopwatchFactory,
         variant,
         null,
-        tooManyPingsRunnable);
+        tooManyPingsRunnable,
+        null);
     this.connectingCallback = connectingCallback;
     this.connectedFuture = Preconditions.checkNotNull(connectedFuture, "connectedFuture");
   }
@@ -389,13 +403,18 @@ public void ping(final PingCallback callback, Executor executor) {
   }
 
   @Override
-  public OkHttpClientStream newStream(
+  public ClientStream newStream(
       MethodDescriptor<?, ?> method, Metadata headers, CallOptions callOptions,
       ClientStreamTracer[] tracers) {
     Preconditions.checkNotNull(method, "method");
     Preconditions.checkNotNull(headers, "headers");
     StatsTraceContext statsTraceContext =
         StatsTraceContext.newClientContext(tracers, getAttributes(), headers);
+    if (callOptions.getAuthority() != null && channelCredentials instanceof TlsChannelCredentials) {
+      if (x509ExtendedTrustManager == null) {
+
+      }
+    }
     // FIXME: it is likely wrong to pass the transportTracer here as it'll exit the lock's scope
     synchronized (lock) { // to make @GuardedBy linter happy
       return new OkHttpClientStream(

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

@@ -241,7 +241,7 @@ public void testToString() throws Exception {
         /*userAgent=*/ null,
         EAG_ATTRS,
         NO_PROXY,
-        tooManyPingsRunnable);
+        tooManyPingsRunnable, channelCredentials);
     String s = clientTransport.toString();
     assertTrue("Unexpected: " + s, s.contains("OkHttpClientTransport"));
     assertTrue("Unexpected: " + s, s.contains(address.toString()));
@@ -259,7 +259,7 @@ public void testTransportExecutorWithTooFewThreads() throws Exception {
         null,
         EAG_ATTRS,
         NO_PROXY,
-        tooManyPingsRunnable);
+        tooManyPingsRunnable, channelCredentials);
     clientTransport.start(transportListener);
     ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
     verify(transportListener, timeout(TIME_OUT_MS)).transportShutdown(statusCaptor.capture());
@@ -1726,7 +1726,7 @@ public void invalidAuthorityPropagates() {
         "userAgent",
         EAG_ATTRS,
         NO_PROXY,
-        tooManyPingsRunnable);
+        tooManyPingsRunnable, channelCredentials);
 
     String host = clientTransport.getOverridenHost();
     int port = clientTransport.getOverridenPort();
@@ -1744,7 +1744,7 @@ public void unreachableServer() throws Exception {
         "userAgent",
         EAG_ATTRS,
         NO_PROXY,
-        tooManyPingsRunnable);
+        tooManyPingsRunnable, channelCredentials);
 
     ManagedClientTransport.Listener listener = mock(ManagedClientTransport.Listener.class);
     clientTransport.start(listener);
@@ -1774,7 +1774,7 @@ public void customSocketFactory() throws Exception {
             "userAgent",
             EAG_ATTRS,
             NO_PROXY,
-            tooManyPingsRunnable);
+            tooManyPingsRunnable, channelCredentials);
 
     ManagedClientTransport.Listener listener = mock(ManagedClientTransport.Listener.class);
     clientTransport.start(listener);
@@ -1799,7 +1799,7 @@ public void proxy_200() throws Exception {
             .setTargetAddress(targetAddress)
             .setProxyAddress(new InetSocketAddress("localhost", serverSocket.getLocalPort()))
             .build(),
-        tooManyPingsRunnable);
+        tooManyPingsRunnable, channelCredentials);
     clientTransport.start(transportListener);
 
     Socket sock = serverSocket.accept();
@@ -1848,7 +1848,7 @@ public void proxy_500() throws Exception {
             .setTargetAddress(targetAddress)
             .setProxyAddress(new InetSocketAddress("localhost", serverSocket.getLocalPort()))
             .build(),
-        tooManyPingsRunnable);
+        tooManyPingsRunnable, channelCredentials);
     clientTransport.start(transportListener);
 
     Socket sock = serverSocket.accept();
@@ -1896,7 +1896,7 @@ public void proxy_immediateServerClose() throws Exception {
             .setTargetAddress(targetAddress)
             .setProxyAddress(new InetSocketAddress("localhost", serverSocket.getLocalPort()))
             .build(),
-        tooManyPingsRunnable);
+        tooManyPingsRunnable, channelCredentials);
     clientTransport.start(transportListener);
 
     Socket sock = serverSocket.accept();
@@ -1927,7 +1927,7 @@ public void proxy_serverHangs() throws Exception {
             .setTargetAddress(targetAddress)
             .setProxyAddress(new InetSocketAddress("localhost", serverSocket.getLocalPort()))
             .build(),
-        tooManyPingsRunnable);
+        tooManyPingsRunnable, channelCredentials);
     clientTransport.proxySocketTimeout = 10;
     clientTransport.start(transportListener);