Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Validation for incoming compressed data

Browse Source
pull/6243/head
David Garcia Quintas 9 years ago
parent e52be8c8a1
commit 73dcbda5b0
7 changed files with 318 additions and 30 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 3
      include/grpc/compression.h
  2. 12
      src/core/lib/channel/channel_args.c
  3. 133
      src/core/lib/surface/call.c
  4. 16
      src/core/lib/surface/channel.c
  5. 4
      src/core/lib/surface/channel.h
  6. 2
      test/core/channel/channel_args_test.c
  7. 178
      test/core/end2end/tests/compressed_payload.c

3
include/grpc/compression.h
Unescape View File

@ -51,7 +51,8 @@ GRPCAPI int grpc_compression_algorithm_parse(
grpc_compression_algorithm *algorithm);
/** Updates \a name with the encoding name corresponding to a valid \a
* algorithm. Returns 1 upon success, 0 otherwise. */
* algorithm. Note that \a name is statically allocated and must *not* be freed.
* Returns 1 upon success, 0 otherwise. */
GRPCAPI int grpc_compression_algorithm_name(
grpc_compression_algorithm algorithm, char **name);

12
src/core/lib/channel/channel_args.c
Unescape View File

@ -35,6 +35,7 @@
#include <grpc/grpc.h>
#include "src/core/lib/support/string.h"
#include <grpc/compression.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
@ -180,6 +181,7 @@ grpc_compression_algorithm grpc_channel_args_get_compression_algorithm(
grpc_channel_args *grpc_channel_args_set_compression_algorithm(
grpc_channel_args *a, grpc_compression_algorithm algorithm) {
GPR_ASSERT(algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT);
grpc_arg tmp;
tmp.type = GRPC_ARG_INTEGER;
tmp.key = GRPC_COMPRESSION_ALGORITHM_ARG;
@ -212,7 +214,15 @@ grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
const int states_arg_found =
find_compression_algorithm_states_bitset(*a, &states_arg);
if (states_arg_found) {
if (grpc_channel_args_get_compression_algorithm(*a) == algorithm &&
state == 0) {
char *algo_name = NULL;
GPR_ASSERT(grpc_compression_algorithm_name(algorithm, &algo_name) != 0);
gpr_log(GPR_ERROR,
"Tried to disable default compression algorithm '%s'. The "
"operation has been ignored.",
algo_name);
} else if (states_arg_found) {
if (state != 0) {
GPR_BITSET((unsigned *)states_arg, algorithm);
} else {

133
src/core/lib/surface/call.c
Unescape View File

@ -40,6 +40,7 @@
#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/slice.h>
#include <grpc/support/string_util.h>
#include <grpc/support/useful.h>
@ -52,7 +53,9 @@
#include "src/core/lib/surface/call.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/surface/completion_queue.h"
#include "src/core/lib/transport/metadata.h"
#include "src/core/lib/transport/static_metadata.h"
#include "src/core/lib/transport/transport.h"
/** The maximum number of concurrent batches possible.
Based upon the maximum number of individually queueable ops in the batch
@ -240,6 +243,9 @@ static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,
static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
grpc_status_code status,
const char *description);
static grpc_call_error close_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
grpc_status_code status,
const char *description);
static void destroy_call(grpc_exec_ctx *exec_ctx, void *call_stack,
bool success);
static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
@ -410,7 +416,30 @@ static void set_status_code(grpc_call *call, status_source source,
static void set_compression_algorithm(grpc_call *call,
grpc_compression_algorithm algo) {
call->compression_algorithm = algo;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
char *error_msg = NULL;
const grpc_compression_options compression_options =
grpc_channel_get_compression_options(call->channel);
/* check if algorithm is known */
if (algo >= GRPC_COMPRESS_ALGORITHMS_COUNT) {
gpr_asprintf(&error_msg, "Invalid compression algorithm value '%d'.", algo);
gpr_log(GPR_ERROR, error_msg);
close_with_status(&exec_ctx, call, GRPC_STATUS_INTERNAL, error_msg);
} else if (grpc_compression_options_is_algorithm_enabled(&compression_options,
algo) == 0) {
/* check if algorithm is supported by current channel config */
char *algo_name;
grpc_compression_algorithm_name(algo, &algo_name);
gpr_asprintf(&error_msg, "Compression algorithm '%s' is disabled.",
algo_name);
gpr_log(GPR_ERROR, error_msg);
close_with_status(&exec_ctx, call, GRPC_STATUS_INTERNAL, error_msg);
} else {
call->compression_algorithm = algo;
}
gpr_free(error_msg);
grpc_exec_ctx_finish(&exec_ctx);
}
grpc_compression_algorithm grpc_call_test_only_get_compression_algorithm(
@ -694,48 +723,102 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
return r;
}
typedef struct cancel_closure {
typedef struct termination_closure {
grpc_closure closure;
grpc_call *call;
grpc_status_code status;
} cancel_closure;
gpr_slice optional_message;
grpc_closure *op_closure;
enum { TC_CANCEL, TC_CLOSE } type;
} termination_closure;
static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp, bool success) {
termination_closure *tc = tcp;
if (tc->type == TC_CANCEL) {
GRPC_CALL_INTERNAL_UNREF(exec_ctx, tc->call, "cancel");
}
if (tc->type == TC_CLOSE) {
GRPC_CALL_INTERNAL_UNREF(exec_ctx, tc->call, "close");
}
gpr_slice_unref(tc->optional_message);
if (tc->op_closure != NULL) {
grpc_exec_ctx_enqueue(exec_ctx, tc->op_closure, false, NULL);
}
gpr_free(tc);
}
static void done_cancel(grpc_exec_ctx *exec_ctx, void *ccp, bool success) {
cancel_closure *cc = ccp;
GRPC_CALL_INTERNAL_UNREF(exec_ctx, cc->call, "cancel");
gpr_free(cc);
static void send_cancel(grpc_exec_ctx *exec_ctx, void *tcp, bool success) {
grpc_transport_stream_op op;
termination_closure *tc = tcp;
memset(&op, 0, sizeof(op));
op.cancel_with_status = tc->status;
/* reuse closure to catch completion */
grpc_closure_init(&tc->closure, done_termination, tc);
op.on_complete = &tc->closure;
execute_op(exec_ctx, tc->call, &op);
}
static void send_cancel(grpc_exec_ctx *exec_ctx, void *ccp, bool success) {
static void send_close(grpc_exec_ctx *exec_ctx, void *tcp, bool success) {
grpc_transport_stream_op op;
cancel_closure *cc = ccp;
termination_closure *tc = tcp;
memset(&op, 0, sizeof(op));
op.cancel_with_status = cc->status;
tc->optional_message = gpr_slice_ref(tc->optional_message);
grpc_transport_stream_op_add_close(&op, tc->status, &tc->optional_message);
/* reuse closure to catch completion */
grpc_closure_init(&cc->closure, done_cancel, cc);
op.on_complete = &cc->closure;
execute_op(exec_ctx, cc->call, &op);
grpc_closure_init(&tc->closure, done_termination, tc);
tc->op_closure = op.on_complete;
op.on_complete = &tc->closure;
execute_op(exec_ctx, tc->call, &op);
}
static grpc_call_error terminate_with_status(grpc_exec_ctx *exec_ctx,
termination_closure *tc) {
grpc_mdstr *details = NULL;
if (GPR_SLICE_LENGTH(tc->optional_message) > 0) {
tc->optional_message = gpr_slice_ref(tc->optional_message);
details = grpc_mdstr_from_slice(tc->optional_message);
}
set_status_code(tc->call, STATUS_FROM_API_OVERRIDE, (uint32_t)tc->status);
set_status_details(tc->call, STATUS_FROM_API_OVERRIDE, details);
if (tc->type == TC_CANCEL) {
grpc_closure_init(&tc->closure, send_cancel, tc);
GRPC_CALL_INTERNAL_REF(tc->call, "cancel");
} else if (tc->type == TC_CLOSE) {
grpc_closure_init(&tc->closure, send_close, tc);
GRPC_CALL_INTERNAL_REF(tc->call, "close");
}
grpc_exec_ctx_enqueue(exec_ctx, &tc->closure, true, NULL);
return GRPC_CALL_OK;
}
static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
grpc_status_code status,
const char *description) {
grpc_mdstr *details =
description ? grpc_mdstr_from_string(description) : NULL;
cancel_closure *cc = gpr_malloc(sizeof(*cc));
termination_closure *tc = gpr_malloc(sizeof(*tc));
memset(tc, 0, sizeof(termination_closure));
tc->type = TC_CANCEL;
tc->call = c;
tc->optional_message = gpr_slice_from_copied_string(description);
GPR_ASSERT(status != GRPC_STATUS_OK);
tc->status = status;
set_status_code(c, STATUS_FROM_API_OVERRIDE, (uint32_t)status);
set_status_details(c, STATUS_FROM_API_OVERRIDE, details);
return terminate_with_status(exec_ctx, tc);
}
grpc_closure_init(&cc->closure, send_cancel, cc);
cc->call = c;
cc->status = status;
GRPC_CALL_INTERNAL_REF(c, "cancel");
grpc_exec_ctx_enqueue(exec_ctx, &cc->closure, true, NULL);
static grpc_call_error close_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
grpc_status_code status,
const char *description) {
termination_closure *tc = gpr_malloc(sizeof(*tc));
memset(tc, 0, sizeof(termination_closure));
tc->type = TC_CLOSE;
tc->call = c;
tc->optional_message = gpr_slice_from_copied_string(description);
GPR_ASSERT(status != GRPC_STATUS_OK);
tc->status = status;
return GRPC_CALL_OK;
return terminate_with_status(exec_ctx, tc);
}
static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,

16
src/core/lib/surface/channel.c
Unescape View File

@ -36,16 +36,17 @@
#include <stdlib.h>
#include <string.h>
#include <grpc/compression.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/iomgr/iomgr.h"
#include "src/core/lib/support/string.h"
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/call.h"
#include "src/core/lib/surface/channel_init.h"
#include "src/core/lib/surface/init.h"
#include "src/core/lib/transport/static_metadata.h"
/** Cache grpc-status: X mdelems for X = 0..NUM_CACHED_STATUS_ELEMS.
@ -64,6 +65,7 @@ typedef struct registered_call {
struct grpc_channel {
int is_client;
uint32_t max_message_length;
grpc_compression_options compression_options;
grpc_mdelem *default_authority;
gpr_mu registered_call_mu;
@ -111,6 +113,7 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
channel->registered_calls = NULL;
channel->max_message_length = DEFAULT_MAX_MESSAGE_LENGTH;
grpc_compression_options_init(&channel->compression_options);
if (args) {
for (size_t i = 0; i < args->num_args; i++) {
if (0 == strcmp(args->args[i].key, GRPC_ARG_MAX_MESSAGE_LENGTH)) {
@ -153,6 +156,12 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
}
}
}
/* extract compression options */
channel->compression_options.enabled_algorithms_bitset =
(uint32_t)grpc_channel_args_compression_algorithm_get_states(args);
channel->compression_options.default_compression_algorithm =
grpc_channel_args_get_compression_algorithm(args);
grpc_channel_args_destroy(args);
}
@ -306,6 +315,11 @@ grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel) {
return CHANNEL_STACK_FROM_CHANNEL(channel);
}
grpc_compression_options grpc_channel_get_compression_options(
const grpc_channel *channel) {
return channel->compression_options;
}
grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int i) {
char tmp[GPR_LTOA_MIN_BUFSIZE];
switch (i) {

4
src/core/lib/surface/channel.h
Unescape View File

@ -45,6 +45,10 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
/** Get a (borrowed) pointer to this channels underlying channel stack */
grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel);
/** Return the compression options for \a channel */
grpc_compression_options grpc_channel_get_compression_options(
const grpc_channel *channel);
/** Get a grpc_mdelem of grpc-status: X where X is the numeric value of
status_code.

2
test/core/channel/channel_args_test.c
Unescape View File

@ -135,8 +135,10 @@ static void test_compression_algorithm_states(void) {
int main(int argc, char **argv) {
grpc_test_init(argc, argv);
grpc_init();
test_create();
test_set_compression_algorithm();
test_compression_algorithm_states();
grpc_shutdown();
return 0;
}

178
test/core/end2end/tests/compressed_payload.c
Unescape View File

@ -38,8 +38,10 @@
#include <grpc/byte_buffer.h>
#include <grpc/byte_buffer_reader.h>
#include <grpc/compression.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include <grpc/support/time.h>
#include <grpc/support/useful.h>
@ -102,6 +104,170 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_destroy(f->cq);
}
static void request_for_disabled_algorithm(
grpc_end2end_test_config config, const char *test_name,
uint32_t send_flags_bitmask,
grpc_compression_algorithm algorithm_to_disable,
grpc_compression_algorithm requested_client_compression_algorithm,
grpc_status_code expected_error, grpc_metadata *client_metadata) {
grpc_call *c;
grpc_call *s;
gpr_slice request_payload_slice;
grpc_byte_buffer *request_payload;
gpr_timespec deadline = five_seconds_time();
grpc_channel_args *client_args;
grpc_channel_args *server_args;
grpc_end2end_test_fixture f;
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
grpc_byte_buffer *request_payload_recv = NULL;
grpc_call_details call_details;
grpc_status_code status;
grpc_call_error error;
char *details = NULL;
size_t details_capacity = 0;
int was_cancelled = 2;
cq_verifier *cqv;
char str[1024];
memset(str, 'x', 1023);
str[1023] = '\0';
request_payload_slice = gpr_slice_from_copied_string(str);
request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1);
client_args = grpc_channel_args_set_compression_algorithm(
NULL, requested_client_compression_algorithm);
server_args =
grpc_channel_args_set_compression_algorithm(NULL, GRPC_COMPRESS_NONE);
server_args = grpc_channel_args_compression_algorithm_set_state(
&server_args, algorithm_to_disable, false);
f = begin_test(config, test_name, client_args, server_args);
cqv = cq_verifier_create(f.cq);
c = grpc_channel_create_call(f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq,
"/foo", "foo.test.google.fr", deadline, NULL);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
grpc_metadata_array_init(&trailing_metadata_recv);
grpc_metadata_array_init(&request_metadata_recv);
grpc_call_details_init(&call_details);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
if (client_metadata != NULL) {
op->data.send_initial_metadata.count = 1;
op->data.send_initial_metadata.metadata = client_metadata;
} else {
op->data.send_initial_metadata.count = 0;
}
op->flags = 0;
op->reserved = NULL;
op++;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message = request_payload;
op->flags = send_flags_bitmask;
op->reserved = NULL;
op++;
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
op->flags = 0;
op->reserved = NULL;
op++;
op->op = GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata = &initial_metadata_recv;
op->flags = 0;
op->reserved = NULL;
op++;
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
op->data.recv_status_on_client.status = &status;
op->data.recv_status_on_client.status_details = &details;
op->data.recv_status_on_client.status_details_capacity = &details_capacity;
op->flags = 0;
op->reserved = NULL;
op++;
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL);
GPR_ASSERT(GRPC_CALL_OK == error);
error =
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq, f.cq, tag(101));
GPR_ASSERT(GRPC_CALL_OK == error);
cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op->flags = 0;
op->reserved = NULL;
op++;
op->op = GRPC_OP_RECV_MESSAGE;
op->data.recv_message = &request_payload_recv;
op->flags = 0;
op->reserved = NULL;
op++;
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), NULL);
GPR_ASSERT(GRPC_CALL_OK == error);
cq_expect_completion(cqv, tag(102), 0);
cq_verify(cqv);
op = ops;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op->flags = 0;
op->reserved = NULL;
op++;
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), NULL);
GPR_ASSERT(GRPC_CALL_OK == error);
cq_expect_completion(cqv, tag(103), 1);
cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
/* call was cancelled (closed) ... */
GPR_ASSERT(was_cancelled != 0);
/* with a certain error */
GPR_ASSERT(status == expected_error);
char *algo_name = NULL;
GPR_ASSERT(grpc_compression_algorithm_name(algorithm_to_disable, &algo_name));
char *expected_details = NULL;
gpr_asprintf(&expected_details, "Compression algorithm '%s' is disabled.",
algo_name);
/* and we expect a specific reason for it */
GPR_ASSERT(0 == strcmp(details, expected_details));
gpr_free(expected_details);
GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr"));
gpr_free(details);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
grpc_call_destroy(c);
grpc_call_destroy(s);
cq_verifier_destroy(cqv);
gpr_slice_unref(request_payload_slice);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(request_payload_recv);
grpc_channel_args_destroy(client_args);
grpc_channel_args_destroy(server_args);
end_test(&f);
config.tear_down_data(&f);
}
static void request_with_payload_template(
grpc_end2end_test_config config, const char *test_name,
uint32_t send_flags_bitmask,
@ -196,8 +362,8 @@ static void request_with_payload_template(
cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
GPR_ASSERT(
GPR_BITCOUNT(grpc_call_test_only_get_encodings_accepted_by_peer(s)) == 3);
GPR_ASSERT(GPR_BITCOUNT(grpc_call_test_only_get_encodings_accepted_by_peer(
s)) == GRPC_COMPRESS_ALGORITHMS_COUNT);
GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s),
GRPC_COMPRESS_NONE) != 0);
GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s),
@ -330,11 +496,19 @@ static void test_invoke_request_with_compressed_payload_md_override(
GRPC_COMPRESS_DEFLATE, GRPC_COMPRESS_NONE, &none_compression_override);
}
static void test_invoke_request_with_invalid_algorithm(
grpc_end2end_test_config config) {
request_for_disabled_algorithm(
config, "test_invoke_request_with_invalid_algorithm", 0,
GRPC_COMPRESS_GZIP, GRPC_COMPRESS_GZIP, GRPC_STATUS_INTERNAL, NULL);
}
void compressed_payload(grpc_end2end_test_config config) {
test_invoke_request_with_exceptionally_uncompressed_payload(config);
test_invoke_request_with_uncompressed_payload(config);
test_invoke_request_with_compressed_payload(config);
test_invoke_request_with_compressed_payload_md_override(config);
test_invoke_request_with_invalid_algorithm(config);
}
void compressed_payload_pre_init(void) {}

Write Preview
Loading…
Cancel
Save