// 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 #include #include #include #include #include #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 &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(d.status), static_cast(Status::NeedMore)); } Greeting g; const Decode d = decode_greeting(full, sizeof(full), &g); CHECK_EQ(static_cast(d.status), static_cast(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(decode_greeting(v4, sizeof(v4), &g).status), static_cast(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(decode_greeting(none, sizeof(none), &g).status), static_cast(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(d.status), static_cast(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(d2.status), static_cast(Status::Ok)); CHECK_EQ(up.password, std::string("")); const uint8_t wrong_ver[] = {0x05, 0x01, 'a', 0x01, 'b'}; CHECK_EQ( static_cast(decode_userpass(wrong_ver, sizeof(wrong_ver), &up).status), static_cast(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(decode_request(req, sizeof(req), &r).status), static_cast(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(decode_request(req, sizeof(req), &r).status), static_cast(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(decode_request(req, sizeof(req), &r).status), static_cast(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(decode_request(req, sizeof(req), &r).status), static_cast(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(decode_request(zero_name, sizeof(zero_name), &r).status), static_cast(Status::Bad)); // Unknown ATYP. const uint8_t bad_atyp[] = {0x05, 0x01, 0x00, 0x09, 0x01, 0x02}; CHECK_EQ( static_cast(decode_request(bad_atyp, sizeof(bad_atyp), &r).status), static_cast(Status::Bad)); // Non-zero RSV. const uint8_t bad_rsv[] = {0x05, 0x01, 0x77, 0x01, 1, 2, 3, 4, 0, 80}; CHECK_EQ(static_cast(decode_request(bad_rsv, sizeof(bad_rsv), &r).status), static_cast(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(decode_request(truncated, sizeof(truncated), &r).status), static_cast(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(decode_request(req, sizeof(req), &r).status), static_cast(Status::Ok)); CHECK_EQ(static_cast(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(reply_for(make_error_code(Error::ConnectionRefused))), static_cast(Reply::ConnectionRefused)); CHECK_EQ(static_cast(reply_for(make_error_code(Error::HostUnreachable))), static_cast(Reply::HostUnreachable)); CHECK_EQ( static_cast(reply_for(make_error_code(Error::NetworkUnreachable))), static_cast(Reply::NetworkUnreachable)); CHECK_EQ(static_cast(reply_for(make_error_code(Error::NotSupported))), static_cast(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 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(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(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 &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 recv_exact(size_t n, std::chrono::milliseconds budget) { std::vector 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(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 connect_request(const std::string &host, uint16_t port) { std::vector out{0x05, 0x01, 0x00}; append_address(&out, Endpoint(host, port)); return out; } std::vector connect_request_ip(const std::string &ip, uint16_t port) { std::vector 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 direct; std::unique_ptr 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(direct); server = std::make_unique(io, cfg, [this] { return std::static_pointer_cast(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{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 auth{0x01, static_cast(user.size())}; auth.insert(auth.end(), user.begin(), user.end()); auth.push_back(static_cast(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{0x05, 0x01, 0x02}); CHECK_EQ(c.recv_exact(2, 2s).size(), size_t{2}); const std::vector 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{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{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 greeting{0x05, 0x01, 0x02}; for (uint8_t b : greeting) { c.send(std::vector{b}); settle(f.io, 5ms); } CHECK_EQ(c.recv_exact(2, 2s).size(), size_t{2}); const std::vector auth{0x01, 0x01, 'u', 0x01, 'p'}; for (uint8_t b : auth) { c.send(std::vector{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{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(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 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(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> held; for (int i = 0; i < 2; ++i) { held.push_back(std::make_unique(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( 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 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((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(payload.data()), payload.size()); std::error_code sec; us.send_to(asio::buffer(dgram), relay, 0, sec); CHECK(!sec); std::vector 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(h.header_len), buf.begin() + static_cast(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 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 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(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(payload.data()), payload.size()); std::error_code sec; us.send_to(asio::buffer(dgram), relay, 0, sec); bool got = false; std::vector 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(f.direct); f.direct = fresh; // subsequent acquire() hands out the new one f.server->on_promote(old, std::static_pointer_cast(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(f.direct); f.server->on_promote(old, std::static_pointer_cast(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(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(payload.data()), payload.size()); std::error_code sec; us.send_to(asio::buffer(dgram), relay, 0, sec); if (sec) return false; std::vector 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(h.header_len), buf.begin() + static_cast(n)) == payload; }; CHECK(send_and_expect("before")); auto fresh = egress::DirectEgress::create(f.io, f.cfg.dns); auto old = std::static_pointer_cast(f.direct); f.direct = fresh; f.server->on_promote(old, std::static_pointer_cast(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 c; bool done = false; }; std::vector peers; peers.reserve(kN); for (int i = 0; i < kN; ++i) { peers.push_back(Peer{std::make_unique(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{0x05, 0x01, 0x02}); for (auto &p : peers) CHECK_EQ(p.c->recv_exact(2, 5s).size(), size_t{2}); const std::vector 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(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(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