mirror of https://github.com/oxen-io/lokinet
QUIC lokinet integration refactor
Refactors how quic packets get handled: the actual tunnels now live in tunnel.hpp's TunnelManager which holds and manages all the quic<->tcp tunnelling. service::Endpoint now holds a TunnelManager rather than a quic::Server. We only need one quic server, but we need a separate quic client instance per outgoing quic tunnel, and TunnelManager handles all that glue now. Adds QUIC packet handling to get to the right tunnel code. This required multiplexing incoming quic packets, as follows: Adds a very small quic tunnel packet header of 4 bytes: [1, SPORT, ECN] for client->server packets, where SPORT is our source "port" (really: just a uint16_t unique quic instance identifier) or [2, DPORT, ECN] for server->client packets where the DPORT is the SPORT from above. (This also reworks ECN bits to get properly carried over lokinet.) We don't need a destination/source port for the server-side because there is only ever one quic server (and we know we're going to it when the first byte of the header is 1). Removes the config option for quic exposing ports; a full lokinet will simply accept anything incoming on quic and tunnel it to the requested port on the the local endpoint IP (this handler will come in a following commit). Replace ConvoTags with full addresses: we need to carry the port, as well, which the ConvoTag can't give us, so change those to more general SockAddrs from which we can extract both the ConvoTag *and* the port. Add a pending connection queue along with new quic-side handlers to call when a stream becomes available (TunnelManager uses this to wire up pending incoming conns with quic streams as streams open up). Completely get rid of tunnel_server/tunnel_client.cpp code; it is now moved to tunnel.hpp. Add listen()/forget() methods in TunnelManager for setting up quic listening sockets (for liblokinet usage). Add open()/close() methods in TunnelManager for spinning up new quic clients for outgoing quic connections.pull/1576/head
parent
158ea4a951
commit
752879d712
@ -1,143 +0,0 @@
|
||||
#include "connection.hpp"
|
||||
#include "client.hpp"
|
||||
#include "stream.hpp"
|
||||
#include "tunnel.hpp"
|
||||
#include <llarp/util/logging/logger.hpp>
|
||||
|
||||
#include <oxenmq/hex.h>
|
||||
|
||||
#include <util/str.hpp>
|
||||
|
||||
#include <charconv>
|
||||
#include <iterator>
|
||||
|
||||
#include <uvw.hpp>
|
||||
|
||||
/*
|
||||
using namespace std::literals;
|
||||
|
||||
namespace llarp::quic::tunnel
|
||||
{
|
||||
// When we receive a new incoming connection we immediately initiate a new quic stream. This quic
|
||||
// stream in turn causes the other end to initiate a TCP connection on whatever port we specified
|
||||
// in the connection; if the connection is established, it sends back a single byte 0x00
|
||||
// (CONNECT_INIT); otherwise it shuts down the stream with an error code.
|
||||
void
|
||||
on_new_connection(const uvw::ListenEvent&, uvw::TCPHandle& server)
|
||||
{
|
||||
LogDebug("New connection!\n");
|
||||
auto client = server.loop().resource<uvw::TCPHandle>();
|
||||
server.accept(*client);
|
||||
|
||||
auto conn = server.data<llarp::quic::Connection>();
|
||||
std::shared_ptr<llarp::quic::Stream> stream;
|
||||
try
|
||||
{
|
||||
LogTrace("open stream");
|
||||
stream = conn->open_stream(
|
||||
[client](llarp::quic::Stream& stream, llarp::quic::bstring_view bdata) {
|
||||
if (bdata.empty())
|
||||
return;
|
||||
if (auto b0 = bdata[0]; b0 == tunnel::CONNECT_INIT)
|
||||
{
|
||||
// Set up callbacks, which replaces both of these initial callbacks
|
||||
client->read();
|
||||
tunnel::install_stream_forwarding(*client, stream);
|
||||
|
||||
if (bdata.size() > 1)
|
||||
{
|
||||
bdata.remove_prefix(1);
|
||||
stream.data_callback(stream, std::move(bdata));
|
||||
}
|
||||
LogTrace("starting client reading");
|
||||
}
|
||||
else
|
||||
{
|
||||
LogWarn(
|
||||
"Remote connection returned invalid initial byte (0x",
|
||||
oxenmq::to_hex(bdata.begin(), bdata.begin() + 1),
|
||||
"); dropping connection");
|
||||
client->closeReset();
|
||||
stream.close(tunnel::ERROR_BAD_INIT);
|
||||
}
|
||||
stream.io_ready();
|
||||
},
|
||||
[client](llarp::quic::Stream&, std::optional<uint64_t> error_code) mutable {
|
||||
if (error_code && *error_code == tunnel::ERROR_CONNECT)
|
||||
LogDebug("Remote TCP connection failed, closing local connection");
|
||||
else
|
||||
LogWarn(
|
||||
"Stream connection closed ",
|
||||
error_code ? "with error " + std::to_string(*error_code) : "gracefully",
|
||||
"; closing local TCP connection.");
|
||||
auto peer = client->peer();
|
||||
LogDebug("Closing connection to ", peer.ip, ":", peer.port);
|
||||
if (error_code)
|
||||
client->closeReset();
|
||||
else
|
||||
client->close();
|
||||
});
|
||||
stream->io_ready();
|
||||
}
|
||||
catch (const std::exception& e)
|
||||
{
|
||||
LogDebug("open stream failed");
|
||||
client->closeReset();
|
||||
return;
|
||||
}
|
||||
|
||||
LogTrace("done stream setup");
|
||||
conn->io_ready();
|
||||
}
|
||||
|
||||
int
|
||||
usage(std::string_view arg0, std::string_view msg)
|
||||
{
|
||||
std::cerr << msg << "\n\n"
|
||||
<< "Usage: " << arg0
|
||||
<< " [DESTPORT [SERVERPORT [LISTENPORT]]]\n\nDefaults to ports 4444 4242 5555\n";
|
||||
return 1;
|
||||
}
|
||||
|
||||
int
|
||||
main(int argc, char* argv[])
|
||||
{
|
||||
auto loop = uvw::Loop::create();
|
||||
|
||||
std::array<uint16_t, 3> ports{{4444, 4242, 5555}};
|
||||
for (size_t i = 0; i < ports.size(); i++)
|
||||
{
|
||||
if (argc < 2 + (int)i)
|
||||
break;
|
||||
if (!parse_int(argv[1 + i], ports[i]))
|
||||
return usage(argv[0], "Invalid port "s + argv[1 + i]);
|
||||
}
|
||||
auto& [dest_port, server_port, listen_port] = ports;
|
||||
std::cout << "Connecting to quic server at localhost:" << server_port
|
||||
<< " to reach tunneled port " << dest_port
|
||||
<< ", listening on localhost:" << listen_port << "\n";
|
||||
|
||||
signal(SIGPIPE, SIG_IGN);
|
||||
|
||||
LogDebug("Initializing client");
|
||||
auto tunnel_client = std::make_shared<llarp::quic::Client>(
|
||||
llarp::quic::Address{{127, 0, 0, 1}, server_port}, // server addr
|
||||
loop,
|
||||
dest_port // tunnel destination port
|
||||
);
|
||||
tunnel_client->default_stream_buffer_size = 0; // We steal uvw's provided buffers
|
||||
LogDebug("Initialized client");
|
||||
|
||||
// Start listening for TCP connections:
|
||||
auto server = loop->resource<uvw::TCPHandle>();
|
||||
server->data(tunnel_client->get_connection());
|
||||
server->on<uvw::ListenEvent>(llarp::quic::tunnel::on_new_connection);
|
||||
|
||||
server->bind("127.0.0.1", listen_port);
|
||||
server->listen();
|
||||
|
||||
loop->run();
|
||||
}
|
||||
|
||||
} // namespace llarp::quic::tunnel
|
||||
*/
|
@ -1,158 +0,0 @@
|
||||
#include "tunnel_server.hpp"
|
||||
#include "tunnel.hpp"
|
||||
#include "connection.hpp"
|
||||
#include "server.hpp"
|
||||
#include <llarp/util/logging/logger.hpp>
|
||||
|
||||
#include <util/str.hpp>
|
||||
|
||||
#include <uvw/tcp.h>
|
||||
|
||||
/*
|
||||
using namespace std::literals;
|
||||
|
||||
namespace llarp::quic::tunnel
|
||||
{
|
||||
IncomingTunnel::IncomingTunnel(uint16_t localhost_port)
|
||||
: IncomingTunnel{
|
||||
[localhost_port](
|
||||
[[maybe_unused]] const auto& remote, uint16_t port, SockAddr& connect_to) {
|
||||
if (port != localhost_port)
|
||||
return AcceptResult::DECLINE;
|
||||
connect_to.setIPv4(127, 0, 0, 1);
|
||||
connect_to.setPort(port);
|
||||
return AcceptResult::ACCEPT;
|
||||
}}
|
||||
{}
|
||||
|
||||
int
|
||||
usage(std::string_view arg0, std::string_view msg)
|
||||
{
|
||||
std::cerr << msg << "\n\n"
|
||||
<< "Usage: " << arg0
|
||||
<< " [LISTENPORT [ALLOWED ...]]\n\nDefaults to listening on 4242 and allowing "
|
||||
"22,80,4444,8080\n";
|
||||
return 1;
|
||||
}
|
||||
|
||||
int
|
||||
main(int argc, char* argv[])
|
||||
{
|
||||
uint16_t listen_port = 4242;
|
||||
std::set<uint16_t> allowed_ports{{22, 80, 4444, 8080}};
|
||||
|
||||
if (argc >= 2 && !parse_int(argv[1], listen_port))
|
||||
return usage(argv[0], "Invalid port "s + argv[1]);
|
||||
if (argc >= 3)
|
||||
{
|
||||
allowed_ports.clear();
|
||||
for (int i = 2; i < argc; i++)
|
||||
{
|
||||
if (argv[i] == "all"sv)
|
||||
{
|
||||
allowed_ports.clear();
|
||||
break;
|
||||
}
|
||||
uint16_t port;
|
||||
if (!parse_int(argv[i], port))
|
||||
return usage(argv[0], "Invalid port "s + argv[i]);
|
||||
allowed_ports.insert(port);
|
||||
}
|
||||
}
|
||||
|
||||
auto loop = uvw::Loop::create();
|
||||
|
||||
Address listen_addr{{0, 0, 0, 0}, listen_port};
|
||||
|
||||
signal(SIGPIPE, SIG_IGN);
|
||||
|
||||
// The local address we connect to for incoming connections. (localhost for this demo, should
|
||||
// be the localhost.loki address for lokinet).
|
||||
std::string localhost = "127.0.0.1";
|
||||
|
||||
LogInfo("Initializing QUIC server");
|
||||
llarp::quic::Server s{
|
||||
listen_addr,
|
||||
loop,
|
||||
[loop, localhost, allowed_ports](
|
||||
llarp::quic::Server&, llarp::quic::Stream& stream, uint16_t port) {
|
||||
LogDebug("New incoming quic stream ", stream.id(), " to reach ", localhost, ":", port);
|
||||
if (port == 0 || !(allowed_ports.empty() || allowed_ports.count(port)))
|
||||
{
|
||||
LogWarn(
|
||||
"quic stream denied by configuration: ", port, " is not a permitted local port");
|
||||
return false;
|
||||
}
|
||||
|
||||
stream.close_callback = [](llarp::quic::Stream& strm,
|
||||
std::optional<uint64_t> error_code) {
|
||||
LogDebug(
|
||||
error_code ? "Remote side" : "We",
|
||||
" closed the quic stream, closing localhost tcp connection");
|
||||
if (error_code && *error_code > 0)
|
||||
LogWarn("Remote quic stream was closed with error code ", *error_code);
|
||||
auto tcp = strm.data<uvw::TCPHandle>();
|
||||
if (!tcp)
|
||||
LogDebug("Local TCP connection already closed");
|
||||
else
|
||||
tcp->close();
|
||||
};
|
||||
// Try to open a TCP connection to the configured localhost port; if we establish a
|
||||
// connection then we immediately send a CONNECT_INIT back down the stream; if we fail
|
||||
// then we send a fail-to-connect error code. Once we successfully connect both of
|
||||
// these handlers get replaced with the normal tunnel handlers.
|
||||
auto tcp = loop->resource<uvw::TCPHandle>();
|
||||
auto error_handler = tcp->once<uvw::ErrorEvent>(
|
||||
[&stream, localhost, port](const uvw::ErrorEvent&, uvw::TCPHandle&) {
|
||||
LogWarn(
|
||||
"Failed to connect to ", localhost, ":", port, ", shutting down quic stream");
|
||||
stream.close(tunnel::ERROR_CONNECT);
|
||||
});
|
||||
tcp->once<uvw::ConnectEvent>(
|
||||
[streamw = stream.weak_from_this(), error_handler = std::move(error_handler)](
|
||||
const uvw::ConnectEvent&, uvw::TCPHandle& tcp) {
|
||||
auto peer = tcp.peer();
|
||||
auto stream = streamw.lock();
|
||||
if (!stream)
|
||||
{
|
||||
LogWarn(
|
||||
"Connected to ",
|
||||
peer.ip,
|
||||
":",
|
||||
peer.port,
|
||||
" but quic stream has gone away; resetting local connection");
|
||||
tcp.closeReset();
|
||||
return;
|
||||
}
|
||||
LogDebug("Connected to ", peer.ip, ":", peer.port, " for quic ", stream->id());
|
||||
tcp.erase(error_handler);
|
||||
tunnel::install_stream_forwarding(tcp, *stream);
|
||||
assert(stream->used() == 0);
|
||||
|
||||
stream->append_buffer(new std::byte[1]{tunnel::CONNECT_INIT}, 1);
|
||||
tcp.read();
|
||||
});
|
||||
|
||||
// FIXME, need to configure this
|
||||
tcp->connect("127.0.0.1", port);
|
||||
|
||||
return true;
|
||||
}};
|
||||
s.default_stream_buffer_size = 0; // We steal uvw's provided buffers
|
||||
LogDebug("Initialized server");
|
||||
std::cout << "Listening on localhost:" << listen_port
|
||||
<< " with tunnel(s) to localhost port(s):";
|
||||
if (allowed_ports.empty())
|
||||
std::cout << " (any)";
|
||||
for (auto p : allowed_ports)
|
||||
std::cout << ' ' << p;
|
||||
std::cout << '\n';
|
||||
|
||||
loop->run();
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
} // namespace llarp::quic::tunnel
|
||||
|
||||
*/
|
@ -1,80 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include "address.hpp"
|
||||
#include <llarp/net/sock_addr.hpp>
|
||||
#include <llarp/ev/ev.hpp>
|
||||
|
||||
#include <memory>
|
||||
|
||||
namespace llarp::quic::tunnel
|
||||
{
|
||||
enum class AcceptResult : int
|
||||
{
|
||||
ACCEPT = 0, // Accepts a connection
|
||||
DECLINE = -1, // Declines a connection (try other callbacks, refuse if all decline)
|
||||
REFUSE = -2, // Refuses a connection (don't try any more callbacks)
|
||||
};
|
||||
|
||||
// Class that wraps an incoming connection acceptance callback (to allow for callback removal).
|
||||
// This is not directly constructible: you must construct it via the TunnelServer instance.
|
||||
class IncomingTunnel final
|
||||
{
|
||||
public:
|
||||
using AcceptCallback = std::function<AcceptResult(
|
||||
const Address& remote, uint16_t port, llarp::SockAddr& connect_to)>;
|
||||
|
||||
private:
|
||||
AcceptCallback accept;
|
||||
|
||||
friend class TunnelServer;
|
||||
|
||||
// Constructor with a full callback; invoked via TunnelServer::add_incoming_tunnel
|
||||
explicit IncomingTunnel(AcceptCallback accept) : accept{std::move(accept)}
|
||||
{}
|
||||
|
||||
// Constructor for a simple forwarding to a single localhost port. E.g. IncomingTunnel(22)
|
||||
// allows incoming connections to reach port 22 and forwards them to localhost:22.
|
||||
explicit IncomingTunnel(uint16_t localhost_port);
|
||||
|
||||
// Constructor for forwarding everything to the same port; this is used by full clients by
|
||||
// default.
|
||||
IncomingTunnel();
|
||||
};
|
||||
|
||||
// Class that handles incoming quic connections. This class sets itself up in the llarp event
|
||||
// loop on construction and maintains a list of incoming acceptor callbacks. When a new incoming
|
||||
// quic connections is being established we try the callbacks one by one to determine the local
|
||||
// TCP port the tunnel should be connected to until:
|
||||
// - a callback sets connect_to and returns AcceptResult::ACCEPT - we connect it to the returned
|
||||
// address
|
||||
// - a callback returns AcceptResult::REFUSE - we reject the connection
|
||||
//
|
||||
// If a callback returns AcceptResult::DECLINE then we skip that callback and try the next one; if
|
||||
// all callbacks decline (or we have no callbacks at all) then we reject the connection.
|
||||
//
|
||||
// Note that tunnel operations and initialization are done in the event loop thread and so will
|
||||
// not take effect until the next event loop tick when called from some other thread.
|
||||
class TunnelServer : public std::enable_shared_from_this<TunnelServer>
|
||||
{
|
||||
public:
|
||||
explicit TunnelServer(EventLoop_ptr ev);
|
||||
|
||||
// Appends a new tunnel to the end of the queue; all arguments are forwarded to private
|
||||
// constructor(s) of IncomingTunnel.
|
||||
template <typename... Args>
|
||||
std::shared_ptr<IncomingTunnel>
|
||||
add_incoming_tunnel(Args&&... args)
|
||||
{
|
||||
return std::shared_ptr<IncomingTunnel>{new IncomingTunnel{std::forward<Args>(args)...}};
|
||||
}
|
||||
|
||||
// Removes a tunnel acceptor from the acceptor queue.
|
||||
void
|
||||
remove_incoming_tunnel(std::weak_ptr<IncomingTunnel> tunnel);
|
||||
|
||||
private:
|
||||
EventLoop_ptr ev;
|
||||
std::vector<std::shared_ptr<IncomingTunnel>> tunnels;
|
||||
};
|
||||
|
||||
} // namespace llarp::quic::tunnel
|
Loading…
Reference in New Issue