reverse_tunnels: intercept RPING on upstream reverse connections (envoyproxy#43666)

## Commit Message
Intercept RPING on upstream reverse connections

## Additional Description
If a RPING is sent and then the connection is upgraded to http2 via the
conn pool the downstream reverse connection io handle's rping response
causes a 502 protocol error.

More specifically: Remote peer returned unexpected data while we
expected SETTINGS frame

Additional logs: 
```
2026-02-26 17:01:22.937 trace connection[thread=41 func=onReadReady file=source/common/network/connection_impl.cc:721] [Tags: "ConnectionId":"10614"] read ready. dispatch_buffered_data=0
2026-02-26 17:01:22.937 info misc[thread=41 func=read file=./source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface/reverse_connection_io_handle.h:89] ALERT GOT SOME RPING Bytes: RPING
2026-02-26 17:01:22.937 trace connection[thread=41 func=doRead file=source/common/network/raw_buffer_socket.cc:25] [Tags: "ConnectionId":"10614"] read returns: 20745
2026-02-26 17:01:22.937 info misc[thread=41 func=read file=./source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface/reverse_connection_io_handle.h:89] ALERT GOT SOME RPING Bytes: RPING
2026-02-26 17:01:22.937 trace connection[thread=41 func=doRead file=source/common/network/raw_buffer_socket.cc:39] [Tags: "ConnectionId":"10614"] read error: Resource temporarily unavailable, code: 0
2026-02-26 17:01:22.937 trace http2[thread=41 func=dispatch file=source/common/http/http2/codec_impl.cc:1077] [Tags: "ConnectionId":"10614"] dispatching 20745 bytes
2026-02-26 17:01:22.937 debug http2[thread=41 func=onError file=source/common/http/http2/codec_impl.cc:1395] [Tags: "ConnectionId":"10614"] invalid http2: Remote peer returned unexpected data while we expected SETTINGS frame.  Perhaps, peer does not support HTTP/2 properly.
```

The Alert was a custom log line added by me for easier debugging.
Its better for both the downstream and upstream rc io handles to share a
common read implementation.

## Testing
Unit tests.
Additionally, manually validated the example in the docs which also
works after the change.

Signed-off-by: aakugan <aakashganapathy2@gmail.com>

Author: aakugan

source/extensions/bootstrap/reverse_tunnel/common/BUILD

@@ -1,6 +1,7 @@
 load(
     "//bazel:envoy_build_system.bzl",
     "envoy_cc_extension",
+    "envoy_cc_library",
     "envoy_extension_package",
 )
 
@@ -22,3 +23,13 @@ envoy_cc_extension(
         "//source/common/http:headers_lib",
     ],
 )
+
+envoy_cc_library(
+    name = "rping_interceptor_lib",
+    srcs = ["rping_interceptor.cc"],
+    hdrs = ["rping_interceptor.h"],
+    deps = [
+        ":reverse_connection_utility_lib",
+        "//source/common/network:default_socket_interface_lib",
+    ],
+)

source/extensions/bootstrap/reverse_tunnel/common/rping_interceptor.cc

@@ -0,0 +1,76 @@
+#include "source/extensions/bootstrap/reverse_tunnel/common/rping_interceptor.h"
+
+#include "source/extensions/bootstrap/reverse_tunnel/common/reverse_connection_utility.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Bootstrap {
+namespace ReverseConnection {
+
+Api::IoCallUint64Result RpingInterceptor::read(Buffer::Instance& buffer,
+                                               absl::optional<uint64_t> max_length) {
+  // Perform the actual read first.
+  Api::IoCallUint64Result result = IoSocketHandleImpl::read(buffer, max_length);
+  ENVOY_LOG(trace, "RpingInterceptor: read result: {}", result.return_value_);
+
+  // If RPING keepalives are still active, check whether the incoming data is a RPING message.
+  if (ping_echo_active_ && result.err_ == nullptr && result.return_value_ > 0) {
+    const uint64_t expected = ReverseConnectionUtility::PING_MESSAGE.size();
+
+    // Compare up to the expected size using a zero-copy view.
+    const uint64_t len = std::min<uint64_t>(buffer.length(), expected);
+    const char* data = static_cast<const char*>(buffer.linearize(len));
+    absl::string_view peek_sv{data, static_cast<size_t>(len)};
+
+    // Check if we have a complete RPING message.
+    if (len == expected && ReverseConnectionUtility::isPingMessage(peek_sv)) {
+      // Found a complete RPING. Echo and drain it from the buffer.
+      buffer.drain(expected);
+      onPingMessage();
+
+      // If buffer only contained RPING, return showing we processed it.
+      if (buffer.length() == 0) {
+        return Api::IoCallUint64Result{expected, Api::IoError::none()};
+      }
+
+      // RPING followed by application data. Disable echo and return the remaining data.
+      ENVOY_LOG(trace,
+                "RpingInterceptor: received application data after RPING, "
+                "disabling RPING echo for FD: {}",
+                fd_);
+      ping_echo_active_ = false;
+      // The adjusted return value is the number of bytes excluding the drained RPING. It should be
+      // transparent to upper layers that the RPING was processed.
+      const uint64_t adjusted =
+          (result.return_value_ >= expected) ? (result.return_value_ - expected) : 0;
+      return Api::IoCallUint64Result{adjusted, Api::IoError::none()};
+    }
+
+    // If partial data could be the start of RPING (only when fewer than expected bytes).
+    if (len < expected) {
+      const absl::string_view rping_prefix =
+          ReverseConnectionUtility::PING_MESSAGE.substr(0, static_cast<size_t>(len));
+      if (peek_sv == rping_prefix) {
+        ENVOY_LOG(trace,
+                  "RpingInterceptor: partial RPING received ({} bytes), waiting "
+                  "for more.",
+                  len);
+        return result; // Wait for more data.
+      }
+    }
+
+    // Data is not RPING (complete or partial). Disable echo permanently.
+    ENVOY_LOG(trace,
+              "RpingInterceptor: received application data ({} bytes), "
+              "disabling RPING echo for FD: {}",
+              len, fd_);
+    ping_echo_active_ = false;
+  }
+
+  return result;
+}
+
+} // namespace ReverseConnection
+} // namespace Bootstrap
+} // namespace Extensions
+} // namespace Envoy

source/extensions/bootstrap/reverse_tunnel/common/rping_interceptor.h

@@ -0,0 +1,27 @@
+#pragma once
+
+#include "source/common/network/io_socket_handle_impl.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Bootstrap {
+namespace ReverseConnection {
+
+class RpingInterceptor : public virtual Network::IoSocketHandleImpl {
+public:
+  // Intercept reads to handle reverse connection keep-alive pings.
+  Api::IoCallUint64Result read(Buffer::Instance& buffer,
+                               absl::optional<uint64_t> max_length) override;
+
+  virtual void onPingMessage() PURE;
+
+protected:
+  // Whether to actively echo RPING messages while the connection is idle.
+  // Disabled permanently after the first non-RPING application byte is observed.
+  bool ping_echo_active_{true};
+};
+
+} // namespace ReverseConnection
+} // namespace Bootstrap
+} // namespace Extensions
+} // namespace Envoy

source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/BUILD

@@ -89,6 +89,7 @@ envoy_cc_library(
         "//source/common/tls:ssl_handshaker_lib",
         "//source/common/upstream:load_balancer_context_base_lib",
         "//source/extensions/bootstrap/reverse_tunnel/common:reverse_connection_utility_lib",
+        "//source/extensions/bootstrap/reverse_tunnel/common:rping_interceptor_lib",
     ],
 )
 

source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/downstream_reverse_connection_io_handle.cc

@@ -1,7 +1,6 @@
 #include "source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/downstream_reverse_connection_io_handle.h"
 
 #include "source/common/common/logger.h"
-#include "source/extensions/bootstrap/reverse_tunnel/common/reverse_connection_utility.h"
 #include "source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/reverse_connection_io_handle.h"
 
 #include "absl/strings/string_view.h"
@@ -31,79 +30,15 @@ DownstreamReverseConnectionIOHandle::~DownstreamReverseConnectionIOHandle() {
       fd_, connection_key_);
 }
 
-Api::IoCallUint64Result
-DownstreamReverseConnectionIOHandle::read(Buffer::Instance& buffer,
-                                          absl::optional<uint64_t> max_length) {
-  // Perform the actual read first.
-  Api::IoCallUint64Result result = IoSocketHandleImpl::read(buffer, max_length);
-  ENVOY_LOG(trace, "DownstreamReverseConnectionIOHandle: read result: {}", result.return_value_);
-
-  // If RPING keepalives are still active, check whether the incoming data is a RPING message.
-  if (ping_echo_active_ && result.err_ == nullptr && result.return_value_ > 0) {
-    const uint64_t expected =
-        ::Envoy::Extensions::Bootstrap::ReverseConnection::ReverseConnectionUtility::PING_MESSAGE
-            .size();
-
-    // Compare up to the expected size using a zero-copy view.
-    const uint64_t len = std::min<uint64_t>(buffer.length(), expected);
-    const char* data = static_cast<const char*>(buffer.linearize(len));
-    absl::string_view peek_sv{data, static_cast<size_t>(len)};
-
-    // Check if we have a complete RPING message.
-    if (len == expected &&
-        ::Envoy::Extensions::Bootstrap::ReverseConnection::ReverseConnectionUtility::isPingMessage(
-            peek_sv)) {
-      // Found a complete RPING. Echo and drain it from the buffer.
-      buffer.drain(expected);
-      auto echo_rc = ::Envoy::Extensions::Bootstrap::ReverseConnection::ReverseConnectionUtility::
-          sendPingResponse(*this);
-      if (!echo_rc.ok()) {
-        ENVOY_LOG(trace, "DownstreamReverseConnectionIOHandle: failed to send RPING echo on FD: {}",
-                  fd_);
-      } else {
-        ENVOY_LOG(trace, "DownstreamReverseConnectionIOHandle: echoed RPING on FD: {}", fd_);
-      }
-
-      // If buffer only contained RPING, return showing we processed it.
-      if (buffer.length() == 0) {
-        return Api::IoCallUint64Result{expected, Api::IoError::none()};
-      }
-
-      // RPING followed by application data. Disable echo and return the remaining data.
-      ENVOY_LOG(trace,
-                "DownstreamReverseConnectionIOHandle: received application data after RPING, "
-                "disabling RPING echo for FD: {}",
-                fd_);
-      ping_echo_active_ = false;
-      // The adjusted return value is the number of bytes excluding the drained RPING. It should be
-      // transparent to upper layers that the RPING was processed.
-      const uint64_t adjusted =
-          (result.return_value_ >= expected) ? (result.return_value_ - expected) : 0;
-      return Api::IoCallUint64Result{adjusted, Api::IoError::none()};
-    }
-
-    // If partial data could be the start of RPING (only when fewer than expected bytes).
-    if (len < expected) {
-      const absl::string_view rping_prefix =
-          ReverseConnectionUtility::PING_MESSAGE.substr(0, static_cast<size_t>(len));
-      if (peek_sv == rping_prefix) {
-        ENVOY_LOG(trace,
-                  "DownstreamReverseConnectionIOHandle: partial RPING received ({} bytes), waiting "
-                  "for more.",
-                  len);
-        return result; // Wait for more data.
-      }
-    }
-
-    // Data is not RPING (complete or partial). Disable echo permanently.
-    ENVOY_LOG(trace,
-              "DownstreamReverseConnectionIOHandle: received application data ({} bytes), "
-              "disabling RPING echo for FD: {}",
-              len, fd_);
-    ping_echo_active_ = false;
-  }
+void DownstreamReverseConnectionIOHandle::onPingMessage() {
+  auto echo_rc = ReverseConnectionUtility::sendPingResponse(*this);
 
-  return result;
+  if (!echo_rc.ok()) {
+    ENVOY_LOG(trace, "DownstreamReverseConnectionIOHandle: failed to send RPING echo on FD: {}",
+              fd_);
+  } else {
+    ENVOY_LOG(trace, "DownstreamReverseConnectionIOHandle: echoed RPING on FD: {}", fd_);
+  }
 }
 
 // DownstreamReverseConnectionIOHandle close() implementation.

source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/downstream_reverse_connection_io_handle.h

@@ -7,6 +7,8 @@
 
 #include "source/common/common/logger.h"
 #include "source/common/network/io_socket_handle_impl.h"
+#include "source/extensions/bootstrap/reverse_tunnel/common/reverse_connection_utility.h"
+#include "source/extensions/bootstrap/reverse_tunnel/common/rping_interceptor.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -21,7 +23,7 @@ class ReverseConnectionIOHandle;
  * This class is used internally by ReverseConnectionIOHandle to manage the lifecycle
  * of accepted downstream connections.
  */
-class DownstreamReverseConnectionIOHandle : public Network::IoSocketHandleImpl {
+class DownstreamReverseConnectionIOHandle : public RpingInterceptor {
 public:
   /**
    * Constructor that takes ownership of the socket and stores parent pointer and connection key.
@@ -33,12 +35,13 @@ class DownstreamReverseConnectionIOHandle : public Network::IoSocketHandleImpl {
   ~DownstreamReverseConnectionIOHandle() override;
 
   // Network::IoHandle overrides.
-  // Intercept reads to handle reverse connection keep-alive pings.
-  Api::IoCallUint64Result read(Buffer::Instance& buffer,
-                               absl::optional<uint64_t> max_length) override;
   Api::IoCallUint64Result close() override;
   Api::SysCallIntResult shutdown(int how) override;
 
+  // RPING Interceptor overrides.
+  // Send the RPING response from here.
+  void onPingMessage() override;
+
   /**
    * Tell this IO handle to ignore close() and shutdown() calls.
    * This is called by the HTTP filter during socket hand-off to prevent
@@ -60,10 +63,6 @@ class DownstreamReverseConnectionIOHandle : public Network::IoSocketHandleImpl {
   std::string connection_key_;
   // Flag to ignore close and shutdown calls during socket hand-off.
   bool ignore_close_and_shutdown_{false};
-
-  // Whether to actively echo RPING messages while the connection is idle.
-  // Disabled permanently after the first non-RPING application byte is observed.
-  bool ping_echo_active_{true};
 };
 
 } // namespace ReverseConnection

source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface/BUILD

@@ -29,6 +29,7 @@ envoy_cc_extension(
         "//source/common/config:utility_lib",
         "//source/common/network:default_socket_interface_lib",
         "//source/extensions/bootstrap/reverse_tunnel/common:reverse_connection_utility_lib",
+        "//source/extensions/bootstrap/reverse_tunnel/common:rping_interceptor_lib",
         "@envoy_api//envoy/extensions/bootstrap/reverse_tunnel/upstream_socket_interface/v3:pkg_cc_proto",
     ],
 )

source/extensions/bootstrap/reverse_tunnel/upstream_socket_interface/reverse_connection_io_handle.h

@@ -6,6 +6,7 @@
 #include "envoy/network/socket.h"
 
 #include "source/common/network/io_socket_handle_impl.h"
+#include "source/extensions/bootstrap/reverse_tunnel/common/rping_interceptor.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -17,7 +18,7 @@ namespace ReverseConnection {
  * This class implements RAII principles to ensure proper socket cleanup and provides
  * reverse connection semantics where the connection is already established.
  */
-class UpstreamReverseConnectionIOHandle : public Network::IoSocketHandleImpl {
+class UpstreamReverseConnectionIOHandle : public RpingInterceptor {
 public:
   /**
    * Constructs an UpstreamReverseConnectionIOHandle that takes ownership of a socket.
@@ -71,6 +72,10 @@ class UpstreamReverseConnectionIOHandle : public Network::IoSocketHandleImpl {
    */
   void releaseSocketForTest() { owned_socket_.reset(); }
 
+  // Ignore all ping messages.
+  // The connection is passed on to the http2 codec.
+  void onPingMessage() override {}
+
 private:
   // The name of the cluster this reverse connection belongs to.
   std::string cluster_name_;

test/extensions/bootstrap/reverse_tunnel/common/BUILD

@@ -20,3 +20,14 @@ envoy_cc_test(
         "//test/test_common:test_runtime_lib",
     ],
 )
+
+envoy_cc_test(
+    name = "rping_interceptor_test",
+    size = "medium",
+    srcs = ["rping_interceptor_test.cc"],
+    deps = [
+        "//source/common/buffer:buffer_lib",
+        "//source/extensions/bootstrap/reverse_tunnel/common:reverse_connection_utility_lib",
+        "//source/extensions/bootstrap/reverse_tunnel/common:rping_interceptor_lib",
+    ],
+)