reverse_tunnels: add validation in the network filter

Signed-off-by: Rohit Agrawal <[email protected]>

Author: agrawroh

api/envoy/extensions/filters/network/reverse_tunnel/v3/reverse_tunnel.proto

@@ -19,27 +19,98 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // Reverse Tunnel Network Filter :ref:`configuration overview <config_network_filters_reverse_tunnel>`.
 // [#extension: envoy.filters.network.reverse_tunnel]
 
+// Validation configuration for reverse tunnel identifiers.
+// Validates the node ID and cluster ID extracted from reverse tunnel handshake headers
+// against expected values specified using format strings.
+message Validation {
+  // Format string to extract the expected node identifier for validation.
+  // The formatted value is compared against the ``x-envoy-reverse-tunnel-node-id`` header
+  // from the incoming handshake request. If they do not match, the connection is rejected
+  // with HTTP ``403 Forbidden``.
+  //
+  // Supports Envoy's :ref:`command operators <config_access_log_command_operators>`:
+  //
+  // * ``%DYNAMIC_METADATA(namespace:key)%``: Extract expected value from dynamic metadata.
+  // * ``%FILTER_STATE(key)%``: Extract expected value from filter state.
+  // * ``%DOWNSTREAM_REMOTE_ADDRESS%``: Use downstream connection IP address.
+  // * Plain strings: Use a static expected value.
+  //
+  // If empty, node ID validation is skipped.
+  //
+  // Example using dynamic metadata allowlist:
+  //
+  // .. code-block:: yaml
+  //
+  //    node_id_format: "%DYNAMIC_METADATA(envoy.reverse_tunnel.allowlist:expected_node_id)%"
+  //
+  string node_id_format = 1 [(validate.rules).string = {max_len: 1024}];
+
+  // Format string to extract the expected cluster identifier for validation.
+  // The formatted value is compared against the ``x-envoy-reverse-tunnel-cluster-id`` header
+  // from the incoming handshake request. If they do not match, the connection is rejected
+  // with HTTP ``403 Forbidden``.
+  //
+  // Supports the same :ref:`command operators <config_access_log_command_operators>` as
+  // ``node_id_format``.
+  //
+  // If empty, cluster ID validation is skipped.
+  //
+  // Example using filter state:
+  //
+  // .. code-block:: yaml
+  //
+  //    cluster_id_format: "%FILTER_STATE(expected_cluster_id)%"
+  //
+  string cluster_id_format = 2 [(validate.rules).string = {max_len: 1024}];
+
+  // Whether to emit validation results as dynamic metadata.
+  // When enabled, the filter emits metadata under the namespace specified by
+  // ``dynamic_metadata_namespace`` containing:
+  //
+  // * ``node_id``: The actual node ID from the handshake request.
+  // * ``cluster_id``: The actual cluster ID from the handshake request.
+  // * ``validation_result``: Either ``allowed`` or ``denied``.
+  //
+  // This metadata can be used by subsequent filters or for access logging.
+  // Defaults to ``false``.
+  bool emit_dynamic_metadata = 3;
+
+  // Namespace for emitted dynamic metadata when ``emit_dynamic_metadata`` is ``true``.
+  // If not specified, defaults to ``envoy.filters.network.reverse_tunnel``.
+  string dynamic_metadata_namespace = 4 [(validate.rules).string = {max_len: 255}];
+}
+
 // Configuration for the reverse tunnel network filter.
 // This filter handles reverse tunnel connection acceptance and rejection by processing
 // HTTP requests where required identification values are provided via HTTP headers.
+// [#next-free-field: 6]
 message ReverseTunnel {
   // Ping interval for health checks on established reverse tunnel connections.
-  // If not specified, defaults to 2 seconds.
+  // If not specified, defaults to ``2 seconds``.
   google.protobuf.Duration ping_interval = 1 [(validate.rules).duration = {
     lte {seconds: 300}
     gte {nanos: 1000000}
   }];
 
   // Whether to automatically close connections after processing reverse tunnel requests.
-  // When set to true, connections are closed after acceptance or rejection.
-  // When set to false, connections remain open for potential reuse. Defaults to false.
+  //
+  // * When set to ``true``, connections are closed after acceptance or rejection.
+  // * When set to ``false``, connections remain open for potential reuse.
+  //
+  // Defaults to ``false``.
   bool auto_close_connections = 2;
 
   // HTTP path to match for reverse tunnel requests.
-  // If not specified, defaults to "/reverse_connections/request".
+  // If not specified, defaults to ``/reverse_connections/request``.
   string request_path = 3 [(validate.rules).string = {min_len: 1 max_len: 255 ignore_empty: true}];
 
   // HTTP method to match for reverse tunnel requests.
   // If not specified (``METHOD_UNSPECIFIED``), this defaults to ``GET``.
   config.core.v3.RequestMethod request_method = 4 [(validate.rules).enum = {defined_only: true}];
+
+  // Optional validation configuration for node and cluster identifiers.
+  // If specified, the filter validates the ``x-envoy-reverse-tunnel-node-id`` and
+  // ``x-envoy-reverse-tunnel-cluster-id`` headers against expected values extracted
+  // using format strings. Requests that fail validation are rejected with HTTP ``403 Forbidden``.
+  Validation validation = 5;
 }

source/extensions/filters/network/reverse_tunnel/BUILD

@@ -27,13 +27,16 @@ envoy_cc_library(
     hdrs = ["reverse_tunnel_filter.h"],
     deps = [
         "//envoy/buffer:buffer_interface",
+        "//envoy/formatter:substitution_formatter_interface",
         "//envoy/http:codec_interface",
         "//envoy/network:connection_interface",
         "//envoy/network:filter_interface",
         "//envoy/ssl:connection_interface",
         "//envoy/thread_local:thread_local_interface",
         "//source/common/buffer:buffer_lib",
         "//source/common/common:logger_lib",
+        "//source/common/formatter:substitution_format_string_lib",
+        "//source/common/formatter:substitution_formatter_lib",
         "//source/common/http:codes_lib",
         "//source/common/http:header_map_lib",
         "//source/common/http:headers_lib",
@@ -51,7 +54,9 @@ envoy_cc_library(
         "//source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface:reverse_tunnel_acceptor_includes",
         "//source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface:reverse_tunnel_acceptor_lib",
         "//source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface:upstream_socket_manager_lib",
+        "//source/server:generic_factory_context_lib",
         "//source/server:null_overload_manager_lib",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
         "@envoy_api//envoy/extensions/filters/network/reverse_tunnel/v3:pkg_cc_proto",
     ],
 )

source/extensions/filters/network/reverse_tunnel/config.cc

@@ -10,7 +10,12 @@ namespace ReverseTunnel {
 Network::FilterFactoryCb ReverseTunnelFilterConfigFactory::createFilterFactoryFromProtoTyped(
     const envoy::extensions::filters::network::reverse_tunnel::v3::ReverseTunnel& proto_config,
     Server::Configuration::FactoryContext& context) {
-  auto config = std::make_shared<ReverseTunnelFilterConfig>(proto_config, context);
+  auto config_or_error = ReverseTunnelFilterConfig::create(proto_config, context);
+  if (!config_or_error.ok()) {
+    throw EnvoyException(std::string(config_or_error.status().message()));
+  }
+  auto config = config_or_error.value();
+
   // Capture scope and overload manager pointers to avoid dangling references.
   Stats::Scope* scope = &context.scope();
   Server::OverloadManager* overload_manager = &context.serverFactoryContext().overloadManager();

source/extensions/filters/network/reverse_tunnel/reverse_tunnel_filter.cc

@@ -1,10 +1,14 @@
 #include "source/extensions/filters/network/reverse_tunnel/reverse_tunnel_filter.h"
 
 #include "envoy/buffer/buffer.h"
+#include "envoy/config/core/v3/substitution_format_string.pb.h"
 #include "envoy/network/connection.h"
 #include "envoy/server/overload/overload_manager.h"
 
 #include "source/common/buffer/buffer_impl.h"
+#include "source/common/config/datasource.h"
+#include "source/common/formatter/substitution_format_string.h"
+#include "source/common/formatter/substitution_formatter.h"
 #include "source/common/http/codes.h"
 #include "source/common/http/header_map_impl.h"
 #include "source/common/http/headers.h"
@@ -15,6 +19,7 @@
 #include "source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface/reverse_tunnel_acceptor.h"
 #include "source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface/reverse_tunnel_acceptor_extension.h"
 #include "source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface/upstream_socket_manager.h"
+#include "source/server/generic_factory_context.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -29,9 +34,59 @@ ReverseTunnelFilter::ReverseTunnelStats::generateStats(const std::string& prefix
 }
 
 // ReverseTunnelFilterConfig implementation.
+absl::StatusOr<std::shared_ptr<ReverseTunnelFilterConfig>> ReverseTunnelFilterConfig::create(
+    const envoy::extensions::filters::network::reverse_tunnel::v3::ReverseTunnel& proto_config,
+    Server::Configuration::FactoryContext& context) {
+
+  Formatter::FormatterConstSharedPtr node_id_formatter;
+  Formatter::FormatterConstSharedPtr cluster_id_formatter;
+
+  // Create formatters for validation if configured.
+  if (proto_config.has_validation()) {
+    Server::GenericFactoryContextImpl generic_context(context.serverFactoryContext(),
+                                                      context.messageValidationVisitor());
+
+    const auto& validation = proto_config.validation();
+
+    // Create node_id formatter if configured.
+    if (!validation.node_id_format().empty()) {
+      envoy::config::core::v3::SubstitutionFormatString node_id_format_config;
+      node_id_format_config.mutable_text_format_source()->set_inline_string(
+          validation.node_id_format());
+
+      auto formatter_or_error = Formatter::SubstitutionFormatStringUtils::fromProtoConfig(
+          node_id_format_config, generic_context);
+      if (!formatter_or_error.ok()) {
+        return absl::InvalidArgumentError(fmt::format("Failed to parse node_id_format: {}",
+                                                      formatter_or_error.status().message()));
+      }
+      node_id_formatter = std::move(formatter_or_error.value());
+    }
+
+    // Create cluster_id formatter if configured.
+    if (!validation.cluster_id_format().empty()) {
+      envoy::config::core::v3::SubstitutionFormatString cluster_id_format_config;
+      cluster_id_format_config.mutable_text_format_source()->set_inline_string(
+          validation.cluster_id_format());
+
+      auto formatter_or_error = Formatter::SubstitutionFormatStringUtils::fromProtoConfig(
+          cluster_id_format_config, generic_context);
+      if (!formatter_or_error.ok()) {
+        return absl::InvalidArgumentError(fmt::format("Failed to parse cluster_id_format: {}",
+                                                      formatter_or_error.status().message()));
+      }
+      cluster_id_formatter = std::move(formatter_or_error.value());
+    }
+  }
+
+  return std::shared_ptr<ReverseTunnelFilterConfig>(new ReverseTunnelFilterConfig(
+      proto_config, std::move(node_id_formatter), std::move(cluster_id_formatter)));
+}
+
 ReverseTunnelFilterConfig::ReverseTunnelFilterConfig(
     const envoy::extensions::filters::network::reverse_tunnel::v3::ReverseTunnel& proto_config,
-    Server::Configuration::FactoryContext&)
+    Formatter::FormatterConstSharedPtr node_id_formatter,
+    Formatter::FormatterConstSharedPtr cluster_id_formatter)
     : ping_interval_(proto_config.has_ping_interval()
                          ? std::chrono::milliseconds(
                                DurationUtil::durationToMilliseconds(proto_config.ping_interval()))
@@ -46,7 +101,78 @@ ReverseTunnelFilterConfig::ReverseTunnelFilterConfig(
           method = envoy::config::core::v3::GET;
         }
         return envoy::config::core::v3::RequestMethod_Name(method);
-      }()) {}
+      }()),
+      node_id_formatter_(std::move(node_id_formatter)),
+      cluster_id_formatter_(std::move(cluster_id_formatter)),
+      emit_dynamic_metadata_(proto_config.has_validation() &&
+                             proto_config.validation().emit_dynamic_metadata()),
+      dynamic_metadata_namespace_(
+          proto_config.has_validation() &&
+                  !proto_config.validation().dynamic_metadata_namespace().empty()
+              ? proto_config.validation().dynamic_metadata_namespace()
+              : "envoy.filters.network.reverse_tunnel") {}
+
+bool ReverseTunnelFilterConfig::validateIdentifiers(
+    absl::string_view node_id, absl::string_view cluster_id,
+    const StreamInfo::StreamInfo& stream_info) const {
+
+  // If no validation configured, pass validation.
+  if (!node_id_formatter_ && !cluster_id_formatter_) {
+    return true;
+  }
+
+  // Validate node_id if formatter is configured.
+  if (node_id_formatter_) {
+    const std::string expected_node_id = node_id_formatter_->formatWithContext({}, stream_info);
+    if (!expected_node_id.empty() && expected_node_id != node_id) {
+      ENVOY_LOG(debug, "reverse_tunnel: node_id validation failed. Expected: '{}', Actual: '{}'",
+                expected_node_id, node_id);
+      return false;
+    }
+  }
+
+  // Validate cluster_id if formatter is configured.
+  if (cluster_id_formatter_) {
+    const std::string expected_cluster_id =
+        cluster_id_formatter_->formatWithContext({}, stream_info);
+    if (!expected_cluster_id.empty() && expected_cluster_id != cluster_id) {
+      ENVOY_LOG(debug, "reverse_tunnel: cluster_id validation failed. Expected: '{}', Actual: '{}'",
+                expected_cluster_id, cluster_id);
+      return false;
+    }
+  }
+
+  return true;
+}
+
+void ReverseTunnelFilterConfig::emitValidationMetadata(
+    absl::string_view node_id, absl::string_view cluster_id, bool validation_passed,
+    const StreamInfo::StreamInfo& stream_info) const {
+  if (!emit_dynamic_metadata_) {
+    return;
+  }
+
+  Protobuf::Struct metadata;
+  auto& fields = *metadata.mutable_fields();
+
+  // Emit actual identifiers.
+  fields["node_id"].set_string_value(std::string(node_id));
+  fields["cluster_id"].set_string_value(std::string(cluster_id));
+
+  // Emit validation result.
+  fields["validation_result"].set_string_value(validation_passed ? "allowed" : "denied");
+
+  // StreamInfo::setDynamicMetadata is not const, so we need to cast away constness.
+  // This is safe because we're modifying dynamic metadata, which is mutable by design.
+  const_cast<StreamInfo::StreamInfo&>(stream_info)
+      .setDynamicMetadata(dynamic_metadata_namespace_, metadata);
+
+  ENVOY_LOG(trace,
+            "reverse_tunnel: emitted dynamic metadata to namespace '{}': node_id={}, "
+            "cluster_id={}, validation_result={}",
+            dynamic_metadata_namespace_, node_id, cluster_id,
+            validation_passed ? "allowed" : "denied");
+}
 
 // ReverseTunnelFilter implementation.
 ReverseTunnelFilter::ReverseTunnelFilter(ReverseTunnelFilterConfigSharedPtr config,
@@ -191,6 +317,25 @@ void ReverseTunnelFilter::RequestDecoderImpl::processIfComplete(bool end_stream)
   const absl::string_view cluster_id = cluster_vals[0]->value().getStringView();
   const absl::string_view tenant_id = tenant_vals[0]->value().getStringView();
 
+  // Validate node_id and cluster_id if validation is configured.
+  const auto& connection = parent_.read_callbacks_->connection();
+  const bool validation_passed =
+      parent_.config_->validateIdentifiers(node_id, cluster_id, connection.streamInfo());
+
+  // Emit validation metadata if configured.
+  parent_.config_->emitValidationMetadata(node_id, cluster_id, validation_passed,
+                                          connection.streamInfo());
+
+  if (!validation_passed) {
+    parent_.stats_.validation_failed_.inc();
+    ENVOY_CONN_LOG(debug, "reverse_tunnel: validation failed for node '{}', cluster '{}'",
+                   parent_.read_callbacks_->connection(), node_id, cluster_id);
+    sendLocalReply(Http::Code::Forbidden, "Validation failed", nullptr, absl::nullopt,
+                   "reverse_tunnel_validation_failed");
+    parent_.read_callbacks_->connection().close(Network::ConnectionCloseType::FlushWrite);
+    return;
+  }
+
   // Respond with 200 OK.
   auto resp_headers = Http::ResponseHeaderMapImpl::create();
   resp_headers->setStatus(200);