Review comments.

grpc · kannanjgithub · Feb 24, 2025 · Nov 18, 2024 · Nov 26, 2024 · Nov 27, 2024
commit 671dee06f9dbaafcc706b67b5621558d6d17d5bf
diff --git a/core/src/main/java/io/grpc/internal/CertificateUtils.java b/core/src/main/java/io/grpc/internal/CertificateUtils.java
@@ -32,7 +32,7 @@
 /**
  * Contains certificate/key PEM file utility method(s) for internal usage.
  */
-public class CertificateUtils {
+public final class CertificateUtils {
   /**
    * Creates X509TrustManagers using the provided CA certs.
    */

diff --git a/netty/src/main/java/io/grpc/netty/GrpcHttp2OutboundHeaders.java b/netty/src/main/java/io/grpc/netty/GrpcHttp2OutboundHeaders.java
@@ -46,13 +46,14 @@ static GrpcHttp2OutboundHeaders clientRequestHeaders(byte[][] serializedMetadata
     return new GrpcHttp2OutboundHeaders(preHeaders, serializedMetadata);
   }
 
-  String getAuthority() {
+  @Override
+  public String authority() {
     for (int i = 0; i < preHeaders.length / 2; i++) {
       if (preHeaders[i].equals(Http2Headers.PseudoHeaderName.AUTHORITY.value())) {
         return preHeaders[i + 1].toString();
       }
     }
-    return null;
+    return "";
   }
 
   static GrpcHttp2OutboundHeaders serverResponseHeaders(byte[][] serializedMetadata) {

diff --git a/netty/src/main/java/io/grpc/netty/NettyClientHandler.java b/netty/src/main/java/io/grpc/netty/NettyClientHandler.java
@@ -28,6 +28,7 @@
 import io.grpc.Attributes;
 import io.grpc.ChannelLogger;
 import io.grpc.InternalChannelz;
+import io.grpc.InternalStatus;
 import io.grpc.Metadata;
 import io.grpc.Status;
 import io.grpc.StatusException;
@@ -586,17 +587,6 @@ protected boolean isGracefulShutdownComplete() {
         && ((StreamBufferingEncoder) encoder()).numBufferedStreams() == 0;
   }
 
-  private String getAuthorityPseudoHeader(Http2Headers http2Headers) {
-    if (http2Headers instanceof GrpcHttp2OutboundHeaders) {
-      return ((GrpcHttp2OutboundHeaders) http2Headers).getAuthority();
-    }
-    try {
-      return http2Headers.authority().toString();
-    } catch (UnsupportedOperationException e) {
-      return null;
-    }
-  }
-
   /**
    * Attempts to create a new stream from the given command. If there are too many active streams,
    * the creation request is queued.
@@ -613,28 +603,48 @@ private void createStream(CreateStreamCommand command, ChannelPromise promise)
       return;
     }
 
-    String authority = getAuthorityPseudoHeader(command.headers());
+    CharSequence authority = command.headers().authority();
     if (authority != null) {
-      Status authorityVerificationStatus = peerVerificationResults.get(authority);
-      if (authorityVerificationStatus == null && attributes != null
-          && attributes.get(GrpcAttributes.ATTR_AUTHORITY_VERIFIER) != null) {
-        authorityVerificationStatus = attributes.get(GrpcAttributes.ATTR_AUTHORITY_VERIFIER)
-            .verifyAuthority(((GrpcHttp2OutboundHeaders) command.headers()).getAuthority());
-        peerVerificationResults.put(authority, authorityVerificationStatus);
+      Status authorityVerificationStatus = peerVerificationResults.get(authority.toString());
+      if (authorityVerificationStatus == null) {
+        if (attributes != null
+                && attributes.get(GrpcAttributes.ATTR_AUTHORITY_VERIFIER) != null) {
+          authorityVerificationStatus = attributes.get(GrpcAttributes.ATTR_AUTHORITY_VERIFIER)
+                  .verifyAuthority(authority.toString());
+          peerVerificationResults.put(authority.toString(), authorityVerificationStatus);
+          if (!authorityVerificationStatus.isOk() && !enablePerRpcAuthorityCheck) {
+            logger.log(Level.WARNING, String.format("%s.%s",
+                            authorityVerificationStatus.getDescription(), enablePerRpcAuthorityCheck
+                                    ? "" : " This will be an error in the future."),
+                    InternalStatus.asRuntimeExceptionWithoutStacktrace(authorityVerificationStatus, null));
+          }
+        } else {
+          authorityVerificationStatus = Status.UNAVAILABLE.withDescription(
+                  "Authority verifier not found to verify authority");
+          command.stream().setNonExistent();
+          command.stream().transportReportStatus(
+                  authorityVerificationStatus, RpcProgress.DROPPED, true, new Metadata());
+          promise.setFailure(InternalStatus.asRuntimeExceptionWithoutStacktrace(authorityVerificationStatus, null));
+          return;
+        }
       }
       if (authorityVerificationStatus != null && !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());
+                  authorityVerificationStatus, RpcProgress.DROPPED, true, new Metadata());
+          promise.setFailure(InternalStatus.asRuntimeExceptionWithoutStacktrace(authorityVerificationStatus, null));
           return;
         }
       }
+    } else {
+      Status authorityVerificationStatus = Status.UNAVAILABLE.withDescription(
+              "Missing authority header");
+      command.stream().setNonExistent();
+      command.stream().transportReportStatus(
+              Status.UNAVAILABLE, RpcProgress.DROPPED, true, new Metadata());
+      promise.setFailure(InternalStatus.asRuntimeExceptionWithoutStacktrace(authorityVerificationStatus, null));
+      return;
     }
     // Get the stream ID for the new stream.
     int streamId;

diff --git a/netty/src/main/java/io/grpc/netty/ProtocolNegotiators.java b/netty/src/main/java/io/grpc/netty/ProtocolNegotiators.java
@@ -762,8 +762,7 @@ public static ProtocolNegotiator tls(SslContext sslContext,
    */
   public static ProtocolNegotiator tls(SslContext sslContext,
       X509TrustManager x509ExtendedTrustManager) {
-    return tls(sslContext, null, Optional.absent(),
-        x509ExtendedTrustManager);
+    return tls(sslContext, null, Optional.absent(), x509ExtendedTrustManager);
   }
 
   public static ProtocolNegotiator.ClientFactory tlsClientFactory(SslContext sslContext,

diff --git a/netty/src/main/java/io/grpc/netty/X509AuthorityVerifier.java b/netty/src/main/java/io/grpc/netty/X509AuthorityVerifier.java
@@ -28,7 +28,7 @@
 import javax.net.ssl.SSLPeerUnverifiedException;
 import javax.net.ssl.X509TrustManager;
 
-public class X509AuthorityVerifier implements AuthorityVerifier {
+final class X509AuthorityVerifier implements AuthorityVerifier {
   private final SSLEngine sslEngine;
   private final X509TrustManager x509ExtendedTrustManager;
 
@@ -57,17 +57,15 @@ public X509AuthorityVerifier(SSLEngine sslEngine, X509TrustManager x509ExtendedT
 
   @Override
   public Status verifyAuthority(@Nonnull String authority) {
-    // sslEngine won't be set when creating ClientTlsHandler from InternalProtocolNegotiators
-    // for example.
     if (sslEngine == null || x509ExtendedTrustManager == null) {
-      return Status.FAILED_PRECONDITION.withDescription(
+      return Status.UNAVAILABLE.withDescription(
               "Can't allow authority override in rpc when SslEngine or X509ExtendedTrustManager"
                       + " is not available");
     }
     Status peerVerificationStatus;
     try {
       // Because the authority pseudo-header can contain a port number:
-      // https://www.rfc-editor.org/rfc/rfc7540#section-8.1.2.3com
+      // https://www.rfc-editor.org/rfc/rfc7540#section-8.1.2.3
       verifyAuthorityAllowedForPeerCert(removeAnyPortNumber(authority));
       peerVerificationStatus = Status.OK;
     } catch (SSLPeerUnverifiedException | CertificateException | InvocationTargetException
@@ -99,6 +97,9 @@ private void verifyAuthorityAllowedForPeerCert(String authority)
     for (int i = 0; i < peerCertificates.length; i++) {
       x509PeerCertificates[i] = (X509Certificate) peerCertificates[i];
     }
+    if (checkServerTrustedMethod == null) {
+      throw new IllegalStateException("checkServerTrustedMethod not found");
+    }
     checkServerTrustedMethod.invoke(
             x509ExtendedTrustManager, x509PeerCertificates, "RSA", sslEngineWrapper);
   }

diff --git a/netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java b/netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java
@@ -36,6 +36,7 @@
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.same;
@@ -64,6 +65,7 @@
 import io.grpc.internal.ClientStreamListener.RpcProgress;
 import io.grpc.internal.ClientTransport;
 import io.grpc.internal.ClientTransport.PingCallback;
+import io.grpc.internal.GrpcAttributes;
 import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.KeepAliveManager;
 import io.grpc.internal.ManagedClientTransport;
@@ -90,10 +92,12 @@
 import io.netty.handler.codec.http2.Http2Stream;
 import io.netty.util.AsciiString;
 import java.io.InputStream;
+import java.security.cert.CertificateException;
 import java.text.MessageFormat;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Queue;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Handler;
@@ -108,6 +112,7 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
@@ -189,7 +194,11 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
           })
         .when(streamListener)
         .messagesAvailable(ArgumentMatchers.<StreamListener.MessageProducer>any());
-
+    doAnswer((attributes) -> Attributes.newBuilder().set(
+            GrpcAttributes.ATTR_AUTHORITY_VERIFIER,
+            (authority) -> Status.OK).build())
+          .when(listener)
+          .filterTransport(ArgumentMatchers.any(Attributes.class));
     lifecycleManager = new ClientTransportLifecycleManager(listener);
     // This mocks the keepalive manager only for there's in which we verify it. For other tests
     // it'll be null which will be testing if we behave correctly when it's not present.
@@ -919,6 +928,159 @@ public void exceptionCaughtShouldCloseConnection() throws Exception {
     assertFalse(channel().isOpen());
   }
 
+  @Test
+  public void missingAuthorityHeader_streamCreationShouldFail() throws Exception {
+    Http2Headers grpcHeadersWithoutAuthority = new DefaultHttp2Headers()
+            .scheme(HTTPS)
+            .path(as("/fakemethod"))
+            .method(HTTP_METHOD)
+            .add(as("auth"), as("sometoken"))
+            .add(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC)
+            .add(TE_HEADER, TE_TRAILERS);
+    ChannelFuture channelFuture = enqueue(newCreateStreamCommand(grpcHeadersWithoutAuthority, streamTransportState));
+    try {
+      channelFuture.get();
+      fail("Expected stream creation failure");
+    } catch (ExecutionException e) {
+      assertThat(e.getCause().getMessage()).isEqualTo("UNAVAILABLE: Missing authority header");
+    }
+  }
+
+  @Test
+  public void missingAuthorityVerifierInAttributes_streamCreationShouldFail() throws Exception {
+    doAnswer(
+            new Answer<Void>() {
+              @Override
+              public Void answer(InvocationOnMock invocation) throws Throwable {
+                StreamListener.MessageProducer producer =
+                        (StreamListener.MessageProducer) invocation.getArguments()[0];
+                InputStream message;
+                while ((message = producer.next()) != null) {
+                  streamListenerMessageQueue.add(message);
+                }
+                return null;
+              }
+            })
+            .when(streamListener)
+            .messagesAvailable(ArgumentMatchers.<StreamListener.MessageProducer>any());
+    doAnswer((attributes) -> Attributes.EMPTY)
+            .when(listener)
+            .filterTransport(ArgumentMatchers.any(Attributes.class));
+    lifecycleManager = new ClientTransportLifecycleManager(listener);
+    // This mocks the keepalive manager only for there's in which we verify it. For other tests
+    // it'll be null which will be testing if we behave correctly when it's not present.
+    if (setKeepaliveManagerFor.contains(testNameRule.getMethodName())) {
+      mockKeepAliveManager = mock(KeepAliveManager.class);
+    }
+
+    initChannel(new GrpcHttp2ClientHeadersDecoder(GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE));
+    streamTransportState = new TransportStateImpl(
+            handler(),
+            channel().eventLoop(),
+            DEFAULT_MAX_MESSAGE_SIZE,
+            transportTracer);
+    streamTransportState.setListener(streamListener);
+
+    grpcHeaders = new DefaultHttp2Headers()
+            .scheme(HTTPS)
+            .authority(as("www.fake.com"))
+            .path(as("/fakemethod"))
+            .method(HTTP_METHOD)
+            .add(as("auth"), as("sometoken"))
+            .add(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC)
+            .add(TE_HEADER, TE_TRAILERS);
+
+    // Simulate receipt of initial remote settings.
+    ByteBuf serializedSettings = serializeSettings(new Http2Settings());
+    channelRead(serializedSettings);
+    channel().releaseOutbound();
+
+    ChannelFuture channelFuture = createStream();
+    try {
+      channelFuture.get();
+      fail("Expected stream creation failure");
+    } catch (ExecutionException e) {
+      assertThat(e.getCause().getMessage()).isEqualTo("UNAVAILABLE: Authority verifier not found to verify authority");
+    }
+  }
+
+  @Test
+  public void authorityVerificationSuccess_streamCreationSucceeds() throws Exception {
+    NettyClientHandler.enablePerRpcAuthorityCheck = true;
+    try {
+      ChannelFuture channelFuture = createStream();
+      channelFuture.get();
+    } finally {
+      NettyClientHandler.enablePerRpcAuthorityCheck = false;
+    }
+  }
+
+  @Test
+  public void authorityVerificationFailure_streamCreationFails() throws Exception {
+    NettyClientHandler.enablePerRpcAuthorityCheck = true;
+    try {
+      doAnswer(
+              new Answer<Void>() {
+                @Override
+                public Void answer(InvocationOnMock invocation) throws Throwable {
+                  StreamListener.MessageProducer producer =
+                          (StreamListener.MessageProducer) invocation.getArguments()[0];
+                  InputStream message;
+                  while ((message = producer.next()) != null) {
+                    streamListenerMessageQueue.add(message);
+                  }
+                  return null;
+                }
+              })
+              .when(streamListener)
+              .messagesAvailable(ArgumentMatchers.<StreamListener.MessageProducer>any());
+      doAnswer((attributes) -> Attributes.newBuilder().set(
+              GrpcAttributes.ATTR_AUTHORITY_VERIFIER,
+              (authority) -> Status.UNAVAILABLE.withCause(
+                      new CertificateException("Peer verification failed"))).build())
+              .when(listener)
+              .filterTransport(ArgumentMatchers.any(Attributes.class));
+      lifecycleManager = new ClientTransportLifecycleManager(listener);
+      // This mocks the keepalive manager only for there's in which we verify it. For other tests
+      // it'll be null which will be testing if we behave correctly when it's not present.
+      if (setKeepaliveManagerFor.contains(testNameRule.getMethodName())) {
+        mockKeepAliveManager = mock(KeepAliveManager.class);
+      }
+
+      initChannel(new GrpcHttp2ClientHeadersDecoder(GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE));
+      streamTransportState = new TransportStateImpl(
+              handler(),
+              channel().eventLoop(),
+              DEFAULT_MAX_MESSAGE_SIZE,
+              transportTracer);
+      streamTransportState.setListener(streamListener);
+
+      grpcHeaders = new DefaultHttp2Headers()
+              .scheme(HTTPS)
+              .authority(as("www.fake.com"))
+              .path(as("/fakemethod"))
+              .method(HTTP_METHOD)
+              .add(as("auth"), as("sometoken"))
+              .add(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC)
+              .add(TE_HEADER, TE_TRAILERS);
+
+      // Simulate receipt of initial remote settings.
+      ByteBuf serializedSettings = serializeSettings(new Http2Settings());
+      channelRead(serializedSettings);
+      channel().releaseOutbound();
+
+      ChannelFuture channelFuture = createStream();
+      try {
+        channelFuture.get();
+        fail("Expected stream creation failure");
+      } catch (ExecutionException e) {
+        assertThat(e.getMessage()).isEqualTo("io.grpc.InternalStatusRuntimeException: UNAVAILABLE");
+      }
+    } finally {
+      NettyClientHandler.enablePerRpcAuthorityCheck = false;
+    }
+  }
+
   @Override
   protected void makeStream() throws Exception {
     createStream();

diff --git a/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java b/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java
@@ -888,7 +888,7 @@ public void authorityOverrideInCallOptions_noX509ExtendedTrustManager_newStreamC
         Status status = ((StatusException) ex.getCause()).getStatus();
         assertThat(status.getDescription()).isEqualTo("Can't allow authority override in rpc "
                 + "when SslEngine or X509ExtendedTrustManager is not available");
-        assertThat(status.getCode()).isEqualTo(Code.FAILED_PRECONDITION);
+        assertThat(status.getCode()).isEqualTo(Code.UNAVAILABLE);
       }
     } finally {
       NettyClientHandler.enablePerRpcAuthorityCheck = false;