jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
e3b0eebc2b3dc7bd2a7f78f93e7b6df88e8694ce
utun2/src/chat/member_sync.c
T

1513 lines
80 KiB
C
Raw Blame History

#include "../routing_layer/topo_group_connect.h"
#include "member_sync.h"
#include "db_sync.h"
#include "chat_feedback.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/topo_group.h"
#include "../radio/radio.h"
#include "chat_core.h"
#include "chat_core_priv.h"
#include "../utun_instance.h"
#include "../transport_layer/etcp.h"
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include <string.h>
#include <stdlib.h>
#include <errno.h>
#include <limits.h>
#include <sqlite3.h>
#include <openssl/evp.h>
#include "../../lib/json_flat.h"
#define MS_ID "member_sync"
/* ── DB access ── */
static sqlite3* _db(struct UTUN_INSTANCE* inst) {
return inst ? inst->topo_sqlite_db : NULL;
}
static void _peers_table(const char* ch_id, char* buf, size_t sz) {
size_t i = 0;
while (*ch_id && i < sz - 1) {
char c = *ch_id++;
if ((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_')
buf[i++] = c; else buf[i++] = '_';
}
buf[i] = '\0';
char tbl[128]; snprintf(tbl, sizeof(tbl), "peers_%s", buf);
snprintf(buf, sz, "%s", tbl);
}
/* Канонические целые для Merkle и wire; формат подписываемых сообщений задаётся отдельно. */
static void member_write64(uint8_t* p, uint64_t n) {
for (int i = 7; i >= 0; i--) { p[i] = (uint8_t)n; n >>= 8; }
}
static uint64_t member_read64(const uint8_t* p) {
uint64_t n = 0;
for (int i = 0; i < 8; i++) n = (n << 8) | p[i];
return n;
}
/* Версия owner-блока — каноническая десятичная строка1..INT_MAX,без atoi overflow. */
static int owner_version_pair(const char* key, const char* text, void* arg) {
if (strcmp(key, "ver")) return 0;
int value = 0;
if (*text < '1' || *text > '9') goto invalid;
for (const char* p = text; *p; p++) {
if (*p < '0' || *p > '9' || value > (INT_MAX - (*p - '0')) / 10) goto invalid;
value = value * 10 + (*p - '0');
}
*(int*)arg = value; return 0;
invalid:
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member owner: invalid version format/range"); return -1;
}
int member_sync_owner_version(sqlite3* db, const char* tags) {
if (!tags || !tags[0]) return 0;
int version = 0;
if (chat_json_strings(db, tags, MS_ADM_TAGS_MAX, owner_version_pair, &version)) return -1;
if (!version) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member owner: missing version"); return -1; }
return version;
}
int member_sync_build_owner_msg(uint64_t node_id, const char* tags, uint8_t* out, size_t capacity) {
size_t len = tags ? strlen(tags) : 0;
if (!tags || !out || len > MS_ADM_TAGS_MAX || capacity < len + 8) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid owner message node=%016llx bytes=%zu capacity=%zu max=%d",
MS_ID, (unsigned long long)node_id, len, capacity, MS_ADM_TAGS_MAX);
return -1;
}
memcpy(out, tags, len);
memcpy(out + len, &node_id, 8);
return (int)len + 8;
}
static int _compute_member_hash(uint64_t node_id, const uint8_t* x25519, const uint8_t* ed25519,
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,
uint64_t signed_by, const uint8_t* signature, uint8_t hash_out[MT_HASH_SIZE]) {
uint8_t data[MS_ADM_TAGS_MAX + MS_USERINFO_MAX + 512];
size_t off = 0;
member_write64(data + off, node_id); off += 8;
memcpy(data + off, x25519, 32); off += 32;
memcpy(data + off, ed25519, 32); off += 32;
if (join_sig) memcpy(data + off, join_sig, 64); else memset(data + off, 0, 64);
off += 64; member_write64(data + off, join_ts); off += 8;
if (update_sig) memcpy(data + off, update_sig, 64); else memset(data + off, 0, 64);
off += 64; member_write64(data + off, update_ts); off += 8;
size_t ul = userinfo ? strlen(userinfo) : 0, atl = adm_tags ? strlen(adm_tags) : 0;
if (ul > MS_USERINFO_MAX || atl > MS_ADM_TAGS_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: member strings exceed wire limits node=%016llx", MS_ID, (unsigned long long)node_id);
return -1;
}
data[off++] = (uint8_t)(ul >> 8); data[off++] = (uint8_t)ul;
if (ul) { memcpy(data + off, userinfo, ul); off += ul; }
data[off++] = (uint8_t)(atl >> 8); data[off++] = (uint8_t)atl;
if (atl) { memcpy(data + off, adm_tags, atl); off += atl; }
if (adm_tags_sig) memcpy(data + off, adm_tags_sig, 64); else memset(data + off, 0, 64);
off += 64; member_write64(data + off, signed_by); off += 8;
if (signature) memcpy(data + off, signature, 64); else memset(data + off, 0, 64);
off += 64;
if (EVP_Digest(data, off, hash_out, NULL, EVP_sha256(), NULL) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: member hash failed node=%016llx", MS_ID, (unsigned long long)node_id);
return -1;
}
return 0;
}
/* SQLite BLOB pointers are only safe after checking their exact length. */
static int member_blob(sqlite3_stmt* st, int col, int size, int optional) {
int bytes = sqlite3_column_bytes(st, col);
return bytes == size || (optional && bytes == 0);
}
int member_sync_build_update_msg(const uint8_t* join_sig, uint64_t update_ts,
const char* userinfo,
uint8_t* out, int out_sz) {
if (!join_sig || !out || out_sz < 0) return -1;
int off = 0;
if (off + 64 > out_sz) return -1;
memcpy(out + off, join_sig, 64); off += 64;
if (off + 8 > out_sz) return -1;
memcpy(out + off, &update_ts, 8); off += 8;
size_t ul = userinfo ? strlen(userinfo) : 0;
if (off + (int)ul + 1 > out_sz) return -1;
memcpy(out + off, userinfo ? userinfo : "", ul + 1); off += (int)ul + 1;
return off;
}
int member_sync_build_join_msg(const uint8_t* ch_x25519, const uint8_t* ch_ed25519,
uint64_t node_id, const uint8_t* node_x25519, uint64_t join_ts,
uint8_t* out, int out_sz) {
if (!ch_x25519 || !ch_ed25519 || !node_x25519 || !out || out_sz < 0) return -1;
int off = 0;
if (off + 32 > out_sz) return -1;
memcpy(out + off, ch_x25519, 32); off += 32;
if (off + 32 > out_sz) return -1;
memcpy(out + off, ch_ed25519, 32); off += 32;
if (off + 8 > out_sz) return -1;
memcpy(out + off, &node_id, 8); off += 8;
if (off + 32 > out_sz) return -1;
memcpy(out + off, node_x25519, 32); off += 32;
if (off + 8 > out_sz) return -1;
memcpy(out + off, &join_ts, 8); off += 8;
return off;
}
/* ── merkle_sync_data_ops implementation ── */
static int _member_update_bucket_hash(void* ctx, const char* ns, uint8_t level,
uint64_t prefix64, EVP_MD_CTX* sha_ctx) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
sqlite3* db = _db(inst); if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: bucket_hash — db is NULL", MS_ID); return -1; }
if (level != MT_MAX_LEVEL) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: expected leaf, level=%u", MS_ID, level); return -1;
}
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: bucket_hash ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix64);
char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl));
int mask_shift = 64 - (int)level * 5;
uint64_t mask = (mask_shift >= 0 && mask_shift < 64) ? (~0ULL << mask_shift) : UINT64_MAX;
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, signed_by, signature, source, userinfo"
" FROM \"%s\""
" WHERE (node_id & %lld) == %lld ORDER BY node_id ASC",
peers_tbl, (long long)mask, (long long)prefix64);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: bucket_hash prepare failed ns=%s", MS_ID, ns); return -1; }
int count = 0;
int step;
while ((step = sqlite3_step(stmt)) == SQLITE_ROW) {
uint64_t nid = (uint64_t)sqlite3_column_int64(stmt, 0);
const uint8_t* x25 = (const uint8_t*)sqlite3_column_blob(stmt, 1);
const uint8_t* ed = (const uint8_t*)sqlite3_column_blob(stmt, 2);
const uint8_t* sig = (const uint8_t*)sqlite3_column_blob(stmt, 3);
uint64_t jts = (uint64_t)sqlite3_column_int64(stmt, 4);
const uint8_t* usig = (const uint8_t*)sqlite3_column_blob(stmt, 5);
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);
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 (src != 0) continue;
if (!member_blob(stmt, 1, 32, 0) || !member_blob(stmt, 2, 32, 0) || !member_blob(stmt, 3, 64, 1)
|| !member_blob(stmt, 5, 64, 1) || !member_blob(stmt, 8, 64, 1) || !member_blob(stmt, 10, 64, 1)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid hash record ns=%s node=%016llx", MS_ID, ns, (unsigned long long)nid);
sqlite3_finalize(stmt); return -1;
}
uint8_t mh[MT_HASH_SIZE];
const char* userinfo = (const char*)sqlite3_column_text(stmt, 12);
if (_compute_member_hash(nid, x25, ed, sig, jts, usig, uts, userinfo, atags, atsig, sb, sgn, mh) < 0
|| EVP_DigestUpdate(sha_ctx, mh, MT_HASH_SIZE) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: leaf digest failed ns=%s", MS_ID, ns);
sqlite3_finalize(stmt); return -1;
}
count++;
}
sqlite3_finalize(stmt);
if (step != SQLITE_DONE) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: leaf read failed ns=%s: %s", MS_ID, ns, sqlite3_errmsg(db)); return -1;
}
return count;
}
static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int after_valid, const merkle_key_t* after,
uint8_t* buf, size_t* len, merkle_key_t* next, int* more) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
sqlite3* db = _db(inst); if (!db || !buf || !len) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items — invalid args db=%p buf=%p len=%p", MS_ID, (void*)db, (void*)buf, (void*)len); return -1; }
const uint8_t level = MT_MAX_LEVEL;
uint64_t after_id = member_read64(after->bytes);
static const uint8_t zero_tail[MT_KEY_SIZE - 8];
if (after_valid && memcmp(after->bytes + 8, zero_tail, sizeof(zero_tail))) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid member cursor ns=%s", MS_ID, ns); return -1;
}
memset(next, 0, sizeof(*next)); *more = 0;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix);
char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl));
int mask_shift = 64 - (int)level * 5;
uint64_t mask = (mask_shift >= 0 && mask_shift < 64) ? (~0ULL << mask_shift) : UINT64_MAX;
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,"
" signed_by, signature, source"
" FROM \"%s\""
" WHERE (node_id & %lld) == %lld AND (?=0 OR node_id>?) ORDER BY node_id ASC",
peers_tbl, (long long)mask, (long long)prefix);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items prepare failed ns=%s", MS_ID, ns); return -1; }
sqlite3_bind_int(stmt, 1, after_valid);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)after_id);
size_t off = 0;
if (off + 2 > *len) { sqlite3_finalize(stmt); return -2; }
uint16_t cnt = 0; off += 2;
int step;
while ((step = sqlite3_step(stmt)) == SQLITE_ROW) {
uint64_t nid = (uint64_t)sqlite3_column_int64(stmt, 0);
const uint8_t* x25 = (const uint8_t*)sqlite3_column_blob(stmt, 1);
const uint8_t* ed = (const uint8_t*)sqlite3_column_blob(stmt, 2);
const uint8_t* sig = (const uint8_t*)sqlite3_column_blob(stmt, 3);
uint64_t jts = (uint64_t)sqlite3_column_int64(stmt, 4);
const uint8_t* usig = (const uint8_t*)sqlite3_column_blob(stmt, 5);
uint64_t uts = (uint64_t)sqlite3_column_int64(stmt, 6);
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);
int src = sqlite3_column_int(stmt, 12);
if (src != 0) continue;
if (!member_blob(stmt, 1, 32, 0) || !member_blob(stmt, 2, 32, 0) || !member_blob(stmt, 3, 64, 1)
|| !member_blob(stmt, 5, 64, 1) || !member_blob(stmt, 9, 64, 1) || !member_blob(stmt, 11, 64, 1)
|| (nm && strlen(nm) > MS_USERINFO_MAX) || (atags && strlen(atags) > MS_ADM_TAGS_MAX)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid page record ns=%s node=%016llx", MS_ID, ns, (unsigned long long)nid);
sqlite3_finalize(stmt); return -1;
}
size_t nl = nm ? strlen(nm) : 0;
size_t atl = atags ? strlen(atags) : 0;
uint8_t flags = (sig && jts) ? PEERS_FLAG_HAS_JOIN : 0;
size_t need = 8 + 32 + 32 + 1 + (flags ? 72ULL : 0ULL) + 64 + 8 + 2 + nl + 2 + atl + 64 + 8 + 64;
if (off + need > *len) {
if (!cnt) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: record exceeds page capacity ns=%s", MS_ID, ns); sqlite3_finalize(stmt); return -1; }
*more = 1;
break;
}
member_write64(buf + off, nid); off += 8;
memcpy(buf + off, x25, 32); off += 32;
memcpy(buf + off, ed, 32); off += 32;
buf[off++] = flags;
if (flags & PEERS_FLAG_HAS_JOIN) {
memcpy(buf + off, sig, 64); off += 64;
member_write64(buf + off, jts); off += 8;
}
if (usig && uts) {
memcpy(buf + off, usig, 64); off += 64;
member_write64(buf + off, uts); off += 8;
} else {
memset(buf + off, 0, 64); off += 64;
uint64_t z = 0; memcpy(buf + off, &z, 8); off += 8;
}
buf[off++] = (uint8_t)(nl >> 8); buf[off++] = (uint8_t)nl;
if (nl) { memcpy(buf + off, nm, nl); off += nl; }
buf[off++] = (uint8_t)(atl >> 8); buf[off++] = (uint8_t)atl;
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; }
member_write64(buf + off, sb); 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++;
member_write64(next->bytes, nid);
}
if (step != SQLITE_DONE && !*more) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page read failed ns=%s: %s", MS_ID, ns, sqlite3_errmsg(db)); sqlite3_finalize(stmt); return -1; }
sqlite3_finalize(stmt);
buf[0] = (uint8_t)(cnt >> 8); buf[1] = (uint8_t)cnt;
*len = off;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items ns=%s L%d/P%016llx mask=%016llx count=%u off=%zu",
MS_ID, ns, level, (unsigned long long)prefix, (unsigned long long)mask, cnt, off);
return 0;
}
/* ── node_props_changed (adm_tags) callbacks ── */
struct ms_props_cbk { node_props_changed_fn fn; void* arg; struct ms_props_cbk* next; };
static int member_callback_channel(const char* channel, char out[64]) {
size_t size = channel ? strnlen(channel, 64) : 0;
if (!size || size == 64) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member callback: invalid namespace"); return -1; }
memcpy(out, channel, size + 1); return 0;
}
static void _fire_props_changed(struct UTUN_INSTANCE* inst, uint64_t node_id, const char* adm_tags, const char* channel_id) {
if (!inst || !CC(inst)) return;
char channel[64]; if (member_callback_channel(channel_id, channel)) return;
struct chat_core_ctx* cc = CC(inst);
cc->props_dispatch_depth++;
for (struct ms_props_cbk* pc = cc->props_cbks; pc; pc = pc->next)
if (pc->fn) pc->fn(node_id, adm_tags, channel, pc->arg);
if (--cc->props_dispatch_depth == 0) {
struct ms_props_cbk** p = &cc->props_cbks;
while (*p) {
if (!(*p)->fn) { struct ms_props_cbk* removed = *p; *p = removed->next; u_free(removed); }
else p = &(*p)->next;
}
}
}
int member_sync_add_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed_fn fn, void* arg) {
if (!inst || !fn || !CC(inst)) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "props callback invalid arguments"); return -1; }
struct ms_props_cbk* e = u_malloc(sizeof(*e));
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "props callback allocation failed"); return -1; }
e->fn = fn; e->arg = arg;
e->next = CC(inst)->props_cbks;
CC(inst)->props_cbks = e;
return 0;
}
void member_sync_remove_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed_fn fn, void* arg) {
if (!inst || !fn || !CC(inst)) return;
struct ms_props_cbk** p = &CC(inst)->props_cbks;
while (*p) {
if ((*p)->fn == fn && (*p)->arg == arg) {
struct ms_props_cbk* r = *p;
if (CC(inst)->props_dispatch_depth) { r->fn = NULL; r->arg = NULL; }
else { *p = r->next; u_free(r); }
return;
}
p = &(*p)->next;
}
}
/* ── Коллбэк применения рекорда мембера (ready-хендлинг join на connection-узле) ── */
struct ms_apply_cbk { member_apply_cbk_fn fn; void* arg; struct ms_apply_cbk* next; };
static void _fire_apply_cbk(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t node_id,
const uint8_t* x25519, uint64_t signed_by,
const uint8_t* signature) {
if (!inst || !CC(inst)) return;
char channel[64]; if (member_callback_channel(ch_id, channel)) return;
struct chat_core_ctx* cc = CC(inst); cc->apply_dispatch_depth++;
for (struct ms_apply_cbk* ac = cc->apply_cbks; ac; ac = ac->next)
if (ac->fn) ac->fn(channel, node_id, x25519, signed_by, signature, ac->arg);
if (--cc->apply_dispatch_depth == 0) {
struct ms_apply_cbk** p = &cc->apply_cbks;
while (*p) {
if (!(*p)->fn) { struct ms_apply_cbk* removed = *p; *p = removed->next; u_free(removed); }
else p = &(*p)->next;
}
}
}
int member_sync_add_apply_cbk(struct UTUN_INSTANCE* inst, member_apply_cbk_fn fn, void* arg) {
if (!inst || !fn || !CC(inst)) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "apply callback: invalid registration"); return -1; }
struct ms_apply_cbk* e = u_malloc(sizeof(*e));
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "apply callback: allocation failed"); return -1; }
e->fn = fn; e->arg = arg;
e->next = CC(inst)->apply_cbks;
CC(inst)->apply_cbks = e;
return 0;
}
void member_sync_remove_apply_cbk(struct UTUN_INSTANCE* inst, member_apply_cbk_fn fn, void* arg) {
if (!inst || !fn || !CC(inst)) return;
struct ms_apply_cbk** p = &CC(inst)->apply_cbks;
while (*p) {
if ((*p)->fn == fn && (*p)->arg == arg) {
struct ms_apply_cbk* r = *p;
if (CC(inst)->apply_dispatch_depth) { r->fn = NULL; r->arg = NULL; }
else { *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 потребителям. ── */
static void _ms_on_bgp_node(struct TOPO_GROUP* group, uint64_t node_id, int event, void* arg) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
if (!inst || !group) return;
sqlite3* db = _db(inst); if (!db) return;
/* REMOVE не удаляет member (удаление — только битая подпись при верификации) */
char tags[MS_ADM_TAGS_MAX + 1] = "";
char peers_tbl[128]; _peers_table(group->channel_id, peers_tbl, sizeof(peers_tbl));
sqlite3_stmt* st = NULL;
char sql[256]; snprintf(sql, sizeof(sql), "SELECT COALESCE(adm_tags,'') FROM \"%s\" WHERE node_id=?", peers_tbl);
int rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc == SQLITE_ROW) {
const char* text = (const char*)sqlite3_column_text(st, 0); int bytes = sqlite3_column_bytes(st, 0);
if (!text || sqlite3_column_type(st, 0) != SQLITE_TEXT || bytes > MS_ADM_TAGS_MAX || strlen(text) != (size_t)bytes) rc = SQLITE_CORRUPT;
else memcpy(tags, text, (size_t)bytes + 1);
}
int final = sqlite3_finalize(st);
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member properties read node=%016llx rc=%d finalize=%d: %s",
(unsigned long long)node_id, rc, final, sqlite3_errmsg(db)); return;
}
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: bgp_node ev=%d nid=0x%016llx tags_bytes=%zu",
MS_ID, event, (unsigned long long)node_id, strlen(tags));
_fire_props_changed(inst, node_id, tags[0] ? tags : NULL, group->channel_id);
}
static int _sig_is_zero64(const uint8_t* sig) {
if (!sig) return 1;
for (int i = 0; i < 64; i++) if (sig[i]) return 0;
return 1;
}
/* Верифицирует join_sig: sign(ch_x25519||ch_ed25519||node_id||x25519||join_ts) ключом ed25519. */
static int _verify_join_sig(sqlite3* db, const char* ch_id, const struct ms_member_rec* m) {
if (!m->join_sig || _sig_is_zero64(m->join_sig)) return 0;
uint8_t ch_x[32], ch_ed[32];
int lookup = topo_node_sqlite_channel_get(db, ch_id, NULL, 0, NULL, ch_x, ch_ed, NULL);
if (lookup != 0) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: join verification unavailable ns=%s lookup=%d", MS_ID, ch_id, lookup);
return lookup == TOPO_SQLITE_LOOKUP_MISSING ? MS_VERIFY_DEFERRED : MS_VERIFY_ERROR;
}
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) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: join message build failed ns=%s", MS_ID, ch_id); return MS_VERIFY_ERROR;
}
int rc = sc_ed25519_verify(m->ed25519, vmsg, (size_t)vlen, m->join_sig);
if (rc != SC_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: join verification ns=%s node=%016llx rc=%d",
MS_ID, ch_id, (unsigned long long)m->node_id, rc);
return rc == SC_OK ? 1 : rc == SC_ERR_AUTH_FAILED ? 0 : MS_VERIFY_ERROR;
}
/* ── Дерево приглашений: подпись 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) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid invitation verification arguments", MS_ID); return MS_VERIFY_ERROR;
}
if (!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) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invitation message build failed", MS_ID); return MS_VERIFY_ERROR;
}
uint8_t signer_pub[32];
int lookup = m->signed_by == 0
? topo_node_sqlite_channel_get(db, ch_id, NULL, 0, NULL, NULL, signer_pub, NULL)
: topo_node_sqlite_get_ed25519_pubkey(db, m->signed_by, signer_pub);
if (lookup != 0) {
if (lookup == TOPO_SQLITE_LOOKUP_MISSING) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invitation deferred ns=%s signer=%016llx",
MS_ID, ch_id, (unsigned long long)m->signed_by); return MS_VERIFY_DEFERRED;
}
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invitation key lookup failed ns=%s signer=%016llx",
MS_ID, ch_id, (unsigned long long)m->signed_by); return MS_VERIFY_ERROR;
}
int rc = sc_ed25519_verify(signer_pub, msg, sizeof(msg), m->signature);
if (rc != SC_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invitation verification ns=%s node=%016llx signer=%016llx rc=%d",
MS_ID, ch_id, (unsigned long long)m->node_id, (unsigned long long)m->signed_by, rc);
return rc == SC_OK ? 1 : rc == SC_ERR_AUTH_FAILED ? 0 : MS_VERIFY_ERROR;
}
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);
}
static int member_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) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record — invalid args", MS_ID);
return -1;
}
sqlite3* db = _db(inst); if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record — db is NULL", MS_ID); return -1; }
int stale = 0, changed = 0, node_key_changed = 0;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record ch=%s nid=%016llx uts=%llu tags_bytes=%zu",
MS_ID, ch_id, (unsigned long long)m->node_id, (unsigned long long)m->update_ts,
m->adm_tags ? strlen(m->adm_tags) : 0);
char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
/* ── загрузить локальную запись ── */
int has_local = 0;
uint8_t local_join_sig[64] = {0}; uint64_t local_join_ts = 0;
uint64_t local_update_ts = 0;
char local_tags[MS_ADM_TAGS_MAX + 1] = "";
uint64_t local_signed_by = 0;
uint8_t local_signature[64] = {0};
int local_src = 0, local_identity_matches = 0, local_deleted = 0;
{
sqlite3_stmt* st = NULL;
char sql[512]; snprintf(sql, sizeof(sql),
"SELECT join_sig,join_ts,update_ts,adm_tags,signed_by,signature,source,x25519_pubkey,ed25519_pubkey,deleted"
" FROM \"%s\" WHERE node_id=?", peers_tbl);
int rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)m->node_id);
if (rc == SQLITE_OK) {
rc = sqlite3_step(st);
if (rc == SQLITE_ROW && (!member_blob(st, 0, 64, 1) || !member_blob(st, 5, 64, 1))) rc = SQLITE_CORRUPT;
if (rc == SQLITE_ROW) {
has_local = 1;
const uint8_t* js = (const uint8_t*)sqlite3_column_blob(st, 0);
if (js) memcpy(local_join_sig, js, 64);
local_join_ts = (uint64_t)sqlite3_column_int64(st, 1);
local_update_ts = (uint64_t)sqlite3_column_int64(st, 2);
const char* tags = (const char*)sqlite3_column_text(st, 3);
int tags_type = sqlite3_column_type(st, 3), tags_bytes = sqlite3_column_bytes(st, 3);
if (tags_type != SQLITE_NULL && (tags_type != SQLITE_TEXT || !tags || tags_bytes > MS_ADM_TAGS_MAX ||
strlen(tags) != (size_t)tags_bytes)) rc = SQLITE_CORRUPT;
else if (tags) memcpy(local_tags, tags, (size_t)tags_bytes + 1);
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);
local_src = sqlite3_column_int(st, 6);
const void* old_x = sqlite3_column_blob(st, 7); const void* old_ed = sqlite3_column_blob(st, 8);
local_identity_matches = old_x && old_ed && sqlite3_column_bytes(st, 7) == 32 && sqlite3_column_bytes(st, 8) == 32
&& !memcmp(old_x, m->x25519, 32) && !memcmp(old_ed, m->ed25519, 32);
local_deleted = sqlite3_column_int(st, 9);
if (sqlite3_errcode(db) == SQLITE_NOMEM) rc = SQLITE_NOMEM;
}
}
int final = sqlite3_finalize(st);
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: local record read failed ns=%s node=%016llx rc=%d finalize=%d: %s",
MS_ID, ch_id, (unsigned long long)m->node_id, rc, final, sqlite3_errmsg(db)); return MT_ERR_IO;
}
}
/* ── Блок A (мембер): verify update_sig (или join_sig) + сравнить ver ── */
int is_local = (from_peer == inst->node_id);
int identity_ok = 1;
int has_signature = m->signature && !_sig_is_zero64(m->signature);
if (!is_local) {
/* жёсткая привязка идентичности: node_id = derive(x25519), ed25519 — доверенный из nodes */
uint64_t derived = sc_derive_node_id_from_pubkey(m->x25519);
if (!derived) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: identity verification unavailable ns=%s", MS_ID, ch_id); return MT_ERR_IO;
}
if (derived != m->node_id) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record node_id/x25519 mismatch (forgery) nid=0x%016llx ns=%s", MS_ID, (unsigned long long)m->node_id, ch_id);
identity_ok = 0;
} else {
uint8_t trusted_ed[32];
int lookup = topo_node_sqlite_get_ed25519_pubkey(db, m->node_id, trusted_ed);
if (lookup == TOPO_SQLITE_LOOKUP_ERROR) return MT_ERR_IO;
if (lookup == 0 && memcmp(m->ed25519, trusted_ed, 32) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record ed25519 mismatch trusted (forgery) nid=0x%016llx ns=%s", MS_ID, (unsigned long long)m->node_id, ch_id);
identity_ok = 0;
}
}
/* подпись дерева приглашений: -1 = родитель ещё не пришёл (defer до verify_local), 0 = явная подделка */
if (identity_ok && has_signature) {
int sv = member_sync_verify_signature(db, ch_id, m);
if (sv == MS_VERIFY_ERROR) return MT_ERR_IO;
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;
if (is_local) {
/* локальная запись (member_sync_put) — данные доверенные, без верификации */
block_a_ok = 1;
} else if (!identity_ok) {
block_a_ok = 0;
} else if (m->update_sig && !_sig_is_zero64(m->update_sig)) {
const uint8_t* jsig = m->join_sig ? m->join_sig : (has_local ? local_join_sig : NULL);
if (jsig) {
uint8_t vmsg[8192];
int vlen = member_sync_build_update_msg(jsig, m->update_ts,
m->userinfo ? m->userinfo : "", vmsg, (int)sizeof(vmsg));
if (vlen >= 0) {
int verified = sc_ed25519_verify(m->ed25519, vmsg, (size_t)vlen, m->update_sig);
if (verified == SC_OK) block_a_ok = 1;
else {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: update verification ns=%s node=%016llx rc=%d",
MS_ID, ch_id, (unsigned long long)m->node_id, verified);
if (verified != SC_ERR_AUTH_FAILED) return MT_ERR_IO;
}
}
}
} else if (m->join_sig && !_sig_is_zero64(m->join_sig)) {
/* join-only запись — верифицируем join_sig напрямую */
int verified = _verify_join_sig(db, ch_id, m);
if (verified == MS_VERIFY_ERROR) return MT_ERR_IO;
block_a_ok = verified == 1;
}
int block_a_changed = 0, block_a_stale = 0;
if (block_a_ok) {
if (!has_local) {
block_a_changed = 1;
} else {
if (m->join_sig && m->join_ts > local_join_ts) block_a_changed = 1; /* re-join / key rotation */
else if (m->join_sig && m->join_ts < local_join_ts) block_a_stale = 1; /* старая эпоха */
else {
if (m->update_ts > local_update_ts) block_a_changed = 1;
else if (m->update_ts < local_update_ts) block_a_stale = 1;
}
}
}
/* ── Блок B (владелец): verify adm_tags_sig + сравнить ver (adm_tags.ver) ── */
int block_b_ok = 0, block_b_changed = 0, block_b_stale = 0;
int adm_ver = 0, adm_storage = 0;
if (m->adm_tags && m->adm_tags[0] && m->adm_tags_sig && !_sig_is_zero64(m->adm_tags_sig)) {
char ver_str[32];
adm_ver = member_sync_owner_version(db, m->adm_tags);
if (adm_ver < 0) return MT_ERR_IO;
adm_storage = (json_flat_get(m->adm_tags, "storage", ver_str, sizeof(ver_str)) == 0 && strcmp(ver_str, "yes") == 0) ? 1 : 0;
uint8_t ch_ed_pub[32] = {0};
int lookup = topo_node_sqlite_channel_get(db, ch_id, NULL, 0, NULL, NULL, ch_ed_pub, NULL);
if (lookup == 0) {
uint8_t amsg[MS_ADM_TAGS_MAX + 8];
int aoff = member_sync_build_owner_msg(m->node_id, m->adm_tags, amsg, sizeof(amsg));
if (aoff < 0) return MT_ERR_DATA;
int verified = sc_ed25519_verify(ch_ed_pub, amsg, (size_t)aoff, m->adm_tags_sig);
if (verified == SC_OK) block_b_ok = 1;
else {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: owner verification ns=%s node=%016llx rc=%d",
MS_ID, ch_id, (unsigned long long)m->node_id, verified);
if (verified != SC_ERR_AUTH_FAILED) return MT_ERR_IO;
}
} else {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record cannot verify adm_tags_sig — no channel key for ns=%s", MS_ID, ch_id);
if (lookup != TOPO_SQLITE_LOOKUP_MISSING) return MT_ERR_IO;
}
}
if (block_b_ok) {
int local_ver = 0;
if (has_local && local_tags[0]) {
local_ver = member_sync_owner_version(db, local_tags);
if (local_ver < 0) return MT_ERR_IO;
}
if (adm_ver > local_ver) block_b_changed = 1;
else if (adm_ver < local_ver) block_b_stale = 1;
}
if (block_b_changed && local_src != 0)
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: adm_tags change on placeholder member (source=%d) — ignored by merkle, won't propagate nid=0x%016llx ns=%s",
MS_ID, local_src, (unsigned long long)m->node_id, ch_id);
if (!identity_ok || !block_a_ok || (m->adm_tags && m->adm_tags[0] && !block_b_ok)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: rejected record ns=%s node=%016llx identity=%d member=%d owner=%d",
MS_ID, ch_id, (unsigned long long)m->node_id, identity_ok, block_a_ok, block_b_ok);
return MT_ERR_DATA;
}
/* Подпись приглашения не зависит от версии профиля: подтверждаем уже сохранённого мембера. */
int signature_changed = has_local && has_signature && _sig_is_zero64(local_signature);
if (signature_changed) {
int verified = member_sync_verify_signature(db, ch_id, m);
if (verified == MS_VERIFY_ERROR) return MT_ERR_IO;
signature_changed = verified == 1;
}
/* ── запись изменённых блоков ── */
if (block_a_changed) {
const uint8_t* jsig = m->join_sig ? m->join_sig : (has_local ? local_join_sig : NULL);
/* Нулевой блок на проводе означает отсутствие подписи и не стирает сохранённое подтверждение. */
uint64_t w_signed_by = has_signature ? m->signed_by : local_signed_by;
const uint8_t* w_signature = has_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, 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;
}
changed = 1;
/* узел в nodes пишет только merkle-путь (verified identity) */
char node_name[128] = "";
if (m->userinfo) json_flat_get(m->userinfo, "name", node_name, sizeof(node_name));
if (merkle_sync_registered(inst, MT_DATA_FEEDBACK)) {
uint8_t previous_key[32];
int lookup = topo_node_sqlite_get_ed25519_pubkey(db, m->node_id, previous_key);
if (lookup == TOPO_SQLITE_LOOKUP_ERROR) return MT_ERR_IO;
node_key_changed = lookup == TOPO_SQLITE_LOOKUP_MISSING || memcmp(previous_key, m->ed25519, 32);
}
if (topo_node_sqlite_node_update_verified(db, m->node_id, node_name, m->x25519, m->ed25519,
m->update_ts ? m->update_ts : m->join_ts,
ntp_time_get_seconds(inst)) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: node update failed ns=%s node=%016llx", MS_ID, ch_id,
(unsigned long long)m->node_id); return MT_ERR_IO;
}
}
if (signature_changed && !block_a_changed) {
char sql[256]; sqlite3_stmt* st = NULL;
snprintf(sql, sizeof(sql), "UPDATE \"%s\" SET signed_by=?,signature=? WHERE node_id=?", peers_tbl);
int rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)m->signed_by);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, m->signature, 64, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)m->node_id);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
sqlite3_finalize(st);
if (rc != SQLITE_DONE) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invitation update failed ns=%s node=%016llx rc=%d: %s",
MS_ID, ch_id, (unsigned long long)m->node_id, rc, sqlite3_errmsg(db)); return MT_ERR_IO;
}
changed = 1;
}
if (signature_changed) DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invitation confirmed ns=%s node=%016llx signer=%016llx",
MS_ID, ch_id, (unsigned long long)m->node_id, (unsigned long long)m->signed_by);
if (block_b_changed) {
if (topo_node_sqlite_member_owner_put(db, ch_id, m->node_id, m->adm_tags, m->adm_tags_sig, adm_storage) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record owner_put FAILED ch=%s nid=0x%016llx", MS_ID, ch_id, (unsigned long long)m->node_id);
return -1;
}
changed = 1;
}
if (block_a_stale || block_b_stale) stale = 1;
/* пересчёт классификации (node_type/storage) — DB-хелпер, читает node_addresses (адреса BGP) */
if (changed && topo_node_sqlite_nodeinfo_updated(db, m->node_id) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: classification update failed ns=%s node=%016llx", MS_ID, ch_id,
(unsigned long long)m->node_id); return MT_ERR_IO;
}
char deleted_value[16];
int deleted_changed = block_b_changed && local_deleted != (json_flat_get(m->adm_tags, "deleted",
deleted_value, sizeof(deleted_value)) == 0 && !strcmp(deleted_value, "yes"));
int credentials_changed = block_a_changed && (!has_local || local_src != 0 || !local_identity_matches
|| m->join_ts != local_join_ts || (m->join_sig && memcmp(m->join_sig, local_join_sig, 64))
|| (has_signature && (m->signed_by != local_signed_by || memcmp(m->signature, local_signature, 64))));
int feedback_changed = merkle_sync_registered(inst, MT_DATA_FEEDBACK)
&& (credentials_changed || signature_changed || deleted_changed || node_key_changed);
/* Профиль/stream options не меняют допуск feedback и не требуют новых его раундов. */
if (feedback_changed && chat_feedback_member_update(inst, m->node_id) < 0) return MT_ERR_IO;
/* ── recompute и подтвердить реальное изменение по хешу ── */
if (changed) {
int rc = merkle_sync_recompute_path(inst, MT_DATA_MEMBERS, ch_id, m->node_id);
if (rc < 0) return -1;
if (rc == 0) changed = 0; /* запись была no-op (данные идентичны) */
}
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record ch=%s nid=0x%016llx → changed=%d stale=%d feedback=%d",
MS_ID, ch_id, (unsigned long long)m->node_id, changed, stale, feedback_changed);
return (changed ? MS_APPLY_CHANGED : 0) | (stale ? MS_APPLY_STALE : 0) | (feedback_changed ? MS_APPLY_FEEDBACK : 0);
}
void member_sync_record_committed(struct UTUN_INSTANCE* inst, const char* ns, uint64_t node_id, unsigned changes) {
char channel[64]; if (!inst || member_callback_channel(ns, channel)) return;
merkle_sync_changed(inst, MT_DATA_MEMBERS, channel);
if (changes & MS_APPLY_FEEDBACK) chat_feedback_member_committed(inst, node_id);
db_sync_members_changed(inst, channel);
uint64_t group_id = strtoull(channel, NULL, 10);
struct TOPO_GROUP* g = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
if (g && g->group_type == TOPO_GROUP_TYPE_CHAT) {
_ms_on_bgp_node(g, node_id, 0, inst);
/* Получатель свойств вправе удалить группу/соединение;после callback берём только актуальные handles. */
g = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
if (!g || g->group_type != TOPO_GROUP_TYPE_CHAT) return;
struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
/* Существующий UP сохраняем; новый транспорт выбирает только автопланировщик. */
if (conn && conn->links_up && !conn->close_requested && topo_group_new_conn(g, conn) < 0)
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: cannot attach UP member channel=%s node=%016llx", MS_ID, channel,
(unsigned long long)node_id);
g = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
if (g && g->group_type == TOPO_GROUP_TYPE_CHAT) topo_group_connect_changed(g);
}
}
int member_sync_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
uint64_t from_peer, const struct ms_member_rec* m) {
sqlite3* db = _db(inst);
if (!db || !ch_id || !m) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid apply args", MS_ID); return -1; }
int own = sqlite3_get_autocommit(db);
if (sqlite3_exec(db, "SAVEPOINT member_record", NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: record begin: %s", MS_ID, sqlite3_errmsg(db)); return -1;
}
int rc = member_apply_record(inst, ch_id, from_peer, m);
if (rc < 0) goto error;
int sql_rc = sqlite3_exec(db, "RELEASE member_record", NULL, NULL, NULL);
if (sql_rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: record commit ch=%s rc=%d: %s", MS_ID, ch_id, sql_rc, sqlite3_errmsg(db)); goto error;
}
if (own && rc >= 0 && (rc & (MS_APPLY_CHANGED | MS_APPLY_FEEDBACK)))
member_sync_record_committed(inst, ch_id, m->node_id, (unsigned)rc);
return rc;
error:
if (!sqlite3_get_autocommit(db)) {
int rollback = sqlite3_exec(db, "ROLLBACK TO member_record", NULL, NULL, NULL);
if (rollback == SQLITE_OK) rollback = sqlite3_exec(db, "RELEASE member_record", NULL, NULL, NULL);
if (rollback != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC,
"%s: record rollback ch=%s rc=%d own=%d;partial savepoint not released: %s", MS_ID, ch_id, rollback, own, sqlite3_errmsg(db));
if (own && !sqlite3_get_autocommit(db)) {
rollback = sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL);
if (rollback != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: record full rollback ch=%s rc=%d: %s",
MS_ID, ch_id, rollback, sqlite3_errmsg(db));
if (!sqlite3_get_autocommit(db)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: record transaction remains open;requesting owner stop/DB close", MS_ID);
utun_instance_stop(inst);
}
}
}
return -1;
}
/* ── Верификация локальной записи при чтении из БД ──
Единственный сценарий удаления мембера: подпись присутствует и невалидна,
либо node_id != derive(x25519). Отсутствие подписи = «не верифицировано», не удаляем. */
int member_sync_verify_local_record(struct sqlite3* db, const char* ch_id, const struct ms_member_rec* m) {
if (!db || !ch_id || !m) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid local verification arguments", MS_ID); return -1;
}
if (!m->x25519 || !m->ed25519) return 1; /* неполная запись — не доказательство порчи */
/* жёсткая привязка идентичности */
uint64_t derived = sc_derive_node_id_from_pubkey(m->x25519);
if (!derived) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: local identity verification unavailable ns=%s", MS_ID, ch_id); return -1;
}
if (derived != m->node_id) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: verify_local — node_id/x25519 mismatch nid=0x%016llx ns=%s",
MS_ID, (unsigned long long)m->node_id, ch_id);
return 0;
}
/* update_sig присутствует — верифицируем */
if (m->update_sig && !_sig_is_zero64(m->update_sig)) {
uint8_t vmsg[8192];
int vlen = member_sync_build_update_msg(m->join_sig, m->update_ts,
m->userinfo ? m->userinfo : "", vmsg, (int)sizeof(vmsg));
if (vlen < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid local update message ns=%s node=%016llx",
MS_ID, ch_id, (unsigned long long)m->node_id); return 0;
}
int verified = sc_ed25519_verify(m->ed25519, vmsg, (size_t)vlen, m->update_sig);
if (verified != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: local update verification ns=%s node=%016llx rc=%d",
MS_ID, ch_id, (unsigned long long)m->node_id, verified);
return verified == SC_ERR_AUTH_FAILED ? 0 : -1;
}
} else if (m->join_sig && !_sig_is_zero64(m->join_sig)) {
/* join-only — верифицируем join_sig напрямую */
int verified = _verify_join_sig(db, ch_id, m);
if (verified == MS_VERIFY_ERROR) return -1;
if (verified == 0) return 0;
}
/* adm_tags присутствует — верифицируем adm_tags_sig */
if (m->adm_tags && m->adm_tags[0] && m->adm_tags_sig && !_sig_is_zero64(m->adm_tags_sig)) {
uint8_t ch_ed_pub[32] = {0};
int lookup = topo_node_sqlite_channel_get(db, ch_id, NULL, 0, NULL, NULL, ch_ed_pub, NULL);
if (lookup == TOPO_SQLITE_LOOKUP_ERROR) return -1;
if (lookup == 0) {
uint8_t amsg[MS_ADM_TAGS_MAX + 8];
int aoff = member_sync_build_owner_msg(m->node_id, m->adm_tags, amsg, sizeof(amsg));
if (aoff < 0) return 0;
int verified = sc_ed25519_verify(ch_ed_pub, amsg, (size_t)aoff, m->adm_tags_sig);
if (verified != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: local owner verification ns=%s node=%016llx rc=%d",
MS_ID, ch_id, (unsigned long long)m->node_id, verified);
return verified == SC_ERR_AUTH_FAILED ? 0 : -1;
}
} else {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: local owner verification deferred ns=%s", MS_ID, ch_id);
}
}
/* дерево приглашений: подпись есть и невалидна (подписант резолвится) — битая запись.
-1 = родитель ещё не пришёл — не трогаем (не битая). */
if (m->signature && !_sig_is_zero64(m->signature)) {
int sv = member_sync_verify_signature(db, ch_id, m);
if (sv == MS_VERIFY_ERROR) return -1;
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;
}
/* Владеет транзакцией канала; публикует удаления только после commit данных и индекса. */
int member_sync_verify_and_purge(struct UTUN_INSTANCE* inst, const char* ch_id) {
sqlite3* db = _db(inst);
size_t ns_len = ch_id ? strlen(ch_id) : 0;
if (!db || !inst->msync || !ns_len || ns_len > 20 || strspn(ch_id, "0123456789") != ns_len
|| (ch_id[0] == '0' && ns_len > 1) || !sqlite3_get_autocommit(db)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge rejected ns=%s db=%p external_transaction=%d",
MS_ID, ch_id ? ch_id : "-", (void*)db, db ? !sqlite3_get_autocommit(db) : 0); return -1;
}
errno = 0;
uint64_t group_id = strtoull(ch_id, NULL, 10);
if (errno) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge namespace overflow ns=%s", MS_ID, ch_id); return -1; }
char ns[21]; memcpy(ns, ch_id, ns_len + 1);
char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl));
uint64_t* removed = NULL;
uint32_t capacity = 0;
int count = 0, rc = SQLITE_OK;
const char* stage = "begin";
sqlite3_stmt* st = NULL;
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge begin failed ns=%s: %s", MS_ID, ns, sqlite3_errmsg(db)); return -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge begin ns=%s", MS_ID, ns);
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,"
" signed_by, signature FROM \"%s\" WHERE source=0 ORDER BY node_id ASC", peers_tbl);
stage = "read";
rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
uint64_t nid = (uint64_t)sqlite3_column_int64(st, 0);
const uint8_t* x25 = (const uint8_t*)sqlite3_column_blob(st, 1);
const uint8_t* ed = (const uint8_t*)sqlite3_column_blob(st, 2);
const uint8_t* jsig = (const uint8_t*)sqlite3_column_blob(st, 3);
uint64_t jts = (uint64_t)sqlite3_column_int64(st, 4);
const uint8_t* usig = (const uint8_t*)sqlite3_column_blob(st, 5);
uint64_t uts = (uint64_t)sqlite3_column_int64(st, 6);
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);
uint64_t sb = (uint64_t)sqlite3_column_int64(st, 10);
const uint8_t* sgn = (const uint8_t*)sqlite3_column_blob(st, 11);
if (sqlite3_errcode(db) == SQLITE_NOMEM) { rc = SQLITE_NOMEM; goto fail; }
if (!sqlite3_column_bytes(st, 1) || !sqlite3_column_bytes(st, 2)) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge keeps incomplete keys ns=%s node=%016llx",
MS_ID, ns, (unsigned long long)nid); continue;
}
int ul = sqlite3_column_bytes(st, 7), atl = sqlite3_column_bytes(st, 8);
int malformed = !member_blob(st, 1, 32, 0) || !member_blob(st, 2, 32, 0) || !member_blob(st, 3, 64, 1)
|| !member_blob(st, 5, 64, 1) || !member_blob(st, 9, 64, 1) || !member_blob(st, 11, 64, 1)
|| ul > MS_USERINFO_MAX || atl > MS_ADM_TAGS_MAX || (nm && memchr(nm, 0, (size_t)ul))
|| (atags && memchr(atags, 0, (size_t)atl));
struct ms_member_rec m = { .node_id = nid, .x25519 = x25, .ed25519 = ed, .join_sig = jsig, .join_ts = jts,
.update_sig = usig, .update_ts = uts, .userinfo = nm, .adm_tags = atags, .adm_tags_sig = atsig,
.signed_by = sb, .signature = sgn };
int verified = malformed ? 0 : member_sync_verify_local_record(db, ns, &m);
if (verified < 0) { stage = "verify"; goto fail; }
if (verified == 0) {
if (count == INT_MAX || (uint32_t)count == capacity) {
if (count == INT_MAX || capacity > UINT32_MAX / sizeof(*removed) / 2) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge list too large ns=%s count=%d", MS_ID, ns, count);
stage = "allocate"; goto fail;
}
uint32_t next = capacity ? capacity * 2 : 8;
uint64_t* ids = u_realloc(removed, next * sizeof(*removed));
if (!ids) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge allocation failed ns=%s count=%d", MS_ID, ns, count);
stage = "allocate"; goto fail;
}
removed = ids; capacity = next;
}
removed[count++] = nid;
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge selected ns=%s node=%016llx malformed=%d",
MS_ID, ns, (unsigned long long)nid, malformed);
}
}
if (rc != SQLITE_DONE) goto fail;
rc = sqlite3_finalize(st); st = NULL;
if (rc != SQLITE_OK) goto fail;
stage = "delete";
for (int i = 0; i < count; i++) if (topo_node_sqlite_member_del(db, ns, removed[i]) < 0) goto fail;
stage = "index";
for (int i = 0; i < count; i++)
if (merkle_sync_recompute_path(inst, MT_DATA_MEMBERS, ns, removed[i]) < 0
|| chat_feedback_member_update(inst, removed[i]) < 0) goto fail;
stage = "commit";
rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
if (rc != SQLITE_OK) goto fail;
if (count) {
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge committed ns=%s removed=%d", MS_ID, ns, count);
merkle_sync_changed(inst, MT_DATA_MEMBERS, ns);
}
for (int i = 0; i < count; i++) {
uint64_t nid = removed[i];
chat_feedback_member_committed(inst, nid);
radio_forget_member_mute(inst, group_id, nid);
struct TOPO_GROUP* g = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
struct ETCP_CONN* conn = instance_find_conn(inst, nid);
if (g && g->group_type == TOPO_GROUP_TYPE_CHAT && conn) topo_group_remove_conn(g, conn, TOPO_REMOVE_MEMBER_INVALID);
_fire_props_changed(inst, nid, NULL, ns);
}
u_free(removed);
return count;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge aborted ns=%s stage=%s selected=%d rc=%d: %s",
MS_ID, ns, stage, count, rc, sqlite3_errmsg(db));
rc = sqlite3_finalize(st);
if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge finalize ns=%s rc=%d", MS_ID, ns, rc);
if (!sqlite3_get_autocommit(db)) {
rc = sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge rollback failed ns=%s rc=%d: %s", MS_ID, ns, rc, sqlite3_errmsg(db));
if (!sqlite3_get_autocommit(db)) {
inst->running = 0;
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: database transaction unrecovered; application restart required", MS_ID);
}
}
}
u_free(removed);
return -1;
}
/* Обходит каналы без живого SQLite cursor во время публикации очистки. */
int member_sync_verify_and_purge_all(struct UTUN_INSTANCE* inst) {
sqlite3* db = _db(inst);
if (!db || !sqlite3_get_autocommit(db)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid purge-all context", MS_ID); return -1;
}
char last[128] = "";
sqlite3_stmt* st = NULL;
for (;;) {
int rc = sqlite3_prepare_v2(db, "SELECT name FROM sqlite_master WHERE type='table' AND name GLOB 'peers_*'"
" AND name>? ORDER BY name LIMIT 1", -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, last, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc == SQLITE_ROW) {
const char* table = (const char*)sqlite3_column_text(st, 0);
int len = sqlite3_column_bytes(st, 0);
if (!table || len < 7 || len >= (int)sizeof(last)) rc = SQLITE_CORRUPT;
else memcpy(last, table, (size_t)len + 1);
}
int final_rc = sqlite3_finalize(st); st = NULL;
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final_rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: purge-all read failed rc=%d finalize=%d: %s",
MS_ID, rc, final_rc, sqlite3_errmsg(db)); return -1;
}
if (rc == SQLITE_DONE) return 0;
if (member_sync_verify_and_purge(inst, last + 6) < 0) return -1;
}
}
struct member_wire {
struct ms_member_rec rec;
char userinfo[MS_USERINFO_MAX + 1], tags[MS_ADM_TAGS_MAX + 1];
uint8_t local_join[64];
};
/* Одна строгая раскладка для проверки и применения. Нельзя молча принимать усечённую запись. */
static int member_decode(const uint8_t* p, size_t len, struct member_wire* w, size_t* used) {
memset(w, 0, sizeof(*w));
if (len < 73) return -1;
struct ms_member_rec* m = &w->rec;
size_t off = 0;
m->node_id = member_read64(p + off); off += 8;
m->x25519 = p + off; off += 32; m->ed25519 = p + off; off += 32;
uint8_t flags = p[off++];
if (flags & ~PEERS_FLAG_HAS_JOIN) return -1;
if (flags & PEERS_FLAG_HAS_JOIN) {
if (len - off < 72) return -1;
m->join_sig = p + off; off += 64; m->join_ts = member_read64(p + off); off += 8;
}
if (len - off < 74) return -1;
m->update_sig = p + off; off += 64; m->update_ts = member_read64(p + off); off += 8;
size_t n = ((size_t)p[off] << 8) | p[off + 1]; off += 2;
if (n > MS_USERINFO_MAX || len - off < n + 2 || memchr(p + off, 0, n)) return -1;
memcpy(w->userinfo, p + off, n); off += n;
size_t tags_len = ((size_t)p[off] << 8) | p[off + 1]; off += 2;
if (tags_len > MS_ADM_TAGS_MAX || len - off < tags_len + 136 || memchr(p + off, 0, tags_len)) return -1;
memcpy(w->tags, p + off, tags_len); off += tags_len;
m->adm_tags_sig = p + off; off += 64;
m->signed_by = member_read64(p + off); off += 8;
m->signature = p + off; off += 64;
m->userinfo = w->userinfo; m->adm_tags = w->tags;
*used = off;
return 0;
}
static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer, const uint8_t* data, size_t len) {
struct UTUN_INSTANCE* inst = ctx;
sqlite3* db = _db(inst);
if (!db || len < 2) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid member page ns=%s bytes=%zu", MS_ID, ns, len); return MT_ERR_DATA;
}
uint16_t count = ((uint16_t)data[0] << 8) | data[1];
if (count > (len - 2) / 285) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: impossible page count=%u bytes=%zu", MS_ID, count, len); return MT_ERR_DATA;
}
size_t off = 2;
for (unsigned i = 0; i < count; i++) {
struct member_wire w; size_t used;
if (member_decode(data + off, len - off, &w, &used) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: malformed member page ns=%s record=%u offset=%zu", MS_ID, ns, i, off);
return MT_ERR_DATA;
}
off += used;
}
if (off != len) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: trailing page bytes ns=%s bytes=%zu", MS_ID, ns, len - off); return MT_ERR_DATA; }
if (!count) return 0;
uint8_t* changed = u_calloc(count, 1);
if (!changed) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page allocation failed", MS_ID); return MT_ERR_IO; }
if (sqlite3_exec(db, "SAVEPOINT member_page", NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page begin: %s", MS_ID, sqlite3_errmsg(db)); u_free(changed); return MT_ERR_IO;
}
int rc = 0;
off = 2;
for (unsigned i = 0; i < count; i++) {
struct member_wire w; size_t used;
if (member_decode(data + off, len - off, &w, &used) < 0) { rc = MT_ERR_DATA; break; }
if (!w.rec.join_sig && topo_node_sqlite_member_get_join(db, ns, w.rec.node_id, w.local_join, &w.rec.join_ts) == 0)
w.rec.join_sig = w.local_join;
int r = member_sync_apply_record(inst, ns, from_peer, &w.rec);
if (r < 0) { rc = r; break; }
changed[i] = r & (MS_APPLY_CHANGED | MS_APPLY_FEEDBACK);
off += used;
}
if (rc < 0 && sqlite3_exec(db, "ROLLBACK TO member_page", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page rollback: %s", MS_ID, sqlite3_errmsg(db));
if (sqlite3_exec(db, "RELEASE member_page", NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page commit: %s", MS_ID, sqlite3_errmsg(db)); rc = MT_ERR_IO;
if (sqlite3_exec(db, "ROLLBACK TO member_page; RELEASE member_page", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page cleanup: %s", MS_ID, sqlite3_errmsg(db));
}
if (rc == 0) {
off = 2;
for (unsigned i = 0; i < count; i++) {
struct member_wire w; size_t used;
if (member_decode(data + off, len - off, &w, &used) < 0) { rc = MT_ERR_DATA; break; }
if (changed[i]) {
member_sync_record_committed(inst, ns, w.rec.node_id, changed[i]);
if (changed[i] & MS_APPLY_CHANGED) {
if (!inst->topo_groups || !topo_groups_find(inst->topo_groups, strtoull(ns, NULL, 10)))
_fire_props_changed(inst, w.rec.node_id, w.tags[0] ? w.tags : NULL, ns);
_fire_apply_cbk(inst, ns, w.rec.node_id, w.rec.x25519, w.rec.signed_by, w.rec.signature);
}
}
off += used;
}
}
u_free(changed);
return rc;
}
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 int member_finish(void* ctx, const char* ns) {
sqlite3* db = _db(ctx);
char table[128], sql[320]; _peers_table(ns, table, sizeof(table));
snprintf(sql, sizeof(sql), "SELECT node_id,x25519_pubkey,signed_by,signature FROM \"%s\" WHERE source=0", table);
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: finish query ns=%s: %s", MS_ID, ns, sqlite3_errmsg(db)); return -1;
}
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
struct ms_member_rec m = {0};
m.node_id = (uint64_t)sqlite3_column_int64(st, 0);
m.x25519 = sqlite3_column_blob(st, 1);
m.signed_by = (uint64_t)sqlite3_column_int64(st, 2);
m.signature = sqlite3_column_blob(st, 3);
if (!m.signature || sqlite3_column_bytes(st, 3) == 0) continue;
if (sqlite3_column_bytes(st, 1) != 32 || sqlite3_column_bytes(st, 3) != 64) { rc = SQLITE_CORRUPT; break; }
if (_sig_is_zero64(m.signature)) continue;
int verified = member_sync_verify_signature(db, ns, &m);
if (verified != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: cannot confirm ns=%s node=%016llx signer=%016llx verification=%d",
MS_ID, ns, (unsigned long long)m.node_id, (unsigned long long)m.signed_by, verified);
rc = SQLITE_CONSTRAINT; break;
}
}
sqlite3_finalize(st);
if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: finish failed ns=%s rc=%d", MS_ID, ns, rc); return -1; }
return 0;
}
static const struct merkle_sync_data_ops g_member_ops = {
.update_bucket_hash = _member_update_bucket_hash,
.get_page = _member_get_page,
.apply_items = _member_apply_items,
.validate_peer = _member_validate_peer,
.finish = member_finish,
};
/* ── Public API ── */
static int _rebuild_all_trees(struct UTUN_INSTANCE* inst) {
sqlite3* db = _db(inst);
sqlite3_stmt* tables = NULL;
int rc = sqlite3_prepare_v2(db, "SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'peers_%'", -1, &tables, NULL);
if (rc != SQLITE_OK) goto fail;
if (sqlite3_exec(db, "SAVEPOINT member_rebuild", NULL, NULL, NULL) != SQLITE_OK) goto fail;
while ((rc = sqlite3_step(tables)) == SQLITE_ROW) {
const char* table = (const char*)sqlite3_column_text(tables, 0);
if (!table || strncmp(table, "peers_", 6)) continue;
const char* ns = table + 6;
uint64_t mask = merkle_sync_level_prefix(UINT64_MAX, MT_MAX_LEVEL);
char sql[256];
snprintf(sql, sizeof(sql), "SELECT DISTINCT (node_id & %lld) FROM \"%s\" WHERE source=0", (long long)mask, table);
sqlite3_stmt* leaves = NULL;
int lr = sqlite3_prepare_v2(db, sql, -1, &leaves, NULL);
if (lr == SQLITE_OK) {
while ((lr = sqlite3_step(leaves)) == SQLITE_ROW) {
uint64_t key = (uint64_t)sqlite3_column_int64(leaves, 0);
if (merkle_sync_recompute_path(inst, MT_DATA_MEMBERS, ns, key) < 0) { lr = SQLITE_ERROR; break; }
}
}
sqlite3_finalize(leaves);
if (lr != SQLITE_DONE) { rc = lr; break; }
}
if (rc != SQLITE_DONE) {
if (sqlite3_exec(db, "ROLLBACK TO member_rebuild", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: rebuild rollback failed: %s", MS_ID, sqlite3_errmsg(db));
}
int release = sqlite3_exec(db, "RELEASE member_rebuild", NULL, NULL, NULL);
sqlite3_finalize(tables); tables = NULL;
if (rc != SQLITE_DONE || release != SQLITE_OK) goto fail;
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: trees rebuilt from distinct leaves", MS_ID);
return 0;
fail:
sqlite3_finalize(tables);
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: tree rebuild failed: %s", MS_ID, db ? sqlite3_errmsg(db) : "no DB");
return -1;
}
/* Подписаться на BGP node-события chat-группы (BGP → member_sync → node_props_changed). */
int member_sync_subscribe_group(struct UTUN_INSTANCE* inst, struct TOPO_GROUP* group) {
if (!inst || !group || group->group_type != TOPO_GROUP_TYPE_CHAT) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "invalid group subscription"); return -1;
}
for (struct topo_node_cbk_entry* cb = group->node_cbks; cb; cb = cb->next)
if (cb->fn == _ms_on_bgp_node && cb->arg == inst) return 0;
topo_group_add_node_cbk(group, _ms_on_bgp_node, inst);
for (struct topo_node_cbk_entry* cb = group->node_cbks; cb; cb = cb->next)
if (cb->fn == _ms_on_bgp_node && cb->arg == inst) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "group subscription allocation failed ch=%s", group->channel_id); return -1;
}
int member_sync_init(struct UTUN_INSTANCE* inst) {
sqlite3* db = _db(inst);
if (!db || !sqlite3_get_autocommit(db)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: init requires database outside transaction", MS_ID); return -1;
}
int rc = merkle_sync_init(inst, ETCP_ID_MEMBER_SYNC, &g_member_ops, inst);
if (rc != 0) return rc;
/* очистка битых записей во всех каналах перед построением дерева */
if (member_sync_verify_and_purge_all(inst) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: initialization stopped after purge failure", MS_ID);
merkle_sync_destroy(inst); return -1;
}
if (_rebuild_all_trees(inst) < 0) { merkle_sync_destroy(inst); return -1; }
/* подписка на BGP-события существующих chat-групп */
if (inst->topo_groups && inst->topo_groups->group_list) {
struct ll_entry* ge = inst->topo_groups->group_list->head;
while (ge) {
struct TOPO_GROUP* g = (struct TOPO_GROUP*)ge;
if (g->group_type == TOPO_GROUP_TYPE_CHAT && g->channel_id[0])
if (member_sync_subscribe_group(inst, g) < 0) { member_sync_destroy(inst); return -1; }
ge = ge->next;
}
}
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: initialized, merkle_rc=%d", MS_ID, rc);
return 0;
}
void member_sync_destroy(struct UTUN_INSTANCE* inst) {
if (!inst) return;
if (inst->topo_groups && inst->topo_groups->group_list) {
struct ll_entry* ge = inst->topo_groups->group_list->head;
while (ge) {
struct TOPO_GROUP* g = (struct TOPO_GROUP*)ge;
if (g->group_type == TOPO_GROUP_TYPE_CHAT && g->channel_id[0])
topo_group_remove_node_cbk(g, _ms_on_bgp_node, inst);
ge = ge->next;
}
}
/* освободить callback-списки (иначе утекут при chat_core_destroy) */
if (CC(inst)) {
struct ms_props_cbk* pc = CC(inst)->props_cbks;
while (pc) { struct ms_props_cbk* n = pc->next; u_free(pc); pc = n; }
CC(inst)->props_cbks = NULL;
struct ms_apply_cbk* ac = CC(inst)->apply_cbks;
while (ac) { struct ms_apply_cbk* n = ac->next; u_free(ac); ac = n; }
CC(inst)->apply_cbks = NULL;
}
merkle_sync_destroy(inst);
}
int member_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
const char* ch_id, merkle_sync_done_cb done_cb, void* arg) {
if (member_sync_verify_and_purge(inst, ch_id) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: start rejected after purge failure ns=%s peer=%016llx",
MS_ID, ch_id ? ch_id : "-", (unsigned long long)peer); return -1;
}
return merkle_sync_start(inst, MT_DATA_MEMBERS, peer, ch_id, done_cb, arg);
}
void member_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id) {
merkle_sync_cancel(inst, MT_DATA_MEMBERS, peer, ch_id);
}
void member_sync_cancel_channel(struct UTUN_INSTANCE* inst, const char* ch_id) {
merkle_sync_cancel_ns(inst, ch_id);
}
int member_sync_put(struct UTUN_INSTANCE* inst, const char* ch_id,
uint64_t member_id, const uint8_t* x25519,
const uint8_t* ed25519,
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,
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 tags_bytes=%zu storage=%d signed_by=%016llx",
MS_ID, ch_id, (unsigned long long)member_id, adm_tags ? strlen(adm_tags) : 0, storage, (unsigned long long)signed_by);
struct ms_member_rec m;
memset(&m, 0, sizeof(m));
m.node_id = member_id;
m.x25519 = x25519; m.ed25519 = ed25519;
m.join_sig = join_sig; m.join_ts = join_ts;
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);
}
int member_sync_count(struct UTUN_INSTANCE* inst, const char* ch_id) {
sqlite3* db = _db(inst);
if (!db || !ch_id) return 0;
char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
char sql[256]; snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM \"%s\"", peers_tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return 0;
int c = 0;
if (sqlite3_step(stmt) == SQLITE_ROW) c = sqlite3_column_int(stmt, 0);
sqlite3_finalize(stmt);
return c;
}
/* Проверяет удостоверение целиком; ожидание зависимости сохраняет возможность позднего догоняния. */
enum ms_author_status member_sync_author_key(struct UTUN_INSTANCE* inst, const char* channel,
uint64_t author, uint8_t key[32]) {
if (!inst || !inst->topo_sqlite_db || !key || !author || !channel || !channel[0]
|| strlen(channel) > 20 || strspn(channel, "0123456789") != strlen(channel)
|| (channel[0] == '0' && channel[1])) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "author verification: invalid arguments"); return MS_AUTHOR_ERROR;
}
char sql[384];
snprintf(sql, sizeof(sql), "SELECT x25519_pubkey,ed25519_pubkey,signed_by,signature,join_sig,join_ts"
" FROM peers_%s WHERE node_id=? AND source=0 AND deleted=0", channel);
sqlite3_stmt* st = NULL; sqlite3* db = inst->topo_sqlite_db;
enum ms_author_status status = MS_AUTHOR_ERROR;
const char* reason = "member lookup";
uint8_t verified_key[32]; int final;
int rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)author);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc == SQLITE_DONE) { status = MS_AUTHOR_INACTIVE; reason = "member inactive"; goto done; }
if (rc != SQLITE_ROW) goto done;
int x_bytes = sqlite3_column_bytes(st, 0), ed_bytes = sqlite3_column_bytes(st, 1);
int tree_bytes = sqlite3_column_bytes(st, 3), join_bytes = sqlite3_column_bytes(st, 4);
if (sqlite3_errcode(db) == SQLITE_NOMEM) { rc = SQLITE_NOMEM; goto done; }
status = MS_AUTHOR_INVALID;
reason = "invalid member fields";
if (x_bytes != 32 || ed_bytes != 32 || (tree_bytes != 0 && tree_bytes != 64)
|| (join_bytes != 0 && join_bytes != 64)) goto done;
struct ms_member_rec member = { .node_id = author, .x25519 = sqlite3_column_blob(st, 0),
.ed25519 = sqlite3_column_blob(st, 1), .signed_by = (uint64_t)sqlite3_column_int64(st, 2),
.signature = sqlite3_column_blob(st, 3), .join_sig = sqlite3_column_blob(st, 4),
.join_ts = (uint64_t)sqlite3_column_int64(st, 5) };
if (sqlite3_errcode(db) == SQLITE_NOMEM || !member.x25519 || !member.ed25519
|| (tree_bytes && !member.signature) || (join_bytes && !member.join_sig)) {
status = MS_AUTHOR_ERROR; reason = "member column allocation"; rc = SQLITE_NOMEM; goto done;
}
if (!tree_bytes || !join_bytes || sqlite3_column_int64(st, 5) <= 0
|| _sig_is_zero64(member.signature) || _sig_is_zero64(member.join_sig)) {
status = MS_AUTHOR_PENDING; reason = "join or invitation pending"; goto done;
}
reason = "node_id/X25519 mismatch";
uint64_t derived = sc_derive_node_id_from_pubkey(member.x25519);
if (!derived) { status = MS_AUTHOR_ERROR; reason = "identity digest"; rc = SC_ERR_CRYPTO; goto done; }
if (derived != author) goto done;
uint8_t trusted[32], x[32], ed[32], body[112];
reason = "trusted Ed25519 key unavailable";
int lookup = topo_node_sqlite_get_ed25519_pubkey(db, author, trusted);
if (lookup != 0) {
status = lookup == TOPO_SQLITE_LOOKUP_MISSING ? MS_AUTHOR_PENDING : MS_AUTHOR_ERROR; rc = lookup; goto done;
}
reason = "trusted Ed25519 key mismatch";
if (memcmp(trusted, member.ed25519, 32)) goto done;
reason = "channel keys unavailable";
lookup = topo_node_sqlite_channel_get(db, channel, NULL, 0, NULL, x, ed, NULL);
if (lookup != 0) {
status = lookup == TOPO_SQLITE_LOOKUP_MISSING ? MS_AUTHOR_PENDING : MS_AUTHOR_ERROR; rc = lookup; goto done;
}
reason = "invalid join signature";
int bytes = member_sync_build_join_msg(x, ed, author, member.x25519, member.join_ts, body, sizeof(body));
if (bytes != 112) { status = MS_AUTHOR_ERROR; reason = "join message build"; goto done; }
int signature = sc_ed25519_verify(member.ed25519, body, bytes, member.join_sig);
if (signature != SC_OK) {
status = signature == SC_ERR_AUTH_FAILED ? MS_AUTHOR_INVALID : MS_AUTHOR_ERROR; rc = signature; goto done;
}
reason = "invitation verification";
signature = member_sync_verify_signature(db, channel, &member);
if (signature != 1) {
status = signature == MS_VERIFY_DEFERRED ? MS_AUTHOR_PENDING
: signature == MS_VERIFY_ERROR ? MS_AUTHOR_ERROR : MS_AUTHOR_INVALID;
rc = signature;
goto done;
}
memcpy(verified_key, member.ed25519, 32); status = MS_AUTHOR_ACTIVE; reason = "verified";
done:
final = sqlite3_finalize(st);
if (final != SQLITE_OK) { status = MS_AUTHOR_ERROR; reason = "member finalize"; rc = final; }
if (status == MS_AUTHOR_ACTIVE) memcpy(key, verified_key, 32);
if (status == MS_AUTHOR_ERROR) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC,
"author verification failed ch=%s node=%016llx reason=%s result=%d",
channel, (unsigned long long)author, reason, rc);
else if (status == MS_AUTHOR_INVALID) DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC,
"author invalid ch=%s node=%016llx reason=%s", channel, (unsigned long long)author, reason);
else if (status != MS_AUTHOR_ACTIVE) DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC,
"author unavailable ch=%s node=%016llx status=%d reason=%s", channel, (unsigned long long)author, status, reason);
return status;
}
/* Строгий допуск radio/attachments: ни ожидание, ни исключение не разрешают действие. */
int member_sync_verified_ed25519(struct UTUN_INSTANCE* inst, const char* channel, uint64_t author, uint8_t key[32]) {
enum ms_author_status status = member_sync_author_key(inst, channel, author, key);
if (status == MS_AUTHOR_ACTIVE) return 0;
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "member verification denied ch=%s node=%016llx status=%d",
channel ? channel : "-", (unsigned long long)author, status); return -1;
}
Reference in new issue View Git Blame Copy Permalink