interop: create PSM security xDS interop tests - server & client (#7609)

Co-authored-by: Sergii Tkachenko <hi@sergii.org>

interop-testing/build.gradle

@@ -26,6 +26,7 @@ dependencies {
             project(':grpc-services'),
             project(':grpc-stub'),
             project(':grpc-testing'),
+            project(path: ':grpc-xds', configuration: 'shadow'),
             libraries.google_auth_oauth2_http,
             libraries.junit,
             libraries.truth
@@ -33,8 +34,7 @@ dependencies {
     compileOnly libraries.javax_annotation
     runtimeOnly libraries.opencensus_impl,
             libraries.netty_tcnative,
-            project(':grpc-grpclb'),
+            project(':grpc-grpclb')
-            project(path: ':grpc-xds', configuration: 'shadow')
     testImplementation project(':grpc-context').sourceSets.test.output,
             libraries.mockito
     alpnagent libraries.jetty_alpn_agent

interop-testing/src/main/java/io/grpc/testing/integration/XdsTestClient.java

@@ -33,11 +33,11 @@ import io.grpc.ClientInterceptor;
 import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
 import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
 import io.grpc.Grpc;
+import io.grpc.InsecureChannelCredentials;
 import io.grpc.ManagedChannel;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.Server;
-import io.grpc.netty.NettyChannelBuilder;
 import io.grpc.netty.NettyServerBuilder;
 import io.grpc.stub.StreamObserver;
 import io.grpc.testing.integration.Messages.ClientConfigureRequest;
@@ -49,6 +49,7 @@ import io.grpc.testing.integration.Messages.LoadBalancerStatsRequest;
 import io.grpc.testing.integration.Messages.LoadBalancerStatsResponse;
 import io.grpc.testing.integration.Messages.SimpleRequest;
 import io.grpc.testing.integration.Messages.SimpleResponse;
+import io.grpc.xds.XdsChannelCredentials;
 import java.util.ArrayList;
 import java.util.EnumMap;
 import java.util.HashMap;
@@ -81,6 +82,7 @@ public final class XdsTestClient {
   private int qps = 1;
   private volatile RpcConfig rpcConfig;
   private int rpcTimeoutSec = 20;
+  private boolean secureMode = false;
   private String server = "localhost:8080";
   private int statsPort = 8081;
   private Server statsServer;
@@ -147,6 +149,8 @@ public final class XdsTestClient {
         server = value;
       } else if ("stats_port".equals(key)) {
         statsPort = Integer.valueOf(value);
+      } else if ("secureMode".equals(key)) {
+        secureMode = Boolean.valueOf(value);
       } else {
         System.err.println("Unknown argument: " + key);
         usage = true;
@@ -176,6 +180,8 @@ public final class XdsTestClient {
               + c.rpcTimeoutSec
               + "\n  --server=host:port     Address of server. Default: "
               + c.server
+              + "\n  --secureMode=BOOLEAN   Use true to enable XdsCredentials. Default: "
+              + c.secureMode
               + "\n  --stats_port=INT       Port to expose peer distribution stats service. "
               + "Default: "
               + c.statsPort);
@@ -231,7 +237,13 @@ public final class XdsTestClient {
     try {
       statsServer.start();
       for (int i = 0; i < numChannels; i++) {
-        channels.add(NettyChannelBuilder.forTarget(server).usePlaintext().build());
+        channels.add(
+            Grpc.newChannelBuilder(
+                    server,
+                    secureMode
+                        ? XdsChannelCredentials.create(InsecureChannelCredentials.create())
+                        : InsecureChannelCredentials.create())
+                .build());
       }
       exec = MoreExecutors.listeningDecorator(Executors.newSingleThreadScheduledExecutor());
       runQps();

interop-testing/src/main/java/io/grpc/testing/integration/XdsTestServer.java

@@ -19,6 +19,7 @@ package io.grpc.testing.integration;
 import io.grpc.Context;
 import io.grpc.Contexts;
 import io.grpc.ForwardingServerCall.SimpleForwardingServerCall;
+import io.grpc.InsecureServerCredentials;
 import io.grpc.Metadata;
 import io.grpc.Server;
 import io.grpc.ServerCall;
@@ -32,6 +33,8 @@ import io.grpc.services.HealthStatusManager;
 import io.grpc.stub.StreamObserver;
 import io.grpc.testing.integration.Messages.SimpleRequest;
 import io.grpc.testing.integration.Messages.SimpleResponse;
+import io.grpc.xds.XdsServerBuilder;
+import io.grpc.xds.XdsServerCredentials;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.concurrent.TimeUnit;
@@ -51,9 +54,12 @@ public final class XdsTestServer {
   private static Logger logger = Logger.getLogger(XdsTestServer.class.getName());
 
   private int port = 8080;
+  private int maintenancePort = 8080;
+  private boolean secureMode = false;
   private String serverId = "java_server";
   private HealthStatusManager health;
   private Server server;
+  private Server maintenanceServer;
   private String host;
 
   /**
@@ -103,6 +109,10 @@ public final class XdsTestServer {
       String value = parts[1];
       if ("port".equals(key)) {
         port = Integer.valueOf(value);
+      } else if ("maintenance_port".equals(key)) {
+        maintenancePort = Integer.valueOf(value);
+      } else if ("secure_mode".equals(key)) {
+        secureMode = Boolean.parseBoolean(value);
       } else if ("server_id".equals(key)) {
         serverId = value;
       } else {
@@ -112,14 +122,30 @@ public final class XdsTestServer {
       }
     }
 
+    if (secureMode && (port == maintenancePort)) {
+      System.err.println(
+          "port and maintenance_port should be different for secure mode: port="
+              + port
+              + ", maintenance_port="
+              + maintenancePort);
+      usage = true;
+    }
+
     if (usage) {
       XdsTestServer s = new XdsTestServer();
       System.err.println(
           "Usage: [ARGS...]"
               + "\n"
-              + "\n  --port=INT          listening port for server."
+              + "\n  --port=INT          listening port for test server."
               + "\n                      Default: "
               + s.port
+              + "\n  --maintenance_port=INT      listening port for other servers."
+              + "\n                      Default: "
+              + s.maintenancePort
+              + "\n  --secure_mode=BOOLEAN Use true to enable XdsCredentials."
+              + " port and maintenance_port should be different for secure mode."
+              + "\n                      Default: "
+              + s.secureMode
               + "\n  --server_id=STRING  server ID for response."
               + "\n                      Default: "
               + s.serverId);
@@ -135,30 +161,58 @@ public final class XdsTestServer {
       throw new RuntimeException(e);
     }
     health = new HealthStatusManager();
-    server =
+    if (secureMode) {
-        NettyServerBuilder.forPort(port)
+      server =
-            .addService(
+          XdsServerBuilder.forPort(
-                ServerInterceptors.intercept(
+                  port, XdsServerCredentials.create(InsecureServerCredentials.create()))
-                    new TestServiceImpl(serverId, host), new TestInfoInterceptor(host)))
+              .addService(
-            .addService(new XdsUpdateHealthServiceImpl(health))
+                  ServerInterceptors.intercept(
-            .addService(health.getHealthService())
+                      new TestServiceImpl(serverId, host), new TestInfoInterceptor(host)))
-            .addService(ProtoReflectionService.newInstance())
+              .build()
-            .build()
+              .start();
-            .start();
+      maintenanceServer =
+          NettyServerBuilder.forPort(maintenancePort)
+              .addService(new XdsUpdateHealthServiceImpl(health))
+              .addService(health.getHealthService())
+              .addService(ProtoReflectionService.newInstance())
+              .build()
+              .start();
+    } else {
+      server =
+          NettyServerBuilder.forPort(port)
+              .addService(
+                  ServerInterceptors.intercept(
+                      new TestServiceImpl(serverId, host), new TestInfoInterceptor(host)))
+              .addService(new XdsUpdateHealthServiceImpl(health))
+              .addService(health.getHealthService())
+              .addService(ProtoReflectionService.newInstance())
+              .build()
+              .start();
+      maintenanceServer = null;
+    }
     health.setStatus("", ServingStatus.SERVING);
   }
 
   private void stop() throws Exception {
     server.shutdownNow();
+    if (maintenanceServer != null) {
+      maintenanceServer.shutdownNow();
+    }
     if (!server.awaitTermination(5, TimeUnit.SECONDS)) {
       System.err.println("Timed out waiting for server shutdown");
     }
+    if (maintenanceServer != null && !maintenanceServer.awaitTermination(5, TimeUnit.SECONDS)) {
+      System.err.println("Timed out waiting for maintenanceServer shutdown");
+    }
   }
 
   private void blockUntilShutdown() throws InterruptedException {
     if (server != null) {
       server.awaitTermination();
     }
+    if (maintenanceServer != null) {
+      maintenanceServer.awaitTermination();
+    }
   }
 
   private static class TestServiceImpl extends TestServiceGrpc.TestServiceImplBase {