grpc-web: Fix 503 handling for gRPC Web text

source/extensions/filters/http/grpc_web/grpc_web_filter.cc

@@ -7,7 +7,9 @@
 #include "common/common/assert.h"
 #include "common/common/base64.h"
 #include "common/common/empty_string.h"
+#include "common/common/enum_to_int.h"
 #include "common/common/utility.h"
+#include "common/grpc/common.h"
 #include "common/grpc/context_impl.h"
 #include "common/http/headers.h"
 #include "common/http/utility.h"
@@ -54,6 +56,22 @@ bool GrpcWebFilter::isGrpcWebRequest(const Http::RequestHeaderMap& headers) {
   return false;
 }
 
+bool GrpcWebFilter::hasGrpcWebContentType(const Http::RequestOrResponseHeaderMap& headers) const {
+  const Http::HeaderEntry* content_type = headers.ContentType();
+  if (content_type != nullptr) {
+    return gRpcWebContentTypes().count(content_type->value().getStringView()) > 0;
+  }
+  return false;
+}
+
+// Valid response headers contain gRPC or gRPC-Web response headers.
+bool GrpcWebFilter::isValidResponseHeaders(Http::ResponseHeaderMap& headers,
+                                           bool end_stream) const {
+  return Grpc::Common::isGrpcResponseHeaders(headers, end_stream) ||
+         (Http::Utility::getResponseStatus(headers) == enumToInt(Http::Code::OK) &&
+          hasGrpcWebContentType(headers));
+}
+
 // Implements StreamDecoderFilter.
 // TODO(fengli): Implements the subtypes of gRPC-Web content-type other than proto, like +json, etc.
 Http::FilterHeadersStatus GrpcWebFilter::decodeHeaders(Http::RequestHeaderMap& headers, bool) {
@@ -143,7 +161,8 @@ Http::FilterDataStatus GrpcWebFilter::decodeData(Buffer::Instance& data, bool en
 }
 
 // Implements StreamEncoderFilter.
-Http::FilterHeadersStatus GrpcWebFilter::encodeHeaders(Http::ResponseHeaderMap& headers, bool) {
+Http::FilterHeadersStatus GrpcWebFilter::encodeHeaders(Http::ResponseHeaderMap& headers,
+                                                       bool end_stream) {
   if (!is_grpc_web_request_) {
     return Http::FilterHeadersStatus::Continue;
   }
@@ -156,14 +175,44 @@ Http::FilterHeadersStatus GrpcWebFilter::encodeHeaders(Http::ResponseHeaderMap&
   } else {
     headers.setReferenceContentType(Http::Headers::get().ContentTypeValues.GrpcWebProto);
   }
-  return Http::FilterHeadersStatus::Continue;
+
+  if (end_stream || isValidResponseHeaders(headers, end_stream)) {
+    return Http::FilterHeadersStatus::Continue;
+  }
+
+  ENVOY_LOG(debug, "received invalid response headers since the upstream response or local reply "
+                   "is not a gRPC or gRPC-Web response");
+  response_headers_ = &headers;
+  return Http::FilterHeadersStatus::StopIteration;
 }
 
-Http::FilterDataStatus GrpcWebFilter::encodeData(Buffer::Instance& data, bool) {
+Http::FilterDataStatus GrpcWebFilter::encodeData(Buffer::Instance& data, bool end_stream) {
   if (!is_grpc_web_request_) {
     return Http::FilterDataStatus::Continue;
   }
 
+  // When the upstream response (this is also relevant for local reply, since gRPC-Web request is
+  // not a gRPC request which makes the local reply's is_grpc_request set to false) is not a gRPC
+  // response, we set the "grpc-message" header with the upstream body content.
+  if (response_headers_ != nullptr) {
+    if (!end_stream) {
+      return Http::FilterDataStatus::StopIterationAndBuffer;
+    }
+
+    constexpr uint64_t max_grpc_message_length = 1024;
+    // Take the last frame as the grpc-message value, but the size of it is limited by
+    // max_grpc_message_length.
+    const auto message =
+        Http::Utility::PercentEncoding::encode(data.toString().substr(0, max_grpc_message_length));
+    data.drain(data.length());
+
+    response_headers_->setGrpcStatus(Grpc::Utility::httpToGrpcStatus(
+        enumToInt(Http::Utility::getResponseStatus(*response_headers_))));
+    response_headers_->setGrpcMessage(message);
+    response_headers_->setContentLength(0);
+    return Http::FilterDataStatus::Continue;
+  }
+
   if (!is_text_response_) {
     // No additional transcoding required if gRPC-Web client asked for binary response.
     return Http::FilterDataStatus::Continue;

source/extensions/filters/http/grpc_web/grpc_web_filter.h

@@ -16,7 +16,9 @@ namespace GrpcWeb {
 /**
  * See docs/configuration/http_filters/grpc_web_filter.rst
  */
-class GrpcWebFilter : public Http::StreamFilter, NonCopyable {
+class GrpcWebFilter : public Http::StreamFilter,
+                      NonCopyable,
+                      public Logger::Loggable<Logger::Id::filter> {
 public:
   explicit GrpcWebFilter(Grpc::Context& context) : context_(context) {}
   ~GrpcWebFilter() override = default;
@@ -56,6 +58,8 @@ class GrpcWebFilter : public Http::StreamFilter, NonCopyable {
   void chargeStat(const Http::ResponseHeaderOrTrailerMap& headers);
   void setupStatTracking(const Http::RequestHeaderMap& headers);
   bool isGrpcWebRequest(const Http::RequestHeaderMap& headers);
+  bool hasGrpcWebContentType(const Http::RequestOrResponseHeaderMap& headers) const;
+  bool isValidResponseHeaders(Http::ResponseHeaderMap& headers, bool end_stream) const;
 
   static const uint8_t GRPC_WEB_TRAILER;
   const absl::flat_hash_set<std::string>& gRpcWebContentTypes() const;
@@ -70,6 +74,7 @@ class GrpcWebFilter : public Http::StreamFilter, NonCopyable {
   absl::optional<Grpc::Context::RequestStatNames> request_stat_names_;
   bool is_grpc_web_request_{};
   Grpc::Context& context_;
+  Http::ResponseHeaderMap* response_headers_{nullptr};
 };
 
 } // namespace GrpcWeb

test/extensions/filters/http/grpc_web/grpc_web_filter_integration_test.cc

@@ -1,5 +1,7 @@
 #include <memory>
 
+#include "common/common/base64.h"
+
 #include "extensions/filters/http/well_known_names.h"
 
 #include "test/integration/http_integration.h"
@@ -9,9 +11,14 @@
 namespace Envoy {
 namespace {
 
+constexpr absl::string_view text{"application/grpc-web-text"};
+constexpr absl::string_view binary{"application/grpc-web"};
+
 using SkipEncodingEmptyTrailers = bool;
-using TestParams =
-    std::tuple<Network::Address::IpVersion, Http::CodecClient::Type, SkipEncodingEmptyTrailers>;
+using ContentType = std::string;
+using Accept = std::string;
+using TestParams = std::tuple<Network::Address::IpVersion, Http::CodecClient::Type,
+                              SkipEncodingEmptyTrailers, ContentType, Accept>;
 
 class GrpcWebFilterIntegrationTest : public testing::TestWithParam<TestParams>,
                                      public HttpIntegrationTest {
@@ -32,11 +39,13 @@ class GrpcWebFilterIntegrationTest : public testing::TestWithParam<TestParams>,
 
   static std::string testParamsToString(const testing::TestParamInfo<TestParams> params) {
     return fmt::format(
-        "{}_{}_{}",
+        "{}_{}_{}_{}_{}",
         TestUtility::ipTestParamsToString(testing::TestParamInfo<Network::Address::IpVersion>(
             std::get<0>(params.param), params.index)),
         std::get<1>(params.param) == Http::CodecClient::Type::HTTP2 ? "Http2" : "Http",
-        std::get<2>(params.param) ? "SkipEncodingEmptyTrailers" : "SubmitEncodingEmptyTrailers");
+        std::get<2>(params.param) ? "SkipEncodingEmptyTrailers" : "SubmitEncodingEmptyTrailers",
+        std::get<3>(params.param) == text ? "SendText" : "SendBinary",
+        std::get<4>(params.param) == text ? "AcceptText" : "AcceptBinary");
   }
 };
 
@@ -45,12 +54,16 @@ INSTANTIATE_TEST_SUITE_P(
     testing::Combine(
         testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),
         testing::Values(Http::CodecClient::Type::HTTP1, Http::CodecClient::Type::HTTP2),
-        testing::Values(SkipEncodingEmptyTrailers{true}, SkipEncodingEmptyTrailers{false})),
+        testing::Values(SkipEncodingEmptyTrailers{true}, SkipEncodingEmptyTrailers{false}),
+        testing::Values(ContentType{text}, ContentType{binary}),
+        testing::Values(Accept{text}, Accept{binary})),
     GrpcWebFilterIntegrationTest::testParamsToString);
 
 TEST_P(GrpcWebFilterIntegrationTest, GrpcWebTrailersNotDuplicated) {
   const auto downstream_protocol = std::get<1>(GetParam());
   const bool http2_skip_encoding_empty_trailers = std::get<2>(GetParam());
+  const ContentType& content_type = std::get<3>(GetParam());
+  const Accept& accept = std::get<4>(GetParam());
 
   if (downstream_protocol == Http::CodecClient::Type::HTTP1) {
     config_helper_.addConfigModifier(setEnableDownstreamTrailersHttp1());
@@ -67,30 +80,37 @@ TEST_P(GrpcWebFilterIntegrationTest, GrpcWebTrailersNotDuplicated) {
 
   initialize();
   codec_client_ = makeHttpConnection(lookupPort("http"));
-  auto encoder_decoder = codec_client_->startRequest(
-      Http::TestRequestHeaderMapImpl{{":method", "POST"},
-                                     {":path", "/test/long/url"},
-                                     {":scheme", "http"},
-                                     {"content-type", "application/grpc-web"},
-                                     {":authority", "host"}});
+  auto encoder_decoder =
+      codec_client_->startRequest(Http::TestRequestHeaderMapImpl{{":method", "POST"},
+                                                                 {":path", "/test/long/url"},
+                                                                 {":scheme", "http"},
+                                                                 {"content-type", content_type},
+                                                                 {"accept", accept},
+                                                                 {":authority", "host"}});
   request_encoder_ = &encoder_decoder.first;
   auto response = std::move(encoder_decoder.second);
-  codec_client_->sendData(*request_encoder_, 1, false);
+
+  const std::string body = "hello";
+  const std::string encoded_body =
+      content_type == text ? Base64::encode(body.data(), body.length()) : body;
+  codec_client_->sendData(*request_encoder_, encoded_body, false);
   codec_client_->sendTrailers(*request_encoder_, request_trailers);
   waitForNextUpstreamRequest();
+  default_response_headers_.setReferenceContentType(Http::Headers::get().ContentTypeValues.Grpc);
   upstream_request_->encodeHeaders(default_response_headers_, false);
   upstream_request_->encodeData(1, false);
   upstream_request_->encodeTrailers(response_trailers);
   response->waitForEndStream();
 
   EXPECT_TRUE(upstream_request_->complete());
-  EXPECT_EQ(1, upstream_request_->bodyLength());
+  EXPECT_EQ(body.length(), upstream_request_->bodyLength());
   EXPECT_THAT(*upstream_request_->trailers(), HeaderMapEqualRef(&request_trailers));
 
   EXPECT_TRUE(response->complete());
   EXPECT_EQ("200", response->headers().getStatusValue());
-  EXPECT_TRUE(absl::StrContains(response->body(), "response1:trailer1"));
-  EXPECT_TRUE(absl::StrContains(response->body(), "response2:trailer2"));
+  const auto response_body = accept == text ? Base64::decode(response->body()) : response->body();
+  EXPECT_TRUE(absl::StrContains(response_body, "response1:trailer1"));
+  EXPECT_TRUE(absl::StrContains(response_body, "response2:trailer2"));
 
   if (downstream_protocol == Http::CodecClient::Type::HTTP1) {
     // When the downstream protocol is HTTP/1.1 we expect the trailers to be in the response-body.
@@ -113,6 +133,8 @@ TEST_P(GrpcWebFilterIntegrationTest, GrpcWebTrailersNotDuplicated) {
 TEST_P(GrpcWebFilterIntegrationTest, UpstreamDisconnect) {
   const auto downstream_protocol = std::get<1>(GetParam());
   const bool http2_skip_encoding_empty_trailers = std::get<2>(GetParam());
+  const ContentType& content_type = std::get<3>(GetParam());
+  const Accept& accept = std::get<4>(GetParam());
 
   if (downstream_protocol == Http::CodecClient::Type::HTTP1) {
     config_helper_.addConfigModifier(setEnableDownstreamTrailersHttp1());
@@ -127,12 +149,13 @@ TEST_P(GrpcWebFilterIntegrationTest, UpstreamDisconnect) {
 
   initialize();
   codec_client_ = makeHttpConnection(lookupPort("http"));
-  auto encoder_decoder = codec_client_->startRequest(
-      Http::TestRequestHeaderMapImpl{{":method", "POST"},
-                                     {":path", "/test/long/url"},
-                                     {":scheme", "http"},
-                                     {"content-type", "application/grpc-web"},
-                                     {":authority", "host"}});
+  auto encoder_decoder =
+      codec_client_->startRequest(Http::TestRequestHeaderMapImpl{{":method", "POST"},
+                                                                 {":path", "/test/long/url"},
+                                                                 {":scheme", "http"},
+                                                                 {"content-type", content_type},
+                                                                 {"accept", accept},
+                                                                 {":authority", "host"}});
   request_encoder_ = &encoder_decoder.first;
   auto response = std::move(encoder_decoder.second);
   codec_client_->sendData(*request_encoder_, 1, false);
@@ -141,7 +164,14 @@ TEST_P(GrpcWebFilterIntegrationTest, UpstreamDisconnect) {
 
   ASSERT_TRUE(fake_upstream_connection_->close());
   response->waitForEndStream();
+  EXPECT_TRUE(response->complete());
+
   EXPECT_EQ("503", response->headers().getStatusValue());
+  EXPECT_EQ(absl::StrCat(accept, "+proto"), response->headers().getContentTypeValue());
+  EXPECT_EQ("upstream connect error or disconnect/reset before headers. reset reason: connection "
+            "termination",
+            response->headers().getGrpcMessageValue());
+  EXPECT_EQ(0U, response->body().length());
 
   codec_client_->close();
 }

test/extensions/filters/http/grpc_web/grpc_web_filter_test.cc

@@ -195,6 +195,28 @@ TEST_F(GrpcWebFilterTest, Base64NoPadding) {
   EXPECT_EQ(decoder_callbacks_.details(), "grpc_base_64_decode_failed_bad_size");
 }
 
+TEST_F(GrpcWebFilterTest, InvalidUpstreamResponseForText) {
+  Http::TestRequestHeaderMapImpl request_headers{
+      {"content-type", Http::Headers::get().ContentTypeValues.GrpcWebText}, {":path", "/"}};
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_.decodeHeaders(request_headers, false));
+
+  Http::TestResponseHeaderMapImpl response_headers{{":status", "400"}};
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
+            filter_.encodeHeaders(response_headers, false));
+  Buffer::OwnedImpl data1("start");
+  EXPECT_EQ(Http::FilterDataStatus::StopIterationAndBuffer, filter_.encodeData(data1, false));
+  Buffer::OwnedImpl data2("hello");
+  EXPECT_EQ(Http::FilterDataStatus::StopIterationAndBuffer, filter_.encodeData(data2, false));
+  Buffer::OwnedImpl data3("end");
+  EXPECT_EQ(Http::FilterDataStatus::Continue, filter_.encodeData(data3, true));
+
+  // We only get the last frame here.
+  EXPECT_EQ("end", response_headers.get_(Http::Headers::get().GrpcMessage));
+
+  Http::TestResponseTrailerMapImpl response_trailers{{"grpc-status", "0"}};
+  EXPECT_EQ(Http::FilterTrailersStatus::Continue, filter_.encodeTrailers(response_trailers));
+}
+
 TEST_P(GrpcWebFilterTest, StatsNoCluster) {
   Http::TestRequestHeaderMapImpl request_headers{
       {"content-type", request_content_type()},