xds: listener type validation

xds/src/main/java/io/grpc/xds/XdsNameResolver.java

@@ -660,6 +660,10 @@
       }
       logger.log(XdsLogLevel.INFO, "Receive LDS resource update: {0}", update);
       HttpConnectionManager httpConnectionManager = update.httpConnectionManager();
+      if (httpConnectionManager == null) {
+        onResourceDoesNotExist("API Listener: httpConnectionManager");
+        return;
+      }
       List<VirtualHost> virtualHosts = httpConnectionManager.virtualHosts();
       String rdsName = httpConnectionManager.rdsName();
       ImmutableList<NamedFilterConfig> filterConfigs = httpConnectionManager.httpFilterConfigs();

xds/src/main/java/io/grpc/xds/XdsServerWrapper.java

@@ -24,6 +24,8 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.net.HostAndPort;
+import com.google.common.net.InetAddresses;
 import com.google.common.util.concurrent.SettableFuture;
 import io.grpc.Attributes;
 import io.grpc.InternalServerInterceptors;
@@ -57,6 +59,7 @@
 import io.grpc.xds.client.XdsClient.ResourceWatcher;
 import io.grpc.xds.internal.security.SslContextProviderSupplier;
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.SocketAddress;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -383,7 +386,20 @@
         return;
       }
       logger.log(Level.FINEST, "Received Lds update {0}", update);
-      checkNotNull(update.listener(), "update");
+      if (update.listener() == null) {
+        onResourceDoesNotExist("Non-API");
+        return;
+      }
+
+      String ldsAddress = update.listener().address();
+      if (ldsAddress != null && !ipAddressesMatch(ldsAddress)) {
+        handleConfigNotFoundOrMismatch(
+            Status.UNKNOWN.withDescription(
+                String.format(
+                    "Listener address mismatch: expected %s, but got %s.",
+                    listenerAddress, ldsAddress)).asException());
+        return;
+      }
       if (!pendingRds.isEmpty()) {
         // filter chain state has not yet been applied to filterChainSelectorManager and there
         // are two sets of sslContextProviderSuppliers, so we release the old ones.
@@ -432,6 +448,20 @@
       }
     }
 
+    private boolean ipAddressesMatch(String ldsAddress) {
+      HostAndPort ldsAddressHnP = HostAndPort.fromString(ldsAddress);
+      HostAndPort listenerAddressHnP = HostAndPort.fromString(listenerAddress);
+
+      InetAddress listenerIp = InetAddresses.forString(listenerAddressHnP.getHost());
+      InetAddress ldsIp = InetAddresses.forString(ldsAddressHnP.getHost());
+      if (ldsAddressHnP.hasPort() && listenerAddressHnP.hasPort()
+          && ldsAddressHnP.getPort() != listenerAddressHnP.getPort()) {
+        return false;
+      }
+
+      return listenerIp.equals(ldsIp);
+    }
+
     @Override
     public void onResourceDoesNotExist(final String resourceName) {
       if (stopped) {
@@ -440,7 +470,7 @@
       StatusException statusException = Status.UNAVAILABLE.withDescription(
           String.format("Listener %s unavailable, xDS node ID: %s", resourceName,
               xdsClient.getBootstrapInfo().node().getId())).asException();
-      handleConfigNotFound(statusException);
+      handleConfigNotFoundOrMismatch(statusException);
     }
 
     @Override
@@ -673,7 +703,7 @@
       };
     }
 
-    private void handleConfigNotFound(StatusException exception) {
+    private void handleConfigNotFoundOrMismatch(StatusException exception) {
       cleanUpRouteDiscoveryStates();
       shutdownActiveFilters();
       List<SslContextProviderSupplier> toRelease = getSuppliersInUse();

xds/src/test/java/io/grpc/xds/XdsClientWrapperForServerSdsTestMisc.java

@@ -165,7 +165,7 @@ public void run() {
     EnvoyServerProtoData.Listener tcpListener =
         EnvoyServerProtoData.Listener.create(
             "listener1",
-            "10.1.2.3",
+            "0.0.0.0:7000",
             ImmutableList.of(),
             null);
     LdsUpdate listenerUpdate = LdsUpdate.forTcpListener(tcpListener);

xds/src/test/java/io/grpc/xds/XdsSecurityClientServerTest.java

@@ -592,7 +592,7 @@ private void buildServer(
     tlsContextManagerForServer = new TlsContextManagerImpl(bootstrapInfoForServer);
     XdsServerWrapper xdsServer = (XdsServerWrapper) builder.build();
     SettableFuture<Throwable> startFuture = startServerAsync(xdsServer);
-    EnvoyServerProtoData.Listener listener = buildListener("listener1", "10.1.2.3",
+    EnvoyServerProtoData.Listener listener = buildListener("listener1", "0.0.0.0:0",
             downstreamTlsContext, tlsContextManagerForServer);
     LdsUpdate listenerUpdate = LdsUpdate.forTcpListener(listener);
     xdsClient.deliverLdsUpdate(listenerUpdate);

xds/src/test/java/io/grpc/xds/XdsServerBuilderTest.java

@@ -17,6 +17,7 @@
 package io.grpc.xds;
 
 import static com.google.common.truth.Truth.assertThat;
+import static io.grpc.xds.XdsServerTestHelper.buildTestListener;
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.mock;
@@ -26,13 +27,15 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.SettableFuture;
 import io.grpc.BindableService;
 import io.grpc.InsecureServerCredentials;
 import io.grpc.ServerServiceDefinition;
 import io.grpc.Status;
 import io.grpc.StatusException;
 import io.grpc.testing.GrpcCleanupRule;
+import io.grpc.xds.XdsListenerResource.LdsUpdate;
 import io.grpc.xds.XdsServerTestHelper.FakeXdsClient;
 import io.grpc.xds.XdsServerTestHelper.FakeXdsClientPoolFactory;
 import io.grpc.xds.internal.security.CommonTlsContextTestsUtil;
@@ -221,10 +224,13 @@ public void xdsServer_startError()
     buildServer(mockXdsServingStatusListener);
     Future<Throwable> future = startServerAsync();
     // create port conflict for start to fail
-    XdsServerTestHelper.generateListenerUpdate(
-        xdsClient,
+    EnvoyServerProtoData.Listener listener = buildTestListener(
+        "listener1", "0.0.0.0:" + port, ImmutableList.of(),
         CommonTlsContextTestsUtil.buildTestInternalDownstreamTlsContext("CERT1", "VA1"),
-            tlsContextManager);
+        null, tlsContextManager);
+    LdsUpdate listenerUpdate = LdsUpdate.forTcpListener(listener);
+    xdsClient.deliverLdsUpdate(listenerUpdate);
+
     Throwable exception = future.get(5, TimeUnit.SECONDS);
     assertThat(exception).isInstanceOf(IOException.class);
     assertThat(exception).hasMessageThat().contains("Failed to bind");

xds/src/test/java/io/grpc/xds/XdsServerTestHelper.java

@@ -74,7 +74,7 @@ public class XdsServerTestHelper {
   static void generateListenerUpdate(FakeXdsClient xdsClient,
                                      EnvoyServerProtoData.DownstreamTlsContext tlsContext,
                                      TlsContextManager tlsContextManager) {
-    EnvoyServerProtoData.Listener listener = buildTestListener("listener1", "10.1.2.3",
+    EnvoyServerProtoData.Listener listener = buildTestListener("listener1", "0.0.0.0:0",
         ImmutableList.of(), tlsContext, null, tlsContextManager);
     LdsUpdate listenerUpdate = LdsUpdate.forTcpListener(listener);
     xdsClient.deliverLdsUpdate(listenerUpdate);
@@ -85,7 +85,8 @@ static void generateListenerUpdate(
       EnvoyServerProtoData.DownstreamTlsContext tlsContext,
       EnvoyServerProtoData.DownstreamTlsContext tlsContextForDefaultFilterChain,
       TlsContextManager tlsContextManager) {
-    EnvoyServerProtoData.Listener listener = buildTestListener("listener1", "10.1.2.3", sourcePorts,
+    EnvoyServerProtoData.Listener listener = buildTestListener(
+        "listener1", "0.0.0.0:7000", sourcePorts,
         tlsContext, tlsContextForDefaultFilterChain, tlsContextManager);
     LdsUpdate listenerUpdate = LdsUpdate.forTcpListener(listener);
     xdsClient.deliverLdsUpdate(listenerUpdate);

xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java

@@ -54,13 +54,15 @@
 import io.grpc.xds.EnvoyServerProtoData.CidrRange;
 import io.grpc.xds.EnvoyServerProtoData.FilterChain;
 import io.grpc.xds.EnvoyServerProtoData.FilterChainMatch;
+import io.grpc.xds.EnvoyServerProtoData.Listener;
 import io.grpc.xds.Filter.FilterConfig;
 import io.grpc.xds.Filter.NamedFilterConfig;
 import io.grpc.xds.FilterChainMatchingProtocolNegotiators.FilterChainMatchingHandler.FilterChainSelector;
 import io.grpc.xds.StatefulFilter.Config;
 import io.grpc.xds.VirtualHost.Route;
 import io.grpc.xds.VirtualHost.Route.RouteMatch;
 import io.grpc.xds.VirtualHost.Route.RouteMatch.PathMatcher;
+import io.grpc.xds.XdsListenerResource.LdsUpdate;
 import io.grpc.xds.XdsRouteConfigureResource.RdsUpdate;
 import io.grpc.xds.XdsServerBuilder.XdsServingStatusListener;
 import io.grpc.xds.XdsServerTestHelper.FakeXdsClient;
@@ -538,6 +540,46 @@ public void run() {
     verify(mockServer).start();
   }
 
+  @Test
+  public void onChanged_listenerAddressMismatch()
+      throws ExecutionException, InterruptedException, TimeoutException {
+
+    when(mockBuilder.build()).thenReturn(mockServer);
+    xdsServerWrapper = new XdsServerWrapper("10.1.2.3:1", mockBuilder, listener,
+        selectorManager, new FakeXdsClientPoolFactory(xdsClient),
+        filterRegistry, executor.getScheduledExecutorService());
+
+    final SettableFuture<Server> start = SettableFuture.create();
+    Executors.newSingleThreadExecutor().execute(new Runnable() {
+      @Override
+      public void run() {
+        try {
+          start.set(xdsServerWrapper.start());
+        } catch (Exception ex) {
+          start.setException(ex);
+        }
+      }
+    });
+    String ldsResource = xdsClient.ldsResource.get(5, TimeUnit.SECONDS);
+    assertThat(ldsResource).isEqualTo("grpc/server?udpa.resource.listening_address=10.1.2.3:1");
+
+    VirtualHost virtualHost =
+        VirtualHost.create(
+            "virtual-host", Collections.singletonList("auth"), new ArrayList<Route>(),
+            ImmutableMap.<String, FilterConfig>of());
+    HttpConnectionManager httpConnectionManager = HttpConnectionManager.forVirtualHosts(
+        0L, Collections.singletonList(virtualHost), new ArrayList<NamedFilterConfig>());
+    EnvoyServerProtoData.FilterChain filterChain = EnvoyServerProtoData.FilterChain.create(
+        "filter-chain-foo", createMatch(), httpConnectionManager, createTls(),
+        mock(TlsContextManager.class));
+
+    LdsUpdate listenerUpdate = LdsUpdate.forTcpListener(
+        Listener.create("listener", "20.3.4.5:1",
+            ImmutableList.copyOf(Collections.singletonList(filterChain)), null));
+    xdsClient.deliverLdsUpdate(listenerUpdate);
+    verify(listener, timeout(10000)).onNotServing(any());
+  }
+
   @Test
   public void discoverState_rds() throws Exception {
     final SettableFuture<Server> start = SettableFuture.create();