legacy-libs/grpc/deps/grpc/src/core/ext/filters/message_size/message_size_filter.cc

   1 //
   2 // Copyright 2016 gRPC authors.
   3 //
   4 // Licensed under the Apache License, Version 2.0 (the "License");
   5 // you may not use this file except in compliance with the License.
   6 // You may obtain a copy of the License at
   7 //
   8 //     http://www.apache.org/licenses/LICENSE-2.0
   9 //
  10 // Unless required by applicable law or agreed to in writing, software
  11 // distributed under the License is distributed on an "AS IS" BASIS,
  12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 // See the License for the specific language governing permissions and
  14 // limitations under the License.
  15 //
  16
  17 #include <grpc/support/port_platform.h>
  18
  19 #include "src/core/ext/filters/message_size/message_size_filter.h"
  20
  21 #include <limits.h>
  22 #include <string.h>
  23
  24 #include <grpc/impl/codegen/grpc_types.h>
  25 #include <grpc/support/alloc.h>
  26 #include <grpc/support/log.h>
  27 #include <grpc/support/string_util.h>
  28
  29 #include "src/core/ext/filters/client_channel/service_config.h"
  30 #include "src/core/lib/channel/channel_args.h"
  31 #include "src/core/lib/channel/channel_stack_builder.h"
  32 #include "src/core/lib/gpr/string.h"
  33 #include "src/core/lib/gprpp/ref_counted.h"
  34 #include "src/core/lib/gprpp/ref_counted_ptr.h"
  35 #include "src/core/lib/surface/call.h"
  36 #include "src/core/lib/surface/channel_init.h"
  37
  38 static void recv_message_ready(void* user_data, grpc_error* error);
  39 static void recv_trailing_metadata_ready(void* user_data, grpc_error* error);
  40
  41 namespace grpc_core {
  42
  43 namespace {
  44 size_t g_message_size_parser_index;
  45 }  // namespace
  46
  47 UniquePtr<ServiceConfig::ParsedConfig> MessageSizeParser::ParsePerMethodParams(
  48     const grpc_json* json, grpc_error** error) {
  49   GPR_DEBUG_ASSERT(error != nullptr && *error == GRPC_ERROR_NONE);
  50   int max_request_message_bytes = -1;
  51   int max_response_message_bytes = -1;
  52   InlinedVector<grpc_error*, 4> error_list;
  53   for (grpc_json* field = json->child; field != nullptr; field = field->next) {
  54     if (field->key == nullptr) continue;
  55     if (strcmp(field->key, "maxRequestMessageBytes") == 0) {
  56       if (max_request_message_bytes >= 0) {
  57         error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
  58             "field:maxRequestMessageBytes error:Duplicate entry"));
  59       }  // Duplicate, continue parsing.
  60       if (field->type != GRPC_JSON_STRING && field->type != GRPC_JSON_NUMBER) {
  61         error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
  62             "field:maxRequestMessageBytes error:should be of type number"));
  63       } else {
  64         max_request_message_bytes = gpr_parse_nonnegative_int(field->value);
  65         if (max_request_message_bytes == -1) {
  66           error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
  67               "field:maxRequestMessageBytes error:should be non-negative"));
  68         }
  69       }
  70     } else if (strcmp(field->key, "maxResponseMessageBytes") == 0) {
  71       if (max_response_message_bytes >= 0) {
  72         error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
  73             "field:maxResponseMessageBytes error:Duplicate entry"));
  74       }  // Duplicate, continue parsing
  75       if (field->type != GRPC_JSON_STRING && field->type != GRPC_JSON_NUMBER) {
  76         error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
  77             "field:maxResponseMessageBytes error:should be of type number"));
  78       } else {
  79         max_response_message_bytes = gpr_parse_nonnegative_int(field->value);
  80         if (max_response_message_bytes == -1) {
  81           error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
  82               "field:maxResponseMessageBytes error:should be non-negative"));
  83         }
  84       }
  85     }
  86   }
  87   if (!error_list.empty()) {
  88     *error = GRPC_ERROR_CREATE_FROM_VECTOR("Message size parser", &error_list);
  89     return nullptr;
  90   }
  91   return UniquePtr<ServiceConfig::ParsedConfig>(New<MessageSizeParsedConfig>(
  92       max_request_message_bytes, max_response_message_bytes));
  93 }
  94
  95 void MessageSizeParser::Register() {
  96   g_message_size_parser_index = ServiceConfig::RegisterParser(
  97       UniquePtr<ServiceConfig::Parser>(New<MessageSizeParser>()));
  98 }
  99
 100 size_t MessageSizeParser::ParserIndex() { return g_message_size_parser_index; }
 101 }  // namespace grpc_core
 102
 103 namespace {
 104 struct channel_data {
 105   grpc_core::MessageSizeParsedConfig::message_size_limits limits;
 106   grpc_core::RefCountedPtr<grpc_core::ServiceConfig> svc_cfg;
 107 };
 108
 109 struct call_data {
 110   call_data(grpc_call_element* elem, const channel_data& chand,
 111             const grpc_call_element_args& args)
 112       : call_combiner(args.call_combiner), limits(chand.limits) {
 113     GRPC_CLOSURE_INIT(&recv_message_ready, ::recv_message_ready, elem,
 114                       grpc_schedule_on_exec_ctx);
 115     GRPC_CLOSURE_INIT(&recv_trailing_metadata_ready,
 116                       ::recv_trailing_metadata_ready, elem,
 117                       grpc_schedule_on_exec_ctx);
 118     // Get max sizes from channel data, then merge in per-method config values.
 119     // Note: Per-method config is only available on the client, so we
 120     // apply the max request size to the send limit and the max response
 121     // size to the receive limit.
 122     const grpc_core::MessageSizeParsedConfig* limits = nullptr;
 123     grpc_core::ServiceConfig::CallData* svc_cfg_call_data = nullptr;
 124     if (args.context != nullptr) {
 125       svc_cfg_call_data = static_cast<grpc_core::ServiceConfig::CallData*>(
 126           args.context[GRPC_CONTEXT_SERVICE_CONFIG_CALL_DATA].value);
 127     }
 128     if (svc_cfg_call_data != nullptr) {
 129       limits = static_cast<const grpc_core::MessageSizeParsedConfig*>(
 130           svc_cfg_call_data->GetMethodParsedConfig(
 131               grpc_core::MessageSizeParser::ParserIndex()));
 132     } else if (chand.svc_cfg != nullptr) {
 133       const auto* objs_vector =
 134           chand.svc_cfg->GetMethodParsedConfigVector(args.path);
 135       if (objs_vector != nullptr) {
 136         limits = static_cast<const grpc_core::MessageSizeParsedConfig*>(
 137             (*objs_vector)[grpc_core::MessageSizeParser::ParserIndex()].get());
 138       }
 139     }
 140     if (limits != nullptr) {
 141       if (limits->limits().max_send_size >= 0 &&
 142           (limits->limits().max_send_size < this->limits.max_send_size ||
 143            this->limits.max_send_size < 0)) {
 144         this->limits.max_send_size = limits->limits().max_send_size;
 145       }
 146       if (limits->limits().max_recv_size >= 0 &&
 147           (limits->limits().max_recv_size < this->limits.max_recv_size ||
 148            this->limits.max_recv_size < 0)) {
 149         this->limits.max_recv_size = limits->limits().max_recv_size;
 150       }
 151     }
 152   }
 153
 154   ~call_data() { GRPC_ERROR_UNREF(error); }
 155
 156   grpc_core::CallCombiner* call_combiner;
 157   grpc_core::MessageSizeParsedConfig::message_size_limits limits;
 158   // Receive closures are chained: we inject this closure as the
 159   // recv_message_ready up-call on transport_stream_op, and remember to
 160   // call our next_recv_message_ready member after handling it.
 161   grpc_closure recv_message_ready;
 162   grpc_closure recv_trailing_metadata_ready;
 163   // The error caused by a message that is too large, or GRPC_ERROR_NONE
 164   grpc_error* error = GRPC_ERROR_NONE;
 165   // Used by recv_message_ready.
 166   grpc_core::OrphanablePtr<grpc_core::ByteStream>* recv_message = nullptr;
 167   // Original recv_message_ready callback, invoked after our own.
 168   grpc_closure* next_recv_message_ready = nullptr;
 169   // Original recv_trailing_metadata callback, invoked after our own.
 170   grpc_closure* original_recv_trailing_metadata_ready;
 171   bool seen_recv_trailing_metadata = false;
 172   grpc_error* recv_trailing_metadata_error;
 173 };
 174
 175 }  // namespace
 176
 177 // Callback invoked when we receive a message.  Here we check the max
 178 // receive message size.
 179 static void recv_message_ready(void* user_data, grpc_error* error) {
 180   grpc_call_element* elem = static_cast<grpc_call_element*>(user_data);
 181   call_data* calld = static_cast<call_data*>(elem->call_data);
 182   if (*calld->recv_message != nullptr && calld->limits.max_recv_size >= 0 &&
 183       (*calld->recv_message)->length() >
 184           static_cast<size_t>(calld->limits.max_recv_size)) {
 185     char* message_string;
 186     gpr_asprintf(&message_string,
 187                  "Received message larger than max (%u vs. %d)",
 188                  (*calld->recv_message)->length(), calld->limits.max_recv_size);
 189     grpc_error* new_error = grpc_error_set_int(
 190         GRPC_ERROR_CREATE_FROM_COPIED_STRING(message_string),
 191         GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED);
 192     GRPC_ERROR_UNREF(calld->error);
 193     if (error == GRPC_ERROR_NONE) {
 194       error = new_error;
 195     } else {
 196       error = grpc_error_add_child(error, new_error);
 197     }
 198     calld->error = GRPC_ERROR_REF(error);
 199     gpr_free(message_string);
 200   } else {
 201     GRPC_ERROR_REF(error);
 202   }
 203   // Invoke the next callback.
 204   grpc_closure* closure = calld->next_recv_message_ready;
 205   calld->next_recv_message_ready = nullptr;
 206   if (calld->seen_recv_trailing_metadata) {
 207     /* We might potentially see another RECV_MESSAGE op. In that case, we do not
 208      * want to run the recv_trailing_metadata_ready closure again. The newer
 209      * RECV_MESSAGE op cannot cause any errors since the transport has already
 210      * invoked the recv_trailing_metadata_ready closure and all further
 211      * RECV_MESSAGE ops will get null payloads. */
 212     calld->seen_recv_trailing_metadata = false;
 213     GRPC_CALL_COMBINER_START(calld->call_combiner,
 214                              &calld->recv_trailing_metadata_ready,
 215                              calld->recv_trailing_metadata_error,
 216                              "continue recv_trailing_metadata_ready");
 217   }
 218   GRPC_CLOSURE_RUN(closure, error);
 219 }
 220
 221 // Callback invoked on completion of recv_trailing_metadata
 222 // Notifies the recv_trailing_metadata batch of any message size failures
 223 static void recv_trailing_metadata_ready(void* user_data, grpc_error* error) {
 224   grpc_call_element* elem = static_cast<grpc_call_element*>(user_data);
 225   call_data* calld = static_cast<call_data*>(elem->call_data);
 226   if (calld->next_recv_message_ready != nullptr) {
 227     calld->seen_recv_trailing_metadata = true;
 228     calld->recv_trailing_metadata_error = GRPC_ERROR_REF(error);
 229     GRPC_CALL_COMBINER_STOP(calld->call_combiner,
 230                             "deferring recv_trailing_metadata_ready until "
 231                             "after recv_message_ready");
 232     return;
 233   }
 234   error =
 235       grpc_error_add_child(GRPC_ERROR_REF(error), GRPC_ERROR_REF(calld->error));
 236   // Invoke the next callback.
 237   GRPC_CLOSURE_RUN(calld->original_recv_trailing_metadata_ready, error);
 238 }
 239
 240 // Start transport stream op.
 241 static void start_transport_stream_op_batch(
 242     grpc_call_element* elem, grpc_transport_stream_op_batch* op) {
 243   call_data* calld = static_cast<call_data*>(elem->call_data);
 244   // Check max send message size.
 245   if (op->send_message && calld->limits.max_send_size >= 0 &&
 246       op->payload->send_message.send_message->length() >
 247           static_cast<size_t>(calld->limits.max_send_size)) {
 248     char* message_string;
 249     gpr_asprintf(&message_string, "Sent message larger than max (%u vs. %d)",
 250                  op->payload->send_message.send_message->length(),
 251                  calld->limits.max_send_size);
 252     grpc_transport_stream_op_batch_finish_with_failure(
 253         op,
 254         grpc_error_set_int(GRPC_ERROR_CREATE_FROM_COPIED_STRING(message_string),
 255                            GRPC_ERROR_INT_GRPC_STATUS,
 256                            GRPC_STATUS_RESOURCE_EXHAUSTED),
 257         calld->call_combiner);
 258     gpr_free(message_string);
 259     return;
 260   }
 261   // Inject callback for receiving a message.
 262   if (op->recv_message) {
 263     calld->next_recv_message_ready =
 264         op->payload->recv_message.recv_message_ready;
 265     calld->recv_message = op->payload->recv_message.recv_message;
 266     op->payload->recv_message.recv_message_ready = &calld->recv_message_ready;
 267   }
 268   // Inject callback for receiving trailing metadata.
 269   if (op->recv_trailing_metadata) {
 270     calld->original_recv_trailing_metadata_ready =
 271         op->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
 272     op->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
 273         &calld->recv_trailing_metadata_ready;
 274   }
 275   // Chain to the next filter.
 276   grpc_call_next_op(elem, op);
 277 }
 278
 279 // Constructor for call_data.
 280 static grpc_error* init_call_elem(grpc_call_element* elem,
 281                                   const grpc_call_element_args* args) {
 282   channel_data* chand = static_cast<channel_data*>(elem->channel_data);
 283   new (elem->call_data) call_data(elem, *chand, *args);
 284   return GRPC_ERROR_NONE;
 285 }
 286
 287 // Destructor for call_data.
 288 static void destroy_call_elem(grpc_call_element* elem,
 289                               const grpc_call_final_info* final_info,
 290                               grpc_closure* ignored) {
 291   call_data* calld = (call_data*)elem->call_data;
 292   calld->~call_data();
 293 }
 294
 295 static int default_size(const grpc_channel_args* args,
 296                         int without_minimal_stack) {
 297   if (grpc_channel_args_want_minimal_stack(args)) {
 298     return -1;
 299   }
 300   return without_minimal_stack;
 301 }
 302
 303 grpc_core::MessageSizeParsedConfig::message_size_limits get_message_size_limits(
 304     const grpc_channel_args* channel_args) {
 305   grpc_core::MessageSizeParsedConfig::message_size_limits lim;
 306   lim.max_send_size =
 307       default_size(channel_args, GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH);
 308   lim.max_recv_size =
 309       default_size(channel_args, GRPC_DEFAULT_MAX_RECV_MESSAGE_LENGTH);
 310   for (size_t i = 0; i < channel_args->num_args; ++i) {
 311     if (strcmp(channel_args->args[i].key, GRPC_ARG_MAX_SEND_MESSAGE_LENGTH) ==
 312         0) {
 313       const grpc_integer_options options = {lim.max_send_size, -1, INT_MAX};
 314       lim.max_send_size =
 315           grpc_channel_arg_get_integer(&channel_args->args[i], options);
 316     }
 317     if (strcmp(channel_args->args[i].key,
 318                GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH) == 0) {
 319       const grpc_integer_options options = {lim.max_recv_size, -1, INT_MAX};
 320       lim.max_recv_size =
 321           grpc_channel_arg_get_integer(&channel_args->args[i], options);
 322     }
 323   }
 324   return lim;
 325 }
 326
 327 // Constructor for channel_data.
 328 static grpc_error* init_channel_elem(grpc_channel_element* elem,
 329                                      grpc_channel_element_args* args) {
 330   GPR_ASSERT(!args->is_last);
 331   channel_data* chand = static_cast<channel_data*>(elem->channel_data);
 332   new (chand) channel_data();
 333   chand->limits = get_message_size_limits(args->channel_args);
 334   // TODO(yashykt): We only need to read GRPC_ARG_SERVICE_CONFIG in the case of
 335   // direct channels. (Service config is otherwise stored in the call_context by
 336   // client_channel filter.) If we ever need a second filter that also needs to
 337   // parse GRPC_ARG_SERVICE_CONFIG, we should refactor this code and add a
 338   // separate filter that reads GRPC_ARG_SERVICE_CONFIG and saves the parsed
 339   // config in the call_context.
 340   const grpc_arg* channel_arg =
 341       grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVICE_CONFIG);
 342   const char* service_config_str = grpc_channel_arg_get_string(channel_arg);
 343   if (service_config_str != nullptr) {
 344     grpc_error* service_config_error = GRPC_ERROR_NONE;
 345     auto svc_cfg = grpc_core::ServiceConfig::Create(service_config_str,
 346                                                     &service_config_error);
 347     if (service_config_error == GRPC_ERROR_NONE) {
 348       chand->svc_cfg = std::move(svc_cfg);
 349     } else {
 350       gpr_log(GPR_ERROR, "%s", grpc_error_string(service_config_error));
 351     }
 352     GRPC_ERROR_UNREF(service_config_error);
 353   }
 354   return GRPC_ERROR_NONE;
 355 }
 356
 357 // Destructor for channel_data.
 358 static void destroy_channel_elem(grpc_channel_element* elem) {
 359   channel_data* chand = static_cast<channel_data*>(elem->channel_data);
 360   chand->~channel_data();
 361 }
 362
 363 const grpc_channel_filter grpc_message_size_filter = {
 364     start_transport_stream_op_batch,
 365     grpc_channel_next_op,
 366     sizeof(call_data),
 367     init_call_elem,
 368     grpc_call_stack_ignore_set_pollset_or_pollset_set,
 369     destroy_call_elem,
 370     sizeof(channel_data),
 371     init_channel_elem,
 372     destroy_channel_elem,
 373     grpc_channel_next_get_info,
 374     "message_size"};
 375
 376 // Used for GRPC_CLIENT_SUBCHANNEL
 377 static bool maybe_add_message_size_filter_subchannel(
 378     grpc_channel_stack_builder* builder, void* arg) {
 379   const grpc_channel_args* channel_args =
 380       grpc_channel_stack_builder_get_channel_arguments(builder);
 381   if (grpc_channel_args_want_minimal_stack(channel_args)) {
 382     return true;
 383   }
 384   return grpc_channel_stack_builder_prepend_filter(
 385       builder, &grpc_message_size_filter, nullptr, nullptr);
 386 }
 387
 388 // Used for GRPC_CLIENT_DIRECT_CHANNEL and GRPC_SERVER_CHANNEL. Adds the filter
 389 // only if message size limits or service config is specified.
 390 static bool maybe_add_message_size_filter(grpc_channel_stack_builder* builder,
 391                                           void* arg) {
 392   const grpc_channel_args* channel_args =
 393       grpc_channel_stack_builder_get_channel_arguments(builder);
 394   bool enable = false;
 395   grpc_core::MessageSizeParsedConfig::message_size_limits lim =
 396       get_message_size_limits(channel_args);
 397   if (lim.max_send_size != -1 || lim.max_recv_size != -1) {
 398     enable = true;
 399   }
 400   const grpc_arg* a =
 401       grpc_channel_args_find(channel_args, GRPC_ARG_SERVICE_CONFIG);
 402   const char* svc_cfg_str = grpc_channel_arg_get_string(a);
 403   if (svc_cfg_str != nullptr) {
 404     enable = true;
 405   }
 406   if (enable) {
 407     return grpc_channel_stack_builder_prepend_filter(
 408         builder, &grpc_message_size_filter, nullptr, nullptr);
 409   } else {
 410     return true;
 411   }
 412 }
 413
 414 void grpc_message_size_filter_init(void) {
 415   grpc_channel_init_register_stage(
 416       GRPC_CLIENT_SUBCHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
 417       maybe_add_message_size_filter_subchannel, nullptr);
 418   grpc_channel_init_register_stage(GRPC_CLIENT_DIRECT_CHANNEL,
 419                                    GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
 420                                    maybe_add_message_size_filter, nullptr);
 421   grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL,
 422                                    GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
 423                                    maybe_add_message_size_filter, nullptr);
 424   grpc_core::MessageSizeParser::Register();
 425 }
 426
 427 void grpc_message_size_filter_shutdown(void) {}