filters: support on_downstream_data in set_filter_state (#43760)

<!--
!!!ATTENTION!!!

If you are fixing *any* crash or *any* potential security issue, *do
not*
open a pull request in this repo. Please report the issue via emailing
envoy-security@googlegroups.com where the issue will be triaged
appropriately.
Thank you in advance for helping to keep Envoy secure.

!!!ATTENTION!!!

For an explanation of how to fill out the fields, please see the
relevant section
in
[PULL_REQUESTS.md](https://github.com/envoyproxy/envoy/blob/main/PULL_REQUESTS.md)

!!!ATTENTION!!!

Please check the [use of generative AI
policy](https://github.com/envoyproxy/envoy/blob/main/CONTRIBUTING.md?plain=1#L41).

You may use generative AI only if you fully understand the code. You
need to disclose
this usage in the PR description to ensure transparency.
-->

Commit Message: filters: support on_downstream_data in set_filter_state
Additional Description: Similar to ON_DOWNSTREAM_TLS_HANDSHAKE support
in
set_filter_state, this PR adds the ability to set_filter_state when the
filter first
receives data. This allows to set filter state based on dynamic metadata
added
 by other filters when they first receive data.
Risk Level: Low
Testing: Integration tests added.
Docs Changes: Described added feature
Release Notes: filters: Added support for set_filter_state at
on_downstream_data time
Platform Specific Features:
Fixes #43640

---------

Signed-off-by: Guilherme Silva <26658340+GuilhermeJSilva@users.noreply.github.com>

Author: GuilhermeJSilva

api/envoy/extensions/filters/network/set_filter_state/v3/set_filter_state.proto

@@ -31,4 +31,8 @@ message Config {
   // For non-TLS downstream connections (where there is no TLS handshake), this
   // list is applied when a new connection is received.
   repeated common.set_filter_state.v3.FilterStateValue on_downstream_tls_handshake = 2;
+
+  // A sequence of the filter state values to apply in the specified order
+  // when data is first received from the downstream connection.
+  repeated common.set_filter_state.v3.FilterStateValue on_downstream_data = 3;
 }

changelogs/current.yaml

@@ -687,5 +687,18 @@ new_features:
     Added :ref:`set() <config_http_filters_lua_stream_info_filter_state_wrapper>` to the Lua filter
     state API, allowing Lua scripts to create and store filter state objects dynamically using
     registered object factories.
+- area: http
+  change: |
+    Fixed an issue where filter chain execution could continue on HTTP streams that had been reset but not yet
+    destroyed. This could cause use-after-free conditions when filter callbacks were invoked on filters that
+    had already received ``onDestroy()``. The fix ensures that ``decodeHeaders()``, ``decodeData()``,
+    ``decodeTrailers()``, and ``decodeMetadata()`` are blocked after a downstream reset.
+- area: network_filter
+  change: |
+    Added support for
+    ``on_downstream_data`` (see
+    :ref:`envoy_v3_api_field_extensions.filters.network.set_filter_state.v3.Config.on_downstream_data`)
+    to the :ref:`set_filter_state network filter <config_network_filters_set_filter_state>`, allowing
+    connection filter state to be populated after first receiving data from the downstream connection.
 
 deprecated:

docs/root/configuration/listeners/network_filters/set_filter_state.rst

@@ -17,6 +17,7 @@ The filter can apply values at different points in the connection lifecycle:
 * ``on_downstream_tls_handshake``: applied when the downstream TLS handshake is complete. For
   non-TLS downstream connections (where there is no TLS handshake), this list is applied when the
   new connection is accepted.
+* ``on_downstream_data``: applied when data is first received from the downstream connection.
 
 .. warning::
     This filter allows overriding the behavior of other extensions and
@@ -86,6 +87,20 @@ TLS handshake completes (e.g., downstream peer certificate SANs), use
       text_format_source:
         inline_string: "%DOWNSTREAM_PEER_URI_SAN%"
 
+When you need to populate filter state using information that is only available after receiving
+data from downstream (e.g., dynamic metadata set by another filter when data is received), use
+``on_downstream_data``:
+
+.. validated-code-block:: yaml
+  :type-name: envoy.extensions.filters.network.set_filter_state.v3.Config
+
+  on_downstream_data:
+  - object_key: my.custom.filter_state
+    factory_key: envoy.string
+    format_string:
+      text_format_source:
+        inline_string: "%DYNAMIC_METADATA(envoy.filters.network.ext_authz:key)%"
+
 
 Statistics
 ----------

source/extensions/filters/network/set_filter_state/config.cc

@@ -26,6 +26,14 @@ Network::FilterStatus SetFilterState::onNewConnection() {
   return Network::FilterStatus::Continue;
 }
 
+Network::FilterStatus SetFilterState::onData(Buffer::Instance&, bool) {
+  if (on_downstream_data_ != nullptr && waiting_for_downstream_data_) {
+    waiting_for_downstream_data_ = false;
+    on_downstream_data_->updateFilterState({}, read_callbacks_->connection().streamInfo());
+  }
+  return Network::FilterStatus::Continue;
+}
+
 void SetFilterState::onEvent(Network::ConnectionEvent event) {
   // For SSL connections the Connected event is raised after the downstream TLS handshake completes.
   // Mirror tcp_proxy's behavior: only run the TLS hook for downstream TLS connections and only
@@ -79,10 +87,17 @@ class SetFilterStateConfigFactory
               StreamInfo::FilterState::LifeSpan::Connection, context);
     }
 
-    return [on_new_connection_config,
-            on_downstream_tls_handshake_config](Network::FilterManager& filter_manager) -> void {
+    Filters::Common::SetFilterState::ConfigSharedPtr on_downstream_data_config;
+    if (!proto_config.on_downstream_data().empty()) {
+      on_downstream_data_config = std::make_shared<Filters::Common::SetFilterState::Config>(
+          proto_config.on_downstream_data(), StreamInfo::FilterState::LifeSpan::Connection,
+          context);
+    }
+
+    return [on_new_connection_config, on_downstream_tls_handshake_config,
+            on_downstream_data_config](Network::FilterManager& filter_manager) -> void {
       filter_manager.addReadFilter(std::make_shared<SetFilterState>(
-          on_new_connection_config, on_downstream_tls_handshake_config));
+          on_new_connection_config, on_downstream_tls_handshake_config, on_downstream_data_config));
     };
   }
 

source/extensions/filters/network/set_filter_state/config.h

@@ -15,14 +15,14 @@ class SetFilterState : public Network::ReadFilter,
                        Logger::Loggable<Logger::Id::filter> {
 public:
   SetFilterState(Filters::Common::SetFilterState::ConfigSharedPtr on_new_connection,
-                 Filters::Common::SetFilterState::ConfigSharedPtr on_downstream_tls_handshake)
+                 Filters::Common::SetFilterState::ConfigSharedPtr on_downstream_tls_handshake,
+                 Filters::Common::SetFilterState::ConfigSharedPtr on_downstream_data)
       : on_new_connection_(std::move(on_new_connection)),
-        on_downstream_tls_handshake_(std::move(on_downstream_tls_handshake)) {}
+        on_downstream_tls_handshake_(std::move(on_downstream_tls_handshake)),
+        on_downstream_data_(std::move(on_downstream_data)) {}
 
   // Network::ReadFilter
-  Network::FilterStatus onData(Buffer::Instance&, bool) override {
-    return Network::FilterStatus::Continue;
-  }
+  Network::FilterStatus onData(Buffer::Instance&, bool) override;
   Network::FilterStatus onNewConnection() override;
   void initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) override {
     read_callbacks_ = &callbacks;
@@ -36,6 +36,10 @@ class SetFilterState : public Network::ReadFilter,
         apply_downstream_tls_handshake_on_new_connection_ = true;
       }
     }
+
+    if (on_downstream_data_ != nullptr) {
+      waiting_for_downstream_data_ = true;
+    }
   }
 
   // Network::ConnectionCallbacks
@@ -48,10 +52,12 @@ class SetFilterState : public Network::ReadFilter,
 
   const Filters::Common::SetFilterState::ConfigSharedPtr on_new_connection_;
   const Filters::Common::SetFilterState::ConfigSharedPtr on_downstream_tls_handshake_;
+  const Filters::Common::SetFilterState::ConfigSharedPtr on_downstream_data_;
   Network::ReadFilterCallbacks* read_callbacks_{};
-  bool waiting_for_downstream_tls_handshake_{false};
-  bool apply_downstream_tls_handshake_on_new_connection_{false};
-  bool downstream_tls_handshake_{false};
+  bool waiting_for_downstream_tls_handshake_ : 1 {false};
+  bool apply_downstream_tls_handshake_on_new_connection_ : 1 {false};
+  bool downstream_tls_handshake_ : 1 {false};
+  bool waiting_for_downstream_data_ : 1 {false};
 };
 
 } // namespace SetFilterState

test/extensions/filters/network/set_filter_state/integration_test.cc

@@ -49,14 +49,20 @@ class SetFilterStateIntegrationTest : public testing::TestWithParam<Network::Add
             format_string:
               text_format_source:
                 inline_string: "baz"
+          on_downstream_data:
+          - object_key: on_data
+            factory_key: foo
+            format_string:
+              text_format_source:
+                inline_string: "on_data_set"
       - name: envoy.filters.network.echo
         typed_config:
             "@type": type.googleapis.com/envoy.extensions.filters.network.echo.v3.Echo
       )EOF");
   }
 
   void SetUp() override {
-    useListenerAccessLog("%FILTER_STATE(early)%|%FILTER_STATE(late)%");
+    useListenerAccessLog("%FILTER_STATE(early)%|%FILTER_STATE(late)%|%FILTER_STATE(on_data)%");
     BaseIntegrationTest::initialize();
   }
 };
@@ -70,7 +76,8 @@ TEST_P(SetFilterStateIntegrationTest, PlaintextConnectionAppliesBothLifecycleLis
   ASSERT_TRUE(tcp_client->write("hello"));
   ASSERT_TRUE(tcp_client->connected());
   tcp_client->close();
-  EXPECT_THAT(waitForAccessLog(listener_access_log_name_), testing::HasSubstr("\"bar\"|\"baz\""));
+  EXPECT_THAT(waitForAccessLog(listener_access_log_name_),
+              testing::HasSubstr("\"bar\"|\"baz\"|\"on_data_set\""));
 }
 
 class SetFilterStateDownstreamTlsIntegrationTest