xds: listener type validation

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

@@ -22,6 +22,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.util.Durations;
+import io.envoyproxy.envoy.config.core.v3.SocketAddress.Protocol;
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.CommonTlsContext;
 import io.grpc.Internal;
 import io.grpc.xds.client.EnvoyProtoData;
@@ -248,13 +249,17 @@ abstract static class Listener {
     @Nullable
     abstract FilterChain defaultFilterChain();
 
+    @Nullable
+    abstract Protocol protocol();
+
     static Listener create(
         String name,
         @Nullable String address,
         ImmutableList<FilterChain> filterChains,
-        @Nullable FilterChain defaultFilterChain) {
+        @Nullable FilterChain defaultFilterChain,
+        @Nullable Protocol protocol) {
       return new AutoValue_EnvoyServerProtoData_Listener(name, address, filterChains,
-          defaultFilterChain);
+          defaultFilterChain, protocol);
     }
   }
 

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

@@ -162,13 +162,16 @@ static EnvoyServerProtoData.Listener parseServerSideListener(
     }
 
     String address = null;
+    SocketAddress socketAddress = null;
     if (proto.getAddress().hasSocketAddress()) {
-      SocketAddress socketAddress = proto.getAddress().getSocketAddress();
+      socketAddress = proto.getAddress().getSocketAddress();
       address = socketAddress.getAddress();
+      if (address.isEmpty()) {
+        throw new ResourceInvalidException("Invalid address: Empty address is not allowed.");
+      }
       switch (socketAddress.getPortSpecifierCase()) {
         case NAMED_PORT:
-          address = address + ":" + socketAddress.getNamedPort();
-          break;
+          throw new ResourceInvalidException("NAMED_PORT is not supported in gRPC.");
         case PORT_VALUE:
           address = address + ":" + socketAddress.getPortValue();
           break;
@@ -209,8 +212,8 @@ static EnvoyServerProtoData.Listener parseServerSideListener(
           null, certProviderInstances, args);
     }
 
-    return EnvoyServerProtoData.Listener.create(
-        proto.getName(), address, filterChains.build(), defaultFilterChain);
+    return EnvoyServerProtoData.Listener.create(proto.getName(), address, filterChains.build(),
+        defaultFilterChain, socketAddress == null ? null : socketAddress.getProtocol());
   }
 
   @VisibleForTesting

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

@@ -676,6 +676,13 @@ public void onUpdate(StatusOr<XdsConfig> updateOrStatus) {
       // Process Route
       XdsConfig update = updateOrStatus.getValue();
       HttpConnectionManager httpConnectionManager = update.getListener().httpConnectionManager();
+      if (httpConnectionManager == null) {
+        logger.log(XdsLogLevel.INFO, "API Listener: httpConnectionManager does not exist.");
+        updateActiveFilters(null);
+        cleanUpRoutes(updateOrStatus.getStatus());
+        return;
+      }
+
       VirtualHost virtualHost = update.getVirtualHost();
       ImmutableList<NamedFilterConfig> filterConfigs = httpConnectionManager.httpFilterConfigs();
       long streamDurationNano = httpConnectionManager.httpMaxStreamDurationNano();

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

@@ -24,7 +24,10 @@
 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.envoyproxy.envoy.config.core.v3.SocketAddress.Protocol;
 import io.grpc.Attributes;
 import io.grpc.InternalServerInterceptors;
 import io.grpc.Metadata;
@@ -57,6 +60,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 +387,21 @@ public void onChanged(final LdsUpdate update) {
         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 || update.listener().protocol() != Protocol.TCP
+          || !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 +450,18 @@ public void onChanged(final LdsUpdate update) {
       }
     }
 
+    private boolean ipAddressesMatch(String ldsAddress) {
+      HostAndPort ldsAddressHnP = HostAndPort.fromString(ldsAddress);
+      HostAndPort listenerAddressHnP = HostAndPort.fromString(listenerAddress);
+      if (!ldsAddressHnP.hasPort() || !listenerAddressHnP.hasPort()
+          || ldsAddressHnP.getPort() != listenerAddressHnP.getPort()) {
+        return false;
+      }
+      InetAddress listenerIp = InetAddresses.forString(listenerAddressHnP.getHost());
+      InetAddress ldsIp = InetAddresses.forString(ldsAddressHnP.getHost());
+      return listenerIp.equals(ldsIp);
+    }
+
     @Override
     public void onResourceDoesNotExist(final String resourceName) {
       if (stopped) {
@@ -440,7 +470,7 @@ public void onResourceDoesNotExist(final String resourceName) {
       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 @@ public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
       };
     }
 
-    private void handleConfigNotFound(StatusException exception) {
+    private void handleConfigNotFoundOrMismatch(StatusException exception) {
       cleanUpRouteDiscoveryStates();
       shutdownActiveFilters();
       List<SslContextProviderSupplier> toRelease = getSuppliersInUse();

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

@@ -366,10 +366,14 @@ static Listener buildServerListener() {
         .setFilterChainMatch(filterChainMatch)
         .addFilters(filter)
         .build();
+    Address address = Address.newBuilder()
+        .setSocketAddress(SocketAddress.newBuilder().setAddress("0.0.0.0").setPortValue(0))
+        .build();
     return Listener.newBuilder()
         .setName(SERVER_LISTENER_TEMPLATE_NO_REPLACEMENT)
         .setTrafficDirection(TrafficDirection.INBOUND)
         .addFilterChains(filterChain)
+        .setAddress(address)
         .build();
   }
 }

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

@@ -2660,6 +2660,41 @@ public void parseServerSideListener_useOriginalDst() throws ResourceInvalidExcep
         listener,null, filterRegistry, null, getXdsResourceTypeArgs(true));
   }
 
+  @Test
+  public void parseServerSideListener_emptyAddress() throws ResourceInvalidException {
+    Listener listener =
+        Listener.newBuilder()
+            .setName("listener1")
+            .setTrafficDirection(TrafficDirection.INBOUND)
+            .setAddress(Address.newBuilder()
+                .setSocketAddress(
+                    SocketAddress.newBuilder()))
+            .build();
+    thrown.expect(ResourceInvalidException.class);
+    thrown.expectMessage("Invalid address: Empty address is not allowed.");
+
+    XdsListenerResource.parseServerSideListener(
+        listener,null, filterRegistry, null, getXdsResourceTypeArgs(true));
+  }
+
+  @Test
+  public void parseServerSideListener_namedPort() throws ResourceInvalidException {
+    Listener listener =
+        Listener.newBuilder()
+            .setName("listener1")
+            .setTrafficDirection(TrafficDirection.INBOUND)
+            .setAddress(Address.newBuilder()
+                .setSocketAddress(
+                    SocketAddress.newBuilder()
+                        .setAddress("172.14.14.5").setNamedPort("")))
+            .build();
+    thrown.expect(ResourceInvalidException.class);
+    thrown.expectMessage("NAMED_PORT is not supported in gRPC.");
+
+    XdsListenerResource.parseServerSideListener(
+        listener,null, filterRegistry, null, getXdsResourceTypeArgs(true));
+  }
+
   @Test
   public void parseServerSideListener_nonUniqueFilterChainMatch() throws ResourceInvalidException {
     Filter filter1 = buildHttpConnectionManagerFilter(

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

@@ -32,6 +32,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.SettableFuture;
+import io.envoyproxy.envoy.config.core.v3.SocketAddress.Protocol;
 import io.grpc.Server;
 import io.grpc.ServerBuilder;
 import io.grpc.Status;
@@ -165,9 +166,10 @@ public void run() {
     EnvoyServerProtoData.Listener tcpListener =
         EnvoyServerProtoData.Listener.create(
             "listener1",
-            "10.1.2.3",
+            "0.0.0.0:7000",
             ImmutableList.of(),
-            null);
+            null,
+            Protocol.TCP);
     LdsUpdate listenerUpdate = LdsUpdate.forTcpListener(tcpListener);
     xdsClient.ldsWatcher.onChanged(listenerUpdate);
     verify(listener, timeout(5000)).onServing();

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

@@ -35,6 +35,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.SettableFuture;
+import io.envoyproxy.envoy.config.core.v3.SocketAddress.Protocol;
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.CertificateValidationContext;
 import io.grpc.Attributes;
 import io.grpc.EquivalentAddressGroup;
@@ -488,7 +489,7 @@ public void mtlsClientServer_changeServerContext_expectException()
     DownstreamTlsContext downstreamTlsContext =
         CommonTlsContextTestsUtil.buildDownstreamTlsContext(
             "cert-instance-name2", true, true);
-    EnvoyServerProtoData.Listener listener = buildListener("listener1", "0.0.0.0",
+    EnvoyServerProtoData.Listener listener = buildListener("listener1", "0.0.0.0:0",
             downstreamTlsContext,
             tlsContextManagerForServer);
     xdsClient.deliverLdsUpdate(LdsUpdate.forTcpListener(listener));
@@ -592,7 +593,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);
@@ -633,7 +634,7 @@ static EnvoyServerProtoData.Listener buildListener(
         "filter-chain-foo", filterChainMatch, httpConnectionManager, tlsContext,
         tlsContextManager);
     EnvoyServerProtoData.Listener listener = EnvoyServerProtoData.Listener.create(
-        name, address, ImmutableList.of(defaultFilterChain), null);
+        name, address, ImmutableList.of(defaultFilterChain), null, Protocol.TCP);
     return listener;
   }
 

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

@@ -21,6 +21,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.SettableFuture;
+import io.envoyproxy.envoy.config.core.v3.SocketAddress.Protocol;
 import io.grpc.InsecureChannelCredentials;
 import io.grpc.MetricRecorder;
 import io.grpc.internal.ObjectPool;
@@ -74,7 +75,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 +86,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);
@@ -130,7 +132,7 @@ static EnvoyServerProtoData.Listener buildTestListener(
         tlsContextForDefaultFilterChain, tlsContextManager);
     EnvoyServerProtoData.Listener listener =
         EnvoyServerProtoData.Listener.create(
-            name, address, ImmutableList.of(filterChain1), defaultFilterChain);
+            name, address, ImmutableList.of(filterChain1), defaultFilterChain, Protocol.TCP);
     return listener;
   }
 
@@ -290,15 +292,23 @@ private String awaitLdsResource(Duration timeout) {
       }
     }
 
+    void deliverLdsUpdateWithApiListener(long httpMaxStreamDurationNano,
+        List<VirtualHost> virtualHosts) {
+      execute(() -> {
+        ldsWatcher.onChanged(LdsUpdate.forApiListener(HttpConnectionManager.forVirtualHosts(
+            httpMaxStreamDurationNano, virtualHosts, null)));
+      });
+    }
+
     void deliverLdsUpdate(LdsUpdate ldsUpdate) {
       execute(() -> ldsWatcher.onChanged(ldsUpdate));
     }
 
     void deliverLdsUpdate(
         List<FilterChain> filterChains,
         @Nullable FilterChain defaultFilterChain) {
-      deliverLdsUpdate(LdsUpdate.forTcpListener(Listener.create(
-          "listener", "0.0.0.0:1", ImmutableList.copyOf(filterChains), defaultFilterChain)));
+      deliverLdsUpdate(LdsUpdate.forTcpListener(Listener.create("listener", "0.0.0.0:1",
+          ImmutableList.copyOf(filterChains), defaultFilterChain, Protocol.TCP)));
     }
 
     void deliverLdsUpdate(FilterChain filterChain, @Nullable FilterChain defaultFilterChain) {