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: 101 GrpcLBAddress(std::string address, bool is_balancer) 102 : is_balancer(is_balancer), address(std::move(address)) {} 103 104 bool operator==(const GrpcLBAddress& other) const { 105 return this->is_balancer == other.is_balancer && 106 this->address == other.address; 107 } 108 109 bool operator!=(const GrpcLBAddress& other) const { 110 return !(*this == other); 111 } 112 113 bool is_balancer; 114 std::string address; 115 }; 116 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 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 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 179 void DoNothing(void* arg, grpc_error* error) {} 180 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 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 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 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 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 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 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 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 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 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 458 TEST(ResolverComponentTest, TestResolvesRelevantRecords) { 459 RunResolvesRelevantRecordsTest(CheckResolverResultLocked); 460 } 461 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 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
