Skip to content

ext_proc: Support CONTINUE_AND_REPLACE from header callbacks #16437

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
snowp merged 3 commits into from
May 20, 2021
Merged

ext_proc: Support CONTINUE_AND_REPLACE from header callbacks #16437

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b3a0f02
ext_proc: Support CONTINUE_AND_REPLACE from header callbacks
gbrail May 10, 2021
03c6212
ext_proc: Address a code review comment
gbrail May 14, 2021
fba8cd3
Merge branch 'main' into ext-proc-replace
gbrail May 14, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 12 additions & 9 deletions api/envoy/service/ext_proc/v3alpha/external_processor.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,15 +231,18 @@ message CommonResponse {
// stream as normal. This is the default.
CONTINUE = 0;

// [#not-implemented-hide:]
// Replace the request or response with the contents
// of this message. If header_mutation is set, apply it to the
// headers. If body_mutation is set and contains a body, then add that
// body to the request or response, even if one does not already exist --
// otherwise, clear the body. Any additional body and trailers
// received from downstream or upstream will be ignored.
// This can be used to add a body to a request or response that does not
// have one already.
// Apply the specified header mutation, replace the body with the body
// specified in the body mutation (if present), and do not send any
// further messages for this request or response even if the processing
// mode is configured to do so.
//
// When used in response to a request_headers or response_headers message,
// this status makes it possible to either completely replace the body
// while discarding the original body, or to add a body to a message that
// formerly did not have one.
//
// In other words, this response makes it possible to turn an HTTP GET
// into a POST, PUT, or PATCH.
CONTINUE_AND_REPLACE = 1;
}

Expand Down
21 changes: 12 additions & 9 deletions generated_api_shadow/envoy/service/ext_proc/v3alpha/external_processor.proto
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions source/extensions/filters/http/ext_proc/BUILD
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ envoy_cc_library(
"//include/envoy/http:filter_interface",
"//include/envoy/http:header_map_interface",
"//include/envoy/stats:stats_macros",
"//source/common/buffer:buffer_lib",
"//source/extensions/filters/http/common:pass_through_filter_lib",
"@com_google_absl//absl/strings:str_format",
"@envoy_api//envoy/extensions/filters/http/ext_proc/v3alpha:pkg_cc_proto",
Expand Down
49 changes: 22 additions & 27 deletions source/extensions/filters/http/ext_proc/ext_proc.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,8 +60,8 @@ void Filter::onDestroy() {
}
}

FilterHeadersStatus Filter::onHeaders(ProcessorState& state, Http::HeaderMap& headers,
bool end_stream) {
FilterHeadersStatus Filter::onHeaders(ProcessorState& state,
Http::RequestOrResponseHeaderMap& headers, bool end_stream) {
switch (openStream()) {
case StreamOpenState::Error:
return FilterHeadersStatus::StopIteration;
Expand All @@ -75,7 +75,7 @@ FilterHeadersStatus Filter::onHeaders(ProcessorState& state, Http::HeaderMap& he
state.setHeaders(&headers);
ProcessingRequest req;
auto* headers_req = state.mutableHeaders(req);
MutationUtils::buildHttpHeaders(headers, *headers_req->mutable_headers());
MutationUtils::headersToProto(headers, *headers_req->mutable_headers());
headers_req->set_end_of_stream(end_stream);
state.setCallbackState(ProcessorState::CallbackState::HeadersCallback);
state.startMessageTimer(std::bind(&Filter::onMessageTimeout, this), config_->messageTimeout());
Expand All @@ -101,12 +101,22 @@ FilterHeadersStatus Filter::decodeHeaders(RequestHeaderMap& headers, bool end_st
return status;
}

Http::FilterDataStatus Filter::onData(ProcessorState& state, Buffer::Instance& data,
bool end_stream) {
FilterDataStatus Filter::onData(ProcessorState& state, Buffer::Instance& data, bool end_stream) {
if (end_stream) {
state.setCompleteBodyAvailable(true);
}

if (state.bodyReplaced()) {
ENVOY_LOG(trace, "Clearing body chunk because CONTINUE_AND_REPLACE was returned");
data.drain(data.length());
return FilterDataStatus::Continue;
}

if (processing_complete_) {
ENVOY_LOG(trace, "Continuing (processing complete)");
return FilterDataStatus::Continue;
}

bool just_added_trailers = false;
Http::HeaderMap* new_trailers = nullptr;
if (end_stream && state.sendTrailers()) {
Expand Down Expand Up @@ -192,17 +202,17 @@ Http::FilterDataStatus Filter::onData(ProcessorState& state, Buffer::Instance& d

FilterDataStatus Filter::decodeData(Buffer::Instance& data, bool end_stream) {
ENVOY_LOG(trace, "decodeData({}): end_stream = {}", data.length(), end_stream);
if (processing_complete_) {
ENVOY_LOG(trace, "decodeData: Continue (complete)");
return FilterDataStatus::Continue;
}

const auto status = onData(decoding_state_, data, end_stream);
ENVOY_LOG(trace, "decodeData returning {}", status);
return status;
}

FilterTrailersStatus Filter::onTrailers(ProcessorState& state, Http::HeaderMap& trailers) {
if (processing_complete_) {
ENVOY_LOG(trace, "trailers: Continue");
return FilterTrailersStatus::Continue;
}

bool body_delivered = state.completeBodyAvailable();
state.setCompleteBodyAvailable(true);
state.setTrailersAvailable(true);
Expand Down Expand Up @@ -243,11 +253,6 @@ FilterTrailersStatus Filter::onTrailers(ProcessorState& state, Http::HeaderMap&

FilterTrailersStatus Filter::decodeTrailers(RequestTrailerMap& trailers) {
ENVOY_LOG(trace, "decodeTrailers");
if (processing_complete_) {
ENVOY_LOG(trace, "decodeTrailers: Continue");
return FilterTrailersStatus::Continue;
}

const auto status = onTrailers(decoding_state_, trailers);
ENVOY_LOG(trace, "encodeTrailers returning {}", status);
return status;
Expand All @@ -271,23 +276,13 @@ FilterHeadersStatus Filter::encodeHeaders(ResponseHeaderMap& headers, bool end_s

FilterDataStatus Filter::encodeData(Buffer::Instance& data, bool end_stream) {
ENVOY_LOG(trace, "encodeData({}): end_stream = {}", data.length(), end_stream);
if (processing_complete_) {
ENVOY_LOG(trace, "encodeData: Continue (complete)");
return FilterDataStatus::Continue;
}

const auto status = onData(encoding_state_, data, end_stream);
ENVOY_LOG(trace, "encodeData returning {}", status);
return status;
}

FilterTrailersStatus Filter::encodeTrailers(ResponseTrailerMap& trailers) {
ENVOY_LOG(trace, "encodeTrailers");
if (processing_complete_) {
ENVOY_LOG(trace, "encodeTrailers: Continue");
return FilterTrailersStatus::Continue;
}

const auto status = onTrailers(encoding_state_, trailers);
ENVOY_LOG(trace, "encodeTrailers returning {}", status);
return status;
Expand All @@ -308,7 +303,7 @@ void Filter::sendBodyChunk(ProcessorState& state, const Buffer::Instance& data,
void Filter::sendTrailers(ProcessorState& state, const Http::HeaderMap& trailers) {
ProcessingRequest req;
auto* trailers_req = state.mutableTrailers(req);
MutationUtils::buildHttpHeaders(trailers, *trailers_req->mutable_trailers());
MutationUtils::headersToProto(trailers, *trailers_req->mutable_trailers());
state.setCallbackState(ProcessorState::CallbackState::TrailersCallback);
state.startMessageTimer(std::bind(&Filter::onMessageTimeout, this), config_->messageTimeout());
ENVOY_LOG(debug, "Sending trailers message");
Expand Down Expand Up @@ -459,7 +454,7 @@ void Filter::sendImmediateResponse(const ImmediateResponse& response) {
: absl::nullopt;
const auto mutate_headers = [&response](Http::ResponseHeaderMap& headers) {
if (response.has_headers()) {
MutationUtils::applyHeaderMutations(response.headers(), headers);
MutationUtils::applyHeaderMutations(response.headers(), headers, false);
}
};

Expand Down
4 changes: 2 additions & 2 deletions source/extensions/filters/http/ext_proc/ext_proc.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,8 +133,8 @@ class Filter : public Logger::Loggable<Logger::Id::filter>,
void sendImmediateResponse(const envoy::service::ext_proc::v3alpha::ImmediateResponse& response);
void sendBodyChunk(ProcessorState& state, const Buffer::Instance& data, bool end_stream);

Http::FilterHeadersStatus onHeaders(ProcessorState& state, Http::HeaderMap& headers,
bool end_stream);
Http::FilterHeadersStatus onHeaders(ProcessorState& state,
Http::RequestOrResponseHeaderMap& headers, bool end_stream);
Http::FilterDataStatus onData(ProcessorState& state, Buffer::Instance& data, bool end_stream);
Http::FilterTrailersStatus onTrailers(ProcessorState& state, Http::HeaderMap& trailers);

Expand Down
34 changes: 23 additions & 11 deletions source/extensions/filters/http/ext_proc/mutation_utils.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,14 @@ using Http::LowerCaseString;

using envoy::service::ext_proc::v3alpha::BodyMutation;
using envoy::service::ext_proc::v3alpha::BodyResponse;
using envoy::service::ext_proc::v3alpha::CommonResponse;
using envoy::service::ext_proc::v3alpha::HeaderMutation;
using envoy::service::ext_proc::v3alpha::HeadersResponse;

void MutationUtils::buildHttpHeaders(const Http::HeaderMap& headers_in,
envoy::config::core::v3::HeaderMap& headers_out) {
headers_in.iterate([&headers_out](const Http::HeaderEntry& e) -> Http::HeaderMap::Iterate {
auto* new_header = headers_out.add_headers();
void MutationUtils::headersToProto(const Http::HeaderMap& headers_in,
envoy::config::core::v3::HeaderMap& proto_out) {
headers_in.iterate([&proto_out](const Http::HeaderEntry& e) -> Http::HeaderMap::Iterate {
auto* new_header = proto_out.add_headers();
new_header->set_key(std::string(e.key().getStringView()));
new_header->set_value(std::string(e.value().getStringView()));
return Http::HeaderMap::Iterate::Continue;
Expand All @@ -34,12 +35,14 @@ void MutationUtils::applyCommonHeaderResponse(const HeadersResponse& response,
if (response.has_response()) {
const auto& common_response = response.response();
if (common_response.has_header_mutation()) {
applyHeaderMutations(common_response.header_mutation(), headers);
applyHeaderMutations(common_response.header_mutation(), headers,
common_response.status() == CommonResponse::CONTINUE_AND_REPLACE);
}
}
}

void MutationUtils::applyHeaderMutations(const HeaderMutation& mutation, Http::HeaderMap& headers) {
void MutationUtils::applyHeaderMutations(const HeaderMutation& mutation, Http::HeaderMap& headers,
bool replacing_message) {
for (const auto& remove_header : mutation.remove_headers()) {
if (Http::HeaderUtility::isRemovableHeader(remove_header)) {
ENVOY_LOG(trace, "Removing header {}", remove_header);
Expand All @@ -53,7 +56,7 @@ void MutationUtils::applyHeaderMutations(const HeaderMutation& mutation, Http::H
if (!sh.has_header()) {
continue;
}
if (isSettableHeader(sh.header().key())) {
if (isSettableHeader(sh.header().key(), replacing_message)) {
// Make "false" the default. This is logical and matches the ext_authz
// filter. However, the router handles this same protobuf and uses "true"
// as the default instead.
Expand All @@ -70,14 +73,20 @@ void MutationUtils::applyHeaderMutations(const HeaderMutation& mutation, Http::H
}
}

void MutationUtils::applyCommonBodyResponse(const BodyResponse& response, Http::HeaderMap* headers,
void MutationUtils::applyCommonBodyResponse(const BodyResponse& response,
Http::RequestOrResponseHeaderMap* headers,
Buffer::Instance& buffer) {
if (response.has_response()) {
const auto& common_response = response.response();
if (headers != nullptr && common_response.has_header_mutation()) {
applyHeaderMutations(common_response.header_mutation(), *headers);
applyHeaderMutations(common_response.header_mutation(), *headers,
common_response.status() == CommonResponse::CONTINUE_AND_REPLACE);
}
if (common_response.has_body_mutation()) {
if (headers != nullptr) {
// Always clear content length if we can before modifying body
headers->removeContentLength();
}
applyBodyMutations(common_response.body_mutation(), buffer);
}
}
Expand All @@ -87,10 +96,13 @@ void MutationUtils::applyBodyMutations(const BodyMutation& mutation, Buffer::Ins
switch (mutation.mutation_case()) {
case BodyMutation::MutationCase::kClearBody:
if (mutation.clear_body()) {
ENVOY_LOG(trace, "Clearing HTTP body");
buffer.drain(buffer.length());
}
break;
case BodyMutation::MutationCase::kBody:
ENVOY_LOG(trace, "Replacing body of {} bytes with new body of {} bytes", buffer.length(),
mutation.body().size());
buffer.drain(buffer.length());
buffer.add(mutation.body());
break;
Expand All @@ -103,11 +115,11 @@ void MutationUtils::applyBodyMutations(const BodyMutation& mutation, Buffer::Ins
// Ignore attempts to set certain sensitive headers that can break later processing.
// We may re-enable some of these after further testing. This logic is specific
// to the ext_proc filter so it is not shared with HeaderUtils.
bool MutationUtils::isSettableHeader(absl::string_view key) {
bool MutationUtils::isSettableHeader(absl::string_view key, bool replacing_message) {
const auto& headers = Headers::get();
return !absl::EqualsIgnoreCase(key, headers.HostLegacy.get()) &&
!absl::EqualsIgnoreCase(key, headers.Host.get()) &&
!absl::EqualsIgnoreCase(key, headers.Method.get()) &&
(!absl::EqualsIgnoreCase(key, headers.Method.get()) || replacing_message) &&
!absl::EqualsIgnoreCase(key, headers.Scheme.get()) &&
!absl::StartsWithIgnoreCase(key, headers.prefix());
}
Expand Down
11 changes: 6 additions & 5 deletions source/extensions/filters/http/ext_proc/mutation_utils.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@ namespace ExternalProcessing {
class MutationUtils : public Logger::Loggable<Logger::Id::filter> {
public:
// Convert a header map until a protobuf
Copy link
Copy Markdown
Contributor

@snowp snowp May 14, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

into?

Copy link
Copy Markdown
Contributor Author

@gbrail gbrail May 14, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we change the other thing here, then I'll certainly do it -- if not, then I'd like to do it later.

Copy link
Copy Markdown

@mjallday mjallday Jun 28, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i believe it was a suggestion on fixing a typo

Suggested change
// Convert a header map until a protobuf
// Convert a header map into a protobuf

static void buildHttpHeaders(const Http::HeaderMap& headers_in,
envoy::config::core::v3::HeaderMap& headers_out);
static void headersToProto(const Http::HeaderMap& headers_in,
envoy::config::core::v3::HeaderMap& proto_out);
Comment on lines +17 to +18
Copy link
Copy Markdown
Contributor

@snowp snowp May 14, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Old code but I think unless there is a good reason this should be returning the HeaderMap instead of taking an out param?

Copy link
Copy Markdown
Contributor Author

@gbrail gbrail May 14, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Interesting question, actually. The only places we call this method we're using it to set the fields of an embedded message field in protobuf. So I think that if we change it, then the code that calls this function has to first get a pointer to an embedded message by calling "mutable_whatever" on the field, and then needs to copy the result of this function into the result. Doing it this way saves the copy which seems like a good thing.

Copy link
Copy Markdown
Contributor

@snowp snowp May 14, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Alright that sounds like a good enough reason for me :)


// Apply mutations that are common to header responses.
static void
Expand All @@ -25,19 +25,20 @@ class MutationUtils : public Logger::Loggable<Logger::Id::filter> {
// Modify header map based on a set of mutations from a protobuf
static void
applyHeaderMutations(const envoy::service::ext_proc::v3alpha::HeaderMutation& mutation,
Http::HeaderMap& headers);
Http::HeaderMap& headers, bool replacing_message);

// Apply mutations that are common to body responses.
// Mutations will be applied to the header map if it is not null.
static void applyCommonBodyResponse(const envoy::service::ext_proc::v3alpha::BodyResponse& body,
Http::HeaderMap* headers, Buffer::Instance& buffer);
Http::RequestOrResponseHeaderMap* headers,
Buffer::Instance& buffer);

// Modify a buffer based on a set of mutations from a protobuf
static void applyBodyMutations(const envoy::service::ext_proc::v3alpha::BodyMutation& mutation,
Buffer::Instance& buffer);

private:
static bool isSettableHeader(absl::string_view key);
static bool isSettableHeader(absl::string_view key, bool replacing_message);
};

} // namespace ExternalProcessing
Expand Down
Loading