Move decompression into gRPC Core

2020-04-02 18:54:18 -07:00 · 2020-04-02 18:54:18 -07:00 · 3c34eae97e
parent b25682427c
commit 3c34eae97e
21 changed files with 432 additions and 0 deletions
--- a/2
+++ b/2
@ -1193,11 +1193,13 @@ grpc_cc_library(
        "src/core/ext/filters/http/client/http_client_filter.cc",
        "src/core/ext/filters/http/http_filters_plugin.cc",
        "src/core/ext/filters/http/message_compress/message_compress_filter.cc",
+        "src/core/ext/filters/http/message_decompress/message_decompress_filter.cc",
        "src/core/ext/filters/http/server/http_server_filter.cc",
    ],
    hdrs = [
        "src/core/ext/filters/http/client/http_client_filter.h",
        "src/core/ext/filters/http/message_compress/message_compress_filter.h",
+        "src/core/ext/filters/http/message_decompress/message_decompress_filter.h",
        "src/core/ext/filters/http/server/http_server_filter.h",
    ],
    language = "c++",
--- a/BUILD.gn
+++ b/BUILD.gn
@ -318,6 +318,8 @@ config("grpc_config") {
        "src/core/ext/filters/http/http_filters_plugin.cc",
        "src/core/ext/filters/http/message_compress/message_compress_filter.cc",
        "src/core/ext/filters/http/message_compress/message_compress_filter.h",
+        "src/core/ext/filters/http/message_decompress/message_decompress_filter.cc",
+        "src/core/ext/filters/http/message_decompress/message_decompress_filter.h",
        "src/core/ext/filters/http/server/http_server_filter.cc",
        "src/core/ext/filters/http/server/http_server_filter.h",
        "src/core/ext/filters/max_age/max_age_filter.cc",
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@ -1369,6 +1369,7 @@ add_library(grpc
  src/core/ext/filters/http/client_authority_filter.cc
  src/core/ext/filters/http/http_filters_plugin.cc
  src/core/ext/filters/http/message_compress/message_compress_filter.cc
+  src/core/ext/filters/http/message_decompress/message_decompress_filter.cc
  src/core/ext/filters/http/server/http_server_filter.cc
  src/core/ext/filters/max_age/max_age_filter.cc
  src/core/ext/filters/message_size/message_size_filter.cc
@ -2028,6 +2029,7 @@ add_library(grpc_unsecure
  src/core/ext/filters/http/client_authority_filter.cc
  src/core/ext/filters/http/http_filters_plugin.cc
  src/core/ext/filters/http/message_compress/message_compress_filter.cc
+  src/core/ext/filters/http/message_decompress/message_decompress_filter.cc
  src/core/ext/filters/http/server/http_server_filter.cc
  src/core/ext/filters/max_age/max_age_filter.cc
  src/core/ext/filters/message_size/message_size_filter.cc
--- a/2
+++ b/2
@ -3694,6 +3694,7 @@ LIBGRPC_SRC = \
    src/core/ext/filters/http/client_authority_filter.cc \
    src/core/ext/filters/http/http_filters_plugin.cc \
    src/core/ext/filters/http/message_compress/message_compress_filter.cc \
+    src/core/ext/filters/http/message_decompress/message_decompress_filter.cc \
    src/core/ext/filters/http/server/http_server_filter.cc \
    src/core/ext/filters/max_age/max_age_filter.cc \
    src/core/ext/filters/message_size/message_size_filter.cc \
@ -4327,6 +4328,7 @@ LIBGRPC_UNSECURE_SRC = \
    src/core/ext/filters/http/client_authority_filter.cc \
    src/core/ext/filters/http/http_filters_plugin.cc \
    src/core/ext/filters/http/message_compress/message_compress_filter.cc \
+    src/core/ext/filters/http/message_decompress/message_decompress_filter.cc \
    src/core/ext/filters/http/server/http_server_filter.cc \
    src/core/ext/filters/max_age/max_age_filter.cc \
    src/core/ext/filters/message_size/message_size_filter.cc \
--- a/build_autogenerated.yaml
+++ b/build_autogenerated.yaml
@ -423,6 +423,7 @@ libs:
  - src/core/ext/filters/http/client/http_client_filter.h
  - src/core/ext/filters/http/client_authority_filter.h
  - src/core/ext/filters/http/message_compress/message_compress_filter.h
+  - src/core/ext/filters/http/message_decompress/message_decompress_filter.h
  - src/core/ext/filters/http/server/http_server_filter.h
  - src/core/ext/filters/max_age/max_age_filter.h
  - src/core/ext/filters/message_size/message_size_filter.h
@ -795,6 +796,7 @@ libs:
  - src/core/ext/filters/http/client_authority_filter.cc
  - src/core/ext/filters/http/http_filters_plugin.cc
  - src/core/ext/filters/http/message_compress/message_compress_filter.cc
+  - src/core/ext/filters/http/message_decompress/message_decompress_filter.cc
  - src/core/ext/filters/http/server/http_server_filter.cc
  - src/core/ext/filters/max_age/max_age_filter.cc
  - src/core/ext/filters/message_size/message_size_filter.cc
@ -1325,6 +1327,7 @@ libs:
  - src/core/ext/filters/http/client/http_client_filter.h
  - src/core/ext/filters/http/client_authority_filter.h
  - src/core/ext/filters/http/message_compress/message_compress_filter.h
+  - src/core/ext/filters/http/message_decompress/message_decompress_filter.h
  - src/core/ext/filters/http/server/http_server_filter.h
  - src/core/ext/filters/max_age/max_age_filter.h
  - src/core/ext/filters/message_size/message_size_filter.h
@ -1632,6 +1635,7 @@ libs:
  - src/core/ext/filters/http/client_authority_filter.cc
  - src/core/ext/filters/http/http_filters_plugin.cc
  - src/core/ext/filters/http/message_compress/message_compress_filter.cc
+  - src/core/ext/filters/http/message_decompress/message_decompress_filter.cc
  - src/core/ext/filters/http/server/http_server_filter.cc
  - src/core/ext/filters/max_age/max_age_filter.cc
  - src/core/ext/filters/message_size/message_size_filter.cc
--- a/config.m4
+++ b/config.m4
@ -104,6 +104,7 @@ if test "$PHP_GRPC" != "no"; then
    src/core/ext/filters/http/client_authority_filter.cc \
    src/core/ext/filters/http/http_filters_plugin.cc \
    src/core/ext/filters/http/message_compress/message_compress_filter.cc \
+    src/core/ext/filters/http/message_decompress/message_decompress_filter.cc \
    src/core/ext/filters/http/server/http_server_filter.cc \
    src/core/ext/filters/max_age/max_age_filter.cc \
    src/core/ext/filters/message_size/message_size_filter.cc \
@ -841,6 +842,7 @@ if test "$PHP_GRPC" != "no"; then
  PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/http)
  PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/http/client)
  PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/http/message_compress)
+  PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/http/message_decompress)
  PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/http/server)
  PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/max_age)
  PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/message_size)
--- a/config.w32
+++ b/config.w32
@ -73,6 +73,7 @@ if (PHP_GRPC != "no") {
    "src\\core\\ext\\filters\\http\\client_authority_filter.cc " +
    "src\\core\\ext\\filters\\http\\http_filters_plugin.cc " +
    "src\\core\\ext\\filters\\http\\message_compress\\message_compress_filter.cc " +
+    "src\\core\\ext\\filters\\http\\message_decompress\\message_decompress_filter.cc " +
    "src\\core\\ext\\filters\\http\\server\\http_server_filter.cc " +
    "src\\core\\ext\\filters\\max_age\\max_age_filter.cc " +
    "src\\core\\ext\\filters\\message_size\\message_size_filter.cc " +
@ -842,6 +843,7 @@ if (PHP_GRPC != "no") {
  FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\http");
  FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\http\\client");
  FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\http\\message_compress");
+  FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\http\\message_decompress");
  FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\http\\server");
  FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\max_age");
  FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\message_size");
--- a/gRPC-C++.podspec
+++ b/gRPC-C++.podspec
@ -274,6 +274,7 @@ Pod::Spec.new do |s|
                      'src/core/ext/filters/http/client/http_client_filter.h',
                      'src/core/ext/filters/http/client_authority_filter.h',
                      'src/core/ext/filters/http/message_compress/message_compress_filter.h',
+                      'src/core/ext/filters/http/message_decompress/message_decompress_filter.h',
                      'src/core/ext/filters/http/server/http_server_filter.h',
                      'src/core/ext/filters/max_age/max_age_filter.h',
                      'src/core/ext/filters/message_size/message_size_filter.h',
@ -725,6 +726,7 @@ Pod::Spec.new do |s|
                              'src/core/ext/filters/http/client/http_client_filter.h',
                              'src/core/ext/filters/http/client_authority_filter.h',
                              'src/core/ext/filters/http/message_compress/message_compress_filter.h',
+                              'src/core/ext/filters/http/message_decompress/message_decompress_filter.h',
                              'src/core/ext/filters/http/server/http_server_filter.h',
                              'src/core/ext/filters/max_age/max_age_filter.h',
                              'src/core/ext/filters/message_size/message_size_filter.h',
--- a/gRPC-Core.podspec
+++ b/gRPC-Core.podspec
@ -301,6 +301,8 @@ Pod::Spec.new do |s|
                      'src/core/ext/filters/http/http_filters_plugin.cc',
                      'src/core/ext/filters/http/message_compress/message_compress_filter.cc',
                      'src/core/ext/filters/http/message_compress/message_compress_filter.h',
+                      'src/core/ext/filters/http/message_decompress/message_decompress_filter.cc',
+                      'src/core/ext/filters/http/message_decompress/message_decompress_filter.h',
                      'src/core/ext/filters/http/server/http_server_filter.cc',
                      'src/core/ext/filters/http/server/http_server_filter.h',
                      'src/core/ext/filters/max_age/max_age_filter.cc',
@ -1078,6 +1080,7 @@ Pod::Spec.new do |s|
                              'src/core/ext/filters/http/client/http_client_filter.h',
                              'src/core/ext/filters/http/client_authority_filter.h',
                              'src/core/ext/filters/http/message_compress/message_compress_filter.h',
+                              'src/core/ext/filters/http/message_decompress/message_decompress_filter.h',
                              'src/core/ext/filters/http/server/http_server_filter.h',
                              'src/core/ext/filters/max_age/max_age_filter.h',
                              'src/core/ext/filters/message_size/message_size_filter.h',
--- a/grpc.gemspec
+++ b/grpc.gemspec
@ -223,6 +223,8 @@ Gem::Specification.new do |s|
  s.files += %w( src/core/ext/filters/http/http_filters_plugin.cc )
  s.files += %w( src/core/ext/filters/http/message_compress/message_compress_filter.cc )
  s.files += %w( src/core/ext/filters/http/message_compress/message_compress_filter.h )
+  s.files += %w( src/core/ext/filters/http/message_decompress/message_decompress_filter.cc )
+  s.files += %w( src/core/ext/filters/http/message_decompress/message_decompress_filter.h )
  s.files += %w( src/core/ext/filters/http/server/http_server_filter.cc )
  s.files += %w( src/core/ext/filters/http/server/http_server_filter.h )
  s.files += %w( src/core/ext/filters/max_age/max_age_filter.cc )
--- a/grpc.gyp
+++ b/grpc.gyp
@ -497,6 +497,7 @@
        'src/core/ext/filters/http/client_authority_filter.cc',
        'src/core/ext/filters/http/http_filters_plugin.cc',
        'src/core/ext/filters/http/message_compress/message_compress_filter.cc',
+        'src/core/ext/filters/http/message_decompress/message_decompress_filter.cc',
        'src/core/ext/filters/http/server/http_server_filter.cc',
        'src/core/ext/filters/max_age/max_age_filter.cc',
        'src/core/ext/filters/message_size/message_size_filter.cc',
@ -992,6 +993,7 @@
        'src/core/ext/filters/http/client_authority_filter.cc',
        'src/core/ext/filters/http/http_filters_plugin.cc',
        'src/core/ext/filters/http/message_compress/message_compress_filter.cc',
+        'src/core/ext/filters/http/message_decompress/message_decompress_filter.cc',
        'src/core/ext/filters/http/server/http_server_filter.cc',
        'src/core/ext/filters/max_age/max_age_filter.cc',
        'src/core/ext/filters/message_size/message_size_filter.cc',
--- a/include/grpc/impl/codegen/grpc_types.h
+++ b/include/grpc/impl/codegen/grpc_types.h
@ -174,6 +174,14 @@ typedef struct {
 /** Enable/disable support for per-message compression. Defaults to 1, unless
    GRPC_ARG_MINIMAL_STACK is enabled, in which case it defaults to 0. */
 #define GRPC_ARG_ENABLE_PER_MESSAGE_COMPRESSION "grpc.per_message_compression"
+/** Experimental Arg. Enable/disable support for per-message decompression.
+   Defaults to 1, unless GRPC_ARG_MINIMAL_STACK is enabled, in which case it
+   defaults to 0. If disabled, decompression will be performed lazily by
+   grpc_byte_buffer_reader. This arg also determines whether max message limits
+   will be applied to the decompressed buffer or the non-decompressed buffer. It
+   is recommended to keep this enabled to protect against zip bomb attacks. */
+#define GRPC_ARG_ENABLE_PER_MESSAGE_DECOMPRESSION \
+  "grpc.per_message_decompression"
 /** Enable/disable support for deadline checking. Defaults to 1, unless
    GRPC_ARG_MINIMAL_STACK is enabled, in which case it defaults to 0 */
 #define GRPC_ARG_ENABLE_DEADLINE_CHECKS "grpc.enable_deadline_checking"
--- a/package.xml
+++ b/package.xml
@ -203,6 +203,8 @@
    <file baseinstalldir="/" name="src/core/ext/filters/http/http_filters_plugin.cc" role="src" />
    <file baseinstalldir="/" name="src/core/ext/filters/http/message_compress/message_compress_filter.cc" role="src" />
    <file baseinstalldir="/" name="src/core/ext/filters/http/message_compress/message_compress_filter.h" role="src" />
+    <file baseinstalldir="/" name="src/core/ext/filters/http/message_decompress/message_decompress_filter.cc" role="src" />
+    <file baseinstalldir="/" name="src/core/ext/filters/http/message_decompress/message_decompress_filter.h" role="src" />
    <file baseinstalldir="/" name="src/core/ext/filters/http/server/http_server_filter.cc" role="src" />
    <file baseinstalldir="/" name="src/core/ext/filters/http/server/http_server_filter.h" role="src" />
    <file baseinstalldir="/" name="src/core/ext/filters/max_age/max_age_filter.cc" role="src" />
--- a/src/core/ext/filters/http/http_filters_plugin.cc
+++ b/src/core/ext/filters/http/http_filters_plugin.cc
@ -22,6 +22,7 @@

 #include "src/core/ext/filters/http/client/http_client_filter.h"
 #include "src/core/ext/filters/http/message_compress/message_compress_filter.h"
+#include "src/core/ext/filters/http/message_decompress/message_decompress_filter.h"
 #include "src/core/ext/filters/http/server/http_server_filter.h"
 #include "src/core/lib/channel/channel_stack_builder.h"
 #include "src/core/lib/surface/call.h"
@ -36,6 +37,9 @@ typedef struct {
 static optional_filter compress_filter = {
    &grpc_message_compress_filter, GRPC_ARG_ENABLE_PER_MESSAGE_COMPRESSION};

+static optional_filter decompress_filter = {
+    &grpc_message_decompress_filter, GRPC_ARG_ENABLE_PER_MESSAGE_DECOMPRESSION};
+
 static bool is_building_http_like_transport(
    grpc_channel_stack_builder* builder) {
  grpc_transport* t = grpc_channel_stack_builder_get_transport(builder);
@ -75,6 +79,15 @@ void grpc_http_filters_init(void) {
  grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL,
                                   GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
                                   maybe_add_optional_filter, &compress_filter);
+  grpc_channel_init_register_stage(
+      GRPC_CLIENT_SUBCHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
+      maybe_add_optional_filter, &decompress_filter);
+  grpc_channel_init_register_stage(
+      GRPC_CLIENT_DIRECT_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
+      maybe_add_optional_filter, &decompress_filter);
+  grpc_channel_init_register_stage(
+      GRPC_SERVER_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
+      maybe_add_optional_filter, &decompress_filter);
  grpc_channel_init_register_stage(
      GRPC_CLIENT_SUBCHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
      maybe_add_required_filter, (void*)&grpc_http_client_filter);
--- a/src/core/ext/filters/http/message_decompress/message_decompress_filter.cc
+++ b/src/core/ext/filters/http/message_decompress/message_decompress_filter.cc
@ -0,0 +1,339 @@
+/*
+ *
+ * Copyright 2015 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include <grpc/support/port_platform.h>
+
+#include <assert.h>
+#include <string.h>
+
+#include <grpc/compression.h>
+#include <grpc/slice_buffer.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+
+#include "src/core/ext/filters/http/message_decompress/message_decompress_filter.h"
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/compression/algorithm_metadata.h"
+#include "src/core/lib/compression/compression_args.h"
+#include "src/core/lib/compression/compression_internal.h"
+#include "src/core/lib/compression/message_compress.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gprpp/manual_constructor.h"
+#include "src/core/lib/profiling/timers.h"
+#include "src/core/lib/slice/slice_internal.h"
+#include "src/core/lib/slice/slice_string_helpers.h"
+#include "src/core/lib/surface/call.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+namespace {
+
+class ChannelData {};
+
+class CallData {
+ public:
+  CallData(const grpc_call_element_args& args)
+      : call_combiner_(args.call_combiner) {
+    // Initialize state for recv_initial_metadata_ready callback
+    GRPC_CLOSURE_INIT(&on_recv_initial_metadata_ready_,
+                      OnRecvInitialMetadataReady, this,
+                      grpc_schedule_on_exec_ctx);
+    // Initialize state for recv_message_ready callback
+    grpc_slice_buffer_init(&recv_slices_);
+    GRPC_CLOSURE_INIT(&on_recv_message_next_done_, OnRecvMessageNextDone, this,
+                      grpc_schedule_on_exec_ctx);
+    GRPC_CLOSURE_INIT(&on_recv_message_ready_, OnRecvMessageReady, this,
+                      grpc_schedule_on_exec_ctx);
+    // Initialize state for recv_trailing_metadata_ready callback
+    GRPC_CLOSURE_INIT(&on_recv_trailing_metadata_ready_,
+                      OnRecvTrailingMetadataReady, this,
+                      grpc_schedule_on_exec_ctx);
+  }
+
+  static void DecompressStartTransportStreamOpBatch(
+      grpc_call_element* elem, grpc_transport_stream_op_batch* batch);
+
+  static void OnRecvInitialMetadataReady(void* arg, grpc_error* error);
+
+  // Methods for processing a receive message event
+  void MaybeResumeOnRecvMessageReady();
+  static void OnRecvMessageReady(void* arg, grpc_error* error);
+  static void OnRecvMessageNextDone(void* arg, grpc_error* error);
+  grpc_error* PullSliceFromRecvMessage();
+  void ContinueReadingRecvMessage();
+  void FinishRecvMessage();
+  void ContinueRecvMessageReadyCallback(grpc_error* error);
+
+  // Methods for processing a recv_trailing_metadata event
+  void MaybeResumeOnRecvTrailingMetadataReady();
+  static void OnRecvTrailingMetadataReady(void* arg, grpc_error* error);
+
+ private:
+  grpc_core::CallCombiner* call_combiner_ = nullptr;
+  // Overall error for the call
+  grpc_error* error_ = GRPC_ERROR_NONE;
+  // Fields for handling recv_initial_metadata_ready callback
+  grpc_closure on_recv_initial_metadata_ready_;
+  grpc_closure* original_recv_initial_metadata_ready_ = nullptr;
+  grpc_metadata_batch* recv_initial_metadata_ = nullptr;
+  // Fields for handling recv_message_ready callback
+  bool seen_recv_message_ready_ = false;
+  grpc_message_compression_algorithm algorithm_ = GRPC_MESSAGE_COMPRESS_NONE;
+  grpc_closure on_recv_message_ready_;
+  grpc_closure* original_recv_message_ready_ = nullptr;
+  grpc_closure on_recv_message_next_done_;
+  grpc_core::OrphanablePtr<grpc_core::ByteStream>* recv_message_ = nullptr;
+  // recv_slices_ holds the slices read from the original recv_message stream.
+  // It is initialized during construction and reset when a new stream is
+  // created using it.
+  grpc_slice_buffer recv_slices_;
+  grpc_core::ManualConstructor<grpc_core::SliceBufferByteStream>
+      recv_replacement_stream_;
+  // Fields for handling recv_trailing_metadata_ready callback
+  bool seen_recv_trailing_metadata_ready_ = false;
+  grpc_closure on_recv_trailing_metadata_ready_;
+  grpc_closure* original_recv_trailing_metadata_ready_ = nullptr;
+  grpc_error* on_recv_trailing_metadata_ready_error_ = GRPC_ERROR_NONE;
+};
+
+grpc_message_compression_algorithm DecodeMessageCompressionAlgorithm(
+    grpc_mdelem md) {
+  grpc_message_compression_algorithm algorithm =
+      grpc_message_compression_algorithm_from_slice(GRPC_MDVALUE(md));
+  if (algorithm == GRPC_MESSAGE_COMPRESS_ALGORITHMS_COUNT) {
+    char* md_c_str = grpc_slice_to_c_string(GRPC_MDVALUE(md));
+    gpr_log(GPR_ERROR,
+            "Invalid incoming message compression algorithm: '%s'. "
+            "Interpreting incoming data as uncompressed.",
+            md_c_str);
+    gpr_free(md_c_str);
+    return GRPC_MESSAGE_COMPRESS_NONE;
+  }
+  return algorithm;
+}
+
+void CallData::OnRecvInitialMetadataReady(void* arg, grpc_error* error) {
+  CallData* calld = static_cast<CallData*>(arg);
+  if (error == GRPC_ERROR_NONE) {
+    grpc_linked_mdelem* grpc_encoding =
+        calld->recv_initial_metadata_->idx.named.grpc_encoding;
+    if (grpc_encoding != nullptr) {
+      calld->algorithm_ = DecodeMessageCompressionAlgorithm(grpc_encoding->md);
+      grpc_metadata_batch_remove(calld->recv_initial_metadata_,
+                                 GRPC_BATCH_GRPC_ENCODING);
+    }
+  }
+  calld->MaybeResumeOnRecvMessageReady();
+  calld->MaybeResumeOnRecvTrailingMetadataReady();
+  grpc_closure* closure = calld->original_recv_initial_metadata_ready_;
+  calld->original_recv_initial_metadata_ready_ = nullptr;
+  grpc_core::Closure::Run(DEBUG_LOCATION, closure, GRPC_ERROR_REF(error));
+}
+
+void CallData::MaybeResumeOnRecvMessageReady() {
+  if (seen_recv_message_ready_) {
+    seen_recv_message_ready_ = false;
+    GRPC_CALL_COMBINER_START(call_combiner_, &on_recv_message_ready_,
+                             GRPC_ERROR_NONE,
+                             "continue recv_message_ready callback");
+  }
+}
+
+void CallData::OnRecvMessageReady(void* arg, grpc_error* error) {
+  CallData* calld = static_cast<CallData*>(arg);
+  if (error == GRPC_ERROR_NONE &&
+      calld->algorithm_ != GRPC_MESSAGE_COMPRESS_NONE) {
+    // recv_message can be NULL if trailing metadata is received instead of
+    // message.
+    if (*calld->recv_message_ == nullptr ||
+        (*calld->recv_message_)->length() == 0) {
+      calld->ContinueRecvMessageReadyCallback(GRPC_ERROR_NONE);
+      return;
+    }
+    if (calld->original_recv_initial_metadata_ready_ != nullptr) {
+      calld->seen_recv_message_ready_ = true;
+      GRPC_CALL_COMBINER_STOP(calld->call_combiner_,
+                              "Deferring OnRecvMessageReady until after "
+                              "OnRecvInitialMetadataReady");
+      return;
+    }
+    calld->ContinueReadingRecvMessage();
+  } else {
+    calld->ContinueRecvMessageReadyCallback(GRPC_ERROR_REF(error));
+  }
+}
+
+void CallData::ContinueReadingRecvMessage() {
+  while ((*recv_message_)
+             ->Next(~static_cast<size_t>(0), &on_recv_message_next_done_)) {
+    grpc_error* error = PullSliceFromRecvMessage();
+    if (error != GRPC_ERROR_NONE) {
+      return ContinueRecvMessageReadyCallback(error);
+    }
+    // We have read the entire message.
+    if (recv_slices_.length == (*recv_message_)->length()) {
+      return FinishRecvMessage();
+    }
+  }
+}
+
+grpc_error* CallData::PullSliceFromRecvMessage() {
+  grpc_slice incoming_slice;
+  grpc_error* error = (*recv_message_)->Pull(&incoming_slice);
+  if (error == GRPC_ERROR_NONE) {
+    grpc_slice_buffer_add(&recv_slices_, incoming_slice);
+  }
+  return error;
+}
+
+void CallData::OnRecvMessageNextDone(void* arg, grpc_error* error) {
+  CallData* calld = static_cast<CallData*>(arg);
+  if (error != GRPC_ERROR_NONE) {
+    return calld->ContinueRecvMessageReadyCallback(GRPC_ERROR_REF(error));
+  }
+  error = calld->PullSliceFromRecvMessage();
+  if (error != GRPC_ERROR_NONE) {
+    return calld->ContinueRecvMessageReadyCallback(error);
+  }
+  if (calld->recv_slices_.length == (*calld->recv_message_)->length()) {
+    calld->FinishRecvMessage();
+  } else {
+    calld->ContinueReadingRecvMessage();
+  }
+}
+
+void CallData::FinishRecvMessage() {
+  grpc_slice_buffer decompressed_slices;
+  if (grpc_msg_decompress(algorithm_, &recv_slices_, &decompressed_slices) ==
+      0) {
+    gpr_log(
+        GPR_ERROR,
+        "Unexpected error decompressing data for algorithm with enum value %d",
+        algorithm_);
+  }
+  uint32_t recv_flags = (*recv_message_)->flags();
+  // Swap out the original receive byte stream with our new one and send the
+  // batch down.
+  recv_replacement_stream_.Init(&recv_slices_, recv_flags);
+  recv_message_->reset(recv_replacement_stream_.get());
+  recv_message_ = nullptr;
+  ContinueRecvMessageReadyCallback(GRPC_ERROR_NONE);
+}
+
+void CallData::ContinueRecvMessageReadyCallback(grpc_error* error) {
+  MaybeResumeOnRecvTrailingMetadataReady();
+  // The surface will clean up the receiving stream if there is an error.
+  grpc_closure* closure = original_recv_message_ready_;
+  original_recv_message_ready_ = nullptr;
+  grpc_core::Closure::Run(DEBUG_LOCATION, closure, error);
+}
+
+void CallData::MaybeResumeOnRecvTrailingMetadataReady() {
+  if (seen_recv_trailing_metadata_ready_) {
+    seen_recv_trailing_metadata_ready_ = false;
+    grpc_error* error = on_recv_trailing_metadata_ready_error_;
+    on_recv_trailing_metadata_ready_error_ = GRPC_ERROR_NONE;
+    GRPC_CALL_COMBINER_START(call_combiner_, &on_recv_trailing_metadata_ready_,
+                             error, "Continuing OnRecvTrailingMetadataReady");
+  }
+}
+
+void CallData::OnRecvTrailingMetadataReady(void* arg, grpc_error* error) {
+  CallData* calld = static_cast<CallData*>(arg);
+  if (calld->original_recv_initial_metadata_ready_ != nullptr ||
+      calld->original_recv_message_ready_ != nullptr) {
+    calld->seen_recv_trailing_metadata_ready_ = true;
+    calld->on_recv_trailing_metadata_ready_error_ = GRPC_ERROR_REF(error);
+    GRPC_CALL_COMBINER_STOP(
+        calld->call_combiner_,
+        "Deferring OnRecvTrailingMetadataReady until after "
+        "OnRecvInitialMetadataReady and OnRecvMessageReady");
+    return;
+  }
+  error = grpc_error_add_child(GRPC_ERROR_REF(error), calld->error_);
+  calld->error_ = GRPC_ERROR_NONE;
+  grpc_closure* closure = calld->original_recv_trailing_metadata_ready_;
+  calld->original_recv_trailing_metadata_ready_ = nullptr;
+  grpc_core::Closure::Run(DEBUG_LOCATION, closure, error);
+}
+
+void CallData::DecompressStartTransportStreamOpBatch(
+    grpc_call_element* elem, grpc_transport_stream_op_batch* batch) {
+  GPR_TIMER_SCOPE("compress_start_transport_stream_op_batch", 0);
+  CallData* calld = static_cast<CallData*>(elem->call_data);
+  // Handle recv_initial_metadata.
+  if (batch->recv_initial_metadata) {
+    calld->recv_initial_metadata_ =
+        batch->payload->recv_initial_metadata.recv_initial_metadata;
+    calld->original_recv_initial_metadata_ready_ =
+        batch->payload->recv_initial_metadata.recv_initial_metadata_ready;
+    batch->payload->recv_initial_metadata.recv_initial_metadata_ready =
+        &calld->on_recv_initial_metadata_ready_;
+  }
+  // Handle recv_message
+  if (batch->recv_message) {
+    calld->recv_message_ = batch->payload->recv_message.recv_message;
+    calld->original_recv_message_ready_ =
+        batch->payload->recv_message.recv_message_ready;
+    batch->payload->recv_message.recv_message_ready =
+        &calld->on_recv_message_ready_;
+  }
+  // Handle recv_trailing_metadata
+  if (batch->recv_trailing_metadata) {
+    calld->original_recv_trailing_metadata_ready_ =
+        batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
+    batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
+        &calld->on_recv_trailing_metadata_ready_;
+  }
+  // Pass control down the stack.
+  grpc_call_next_op(elem, batch);
+}
+
+static grpc_error* DecompressInitCallElem(grpc_call_element* elem,
+                                          const grpc_call_element_args* args) {
+  new (elem->call_data) CallData(*args);
+  return GRPC_ERROR_NONE;
+}
+
+static void DecompressDestroyCallElem(
+    grpc_call_element* elem, const grpc_call_final_info* /*final_info*/,
+    grpc_closure* /*ignored*/) {
+  CallData* calld = static_cast<CallData*>(elem->call_data);
+  calld->~CallData();
+}
+
+static grpc_error* DecompressInitChannelElem(
+    grpc_channel_element* /*elem*/, grpc_channel_element_args* /*args*/) {
+  return GRPC_ERROR_NONE;
+}
+
+void DecompressDestroyChannelElem(grpc_channel_element* /*elem*/) { return; }
+
+}  // namespace
+
+const grpc_channel_filter grpc_message_decompress_filter = {
+    CallData::DecompressStartTransportStreamOpBatch,
+    grpc_channel_next_op,
+    sizeof(CallData),
+    DecompressInitCallElem,
+    grpc_call_stack_ignore_set_pollset_or_pollset_set,
+    DecompressDestroyCallElem,
+    0,  // sizeof(ChannelData)
+    DecompressInitChannelElem,
+    DecompressDestroyChannelElem,
+    grpc_channel_next_get_info,
+    "message_decompress"};
--- a/src/core/ext/filters/http/message_decompress/message_decompress_filter.h
+++ b/src/core/ext/filters/http/message_decompress/message_decompress_filter.h
@ -0,0 +1,29 @@
+/*
+ *
+ * Copyright 2015 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPC_CORE_EXT_FILTERS_HTTP_MESSAGE_DECOMPRESS_MESSAGE_DECOMPRESS_FILTER_H
+#define GRPC_CORE_EXT_FILTERS_HTTP_MESSAGE_DECOMPRESS_MESSAGE_DECOMPRESS_FILTER_H
+
+#include <grpc/support/port_platform.h>
+
+#include "src/core/lib/channel/channel_stack.h"
+
+extern const grpc_channel_filter grpc_message_decompress_filter;
+
+#endif /* GRPC_CORE_EXT_FILTERS_HTTP_MESSAGE_DECOMPRESS_MESSAGE_DECOMPRESS_FILTER_H \
+        */
--- a/src/core/lib/surface/call.cc
+++ b/src/core/lib/surface/call.cc
@ -200,6 +200,8 @@ struct grpc_call {
  /* Stream compression algorithm for *incoming* data */
  grpc_stream_compression_algorithm incoming_stream_compression_algorithm =
      GRPC_STREAM_COMPRESS_NONE;
+  /* Maximum size for uncompressed receive message in bytes. -1 for unlimited */
+  int max_uncompressed_receive_message_length = -1;
  /* Supported encodings (compression algorithms), a bitset.
   * Always support no compression. */
  uint32_t encodings_accepted_by_peer = 1 << GRPC_MESSAGE_COMPRESS_NONE;
@ -497,6 +499,11 @@ void grpc_call_set_completion_queue(grpc_call* call,
                                             &call->pollent);
 }

+void grpc_call_set_max_uncompressed_receive_message_length(grpc_call* call,
+                                                           int limit) {
+  call->max_uncompressed_receive_message_length = limit;
+}
+
 #ifndef NDEBUG
 #define REF_REASON reason
 #define REF_ARG , const char* reason
--- a/src/core/lib/surface/call.h
+++ b/src/core/lib/surface/call.h
@ -59,6 +59,10 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args,

 void grpc_call_set_completion_queue(grpc_call* call, grpc_completion_queue* cq);

+/* Sets the max uncompressed receive message length for the call. */
+void grpc_call_set_max_uncompressed_receive_message_length(grpc_call* call,
+                                                           int limit);
+
 #ifndef NDEBUG
 void grpc_call_internal_ref(grpc_call* call, const char* reason);
 void grpc_call_internal_unref(grpc_call* call, const char* reason);
--- a/src/python/grpcio/grpc_core_dependencies.py
+++ b/src/python/grpcio/grpc_core_dependencies.py
@ -82,6 +82,7 @@ CORE_SOURCE_FILES = [
    'src/core/ext/filters/http/client_authority_filter.cc',
    'src/core/ext/filters/http/http_filters_plugin.cc',
    'src/core/ext/filters/http/message_compress/message_compress_filter.cc',
+    'src/core/ext/filters/http/message_decompress/message_decompress_filter.cc',
    'src/core/ext/filters/http/server/http_server_filter.cc',
    'src/core/ext/filters/max_age/max_age_filter.cc',
    'src/core/ext/filters/message_size/message_size_filter.cc',
--- a/tools/doxygen/Doxyfile.c++.internal
+++ b/tools/doxygen/Doxyfile.c++.internal
@ -1186,6 +1186,8 @@ src/core/ext/filters/http/client_authority_filter.h \
 src/core/ext/filters/http/http_filters_plugin.cc \
 src/core/ext/filters/http/message_compress/message_compress_filter.cc \
 src/core/ext/filters/http/message_compress/message_compress_filter.h \
+src/core/ext/filters/http/message_decompress/message_decompress_filter.cc \
+src/core/ext/filters/http/message_decompress/message_decompress_filter.h \
 src/core/ext/filters/http/server/http_server_filter.cc \
 src/core/ext/filters/http/server/http_server_filter.h \
 src/core/ext/filters/max_age/max_age_filter.cc \
--- a/tools/doxygen/Doxyfile.core.internal
+++ b/tools/doxygen/Doxyfile.core.internal
@ -986,6 +986,8 @@ src/core/ext/filters/http/client_authority_filter.h \
 src/core/ext/filters/http/http_filters_plugin.cc \
 src/core/ext/filters/http/message_compress/message_compress_filter.cc \
 src/core/ext/filters/http/message_compress/message_compress_filter.h \
+src/core/ext/filters/http/message_decompress/message_decompress_filter.cc \
+src/core/ext/filters/http/message_decompress/message_decompress_filter.h \
 src/core/ext/filters/http/server/http_server_filter.cc \
 src/core/ext/filters/http/server/http_server_filter.h \
 src/core/ext/filters/max_age/max_age_filter.cc \