CVE-2021-43825

Response filter manager crash

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

source/common/http/conn_manager_impl.cc

@@ -205,8 +205,8 @@ void ConnectionManagerImpl::doEndStream(ActiveStream& stream) {
   // 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);
@@ -249,7 +249,7 @@ void ConnectionManagerImpl::doEndStream(ActiveStream& stream) {
   // 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));
 }
@@ -1432,7 +1432,7 @@ 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;
     }

source/common/http/filter_manager.cc

@@ -303,6 +303,12 @@ bool ActiveStreamDecoderFilter::canContinue() {
   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(); },
@@ -316,7 +322,7 @@ 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);
@@ -832,8 +838,8 @@ void FilterManager::decodeMetadata(ActiveStreamDecoderFilter* filter, MetadataMa
 }
 
 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);
@@ -1356,6 +1362,8 @@ void FilterManager::encodeTrailers(ActiveStreamEncoderFilter* filter,
 
 void FilterManager::maybeEndEncode(bool end_stream) {
   if (end_stream) {
+    ASSERT(!state_.remote_encode_complete_);
+    state_.remote_encode_complete_ = true;
     filter_manager_callbacks_.endStream();
   }
 }
@@ -1646,6 +1654,7 @@ Http1StreamEncoderOptionsOptRef ActiveStreamEncoderFilter::http1StreamEncoderOpt
 }
 
 void ActiveStreamEncoderFilter::responseDataTooLarge() {
+  ENVOY_STREAM_LOG(debug, "response data too large watermark exceeded", parent_);
   if (parent_.state_.encoder_filters_streaming_) {
     onEncoderFilterAboveWriteBufferHighWatermark();
   } else {

source/common/http/filter_manager.h

@@ -317,7 +317,7 @@ struct ActiveStreamEncoderFilter : public ActiveStreamFilterBase,
       : 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;
@@ -907,7 +907,7 @@ class FilterManager : public ScopeTrackedObject,
   /**
    * 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
@@ -1058,15 +1058,15 @@ class FilterManager : public ScopeTrackedObject,
 
   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.

test/integration/BUILD

@@ -517,6 +517,7 @@ envoy_cc_test_library(
         "//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/BUILD

@@ -545,6 +545,20 @@ envoy_cc_test_library(
     ],
 )
 
+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/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

test/integration/protocol_integration_test.cc

@@ -3450,6 +3450,58 @@ TEST_P(ProtocolIntegrationTest, FragmentStrippedFromPathWithOverride) {
   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"));