diff options
author | David Garcia Quintas <dgq@google.com> | 2016-04-23 00:17:05 -0700 |
---|---|---|
committer | David Garcia Quintas <dgq@google.com> | 2016-04-23 00:17:05 -0700 |
commit | 73dcbda5b006fac4651480f7692b30db993b79b9 (patch) | |
tree | ab5ce0d90300dedfdbc1954679f5cf3c4bcd1583 | |
parent | e52be8c8a1e35e5fd709bfe99f95a12f90029aed (diff) |
Validation for incoming compressed data
-rw-r--r-- | include/grpc/compression.h | 3 | ||||
-rw-r--r-- | src/core/lib/channel/channel_args.c | 12 | ||||
-rw-r--r-- | src/core/lib/surface/call.c | 133 | ||||
-rw-r--r-- | src/core/lib/surface/channel.c | 16 | ||||
-rw-r--r-- | src/core/lib/surface/channel.h | 4 | ||||
-rw-r--r-- | test/core/channel/channel_args_test.c | 2 | ||||
-rw-r--r-- | test/core/end2end/tests/compressed_payload.c | 178 |
7 files changed, 318 insertions, 30 deletions
diff --git a/include/grpc/compression.h b/include/grpc/compression.h index 8de4b133d4..22bcf0e302 100644 --- a/include/grpc/compression.h +++ b/include/grpc/compression.h @@ -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); diff --git a/src/core/lib/channel/channel_args.c b/src/core/lib/channel/channel_args.c index 28d2d78d00..d6b95fe644 100644 --- a/src/core/lib/channel/channel_args.c +++ b/src/core/lib/channel/channel_args.c @@ -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 { diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c index 6581bbd3d1..2462adc26a 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -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, diff --git a/src/core/lib/surface/channel.c b/src/core/lib/surface/channel.c index b6b760b5d8..72d0cfab2b 100644 --- a/src/core/lib/surface/channel.c +++ b/src/core/lib/surface/channel.c @@ -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) { diff --git a/src/core/lib/surface/channel.h b/src/core/lib/surface/channel.h index 22dae930e4..8f153fbc55 100644 --- a/src/core/lib/surface/channel.h +++ b/src/core/lib/surface/channel.h @@ -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. diff --git a/test/core/channel/channel_args_test.c b/test/core/channel/channel_args_test.c index c7fc25960c..1aa8c5dd2a 100644 --- a/test/core/channel/channel_args_test.c +++ b/test/core/channel/channel_args_test.c @@ -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; } diff --git a/test/core/end2end/tests/compressed_payload.c b/test/core/end2end/tests/compressed_payload.c index 589bc314f8..b713d399c0 100644 --- a/test/core/end2end/tests/compressed_payload.c +++ b/test/core/end2end/tests/compressed_payload.c @@ -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) {} |