hcm: Ensure operations are not called on deleted stream decoders (envoyproxy#39346)

Previously, it was possible for the `ActiveStream` in the
`HttpConnectionManager` to get deleted while still trying to process
packets from the codec.

This change uses the `ActiveStreamHandle` with weak pointer semantics to
ensure that even with incoming data packets, methods are not called on a
deleted `ActiveStream`, which represents the HCM's `RequestDecoder`.

---------

Signed-off-by: Ali Beyad <[email protected]>

Author: abeyad

changelogs/current.yaml

@@ -49,6 +49,10 @@ bug_fixes:
     draining. This can result in not enough connections being established for current pending requests. This is most problematic for
     long-lived requests (such as streaming gRPC requests or long-poll requests) because a connection could be in the draining state
     for a long time.
+- area: hcm
+  change: |
+    Fixes a bug where the lifetime of the HttpConnectionManager's ActiveStream can be out of sync
+    with the lifetime of the codec stream.
 
 removed_config_or_runtime:
 # *Normally occurs at the end of the* :ref:`deprecation period <deprecated>`

envoy/http/api_listener.h

@@ -5,17 +5,6 @@
 namespace Envoy {
 namespace Http {
 
-class RequestDecoderHandle {
-public:
-  virtual ~RequestDecoderHandle() = default;
-
-  /**
-   * @return a reference to the underlying decoder if it is still valid.
-   */
-  virtual OptRef<RequestDecoder> get() PURE;
-};
-using RequestDecoderHandlePtr = std::unique_ptr<RequestDecoderHandle>;
-
 /**
  * ApiListener that allows consumers to interact with HTTP streams via API calls.
  */

envoy/http/codec.h

@@ -7,6 +7,7 @@
 #include "envoy/access_log/access_log.h"
 #include "envoy/buffer/buffer.h"
 #include "envoy/common/matchers.h"
+#include "envoy/common/optref.h"
 #include "envoy/common/pure.h"
 #include "envoy/grpc/status.h"
 #include "envoy/http/header_formatter.h"
@@ -53,6 +54,17 @@ const char MaxResponseHeadersSizeOverrideKey[] =
 class Stream;
 class RequestDecoder;
 
+class RequestDecoderHandle {
+public:
+  virtual ~RequestDecoderHandle() = default;
+
+  /**
+   * @return a reference to the underlying decoder if it is still valid.
+   */
+  virtual OptRef<RequestDecoder> get() PURE;
+};
+using RequestDecoderHandlePtr = std::unique_ptr<RequestDecoderHandle>;
+
 /**
  * Error codes used to convey the reason for a GOAWAY.
  */
@@ -264,6 +276,12 @@ class RequestDecoder : public virtual StreamDecoder {
    * @return List of shared pointers to access loggers for this stream.
    */
   virtual AccessLog::InstanceSharedPtrVector accessLogHandlers() PURE;
+
+  /**
+   * @return A handle to the request decoder. Caller can check the request decoder's liveness via
+   * the handle.
+   */
+  virtual RequestDecoderHandlePtr getRequestDecoderHandle() PURE;
 };
 
 /**

source/common/http/conn_manager_impl.h

@@ -207,6 +207,11 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
                                    filter_log_handlers.end());
       return combined_log_handlers;
     }
+
+    RequestDecoderHandlePtr getRequestDecoderHandle() override {
+      return std::make_unique<ActiveStreamHandle>(*this);
+    }
+
     // Hand off headers/trailers and stream info to the codec's response encoder, for logging later
     // (i.e. possibly after this stream has been destroyed).
     //

source/common/http/http2/codec_impl.cc

@@ -498,6 +498,7 @@ void ConnectionImpl::StreamImpl::decodeData() {
         stream_manager_.decodeAsChunks() &&
         pending_recv_data_->length() > stream_manager_.defer_processing_segment_size_;
 
+    StreamDecoder* stream_decoder = decoder();
     if (decode_data_in_chunk) {
       Buffer::OwnedImpl chunk_buffer;
       // TODO(kbaichoo): Consider implementing an approximate move for chunking.
@@ -508,15 +509,19 @@ void ConnectionImpl::StreamImpl::decodeData() {
       stream_manager_.body_buffered_ = true;
       ASSERT(pending_recv_data_->length() > 0);
 
-      decoder().decodeData(chunk_buffer, sendEndStream());
+      if (stream_decoder) {
+        stream_decoder->decodeData(chunk_buffer, sendEndStream());
+      }
       already_drained_data = true;
 
       if (!buffersOverrun()) {
         scheduleProcessingOfBufferedData(true);
       }
     } else {
       // Send the entire buffer through.
-      decoder().decodeData(*pending_recv_data_, sendEndStream());
+      if (stream_decoder) {
+        stream_decoder->decodeData(*pending_recv_data_, sendEndStream());
+      }
     }
   }
 
@@ -597,7 +602,11 @@ void ConnectionImpl::ServerStreamImpl::decodeHeaders() {
     Http::Utility::transformUpgradeRequestFromH2toH1(*headers);
   }
 #endif
-  request_decoder_->decodeHeaders(std::move(headers), sendEndStream());
+  RequestDecoder* request_decoder = request_decoder_handle_->get().ptr();
+  ENVOY_BUG(request_decoder != nullptr, "Missing request_decoder_");
+  if (request_decoder) {
+    request_decoder->decodeHeaders(std::move(headers), sendEndStream());
+  }
 }
 
 void ConnectionImpl::ServerStreamImpl::decodeTrailers() {
@@ -608,8 +617,12 @@ void ConnectionImpl::ServerStreamImpl::decodeTrailers() {
   // Consume any buffered trailers.
   stream_manager_.trailers_buffered_ = false;
 
-  request_decoder_->decodeTrailers(
-      std::move(absl::get<RequestTrailerMapPtr>(headers_or_trailers_)));
+  RequestDecoder* request_decoder = request_decoder_handle_->get().ptr();
+  ENVOY_BUG(request_decoder != nullptr, "Missing request_decoder_");
+  if (request_decoder) {
+    request_decoder->decodeTrailers(
+        std::move(absl::get<RequestTrailerMapPtr>(headers_or_trailers_)));
+  }
 }
 
 void ConnectionImpl::StreamImpl::pendingSendBufferHighWatermark() {
@@ -827,7 +840,10 @@ void ConnectionImpl::StreamImpl::onMetadataDecoded(MetadataMapPtr&& metadata_map
     ENVOY_CONN_LOG(debug, "decode metadata called with empty map, skipping", parent_.connection_);
     parent_.stats_.metadata_empty_frames_.inc();
   } else {
-    decoder().decodeMetadata(std::move(metadata_map_ptr));
+    StreamDecoder* stream_decoder = decoder();
+    if (stream_decoder) {
+      stream_decoder->decodeMetadata(std::move(metadata_map_ptr));
+    }
   }
 }
 

source/common/http/http2/codec_impl.h

@@ -324,7 +324,7 @@ class ConnectionImpl : public virtual Connection,
     // sendPendingFrames so pending outbound frames have one final chance to be flushed. If we
     // submit a reset, nghttp2 will cancel outbound frames that have not yet been sent.
     virtual bool useDeferredReset() const PURE;
-    virtual StreamDecoder& decoder() PURE;
+    virtual StreamDecoder* decoder() PURE;
     virtual HeaderMap& headers() PURE;
     virtual void allocTrailers() PURE;
     virtual HeaderMapPtr cloneTrailers(const HeaderMap& trailers) PURE;
@@ -523,7 +523,7 @@ class ConnectionImpl : public virtual Connection,
     HeadersState headersState() const override { return headers_state_; }
     // Do not use deferred reset on upstream connections.
     bool useDeferredReset() const override { return false; }
-    StreamDecoder& decoder() override { return response_decoder_; }
+    StreamDecoder* decoder() override { return &response_decoder_; }
     void decodeHeaders() override;
     void decodeTrailers() override;
     HeaderMap& headers() override {
@@ -585,7 +585,7 @@ class ConnectionImpl : public virtual Connection,
     // written out before force resetting the stream, assuming there is enough H2 connection flow
     // control window is available.
     bool useDeferredReset() const override { return true; }
-    StreamDecoder& decoder() override { return *request_decoder_; }
+    StreamDecoder* decoder() override { return request_decoder_handle_->get().ptr(); }
     void decodeHeaders() override;
     void decodeTrailers() override;
     HeaderMap& headers() override {
@@ -610,7 +610,9 @@ class ConnectionImpl : public virtual Connection,
     void encodeTrailers(const ResponseTrailerMap& trailers) override {
       encodeTrailersBase(trailers);
     }
-    void setRequestDecoder(Http::RequestDecoder& decoder) override { request_decoder_ = &decoder; }
+    void setRequestDecoder(Http::RequestDecoder& decoder) override {
+      request_decoder_handle_ = decoder.getRequestDecoderHandle();
+    }
     void setDeferredLoggingHeadersAndTrailers(Http::RequestHeaderMapConstSharedPtr,
                                               Http::ResponseHeaderMapConstSharedPtr,
                                               Http::ResponseTrailerMapConstSharedPtr,
@@ -626,7 +628,7 @@ class ConnectionImpl : public virtual Connection,
     }
 
   private:
-    RequestDecoder* request_decoder_{};
+    RequestDecoderHandlePtr request_decoder_handle_;
     HeadersState headers_state_ = HeadersState::Request;
   };
 

source/common/quic/envoy_quic_server_stream.cc

@@ -4,6 +4,7 @@
 #include <openssl/evp.h>
 
 #include <memory>
+#include <utility>
 
 #include "source/common/buffer/buffer_impl.h"
 #include "source/common/common/assert.h"
@@ -214,7 +215,10 @@ void EnvoyQuicServerStream::OnInitialHeadersComplete(bool fin, size_t frame_len,
   }
 #endif
 
-  request_decoder_->decodeHeaders(std::move(headers), /*end_stream=*/fin);
+  Http::RequestDecoder* decoder = request_decoder_->get().ptr();
+  if (decoder != nullptr) {
+    decoder->decodeHeaders(std::move(headers), /*end_stream=*/fin);
+  }
   ConsumeHeaderList();
 }
 
@@ -262,7 +266,10 @@ void EnvoyQuicServerStream::OnBodyAvailable() {
       // A stream error has occurred, stop processing.
       return;
     }
-    request_decoder_->decodeData(*buffer, fin_read_and_no_trailers);
+    Http::RequestDecoder* decoder = request_decoder_->get().ptr();
+    if (decoder != nullptr) {
+      decoder->decodeData(*buffer, fin_read_and_no_trailers);
+    }
   }
 
   if (!sequencer()->IsClosed() || read_side_closed()) {
@@ -314,7 +321,10 @@ void EnvoyQuicServerStream::maybeDecodeTrailers() {
       onStreamError(close_connection_upon_invalid_header_, rst);
       return;
     }
-    request_decoder_->decodeTrailers(std::move(trailers));
+    Http::RequestDecoder* decoder = request_decoder_->get().ptr();
+    if (decoder != nullptr) {
+      decoder->decodeTrailers(std::move(trailers));
+    }
     MarkTrailersConsumed();
   }
 }
@@ -499,7 +509,10 @@ void EnvoyQuicServerStream::OnMetadataComplete(size_t /*frame_len*/,
     return;
   }
   if (!header_list.empty()) {
-    request_decoder_->decodeMetadata(metadataMapFromHeaderList(header_list));
+    Http::RequestDecoder* decoder = request_decoder_->get().ptr();
+    if (decoder != nullptr) {
+      decoder->decodeMetadata(metadataMapFromHeaderList(header_list));
+    }
   }
 }
 
@@ -542,8 +555,8 @@ bool EnvoyQuicServerStream::hasPendingData() {
 #ifdef ENVOY_ENABLE_HTTP_DATAGRAMS
 void EnvoyQuicServerStream::useCapsuleProtocol() {
   http_datagram_handler_ = std::make_unique<HttpDatagramHandler>(*this);
-  ASSERT(request_decoder_ != nullptr);
-  http_datagram_handler_->setStreamDecoder(request_decoder_);
+  ASSERT(request_decoder_->get().has_value());
+  http_datagram_handler_->setStreamDecoder(request_decoder_->get().ptr());
   RegisterHttp3DatagramVisitor(http_datagram_handler_.get());
 }
 #endif

source/common/quic/envoy_quic_server_stream.h

@@ -27,8 +27,8 @@ class EnvoyQuicServerStream : public quic::QuicSpdyServerStreamBase,
                             headers_with_underscores_action);
 
   void setRequestDecoder(Http::RequestDecoder& decoder) override {
-    request_decoder_ = &decoder;
-    stats_gatherer_->setAccessLogHandlers(request_decoder_->accessLogHandlers());
+    request_decoder_ = decoder.getRequestDecoderHandle();
+    stats_gatherer_->setAccessLogHandlers(request_decoder_->get()->accessLogHandlers());
   }
 
   // Http::StreamEncoder
@@ -122,7 +122,7 @@ class EnvoyQuicServerStream : public quic::QuicSpdyServerStreamBase,
   void useCapsuleProtocol();
 #endif
 
-  Http::RequestDecoder* request_decoder_{nullptr};
+  Http::RequestDecoderHandlePtr request_decoder_;
   envoy::config::core::v3::HttpProtocolOptions::HeadersWithUnderscoresAction
       headers_with_underscores_action_;
 

source/common/quic/http_datagram_handler.cc

@@ -18,10 +18,11 @@ void HttpDatagramHandler::decodeCapsule(const quiche::Capsule& capsule) {
   quiche::QuicheBuffer serialized_capsule = SerializeCapsule(capsule, &capsule_buffer_allocator_);
   Buffer::InstancePtr buffer = std::make_unique<Buffer::OwnedImpl>();
   buffer->add(serialized_capsule.AsStringView());
-  if (!stream_decoder_) {
+  if (stream_decoder_) {
+    stream_decoder_->decodeData(*buffer, stream_.IsDoneReading());
+  } else {
     IS_ENVOY_BUG("HTTP/3 Datagram received before a stream decoder is set.");
   }
-  stream_decoder_->decodeData(*buffer, stream_.IsDoneReading());
 }
 
 void HttpDatagramHandler::OnHttp3Datagram(quic::QuicStreamId stream_id, absl::string_view payload) {

test/common/http/http2/codec_impl_test.cc

@@ -248,7 +248,7 @@ class Http2CodecImplTestFixture {
     server_wrapper_ = std::make_unique<ConnectionWrapper>(server_.get());
     createHeaderValidator();
     request_encoder_ = &client_->newStream(response_decoder_);
-    setupDefaultConnectionMocks();
+    setupDefaultMocks();
     driveToCompletion();
 
     EXPECT_CALL(server_callbacks_, newStream(_, _))
@@ -267,7 +267,16 @@ class Http2CodecImplTestFixture {
         .WillByDefault(Return(true));
   }
 
-  void setupDefaultConnectionMocks() {
+  void setupRequestDecoderMock(MockRequestDecoder& request_decoder) {
+    EXPECT_CALL(request_decoder, getRequestDecoderHandle())
+        .WillRepeatedly(Invoke([&request_decoder]() {
+          auto handle = std::make_unique<NiceMock<MockRequestDecoderHandle>>();
+          ON_CALL(*handle, get()).WillByDefault(Return(OptRef<RequestDecoder>(request_decoder)));
+          return handle;
+        }));
+  }
+
+  void setupDefaultMocks() {
     ON_CALL(client_connection_, write(_, _))
         .WillByDefault(Invoke([&](Buffer::Instance& data, bool) -> void {
           if (corrupt_metadata_frame_) {
@@ -280,6 +289,7 @@ class Http2CodecImplTestFixture {
             [&](Buffer::Instance& data, bool) -> void { client_wrapper_->buffer_.add(data); }));
     // Set to the small read size (reads are suggested to be 16k aligned).
     ON_CALL(server_connection_, bufferLimit()).WillByDefault(Return(16 * 1024));
+    setupRequestDecoderMock(request_decoder_);
   }
 
   void http2OptionsFromTuple(envoy::config::core::v3::Http2ProtocolOptions& options,
@@ -1895,6 +1905,7 @@ TEST_P(Http2CodecImplFlowControlTest, TestFlowControlInPendingSendData) {
   StreamEncoder* response_encoder2;
   MockStreamCallbacks server_stream_callbacks2;
   MockRequestDecoder request_decoder2;
+  setupRequestDecoderMock(request_decoder2);
   // When the server stream is created it should check the status of the
   // underlying connection. Pretend it is overrun.
   EXPECT_CALL(server_connection_, aboveHighWatermark()).WillOnce(Return(true));
@@ -2505,7 +2516,7 @@ TEST_P(Http2CodecImplStreamLimitTest, MaxClientStreams) {
       server_http2_options_, random_, max_request_headers_kb_, max_request_headers_count_,
       headers_with_underscores_action_);
   server_wrapper_ = std::make_unique<ConnectionWrapper>(server_.get());
-  setupDefaultConnectionMocks();
+  setupDefaultMocks();
   driveToCompletion();
   for (int i = 0; i < 101; ++i) {
     request_encoder_ = &client_->newStream(response_decoder_);
@@ -4351,6 +4362,7 @@ TEST_P(Http2CodecImplTest, CheckHeaderPaddedWhitespaceValidation) {
   MockStreamCallbacks server_stream_callbacks;
   MockRequestDecoder request_decoder;
 
+  setupRequestDecoderMock(request_decoder);
   EXPECT_CALL(server_callbacks_, newStream(_, _))
       .WillOnce(Invoke([&](ResponseEncoder& encoder, bool) -> RequestDecoder& {
         response_encoder = &encoder;
@@ -4437,6 +4449,8 @@ TEST_P(Http2CodecImplTest, CheckHeaderValueValidation) {
 
   scoped_runtime_.mergeValues({{"envoy.reloadable_features.validate_upstream_headers", "false"}});
   stream_error_on_invalid_http_messaging_ = true;
+
+  setupRequestDecoderMock(request_decoder_);
   initialize();
 
 #ifdef ENVOY_ENABLE_UHV
@@ -4471,6 +4485,7 @@ TEST_P(Http2CodecImplTest, CheckHeaderValueValidation) {
     MockStreamCallbacks server_stream_callbacks;
     MockRequestDecoder request_decoder;
 
+    setupRequestDecoderMock(request_decoder);
     EXPECT_CALL(server_callbacks_, newStream(_, _))
         .WillOnce(Invoke([&](ResponseEncoder& encoder, bool) -> RequestDecoder& {
           response_encoder = &encoder;
@@ -4591,7 +4606,7 @@ class Http2CodecMetadataTest : public Http2CodecImplTestFixture, public ::testin
         server_http2_options_, random_, max_request_headers_kb_, max_request_headers_count_,
         headers_with_underscores_action_);
     server_wrapper_ = std::make_unique<ConnectionWrapper>(server_.get());
-    setupDefaultConnectionMocks();
+    setupDefaultMocks();
     driveToCompletion();
   }