From 7319f900c47b0c78de0dca17db03bdbd474e6197 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 4 Nov 2026 00:46:27 +0100 Subject: [PATCH] Exercise concurrent member versions, fallback, SQL rollback and restart --- src/chat/member_sync.c | 7 +- tests/Makefile.am | 4 ++ tests/test_member_versions.c | 134 +++++++++++++++++++++++++++++++++++ 3 files changed, 142 insertions(+), 3 deletions(-) create mode 100644 tests/test_member_versions.c diff --git a/src/chat/member_sync.c b/src/chat/member_sync.c index 23a3db38a..00fbaa4a3 100644 --- a/src/chat/member_sync.c +++ b/src/chat/member_sync.c @@ -1613,22 +1613,23 @@ static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer, co static int _member_validate_peer(void* ctx, const char* ns, uint64_t peer) { struct UTUN_INSTANCE* inst = ctx; struct ETCP_CONN* conn = inst ? instance_find_conn(inst, peer) : NULL; - return conn && conn->initialized && conn->links_up && conn->state != 2 && !conn->close_requested && + return ns && conn && conn->initialized && conn->links_up && conn->state != 2 && !conn->close_requested && topo_group_chat_allowed(inst, chat_channel_id(ns), peer) && - topo_node_sqlite_member_key_allowed(_db(inst), ns, conn->peer_pubkey); + topo_node_sqlite_member_key_allowed(_db(inst), ns, conn->crypto_ctx.peer_public_key); } /* Разрешение только спросить: канал выбран локально, ключ источника уже связан с его JOIN. * Никаких страниц/хешей в ответ этот режим не выдаёт; источник сам проверяет наш допуск. */ static int member_validate_request(void* ctx, const char* ns, uint64_t peer) { struct UTUN_INSTANCE* inst = ctx; + if (!ns) return 0; struct TOPO_GROUP* group = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, chat_channel_id(ns)) : NULL; struct ETCP_CONN* conn = inst ? instance_find_conn(inst, peer) : NULL; if (!group || group->group_type != TOPO_GROUP_TYPE_CHAT || group->stopping || strcmp(group->channel_id, ns) || !conn || !conn->initialized || !conn->links_up || conn->state == 2 || conn->close_requested) return 0; struct member_saved record; int found = member_read_saved(_db(inst), ns, peer, 0, 0, &record); - if (found != 1 || record.source || record.malformed || memcmp(record.x, conn->peer_pubkey, 32) || + if (found != 1 || record.source || record.malformed || memcmp(record.x, conn->crypto_ctx.peer_public_key, 32) || sc_derive_node_id_from_pubkey(record.x) != peer || _verify_join_sig(_db(inst), ns, &record.rec) != 1) return 0; uint8_t key[32]; enum ms_author_status status = member_sync_author_key(inst, ns, peer, key); return status == MS_AUTHOR_ACTIVE || status == MS_AUTHOR_PENDING; diff --git a/tests/Makefile.am b/tests/Makefile.am index d928dd59c..66b059821 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -67,6 +67,7 @@ check_PROGRAMS = \ test_multicast \ test_group_ownership \ test_merkle_protocol \ + test_member_versions \ test_chat_sync_schedule \ test_ncd_config \ test_node_snapshot \ @@ -1004,6 +1005,9 @@ test_chat_feedback_LDFLAGS = -Wl,--wrap=sqlite3_exec -Wl,--wrap=sqlite3_prepare_ test_merkle_protocol_SOURCES = test_merkle_protocol.c test_merkle_protocol_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/lib test_merkle_protocol_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) +test_member_versions_SOURCES = test_member_versions.c +test_member_versions_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/lib +test_member_versions_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) test_chat_sync_schedule_SOURCES = test_chat_sync_schedule.c $(test_clock_sources) test_chat_sync_schedule_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/lib test_chat_sync_schedule_LDFLAGS = $(test_clock_ldflags) diff --git a/tests/test_member_versions.c b/tests/test_member_versions.c new file mode 100644 index 000000000..662a4e0f7 --- /dev/null +++ b/tests/test_member_versions.c @@ -0,0 +1,134 @@ +/* Проверенные/новые/старые версии в настоящей SQLite: конкурентные пиры, отказ и отмена. */ +#include "../src/chat/member_sync.c" +#include + +struct identity { uint64_t id; uint8_t x[32], ed[32], secret[32], join[64], tree[64]; }; +static struct identity identities[5]; +static uint8_t group_secret[32] = {91}, group_public[32], group_x[32] = {92}; + +static void public_key(const uint8_t secret[32], uint8_t public[32]) { + EVP_PKEY* key = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL, secret, 32); assert(key); + size_t bytes = 32; assert(EVP_PKEY_get_raw_public_key(key, public, &bytes) == 1); EVP_PKEY_free(key); +} + +static void init_identity(unsigned i) { + struct identity* p = &identities[i]; p->x[0] = i + 11; p->secret[0] = i + 31; + public_key(p->secret, p->ed); p->id = sc_derive_node_id_from_pubkey(p->x); + uint8_t message[112]; assert(member_sync_build_join_msg(group_x, group_public, p->id, p->x, 1, message, sizeof(message)) == 112); + assert(sc_ed25519_sign(p->secret, message, sizeof(message), p->join) == SC_OK); + assert(!member_sync_sign_join(group_secret, p->join, p->tree)); +} + +static void profile(struct UTUN_INSTANCE* inst, unsigned person, uint64_t version) { + struct identity* p = &identities[person]; uint8_t message[256], signature[64]; char info[64]; + snprintf(info, sizeof(info), "{\"name\":\"profile-%llu\"}", (unsigned long long)version); + int bytes = member_sync_build_update_msg(p->join, version, info, message, sizeof(message)); assert(bytes > 0); + assert(sc_ed25519_sign(p->secret, message, bytes, signature) == SC_OK); + assert(member_sync_put(inst, "1001", p->id, p->x, p->ed, p->join, 1, signature, version, info, + NULL, NULL, 0, 0, p->tree) >= 0); +} + +static void setup(struct UTUN_INSTANCE* inst, struct UASYNC* ua, unsigned self) { + memset(inst, 0, sizeof(*inst)); inst->node_id = identities[self].id; inst->ua = ua; + assert(sqlite3_open(":memory:", &inst->topo_sqlite_db) == SQLITE_OK); + assert(!topo_node_sqlite_init(inst->topo_sqlite_db)); + assert(!topo_node_sqlite_channel_put(inst->topo_sqlite_db, "1001", "versions", identities[0].id, + group_x, NULL, group_public, NULL, (uint8_t[64]){0})); + assert(!member_sync_init(inst)); + for (unsigned i = 0; i < 3; i++) profile(inst, i, 1); +} + +static sqlite3_int64 scalar(struct UTUN_INSTANCE* inst, const char* sql) { + sqlite3_stmt* st = NULL; assert(sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) == SQLITE_OK); + assert(sqlite3_step(st) == SQLITE_ROW); sqlite3_int64 value = sqlite3_column_int64(st, 0); + assert(sqlite3_finalize(st) == SQLITE_OK); return value; +} + +static uint64_t version(struct UTUN_INSTANCE* inst) { + struct member_saved record; assert(member_read_saved(inst->topo_sqlite_db, "1001", identities[2].id, 0, 0, &record) == 1); + return record.rec.update_ts; +} + +static struct member_pull* receive(struct UTUN_INSTANCE* source, struct UTUN_INSTANCE* target, unsigned person, uint64_t peer) { + void* state = NULL; assert(!member_begin_pull(target, "1001", peer, 1, &state)); + uint64_t prefix = merkle_sync_level_prefix(identities[person].id, MT_MAX_LEVEL); + uint8_t bytes[MT_PAGE_SIZE]; size_t length = sizeof(bytes); int more; merkle_key_t after = {{0}}, next; + assert(!_member_get_page(source, "1001", prefix, 0, &after, bytes, &length, &next, &more) && !more); + assert(!member_stage_page(target, state, prefix, 0, &after, &next, more, bytes, length)); return state; +} + +static void accept(struct UTUN_INSTANCE* inst, struct member_pull* pull) { + assert(!member_commit_pull(inst, pull)); member_publish_pull(inst, pull); member_abort_pull(inst, pull); + assert(sqlite3_get_autocommit(inst->topo_sqlite_db)); +} + +static void corrupt(struct UTUN_INSTANCE* inst) { + char sql[192]; snprintf(sql, sizeof(sql), "UPDATE peers_1001 SET update_sig=zeroblob(64)||X'01' WHERE node_id=%lld AND record_state=0", + (long long)identities[2].id); + assert(sqlite3_exec(inst->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK); +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_INFO); + size_t baseline = u_get_allocated_count(); struct UASYNC* ua = uasync_create(); assert(ua); + public_key(group_secret, group_public); + for (unsigned i = 0; i < 5; i++) init_identity(i); + struct UTUN_INSTANCE source, target; setup(&source, ua, 1); setup(&target, ua, 0); + uint8_t original[32], staged[32]; assert(!merkle_sync_read_hash(&target, MT_DATA_MEMBERS, "1001", 0, 0, original)); + profile(&source, 2, 2); + struct member_pull* a = receive(&source, &target, 2, identities[1].id); + assert(version(&target) == 1 && member_sync_count(&target, "1001") == 3); + assert(!merkle_sync_read_hash(&target, MT_DATA_MEMBERS, "1001", 0, 0, staged) && !memcmp(original, staged, 32)); + assert(scalar(&target, "SELECT COUNT(*) FROM nodes WHERE name='profile-2'") == 0); + profile(&source, 2, 3); + struct member_pull* b = receive(&source, &target, 2, identities[3].id); + assert(a->id != b->id && scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state=1") == 4); + accept(&target, b); assert(version(&target) == 3); + assert(scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state=1") == 2); + accept(&target, a); assert(version(&target) == 3); + assert(!scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state<>0")); + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS two peers: isolated staging, no early profile/hash change, stale completion preserves newer commit"); + + profile(&source, 2, 4); a = receive(&source, &target, 2, identities[1].id); b = receive(&source, &target, 2, identities[3].id); + member_abort_pull(&target, a); + assert(scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state=1") == 2); + accept(&target, b); assert(version(&target) == 4); + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS cancelling one peer leaves the other exchange intact"); + + profile(&source, 2, 5); + char sql[256]; snprintf(sql, sizeof(sql), "UPDATE peers_1001 SET update_sig=randomblob(64) WHERE node_id=%lld AND record_state=0", + (long long)identities[2].id); + assert(sqlite3_exec(source.topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK); + a = receive(&source, &target, 2, identities[1].id); accept(&target, a); assert(version(&target) == 4); + assert(sqlite3_exec(target.topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK); + a = receive(&source, &target, 2, identities[1].id); accept(&target, a); + assert(!topo_node_sqlite_member_in_channel(target.topo_sqlite_db, "1001", identities[2].id)); + assert(member_sync_count(&target, "1001") == 2 && topo_node_sqlite_member_allowed(target.topo_sqlite_db, "1001", identities[0].id)); + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS bad new keeps good old; both invalid removes only that member"); + + profile(&source, 2, 6); a = receive(&source, &target, 2, identities[1].id); accept(&target, a); assert(version(&target) == 6); + profile(&source, 2, 7); a = receive(&source, &target, 2, identities[1].id); + snprintf(sql, sizeof(sql), "CREATE TRIGGER fail_identity BEFORE INSERT ON nodes WHEN NEW.node_id=%lld" + " BEGIN SELECT RAISE(ABORT,'injected identity failure'); END", (long long)identities[2].id); + assert(sqlite3_exec(target.topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK); + assert(member_commit_pull(&target, a) < 0 && version(&target) == 6); + member_publish_pull(&target, a); member_abort_pull(&target, a); + assert(!scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state<>0")); + assert(sqlite3_exec(target.topo_sqlite_db, "DROP TRIGGER fail_identity", NULL, NULL, NULL) == SQLITE_OK); + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS SQL failure rolls back versions, identities and admission together"); + + a = receive(&source, &target, 2, identities[1].id); sqlite3_int64 previous = a->id; + member_sync_destroy(&target); assert(!member_sync_init(&target)); + assert(version(&target) == 6 && !scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state<>0")); + b = receive(&source, &target, 2, identities[3].id); assert(b->id > previous); + member_abort_pull(&target, a); accept(&target, b); assert(version(&target) == 7); + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS restart removes abandoned versions without reusing exchange identifiers"); + + corrupt(&target); profile(&source, 2, 8); + a = receive(&source, &target, 2, identities[1].id); accept(&target, a); assert(version(&target) == 8); + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS valid new version repairs a malformed previous version"); + member_sync_destroy(&source); member_sync_destroy(&target); + assert(sqlite3_close(source.topo_sqlite_db) == SQLITE_OK && sqlite3_close(target.topo_sqlite_db) == SQLITE_OK); + uasync_destroy(ua, 0); assert(u_get_allocated_count() == baseline); + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "ALL PASS member version isolation and recovery"); return 0; +}