ftp_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/ftp_mock_server.h"
 
#include <fcntl.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>
 
#include <algorithm>
#include <cerrno>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <limits>
#include <memory>
#include <string>
#include <string_view>
 
#include "proto_fuzzer/allocation_fault_injection.h"
#include "proto_fuzzer/mock_server.h"
#include "proto_fuzzer/scenario_limits.h"
 
namespace proto_fuzzer {

FtpMockServer::FtpMockServer()
    : scenario_(nullptr),
      next_data_script_(0),
      active_listener_fd_(-1),
      active_listener_client_fd_(CURL_SOCKET_BAD),
      pending_active_channel_(kMaxDataChannels),
      control_reply_count_(0),
      next_control_reply_(0),
      active_pollset_probe_count_(0),
      control_opened_(false) {}

FtpMockServer::~FtpMockServer() { ReleaseConnections(); }

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

bool FtpMockServer::SupportsActiveMode() const { return true; }

void FtpMockServer::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);
}

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

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

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

std::size_t FtpMockServer::opened_data_connection_count() const { return next_data_script_; }

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

void FtpMockServer::ReleaseConnections() {
  CloseActiveDescriptors();
  control_connection_.reset();
  for (DataChannel& channel : data_channels_) {
    channel.connection.reset();
    channel.incoming_data.clear();
  }
}

void FtpMockServer::ResetForScenario(const curl::fuzzer::proto::Scenario& scenario) {
  ReleaseConnections();
  for (DataChannel& channel : data_channels_) {
    channel.script = nullptr;
    channel.transfer_started = false;
    channel.upload = false;
  }
 
  scenario_ = &scenario;
  next_data_script_ = 0;
  pending_active_channel_ = kMaxDataChannels;
  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;
  control_opened_ = false;
}

void FtpMockServer::CloseActiveDescriptors() {
  active_listener_client_fd_ = CURL_SOCKET_BAD;
  if (active_listener_fd_ >= 0) {
    close(active_listener_fd_);
    active_listener_fd_ = -1;
  }
  for (DataChannel& channel : data_channels_) {
    if (channel.active_fd >= 0) {
      close(channel.active_fd);
      channel.active_fd = -1;
    }
  }
  pending_active_channel_ = kMaxDataChannels;
}

void FtpMockServer::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()));
}

bool FtpMockServer::UsesActiveMode() const {
  if (scenario_ == nullptr) {
    return false;
  }
  for (int index = scenario_->options_size() - 1; index >= 0; --index) {
    const auto& option = scenario_->options(index);
    if (option.option_id() == curl::fuzzer::proto::CURLOPT_FTPPORT) {
      return option.value_case() == curl::fuzzer::proto::SetOption::kStringValue && !option.string_value().empty();
    }
  }
  return false;
}

SocketSetupDisposition FtpMockServer::GetSocketSetupDisposition(curl_socket_t curlfd, curlsocktype purpose) const {
  if (purpose == CURLSOCKTYPE_ACCEPT || curlfd == active_listener_client_fd_) {
    return SocketSetupDisposition::kNeedsSetup;
  }
  return SocketSetupDisposition::kAlreadyConnected;
}

curl_socket_t FtpMockServer::OpenActiveListener(struct curl_sockaddr* address) {
  if (address == nullptr || (address->family != AF_INET && address->family != AF_INET6)) {
    return CURL_SOCKET_BAD;
  }
 
  if (pending_active_channel_ == kMaxDataChannels) {
    const std::size_t available_scripts =
        std::min<std::size_t>(kMaxDataChannels, static_cast<std::size_t>(scenario_->subsequent_connections_size()));
    if (next_data_script_ >= available_scripts) {
      return CURL_SOCKET_BAD;
    }
    pending_active_channel_ = next_data_script_++;
    DataChannel& channel = data_channels_[pending_active_channel_];
    channel.script = &scenario_->subsequent_connections(static_cast<int>(pending_active_channel_));
  }
 
  if (active_listener_fd_ >= 0) {
    close(active_listener_fd_);
    active_listener_fd_ = -1;
  }
  active_listener_client_fd_ = CURL_SOCKET_BAD;
 
  const int listener = socket(address->family, address->socktype, address->protocol);
  if (listener < 0) {
    return CURL_SOCKET_BAD;
  }
  active_listener_fd_ = dup(listener);
  if (active_listener_fd_ < 0) {
    close(listener);
    return CURL_SOCKET_BAD;
  }
  active_listener_client_fd_ = listener;
  (void)fcntl(active_listener_fd_, F_SETFD, FD_CLOEXEC);
  return listener;
}

curl_socket_t FtpMockServer::HandleOpenSocket(curlsocktype purpose, struct curl_sockaddr* address) {
  if (scenario_ == nullptr || purpose != CURLSOCKTYPE_IPCXN) {
    return CURL_SOCKET_BAD;
  }
 
  if (!control_opened_) {
    control_connection_ = CreateConnection();
    if (!control_connection_ || !control_connection_->ok()) {
      control_connection_.reset();
      return CURL_SOCKET_BAD;
    }
    control_connection_->SetIncomingDataObserver(this);
 
    ApplyScriptBackpressure(control_connection_.get(), scenario_->connection());
    const std::string& greeting = scenario_->connection().initial_response();
    if (!greeting.empty() &&
        !control_connection_->WriteAll(reinterpret_cast<const unsigned char*>(greeting.data()), greeting.size())) {
      control_connection_.reset();
      return CURL_SOCKET_BAD;
    }
    // Do not consume the unique control slot until its peer is usable. Curl
    // may retry a failed application socket callback, and that retry must not
    // be mistaken for the first passive data connection.
    control_opened_ = true;
    return control_connection_->take_client_fd();
  }
 
  if (UsesActiveMode()) {
    // A TLS peer owns only wrapped socketpairs, so it cannot hand curl a real
    // listener that its data TLS server could later answer. Reject the request
    // immediately: the target policy already removes CURLOPT_FTPPORT, and this
    // guard keeps direct callers on a bounded failure instead of an unusable
    // plaintext data channel.
    if (!SupportsActiveMode()) {
      return CURL_SOCKET_BAD;
    }
    return OpenActiveListener(address);
  }
 
  const std::size_t available_scripts =
      std::min<std::size_t>(kMaxDataChannels, static_cast<std::size_t>(scenario_->subsequent_connections_size()));
  if (next_data_script_ >= available_scripts) {
    return CURL_SOCKET_BAD;
  }
 
  DataChannel& channel = data_channels_[next_data_script_];
  channel.script = &scenario_->subsequent_connections(static_cast<int>(next_data_script_));
  channel.connection = CreateConnection();
  if (!channel.connection || !channel.connection->ok()) {
    channel.connection.reset();
    channel.script = nullptr;
    return CURL_SOCKET_BAD;
  }
  channel.connection->SetIncomingDataObserver(&channel);
 
  ApplyScriptBackpressure(channel.connection.get(), *channel.script);
  ++next_data_script_;
  return channel.connection->take_client_fd();
}

bool FtpMockServer::WriteActiveBytes(int fd, const unsigned char* data, std::size_t size) {
  std::size_t offset = 0;
  while (offset < size) {
    const ssize_t written = send(fd, data + offset, size - offset, MSG_NOSIGNAL | MSG_DONTWAIT);
    if (written > 0) {
      offset += static_cast<std::size_t>(written);
      continue;
    }
    if (written < 0 && errno == EINTR) {
      continue;
    }
    return false;
  }
  return true;
}

bool FtpMockServer::ReadActiveBytes(int fd, std::string* output) {
  bool made_progress = false;
  char buffer[16 * 1024];
  for (;;) {
    const ssize_t amount = recv(fd, buffer, sizeof(buffer), MSG_DONTWAIT);
    if (amount > 0) {
      output->append(buffer, static_cast<std::size_t>(amount));
      made_progress = true;
      continue;
    }
    if (amount < 0 && errno == EINTR) {
      continue;
    }
    return made_progress;
  }
}

void FtpMockServer::ConnectActiveDataChannel() {
  if (active_listener_fd_ < 0 || pending_active_channel_ >= kMaxDataChannels) {
    return;
  }
 
  struct sockaddr_storage address;
  std::memset(&address, 0, sizeof(address));
  socklen_t address_length = sizeof(address);
  if (getsockname(active_listener_fd_, reinterpret_cast<struct sockaddr*>(&address), &address_length) != 0) {
    return;
  }
 
  const int fd = socket(address.ss_family, SOCK_STREAM, IPPROTO_TCP);
  if (fd < 0) {
    return;
  }
  if (connect(fd, reinterpret_cast<struct sockaddr*>(&address), address_length) != 0) {
    close(fd);
    return;
  }
  const int flags = fcntl(fd, F_GETFL, 0);
  if (flags >= 0) {
    (void)fcntl(fd, F_SETFL, flags | O_NONBLOCK);
  }
  (void)fcntl(fd, F_SETFD, FD_CLOEXEC);
 
  DataChannel& channel = data_channels_[pending_active_channel_];
  channel.active_fd = fd;
  close(active_listener_fd_);
  active_listener_fd_ = -1;
  active_listener_client_fd_ = CURL_SOCKET_BAD;
  pending_active_channel_ = kMaxDataChannels;
}

void FtpMockServer::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));
}

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

bool FtpMockServer::VerbEquals(std::string_view verb, std::string_view expected_uppercase) {
  if (verb.size() != expected_uppercase.size()) {
    return false;
  }
  for (std::size_t index = 0; index < verb.size(); ++index) {
    unsigned char actual = static_cast<unsigned char>(verb[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;
}

FtpMockServer::TransferDirection FtpMockServer::DirectionForVerb(std::string_view verb) {
  if (VerbEquals(verb, "STOR") || VerbEquals(verb, "APPE")) {
    return TransferDirection::kUpload;
  }
  if (VerbEquals(verb, "RETR") || VerbEquals(verb, "LIST") || VerbEquals(verb, "NLST") || VerbEquals(verb, "MLSD")) {
    return TransferDirection::kDownload;
  }
  return TransferDirection::kNone;
}

bool FtpMockServer::IsConfiguredCustomDownload(std::string_view verb) const {
  if (scenario_ == nullptr || verb.empty()) {
    return false;
  }
 
  for (int index = scenario_->options_size() - 1; index >= 0; --index) {
    const auto& option = scenario_->options(index);
    if (option.option_id() != curl::fuzzer::proto::CURLOPT_CUSTOMREQUEST) {
      continue;
    }
    if (option.value_case() != curl::fuzzer::proto::SetOption::kStringValue) {
      return false;
    }
    return VerbEquals(verb, CommandVerb(option.string_value()));
  }
  return false;
}

int FtpMockServer::FirstReplyCode(std::string_view response) {
  std::size_t line_start = 0;
  while (line_start < response.size()) {
    const std::size_t newline = response.find('\n', line_start);
    if (newline == std::string_view::npos) {
      return -1;
    }
    const std::size_t line_size = newline - line_start + 1;
    if (line_size > 3) {
      const unsigned char first = static_cast<unsigned char>(response[line_start]);
      const unsigned char second = static_cast<unsigned char>(response[line_start + 1]);
      const unsigned char third = static_cast<unsigned char>(response[line_start + 2]);
      if (first >= '0' && first <= '9' && second >= '0' && second <= '9' && third >= '0' && third <= '9' &&
          response[line_start + 3] == ' ') {
        return 100 * (first - '0') + 10 * (second - '0') + (third - '0');
      }
    }
    line_start = newline + 1;
  }
  return -1;
}

bool FtpMockServer::TailNeedsOnlyNewline(std::string_view response) {
  const std::size_t last_newline = response.rfind('\n');
  const std::size_t tail = last_newline == std::string_view::npos ? 0 : last_newline + 1;
  if (response.size() - tail <= 3) {
    return false;
  }
  const unsigned char first = static_cast<unsigned char>(response[tail]);
  const unsigned char second = static_cast<unsigned char>(response[tail + 1]);
  const unsigned char third = static_cast<unsigned char>(response[tail + 2]);
  return first >= '0' && first <= '9' && second >= '0' && second <= '9' && third >= '0' && third <= '9' &&
         response[tail + 3] == ' ';
}

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

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

void FtpMockServer::QueueTransferCompletion() {
  const std::string* response = NextControlReply();
  if (response != nullptr && response->empty()) {
    if (control_connection_ != nullptr) {
      control_connection_->ShutdownWrite();
    }
    return;
  }
  if (response != nullptr && !response->empty()) {
    (void)WriteControlBytes(*response);
    if (FirstReplyCode(*response) >= 0) {
      return;
    }
    if (TailNeedsOnlyNewline(*response)) {
      (void)WriteControlBytes("\r\n");
      return;
    }
    if (response->back() != '\n') {
      (void)WriteControlBytes("\r\n");
    }
  }
 
  // This line is a liveness guard, not a forced successful outcome: any
  // complete fuzzed final response above remains the reply curl consumes.
  (void)WriteControlBytes("226 mock transfer complete\r\n");
}

void FtpMockServer::PreloadDownload(DataChannel* channel) {
  if (channel == nullptr || channel->script == nullptr || (channel->connection == nullptr && channel->active_fd < 0)) {
    return;
  }
 
  const std::string& initial = channel->script->initial_response();
  const auto write_bytes = [channel](const std::string& bytes) {
    if (bytes.empty()) {
      return true;
    }
    if (channel->connection != nullptr) {
      return channel->connection->WriteAll(reinterpret_cast<const unsigned char*>(bytes.data()), bytes.size());
    }
    return WriteActiveBytes(channel->active_fd, reinterpret_cast<const unsigned char*>(bytes.data()), bytes.size());
  };
  bool complete = write_bytes(initial);
  const std::size_t chunk_count = std::min<std::size_t>(scenario_limits::kMaxResponseChunks,
                                                        static_cast<std::size_t>(channel->script->on_readable_size()));
  for (std::size_t index = 0; complete && index < chunk_count; ++index) {
    const std::string& chunk = channel->script->on_readable(static_cast<int>(index));
    complete = write_bytes(chunk);
  }
  (void)complete;
  if (channel->connection != nullptr) {
    channel->connection->ShutdownWrite();
  } else {
    (void)shutdown(channel->active_fd, SHUT_WR);
  }
}

void FtpMockServer::StartNextTransfer(TransferDirection direction) {
  for (std::size_t index = 0; index < next_data_script_; ++index) {
    DataChannel& channel = data_channels_[index];
    if ((channel.connection == nullptr && channel.active_fd < 0) || channel.transfer_started) {
      continue;
    }
 
    channel.transfer_started = true;
    channel.upload = direction == TransferDirection::kUpload;
    if (!channel.upload) {
      PreloadDownload(&channel);
    }
    // Upload peers keep their write half open until cleanup. A real FTP server
    // does not send an early FIN while receiving a file, and a readiness
    // backend could otherwise report that FIN alongside POLLOUT before curl
    // has delivered the upload callback's bytes.
    return;
  }
}

void FtpMockServer::HandleControlCommand(std::string_view command) {
  const std::string* response = NextControlReply();
  if (response == nullptr) {
    return;
  }
 
  const int response_code = FirstReplyCode(*response);
  const bool response_written = WriteControlBytes(*response);
 
  const std::string_view verb = CommandVerb(command);
  if (response_written && scenario_->ftp_auth_tls_filter_allocation_failure() && VerbEquals(verb, "AUTH") &&
      (response_code == 234 || response_code == 334)) {
    allocation_fault_injection::ArmNextAllocationFailure();
  }
  if ((VerbEquals(verb, "EPRT") || VerbEquals(verb, "PORT")) && response_code >= 200 && response_code < 300) {
    ConnectActiveDataChannel();
  }
  TransferDirection direction = DirectionForVerb(verb);
  if (direction == TransferDirection::kNone && IsConfiguredCustomDownload(verb)) {
    direction = TransferDirection::kDownload;
  }
  const bool starts_download =
      direction == TransferDirection::kDownload && (response_code == 125 || response_code == 150);
  // curl's STOR handler accepts every response below 400 before initiating
  // the upload, even though 125/150 are the conventional successful replies.
  const bool starts_upload = direction == TransferDirection::kUpload && response_code >= 100 && response_code < 400;
  if (starts_download || starts_upload) {
    StartNextTransfer(direction);
    QueueTransferCompletion();
  }
}

bool FtpMockServer::ServiceControlConnection() {
  if (control_connection_ == nullptr) {
    return false;
  }
 
  const std::size_t size_before_read = pending_control_bytes_.size();
  const std::size_t transport_progress = control_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()) {
    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_);
    HandleControlCommand(command);
    consumed += command_size;
    made_progress = true;
  }
 
  if (consumed != 0) {
    pending_control_bytes_.erase(0, consumed);
    // Replies queued while handling a command must reach the wire before the
    // next curl transition. FTP performs its final transfer-reply read inside
    // curl_multi_perform, so a deferred TLS record would turn a completed
    // download or upload into a control-response timeout.
    const std::size_t flush_progress = control_connection_->DrainIncoming();
    made_progress = made_progress || flush_progress != 0;
  }
  return made_progress;
}

bool FtpMockServer::ServiceDataConnections() {
  bool made_progress = false;
  for (std::size_t index = 0; index < next_data_script_; ++index) {
    DataChannel& channel = data_channels_[index];
    if (channel.connection != nullptr) {
      const std::size_t transport_progress = channel.connection->DrainIncoming();
      made_progress = made_progress || transport_progress != 0;
      if (channel.upload && !channel.incoming_data.empty()) {
        CapturePrefix(channel.incoming_data, kMaxCapturedUploadBytes, &uploaded_data_);
        channel.incoming_data.clear();
        made_progress = true;
      }
      continue;
    }
 
    if (channel.upload && channel.active_fd >= 0) {
      std::string bytes;
      (void)ReadActiveBytes(channel.active_fd, &bytes);
      if (!bytes.empty()) {
        CapturePrefix(bytes, kMaxCapturedUploadBytes, &uploaded_data_);
        made_progress = true;
      }
    }
  }
  return made_progress;
}

void FtpMockServer::FinishConnections(bool completed) {
  if (control_connection_ != nullptr) {
    if (completed) {
      (void)WriteControlBytes("221 mock closing control connection\r\n");
    }
    control_connection_->ShutdownWrite();
  }
  for (std::size_t index = 0; index < next_data_script_; ++index) {
    if (data_channels_[index].connection != nullptr) {
      data_channels_[index].connection->ShutdownWrite();
    }
    if (data_channels_[index].active_fd >= 0) {
      (void)shutdown(data_channels_[index].active_fd, SHUT_WR);
    }
  }
  if (active_listener_fd_ >= 0) {
    close(active_listener_fd_);
    active_listener_fd_ = -1;
  }
  active_listener_client_fd_ = CURL_SOCKET_BAD;
}

void FtpMockServer::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 opt-in coverage work. Waiting until the socket
    // callback has run and the transfer remains active reaches FTP's protocol
    // and DO_MORE pollsets without adding a syscall to the default hot loop.
    if (scenario.probe_multi_pollset() && still_running && control_opened_) {
      ProbeMultiPollset(multi);
      ++active_pollset_probe_count_;
    }
 
    bool made_progress = still_running != running_before;
    made_progress = ServiceControlConnection() || made_progress;
    made_progress = ServiceDataConnections() || made_progress;
    if (made_progress) {
      idle_iterations = 0;
    } else {
      ++idle_iterations;
    }
  }
 
  // The final perform may have emitted upload or control bytes immediately
  // before marking the transfer done. Drain/capture them before cleanup.
  (void)ServiceControlConnection();
  (void)ServiceDataConnections();
  FinishConnections(still_running == 0 && result == CURLM_OK);
  scenario_ = nullptr;
}

void FtpMockServer::ObserveState() {
#if defined(PROTO_FUZZER_OBSERVE_STATE)
  MockServerBase::ObserveState();
  const volatile std::size_t ftp_observation[] = {control_transcript().size(), uploaded_data().size(),
                                                  opened_data_connection_count()};
  (void)ftp_observation;
#endif
}
 
}  // namespace proto_fuzzer