xDS: fix crash on wrong listener type (both client and server side) (#31684)

* xds resolver: fix crash on wrong listener type

* fix same bug on server side

* fix clang-tidy and add requested TODO

src/core/ext/filters/client_channel/resolver/xds/xds_resolver.cc

@@ -877,8 +877,13 @@ void XdsResolver::OnListenerUpdate(XdsListenerResource listener) {
     gpr_log(GPR_INFO, "[xds_resolver %p] received updated listener data", this);
   }
   if (xds_client_ == nullptr) return;
-  current_listener_ = std::move(
-      absl::get<XdsListenerResource::HttpConnectionManager>(listener.listener));
+  auto* hcm = absl::get_if<XdsListenerResource::HttpConnectionManager>(
+      &listener.listener);
+  if (hcm == nullptr) {
+    return OnError(lds_resource_name_,
+                   absl::UnavailableError("not an API listener"));
+  }
+  current_listener_ = std::move(*hcm);
   MatchMutable(
       &current_listener_.route_config,
       // RDS resource name

src/core/ext/xds/xds_server_config_fetcher.cc

@@ -559,9 +559,15 @@ void XdsServerConfigFetcher::ListenerWatcher::OnResourceChanged(
             "[ListenerWatcher %p] Received LDS update from xds client %p: %s",
             this, xds_client_.get(), listener.ToString().c_str());
   }
-  auto& tcp_listener =
-      absl::get<XdsListenerResource::TcpListener>(listener.listener);
-  if (tcp_listener.address != listening_address_) {
+  auto* tcp_listener =
+      absl::get_if<XdsListenerResource::TcpListener>(&listener.listener);
+  if (tcp_listener == nullptr) {
+    MutexLock lock(&mu_);
+    OnFatalError(
+        absl::FailedPreconditionError("LDS resource is not a TCP listener"));
+    return;
+  }
+  if (tcp_listener->address != listening_address_) {
     MutexLock lock(&mu_);
     OnFatalError(absl::FailedPreconditionError(
         "Address in LDS update does not match listening address"));
@@ -569,8 +575,8 @@ void XdsServerConfigFetcher::ListenerWatcher::OnResourceChanged(
   }
   auto new_filter_chain_match_manager = MakeRefCounted<FilterChainMatchManager>(
       xds_client_->Ref(DEBUG_LOCATION, "FilterChainMatchManager"),
-      std::move(tcp_listener.filter_chain_map),
-      std::move(tcp_listener.default_filter_chain));
+      std::move(tcp_listener->filter_chain_map),
+      std::move(tcp_listener->default_filter_chain));
   MutexLock lock(&mu_);
   if (filter_chain_match_manager_ == nullptr ||
       !(new_filter_chain_match_manager->filter_chain_map() ==

test/cpp/end2end/xds/xds_end2end_test.cc

@@ -784,6 +784,23 @@ TEST_P(XdsEnabledServerTest, BadLdsUpdateNoApiListenerNorAddress) {
                   "ApiListener]")));
 }
 
+// Verify that a non-TCP listener results in "not serving" status.
+TEST_P(XdsEnabledServerTest, NonTcpListener) {
+  DoSetUp();
+  Listener listener = default_listener_;  // Client-side listener.
+  listener = PopulateServerListenerNameAndPort(listener, backends_[0]->port());
+  auto hcm = ClientHcmAccessor().Unpack(listener);
+  auto* rds = hcm.mutable_rds();
+  rds->set_route_config_name(kDefaultRouteConfigurationName);
+  rds->mutable_config_source()->mutable_self();
+  ClientHcmAccessor().Pack(hcm, &listener);
+  balancer_->ads_service()->SetLdsResource(listener);
+  backends_[0]->Start();
+  backends_[0]->notifier()->WaitOnServingStatusChange(
+      absl::StrCat(ipv6_only_ ? "[::1]:" : "127.0.0.1:", backends_[0]->port()),
+      grpc::StatusCode::FAILED_PRECONDITION);
+}
+
 // Verify that a mismatch of listening address results in "not serving"
 // status.
 TEST_P(XdsEnabledServerTest, ListenerAddressMismatch) {

test/cpp/end2end/xds/xds_routing_end2end_test.cc

@@ -50,6 +50,35 @@ TEST_P(LdsTest, NacksInvalidListener) {
       ::testing::HasSubstr("Listener has neither address nor ApiListener"));
 }
 
+// Tests that we go into TRANSIENT_FAILURE if the Listener is not an API
+// listener.
+TEST_P(LdsTest, NotAnApiListener) {
+  Listener listener = default_server_listener_;
+  listener.set_name(kServerName);
+  auto hcm = ServerHcmAccessor().Unpack(listener);
+  auto* rds = hcm.mutable_rds();
+  rds->set_route_config_name(kDefaultRouteConfigurationName);
+  rds->mutable_config_source()->mutable_self();
+  ServerHcmAccessor().Pack(hcm, &listener);
+  balancer_->ads_service()->SetLdsResource(listener);
+  // RPCs should fail.
+  CheckRpcSendFailure(
+      DEBUG_LOCATION, StatusCode::UNAVAILABLE,
+      absl::StrCat(kServerName, ": UNAVAILABLE: not an API listener"));
+  // We should have ACKed the LDS resource.
+  const auto deadline =
+      absl::Now() + (absl::Seconds(30) * grpc_test_slowdown_factor());
+  while (true) {
+    ASSERT_LT(absl::Now(), deadline) << "timed out waiting for LDS ACK";
+    auto response_state = balancer_->ads_service()->lds_response_state();
+    if (response_state.has_value()) {
+      EXPECT_EQ(response_state->state, AdsServiceImpl::ResponseState::ACKED);
+      break;
+    }
+    absl::SleepFor(absl::Seconds(1) * grpc_test_slowdown_factor());
+  }
+}
+
 class LdsDeletionTest : public XdsEnd2endTest {
  protected:
   void SetUp() override {}  // Individual tests call InitClient().

test/cpp/end2end/xds/xds_server.h

@@ -132,6 +132,8 @@ class AdsServiceImpl
   }
 
   // Get the list of response state for each resource type.
+  // TODO(roth): Consider adding an absl::Notification-based mechanism
+  // here to avoid the need for tests to poll the response state.
   absl::optional<ResponseState> GetResponseState(const std::string& type_url) {
     grpc_core::MutexLock lock(&ads_mu_);
     if (resource_type_response_state_[type_url].empty()) {