chat_join: модуль join-протокола, invite-ссылки v2, синхронизация мемберов

This commit is contained in:
evgeny committed 2026-09-14 19:57:27 +03:00
1 parent a01a4ddba4
commit 75e39d2fee
41 files changed
+2206 -471

No files matched your search

+2
View File
@@ -86,6 +86,7 @@ utun_CORE_SOURCES = \
chat/chat_status.c \
chat/chat_sync.c \
chat/chat_admin.c \
chat/chat_join.c \
chat/member_sync.c \
chat/merkle_sync.c \
chat/invite_link.c \
@@ -176,6 +177,7 @@ libutun_a_SOURCES = \
chat/chat_status.c \
chat/chat_sync.c \
chat/chat_admin.c \
chat/chat_join.c \
chat/member_sync.c \
chat/merkle_sync.c \
chat/invite_link.c \
+4 -1
View File
@@ -124,9 +124,12 @@ void chat_core_create_channel(struct chat_channel_create* req) {
char juser3[256]; snprintf(juser3, sizeof(juser3), "{\"name\":\"%s\"}",
g_cc.inst->name[0] ? g_cc.inst->name : "");
/* корень дерева приглашений: подписан ключом группы (канальный ed25519) */
uint8_t root_sig[64]; memset(root_sig, 0, 64);
member_sync_sign_pubkey(req->ed25519_privkey, g_cc.inst->my_keys.public_key, root_sig);
int mrc = member_sync_put(g_cc.inst, req->channel_id, myid,
g_cc.inst->my_keys.public_key, g_cc.inst->my_ed25519_pubkey,
join_sig, join_ts, NULL, 0, juser3, NULL, NULL, 0);
join_sig, join_ts, NULL, 0, juser3, NULL, NULL, 0, 0, root_sig);
if (mrc < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: member_sync_put(self) FAILED ch=%s rc=%d",
CC_ID, req->channel_id, mrc);
+1 -1
View File
@@ -655,7 +655,7 @@ int chat_member_tags_commit(struct chat_member_tags* t) {
if (stg_val && strcmp(stg_val, "yes") == 0) stg = 1;
int rc = member_sync_put(g_cc.inst, t->ch_id, t->node_id, x25, ed, jsig, jts, usig, uts,
uinfo ? uinfo : "", json, sig, stg);
uinfo ? uinfo : "", json, sig, stg, 0, NULL);
sqlite3_finalize(st);
if (rc < 0) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: tags_commit — member_sync_put failed rc=%d", CC_ID, rc); return -1; }
if (rc & MS_APPLY_CHANGED) member_sync_broadcast_one(g_cc.inst, t->ch_id, t->node_id);
+1 -1
View File
@@ -464,7 +464,7 @@ static void hc_handle_connect(struct headless_client* cli, int id, const char* j
chat_core_set_my_node_id(g_hc.inst->node_id);
chat_sync_connect_from_invite(d.channelId, d.nodeId, d.pubkey,
addrs_buf, d.addrCount, addrs_len, d.password_len ? d.password : NULL);
addrs_buf, d.addrCount, addrs_len, d.password_len ? d.password : NULL, d.join_key);
char resp[512]; snprintf(resp, sizeof(resp),
"{\"channel_id\":%llu,\"node_id\":\"0x%016llx\",\"addrs\":%d,\"connecting\":true}",
+320
View File
@@ -0,0 +1,320 @@
/*
* chat_join.c — авторизация входа в канал через connection-узел и инвайтера.
* Маршрутизируемая (etcp_router) часть: регистрация ключа, форвард запроса,
* обработка на инвайтере (верификация + добавление мембера + подпись дерева).
* См. описание flow в chat_join.h.
*/
#include "chat_join.h"
#include "member_sync.h"
#include "../utun_instance.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/etcp_router.h"
#include "../routing_layer/route_crypto.h"
#include "../transport_layer/etcp_api.h"
#include "../transport_layer/etcp.h"
#include "../transport_layer/secure_channel.h"
#include "../ntp_time.h"
#include "../../lib/mem.h"
#include "../../lib/ll_queue.h"
#include "../../lib/u_async.h"
#include "../../lib/debug_config.h"
#include "../../lib/platform_compat.h"
#include <string.h>
#include <stdlib.h>
#define CJ_ID "chat_join"
/* ── Хранилище ключей на connection-узле ── */
struct join_key_entry {
struct join_key_entry* next;
uint64_t join_key;
uint64_t channel_id;
uint64_t inviter_id; /* кто пригласил (src KEY_REGISTER) */
uint64_t expiry_tb; /* timebase (0.1ms) истечения */
};
static struct join_key_entry* g_keys = NULL;
static uint64_t _now_tb(void) { return get_time_tb(); }
static void _keys_purge_expired(void) {
uint64_t now = _now_tb();
struct join_key_entry** p = &g_keys;
while (*p) {
if ((*p)->expiry_tb <= now) {
struct join_key_entry* r = *p;
*p = r->next;
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "%s: join-key expired key=%016llx ch=%llu inviter=%016llx",
CJ_ID, (unsigned long long)r->join_key, (unsigned long long)r->channel_id,
(unsigned long long)r->inviter_id);
u_free(r);
} else {
p = &(*p)->next;
}
}
}
static void _keys_put(uint64_t channel_id, uint64_t join_key, uint64_t inviter_id) {
_keys_purge_expired();
struct join_key_entry* e = u_malloc(sizeof(*e));
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: join-key alloc failed", CJ_ID); return; }
e->join_key = join_key; e->channel_id = channel_id; e->inviter_id = inviter_id;
e->expiry_tb = _now_tb() + (uint64_t)JOIN_KEY_TTL_SECONDS * 10000ULL;
e->next = g_keys;
g_keys = e;
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: join-key stored key=%016llx ch=%llu inviter=%016llx ttl=%ds",
CJ_ID, (unsigned long long)join_key, (unsigned long long)channel_id,
(unsigned long long)inviter_id, JOIN_KEY_TTL_SECONDS);
}
static struct join_key_entry* _keys_find(uint64_t channel_id, uint64_t join_key) {
_keys_purge_expired();
for (struct join_key_entry* e = g_keys; e; e = e->next)
if (e->channel_id == channel_id && e->join_key == join_key) return e;
return NULL;
}
static void _keys_free_all(void) {
struct join_key_entry* e = g_keys;
while (e) { struct join_key_entry* n = e->next; u_free(e); e = n; }
g_keys = NULL;
}
/* ── Сериализация member-блока ── */
int chat_join_serialize_member(const struct join_member_data* m,
uint8_t* out, size_t out_sz, size_t* out_len) {
if (!m || !out || !out_len || !m->x25519 || !m->ed25519) return -1;
size_t ul = strnlen(m->userinfo, 255);
size_t need = 8 + 32 + 32 + 64 + 8 + 64 + 8 + 1 + ul;
if (need > out_sz) return -1;
size_t off = 0;
memcpy(out + off, &m->node_id, 8); off += 8;
memcpy(out + off, m->x25519, 32); off += 32;
memcpy(out + off, m->ed25519, 32); off += 32;
if (m->join_sig) memcpy(out + off, m->join_sig, 64); else memset(out + off, 0, 64); off += 64;
memcpy(out + off, &m->join_ts, 8); off += 8;
if (m->update_sig) memcpy(out + off, m->update_sig, 64); else memset(out + off, 0, 64); off += 64;
memcpy(out + off, &m->update_ts, 8); off += 8;
out[off++] = (uint8_t)ul;
if (ul) memcpy(out + off, m->userinfo, ul);
off += ul;
*out_len = off;
return 0;
}
int chat_join_parse_member(const uint8_t* data, size_t len, struct join_member_data* m) {
if (!data || !m || len < 8 + 32 + 32 + 64 + 8 + 64 + 8 + 1) return -1;
memset(m, 0, sizeof(*m));
size_t off = 0;
memcpy(&m->node_id, data + off, 8); off += 8;
m->x25519 = data + off; off += 32;
m->ed25519 = data + off; off += 32;
m->join_sig = data + off; off += 64;
memcpy(&m->join_ts, data + off, 8); off += 8;
m->update_sig = data + off; off += 64;
memcpy(&m->update_ts, data + off, 8); off += 8;
uint8_t ul = data[off++];
if (off + ul > len) return -1;
if (ul > 255) ul = 255;
memcpy(m->userinfo, data + off, ul);
m->userinfo[ul] = '\0';
return 0;
}
/* ── Обработка на инвайтере: верификация + добавление мембера + подпись дерева ── */
void chat_join_process_request(struct UTUN_INSTANCE* inst, uint64_t group_id,
uint64_t from_node, uint64_t join_key,
const uint8_t* member_data, size_t member_len) {
struct join_member_data m;
if (chat_join_parse_member(member_data, member_len, &m) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "%s: process_request parse failed ch=%llu from=%016llx", CJ_ID, (unsigned long long)group_id, (unsigned long long)from_node);
return;
}
char ch_id[32]; snprintf(ch_id, sizeof(ch_id), "%llu", (unsigned long long)group_id);
sqlite3* db = inst->topo_sqlite_db;
/* 1. привязка идентичности: node_id = derive(x25519) */
if (sc_derive_node_id_from_pubkey(m.x25519) != m.node_id) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: process_request node_id/x25519 mismatch (forgery) nid=0x%016llx ch=%s",
CJ_ID, (unsigned long long)m.node_id, ch_id);
return;
}
/* 2. верификация join_sig: sign(ch_x||ch_ed||node_id||x25519||join_ts) ключом ed25519 */
uint8_t ch_x[32], ch_ed[32];
if (!db || topo_node_sqlite_channel_get(db, ch_id, NULL, 0, NULL, ch_x, ch_ed, NULL) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "%s: process_request no channel ch=%s", CJ_ID, ch_id);
return;
}
uint8_t vmsg[128];
int vlen = member_sync_build_join_msg(ch_x, ch_ed, m.node_id, m.x25519, m.join_ts, vmsg, (int)sizeof(vmsg));
if (vlen < 0 || sc_ed25519_verify(m.ed25519, vmsg, (size_t)vlen, m.join_sig) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: process_request invalid join_sig nid=0x%016llx ch=%s",
CJ_ID, (unsigned long long)m.node_id, ch_id);
return;
}
/* 3. добавить мембера + подписать его pubkey (дерево приглашений) */
uint8_t tree_sig[64];
if (member_sync_sign_pubkey(inst->my_ed25519_privkey, m.x25519, tree_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: process_request sign_pubkey FAILED nid=0x%016llx", CJ_ID, (unsigned long long)m.node_id);
return;
}
int rc = member_sync_put(inst, ch_id, m.node_id, m.x25519, m.ed25519,
m.join_sig, m.join_ts, m.update_sig, m.update_ts,
m.userinfo, NULL, NULL, 0, inst->node_id, tree_sig);
if (rc < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: process_request member_sync_put FAILED ch=%s nid=0x%016llx rc=%d",
CJ_ID, ch_id, (unsigned long long)m.node_id, rc);
return;
}
if (rc & MS_APPLY_CHANGED)
member_sync_broadcast_one(inst, ch_id, m.node_id);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: process_request accepted joiner=0x%016llx ch=%s via=0x%016llx key=%016llx — signed by me",
CJ_ID, (unsigned long long)m.node_id, ch_id, (unsigned long long)from_node, (unsigned long long)join_key);
}
/* ── etcp_router recv ── */
static void join_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!entry || !conn || !entry->dgram || entry->len < ROUTER_SVC_PAYLOAD_OFF + 1) {
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
return;
}
struct UTUN_INSTANCE* inst = conn->instance;
if (!inst) { queue_dgram_free(entry); queue_entry_free(entry); return; }
const uint8_t* d = entry->dgram;
uint64_t from_node; memcpy(&from_node, d + ROUTER_SVC_SRC_OFF, 8);
uint64_t group_id; memcpy(&group_id, d + ROUTER_SVC_GROUP_OFF, 8);
const uint8_t* payload = d + ROUTER_SVC_PAYLOAD_OFF;
size_t plen = entry->len - ROUTER_SVC_PAYLOAD_OFF;
uint8_t subcmd = payload[0];
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "%s: recv subcmd=%02x from=%016llx ch=%llu len=%zu",
CJ_ID, subcmd, (unsigned long long)from_node, (unsigned long long)group_id, plen);
if (subcmd == JOIN_SUBCMD_KEY_REGISTER) {
/* payload: [subcmd:1][join_key:8] */
if (plen < 1 + 8) { queue_dgram_free(entry); queue_entry_free(entry); return; }
uint64_t join_key; memcpy(&join_key, payload + 1, 8);
_keys_put(group_id, join_key, from_node);
} else if (subcmd == JOIN_SUBCMD_REQUEST_FWD) {
/* payload: [subcmd:1][join_key:8][member...] */
if (plen < 1 + 8) { queue_dgram_free(entry); queue_entry_free(entry); return; }
uint64_t join_key; memcpy(&join_key, payload + 1, 8);
chat_join_process_request(inst, group_id, from_node, join_key, payload + 1 + 8, plen - 1 - 8);
} else {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "%s: unknown subcmd=%02x from=%016llx", CJ_ID, subcmd, (unsigned long long)from_node);
}
queue_dgram_free(entry); queue_entry_free(entry);
}
/* ── Жизненный цикл ── */
int chat_join_init(struct UTUN_INSTANCE* inst) {
if (!inst) return -1;
if (etcp_router_bind(inst, ETCP_RT_ID_JOIN, join_recv_cb) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: etcp_router_bind failed", CJ_ID);
return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: initialized", CJ_ID);
return 0;
}
void chat_join_destroy(struct UTUN_INSTANCE* inst) {
if (!inst) return;
etcp_router_unbind(inst, ETCP_RT_ID_JOIN);
_keys_free_all();
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: destroyed", CJ_ID);
}
/* ── Публичные операции ── */
void chat_join_register_key(struct UTUN_INSTANCE* inst, uint64_t channel_id,
uint64_t target_node_id, uint64_t join_key) {
if (!inst || channel_id == 0 || join_key == 0) return;
if (target_node_id == inst->node_id || target_node_id == 0) {
/* инвайтер и есть connection-узел — храним ключ локально */
_keys_put(channel_id, join_key, inst->node_id);
return;
}
uint8_t payload[1 + 8];
payload[0] = JOIN_SUBCMD_KEY_REGISTER;
memcpy(payload + 1, &join_key, 8);
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: register_key queue_entry_new failed", CJ_ID); return; }
entry->dgram = u_malloc(1 + sizeof(payload));
if (!entry->dgram) { queue_entry_free(entry); return; }
entry->dgram[0] = ETCP_RT_ID_JOIN;
memcpy(entry->dgram + 1, payload, sizeof(payload));
entry->len = 1 + (uint16_t)sizeof(payload);
int rc = etcp_route_send(inst, channel_id, target_node_id, entry, 1, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
if (rc != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: register_key etcp_route_send FAILED rc=%d ch=%llu → 0x%016llx",
CJ_ID, rc, (unsigned long long)channel_id, (unsigned long long)target_node_id);
queue_dgram_free(entry); queue_entry_free(entry);
return;
}
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: join-key registered ch=%llu → connection=0x%016llx key=%016llx",
CJ_ID, (unsigned long long)channel_id, (unsigned long long)target_node_id, (unsigned long long)join_key);
}
uint64_t chat_join_lookup_inviter(struct UTUN_INSTANCE* inst, uint64_t channel_id, uint64_t join_key) {
(void)inst;
struct join_key_entry* e = _keys_find(channel_id, join_key);
return e ? e->inviter_id : 0;
}
int chat_join_forward_request(struct UTUN_INSTANCE* inst, uint64_t channel_id,
uint64_t inviter_id, uint64_t join_key,
const struct join_member_data* m) {
if (!inst || !m || inviter_id == 0) return -1;
uint8_t mbuf[512]; size_t mlen = 0;
if (chat_join_serialize_member(m, mbuf, sizeof(mbuf), &mlen) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: forward serialize failed", CJ_ID);
return -1;
}
size_t plen = 1 + 8 + mlen;
uint8_t* payload = u_malloc(plen);
if (!payload) return -1;
payload[0] = JOIN_SUBCMD_REQUEST_FWD;
memcpy(payload + 1, &join_key, 8);
memcpy(payload + 1 + 8, mbuf, mlen);
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { u_free(payload); return -1; }
entry->dgram = u_malloc(1 + plen);
if (!entry->dgram) { u_free(payload); queue_entry_free(entry); return -1; }
entry->dgram[0] = ETCP_RT_ID_JOIN;
memcpy(entry->dgram + 1, payload, plen);
entry->len = 1 + (uint16_t)plen;
u_free(payload);
int rc = etcp_route_send(inst, channel_id, inviter_id, entry, 1, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
if (rc != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: forward etcp_route_send FAILED rc=%d ch=%llu → 0x%016llx",
CJ_ID, rc, (unsigned long long)channel_id, (unsigned long long)inviter_id);
queue_dgram_free(entry); queue_entry_free(entry);
return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: join request forwarded ch=%llu → inviter=0x%016llx joiner=0x%016llx",
CJ_ID, (unsigned long long)channel_id, (unsigned long long)inviter_id, (unsigned long long)m->node_id);
return 0;
}
+94
View File
@@ -0,0 +1,94 @@
/*
* chat_join.h — авторизация входа в канал (join) через connection-узел и инвайтера
*
* Flow (замена старого CHANNEL_INFO_REQ/CHANNEL_JOIN/WELCOME):
*
* инвайтер A connection-узел C джойнер J
* │ KEY_REGISTER(key,ch) │ │
* ├──────etcp_router─────────▶│ хранит {key→A, 10 мин} │
* │ │◀─── JOIN_INFO_REQ(key) ──────┤ (J подключился к C по ссылке)
* │ │── JOIN_INFO_RESP ───────────▶│ (ch_x25519, ch_ed25519, ch_sig, C-мембер)
* │ │◀─── JOIN_REQUEST ────────────┤ (key + полный набор мембера J)
* │ │ (валиден ключ → BGP new_conn, pending status=sent)
* │◀── REQUEST_FWD ───────────┤ │
* │ verif join_sig, node_id=derive(x25519) │
* │ member_sync_put(J, signed_by=A, signature=sign(A, J_x25519||"group_join"))
* │ ── merkle broadcast ──▶ (рекорд J дошёл по merkle) │
* │ │── JOIN_READY(ch) ────────────▶│ → topo_group_new_conn + ensure_channel_ready
*
* Модуль отвечает только за маршрутизируемую (etcp_router) часть:
* - регистрация ключа (инвайтер → connection),
* - форвард join-запроса (connection → инвайтер),
* - обработка на инвайтере: верификация + добавление мембера + подпись дерева.
* Прямой обмен (джойнер ↔ connection, JOIN_INFO_REQ/RESP, JOIN_REQUEST, JOIN_READY)
* и pending-трекинг готовности — в chat_sync.c.
*/
#ifndef CHAT_JOIN_H
#define CHAT_JOIN_H
#include <stdint.h>
#include <stddef.h>
struct UTUN_INSTANCE;
#ifdef __cplusplus
extern "C" {
#endif
/* etcp_router service id (connection ↔ инвайтер) */
#define ETCP_RT_ID_JOIN 0x33
/* subcommands (первый байт payload etcp_router) */
#define JOIN_SUBCMD_KEY_REGISTER 0x01 /* инвайтер → connection: {join_key:8} */
#define JOIN_SUBCMD_REQUEST_FWD 0x02 /* connection → инвайтер: {join_key:8, member...} */
#define JOIN_KEY_TTL_SECONDS 600 /* 10 минут хранения ключа на connection-узле */
/* Полный набор мембера джойнера (без подписи дерева — её ставит инвайтер). */
struct join_member_data {
uint64_t node_id;
const uint8_t* x25519; /* 32 */
const uint8_t* ed25519; /* 32 */
const uint8_t* join_sig; /* 64, может быть NULL */
uint64_t join_ts;
const uint8_t* update_sig; /* 64, может быть NULL */
uint64_t update_ts;
char userinfo[256]; /* JSON, null-terminated */
};
/* Сериализовать member-блок (без join_key). Возвращает 0 и *out_len, -1 если не влез. */
int chat_join_serialize_member(const struct join_member_data* m,
uint8_t* out, size_t out_sz, size_t* out_len);
/* Разобрать member-блок. Указатели ссылаются на data (не копируют). 0=ок, -1=ошибка. */
int chat_join_parse_member(const uint8_t* data, size_t len, struct join_member_data* m);
/* Инициализация: etcp_router_bind(ETCP_RT_ID_JOIN). */
int chat_join_init(struct UTUN_INSTANCE* inst);
/* Деинициализация. */
void chat_join_destroy(struct UTUN_INSTANCE* inst);
/* Инвайтер: зарегистрировать join-ключ на connection-узле.
* target==self → храним локально (инвайтер и есть connection-узел). */
void chat_join_register_key(struct UTUN_INSTANCE* inst, uint64_t channel_id,
uint64_t target_node_id, uint64_t join_key);
/* Connection-узел: найти инвайтера по ключу. 0 = нет/истёк. */
uint64_t chat_join_lookup_inviter(struct UTUN_INSTANCE* inst, uint64_t channel_id, uint64_t join_key);
/* Connection-узел: форварднуть join-запрос инвайтеру по etcp_router. 0=ок, -1=ошибка. */
int chat_join_forward_request(struct UTUN_INSTANCE* inst, uint64_t channel_id,
uint64_t inviter_id, uint64_t join_key,
const struct join_member_data* m);
/* Инвайтер (локально, когда connection==инвайтер, или из recv REQUEST_FWD):
* верификация join_sig + добавление мембера + подпись дерева. */
void chat_join_process_request(struct UTUN_INSTANCE* inst, uint64_t group_id,
uint64_t from_node, uint64_t join_key,
const uint8_t* member_data, size_t member_len);
#ifdef __cplusplus
}
#endif
#endif /* CHAT_JOIN_H */
+1 -1
View File
@@ -92,7 +92,7 @@ void chat_core_update_my_member(void) {
int r2 = member_sync_put(g_cc.inst, ch, myid, g_cc.inst->my_keys.public_key,
g_cc.inst->my_ed25519_pubkey, join_sig, join_ts,
update_sig, update_ts, juser, NULL, NULL, 0);
update_sig, update_ts, juser, NULL, NULL, 0, 0, NULL);
if (r2 < 0) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: member_sync_put(self) FAILED ch=%s rc=%d", CC_ID, ch, r2);
else if (r2 & MS_APPLY_CHANGED) member_sync_broadcast_one(g_cc.inst, ch, myid);
+301 -414
View File
@@ -5,6 +5,7 @@
#include "chat_setting.h"
#include "chat_event.h"
#include "chat_admin.h"
#include "chat_join.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/topo_group.h"
#include "../routing_layer/topo_group_invite.h"
@@ -45,6 +46,13 @@ struct channel_cache {
uint8_t synced;
};
struct join_pending {
uint64_t node_id; /* joiner node_id */
char ch_id[64];
struct ETCP_CONN* conn; /* прямое соединение к джойнеру */
struct join_pending* next;
};
struct chat_sync {
struct UTUN_INSTANCE* inst;
struct channel_cache* channels;
@@ -59,8 +67,10 @@ struct chat_sync {
uint8_t sync_scheduled;
uint64_t pending_invite_ch_id;
uint64_t pending_invite_node_id;
uint8_t pending_invite_is_inviter; /* 1 = inviter sends CHANNEL_INVITE on conn_up, 0 = joiner sends CHANNEL_INFO_REQ */
char pending_password[128]; /* пароль для CHANNEL_INFO_REQ */
uint8_t pending_invite_is_inviter; /* 1 = inviter sends CHANNEL_INVITE on conn_up, 0 = joiner sends JOIN_INFO_REQ */
char pending_password[128]; /* пароль для JOIN_INFO_REQ */
uint64_t pending_join_key; /* join-ключ из invite-ссылки (джойнер) */
struct join_pending* join_pending; /* connection-узел: ожидающие ready джойнеры */
/* Pending incoming invites (для ask/deny) */
uint64_t pending_inv_in_ch_id[4];
uint64_t pending_inv_in_node_id[4];
@@ -81,13 +91,12 @@ static const char* cs_msg_name(uint8_t type) {
case CS_MSG_PUSH: return "PUSH";
case CS_MSG_ACK_PUSH: return "ACK_PUSH";
case CS_MSG_SYNC_DONE: return "SYNC_DONE";
case CS_MSG_CHANNEL_INFO_REQ: return "CHANNEL_INFO_REQ";
case CS_MSG_CHANNEL_INFO_RESP: return "CHANNEL_INFO_RESP";
case CS_MSG_CHANNEL_JOIN: return "CHANNEL_JOIN";
case CS_MSG_WELCOME: return "WELCOME";
case CS_MSG_PEER_UPSERT: return "PEER_UPSERT";
case CS_MSG_ERROR: return "ERROR";
case CS_MSG_CHANNEL_INVITE: return "CHANNEL_INVITE";
case CS_MSG_JOIN_INFO_REQ: return "JOIN_INFO_REQ";
case CS_MSG_JOIN_INFO_RESP: return "JOIN_INFO_RESP";
case CS_MSG_JOIN_REQUEST: return "JOIN_REQUEST";
case CS_MSG_JOIN_READY: return "JOIN_READY";
default: return "???";
}
}
@@ -156,15 +165,13 @@ static struct channel_cache* cs_find(struct chat_sync* cs, const char* ch_id) {
static void cs_propagate(struct chat_sync* cs, const char* ch_id, uint64_t exclude_id,
const uint8_t* payload, size_t len);
static void cs_handle_channel_info_req(struct chat_sync* cs, uint64_t peer,
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);
static void cs_handle_channel_info_resp(struct chat_sync* cs, uint64_t peer,
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);
static void cs_handle_channel_join(struct chat_sync* cs, uint64_t peer,
static void cs_handle_join_request(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len);
static void cs_handle_welcome(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len);
static void cs_handle_peer_upsert(struct chat_sync* cs, uint64_t peer,
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,
const char* ch_id, const uint8_t* pl, size_t len);
@@ -173,6 +180,10 @@ static int cs_send_channel_invite(struct chat_sync* cs, struct UTUN_INSTANCE* in
const char* ch_name, uint64_t owner,
const uint8_t* ch_x25519, const uint8_t* ch_ed,
const uint8_t* ch_sig);
static void cs_on_member_applied(const char* ch_id, uint64_t node_id,
const uint8_t* x25519, uint64_t signed_by,
const uint8_t* signature, void* arg);
static void cs_pending_free_all(struct chat_sync* cs);
/* ── Recv dispatcher ── */
static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
@@ -193,9 +204,9 @@ static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
uint8_t type = d[9];
char ch_id[64];
if (cs_group_id_to_ch_id(g_cs->inst, group_id, ch_id, sizeof(ch_id)) != 0) {
/* CHANNEL_INVITE/INFO_REQ приходят для ещё не созданного у нас канала:
/* CHANNEL_INVITE/JOIN_INFO_REQ приходят для ещё не созданного у нас канала:
group_id == channel_id (numeric), поэтому выводим ch_id напрямую. */
if (type == CS_MSG_CHANNEL_INVITE || type == CS_MSG_CHANNEL_INFO_REQ) {
if (type == CS_MSG_CHANNEL_INVITE || type == CS_MSG_JOIN_INFO_REQ || type == CS_MSG_JOIN_REQUEST) {
snprintf(ch_id, sizeof(ch_id), "%llu", (unsigned long long)group_id);
} else {
u_free(entry->dgram); queue_entry_free(entry); return;
@@ -210,11 +221,10 @@ static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
CS_ID, cs_msg_name(type), (unsigned long long)peer, ch_id, plen);
switch (type) {
case CS_MSG_CHANNEL_INFO_REQ: DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle INFO_REQ", CS_ID); cs_handle_channel_info_req(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_CHANNEL_INFO_RESP:DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle INFO_RESP", CS_ID); cs_handle_channel_info_resp(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_CHANNEL_JOIN: DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle JOIN", CS_ID); cs_handle_channel_join(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_WELCOME: DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle WELCOME", CS_ID); cs_handle_welcome(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_PEER_UPSERT: DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle PEER_UPSERT", CS_ID); cs_handle_peer_upsert(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_JOIN_INFO_REQ: DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle JOIN_INFO_REQ", CS_ID); cs_handle_join_info_req(g_cs, peer, ch_id, pl, plen); 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(g_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(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_JOIN_READY: DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → handle JOIN_READY", CS_ID); cs_handle_join_ready(g_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);
if (g_cs->info_req_timer) { uasync_cancel_timeout(g_cs->inst->ua, g_cs->info_req_timer); g_cs->info_req_timer = NULL; }
@@ -448,22 +458,25 @@ static void _on_invite_sync_done(uint64_t peer, const char* ns, int result, void
u_free(sa);
}
/* ── send CHANNEL_INFO_REQ to start join handshake ── */
/* ── send JOIN_INFO_REQ to start join handshake ── */
static void cs_start_channel_join(struct chat_sync* cs, uint64_t peer) {
if (!cs || cs->info_req_timer) return;
char ch_id_str[64];
snprintf(ch_id_str, sizeof(ch_id_str), "%llu", (unsigned long long)cs->pending_invite_ch_id);
/* payload: [join_key:8][pass_len:1][password] */
size_t plen = strlen(cs->pending_password);
uint8_t req[130];
req[0] = CS_MSG_CHANNEL_INFO_REQ;
req[1] = (uint8_t)(plen > 127 ? 127 : plen);
if (plen > 0) memcpy(req + 2, cs->pending_password, plen > 127 ? 127 : plen);
cs_send(cs, ch_id_str, peer, req, 2 + (plen > 127 ? 127 : (int)plen));
uint8_t req[140];
req[0] = CS_MSG_JOIN_INFO_REQ;
memcpy(req + 1, &cs->pending_join_key, 8);
req[9] = (uint8_t)(plen > 127 ? 127 : plen);
if (plen > 0) memcpy(req + 10, cs->pending_password, plen > 127 ? 127 : plen);
cs_send(cs, ch_id_str, peer, req, 10 + (plen > 127 ? 127 : (int)plen));
cs->info_req_timer = uasync_set_timeout(cs->inst->ua,
CS_INFO_REQ_TIMEOUT_MS * 10, cs, cs_info_req_timeout_cb, "cs_info_req");
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: sent CHANNEL_INFO_REQ ch=%llu peer=0x%016llx pass_len=%zu",
CS_ID, (unsigned long long)cs->pending_invite_ch_id, (unsigned long long)peer, plen);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: sent JOIN_INFO_REQ ch=%llu peer=0x%016llx key=%016llx pass_len=%zu",
CS_ID, (unsigned long long)cs->pending_invite_ch_id, (unsigned long long)peer,
(unsigned long long)cs->pending_join_key, plen);
}
/* ── Connection callbacks ── */
@@ -633,7 +646,9 @@ int chat_sync_init(struct UTUN_INSTANCE* inst) {
cs->join_timer = NULL;
member_sync_init(inst);
member_sync_add_apply_cbk(cs_on_member_applied, cs);
chat_admin_init(inst);
chat_join_init(inst);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: initialized", CS_ID);
return 0;
@@ -646,8 +661,11 @@ void chat_sync_destroy(struct UTUN_INSTANCE* inst) {
if (cs->sync_timer) { uasync_cancel_timeout(inst->ua, cs->sync_timer); cs->sync_timer = NULL; }
if (cs->invite_send_timer) { uasync_cancel_timeout(inst->ua, cs->invite_send_timer); cs->invite_send_timer = NULL; }
if (cs->sync_scheduled) { cs->sync_scheduled = 0; cs_flush_sync(cs); }
member_sync_remove_apply_cbk(cs_on_member_applied, cs);
cs_pending_free_all(cs);
member_sync_destroy(inst);
chat_admin_destroy(inst);
chat_join_destroy(inst);
cs->initialized = 0; g_cs = NULL;
etcp_unbind(inst, ETCP_RT_ID_CHAT_SYNC);
@@ -704,20 +722,6 @@ struct chat_invite {
int addr_count;
};
/* коллбэк invite-подключения (topo_group_invite): nq->handle уже установлен модулем,
здесь только лог результата. Реальный join-протокол запускается через cs_on_conn_up
(глобальный callback), сохранение в БД — через cs_handle_channel_info_resp. */
static void cs_tgi_cb(uint64_t node_id, uint64_t group_id, int event, void* arg) {
(void)arg; (void)group_id;
if (event == TGI_EVENT_JOIN) {
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "channel invite: JOINED — connected to 0x%016llx, online now",
(unsigned long long)node_id);
} else {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "channel invite: TIMEOUT — could not connect to 0x%016llx",
(unsigned long long)node_id);
}
}
struct cm_invite_wrap { struct chat_invite inv; };
static void cm_invite_trampoline(void* arg) {
@@ -777,13 +781,13 @@ static void cm_invite_trampoline(void* arg) {
u_free(ni); u_free(w); return;
}
/* topo_group_invite_join сам создаст TOPO_GROUP, найдёт/создаст соединение (ncd),
запустит INVITE_INFO. DB channel + sync создаются позже в cs_handle_channel_info_resp после верификации */
/* topo_group_invite_to_channel: прямое ncd-подключение к connection-узлу БЕЗ INVITE_INFO.
Реальный join-протокол (JOIN_INFO_REQ → … → JOIN_READY) запускается по cs_on_conn_up. */
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "channel invite: connecting to 0x%016llx via %d addresses",
(unsigned long long)node_id, inv->addr_count);
int r = topo_group_invite_join(inst, channel_id, ni, node_id, cs_tgi_cb, NULL);
int r = topo_group_invite_to_channel(inst, channel_id, ni, node_id);
if (r < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "channel invite: FAILED — topo_group_invite_join error=%d", r);
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "channel invite: FAILED — topo_group_invite_to_channel error=%d", r);
}
/* free caller-owned ni */
@@ -797,7 +801,7 @@ void chat_sync_connect_from_invite(uint64_t channel_id, uint64_t node_id,
const uint8_t* pubkey_bin,
const uint8_t* addrs_data, int addr_count,
int addrs_data_len,
const char* password) {
const char* password, uint64_t join_key) {
if (!g_cs || !g_cs->inst || !g_cs->inst->ua) {
int r = -7;
#ifdef __ANDROID__
@@ -826,6 +830,7 @@ void chat_sync_connect_from_invite(uint64_t channel_id, uint64_t node_id,
g_cs->pending_invite_ch_id = channel_id;
g_cs->pending_invite_node_id = node_id;
g_cs->pending_join_key = join_key;
if (password && password[0]) {
size_t pl = strlen(password);
if (pl >= sizeof(g_cs->pending_password)) pl = sizeof(g_cs->pending_password) - 1;
@@ -933,481 +938,363 @@ static int _get_node_name(sqlite3* db, uint64_t node_id, char* out, size_t sz) {
return rc;
}
/* ─── CHANNEL_INFO_REQ (0x09): joiner → inviter ─── */
/* ─── JOIN_INFO_REQ (0x10): joiner → connection-узел ─── */
static void cs_handle_channel_info_req(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: INFO_REQ ch=%s from=%016llx len=%zu", CS_ID, ch_id, (unsigned long long)peer, len);
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) {
/* payload: [join_key:8][pass_len:1][password] */
if (len < 9) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_REQ too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
uint64_t join_key; memcpy(&join_key, pl, 8);
uint64_t channel_id = strtoull(ch_id, NULL, 10);
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_REQ ch=%s from=%016llx key=%016llx len=%zu",
CS_ID, ch_id, (unsigned long long)peer, (unsigned long long)join_key, len);
/* joiner responded to our CHANNEL_INVITE — cancel invite send timer */
if (cs->invite_send_timer) { uasync_cancel_timeout(cs->inst->ua, cs->invite_send_timer); cs->invite_send_timer = NULL; }
/* Parse password from request (v2 format: type + pass_len + optional_password) */
char req_password[128] = "";
if (len >= 1) {
uint8_t plen2 = pl[0];
if (plen2 > 0 && plen2 <= len - 1) {
if (plen2 > 127) plen2 = 127;
memcpy(req_password, pl + 1, plen2); req_password[plen2] = '\0';
}
/* 1. валидация join-ключа (зарегистрирован инвайтером на connection-узле, TTL 10 мин) */
uint64_t inviter = chat_join_lookup_inviter(cs->inst, channel_id, join_key);
if (inviter == 0) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_REQ invalid/expired join key ch=%s from=%016llx key=%016llx",
CS_ID, ch_id, (unsigned long long)peer, (unsigned long long)join_key);
uint8_t err[1] = { CS_MSG_ERROR }; cs_send(cs, ch_id, peer, err, 1);
return;
}
/* Check password if server has one configured */
/* 2. пароль (если настроен на сервере) */
char req_password[128] = "";
if (len >= 10) {
uint8_t plen2 = pl[9];
if (plen2 > 0 && plen2 <= len - 10) {
if (plen2 > 127) plen2 = 127;
memcpy(req_password, pl + 10, plen2); req_password[plen2] = '\0';
}
}
{
struct global_config *gcfg = &cs->inst->config->global;
if (gcfg->chatserver_enabled && gcfg->chatserver_join_password[0]) {
if (strcmp(req_password, gcfg->chatserver_join_password) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_REQ wrong password ch=%s from=%016llx (got '%s', expected '%s')",
CS_ID, ch_id, (unsigned long long)peer, req_password, gcfg->chatserver_join_password);
uint8_t err[2] = { CS_MSG_ERROR, 0x01 }; /* 0x01 = wrong password */
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_REQ wrong password ch=%s from=%016llx", CS_ID, ch_id, (unsigned long long)peer);
uint8_t err[2] = { CS_MSG_ERROR, 0x01 };
cs_send(cs, ch_id, peer, err, 2);
return;
}
}
}
char name[128]; uint64_t owner; uint8_t x25519[32], ed_pub[32], ch_sig[64];
/* 3. инфо канала */
char name[128]; uint64_t owner; uint8_t ch_x25519[32], ch_ed[32], ch_sig[64];
if (topo_node_sqlite_channel_get(cs->inst->topo_sqlite_db,
ch_id, name, (int)sizeof(name), &owner, x25519, ed_pub, ch_sig) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_REQ unknown ch=%s from=%016llx", CS_ID, ch_id, (unsigned long long)peer);
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(cs, ch_id, peer, err, 1);
return;
}
/* 4. мой мембер-рекорд (join_sig/update_sig + дерево приглашений) */
uint64_t myid = cs->inst->node_id;
if (memcmp(cs->inst->my_keys.public_key, x25519, 32) != 0)
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_RESP pubkey MISMATCH: my=%016llx ch=%016llx — channel was created with DIFFERENT keys!",
CS_ID, *(const uint64_t*)cs->inst->my_keys.public_key, *(const uint64_t*)x25519);
/* load or create join_sig */
uint8_t my_join_sig[64] = {0}; uint64_t my_join_ts = 0;
{
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) {
uint8_t join_msg[128];
my_join_ts = (uint64_t)ntp_time_get_seconds(cs->inst);
int jmlen = member_sync_build_join_msg(x25519, ed_pub, myid,
cs->inst->my_keys.public_key, my_join_ts,
int jmlen = member_sync_build_join_msg(ch_x25519, ch_ed, myid, cs->inst->my_keys.public_key, my_join_ts,
join_msg, (int)sizeof(join_msg));
if (jmlen >= 0) cs_ed25519_sign(cs->inst->my_ed25519_privkey, join_msg, (size_t)jmlen, my_join_sig);
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_RESP created NEW join_sig: myid=%016llx ts=%llu",
CS_ID, (unsigned long long)myid, (unsigned long long)my_join_ts);
}
}
/* generate update_sig */
uint8_t my_update_sig[64] = {0}; uint64_t my_update_ts;
{
char juser[256]; snprintf(juser, sizeof(juser), "{\"name\":\"%s\"}",
cs->inst->name[0] ? cs->inst->name : "");
char juser[256]; snprintf(juser, sizeof(juser), "{\"name\":\"%s\"}", cs->inst->name[0] ? cs->inst->name : "");
my_update_ts = (uint64_t)ntp_time_get_seconds(cs->inst);
uint8_t umsg[256];
int ulen = member_sync_build_update_msg(my_join_sig, my_update_ts, juser,
umsg, (int)sizeof(umsg));
int ulen = member_sync_build_update_msg(my_join_sig, my_update_ts, juser, umsg, (int)sizeof(umsg));
if (ulen >= 0) cs_ed25519_sign(cs->inst->my_ed25519_privkey, umsg, (size_t)ulen, my_update_sig);
}
uint64_t my_signed_by = 0; uint8_t my_signature[64] = {0};
topo_node_sqlite_member_get_sign(cs->inst->topo_sqlite_db, ch_id, myid, &my_signed_by, my_signature);
uint8_t buf[1024]; size_t boff = 0;
buf[boff++] = CS_MSG_CHANNEL_INFO_RESP;
uint8_t nl = (uint8_t)strlen(name);
buf[boff++] = nl; memcpy(buf + boff, name, nl); boff += nl;
/* 5. JOIN_INFO_RESP: инфо канала + мой мембер-рекорд */
char juser[256]; snprintf(juser, sizeof(juser), "{\"name\":\"%s\"}", cs->inst->name[0] ? cs->inst->name : "");
uint8_t il = (uint8_t)strlen(juser);
uint8_t buf[1536]; size_t boff = 0;
buf[boff++] = CS_MSG_JOIN_INFO_RESP;
uint8_t cnl = (uint8_t)strlen(name); buf[boff++] = cnl; memcpy(buf + boff, name, cnl); boff += cnl;
memcpy(buf + boff, &owner, 8); boff += 8;
memcpy(buf + boff, x25519, 32); boff += 32;
memcpy(buf + boff, ed_pub, 32); boff += 32;
memcpy(buf + boff, ch_x25519, 32); boff += 32;
memcpy(buf + boff, ch_ed, 32); boff += 32;
memcpy(buf + boff, ch_sig, 64); boff += 64;
uint8_t inv_flags = (my_join_sig[0] || my_join_ts) ? PEERS_FLAG_HAS_JOIN : 0;
buf[boff++] = inv_flags;
if (inv_flags & PEERS_FLAG_HAS_JOIN) {
memcpy(buf + boff, my_join_sig, 64); boff += 64;
memcpy(buf + boff, &my_join_ts, 8); boff += 8;
}
memcpy(buf + boff, &myid, 8); boff += 8;
memcpy(buf + boff, cs->inst->my_keys.public_key, 32); boff += 32;
memcpy(buf + boff, cs->inst->my_ed25519_pubkey, 32); boff += 32;
memcpy(buf + boff, my_join_sig, 64); boff += 64;
memcpy(buf + boff, &my_join_ts, 8); boff += 8;
memcpy(buf + boff, my_update_sig, 64); boff += 64;
memcpy(buf + boff, &my_update_ts, 8); boff += 8;
char juser[256]; snprintf(juser, sizeof(juser), "{\"name\":\"%s\"}",
cs->inst->name[0] ? cs->inst->name : "");
uint8_t il = (uint8_t)strlen(juser);
buf[boff++] = 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(cs, ch_id, peer, buf, boff);
}
/* ─── CHANNEL_INFO_RESP (0x0A): inviter → joiner ─── */
/* ─── JOIN_INFO_RESP (0x11): connection-узел → joiner ─── */
static void cs_handle_channel_info_resp(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: INFO_RESP ch=%s from=%016llx len=%zu", CS_ID, ch_id, (unsigned long long)peer, len);
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) {
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 (cs->info_req_timer) { uasync_cancel_timeout(cs->inst->ua, cs->info_req_timer); cs->info_req_timer = NULL; }
if (len < 1) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_RESP too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
uint8_t nl = pl[0];
if (1 + nl + 8 + 32 + 32 + 64 + 1 + 64 + 8 + 1 > len) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_RESP truncated len=%zu", CS_ID, len); return; }
const uint8_t* p = pl + 1;
char name[128]; memcpy(name, p, nl); name[nl] = '\0'; p += nl;
uint64_t owner; memcpy(&owner, p, 8); p += 8;
const uint8_t* x25519 = p; p += 32;
const uint8_t* ed_pub = p; p += 32;
const uint8_t* ch_sig = p; p += 64;
uint8_t inv_flags = *p++;
const uint8_t* inviter_join_sig = NULL; uint64_t inviter_join_ts = 0;
if (inv_flags & PEERS_FLAG_HAS_JOIN) {
if (p + 72 > pl + len) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_RESP truncated at inviter_join_sig peer=%016llx", CS_ID, (unsigned long long)peer); return; }
inviter_join_sig = p; p += 64; memcpy(&inviter_join_ts, p, 8); p += 8;
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;
uint8_t cnl = *p++;
if (cnl + 8 + 32 + 32 + 64 + 8 + 32 + 32 + 64 + 8 + 64 + 8 + 1 + 8 + 64 > len) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_RESP truncated len=%zu", CS_ID, len); return;
}
const uint8_t* inviter_update_sig = p; p += 64; uint64_t inviter_update_ts;
memcpy(&inviter_update_ts, p, 8); p += 8;
uint8_t inv_userinfo_len = *p++;
char inv_userinfo[128] = "";
if (p + inv_userinfo_len <= pl + len) { memcpy(inv_userinfo, p, inv_userinfo_len); inv_userinfo[inv_userinfo_len] = '\0'; }
char name[128]; size_t cn = cnl > 127 ? 127 : cnl; memcpy(name, p, cn); name[cn] = '\0'; p += cnl;
uint64_t owner; memcpy(&owner, p, 8); p += 8;
const uint8_t* ch_x25519 = p; p += 32;
const uint8_t* ch_ed = p; p += 32;
const uint8_t* ch_sig = p; p += 64;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_RESP from peer=%016llx owner=%016llx ch_x25519=%016llx ch_ed=%016llx name=%s",
uint64_t c_node_id; memcpy(&c_node_id, p, 8); p += 8;
const uint8_t* c_x25519 = p; p += 32;
const uint8_t* c_ed = p; p += 32;
const uint8_t* c_join_sig = p; p += 64;
uint64_t c_join_ts; memcpy(&c_join_ts, p, 8); p += 8;
const uint8_t* c_update_sig = p; p += 64;
uint64_t c_update_ts; memcpy(&c_update_ts, p, 8); p += 8;
uint8_t c_uil = *p++;
char c_userinfo[256] = "";
if (c_uil && p + c_uil <= pl + len) { size_t u = c_uil > 255 ? 255 : c_uil; memcpy(c_userinfo, p, u); c_userinfo[u] = '\0'; p += c_uil; }
uint64_t c_signed_by; memcpy(&c_signed_by, p, 8); p += 8;
const uint8_t* c_signature = p; p += 64;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_RESP from=%016llx owner=%016llx ch_x=%016llx ch_ed=%016llx name=%s conn_node=%016llx",
CS_ID, (unsigned long long)peer, (unsigned long long)owner,
*(const uint64_t*)x25519, *(const uint64_t*)ed_pub, name);
*(const uint64_t*)ch_x25519, *(const uint64_t*)ch_ed, name, (unsigned long long)c_node_id);
/* verify channel signature */
uint8_t vmsg[1024]; size_t vlen = 0;
vlen += snprintf((char*)vmsg + vlen, sizeof(vmsg) - vlen, "%s", ch_id) + 1;
vlen += snprintf((char*)vmsg + vlen, sizeof(vmsg) - vlen, "%s", name) + 1;
memcpy(vmsg + vlen, &owner, 8); vlen += 8;
memcpy(vmsg + vlen, x25519, 32); vlen += 32;
memcpy(vmsg + vlen, ed_pub, 32); vlen += 32;
if (cs_ed25519_verify(ed_pub, vmsg, vlen, ch_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_RESP invalid ch_sig ch=%s", CS_ID, ch_id);
memcpy(vmsg + vlen, ch_x25519, 32); vlen += 32;
memcpy(vmsg + vlen, ch_ed, 32); vlen += 32;
if (cs_ed25519_verify(ch_ed, vmsg, vlen, ch_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_RESP invalid ch_sig ch=%s", CS_ID, ch_id);
return;
}
/* save channel to local DB */
int cprc = topo_node_sqlite_channel_put(cs->inst->topo_sqlite_db,
ch_id, name, owner, x25519, NULL, ed_pub, NULL, ch_sig);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: INFO_RESP channel_put rc=%d ch=%s owner=%016llx ch_x25519=%016llx",
CS_ID, cprc, ch_id, (unsigned long long)owner, *(const uint64_t*)x25519);
/* save channel */
topo_node_sqlite_channel_put(cs->inst->topo_sqlite_db, ch_id, name, owner, ch_x25519, NULL, ch_ed, NULL, ch_sig);
chat_core_ensure_channel_ready(ch_id);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: channel ready for sync ch=%s, db_sync will pick up via periodic check or active conn",
CS_ID, ch_id);
/* verify inviter's join_sig AND save node info using ETCP-authenticated keys */
{ struct ETCP_CONN* inv_conn = cs_find_conn_for_node(cs->inst, peer);
if (!inv_conn) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_RESP no ETCP conn for inviter peer=%016llx", CS_ID, (unsigned long long)peer);
return;
}
const uint8_t* inv_x25519 = inv_conn->crypto_ctx.peer_public_key;
const uint8_t* inv_ed = inv_conn->peer_ed25519_pubkey;
static const uint8_t zero_sig[64] = {0};
const uint8_t* jsig = inviter_join_sig ? inviter_join_sig : zero_sig;
uint8_t ivmsg[256];
int ilen = member_sync_build_update_msg(jsig, inviter_update_ts,
inv_userinfo[0] ? inv_userinfo : "",
ivmsg, (int)sizeof(ivmsg));
if (ilen < 0 || cs_ed25519_verify(inv_ed, ivmsg, (size_t)ilen, inviter_update_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INFO_RESP invalid inviter_update_sig peer=%016llx ts=%llu",
CS_ID, (unsigned long long)peer, (unsigned long long)inviter_update_ts);
}
sqlite3* vdb = cs->inst->topo_sqlite_db;
int rc = topo_node_sqlite_member_block_put(vdb, ch_id, peer, inviter_join_sig, inviter_join_ts, inviter_update_sig, inviter_update_ts, inv_x25519, inv_ed, inv_userinfo, inviter_join_sig);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: member_put(inviter) FAILED ch=%s peer=%016llx rc=%d", CS_ID, ch_id, (unsigned long long)peer, rc);
char inv_name[128]; json_flat_get(inv_userinfo, "name", inv_name, sizeof(inv_name));
topo_node_sqlite_node_update_verified(vdb, peer, inv_name, inv_x25519, inv_ed, inviter_join_ts, ntp_time_get_seconds(cs->inst));
}
/* добавить connection-узел в мемберы (дерево приглашений из INFO_RESP) */
sqlite3* vdb = cs->inst->topo_sqlite_db;
topo_node_sqlite_member_block_put(vdb, ch_id, c_node_id, c_join_sig, c_join_ts,
c_update_sig, c_update_ts, c_x25519, c_ed,
c_userinfo, NULL, c_signed_by, c_signature);
{ char cn_name[128]; json_flat_get(c_userinfo, "name", cn_name, sizeof(cn_name));
topo_node_sqlite_node_update_verified(vdb, c_node_id, cn_name, c_x25519, c_ed, c_join_ts, ntp_time_get_seconds(cs->inst)); }
/* generate our own join_sig */
/* построить собственный join_sig */
uint64_t myid = cs->inst->node_id;
uint8_t my_x25519[32];
memcpy(my_x25519, cs->inst->my_keys.public_key, 32);
const uint8_t* my_x25519 = cs->inst->my_keys.public_key;
uint8_t join_sig[64]; uint64_t join_ts;
{
uint8_t msg[128];
join_ts = (uint64_t)ntp_time_get_seconds(cs->inst);
int mlen = member_sync_build_join_msg(x25519, ed_pub, myid, my_x25519, join_ts,
msg, (int)sizeof(msg));
if (mlen >= 0) cs_ed25519_sign(cs->inst->my_ed25519_privkey, msg, (size_t)mlen, join_sig);
int mlen = member_sync_build_join_msg(ch_x25519, ch_ed, myid, my_x25519, join_ts, msg, (int)sizeof(msg));
if (mlen < 0) { memset(join_sig, 0, 64); DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_RESP build_join_msg failed", CS_ID); return; }
cs_ed25519_sign(cs->inst->my_ed25519_privkey, msg, (size_t)mlen, join_sig);
}
uint8_t my_update_sig[64] = {0}; uint64_t my_update_ts;
{
char juser[256]; snprintf(juser, sizeof(juser), "{\"name\":\"%s\"}", cs->inst->name[0] ? cs->inst->name : "");
my_update_ts = (uint64_t)ntp_time_get_seconds(cs->inst);
uint8_t umsg[256];
int ulen = member_sync_build_update_msg(join_sig, my_update_ts, juser, umsg, (int)sizeof(umsg));
if (ulen >= 0) cs_ed25519_sign(cs->inst->my_ed25519_privkey, umsg, (size_t)ulen, my_update_sig);
}
/* send JOIN_CHANNEL */
uint8_t jbuf[512]; size_t joff = 0;
jbuf[joff++] = CS_MSG_CHANNEL_JOIN;
memcpy(jbuf + joff, &myid, 8); joff += 8;
memcpy(jbuf + joff, my_x25519, 32); joff += 32;
memcpy(jbuf + joff, cs->inst->my_ed25519_pubkey, 32); joff += 32;
memcpy(jbuf + joff, join_sig, 64); joff += 64;
memcpy(jbuf + joff, &join_ts, 8); joff += 8;
char juser2[256]; snprintf(juser2, sizeof(juser2), "{\"name\":\"%s\"}",
cs->inst->name[0] ? cs->inst->name : "");
uint8_t nml = (uint8_t)strlen(juser2);
jbuf[joff++] = nml; memcpy(jbuf + joff, juser2, nml); joff += nml;
/* send JOIN_REQUEST: [join_key:8][member...] */
struct join_member_data jm;
memset(&jm, 0, sizeof(jm));
jm.node_id = myid; jm.x25519 = my_x25519; jm.ed25519 = cs->inst->my_ed25519_pubkey;
jm.join_sig = join_sig; jm.join_ts = join_ts;
jm.update_sig = my_update_sig; jm.update_ts = my_update_ts;
snprintf(jm.userinfo, sizeof(jm.userinfo), "{\"name\":\"%s\"}", cs->inst->name[0] ? cs->inst->name : "");
uint8_t jm_buf[512]; size_t jm_len = 0;
if (chat_join_serialize_member(&jm, jm_buf, sizeof(jm_buf), &jm_len) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_REQUEST serialize failed", CS_ID);
return;
}
uint8_t jbuf[520]; size_t joff = 0;
jbuf[joff++] = CS_MSG_JOIN_REQUEST;
memcpy(jbuf + joff, &cs->pending_join_key, 8); joff += 8;
memcpy(jbuf + joff, jm_buf, jm_len); joff += jm_len;
int jrc = cs_send(cs, ch_id, peer, jbuf, joff);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN sent myid=%016llx x25519=%016llx join_ts=%llu ch=%s to=0x%016llx rc=%d",
CS_ID, (unsigned long long)myid, *(const uint64_t*)my_x25519, (unsigned long long)join_ts, ch_id, (unsigned long long)peer, jrc);
cs->join_timer = uasync_set_timeout(cs->inst->ua,
CS_JOIN_TIMEOUT_MS * 10, cs, cs_join_timeout_cb, "cs_join");
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_REQUEST sent myid=%016llx key=%016llx ch=%s to=0x%016llx rc=%d",
CS_ID, (unsigned long long)myid, (unsigned long long)cs->pending_join_key, ch_id, (unsigned long long)peer, jrc);
cs->join_timer = uasync_set_timeout(cs->inst->ua, CS_JOIN_TIMEOUT_MS * 10, cs, cs_join_timeout_cb, "cs_join");
}
/* ─── CHANNEL_JOIN (0x0B): joiner → inviter ─── */
/* ─── pending-джойнеры (connection-узел): node_id → прямое соединение ─── */
static void cs_handle_channel_join(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
if (len < 8 + 32 + 32 + 64 + 8 + 1) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_JOIN too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
const uint8_t* p = pl;
uint64_t node_id; memcpy(&node_id, p, 8); p += 8;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN ch=%s node=%016llx from=%016llx", CS_ID, ch_id, (unsigned long long)node_id, (unsigned long long)peer);
const uint8_t* x25519 = p; p += 32;
const uint8_t* ed_pub = p; p += 32;
const uint8_t* join_sig = p; p += 64;
uint64_t join_ts; memcpy(&join_ts, p, 8); p += 8;
uint8_t userinfo_len = *p++;
char joiner_userinfo[128] = "";
if (p + userinfo_len <= pl + len) { memcpy(joiner_userinfo, p, userinfo_len); joiner_userinfo[userinfo_len] = '\0'; p += userinfo_len; }
static struct join_pending* cs_pending_find(struct chat_sync* cs, uint64_t node_id, const char* ch_id) {
for (struct join_pending* jp = cs->join_pending; jp; jp = jp->next)
if (jp->node_id == node_id && strcmp(jp->ch_id, ch_id) == 0) return jp;
return NULL;
}
/* verify join_sig */
uint8_t ch_x25519[32], ch_ed25519[32];
if (topo_node_sqlite_channel_get(cs->inst->topo_sqlite_db, ch_id,
NULL, 0, NULL, ch_x25519, ch_ed25519, NULL) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN channel not in DB ch=%s", CS_ID, ch_id);
return;
}
uint8_t vmsg[128];
int vlen = member_sync_build_join_msg(ch_x25519, ch_ed25519, node_id, x25519, join_ts,
vmsg, (int)sizeof(vmsg));
if (vlen < 0 || cs_ed25519_verify(ed_pub, vmsg, (size_t)vlen, join_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN invalid sig node=0x%016llx ch=%s — signed(node=%016llx x25519=%016llx ed=%016llx ts=%llu)", CS_ID,
(unsigned long long)node_id, ch_id,
(unsigned long long)node_id, *(const uint64_t*)x25519, *(const uint64_t*)ed_pub, (unsigned long long)join_ts);
return;
}
static void cs_pending_add(struct chat_sync* cs, uint64_t node_id, const char* ch_id, struct ETCP_CONN* conn) {
if (cs_pending_find(cs, node_id, ch_id)) return;
struct join_pending* jp = u_malloc(sizeof(*jp));
if (!jp) return;
jp->node_id = node_id; snprintf(jp->ch_id, sizeof(jp->ch_id), "%s", ch_id); jp->conn = conn;
jp->next = cs->join_pending; cs->join_pending = jp;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: pending join added node=0x%016llx ch=%s", CS_ID, (unsigned long long)node_id, ch_id);
}
sqlite3* db = cs->inst->topo_sqlite_db;
int rc = topo_node_sqlite_member_block_put(db, ch_id, node_id, join_sig, join_ts, NULL, 0, x25519, ed_pub, joiner_userinfo, NULL);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: member_put(joiner) FAILED ch=%s node=0x%016llx rc=%d", CS_ID, ch_id, (unsigned long long)node_id, rc);
{ char j_name[128]; json_flat_get(joiner_userinfo, "name", j_name, sizeof(j_name));
topo_node_sqlite_node_update_verified(db, node_id, j_name, x25519, ed_pub, join_ts, ntp_time_get_seconds(cs->inst)); }
/* build WELCOME with all current peers */
uint8_t wbuf[8192];
{
uint8_t peers_data[8192]; size_t peers_len = 0;
if (topo_node_sqlite_channel_peers_all(db, ch_id,
peers_data, sizeof(peers_data), &peers_len) == 0) {
size_t woff = 0;
wbuf[woff++] = CS_MSG_WELCOME;
if (woff + peers_len <= sizeof(wbuf))
{ memcpy(wbuf + woff, peers_data, peers_len); woff += peers_len; }
cs_send(cs, ch_id, peer, wbuf, woff);
static void cs_pending_remove(struct chat_sync* cs, uint64_t node_id, const char* ch_id) {
struct join_pending** p = &cs->join_pending;
while (*p) {
if ((*p)->node_id == node_id && strcmp((*p)->ch_id, ch_id) == 0) {
struct join_pending* r = *p; *p = r->next; u_free(r);
return;
}
p = &(*p)->next;
}
}
/* propagate PEER_UPSERT to other channel members */
{
uint8_t ubuf[512]; size_t uoff = 0;
ubuf[uoff++] = CS_MSG_PEER_UPSERT;
memcpy(ubuf + uoff, &node_id, 8); uoff += 8;
memcpy(ubuf + uoff, x25519, 32); uoff += 32;
memcpy(ubuf + uoff, ed_pub, 32); uoff += 32;
memcpy(ubuf + uoff, join_sig, 64); uoff += 64;
memcpy(ubuf + uoff, &join_ts, 8); uoff += 8;
ubuf[uoff++] = userinfo_len;
memcpy(ubuf + uoff, joiner_userinfo, userinfo_len); uoff += userinfo_len;
cs_propagate(cs, ch_id, peer, ubuf, uoff);
}
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; }
cs->join_pending = NULL;
}
/* add to channel cache — always refresh to include new peer */
cs_refresh_channels(cs);
/* ─── отправка JOIN_READY джойнеру + снятие pending (connection-узел) ─── */
{ uint8_t mevt[1 + 64 + 1 + CHAT_MEMBER_DISPLAY_SIZE]; uint8_t cl = (uint8_t)strlen(ch_id);
mevt[0] = cl; memcpy(mevt + 1, ch_id, cl); mevt[1 + cl] = 1;
if (chat_core_get_single_member(ch_id, node_id, mevt + 1 + cl + 1) == 0)
chat_event_post(CHAT_EVT_MEMBER_UPDATED, mevt, 1 + cl + 1 + CHAT_MEMBER_DISPLAY_SIZE); }
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN calling cs_post_channel_online ch=%s", CS_ID, ch_id);
cs_post_channel_online(cs, ch_id);
/* initiate member sync with the joiner (server side) */
struct invite_sync_arg* sa = u_malloc(sizeof(struct invite_sync_arg));
if (sa) { sa->node_id = node_id;
strncpy(sa->ch_id, ch_id, sizeof(sa->ch_id) - 1); sa->ch_id[sizeof(sa->ch_id) - 1] = '\0';
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN alloc sa=%p node=0x%016llx ch=%s", CS_ID, (void*)sa, (unsigned long long)sa->node_id, sa->ch_id);
member_sync_start(cs->inst, node_id, ch_id, _on_invite_sync_done, sa); }
else member_sync_start(cs->inst, node_id, ch_id, NULL, NULL);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN starting member_sync with joiner node=0x%016llx ch=%s done_cb=%s",
CS_ID, (unsigned long long)node_id, ch_id, sa ? "yes" : "no");
/* if we're the inviter — joiner joined successfully, notify GUI */
if (cs->pending_invite_ch_id != 0 && cs->pending_invite_is_inviter) {
uint64_t ch_id_num = strtoull(ch_id, NULL, 10);
uint8_t evt[20]; memcpy(evt, &node_id, 8); int r = 0; memcpy(evt + 8, &r, 4);
memcpy(evt + 12, &ch_id_num, 8);
chat_event_post(CHAT_EVT_CONNECT_RESULT, evt, 20);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN posted CHAT_EVT_CONNECT_RESULT to GUI node=0x%016llx ch=%s",
CS_ID, (unsigned long long)node_id, ch_id);
cs->pending_invite_node_id = 0; cs->pending_invite_ch_id = 0;
cs->pending_invite_is_inviter = 0;
if (cs->invite_send_timer) { uasync_cancel_timeout(cs->inst->ua, cs->invite_send_timer); cs->invite_send_timer = NULL; }
}
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN accepted node=0x%016llx ch=%s",
static void cs_send_join_ready(struct chat_sync* cs, const char* ch_id, uint64_t node_id) {
uint64_t group_id = strtoull(ch_id, NULL, 10);
uint8_t rbuf[9];
rbuf[0] = CS_MSG_JOIN_READY;
memcpy(rbuf + 1, &group_id, 8);
cs_send(cs, ch_id, node_id, rbuf, 9);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_READY sent to 0x%016llx ch=%s",
CS_ID, (unsigned long long)node_id, ch_id);
cs_pending_remove(cs, node_id, ch_id);
}
/* ─── WELCOME (0x0C): inviter → joiner with full peer list ─── */
/* ─── коллбэк member_sync: пришёл рекорд мембера → отправить JOIN_READY джойнеру ─── */
static void cs_handle_welcome(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
if (cs->join_timer) { uasync_cancel_timeout(cs->inst->ua, cs->join_timer); cs->join_timer = NULL; }
(void)peer;
if (len < 2) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
sqlite3* db = cs->inst->topo_sqlite_db;
static void cs_on_member_applied(const char* ch_id, uint64_t node_id,
const uint8_t* x25519, uint64_t signed_by,
const uint8_t* signature, void* arg) {
(void)x25519; (void)signed_by; (void)signature;
struct chat_sync* cs = (struct chat_sync*)arg;
if (!cs || !cs->initialized) return;
struct join_pending* jp = cs_pending_find(cs, node_id, ch_id);
if (!jp || !jp->conn) return;
cs_send_join_ready(cs, ch_id, node_id);
}
const uint8_t* p = pl;
uint16_t pc; memcpy(&pc, p, 2); p += 2;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME ch=%s from=%016llx peers=%u", CS_ID, ch_id, (unsigned long long)peer, pc);
for (uint16_t i = 0; i < pc; i++) {
if ((size_t)(p - pl) + 8 + 32 + 32 + 1 + 64 + 8 + 1 + 1 + 64 > len) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME peer[%u] truncated at %td len=%zu", CS_ID, i, p - pl, len); break; } /* min: id+x25519+ed+flags+update_sig(64+8)+nl+atl+adm_tags_sig(64) */
uint64_t node_id; memcpy(&node_id, p, 8); p += 8;
const uint8_t* x25519 = p; p += 32;
const uint8_t* ed_pub = p; p += 32;
uint8_t flags = *p++;
const uint8_t* join_sig = NULL; uint64_t join_ts = 0;
if (flags & PEERS_FLAG_HAS_JOIN) {
if (p + 64 + 8 > pl + len) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME join_sig truncated node=0x%016llx i=%u", CS_ID, (unsigned long long)node_id, i); break; }
join_sig = p; p += 64;
memcpy(&join_ts, p, 8); p += 8;
}
const uint8_t* update_sig = p; p += 64;
uint64_t update_ts; memcpy(&update_ts, p, 8); p += 8;
(void)update_sig; (void)update_ts;
uint8_t nl = *p++;
char peer_userinfo[256] = "";
if (nl && p + nl <= pl + len) { memcpy(peer_userinfo, p, nl); peer_userinfo[nl] = '\0'; p += nl; }
uint8_t atl = *p++;
if (atl > 0 && p + atl <= pl + len) p += atl;
p += 64; /* adm_tags_sig */
/* ─── 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) {
/* 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);
uint64_t channel_id = strtoull(ch_id, NULL, 10);
const uint8_t* member_data = pl + 8; size_t member_len = len - 8;
/* verify join_sig if present */
if (join_sig) {
uint8_t ch_x2[32], ch_ed2[32];
if (topo_node_sqlite_channel_get(db, ch_id,
NULL, 0, NULL, ch_x2, ch_ed2, NULL) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME channel not in DB ch=%s", CS_ID, ch_id);
continue;
}
uint8_t vmsg[128];
int vlen = member_sync_build_join_msg(ch_x2, ch_ed2, node_id, x25519, join_ts,
vmsg, (int)sizeof(vmsg));
if (vlen < 0 || cs_ed25519_verify(ed_pub, vmsg, (size_t)vlen, join_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME invalid join_sig node=0x%016llx — skipping", CS_ID, (unsigned long long)node_id);
continue;
}
}
struct join_member_data m;
if (chat_join_parse_member(member_data, member_len, &m) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_REQUEST parse failed ch=%s from=%016llx", CS_ID, ch_id, (unsigned long long)peer);
return;
}
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_REQUEST ch=%s joiner=%016llx from=%016llx key=%016llx",
CS_ID, ch_id, (unsigned long long)m.node_id, (unsigned long long)peer, (unsigned long long)join_key);
int rc = topo_node_sqlite_member_block_put(db, ch_id, node_id, join_sig, join_ts, NULL, 0, x25519, ed_pub, peer_userinfo, NULL);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: member_put(welcome) FAILED ch=%s node=0x%016llx rc=%d", CS_ID, ch_id, (unsigned long long)node_id, rc);
{ char pn[256]; json_flat_get(peer_userinfo, "name", pn, sizeof(pn));
topo_node_sqlite_node_update_verified(db, node_id, pn, x25519, ed_pub, join_ts, ntp_time_get_seconds(cs->inst)); }
/* привязка идентичности: pubkey из ETCP-соединения == заявленный x25519 */
struct ETCP_CONN* jconn = cs_find_conn_for_node(cs->inst, peer);
if (!jconn || memcmp(jconn->crypto_ctx.peer_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;
}
member_sync_cancel(g_cs->inst, node_id, ch_id);
/* валидация ключа */
uint64_t inviter = chat_join_lookup_inviter(cs->inst, channel_id, join_key);
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(cs, ch_id, peer, err, 1);
return;
}
/* валидный ключ → сразу добавляем узел в группу для BGP sync */
{
uint64_t gid = strtoull(ch_id, NULL, 10);
struct TOPO_GROUP* g = topo_groups_find(cs->inst->topo_groups, gid);
if (g && g->group_type == TOPO_GROUP_TYPE_CHAT)
topo_group_new_conn(g, jconn);
}
/* добавить в pending (ждут JOIN_READY по приходу рекорда через merkle) */
cs_pending_add(cs, m.node_id, ch_id, jconn);
/* форвард инвайтеру (или локальная обработка, если инвайтер == мы) */
if (inviter == cs->inst->node_id) {
chat_join_process_request(cs->inst, channel_id, peer, join_key, member_data, member_len);
/* мы и есть connection-узел + инвайтер: рекорд добавлен локально,
merkle-apply-коллбэк (cs_on_member_applied) здесь не сработает —
отправляем JOIN_READY напрямую, если мембер реально добавлен. */
if (topo_node_sqlite_member_in_channel(cs->inst->topo_sqlite_db, ch_id, m.node_id))
cs_send_join_ready(cs, ch_id, m.node_id);
} else {
chat_join_forward_request(cs->inst, channel_id, inviter, join_key, &m);
}
cs_refresh_channels(cs);
cs_post_channel_online(cs, ch_id);
}
/* add self to local peers table (server added joiner to its DB; client must do the same) */
/* ─── JOIN_READY (0x13): connection-узел → joiner (можно запускать BGP) ─── */
static void cs_handle_join_ready(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
if (cs->join_timer) { uasync_cancel_timeout(cs->inst->ua, cs->join_timer); cs->join_timer = NULL; }
if (len < 8) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_READY too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
uint64_t group_id; memcpy(&group_id, pl, 8);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_READY ch=%s group=%016llx from=%016llx — joining complete, starting BGP",
CS_ID, ch_id, (unsigned long long)group_id, (unsigned long long)peer);
/* запуск BGP + полноценное участие в канале */
{
uint64_t myid = cs->inst->node_id;
const uint8_t* my_x25 = cs->inst->my_keys.public_key;
const uint8_t* my_ed = cs->inst->my_ed25519_pubkey;
const char* my_name = cs->inst->name[0] ? cs->inst->name : "";
int src = topo_node_sqlite_member_block_put(db, ch_id, myid, NULL, 0, NULL, 0, my_x25, my_ed, my_name, NULL);
if (src != 0)
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: member_put(self) FAILED in WELCOME ch=%s rc=%d", CS_ID, ch_id, src);
else
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME added self to peers ch=%s node=0x%016llx x25519=%016llx name=%s",
CS_ID, ch_id, (unsigned long long)myid, *(const uint64_t*)my_x25, my_name);
struct TOPO_GROUP* g = topo_groups_find(cs->inst->topo_groups, group_id);
if (g && g->group_type == TOPO_GROUP_TYPE_CHAT) {
struct ETCP_CONN* c = cs_find_conn_for_node(cs->inst, peer);
if (c) topo_group_new_conn(g, c);
}
}
chat_core_ensure_channel_ready(ch_id);
uint8_t evt[65]; uint8_t ch_id_len = (uint8_t)strlen(ch_id);
evt[0] = ch_id_len; memcpy(evt + 1, ch_id, ch_id_len);
chat_event_post(CHAT_EVT_CHANNEL_UPDATED, evt, 1 + ch_id_len);
/* full member list for the newly joined channel */
{ uint8_t* members = NULL; int mc = 0;
if (chat_core_get_member_list(ch_id, &members, &mc) == 0 && members) {
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME member_list count=%d ch=%s", CS_ID, mc, ch_id);
size_t esz = 1 + ch_id_len + 2 + (size_t)mc * CHAT_MEMBER_DISPLAY_SIZE;
uint8_t* mevt = u_malloc(esz);
if (mevt) {
uint16_t u16 = (uint16_t)mc;
mevt[0] = ch_id_len; memcpy(mevt + 1, ch_id, ch_id_len);
memcpy(mevt + 1 + ch_id_len, &u16, 2);
memcpy(mevt + 1 + ch_id_len + 2, members, (size_t)mc * CHAT_MEMBER_DISPLAY_SIZE);
chat_event_post(CHAT_EVT_MEMBER_LIST, mevt, (int)esz);
u_free(mevt);
}
u_free(members);
}
}
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME calling cs_post_channel_online ch=%s", CS_ID, ch_id);
cs_post_channel_online(cs, ch_id);
cs_resume_db_sync(ch_id);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME processed ch=%s peers=%d — message sync triggered via db_sync (active conn or peer_check timer every 5s)",
CS_ID, ch_id, pc);
if (cs->pending_invite_ch_id != 0) {
uint64_t ch_id_num = strtoull(ch_id, NULL, 10);
uint8_t cevt[20]; memcpy(cevt, &peer, 8); int r = 0; memcpy(cevt + 8, &r, 4); memcpy(cevt + 12, &ch_id_num, 8);
chat_event_post(CHAT_EVT_CONNECT_RESULT, cevt, 20);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME invite success ch=%s peer=%016llx — posted CHAT_EVT_CONNECT_RESULT",
CS_ID, ch_id, (unsigned long long)peer);
cs->pending_invite_node_id = 0; cs->pending_invite_ch_id = 0; if (cs->join_timer) { uasync_cancel_timeout(cs->inst->ua, cs->join_timer); cs->join_timer = NULL; }
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_READY invite success ch=%s peer=%016llx", CS_ID, ch_id, (unsigned long long)peer);
cs->pending_invite_node_id = 0; cs->pending_invite_ch_id = 0; cs->pending_join_key = 0;
}
}
/* ─── PEER_UPSERT (0x0D): propagate new peer ─── */
static void cs_handle_peer_upsert(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
if (len < 8 + 32 + 32 + 64 + 8 + 1) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: PEER_UPSERT too short len=%zu", CS_ID, len); return; }
const uint8_t* p = pl;
uint64_t node_id; memcpy(&node_id, p, 8); p += 8;
const uint8_t* x25519 = p; p += 32;
const uint8_t* ed_pub = p; p += 32;
const uint8_t* join_sig = p; p += 64;
uint64_t join_ts; memcpy(&join_ts, p, 8); p += 8;
uint8_t userinfo_len = *p++;
char peer_userinfo[128] = "";
if (p + userinfo_len <= pl + len) { memcpy(peer_userinfo, p, userinfo_len); peer_userinfo[userinfo_len] = '\0'; p += userinfo_len; }
/* verify join_sig */
uint8_t ch_x3[32], ch_ed3[32];
if (topo_node_sqlite_channel_get(cs->inst->topo_sqlite_db, ch_id,
NULL, 0, NULL, ch_x3, ch_ed3, NULL) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: PEER_UPSERT channel not in DB ch=%s", CS_ID, ch_id);
return;
}
uint8_t vmsg[128];
int vlen = member_sync_build_join_msg(ch_x3, ch_ed3, node_id, x25519, join_ts,
vmsg, (int)sizeof(vmsg));
if (vlen < 0 || cs_ed25519_verify(ed_pub, vmsg, (size_t)vlen, join_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: PEER_UPSERT invalid sig node=0x%016llx", CS_ID,
(unsigned long long)node_id);
return;
}
sqlite3* db = cs->inst->topo_sqlite_db;
int rc = topo_node_sqlite_member_block_put(db, ch_id, node_id, join_sig, join_ts, NULL, 0, x25519, ed_pub, peer_userinfo, NULL);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: member_put(peer_upsert) FAILED ch=%s node=0x%016llx rc=%d", CS_ID, ch_id, (unsigned long long)node_id, rc);
{ char pn2[128]; json_flat_get(peer_userinfo, "name", pn2, sizeof(pn2));
topo_node_sqlite_node_update_verified(db, node_id, pn2, x25519, ed_pub, join_ts, ntp_time_get_seconds(cs->inst)); }
cs_refresh_channels(cs);
{ uint8_t mevt[1 + 64 + 1 + CHAT_MEMBER_DISPLAY_SIZE]; uint8_t ml = (uint8_t)strlen(ch_id);
mevt[0] = ml; memcpy(mevt + 1, ch_id, ml); mevt[1 + ml] = 1;
if (chat_core_get_single_member(ch_id, node_id, mevt + 1 + ml + 1) == 0)
chat_event_post(CHAT_EVT_MEMBER_UPDATED, mevt, 1 + ml + 1 + CHAT_MEMBER_DISPLAY_SIZE); }
/* propagate to others (except sender and the subject node) */
cs_propagate(cs, ch_id, peer, pl, len);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: PEER_UPSERT node=0x%016llx ch=%s",
CS_ID, (unsigned long long)node_id, ch_id);
}
/* ─── CHANNEL_INVITE (0x0F): inviter → joiner (push-приглашение в канал) ─── */
static void cs_handle_channel_invite(struct chat_sync* cs, uint64_t peer,
+10 -10
View File
@@ -6,12 +6,12 @@
* - запускает member_sync (синхронизацию участников каналов) через merkle_sync
* - управляет кешем каналов (периодический refresh из БД, 30s)
*
* Обслуживает протокол входа в канал между joiner и inviter:
* CHANNEL_INFO_REQ → CHANNEL_INFO_RESP → CHANNEL_JOIN → WELCOME
* Все сообщения с Ed25519-верификацией.
* Обслуживает протокол входа в канал (join) между джойнером, connection-узлом и инвайтером:
* JOIN_INFO_REQ → JOIN_INFO_RESP → JOIN_REQUEST → (форвард инвайтеру) → JOIN_READY.
* Полное описание flow и дерева приглашений — в chat_join.h.
*
* Обрабатывает invite-ссылки (chat_sync_connect_from_invite):
* - сохраняет pubkey и адреса invite-узла в БД (nodes, node_addresses)
* - сохраняет pubkey, join_key и адреса invite-узла в БД (nodes, node_addresses)
* - создаёт TOPO_GROUP, запускает conn_mgr для подключения
* - после поднятия ETCP-соединения запускает join-протокол
*
@@ -47,12 +47,11 @@ struct UASYNC;
#define CS_MSG_ACK_PUSH 0x05
#define CS_MSG_SYNC_DONE 0x06
#define CS_MSG_ERROR 0x07
#define CS_MSG_CHANNEL_INFO_REQ 0x09
#define CS_MSG_CHANNEL_INFO_RESP 0x0A
#define CS_MSG_CHANNEL_JOIN 0x0B
#define CS_MSG_WELCOME 0x0C
#define CS_MSG_PEER_UPSERT 0x0D
#define CS_MSG_CHANNEL_INVITE 0x0F /* push-invite: inviter → joiner */
#define CS_MSG_JOIN_INFO_REQ 0x10 /* joiner → connection: {join_key:8} */
#define CS_MSG_JOIN_INFO_RESP 0x11 /* connection → joiner: channel info + connection's member record */
#define CS_MSG_JOIN_REQUEST 0x12 /* joiner → connection: {join_key:8, member...} */
#define CS_MSG_JOIN_READY 0x13 /* connection → joiner: {group_id:8} */
/* Protocol constants */
#define CS_SEND_DATA_MAX 32
@@ -73,12 +72,13 @@ void chat_sync_destroy(struct UTUN_INSTANCE* inst);
void chat_sync_connect_node(struct UTUN_INSTANCE* inst, uint64_t node_id);
/* Подключиться к пиру по данным invite-ссылки (вызывается из uasync-потока).
join_key — 64-битный ключ входа из ссылки (регистрируется инвайтером на connection-узле).
password может быть NULL — тогда отправляется без пароля (совместимость с v1) */
void chat_sync_connect_from_invite(uint64_t channel_id, uint64_t node_id,
const uint8_t* pubkey_bin,
const uint8_t* addrs_data, int addr_count,
int addrs_data_len,
const char* password);
const char* password, uint64_t join_key);
/* Пригласить target_node_id в канал ch_id с адресами из invite-ссылки (inviter → joiner).
Принимает pubkey и адреса целевого узла. Если соединения ещё нет — строит TOPO_NODE
+11
View File
@@ -6,6 +6,7 @@
#include "../utun_instance.h"
#include "../routing_layer/topo_group.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "member_sync.h"
#include "../transport_layer/etcp_dump.h"
#include "../transport_layer/secure_channel.h"
#include "../../lib/debug_config.h"
@@ -1255,6 +1256,16 @@ static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
queue_dgram_free(entry); queue_entry_free(entry); return;
}
/* валидация клиента: pubkey из ETCP-соединения должен быть в таблице мемберов канала */
{
char ch_id[32]; snprintf(ch_id, sizeof(ch_id), "%llu", (unsigned long long)group_id);
if (conn && !member_sync_validate_pubkey(conn->instance, ch_id, conn->crypto_ctx.peer_public_key)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "sync: recv DROP — peer=0x%016llx not member of ch=%s (type=%02x)",
(unsigned long long)src, ch_id, type);
queue_dgram_free(entry); queue_entry_free(entry); return;
}
}
switch (type) {
case DB_MSG_INIT_SYNC: DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "→ handle INIT_SYNC"); db_handle_init_sync(si, src, payload, plen); break;
case DB_MSG_INIT_RESP: DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "→ handle INIT_RESP"); db_handle_init_resp(si, src, payload, plen); break;
+11
View File
@@ -9,6 +9,7 @@
#include "invite_link.h"
#include "chat_core_priv.h"
#include "chat_event.h"
#include "chat_join.h"
#include "../routing_layer/topo_group.h"
#include "../routing_layer/topo_node.h"
@@ -20,6 +21,7 @@
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <openssl/rand.h>
/* ── живость узла в канале (BGP/прямое/через посредника) ── */
@@ -268,6 +270,11 @@ int chat_invite_build_link(uint64_t channel_id, uint64_t target_node_id,
memset(&data, 0, sizeof(data));
data.channelId = channel_id;
data.nodeId = target_nid;
/* join-ключ: генерируется инвайтером, кладётся в ссылку и регистрируется на connection-узле */
if (RAND_bytes((uint8_t*)&data.join_key, sizeof(data.join_key)) != 1 || data.join_key == 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "invite_build: join_key generation failed");
return -1;
}
if (get_node_pubkey(target_nid, use_self, data.pubkey) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "invite_build: no pubkey for node=0x%016llx",
(unsigned long long)target_nid);
@@ -296,6 +303,10 @@ int chat_invite_build_link(uint64_t channel_id, uint64_t target_node_id,
(unsigned long long)channel_id);
return -1;
}
/* ── регистрация join-ключа на connection-узле (или локально, если target==self) ── */
chat_join_register_key(g_cc.inst, channel_id, target_nid, data.join_key);
return 0;
}
+7
View File
@@ -83,6 +83,12 @@ int invite_link_decode(const char* link, size_t link_len, struct InviteData* out
uint64_t chId = 0;
for (int i = 0; i < 8; i++) chId = (chId << 8) | raw[off++];
out->channelId = chId;
if (off + 8 > raw_len) { if (error_buf && error_buf_size) snprintf(error_buf, error_buf_size, "truncated at join_key"); return -1; }
uint64_t jk = 0;
for (int i = 0; i < 8; i++) jk = (jk << 8) | raw[off++];
out->join_key = jk;
memset(out->pubkey, 0, INVITE_PUBKEY_SIZE);
out->addrCount = 0;
@@ -137,6 +143,7 @@ int invite_link_encode(const struct InviteData* data, const char* password, char
raw[pos++] = (uint8_t)plen; memcpy(raw + pos, password, plen); pos += plen;
}
for (int i = 7; i >= 0; i--) raw[pos++] = (uint8_t)((data->channelId >> (i * 8)) & 0xFF);
for (int i = 7; i >= 0; i--) raw[pos++] = (uint8_t)((data->join_key >> (i * 8)) & 0xFF);
uint8_t pk[32]; memcpy(pk, data->pubkey, 32);
int ia = 0;
+1
View File
@@ -32,6 +32,7 @@ struct InviteAddr {
struct InviteData {
uint64_t channelId;
uint64_t join_key; /* 64-битный ключ входа (генерируется инвайтером, хранится на connection-узле 10 мин) */
uint8_t pubkey[INVITE_PUBKEY_SIZE];
uint64_t nodeId;
uint8_t addrCount;
+168 -14
View File
@@ -41,6 +41,7 @@ static void _compute_member_hash(uint64_t node_id, const uint8_t* x25519,
const uint8_t* join_sig, uint64_t join_ts,
const uint8_t* update_sig, uint64_t update_ts,
const char* adm_tags, const uint8_t* adm_tags_sig,
uint64_t signed_by, const uint8_t* signature,
uint8_t hash_out[MT_HASH_SIZE]) {
EVP_MD_CTX* ctx = EVP_MD_CTX_new();
EVP_DigestInit_ex(ctx, EVP_sha256(), NULL);
@@ -57,6 +58,9 @@ static void _compute_member_hash(uint64_t node_id, const uint8_t* x25519,
if (atl) EVP_DigestUpdate(ctx, adm_tags, atl);
if (adm_tags_sig) EVP_DigestUpdate(ctx, adm_tags_sig, 64); else { static const uint8_t z64[64]; EVP_DigestUpdate(ctx, z64, 64); }
EVP_DigestUpdate(ctx, &signed_by, 8);
if (signature) EVP_DigestUpdate(ctx, signature, 64); else { static const uint8_t zsig[64]; EVP_DigestUpdate(ctx, zsig, 64); }
EVP_DigestFinal_ex(ctx, hash_out, NULL);
EVP_MD_CTX_free(ctx);
}
@@ -119,7 +123,7 @@ static int _member_update_bucket_hash(void* ctx, const char* ns, uint8_t level,
char sql[512]; snprintf(sql, sizeof(sql),
"SELECT node_id, x25519_pubkey, ed25519_pubkey,"
" join_sig, join_ts, update_sig, update_ts, adm_tags, adm_tags_sig, source"
" join_sig, join_ts, update_sig, update_ts, adm_tags, adm_tags_sig, signed_by, signature, source"
" FROM \"%s\""
" WHERE (node_id & %lld) == %lld ORDER BY node_id ASC",
peers_tbl, (long long)mask, (long long)prefix64);
@@ -138,11 +142,13 @@ static int _member_update_bucket_hash(void* ctx, const char* ns, uint8_t level,
uint64_t uts = (uint64_t)sqlite3_column_int64(stmt, 6);
const char* atags = (const char*)sqlite3_column_text(stmt, 7);
const uint8_t* atsig = (const uint8_t*)sqlite3_column_blob(stmt, 8);
int src = sqlite3_column_int(stmt, 9);
uint64_t sb = (uint64_t)sqlite3_column_int64(stmt, 9);
const uint8_t* sgn = (const uint8_t*)sqlite3_column_blob(stmt, 10);
int src = sqlite3_column_int(stmt, 11);
if (!x25 || !ed || src != 0) continue;
uint8_t mh[MT_HASH_SIZE];
_compute_member_hash(nid, x25, ed, sig, jts, usig, uts, atags, atsig, mh);
_compute_member_hash(nid, x25, ed, sig, jts, usig, uts, atags, atsig, sb, sgn, mh);
EVP_DigestUpdate(sha_ctx, mh, MT_HASH_SIZE);
count++;
}
@@ -164,7 +170,8 @@ static int _member_get_items(void* ctx, const char* ns, uint8_t level,
char sql[512]; snprintf(sql, sizeof(sql),
"SELECT node_id, x25519_pubkey, ed25519_pubkey,"
" join_sig, join_ts, update_sig, update_ts, userinfo, adm_tags, adm_tags_sig, source"
" join_sig, join_ts, update_sig, update_ts, userinfo, adm_tags, adm_tags_sig,"
" signed_by, signature, source"
" FROM \"%s\""
" WHERE (node_id & %lld) == %lld ORDER BY node_id ASC",
peers_tbl, (long long)mask, (long long)prefix);
@@ -187,13 +194,15 @@ static int _member_get_items(void* ctx, const char* ns, uint8_t level,
const char* nm = (const char*)sqlite3_column_text(stmt, 7);
const char* atags = (const char*)sqlite3_column_text(stmt, 8);
const uint8_t* atsig = (const uint8_t*)sqlite3_column_blob(stmt, 9);
int src = sqlite3_column_int(stmt, 10);
uint64_t sb = (uint64_t)sqlite3_column_int64(stmt, 10);
const uint8_t* sgn = (const uint8_t*)sqlite3_column_blob(stmt, 11);
int src = sqlite3_column_int(stmt, 12);
if (!x25 || !ed || src != 0) continue;
uint8_t nl = nm ? (uint8_t)strnlen(nm, 255) : 0;
uint8_t atl = atags ? (uint8_t)strnlen(atags, 255) : 0;
uint8_t flags = (sig && jts) ? PEERS_FLAG_HAS_JOIN : 0;
size_t need = 8 + 32 + 32 + 1 + (flags ? 72ULL : 0ULL) + 64 + 8 + 1 + (size_t)nl + 1 + (size_t)atl + 64;
size_t need = 8 + 32 + 32 + 1 + (flags ? 72ULL : 0ULL) + 64 + 8 + 1 + (size_t)nl + 1 + (size_t)atl + 64 + 8 + 64;
if (off + need > *len) { sqlite3_finalize(stmt); return -2; }
memcpy(buf + off, &nid, 8); off += 8;
memcpy(buf + off, x25, 32); off += 32;
@@ -216,6 +225,9 @@ static int _member_get_items(void* ctx, const char* ns, uint8_t level,
if (atl) { memcpy(buf + off, atags, atl); off += atl; }
if (atsig && sqlite3_column_bytes(stmt, 9) >= 64) { memcpy(buf + off, atsig, 64); off += 64; }
else { memset(buf + off, 0, 64); off += 64; }
memcpy(buf + off, &sb, 8); off += 8;
if (sgn && sqlite3_column_bytes(stmt, 11) >= 64) { memcpy(buf + off, sgn, 64); off += 64; }
else { memset(buf + off, 0, 64); off += 64; }
(*cnt)++;
}
sqlite3_finalize(stmt);
@@ -257,6 +269,40 @@ void member_sync_remove_props_cbk(node_props_changed_fn fn, void* arg) {
}
}
/* ── Коллбэк применения рекорда мембера (ready-хендлинг join на connection-узле) ── */
struct ms_apply_cbk { member_apply_cbk_fn fn; void* arg; struct ms_apply_cbk* next; };
static struct ms_apply_cbk* g_apply_cbks = NULL;
static void _fire_apply_cbk(const char* ch_id, uint64_t node_id,
const uint8_t* x25519, uint64_t signed_by,
const uint8_t* signature) {
struct ms_apply_cbk* ac = g_apply_cbks;
while (ac) { ac->fn(ch_id, node_id, x25519, signed_by, signature, ac->arg); ac = ac->next; }
}
void member_sync_add_apply_cbk(member_apply_cbk_fn fn, void* arg) {
if (!fn) return;
struct ms_apply_cbk* e = u_malloc(sizeof(*e));
if (!e) return;
e->fn = fn; e->arg = arg;
e->next = g_apply_cbks;
g_apply_cbks = e;
}
void member_sync_remove_apply_cbk(member_apply_cbk_fn fn, void* arg) {
if (!fn) return;
struct ms_apply_cbk** p = &g_apply_cbks;
while (*p) {
if ((*p)->fn == fn && (*p)->arg == arg) {
struct ms_apply_cbk* r = *p;
*p = r->next; u_free(r);
return;
}
p = &(*p)->next;
}
}
/* ── BGP node event callback: BGP сообщает member_sync об обновлении адресов/классификации.
Классификацию node_type/storage пересчитывает сам BGP (nodeinfo_updated);
member_sync только рассылает node_props_changed потребителям. ── */
@@ -316,6 +362,60 @@ static int _verify_join_sig(sqlite3* db, const char* ch_id, const struct ms_memb
return ok ? 1 : 0;
}
/* ── Дерево приглашений: подпись pubkey узла ── */
int member_sync_build_sign_msg(const uint8_t* x25519, uint8_t* out, int out_sz) {
if (!x25519 || !out || out_sz < MEMBER_SIGN_MSG_SIZE) return -1;
memcpy(out, x25519, SC_PUBKEY_SIZE);
memcpy(out + SC_PUBKEY_SIZE, MEMBER_SIGN_SUFFIX, MEMBER_SIGN_SUFFIX_LEN);
return MEMBER_SIGN_MSG_SIZE;
}
int member_sync_sign_pubkey(const uint8_t* signer_ed_priv, const uint8_t* x25519, uint8_t sig[64]) {
if (!signer_ed_priv || !x25519 || !sig) return -1;
uint8_t msg[MEMBER_SIGN_MSG_SIZE];
if (member_sync_build_sign_msg(x25519, msg, (int)sizeof(msg)) < 0) return -1;
if (sc_ed25519_sign(signer_ed_priv, msg, sizeof(msg), sig) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: sign_pubkey — sc_ed25519_sign FAILED", MS_ID);
return -1;
}
return 0;
}
int member_sync_verify_signature(sqlite3* db, const char* ch_id, const struct ms_member_rec* m) {
if (!db || !ch_id || !m || !m->x25519 || !m->signature || _sig_is_zero64(m->signature)) return 0;
uint8_t msg[MEMBER_SIGN_MSG_SIZE];
if (member_sync_build_sign_msg(m->x25519, msg, (int)sizeof(msg)) < 0) return -1;
uint8_t signer_pub[32];
if (m->signed_by == 0) {
if (topo_node_sqlite_channel_get(db, ch_id, NULL, 0, NULL, NULL, signer_pub, NULL) != 0) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: verify_signature — no channel key ns=%s (defer)", MS_ID, ch_id);
return -1;
}
} else {
if (topo_node_sqlite_get_ed25519_pubkey(db, m->signed_by, signer_pub) != 0) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: verify_signature — no ed25519 pubkey for signed_by=0x%016llx ns=%s (defer)", MS_ID, (unsigned long long)m->signed_by, ch_id);
return -1;
}
}
int ok = sc_ed25519_verify(signer_pub, msg, sizeof(msg), m->signature) == SC_OK;
if (!ok)
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: verify_signature FAIL nid=0x%016llx signed_by=0x%016llx ns=%s",
MS_ID, (unsigned long long)m->node_id, (unsigned long long)m->signed_by, ch_id);
return ok ? 1 : 0;
}
int member_sync_validate_pubkey(struct UTUN_INSTANCE* inst, const char* ch_id,
const uint8_t* x25519_pubkey) {
if (!inst || !ch_id || !x25519_pubkey) return 0;
sqlite3* db = _db(inst); if (!db) return 0;
uint64_t node_id = sc_derive_node_id_from_pubkey(x25519_pubkey);
if (node_id == 0) return 0;
return topo_node_sqlite_member_in_channel(db, ch_id, node_id);
}
int member_sync_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
uint64_t from_peer, const struct ms_member_rec* m) {
if (!inst || !ch_id || !m || !m->x25519 || !m->ed25519) {
@@ -336,10 +436,12 @@ int member_sync_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
uint8_t local_join_sig[64] = {0}; uint64_t local_join_ts = 0;
uint64_t local_update_ts = 0;
char local_tags[256] = "";
uint64_t local_signed_by = 0;
uint8_t local_signature[64] = {0};
{
sqlite3_stmt* st = NULL;
char sql[512]; snprintf(sql, sizeof(sql),
"SELECT join_sig, join_ts, update_ts, adm_tags FROM \"%s\" WHERE node_id=?", peers_tbl);
"SELECT join_sig, join_ts, update_ts, adm_tags, signed_by, signature FROM \"%s\" WHERE node_id=?", peers_tbl);
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_int64(st, 1, (sqlite3_int64)m->node_id);
if (sqlite3_step(st) == SQLITE_ROW) {
@@ -350,6 +452,9 @@ int member_sync_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
local_update_ts = (uint64_t)sqlite3_column_int64(st, 2);
const char* tags = (const char*)sqlite3_column_text(st, 3);
if (tags) snprintf(local_tags, sizeof(local_tags), "%s", tags);
local_signed_by = (uint64_t)sqlite3_column_int64(st, 4);
const uint8_t* ls = (const uint8_t*)sqlite3_column_blob(st, 5);
if (ls && sqlite3_column_bytes(st, 5) >= 64) memcpy(local_signature, ls, 64);
}
sqlite3_finalize(st);
}
@@ -371,6 +476,15 @@ int member_sync_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
identity_ok = 0;
}
}
/* подпись дерева приглашений: -1 = родитель ещё не пришёл (defer до verify_local), 0 = явная подделка */
if (identity_ok && m->signature && !_sig_is_zero64(m->signature)) {
int sv = member_sync_verify_signature(db, ch_id, m);
if (sv == 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record invalid tree-signature (forgery) nid=0x%016llx signed_by=0x%016llx ns=%s",
MS_ID, (unsigned long long)m->node_id, (unsigned long long)m->signed_by, ch_id);
identity_ok = 0;
}
}
}
int block_a_ok = 0;
@@ -462,9 +576,12 @@ int member_sync_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
/* ── запись изменённых блоков ── */
if (block_a_changed) {
const uint8_t* jsig = m->join_sig ? m->join_sig : (has_local ? local_join_sig : NULL);
/* дерево приглашений: если в обновлении подписи нет — сохраняем локальную (set один раз при join/create) */
uint64_t w_signed_by = m->signature ? m->signed_by : local_signed_by;
const uint8_t* w_signature = m->signature ? m->signature : (has_local ? local_signature : NULL);
if (topo_node_sqlite_member_block_put(db, ch_id, m->node_id, jsig, m->join_ts,
m->update_sig, m->update_ts, m->x25519, m->ed25519,
m->userinfo ? m->userinfo : "", NULL) != 0) {
m->userinfo ? m->userinfo : "", NULL, w_signed_by, w_signature) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record block_put FAILED ch=%s nid=0x%016llx", MS_ID, ch_id, (unsigned long long)m->node_id);
return -1;
}
@@ -572,6 +689,17 @@ int member_sync_verify_local_record(struct sqlite3* db, const char* ch_id, const
return 0;
}
}
/* дерево приглашений: подпись есть и невалидна (подписант резолвится) — битая запись.
-1 = родитель ещё не пришёл — не трогаем (не битая). */
if (m->signature && !_sig_is_zero64(m->signature)) {
int sv = member_sync_verify_signature(db, ch_id, m);
if (sv == 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: verify_local — invalid tree-signature nid=0x%016llx signed_by=0x%016llx ns=%s",
MS_ID, (unsigned long long)m->node_id, (unsigned long long)m->signed_by, ch_id);
return 0;
}
}
return 1;
}
@@ -584,7 +712,8 @@ int member_sync_verify_and_purge(struct UTUN_INSTANCE* inst, const char* ch_id)
sqlite3_stmt* st = NULL;
char sql[512]; snprintf(sql, sizeof(sql),
"SELECT node_id, x25519_pubkey, ed25519_pubkey,"
" join_sig, join_ts, update_sig, update_ts, userinfo, adm_tags, adm_tags_sig, source"
" join_sig, join_ts, update_sig, update_ts, userinfo, adm_tags, adm_tags_sig,"
" signed_by, signature, source"
" FROM \"%s\" ORDER BY node_id ASC", peers_tbl);
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) return 0;
@@ -600,7 +729,9 @@ int member_sync_verify_and_purge(struct UTUN_INSTANCE* inst, const char* ch_id)
const char* nm = (const char*)sqlite3_column_text(st, 7);
const char* atags = (const char*)sqlite3_column_text(st, 8);
const uint8_t* atsig = (const uint8_t*)sqlite3_column_blob(st, 9);
int src = sqlite3_column_int(st, 10);
uint64_t sb = (uint64_t)sqlite3_column_int64(st, 10);
const uint8_t* sgn = (const uint8_t*)sqlite3_column_blob(st, 11);
int src = sqlite3_column_int(st, 12);
if (!x25 || !ed) continue;
if (src != 0) continue; /* плейсхолдер topo_group — не верифицируем */
@@ -612,6 +743,7 @@ int member_sync_verify_and_purge(struct UTUN_INSTANCE* inst, const char* ch_id)
m.update_sig = usig; m.update_ts = uts;
m.userinfo = nm;
m.adm_tags = atags; m.adm_tags_sig = atsig;
m.signed_by = sb; m.signature = sgn;
if (!member_sync_verify_local_record(db, ch_id, &m)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: verify_and_purge — PURGE broken record nid=0x%016llx ns=%s",
@@ -688,6 +820,8 @@ static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer,
const uint8_t* atsig = NULL;
if (atl && mrem >= atl) { memcpy(atags, mp, atl); atags[atl] = '\0'; mp += atl; mrem -= atl; }
if (mrem >= 64) { atsig = mp; mp += 64; mrem -= 64; }
uint64_t sb = 0; const uint8_t* sgn = NULL;
if (mrem >= 72) { memcpy(&sb, mp, 8); mp += 8; mrem -= 8; sgn = mp; mp += 64; mrem -= 64; }
struct ms_member_rec m;
memset(&m, 0, sizeof(m));
@@ -698,6 +832,7 @@ static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer,
m.userinfo = nm;
m.adm_tags = atags[0] ? atags : NULL;
m.adm_tags_sig = atsig;
m.signed_by = sb; m.signature = sgn;
int r = member_sync_apply_record(inst, ns, from_peer, &m);
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: member_sync recv ns=%s nid=0x%016llx r=%d", MS_ID, ns, (unsigned long long)nid, r);
@@ -710,6 +845,8 @@ static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer,
const char* tags_final = atags[0] ? atags : NULL;
while (pc) { pc->fn(nid, tags_final, ns, pc->arg); pc = pc->next; }
}
/* fire apply callbacks (ready-хендлинг join на connection-узле) */
_fire_apply_cbk(ns, nid, x25, sb, sgn);
}
if (r & MS_APPLY_STALE) {
/* у нас версия новее — отправить наш полный рекорд автору */
@@ -721,11 +858,19 @@ static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer,
return 0;
}
static int _member_validate_peer(void* ctx, const char* ns, uint64_t peer) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
sqlite3* db = _db(inst);
if (!db) return 0;
return topo_node_sqlite_member_in_channel(db, ns, peer);
}
static const struct merkle_sync_data_ops g_member_ops = {
.update_bucket_hash = _member_update_bucket_hash,
.get_items = _member_get_items,
.apply_items = _member_apply_items,
.apply_update = NULL, /* online-статус больше не рассылается через merkle */
.validate_peer = _member_validate_peer,
};
/* ── Public API ── */
@@ -828,7 +973,8 @@ static int _serialize_member(struct UTUN_INSTANCE* inst, const char* ch_id, uint
sqlite3_stmt* stmt = NULL;
char sql[512]; snprintf(sql, sizeof(sql),
"SELECT node_id, x25519_pubkey, ed25519_pubkey,"
" join_sig, join_ts, update_sig, update_ts, userinfo, adm_tags, adm_tags_sig"
" join_sig, join_ts, update_sig, update_ts, userinfo, adm_tags, adm_tags_sig,"
" signed_by, signature"
" FROM \"%s\" WHERE node_id=?", peers_tbl);
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: serialize_member — query failed ch=%s nid=0x%016llx", MS_ID, ch_id, (unsigned long long)member_id);
@@ -847,13 +993,15 @@ static int _serialize_member(struct UTUN_INSTANCE* inst, const char* ch_id, uint
const char* nm = (const char*)sqlite3_column_text(stmt, 7);
const char* atags = (const char*)sqlite3_column_text(stmt, 8);
const uint8_t* atsig = (const uint8_t*)sqlite3_column_blob(stmt, 9);
uint64_t sb = (uint64_t)sqlite3_column_int64(stmt, 10);
const uint8_t* sgn = (const uint8_t*)sqlite3_column_blob(stmt, 11);
if (!x25 || !ed) { sqlite3_finalize(stmt); return -1; }
uint8_t nl = nm ? (uint8_t)strnlen(nm, 255) : 0;
uint8_t atl = atags ? (uint8_t)strnlen(atags, 255) : 0;
uint8_t flags = (sig && jts) ? PEERS_FLAG_HAS_JOIN : 0;
size_t need = 2 + 8 + 32 + 32 + 1 + (flags & PEERS_FLAG_HAS_JOIN ? 72 : 0)
+ 72 + 1 + (size_t)nl + 1 + (size_t)atl + 64;
+ 72 + 1 + (size_t)nl + 1 + (size_t)atl + 64 + 8 + 64;
if (need > buf_sz) { sqlite3_finalize(stmt); return -1; }
size_t off = 0;
uint16_t wcnt = 1; memcpy(buf, &wcnt, 2); off += 2;
@@ -869,6 +1017,10 @@ static int _serialize_member(struct UTUN_INSTANCE* inst, const char* ch_id, uint
if (atsig && sqlite3_column_bytes(stmt, 9) >= 64) { memcpy(buf + off, atsig, 64); }
else { memset(buf + off, 0, 64); }
off += 64;
memcpy(buf + off, &sb, 8); off += 8;
if (sgn && sqlite3_column_bytes(stmt, 11) >= 64) { memcpy(buf + off, sgn, 64); }
else { memset(buf + off, 0, 64); }
off += 64;
sqlite3_finalize(stmt);
*out_len = off;
@@ -896,9 +1048,10 @@ int member_sync_put(struct UTUN_INSTANCE* inst, const char* ch_id,
const uint8_t* join_sig, uint64_t join_ts,
const uint8_t* update_sig, uint64_t update_ts,
const char* userinfo,
const char* adm_tags, const uint8_t* adm_tags_sig, int storage) {
const char* adm_tags, const uint8_t* adm_tags_sig, int storage,
uint64_t signed_by, const uint8_t* signature) {
if (!inst || !ch_id) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: put — inst=%p ch_id=%s", MS_ID, (void*)inst, ch_id ? ch_id : "(null)"); return -1; }
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: put ch=%s nid=%016llx userinfo=%s adm_tags=%s storage=%d", MS_ID, ch_id, (unsigned long long)member_id, userinfo ? userinfo : "", adm_tags ? adm_tags : "", storage);
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: put ch=%s nid=%016llx userinfo=%s adm_tags=%s storage=%d signed_by=%016llx", MS_ID, ch_id, (unsigned long long)member_id, userinfo ? userinfo : "", adm_tags ? adm_tags : "", storage, (unsigned long long)signed_by);
struct ms_member_rec m;
memset(&m, 0, sizeof(m));
@@ -908,6 +1061,7 @@ int member_sync_put(struct UTUN_INSTANCE* inst, const char* ch_id,
m.update_sig = update_sig; m.update_ts = update_ts;
m.userinfo = userinfo;
m.adm_tags = adm_tags; m.adm_tags_sig = adm_tags_sig; m.storage = storage;
m.signed_by = signed_by; m.signature = signature;
return member_sync_apply_record(inst, ch_id, inst->node_id, &m);
}
+35 -1
View File
@@ -68,8 +68,41 @@ struct ms_member_rec {
const char* adm_tags; /* JSON владельца с "ver", может быть NULL */
const uint8_t* adm_tags_sig; /* 64 байта, может быть NULL */
int storage; /* 0/1, производная от adm_tags.storage */
uint64_t signed_by; /* дерево приглашений: 0 = корень (подписан ключом группы), иначе node_id инвайтера */
const uint8_t* signature; /* 64 байта, Ed25519 подпись x25519||"group_join" ключом родителя (или канальным ed25519) */
};
/* ── Дерево приглашений (подпись pubkey узла) ── */
#define MEMBER_SIGN_SUFFIX "group_join"
#define MEMBER_SIGN_SUFFIX_LEN 10
#define MEMBER_SIGN_MSG_SIZE (SC_PUBKEY_SIZE + MEMBER_SIGN_SUFFIX_LEN) /* 42 */
/* Собрать сообщение подписи: x25519(32) || "group_join". Возвращает 42 или -1. */
int member_sync_build_sign_msg(const uint8_t* x25519, uint8_t* out, int out_sz);
/* Подписать pubkey узла: sign(signer_ed_priv, x25519||"group_join"). Возвращает 0/-1. */
int member_sync_sign_pubkey(const uint8_t* signer_ed_priv, const uint8_t* x25519, uint8_t sig[64]);
/* Верифицировать подпись дерева мембера:
* signed_by==0 → ключ канала (ch_ed25519), иначе ed25519 pubkey узла signed_by.
* Возвращает 1 если подпись валидна, 0 если невалидна, -1 если подписанта нет в БД
* (родитель ещё не пришёл по merkle — верификацию откладываем). */
int member_sync_verify_signature(struct sqlite3* db, const char* ch_id, const struct ms_member_rec* m);
/* Проверить, что x25519 pubkey (из ETCP-соединения) есть в таблице мемберов канала.
* node_id = derive(x25519) → SELECT 1 FROM peers_<ch_id>. Возвращает 1/0. */
int member_sync_validate_pubkey(struct UTUN_INSTANCE* inst, const char* ch_id,
const uint8_t* x25519_pubkey);
/* ── Коллбэк приёма/применения рекорда мембера (ready-хендлинг join) ── */
typedef void (*member_apply_cbk_fn)(const char* ch_id, uint64_t node_id,
const uint8_t* x25519, uint64_t signed_by,
const uint8_t* signature, void* arg);
void member_sync_add_apply_cbk(member_apply_cbk_fn fn, void* arg);
void member_sync_remove_apply_cbk(member_apply_cbk_fn fn, void* arg);
/*
* Инициализировать модуль: создаёт merkle_sync с коллбэками для мемберов
* (ETCP сервис 0x31), регистрирует _on_node_updated для пересчёта дерева
@@ -129,7 +162,8 @@ int member_sync_put(struct UTUN_INSTANCE* inst, const char* ch_id,
const uint8_t* join_sig, uint64_t join_ts,
const uint8_t* update_sig, uint64_t update_ts,
const char* userinfo,
const char* adm_tags, const uint8_t* adm_tags_sig, int storage);
const char* adm_tags, const uint8_t* adm_tags_sig, int storage,
uint64_t signed_by, const uint8_t* signature);
/*
* Строит сообщение для update_sig блока A (мембер):
+7
View File
@@ -698,6 +698,13 @@ static void _recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
char ns[64];
if (ms_group_id_to_ns(inst, group_id, ns, sizeof(ns)) != 0) { u_free(entry->dgram); queue_entry_free(entry); return; }
uint8_t type = d[9];
/* валидация пира: только авторизованные узлы (мемберы канала) могут синхронизироваться */
if (ms->ops->validate_peer && !ms->ops->validate_peer(ms->data_ctx, ns, peer)) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: recv DROP — peer=0x%016llx not authorized for ns=%s type=%02x",
MS_ID, (unsigned long long)peer, ns, type);
u_free(entry->dgram); queue_entry_free(entry); return;
}
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: recv type=%02x from=%016llx group=%016llx ns=%s len=%zu", MS_ID, type, (unsigned long long)peer, (unsigned long long)group_id, ns, dlen);
const uint8_t* pl = d + 10;
size_t plen = dlen - 10;
+8
View File
@@ -154,6 +154,14 @@ struct merkle_sync_data_ops {
*/
int (*apply_update)(void* ctx, const char* ns, uint64_t key,
uint8_t type, const uint8_t* data, size_t len);
/*
* Проверить, что пир авторизован для namespace ns (напр. его pubkey есть
* в таблице мемберов канала). Вызывается в _recv_cb для КАЖДОГО входящего
* пакета до диспатча. Возвращает 1 если разрешено, 0 если нет.
* Может быть NULL — тогда проверка пропускается (generic-потребитель).
*/
int (*validate_peer)(void* ctx, const char* ns, uint64_t peer);
};
/*
+44 -4
View File
@@ -351,6 +351,8 @@ int topo_node_sqlite_channel_put(sqlite3* db, const char* channel_id,
" update_ts INTEGER DEFAULT 0," /* NTP timestamp подписи update_sig */
" userinfo TEXT NOT NULL DEFAULT ''," /* JSON с информацией о пользователе: {"name":"<display_name>"} */
" creator_sig BLOB," /* Ed25519 подпись создателя (PEERS_JOIN_SIG_SIZE=64), пока заглушка */
" signed_by INTEGER," /* node_id инвайтера (родитель в дереве приглашений); 0 = корень, подписан ключом группы */
" signature BLOB," /* Ed25519 подпись x25519_pubkey||"group_join" ключом родителя (или канальным ed25519 для корня) */
" adm_tags TEXT," /* JSON-теги администратора: {"key":"val",...,"ver":"N"}.
подпись: sc_ed25519_sign(ch_ed_priv, json||node_id_8LE) */
" adm_tags_sig BLOB," /* Ed25519 подпись над adm_tags (64 байта) канальным ключом */
@@ -480,14 +482,15 @@ int topo_node_sqlite_member_block_put(sqlite3* db, const char* channel_id, uint6
const uint8_t* join_sig, uint64_t join_ts,
const uint8_t* update_sig, uint64_t update_ts,
const uint8_t* x25519_pubkey, const uint8_t* ed25519_pubkey,
const char* userinfo, const uint8_t* creator_sig) {
const char* userinfo, const uint8_t* creator_sig,
uint64_t signed_by, const uint8_t* signature) {
if (!db || !channel_id || !x25519_pubkey || !ed25519_pubkey) return -1;
char peers_tbl[80]; peers_table_name(channel_id, peers_tbl, sizeof(peers_tbl));
char sql[640];
char sql[768];
snprintf(sql, sizeof(sql),
"INSERT INTO \"%s\"(node_id, x25519_pubkey, ed25519_pubkey,"
" join_sig, join_ts, update_sig, update_ts, userinfo, creator_sig)"
" VALUES(?,?,?,?,?,?,?,?,?)"
" join_sig, join_ts, update_sig, update_ts, userinfo, creator_sig, signed_by, signature)"
" VALUES(?,?,?,?,?,?,?,?,?,?,?)"
" ON CONFLICT(node_id) DO UPDATE SET"
" x25519_pubkey=excluded.x25519_pubkey,"
" ed25519_pubkey=excluded.ed25519_pubkey,"
@@ -497,6 +500,8 @@ int topo_node_sqlite_member_block_put(sqlite3* db, const char* channel_id, uint6
" update_ts=excluded.update_ts,"
" userinfo=excluded.userinfo,"
" creator_sig=excluded.creator_sig,"
" signed_by=excluded.signed_by,"
" signature=excluded.signature,"
" source=0", peers_tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1;
@@ -512,6 +517,9 @@ int topo_node_sqlite_member_block_put(sqlite3* db, const char* channel_id, uint6
sqlite3_bind_text(stmt, 8, userinfo ? userinfo : "", -1, SQLITE_STATIC);
if (creator_sig) sqlite3_bind_blob(stmt, 9, creator_sig, PEERS_JOIN_SIG_SIZE, SQLITE_STATIC);
else sqlite3_bind_null(stmt, 9);
sqlite3_bind_int64(stmt, 10, (sqlite3_int64)signed_by);
if (signature) sqlite3_bind_blob(stmt, 11, signature, 64, SQLITE_STATIC);
else sqlite3_bind_null(stmt, 11);
int rc = sqlite3_step(stmt); sqlite3_finalize(stmt);
return rc == SQLITE_DONE ? 0 : -1;
}
@@ -583,6 +591,38 @@ int topo_node_sqlite_member_get_join(sqlite3* db, const char* channel_id, uint64
return rc;
}
int topo_node_sqlite_member_get_sign(sqlite3* db, const char* channel_id, uint64_t node_id,
uint64_t* signed_by_out, uint8_t signature_out[64]) {
if (!db || !channel_id || !signed_by_out || !signature_out) return -1;
char peers_tbl[80]; peers_table_name(channel_id, peers_tbl, sizeof(peers_tbl));
char sql[256];
snprintf(sql, sizeof(sql), "SELECT signed_by, signature FROM \"%s\" WHERE node_id=?", peers_tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1;
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)node_id);
int rc = -1;
if (sqlite3_step(stmt) == SQLITE_ROW) {
*signed_by_out = (uint64_t)sqlite3_column_int64(stmt, 0);
const uint8_t* sig = (const uint8_t*)sqlite3_column_blob(stmt, 1);
if (sig && sqlite3_column_bytes(stmt, 1) >= 64) { memcpy(signature_out, sig, 64); rc = 0; }
}
sqlite3_finalize(stmt);
return rc;
}
int topo_node_sqlite_member_in_channel(sqlite3* db, const char* channel_id, uint64_t node_id) {
if (!db || !channel_id || node_id == 0) return 0;
char peers_tbl[80]; peers_table_name(channel_id, peers_tbl, sizeof(peers_tbl));
char sql[256];
snprintf(sql, sizeof(sql), "SELECT 1 FROM \"%s\" WHERE node_id=?", peers_tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return 0;
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)node_id);
int found = (sqlite3_step(stmt) == SQLITE_ROW);
sqlite3_finalize(stmt);
return found;
}
int topo_node_sqlite_member_del(sqlite3* db, const char* channel_id, uint64_t node_id) {
if (!db || !channel_id) return -1;
char peers_tbl[80]; peers_table_name(channel_id, peers_tbl, sizeof(peers_tbl));
+10 -2
View File
@@ -41,12 +41,20 @@ int topo_node_sqlite_channel_delete(sqlite3* db, const char* channel_id);
int topo_node_sqlite_member_placeholder_put(sqlite3* db, const char* channel_id, uint64_t node_id,
const uint8_t* x25519_pubkey, const uint8_t* ed25519_pubkey);
/* Обновляет только блок мембера (keys/join/update_sig/update_ts/userinfo), не трогая блок владельца. */
/* Обновляет только блок мембера (keys/join/update_sig/update_ts/userinfo), не трогая блок владельца.
* signed_by/signature — дерево приглашений: кто подписал pubkey узла (0 = корень, ключом группы). */
int topo_node_sqlite_member_block_put(sqlite3* db, const char* channel_id, uint64_t node_id,
const uint8_t* join_sig, uint64_t join_ts,
const uint8_t* update_sig, uint64_t update_ts,
const uint8_t* x25519_pubkey, const uint8_t* ed25519_pubkey, const char* userinfo,
const uint8_t* creator_sig);
const uint8_t* creator_sig, uint64_t signed_by, const uint8_t* signature);
/* Прочитать дерево приглашений мембера (signed_by + signature). 0=ок, -1=нет записи/полей. */
int topo_node_sqlite_member_get_sign(sqlite3* db, const char* channel_id, uint64_t node_id,
uint64_t* signed_by_out, uint8_t signature_out[64]);
/* Есть ли узел в таблице мемберов канала peers_<ch_id> (source 0 или 1). 1=да, 0=нет. */
int topo_node_sqlite_member_in_channel(sqlite3* db, const char* channel_id, uint64_t node_id);
/* Обновляет только блок владельца (adm_tags/adm_tags_sig/storage). */
int topo_node_sqlite_member_owner_put(sqlite3* db, const char* channel_id, uint64_t node_id,