From d14555db464cba32719179ecda5774013d714697 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 22:16:17 +0100 Subject: [PATCH] test: exercise private mailbox across pairwise CHAT groups --- src/dm/dm_mailbox.h | 2 +- tests/Makefile.am | 5 + tests/test_dm_group_routes.c | 211 +++++++++++++++++++++++++++++++++++ 3 files changed, 217 insertions(+), 1 deletion(-) create mode 100644 tests/test_dm_group_routes.c diff --git a/src/dm/dm_mailbox.h b/src/dm/dm_mailbox.h index f72ed8944..84c2df21e 100644 --- a/src/dm/dm_mailbox.h +++ b/src/dm/dm_mailbox.h @@ -33,7 +33,7 @@ extern "C" { #endif /* Все API в uasync. init после chat_core_init, с живыми SQLite/ua/router ядра: - * таблицы уже созданы dm_core_init; создаёт media-координатор и таймер; 0=успех, -1=ошибка. + * создаёт таблицы, media-координатор и таймер независимо от dm_core_init; 0=успех, -1=ошибка. * destroy отменяет таймеры/bind до освобождения owner, сохраняет pending/receipts в SQLite. * Вызывать после остановки dm_media, до chat_core_destroy; inst=NULL допустим для destroy. */ int dm_mailbox_init(struct UTUN_INSTANCE* inst); diff --git a/tests/Makefile.am b/tests/Makefile.am index b6c84514a..d928dd59c 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -1040,6 +1040,11 @@ test_dm_e2e_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir test_dm_e2e_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) test_dm_e2e_LDFLAGS = $(test_clock_ldflags) +check_PROGRAMS += test_dm_group_routes +test_dm_group_routes_SOURCES = test_dm_group_routes.c test_membership.h test_net.h +test_dm_group_routes_CFLAGS = $(test_dm_e2e_CFLAGS) +test_dm_group_routes_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) + test_dm_invite_SOURCES = test_dm_invite.c $(test_clock_sources) test_dm_invite_CFLAGS = $(test_dm_e2e_CFLAGS) test_dm_invite_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) diff --git a/tests/test_dm_group_routes.c b/tests/test_dm_group_routes.c new file mode 100644 index 000000000..f350ed76e --- /dev/null +++ b/tests/test_dm_group_routes.c @@ -0,0 +1,211 @@ +/* Личные сообщения проходят через попарные группы; общей группы A/S/M/B нет. + * Подписанные участники, настоящий ETCP/router и отдельная SQLite каждого узла. */ +#include +#include +#include +#include "utun_instance.h" +#include "chat/chat_core.h" +#include "chat/member_sync.h" +#include "dm/dm_core.h" +#include "dm/dm_mailbox.h" +#include "routing_layer/topo_group.h" +#include "routing_layer/topo_node_sqlite.h" +#include "routing_layer/etcp_router.h" +#include "routing_layer/route_crypto.h" +#include "transport_layer/etcp.h" +#include "transport_layer/etcp_api.h" +#include "../lib/debug_config.h" +#include "../lib/mem.h" +#include "test_net.h" +#include "test_membership.h" +#include "test_utils.h" + +enum { A, S, M, B, NODES }; +static const unsigned pairs[][2] = {{A,S}, {A,S}, {S,B}, {S,M}, {A,M}, {M,B}, {A,B}}; +#define GROUPS (sizeof(pairs) / sizeof(pairs[0])) +#define REQUIRE(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "DM routes: %s line=%d", #x, __LINE__); return -1; } } while (0) +#define WAIT(x) do { uint64_t end = get_time_tb() + 70000; while (!(x) && get_time_tb() < end) uasync_poll(f.ua, 1); REQUIRE(x); } while (0) + +static struct { + struct UASYNC* ua; + struct UTUN_INSTANCE* node[NODES]; + struct TOPO_GROUP* group[NODES][GROUPS]; + char directory[512], conv[64]; + etcp_recv_fn receiver; + unsigned observed, passed, failed; +} f; + +static void check(const char* name, int ok) { + if (ok) { f.passed++; DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS %s", name); } + else { f.failed++; DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "FAIL %s", name); } +} + +static int value(unsigned node, const char* sql) { + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(f.node[node]->topo_sqlite_db, sql, -1, &st, NULL), result = -1; + if (rc == SQLITE_OK) rc = sqlite3_step(st); + if (rc == SQLITE_ROW) result = sqlite3_column_int(st, 0); + int final = sqlite3_finalize(st); + if (rc != SQLITE_ROW || final != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "DM routes SQL: %s rc=%d final=%d", sql, rc, final); + return result; +} + +static int member(unsigned node, unsigned group, unsigned peer, const char* field, int state) { + char sql[192]; + snprintf(sql, sizeof(sql), "UPDATE peers_%u SET %s=%d WHERE node_id=%lld", 101 + group, field, state, + (long long)f.node[peer]->node_id); + REQUIRE(sqlite3_exec(f.node[node]->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK); + return 0; +} + +static void observed(struct ETCP_CONN* conn, struct ll_entry* entry) { + if (entry->len >= sizeof(struct SVC_ROUTE_HDR) && entry->dgram[ROUTER_SVC_ID_OFF] == ETCP_RT_ID_DM_MAILBOX) f.observed++; + f.receiver(conn, entry); +} + +/* Инъекция подписана настоящим A; обходим только исходящую политику отправителя. */ +static int inject(unsigned group, const uint8_t* payload, size_t len, int signed_packet) { + struct SVC_ROUTE_HDR header = {.cmd = ETCP_ID_SVC_ROUTE, .group_id = 101 + group, + .src_node_id = f.node[A]->node_id, .dst_node_id = f.node[S]->node_id, + .svc_id = ETCP_RT_ID_DM_MAILBOX, .reset_id = 123, .flags = ROUTER_FLAG_START}; + uint8_t plain[2048]; REQUIRE(sizeof(header) + len <= sizeof(plain)); + memcpy(plain, &header, sizeof(header)); memcpy(plain + sizeof(header), payload, len); + uint8_t* encoded = NULL; size_t size = 0; + REQUIRE(!route_crypto_encode(f.node[A], header.dst_node_id, plain, sizeof(header) + len, + signed_packet ? ROUTE_CRYPTO_SIGN : 0, &encoded, &size)); + struct ll_entry* entry = queue_entry_new(0); REQUIRE(entry); + entry->dgram = u_malloc(size + 9); REQUIRE(entry->dgram); + memcpy(entry->dgram, encoded, size); u_free(encoded); + memcpy(entry->dgram + size, &header.src_node_id, 8); entry->dgram[size + 8] = 1; + entry->len = (uint16_t)(size + 9); + struct ETCP_CONN* conn = instance_find_conn(f.node[A], f.node[S]->node_id); REQUIRE(conn && conn->links_up); + unsigned before = f.observed; + REQUIRE(!etcp_send(conn, entry)); + WAIT(f.observed > before); + return 0; +} + +static int setup(void) { + snprintf(f.directory, sizeof(f.directory), "/tmp/utun_dm_routes_XXXXXX"); REQUIRE(!test_mkdtemp(f.directory)); + struct SC_MYKEYS keys[NODES]; char pub[NODES][65], priv[NODES][65]; + for (unsigned i = 0; i < NODES; i++) { + REQUIRE(!sc_generate_keypair(&keys[i])); + bytes_to_hex(keys[i].public_key, 32, pub[i], 65); bytes_to_hex(keys[i].private_key, 32, priv[i], 65); + } + f.ua = uasync_create(); REQUIRE(f.ua); + for (unsigned i = 0; i < NODES; i++) { + char config[4096]; + int len = snprintf(config, sizeof(config), "[global]\nauto_sockets=no\nmy_public_key=%s\nmy_private_key=%s\n" + "db_path=%s/node%u\n[server: udp]\naddr=127.0.0.1:0\ntype=public\n[allowed_keys]\nallow_all=1\n" + "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=1\n", pub[i], priv[i], f.directory, i); + for (unsigned j = 0; j < i; j++) len += snprintf(config + len, sizeof(config) - len, + "[client: peer%u]\npeer_public_key=%s\nlink=udp:127.0.0.1:0\n", j, pub[j]); + f.node[i] = utun_instance_create_from_str(f.ua, config); REQUIRE(f.node[i]); + } + for (unsigned i = 0; i < NODES; i++) { + for (unsigned j = 0; j < i; j++) { + char name[32]; snprintf(name, sizeof(name), "peer%u", j); test_etcp_set_peer(f.node[i], name, f.node[j], "udp"); + } + REQUIRE(!utun_core_start(f.node[i]) && !chat_service_start(f.node[i])); + } + struct UTUN_INSTANCE* owner = f.node[A]; + for (unsigned g = 0; g < GROUPS; g++) { + char channel[32]; snprintf(channel, sizeof(channel), "%u", 101 + g); + uint8_t signature[64]; + REQUIRE(!sc_ed25519_sign(owner->my_ed25519_privkey, (const uint8_t*)channel, strlen(channel), signature)); + for (unsigned side = 0; side < 2; side++) { + unsigned local = pairs[g][side]; struct UTUN_INSTANCE* inst = f.node[local]; + REQUIRE(!topo_node_sqlite_channel_put(inst->topo_sqlite_db, channel, "dm-routes", owner->node_id, + owner->my_keys.public_key, NULL, owner->my_ed25519_pubkey, NULL, signature)); + for (unsigned p = 0; p < 2; p++) { + unsigned index = pairs[g][p]; struct UTUN_INSTANCE* node = f.node[index]; + const char* tags = index == S && g != 1 ? "{\"supernode\":\"yes\",\"ver\":\"1\"}" : + index == M ? "{\"storage\":\"yes\",\"ver\":\"1\"}" : "{\"ver\":\"1\"}"; + uint8_t body[256], tag_signature[64]; + int size = member_sync_build_owner_msg(node->node_id, tags, body, sizeof(body)); REQUIRE(size > 0); + REQUIRE(!sc_ed25519_sign(owner->my_ed25519_privkey, body, size, tag_signature)); + REQUIRE(test_member_seed(inst, channel, node->my_keys.public_key, node->my_ed25519_pubkey, + node->my_ed25519_privkey, owner->my_ed25519_privkey, "{}", tags, tag_signature) >= 0); + } + f.group[local][g] = topo_groups_create_group(inst->topo_groups, 101 + g, TOPO_GROUP_TYPE_CHAT, channel); + REQUIRE(f.group[local][g]); + } + } + for (unsigned g = 0; g < GROUPS; g++) for (unsigned side = 0; side < 2; side++) { + unsigned local = pairs[g][side], peer = pairs[g][1 - side]; + struct ETCP_CONN* conn; + WAIT((conn = instance_find_conn(f.node[local], f.node[peer]->node_id)) && conn->links_up); + REQUIRE(!topo_group_new_conn(f.group[local][g], conn)); + } + for (unsigned g = 0; g < GROUPS; g++) for (unsigned side = 0; side < 2; side++) + WAIT(topo_group_peer_ready(f.group[pairs[g][side]][g], f.node[pairs[g][1-side]]->node_id)); + f.receiver = f.node[S]->api_bindings.callbacks[ETCP_ID_SVC_ROUTE]; REQUIRE(f.receiver); + f.node[S]->api_bindings.callbacks[ETCP_ID_SVC_ROUTE] = observed; + uint64_t conv; + REQUIRE(!dm_start(owner, f.node[B]->node_id, f.node[B]->my_keys.public_key, f.node[B]->my_ed25519_pubkey, + "recipient", "107", DM_MODE_DEVICE, &conv)); + snprintf(f.conv, sizeof(f.conv), "%llu", (unsigned long long)conv); + return 0; +} + +static int run(void) { + uint64_t route = 0; + check("role in 101 permits S while route uses 102", dm_peer_role(f.node[A], f.node[S]->node_id, "supernode") && + dm_route_group(f.node[A], f.node[S]->node_id, 102, &route) && route == 102); + const char* fields[] = {"admitted", "deleted", "source"}; + for (unsigned i = 0; i < 3; i++) { + REQUIRE(!member(A, 0, S, fields[i], i ? 1 : 0)); + check("unadmitted/deleted/foreign-source row gives no role", !dm_peer_role(f.node[A], f.node[S]->node_id, "supernode")); + REQUIRE(!member(A, 0, S, fields[i], i ? 0 : 1)); + } + REQUIRE(!member(A, 0, A, "admitted", 0)); + check("self revocation removes role from that group", !dm_peer_role(f.node[A], f.node[S]->node_id, "supernode")); + REQUIRE(!member(A, 0, A, "admitted", 1)); + /* Отдельная A/B-группа нужна только для начала беседы, затем прямого маршрута нет. */ + REQUIRE(!member(A, 6, B, "admitted", 0) && !member(B, 6, A, "admitted", 0)); + topo_group_revalidate_members(f.group[A][6]); topo_group_revalidate_members(f.group[B][6]); + REQUIRE(!dm_send(f.node[A], f.conv, "text", (const uint8_t*)"pairwise mailbox", 16)); + uint8_t packet[2048] = {1}; + memcpy(packet + 1, &f.node[B]->node_id, 8); + sqlite3_stmt* st = NULL; + REQUIRE(sqlite3_prepare_v2(f.node[A]->topo_sqlite_db, "SELECT body FROM dm_outbox", -1, &st, NULL) == SQLITE_OK); + REQUIRE(sqlite3_step(st) == SQLITE_ROW); int bytes = sqlite3_column_bytes(st, 0); + REQUIRE(bytes > 0 && bytes + 9 <= (int)sizeof(packet)); memcpy(packet + 9, sqlite3_column_blob(st, 0), bytes); + REQUIRE(sqlite3_finalize(st) == SQLITE_OK); + REQUIRE(!member(S, 1, A, "admitted", 0)); + REQUIRE(!inject(1, packet, bytes + 9, 1)); + check("signed PUT through revoked group does not store pending", value(S, "SELECT count(*) FROM dm_pending_messages") == 0); + REQUIRE(!member(S, 1, A, "admitted", 1)); + REQUIRE(!inject(1, packet, bytes + 9, 0)); + check("unsigned mailbox PUT does not store pending", value(S, "SELECT count(*) FROM dm_pending_messages") == 0); + struct ETCP_ROUTER_CONN* conn = etcp_router_conn_get(f.node[A], 102, f.node[S]->node_id, ETCP_RT_ID_DM_MAILBOX); REQUIRE(conn); + REQUIRE(!etcp_router_conn_send(conn, packet, bytes + 9, ROUTER_FLAG_SIGNED)); + WAIT(value(B, "SELECT count(*) FROM dm_messages") == 1); + WAIT(value(A, "SELECT count(*) FROM dm_outbox") == 0); + check("mailbox crosses A/S and S/B groups without common A/S/B group", + value(S, "SELECT count(*) FROM dm_pending_messages") == 0 && value(S, "SELECT count(*) FROM dm_mail_receipts") == 1); + conn = etcp_router_conn_get(f.node[A], 101, f.node[S]->node_id, ETCP_RT_ID_DM_MAILBOX); REQUIRE(conn); + REQUIRE(!etcp_router_conn_send(conn, packet, bytes + 9, ROUTER_FLAG_SIGNED)); + uint64_t end = get_time_tb() + 2000; while (get_time_tb() < end) uasync_poll(f.ua, 1); + check("same PUT over another group preserves receipt and single message", value(S, "SELECT count(*) FROM dm_pending_messages") == 0 && + value(B, "SELECT count(*) FROM dm_messages") == 1 && value(S, "SELECT count(*) FROM dm_mail_receipts") == 1); + check("mailbox persists identities without route column", value(S, + "SELECT count(*) FROM pragma_table_info('dm_pending_messages') WHERE name='group_id'") == 0); + return 0; +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO); + if (getenv("UTUN_TEST_DEBUG")) { + debug_set_category_level(DEBUG_CATEGORY_DM, DEBUG_LEVEL_TRACE); + debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_DEBUG); + debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_INFO); + } + size_t before = u_get_allocated_count(); + if (setup() || run()) f.failed++; + for (unsigned i = 0; i < NODES; i++) if (f.node[i]) utun_instance_destroy(f.node[i]); + if (f.ua) uasync_destroy(f.ua, 0); + check("fixture releases all allocations", before == u_get_allocated_count()); + printf("test_dm_group_routes: %u PASS %u FAIL artifacts=%s\n", f.passed, f.failed, f.directory); + return f.failed ? 1 : 0; +}