Validate media role changes through signed CHAT membership

tests/test_media_delivery_chat.c

@@ -216,16 +216,6 @@ static void ensure_supernode_type(int idx) {
     sqlite3_exec(g_inst[idx]->topo_sqlite_db, sql, NULL, NULL, NULL);
 }
 
-/* прямой send media-пакета через CHAT-группу (доставка идёт по группе канала) */
-static int msend(struct UTUN_INSTANCE* inst, uint64_t dst, const uint8_t* data, size_t len) {
-    struct ll_entry* e = queue_entry_new(0); if (!e) return -1;
-    e->dgram = u_malloc(len + 1); if (!e->dgram) { queue_entry_free(e); return -1; }
-    e->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY; memcpy(e->dgram + 1, data, len); e->len = (uint16_t)(len + 1);
-    int rc = etcp_route_send(inst, g_group_id, dst, e, 1, 0);
-    if (rc != 0) { u_free(e->dgram); queue_entry_free(e); }
-    return rc;
-}
-
 /* ── завершение загрузки ── */
 
 static void dl_done_cb(void* arg, int err) {
@@ -482,20 +472,16 @@ static void setup_chat_group(void) {
     }
 
     TEST("media_delivery supernode s1,s2"); {
-        media_delivery_set_supernode(g_inst[I_S1], 1);
-        media_delivery_set_supernode(g_inst[I_S2], 1);
+        media_delivery_refresh_roles(g_inst[I_S1]);
+        media_delivery_refresh_roles(g_inst[I_S2]);
         if (g_inst[I_S1]->md.is_supernode && g_inst[I_S2]->md.is_supernode) OK(); else FAIL();
     }
 
-    TEST("SUPER_HELLO s1<->s2 (via CHAT)"); {
-        struct media_pkt_super_hello h; memset(&h, 0, sizeof(h));
-        h.subcmd = MEDIA_SUBCMD_SUPER_HELLO; h.group_id = g_group_id;
-        msend(g_inst[I_S1], g_nid[I_S2], (const uint8_t*)&h, sizeof(h));
-        msend(g_inst[I_S2], g_nid[I_S1], (const uint8_t*)&h, sizeof(h));
+    TEST("automatic SYNC s1<->s2 (via CHAT)"); {
         int a = 0, ok = 0;
         while (a < 1000) {
-            ok = (g_inst[I_S1]->md.super_peers && g_inst[I_S1]->md.super_peers->head)
-              && (g_inst[I_S2]->md.super_peers && g_inst[I_S2]->md.super_peers->head);
+            ok = db_count(g_inst[I_S1]->md.db, "super_sync", "group_id", g_group_id) > 0
+              && db_count(g_inst[I_S2]->md.db, "super_sync", "group_id", g_group_id) > 0;
             if (ok) break;
             uasync_poll(g_ua, POLL_MS); a++;
         }
@@ -678,8 +664,9 @@ int main(void) {
         uint64_t start = get_time_tb();
         while ((get_time_tb() - start) < (uint64_t)60000) {
             sqlite3_stmt* st = NULL; uint64_t lr = 0;
-            sqlite3_prepare_v2(g_inst[I_S2]->topo_sqlite_db, "SELECT last_recv_id FROM super_sync WHERE peer_node_id=?", -1, &st, NULL);
-            sqlite3_bind_int64(st, 1, (sqlite3_int64)g_nid[I_S1]);
+            sqlite3_prepare_v2(g_inst[I_S2]->topo_sqlite_db, "SELECT last_recv_id FROM super_sync WHERE group_id=? AND peer_node_id=?", -1, &st, NULL);
+            sqlite3_bind_int64(st, 1, (sqlite3_int64)g_group_id);
+            sqlite3_bind_int64(st, 2, (sqlite3_int64)g_nid[I_S1]);
             if (sqlite3_step(st) == SQLITE_ROW) lr = (uint64_t)sqlite3_column_int64(st, 0);
             sqlite3_finalize(st);
             if (lr > 0) { ok = 1; break; }

tests/test_media_delivery_integration.c

@@ -2,11 +2,11 @@
 //
 // Проверки:
 //   1. media_delivery_init на реальных инстансах (таблицы созданы)
-//   2. media_delivery_set_supernode → super_peers/served_nodes созданы
+//   2. Подписанная роль суперузла CHAT-группы → обработка объявлений
 //   3. SERVE_REG → добавление в served_nodes
 //   4. HAVE_BLOCK → вставка в block_availability + ACK
 //   5. QUERY → поиск в block_availability → QUERY_RESP
-//   6. SUPER_HELLO → обмен last_recv_id
+//   6. Смена подписанной роли → очистка каталога этой группы
 //   7. SERVE_LEAVE → удаление из served_nodes
 //   8. Выключение суперузла → чистка block_availability
 
@@ -18,6 +18,9 @@
 #include "../config_updater.h"
 #include "../routing_layer/topo_group.h"
 #include "../routing_layer/topo_node.h"
+#include "../routing_layer/topo_node_sqlite.h"
+#include "../chat/member_sync.h"
+#include "../transport_layer/secure_channel.h"
 #include "../lib/u_async.h"
 #include "../lib/debug_config.h"
 #include "../lib/mem.h"
@@ -36,6 +39,8 @@ static int g_passed = 0, g_failed = 0, g_total = 0;
 
 #define TIMEOUT_TB 300000
 #define POLL_MS 5
+#define CHAT_GROUP 111
+#define CHAT_CHANNEL "111"
 
 static struct UASYNC* g_ua = NULL;
 static struct UTUN_INSTANCE* g_a = NULL, *g_b = NULL;
@@ -45,6 +50,37 @@ static char g_tdir[] = "/tmp/utun_mdl_XXXXXX";
 static char g_ca[256], g_cb[256];
 static uint8_t g_a_pubkey[32], g_b_pubkey[32];
 
+/* Меняем роль через подписанный блок владельца, как в рабочем member_sync. */
+static int set_super_role(int enabled) {
+    static unsigned version;
+    char tags[96]; uint8_t message[MS_ADM_TAGS_MAX + 8], sig[64];
+    snprintf(tags, sizeof(tags), "{\"ver\":\"%u\",\"supernode\":\"%s\"}", ++version, enabled ? "yes" : "no");
+    int n = member_sync_build_owner_msg(g_nid_a, tags, message, sizeof(message));
+    if (n < 0 || sc_ed25519_sign(g_a->my_ed25519_privkey, message, n, sig) != SC_OK) return -1;
+    struct UTUN_INSTANCE* nodes[2] = {g_a, g_b};
+    for (int i = 0; i < 2; i++) {
+        if (member_sync_put(nodes[i], CHAT_CHANNEL, g_nid_a, g_a->my_keys.public_key, g_a->my_ed25519_pubkey,
+            NULL, 0, NULL, 0, "{}", tags, sig, 0, 0, NULL) < 0) return -1;
+        media_delivery_refresh_roles(nodes[i]);
+    }
+    return 0;
+}
+
+static int setup_chat(void) {
+    struct UTUN_INSTANCE* nodes[2] = {g_a, g_b};
+    uint8_t sig[64];
+    if (sc_ed25519_sign(g_a->my_ed25519_privkey, (const uint8_t*)CHAT_CHANNEL, strlen(CHAT_CHANNEL), sig) != SC_OK) return -1;
+    for (int i = 0; i < 2; i++) {
+        struct UTUN_INSTANCE* inst = nodes[i];
+        if (member_sync_init(inst) < 0 || topo_node_sqlite_channel_put(inst->topo_sqlite_db, CHAT_CHANNEL, "media", g_nid_a,
+            g_a->my_keys.public_key, NULL, g_a->my_ed25519_pubkey, NULL, sig) < 0) return -1;
+        if (!topo_groups_create_group(inst->topo_groups, CHAT_GROUP, TOPO_GROUP_TYPE_CHAT, CHAT_CHANNEL)) return -1;
+        for (int j = 0; j < 2; j++) if (member_sync_put(inst, CHAT_CHANNEL, nodes[j]->node_id,
+            nodes[j]->my_keys.public_key, nodes[j]->my_ed25519_pubkey, NULL, 0, NULL, 0, "{}", NULL, NULL, 0, 0, NULL) < 0) return -1;
+    }
+    return set_super_role(1);
+}
+
 /* ── helpers ── */
 
 static int wf(const char* p, const char* f, ...) {
@@ -117,7 +153,7 @@ static int msend(struct UTUN_INSTANCE* inst, uint64_t dst, const uint8_t* data,
     e->dgram = u_malloc(len + 1); if (!e->dgram) { queue_entry_free(e); return -1; }
     e->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY;
     memcpy(e->dgram + 1, data, len); e->len = (uint16_t)(len + 1);
-    int rc = etcp_route_send(inst, TOPO_GROUP_UTUN, dst, e, 1, 0);
+    int rc = etcp_route_send(inst, CHAT_GROUP, dst, e, 1, 0);
     return rc;
 }
 
@@ -139,7 +175,7 @@ static void query_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
         uint8_t block[16], request[16];
         for (int i = 0; i < 16; i++) block[i] = (uint8_t)(0x10 + i);
         memset(request, 0x91, 16);
-        query_valid = source == g_nid_a && group == TOPO_GROUP_UTUN && r->status == MEDIA_QUERY_OK && r->num_blocks == 1
+        query_valid = source == g_nid_a && group == CHAT_GROUP && r->status == MEDIA_QUERY_OK && r->num_blocks == 1
             && !memcmp(r->request_id, request, 16) && !memcmp(r->media_id, block, 16) && !memcmp(e->block_id, block, 16)
             && e->num_holders == 1 && e->holders[0] == g_nid_a;
     }
@@ -153,7 +189,7 @@ static void phase1_serve_reg(void) {
         struct media_pkt_serve_reg pkt;
         memset(&pkt, 0, sizeof(pkt));
         pkt.subcmd = MEDIA_SUBCMD_SERVE_REG;
-        pkt.group_id = TOPO_GROUP_UTUN;
+        pkt.group_id = CHAT_GROUP;
 
         int rc = msend(g_b, g_nid_a, (const uint8_t*)&pkt, sizeof(pkt));
         /* BGP routes need time — poll up to 10s */
@@ -171,7 +207,7 @@ static void phase1_serve_reg(void) {
 static void phase2_have_block(void) {
     TEST("HAVE_BLOCK B→A inserts into block_availability"); {
         uint8_t block_uuid[16]; for (int i = 0; i < 16; i++) block_uuid[i] = (uint8_t)(0x10 + i);
-        uint64_t group_id = TOPO_GROUP_UTUN;
+        uint64_t group_id = CHAT_GROUP;
 
         struct media_pkt_have_block hb;
         memset(&hb, 0, sizeof(hb));
@@ -205,7 +241,7 @@ static void phase3_query(void) {
         memcpy(q->media_id, block_uuid, 16);
         q->num_blocks = 1;
         memcpy(pkt + MEDIA_QUERY_HDR_SIZE, block_uuid, 16);
-        media_delivery_send_have_block(g_a, TOPO_GROUP_UTUN, g_nid_a, block_uuid, block_uuid, 0);
+        media_delivery_send_have_block(g_a, CHAT_GROUP, g_nid_a, block_uuid, block_uuid, 0);
         saved_media_recv = g_b->router_bindings.callbacks[ETCP_RT_ID_MEDIA_DELIVERY];
         g_b->router_bindings.callbacks[ETCP_RT_ID_MEDIA_DELIVERY] = query_recv;
         query_received = query_valid = 0;
@@ -222,7 +258,7 @@ static void phase4_serve_leave(void) {
         struct media_pkt_serve_leave pkt;
         memset(&pkt, 0, sizeof(pkt));
         pkt.subcmd = MEDIA_SUBCMD_SERVE_LEAVE;
-        pkt.group_id = TOPO_GROUP_UTUN;
+        pkt.group_id = CHAT_GROUP;
 
         msend(g_b, g_nid_a, (const uint8_t*)&pkt, sizeof(pkt));
         int attempts = 0;
@@ -238,13 +274,13 @@ static void phase4_serve_leave(void) {
 
 static void phase5_supernode_stop(void) {
     TEST("supernode STOP cleans block_availability"); {
-        media_delivery_set_supernode(g_a, 0);
+        if (set_super_role(0)) { FAIL("role update failed"); return; }
         int n = count_rows(g_a->md.db, "block_availability", NULL, 0);
         if (n == 0) OK(); else FAIL("%d entries remain", n);
     }
 
     TEST("supernode START re-enables"); {
-        media_delivery_set_supernode(g_a, 1);
+        if (set_super_role(1)) { FAIL("role update failed"); return; }
         if (g_a->md.is_supernode) OK(); else FAIL();
     }
 }
@@ -312,8 +348,7 @@ int main(void) {
     utun_instance_init(g_a); utun_instance_init(g_b);
     media_delivery_init(g_a); media_delivery_init(g_b);
 
-    /* make A supernode */
-    media_delivery_set_supernode(g_a, 1);
+    if (setup_chat()) { FAIL("CHAT setup failed"); goto done; }
 
     /* wait for ETCP connection */
     mon(NULL);