1 //
2 //
3 // Copyright 2016 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 <inttypes.h>
20 #include <limits.h>
21 #include <string.h>
22
23 #include <memory>
24 #include <string>
25 #include <vector>
26
27 #include <grpc/grpc.h>
28 #include <grpc/grpc_security.h>
29 #include <grpc/impl/propagation_bits.h>
30 #include <grpc/slice.h>
31 #include <grpc/slice_buffer.h>
32 #include <grpc/status.h>
33 #include <grpc/support/alloc.h>
34 #include <grpc/support/atm.h>
35 #include <grpc/support/log.h>
36 #include <grpc/support/sync.h>
37 #include <grpc/support/time.h>
38
39 #include "src/core/lib/event_engine/shim.h"
40 #include "src/core/lib/gpr/string.h"
41 #include "src/core/lib/gprpp/host_port.h"
42 #include "src/core/lib/gprpp/notification.h"
43 #include "src/core/lib/gprpp/status_helper.h"
44 #include "src/core/lib/gprpp/thd.h"
45 #include "src/core/lib/iomgr/closure.h"
46 #include "src/core/lib/iomgr/endpoint.h"
47 #include "src/core/lib/iomgr/error.h"
48 #include "src/core/lib/iomgr/exec_ctx.h"
49 #include "src/core/lib/iomgr/iomgr_fwd.h"
50 #include "src/core/lib/iomgr/tcp_server.h"
51 #include "src/core/lib/slice/slice_string_helpers.h"
52 #include "test/core/end2end/cq_verifier.h"
53 #include "test/core/util/port.h"
54 #include "test/core/util/test_config.h"
55 #include "test/core/util/test_tcp_server.h"
56
57 #define HTTP1_RESP_400 \
58 "HTTP/1.0 400 Bad Request\n" \
59 "Content-Type: text/html; charset=UTF-8\n" \
60 "Content-Length: 0\n" \
61 "Date: Tue, 07 Jun 2016 17:43:20 GMT\n\n"
62
63 #define HTTP2_SETTINGS_FRAME "\x00\x00\x00\x04\x00\x00\x00\x00\x00"
64
65 #define HTTP2_RESP(STATUS_CODE) \
66 "\x00\x00>\x01\x04\x00\x00\x00\x01" \
67 "\x10\x0e" \
68 "content-length\x01" \
69 "0" \
70 "\x10\x0c" \
71 "content-type\x10" \
72 "application/grpc" \
73 "\x10\x07:status\x03" #STATUS_CODE
74
75 #define UNPARSEABLE_RESP "Bad Request\n"
76
77 #define HTTP2_DETAIL_MSG(STATUS_CODE) \
78 "Received http2 header with status: " #STATUS_CODE
79
80 // TODO(zyc) Check the content of incoming data instead of using this length
81 // The 'bad' server will start sending responses after reading this amount of
82 // data from the client.
83 #define SERVER_INCOMING_DATA_LENGTH_LOWER_THRESHOLD (size_t)200
84
85 struct rpc_state {
86 std::string target;
87 grpc_completion_queue* cq;
88 grpc_channel* channel;
89 grpc_call* call;
90 size_t incoming_data_length;
91 grpc_slice_buffer temp_incoming_buffer;
92 grpc_slice_buffer outgoing_buffer;
93 grpc_endpoint* tcp;
94 gpr_atm done_atm;
95 bool http2_response;
96 bool send_settings;
97 const char* response_payload;
98 size_t response_payload_length;
99 bool connection_attempt_made;
100 std::unique_ptr<grpc_core::Notification> on_connect_done;
101 };
102
103 static int server_port;
104 static struct rpc_state state;
105 static grpc_closure on_read;
106 static grpc_closure on_writing_settings_frame;
107 static grpc_closure on_write;
108
tag(intptr_t t)109 static void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
110
done_write(void *,grpc_error_handle error)111 static void done_write(void* /*arg*/, grpc_error_handle error) {
112 GPR_ASSERT(error.ok());
113 gpr_atm_rel_store(&state.done_atm, 1);
114 }
115
done_writing_settings_frame(void *,grpc_error_handle error)116 static void done_writing_settings_frame(void* /* arg */,
117 grpc_error_handle error) {
118 GPR_ASSERT(error.ok());
119 grpc_endpoint_read(state.tcp, &state.temp_incoming_buffer, &on_read,
120 /*urgent=*/false, /*min_progress_size=*/1);
121 }
122
handle_write()123 static void handle_write() {
124 grpc_slice slice = grpc_slice_from_copied_buffer(
125 state.response_payload, state.response_payload_length);
126
127 grpc_slice_buffer_reset_and_unref(&state.outgoing_buffer);
128 grpc_slice_buffer_add(&state.outgoing_buffer, slice);
129 grpc_endpoint_write(state.tcp, &state.outgoing_buffer, &on_write, nullptr,
130 /*max_frame_size=*/INT_MAX);
131 }
132
handle_read(void *,grpc_error_handle error)133 static void handle_read(void* /*arg*/, grpc_error_handle error) {
134 if (!error.ok()) {
135 gpr_log(GPR_ERROR, "handle_read error: %s",
136 grpc_core::StatusToString(error).c_str());
137 return;
138 }
139 state.incoming_data_length += state.temp_incoming_buffer.length;
140
141 size_t i;
142 for (i = 0; i < state.temp_incoming_buffer.count; i++) {
143 char* dump = grpc_dump_slice(state.temp_incoming_buffer.slices[i],
144 GPR_DUMP_HEX | GPR_DUMP_ASCII);
145 gpr_log(GPR_DEBUG, "Server received: %s", dump);
146 gpr_free(dump);
147 }
148
149 gpr_log(GPR_DEBUG,
150 "got %" PRIuPTR " bytes, expected %" PRIuPTR
151 " bytes or a non-HTTP2 response to be sent",
152 state.incoming_data_length,
153 SERVER_INCOMING_DATA_LENGTH_LOWER_THRESHOLD);
154 if (state.incoming_data_length >=
155 SERVER_INCOMING_DATA_LENGTH_LOWER_THRESHOLD ||
156 !state.http2_response) {
157 handle_write();
158 } else {
159 grpc_endpoint_read(state.tcp, &state.temp_incoming_buffer, &on_read,
160 /*urgent=*/false, /*min_progress_size=*/1);
161 }
162 }
163
on_connect(void * arg,grpc_endpoint * tcp,grpc_pollset *,grpc_tcp_server_acceptor * acceptor)164 static void on_connect(void* arg, grpc_endpoint* tcp,
165 grpc_pollset* /*accepting_pollset*/,
166 grpc_tcp_server_acceptor* acceptor) {
167 gpr_free(acceptor);
168 test_tcp_server* server = static_cast<test_tcp_server*>(arg);
169 GRPC_CLOSURE_INIT(&on_read, handle_read, nullptr, grpc_schedule_on_exec_ctx);
170 GRPC_CLOSURE_INIT(&on_writing_settings_frame, done_writing_settings_frame,
171 nullptr, grpc_schedule_on_exec_ctx);
172 GRPC_CLOSURE_INIT(&on_write, done_write, nullptr, grpc_schedule_on_exec_ctx);
173 grpc_slice_buffer_init(&state.temp_incoming_buffer);
174 grpc_slice_buffer_init(&state.outgoing_buffer);
175 state.connection_attempt_made = true;
176 state.tcp = tcp;
177 state.incoming_data_length = 0;
178 grpc_endpoint_add_to_pollset(tcp, server->pollset[0]);
179 if (state.send_settings) {
180 // Send settings frame from server
181 grpc_slice slice = grpc_slice_from_static_buffer(
182 HTTP2_SETTINGS_FRAME, sizeof(HTTP2_SETTINGS_FRAME) - 1);
183 grpc_slice_buffer_add(&state.outgoing_buffer, slice);
184 grpc_endpoint_write(state.tcp, &state.outgoing_buffer,
185 &on_writing_settings_frame, nullptr,
186 /*max_frame_size=*/INT_MAX);
187 } else {
188 grpc_endpoint_read(state.tcp, &state.temp_incoming_buffer, &on_read,
189 /*urgent=*/false, /*min_progress_size=*/1);
190 }
191 state.on_connect_done->Notify();
192 }
193
n_sec_deadline(int seconds)194 static gpr_timespec n_sec_deadline(int seconds) {
195 return gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
196 gpr_time_from_seconds(seconds, GPR_TIMESPAN));
197 }
198
start_rpc(int target_port,grpc_status_code expected_status,const char * expected_detail)199 static void start_rpc(int target_port, grpc_status_code expected_status,
200 const char* expected_detail) {
201 grpc_op ops[6];
202 grpc_op* op;
203 grpc_metadata_array initial_metadata_recv;
204 grpc_metadata_array trailing_metadata_recv;
205 grpc_status_code status;
206 grpc_call_error error;
207 grpc_slice details;
208
209 state.cq = grpc_completion_queue_create_for_next(nullptr);
210 grpc_core::CqVerifier cqv(state.cq);
211 state.target = grpc_core::JoinHostPort("127.0.0.1", target_port);
212
213 grpc_channel_credentials* creds = grpc_insecure_credentials_create();
214 state.channel = grpc_channel_create(state.target.c_str(), creds, nullptr);
215 grpc_channel_credentials_release(creds);
216 grpc_slice host = grpc_slice_from_static_string("localhost");
217 // The default connect deadline is 20 seconds, so reduce the RPC deadline to 1
218 // second. This helps us verify - a) If the server responded with a non-HTTP2
219 // response, the connect fails immediately resulting in
220 // GRPC_STATUS_UNAVAILABLE instead of GRPC_STATUS_DEADLINE_EXCEEDED. b) If the
221 // server does not send a HTTP2 SETTINGs frame, the RPC fails with a
222 // DEADLINE_EXCEEDED.
223 state.call = grpc_channel_create_call(
224 state.channel, nullptr, GRPC_PROPAGATE_DEFAULTS, state.cq,
225 grpc_slice_from_static_string("/Service/Method"), &host,
226 n_sec_deadline(5), nullptr);
227
228 grpc_metadata_array_init(&initial_metadata_recv);
229 grpc_metadata_array_init(&trailing_metadata_recv);
230
231 memset(ops, 0, sizeof(ops));
232 op = ops;
233 op->op = GRPC_OP_SEND_INITIAL_METADATA;
234 op->data.send_initial_metadata.count = 0;
235 op->flags = 0;
236 op->reserved = nullptr;
237 op++;
238 op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
239 op->flags = 0;
240 op->reserved = nullptr;
241 op++;
242 op->op = GRPC_OP_RECV_INITIAL_METADATA;
243 op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
244 op->flags = 0;
245 op->reserved = nullptr;
246 op++;
247 op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
248 op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
249 op->data.recv_status_on_client.status = &status;
250 op->data.recv_status_on_client.status_details = &details;
251 op->flags = 0;
252 op->reserved = nullptr;
253 op++;
254 error = grpc_call_start_batch(state.call, ops, static_cast<size_t>(op - ops),
255 tag(1), nullptr);
256
257 GPR_ASSERT(GRPC_CALL_OK == error);
258
259 cqv.Expect(tag(1), true);
260 cqv.Verify();
261
262 GPR_ASSERT(status == expected_status);
263 if (expected_detail != nullptr) {
264 GPR_ASSERT(-1 != grpc_slice_slice(details, grpc_slice_from_static_string(
265 expected_detail)));
266 }
267
268 grpc_metadata_array_destroy(&initial_metadata_recv);
269 grpc_metadata_array_destroy(&trailing_metadata_recv);
270 grpc_slice_unref(details);
271 }
272
cleanup_rpc()273 static void cleanup_rpc() {
274 grpc_event ev;
275 grpc_slice_buffer_destroy(&state.temp_incoming_buffer);
276 grpc_slice_buffer_destroy(&state.outgoing_buffer);
277 grpc_call_unref(state.call);
278 grpc_completion_queue_shutdown(state.cq);
279 do {
280 ev = grpc_completion_queue_next(state.cq, n_sec_deadline(1), nullptr);
281 } while (ev.type != GRPC_QUEUE_SHUTDOWN);
282 grpc_completion_queue_destroy(state.cq);
283 grpc_channel_destroy(state.channel);
284 state.target.clear();
285 }
286
287 typedef struct {
288 test_tcp_server* server;
289 gpr_event* signal_when_done;
290 } poll_args;
291
actually_poll_server(void * arg)292 static void actually_poll_server(void* arg) {
293 poll_args* pa = static_cast<poll_args*>(arg);
294 gpr_timespec deadline = n_sec_deadline(5);
295 while (true) {
296 bool done = gpr_atm_acq_load(&state.done_atm) != 0;
297 gpr_timespec time_left =
298 gpr_time_sub(deadline, gpr_now(GPR_CLOCK_REALTIME));
299 gpr_log(GPR_DEBUG, "done=%d, time_left=%" PRId64 ".%09d", done,
300 time_left.tv_sec, time_left.tv_nsec);
301 if (done || gpr_time_cmp(time_left, gpr_time_0(GPR_TIMESPAN)) < 0) {
302 break;
303 }
304 int milliseconds = 1000;
305 if (grpc_event_engine::experimental::UseEventEngineListener()) {
306 milliseconds = 10;
307 }
308 test_tcp_server_poll(pa->server, milliseconds);
309 }
310 gpr_event_set(pa->signal_when_done, reinterpret_cast<void*>(1));
311 gpr_free(pa);
312 }
313
poll_server_until_read_done(test_tcp_server * server,gpr_event * signal_when_done)314 static grpc_core::Thread* poll_server_until_read_done(
315 test_tcp_server* server, gpr_event* signal_when_done) {
316 gpr_atm_rel_store(&state.done_atm, 0);
317 state.connection_attempt_made = false;
318 poll_args* pa = static_cast<poll_args*>(gpr_malloc(sizeof(*pa)));
319 pa->server = server;
320 pa->signal_when_done = signal_when_done;
321 auto* th =
322 new grpc_core::Thread("grpc_poll_server", actually_poll_server, pa);
323 th->Start();
324 return th;
325 }
326
run_test(bool http2_response,bool send_settings,const char * response_payload,size_t response_payload_length,grpc_status_code expected_status,const char * expected_detail)327 static void run_test(bool http2_response, bool send_settings,
328 const char* response_payload,
329 size_t response_payload_length,
330 grpc_status_code expected_status,
331 const char* expected_detail) {
332 test_tcp_server test_server;
333 grpc_core::ExecCtx exec_ctx;
334 gpr_event ev;
335
336 grpc_init();
337 gpr_event_init(&ev);
338 server_port = grpc_pick_unused_port_or_die();
339 test_tcp_server_init(&test_server, on_connect, &test_server);
340 test_tcp_server_start(&test_server, server_port);
341 state.on_connect_done = std::make_unique<grpc_core::Notification>();
342 state.http2_response = http2_response;
343 state.send_settings = send_settings;
344 state.response_payload = response_payload;
345 state.response_payload_length = response_payload_length;
346
347 // poll server until sending out the response
348 std::unique_ptr<grpc_core::Thread> thdptr(
349 poll_server_until_read_done(&test_server, &ev));
350 start_rpc(server_port, expected_status, expected_detail);
351 gpr_event_wait(&ev, gpr_inf_future(GPR_CLOCK_REALTIME));
352 thdptr->Join();
353 state.on_connect_done->WaitForNotification();
354 // Proof that the server accepted the TCP connection.
355 GPR_ASSERT(state.connection_attempt_made == true);
356 // clean up
357 grpc_endpoint_shutdown(state.tcp, GRPC_ERROR_CREATE("Test Shutdown"));
358 grpc_endpoint_destroy(state.tcp);
359 cleanup_rpc();
360 grpc_core::ExecCtx::Get()->Flush();
361 test_tcp_server_destroy(&test_server);
362
363 grpc_shutdown();
364 }
365
main(int argc,char ** argv)366 int main(int argc, char** argv) {
367 grpc::testing::TestEnvironment env(&argc, argv);
368 grpc_init();
369 // status defined in hpack static table
370 run_test(true, true, HTTP2_RESP(204), sizeof(HTTP2_RESP(204)) - 1,
371 GRPC_STATUS_UNKNOWN, HTTP2_DETAIL_MSG(204));
372 run_test(true, true, HTTP2_RESP(206), sizeof(HTTP2_RESP(206)) - 1,
373 GRPC_STATUS_UNKNOWN, HTTP2_DETAIL_MSG(206));
374 run_test(true, true, HTTP2_RESP(304), sizeof(HTTP2_RESP(304)) - 1,
375 GRPC_STATUS_UNKNOWN, HTTP2_DETAIL_MSG(304));
376 run_test(true, true, HTTP2_RESP(400), sizeof(HTTP2_RESP(400)) - 1,
377 GRPC_STATUS_INTERNAL, HTTP2_DETAIL_MSG(400));
378 run_test(true, true, HTTP2_RESP(404), sizeof(HTTP2_RESP(404)) - 1,
379 GRPC_STATUS_UNIMPLEMENTED, HTTP2_DETAIL_MSG(404));
380 run_test(true, true, HTTP2_RESP(500), sizeof(HTTP2_RESP(500)) - 1,
381 GRPC_STATUS_UNKNOWN, HTTP2_DETAIL_MSG(500));
382
383 // status not defined in hpack static table
384 run_test(true, true, HTTP2_RESP(401), sizeof(HTTP2_RESP(401)) - 1,
385 GRPC_STATUS_UNAUTHENTICATED, HTTP2_DETAIL_MSG(401));
386 run_test(true, true, HTTP2_RESP(403), sizeof(HTTP2_RESP(403)) - 1,
387 GRPC_STATUS_PERMISSION_DENIED, HTTP2_DETAIL_MSG(403));
388 run_test(true, true, HTTP2_RESP(429), sizeof(HTTP2_RESP(429)) - 1,
389 GRPC_STATUS_UNAVAILABLE, HTTP2_DETAIL_MSG(429));
390 run_test(true, true, HTTP2_RESP(499), sizeof(HTTP2_RESP(499)) - 1,
391 GRPC_STATUS_UNKNOWN, HTTP2_DETAIL_MSG(499));
392 run_test(true, true, HTTP2_RESP(502), sizeof(HTTP2_RESP(502)) - 1,
393 GRPC_STATUS_UNAVAILABLE, HTTP2_DETAIL_MSG(502));
394 run_test(true, true, HTTP2_RESP(503), sizeof(HTTP2_RESP(503)) - 1,
395 GRPC_STATUS_UNAVAILABLE, HTTP2_DETAIL_MSG(503));
396 run_test(true, true, HTTP2_RESP(504), sizeof(HTTP2_RESP(504)) - 1,
397 GRPC_STATUS_UNAVAILABLE, HTTP2_DETAIL_MSG(504));
398 // unparseable response. RPC should fail immediately due to a connect
399 // failure.
400 //
401 run_test(false, false, UNPARSEABLE_RESP, sizeof(UNPARSEABLE_RESP) - 1,
402 GRPC_STATUS_UNAVAILABLE, nullptr);
403
404 // http1 response. RPC should fail immediately due to a connect failure.
405 run_test(false, false, HTTP1_RESP_400, sizeof(HTTP1_RESP_400) - 1,
406 GRPC_STATUS_UNAVAILABLE, nullptr);
407
408 // http2 response without sending a SETTINGs frame. RPC should fail with
409 // DEADLINE_EXCEEDED since the RPC deadline is lower than the connection
410 // attempt deadline.
411 run_test(true, false, HTTP2_RESP(404), sizeof(HTTP2_RESP(404)) - 1,
412 GRPC_STATUS_DEADLINE_EXCEEDED, nullptr);
413 grpc_shutdown();
414 return 0;
415 }
416