src/shrpx_connection_handler.cc

   1 /*
   2  * nghttp2 - HTTP/2 C Library
   3  *
   4  * Copyright (c) 2012 Tatsuhiro Tsujikawa
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining
   7  * a copy of this software and associated documentation files (the
   8  * "Software"), to deal in the Software without restriction, including
   9  * without limitation the rights to use, copy, modify, merge, publish,
  10  * distribute, sublicense, and/or sell copies of the Software, and to
  11  * permit persons to whom the Software is furnished to do so, subject to
  12  * the following conditions:
  13  *
  14  * The above copyright notice and this permission notice shall be
  15  * included in all copies or substantial portions of the Software.
  16  *
  17  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  18  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  19  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  20  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
  21  * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
  22  * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
  23  * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  24  */
  25 #include "shrpx_connection_handler.h"
  26
  27 #ifdef HAVE_UNISTD_H
  28 #include <unistd.h>
  29 #endif // HAVE_UNISTD_H
  30 #include <sys/types.h>
  31 #include <sys/wait.h>
  32
  33 #include <cerrno>
  34 #include <thread>
  35
  36 #include "shrpx_client_handler.h"
  37 #include "shrpx_ssl.h"
  38 #include "shrpx_worker.h"
  39 #include "shrpx_config.h"
  40 #include "shrpx_http2_session.h"
  41 #include "shrpx_connect_blocker.h"
  42 #include "shrpx_downstream_connection.h"
  43 #include "shrpx_accept_handler.h"
  44 #include "util.h"
  45 #include "template.h"
  46
  47 using namespace nghttp2;
  48
  49 namespace shrpx {
  50
  51 namespace {
  52 void acceptor_disable_cb(struct ev_loop *loop, ev_timer *w, int revent) {
  53   auto h = static_cast<ConnectionHandler *>(w->data);
  54
  55   // If we are in graceful shutdown period, we must not enable
  56   // acceptors again.
  57   if (h->get_graceful_shutdown()) {
  58     return;
  59   }
  60
  61   h->enable_acceptor();
  62 }
  63 } // namespace
  64
  65 namespace {
  66 void ocsp_cb(struct ev_loop *loop, ev_timer *w, int revent) {
  67   auto h = static_cast<ConnectionHandler *>(w->data);
  68
  69   // If we are in graceful shutdown period, we won't do ocsp query.
  70   if (h->get_graceful_shutdown()) {
  71     return;
  72   }
  73
  74   LOG(NOTICE) << "Start ocsp update";
  75
  76   h->proceed_next_cert_ocsp();
  77 }
  78 } // namespace
  79
  80 namespace {
  81 void ocsp_read_cb(struct ev_loop *loop, ev_io *w, int revent) {
  82   auto h = static_cast<ConnectionHandler *>(w->data);
  83
  84   h->read_ocsp_chunk();
  85 }
  86 } // namespace
  87
  88 namespace {
  89 void ocsp_chld_cb(struct ev_loop *loop, ev_child *w, int revent) {
  90   auto h = static_cast<ConnectionHandler *>(w->data);
  91
  92   h->handle_ocsp_complete();
  93 }
  94 } // namespace
  95
  96 ConnectionHandler::ConnectionHandler(struct ev_loop *loop)
  97     : single_worker_(nullptr), loop_(loop), worker_round_robin_cnt_(0),
  98       graceful_shutdown_(false) {
  99   ev_timer_init(&disable_acceptor_timer_, acceptor_disable_cb, 0., 0.);
 100   disable_acceptor_timer_.data = this;
 101
 102   ev_timer_init(&ocsp_timer_, ocsp_cb, 0., 0.);
 103   ocsp_timer_.data = this;
 104
 105   ev_io_init(&ocsp_.rev, ocsp_read_cb, -1, EV_READ);
 106   ocsp_.rev.data = this;
 107
 108   ev_child_init(&ocsp_.chldev, ocsp_chld_cb, 0, 0);
 109   ocsp_.chldev.data = this;
 110
 111   ocsp_.next = 0;
 112   ocsp_.fd = -1;
 113
 114   reset_ocsp();
 115 }
 116
 117 ConnectionHandler::~ConnectionHandler() {
 118   ev_timer_stop(loop_, &disable_acceptor_timer_);
 119   ev_timer_stop(loop_, &ocsp_timer_);
 120 }
 121
 122 void ConnectionHandler::worker_reopen_log_files() {
 123   WorkerEvent wev;
 124
 125   memset(&wev, 0, sizeof(wev));
 126   wev.type = REOPEN_LOG;
 127
 128   for (auto &worker : workers_) {
 129     worker->send(wev);
 130   }
 131 }
 132
 133 void ConnectionHandler::worker_renew_ticket_keys(
 134     const std::shared_ptr<TicketKeys> &ticket_keys) {
 135   WorkerEvent wev;
 136
 137   memset(&wev, 0, sizeof(wev));
 138   wev.type = RENEW_TICKET_KEYS;
 139   wev.ticket_keys = ticket_keys;
 140
 141   for (auto &worker : workers_) {
 142     worker->send(wev);
 143   }
 144 }
 145
 146 void ConnectionHandler::create_single_worker() {
 147   auto cert_tree = ssl::create_cert_lookup_tree();
 148   auto sv_ssl_ctx = ssl::setup_server_ssl_context(all_ssl_ctx_, cert_tree);
 149   auto cl_ssl_ctx = ssl::setup_client_ssl_context();
 150
 151   single_worker_ = make_unique<Worker>(loop_, sv_ssl_ctx, cl_ssl_ctx, cert_tree,
 152                                        ticket_keys_);
 153 }
 154
 155 void ConnectionHandler::create_worker_thread(size_t num) {
 156 #ifndef NOTHREADS
 157   assert(workers_.size() == 0);
 158
 159   auto cert_tree = ssl::create_cert_lookup_tree();
 160   auto sv_ssl_ctx = ssl::setup_server_ssl_context(all_ssl_ctx_, cert_tree);
 161   auto cl_ssl_ctx = ssl::setup_client_ssl_context();
 162
 163   for (size_t i = 0; i < num; ++i) {
 164     auto loop = ev_loop_new(0);
 165
 166     auto worker = make_unique<Worker>(loop, sv_ssl_ctx, cl_ssl_ctx, cert_tree,
 167                                       ticket_keys_);
 168     worker->run_async();
 169     workers_.push_back(std::move(worker));
 170
 171     if (LOG_ENABLED(INFO)) {
 172       LLOG(INFO, this) << "Created thread #" << workers_.size() - 1;
 173     }
 174   }
 175 #endif // NOTHREADS
 176 }
 177
 178 void ConnectionHandler::join_worker() {
 179 #ifndef NOTHREADS
 180   int n = 0;
 181
 182   if (LOG_ENABLED(INFO)) {
 183     LLOG(INFO, this) << "Waiting for worker thread to join: n="
 184                      << workers_.size();
 185   }
 186
 187   for (auto &worker : workers_) {
 188     worker->wait();
 189     if (LOG_ENABLED(INFO)) {
 190       LLOG(INFO, this) << "Thread #" << n << " joined";
 191     }
 192     ++n;
 193   }
 194 #endif // NOTHREADS
 195 }
 196
 197 void ConnectionHandler::graceful_shutdown_worker() {
 198   if (get_config()->num_worker == 1) {
 199     return;
 200   }
 201
 202   WorkerEvent wev;
 203   memset(&wev, 0, sizeof(wev));
 204   wev.type = GRACEFUL_SHUTDOWN;
 205
 206   if (LOG_ENABLED(INFO)) {
 207     LLOG(INFO, this) << "Sending graceful shutdown signal to worker";
 208   }
 209
 210   for (auto &worker : workers_) {
 211
 212     worker->send(wev);
 213   }
 214 }
 215
 216 int ConnectionHandler::handle_connection(int fd, sockaddr *addr, int addrlen) {
 217   if (LOG_ENABLED(INFO)) {
 218     LLOG(INFO, this) << "Accepted connection. fd=" << fd;
 219   }
 220
 221   if (get_config()->num_worker == 1) {
 222
 223     if (single_worker_->get_worker_stat()->num_connections >=
 224         get_config()->worker_frontend_connections) {
 225
 226       if (LOG_ENABLED(INFO)) {
 227         LLOG(INFO, this) << "Too many connections >="
 228                          << get_config()->worker_frontend_connections;
 229       }
 230
 231       close(fd);
 232       return -1;
 233     }
 234
 235     auto client =
 236         ssl::accept_connection(single_worker_.get(), fd, addr, addrlen);
 237     if (!client) {
 238       LLOG(ERROR, this) << "ClientHandler creation failed";
 239
 240       close(fd);
 241       return -1;
 242     }
 243
 244     return 0;
 245   }
 246
 247   size_t idx = worker_round_robin_cnt_ % workers_.size();
 248   if (LOG_ENABLED(INFO)) {
 249     LOG(INFO) << "Dispatch connection to worker #" << idx;
 250   }
 251   ++worker_round_robin_cnt_;
 252   WorkerEvent wev;
 253   memset(&wev, 0, sizeof(wev));
 254   wev.type = NEW_CONNECTION;
 255   wev.client_fd = fd;
 256   memcpy(&wev.client_addr, addr, addrlen);
 257   wev.client_addrlen = addrlen;
 258
 259   workers_[idx]->send(wev);
 260
 261   return 0;
 262 }
 263
 264 struct ev_loop *ConnectionHandler::get_loop() const {
 265   return loop_;
 266 }
 267
 268 Worker *ConnectionHandler::get_single_worker() const {
 269   return single_worker_.get();
 270 }
 271
 272 void ConnectionHandler::set_acceptor(std::unique_ptr<AcceptHandler> h) {
 273   acceptor_ = std::move(h);
 274 }
 275
 276 AcceptHandler *ConnectionHandler::get_acceptor() const {
 277   return acceptor_.get();
 278 }
 279
 280 void ConnectionHandler::set_acceptor6(std::unique_ptr<AcceptHandler> h) {
 281   acceptor6_ = std::move(h);
 282 }
 283
 284 AcceptHandler *ConnectionHandler::get_acceptor6() const {
 285   return acceptor6_.get();
 286 }
 287
 288 void ConnectionHandler::enable_acceptor() {
 289   if (acceptor_) {
 290     acceptor_->enable();
 291   }
 292
 293   if (acceptor6_) {
 294     acceptor6_->enable();
 295   }
 296 }
 297
 298 void ConnectionHandler::disable_acceptor() {
 299   if (acceptor_) {
 300     acceptor_->disable();
 301   }
 302
 303   if (acceptor6_) {
 304     acceptor6_->disable();
 305   }
 306 }
 307
 308 void ConnectionHandler::disable_acceptor_temporary(ev_tstamp t) {
 309   if (t == 0. || ev_is_active(&disable_acceptor_timer_)) {
 310     return;
 311   }
 312
 313   disable_acceptor();
 314
 315   ev_timer_set(&disable_acceptor_timer_, t, 0.);
 316   ev_timer_start(loop_, &disable_acceptor_timer_);
 317 }
 318
 319 void ConnectionHandler::accept_pending_connection() {
 320   if (acceptor_) {
 321     acceptor_->accept_connection();
 322   }
 323   if (acceptor6_) {
 324     acceptor6_->accept_connection();
 325   }
 326 }
 327
 328 void
 329 ConnectionHandler::set_ticket_keys(std::shared_ptr<TicketKeys> ticket_keys) {
 330   ticket_keys_ = std::move(ticket_keys);
 331   if (single_worker_) {
 332     single_worker_->set_ticket_keys(ticket_keys_);
 333   }
 334 }
 335
 336 const std::shared_ptr<TicketKeys> &ConnectionHandler::get_ticket_keys() const {
 337   return ticket_keys_;
 338 }
 339
 340 void ConnectionHandler::set_graceful_shutdown(bool f) {
 341   graceful_shutdown_ = f;
 342   if (single_worker_) {
 343     single_worker_->set_graceful_shutdown(f);
 344   }
 345 }
 346
 347 bool ConnectionHandler::get_graceful_shutdown() const {
 348   return graceful_shutdown_;
 349 }
 350
 351 void ConnectionHandler::cancel_ocsp_update() {
 352   if (ocsp_.pid == 0) {
 353     return;
 354   }
 355
 356   kill(ocsp_.pid, SIGTERM);
 357 }
 358
 359 // inspired by h2o_read_command function from h2o project:
 360 // https://github.com/h2o/h2o
 361 int ConnectionHandler::start_ocsp_update(const char *cert_file) {
 362   int rv;
 363   int pfd[2];
 364
 365   if (LOG_ENABLED(INFO)) {
 366     LOG(INFO) << "Start ocsp update for " << cert_file;
 367   }
 368
 369   assert(!ev_is_active(&ocsp_.rev));
 370   assert(!ev_is_active(&ocsp_.chldev));
 371
 372   char *const argv[] = {
 373       const_cast<char *>(get_config()->fetch_ocsp_response_file.get()),
 374       const_cast<char *>(cert_file), nullptr};
 375   char *const envp[] = {nullptr};
 376
 377 #ifdef O_CLOEXEC
 378   if (pipe2(pfd, O_CLOEXEC) == -1) {
 379     return -1;
 380   }
 381 #else  // !O_CLOEXEC
 382   if (pipe(pfd) == -1) {
 383     return -1;
 384   }
 385   util::make_socket_closeonexec(pfd[0]);
 386   util::make_socket_closeonexec(pfd[1]);
 387 #endif // !O_CLOEXEC
 388
 389   auto closer = defer([&pfd]() {
 390     if (pfd[0] != -1) {
 391       close(pfd[0]);
 392     }
 393
 394     if (pfd[1] != -1) {
 395       close(pfd[1]);
 396     }
 397   });
 398
 399   auto pid = fork();
 400   if (pid == -1) {
 401     auto error = errno;
 402     LOG(WARN) << "Could not execute ocsp query command for " << cert_file
 403               << ": " << argv[0] << ", fork() failed, errno=" << error;
 404     return -1;
 405   }
 406
 407   if (pid == 0) {
 408     // child process
 409     dup2(pfd[1], 1);
 410     close(pfd[0]);
 411
 412     rv = execve(argv[0], argv, envp);
 413     if (rv == -1) {
 414       auto error = errno;
 415       LOG(WARN) << "Could not execute ocsp query command: " << argv[0]
 416                 << ", execve() faild, errno=" << error;
 417       _Exit(EXIT_FAILURE);
 418     }
 419     // unreachable
 420   }
 421
 422   // parent process
 423   close(pfd[1]);
 424   pfd[1] = -1;
 425
 426   ocsp_.pid = pid;
 427   ocsp_.fd = pfd[0];
 428   pfd[0] = -1;
 429
 430   util::make_socket_nonblocking(ocsp_.fd);
 431   ev_io_set(&ocsp_.rev, ocsp_.fd, EV_READ);
 432   ev_io_start(loop_, &ocsp_.rev);
 433
 434   ev_child_set(&ocsp_.chldev, ocsp_.pid, 0);
 435   ev_child_start(loop_, &ocsp_.chldev);
 436
 437   return 0;
 438 }
 439
 440 void ConnectionHandler::read_ocsp_chunk() {
 441   std::array<uint8_t, 4096> buf;
 442   for (;;) {
 443     ssize_t n;
 444     while ((n = read(ocsp_.fd, buf.data(), buf.size())) == -1 && errno == EINTR)
 445       ;
 446
 447     if (n == -1) {
 448       if (errno == EAGAIN || errno == EWOULDBLOCK) {
 449         return;
 450       }
 451       auto error = errno;
 452       LOG(WARN) << "Reading from ocsp query command failed: errno=" << error;
 453       ocsp_.error = error;
 454
 455       break;
 456     }
 457
 458     if (n == 0) {
 459       break;
 460     }
 461
 462     std::copy_n(std::begin(buf), n, std::back_inserter(ocsp_.resp));
 463   }
 464
 465   ev_io_stop(loop_, &ocsp_.rev);
 466 }
 467
 468 void ConnectionHandler::handle_ocsp_complete() {
 469   ev_io_stop(loop_, &ocsp_.rev);
 470   ev_child_stop(loop_, &ocsp_.chldev);
 471
 472   assert(ocsp_.next < all_ssl_ctx_.size());
 473
 474   auto ssl_ctx = all_ssl_ctx_[ocsp_.next];
 475   auto tls_ctx_data =
 476       static_cast<ssl::TLSContextData *>(SSL_CTX_get_app_data(ssl_ctx));
 477
 478   auto rstatus = ocsp_.chldev.rstatus;
 479   auto status = WEXITSTATUS(rstatus);
 480   if (ocsp_.error || !WIFEXITED(rstatus) || status != 0) {
 481     LOG(WARN) << "ocsp query command for " << tls_ctx_data->cert_file
 482               << " failed: error=" << ocsp_.error << ", rstatus=" << rstatus
 483               << ", status=" << status;
 484     ++ocsp_.next;
 485     proceed_next_cert_ocsp();
 486     return;
 487   }
 488
 489   if (LOG_ENABLED(INFO)) {
 490     LOG(INFO) << "ocsp update for " << tls_ctx_data->cert_file
 491               << " finished successfully";
 492   }
 493
 494   {
 495     std::lock_guard<std::mutex> g(tls_ctx_data->mu);
 496     tls_ctx_data->ocsp_data = std::move(ocsp_.resp);
 497   }
 498
 499   ++ocsp_.next;
 500   proceed_next_cert_ocsp();
 501 }
 502
 503 void ConnectionHandler::reset_ocsp() {
 504   if (ocsp_.fd != -1) {
 505     close(ocsp_.fd);
 506   }
 507
 508   ocsp_.fd = -1;
 509   ocsp_.pid = 0;
 510   ocsp_.error = 0;
 511   ocsp_.resp = std::vector<uint8_t>();
 512 }
 513
 514 void ConnectionHandler::proceed_next_cert_ocsp() {
 515   for (;;) {
 516     reset_ocsp();
 517     if (ocsp_.next == all_ssl_ctx_.size()) {
 518       ocsp_.next = 0;
 519       // We have updated all ocsp response, and schedule next update.
 520       ev_timer_set(&ocsp_timer_, get_config()->ocsp_update_interval, 0.);
 521       ev_timer_start(loop_, &ocsp_timer_);
 522       return;
 523     }
 524
 525     auto ssl_ctx = all_ssl_ctx_[ocsp_.next];
 526     auto tls_ctx_data =
 527         static_cast<ssl::TLSContextData *>(SSL_CTX_get_app_data(ssl_ctx));
 528     auto cert_file = tls_ctx_data->cert_file;
 529
 530     if (start_ocsp_update(cert_file) != 0) {
 531       ++ocsp_.next;
 532       continue;
 533     }
 534
 535     break;
 536   }
 537 }
 538
 539 } // namespace shrpx