jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

test: exercise private mailbox across pairwise CHAT groups

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 22:16:17 +01:00
1 parent c09ab9f344
commit d14555db46
3 files changed
+217 -1
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dm/dm_mailbox.h
+1 -1
View File
@@ -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);
tests/Makefile.am
+5
View File
@@ -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)
tests/test_dm_group_routes.c
+211
View File
@@ -0,0 +1,211 @@
/* Личные сообщения проходят через попарные группы; общей группы A/S/M/B нет.
* Подписанные участники, настоящий ETCP/router и отдельная SQLite каждого узла. */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#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;
}