From fbe3e7a27953355821d64a1ed8a9ab3e3f76601d Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 12:29:00 +0100 Subject: [PATCH] Make member cleanup atomic and publish removals after commit --- src/chat/chat_core_priv.h | 1 + src/chat/member_sync.c | 389 +++++++++++++++++++++++--------------- src/chat/member_sync.h | 56 ++++-- 3 files changed, 279 insertions(+), 167 deletions(-) diff --git a/src/chat/chat_core_priv.h b/src/chat/chat_core_priv.h index bc45b2ad..3e94de5b 100644 --- a/src/chat/chat_core_priv.h +++ b/src/chat/chat_core_priv.h @@ -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 */ diff --git a/src/chat/member_sync.c b/src/chat/member_sync.c index 9f0bc6b0..11f48915 100644 --- a/src/chat/member_sync.c +++ b/src/chat/member_sync.c @@ -13,6 +13,8 @@ #include #include +#include +#include #include #include #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); } diff --git a/src/chat/member_sync.h b/src/chat/member_sync.h index 0da4adba..fe4fbcd6 100644 --- a/src/chat/member_sync.h +++ b/src/chat/member_sync.h @@ -1,7 +1,8 @@ /* * member_sync — подписанные записи участников каналов и их обмен через merkle_sync. * Запись содержит блок участника, блок владельца и подпись дерева приглашений. - * Версии блоков сравниваются отдельно; запись и Merkle-хеш меняются атомарно. + * Версии блоков сравниваются отдельно; apply_record/put меняют запись и Merkle-хеш атомарно. + * Очистка повреждённых записей владеет транзакцией канала и публикует удаления после commit. * Локальный put запускает распространение после commit; start ждёт сверки с пиром. * Участие в CHAT-группе определяется локальной таблицей peers_. * Сетевые адреса принадлежат 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_. Возвращает 1/0. */ +/* Проверить наличие derive(x25519) в peers_; 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