#include "vpn/internal/tunnel.h"

#include <algorithm>

#include <atomic>

#include <bitset>

#include <cassert>

#include <optional>

#include <utility>

#include <vector>

#include <magic_enum/magic_enum.hpp>

#include "common/net_utils.h"

#include "connection_statistics.h"

#include "dns_handler.h"

#include "fake_upstream.h"

#include "net/quic_utils.h"

#include "net/utils.h"

#include "socks_listener.h"

#include "vpn/internal/utils.h"

#include "vpn/internal/vpn_client.h"

#include "vpn/utils.h"

#define log_tun(tun_, lvl_, fmt_, ...) lvl_##log((tun_)->log, "[{}] " fmt_, (tun_)->id, ##__VA_ARGS__)

#define log_conn(tun_, conn_, lvl_, fmt_, ...) \

lvl_##log((tun_)->log, "[{}] [L:{}-R:{}] " fmt_, (tun_)->id, (int64_t) (conn_)->client_id, \

(int64_t) (conn_)->server_id, ##__VA_ARGS__)

// TODO: either remove quic blocking code or fix it to correctly tell QUIC libraries that connection is unreachable.

#define BLOCK_OLD_QUIC_CONNS 0

namespace ag {

static std::atomic_int g_next_tunnel_id = 0; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)

struct CompleteConnectRequestCtx {

Tunnel *tunnel = nullptr;

uint64_t listener_conn_id = NON_ID;

int err_code = -1;

};

// Func: void(VpnConnection *)

// Call `func` for each value in `table`.

template <typename Func>

void vpn_connections_foreach(const khash_t(connections_by_id) * table, Func &&func) {

auto f = std::forward<Func>(func);

for (auto it = kh_begin(table); it != kh_end(table); ++it) {

if (!kh_exist(table, it)) {

continue;

}

VpnConnection *conn = kh_value(table, it);

f(conn);

}

}

static VpnConnection *vpn_connection_get_by_id(const khash_t(connections_by_id) * table, uint64_t id) {

VpnConnection *conn = nullptr;

khiter_t iter = kh_get(connections_by_id, table, id);

if (iter != kh_end(table)) {

conn = kh_value(table, iter);

}

return conn;

}

static void vpn_connection_put(khash_t(connections_by_id) * table, uint64_t id, VpnConnection *conn) {

int r = 0;

khiter_t iter = kh_put(connections_by_id, table, id, &r);

assert(r >= 0);

if (r >= 0) {

kh_value(table, iter) = conn;

}

}

static void vpn_connection_remove(khash_t(connections_by_id) * table, uint64_t id) {

khiter_t iter = kh_get(connections_by_id, table, id);

if (iter != kh_end(table)) {

kh_del(connections_by_id, table, iter);

}

}

static void add_connection(Tunnel *tunnel, VpnConnection *conn) {

assert(conn->client_id != NON_ID || conn->server_id != NON_ID);

if (conn->client_id != NON_ID) {

vpn_connection_put(tunnel->connections.by_client_id, conn->client_id, conn);

}

if (conn->server_id != NON_ID) {

vpn_connection_put(tunnel->connections.by_server_id, conn->server_id, conn);

}

}

static void cancel_destination_resolve(Tunnel *self, VpnConnection *conn) {

auto it = std::find_if(

self->dns_resolve_waiters.begin(), self->dns_resolve_waiters.end(), [id = conn->client_id](const auto &i) {

return i.second.conn_client_id == id;

});

if (it != self->dns_resolve_waiters.end()) {

self->dns_resolver->cancel(it->first);

self->dns_resolve_waiters.erase(it);

}

}

static void destroy_connection(Tunnel *tunnel, uint64_t client_id, uint64_t server_id) {

assert(client_id != NON_ID || server_id != NON_ID);

VpnConnection *conn = nullptr;

if (client_id != NON_ID) {

conn = vpn_connection_get_by_id(tunnel->connections.by_client_id, client_id);

}

if (conn == nullptr && server_id != NON_ID) {

conn = vpn_connection_get_by_id(tunnel->connections.by_server_id, server_id);

}

if (conn != nullptr) {

if (conn->state == CONNS_WAITING_RESOLVE) {

cancel_destination_resolve(tunnel, conn);

}

vpn_connection_remove(tunnel->connections.by_client_id, conn->client_id);

vpn_connection_remove(tunnel->connections.by_server_id, conn->server_id);

if (conn->proto == IPPROTO_UDP && tunnel->udp_close_wait_hostname_cache) {

std::scoped_lock l(tunnel->udp_close_wait_hostname_cache->mtx);

tunnel->udp_close_wait_hostname_cache->val.insert(conn->addr, conn->domain_extractor_result);

}

log_conn(tunnel, conn, dbg, "Destroyed (download={}, upload={})", conn->incoming_bytes, conn->outgoing_bytes);

if (conn->flags.test(CONNF_MONITOR_STATS)) {

tunnel->statistics_monitor->unregister_conn(conn->client_id, /*do_report*/ true);

}

if (!conn->listener.expired() && conn->listener.lock().get() == tunnel->vpn->client_listener.get()) {

VpnTunnelConnectionClosedEvent event{

.id = conn->client_id,

};

const vpn_client::Handler &handler = tunnel->vpn->parameters.handler;

handler.func(handler.arg, vpn_client::EVENT_CONNECTION_CLOSED, &event);

}

assert(nullptr == vpn_connection_get_by_id(tunnel->connections.by_client_id, conn->client_id));

assert(nullptr == vpn_connection_get_by_id(tunnel->connections.by_server_id, conn->server_id));

delete conn;

} else {

log_tun(tunnel, dbg, "Trying to destroy non-existent connection: L:{}-R:{}", client_id, server_id);

return;

}

log_tun(tunnel, dbg, "Remaining connections: client-side={} server-side={}",

(int) kh_size(tunnel->connections.by_client_id), (int) kh_size(tunnel->connections.by_server_id));

}

static ClientConnectResult server_error_to_connect_result(int err) {

switch (err) {

case 0:

return CCR_PASS;

case ag::utils::AG_ETIMEDOUT:

return CCR_DROP;

case AG_ENETUNREACH:

case AG_EHOSTUNREACH:

return CCR_UNREACH;

default:

return CCR_REJECT;

}

}

static void complete_connect_request_task(void *arg, TaskId) {

auto *ctx = (CompleteConnectRequestCtx *) arg;

Tunnel *tunnel = ctx->tunnel;

VpnConnection *conn = vpn_connection_get_by_id(tunnel->connections.by_client_id, ctx->listener_conn_id);

if (conn == nullptr) {

log_tun(tunnel, dbg, "Connection not found: L:{}", ctx->listener_conn_id);

assert(0);

return;

}

conn->complete_connect_request_task.release();

std::shared_ptr<ClientListener> listener = conn->listener.lock();

if (listener == nullptr) {

log_conn(tunnel, conn, dbg, "Listener was deleted");

destroy_connection(tunnel, ctx->listener_conn_id, NON_ID);

return;

}

switch (conn->state) {

case CONNS_WAITING_RESOLVE:

case CONNS_WAITING_RESPONSE:

case CONNS_WAITING_ACTION: {

ClientConnectResult result = server_error_to_connect_result(ctx->err_code);

listener->complete_connect_request(conn->client_id, result);

if (result != CCR_PASS) {

// Relookup connection since complete_connect_request may lead to it's close.

conn = vpn_connection_get_by_id(tunnel->connections.by_client_id, ctx->listener_conn_id);

if (conn != nullptr) {

conn->state = CONNS_REJECTED;

}

}

break;

}

case CONNS_REJECTED:

case CONNS_WAITING_ACCEPT:

case CONNS_CONNECTED:

case CONNS_CONNECTED_MIGRATING:

case CONNS_WAITING_RESPONSE_MIGRATING:

log_conn(tunnel, conn, dbg, "Invalid connection state: {}", magic_enum::enum_name(conn->state));

listener->close_connection(ctx->listener_conn_id, false, false);

assert(0);

break;

}

}

static void close_client_side_connection(Tunnel *self, VpnConnection *conn, int err_code, bool async) {

std::shared_ptr<ClientListener> listener = conn->listener.lock();

if (listener == nullptr) {

log_conn(self, conn, dbg, "Listener was deleted");

destroy_connection(self, conn->client_id, conn->server_id);

return;

}

switch (conn->state) {

case CONNS_WAITING_ACCEPT:

case CONNS_CONNECTED:

case CONNS_CONNECTED_MIGRATING:

case CONNS_REJECTED:

// will be deleted in connection closed event

listener->close_connection(conn->client_id, err_code == 0, async);

break;

case CONNS_WAITING_RESOLVE:

case CONNS_WAITING_RESPONSE:

case CONNS_WAITING_ACTION:

err_code = (err_code == 0) ? ag::utils::AG_ECONNREFUSED : err_code;

if (!async) {

auto result = server_error_to_connect_result(err_code);

uint64_t conn_client_id = conn->client_id;

listener->complete_connect_request(conn->client_id, result);

if (result != CCR_PASS) {

conn = vpn_connection_get_by_id(self->connections.by_client_id, conn_client_id);

if (conn != nullptr) {

conn->state = CONNS_REJECTED;

}

}

break;

}

if (conn->complete_connect_request_task.has_value()) {

break;

}

conn->complete_connect_request_task = event_loop::submit(self->vpn->parameters.ev_loop,

{

new CompleteConnectRequestCtx{self, conn->client_id, err_code},

complete_connect_request_task,

[](void *arg) {

delete (CompleteConnectRequestCtx *) arg;

},

});

break;

case CONNS_WAITING_RESPONSE_MIGRATING:

// do nothing

break;

}

}

static void close_client_side(Tunnel *tunnel, ServerUpstream *upstream) {

if (tunnel->connections.by_client_id == nullptr) {

return;

}

kh_connections_by_id_t *table = tunnel->connections.by_client_id;

// `close_client_side_connection()` might insert new connections into the table,

// leading to undefined behaviour if called within the foreach loop

std::vector<uint64_t> ids;

ids.reserve(kh_size(table));

vpn_connections_foreach(table, [&](VpnConnection *conn) {

if (!conn->upstream.expired() && conn->upstream.lock().get() == upstream) {

ids.push_back(conn->client_id);

}

});

for (uint64_t id : ids) {

if (VpnConnection *conn = vpn_connection_get_by_id(table, id)) {

conn->flags.set(CONNF_SESSION_CLOSED);

close_client_side_connection(tunnel, conn, -1, false);

}

}

}

enum TunnelDomainLookupAction {

TDLA_DONE, // domain lookup has nothing more to do

TDLA_PASS, // domain lookup wants to process the next packet, the current one can be sent

TDLA_DROP, // domain lookup wants to process the next packet, the current one can be dropped

TDLA_SHUTDOWN, // domain lookup realized that connection should've been routed to another upstream

TDLA_BLOCK, // domain lookup realized that connection shouldn't be processed

TDLA_WANT_MORE, // the same as `ALUA_PASS`, but waiting for a packet divided by AntiDpi

};

static std::variant<TunnelDomainLookupAction, DomainExtractorResult> domain_lookup_process_udp(

Tunnel *tunnel, VpnConnection *conn, DomainExtractorPacketDirection dir, const uint8_t *data, size_t length) {

// parse quic header

auto quic_header = ag::quic_utils::parse_quic_header({data, length});

if (quic_header.has_value()) {

if (quic_header->type == ag::quic_utils::ZERO_RTT) {

return TDLA_DROP;

}

if (quic_header->type == ag::quic_utils::INITIAL) {

// quic traffic

auto quic_data = ag::quic_utils::prepare_for_domain_lookup({data, length}, quic_header.value());

if (!quic_data.has_value()) {

// no data for domain lookup

return TDLA_DONE;

}

return conn->domain_extractor.proceed(dir, conn->proto, quic_data->data(), quic_data->size());

}

}

if (tunnel->udp_close_wait_hostname_cache) {

std::scoped_lock l(tunnel->udp_close_wait_hostname_cache->mtx);

if (auto value = tunnel->udp_close_wait_hostname_cache->val.get(conn->addr); value) {

log_conn(tunnel, conn, dbg, "Found cached result for pass_through_lookuper");

return DomainExtractorResult{*value};

}

}

#if BLOCK_OLD_QUIC_CONNS

// get destination port

const SocketAddress *dst = std::get_if<SocketAddress>(&conn->addr.dst);

assert(dst != nullptr);

auto dst_port = dst->port();

// Check if destination is default quic port

if (dst_port == ag::quic_utils::DEFAULT_QUIC_PORT) {

// expected to get quic initial, but got wrong data, block it

return TDLA_BLOCK;

}

#endif

// simple udp traffic

return TDLA_DONE;

}

static TunnelDomainLookupAction pass_through_domain_lookup(

Tunnel *tunnel, VpnConnection *conn, DomainExtractorPacketDirection dir, const uint8_t *data, size_t length) {

DomainExtractor *domain_extractor = &conn->domain_extractor;

DomainExtractorResult r;

TunnelDomainLookupAction action = TDLA_DONE;

DomainFilter *filter = &tunnel->vpn->domain_filter;

if (conn->proto == IPPROTO_UDP) {

auto result = domain_lookup_process_udp(tunnel, conn, dir, data, length);

if (auto *result_action = std::get_if<TunnelDomainLookupAction>(&result)) {

return *result_action;

}

r = std::get<DomainExtractorResult>(result);

} else {

// tcp traffic

r = domain_extractor->proceed(dir, conn->proto, data, length);

}

conn->domain_extractor_result = r;

switch (r.status) {

case DES_NOTFOUND:

log_conn(tunnel, conn, dbg, "Gave up to find domain name");

domain_extractor->reset();

break;

case DES_FOUND: {

log_conn(tunnel, conn, dbg, "Found domain in {} data: {}", (dir == DEPD_OUTGOING) ? "outgoing" : "incoming",

r.domain);

if (DFMS_EXCLUSION == filter->match_domain(r.domain)) {

action = TDLA_SHUTDOWN;

}

filter->add_resolved_tag(conn->make_tag(), std::move(r.domain));

domain_extractor->reset();

break;

}

case DES_PASS:

action = TDLA_PASS;

break;

case DES_WANT_MORE:

action = TDLA_WANT_MORE;

break;

}

return action;

}

static bool is_domain_scannable_port(uint16_t port) {

static constexpr uint16_t SCANNABLE_PORTS[] = {443, 80, 8080, 8008};

return std::any_of(std::begin(SCANNABLE_PORTS), std::end(SCANNABLE_PORTS), [port](uint16_t i) {

return port == i;

});

}

static void dns_resolver_handler(void *arg, ClientEvent what, void *data) {

auto *self = (Tunnel *) arg;

self->listener_handler(self->dns_resolver, what, data);

}

// Send buffered data, if there is any, possibly not completely,

// and turn listener read on/off.

//

// Return `false` on fatal error, `true` otherwise.

//

// Caller should check `conn->buffered_packets` after this function returns

// if handling a listener read event.

static bool send_buffered_data(const Tunnel *self, uint64_t conn_client_id) {

VpnConnection *conn = vpn_connection_get_by_id(self->connections.by_client_id, conn_client_id);

if (conn == nullptr) {

log_tun(self, dbg, "Connection not found: [L:{}]", conn_client_id);

return false;

}

conn->send_buffered_task.release();

std::shared_ptr<ServerUpstream> upstream = conn->upstream.lock();

if (upstream == nullptr) {

log_conn(self, conn, dbg, "Upstream was deleted");

return false;

}

std::shared_ptr<ClientListener> listener = conn->listener.lock();

if (listener == nullptr) {

log_conn(self, conn, dbg, "Listener was deleted");

return false;

}

bool sent_zero_bytes = false;

for (auto it = conn->buffered_packets.begin(); it != conn->buffered_packets.end();) {

auto &packet = *it;

log_conn(self, conn, trace, "Sending {} bytes from buffered packets", packet.size());

ssize_t r = upstream->send(conn->server_id, packet.data(), packet.size());

if (r >= 0 && size_t(r) == packet.size()) {

conn->outgoing_bytes += r;

if (conn->flags.test(CONNF_MONITOR_STATS)) {

self->statistics_monitor->update_upload(conn_client_id, r);

}

it = conn->buffered_packets.erase(it);

continue;

}

if (r < 0) {

log_conn(self, conn, dbg, "Failed to send data: error={}", r);

return false;

}

if (r == 0) {

log_conn(self, conn, dbg, "Sent partially: 0 bytes out of {}, turning off client reads", packet.size());

sent_zero_bytes = true;

break;

}

log_conn(self, conn, dbg, "Sent partially: {} bytes out of {}", r, packet.size());

packet.erase(packet.begin(), packet.begin() + r);

break;

}

size_t server_can_send = upstream->available_to_send(conn->server_id);

log_conn(self, conn, trace, "Can send to server side: {} bytes, upstream sent zero bytes: {}", server_can_send,

sent_zero_bytes);

listener->turn_read(conn->client_id, server_can_send > 0 && !sent_zero_bytes);

upstream->update_flow_control(conn->server_id, listener->flow_control_info(conn_client_id));

return true;

}

static VpnAddress tunnel_to_vpn_address(const TunnelAddress *tunnel);

static VpnFinalConnectionAction get_final_action(VpnMode mode, VpnConnectAction action) {

if (action == VPN_CA_DEFAULT) {

switch (mode) {

case VPN_MODE_GENERAL:

return VpnFinalConnectionAction::VPN_FCA_TUNNEL;

case VPN_MODE_SELECTIVE:

return VpnFinalConnectionAction::VPN_FCA_BYPASS;

}

}

if (action == VPN_CA_FORCE_BYPASS) {

return VpnFinalConnectionAction::VPN_FCA_BYPASS;

}

if (action == VPN_CA_FORCE_REDIRECT) {

return VpnFinalConnectionAction::VPN_FCA_TUNNEL;

}

return VpnFinalConnectionAction::VPN_FCA_REJECT;

}

static void report_connection_info(const Tunnel *self, VpnConnection *conn, const char *domain) {

// Don't report connection info for loopback, DNS, or already reported connections. Also don't report if we are

// looking up the domain.

if (conn->flags.test(CONNF_PLAIN_DNS_CONNECTION) || conn->flags.test(CONNF_LOOKINGUP_DOMAIN)

|| conn->flags.test(CONNF_CONN_INFO_SENT)) {

return;

}

if (const SocketAddress *dst = std::get_if<SocketAddress>(&conn->addr.dst); dst && dst->is_loopback()) {

return;

}

conn->flags.set(CONNF_CONN_INFO_SENT);

VpnConnectionInfoEvent info;

info.src = conn->addr.src.c_storage();

if (const SocketAddress *dst = std::get_if<SocketAddress>(&conn->addr.dst)) {

info.dst = dst->c_storage();

info.domain = domain;

} else {

info.dst = nullptr;

info.domain = std::get<NamePort>(conn->addr.dst).name.c_str();

}

info.proto = conn->proto;

info.action = get_final_action(self->vpn->domain_filter.get_mode(), conn->action.value_or(VPN_CA_DEFAULT));

if (info.domain && info.domain[0] == '\0') {

info.domain = nullptr;

}

self->vpn->parameters.handler.func(self->vpn->parameters.handler.arg, vpn_client::EVENT_CONNECTION_INFO, &info);

}

static std::optional<DnsHandlerParameters> make_dns_handler_parameters(Tunnel *self) {

DnsHandlerParameters parameters{.cert_verify_handler = self->vpn->parameters.cert_verify_handler};

if (self->vpn->listener_config.dns_upstreams.size) {

if (!self->vpn->dns_proxy_listener) {

log_tun(self, warn, "There are user-provided DNS upstreams, but the DNS proxy listener is not initialized");

return std::nullopt;

}

for (size_t i = 0; i < self->vpn->listener_config.dns_upstreams.size; ++i) {

parameters.dns_upstreams.emplace_back(

DnsProxyAccessor::Upstream{.address = self->vpn->listener_config.dns_upstreams.data[i]});

}

parameters.dns_proxy_listener_address =

((SocksListener *) self->vpn->dns_proxy_listener.get())->get_listen_address();

parameters.dns_proxy_listener_username = std::string(vpn_client::DNS_PROXY_LISTENER_USERNAME);

parameters.dns_proxy_listener_password = self->vpn->dns_proxy_listener_password;

}

return parameters;

}

static void schedule_send_buffered_data(Tunnel *tunnel, VpnConnection *conn) {

struct Ctx {

Tunnel *tunnel;

uint64_t conn_client_id;

};

conn->send_buffered_task = event_loop::submit(tunnel->vpn->parameters.ev_loop,

{

new Ctx{tunnel, conn->client_id},

[](void *arg, TaskId) {

auto *ctx = (Ctx *) arg;

send_buffered_data(ctx->tunnel, ctx->conn_client_id);

},

[](void *arg) {

delete (Ctx *) arg;

},

});

}

void Tunnel::upstream_handler(const std::shared_ptr<ServerUpstream> &upstream, ServerEvent what, void *data) {

switch (what) {

case SERVER_EVENT_SESSION_OPENED:

if (upstream.get() == this->vpn->endpoint_upstream.get()) {

assert(!this->endpoint_upstream_connected);

this->endpoint_upstream_connected = true;

}

break;

case SERVER_EVENT_HEALTH_CHECK_ERROR:

// Health check results are handled by `VpnClient`.

break;

case SERVER_EVENT_SESSION_CLOSED:

close_client_side(this, upstream.get());

break;

case SERVER_EVENT_CONNECTION_OPENED: {

uint64_t id = *(uint64_t *) data;

VpnConnection *conn = vpn_connection_get_by_id(this->connections.by_server_id, id);

if (conn == nullptr) {

log_tun(this, warn, "Got server connect result for inexistent or already closed connection: {}", id);

upstream->close_connection(id, false, true);

assert(0);

break;

}

if (upstream.get() != conn->upstream.lock().get()) {

log_conn(this, conn, warn, "Upstream mismatch");

upstream->close_connection(id, false, true);

assert(0);

break;

}

std::shared_ptr<ClientListener> listener = conn->listener.lock();

if (listener == nullptr) {

log_conn(this, conn, dbg, "Listener was deleted");

upstream->close_connection(id, false, true);

assert(0);

break;

}

switch (conn->state) {

case CONNS_WAITING_RESPONSE: {

log_conn(this, conn, dbg, "Successfully made tunnel ({})",

upstream.get() == this->vpn->endpoint_upstream.get() ? "endpoint upstream"

: upstream.get() == this->vpn->bypass_upstream.get() ? "bypass upstream"

: upstream.get() == this->fake_upstream.get() ? "fake upstream"

: upstream.get() == this->dns_handler.get() ? "DNS handler upstream"

: "unknown upstream");

report_connection_info(this, conn, conn->domain_extractor_result.domain.c_str());

conn->state = CONNS_WAITING_ACCEPT;

listener->complete_connect_request(conn->client_id, CCR_PASS);

break;

}

case CONNS_WAITING_RESPONSE_MIGRATING: {

VpnConnection *src_conn =

vpn_connection_get_by_id(this->connections.by_client_id, conn->migrating_client_id);

if (src_conn == nullptr) {

log_conn(this, conn, dbg, "Migrating connection closed while had being connecting to another upstream");

upstream->close_connection(id, false, true);

break;

}

std::swap(conn->client_id, src_conn->client_id);

conn->state = CONNS_CONNECTED;

conn->migrating_client_id = NON_ID;

add_connection(this, conn);

if (conn->proto == IPPROTO_UDP) {

std::swap(((UdpVpnConnection *) conn)->buffered_packets,

((UdpVpnConnection *) src_conn)->buffered_packets);

}

if (std::shared_ptr<ServerUpstream> src_conn_upstream = src_conn->upstream.lock();

src_conn_upstream != nullptr) {

src_conn_upstream->close_connection(src_conn->server_id, false, false);

}

log_conn(this, conn, dbg, "Upstream has been switched successfully");

if (!conn->buffered_packets.empty()) {

schedule_send_buffered_data(this, conn);

break;

}

listener->turn_read(conn->client_id, true);

upstream->update_flow_control(conn->server_id, listener->flow_control_info(conn->client_id));

break;

}

case CONNS_REJECTED: {

log_conn(this, conn, dbg, "Connection was rejected while waiting for upstream response, state={}, event={}",

magic_enum::enum_name(conn->state), magic_enum::enum_name(what));

upstream->close_connection(id, false, true);

break;

}

default:

log_conn(this, conn, err, "Connection has invalid state: {} (event={})", magic_enum::enum_name(conn->state),

magic_enum::enum_name(what));

assert(0);

upstream->close_connection(id, false, true);

break;

}

break;

}

case SERVER_EVENT_CONNECTION_CLOSED: {

uint64_t id = *(uint64_t *) data;

VpnConnection *conn = vpn_connection_get_by_id(this->connections.by_server_id, id);

if (conn == nullptr) {

log_tun(this, dbg, "Got close event for nonexistent or already closed connection: R:{}", id);

destroy_connection(this, NON_ID, id);

break;

}

log_conn(this, conn, dbg, "Connection closed");

if (std::shared_ptr<ClientListener> listener;

nullptr != vpn_connection_get_by_id(this->connections.by_client_id, conn->client_id)

&& nullptr != (listener = conn->listener.lock())) {

vpn_connection_remove(this->connections.by_server_id, conn->server_id);

listener->turn_read(conn->client_id, false);

close_client_side_connection(this, conn, 0, false);

} else {

destroy_connection(this, conn->client_id, conn->server_id);

}

break;

}

case SERVER_EVENT_READ: {

auto *event = (ServerReadEvent *) data;

VpnConnection *conn = vpn_connection_get_by_id(this->connections.by_server_id, event->id);

if (conn == nullptr) {

log_tun(this, dbg, "Got data from server for inexistent or already closed connection: {}", event->id);

event->result = -1;

assert(0);

break;

}

if (upstream.get() != conn->upstream.lock().get()) {

log_conn(this, conn, warn, "Upstream mismatch");

event->result = -1;

assert(0);

break;

}

std::shared_ptr<ClientListener> listener = conn->listener.lock();

if (listener == nullptr) {

log_conn(this, conn, dbg, "Listener was deleted");

event->result = -1;

assert(0);

break;

}

log_conn(this, conn, trace, "Received {} bytes from server", event->length);

if (conn->flags.test(CONNF_LOOKINGUP_DOMAIN)) {

TunnelDomainLookupAction action =

pass_through_domain_lookup(this, conn, DEPD_INCOMING, event->data, event->length);

log_conn(this, conn, dbg, "tdla={}", magic_enum::enum_name(action));

switch (action) {

case TDLA_DONE:

conn->flags.reset(CONNF_LOOKINGUP_DOMAIN);

break;

case TDLA_PASS:

case TDLA_WANT_MORE:

case TDLA_DROP:

break;

case TDLA_SHUTDOWN:

log_conn(this, conn, dbg, "Server: Connection had been routed {}, closing it to re-route",

(upstream.get() == this->vpn->endpoint_upstream.get()) ? "through VPN endpoint"

: (upstream.get() == this->vpn->bypass_upstream.get()) ? "directly to target host"

: (upstream.get() == this->fake_upstream.get()) ? "through fake endpoint"

: "through unknown endpoint");

close_client_side_connection(this, conn, 0, false);

return;

case TDLA_BLOCK:

log_conn(this, conn, dbg, "Dropped QUIC connection");

upstream->update_flow_control(event->id, {});

close_client_side_connection(this, conn, -1, /* async */ false);

return;

}

}

event->result = (int) listener->send(conn->client_id, event->data, event->length);

if (event->result == 0) {

upstream->update_flow_control(conn->server_id, {});

} else if (event->result > 0) {

conn->incoming_bytes += event->result;

if (conn->flags.test(CONNF_MONITOR_STATS)) {

this->statistics_monitor->update_download(conn->client_id, event->result);

}

TcpFlowCtrlInfo info = listener->flow_control_info(conn->client_id);

log_conn(this, conn, trace, "Can send to client side: {} bytes", info.send_buffer_size);

upstream->update_flow_control(conn->server_id, info);

} else {

log_conn(this, conn, dbg, "Failed to send data from server");

// connection will be closed inside upstream

}

break;

}

case SERVER_EVENT_DATA_SENT: {

const ServerDataSentEvent *event = (ServerDataSentEvent *) data;

VpnConnection *conn = vpn_connection_get_by_id(this->connections.by_server_id, event->id);

if (conn == nullptr) {

break;

}

if (upstream.get() != conn->upstream.lock().get()) {

log_conn(this, conn, warn, "Upstream mismatch");

upstream->close_connection(id, false, true);

assert(0);

break;

}

std::shared_ptr<ClientListener> listener = conn->listener.lock();

if (listener == nullptr) {

log_conn(this, conn, dbg, "Listener was deleted");

upstream->close_connection(id, false, true);

assert(0);

break;

}

switch (conn->state) {

case CONNS_CONNECTED: {

listener->consume(conn->client_id, event->length);

size_t server_can_send = upstream->available_to_send(conn->server_id);

listener->turn_read(conn->client_id, server_can_send > 0);

if (event->length > 0) {

log_conn(this, conn, trace, "{} bytes sent to server (can send to server side: {} bytes)",

event->length, server_can_send);

}

if (!conn->buffered_packets.empty()) {

schedule_send_buffered_data(this, conn);

break;

}

break;

}

default:

// do nothing

break;

}

break;

}

case SERVER_EVENT_GET_AVAILABLE_TO_SEND: {

auto *event = (ServerAvailableToSendEvent *) data;

VpnConnection *conn = vpn_connection_get_by_id(this->connections.by_server_id, event->id);

if (conn == nullptr) {

break;

}

if (std::shared_ptr<ClientListener> listener = conn->listener.lock(); listener != nullptr) {

event->length = listener->flow_control_info(conn->client_id).send_buffer_size;

}

break;

}

case SERVER_EVENT_ERROR: {

const ServerError *event = (ServerError *) data;

if (event->id == NON_ID) {

close_client_side(this, upstream.get());

break;

}

VpnConnection *conn = vpn_connection_get_by_id(this->connections.by_server_id, event->id);

if (conn == nullptr) {

destroy_connection(this, NON_ID, event->id);

break;

}

bool need_close_client_side = true;

switch (conn->state) {

case CONNS_WAITING_RESPONSE: {

log_conn(this, conn, dbg, "Failed to make tunnel: {} ({})", safe_to_string_view(event->error.text),

event->error.code);

break;

}

case CONNS_WAITING_RESPONSE_MIGRATING: {

log_conn(this, conn, dbg, "Failed to switch upstream: {} ({})", safe_to_string_view(event->error.text),

event->error.code);

VpnConnection *src_conn =

vpn_connection_get_by_id(this->connections.by_client_id, conn->migrating_client_id);

if (src_conn == nullptr) {

// do nothing

} else {

upstream->close_connection(src_conn->server_id, false, true);

}

if (std::shared_ptr<ServerUpstream> conn_upstream = conn->upstream.lock(); conn_upstream != nullptr) {

conn_upstream->close_connection(conn->server_id, false, true);

}

need_close_client_side = false;

break;

}

default:

log_conn(this, conn, dbg, "Server closed connection in state: {}, with error: {} ({})",

magic_enum::enum_name(conn->state), safe_to_string_view(event->error.text), event->error.code);

break;

}

if (need_close_client_side) {

if (nullptr != vpn_connection_get_by_id(this->connections.by_client_id, conn->client_id)) {

vpn_connection_remove(this->connections.by_server_id, conn->server_id);

close_client_side_connection(this, conn, event->error.code, false);

} else {

destroy_connection(this, conn->client_id, conn->server_id);

}

}

break;

}

case SERVER_EVENT_ECHO_REPLY: {

auto &reply = *(IcmpEchoReply *) data;

switch (this->icmp_manager.register_reply(reply)) {

case IM_MSGS_PASS:

this->vpn->client_listener->process_icmp_reply(reply);

break;

case IM_MSGS_DROP:

break;

}

break;

}

}

}

constexpr VpnConnectAction vpn_mode_to_action(VpnMode mode) {

VpnConnectAction action = VPN_CA_DEFAULT;

switch (mode) {

case VPN_MODE_GENERAL:

action = VPN_CA_FORCE_REDIRECT;

break;

case VPN_MODE_SELECTIVE:

action = VPN_CA_FORCE_BYPASS;

break;

}

return action;

}

constexpr VpnConnectAction invert_action(VpnConnectAction action) {

switch (action) {

case VPN_CA_DEFAULT:

case VPN_CA_REJECT:

break;

case VPN_CA_FORCE_BYPASS:

action = VPN_CA_FORCE_REDIRECT;

break;

case VPN_CA_FORCE_REDIRECT:

action = VPN_CA_FORCE_BYPASS;

break;

}

return action;

}

static std::shared_ptr<ServerUpstream> select_upstream(

const Tunnel *self, VpnConnectAction action, VpnConnection *conn) {

switch (action) {

case VPN_CA_DEFAULT:

if (conn != nullptr) {

if (conn->listener.lock().get() == self->dns_handler.get()) {

return select_upstream(self, VPN_CA_FORCE_REDIRECT, nullptr);

}

if (conn->flags.test(CONNF_PLAIN_DNS_CONNECTION)) {

return self->dns_handler;

}

}

if (const SocketAddress *dst; // NOLINT(cppcoreguidelines-init-variables)

conn != nullptr && conn->flags.test(CONNF_LOOKINGUP_DOMAIN) && conn->flags.test(CONNF_SUSPECT_EXCLUSION)

&& nullptr != (dst = std::get_if<SocketAddress>(&conn->addr.dst))

&& is_domain_scannable_port(dst->port())) {

return self->fake_upstream;

}

return select_upstream(self, vpn_mode_to_action(self->vpn->domain_filter.get_mode()), nullptr);

case VPN_CA_FORCE_BYPASS:

return self->vpn->bypass_upstream;

break;

case VPN_CA_FORCE_REDIRECT:

return self->vpn->endpoint_upstream;

break;

case VPN_CA_REJECT:

assert(0);

break;

}

return nullptr;

}

/**

* @return The result code which should be set to the event

*/

[[nodiscard]] static ssize_t initiate_connection_migration(

Tunnel *self, VpnConnection *conn, std::shared_ptr<ServerUpstream> upstream, U8View packet) {

if (upstream == nullptr) {

log_conn(self, conn, dbg, "Can't start migration due to upstream isn't selected");

return -1;

}

log_conn(self, conn, dbg, "Migrating to {} upstream",

(upstream.get() == self->vpn->endpoint_upstream.get()) ? "VPN endpoint"

: (upstream.get() == self->vpn->bypass_upstream.get()) ? "direct"

: (upstream.get() == self->fake_upstream.get()) ? "fake"

: "unknown");

uint64_t server_id = upstream->open_connection(&conn->addr, conn->proto, {});

if (server_id == NON_ID) {

log_conn(self, conn, dbg, "Couldn't open connection on new upstream");

return -1;

}

size_t processed = 0;

VpnConnection *sw_conn = VpnConnection::make(NON_ID, conn->addr, conn->proto);

sw_conn->server_id = server_id;

sw_conn->listener = conn->listener;

sw_conn->upstream = upstream;

sw_conn->state = CONNS_WAITING_RESPONSE_MIGRATING;

sw_conn->flags.set(CONNF_LOOKINGUP_DOMAIN, conn->flags.test(CONNF_LOOKINGUP_DOMAIN));

sw_conn->migrating_client_id = conn->client_id;

sw_conn->app_name = conn->app_name;

sw_conn->domain_extractor_result = conn->domain_extractor_result;

add_connection(self, sw_conn);

if (conn->proto == IPPROTO_UDP) {

// do not turn off reads on migrating UDP connections,

// because otherwise the unread packets might be dropped

conn->buffered_packets.emplace_back(packet.begin(), packet.end());

processed = packet.size();

} else {

sw_conn->buffered_packets = std::move(conn->buffered_packets);

if (std::shared_ptr<ClientListener> listener = conn->listener.lock(); listener != nullptr) {

listener->turn_read(conn->client_id, false);

} else {

log_conn(self, conn, dbg, "Listener was deleted");

return -1;

}

}

conn->state = CONNS_CONNECTED_MIGRATING;

if (std::shared_ptr<ServerUpstream> conn_upstream = conn->upstream.lock(); conn_upstream != nullptr) {

conn_upstream->update_flow_control(conn->server_id, {});

}

self->do_health_check(upstream);

log_conn(self, sw_conn, trace, "Connecting...");

return static_cast<ssize_t>(processed);

}