CVE-2021-43825: Use After Free (CWE-416)
value: Envoy is an open source edge and service proxy, designed for cloud-native applications. Sending a locally generated response must stop further processing of request or response data. Envoy tracks the amount of buffered request and response data and aborts the request if the amount of buffered data is over the limit by sending 413 or 500 responses. However when the buffer overflows while response is processed by the filter chain the operation may not be aborted correctly and result in accessing a freed memory block. If this happens Envoy will crash resulting in a denial of service.
CWE description: The product reuses or references memory after it has been freed. At some point afterward, the memory may be allocated again and saved in another pointer, while the original pointer references a location somewhere within the new allocation. Any operations using the original pointer are no longer valid because the memory "belongs" to the code that operates on the new pointer.
CVSS v3 base score: 7.5
Published: 2022-02-22T23:15Z
Repository: https://github.com/envoyproxy/envoy
Language: C++
Files changed: source/common/http/conn_manager_impl.cc, source/common/http/filter_manager.cc, source/common/http/filter_manager.h, test/integration/BUILD, test/integration/filters/BUILD, test/integration/filters/buffer_continue_filter.cc, test/integration/protocol_integration_test.cc
Commit date: 2022-02-22 17:58:22 +0000
Commit message: CVE-2021-43825

Response filter manager crash

Signed-off-by: Yan Avlasov <yavlasov@google.com>

Diff: diff --git a/source/common/http/conn_manager_impl.cc b/source/common/http/conn_manager_impl.cc
index 3a76359545..0f824751f6 100644
--- a/source/common/http/conn_manager_impl.cc
+++ b/source/common/http/conn_manager_impl.cc
@@ -203,12 +203,12 @@ void ConnectionManagerImpl::doEndStream(ActiveStream& stream) {
   bool reset_stream = false;
   // If the response encoder is still associated with the stream, reset the stream. The exception
   // here is when Envoy "ends" the stream by calling recreateStream at which point recreateStream
   // explicitly nulls out response_encoder to avoid the downstream being notified of the
   // Envoy-internal stream instance being ended.
-  if (stream.response_encoder_ != nullptr &&
-      (!stream.filter_manager_.remoteComplete() || !stream.state_.codec_saw_local_complete_)) {
+  if (stream.response_encoder_ != nullptr && (!stream.filter_manager_.remoteDecodeComplete() ||
+                                              !stream.state_.codec_saw_local_complete_)) {
     // Indicate local is complete at this point so that if we reset during a continuation, we don't
     // raise further data or trailers.
     ENVOY_STREAM_LOG(debug, "doEndStream() resetting stream", stream);
     // TODO(snowp): This call might not be necessary, try to clean up + remove setter function.
     stream.filter_manager_.setLocalComplete();
@@ -247,11 +247,11 @@ void ConnectionManagerImpl::doEndStream(ActiveStream& stream) {
   bool http_10_sans_cl = (codec_->protocol() == Protocol::Http10) &&
                          (!stream.response_headers_ || !stream.response_headers_->ContentLength());
   // We also don't delay-close in the case of HTTP/1.1 where the request is
   // fully read, as there's no race condition to avoid.
   bool connection_close = stream.state_.saw_connection_close_;
-  bool request_complete = stream.filter_manager_.remoteComplete();
+  bool request_complete = stream.filter_manager_.remoteDecodeComplete();
 
   checkForDeferredClose(connection_close && (request_complete || http_10_sans_cl));
 }
 
 void ConnectionManagerImpl::doDeferredStreamDestroy(ActiveStream& stream) {
@@ -1430,11 +1430,11 @@ void ConnectionManagerImpl::ActiveStream::encodeHeaders(ResponseHeaderMap& heade
   }
 
   // If we are destroying a stream before remote is complete and the connection does not support
   // multiplexing, we should disconnect since we don't want to wait around for the request to
   // finish.
-  if (!filter_manager_.remoteComplete()) {
+  if (!filter_manager_.remoteDecodeComplete()) {
     if (connection_manager_.codec_->protocol() < Protocol::Http2) {
       connection_manager_.drain_state_ = DrainState::Closing;
     }
 
     connection_manager_.stats_.named_.downstream_rq_response_before_rq_complete_.inc();
diff --git a/source/common/http/filter_manager.cc b/source/common/http/filter_manager.cc
index 7dd385416d..ea5c18daca 100644
--- a/source/common/http/filter_manager.cc
+++ b/source/common/http/filter_manager.cc
@@ -301,10 +301,16 @@ bool ActiveStreamDecoderFilter::canContinue() {
   // last data frame comes in and the filter continues, but the final buffering takes the stream
   // over the high watermark such that a 413 is returned.
   return !parent_.state_.local_complete_;
 }
 
+bool ActiveStreamEncoderFilter::canContinue() {
+  // As with ActiveStreamDecoderFilter::canContinue() make sure we do not
+  // continue if a local reply has been sent.
+  return !parent_.state_.remote_encode_complete_;
+}
+
 Buffer::InstancePtr ActiveStreamDecoderFilter::createBuffer() {
   auto buffer = dispatcher().getWatermarkFactory().createBuffer(
       [this]() -> void { this->requestDataDrained(); },
       [this]() -> void { this->requestDataTooLarge(); },
       []() -> void { /* TODO(adisuissa): Handle overflow watermark */ });
@@ -314,11 +320,11 @@ Buffer::InstancePtr ActiveStreamDecoderFilter::createBuffer() {
 
 Buffer::InstancePtr& ActiveStreamDecoderFilter::bufferedData() {
   return parent_.buffered_request_data_;
 }
 
-bool ActiveStreamDecoderFilter::complete() { return parent_.state_.remote_complete_; }
+bool ActiveStreamDecoderFilter::complete() { return parent_.state_.remote_decode_complete_; }
 
 void ActiveStreamDecoderFilter::doHeaders(bool end_stream) {
   parent_.decodeHeaders(this, *parent_.filter_manager_callbacks_.requestHeaders(), end_stream);
 }
 
@@ -830,12 +836,12 @@ void FilterManager::decodeMetadata(ActiveStreamDecoderFilter* filter, MetadataMa
                      metadata_map);
   }
 }
 
 void FilterManager::maybeEndDecode(bool end_stream) {
-  ASSERT(!state_.remote_complete_);
-  state_.remote_complete_ = end_stream;
+  ASSERT(!state_.remote_decode_complete_);
+  state_.remote_decode_complete_ = end_stream;
   if (end_stream) {
     stream_info_.downstreamTiming().onLastDownstreamRxByteReceived(dispatcher().timeSource());
     ENVOY_STREAM_LOG(debug, "request end stream", *this);
   }
 }
@@ -1354,10 +1360,12 @@ void FilterManager::encodeTrailers(ActiveStreamEncoderFilter* filter,
   maybeEndEncode(true);
 }
 
 void FilterManager::maybeEndEncode(bool end_stream) {
   if (end_stream) {
+    ASSERT(!state_.remote_encode_complete_);
+    state_.remote_encode_complete_ = true;
     filter_manager_callbacks_.endStream();
   }
 }
 
 bool FilterManager::processNewlyAddedMetadata() {
@@ -1644,10 +1652,11 @@ Http1StreamEncoderOptionsOptRef ActiveStreamEncoderFilter::http1StreamEncoderOpt
   // we give the filter direct access to the encoder options.
   return parent_.filter_manager_callbacks_.http1StreamEncoderOptions();
 }
 
 void ActiveStreamEncoderFilter::responseDataTooLarge() {
+  ENVOY_STREAM_LOG(debug, "response data too large watermark exceeded", parent_);
   if (parent_.state_.encoder_filters_streaming_) {
     onEncoderFilterAboveWriteBufferHighWatermark();
   } else {
     parent_.filter_manager_callbacks_.onResponseDataTooLarge();
 
diff --git a/source/common/http/filter_manager.h b/source/common/http/filter_manager.h
index ff3c02ab85..d9c2a3d2ad 100644
--- a/source/common/http/filter_manager.h
+++ b/source/common/http/filter_manager.h
@@ -315,11 +315,11 @@ struct ActiveStreamEncoderFilter : public ActiveStreamFilterBase,
   ActiveStreamEncoderFilter(FilterManager& parent, StreamEncoderFilterSharedPtr filter,
                             FilterMatchStateSharedPtr match_state, bool dual_filter)
       : ActiveStreamFilterBase(parent, dual_filter, std::move(match_state)), handle_(filter) {}
 
   // ActiveStreamFilterBase
-  bool canContinue() override { return true; }
+  bool canContinue() override;
   Buffer::InstancePtr createBuffer() override;
   Buffer::InstancePtr& bufferedData() override;
   bool complete() override;
   bool has1xxheaders() override;
   void do1xxHeaders() override;
@@ -905,11 +905,11 @@ public:
   bool destroyed() const { return state_.destroyed_; }
 
   /**
    * Whether remote processing has been marked as complete.
    */
-  bool remoteComplete() const { return state_.remote_complete_; }
+  bool remoteDecodeComplete() const { return state_.remote_decode_complete_; }
 
   /**
    * Instructs the FilterManager to not create a filter chain. This makes it possible to issue
    * a local reply without the overhead of creating and traversing the filters.
    */
@@ -1056,19 +1056,19 @@ private:
     };
   // clang-format on
 
   struct State {
     State()
-        : remote_complete_(false), local_complete_(false), has_1xx_headers_(false),
-          created_filter_chain_(false), is_head_request_(false), is_grpc_request_(false),
-          non_100_response_headers_encoded_(false), under_on_local_reply_(false),
-          decoder_filter_chain_aborted_(false), encoder_filter_chain_aborted_(false),
-          saw_downstream_reset_(false) {}
-
+        : remote_encode_complete_(false), remote_decode_complete_(false), local_complete_(false),
+          has_1xx_headers_(false), created_filter_chain_(false), is_head_request_(false),
+          is_grpc_request_(false), non_100_response_headers_encoded_(false),
+          under_on_local_reply_(false), decoder_filter_chain_aborted_(false),
+          encoder_filter_chain_aborted_(false), saw_downstream_reset_(false) {}
     uint32_t filter_call_state_{0};
 
-    bool remote_complete_ : 1;
+    bool remote_encode_complete_ : 1;
+    bool remote_decode_complete_ : 1;
     bool local_complete_ : 1; // This indicates that local is complete prior to filter processing.
                               // A filter can still stop the stream from being complete as seen
                               // by the codec.
     // By default, we will assume there are no 1xx. If encode1xxHeaders
     // is ever called, this is set to true so commonContinue resumes processing the 1xx.
diff --git a/test/integration/BUILD b/test/integration/BUILD
index 990e17b2d0..d93a55ba7d 100644
--- a/test/integration/BUILD
+++ b/test/integration/BUILD
@@ -515,10 +515,11 @@ envoy_cc_test_library(
         ":socket_interface_swap_lib",
         "//source/common/http:header_map_lib",
         "//source/extensions/filters/http/buffer:config",
         "//test/common/http/http2:http2_frame",
         "//test/integration/filters:add_invalid_data_filter_lib",
+        "//test/integration/filters:buffer_continue_filter_lib",
         "//test/integration/filters:continue_after_local_reply_filter_lib",
         "//test/integration/filters:continue_headers_only_inject_body",
         "//test/integration/filters:encoder_decoder_buffer_filter_lib",
         "//test/integration/filters:invalid_header_filter_lib",
         "//test/integration/filters:local_reply_during_encoding_data_filter_lib",
diff --git a/test/integration/filters/BUILD b/test/integration/filters/BUILD
index 7a39648ae1..e29fd05bd6 100644
--- a/test/integration/filters/BUILD
+++ b/test/integration/filters/BUILD
@@ -543,10 +543,24 @@ envoy_cc_test_library(
         "//source/extensions/filters/http/common:pass_through_filter_lib",
         "//test/extensions/filters/http/common:empty_http_filter_config_lib",
     ],
 )
 
+envoy_cc_test_library(
+    name = "buffer_continue_filter_lib",
+    srcs = [
+        "buffer_continue_filter.cc",
+    ],
+    deps = [
+        "//envoy/http:filter_interface",
+        "//envoy/registry",
+        "//envoy/server:filter_config_interface",
+        "//source/extensions/filters/http/common:pass_through_filter_lib",
+        "//test/extensions/filters/http/common:empty_http_filter_config_lib",
+    ],
+)
+
 envoy_cc_test_library(
     name = "test_socket_interface_lib",
     srcs = [
         "test_socket_interface.cc",
     ],
diff --git a/test/integration/filters/buffer_continue_filter.cc b/test/integration/filters/buffer_continue_filter.cc
new file mode 100644
index 0000000000..d5f5dadd69
--- /dev/null
+++ b/test/integration/filters/buffer_continue_filter.cc
@@ -0,0 +1,74 @@
+#include <string>
+
+#include "envoy/http/filter.h"
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+
+#include "source/extensions/filters/http/common/pass_through_filter.h"
+
+#include "test/extensions/filters/http/common/empty_http_filter_config.h"
+
+namespace Envoy {
+
+// A filter that buffers until the limit is reached and then continues.
+class BufferContinueStreamFilter : public Http::PassThroughFilter {
+public:
+  Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap&, bool end_stream) override {
+    return end_stream ? Http::FilterHeadersStatus::Continue
+                      : Http::FilterHeadersStatus::StopIteration;
+  }
+
+  Http::FilterDataStatus decodeData(Buffer::Instance&, bool end_stream) override {
+    return end_stream ? Http::FilterDataStatus::Continue
+                      : Http::FilterDataStatus::StopIterationAndBuffer;
+  }
+
+  Http::FilterHeadersStatus encodeHeaders(Http::ResponseHeaderMap& headers, bool) override {
+    response_headers_ = &headers;
+    return Http::FilterHeadersStatus::StopIteration;
+  }
+
+  Http::FilterDataStatus encodeData(Buffer::Instance& data, bool end_stream) override {
+    data_total_ += data.length();
+
+    const auto limit = encoder_callbacks_->encoderBufferLimit();
+    const auto header_size = response_headers_->byteSize();
+
+    if (limit && header_size + data_total_ > limit) {
+      // Give up since we've reached the buffer limit, Envoy should generate
+      // a 500 since it couldn't finished encoding.
+      return Http::FilterDataStatus::Continue;
+    }
+
+    encoder_callbacks_->addEncodedData(data, false);
+
+    if (!end_stream) {
+      return Http::FilterDataStatus::StopIterationAndBuffer;
+    }
+
+    return Http::FilterDataStatus::Continue;
+  }
+
+private:
+  Http::ResponseHeaderMap* response_headers_;
+  uint64_t data_total_{0};
+};
+
+class BufferContinueFilterConfig : public Extensions::HttpFilters::Common::EmptyHttpFilterConfig {
+public:
+  BufferContinueFilterConfig() : EmptyHttpFilterConfig("buffer-continue-filter") {}
+
+  Http::FilterFactoryCb createFilter(const std::string&,
+                                     Server::Configuration::FactoryContext&) override {
+    return [](Http::FilterChainFactoryCallbacks& callbacks) -> void {
+      callbacks.addStreamFilter(std::make_shared<::Envoy::BufferContinueStreamFilter>());
+    };
+  }
+};
+
+// perform static registration
+static Registry::RegisterFactory<BufferContinueFilterConfig,
+                                 Server::Configuration::NamedHttpFilterConfigFactory>
+    register_;
+
+} // namespace Envoy
diff --git a/test/integration/protocol_integration_test.cc b/test/integration/protocol_integration_test.cc
index 289ae403ca..33a65d23b4 100644
--- a/test/integration/protocol_integration_test.cc
+++ b/test/integration/protocol_integration_test.cc
@@ -3448,10 +3448,62 @@ TEST_P(ProtocolIntegrationTest, FragmentStrippedFromPathWithOverride) {
   EXPECT_TRUE(upstream_request_->complete());
   ASSERT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().getStatusValue());
 }
 
+// Test buffering and then continuing after too many response bytes to buffer.
+TEST_P(ProtocolIntegrationTest, BufferContinue) {
+  // Bytes sent is configured for http/2 flow control windows.
+  if (upstreamProtocol() != Http::CodecType::HTTP2) {
+    return;
+  }
+  config_helper_.addConfigModifier(
+      [&](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
+              hcm) -> void {
+        auto* route_config = hcm.mutable_route_config();
+        auto* virtual_host = route_config->mutable_virtual_hosts(0);
+        auto* header = virtual_host->mutable_response_headers_to_add()->Add()->mutable_header();
+        header->set_key("foo");
+        header->set_value("bar");
+      });
+
+  useAccessLog();
+  config_helper_.addFilter("{ name: buffer-continue-filter, typed_config: { \"@type\": "
+                           "type.googleapis.com/google.protobuf.Empty } }");
+  config_helper_.setBufferLimits(1024, 1024);
+  initialize();
+
+  // Send the request.
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  auto encoder_decoder = codec_client_->startRequest(default_request_headers_);
+  auto downstream_request = &encoder_decoder.first;
+  auto response = std::move(encoder_decoder.second);
+  Buffer::OwnedImpl data("HTTP body content goes here");
+  codec_client_->sendData(*downstream_request, data, true);
+  waitForNextUpstreamRequest();
+
+  // Send the response headers.
+  upstream_request_->encodeHeaders(default_response_headers_, false);
+
+  // Now send an overly large response body. At some point, too much data will
+  // be buffered, the stream will be reset, and the connection will disconnect.
+  upstream_request_->encodeData(512, false);
+  upstream_request_->encodeData(1024 * 100, false);
+
+  if (upstreamProtocol() == Http::CodecType::HTTP1) {
+    ASSERT_TRUE(fake_upstream_connection_->waitForDisconnect());
+  } else {
+    ASSERT_TRUE(upstream_request_->waitForReset());
+    ASSERT_TRUE(fake_upstream_connection_->close());
+    ASSERT_TRUE(fake_upstream_connection_->waitForDisconnect());
+  }
+
+  ASSERT_TRUE(response->waitForEndStream());
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("500", response->headers().getStatusValue());
+}
+
 TEST_P(DownstreamProtocolIntegrationTest, ContentLengthSmallerThanPayload) {
   initialize();
   codec_client_ = makeHttpConnection(lookupPort("http"));
   auto response =
       codec_client_->makeRequestWithBody(Http::TestRequestHeaderMapImpl{{":method", "POST"},

Vulnerable code:   if (stream.response_encoder_ != nullptr &&
      (!stream.filter_manager_.remoteComplete() || !stream.state_.codec_saw_local_complete_)) {
  bool request_complete = stream.filter_manager_.remoteComplete();
  if (!filter_manager_.remoteComplete()) {
bool ActiveStreamDecoderFilter::complete() { return parent_.state_.remote_complete_; }
  ASSERT(!state_.remote_complete_);
  state_.remote_complete_ = end_stream;
  bool canContinue() override { return true; }
  bool remoteComplete() const { return state_.remote_complete_; }
        : remote_complete_(false), local_complete_(false), has_1xx_headers_(false),
          created_filter_chain_(false), is_head_request_(false), is_grpc_request_(false),
          non_100_response_headers_encoded_(false), under_on_local_reply_(false),
          decoder_filter_chain_aborted_(false), encoder_filter_chain_aborted_(false),
          saw_downstream_reset_(false) {}

    bool remote_complete_ : 1;
Fixed code:   if (stream.response_encoder_ != nullptr && (!stream.filter_manager_.remoteDecodeComplete() ||
                                              !stream.state_.codec_saw_local_complete_)) {
  bool request_complete = stream.filter_manager_.remoteDecodeComplete();
  if (!filter_manager_.remoteDecodeComplete()) {
bool ActiveStreamEncoderFilter::canContinue() {
  // As with ActiveStreamDecoderFilter::canContinue() make sure we do not
  // continue if a local reply has been sent.
  return !parent_.state_.remote_encode_complete_;
}

bool ActiveStreamDecoderFilter::complete() { return parent_.state_.remote_decode_complete_; }
  ASSERT(!state_.remote_decode_complete_);
  state_.remote_decode_complete_ = end_stream;
    ASSERT(!state_.remote_encode_complete_);
    state_.remote_encode_complete_ = true;
  ENVOY_STREAM_LOG(debug, "response data too large watermark exceeded", parent_);
  bool canContinue() override;
  bool remoteDecodeComplete() const { return state_.remote_decode_complete_; }
        : remote_encode_complete_(false), remote_decode_complete_(false), local_complete_(false),
          has_1xx_headers_(false), created_filter_chain_(false), is_head_request_(false),
          is_grpc_request_(false), non_100_response_headers_encoded_(false),
          under_on_local_reply_(false), decoder_filter_chain_aborted_(false),
          encoder_filter_chain_aborted_(false), saw_downstream_reset_(false) {}
    bool remote_encode_complete_ : 1;
    bool remote_decode_complete_ : 1;
        "//test/integration/filters:buffer_continue_filter_lib",
envoy_cc_test_library(
    name = "buffer_continue_filter_lib",
    srcs = [
        "buffer_continue_filter.cc",
    ],
    deps = [
        "//envoy/http:filter_interface",
        "//envoy/registry",
        "//envoy/server:filter_config_interface",
        "//source/extensions/filters/http/common:pass_through_filter_lib",
        "//test/extensions/filters/http/common:empty_http_filter_config_lib",
    ],
)

#include <string>

#include "envoy/http/filter.h"
#include "envoy/registry/registry.h"
#include "envoy/server/filter_config.h"

#include "source/extensions/filters/http/common/pass_through_filter.h"

#include "test/extensions/filters/http/common/empty_http_filter_config.h"

namespace Envoy {

// A filter that buffers until the limit is reached and then continues.
class BufferContinueStreamFilter : public Http::PassThroughFilter {
public:
  Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap&, bool end_stream) override {
    return end_stream ? Http::FilterHeadersStatus::Continue
                      : Http::FilterHeadersStatus::StopIteration;
  }

  Http::FilterDataStatus decodeData(Buffer::Instance&, bool end_stream) override {
    return end_stream ? Http::FilterDataStatus::Continue
                      : Http::FilterDataStatus::StopIterationAndBuffer;
  }

  Http::FilterHeadersStatus encodeHeaders(Http::ResponseHeaderMap& headers, bool) override {
    response_headers_ = &headers;
    return Http::FilterHeadersStatus::StopIteration;
  }

  Http::FilterDataStatus encodeData(Buffer::Instance& data, bool end_stream) override {
    data_total_ += data.length();

    const auto limit = encoder_callbacks_->encoderBufferLimit();
    const auto header_size = response_headers_->byteSize();

    if (limit && header_size + data_total_ > limit) {
      // Give up since we've reached the buffer limit, Envoy should generate
      // a 500 since it couldn't finished encoding.
      return Http::FilterDataStatus::Continue;
    }

    encoder_callbacks_->addEncodedData(data, false);

    if (!end_stream) {
      return Http::FilterDataStatus::StopIterationAndBuffer;
    }

    return Http::FilterDataStatus::Continue;
  }

private:
  Http::ResponseHeaderMap* response_headers_;
  uint64_t data_total_{0};
};

class BufferContinueFilterConfig : public Extensions::HttpFilters::Common::EmptyHttpFilterConfig {
public:
  BufferContinueFilterConfig() : EmptyHttpFilterConfig("buffer-continue-filter") {}

  Http::FilterFactoryCb createFilter(const std::string&,
                                     Server::Configuration::FactoryContext&) override {
    return [](Http::FilterChainFactoryCallbacks& callbacks) -> void {
      callbacks.addStreamFilter(std::make_shared<::Envoy::BufferContinueStreamFilter>());
    };
  }
};

// perform static registration
static Registry::RegisterFactory<BufferContinueFilterConfig,
                                 Server::Configuration::NamedHttpFilterConfigFactory>
    register_;

} // namespace Envoy
// Test buffering and then continuing after too many response bytes to buffer.
TEST_P(ProtocolIntegrationTest, BufferContinue) {
  // Bytes sent is configured for http/2 flow control windows.
  if (upstreamProtocol() != Http::CodecType::HTTP2) {
    return;
  }
  config_helper_.addConfigModifier(
      [&](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
              hcm) -> void {
        auto* route_config = hcm.mutable_route_config();
        auto* virtual_host = route_config->mutable_virtual_hosts(0);
        auto* header = virtual_host->mutable_response_headers_to_add()->Add()->mutable_header();
        header->set_key("foo");
        header->set_value("bar");
      });

  useAccessLog();
  config_helper_.addFilter("{ name: buffer-continue-filter, typed_config: { \"@type\": "
                           "type.googleapis.com/google.protobuf.Empty } }");
  config_helper_.setBufferLimits(1024, 1024);
  initialize();

  // Send the request.
  codec_client_ = makeHttpConnection(lookupPort("http"));
  auto encoder_decoder = codec_client_->startRequest(default_request_headers_);
  auto downstream_request = &encoder_decoder.first;
  auto response = std::move(encoder_decoder.second);
  Buffer::OwnedImpl data("HTTP body content goes here");
  codec_client_->sendData(*downstream_request, data, true);
  waitForNextUpstreamRequest();

  // Send the response headers.
  upstream_request_->encodeHeaders(default_response_headers_, false);

  // Now send an overly large response body. At some point, too much data will
  // be buffered, the stream will be reset, and the connection will disconnect.
  upstream_request_->encodeData(512, false);
  upstream_request_->encodeData(1024 * 100, false);

  if (upstreamProtocol() == Http::CodecType::HTTP1) {
    ASSERT_TRUE(fake_upstream_connection_->waitForDisconnect());
  } else {
    ASSERT_TRUE(upstream_request_->waitForReset());
    ASSERT_TRUE(fake_upstream_connection_->close());
    ASSERT_TRUE(fake_upstream_connection_->waitForDisconnect());
  }

  ASSERT_TRUE(response->waitForEndStream());
  EXPECT_TRUE(response->complete());
  EXPECT_EQ("500", response->headers().getStatusValue());
}