udp: Add use_original_src_ip feature on udp proxy

api/envoy/extensions/filters/udp/udp_proxy/v3/udp_proxy.proto

@@ -35,4 +35,19 @@ message UdpProxyConfig {
   // The idle timeout for sessions. Idle is defined as no datagrams between received or sent by
   // the session. The default if not specified is 1 minute.
   google.protobuf.Duration idle_timeout = 3;
+
+  // Use the remote downstream IP address as the sender IP address when sending packets to upstream hosts.
+  // This option requires Envoy to be run with the *CAP_NET_ADMIN* capability on Linux.
+  // This option does not preserve the remote downstream port.
+  // If this option is enabled, the IP address of sent datagrams will be changed to the remote downstream IP address.
+  // This means that Envoy will not receive packets that are sent by upstream hosts because the upstream hosts
+  // will send the packets with the remote downstream IP address as the destination. All packets will be routed
+  // to the remote downstream directly if there are route rules on the upstream host side.
+  // There are two options to return the packets back to the remote downstream.
+  // The first one is to use DSR (Direct Server Return).
+  // The other one is to configure routing rules on the upstream hosts to forward
+  // all packets back to Envoy and configure iptables rules on the host running Envoy to
+  // forward all packets from upstream hosts to the Envoy process so that Envoy can forward the packets to the downstream.
+  // If the platform does not support this option, Envoy will raise a configuration error.
+  bool use_original_src_ip = 4;
 }

docs/root/configuration/listeners/udp_filters/udp_proxy.rst

@@ -20,6 +20,10 @@ Each session is index by the 4-tuple consisting of source IP/port and local IP/p
 datagram is received on. Sessions last until the :ref:`idle timeout
 <envoy_v3_api_field_extensions.filters.udp.udp_proxy.v3.UdpProxyConfig.idle_timeout>` is reached.
 
+The UDP proxy listener filter also can operate as a *transparent* proxy if the
+:ref:`use_original_src_ip <envoy_v3_api_msg_extensions.filters.udp.udp_proxy.v3.UdpProxyConfig>`
+field is set. But please keep in mind that it does not forward the port to upstreams. It forwards only the IP address to upstreams.
+
 Load balancing and unhealthy host handling
 ------------------------------------------
 

docs/root/version_history/current.rst

@@ -91,6 +91,7 @@ New Features
 * tap: added :ref:`generic body matcher<envoy_v3_api_msg_config.tap.v3.HttpGenericBodyMatch>` to scan http requests and responses for text or hex patterns.
 * tcp: switched the TCP connection pool to the new "shared" connection pool, sharing a common code base with HTTP and HTTP/2. Any unexpected behavioral changes can be temporarily reverted by setting `envoy.reloadable_features.new_tcp_connection_pool` to false.
 * tcp_proxy: allow earlier network filters to set metadataMatchCriteria on the connection StreamInfo to influence load balancing.
+* udp_proxy: added :ref:`use_original_src_ip <envoy_v3_api_msg_extensions.filters.udp.udp_proxy.v3.UdpProxyConfig>` option to replicate the downstream remote address of the packets on the upstream side of Envoy. It is similar to :ref:`original source filter <envoy_v3_api_msg_extensions.filters.listener.original_src.v3.OriginalSrc>`.
 * watchdog: support randomizing the watchdog's kill timeout to prevent synchronized kills via a maximium jitter parameter :ref:`max_kill_timeout_jitter<envoy_v3_api_field_config.bootstrap.v3.Watchdog.max_kill_timeout_jitter>`.
 * watchdog: supports an extension point where actions can be registered to fire on watchdog events such as miss, megamiss, kill and multikill. See ref:`watchdog actions<envoy_v3_api_field_config.bootstrap.v3.Watchdog.actions>`.
 * xds: added :ref:`extension config discovery<envoy_v3_api_msg_config.core.v3.ExtensionConfigSource>` support for HTTP filters.

include/envoy/api/os_sys_calls.h

@@ -72,6 +72,13 @@ class OsSysCalls {
    */
   virtual bool supportsUdpGso() const PURE;
 
+  /**
+   * return true if the OS support IP_TRANSPARENT or IPV6_TRANSPARENT options
+   * @param check_v4only If it is true, check if the OS supports only IP_TRANSPARENT option, else
+   * check if the OS supports IPV6_TRANSPARENT option as well.
+   */
+  virtual bool supportsIpTransparent(bool check_v4only) const PURE;
+
   /**
    * Release all resources allocated for fd.
    * @return zero on success, -1 returned otherwise.

source/common/api/posix/os_sys_calls_impl.cc

@@ -110,6 +110,49 @@ bool OsSysCallsImpl::supportsUdpGso() const {
 #endif
 }
 
+bool OsSysCallsImpl::supportsIpTransparent(bool check_v4only) const {
+#if !defined(__linux__)
+  UNREFERENCED_PARAMETER(check_v4only);
+  return false;
+#else
+  // The linux kernel supports IP_TRANSPARENT by following patch(starting from v2.6.28) :
+  // https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/commit/net/ipv4/ip_sockglue.c?id=f5715aea4564f233767ea1d944b2637a5fd7cd2e
+  //
+  // The linux kernel supports IPV6_TRANSPARENT by following patch(starting from v2.6.37) :
+  // https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/commit/net/ipv6/ipv6_sockglue.c?id=6c46862280c5f55eda7750391bc65cd7e08c7535
+  //
+  // So, almost recent linux kernel supports both IP_TRANSPARENT and IPV6_TRANSPARENT options.
+  //
+  // And these socket options need CAP_NET_ADMIN capability to be applied.
+  // The CAP_NET_ADMIN capability should be applied by root user before call this function.
+  static const bool is_supported = [check_v4only] {
+    // Check ipv4 case
+    int fd = ::socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP);
+    if (fd < 0) {
+      return false;
+    }
+    int val = 1;
+    bool result = (0 == ::setsockopt(fd, IPPROTO_IP, IP_TRANSPARENT, &val, sizeof(val)));
+    ::close(fd);
+    if (!result) {
+      return false;
+    }
+    if (!check_v4only) {
+      // Check ipv6 case
+      fd = ::socket(AF_INET6, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP);
+      if (fd < 0) {
+        return false;
+      }
+      val = 1;
+      result = (0 == ::setsockopt(fd, IPPROTO_IPV6, IPV6_TRANSPARENT, &val, sizeof(val)));
+      ::close(fd);
+    }
+    return result;
+  }();
+  return is_supported;
+#endif
+}
+
 SysCallIntResult OsSysCallsImpl::ftruncate(int fd, off_t length) {
   const int rc = ::ftruncate(fd, length);
   return {rc, rc != -1 ? 0 : errno};

source/common/api/posix/os_sys_calls_impl.h

@@ -24,6 +24,7 @@ class OsSysCallsImpl : public OsSysCalls {
   bool supportsMmsg() const override;
   bool supportsUdpGro() const override;
   bool supportsUdpGso() const override;
+  bool supportsIpTransparent(bool check_v4only) const override;
   SysCallIntResult close(os_fd_t fd) override;
   SysCallIntResult ftruncate(int fd, off_t length) override;
   SysCallPtrResult mmap(void* addr, size_t length, int prot, int flags, int fd,

source/common/api/win32/os_sys_calls_impl.cc

@@ -179,6 +179,11 @@ bool OsSysCallsImpl::supportsUdpGso() const {
   return false;
 }
 
+bool OsSysCallsImpl::supportsIpTransparent(bool) const {
+  // Windows doesn't support it.
+  return false;
+}
+
 SysCallIntResult OsSysCallsImpl::ftruncate(int fd, off_t length) {
   const int rc = ::_chsize_s(fd, length);
   return {rc, rc == 0 ? 0 : errno};

source/common/api/win32/os_sys_calls_impl.h

@@ -24,6 +24,7 @@ class OsSysCallsImpl : public OsSysCalls {
   bool supportsMmsg() const override;
   bool supportsUdpGro() const override;
   bool supportsUdpGso() const override;
+  bool supportsIpTransparent(bool check_v4only) const override;
   SysCallIntResult close(os_fd_t fd) override;
   SysCallIntResult ftruncate(int fd, off_t length) override;
   SysCallPtrResult mmap(void* addr, size_t length, int prot, int flags, int fd,

source/extensions/filters/udp/udp_proxy/BUILD

@@ -16,9 +16,13 @@ envoy_cc_library(
     deps = [
         "//include/envoy/event:file_event_interface",
         "//include/envoy/event:timer_interface",
+        "//include/envoy/network:exception_interface",
         "//include/envoy/network:filter_interface",
         "//include/envoy/network:listener_interface",
         "//include/envoy/upstream:cluster_manager_interface",
+        "//source/common/api:os_sys_calls_lib",
+        "//source/common/network:socket_lib",
+        "//source/common/network:socket_option_factory_lib",
         "//source/common/network:utility_lib",
         "@envoy_api//envoy/extensions/filters/udp/udp_proxy/v3:pkg_cc_proto",
     ],

source/extensions/filters/udp/udp_proxy/udp_proxy_filter.cc

@@ -1,7 +1,10 @@
 #include "extensions/filters/udp/udp_proxy/udp_proxy_filter.h"
 
+#include "envoy/network/exception.h"
 #include "envoy/network/listener.h"
 
+#include "common/network/socket_option_factory.h"
+
 namespace Envoy {
 namespace Extensions {
 namespace UdpFilters {
@@ -12,6 +15,16 @@ UdpProxyFilter::UdpProxyFilter(Network::UdpReadFilterCallbacks& callbacks,
     : UdpListenerReadFilter(callbacks), config_(config),
       cluster_update_callbacks_(
           config->clusterManager().addThreadLocalClusterUpdateCallbacks(*this)) {
+  bool check_v4only =
+      callbacks.udpListener().localAddress()->ip()->version() == Network::Address::IpVersion::v4;
+  if (config_->usingOriginalSrcIp() &&
+      !Api::OsSysCallsSingleton::get().supportsIpTransparent(check_v4only)) {
+    throw Network::CreateListenerException(
+        fmt::format("The platform does not support either {} or the envoy is not running with the "
+                    "CAP_NET_ADMIN capability.",
+                    check_v4only ? "IP_TRANSPARENT" : "IPV6_TRANSPARENT"));
+  }
+
   Upstream::ThreadLocalCluster* cluster = config->clusterManager().get(config->cluster());
   if (cluster != nullptr) {
     onClusterAddOrUpdate(*cluster);
@@ -152,13 +165,14 @@ void UdpProxyFilter::ClusterInfo::removeSession(const ActiveSession* session) {
 UdpProxyFilter::ActiveSession::ActiveSession(ClusterInfo& cluster,
                                              Network::UdpRecvData::LocalPeerAddresses&& addresses,
                                              const Upstream::HostConstSharedPtr& host)
-    : cluster_(cluster), addresses_(std::move(addresses)), host_(host),
+    : cluster_(cluster), use_original_src_ip_(cluster_.filter_.config_->usingOriginalSrcIp()),
+      addresses_(std::move(addresses)), host_(host),
       idle_timer_(cluster.filter_.read_callbacks_->udpListener().dispatcher().createTimer(
           [this] { onIdleTimer(); })),
       // NOTE: The socket call can only fail due to memory/fd exhaustion. No local ephemeral port
       //       is bound until the first packet is sent to the upstream host.
-      io_handle_(cluster.filter_.createIoHandle(host)),
-      socket_event_(io_handle_->createFileEvent(
+      socket_(cluster.filter_.createSocket(host)),
+      socket_event_(socket_->ioHandle().createFileEvent(
           cluster.filter_.read_callbacks_->udpListener().dispatcher(),
           [this](uint32_t) { onReadReady(); }, Event::PlatformDefaultTriggerType,
           Event::FileReadyType::Read)) {
@@ -172,6 +186,17 @@ UdpProxyFilter::ActiveSession::ActiveSession(ClusterInfo& cluster,
       .connections()
       .inc();
 
+  if (use_original_src_ip_) {
+    const Network::Socket::OptionsSharedPtr socket_options =
+        Network::SocketOptionFactory::buildIpTransparentOptions();
+    const bool ok = Network::Socket::applyOptions(
+        socket_options, *socket_, envoy::config::core::v3::SocketOption::STATE_PREBIND);
+
+    RELEASE_ASSERT(ok, "Should never occur!");
+    ENVOY_LOG(debug, "The original src is enabled for address {}.",
+              addresses_.peer_->asStringView());
+  }
+
   // TODO(mattklein123): Enable dropped packets socket option. In general the Socket abstraction
   // does not work well right now for client sockets. It's too heavy weight and is aimed at listener
   // sockets. We need to figure out how to either refactor Socket into something that works better
@@ -204,7 +229,7 @@ void UdpProxyFilter::ActiveSession::onReadReady() {
   //                     not trying to populate the local address for received packets.
   uint32_t packets_dropped = 0;
   const Api::IoErrorPtr result = Network::Utility::readPacketsFromSocket(
-      *io_handle_, *addresses_.local_, *this, cluster_.filter_.config_->timeSource(),
+      socket_->ioHandle(), *addresses_.local_, *this, cluster_.filter_.config_->timeSource(),
       packets_dropped);
   // TODO(mattklein123): Handle no error when we limit the number of packets read.
   if (result->getErrorCode() != Api::IoError::IoErrorCode::Again) {
@@ -226,10 +251,12 @@ void UdpProxyFilter::ActiveSession::write(const Buffer::Instance& buffer) {
 
   // NOTE: On the first write, a local ephemeral port is bound, and thus this write can fail due to
   //       port exhaustion.
-  // NOTE: We do not specify the local IP to use for the sendmsg call. We allow the OS to select
-  //       the right IP based on outbound routing rules.
+  // NOTE: We do not specify the local IP to use for the sendmsg call if use_original_src_ip_ is not
+  //       set. We allow the OS to select the right IP based on outbound routing rules if
+  //       use_original_src_ip_ is not set, else use downstream peer IP as local IP.
+  const Network::Address::Ip* local_ip = use_original_src_ip_ ? addresses_.peer_->ip() : nullptr;
   Api::IoCallUint64Result rc =
-      Network::Utility::writeToSocket(*io_handle_, buffer, nullptr, *host_->address());
+      Network::Utility::writeToSocket(socket_->ioHandle(), buffer, local_ip, *host_->address());
   if (!rc.ok()) {
     cluster_.cluster_stats_.sess_tx_errors_.inc();
   } else {

source/extensions/filters/udp/udp_proxy/udp_proxy_filter.h

@@ -6,6 +6,8 @@
 #include "envoy/network/filter.h"
 #include "envoy/upstream/cluster_manager.h"
 
+#include "common/api/os_sys_calls_impl.h"
+#include "common/network/socket_impl.h"
 #include "common/network/socket_interface.h"
 #include "common/network/utility.h"
 
@@ -63,11 +65,13 @@ class UdpProxyFilterConfig {
                        const envoy::extensions::filters::udp::udp_proxy::v3::UdpProxyConfig& config)
       : cluster_manager_(cluster_manager), time_source_(time_source), cluster_(config.cluster()),
         session_timeout_(PROTOBUF_GET_MS_OR_DEFAULT(config, idle_timeout, 60 * 1000)),
+        use_original_src_ip_(config.use_original_src_ip()),
         stats_(generateStats(config.stat_prefix(), root_scope)) {}
 
   const std::string& cluster() const { return cluster_; }
   Upstream::ClusterManager& clusterManager() const { return cluster_manager_; }
   std::chrono::milliseconds sessionTimeout() const { return session_timeout_; }
+  bool usingOriginalSrcIp() const { return use_original_src_ip_; }
   UdpProxyDownstreamStats& stats() const { return stats_; }
   TimeSource& timeSource() const { return time_source_; }
 
@@ -83,6 +87,7 @@ class UdpProxyFilterConfig {
   TimeSource& time_source_;
   const std::string cluster_;
   const std::chrono::milliseconds session_timeout_;
+  const bool use_original_src_ip_;
   mutable UdpProxyDownstreamStats stats_;
 };
 
@@ -135,6 +140,7 @@ class UdpProxyFilter : public Network::UdpListenerReadFilter,
     }
 
     ClusterInfo& cluster_;
+    const bool use_original_src_ip_;
     const Network::UdpRecvData::LocalPeerAddresses addresses_;
     const Upstream::HostConstSharedPtr host_;
     // TODO(mattklein123): Consider replacing an idle timer for each session with a last used
@@ -143,10 +149,10 @@ class UdpProxyFilter : public Network::UdpListenerReadFilter,
     // idle timeouts work so we should consider unifying the implementation if we move to a time
     // stamp and scan approach.
     const Event::TimerPtr idle_timer_;
-    // The IO handle is used for writing packets to the selected upstream host as well as receiving
+    // The socket is used for writing packets to the selected upstream host as well as receiving
     // packets from the upstream host. Note that a a local ephemeral port is bound on the first
     // write to the upstream host.
-    const Network::IoHandlePtr io_handle_;
+    const Network::SocketPtr socket_;
     const Event::FileEventPtr socket_event_;
   };
 
@@ -220,9 +226,9 @@ class UdpProxyFilter : public Network::UdpListenerReadFilter,
         host_to_sessions_;
   };
 
-  virtual Network::IoHandlePtr createIoHandle(const Upstream::HostConstSharedPtr& host) {
+  virtual Network::SocketPtr createSocket(const Upstream::HostConstSharedPtr& host) {
     // Virtual so this can be overridden in unit tests.
-    return Network::ioHandleForAddr(Network::Socket::Type::Datagram, host->address());
+    return std::make_unique<Network::SocketImpl>(Network::Socket::Type::Datagram, host->address());
   }
 
   // Upstream::ClusterUpdateCallbacks

test/extensions/filters/udp/udp_proxy/BUILD

@@ -16,13 +16,16 @@ envoy_extension_cc_test(
     srcs = ["udp_proxy_filter_test.cc"],
     extension_name = "envoy.filters.udp_listener.udp_proxy",
     deps = [
+        "//include/envoy/network:exception_interface",
         "//source/extensions/filters/udp/udp_proxy:udp_proxy_filter_lib",
-        "//test/mocks/network:io_handle_mocks",
+        "//test/mocks/api:api_mocks",
+        "//test/mocks/network:socket_mocks",
         "//test/mocks/upstream:cluster_manager_mocks",
         "//test/mocks/upstream:cluster_update_callbacks_handle_mocks",
         "//test/mocks/upstream:cluster_update_callbacks_mocks",
         "//test/mocks/upstream:host_mocks",
         "//test/mocks/upstream:thread_local_cluster_mocks",
+        "//test/test_common:threadsafe_singleton_injector_lib",
         "@envoy_api//envoy/extensions/filters/udp/udp_proxy/v3:pkg_cc_proto",
     ],
 )