header mutation: add query parameter mutation support (#37555)

This added query parameter mutation support for header mutation filter.

Risk Level: low.
Testing: unit.
Docs Changes: n/a.
Release Notes: added.
Platform Specific Features: n/a.

---------

Signed-off-by: wangbaiping(wbpcode) <[email protected]>
Signed-off-by: code <[email protected]>
Co-authored-by: Adi (Suissa) Peleg <[email protected]>

api/envoy/config/core/v3/base.proto

@@ -303,12 +303,31 @@ message RuntimeFeatureFlag {
   string runtime_key = 2 [(validate.rules).string = {min_len: 1}];
 }
 
+// Please use :ref:`KeyValuePair <envoy_api_msg_config.core.v3.KeyValuePair>` instead.
+// [#not-implemented-hide:]
 message KeyValue {
+  // The key of the key/value pair.
+  string key = 1 [
+    deprecated = true,
+    (validate.rules).string = {min_len: 1 max_bytes: 16384},
+    (envoy.annotations.deprecated_at_minor_version) = "3.0"
+  ];
+
+  // The value of the key/value pair.
+  //
+  // The ``bytes`` type is used. This means if JSON or YAML is used to to represent the
+  // configuration, the value must be base64 encoded. This is unfriendly for users in most
+  // use scenarios of this message.
+  //
+  bytes value = 2 [deprecated = true, (envoy.annotations.deprecated_at_minor_version) = "3.0"];
+}
+
+message KeyValuePair {
   // The key of the key/value pair.
   string key = 1 [(validate.rules).string = {min_len: 1 max_bytes: 16384}];
 
   // The value of the key/value pair.
-  bytes value = 2;
+  google.protobuf.Value value = 2;
 }
 
 // Key/value pair plus option to control append behavior. This is used to specify
@@ -339,8 +358,18 @@ message KeyValueAppend {
     OVERWRITE_IF_EXISTS = 3;
   }
 
-  // Key/value pair entry that this option to append or overwrite.
-  KeyValue entry = 1 [(validate.rules).message = {required: true}];
+  // The single key/value pair record to be appended or overridden. This field must be set.
+  KeyValuePair record = 3;
+
+  // Key/value pair entry that this option to append or overwrite. This field is deprecated
+  // and please use :ref:`record <envoy_v3_api_field_config.core.v3.KeyValueAppend.record>`
+  // as replacement.
+  // [#not-implemented-hide:]
+  KeyValue entry = 1 [
+    deprecated = true,
+    (validate.rules).message = {skip: true},
+    (envoy.annotations.deprecated_at_minor_version) = "3.0"
+  ];
 
   // Describes the action taken to append/overwrite the given value for an existing
   // key or to only add this key if it's absent.
@@ -349,10 +378,12 @@ message KeyValueAppend {
 
 // Key/value pair to append or remove.
 message KeyValueMutation {
-  // Key/value pair to append or overwrite. Only one of ``append`` or ``remove`` can be set.
+  // Key/value pair to append or overwrite. Only one of ``append`` or ``remove`` can be set or
+  // the configuration will be rejected.
   KeyValueAppend append = 1;
 
-  // Key to remove. Only one of ``append`` or ``remove`` can be set.
+  // Key to remove. Only one of ``append`` or ``remove`` can be set or the configuration will be
+  // rejected.
   string remove = 2 [(validate.rules).string = {max_bytes: 16384}];
 }
 

api/envoy/extensions/filters/http/header_mutation/v3/BUILD

@@ -7,6 +7,7 @@ licenses(["notice"])  # Apache 2
 api_proto_package(
     deps = [
         "//envoy/config/common/mutation_rules/v3:pkg",
+        "//envoy/config/core/v3:pkg",
         "@com_github_cncf_xds//udpa/annotations:pkg",
     ],
 )

api/envoy/extensions/filters/http/header_mutation/v3/header_mutation.proto

@@ -3,6 +3,7 @@ syntax = "proto3";
 package envoy.extensions.filters.http.header_mutation.v3;
 
 import "envoy/config/common/mutation_rules/v3/mutation_rules.proto";
+import "envoy/config/core/v3/base.proto";
 
 import "udpa/annotations/status.proto";
 
@@ -19,6 +20,10 @@ message Mutations {
   // The request mutations are applied before the request is forwarded to the upstream cluster.
   repeated config.common.mutation_rules.v3.HeaderMutation request_mutations = 1;
 
+  // The ``path`` header query parameter mutations are applied after ``request_mutations`` and before the request
+  // is forwarded to the next filter in the filter chain.
+  repeated config.core.v3.KeyValueMutation query_parameter_mutations = 3;
+
   // The response mutations are applied before the response is sent to the downstream client.
   repeated config.common.mutation_rules.v3.HeaderMutation response_mutations = 2;
 }

changelogs/current.yaml

@@ -364,6 +364,12 @@ new_features:
   change: |
     Added new health check filter stats including total requests, successful/failed checks, cached responses, and
     cluster health status counters. These stats help track health check behavior and cluster health state.
+- area: http
+  change: |
+    Add :ref:`query parameter mutations
+    <envoy_v3_api_field_extensions.filters.http.header_mutation.v3.Mutations.query_parameter_mutations>`
+    to :ref:`Header Mutation Filter <envoy_v3_api_msg_extensions.filters.http.header_mutation.v3.HeaderMutation>`
+    for adding/removing query parameters on a request.
 - area: local_ratelimit
   change: |
     Added per descriptor custom hits addend support for local rate limit filter. See :ref:`hits_addend

source/extensions/filters/http/header_mutation/config.cc

@@ -13,7 +13,10 @@ absl::StatusOr<Http::FilterFactoryCb>
 HeaderMutationFactoryConfig::createFilterFactoryFromProtoTyped(
     const ProtoConfig& config, const std::string&, DualInfo,
     Server::Configuration::ServerFactoryContext&) {
-  auto filter_config = std::make_shared<HeaderMutationConfig>(config);
+  absl::Status creation_status = absl::OkStatus();
+  auto filter_config = std::make_shared<HeaderMutationConfig>(config, creation_status);
+  RETURN_IF_NOT_OK_REF(creation_status);
+
   return [filter_config](Http::FilterChainFactoryCallbacks& callbacks) -> void {
     callbacks.addStreamFilter(std::make_shared<HeaderMutation>(filter_config));
   };
@@ -23,7 +26,10 @@ absl::StatusOr<Router::RouteSpecificFilterConfigConstSharedPtr>
 HeaderMutationFactoryConfig::createRouteSpecificFilterConfigTyped(
     const PerRouteProtoConfig& proto_config, Server::Configuration::ServerFactoryContext&,
     ProtobufMessage::ValidationVisitor&) {
-  return std::make_shared<PerRouteHeaderMutation>(proto_config);
+  absl::Status creation_status = absl::OkStatus();
+  auto route_config = std::make_shared<PerRouteHeaderMutation>(proto_config, creation_status);
+  RETURN_IF_NOT_OK_REF(creation_status);
+  return route_config;
 }
 
 using UpstreamHeaderMutationFactoryConfig = HeaderMutationFactoryConfig;

source/extensions/filters/http/header_mutation/header_mutation.cc

@@ -12,70 +12,164 @@ namespace Extensions {
 namespace HttpFilters {
 namespace HeaderMutation {
 
-void Mutations::mutateRequestHeaders(Http::HeaderMap& headers,
-                                     const Formatter::HttpFormatterContext& ctx,
+void QueryParameterMutationAppend::mutateQueryParameter(
+    Http::Utility::QueryParamsMulti& params, const Formatter::HttpFormatterContext& context,
+    const StreamInfo::StreamInfo& stream_info) const {
+
+  switch (action_) {
+    PANIC_ON_PROTO_ENUM_SENTINEL_VALUES;
+  case ParameterAppendProto::APPEND_IF_EXISTS_OR_ADD:
+    params.add(key_, formatter_->formatWithContext(context, stream_info));
+    return;
+  case ParameterAppendProto::ADD_IF_ABSENT: {
+    auto iter = params.data().find(key_);
+    if (iter == params.data().end()) {
+      params.add(key_, formatter_->formatWithContext(context, stream_info));
+    }
+    break;
+  }
+  case ParameterAppendProto::OVERWRITE_IF_EXISTS: {
+    auto iter = params.data().find(key_);
+    if (iter != params.data().end()) {
+      params.overwrite(key_, formatter_->formatWithContext(context, stream_info));
+    }
+    break;
+  }
+  case ParameterAppendProto::OVERWRITE_IF_EXISTS_OR_ADD:
+    params.overwrite(key_, formatter_->formatWithContext(context, stream_info));
+    break;
+  }
+}
+
+Mutations::Mutations(const MutationsProto& config, absl::Status& creation_status) {
+  auto request_mutations_or_error = HeaderMutations::create(config.request_mutations());
+  SET_AND_RETURN_IF_NOT_OK(request_mutations_or_error.status(), creation_status);
+  request_mutations_ = std::move(request_mutations_or_error.value());
+
+  auto response_mutations_or_error = HeaderMutations::create(config.response_mutations());
+  SET_AND_RETURN_IF_NOT_OK(response_mutations_or_error.status(), creation_status);
+  response_mutations_ = std::move(response_mutations_or_error.value());
+
+  query_query_parameter_mutations_.reserve(config.query_parameter_mutations_size());
+  for (const auto& mutation : config.query_parameter_mutations()) {
+    if (mutation.has_append()) {
+      if (!mutation.remove().empty()) {
+        creation_status =
+            absl::InvalidArgumentError("Only one of 'append'/'remove can be specified.");
+        return;
+      }
+
+      if (!mutation.append().has_record()) {
+        creation_status = absl::InvalidArgumentError("No record specified for append mutation.");
+        return;
+      }
+      if (!mutation.append().record().value().has_string_value()) {
+        creation_status =
+            absl::InvalidArgumentError("Only string value is allowed for record value.");
+        return;
+      }
+
+      auto value_or_error =
+          Formatter::FormatterImpl::create(mutation.append().record().value().string_value(), true);
+      SET_AND_RETURN_IF_NOT_OK(value_or_error.status(), creation_status);
+      query_query_parameter_mutations_.emplace_back(std::make_unique<QueryParameterMutationAppend>(
+          mutation.append().record().key(), std::move(value_or_error.value()),
+          mutation.append().action()));
+
+    } else if (!mutation.remove().empty()) {
+      query_query_parameter_mutations_.emplace_back(
+          std::make_unique<QueryParameterMutationRemove>(mutation.remove()));
+    } else {
+      creation_status = absl::InvalidArgumentError("One of 'append'/'remove' must be specified.");
+      return;
+    }
+  }
+}
+
+void Mutations::mutateRequestHeaders(Http::RequestHeaderMap& headers,
+                                     const Formatter::HttpFormatterContext& context,
                                      const StreamInfo::StreamInfo& stream_info) const {
-  request_mutations_->evaluateHeaders(headers, ctx, stream_info);
+  request_mutations_->evaluateHeaders(headers, context, stream_info);
+
+  if (query_query_parameter_mutations_.empty() || headers.Path() == nullptr) {
+    return;
+  }
+
+  // Mutate query parameters.
+  Http::Utility::QueryParamsMulti params =
+      Http::Utility::QueryParamsMulti::parseAndDecodeQueryString(headers.getPathValue());
+  for (const auto& mutation : query_query_parameter_mutations_) {
+    mutation->mutateQueryParameter(params, context, stream_info);
+  }
+  headers.setPath(params.replaceQueryString(headers.Path()->value()));
 }
 
-void Mutations::mutateResponseHeaders(Http::HeaderMap& headers,
-                                      const Formatter::HttpFormatterContext& ctx,
+void Mutations::mutateResponseHeaders(Http::ResponseHeaderMap& headers,
+                                      const Formatter::HttpFormatterContext& context,
                                       const StreamInfo::StreamInfo& stream_info) const {
-  response_mutations_->evaluateHeaders(headers, ctx, stream_info);
+  response_mutations_->evaluateHeaders(headers, context, stream_info);
 }
 
-PerRouteHeaderMutation::PerRouteHeaderMutation(const PerRouteProtoConfig& config)
-    : mutations_(config.mutations()) {}
+PerRouteHeaderMutation::PerRouteHeaderMutation(const PerRouteProtoConfig& config,
+                                               absl::Status& creation_status)
+    : mutations_(config.mutations(), creation_status) {}
 
-HeaderMutationConfig::HeaderMutationConfig(const ProtoConfig& config)
-    : mutations_(config.mutations()),
+HeaderMutationConfig::HeaderMutationConfig(const ProtoConfig& config, absl::Status& creation_status)
+    : mutations_(config.mutations(), creation_status),
       most_specific_header_mutations_wins_(config.most_specific_header_mutations_wins()) {}
 
-Http::FilterHeadersStatus HeaderMutation::decodeHeaders(Http::RequestHeaderMap& headers, bool) {
-  Formatter::HttpFormatterContext ctx{&headers};
-  config_->mutations().mutateRequestHeaders(headers, ctx, decoder_callbacks_->streamInfo());
+void HeaderMutation::maybeInitializeRouteConfigs(Http::StreamFilterCallbacks* callbacks) {
+  // Ensure that route configs are initialized only once and the same route configs are used
+  // for both decoding and encoding paths.
+  // An independent flag is used to ensure even at the case where the route configs is empty,
+  // we still won't try to initialize it again.
+  if (route_configs_initialized_) {
+    return;
+  }
+  route_configs_initialized_ = true;
 
   // Traverse through all route configs to retrieve all available header mutations.
   // `getAllPerFilterConfig` returns in ascending order of specificity (i.e., route table
   // first, then virtual host, then per route).
-  route_configs_ = Http::Utility::getAllPerFilterConfig<PerRouteHeaderMutation>(decoder_callbacks_);
-
+  route_configs_ = Http::Utility::getAllPerFilterConfig<PerRouteHeaderMutation>(callbacks);
+
+  // The order of returned route configs is route table first, then virtual host, then
+  // per route. That means the per route configs will be evaluated at the end and will
+  // override the more general virtual host and route table configs.
+  //
+  // So, if most_specific_header_mutations_wins is false, we need to reverse the order
+  // of route configs.
   if (!config_->mostSpecificHeaderMutationsWins()) {
-    // most_specific_wins means that most specific level per filter config is evaluated last. In
-    // other words, header mutations are evaluated in ascending order of specificity (same order as
-    // `getAllPerFilterConfig` above returns).
-    // Thus, here we reverse iterate the vector when `most_specific_wins` is false.
-    for (auto it = route_configs_.rbegin(); it != route_configs_.rend(); ++it) {
-      (*it).get().mutations().mutateRequestHeaders(headers, ctx, decoder_callbacks_->streamInfo());
-    }
-  } else {
-    for (const PerRouteHeaderMutation& route_config : route_configs_) {
-      route_config.mutations().mutateRequestHeaders(headers, ctx, decoder_callbacks_->streamInfo());
-    }
+    std::reverse(route_configs_.begin(), route_configs_.end());
+  }
+}
+
+Http::FilterHeadersStatus HeaderMutation::decodeHeaders(Http::RequestHeaderMap& headers, bool) {
+  Formatter::HttpFormatterContext context{&headers};
+  config_->mutations().mutateRequestHeaders(headers, context, decoder_callbacks_->streamInfo());
+
+  maybeInitializeRouteConfigs(decoder_callbacks_);
+
+  for (const PerRouteHeaderMutation& route_config : route_configs_) {
+    route_config.mutations().mutateRequestHeaders(headers, context,
+                                                  decoder_callbacks_->streamInfo());
   }
 
   return Http::FilterHeadersStatus::Continue;
 }
 
 Http::FilterHeadersStatus HeaderMutation::encodeHeaders(Http::ResponseHeaderMap& headers, bool) {
-  Formatter::HttpFormatterContext ctx{encoder_callbacks_->requestHeaders().ptr(), &headers};
-  config_->mutations().mutateResponseHeaders(headers, ctx, encoder_callbacks_->streamInfo());
+  Formatter::HttpFormatterContext context{encoder_callbacks_->requestHeaders().ptr(), &headers};
+  config_->mutations().mutateResponseHeaders(headers, context, encoder_callbacks_->streamInfo());
 
-  // If we haven't already traversed the route configs, do so now.
-  if (route_configs_.empty()) {
-    route_configs_ =
-        Http::Utility::getAllPerFilterConfig<PerRouteHeaderMutation>(encoder_callbacks_);
-  }
+  // Note if the filter before this one has send local reply then the decodeHeaders() will not be
+  // be called and the route config will not be initialized. Try it again here and it will be
+  // no-op if already initialized.
+  maybeInitializeRouteConfigs(encoder_callbacks_);
 
-  if (!config_->mostSpecificHeaderMutationsWins()) {
-    for (auto it = route_configs_.rbegin(); it != route_configs_.rend(); ++it) {
-      (*it).get().mutations().mutateResponseHeaders(headers, ctx, encoder_callbacks_->streamInfo());
-    }
-  } else {
-    for (const PerRouteHeaderMutation& route_config : route_configs_) {
-      route_config.mutations().mutateResponseHeaders(headers, ctx,
-                                                     encoder_callbacks_->streamInfo());
-    }
+  for (const PerRouteHeaderMutation& route_config : route_configs_) {
+    route_config.mutations().mutateResponseHeaders(headers, context,
+                                                   encoder_callbacks_->streamInfo());
   }
 
   return Http::FilterHeadersStatus::Continue;