#include "source/common/http/filter_manager.h"

#include <functional>

#include "envoy/http/header_map.h"

#include "envoy/matcher/matcher.h"

#include "source/common/common/enum_to_int.h"

#include "source/common/common/scope_tracked_object_stack.h"

#include "source/common/common/scope_tracker.h"

#include "source/common/http/codes.h"

#include "source/common/http/header_map_impl.h"

#include "source/common/http/header_utility.h"

#include "source/common/http/utility.h"

#include "source/common/runtime/runtime_features.h"

#include "matching/data_impl.h"

namespace Envoy {

namespace Http {

namespace {

// Shared helper for recording the latest filter used.

template <class Filters>

void recordLatestDataFilter(typename Filters::Iterator current_filter,

typename Filters::Element*& latest_filter, Filters& filters) {

// If this is the first time we're calling onData, just record the current filter.

if (latest_filter == nullptr) {

latest_filter = current_filter->get();

return;

}

// We want to keep this pointing at the latest filter in the filter list that has received the

// onData callback. To do so, we compare the current latest with the *previous* filter. If they

// match, then we must be processing a new filter for the first time. We omit this check if we're

// the first filter, since the above check handles that case.

//

// We compare against the previous filter to avoid multiple filter iterations from resetting the

// pointer: If we just set latest to current, then the first onData filter iteration would

// correctly iterate over the filters and set latest, but on subsequent onData iterations

// we'd start from the beginning again, potentially allowing filter N to modify the buffer even

// though filter M > N was the filter that inserted data into the buffer.

if (current_filter != filters.begin() && latest_filter == std::prev(current_filter)->get()) {

latest_filter = current_filter->get();

}

}

void finalizeHeaders(FilterManagerCallbacks& callbacks, StreamInfo::StreamInfo& stream_info,

ResponseHeaderMap& headers) {

const auto route = stream_info.route();

if (route && route->routeEntry() != nullptr) {

const Formatter::Context formatter_context{

callbacks.requestHeaders().ptr(), &headers, {}, {}, {}, &callbacks.activeSpan()};

route->routeEntry()->finalizeResponseHeaders(headers, formatter_context, stream_info);

}

}

} // namespace

void ActiveStreamFilterBase::commonContinue() {

if (!canContinue()) {

ENVOY_STREAM_LOG(trace, "cannot continue filter chain: filter={}", *this,

static_cast<const void*>(this));

return;

}

// Set ScopeTrackerScopeState if there's no existing crash context.

ScopeTrackedObjectStack encapsulated_object;

absl::optional<ScopeTrackerScopeState> state;

if (parent_.dispatcher_.trackedObjectStackIsEmpty()) {

restoreContextOnContinue(encapsulated_object);

state.emplace(&encapsulated_object, parent_.dispatcher_);

}

ENVOY_STREAM_LOG(trace, "continuing filter chain: filter={}", *this,

static_cast<const void*>(this));

ASSERT(!canIterate(),

"Attempting to continue iteration while the IterationState is already Continue");

// If iteration has stopped for all frame types, set iterate_from_current_filter_ to true so the

// filter iteration starts with the current filter instead of the next one.

if (stoppedAll()) {

iterate_from_current_filter_ = true;

}

allowIteration();

// Only resume with do1xxHeaders() if we've actually seen 1xx headers.

if (has1xxHeaders()) {

continued_1xx_headers_ = true;

do1xxHeaders();

// If the response headers have not yet come in, don't continue on with

// headers and body. doHeaders expects request headers to exist.

if (!parent_.filter_manager_callbacks_.responseHeaders()) {

return;

}

}

if (!canContinue()) {

ENVOY_STREAM_LOG(trace, "cannot continue filter chain: filter={}", *this,

static_cast<const void*>(this));

return;

}

// Make sure that we handle the zero byte data frame case. We make no effort to optimize this

// case in terms of merging it into a header only request/response. This could be done in the

// future.

if (!headers_continued_) {

headers_continued_ = true;

doHeaders(observedEndStream() && !bufferedData() && !hasTrailers());

}

if (!canContinue()) {

ENVOY_STREAM_LOG(trace, "cannot continue filter chain: filter={}", *this,

static_cast<const void*>(this));

return;

}

doMetadata();

if (!canContinue()) {

ENVOY_STREAM_LOG(trace, "cannot continue filter chain: filter={}", *this,

static_cast<const void*>(this));

return;

}

// It is possible for trailers to be added during doData(). doData() itself handles continuation

// of trailers for the non-continuation case. Thus, we must keep track of whether we had

// trailers prior to calling doData(). If we do, then we continue them here, otherwise we rely

// on doData() to do so.

const bool had_trailers_before_data = hasTrailers();

if (bufferedData()) {

doData(observedEndStream() && !had_trailers_before_data);

}

if (!canContinue()) {

ENVOY_STREAM_LOG(trace, "cannot continue filter chain: filter={}", *this,

static_cast<const void*>(this));

return;

}

if (had_trailers_before_data) {

doTrailers();

}

iterate_from_current_filter_ = false;

}

bool ActiveStreamFilterBase::commonHandleAfter1xxHeadersCallback(Filter1xxHeadersStatus status) {

ASSERT(parent_.state_.has_1xx_headers_);

ASSERT(!continued_1xx_headers_);

ASSERT(canIterate());

switch (status) {

case Filter1xxHeadersStatus::Continue:

continued_1xx_headers_ = true;

return true;

case Filter1xxHeadersStatus::StopIteration:

iteration_state_ = IterationState::StopSingleIteration;

return false;

}

PANIC_DUE_TO_CORRUPT_ENUM;

}

bool ActiveStreamFilterBase::commonHandleAfterHeadersCallback(FilterHeadersStatus status,

bool& end_stream) {

ASSERT(!headers_continued_);

ASSERT(canIterate());

switch (status) {

case FilterHeadersStatus::StopIteration:

iteration_state_ = IterationState::StopSingleIteration;

break;

case FilterHeadersStatus::StopAllIterationAndBuffer:

iteration_state_ = IterationState::StopAllBuffer;

break;

case FilterHeadersStatus::StopAllIterationAndWatermark:

iteration_state_ = IterationState::StopAllWatermark;

break;

case FilterHeadersStatus::ContinueAndDontEndStream:

end_stream = false;

headers_continued_ = true;

ENVOY_STREAM_LOG(debug, "converting to headers and body (body not available yet)", parent_);

break;

case FilterHeadersStatus::Continue:

headers_continued_ = true;

break;

}

handleMetadataAfterHeadersCallback();

if (stoppedAll() || status == FilterHeadersStatus::StopIteration) {

return false;

} else {

return true;

}

}

void ActiveStreamFilterBase::commonHandleBufferData(Buffer::Instance& provided_data) {

// The way we do buffering is a little complicated which is why we have this common function

// which is used for both encoding and decoding. When data first comes into our filter pipeline,

// we send it through. Any filter can choose to stop iteration and buffer or not. If we then

// continue iteration in the future, we use the buffered data. A future filter can stop and

// buffer again. In this case, since we are already operating on buffered data, we don't

// rebuffer, because we assume the filter has modified the buffer as it wishes in place.

if (bufferedData().get() != &provided_data) {

if (!bufferedData()) {

bufferedData() = createBuffer();

}

bufferedData()->move(provided_data);

}

}

bool ActiveStreamFilterBase::commonHandleAfterDataCallback(FilterDataStatus status,

Buffer::Instance& provided_data,

bool& buffer_was_streaming) {

if (status == FilterDataStatus::Continue) {

if (iteration_state_ == IterationState::StopSingleIteration) {

commonHandleBufferData(provided_data);

commonContinue();

return false;

} else {

ASSERT(headers_continued_);

}

} else {

iteration_state_ = IterationState::StopSingleIteration;

if (status == FilterDataStatus::StopIterationAndBuffer ||

status == FilterDataStatus::StopIterationAndWatermark) {

buffer_was_streaming = status == FilterDataStatus::StopIterationAndWatermark;

commonHandleBufferData(provided_data);

} else if (observedEndStream() && !hasTrailers() && !bufferedData() &&

// If the stream is destroyed, no need to handle the data buffer or trailers.

// This can occur if the filter calls sendLocalReply.

!parent_.state_.destroyed_) {

// If this filter is doing StopIterationNoBuffer and this stream is terminated with a zero

// byte data frame, we need to create an empty buffer to make sure that when commonContinue

// is called, the pipeline resumes with an empty data frame with end_stream = true

ASSERT(end_stream_);

bufferedData() = createBuffer();

}

return false;

}

return true;

}

bool ActiveStreamFilterBase::commonHandleAfterTrailersCallback(FilterTrailersStatus status) {

if (status == FilterTrailersStatus::Continue) {

if (iteration_state_ == IterationState::StopSingleIteration) {

commonContinue();

return false;

} else {

ASSERT(headers_continued_);

}

} else if (status == FilterTrailersStatus::StopIteration) {

if (canIterate()) {

iteration_state_ = IterationState::StopSingleIteration;

}

return false;

}

return true;

}

OptRef<const Network::Connection> ActiveStreamFilterBase::connection() {

return parent_.connection();

}

Event::Dispatcher& ActiveStreamFilterBase::dispatcher() { return parent_.dispatcher_; }

StreamInfo::StreamInfo& ActiveStreamFilterBase::streamInfo() { return parent_.streamInfo(); }

Tracing::Span& ActiveStreamFilterBase::activeSpan() {

return parent_.filter_manager_callbacks_.activeSpan();

}

const ScopeTrackedObject& ActiveStreamFilterBase::scope() {

return parent_.filter_manager_callbacks_.scope();

}

void ActiveStreamFilterBase::restoreContextOnContinue(

ScopeTrackedObjectStack& tracked_object_stack) {

parent_.contextOnContinue(tracked_object_stack);

}

OptRef<const Tracing::Config> ActiveStreamFilterBase::tracingConfig() const {

return parent_.filter_manager_callbacks_.tracingConfig();

}

OptRef<const Upstream::ClusterInfo> ActiveStreamFilterBase::clusterInfo() {

return parent_.filter_manager_callbacks_.clusterInfo();

}

Upstream::ClusterInfoConstSharedPtr ActiveStreamFilterBase::clusterInfoSharedPtr() {

return parent_.filter_manager_callbacks_.clusterInfoSharedPtr();

}

OptRef<const Router::Route> ActiveStreamFilterBase::route() { return getRoute(); }

OptRef<const Router::Route> ActiveStreamFilterBase::getRoute() const {

if (parent_.filter_manager_callbacks_.downstreamCallbacks()) {

return parent_.filter_manager_callbacks_.downstreamCallbacks()->route(nullptr);

}

return parent_.streamInfo().route();

}

Router::RouteConstSharedPtr ActiveStreamFilterBase::routeSharedPtr() { return getRouteSharedPtr(); }

Router::RouteConstSharedPtr ActiveStreamFilterBase::getRouteSharedPtr() const {

if (parent_.filter_manager_callbacks_.downstreamCallbacks()) {

return parent_.filter_manager_callbacks_.downstreamCallbacks()->routeSharedPtr(nullptr);

}

return parent_.streamInfo().routeSharedPtr();

}

void ActiveStreamFilterBase::resetIdleTimer() {

parent_.filter_manager_callbacks_.resetIdleTimer();

}

const Router::RouteSpecificFilterConfig*

ActiveStreamFilterBase::mostSpecificPerFilterConfig() const {

const auto current_route = getRoute();

if (!current_route) {

return nullptr;

}

return current_route->mostSpecificPerFilterConfig(filter_context_.config_name);

}

Router::RouteSpecificFilterConfigs ActiveStreamFilterBase::perFilterConfigs() const {

const auto current_route = getRoute();

if (!current_route) {

return {};

}

return current_route->perFilterConfigs(filter_context_.config_name);

}

Http1StreamEncoderOptionsOptRef ActiveStreamFilterBase::http1StreamEncoderOptions() {

// TODO(mattklein123): At some point we might want to actually wrap this interface but for now

// we give the filter direct access to the encoder options.

return parent_.filter_manager_callbacks_.http1StreamEncoderOptions();

}

OptRef<DownstreamStreamFilterCallbacks> ActiveStreamFilterBase::downstreamCallbacks() {

return parent_.filter_manager_callbacks_.downstreamCallbacks();

}

OptRef<UpstreamStreamFilterCallbacks> ActiveStreamFilterBase::upstreamCallbacks() {

return parent_.filter_manager_callbacks_.upstreamCallbacks();

}

RequestHeaderMapOptRef ActiveStreamFilterBase::requestHeaders() {

return parent_.filter_manager_callbacks_.requestHeaders();

}

RequestTrailerMapOptRef ActiveStreamFilterBase::requestTrailers() {

return parent_.filter_manager_callbacks_.requestTrailers();

}

ResponseHeaderMapOptRef ActiveStreamFilterBase::informationalHeaders() {

return parent_.filter_manager_callbacks_.informationalHeaders();

}

ResponseHeaderMapOptRef ActiveStreamFilterBase::responseHeaders() {

return parent_.filter_manager_callbacks_.responseHeaders();

}

ResponseTrailerMapOptRef ActiveStreamFilterBase::responseTrailers() {

return parent_.filter_manager_callbacks_.responseTrailers();

}

void ActiveStreamFilterBase::setBufferLimit(uint64_t limit) { parent_.setBufferLimit(limit); }

uint64_t ActiveStreamFilterBase::bufferLimit() { return parent_.buffer_limit_; }

void ActiveStreamFilterBase::sendLocalReply(

Code code, absl::string_view body,

std::function<void(ResponseHeaderMap& headers)> modify_headers,

const absl::optional<Grpc::Status::GrpcStatus> grpc_status, absl::string_view details) {

if (!streamInfo().filterState()->hasData<LocalReplyOwnerObject>(LocalReplyFilterStateKey)) {

streamInfo().filterState()->setData(

LocalReplyFilterStateKey,

std::make_shared<LocalReplyOwnerObject>(filter_context_.config_name),

StreamInfo::FilterState::StateType::ReadOnly,

StreamInfo::FilterState::LifeSpan::FilterChain);

}

parent_.sendLocalReply(code, body, modify_headers, grpc_status, details);

}

bool ActiveStreamDecoderFilter::canContinue() {

// It is possible for the connection manager to respond directly to a request even while

// a filter is trying to continue. If a response has already happened, we should not

// continue to further filters. A concrete example of this is a filter buffering data, the

// last data frame comes in and the filter continues, but the final buffering takes the stream

// over the high watermark such that a 413 is returned.

return !parent_.stopDecoderFilterChain();

}

bool ActiveStreamEncoderFilter::canContinue() {

// As with ActiveStreamDecoderFilter::canContinue() make sure we do not

// continue if a local reply has been sent or ActiveStreamDecoderFilter::recreateStream() is

// called, etc.

return !parent_.state_.encoder_filter_chain_complete_ && !parent_.stopEncoderFilterChain();

}

Buffer::InstancePtr ActiveStreamDecoderFilter::createBuffer() {

auto buffer = dispatcher().getWatermarkFactory().createBuffer(

[this]() -> void { this->requestDataDrained(); },

[this]() -> void { this->requestDataTooLarge(); },

[]() -> void { /* TODO(adisuissa): Handle overflow watermark */ });

buffer->setWatermarks(parent_.buffer_limit_);

return buffer;

}

Buffer::InstancePtr& ActiveStreamDecoderFilter::bufferedData() {

return parent_.buffered_request_data_;

}

bool ActiveStreamDecoderFilter::observedEndStream() { return parent_.decoderObservedEndStream(); }

void ActiveStreamDecoderFilter::doHeaders(bool end_stream) {

parent_.decodeHeaders(this, *parent_.filter_manager_callbacks_.requestHeaders(), end_stream);

}

void ActiveStreamDecoderFilter::doData(bool end_stream) {

parent_.decodeData(this, *parent_.buffered_request_data_, end_stream,

FilterManager::FilterIterationStartState::CanStartFromCurrent);

}

void ActiveStreamDecoderFilter::doTrailers() {

parent_.decodeTrailers(this, *parent_.filter_manager_callbacks_.requestTrailers());

}

bool ActiveStreamDecoderFilter::hasTrailers() {

return parent_.filter_manager_callbacks_.requestTrailers().has_value();

}

void ActiveStreamDecoderFilter::drainSavedRequestMetadata() {

ASSERT(saved_request_metadata_ != nullptr);

for (auto& metadata_map : *getSavedRequestMetadata()) {

parent_.decodeMetadata(this, *metadata_map);

}

getSavedRequestMetadata()->clear();

}

void ActiveStreamDecoderFilter::handleMetadataAfterHeadersCallback() {

if (parent_.state_.decoder_filter_chain_aborted_) {

// The decoder filter chain has been aborted, possibly due to a local reply. In this case,

// there's no reason to decode saved metadata.

getSavedRequestMetadata()->clear();

return;

}

// If we drain accumulated metadata, the iteration must start with the current filter.

const bool saved_state = iterate_from_current_filter_;

iterate_from_current_filter_ = true;

// If decodeHeaders() returns StopAllIteration, we should skip draining metadata, and wait

// for doMetadata() to drain the metadata after iteration continues.

if (!stoppedAll() && saved_request_metadata_ != nullptr && !getSavedRequestMetadata()->empty()) {

drainSavedRequestMetadata();

}

// Restores the original value of iterate_from_current_filter_.

iterate_from_current_filter_ = saved_state;

}

RequestTrailerMap& ActiveStreamDecoderFilter::addDecodedTrailers() {

return parent_.addDecodedTrailers();

}

void ActiveStreamDecoderFilter::addDecodedData(Buffer::Instance& data, bool streaming) {

parent_.addDecodedData(*this, data, streaming);

}

MetadataMapVector& ActiveStreamDecoderFilter::addDecodedMetadata() {

return parent_.addDecodedMetadata();

}

void ActiveStreamDecoderFilter::injectDecodedDataToFilterChain(Buffer::Instance& data,

bool end_stream) {

if (!headers_continued_) {

headers_continued_ = true;

doHeaders(false);

}

if (Runtime::runtimeFeatureEnabled(

"envoy.reloadable_features.ext_proc_inject_data_with_state_update")) {

parent_.state().observed_decode_end_stream_ = end_stream;

ENVOY_STREAM_LOG(trace, "injectDecodedDataToFilterChain with end_stream updated: {}", parent_,

end_stream);

}

parent_.decodeData(this, data, end_stream,

FilterManager::FilterIterationStartState::CanStartFromCurrent);

}

void ActiveStreamDecoderFilter::continueDecoding() { commonContinue(); }

const Buffer::Instance* ActiveStreamDecoderFilter::decodingBuffer() {

return parent_.buffered_request_data_.get();

}

bool ActiveStreamDecoderFilter::shouldLoadShed() const { return parent_.shouldLoadShed(); }

void ActiveStreamDecoderFilter::modifyDecodingBuffer(

std::function<void(Buffer::Instance&)> callback) {

ASSERT(parent_.state_.latest_data_decoding_filter_ == this);

callback(*parent_.buffered_request_data_.get());

}

void ActiveStreamDecoderFilter::sendLocalReply(

Code code, absl::string_view body,

std::function<void(ResponseHeaderMap& headers)> modify_headers,

const absl::optional<Grpc::Status::GrpcStatus> grpc_status, absl::string_view details) {

ActiveStreamFilterBase::sendLocalReply(code, body, modify_headers, grpc_status, details);

}

void ActiveStreamDecoderFilter::sendGoAwayAndClose(bool graceful) {

parent_.sendGoAwayAndClose(graceful);

}

void ActiveStreamDecoderFilter::encode1xxHeaders(ResponseHeaderMapPtr&& headers) {

// If Envoy is not configured to proxy 100-Continue responses, swallow the 100 Continue

// here. This avoids the potential situation where Envoy strips Expect: 100-Continue and sends a

// 100-Continue, then proxies a duplicate 100 Continue from upstream.

if (parent_.proxy_100_continue_) {

parent_.filter_manager_callbacks_.setInformationalHeaders(std::move(headers));

parent_.encode1xxHeaders(nullptr, *parent_.filter_manager_callbacks_.informationalHeaders());

}

}

void ActiveStreamDecoderFilter::stopDecodingIfNonTerminalFilterEncodedEndStream(

bool encoded_end_stream) {

// Encoding end_stream by a non-terminal filters (i.e. cache filter) always causes the decoding to

// be stopped even if independent half-close is enabled. For simplicity, independent half-close is

// allowed only when the terminal (router) filter is encoding the response.

if (encoded_end_stream && !parent_.isTerminalDecoderFilter(*this) &&

!parent_.state_.decoder_filter_chain_complete_) {

parent_.state_.decoder_filter_chain_aborted_ = true;

}

}

void ActiveStreamDecoderFilter::encodeHeaders(ResponseHeaderMapPtr&& headers, bool end_stream,

absl::string_view details) {

stopDecodingIfNonTerminalFilterEncodedEndStream(end_stream);

parent_.streamInfo().setResponseCodeDetails(details);

parent_.filter_manager_callbacks_.setResponseHeaders(std::move(headers));

parent_.encodeHeaders(nullptr, *parent_.filter_manager_callbacks_.responseHeaders(), end_stream);

}

void ActiveStreamDecoderFilter::encodeData(Buffer::Instance& data, bool end_stream) {

stopDecodingIfNonTerminalFilterEncodedEndStream(end_stream);

parent_.encodeData(nullptr, data, end_stream,

FilterManager::FilterIterationStartState::CanStartFromCurrent);

}

void ActiveStreamDecoderFilter::encodeTrailers(ResponseTrailerMapPtr&& trailers) {

stopDecodingIfNonTerminalFilterEncodedEndStream(true);

parent_.filter_manager_callbacks_.setResponseTrailers(std::move(trailers));

parent_.encodeTrailers(nullptr, *parent_.filter_manager_callbacks_.responseTrailers());

}

void ActiveStreamDecoderFilter::encodeMetadata(MetadataMapPtr&& metadata_map_ptr) {

parent_.encodeMetadata(nullptr, std::move(metadata_map_ptr));

}

void ActiveStreamDecoderFilter::onDecoderFilterAboveWriteBufferHighWatermark() {

parent_.filter_manager_callbacks_.onDecoderFilterAboveWriteBufferHighWatermark();

}

void ActiveStreamDecoderFilter::requestDataTooLarge() {

ENVOY_STREAM_LOG(debug, "request data too large watermark exceeded", parent_);

if (parent_.state_.decoder_filters_streaming_) {

onDecoderFilterAboveWriteBufferHighWatermark();

} else {

parent_.filter_manager_callbacks_.onRequestDataTooLarge();

sendLocalReply(Code::PayloadTooLarge, CodeUtility::toString(Code::PayloadTooLarge), nullptr,

absl::nullopt, StreamInfo::ResponseCodeDetails::get().RequestPayloadTooLarge);

}

}

void FilterManager::maybeContinueDecoding(StreamDecoderFilters::Iterator continue_data_entry) {

if (continue_data_entry != decoder_filters_.end()) {

// We use the continueDecoding() code since it will correctly handle not calling

// decodeHeaders() again. Fake setting StopSingleIteration since the continueDecoding() code

// expects it.

ASSERT(buffered_request_data_);

(*continue_data_entry)->iteration_state_ =

ActiveStreamFilterBase::IterationState::StopSingleIteration;

(*continue_data_entry)->continueDecoding();

}

}

void FilterManager::decodeHeaders(ActiveStreamDecoderFilter* filter, RequestHeaderMap& headers,

bool end_stream) {

// If the stream has been reset, do not process any more frames.

if (stopDecoderFilterChain()) {

return;

}

// Headers filter iteration should always start with the next filter if available.

StreamDecoderFilters::Iterator entry =

commonDecodePrefix(filter, FilterIterationStartState::AlwaysStartFromNext);

StreamDecoderFilters::Iterator continue_data_entry = decoder_filters_.end();

bool terminal_filter_decoded_end_stream = false;

ASSERT(!state_.decoder_filter_chain_complete_ || entry == decoder_filters_.end() ||

(*entry)->end_stream_);

for (; entry != decoder_filters_.end(); entry++) {

ENVOY_EXECUTION_SCOPE(trackedStream(), &(*entry)->filter_context_);

ASSERT(!(state_.filter_call_state_ & FilterCallState::DecodeHeaders));

state_.filter_call_state_ |= FilterCallState::DecodeHeaders;

(*entry)->end_stream_ = (end_stream && continue_data_entry == decoder_filters_.end());

if ((*entry)->end_stream_) {

state_.filter_call_state_ |= FilterCallState::EndOfStream;

}

FilterHeadersStatus status = (*entry)->decodeHeaders(headers, (*entry)->end_stream_);

state_.filter_call_state_ &= ~FilterCallState::DecodeHeaders;

if ((*entry)->end_stream_) {

state_.filter_call_state_ &= ~FilterCallState::EndOfStream;

}

if (state_.decoder_filter_chain_aborted_) {

executeLocalReplyIfPrepared();

ENVOY_STREAM_LOG(trace,

"decodeHeaders filter iteration aborted due to local reply: filter={}",

*this, (*entry)->filter_context_.config_name);

status = FilterHeadersStatus::StopIteration;

}

ASSERT(!(status == FilterHeadersStatus::ContinueAndDontEndStream && !(*entry)->end_stream_),

"Filters should not return FilterHeadersStatus::ContinueAndDontEndStream from "

"decodeHeaders when end_stream is already false");

ENVOY_STREAM_LOG(trace, "decode headers called: filter={} status={}", *this,

(*entry)->filter_context_.config_name, static_cast<uint64_t>(status));

(*entry)->processed_headers_ = true;

const auto continue_iteration = (*entry)->commonHandleAfterHeadersCallback(status, end_stream);

ENVOY_BUG(!continue_iteration || !stopDecoderFilterChain(),

fmt::format(

"filter={} did not return StopAll or StopIteration after sending a local reply.",

(*entry)->filter_context_.config_name));

// If this filter ended the stream, decodeComplete() should be called for it.

if ((*entry)->end_stream_) {

(*entry)->handle_->decodeComplete();

}

// Skip processing metadata after sending local reply

if (stopDecoderFilterChain() && std::next(entry) != decoder_filters_.end()) {

maybeContinueDecoding(continue_data_entry);

return;

}

const bool new_metadata_added = processNewlyAddedMetadata();

// If end_stream is set in headers, and a filter adds new metadata, we need to delay end_stream

// in headers by inserting an empty data frame with end_stream set. The empty data frame is sent

// after the new metadata.

if ((*entry)->end_stream_ && new_metadata_added && !buffered_request_data_) {

Buffer::OwnedImpl empty_data("");

ENVOY_STREAM_LOG(

trace, "inserting an empty data frame for end_stream due metadata being added.", *this);

// Metadata frame doesn't carry end of stream bit. We need an empty data frame to end the

// stream.

addDecodedData(*(*entry), empty_data, true);

}

if (!continue_iteration && std::next(entry) != decoder_filters_.end()) {

// Stop iteration IFF this is not the last filter. If it is the last filter, continue with

// processing since we need to handle the case where a terminal filter wants to buffer, but

// a previous filter has added body.

maybeContinueDecoding(continue_data_entry);

return;

}

// Here we handle the case where we have a header only request, but a filter adds a body

// to it. We need to not raise end_stream = true to further filters during inline iteration.

if (end_stream && buffered_request_data_ && continue_data_entry == decoder_filters_.end()) {

continue_data_entry = entry;

}

// The decoder filter chain is finished here if the following is true:

// 1. the last filter has observed the end_stream AND

// 2. no filter, including the last filter, has injected body during header iteration.

// If body was injected the end_stream will be processed in the `decodeData()` method below.

const bool no_body_was_injected = continue_data_entry == decoder_filters_.end();

terminal_filter_decoded_end_stream =

(std::next(entry) == decoder_filters_.end() && (*entry)->end_stream_) &&

no_body_was_injected;

}

maybeContinueDecoding(continue_data_entry);

if (end_stream) {

disarmRequestTimeout();

}

maybeEndDecode(terminal_filter_decoded_end_stream);

}

void FilterManager::decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instance& data,

bool end_stream,

FilterIterationStartState filter_iteration_start_state) {

ScopeTrackerScopeState scope(this, dispatcher_);

filter_manager_callbacks_.resetIdleTimer();

// If a response is complete or a reset has been sent, filters do not care about further body

// data. Just drop it.

if (stopDecoderFilterChain()) {

return;

}

auto trailers_added_entry = decoder_filters_.end();

const bool trailers_exists_at_start = filter_manager_callbacks_.requestTrailers().has_value();

// Filter iteration may start at the current filter.

StreamDecoderFilters::Iterator entry = commonDecodePrefix(filter, filter_iteration_start_state);

bool terminal_filter_decoded_end_stream = false;

ASSERT(!state_.decoder_filter_chain_complete_ || entry == decoder_filters_.end() ||

(*entry)->end_stream_);

for (; entry != decoder_filters_.end(); entry++) {

ENVOY_EXECUTION_SCOPE(trackedStream(), &(*entry)->filter_context_);

// If the filter pointed by entry has stopped for all frame types, return now.

if (handleDataIfStopAll(**entry, data, state_.decoder_filters_streaming_)) {

return;

}

// If end_stream_ is marked for a filter, the data is not for this filter and filters after.

//

// In following case, ActiveStreamFilterBase::commonContinue() could be called recursively and

// its doData() is called with wrong data.

//

// There are 3 decode filters and "wrapper" refers to ActiveStreamFilter object.

//

// filter0->decodeHeaders(_, true)

// return STOP

// filter0->continueDecoding()

// wrapper0->commonContinue()

// wrapper0->decodeHeaders(_, _, true)

// filter1->decodeHeaders(_, true)

// filter1->addDecodeData()

// return CONTINUE

// filter2->decodeHeaders(_, false)

// return CONTINUE

// wrapper1->commonContinue() // Detects data is added.

// wrapper1->doData()

// wrapper1->decodeData()

// filter2->decodeData(_, true)

// return CONTINUE

// wrapper0->doData() // This should not be called

// wrapper0->decodeData()

// filter1->decodeData(_, true) // It will cause assertions.

//

// One way to solve this problem is to mark end_stream_ for each filter.

// If a filter is already marked as end_stream_ when decodeData() is called, bails out the

// whole function. If just skip the filter, the codes after the loop will be called with

// wrong data. For encodeData, the response_encoder->encode() will be called.

if ((*entry)->end_stream_) {

return;

}

ASSERT(!(state_.filter_call_state_ & FilterCallState::DecodeData));

// We check the request_trailers_ pointer here in case addDecodedTrailers

// is called in decodeData during a previous filter invocation, at which point we communicate to

// the current and future filters that the stream has not yet ended.

if (end_stream) {

state_.filter_call_state_ |= FilterCallState::EndOfStream;

}

recordLatestDataFilter(entry, state_.latest_data_decoding_filter_, decoder_filters_);

state_.filter_call_state_ |= FilterCallState::DecodeData;

(*entry)->end_stream_ = end_stream && !filter_manager_callbacks_.requestTrailers();

FilterDataStatus status = (*entry)->handle_->decodeData(data, (*entry)->end_stream_);

if ((*entry)->end_stream_) {

(*entry)->handle_->decodeComplete();

}

state_.filter_call_state_ &= ~FilterCallState::DecodeData;

if (end_stream) {

state_.filter_call_state_ &= ~FilterCallState::EndOfStream;

}

ENVOY_STREAM_LOG(trace, "decode data called: filter={} status={}", *this,

(*entry)->filter_context_.config_name, static_cast<uint64_t>(status));

if (state_.decoder_filter_chain_aborted_) {

executeLocalReplyIfPrepared();

ENVOY_STREAM_LOG(trace, "decodeData filter iteration aborted due to local reply: filter={}",

*this, (*entry)->filter_context_.config_name);

return;

}

processNewlyAddedMetadata();

if (!trailers_exists_at_start && filter_manager_callbacks_.requestTrailers() &&

trailers_added_entry == decoder_filters_.end()) {

end_stream = false;

trailers_added_entry = entry;

}

// The decoder filter chain is finished here if the following is true:

// 1. the last filter has observed the end_stream AND

// 2. no filter, including the last filter, has injected trailers during header iteration.

// NOTE: end_stream is set to false above if a filter injected trailers.

// If trailers were injected the end_stream will be processed in the `decodeTrailers()` method

// below.

terminal_filter_decoded_end_stream = end_stream && std::next(entry) == decoder_filters_.end();

if (!(*entry)->commonHandleAfterDataCallback(status, data, state_.decoder_filters_streaming_) &&

std::next(entry) != decoder_filters_.end()) {

// Stop iteration IFF this is not the last filter. If it is the last filter, continue with

// processing since we need to handle the case where a terminal filter wants to buffer, but

// a previous filter has added trailers.

break;

}

}

// If trailers were adding during decodeData we need to trigger decodeTrailers in order

// to allow filters to process the trailers.

if (trailers_added_entry != decoder_filters_.end()) {

decodeTrailers(trailers_added_entry->get(), *filter_manager_callbacks_.requestTrailers());

}

if (end_stream) {

disarmRequestTimeout();

}

maybeEndDecode(terminal_filter_decoded_end_stream);

}

RequestTrailerMap& FilterManager::addDecodedTrailers() {

// Trailers can only be added during the last data frame (i.e. end_stream = true).

ASSERT(state_.filter_call_state_ & FilterCallState::EndOfStream);

filter_manager_callbacks_.setRequestTrailers(RequestTrailerMapImpl::create());

return *filter_manager_callbacks_.requestTrailers();

}

void FilterManager::addDecodedData(ActiveStreamDecoderFilter& filter, Buffer::Instance& data,

bool streaming) {

if (state_.filter_call_state_ == 0 ||

(state_.filter_call_state_ & FilterCallState::DecodeHeaders) ||

(state_.filter_call_state_ & FilterCallState::DecodeData) ||

((state_.filter_call_state_ & FilterCallState::DecodeTrailers) && !filter.canIterate())) {

// Make sure if this triggers watermarks, the correct action is taken.

state_.decoder_filters_streaming_ = streaming;

// If no call is happening or we are in the decode headers/data callback, buffer the data.

// Inline processing happens in the decodeHeaders() callback if necessary.

filter.commonHandleBufferData(data);

} else if (state_.filter_call_state_ & FilterCallState::DecodeTrailers) {

// In this case we need to inline dispatch the data to further filters. If those filters

// choose to buffer/stop iteration that's fine.

decodeData(&filter, data, false, FilterIterationStartState::AlwaysStartFromNext);

} else {

IS_ENVOY_BUG("Invalid request data");

sendLocalReply(Http::Code::BadGateway, "Filter error", nullptr, absl::nullopt,

StreamInfo::ResponseCodeDetails::get().FilterAddedInvalidRequestData);

}

}

MetadataMapVector& FilterManager::addDecodedMetadata() { return *getRequestMetadataMapVector(); }

void FilterManager::decodeTrailers(ActiveStreamDecoderFilter* filter, RequestTrailerMap& trailers) {

// If a response is complete or a reset has been sent, filters do not care about further body

// data. Just drop it.

if (stopDecoderFilterChain()) {

return;

}

// Filter iteration may start at the current filter.

StreamDecoderFilters::Iterator entry =

commonDecodePrefix(filter, FilterIterationStartState::CanStartFromCurrent);

bool terminal_filter_reached = false;

ASSERT(!state_.decoder_filter_chain_complete_ || entry == decoder_filters_.end());

for (; entry != decoder_filters_.end(); entry++) {

ENVOY_EXECUTION_SCOPE(trackedStream(), &(*entry)->filter_context_);

// If the filter pointed by entry has stopped for all frame type, return now.

if ((*entry)->stoppedAll()) {

return;

}

ASSERT(!(state_.filter_call_state_ & FilterCallState::DecodeTrailers));

state_.filter_call_state_ |= FilterCallState::DecodeTrailers;

FilterTrailersStatus status = (*entry)->handle_->decodeTrailers(trailers);

(*entry)->handle_->decodeComplete();

(*entry)->end_stream_ = true;

state_.filter_call_state_ &= ~FilterCallState::DecodeTrailers;

ENVOY_STREAM_LOG(trace, "decode trailers called: filter={} status={}", *this,

(*entry)->filter_context_.config_name, static_cast<uint64_t>(status));

if (state_.decoder_filter_chain_aborted_) {

executeLocalReplyIfPrepared();

ENVOY_STREAM_LOG(trace,

"decodeTrailers filter iteration aborted due to local reply: filter={}",

*this, (*entry)->filter_context_.config_name);

status = FilterTrailersStatus::StopIteration;

}

processNewlyAddedMetadata();

// Check if the last filter has processed trailers

terminal_filter_reached = std::next(entry) == decoder_filters_.end();

if (!(*entry)->commonHandleAfterTrailersCallback(status) && !terminal_filter_reached) {

return;

}

}

disarmRequestTimeout();

maybeEndDecode(terminal_filter_reached);

}

void FilterManager::decodeMetadata(ActiveStreamDecoderFilter* filter, MetadataMap& metadata_map) {

ScopeTrackerScopeState scope(&*this, dispatcher_);

filter_manager_callbacks_.resetIdleTimer();

// If the stream has been reset, do not process any more frames.

if (stopDecoderFilterChain()) {

return;

}

// Filter iteration may start at the current filter.

StreamDecoderFilters::Iterator entry =

commonDecodePrefix(filter, FilterIterationStartState::CanStartFromCurrent);

ASSERT(!(state_.filter_call_state_ & FilterCallState::DecodeMetadata));

for (; entry != decoder_filters_.end(); entry++) {

ENVOY_EXECUTION_SCOPE(trackedStream(), &(*entry)->filter_context_);

// If the filter pointed by entry has stopped for all frame type, stores metadata and returns.

// If the filter pointed by entry hasn't returned from decodeHeaders, stores newly added

// metadata in case decodeHeaders returns StopAllIteration. The latter can happen when headers

// callbacks generate new metadata.

if (!(*entry)->processed_headers_ || (*entry)->stoppedAll()) {

Http::MetadataMapPtr metadata_map_ptr = std::make_unique<Http::MetadataMap>(metadata_map);

(*entry)->getSavedRequestMetadata()->emplace_back(std::move(metadata_map_ptr));

return;

}

state_.filter_call_state_ |= FilterCallState::DecodeMetadata;

FilterMetadataStatus status = (*entry)->handle_->decodeMetadata(metadata_map);

state_.filter_call_state_ &= ~FilterCallState::DecodeMetadata;

ENVOY_STREAM_LOG(trace, "decode metadata called: filter={} status={}, metadata: {}", *this,

(*entry)->filter_context_.config_name, static_cast<uint64_t>(status),

metadata_map);

if (state_.decoder_filter_chain_aborted_) {

// If the decoder filter chain has been aborted, then either:

// 1. This filter has sent a local reply or GoAway from decode metadata.

// 2. This filter is the terminal http filter, and an upstream HTTP filter has sent a local

// reply.

ASSERT((status == FilterMetadataStatus::StopIterationForLocalReply) ||

(std::next(entry) == decoder_filters_.end()));

executeLocalReplyIfPrepared();

ENVOY_STREAM_LOG(trace,

"decodeMetadata filter iteration aborted due to local reply: filter={}",

*this, (*entry)->filter_context_.config_name);

return;

}

// Add the processed metadata to the next entry.

if (status == FilterMetadataStatus::ContinueAll && !(*entry)->canIterate()) {

if (std::next(entry) != decoder_filters_.end()) {

(*std::next(entry))

->getSavedRequestMetadata()

->emplace_back(std::make_unique<MetadataMap>(metadata_map));

}

(*entry)->commonContinue();

return;

}

}

}

void FilterManager::disarmRequestTimeout() { filter_manager_callbacks_.disarmRequestTimeout(); }

StreamEncoderFilters::Iterator

FilterManager::commonEncodePrefix(ActiveStreamEncoderFilter* filter, bool end_stream,

FilterIterationStartState filter_iteration_start_state) {

// Only do base state setting on the initial call. Subsequent calls for filtering do not touch

// the base state.

ENVOY_STREAM_LOG(trace, "commonEncodePrefix end_stream: {}, isHalfCloseEnabled: {}", *this,

end_stream, filter_manager_callbacks_.isHalfCloseEnabled());

if (filter == nullptr) {

if (end_stream) {

ASSERT(!state_.observed_encode_end_stream_);

state_.observed_encode_end_stream_ = true;

// When half close semantics are disabled, receiving end stream from the upstream causes

// decoder filter to stop, as neither filters nor upstream is interested in downstream data.

// half close is enabled in case tcp proxying is done with http1 encoder. In this case, we

// should not stop decoder filter chain when end_stream is true, as it will cause any data

// sent in the upstream direction to be

// dropped.

if (!filter_manager_callbacks_.isHalfCloseEnabled()) {

state_.decoder_filter_chain_aborted_ = true;

}

}

return encoder_filters_.begin();

}

if (filter_iteration_start_state == FilterIterationStartState::CanStartFromCurrent &&

(*(filter->entry()))->iterate_from_current_filter_) {

// The filter iteration has been stopped for all frame types, and now the iteration continues.

// The current filter's encoding callback has not be called. Call it now.

return filter->entry();

}

return std::next(filter->entry());

}

StreamDecoderFilters::Iterator

FilterManager::commonDecodePrefix(ActiveStreamDecoderFilter* filter,

FilterIterationStartState filter_iteration_start_state) {

if (!filter) {

return decoder_filters_.begin();