Add routing capabilities to tcp_proxy

configs/envoy_service_to_service.template.json

@@ -245,8 +245,14 @@
       "type": "read",
       "name": "tcp_proxy",
       "config": {
-        "cluster": "mongo_{{ key }}",
-        "stat_prefix": "mongo_{{ key }}"
+        "stat_prefix": "mongo_{{ key }}",
+        "route_config": {
+          "routes": [
+            {
+              "cluster": "mongo_{{ key }}"
+            }
+          ]
+        }
       }
     }]
   }{% if not loop.last %},{% endif -%}

docs/configuration/network_filters/tcp_proxy_filter.rst

@@ -11,19 +11,114 @@ TCP proxy :ref:`architecture overview <arch_overview_tcp_proxy>`.
     "type": "read",
     "name": "tcp_proxy",
     "config": {
-      "cluster": "...",
-      "stat_prefix": "..."
+      "stat_prefix": "...",
+      "route_config": "{...}"
     }
   }
 
-cluster
-  *(required, string)* The :ref:`cluster manager <arch_overview_cluster_manager>` cluster to connect
-  to when a new downstream network connection is received.
+:ref:`route_config <config_network_filters_tcp_proxy_route_config>`
+  *(required, object)* The route table for the filter.
+  All filter instances must have a route table, even if it is empty.
 
 stat_prefix
   *(required, string)* The prefix to use when emitting :ref:`statistics
   <config_network_filters_tcp_proxy_stats>`.
 
+.. _config_network_filters_tcp_proxy_route_config:
+
+Route Configuration
+-------------------
+
+.. code-block:: json
+
+  {
+    "routes": []
+  }
+
+:ref:`routes <config_network_filters_tcp_proxy_route>`
+  *(required, array)* An array of route entries that make up the route table.
+
+.. _config_network_filters_tcp_proxy_route:
+
+Route
+-----
+
+A TCP proxy route consists of a set of optional L4 criteria and the name of a
+:ref:`cluster <config_cluster_manager_cluster>`. If a downstream connection matches
+all the specified criteria, the cluster in the route is used for the corresponding upstream
+connection. Routes are tried in the order specified until a match is found. If no match is
+found, the connection is closed. A route with no criteria is valid and always produces a match.
+
+.. code-block:: json
+
+  {
+    "cluster": "...",
+    "destination_ip_list": [],
+    "destination_ports": "...",
+    "source_ip_list": [],
+    "source_ports": "..."
+  }
+
+cluster
+  *(required, string)* The :ref:`cluster <config_cluster_manager_cluster>` to connect
+  to when a the downstream network connection matches the specified criteria.
+
+destination_ip_list
+  *(optional, array)*  An optional list of IPv4 subnets in the form "a.b.c.d/xx".
+  The criteria is satisfied if the destination IP address of the downstream connection is
+  contained in at least one of the specified subnets.
+  If the parameter is not specified or the list is empty, the destination IP address is ignored.
+  The destination IP address of the downstream connection might be different from the addresses
+  on which the proxy is listening if the connection has been redirected.  Example:
+
+ .. code-block:: json
+
+    [
+      "192.168.3.0/24",
+      "50.1.2.3/32",
+      "10.15.0.0/16"
+    ]
+
+destination_ports
+  *(optional, string)* An optional string containing a comma-separated list of port numbers or
+  ranges. The criteria is satisfied if the destination port of the downstream connection
+  is contained in at least one of the specified ranges.
+  If the parameter is not specified or the list is empty, the destination port is ignored.
+  The destination port address of the downstream connection might be different from the port
+  on which the proxy is listening if the connection has been redirected. Example:
+
+ .. code-block:: json
+
+  {
+    "destination_ports": "1-1024,2048-4096,12345"
+  }
+
+source_ip_list
+  *(optional, array)*  An optional list of IPv4 subnets in the form "a.b.c.d/xx".
+  The criteria is satisfied if the source IP address of the downstream connection is contained
+  in at least one of the specified subnets. If the parameter is not specified or the list is empty,
+  the source IP address is ignored. Example:
+
+ .. code-block:: json
+
+    [
+      "192.168.3.0/24",
+      "50.1.2.3/32",
+      "10.15.0.0/16"
+    ]
+
+source_ports
+  *(optional, string)* An optional string containing a comma-separated list of port numbers or
+  ranges. The criteria is satisfied if the source port of the downstream connection is contained
+  in at least one of the specified ranges. If the parameter is not specified or the list is empty,
+  the source port is ignored.  Example:
+
+ .. code-block:: json
+
+  {
+    "source_ports": "1-1024,2048-4096,12345"
+  }
+
 .. _config_network_filters_tcp_proxy_stats:
 
 Statistics
@@ -37,5 +132,8 @@ statistics are rooted at *tcp.<stat_prefix>.* with the following statistics:
   :header: Name, Type, Description
   :widths: 1, 1, 2
 
+  downstream_cx_total, Counter, Total number of connections handled by the filter.
+  downstream_cx_no_route, Counter, Number of connections for which no matching route was found.
   downstream_cx_tx_bytes_total, Counter, Total bytes written to the downstream connection.
   downstream_cx_tx_bytes_buffered, Gauge, Total bytes currently buffered to the downstream connection.
+

include/envoy/network/connection.h

@@ -111,10 +111,18 @@ class Connection : public Event::DeferredDeletable, public FilterManager {
   virtual bool readEnabled() PURE;
 
   /**
-   * @return The address of the remote client
+   * @return The address of the remote client.
    */
   virtual const std::string& remoteAddress() PURE;
 
+  /**
+   * @return the local address of the connection. For client connections, this is the origin
+   * address. For server connections, this is the local destination address. For server connections
+   * it can be different from the proxy address if the downstream connection has been redirected or
+   * the proxy is operating in transparent mode.
+   */
+  virtual const std::string& localAddress() PURE;
+
   /**
    * Set the buffer stats to update when the connection's read/write buffers change. Note that
    * for performance reasons these stats are eventually consistent and may not always accurately

source/common/filter/auth/client_ssl.cc

@@ -8,6 +8,7 @@
 #include "common/http/message_impl.h"
 #include "common/http/utility.h"
 #include "common/json/config_schemas.h"
+#include "common/network/utility.h"
 
 namespace Filter {
 namespace Auth {
@@ -18,7 +19,7 @@ Config::Config(const Json::Object& config, ThreadLocal::Instance& tls, Upstream:
                Runtime::RandomGenerator& random)
     : RestApiFetcher(cm, config.getString("auth_api_cluster"), dispatcher, random,
                      std::chrono::milliseconds(config.getInteger("refresh_interval_ms", 60000))),
-      tls_(tls), tls_slot_(tls.allocateSlot()), ip_white_list_(config),
+      tls_(tls), tls_slot_(tls.allocateSlot()), ip_white_list_(config, "ip_white_list"),
       stats_(generateStats(stats_store, config.getString("stat_prefix"))) {
 
   config.validateSchema(Json::Schema::CLIENT_SSL_NETWORK_FILTER_SCHEMA);
@@ -97,7 +98,8 @@ void Instance::onEvent(uint32_t events) {
   }
 
   ASSERT(read_callbacks_->connection().ssl());
-  if (config_->ipWhiteList().contains(read_callbacks_->connection().remoteAddress())) {
+  if (config_->ipWhiteList().contains(
+          Network::Utility::hostFromUrl(read_callbacks_->connection().remoteAddress()))) {
     config_->stats().auth_ip_white_list_.inc();
     read_callbacks_->continueReading();
     return;

source/common/filter/auth/client_ssl.h

@@ -74,7 +74,7 @@ class Config : public Http::RestApiFetcher {
                           Stats::Store& stats_store, Runtime::RandomGenerator& random);
 
   const AllowedPrincipals& allowedPrincipals();
-  const Network::IpWhiteList& ipWhiteList() { return ip_white_list_; }
+  const Network::IpList& ipWhiteList() { return ip_white_list_; }
   GlobalStats& stats() { return stats_; }
 
 private:
@@ -92,7 +92,7 @@ class Config : public Http::RestApiFetcher {
 
   ThreadLocal::Instance& tls_;
   uint32_t tls_slot_;
-  Network::IpWhiteList ip_white_list_;
+  Network::IpList ip_white_list_;
   GlobalStats stats_;
 };
 

source/common/filter/tcp_proxy.cc

@@ -8,23 +8,97 @@
 #include "envoy/upstream/upstream.h"
 
 #include "common/common/assert.h"
+#include "common/common/empty_string.h"
 #include "common/json/config_schemas.h"
 #include "common/json/json_loader.h"
 
 namespace Filter {
 
+TcpProxyConfig::Route::Route(const Json::Object& config) {
+  cluster_name_ = config.getString("cluster");
+
+  if (config.hasObject("source_ip_list")) {
+    source_ips_ = Network::IpList(config.getStringArray("source_ip_list"));
+  }
+
+  if (config.hasObject("source_ports")) {
+    const std::string source_ports = config.getString("source_ports");
+    if (source_ports.empty()) {
+      throw EnvoyException("source_ports cannot be empty");
+    }
+    Network::Utility::parsePortRangeList(source_ports, source_port_ranges_);
+  }
+
+  if (config.hasObject("destination_ip_list")) {
+    destination_ips_ = Network::IpList(config.getStringArray("destination_ip_list"));
+  }
+
+  if (config.hasObject("destination_ports")) {
+    const std::string destination_ports = config.getString("destination_ports");
+    if (destination_ports.empty()) {
+      throw EnvoyException("destination_ports cannot be empty");
+    }
+    Network::Utility::parsePortRangeList(destination_ports, destination_port_ranges_);
+  }
+}
+
 TcpProxyConfig::TcpProxyConfig(const Json::Object& config,
                                Upstream::ClusterManager& cluster_manager, Stats::Store& stats_store)
-    : cluster_name_(config.getString("cluster")),
-      stats_(generateStats(config.getString("stat_prefix"), stats_store)) {
-
+    : stats_(generateStats(config.getString("stat_prefix"), stats_store)) {
   config.validateSchema(Json::Schema::TCP_PROXY_NETWORK_FILTER_SCHEMA);
 
-  if (!cluster_manager.get(cluster_name_)) {
-    throw EnvoyException(fmt::format("tcp proxy: unknown cluster '{}'", cluster_name_));
+  for (const Json::ObjectPtr& route_desc :
+       config.getObject("route_config")->getObjectArray("routes")) {
+    routes_.emplace_back(Route(*route_desc));
+
+    if (!cluster_manager.get(route_desc->getString("cluster"))) {
+      throw EnvoyException(fmt::format("tcp proxy: unknown cluster '{}' in TCP route",
+                                       route_desc->getString("cluster")));
+    }
   }
 }
 
+const std::string& TcpProxyConfig::getRouteFromEntries(Network::Connection& connection) {
+  for (const TcpProxyConfig::Route& route : routes_) {
+    if (!route.source_port_ranges_.empty() &&
+        !Network::Utility::portInRangeList(
+            Network::Utility::portFromUrl(connection.remoteAddress()), route.source_port_ranges_)) {
+      continue; // no match, try next route
+    }
+
+    if (!route.source_ips_.empty() &&
+        !route.source_ips_.contains(Network::Utility::hostFromUrl(connection.remoteAddress()))) {
+      continue; // no match, try next route
+    }
+
+    // If the route needs to match on destination address and port but they are not available
+    // (localAddress is empty), we skip it. The connection has a chance to match a different
+    // route that does not depend on destination address and port.
+    if ((!route.destination_port_ranges_.empty() || !route.destination_ips_.empty()) &&
+        connection.localAddress().empty()) {
+      continue;
+    }
+
+    if (!route.destination_port_ranges_.empty() &&
+        !Network::Utility::portInRangeList(Network::Utility::portFromUrl(connection.localAddress()),
+                                           route.destination_port_ranges_)) {
+      continue; // no match, try next route
+    }
+
+    if (!route.destination_ips_.empty() &&
+        !route.destination_ips_.contains(
+            Network::Utility::hostFromUrl(connection.localAddress()))) {
+      continue; // no match, try next route
+    }
+
+    // if we made it past all checks, the route matches
+    return route.cluster_name_;
+  }
+
+  // no match, no more routes to try
+  return EMPTY_STRING;
+}
+
 TcpProxy::TcpProxy(TcpProxyConfigPtr config, Upstream::ClusterManager& cluster_manager)
     : config_(config), cluster_manager_(cluster_manager), downstream_callbacks_(*this),
       upstream_callbacks_(new UpstreamCallbacks(*this)) {}
@@ -52,6 +126,7 @@ TcpProxyStats TcpProxyConfig::generateStats(const std::string& name, Stats::Stor
 void TcpProxy::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) {
   read_callbacks_ = &callbacks;
   conn_log_info("new tcp proxy session", read_callbacks_->connection());
+  config_->stats().downstream_cx_total_.inc();
   read_callbacks_->connection().addConnectionCallbacks(downstream_callbacks_);
   read_callbacks_->connection().setBufferStats({config_->stats().downstream_cx_rx_bytes_total_,
                                                 config_->stats().downstream_cx_rx_bytes_buffered_,
@@ -60,14 +135,25 @@ void TcpProxy::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callb
 }
 
 Network::FilterStatus TcpProxy::initializeUpstreamConnection() {
-  Upstream::ClusterInfoPtr cluster = cluster_manager_.get(config_->clusterName());
+  const std::string& cluster_name = config_->getRouteFromEntries(read_callbacks_->connection());
+
+  Upstream::ClusterInfoPtr cluster = cluster_manager_.get(cluster_name);
+
+  if (cluster) {
+    conn_log_debug("Creating connection to cluster {}", read_callbacks_->connection(),
+                   cluster_name);
+  } else {
+    config_->stats().downstream_cx_no_route_.inc();
+    read_callbacks_->connection().close(Network::ConnectionCloseType::NoFlush);
+    return Network::FilterStatus::StopIteration;
+  }
+
   if (!cluster->resourceManager(Upstream::ResourcePriority::Default).connections().canCreate()) {
     cluster->stats().upstream_cx_overflow_.inc();
     read_callbacks_->connection().close(Network::ConnectionCloseType::NoFlush);
     return Network::FilterStatus::StopIteration;
   }
-  Upstream::Host::CreateConnectionData conn_info =
-      cluster_manager_.tcpConnForCluster(config_->clusterName());
+  Upstream::Host::CreateConnectionData conn_info = cluster_manager_.tcpConnForCluster(cluster_name);
 
   upstream_connection_ = std::move(conn_info.connection_);
   read_callbacks_->upstreamHost(conn_info.host_description_);