diff --git a/src/chat/chat_sync.c b/src/chat/chat_sync.c index f7374bb11..a8073277c 100644 --- a/src/chat/chat_sync.c +++ b/src/chat/chat_sync.c @@ -65,6 +65,7 @@ struct join_pending { char ch_id[64]; uint64_t attempt; /* идентификатор конкретного JOIN_REQUEST */ uint64_t expires; /* deadline ожидания записи, timebase */ + struct chat_join_route route; /* Подтверждённый путь запроса, сохраняется до READY. */ struct join_pending* next; }; @@ -86,6 +87,7 @@ struct chat_sync { struct { uint64_t channel, peer, key, request, attempt; uint8_t inviter, phase; /* 0=idle, 1=connect, 2=info, 3=ready, 4=offer, 5=prepare */ + struct chat_join_route route; char password[INVITE_PASS_MAX]; struct NODE_CONN_DIRECT* handle; void* transport_timer; @@ -109,6 +111,7 @@ struct chat_sync { #define CS_JOIN_TIMEOUT_MS 8000 #define CS_INVITE_SEND_TIMEOUT_MS 3000 /* inviter waits for joiner to respond */ #define CS_SYNC_INTERVAL_MS 1000 +#define CS_CARRIER_TIMEOUT_MS 20000 /* GATT discovery и фрагментация JOIN при минимальном ATT MTU. */ static void cs_info_req_timeout_cb(void* arg); static void cs_join_finish(struct chat_sync* cs, int result); @@ -203,6 +206,7 @@ static void cs_join_end(struct chat_sync* cs, int result, int notify) { struct UTUN_INSTANCE* inst = cs->inst; uint64_t channel = cs->join.channel, peer = cs->join.peer, request = cs->join.request; struct NODE_CONN_DIRECT* handle = cs->join.handle; + uint64_t carrier_handle = cs->join.route.handle; struct chat_invite_build* build = cs->join.offer_build; uint64_t origin = cs->join.origin_operation; cs_cancel_proto_timers(cs); @@ -216,6 +220,8 @@ static void cs_join_end(struct chat_sync* cs, int result, int notify) { cs_server_join_reply(cs, name, peer, origin, result); } if (handle) node_conn_direct_close(handle); + if (carrier_handle && inst->join_carrier) + inst->join_carrier->close(inst->join_carrier->ctx, carrier_handle, result == 0); if (notify) chat_sync_join_result(inst, request, channel, peer, result); } @@ -232,6 +238,7 @@ static int cs_join_reserve(struct chat_sync* cs, uint64_t channel, uint64_t peer } cs->join.channel = channel; cs->join.peer = peer; cs->join.request = request; cs->join.attempt = attempt; cs->join.inviter = inviter; cs->join.phase = 5; + cs->join.route.peer = peer; return 0; } @@ -259,7 +266,7 @@ static int cs_group_id_to_ch_id(struct UTUN_INSTANCE* inst, uint64_t group_id, c } static int cs_send(struct chat_sync* cs, const char* ch_id, uint64_t dst, - const uint8_t* payload, size_t len) { + const uint8_t* payload, size_t len, const struct chat_join_route* route) { if (len < 1) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: cs_send len<1", CS_ID); return -1; } size_t total = 1 + 8 + len; uint8_t* buf = u_malloc(total); @@ -268,6 +275,17 @@ static int cs_send(struct chat_sync* cs, const char* ch_id, uint64_t dst, *p++ = ETCP_ID_CHAT_SYNC; cs_write_group_id(p, ch_id); p += 8; memcpy(p, payload, len); + if (route && route->handle) { + const struct chat_join_carrier* carrier = cs->inst->join_carrier; + int rc = carrier && route->peer == dst && total <= CHAT_JOIN_WIRE_MAX ? + carrier->send(carrier->ctx, route->handle, buf, total) : -1; + u_free(buf); + if (rc) DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN carrier send failed handle=%016llx type=%u bytes=%zu", + (unsigned long long)route->handle, payload[0], total); + else DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN carrier SEND %s peer=%016llx handle=%016llx bytes=%zu", + cs_msg_name(payload[0]), (unsigned long long)dst, (unsigned long long)route->handle, total); + return rc; + } struct ll_entry* entry = queue_entry_new(0); if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: cs_send queue_entry_new failed", CS_ID); u_free(buf); return -1; } entry->dgram = buf; entry->len = total; @@ -282,13 +300,13 @@ static int cs_send(struct chat_sync* cs, const char* ch_id, uint64_t dst, } static int cs_send_attempt(struct chat_sync* cs, const char* channel, uint64_t peer, - const uint8_t* payload, size_t len, uint64_t attempt) { + const uint8_t* payload, size_t len, uint64_t attempt, const struct chat_join_route* route) { if (!attempt || !len) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "missing join attempt"); return -1; } uint8_t* data = u_malloc(len + 8); if (!data) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join wire allocation failed"); return -1; } data[0] = payload[0]; uint64_t wire = htobe64(attempt); memcpy(data + 1, &wire, 8); memcpy(data + 9, payload + 1, len - 1); - int rc = cs_send(cs, channel, peer, data, len + 8); u_free(data); return rc; + int rc = cs_send(cs, channel, peer, data, len + 8, route); u_free(data); return rc; } /* ── Channel cache ── */ @@ -298,7 +316,7 @@ static int cs_server_join_reply(struct chat_sync* cs, const char* channel, uint6 uint32_t code = htonl((uint32_t)result); memcpy(data + 1, &code, sizeof(code)); DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "server join reply operation=%016llx channel=%s peer=%016llx result=%d", (unsigned long long)operation, channel, (unsigned long long)peer, result); - int rc = cs_send_attempt(cs, channel, peer, data, sizeof(data), operation); + int rc = cs_send_attempt(cs, channel, peer, data, sizeof(data), operation, NULL); if (rc) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "server join reply failed operation=%016llx rc=%d", (unsigned long long)operation, rc); return rc; } @@ -359,11 +377,11 @@ static void cs_handle_server_join_result(struct chat_sync* cs, uint64_t peer, co /* ── Forward declarations for new message handlers ── */ static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, - const char* ch_id, const uint8_t* pl, size_t len, uint64_t attempt); + const char* ch_id, const uint8_t* pl, size_t len, uint64_t attempt, const struct chat_join_route* route); static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, - const char* ch_id, const uint8_t* pl, size_t len); + const char* ch_id, const uint8_t* pl, size_t len, const struct chat_join_route* route); static void cs_handle_join_request(struct chat_sync* cs, uint64_t peer, - const char* ch_id, const uint8_t* pl, size_t len, uint64_t attempt); + const char* ch_id, const uint8_t* pl, size_t len, uint64_t attempt, const struct chat_join_route* route); static void cs_handle_join_ready(struct chat_sync* cs, uint64_t peer, const char* ch_id, const uint8_t* pl, size_t len); static void cs_handle_channel_invite(struct chat_sync* cs, uint64_t peer, @@ -380,26 +398,18 @@ static void cs_pending_free_all(struct chat_sync* cs); static void cs_pending_prune(struct chat_sync* cs); /* ── Recv dispatcher ── */ -static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { - if (!entry || !entry->dgram || !conn || entry->len < 10) { - if (entry) { - DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: recv short entry len=%zu conn=%s", CS_ID, entry->len, conn ? conn->log_name : "?"); - if (entry->dgram) u_free(entry->dgram); queue_entry_free(entry); - } - return; +/* Общий автомат JOIN получает уже подтверждённую личность и явный ответный маршрут. */ +static void cs_receive(struct chat_sync* cs, const struct chat_join_route* route, const uint8_t* d, size_t dlen) { + if (!cs || !cs->initialized) return; + if (!d || dlen < 10 || d[0] != ETCP_ID_CHAT_SYNC) { + DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "chat_sync: invalid frame bytes=%zu", dlen); return; } - struct chat_sync* cs = cs_of(conn->instance); - if (!cs || !cs->initialized) { u_free(entry->dgram); queue_entry_free(entry); return; } - - uint64_t peer = conn ? conn->peer_node_id : 0; - const uint8_t* d = entry->dgram; - size_t dlen = entry->len; - + uint64_t peer = route->peer; uint64_t group_id; memcpy(&group_id, d + 1, sizeof(group_id)); group_id = be64toh(group_id); if (!group_id || group_id > INT64_MAX) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "chat_sync: invalid wire channel=%llu", (unsigned long long)group_id); - u_free(entry->dgram); queue_entry_free(entry); return; + return; } uint8_t type = d[9]; char ch_id[64]; @@ -411,7 +421,7 @@ static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { type == CS_MSG_SERVER_JOIN_REQ || type == CS_MSG_SERVER_JOIN_RESULT) { snprintf(ch_id, sizeof(ch_id), "%llu", (unsigned long long)group_id); } else { - u_free(entry->dgram); queue_entry_free(entry); return; + return; } } DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: recv %s(%02x) from=%016llx ch=%s len=%zu", @@ -425,22 +435,24 @@ static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { uint64_t attempt = 0; if (type == CS_MSG_JOIN_INFO_REQ || type == CS_MSG_JOIN_INFO_RESP || type == CS_MSG_JOIN_REQUEST || type == CS_MSG_JOIN_READY || type == CS_MSG_ERROR || type == CS_MSG_SERVER_JOIN_REQ || type == CS_MSG_SERVER_JOIN_RESULT) { - if (plen < 8) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "join missing attempt type=%u", type); goto received; } + if (plen < 8) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "join missing attempt type=%u", type); return; } memcpy(&attempt, pl, 8); attempt = be64toh(attempt); pl += 8; plen -= 8; - if (!attempt) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "join zero attempt type=%u", type); goto received; } + if (!attempt) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "join zero attempt type=%u", type); return; } } if ((type == CS_MSG_JOIN_INFO_RESP || type == CS_MSG_JOIN_READY || type == CS_MSG_ERROR) && (cs->join.peer != peer || cs->join.channel != group_id || cs->join.attempt != attempt || + cs->join.route.handle != route->handle || + (route->handle && memcmp(cs->join.route.public_key, route->public_key, 32)) || (type == CS_MSG_JOIN_INFO_RESP && cs->join.phase != 2) || (type == CS_MSG_JOIN_READY && cs->join.phase != 3))) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "ignore stale join response peer=%016llx group=%016llx type=%u", (unsigned long long)peer, (unsigned long long)group_id, type); - u_free(entry->dgram); queue_entry_free(entry); return; + return; } switch (type) { - case CS_MSG_JOIN_INFO_REQ: DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle JOIN_INFO_REQ", CS_ID); cs_handle_join_info_req(cs, peer, ch_id, pl, plen, attempt); break; - case CS_MSG_JOIN_INFO_RESP: DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle JOIN_INFO_RESP", CS_ID); cs_handle_join_info_resp(cs, peer, ch_id, pl, plen); break; - case CS_MSG_JOIN_REQUEST: DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle JOIN_REQUEST", CS_ID); cs_handle_join_request(cs, peer, ch_id, pl, plen, attempt); break; + case CS_MSG_JOIN_INFO_REQ: cs_handle_join_info_req(cs, peer, ch_id, pl, plen, attempt, route); break; + case CS_MSG_JOIN_INFO_RESP: cs_handle_join_info_resp(cs, peer, ch_id, pl, plen, route); break; + case CS_MSG_JOIN_REQUEST: cs_handle_join_request(cs, peer, ch_id, pl, plen, attempt, route); break; case CS_MSG_JOIN_READY: DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle JOIN_READY", CS_ID); cs_handle_join_ready(cs, peer, ch_id, pl, plen); break; case CS_MSG_ERROR: { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: RECV ERROR from=%016llx ch=%s", CS_ID, (unsigned long long)peer, ch_id); @@ -456,8 +468,29 @@ static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { break; default: DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: UNKNOWN msg type=%02x from=%016llx", CS_ID, type, (unsigned long long)peer); break; } -received: - u_free(entry->dgram); queue_entry_free(entry); +} + +/* ETCP передаёт владение entry; обработчик JOIN заимствует только подтверждённые key и bytes. */ +static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { + if (conn && entry && entry->dgram) { + struct chat_join_route route = {.peer = conn->peer_node_id}; + memcpy(route.public_key, conn->crypto_ctx.peer_public_key, 32); + cs_receive(cs_of(conn->instance), &route, entry->dgram, entry->len); + } else DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "chat_sync: missing ETCP connection/frame"); + if (entry) { u_free(entry->dgram); queue_entry_free(entry); } +} + +/* Carrier не имеет доступа к остальным командам CHAT_SYNC даже при корректной криптографии. */ +void chat_sync_join_carrier_receive(struct UTUN_INSTANCE* inst, uint64_t handle, const uint8_t public_key[32], + const uint8_t* data, size_t len) { + if (!inst || !inst->join_carrier || !handle || !public_key || !data || len < 18 || len > CHAT_JOIN_WIRE_MAX || + (data[9] != CS_MSG_JOIN_INFO_REQ && data[9] != CS_MSG_JOIN_INFO_RESP && data[9] != CS_MSG_JOIN_REQUEST && + data[9] != CS_MSG_JOIN_READY && data[9] != CS_MSG_ERROR)) { + DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN carrier rejected frame handle=%016llx bytes=%zu", (unsigned long long)handle, len); + return; + } + struct chat_join_route route = {.peer = sc_derive_node_id_from_pubkey(public_key), .handle = handle}; + memcpy(route.public_key, public_key, 32); cs_receive(cs_of(inst), &route, data, len); } /* ── Connection callbacks ── */ @@ -656,7 +689,7 @@ static void cs_on_peer_sleep(struct UTUN_INSTANCE* inst, uint64_t peer_node_id, static void cs_send_sound_setting(struct chat_sync* cs, const char* ch_id, uint64_t dst, int on) { uint8_t p[2] = { CS_MSG_SOUND_SETTING, (uint8_t)(on ? 1 : 0) }; - cs_send(cs, ch_id, dst, p, sizeof(p)); + cs_send(cs, ch_id, dst, p, sizeof(p), NULL); } void chat_sync_broadcast_sound_setting(struct UTUN_INSTANCE* inst, const char* ch_id, int on) { @@ -742,17 +775,73 @@ static int cs_start_channel_join(struct chat_sync* cs, uint64_t peer) { size_t len = strlen(cs->join.password); uint8_t data[140]; data[0] = CS_MSG_JOIN_INFO_REQ; memcpy(data + 1, &cs->join.key, 8); data[9] = (uint8_t)len; memcpy(data + 10, cs->join.password, len); - cs->info_req_timer = uasync_set_timeout(cs->inst->ua, CS_INFO_REQ_TIMEOUT_MS * 10, cs, + unsigned timeout = cs->join.route.handle ? CS_CARRIER_TIMEOUT_MS : CS_INFO_REQ_TIMEOUT_MS; + cs->info_req_timer = uasync_set_timeout(cs->inst->ua, timeout * 10, cs, cs_info_req_timeout_cb, "cs_info_req"); if (!cs->info_req_timer) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join info deadline allocation failed"); return -1; } cs->join.phase = 2; - if (cs_send_attempt(cs, channel, peer, data, len + 10, cs->join.attempt)) return -1; + if (cs_send_attempt(cs, channel, peer, data, len + 10, cs->join.attempt, &cs->join.route)) return -1; if (cs->join.transport_timer) { uasync_cancel_timeout(cs->inst->ua, cs->join.transport_timer); cs->join.transport_timer = NULL; } DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "join info requested channel=%s attempt=%016llx", channel, (unsigned long long)cs->join.attempt); return 0; } +/* Событие относится к сохранённому handle, а не к любому соединению с тем же node_id. */ +void chat_sync_join_carrier_ready(struct UTUN_INSTANCE* inst, uint64_t handle, const uint8_t public_key[32]) { + struct chat_sync* cs = cs_of(inst); + if (!cs || !cs->initialized || !handle || cs->join.route.handle != handle || cs->join.phase != 1) { + DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "ignore inactive JOIN carrier ready handle=%016llx", (unsigned long long)handle); return; + } + if (!public_key || memcmp(public_key, cs->join.route.public_key, 32) || + sc_derive_node_id_from_pubkey(public_key) != cs->join.peer) { + DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN carrier identity differs from invitation"); cs_join_finish(cs, CHAT_CONNECT_AUTH); return; + } + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN carrier authenticated peer=%016llx handle=%016llx", + (unsigned long long)cs->join.peer, (unsigned long long)handle); + if (cs_start_channel_join(cs, cs->join.peer) < 0) cs_join_finish(cs, CHAT_CONNECT_ERROR); +} + +/* Закрытие затрагивает только попытки этого пути; сохранённое членство не откатывается. */ +void chat_sync_join_carrier_closed(struct UTUN_INSTANCE* inst, uint64_t handle) { + struct chat_sync* cs = cs_of(inst); + if (!cs || !handle) return; + if (cs->join.route.handle == handle) { + DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN carrier closed handle=%016llx phase=%u", (unsigned long long)handle, cs->join.phase); + cs_join_finish(cs, CHAT_CONNECT_ERROR); + } + struct join_pending** p = &cs->join_pending; + while (*p) { + struct join_pending* item = *p; + if (item->route.handle != handle) { p = &item->next; continue; } + DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN waiter cancelled with carrier handle=%016llx attempt=%016llx", + (unsigned long long)handle, (unsigned long long)item->attempt); + *p = item->next; u_free(item); + } +} + +/* Provider переживает restart чата; при его снятии живые пользователи отменяются до смены указателя. */ +int chat_sync_set_join_carrier(struct UTUN_INSTANCE* inst, const struct chat_join_carrier* carrier) { + if (!inst || (carrier && (!carrier->ctx || !carrier->open || !carrier->send || !carrier->close))) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "invalid JOIN carrier registration"); return -1; + } + if (inst->join_carrier == carrier) return 0; + struct chat_sync* cs = cs_of(inst); + if (cs) { + if (cs->join.route.handle) cs_join_finish(cs, CHAT_CONNECT_CANCELLED); + struct join_pending** p = &cs->join_pending; + while (*p) { + struct join_pending* item = *p; + if (!item->route.handle) { p = &item->next; continue; } + if (inst->join_carrier) inst->join_carrier->close(inst->join_carrier->ctx, item->route.handle, 0); + *p = item->next; u_free(item); + } + } + inst->join_carrier = carrier; + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN carrier registered=%d", carrier != NULL); + return 0; +} + static int cs_send_server_offer(struct chat_sync* cs) { char channel[64]; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)cs->join.channel); size_t len = strlen(cs->join.offer_link); @@ -760,7 +849,7 @@ static int cs_send_server_offer(struct chat_sync* cs) { memcpy(data + 1, cs->join.add_key, INVITE_ADD_KEY_SIZE); data[33] = (uint8_t)(len >> 8); data[34] = (uint8_t)len; memcpy(data + 35, cs->join.offer_link, len); cs->join.phase = 4; - int rc = cs_send_attempt(cs, channel, cs->join.peer, data, len + 35, cs->join.attempt); + int rc = cs_send_attempt(cs, channel, cs->join.peer, data, len + 35, cs->join.attempt, NULL); OPENSSL_cleanse(data, sizeof(data)); if (rc) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "server join request send failed channel=%s rc=%d", channel, rc); return rc; @@ -824,7 +913,7 @@ static void cs_on_conn_up(struct ETCP_CONN* conn, int event, void* arg) { DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: conn_up peer=%016llx pending_invite=%llu links_up=%d", CS_ID, (unsigned long long)peer, (unsigned long long)cs->join.channel, conn->links_up); cs_schedule_peer_sync(cs, peer, event == ETCP_CBK_EVENT_REINIT ? "reinit" : "transport_up"); - if (cs->join.channel != 0 && cs->join.peer == peer && cs->join.phase == 1) { + if (cs->join.channel != 0 && cs->join.peer == peer && !cs->join.route.handle && cs->join.phase == 1) { if (cs->join.transport_timer) { uasync_cancel_timeout(cs->inst->ua, cs->join.transport_timer); cs->join.transport_timer = NULL; } @@ -847,7 +936,7 @@ static void cs_on_conn_down(struct ETCP_CONN* conn, int event, void* arg) { (voi cs_on_peer_status_changed(cs, peer, 0); - if (cs->join.peer == peer && cs->join.phase != 5) { + if (cs->join.peer == peer && !cs->join.route.handle && cs->join.phase != 5) { cs_cancel_proto_timers(cs); cs->join.phase = 1; if (cs->join.transport_timer) uasync_cancel_timeout(cs->inst->ua, cs->join.transport_timer); @@ -1159,6 +1248,21 @@ static int cs_connect_invite(struct UTUN_INSTANCE* inst, uint64_t request, const } } cs->join.key = invite.join_key; memcpy(cs->join.password, invite.password, sizeof(cs->join.password)); + memcpy(cs->join.route.public_key, invite.pubkey, 32); + if (!offer_channel && inst->join_carrier && !cs_find_conn_for_node(inst, invite.nodeId)) { + uint64_t handle = 0; + int selected = inst->join_carrier->open(inst->join_carrier->ctx, invite.pubkey, &handle); + if (selected < 0 || (!selected && !handle)) goto failure; + if (!selected) { + cs->join.route.handle = handle; cs->join.phase = 1; + cs->join.transport_timer = uasync_set_timeout(inst->ua, CS_CARRIER_TIMEOUT_MS * 10, cs, + cs_join_connect_timeout, "join_carrier_open"); + if (!cs->join.transport_timer) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN carrier deadline allocation failed"); goto failure; } + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN carrier selected peer=%016llx handle=%016llx attempt=%016llx", + (unsigned long long)invite.nodeId, (unsigned long long)handle, (unsigned long long)cs->join.attempt); + cs_node_clear(&node); return 0; + } + } rc = cs_join_hold_transport(cs, invite.nodeId, &node); if (rc < 0) goto failure; struct ETCP_CONN* conn = node_conn_direct_get_conn(cs->join.handle); @@ -1204,7 +1308,8 @@ static int cs_join_existing(struct UTUN_INSTANCE* inst, const char* channel, uin /* ─── JOIN_INFO_REQ (0x10): joiner → connection-узел ─── */ static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, - const char* ch_id, const uint8_t* pl, size_t len, uint64_t attempt) { + const char* ch_id, const uint8_t* pl, size_t len, uint64_t attempt, + const struct chat_join_route* route) { /* payload: [join_key:8][pass_len:1][password] */ if (len < 9 || pl[8] >= INVITE_PASS_MAX || len != 9 + (size_t)pl[8] || memchr(pl + 9, 0, pl[8])) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid JOIN_INFO_REQ size=%zu", CS_ID, len); return; @@ -1219,7 +1324,7 @@ static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, if (inviter == 0) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_REQ invalid/expired join key ch=%s from=%016llx", CS_ID, ch_id, (unsigned long long)peer); - uint8_t err[1] = { CS_MSG_ERROR }; cs_send_attempt(cs, ch_id, peer, err, 1, attempt); + uint8_t err[1] = { CS_MSG_ERROR }; cs_send_attempt(cs, ch_id, peer, err, 1, attempt, route); return; } @@ -1228,7 +1333,7 @@ static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, if (topo_node_sqlite_channel_get(cs->inst->topo_sqlite_db, ch_id, name, (int)sizeof(name), &owner, ch_x25519, ch_ed, ch_sig) != 0) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_REQ unknown ch=%s from=%016llx", CS_ID, ch_id, (unsigned long long)peer); - uint8_t err[1] = { CS_MSG_ERROR }; cs_send_attempt(cs, ch_id, peer, err, 1, attempt); + uint8_t err[1] = { CS_MSG_ERROR }; cs_send_attempt(cs, ch_id, peer, err, 1, attempt, route); return; } @@ -1239,7 +1344,7 @@ static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, sqlite3* vdb = cs->inst->topo_sqlite_db; if (topo_node_sqlite_member_get_join(vdb, ch_id, myid, my_join_sig, &my_join_ts) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN_INFO_REQ: own JOIN unavailable ch=%s", ch_id); - uint8_t error = CS_MSG_ERROR; cs_send_attempt(cs, ch_id, peer, &error, 1, attempt); return; + uint8_t error = CS_MSG_ERROR; cs_send_attempt(cs, ch_id, peer, &error, 1, attempt, route); return; } } uint8_t my_update_sig[64] = {0}; uint64_t my_update_ts; @@ -1247,13 +1352,13 @@ static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, if (chat_core_get_my_profile(cs->inst, ch_id, juser, sizeof(juser), my_update_sig, &my_update_ts) < 0) { chat_core_update_my_member(cs->inst); if (chat_core_get_my_profile(cs->inst, ch_id, juser, sizeof(juser), my_update_sig, &my_update_ts) < 0) { - uint8_t err[1] = { CS_MSG_ERROR }; cs_send_attempt(cs, ch_id, peer, err, sizeof(err), attempt); return; + uint8_t err[1] = { CS_MSG_ERROR }; cs_send_attempt(cs, ch_id, peer, err, sizeof(err), attempt, route); return; } } uint64_t my_signed_by = 0; uint8_t my_signature[64] = {0}; if (topo_node_sqlite_member_get_sign(cs->inst->topo_sqlite_db, ch_id, myid, &my_signed_by, my_signature)) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN_INFO_REQ: own invitation unavailable ch=%s", ch_id); - uint8_t error = CS_MSG_ERROR; cs_send_attempt(cs, ch_id, peer, &error, 1, attempt); return; + uint8_t error = CS_MSG_ERROR; cs_send_attempt(cs, ch_id, peer, &error, 1, attempt, route); return; } /* 5. JOIN_INFO_RESP: инфо канала + мой мембер-рекорд */ @@ -1276,13 +1381,13 @@ static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, buf[boff++] = (uint8_t)(il >> 8); buf[boff++] = (uint8_t)il; memcpy(buf + boff, juser, il); boff += il; memcpy(buf + boff, &my_signed_by, 8); boff += 8; memcpy(buf + boff, my_signature, 64); boff += 64; - cs_send_attempt(cs, ch_id, peer, buf, boff, attempt); + cs_send_attempt(cs, ch_id, peer, buf, boff, attempt, route); } /* ─── JOIN_INFO_RESP (0x11): connection-узел → joiner ─── */ static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, - const char* ch_id, const uint8_t* pl, size_t len) { + const char* ch_id, const uint8_t* pl, size_t len, const struct chat_join_route* route) { DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_RESP ch=%s from=%016llx len=%zu", CS_ID, ch_id, (unsigned long long)peer, len); if (len < 1) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_RESP too short len=%zu", CS_ID, len); return; } const uint8_t* p = pl; @@ -1343,8 +1448,7 @@ static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, .join_sig = c_join_sig, .join_ts = c_join_ts, .update_sig = c_update_sig, .update_ts = c_update_ts, .userinfo = c_userinfo, .signed_by = c_signed_by, .signature = c_signature }; - struct ETCP_CONN* response_conn = cs_find_conn_for_node(cs->inst, peer); - if (!response_conn || c_node_id != peer || memcmp(c_x25519, response_conn->crypto_ctx.peer_public_key, 32)) { + if (c_node_id != peer || memcmp(c_x25519, route->public_key, 32)) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "join response connection identity mismatch"); cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } if (!sqlite3_get_autocommit(db) || sqlite3_exec(db, "SAVEPOINT join_channel", NULL, NULL, NULL) != SQLITE_OK) { @@ -1410,10 +1514,11 @@ static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, jbuf[joff++] = CS_MSG_JOIN_REQUEST; memcpy(jbuf + joff, &cs->join.key, 8); joff += 8; memcpy(jbuf + joff, jm_buf, jm_len); joff += jm_len; - cs->join_timer = uasync_set_timeout(cs->inst->ua, CS_JOIN_TIMEOUT_MS * 10, cs, cs_join_timeout_cb, "cs_join"); + unsigned timeout = route->handle ? CS_CARRIER_TIMEOUT_MS : CS_JOIN_TIMEOUT_MS; + cs->join_timer = uasync_set_timeout(cs->inst->ua, timeout * 10, cs, cs_join_timeout_cb, "cs_join"); if (!cs->join_timer) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join ready deadline allocation failed"); cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } cs->join.phase = 3; - int jrc = cs_send_attempt(cs, ch_id, peer, jbuf, joff, cs->join.attempt); + int jrc = cs_send_attempt(cs, ch_id, peer, jbuf, joff, cs->join.attempt, &cs->join.route); DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_REQUEST sent myid=%016llx ch=%s to=0x%016llx rc=%d", CS_ID, (unsigned long long)myid, ch_id, (unsigned long long)peer, jrc); if (jrc) cs_join_finish(cs, CHAT_CONNECT_ERROR); @@ -1421,15 +1526,18 @@ static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, /* ─── pending-джойнеры (connection-узел): node_id → прямое соединение ─── */ -static int cs_pending_add(struct chat_sync* cs, uint64_t node_id, const char* ch_id, struct ETCP_CONN* conn, uint64_t attempt) { - (void)conn; +static int cs_pending_add(struct chat_sync* cs, uint64_t node_id, const char* ch_id, + const struct chat_join_route* route, uint64_t attempt) { unsigned count = 0; uint64_t now = get_time_tb(); struct join_pending** link = &cs->join_pending; while (*link) { struct join_pending* item = *link; if (item->expires <= now) { *link = item->next; u_free(item); continue; } if (item->node_id == node_id) { - if (item->attempt == attempt && !strcmp(item->ch_id, ch_id)) return 0; + if (item->attempt == attempt && !strcmp(item->ch_id, ch_id)) { + if (item->route.handle == route->handle && !memcmp(item->route.public_key, route->public_key, 32)) return 0; + DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN attempt changed authenticated route"); return -1; + } count++; } link = &item->next; @@ -1437,14 +1545,20 @@ static int cs_pending_add(struct chat_sync* cs, uint64_t node_id, const char* ch if (count >= 8) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "too many pending join attempts for peer"); return -1; } struct join_pending* jp = u_calloc(1, sizeof(*jp)); if (!jp) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join waiter allocation failed"); return -1; } - jp->attempt = attempt; jp->node_id = node_id; jp->expires = now + CS_JOIN_TIMEOUT_MS * 10; + jp->route = *route; jp->attempt = attempt; jp->node_id = node_id; + jp->expires = now + (route->handle ? CS_CARRIER_TIMEOUT_MS : CS_JOIN_TIMEOUT_MS) * 10; snprintf(jp->ch_id, sizeof(jp->ch_id), "%s", ch_id); jp->next = cs->join_pending; cs->join_pending = jp; return 0; } static void cs_pending_free_all(struct chat_sync* cs) { struct join_pending* jp = cs->join_pending; - while (jp) { struct join_pending* n = jp->next; u_free(jp); jp = n; } + while (jp) { + struct join_pending* n = jp->next; + if (jp->route.handle && cs->inst->join_carrier) + cs->inst->join_carrier->close(cs->inst->join_carrier->ctx, jp->route.handle, 0); + u_free(jp); jp = n; + } cs->join_pending = NULL; } @@ -1456,6 +1570,8 @@ static void cs_pending_prune(struct chat_sync* cs) { if (item->expires > now) { link = &item->next; continue; } DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "join waiter expired peer=%016llx attempt=%016llx", (unsigned long long)item->node_id, (unsigned long long)item->attempt); + if (item->route.handle && cs->inst->join_carrier) + cs->inst->join_carrier->close(cs->inst->join_carrier->ctx, item->route.handle, 0); *link = item->next; u_free(item); } } @@ -1469,7 +1585,9 @@ static void cs_send_join_ready(struct chat_sync* cs, const char* ch_id, uint64_t while (*link) { struct join_pending* pending = *link; if (pending->node_id != node_id || strcmp(pending->ch_id, ch_id)) { link = &pending->next; continue; } - if (cs_send_attempt(cs, ch_id, node_id, data, sizeof(data), pending->attempt)) { link = &pending->next; continue; } + if (cs_send_attempt(cs, ch_id, node_id, data, sizeof(data), pending->attempt, &pending->route)) { link = &pending->next; continue; } + if (pending->route.handle && cs->inst->join_carrier) + cs->inst->join_carrier->close(cs->inst->join_carrier->ctx, pending->route.handle, 1); DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "join ready peer=%016llx attempt=%016llx", (unsigned long long)node_id, (unsigned long long)pending->attempt); *link = pending->next; u_free(pending); @@ -1488,7 +1606,8 @@ static void cs_on_member_applied(const char* ch_id, uint64_t node_id, const uint /* ─── JOIN_REQUEST (0x12): joiner → connection-узел ─── */ static void cs_handle_join_request(struct chat_sync* cs, uint64_t peer, - const char* ch_id, const uint8_t* pl, size_t len, uint64_t attempt) { + const char* ch_id, const uint8_t* pl, size_t len, uint64_t attempt, + const struct chat_join_route* route) { /* payload: [join_key:8][member...] */ if (len < 8 + 1) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_REQUEST too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; } uint64_t join_key; memcpy(&join_key, pl, 8); @@ -1503,9 +1622,8 @@ static void cs_handle_join_request(struct chat_sync* cs, uint64_t peer, DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_REQUEST ch=%s joiner=%016llx from=%016llx", CS_ID, ch_id, (unsigned long long)m.node_id, (unsigned long long)peer); - /* привязка идентичности: pubkey из ETCP-соединения == заявленный x25519 */ - struct ETCP_CONN* jconn = cs_find_conn_for_node(cs->inst, peer); - if (!jconn || m.node_id != peer || memcmp(jconn->crypto_ctx.peer_public_key, m.x25519, 32) != 0) { + /* Оба транспорта подтверждают владение X25519 до входа в общий обработчик. */ + if (m.node_id != peer || memcmp(route->public_key, m.x25519, 32) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_REQUEST identity mismatch joiner=0x%016llx peer=0x%016llx — drop", CS_ID, (unsigned long long)m.node_id, (unsigned long long)peer); return; @@ -1516,7 +1634,7 @@ static void cs_handle_join_request(struct chat_sync* cs, uint64_t peer, if (inviter == 0) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_REQUEST invalid/expired join key ch=%s joiner=0x%016llx", CS_ID, ch_id, (unsigned long long)m.node_id); - uint8_t err[1] = { CS_MSG_ERROR }; cs_send_attempt(cs, ch_id, peer, err, 1, attempt); + uint8_t err[1] = { CS_MSG_ERROR }; cs_send_attempt(cs, ch_id, peer, err, 1, attempt, route); return; } @@ -1524,7 +1642,7 @@ static void cs_handle_join_request(struct chat_sync* cs, uint64_t peer, локально либо через Merkle; существующий транспорт обслуживает JOIN_REQUEST. */ /* добавить в pending (ждут JOIN_READY по приходу рекорда через merkle) */ - if (cs_pending_add(cs, m.node_id, ch_id, jconn, attempt) < 0) return; + if (cs_pending_add(cs, m.node_id, ch_id, route, attempt) < 0) return; /* форвард инвайтеру (или локальная обработка, если инвайтер == мы) */ if (inviter == cs->inst->node_id) { @@ -1538,11 +1656,11 @@ static void cs_handle_join_request(struct chat_sync* cs, uint64_t peer, cs->join.peer == m.node_id && cs->join.key == join_key) cs_join_finish(cs, 0); } else { - uint8_t error = CS_MSG_ERROR; cs_send_attempt(cs, ch_id, peer, &error, 1, attempt); + uint8_t error = CS_MSG_ERROR; cs_send_attempt(cs, ch_id, peer, &error, 1, attempt, route); } } else { if (chat_join_forward_request(cs->inst, channel_id, inviter, join_key, &m) < 0) { - uint8_t error = CS_MSG_ERROR; cs_send_attempt(cs, ch_id, peer, &error, 1, attempt); + uint8_t error = CS_MSG_ERROR; cs_send_attempt(cs, ch_id, peer, &error, 1, attempt, route); } else if (topo_node_sqlite_member_allowed(cs->inst->topo_sqlite_db, ch_id, m.node_id)) cs_send_join_ready(cs, ch_id, m.node_id); } @@ -1733,7 +1851,7 @@ static int cs_send_channel_invite(struct chat_sync* cs, struct UTUN_INSTANCE* in size_t il = strlen(juser); buf[boff++] = (uint8_t)(il >> 8); buf[boff++] = (uint8_t)il; memcpy(buf + boff, juser, il); boff += il; memcpy(buf + boff, &join_key, 8); boff += 8; - int rc = cs_send(cs, ch_id, target_node_id, buf, boff); + int rc = cs_send(cs, ch_id, target_node_id, buf, boff, NULL); DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: sent CHANNEL_INVITE to 0x%016llx ch=%s rc=%d", CS_ID, (unsigned long long)target_node_id, ch_id, rc); return rc;