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

Share JOIN state machine across ETCP and authenticated carriers

Browse Source
This commit is contained in:
evgeny committed 2026-10-30 23:33:57 +01:00
1 parent 8a0cc6f5e1
commit 6344ff2ecb
1 file changed
+182 -64
Download Patch File Download Diff File Expand all files Collapse all files
src/chat/chat_sync.c
+182 -64
View File
@@ -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;