|
|
|
|
@ -24,6 +24,7 @@ |
|
|
|
|
#include <grpc/support/alloc.h> |
|
|
|
|
#include <grpc/support/log.h> |
|
|
|
|
|
|
|
|
|
#include "src/core/lib/gprpp/sync.h" |
|
|
|
|
#include "src/core/lib/slice/slice_internal.h" |
|
|
|
|
#include "src/core/lib/surface/call.h" |
|
|
|
|
#include "src/core/lib/surface/channel.h" |
|
|
|
|
@ -39,8 +40,18 @@ struct alts_handshaker_client { |
|
|
|
|
const alts_handshaker_client_vtable* vtable; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
struct recv_message_result { |
|
|
|
|
tsi_result status; |
|
|
|
|
const unsigned char* bytes_to_send; |
|
|
|
|
size_t bytes_to_send_size; |
|
|
|
|
tsi_handshaker_result* result; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
typedef struct alts_grpc_handshaker_client { |
|
|
|
|
alts_handshaker_client base; |
|
|
|
|
/* One ref is held by the entity that created this handshaker_client, and
|
|
|
|
|
* another ref is held by the pending RECEIVE_STATUS_ON_CLIENT op. */ |
|
|
|
|
gpr_refcount refs; |
|
|
|
|
alts_tsi_handshaker* handshaker; |
|
|
|
|
grpc_call* call; |
|
|
|
|
/* A pointer to a function handling the interaction with handshaker service.
|
|
|
|
|
@ -77,6 +88,18 @@ typedef struct alts_grpc_handshaker_client { |
|
|
|
|
/* a buffer containing data to be sent to the grpc client or server's peer. */ |
|
|
|
|
unsigned char* buffer; |
|
|
|
|
size_t buffer_size; |
|
|
|
|
/** callback for receiving handshake call status */ |
|
|
|
|
grpc_closure on_status_received; |
|
|
|
|
/** gRPC status code of handshake call */ |
|
|
|
|
grpc_status_code handshake_status_code; |
|
|
|
|
/** gRPC status details of handshake call */ |
|
|
|
|
grpc_slice handshake_status_details; |
|
|
|
|
/* mu synchronizes all fields below including their internal fields. */ |
|
|
|
|
gpr_mu mu; |
|
|
|
|
/* indicates if the handshaker call's RECV_STATUS_ON_CLIENT op is done. */ |
|
|
|
|
bool receive_status_finished; |
|
|
|
|
/* if non-null, contains arguments to complete a TSI next callback. */ |
|
|
|
|
recv_message_result* pending_recv_message_result; |
|
|
|
|
} alts_grpc_handshaker_client; |
|
|
|
|
|
|
|
|
|
static void handshaker_client_send_buffer_destroy( |
|
|
|
|
@ -94,6 +117,95 @@ static bool is_handshake_finished_properly(grpc_gcp_HandshakerResp* resp) { |
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void alts_grpc_handshaker_client_unref( |
|
|
|
|
alts_grpc_handshaker_client* client) { |
|
|
|
|
if (gpr_unref(&client->refs)) { |
|
|
|
|
if (client->base.vtable != nullptr && |
|
|
|
|
client->base.vtable->destruct != nullptr) { |
|
|
|
|
client->base.vtable->destruct(&client->base); |
|
|
|
|
} |
|
|
|
|
grpc_byte_buffer_destroy(client->send_buffer); |
|
|
|
|
grpc_byte_buffer_destroy(client->recv_buffer); |
|
|
|
|
client->send_buffer = nullptr; |
|
|
|
|
client->recv_buffer = nullptr; |
|
|
|
|
grpc_metadata_array_destroy(&client->recv_initial_metadata); |
|
|
|
|
grpc_slice_unref_internal(client->recv_bytes); |
|
|
|
|
grpc_slice_unref_internal(client->target_name); |
|
|
|
|
grpc_alts_credentials_options_destroy(client->options); |
|
|
|
|
gpr_free(client->buffer); |
|
|
|
|
grpc_slice_unref_internal(client->handshake_status_details); |
|
|
|
|
gpr_mu_destroy(&client->mu); |
|
|
|
|
gpr_free(client); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void maybe_complete_tsi_next( |
|
|
|
|
alts_grpc_handshaker_client* client, bool receive_status_finished, |
|
|
|
|
recv_message_result* pending_recv_message_result) { |
|
|
|
|
recv_message_result* r; |
|
|
|
|
{ |
|
|
|
|
grpc_core::MutexLock lock(&client->mu); |
|
|
|
|
client->receive_status_finished |= receive_status_finished; |
|
|
|
|
if (pending_recv_message_result != nullptr) { |
|
|
|
|
GPR_ASSERT(client->pending_recv_message_result == nullptr); |
|
|
|
|
client->pending_recv_message_result = pending_recv_message_result; |
|
|
|
|
} |
|
|
|
|
if (client->pending_recv_message_result == nullptr) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
const bool have_final_result = |
|
|
|
|
client->pending_recv_message_result->result != nullptr || |
|
|
|
|
client->pending_recv_message_result->status != TSI_OK; |
|
|
|
|
if (have_final_result && !client->receive_status_finished) { |
|
|
|
|
// If we've received the final message from the handshake
|
|
|
|
|
// server, or we're about to invoke the TSI next callback
|
|
|
|
|
// with a status other than TSI_OK (which terminates the
|
|
|
|
|
// handshake), then first wait for the RECV_STATUS op to complete.
|
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
r = client->pending_recv_message_result; |
|
|
|
|
client->pending_recv_message_result = nullptr; |
|
|
|
|
} |
|
|
|
|
client->cb(r->status, client->user_data, r->bytes_to_send, |
|
|
|
|
r->bytes_to_send_size, r->result); |
|
|
|
|
gpr_free(r); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void on_status_received(void* arg, grpc_error* error) { |
|
|
|
|
alts_grpc_handshaker_client* client = |
|
|
|
|
static_cast<alts_grpc_handshaker_client*>(arg); |
|
|
|
|
if (client->handshake_status_code != GRPC_STATUS_OK) { |
|
|
|
|
// TODO(apolcyn): consider overriding the handshake result's
|
|
|
|
|
// status from the final ALTS message with the status here.
|
|
|
|
|
char* status_details = |
|
|
|
|
grpc_slice_to_c_string(client->handshake_status_details); |
|
|
|
|
gpr_log(GPR_INFO, |
|
|
|
|
"alts_grpc_handshaker_client:%p on_status_received " |
|
|
|
|
"status:%d details:|%s| error:|%s|", |
|
|
|
|
client, client->handshake_status_code, status_details, |
|
|
|
|
grpc_error_string(error)); |
|
|
|
|
gpr_free(status_details); |
|
|
|
|
} |
|
|
|
|
maybe_complete_tsi_next(client, true /* receive_status_finished */, |
|
|
|
|
nullptr /* pending_recv_message_result */); |
|
|
|
|
alts_grpc_handshaker_client_unref(client); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void handle_response_done(alts_grpc_handshaker_client* client, |
|
|
|
|
tsi_result status, |
|
|
|
|
const unsigned char* bytes_to_send, |
|
|
|
|
size_t bytes_to_send_size, |
|
|
|
|
tsi_handshaker_result* result) { |
|
|
|
|
recv_message_result* p = |
|
|
|
|
static_cast<recv_message_result*>(gpr_zalloc(sizeof(*p))); |
|
|
|
|
p->status = status; |
|
|
|
|
p->bytes_to_send = bytes_to_send; |
|
|
|
|
p->bytes_to_send_size = bytes_to_send_size; |
|
|
|
|
p->result = result; |
|
|
|
|
maybe_complete_tsi_next(client, false /* receive_status_finished */, |
|
|
|
|
p /* pending_recv_message_result */); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void alts_handshaker_client_handle_response(alts_handshaker_client* c, |
|
|
|
|
bool is_ok) { |
|
|
|
|
GPR_ASSERT(c != nullptr); |
|
|
|
|
@ -101,38 +213,35 @@ void alts_handshaker_client_handle_response(alts_handshaker_client* c, |
|
|
|
|
reinterpret_cast<alts_grpc_handshaker_client*>(c); |
|
|
|
|
grpc_byte_buffer* recv_buffer = client->recv_buffer; |
|
|
|
|
grpc_status_code status = client->status; |
|
|
|
|
tsi_handshaker_on_next_done_cb cb = client->cb; |
|
|
|
|
void* user_data = client->user_data; |
|
|
|
|
alts_tsi_handshaker* handshaker = client->handshaker; |
|
|
|
|
|
|
|
|
|
/* Invalid input check. */ |
|
|
|
|
if (cb == nullptr) { |
|
|
|
|
if (client->cb == nullptr) { |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"cb is nullptr in alts_tsi_handshaker_handle_response()"); |
|
|
|
|
"client->cb is nullptr in alts_tsi_handshaker_handle_response()"); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
if (handshaker == nullptr) { |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"handshaker is nullptr in alts_tsi_handshaker_handle_response()"); |
|
|
|
|
cb(TSI_INTERNAL_ERROR, user_data, nullptr, 0, nullptr); |
|
|
|
|
handle_response_done(client, TSI_INTERNAL_ERROR, nullptr, 0, nullptr); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
/* TSI handshake has been shutdown. */ |
|
|
|
|
if (alts_tsi_handshaker_has_shutdown(handshaker)) { |
|
|
|
|
gpr_log(GPR_ERROR, "TSI handshake shutdown"); |
|
|
|
|
cb(TSI_HANDSHAKE_SHUTDOWN, user_data, nullptr, 0, nullptr); |
|
|
|
|
handle_response_done(client, TSI_HANDSHAKE_SHUTDOWN, nullptr, 0, nullptr); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
/* Failed grpc call check. */ |
|
|
|
|
if (!is_ok || status != GRPC_STATUS_OK) { |
|
|
|
|
gpr_log(GPR_ERROR, "grpc call made to handshaker service failed"); |
|
|
|
|
cb(TSI_INTERNAL_ERROR, user_data, nullptr, 0, nullptr); |
|
|
|
|
handle_response_done(client, TSI_INTERNAL_ERROR, nullptr, 0, nullptr); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
if (recv_buffer == nullptr) { |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"recv_buffer is nullptr in alts_tsi_handshaker_handle_response()"); |
|
|
|
|
cb(TSI_INTERNAL_ERROR, user_data, nullptr, 0, nullptr); |
|
|
|
|
handle_response_done(client, TSI_INTERNAL_ERROR, nullptr, 0, nullptr); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
upb::Arena arena; |
|
|
|
|
@ -143,14 +252,14 @@ void alts_handshaker_client_handle_response(alts_handshaker_client* c, |
|
|
|
|
/* Invalid handshaker response check. */ |
|
|
|
|
if (resp == nullptr) { |
|
|
|
|
gpr_log(GPR_ERROR, "alts_tsi_utils_deserialize_response() failed"); |
|
|
|
|
cb(TSI_DATA_CORRUPTED, user_data, nullptr, 0, nullptr); |
|
|
|
|
handle_response_done(client, TSI_DATA_CORRUPTED, nullptr, 0, nullptr); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
const grpc_gcp_HandshakerStatus* resp_status = |
|
|
|
|
grpc_gcp_HandshakerResp_status(resp); |
|
|
|
|
if (resp_status == nullptr) { |
|
|
|
|
gpr_log(GPR_ERROR, "No status in HandshakerResp"); |
|
|
|
|
cb(TSI_DATA_CORRUPTED, user_data, nullptr, 0, nullptr); |
|
|
|
|
handle_response_done(client, TSI_DATA_CORRUPTED, nullptr, 0, nullptr); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
upb_strview out_frames = grpc_gcp_HandshakerResp_out_frames(resp); |
|
|
|
|
@ -184,8 +293,12 @@ void alts_handshaker_client_handle_response(alts_handshaker_client* c, |
|
|
|
|
gpr_free(error_details); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
cb(alts_tsi_utils_convert_to_tsi_result(code), user_data, bytes_to_send, |
|
|
|
|
bytes_to_send_size, result); |
|
|
|
|
// TODO(apolcyn): consider short ciruiting handle_response_done and
|
|
|
|
|
// invoking the TSI callback directly if we aren't done yet, if
|
|
|
|
|
// handle_response_done's allocation per message received causes
|
|
|
|
|
// a performance issue.
|
|
|
|
|
handle_response_done(client, alts_tsi_utils_convert_to_tsi_result(code), |
|
|
|
|
bytes_to_send, bytes_to_send_size, result); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
@ -200,6 +313,23 @@ static tsi_result make_grpc_call(alts_handshaker_client* c, bool is_start) { |
|
|
|
|
memset(ops, 0, sizeof(ops)); |
|
|
|
|
grpc_op* op = ops; |
|
|
|
|
if (is_start) { |
|
|
|
|
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; |
|
|
|
|
op->data.recv_status_on_client.trailing_metadata = nullptr; |
|
|
|
|
op->data.recv_status_on_client.status = &client->handshake_status_code; |
|
|
|
|
op->data.recv_status_on_client.status_details = |
|
|
|
|
&client->handshake_status_details; |
|
|
|
|
op->flags = 0; |
|
|
|
|
op->reserved = nullptr; |
|
|
|
|
op++; |
|
|
|
|
GPR_ASSERT(op - ops <= kHandshakerClientOpNum); |
|
|
|
|
gpr_ref(&client->refs); |
|
|
|
|
grpc_call_error call_error = |
|
|
|
|
client->grpc_caller(client->call, ops, static_cast<size_t>(op - ops), |
|
|
|
|
&client->on_status_received); |
|
|
|
|
// TODO(apolcyn): return the error here instead, as done for other ops?
|
|
|
|
|
GPR_ASSERT(call_error == GRPC_CALL_OK); |
|
|
|
|
memset(ops, 0, sizeof(ops)); |
|
|
|
|
op = ops; |
|
|
|
|
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
|
|
|
|
op->data.send_initial_metadata.count = 0; |
|
|
|
|
op++; |
|
|
|
|
@ -455,6 +585,8 @@ alts_handshaker_client* alts_grpc_handshaker_client_create( |
|
|
|
|
} |
|
|
|
|
alts_grpc_handshaker_client* client = |
|
|
|
|
static_cast<alts_grpc_handshaker_client*>(gpr_zalloc(sizeof(*client))); |
|
|
|
|
gpr_mu_init(&client->mu); |
|
|
|
|
gpr_ref_init(&client->refs, 1); |
|
|
|
|
client->grpc_caller = grpc_call_start_batch_and_execute; |
|
|
|
|
client->handshaker = handshaker; |
|
|
|
|
client->cb = cb; |
|
|
|
|
@ -481,6 +613,8 @@ alts_handshaker_client* alts_grpc_handshaker_client_create( |
|
|
|
|
vtable_for_testing == nullptr ? &vtable : vtable_for_testing; |
|
|
|
|
GRPC_CLOSURE_INIT(&client->on_handshaker_service_resp_recv, grpc_cb, client, |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
GRPC_CLOSURE_INIT(&client->on_status_received, on_status_received, client, |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
grpc_slice_unref_internal(slice); |
|
|
|
|
return &client->base; |
|
|
|
|
} |
|
|
|
|
@ -590,6 +724,21 @@ grpc_closure* alts_handshaker_client_get_closure_for_testing( |
|
|
|
|
return &client->on_handshaker_service_resp_recv; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void alts_handshaker_client_ref_for_testing(alts_handshaker_client* c) { |
|
|
|
|
alts_grpc_handshaker_client* client = |
|
|
|
|
reinterpret_cast<alts_grpc_handshaker_client*>(c); |
|
|
|
|
gpr_ref(&client->refs); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void alts_handshaker_client_on_status_received_for_testing( |
|
|
|
|
alts_handshaker_client* c, grpc_status_code status, grpc_error* error) { |
|
|
|
|
alts_grpc_handshaker_client* client = |
|
|
|
|
reinterpret_cast<alts_grpc_handshaker_client*>(c); |
|
|
|
|
client->handshake_status_code = status; |
|
|
|
|
client->handshake_status_details = grpc_empty_slice(); |
|
|
|
|
grpc_core::Closure::Run(DEBUG_LOCATION, &client->on_status_received, error); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
} // namespace internal
|
|
|
|
|
} // namespace grpc_core
|
|
|
|
|
|
|
|
|
|
@ -634,20 +783,8 @@ void alts_handshaker_client_shutdown(alts_handshaker_client* client) { |
|
|
|
|
|
|
|
|
|
void alts_handshaker_client_destroy(alts_handshaker_client* c) { |
|
|
|
|
if (c != nullptr) { |
|
|
|
|
if (c->vtable != nullptr && c->vtable->destruct != nullptr) { |
|
|
|
|
c->vtable->destruct(c); |
|
|
|
|
} |
|
|
|
|
alts_grpc_handshaker_client* client = |
|
|
|
|
reinterpret_cast<alts_grpc_handshaker_client*>(c); |
|
|
|
|
grpc_byte_buffer_destroy(client->send_buffer); |
|
|
|
|
grpc_byte_buffer_destroy(client->recv_buffer); |
|
|
|
|
client->send_buffer = nullptr; |
|
|
|
|
client->recv_buffer = nullptr; |
|
|
|
|
grpc_metadata_array_destroy(&client->recv_initial_metadata); |
|
|
|
|
grpc_slice_unref_internal(client->recv_bytes); |
|
|
|
|
grpc_slice_unref_internal(client->target_name); |
|
|
|
|
grpc_alts_credentials_options_destroy(client->options); |
|
|
|
|
gpr_free(client->buffer); |
|
|
|
|
gpr_free(client); |
|
|
|
|
alts_grpc_handshaker_client_unref(client); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
apolcyn
5 years ago
committed by
GitHub