• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 */
thd_func(void * arg)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 */
set_done_write(void * arg,grpc_error * error)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 
server_setup_transport(void * ts,grpc_transport * transport)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));
70 }
71 
72 /* Sets the read_done event */
set_read_done(void * arg,grpc_error * error)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 */
shutdown_client(grpc_endpoint ** client_fd)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 */
grpc_run_client_side_validator(grpc_bad_client_arg * arg,uint32_t flags,grpc_endpoint_pair * sfd,grpc_completion_queue * client_cq)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         grpc_core::ExecCtx::Get()->Flush();
148         do {
149           GPR_ASSERT(gpr_time_cmp(deadline, gpr_now(deadline.clock_type)) > 0);
150           /* Perform a cq next just to provide a thread that can read incoming
151            bytes on the client fd */
152           GPR_ASSERT(grpc_completion_queue_next(
153                          client_cq, grpc_timeout_milliseconds_to_deadline(100),
154                          nullptr)
155                          .type == GRPC_QUEUE_TIMEOUT);
156         } while (!gpr_event_get(&read_done_event));
157         if (arg->client_validator(&incoming, arg->client_validator_arg)) break;
158         gpr_log(GPR_INFO,
159                 "client validator failed; trying additional read "
160                 "in case we didn't get all the data");
161       }
162       grpc_slice_buffer_destroy_internal(&incoming);
163     }
164     grpc_core::ExecCtx::Get()->Flush();
165   }
166 
167   /* If the request was too large, then we need to forcefully shut down the
168    * client, so that the write can be considered completed */
169   if (flags & GRPC_BAD_CLIENT_LARGE_REQUEST) {
170     shutdown_client(&sfd->client);
171   }
172 
173   /* Make sure that the client is done writing */
174   while (!gpr_event_get(&done_write)) {
175     GPR_ASSERT(
176         grpc_completion_queue_next(
177             client_cq, grpc_timeout_milliseconds_to_deadline(100), nullptr)
178             .type == GRPC_QUEUE_TIMEOUT);
179   }
180 
181   grpc_slice_buffer_destroy_internal(&outgoing);
182   grpc_core::ExecCtx::Get()->Flush();
183 }
184 
grpc_run_bad_client_test(grpc_bad_client_server_side_validator server_validator,grpc_bad_client_arg args[],int num_args,uint32_t flags)185 void grpc_run_bad_client_test(
186     grpc_bad_client_server_side_validator server_validator,
187     grpc_bad_client_arg args[], int num_args, uint32_t flags) {
188   grpc_endpoint_pair sfd;
189   thd_args a;
190   grpc_transport* transport;
191   grpc_core::ExecCtx exec_ctx;
192   grpc_completion_queue* shutdown_cq;
193   grpc_completion_queue* client_cq;
194 
195   /* Init grpc */
196   grpc_init();
197 
198   /* Create endpoints */
199   sfd = grpc_iomgr_create_endpoint_pair("fixture", nullptr);
200 
201   /* Create server, completion events */
202   a.server = grpc_server_create(nullptr, nullptr);
203   a.cq = grpc_completion_queue_create_for_next(nullptr);
204   client_cq = grpc_completion_queue_create_for_next(nullptr);
205 
206   grpc_server_register_completion_queue(a.server, a.cq, nullptr);
207   a.registered_method =
208       grpc_server_register_method(a.server, GRPC_BAD_CLIENT_REGISTERED_METHOD,
209                                   GRPC_BAD_CLIENT_REGISTERED_HOST,
210                                   GRPC_SRM_PAYLOAD_READ_INITIAL_BYTE_BUFFER, 0);
211   grpc_server_start(a.server);
212   transport = grpc_create_chttp2_transport(nullptr, sfd.server, false);
213   server_setup_transport(&a, transport);
214   grpc_chttp2_transport_start_reading(transport, nullptr, nullptr);
215 
216   /* Bind fds to pollsets */
217   grpc_endpoint_add_to_pollset(sfd.client, grpc_cq_pollset(client_cq));
218   grpc_endpoint_add_to_pollset(sfd.server, grpc_cq_pollset(a.cq));
219 
220   /* Check a ground truth */
221   GPR_ASSERT(grpc_server_has_open_connections(a.server));
222 
223   gpr_event_init(&a.done_thd);
224   a.validator = server_validator;
225   /* Start validator */
226 
227   grpc_core::Thread server_validator_thd("grpc_bad_client", thd_func, &a);
228   server_validator_thd.Start();
229   for (int i = 0; i < num_args; i++) {
230     grpc_run_client_side_validator(&args[i], i == (num_args - 1) ? flags : 0,
231                                    &sfd, client_cq);
232   }
233   /* Wait for server thread to finish */
234   GPR_ASSERT(gpr_event_wait(&a.done_thd, grpc_timeout_seconds_to_deadline(1)));
235 
236   /* Shutdown. */
237   shutdown_client(&sfd.client);
238   server_validator_thd.Join();
239 
240   shutdown_cq = grpc_completion_queue_create_for_pluck(nullptr);
241   grpc_server_shutdown_and_notify(a.server, shutdown_cq, nullptr);
242   GPR_ASSERT(grpc_completion_queue_pluck(shutdown_cq, nullptr,
243                                          grpc_timeout_seconds_to_deadline(1),
244                                          nullptr)
245                  .type == GRPC_OP_COMPLETE);
246   grpc_completion_queue_destroy(shutdown_cq);
247   grpc_server_destroy(a.server);
248   grpc_completion_queue_destroy(a.cq);
249   grpc_completion_queue_destroy(client_cq);
250   grpc_shutdown();
251 }
252 
client_connection_preface_validator(grpc_slice_buffer * incoming,void * arg)253 bool client_connection_preface_validator(grpc_slice_buffer* incoming,
254                                          void* arg) {
255   if (incoming->count < 1) {
256     return false;
257   }
258   grpc_slice slice = incoming->slices[0];
259   /* There should be atleast a settings frame present */
260   if (GRPC_SLICE_LENGTH(slice) < MIN_HTTP2_FRAME_SIZE) {
261     return false;
262   }
263   const uint8_t* p = GRPC_SLICE_START_PTR(slice);
264   /* Check the frame type (SETTINGS) */
265   if (*(p + 3) != 4) {
266     return false;
267   }
268   return true;
269 }
270 
271 /* connection preface and settings frame to be sent by the client */
272 #define CONNECTION_PREFACE_FROM_CLIENT \
273   "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"   \
274   "\x00\x00\x00\x04\x00\x00\x00\x00\x00"
275 
276 grpc_bad_client_arg connection_preface_arg = {
277     client_connection_preface_validator, nullptr,
278     CONNECTION_PREFACE_FROM_CLIENT, sizeof(CONNECTION_PREFACE_FROM_CLIENT) - 1};
279 
rst_stream_client_validator(grpc_slice_buffer * incoming,void * arg)280 bool rst_stream_client_validator(grpc_slice_buffer* incoming, void* arg) {
281   // Get last frame from incoming slice buffer.
282   grpc_slice_buffer last_frame_buffer;
283   grpc_slice_buffer_init(&last_frame_buffer);
284   grpc_slice_buffer_trim_end(incoming, 13, &last_frame_buffer);
285   GPR_ASSERT(last_frame_buffer.count == 1);
286   grpc_slice last_frame = last_frame_buffer.slices[0];
287 
288   const uint8_t* p = GRPC_SLICE_START_PTR(last_frame);
289   bool success =
290       // Length == 4
291       *p++ != 0 || *p++ != 0 || *p++ != 4 ||
292       // Frame type (RST_STREAM)
293       *p++ != 3 ||
294       // Flags
295       *p++ != 0 ||
296       // Stream ID.
297       *p++ != 0 || *p++ != 0 || *p++ != 0 || *p++ != 1 ||
298       // Payload (error code)
299       *p++ == 0 || *p++ == 0 || *p++ == 0 || *p == 0 || *p == 11;
300 
301   if (!success) {
302     gpr_log(GPR_INFO, "client expected RST_STREAM frame, not found");
303   }
304 
305   grpc_slice_buffer_destroy(&last_frame_buffer);
306   return success;
307 }
308 
tag(intptr_t t)309 static void* tag(intptr_t t) { return (void*)t; }
310 
server_verifier_request_call(grpc_server * server,grpc_completion_queue * cq,void * registered_method)311 void server_verifier_request_call(grpc_server* server,
312                                   grpc_completion_queue* cq,
313                                   void* registered_method) {
314   grpc_call_error error;
315   grpc_call* s;
316   grpc_call_details call_details;
317   cq_verifier* cqv = cq_verifier_create(cq);
318   grpc_metadata_array request_metadata_recv;
319 
320   grpc_call_details_init(&call_details);
321   grpc_metadata_array_init(&request_metadata_recv);
322 
323   error = grpc_server_request_call(server, &s, &call_details,
324                                    &request_metadata_recv, cq, cq, tag(101));
325   GPR_ASSERT(GRPC_CALL_OK == error);
326   CQ_EXPECT_COMPLETION(cqv, tag(101), 1);
327   cq_verify(cqv);
328 
329   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.host, "localhost"));
330   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo/bar"));
331 
332   grpc_metadata_array_destroy(&request_metadata_recv);
333   grpc_call_details_destroy(&call_details);
334   grpc_call_unref(s);
335   cq_verifier_destroy(cqv);
336 }
337