ovgate/tests/test_socks5.cpp

// SOCKS5 tests: the codec on its own, then the whole proxy over loopback.
//
// The codec tests are the fuzz-shaped ones -- truncated messages, zero-length
// names, an ATYP nobody has heard of -- because socks5/protocol.h is pure
// functions and those cases are cheap to state here and expensive to debug in a
// relay.
//
// The server tests run the real Server, the real Session, the real
// Authenticator and a real DirectEgress against a real loopback echo server.
// Nothing in the SOCKS5 path is faked: what is swapped out is the *tunnel*,
// which is precisely what DirectEgress exists for (egress/direct_egress.h). So
// these exercise the handshake, the three timers, half-close propagation,
// admission control, UDP ASSOCIATE and the zero-progress rehome exactly as they
// will run in production.

#include <asio.hpp>

#include <chrono>
#include <cstring>
#include <memory>
#include <string>
#include <vector>

#include "common/config.h"
#include "common/error.h"
#include "egress/direct_egress.h"
#include "harness.h"
#include "socks5/protocol.h"
#include "socks5/server.h"

using namespace ovg;
using namespace ovg::socks5;
using namespace std::chrono_literals;

namespace {

using Clock = std::chrono::steady_clock;

bool run_until(asio::io_context &io, const std::function<bool()> &pred,
               std::chrono::milliseconds budget) {
  const auto deadline = Clock::now() + budget;
  while (!pred() && Clock::now() < deadline) {
    io.restart();
    io.run_for(2ms);
  }
  return pred();
}

void settle(asio::io_context &io, std::chrono::milliseconds d = 40ms) {
  const auto deadline = Clock::now() + d;
  while (Clock::now() < deadline) {
    io.restart();
    io.run_for(2ms);
  }
}

// ---------------------------------------------------------------------------
// Codec
// ---------------------------------------------------------------------------

OVG_TEST(protocol_decodes_a_greeting_and_reports_partial_input) {
  const uint8_t full[] = {0x05, 0x02, 0x00, 0x02};

  // Every proper prefix must ask for more, not fail. A client is entitled to
  // dribble the greeting one byte per segment.
  for (size_t n = 0; n < sizeof(full); ++n) {
    Greeting g;
    const Decode d = decode_greeting(full, n, &g);
    CHECK_EQ(static_cast<int>(d.status), static_cast<int>(Status::NeedMore));
  }

  Greeting g;
  const Decode d = decode_greeting(full, sizeof(full), &g);
  CHECK_EQ(static_cast<int>(d.status), static_cast<int>(Status::Ok));
  CHECK_EQ(d.used, sizeof(full));
  CHECK(g.offers(Method::NoAuth));
  CHECK(g.offers(Method::UserPass));
  CHECK(!g.offers(Method::Gssapi));
}

OVG_TEST(protocol_rejects_impossible_greetings) {
  // Wrong version: SOCKS4 knocking on a SOCKS5 port.
  const uint8_t v4[] = {0x04, 0x01, 0x00};
  Greeting g;
  CHECK_EQ(static_cast<int>(decode_greeting(v4, sizeof(v4), &g).status),
           static_cast<int>(Status::Bad));

  // NMETHODS == 0 can never become valid, so it must not sit in NeedMore
  // forever waiting for a method that is not coming.
  const uint8_t none[] = {0x05, 0x00};
  CHECK_EQ(static_cast<int>(decode_greeting(none, sizeof(none), &g).status),
           static_cast<int>(Status::Bad));
}

OVG_TEST(protocol_decodes_userpass_including_empty_fields) {
  const uint8_t msg[] = {0x01, 0x03, 'b', 'o', 'b', 0x02, 'h', 'i'};
  UserPass up;
  const Decode d = decode_userpass(msg, sizeof(msg), &up);
  CHECK_EQ(static_cast<int>(d.status), static_cast<int>(Status::Ok));
  CHECK_EQ(up.username, std::string("bob"));
  CHECK_EQ(up.password, std::string("hi"));

  // A zero-length password is well-formed per the grammar; rejecting it is the
  // authenticator's job, not the parser's.
  const uint8_t empty_pw[] = {0x01, 0x01, 'a', 0x00};
  const Decode d2 = decode_userpass(empty_pw, sizeof(empty_pw), &up);
  CHECK_EQ(static_cast<int>(d2.status), static_cast<int>(Status::Ok));
  CHECK_EQ(up.password, std::string(""));

  const uint8_t wrong_ver[] = {0x05, 0x01, 'a', 0x01, 'b'};
  CHECK_EQ(
      static_cast<int>(decode_userpass(wrong_ver, sizeof(wrong_ver), &up).status),
      static_cast<int>(Status::Bad));
}

OVG_TEST(protocol_decodes_all_three_address_types) {
  {
    const uint8_t req[] = {0x05, 0x01, 0x00, 0x01, 192, 0, 2, 1, 0x01, 0xBB};
    Request r;
    CHECK_EQ(static_cast<int>(decode_request(req, sizeof(req), &r).status),
             static_cast<int>(Status::Ok));
    CHECK_EQ(r.target.to_string(), std::string("192.0.2.1:443"));
  }
  {
    uint8_t req[4 + 16 + 2] = {0x05, 0x01, 0x00, 0x04};
    req[4 + 15] = 1;  // ::1
    req[4 + 16] = 0x00;
    req[4 + 17] = 0x50;
    Request r;
    CHECK_EQ(static_cast<int>(decode_request(req, sizeof(req), &r).status),
             static_cast<int>(Status::Ok));
    CHECK_EQ(r.target.to_string(), std::string("[::1]:80"));
  }
  {
    const uint8_t req[] = {0x05, 0x01, 0x00, 0x03, 0x0B, 'e', 'x', 'a',
                           'm',  'p',  'l',  'e',  '.',  'c', 'o', 'm',
                           0x01, 0xBB};
    Request r;
    CHECK_EQ(static_cast<int>(decode_request(req, sizeof(req), &r).status),
             static_cast<int>(Status::Ok));
    CHECK(r.target.is_domain());
    CHECK_EQ(r.target.domain(), std::string("example.com"));
    CHECK_EQ(r.target.port(), uint16_t{443});
  }
}

OVG_TEST(protocol_normalises_a_literal_sent_as_a_domain) {
  // Some clients put an IP in an ATYP=3 field. Leaving it as a name would make
  // the egress spend a DNS query resolving "192.0.2.7" to itself.
  const uint8_t req[] = {0x05, 0x01, 0x00, 0x03, 0x09, '1', '9', '2',
                         '.',  '0',  '.',  '2',  '.',  '7', 0x00, 0x50};
  Request r;
  CHECK_EQ(static_cast<int>(decode_request(req, sizeof(req), &r).status),
           static_cast<int>(Status::Ok));
  CHECK(!r.target.is_domain());
  CHECK_EQ(r.target.to_string(), std::string("192.0.2.7:80"));
}

OVG_TEST(protocol_rejects_malformed_requests) {
  Request r;

  // A zero-length domain: no more bytes can rescue it.
  const uint8_t zero_name[] = {0x05, 0x01, 0x00, 0x03, 0x00, 0x00, 0x50};
  CHECK_EQ(
      static_cast<int>(decode_request(zero_name, sizeof(zero_name), &r).status),
      static_cast<int>(Status::Bad));

  // Unknown ATYP.
  const uint8_t bad_atyp[] = {0x05, 0x01, 0x00, 0x09, 0x01, 0x02};
  CHECK_EQ(
      static_cast<int>(decode_request(bad_atyp, sizeof(bad_atyp), &r).status),
      static_cast<int>(Status::Bad));

  // Non-zero RSV.
  const uint8_t bad_rsv[] = {0x05, 0x01, 0x77, 0x01, 1, 2, 3, 4, 0, 80};
  CHECK_EQ(static_cast<int>(decode_request(bad_rsv, sizeof(bad_rsv), &r).status),
           static_cast<int>(Status::Bad));

  // A truncated domain is only NeedMore -- the length byte says 11, we have 4.
  const uint8_t truncated[] = {0x05, 0x01, 0x00, 0x03, 0x0B, 'e', 'x', 'a'};
  CHECK_EQ(
      static_cast<int>(decode_request(truncated, sizeof(truncated), &r).status),
      static_cast<int>(Status::NeedMore));
}

OVG_TEST(protocol_keeps_an_unknown_command_parsable) {
  // The point: an unrecognised CMD must still yield a decoded address so the
  // session can answer "command not supported" in-band instead of hanging up.
  const uint8_t req[] = {0x05, 0x77, 0x00, 0x01, 10, 0, 0, 1, 0x00, 0x50};
  Request r;
  CHECK_EQ(static_cast<int>(decode_request(req, sizeof(req), &r).status),
           static_cast<int>(Status::Ok));
  CHECK_EQ(static_cast<int>(r.command), 0x77);
  CHECK_EQ(r.target.to_string(), std::string("10.0.0.1:80"));
}

OVG_TEST(protocol_encodes_a_failure_reply_with_a_wellformed_address) {
  // RFC 1928 requires BND even on failure. An empty endpoint must come out as
  // 0.0.0.0:0, not as a short packet a client will choke on.
  const auto out = encode_reply(Reply::ConnectionRefused, Endpoint());
  CHECK_EQ(out.size(), size_t{10});
  CHECK_EQ(out[0], uint8_t{0x05});
  CHECK_EQ(out[1], uint8_t{0x05});
  CHECK_EQ(out[2], uint8_t{0x00});
  CHECK_EQ(out[3], uint8_t{0x01});
  for (size_t i = 4; i < 10; ++i) CHECK_EQ(out[i], uint8_t{0});
}

OVG_TEST(protocol_maps_errors_onto_rfc1928_reply_codes) {
  CHECK_EQ(static_cast<int>(reply_for(make_error_code(Error::ConnectionRefused))),
           static_cast<int>(Reply::ConnectionRefused));
  CHECK_EQ(static_cast<int>(reply_for(make_error_code(Error::HostUnreachable))),
           static_cast<int>(Reply::HostUnreachable));
  CHECK_EQ(
      static_cast<int>(reply_for(make_error_code(Error::NetworkUnreachable))),
      static_cast<int>(Reply::NetworkUnreachable));
  CHECK_EQ(static_cast<int>(reply_for(make_error_code(Error::NotSupported))),
           static_cast<int>(Reply::CommandNotSupported));
}

OVG_TEST(protocol_round_trips_a_udp_datagram_header) {
  const uint8_t payload[] = {0xDE, 0xAD, 0xBE, 0xEF};
  const Endpoint from(*IpAddress::parse("198.51.100.9"), 53);
  const auto dgram = encode_udp_datagram(from, payload, sizeof(payload));

  UdpHeader h;
  CHECK(decode_udp_header(dgram.data(), dgram.size(), &h));
  CHECK_EQ(h.frag, uint8_t{0});
  CHECK_EQ(h.target.to_string(), std::string("198.51.100.9:53"));
  CHECK_EQ(dgram.size() - h.header_len, sizeof(payload));
  CHECK(std::memcmp(dgram.data() + h.header_len, payload, sizeof(payload)) == 0);
}

OVG_TEST(protocol_udp_header_is_all_or_nothing) {
  // A datagram is a complete message or it is garbage; there is no "wait for
  // more", so a truncated one must be rejected rather than half-decoded.
  const uint8_t truncated[] = {0x00, 0x00, 0x00, 0x01, 1, 2};
  UdpHeader h;
  CHECK(!decode_udp_header(truncated, sizeof(truncated), &h));

  const uint8_t bad_rsv[] = {0x01, 0x00, 0x00, 0x01, 1, 2, 3, 4, 0, 53};
  CHECK(!decode_udp_header(bad_rsv, sizeof(bad_rsv), &h));

  // FRAG is surfaced, not rejected: dropping it is the relay's decision and it
  // has to count the drop.
  const uint8_t fragged[] = {0x00, 0x00, 0x07, 0x01, 1, 2, 3, 4, 0, 53};
  CHECK(decode_udp_header(fragged, sizeof(fragged), &h));
  CHECK_EQ(h.frag, uint8_t{7});
}

// ---------------------------------------------------------------------------
// Authenticator
// ---------------------------------------------------------------------------

OVG_TEST(authenticator_accepts_only_the_right_password) {
  std::vector<Credential> users{make_credential("alice", "s3cret"),
                                make_credential("bob", "hunter2")};
  Authenticator a(users, true);

  CHECK(a.check("alice", "s3cret"));
  CHECK(a.check("bob", "hunter2"));
  CHECK(!a.check("alice", "hunter2"));
  CHECK(!a.check("alice", ""));
  CHECK(!a.check("", "s3cret"));
  CHECK(!a.check("mallory", "s3cret"));
  CHECK_EQ(a.successes(), uint64_t{2});
  CHECK_EQ(a.failures(), uint64_t{4});
}

OVG_TEST(authenticator_hashes_even_for_an_unknown_user) {
  // The property that matters is that a miss does real work: an attacker must
  // not be able to enumerate usernames by timing. Measuring wall-clock in a
  // unit test is flaky, so this asserts the observable proxy for it -- a miss
  // against an empty credential set still returns false rather than
  // short-circuiting on "no users configured".
  Authenticator empty({}, true);
  CHECK(empty.empty());
  CHECK(!empty.check("anyone", "anything"));
  CHECK_EQ(empty.failures(), uint64_t{1});
}

OVG_TEST(authenticator_reload_replaces_the_whole_set) {
  Authenticator a({make_credential("old", "pw")}, true);
  CHECK(a.check("old", "pw"));
  CHECK_EQ(a.replace({make_credential("new", "pw2")}), size_t{1});
  CHECK(!a.check("old", "pw"));
  CHECK(a.check("new", "pw2"));
}

// ---------------------------------------------------------------------------
// A loopback echo server, and a minimal SOCKS5 client to drive the proxy
// ---------------------------------------------------------------------------

// Echoes, and can be told to half-close after N bytes so the proxy's EOF
// propagation is observable from the client side.
class EchoServer {
 public:
  explicit EchoServer(asio::io_context &io)
      : acceptor_(io, asio::ip::tcp::endpoint(
                          asio::ip::make_address("127.0.0.1"), 0)) {
    accept();
  }

  uint16_t port() const { return acceptor_.local_endpoint().port(); }
  void stop() {
    std::error_code ignored;
    acceptor_.close(ignored);
  }
  // When set, the connection sends this greeting then immediately half-closes.
  void say_then_fin(std::string s) { greeting_ = std::move(s); }

 private:
  struct Conn : std::enable_shared_from_this<Conn> {
    Conn(asio::ip::tcp::socket s, std::string greeting)
        : sock(std::move(s)), greeting(std::move(greeting)) {}
    asio::ip::tcp::socket sock;
    std::string greeting;
    char buf[4096];

    void go() {
      if (!greeting.empty()) {
        auto self = shared_from_this();
        asio::async_write(sock, asio::buffer(greeting),
                          [self](const std::error_code &ec, size_t) {
                            if (ec) return;
                            std::error_code ignored;
                            // Half-close: the client must still be able to send.
                            self->sock.shutdown(
                                asio::ip::tcp::socket::shutdown_send, ignored);
                            self->drain();
                          });
        return;
      }
      echo();
    }

    void echo() {
      auto self = shared_from_this();
      sock.async_read_some(
          asio::buffer(buf), [self](const std::error_code &ec, size_t n) {
            if (ec) return;
            asio::async_write(self->sock, asio::buffer(self->buf, n),
                              [self](const std::error_code &wec, size_t) {
                                if (!wec) self->echo();
                              });
          });
    }

    void drain() {
      auto self = shared_from_this();
      sock.async_read_some(asio::buffer(buf),
                           [self](const std::error_code &ec, size_t) {
                             if (!ec) self->drain();
                           });
    }
  };

  void accept() {
    acceptor_.async_accept(
        [this](const std::error_code &ec, asio::ip::tcp::socket s) {
          if (ec) return;
          std::make_shared<Conn>(std::move(s), greeting_)->go();
          accept();
        });
  }

  asio::ip::tcp::acceptor acceptor_;
  std::string greeting_;
};

class UdpEcho {
 public:
  explicit UdpEcho(asio::io_context &io)
      : sock_(io, asio::ip::udp::endpoint(asio::ip::make_address("127.0.0.1"),
                                          0)) {
    recv();
  }
  uint16_t port() const { return sock_.local_endpoint().port(); }
  void stop() {
    std::error_code ignored;
    sock_.close(ignored);
  }

 private:
  void recv() {
    sock_.async_receive_from(asio::buffer(buf_), from_,
                             [this](const std::error_code &ec, size_t n) {
                               if (ec) return;
                               std::error_code ignored;
                               sock_.send_to(asio::buffer(buf_, n), from_, 0,
                                             ignored);
                               recv();
                             });
  }
  asio::ip::udp::socket sock_;
  asio::ip::udp::endpoint from_;
  char buf_[2048];
};

// A blocking SOCKS5 client. Blocking is fine and preferable here: it runs on
// the test thread while the io_context is pumped by `run_until`, so the test
// reads like the protocol exchange it is checking.
//
// Every call needs the io_context pumped concurrently, which is why the helpers
// below take it and spin -- there is no second thread to hide the ordering in.
class Client {
 public:
  Client(asio::io_context &io, uint16_t port) : io_(io), sock_(io) {
    std::error_code ec;
    sock_.connect(asio::ip::tcp::endpoint(asio::ip::make_address("127.0.0.1"),
                                          port),
                  ec);
    connected_ = !ec;
  }

  bool connected() const { return connected_; }
  asio::ip::tcp::socket &socket() { return sock_; }

  void send(const std::vector<uint8_t> &b) {
    std::error_code ec;
    asio::write(sock_, asio::buffer(b), ec);
  }
  void send(const std::string &s) {
    std::error_code ec;
    asio::write(sock_, asio::buffer(s), ec);
  }

  // Reads exactly n bytes while pumping the io_context. Returns fewer on error
  // or timeout, which every caller checks.
  std::vector<uint8_t> recv_exact(size_t n, std::chrono::milliseconds budget) {
    std::vector<uint8_t> out;
    out.resize(n);
    size_t got = 0;
    bool done = false, failed = false;
    asio::async_read(sock_, asio::buffer(out),
                     [&](const std::error_code &ec, size_t transferred) {
                       got = transferred;
                       done = true;
                       failed = static_cast<bool>(ec);
                     });
    run_until(io_, [&] { return done; }, budget);
    if (!done || failed) {
      std::error_code ignored;
      sock_.cancel(ignored);
      run_until(io_, [&] { return done; }, 200ms);
    }
    out.resize(got);
    return out;
  }

  bool eof(std::chrono::milliseconds budget) {
    uint8_t b = 0;
    bool done = false, is_eof = false;
    sock_.async_read_some(asio::buffer(&b, 1),
                          [&](const std::error_code &ec, size_t) {
                            done = true;
                            is_eof = (ec == asio::error::eof);
                          });
    run_until(io_, [&] { return done; }, budget);
    if (!done) {
      std::error_code ignored;
      sock_.cancel(ignored);
      run_until(io_, [&] { return done; }, 200ms);
      return false;
    }
    return is_eof;
  }

  void half_close() {
    std::error_code ignored;
    sock_.shutdown(asio::ip::tcp::socket::shutdown_send, ignored);
  }
  void close() {
    std::error_code ignored;
    sock_.close(ignored);
  }

 private:
  asio::io_context &io_;
  asio::ip::tcp::socket sock_;
  bool connected_ = false;
};

std::vector<uint8_t> connect_request(const std::string &host, uint16_t port) {
  std::vector<uint8_t> out{0x05, 0x01, 0x00};
  append_address(&out, Endpoint(host, port));
  return out;
}

std::vector<uint8_t> connect_request_ip(const std::string &ip, uint16_t port) {
  std::vector<uint8_t> out{0x05, 0x01, 0x00};
  append_address(&out, Endpoint(*IpAddress::parse(ip), port));
  return out;
}

// ---------------------------------------------------------------------------
// Fixture
// ---------------------------------------------------------------------------

struct ProxyFixture {
  asio::io_context io;
  Config cfg;
  EchoServer echo{io};
  UdpEcho udp_echo{io};
  std::shared_ptr<egress::DirectEgress> direct;
  std::unique_ptr<Server> server;

  ProxyFixture() {
    cfg.socks5.listen_address = "127.0.0.1";
    cfg.socks5.listen_port = 0;  // ephemeral; server.port() reports the real one
    cfg.socks5.require_auth = true;
    cfg.socks5.users = {make_credential("u", "p")};
    cfg.socks5.handshake_timeout = 2000ms;
    cfg.socks5.connect_timeout = 2000ms;
    cfg.socks5.idle_timeout = 60000ms;
    cfg.socks5.udp_idle_timeout = 5000ms;
    cfg.socks5.relay_buffer_size = 4096;
  }

  // Deferred so a test can adjust cfg first.
  bool start(std::string *err) {
    direct = egress::DirectEgress::create(io, cfg.dns);
    auto e = std::static_pointer_cast<egress::Egress>(direct);
    server = std::make_unique<Server>(io, cfg, [this] {
      return std::static_pointer_cast<egress::Egress>(direct);
    });
    (void)e;
    return server->start(err);
  }

  uint16_t port() const { return server->port(); }

  ~ProxyFixture() {
    if (server) server->stop();
    echo.stop();
    udp_echo.stop();
    settle(io, 60ms);
    if (direct) direct->shutdown();
    settle(io, 60ms);
  }

  // Greeting + userpass, leaving the connection ready for a request.
  bool handshake(Client *c, const std::string &user = "u",
                 const std::string &pw = "p") {
    c->send(std::vector<uint8_t>{0x05, 0x01, 0x02});
    const auto sel = c->recv_exact(2, 2s);
    if (sel.size() != 2 || sel[0] != 0x05 || sel[1] != 0x02) return false;

    std::vector<uint8_t> auth{0x01, static_cast<uint8_t>(user.size())};
    auth.insert(auth.end(), user.begin(), user.end());
    auth.push_back(static_cast<uint8_t>(pw.size()));
    auth.insert(auth.end(), pw.begin(), pw.end());
    c->send(auth);

    const auto ok = c->recv_exact(2, 2s);
    return ok.size() == 2 && ok[0] == 0x01 && ok[1] == 0x00;
  }
};

// ---------------------------------------------------------------------------
// Handshake and authentication
// ---------------------------------------------------------------------------

OVG_TEST(server_relays_a_connect_after_authenticating) {
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));

  c.send(connect_request_ip("127.0.0.1", f.echo.port()));
  const auto rep = c.recv_exact(10, 2s);
  CHECK_EQ(rep.size(), size_t{10});
  CHECK_EQ(rep[0], uint8_t{0x05});
  CHECK_EQ(rep[1], uint8_t{0x00});  // succeeded

  c.send(std::string("hello proxy"));
  const auto back = c.recv_exact(11, 2s);
  CHECK_EQ(std::string(back.begin(), back.end()), std::string("hello proxy"));

  CHECK_EQ(f.server->stats().accepted, uint64_t{1});
  CHECK_EQ(f.server->stats().auth_ok, uint64_t{1});
}

OVG_TEST(server_rejects_a_bad_password_and_closes) {
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  c.send(std::vector<uint8_t>{0x05, 0x01, 0x02});
  CHECK_EQ(c.recv_exact(2, 2s).size(), size_t{2});

  const std::vector<uint8_t> auth{0x01, 0x01, 'u', 0x04, 'n', 'o', 'p', 'e'};
  c.send(auth);
  const auto rep = c.recv_exact(2, 2s);
  CHECK_EQ(rep.size(), size_t{2});
  CHECK_EQ(rep[0], uint8_t{0x01});
  CHECK_NE(rep[1], uint8_t{0x00});
  // RFC 1929 §2: the server must close after a failed attempt, so the client
  // cannot retry passwords on the same connection.
  CHECK(c.eof(2s));
  CHECK_EQ(f.server->stats().auth_failed, uint64_t{1});
}

OVG_TEST(server_refuses_noauth_when_credentials_are_required) {
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  c.send(std::vector<uint8_t>{0x05, 0x01, 0x00});  // NO_AUTH only
  const auto sel = c.recv_exact(2, 2s);
  CHECK_EQ(sel.size(), size_t{2});
  CHECK_EQ(sel[1], uint8_t{0xFF});  // no acceptable methods
  CHECK(c.eof(2s));
}

OVG_TEST(server_allows_noauth_only_when_configured) {
  ProxyFixture f;
  f.cfg.socks5.require_auth = false;
  f.cfg.socks5.users.clear();
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  c.send(std::vector<uint8_t>{0x05, 0x01, 0x00});
  const auto sel = c.recv_exact(2, 2s);
  CHECK_EQ(sel.size(), size_t{2});
  CHECK_EQ(sel[1], uint8_t{0x00});

  c.send(connect_request_ip("127.0.0.1", f.echo.port()));
  const auto rep = c.recv_exact(10, 2s);
  CHECK_EQ(rep.size(), size_t{10});
  CHECK_EQ(rep[1], uint8_t{0x00});
}

OVG_TEST(server_survives_a_byte_at_a_time_handshake) {
  // The incremental-parse contract, exercised for real: every message arrives
  // in single-byte segments and the session must never mistake a prefix for a
  // protocol error.
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());

  const std::vector<uint8_t> greeting{0x05, 0x01, 0x02};
  for (uint8_t b : greeting) {
    c.send(std::vector<uint8_t>{b});
    settle(f.io, 5ms);
  }
  CHECK_EQ(c.recv_exact(2, 2s).size(), size_t{2});

  const std::vector<uint8_t> auth{0x01, 0x01, 'u', 0x01, 'p'};
  for (uint8_t b : auth) {
    c.send(std::vector<uint8_t>{b});
    settle(f.io, 5ms);
  }
  const auto ok = c.recv_exact(2, 2s);
  CHECK_EQ(ok.size(), size_t{2});
  CHECK_EQ(ok[1], uint8_t{0x00});

  const auto req = connect_request_ip("127.0.0.1", f.echo.port());
  for (uint8_t b : req) {
    c.send(std::vector<uint8_t>{b});
    settle(f.io, 5ms);
  }
  const auto rep = c.recv_exact(10, 2s);
  CHECK_EQ(rep.size(), size_t{10});
  CHECK_EQ(rep[1], uint8_t{0x00});
}

OVG_TEST(server_forwards_bytes_pipelined_behind_the_request) {
  // A client that writes the CONNECT request and its first payload in one
  // segment. Those payload bytes sit in the handshake buffer when the reply is
  // sent, and dropping them is a classic proxy bug.
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));

  auto req = connect_request_ip("127.0.0.1", f.echo.port());
  const std::string payload = "pipelined";
  req.insert(req.end(), payload.begin(), payload.end());
  c.send(req);

  const auto rep = c.recv_exact(10, 2s);
  CHECK_EQ(rep.size(), size_t{10});
  CHECK_EQ(rep[1], uint8_t{0x00});

  const auto back = c.recv_exact(payload.size(), 2s);
  CHECK_EQ(std::string(back.begin(), back.end()), payload);
}

OVG_TEST(server_times_out_a_silent_handshake) {
  ProxyFixture f;
  f.cfg.socks5.handshake_timeout = 150ms;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  // Say nothing at all.
  CHECK(c.eof(3s));
  CHECK(run_until(f.io, [&] { return f.server->stats().active == 0; }, 2s));
}

// ---------------------------------------------------------------------------
// CONNECT behaviour
// ---------------------------------------------------------------------------

OVG_TEST(server_reports_a_refused_connection_in_band) {
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  // A port nothing is listening on. Bind and immediately close to get one that
  // is very unlikely to be reused before the test finishes.
  uint16_t dead_port = 0;
  {
    asio::ip::tcp::acceptor a(
        f.io, asio::ip::tcp::endpoint(asio::ip::make_address("127.0.0.1"), 0));
    dead_port = a.local_endpoint().port();
  }

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));
  c.send(connect_request_ip("127.0.0.1", dead_port));

  const auto rep = c.recv_exact(10, 3s);
  CHECK_EQ(rep.size(), size_t{10});
  CHECK_EQ(rep[0], uint8_t{0x05});
  if (rep[1] == 0x00) {
    // Some sandboxes transparently accept every outbound connection; there is
    // then no refusal to observe and the assertion would be about the network,
    // not the proxy.
    SKIP("this environment accepts connections to a closed port");
  }
  CHECK_EQ(rep[1], static_cast<uint8_t>(Reply::ConnectionRefused));
  CHECK(c.eof(2s));
}

OVG_TEST(server_answers_bind_with_command_not_supported) {
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));

  std::vector<uint8_t> req{0x05, 0x02, 0x00};  // BIND
  append_address(&req, Endpoint(*IpAddress::parse("127.0.0.1"), 80));
  c.send(req);

  const auto rep = c.recv_exact(10, 2s);
  CHECK_EQ(rep.size(), size_t{10});
  CHECK_EQ(rep[1], static_cast<uint8_t>(Reply::CommandNotSupported));
  CHECK(c.eof(2s));
}

OVG_TEST(server_propagates_a_client_half_close_without_truncating_the_reply) {
  // The bug this exists to prevent: treating the client's EOF as "session over"
  // and closing the response direction with it.
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));
  c.send(connect_request_ip("127.0.0.1", f.echo.port()));
  CHECK_EQ(c.recv_exact(10, 2s).size(), size_t{10});

  c.send(std::string("last request"));
  c.half_close();  // we are done sending, but still want the answer

  const auto back = c.recv_exact(12, 3s);
  CHECK_EQ(std::string(back.begin(), back.end()), std::string("last request"));
}

OVG_TEST(server_propagates_a_server_half_close_to_the_client) {
  ProxyFixture f;
  f.echo.say_then_fin("HTTP/1.0 200 OK\r\n\r\n");
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));
  c.send(connect_request_ip("127.0.0.1", f.echo.port()));
  CHECK_EQ(c.recv_exact(10, 2s).size(), size_t{10});

  const std::string expected = "HTTP/1.0 200 OK\r\n\r\n";
  const auto body = c.recv_exact(expected.size(), 2s);
  CHECK_EQ(std::string(body.begin(), body.end()), expected);
  // The far side half-closed; we must see EOF and not a reset.
  CHECK(c.eof(2s));

  // ...and the session only ends once we close our own direction too.
  c.half_close();
  CHECK(run_until(f.io, [&] { return f.server->stats().active == 0; }, 3s));
}

OVG_TEST(server_closes_an_idle_session) {
  ProxyFixture f;
  f.cfg.socks5.idle_timeout = 200ms;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));
  c.send(connect_request_ip("127.0.0.1", f.echo.port()));
  CHECK_EQ(c.recv_exact(10, 2s).size(), size_t{10});

  CHECK(c.eof(3s));
  CHECK(run_until(f.io, [&] { return f.server->stats().active == 0; }, 2s));
}

// ---------------------------------------------------------------------------
// Admission control
// ---------------------------------------------------------------------------

OVG_TEST(server_refuses_connections_past_max_sessions) {
  ProxyFixture f;
  f.cfg.socks5.max_sessions = 2;
  std::string err;
  CHECK(f.start(&err));

  std::vector<std::unique_ptr<Client>> held;
  for (int i = 0; i < 2; ++i) {
    held.push_back(std::make_unique<Client>(f.io, f.port()));
    CHECK(held.back()->connected());
    CHECK(f.handshake(held.back().get()));
  }
  CHECK(run_until(f.io, [&] { return f.server->stats().active == 2; }, 2s));

  Client over(f.io, f.port());
  // The listen backlog means connect() itself succeeds; the refusal shows up
  // as an immediate EOF with no bytes, before any SOCKS5 exchange.
  CHECK(over.connected());
  CHECK(over.eof(2s));
  CHECK(run_until(f.io, [&] { return f.server->stats().rejected == 1; }, 2s));
  CHECK_EQ(f.server->stats().active, int64_t{2});

  // Once one goes away, the next is admitted.
  held.pop_back();
  CHECK(run_until(f.io, [&] { return f.server->stats().active == 1; }, 3s));
  Client again(f.io, f.port());
  CHECK(again.connected());
  CHECK(f.handshake(&again));
}

OVG_TEST(server_refuses_connections_when_there_is_no_egress) {
  // What the proxy must do when the tunnel is down: refuse, not fall back to
  // the host network. A silent leak outside the VPN is the worst possible
  // failure for this program.
  ProxyFixture f;
  std::string err;
  f.direct = egress::DirectEgress::create(f.io, f.cfg.dns);
  f.server = std::make_unique<Server>(
      f.io, f.cfg, []() -> egress::EgressPtr { return nullptr; });
  CHECK(f.server->start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(c.eof(2s));
  CHECK(run_until(f.io, [&] { return f.server->stats().no_egress == 1; }, 2s));
}

// ---------------------------------------------------------------------------
// UDP ASSOCIATE
// ---------------------------------------------------------------------------

// Performs the ASSOCIATE handshake and returns the advertised relay endpoint.
bool associate(ProxyFixture *f, Client *c, asio::ip::udp::endpoint *relay) {
  if (!f->handshake(c)) return false;
  std::vector<uint8_t> req{0x05, 0x03, 0x00};
  // 0.0.0.0:0 -- "I do not know what address I will send from", which is what
  // most clients actually say.
  append_address(&req, Endpoint(*IpAddress::parse("0.0.0.0"), 0));
  c->send(req);

  const auto rep = c->recv_exact(10, 2s);
  if (rep.size() != 10 || rep[1] != 0x00) return false;
  const asio::ip::address_v4::bytes_type addr{rep[4], rep[5], rep[6], rep[7]};
  const uint16_t port = static_cast<uint16_t>((rep[8] << 8) | rep[9]);
  *relay = asio::ip::udp::endpoint(asio::ip::address_v4(addr), port);
  return true;
}

OVG_TEST(server_relays_a_udp_datagram_both_ways) {
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  asio::ip::udp::endpoint relay;
  CHECK(associate(&f, &c, &relay));
  CHECK(relay.port() != 0);

  asio::ip::udp::socket us(f.io, asio::ip::udp::endpoint(
                                     asio::ip::make_address("127.0.0.1"), 0));
  const std::string payload = "ping";
  const Endpoint target(*IpAddress::parse("127.0.0.1"), f.udp_echo.port());
  const auto dgram = encode_udp_datagram(
      target, reinterpret_cast<const uint8_t *>(payload.data()), payload.size());

  std::error_code sec;
  us.send_to(asio::buffer(dgram), relay, 0, sec);
  CHECK(!sec);

  std::vector<uint8_t> buf(2048);
  asio::ip::udp::endpoint from;
  bool got = false;
  size_t got_n = 0;
  us.async_receive_from(asio::buffer(buf), from,
                        [&](const std::error_code &ec, size_t n) {
                          got = !ec;
                          got_n = n;
                        });
  CHECK(run_until(f.io, [&] { return got; }, 3s));

  UdpHeader h;
  CHECK(decode_udp_header(buf.data(), got_n, &h));
  CHECK_EQ(h.frag, uint8_t{0});
  CHECK_EQ(h.target.port(), f.udp_echo.port());
  CHECK_EQ(std::string(buf.begin() + static_cast<long>(h.header_len),
                       buf.begin() + static_cast<long>(got_n)),
           payload);

  std::error_code ignored;
  us.close(ignored);
}

OVG_TEST(server_drops_a_fragmented_datagram) {
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  asio::ip::udp::endpoint relay;
  CHECK(associate(&f, &c, &relay));

  asio::ip::udp::socket us(f.io, asio::ip::udp::endpoint(
                                     asio::ip::make_address("127.0.0.1"), 0));

  // FRAG != 0: must be dropped, per docs/FEASIBILITY.md §5.1.
  std::vector<uint8_t> fragged{0x00, 0x00, 0x01};
  append_address(&fragged,
                 Endpoint(*IpAddress::parse("127.0.0.1"), f.udp_echo.port()));
  fragged.push_back('x');
  std::error_code sec;
  us.send_to(asio::buffer(fragged), relay, 0, sec);
  CHECK(!sec);

  bool got = false;
  std::vector<uint8_t> buf(2048);
  asio::ip::udp::endpoint from;
  us.async_receive_from(asio::buffer(buf), from,
                        [&](const std::error_code &ec, size_t) { got = !ec; });
  // Nothing should come back.
  settle(f.io, 400ms);
  CHECK(!got);

  // ...and the association still works for a well-formed datagram, i.e. the
  // drop did not tear anything down.
  const std::string payload = "still here";
  const auto dgram = encode_udp_datagram(
      Endpoint(*IpAddress::parse("127.0.0.1"), f.udp_echo.port()),
      reinterpret_cast<const uint8_t *>(payload.data()), payload.size());
  us.send_to(asio::buffer(dgram), relay, 0, sec);
  CHECK(run_until(f.io, [&] { return got; }, 3s));

  std::error_code ignored;
  us.cancel(ignored);
  us.close(ignored);
  settle(f.io, 40ms);
}

OVG_TEST(server_tears_down_the_association_with_its_control_connection) {
  // RFC 1928 §7 requires it, and without it a client that walks away leaks a
  // socket and a tunnel PCB.
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  asio::ip::udp::endpoint relay;
  CHECK(associate(&f, &c, &relay));
  CHECK(run_until(f.io, [&] { return f.server->stats().active == 1; }, 2s));

  c.close();
  CHECK(run_until(f.io, [&] { return f.server->stats().active == 0; }, 3s));

  // The relay socket is gone: a datagram to it is no longer forwarded.
  asio::ip::udp::socket us(f.io, asio::ip::udp::endpoint(
                                     asio::ip::make_address("127.0.0.1"), 0));
  const std::string payload = "orphan";
  const auto dgram = encode_udp_datagram(
      Endpoint(*IpAddress::parse("127.0.0.1"), f.udp_echo.port()),
      reinterpret_cast<const uint8_t *>(payload.data()), payload.size());
  std::error_code sec;
  us.send_to(asio::buffer(dgram), relay, 0, sec);

  bool got = false;
  std::vector<uint8_t> buf(2048);
  asio::ip::udp::endpoint from;
  us.async_receive_from(asio::buffer(buf), from,
                        [&](const std::error_code &ec, size_t) { got = !ec; });
  settle(f.io, 300ms);
  CHECK(!got);

  std::error_code ignored;
  us.cancel(ignored);
  us.close(ignored);
  settle(f.io, 40ms);
}

// ---------------------------------------------------------------------------
// Switching: what the server does when the egress underneath it changes
// ---------------------------------------------------------------------------

OVG_TEST(server_rehomes_a_session_that_has_moved_no_bytes) {
  // ARCHITECTURE §5.3. The session is established but has not exchanged a
  // single byte, so it carries no stream state and can be re-dialled on the
  // new egress without the client noticing.
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));
  c.send(connect_request_ip("127.0.0.1", f.echo.port()));
  CHECK_EQ(c.recv_exact(10, 2s).size(), size_t{10});
  CHECK(run_until(f.io, [&] { return f.server->stats().active == 1; }, 2s));

  auto fresh = egress::DirectEgress::create(f.io, f.cfg.dns);
  auto old = std::static_pointer_cast<egress::Egress>(f.direct);
  f.direct = fresh;  // subsequent acquire() hands out the new one
  f.server->on_promote(old, std::static_pointer_cast<egress::Egress>(fresh));

  // The proof it really moved: the old egress ends up with no live streams and
  // the session keeps working.
  CHECK(run_until(f.io, [&] { return old->stats().tcp_active == 0; }, 3s));
  CHECK_EQ(f.server->stats().active, int64_t{1});

  c.send(std::string("after the switch"));
  const auto back = c.recv_exact(16, 3s);
  CHECK_EQ(std::string(back.begin(), back.end()),
           std::string("after the switch"));

  old->shutdown();
  settle(f.io, 60ms);
  fresh->shutdown();
  settle(f.io, 60ms);
}

OVG_TEST(server_leaves_a_session_that_has_carried_bytes_alone) {
  // The other half of the honest answer: once a byte has crossed, the TCP state
  // lives on the old node and cannot be recreated. The session stays put and
  // drains -- it is not silently broken and not silently moved.
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));
  c.send(connect_request_ip("127.0.0.1", f.echo.port()));
  CHECK_EQ(c.recv_exact(10, 2s).size(), size_t{10});

  c.send(std::string("progress"));
  CHECK_EQ(c.recv_exact(8, 2s).size(), size_t{8});

  auto fresh = egress::DirectEgress::create(f.io, f.cfg.dns);
  auto old = std::static_pointer_cast<egress::Egress>(f.direct);
  f.server->on_promote(old, std::static_pointer_cast<egress::Egress>(fresh));
  settle(f.io, 200ms);

  // Still on the old egress, and still working.
  CHECK_EQ(old->stats().tcp_active, int64_t{1});
  c.send(std::string("more"));
  const auto back = c.recv_exact(4, 2s);
  CHECK_EQ(std::string(back.begin(), back.end()), std::string("more"));

  fresh->shutdown();
  settle(f.io, 60ms);
}

OVG_TEST(server_closes_stragglers_when_the_drain_window_expires) {
  // "If you cannot keep the old connections, drop them" -- scoped to exactly
  // the sessions that could not be moved, and only once the grace period is up.
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));
  c.send(connect_request_ip("127.0.0.1", f.echo.port()));
  CHECK_EQ(c.recv_exact(10, 2s).size(), size_t{10});
  c.send(std::string("x"));
  CHECK_EQ(c.recv_exact(1, 2s).size(), size_t{1});

  auto old = std::static_pointer_cast<egress::Egress>(f.direct);
  f.server->on_drain_expired(old);

  CHECK(c.eof(3s));
  CHECK(run_until(f.io, [&] { return f.server->stats().active == 0; }, 3s));
}

OVG_TEST(server_rehomes_a_udp_association_in_place) {
  // ARCHITECTURE §5.4 / FEASIBILITY §5.3: UDP has no sequence state, so only
  // the egress-side socket is replaced. The client keeps talking to the same
  // relay address it was given.
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  asio::ip::udp::endpoint relay;
  CHECK(associate(&f, &c, &relay));

  asio::ip::udp::socket us(f.io, asio::ip::udp::endpoint(
                                     asio::ip::make_address("127.0.0.1"), 0));
  const Endpoint target(*IpAddress::parse("127.0.0.1"), f.udp_echo.port());

  auto send_and_expect = [&](const std::string &payload) {
    const auto dgram = encode_udp_datagram(
        target, reinterpret_cast<const uint8_t *>(payload.data()),
        payload.size());
    std::error_code sec;
    us.send_to(asio::buffer(dgram), relay, 0, sec);
    if (sec) return false;

    std::vector<uint8_t> buf(2048);
    asio::ip::udp::endpoint from;
    bool got = false;
    size_t n = 0;
    us.async_receive_from(asio::buffer(buf), from,
                          [&](const std::error_code &ec, size_t got_n) {
                            got = !ec;
                            n = got_n;
                          });
    if (!run_until(f.io, [&] { return got; }, 3s)) {
      std::error_code ignored;
      us.cancel(ignored);
      settle(f.io, 50ms);
      return false;
    }
    UdpHeader h;
    if (!decode_udp_header(buf.data(), n, &h)) return false;
    return std::string(buf.begin() + static_cast<long>(h.header_len),
                       buf.begin() + static_cast<long>(n)) == payload;
  };

  CHECK(send_and_expect("before"));

  auto fresh = egress::DirectEgress::create(f.io, f.cfg.dns);
  auto old = std::static_pointer_cast<egress::Egress>(f.direct);
  f.direct = fresh;
  f.server->on_promote(old, std::static_pointer_cast<egress::Egress>(fresh));
  settle(f.io, 150ms);

  // Same relay endpoint, same client socket -- only the exit changed.
  CHECK(send_and_expect("after"));
  CHECK_EQ(f.server->stats().active, int64_t{1});

  std::error_code ignored;
  us.close(ignored);
  old->shutdown();
  settle(f.io, 60ms);
  fresh->shutdown();
  settle(f.io, 60ms);
}

// ---------------------------------------------------------------------------
// Concurrency
// ---------------------------------------------------------------------------

OVG_TEST(server_handles_many_concurrent_sessions_without_a_thread_each) {
  // Not a load test -- it is a structural one. 64 sessions run to completion on
  // a single io_context with no thread per connection, which is the property
  // the requirement asked for. The number is small enough to stay fast and
  // large enough that a per-session thread would be obvious.
  constexpr int kN = 64;
  ProxyFixture f;
  f.cfg.socks5.max_sessions = 256;
  std::string err;
  CHECK(f.start(&err));

  struct Peer {
    std::unique_ptr<Client> c;
    bool done = false;
  };
  std::vector<Peer> peers;
  peers.reserve(kN);

  for (int i = 0; i < kN; ++i) {
    peers.push_back(Peer{std::make_unique<Client>(f.io, f.port()), false});
    CHECK(peers.back().c->connected());
  }

  // Drive all of them through the handshake and a CONNECT, interleaved rather
  // than one at a time, so they are genuinely concurrent inside the server.
  for (auto &p : peers) p.c->send(std::vector<uint8_t>{0x05, 0x01, 0x02});
  for (auto &p : peers) CHECK_EQ(p.c->recv_exact(2, 5s).size(), size_t{2});

  const std::vector<uint8_t> auth{0x01, 0x01, 'u', 0x01, 'p'};
  for (auto &p : peers) p.c->send(auth);
  for (auto &p : peers) {
    const auto ok = p.c->recv_exact(2, 5s);
    CHECK_EQ(ok.size(), size_t{2});
    CHECK_EQ(ok[1], uint8_t{0x00});
  }

  const auto req = connect_request_ip("127.0.0.1", f.echo.port());
  for (auto &p : peers) p.c->send(req);
  for (auto &p : peers) {
    const auto rep = p.c->recv_exact(10, 5s);
    CHECK_EQ(rep.size(), size_t{10});
    CHECK_EQ(rep[1], uint8_t{0x00});
  }

  CHECK(run_until(f.io, [&] { return f.server->stats().active == kN; }, 5s));

  for (int i = 0; i < kN; ++i) {
    peers[static_cast<size_t>(i)].c->send("id" + std::to_string(i));
  }
  for (int i = 0; i < kN; ++i) {
    const std::string expect = "id" + std::to_string(i);
    const auto back = peers[static_cast<size_t>(i)].c->recv_exact(
        expect.size(), 5s);
    CHECK_EQ(std::string(back.begin(), back.end()), expect);
  }

  peers.clear();
  CHECK(run_until(f.io, [&] { return f.server->stats().active == 0; }, 5s));
  CHECK_EQ(f.server->stats().rejected, uint64_t{0});
}

OVG_TEST(server_lists_live_sessions_for_the_admin_endpoint) {
  ProxyFixture f;
  std::string err;
  CHECK(f.start(&err));

  Client c(f.io, f.port());
  CHECK(c.connected());
  CHECK(f.handshake(&c));
  c.send(connect_request("example.test", 8080));
  // The resolve will fail (or not) depending on the sandbox; either way the
  // session must be listed with its target while it is trying.
  CHECK(run_until(f.io,
                  [&] {
                    const auto v = f.server->sessions(10);
                    return !v.empty() && v[0].target == "example.test:8080";
                  },
                  2s));

  const auto v = f.server->sessions(10);
  CHECK_EQ(v.size(), size_t{1});
  CHECK_EQ(v[0].command, std::string("CONNECT"));
  CHECK_EQ(v[0].egress_label, std::string("direct"));
  CHECK(v[0].age_ms >= 0);
}

}  // namespace