test/core/bad_client/bad_client.cc

   1 /*
   2  *
   3  * Copyright 2015 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include "test/core/bad_client/bad_client.h"
  20
  21 #include <stdio.h>
  22
  23 #include <grpc/support/alloc.h>
  24 #include <grpc/support/string_util.h>
  25 #include <grpc/support/sync.h>
  26
  27 #include "src/core/ext/filters/http/server/http_server_filter.h"
  28 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
  29 #include "src/core/lib/channel/channel_stack.h"
  30 #include "src/core/lib/gpr/murmur_hash.h"
  31 #include "src/core/lib/gpr/string.h"
  32 #include "src/core/lib/gprpp/thd.h"
  33 #include "src/core/lib/iomgr/endpoint_pair.h"
  34 #include "src/core/lib/slice/slice_internal.h"
  35 #include "src/core/lib/surface/completion_queue.h"
  36 #include "src/core/lib/surface/server.h"
  37 #include "test/core/end2end/cq_verifier.h"
  38
  39 #define MIN_HTTP2_FRAME_SIZE 9
  40
  41 /* Args to provide to thread running server side validator */
  42 typedef struct {
  43   grpc_server* server;
  44   grpc_completion_queue* cq;
  45   grpc_bad_client_server_side_validator validator;
  46   void* registered_method;
  47   gpr_event done_thd;
  48 } thd_args;
  49
  50 /* Run the server side validator and set done_thd once done */
  51 static void thd_func(void* arg) {
  52   thd_args* a = static_cast<thd_args*>(arg);
  53   if (a->validator != nullptr) {
  54     a->validator(a->server, a->cq, a->registered_method);
  55   }
  56   gpr_event_set(&a->done_thd, (void*)1);
  57 }
  58
  59 /* Sets the done_write event */
  60 static void set_done_write(void* arg, grpc_error* error) {
  61   gpr_event* done_write = static_cast<gpr_event*>(arg);
  62   gpr_event_set(done_write, (void*)1);
  63 }
  64
  65 static void server_setup_transport(void* ts, grpc_transport* transport) {
  66   thd_args* a = static_cast<thd_args*>(ts);
  67   grpc_core::ExecCtx exec_ctx;
  68   grpc_server_setup_transport(a->server, transport, nullptr,
  69                               grpc_server_get_channel_args(a->server), nullptr);
  70 }
  71
  72 /* Sets the read_done event */
  73 static void set_read_done(void* arg, grpc_error* error) {
  74   gpr_event* read_done = static_cast<gpr_event*>(arg);
  75   gpr_event_set(read_done, (void*)1);
  76 }
  77
  78 /* shutdown client */
  79 static void shutdown_client(grpc_endpoint** client_fd) {
  80   if (*client_fd != nullptr) {
  81     grpc_endpoint_shutdown(
  82         *client_fd, GRPC_ERROR_CREATE_FROM_STATIC_STRING("Forced Disconnect"));
  83     grpc_endpoint_destroy(*client_fd);
  84     grpc_core::ExecCtx::Get()->Flush();
  85     *client_fd = nullptr;
  86   }
  87 }
  88
  89 /* Runs client side validator */
  90 void grpc_run_client_side_validator(grpc_bad_client_arg* arg, uint32_t flags,
  91                                     grpc_endpoint_pair* sfd,
  92                                     grpc_completion_queue* client_cq) {
  93   char* hex;
  94   gpr_event done_write;
  95   if (arg->client_payload_length < 4 * 1024) {
  96     hex = gpr_dump(arg->client_payload, arg->client_payload_length,
  97                    GPR_DUMP_HEX | GPR_DUMP_ASCII);
  98     /* Add a debug log */
  99     gpr_log(GPR_INFO, "TEST: %s", hex);
 100     gpr_free(hex);
 101   } else {
 102     gpr_log(GPR_INFO, "TEST: (%" PRIdPTR " byte long string)",
 103             arg->client_payload_length);
 104   }
 105
 106   grpc_slice slice = grpc_slice_from_copied_buffer(arg->client_payload,
 107                                                    arg->client_payload_length);
 108   grpc_slice_buffer outgoing;
 109   grpc_closure done_write_closure;
 110   gpr_event_init(&done_write);
 111
 112   grpc_slice_buffer_init(&outgoing);
 113   grpc_slice_buffer_add(&outgoing, slice);
 114   GRPC_CLOSURE_INIT(&done_write_closure, set_done_write, &done_write,
 115                     grpc_schedule_on_exec_ctx);
 116
 117   /* Write data */
 118   grpc_endpoint_write(sfd->client, &outgoing, &done_write_closure, nullptr);
 119   grpc_core::ExecCtx::Get()->Flush();
 120
 121   /* Await completion, unless the request is large and write may not finish
 122    * before the peer shuts down. */
 123   if (!(flags & GRPC_BAD_CLIENT_LARGE_REQUEST)) {
 124     GPR_ASSERT(
 125         gpr_event_wait(&done_write, grpc_timeout_seconds_to_deadline(5)));
 126   }
 127
 128   if (flags & GRPC_BAD_CLIENT_DISCONNECT) {
 129     shutdown_client(&sfd->client);
 130   }
 131
 132   if (sfd->client != nullptr) {
 133     /* Validate client stream, if requested. */
 134     if (arg->client_validator != nullptr) {
 135       gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
 136       grpc_slice_buffer incoming;
 137       grpc_slice_buffer_init(&incoming);
 138       /* We may need to do multiple reads to read the complete server
 139        * response. */
 140       while (true) {
 141         gpr_event read_done_event;
 142         gpr_event_init(&read_done_event);
 143         grpc_closure read_done_closure;
 144         GRPC_CLOSURE_INIT(&read_done_closure, set_read_done, &read_done_event,
 145                           grpc_schedule_on_exec_ctx);
 146         grpc_endpoint_read(sfd->client, &incoming, &read_done_closure,
 147                            /*urgent=*/true);
 148         grpc_core::ExecCtx::Get()->Flush();
 149         do {
 150           GPR_ASSERT(gpr_time_cmp(deadline, gpr_now(deadline.clock_type)) > 0);
 151           /* Perform a cq next just to provide a thread that can read incoming
 152            bytes on the client fd */
 153           GPR_ASSERT(grpc_completion_queue_next(
 154                          client_cq, grpc_timeout_milliseconds_to_deadline(100),
 155                          nullptr)
 156                          .type == GRPC_QUEUE_TIMEOUT);
 157         } while (!gpr_event_get(&read_done_event));
 158         if (arg->client_validator(&incoming, arg->client_validator_arg)) break;
 159         gpr_log(GPR_INFO,
 160                 "client validator failed; trying additional read "
 161                 "in case we didn't get all the data");
 162       }
 163       grpc_slice_buffer_destroy_internal(&incoming);
 164     }
 165     grpc_core::ExecCtx::Get()->Flush();
 166   }
 167
 168   /* If the request was too large, then we need to forcefully shut down the
 169    * client, so that the write can be considered completed */
 170   if (flags & GRPC_BAD_CLIENT_LARGE_REQUEST) {
 171     shutdown_client(&sfd->client);
 172   }
 173
 174   /* Make sure that the client is done writing */
 175   while (!gpr_event_get(&done_write)) {
 176     GPR_ASSERT(
 177         grpc_completion_queue_next(
 178             client_cq, grpc_timeout_milliseconds_to_deadline(100), nullptr)
 179             .type == GRPC_QUEUE_TIMEOUT);
 180   }
 181
 182   grpc_slice_buffer_destroy_internal(&outgoing);
 183   grpc_core::ExecCtx::Get()->Flush();
 184 }
 185
 186 void grpc_run_bad_client_test(
 187     grpc_bad_client_server_side_validator server_validator,
 188     grpc_bad_client_arg args[], int num_args, uint32_t flags) {
 189   grpc_endpoint_pair sfd;
 190   thd_args a;
 191   grpc_transport* transport;
 192   grpc_core::ExecCtx exec_ctx;
 193   grpc_completion_queue* shutdown_cq;
 194   grpc_completion_queue* client_cq;
 195
 196   /* Init grpc */
 197   grpc_init();
 198
 199   /* Create endpoints */
 200   sfd = grpc_iomgr_create_endpoint_pair("fixture", nullptr);
 201
 202   /* Create server, completion events */
 203   a.server = grpc_server_create(nullptr, nullptr);
 204   a.cq = grpc_completion_queue_create_for_next(nullptr);
 205   client_cq = grpc_completion_queue_create_for_next(nullptr);
 206
 207   grpc_server_register_completion_queue(a.server, a.cq, nullptr);
 208   a.registered_method =
 209       grpc_server_register_method(a.server, GRPC_BAD_CLIENT_REGISTERED_METHOD,
 210                                   GRPC_BAD_CLIENT_REGISTERED_HOST,
 211                                   GRPC_SRM_PAYLOAD_READ_INITIAL_BYTE_BUFFER, 0);
 212   grpc_server_start(a.server);
 213   transport = grpc_create_chttp2_transport(nullptr, sfd.server, false);
 214   server_setup_transport(&a, transport);
 215   grpc_chttp2_transport_start_reading(transport, nullptr, nullptr);
 216
 217   /* Bind fds to pollsets */
 218   grpc_endpoint_add_to_pollset(sfd.client, grpc_cq_pollset(client_cq));
 219   grpc_endpoint_add_to_pollset(sfd.server, grpc_cq_pollset(a.cq));
 220
 221   /* Check a ground truth */
 222   GPR_ASSERT(grpc_server_has_open_connections(a.server));
 223
 224   gpr_event_init(&a.done_thd);
 225   a.validator = server_validator;
 226   /* Start validator */
 227
 228   grpc_core::Thread server_validator_thd("grpc_bad_client", thd_func, &a);
 229   server_validator_thd.Start();
 230   for (int i = 0; i < num_args; i++) {
 231     grpc_run_client_side_validator(&args[i], i == (num_args - 1) ? flags : 0,
 232                                    &sfd, client_cq);
 233   }
 234   /* Wait for server thread to finish */
 235   GPR_ASSERT(gpr_event_wait(&a.done_thd, grpc_timeout_seconds_to_deadline(1)));
 236
 237   /* Shutdown. */
 238   shutdown_client(&sfd.client);
 239   server_validator_thd.Join();
 240
 241   shutdown_cq = grpc_completion_queue_create_for_pluck(nullptr);
 242   grpc_server_shutdown_and_notify(a.server, shutdown_cq, nullptr);
 243   GPR_ASSERT(grpc_completion_queue_pluck(shutdown_cq, nullptr,
 244                                          grpc_timeout_seconds_to_deadline(1),
 245                                          nullptr)
 246                  .type == GRPC_OP_COMPLETE);
 247   grpc_completion_queue_destroy(shutdown_cq);
 248   grpc_server_destroy(a.server);
 249   grpc_completion_queue_destroy(a.cq);
 250   grpc_completion_queue_destroy(client_cq);
 251   grpc_shutdown();
 252 }
 253
 254 bool client_connection_preface_validator(grpc_slice_buffer* incoming,
 255                                          void* arg) {
 256   if (incoming->count < 1) {
 257     return false;
 258   }
 259   grpc_slice slice = incoming->slices[0];
 260   /* There should be at least one settings frame present */
 261   if (GRPC_SLICE_LENGTH(slice) < MIN_HTTP2_FRAME_SIZE) {
 262     return false;
 263   }
 264   const uint8_t* p = GRPC_SLICE_START_PTR(slice);
 265   /* Check the frame type (SETTINGS) */
 266   if (*(p + 3) != 4) {
 267     return false;
 268   }
 269   return true;
 270 }
 271
 272 /* connection preface and settings frame to be sent by the client */
 273 #define CONNECTION_PREFACE_FROM_CLIENT \
 274   "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"   \
 275   "\x00\x00\x00\x04\x00\x00\x00\x00\x00"
 276
 277 grpc_bad_client_arg connection_preface_arg = {
 278     client_connection_preface_validator, nullptr,
 279     CONNECTION_PREFACE_FROM_CLIENT, sizeof(CONNECTION_PREFACE_FROM_CLIENT) - 1};
 280
 281 bool rst_stream_client_validator(grpc_slice_buffer* incoming, void* arg) {
 282   // Get last frame from incoming slice buffer.
 283   grpc_slice_buffer last_frame_buffer;
 284   grpc_slice_buffer_init(&last_frame_buffer);
 285   grpc_slice_buffer_trim_end(incoming, 13, &last_frame_buffer);
 286   GPR_ASSERT(last_frame_buffer.count == 1);
 287   grpc_slice last_frame = last_frame_buffer.slices[0];
 288
 289   const uint8_t* p = GRPC_SLICE_START_PTR(last_frame);
 290   bool success =
 291       // Length == 4
 292       *p++ != 0 || *p++ != 0 || *p++ != 4 ||
 293       // Frame type (RST_STREAM)
 294       *p++ != 3 ||
 295       // Flags
 296       *p++ != 0 ||
 297       // Stream ID.
 298       *p++ != 0 || *p++ != 0 || *p++ != 0 || *p++ != 1 ||
 299       // Payload (error code)
 300       *p++ == 0 || *p++ == 0 || *p++ == 0 || *p == 0 || *p == 11;
 301
 302   if (!success) {
 303     gpr_log(GPR_INFO, "client expected RST_STREAM frame, not found");
 304   }
 305
 306   grpc_slice_buffer_destroy(&last_frame_buffer);
 307   return success;
 308 }
 309
 310 static void* tag(intptr_t t) { return (void*)t; }
 311
 312 void server_verifier_request_call(grpc_server* server,
 313                                   grpc_completion_queue* cq,
 314                                   void* registered_method) {
 315   grpc_call_error error;
 316   grpc_call* s;
 317   grpc_call_details call_details;
 318   cq_verifier* cqv = cq_verifier_create(cq);
 319   grpc_metadata_array request_metadata_recv;
 320
 321   grpc_call_details_init(&call_details);
 322   grpc_metadata_array_init(&request_metadata_recv);
 323
 324   error = grpc_server_request_call(server, &s, &call_details,
 325                                    &request_metadata_recv, cq, cq, tag(101));
 326   GPR_ASSERT(GRPC_CALL_OK == error);
 327   CQ_EXPECT_COMPLETION(cqv, tag(101), 1);
 328   cq_verify(cqv);
 329
 330   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.host, "localhost"));
 331   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo/bar"));
 332
 333   grpc_metadata_array_destroy(&request_metadata_recv);
 334   grpc_call_details_destroy(&call_details);
 335   grpc_call_unref(s);
 336   cq_verifier_destroy(cqv);
 337 }