Use one NCD address matcher for opens updates and new sockets
This commit is contained in:
1 parent
b23d5e6047
commit
772c0728e1
1 file changed
+100
-190
@@ -148,42 +148,39 @@ static struct TOPO_NODE* ncd_lookup_node(struct UTUN_INSTANCE* inst, uint64_t no
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* ═══════════ создание линков (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 больше не нужен)
|
||||
*/
|
||||
/* Все разрешённые пары адрес/сокет: UDP, TCP, IPv4/IPv6 и REALITY.
|
||||
* Явный specific_sock в open обходит локальную политику, но не проверку семейства/протокола/цели. */
|
||||
|
||||
/* Создаёт один UDP-линк на use_sock → sa, с вытеснением stale-линка и дедупом. */
|
||||
#define NCD_MAX_SOCKET_VIEWS 32
|
||||
#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) return 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;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "[ncd] evict stale link node=0x%016llx(old pk=%016llx) for new node=0x%016llx(pk=%016llx)",
|
||||
(unsigned long long)old->peer_node_id, *(const uint64_t*)old->crypto_ctx.peer_public_key,
|
||||
(unsigned long long)conn->peer_node_id, *(const uint64_t*)conn->crypto_ctx.peer_public_key);
|
||||
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);
|
||||
etcp_cbk_fire(old, ETCP_CBK_EVENT_NODE_CHANGED);
|
||||
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 0;
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
if (stale && stale->etcp != conn
|
||||
@@ -195,9 +192,12 @@ static int ncd_add_udp_link(struct ETCP_CONN* conn, struct ETCP_SOCKET* use_sock
|
||||
(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 0;
|
||||
return -1;
|
||||
}
|
||||
return etcp_link_new(conn, use_sock, sa, 0) ? 1 : 0;
|
||||
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) {
|
||||
@@ -220,6 +220,7 @@ 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);
|
||||
@@ -229,23 +230,28 @@ static int ncd_add_tcp_link(struct ETCP_CONN* conn, struct ETCP_SOCKET* sock,
|
||||
(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->link_status && !link->tcp_link && !link->tcp_reconnect_timer) {
|
||||
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);
|
||||
}
|
||||
return 0;
|
||||
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);
|
||||
return 1;
|
||||
int cancelled = link->closing || conn->close_requested;
|
||||
etcp_link_release(link);
|
||||
return cancelled ? -1 : 1;
|
||||
}
|
||||
|
||||
/* Проверка совместимости и создание используют один обход пар, включая TCP fallback и IPv6. */
|
||||
static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni,
|
||||
struct ETCP_SOCKET* specific_sock, struct UTUN_INSTANCE* check_inst) {
|
||||
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",
|
||||
@@ -275,7 +281,7 @@ static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni,
|
||||
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; memcpy(&sa, &sin, sizeof(sin));
|
||||
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)
|
||||
@@ -295,28 +301,29 @@ static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni,
|
||||
const struct sock_view* v = &views[vi];
|
||||
if (!v->is_tcp) continue;
|
||||
tcp_sockets++;
|
||||
if (specific_sock ? v->sock != specific_sock : !sock_match(v, a->type, tcfg, profile, tip)) continue;
|
||||
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) link_count += added;
|
||||
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) link_count += added;
|
||||
if (added < 0 || entry->closed || conn->close_requested) return -1;
|
||||
link_count += added;
|
||||
}
|
||||
}
|
||||
if (a->protocol & TOPO_PROTO_UDP) {
|
||||
if (specific_sock && !specific_sock->is_tcp && specific_sock->local_addr.ss_family == AF_INET) {
|
||||
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;
|
||||
link_count += ncd_add_udp_link(conn, specific_sock, &sa);
|
||||
} else if (!specific_sock) {
|
||||
for (int vi = 0; vi < nviews; vi++) {
|
||||
const struct sock_view* v = &views[vi];
|
||||
if (v->is_tcp || !sock_match(v, a->type, tcfg, profile, tip)) continue;
|
||||
if (check_inst) return 1;
|
||||
link_count += ncd_add_udp_link(conn, v->sock, &sa);
|
||||
}
|
||||
int added = ncd_add_udp_link(conn, v->sock, &sa);
|
||||
if (added < 0 || entry->closed || conn->close_requested) return -1;
|
||||
link_count += added;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -344,39 +351,36 @@ static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni,
|
||||
(unsigned long long)nid, a->socket_id); continue;
|
||||
}
|
||||
int tcp_sockets = 0;
|
||||
if (is_ll) { struct ETCP_SOCKET* sv = inst->etcp_sockets;
|
||||
while (sv) { if (!sv->probe_only && sv->local_addr.ss_family == AF_INET6 && sv->netif_index) { sin6.sin6_scope_id = sv->netif_index; break; } sv = sv->next; } }
|
||||
struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6));
|
||||
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 : global && s->type == CFG_SERVER_TYPE_LOCAL) continue;
|
||||
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) link_count += added;
|
||||
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) link_count += added;
|
||||
if (added < 0 || entry->closed || conn->close_requested) return -1;
|
||||
link_count += added;
|
||||
}
|
||||
}
|
||||
if (a->protocol & TOPO_PROTO_UDP) {
|
||||
if (specific_sock && !specific_sock->is_tcp && specific_sock->local_addr.ss_family == AF_INET6) {
|
||||
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 = specific_sock->netif_index;
|
||||
struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6));
|
||||
link_count += ncd_add_udp_link(conn, specific_sock, &sa);
|
||||
} else if (!specific_sock) {
|
||||
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 ||
|
||||
(global && 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; memcpy(&sa, &sin6, sizeof(sin6));
|
||||
link_count += ncd_add_udp_link(conn, s, &sa);
|
||||
}
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -394,7 +398,7 @@ int node_conn_direct_compatible(struct UTUN_INSTANCE* inst, uint64_t 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) { 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, inst) > 0) return 1;
|
||||
if (ncd_create_links(NULL, node, NULL, 0, inst) > 0) return 1;
|
||||
/* Входящее подключение тоже допустимо: REVERSE_REQ просит пира открыть наш опубликованный адрес.
|
||||
* TCP имеет собственный исходящий socket fallback; для UDP нужен известный UDP-сокет пира.
|
||||
* Это возможность попытки, не подтверждение маршрута/работоспособности скрытых сокетов пира. */
|
||||
@@ -448,7 +452,7 @@ int node_conn_direct_update_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
|
||||
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, NULL);
|
||||
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);
|
||||
@@ -818,7 +822,7 @@ static int ncd_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id, ncd_callb
|
||||
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, NULL) < 0) goto done;
|
||||
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;
|
||||
}
|
||||
@@ -979,135 +983,41 @@ struct ETCP_CONN* node_conn_direct_get_conn(struct NODE_CONN_DIRECT* h) {
|
||||
|
||||
/* ═══════════ Управление линками на конкретном сокете (для auto_socket) ═══════════ */
|
||||
|
||||
/* ncd_add_udp_link_random / ncd_add_tcp_link_random — собирают совместимые адреса пира,
|
||||
* случайно выбирают один, проверяют что линк на (sock, addr, proto) ещё не существует,
|
||||
* и создают новый. */
|
||||
static int ncd_add_udp_link_random(struct ETCP_CONN* conn, struct TOPO_NODE* ni, struct ETCP_SOCKET* sock) {
|
||||
int family = sock->local_addr.ss_family;
|
||||
if (family == AF_INET) {
|
||||
struct TOPO_ADDR4* compat[64]; int count = 0;
|
||||
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) {
|
||||
if (!a->port || !(a->protocol & TOPO_PROTO_UDP)) continue;
|
||||
int zero = 1; for (int j = 0; j < 4; j++) if (a->addr[j]) { zero = 0; break; }
|
||||
if (zero) continue;
|
||||
if (count < 64) compat[count++] = (struct TOPO_ADDR4*)a;
|
||||
}
|
||||
if (count == 0) return 0;
|
||||
uint32_t r; random_bytes((uint8_t*)&r, sizeof(r));
|
||||
struct TOPO_ADDR4* a = compat[r % count];
|
||||
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; memcpy(&sa, &sin, sizeof(sin));
|
||||
if (etcp_link_find_by_addr(sock, &sa, 0)) return 0;
|
||||
if (etcp_link_new(conn, sock, &sa, 0)) return 1;
|
||||
return 0;
|
||||
}
|
||||
if (family == AF_INET6) {
|
||||
struct TOPO_ADDR6* compat[64]; int count = 0;
|
||||
for (const struct TOPO_ADDR6* a = ni->v6_addrs; a; a = a->next) {
|
||||
if (!a->port || !(a->protocol & TOPO_PROTO_UDP)) continue;
|
||||
int zero = 1; for (int j = 0; j < 16; j++) if (a->addr[j]) { zero = 0; break; }
|
||||
if (zero) continue;
|
||||
if (count < 64) compat[count++] = (struct TOPO_ADDR6*)a;
|
||||
}
|
||||
if (count == 0) return 0;
|
||||
uint32_t r; random_bytes((uint8_t*)&r, sizeof(r));
|
||||
struct TOPO_ADDR6* a = compat[r % count];
|
||||
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->addr[0] == 0xfe && (a->addr[1] & 0xc0) == 0x80)) sin6.sin6_scope_id = sock->netif_index;
|
||||
struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6));
|
||||
if (etcp_link_find_by_addr(sock, &sa, 0)) return 0;
|
||||
if (etcp_link_new(conn, sock, &sa, 0)) return 1;
|
||||
return 0;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int ncd_add_tcp_link_random(struct ETCP_CONN* conn, struct TOPO_NODE* ni, struct ETCP_SOCKET* sock) {
|
||||
int family = sock->local_addr.ss_family;
|
||||
if (family == AF_INET) {
|
||||
struct TOPO_ADDR4* compat[64]; int count = 0;
|
||||
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) {
|
||||
if (!a->port || !(a->protocol & TOPO_PROTO_TCP)) continue;
|
||||
int zero = 1; for (int j = 0; j < 4; j++) if (a->addr[j]) { zero = 0; break; }
|
||||
if (zero) continue;
|
||||
if (count < 64) compat[count++] = (struct TOPO_ADDR4*)a;
|
||||
}
|
||||
if (count == 0) return 0;
|
||||
uint32_t r; random_bytes((uint8_t*)&r, sizeof(r));
|
||||
struct TOPO_ADDR4* a = compat[r % count];
|
||||
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; memcpy(&sa, &sin, sizeof(sin));
|
||||
if (etcp_link_find_by_addr(sock, &sa, 1)) return 0;
|
||||
struct ETCP_LINK* tlink = etcp_link_new(conn, sock, &sa, 0);
|
||||
if (!tlink) return 0;
|
||||
tlink->is_tcp = 1;
|
||||
topo_node_apply_reality(tlink, topo_node_find_reality_sock(ni, a->socket_id));
|
||||
etcp_tcp_link_start_connect(tlink, &sa, a->port);
|
||||
return 1;
|
||||
}
|
||||
if (family == AF_INET6) {
|
||||
struct TOPO_ADDR6* compat[64]; int count = 0;
|
||||
for (const struct TOPO_ADDR6* a = ni->v6_addrs; a; a = a->next) {
|
||||
if (!a->port || !(a->protocol & TOPO_PROTO_TCP)) continue;
|
||||
int zero = 1; for (int j = 0; j < 16; j++) if (a->addr[j]) { zero = 0; break; }
|
||||
if (zero) continue;
|
||||
if (count < 64) compat[count++] = (struct TOPO_ADDR6*)a;
|
||||
}
|
||||
if (count == 0) return 0;
|
||||
uint32_t r; random_bytes((uint8_t*)&r, sizeof(r));
|
||||
struct TOPO_ADDR6* a = compat[r % count];
|
||||
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->addr[0] == 0xfe && (a->addr[1] & 0xc0) == 0x80)) sin6.sin6_scope_id = sock->netif_index;
|
||||
struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6));
|
||||
if (etcp_link_find_by_addr(sock, &sa, 1)) return 0;
|
||||
struct ETCP_LINK* tlink = etcp_link_new(conn, sock, &sa, 0);
|
||||
if (!tlink) return 0;
|
||||
tlink->is_tcp = 1;
|
||||
topo_node_apply_reality(tlink, topo_node_find_reality_sock(ni, a->socket_id));
|
||||
etcp_tcp_link_start_connect(tlink, &sa, a->port);
|
||||
return 1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Снимок entry удерживает список через callbacks, которые могут закрыть соседних владельцев. */
|
||||
int ncd_add_socket_links(struct UTUN_INSTANCE* inst, struct ETCP_SOCKET* sock) {
|
||||
if (!inst || !sock) return 0;
|
||||
int added = 0, total_conns = 0, skipped_closed = 0, skipped_no_addr = 0;
|
||||
|
||||
struct ll_entry* qe = inst->connections ? inst->connections->head : NULL;
|
||||
while (qe) {
|
||||
struct conn_queue_entry* ce = (struct conn_queue_entry*)qe->data;
|
||||
struct ETCP_CONN* conn = ce->conn;
|
||||
qe = qe->next;
|
||||
if (!conn) continue;
|
||||
total_conns++;
|
||||
|
||||
if (conn->state == 2) { skipped_closed++; continue; }
|
||||
|
||||
struct TOPO_NODE* ni = ncd_lookup_node(inst, ce->peer_node_id);
|
||||
if (!ni) { skipped_no_addr++; continue; }
|
||||
|
||||
int n;
|
||||
if (sock->is_tcp) n = ncd_add_tcp_link_random(conn, ni, sock);
|
||||
else n = ncd_add_udp_link_random(conn, ni, sock);
|
||||
if (n > 0) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] add_socket_links sock=%s fd=%d conn=[%s] node=0x%016llx addr=%s is_tcp=%d",
|
||||
sock->name, (int)sock->fd, conn->log_name,
|
||||
(unsigned long long)ce->peer_node_id,
|
||||
sockaddr_storage_to_str(&sock->local_addr).str, sock->is_tcp);
|
||||
added += n;
|
||||
}
|
||||
topo_node_registry_unref(inst->topo_groups, ni->node_id);
|
||||
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;
|
||||
}
|
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_NCD,
|
||||
"[ncd] add_socket_links done: sock=%s is_tcp=%d total_conns=%d skipped(cl=%d,no_addr=%d) added=%d",
|
||||
sock->name, sock->is_tcp, total_conns, skipped_closed, skipped_no_addr, added);
|
||||
return added;
|
||||
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) {
|
||||
|
||||
Reference in new issue
Block a user