jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix: NCD UAF, conn_mgr invite protocol, node_id derivation, link reuse

Browse Source 
node_conn_direct:
- ncd_init_cb: skip UP when links_up==0, ncd_up_cb delivers when link comes up
- close/force_close: handle_count safety checks, callback clearing
- open/open_node case 2: create links on reused dead conn
- ncd_create_links: skip duplicate link on same conn (stale->etcp==conn)
- set_callback: optional callback change on live handle

conn_mgr:
- cm_ncd_callback: DOWN only delivers event (handle stays alive)
- cm_ncd_callback: TIMEOUT force_close BEFORE cm_deliver_event
- cm_deliver_event: read handle fields to locals before callback (reentrany-safe)
- INVITE_INFO_REQ/RESP: add duplicate cmd byte for direct-send protocol
- cm_handle_invite_info_resp: search invite by group_id in channel conn_mgr

Android:
- node_id computed by sc_derive_node_id_from_pubkey in C, not Kotlin
- utun_bridge_join_channel: recompute node_id, log mismatch
- invite_link_c: compute nodeId from pubkey via sc_derive_node_id_from_pubkey
- NativeLib.deriveNodeId, InviteLink.kt uses C function
This commit is contained in:
evgeny committed 2026-08-03 09:51:06 +03:00
1 parent 4b07d59043
commit 740ed2a93a
10 files changed
+351 -53
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/routing_layer/conn_mgr.h
+4 -4
View File
@@ -55,10 +55,10 @@ struct TOPO_GROUP;
*/
enum conn_mgr_event {
CONN_EVENT_UP = 0, /* обычное соединение установлено */
CONN_EVENT_DOWN = 1, /* соединение потеряно */
CONN_EVENT_TIMEOUT = 2, /* таймаут / ошибка */
CONN_EVENT_JOIN = 3, /* invite успешен (группа создана, соединение готово) */
CONN_EVENT_UP = 0, /* связь есть. handle жив, можно отправлять данные */
CONN_EVENT_DOWN = 1, /* временный обрыв. handle жив, само восстановится. НЕ закрывать */
CONN_EVENT_TIMEOUT = 2, /* подключение не удалось. handle жив, закройте сами */
CONN_EVENT_JOIN = 3, /* invite принят, группа создана. handle жив */
};
/**
src/routing_layer/conn_mgr_core.c
+46 -19
View File
@@ -28,7 +28,7 @@
void conn_mgr_router_recv_handler(struct ETCP_CONN* conn, struct ll_entry* entry);
static void cm_handle_invite_info_req(struct CONN_MGR* mgr, struct ETCP_CONN* conn, const uint8_t* data, size_t len);
static void cm_handle_invite_info_resp(struct CONN_MGR* mgr, const uint8_t* data, size_t len);
static void cm_handle_invite_info_resp(struct CONN_MGR* mgr, struct ETCP_CONN* conn, const uint8_t* data, size_t len);
/* ═══════ утилиты ═══════ */
@@ -54,11 +54,20 @@ static void cm_deliver_up_cb(void* arg) {
}
/* Рассылает событие ВСЕМ handle'ам entry. Одно соединение — много слушателей:
* каждый conn_mgr_open создал свой handle, все получают UP/DOWN/TIMEOUT. */
* каждый conn_mgr_open создал свой handle, все получают UP/DOWN/TIMEOUT.
* Все поля handle'а читаются до вызова коллбэка — коллбэк может закрыть
* handle (conn_mgr_close), после вызова к handle не обращаемся. */
void cm_deliver_event(struct CONN_MGR_ENTRY* entry, enum conn_mgr_event event) {
uint64_t gid = entry->mgr->group->group_id;
struct CONN_MGR_HANDLE* h = entry->handles;
while (h) { struct CONN_MGR_HANDLE* next = h->next; if (h->cb) h->cb(h, h->node_id, gid, event, h->cb_arg); h = next; }
while (h) {
struct CONN_MGR_HANDLE* next = h->next;
uint64_t node_id = h->node_id;
conn_mgr_cb_t cb = h->cb;
void* cb_arg = h->cb_arg;
if (cb) cb(h, node_id, gid, event, cb_arg);
h = next;
}
}
/* ═══════ NAT / адресные утилиты ═══════ */
@@ -184,7 +193,7 @@ void cm_entry_cleanup(struct CONN_MGR_ENTRY* entry) {
} else pp = &(*pp)->next;
}
}
if (entry->ncd_handle) { node_conn_direct_close(entry->ncd_handle); entry->ncd_handle = NULL; }
if (entry->ncd_handle) { node_conn_direct_force_close(entry->ncd_handle); entry->ncd_handle = NULL; }
entry->handles = NULL;
cm_clear_nodeinfo(entry->mgr, entry->node_id);
entry->state = CONN_MGR_STATE_DISCONNECTED; entry->conn_type = CONN_TYPE_NONE;
@@ -228,7 +237,7 @@ void cm_cleanup_db_node(struct CONN_MGR_ENTRY* entry) {
/* ═══════ NCD коллбэки ═══════ */
/* NCD события → conn_mgr для DIRECT/REVERSE фаз:
/* NCD (node conn direct) события → conn_mgr для DIRECT/REVERSE фаз:
* UP — соединение поднялось, помечаем CONNECTED, доставляем UP всем handle'ам.
* TIMEOUT — NCD таймаут истёк, переходим к REVERSE или INDIRECT, либо фейлим db_node.
* DOWN — соединение упало, доставляем DOWN. */
@@ -248,9 +257,10 @@ void cm_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void
case NCD_EVENT_TIMEOUT:
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "link: ETCP connect TIMEOUT to 0x%016llx — no response, trying fallback",
(unsigned long long)entry->node_id);
entry->ncd_handle = NULL;
if (entry->db_loaded) {
entry->main_connect_state = CM_TRY_FAILED;
{ struct NODE_CONN_DIRECT* ncd = entry->ncd_handle; entry->ncd_handle = NULL;
if (ncd) node_conn_direct_force_close(ncd); }
cm_cleanup_db_node(entry); cm_deliver_event(entry, CONN_EVENT_TIMEOUT);
return;
}
@@ -258,17 +268,20 @@ void cm_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void
{ struct TOPO_GROUP* g = entry->mgr->group;
struct TOPO_GROUP_NODE* t = topo_node_find_by_id(g, entry->node_id);
if (entry->local_scan_state != CM_TRY_OK) {
{ struct NODE_CONN_DIRECT* ncd = entry->ncd_handle; entry->ncd_handle = NULL;
if (ncd) node_conn_direct_force_close(ncd); }
if (cm_has_direct_ip(entry->mgr, g->local_node) && !(t && cm_has_direct_ip(entry->mgr, t)))
{ cm_start_phase_reverse(entry); return; }
cm_start_phase_indirect(entry); return;
}
}
if (entry->local_scan_state == CM_TRY_FAILED) cm_deliver_event(entry, CONN_EVENT_TIMEOUT);
{ struct NODE_CONN_DIRECT* ncd = entry->ncd_handle; entry->ncd_handle = NULL;
if (ncd) node_conn_direct_force_close(ncd); }
break;
case NCD_EVENT_DOWN:
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "link: ETCP link DROPPED to 0x%016llx",
(unsigned long long)entry->node_id);
entry->ncd_handle = NULL;
cm_deliver_event(entry, CONN_EVENT_DOWN);
break;
}
@@ -284,12 +297,13 @@ void cm_invite_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_e
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: ETCP handshake OK with 0x%016llx — sending membership check",
(unsigned long long)inv->node_id);
inv->state = CM_INVITE_WAIT_INFO;
{ struct CM_INVITE_REQ req; memset(&req, 0, sizeof(req));
req.cmd = ETCP_RT_ID_CONN_MGR; req.subcmd = CM_SUBCMD_INVITE_INFO_REQ; req.group_id = inv->mgr->group->group_id;
{ uint8_t buf[11]; /* dgram[0]=cmd(dispatch), dgram[1]=cmd(handler skip), dgram[2]=subcmd, dgram[3..10]=group_id */
buf[0] = ETCP_RT_ID_CONN_MGR; buf[1] = ETCP_RT_ID_CONN_MGR; buf[2] = CM_SUBCMD_INVITE_INFO_REQ;
memcpy(buf + 3, &inv->mgr->group->group_id, 8);
struct ll_entry* qe = queue_entry_new(0);
if (!qe) { cm_invite_fail(inv); return; }
qe->dgram = u_malloc(sizeof(req)); if (!qe->dgram) { queue_entry_free(qe); cm_invite_fail(inv); return; }
memcpy(qe->dgram, &req, sizeof(req)); qe->len = sizeof(req);
qe->dgram = u_malloc(sizeof(buf)); if (!qe->dgram) { queue_entry_free(qe); cm_invite_fail(inv); return; }
memcpy(qe->dgram, buf, sizeof(buf)); qe->len = sizeof(buf);
etcp_send(node_conn_direct_get_conn(ncd_h), qe);
}
break;
@@ -711,7 +725,7 @@ void conn_mgr_router_recv_handler(struct ETCP_CONN* conn, struct ll_entry* entry
case CM_SUBCMD_INTERM_SELECTED: if (mgr) cm_handle_interm_selected(mgr, d, len); break;
case CM_SUBCMD_DISCONNECT: if (len >= CM_DISCONNECT_SIZE && mgr) cm_handle_disconnect(mgr, ((struct CM_DISCONNECT*)d)->node_id); break;
case CM_SUBCMD_INVITE_INFO_REQ: if (mgr && len >= sizeof(struct CM_INVITE_REQ)) cm_handle_invite_info_req(mgr, conn, d, len); break;
case CM_SUBCMD_INVITE_INFO_RESP: if (mgr) cm_handle_invite_info_resp(mgr, d, len); break;
case CM_SUBCMD_INVITE_INFO_RESP: if (mgr) cm_handle_invite_info_resp(mgr, conn, d, len); break;
}
}
@@ -730,17 +744,29 @@ void cm_handle_invite_info_req(struct CONN_MGR* mgr, struct ETCP_CONN* conn, con
memcpy(resp->ed25519_pubkey, sg->ed25519_public_key, SC_PUBKEY_SIZE); resp->node_name_len=(uint8_t)nl;
if (nl) memcpy(resp->node_name, nm, nl);
struct ll_entry* qe = queue_entry_new(0);
if (qe) { qe->dgram=(uint8_t*)resp; qe->len=(uint16_t)rs; etcp_send(conn, qe); } else u_free(resp);
if (qe) {
size_t new_len = rs + 1; /* +1 for duplicate cmd: dgram[0]=dispatch, dgram[1]=handler-skip, dgram[2]=subcmd, dgram[3..]=payload */
uint8_t* buf = u_malloc(new_len);
if (buf) { buf[0]=ETCP_RT_ID_CONN_MGR; buf[1]=ETCP_RT_ID_CONN_MGR; buf[2]=CM_SUBCMD_INVITE_INFO_RESP;
memcpy(buf+3, (uint8_t*)resp+2, rs-2); qe->dgram=buf; qe->len=(uint16_t)new_len; etcp_send(conn, qe); u_free(resp); }
else { queue_entry_free(qe); u_free(resp); }
} else u_free(resp);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: membership check RECEIVED from 0x%016llx — responding is_member=yes name=\"%.*s\"",
(unsigned long long)conn->peer_node_id, (int)nl, nm ? nm : "");
}
void cm_handle_invite_info_resp(struct CONN_MGR* mgr, const uint8_t* data, size_t len) {
void cm_handle_invite_info_resp(struct CONN_MGR* mgr, struct ETCP_CONN* conn, const uint8_t* data, size_t len) {
if (len < sizeof(struct CM_INVITE_RESP)) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "invite: INVITE_INFO_RESP too short (%zu)", len); return; }
struct CM_INVITE_RESP* resp = (struct CM_INVITE_RESP*)data;
if (len < sizeof(struct CM_INVITE_RESP) + (size_t)resp->node_name_len) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "invite: INVITE_INFO_RESP truncated"); return; }
struct cm_invite_pending* inv = NULL;
for (struct cm_invite_pending* p = mgr->invite_list; p; p = p->next) if (p->state == CM_INVITE_WAIT_INFO) { inv = p; break; }
{ struct TOPO_GROUP* grp = topo_groups_find(mgr->instance->topo_groups, resp->group_id);
if (grp && grp->conn_mgr) {
for (struct cm_invite_pending* p = grp->conn_mgr->invite_list; p; p = p->next)
if (p->state == CM_INVITE_WAIT_INFO && p->node_id == conn->peer_node_id) { inv = p; break; }
}
}
if (!inv) { for (struct cm_invite_pending* p = mgr->invite_list; p; p = p->next) if (p->state == CM_INVITE_WAIT_INFO) { inv = p; break; } }
if (!inv) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "invite: INVITE_INFO_RESP — no pending invite in WAIT_INFO state"); return; }
if (!resp->is_member) {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "invite: membership DENIED by 0x%016llx — not in channel", (unsigned long long)inv->node_id);
@@ -815,11 +841,12 @@ static void cm_tcp_ready_cb(struct stcp_link* link, void* arg) {
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: TCP handshake OK with 0x%016llx — sending membership check",
(unsigned long long)inv->node_id);
inv->state = CM_INVITE_WAIT_INFO;
{ struct CM_INVITE_REQ req; memset(&req,0,sizeof(req));
req.cmd=ETCP_RT_ID_CONN_MGR; req.subcmd=CM_SUBCMD_INVITE_INFO_REQ; req.group_id=inv->mgr->group->group_id;
{ uint8_t buf[11]; /* dgram[0]=cmd(dispatch), dgram[1]=cmd(handler skip), dgram[2]=subcmd, dgram[3..10]=group_id */
buf[0] = ETCP_RT_ID_CONN_MGR; buf[1] = ETCP_RT_ID_CONN_MGR; buf[2] = CM_SUBCMD_INVITE_INFO_REQ;
memcpy(buf + 3, &inv->mgr->group->group_id, 8);
struct ll_entry* qe2 = queue_entry_new(0); if (!qe2) { cm_invite_fail(inv); return; }
qe2->dgram=u_malloc(sizeof(req)); if (!qe2->dgram) { queue_entry_free(qe2); cm_invite_fail(inv); return; }
memcpy(qe2->dgram,&req,sizeof(req)); qe2->len=sizeof(req); etcp_send(etcp, qe2);
qe2->dgram=u_malloc(sizeof(buf)); if (!qe2->dgram) { queue_entry_free(qe2); cm_invite_fail(inv); return; }
memcpy(qe2->dgram,buf,sizeof(buf)); qe2->len=sizeof(buf); etcp_send(etcp, qe2);
}
}
src/transport_layer/node_conn_direct.c
+220 -2
View File
@@ -66,6 +66,11 @@ static void ncd_deferred_close(void* arg);
static void ncd_deferred_close_conn(void* arg);
/* ═══════════ реестр ═══════════ */
/*
* Реестр 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;
@@ -81,6 +86,12 @@ static void ncd_registry_remove(struct UTUN_INSTANCE* inst, struct ncd_entry* en
}
/* ═══════════ поиск узла ═══════════ */
/*
* Загружает информацию об узле (адреса, 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) {
@@ -99,6 +110,20 @@ static struct TOPO_NODE* ncd_lookup_node(struct UTUN_INSTANCE* inst, uint64_t no
}
/* ═══════════ создание линков (round‑robin, все сокеты кроме PRIVATE) ═══════════ */
/*
* Создаёт по одному ETCP_LINK на каждый адрес пира (IPv4 и IPv6).
*
* Все адреса пира распределяются round-robin по локальным сокетам.
* Если specific_sock задан — использует только его. PRIVATE-сокеты пропускаются.
*
* Особые случаи:
* - Нулевые адреса и нулевые порты пропускаются
* - Не-UDP протоколы пропускаются
* - IPv6 link-local: автоматически выставляется scope_id по netif_index сокета
* - Stale-линки: если на том же addr:port висит старый conn с другим pubkey
* и старый conn ещё не поднялся — stale-линк вытесняется (узел пересоздался
* с новым ключом, старый conn больше не нужен)
*/
static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni,
struct ETCP_SOCKET* specific_sock) {
@@ -130,6 +155,7 @@ static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni,
struct ETCP_SOCKET* use_sock = socks[rr++ % sock_count];
{
struct ETCP_LINK* stale = etcp_link_find_by_addr(use_sock, &sa);
if (stale && stale->etcp == conn) continue;
if (stale && stale->etcp != conn
&& memcmp(conn->crypto_ctx.peer_public_key, stale->etcp->crypto_ctx.peer_public_key, SC_PUBKEY_SIZE))
{
@@ -178,6 +204,7 @@ static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni,
struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6));
{
struct ETCP_LINK* stale = etcp_link_find_by_addr(use_sock, &sa);
if (stale && stale->etcp == conn) continue;
if (stale && stale->etcp != conn
&& memcmp(conn->crypto_ctx.peer_public_key, stale->etcp->crypto_ctx.peer_public_key, SC_PUBKEY_SIZE))
{
@@ -203,6 +230,14 @@ static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni,
}
/* ═══════════ единая диспетчеризация событий ═══════════ */
/*
* Центральный диспетчер: принимает событие (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) return;
@@ -230,17 +265,32 @@ static void ncd_event_dispatch(struct ncd_entry* entry, enum ncd_event event) {
}
/* ═══════════ ETCP коллбэки (прокидывают в ncd_event_dispatch) ═══════════ */
/*
* ETCP сообщил что соединение работает: INIT — handshake завершён,
* UP — линки восстановились после DOWN. Транслируем в NCD_EVENT_UP.
* Не транслируем если уже сработал таймаут подключения (timed_out)
* или conn в состоянии fin_wait (ждём закрытия).
* Если INIT пришёл а линков нет — UP не доставляем: ncd_up_cb сам
* доставит когда линки поднимутся. Так гарантируем что пользователь
* получает UP только когда conn реально готов отправлять данные.
*/
static void ncd_init_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)event;
struct ncd_entry* entry = (struct ncd_entry*)arg;
if (!entry || entry->timed_out || conn->fin_wait) return;
ncd_event_dispatch(entry, NCD_EVENT_UP);
if (conn->links_up > 0) ncd_event_dispatch(entry, NCD_EVENT_UP);
}
static void ncd_up_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)event;
struct ncd_entry* entry = (struct ncd_entry*)arg;
if (!entry || conn->fin_wait) return;
ncd_event_dispatch(entry, NCD_EVENT_UP);
}
/*
* Отложенное (uasync_call_soon) уничтожение ncd_entry вместе с conn.
* Нельзя вызывать напрямую из ETCP-коллбэка: conn может использоваться
* после возврата из коллбэка. Поэтому очистка откладывается на следующий цикл событий.
* Снимает все ETCP-коллбэки, удаляет из реестра, закрывает conn, освобождает память.
*/
static void ncd_deferred_close(void* arg) {
struct ncd_entry* entry = (struct ncd_entry*)arg;
struct ETCP_CONN* conn = entry->conn;
@@ -253,10 +303,20 @@ static void ncd_deferred_close(void* arg) {
}
u_free(entry);
}
/*
* Отложенное закрытие conn без ncd_entry.
* Используется когда пришёл CLOSE от пира, а ncd_entry для этого пира уже нет
* (закрыли раньше — например другой модуль уже удалил все handle'ы).
*/
static void ncd_deferred_close_conn(void* arg) {
struct ETCP_CONN* conn = (struct ETCP_CONN*)arg;
if (conn) etcp_connection_close(conn);
}
/*
* ETCP сообщил что все линки упали.
* Если conn в fin_wait и handle'ов нет — peer подтвердил закрытие,
* делаем немедленную очистку. Иначе транслируем NCD_EVENT_DOWN потребителям.
*/
static void ncd_down_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)event;
struct ncd_entry* entry = (struct ncd_entry*)arg;
if (!entry) return;
@@ -271,6 +331,11 @@ static void ncd_down_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)ev
}
/* ═══════════ таймер подключения ═══════════ */
/*
* Таймер первого подключения истёк — соединение не установилось за etcp_connect_timeout_tb.
* Транслирует NCD_EVENT_TIMEOUT всем handle'ам. После этого init/up коллбэки
* больше не транслируются (entry->timed_out = 1).
*/
static void ncd_connect_timeout_cb(void* arg) {
struct ncd_entry* entry = (struct ncd_entry*)arg;
@@ -282,6 +347,12 @@ static void ncd_connect_timeout_cb(void* arg) {
}
/* ═══════════ асинхронная доставка 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;
@@ -289,13 +360,29 @@ static void ncd_deliver_up_cb(void* arg) {
}
/* ═══════════ FIN_WAIT ═══════════ */
/*
* fin_wait — состояние ожидания подтверждения закрытия от пира.
* Вызывающая сторона: мы отправили CLOSE и ждём ответа (KEEP_ALIVE или DOWN).
* Максимальное время ожидания — NCD_FIN_WAIT_TIMEOUT_TB (5 сек), после чего
* соединение форсированно закрывается.
*/
/*
* 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 за NCD_FIN_WAIT_TIMEOUT_TB (5 сек).
* Форсированно закрываем conn и освобождаем entry — соединение разорвано
* без подтверждения от пира.
*/
static void ncd_fin_wait_timeout_cb(void* arg) {
struct ncd_entry* entry = (struct ncd_entry*)arg;
if (!entry || !entry->conn) return;
@@ -312,6 +399,17 @@ static void ncd_fin_wait_timeout_cb(void* arg) {
}
/* ═══════════ приём CLOSE / KEEP_ALIVE ═══════════ */
/*
* Обрабатывает управляющие сообщения от пира по протоколу graceful shutdown.
*
* CLOSE (пир хочет закрыть conn — у него закончились handle'ы):
* - Если у нас ещё есть handle'ы — отказываем: шлём KEEP_ALIVE, conn живёт.
* - Если handle'ов нет — соглашаемся, закрываем conn.
*
* KEEP_ALIVE (пир просит не закрывать conn — у него ещё есть handle'ы):
* - Если мы в fin_wait — отменяем его, conn продолжает работать.
* - Если у нас handle'ов нет — перезапускаем fin_wait (ждём ещё 5 сек).
*/
static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e) {
if (!conn || !e || e->len < NCD_CONTROL_MSG_SIZE) {
@@ -363,6 +461,11 @@ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e)
}
/* ═══════════ инициализация глобального обработчика ═══════════ */
/*
* Однократная привязка обработчика NCD control-сообщений к ETCP (etcp_bind).
* Вызывается при первом node_conn_direct_open, повторные вызовы безвредны
* (защита через inst->ncd_control_bound).
*/
static void ncd_init_control_binding(struct UTUN_INSTANCE* inst) {
if (inst->ncd_control_bound) return;
@@ -373,6 +476,20 @@ static void ncd_init_control_binding(struct UTUN_INSTANCE* inst) {
/* ═══════════ API ═══════════ */
/*
* Открыть handle для связи с удалённым узлом.
*
* Три сценария (прозрачно для вызывающего):
* 1. Узел уже в NCD-реестре — другой модуль уже открыл соединение.
* Добавляем ещё один handle. Если conn работает — сразу шлём UP.
* 2. ETCP-соединение существует (входящее / etcp_connect), но NCD о нём не знает.
* Оборачиваем в ncd_entry, подписываемся на события.
* 3. Ничего нет — загружаем адреса/pupkey узла, создаём новый ETCP_CONN,
* инициализируем шифрование, создаём линки, запускаем таймер подключения.
*
* Возвращает NCD_NEW (новое), NCD_REUSED (переиспользовано) или NCD_ERR.
* specific_sock=NULL — авто-подбор всех локальных сокетов.
*/
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,
@@ -435,6 +552,9 @@ int node_conn_direct_open(struct UTUN_INSTANCE* inst, uint64_t node_id,
uasync_call_soon(inst->ua, h, ncd_deliver_up_cb);
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open REUSED new-entry (ready) node=0x%016llx conn=%p", (unsigned long long)node_id, (void*)conn);
} else {
{ struct TOPO_NODE* ni = ncd_lookup_node(inst, node_id);
if (ni) { ncd_create_links(entry, ni, specific_sock); topo_node_registry_unref(inst->topo_groups, ni->node_id); }
}
entry->connect_timer = uasync_set_timeout(inst->ua, (int)inst->etcp_connect_timeout_tb, entry, ncd_connect_timeout_cb, "ncd_connect");
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open REUSED new-entry (pending) node=0x%016llx conn=%p", (unsigned long long)node_id, (void*)conn);
}
@@ -514,6 +634,12 @@ int node_conn_direct_open(struct UTUN_INSTANCE* inst, uint64_t node_id,
return NCD_NEW;
}
/*
* То же что open, но TOPO_NODE (адреса + pubkey) уже загружен вызывающим.
* Экономит поиск узла (ncd_lookup_node) — полезно когда адреса/pupkey
* известны заранее (например из BGP-анонса). ni не владеем — можно
* передать временную структуру, копия не делается.
*/
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,
@@ -575,6 +701,7 @@ int node_conn_direct_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id,
uasync_call_soon(inst->ua, h, ncd_deliver_up_cb);
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open_node REUSED new-entry (ready) node=0x%016llx conn=%p", (unsigned long long)node_id, (void*)conn);
} else {
ncd_create_links(entry, ni, specific_sock);
entry->connect_timer = uasync_set_timeout(inst->ua, (int)inst->etcp_connect_timeout_tb, entry, ncd_connect_timeout_cb, "ncd_connect_node");
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open_node REUSED new-entry (pending) node=0x%016llx conn=%p", (unsigned long long)node_id, (void*)conn);
}
@@ -645,6 +772,18 @@ int node_conn_direct_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id,
}
}
/*
* Закрыть handle (graceful shutdown).
*
* Если остались другие handle'ы — просто удаляется из списка, conn продолжает работу.
*
* Если это был последний handle (handle_count стал 0):
* - Запускается протокол graceful shutdown: пиру отправляется CLOSE, conn
* переводится в fin_wait, ставится таймер на 5 сек.
* - Если пир ответит KEEP_ALIVE (у него ещё есть handle'ы) — conn останется жив,
* просто без наших handle'ов.
* - Если пир не ответит — conn закрывается форсированно по таймауту.
*/
void node_conn_direct_close(struct NODE_CONN_DIRECT* h) {
if (!h) return;
struct ncd_entry* entry = h->entry;
@@ -658,9 +797,21 @@ void node_conn_direct_close(struct NODE_CONN_DIRECT* h) {
while (*pp) { if (*pp == h) { *pp = h->next; break; } pp = &(*pp)->next; }
entry->handle_count--;
if (entry->handle_count < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] FATAL: handle_count=%d < 0 node=0x%016llx",
entry->handle_count, (unsigned long long)node_id);
h->entry = NULL; u_free(h); return;
}
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] close node=0x%016llx remaining=%d", (unsigned long long)node_id, entry->handle_count);
if (entry->handle_count <= 0) {
if (entry->handle_count == 0) {
if (entry->handles) {
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] FATAL: handle_count=0 but handles non-empty node=0x%016llx",
(unsigned long long)node_id);
entry->handles = NULL;
}
h->cb = NULL; h->cb_arg = NULL;
if (entry->connect_timer) { uasync_cancel_timeout(entry->ua, entry->connect_timer); entry->connect_timer = NULL; }
if (conn && conn->state != 2) {
/* устанавливаем fin_wait, отправляем CLOSE, ставим таймер */
@@ -694,6 +845,73 @@ void node_conn_direct_close(struct NODE_CONN_DIRECT* h) {
u_free(h);
}
/*
* Немедленное жёсткое закрытие handle. Никакого CLOSE/KEEP_ALIVE — сразу
* снимает ETCP-коллбэки и (если последний handle) закрывает conn.
*
* Используется при разрушении CM entry чтобы гарантировать что коллбэки
* не вызовутся в уже освобождённую память. После force_close никакие
* NCD-события не будут доставлены.
*/
void node_conn_direct_force_close(struct NODE_CONN_DIRECT* h) {
if (!h) return;
struct ncd_entry* entry = h->entry;
if (!entry) { u_free(h); return; }
uint64_t node_id = entry->node_id;
struct ETCP_CONN* conn = entry->conn;
struct NODE_CONN_DIRECT** pp = &entry->handles;
while (*pp) { if (*pp == h) { *pp = h->next; break; } pp = &(*pp)->next; }
entry->handle_count--;
if (entry->handle_count < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] FATAL: force_close handle_count=%d < 0 node=0x%016llx",
entry->handle_count, (unsigned long long)node_id);
h->entry = NULL; u_free(h); return;
}
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] force_close node=0x%016llx remaining=%d",
(unsigned long long)node_id, entry->handle_count);
if (entry->handle_count == 0) {
if (entry->handles) {
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] FATAL: force_close handle_count=0 but handles non-empty node=0x%016llx",
(unsigned long long)node_id);
entry->handles = NULL;
}
h->cb = NULL; h->cb_arg = NULL;
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 (conn) {
conn->fin_wait = 0; conn->fin_wait_clear_cb = NULL; conn->fin_wait_clear_arg = NULL;
etcp_conn_remove_cbk(conn, ncd_init_cb, entry);
etcp_conn_remove_cbk(conn, ncd_up_cb, entry);
etcp_conn_remove_cbk(conn, ncd_down_cb, entry);
if (conn->state != 2) uasync_call_soon(entry->ua, conn, ncd_deferred_close_conn);
ncd_registry_remove(conn->instance, entry);
}
u_free(entry);
}
h->entry = NULL;
u_free(h);
}
/*
* Сменить или сбросить (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;
}
/*
* Прямой доступ к ETCP_CONN из handle.
* Нужен для отправки данных (etcp_send), проверки статуса, и т.д.
*/
struct ETCP_CONN* node_conn_direct_get_conn(struct NODE_CONN_DIRECT* h) {
if (!h || !h->entry) return NULL;
return h->entry->conn;
src/transport_layer/node_conn_direct.h
+28
View File
@@ -1,6 +1,23 @@
#ifndef NODE_CONN_DIRECT_H
#define NODE_CONN_DIRECT_H
/*
* node_conn_direct.h — handle-based прямое подключение к удалённому узлу (node_id)
*
* Модуль управляет ETCP-соединениями через непрозрачные handle'ы (NODE_CONN_DIRECT).
* Несколько handle'ов могут разделять одно ETCP_CONN — закрытие происходит только
* когда все handle'ы закрыты (refcounting). При закрытии последнего handle
* используется graceful shutdown: CLOSE/KEEP_ALIVE протокол (fin_wait 5 сек).
*
* Кто использует:
* - conn_mgr (Connection Manager) — создаёт handle'ы для DIR/REV/IND-соединений
* - chat_core / topo_group — связь с узлами канала/группы
* - любой модуль, кому нужно надёжное ETCP-соединение с конкретным node_id
*
* События (UP/DOWN/TIMEOUT) доставляются через единый callback асинхронно
* (uasync_call_soon), что исключает рекурсию и проблемы с порядком инициализации.
*/
#include <stdint.h>
struct UTUN_INSTANCE;
@@ -30,6 +47,7 @@ typedef void (*ncd_callback)(struct NODE_CONN_DIRECT* h, enum ncd_event event, v
* Node info ищется через node_registry, fallback — SQLite.
* cb вызывается при изменении статуса / таймауте.
* Если conn уже готов — cb(NCD_EVENT_UP) через uasync_call_soon (не синхронно).
* Пока не придёт событие UP - отправляения могут теряться. Поэтому перед первой отправкой всегда ждём UP.
*/
int node_conn_direct_open(struct UTUN_INSTANCE* inst, uint64_t node_id,
ncd_callback cb, void* cb_arg,
@@ -48,6 +66,16 @@ int node_conn_direct_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id,
void node_conn_direct_close(struct NODE_CONN_DIRECT* h);
/* Force close без fin_wait: немедленно удаляет ETCP-коллбэки и, если это
* последний handle, закрывает соединение и освобождает ncd_entry.
* Используется при уничтожении CM entry — гарантирует, что после вызова
* никакие NCD-коллбэки не доставят события в освобождённую память. */
void node_conn_direct_force_close(struct NODE_CONN_DIRECT* h);
/* Сменить или сбросить (cb=NULL) callback на уже открытом handle.
* Не влияет на refcounting и состояние conn. */
void node_conn_direct_set_callback(struct NODE_CONN_DIRECT* h, ncd_callback cb, void* cb_arg);
struct ETCP_CONN* node_conn_direct_get_conn(struct NODE_CONN_DIRECT* h);
/* ─── Протокол CLOSE / KEEP_ALIVE (ETCP_RT_ID_NCD_CONTROL = 0x12) ─── */