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