racel/ovgate
1
0
Fork 0
forked from cloud/ovgate
Code Pull Requests Actions Packages Activity
Files
master
ovgate/tests/test_netstack.cpp
T
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

1132 lines
38 KiB
C++
Raw Permalink Blame History

// Netstack tests.
//
// The interesting part here is that the whole TCP path is exercised without a
// VPN, without root and without a network: a PacketLink that loops back into a
// hand-written TCP peer. The peer speaks just enough of RFC 793 to complete a
// handshake, carry data both ways and close -- which is exactly the surface
// LwipTcpStream sits on. That makes connect, half-close, EOF and the receive
// backpressure testable in-process, months before a real tunnel is involved.
//
// It also pins down the address-collision limitation documented in
// netstack/lwip_stack.h, because a limitation nobody tests is a limitation that
// quietly stops holding.
#include <asio.hpp>
#include <algorithm>
#include <chrono>
#include <cstring>
#include <deque>
#include <functional>
#include <string>
#include <vector>
#include "common/error.h"
#include "harness.h"
#include "netstack/dns_resolver.h"
#include "netstack/lwip_stack.h"
#include "netstack/packet_link.h"
namespace {
using namespace ovg;
using namespace ovg::netstack;
using Clock = std::chrono::steady_clock;
// ---------------------------------------------------------------------------
// Loopback link
// ---------------------------------------------------------------------------
class LoopbackLink final : public PacketLink {
public:
// Invoked (on the stack's strand, from inside lwIP's output path) for every
// packet the stack transmits. Must not call back into lwIP.
std::function<void(const std::vector<uint8_t> &)> on_tx;
bool send_packet(const void *data, size_t len) override {
const auto *p = static_cast<const uint8_t *>(data);
std::vector<uint8_t> pkt(p, p + len);
++tx_packets;
if (on_tx) on_tx(pkt);
return true;
}
void async_receive(asio::mutable_buffer buf, const asio::any_io_executor &ex,
RecvHandler h) override {
buf_ = buf;
ex_ = ex;
h_ = std::move(h);
flush();
}
void cancel() override {
if (h_) {
auto h = std::move(h_);
h_ = nullptr;
asio::post(ex_, [h = std::move(h)] {
h(asio::error::operation_aborted, 0);
});
}
}
bool is_open() const override { return open_; }
size_t max_packet_size() const override { return 2048; }
// Hands a packet to the stack.
void inject(std::vector<uint8_t> pkt) {
inbox_.push_back(std::move(pkt));
flush();
}
void close() {
open_ = false;
cancel();
}
size_t tx_packets = 0;
private:
void flush() {
if (!h_ || inbox_.empty()) return;
std::vector<uint8_t> pkt = std::move(inbox_.front());
inbox_.erase(inbox_.begin());
const size_t n = std::min(pkt.size(), buf_.size());
std::memcpy(buf_.data(), pkt.data(), n);
auto h = std::move(h_);
h_ = nullptr;
asio::post(ex_, [h = std::move(h), n] { h(std::error_code{}, n); });
}
asio::mutable_buffer buf_;
asio::any_io_executor ex_;
RecvHandler h_;
std::vector<std::vector<uint8_t>> inbox_;
bool open_ = true;
};
// ---------------------------------------------------------------------------
// A minimal TCP peer
// ---------------------------------------------------------------------------
constexpr uint8_t kFin = 0x01;
constexpr uint8_t kSyn = 0x02;
constexpr uint8_t kRst = 0x04;
constexpr uint8_t kPsh = 0x08;
constexpr uint8_t kAck = 0x10;
uint16_t csum16(const uint8_t *d, size_t n, uint32_t init) {
uint32_t sum = init;
size_t i = 0;
for (; i + 1 < n; i += 2) sum += (uint32_t(d[i]) << 8) | d[i + 1];
if (i < n) sum += uint32_t(d[i]) << 8;
while (sum >> 16) sum = (sum & 0xFFFF) + (sum >> 16);
return static_cast<uint16_t>(~sum);
}
void put16(std::vector<uint8_t> &v, size_t at, uint16_t x) {
v[at] = static_cast<uint8_t>(x >> 8);
v[at + 1] = static_cast<uint8_t>(x & 0xFF);
}
void put32(std::vector<uint8_t> &v, size_t at, uint32_t x) {
v[at] = static_cast<uint8_t>(x >> 24);
v[at + 1] = static_cast<uint8_t>(x >> 16);
v[at + 2] = static_cast<uint8_t>(x >> 8);
v[at + 3] = static_cast<uint8_t>(x);
}
uint16_t get16(const uint8_t *d) { return uint16_t((d[0] << 8) | d[1]); }
uint32_t get32(const uint8_t *d) {
return (uint32_t(d[0]) << 24) | (uint32_t(d[1]) << 16) |
(uint32_t(d[2]) << 8) | uint32_t(d[3]);
}
struct Segment {
uint32_t src = 0, dst = 0;
uint16_t sport = 0, dport = 0;
uint32_t seq = 0, ack = 0;
uint8_t flags = 0;
uint16_t window = 0;
std::vector<uint8_t> payload;
};
bool parse_segment(const std::vector<uint8_t> &pkt, Segment *s) {
if (pkt.size() < 20 || (pkt[0] >> 4) != 4) return false;
const size_t ihl = (pkt[0] & 0x0F) * 4u;
if (ihl < 20 || pkt.size() < ihl + 20) return false;
if (pkt[9] != 6) return false; // not TCP
s->src = get32(&pkt[12]);
s->dst = get32(&pkt[16]);
const uint8_t *t = &pkt[ihl];
const size_t doff = (t[12] >> 4) * 4u;
if (doff < 20 || ihl + doff > pkt.size()) return false;
s->sport = get16(t);
s->dport = get16(t + 2);
s->seq = get32(t + 4);
s->ack = get32(t + 8);
s->flags = t[13];
// Unscaled: our SYN-ACK carries no window-scale option, so RFC 7323 leaves
// scaling off for both directions no matter what lwIP offered in its SYN.
s->window = get16(t + 14);
const size_t total = get16(&pkt[2]);
const size_t end = std::min(total, pkt.size());
if (end > ihl + doff) {
s->payload.assign(pkt.begin() + static_cast<long>(ihl + doff),
pkt.begin() + static_cast<long>(end));
} else {
s->payload.clear();
}
return true;
}
std::vector<uint8_t> build_segment(uint32_t src, uint32_t dst, uint16_t sport,
uint16_t dport, uint32_t seq, uint32_t ack,
uint8_t flags, const uint8_t *data,
size_t dlen, bool with_mss) {
const size_t opt = with_mss ? 4 : 0;
const size_t tcp_len = 20 + opt + dlen;
std::vector<uint8_t> p(20 + tcp_len, 0);
p[0] = 0x45;
put16(p, 2, static_cast<uint16_t>(p.size()));
put16(p, 4, 0x4242);
p[8] = 64;
p[9] = 6;
put32(p, 12, src);
put32(p, 16, dst);
put16(p, 10, csum16(p.data(), 20, 0));
const size_t t = 20;
put16(p, t + 0, sport);
put16(p, t + 2, dport);
put32(p, t + 4, seq);
put32(p, t + 8, ack);
p[t + 12] = static_cast<uint8_t>(((20 + opt) / 4) << 4);
p[t + 13] = flags;
put16(p, t + 14, 32768); // window
if (with_mss) {
p[t + 20] = 2;
p[t + 21] = 4;
put16(p, t + 22, 1360);
}
if (dlen > 0) std::memcpy(&p[t + 20 + opt], data, dlen);
// TCP checksum over the pseudo-header + segment.
uint32_t pseudo = 0;
pseudo += (src >> 16) & 0xFFFF;
pseudo += src & 0xFFFF;
pseudo += (dst >> 16) & 0xFFFF;
pseudo += dst & 0xFFFF;
pseudo += 6;
pseudo += static_cast<uint32_t>(tcp_len);
put16(p, t + 16, csum16(p.data() + t, tcp_len, pseudo));
return p;
}
// Enough of a TCP endpoint to be the far side of one connection.
struct TcpPeer {
LoopbackLink *link = nullptr;
bool echo = true;
bool refuse = false; // answer the SYN with RST
uint32_t my_seq = 700000;
uint32_t their_seq = 0;
uint32_t my_ip = 0, their_ip = 0;
uint16_t my_port = 0, their_port = 0;
bool saw_syn = false;
bool established = false;
bool saw_fin = false;
bool saw_rst = false;
std::vector<uint8_t> received;
// Outbound bytes waiting on the receive window. Everything the peer sends
// goes through here rather than straight onto the link, because a peer that
// ignores the advertised window would paper over exactly the backpressure
// this stack is supposed to apply.
std::vector<uint8_t> outbox;
size_t outbox_sent = 0;
uint16_t their_win = 0;
uint32_t last_ack = 0;
bool have_ack = false;
void send(uint8_t flags, const uint8_t *data = nullptr, size_t len = 0,
bool mss = false) {
link->inject(build_segment(my_ip, their_ip, my_port, their_port, my_seq,
their_seq, flags, data, len, mss));
my_seq += static_cast<uint32_t>(len);
if (flags & (kSyn | kFin)) my_seq += 1;
}
void queue(const uint8_t *data, size_t len) {
outbox.insert(outbox.end(), data, data + len);
}
void pump() {
if (!established || !have_ack) return;
while (outbox_sent < outbox.size()) {
// Unsigned arithmetic, so this stays right across a sequence wrap.
const uint32_t in_flight = my_seq - last_ack;
if (in_flight >= their_win) break;
const size_t room = their_win - in_flight;
const size_t chunk =
std::min({outbox.size() - outbox_sent, room, size_t{1360}});
if (chunk == 0) break;
send(kPsh | kAck, outbox.data() + outbox_sent, chunk);
outbox_sent += chunk;
}
}
void on_tx(const std::vector<uint8_t> &pkt) {
Segment s;
if (!parse_segment(pkt, &s)) return;
if (s.flags & kRst) {
saw_rst = true;
return;
}
if ((s.flags & kSyn) && !(s.flags & kAck)) {
saw_syn = true;
my_ip = s.dst;
their_ip = s.src;
my_port = s.dport;
their_port = s.sport;
their_seq = s.seq + 1;
if (refuse) {
// RST carrying the ACK of the SYN: what a closed port answers.
link->inject(build_segment(my_ip, their_ip, my_port, their_port, 0,
their_seq, kRst | kAck, nullptr, 0, false));
return;
}
send(kSyn | kAck, nullptr, 0, /*mss=*/true);
established = true;
return;
}
if (s.flags & kAck) {
last_ack = s.ack;
have_ack = true;
}
their_win = s.window;
if (!s.payload.empty()) {
received.insert(received.end(), s.payload.begin(), s.payload.end());
their_seq = s.seq + static_cast<uint32_t>(s.payload.size());
send(kAck);
if (echo) queue(s.payload.data(), s.payload.size());
}
if (s.flags & kFin) {
saw_fin = true;
their_seq = s.seq + static_cast<uint32_t>(s.payload.size()) + 1;
send(kAck);
send(kFin | kAck);
return;
}
pump();
}
};
// ---------------------------------------------------------------------------
// A minimal UDP peer: echoes every datagram back to its sender.
// ---------------------------------------------------------------------------
struct UdpPeer {
LoopbackLink *link = nullptr;
std::vector<std::vector<uint8_t>> received;
void on_tx(const std::vector<uint8_t> &pkt) {
if (pkt.size() < 20 || (pkt[0] >> 4) != 4) return;
const size_t ihl = (pkt[0] & 0x0F) * 4u;
if (pkt[9] != 17 || pkt.size() < ihl + 8) return; // not UDP
const uint32_t src = get32(&pkt[12]);
const uint32_t dst = get32(&pkt[16]);
const uint8_t *u = &pkt[ihl];
const uint16_t sport = get16(u);
const uint16_t dport = get16(u + 2);
const size_t ulen = get16(u + 4);
if (ulen < 8 || ihl + ulen > pkt.size()) return;
std::vector<uint8_t> payload(u + 8, u + ulen);
received.push_back(payload);
// Straight back the way it came.
const size_t tot = 8 + payload.size();
std::vector<uint8_t> p(20 + tot, 0);
p[0] = 0x45;
put16(p, 2, static_cast<uint16_t>(p.size()));
p[8] = 64;
p[9] = 17;
put32(p, 12, dst);
put32(p, 16, src);
put16(p, 10, csum16(p.data(), 20, 0));
put16(p, 20 + 0, dport);
put16(p, 20 + 2, sport);
put16(p, 20 + 4, static_cast<uint16_t>(tot));
if (!payload.empty()) {
std::memcpy(&p[28], payload.data(), payload.size());
}
uint32_t pseudo = ((dst >> 16) & 0xFFFF) + (dst & 0xFFFF) +
((src >> 16) & 0xFFFF) + (src & 0xFFFF) + 17u +
static_cast<uint32_t>(tot);
uint16_t ck = csum16(p.data() + 20, tot, pseudo);
// A zero checksum means "not computed" on the wire, so the all-ones form is
// the one that must be sent (RFC 768).
if (ck == 0) ck = 0xFFFF;
put16(p, 20 + 6, ck);
link->inject(std::move(p));
}
};
// ---------------------------------------------------------------------------
// Fixture
// ---------------------------------------------------------------------------
NetifConfig test_cfg(const char *addr, const char *label) {
NetifConfig c;
c.address = *IpAddress::parse(addr);
c.prefix = 24;
c.gateway = *IpAddress::parse("10.0.0.1");
c.mtu = 1500;
c.label = label;
return c;
}
// Everything a netstack test needs, in the one declaration order that is safe,
// draining on the way out.
//
// Both halves of that matter, and neither is obvious. A Netif is destroyed from
// a handler posted to the strand (StrandDeleter), not where the last reference
// is dropped -- and ~Netif calls back into both the PacketLink and the Stack. So
// `link` is declared before `io`, because destroying the io_context is what
// finally runs those handlers; and the destructor drains while every member is
// still alive, rather than leaving it to unwinding order.
//
// Doing this here instead of at the end of each test body is what makes a
// failed CHECK report a failure: the exception unwinds through ~Fixture, which
// still tears down correctly, rather than aborting the process on a call into a
// half-destroyed object.
struct Fixture {
// A deque so that references stay valid as more are added, and declared first
// so they outlive `io` -- see above.
std::deque<LoopbackLink> links;
asio::io_context io;
Stack stack{io};
// Registered so the destructor can close them in dependency order. Tests hold
// their own references too; these are just the handles teardown needs.
std::vector<std::shared_ptr<Netif>> netifs;
std::vector<TcpStreamPtr> streams;
std::vector<UdpSocketPtr> socks;
Fixture() { links.emplace_back(); }
LoopbackLink &link() { return links.front(); }
LoopbackLink &new_link() { return links.emplace_back(); }
~Fixture() {
for (auto &s : streams) {
if (s) s->close();
}
for (auto &s : socks) {
if (s) s->close();
}
settle();
streams.clear();
socks.clear();
settle();
for (auto &n : netifs) {
if (n) n->shutdown();
}
settle();
netifs.clear();
settle();
}
// Runs the io_context for a fixed slice. The Stack's timer re-arms forever,
// so there is no "until idle" to wait for -- only "long enough".
void settle(std::chrono::milliseconds d = std::chrono::milliseconds(30)) {
const auto deadline = Clock::now() + d;
while (Clock::now() < deadline) {
io.restart();
io.run_for(std::chrono::milliseconds(2));
}
}
bool run_until(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(std::chrono::milliseconds(2));
}
return pred();
}
std::shared_ptr<Netif> attach(const NetifConfig &cfg,
std::error_code *ec_out = nullptr) {
return attach_on(link(), cfg, ec_out);
}
std::shared_ptr<Netif> attach_on(LoopbackLink &l, const NetifConfig &cfg,
std::error_code *ec_out = nullptr) {
std::shared_ptr<Netif> netif;
std::error_code ec;
bool done = false;
stack.async_attach(l, cfg, io.get_executor(),
[&](const std::error_code &e, std::shared_ptr<Netif> n) {
ec = e;
netif = std::move(n);
done = true;
});
run_until([&] { return done; }, std::chrono::milliseconds(500));
if (ec_out) *ec_out = ec;
if (netif) netifs.push_back(netif);
return netif;
}
TcpStreamPtr connect(const std::shared_ptr<Netif> &netif, const char *addr,
uint16_t port, Millis timeout,
std::error_code *ec_out = nullptr) {
TcpStreamPtr stream;
std::error_code ec;
bool done = false;
netif->async_connect_tcp(*IpAddress::parse(addr), port, timeout,
io.get_executor(),
[&](const std::error_code &e, TcpStreamPtr s) {
ec = e;
stream = std::move(s);
done = true;
});
run_until([&] { return done; }, std::chrono::seconds(5));
if (ec_out) *ec_out = ec;
if (stream) streams.push_back(stream);
return stream;
}
UdpSocketPtr open_udp(const std::shared_ptr<Netif> &netif,
std::error_code *ec_out = nullptr) {
UdpSocketPtr sock;
std::error_code ec;
bool done = false;
netif->async_open_udp(io.get_executor(),
[&](const std::error_code &e, UdpSocketPtr s) {
ec = e;
sock = std::move(s);
done = true;
});
run_until([&] { return done; }, std::chrono::milliseconds(500));
if (ec_out) *ec_out = ec;
if (sock) socks.push_back(sock);
return sock;
}
};
} // namespace
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
OVG_TEST(netstack_attach_and_teardown) {
Fixture fx;
auto netif = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(netif != nullptr);
CHECK(netif->is_up());
CHECK_EQ(netif->address().to_string(), std::string("10.0.0.2"));
CHECK_EQ(fx.stack.global_stats().netifs, size_t{1});
netif->shutdown();
fx.settle();
CHECK(!netif->is_up());
CHECK_EQ(fx.stack.global_stats().netifs, size_t{0});
}
OVG_TEST(netstack_rejects_a_second_stack) {
Fixture fx;
bool threw = false;
try {
Stack second(fx.io);
} catch (const std::logic_error &) {
threw = true;
}
CHECK(threw);
}
// The documented make-before-break limitation: two nodes that push the same
// address cannot both be live, because lwIP attributes inbound packets by
// destination address and could not tell them apart.
OVG_TEST(netstack_rejects_a_colliding_address) {
Fixture fx;
LoopbackLink &link_b = fx.new_link();
auto a = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(a != nullptr);
std::error_code ec;
auto b = fx.attach_on(link_b, test_cfg("10.0.0.2", "node-b"), &ec);
CHECK(b == nullptr);
CHECK_EQ(ec, make_error_code(Error::ResourceExhausted));
// A different address on the same stack is fine -- that is the case
// make-before-break actually relies on.
auto c = fx.attach_on(link_b, test_cfg("10.9.0.2", "node-c"));
CHECK(c != nullptr);
CHECK_EQ(fx.stack.global_stats().netifs, size_t{2});
}
OVG_TEST(netstack_tcp_connect_echo_and_half_close) {
Fixture fx;
TcpPeer peer;
peer.link = &fx.link();
fx.link().on_tx = [&](const std::vector<uint8_t> &p) { peer.on_tx(p); };
auto netif = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(netif != nullptr);
// --- connect ---
std::error_code connect_ec;
auto stream =
fx.connect(netif, "93.184.216.34", 80, Millis{3000}, &connect_ec);
CHECK_EQ(connect_ec, std::error_code{});
CHECK(stream != nullptr);
CHECK(stream->is_open());
CHECK(peer.saw_syn);
CHECK_EQ(stream->remote_endpoint().to_string(),
std::string("93.184.216.34:80"));
// The source address comes from the netif, which is what proves the PCB was
// pinned rather than routed through whatever netif happened to be default.
CHECK_EQ(stream->local_endpoint().address().to_string(),
std::string("10.0.0.2"));
// --- write, and read the echo back ---
const std::string msg = "GET / HTTP/1.0\r\n\r\n";
std::error_code write_ec;
size_t written = 0;
bool wrote = false;
stream->async_write(asio::buffer(msg),
[&](const std::error_code &ec, size_t n) {
write_ec = ec;
written = n;
wrote = true;
});
CHECK(fx.run_until([&] { return wrote; }, std::chrono::seconds(3)));
CHECK_EQ(write_ec, std::error_code{});
CHECK_EQ(written, msg.size());
CHECK(fx.run_until([&] { return peer.received.size() >= msg.size(); },
std::chrono::seconds(3)));
CHECK_EQ(std::string(peer.received.begin(), peer.received.end()), msg);
std::vector<char> rbuf(256);
std::error_code read_ec;
size_t got = 0;
bool read_done = false;
stream->async_read_some(asio::buffer(rbuf),
[&](const std::error_code &ec, size_t n) {
read_ec = ec;
got = n;
read_done = true;
});
CHECK(fx.run_until([&] { return read_done; }, std::chrono::seconds(3)));
CHECK_EQ(read_ec, std::error_code{});
CHECK_EQ(std::string(rbuf.data(), got), msg);
CHECK_EQ(stream->bytes_written(), uint64_t{msg.size()});
CHECK_GT(stream->bytes_read(), uint64_t{0});
// --- half close: our FIN must reach the peer, and its FIN must surface as
// EOF rather than as an error ---
stream->shutdown_send();
CHECK(fx.run_until([&] { return peer.saw_fin; }, std::chrono::seconds(3)));
bool eof_done = false;
std::error_code eof_ec;
stream->async_read_some(asio::buffer(rbuf),
[&](const std::error_code &ec, size_t n) {
eof_ec = ec;
(void)n;
eof_done = true;
});
CHECK(fx.run_until([&] { return eof_done; }, std::chrono::seconds(3)));
CHECK_EQ(eof_ec, std::error_code(asio::error::eof));
const auto s = netif->stats();
CHECK_EQ(s.tcp_opened, uint64_t{1});
CHECK_EQ(s.tcp_failed, uint64_t{0});
CHECK_GT(s.tx_packets, uint64_t{0});
CHECK_GT(s.rx_packets, uint64_t{0});
}
// The point of the explicit-backpressure design: lwIP's window must stay closed
// until the consumer actually takes the bytes, and must reopen by exactly what
// was taken. A peer that respects the advertised window can only push the whole
// transfer through if that loop works -- and the transfer here is twice TCP_WND,
// so it cannot complete on the initial window alone.
OVG_TEST(netstack_tcp_receive_backpressure) {
Fixture fx;
TcpPeer peer;
peer.link = &fx.link();
peer.echo = false;
fx.link().on_tx = [&](const std::vector<uint8_t> &p) { peer.on_tx(p); };
auto netif = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(netif != nullptr);
auto stream = fx.connect(netif, "93.184.216.34", 80, Millis{3000});
CHECK(stream != nullptr);
constexpr size_t kTotal = 128 * 1024; // 2x TCP_WND
std::vector<uint8_t> payload(kTotal);
for (size_t i = 0; i < kTotal; ++i) {
payload[i] = static_cast<uint8_t>((i * 31 + (i >> 8)) & 0xFF);
}
peer.queue(payload.data(), payload.size());
peer.pump();
// Phase 1: no reader. The whole 128 KB is queued and the peer is willing to
// send all of it, so the only thing that can stop it is lwIP's window
// closing -- which it can only do because nothing calls tcp_recved() until
// the consumer takes the bytes. Without that, this would run to completion.
//
// The bound is two-sided on purpose. Above: a stack that never opened the
// window at all would also "stall", so the peer has to get most of a window
// through before it stops. Below: 65535 is what a 16-bit window field can
// advertise once the peer's SYN-ACK omits the scale option (RFC 7323 needs it
// on both SYNs), so a byte past that means tcp_recved() ran without a reader.
fx.settle(std::chrono::milliseconds(150));
CHECK_GT(peer.outbox_sent, size_t{48 * 1024});
CHECK(peer.outbox_sent <= 65535);
// Phase 2: small reads on purpose. The window has to reopen incrementally,
// and a pbuf gets consumed across several reads (the queue_offset_ path).
std::vector<uint8_t> got;
got.reserve(kTotal);
std::vector<char> rbuf(4096);
std::error_code read_ec;
bool failed = false;
std::function<void()> read_more = [&] {
stream->async_read_some(asio::buffer(rbuf),
[&](const std::error_code &ec, size_t n) {
if (ec) {
read_ec = ec;
failed = true;
return;
}
got.insert(got.end(), rbuf.begin(),
rbuf.begin() + static_cast<long>(n));
if (got.size() < kTotal) read_more();
});
};
read_more();
CHECK(fx.run_until([&] { return failed || got.size() >= kTotal; },
std::chrono::seconds(20)));
CHECK_EQ(read_ec, std::error_code{});
CHECK_EQ(got.size(), kTotal);
CHECK(got == payload);
CHECK_EQ(stream->bytes_read(), uint64_t{kTotal});
}
OVG_TEST(netstack_tcp_connect_refused) {
Fixture fx;
TcpPeer peer;
peer.link = &fx.link();
peer.refuse = true;
fx.link().on_tx = [&](const std::vector<uint8_t> &p) { peer.on_tx(p); };
auto netif = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(netif != nullptr);
std::error_code ec;
auto stream = fx.connect(netif, "93.184.216.34", 81, Millis{3000}, &ec);
CHECK(stream == nullptr);
CHECK_EQ(ec, make_error_code(Error::ConnectionRefused));
CHECK_EQ(netif->stats().tcp_failed, uint64_t{1});
CHECK_EQ(netif->stats().tcp_opened, uint64_t{0});
}
// A connect whose SYN is never answered must fail with Timeout on our own
// deadline, well before lwIP's SYN retransmit budget runs out.
OVG_TEST(netstack_tcp_connect_timeout) {
Fixture fx; // no peer attached: packets go nowhere
auto netif = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(netif != nullptr);
std::error_code ec;
auto stream = fx.connect(netif, "93.184.216.34", 80, Millis{150}, &ec);
CHECK(stream == nullptr);
CHECK_EQ(ec, make_error_code(Error::Timeout));
CHECK_GT(fx.link().tx_packets, size_t{0});
}
// Tearing the netif down under a live connection must abort it, not leave a PCB
// pointing at a removed interface.
OVG_TEST(netstack_shutdown_aborts_live_streams) {
Fixture fx;
TcpPeer peer;
peer.link = &fx.link();
fx.link().on_tx = [&](const std::vector<uint8_t> &p) { peer.on_tx(p); };
auto netif = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(netif != nullptr);
auto stream = fx.connect(netif, "93.184.216.34", 80, Millis{3000});
CHECK(stream != nullptr);
CHECK(stream->is_open());
std::vector<char> rbuf(64);
std::error_code read_ec;
bool read_done = false;
stream->async_read_some(asio::buffer(rbuf),
[&](const std::error_code &ec, size_t n) {
read_ec = ec;
(void)n;
read_done = true;
});
netif->shutdown();
CHECK(fx.run_until([&] { return read_done; }, std::chrono::seconds(3)));
CHECK_EQ(read_ec, make_error_code(Error::EgressGone));
CHECK(!stream->is_open());
// And the stream stays usable-as-an-object afterwards. This is not a detail:
// abort_from_netif() drops the stream's reference to the netif so the netif
// can finish dying, while the caller still holds the stream -- and the caller
// is a SOCKS5 relay whose next move on a failed read is to close both sides.
// Every method here has to answer rather than reach through the netif it no
// longer has.
std::error_code late_read_ec, late_write_ec;
bool late_read = false, late_write = false;
stream->shutdown_send();
stream->async_read_some(asio::buffer(rbuf),
[&](const std::error_code &ec, size_t) {
late_read_ec = ec;
late_read = true;
});
stream->async_write(asio::buffer("x", 1),
[&](const std::error_code &ec, size_t) {
late_write_ec = ec;
late_write = true;
});
stream->close();
stream->close(); // idempotent
CHECK(fx.run_until([&] { return late_read && late_write; },
std::chrono::seconds(3)));
CHECK_NE(late_read_ec, std::error_code{});
CHECK_NE(late_write_ec, std::error_code{});
}
// The same contract on the UDP side, which releases its netif the same way.
OVG_TEST(netstack_shutdown_aborts_live_udp_sockets) {
Fixture fx;
UdpPeer peer;
peer.link = &fx.link();
fx.link().on_tx = [&](const std::vector<uint8_t> &p) { peer.on_tx(p); };
auto netif = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(netif != nullptr);
auto sock = fx.open_udp(netif);
CHECK(sock != nullptr);
std::vector<char> rbuf(512);
std::error_code recv_ec;
bool received = false;
sock->async_receive_from(
asio::buffer(rbuf),
[&](const std::error_code &ec, size_t, const Endpoint &) {
recv_ec = ec;
received = true;
});
netif->shutdown();
CHECK(fx.run_until([&] { return received; }, std::chrono::seconds(3)));
CHECK_EQ(recv_ec, make_error_code(Error::EgressGone));
CHECK(!sock->is_open());
std::error_code late_send_ec;
bool late_send = false;
sock->async_send_to(asio::buffer("x", 1),
Endpoint(*IpAddress::parse("8.8.8.8"), 53),
[&](const std::error_code &ec, size_t) {
late_send_ec = ec;
late_send = true;
});
sock->close();
sock->close();
CHECK(fx.run_until([&] { return late_send; }, std::chrono::seconds(3)));
CHECK_NE(late_send_ec, std::error_code{});
}
// ---------------------------------------------------------------------------
// UDP
// ---------------------------------------------------------------------------
OVG_TEST(netstack_udp_round_trip) {
Fixture fx;
UdpPeer peer;
peer.link = &fx.link();
fx.link().on_tx = [&](const std::vector<uint8_t> &p) { peer.on_tx(p); };
auto netif = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(netif != nullptr);
auto sock = fx.open_udp(netif);
CHECK(sock != nullptr);
CHECK(sock->is_open());
CHECK_EQ(sock->local_endpoint().address().to_string(),
std::string("10.0.0.2"));
CHECK_NE(sock->local_endpoint().port(), uint16_t{0});
CHECK_EQ(netif->stats().udp_active, int64_t{1});
const Endpoint server(*IpAddress::parse("8.8.8.8"), 53);
const std::string msg = "\x12\x34 hello over udp";
std::vector<char> rbuf(512);
std::error_code recv_ec;
size_t got = 0;
Endpoint from;
bool received = false;
sock->async_receive_from(
asio::buffer(rbuf),
[&](const std::error_code &ec, size_t n, const Endpoint &f) {
recv_ec = ec;
got = n;
from = f;
received = true;
});
std::error_code send_ec;
bool sent = false;
sock->async_send_to(asio::buffer(msg), server,
[&](const std::error_code &ec, size_t n) {
send_ec = ec;
(void)n;
sent = true;
});
CHECK(fx.run_until([&] { return sent; }, std::chrono::seconds(3)));
CHECK_EQ(send_ec, std::error_code{});
CHECK(fx.run_until([&] { return received; }, std::chrono::seconds(3)));
CHECK_EQ(recv_ec, std::error_code{});
CHECK_EQ(std::string(rbuf.data(), got), msg);
// Inbound datagrams must carry the real sender, not the socket's peer: SOCKS5
// UDP ASSOCIATE has to put it back in the reply header.
CHECK_EQ(from.to_string(), std::string("8.8.8.8:53"));
CHECK_EQ(peer.received.size(), size_t{1});
sock->close();
fx.settle();
CHECK(!sock->is_open());
CHECK_EQ(netif->stats().udp_active, int64_t{0});
}
// A datagram larger than the caller's buffer is truncated, not an error, and
// not a buffer overrun -- the recvfrom(2) contract the header promises.
OVG_TEST(netstack_udp_truncates_oversized_datagrams) {
Fixture fx;
UdpPeer peer;
peer.link = &fx.link();
fx.link().on_tx = [&](const std::vector<uint8_t> &p) { peer.on_tx(p); };
auto netif = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(netif != nullptr);
auto sock = fx.open_udp(netif);
CHECK(sock != nullptr);
const std::string msg(600, 'x');
std::vector<char> rbuf(64);
size_t got = 0;
std::error_code recv_ec;
bool received = false;
sock->async_receive_from(
asio::buffer(rbuf),
[&](const std::error_code &ec, size_t n, const Endpoint &) {
recv_ec = ec;
got = n;
received = true;
});
bool sent = false;
sock->async_send_to(asio::buffer(msg),
Endpoint(*IpAddress::parse("8.8.8.8"), 53),
[&](const std::error_code &, size_t) { sent = true; });
CHECK(fx.run_until([&] { return sent && received; },
std::chrono::seconds(3)));
CHECK_EQ(recv_ec, std::error_code{});
CHECK_EQ(got, size_t{64});
CHECK_EQ(std::string(rbuf.data(), got), msg.substr(0, 64));
}
// Resolving is the caller's job, so that it happens through the same egress the
// datagram will take (stream.h). A name here is a programming error, not
// something to quietly look up.
OVG_TEST(netstack_udp_rejects_unresolved_destinations) {
Fixture fx;
auto netif = fx.attach(test_cfg("10.0.0.2", "node-a"));
CHECK(netif != nullptr);
auto sock = fx.open_udp(netif);
CHECK(sock != nullptr);
std::error_code ec;
bool done = false;
sock->async_send_to(asio::buffer("x", 1), Endpoint("example.com", 53),
[&](const std::error_code &e, size_t) {
ec = e;
done = true;
});
CHECK(fx.run_until([&] { return done; }, std::chrono::seconds(3)));
CHECK_EQ(ec, make_error_code(Error::NotSupported));
}
// ---------------------------------------------------------------------------
// DNS wire format
// ---------------------------------------------------------------------------
OVG_TEST(dns_build_query_shape) {
std::vector<uint8_t> q;
CHECK(dns::build_query("www.example.com", 0xBEEF, &q));
CHECK_EQ(q.size(), size_t{12 + 17 + 4});
CHECK_EQ(int(q[0]), 0xBE);
CHECK_EQ(int(q[1]), 0xEF);
CHECK_EQ(int(q[2]), 0x01); // RD
CHECK_EQ(int(q[5]), 1); // QDCOUNT
CHECK_EQ(int(q[12]), 3);
CHECK_EQ(std::string(reinterpret_cast<const char *>(&q[13]), 3),
std::string("www"));
CHECK_EQ(int(q[q.size() - 3]), 1); // QTYPE = A
CHECK_EQ(int(q[q.size() - 1]), 1); // QCLASS = IN
// A trailing root dot is implied by the encoding, not an extra label.
std::vector<uint8_t> q2;
CHECK(dns::build_query("www.example.com.", 1, &q2));
CHECK_EQ(q2.size(), q.size());
}
OVG_TEST(dns_build_query_rejects_bad_names) {
std::vector<uint8_t> q;
CHECK(!dns::build_query("", 1, &q));
CHECK(!dns::build_query(".", 1, &q));
CHECK(!dns::build_query("a..b", 1, &q));
CHECK(!dns::build_query(std::string(64, 'a') + ".com", 1, &q));
CHECK(!dns::build_query(std::string(300, 'a'), 1, &q));
}
namespace {
// Appends a compressed name pointing at offset 12 (the question's QNAME).
void append_ptr_name(std::vector<uint8_t> &v) {
v.push_back(0xC0);
v.push_back(0x0C);
}
void append_a_record(std::vector<uint8_t> &v, uint32_t ttl, const uint8_t ip4[4]) {
append_ptr_name(v);
v.insert(v.end(), {0x00, 0x01, 0x00, 0x01}); // A, IN
v.push_back(uint8_t(ttl >> 24));
v.push_back(uint8_t(ttl >> 16));
v.push_back(uint8_t(ttl >> 8));
v.push_back(uint8_t(ttl));
v.insert(v.end(), {0x00, 0x04});
v.insert(v.end(), ip4, ip4 + 4);
}
} // namespace
OVG_TEST(dns_parse_response_extracts_a_records) {
std::vector<uint8_t> r;
CHECK(dns::build_query("example.com", 0x1234, &r));
r[2] = 0x81; // QR + RD
r[3] = 0x80; // RA, rcode 0
r[7] = 3; // ANCOUNT
const uint8_t a1[4] = {93, 184, 216, 34};
const uint8_t a2[4] = {1, 2, 3, 4};
append_a_record(r, 300, a1);
// A CNAME in the middle must be stepped over, not mistaken for an address.
append_ptr_name(r);
r.insert(r.end(), {0x00, 0x05, 0x00, 0x01, 0, 0, 1, 44, 0x00, 0x02, 0xC0, 0x0C});
append_a_record(r, 60, a2);
dns::ParseResult out;
CHECK(dns::parse_response(r.data(), r.size(), &out));
CHECK_EQ(out.id, uint16_t{0x1234});
CHECK_EQ(out.rcode, 0);
CHECK_EQ(out.addrs.size(), size_t{2});
CHECK_EQ(out.addrs[0].to_string(), std::string("93.184.216.34"));
CHECK_EQ(out.addrs[1].to_string(), std::string("1.2.3.4"));
// The shortest TTL in the set governs the whole set.
CHECK_EQ(out.min_ttl, uint32_t{60});
}
OVG_TEST(dns_parse_response_reports_rcode) {
std::vector<uint8_t> r;
CHECK(dns::build_query("nope.example", 7, &r));
r[2] = 0x81;
r[3] = 0x83; // NXDOMAIN
dns::ParseResult out;
CHECK(dns::parse_response(r.data(), r.size(), &out));
CHECK_EQ(out.rcode, 3);
CHECK(out.addrs.empty());
}
OVG_TEST(dns_parse_response_rejects_malformed) {
dns::ParseResult out;
const uint8_t tiny[4] = {0, 1, 2, 3};
CHECK(!dns::parse_response(tiny, sizeof(tiny), &out));
// A well-formed header claiming an answer that is not there.
std::vector<uint8_t> r;
CHECK(dns::build_query("example.com", 1, &r));
r[2] = 0x81;
r[7] = 1; // ANCOUNT = 1, but no answer follows
CHECK(!dns::parse_response(r.data(), r.size(), &out));
// A query, not a response.
std::vector<uint8_t> q;
CHECK(dns::build_query("example.com", 1, &q));
CHECK(!dns::parse_response(q.data(), q.size(), &out));
}
// Regression, and the bug that hid behind every green test in this file: the
// periodic timer re-arms itself for as long as the Stack is alive, so it is
// outstanding io_context work that no work-guard release can retire. Tests
// never noticed because they drive the context in slices and stop caring; a
// service calls run() once and waits for it to return. It waited forever --
// after logging a complete, correct graceful shutdown. `direct` mode, which
// builds no stack at all, exited fine, which is exactly why this went unseen.
OVG_TEST(stack_stop_lets_the_io_context_drain) {
asio::io_context io;
Stack stack{io};
// The service's work guard, released the way shutdown releases it. On its own
// this settles nothing: the timer is not a guard, it is real pending work.
auto work = asio::make_work_guard(io);
work.reset();
const auto t0 = Clock::now();
io.run_for(std::chrono::milliseconds(120));
// Still ticking, so the slice was used up rather than returning early.
CHECK(Clock::now() - t0 >= std::chrono::milliseconds(100));
stack.stop();
const auto t1 = Clock::now();
io.run_for(std::chrono::seconds(3));
const auto took = Clock::now() - t1;
CHECK(took < std::chrono::seconds(1));
CHECK(io.stopped());
// Idempotent, and safe after the context has drained: the post has nowhere to
// run, which must not be a problem for a shutdown path that may be reached
// twice.
stack.stop();
}
View Git Blame Copy Permalink