mock_server.cc

Go to the documentation of this file.

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

 
#include "proto_fuzzer/mock_server.h"
 
#include <fcntl.h>
#include <string.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
 
#include <algorithm>
#include <cerrno>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <string>
#include <string_view>
#include <vector>
 
#include "proto_fuzzer/multi_socket_driver.h"
#include "proto_fuzzer/scenario_limits.h"
#include "proto_fuzzer/ws_frame.h"
 
namespace proto_fuzzer {
 
// curl's debug-only event-based easy entrypoint is intentionally used by the
// API coverage lane. curl-fuzzer builds every bundled curl with ENABLE_DEBUG
// and visible symbols, matching curl's own command-line tool declaration.
extern "C" CURLcode curl_easy_perform_ev(CURL* easy);
 
namespace {
 
constexpr int kMaxConnectOnlyIoIterations = 16;
constexpr std::size_t kConnectOnlyBufferSize = 1024;
 
// fd_set can only represent file descriptors < FD_SETSIZE. Reject any pair that couldn't participate in select()
// without memory corruption.
bool FdFitsInFdSet(int fd) { return fd >= 0 && fd < FD_SETSIZE; }
 
}  // namespace


MockConnection::MockConnection() : server_fd_(-1), client_fd_(-1), drain_limit_(0), incoming_data_observer_(nullptr) {
  int fds[2];
 
  if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) != 0) {
    return;
  }
 
  // The fds must be small enough to fit in an fd_set for select(). If not, close them and fail the constructor.
  if (!FdFitsInFdSet(fds[0]) || !FdFitsInFdSet(fds[1])) {
    close(fds[0]);
    close(fds[1]);
    return;
  }
 
  // Set the server-side fd non-blocking so we can write to it without risk of hanging the fuzzer.
  int flags = fcntl(fds[0], F_GETFL, 0);
  if (flags < 0 || fcntl(fds[0], F_SETFL, flags | O_NONBLOCK) < 0) {
    close(fds[0]);
    close(fds[1]);
    return;
  }
 
  // Success: store the file descriptors.
  server_fd_ = fds[0];
  client_fd_ = fds[1];
}

MockConnection::MockConnection(int adopted_server_fd)
    : server_fd_(adopted_server_fd), client_fd_(-1), drain_limit_(0), incoming_data_observer_(nullptr) {
  if (server_fd_ < 0) {
    return;
  }
  const int flags = fcntl(server_fd_, F_GETFL, 0);
  if (!FdFitsInFdSet(server_fd_) || flags < 0 || fcntl(server_fd_, F_SETFL, flags | O_NONBLOCK) < 0) {
    close(server_fd_);
    server_fd_ = -1;
  }
}

MockConnection::~MockConnection() {
  if (server_fd_ >= 0) {
    close(server_fd_);
  }
  if (client_fd_ >= 0) {
    close(client_fd_);
  }
}

bool MockConnection::ok() const { return server_fd_ >= 0; }

int MockConnection::server_fd() const { return server_fd_; }

std::size_t MockConnection::client_send_buffer_size() const {
  if (client_fd_ < 0) {
    return 0;
  }
  int size = 0;
  socklen_t length = sizeof(size);
  if (getsockopt(client_fd_, SOL_SOCKET, SO_SNDBUF, &size, &length) != 0 || size <= 0) {
    return 0;
  }
  return static_cast<std::size_t>(size);
}

bool MockConnection::EnsureClientSendBufferSize(std::size_t minimum) {
  if (client_send_buffer_size() >= minimum) {
    return true;
  }
  if (client_fd_ < 0 || minimum > static_cast<std::size_t>(std::numeric_limits<int>::max())) {
    return false;
  }
  const int requested = static_cast<int>(minimum);
  if (setsockopt(client_fd_, SOL_SOCKET, SO_SNDBUF, &requested, sizeof(requested)) != 0) {
    return false;
  }
  return client_send_buffer_size() >= minimum;
}

bool MockConnection::FillClientSendBuffer() {
  if (client_fd_ < 0) {
    return false;
  }
  const int flags = fcntl(client_fd_, F_GETFL, 0);
  if (flags < 0 || fcntl(client_fd_, F_SETFL, flags | O_NONBLOCK) < 0) {
    return false;
  }
 
  static constexpr unsigned char kPadding[4096] = {};
  bool full = false;
  while (true) {
    const ssize_t written = ::send(client_fd_, kPadding, sizeof(kPadding), MSG_NOSIGNAL);
    if (written > 0) {
      continue;
    }
    full = written < 0 && (errno == EAGAIN || errno == EWOULDBLOCK);
    break;
  }
  if (fcntl(client_fd_, F_SETFL, flags) < 0) {
    return false;
  }
  return full;
}

curl_socket_t MockConnection::take_client_fd() {
  int fd = client_fd_;
  client_fd_ = -1;
  return static_cast<curl_socket_t>(fd);
}

bool MockConnection::WriteAll(const unsigned char* data, std::size_t size) {
  if (server_fd_ < 0) {
    return false;
  }
  std::size_t written = 0;
  while (written < size) {
    ssize_t n = ::send(server_fd_, data + written, size - written, MSG_NOSIGNAL);
    if (n <= 0) {
      return false;
    }
    written += static_cast<std::size_t>(n);
  }
  return true;
}

std::size_t MockConnection::DrainIncoming() {
  if (server_fd_ < 0) {
    return 0;
  }
  unsigned char scratch[4096];
  std::size_t drained = 0;
  while (drain_limit_ == 0 || drained < drain_limit_) {
    std::size_t want = sizeof(scratch);
    if (drain_limit_ != 0) {
      const std::size_t remaining = drain_limit_ - drained;
      if (remaining < want) {
        want = remaining;
      }
    }
    ssize_t n = ::read(server_fd_, scratch, want);
    if (n <= 0) {
      break;
    }
    NotifyIncomingData(scratch, static_cast<std::size_t>(n));
    drained += static_cast<std::size_t>(n);
  }
  return drained;
}

void MockConnection::SetIncomingDataObserver(IncomingDataObserver* observer) { incoming_data_observer_ = observer; }

void MockConnection::NotifyIncomingData(const unsigned char* data, std::size_t size) {
  if (incoming_data_observer_ != nullptr) {
    incoming_data_observer_->ObserveIncomingData(data, size);
  }
}

void MockConnection::ApplyBackpressure(int recv_buf_bytes, std::size_t drain_limit) {
  if (recv_buf_bytes > 0) {
    if (server_fd_ >= 0) {
      (void)setsockopt(server_fd_, SOL_SOCKET, SO_RCVBUF, &recv_buf_bytes, sizeof(recv_buf_bytes));
    }
    if (client_fd_ >= 0) {
      (void)setsockopt(client_fd_, SOL_SOCKET, SO_SNDBUF, &recv_buf_bytes, sizeof(recv_buf_bytes));
    }
  }
  drain_limit_ = drain_limit;
}

void MockConnection::ReadAvailable(std::string* out) {
  if (server_fd_ < 0 || out == nullptr) {
    return;
  }
  unsigned char scratch[4096];
  while (true) {
    ssize_t n = ::read(server_fd_, scratch, sizeof(scratch));
    if (n <= 0) {
      break;
    }
    out->append(reinterpret_cast<const char*>(scratch), static_cast<std::size_t>(n));
  }
}

void MockConnection::ShutdownWrite() {
  if (server_fd_ < 0) {
    return;
  }
  ::shutdown(server_fd_, SHUT_WR);
}


MockServer::MockServer(MultiDrivePolicy drive_policy)
    : MockServerBase(drive_policy),
      script_count_(0),
      next_script_(0),
      active_script_(nullptr),
      preload_all_chunks_(false),
      keep_connections_open_(false) {}

MockServer::~MockServer() = default;

void MockServer::SetScripts(const curl::fuzzer::proto::Scenario& scenario) {
  ResetConnections();
  script_count_ = 0;
  next_script_ = 0;
 
  const auto append_script = [this](const curl::fuzzer::proto::Connection& connection) {
    ConnectionScript& script = scripts_[script_count_++];
    script.connection = &connection;
    script.raw_chunk_count = std::min<std::size_t>(scenario_limits::kMaxResponseChunks, connection.on_readable_size());
    const std::size_t frame_budget = scenario_limits::kMaxResponseChunks - script.raw_chunk_count;
    script.frame_chunk_count = std::min<std::size_t>(frame_budget, connection.server_frames_size());
    script.next_chunk = 0;
  };
 
  append_script(scenario.connection());
  const std::size_t subsequent_count = std::min<std::size_t>(
      scenario_limits::kMaxConnections - 1, static_cast<std::size_t>(scenario.subsequent_connections_size()));
  for (std::size_t i = 0; i < subsequent_count; ++i) {
    append_script(scenario.subsequent_connections(static_cast<int>(i)));
  }
}

void MockServer::SetKeepConnectionsOpen(bool keep_open) { keep_connections_open_ = keep_open; }

bool MockServer::has_more_chunks() const {
  return active_script_ != nullptr && active_script_->next_chunk < active_script_->chunk_count();
}

std::unique_ptr<MockConnection> MockServer::CreateConnection() { return std::make_unique<MockConnection>(); }

void MockServer::ResetConnections() {
  active_script_ = nullptr;
  connection_.reset();
  previous_connections_.clear();
}

void MockServer::ObserveActiveTransfer(CURL* /*easy*/) {}

curl_socket_t MockServer::HandleOpenSocket(curlsocktype purpose, struct curl_sockaddr* address) {
  (void)purpose;
  (void)address;
  if (next_script_ >= script_count_) {
    // Refusing a fifth socket keeps redirect loops bounded even if curl's own
    // redirect limit is mutated upward or an authentication scheme retries.
    return CURL_SOCKET_BAD;
  }
 
  if (connection_) {
    // libcurl owns the corresponding client fd, so retain the server half
    // until the easy handle leaves the multi instead of closing it at the
    // moment a redirect or retry opens its replacement.
    previous_connections_.push_back(std::move(connection_));
  }
 
  active_script_ = &scripts_[next_script_++];
  const curl::fuzzer::proto::Connection& script_connection = *active_script_->connection;
  connection_ = CreateConnection();
  if (!connection_ || !connection_->ok()) {
    connection_.reset();
    active_script_ = nullptr;
    return CURL_SOCKET_BAD;
  }
 
  // Target policies clamp/clear these values before execution. Saturating the
  // compatibility lane's raw uint32 avoids implementation-defined narrowing
  // while preserving its ability to request any representable socket size.
  const std::uint32_t int_max = static_cast<std::uint32_t>(std::numeric_limits<int>::max());
  const auto& backpressure = script_connection.backpressure();
  const int recv_buf_bytes = static_cast<int>(std::min(backpressure.recv_buf_bytes(), int_max));
  connection_->ApplyBackpressure(recv_buf_bytes, static_cast<std::size_t>(backpressure.drain_limit()));
 
  const std::string& initial_response = script_connection.initial_response();
  if (!initial_response.empty()) {
    if (!connection_->WriteAll(reinterpret_cast<const unsigned char*>(initial_response.data()),
                               initial_response.size())) {
      connection_.reset();
      active_script_ = nullptr;
      return CURL_SOCKET_BAD;
    }
  }
  if (preload_all_chunks_) {
    while (has_more_chunks()) {
      (void)DeliverNextChunk();
    }
  }
  if (active_script_->chunk_count() == 0 && !keep_connections_open_) {
    connection_->ShutdownWrite();
  }
  return connection_->take_client_fd();
}

CURLcode MockServer::DriveEasyScenario(CURL* easy, const curl::fuzzer::proto::Scenario& scenario, bool use_events) {
  SetScripts(scenario);
  preload_all_chunks_ = true;
  (void)curl_easy_setopt(easy, CURLOPT_TIMEOUT_MS, 50L);
  (void)curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 50L);
  // CONNECT_ONLY makes curl's timeleft check report "no limit" for the whole
  // post-connect phase, so the timeouts above stop being enforced; value 2
  // additionally skips the connect-only shortcut and runs a complete transfer.
  // Clearing it is not a coverage loss: the multi-driven WebSocket lanes still
  // exercise CONNECT_ONLY under their own iteration budget. Zero rather than
  // one because CONNECT_ONLY=1 also disables the timeout and stays bounded only
  // through a curl-internal shortcut this harness should not depend on.
  (void)curl_easy_setopt(easy, CURLOPT_CONNECT_ONLY, 0L);
  const CURLcode result = use_events ? curl_easy_perform_ev(easy) : curl_easy_perform(easy);
  preload_all_chunks_ = false;
  return result;
}

ConnectOnlyRunStats MockServer::DriveConnectOnlyScenario(CURL* easy, const curl::fuzzer::proto::Scenario& scenario) {
  ConnectOnlyRunStats stats;
  SetScripts(scenario);
  preload_all_chunks_ = true;
  (void)curl_easy_setopt(easy, CURLOPT_TIMEOUT_MS, 50L);
  (void)curl_easy_setopt(easy, CURLOPT_CONNECTTIMEOUT_MS, 50L);
  (void)curl_easy_setopt(easy, CURLOPT_CONNECT_ONLY, 1L);
  stats.connect_result = curl_easy_perform(easy);
  preload_all_chunks_ = false;
  if (stats.connect_result != CURLE_OK || connection_ == nullptr) {
    return stats;
  }
 
  constexpr std::string_view kFallbackRequest = "GET / HTTP/1.0\r\n\r\n";
  const std::string& upload = scenario.upload().data();
  const std::string_view source = upload.empty() ? kFallbackRequest : std::string_view(upload);
  const std::string_view outgoing = source.substr(0, scenario_limits::kMaxApiStringBytes);
  stats.send_result = CURLE_OK;
  for (int iteration = 0; stats.sent_bytes < outgoing.size() && iteration < kMaxConnectOnlyIoIterations; ++iteration) {
    std::size_t sent = 0;
    stats.send_result =
        curl_easy_send(easy, outgoing.data() + stats.sent_bytes, outgoing.size() - stats.sent_bytes, &sent);
    stats.sent_bytes += sent;
    (void)connection_->DrainIncoming();
    if (stats.send_result != CURLE_OK && stats.send_result != CURLE_AGAIN) {
      break;
    }
  }
 
  std::array<unsigned char, kConnectOnlyBufferSize> incoming{};
  for (int iteration = 0; iteration < kMaxConnectOnlyIoIterations; ++iteration) {
    std::size_t received = 0;
    stats.recv_result = curl_easy_recv(easy, incoming.data(), incoming.size(), &received);
    stats.received_bytes += received;
    if ((stats.recv_result != CURLE_OK && stats.recv_result != CURLE_AGAIN) ||
        (stats.recv_result == CURLE_OK && received == 0)) {
      break;
    }
  }
  return stats;
}

bool MockServer::DeliverNextChunk() {
  if (!connection_ || !has_more_chunks()) {
    return false;
  }
  connection_->DrainIncoming();
  const std::size_t chunk_index = active_script_->next_chunk++;
  const auto& script_connection = *active_script_->connection;
  if (chunk_index < active_script_->raw_chunk_count) {
    const std::string& chunk = script_connection.on_readable(static_cast<int>(chunk_index));
    if (!chunk.empty()) {
      connection_->WriteAll(reinterpret_cast<const unsigned char*>(chunk.data()), chunk.size());
    }
  } else {
    // HTTP scenarios historically accept structured WebSocket frames as raw
    // response bytes after every on_readable chunk. Serialize only the frame
    // curl is about to receive: follow-on scripts and capped suffix frames may
    // never be consumed, so eagerly materialising all of them wastes mutations.
    const std::size_t frame_index = chunk_index - active_script_->raw_chunk_count;
    const std::string chunk = SerializeWebSocketFrame(script_connection.server_frames(static_cast<int>(frame_index)));
    if (!chunk.empty()) {
      connection_->WriteAll(reinterpret_cast<const unsigned char*>(chunk.data()), chunk.size());
    }
  }
  if (!has_more_chunks() && !keep_connections_open_) {
    connection_->ShutdownWrite();
  }
  return true;
}

std::size_t MockServer::DrainIncomingConnections() {
  std::size_t drained = 0;
  for (const auto& previous : previous_connections_) {
    drained += previous->DrainIncoming();
  }
  if (connection_) {
    drained += connection_->DrainIncoming();
  }
  return drained;
}

bool MockServer::ServiceConnections() {
  bool made_progress = DrainIncomingConnections() != 0;
  if (has_more_chunks()) {
    made_progress = DeliverNextChunk() || made_progress;
  }
  return made_progress;
}

std::size_t MockServer::opened_connection_count() const { return next_script_; }

void MockServer::RunSocketActionLoop(CURLM* multi, CURL* easy) {
  MultiSocketDriver* driver = multi_socket_driver();
  if (driver == nullptr) {
    return;
  }
 
  int still_running = 1;
  CURLMcode rc = driver->Start(&still_running);
  ObserveActiveTransfer(easy);
  ResumeResponseIfRequested(easy);
  int idle_iterations = 0;
  int drive_iterations = 0;
  while (rc == CURLM_OK && still_running && idle_iterations < kMaxIdleIterations &&
         drive_iterations++ < kMaxDriveIterations) {
    bool made_progress = ServiceConnections();
    const MultiSocketDriver::DriveResult drive_result = driver->DriveReady(&still_running);
    rc = drive_result.code;
    ObserveActiveTransfer(easy);
    ResumeResponseIfRequested(easy);
    made_progress = drive_result.made_progress || made_progress;
    if (made_progress) {
      idle_iterations = 0;
    } else {
      ++idle_iterations;
    }
  }
}

bool MockServer::SupportsRepeatedPollsetProbe() const { return false; }

void MockServer::RunLoop(CURLM* multi, CURL* easy, const curl::fuzzer::proto::Scenario& scenario) {
  SetScripts(scenario);
 
  if (multi_socket_driver() != nullptr) {
    RunSocketActionLoop(multi, easy);
    return;
  }
 
  int still_running = 1;
  int idle_iterations = 0;
  int drive_iterations = 0;
  bool pollset_probed = false;
  // Each newly opened socket can carry its own pressure settings. Inspect only
  // the borrowed, runtime-bounded scripts: ignored repeated protobuf entries must
  // not opt a fast transfer into timed waits.
  const bool timed_drive =
      std::any_of(scripts_.begin(), scripts_.begin() + script_count_, [](const ConnectionScript& script) {
        const auto& backpressure = script.connection->backpressure();
        return backpressure.recv_buf_bytes() != 0 || backpressure.drain_limit() != 0;
      });
  const int idle_limit = timed_drive ? kMaxTimedIdleIterations : kMaxIdleIterations;
  const bool repeated_pollset_probe = scenario.probe_multi_pollset() && SupportsRepeatedPollsetProbe();
  CURLMcode rc = CURLM_OK;
 
  while (still_running && idle_iterations < idle_limit && drive_iterations++ < kMaxDriveIterations) {
    bool made_progress = false;
    const int running_before = still_running;
    rc = curl_multi_perform(multi, &still_running);
    if (rc != CURLM_OK) {
      break;
    }
    ObserveActiveTransfer(easy);
    ResumeResponseIfRequested(easy);
    made_progress = still_running != running_before;
    if (still_running && ((timed_drive && !pollset_probed) || repeated_pollset_probe)) {
      // Probe only after curl has built the socket/filter chain. Timing cases
      // need one zero-timeout poll; an explicit protocol opt-in probes each
      // bounded perform turn so filters installed after a SOCKS handshake
      // (notably HAProxy) are observed while active. Default fast paths remain
      // free of a per-input polling syscall.
      ProbeMultiPollset(multi);
      pollset_probed = true;
    }
    if (!still_running) {
      break;
    }
 
    // Always drain whatever curl has written. Under backpressure the kernel
    // recv buffer would otherwise stay full — curl short-writes, the mock
    // never consumes, and the transfer wedges until the drive budget. With
    // drain_limit set this still honours the per-tick byte budget.
    made_progress = ServiceConnections() || made_progress;
 
    if (made_progress) {
      idle_iterations = 0;
      continue;
    }
 
    // Ordinary socketpair scenarios never sleep: repeated no-progress
    // performs are enough to settle curl's local state machine. Only an
    // explicit BackpressureConfig opts into short waits so timeout-related
    // behaviour remains fuzzable without taxing every corpus entry.
    if (timed_drive) {
      (void)WaitOnMultiFdset(multi, &rc);
      if (rc != CURLM_OK) {
        break;
      }
    }
    ++idle_iterations;
  }
}
 
}  // namespace proto_fuzzer