[common][server] Bug fixes and clean of ACL handlers for GRPC (#982)

- Consolidate boiler plate code in ACL handlers for access validations
- Fix code path that throws exception in GRPC and close the calls with the appropriate error messages
- Fix the flow that invokes onMessage twice in GRPC call intercept to prevent exceptions
- Rename GrpcSSLUtils to GrpcUtils to accommodate general utility methods
- Add reflection service to GRPC server to interact with Grpcurl
- Reuse the header for skipping ACLs from router since the store ACL handling is performed at the router for non FC clients
- [code readability] Move public methods together followed by overridable and protected methods followed by package private and private methods

Co-authored-by: Bharath Kumarasubramanian <[email protected]>

build.gradle

@@ -82,6 +82,7 @@ ext.libraries = [
     fastUtil: 'it.unimi.dsi:fastutil:8.3.0',
     grpcNettyShaded: "io.grpc:grpc-netty-shaded:${grpcVersion}",
     grpcProtobuf: "io.grpc:grpc-protobuf:${grpcVersion}",
+    grpcServices: "io.grpc:grpc-services:${grpcVersion}",
     grpcStub: "io.grpc:grpc-stub:${grpcVersion}",
     hadoopCommon: "org.apache.hadoop:hadoop-common:${hadoopVersion}",
     helix: 'org.apache.helix:helix-core:1.1.0',
@@ -266,6 +267,7 @@ subprojects {
     avroCompiler 'org.slf4j:slf4j-simple:1.7.32'
     implementation libraries.grpcNettyShaded
     implementation libraries.grpcProtobuf
+    implementation libraries.grpcServices
     implementation libraries.grpcStub
     compileOnly libraries.tomcatAnnotations
   }

clients/venice-client/src/main/java/com/linkedin/venice/fastclient/transport/GrpcTransportClient.java

@@ -11,7 +11,7 @@
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.fastclient.GrpcClientConfig;
 import com.linkedin.venice.grpc.GrpcErrorCodes;
-import com.linkedin.venice.grpc.GrpcSslUtils;
+import com.linkedin.venice.grpc.GrpcUtils;
 import com.linkedin.venice.protocols.VeniceClientRequest;
 import com.linkedin.venice.protocols.VeniceReadServiceGrpc;
 import com.linkedin.venice.protocols.VeniceServerResponse;
@@ -64,8 +64,8 @@ public GrpcTransportClient(GrpcClientConfig grpcClientConfig) {
   private void initChannelCredentials() {
     try {
       TlsChannelCredentials.Builder tlsBuilder = TlsChannelCredentials.newBuilder()
-          .keyManager(GrpcSslUtils.getKeyManagers(sslFactory))
-          .trustManager(GrpcSslUtils.getTrustManagers(sslFactory));
+          .keyManager(GrpcUtils.getKeyManagers(sslFactory))
+          .trustManager(GrpcUtils.getTrustManagers(sslFactory));
       channelCredentials = tlsBuilder.build();
     } catch (Exception e) {
       throw new VeniceClientException(

internal/venice-common/src/main/java/com/linkedin/venice/grpc/GrpcSslUtils.java → internal/venice-common/src/main/java/com/linkedin/venice/grpc/GrpcUtils.java

@@ -1,7 +1,13 @@
 package com.linkedin.venice.grpc;
 
+import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.security.SSLConfig;
 import com.linkedin.venice.security.SSLFactory;
+import com.linkedin.venice.utils.SslUtils;
+import io.grpc.Grpc;
+import io.grpc.ServerCall;
+import io.grpc.Status;
+import io.netty.handler.codec.http.HttpResponseStatus;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.file.Files;
@@ -11,13 +17,57 @@
 import java.security.NoSuchAlgorithmException;
 import java.security.UnrecoverableKeyException;
 import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
 import javax.net.ssl.KeyManager;
 import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLPeerUnverifiedException;
+import javax.net.ssl.SSLSession;
 import javax.net.ssl.TrustManager;
 import javax.net.ssl.TrustManagerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 
-public class GrpcSslUtils {
+public final class GrpcUtils {
+  private static final Logger LOGGER = LogManager.getLogger(GrpcUtils.class);
+
+  public static KeyManager[] getKeyManagers(SSLFactory sslFactory)
+      throws UnrecoverableKeyException, CertificateException, KeyStoreException, IOException, NoSuchAlgorithmException {
+    String algorithm = KeyManagerFactory.getDefaultAlgorithm();
+    String password = sslFactory.getSSLConfig().getKeyStorePassword();
+    KeyStore keyStore = loadKeyStore(sslFactory, sslFactory.getSSLConfig().getKeyStoreType());
+    KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(algorithm);
+    keyManagerFactory.init(keyStore, password.toCharArray());
+    return keyManagerFactory.getKeyManagers();
+  }
+
+  public static TrustManager[] getTrustManagers(SSLFactory sslFactory)
+      throws CertificateException, KeyStoreException, IOException, NoSuchAlgorithmException {
+    String algorithm = TrustManagerFactory.getDefaultAlgorithm();
+    KeyStore trustStore = loadTrustStore(sslFactory, sslFactory.getSSLConfig().getTrustStoreType());
+    TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(algorithm);
+    trustManagerFactory.init(trustStore);
+    return trustManagerFactory.getTrustManagers();
+  }
+
+  public static Status httpResponseStatusToGrpcStatus(HttpResponseStatus status, String errorMessage) {
+    if (status.equals(HttpResponseStatus.FORBIDDEN) || status.equals(HttpResponseStatus.UNAUTHORIZED)) {
+      return Status.PERMISSION_DENIED.withDescription(errorMessage);
+    }
+
+    return Status.UNKNOWN.withDescription(errorMessage);
+  }
+
+  public static X509Certificate extractGrpcClientCert(ServerCall<?, ?> call) throws SSLPeerUnverifiedException {
+    SSLSession sslSession = call.getAttributes().get(Grpc.TRANSPORT_ATTR_SSL_SESSION);
+    if (sslSession == null) {
+      LOGGER.error("Cannot obtain SSLSession for authority {}", call.getAuthority());
+      throw new VeniceException("Failed to obtain SSL session");
+    }
+
+    return SslUtils.getX509Certificate(sslSession.getPeerCertificates()[0]);
+  }
+
   private static KeyStore loadKeyStore(SSLFactory sslFactory, String type)
       throws CertificateException, KeyStoreException, IOException, NoSuchAlgorithmException {
     SSLConfig config = sslFactory.getSSLConfig();
@@ -42,43 +92,4 @@ private static KeyStore loadStore(String path, char[] password, String type)
     }
     return keyStore;
   }
-
-  public static KeyManager[] getKeyManagers(SSLFactory sslFactory)
-      throws UnrecoverableKeyException, CertificateException, KeyStoreException, IOException, NoSuchAlgorithmException {
-    String algorithm = KeyManagerFactory.getDefaultAlgorithm();
-    return getKeyManagers(sslFactory, algorithm);
-  }
-
-  public static TrustManager[] getTrustManagers(SSLFactory sslFactory)
-      throws CertificateException, KeyStoreException, IOException, NoSuchAlgorithmException {
-    String algorithm = TrustManagerFactory.getDefaultAlgorithm();
-    return getTrustManagers(sslFactory, algorithm);
-  }
-
-  public static KeyManager[] getKeyManagers(SSLFactory sslFactory, String algorithm)
-      throws CertificateException, KeyStoreException, IOException, NoSuchAlgorithmException, UnrecoverableKeyException {
-    String password = sslFactory.getSSLConfig().getKeyStorePassword();
-    KeyStore keyStore = loadKeyStore(sslFactory, sslFactory.getSSLConfig().getKeyStoreType());
-    return createKeyManagers(keyStore, algorithm, password);
-  }
-
-  public static TrustManager[] getTrustManagers(SSLFactory sslFactory, String algorithm)
-      throws CertificateException, KeyStoreException, IOException, NoSuchAlgorithmException {
-    KeyStore trustStore = loadTrustStore(sslFactory, sslFactory.getSSLConfig().getTrustStoreType());
-    return createTrustManagers(trustStore, algorithm);
-  }
-
-  private static KeyManager[] createKeyManagers(KeyStore keyStore, String algorithm, String password)
-      throws NoSuchAlgorithmException, UnrecoverableKeyException, KeyStoreException {
-    KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(algorithm);
-    keyManagerFactory.init(keyStore, password.toCharArray());
-    return keyManagerFactory.getKeyManagers();
-  }
-
-  private static TrustManager[] createTrustManagers(KeyStore trustStore, String algorithm)
-      throws NoSuchAlgorithmException, KeyStoreException {
-    TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(algorithm);
-    trustManagerFactory.init(trustStore);
-    return trustManagerFactory.getTrustManagers();
-  }
 }

internal/venice-common/src/main/java/com/linkedin/venice/listener/ServerHandlerUtils.java

@@ -1,7 +1,11 @@
 package com.linkedin.venice.listener;
 
+import com.linkedin.venice.exceptions.VeniceException;
+import com.linkedin.venice.utils.SslUtils;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.ssl.SslHandler;
+import java.security.cert.X509Certificate;
+import javax.net.ssl.SSLPeerUnverifiedException;
 
 
 public class ServerHandlerUtils {
@@ -21,4 +25,13 @@ public static SslHandler extractSslHandler(ChannelHandlerContext ctx) {
     }
     return sslHandler;
   }
+
+  public static X509Certificate extractClientCert(ChannelHandlerContext ctx) throws SSLPeerUnverifiedException {
+    SslHandler sslHandler = ServerHandlerUtils.extractSslHandler(ctx);
+    if (sslHandler != null) {
+      return SslUtils.getX509Certificate(sslHandler.engine().getSession().getPeerCertificates()[0]);
+    } else {
+      throw new VeniceException("Failed to extract client cert from the incoming request");
+    }
+  }
 }

internal/venice-common/src/test/java/com/linkedin/venice/grpc/GrpcUtilsTest.java

@@ -0,0 +1,74 @@
+package com.linkedin.venice.grpc;
+
+import static org.mockito.Mockito.*;
+import static org.testng.Assert.*;
+
+import com.linkedin.venice.security.SSLFactory;
+import com.linkedin.venice.utils.SslUtils;
+import io.grpc.Status;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.TrustManager;
+import org.testng.annotations.BeforeTest;
+import org.testng.annotations.Test;
+
+
+public class GrpcUtilsTest {
+  private static SSLFactory sslFactory;
+
+  @BeforeTest
+  public static void setup() {
+    sslFactory = SslUtils.getVeniceLocalSslFactory();
+  }
+
+  @Test
+  public void testGetTrustManagers() throws Exception {
+    TrustManager[] trustManagers = GrpcUtils.getTrustManagers(sslFactory);
+
+    assertNotNull(trustManagers);
+    assertTrue(trustManagers.length > 0);
+  }
+
+  @Test
+  public void testGetKeyManagers() throws Exception {
+    KeyManager[] keyManagers = GrpcUtils.getKeyManagers(sslFactory);
+
+    assertNotNull(keyManagers);
+    assertTrue(keyManagers.length > 0);
+  }
+
+  @Test
+  public void testHttpResponseStatusToGrpcStatus() {
+    final String permissionDeniedErrorMessage = "permission denied error message";
+    Status grpcStatus =
+        GrpcUtils.httpResponseStatusToGrpcStatus(HttpResponseStatus.FORBIDDEN, permissionDeniedErrorMessage);
+
+    assertEquals(
+        grpcStatus.getCode(),
+        Status.PERMISSION_DENIED.getCode(),
+        "Mismatch in GRPC status for the http response status permission denied");
+    assertEquals(
+        permissionDeniedErrorMessage,
+        grpcStatus.getDescription(),
+        "Mismatch in error description for the mapped grpc status");
+
+    final String unauthorizedErrorMessage = "unauthorized error message";
+    grpcStatus = GrpcUtils.httpResponseStatusToGrpcStatus(HttpResponseStatus.UNAUTHORIZED, unauthorizedErrorMessage);
+    assertEquals(
+        grpcStatus.getCode(),
+        Status.PERMISSION_DENIED.getCode(),
+        "Mismatch in GRPC status for the http response status unauthorized");
+    assertEquals(
+        unauthorizedErrorMessage,
+        grpcStatus.getDescription(),
+        "Mismatch in error description for the mapped grpc status");
+
+    final String badRequestErrorMessage = "bad request error message";
+    grpcStatus = GrpcUtils.httpResponseStatusToGrpcStatus(HttpResponseStatus.BAD_REQUEST, badRequestErrorMessage);
+    assertEquals(grpcStatus.getCode(), Status.UNKNOWN.getCode(), "Expected unknown status for everything else");
+    assertEquals(
+        badRequestErrorMessage,
+        grpcStatus.getDescription(),
+        "Mismatch in error description for the mapped grpc status");
+  }
+}

services/venice-server/src/main/java/com/linkedin/venice/grpc/VeniceGrpcServer.java

@@ -8,6 +8,7 @@
 import io.grpc.ServerCredentials;
 import io.grpc.ServerInterceptors;
 import io.grpc.TlsServerCredentials;
+import io.grpc.protobuf.services.ProtoReflectionService;
 import java.io.IOException;
 import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
@@ -40,6 +41,7 @@ public VeniceGrpcServer(VeniceGrpcServerConfig config) {
     server = Grpc.newServerBuilderForPort(config.getPort(), credentials)
         .executor(executor) // TODO: experiment with different executors for best performance
         .addService(ServerInterceptors.intercept(config.getService(), config.getInterceptors()))
+        .addService(ProtoReflectionService.newInstance())
         .build();
   }
 
@@ -58,8 +60,8 @@ private void initServerCredentials() {
 
     try {
       credentials = TlsServerCredentials.newBuilder()
-          .keyManager(GrpcSslUtils.getKeyManagers(sslFactory))
-          .trustManager(GrpcSslUtils.getTrustManagers(sslFactory))
+          .keyManager(GrpcUtils.getKeyManagers(sslFactory))
+          .trustManager(GrpcUtils.getTrustManagers(sslFactory))
           .clientAuth(TlsServerCredentials.ClientAuth.REQUIRE)
           .build();
     } catch (UnrecoverableKeyException | KeyStoreException | CertificateException | IOException