pdns/dnsdistdist/dnsdist-nghttp2.cc

   1 /*
   2  * This file is part of PowerDNS or dnsdist.
   3  * Copyright -- PowerDNS.COM B.V. and its contributors
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of version 2 of the GNU General Public License as
   7  * published by the Free Software Foundation.
   8  *
   9  * In addition, for the avoidance of any doubt, permission is granted to
  10  * link this program with OpenSSL and to (re)distribute the binaries
  11  * produced as the result of such linking.
  12  *
  13  * This program is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  21  */
  22
  23 #include "config.h"
  24
  25 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
  26 #include <nghttp2/nghttp2.h>
  27 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
  28
  29 #include "dnsdist-nghttp2.hh"
  30 #include "dnsdist-nghttp2-in.hh"
  31 #include "dnsdist-tcp.hh"
  32 #include "dnsdist-tcp-downstream.hh"
  33 #include "dnsdist-downstream-connection.hh"
  34
  35 #include "dolog.hh"
  36 #include "channel.hh"
  37 #include "iputils.hh"
  38 #include "libssl.hh"
  39 #include "noinitvector.hh"
  40 #include "tcpiohandler.hh"
  41 #include "threadname.hh"
  42 #include "sstuff.hh"
  43
  44 std::atomic<uint64_t> g_dohStatesDumpRequested{0};
  45 std::unique_ptr<DoHClientCollection> g_dohClientThreads{nullptr};
  46 std::optional<uint16_t> g_outgoingDoHWorkerThreads{std::nullopt};
  47
  48 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
  49 class DoHConnectionToBackend : public ConnectionToBackend
  50 {
  51 public:
  52   DoHConnectionToBackend(const std::shared_ptr<DownstreamState>& ds, std::unique_ptr<FDMultiplexer>& mplexer, const struct timeval& now, std::string&& proxyProtocolPayload);
  53
  54   void handleTimeout(const struct timeval& now, bool write) override;
  55   void queueQuery(std::shared_ptr<TCPQuerySender>& sender, TCPQuery&& query) override;
  56
  57   std::string toString() const override
  58   {
  59     ostringstream o;
  60     o << "DoH connection to backend " << (d_ds ? d_ds->getName() : "empty") << " over FD " << (d_handler ? std::to_string(d_handler->getDescriptor()) : "no socket") << ", " << getConcurrentStreamsCount() << " streams";
  61     return o.str();
  62   }
  63
  64   void setHealthCheck(bool h)
  65   {
  66     d_healthCheckQuery = h;
  67   }
  68
  69   void stopIO() override;
  70   bool reachedMaxConcurrentQueries() const override;
  71   bool reachedMaxStreamID() const override;
  72   bool isIdle() const override;
  73   void release() override
  74   {
  75   }
  76
  77 private:
  78   static ssize_t send_callback(nghttp2_session* session, const uint8_t* data, size_t length, int flags, void* user_data);
  79   static int on_frame_recv_callback(nghttp2_session* session, const nghttp2_frame* frame, void* user_data);
  80   static int on_data_chunk_recv_callback(nghttp2_session* session, uint8_t flags, int32_t stream_id, const uint8_t* data, size_t len, void* user_data);
  81   static int on_stream_close_callback(nghttp2_session* session, int32_t stream_id, uint32_t error_code, void* user_data);
  82   static int on_header_callback(nghttp2_session* session, const nghttp2_frame* frame, const uint8_t* name, size_t namelen, const uint8_t* value, size_t valuelen, uint8_t flags, void* user_data);
  83   static int on_error_callback(nghttp2_session* session, int lib_error_code, const char* msg, size_t len, void* user_data);
  84   static void handleReadableIOCallback(int fd, FDMultiplexer::funcparam_t& param);
  85   static void handleWritableIOCallback(int fd, FDMultiplexer::funcparam_t& param);
  86
  87   class PendingRequest
  88   {
  89   public:
  90     std::shared_ptr<TCPQuerySender> d_sender{nullptr};
  91     TCPQuery d_query;
  92     PacketBuffer d_buffer;
  93     size_t d_queryPos{0};
  94     uint16_t d_responseCode{0};
  95     bool d_finished{false};
  96   };
  97   void updateIO(IOState newState, FDMultiplexer::callbackfunc_t callback, bool noTTD = false);
  98   void watchForRemoteHostClosingConnection();
  99   void handleResponse(PendingRequest&& request);
 100   void handleResponseError(PendingRequest&& request, const struct timeval& now);
 101   void handleIOError();
 102   uint32_t getConcurrentStreamsCount() const;
 103
 104   size_t getUsageCount() const
 105   {
 106     auto ref = shared_from_this();
 107     return ref.use_count();
 108   }
 109
 110   static const std::unordered_map<std::string, std::string> s_constants;
 111
 112   std::unordered_map<int32_t, PendingRequest> d_currentStreams;
 113   std::string d_proxyProtocolPayload;
 114   PacketBuffer d_out;
 115   PacketBuffer d_in;
 116   std::unique_ptr<nghttp2_session, void (*)(nghttp2_session*)> d_session{nullptr, nghttp2_session_del};
 117   size_t d_outPos{0};
 118   size_t d_inPos{0};
 119   bool d_healthCheckQuery{false};
 120   bool d_firstWrite{true};
 121 };
 122
 123 using DownstreamDoHConnectionsManager = DownstreamConnectionsManager<DoHConnectionToBackend>;
 124 thread_local DownstreamDoHConnectionsManager t_downstreamDoHConnectionsManager;
 125
 126 uint32_t DoHConnectionToBackend::getConcurrentStreamsCount() const
 127 {
 128   return d_currentStreams.size();
 129 }
 130
 131 void DoHConnectionToBackend::handleResponse(PendingRequest&& request)
 132 {
 133   struct timeval now
 134   {
 135     .tv_sec = 0, .tv_usec = 0
 136   };
 137
 138   gettimeofday(&now, nullptr);
 139   try {
 140     if (!d_healthCheckQuery) {
 141       const double udiff = request.d_query.d_idstate.queryRealTime.udiff();
 142       d_ds->updateTCPLatency(udiff);
 143       if (request.d_buffer.size() >= sizeof(dnsheader)) {
 144         dnsheader dh;
 145         memcpy(&dh, request.d_buffer.data(), sizeof(dh));
 146         d_ds->reportResponse(dh.rcode);
 147       }
 148       else {
 149         d_ds->reportTimeoutOrError();
 150       }
 151     }
 152
 153     TCPResponse response(std::move(request.d_query));
 154     response.d_buffer = std::move(request.d_buffer);
 155     response.d_connection = shared_from_this();
 156     response.d_ds = d_ds;
 157     request.d_sender->handleResponse(now, std::move(response));
 158   }
 159   catch (const std::exception& e) {
 160     vinfolog("Got exception while handling response for cross-protocol DoH: %s", e.what());
 161   }
 162 }
 163
 164 void DoHConnectionToBackend::handleResponseError(PendingRequest&& request, const struct timeval& now)
 165 {
 166   try {
 167     if (!d_healthCheckQuery) {
 168       d_ds->reportTimeoutOrError();
 169     }
 170
 171     TCPResponse response(PacketBuffer(), std::move(request.d_query.d_idstate), nullptr, nullptr);
 172     request.d_sender->notifyIOError(now, std::move(response));
 173   }
 174   catch (const std::exception& e) {
 175     vinfolog("Got exception while handling response for cross-protocol DoH: %s", e.what());
 176   }
 177 }
 178
 179 void DoHConnectionToBackend::handleIOError()
 180 {
 181   d_connectionDied = true;
 182   nghttp2_session_terminate_session(d_session.get(), NGHTTP2_PROTOCOL_ERROR);
 183
 184   struct timeval now
 185   {
 186     .tv_sec = 0, .tv_usec = 0
 187   };
 188
 189   gettimeofday(&now, nullptr);
 190   for (auto& request : d_currentStreams) {
 191     handleResponseError(std::move(request.second), now);
 192   }
 193
 194   d_currentStreams.clear();
 195   stopIO();
 196 }
 197
 198 void DoHConnectionToBackend::handleTimeout(const struct timeval& now, bool write)
 199 {
 200   if (write) {
 201     if (d_firstWrite) {
 202       ++d_ds->tcpConnectTimeouts;
 203     }
 204     else {
 205       ++d_ds->tcpWriteTimeouts;
 206     }
 207   }
 208   else {
 209     ++d_ds->tcpReadTimeouts;
 210   }
 211
 212   handleIOError();
 213 }
 214
 215 bool DoHConnectionToBackend::reachedMaxStreamID() const
 216 {
 217   const uint32_t maximumStreamID = (static_cast<uint32_t>(1) << 31) - 1;
 218   return d_highestStreamID == maximumStreamID;
 219 }
 220
 221 bool DoHConnectionToBackend::reachedMaxConcurrentQueries() const
 222 {
 223   // cerr<<"Got "<<getConcurrentStreamsCount()<<" concurrent streams, max is "<<nghttp2_session_get_remote_settings(d_session.get(), NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS)<<endl;
 224   if (nghttp2_session_get_remote_settings(d_session.get(), NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS) <= getConcurrentStreamsCount()) {
 225     return true;
 226   }
 227   return false;
 228 }
 229
 230 bool DoHConnectionToBackend::isIdle() const
 231 {
 232   return getConcurrentStreamsCount() == 0;
 233 }
 234
 235 void DoHConnectionToBackend::queueQuery(std::shared_ptr<TCPQuerySender>& sender, TCPQuery&& query)
 236 {
 237   auto payloadSize = std::to_string(query.d_buffer.size());
 238
 239   bool addXForwarded = d_ds->d_config.d_addXForwardedHeaders;
 240
 241   /* We use nghttp2_nv_flag.NGHTTP2_NV_FLAG_NO_COPY_NAME and nghttp2_nv_flag.NGHTTP2_NV_FLAG_NO_COPY_VALUE
 242      to avoid a copy and lowercasing but we need to make sure that the data will outlive the request (nghttp2_on_frame_send_callback called), and that it is already lowercased. */
 243   std::vector<nghttp2_nv> headers;
 244   // these need to live until after the request headers have been processed
 245   std::string remote;
 246   std::string remotePort;
 247   headers.reserve(8 + (addXForwarded ? 3 : 0));
 248
 249   /* Pseudo-headers need to come first (rfc7540 8.1.2.1) */
 250   NGHTTP2Headers::addStaticHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::METHOD_NAME, NGHTTP2Headers::HeaderConstantIndexes::METHOD_VALUE);
 251   NGHTTP2Headers::addStaticHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::SCHEME_NAME, NGHTTP2Headers::HeaderConstantIndexes::SCHEME_VALUE);
 252   NGHTTP2Headers::addDynamicHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::AUTHORITY_NAME, d_ds->d_config.d_tlsSubjectName);
 253   NGHTTP2Headers::addDynamicHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::PATH_NAME, d_ds->d_config.d_dohPath);
 254   NGHTTP2Headers::addStaticHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::ACCEPT_NAME, NGHTTP2Headers::HeaderConstantIndexes::ACCEPT_VALUE);
 255   NGHTTP2Headers::addStaticHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::CONTENT_TYPE_NAME, NGHTTP2Headers::HeaderConstantIndexes::CONTENT_TYPE_VALUE);
 256   NGHTTP2Headers::addStaticHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::USER_AGENT_NAME, NGHTTP2Headers::HeaderConstantIndexes::USER_AGENT_VALUE);
 257   NGHTTP2Headers::addDynamicHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::CONTENT_LENGTH_NAME, payloadSize);
 258   /* no need to add these headers for health-check queries */
 259   if (addXForwarded && query.d_idstate.origRemote.getPort() != 0) {
 260     remote = query.d_idstate.origRemote.toString();
 261     remotePort = std::to_string(query.d_idstate.origRemote.getPort());
 262     NGHTTP2Headers::addDynamicHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_FOR_NAME, remote);
 263     NGHTTP2Headers::addDynamicHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PORT_NAME, remotePort);
 264     if (query.d_idstate.cs != nullptr) {
 265       if (query.d_idstate.cs->isUDP()) {
 266         NGHTTP2Headers::addStaticHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PROTO_NAME, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PROTO_VALUE_DNS_OVER_UDP);
 267       }
 268       else if (query.d_idstate.cs->isDoH()) {
 269         if (query.d_idstate.cs->hasTLS()) {
 270           NGHTTP2Headers::addStaticHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PROTO_NAME, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PROTO_VALUE_DNS_OVER_HTTPS);
 271         }
 272         else {
 273           NGHTTP2Headers::addStaticHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PROTO_NAME, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PROTO_VALUE_DNS_OVER_HTTP);
 274         }
 275       }
 276       else if (query.d_idstate.cs->hasTLS()) {
 277         NGHTTP2Headers::addStaticHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PROTO_NAME, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PROTO_VALUE_DNS_OVER_TLS);
 278       }
 279       else {
 280         NGHTTP2Headers::addStaticHeader(headers, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PROTO_NAME, NGHTTP2Headers::HeaderConstantIndexes::X_FORWARDED_PROTO_VALUE_DNS_OVER_TCP);
 281       }
 282     }
 283   }
 284
 285   PendingRequest pending;
 286   pending.d_query = std::move(query);
 287   pending.d_sender = std::move(sender);
 288
 289   uint32_t streamId = nghttp2_session_get_next_stream_id(d_session.get());
 290   auto insertPair = d_currentStreams.insert({streamId, std::move(pending)});
 291   if (!insertPair.second) {
 292     /* there is a stream ID collision, something is very wrong! */
 293     d_connectionDied = true;
 294     nghttp2_session_terminate_session(d_session.get(), NGHTTP2_NO_ERROR);
 295     throw std::runtime_error("Stream ID collision");
 296   }
 297
 298   /* if data_prd is not NULL, it provides data which will be sent in subsequent DATA frames. In this case, a method that allows request message bodies (https://tools.ietf.org/html/rfc7231#section-4) must be specified with :method key (e.g. POST). This function does not take ownership of the data_prd. The function copies the members of the data_prd. If data_prd is NULL, HEADERS have END_STREAM set.
 299    */
 300   nghttp2_data_provider data_provider;
 301
 302   data_provider.source.ptr = this;
 303   data_provider.read_callback = [](nghttp2_session* session, int32_t stream_id, uint8_t* buf, size_t length, uint32_t* data_flags, nghttp2_data_source* source, void* user_data) -> ssize_t {
 304     auto* conn = static_cast<DoHConnectionToBackend*>(user_data);
 305     auto& request = conn->d_currentStreams.at(stream_id);
 306     size_t toCopy = 0;
 307     if (request.d_queryPos < request.d_query.d_buffer.size()) {
 308       size_t remaining = request.d_query.d_buffer.size() - request.d_queryPos;
 309       toCopy = length > remaining ? remaining : length;
 310       memcpy(buf, &request.d_query.d_buffer.at(request.d_queryPos), toCopy);
 311       request.d_queryPos += toCopy;
 312     }
 313
 314     if (request.d_queryPos >= request.d_query.d_buffer.size()) {
 315       *data_flags |= NGHTTP2_DATA_FLAG_EOF;
 316     }
 317     return toCopy;
 318   };
 319
 320   auto newStreamId = nghttp2_submit_request(d_session.get(), nullptr, headers.data(), headers.size(), &data_provider, this);
 321   if (newStreamId < 0) {
 322     d_connectionDied = true;
 323     ++d_ds->tcpDiedSendingQuery;
 324     d_currentStreams.erase(streamId);
 325     throw std::runtime_error("Error submitting HTTP request:" + std::string(nghttp2_strerror(newStreamId)));
 326   }
 327
 328   auto rv = nghttp2_session_send(d_session.get());
 329   if (rv != 0) {
 330     d_connectionDied = true;
 331     ++d_ds->tcpDiedSendingQuery;
 332     d_currentStreams.erase(streamId);
 333     throw std::runtime_error("Error in nghttp2_session_send:" + std::to_string(rv));
 334   }
 335
 336   d_highestStreamID = newStreamId;
 337 }
 338
 339 class DoHClientThreadData
 340 {
 341 public:
 342   DoHClientThreadData(pdns::channel::Receiver<CrossProtocolQuery>&& receiver) :
 343     mplexer(std::unique_ptr<FDMultiplexer>(FDMultiplexer::getMultiplexerSilent())),
 344     d_receiver(std::move(receiver))
 345   {
 346   }
 347
 348   std::unique_ptr<FDMultiplexer> mplexer{nullptr};
 349   pdns::channel::Receiver<CrossProtocolQuery> d_receiver;
 350 };
 351
 352 void DoHConnectionToBackend::handleReadableIOCallback(int fd, FDMultiplexer::funcparam_t& param)
 353 {
 354   auto conn = boost::any_cast<std::shared_ptr<DoHConnectionToBackend>>(param);
 355   if (fd != conn->getHandle()) {
 356     throw std::runtime_error("Unexpected socket descriptor " + std::to_string(fd) + " received in " + std::string(__PRETTY_FUNCTION__) + ", expected " + std::to_string(conn->getHandle()));
 357   }
 358
 359   IOStateGuard ioGuard(conn->d_ioState);
 360   do {
 361     conn->d_inPos = 0;
 362     conn->d_in.resize(conn->d_in.size() + 512);
 363     // cerr<<"trying to read "<<conn->d_in.size()<<endl;
 364     try {
 365       IOState newState = conn->d_handler->tryRead(conn->d_in, conn->d_inPos, conn->d_in.size(), true);
 366       // cerr<<"got a "<<(int)newState<<" state and "<<conn->d_inPos<<" bytes"<<endl;
 367       conn->d_in.resize(conn->d_inPos);
 368
 369       if (conn->d_inPos > 0) {
 370         /* we got something */
 371         auto readlen = nghttp2_session_mem_recv(conn->d_session.get(), conn->d_in.data(), conn->d_inPos);
 372         // cerr<<"nghttp2_session_mem_recv returned "<<readlen<<endl;
 373         /* as long as we don't require a pause by returning nghttp2_error.NGHTTP2_ERR_PAUSE from a CB,
 374            all data should be consumed before returning */
 375         if (readlen > 0 && static_cast<size_t>(readlen) < conn->d_inPos) {
 376           throw std::runtime_error("Fatal error while passing received data to nghttp2: " + std::string(nghttp2_strerror((int)readlen)));
 377         }
 378
 379         struct timeval now
 380         {
 381           .tv_sec = 0, .tv_usec = 0
 382         };
 383
 384         gettimeofday(&now, nullptr);
 385         conn->d_lastDataReceivedTime = now;
 386
 387         // cerr<<"after read send"<<endl;
 388         nghttp2_session_send(conn->d_session.get());
 389       }
 390
 391       if (newState == IOState::Done) {
 392         if (conn->isIdle()) {
 393           conn->stopIO();
 394           conn->watchForRemoteHostClosingConnection();
 395           ioGuard.release();
 396           break;
 397         }
 398       }
 399       else {
 400         if (newState == IOState::NeedWrite) {
 401           // cerr<<"need write"<<endl;
 402           conn->updateIO(IOState::NeedWrite, handleReadableIOCallback);
 403         }
 404         ioGuard.release();
 405         break;
 406       }
 407     }
 408     catch (const std::exception& e) {
 409       vinfolog("Exception while trying to read from HTTP backend connection: %s", e.what());
 410       ++conn->d_ds->tcpDiedReadingResponse;
 411       conn->handleIOError();
 412       break;
 413     }
 414   } while (conn->getConcurrentStreamsCount() > 0);
 415 }
 416
 417 void DoHConnectionToBackend::handleWritableIOCallback(int fd, FDMultiplexer::funcparam_t& param)
 418 {
 419   auto conn = boost::any_cast<std::shared_ptr<DoHConnectionToBackend>>(param);
 420   if (fd != conn->getHandle()) {
 421     throw std::runtime_error("Unexpected socket descriptor " + std::to_string(fd) + " received in " + std::string(__PRETTY_FUNCTION__) + ", expected " + std::to_string(conn->getHandle()));
 422   }
 423   IOStateGuard ioGuard(conn->d_ioState);
 424
 425   // cerr<<"in "<<__PRETTY_FUNCTION__<<" trying to write "<<conn->d_out.size()-conn->d_outPos<<endl;
 426   try {
 427     IOState newState = conn->d_handler->tryWrite(conn->d_out, conn->d_outPos, conn->d_out.size());
 428     // cerr<<"got a "<<(int)newState<<" state, "<<conn->d_out.size()-conn->d_outPos<<" bytes remaining"<<endl;
 429     if (newState == IOState::NeedRead) {
 430       conn->updateIO(IOState::NeedRead, handleWritableIOCallback);
 431     }
 432     else if (newState == IOState::Done) {
 433       // cerr<<"done, buffer size was "<<conn->d_out.size()<<", pos was "<<conn->d_outPos<<endl;
 434       conn->d_firstWrite = false;
 435       conn->d_out.clear();
 436       conn->d_outPos = 0;
 437       conn->stopIO();
 438       if (!conn->isIdle()) {
 439         conn->updateIO(IOState::NeedRead, handleReadableIOCallback);
 440       }
 441       else {
 442         conn->watchForRemoteHostClosingConnection();
 443       }
 444     }
 445     ioGuard.release();
 446   }
 447   catch (const std::exception& e) {
 448     vinfolog("Exception while trying to write (ready) to HTTP backend connection: %s", e.what());
 449     ++conn->d_ds->tcpDiedSendingQuery;
 450     conn->handleIOError();
 451   }
 452 }
 453
 454 void DoHConnectionToBackend::stopIO()
 455 {
 456   d_ioState->reset();
 457
 458   if (isIdle()) {
 459     auto shared = std::dynamic_pointer_cast<DoHConnectionToBackend>(shared_from_this());
 460     if (!willBeReusable(false)) {
 461       /* remove ourselves from the connection cache, this might mean that our
 462          reference count drops to zero after that, so we need to be careful */
 463       t_downstreamDoHConnectionsManager.removeDownstreamConnection(shared);
 464     }
 465     else {
 466       t_downstreamDoHConnectionsManager.moveToIdle(shared);
 467     }
 468   }
 469 }
 470
 471 void DoHConnectionToBackend::updateIO(IOState newState, FDMultiplexer::callbackfunc_t callback, bool noTTD)
 472 {
 473   struct timeval now
 474   {
 475     .tv_sec = 0, .tv_usec = 0
 476   };
 477
 478   gettimeofday(&now, nullptr);
 479   boost::optional<struct timeval> ttd{boost::none};
 480   if (!noTTD) {
 481     if (d_healthCheckQuery) {
 482       ttd = getBackendHealthCheckTTD(now);
 483     }
 484     else if (newState == IOState::NeedRead) {
 485       ttd = getBackendReadTTD(now);
 486     }
 487     else if (isFresh() && d_firstWrite) {
 488       /* first write just after the non-blocking connect */
 489       ttd = getBackendConnectTTD(now);
 490     }
 491     else {
 492       ttd = getBackendWriteTTD(now);
 493     }
 494   }
 495
 496   auto shared = std::dynamic_pointer_cast<DoHConnectionToBackend>(shared_from_this());
 497   if (shared) {
 498     if (newState == IOState::NeedRead) {
 499       d_ioState->update(newState, callback, shared, ttd);
 500     }
 501     else if (newState == IOState::NeedWrite) {
 502       d_ioState->update(newState, callback, shared, ttd);
 503     }
 504   }
 505 }
 506
 507 void DoHConnectionToBackend::watchForRemoteHostClosingConnection()
 508 {
 509   if (willBeReusable(false) && !d_healthCheckQuery) {
 510     updateIO(IOState::NeedRead, handleReadableIOCallback, false);
 511   }
 512 }
 513
 514 ssize_t DoHConnectionToBackend::send_callback(nghttp2_session* session, const uint8_t* data, size_t length, int flags, void* user_data)
 515 {
 516   DoHConnectionToBackend* conn = reinterpret_cast<DoHConnectionToBackend*>(user_data);
 517   bool bufferWasEmpty = conn->d_out.empty();
 518   if (!conn->d_proxyProtocolPayloadSent && !conn->d_proxyProtocolPayload.empty()) {
 519     conn->d_out.insert(conn->d_out.end(), conn->d_proxyProtocolPayload.begin(), conn->d_proxyProtocolPayload.end());
 520     conn->d_proxyProtocolPayloadSent = true;
 521   }
 522
 523   conn->d_out.insert(conn->d_out.end(), data, data + length);
 524
 525   if (bufferWasEmpty) {
 526     try {
 527       // cerr<<"in "<<__PRETTY_FUNCTION__<<" trying to write "<<conn->d_out.size()-conn->d_outPos<<endl;
 528       auto state = conn->d_handler->tryWrite(conn->d_out, conn->d_outPos, conn->d_out.size());
 529       // cerr<<"got a "<<(int)state<<" state, "<<conn->d_out.size()-conn->d_outPos<<" bytes remaining"<<endl;
 530       if (state == IOState::Done) {
 531         conn->d_firstWrite = false;
 532         conn->d_out.clear();
 533         conn->d_outPos = 0;
 534         conn->stopIO();
 535         if (!conn->isIdle()) {
 536           conn->updateIO(IOState::NeedRead, handleReadableIOCallback);
 537         }
 538         else {
 539           conn->watchForRemoteHostClosingConnection();
 540         }
 541       }
 542       else {
 543         conn->updateIO(state, handleWritableIOCallback);
 544       }
 545     }
 546     catch (const std::exception& e) {
 547       vinfolog("Exception while trying to write (send) to HTTP backend connection: %s", e.what());
 548       conn->handleIOError();
 549       ++conn->d_ds->tcpDiedSendingQuery;
 550     }
 551   }
 552
 553   return length;
 554 }
 555
 556 int DoHConnectionToBackend::on_frame_recv_callback(nghttp2_session* session, const nghttp2_frame* frame, void* user_data)
 557 {
 558   DoHConnectionToBackend* conn = reinterpret_cast<DoHConnectionToBackend*>(user_data);
 559   // cerr<<"Frame type is "<<std::to_string(frame->hd.type)<<endl;
 560 #if 0
 561   switch (frame->hd.type) {
 562   case NGHTTP2_HEADERS:
 563     cerr<<"got headers"<<endl;
 564     if (frame->headers.cat == NGHTTP2_HCAT_RESPONSE) {
 565       cerr<<"All headers received"<<endl;
 566     }
 567     break;
 568   case NGHTTP2_WINDOW_UPDATE:
 569     cerr<<"got window update"<<endl;
 570     break;
 571   case NGHTTP2_SETTINGS:
 572     cerr<<"got settings"<<endl;
 573     cerr<<frame->settings.niv<<endl;
 574     for (size_t idx = 0; idx < frame->settings.niv; idx++) {
 575       cerr<<"- "<<frame->settings.iv[idx].settings_id<<" "<<frame->settings.iv[idx].value<<endl;
 576     }
 577     break;
 578   case NGHTTP2_DATA:
 579     cerr<<"got data"<<endl;
 580     break;
 581   }
 582 #endif
 583
 584   if (frame->hd.type == NGHTTP2_GOAWAY) {
 585     conn->d_connectionDied = true;
 586   }
 587
 588   /* is this the last frame for this stream? */
 589   else if ((frame->hd.type == NGHTTP2_HEADERS || frame->hd.type == NGHTTP2_DATA) && frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
 590     auto stream = conn->d_currentStreams.find(frame->hd.stream_id);
 591     if (stream != conn->d_currentStreams.end()) {
 592       // cerr<<"Stream "<<frame->hd.stream_id<<" is now finished"<<endl;
 593       stream->second.d_finished = true;
 594       ++conn->d_queries;
 595
 596       auto request = std::move(stream->second);
 597       conn->d_currentStreams.erase(stream->first);
 598       if (request.d_responseCode == 200U) {
 599         conn->handleResponse(std::move(request));
 600       }
 601       else {
 602         vinfolog("HTTP response has a non-200 status code: %d", request.d_responseCode);
 603         struct timeval now
 604         {
 605           .tv_sec = 0, .tv_usec = 0
 606         };
 607
 608         gettimeofday(&now, nullptr);
 609
 610         conn->handleResponseError(std::move(request), now);
 611       }
 612
 613       if (conn->isIdle()) {
 614         conn->stopIO();
 615         conn->watchForRemoteHostClosingConnection();
 616       }
 617     }
 618     else {
 619       vinfolog("Stream %d NOT FOUND", frame->hd.stream_id);
 620       conn->d_connectionDied = true;
 621       ++conn->d_ds->tcpDiedReadingResponse;
 622       return NGHTTP2_ERR_CALLBACK_FAILURE;
 623     }
 624   }
 625
 626   return 0;
 627 }
 628
 629 int DoHConnectionToBackend::on_data_chunk_recv_callback(nghttp2_session* session, uint8_t flags, int32_t stream_id, const uint8_t* data, size_t len, void* user_data)
 630 {
 631   DoHConnectionToBackend* conn = reinterpret_cast<DoHConnectionToBackend*>(user_data);
 632   // cerr<<"Got data of size "<<len<<" for stream "<<stream_id<<endl;
 633   auto stream = conn->d_currentStreams.find(stream_id);
 634   if (stream == conn->d_currentStreams.end()) {
 635     vinfolog("Unable to match the stream ID %d to a known one!", stream_id);
 636     conn->d_connectionDied = true;
 637     ++conn->d_ds->tcpDiedReadingResponse;
 638     return NGHTTP2_ERR_CALLBACK_FAILURE;
 639   }
 640   if (len > std::numeric_limits<uint16_t>::max() || (std::numeric_limits<uint16_t>::max() - stream->second.d_buffer.size()) < len) {
 641     vinfolog("Data frame of size %d is too large for a DNS response (we already have %d)", len, stream->second.d_buffer.size());
 642     conn->d_connectionDied = true;
 643     ++conn->d_ds->tcpDiedReadingResponse;
 644     return NGHTTP2_ERR_CALLBACK_FAILURE;
 645   }
 646
 647   stream->second.d_buffer.insert(stream->second.d_buffer.end(), data, data + len);
 648   if (stream->second.d_finished) {
 649     // cerr<<"we now have the full response!"<<endl;
 650     // cerr<<std::string(reinterpret_cast<const char*>(data), len)<<endl;
 651
 652     auto request = std::move(stream->second);
 653     conn->d_currentStreams.erase(stream->first);
 654     if (request.d_responseCode == 200U) {
 655       conn->handleResponse(std::move(request));
 656     }
 657     else {
 658       vinfolog("HTTP response has a non-200 status code: %d", request.d_responseCode);
 659       struct timeval now
 660       {
 661         .tv_sec = 0, .tv_usec = 0
 662       };
 663
 664       gettimeofday(&now, nullptr);
 665
 666       conn->handleResponseError(std::move(request), now);
 667     }
 668     if (conn->isIdle()) {
 669       conn->stopIO();
 670       conn->watchForRemoteHostClosingConnection();
 671     }
 672   }
 673
 674   return 0;
 675 }
 676
 677 int DoHConnectionToBackend::on_stream_close_callback(nghttp2_session* session, int32_t stream_id, uint32_t error_code, void* user_data)
 678 {
 679   DoHConnectionToBackend* conn = reinterpret_cast<DoHConnectionToBackend*>(user_data);
 680
 681   if (error_code == 0) {
 682     return 0;
 683   }
 684
 685   // cerr << "Stream " << stream_id << " closed with error_code=" << error_code << endl;
 686   conn->d_connectionDied = true;
 687   ++conn->d_ds->tcpDiedReadingResponse;
 688
 689   auto stream = conn->d_currentStreams.find(stream_id);
 690   if (stream == conn->d_currentStreams.end()) {
 691     /* we don't care, then */
 692     return 0;
 693   }
 694
 695   struct timeval now
 696   {
 697     .tv_sec = 0, .tv_usec = 0
 698   };
 699
 700   gettimeofday(&now, nullptr);
 701   auto request = std::move(stream->second);
 702   conn->d_currentStreams.erase(stream->first);
 703
 704   // cerr<<"Query has "<<request.d_query.d_downstreamFailures<<" failures, backend limit is "<<conn->d_ds->d_retries<<endl;
 705   if (request.d_query.d_downstreamFailures < conn->d_ds->d_config.d_retries) {
 706     // cerr<<"in "<<__PRETTY_FUNCTION__<<", looking for a connection to send a query of size "<<request.d_query.d_buffer.size()<<endl;
 707     ++request.d_query.d_downstreamFailures;
 708     auto downstream = t_downstreamDoHConnectionsManager.getConnectionToDownstream(conn->d_mplexer, conn->d_ds, now, std::string(conn->d_proxyProtocolPayload));
 709     downstream->queueQuery(request.d_sender, std::move(request.d_query));
 710   }
 711   else {
 712     conn->handleResponseError(std::move(request), now);
 713   }
 714
 715   // cerr<<"we now have "<<conn->getConcurrentStreamsCount()<<" concurrent connections"<<endl;
 716   if (conn->isIdle()) {
 717     // cerr<<"stopping IO"<<endl;
 718     conn->stopIO();
 719     conn->watchForRemoteHostClosingConnection();
 720   }
 721
 722   return 0;
 723 }
 724
 725 int DoHConnectionToBackend::on_header_callback(nghttp2_session* session, const nghttp2_frame* frame, const uint8_t* name, size_t namelen, const uint8_t* value, size_t valuelen, uint8_t flags, void* user_data)
 726 {
 727   DoHConnectionToBackend* conn = reinterpret_cast<DoHConnectionToBackend*>(user_data);
 728
 729   const std::string status(":status");
 730   if (frame->hd.type == NGHTTP2_HEADERS && frame->headers.cat == NGHTTP2_HCAT_RESPONSE) {
 731     // cerr<<"got header for "<<frame->hd.stream_id<<":"<<endl;
 732     // cerr<<"- "<<std::string(reinterpret_cast<const char*>(name), namelen)<<endl;
 733     // cerr<<"- "<<std::string(reinterpret_cast<const char*>(value), valuelen)<<endl;
 734     if (namelen == status.size() && memcmp(status.data(), name, status.size()) == 0) {
 735       auto stream = conn->d_currentStreams.find(frame->hd.stream_id);
 736       if (stream == conn->d_currentStreams.end()) {
 737         vinfolog("Unable to match the stream ID %d to a known one!", frame->hd.stream_id);
 738         conn->d_connectionDied = true;
 739         return NGHTTP2_ERR_CALLBACK_FAILURE;
 740       }
 741       try {
 742         pdns::checked_stoi_into(stream->second.d_responseCode, std::string(reinterpret_cast<const char*>(value), valuelen));
 743       }
 744       catch (...) {
 745         vinfolog("Error parsing the status header for stream ID %d", frame->hd.stream_id);
 746         conn->d_connectionDied = true;
 747         ++conn->d_ds->tcpDiedReadingResponse;
 748         return NGHTTP2_ERR_CALLBACK_FAILURE;
 749       }
 750     }
 751   }
 752   return 0;
 753 }
 754
 755 int DoHConnectionToBackend::on_error_callback(nghttp2_session* session, int lib_error_code, const char* msg, size_t len, void* user_data)
 756 {
 757   vinfolog("Error in HTTP/2 connection: %s", std::string(msg, len));
 758
 759   DoHConnectionToBackend* conn = reinterpret_cast<DoHConnectionToBackend*>(user_data);
 760   conn->d_connectionDied = true;
 761   ++conn->d_ds->tcpDiedReadingResponse;
 762
 763   return 0;
 764 }
 765
 766 DoHConnectionToBackend::DoHConnectionToBackend(const std::shared_ptr<DownstreamState>& ds, std::unique_ptr<FDMultiplexer>& mplexer, const struct timeval& now, std::string&& proxyProtocolPayload) :
 767   ConnectionToBackend(ds, mplexer, now), d_proxyProtocolPayload(std::move(proxyProtocolPayload))
 768 {
 769   // inherit most of the stuff from the ConnectionToBackend()
 770   d_ioState = make_unique<IOStateHandler>(*d_mplexer, d_handler->getDescriptor());
 771
 772   nghttp2_session_callbacks* cbs = nullptr;
 773   if (nghttp2_session_callbacks_new(&cbs) != 0) {
 774     d_connectionDied = true;
 775     ++d_ds->tcpDiedSendingQuery;
 776     vinfolog("Unable to create a callback object for a new HTTP/2 session");
 777     return;
 778   }
 779   std::unique_ptr<nghttp2_session_callbacks, void (*)(nghttp2_session_callbacks*)> callbacks(cbs, nghttp2_session_callbacks_del);
 780   cbs = nullptr;
 781
 782   nghttp2_session_callbacks_set_send_callback(callbacks.get(), send_callback);
 783   nghttp2_session_callbacks_set_on_frame_recv_callback(callbacks.get(), on_frame_recv_callback);
 784   nghttp2_session_callbacks_set_on_data_chunk_recv_callback(callbacks.get(), on_data_chunk_recv_callback);
 785   nghttp2_session_callbacks_set_on_stream_close_callback(callbacks.get(), on_stream_close_callback);
 786   nghttp2_session_callbacks_set_on_header_callback(callbacks.get(), on_header_callback);
 787   nghttp2_session_callbacks_set_error_callback2(callbacks.get(), on_error_callback);
 788
 789   nghttp2_session* sess = nullptr;
 790   if (nghttp2_session_client_new(&sess, callbacks.get(), this) != 0) {
 791     d_connectionDied = true;
 792     ++d_ds->tcpDiedSendingQuery;
 793     vinfolog("Coult not allocate a new HTTP/2 session");
 794     return;
 795   }
 796
 797   d_session = std::unique_ptr<nghttp2_session, void (*)(nghttp2_session*)>(sess, nghttp2_session_del);
 798   sess = nullptr;
 799
 800   callbacks.reset();
 801
 802   nghttp2_settings_entry iv[] = {
 803     /* rfc7540 section-8.2.2:
 804        "Advertising a SETTINGS_MAX_CONCURRENT_STREAMS value of zero disables
 805        server push by preventing the server from creating the necessary
 806        streams."
 807     */
 808     {NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, 0},
 809     {NGHTTP2_SETTINGS_ENABLE_PUSH, 0},
 810     /* we might want to make the initial window size configurable, but 16M is a large enough default */
 811     {NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, 16 * 1024 * 1024}};
 812   /* client 24 bytes magic string will be sent by nghttp2 library */
 813   int rv = nghttp2_submit_settings(d_session.get(), NGHTTP2_FLAG_NONE, iv, sizeof(iv) / sizeof(*iv));
 814   if (rv != 0) {
 815     d_connectionDied = true;
 816     ++d_ds->tcpDiedSendingQuery;
 817     vinfolog("Could not submit SETTINGS: %s", nghttp2_strerror(rv));
 818     return;
 819   }
 820 }
 821
 822 static void handleCrossProtocolQuery(int pipefd, FDMultiplexer::funcparam_t& param)
 823 {
 824   auto threadData = boost::any_cast<DoHClientThreadData*>(param);
 825
 826   std::unique_ptr<CrossProtocolQuery> cpq{nullptr};
 827   try {
 828     auto tmp = threadData->d_receiver.receive();
 829     if (!tmp) {
 830       return;
 831     }
 832     cpq = std::move(*tmp);
 833   }
 834   catch (const std::exception& e) {
 835     throw std::runtime_error("Error while reading from the DoH cross-protocol channel:" + std::string(e.what()));
 836   }
 837
 838   struct timeval now
 839   {
 840     .tv_sec = 0, .tv_usec = 0
 841   };
 842   gettimeofday(&now, nullptr);
 843
 844   std::shared_ptr<TCPQuerySender> tqs = cpq->getTCPQuerySender();
 845   auto query = std::move(cpq->query);
 846   auto downstreamServer = std::move(cpq->downstream);
 847   cpq.reset();
 848
 849   try {
 850     auto downstream = t_downstreamDoHConnectionsManager.getConnectionToDownstream(threadData->mplexer, downstreamServer, now, std::move(query.d_proxyProtocolPayload));
 851     downstream->queueQuery(tqs, std::move(query));
 852   }
 853   catch (...) {
 854     TCPResponse response(std::move(query));
 855     tqs->notifyIOError(now, std::move(response));
 856   }
 857 }
 858
 859 static void dohClientThread(pdns::channel::Receiver<CrossProtocolQuery>&& receiver)
 860 {
 861   setThreadName("dnsdist/dohClie");
 862
 863   try {
 864     DoHClientThreadData data(std::move(receiver));
 865     data.mplexer->addReadFD(data.d_receiver.getDescriptor(), handleCrossProtocolQuery, &data);
 866
 867     struct timeval now
 868     {
 869       .tv_sec = 0, .tv_usec = 0
 870     };
 871
 872     gettimeofday(&now, nullptr);
 873     time_t lastTimeoutScan = now.tv_sec;
 874
 875     for (;;) {
 876       data.mplexer->run(&now, 1000);
 877
 878       if (now.tv_sec > lastTimeoutScan) {
 879         lastTimeoutScan = now.tv_sec;
 880
 881         try {
 882           t_downstreamDoHConnectionsManager.cleanupClosedConnections(now);
 883           handleH2Timeouts(*data.mplexer, now);
 884
 885           if (g_dohStatesDumpRequested > 0) {
 886             /* just to keep things clean in the output, debug only */
 887             static std::mutex s_lock;
 888             std::lock_guard<decltype(s_lock)> lck(s_lock);
 889             if (g_dohStatesDumpRequested > 0) {
 890               /* no race here, we took the lock so it can only be increased in the meantime */
 891               --g_dohStatesDumpRequested;
 892               infolog("Dumping the DoH client states, as requested:");
 893               data.mplexer->runForAllWatchedFDs([](bool isRead, int fd, const FDMultiplexer::funcparam_t& param, struct timeval ttd) {
 894                 struct timeval lnow;
 895                 gettimeofday(&lnow, nullptr);
 896                 if (ttd.tv_sec > 0) {
 897                   infolog("- Descriptor %d is in %s state, TTD in %d", fd, (isRead ? "read" : "write"), (ttd.tv_sec - lnow.tv_sec));
 898                 }
 899                 else {
 900                   infolog("- Descriptor %d is in %s state, no TTD set", fd, (isRead ? "read" : "write"));
 901                 }
 902
 903                 if (param.type() == typeid(std::shared_ptr<DoHConnectionToBackend>)) {
 904                   auto conn = boost::any_cast<std::shared_ptr<DoHConnectionToBackend>>(param);
 905                   infolog(" - %s", conn->toString());
 906                 }
 907                 else if (param.type() == typeid(DoHClientThreadData*)) {
 908                   infolog(" - Worker thread pipe");
 909                 }
 910               });
 911               infolog("The DoH client cache has %d active and %d idle outgoing connections cached", t_downstreamDoHConnectionsManager.getActiveCount(), t_downstreamDoHConnectionsManager.getIdleCount());
 912             }
 913           }
 914         }
 915         catch (const std::exception& e) {
 916           warnlog("Error in outgoing DoH thread: %s", e.what());
 917         }
 918       }
 919     }
 920   }
 921   catch (const std::exception& e) {
 922     errlog("Fatal error in outgoing DoH thread: %s", e.what());
 923   }
 924 }
 925
 926 static bool select_next_proto_callback(unsigned char** out, unsigned char* outlen, const unsigned char* in, unsigned int inlen)
 927 {
 928   if (nghttp2_select_next_protocol(out, outlen, in, inlen) <= 0) {
 929     vinfolog("The remote DoH backend did not advertise " NGHTTP2_PROTO_VERSION_ID);
 930     return false;
 931   }
 932   return true;
 933 }
 934
 935 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
 936
 937 struct DoHClientCollection::DoHWorkerThread
 938 {
 939   DoHWorkerThread()
 940   {
 941   }
 942
 943   DoHWorkerThread(pdns::channel::Sender<CrossProtocolQuery>&& sender) :
 944     d_sender(std::move(sender))
 945   {
 946   }
 947
 948   DoHWorkerThread(DoHWorkerThread&& rhs) :
 949     d_sender(std::move(rhs.d_sender))
 950   {
 951   }
 952
 953   DoHWorkerThread& operator=(DoHWorkerThread&& rhs)
 954   {
 955     d_sender = std::move(rhs.d_sender);
 956     return *this;
 957   }
 958
 959   DoHWorkerThread(const DoHWorkerThread& rhs) = delete;
 960   DoHWorkerThread& operator=(const DoHWorkerThread&) = delete;
 961
 962   pdns::channel::Sender<CrossProtocolQuery> d_sender;
 963 };
 964
 965 DoHClientCollection::DoHClientCollection(size_t numberOfThreads) :
 966   d_clientThreads(numberOfThreads)
 967 {
 968 }
 969
 970 bool DoHClientCollection::passCrossProtocolQueryToThread(std::unique_ptr<CrossProtocolQuery>&& cpq)
 971 {
 972   if (d_numberOfThreads == 0) {
 973     throw std::runtime_error("No DoH worker thread yet");
 974   }
 975
 976   uint64_t pos = d_pos++;
 977   if (!d_clientThreads.at(pos % d_numberOfThreads).d_sender.send(std::move(cpq))) {
 978     ++dnsdist::metrics::g_stats.outgoingDoHQueryPipeFull;
 979     return false;
 980   }
 981
 982   return true;
 983 }
 984
 985 void DoHClientCollection::addThread()
 986 {
 987 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
 988   try {
 989     auto [sender, receiver] = pdns::channel::createObjectQueue<CrossProtocolQuery>(pdns::channel::SenderBlockingMode::SenderNonBlocking, pdns::channel::ReceiverBlockingMode::ReceiverNonBlocking, g_tcpInternalPipeBufferSize);
 990
 991     vinfolog("Adding DoH Client thread");
 992     std::lock_guard<std::mutex> lock(d_mutex);
 993
 994     if (d_numberOfThreads >= d_clientThreads.size()) {
 995       vinfolog("Adding a new DoH client thread would exceed the vector size (%d/%d), skipping. Consider increasing the maximum amount of DoH client threads with setMaxDoHClientThreads() in the configuration.", d_numberOfThreads, d_clientThreads.size());
 996       return;
 997     }
 998
 999     DoHWorkerThread worker(std::move(sender));
1000     try {
1001       std::thread t1(dohClientThread, std::move(receiver));
1002       t1.detach();
1003     }
1004     catch (const std::runtime_error& e) {
1005       /* the thread creation failed */
1006       errlog("Error creating a DoH thread: %s", e.what());
1007       return;
1008     }
1009
1010     d_clientThreads.at(d_numberOfThreads) = std::move(worker);
1011     ++d_numberOfThreads;
1012   }
1013   catch (const std::exception& e) {
1014     errlog("Error creating the DoH channel: %s", e.what());
1015     return;
1016   }
1017 #else /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1018   throw std::runtime_error("DoHClientCollection::addThread() called but nghttp2 support is not available");
1019 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1020 }
1021
1022 bool initDoHWorkers()
1023 {
1024 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
1025   if (!g_outgoingDoHWorkerThreads) {
1026     /* Unless the value has been set to 0 explicitly, always start at least one outgoing DoH worker thread, in case a DoH backend
1027        is added at a later time. */
1028     g_outgoingDoHWorkerThreads = 1;
1029   }
1030
1031   if (g_outgoingDoHWorkerThreads && *g_outgoingDoHWorkerThreads > 0) {
1032     g_dohClientThreads = std::make_unique<DoHClientCollection>(*g_outgoingDoHWorkerThreads);
1033     for (size_t idx = 0; idx < *g_outgoingDoHWorkerThreads; idx++) {
1034       g_dohClientThreads->addThread();
1035     }
1036   }
1037   return true;
1038 #else
1039   return false;
1040 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1041 }
1042
1043 bool setupDoHClientProtocolNegotiation(std::shared_ptr<TLSCtx>& ctx)
1044 {
1045   if (ctx == nullptr) {
1046     return false;
1047   }
1048 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
1049   /* we want to set the ALPN to h2, if only to mitigate the ALPACA attack */
1050   const std::vector<std::vector<uint8_t>> h2Alpns = {{'h', '2'}};
1051   ctx->setALPNProtos(h2Alpns);
1052   ctx->setNextProtocolSelectCallback(select_next_proto_callback);
1053   return true;
1054 #else /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1055   return false;
1056 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1057 }
1058
1059 bool sendH2Query(const std::shared_ptr<DownstreamState>& ds, std::unique_ptr<FDMultiplexer>& mplexer, std::shared_ptr<TCPQuerySender>& sender, InternalQuery&& query, bool healthCheck)
1060 {
1061 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
1062   struct timeval now
1063   {
1064     .tv_sec = 0, .tv_usec = 0
1065   };
1066   gettimeofday(&now, nullptr);
1067
1068   if (healthCheck) {
1069     /* always do health-checks over a new connection */
1070     auto newConnection = std::make_shared<DoHConnectionToBackend>(ds, mplexer, now, std::move(query.d_proxyProtocolPayload));
1071     newConnection->setHealthCheck(healthCheck);
1072     newConnection->queueQuery(sender, std::move(query));
1073   }
1074   else {
1075     auto connection = t_downstreamDoHConnectionsManager.getConnectionToDownstream(mplexer, ds, now, std::move(query.d_proxyProtocolPayload));
1076     connection->queueQuery(sender, std::move(query));
1077   }
1078
1079   return true;
1080 #else /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1081   return false;
1082 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1083 }
1084
1085 size_t clearH2Connections()
1086 {
1087   size_t cleared = 0;
1088 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
1089   cleared = t_downstreamDoHConnectionsManager.clear();
1090 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1091   return cleared;
1092 }
1093
1094 size_t handleH2Timeouts(FDMultiplexer& mplexer, const struct timeval& now)
1095 {
1096   size_t got = 0;
1097 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
1098   auto expiredReadConns = mplexer.getTimeouts(now, false);
1099   for (const auto& cbData : expiredReadConns) {
1100     if (cbData.second.type() == typeid(std::shared_ptr<DoHConnectionToBackend>)) {
1101       auto conn = boost::any_cast<std::shared_ptr<DoHConnectionToBackend>>(cbData.second);
1102       vinfolog("Timeout (read) from remote DoH backend %s", conn->getBackendName());
1103       conn->handleTimeout(now, false);
1104       ++got;
1105     }
1106   }
1107
1108   auto expiredWriteConns = mplexer.getTimeouts(now, true);
1109   for (const auto& cbData : expiredWriteConns) {
1110     if (cbData.second.type() == typeid(std::shared_ptr<DoHConnectionToBackend>)) {
1111       auto conn = boost::any_cast<std::shared_ptr<DoHConnectionToBackend>>(cbData.second);
1112       vinfolog("Timeout (write) from remote DoH backend %s", conn->getBackendName());
1113       conn->handleTimeout(now, true);
1114       ++got;
1115     }
1116   }
1117 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1118   return got;
1119 }
1120
1121 void setDoHDownstreamCleanupInterval(uint16_t max)
1122 {
1123 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
1124   DownstreamDoHConnectionsManager::setCleanupInterval(max);
1125 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1126 }
1127
1128 void setDoHDownstreamMaxIdleTime(uint16_t max)
1129 {
1130 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
1131   DownstreamDoHConnectionsManager::setMaxIdleTime(max);
1132 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1133 }
1134
1135 void setDoHDownstreamMaxIdleConnectionsPerBackend(size_t max)
1136 {
1137 #if defined(HAVE_DNS_OVER_HTTPS) && defined(HAVE_NGHTTP2)
1138   DownstreamDoHConnectionsManager::setMaxIdleConnectionsPerDownstream(max);
1139 #endif /* HAVE_DNS_OVER_HTTPS && HAVE_NGHTTP2 */
1140 }