/*

 * 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),

      active_pollset_probe_count_(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() { ReleaseConnections(); }

/// Create the default plaintext transport for the control and passive data

/// connections. The implicit-TLS FTPS subclass overrides this factory while

/// retaining the command parser and channel bookkeeping.

/// @return A new socketpair-backed connection.

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

/// Plain FTP can create and publish a real loopback listener for PORT/EPRT.

/// Subclasses that only expose wrapped socketpairs override this to keep the

/// active-mode request from producing an unusable plaintext endpoint.

/// @return true for the plaintext FTP peer.

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

/// Retain decoded client bytes for the control parser. Data channels install

/// their own observer so upload bodies are routed without mixing them into the

/// command stream.

/// @param data First decoded byte.

/// @param size Number of bytes available at `data`.

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);

}

/// Append decoded bytes from one passive data channel. The bytes are captured

/// only while the channel is an upload; download channels still drain their

/// transport but discard any unexpected client payload after decoding.

/// @param data First decoded byte.

/// @param size Number of bytes available at `data`.

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);

}

/// 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_; }

/// Report how many requested zero-timeout multi-poll probes ran during the

/// most recent drive. RunLoop increments this only after a successful perform

/// turn while curl still owns an active transfer and the control socket is

/// open.

/// @return Number of active FTP pollset probes in the previous drive.

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

/// Close every server-owned transport before the borrowed scenario or a

/// subclass transport disappears. The out-of-line definition keeps

/// MockConnection complete here while the header only forward-declares it.

void FtpMockServer::ReleaseConnections() {

  CloseActiveDescriptors();

  control_connection_.reset();

  for (DataChannel& channel : data_channels_) {

    channel.connection.reset();

    channel.incoming_data.clear();

  }

}

/// 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) {

  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;

}

/// 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_ = 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();

}

/// 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();

  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;

}

/// Advance every data transport so a TLS handshake and queued response records

/// make progress even when no upload is active. Decoded upload bytes are

/// appended by each channel's observer and copied into a temporary capture so

/// data beyond the observable cap is still removed from the transport. This

/// keeps curl's progress independent of how much a unit test chooses to retain.

/// @return true when a transport or captured upload advanced.

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;

}

/// 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;

    }

    // 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;

}

}  // namespace proto_fuzzer