diff --git a/docs/root/version_history/v1.15.1.rst b/docs/root/version_history/v1.15.1.rst
new file mode 100644
index 000000000000..57b2acb3f307
--- /dev/null
+++ b/docs/root/version_history/v1.15.1.rst
@@ -0,0 +1,27 @@
+1.15.1 (TBD)
+============
+
+Changes
+-------
+* http: fixed CVE-2020-25017. Previously header matching did not match on all headers for non-inline
+  headers. This patch changes the default behavior to always logically match on all headers.
+  Multiple individual headers will be logically concatenated with ',' similar to what is done with
+  inline headers. This makes the behavior effectively consistent. This behavior can be temporary
+  reverted by setting the runtime value "envoy.reloadable_features.header_match_on_all_headers" to
+  "false".
+
+  Targeted fixes have been additionally performed on the following extensions which make them
+  consider all duplicate headers by default as a comma concatenated list:
+
+    1. Any extension using CEL matching on headers.
+    2. The header to metadata filter.
+    3. The JWT filter.
+    4. The Lua filter.
+
+  Like primary header matching used in routing, RBAC, etc. this behavior can be disabled by setting
+  the runtime value "envoy.reloadable_features.header_match_on_all_headers" to false.
+* http: The setCopy() header map API previously only set the first header in the case of duplicate
+  non-inline headers. setCopy() now behaves similarly to the other set*() APIs and replaces all found
+  headers with a single value. This may have had security implications in the extauth filter which
+  uses this API. This behavior can be disabled by setting the runtime value
+  "envoy.reloadable_features.http_set_copy_replace_all_headers" to false.
diff --git a/docs/root/version_history/version_history.rst b/docs/root/version_history/version_history.rst
index 07db664892d4..d9d9993e500c 100644
--- a/docs/root/version_history/version_history.rst
+++ b/docs/root/version_history/version_history.rst
@@ -7,6 +7,7 @@ Version history
   :titlesonly:
 
   current
+  v1.15.1
   v1.15.0
   v1.14.3
   v1.14.2
diff --git a/include/envoy/http/header_map.h b/include/envoy/http/header_map.h
index 358b642ed4d1..daa5bbeb1988 100644
--- a/include/envoy/http/header_map.h
+++ b/include/envoy/http/header_map.h
@@ -512,6 +512,32 @@ class HeaderMap {
    */
   virtual const HeaderEntry* get(const LowerCaseString& key) const PURE;
 
+  /**
+   * This is a wrapper for the return result from getAll(). It avoids a copy when translating from
+   * non-const HeaderEntry to const HeaderEntry and only provides const access to the result.
+   */
+  using NonConstGetResult = absl::InlinedVector<HeaderEntry*, 1>;
+  class GetResult {
+  public:
+    GetResult() = default;
+    explicit GetResult(NonConstGetResult&& result) : result_(std::move(result)) {}
+    void operator=(GetResult&& rhs) noexcept { result_ = std::move(rhs.result_); }
+
+    bool empty() const { return result_.empty(); }
+    size_t size() const { return result_.size(); }
+    const HeaderEntry* operator[](size_t i) const { return result_[i]; }
+
+  private:
+    NonConstGetResult result_;
+  };
+
+  /**
+   * Get a header by key.
+   * @param key supplies the header key.
+   * @return all header entries matching the key.
+   */
+  virtual GetResult getAll(const LowerCaseString& key) const PURE;
+
   // aliases to make iterate() and iterateReverse() callbacks easier to read
   enum class Iterate { Continue, Break };
 
diff --git a/source/common/http/BUILD b/source/common/http/BUILD
index 580a30ac64bc..fcc08663dcc2 100644
--- a/source/common/http/BUILD
+++ b/source/common/http/BUILD
@@ -306,6 +306,7 @@ envoy_cc_library(
         "//source/common/common:empty_string",
         "//source/common/common:non_copyable",
         "//source/common/common:utility_lib",
+        "//source/common/runtime:runtime_features_lib",
         "//source/common/singleton:const_singleton",
     ],
 )
diff --git a/source/common/http/header_map_impl.cc b/source/common/http/header_map_impl.cc
index ce63493486b7..2f6dec47dd22 100644
--- a/source/common/http/header_map_impl.cc
+++ b/source/common/http/header_map_impl.cc
@@ -10,6 +10,7 @@
 #include "common/common/assert.h"
 #include "common/common/dump_state_utils.h"
 #include "common/common/empty_string.h"
+#include "common/runtime/runtime_features.h"
 #include "common/singleton/const_singleton.h"
 
 #include "absl/strings/match.h"
@@ -383,9 +384,9 @@ void HeaderMapImpl::addCopy(const LowerCaseString& key, absl::string_view value)
 
 void HeaderMapImpl::appendCopy(const LowerCaseString& key, absl::string_view value) {
   // TODO(#9221): converge on and document a policy for coalescing multiple headers.
-  auto* entry = getExisting(key);
-  if (entry) {
-    const uint64_t added_size = appendToHeader(entry->value(), value);
+  auto entry = getExisting(key);
+  if (!entry.empty()) {
+    const uint64_t added_size = appendToHeader(entry[0]->value(), value);
     addSize(added_size);
   } else {
     addCopy(key, value);
@@ -393,29 +394,27 @@ void HeaderMapImpl::appendCopy(const LowerCaseString& key, absl::string_view val
 }
 
 void HeaderMapImpl::setReference(const LowerCaseString& key, absl::string_view value) {
-  HeaderString ref_key(key);
-  HeaderString ref_value(value);
   remove(key);
-  insertByKey(std::move(ref_key), std::move(ref_value));
+  addReference(key, value);
 }
 
 void HeaderMapImpl::setReferenceKey(const LowerCaseString& key, absl::string_view value) {
-  HeaderString ref_key(key);
-  HeaderString new_value;
-  new_value.setCopy(value);
   remove(key);
-  insertByKey(std::move(ref_key), std::move(new_value));
-  ASSERT(new_value.empty()); // NOLINT(bugprone-use-after-move)
+  addReferenceKey(key, value);
 }
 
 void HeaderMapImpl::setCopy(const LowerCaseString& key, absl::string_view value) {
-  // Replaces the first occurrence of a header if it exists, otherwise adds by copy.
-  // TODO(#9221): converge on and document a policy for coalescing multiple headers.
-  auto* entry = getExisting(key);
-  if (entry) {
-    updateSize(entry->value().size(), value.size());
-    entry->value(value);
+  if (!Runtime::runtimeFeatureEnabled(
+          "envoy.reloadable_features.http_set_copy_replace_all_headers")) {
+    auto entry = getExisting(key);
+    if (!entry.empty()) {
+      updateSize(entry[0]->value().size(), value.size());
+      entry[0]->value(value);
+    } else {
+      addCopy(key, value);
+    }
   } else {
+    remove(key);
     addCopy(key, value);
   }
 }
@@ -433,17 +432,26 @@ void HeaderMapImpl::verifyByteSizeInternalForTest() const {
 }
 
 const HeaderEntry* HeaderMapImpl::get(const LowerCaseString& key) const {
-  return const_cast<HeaderMapImpl*>(this)->getExisting(key);
+  const auto result = getAll(key);
+  return result.empty() ? nullptr : result[0];
 }
 
-HeaderEntry* HeaderMapImpl::getExisting(const LowerCaseString& key) {
+HeaderMap::GetResult HeaderMapImpl::getAll(const LowerCaseString& key) const {
+  return HeaderMap::GetResult(const_cast<HeaderMapImpl*>(this)->getExisting(key));
+}
+
+HeaderMap::NonConstGetResult HeaderMapImpl::getExisting(const LowerCaseString& key) {
   // Attempt a trie lookup first to see if the user is requesting an O(1) header. This may be
   // relatively common in certain header matching / routing patterns.
   // TODO(mattklein123): Add inline handle support directly to the header matcher code to support
   // this use case more directly.
+  HeaderMap::NonConstGetResult ret;
   auto lookup = staticLookup(key.get());
   if (lookup.has_value()) {
-    return *lookup.value().entry_;
+    if (*lookup.value().entry_ != nullptr) {
+      ret.push_back(*lookup.value().entry_);
+    }
+    return ret;
   }
 
   // If the requested header is not an O(1) header we do a full scan. Doing the trie lookup is
@@ -454,11 +462,11 @@ HeaderEntry* HeaderMapImpl::getExisting(const LowerCaseString& key) {
   // implementation or potentially create a lazy map if the size of the map is above a threshold.
   for (HeaderEntryImpl& header : headers_) {
     if (header.key() == key.get().c_str()) {
-      return &header;
+      ret.push_back(&header);
     }
   }
 
-  return nullptr;
+  return ret;
 }
 
 void HeaderMapImpl::iterate(HeaderMap::ConstIterateCb cb) const {
diff --git a/source/common/http/header_map_impl.h b/source/common/http/header_map_impl.h
index d4cb88bdcacb..4ac2839b2a64 100644
--- a/source/common/http/header_map_impl.h
+++ b/source/common/http/header_map_impl.h
@@ -86,6 +86,7 @@ class HeaderMapImpl : NonCopyable {
   void setCopy(const LowerCaseString& key, absl::string_view value);
   uint64_t byteSize() const;
   const HeaderEntry* get(const LowerCaseString& key) const;
+  HeaderMap::GetResult getAll(const LowerCaseString& key) const;
   void iterate(HeaderMap::ConstIterateCb cb) const;
   void iterateReverse(HeaderMap::ConstIterateCb cb) const;
   void clear();
@@ -241,7 +242,7 @@ class HeaderMapImpl : NonCopyable {
   HeaderEntryImpl& maybeCreateInline(HeaderEntryImpl** entry, const LowerCaseString& key);
   HeaderEntryImpl& maybeCreateInline(HeaderEntryImpl** entry, const LowerCaseString& key,
                                      HeaderString&& value);
-  HeaderEntry* getExisting(const LowerCaseString& key);
+  HeaderMap::NonConstGetResult getExisting(const LowerCaseString& key);
   size_t removeInline(HeaderEntryImpl** entry);
   void updateSize(uint64_t from_size, uint64_t to_size);
   void addSize(uint64_t size);
@@ -299,6 +300,9 @@ template <class Interface> class TypedHeaderMapImpl : public HeaderMapImpl, publ
   const HeaderEntry* get(const LowerCaseString& key) const override {
     return HeaderMapImpl::get(key);
   }
+  HeaderMap::GetResult getAll(const LowerCaseString& key) const override {
+    return HeaderMapImpl::getAll(key);
+  }
   void iterate(HeaderMap::ConstIterateCb cb) const override { HeaderMapImpl::iterate(cb); }
   void iterateReverse(HeaderMap::ConstIterateCb cb) const override {
     HeaderMapImpl::iterateReverse(cb);
diff --git a/source/common/http/header_utility.cc b/source/common/http/header_utility.cc
index 3b1726d0304e..91f15b4aeb0d 100644
--- a/source/common/http/header_utility.cc
+++ b/source/common/http/header_utility.cc
@@ -105,39 +105,68 @@ bool HeaderUtility::matchHeaders(const HeaderMap& request_headers,
   return true;
 }
 
+HeaderUtility::GetAllOfHeaderAsStringResult
+HeaderUtility::getAllOfHeaderAsString(const HeaderMap& headers, const Http::LowerCaseString& key) {
+  GetAllOfHeaderAsStringResult result;
+  const auto header_value = headers.getAll(key);
+
+  if (header_value.empty()) {
+    // Empty for clarity. Avoid handling the empty case in the block below if the runtime feature
+    // is disabled.
+  } else if (header_value.size() == 1 ||
+             !Runtime::runtimeFeatureEnabled(
+                 "envoy.reloadable_features.http_match_on_all_headers")) {
+    result.result_ = header_value[0]->value().getStringView();
+  } else {
+    // In this case we concatenate all found headers using a ',' delimiter before performing the
+    // final match. We use an InlinedVector of absl::string_view to invoke the optimized join
+    // algorithm. This requires a copying phase before we invoke join. The 3 used as the inline
+    // size has been arbitrarily chosen.
+    // TODO(mattklein123): Do we need to normalize any whitespace here?
+    absl::InlinedVector<absl::string_view, 3> string_view_vector;
+    string_view_vector.reserve(header_value.size());
+    for (size_t i = 0; i < header_value.size(); i++) {
+      string_view_vector.push_back(header_value[i]->value().getStringView());
+    }
+    result.result_backing_string_ = absl::StrJoin(string_view_vector, ",");
+  }
+
+  return result;
+}
+
 bool HeaderUtility::matchHeaders(const HeaderMap& request_headers, const HeaderData& header_data) {
-  const HeaderEntry* header = request_headers.get(header_data.name_);
+  const auto header_value = getAllOfHeaderAsString(request_headers, header_data.name_);
 
-  if (header == nullptr) {
+  if (!header_value.result().has_value()) {
     return header_data.invert_match_ && header_data.header_match_type_ == HeaderMatchType::Present;
   }
 
   bool match;
-  const absl::string_view header_view = header->value().getStringView();
   switch (header_data.header_match_type_) {
   case HeaderMatchType::Value:
-    match = header_data.value_.empty() || header_view == header_data.value_;
+    match = header_data.value_.empty() || header_value.result().value() == header_data.value_;
     break;
   case HeaderMatchType::Regex:
-    match = header_data.regex_->match(header_view);
+    match = header_data.regex_->match(header_value.result().value());
     break;
   case HeaderMatchType::Range: {
-    int64_t header_value = 0;
-    match = absl::SimpleAtoi(header_view, &header_value) &&
-            header_value >= header_data.range_.start() && header_value < header_data.range_.end();
+    int64_t header_int_value = 0;
+    match = absl::SimpleAtoi(header_value.result().value(), &header_int_value) &&
+            header_int_value >= header_data.range_.start() &&
+            header_int_value < header_data.range_.end();
     break;
   }
   case HeaderMatchType::Present:
     match = true;
     break;
   case HeaderMatchType::Prefix:
-    match = absl::StartsWith(header_view, header_data.value_);
+    match = absl::StartsWith(header_value.result().value(), header_data.value_);
     break;
   case HeaderMatchType::Suffix:
-    match = absl::EndsWith(header_view, header_data.value_);
+    match = absl::EndsWith(header_value.result().value(), header_data.value_);
     break;
   case HeaderMatchType::Contains:
-    match = absl::StrContains(header_view, header_data.value_);
+    match = absl::StrContains(header_value.result().value(), header_data.value_);
     break;
   default:
     NOT_REACHED_GCOVR_EXCL_LINE;
diff --git a/source/common/http/header_utility.h b/source/common/http/header_utility.h
index 27d2b9907361..cd9669efcf14 100644
--- a/source/common/http/header_utility.h
+++ b/source/common/http/header_utility.h
@@ -33,6 +33,35 @@ class HeaderUtility {
   static void getAllOfHeader(const HeaderMap& headers, absl::string_view key,
                              std::vector<absl::string_view>& out);
 
+  /**
+   * Get all header values as a single string. Multiple headers are concatenated with ','.
+   */
+  class GetAllOfHeaderAsStringResult {
+  public:
+    // The ultimate result of the concatenation. If absl::nullopt, no header values were found.
+    // If the final string required a string allocation, the memory is held in
+    // backingString(). This allows zero allocation in the common case of a single header
+    // value.
+    absl::optional<absl::string_view> result() const {
+      // This is safe for move/copy of this class as the backing string will be moved or copied.
+      // Otherwise result_ is valid. The assert verifies that both are empty or only 1 is set.
+      ASSERT((!result_.has_value() && result_backing_string_.empty()) ||
+             (result_.has_value() ^ !result_backing_string_.empty()));
+      return !result_backing_string_.empty() ? result_backing_string_ : result_;
+    }
+
+    const std::string& backingString() const { return result_backing_string_; }
+
+  private:
+    absl::optional<absl::string_view> result_;
+    // Valid only if result_ relies on memory allocation that must live beyond the call. See above.
+    std::string result_backing_string_;
+
+    friend class HeaderUtility;
+  };
+  static GetAllOfHeaderAsStringResult getAllOfHeaderAsString(const HeaderMap& headers,
+                                                             const Http::LowerCaseString& key);
+
   // A HeaderData specifies one of exact value or regex or range element
   // to match in a request's header, specified in the header_match_type_ member.
   // It is the runtime equivalent of the HeaderMatchSpecifier proto in RDS API.
diff --git a/source/common/runtime/runtime_features.cc b/source/common/runtime/runtime_features.cc
index 3415be06f081..173286c1489b 100644
--- a/source/common/runtime/runtime_features.cc
+++ b/source/common/runtime/runtime_features.cc
@@ -76,6 +76,8 @@ constexpr const char* runtime_features[] = {
     "envoy.reloadable_features.fixed_connection_close",
     "envoy.reloadable_features.hcm_stream_error_on_invalid_message",
     "envoy.reloadable_features.http_default_alpn",
+    "envoy.reloadable_features.http_match_on_all_headers",
+    "envoy.reloadable_features.http_set_copy_replace_all_headers",
     "envoy.reloadable_features.http_transport_failure_reason_in_body",
     "envoy.reloadable_features.http2_skip_encoding_empty_trailers",
     "envoy.reloadable_features.listener_in_place_filterchain_update",
diff --git a/source/extensions/filters/common/expr/context.cc b/source/extensions/filters/common/expr/context.cc
index 0655b1c735d8..a33228bcd73a 100644
--- a/source/extensions/filters/common/expr/context.cc
+++ b/source/extensions/filters/common/expr/context.cc
@@ -23,6 +23,19 @@ absl::optional<CelValue> convertHeaderEntry(const Http::HeaderEntry* header) {
   return CelValue::CreateStringView(header->value().getStringView());
 }
 
+absl::optional<CelValue>
+convertHeaderEntry(Protobuf::Arena& arena,
+                   Http::HeaderUtility::GetAllOfHeaderAsStringResult&& result) {
+  if (!result.result().has_value()) {
+    return {};
+  } else if (!result.backingString().empty()) {
+    return CelValue::CreateString(
+        Protobuf::Arena::Create<std::string>(&arena, result.backingString()));
+  } else {
+    return CelValue::CreateStringView(result.result().value());
+  }
+}
+
 namespace {
 
 absl::optional<CelValue> extractSslInfo(const Ssl::ConnectionInfo& ssl_info,
diff --git a/source/extensions/filters/common/expr/context.h b/source/extensions/filters/common/expr/context.h
index 09bf8d6a559e..fd4b386a9a32 100644
--- a/source/extensions/filters/common/expr/context.h
+++ b/source/extensions/filters/common/expr/context.h
@@ -4,6 +4,7 @@
 #include "envoy/stream_info/stream_info.h"
 
 #include "common/grpc/status.h"
+#include "common/http/header_utility.h"
 #include "common/http/headers.h"
 
 #include "eval/public/cel_value.h"
@@ -76,10 +77,13 @@ constexpr absl::string_view UpstreamTransportFailureReason = "transport_failure_
 class RequestWrapper;
 
 absl::optional<CelValue> convertHeaderEntry(const Http::HeaderEntry* header);
+absl::optional<CelValue>
+convertHeaderEntry(Protobuf::Arena& arena,
+                   Http::HeaderUtility::GetAllOfHeaderAsStringResult&& result);
 
 template <class T> class HeadersWrapper : public google::api::expr::runtime::CelMap {
 public:
-  HeadersWrapper(const T* value) : value_(value) {}
+  HeadersWrapper(Protobuf::Arena& arena, const T* value) : arena_(arena), value_(value) {}
   absl::optional<CelValue> operator[](CelValue key) const override {
     if (value_ == nullptr || !key.IsString()) {
       return {};
@@ -89,8 +93,8 @@ template <class T> class HeadersWrapper : public google::api::expr::runtime::Cel
       // Reject key if it is an invalid header string
       return {};
     }
-    auto out = value_->get(Http::LowerCaseString(str));
-    return convertHeaderEntry(out);
+    return convertHeaderEntry(
+        arena_, Http::HeaderUtility::getAllOfHeaderAsString(*value_, Http::LowerCaseString(str)));
   }
   int size() const override { return value_ == nullptr ? 0 : value_->size(); }
   bool empty() const override { return value_ == nullptr ? true : value_->empty(); }
@@ -101,6 +105,7 @@ template <class T> class HeadersWrapper : public google::api::expr::runtime::Cel
 private:
   friend class RequestWrapper;
   friend class ResponseWrapper;
+  Protobuf::Arena& arena_;
   const T* value_;
 };
 
@@ -127,8 +132,9 @@ class BaseWrapper : public google::api::expr::runtime::CelMap,
 
 class RequestWrapper : public BaseWrapper {
 public:
-  RequestWrapper(const Http::RequestHeaderMap* headers, const StreamInfo::StreamInfo& info)
-      : headers_(headers), info_(info) {}
+  RequestWrapper(Protobuf::Arena& arena, const Http::RequestHeaderMap* headers,
+                 const StreamInfo::StreamInfo& info)
+      : headers_(arena, headers), info_(info) {}
   absl::optional<CelValue> operator[](CelValue key) const override;
 
 private:
@@ -138,9 +144,9 @@ class RequestWrapper : public BaseWrapper {
 
 class ResponseWrapper : public BaseWrapper {
 public:
-  ResponseWrapper(const Http::ResponseHeaderMap* headers, const Http::ResponseTrailerMap* trailers,
-                  const StreamInfo::StreamInfo& info)
-      : headers_(headers), trailers_(trailers), info_(info) {}
+  ResponseWrapper(Protobuf::Arena& arena, const Http::ResponseHeaderMap* headers,
+                  const Http::ResponseTrailerMap* trailers, const StreamInfo::StreamInfo& info)
+      : headers_(arena, headers), trailers_(arena, trailers), info_(info) {}
   absl::optional<CelValue> operator[](CelValue key) const override;
 
 private:
diff --git a/source/extensions/filters/common/expr/evaluator.cc b/source/extensions/filters/common/expr/evaluator.cc
index fee62c0ad9fa..d2c376684f72 100644
--- a/source/extensions/filters/common/expr/evaluator.cc
+++ b/source/extensions/filters/common/expr/evaluator.cc
@@ -11,14 +11,15 @@ namespace Filters {
 namespace Common {
 namespace Expr {
 
-ActivationPtr createActivation(const StreamInfo::StreamInfo& info,
+ActivationPtr createActivation(Protobuf::Arena& arena, const StreamInfo::StreamInfo& info,
                                const Http::RequestHeaderMap* request_headers,
                                const Http::ResponseHeaderMap* response_headers,
                                const Http::ResponseTrailerMap* response_trailers) {
   auto activation = std::make_unique<Activation>();
-  activation->InsertValueProducer(Request, std::make_unique<RequestWrapper>(request_headers, info));
-  activation->InsertValueProducer(
-      Response, std::make_unique<ResponseWrapper>(response_headers, response_trailers, info));
+  activation->InsertValueProducer(Request,
+                                  std::make_unique<RequestWrapper>(arena, request_headers, info));
+  activation->InsertValueProducer(Response, std::make_unique<ResponseWrapper>(
+                                                arena, response_headers, response_trailers, info));
   activation->InsertValueProducer(Connection, std::make_unique<ConnectionWrapper>(info));
   activation->InsertValueProducer(Upstream, std::make_unique<UpstreamWrapper>(info));
   activation->InsertValueProducer(Source, std::make_unique<PeerWrapper>(info, false));
@@ -67,13 +68,14 @@ ExpressionPtr createExpression(Builder& builder, const google::api::expr::v1alph
   return std::move(cel_expression_status.value());
 }
 
-absl::optional<CelValue> evaluate(const Expression& expr, Protobuf::Arena* arena,
+absl::optional<CelValue> evaluate(const Expression& expr, Protobuf::Arena& arena,
                                   const StreamInfo::StreamInfo& info,
                                   const Http::RequestHeaderMap* request_headers,
                                   const Http::ResponseHeaderMap* response_headers,
                                   const Http::ResponseTrailerMap* response_trailers) {
-  auto activation = createActivation(info, request_headers, response_headers, response_trailers);
-  auto eval_status = expr.Evaluate(*activation, arena);
+  auto activation =
+      createActivation(arena, info, request_headers, response_headers, response_trailers);
+  auto eval_status = expr.Evaluate(*activation, &arena);
   if (!eval_status.ok()) {
     return {};
   }
@@ -84,7 +86,7 @@ absl::optional<CelValue> evaluate(const Expression& expr, Protobuf::Arena* arena
 bool matches(const Expression& expr, const StreamInfo::StreamInfo& info,
              const Http::RequestHeaderMap& headers) {
   Protobuf::Arena arena;
-  auto eval_status = Expr::evaluate(expr, &arena, info, &headers, nullptr, nullptr);
+  auto eval_status = Expr::evaluate(expr, arena, info, &headers, nullptr, nullptr);
   if (!eval_status.has_value()) {
     return false;
   }
diff --git a/source/extensions/filters/common/expr/evaluator.h b/source/extensions/filters/common/expr/evaluator.h
index 116239fde1a5..37fdf63ab1bf 100644
--- a/source/extensions/filters/common/expr/evaluator.h
+++ b/source/extensions/filters/common/expr/evaluator.h
@@ -25,7 +25,7 @@ using ExpressionPtr = std::unique_ptr<Expression>;
 
 // Creates an activation providing the common context attributes.
 // The activation lazily creates wrappers during an evaluation using the evaluation arena.
-ActivationPtr createActivation(const StreamInfo::StreamInfo& info,
+ActivationPtr createActivation(Protobuf::Arena& arena, const StreamInfo::StreamInfo& info,
                                const Http::RequestHeaderMap* request_headers,
                                const Http::ResponseHeaderMap* response_headers,
                                const Http::ResponseTrailerMap* response_trailers);
@@ -41,7 +41,7 @@ ExpressionPtr createExpression(Builder& builder, const google::api::expr::v1alph
 
 // Evaluates an expression for a request. The arena is used to hold intermediate computational
 // results and potentially the final value.
-absl::optional<CelValue> evaluate(const Expression& expr, Protobuf::Arena* arena,
+absl::optional<CelValue> evaluate(const Expression& expr, Protobuf::Arena& arena,
                                   const StreamInfo::StreamInfo& info,
                                   const Http::RequestHeaderMap* request_headers,
                                   const Http::ResponseHeaderMap* response_headers,
diff --git a/source/extensions/filters/http/header_to_metadata/BUILD b/source/extensions/filters/http/header_to_metadata/BUILD
index 1bbe574312e6..ad4f9bcf8cfe 100644
--- a/source/extensions/filters/http/header_to_metadata/BUILD
+++ b/source/extensions/filters/http/header_to_metadata/BUILD
@@ -19,6 +19,7 @@ envoy_cc_library(
     deps = [
         "//include/envoy/server:filter_config_interface",
         "//source/common/common:base64_lib",
+        "//source/common/http:header_utility_lib",
         "//source/common/http:utility_lib",
         "//source/extensions/filters/http:well_known_names",
         "@envoy_api//envoy/extensions/filters/http/header_to_metadata/v3:pkg_cc_proto",
diff --git a/source/extensions/filters/http/header_to_metadata/header_to_metadata_filter.cc b/source/extensions/filters/http/header_to_metadata/header_to_metadata_filter.cc
index 9550e243c6a1..2d5358f89aeb 100644
--- a/source/extensions/filters/http/header_to_metadata/header_to_metadata_filter.cc
+++ b/source/extensions/filters/http/header_to_metadata/header_to_metadata_filter.cc
@@ -5,6 +5,7 @@
 #include "common/common/base64.h"
 #include "common/common/regex.h"
 #include "common/config/well_known_names.h"
+#include "common/http/header_utility.h"
 #include "common/http/utility.h"
 #include "common/protobuf/protobuf.h"
 
@@ -20,12 +21,12 @@ namespace HeaderToMetadataFilter {
 
 // Extract the value of the header.
 absl::optional<std::string> HeaderValueSelector::extract(Http::HeaderMap& map) const {
-  const Http::HeaderEntry* header_entry = map.get(header_);
-  if (header_entry == nullptr) {
+  const auto header_value = Http::HeaderUtility::getAllOfHeaderAsString(map, header_);
+  if (!header_value.result().has_value()) {
     return absl::nullopt;
   }
   // Catch the value in the header before removing.
-  absl::optional<std::string> value = std::string(header_entry->value().getStringView());
+  absl::optional<std::string> value = std::string(header_value.result().value());
   if (remove_) {
     map.remove(header_);
   }
diff --git a/source/extensions/filters/http/jwt_authn/BUILD b/source/extensions/filters/http/jwt_authn/BUILD
index f0249b014ea1..1df6425b585f 100644
--- a/source/extensions/filters/http/jwt_authn/BUILD
+++ b/source/extensions/filters/http/jwt_authn/BUILD
@@ -14,6 +14,7 @@ envoy_cc_library(
     srcs = ["extractor.cc"],
     hdrs = ["extractor.h"],
     deps = [
+        "//source/common/http:header_utility_lib",
         "//source/common/http:utility_lib",
         "@envoy_api//envoy/extensions/filters/http/jwt_authn/v3:pkg_cc_proto",
     ],
diff --git a/source/extensions/filters/http/jwt_authn/extractor.cc b/source/extensions/filters/http/jwt_authn/extractor.cc
index b84f9fb4178f..b1ec12058ea9 100644
--- a/source/extensions/filters/http/jwt_authn/extractor.cc
+++ b/source/extensions/filters/http/jwt_authn/extractor.cc
@@ -5,6 +5,7 @@
 #include "envoy/extensions/filters/http/jwt_authn/v3/config.pb.h"
 
 #include "common/common/utility.h"
+#include "common/http/header_utility.h"
 #include "common/http/headers.h"
 #include "common/http/utility.h"
 #include "common/singleton/const_singleton.h"
@@ -187,9 +188,10 @@ ExtractorImpl::extract(const Http::RequestHeaderMap& headers) const {
   for (const auto& location_it : header_locations_) {
     const auto& location_spec = location_it.second;
     ENVOY_LOG(debug, "extract {}", location_it.first);
-    const Http::HeaderEntry* entry = headers.get(location_spec->header_);
-    if (entry) {
-      auto value_str = entry->value().getStringView();
+    const auto result =
+        Http::HeaderUtility::getAllOfHeaderAsString(headers, location_spec->header_);
+    if (result.result().has_value()) {
+      auto value_str = result.result().value();
       if (!location_spec->value_prefix_.empty()) {
         const auto pos = value_str.find(location_spec->value_prefix_);
         if (pos == absl::string_view::npos) {
diff --git a/source/extensions/filters/http/lua/BUILD b/source/extensions/filters/http/lua/BUILD
index 25b02828a28f..188b0c484752 100644
--- a/source/extensions/filters/http/lua/BUILD
+++ b/source/extensions/filters/http/lua/BUILD
@@ -43,6 +43,7 @@ envoy_cc_library(
         "//include/envoy/http:header_map_interface",
         "//include/envoy/stream_info:stream_info_interface",
         "//source/common/crypto:utility_lib",
+        "//source/common/http:header_utility_lib",
         "//source/common/http:utility_lib",
         "//source/extensions/common/crypto:utility_lib",
         "//source/extensions/filters/common/lua:lua_lib",
diff --git a/source/extensions/filters/http/lua/wrappers.cc b/source/extensions/filters/http/lua/wrappers.cc
index f716519dbefb..cb31e695f8be 100644
--- a/source/extensions/filters/http/lua/wrappers.cc
+++ b/source/extensions/filters/http/lua/wrappers.cc
@@ -1,5 +1,6 @@
 #include "extensions/filters/http/lua/wrappers.h"
 
+#include "common/http/header_utility.h"
 #include "common/http/utility.h"
 
 #include "extensions/filters/common/lua/wrappers.h"
@@ -42,10 +43,10 @@ int HeaderMapWrapper::luaAdd(lua_State* state) {
 
 int HeaderMapWrapper::luaGet(lua_State* state) {
   const char* key = luaL_checkstring(state, 2);
-  const Http::HeaderEntry* entry = headers_.get(Http::LowerCaseString(key));
-  if (entry != nullptr) {
-    lua_pushlstring(state, entry->value().getStringView().data(),
-                    entry->value().getStringView().length());
+  const auto value =
+      Http::HeaderUtility::getAllOfHeaderAsString(headers_, Http::LowerCaseString(key));
+  if (value.result().has_value()) {
+    lua_pushlstring(state, value.result().value().data(), value.result().value().length());
     return 1;
   } else {
     return 0;
diff --git a/test/common/http/BUILD b/test/common/http/BUILD
index 83668959f511..9acc034717dc 100644
--- a/test/common/http/BUILD
+++ b/test/common/http/BUILD
@@ -263,6 +263,7 @@ envoy_cc_test(
         "//source/common/http:header_list_view_lib",
         "//source/common/http:header_map_lib",
         "//source/common/http:header_utility_lib",
+        "//test/test_common:test_runtime_lib",
         "//test/test_common:utility_lib",
     ],
 )
@@ -304,6 +305,7 @@ envoy_cc_test(
     srcs = ["header_utility_test.cc"],
     deps = [
         "//source/common/http:header_utility_lib",
+        "//test/test_common:test_runtime_lib",
         "//test/test_common:utility_lib",
         "@envoy_api//envoy/config/route/v3:pkg_cc_proto",
     ],
diff --git a/test/common/http/header_map_impl_test.cc b/test/common/http/header_map_impl_test.cc
index 0e5b0c3df8cd..94995001405f 100644
--- a/test/common/http/header_map_impl_test.cc
+++ b/test/common/http/header_map_impl_test.cc
@@ -7,6 +7,7 @@
 #include "common/http/header_utility.h"
 
 #include "test/test_common/printers.h"
+#include "test/test_common/test_runtime.h"
 #include "test/test_common/utility.h"
 
 #include "gtest/gtest.h"
@@ -728,7 +729,11 @@ TEST(HeaderMapImplTest, SetReferenceKey) {
   EXPECT_EQ("monde", headers.get(foo)->value().getStringView());
 }
 
-TEST(HeaderMapImplTest, SetCopy) {
+TEST(HeaderMapImplTest, SetCopyOldBehavior) {
+  TestScopedRuntime runtime;
+  Runtime::LoaderSingleton::getExisting()->mergeValues(
+      {{"envoy.reloadable_features.http_set_copy_replace_all_headers", "false"}});
+
   TestRequestHeaderMapImpl headers;
   LowerCaseString foo("hello");
   headers.setCopy(foo, "world");
@@ -773,6 +778,50 @@ TEST(HeaderMapImplTest, SetCopy) {
   EXPECT_EQ(headers.getPathValue(), "/foo");
 }
 
+TEST(HeaderMapImplTest, SetCopyNewBehavior) {
+  TestRequestHeaderMapImpl headers;
+  LowerCaseString foo("hello");
+  headers.setCopy(foo, "world");
+  EXPECT_EQ("world", headers.get(foo)->value().getStringView());
+
+  // Overwrite value.
+  headers.setCopy(foo, "monde");
+  EXPECT_EQ("monde", headers.get(foo)->value().getStringView());
+
+  // Add another foo header.
+  headers.addCopy(foo, "monde2");
+  EXPECT_EQ(headers.size(), 2);
+
+  // The foo header is overridden.
+  headers.setCopy(foo, "override-monde");
+  EXPECT_EQ(headers.size(), 1);
+
+  HeaderAndValueCb cb;
+
+  InSequence seq;
+  EXPECT_CALL(cb, Call("hello", "override-monde"));
+  headers.iterate(cb.asIterateCb());
+
+  // Test setting an empty string and then overriding.
+  EXPECT_EQ(1UL, headers.remove(foo));
+  EXPECT_EQ(headers.size(), 0);
+  const std::string empty;
+  headers.setCopy(foo, empty);
+  EXPECT_EQ(headers.size(), 1);
+  headers.setCopy(foo, "not-empty");
+  EXPECT_EQ(headers.get(foo)->value().getStringView(), "not-empty");
+
+  // Use setCopy with inline headers both indirectly and directly.
+  headers.clear();
+  EXPECT_EQ(headers.size(), 0);
+  headers.setCopy(Headers::get().Path, "/");
+  EXPECT_EQ(headers.size(), 1);
+  EXPECT_EQ(headers.getPathValue(), "/");
+  headers.setPath("/foo");
+  EXPECT_EQ(headers.size(), 1);
+  EXPECT_EQ(headers.getPathValue(), "/foo");
+}
+
 TEST(HeaderMapImplTest, AddCopy) {
   TestRequestHeaderMapImpl headers;
 
diff --git a/test/common/http/header_utility_test.cc b/test/common/http/header_utility_test.cc
index ae0aaba39c42..068eb67eec8b 100644
--- a/test/common/http/header_utility_test.cc
+++ b/test/common/http/header_utility_test.cc
@@ -8,6 +8,7 @@
 #include "common/http/header_utility.h"
 #include "common/json/json_loader.h"
 
+#include "test/test_common/test_runtime.h"
 #include "test/test_common/utility.h"
 
 #include "gtest/gtest.h"
@@ -71,6 +72,53 @@ TEST_F(HeaderUtilityTest, RemovePortsFromHostConnect) {
   }
 }
 
+TEST(GetAllOfHeaderAsStringTest, All) {
+  const LowerCaseString test_header("test");
+  {
+    TestRequestHeaderMapImpl headers;
+    const auto ret = HeaderUtility::getAllOfHeaderAsString(headers, test_header);
+    EXPECT_FALSE(ret.result().has_value());
+    EXPECT_TRUE(ret.backingString().empty());
+  }
+  {
+    TestRequestHeaderMapImpl headers{{"test", "foo"}};
+    const auto ret = HeaderUtility::getAllOfHeaderAsString(headers, test_header);
+    EXPECT_EQ("foo", ret.result().value());
+    EXPECT_TRUE(ret.backingString().empty());
+  }
+  {
+    TestRequestHeaderMapImpl headers{{"test", "foo"}, {"test", "bar"}};
+    const auto ret = HeaderUtility::getAllOfHeaderAsString(headers, test_header);
+    EXPECT_EQ("foo,bar", ret.result().value());
+    EXPECT_EQ("foo,bar", ret.backingString());
+  }
+  {
+    TestRequestHeaderMapImpl headers{{"test", ""}, {"test", "bar"}};
+    const auto ret = HeaderUtility::getAllOfHeaderAsString(headers, test_header);
+    EXPECT_EQ(",bar", ret.result().value());
+    EXPECT_EQ(",bar", ret.backingString());
+  }
+  {
+    TestRequestHeaderMapImpl headers{{"test", ""}, {"test", ""}};
+    const auto ret = HeaderUtility::getAllOfHeaderAsString(headers, test_header);
+    EXPECT_EQ(",", ret.result().value());
+    EXPECT_EQ(",", ret.backingString());
+  }
+  {
+    TestRequestHeaderMapImpl headers{
+        {"test", "a"}, {"test", "b"}, {"test", "c"}, {"test", ""}, {"test", ""}};
+    const auto ret = HeaderUtility::getAllOfHeaderAsString(headers, test_header);
+    EXPECT_EQ("a,b,c,,", ret.result().value());
+    EXPECT_EQ("a,b,c,,", ret.backingString());
+    // Make sure copying the return value works correctly.
+    const auto ret2 = ret; // NOLINT(performance-unnecessary-copy-initialization)
+    EXPECT_EQ(ret2.result(), ret.result());
+    EXPECT_EQ(ret2.backingString(), ret.backingString());
+    EXPECT_EQ(ret2.result().value().data(), ret2.backingString().data());
+    EXPECT_NE(ret2.result().value().data(), ret.backingString().data());
+  }
+}
+
 TEST(HeaderDataConstructorTest, NoSpecifierSet) {
   const std::string yaml = R"EOF(
 name: test-header
@@ -236,8 +284,32 @@ regex_match: (a|b)
   EXPECT_FALSE(HeaderUtility::matchHeaders(headers, header_data));
 
   headers.addCopy("match-header", "a");
+  // With a single "match-header" this regex will match.
   EXPECT_TRUE(HeaderUtility::matchHeaders(headers, header_data));
+
   headers.addCopy("match-header", "b");
+  // With two "match-header" we now logically have "a,b" as the value, so the regex will not match.
+  EXPECT_FALSE(HeaderUtility::matchHeaders(headers, header_data));
+
+  header_data[0] = std::make_unique<HeaderUtility::HeaderData>(parseHeaderMatcherFromYaml(R"EOF(
+name: match-header
+exact_match: a,b
+  )EOF"));
+  // Make sure that an exact match on "a,b" does in fact work.
+  EXPECT_TRUE(HeaderUtility::matchHeaders(headers, header_data));
+
+  TestScopedRuntime runtime;
+  Runtime::LoaderSingleton::getExisting()->mergeValues(
+      {{"envoy.reloadable_features.http_match_on_all_headers", "false"}});
+  // Flipping runtime to false should make "a,b" no longer match because we will match on the first
+  // header only.
+  EXPECT_FALSE(HeaderUtility::matchHeaders(headers, header_data));
+
+  header_data[0] = std::make_unique<HeaderUtility::HeaderData>(parseHeaderMatcherFromYaml(R"EOF(
+name: match-header
+exact_match: a
+  )EOF"));
+  // With runtime off, exact match on "a" should pass.
   EXPECT_TRUE(HeaderUtility::matchHeaders(headers, header_data));
 }
 
diff --git a/test/extensions/filters/common/expr/context_test.cc b/test/extensions/filters/common/expr/context_test.cc
index bdcb4cef091d..0ac086e93603 100644
--- a/test/extensions/filters/common/expr/context_test.cc
+++ b/test/extensions/filters/common/expr/context_test.cc
@@ -25,7 +25,8 @@ namespace {
 constexpr absl::string_view Undefined = "undefined";
 
 TEST(Context, EmptyHeadersAttributes) {
-  HeadersWrapper<Http::RequestHeaderMap> headers(nullptr);
+  Protobuf::Arena arena;
+  HeadersWrapper<Http::RequestHeaderMap> headers(arena, nullptr);
   auto header = headers[CelValue::CreateStringView(Referer)];
   EXPECT_FALSE(header.has_value());
   EXPECT_EQ(0, headers.size());
@@ -34,7 +35,8 @@ TEST(Context, EmptyHeadersAttributes) {
 
 TEST(Context, InvalidRequest) {
   Http::TestRequestHeaderMapImpl header_map{{"referer", "dogs.com"}};
-  HeadersWrapper<Http::RequestHeaderMap> headers(&header_map);
+  Protobuf::Arena arena;
+  HeadersWrapper<Http::RequestHeaderMap> headers(arena, &header_map);
   auto header = headers[CelValue::CreateStringView("dogs.com\n")];
   EXPECT_FALSE(header.has_value());
 }
@@ -45,10 +47,11 @@ TEST(Context, RequestAttributes) {
   Http::TestRequestHeaderMapImpl header_map{
       {":method", "POST"},           {":scheme", "http"},      {":path", "/meow?yes=1"},
       {":authority", "kittens.com"}, {"referer", "dogs.com"},  {"user-agent", "envoy-mobile"},
-      {"content-length", "10"},      {"x-request-id", "blah"},
-  };
-  RequestWrapper request(&header_map, info);
-  RequestWrapper empty_request(nullptr, empty_info);
+      {"content-length", "10"},      {"x-request-id", "blah"}, {"double-header", "foo"},
+      {"double-header", "bar"}};
+  Protobuf::Arena arena;
+  RequestWrapper request(arena, &header_map, info);
+  RequestWrapper empty_request(arena, nullptr, empty_info);
 
   EXPECT_CALL(info, bytesReceived()).WillRepeatedly(Return(10));
   // "2018-04-03T23:06:09.123Z".
@@ -145,7 +148,7 @@ TEST(Context, RequestAttributes) {
     EXPECT_TRUE(value.has_value());
     ASSERT_TRUE(value.value().IsInt64());
     // this includes the headers size
-    EXPECT_EQ(138, value.value().Int64OrDie());
+    EXPECT_EQ(170, value.value().Int64OrDie());
   }
 
   {
@@ -169,12 +172,17 @@ TEST(Context, RequestAttributes) {
     ASSERT_TRUE(value.value().IsMap());
     auto& map = *value.value().MapOrDie();
     EXPECT_FALSE(map.empty());
-    EXPECT_EQ(8, map.size());
+    EXPECT_EQ(10, map.size());
 
     auto header = map[CelValue::CreateStringView(Referer)];
     EXPECT_TRUE(header.has_value());
     ASSERT_TRUE(header.value().IsString());
     EXPECT_EQ("dogs.com", header.value().StringOrDie().value());
+
+    auto header2 = map[CelValue::CreateStringView("double-header")];
+    EXPECT_TRUE(header2.has_value());
+    ASSERT_TRUE(header2.value().IsString());
+    EXPECT_EQ("foo,bar", header2.value().StringOrDie().value());
   }
 
   {
@@ -209,7 +217,8 @@ TEST(Context, RequestFallbackAttributes) {
       {":scheme", "http"},
       {":path", "/meow?yes=1"},
   };
-  RequestWrapper request(&header_map, info);
+  Protobuf::Arena arena;
+  RequestWrapper request(arena, &header_map, info);
 
   EXPECT_CALL(info, bytesReceived()).WillRepeatedly(Return(10));
 
@@ -236,8 +245,9 @@ TEST(Context, ResponseAttributes) {
   const std::string grpc_status = "grpc-status";
   Http::TestResponseHeaderMapImpl header_map{{header_name, "a"}};
   Http::TestResponseTrailerMapImpl trailer_map{{trailer_name, "b"}, {grpc_status, "8"}};
-  ResponseWrapper response(&header_map, &trailer_map, info);
-  ResponseWrapper empty_response(nullptr, nullptr, empty_info);
+  Protobuf::Arena arena;
+  ResponseWrapper response(arena, &header_map, &trailer_map, info);
+  ResponseWrapper empty_response(arena, nullptr, nullptr, empty_info);
 
   EXPECT_CALL(info, responseCode()).WillRepeatedly(Return(404));
   EXPECT_CALL(info, bytesSent()).WillRepeatedly(Return(123));
@@ -354,7 +364,8 @@ TEST(Context, ResponseAttributes) {
   {
     Http::TestResponseHeaderMapImpl header_map{{header_name, "a"}, {grpc_status, "7"}};
     Http::TestResponseTrailerMapImpl trailer_map{{trailer_name, "b"}};
-    ResponseWrapper response_header_status(&header_map, &trailer_map, info);
+    Protobuf::Arena arena;
+    ResponseWrapper response_header_status(arena, &header_map, &trailer_map, info);
     auto value = response_header_status[CelValue::CreateStringView(GrpcStatus)];
     EXPECT_TRUE(value.has_value());
     ASSERT_TRUE(value.value().IsInt64());
@@ -363,7 +374,8 @@ TEST(Context, ResponseAttributes) {
   {
     Http::TestResponseHeaderMapImpl header_map{{header_name, "a"}};
     Http::TestResponseTrailerMapImpl trailer_map{{trailer_name, "b"}};
-    ResponseWrapper response_no_status(&header_map, &trailer_map, info);
+    Protobuf::Arena arena;
+    ResponseWrapper response_no_status(arena, &header_map, &trailer_map, info);
     auto value = response_no_status[CelValue::CreateStringView(GrpcStatus)];
     EXPECT_TRUE(value.has_value());
     ASSERT_TRUE(value.value().IsInt64());
@@ -373,7 +385,8 @@ TEST(Context, ResponseAttributes) {
     NiceMock<StreamInfo::MockStreamInfo> info_without_code;
     Http::TestResponseHeaderMapImpl header_map{{header_name, "a"}};
     Http::TestResponseTrailerMapImpl trailer_map{{trailer_name, "b"}};
-    ResponseWrapper response_no_status(&header_map, &trailer_map, info_without_code);
+    Protobuf::Arena arena;
+    ResponseWrapper response_no_status(arena, &header_map, &trailer_map, info_without_code);
     auto value = response_no_status[CelValue::CreateStringView(GrpcStatus)];
     EXPECT_FALSE(value.has_value());
   }
diff --git a/test/extensions/filters/common/expr/evaluator_fuzz_test.cc b/test/extensions/filters/common/expr/evaluator_fuzz_test.cc
index 2f29c9f0023c..6c3867747ede 100644
--- a/test/extensions/filters/common/expr/evaluator_fuzz_test.cc
+++ b/test/extensions/filters/common/expr/evaluator_fuzz_test.cc
@@ -44,7 +44,7 @@ DEFINE_PROTO_FUZZER(const test::extensions::filters::common::expr::EvaluatorTest
 
     // Evaluate the CEL expression.
     Protobuf::Arena arena;
-    Expr::evaluate(*expr, &arena, *stream_info, &request_headers, &response_headers,
+    Expr::evaluate(*expr, arena, *stream_info, &request_headers, &response_headers,
                    &response_trailers);
   } catch (const CelException& e) {
     ENVOY_LOG_MISC(debug, "CelException: {}", e.what());
diff --git a/test/extensions/filters/http/header_to_metadata/header_to_metadata_filter_test.cc b/test/extensions/filters/http/header_to_metadata/header_to_metadata_filter_test.cc
index fc9c81431ae1..9a88da9ee050 100644
--- a/test/extensions/filters/http/header_to_metadata/header_to_metadata_filter_test.cc
+++ b/test/extensions/filters/http/header_to_metadata/header_to_metadata_filter_test.cc
@@ -109,6 +109,24 @@ TEST_F(HeaderToMetadataTest, BasicRequestTest) {
   filter_->onDestroy();
 }
 
+// Verify concatenation works.
+TEST_F(HeaderToMetadataTest, BasicRequestDoubleHeadersTest) {
+  initializeFilter(request_config_yaml);
+  Http::TestRequestHeaderMapImpl incoming_headers{{"X-VERSION", "foo"}, {"X-VERSION", "bar"}};
+  std::map<std::string, std::string> expected = {{"version", "foo,bar"}};
+
+  EXPECT_CALL(decoder_callbacks_, streamInfo()).WillRepeatedly(ReturnRef(req_info_));
+  EXPECT_CALL(req_info_, setDynamicMetadata("envoy.lb", MapEq(expected)));
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_->decodeHeaders(incoming_headers, false));
+  Http::MetadataMap metadata_map{{"metadata", "metadata"}};
+  EXPECT_EQ(Http::FilterMetadataStatus::Continue, filter_->decodeMetadata(metadata_map));
+  Buffer::OwnedImpl data("data");
+  EXPECT_EQ(Http::FilterDataStatus::Continue, filter_->decodeData(data, false));
+  Http::TestRequestTrailerMapImpl incoming_trailers;
+  EXPECT_EQ(Http::FilterTrailersStatus::Continue, filter_->decodeTrailers(incoming_trailers));
+  filter_->onDestroy();
+}
+
 TEST_F(HeaderToMetadataTest, PerRouteOverride) {
   // Global config is empty.
   initializeFilter("{}");
diff --git a/test/extensions/filters/http/jwt_authn/extractor_test.cc b/test/extensions/filters/http/jwt_authn/extractor_test.cc
index d91f2c7dfee1..a32d819fbeae 100644
--- a/test/extensions/filters/http/jwt_authn/extractor_test.cc
+++ b/test/extensions/filters/http/jwt_authn/extractor_test.cc
@@ -170,6 +170,14 @@ TEST_F(ExtractorTest, TestCustomHeaderToken) {
   EXPECT_FALSE(headers.get(Http::LowerCaseString("token-header")));
 }
 
+// Make sure a double custom header concatenates the token
+TEST_F(ExtractorTest, TestDoubleCustomHeaderToken) {
+  auto headers = TestRequestHeaderMapImpl{{"token-header", "jwt_token"}, {"token-header", "foo"}};
+  auto tokens = extractor_->extract(headers);
+  EXPECT_EQ(tokens.size(), 1);
+  EXPECT_EQ(tokens[0]->token(), "jwt_token,foo");
+}
+
 // Test extracting token from the custom header: "prefix-header"
 // value prefix doesn't match. It has to be either "AAA" or "AAABBB".
 TEST_F(ExtractorTest, TestPrefixHeaderNotMatch) {
diff --git a/test/extensions/filters/http/lua/wrappers_test.cc b/test/extensions/filters/http/lua/wrappers_test.cc
index 7ccf297bf343..e3b03f538eaf 100644
--- a/test/extensions/filters/http/lua/wrappers_test.cc
+++ b/test/extensions/filters/http/lua/wrappers_test.cc
@@ -44,6 +44,10 @@ TEST_F(LuaHeaderMapWrapperTest, Methods) {
       for key, value in pairs(object) do
         testPrint(string.format("'%s' '%s'", key, value))
       end
+
+      object:add("header3", "foo")
+      object:add("header3", "bar")
+      testPrint(object:get("header3"))
     end
   )EOF"};
 
@@ -58,6 +62,7 @@ TEST_F(LuaHeaderMapWrapperTest, Methods) {
   EXPECT_CALL(printer_, testPrint("'header2' 'foo'"));
   EXPECT_CALL(printer_, testPrint("'hello' 'WORLD'"));
   EXPECT_CALL(printer_, testPrint("'header2' 'foo'"));
+  EXPECT_CALL(printer_, testPrint("foo,bar"));
   start("callMe");
 }
 
diff --git a/test/extensions/filters/http/rbac/rbac_filter_integration_test.cc b/test/extensions/filters/http/rbac/rbac_filter_integration_test.cc
index 49455bef6789..89b73bbef081 100644
--- a/test/extensions/filters/http/rbac/rbac_filter_integration_test.cc
+++ b/test/extensions/filters/http/rbac/rbac_filter_integration_test.cc
@@ -92,6 +92,33 @@ name: rbac
           - any: true
 )EOF";
 
+const std::string RBAC_CONFIG_HEADER_MATCH_CONDITION = R"EOF(
+name: rbac
+typed_config:
+  "@type": type.googleapis.com/envoy.extensions.filters.http.rbac.v3.RBAC
+  rules:
+    policies:
+      foo:
+        permissions:
+          - any: true
+        principals:
+          - any: true
+        condition:
+          call_expr:
+            function: _==_
+            args:
+            - select_expr:
+                operand:
+                  select_expr:
+                    operand:
+                      ident_expr:
+                        name: request
+                    field: headers
+                field: xxx
+            - const_expr:
+               string_value: {}
+)EOF";
+
 using RBACIntegrationTest = HttpProtocolIntegrationTest;
 
 INSTANTIATE_TEST_SUITE_P(Protocols, RBACIntegrationTest,
@@ -357,5 +384,78 @@ TEST_P(RBACIntegrationTest, LogConnectionAllow) {
   EXPECT_EQ("200", response->headers().getStatusValue());
 }
 
+// Basic CEL match on a header value.
+TEST_P(RBACIntegrationTest, HeaderMatchCondition) {
+  config_helper_.addFilter(fmt::format(RBAC_CONFIG_HEADER_MATCH_CONDITION, "yyy"));
+  initialize();
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  auto response = codec_client_->makeRequestWithBody(
+      Http::TestRequestHeaderMapImpl{
+          {":method", "POST"},
+          {":path", "/path"},
+          {":scheme", "http"},
+          {":authority", "host"},
+          {"xxx", "yyy"},
+      },
+      1024);
+  waitForNextUpstreamRequest();
+  upstream_request_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "200"}}, true);
+
+  response->waitForEndStream();
+  ASSERT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().getStatusValue());
+}
+
+// CEL match on a header value in which the header is a duplicate. Verifies we handle string
+// copying correctly inside the CEL expression.
+TEST_P(RBACIntegrationTest, HeaderMatchConditionDuplicateHeaderNoMatch) {
+  config_helper_.addFilter(fmt::format(RBAC_CONFIG_HEADER_MATCH_CONDITION, "yyy"));
+  initialize();
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  auto response = codec_client_->makeRequestWithBody(
+      Http::TestRequestHeaderMapImpl{
+          {":method", "POST"},
+          {":path", "/path"},
+          {":scheme", "http"},
+          {":authority", "host"},
+          {"xxx", "yyy"},
+          {"xxx", "zzz"},
+      },
+      1024);
+  response->waitForEndStream();
+  ASSERT_TRUE(response->complete());
+  EXPECT_EQ("403", response->headers().getStatusValue());
+}
+
+// CEL match on a header value in which the header is a duplicate. Verifies we handle string
+// copying correctly inside the CEL expression.
+TEST_P(RBACIntegrationTest, HeaderMatchConditionDuplicateHeaderMatch) {
+  config_helper_.addFilter(fmt::format(RBAC_CONFIG_HEADER_MATCH_CONDITION, "yyy,zzz"));
+  initialize();
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  auto response = codec_client_->makeRequestWithBody(
+      Http::TestRequestHeaderMapImpl{
+          {":method", "POST"},
+          {":path", "/path"},
+          {":scheme", "http"},
+          {":authority", "host"},
+          {"xxx", "yyy"},
+          {"xxx", "zzz"},
+      },
+      1024);
+  waitForNextUpstreamRequest();
+  upstream_request_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "200"}}, true);
+
+  response->waitForEndStream();
+  ASSERT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().getStatusValue());
+}
+
 } // namespace
 } // namespace Envoy
diff --git a/test/test_common/utility.h b/test/test_common/utility.h
index 20a20246ef82..309c5e3b9993 100644
--- a/test/test_common/utility.h
+++ b/test/test_common/utility.h
@@ -937,6 +937,9 @@ template <class Interface, class Impl> class TestHeaderMapImplBase : public Inte
   const HeaderEntry* get(const LowerCaseString& key) const override {
     return header_map_->get(key);
   }
+  HeaderMap::GetResult getAll(const LowerCaseString& key) const override {
+    return header_map_->getAll(key);
+  }
   void iterate(HeaderMap::ConstIterateCb cb) const override { header_map_->iterate(cb); }
   void iterateReverse(HeaderMap::ConstIterateCb cb) const override {
     header_map_->iterateReverse(cb);