1196 lines
69 KiB
C
1196 lines
69 KiB
C
/*
|
||
* node_conn_direct.c — handle-based ETCP connection layer
|
||
*
|
||
* Тонкая прослойка над ETCP: управляет прямыми подключениями по node_id.
|
||
* Несколько handle'ов могут разделять одно ETCP_CONN.
|
||
* Conn закрывается при закрытии последнего handle (с протоколом CLOSE/KEEP_ALIVE).
|
||
*/
|
||
|
||
#include "node_conn_direct.h"
|
||
#include "etcp_api.h"
|
||
#include "etcp.h"
|
||
#include "etcp_connections.h"
|
||
#include "call_ring.h"
|
||
#include "secure_channel.h"
|
||
#include "utun_instance.h"
|
||
#include "topo_node.h"
|
||
#include "topo_node_sqlite.h"
|
||
#include "sock_match.h"
|
||
#include "nat_detection.h"
|
||
#include "topo_group.h"
|
||
#include "ntp_time.h"
|
||
#include "config_parser.h"
|
||
#include "../local_transport/local_transport.h"
|
||
#include "../lib/debug_config.h"
|
||
#include "../lib/mem.h"
|
||
#include "../lib/u_async.h"
|
||
#include "../lib/ll_queue.h"
|
||
#include "../lib/socket_compat.h"
|
||
#include "../lib/platform_compat.h"
|
||
#include <string.h>
|
||
#include <limits.h>
|
||
|
||
#define DEBUG_CATEGORY_NCD DEBUG_CATEGORY_ETCP
|
||
|
||
/* ═══════════ локальные константы ═══════════ */
|
||
|
||
#define NCD_FIN_WAIT_TIMEOUT_TB 50000 /* 5 сек в 0.1ms */
|
||
|
||
/* ═══════════ внутренние структуры ═══════════ */
|
||
|
||
struct ncd_entry {
|
||
uint64_t node_id;
|
||
struct ETCP_CONN* conn;
|
||
struct UASYNC* ua;
|
||
struct UTUN_INSTANCE* inst;
|
||
unsigned active_calls; /* удержание entry внутри операций и callback-рассылок */
|
||
uint64_t event_generation;
|
||
uint8_t closed;
|
||
uint8_t node_ref;
|
||
uint64_t node_timestamp;
|
||
int handle_count;
|
||
|
||
struct NODE_CONN_DIRECT* handles; /* связный список всех handle'ов */
|
||
|
||
void* connect_timer; /* однократный таймер первого подъёма */
|
||
void* fin_wait_timer; /* таймер ожидания ответа на CLOSE */
|
||
void* lan_timer; /* очистка неподтверждённых LAN-проб */
|
||
uint8_t up : 1; /* текущий статус: 1=есть живой линк */
|
||
uint8_t timed_out : 1; /* connect_timer уже сработал */
|
||
|
||
struct ncd_entry* next; /* цепочка статического реестра */
|
||
};
|
||
|
||
struct NODE_CONN_DIRECT {
|
||
struct ncd_entry* entry;
|
||
ncd_callback cb;
|
||
void* cb_arg;
|
||
struct NODE_CONN_DIRECT* next;
|
||
uint8_t closed;
|
||
void* deliver_up_token; /* handle uasync_call_soon отложенной доставки UP (для отмены при close) */
|
||
};
|
||
|
||
/* ─── forward declarations ─── */
|
||
static void ncd_event_dispatch(struct ncd_entry* entry, enum ncd_event event);
|
||
static void ncd_connect_timeout_cb(void* arg);
|
||
static void ncd_deliver_up_cb(void* arg);
|
||
static void ncd_conn_event(struct ETCP_CONN* conn, int event, void* arg);
|
||
static void ncd_finish(struct ncd_entry* entry, int close_conn);
|
||
static void ncd_collect(struct ncd_entry* entry);
|
||
static void ncd_cancel_deliver_up(struct NODE_CONN_DIRECT* h, struct UASYNC* ua);
|
||
static struct TOPO_NODE* ncd_copy_node(struct UTUN_INSTANCE* inst, struct TOPO_NODE* source);
|
||
|
||
/* ═══════════ реестр ═══════════ */
|
||
/*
|
||
* Реестр ncd_entry по node_id — собственный связный список (не inst->connections).
|
||
* ncd_entry хранит состояние, которого нет в ETCP_CONN: handles, таймеры, fin_wait.
|
||
* Нужен чтобы при повторном open не создавать дублирующий conn для того же node_id.
|
||
*/
|
||
|
||
static struct ncd_entry* ncd_registry_find(struct UTUN_INSTANCE* inst, uint64_t node_id) {
|
||
struct ncd_entry* e = (struct ncd_entry*)inst->ncd_registry;
|
||
while (e) { if (e->node_id == node_id && !e->closed && !e->conn->close_requested && e->conn->state != 2) return e; e = e->next; }
|
||
return NULL;
|
||
}
|
||
|
||
// Проверка владения по объекту conn исключает смешение старого и нового транспорта одного пира.
|
||
int node_conn_direct_handle_count(const struct ETCP_CONN* conn) {
|
||
if (!conn || !conn->instance) return 0;
|
||
for (const struct ncd_entry* entry = conn->instance->ncd_registry; entry; entry = entry->next)
|
||
if (entry->conn == conn && !entry->closed) return entry->handle_count;
|
||
return 0;
|
||
}
|
||
|
||
static int ncd_registry_add(struct UTUN_INSTANCE* inst, struct ncd_entry* entry) {
|
||
entry->inst = inst;
|
||
if (etcp_conn_ref_take(entry->conn) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] register closing conn"); return -1; }
|
||
int events = ETCP_CBK_EVENT_INIT | ETCP_CBK_EVENT_UP | ETCP_CBK_EVENT_DOWN | ETCP_CBK_EVENT_DELETE;
|
||
etcp_conn_add_cbk(entry->conn, ncd_conn_event, entry, events);
|
||
struct etcp_cbk_entry* subscription = entry->conn->cbks;
|
||
if (!subscription || subscription->fn != ncd_conn_event || subscription->arg != entry || subscription->event_mask != events) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot subscribe to transport node=%016llx", (unsigned long long)entry->node_id);
|
||
etcp_conn_ref_free(entry->conn); return -1;
|
||
}
|
||
entry->next = (struct ncd_entry*)inst->ncd_registry; inst->ncd_registry = entry;
|
||
return 0;
|
||
}
|
||
static void ncd_registry_remove(struct UTUN_INSTANCE* inst, struct ncd_entry* entry) {
|
||
struct ncd_entry** pp = (struct ncd_entry**)&inst->ncd_registry;
|
||
while (*pp) { if (*pp == entry) { *pp = entry->next; return; } pp = &(*pp)->next; }
|
||
}
|
||
|
||
/* ═══════════ поиск узла ═══════════ */
|
||
/*
|
||
* Загружает информацию об узле (адреса, pubkey) для создания линков.
|
||
* Сначала ищет в памяти (node_registry — туда попадают узлы из BGP/topo),
|
||
* если нет в памяти — подгружает из SQLite и помещает в реестр.
|
||
* Возвращает владеющую ссылку (ref++), вызывающий обязан сделать topo_node_registry_unref.
|
||
*/
|
||
|
||
static struct TOPO_NODE* ncd_lookup_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
|
||
if (inst->topo_groups && inst->topo_groups->node_registry) {
|
||
struct ll_entry* e = queue_find_data_by_index(inst->topo_groups->node_registry, (const uint8_t*)&node_id);
|
||
if (e) {
|
||
struct TOPO_NODE* node;
|
||
memcpy(&node, e->data + 8, sizeof(node));
|
||
if (node) { topo_node_registry_ref(inst->topo_groups, node_id); return node; }
|
||
}
|
||
}
|
||
if (inst->topo_sqlite_db && inst->topo_groups) {
|
||
struct local_route route;
|
||
int lan = local_transport_route(inst->local_transport, node_id, 0, &route);
|
||
struct TOPO_NODE* ni = lan ? topo_node_sqlite_snapshot_load(inst->topo_sqlite_db, inst->topo_groups, node_id) :
|
||
topo_node_sqlite_node_load(inst->topo_sqlite_db, inst->topo_groups, node_id);
|
||
if (lan && !ni) ni = topo_node_sqlite_identity_load(inst->topo_sqlite_db, node_id);
|
||
if (ni) {
|
||
struct TOPO_NODE* stored = topo_node_registry_store(inst->topo_groups, ni);
|
||
if (!stored) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot register loaded node=%016llx", (unsigned long long)node_id);
|
||
topo_node_destroy(inst->topo_groups, ni);
|
||
}
|
||
return stored;
|
||
}
|
||
}
|
||
return NULL;
|
||
}
|
||
|
||
/* Все разрешённые пары адрес/сокет: UDP, TCP, IPv4/IPv6 и REALITY.
|
||
* Явный specific_sock в open обходит локальную политику, но не проверку семейства/протокола/цели. */
|
||
|
||
/* Создаёт один UDP-линк на use_sock → sa, с вытеснением stale-линка и дедупом. */
|
||
#define NCD_MAX_SOCKET_VIEWS 256 /* sock_id уникален среди живых сокетов и имеет тип uint8_t. */
|
||
static int ncd_add_udp_link(struct ETCP_CONN* conn, struct ETCP_SOCKET* use_sock, struct sockaddr_storage* sa) {
|
||
struct ETCP_LINK* stale = etcp_link_find_by_addr(use_sock, sa, 0);
|
||
if (stale && stale->etcp == conn) {
|
||
if (!stale->is_server && !stale->link_status && !stale->init_timer) {
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] resume idle UDP probe node=%016llx link=%u",
|
||
(unsigned long long)conn->peer_node_id, stale->local_link_id);
|
||
etcp_link_enter_init(stale);
|
||
}
|
||
return conn->close_requested ? -1 : 0;
|
||
}
|
||
if (stale && stale->etcp != conn
|
||
&& memcmp(conn->crypto_ctx.peer_public_key, stale->etcp->crypto_ctx.peer_public_key, SC_PUBKEY_SIZE))
|
||
{
|
||
if (!stale->etcp->links_up || !stale->etcp->initialized) {
|
||
struct ETCP_CONN* old = stale->etcp;
|
||
if (etcp_conn_ref_take(old) < 0) return -1;
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] evict stale endpoint=%s old_node=%016llx new_node=%016llx",
|
||
sockaddr_storage_to_str(sa).str, (unsigned long long)old->peer_node_id,
|
||
(unsigned long long)conn->peer_node_id);
|
||
etcp_link_close(stale);
|
||
if (!old->close_requested) etcp_cbk_fire(old, ETCP_CBK_EVENT_NODE_CHANGED);
|
||
etcp_conn_ref_free(old);
|
||
if (conn->close_requested) return -1;
|
||
stale = NULL;
|
||
} else {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "[ncd] addr conflict: port already used by live conn 0x%016llx for node 0x%016llx",
|
||
(unsigned long long)stale->etcp->peer_node_id, (unsigned long long)conn->peer_node_id);
|
||
return -1;
|
||
}
|
||
}
|
||
if (stale && stale->etcp != conn
|
||
&& !memcmp(conn->crypto_ctx.peer_public_key, stale->etcp->crypto_ctx.peer_public_key, SC_PUBKEY_SIZE))
|
||
{
|
||
DEBUG_ERROR(DEBUG_CATEGORY_ETCP,
|
||
"[ncd] add_udp_link: SAME-PUBKEY DUPLICATE conn at %s — new=%p [%s] state=%d / old=%p [%s] state=%d init=%d up=%d",
|
||
sockaddr_storage_to_str(sa).str,
|
||
(void*)conn, conn->log_name, conn->state,
|
||
(void*)stale->etcp, stale->etcp->log_name, stale->etcp->state,
|
||
stale->etcp->initialized, stale->etcp->links_up);
|
||
return -1;
|
||
}
|
||
if (etcp_link_new(conn, use_sock, sa, 0)) return 1;
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] UDP link creation failed node=%016llx address=%s",
|
||
(unsigned long long)conn->peer_node_id, sockaddr_storage_to_str(sa).str);
|
||
return -1;
|
||
}
|
||
|
||
static int ncd_same_address(const struct sockaddr_storage* a, const struct sockaddr_storage* b) {
|
||
if (a->ss_family != b->ss_family) return 0;
|
||
if (a->ss_family == AF_INET) {
|
||
const struct sockaddr_in* x = (const struct sockaddr_in*)a;
|
||
const struct sockaddr_in* y = (const struct sockaddr_in*)b;
|
||
return x->sin_port == y->sin_port && x->sin_addr.s_addr == y->sin_addr.s_addr;
|
||
}
|
||
if (a->ss_family == AF_INET6) {
|
||
const struct sockaddr_in6* x = (const struct sockaddr_in6*)a;
|
||
const struct sockaddr_in6* y = (const struct sockaddr_in6*)b;
|
||
return x->sin6_port == y->sin6_port && x->sin6_scope_id == y->sin6_scope_id &&
|
||
!memcmp(&x->sin6_addr, &y->sin6_addr, sizeof(x->sin6_addr));
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
static int ncd_add_tcp_link(struct ETCP_CONN* conn, struct ETCP_SOCKET* sock,
|
||
struct sockaddr_storage* addr, const struct TOPO_REALITY_SOCK* reality) {
|
||
for (struct ETCP_LINK* link = conn->links; link; link = link->next) {
|
||
if (!link->is_tcp || link->is_server || link->conn != sock || !ncd_same_address(&link->remote_addr, addr)) continue;
|
||
link->active_users++;
|
||
struct reality_client_config before; memcpy(&before, &link->reality, sizeof(before));
|
||
int had_reality = link->reality_set;
|
||
if (reality) topo_node_apply_reality(link, reality);
|
||
else { memset(&link->reality, 0, sizeof(link->reality)); link->reality_set = 0; }
|
||
if (had_reality != link->reality_set || memcmp(&before, &link->reality, sizeof(before))) {
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] TCP parameters updated node=%016llx link=%d up=%d",
|
||
(unsigned long long)conn->peer_node_id, link->local_link_id, link->link_status);
|
||
if (!link->link_status) etcp_tcp_link_start_reconnect(link);
|
||
}
|
||
if (!link->closing && !conn->close_requested && !link->link_status && !link->tcp_link && !link->tcp_reconnect_timer) {
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] resume idle TCP probe node=%016llx link=%u",
|
||
(unsigned long long)conn->peer_node_id, link->local_link_id);
|
||
etcp_tcp_link_start_reconnect(link);
|
||
}
|
||
int cancelled = link->closing || conn->close_requested;
|
||
etcp_link_release(link);
|
||
return cancelled ? -1 : 0;
|
||
}
|
||
struct ETCP_LINK* link = etcp_link_new(conn, sock, addr, 0);
|
||
if (!link) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] TCP link allocation failed"); return -1; }
|
||
link->active_users++;
|
||
topo_node_apply_reality(link, reality);
|
||
etcp_tcp_link_start_connect(link, addr, 0);
|
||
int cancelled = link->closing || conn->close_requested;
|
||
etcp_link_release(link);
|
||
return cancelled ? -1 : 1;
|
||
}
|
||
|
||
/* Проверка совместимости и создание используют один обход пар, включая TCP fallback и IPv6. */
|
||
/* Сверить неподтверждённые линки с живым discovery-кешем; закрытие может вызвать callbacks NCD. */
|
||
static int ncd_prune_lan(struct ncd_entry* entry, int force) {
|
||
struct ETCP_CONN* conn = entry->conn;
|
||
for (;;) {
|
||
struct ETCP_LINK* stale = NULL;
|
||
int pending = 0;
|
||
for (struct ETCP_LINK* link = conn->links; link; link = link->next) {
|
||
if (!link->lan_generation || link->closing) continue;
|
||
if (link->link_status) { link->lan_generation = 0; continue; }
|
||
int found = 0;
|
||
struct local_route route;
|
||
for (size_t i = 0; !force && local_transport_route(entry->inst->local_transport, entry->node_id, i, &route); i++) {
|
||
if (route.socket.generation == link->lan_generation && route.interface_index == link->lan_interface &&
|
||
!!(route.socket.protocol & LOCAL_ROUTE_TCP) == !!link->is_tcp &&
|
||
ncd_same_address(&route.address, &link->remote_addr) && sock_matches_lan(link->conn, &route.local_address,
|
||
route.interface_index)) { found = 1; break; }
|
||
}
|
||
if (!found) { stale = link; break; }
|
||
pending++;
|
||
}
|
||
if (!stale) return pending;
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] retire LAN probe node=%016llx link=%u generation=%016llx interface=%u",
|
||
(unsigned long long)entry->node_id, stale->local_link_id,
|
||
(unsigned long long)stale->lan_generation, stale->lan_interface);
|
||
etcp_link_close(stale);
|
||
if (entry->closed || conn->close_requested) return 0;
|
||
}
|
||
}
|
||
|
||
/* TTL относится к пробе: после успешного ETCP-handshake линк живёт по обычным правилам транспорта. */
|
||
static void ncd_lan_timer(void* arg) {
|
||
struct ncd_entry* entry = arg;
|
||
entry->lan_timer = NULL;
|
||
struct ETCP_CONN* conn = entry->conn;
|
||
if (etcp_conn_ref_take(conn) < 0) return;
|
||
entry->active_calls++;
|
||
if (ncd_prune_lan(entry, 0) && !entry->closed && !conn->close_requested) {
|
||
entry->lan_timer = uasync_set_timeout(entry->ua, 10000, entry, ncd_lan_timer, "ncd_lan_probe");
|
||
if (!entry->lan_timer) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] LAN cleanup timer failed node=%016llx", (unsigned long long)entry->node_id);
|
||
ncd_prune_lan(entry, 1);
|
||
}
|
||
}
|
||
entry->active_calls--; ncd_collect(entry); etcp_conn_ref_free(conn);
|
||
}
|
||
|
||
/* LAN-кандидаты имеют свою подпись/TTL и не подменяют содержимое или timestamp NODEINFO. */
|
||
static int ncd_create_lan_links(struct ncd_entry* entry, const struct TOPO_NODE* ni,
|
||
struct ETCP_SOCKET* specific_sock, struct UTUN_INSTANCE* check_inst) {
|
||
struct UTUN_INSTANCE* inst = check_inst ? check_inst : entry->inst;
|
||
if (!inst->local_transport) return 0;
|
||
struct ETCP_CONN* conn = check_inst ? NULL : entry->conn;
|
||
uint64_t id = check_inst ? ni->node_id : entry->node_id;
|
||
if (conn && (conn->fin_wait || conn->close_requested)) return 0;
|
||
if (conn) ncd_prune_lan(entry, 0);
|
||
if (conn && (entry->closed || conn->close_requested)) return -1;
|
||
int count = 0;
|
||
struct local_route route;
|
||
for (size_t i = 0; local_transport_route(inst->local_transport, id, i, &route); i++) {
|
||
if ((route.socket.protocol & LOCAL_ROUTE_TCP) && inst->config && inst->config->global.socks_enabled) {
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] LAN TCP skipped node=%016llx socket=%u reason=global SOCKS policy",
|
||
(unsigned long long)id, route.socket.socket_id); continue;
|
||
}
|
||
const struct TOPO_REALITY_SOCK* reality = NULL;
|
||
if (route.socket.protocol & LOCAL_ROUTE_REALITY) {
|
||
reality = topo_node_find_reality_sock(ni, route.socket.socket_id);
|
||
if (!reality) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] LAN REALITY metadata unavailable node=%016llx socket=%u",
|
||
(unsigned long long)id, route.socket.socket_id); continue;
|
||
}
|
||
}
|
||
for (struct ETCP_SOCKET* sock = inst->etcp_sockets; sock; sock = sock->next) {
|
||
if ((specific_sock && sock != specific_sock) || !!sock->is_tcp != !!(route.socket.protocol & LOCAL_ROUTE_TCP) ||
|
||
!sock_matches_lan(sock, &route.local_address, route.interface_index)) continue;
|
||
if (check_inst) return 1;
|
||
if (!entry->lan_timer) {
|
||
entry->lan_timer = uasync_set_timeout(entry->ua, 10000, entry, ncd_lan_timer, "ncd_lan_probe");
|
||
if (!entry->lan_timer) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] LAN cleanup timer allocation failed"); return -1; }
|
||
}
|
||
int added = sock->is_tcp ? ncd_add_tcp_link(conn, sock, &route.address, reality) :
|
||
ncd_add_udp_link(conn, sock, &route.address);
|
||
if (added < 0 || entry->closed || conn->close_requested) return -1;
|
||
for (struct ETCP_LINK* link = conn->links; link; link = link->next) {
|
||
if (link->conn != sock || !!link->is_tcp != !!sock->is_tcp || !ncd_same_address(&link->remote_addr, &route.address)) continue;
|
||
if ((added || link->lan_generation) && !link->link_status) {
|
||
link->lan_generation = route.socket.generation; link->lan_interface = route.interface_index;
|
||
}
|
||
break;
|
||
}
|
||
if (added) DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] LAN probe node=%016llx socket=%u generation=%016llx interface=%u address=%s",
|
||
(unsigned long long)id, route.socket.socket_id, (unsigned long long)route.socket.generation,
|
||
route.interface_index, sockaddr_storage_to_str(&route.address).str);
|
||
count += added;
|
||
}
|
||
}
|
||
return count;
|
||
}
|
||
|
||
static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni,
|
||
struct ETCP_SOCKET* specific_sock, int force_socket, struct UTUN_INSTANCE* check_inst) {
|
||
if (!ni) { DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] link matching without node snapshot"); return 0; }
|
||
if (!check_inst && ni->timestamp < entry->node_timestamp) {
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] ignore stale addresses node=%016llx current=%llu incoming=%llu",
|
||
(unsigned long long)entry->node_id, (unsigned long long)entry->node_timestamp,
|
||
(unsigned long long)(ni ? ni->timestamp : 0)); return 0;
|
||
}
|
||
struct ETCP_CONN* conn = check_inst ? NULL : entry->conn;
|
||
struct UTUN_INSTANCE* inst = check_inst ? check_inst : conn->instance;
|
||
if (!check_inst && memcmp(ni->public_key, conn->crypto_ctx.peer_public_key, SC_PUBKEY_SIZE)) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] address update key mismatch node=%016llx", (unsigned long long)entry->node_id);
|
||
return -1;
|
||
}
|
||
if (!check_inst && !entry->node_ref && topo_node_registry_find(inst->topo_groups, entry->node_id)) {
|
||
topo_node_registry_ref(inst->topo_groups, entry->node_id); entry->node_ref = 1;
|
||
}
|
||
if (!check_inst) entry->node_timestamp = ni->timestamp;
|
||
int link_count = 0, v4_skip = 0, v6_skip = 0, v4_addrs = 0, v6_addrs = 0;
|
||
uint64_t nid = check_inst ? ni->node_id : entry->node_id;
|
||
|
||
/* ---- IPv4 ---- */
|
||
struct sock_view views[NCD_MAX_SOCKET_VIEWS];
|
||
int nviews = sock_collect_views(inst, views, NCD_MAX_SOCKET_VIEWS);
|
||
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) {
|
||
v4_addrs++;
|
||
if (a->port == 0 || !(a->protocol & (TOPO_PROTO_UDP | TOPO_PROTO_TCP))) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[ncd] v4 skip proto=%02x port=%d node=0x%016llx", a->protocol, (int)a->port, (unsigned long long)nid); v4_skip++; continue; }
|
||
int zero = 1; for (int j = 0; j < 4; j++) if (a->addr[j] != 0) { zero = 0; break; }
|
||
if (zero) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[ncd] v4 skip zero-addr node=0x%016llx", (unsigned long long)nid); continue; }
|
||
struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET;
|
||
memcpy(&sin.sin_addr.s_addr, a->addr, 4); sin.sin_port = htons(a->port);
|
||
struct sockaddr_storage sa = {0}; memcpy(&sa, &sin, sizeof(sin));
|
||
uint32_t tip; memcpy(&tip, a->addr, 4);
|
||
const struct TOPO_SOCKMETA4* m = NULL;
|
||
for (const struct TOPO_SOCKMETA4* mm = ni->v4_sock_meta; mm; mm = mm->next)
|
||
if (mm->id == a->socket_id) { m = mm; break; }
|
||
uint8_t tcfg = m ? m->config_type : CFG_SERVER_TYPE_UNKNOWN;
|
||
const struct address_profile *profile = m ? &m->address_profile : NULL;
|
||
uint8_t target_class = sock_target_class(a->type, tcfg, profile, ip_is_public(tip));
|
||
if (target_class == SOCK_TC_UNREACH) { v4_skip++; continue; }
|
||
|
||
if (a->protocol & TOPO_PROTO_TCP) {
|
||
if ((a->protocol & TOPO_PROTO_REALITY) && !topo_node_find_reality_sock(ni, a->socket_id)) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] missing REALITY metadata node=%016llx socket=%u",
|
||
(unsigned long long)nid, a->socket_id); continue;
|
||
}
|
||
int tcp_sockets = 0;
|
||
for (int vi = 0; vi < nviews; vi++) {
|
||
const struct sock_view* v = &views[vi];
|
||
if (!v->is_tcp) continue;
|
||
tcp_sockets++;
|
||
if ((specific_sock && v->sock != specific_sock) ||
|
||
(!force_socket && !sock_match(v, a->type, tcfg, profile, tip))) continue;
|
||
if (check_inst) return 1;
|
||
int added = ncd_add_tcp_link(conn, v->sock, &sa, topo_node_find_reality_sock(ni, a->socket_id));
|
||
if (added < 0 || entry->closed || conn->close_requested) return -1;
|
||
link_count += added;
|
||
}
|
||
if (!specific_sock && !tcp_sockets) {
|
||
if (check_inst) return 1;
|
||
int added = ncd_add_tcp_link(conn, NULL, &sa, topo_node_find_reality_sock(ni, a->socket_id));
|
||
if (added < 0 || entry->closed || conn->close_requested) return -1;
|
||
link_count += added;
|
||
}
|
||
}
|
||
if (a->protocol & TOPO_PROTO_UDP) {
|
||
for (int vi = 0; vi < nviews; vi++) {
|
||
const struct sock_view* v = &views[vi];
|
||
if (v->is_tcp || (specific_sock && v->sock != specific_sock) ||
|
||
(!force_socket && !sock_match(v, a->type, tcfg, profile, tip))) continue;
|
||
if (check_inst) return 1;
|
||
int added = ncd_add_udp_link(conn, v->sock, &sa);
|
||
if (added < 0 || entry->closed || conn->close_requested) return -1;
|
||
link_count += added;
|
||
}
|
||
}
|
||
}
|
||
|
||
/* ---- IPv6 ---- */
|
||
for (const struct TOPO_ADDR6* a = ni->v6_addrs; a; a = a->next) {
|
||
v6_addrs++;
|
||
if (a->port == 0 || !(a->protocol & (TOPO_PROTO_UDP | TOPO_PROTO_TCP))) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[ncd] v6 skip proto=%02x port=%d node=0x%016llx", a->protocol, (int)a->port, (unsigned long long)nid); v6_skip++; continue; }
|
||
int zero = 1; for (int j = 0; j < 16; j++) if (a->addr[j] != 0) { zero = 0; break; }
|
||
if (zero) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[ncd] v6 skip zero-addr node=0x%016llx", (unsigned long long)nid); continue; }
|
||
int is_ll = (a->addr[0] == 0xfe && (a->addr[1] & 0xc0) == 0x80);
|
||
if (a->addr[0] == 0xff) { v6_skip++; continue; }
|
||
const struct TOPO_SOCKMETA6 *metadata = ni->v6_sock_meta;
|
||
while (metadata && metadata->id != a->socket_id) metadata = metadata->next;
|
||
int global = (a->addr[0] & 0xe0) == 0x20;
|
||
uint8_t target_class = sock_target_class(a->type, metadata ? metadata->config_type : CFG_SERVER_TYPE_UNKNOWN,
|
||
metadata ? &metadata->address_profile : NULL, global);
|
||
if (target_class == SOCK_TC_UNREACH) { v6_skip++; continue; }
|
||
struct sockaddr_in6 sin6; memset(&sin6, 0, sizeof(sin6)); sin6.sin6_family = AF_INET6;
|
||
memcpy(&sin6.sin6_addr, a->addr, 16); sin6.sin6_port = htons(a->port);
|
||
|
||
if (a->protocol & TOPO_PROTO_TCP) {
|
||
if ((a->protocol & TOPO_PROTO_REALITY) && !topo_node_find_reality_sock(ni, a->socket_id)) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] missing REALITY metadata node=%016llx socket=%u",
|
||
(unsigned long long)nid, a->socket_id); continue;
|
||
}
|
||
int tcp_sockets = 0;
|
||
struct sockaddr_storage sa = {0}; memcpy(&sa, &sin6, sizeof(sin6));
|
||
for (struct ETCP_SOCKET* s = inst->etcp_sockets; s; s = s->next) {
|
||
if (!s->is_tcp || s->probe_only || s->local_addr.ss_family != AF_INET6) continue;
|
||
tcp_sockets++;
|
||
if ((specific_sock && s != specific_sock) || (is_ll && !s->netif_index) ||
|
||
(!force_socket && target_class != SOCK_TC_PRIV && s->type == CFG_SERVER_TYPE_LOCAL)) continue;
|
||
if (check_inst) return 1;
|
||
if (is_ll) { sin6.sin6_scope_id = s->netif_index; memcpy(&sa, &sin6, sizeof(sin6)); }
|
||
int added = ncd_add_tcp_link(conn, s, &sa, topo_node_find_reality_sock(ni, a->socket_id));
|
||
if (added < 0 || entry->closed || conn->close_requested) return -1;
|
||
link_count += added;
|
||
}
|
||
if (!specific_sock && !tcp_sockets && !is_ll) {
|
||
if (check_inst) return 1;
|
||
int added = ncd_add_tcp_link(conn, NULL, &sa, topo_node_find_reality_sock(ni, a->socket_id));
|
||
if (added < 0 || entry->closed || conn->close_requested) return -1;
|
||
link_count += added;
|
||
}
|
||
}
|
||
if (a->protocol & TOPO_PROTO_UDP) {
|
||
for (struct ETCP_SOCKET* s = inst->etcp_sockets; s; s = s->next) {
|
||
if (s->is_tcp || s->probe_only || s->local_addr.ss_family != AF_INET6 ||
|
||
(specific_sock && s != specific_sock) || (is_ll && !s->netif_index) ||
|
||
(!force_socket && target_class != SOCK_TC_PRIV && s->type == CFG_SERVER_TYPE_LOCAL)) continue;
|
||
if (check_inst) return 1;
|
||
if (is_ll) sin6.sin6_scope_id = s->netif_index;
|
||
struct sockaddr_storage sa = {0}; memcpy(&sa, &sin6, sizeof(sin6));
|
||
int added = ncd_add_udp_link(conn, s, &sa);
|
||
if (added < 0 || entry->closed || conn->close_requested) return -1;
|
||
link_count += added;
|
||
}
|
||
}
|
||
}
|
||
|
||
int lan = ncd_create_lan_links(entry, ni, specific_sock, check_inst);
|
||
if (lan < 0) return -1;
|
||
if (check_inst && lan) return 1;
|
||
link_count += lan;
|
||
if (link_count > 0) {
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] create_links: node=0x%016llx v4=%d/skp=%d v6=%d/skp=%d → %d new links (UDP+TCP)",
|
||
(unsigned long long)nid, v4_addrs, v4_skip, v6_addrs, v6_skip, link_count);
|
||
}
|
||
return link_count;
|
||
}
|
||
|
||
int node_conn_direct_compatible(struct UTUN_INSTANCE* inst, uint64_t node_id) {
|
||
if (!inst || !node_id) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] compatibility invalid arguments"); return 0; }
|
||
struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
|
||
if (conn && conn->state != 2 && !conn->close_requested && conn->links_up) return 1;
|
||
struct TOPO_NODE* node = topo_node_registry_find(inst->topo_groups, node_id);
|
||
if (!node) {
|
||
struct local_route route;
|
||
if (inst->topo_sqlite_db && local_transport_route(inst->local_transport, node_id, 0, &route)) {
|
||
struct TOPO_NODE* identity = topo_node_sqlite_identity_load(inst->topo_sqlite_db, node_id);
|
||
int possible = identity ? ncd_create_lan_links(NULL, identity, NULL, inst) : 0;
|
||
topo_node_destroy(inst->topo_groups, identity);
|
||
return possible > 0;
|
||
}
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] compatibility unknown snapshot node=%016llx", (unsigned long long)node_id); return 0;
|
||
}
|
||
if (ncd_create_links(NULL, node, NULL, 0, inst) > 0) return 1;
|
||
/* Входящее подключение тоже допустимо: REVERSE_REQ просит пира открыть наш опубликованный адрес.
|
||
* TCP имеет собственный исходящий socket fallback; для UDP нужен известный UDP-сокет пира.
|
||
* Это возможность попытки, не подтверждение маршрута/работоспособности скрытых сокетов пира. */
|
||
const struct TOPO_NODE* self = topo_node_registry_find(inst->topo_groups, inst->node_id);
|
||
if (!self) return 0;
|
||
for (const struct TOPO_ADDR4* a = self->v4_addrs; a; a = a->next) {
|
||
uint32_t ip; memcpy(&ip, a->addr, 4);
|
||
if (!ip || !a->port) continue;
|
||
const struct TOPO_SOCKMETA4* meta = self->v4_sock_meta;
|
||
while (meta && meta->id != a->socket_id) meta = meta->next;
|
||
uint8_t cfg = meta ? meta->config_type : CFG_SERVER_TYPE_UNKNOWN;
|
||
const struct address_profile *profile = meta ? &meta->address_profile : NULL;
|
||
uint8_t target_class = sock_target_class(a->type, cfg, profile, ip_is_public(ip));
|
||
if (target_class == SOCK_TC_PRIV || target_class == SOCK_TC_UNREACH) continue;
|
||
if ((a->protocol & TOPO_PROTO_REALITY) && !topo_node_find_reality_sock(self, a->socket_id)) continue;
|
||
if (a->protocol & TOPO_PROTO_TCP) return 1;
|
||
if (!(a->protocol & TOPO_PROTO_UDP)) continue;
|
||
for (const struct TOPO_ADDR4* remote = node->v4_addrs; remote; remote = remote->next) {
|
||
if (!(remote->protocol & TOPO_PROTO_UDP)) continue;
|
||
const struct TOPO_SOCKMETA4* rm = node->v4_sock_meta;
|
||
while (rm && rm->id != remote->socket_id) rm = rm->next;
|
||
struct sock_view view = {.has_inet = 1};
|
||
if (rm) view.address_profile = rm->address_profile;
|
||
for (const struct TOPO_ADDR4* ra = node->v4_addrs; ra; ra = ra->next)
|
||
if (ra->socket_id == remote->socket_id && ra->type == TOPO_ADDR_NAT) memcpy(&view.public_ip4, ra->addr, 4);
|
||
if (sock_match(&view, a->type, cfg, profile, ip)) return 1;
|
||
}
|
||
}
|
||
for (const struct TOPO_ADDR6* a = self->v6_addrs; a; a = a->next) {
|
||
int nonzero = 0; for (unsigned i = 0; i < 16; i++) nonzero |= a->addr[i];
|
||
if (!nonzero || !a->port || (a->addr[0] == 0xfe && (a->addr[1] & 0xc0) == 0x80)) continue;
|
||
const struct TOPO_SOCKMETA6* meta = self->v6_sock_meta;
|
||
while (meta && meta->id != a->socket_id) meta = meta->next;
|
||
if (meta && (meta->config_type == CFG_SERVER_TYPE_PRIVATE || meta->config_type == CFG_SERVER_TYPE_LOCAL)) continue;
|
||
if ((a->protocol & TOPO_PROTO_REALITY) && !topo_node_find_reality_sock(self, a->socket_id)) continue;
|
||
if (a->protocol & TOPO_PROTO_TCP) return 1;
|
||
if (!(a->protocol & TOPO_PROTO_UDP)) continue;
|
||
for (const struct TOPO_ADDR6* remote = node->v6_addrs; remote; remote = remote->next)
|
||
if (remote->protocol & TOPO_PROTO_UDP) return 1;
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
int node_conn_direct_update_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
|
||
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] update without instance"); return -1; }
|
||
struct ncd_entry* entry = ncd_registry_find(inst, node_id);
|
||
struct TOPO_NODE* ni = topo_node_registry_find(inst->topo_groups, node_id);
|
||
if (!entry || !ni) return 0;
|
||
struct TOPO_NODE* copy = ncd_copy_node(inst, ni);
|
||
if (!copy) return -1;
|
||
struct ETCP_CONN* conn = entry->conn;
|
||
if (etcp_conn_ref_take(conn) < 0) { topo_node_destroy(inst->topo_groups, copy); return -1; }
|
||
entry->active_calls++;
|
||
int added = ncd_create_links(entry, copy, NULL, 0, NULL);
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] apply node=%016llx timestamp=%llu up=%d added=%d",
|
||
(unsigned long long)node_id, (unsigned long long)copy->timestamp, entry->up, added);
|
||
entry->active_calls--; ncd_collect(entry);
|
||
etcp_conn_ref_free(conn);
|
||
topo_node_destroy(inst->topo_groups, copy);
|
||
return added;
|
||
}
|
||
|
||
/* Адрес получен отдельно от BGP; нет транспорта — сохраняем только discovery-кеш. */
|
||
int node_conn_direct_update_local(struct UTUN_INSTANCE* inst, uint64_t node_id) {
|
||
if (!inst || !node_id) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] invalid LAN update"); return -1; }
|
||
struct ncd_entry* entry = ncd_registry_find(inst, node_id);
|
||
if (!entry || entry->conn->fin_wait) return 0;
|
||
struct ETCP_CONN* conn = entry->conn;
|
||
if (etcp_conn_ref_take(conn) < 0) return -1;
|
||
entry->active_calls++;
|
||
struct local_route route;
|
||
int has_routes = local_transport_route(inst->local_transport, node_id, 0, &route);
|
||
struct TOPO_NODE* known = has_routes ? ncd_lookup_node(inst, node_id) : NULL;
|
||
struct TOPO_NODE* copy = known ? ncd_copy_node(inst, known) : NULL;
|
||
if (known) topo_node_registry_unref(inst->topo_groups, node_id);
|
||
int result = copy ? ncd_create_lan_links(entry, copy, NULL, NULL) : has_routes ? -1 : 0;
|
||
if (!copy) ncd_prune_lan(entry, 0);
|
||
topo_node_destroy(inst->topo_groups, copy);
|
||
entry->active_calls--; ncd_collect(entry); etcp_conn_ref_free(conn);
|
||
return result;
|
||
}
|
||
|
||
/* ═══════════ единая диспетчеризация событий ═══════════ */
|
||
/*
|
||
* Центральный диспетчер: принимает событие (UP/DOWN/TIMEOUT), проверяет что оно
|
||
* действительно меняет состояние (повторный UP когда уже up — игнорируется),
|
||
* обновляет entry->up/timed_out и рассылает callback всем handle'ам.
|
||
*
|
||
* Именно через эту функцию все потребители узнают об изменении состояния соединения.
|
||
* Один вызов — один проход по всем handle'ам.
|
||
*/
|
||
|
||
static void ncd_event_dispatch(struct ncd_entry* entry, enum ncd_event event) {
|
||
if (!entry || (entry->closed && event != NCD_EVENT_CLOSED)) return;
|
||
switch (event) {
|
||
case NCD_EVENT_UP:
|
||
if (entry->up) return;
|
||
if (entry->timed_out) DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] connected after timeout node=%016llx",
|
||
(unsigned long long)entry->node_id);
|
||
entry->timed_out = 0;
|
||
if (entry->connect_timer) { uasync_cancel_timeout(entry->ua, entry->connect_timer); entry->connect_timer = NULL; }
|
||
entry->up = 1; break;
|
||
case NCD_EVENT_DOWN:
|
||
if (!entry->up) return;
|
||
entry->up = 0; break;
|
||
case NCD_EVENT_TIMEOUT:
|
||
if (entry->timed_out) return;
|
||
entry->timed_out = 1; entry->connect_timer = NULL; break;
|
||
case NCD_EVENT_CLOSED: entry->up = 0; break;
|
||
}
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] event=%d node=%016llx handles=%d depth=%u", event,
|
||
(unsigned long long)entry->node_id, entry->handle_count, entry->active_calls);
|
||
entry->active_calls++;
|
||
uint64_t generation = ++entry->event_generation;
|
||
// Закрытые элементы не освобождаются до выхода из внешней рассылки. Новые добавляются в голову.
|
||
for (struct NODE_CONN_DIRECT* h = entry->handles; h; h = h->next) {
|
||
if (entry->event_generation != generation) {
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] stop superseded event=%d node=%016llx", event, (unsigned long long)entry->node_id);
|
||
break;
|
||
}
|
||
ncd_cancel_deliver_up(h, entry->ua);
|
||
if (!h->closed && h->cb && (!entry->closed || event == NCD_EVENT_CLOSED)) h->cb(h, event, h->cb_arg);
|
||
}
|
||
entry->active_calls--;
|
||
ncd_collect(entry);
|
||
}
|
||
|
||
/* ═══════════ ETCP коллбэки (прокидывают в ncd_event_dispatch) ═══════════ */
|
||
/*
|
||
* ETCP сообщил что соединение работает: INIT — handshake завершён,
|
||
* UP — линки восстановились после DOWN. Транслируем в NCD_EVENT_UP.
|
||
* TIMEOUT сообщает об истечении срока первой попытки, но удерживаемый handle
|
||
* продолжает следить за транспортом (в том числе при REVERSE). FIN_WAIT подавляет UP.
|
||
* Если INIT пришёл а линков нет — UP не доставляем: ETCP_CBK_EVENT_UP
|
||
* доставит когда линки поднимутся. Так гарантируем что пользователь
|
||
* получает UP только когда conn реально готов отправлять данные.
|
||
*/
|
||
|
||
static void ncd_conn_event(struct ETCP_CONN* conn, int event, void* arg) {
|
||
struct ncd_entry* entry = arg;
|
||
if (event == ETCP_CBK_EVENT_DELETE) { ncd_finish(entry, 0); return; }
|
||
if (event == ETCP_CBK_EVENT_DOWN) {
|
||
if (conn->fin_wait && !entry->handle_count) {
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] DOWN confirms FIN_WAIT node=%016llx", (unsigned long long)entry->node_id);
|
||
ncd_finish(entry, 1); return;
|
||
}
|
||
ncd_event_dispatch(entry, NCD_EVENT_DOWN); return;
|
||
}
|
||
if (!conn->fin_wait && !conn->close_requested && conn->state == 1 && conn->links_up)
|
||
ncd_event_dispatch(entry, NCD_EVENT_UP);
|
||
}
|
||
|
||
static void ncd_collect(struct ncd_entry* entry) {
|
||
if (entry->active_calls) return;
|
||
struct NODE_CONN_DIRECT** pp = &entry->handles;
|
||
while (*pp) {
|
||
struct NODE_CONN_DIRECT* h = *pp;
|
||
if (h->closed) { *pp = h->next; u_free(h); } else pp = &h->next;
|
||
}
|
||
if (entry->closed && !entry->handle_count) {
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] free node=%016llx", (unsigned long long)entry->node_id);
|
||
u_free(entry);
|
||
}
|
||
}
|
||
|
||
// Сначала исключение из реестра и отсоединение транспорта, затем уведомление владельцев.
|
||
static void ncd_finish(struct ncd_entry* entry, int close_conn) {
|
||
if (entry->closed) return;
|
||
entry->active_calls++;
|
||
entry->closed = 1;
|
||
ncd_registry_remove(entry->inst, entry);
|
||
if (entry->node_ref) {
|
||
topo_node_registry_unref(entry->inst->topo_groups, entry->node_id); entry->node_ref = 0;
|
||
}
|
||
if (entry->connect_timer) { uasync_cancel_timeout(entry->ua, entry->connect_timer); entry->connect_timer = NULL; }
|
||
if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
|
||
if (entry->lan_timer) { uasync_cancel_timeout(entry->ua, entry->lan_timer); entry->lan_timer = NULL; }
|
||
struct ETCP_CONN* conn = entry->conn; entry->conn = NULL;
|
||
if (conn) {
|
||
if (conn->fin_wait_clear_arg == entry) {
|
||
conn->fin_wait = 0; conn->fin_wait_clear_cb = NULL; conn->fin_wait_clear_arg = NULL;
|
||
}
|
||
etcp_conn_remove_cbk(conn, ncd_conn_event, entry);
|
||
if (close_conn) etcp_connection_close(conn); /* ETCP сам откладывает close из своих callbacks. */
|
||
etcp_conn_ref_free(conn);
|
||
}
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] detached node=%016llx handles=%d", (unsigned long long)entry->node_id, entry->handle_count);
|
||
ncd_event_dispatch(entry, NCD_EVENT_CLOSED);
|
||
entry->active_calls--; ncd_collect(entry);
|
||
}
|
||
|
||
/* ═══════════ таймер подключения ═══════════ */
|
||
/*
|
||
* Таймер первого подключения истёк — соединение не установилось за etcp_connect_timeout_tb.
|
||
* Транслирует NCD_EVENT_TIMEOUT один раз. Владелец может закрыть handle или
|
||
* оставить его для позднего UP; окончательное завершение обозначается CLOSED.
|
||
*/
|
||
|
||
static void ncd_connect_timeout_cb(void* arg) {
|
||
struct ncd_entry* entry = (struct ncd_entry*)arg;
|
||
if (!entry || entry->timed_out) return;
|
||
entry->connect_timer = NULL;
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] connect TIMEOUT: node=0x%016llx now_tb=%llu handles=%d",
|
||
(unsigned long long)entry->node_id, (unsigned long long)get_time_tb(), entry->handle_count);
|
||
nat_detection_connection_failed(entry->inst->nat_det, entry->conn, 0);
|
||
ncd_event_dispatch(entry, NCD_EVENT_TIMEOUT);
|
||
}
|
||
|
||
/* ═══════════ асинхронная доставка UP ═══════════ */
|
||
/*
|
||
* Доставляет NCD_EVENT_UP одному конкретному handle'у (не всем).
|
||
* Используется когда handle добавляется к уже работающему conn: доставка отложенная
|
||
* (uasync_call_soon), чтобы вызывающий успел сохранить указатель handle
|
||
* до вызова callback'а. Синхронная доставка привела бы к гонке.
|
||
*/
|
||
|
||
static void ncd_deliver_up_cb(void* arg) {
|
||
struct NODE_CONN_DIRECT* h = (struct NODE_CONN_DIRECT*)arg;
|
||
h->deliver_up_token = NULL; /* токен consumed — close больше не отменит освобождённый узел */
|
||
struct ETCP_CONN* conn = h->entry->conn;
|
||
if (!h->closed && !h->entry->closed && h->entry->up && !h->entry->timed_out && h->cb &&
|
||
conn && conn->state == 1 && conn->links_up && !conn->close_requested && !conn->fin_wait)
|
||
h->cb(h, NCD_EVENT_UP, h->cb_arg);
|
||
}
|
||
|
||
/* Планирует отложенную доставку UP и сохраняет токен для возможной отмены при close. */
|
||
static int ncd_schedule_deliver_up(struct UASYNC* ua, struct NODE_CONN_DIRECT* h) {
|
||
ncd_cancel_deliver_up(h, ua);
|
||
h->deliver_up_token = uasync_call_soon(ua, h, ncd_deliver_up_cb);
|
||
if (!h->deliver_up_token) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot schedule initial UP node=%016llx", (unsigned long long)h->entry->node_id);
|
||
return -1;
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
/* Отменяет pending-доставку UP (если ещё не сработала). Без этого h освобождается,
|
||
* а отложенный ncd_deliver_up_cb читает freed-память — UAF. */
|
||
static void ncd_cancel_deliver_up(struct NODE_CONN_DIRECT* h, struct UASYNC* ua) {
|
||
if (h->deliver_up_token) { uasync_call_soon_cancel(ua, h->deliver_up_token); h->deliver_up_token = NULL; }
|
||
}
|
||
|
||
/* ═══════════ FIN_WAIT ═══════════ */
|
||
/*
|
||
* fin_wait — состояние ожидания подтверждения закрытия от пира.
|
||
* Вызывающая сторона: мы отправили CLOSE и ждём ответа (KEEP_ALIVE или DOWN).
|
||
* Обычное ожидание — NCD_FIN_WAIT_TIMEOUT_TB (5 сек); объявленный сон продлевает
|
||
* его до RX deadline живого линка. По истечении закрываем транспорт.
|
||
*/
|
||
|
||
/*
|
||
* ETCP отменяет наш fin_wait — peer переподключился и прислал новый INIT
|
||
* (или другая причина отмены на уровне ETCP). Отменяем локальный таймер,
|
||
* соединение продолжает работать.
|
||
*/
|
||
static void ncd_fin_wait_cancelled(struct ETCP_CONN* conn, void* arg) {
|
||
struct ncd_entry* entry = (struct ncd_entry*)arg;
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] fin_wait cancelled node=0x%016llx", (unsigned long long)entry->node_id);
|
||
if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
|
||
}
|
||
|
||
/*
|
||
* Таймаут fin_wait: peer не ответил на CLOSE в отведённый срок с учётом сна.
|
||
* Форсированно закрываем conn и освобождаем entry — соединение разорвано
|
||
* без подтверждения от пира.
|
||
*/
|
||
static void ncd_fin_wait_timeout_cb(void* arg) {
|
||
struct ncd_entry* entry = (struct ncd_entry*)arg;
|
||
if (!entry || !entry->conn) return;
|
||
entry->fin_wait_timer = NULL;
|
||
if (!entry->conn->fin_wait) return;
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] fin_wait timeout, force close node=0x%016llx", (unsigned long long)entry->node_id);
|
||
ncd_finish(entry, 1);
|
||
}
|
||
|
||
/* ═══════════ приём CLOSE / KEEP_ALIVE ═══════════ */
|
||
/*
|
||
* Обрабатывает управляющие сообщения от пира по протоколу graceful shutdown.
|
||
*
|
||
* CLOSE (пир хочет закрыть conn — у него закончились handle'ы):
|
||
* - Если у нас ещё есть handle'ы — отказываем: шлём KEEP_ALIVE, conn живёт.
|
||
* - Если handle'ов нет — соглашаемся, закрываем conn.
|
||
*
|
||
* KEEP_ALIVE (пир просит не закрывать conn — у него ещё есть handle'ы):
|
||
* - Если мы в fin_wait — отменяем его, conn продолжает работать.
|
||
* - Отсутствие наших handle'ов не отменяет потребность пира: FIN_WAIT снимается полностью.
|
||
*/
|
||
|
||
static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e) {
|
||
if (!conn || !e || !e->dgram || e->len != NCD_CONTROL_MSG_SIZE || e->dgram[0] != ETCP_ID_NCD_CONTROL) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] malformed control conn=%p size=%zu", (void*)conn, e ? (size_t)e->len : 0);
|
||
if (e) { queue_dgram_free(e); queue_entry_free(e); }
|
||
return;
|
||
}
|
||
struct ncd_control_msg message;
|
||
memcpy(&message, e->dgram, sizeof(message));
|
||
const struct ncd_control_msg* msg = &message;
|
||
uint64_t sender_id = 0;
|
||
for (unsigned i = 0; i < 8; i++) sender_id |= (uint64_t)e->dgram[2 + i] << (8 * i);
|
||
struct ncd_entry* entry = ncd_registry_find(conn->instance, sender_id);
|
||
if (!sender_id || sender_id != conn->peer_node_id || (entry && entry->conn != conn)) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] control from wrong transport node=%016llx", (unsigned long long)sender_id);
|
||
queue_dgram_free(e); queue_entry_free(e); return;
|
||
}
|
||
|
||
switch (msg->subcmd) {
|
||
case NCD_SUBCMD_CLOSE:
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] recv CLOSE from 0x%016llx", (unsigned long long)sender_id);
|
||
if (!entry) {
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] no entry for CLOSE sender 0x%016llx, deferred close conn", (unsigned long long)sender_id);
|
||
etcp_connection_close(conn);
|
||
break;
|
||
}
|
||
if (entry->handle_count > 0) {
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] refusing CLOSE, still have %d handles for 0x%016llx",
|
||
entry->handle_count, (unsigned long long)sender_id);
|
||
{ struct ncd_control_msg resp = { ETCP_ID_NCD_CONTROL, NCD_SUBCMD_KEEP_ALIVE, conn->instance->node_id };
|
||
struct ll_entry* qe = ll_alloc_lldgram(sizeof(resp));
|
||
if (!qe) DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE allocation failed");
|
||
else {
|
||
memcpy(qe->dgram, &resp, sizeof(resp)); qe->len = sizeof(resp);
|
||
for (unsigned i = 0; i < 8; i++) qe->dgram[2 + i] = (uint8_t)(conn->instance->node_id >> (8 * i));
|
||
if (etcp_send(conn, qe) != 0) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE send failed node=%016llx", (unsigned long long)sender_id);
|
||
queue_dgram_free(qe); queue_entry_free(qe);
|
||
}
|
||
}
|
||
}
|
||
} else {
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] deferred close entry for CLOSE from 0x%016llx", (unsigned long long)sender_id);
|
||
ncd_finish(entry, 1);
|
||
}
|
||
break;
|
||
case NCD_SUBCMD_KEEP_ALIVE:
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] recv KEEP_ALIVE from 0x%016llx", (unsigned long long)sender_id);
|
||
if (entry && entry->conn->fin_wait) {
|
||
entry->conn->fin_wait = 0; entry->conn->fin_wait_clear_cb = NULL; entry->conn->fin_wait_clear_arg = NULL;
|
||
if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE: fin_wait cleared, conn stays alive handles=%d", entry->handle_count);
|
||
}
|
||
break;
|
||
default:
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] unknown control subcmd=%u node=%016llx", msg->subcmd, (unsigned long long)sender_id);
|
||
break;
|
||
}
|
||
queue_dgram_free(e); queue_entry_free(e);
|
||
}
|
||
|
||
/* ═══════════ инициализация глобального обработчика ═══════════ */
|
||
/*
|
||
* Однократная привязка обработчика NCD control-сообщений к ETCP (etcp_bind).
|
||
* Вызывается при запуске сети и node_conn_direct_open, повторные вызовы безвредны
|
||
* (защита через inst->ncd_control_bound).
|
||
*/
|
||
|
||
int node_conn_direct_init(struct UTUN_INSTANCE* inst) {
|
||
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] init without instance"); return -1; }
|
||
if (inst->ncd_control_bound) return 0;
|
||
if (etcp_bind(inst, ETCP_ID_NCD_CONTROL, ncd_recv_control_handler) < 0) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot bind CLOSE/KEEP_ALIVE handler"); return -1;
|
||
}
|
||
inst->ncd_control_bound = 1;
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] bound ETCP_ID_NCD_CONTROL handler");
|
||
return 0;
|
||
}
|
||
|
||
/* Своя копия адресов: callbacks линков могут заменить списки в общем node_registry. */
|
||
static struct TOPO_NODE* ncd_copy_node(struct UTUN_INSTANCE* inst, struct TOPO_NODE* source) {
|
||
uint8_t wire[TOPO_NODE_WIRE_MAX_SIZE];
|
||
struct TOPO_GROUP_NODE node = {0};
|
||
struct TOPO_GROUP group = {.instance = inst};
|
||
struct TOPO_NODE* copy = NULL;
|
||
int bytes = topo_node_serialize(source, &node, 0, 0, wire, sizeof(wire), 0);
|
||
if (bytes < 0 || !inst->topo_groups ||
|
||
topo_node_deserialize(&group, wire, (size_t)bytes, ©, NULL, NULL, NULL, NULL) < 0) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot copy address snapshot node=%016llx", (unsigned long long)source->node_id);
|
||
return NULL;
|
||
}
|
||
return copy;
|
||
}
|
||
|
||
/* Общий запуск: подготовить ресурсы, обновить транспорт и только затем опубликовать handle.
|
||
* entry/conn удерживаются через callbacks; новый владелец не получает событий до возврата open. */
|
||
static int ncd_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id, ncd_callback cb, void* cb_arg,
|
||
struct NODE_CONN_DIRECT** out_handle, struct TOPO_NODE* source,
|
||
struct ETCP_SOCKET* specific_sock) {
|
||
if (out_handle) *out_handle = NULL;
|
||
if (!inst || !inst->ua || !inst->connections || !out_handle || !node_id || node_id == inst->node_id ||
|
||
(specific_sock && (specific_sock->instance != inst || specific_sock->probe_only ||
|
||
(specific_sock->local_addr.ss_family != AF_INET && specific_sock->local_addr.ss_family != AF_INET6)))) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] invalid open node=%016llx socket=%p", (unsigned long long)node_id, (void*)specific_sock);
|
||
return NCD_ERR;
|
||
}
|
||
if (node_conn_direct_init(inst) < 0) return NCD_ERR;
|
||
struct NODE_CONN_DIRECT* handle = u_calloc(1, sizeof(*handle));
|
||
if (!handle) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] handle allocation failed node=%016llx", (unsigned long long)node_id); return NCD_ERR; }
|
||
handle->cb = cb; handle->cb_arg = cb_arg;
|
||
struct ncd_entry* entry = ncd_registry_find(inst, node_id);
|
||
struct ETCP_CONN* conn = entry ? entry->conn : instance_find_conn(inst, node_id);
|
||
if (conn && (conn->close_requested || conn->state == 2)) conn = NULL;
|
||
struct TOPO_NODE* copy = NULL;
|
||
int created_conn = 0, created_entry = 0, held = 0, pinned = 0, result = NCD_ERR;
|
||
if (!source && (!conn || !entry || entry->timed_out || specific_sock)) {
|
||
struct TOPO_NODE* known = ncd_lookup_node(inst, node_id);
|
||
if (known) { copy = ncd_copy_node(inst, known); topo_node_registry_unref(inst->topo_groups, node_id); }
|
||
if (known && !copy) goto done;
|
||
} else if (source) {
|
||
copy = ncd_copy_node(inst, source);
|
||
if (!copy) goto done;
|
||
}
|
||
if (!conn) {
|
||
if (!copy || copy->node_id != node_id || sc_derive_node_id_from_pubkey(copy->public_key) != node_id) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] missing/invalid identity node=%016llx", (unsigned long long)node_id); goto done;
|
||
}
|
||
char name[MAX_CONN_NAME_LEN];
|
||
if (copy->node_name && copy->node_name[0]) snprintf(name, sizeof(name), "%s", copy->node_name);
|
||
else snprintf(name, sizeof(name), "n_%016llx", (unsigned long long)node_id);
|
||
conn = etcp_connection_create(inst, name);
|
||
if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] transport allocation failed node=%016llx", (unsigned long long)node_id); goto done; }
|
||
created_conn = 1; conn->peer_node_id = node_id; etcp_update_log_name(conn);
|
||
if (sc_init_ctx(&conn->crypto_ctx, &inst->my_keys) != SC_OK ||
|
||
sc_set_peer_public_key(&conn->crypto_ctx, copy->public_key, SC_PEER_PUBKEY_BIN) != SC_OK) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] transport key setup failed node=%016llx", (unsigned long long)node_id); goto done;
|
||
}
|
||
if (etcp_conn_index_peer(conn) < 0) goto done;
|
||
}
|
||
if (etcp_conn_ref_take(conn) < 0) goto done;
|
||
held = 1;
|
||
if (!entry) {
|
||
entry = u_calloc(1, sizeof(*entry));
|
||
if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] entry allocation failed node=%016llx", (unsigned long long)node_id); goto done; }
|
||
entry->node_id = node_id; entry->conn = conn; entry->ua = inst->ua;
|
||
entry->up = conn->state == 1 && conn->links_up;
|
||
if (ncd_registry_add(inst, entry) < 0) { u_free(entry); entry = NULL; goto done; }
|
||
created_entry = 1;
|
||
}
|
||
entry->active_calls++; pinned = 1; handle->entry = entry;
|
||
if (entry->up && cb && ncd_schedule_deliver_up(inst->ua, handle) < 0) goto done;
|
||
if (!entry->up && (!entry->connect_timer || entry->timed_out)) {
|
||
int delay = inst->etcp_connect_timeout_tb > INT_MAX ? INT_MAX : (int)inst->etcp_connect_timeout_tb;
|
||
void* timer = uasync_set_timeout(inst->ua, delay, entry, ncd_connect_timeout_cb, "ncd_connect");
|
||
if (!timer) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] connect timer failed node=%016llx", (unsigned long long)node_id); goto done; }
|
||
if (entry->connect_timer) uasync_cancel_timeout(inst->ua, entry->connect_timer);
|
||
entry->connect_timer = timer; entry->timed_out = 0;
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] connect deadline node=%016llx timeout_tb=%d", (unsigned long long)node_id, delay);
|
||
}
|
||
if (copy && ncd_create_links(entry, copy, specific_sock, specific_sock != NULL, NULL) < 0) goto done;
|
||
if (entry->closed || conn->close_requested || conn->state == 2) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] open cancelled by transport callback node=%016llx", (unsigned long long)node_id); goto done;
|
||
}
|
||
if (conn->state == 1 && conn->links_up && !entry->up) {
|
||
ncd_event_dispatch(entry, NCD_EVENT_UP);
|
||
if (entry->closed || conn->close_requested) goto done;
|
||
}
|
||
if (entry->up && cb && !handle->deliver_up_token && ncd_schedule_deliver_up(inst->ua, handle) < 0) goto done;
|
||
if (!entry->up) ncd_cancel_deliver_up(handle, inst->ua);
|
||
if (conn->fin_wait) {
|
||
conn->fin_wait = 0; conn->fin_wait_clear_cb = NULL; conn->fin_wait_clear_arg = NULL;
|
||
if (entry->fin_wait_timer) { uasync_cancel_timeout(inst->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] new owner cancels FIN_WAIT node=%016llx", (unsigned long long)node_id);
|
||
}
|
||
handle->next = entry->handles; entry->handles = handle; entry->handle_count++;
|
||
*out_handle = handle; handle = NULL; result = created_conn ? NCD_NEW : NCD_REUSED;
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open %s node=%016llx ready=%u handles=%d", created_conn ? "NEW" : "REUSED",
|
||
(unsigned long long)node_id, entry->up, entry->handle_count);
|
||
done:
|
||
if (handle) { ncd_cancel_deliver_up(handle, inst->ua); u_free(handle); }
|
||
if (result == NCD_ERR) {
|
||
if (created_entry && !entry->closed && !entry->handle_count) ncd_finish(entry, created_conn);
|
||
else if (created_conn && (!entry || entry->closed || !entry->handle_count)) etcp_connection_close(conn);
|
||
}
|
||
if (pinned) { entry->active_calls--; ncd_collect(entry); }
|
||
if (held) etcp_conn_ref_free(conn);
|
||
if (copy) topo_node_destroy(inst->topo_groups, copy);
|
||
return result;
|
||
}
|
||
|
||
/* Поиск записи выполняется общим запуском только при необходимости создать/обновить линки. */
|
||
int node_conn_direct_open(struct UTUN_INSTANCE* inst, uint64_t node_id, ncd_callback cb, void* cb_arg,
|
||
struct NODE_CONN_DIRECT** out_handle, struct ETCP_SOCKET* specific_sock) {
|
||
return ncd_open_node(inst, node_id, cb, cb_arg, out_handle, NULL, specific_sock);
|
||
}
|
||
|
||
int node_conn_direct_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id,
|
||
ncd_callback cb, void* cb_arg, struct NODE_CONN_DIRECT** out_handle,
|
||
struct TOPO_NODE* ni, struct ETCP_SOCKET* specific_sock) {
|
||
if (out_handle) *out_handle = NULL;
|
||
if (!inst || !out_handle || !ni || node_id != ni->node_id ||
|
||
node_id != sc_derive_node_id_from_pubkey(ni->public_key) || (ni->timestamp && topo_node_verify(ni) < 0)) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] invalid node passed to open"); return NCD_ERR;
|
||
}
|
||
struct TOPO_NODE* known = ncd_lookup_node(inst, node_id);
|
||
if (ni->timestamp && (!known || ni->timestamp > known->timestamp)) {
|
||
struct TOPO_NODE* copy = ncd_copy_node(inst, ni);
|
||
if (!copy) {
|
||
if (known) topo_node_registry_unref(inst->topo_groups, node_id);
|
||
return NCD_ERR;
|
||
}
|
||
struct TOPO_NODE* stored = topo_node_registry_store(inst->topo_groups, copy);
|
||
if (known) topo_node_registry_unref(inst->topo_groups, node_id);
|
||
known = stored;
|
||
if (!stored) { topo_node_destroy(inst->topo_groups, copy); return NCD_ERR; }
|
||
if (inst->topo_sqlite_db && topo_node_sqlite_snapshot_put(inst->topo_sqlite_db, stored) < 0)
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot persist received node snapshot");
|
||
}
|
||
if (known && ni->timestamp && known->timestamp == ni->timestamp &&
|
||
memcmp(known->x25519_self_sig, ni->x25519_self_sig, 64)) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] conflicting node snapshot node=%016llx", (unsigned long long)node_id);
|
||
topo_node_registry_unref(inst->topo_groups, node_id); return NCD_ERR;
|
||
}
|
||
struct TOPO_NODE* selected = known && known->timestamp ? known : ni;
|
||
int result = ncd_open_node(inst, node_id, cb, cb_arg, out_handle, selected, specific_sock);
|
||
if (known) topo_node_registry_unref(inst->topo_groups, node_id);
|
||
return result;
|
||
}
|
||
|
||
/*
|
||
* Закрыть handle (graceful shutdown).
|
||
*
|
||
* Если остались другие handle'ы — просто удаляется из списка, conn продолжает работу.
|
||
*
|
||
* Если это был последний handle (handle_count стал 0):
|
||
* - Запускается протокол graceful shutdown: пиру отправляется CLOSE, conn
|
||
* переводится в fin_wait, ставится таймер на 5 сек либо до пробуждения пира.
|
||
* - Если пир ответит KEEP_ALIVE (у него ещё есть handle'ы) — conn останется жив,
|
||
* просто без наших handle'ов.
|
||
* - Если пир не ответит — conn закрывается форсированно по таймауту.
|
||
*/
|
||
static void ncd_handle_close(struct NODE_CONN_DIRECT* h, int force) {
|
||
if (!h || h->closed) return;
|
||
struct ncd_entry* entry = h->entry;
|
||
entry->active_calls++;
|
||
struct ETCP_CONN* conn = entry->conn;
|
||
int held = conn && etcp_conn_ref_take(conn) == 0;
|
||
ncd_cancel_deliver_up(h, entry->ua);
|
||
h->closed = 1; h->cb = NULL; h->cb_arg = NULL;
|
||
entry->handle_count--;
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] handle close node=%016llx remaining=%d force=%d",
|
||
(unsigned long long)entry->node_id, entry->handle_count, force);
|
||
if (entry->handle_count || entry->closed) goto done;
|
||
if (force || !conn || conn->close_requested || conn->state == 2 || entry->timed_out || (!conn->state && !conn->links_up)) {
|
||
ncd_finish(entry, 1); goto done;
|
||
}
|
||
if (!conn->links_up) {
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] last handle released while DOWN, retaining until ETCP timeout node=%016llx",
|
||
(unsigned long long)entry->node_id);
|
||
ncd_finish(entry, 0);
|
||
if (etcp_conn_schedule_down_close(conn) < 0) etcp_connection_close(conn);
|
||
goto done;
|
||
}
|
||
if (entry->connect_timer) { uasync_cancel_timeout(entry->ua, entry->connect_timer); entry->connect_timer = NULL; }
|
||
conn->fin_wait = 1; conn->fin_wait_clear_cb = ncd_fin_wait_cancelled; conn->fin_wait_clear_arg = entry;
|
||
struct ncd_control_msg msg = { ETCP_ID_NCD_CONTROL, NCD_SUBCMD_CLOSE, entry->inst->node_id };
|
||
struct ll_entry* e = ll_alloc_lldgram(sizeof(msg));
|
||
if (!e) DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] CLOSE allocation failed");
|
||
else {
|
||
memcpy(e->dgram, &msg, sizeof(msg)); e->len = sizeof(msg);
|
||
for (unsigned i = 0; i < 8; i++) e->dgram[2 + i] = (uint8_t)(entry->inst->node_id >> (8 * i));
|
||
if (etcp_send(conn,e) != 0) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] CLOSE send failed"); queue_dgram_free(e); queue_entry_free(e);
|
||
}
|
||
}
|
||
if (entry->closed || conn->close_requested || !conn->fin_wait || entry->handle_count) goto done;
|
||
uint64_t now = get_time_tb(), wait_tb = NCD_FIN_WAIT_TIMEOUT_TB;
|
||
for (struct ETCP_LINK* link = conn->links; link; link = link->next) {
|
||
if (!link->link_status || (!link->ka_my_sleeping && !link->ka_peer_sleeping)) continue;
|
||
if (link->rx_deadline_tb > now && link->rx_deadline_tb - now > wait_tb) wait_tb = link->rx_deadline_tb - now;
|
||
}
|
||
int delay = wait_tb > INT_MAX ? INT_MAX : (int)wait_tb;
|
||
entry->fin_wait_timer = uasync_set_timeout(entry->ua, delay, entry, ncd_fin_wait_timeout_cb, "ncd_fin_wait");
|
||
if (!entry->fin_wait_timer) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot schedule FIN_WAIT node=%016llx", (unsigned long long)entry->node_id);
|
||
ncd_finish(entry, 1); goto done;
|
||
}
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] fin_wait node=%016llx timeout_ms=%d", (unsigned long long)entry->node_id, delay / 10);
|
||
done:
|
||
entry->active_calls--; ncd_collect(entry);
|
||
if (held) etcp_conn_ref_free(conn);
|
||
}
|
||
void node_conn_direct_close(struct NODE_CONN_DIRECT* h) { ncd_handle_close(h, 0); }
|
||
void node_conn_direct_force_close(struct NODE_CONN_DIRECT* h) { ncd_handle_close(h, 1); }
|
||
|
||
/*
|
||
* Сменить или сбросить (cb=NULL) callback на уже открытом handle.
|
||
* Не влияет на refcounting и состояние conn.
|
||
*/
|
||
void node_conn_direct_set_callback(struct NODE_CONN_DIRECT* h, ncd_callback cb, void* cb_arg) {
|
||
if (!h || !h->entry) return;
|
||
h->cb = cb;
|
||
h->cb_arg = cb_arg;
|
||
}
|
||
|
||
void node_conn_direct_transfer(struct NODE_CONN_DIRECT* h, ncd_callback new_cb, void* new_cb_arg) {
|
||
if (!h || !h->entry) return;
|
||
h->cb = new_cb;
|
||
h->cb_arg = new_cb_arg;
|
||
}
|
||
|
||
/*
|
||
* Прямой доступ к ETCP_CONN из handle.
|
||
* Нужен для отправки данных (etcp_send), проверки статуса, и т.д.
|
||
*/
|
||
uint64_t node_conn_direct_node_id(const struct NODE_CONN_DIRECT* h) {
|
||
return h && h->entry ? h->entry->node_id : 0;
|
||
}
|
||
|
||
struct ETCP_CONN* node_conn_direct_get_conn(struct NODE_CONN_DIRECT* h) {
|
||
if (!h || h->closed || !h->entry || h->entry->closed) return NULL;
|
||
return h->entry->conn;
|
||
}
|
||
|
||
/* ═══════════ Управление линками на конкретном сокете (для auto_socket) ═══════════ */
|
||
|
||
/* Снимок entry удерживает список через callbacks, которые могут закрыть соседних владельцев. */
|
||
int ncd_add_socket_links(struct UTUN_INSTANCE* inst, struct ETCP_SOCKET* sock) {
|
||
if (!inst || !sock || sock->instance != inst || sock->probe_only ||
|
||
(sock->local_addr.ss_family != AF_INET && sock->local_addr.ss_family != AF_INET6)) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] invalid added socket=%p", (void*)sock); return -1;
|
||
}
|
||
size_t count = 0;
|
||
for (struct ncd_entry* entry = inst->ncd_registry; entry; entry = entry->next) count++;
|
||
if (!count) return 0;
|
||
struct ncd_entry** entries = u_calloc(count, sizeof(*entries));
|
||
if (!entries) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] added socket snapshot allocation failed"); return -1; }
|
||
size_t held = 0;
|
||
for (struct ncd_entry* entry = inst->ncd_registry; entry; entry = entry->next) {
|
||
entry->active_calls++; entries[held++] = entry;
|
||
}
|
||
int added = 0, failed = 0;
|
||
for (size_t i = 0; i < held; i++) {
|
||
struct ncd_entry* entry = entries[i];
|
||
struct ETCP_CONN* conn = entry->conn;
|
||
if (!entry->closed && !conn->fin_wait && etcp_conn_ref_take(conn) == 0) {
|
||
struct TOPO_NODE* known = ncd_lookup_node(inst, entry->node_id);
|
||
struct TOPO_NODE* copy = known ? ncd_copy_node(inst, known) : NULL;
|
||
if (known) topo_node_registry_unref(inst->topo_groups, entry->node_id);
|
||
if (copy) {
|
||
int result = ncd_create_links(entry, copy, sock, 0, NULL);
|
||
if (result < 0) failed = 1; else added += result;
|
||
topo_node_destroy(inst->topo_groups, copy);
|
||
} else if (known) failed = 1;
|
||
etcp_conn_ref_free(conn);
|
||
}
|
||
entry->active_calls--; ncd_collect(entry);
|
||
}
|
||
u_free(entries);
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] added socket=%s transports=%zu links=%d failed=%d", sock->name, held, added, failed);
|
||
return failed ? -1 : added;
|
||
}
|
||
|
||
int ncd_remove_socket_links(struct UTUN_INSTANCE* inst, struct ETCP_SOCKET* sock) {
|
||
if (!inst || !sock) return 0;
|
||
int removed = 0;
|
||
size_t count = 0;
|
||
for (struct ncd_entry* e = inst->ncd_registry; e; e = e->next) count++;
|
||
if (!count) return 0;
|
||
struct ETCP_CONN** conns = u_calloc(count,sizeof(*conns));
|
||
if (!conns) { DEBUG_ERROR(DEBUG_CATEGORY_NCD,"[ncd] remove_socket_links: snapshot allocation failed"); return -1; }
|
||
size_t held = 0;
|
||
for (struct ncd_entry* e = inst->ncd_registry; e; e = e->next)
|
||
if (etcp_conn_ref_take(e->conn) == 0) conns[held++] = e->conn;
|
||
// DOWN callback может удалить текущий и следующий entry. Снимок удерживает только транспорт.
|
||
for (size_t i = 0; i < held; i++) {
|
||
for (;;) {
|
||
struct ETCP_LINK* link = conns[i]->links;
|
||
while (link && link->conn != sock) link = link->next;
|
||
if (!link) break;
|
||
etcp_link_close(link); removed++;
|
||
}
|
||
etcp_conn_ref_free(conns[i]);
|
||
}
|
||
u_free(conns);
|
||
if (removed > 0) {
|
||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] remove_socket_links socket=%s closed=%d", sock->name, removed);
|
||
}
|
||
return removed;
|
||
}
|