/src/curl_fuzzer/proto_fuzzer/ftp_mock_server.cc

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

/// @file
/// @brief Implementation of the command-aware in-process FTP peer.

#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/mock_server.h"
#include "proto_fuzzer/scenario_limits.h"

namespace proto_fuzzer {

/// Begin without a borrowed protobuf or open socketpairs. RunLoop borrows a
/// Scenario only for the synchronous DriveScenario call in which socket callbacks
/// can run. Keeping all session state in the object avoids process globals that
/// could leak one libFuzzer iteration's protocol position into the next.
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),
      control_opened_(false) {}

/// Release server-owned descriptors and socketpair peers after curl has returned
/// its client fds. Loopback peers are unique_ptr-owned transports; the out-of-line
/// destructor also permits MockConnection to remain forward-declared in the
/// header.
FtpMockServer::~FtpMockServer() { CloseActiveDescriptors(); }

/// Expose the bounded command capture from the most recent drive so tests can
/// verify FTP states without coupling assertions to curl debug output. The
/// capture is bounded independently of protocol processing.
/// @return A prefix of complete control commands, including line endings.
const std::string& FtpMockServer::control_transcript() const { return control_transcript_; }

/// Expose a bounded upload prefix from the most recent drive while continuing
/// to drain all client bytes, so test observability cannot create backpressure.
/// @return Bytes curl sent over passive upload connections.
const std::string& FtpMockServer::uploaded_data() const { return uploaded_data_; }

/// Report how many bounded data scripts curl assigned to passive or active
/// transfer sockets, rather than how many scripts the protobuf contains. Tests
/// can distinguish control-only failures from paths that reached data setup.
/// @return Number of bounded data scripts assigned to a transfer socket.
std::size_t FtpMockServer::opened_data_connection_count() const { return next_data_script_; }

/// Reset every borrowed pointer, descriptor, parser cursor, and observable
/// capture before borrowing the next Scenario. A server can be reused serially,
/// but no connection or response cursor is meaningful across easy-handle drives.
void FtpMockServer::ResetForScenario(const curl::fuzzer::proto::Scenario& scenario) {
  CloseActiveDescriptors();
  control_connection_.reset();
  for (DataChannel& channel : data_channels_) {
    channel.connection.reset();
    channel.active_fd = -1;
    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();
  control_opened_ = false;
}

/// Close only peer-owned active-mode descriptors before reuse or destruction.
/// Curl owns every listener fd returned by HandleOpenSocket;
/// active_listener_fd_ is a private duplicate.
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;
}

/// Apply bounded transport knobs when creating a scripted socket. Clamp a
/// compatibility input's raw protobuf socket size before crossing the platform
/// int boundary. Target policies normally clear FTP backpressure, but direct
/// corpus replay must retain deterministic, well-defined behavior.
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()));
}

/// Return whether the final FTPPORT setopt retained by the Scenario selected
/// active mode. Target policy constrains the actual value to loopback.
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;
}

/// Control/passive sockets are connected AF_UNIX pairs, while active FTP
/// needs curl to bind and listen on the real TCP descriptor returned by
/// OpenActiveListener. The descriptor identity is recorded when opened, so
/// this decision never depends on a sandbox-sensitive getsockopt call.
/// @param curlfd Descriptor returned by HandleOpenSocket.
/// @param purpose Role curl assigned to the descriptor.
/// @return Whether curl must perform its normal socket setup.
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;
}

/// Give curl a genuine INET socket for bind/listen while retaining a duplicate
/// that can discover the ephemeral loopback port afterwards, without parsing
/// curl's generated PORT/EPRT command.
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;
}

/// Allocate one control socket followed by a bounded sequence of passive data
/// sockets, or a real listener when CURLOPT_FTPPORT selected active mode.
/// Data is intentionally not preloaded here: curl has not yet sent the command
/// that determines whether the stream is a listing, download, or upload.
/// @param purpose Socket role requested by curl.
/// @param address Mutable description of curl's intended destination.
/// @return Client descriptor for the selected FTP channel, or
///         CURL_SOCKET_BAD when no channel is available.
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_ = std::make_unique<MockConnection>();
    if (!control_connection_->ok()) {
      control_connection_.reset();
      return CURL_SOCKET_BAD;
    }

    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()) {
    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 = std::make_unique<MockConnection>();
  if (!channel.connection->ok()) {
    channel.connection.reset();
    channel.script = nullptr;
    return CURL_SOCKET_BAD;
  }

  ApplyScriptBackpressure(channel.connection.get(), *channel.script);
  ++next_data_script_;
  return channel.connection->take_client_fd();
}

/// Retry writes only while they can make immediate progress. Scenario bytes
/// are bounded, and dropping a suffix on EAGAIN is preferable to introducing
/// a wall-clock wait into one fuzz iteration.
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;
}

/// Drain an active upload without waiting for curl. The accepted peer is
/// nonblocking and the outer perform loop calls this again after more output.
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;
  }
}

/// Connect to curl's exact listener after queuing a successful PORT/EPRT
/// response. Since active_listener_fd_ duplicates that socket, getsockname
/// observes the assigned port without trusting or reparsing control bytes.
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;
}

/// Retain only a bounded prefix for assertions. Parsing and draining continue
/// against the full transport bytes, so reaching the cap cannot change curl's
/// behavior or manufacture socket backpressure.
/// @param source Newly observed bytes.
/// @param limit Maximum retained destination size.
/// @param destination Capture buffer receiving the available prefix.
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));
}

/// Remove FTP's optional leading horizontal whitespace and return just the
/// verb without allocating. Arguments remain untouched because only curl, not
/// the mock harness, should interpret fuzzed paths and offsets.
/// @param command Complete command line.
/// @return View of its first non-whitespace token.
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);
}

/// Fold only ASCII lowercase command letters. FTP verbs are ASCII tokens, so
/// locale-aware case conversion would add state and undefined signed-char
/// behavior without accepting anything curl can legitimately send.
/// @param verb Parsed command token.
/// @param expected_uppercase Literal uppercase command name.
/// @return true when the verbs are equal ignoring ASCII case.
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;
}

/// Recognize curl's built-in data commands plus MLSD, the useful structured
/// listing custom request. PRET deliberately remains control-only even though
/// its argument can contain RETR/STOR: no data transfer has begun at that
/// point.
/// @param verb Parsed FTP command verb.
/// @return transfer direction, or kNone for control-only commands.
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;
}

/// Recognize the final CURLOPT_CUSTOMREQUEST value as a data command. Curl uses
/// it in place of LIST, so mutations need not rediscover a built-in allowed verb
/// before releasing the passive payload. Scan backwards because repeated setopt
/// calls are legal and libcurl keeps the last value. Comparing only the verb
/// preserves arbitrary custom arguments while avoiding false positives on
/// preparatory commands that receive a fuzzed 125/150 reply after EPSV.
/// @param verb Parsed command token sent by curl.
/// @return true when verb matches the effective custom request option.
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;
}

/// Scan complete lines because a legal FTP response fragment may start with
/// one or more informational lines. curl ends a response at the first line
/// whose first three bytes are digits and whose fourth byte is a space.
/// @param response Raw control fragment.
/// @return numeric status, or -1 when no terminating line is present.
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;
}

/// A raw completion such as "226 done" needs only its missing newline. In
/// that common mutation, appending a second synthetic reply would leave bytes
/// in curl's control cache and misalign a later wildcard transfer.
/// @param response Raw control fragment.
/// @return true when the unterminated tail begins with "ddd ".
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] == ' ';
}

/// Return the next command-aligned primary response within the shared
/// response-count budget. Empty strings still advance the cursor: they are
/// meaningful truncation mutations for ordinary command states and explicitly
/// select control EOF at transfer completion.
/// @return Scenario-owned reply, or nullptr after the bounded prefix.
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_++));
}

/// Keep failed peer writes local to the mock. Response cursors must still
/// advance after curl closes early, otherwise a replacement socket could see
/// a reply intended for an earlier command.
/// @param bytes Response fragment to queue.
/// @return true when the complete fragment was queued.
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()));
}

/// Queue a scenario-provided final reply before curl observes data EOF.
/// ftp_done_control_reply() calls getftpresponse() synchronously inside
/// curl_multi_perform, so the outer loop cannot provide the reply afterwards.
/// If the fuzzed fragment lacks a terminating numeric line, append the smallest
/// completion needed so curl returns to the harness immediately. An explicitly
/// empty repeated value instead half-closes the peer, retaining deterministic
/// liveness while making curl's missing-completion error path seedable.
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");
}

/// Send every runtime-visible data fragment once RETR/LIST has been accepted,
/// then half-close the peer. Delaying until the command prevents early EOF from
/// perturbing passive setup. Sending the complete bounded prefix lets curl drain
/// its data state without event-loop sleeps while each protobuf fragment still
/// affects byte content and parser behavior.
/// @param channel Data peer whose script supplies the download bytes.
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);
  }
}

/// Pair the next transfer command with the oldest passive socket that EPSV or
/// PASV opened but no command has used. FTP serializes data transfers, so this
/// simple cursor is sufficient and avoids interpreting fuzzed port numbers.
/// @param direction Whether curl will read or write the data connection.
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;
  }
}

/// Advance one reply for every complete command. Accepted transfer commands
/// additionally consume their final reply now, because waiting until data EOF
/// would be too late to escape curl's blocking completion read.
/// @param command One newline-terminated command from curl.
void FtpMockServer::HandleControlCommand(std::string_view command) {
  const std::string* response = NextControlReply();
  if (response == nullptr) {
    return;
  }

  const int response_code = FirstReplyCode(*response);
  (void)WriteControlBytes(*response);

  const std::string_view verb = CommandVerb(command);
  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();
  }
}

/// Read without waiting, retain partial final commands, and process all full
/// lines already emitted by curl. The number of replies is bounded even if a
/// malformed command stream contains many newlines.
/// @return true when bytes or a command/reply cursor advanced.
bool FtpMockServer::ServiceControlConnection() {
  if (control_connection_ == nullptr) {
    return false;
  }

  const std::size_t size_before_read = pending_control_bytes_.size();
  control_connection_->ReadAvailable(&pending_control_bytes_);
  bool made_progress = pending_control_bytes_.size() != size_before_read;

  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);
  }
  return made_progress;
}

/// Drain uploads into a temporary buffer so data beyond the observable cap is
/// still removed from the socket. This keeps curl's progress independent of
/// how much a unit test chooses to retain.
/// @return true when curl produced at least one data byte.
bool FtpMockServer::ServiceUploadConnections() {
  bool made_progress = false;
  for (std::size_t index = 0; index < next_data_script_; ++index) {
    DataChannel& channel = data_channels_[index];
    if (!channel.upload || (channel.connection == nullptr && channel.active_fd < 0)) {
      continue;
    }

    std::string bytes;
    if (channel.connection != nullptr) {
      channel.connection->ReadAvailable(&bytes);
    } else {
      (void)ReadActiveBytes(channel.active_fd, &bytes);
    }
    if (!bytes.empty()) {
      CapturePrefix(bytes, kMaxCapturedUploadBytes, &uploaded_data_);
      made_progress = true;
    }
  }
  return made_progress;
}

/// Prepare cleanup while the mock can still write to curl's control socket.
/// A completed easy remains in multi's connection cache until cleanup, where
/// FTP sends QUIT and performs a blocking read. Preloading 221 handles that
/// path; an unfinished drive instead gets EOF so cleanup fails fast.
/// @param completed Whether curl reported that the transfer stopped.
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;
}

/// Alternate curl transitions with immediate peer service until the transfer
/// completes or deterministic idle/operation caps win. No select(), poll(), or
/// sleep is needed: every transport is in-process, and malformed scripts are
/// terminated by half-close after the bounded loop.
/// @param multi Multi handle containing `easy`.
/// @param easy Easy handle already installed on this mock.
/// @param scenario Control and passive-data scripts to borrow.
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;
    }

    bool made_progress = still_running != running_before;
    made_progress = ServiceControlConnection() || made_progress;
    made_progress = ServiceUploadConnections() || 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)ServiceUploadConnections();
  FinishConnections(still_running == 0 && result == CURLM_OK);
  scenario_ = nullptr;
}

}  // namespace proto_fuzzer

Coverage Report

Created: 2026-10-22 06:56