netty: Per-rpc call option authority verification against peer cert subject names

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

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2025 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import io.grpc.Status;
+
+/** Verifier for the outgoing authority pseudo-header against peer cert. */
+public interface AuthorityVerifier {
+  Status verifyAuthority(String authority);
+}

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

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+import java.security.cert.Certificate;
+import java.security.cert.CertificateException;
+import java.security.cert.CertificateFactory;
+import java.security.cert.X509Certificate;
+import java.util.Collection;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.security.auth.x500.X500Principal;
+
+/**
+ * Contains certificate/key PEM file utility method(s) for internal usage.
+ */
+public final class CertificateUtils {
+  /**
+   * Creates X509TrustManagers using the provided CA certs.
+   */
+  public static TrustManager[] createTrustManager(InputStream rootCerts)
+      throws GeneralSecurityException {
+    KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
+    try {
+      ks.load(null, null);
+    } catch (IOException ex) {
+      // Shouldn't really happen, as we're not loading any data.
+      throw new GeneralSecurityException(ex);
+    }
+    X509Certificate[] certs = CertificateUtils.getX509Certificates(rootCerts);
+    for (X509Certificate cert : certs) {
+      X500Principal principal = cert.getSubjectX500Principal();
+      ks.setCertificateEntry(principal.getName("RFC2253"), cert);
+    }
+
+    TrustManagerFactory trustManagerFactory =
+        TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+    trustManagerFactory.init(ks);
+    return trustManagerFactory.getTrustManagers();
+  }
+
+  private static X509Certificate[] getX509Certificates(InputStream inputStream)
+      throws CertificateException {
+    CertificateFactory factory = CertificateFactory.getInstance("X.509");
+    Collection<? extends Certificate> certs = factory.generateCertificates(inputStream);
+    return certs.toArray(new X509Certificate[0]);
+  }
+}

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

@@ -42,5 +42,8 @@ public final class GrpcAttributes {
   public static final Attributes.Key<Attributes> ATTR_CLIENT_EAG_ATTRS =
       Attributes.Key.create("io.grpc.internal.GrpcAttributes.clientEagAttrs");
 
+  public static final Attributes.Key<AuthorityVerifier> ATTR_AUTHORITY_VERIFIER =
+          Attributes.Key.create("io.grpc.internal.GrpcAttributes.authorityVerifier");
+
   private GrpcAttributes() {}
 }

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

@@ -0,0 +1,132 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import java.security.Principal;
+import java.security.cert.Certificate;
+import javax.net.ssl.SSLPeerUnverifiedException;
+import javax.net.ssl.SSLSession;
+import javax.net.ssl.SSLSessionContext;
+
+/** A no-op ssl session, to facilitate overriding only the required methods in specific
+ * implementations.
+ */
+public class NoopSslSession implements SSLSession {
+  @Override
+  public byte[] getId() {
+    return new byte[0];
+  }
+
+  @Override
+  public SSLSessionContext getSessionContext() {
+    return null;
+  }
+
+  @Override
+  @SuppressWarnings("deprecation")
+  public javax.security.cert.X509Certificate[] getPeerCertificateChain() {
+    throw new UnsupportedOperationException("This method is deprecated and marked for removal. "
+            + "Use the getPeerCertificates() method instead.");
+  }
+
+  @Override
+  public long getCreationTime() {
+    return 0;
+  }
+
+  @Override
+  public long getLastAccessedTime() {
+    return 0;
+  }
+
+  @Override
+  public void invalidate() {
+  }
+
+  @Override
+  public boolean isValid() {
+    return false;
+  }
+
+  @Override
+  public void putValue(String s, Object o) {
+  }
+
+  @Override
+  public Object getValue(String s) {
+    return null;
+  }
+
+  @Override
+  public void removeValue(String s) {
+  }
+
+  @Override
+  public String[] getValueNames() {
+    return new String[0];
+  }
+
+  @Override
+  public Certificate[] getPeerCertificates() throws SSLPeerUnverifiedException {
+    return new Certificate[0];
+  }
+
+  @Override
+  public Certificate[] getLocalCertificates() {
+    return new Certificate[0];
+  }
+
+  @Override
+  public Principal getPeerPrincipal() throws SSLPeerUnverifiedException {
+    return null;
+  }
+
+  @Override
+  public Principal getLocalPrincipal() {
+    return null;
+  }
+
+  @Override
+  public String getCipherSuite() {
+    return null;
+  }
+
+  @Override
+  public String getProtocol() {
+    return null;
+  }
+
+  @Override
+  public String getPeerHost() {
+    return null;
+  }
+
+  @Override
+  public int getPeerPort() {
+    return 0;
+  }
+
+  @Override
+  public int getPacketBufferSize() {
+    return 0;
+  }
+
+  @Override
+  public int getApplicationBufferSize() {
+    return 0;
+  }
+}

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

@@ -66,6 +66,16 @@
     this.preHeaders = preHeaders;
   }
 
+  @Override
+  public CharSequence authority() {
+    for (int i = 0; i < preHeaders.length / 2; i++) {
+      if (preHeaders[i * 2].equals(Http2Headers.PseudoHeaderName.AUTHORITY.value())) {
+        return preHeaders[i * 2 + 1];
+      }
+    }
+    return null;
+  }
+
   @Override
   @SuppressWarnings("ReferenceEquality") // STATUS.value() never changes.
   public CharSequence status() {

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

@@ -44,7 +44,7 @@
           ObjectPool<? extends Executor> executorPool,
           Optional<Runnable> handshakeCompleteRunnable) {
     final io.grpc.netty.ProtocolNegotiator negotiator = ProtocolNegotiators.tls(sslContext,
-        executorPool, handshakeCompleteRunnable);
+        executorPool, handshakeCompleteRunnable, null);
     final class TlsNegotiator implements InternalProtocolNegotiator.ProtocolNegotiator {
 
       @Override
@@ -170,7 +170,7 @@
       ChannelHandler next, SslContext sslContext, String authority,
       ChannelLogger negotiationLogger) {
     return new ClientTlsHandler(next, sslContext, authority, null, negotiationLogger,
-        Optional.absent());
+        Optional.absent(), null, null);
   }
 
   public static class ProtocolNegotiationHandler

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

@@ -652,7 +652,7 @@ static ProtocolNegotiator createProtocolNegotiatorByType(
       case PLAINTEXT_UPGRADE:
         return ProtocolNegotiators.plaintextUpgrade();
       case TLS:
-        return ProtocolNegotiators.tls(sslContext, executorPool, Optional.absent());
+        return ProtocolNegotiators.tls(sslContext, executorPool, Optional.absent(), null);
       default:
         throw new IllegalArgumentException("Unsupported negotiationType: " + negotiationType);
     }

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;
@@ -83,6 +84,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 +97,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 +133,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 +603,56 @@ private void createStream(CreateStreamCommand command, ChannelPromise promise)
       return;
     }
 
+    CharSequence authorityHeader = command.headers().authority();
+    if (authorityHeader == null) {
+      Status authorityVerificationStatus = Status.UNAVAILABLE.withDescription(
+              "Missing authority header");
+      command.stream().setNonExistent();
+      command.stream().transportReportStatus(
+              Status.UNAVAILABLE, RpcProgress.PROCESSED, true, new Metadata());
+      promise.setFailure(InternalStatus.asRuntimeExceptionWithoutStacktrace(
+              authorityVerificationStatus, null));
+      return;
+    }
+    // No need to verify authority for the rpc outgoing header if it is same as the authority
+    // for the transport
+    if (!authority.contentEquals(authorityHeader)) {
+      Status authorityVerificationStatus = peerVerificationResults.get(
+              authorityHeader.toString());
+      if (authorityVerificationStatus == null) {
+        if (attributes.get(GrpcAttributes.ATTR_AUTHORITY_VERIFIER) == null) {
+          authorityVerificationStatus = Status.UNAVAILABLE.withDescription(
+                  "Authority verifier not found to verify authority");
+          command.stream().setNonExistent();
+          command.stream().transportReportStatus(
+                  authorityVerificationStatus, RpcProgress.PROCESSED, true, new Metadata());
+          promise.setFailure(InternalStatus.asRuntimeExceptionWithoutStacktrace(
+                  authorityVerificationStatus, null));
+          return;
+        }
+        authorityVerificationStatus = attributes.get(GrpcAttributes.ATTR_AUTHORITY_VERIFIER)
+                .verifyAuthority(authorityHeader.toString());
+        peerVerificationResults.put(authorityHeader.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));
+        }
+      }
+      if (!authorityVerificationStatus.isOk()) {
+        if (enablePerRpcAuthorityCheck) {
+          command.stream().setNonExistent();
+          command.stream().transportReportStatus(
+                  authorityVerificationStatus, RpcProgress.PROCESSED, true, new Metadata());
+          promise.setFailure(InternalStatus.asRuntimeExceptionWithoutStacktrace(
+                  authorityVerificationStatus, null));
+          return;
+        }
+      }
+    }
     // Get the stream ID for the new stream.
     int streamId;
     try {

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

@@ -106,6 +106,7 @@ class NettyClientTransport implements ConnectionClientTransport {
   private final boolean useGetForSafeMethods;
   private final Ticker ticker;
 
+
   NettyClientTransport(
       SocketAddress address,
       ChannelFactory<? extends Channel> channelFactory,

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

@@ -34,6 +34,6 @@
     Preconditions.checkArgument(sslContext.isClient(),
         "Server SSL context can not be used for client channel");
     GrpcSslContexts.ensureAlpnAndH2Enabled(sslContext.applicationProtocolNegotiator());
-    return NettyChannelCredentials.create(ProtocolNegotiators.tlsClientFactory(sslContext));
+    return NettyChannelCredentials.create(ProtocolNegotiators.tlsClientFactory(sslContext, null));
   }
 }