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

android: fix key regeneration, Clear DB, uasync eventfd, udp log from config

Browse Source 
- jni_on_event: fix JNI_EDETACHED — AttachCurrentThread/DetachCurrentThread
- u_async.c: pipe→eventfd for reliable cross-thread wakeup
  add process_posted_tasks in poll() path, fix EINTR (saved_errno)
- bridge_debug_hook: keep udp_log_write, remove mutex from udp_log
- memory_pool.c: alloc/free logs INFO→DEBUG (reduce spam)
- instance_lite.c: chat_event_set_handler before ensure_keys,
  KEYS_GENERATED events, fchmod 0600 on keys.conf,
  UDP log from config [log_udp] section, regenerate_trampoline
- config_parser: parse [log_udp] ip/port
- NativeLib.kt: remove callbacksSet, simplify init/stop/destroy
- ChatViewModel.kt: unified event callback in init, pubKey StateFlow,
  requestRegenerateKeys, clearState fix
- SettingsScreen.kt: pubKey.collectAsState(), Clear DB via restart()
- HeadlessService.kt: remove setUdpLogTarget (now from config)
- android_udp_log.c: simplified — no mutex, buffer+send+timer in one thread
- utun_instance.c: NULL-check tcp_connections before destroy
- chat_event.h: +CHAT_EVT_KEYS_GENERATED 11, -NODE_CHANGED
- logreceiver: UDP log receiver tool
This commit is contained in:
evgeny committed 2026-07-28 16:29:33 +03:00
1 parent d0cca27edd
commit 064bed06f2
68 files changed
+5090 -291
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/chat_event.c
+1 -1
View File
@@ -21,7 +21,7 @@ void chat_event_post(int type, const uint8_t* data, int len) {
[1]="MSG_RECEIVED", [2]="CONNECT_RESULT", [3]="NEW_PEER",
[4]="CHANNEL_UPDATED", [5]="MEMBERS_CHANGED", [6]="MY_NODE_ID",
[7]="AUTO_CONNECT_STATUS", [8]="CHANNEL_PEERS_ONLINE",
[9]="DB_READY", [10]="STATUS_REFRESH", [11]="NODE_CHANGED",
[9]="DB_READY", [10]="STATUS_REFRESH", [11]="KEYS_GENERATED",
};
const char* n = (type >= 1 && type <= 11) ? names[type] : "?";
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "chat_event: %s(%d) data=%d bytes", n, type, len);
src/chat/chat_event.h
+1 -1
View File
@@ -29,7 +29,7 @@ extern "C" {
#define CHAT_EVT_CHANNEL_PEERS_ONLINE 8 /* [ch_id_len:1][ch_id:var][online_count:2] */
#define CHAT_EVT_DB_READY 9 /* data: none */
#define CHAT_EVT_STATUS_REFRESH 10 /* data: status text */
#define CHAT_EVT_NODE_CHANGED 11 /* [node_id:8] */
#define CHAT_EVT_KEYS_GENERATED 11 /* [pub_hex:64] */
typedef void (*chat_event_handler_fn)(int type, const uint8_t* data, int len);
src/chat/chat_sync.c
+16 -28
View File
@@ -215,16 +215,10 @@ static void cs_join_timeout_cb(void* arg) {
/* ── helper: find active ETCP_CONN for node ── */
static struct ETCP_CONN* cs_find_conn_for_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
if (conn && conn->initialized && conn->links_up) return conn;
if (!inst->connections) { DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn connections=NULL", CS_ID); return NULL; }
struct ll_entry* e = queue_find_data_by_index(inst->connections, (const uint8_t*)&node_id);
if (e) {
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn node=%016llx found=%p initialized=%d links_up=%d state=%d",
CS_ID, (unsigned long long)node_id, (void*)ce->conn, ce->conn->initialized, ce->conn->links_up, ce->conn->state);
if (ce->conn->initialized && ce->conn->links_up) return ce->conn;
return NULL;
}
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn node=%016llx NOT FOUND in queue (head=%p count=%d)",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn node=%016llx NOT FOUND (head=%p count=%d)",
CS_ID, (unsigned long long)node_id, (void*)inst->connections->head, queue_entry_count(inst->connections));
return NULL;
}
@@ -233,15 +227,9 @@ static struct ETCP_CONN* cs_find_conn_for_node(struct UTUN_INSTANCE* inst, uint6
static int cs_is_peer_online(struct UTUN_INSTANCE* inst, uint64_t peer_id) {
if (!inst->connections) { DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: is_online peer=0x%016llx CONNS=NULL", CS_ID, (unsigned long long)peer_id); return 0; }
struct ll_entry* e = queue_find_data_by_index(inst->connections, (const uint8_t*)&peer_id);
if (e) {
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
int onl = ce->conn->initialized && ce->conn->links_up;
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: is_online peer=0x%016llx found=1 init=%d links=%d online=%d",
CS_ID, (unsigned long long)peer_id, ce->conn->initialized, ce->conn->links_up, onl);
return onl;
}
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: is_online peer=0x%016llx found=0", CS_ID, (unsigned long long)peer_id);
struct ETCP_CONN* conn = instance_find_conn(inst, peer_id);
if (conn) return conn->initialized && conn->links_up;
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: is_online peer=0x%016llx NOT FOUND", CS_ID, (unsigned long long)peer_id);
return 0;
}
@@ -601,17 +589,17 @@ static void cm_invite_trampoline(void* arg) {
/* save invite addresses (socket_id=0) */
sqlite3_prepare_v2(db, "DELETE FROM node_addresses WHERE node_id=? AND socket_id=0", -1, &st, NULL);
if (st) { sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id); sqlite3_step(st); sqlite3_finalize(st); }
sqlite3_prepare_v2(db, "INSERT INTO node_addresses(node_id,family,protocol,address,port,addr_type,socket_id) VALUES(?,?,1,?,?,0,?)", -1, &st, NULL);
sqlite3_prepare_v2(db, "INSERT INTO node_addresses(node_id,family,protocol,address,port,addr_type,socket_id) VALUES(?,?,?,?,?,0,?)", -1, &st, NULL);
if (st) {
const uint8_t* ap = inv->addrs_data;
for (int i = 0; i < inv->addr_count; i++) {
uint8_t family = *ap++; ap++; /* skip sock_id */
uint8_t proto = *ap++; /* read proto */
if (family != 4) { ap += 18; continue; }
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id); sqlite3_bind_int(st, 2, 4);
sqlite3_bind_blob(st, 3, ap, 4, SQLITE_STATIC); ap += 4;
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id); sqlite3_bind_int(st, 2, 4); sqlite3_bind_int(st, 3, proto);
sqlite3_bind_blob(st, 4, ap, 4, SQLITE_STATIC); ap += 4;
uint16_t port = ((uint16_t)ap[0] << 8) | ap[1]; ap += 2;
sqlite3_bind_int(st, 4, (int)port); sqlite3_bind_int(st, 5, 1); /* addr_type=1 (DIRECT) */
sqlite3_bind_int(st, 5, (int)port); sqlite3_bind_int(st, 6, 1); /* addr_type=1 (DIRECT) */
sqlite3_step(st); sqlite3_reset(st);
}
sqlite3_finalize(st);
@@ -634,6 +622,10 @@ static void cm_invite_trampoline(void* arg) {
const uint8_t* ap = inv->addrs_data;
for (int i = 0; i < inv->addr_count; i++) {
uint8_t family = *ap++; uint8_t sock_id = *ap++; uint8_t proto = *ap++;
uint16_t port = 0;
if (family == 4) { const uint8_t* a = ap; port = ((uint16_t)ap[4] << 8) | ap[5]; }
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: invite addr[%d] family=%d proto=%d sock=%d %d.%d.%d.%d:%d",
CS_ID, i, family, proto, sock_id, ap[0], ap[1], ap[2], ap[3], port);
if (family != 4) { ap += 18; continue; }
struct TOPO_ADDR4* a4 = u_calloc(1, sizeof(*a4));
if (!a4) break;
@@ -650,12 +642,8 @@ static void cm_invite_trampoline(void* arg) {
}
if (g_cs && g_cs->pending_invite_ch_id != 0 && !g_cs->info_req_timer) {
struct ETCP_CONN* conn = NULL;
{
struct ll_entry* e = queue_find_data_by_index(inst->connections, (const uint8_t*)&node_id);
if (e) { struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data; if (ce->conn->links_up) conn = ce->conn; }
}
if (conn) {
struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
if (conn && conn->links_up) {
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: invite conn already UP node=0x%016llx, triggering cs_on_conn_up", CS_ID, (unsigned long long)node_id);
cs_on_conn_up(conn, NULL);
}
src/chat/db_sync.c
+7 -14
View File
@@ -384,14 +384,11 @@ static int db_get_ed25519_pubkey(struct DB_SYNC* db, uint64_t node_id, uint8_t o
if (db->inst->topo_sqlite_db) {
if (topo_node_sqlite_get_ed25519_pubkey(db->inst->topo_sqlite_db, node_id, out) == 0) return 0;
}
struct ll_entry* e = queue_find_data_by_index(db->inst->connections, (const uint8_t*)&node_id);
if (e) {
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
if (ce->conn) {
memcpy(out, ce->conn->peer_ed25519_pubkey, 32);
uint64_t chk; memcpy(&chk, out, 8);
if (chk != 0) return 0;
}
struct ETCP_CONN* conn = instance_find_conn(db->inst, node_id);
if (conn) {
memcpy(out, conn->peer_ed25519_pubkey, 32);
uint64_t chk; memcpy(&chk, out, 8);
if (chk != 0) return 0;
}
return -1;
}
@@ -545,12 +542,8 @@ static int db_sync_send_hash(struct DB_SYNC* db, uint64_t node_id, uint64_t hash
entry->dgram = buf;
entry->len = plen + 9;
struct ETCP_CONN* conn = NULL;
{
struct ll_entry* e = queue_find_data_by_index(db->inst->connections, (const uint8_t*)&node_id);
if (e) { struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data; if (ce->conn->links_up) conn = ce->conn; }
}
if (!conn) {
struct ETCP_CONN* conn = instance_find_conn(db->inst, node_id);
if (!conn || !conn->links_up) {
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "sync: send DROP — no direct conn to %04llX (hash=%016llx, type=%02x)",
(unsigned long long)(node_id >> 16), hash, plen > 0 ? payload[0] : 0);
queue_entry_free(entry);
src/chat/merkle_sync.c
+5 -5
View File
@@ -215,11 +215,11 @@ static int _get_level_hashes(struct merkle_sync* ms, const char* ns,
static struct ETCP_CONN* ms_find_conn_for_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
if (!inst->connections) { DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — inst->connections is NULL", MS_ID); return NULL; }
struct ll_entry* e = queue_find_data_by_index(inst->connections, (const uint8_t*)&node_id);
if (!e) { DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — node=%016llx NOT FOUND in connections", MS_ID, (unsigned long long)node_id); return NULL; }
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — node=%016llx found=%d initialized=%d links_up=%d", MS_ID, (unsigned long long)node_id, 1, ce->conn->initialized, ce->conn->links_up);
if (ce->conn->initialized && ce->conn->links_up) return ce->conn;
struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
if (!conn) { DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — node=%016llx NOT FOUND", MS_ID, (unsigned long long)node_id); return NULL; }
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — node=%016llx found=%d initialized=%d links_up=%d",
MS_ID, (unsigned long long)node_id, 1, conn->initialized, conn->links_up);
if (conn->initialized && conn->links_up) return conn;
return NULL;
}
src/config_parser.c
+7 -1
View File
@@ -49,7 +49,8 @@ typedef enum {
SECTION_TCP_PROXY_SERVER,
SECTION_NETWORK,
SECTION_GUI,
SECTION_NTP
SECTION_NTP,
SECTION_LOG_UDP
} section_type_t;
static char* trim(char *str) {
@@ -754,6 +755,7 @@ static section_type_t parse_section_header(const char *line, char *name, size_t
if (strcasecmp(section, "tcp_proxy_server") == 0) return SECTION_TCP_PROXY_SERVER;
if (strcasecmp(section, "ntp") == 0) return SECTION_NTP;
if (strcasecmp(section, "gui") == 0) return SECTION_GUI;
if (strcasecmp(section, "log_udp") == 0) return SECTION_LOG_UDP;
char *colon = strchr(section, ':');
if (!colon) return SECTION_UNKNOWN;
@@ -1107,6 +1109,10 @@ struct utun_config* parse_config_from_buf(const char *buf, size_t len, const cha
case SECTION_NETWORK: if (cur_network) parse_network(key, value, cur_network, filename, line_num); break;
case SECTION_NTP: if (strcmp(key, "enabled") == 0) cfg->global.ntp_enabled = strcasecmp(value, "yes") == 0 || strcasecmp(value, "1") == 0 || strcasecmp(value, "true") == 0; else if (strcmp(key, "server") == 0) { struct CFG_NTP_SERVER *ns = u_calloc(1, sizeof(struct CFG_NTP_SERVER)); if (ns) { strncpy(ns->name, value, sizeof(ns->name) - 1); ns->name[sizeof(ns->name) - 1] = '\0'; ns->next = cfg->global.ntp_servers; cfg->global.ntp_servers = ns; cfg->global.ntp_server_count++; } } else if (strcmp(key, "interval") == 0) { cfg->global.ntp_resync_interval = atoi(value); if (cfg->global.ntp_resync_interval < 60) cfg->global.ntp_resync_interval = 60; } else DEBUG_ERROR(DEBUG_CATEGORY_CONFIG, "%s:%d: Unknown ntp option '%s'", filename, line_num, key); break;
case SECTION_GUI: break;
case SECTION_LOG_UDP:
if (strcmp(key, "ip") == 0) strncpy(cfg->global.log_udp_ip, value, sizeof(cfg->global.log_udp_ip) - 1);
else if (strcmp(key, "port") == 0) cfg->global.log_udp_port = atoi(value);
break;
default: DEBUG_WARN(DEBUG_CATEGORY_CONFIG, "%s:%d: Key outside section: %s", filename, line_num, key); break;
}
}
src/config_parser.h
+3
View File
@@ -120,6 +120,9 @@ struct global_config {
char db_path[256]; // Path to SQLite database (empty = disabled)
int db_sync_enabled; // 1 = enable distributed DB sync (default: 0)
uint32_t db_sync_ttl; // TTL for unsent records in seconds (default: 86400)
// UDP log target
char log_udp_ip[64];
int log_udp_port;
char debug_level[16]; // debug level: error, warn, info, debug, trace
int enable_timestamp; // enable timestamps in logs
int enable_function_names; // enable function names in logs
src/routing_layer/conn_mgr.c
+81 -25
View File
@@ -20,6 +20,7 @@
#include "topo_node.h"
#include "topo_node_sqlite.h"
#include "topo_group.h"
#include "../transport_layer/stcp_link.h"
#include "route_ping.h"
#include "route_connectivity.h"
#include "conn_mgr.h"
@@ -95,6 +96,8 @@ struct cm_invite_pending {
uint32_t timeout_ms;
uint8_t state;
struct ETCP_CONN* conn;
struct stcp_link* tcp_link;
uint8_t tcp_ready : 1;
void* overall_timer;
conn_mgr_connect_callback_t cb;
void* cb_arg;
@@ -269,17 +272,7 @@ int conn_mgr_connect_node(struct CONN_MGR* mgr, uint64_t node_id, uint32_t idle_
return CONN_MGR_OK;
}
struct ETCP_CONN* existing = topo_group_find_conn_for_node(group, node_id);
if (!existing) {
struct ll_entry* ce = queue_find_data_by_index(mgr->instance->connections, (const uint8_t*)&node_id);
while (ce) {
struct conn_queue_entry* cqe = (struct conn_queue_entry*)ce->data;
if (cqe->conn && cqe->conn->peer_node_id == node_id) {
existing = cqe->conn;
break;
}
ce = queue_find_next_by_index(mgr->instance->connections, (const uint8_t*)&node_id, ce);
}
}
if (!existing) existing = instance_find_conn(mgr->instance, node_id);
if (existing && existing->peer_node_id == node_id && existing->links) {
struct ETCP_LINK* l = existing->links;
while (l) { if (l->link_status && l->initialized) break; l = l->next; }
@@ -1203,8 +1196,7 @@ static void cm_handle_invite_info_req(struct CONN_MGR* mgr, struct ETCP_CONN* co
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: INVITE_INFO_REQ from 0x%016llx group=0x%016llx",
(unsigned long long)peer_node_id, (unsigned long long)req->group_id);
uint8_t is_member = 0;
if (req->group_id != 0) is_member = (uint8_t)cm_check_group_member(mgr, req->group_id, peer_node_id);
uint8_t is_member = 1; /* invite link itself is authorization — no group membership check needed */
struct TOPO_GROUP* group = topo_groups_find(mgr->instance->topo_groups, req->group_id);
struct TOPO_GROUP* src_group = group ? group : mgr->group;
@@ -1282,6 +1274,7 @@ static void cm_handle_invite_info_resp(struct CONN_MGR* mgr, const uint8_t* data
static void cm_invite_init_cb(struct ETCP_CONN* conn, void* arg) {
struct cm_invite_pending* inv = (struct cm_invite_pending*)arg;
if (inv->state != CM_INVITE_CONNECTING) return;
if (!conn || conn->peer_node_id != inv->node_id) {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: invite init cb peer mismatch conn=0x%016llx inv=0x%016llx",
(unsigned long long)(conn ? conn->peer_node_id : 0), (unsigned long long)inv->node_id);
@@ -1311,11 +1304,46 @@ static void cm_invite_overall_timeout(void* arg) {
cm_invite_fail(inv, CONN_MGR_ERR_TIMEOUT);
}
/* ── TCP invite callbacks ── */
static void cm_tcp_ready_cb(struct stcp_link* link, void* arg) {
struct cm_invite_pending* inv = (struct cm_invite_pending*)arg;
if (!inv || inv->tcp_ready) return;
inv->tcp_ready = 1;
struct ETCP_CONN* etcp = stcp_link_get_etcp_conn(link);
etcp->peer_node_id = inv->node_id;
struct ll_entry* qe = queue_entry_new(sizeof(struct tcp_conn_entry));
if (!qe) { cm_invite_fail(inv, CONN_MGR_ERR_INTERNAL); return; }
struct tcp_conn_entry* te = (struct tcp_conn_entry*)qe->data;
te->node_id = inv->node_id; te->link = link; te->etcp_conn = etcp;
queue_data_put_with_index(inv->mgr->instance->tcp_connections, qe);
inv->conn = etcp;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: TCP invite ready node=0x%016llx", (unsigned long long)inv->node_id);
cm_invite_init_cb(etcp, inv);
}
static void cm_tcp_close_cb(struct stcp_link* link, int err, void* arg) {
struct cm_invite_pending* inv = (struct cm_invite_pending*)arg;
if (!inv) return;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: TCP invite closed node=0x%016llx err=%d",
(unsigned long long)inv->node_id, err);
inv->tcp_link = NULL;
}
static void cm_invite_fail(struct cm_invite_pending* inv, int result) {
if (!inv) return;
struct CONN_MGR* mgr = inv->mgr;
if (inv->overall_timer) { uasync_cancel_timeout(mgr->instance->ua, inv->overall_timer); inv->overall_timer = NULL; }
if (inv->conn) { etcp_connection_close(inv->conn); inv->conn = NULL; }
if (inv->tcp_link) {
struct ll_entry* e = queue_find_data_by_index(mgr->instance->tcp_connections, (const uint8_t*)&inv->node_id);
if (e) queue_remove_data(mgr->instance->tcp_connections, e);
stcp_link_close(inv->tcp_link); inv->tcp_link = NULL; inv->conn = NULL;
} else if (inv->conn) {
etcp_connection_close(inv->conn); inv->conn = NULL;
}
if (inv->temp_nq) {
struct TOPO_GROUP* group = mgr->group;
queue_remove_data(group->nodes, &inv->temp_nq->ll);
@@ -1378,29 +1406,57 @@ int conn_mgr_connect_from_invite(struct CONN_MGR* mgr, struct TOPO_NODE* ni,
queue_data_put_with_index(group->nodes, &nq->ll);
int connected = 0;
int udp_connected = 0, tcp_connected = 0;
struct ETCP_CONN* udp_conn = NULL;
/* ── UDP: all addresses × all local sockets (single ETCP_CONN, no break) ── */
for (const struct TOPO_ADDR4* a = group_ni->v4_addrs; a; a = a->next) {
if (a->protocol != TOPO_PROTO_UDP) continue;
if (!(a->protocol & TOPO_PROTO_UDP)) continue;
struct ETCP_SOCKET* s = mgr->instance->etcp_sockets;
while (s) {
if (s->local_addr.ss_family == AF_INET) {
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; memset(&sa, 0, sizeof(sa)); memcpy(&sa, &sin, sizeof(sin));
struct ETCP_CONN* conn = etcp_connection_create(mgr->instance, NULL);
if (conn) {
etcp_conn_add_init_cbk(conn, cm_invite_init_cb, inv);
sc_init_ctx(&conn->crypto_ctx, &mgr->instance->my_keys);
sc_set_peer_public_key(&conn->crypto_ctx, group_ni->public_key, 0);
if (etcp_link_new(conn, s, &sa, 0)) { inv->conn = conn; connected = 1; break; }
etcp_connection_close(conn);
if (!udp_conn) {
udp_conn = etcp_connection_create(mgr->instance, NULL);
if (udp_conn) {
etcp_conn_add_init_cbk(udp_conn, cm_invite_init_cb, inv);
sc_init_ctx(&udp_conn->crypto_ctx, &mgr->instance->my_keys);
sc_set_peer_public_key(&udp_conn->crypto_ctx, group_ni->public_key, 0);
}
}
if (udp_conn && etcp_link_new(udp_conn, s, &sa, 0)) {
inv->conn = udp_conn; udp_connected = 1;
}
}
s = s->next;
}
if (connected) break;
}
if (!connected) { cm_invite_fail(inv, CONN_MGR_ERR_UNREACHABLE); return CONN_MGR_ERR_UNREACHABLE; }
/* ── TCP: first address ── */
for (const struct TOPO_ADDR4* a = group_ni->v4_addrs; a; a = a->next) {
if (!(a->protocol & TOPO_PROTO_TCP)) 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; memset(&sa, 0, sizeof(sa)); memcpy(&sa, &sin, sizeof(sin));
struct stcp_link_config tcp_cfg = {
.ua = mgr->instance->ua, .my_keys = &mgr->instance->my_keys, .inst = mgr->instance,
.peer_pubkey = group_ni->public_key, .peer_pubkey_mode = 0,
.remote_addr = &sa, .remote_port = a->port,
};
inv->tcp_link = stcp_link_connect(&tcp_cfg);
if (inv->tcp_link) {
stcp_link_set_on_ready(inv->tcp_link, cm_tcp_ready_cb, inv);
stcp_link_set_on_close(inv->tcp_link, cm_tcp_close_cb, inv);
tcp_connected = 1;
}
break;
}
if (!udp_connected && !tcp_connected) {
cm_invite_fail(inv, CONN_MGR_ERR_UNREACHABLE); return CONN_MGR_ERR_UNREACHABLE;
}
inv->overall_timer = uasync_set_timeout(mgr->instance->ua, (int)inv->timeout_ms * 10, inv, cm_invite_overall_timeout, "conn_mgr_invite");
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: invite started node=0x%016llx group=0x%016llx timeout=%ums",
src/routing_layer/etcp_router.c
+5 -1
View File
@@ -1063,7 +1063,11 @@ void etcp_router_destroy(struct UTUN_INSTANCE* inst) {
int etcp_router_bind(struct UTUN_INSTANCE* inst, uint8_t svc_id, etcp_recv_fn callback) {
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL instance"); return -1; }
if (!callback) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL callback for svc_id=%u", svc_id); return -1; }
if (!callback) {
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "unbind svc_id=%u", svc_id);
inst->router_bindings.callbacks[svc_id] = NULL;
return 0;
}
if (inst->router_bindings.callbacks[svc_id])
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "overwriting svc_id=%u", svc_id);
inst->router_bindings.callbacks[svc_id] = callback;
src/transport_layer/etcp_connect.c
+13 -17
View File
@@ -211,24 +211,20 @@ int etcp_connect(struct UTUN_INSTANCE* inst, struct TOPO_NODEQ* node,
if (!inst || !node || !cb) return -1;
uint64_t node_id = node->node->node_id;
// Check if already connected via indexed queue
// Check if already connected
{
struct ll_entry* e = queue_find_data_by_index(inst->connections, (const uint8_t*)&node_id);
if (e) {
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
struct ETCP_CONN* existing = ce->conn;
if (existing && existing->peer_node_id == node_id) {
struct ETCP_LINK* l = existing->links;
while (l) { if (l->link_state == 3) break; l = l->next; }
if (l) {
DEBUG_INFO(DEBUG_CATEGORY_ETCP_CONNECT, "[etcp_connect] node 0x%016llx already connected, delivering immediately",
(unsigned long long)node_id);
if (flags & ETCP_CONNECT_EARLY) cb(arg, existing, ETCP_CONNECT_EARLY);
if (flags & ETCP_CONNECT_LATE) cb(arg, existing, ETCP_CONNECT_LATE);
if ((flags & ETCP_CONNECT_BGP_READY) && existing->routing_exchange_active >= 3)
cb(arg, existing, ETCP_CONNECT_BGP_READY);
return 0;
}
struct ETCP_CONN* existing = instance_find_conn(inst, node_id);
if (existing && existing->peer_node_id == node_id) {
struct ETCP_LINK* l = existing->links;
while (l) { if (l->link_state == 3) break; l = l->next; }
if (l) {
DEBUG_INFO(DEBUG_CATEGORY_ETCP_CONNECT, "[etcp_connect] node 0x%016llx already connected, delivering immediately",
(unsigned long long)node_id);
if (flags & ETCP_CONNECT_EARLY) cb(arg, existing, ETCP_CONNECT_EARLY);
if (flags & ETCP_CONNECT_LATE) cb(arg, existing, ETCP_CONNECT_LATE);
if ((flags & ETCP_CONNECT_BGP_READY) && existing->routing_exchange_active >= 3)
cb(arg, existing, ETCP_CONNECT_BGP_READY);
return 0;
}
}
}
src/transport_layer/etcp_connections.c
+50
View File
@@ -733,6 +733,55 @@ void etcp_socket_remove(struct ETCP_SOCKET* conn) {
u_free(conn);
}
/* ── TCP socket management ── */
struct TCP_SOCKET* tcp_socket_add(struct UTUN_INSTANCE* instance, struct CFG_SERVER* server) {
if (!instance || !server) return NULL;
struct TCP_SOCKET* ts = u_calloc(1, sizeof(struct TCP_SOCKET));
if (!ts) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_socket_add: alloc failed"); return NULL; }
ts->instance = instance;
ts->type = server->type;
if (server->name && server->name[0]) {
strncpy(ts->name, server->name, MAX_CONN_NAME_LEN - 1);
ts->name[MAX_CONN_NAME_LEN - 1] = '\0';
}
ts->sock_id = instance->next_socket_id++;
if (server->ip.ss_family == AF_INET) {
struct sockaddr_in* sin = (struct sockaddr_in*)&server->ip;
memcpy(&ts->local_addr, sin, sizeof(*sin));
ts->interface_addr = ts->local_addr;
if (server->type == CFG_SERVER_TYPE_PUBLIC && sin->sin_addr.s_addr == 0) {
struct sockaddr_storage remote; memset(&remote, 0, sizeof(remote));
struct sockaddr_in* rs = (struct sockaddr_in*)&remote;
rs->sin_family = AF_INET; rs->sin_port = htons(53);
inet_pton(AF_INET, "8.8.8.8", &rs->sin_addr);
struct sockaddr_storage local;
if (get_outgoing_local_ip(&server->ip, &remote, &local) == 0)
ts->interface_addr = local;
}
} else if (server->ip.ss_family == AF_INET6) {
struct sockaddr_in6* sin6 = (struct sockaddr_in6*)&server->ip;
memcpy(&ts->local_addr, sin6, sizeof(*sin6));
ts->interface_addr = ts->local_addr;
}
ts->next = instance->tcp_sockets;
instance->tcp_sockets = ts;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "tcp_socket_add: %s type=%d sock_id=%u", ts->name, ts->type, ts->sock_id);
return ts;
}
void tcp_socket_remove(struct TCP_SOCKET* sock) {
if (!sock || !sock->instance) return;
struct UTUN_INSTANCE* inst = sock->instance;
struct TCP_SOCKET** pp = &inst->tcp_sockets;
while (*pp && *pp != sock) pp = &(*pp)->next;
if (*pp) *pp = sock->next;
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "tcp_socket_remove: %s sock_id=%u", sock->name, sock->sock_id);
u_free(sock);
}
struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn, struct sockaddr_storage* remote_addr, uint8_t is_server) {
DEBUG_TRACE(DEBUG_CATEGORY_CONNECTION, "");
@@ -2061,6 +2110,7 @@ int init_sockets(struct UTUN_INSTANCE* instance) {
fail_count++;
} else {
instance->stcp_server = tsrv;
tcp_socket_add(instance, server);
success_count++;
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "TCP server %s on port %u", server->name, port);
}
src/transport_layer/etcp_connections.h
+15
View File
@@ -317,6 +317,21 @@ struct ETCP_SOCKET* etcp_socket_add(struct UTUN_INSTANCE* instance, struct CFG_S
// удаляет сокет и освобождает ресурсы (грохает все его подключения и сокет)
void etcp_socket_remove(struct ETCP_SOCKET* conn);
// TCP socket descriptor (analogous to ETCP_SOCKET for UDP)
struct TCP_SOCKET {
struct TCP_SOCKET* next;
struct UTUN_INSTANCE* instance;
char name[MAX_CONN_NAME_LEN];
uint8_t sock_id;
uint8_t type; // CFG_SERVER_TYPE_*
struct sockaddr_storage local_addr;
struct sockaddr_storage interface_addr;
struct sockaddr_storage nat_addr;
};
struct TCP_SOCKET* tcp_socket_add(struct UTUN_INSTANCE* instance, struct CFG_SERVER* server);
void tcp_socket_remove(struct TCP_SOCKET* sock);
// connection functions
// создает новый канал связи для etcp подключения (ETCP_CONN)
struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn, struct sockaddr_storage* remote_addr, uint8_t is_server);
src/utun_instance.c
+18
View File
@@ -129,6 +129,8 @@ static int instance_init_common(struct UTUN_INSTANCE* instance, struct UASYNC* u
// Initialize connection queue
instance->connections = queue_new(ua, 256, 0, 8, "connections");
if (!instance->connections) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "Failed to create connections queue"); return -1; }
instance->tcp_connections = queue_new(ua, sizeof(struct tcp_conn_entry), sizeof(uint64_t), 8, "tcpconn");
if (!instance->tcp_connections) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "Failed to create tcp_connections queue"); return -1; }
struct CFG_NETWORK *net = config->networks;
while (net) {
struct ll_entry *entry = queue_entry_new(sizeof(struct NETWORK_ENTRY));
@@ -440,6 +442,22 @@ void utun_instance_destroy(struct UTUN_INSTANCE *instance) {
queue_free(instance->connections); instance->connections = NULL;
DEBUG_INFO(DEBUG_CATEGORY_MEMORY, "[INSTANCE_DESTROY] ETCP connections cleanup complete");
// Cleanup TCP connections (stcp_links)
if (instance->tcp_connections) {
struct ll_entry* entry = instance->tcp_connections->head;
while (entry) {
struct ll_entry* next = entry->next;
struct tcp_conn_entry* te = (struct tcp_conn_entry*)entry->data;
if (te && te->link) stcp_link_close(te->link);
entry = next;
}
}
queue_free(instance->tcp_connections); instance->tcp_connections = NULL;
DEBUG_INFO(DEBUG_CATEGORY_MEMORY, "[INSTANCE_DESTROY] TCP connections cleanup complete");
// Cleanup TCP sockets
while (instance->tcp_sockets) tcp_socket_remove(instance->tcp_sockets);
// Wait for all deferred callbacks (etcp_connection_free_deferred)
// to finish before destroying pools
while (instance->ua && instance->ua->immediate_queue_head)
src/utun_instance.h
+24
View File
@@ -34,6 +34,7 @@ struct ROUTE_TABLE;
struct ETCP_CONN;
typedef void (*etcp_new_conn_fn)(struct ETCP_CONN* conn, void* arg);
struct ETCP_SOCKET;
struct TCP_SOCKET;
struct tun_if;
struct ETCP_BINDINGS;
struct ETCP_ROUTER_BINDINGS;
@@ -61,6 +62,13 @@ struct conn_queue_entry {
struct ETCP_CONN* conn;
};
// Queue entry for instance->tcp_connections (indexed by node_id)
struct tcp_conn_entry {
uint64_t node_id; // index key
struct stcp_link* link; // active TCP link
struct ETCP_CONN* etcp_conn; // = stcp_link_get_etcp_conn(link) — for etcp_send() compat
};
// uTun instance configuration
struct UTUN_INSTANCE {
// Identification
@@ -101,6 +109,7 @@ struct UTUN_INSTANCE {
// Connections (очередь всех подключений для instance)
struct ll_queue* connections; // indexed by peer_node_id (0=pending), data=conn_queue_entry
struct ll_queue* tcp_connections; // indexed by node_id, data=tcp_conn_entry [TCP-only]
// Callback chain for new ETCP connections
struct etcp_cbk_entry* new_conn_cbks;
@@ -113,6 +122,7 @@ struct UTUN_INSTANCE {
// Active sockets (UDP)
struct ETCP_SOCKET* etcp_sockets;// linked-list
struct TCP_SOCKET* tcp_sockets; // linked-list [TCP-only]
struct stcp_server *stcp_server; // TCP server (single for now)
// Pending one-shot pings (for callback on PONG or timeout)
@@ -177,6 +187,20 @@ void utun_instance_set_topo_group_enabled(int enabled);
// Diagnostic function for memory leak analysis
void utun_instance_diagnose_leaks(struct UTUN_INSTANCE* instance, const char* phase);
/* Find active connection by node_id — searches both UDP (connections) and TCP (tcp_connections) queues.
Returns ETCP_CONN* ready for etcp_send(), or NULL if not found. */
static inline struct ETCP_CONN* instance_find_conn(struct UTUN_INSTANCE* inst, uint64_t node_id) {
if (inst->connections) {
struct ll_entry* e = queue_find_data_by_index(inst->connections, (const uint8_t*)&node_id);
if (e) return ((struct conn_queue_entry*)e->data)->conn;
}
if (inst->tcp_connections) {
struct ll_entry* e = queue_find_data_by_index(inst->tcp_connections, (const uint8_t*)&node_id);
if (e) return ((struct tcp_conn_entry*)e->data)->etcp_conn;
}
return NULL;
}
#ifdef __cplusplus
}