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

Make member cleanup atomic and publish removals after commit

Browse Source
This commit is contained in:
evgeny committed 2026-10-26 14:42:59 +01:00
1 parent 1f3f1d9a34
commit fbe3e7a279
3 files changed
+279 -167
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/chat_core_priv.h
+1
View File
@@ -90,6 +90,7 @@ struct chat_core_ctx {
uint8_t initialized;
/* per-instance sub-module state (was file-scope statics) */
struct ms_props_cbk* props_cbks; /* was member_sync.c g_props_cbks */
unsigned props_dispatch_depth; /* снятие props-подписки откладывает free до выхода из вложенных callbacks */
struct ms_apply_cbk* apply_cbks; /* was member_sync.c g_apply_cbks */
struct group_avatar* group_avatar;
uint64_t media_cache_bytes; /* was chat_msg.c g_media_cache_bytes */
src/chat/member_sync.c
+239 -150
View File
@@ -13,6 +13,8 @@
#include <string.h>
#include <stdlib.h>
#include <errno.h>
#include <limits.h>
#include <sqlite3.h>
#include <openssl/evp.h>
#include "../../lib/json_flat.h"
@@ -313,8 +315,17 @@ struct ms_props_cbk { node_props_changed_fn fn; void* arg; struct ms_props_cbk*
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;
struct ms_props_cbk* pc = CC(inst)->props_cbks;
while (pc) { pc->fn(node_id, adm_tags, channel_id, pc->arg); pc = pc->next; }
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_id, 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) {
@@ -333,7 +344,8 @@ void member_sync_remove_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed
while (*p) {
if ((*p)->fn == fn && (*p)->arg == arg) {
struct ms_props_cbk* r = *p;
*p = r->next; u_free(r);
if (CC(inst)->props_dispatch_depth) { r->fn = NULL; r->arg = NULL; }
else { *p = r->next; u_free(r); }
return;
}
p = &(*p)->next;
@@ -413,24 +425,21 @@ static int _sig_is_zero64(const uint8_t* sig) {
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];
if (topo_node_sqlite_channel_get(db, ch_id, NULL, 0, NULL, ch_x, ch_ed, NULL) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: verify_join_sig — no channel key ns=%s", MS_ID, ch_id);
return 0;
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) return 0;
EVP_PKEY* pk = EVP_PKEY_new_raw_public_key(EVP_PKEY_ED25519, NULL, m->ed25519, 32);
if (!pk) return 0;
EVP_MD_CTX* vctx = EVP_MD_CTX_new();
int ok = vctx && (EVP_DigestVerifyInit(vctx, NULL, NULL, NULL, pk) == 1)
&& (EVP_DigestVerify(vctx, m->join_sig, 64, vmsg, (size_t)vlen) == 1);
if (vctx) EVP_MD_CTX_free(vctx);
EVP_PKEY_free(pk);
if (!ok) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record invalid join_sig node=0x%016llx ns=%s", MS_ID, (unsigned long long)m->node_id, ch_id);
return ok ? 1 : 0;
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 узла ── */
@@ -454,28 +463,32 @@ int member_sync_sign_pubkey(const uint8_t* signer_ed_priv, const uint8_t* x25519
}
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;
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) 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;
}
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;
}
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;
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,
@@ -560,6 +573,7 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
/* подпись дерева приглашений: -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);
@@ -581,23 +595,20 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
int vlen = member_sync_build_update_msg(jsig, m->update_ts,
m->userinfo ? m->userinfo : "", vmsg, (int)sizeof(vmsg));
if (vlen >= 0) {
EVP_PKEY* pkey = EVP_PKEY_new_raw_public_key(EVP_PKEY_ED25519, NULL, m->ed25519, 32);
if (pkey) {
EVP_MD_CTX* ver = EVP_MD_CTX_new();
if (ver) {
int ok = (EVP_DigestVerifyInit(ver, NULL, NULL, NULL, pkey) == 1)
&& (EVP_DigestVerify(ver, m->update_sig, 64, vmsg, (size_t)vlen) == 1);
if (ok) block_a_ok = 1;
else DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record invalid update_sig node=0x%016llx ns=%s", MS_ID, (unsigned long long)m->node_id, ch_id);
EVP_MD_CTX_free(ver);
}
EVP_PKEY_free(pkey);
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 напрямую */
block_a_ok = _verify_join_sig(db, ch_id, m);
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;
@@ -622,24 +633,21 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
adm_ver = json_flat_get(m->adm_tags, "ver", ver_str, sizeof(ver_str)) == 0 ? atoi(ver_str) : 0;
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};
if (topo_node_sqlite_channel_get(db, ch_id, NULL, 0, NULL, NULL, ch_ed_pub, NULL) == 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;
EVP_PKEY* pkey = EVP_PKEY_new_raw_public_key(EVP_PKEY_ED25519, NULL, ch_ed_pub, 32);
if (pkey) {
EVP_MD_CTX* vctx = EVP_MD_CTX_new();
if (vctx) {
int ok = (EVP_DigestVerifyInit(vctx, NULL, NULL, NULL, pkey) == 1)
&& (EVP_DigestVerify(vctx, m->adm_tags_sig, 64, amsg, aoff) == 1);
if (!ok) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: apply_record invalid adm_tags_sig node=0x%016llx ns=%s bytes=%zu", MS_ID, (unsigned long long)m->node_id, ch_id, strlen(m->adm_tags));
else block_b_ok = 1;
EVP_MD_CTX_free(vctx);
}
EVP_PKEY_free(pkey);
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) {
@@ -773,10 +781,17 @@ int member_sync_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
либо 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 || !m->x25519 || !m->ed25519) return 1; /* нет данных — не битая */
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; /* неполная запись — не доказательство порчи */
/* жёсткая привязка идентичности */
if (sc_derive_node_id_from_pubkey(m->x25519) != m->node_id) {
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;
@@ -787,43 +802,40 @@ int member_sync_verify_local_record(struct sqlite3* db, const char* ch_id, const
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 || !m->join_sig) return 0;
EVP_PKEY* pkey = EVP_PKEY_new_raw_public_key(EVP_PKEY_ED25519, NULL, m->ed25519, 32);
if (!pkey) return 0;
EVP_MD_CTX* ver = EVP_MD_CTX_new();
int ok = ver && (EVP_DigestVerifyInit(ver, NULL, NULL, NULL, pkey) == 1)
&& (EVP_DigestVerify(ver, m->update_sig, 64, vmsg, (size_t)vlen) == 1);
if (ver) EVP_MD_CTX_free(ver);
EVP_PKEY_free(pkey);
if (!ok) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: verify_local — invalid update_sig nid=0x%016llx ns=%s",
MS_ID, (unsigned long long)m->node_id, ch_id);
return 0;
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 напрямую */
if (!_verify_join_sig(db, ch_id, m)) return 0;
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};
if (topo_node_sqlite_channel_get(db, ch_id, NULL, 0, NULL, NULL, ch_ed_pub, NULL) != 0)
return 1; /* нет канального ключа — не можем проверить, не удаляем */
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;
EVP_PKEY* pkey = EVP_PKEY_new_raw_public_key(EVP_PKEY_ED25519, NULL, ch_ed_pub, 32);
if (!pkey) return 1;
EVP_MD_CTX* vctx = EVP_MD_CTX_new();
int ok = vctx && (EVP_DigestVerifyInit(vctx, NULL, NULL, NULL, pkey) == 1)
&& (EVP_DigestVerify(vctx, m->adm_tags_sig, 64, amsg, aoff) == 1);
if (vctx) EVP_MD_CTX_free(vctx);
EVP_PKEY_free(pkey);
if (!ok) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: verify_local — invalid adm_tags_sig nid=0x%016llx ns=%s bytes=%zu",
MS_ID, (unsigned long long)m->node_id, ch_id, strlen(m->adm_tags));
return 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);
}
}
@@ -831,6 +843,7 @@ int member_sync_verify_local_record(struct sqlite3* db, const char* ch_id, const
-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);
@@ -840,22 +853,37 @@ int member_sync_verify_local_record(struct sqlite3* db, const char* ch_id, const
return 1;
}
/* Прогоняет все записи канала: битая → удалить + пересчитать хеши. */
/* Владеет транзакцией канала; публикует удаления только после commit данных и индекса. */
int member_sync_verify_and_purge(struct UTUN_INSTANCE* inst, const char* ch_id) {
if (!inst || !ch_id) return 0;
sqlite3* db = _db(inst); if (!db) return 0;
char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
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, source"
" FROM \"%s\" ORDER BY node_id ASC", peers_tbl);
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) return 0;
int purged = 0;
while (sqlite3_step(st) == SQLITE_ROW) {
" 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);
@@ -868,59 +896,111 @@ int member_sync_verify_and_purge(struct UTUN_INSTANCE* inst, const char* ch_id)
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);
int src = sqlite3_column_int(st, 12);
if (!x25 || !ed) continue;
if (src != 0) continue; /* плейсхолдер topo_group — не верифицируем */
struct ms_member_rec m;
memset(&m, 0, sizeof(m));
m.node_id = nid;
m.x25519 = x25; m.ed25519 = ed;
m.join_sig = jsig; m.join_ts = jts;
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",
MS_ID, (unsigned long long)nid, ch_id);
topo_node_sqlite_member_del(db, ch_id, nid);
radio_forget_member_mute(inst, strtoull(ch_id, NULL, 10), nid);
merkle_sync_recompute_path(inst, MT_DATA_MEMBERS, ch_id, nid);
_fire_props_changed(inst, nid, NULL, ch_id);
/* пассивное удаление: если узел подключён напрямую — убрать из CHAT-группы */
if (inst->topo_groups) {
uint64_t gid = strtoull(ch_id, NULL, 10);
struct TOPO_GROUP* g = topo_groups_find(inst->topo_groups, gid);
if (g && g->group_type == TOPO_GROUP_TYPE_CHAT) {
struct ETCP_CONN* c = instance_find_conn(inst, nid);
if (c) topo_group_remove_conn(g, c, TOPO_REMOVE_MEMBER_INVALID);
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;
}
purged++;
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);
}
}
sqlite3_finalize(st);
if (purged > 0)
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: verify_and_purge — purged %d broken records ns=%s",
MS_ID, purged, ch_id);
return purged;
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) 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];
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;
}
/* Прогнать очистку битых записей по всем каналам. */
void member_sync_verify_and_purge_all(struct UTUN_INSTANCE* inst) {
sqlite3* db = _db(inst); if (!db) return;
/* Обходит каналы без живого 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;
if (sqlite3_prepare_v2(db,
"SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'peers_%'",
-1, &st, NULL) != SQLITE_OK) return;
while (sqlite3_step(st) == SQLITE_ROW) {
const char* tbl = (const char*)sqlite3_column_text(st, 0);
if (!tbl || strncmp(tbl, "peers_", 6) != 0) continue;
member_sync_verify_and_purge(inst, tbl + 6);
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;
}
sqlite3_finalize(st);
}
struct member_wire {
@@ -1122,11 +1202,17 @@ int member_sync_subscribe_group(struct UTUN_INSTANCE* inst, struct TOPO_GROUP* g
}
int member_sync_init(struct UTUN_INSTANCE* inst) {
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: init — inst is NULL", MS_ID); return -1; }
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;
/* очистка битых записей во всех каналах перед построением дерева */
member_sync_verify_and_purge_all(inst);
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) {
@@ -1167,7 +1253,10 @@ void member_sync_destroy(struct UTUN_INSTANCE* inst) {
int member_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
const char* ch_id, merkle_sync_done_cb done_cb, void* arg) {
member_sync_verify_and_purge(inst, ch_id);
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);
}
src/chat/member_sync.h
+39 -17
View File
@@ -1,7 +1,8 @@
/*
* member_sync — подписанные записи участников каналов и их обмен через merkle_sync.
* Запись содержит блок участника, блок владельца и подпись дерева приглашений.
* Версии блоков сравниваются отдельно; запись и Merkle-хеш меняются атомарно.
* Версии блоков сравниваются отдельно; apply_record/put меняют запись и Merkle-хеш атомарно.
* Очистка повреждённых записей владеет транзакцией канала и публикует удаления после commit.
* Локальный put запускает распространение после commit; start ждёт сверки с пиром.
* Участие в CHAT-группе определяется локальной таблицей peers_<channel_id>.
* Сетевые адреса принадлежат topo_node и не входят в блок участника.
@@ -24,12 +25,17 @@ struct UTUN_INSTANCE;
struct TOPO_GROUP;
struct sqlite3;
/* Результат применения рекорда мембера (compare+update по версиям двух блоков). */
/* Полная проверка членства, join и дерева приглашений; возвращает доверенный ключ, 0/-1. */
/* Проверить source=0/deleted=0, привязку node_id/X25519 и доверенный Ed25519,
* join_sig и подпись непосредственного пригласившего (либо ключом канала для signed_by=0).
* Не обходит цепочку предков. inst/channel/key обязательны, channel — канонический uint64 decimal.
* 0 — ключ записан в key[32], -1 — отказ, key не меняется. */
int member_sync_verified_ed25519(struct UTUN_INSTANCE* inst, const char* channel, uint64_t author, uint8_t key[32]);
/* Результат применения рекорда мембера (compare+update по версиям двух блоков). */
#define MS_APPLY_CHANGED 0x01 /* запись и дерево обновлены */
#define MS_APPLY_STALE 0x02 /* локальная версия новее; обратный проход доставит её пиру */
#define MS_VERIFY_DEFERRED (-1) /* зависимость подписи ещё не получена */
#define MS_VERIFY_ERROR (-2) /* SQL/crypto/аргументы: проверка не выполнена */
/* UTF-8 JSON bytes, excluding NUL. A complete record fits in one Merkle page. */
#define MS_ADM_TAGS_MAX 8192
@@ -50,7 +56,7 @@ struct ms_member_rec {
const char* userinfo; /* JSON {"name":...}, может быть NULL */
const char* adm_tags; /* JSON владельца с "ver", может быть NULL */
const uint8_t* adm_tags_sig; /* 64 байта, может быть NULL */
int storage; /* 0/1, производная от adm_tags.storage */
int storage; /* вход не используется: сохраняемое значение выводится из adm_tags.storage="yes" */
uint64_t signed_by; /* дерево приглашений: 0 = корень (подписан ключом группы), иначе node_id инвайтера */
const uint8_t* signature; /* 64 байта, Ed25519 подпись x25519||"group_join" ключом родителя (или канальным ed25519) */
};
@@ -69,12 +75,13 @@ int member_sync_sign_pubkey(const uint8_t* signer_ed_priv, const uint8_t* x2551
/* Верифицировать подпись дерева мембера:
* signed_by==0 → ключ канала (ch_ed25519), иначе ed25519 pubkey узла signed_by.
* Возвращает 1 если подпись валидна, 0 если невалидна, -1 если подписанта нет в БД
* (родитель ещё не пришёл по merkle — верификацию откладываем). */
* Возвращает 1 если подпись валидна, 0 если невалидна, MS_VERIFY_DEFERRED если нет ключа,
* MS_VERIFY_ERROR при отказе SQL/crypto. Отсутствие ключа не доказывает порчу записи. */
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. */
/* Проверить наличие derive(x25519) в peers_<ch_id>; 1/0 (ошибки тоже дают 0).
* Не проверяет source, deleted, подписи или совпадение сохранённого ключа.
* Для допуска в CHAT-группу используется отдельный topo_node_sqlite_member_allowed. */
int member_sync_validate_pubkey(struct UTUN_INSTANCE* inst, const char* ch_id,
const uint8_t* x25519_pubkey);
@@ -103,7 +110,9 @@ int member_sync_subscribe_group(struct UTUN_INSTANCE* inst, struct TOPO_GROUP* g
/*
* Запустить синхронизацию мемберов канала ch_id с пиром peer.
* Делегирует в merkle_sync_start().
* Сначала вызывает verify_and_purge, затем merkle_sync_start(MEMBERS).
* 0 — запрос принят; <0 — не принят, done_cb не вызывается (см. merkle_sync.h).
* Ошибка очистки отклоняет запрос до merkle_sync_start, без done callback.
*/
int member_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
const char* ch_id, merkle_sync_done_cb done_cb, void* arg);
@@ -176,10 +185,16 @@ int member_sync_build_join_msg(const uint8_t* ch_x25519, const uint8_t* ch_ed25
* Применить рекорд мембера с по-блочным сравнением версий:
* - блок A (мембер): ver = update_ts (внутри эпохи join_ts), подписан ключом мембера;
* - блок B (владелец): ver = adm_tags.ver, подписан канальным ключом (adm_tags_sig).
* Каждый блок обновляется независимо, только если его ver увеличился.
* Блок A сравнивает сначала join_ts, затем update_ts; новая эпоха может снизить update_ts.
* Блок B обновляется только при увеличении adm_tags.ver (разбирается как int).
* Отсутствующее подтверждение приглашения может дополниться без увеличения версий.
* from_peer == inst->node_id отключает проверки идентичности/подписей блока A и дерева;
* подпись непустого блока B проверяется также для локальной записи.
* Если ver пришедшего блока меньше локального — блок игнорируется и выставляется
* MS_APPLY_STALE (локальная версия новее; её доставит обратный проход).
* Возвращает битовую маску MS_APPLY_*, <0 — ошибка.
* При внешней транзакции caller после commit вызывает member_sync_record_committed
* для изменённой записи; до commit автоматических уведомлений нет.
* Возвращает битовую маску MS_APPLY_* (0 — без изменения индекса), <0 — ошибка.
*/
int member_sync_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
uint64_t from_peer, const struct ms_member_rec* m);
@@ -188,18 +203,24 @@ int member_sync_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
* Верифицировать локальную запись мембера (чтение из БД). Единственный сценарий
* удаления: подпись присутствует и невалидна, либо node_id != derive(x25519).
* Отсутствие подписи = «не верифицировано» (self-add/плейсхолдер) — не битая.
* Возвращает 1 если запись валидна/не подписана, 0 если битая.
* Возвращает 1 если запись сохраняется (валидна/не подписана/ждёт зависимость),
* 0 если доказанно битая, -1 при операционной ошибке. Только 0 разрешает удаление.
*/
int member_sync_verify_local_record(struct sqlite3* db, const char* ch_id,
const struct ms_member_rec* m);
/* Прогнать все записи канала: битую удалить + пересчитать хеши. Возвращает число удалённых. */
/* Проверить source=0 записи канала и атомарно удалить доказанно повреждённые вместе с их индексом.
* Требуются initialized member_sync, общая БД в autocommit и канонический uint64 decimal ch_id.
* Чужая транзакция отклоняется до изменений; недостающие ключи/зависимости сохраняются.
* >=0 — число удалений после commit, -1 — отказ; до commit уведомления/mute/группа не меняются.
* Props callbacks можно снимать/добавлять и вызывать очистку повторно; stop/destroy — вне callbacks. */
int member_sync_verify_and_purge(struct UTUN_INSTANCE* inst, const char* ch_id);
/* Прогнать очистку битых записей по всем каналам (при init). */
void member_sync_verify_and_purge_all(struct UTUN_INSTANCE* inst);
/* Очистить все peers_*: 0/-1, остановиться на первом отказе. Атомарность — по каналу;
* ранее завершённые каналы сохраняют результат. Не держит SQLite cursor через callbacks. */
int member_sync_verify_and_purge_all(struct UTUN_INSTANCE* inst);
/* Количество мемберов в канале. */
/* COUNT(*) всех строк, включая placeholders/deleted; неверный аргумент/ошибка SQL дают 0. */
int member_sync_count(struct UTUN_INSTANCE* inst, const char* ch_id);
/*
@@ -213,7 +234,8 @@ typedef void (*node_props_changed_fn)(uint64_t node_id, const char* adm_tags, co
int member_sync_add_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed_fn fn, void* arg);
void member_sync_remove_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed_fn fn, void* arg);
/* Опубликовать изменения после внешнего commit; uasync, БД уже вне транзакции. */
/* После внешнего commit уведомить Merkle, свойства и планировщик CHAT-группы.
* inst/channel обязательны; uasync, БД уже вне транзакции. Не вызывает apply callbacks. */
void member_sync_record_committed(struct UTUN_INSTANCE* inst, const char* channel, uint64_t node_id);
#ifdef __cplusplus