listener: check socket type matches the scheme in url (#20584) (#20605)

comparing address is not sufficient to find the listener since udp and tcp listeners can share the same address. use the scheme in the address url to match the socket type as well to avoid finding a listener with a different socket type. Signed-off-by: Ji-Young Park <jpark@twosigma.com>
envoyproxy · Apr 5, 2022 · adc7c80 · adc7c80
1 parent 90a1d4b
commit adc7c80
Show file tree

Hide file tree

Showing 6 changed files with 123 additions and 8 deletions.
diff --git a/docs/root/version_history/current.rst b/docs/root/version_history/current.rst
@@ -53,6 +53,7 @@ Bug Fixes
 * data plane: fix crash when internal redirect selects a route configured with direct response or redirect actions.
 * data plane: fixing error handling where writing to a socket failed while under the stack of processing. This should genreally affect HTTP/3. This behavioral change can be reverted by setting ``envoy.reloadable_features.allow_upstream_inline_write`` to false.
 * eds: fix the eds cluster update by allowing update on the locality of the cluster endpoints. This behavioral change can be temporarily reverted by setting runtime guard ``envoy.reloadable_features.support_locality_update_on_eds_cluster_endpoints`` to false.
+* hot restart: fixed a bug where an incorrect fd was passed to child when a tcp listener and a udp listener listen to the same address because socket type was not used to find the matching listener for a url.
 * http: fixed a bug where %RESPONSE_CODE_DETAILS% was not set correctly in :ref:`request_headers_to_add <envoy_v3_api_field_config.route.v3.RouteConfiguration.request_headers_to_add>`.
 * http: fixed a bug where ``100-continue`` comparison in the ``Expect`` request header field was case sensitive. This RFC compliant behavior can be temporarily reverted by setting runtime guard ``envoy.reloadable_features.http_100_continue_case_insensitive`` to false.
 * jwt_authn: fixed a bug where a JWT with empty "iss" is passed even the field :ref:`issuer <envoy_v3_api_field_extensions.filters.http.jwt_authn.v3.JwtProvider.issuer>` is specified. If the "issuer" field is specified, "iss" in the JWT should match it.

diff --git a/source/common/network/utility.cc b/source/common/network/utility.cc
@@ -52,6 +52,18 @@ Address::InstanceConstSharedPtr Utility::resolveUrl(const std::string& url) {
   }
 }
 
+StatusOr<Socket::Type> Utility::socketTypeFromUrl(const std::string& url) {
+  if (urlIsTcpScheme(url)) {
+    return Socket::Type::Stream;
+  } else if (urlIsUdpScheme(url)) {
+    return Socket::Type::Datagram;
+  } else if (urlIsUnixScheme(url)) {
+    return Socket::Type::Stream;
+  } else {
+    return absl::InvalidArgumentError(absl::StrCat("unknown protocol scheme: ", url));
+  }
+}
+
 bool Utility::urlIsTcpScheme(absl::string_view url) { return absl::StartsWith(url, TCP_SCHEME); }
 
 bool Utility::urlIsUdpScheme(absl::string_view url) { return absl::StartsWith(url, UDP_SCHEME); }

diff --git a/source/common/network/utility.h b/source/common/network/utility.h
@@ -9,6 +9,8 @@
 #include "envoy/network/connection.h"
 #include "envoy/network/listener.h"
 
+#include "source/common/common/statusor.h"
+
 #include "absl/strings/string_view.h"
 
 namespace Envoy {
@@ -100,6 +102,14 @@ class Utility {
    */
   static Address::InstanceConstSharedPtr resolveUrl(const std::string& url);
 
+  /**
+   * Determine the socket type for a URL.
+   *
+   * @param url supplies the url to resolve.
+   * @return StatusOr<Socket::Type> of the socket type, or an error status if url is invalid.
+   */
+  static StatusOr<Socket::Type> socketTypeFromUrl(const std::string& url);
+
   /**
    * Match a URL to the TCP scheme
    * @param url supplies the URL to match.

diff --git a/source/server/hot_restarting_parent.cc b/source/server/hot_restarting_parent.cc
@@ -106,18 +106,26 @@ HotRestartingParent::Internal::getListenSocketsForChild(const HotRestartMessage:
   wrapped_reply.mutable_reply()->mutable_pass_listen_socket()->set_fd(-1);
   Network::Address::InstanceConstSharedPtr addr =
       Network::Utility::resolveUrl(request.pass_listen_socket().address());
+
   for (const auto& listener : server_->listenerManager().listeners()) {
     Network::ListenSocketFactory& socket_factory = listener.get().listenSocketFactory();
     if (*socket_factory.localAddress() == *addr && listener.get().bindToPort()) {
-      // worker_index() will default to 0 if not set which is the behavior before this field
-      // was added. Thus, this should be safe for both roll forward and roll back.
-      if (request.pass_listen_socket().worker_index() < server_->options().concurrency()) {
-        wrapped_reply.mutable_reply()->mutable_pass_listen_socket()->set_fd(
-            socket_factory.getListenSocket(request.pass_listen_socket().worker_index())
-                ->ioHandle()
-                .fdDoNotUse());
+      StatusOr<Network::Socket::Type> socket_type =
+          Network::Utility::socketTypeFromUrl(request.pass_listen_socket().address());
+      // socketTypeFromUrl should return a valid value since resolveUrl returned a valid address.
+      ASSERT(socket_type.ok());
+
+      if (socket_factory.socketType() == *socket_type) {
+        // worker_index() will default to 0 if not set which is the behavior before this field
+        // was added. Thus, this should be safe for both roll forward and roll back.
+        if (request.pass_listen_socket().worker_index() < server_->options().concurrency()) {
+          wrapped_reply.mutable_reply()->mutable_pass_listen_socket()->set_fd(
+              socket_factory.getListenSocket(request.pass_listen_socket().worker_index())
+                  ->ioHandle()
+                  .fdDoNotUse());
+        }
+        break;
       }
-      break;
     }
   }
   return wrapped_reply;

diff --git a/test/common/network/utility_test.cc b/test/common/network/utility_test.cc
@@ -106,6 +106,23 @@ TEST(NetworkUtility, resolveUrl) {
   EXPECT_EQ("[a:b:c:d::]:0", Utility::resolveUrl("udp://[a:b:c:d::]:0")->asString());
 }
 
+TEST(NetworkUtility, socketTypeFromUrl) {
+  EXPECT_FALSE(Utility::socketTypeFromUrl("foo").ok());
+  EXPECT_FALSE(Utility::socketTypeFromUrl("abc://foo").ok());
+
+  EXPECT_EQ(Network::Socket::Type::Stream, *Utility::socketTypeFromUrl("unix://"));
+  EXPECT_EQ(Network::Socket::Type::Stream, *Utility::socketTypeFromUrl("unix://foo"));
+  EXPECT_EQ(Network::Socket::Type::Stream, *Utility::socketTypeFromUrl("unix://tmp/server"));
+
+  EXPECT_EQ(Network::Socket::Type::Stream, *Utility::socketTypeFromUrl("tcp://1.2.3.4:1234"));
+  EXPECT_EQ(Network::Socket::Type::Stream, *Utility::socketTypeFromUrl("tcp://0.0.0.0:0"));
+  EXPECT_EQ(Network::Socket::Type::Stream, *Utility::socketTypeFromUrl("tcp://[::1]:1"));
+
+  EXPECT_EQ(Network::Socket::Type::Datagram, *Utility::socketTypeFromUrl("udp://1.2.3.4:1234"));
+  EXPECT_EQ(Network::Socket::Type::Datagram, *Utility::socketTypeFromUrl("udp://0.0.0.0:0"));
+  EXPECT_EQ(Network::Socket::Type::Datagram, *Utility::socketTypeFromUrl("udp://[::1]:1"));
+}
+
 TEST(NetworkUtility, ParseInternetAddress) {
   EXPECT_THROW(Utility::parseInternetAddress(""), EnvoyException);
   EXPECT_THROW(Utility::parseInternetAddress("1.2.3"), EnvoyException);

diff --git a/test/server/hot_restarting_parent_test.cc b/test/server/hot_restarting_parent_test.cc
@@ -1,5 +1,6 @@
 #include <memory>
 
+#include "source/common/network/address_impl.h"
 #include "source/server/hot_restarting_child.h"
 #include "source/server/hot_restarting_parent.h"
 
@@ -64,6 +65,72 @@ TEST_F(HotRestartingParentTest, GetListenSocketsForChildNotBindPort) {
   EXPECT_EQ(-1, message.reply().pass_listen_socket().fd());
 }
 
+TEST_F(HotRestartingParentTest, GetListenSocketsForChildSocketType) {
+  MockListenerManager listener_manager;
+  Network::MockListenerConfig tcp_listener_config;
+  Network::MockListenerConfig udp_listener_config;
+  MockOptions options;
+  std::vector<std::reference_wrapper<Network::ListenerConfig>> listeners;
+  InSequence s;
+
+  listeners.push_back(std::ref(*static_cast<Network::ListenerConfig*>(&tcp_listener_config)));
+  listeners.push_back(std::ref(*static_cast<Network::ListenerConfig*>(&udp_listener_config)));
+
+  EXPECT_CALL(server_, listenerManager()).WillOnce(ReturnRef(listener_manager));
+  EXPECT_CALL(listener_manager, listeners(ListenerManager::ListenerState::ACTIVE))
+      .WillOnce(Return(listeners));
+  EXPECT_CALL(tcp_listener_config, listenSocketFactory());
+  EXPECT_CALL(tcp_listener_config.socket_factory_, localAddress());
+  EXPECT_CALL(tcp_listener_config, bindToPort()).WillOnce(Return(true));
+  EXPECT_CALL(tcp_listener_config.socket_factory_, socketType())
+      .WillOnce(Return(Network::Socket::Type::Stream));
+
+  EXPECT_CALL(udp_listener_config, listenSocketFactory());
+  EXPECT_CALL(udp_listener_config.socket_factory_, localAddress());
+  EXPECT_CALL(udp_listener_config, bindToPort()).WillOnce(Return(true));
+  EXPECT_CALL(udp_listener_config.socket_factory_, socketType())
+      .WillOnce(Return(Network::Socket::Type::Datagram));
+
+  EXPECT_CALL(server_, options()).WillOnce(ReturnRef(options));
+  EXPECT_CALL(options, concurrency()).WillOnce(Return(1));
+  EXPECT_CALL(udp_listener_config.socket_factory_, getListenSocket(_));
+
+  HotRestartMessage::Request request;
+  request.mutable_pass_listen_socket()->set_address("udp://0.0.0.0:80");
+  HotRestartMessage message = hot_restarting_parent_.getListenSocketsForChild(request);
+  EXPECT_EQ(0, message.reply().pass_listen_socket().fd());
+}
+
+TEST_F(HotRestartingParentTest, GetListenSocketsForChildUnixDomainSocket) {
+  MockListenerManager listener_manager;
+  Network::MockListenerConfig listener_config;
+  std::vector<std::reference_wrapper<Network::ListenerConfig>> listeners;
+  Network::Address::InstanceConstSharedPtr local_address =
+      std::make_shared<Network::Address::PipeInstance>("domain.socket");
+  MockOptions options;
+  InSequence s;
+
+  listeners.push_back(std::ref(*static_cast<Network::ListenerConfig*>(&listener_config)));
+
+  EXPECT_CALL(server_, listenerManager()).WillOnce(ReturnRef(listener_manager));
+  EXPECT_CALL(listener_manager, listeners(ListenerManager::ListenerState::ACTIVE))
+      .WillOnce(Return(listeners));
+  EXPECT_CALL(listener_config, listenSocketFactory());
+  EXPECT_CALL(listener_config.socket_factory_, localAddress()).WillOnce(ReturnRef(local_address));
+  EXPECT_CALL(listener_config, bindToPort()).WillOnce(Return(true));
+  EXPECT_CALL(listener_config.socket_factory_, socketType())
+      .WillOnce(Return(Network::Socket::Type::Stream));
+
+  EXPECT_CALL(server_, options()).WillOnce(ReturnRef(options));
+  EXPECT_CALL(options, concurrency()).WillOnce(Return(1));
+  EXPECT_CALL(listener_config.socket_factory_, getListenSocket(_));
+
+  HotRestartMessage::Request request;
+  request.mutable_pass_listen_socket()->set_address("unix://domain.socket");
+  HotRestartMessage message = hot_restarting_parent_.getListenSocketsForChild(request);
+  EXPECT_EQ(0, message.reply().pass_listen_socket().fd());
+}
+
 TEST_F(HotRestartingParentTest, ExportStatsToChild) {
   Stats::TestUtil::TestStore store;
   MockListenerManager listener_manager;