Envoy crashes when HTTP ext_proc processes local replies

@@ -105,6 +105,16 @@ bug_fixes:
     <envoy_v3_api_msg_extensions.http.ext_proc.response_processors.save_processing_response.v3.SaveProcessingResponse>` where the
     :ref:`response <envoy_v3_api_msg_service.ext_proc.v3.ProcessingResponse>` from the external processor was being saved to
     filter state after iterating through the filter chain.
+- area: ext_proc
+  change: |
+    Fixes a bug where local replies were incorrectly sent to the ext_proc server for external processing.
+    This change can be temporarily reverted by setting runtime guard ``envoy_reloadable_features_skip_ext_proc_on_local_reply``
+    to ``false``.
+- area: router
+  change: |
+    Fixes an Envoy crash issue when a local reply is sent.
+    This change can be temporarily reverted by setting runtime guard
+    ``envoy_reloadable_features_router_filter_resetall_on_local_reply`` to ``false``.
 
 removed_config_or_runtime:
 # *Normally occurs at the end of the* :ref:`deprecation period <deprecated>`

@@ -327,13 +327,19 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   // Http::StreamFilterBase
   void onDestroy() override;
 
-  // If there's a local reply (e.g. timeout) during host selection, cancel host
-  // selection.
   Http::LocalErrorStatus onLocalReply(const LocalReplyData&) override {
+    // If there's a local reply (e.g. timeout) during host selection, cancel host
+    // selection.
     if (host_selection_cancelable_) {
       host_selection_cancelable_->cancel();
       host_selection_cancelable_.reset();
     }
+
+    // Clean up the upstream_requests_.
+    if (Runtime::runtimeFeatureEnabled(
+            "envoy.reloadable_features.router_filter_resetall_on_local_reply")) {
+      resetAll();
+    }
     return Http::LocalErrorStatus::Continue;
   }
 

@@ -88,9 +88,11 @@ RUNTIME_GUARD(envoy_reloadable_features_quic_upstream_reads_fixed_number_packets
 RUNTIME_GUARD(envoy_reloadable_features_quic_upstream_socket_use_address_cache_for_read);
 RUNTIME_GUARD(envoy_reloadable_features_report_load_with_rq_issued);
 RUNTIME_GUARD(envoy_reloadable_features_report_stream_reset_error_code);
+RUNTIME_GUARD(envoy_reloadable_features_router_filter_resetall_on_local_reply);
 RUNTIME_GUARD(envoy_reloadable_features_sanitize_sni_in_access_log);
 RUNTIME_GUARD(envoy_reloadable_features_shadow_policy_inherit_trace_sampling);
 RUNTIME_GUARD(envoy_reloadable_features_skip_dns_lookup_for_proxied_requests);
+RUNTIME_GUARD(envoy_reloadable_features_skip_ext_proc_on_local_reply);
 RUNTIME_GUARD(envoy_reloadable_features_streaming_shadow);
 RUNTIME_GUARD(envoy_reloadable_features_tcp_proxy_retry_on_different_event_loop);
 RUNTIME_GUARD(envoy_reloadable_features_tcp_tunneling_send_downstream_fin_on_upstream_trailers);

@@ -476,6 +476,18 @@ class Filter : public Logger::Loggable<Logger::Id::ext_proc>,
                              absl::Status status,
                              envoy::config::core::v3::TrafficDirection traffic_direction);
 
+  Envoy::Http::LocalErrorStatus
+  onLocalReply(const Envoy::Http::StreamFilterBase::LocalReplyData&) override {
+    if (Runtime::runtimeFeatureEnabled("envoy.reloadable_features.skip_ext_proc_on_local_reply")) {
+      ENVOY_STREAM_LOG(debug,
+                       "When onLocalReply() is called, set processing_complete_ to true to skip "
+                       "external processing",
+                       *decoder_callbacks_);
+      processing_complete_ = true;
+    }
+    return ::Envoy::Http::LocalErrorStatus::Continue;
+  }
+
 private:
   void mergePerRouteConfig();
   StreamOpenState openStream();

@@ -38,7 +38,7 @@ class ExtProcMiscIntegrationTest : public HttpIntegrationTest,
 
     // Create separate "upstreams" for ExtProc gRPC servers
     for (int i = 0; i < grpc_upstream_count_; ++i) {
-      grpc_upstreams_.push_back(&addFakeUpstream(Http::CodecType::HTTP2));
+      grpc_upstreams_.push_back(&addFakeUpstream(http_codec_type_));
     }
   }
 
@@ -175,12 +175,60 @@ class ExtProcMiscIntegrationTest : public HttpIntegrationTest,
 
   bool IsEnvoyGrpc() { return std::get<1>(GetParam()) == Envoy::Grpc::ClientType::EnvoyGrpc; }
 
+  void websocketExtProcTest() {
+    if (!IsEnvoyGrpc()) {
+      return;
+    }
+
+    http_codec_type_ = Http::CodecType::HTTP1;
+    scoped_runtime_.mergeValues(
+        {{"envoy.reloadable_features.check_switch_protocol_websocket_handshake", "true"}});
+    auto* forward_rules = proto_config_.mutable_forward_rules();
+    auto* allowed_headers = forward_rules->mutable_allowed_headers();
+    allowed_headers->add_patterns()->set_exact("upgrade");
+    allowed_headers->add_patterns()->set_exact("connection");
+
+    config_helper_.addConfigModifier(
+        [](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
+               hcm) { hcm.add_upgrade_configs()->set_upgrade_type("websocket"); });
+
+    const std::string local_reply_yaml = R"EOF(
+body_format:
+  json_format:
+    code: "%RESPONSE_CODE%"
+    message: "%LOCAL_REPLY_BODY%"
+  )EOF";
+    envoy::extensions::filters::network::http_connection_manager::v3::LocalReplyConfig
+        local_reply_config;
+    TestUtility::loadFromYaml(local_reply_yaml, local_reply_config);
+    config_helper_.setLocalReply(local_reply_config);
+
+    initializeConfig();
+    HttpIntegrationTest::initialize();
+
+    auto response = sendDownstreamRequest([](Http::HeaderMap& headers) {
+      headers.addCopy(LowerCaseString("upgrade"), "websocket");
+      headers.addCopy(LowerCaseString("connection"), "Upgrade");
+    });
+
+    processRequestHeadersMessage(*grpc_upstreams_[0], true, absl::nullopt);
+
+    ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_));
+    ASSERT_TRUE(fake_upstream_connection_->waitForNewStream(*dispatcher_, upstream_request_));
+    upstream_request_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "401"}}, true);
+    if (!Runtime::runtimeFeatureEnabled("envoy.reloadable_features.skip_ext_proc_on_local_reply")) {
+      processResponseHeadersMessage(*grpc_upstreams_[0], false, absl::nullopt);
+    }
+    verifyDownstreamResponse(*response, 401);
+  }
+
   envoy::extensions::filters::http::ext_proc::v3::ExternalProcessor proto_config_{};
   std::vector<FakeUpstream*> grpc_upstreams_;
   FakeHttpConnectionPtr processor_connection_;
   FakeStreamPtr processor_stream_;
   TestScopedRuntime scoped_runtime_;
   int grpc_upstream_count_ = 2;
+  Http::CodecType http_codec_type_ = Http::CodecType::HTTP2;
 };
 
 INSTANTIATE_TEST_SUITE_P(IpVersionsClientTypeDeferredProcessing, ExtProcMiscIntegrationTest,
@@ -246,4 +294,21 @@ TEST_P(ExtProcMiscIntegrationTest, SendEmptyLastBodyChunk) {
   verifyDownstreamResponse(*response, 200);
 }
 
+// Test Ext_Proc filter and WebSocket configuration combination.
+TEST_P(ExtProcMiscIntegrationTest, WebSocketExtProcCombo) {
+  scoped_runtime_.mergeValues({{"envoy.reloadable_features.skip_ext_proc_on_local_reply", "true"}});
+  scoped_runtime_.mergeValues(
+      {{"envoy.reloadable_features.router_filter_resetall_on_local_reply", "false"}});
+  websocketExtProcTest();
+}
+
+// TODO(yanjunxiang-google): Delete this test after both runtime flags are removed.
+TEST_P(ExtProcMiscIntegrationTest, UpstreamRequestEncoderDanglingPointerTest) {
+  scoped_runtime_.mergeValues(
+      {{"envoy.reloadable_features.skip_ext_proc_on_local_reply", "false"}});
+  scoped_runtime_.mergeValues(
+      {{"envoy.reloadable_features.router_filter_resetall_on_local_reply", "true"}});
+  websocketExtProcTest();
+}
+
 } // namespace Envoy