pingpong_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/pingpong_mock_server.h"
 
#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <memory>
#include <string>
#include <string_view>
 
#include "proto_fuzzer/mock_server.h"
 
namespace proto_fuzzer {

PingpongMockServer::PingpongMockServer(MultiDrivePolicy drive_policy)
    : MockServerBase(drive_policy),
      scenario_(nullptr),
      control_reply_count_(0),
      next_control_reply_(0),
      active_pollset_probe_count_(0),
      writable_pollset_probe_count_(0),
      body_length_remaining_(0),
      body_terminator_remaining_(0),
      body_mode_(BodyMode::kCommand),
      control_socket_(CURL_SOCKET_BAD),
      control_opened_(false) {}

PingpongMockServer::~PingpongMockServer() = default;

const std::string& PingpongMockServer::control_transcript() const { return control_transcript_; }

const std::string& PingpongMockServer::uploaded_data() const { return uploaded_data_; }

std::size_t PingpongMockServer::active_pollset_probe_count() const { return active_pollset_probe_count_; }

std::size_t PingpongMockServer::writable_pollset_probe_count() const { return writable_pollset_probe_count_; }

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

void PingpongMockServer::BeginMarkerBody(std::string_view terminator) {
  body_mode_ = BodyMode::kMarker;
  body_marker_.assign(terminator.data(), terminator.size());
  body_length_remaining_ = 0;
  body_terminator_remaining_ = 0;
}

void PingpongMockServer::BeginLiteralBody(std::size_t length) {
  body_mode_ = BodyMode::kLiteral;
  body_marker_.clear();
  body_length_remaining_ = length;
  body_terminator_remaining_ = 2;
}

void PingpongMockServer::BeginExactBody(std::size_t length) {
  body_mode_ = BodyMode::kExact;
  body_marker_.clear();
  body_length_remaining_ = length;
  body_terminator_remaining_ = 0;
}

void PingpongMockServer::BeginOpaqueDrain() {
  body_mode_ = BodyMode::kOpaque;
  body_marker_.clear();
  body_length_remaining_ = 0;
  body_terminator_remaining_ = 0;
}

void PingpongMockServer::ObserveIncomingData(const unsigned char* data, std::size_t size) {
  if (data == nullptr || size == 0) {
    return;
  }
  pending_control_bytes_.append(reinterpret_cast<const char*>(data), size);
}

curl_socket_t PingpongMockServer::HandleOpenSocket(curlsocktype purpose, struct curl_sockaddr* /*address*/) {
  if (scenario_ == nullptr || purpose != CURLSOCKTYPE_IPCXN || control_opened_) {
    return CURL_SOCKET_BAD;
  }
 
  connection_ = CreateConnection();
  if (!connection_ || !connection_->ok()) {
    connection_.reset();
    return CURL_SOCKET_BAD;
  }
  connection_->SetIncomingDataObserver(this);
 
  ApplyScriptBackpressure(connection_.get(), scenario_->connection());
  const std::string& greeting = scenario_->connection().initial_response();
  if (!greeting.empty() &&
      !connection_->WriteAll(reinterpret_cast<const unsigned char*>(greeting.data()), greeting.size())) {
    connection_.reset();
    return CURL_SOCKET_BAD;
  }
 
  control_opened_ = true;
  control_socket_ = connection_->take_client_fd();
  return control_socket_;
}

void PingpongMockServer::ResetForScenario(const curl::fuzzer::proto::Scenario& scenario) {
  connection_.reset();
  scenario_ = &scenario;
  control_reply_count_ = std::min<std::size_t>(scenario_limits::kMaxResponseChunks,
                                               static_cast<std::size_t>(scenario.connection().on_readable_size()));
  next_control_reply_ = 0;
  pending_control_bytes_.clear();
  control_transcript_.clear();
  uploaded_data_.clear();
  active_pollset_probe_count_ = 0;
  writable_pollset_probe_count_ = 0;
  body_marker_.clear();
  body_length_remaining_ = 0;
  body_terminator_remaining_ = 0;
  body_mode_ = BodyMode::kCommand;
  control_socket_ = CURL_SOCKET_BAD;
  control_opened_ = false;
  ResetProtocolState();
}

void PingpongMockServer::ApplyScriptBackpressure(MockConnection* connection,
                                                 const curl::fuzzer::proto::Connection& script) {
  if (connection == nullptr) {
    return;
  }
 
  const std::uint32_t int_max = static_cast<std::uint32_t>(std::numeric_limits<int>::max());
  const auto& backpressure = script.backpressure();
  const int receive_buffer = static_cast<int>(std::min(backpressure.recv_buf_bytes(), int_max));
  connection->ApplyBackpressure(receive_buffer, static_cast<std::size_t>(backpressure.drain_limit()));
}

const std::string* PingpongMockServer::NextControlReply() {
  if (scenario_ == nullptr || next_control_reply_ >= control_reply_count_) {
    return nullptr;
  }
  return &scenario_->connection().on_readable(static_cast<int>(next_control_reply_++));
}

bool PingpongMockServer::WriteControlBytes(std::string_view bytes) {
  return bytes.empty() || (connection_ != nullptr &&
                           connection_->WriteAll(reinterpret_cast<const unsigned char*>(bytes.data()), bytes.size()));
}

bool PingpongMockServer::WriteNextReply() {
  const std::string* response = NextControlReply();
  return response == nullptr ? false : WriteControlBytes(*response);
}

void PingpongMockServer::ShutdownControlWrite() {
  if (connection_ != nullptr) {
    connection_->ShutdownWrite();
  }
}

std::string_view PingpongMockServer::FirstToken(std::string_view line) {
  std::size_t begin = 0;
  while (begin < line.size() && (line[begin] == ' ' || line[begin] == '\t')) {
    ++begin;
  }
 
  std::size_t end = begin;
  while (end < line.size() && line[end] != ' ' && line[end] != '\t' && line[end] != '\r' && line[end] != '\n') {
    ++end;
  }
  return line.substr(begin, end - begin);
}

std::string_view PingpongMockServer::SecondToken(std::string_view line) {
  std::size_t position = 0;
  while (position < line.size() && (line[position] == ' ' || line[position] == '\t')) {
    ++position;
  }
  while (position < line.size() && line[position] != ' ' && line[position] != '\t' && line[position] != '\r' &&
         line[position] != '\n') {
    ++position;
  }
  while (position < line.size() && (line[position] == ' ' || line[position] == '\t')) {
    ++position;
  }
 
  std::size_t end = position;
  while (end < line.size() && line[end] != ' ' && line[end] != '\t' && line[end] != '\r' && line[end] != '\n') {
    ++end;
  }
  return line.substr(position, end - position);
}

bool PingpongMockServer::TokenEquals(std::string_view token, std::string_view expected_uppercase) {
  if (token.size() != expected_uppercase.size()) {
    return false;
  }
  for (std::size_t index = 0; index < token.size(); ++index) {
    unsigned char actual = static_cast<unsigned char>(token[index]);
    if (actual >= 'a' && actual <= 'z') {
      actual = static_cast<unsigned char>(actual - 'a' + 'A');
    }
    if (actual != static_cast<unsigned char>(expected_uppercase[index])) {
      return false;
    }
  }
  return true;
}

void PingpongMockServer::CapturePrefix(std::string_view source, std::size_t limit, std::string* destination) {
  if (destination == nullptr || destination->size() >= limit) {
    return;
  }
  const std::size_t available = limit - destination->size();
  destination->append(source.data(), std::min(source.size(), available));
}

bool PingpongMockServer::ServiceControlConnection() {
  if (connection_ == nullptr) {
    return false;
  }
 
  const std::size_t size_before_read = pending_control_bytes_.size();
  const std::size_t transport_progress = connection_->DrainIncoming();
  bool made_progress = pending_control_bytes_.size() != size_before_read || transport_progress != 0;
 
  std::size_t consumed = 0;
  while (consumed < pending_control_bytes_.size()) {
    if (body_mode_ == BodyMode::kOpaque) {
      consumed = pending_control_bytes_.size();
      made_progress = true;
      break;
    }
 
    if (body_mode_ == BodyMode::kMarker) {
      const std::size_t marker = pending_control_bytes_.find(body_marker_, consumed);
      if (marker == std::string::npos) {
        break;
      }
      const std::size_t body_end = marker + body_marker_.size();
      CapturePrefix(std::string_view(pending_control_bytes_.data() + consumed, body_end - consumed),
                    kMaxCapturedUploadBytes, &uploaded_data_);
      consumed = body_end;
      body_mode_ = BodyMode::kCommand;
      made_progress = true;
      HandleBodyComplete();
      continue;
    }
 
    if (body_mode_ == BodyMode::kExact) {
      const std::size_t available = pending_control_bytes_.size() - consumed;
      const std::size_t take = std::min(available, body_length_remaining_);
      CapturePrefix(std::string_view(pending_control_bytes_.data() + consumed, take), kMaxCapturedUploadBytes,
                    &uploaded_data_);
      consumed += take;
      body_length_remaining_ -= take;
      if (take != 0) {
        made_progress = true;
      }
      if (body_length_remaining_ == 0) {
        body_mode_ = BodyMode::kCommand;
        made_progress = true;
        HandleBodyComplete();
        continue;
      }
      break;
    }
 
    if (body_mode_ == BodyMode::kLiteral) {
      if (body_length_remaining_ != 0) {
        const std::size_t available = pending_control_bytes_.size() - consumed;
        const std::size_t take = std::min(available, body_length_remaining_);
        CapturePrefix(std::string_view(pending_control_bytes_.data() + consumed, take), kMaxCapturedUploadBytes,
                      &uploaded_data_);
        consumed += take;
        body_length_remaining_ -= take;
        made_progress = true;
        continue;
      }
 
      const std::size_t available = pending_control_bytes_.size() - consumed;
      if (body_terminator_remaining_ == 2 && available >= 2 && pending_control_bytes_[consumed] == '\r' &&
          pending_control_bytes_[consumed + 1] == '\n') {
        consumed += 2;
        body_terminator_remaining_ = 0;
        body_mode_ = BodyMode::kCommand;
        made_progress = true;
        HandleBodyComplete();
        continue;
      }
      if (available >= 1) {
        const char byte = pending_control_bytes_[consumed];
        if (byte == '\r' || byte == '\n') {
          ++consumed;
          if (byte == '\n' || body_terminator_remaining_ == 0) {
            body_terminator_remaining_ = 0;
            body_mode_ = BodyMode::kCommand;
            HandleBodyComplete();
          } else {
            body_terminator_remaining_ = 1;
          }
          made_progress = true;
          continue;
        }
        // No terminating empty line is recognisable. Resume command parsing so
        // malformed uploads still make bounded progress.
        body_terminator_remaining_ = 0;
        body_mode_ = BodyMode::kCommand;
        continue;
      }
      break;
    }
 
    const std::size_t newline = pending_control_bytes_.find('\n', consumed);
    if (newline == std::string::npos) {
      break;
    }
    const std::size_t command_size = newline - consumed + 1;
    const std::string_view command(pending_control_bytes_.data() + consumed, command_size);
    CapturePrefix(command, kMaxCapturedControlBytes, &control_transcript_);
    HandleLine(command);
    consumed += command_size;
    made_progress = true;
  }
 
  if (consumed != 0) {
    pending_control_bytes_.erase(0, consumed);
  }
  return made_progress;
}

void PingpongMockServer::FinishConnections() {
  if (connection_ != nullptr) {
    QueueShutdownReply();
    connection_->ShutdownWrite();
  }
}

void PingpongMockServer::HandleDetachedMulti(CurlMultiPtr multi) {
  multi.reset();
  (void)ServiceControlConnection();
}

void PingpongMockServer::RunLoop(CURLM* multi, CURL* easy, const curl::fuzzer::proto::Scenario& scenario) {
  (void)easy;
  ResetForScenario(scenario);
 
  int still_running = 1;
  int idle_iterations = 0;
  int drive_iterations = 0;
  CURLMcode result = CURLM_OK;
  while (still_running && idle_iterations < kMaxIdleIterations && drive_iterations++ < kMaxDriveIterations) {
    const int running_before = still_running;
    result = curl_multi_perform(multi, &still_running);
    if (result != CURLM_OK) {
      break;
    }
 
    // Pollset construction is coverage work, not part of the default driver:
    // keep the extra public-API call behind the scenario opt-in. Waiting until
    // the socket callback has run and the transfer remains active ensures the
    // pollset belongs to a live protocol state rather than setup or cleanup.
    if (scenario.probe_multi_pollset() && still_running && control_opened_) {
      ProbeMultiPollset(multi);
      ++active_pollset_probe_count_;
      const bool incomplete_command = body_mode_ == BodyMode::kCommand && !pending_control_bytes_.empty() &&
                                      pending_control_bytes_.find('\n') == std::string::npos;
      if (incomplete_command && MultiPollsetWantsWrite(multi, control_socket_)) {
        ++writable_pollset_probe_count_;
      }
    }
 
    bool made_progress = still_running != running_before;
    made_progress = ServiceControlConnection() || made_progress;
    if (made_progress) {
      idle_iterations = 0;
    } else {
      ++idle_iterations;
    }
  }
 
  // The final perform may have emitted the last command or body bytes
  // immediately before marking the transfer done. Drain them before cleanup.
  (void)ServiceControlConnection();
  FinishConnections();
  scenario_ = nullptr;
}
 
}  // namespace proto_fuzzer