contrib/golang: address race condition in Go HTTP filter

contrib/golang/filters/http/source/go/pkg/http/shim.go

@@ -157,6 +157,9 @@ func getRequest(r *C.httpRequest) *httpRequest {
 func getState(s *C.processState) *processState {
 	r := s.req
 	req := getRequest(r)
+	if req == nil {
+		return nil
+	}
 	if s.is_encoding == 0 {
 		return &req.decodingState.processState
 	}
@@ -237,6 +240,10 @@ func envoyGoFilterOnHttpHeader(s *C.processState, endStream, headerNum, headerBy
 //export envoyGoFilterOnHttpData
 func envoyGoFilterOnHttpData(s *C.processState, endStream, buffer, length uint64) uint64 {
 	state := getState(s)
+	if state == nil {
+		// safe to do as the C++ side hasDestroyed() check prevents acting on the returned value
+		return uint64(api.Continue)
+	}
 
 	req := state.request
 	if req.pInfo.paniced {
@@ -389,6 +396,9 @@ func envoyGoFilterOnHttpDestroy(r *C.httpRequest, reason uint64) {
 //export envoyGoRequestSemaDec
 func envoyGoRequestSemaDec(r *C.httpRequest) {
 	req := getRequest(r)
+	if req == nil {
+		return
+	}
 	defer req.recoverPanic()
 	req.resumeWaitCallback()
 }

contrib/golang/filters/http/source/golang_filter.cc

@@ -404,12 +404,19 @@ void Filter::continueStatusInternal(ProcessorState& state, GolangStatus status)
 
     case FilterState::ProcessingTrailer:
       state.continueDoData();
+      if (hasDestroyed()) {
+        return;
+      }
       state.continueProcessing();
       break;
 
     default:
       ASSERT(0, "unexpected state");
     }
+
+    if (hasDestroyed()) {
+      return;
+    }
   }
 
   ENVOY_LOG(debug,
@@ -426,6 +433,9 @@ void Filter::continueStatusInternal(ProcessorState& state, GolangStatus status)
     auto done = doDataGo(state, state.getBufferData(), state.getEndStream());
     if (done) {
       state.continueDoData();
+      if (hasDestroyed()) {
+        return;
+      }
     } else {
       // do not process trailers when data is not finished
       return;

contrib/golang/filters/http/source/golang_filter.h

@@ -348,6 +348,8 @@ class Filter : public Http::StreamFilter,
   void deferredDeleteRequest(HttpRequestInternal* req);
 
 private:
+  friend class TestFilter;
+
   bool hasDestroyed() {
     Thread::LockGuard lock(mutex_);
     return has_destroyed_;

contrib/golang/filters/http/test/BUILD

@@ -33,7 +33,9 @@ envoy_cc_test(
     ],
     rbe_pool = "6gig",
     deps = [
+        "//contrib/golang/common/dso/test:dso_mocks",
         "//contrib/golang/filters/http/source:golang_filter_lib",
+        "//source/common/network:address_lib",
         "//source/common/stream_info:stream_info_lib",
         "//test/mocks/api:api_mocks",
         "//test/mocks/http:http_mocks",

contrib/golang/filters/http/test/golang_filter_test.cc

@@ -5,6 +5,7 @@
 
 #include "source/common/buffer/buffer_impl.h"
 #include "source/common/http/message_impl.h"
+#include "source/common/network/address_impl.h"
 #include "source/common/stream_info/stream_info_impl.h"
 
 #include "test/common/stats/stat_test_utility.h"
@@ -21,25 +22,32 @@
 #include "test/test_common/utility.h"
 
 #include "absl/strings/str_format.h"
+#include "contrib/golang/common/dso/test/mocks.h"
 #include "contrib/golang/filters/http/source/golang_filter.h"
 #include "gmock/gmock.h"
 
 using testing::_;
 using testing::AtLeast;
 using testing::InSequence;
 using testing::Invoke;
+using testing::Return;
+using testing::SaveArg;
 
 namespace Envoy {
 namespace Extensions {
 namespace HttpFilters {
 namespace Golang {
-namespace {
 
 class TestFilter : public Filter {
 public:
+  using Filter::continueStatusInternal;
   using Filter::Filter;
+  DecodingProcessorState& testDecodingState() { return decoding_state_; }
+  HttpRequestInternal* testReq() { return req_; }
 };
 
+namespace {
+
 class GolangHttpFilterTest : public testing::Test {
 public:
   GolangHttpFilterTest() {
@@ -191,6 +199,72 @@ TEST_F(GolangHttpFilterTest, InvalidConfigForRouteConfigFilter) {
                           "golang filter failed to parse plugin config");
 }
 
+// Regression test for https://github.com/envoyproxy/envoy/issues/44320.
+TEST_F(GolangHttpFilterTest, BufferedDataAfterDestroyDuringContinue) {
+  auto dso_lib = std::make_shared<NiceMock<Dso::MockHttpFilterDsoImpl>>();
+  ON_CALL(*dso_lib, envoyGoFilterNewHttpPluginConfig(_)).WillByDefault(Return(1));
+  ON_CALL(*dso_lib, envoyGoFilterOnHttpHeader(_, _, _, _))
+      .WillByDefault(Return(static_cast<uint64_t>(GolangStatus::Running)));
+
+  bool destroyed = false;
+  bool data_called_after_destroy = false;
+  ON_CALL(*dso_lib, envoyGoFilterOnHttpData(_, _, _, _))
+      .WillByDefault(Invoke(
+          [&destroyed, &data_called_after_destroy](processState*, GoUint64, GoUint64, GoUint64) {
+            if (destroyed) {
+              data_called_after_destroy = true;
+            }
+            return static_cast<uint64_t>(GolangStatus::Continue);
+          }));
+  ON_CALL(*dso_lib, envoyGoFilterOnHttpDestroy(_, _))
+      .WillByDefault(Invoke([&destroyed](httpRequest*, int) { destroyed = true; }));
+
+  const auto yaml = R"EOF(
+    library_id: test
+    library_path: test
+    plugin_name: test
+    )EOF";
+  envoy::extensions::filters::http::golang::v3alpha::Config proto_config;
+  TestUtility::loadFromYaml(yaml, proto_config);
+  NiceMock<Server::Configuration::MockFactoryContext> mock_context;
+  auto config = std::make_shared<FilterConfig>(proto_config, dso_lib, "", mock_context);
+  config->newGoPluginConfig();
+
+  Network::Address::InstanceConstSharedPtr addr(
+      (*Network::Address::PipeInstance::create("/test/test.sock")).release());
+  NiceMock<Http::MockStreamDecoderFilterCallbacks> mock_callbacks;
+  NiceMock<Http::MockStreamEncoderFilterCallbacks> mock_enc_callbacks;
+  NiceMock<Envoy::Network::MockConnection> mock_connection;
+  ON_CALL(mock_callbacks, connection())
+      .WillByDefault(Return(OptRef<const Network::Connection>{mock_connection}));
+  mock_connection.stream_info_.downstream_connection_info_provider_->setRemoteAddress(addr);
+  mock_connection.stream_info_.downstream_connection_info_provider_->setLocalAddress(addr);
+  EXPECT_CALL(mock_callbacks.dispatcher_, isThreadSafe()).WillRepeatedly(Return(true));
+
+  auto filter = std::make_shared<TestFilter>(config, dso_lib, 0);
+  filter->setDecoderFilterCallbacks(mock_callbacks);
+  filter->setEncoderFilterCallbacks(mock_enc_callbacks);
+
+  Http::TestRequestHeaderMapImpl request_headers{{":path", "/"}};
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
+            filter->decodeHeaders(request_headers, false));
+
+  Buffer::OwnedImpl body("request body");
+  EXPECT_EQ(Http::FilterDataStatus::StopIterationNoBuffer, filter->decodeData(body, false));
+
+  EXPECT_CALL(mock_callbacks, continueDecoding()).WillOnce(Invoke([&filter]() {
+    filter->onDestroy();
+  }));
+
+  filter->continueStatusInternal(filter->testDecodingState(), GolangStatus::Continue);
+
+  EXPECT_FALSE(data_called_after_destroy)
+      << "envoyGoFilterOnHttpData must not be called after onDestroy";
+
+  ASSERT_NE(nullptr, filter->testReq());
+  delete filter->testReq();
+}
+
 } // namespace
 } // namespace Golang
 } // namespace HttpFilters