multi_transfer_runner.cc

Go to the documentation of this file.

/*
 * Copyright (C) Max Dymond, <cmeister2@gmail.com>, et al.
 *
 * SPDX-License-Identifier: curl
 */

 
#include "proto_fuzzer/multi_transfer_runner.h"
 
#define CURL_ALLOW_OLD_MULTI_SOCKET
#include <curl/curl.h>
#include <curl/multi.h>
 
#include <algorithm>
#include <array>
#include <cstddef>
#include <cstdio>
#include <memory>
#include <string>
 
#include "proto_fuzzer/curl_raii.h"
#include "proto_fuzzer/mock_server.h"
#include "proto_fuzzer/multi_socket_driver.h"
#include "proto_fuzzer/option_apply.h"
#include "proto_fuzzer/request_data.h"
#include "proto_fuzzer/scenario_limits.h"
#include "proto_fuzzer/transfer_session.h"
 
namespace proto_fuzzer {
 
namespace {

struct TransferState {
  TransferSession session;
  bool attached = false;
};
 
constexpr int kMaxDriveIterations = 512;
constexpr int kMaxIdleIterations = 8;
constexpr curl_off_t kStateProbeReceiveRate = 1;
constexpr long kStateProbeTimeoutMs = 20;
 
struct FileCloser {
  void operator()(FILE* file) const {
    if (file != nullptr) {
      std::fclose(file);
    }
  }
};
 
using FilePtr = std::unique_ptr<FILE, FileCloser>;

bool ApplyStateProbeOptions(CURL* easy, curl::fuzzer::proto::MultiStateProbeProfile profile, FILE* progress_sink) {
  if (profile != curl::fuzzer::proto::MULTI_STATE_PROBE_PROGRESS_RATE_LIMIT || progress_sink == nullptr) {
    return false;
  }
  return curl_easy_setopt(easy, CURLOPT_STDERR, progress_sink) == CURLE_OK &&
         curl_easy_setopt(easy, CURLOPT_TIMEOUT_MS, kStateProbeTimeoutMs) == CURLE_OK &&
         curl_easy_setopt(easy, CURLOPT_NOPROGRESS, 0L) == CURLE_OK &&
         curl_easy_setopt(easy, CURLOPT_MAX_RECV_SPEED_LARGE, kStateProbeReceiveRate) == CURLE_OK;
}
 
std::size_t RuntimeTransferCount(const curl::fuzzer::proto::MultiPlan& plan) {
  return std::max(scenario_limits::kMinMultiTransfers,
                  std::min(static_cast<std::size_t>(plan.transfer_count()), scenario_limits::kMaxMultiTransfers));
}
 
void RecordNotification(CURLM*, unsigned int notification, CURL*, void* userdata) {
  auto* stats = static_cast<MultiTransferRunStats*>(userdata);
  if (stats == nullptr) {
    return;
  }
  if (notification == CURLMNOTIFY_INFO_READ) {
    ++stats->info_read_notifications;
  } else if (notification == CURLMNOTIFY_EASY_DONE) {
    ++stats->easy_done_notifications;
  }
}

class MultiTransferDriver {
 public:
  MultiTransferDriver(CURLM* multi, std::array<TransferState, scenario_limits::kMaxMultiTransfers>& transfers,
                      std::size_t transfer_count, MultiTransferRunStats& stats)
      : multi_(multi), transfers_(transfers), transfer_count_(transfer_count), stats_(stats) {}

  bool DrainCompletionMessages() {
    bool consumed = false;
    int messages_remaining = 0;
    CURLMsg* message = nullptr;
    while ((message = curl_multi_info_read(multi_, &messages_remaining)) != nullptr) {
      consumed = true;
      if (message->msg != CURLMSG_DONE) {
        continue;
      }
      for (std::size_t index = 0; index < transfer_count_; ++index) {
        TransferState& transfer = transfers_[index];
        if (transfer.session.easy() == message->easy_handle) {
          ++stats_.completion_messages;
          break;
        }
      }
    }
    return consumed;
  }

  bool ApplyAction(const curl::fuzzer::proto::MultiAction& action) {
    if (transfer_count_ == 0) {
      return false;
    }
    TransferState& transfer = transfers_[static_cast<std::size_t>(action.transfer_selector()) % transfer_count_];
    CURL* easy = transfer.session.easy();
    if (easy == nullptr) {
      return false;
    }
 
    switch (action.kind()) {
      case curl::fuzzer::proto::MULTI_ACTION_PAUSE_RECV:
        return curl_easy_pause(easy, CURLPAUSE_RECV) == CURLE_OK;
      case curl::fuzzer::proto::MULTI_ACTION_PAUSE_SEND:
        return curl_easy_pause(easy, CURLPAUSE_SEND) == CURLE_OK;
      case curl::fuzzer::proto::MULTI_ACTION_PAUSE_ALL:
        return curl_easy_pause(easy, CURLPAUSE_ALL) == CURLE_OK;
      case curl::fuzzer::proto::MULTI_ACTION_RESUME:
        return curl_easy_pause(easy, CURLPAUSE_CONT) == CURLE_OK;
      case curl::fuzzer::proto::MULTI_ACTION_REMOVE:
        if (transfer.attached && curl_multi_remove_handle(multi_, easy) == CURLM_OK) {
          transfer.attached = false;
          return true;
        }
        return false;
      case curl::fuzzer::proto::MULTI_ACTION_READD:
        if (!transfer.attached && curl_multi_add_handle(multi_, easy) == CURLM_OK) {
          transfer.attached = true;
          return true;
        }
        return false;
      case curl::fuzzer::proto::MULTI_ACTION_NONE:
      default:
        return false;
    }
  }
 
 private:
  CURLM* multi_;
  std::array<TransferState, scenario_limits::kMaxMultiTransfers>& transfers_;
  const std::size_t transfer_count_;
  MultiTransferRunStats& stats_;
};

void ProbeMultiWaitApis(CURLM* multi) {
  int numfds = 0;
  (void)curl_multi_poll(multi, nullptr, 0, 0, &numfds);
  (void)curl_multi_wait(multi, nullptr, 0, 0, &numfds);
 
  unsigned int waitfd_count = 0;
  (void)curl_multi_waitfds(multi, nullptr, 0, &waitfd_count);
  std::array<curl_waitfd, scenario_limits::kMaxMultiTransfers * 2> waitfds{};
  (void)curl_multi_waitfds(multi, waitfds.data(), static_cast<unsigned int>(waitfds.size()), &waitfd_count);
 
  fd_set readfds;
  fd_set writefds;
  fd_set exceptfds;
  FD_ZERO(&readfds);
  FD_ZERO(&writefds);
  FD_ZERO(&exceptfds);
  int maxfd = -1;
  (void)curl_multi_fdset(multi, &readfds, &writefds, &exceptfds, &maxfd);
  long timeout_ms = -1;
  (void)curl_multi_timeout(multi, &timeout_ms);
 
  CURL** handles = curl_multi_get_handles(multi);
  curl_free(handles);
 
  constexpr CURLMinfo_offt kOffsetInfo[] = {
      CURLMINFO_XFERS_CURRENT, CURLMINFO_XFERS_RUNNING, CURLMINFO_XFERS_PENDING,
      CURLMINFO_XFERS_DONE,    CURLMINFO_XFERS_ADDED,
  };
  for (const CURLMinfo_offt info : kOffsetInfo) {
    curl_off_t value = 0;
    (void)curl_multi_get_offt(multi, info, &value);
  }
}

void ProbeLegacyMultiSocketApis(CURLM* multi) {
  int running_handles = 0;
#if defined(__GNUC__)
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
#endif
  (void)curl_multi_socket(multi, CURL_SOCKET_TIMEOUT, &running_handles);
  (void)curl_multi_socket_all(multi, &running_handles);
#if defined(__GNUC__)
#pragma GCC diagnostic pop
#endif
}
 
}  // namespace

MultiTransferRunStats RunMultiTransferScenario(const curl::fuzzer::proto::Scenario& scenario) {
  MultiTransferRunStats stats;
  const auto& plan = scenario.multi_plan();
  const std::size_t transfer_count = RuntimeTransferCount(plan);
  stats.configured_handles = transfer_count;
 
  FilePtr progress_sink;
  if (plan.state_probe_profile() == curl::fuzzer::proto::MULTI_STATE_PROBE_PROGRESS_RATE_LIMIT) {
    progress_sink.reset(std::tmpfile());
  }
 
  // The peer and socket callback state must outlive multi cleanup, which can
  // synchronously announce socket removals and close cached connections.
  MockServer mock;
  mock.SetScripts(scenario);
  mock.SetKeepConnectionsOpen(plan.keep_connections_open());
  MultiSocketDriver socket_driver;
  CurlMultiPtr multi(curl_multi_init());
  if (!multi) {
    return stats;
  }
 
  const long max_host_connections =
      static_cast<long>(std::min<std::size_t>(plan.max_host_connections(), transfer_count));
  const long max_total_connections =
      static_cast<long>(std::min<std::size_t>(plan.max_total_connections(), transfer_count));
  const long cache_size =
      static_cast<long>(std::min<std::size_t>(plan.connection_cache_size(), scenario_limits::kMaxMultiTransfers * 2));
  (void)curl_multi_setopt(multi.get(), CURLMOPT_MAX_HOST_CONNECTIONS, max_host_connections);
  (void)curl_multi_setopt(multi.get(), CURLMOPT_MAX_TOTAL_CONNECTIONS, max_total_connections);
  (void)curl_multi_setopt(multi.get(), CURLMOPT_MAXCONNECTS, cache_size);
  (void)curl_multi_setopt(multi.get(), CURLMOPT_PIPELINING,
                          plan.multiplex() ? static_cast<long>(CURLPIPE_MULTIPLEX) : 0L);
  (void)curl_multi_setopt(multi.get(), CURLMOPT_NOTIFYFUNCTION, &RecordNotification);
  (void)curl_multi_setopt(multi.get(), CURLMOPT_NOTIFYDATA, &stats);
  (void)curl_multi_notify_enable(multi.get(), CURLMNOTIFY_INFO_READ);
  (void)curl_multi_notify_enable(multi.get(), CURLMNOTIFY_EASY_DONE);
  (void)curl_multi_notify_disable(multi.get(), CURLMNOTIFY_INFO_READ);
  (void)curl_multi_notify_enable(multi.get(), CURLMNOTIFY_INFO_READ);
  (void)curl_multi_notify_disable(multi.get(), CURLMNOTIFY_EASY_DONE);
  (void)curl_multi_notify_enable(multi.get(), CURLMNOTIFY_EASY_DONE);
 
  const bool socket_mode = plan.drive_mode() == curl::fuzzer::proto::MULTI_DRIVE_SOCKET;
  const bool socket_driver_installed = socket_mode && socket_driver.Install(multi.get());
  std::array<TransferState, scenario_limits::kMaxMultiTransfers> transfers;
  const std::string url = "http://" + scenario.host_path();
  for (std::size_t index = 0; index < transfer_count; ++index) {
    TransferState& transfer = transfers[index];
    if (!transfer.session.Initialize()) {
      continue;
    }
    transfer.session.ApplyBaseline(scenario);
    (void)curl_easy_setopt(transfer.session.easy(), CURLOPT_URL, url.c_str());
    mock.Install(transfer.session.easy());
    (void)ApplyScenarioOptions(transfer.session.easy(), scenario);
    if (ApplyStateProbeOptions(transfer.session.easy(), plan.state_probe_profile(), progress_sink.get())) {
      ++stats.state_probe_handles;
    }
    mock.ConfigureRequestData(transfer.session.InstallRequestData(scenario));
    if (curl_multi_add_handle(multi.get(), transfer.session.easy()) == CURLM_OK) {
      transfer.attached = true;
      ++stats.added_handles;
    }
  }
 
  if (stats.added_handles == 0) {
    multi.reset();
    return stats;
  }
 
  MultiTransferDriver driver(multi.get(), transfers, transfer_count, stats);
  ProbeLegacyMultiSocketApis(multi.get());
 
  const std::size_t action_count = std::min<std::size_t>(plan.actions_size(), scenario_limits::kMaxMultiActions);
  std::size_t next_action = 0;
  bool initial_action_changed = false;
  if (next_action < action_count) {
    initial_action_changed = driver.ApplyAction(plan.actions(static_cast<int>(next_action++)));
    ++stats.actions_consumed;
  }
 
  int running_handles = 0;
  CURLMcode result = socket_driver_installed ? socket_driver.Start(&running_handles)
                                             : curl_multi_perform(multi.get(), &running_handles);
  bool completion_progress = driver.DrainCompletionMessages();
  ProbeMultiWaitApis(multi.get());
  if (plan.wake_multi()) {
    if (socket_driver_installed) {
      socket_driver.ProbeControlApis();
    } else {
      (void)curl_multi_wakeup(multi.get());
    }
  }
 
  int idle_iterations = initial_action_changed || completion_progress ? 0 : 1;
  for (int iteration = 0; result == CURLM_OK && iteration < kMaxDriveIterations; ++iteration) {
    if (running_handles == 0 && next_action >= action_count) {
      break;
    }
 
    bool made_progress = mock.ServiceConnections();
    if (next_action < action_count) {
      made_progress = driver.ApplyAction(plan.actions(static_cast<int>(next_action++))) || made_progress;
      ++stats.actions_consumed;
      // Consuming an ordered no-op is still harness progress: continue to the
      // next action instead of letting an idle transport hide its suffix.
      made_progress = true;
    }
 
    const int running_before = running_handles;
    if (socket_driver_installed) {
      const MultiSocketDriver::DriveResult drive = socket_driver.DriveReady(&running_handles);
      result = drive.code;
      made_progress = drive.made_progress || made_progress;
    } else {
      result = curl_multi_perform(multi.get(), &running_handles);
      made_progress = running_handles != running_before || made_progress;
    }
    made_progress = driver.DrainCompletionMessages() || made_progress;
 
    if (made_progress) {
      idle_iterations = 0;
    } else if (++idle_iterations >= kMaxIdleIterations) {
      break;
    }
  }
 
  (void)driver.DrainCompletionMessages();
  stats.opened_connections = mock.opened_connection_count();
  for (std::size_t index = 0; index < transfer_count; ++index) {
    TransferState& transfer = transfers[index];
    if (transfer.session.easy() == nullptr) {
      continue;
    }
    (void)curl_easy_pause(transfer.session.easy(), CURLPAUSE_CONT);
    if (transfer.attached) {
      (void)curl_multi_remove_handle(multi.get(), transfer.session.easy());
      transfer.attached = false;
    }
  }
 
  // Destroy the multi while both its socket callback storage and mock peer are
  // live, then detach request-data pointers before cleaning each easy handle.
  multi.reset();
  stats.watch_exhausted = socket_driver.watch_exhausted();
  if (stats.watch_exhausted) {
    std::fputs("curl-fuzzer: socket watch capacity exhausted\n", stderr);
  }
  for (std::size_t index = 0; index < transfer_count; ++index) {
    transfers[index].session.Close();
  }
  return stats;
}
 
}  // namespace proto_fuzzer