[filters] Prevent a filter from sending local reply and continue

source/common/http/filter_manager.cc

@@ -447,14 +447,18 @@ void FilterManager::decodeHeaders(ActiveStreamDecoderFilter* filter, RequestHead
     ASSERT(!(status == FilterHeadersStatus::ContinueAndDontEndStream && !(*entry)->end_stream_),
            "Filters should not return FilterHeadersStatus::ContinueAndDontEndStream from "
            "decodeHeaders when end_stream is already false");
+    ENVOY_BUG(
+        !(state_.local_complete_) || status == FilterHeadersStatus::StopIteration,
+        "Filters should return FilterHeadersStatus::StopIteration after sending a local reply.");
 
     state_.filter_call_state_ &= ~FilterCallState::DecodeHeaders;
     ENVOY_STREAM_LOG(trace, "decode headers called: filter={} status={}", *this,
                      static_cast<const void*>((*entry).get()), static_cast<uint64_t>(status));
 
     (*entry)->decode_headers_called_ = true;
 
-    const auto continue_iteration = (*entry)->commonHandleAfterHeadersCallback(status, end_stream);
+    const auto continue_iteration =
+        (*entry)->commonHandleAfterHeadersCallback(status, end_stream) && !state_.local_complete_;
 
     // If this filter ended the stream, decodeComplete() should be called for it.
     if ((*entry)->end_stream_) {

source/extensions/filters/http/oauth2/filter.cc

@@ -212,12 +212,12 @@ Http::FilterHeadersStatus OAuth2Filter::decodeHeaders(Http::RequestHeaderMap& he
       Http::Utility::Url state_url;
       if (!state_url.initialize(state, false)) {
         sendUnauthorizedResponse();
-        return Http::FilterHeadersStatus::StopAllIterationAndBuffer;
+        return Http::FilterHeadersStatus::StopIteration;
       }
       // Avoid infinite redirect storm
       if (config_->redirectPathMatcher().match(state_url.pathAndQueryParams())) {
         sendUnauthorizedResponse();
-        return Http::FilterHeadersStatus::StopAllIterationAndBuffer;
+        return Http::FilterHeadersStatus::StopIteration;
       }
       Http::ResponseHeaderMapPtr response_headers{
           Http::createHeaderMap<Http::ResponseHeaderMapImpl>(
@@ -283,7 +283,7 @@ Http::FilterHeadersStatus OAuth2Filter::decodeHeaders(Http::RequestHeaderMap& he
 
     config_->stats().oauth_unauthorized_rq_.inc();
 
-    return Http::FilterHeadersStatus::StopAllIterationAndBuffer;
+    return Http::FilterHeadersStatus::StopIteration;
   }
 
   // At this point, we *are* on /_oauth. We believe this request comes from the authorization
@@ -292,14 +292,14 @@ Http::FilterHeadersStatus OAuth2Filter::decodeHeaders(Http::RequestHeaderMap& he
   const auto query_parameters = Http::Utility::parseQueryString(path_str);
   if (query_parameters.find(queryParamsError()) != query_parameters.end()) {
     sendUnauthorizedResponse();
-    return Http::FilterHeadersStatus::StopAllIterationAndBuffer;
+    return Http::FilterHeadersStatus::StopIteration;
   }
 
   // if the data we need is not present on the URL, stop execution
   if (query_parameters.find(queryParamsCode()) == query_parameters.end() ||
       query_parameters.find(queryParamsState()) == query_parameters.end()) {
     sendUnauthorizedResponse();
-    return Http::FilterHeadersStatus::StopAllIterationAndBuffer;
+    return Http::FilterHeadersStatus::StopIteration;
   }
 
   auth_code_ = query_parameters.at(queryParamsCode());
@@ -308,7 +308,7 @@ Http::FilterHeadersStatus OAuth2Filter::decodeHeaders(Http::RequestHeaderMap& he
   Http::Utility::Url state_url;
   if (!state_url.initialize(state_, false)) {
     sendUnauthorizedResponse();
-    return Http::FilterHeadersStatus::StopAllIterationAndBuffer;
+    return Http::FilterHeadersStatus::StopIteration;
   }
 
   Formatter::FormatterImpl formatter(config_->redirectUri());
@@ -319,7 +319,7 @@ Http::FilterHeadersStatus OAuth2Filter::decodeHeaders(Http::RequestHeaderMap& he
                                      redirect_uri);
 
   // pause while we await the next step from the OAuth server
-  return Http::FilterHeadersStatus::StopAllIterationAndBuffer;
+  return Http::FilterHeadersStatus::StopIteration;
 }
 
 // Defines a sequence of checks determining whether we should initiate a new OAuth flow or skip to
@@ -359,7 +359,7 @@ Http::FilterHeadersStatus OAuth2Filter::signOutUser(const Http::RequestHeaderMap
   response_headers->setLocation(new_path);
   decoder_callbacks_->encodeHeaders(std::move(response_headers), true, SIGN_OUT);
 
-  return Http::FilterHeadersStatus::StopAllIterationAndBuffer;
+  return Http::FilterHeadersStatus::StopIteration;
 }
 
 void OAuth2Filter::onGetAccessTokenSuccess(const std::string& access_code,

test/common/http/conn_manager_impl_test_2.cc

@@ -1453,7 +1453,7 @@ TEST_F(HttpConnectionManagerImplTest, FilterHeadReply) {
       .WillOnce(InvokeWithoutArgs([&]() -> FilterHeadersStatus {
         decoder_filters_[0]->callbacks_->sendLocalReply(Code::BadRequest, "Bad request", nullptr,
                                                         absl::nullopt, "");
-        return FilterHeadersStatus::Continue;
+        return FilterHeadersStatus::StopIteration;
       }));
 
   EXPECT_CALL(response_encoder_, streamErrorOnInvalidHttpMessage()).WillOnce(Return(true));
@@ -1483,7 +1483,7 @@ TEST_F(HttpConnectionManagerImplTest, ResetWithStoppedFilter) {
       .WillOnce(InvokeWithoutArgs([&]() -> FilterHeadersStatus {
         decoder_filters_[0]->callbacks_->sendLocalReply(Code::BadRequest, "Bad request", nullptr,
                                                         absl::nullopt, "");
-        return FilterHeadersStatus::Continue;
+        return FilterHeadersStatus::StopIteration;
       }));
 
   EXPECT_CALL(response_encoder_, streamErrorOnInvalidHttpMessage()).WillOnce(Return(true));

test/extensions/filters/http/oauth2/filter_test.cc

@@ -174,7 +174,7 @@ TEST_F(OAuth2Test, RequestSignout) {
   };
   EXPECT_CALL(decoder_callbacks_, encodeHeaders_(HeaderMapEqualRef(&response_headers), true));
 
-  EXPECT_EQ(Http::FilterHeadersStatus::StopAllIterationAndBuffer,
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
             filter_->decodeHeaders(request_headers, false));
 }
 
@@ -255,7 +255,7 @@ TEST_F(OAuth2Test, OAuthErrorNonOAuthHttpCallback) {
 
   EXPECT_CALL(decoder_callbacks_, encodeHeaders_(HeaderMapEqualRef(&response_headers), true));
 
-  EXPECT_EQ(Http::FilterHeadersStatus::StopAllIterationAndBuffer,
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
             filter_->decodeHeaders(request_headers, false));
 }
 
@@ -281,7 +281,7 @@ TEST_F(OAuth2Test, OAuthErrorQueryString) {
   EXPECT_CALL(decoder_callbacks_, encodeHeaders_(HeaderMapEqualRef(&response_headers), false));
   EXPECT_CALL(decoder_callbacks_, encodeData(_, true));
 
-  EXPECT_EQ(Http::FilterHeadersStatus::StopAllIterationAndBuffer,
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
             filter_->decodeHeaders(request_headers, false));
 
   EXPECT_EQ(scope_.counterFromString("test.oauth_failure").value(), 1);
@@ -309,7 +309,7 @@ TEST_F(OAuth2Test, OAuthCallbackStartsAuthentication) {
   EXPECT_CALL(*oauth_client_, asyncGetAccessToken("123", TEST_CLIENT_ID, "asdf_client_secret_fdsa",
                                                   "https://traffic.example.com" + TEST_CALLBACK));
 
-  EXPECT_EQ(Http::FilterHeadersStatus::StopAllIterationAndBuffer,
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
             filter_->decodeHeaders(request_headers, false));
 }
 
@@ -421,7 +421,7 @@ TEST_F(OAuth2Test, OAuthTestInvalidUrlInStateQueryParam) {
   EXPECT_CALL(*validator_, token()).WillRepeatedly(ReturnRef(legit_token));
 
   EXPECT_CALL(decoder_callbacks_, encodeHeaders_(HeaderMapEqualRef(&expected_headers), false));
-  EXPECT_EQ(Http::FilterHeadersStatus::StopAllIterationAndBuffer,
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
             filter_->decodeHeaders(request_headers, false));
 }
 
@@ -455,7 +455,7 @@ TEST_F(OAuth2Test, OAuthTestCallbackUrlInStateQueryParam) {
 
   EXPECT_CALL(decoder_callbacks_,
               encodeHeaders_(HeaderMapEqualRef(&expected_response_headers), false));
-  EXPECT_EQ(Http::FilterHeadersStatus::StopAllIterationAndBuffer,
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
             filter_->decodeHeaders(request_headers, false));
 
   Http::TestRequestHeaderMapImpl final_request_headers{
@@ -564,7 +564,7 @@ TEST_F(OAuth2Test, OAuthTestFullFlowPostWithParameters) {
   EXPECT_CALL(decoder_callbacks_, encodeHeaders_(HeaderMapEqualRef(&first_response_headers), true));
 
   // This represents the beginning of the OAuth filter.
-  EXPECT_EQ(Http::FilterHeadersStatus::StopAllIterationAndBuffer,
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
             filter_->decodeHeaders(first_request_headers, false));
 
   // This represents the callback request from the authorization server.
@@ -584,7 +584,7 @@ TEST_F(OAuth2Test, OAuthTestFullFlowPostWithParameters) {
                                                   "https://traffic.example.com" + TEST_CALLBACK));
 
   // Invoke the callback logic. As a side effect, state_ will be populated.
-  EXPECT_EQ(Http::FilterHeadersStatus::StopAllIterationAndBuffer,
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
             filter_->decodeHeaders(second_request_headers, false));
 
   EXPECT_EQ(1, config_->stats().oauth_unauthorized_rq_.value());

test/integration/BUILD

@@ -437,11 +437,13 @@ envoy_cc_test(
         "//source/extensions/filters/http/buffer:config",
         "//source/extensions/filters/http/health_check:config",
         "//test/common/http/http2:http2_frame",
+        "//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_filter_lib",
         "//test/integration/filters:random_pause_filter_lib",
+        "//test/test_common:logging_lib",
         "//test/test_common:utility_lib",
         "@envoy_api//envoy/config/bootstrap/v3:pkg_cc_proto",
         "@envoy_api//envoy/config/route/v3:pkg_cc_proto",

test/integration/filters/BUILD

@@ -39,6 +39,21 @@ envoy_cc_test_library(
     ],
 )
 
+envoy_cc_test_library(
+    name = "continue_after_local_reply_filter_lib",
+    srcs = [
+        "continue_after_local_reply_filter.cc",
+    ],
+    deps = [
+        ":common_lib",
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/registry",
+        "//include/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 = "continue_headers_only_inject_body",
     srcs = [

test/integration/filters/continue_after_local_reply_filter.cc

@@ -0,0 +1,30 @@
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+
+#include "extensions/filters/http/common/pass_through_filter.h"
+
+#include "test/extensions/filters/http/common/empty_http_filter_config.h"
+#include "test/integration/filters/common.h"
+
+#include "gtest/gtest.h"
+
+namespace Envoy {
+
+// A filter that only calls Http::FilterHeadersStatus::Continue after a local reply.
+class ContinueAfterLocalReplyFilter : public Http::PassThroughFilter {
+public:
+  constexpr static char name[] = "continue-after-local-reply-filter";
+
+  Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap&, bool) override {
+    decoder_callbacks_->sendLocalReply(Envoy::Http::Code::OK, "", nullptr, absl::nullopt,
+                                       "ContinueAfterLocalReplyFilter is ready");
+    return Http::FilterHeadersStatus::Continue;
+  }
+};
+
+constexpr char ContinueAfterLocalReplyFilter::name[];
+static Registry::RegisterFactory<SimpleFilterConfig<ContinueAfterLocalReplyFilter>,
+                                 Server::Configuration::NamedHttpFilterConfigFactory>
+    register_;
+
+} // namespace Envoy

test/integration/protocol_integration_test.cc

@@ -34,6 +34,7 @@
 #include "test/mocks/upstream/retry_priority.h"
 #include "test/mocks/upstream/retry_priority_factory.h"
 #include "test/test_common/environment.h"
+#include "test/test_common/logging.h"
 #include "test/test_common/network_utility.h"
 #include "test/test_common/registry.h"
 
@@ -287,6 +288,34 @@ TEST_P(ProtocolIntegrationTest, ContinueHeadersOnlyInjectBodyFilter) {
   EXPECT_EQ(response->body(), "body");
 }
 
+// Tests a filter that returns a FilterHeadersStatus::Continue after a local reply. In debug mode,
+// this fails on ENVOY_BUG. In opt mode, the status is corrected and the failure is logged.
+TEST_P(ProtocolIntegrationTest, ContinueAfterLocalReply) {
+#ifdef NDEBUG
+  config_helper_.addFilter(R"EOF(
+  name: continue-after-local-reply-filter
+  typed_config:
+    "@type": type.googleapis.com/google.protobuf.Empty
+  )EOF");
+  initialize();
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  // Send a headers only request.
+  IntegrationStreamDecoderPtr response;
+  EXPECT_LOG_CONTAINS("error",
+                      "envoy bug failure: !(state_.local_complete_) || status == "
+                      "FilterHeadersStatus::StopIteration. Details: Filters should return "
+                      "FilterHeadersStatus::StopIteration after sending a local reply.",
+                      {
+                        response = codec_client_->makeHeaderOnlyRequest(default_request_headers_);
+                        response->waitForEndStream();
+                      });
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().getStatusValue());
+#endif
+}
+
 TEST_P(ProtocolIntegrationTest, AddEncodedTrailers) {
   config_helper_.addFilter(R"EOF(
 name: add-trailers-filter