From bc745a5f4e30590cdcba1c600b51e4e9d38ad360 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Mon, 15 Jun 2015 18:29:23 -0700 Subject: [PATCH 1/5] fix json_rewrite_test CRLF handling --- test/core/json/json_rewrite_test.c | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/test/core/json/json_rewrite_test.c b/test/core/json/json_rewrite_test.c index ec6deebe76c..fa46830d5ba 100644 --- a/test/core/json/json_rewrite_test.c +++ b/test/core/json/json_rewrite_test.c @@ -64,6 +64,11 @@ typedef struct json_reader_userdata { static void json_writer_output_char(void* userdata, char c) { json_writer_userdata* state = userdata; int cmp = fgetc(state->cmp); + + /* treat CRLF as LF */ + if (cmp == '\r' && c == '\n') { + cmp = fgetc(state->cmp); + } GPR_ASSERT(cmp == c); } From d1eb527f78bbe962b02871e54c917b6c4dc984e2 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Tue, 16 Jun 2015 13:15:27 -0700 Subject: [PATCH 2/5] fix indentation --- test/core/json/json_rewrite_test.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/core/json/json_rewrite_test.c b/test/core/json/json_rewrite_test.c index fa46830d5ba..f5859322ea1 100644 --- a/test/core/json/json_rewrite_test.c +++ b/test/core/json/json_rewrite_test.c @@ -67,7 +67,7 @@ static void json_writer_output_char(void* userdata, char c) { /* treat CRLF as LF */ if (cmp == '\r' && c == '\n') { - cmp = fgetc(state->cmp); + cmp = fgetc(state->cmp); } GPR_ASSERT(cmp == c); } From d1531323e7dc5ee8e4f562bc251d1de5eb0a8b0c Mon Sep 17 00:00:00 2001 From: Julien Boeuf Date: Thu, 18 Jun 2015 11:05:39 +0200 Subject: [PATCH 3/5] Adding NPN support if ALPN support is not available. - ALPN may not be available: - because OpenSSL does not support it. - because one of the peers does not support it. - Tested manually end to end by forcing TSI_OPENSSL_ALPN_SUPPORT to 0. Also tested if only one of the peers supports ALPN. --- src/core/tsi/ssl_transport_security.c | 124 ++++++++++++++++++-------- 1 file changed, 87 insertions(+), 37 deletions(-) diff --git a/src/core/tsi/ssl_transport_security.c b/src/core/tsi/ssl_transport_security.c index 63b4c42131b..bd4c15a0686 100644 --- a/src/core/tsi/ssl_transport_security.c +++ b/src/core/tsi/ssl_transport_security.c @@ -54,6 +54,10 @@ #define TSI_SSL_MAX_PROTECTED_FRAME_SIZE_UPPER_BOUND 16384 #define TSI_SSL_MAX_PROTECTED_FRAME_SIZE_LOWER_BOUND 1024 +#ifndef TSI_OPENSSL_ALPN_SUPPORT +#define TSI_OPENSSL_ALPN_SUPPORT 1 +#endif + /* TODO(jboeuf): I have not found a way to get this number dynamically from the * SSL structure. This is what we would ultimately want though... */ #define TSI_SSL_MAX_PROTECTION_OVERHEAD 100 @@ -70,6 +74,8 @@ struct tsi_ssl_handshaker_factory { typedef struct { tsi_ssl_handshaker_factory base; SSL_CTX* ssl_context; + unsigned char* alpn_protocol_list; + size_t alpn_protocol_list_length; } tsi_ssl_client_handshaker_factory; typedef struct { @@ -841,7 +847,7 @@ static tsi_result ssl_handshaker_process_bytes_from_peer( static tsi_result ssl_handshaker_extract_peer(tsi_handshaker* self, tsi_peer* peer) { tsi_result result = TSI_OK; - const unsigned char* alpn_selected; + const unsigned char* alpn_selected = NULL; unsigned int alpn_selected_len; tsi_ssl_handshaker* impl = (tsi_ssl_handshaker*)self; X509* peer_cert = SSL_get_peer_certificate(impl->ssl); @@ -850,7 +856,14 @@ static tsi_result ssl_handshaker_extract_peer(tsi_handshaker* self, X509_free(peer_cert); if (result != TSI_OK) return result; } +#if TSI_OPENSSL_ALPN_SUPPORT SSL_get0_alpn_selected(impl->ssl, &alpn_selected, &alpn_selected_len); +#endif /* TSI_OPENSSL_ALPN_SUPPORT */ + if (alpn_selected == NULL) { + /* Try npn. */ + SSL_get0_next_proto_negotiated(impl->ssl, &alpn_selected, + &alpn_selected_len); + } if (alpn_selected != NULL) { size_t i; tsi_peer_property* new_properties = @@ -1012,6 +1025,32 @@ static tsi_result create_tsi_ssl_handshaker(SSL_CTX* ctx, int is_client, return TSI_OK; } +static int select_protocol_list(const unsigned char** out, + unsigned char* outlen, + const unsigned char* client_list, + unsigned int client_list_len, + const unsigned char* server_list, + unsigned int server_list_len) { + const unsigned char* client_current = client_list; + while ((unsigned int)(client_current - client_list) < client_list_len) { + unsigned char client_current_len = *(client_current++); + const unsigned char* server_current = server_list; + while ((server_current >= server_list) && + (gpr_uintptr)(server_current - server_list) < server_list_len) { + unsigned char server_current_len = *(server_current++); + if ((client_current_len == server_current_len) && + !memcmp(client_current, server_current, server_current_len)) { + *out = server_current; + *outlen = server_current_len; + return SSL_TLSEXT_ERR_OK; + } + server_current += server_current_len; + } + client_current += client_current_len; + } + return SSL_TLSEXT_ERR_NOACK; +} + /* --- tsi_ssl__client_handshaker_factory methods implementation. --- */ static tsi_result ssl_client_handshaker_factory_create_handshaker( @@ -1027,10 +1066,21 @@ static void ssl_client_handshaker_factory_destroy( tsi_ssl_handshaker_factory* self) { tsi_ssl_client_handshaker_factory* impl = (tsi_ssl_client_handshaker_factory*)self; - SSL_CTX_free(impl->ssl_context); + if (impl->ssl_context != NULL) SSL_CTX_free(impl->ssl_context); + if (impl->alpn_protocol_list != NULL) free(impl->alpn_protocol_list); free(impl); } +static int client_handshaker_factory_npn_callback( + SSL* ssl, unsigned char** out, unsigned char* outlen, + const unsigned char* in, unsigned int inlen, void* arg) { + tsi_ssl_client_handshaker_factory* factory = + (tsi_ssl_client_handshaker_factory*)arg; + return select_protocol_list((const unsigned char**)out, outlen, + factory->alpn_protocol_list, + factory->alpn_protocol_list_length, in, inlen); +} + /* --- tsi_ssl_server_handshaker_factory methods implementation. --- */ static tsi_result ssl_server_handshaker_factory_create_handshaker( @@ -1134,30 +1184,25 @@ static int ssl_server_handshaker_factory_servername_callback(SSL* ssl, int* ap, return SSL_TLSEXT_ERR_ALERT_WARNING; } +#if TSI_OPENSSL_ALPN_SUPPORT static int server_handshaker_factory_alpn_callback( SSL* ssl, const unsigned char** out, unsigned char* outlen, const unsigned char* in, unsigned int inlen, void* arg) { tsi_ssl_server_handshaker_factory* factory = (tsi_ssl_server_handshaker_factory*)arg; - const unsigned char* client_current = in; - while ((unsigned int)(client_current - in) < inlen) { - unsigned char client_current_len = *(client_current++); - const unsigned char* server_current = factory->alpn_protocol_list; - while ((server_current >= factory->alpn_protocol_list) && - (gpr_uintptr)(server_current - factory->alpn_protocol_list) < - factory->alpn_protocol_list_length) { - unsigned char server_current_len = *(server_current++); - if ((client_current_len == server_current_len) && - !memcmp(client_current, server_current, server_current_len)) { - *out = server_current; - *outlen = server_current_len; - return SSL_TLSEXT_ERR_OK; - } - server_current += server_current_len; - } - client_current += client_current_len; - } - return SSL_TLSEXT_ERR_NOACK; + return select_protocol_list(out, outlen, in, inlen, + factory->alpn_protocol_list, + factory->alpn_protocol_list_length); +} +#endif /* TSI_OPENSSL_ALPN_SUPPORT */ + +static int server_handshaker_factory_npn_advertised_callback( + SSL* ssl, const unsigned char** out, unsigned int* outlen, void* arg) { + tsi_ssl_server_handshaker_factory* factory = + (tsi_ssl_server_handshaker_factory*)arg; + *out = factory->alpn_protocol_list; + *outlen = factory->alpn_protocol_list_length; + return SSL_TLSEXT_ERR_OK; } /* --- tsi_ssl_handshaker_factory constructors. --- */ @@ -1184,6 +1229,14 @@ tsi_result tsi_create_ssl_client_handshaker_factory( gpr_log(GPR_ERROR, "Could not create ssl context."); return TSI_INVALID_ARGUMENT; } + + impl = calloc(1, sizeof(tsi_ssl_client_handshaker_factory)); + if (impl == NULL) { + SSL_CTX_free(ssl_context); + return TSI_OUT_OF_RESOURCES; + } + impl->ssl_context = ssl_context; + do { result = populate_ssl_context(ssl_context, pem_private_key, pem_private_key_size, @@ -1197,41 +1250,33 @@ tsi_result tsi_create_ssl_client_handshaker_factory( } if (num_alpn_protocols != 0) { - unsigned char* alpn_protocol_list = NULL; - size_t alpn_protocol_list_length = 0; - int ssl_failed; result = build_alpn_protocol_name_list( alpn_protocols, alpn_protocols_lengths, num_alpn_protocols, - &alpn_protocol_list, &alpn_protocol_list_length); + &impl->alpn_protocol_list, &impl->alpn_protocol_list_length); if (result != TSI_OK) { gpr_log(GPR_ERROR, "Building alpn list failed with error %s.", tsi_result_to_string(result)); - free(alpn_protocol_list); break; } - ssl_failed = SSL_CTX_set_alpn_protos(ssl_context, alpn_protocol_list, - alpn_protocol_list_length); - free(alpn_protocol_list); - if (ssl_failed) { +#if TSI_OPENSSL_ALPN_SUPPORT + if (SSL_CTX_set_alpn_protos(ssl_context, impl->alpn_protocol_list, + impl->alpn_protocol_list_length)) { gpr_log(GPR_ERROR, "Could not set alpn protocol list to context."); result = TSI_INVALID_ARGUMENT; break; } +#endif /* TSI_OPENSSL_ALPN_SUPPORT */ + SSL_CTX_set_next_proto_select_cb( + ssl_context, client_handshaker_factory_npn_callback, impl); } } while (0); if (result != TSI_OK) { - SSL_CTX_free(ssl_context); + ssl_client_handshaker_factory_destroy(&impl->base); return result; } SSL_CTX_set_verify(ssl_context, SSL_VERIFY_PEER, NULL); /* TODO(jboeuf): Add revocation verification. */ - impl = calloc(1, sizeof(tsi_ssl_client_handshaker_factory)); - if (impl == NULL) { - SSL_CTX_free(ssl_context); - return TSI_OUT_OF_RESOURCES; - } - impl->ssl_context = ssl_context; impl->base.create_handshaker = ssl_client_handshaker_factory_create_handshaker; impl->base.destroy = ssl_client_handshaker_factory_destroy; @@ -1322,8 +1367,13 @@ tsi_result tsi_create_ssl_server_handshaker_factory( impl->ssl_contexts[i], ssl_server_handshaker_factory_servername_callback); SSL_CTX_set_tlsext_servername_arg(impl->ssl_contexts[i], impl); +#if TSI_OPENSSL_ALPN_SUPPORT SSL_CTX_set_alpn_select_cb(impl->ssl_contexts[i], server_handshaker_factory_alpn_callback, impl); +#endif /* TSI_OPENSSL_ALPN_SUPPORT */ + SSL_CTX_set_next_protos_advertised_cb( + impl->ssl_contexts[i], + server_handshaker_factory_npn_advertised_callback, impl); } while (0); if (result != TSI_OK) { From 3e29de19a3768f277a76e1da2f26ed17c591cc37 Mon Sep 17 00:00:00 2001 From: Julien Boeuf Date: Thu, 18 Jun 2015 23:29:13 +0200 Subject: [PATCH 4/5] Adressing comments. --- src/core/json/json.h | 2 +- src/core/tsi/ssl_transport_security.c | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/src/core/json/json.h b/src/core/json/json.h index 69cbac17dc3..b78b42a5b2f 100644 --- a/src/core/json/json.h +++ b/src/core/json/json.h @@ -60,7 +60,7 @@ typedef struct grpc_json { * strings in the tree. The input stream's UTF-8 isn't validated, * as in, what you input is what you get as an output. * - * All the keys and values in the grpc_json_t objects will be strings + * All the keys and values in the grpc_json objects will be strings * pointing at your input buffer. * * Delete the allocated tree afterward using grpc_json_destroy(). diff --git a/src/core/tsi/ssl_transport_security.c b/src/core/tsi/ssl_transport_security.c index bd4c15a0686..6156a39d093 100644 --- a/src/core/tsi/ssl_transport_security.c +++ b/src/core/tsi/ssl_transport_security.c @@ -54,12 +54,16 @@ #define TSI_SSL_MAX_PROTECTED_FRAME_SIZE_UPPER_BOUND 16384 #define TSI_SSL_MAX_PROTECTED_FRAME_SIZE_LOWER_BOUND 1024 + +/* Putting a macro like this and littering the source file with #if is really + bad practice. + TODO(jboeuf): refactor all the #if / #endif in a separate module. */ #ifndef TSI_OPENSSL_ALPN_SUPPORT #define TSI_OPENSSL_ALPN_SUPPORT 1 #endif /* TODO(jboeuf): I have not found a way to get this number dynamically from the - * SSL structure. This is what we would ultimately want though... */ + SSL structure. This is what we would ultimately want though... */ #define TSI_SSL_MAX_PROTECTION_OVERHEAD 100 /* --- Structure definitions. ---*/ From aa57bab3ca99f621f042b0a3985b36a0e0e2d6f6 Mon Sep 17 00:00:00 2001 From: Tim Emiola Date: Wed, 17 Jun 2015 11:52:51 -0700 Subject: [PATCH 5/5] Corrects the cancel_after_first_response behaviour - introduces a #wait method on the call operation view - invokes #wait on a Notifier that is created for all operations - ensures the Notifier is invoked if necessary whenever a client request completes - updates the interop_test to use op.wait before checking if the call was cancelled. --- src/ruby/.rubocop_todo.yml | 2 +- src/ruby/bin/interop/interop_client.rb | 3 ++- src/ruby/lib/grpc/generic/active_call.rb | 21 +++++++++++++++++++-- src/ruby/lib/grpc/generic/bidi_call.rb | 14 +++++++++++++- src/ruby/lib/grpc/logconfig.rb | 2 +- 5 files changed, 36 insertions(+), 6 deletions(-) diff --git a/src/ruby/.rubocop_todo.yml b/src/ruby/.rubocop_todo.yml index c35e970df66..05db4045825 100644 --- a/src/ruby/.rubocop_todo.yml +++ b/src/ruby/.rubocop_todo.yml @@ -12,7 +12,7 @@ Metrics/AbcSize: # Offense count: 3 # Configuration parameters: CountComments. Metrics/ClassLength: - Max: 192 + Max: 200 # Offense count: 35 # Configuration parameters: CountComments. diff --git a/src/ruby/bin/interop/interop_client.rb b/src/ruby/bin/interop/interop_client.rb index 16fb1b199dd..da4caa842b6 100755 --- a/src/ruby/bin/interop/interop_client.rb +++ b/src/ruby/bin/interop/interop_client.rb @@ -284,7 +284,8 @@ class NamedTests op = @stub.full_duplex_call(ppp.each_item, return_op: true) ppp.canceller_op = op # causes ppp to cancel after the 1st message op.execute.each { |r| ppp.queue.push(r) } - assert(op.cancelled, 'call operation should be CANCELLED') + op.wait + assert(op.cancelled, 'call operation was not CANCELLED') p 'OK: cancel_after_first_response' end diff --git a/src/ruby/lib/grpc/generic/active_call.rb b/src/ruby/lib/grpc/generic/active_call.rb index 3814ef34b41..215c0069a3f 100644 --- a/src/ruby/lib/grpc/generic/active_call.rb +++ b/src/ruby/lib/grpc/generic/active_call.rb @@ -120,6 +120,7 @@ module GRPC @started = started @unmarshal = unmarshal @metadata_tag = metadata_tag + @op_notifier = nil end # output_metadata are provides access to hash that can be used to @@ -148,6 +149,7 @@ module GRPC # operation provides a restricted view of this ActiveCall for use as # a Operation. def operation + @op_notifier = Notifier.new Operation.new(self) end @@ -167,6 +169,7 @@ module GRPC batch_result = @call.run_batch(@cq, self, INFINITE_FUTURE, ops) return unless assert_finished @call.status = batch_result.status + op_is_done batch_result.check_status end @@ -184,6 +187,7 @@ module GRPC end end @call.status = batch_result.status + op_is_done batch_result.check_status end @@ -415,7 +419,7 @@ module GRPC def bidi_streamer(requests, **kw, &blk) start_call(**kw) unless @started bd = BidiCall.new(@call, @cq, @marshal, @unmarshal, @deadline) - bd.run_on_client(requests, &blk) + bd.run_on_client(requests, @op_notifier, &blk) end # run_server_bidi orchestrates a BiDi stream processing on a server. @@ -434,6 +438,19 @@ module GRPC bd.run_on_server(gen_each_reply) end + # Waits till an operation completes + def wait + return if @op_notifier.nil? + GRPC.logger.debug("active_call.wait: on #{@op_notifier}") + @op_notifier.wait + end + + # Signals that an operation is done + def op_is_done + return if @op_notifier.nil? + @op_notifier.notify(self) + end + private # Starts the call if not already started @@ -468,6 +485,6 @@ module GRPC # Operation limits access to an ActiveCall's methods for use as # a Operation on the client. Operation = view_class(:cancel, :cancelled, :deadline, :execute, - :metadata, :status, :start_call) + :metadata, :status, :start_call, :wait) end end diff --git a/src/ruby/lib/grpc/generic/bidi_call.rb b/src/ruby/lib/grpc/generic/bidi_call.rb index 489dd5162a0..3b0c71395ce 100644 --- a/src/ruby/lib/grpc/generic/bidi_call.rb +++ b/src/ruby/lib/grpc/generic/bidi_call.rb @@ -66,6 +66,7 @@ module GRPC @cq = q @deadline = deadline @marshal = marshal + @op_notifier = nil # signals completion on clients @readq = Queue.new @unmarshal = unmarshal end @@ -76,8 +77,10 @@ module GRPC # block that can be invoked with each response. # # @param requests the Enumerable of requests to send + # @op_notifier a Notifier used to signal completion # @return an Enumerator of requests to yield - def run_on_client(requests, &blk) + def run_on_client(requests, op_notifier, &blk) + @op_notifier = op_notifier @enq_th = Thread.new { write_loop(requests) } @loop_th = start_read_loop each_queued_msg(&blk) @@ -105,6 +108,13 @@ module GRPC END_OF_READS = :end_of_reads END_OF_WRITES = :end_of_writes + # signals that bidi operation is complete + def notify_done + return unless @op_notifier + GRPC.logger.debug("bidi-notify-done: notifying #{@op_notifier}") + @op_notifier.notify(self) + end + # each_queued_msg yields each message on this instances readq # # - messages are added to the readq by #read_loop @@ -143,11 +153,13 @@ module GRPC @call.status = batch_result.status batch_result.check_status GRPC.logger.debug("bidi-write-loop: done status #{@call.status}") + notify_done end GRPC.logger.debug('bidi-write-loop: finished') rescue StandardError => e GRPC.logger.warn('bidi-write-loop: failed') GRPC.logger.warn(e) + notify_done raise e end diff --git a/src/ruby/lib/grpc/logconfig.rb b/src/ruby/lib/grpc/logconfig.rb index 96812170ba8..e9b4aa3c954 100644 --- a/src/ruby/lib/grpc/logconfig.rb +++ b/src/ruby/lib/grpc/logconfig.rb @@ -38,6 +38,6 @@ Logging.logger.root.appenders = Logging.appenders.stdout Logging.logger.root.level = :info # TODO: provide command-line configuration for logging -Logging.logger['GRPC'].level = :debug +Logging.logger['GRPC'].level = :info Logging.logger['GRPC::ActiveCall'].level = :info Logging.logger['GRPC::BidiCall'].level = :info