diff --git a/VERSION b/VERSION
index 80138e714669..0b33e0b5d033 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-1.13.4
+1.13.5-dev
diff --git a/docs/root/intro/version_history.rst b/docs/root/intro/version_history.rst
index 4d1136dfc257..4a25e0005076 100644
--- a/docs/root/intro/version_history.rst
+++ b/docs/root/intro/version_history.rst
@@ -1,6 +1,32 @@
 Version history
 ---------------
 
+1.13.5 (Pending)
+================
+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: fixed CVE-2020-25017. 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.
+
 1.13.4 (July 7, 2020)
 =====================
 * tls: fixed a bug where wilcard matching for "\*.foo.com" also matched domains of the form "a.b.foo.com". This behavior can be temporarily reverted by setting runtime feature `envoy.reloadable_features.fix_wildcard_matching` to false.
diff --git a/include/envoy/http/BUILD b/include/envoy/http/BUILD
index 36217efeeeb8..3ded730dbfde 100644
--- a/include/envoy/http/BUILD
+++ b/include/envoy/http/BUILD
@@ -93,6 +93,9 @@ envoy_cc_library(
 envoy_cc_library(
     name = "header_map_interface",
     hdrs = ["header_map.h"],
+    external_deps = [
+        "abseil_inlined_vector",
+    ],
     deps = [
         "//source/common/common:assert_lib",
         "//source/common/common:hash_lib",
diff --git a/include/envoy/http/header_map.h b/include/envoy/http/header_map.h
index 9c65dfa1bc9b..de2eba246513 100644
--- a/include/envoy/http/header_map.h
+++ b/include/envoy/http/header_map.h
@@ -15,6 +15,7 @@
 #include "common/common/hash.h"
 #include "common/common/macros.h"
 
+#include "absl/container/inlined_vector.h"
 #include "absl/strings/string_view.h"
 
 namespace Envoy {
@@ -129,6 +130,12 @@ class HeaderString {
    */
   char* buffer() { return buffer_.dynamic_; }
 
+  /**
+   * Trim trailing whitespaces from the HeaderString.
+   * v1.13 supports both Inline and Dynamic, but not Reference type.
+   */
+  void rtrim();
+
   /**
    * Get an absl::string_view. It will NOT be NUL terminated!
    *
@@ -503,6 +510,31 @@ 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)) {}
+
+    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 e4a2634fa59b..6560450e0e98 100644
--- a/source/common/http/BUILD
+++ b/source/common/http/BUILD
@@ -251,6 +251,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 ecc98017b841..14463409d5a7 100644
--- a/source/common/http/header_map_impl.cc
+++ b/source/common/http/header_map_impl.cc
@@ -9,6 +9,7 @@
 #include "common/common/dump_state_utils.h"
 #include "common/common/empty_string.h"
 #include "common/common/utility.h"
+#include "common/runtime/runtime_features.h"
 #include "common/singleton/const_singleton.h"
 
 #include "absl/strings/match.h"
@@ -150,6 +151,15 @@ void HeaderString::append(const char* data, uint32_t size) {
   string_length_ += size;
 }
 
+void HeaderString::rtrim() {
+  ASSERT(type() == Type::Inline || type() == Type::Dynamic);
+  absl::string_view original = getStringView();
+  absl::string_view rtrimmed = StringUtil::rtrim(original);
+  if (original.size() != rtrimmed.size()) {
+    string_length_ = rtrimmed.size();
+  }
+}
+
 void HeaderString::clear() {
   switch (type_) {
   case Type::Reference: {
@@ -462,9 +472,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);
@@ -474,31 +484,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));
-  verifyByteSize();
+  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)
-  verifyByteSize();
+  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);
   }
   verifyByteSize();
@@ -517,23 +523,41 @@ uint64_t HeaderMapImpl::byteSizeInternal() const {
 }
 
 const HeaderEntry* HeaderMapImpl::get(const LowerCaseString& key) const {
-  for (const HeaderEntryImpl& header : headers_) {
-    if (header.key() == key.get().c_str()) {
-      return &header;
-    }
-  }
+  const auto result = getAll(key);
+  return result.empty() ? nullptr : result[0];
+}
 
-  return nullptr;
+HeaderMap::GetResult HeaderMapImpl::getAll(const LowerCaseString& key) const {
+  return HeaderMap::GetResult(const_cast<HeaderMapImpl*>(this)->getExisting(key));
 }
 
-HeaderEntry* HeaderMapImpl::getExisting(const LowerCaseString& 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;
+
+  EntryCb cb = ConstSingleton<StaticLookupTable>::get().find(key.get());
+  if (cb) {
+    StaticLookupResponse ref_lookup_response = cb(*this);
+    if (*ref_lookup_response.entry_) {
+      ret.push_back(*ref_lookup_response.entry_);
+    }
+    return ret;
+  }
+  // If the requested header is not an O(1) header we do a full scan. Doing the trie lookup is
+  // wasteful in the miss case, but is present for code consistency with other functions that do
+  // similar things.
+  // TODO(mattklein123): The full scan here and in remove() are the biggest issues with this
+  // implementation for certain use cases. We can either replace this with a totally different
+  // 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(ConstIterateCb cb, void* context) const {
diff --git a/source/common/http/header_map_impl.h b/source/common/http/header_map_impl.h
index bba7221c7d36..984411bb2132 100644
--- a/source/common/http/header_map_impl.h
+++ b/source/common/http/header_map_impl.h
@@ -88,6 +88,7 @@ class HeaderMapImpl : public HeaderMap, NonCopyable {
   void setCopy(const LowerCaseString& key, absl::string_view value) override;
   uint64_t byteSize() const override;
   const HeaderEntry* get(const LowerCaseString& key) const override;
+  HeaderMap::GetResult getAll(const LowerCaseString& key) const override;
   void iterate(ConstIterateCb cb, void* context) const override;
   void iterateReverse(ConstIterateCb cb, void* context) const override;
   Lookup lookup(const LowerCaseString& key, const HeaderEntry** entry) const override;
@@ -213,7 +214,7 @@ class HeaderMapImpl : public HeaderMap, 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);
   HeaderEntryImpl* getExistingInline(absl::string_view key);
 
   void removeInline(HeaderEntryImpl** entry);
diff --git a/source/common/http/header_utility.cc b/source/common/http/header_utility.cc
index 2c1904293603..a88af1440684 100644
--- a/source/common/http/header_utility.cc
+++ b/source/common/http/header_utility.cc
@@ -105,36 +105,65 @@ 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;
   default:
     NOT_REACHED_GCOVR_EXCL_LINE;
diff --git a/source/common/http/header_utility.h b/source/common/http/header_utility.h
index ce7a633b4e72..3b15665b9e3a 100644
--- a/source/common/http/header_utility.h
+++ b/source/common/http/header_utility.h
@@ -32,6 +32,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/http/http1/codec_impl.cc b/source/common/http/http1/codec_impl.cc
index f2f58328f155..5a68a7d76bc7 100644
--- a/source/common/http/http1/codec_impl.cc
+++ b/source/common/http/http1/codec_impl.cc
@@ -461,6 +461,10 @@ void ConnectionImpl::completeLastHeader() {
   checkHeaderNameForUnderscores();
   if (!current_header_field_.empty()) {
     toLowerTable().toLowerCase(current_header_field_.buffer(), current_header_field_.size());
+    // Strip trailing whitespace of the current header value if any. Leading whitespace was trimmed
+    // in ConnectionImpl::onHeaderValue. http_parser does not strip leading or trailing whitespace
+    // as the spec requires: https://tools.ietf.org/html/rfc7230#section-3.2.4
+    current_header_value_.rtrim();
     current_header_map_->addViaMove(std::move(current_header_field_),
                                     std::move(current_header_value_));
   }
@@ -584,9 +588,8 @@ void ConnectionImpl::onHeaderValue(const char* data, size_t length) {
   if (!current_header_map_) {
     current_header_map_ = std::make_unique<HeaderMapImpl>();
   }
-  // Work around a bug in http_parser where trailing whitespace is not trimmed
-  // as the spec requires: https://tools.ietf.org/html/rfc7230#section-3.2.4
-  const absl::string_view header_value = StringUtil::trim(absl::string_view(data, length));
+
+  absl::string_view header_value{data, length};
 
   if (strict_header_validation_) {
     if (!Http::HeaderUtility::headerValueIsValid(header_value)) {
@@ -604,6 +607,13 @@ void ConnectionImpl::onHeaderValue(const char* data, size_t length) {
   }
 
   header_parsing_state_ = HeaderParsingState::Value;
+  if (current_header_value_.empty()) {
+    // Strip leading whitespace if the current header value input contains the first bytes of the
+    // encoded header value. Trailing whitespace is stripped once the full header value is known in
+    // ConnectionImpl::completeLastHeader. http_parser does not strip leading or trailing whitespace
+    // as the spec requires: https://tools.ietf.org/html/rfc7230#section-3.2.4 .
+    header_value = StringUtil::ltrim(header_value);
+  }
   current_header_value_.append(header_value.data(), header_value.length());
 
   checkMaxHeadersSize();
diff --git a/source/common/runtime/BUILD b/source/common/runtime/BUILD
index a5ac6bcaef76..98710d3c40f7 100644
--- a/source/common/runtime/BUILD
+++ b/source/common/runtime/BUILD
@@ -17,6 +17,8 @@ envoy_cc_library(
         "runtime_features.h",
     ],
     deps = [
+        "//include/envoy/runtime:runtime_interface",
+        "//source/common/common:assert_lib",
         "//source/common/common:hash_lib",
         "//source/common/singleton:const_singleton",
     ],
diff --git a/source/common/runtime/runtime_features.cc b/source/common/runtime/runtime_features.cc
index 7a415ff8db0a..957cca68b46a 100644
--- a/source/common/runtime/runtime_features.cc
+++ b/source/common/runtime/runtime_features.cc
@@ -1,8 +1,37 @@
 #include "common/runtime/runtime_features.h"
 
+#include "common/common/assert.h"
+
 namespace Envoy {
 namespace Runtime {
 
+bool isRuntimeFeature(absl::string_view feature) {
+  return RuntimeFeaturesDefaults::get().enabledByDefault(feature) ||
+         RuntimeFeaturesDefaults::get().existsButDisabled(feature);
+}
+
+bool runtimeFeatureEnabled(absl::string_view feature) {
+  ASSERT(isRuntimeFeature(feature));
+  if (Runtime::LoaderSingleton::getExisting()) {
+    return Runtime::LoaderSingleton::getExisting()->threadsafeSnapshot()->runtimeFeatureEnabled(
+        feature);
+  }
+  ENVOY_LOG_TO_LOGGER(Envoy::Logger::Registry::getLog(Envoy::Logger::Id::runtime), warn,
+                      "Unable to use runtime singleton for feature {}", feature);
+  return RuntimeFeaturesDefaults::get().enabledByDefault(feature);
+}
+
+uint64_t getInteger(absl::string_view feature, uint64_t default_value) {
+  ASSERT(absl::StartsWith(feature, "envoy."));
+  if (Runtime::LoaderSingleton::getExisting()) {
+    return Runtime::LoaderSingleton::getExisting()->threadsafeSnapshot()->getInteger(
+        std::string(feature), default_value);
+  }
+  ENVOY_LOG_TO_LOGGER(Envoy::Logger::Registry::getLog(Envoy::Logger::Id::runtime), warn,
+                      "Unable to use runtime singleton for feature {}", feature);
+  return default_value;
+}
+
 // Add additional features here to enable the new code paths by default.
 //
 // Per documentation in CONTRIBUTING.md is expected that new high risk code paths be guarded
@@ -34,6 +63,8 @@ constexpr const char* runtime_features[] = {
     "envoy.reloadable_features.reject_unsupported_transfer_encodings",
     "envoy.reloadable_features.strict_method_validation",
     "envoy.reloadable_features.fix_wildcard_matching",
+    "envoy.reloadable_features.http_match_on_all_headers",
+    "envoy.reloadable_features.http_set_copy_replace_all_headers",
 };
 
 // This is a section for officially sanctioned runtime features which are too
diff --git a/source/common/runtime/runtime_features.h b/source/common/runtime/runtime_features.h
index e8c7b2bc4851..38bc633ea7ce 100644
--- a/source/common/runtime/runtime_features.h
+++ b/source/common/runtime/runtime_features.h
@@ -2,13 +2,21 @@
 
 #include <string>
 
+#include "envoy/runtime/runtime.h"
+
 #include "common/singleton/const_singleton.h"
+#include "common/singleton/threadsafe_singleton.h"
 
 #include "absl/container/flat_hash_set.h"
+#include "absl/strings/match.h"
 
 namespace Envoy {
 namespace Runtime {
 
+bool isRuntimeFeature(absl::string_view feature);
+bool runtimeFeatureEnabled(absl::string_view feature);
+uint64_t getInteger(absl::string_view feature, uint64_t default_value);
+
 class RuntimeFeatures {
 public:
   RuntimeFeatures();
diff --git a/source/common/runtime/runtime_impl.cc b/source/common/runtime/runtime_impl.cc
index 8f6927ad9e19..ba8912edb636 100644
--- a/source/common/runtime/runtime_impl.cc
+++ b/source/common/runtime/runtime_impl.cc
@@ -45,36 +45,8 @@ bool isLegacyFeature(absl::string_view feature) {
   return absl::StartsWith(feature, "envoy.reloadable_features.http2_protocol_options.") ||
          absl::StartsWith(feature, "envoy.reloadable_features.max_re");
 }
-
-bool isRuntimeFeature(absl::string_view feature) {
-  return RuntimeFeaturesDefaults::get().enabledByDefault(feature) ||
-         RuntimeFeaturesDefaults::get().existsButDisabled(feature);
-}
-
 } // namespace
 
-bool runtimeFeatureEnabled(absl::string_view feature) {
-  ASSERT(isRuntimeFeature(feature));
-  if (Runtime::LoaderSingleton::getExisting()) {
-    return Runtime::LoaderSingleton::getExisting()->threadsafeSnapshot()->runtimeFeatureEnabled(
-        feature);
-  }
-  ENVOY_LOG_TO_LOGGER(Envoy::Logger::Registry::getLog(Envoy::Logger::Id::runtime), warn,
-                      "Unable to use runtime singleton for feature {}", feature);
-  return RuntimeFeaturesDefaults::get().enabledByDefault(feature);
-}
-
-uint64_t getInteger(absl::string_view feature, uint64_t default_value) {
-  ASSERT(absl::StartsWith(feature, "envoy."));
-  if (Runtime::LoaderSingleton::getExisting()) {
-    return Runtime::LoaderSingleton::getExisting()->threadsafeSnapshot()->getInteger(
-        std::string(feature), default_value);
-  }
-  ENVOY_LOG_TO_LOGGER(Envoy::Logger::Registry::getLog(Envoy::Logger::Id::runtime), warn,
-                      "Unable to use runtime singleton for feature {}", feature);
-  return default_value;
-}
-
 const size_t RandomGeneratorImpl::UUID_LENGTH = 36;
 
 uint64_t RandomGeneratorImpl::random() {
diff --git a/source/extensions/common/wasm/context.cc b/source/extensions/common/wasm/context.cc
index 3e459c770122..be81b329b0fb 100644
--- a/source/extensions/common/wasm/context.cc
+++ b/source/extensions/common/wasm/context.cc
@@ -448,13 +448,13 @@ Context::FindValue(absl::string_view name, Protobuf::Arena* arena) const {
   case PropertyToken::REQUEST:
     if (info) {
       return CelValue::CreateMap(Protobuf::Arena::Create<Filters::Common::Expr::RequestWrapper>(
-          arena, request_headers_ ? request_headers_ : access_log_request_headers_, *info));
+          arena, *arena, request_headers_ ? request_headers_ : access_log_request_headers_, *info));
     }
     break;
   case PropertyToken::RESPONSE:
     if (info) {
       return CelValue::CreateMap(Protobuf::Arena::Create<Filters::Common::Expr::ResponseWrapper>(
-          arena, response_headers_ ? response_headers_ : access_log_response_headers_,
+          arena, *arena, response_headers_ ? response_headers_ : access_log_response_headers_,
           response_trailers_ ? response_trailers_ : access_log_response_trailers_, *info));
     }
     break;
diff --git a/source/extensions/filters/common/expr/BUILD b/source/extensions/filters/common/expr/BUILD
index 316c36b05b19..052581fea78b 100644
--- a/source/extensions/filters/common/expr/BUILD
+++ b/source/extensions/filters/common/expr/BUILD
@@ -1,11 +1,11 @@
-licenses(["notice"])  # Apache 2
-
 load(
     "//bazel:envoy_build_system.bzl",
     "envoy_cc_library",
     "envoy_package",
 )
 
+licenses(["notice"])  # Apache 2
+
 envoy_package()
 
 envoy_cc_library(
@@ -30,6 +30,7 @@ envoy_cc_library(
     deps = [
         "//source/common/grpc:common_lib",
         "//source/common/http:header_map_lib",
+        "//source/common/http:header_utility_lib",
         "//source/common/http:utility_lib",
         "//source/common/stream_info:utility_lib",
         "@com_google_cel_cpp//eval/public:cel_value",
diff --git a/source/extensions/filters/common/expr/context.cc b/source/extensions/filters/common/expr/context.cc
index ac5338cb86e9..d373f8c5545b 100644
--- a/source/extensions/filters/common/expr/context.cc
+++ b/source/extensions/filters/common/expr/context.cc
@@ -22,6 +22,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());
+  }
+}
+
 absl::optional<CelValue> extractSslInfo(const Ssl::ConnectionInfo& ssl_info,
                                         absl::string_view value) {
   if (value == TLSVersion) {
@@ -56,8 +69,9 @@ absl::optional<CelValue> HeadersWrapper::operator[](CelValue key) const {
   if (value_ == nullptr || !key.IsString()) {
     return {};
   }
-  auto out = value_->get(Http::LowerCaseString(std::string(key.StringOrDie().value())));
-  return convertHeaderEntry(out);
+  return convertHeaderEntry(
+      arena_, Http::HeaderUtility::getAllOfHeaderAsString(
+                  *value_, Http::LowerCaseString(std::string(key.StringOrDie().value()))));
 }
 
 absl::optional<CelValue> RequestWrapper::operator[](CelValue key) const {
diff --git a/source/extensions/filters/common/expr/context.h b/source/extensions/filters/common/expr/context.h
index a1eb7e8d4282..4797cea2a6cc 100644
--- a/source/extensions/filters/common/expr/context.h
+++ b/source/extensions/filters/common/expr/context.h
@@ -3,6 +3,7 @@
 #include "envoy/config/core/v3/base.pb.h"
 #include "envoy/stream_info/stream_info.h"
 
+#include "common/http/header_utility.h"
 #include "common/grpc/status.h"
 #include "common/http/headers.h"
 
@@ -71,7 +72,8 @@ class RequestWrapper;
 
 class HeadersWrapper : public google::api::expr::runtime::CelMap {
 public:
-  HeadersWrapper(const Http::HeaderMap* value) : value_(value) {}
+  HeadersWrapper(Protobuf::Arena& arena, const Http::HeaderMap* value)
+      : arena_(arena), value_(value) {}
   absl::optional<CelValue> operator[](CelValue key) const override;
   int size() const override { return value_ == nullptr ? 0 : value_->size(); }
   bool empty() const override { return value_ == nullptr ? true : value_->empty(); }
@@ -82,6 +84,7 @@ class HeadersWrapper : public google::api::expr::runtime::CelMap {
 private:
   friend class RequestWrapper;
   friend class ResponseWrapper;
+  Protobuf::Arena& arena_;
   const Http::HeaderMap* value_;
 };
 
@@ -98,8 +101,9 @@ class BaseWrapper : public google::api::expr::runtime::CelMap,
 
 class RequestWrapper : public BaseWrapper {
 public:
-  RequestWrapper(const Http::HeaderMap* headers, const StreamInfo::StreamInfo& info)
-      : headers_(headers), info_(info) {}
+  RequestWrapper(Protobuf::Arena& arena, const Http::HeaderMap* headers,
+                 const StreamInfo::StreamInfo& info)
+      : headers_(arena, headers), info_(info) {}
   absl::optional<CelValue> operator[](CelValue key) const override;
 
 private:
@@ -109,9 +113,9 @@ class RequestWrapper : public BaseWrapper {
 
 class ResponseWrapper : public BaseWrapper {
 public:
-  ResponseWrapper(const Http::HeaderMap* headers, const Http::HeaderMap* trailers,
-                  const StreamInfo::StreamInfo& info)
-      : headers_(headers), trailers_(trailers), info_(info) {}
+  ResponseWrapper(Protobuf::Arena& arena, const Http::HeaderMap* headers,
+                  const Http::HeaderMap* 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 dff584d3ee9e..59aa9872dfb5 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::HeaderMap* request_headers,
                                const Http::HeaderMap* response_headers,
                                const Http::HeaderMap* 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));
@@ -65,13 +66,14 @@ ExpressionPtr createExpression(Builder& builder, const google::api::expr::v1alph
   return std::move(cel_expression_status.ValueOrDie());
 }
 
-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::HeaderMap* request_headers,
                                   const Http::HeaderMap* response_headers,
                                   const Http::HeaderMap* response_trailers) {
-  auto activation = createActivation(info, request_headers, response_headers, response_trailers);
-  auto eval_status = expr.Evaluate(*activation.get(), arena);
+  auto activation =
+      createActivation(arena, info, request_headers, response_headers, response_trailers);
+  auto eval_status = expr.Evaluate(*activation.get(), &arena);
   if (!eval_status.ok()) {
     return {};
   }
@@ -82,7 +84,7 @@ absl::optional<CelValue> evaluate(const Expression& expr, Protobuf::Arena* arena
 bool matches(const Expression& expr, const StreamInfo::StreamInfo& info,
              const Http::HeaderMap& 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 16cdada5953d..8decd4bb0b28 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::HeaderMap* request_headers,
                                const Http::HeaderMap* response_headers,
                                const Http::HeaderMap* 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::HeaderMap* request_headers,
                                   const Http::HeaderMap* response_headers,
diff --git a/source/extensions/filters/http/header_to_metadata/BUILD b/source/extensions/filters/http/header_to_metadata/BUILD
index 1c9cbed21a84..c98500e5f934 100644
--- a/source/extensions/filters/http/header_to_metadata/BUILD
+++ b/source/extensions/filters/http/header_to_metadata/BUILD
@@ -19,6 +19,8 @@ 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 81a939442422..44a07901958a 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
@@ -4,6 +4,8 @@
 
 #include "common/common/base64.h"
 #include "common/config/well_known_names.h"
+#include "common/http/header_utility.h"
+#include "common/http/utility.h"
 #include "common/protobuf/protobuf.h"
 
 #include "extensions/filters/http/well_known_names.h"
@@ -157,13 +159,12 @@ void HeaderToMetadataFilter::writeHeaderToMetadata(Http::HeaderMap& headers,
   for (const auto& rulePair : rules) {
     const auto& header = rulePair.first;
     const auto& rule = rulePair.second;
-    const Http::HeaderEntry* header_entry = headers.get(header);
+    const auto header_value = Http::HeaderUtility::getAllOfHeaderAsString(headers, header);
 
-    if (header_entry != nullptr && rule.has_on_header_present()) {
+    if (header_value.result().has_value() && rule.has_on_header_present()) {
       const auto& keyval = rule.on_header_present();
-      absl::string_view value = keyval.value().empty() ? header_entry->value().getStringView()
+      absl::string_view value = keyval.value().empty() ? header_value.result().value()
                                                        : absl::string_view(keyval.value());
-
       if (!value.empty()) {
         const auto& nspace = decideNamespace(keyval.metadata_namespace());
         addMetadata(structs_by_namespace, nspace, keyval.key(), value, keyval.type(),
@@ -175,7 +176,8 @@ void HeaderToMetadataFilter::writeHeaderToMetadata(Http::HeaderMap& headers,
       if (rule.remove()) {
         headers.remove(header);
       }
-    } else if (rule.has_on_header_missing()) {
+    }
+    if (!header_value.result().has_value() && rule.has_on_header_missing()) {
       // Add metadata for the header missing case.
       const auto& keyval = rule.on_header_missing();
 
diff --git a/source/extensions/filters/http/jwt_authn/BUILD b/source/extensions/filters/http/jwt_authn/BUILD
index 462d1b193db5..5260e579d3bb 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 2ff31f30051a..c17bad423290 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"
@@ -186,9 +187,10 @@ std::vector<JwtLocationConstPtr> ExtractorImpl::extract(const Http::HeaderMap& h
   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 3a4be3465226..a084db5c40d6 100644
--- a/source/extensions/filters/http/lua/BUILD
+++ b/source/extensions/filters/http/lua/BUILD
@@ -39,6 +39,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 a772f3c1edfe..87b672092478 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"
@@ -45,10 +46,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 593d22653e68..fa0d932f614e 100644
--- a/test/common/http/BUILD
+++ b/test/common/http/BUILD
@@ -255,6 +255,7 @@ envoy_cc_test(
     deps = [
         "//source/common/http:header_map_lib",
         "//source/common/http:header_utility_lib",
+        "//test/test_common:test_runtime_lib",
         "//test/test_common:utility_lib",
     ],
 )
@@ -291,6 +292,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 74f2d225e308..7858bf82e876 100644
--- a/test/common/http/header_map_impl_test.cc
+++ b/test/common/http/header_map_impl_test.cc
@@ -5,6 +5,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"
@@ -115,6 +116,37 @@ TEST(HeaderStringTest, All) {
     EXPECT_EQ("HELLO", string.getStringView());
   }
 
+  // Inline rtrim removes trailing whitespace only.
+  {
+    // This header string is short enough to fit into Inline type.
+    const std::string data_with_leading_lws = " \t\f\v  data";
+    const std::string data_with_leading_and_trailing_lws = data_with_leading_lws + " \t\f\v";
+    HeaderString string;
+    string.append(data_with_leading_and_trailing_lws.data(),
+                  data_with_leading_and_trailing_lws.size());
+    EXPECT_EQ(data_with_leading_and_trailing_lws, string.getStringView());
+    EXPECT_EQ(string.type(), HeaderString::Type::Inline);
+    string.rtrim();
+    EXPECT_NE(data_with_leading_and_trailing_lws, string.getStringView());
+    EXPECT_EQ(data_with_leading_lws, string.getStringView());
+  }
+
+  // Dynamic rtrim removes trailing whitespace only.
+  {
+    // Making this string longer than Inline can fit.
+    const std::string padding_data_with_leading_lws = " \t\f\v  data" + std::string(128, 'a');
+    const std::string data_with_leading_and_trailing_lws =
+        padding_data_with_leading_lws + " \t\f\v";
+    HeaderString string;
+    string.append(data_with_leading_and_trailing_lws.data(),
+                  data_with_leading_and_trailing_lws.size());
+    EXPECT_EQ(data_with_leading_and_trailing_lws, string.getStringView());
+    EXPECT_EQ(string.type(), HeaderString::Type::Dynamic);
+    string.rtrim();
+    EXPECT_NE(data_with_leading_and_trailing_lws, string.getStringView());
+    EXPECT_EQ(padding_data_with_leading_lws, string.getStringView());
+  }
+
   // Static clear() does nothing.
   {
     std::string static_string("HELLO");
@@ -660,8 +692,12 @@ TEST(HeaderMapImplTest, SetReferenceKey) {
   EXPECT_EQ("monde", headers.get(foo)->value().getStringView());
 }
 
-TEST(HeaderMapImplTest, SetCopy) {
-  VerifiedHeaderMapImpl headers;
+TEST(HeaderMapImplTest, SetCopyOldBehavior) {
+  TestScopedRuntime runtime;
+  Runtime::LoaderSingleton::getExisting()->mergeValues(
+      {{"envoy.reloadable_features.http_set_copy_replace_all_headers", "false"}});
+
+  TestHeaderMapImpl headers;
   LowerCaseString foo("hello");
   headers.setCopy(foo, "world");
   EXPECT_EQ("world", headers.get(foo)->value().getStringView());
@@ -712,6 +748,58 @@ TEST(HeaderMapImplTest, SetCopy) {
   EXPECT_EQ(headers.Path()->value().getStringView(), "/foo");
 }
 
+TEST(HeaderMapImplTest, SetCopyNewBehavior) {
+  TestHeaderMapImpl 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);
+
+  using MockCb = testing::MockFunction<void(const std::string&, const std::string&)>;
+  MockCb cb;
+
+  InSequence seq;
+  EXPECT_CALL(cb, Call("hello", "override-monde"));
+  headers.iterate(
+      [](const Http::HeaderEntry& header, void* cb_v) -> HeaderMap::Iterate {
+        static_cast<MockCb*>(cb_v)->Call(std::string(header.key().getStringView()),
+                                         std::string(header.value().getStringView()));
+        return HeaderMap::Iterate::Continue;
+      },
+      &cb);
+
+  // Test setting an empty string and then overriding.
+  EXPECT_EQ(headers.size(), 1);
+  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.Path()->value(), "/");
+  headers.setPath("/foo");
+  EXPECT_EQ(headers.size(), 1);
+  EXPECT_EQ(headers.Path()->value(), "/foo");
+}
+
 TEST(HeaderMapImplTest, AddCopy) {
   VerifiedHeaderMapImpl headers;
 
diff --git a/test/common/http/header_utility_test.cc b/test/common/http/header_utility_test.cc
index 55d2056aab78..5124643d6db8 100644
--- a/test/common/http/header_utility_test.cc
+++ b/test/common/http/header_utility_test.cc
@@ -7,6 +7,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"
@@ -20,6 +21,65 @@ envoy::config::route::v3::HeaderMatcher parseHeaderMatcherFromYaml(const std::st
   return header_matcher;
 }
 
+class HeaderUtilityTest : public testing::Test {
+public:
+  const HeaderEntry& hostHeaderEntry(const std::string& host_value, bool set_connect = false) {
+    headers_.setHost(host_value);
+    if (set_connect) {
+      headers_.setMethod(Http::Headers::get().MethodValues.Connect);
+    }
+    return *headers_.Host();
+  }
+  TestHeaderMapImpl headers_;
+};
+
+TEST(GetAllOfHeaderAsStringTest, All) {
+  const LowerCaseString test_header("test");
+  {
+    TestHeaderMapImpl headers;
+    const auto ret = HeaderUtility::getAllOfHeaderAsString(headers, test_header);
+    EXPECT_FALSE(ret.result().has_value());
+    EXPECT_TRUE(ret.backingString().empty());
+  }
+  {
+    TestHeaderMapImpl headers{{"test", "foo"}};
+    const auto ret = HeaderUtility::getAllOfHeaderAsString(headers, test_header);
+    EXPECT_EQ("foo", ret.result().value());
+    EXPECT_TRUE(ret.backingString().empty());
+  }
+  {
+    TestHeaderMapImpl 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());
+  }
+  {
+    TestHeaderMapImpl headers{{"test", ""}, {"test", "bar"}};
+    const auto ret = HeaderUtility::getAllOfHeaderAsString(headers, test_header);
+    EXPECT_EQ(",bar", ret.result().value());
+    EXPECT_EQ(",bar", ret.backingString());
+  }
+  {
+    TestHeaderMapImpl headers{{"test", ""}, {"test", ""}};
+    const auto ret = HeaderUtility::getAllOfHeaderAsString(headers, test_header);
+    EXPECT_EQ(",", ret.result().value());
+    EXPECT_EQ(",", ret.backingString());
+  }
+  {
+    TestHeaderMapImpl 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
@@ -170,8 +230,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/common/http/http1/codec_impl_test.cc b/test/common/http/http1/codec_impl_test.cc
index e637094a23cf..12291a878b3e 100644
--- a/test/common/http/http1/codec_impl_test.cc
+++ b/test/common/http/http1/codec_impl_test.cc
@@ -43,6 +43,16 @@ std::string createHeaderFragment(int num_headers) {
   }
   return headers;
 }
+
+Buffer::OwnedImpl createBufferWithNByteSlices(absl::string_view input, size_t max_slice_size) {
+  Buffer::OwnedImpl buffer;
+  for (size_t offset = 0; offset < input.size(); offset += max_slice_size) {
+    buffer.appendSliceForTest(input.substr(offset, max_slice_size));
+  }
+  // Verify that the buffer contains the right number of slices.
+  ASSERT(buffer.getRawSlices(nullptr, 0) == (input.size() + max_slice_size - 1) / max_slice_size);
+  return buffer;
+}
 } // namespace
 
 class Http1ServerConnectionImplTest : public testing::Test {
@@ -72,7 +82,13 @@ class Http1ServerConnectionImplTest : public testing::Test {
   // Then send a response just to clean up.
   void sendAndValidateRequestAndSendResponse(absl::string_view raw_request,
                                              const TestHeaderMapImpl& expected_request_headers) {
-    NiceMock<Http::MockStreamDecoder> decoder;
+    Buffer::OwnedImpl buffer(raw_request);
+    sendAndValidateRequestAndSendResponse(buffer, expected_request_headers);
+  }
+
+  void sendAndValidateRequestAndSendResponse(Buffer::Instance& buffer,
+                                             const TestHeaderMapImpl& expected_request_headers) {
+    NiceMock<MockStreamDecoder> decoder;
     Http::StreamEncoder* response_encoder = nullptr;
     EXPECT_CALL(callbacks_, newStream(_, _))
         .Times(1)
@@ -80,8 +96,7 @@ class Http1ServerConnectionImplTest : public testing::Test {
           response_encoder = &encoder;
           return decoder;
         }));
-    EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_request_headers), true)).Times(1);
-    Buffer::OwnedImpl buffer(raw_request);
+    EXPECT_CALL(decoder, decodeHeaders_(HeaderMapEqual(&expected_request_headers), true));
     codec_->dispatch(buffer);
     EXPECT_EQ(0U, buffer.length());
     response_encoder->encodeHeaders(TestHeaderMapImpl{{":status", "200"}}, true);
@@ -377,6 +392,42 @@ TEST_F(Http1ServerConnectionImplTest, HostWithLWS) {
       "GET / HTTP/1.1\r\nHost: 	 	  host		  	 \r\n\r\n", expected_headers);
 }
 
+// Regression test for https://github.com/envoyproxy/envoy/issues/10270. Linear whitespace at the
+// beginning and end of a header value should be stripped. Whitespace in the middle should be
+// preserved.
+TEST_F(Http1ServerConnectionImplTest, InnerLWSIsPreserved) {
+  initialize();
+
+  // Header with many spaces surrounded by non-whitespace characters to ensure that dispatching is
+  // split across multiple dispatch calls. The threshold used here comes from Envoy preferring 16KB
+  // reads, but the important part is that the header value is split such that the pieces have
+  // leading and trailing whitespace characters.
+  const std::string header_value_with_inner_lws = "v" + std::string(32 * 1024, ' ') + "v";
+  TestHeaderMapImpl expected_headers{{":authority", "host"},
+                                     {":path", "/"},
+                                     {":method", "GET"},
+                                     {"header_field", header_value_with_inner_lws}};
+
+  {
+    // Regression test spaces in the middle are preserved
+    Buffer::OwnedImpl header_buffer = createBufferWithNByteSlices(
+        "GET / HTTP/1.1\r\nHost: host\r\nheader_field: " + header_value_with_inner_lws + "\r\n\r\n",
+        16 * 1024);
+    EXPECT_EQ(3, header_buffer.getRawSlices(nullptr, 0));
+    sendAndValidateRequestAndSendResponse(header_buffer, expected_headers);
+  }
+
+  {
+    // Regression test spaces before and after are removed
+    Buffer::OwnedImpl header_buffer = createBufferWithNByteSlices(
+        "GET / HTTP/1.1\r\nHost: host\r\nheader_field:  " + header_value_with_inner_lws +
+            "  \r\n\r\n",
+        16 * 1024);
+    EXPECT_EQ(3, header_buffer.getRawSlices(nullptr, 0));
+    sendAndValidateRequestAndSendResponse(header_buffer, expected_headers);
+  }
+}
+
 TEST_F(Http1ServerConnectionImplTest, Http10) {
   initialize();
 
diff --git a/test/extensions/common/wasm/BUILD b/test/extensions/common/wasm/BUILD
index 711f3e984a39..4d595f983caf 100644
--- a/test/extensions/common/wasm/BUILD
+++ b/test/extensions/common/wasm/BUILD
@@ -14,6 +14,9 @@ envoy_cc_test(
     data = [
         "//test/extensions/common/wasm/test_data:modules",
     ],
+    tags = [
+        "manual",
+    ],
     deps = [
         "//source/extensions/common/wasm:wasm_vm_lib",
         "//test/test_common:environment_lib",
diff --git a/test/extensions/filters/common/expr/context_test.cc b/test/extensions/filters/common/expr/context_test.cc
index 549bdd842a53..2317227e294a 100644
--- a/test/extensions/filters/common/expr/context_test.cc
+++ b/test/extensions/filters/common/expr/context_test.cc
@@ -23,7 +23,8 @@ namespace {
 constexpr absl::string_view Undefined = "undefined";
 
 TEST(Context, EmptyHeadersAttributes) {
-  HeadersWrapper headers(nullptr);
+  Protobuf::Arena arena;
+  HeadersWrapper headers(arena, nullptr);
   auto header = headers[CelValue::CreateStringView(Referer)];
   EXPECT_FALSE(header.has_value());
   EXPECT_EQ(0, headers.size());
@@ -33,13 +34,14 @@ TEST(Context, EmptyHeadersAttributes) {
 TEST(Context, RequestAttributes) {
   NiceMock<StreamInfo::MockStreamInfo> info;
   NiceMock<StreamInfo::MockStreamInfo> empty_info;
-  Http::TestHeaderMapImpl 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);
+  Http::TestHeaderMapImpl 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"},
+                                     {"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".
@@ -136,7 +138,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());
   }
 
   {
@@ -160,12 +162,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());
   }
 
   {
@@ -200,7 +207,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));
 
@@ -227,8 +235,9 @@ TEST(Context, ResponseAttributes) {
   const std::string grpc_status = "grpc-status";
   Http::TestHeaderMapImpl header_map{{header_name, "a"}};
   Http::TestHeaderMapImpl 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));
@@ -325,7 +334,7 @@ TEST(Context, ResponseAttributes) {
   {
     Http::TestHeaderMapImpl header_map{{header_name, "a"}, {grpc_status, "7"}};
     Http::TestHeaderMapImpl trailer_map{{trailer_name, "b"}};
-    ResponseWrapper response_header_status(&header_map, &trailer_map, info);
+    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());
@@ -334,7 +343,7 @@ TEST(Context, ResponseAttributes) {
   {
     Http::TestHeaderMapImpl header_map{{header_name, "a"}};
     Http::TestHeaderMapImpl trailer_map{{trailer_name, "b"}};
-    ResponseWrapper response_no_status(&header_map, &trailer_map, info);
+    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());
@@ -344,7 +353,7 @@ TEST(Context, ResponseAttributes) {
     NiceMock<StreamInfo::MockStreamInfo> info_without_code;
     Http::TestHeaderMapImpl header_map{{header_name, "a"}};
     Http::TestHeaderMapImpl trailer_map{{trailer_name, "b"}};
-    ResponseWrapper response_no_status(&header_map, &trailer_map, info_without_code);
+    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 f48bc37589f8..fdfc38c7711f 100644
--- a/test/extensions/filters/common/expr/evaluator_fuzz_test.cc
+++ b/test/extensions/filters/common/expr/evaluator_fuzz_test.cc
@@ -42,7 +42,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 abfb361a8f70..a0a49eff6c97 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
@@ -100,6 +100,24 @@ TEST_F(HeaderToMetadataTest, BasicRequestTest) {
   filter_->onDestroy();
 }
 
+// Verify concatenation works.
+TEST_F(HeaderToMetadataTest, BasicRequestDoubleHeadersTest) {
+  initializeFilter(request_config_yaml);
+  Http::TestHeaderMapImpl 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::TestHeaderMapImpl incoming_trailers;
+  EXPECT_EQ(Http::FilterTrailersStatus::Continue, filter_->decodeTrailers(incoming_trailers));
+  filter_->onDestroy();
+}
+
 /**
  * X-version not set, the on missing value should be set.
  */
@@ -404,6 +422,27 @@ TEST_F(HeaderToMetadataTest, NoEmptyValues) {
   EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_->decodeHeaders(headers, false));
 }
 
+/**
+ * Missing case is not executed when header is present.
+ */
+TEST_F(HeaderToMetadataTest, NoMissingWhenHeaderIsPresent) {
+  const std::string config = R"EOF(
+request_rules:
+  - header: x-version
+    on_header_missing:
+      metadata_namespace: envoy.lb
+      key: version
+      value: some_value
+      type: STRING
+)EOF";
+  initializeFilter(config);
+  Http::TestHeaderMapImpl headers{{"x-version", "19"}};
+
+  EXPECT_CALL(decoder_callbacks_, streamInfo()).WillRepeatedly(ReturnRef(req_info_));
+  EXPECT_CALL(req_info_, setDynamicMetadata(_, _)).Times(0);
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_->decodeHeaders(headers, false));
+}
+
 } // namespace
 } // namespace HeaderToMetadataFilter
 } // namespace HttpFilters
diff --git a/test/extensions/filters/http/jwt_authn/extractor_test.cc b/test/extensions/filters/http/jwt_authn/extractor_test.cc
index 5fbe1a84f4bd..ab943e217fec 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 = TestHeaderMapImpl{{"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 32c003951e1f..56ed5b3440de 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(*this, testPrint("'header2' 'foo'"));
   EXPECT_CALL(*this, testPrint("'hello' 'WORLD'"));
   EXPECT_CALL(*this, testPrint("'header2' 'foo'"));
+  EXPECT_CALL(*this, 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 23a4609ab440..273f28aedbf9 100644
--- a/test/extensions/filters/http/rbac/rbac_filter_integration_test.cc
+++ b/test/extensions/filters/http/rbac/rbac_filter_integration_test.cc
@@ -64,6 +64,33 @@ name: envoy.filters.http.rbac
           - any: true
 )EOF";
 
+const std::string RBAC_CONFIG_HEADER_MATCH_CONDITION = R"EOF(
+name: envoy.filters.http.rbac
+typed_config:
+  "@type": type.googleapis.com/envoy.config.filter.http.rbac.v2.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,
@@ -277,5 +304,78 @@ TEST_P(RBACIntegrationTest, PathIgnoreCase) {
   }
 }
 
+// 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::TestHeaderMapImpl{
+          {":method", "POST"},
+          {":path", "/path"},
+          {":scheme", "http"},
+          {":authority", "host"},
+          {"xxx", "yyy"},
+      },
+      1024);
+  waitForNextUpstreamRequest();
+  upstream_request_->encodeHeaders(Http::TestHeaderMapImpl{{":status", "200"}}, true);
+
+  response->waitForEndStream();
+  ASSERT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().Status()->value().getStringView());
+}
+
+// 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::TestHeaderMapImpl{
+          {":method", "POST"},
+          {":path", "/path"},
+          {":scheme", "http"},
+          {":authority", "host"},
+          {"xxx", "yyy"},
+          {"xxx", "zzz"},
+      },
+      1024);
+  response->waitForEndStream();
+  ASSERT_TRUE(response->complete());
+  EXPECT_EQ("403", response->headers().Status()->value().getStringView());
+}
+
+// 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::TestHeaderMapImpl{
+          {":method", "POST"},
+          {":path", "/path"},
+          {":scheme", "http"},
+          {":authority", "host"},
+          {"xxx", "yyy"},
+          {"xxx", "zzz"},
+      },
+      1024);
+  waitForNextUpstreamRequest();
+  upstream_request_->encodeHeaders(Http::TestHeaderMapImpl{{":status", "200"}}, true);
+
+  response->waitForEndStream();
+  ASSERT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().Status()->value().getStringView());
+}
+
 } // namespace
 } // namespace Envoy
diff --git a/test/integration/protocol_integration_test.cc b/test/integration/protocol_integration_test.cc
index e0d69f3774bc..e28d6b9a07bb 100644
--- a/test/integration/protocol_integration_test.cc
+++ b/test/integration/protocol_integration_test.cc
@@ -241,6 +241,39 @@ TEST_P(ProtocolIntegrationTest, DrainClose) {
   test_server_->drainManager().draining_ = false;
 }
 
+// Regression test for https://github.com/envoyproxy/envoy/issues/10270
+TEST_P(ProtocolIntegrationTest, LongHeaderValueWithSpaces) {
+  // Header with at least 20kb of spaces surrounded by non-whitespace characters to ensure that
+  // dispatching is split across 2 dispatch calls. This threshold comes from Envoy preferring 16KB
+  // reads, which the buffer rounds up to about 20KB when allocating slices in
+  // Buffer::OwnedImpl::reserve().
+  const std::string long_header_value_with_inner_lws = "v" + std::string(32 * 1024, ' ') + "v";
+
+  initialize();
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  auto response = codec_client_->makeHeaderOnlyRequest(
+      Http::TestHeaderMapImpl{{":method", "GET"},
+                              {":path", "/test/long/url"},
+                              {":scheme", "http"},
+                              {":authority", "host"},
+                              {"longrequestvalue", long_header_value_with_inner_lws}});
+  waitForNextUpstreamRequest();
+  EXPECT_EQ(long_header_value_with_inner_lws, upstream_request_->headers()
+                                                  .get(Http::LowerCaseString("longrequestvalue"))
+                                                  ->value()
+                                                  .getStringView());
+  upstream_request_->encodeHeaders(
+      Http::TestHeaderMapImpl{{":status", "200"},
+                              {"longresponsevalue", long_header_value_with_inner_lws}},
+      true);
+  response->waitForEndStream();
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().Status()->value().getStringView());
+  EXPECT_EQ(
+      long_header_value_with_inner_lws,
+      response->headers().get(Http::LowerCaseString("longresponsevalue"))->value().getStringView());
+}
+
 TEST_P(ProtocolIntegrationTest, Retry) {
   initialize();
   codec_client_ = makeHttpConnection(lookupPort("http"));