1 /*
2 *
3 * Copyright 2017 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 <grpc/support/port_platform.h>
20
21 #include <grpc/grpc.h>
22 #include <grpc/support/alloc.h>
23 #include <grpc/support/log.h>
24 #include <grpc/support/string_util.h>
25 #include <grpc/support/sync.h>
26 #include <grpc/support/time.h>
27
28 #include <string.h>
29
30 #include <errno.h>
31 #include <fcntl.h>
32 #include <gflags/gflags.h>
33 #include <gmock/gmock.h>
34 #include <thread>
35 #include <vector>
36
37 #include "test/cpp/util/subprocess.h"
38 #include "test/cpp/util/test_config.h"
39
40 #include "src/core/ext/filters/client_channel/client_channel.h"
41 #include "src/core/ext/filters/client_channel/resolver.h"
42 #include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h"
43 #include "src/core/ext/filters/client_channel/resolver_registry.h"
44 #include "src/core/lib/channel/channel_args.h"
45 #include "src/core/lib/gpr/env.h"
46 #include "src/core/lib/gpr/host_port.h"
47 #include "src/core/lib/gpr/string.h"
48 #include "src/core/lib/gprpp/orphanable.h"
49 #include "src/core/lib/iomgr/combiner.h"
50 #include "src/core/lib/iomgr/executor.h"
51 #include "src/core/lib/iomgr/iomgr.h"
52 #include "src/core/lib/iomgr/resolve_address.h"
53 #include "src/core/lib/iomgr/sockaddr_utils.h"
54 #include "test/core/util/port.h"
55 #include "test/core/util/test_config.h"
56
57 // TODO: pull in different headers when enabling this
58 // test on windows. Also set BAD_SOCKET_RETURN_VAL
59 // to INVALID_SOCKET on windows.
60 #ifdef GPR_WINDOWS
61 #include "src/core/lib/iomgr/sockaddr_windows.h"
62 #include "src/core/lib/iomgr/socket_windows.h"
63 #include "src/core/lib/iomgr/tcp_windows.h"
64 #define BAD_SOCKET_RETURN_VAL INVALID_SOCKET
65 #else
66 #include "src/core/lib/iomgr/sockaddr_posix.h"
67 #define BAD_SOCKET_RETURN_VAL -1
68 #endif
69
70 using grpc::SubProcess;
71 using std::vector;
72 using testing::UnorderedElementsAreArray;
73
74 // Hack copied from "test/cpp/end2end/server_crash_test_client.cc"!
75 // In some distros, gflags is in the namespace google, and in some others,
76 // in gflags. This hack is enabling us to find both.
77 namespace google {}
78 namespace gflags {}
79 using namespace google;
80 using namespace gflags;
81
82 DEFINE_string(target_name, "", "Target name to resolve.");
83 DEFINE_string(expected_addrs, "",
84 "List of expected backend or balancer addresses in the form "
85 "'<ip0:port0>,<is_balancer0>;<ip1:port1>,<is_balancer1>;...'. "
86 "'is_balancer' should be bool, i.e. true or false.");
87 DEFINE_string(expected_chosen_service_config, "",
88 "Expected service config json string that gets chosen (no "
89 "whitespace). Empty for none.");
90 DEFINE_string(
91 local_dns_server_address, "",
92 "Optional. This address is placed as the uri authority if present.");
93 DEFINE_string(expected_lb_policy, "",
94 "Expected lb policy name that appears in resolver result channel "
95 "arg. Empty for none.");
96
97 namespace {
98
99 class GrpcLBAddress final {
100 public:
GrpcLBAddress(std::string address,bool is_balancer)101 GrpcLBAddress(std::string address, bool is_balancer)
102 : is_balancer(is_balancer), address(std::move(address)) {}
103
operator ==(const GrpcLBAddress & other) const104 bool operator==(const GrpcLBAddress& other) const {
105 return this->is_balancer == other.is_balancer &&
106 this->address == other.address;
107 }
108
operator !=(const GrpcLBAddress & other) const109 bool operator!=(const GrpcLBAddress& other) const {
110 return !(*this == other);
111 }
112
113 bool is_balancer;
114 std::string address;
115 };
116
ParseExpectedAddrs(std::string expected_addrs)117 vector<GrpcLBAddress> ParseExpectedAddrs(std::string expected_addrs) {
118 std::vector<GrpcLBAddress> out;
119 while (expected_addrs.size() != 0) {
120 // get the next <ip>,<port> (v4 or v6)
121 size_t next_comma = expected_addrs.find(',');
122 if (next_comma == std::string::npos) {
123 gpr_log(GPR_ERROR,
124 "Missing ','. Expected_addrs arg should be a semicolon-separated "
125 "list of <ip-port>,<bool> pairs. Left-to-be-parsed arg is |%s|",
126 expected_addrs.c_str());
127 abort();
128 }
129 std::string next_addr = expected_addrs.substr(0, next_comma);
130 expected_addrs = expected_addrs.substr(next_comma + 1, std::string::npos);
131 // get the next is_balancer 'bool' associated with this address
132 size_t next_semicolon = expected_addrs.find(';');
133 bool is_balancer =
134 gpr_is_true(expected_addrs.substr(0, next_semicolon).c_str());
135 out.emplace_back(GrpcLBAddress(next_addr, is_balancer));
136 if (next_semicolon == std::string::npos) {
137 break;
138 }
139 expected_addrs =
140 expected_addrs.substr(next_semicolon + 1, std::string::npos);
141 }
142 if (out.size() == 0) {
143 gpr_log(GPR_ERROR,
144 "expected_addrs arg should be a semicolon-separated list of "
145 "<ip-port>,<bool> pairs");
146 abort();
147 }
148 return out;
149 }
150
TestDeadline(void)151 gpr_timespec TestDeadline(void) {
152 return grpc_timeout_seconds_to_deadline(100);
153 }
154
155 struct ArgsStruct {
156 gpr_event ev;
157 gpr_atm done_atm;
158 gpr_mu* mu;
159 grpc_pollset* pollset;
160 grpc_pollset_set* pollset_set;
161 grpc_combiner* lock;
162 grpc_channel_args* channel_args;
163 vector<GrpcLBAddress> expected_addrs;
164 std::string expected_service_config_string;
165 std::string expected_lb_policy;
166 };
167
ArgsInit(ArgsStruct * args)168 void ArgsInit(ArgsStruct* args) {
169 gpr_event_init(&args->ev);
170 args->pollset = (grpc_pollset*)gpr_zalloc(grpc_pollset_size());
171 grpc_pollset_init(args->pollset, &args->mu);
172 args->pollset_set = grpc_pollset_set_create();
173 grpc_pollset_set_add_pollset(args->pollset_set, args->pollset);
174 args->lock = grpc_combiner_create();
175 gpr_atm_rel_store(&args->done_atm, 0);
176 args->channel_args = nullptr;
177 }
178
DoNothing(void * arg,grpc_error * error)179 void DoNothing(void* arg, grpc_error* error) {}
180
ArgsFinish(ArgsStruct * args)181 void ArgsFinish(ArgsStruct* args) {
182 GPR_ASSERT(gpr_event_wait(&args->ev, TestDeadline()));
183 grpc_pollset_set_del_pollset(args->pollset_set, args->pollset);
184 grpc_pollset_set_destroy(args->pollset_set);
185 grpc_closure DoNothing_cb;
186 GRPC_CLOSURE_INIT(&DoNothing_cb, DoNothing, nullptr,
187 grpc_schedule_on_exec_ctx);
188 grpc_pollset_shutdown(args->pollset, &DoNothing_cb);
189 // exec_ctx needs to be flushed before calling grpc_pollset_destroy()
190 grpc_channel_args_destroy(args->channel_args);
191 grpc_core::ExecCtx::Get()->Flush();
192 grpc_pollset_destroy(args->pollset);
193 gpr_free(args->pollset);
194 GRPC_COMBINER_UNREF(args->lock, nullptr);
195 }
196
NSecondDeadline(int seconds)197 gpr_timespec NSecondDeadline(int seconds) {
198 return gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
199 gpr_time_from_seconds(seconds, GPR_TIMESPAN));
200 }
201
PollPollsetUntilRequestDone(ArgsStruct * args)202 void PollPollsetUntilRequestDone(ArgsStruct* args) {
203 gpr_timespec deadline = NSecondDeadline(10);
204 while (true) {
205 bool done = gpr_atm_acq_load(&args->done_atm) != 0;
206 if (done) {
207 break;
208 }
209 gpr_timespec time_left =
210 gpr_time_sub(deadline, gpr_now(GPR_CLOCK_REALTIME));
211 gpr_log(GPR_DEBUG, "done=%d, time_left=%" PRId64 ".%09d", done,
212 time_left.tv_sec, time_left.tv_nsec);
213 GPR_ASSERT(gpr_time_cmp(time_left, gpr_time_0(GPR_TIMESPAN)) >= 0);
214 grpc_pollset_worker* worker = nullptr;
215 grpc_core::ExecCtx exec_ctx;
216 gpr_mu_lock(args->mu);
217 GRPC_LOG_IF_ERROR("pollset_work",
218 grpc_pollset_work(args->pollset, &worker,
219 grpc_timespec_to_millis_round_up(
220 NSecondDeadline(1))));
221 gpr_mu_unlock(args->mu);
222 }
223 gpr_event_set(&args->ev, (void*)1);
224 }
225
CheckServiceConfigResultLocked(grpc_channel_args * channel_args,ArgsStruct * args)226 void CheckServiceConfigResultLocked(grpc_channel_args* channel_args,
227 ArgsStruct* args) {
228 const grpc_arg* service_config_arg =
229 grpc_channel_args_find(channel_args, GRPC_ARG_SERVICE_CONFIG);
230 if (args->expected_service_config_string != "") {
231 GPR_ASSERT(service_config_arg != nullptr);
232 GPR_ASSERT(service_config_arg->type == GRPC_ARG_STRING);
233 EXPECT_EQ(service_config_arg->value.string,
234 args->expected_service_config_string);
235 } else {
236 GPR_ASSERT(service_config_arg == nullptr);
237 }
238 }
239
CheckLBPolicyResultLocked(grpc_channel_args * channel_args,ArgsStruct * args)240 void CheckLBPolicyResultLocked(grpc_channel_args* channel_args,
241 ArgsStruct* args) {
242 const grpc_arg* lb_policy_arg =
243 grpc_channel_args_find(channel_args, GRPC_ARG_LB_POLICY_NAME);
244 if (args->expected_lb_policy != "") {
245 GPR_ASSERT(lb_policy_arg != nullptr);
246 GPR_ASSERT(lb_policy_arg->type == GRPC_ARG_STRING);
247 EXPECT_EQ(lb_policy_arg->value.string, args->expected_lb_policy);
248 } else {
249 GPR_ASSERT(lb_policy_arg == nullptr);
250 }
251 }
252
253 #ifdef GPR_WINDOWS
OpenAndCloseSocketsStressLoop(int dummy_port,gpr_event * done_ev)254 void OpenAndCloseSocketsStressLoop(int dummy_port, gpr_event* done_ev) {
255 sockaddr_in6 addr;
256 memset(&addr, 0, sizeof(addr));
257 addr.sin6_family = AF_INET6;
258 addr.sin6_port = htons(dummy_port);
259 ((char*)&addr.sin6_addr)[15] = 1;
260 for (;;) {
261 if (gpr_event_get(done_ev)) {
262 return;
263 }
264 std::vector<int> sockets;
265 for (size_t i = 0; i < 50; i++) {
266 SOCKET s = WSASocket(AF_INET6, SOCK_STREAM, IPPROTO_TCP, nullptr, 0,
267 WSA_FLAG_OVERLAPPED);
268 ASSERT_TRUE(s != BAD_SOCKET_RETURN_VAL)
269 << "Failed to create TCP ipv6 socket";
270 gpr_log(GPR_DEBUG, "Opened socket: %d", s);
271 char val = 1;
272 ASSERT_TRUE(setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val)) !=
273 SOCKET_ERROR)
274 << "Failed to set socketopt reuseaddr. WSA error: " +
275 std::to_string(WSAGetLastError());
276 ASSERT_TRUE(grpc_tcp_set_non_block(s) == GRPC_ERROR_NONE)
277 << "Failed to set socket non-blocking";
278 ASSERT_TRUE(bind(s, (const sockaddr*)&addr, sizeof(addr)) != SOCKET_ERROR)
279 << "Failed to bind socket " + std::to_string(s) +
280 " to [::1]:" + std::to_string(dummy_port) +
281 ". WSA error: " + std::to_string(WSAGetLastError());
282 ASSERT_TRUE(listen(s, 1) != SOCKET_ERROR)
283 << "Failed to listen on socket " + std::to_string(s) +
284 ". WSA error: " + std::to_string(WSAGetLastError());
285 sockets.push_back(s);
286 }
287 // Do a non-blocking accept followed by a close on all of those sockets.
288 // Do this in a separate loop to try to induce a time window to hit races.
289 for (size_t i = 0; i < sockets.size(); i++) {
290 gpr_log(GPR_DEBUG, "non-blocking accept then close on %d", sockets[i]);
291 ASSERT_TRUE(accept(sockets[i], nullptr, nullptr) == INVALID_SOCKET)
292 << "Accept on dummy socket unexpectedly accepted actual connection.";
293 ASSERT_TRUE(WSAGetLastError() == WSAEWOULDBLOCK)
294 << "OpenAndCloseSocketsStressLoop accept on socket " +
295 std::to_string(sockets[i]) +
296 " failed in "
297 "an unexpected way. "
298 "WSA error: " +
299 std::to_string(WSAGetLastError()) +
300 ". Socket use-after-close bugs are likely.";
301 ASSERT_TRUE(closesocket(sockets[i]) != SOCKET_ERROR)
302 << "Failed to close socket: " + std::to_string(sockets[i]) +
303 ". WSA error: " + std::to_string(WSAGetLastError());
304 }
305 }
306 return;
307 }
308 #else
OpenAndCloseSocketsStressLoop(int dummy_port,gpr_event * done_ev)309 void OpenAndCloseSocketsStressLoop(int dummy_port, gpr_event* done_ev) {
310 // The goal of this loop is to catch socket
311 // "use after close" bugs within the c-ares resolver by acting
312 // like some separate thread doing I/O.
313 // It's goal is to try to hit race conditions whereby:
314 // 1) The c-ares resolver closes a socket.
315 // 2) This loop opens a socket with (coincidentally) the same handle.
316 // 3) the c-ares resolver mistakenly uses that same socket without
317 // realizing that its closed.
318 // 4) This loop performs an operation on that socket that should
319 // succeed but instead fails because of what the c-ares
320 // resolver did in the meantime.
321 sockaddr_in6 addr;
322 memset(&addr, 0, sizeof(addr));
323 addr.sin6_family = AF_INET6;
324 addr.sin6_port = htons(dummy_port);
325 ((char*)&addr.sin6_addr)[15] = 1;
326 for (;;) {
327 if (gpr_event_get(done_ev)) {
328 return;
329 }
330 std::vector<int> sockets;
331 // First open a bunch of sockets, bind and listen
332 // '50' is an arbitrary number that, experimentally,
333 // has a good chance of catching bugs.
334 for (size_t i = 0; i < 50; i++) {
335 int s = socket(AF_INET6, SOCK_STREAM, 0);
336 int val = 1;
337 ASSERT_TRUE(setsockopt(s, SOL_SOCKET, SO_REUSEPORT, &val, sizeof(val)) ==
338 0)
339 << "Failed to set socketopt reuseport";
340 ASSERT_TRUE(setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val)) ==
341 0)
342 << "Failed to set socket reuseaddr";
343 ASSERT_TRUE(fcntl(s, F_SETFL, O_NONBLOCK) == 0)
344 << "Failed to set socket non-blocking";
345 ASSERT_TRUE(s != BAD_SOCKET_RETURN_VAL)
346 << "Failed to create TCP ipv6 socket";
347 gpr_log(GPR_DEBUG, "Opened fd: %d", s);
348 ASSERT_TRUE(bind(s, (const sockaddr*)&addr, sizeof(addr)) == 0)
349 << "Failed to bind socket " + std::to_string(s) +
350 " to [::1]:" + std::to_string(dummy_port) +
351 ". errno: " + std::to_string(errno);
352 ASSERT_TRUE(listen(s, 1) == 0) << "Failed to listen on socket " +
353 std::to_string(s) +
354 ". errno: " + std::to_string(errno);
355 sockets.push_back(s);
356 }
357 // Do a non-blocking accept followed by a close on all of those sockets.
358 // Do this in a separate loop to try to induce a time window to hit races.
359 for (size_t i = 0; i < sockets.size(); i++) {
360 gpr_log(GPR_DEBUG, "non-blocking accept then close on %d", sockets[i]);
361 if (accept(sockets[i], nullptr, nullptr)) {
362 // If e.g. a "shutdown" was called on this fd from another thread,
363 // then this accept call should fail with an unexpected error.
364 ASSERT_TRUE(errno == EAGAIN || errno == EWOULDBLOCK)
365 << "OpenAndCloseSocketsStressLoop accept on socket " +
366 std::to_string(sockets[i]) +
367 " failed in "
368 "an unexpected way. "
369 "errno: " +
370 std::to_string(errno) +
371 ". Socket use-after-close bugs are likely.";
372 }
373 ASSERT_TRUE(close(sockets[i]) == 0)
374 << "Failed to close socket: " + std::to_string(sockets[i]) +
375 ". errno: " + std::to_string(errno);
376 }
377 }
378 }
379 #endif
380
CheckResolverResultLocked(void * argsp,grpc_error * err)381 void CheckResolverResultLocked(void* argsp, grpc_error* err) {
382 EXPECT_EQ(err, GRPC_ERROR_NONE);
383 ArgsStruct* args = (ArgsStruct*)argsp;
384 grpc_channel_args* channel_args = args->channel_args;
385 const grpc_arg* channel_arg =
386 grpc_channel_args_find(channel_args, GRPC_ARG_LB_ADDRESSES);
387 GPR_ASSERT(channel_arg != nullptr);
388 GPR_ASSERT(channel_arg->type == GRPC_ARG_POINTER);
389 grpc_lb_addresses* addresses =
390 (grpc_lb_addresses*)channel_arg->value.pointer.p;
391 gpr_log(GPR_INFO, "num addrs found: %" PRIdPTR ". expected %" PRIdPTR,
392 addresses->num_addresses, args->expected_addrs.size());
393 GPR_ASSERT(addresses->num_addresses == args->expected_addrs.size());
394 std::vector<GrpcLBAddress> found_lb_addrs;
395 for (size_t i = 0; i < addresses->num_addresses; i++) {
396 grpc_lb_address addr = addresses->addresses[i];
397 char* str;
398 grpc_sockaddr_to_string(&str, &addr.address, 1 /* normalize */);
399 gpr_log(GPR_INFO, "%s", str);
400 found_lb_addrs.emplace_back(
401 GrpcLBAddress(std::string(str), addr.is_balancer));
402 gpr_free(str);
403 }
404 if (args->expected_addrs.size() != found_lb_addrs.size()) {
405 gpr_log(GPR_DEBUG,
406 "found lb addrs size is: %" PRIdPTR
407 ". expected addrs size is %" PRIdPTR,
408 found_lb_addrs.size(), args->expected_addrs.size());
409 abort();
410 }
411 EXPECT_THAT(args->expected_addrs, UnorderedElementsAreArray(found_lb_addrs));
412 CheckServiceConfigResultLocked(channel_args, args);
413 if (args->expected_service_config_string == "") {
414 CheckLBPolicyResultLocked(channel_args, args);
415 }
416 gpr_atm_rel_store(&args->done_atm, 1);
417 gpr_mu_lock(args->mu);
418 GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(args->pollset, nullptr));
419 gpr_mu_unlock(args->mu);
420 }
421
CheckResolvedWithoutErrorLocked(void * argsp,grpc_error * err)422 void CheckResolvedWithoutErrorLocked(void* argsp, grpc_error* err) {
423 EXPECT_EQ(err, GRPC_ERROR_NONE);
424 ArgsStruct* args = (ArgsStruct*)argsp;
425 gpr_atm_rel_store(&args->done_atm, 1);
426 gpr_mu_lock(args->mu);
427 GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(args->pollset, nullptr));
428 gpr_mu_unlock(args->mu);
429 }
430
RunResolvesRelevantRecordsTest(void (* OnDoneLocked)(void * arg,grpc_error * error))431 void RunResolvesRelevantRecordsTest(void (*OnDoneLocked)(void* arg,
432 grpc_error* error)) {
433 grpc_core::ExecCtx exec_ctx;
434 ArgsStruct args;
435 ArgsInit(&args);
436 args.expected_addrs = ParseExpectedAddrs(FLAGS_expected_addrs);
437 args.expected_service_config_string = FLAGS_expected_chosen_service_config;
438 args.expected_lb_policy = FLAGS_expected_lb_policy;
439 // maybe build the address with an authority
440 char* whole_uri = nullptr;
441 GPR_ASSERT(gpr_asprintf(&whole_uri, "dns://%s/%s",
442 FLAGS_local_dns_server_address.c_str(),
443 FLAGS_target_name.c_str()));
444 // create resolver and resolve
445 grpc_core::OrphanablePtr<grpc_core::Resolver> resolver =
446 grpc_core::ResolverRegistry::CreateResolver(whole_uri, nullptr,
447 args.pollset_set, args.lock);
448 gpr_free(whole_uri);
449 grpc_closure on_resolver_result_changed;
450 GRPC_CLOSURE_INIT(&on_resolver_result_changed, OnDoneLocked, (void*)&args,
451 grpc_combiner_scheduler(args.lock));
452 resolver->NextLocked(&args.channel_args, &on_resolver_result_changed);
453 grpc_core::ExecCtx::Get()->Flush();
454 PollPollsetUntilRequestDone(&args);
455 ArgsFinish(&args);
456 }
457
TEST(ResolverComponentTest,TestResolvesRelevantRecords)458 TEST(ResolverComponentTest, TestResolvesRelevantRecords) {
459 RunResolvesRelevantRecordsTest(CheckResolverResultLocked);
460 }
461
TEST(ResolverComponentTest,TestResolvesRelevantRecordsWithConcurrentFdStress)462 TEST(ResolverComponentTest, TestResolvesRelevantRecordsWithConcurrentFdStress) {
463 // Start up background stress thread
464 int dummy_port = grpc_pick_unused_port_or_die();
465 gpr_event done_ev;
466 gpr_event_init(&done_ev);
467 std::thread socket_stress_thread(OpenAndCloseSocketsStressLoop, dummy_port,
468 &done_ev);
469 // Run the resolver test
470 RunResolvesRelevantRecordsTest(CheckResolvedWithoutErrorLocked);
471 // Shutdown and join stress thread
472 gpr_event_set(&done_ev, (void*)1);
473 socket_stress_thread.join();
474 }
475
476 } // namespace
477
main(int argc,char ** argv)478 int main(int argc, char** argv) {
479 grpc_init();
480 grpc_test_init(argc, argv);
481 ::testing::InitGoogleTest(&argc, argv);
482 ParseCommandLineFlags(&argc, &argv, true);
483 if (FLAGS_target_name == "") {
484 gpr_log(GPR_ERROR, "Missing target_name param.");
485 abort();
486 }
487 if (FLAGS_local_dns_server_address != "") {
488 gpr_log(GPR_INFO, "Specifying authority in uris to: %s",
489 FLAGS_local_dns_server_address.c_str());
490 }
491 auto result = RUN_ALL_TESTS();
492 grpc_shutdown();
493 return result;
494 }
495