Files
ovgate/tests/test_socks5.cpp
iceBear67andClaude Opus 5 b2ba45c9f8 OpenVPN client with an authenticated SOCKS5 front door
A userspace VPN gateway: builds an OpenVPN tunnel to a VPNGate node with
the OpenVPN 3 core, terminates it in-process with lwIP, and serves SOCKS5
(RFC 1928/1929, CONNECT and UDP ASSOCIATE) over it. No root, no tun
device, no routing table changes.

Layout follows the module boundaries in docs/ARCHITECTURE.md:

  vpngate/   directory fetch + CSV parse (lines run to ~13.5 KB, so the
             parser streams rather than splitting on newlines)
  selector/  two-phase pick: cheap prior over the whole list, then real
             TCP handshake timing of the top K
  ovpn/      openvpn3 driven through TunBuilder, packets over a socketpair
  netstack/  lwIP: the TCP/IP stack that makes "no root" possible
  egress/    the swappable way out, and make-before-break switching
  socks5/    the front door
  health/    per-window scoring, and the decision to move
  app/       wiring, admin HTTP, signals

docs/FEASIBILITY.md is the analysis this was built from, including the
one requirement that is not physically possible -- carrying established
TCP connections across a node switch -- and what is done instead
(zero-progress redial, UDP re-homing, grace-period drain).

Tests: 155 without the tunnel egress, 172 with it. The seam is the egress
factory; selection, scoring, history and probing all run for real.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-07-28 04:38:39 +00:00

1286 lines
44 KiB
C++

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