Changes.

Author: kannanjgithub

examples/example-tls/src/main/java/io/grpc/examples/helloworldtls/HelloWorldClientTls.java

@@ -16,6 +16,8 @@
 
 package io.grpc.examples.helloworldtls;
 
+import static io.grpc.examples.helloworld.GreeterGrpc.getSayHelloMethod;
+
 import io.grpc.Channel;
 import io.grpc.Grpc;
 import io.grpc.ManagedChannel;
@@ -48,11 +50,13 @@ public HelloWorldClientTls(Channel channel) {
      * Say hello to server.
      */
     public void greet(String name) {
+        System.setProperty("GRPC_ENABLE_PER_RPC_AUTHORITY_CHECK", "true");
         logger.info("Will try to greet " + name + " ...");
         HelloRequest request = HelloRequest.newBuilder().setName(name).build();
         HelloReply response;
         try {
-            response = blockingStub.sayHello(request);
+            response = io.grpc.stub.ClientCalls.blockingUnaryCall(
+                blockingStub.getChannel(), getSayHelloMethod(), blockingStub.getCallOptions().withAuthority("foo.goog.test.in"), request);
         } catch (StatusRuntimeException e) {
             logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
             return;

netty/src/main/java/io/grpc/netty/GrpcHttp2OutboundHeaders.java

@@ -46,6 +46,15 @@ static GrpcHttp2OutboundHeaders clientRequestHeaders(byte[][] serializedMetadata
     return new GrpcHttp2OutboundHeaders(preHeaders, serializedMetadata);
   }
 
+  String getAuthority() {
+    for (int i = 0; i < preHeaders.length / 2; i++) {
+      if (preHeaders[i] == Http2Headers.PseudoHeaderName.AUTHORITY.value()) {
+        return preHeaders[i + 1].toString();
+      }
+    }
+    return null;
+  }
+
   static GrpcHttp2OutboundHeaders serverResponseHeaders(byte[][] serializedMetadata) {
     AsciiString[] preHeaders = new AsciiString[] {
         Http2Headers.PseudoHeaderName.STATUS.value(), Utils.STATUS_OK,

netty/src/main/java/io/grpc/netty/NettyClientHandler.java

@@ -83,6 +83,8 @@
 import io.perfmark.Tag;
 import io.perfmark.TaskCloseable;
 import java.nio.channels.ClosedChannelException;
+import java.util.LinkedHashMap;
+import java.util.Map;
 import java.util.concurrent.Executor;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -94,6 +96,8 @@
  */
 class NettyClientHandler extends AbstractNettyHandler {
   private static final Logger logger = Logger.getLogger(NettyClientHandler.class.getName());
+  static boolean enablePerRpcAuthorityCheck =
+      GrpcUtil.getFlag("GRPC_ENABLE_PER_RPC_AUTHORITY_CHECK", false);
 
   /**
    * A message that simply passes through the channel without any real processing. It is useful to
@@ -128,6 +132,13 @@ protected void handleNotInUse() {
           lifecycleManager.notifyInUse(false);
         }
       };
+  private final Map<String, Status> peerVerificationResults =
+      new LinkedHashMap<String, Status>() {
+        @Override
+        protected boolean removeEldestEntry(Map.Entry<String, Status> eldest) {
+          return size() > 100;
+        }
+      };
 
   private WriteQueue clientWriteQueue;
   private Http2Ping ping;
@@ -591,6 +602,28 @@ private void createStream(CreateStreamCommand command, ChannelPromise promise)
       return;
     }
 
+    String authority = ((GrpcHttp2OutboundHeaders) command.headers()).getAuthority();
+    if (authority != null) {
+      Status authorityVerificationStatus = peerVerificationResults.get(authority);
+      if (authorityVerificationStatus == null) {
+        authorityVerificationStatus = attributes.get(GrpcAttributes.ATTR_AUTHORITY_VERIFIER)
+            .verifyAuthority(((GrpcHttp2OutboundHeaders) command.headers()).getAuthority());
+        peerVerificationResults.put(authority, authorityVerificationStatus);
+      }
+      if (!authorityVerificationStatus.isOk()) {
+        logger.log(Level.WARNING, String.format("%s.%s",
+                authorityVerificationStatus.getDescription(), enablePerRpcAuthorityCheck
+            ? "" : "This will be an error in the future."),
+            authorityVerificationStatus.getCause());
+        if (enablePerRpcAuthorityCheck) {
+          command.stream().setNonExistent();
+          command.stream().transportReportStatus(
+              authorityVerificationStatus, RpcProgress.DROPPED, true, new Metadata());
+          promise.setFailure(authorityVerificationStatus.getCause());
+          return;
+        }
+      }
+    }
     // Get the stream ID for the new stream.
     int streamId;
     try {

netty/src/main/java/io/grpc/netty/NettyClientTransport.java

@@ -60,12 +60,9 @@
 import io.netty.util.concurrent.GenericFutureListener;
 import java.net.SocketAddress;
 import java.nio.channels.ClosedChannelException;
-import java.util.Collections;
-import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
-import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
 
@@ -110,17 +107,7 @@ class NettyClientTransport implements ConnectionClientTransport {
   private final boolean useGetForSafeMethods;
   private final Ticker ticker;
   private final Logger logger = Logger.getLogger(NettyClientTransport.class.getName());
-  private final Map<String, Status> peerVerificationResults = Collections.synchronizedMap(
-      new LinkedHashMap<String, Status>() {
-        @Override
-        protected boolean removeEldestEntry(Map.Entry<String, Status> eldest) {
-          return size() > 100;
-        }
-      });
 
-  @VisibleForTesting
-  static boolean enablePerRpcAuthorityCheck =
-          GrpcUtil.getFlag("GRPC_ENABLE_PER_RPC_AUTHORITY_CHECK", false);
 
   NettyClientTransport(
       SocketAddress address,
@@ -210,25 +197,6 @@ public ClientStream newStream(
     if (channel == null) {
       return new FailingClientStream(statusExplainingWhyTheChannelIsNull, tracers);
     }
-    if (callOptions.getAuthority() != null) {
-      Status verificationStatus = peerVerificationResults.get(callOptions.getAuthority());
-      if (verificationStatus == null) {
-        verificationStatus = negotiator.verifyAuthority(callOptions.getAuthority());
-        peerVerificationResults.put(callOptions.getAuthority(), verificationStatus);
-        if (!verificationStatus.isOk()) {
-          logger.log(Level.WARNING, String.format("Peer hostname verification during rpc failed "
-                          + "for authority '%s' for method '%s' with the error \"%s\". This will "
-                          + "be an error in the future.", callOptions.getAuthority(),
-                  method.getFullMethodName(), verificationStatus.getDescription()),
-                  verificationStatus.getCause());
-        }
-      }
-      if (!verificationStatus.isOk()) {
-        if (enablePerRpcAuthorityCheck) {
-          return new FailingClientStream(verificationStatus, tracers);
-        }
-      }
-    }
     StatsTraceContext statsTraceCtx =
         StatsTraceContext.newClientContext(tracers, getAttributes(), headers);
     return new NettyClientStream(

netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java

@@ -867,7 +867,7 @@ public void tlsNegotiationServerExecutorShouldSucceed() throws Exception {
   public void authorityOverrideInCallOptions_noX509ExtendedTrustManager_newStreamCreationFails()
           throws IOException, InterruptedException, GeneralSecurityException, ExecutionException,
           TimeoutException {
-    NettyClientTransport.enablePerRpcAuthorityCheck = true;
+    NettyClientHandler.enablePerRpcAuthorityCheck = true;
     try {
       startServer();
       InputStream caCert = TlsTesting.loadCert("ca.pem");
@@ -895,15 +895,15 @@ public void authorityOverrideInCallOptions_noX509ExtendedTrustManager_newStreamC
         assertThat(status.getCode()).isEqualTo(Code.FAILED_PRECONDITION);
       }
     } finally {
-      NettyClientTransport.enablePerRpcAuthorityCheck = false;
+      NettyClientHandler.enablePerRpcAuthorityCheck = false;
     }
   }
 
   @Test
   public void authorityOverrideInCallOptions_doesntMatchServerPeerHost_newStreamCreationFails()
           throws IOException, InterruptedException, GeneralSecurityException, ExecutionException,
           TimeoutException {
-    NettyClientTransport.enablePerRpcAuthorityCheck = true;
+    NettyClientHandler.enablePerRpcAuthorityCheck = true;
     try {
       startServer();
       NettyClientTransport transport = newTransport(newNegotiator());
@@ -930,15 +930,15 @@ public void authorityOverrideInCallOptions_doesntMatchServerPeerHost_newStreamCr
                 "No subject alternative DNS name matching foo.test.google.in found.");
       }
     } finally {
-      NettyClientTransport.enablePerRpcAuthorityCheck = false;
+      NettyClientHandler.enablePerRpcAuthorityCheck = false;
     }
   }
 
   @Test
   public void authorityOverrideInCallOptions_matchesServerPeerHost_newStreamCreationSucceeds()
           throws IOException, InterruptedException, GeneralSecurityException, ExecutionException,
           TimeoutException {
-    NettyClientTransport.enablePerRpcAuthorityCheck = true;
+    NettyClientHandler.enablePerRpcAuthorityCheck = true;
     try {
       startServer();
       NettyClientTransport transport = newTransport(newNegotiator());
@@ -951,15 +951,15 @@ public void authorityOverrideInCallOptions_matchesServerPeerHost_newStreamCreati
 
       new Rpc(transport, new Metadata(), "foo.test.google.fr").waitForResponse();
     } finally {
-      NettyClientTransport.enablePerRpcAuthorityCheck = false;;
+      NettyClientHandler.enablePerRpcAuthorityCheck = false;;
     }
   }
 
   @Test
   public void authorityOverrideInCallOptions_lruCache()
           throws IOException, InterruptedException, GeneralSecurityException, ExecutionException,
           TimeoutException {
-    NettyClientTransport.enablePerRpcAuthorityCheck = true;
+    NettyClientHandler.enablePerRpcAuthorityCheck = true;
     try {
       startServer();
       ProtocolNegotiator mockNegotiator =
@@ -997,7 +997,7 @@ public void authorityOverrideInCallOptions_lruCache()
       assertThat(authorityValues.get(100)).isEqualTo("foo-100.test.google.fr");
       assertThat(authorityValues.get(101)).isEqualTo("foo-0.test.google.fr");
     } finally {
-      NettyClientTransport.enablePerRpcAuthorityCheck = false;;
+      NettyClientHandler.enablePerRpcAuthorityCheck = false;;
     }
   }