diff --git a/src/media_delivery/media_delivery.c b/src/media_delivery/media_delivery.c index 06cda570..76b38e57 100644 --- a/src/media_delivery/media_delivery.c +++ b/src/media_delivery/media_delivery.c @@ -406,6 +406,14 @@ static void md_super_repl_send(struct media_delivery_ctx* md, struct media_super static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, const uint8_t* data, size_t len) { if (!inst || !data || len == 0) return -1; + struct TOPO_GROUP* grp = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL; + struct ETCP_CONN* conn = grp ? topo_group_find_conn_for_node(grp, dst_node_id) : NULL; + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: send dst=0x%016llx group=0x%016llx subcmd=0x%02x len=%zu conn=%p", + MD_ID, (unsigned long long)dst_node_id, (unsigned long long)group_id, + (unsigned)data[0], len, (void*)conn); + if (!conn) + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: no conn to 0x%016llx group=0x%016llx subcmd=0x%02x — packet queued, no route", + MD_ID, (unsigned long long)dst_node_id, (unsigned long long)group_id, (unsigned)data[0]); struct ll_entry* e = queue_entry_new(0); if (!e) return -1; e->dgram = u_malloc(len + 1); diff --git a/src/media_delivery/media_download.c b/src/media_delivery/media_download.c index eeee831b..2779936c 100644 --- a/src/media_delivery/media_download.c +++ b/src/media_delivery/media_download.c @@ -89,8 +89,12 @@ static int md_dl_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t ds if (!inst || !inst->topo_groups || !inst->connections) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: md_dl_send: no routing (inst=%p topo=%p conn=%p)", MDL_ID, (void*)inst, (void*)(inst ? inst->topo_groups : NULL), (void*)(inst ? inst->connections : NULL)); return -1; } { struct TOPO_GROUP* grp = topo_groups_find(inst->topo_groups, group_id); struct ETCP_CONN* gc = grp ? topo_group_find_conn_for_node(grp, dst) : NULL; - DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "%s: md_dl_send group=0x%016llx dst=0x%016llx grp=%p conn_in_group=%p", - MDL_ID, (unsigned long long)group_id, (unsigned long long)dst, (void*)grp, (void*)gc); + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: send dst=0x%016llx group=0x%016llx subcmd=0x%02x len=%zu grp=%p conn=%p", + MDL_ID, (unsigned long long)dst, (unsigned long long)group_id, + (unsigned)data[0], len, (void*)grp, (void*)gc); + if (!gc) + DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: no conn to 0x%016llx subcmd=0x%02x — packet queued, no route", + MDL_ID, (unsigned long long)dst, (unsigned)data[0]); } struct ll_entry* e = queue_entry_new(0); if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new failed for send to 0x%016llx", MDL_ID, (unsigned long long)dst); return -1; } @@ -447,8 +451,12 @@ static void md_dl_send_query(struct UTUN_INSTANCE* inst, struct media_download* for (int i = 0; i < dl->num_blocks; i++) memcpy(pkt + MEDIA_QUERY_HDR_SIZE + (size_t)i * 16, dl->blocks[i].block_id, 16); - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: QUERY to super 0x%016llx blocks=%d", - MDL_ID, (unsigned long long)super, dl->num_blocks); + { + struct TOPO_GROUP* grp = inst->topo_groups ? topo_groups_find(inst->topo_groups, dl->group_id) : NULL; + struct ETCP_CONN* gc = grp ? topo_group_find_conn_for_node(grp, super) : NULL; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: QUERY to super 0x%016llx blocks=%d conn=%d", + MDL_ID, (unsigned long long)super, dl->num_blocks, gc ? 1 : 0); + } md_dl_send(inst, dl->group_id, super, pkt, pkt_len); u_free(pkt); dl->last_activity_tb = get_time_tb(); diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index b3383833..bfb77c0a 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -432,6 +432,9 @@ static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn) { static void router_set_no_route(struct ETCP_ROUTER_CONN* rconn) { if (rconn->no_route) return; rconn->no_route = 1; + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: no route to %016llx svc_id=%u group=%016llx — send_q=%d queued, will retry", + (unsigned long long)rconn->remote_node_id, rconn->svc_id, + (unsigned long long)rconn->group_id, queue_entry_count(rconn->send_q)); if (!rconn->no_route_timer) rconn->no_route_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_NO_ROUTE_RETRY_TB, rconn, router_no_route_retry_cb, "router_no_route"); @@ -907,6 +910,10 @@ static void router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* con DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router_deliver: svc_id=%u len=%zu flags=%02x src=%016llx dst=%016llx", rconn->svc_id, payload_len, rx_flags, (unsigned long long)rconn->remote_node_id, (unsigned long long)inst->node_id); + if (rconn->svc_id == ETCP_RT_ID_MEDIA_DELIVERY) { + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "router_deliver: MEDIA pkt len=%zu subcmd=0x%02x from %016llx", + payload_len, payload_len > 0 ? payload[0] : 0, (unsigned long long)rconn->remote_node_id); + } rconn->c_pkts_rcvd++; cb(conn, svc_entry); } diff --git a/tests/Makefile.am b/tests/Makefile.am index 30a707ba..841e9cfd 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -74,6 +74,7 @@ check_PROGRAMS = \ test_media_download_timeout \ test_media_delivery_integration \ test_media_delivery_full \ + test_media_delivery_chat \ test_etcp_link_stress \ test_reality_hello \ test_reality_bgp \ @@ -415,6 +416,10 @@ test_media_delivery_full_SOURCES = test_media_delivery_full.c test_media_delivery_full_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/media_delivery -I$(top_srcdir)/src/media_async -I$(top_srcdir)/lib -I$(top_srcdir)/src/transport_layer -I$(top_srcdir)/src/routing_layer -I$(top_srcdir)/src/chat test_media_delivery_full_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) +test_media_delivery_chat_SOURCES = test_media_delivery_chat.c +test_media_delivery_chat_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/media_delivery -I$(top_srcdir)/src/media_async -I$(top_srcdir)/lib -I$(top_srcdir)/src/transport_layer -I$(top_srcdir)/src/routing_layer -I$(top_srcdir)/src/chat +test_media_delivery_chat_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) + # Copy test configs to build directory (tests run from build/tests/) all-local: copy-test-configs diff --git a/tests/test_media_delivery_chat.c b/tests/test_media_delivery_chat.c new file mode 100644 index 00000000..eea41838 --- /dev/null +++ b/tests/test_media_delivery_chat.c @@ -0,0 +1,636 @@ +// test_media_delivery_chat.c — сквозной тест распространения медиа через db_sync +// +// 6 узлов в цепочку: +// n1(0) — n2(1) — s1(2) — s2(3) — st1(4) — st2(5) +// автор клиент супер супер storage storage +// +// Сценарий: +// 1. n1 индексирует медиафайл (media_index_commit) и публикует сообщение с метаданными +// через db_sync (реальная репликация по цепочке каскадом). +// 2. storage-узлы (st1/st2) по insert-callback'у запускают auto-download +// (media_download_start), скачивают блоки у автора, собирают файл. +// 3. Проверяем, что суперузлы (s1/s2) узнали о блоках (block_availability от HAVE_BLOCK) +// и отреплицировались между собой (super_sync.last_recv_id > 0). +// 4. Второй клиент n2 скачивает медиа «по запросу» и собирает файл. +// +// Используется настоящий db_sync для доставки сообщения и настоящий media-движок +// для доставки блоков. chat_core НЕ используется (он — глобальный синглтон на процесс), +// поэтому связка «сообщение → auto-download» воспроизведена локальным insert-callback'ом. + +#include "media_delivery.h" +#include "media_delivery_proto.h" +#include "media_download.h" +#include "media_index.h" +#include "member_sync.h" +#include "db_sync.h" +#include "../routing_layer/topo_node_sqlite.h" +#include "../utun_instance.h" +#include "../transport_layer/etcp.h" +#include "../transport_layer/secure_channel.h" +#include "../config_parser.h" +#include "../config_updater.h" +#include "../routing_layer/topo_group.h" +#include "../routing_layer/topo_node.h" +#include "../lib/u_async.h" +#include "../lib/debug_config.h" +#include "../lib/mem.h" + +#include +#define OPENSSL_API_COMPAT 0x10100000L +#include +#include +#include +#include +#include +#include +#include + +static int G_PASSED = 0, G_FAILED = 0, G_TOTAL = 0; +#define TEST(n) do { G_TOTAL++; printf(" %-58s", n); fflush(stdout); } while(0) +#define OK() do { G_PASSED++; printf("OK\n"); } while(0) +#define FAIL(f,...) do { G_FAILED++; printf("FAIL: " f "\n", ##__VA_ARGS__); } while(0) + +#define TIMEOUT_TB 1200000 /* 120 s общий таймаут */ +#define POLL_MS 5 +#define N_NODES 6 +#define CH_ID "31415926535" + +/* роли узлов */ +#define I_N1 0 /* автор */ +#define I_N2 1 /* клиент (скачивает позже) */ +#define I_S1 2 /* суперузел */ +#define I_S2 3 /* суперузел */ +#define I_ST1 4 /* storage */ +#define I_ST2 5 /* storage */ + +#define MEDIA_FNAME "photo.bin" +#define MEDIA_FILE_SIZE 15360 +#define MEDIA_BLOCK_SIZE 5120 +#define MEDIA_NUM_BLOCKS 3 + +static struct UASYNC* g_ua = NULL; +static struct UTUN_INSTANCE* g_inst[N_NODES]; +static uint64_t g_nid[N_NODES]; +static struct DB_SYNC_INSTANCE* g_si[N_NODES]; +static uint64_t g_group_id = 0; + +static char g_tdir[256] = "/tmp/utun_mdc_XXXXXX"; +static char g_cfg[N_NODES][256]; +static char g_db_dir[N_NODES][320]; +static int g_port[N_NODES]; +static int g_connected = 0, g_result = 0; + +/* опубликованное медиа (автор) — глобально для n2-загрузки и проверок */ +static uint8_t g_media_id[16]; +static uint8_t g_content_hash[32]; +static uint8_t g_block_ids[MEDIA_NUM_BLOCKS][16]; +static uint8_t g_block_sigs[MEDIA_NUM_BLOCKS][64]; +static uint8_t g_file_data[MEDIA_FILE_SIZE]; + +/* флаги завершения загрузок */ +static int g_dl_done[N_NODES]; +static int g_dl_err[N_NODES]; + +/* ── helpers ── */ + +static int wf(const char* p, const char* f, ...) { + va_list ap; FILE* fp = fopen(p, "w"); if (!fp) return -1; + va_start(ap, f); vfprintf(fp, f, ap); va_end(ap); fclose(fp); return 0; +} +static char* gv(const char* p, const char* k) { + struct utun_config* c = parse_config(p); if (!c) return NULL; + char* r = (strcmp(k, "pub") == 0) ? u_strdup(c->global.my_public_key_hex) : u_strdup(c->global.my_private_key_hex); + free_config(c); return r; +} +static void t_mkdtemp(char* t) { +#ifdef _WIN32 + char tmp_path[512]; GetTempPathA(sizeof(tmp_path), tmp_path); + snprintf(t, 256, "%s\\utun_test_%08x", tmp_path, (unsigned)rand()); + _mkdir(t); +#else + (void)!mkdtemp(t); +#endif +} +static void t_rmdir(const char* p) { char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", p); system(cmd); } + +static int db_count(sqlite3* db, const char* tbl, const char* wh, uint64_t val) { + char sql[256]; + if (wh) snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM %s WHERE %s=%llu", tbl, wh, (unsigned long long)val); + else snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM %s", tbl); + sqlite3_stmt* s = NULL; + if (sqlite3_prepare_v2(db, sql, -1, &s, NULL) != SQLITE_OK) return -1; + int n = -1; + if (sqlite3_step(s) == SQLITE_ROW) n = sqlite3_column_int(s, 0); + sqlite3_finalize(s); return n; +} + +static void to_cb(void* arg) { (void)arg; fprintf(stderr, " TIMEOUT\n"); g_result = 2; } + +static void mon(void* arg) { + (void)arg; + if (g_result) return; + int ok = 0; + for (int i = 0; i < N_NODES; i++) { + struct ll_entry* e = g_inst[i]->connections->head; + while (e) { struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data; + if (ce->conn && ce->conn->links) { struct ETCP_LINK* l; for (l = ce->conn->links; l; l = l->next) + if (l->initialized && ce->conn->crypto_ctx.initialized) ok++; } e = e->next; } + } + if (ok >= N_NODES * 2 - 2) g_connected = 1; + if (!g_result) uasync_set_timeout(g_ua, 10, NULL, mon, "mdc_mon"); +} + +/* pubkey узла известен в in-memory registry (BGP сошёлся) */ +static int pubkey_known(struct UTUN_INSTANCE* inst, uint64_t node_id) { + struct TOPO_NODE* ni = inst && inst->topo_groups ? topo_node_registry_find(inst->topo_groups, node_id) : NULL; + if (!ni) return 0; + uint64_t chk; memcpy(&chk, ni->ed25519_public_key, 8); + return chk != 0; +} + +/* ждём, пока все узлы узнают ed25519-ключи всех остальных (нужно db_sync для верификации подписи) */ +static int wait_pubkeys(void) { + int a = 0; + while (a < 4000) { + int ok = 1; + for (int i = 0; i < N_NODES; i++) + for (int m = 0; m < N_NODES; m++) + if (m != i && !pubkey_known(g_inst[i], g_nid[m])) ok = 0; + if (ok) return 1; + uasync_poll(g_ua, POLL_MS); a++; + } + return 0; +} + +static void sha256_buf(const uint8_t* data, size_t len, uint8_t out[32]) { + EVP_MD_CTX* ctx = EVP_MD_CTX_new(); + EVP_DigestInit_ex(ctx, EVP_sha256(), NULL); + EVP_DigestUpdate(ctx, data, len); + EVP_DigestFinal_ex(ctx, out, NULL); + EVP_MD_CTX_free(ctx); +} + +static int file_sha256(const char* path, uint8_t out[32]) { + FILE* f = fopen(path, "rb"); if (!f) return -1; + EVP_MD_CTX* ctx = EVP_MD_CTX_new(); + EVP_DigestInit_ex(ctx, EVP_sha256(), NULL); + uint8_t buf[8192]; size_t rd; + while ((rd = fread(buf, 1, sizeof(buf), f)) > 0) EVP_DigestUpdate(ctx, buf, rd); + EVP_DigestFinal_ex(ctx, out, NULL); + EVP_MD_CTX_free(ctx); fclose(f); + return 0; +} + +static int file_exists(const char* p) { return access(p, F_OK) == 0; } + +/* nodeinfo_updated пересчитывает node_type по адресам и может сбросить тип суперузла, + * если узел не имеет DIRECT-адреса. Перед каждой загрузкой принудительно выставляем + * node_type=4 для s1/s2 (md_dl_collect_supernodes читает именно это поле). */ +static void ensure_supernode_type(int idx) { + char sql[256]; + snprintf(sql, sizeof(sql), "UPDATE \"peers_%s\" SET node_type=4 WHERE node_id IN (%llu,%llu)", + CH_ID, (unsigned long long)g_nid[I_S1], (unsigned long long)g_nid[I_S2]); + sqlite3_exec(g_inst[idx]->topo_sqlite_db, sql, NULL, NULL, NULL); +} + +/* ── завершение загрузки ── */ + +static void dl_done_cb(void* arg, int err) { + intptr_t idx = (intptr_t)arg; + g_dl_done[idx] = 1; g_dl_err[idx] = err; + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "[mdc] download done node=%d err=%d", (int)idx, err); +} + +/* media_base узла = каталог db_path (как chat_core/md_start_download) */ +static void node_media_base(int idx, char* out, size_t sz) { snprintf(out, sz, "%s", g_db_dir[idx]); } +static void node_dest(int idx, char* out, size_t sz) { + snprintf(out, sz, "%s/media/%s/%s", g_db_dir[idx], CH_ID, MEDIA_FNAME); +} + +/* ── парсинг тела медиа-сообщения (как md_start_download в chat_msg.c) ── */ + +static int hex_byte(const char* p) { + int v = 0; + for (int i = 0; i < 2; i++) { char c = p[i]; v <<= 4; + v |= (c >= '0' && c <= '9') ? (c - '0') : (c >= 'a' && c <= 'f') ? (c - 'a' + 10) : (c - 'A' + 10); } + return v; +} + +static int parse_media_body(const char* body, struct media_index_result* out) { + const char* p = body; + while (*p && *p != '|') p++; + if (!*p) return -1; p++; + out->file_size = strtoll(p, (char**)&p, 10); + if (*p != '|') return -1; p++; + out->block_size = strtoll(p, (char**)&p, 10); + if (*p != '|') return -1; p++; + int nb = (int)strtol(p, (char**)&p, 10); + while (*p && *p != '|') p++; + if (!*p) return -1; p++; + if (nb <= 0 || nb > 100 || out->file_size <= 0) return -1; + + for (int i = 0; i < 16; i++) out->media_id[i] = (uint8_t)hex_byte(p + i * 2); + p += 32; if (*p != '|') return -1; p++; + for (int i = 0; i < 32; i++) out->content_hash[i] = (uint8_t)hex_byte(p + i * 2); + p += 64; if (*p != '|') return -1; p++; + + out->num_blocks = nb; + out->block_ids = u_malloc((size_t)nb * 16); + out->block_sigs = u_malloc((size_t)nb * 64); + if (!out->block_ids || !out->block_sigs) { media_index_result_free(out); return -1; } + for (int n = 0; n < nb; n++) { + for (int i = 0; i < 16; i++) out->block_ids[n * 16 + i] = (uint8_t)hex_byte(p + i * 2); + p += 32; if (*p != ',') { media_index_result_free(out); return -1; } p++; + for (int i = 0; i < 64; i++) out->block_sigs[n * 64 + i] = (uint8_t)hex_byte(p + i * 2); + p += 128; + if (n < nb - 1 && *p == ',') p++; + } + return 0; +} + +/* ── insert-callback: storage-узлы запускают auto-download ── */ + +static void storage_on_insert(struct DB_SYNC_INSTANCE* si, uint64_t ts, + const char* data, size_t len, uint64_t author, void* arg) { + (void)si; (void)ts; + int idx = (int)(intptr_t)arg; + struct UTUN_INSTANCE* inst = g_inst[idx]; + if (author == inst->node_id) return; + + char buf[8192]; size_t blen = len < sizeof(buf) - 1 ? len : sizeof(buf) - 1; + memcpy(buf, data, blen); buf[blen] = '\0'; + const char* ct = strstr(buf, "\"ct\":\""); + const char* dpos = strstr(buf, "\"d\":\""); + if (!ct || !dpos) return; + + const char* ct_val = ct + 6; + char content_type[32] = {0}; int cti = 0; + while (ct_val[cti] && ct_val[cti] != '\"' && cti < (int)sizeof(content_type) - 1) content_type[cti++] = ct_val[cti]; + + const char* d_start = dpos + 5; + char body[8192]; size_t bi = 0; + while (*d_start && *d_start != '\"' && bi < sizeof(body) - 1) body[bi++] = *d_start++; + body[bi] = '\0'; + + struct media_index_result result; + memset(&result, 0, sizeof(result)); + if (parse_media_body(body, &result) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "[mdc] parse_media_body failed"); return; } + + char dest[1024], mbase[512]; + node_media_base(idx, mbase, sizeof(mbase)); + node_dest(idx, dest, sizeof(dest)); + + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "[mdc] auto-download on node=%d ch=%s ct=%s blocks=%d author=0x%016llx dest=%s", + idx, CH_ID, content_type, result.num_blocks, (unsigned long long)author, dest); + ensure_supernode_type(idx); + media_download_start(inst, g_group_id, &result, dest, mbase, author, dl_done_cb, (void*)(intptr_t)idx, NULL, NULL); + media_index_result_free(&result); +} + +/* ── публикация медиа автором ── */ + +static int author_index_file(void) { + char media_dir[512]; snprintf(media_dir, sizeof(media_dir), "%s/media", g_db_dir[I_N1]); + utun_mkdir(media_dir, 0755); + char dst_path[512]; snprintf(dst_path, sizeof(dst_path), "%s/%s", media_dir, MEDIA_FNAME); + char media_base[512]; snprintf(media_base, sizeof(media_base), "%s", g_db_dir[I_N1]); + + FILE* f = fopen(dst_path, "wb"); if (!f) return -1; + fwrite(g_file_data, 1, MEDIA_FILE_SIZE, f); fclose(f); + + sha256_buf(g_file_data, MEDIA_FILE_SIZE, g_content_hash); + + struct media_index_result result; memset(&result, 0, sizeof(result)); + media_index_init(g_inst[I_N1]->topo_sqlite_db); + media_index_generate_uuid(result.media_id); + memcpy(result.content_hash, g_content_hash, 32); + result.file_size = MEDIA_FILE_SIZE; result.block_size = MEDIA_BLOCK_SIZE; result.num_blocks = MEDIA_NUM_BLOCKS; + result.block_ids = u_malloc(MEDIA_NUM_BLOCKS * 16); + result.block_sigs = u_malloc(MEDIA_NUM_BLOCKS * 64); + for (int n = 0; n < MEDIA_NUM_BLOCKS; n++) { + media_index_generate_uuid(result.block_ids + n * 16); + int off = n * MEDIA_BLOCK_SIZE; + int sz = (n == MEDIA_NUM_BLOCKS - 1) ? MEDIA_FILE_SIZE - off : MEDIA_BLOCK_SIZE; + /* подпись блока = Ed25519(данные блока) — ровно как проверяет download */ + sc_ed25519_sign(g_inst[I_N1]->my_ed25519_privkey, g_file_data + off, (size_t)sz, result.block_sigs + n * 64); + } + int rc = media_index_commit(g_inst[I_N1]->topo_sqlite_db, &result, g_nid[I_N1], + g_inst[I_N1]->my_ed25519_privkey, CH_ID, dst_path, media_base); + + /* сохраняем глобально */ + memcpy(g_media_id, result.media_id, 16); + for (int n = 0; n < MEDIA_NUM_BLOCKS; n++) { + memcpy(g_block_ids[n], result.block_ids + n * 16, 16); + memcpy(g_block_sigs[n], result.block_sigs + n * 64, 64); + } + media_index_result_free(&result); + return rc; +} + +static int author_publish_message(void) { + /* body: ||||||,,... */ + char body[8192]; size_t off = 0; + off += snprintf(body + off, sizeof(body) - off, "%s|%d|%d|%d|", + MEDIA_FNAME, MEDIA_FILE_SIZE, MEDIA_BLOCK_SIZE, MEDIA_NUM_BLOCKS); + for (int i = 0; i < 16; i++) off += snprintf(body + off, sizeof(body) - off, "%02x", g_media_id[i]); + off += snprintf(body + off, sizeof(body) - off, "|"); + for (int i = 0; i < 32; i++) off += snprintf(body + off, sizeof(body) - off, "%02x", g_content_hash[i]); + off += snprintf(body + off, sizeof(body) - off, "|"); + for (int n = 0; n < MEDIA_NUM_BLOCKS; n++) { + if (n > 0) body[off++] = ','; + for (int i = 0; i < 16; i++) off += snprintf(body + off, sizeof(body) - off, "%02x", g_block_ids[n][i]); + body[off++] = ','; + for (int i = 0; i < 64; i++) off += snprintf(body + off, sizeof(body) - off, "%02x", g_block_sigs[n][i]); + } + + char json[8700]; + snprintf(json, sizeof(json), "{\"n\":%llu,\"ch\":\"%s\",\"ct\":\"application/octet-stream\",\"d\":\"%s\"}", + (unsigned long long)g_nid[I_N1], CH_ID, body); + + uint64_t ts = db_sync_next_timestamp(g_si[I_N1]); + size_t jl = strlen(json); + uint8_t sig_msg[8700]; size_t soff = 0; + memcpy(sig_msg + soff, &ts, 8); soff += 8; + memcpy(sig_msg + soff, json, jl); soff += jl; + uint8_t sig[64]; + if (sc_ed25519_sign(g_inst[I_N1]->my_ed25519_privkey, sig_msg, soff, sig) != SC_OK) return -1; + + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "[mdc] publish media message ts=%llu len=%zu", (unsigned long long)ts, jl); + return db_sync_insert_signed(g_si[I_N1], json, jl, sig, 64, ts, NULL); +} + +/* ── фазы ── */ + +static void setup_chat_group(void) { + uint8_t x25519_pub[32] = {0}, ed_pub[32] = {0}, ch_sig[64] = {0}; + int i; + + TEST("generate channel keys"); { + EVP_PKEY* xpkey = NULL, *epkey = NULL; + EVP_PKEY_CTX* ctx = EVP_PKEY_CTX_new_id(EVP_PKEY_X25519, NULL); + if (ctx) { EVP_PKEY_keygen_init(ctx); EVP_PKEY_keygen(ctx, &xpkey); EVP_PKEY_CTX_free(ctx); } + if (xpkey) { size_t l = 32; EVP_PKEY_get_raw_public_key(xpkey, x25519_pub, &l); EVP_PKEY_free(xpkey); } + ctx = EVP_PKEY_CTX_new_id(EVP_PKEY_ED25519, NULL); + if (ctx) { EVP_PKEY_keygen_init(ctx); EVP_PKEY_keygen(ctx, &epkey); EVP_PKEY_CTX_free(ctx); } + if (epkey) { size_t l = 32; EVP_PKEY_get_raw_public_key(epkey, ed_pub, &l); + EVP_MD_CTX* mctx = EVP_MD_CTX_new(); + if (mctx) { EVP_DigestSignInit(mctx, NULL, NULL, NULL, epkey); + size_t sl = 64; EVP_DigestSign(mctx, ch_sig, &sl, (const uint8_t*)CH_ID, strlen(CH_ID)); EVP_MD_CTX_free(mctx); } + EVP_PKEY_free(epkey); + } + if (x25519_pub[0] || ed_pub[0]) OK(); else FAIL(); + } + + TEST("create CHAT group + peers tables"); { + for (i = 0; i < N_NODES; i++) { + sqlite3* db = g_inst[i]->topo_sqlite_db; + struct TOPO_GROUPS* tg = g_inst[i]->topo_groups; + if (!db || !tg) continue; + topo_node_sqlite_channel_put(db, CH_ID, "test", g_nid[0], x25519_pub, NULL, ed_pub, NULL, ch_sig); + if (topo_groups_find(tg, g_group_id)) continue; + topo_groups_create_group(tg, g_group_id, TOPO_GROUP_TYPE_CHAT, CH_ID); + } + OK(); + } + + TEST("add 6 members (s1/s2 supernode, st1/st2 storage)"); { + const char* adm[N_NODES] = { NULL, NULL, "{\"supernode\":\"yes\"}", "{\"supernode\":\"yes\"}", "{\"storage\":\"yes\"}", "{\"storage\":\"yes\"}" }; + for (i = 0; i < N_NODES; i++) { + for (int m = 0; m < N_NODES; m++) { + member_sync_put(g_inst[i], CH_ID, g_nid[m], + g_inst[m]->my_keys.public_key, g_inst[m]->my_ed25519_pubkey, + NULL, 0, NULL, 0, "{\"name\":\"node\"}", adm[m], NULL, 0); + /* заполняем nodes-таблицу (нужно для верификации Ed25519 подписи блоков/автора) */ + topo_node_sqlite_node_update_verified(g_inst[i]->topo_sqlite_db, g_nid[m], + "node", g_inst[m]->my_keys.public_key, g_inst[m]->my_ed25519_pubkey, + (uint64_t)time(NULL), time(NULL)); + } + } + int ok = 1; + for (i = 0; i < N_NODES; i++) if (member_sync_count(g_inst[i], CH_ID) < N_NODES) ok = 0; + if (ok) OK(); else FAIL(); + } + + TEST("supernodes node_type=4 in peers table"); { + char sql[256]; + for (i = 0; i < N_NODES; i++) { + snprintf(sql, sizeof(sql), "UPDATE \"peers_%s\" SET node_type=4 WHERE node_id IN (%llu,%llu)", + CH_ID, (unsigned long long)g_nid[I_S1], (unsigned long long)g_nid[I_S2]); + sqlite3_exec(g_inst[i]->topo_sqlite_db, sql, NULL, NULL, NULL); + } + int ok = 1; + for (i = 0; i < N_NODES; i++) { + int n = db_count(g_inst[i]->topo_sqlite_db, "peers_" CH_ID, "node_type", 4); + if (n < 2) ok = 0; + } + if (ok) OK(); else FAIL(); + } + + TEST("media_delivery supernode s1,s2"); { + media_delivery_set_supernode(g_inst[I_S1], 1); + media_delivery_set_supernode(g_inst[I_S2], 1); + if (g_inst[I_S1]->md.is_supernode && g_inst[I_S2]->md.is_supernode) OK(); else FAIL(); + } +} + +static void setup_db_sync(void) { + TEST("db_sync instances on all nodes"); { + int ok = 1; + for (int i = 0; i < N_NODES; i++) { + g_si[i] = db_sync_instance_add(g_inst[i], "msg_" CH_ID, g_group_id, 1); + if (!g_si[i]) { ok = 0; DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "[mdc] db_sync_instance_add failed node=%d", i); } + } + if (ok) OK(); else FAIL(); + } + + TEST("register auto-download insert-cb on storage nodes"); { + db_sync_set_insert_cb(g_si[I_ST1], storage_on_insert, (void*)(intptr_t)I_ST1); + db_sync_set_insert_cb(g_si[I_ST2], storage_on_insert, (void*)(intptr_t)I_ST2); + OK(); + } +} + +/* ── main ── */ + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); + if (getenv("UTUN_TEST_DEBUG")) { + debug_set_category_level_by_name("media", "trace"); + debug_set_category_level_by_name("general", "info"); + debug_set_category_level_by_name("chatsync", "info"); + debug_set_category_level_by_name("etcproute", "info"); + } + utun_instance_set_tun_init_enabled(0); + srand((unsigned)time(NULL)); + printf("=== test_media_delivery_chat ===\n"); + t_mkdtemp(g_tdir); + + for (int i = 0; i < MEDIA_FILE_SIZE; i++) g_file_data[i] = (uint8_t)((i * 31 + 17) & 0xFF); + + for (int i = 0; i < N_NODES; i++) { + snprintf(g_cfg[i], sizeof(g_cfg[i]), "%s/n%d.conf", g_tdir, i); + snprintf(g_db_dir[i], sizeof(g_db_dir[i]), "%s/db%d", g_tdir, i); + utun_mkdir(g_db_dir[i], 0755); + g_port[i] = 52000 + (getpid() % 5000) + i * 1000; + } + + g_ua = uasync_create(); + for (int i = 0; i < N_NODES; i++) + wf(g_cfg[i], "[global]\ntun_ip=10.98.%d.1/24\ntun_ifname=tun%d0\ndb_path=%s\ndb_sync_enabled=1\n" + "[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n", + i, i, g_db_dir[i], g_port[i]); + for (int i = 0; i < N_NODES; i++) { + config_ensure_keys_and_node_id(g_cfg[i]); + struct utun_config* c = parse_config(g_cfg[i]); g_nid[i] = c->global.my_node_id; free_config(c); + } + + for (int i = 0; i < N_NODES; i++) { + char *pr = gv(g_cfg[i], "priv"), *pu = gv(g_cfg[i], "pub"); + if (i == 0) { + wf(g_cfg[i], "[global]\nmy_private_key=%s\nmy_public_key=%s\ntun_ip=10.98.%d.1/24\ntun_ifname=tun%d0\n" + "db_path=%s\ndb_sync_enabled=1\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n", + pr, pu, i, i, g_db_dir[i], g_port[i]); + } else { + int prev = i - 1; char *pv_pu = gv(g_cfg[prev], "pub"); + char link[256]; snprintf(link, sizeof(link), + "[client: to_n%d]\nkeepalive=1\npeer_public_key=%s\nlink=s1:127.0.0.1:%d\n", prev, pv_pu, g_port[prev]); + wf(g_cfg[i], "[global]\nmy_private_key=%s\nmy_public_key=%s\ntun_ip=10.98.%d.1/24\ntun_ifname=tun%d0\n" + "db_path=%s\ndb_sync_enabled=1\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n%s[allowed_keys]\nallow_all=1\n", + pr, pu, i, i, g_db_dir[i], g_port[i], link); + u_free(pv_pu); + } + u_free(pr); u_free(pu); + } + + for (int i = 0; i < N_NODES; i++) { + g_inst[i] = utun_instance_create(g_ua, g_cfg[i]); + if (!g_inst[i]) { printf("FAIL: create n%d\n", i); goto done; } + utun_instance_init(g_inst[i]); + } + + mon(NULL); + uasync_set_timeout(g_ua, TIMEOUT_TB, NULL, (timeout_callback_t)to_cb, "mdc_to"); + { int el = 0; while (!g_connected && !g_result && el < TIMEOUT_TB + 5000) { uasync_poll(g_ua, POLL_MS); el += POLL_MS; } + if (!g_connected) { printf("FAIL: connection timeout\n"); goto done; } } + printf(" connected\n"); + + g_group_id = strtoull(CH_ID, NULL, 10); + + setup_chat_group(); + setup_db_sync(); + + TEST("BGP pubkeys known (db_sync verification)"); { + if (wait_pubkeys()) OK(); else FAIL(); + } + + TEST("author: index file + media_index_commit"); { + if (author_index_file() == 0 && db_count(g_inst[I_N1]->topo_sqlite_db, "media_files", NULL, 0) == MEDIA_NUM_BLOCKS) OK(); + else FAIL(); + } + + TEST("author: publish media message via db_sync"); { + if (author_publish_message() == 0) OK(); else FAIL(); + } + + /* Фаза 1: сообщение распространилось на все узлы */ + TEST("message propagated to all 6 nodes"); { + int a = 0, all = 0; + while (a < 4000) { + all = 1; + for (int i = 0; i < N_NODES; i++) if (db_sync_count(g_si[i]) < 1) all = 0; + if (all) break; + uasync_poll(g_ua, POLL_MS); a++; + } + if (all) OK(); else FAIL("counts: %d %d %d %d %d %d", + db_sync_count(g_si[0]), db_sync_count(g_si[1]), db_sync_count(g_si[2]), + db_sync_count(g_si[3]), db_sync_count(g_si[4]), db_sync_count(g_si[5])); + } + + /* Фаза 2: storage скачали и собрали файл */ + TEST("storage st1/st2 downloaded + assembled media"); { + char d1[1024], d2[1024]; + node_dest(I_ST1, d1, sizeof(d1)); node_dest(I_ST2, d2, sizeof(d2)); + int a = 0; + while (a < 8000 && (!g_dl_done[I_ST1] || !g_dl_done[I_ST2])) { uasync_poll(g_ua, POLL_MS); a++; } + if (g_dl_done[I_ST1] && g_dl_done[I_ST2] && g_dl_err[I_ST1] == 0 && g_dl_err[I_ST2] == 0 + && file_exists(d1) && file_exists(d2)) OK(); + else FAIL("st1 done=%d err=%d exists=%d; st2 done=%d err=%d exists=%d", + g_dl_done[I_ST1], g_dl_err[I_ST1], file_exists(d1), g_dl_done[I_ST2], g_dl_err[I_ST2], file_exists(d2)); + } + + TEST("storage file content hash matches"); { + char d1[1024], d2[1024]; + node_dest(I_ST1, d1, sizeof(d1)); node_dest(I_ST2, d2, sizeof(d2)); + uint8_t h1[32], h2[32]; + int ok = (file_sha256(d1, h1) == 0 && file_sha256(d2, h2) == 0 + && memcmp(h1, g_content_hash, 32) == 0 && memcmp(h2, g_content_hash, 32) == 0); + if (ok) OK(); else FAIL(); + } + + /* Фаза 3: суперузлы узнали о блоках + репликация между собой */ + TEST("supernodes have block_availability (from storage HAVE_BLOCK)"); { + int a = 0, ok = 0; + while (a < 2000) { + int n1 = db_count(g_inst[I_S1]->topo_sqlite_db, "block_availability", NULL, 0); + int n2 = db_count(g_inst[I_S2]->topo_sqlite_db, "block_availability", NULL, 0); + if (n1 >= 2 && n2 >= 2) { ok = 1; break; } + uasync_poll(g_ua, POLL_MS); a++; + } + if (ok) OK(); else FAIL("s1=%d s2=%d", + db_count(g_inst[I_S1]->topo_sqlite_db, "block_availability", NULL, 0), + db_count(g_inst[I_S2]->topo_sqlite_db, "block_availability", NULL, 0)); + } + + TEST("supernode replication s1↔s2 (super_sync)"); { + int a = 0, ok = 0; + while (a < 2000) { + 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]); + if (sqlite3_step(st) == SQLITE_ROW) lr = (uint64_t)sqlite3_column_int64(st, 0); + sqlite3_finalize(st); + if (lr > 0) { ok = 1; break; } + uasync_poll(g_ua, POLL_MS); a++; + } + if (ok) OK(); else FAIL(); + } + + /* Фаза 4: второй клиент n2 скачивает медиа */ + TEST("n2 downloads media (on-demand)"); { + struct media_index_result r; memset(&r, 0, sizeof(r)); + memcpy(r.media_id, g_media_id, 16); + memcpy(r.content_hash, g_content_hash, 32); + r.file_size = MEDIA_FILE_SIZE; r.block_size = MEDIA_BLOCK_SIZE; r.num_blocks = MEDIA_NUM_BLOCKS; + r.block_ids = u_malloc(MEDIA_NUM_BLOCKS * 16); + r.block_sigs = u_malloc(MEDIA_NUM_BLOCKS * 64); + for (int n = 0; n < MEDIA_NUM_BLOCKS; n++) { memcpy(r.block_ids + n * 16, g_block_ids[n], 16); memcpy(r.block_sigs + n * 64, g_block_sigs[n], 64); } + + char dest[1024], mbase[512]; + node_dest(I_N2, dest, sizeof(dest)); + node_media_base(I_N2, mbase, sizeof(mbase)); + ensure_supernode_type(I_N2); + media_download_start(g_inst[I_N2], g_group_id, &r, dest, mbase, g_nid[I_N1], dl_done_cb, (void*)(intptr_t)I_N2, NULL, NULL); + media_index_result_free(&r); + + int a = 0; + while (a < 8000 && !g_dl_done[I_N2]) { uasync_poll(g_ua, POLL_MS); a++; } + if (g_dl_done[I_N2] && g_dl_err[I_N2] == 0 && file_exists(dest)) OK(); + else FAIL("done=%d err=%d exists=%d", g_dl_done[I_N2], g_dl_err[I_N2], file_exists(dest)); + } + + TEST("n2 file content hash matches"); { + char dest[1024]; node_dest(I_N2, dest, sizeof(dest)); + uint8_t h[32]; + if (file_sha256(dest, h) == 0 && memcmp(h, g_content_hash, 32) == 0) OK(); else FAIL(); + } + + fflush(stdout); fflush(stderr); + printf("\n%d/%d passed, %d failed\n", G_PASSED, G_TOTAL, G_FAILED); fflush(stdout); + +done: + g_result = 1; + for (int i = 0; i < N_NODES; i++) { if (g_inst[i]) { g_inst[i]->running = 0; utun_instance_destroy(g_inst[i]); g_inst[i] = NULL; } } + if (g_ua) { uasync_destroy(g_ua, 0); g_ua = NULL; } + t_rmdir(g_tdir); + return G_FAILED > 0 ? 1 : 0; +} diff --git a/tools/chatgui-android/libutun_lite/instance_lite.c b/tools/chatgui-android/libutun_lite/instance_lite.c index 1bd60ec9..0ed6c0c9 100644 --- a/tools/chatgui-android/libutun_lite/instance_lite.c +++ b/tools/chatgui-android/libutun_lite/instance_lite.c @@ -437,7 +437,7 @@ int instance_lite_start(const char* config_text) { debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_INFO); debug_set_category_level(DEBUG_CATEGORY_CHAT_SYNC, DEBUG_LEVEL_INFO); debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_INFO); - debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO); + debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_DEBUG); debug_set_category_level(DEBUG_CATEGORY_ETCP, DEBUG_LEVEL_INFO); debug_set_category_level(DEBUG_CATEGORY_DUMP, DEBUG_LEVEL_WARN); debug_enable_function_name(0);