Check cleanup retries, request ownership and stream rollback

src/chat/member_sync.c

@@ -559,13 +559,18 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
     int has_signature = m->signature && !_sig_is_zero64(m->signature);
     if (!is_local) {
         /* жёсткая привязка идентичности: node_id = derive(x25519), ed25519 — доверенный из nodes */
-        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: 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];
-            if (topo_node_sqlite_get_ed25519_pubkey(db, m->node_id, trusted_ed) == 0
-                    && memcmp(m->ed25519, trusted_ed, 32) != 0) {
+            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;
             }
@@ -670,8 +675,12 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
     }
 
     /* Подпись приглашения не зависит от версии профиля: подтверждаем уже сохранённого мембера. */
-    int signature_changed = has_local && has_signature && _sig_is_zero64(local_signature)
-        && member_sync_verify_signature(db, ch_id, m) == 1;
+    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) {

tests/test_member_purge.c

@@ -226,7 +226,7 @@ static void test_purge_dependencies(struct UASYNC* ua) {
     step_fault = "SELECT ed25519_pubkey FROM nodes"; expect_purge_error(&f, before, 1);
     finalize_fault = "SELECT ed25519_pubkey FROM nodes"; expect_purge_error(&f, before, 1);
     prepare_fault = "FROM channels"; expect_purge_error(&f, before, 1);
-    assert(topo_node_sqlite_member_add_placeholder(f.cc.db, "1001", 321, f.x, f.pub) == 0);
+    assert(topo_node_sqlite_member_placeholder_put(f.cc.db, "1001", 321, f.x, f.pub) == 0);
     purge_bad(&f, 123);
     assert(sqlite3_exec(f.cc.db, "BEGIN", NULL, NULL, NULL) == SQLITE_OK);
     assert(member_sync_verify_and_purge(&f.inst, "1001") < 0);
@@ -263,10 +263,59 @@ static void test_purge_init_and_growth(struct UASYNC* ua) {
     DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS purge: list growth failure; purge-all read error; init rollback/cleanup/retry");
 }
 
+static void purge_done(uint64_t peer, const char* channel, int result, void* arg) {
+    (void)peer; (void)channel; (void)result;
+    (*(int*)arg)++;
+}
+
+static void test_purge_start(struct UASYNC* ua) {
+    struct purge_fixture f; purge_setup(&f, ua, 0); purge_bad(&f, 123);
+    f.inst.connections = queue_new(ua, 16, 0, 8, "purge test connections"); assert(f.inst.connections);
+    f.conn.instance = &f.inst; f.conn.peer_node_id = 2; f.conn.initialized = 1; f.conn.links_up = 1;
+    f.conn.send_input_q = queue_new(ua, 0, 0, 0, "purge test send"); assert(f.conn.send_input_q);
+    queue_set_waiter_defer(f.conn.send_input_q, 1);
+    struct ll_entry* e = queue_entry_new(sizeof(struct conn_queue_entry)); assert(e);
+    struct conn_queue_entry* entry = (struct conn_queue_entry*)e->data; entry->peer_node_id = 2; entry->conn = &f.conn;
+    assert(queue_data_put_with_index(f.inst.connections, e) == 0);
+    int callbacks = 0;
+    exec_fault = "COMMIT";
+    assert(member_sync_start(&f.inst, 2, "1001", purge_done, &callbacks) < 0);
+    reset_faults();
+    assert(!f.conn.ref_count && !callbacks && member_sync_count(&f.inst, "1001") == 2);
+    assert(member_sync_start(&f.inst, 2, "1001", purge_done, &callbacks) == 0 && f.conn.ref_count == 1);
+    merkle_sync_cancel_peer(&f.inst, 2);
+    assert(!f.conn.ref_count && !callbacks && !f.conn.send_input_q->waiter_head);
+    e = queue_data_get(f.inst.connections); assert(e); queue_entry_free(e);
+    queue_free(f.inst.connections); f.inst.connections = NULL;
+    queue_free(f.conn.send_input_q);
+    purge_destroy(&f);
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS purge: failed start creates no request/callback; retry attaches real session; cancel releases ownership");
+}
+
+/* Настоящие неверные подписи удаляются; короткий BLOB никогда не читается как полный ключ. */
+static void test_purge_invalid_records(struct UASYNC* ua) {
+    const char* fields[] = { "join_sig", "update_sig", "adm_tags_sig", "signature", "x25519_pubkey" };
+    for (unsigned i = 0; i < sizeof(fields) / sizeof(fields[0]); i++) {
+        struct purge_fixture f; purge_setup(&f, ua, i == 0);
+        uint8_t bad[64]; memset(bad, 1, sizeof(bad));
+        char sql[128]; snprintf(sql, sizeof(sql), "UPDATE peers_1001 SET %s=?", fields[i]);
+        sqlite3_stmt* st = NULL; assert(sqlite3_prepare_v2(f.cc.db, sql, -1, &st, NULL) == SQLITE_OK);
+        sqlite3_bind_blob(st, 1, bad, i == 4 ? 1 : 64, SQLITE_STATIC);
+        assert(sqlite3_step(st) == SQLITE_DONE); sqlite3_finalize(st);
+        assert(member_sync_add_props_cbk(&f.inst, purge_event, &f) == 0);
+        assert(member_sync_verify_and_purge(&f.inst, "1001") == 1 && f.events == 1);
+        uint8_t hash[32], empty[32] = {0}; purge_root(&f, hash);
+        assert(!member_sync_count(&f.inst, "1001") && !memcmp(hash, empty, 32));
+        purge_destroy(&f);
+    }
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS purge: invalid join/update/owner/invitation and malformed key removed with empty root");
+}
+
 void test_member_purge(struct UASYNC* ua) {
     size_t baseline = u_get_allocated_count(); reset_faults();
     debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
     test_purge_transactions(ua); test_purge_crypto(ua); test_purge_dependencies(ua); test_purge_init_and_growth(ua);
+    test_purge_start(ua); test_purge_invalid_records(ua);
     assert(u_get_allocated_count() == baseline);
     DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "ALL PASS: atomic purge contracts; no tracked allocation leaks");
     debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_NONE);

tests/test_stream_service.c

@@ -2499,6 +2499,20 @@ int main(int argc, char **argv) {
     sqlite3_bind_blob(broken, 1, bad_signature, sizeof(bad_signature), SQLITE_STATIC);
     sqlite3_bind_int64(broken, 2, (sqlite3_int64)nodes[1]->node_id);
     CHECK(sqlite3_step(broken) == SQLITE_DONE); sqlite3_finalize(broken);
+    uint8_t purge_root_before[32], purge_root_after[32];
+    CHECK(merkle_sync_read_hash(nodes[1], MT_DATA_MEMBERS, "710002", 0, 0, purge_root_before) == 0);
+    int purge_count_before = member_sync_count(nodes[1], "710002");
+    for (int fault = 0; fault < 2; fault++) {
+        const char* trigger = fault == 0
+            ? "CREATE TRIGGER reject_purge BEFORE DELETE ON peers_710002 BEGIN SELECT RAISE(ABORT,'stream purge delete'); END"
+            : "CREATE TRIGGER reject_purge BEFORE DELETE ON merkle_tree_hash BEGIN SELECT RAISE(ABORT,'stream purge index'); END";
+        CHECK(sqlite3_exec(nodes[1]->topo_sqlite_db, trigger, NULL, NULL, NULL) == SQLITE_OK);
+        CHECK(member_sync_verify_and_purge(nodes[1], "710002") < 0);
+        CHECK(stream_handle_active(other_group) && member_sync_count(nodes[1], "710002") == purge_count_before);
+        CHECK(merkle_sync_read_hash(nodes[1], MT_DATA_MEMBERS, "710002", 0, 0, purge_root_after) == 0);
+        CHECK(!memcmp(purge_root_before, purge_root_after, sizeof(purge_root_before)));
+        CHECK(sqlite3_exec(nodes[1]->topo_sqlite_db, "DROP TRIGGER reject_purge", NULL, NULL, NULL) == SQLITE_OK);
+    }
     CHECK(member_sync_verify_and_purge(nodes[1], "710002") == 1);
     CHECK(!stream_handle_active(other_group) && stream_handle_reason(other_group) == STREAM_CLOSE_MEMBER);
     run_until(&closed_events[1], 2); stream_close(other_group); other_group = NULL;