diff options
Diffstat (limited to 'src/core/lib/security')
7 files changed, 314 insertions, 285 deletions
diff --git a/src/core/lib/security/credentials/google_default/google_default_credentials.c b/src/core/lib/security/credentials/google_default/google_default_credentials.c index a2a8e289ee..691d66df69 100644 --- a/src/core/lib/security/credentials/google_default/google_default_credentials.c +++ b/src/core/lib/security/credentials/google_default/google_default_credentials.c @@ -79,7 +79,8 @@ static void on_compute_engine_detection_http_response(grpc_exec_ctx *exec_ctx, detector->is_done = 1; GRPC_LOG_IF_ERROR( "Pollset kick", - grpc_pollset_kick(grpc_polling_entity_pollset(&detector->pollent), NULL)); + grpc_pollset_kick(exec_ctx, + grpc_polling_entity_pollset(&detector->pollent), NULL)); gpr_mu_unlock(g_polling_mu); } diff --git a/src/core/lib/security/transport/client_auth_filter.c b/src/core/lib/security/transport/client_auth_filter.c index 531a88434f..dd7dd44e79 100644 --- a/src/core/lib/security/transport/client_auth_filter.c +++ b/src/core/lib/security/transport/client_auth_filter.c @@ -39,6 +39,8 @@ /* We can have a per-call credentials. */ typedef struct { + grpc_call_stack *owning_call; + grpc_call_combiner *call_combiner; grpc_call_credentials *creds; bool have_host; bool have_method; @@ -49,17 +51,12 @@ typedef struct { pollset_set so that work can progress when this call wants work to progress */ grpc_polling_entity *pollent; - gpr_atm security_context_set; - gpr_mu security_context_mu; grpc_credentials_mdelem_array md_array; grpc_linked_mdelem md_links[MAX_CREDENTIALS_METADATA_COUNT]; grpc_auth_metadata_context auth_md_context; - grpc_closure closure; - // Either 0 (no cancellation and no async operation in flight), - // a grpc_closure* (if the lowest bit is 0), - // or a grpc_error* (if the lowest bit is 1). - gpr_atm cancellation_state; - grpc_closure cancel_closure; + grpc_closure async_result_closure; + grpc_closure check_call_host_cancel_closure; + grpc_closure get_request_metadata_cancel_closure; } call_data; /* We can have a per-channel credentials. */ @@ -68,43 +65,6 @@ typedef struct { grpc_auth_context *auth_context; } channel_data; -static void decode_cancel_state(gpr_atm cancel_state, grpc_closure **func, - grpc_error **error) { - // If the lowest bit is 1, the value is a grpc_error*. - // Otherwise, if non-zdero, the value is a grpc_closure*. - if (cancel_state & 1) { - *error = (grpc_error *)(cancel_state & ~(gpr_atm)1); - } else if (cancel_state != 0) { - *func = (grpc_closure *)cancel_state; - } -} - -static gpr_atm encode_cancel_state_error(grpc_error *error) { - // Set the lowest bit to 1 to indicate that it's an error. - return (gpr_atm)1 | (gpr_atm)error; -} - -// Returns an error if the call has been cancelled. Otherwise, sets the -// cancellation function to be called upon cancellation. -static grpc_error *set_cancel_func(grpc_call_element *elem, - grpc_iomgr_cb_func func) { - call_data *calld = (call_data *)elem->call_data; - // Decode original state. - gpr_atm original_state = gpr_atm_acq_load(&calld->cancellation_state); - grpc_error *original_error = GRPC_ERROR_NONE; - grpc_closure *original_func = NULL; - decode_cancel_state(original_state, &original_func, &original_error); - // If error is set, return it. - if (original_error != GRPC_ERROR_NONE) return GRPC_ERROR_REF(original_error); - // Otherwise, store func. - GRPC_CLOSURE_INIT(&calld->cancel_closure, func, elem, - grpc_schedule_on_exec_ctx); - GPR_ASSERT(((gpr_atm)&calld->cancel_closure & (gpr_atm)1) == 0); - gpr_atm_rel_store(&calld->cancellation_state, - (gpr_atm)&calld->cancel_closure); - return GRPC_ERROR_NONE; -} - static void reset_auth_metadata_context( grpc_auth_metadata_context *auth_md_context) { if (auth_md_context->service_url != NULL) { @@ -153,7 +113,8 @@ static void on_credentials_metadata(grpc_exec_ctx *exec_ctx, void *arg, } else { error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAUTHENTICATED); - grpc_transport_stream_op_batch_finish_with_failure(exec_ctx, batch, error); + grpc_transport_stream_op_batch_finish_with_failure(exec_ctx, batch, error, + calld->call_combiner); } } @@ -191,8 +152,12 @@ static void cancel_get_request_metadata(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { grpc_call_element *elem = (grpc_call_element *)arg; call_data *calld = (call_data *)elem->call_data; - grpc_call_credentials_cancel_get_request_metadata( - exec_ctx, calld->creds, &calld->md_array, GRPC_ERROR_REF(error)); + if (error != GRPC_ERROR_NONE) { + grpc_call_credentials_cancel_get_request_metadata( + exec_ctx, calld->creds, &calld->md_array, GRPC_ERROR_REF(error)); + } + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, + "cancel_get_request_metadata"); } static void send_security_metadata(grpc_exec_ctx *exec_ctx, @@ -223,7 +188,8 @@ static void send_security_metadata(grpc_exec_ctx *exec_ctx, grpc_error_set_int( GRPC_ERROR_CREATE_FROM_STATIC_STRING( "Incompatible credentials set on channel and call."), - GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAUTHENTICATED)); + GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAUTHENTICATED), + calld->call_combiner); return; } } else { @@ -234,22 +200,25 @@ static void send_security_metadata(grpc_exec_ctx *exec_ctx, build_auth_metadata_context(&chand->security_connector->base, chand->auth_context, calld); - grpc_error *cancel_error = set_cancel_func(elem, cancel_get_request_metadata); - if (cancel_error != GRPC_ERROR_NONE) { - grpc_transport_stream_op_batch_finish_with_failure(exec_ctx, batch, - cancel_error); - return; - } GPR_ASSERT(calld->pollent != NULL); - GRPC_CLOSURE_INIT(&calld->closure, on_credentials_metadata, batch, - grpc_schedule_on_exec_ctx); + + GRPC_CLOSURE_INIT(&calld->async_result_closure, on_credentials_metadata, + batch, grpc_schedule_on_exec_ctx); grpc_error *error = GRPC_ERROR_NONE; if (grpc_call_credentials_get_request_metadata( exec_ctx, calld->creds, calld->pollent, calld->auth_md_context, - &calld->md_array, &calld->closure, &error)) { + &calld->md_array, &calld->async_result_closure, &error)) { // Synchronous return; invoke on_credentials_metadata() directly. on_credentials_metadata(exec_ctx, batch, error); GRPC_ERROR_UNREF(error); + } else { + // Async return; register cancellation closure with call combiner. + GRPC_CALL_STACK_REF(calld->owning_call, "cancel_get_request_metadata"); + grpc_call_combiner_set_notify_on_cancel( + exec_ctx, calld->call_combiner, + GRPC_CLOSURE_INIT(&calld->get_request_metadata_cancel_closure, + cancel_get_request_metadata, elem, + grpc_schedule_on_exec_ctx)); } } @@ -258,7 +227,6 @@ static void on_host_checked(grpc_exec_ctx *exec_ctx, void *arg, grpc_transport_stream_op_batch *batch = (grpc_transport_stream_op_batch *)arg; grpc_call_element *elem = batch->handler_private.extra_arg; call_data *calld = elem->call_data; - if (error == GRPC_ERROR_NONE) { send_security_metadata(exec_ctx, elem, batch); } else { @@ -271,7 +239,8 @@ static void on_host_checked(grpc_exec_ctx *exec_ctx, void *arg, exec_ctx, batch, grpc_error_set_int(GRPC_ERROR_CREATE_FROM_COPIED_STRING(error_msg), GRPC_ERROR_INT_GRPC_STATUS, - GRPC_STATUS_UNAUTHENTICATED)); + GRPC_STATUS_UNAUTHENTICATED), + calld->call_combiner); gpr_free(error_msg); } } @@ -281,9 +250,12 @@ static void cancel_check_call_host(grpc_exec_ctx *exec_ctx, void *arg, grpc_call_element *elem = (grpc_call_element *)arg; call_data *calld = (call_data *)elem->call_data; channel_data *chand = (channel_data *)elem->channel_data; - grpc_channel_security_connector_cancel_check_call_host( - exec_ctx, chand->security_connector, &calld->closure, - GRPC_ERROR_REF(error)); + if (error != GRPC_ERROR_NONE) { + grpc_channel_security_connector_cancel_check_call_host( + exec_ctx, chand->security_connector, &calld->async_result_closure, + GRPC_ERROR_REF(error)); + } + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "cancel_check_call_host"); } static void auth_start_transport_stream_op_batch( @@ -295,52 +267,19 @@ static void auth_start_transport_stream_op_batch( call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; - if (batch->cancel_stream) { - while (true) { - // Decode the original cancellation state. - gpr_atm original_state = gpr_atm_acq_load(&calld->cancellation_state); - grpc_error *cancel_error = GRPC_ERROR_NONE; - grpc_closure *func = NULL; - decode_cancel_state(original_state, &func, &cancel_error); - // If we had already set a cancellation error, there's nothing - // more to do. - if (cancel_error != GRPC_ERROR_NONE) break; - // If there's a cancel func, call it. - // Note that even if the cancel func has been changed by some - // other thread between when we decoded it and now, it will just - // be a no-op. - cancel_error = GRPC_ERROR_REF(batch->payload->cancel_stream.cancel_error); - if (func != NULL) { - GRPC_CLOSURE_SCHED(exec_ctx, func, GRPC_ERROR_REF(cancel_error)); - } - // Encode the new error into cancellation state. - if (gpr_atm_full_cas(&calld->cancellation_state, original_state, - encode_cancel_state_error(cancel_error))) { - break; // Success. - } - // The cas failed, so try again. - } - } else { - /* double checked lock over security context to ensure it's set once */ - if (gpr_atm_acq_load(&calld->security_context_set) == 0) { - gpr_mu_lock(&calld->security_context_mu); - if (gpr_atm_acq_load(&calld->security_context_set) == 0) { - GPR_ASSERT(batch->payload->context != NULL); - if (batch->payload->context[GRPC_CONTEXT_SECURITY].value == NULL) { - batch->payload->context[GRPC_CONTEXT_SECURITY].value = - grpc_client_security_context_create(); - batch->payload->context[GRPC_CONTEXT_SECURITY].destroy = - grpc_client_security_context_destroy; - } - grpc_client_security_context *sec_ctx = - batch->payload->context[GRPC_CONTEXT_SECURITY].value; - GRPC_AUTH_CONTEXT_UNREF(sec_ctx->auth_context, "client auth filter"); - sec_ctx->auth_context = - GRPC_AUTH_CONTEXT_REF(chand->auth_context, "client_auth_filter"); - gpr_atm_rel_store(&calld->security_context_set, 1); - } - gpr_mu_unlock(&calld->security_context_mu); + if (!batch->cancel_stream) { + GPR_ASSERT(batch->payload->context != NULL); + if (batch->payload->context[GRPC_CONTEXT_SECURITY].value == NULL) { + batch->payload->context[GRPC_CONTEXT_SECURITY].value = + grpc_client_security_context_create(); + batch->payload->context[GRPC_CONTEXT_SECURITY].destroy = + grpc_client_security_context_destroy; } + grpc_client_security_context *sec_ctx = + batch->payload->context[GRPC_CONTEXT_SECURITY].value; + GRPC_AUTH_CONTEXT_UNREF(sec_ctx->auth_context, "client auth filter"); + sec_ctx->auth_context = + GRPC_AUTH_CONTEXT_REF(chand->auth_context, "client_auth_filter"); } if (batch->send_initial_metadata) { @@ -365,26 +304,27 @@ static void auth_start_transport_stream_op_batch( } } if (calld->have_host) { - grpc_error *cancel_error = set_cancel_func(elem, cancel_check_call_host); - if (cancel_error != GRPC_ERROR_NONE) { - grpc_transport_stream_op_batch_finish_with_failure(exec_ctx, batch, - cancel_error); + batch->handler_private.extra_arg = elem; + GRPC_CLOSURE_INIT(&calld->async_result_closure, on_host_checked, batch, + grpc_schedule_on_exec_ctx); + char *call_host = grpc_slice_to_c_string(calld->host); + grpc_error *error = GRPC_ERROR_NONE; + if (grpc_channel_security_connector_check_call_host( + exec_ctx, chand->security_connector, call_host, + chand->auth_context, &calld->async_result_closure, &error)) { + // Synchronous return; invoke on_host_checked() directly. + on_host_checked(exec_ctx, batch, error); + GRPC_ERROR_UNREF(error); } else { - char *call_host = grpc_slice_to_c_string(calld->host); - batch->handler_private.extra_arg = elem; - grpc_error *error = GRPC_ERROR_NONE; - if (grpc_channel_security_connector_check_call_host( - exec_ctx, chand->security_connector, call_host, - chand->auth_context, - GRPC_CLOSURE_INIT(&calld->closure, on_host_checked, batch, - grpc_schedule_on_exec_ctx), - &error)) { - // Synchronous return; invoke on_host_checked() directly. - on_host_checked(exec_ctx, batch, error); - GRPC_ERROR_UNREF(error); - } - gpr_free(call_host); + // Async return; register cancellation closure with call combiner. + GRPC_CALL_STACK_REF(calld->owning_call, "cancel_check_call_host"); + grpc_call_combiner_set_notify_on_cancel( + exec_ctx, calld->call_combiner, + GRPC_CLOSURE_INIT(&calld->check_call_host_cancel_closure, + cancel_check_call_host, elem, + grpc_schedule_on_exec_ctx)); } + gpr_free(call_host); GPR_TIMER_END("auth_start_transport_stream_op_batch", 0); return; /* early exit */ } @@ -400,8 +340,8 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_element_args *args) { call_data *calld = elem->call_data; - memset(calld, 0, sizeof(*calld)); - gpr_mu_init(&calld->security_context_mu); + calld->owning_call = args->call_stack; + calld->call_combiner = args->call_combiner; return GRPC_ERROR_NONE; } @@ -426,12 +366,6 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_slice_unref_internal(exec_ctx, calld->method); } reset_auth_metadata_context(&calld->auth_md_context); - gpr_mu_destroy(&calld->security_context_mu); - gpr_atm cancel_state = gpr_atm_acq_load(&calld->cancellation_state); - grpc_error *cancel_error = GRPC_ERROR_NONE; - grpc_closure *cancel_func = NULL; - decode_cancel_state(cancel_state, &cancel_func, &cancel_error); - GRPC_ERROR_UNREF(cancel_error); } /* Constructor for channel_data */ @@ -490,6 +424,5 @@ const grpc_channel_filter grpc_client_auth_filter = { sizeof(channel_data), init_channel_elem, destroy_channel_elem, - grpc_call_next_get_peer, grpc_channel_next_get_info, "client-auth"}; diff --git a/src/core/lib/security/transport/secure_endpoint.c b/src/core/lib/security/transport/secure_endpoint.c index 5e41b94ff8..ae5633b82c 100644 --- a/src/core/lib/security/transport/secure_endpoint.c +++ b/src/core/lib/security/transport/secure_endpoint.c @@ -34,7 +34,7 @@ #include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" -#include "src/core/tsi/transport_security_interface.h" +#include "src/core/tsi/transport_security_grpc.h" #define STAGING_BUFFER_SIZE 8192 @@ -42,6 +42,7 @@ typedef struct { grpc_endpoint base; grpc_endpoint *wrapped_ep; struct tsi_frame_protector *protector; + struct tsi_zero_copy_grpc_protector *zero_copy_protector; gpr_mu protector_mu; /* saved upper level callbacks and user_data. */ grpc_closure *read_cb; @@ -67,6 +68,7 @@ static void destroy(grpc_exec_ctx *exec_ctx, secure_endpoint *secure_ep) { secure_endpoint *ep = secure_ep; grpc_endpoint_destroy(exec_ctx, ep->wrapped_ep); tsi_frame_protector_destroy(ep->protector); + tsi_zero_copy_grpc_protector_destroy(exec_ctx, ep->zero_copy_protector); grpc_slice_buffer_destroy_internal(exec_ctx, &ep->leftover_bytes); grpc_slice_unref_internal(exec_ctx, ep->read_staging_buffer); grpc_slice_unref_internal(exec_ctx, ep->write_staging_buffer); @@ -159,51 +161,58 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *user_data, return; } - /* TODO(yangg) check error, maybe bail out early */ - for (i = 0; i < ep->source_buffer.count; i++) { - grpc_slice encrypted = ep->source_buffer.slices[i]; - uint8_t *message_bytes = GRPC_SLICE_START_PTR(encrypted); - size_t message_size = GRPC_SLICE_LENGTH(encrypted); - - while (message_size > 0 || keep_looping) { - size_t unprotected_buffer_size_written = (size_t)(end - cur); - size_t processed_message_size = message_size; - gpr_mu_lock(&ep->protector_mu); - result = tsi_frame_protector_unprotect(ep->protector, message_bytes, - &processed_message_size, cur, - &unprotected_buffer_size_written); - gpr_mu_unlock(&ep->protector_mu); - if (result != TSI_OK) { - gpr_log(GPR_ERROR, "Decryption error: %s", - tsi_result_to_string(result)); - break; - } - message_bytes += processed_message_size; - message_size -= processed_message_size; - cur += unprotected_buffer_size_written; - - if (cur == end) { - flush_read_staging_buffer(ep, &cur, &end); - /* Force to enter the loop again to extract buffered bytes in protector. - The bytes could be buffered because of running out of staging_buffer. - If this happens at the end of all slices, doing another unprotect - avoids leaving data in the protector. */ - keep_looping = 1; - } else if (unprotected_buffer_size_written > 0) { - keep_looping = 1; - } else { - keep_looping = 0; + if (ep->zero_copy_protector != NULL) { + // Use zero-copy grpc protector to unprotect. + result = tsi_zero_copy_grpc_protector_unprotect( + exec_ctx, ep->zero_copy_protector, &ep->source_buffer, ep->read_buffer); + } else { + // Use frame protector to unprotect. + /* TODO(yangg) check error, maybe bail out early */ + for (i = 0; i < ep->source_buffer.count; i++) { + grpc_slice encrypted = ep->source_buffer.slices[i]; + uint8_t *message_bytes = GRPC_SLICE_START_PTR(encrypted); + size_t message_size = GRPC_SLICE_LENGTH(encrypted); + + while (message_size > 0 || keep_looping) { + size_t unprotected_buffer_size_written = (size_t)(end - cur); + size_t processed_message_size = message_size; + gpr_mu_lock(&ep->protector_mu); + result = tsi_frame_protector_unprotect( + ep->protector, message_bytes, &processed_message_size, cur, + &unprotected_buffer_size_written); + gpr_mu_unlock(&ep->protector_mu); + if (result != TSI_OK) { + gpr_log(GPR_ERROR, "Decryption error: %s", + tsi_result_to_string(result)); + break; + } + message_bytes += processed_message_size; + message_size -= processed_message_size; + cur += unprotected_buffer_size_written; + + if (cur == end) { + flush_read_staging_buffer(ep, &cur, &end); + /* Force to enter the loop again to extract buffered bytes in + protector. The bytes could be buffered because of running out of + staging_buffer. If this happens at the end of all slices, doing + another unprotect avoids leaving data in the protector. */ + keep_looping = 1; + } else if (unprotected_buffer_size_written > 0) { + keep_looping = 1; + } else { + keep_looping = 0; + } } + if (result != TSI_OK) break; } - if (result != TSI_OK) break; - } - if (cur != GRPC_SLICE_START_PTR(ep->read_staging_buffer)) { - grpc_slice_buffer_add( - ep->read_buffer, - grpc_slice_split_head( - &ep->read_staging_buffer, - (size_t)(cur - GRPC_SLICE_START_PTR(ep->read_staging_buffer)))); + if (cur != GRPC_SLICE_START_PTR(ep->read_staging_buffer)) { + grpc_slice_buffer_add( + ep->read_buffer, + grpc_slice_split_head( + &ep->read_staging_buffer, + (size_t)(cur - GRPC_SLICE_START_PTR(ep->read_staging_buffer)))); + } } /* TODO(yangg) experiment with moving this block after read_cb to see if it @@ -270,54 +279,62 @@ static void endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep, } } - for (i = 0; i < slices->count; i++) { - grpc_slice plain = slices->slices[i]; - uint8_t *message_bytes = GRPC_SLICE_START_PTR(plain); - size_t message_size = GRPC_SLICE_LENGTH(plain); - while (message_size > 0) { - size_t protected_buffer_size_to_send = (size_t)(end - cur); - size_t processed_message_size = message_size; - gpr_mu_lock(&ep->protector_mu); - result = tsi_frame_protector_protect(ep->protector, message_bytes, - &processed_message_size, cur, - &protected_buffer_size_to_send); - gpr_mu_unlock(&ep->protector_mu); - if (result != TSI_OK) { - gpr_log(GPR_ERROR, "Encryption error: %s", - tsi_result_to_string(result)); - break; - } - message_bytes += processed_message_size; - message_size -= processed_message_size; - cur += protected_buffer_size_to_send; - - if (cur == end) { - flush_write_staging_buffer(ep, &cur, &end); + if (ep->zero_copy_protector != NULL) { + // Use zero-copy grpc protector to protect. + result = tsi_zero_copy_grpc_protector_protect( + exec_ctx, ep->zero_copy_protector, slices, &ep->output_buffer); + } else { + // Use frame protector to protect. + for (i = 0; i < slices->count; i++) { + grpc_slice plain = slices->slices[i]; + uint8_t *message_bytes = GRPC_SLICE_START_PTR(plain); + size_t message_size = GRPC_SLICE_LENGTH(plain); + while (message_size > 0) { + size_t protected_buffer_size_to_send = (size_t)(end - cur); + size_t processed_message_size = message_size; + gpr_mu_lock(&ep->protector_mu); + result = tsi_frame_protector_protect(ep->protector, message_bytes, + &processed_message_size, cur, + &protected_buffer_size_to_send); + gpr_mu_unlock(&ep->protector_mu); + if (result != TSI_OK) { + gpr_log(GPR_ERROR, "Encryption error: %s", + tsi_result_to_string(result)); + break; + } + message_bytes += processed_message_size; + message_size -= processed_message_size; + cur += protected_buffer_size_to_send; + + if (cur == end) { + flush_write_staging_buffer(ep, &cur, &end); + } } - } - if (result != TSI_OK) break; - } - if (result == TSI_OK) { - size_t still_pending_size; - do { - size_t protected_buffer_size_to_send = (size_t)(end - cur); - gpr_mu_lock(&ep->protector_mu); - result = tsi_frame_protector_protect_flush(ep->protector, cur, - &protected_buffer_size_to_send, - &still_pending_size); - gpr_mu_unlock(&ep->protector_mu); if (result != TSI_OK) break; - cur += protected_buffer_size_to_send; - if (cur == end) { - flush_write_staging_buffer(ep, &cur, &end); + } + if (result == TSI_OK) { + size_t still_pending_size; + do { + size_t protected_buffer_size_to_send = (size_t)(end - cur); + gpr_mu_lock(&ep->protector_mu); + result = tsi_frame_protector_protect_flush( + ep->protector, cur, &protected_buffer_size_to_send, + &still_pending_size); + gpr_mu_unlock(&ep->protector_mu); + if (result != TSI_OK) break; + cur += protected_buffer_size_to_send; + if (cur == end) { + flush_write_staging_buffer(ep, &cur, &end); + } + } while (still_pending_size > 0); + if (cur != GRPC_SLICE_START_PTR(ep->write_staging_buffer)) { + grpc_slice_buffer_add( + &ep->output_buffer, + grpc_slice_split_head( + &ep->write_staging_buffer, + (size_t)(cur - + GRPC_SLICE_START_PTR(ep->write_staging_buffer)))); } - } while (still_pending_size > 0); - if (cur != GRPC_SLICE_START_PTR(ep->write_staging_buffer)) { - grpc_slice_buffer_add( - &ep->output_buffer, - grpc_slice_split_head( - &ep->write_staging_buffer, - (size_t)(cur - GRPC_SLICE_START_PTR(ep->write_staging_buffer)))); } } @@ -389,13 +406,16 @@ static const grpc_endpoint_vtable vtable = {endpoint_read, endpoint_get_fd}; grpc_endpoint *grpc_secure_endpoint_create( - struct tsi_frame_protector *protector, grpc_endpoint *transport, - grpc_slice *leftover_slices, size_t leftover_nslices) { + struct tsi_frame_protector *protector, + struct tsi_zero_copy_grpc_protector *zero_copy_protector, + grpc_endpoint *transport, grpc_slice *leftover_slices, + size_t leftover_nslices) { size_t i; secure_endpoint *ep = (secure_endpoint *)gpr_malloc(sizeof(secure_endpoint)); ep->base.vtable = &vtable; ep->wrapped_ep = transport; ep->protector = protector; + ep->zero_copy_protector = zero_copy_protector; grpc_slice_buffer_init(&ep->leftover_bytes); for (i = 0; i < leftover_nslices; i++) { grpc_slice_buffer_add(&ep->leftover_bytes, diff --git a/src/core/lib/security/transport/secure_endpoint.h b/src/core/lib/security/transport/secure_endpoint.h index 1c5555f3df..3323a6ff42 100644 --- a/src/core/lib/security/transport/secure_endpoint.h +++ b/src/core/lib/security/transport/secure_endpoint.h @@ -23,12 +23,17 @@ #include "src/core/lib/iomgr/endpoint.h" struct tsi_frame_protector; +struct tsi_zero_copy_grpc_protector; extern grpc_tracer_flag grpc_trace_secure_endpoint; -/* Takes ownership of protector and to_wrap, and refs leftover_slices. */ +/* Takes ownership of protector, zero_copy_protector, and to_wrap, and refs + * leftover_slices. If zero_copy_protector is not NULL, protector will never be + * used. */ grpc_endpoint *grpc_secure_endpoint_create( - struct tsi_frame_protector *protector, grpc_endpoint *to_wrap, - grpc_slice *leftover_slices, size_t leftover_nslices); + struct tsi_frame_protector *protector, + struct tsi_zero_copy_grpc_protector *zero_copy_protector, + grpc_endpoint *to_wrap, grpc_slice *leftover_slices, + size_t leftover_nslices); #endif /* GRPC_CORE_LIB_SECURITY_TRANSPORT_SECURE_ENDPOINT_H */ diff --git a/src/core/lib/security/transport/security_connector.c b/src/core/lib/security/transport/security_connector.c index a7568b995f..2a9e939d40 100644 --- a/src/core/lib/security/transport/security_connector.c +++ b/src/core/lib/security/transport/security_connector.c @@ -455,14 +455,14 @@ grpc_server_security_connector *grpc_fake_server_security_connector_create( typedef struct { grpc_channel_security_connector base; - tsi_ssl_client_handshaker_factory *handshaker_factory; + tsi_ssl_client_handshaker_factory *client_handshaker_factory; char *target_name; char *overridden_target_name; } grpc_ssl_channel_security_connector; typedef struct { grpc_server_security_connector base; - tsi_ssl_server_handshaker_factory *handshaker_factory; + tsi_ssl_server_handshaker_factory *server_handshaker_factory; } grpc_ssl_server_security_connector; static void ssl_channel_destroy(grpc_exec_ctx *exec_ctx, @@ -470,9 +470,8 @@ static void ssl_channel_destroy(grpc_exec_ctx *exec_ctx, grpc_ssl_channel_security_connector *c = (grpc_ssl_channel_security_connector *)sc; grpc_call_credentials_unref(exec_ctx, c->base.request_metadata_creds); - if (c->handshaker_factory != NULL) { - tsi_ssl_client_handshaker_factory_destroy(c->handshaker_factory); - } + tsi_ssl_client_handshaker_factory_unref(c->client_handshaker_factory); + c->client_handshaker_factory = NULL; if (c->target_name != NULL) gpr_free(c->target_name); if (c->overridden_target_name != NULL) gpr_free(c->overridden_target_name); gpr_free(sc); @@ -482,9 +481,8 @@ static void ssl_server_destroy(grpc_exec_ctx *exec_ctx, grpc_security_connector *sc) { grpc_ssl_server_security_connector *c = (grpc_ssl_server_security_connector *)sc; - if (c->handshaker_factory != NULL) { - tsi_ssl_server_handshaker_factory_destroy(c->handshaker_factory); - } + tsi_ssl_server_handshaker_factory_unref(c->server_handshaker_factory); + c->server_handshaker_factory = NULL; gpr_free(sc); } @@ -496,7 +494,7 @@ static void ssl_channel_add_handshakers(grpc_exec_ctx *exec_ctx, // Instantiate TSI handshaker. tsi_handshaker *tsi_hs = NULL; tsi_result result = tsi_ssl_client_handshaker_factory_create_handshaker( - c->handshaker_factory, + c->client_handshaker_factory, c->overridden_target_name != NULL ? c->overridden_target_name : c->target_name, &tsi_hs); @@ -521,7 +519,7 @@ static void ssl_server_add_handshakers(grpc_exec_ctx *exec_ctx, // Instantiate TSI handshaker. tsi_handshaker *tsi_hs = NULL; tsi_result result = tsi_ssl_server_handshaker_factory_create_handshaker( - c->handshaker_factory, &tsi_hs); + c->server_handshaker_factory, &tsi_hs); if (result != TSI_OK) { gpr_log(GPR_ERROR, "Handshaker creation failed with error %s.", tsi_result_to_string(result)); @@ -852,7 +850,7 @@ grpc_security_status grpc_ssl_channel_security_connector_create( result = tsi_create_ssl_client_handshaker_factory( has_key_cert_pair ? &config->pem_key_cert_pair : NULL, pem_root_certs, ssl_cipher_suites(), alpn_protocol_strings, (uint16_t)num_alpn_protocols, - &c->handshaker_factory); + &c->client_handshaker_factory); if (result != TSI_OK) { gpr_log(GPR_ERROR, "Handshaker factory creation failed with %s.", tsi_result_to_string(result)); @@ -897,7 +895,7 @@ grpc_security_status grpc_ssl_server_security_connector_create( config->pem_root_certs, get_tsi_client_certificate_request_type( config->client_certificate_request), ssl_cipher_suites(), alpn_protocol_strings, (uint16_t)num_alpn_protocols, - &c->handshaker_factory); + &c->server_handshaker_factory); if (result != TSI_OK) { gpr_log(GPR_ERROR, "Handshaker factory creation failed with %s.", tsi_result_to_string(result)); diff --git a/src/core/lib/security/transport/security_handshaker.c b/src/core/lib/security/transport/security_handshaker.c index fc9c9f980f..3d19605617 100644 --- a/src/core/lib/security/transport/security_handshaker.c +++ b/src/core/lib/security/transport/security_handshaker.c @@ -32,6 +32,7 @@ #include "src/core/lib/security/transport/secure_endpoint.h" #include "src/core/lib/security/transport/tsi_error.h" #include "src/core/lib/slice/slice_internal.h" +#include "src/core/tsi/transport_security_grpc.h" #define GRPC_INITIAL_HANDSHAKE_BUFFER_SIZE 256 @@ -127,24 +128,36 @@ static void security_handshake_failed_locked(grpc_exec_ctx *exec_ctx, GRPC_CLOSURE_SCHED(exec_ctx, h->on_handshake_done, error); } -static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg, - grpc_error *error) { - security_handshaker *h = arg; - gpr_mu_lock(&h->mu); +static void on_peer_checked_inner(grpc_exec_ctx *exec_ctx, + security_handshaker *h, grpc_error *error) { if (error != GRPC_ERROR_NONE || h->shutdown) { security_handshake_failed_locked(exec_ctx, h, GRPC_ERROR_REF(error)); - goto done; + return; } - // Create frame protector. - tsi_frame_protector *protector; - tsi_result result = tsi_handshaker_result_create_frame_protector( - h->handshaker_result, NULL, &protector); - if (result != TSI_OK) { + // Create zero-copy frame protector, if implemented. + tsi_zero_copy_grpc_protector *zero_copy_protector = NULL; + tsi_result result = tsi_handshaker_result_create_zero_copy_grpc_protector( + exec_ctx, h->handshaker_result, NULL, &zero_copy_protector); + if (result != TSI_OK && result != TSI_UNIMPLEMENTED) { error = grpc_set_tsi_error_result( - GRPC_ERROR_CREATE_FROM_STATIC_STRING("Frame protector creation failed"), + GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Zero-copy frame protector creation failed"), result); security_handshake_failed_locked(exec_ctx, h, error); - goto done; + return; + } + // Create frame protector if zero-copy frame protector is NULL. + tsi_frame_protector *protector = NULL; + if (zero_copy_protector == NULL) { + result = tsi_handshaker_result_create_frame_protector(h->handshaker_result, + NULL, &protector); + if (result != TSI_OK) { + error = grpc_set_tsi_error_result(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Frame protector creation failed"), + result); + security_handshake_failed_locked(exec_ctx, h, error); + return; + } } // Get unused bytes. const unsigned char *unused_bytes = NULL; @@ -155,12 +168,12 @@ static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg, if (unused_bytes_size > 0) { grpc_slice slice = grpc_slice_from_copied_buffer((char *)unused_bytes, unused_bytes_size); - h->args->endpoint = - grpc_secure_endpoint_create(protector, h->args->endpoint, &slice, 1); + h->args->endpoint = grpc_secure_endpoint_create( + protector, zero_copy_protector, h->args->endpoint, &slice, 1); grpc_slice_unref_internal(exec_ctx, slice); } else { - h->args->endpoint = - grpc_secure_endpoint_create(protector, h->args->endpoint, NULL, 0); + h->args->endpoint = grpc_secure_endpoint_create( + protector, zero_copy_protector, h->args->endpoint, NULL, 0); } tsi_handshaker_result_destroy(h->handshaker_result); h->handshaker_result = NULL; @@ -177,7 +190,13 @@ static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg, // Set shutdown to true so that subsequent calls to // security_handshaker_shutdown() do nothing. h->shutdown = true; -done: +} + +static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + security_handshaker *h = (security_handshaker *)arg; + gpr_mu_lock(&h->mu); + on_peer_checked_inner(exec_ctx, h, error); gpr_mu_unlock(&h->mu); security_handshaker_unref(exec_ctx, h); } @@ -239,7 +258,7 @@ static grpc_error *on_handshake_next_done_locked( static void on_handshake_next_done_grpc_wrapper( tsi_result result, void *user_data, const unsigned char *bytes_to_send, size_t bytes_to_send_size, tsi_handshaker_result *handshaker_result) { - security_handshaker *h = user_data; + security_handshaker *h = (security_handshaker *)user_data; // This callback will be invoked by TSI in a non-grpc thread, so it's // safe to create our own exec_ctx here. grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; @@ -281,7 +300,7 @@ static grpc_error *do_handshaker_next_locked( static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - security_handshaker *h = arg; + security_handshaker *h = (security_handshaker *)arg; gpr_mu_lock(&h->mu); if (error != GRPC_ERROR_NONE || h->shutdown) { security_handshake_failed_locked( @@ -298,7 +317,8 @@ static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx, bytes_received_size += GRPC_SLICE_LENGTH(h->args->read_buffer->slices[i]); } if (bytes_received_size > h->handshake_buffer_size) { - h->handshake_buffer = gpr_realloc(h->handshake_buffer, bytes_received_size); + h->handshake_buffer = + (uint8_t *)gpr_realloc(h->handshake_buffer, bytes_received_size); h->handshake_buffer_size = bytes_received_size; } size_t offset = 0; @@ -323,7 +343,7 @@ static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx, static void on_handshake_data_sent_to_peer(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - security_handshaker *h = arg; + security_handshaker *h = (security_handshaker *)arg; gpr_mu_lock(&h->mu); if (error != GRPC_ERROR_NONE || h->shutdown) { security_handshake_failed_locked( @@ -400,14 +420,15 @@ static const grpc_handshaker_vtable security_handshaker_vtable = { static grpc_handshaker *security_handshaker_create( grpc_exec_ctx *exec_ctx, tsi_handshaker *handshaker, grpc_security_connector *connector) { - security_handshaker *h = gpr_zalloc(sizeof(security_handshaker)); + security_handshaker *h = + (security_handshaker *)gpr_zalloc(sizeof(security_handshaker)); grpc_handshaker_init(&security_handshaker_vtable, &h->base); h->handshaker = handshaker; h->connector = GRPC_SECURITY_CONNECTOR_REF(connector, "handshake"); gpr_mu_init(&h->mu); gpr_ref_init(&h->refs, 1); h->handshake_buffer_size = GRPC_INITIAL_HANDSHAKE_BUFFER_SIZE; - h->handshake_buffer = gpr_malloc(h->handshake_buffer_size); + h->handshake_buffer = (uint8_t *)gpr_malloc(h->handshake_buffer_size); GRPC_CLOSURE_INIT(&h->on_handshake_data_sent_to_peer, on_handshake_data_sent_to_peer, h, grpc_schedule_on_exec_ctx); @@ -450,7 +471,7 @@ static const grpc_handshaker_vtable fail_handshaker_vtable = { fail_handshaker_do_handshake}; static grpc_handshaker *fail_handshaker_create() { - grpc_handshaker *h = gpr_malloc(sizeof(*h)); + grpc_handshaker *h = (grpc_handshaker *)gpr_malloc(sizeof(*h)); grpc_handshaker_init(&fail_handshaker_vtable, h); return h; } diff --git a/src/core/lib/security/transport/server_auth_filter.c b/src/core/lib/security/transport/server_auth_filter.c index 9bf3f0ca0f..7f523c0883 100644 --- a/src/core/lib/security/transport/server_auth_filter.c +++ b/src/core/lib/security/transport/server_auth_filter.c @@ -26,7 +26,15 @@ #include "src/core/lib/security/transport/auth_filters.h" #include "src/core/lib/slice/slice_internal.h" +typedef enum { + STATE_INIT = 0, + STATE_DONE, + STATE_CANCELLED, +} async_state; + typedef struct call_data { + grpc_call_combiner *call_combiner; + grpc_call_stack *owning_call; grpc_transport_stream_op_batch *recv_initial_metadata_batch; grpc_closure *original_recv_initial_metadata_ready; grpc_closure recv_initial_metadata_ready; @@ -34,6 +42,8 @@ typedef struct call_data { const grpc_metadata *consumed_md; size_t num_consumed_md; grpc_auth_context *auth_context; + grpc_closure cancel_closure; + gpr_atm state; // async_state } call_data; typedef struct channel_data { @@ -78,54 +88,94 @@ static grpc_filtered_mdelem remove_consumed_md(grpc_exec_ctx *exec_ctx, return GRPC_FILTERED_MDELEM(md); } -/* called from application code */ -static void on_md_processing_done( - void *user_data, const grpc_metadata *consumed_md, size_t num_consumed_md, - const grpc_metadata *response_md, size_t num_response_md, - grpc_status_code status, const char *error_details) { - grpc_call_element *elem = user_data; +static void on_md_processing_done_inner(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + const grpc_metadata *consumed_md, + size_t num_consumed_md, + const grpc_metadata *response_md, + size_t num_response_md, + grpc_error *error) { call_data *calld = elem->call_data; grpc_transport_stream_op_batch *batch = calld->recv_initial_metadata_batch; - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; /* TODO(jboeuf): Implement support for response_md. */ if (response_md != NULL && num_response_md > 0) { gpr_log(GPR_INFO, "response_md in auth metadata processing not supported for now. " "Ignoring..."); } - grpc_error *error = GRPC_ERROR_NONE; - if (status == GRPC_STATUS_OK) { + if (error == GRPC_ERROR_NONE) { calld->consumed_md = consumed_md; calld->num_consumed_md = num_consumed_md; error = grpc_metadata_batch_filter( - &exec_ctx, batch->payload->recv_initial_metadata.recv_initial_metadata, + exec_ctx, batch->payload->recv_initial_metadata.recv_initial_metadata, remove_consumed_md, elem, "Response metadata filtering error"); - } else { - if (error_details == NULL) { - error_details = "Authentication metadata processing failed."; + } + GRPC_CLOSURE_SCHED(exec_ctx, calld->original_recv_initial_metadata_ready, + error); +} + +// Called from application code. +static void on_md_processing_done( + void *user_data, const grpc_metadata *consumed_md, size_t num_consumed_md, + const grpc_metadata *response_md, size_t num_response_md, + grpc_status_code status, const char *error_details) { + grpc_call_element *elem = user_data; + call_data *calld = elem->call_data; + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + // If the call was not cancelled while we were in flight, process the result. + if (gpr_atm_full_cas(&calld->state, (gpr_atm)STATE_INIT, + (gpr_atm)STATE_DONE)) { + grpc_error *error = GRPC_ERROR_NONE; + if (status != GRPC_STATUS_OK) { + if (error_details == NULL) { + error_details = "Authentication metadata processing failed."; + } + error = grpc_error_set_int( + GRPC_ERROR_CREATE_FROM_COPIED_STRING(error_details), + GRPC_ERROR_INT_GRPC_STATUS, status); } - error = - grpc_error_set_int(GRPC_ERROR_CREATE_FROM_COPIED_STRING(error_details), - GRPC_ERROR_INT_GRPC_STATUS, status); + on_md_processing_done_inner(&exec_ctx, elem, consumed_md, num_consumed_md, + response_md, num_response_md, error); } + // Clean up. for (size_t i = 0; i < calld->md.count; i++) { grpc_slice_unref_internal(&exec_ctx, calld->md.metadata[i].key); grpc_slice_unref_internal(&exec_ctx, calld->md.metadata[i].value); } grpc_metadata_array_destroy(&calld->md); - GRPC_CLOSURE_SCHED(&exec_ctx, calld->original_recv_initial_metadata_ready, - error); + GRPC_CALL_STACK_UNREF(&exec_ctx, calld->owning_call, "server_auth_metadata"); grpc_exec_ctx_finish(&exec_ctx); } +static void cancel_call(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { + grpc_call_element *elem = (grpc_call_element *)arg; + call_data *calld = elem->call_data; + // If the result was not already processed, invoke the callback now. + if (error != GRPC_ERROR_NONE && + gpr_atm_full_cas(&calld->state, (gpr_atm)STATE_INIT, + (gpr_atm)STATE_CANCELLED)) { + on_md_processing_done_inner(exec_ctx, elem, NULL, 0, NULL, 0, + GRPC_ERROR_REF(error)); + } + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "cancel_call"); +} + static void recv_initial_metadata_ready(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_call_element *elem = arg; + grpc_call_element *elem = (grpc_call_element *)arg; channel_data *chand = elem->channel_data; call_data *calld = elem->call_data; grpc_transport_stream_op_batch *batch = calld->recv_initial_metadata_batch; if (error == GRPC_ERROR_NONE) { if (chand->creds != NULL && chand->creds->processor.process != NULL) { + // We're calling out to the application, so we need to make sure + // to drop the call combiner early if we get cancelled. + GRPC_CALL_STACK_REF(calld->owning_call, "cancel_call"); + GRPC_CLOSURE_INIT(&calld->cancel_closure, cancel_call, elem, + grpc_schedule_on_exec_ctx); + grpc_call_combiner_set_notify_on_cancel(exec_ctx, calld->call_combiner, + &calld->cancel_closure); + GRPC_CALL_STACK_REF(calld->owning_call, "server_auth_metadata"); calld->md = metadata_batch_to_md_array( batch->payload->recv_initial_metadata.recv_initial_metadata); chand->creds->processor.process( @@ -159,6 +209,8 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, const grpc_call_element_args *args) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; + calld->call_combiner = args->call_combiner; + calld->owning_call = args->call_stack; GRPC_CLOSURE_INIT(&calld->recv_initial_metadata_ready, recv_initial_metadata_ready, elem, grpc_schedule_on_exec_ctx); @@ -218,6 +270,5 @@ const grpc_channel_filter grpc_server_auth_filter = { sizeof(channel_data), init_channel_elem, destroy_channel_elem, - grpc_call_next_get_peer, grpc_channel_next_get_info, "server-auth"}; |