grpc-js-xds: Implement and enable security interop tests

packages/grpc-js-xds/interop/test-client.Dockerfile

@@ -42,7 +42,7 @@ COPY --from=build /node/src/grpc-node/packages/grpc-js ./packages/grpc-js/
 COPY --from=build /node/src/grpc-node/packages/grpc-js-xds ./packages/grpc-js-xds/
 
 ENV GRPC_VERBOSITY="DEBUG"
-ENV GRPC_TRACE=xds_client,xds_resolver,xds_cluster_manager,cds_balancer,xds_cluster_resolver,xds_cluster_impl,priority,weighted_target,round_robin,resolving_load_balancer,subchannel,keepalive,dns_resolver,fault_injection,http_filter,csds,outlier_detection,server,server_call,ring_hash
+ENV GRPC_TRACE=xds_client,xds_resolver,xds_cluster_manager,cds_balancer,xds_cluster_resolver,xds_cluster_impl,priority,weighted_target,round_robin,resolving_load_balancer,subchannel,keepalive,dns_resolver,fault_injection,http_filter,csds,outlier_detection,server,server_call,ring_hash,transport,certificate_provider,xds_channel_credentials
 ENV NODE_XDS_INTEROP_VERBOSITY=1
 
 ENTRYPOINT [ "/nodejs/bin/node", "/node/src/grpc-node/packages/grpc-js-xds/build/interop/xds-interop-client" ]

packages/grpc-js-xds/interop/test-server.Dockerfile

@@ -46,7 +46,7 @@ COPY --from=build /node/src/grpc-node/packages/grpc-js ./packages/grpc-js/
 COPY --from=build /node/src/grpc-node/packages/grpc-js-xds ./packages/grpc-js-xds/
 
 ENV GRPC_VERBOSITY="DEBUG"
-ENV GRPC_TRACE=xds_client,server,xds_server
+ENV GRPC_TRACE=xds_client,server,xds_server,http_filter,certificate_provider
 
 # tini serves as PID 1 and enables the server to properly respond to signals.
 COPY --from=build /tini /tini

packages/grpc-js-xds/interop/xds-interop-client.ts

@@ -519,7 +519,9 @@ function main() {
      * channels do not share any subchannels. It does not have any
      * inherent function. */
     console.log(`Interop client channel ${i} starting sending ${argv.qps} QPS to ${argv.server}`);
-    sendConstantQps(new loadedProto.grpc.testing.TestService(argv.server, grpc.credentials.createInsecure(), {'unique': i}),
+    const insecureCreds = grpc.credentials.createInsecure();
+    const creds = new grpc_xds.XdsChannelCredentials(insecureCreds);
+    sendConstantQps(new loadedProto.grpc.testing.TestService(argv.server, creds, {'unique': i}),
       argv.qps,
       argv.fail_on_failed_rpcs === 'true',
       callStatsTracker);

packages/grpc-js-xds/interop/xds-interop-server.ts

@@ -30,6 +30,8 @@ import { SimpleRequest__Output } from './generated/grpc/testing/SimpleRequest';
 import { SimpleResponse } from './generated/grpc/testing/SimpleResponse';
 import { ReflectionService } from '@grpc/reflection';
 
+grpc_xds.register();
+
 const packageDefinition = protoLoader.loadSync('grpc/testing/test.proto', {
   keepCase: true,
   defaults: true,
@@ -158,6 +160,10 @@ function adminServiceInterceptor(methodDescriptor: grpc.ServerMethodDefinition<a
   const responder: grpc.Responder = {
     start: next => {
       next(listener);
+    },
+    sendMessage: (message, next) => {
+      console.log(`Responded to request to method ${methodDescriptor.path}: ${JSON.stringify(message)}`);
+      next(message);
     }
   };
   return new grpc.ServerInterceptingCall(call, responder);
@@ -228,11 +234,10 @@ function getIPv6Addresses(): string[] {
 
 async function main() {
   const argv = yargs
-    .string(['port', 'maintenance_port', 'address_type'])
-    .boolean(['secure_mode'])
+    .string(['port', 'maintenance_port', 'address_type', 'secure_mode'])
     .demandOption(['port'])
     .default('address_type', 'IPV4_IPV6')
-    .default('secure_mode', false)
+    .default('secure_mode', 'false')
     .parse()
     console.log('Starting xDS interop server. Args: ', argv);
   const healthImpl = new HealthImplementation({'': 'NOT_SERVING'});
@@ -250,7 +255,8 @@ async function main() {
     services: ['grpc.testing.TestService']
   })
   const addressType = argv.address_type.toUpperCase();
-  if (argv.secure_mode) {
+  const secureMode = argv.secure_mode.toLowerCase() == 'true';
+  if (secureMode) {
     if (addressType !== 'IPV4_IPV6') {
       throw new Error('Secure mode only supports IPV4_IPV6 address type');
     }
@@ -265,7 +271,7 @@ async function main() {
     const xdsCreds = new grpc_xds.XdsServerCredentials(grpc.ServerCredentials.createInsecure());
     await Promise.all([
       serverBindPromise(maintenanceServer, `[::]:${argv.maintenance_port}`, grpc.ServerCredentials.createInsecure()),
-      serverBindPromise(server, `[::]:${argv.port}`, xdsCreds)
+      serverBindPromise(server, `0.0.0.0:${argv.port}`, xdsCreds)
     ]);
   } else {
     const server = new grpc.Server({interceptors: [unifiedInterceptor]});

packages/grpc-js-xds/src/index.ts

@@ -33,10 +33,16 @@ import * as pick_first_lb from './lb-policy-registry/pick-first';
 export { XdsServer } from './server';
 export { XdsChannelCredentials, XdsServerCredentials } from './xds-credentials';
 
+let registered = false;
+
 /**
  * Register the "xds:" name scheme with the @grpc/grpc-js library.
  */
 export function register() {
+  if (registered) {
+    return;
+  }
+  registered = true;
   resolver_xds.setup();
   load_balancer_cds.setup();
   xds_cluster_impl.setup();

packages/grpc-js-xds/src/load-balancer-cds.ts

@@ -84,11 +84,13 @@ class DnsExactValueMatcher implements ValueMatcher {
     }
   }
   apply(entry: string): boolean {
-    let [type, value] = entry.split(':');
-    if (!isSupportedSanType(type)) {
+    const colonIndex = entry.indexOf(':');
+    if (colonIndex < 0) {
       return false;
     }
-    if (!value) {
+    const type = entry.substring(0, colonIndex);
+    let value = entry.substring(colonIndex + 1);
+    if (!isSupportedSanType(type)) {
       return false;
     }
     if (this.ignoreCase) {
@@ -137,14 +139,16 @@ class SanEntryMatcher implements ValueMatcher {
     }
   }
   apply(entry: string): boolean {
-    let [type, value] = entry.split(':');
-    if (!isSupportedSanType(type)) {
+    const colonIndex = entry.indexOf(':');
+    if (colonIndex < 0) {
       return false;
     }
-    value = canonicalizeSanEntryValue(type, value);
-    if (!entry) {
+    const type = entry.substring(0, colonIndex);
+    let value = entry.substring(colonIndex + 1);
+    if (!isSupportedSanType(type)) {
       return false;
     }
+    value = canonicalizeSanEntryValue(type, value);
     return this.childMatcher.apply(value);
   }
   toString(): string {
@@ -433,6 +437,7 @@ export class CdsLoadBalancer implements LoadBalancer {
         if (this.latestSanMatcher === null || !this.latestSanMatcher.equals(sanMatcher)) {
           this.latestSanMatcher = sanMatcher;
         }
+        trace('Configured subject alternative name matcher: ' + sanMatcher);
         childOptions[SAN_MATCHER_KEY] = this.latestSanMatcher;
       }
       this.childBalancer.updateAddressList(childEndpointList, typedChildConfig, childOptions);

packages/grpc-js-xds/src/server.ts

@@ -83,6 +83,7 @@ interface ConfigParameters {
   createConnectionInjector: (credentials: ServerCredentials) => ConnectionInjector;
   drainGraceTimeMs: number;
   listenerResourceNameTemplate: string;
+  unregisterChannelzRef: () => void;
 }
 
 class FilterChainEntry {
@@ -159,22 +160,25 @@ class FilterChainEntry {
     }
     if (credentials instanceof XdsServerCredentials) {
       if (filterChain.transport_socket) {
+        trace('Using secure credentials');
         const downstreamTlsContext = decodeSingleResource(DOWNSTREAM_TLS_CONTEXT_TYPE_URL, filterChain.transport_socket.typed_config!.value);
         const commonTlsContext = downstreamTlsContext.common_tls_context!;
         const instanceCertificateProvider = configParameters.xdsClient.getCertificateProvider(commonTlsContext.tls_certificate_provider_instance!.instance_name);
         if (!instanceCertificateProvider) {
           throw new Error(`Invalid TLS context detected: unrecognized certificate instance name: ${commonTlsContext.tls_certificate_provider_instance!.instance_name}`);
         }
-        let validationContext: CertificateValidationContext__Output | null;
-        switch (commonTlsContext?.validation_context_type) {
-          case 'validation_context':
-            validationContext = commonTlsContext.validation_context!;
-            break;
-          case 'combined_validation_context':
-            validationContext = commonTlsContext.combined_validation_context!.default_validation_context;
-            break;
-          default:
-            throw new Error(`Invalid TLS context detected: invalid validation_context_type: ${commonTlsContext.validation_context_type}`);
+        let validationContext: CertificateValidationContext__Output | null = null;
+        if (commonTlsContext?.validation_context_type) {
+          switch (commonTlsContext?.validation_context_type) {
+            case 'validation_context':
+              validationContext = commonTlsContext.validation_context!;
+              break;
+            case 'combined_validation_context':
+              validationContext = commonTlsContext.combined_validation_context!.default_validation_context;
+              break;
+            default:
+              throw new Error(`Invalid TLS context detected: invalid validation_context_type: ${commonTlsContext.validation_context_type}`);
+          }
         }
         let caCertificateProvider: experimental.CertificateProvider | null = null;
         if (validationContext?.ca_certificate_provider_instance) {
@@ -185,6 +189,7 @@ class FilterChainEntry {
         }
         credentials = experimental.createCertificateProviderServerCredentials(instanceCertificateProvider, caCertificateProvider, downstreamTlsContext.require_client_certificate?.value ?? false);
       } else {
+        trace('Using fallback credentials');
         credentials = credentials.getFallbackCredentials();
       }
     }
@@ -287,6 +292,7 @@ class ListenerConfig {
   handleConnection(socket: net.Socket) {
     const matchingFilter = selectMostSpecificallyMatchingFilter(this.filterChainEntries, socket) ?? this.defaultFilterChain;
     if (!matchingFilter) {
+      trace('Rejecting connection from ' + socket.remoteAddress + ': No filter matched');
       socket.destroy();
       return;
     }
@@ -449,12 +455,25 @@ class BoundPortEntry {
     this.tcpServer.close();
     const resourceName = formatTemplateString(this.configParameters.listenerResourceNameTemplate, this.boundAddress);
     ListenerResourceType.cancelWatch(this.configParameters.xdsClient, resourceName, this.listenerWatcher);
+    this.configParameters.unregisterChannelzRef();
   }
 }
 
 function normalizeFilterChainMatch(filterChainMatch: FilterChainMatch__Output | null): NormalizedFilterChainMatch[] {
   if (!filterChainMatch) {
-    return [];
+    filterChainMatch = {
+      address_suffix: '',
+      application_protocols: [],
+      destination_port: null,
+      direct_source_prefix_ranges: [],
+      prefix_ranges: [],
+      server_names: [],
+      source_ports: [],
+      source_prefix_ranges: [],
+      source_type: 'ANY',
+      suffix_len: null,
+      transport_protocol: 'raw_buffer'
+    };
   }
   if (filterChainMatch.destination_port) {
     return [];
@@ -613,11 +632,13 @@ export class XdsServer extends Server {
     if (!hostPort || !isValidIpPort(hostPort)) {
       throw new Error(`Listening port string must have the format IP:port with non-zero port, got ${port}`);
     }
+    const channelzRef = this.experimentalRegisterListenerToChannelz({host: hostPort.host, port: hostPort.port!});
     const configParameters: ConfigParameters = {
-      createConnectionInjector: (credentials) => this.createConnectionInjector(credentials),
+      createConnectionInjector: (credentials) => this.experimentalCreateConnectionInjectorWithChannelzRef(credentials, channelzRef),
       drainGraceTimeMs: this.drainGraceTimeMs,
       listenerResourceNameTemplate: this.listenerResourceNameTemplate,
-      xdsClient: this.xdsClient
+      xdsClient: this.xdsClient,
+      unregisterChannelzRef: () => this.experimentalUnregisterListenerFromChannelz(channelzRef)
     };
     const portEntry = new BoundPortEntry(configParameters, port, creds);
     const servingStatusListener: ServingStatusListener = statusObject => {

packages/grpc-js-xds/src/xds-credentials.ts

@@ -15,12 +15,18 @@
  *
  */
 
-import { CallCredentials, ChannelCredentials, ChannelOptions, ServerCredentials, VerifyOptions, experimental } from "@grpc/grpc-js";
+import { CallCredentials, ChannelCredentials, ChannelOptions, ServerCredentials, VerifyOptions, experimental, logVerbosity } from "@grpc/grpc-js";
 import { CA_CERT_PROVIDER_KEY, IDENTITY_CERT_PROVIDER_KEY, SAN_MATCHER_KEY, SanMatcher } from "./load-balancer-cds";
 import GrpcUri = experimental.GrpcUri;
 import SecureConnector = experimental.SecureConnector;
 import createCertificateProviderChannelCredentials = experimental.createCertificateProviderChannelCredentials;
 
+const TRACER_NAME = 'xds_channel_credentials';
+
+function trace(text: string) {
+  experimental.trace(logVerbosity.DEBUG, TRACER_NAME, text);
+}
+
 export class XdsChannelCredentials extends ChannelCredentials {
   constructor(private fallbackCredentials: ChannelCredentials) {
     super();
@@ -33,20 +39,23 @@ export class XdsChannelCredentials extends ChannelCredentials {
   }
   _createSecureConnector(channelTarget: GrpcUri, options: ChannelOptions, callCredentials?: CallCredentials): SecureConnector {
     if (options[CA_CERT_PROVIDER_KEY]) {
+      trace('Using secure credentials');
       const verifyOptions: VerifyOptions = {};
       if (options[SAN_MATCHER_KEY]) {
         const matcher = options[SAN_MATCHER_KEY] as SanMatcher;
         verifyOptions.checkServerIdentity = (hostname, cert) => {
           if (cert.subjectaltname && matcher.apply(cert.subjectaltname)) {
             return undefined;
           } else {
+            trace('Subject alternative name not matched: ' + cert.subjectaltname);
             return new Error('No matching subject alternative name found in certificate');
           }
         }
       }
       const certProviderCreds = createCertificateProviderChannelCredentials(options[CA_CERT_PROVIDER_KEY], options[IDENTITY_CERT_PROVIDER_KEY] ?? null, verifyOptions);
       return certProviderCreds._createSecureConnector(channelTarget, options, callCredentials);
     } else {
+      trace('Using fallback credentials');
       return this.fallbackCredentials._createSecureConnector(channelTarget, options, callCredentials);
     }
   }
@@ -55,7 +64,7 @@ export class XdsChannelCredentials extends ChannelCredentials {
 
 export class XdsServerCredentials extends ServerCredentials {
   constructor(private fallbackCredentials: ServerCredentials) {
-    super();
+    super({});
   }
 
   getFallbackCredentials() {

packages/grpc-js-xds/src/xds-dependency-manager.ts

@@ -360,6 +360,11 @@ export class XdsDependencyManager {
   constructor(private xdsClient: XdsClient, private listenerResourceName: string, private dataPlaneAuthority: string, private watcher: XdsConfigWatcher) {
     this.ldsWatcher = new Watcher<Listener__Output>({
       onResourceChanged: (update: Listener__Output) => {
+        if (!update.api_listener) {
+          this.trace('Received Listener resource not usable on client');
+          this.handleListenerDoesNotExist();
+          return;
+        }
         this.latestListener = update;
         const httpConnectionManager = decodeSingleResource(HTTP_CONNECTION_MANGER_TYPE_URL, update.api_listener!.api_listener!.value);
         switch (httpConnectionManager.route_specifier) {
@@ -401,15 +406,7 @@ export class XdsDependencyManager {
       },
       onResourceDoesNotExist: () => {
         this.trace('Resolution error: LDS resource does not exist');
-        if (this.latestRouteConfigName) {
-          this.trace('RDS.cancelWatch(' + this.latestRouteConfigName + '): LDS resource does not exist');
-          RouteConfigurationResourceType.cancelWatch(this.xdsClient, this.latestRouteConfigName, this.rdsWatcher);
-          this.latestRouteConfigName = null;
-          this.latestRouteConfiguration = null;
-          this.clusterRoots = [];
-          this.pruneOrphanClusters();
-        }
-        this.watcher.onResourceDoesNotExist(`Listener ${listenerResourceName}`);
+        this.handleListenerDoesNotExist();
       }
     });
     this.rdsWatcher = new Watcher<RouteConfiguration__Output>({
@@ -435,6 +432,19 @@ export class XdsDependencyManager {
     trace('[' + this.listenerResourceName + '] ' + text);
   }
 
+  private handleListenerDoesNotExist() {
+    if (this.latestRouteConfigName) {
+      this.trace('RDS.cancelWatch(' + this.latestRouteConfigName + '): LDS resource does not exist');
+      RouteConfigurationResourceType.cancelWatch(this.xdsClient, this.latestRouteConfigName, this.rdsWatcher);
+      this.latestRouteConfigName = null;
+      this.latestRouteConfiguration = null;
+      this.clusterRoots = [];
+      this.pruneOrphanClusters();
+    }
+    this.watcher.onResourceDoesNotExist(`Listener ${this.listenerResourceName}`);
+
+  }
+
   private maybeSendUpdate() {
     if (!this.latestListener) {
       this.trace('Not sending update: no Listener update received');