From 29013ebcccff92148f4687aae2255a5dc6e88704 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 22:13:29 +0100 Subject: [PATCH] dm: route mailbox delivery by each pair of node identities --- src/dm/dm_mailbox.c | 80 ++++++++++++++++----------------------------- src/dm/dm_mailbox.h | 12 ++++--- src/dm/dm_schema.c | 14 ++++++++ 3 files changed, 50 insertions(+), 56 deletions(-) diff --git a/src/dm/dm_mailbox.c b/src/dm/dm_mailbox.c index 1dc0ffee2..e8a79e1f7 100644 --- a/src/dm/dm_mailbox.c +++ b/src/dm/dm_mailbox.c @@ -10,6 +10,7 @@ #include "../ntp_time.h" #include "../routing_layer/etcp_router.h" #include "../routing_layer/topo_node_sqlite.h" +#include "../routing_layer/topo_group.h" #include "../transport_layer/etcp.h" #include "../../lib/mem.h" #include "../../lib/ll_queue.h" @@ -37,33 +38,11 @@ struct dm_mb_media* dm_mailbox_get_media(struct UTUN_INSTANCE* inst) { return inst && inst->dm_mailbox ? inst->dm_mailbox->media : NULL; } -/* Проверить роль в конкретной общей группе; storage не заменяет supernode. */ -static int mb_scope(struct dm_mb_state* mb, uint64_t gid, uint64_t super, uint64_t a, uint64_t b) { - if (!gid) return 0; - char ch[64]; - snprintf(ch, sizeof(ch), "%llu", (unsigned long long)gid); - char sql[256]; - snprintf(sql, sizeof(sql), "SELECT adm_tags FROM \"peers_%s\" WHERE node_id=? AND deleted=0", ch); - sqlite3_stmt* st = NULL; - if (sqlite3_prepare_v2(mb->db, sql, -1, &st, NULL) != SQLITE_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: role query failed group=%llu: %s", MB_ID, - (unsigned long long)gid, sqlite3_errmsg(mb->db)); - return 0; - } - sqlite3_bind_int64(st, 1, (sqlite3_int64)super); - int allowed = 0; - char role[16]; - if (sqlite3_step(st) == SQLITE_ROW) { - const char* tags = (const char*)sqlite3_column_text(st, 0); - allowed = tags && json_flat_get(tags, "supernode", role, sizeof(role)) == 0 && !strcmp(role, "yes"); - } - sqlite3_finalize(st); - return allowed && topo_node_sqlite_member_in_channel(mb->db, ch, a) && - topo_node_sqlite_member_in_channel(mb->db, ch, b); -} - /* Подписать контроль и передать владение пакетом router, учитывая backpressure. */ static int mb_send(struct dm_mb_state* mb, uint64_t gid, uint64_t dst, uint8_t cmd, const uint8_t* body, size_t len) { + if (!dm_route_group(mb->inst, dst, gid, &gid)) { + DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: no admitted route peer=%llu cmd=%u", MB_ID, (unsigned long long)dst, cmd); return -1; + } if (!etcp_router_conn_get(mb->inst, gid, dst, ETCP_RT_ID_DM_MAILBOX)) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: router channel allocation failed peer=%llu", MB_ID, (unsigned long long)dst); return -1; @@ -108,7 +87,7 @@ static void mb_put_received(struct dm_mb_state* mb, uint64_t gid, uint64_t sourc if (dm_verify_message(mb->inst, recipient, p, len) != 0) return; memcpy(&seq, p + 8, 8); memcpy(&author, p + 24, 8); - if (source != author || !mb_scope(mb, gid, mb->inst->node_id, author, recipient)) { + if (source != author || !dm_peer_role(mb->inst, mb->inst->node_id, "supernode")) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unauthorized PUT source=%llu recipient=%llu group=%llu", MB_ID, (unsigned long long)source, (unsigned long long)recipient, (unsigned long long)gid); return; @@ -142,14 +121,13 @@ static void mb_put_received(struct dm_mb_state* mb, uint64_t gid, uint64_t sourc sqlite3_finalize(st); st = NULL; rc = sqlite3_prepare_v2(mb->db, - "INSERT INTO dm_pending_messages(recipient,sender,seq,msg,group_id,retry_at) VALUES(?,?,?,?,?,0)" + "INSERT INTO dm_pending_messages(recipient,sender,seq,msg,retry_at) VALUES(?,?,?,?,0)" " ON CONFLICT(recipient,sender,seq) DO NOTHING", -1, &st, NULL); if (rc != SQLITE_OK) goto fail; sqlite3_bind_int64(st, 1, (sqlite3_int64)recipient); sqlite3_bind_int64(st, 2, (sqlite3_int64)author); sqlite3_bind_int64(st, 3, (sqlite3_int64)seq); sqlite3_bind_blob(st, 4, p, (int)len, SQLITE_STATIC); - sqlite3_bind_int64(st, 5, (sqlite3_int64)gid); rc = sqlite3_step(st); if (rc != SQLITE_DONE) goto fail; sqlite3_finalize(st); @@ -188,7 +166,7 @@ static void mb_receipt_received(struct dm_mb_state* mb, uint64_t gid, uint64_t s dm_accept_receipt(mb->inst, r); return; } - if ((source != author && source != recipient) || !mb_scope(mb, gid, mb->inst->node_id, author, recipient)) { + if (source != author && source != recipient) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: receipt outside authorized scope source=%llu group=%llu", MB_ID, (unsigned long long)source, (unsigned long long)gid); return; @@ -208,14 +186,13 @@ static void mb_receipt_received(struct dm_mb_state* mb, uint64_t gid, uint64_t s sqlite3_finalize(st); st = NULL; rc = sqlite3_prepare_v2(mb->db, - "INSERT INTO dm_mail_receipts(recipient,sender,seq,receipt,group_id) VALUES(?,?,?,?,?)" + "INSERT INTO dm_mail_receipts(recipient,sender,seq,receipt) VALUES(?,?,?,?)" " ON CONFLICT(recipient,sender,seq) DO NOTHING", -1, &st, NULL); if (rc != SQLITE_OK) goto fail; sqlite3_bind_int64(st, 1, (sqlite3_int64)recipient); sqlite3_bind_int64(st, 2, (sqlite3_int64)author); sqlite3_bind_int64(st, 3, (sqlite3_int64)seq); sqlite3_bind_blob(st, 4, r, DM_RECEIPT_SIZE, SQLITE_STATIC); - sqlite3_bind_int64(st, 5, (sqlite3_int64)gid); rc = sqlite3_step(st); if (rc != SQLITE_DONE) goto fail; sqlite3_finalize(st); @@ -258,12 +235,14 @@ static void mb_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unsigned control source=%llu", MB_ID, (unsigned long long)source); goto done; } + if (!topo_group_chat_allowed(mb->inst, gid, source)) { + DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: denied group=%llu source=%llu cmd=%u", MB_ID, + (unsigned long long)gid, (unsigned long long)source, p[0]); goto done; + } if (p[0] == MB_PUT || p[0] == MB_MEDIA_REQUEST) mb_put_received(mb, gid, source, p + 1, len, p[0] == MB_MEDIA_REQUEST); else if (p[0] == MB_DELIVER) { if (len < DM_MSG_FIXED_HDR) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: short delivery", MB_ID); goto done; } - uint64_t author; - memcpy(&author, p + 25, 8); - if (!mb_scope(mb, gid, source, author, mb->inst->node_id)) { + if (!dm_peer_role(mb->inst, source, "supernode")) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: delivery from non-supernode source=%llu", MB_ID, (unsigned long long)source); goto done; } @@ -276,7 +255,7 @@ static void mb_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { memcpy(&conv, p + 1, 8); memcpy(&seq, p + 9, 8); memcpy(&recipient, p + 17, 8); - if (!mb_scope(mb, gid, source, mb->inst->node_id, recipient)) { + if (!dm_peer_role(mb->inst, source, "supernode")) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM mailbox: unauthorized STORED super=%llu", (unsigned long long)source); goto done; } char id[64]; snprintf(id, sizeof(id), "%llu", (unsigned long long)conv); @@ -303,7 +282,7 @@ static void mb_tick(void* arg) { for (int i = 0; i < 32; i++) { sqlite3_stmt* st = NULL; int rc = sqlite3_prepare_v2(mb->db, - "SELECT recipient,sender,seq,msg,group_id FROM dm_pending_messages" + "SELECT recipient,sender,seq,msg FROM dm_pending_messages" " WHERE retry_at<=? ORDER BY retry_at,seq LIMIT 1", -1, &st, NULL); if (rc != SQLITE_OK) goto fail; sqlite3_bind_int64(st, 1, now); @@ -311,7 +290,7 @@ static void mb_tick(void* arg) { if (rc == SQLITE_DONE) { sqlite3_finalize(st); break; } if (rc != SQLITE_ROW) goto fail; uint64_t recipient = (uint64_t)sqlite3_column_int64(st, 0), author = (uint64_t)sqlite3_column_int64(st, 1); - uint64_t seq = (uint64_t)sqlite3_column_int64(st, 2), gid = (uint64_t)sqlite3_column_int64(st, 4); + uint64_t seq = (uint64_t)sqlite3_column_int64(st, 2); size_t len = (size_t)sqlite3_column_bytes(st, 3); uint8_t* msg = u_malloc(len); if (!msg) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: delivery allocation failed", MB_ID); sqlite3_finalize(st); break; } @@ -330,9 +309,8 @@ static void mb_tick(void* arg) { sqlite3_finalize(st); st = NULL; if (rc != SQLITE_DONE) { u_free(msg); goto fail; } - uint64_t route_gid = gid; - if (dm_route_group(mb->inst, recipient, gid, &route_gid) && - mb_scope(mb, route_gid, mb->inst->node_id, author, recipient)) { + uint64_t route_gid = 0; + if (dm_route_group(mb->inst, recipient, 0, &route_gid)) { mb_send(mb, route_gid, recipient, MB_DELIVER, msg, len); DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: pending delivery sender=%llu recipient=%llu seq=%llu group=%llu", MB_ID, (unsigned long long)author, (unsigned long long)recipient, (unsigned long long)seq, @@ -357,13 +335,12 @@ int dm_mailbox_init(struct UTUN_INSTANCE* inst) { if (!mb) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: state allocation failed", MB_ID); return -1; } mb->inst = inst; mb->db = inst->topo_sqlite_db; - if (mb_exec(mb, "CREATE TABLE IF NOT EXISTS dm_pending_messages (recipient INTEGER NOT NULL,sender INTEGER NOT NULL," - "seq INTEGER NOT NULL,msg BLOB NOT NULL,group_id INTEGER NOT NULL,retry_at INTEGER NOT NULL DEFAULT 0," - "PRIMARY KEY(recipient,sender,seq))") != 0 || - mb_exec(mb, "CREATE TABLE IF NOT EXISTS dm_mail_receipts (recipient INTEGER NOT NULL,sender INTEGER NOT NULL," - "seq INTEGER NOT NULL,receipt BLOB NOT NULL,group_id INTEGER NOT NULL," - "PRIMARY KEY(recipient,sender,seq))") != 0) goto fail; if (etcp_router_bind(inst, ETCP_RT_ID_DM_MAILBOX, mb_recv) != 0) goto fail; + if (etcp_router_require_crypto(inst, ETCP_RT_ID_DM_MAILBOX, ROUTER_FLAG_SIGNED) || + etcp_router_set_authorize(inst, ETCP_RT_ID_DM_MAILBOX, topo_group_chat_allowed)) { + DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: router policy setup failed", MB_ID); + etcp_router_unbind(inst, ETCP_RT_ID_DM_MAILBOX); goto fail; + } inst->dm_mailbox = mb; mb->media = dm_mailbox_media_create(inst); if (!mb->media) { @@ -398,7 +375,7 @@ void dm_mailbox_destroy(struct UTUN_INSTANCE* inst) { u_free(mb); } -/* Найти суперузлы в общей группе обеих сторон; для PUT выбрать только один. */ +/* Объявления S ищутся среди своих групп, маршрут до выбранного S выбирается отдельно. */ static int mb_to_supernodes(struct dm_mb_state* mb, uint64_t recipient, uint64_t author, uint8_t cmd, const uint8_t* body, size_t len) { uint64_t* ids = NULL; @@ -410,9 +387,9 @@ static int mb_to_supernodes(struct dm_mb_state* mb, uint64_t recipient, uint64_t for (int i = 0; i < count; i++) { char ch[64], sql[256]; snprintf(ch, sizeof(ch), "%llu", (unsigned long long)ids[i]); - if (!topo_node_sqlite_member_in_channel(mb->db, ch, author) || - !topo_node_sqlite_member_in_channel(mb->db, ch, recipient)) continue; - snprintf(sql, sizeof(sql), "SELECT node_id,adm_tags FROM \"peers_%s\" WHERE deleted=0 ORDER BY node_id", ch); + if (!topo_group_chat_allowed(mb->inst, ids[i], mb->inst->node_id)) continue; + snprintf(sql, sizeof(sql), "SELECT node_id,adm_tags FROM \"peers_%s\"" + " WHERE source=0 AND deleted=0 AND admitted=1 ORDER BY node_id", ch); sqlite3_stmt* st = NULL; if (sqlite3_prepare_v2(mb->db, sql, -1, &st, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: supernode query failed: %s", MB_ID, sqlite3_errmsg(mb->db)); @@ -425,8 +402,7 @@ static int mb_to_supernodes(struct dm_mb_state* mb, uint64_t recipient, uint64_t char role[16]; if (!tags || json_flat_get(tags, "supernode", role, sizeof(role)) != 0 || strcmp(role, "yes")) continue; if (node == mb->inst->node_id || node == recipient || node == author) continue; - struct ETCP_ROUTER_PATH path = etcp_router_get_path(mb->inst, ids[i], node, ETCP_RT_ID_DM_MAILBOX); - if (!path.next_hop_node_id) continue; + if (!topo_group_chat_allowed(mb->inst, ids[i], node)) continue; if (!mb_send(mb, ids[i], node, cmd, body, len)) { sent++; if (cmd == MB_PUT || cmd == MB_MEDIA_REQUEST) break; diff --git a/src/dm/dm_mailbox.h b/src/dm/dm_mailbox.h index e17e53ddf..f72ed8944 100644 --- a/src/dm/dm_mailbox.h +++ b/src/dm/dm_mailbox.h @@ -1,12 +1,16 @@ /* * dm_mailbox.h — очередь offline-сообщений на суперузлах * - * Суперузел общей группы хранит непрозрачное подписанное тело и повторяет + * Суперузел хранит непрозрачное подписанное тело и повторяет * доставку при появлении маршрута до получателя, независимо от отправителя. * Копия удаляется только по квитанции получателя после commit в dm_core. * Квитанция сохраняется отдельно для повторов PUT и доставки отправителю. * - * Роль S — supernode=yes в подписанных adm_tags общей CHAT-группы A/B; storage её не заменяет. + * Роль S — supernode=yes в подписанных adm_tags допущенной общей группы проверяющего узла и S. + * Пары A/S и S/B выбирают маршруты независимо; общая группа A/S/B не требуется. + * Каждая передача проверяет текущий допуск self/peer и подпись router, включая controls. + * Группа не записывается в pending/receipt; отзыв пути сохраняет очередь и позволяет другой маршрут. + * storage не заменяет роль supernode. Квитанция точного сообщения завершает прежнюю работу и после отзыва роли S. * PUT фиксирует сообщение; независимый MEDIA_REQUEST регистрирует задание dm_mailbox_media. * Сервис ETCP_RT_ID_DM_MAILBOX (0x35); все пакеты подписаны router, числа native-endian: * PUT=1 [recipient:8][message]; DELIVER=2 [message]; RECEIPT=3 [receipt:136]; STORED=4 [conv:8][seq:8][recipient:8]. @@ -29,13 +33,13 @@ extern "C" { #endif /* Все API в uasync. init после chat_core_init, с живыми SQLite/ua/router ядра: - * создаёт таблицы, media-координатор и таймер; 0=успех (также повторный init), -1=ошибка. + * таблицы уже созданы dm_core_init; создаёт media-координатор и таймер; 0=успех, -1=ошибка. * destroy отменяет таймеры/bind до освобождения owner, сохраняет pending/receipts в SQLite. * Вызывать после остановки dm_media, до chat_core_destroy; inst=NULL допустим для destroy. */ int dm_mailbox_init(struct UTUN_INSTANCE* inst); void dm_mailbox_destroy(struct UTUN_INSTANCE* inst); -/* Поставить PUT выбранному доступному суперузлу общей с recipient группы. +/* Поставить PUT выбранному доступному суперузлу общей с локальным автором группы. * 0 означает принятие router, хранение подтверждается отдельно на проводе. * До квитанции получателя источник сохраняет задачу в durable outbox. * sender должен совпадать с inst->node_id; msg/len — полное подписанное тело dm_core. diff --git a/src/dm/dm_schema.c b/src/dm/dm_schema.c index ea5123af7..e7402a4ac 100644 --- a/src/dm/dm_schema.c +++ b/src/dm/dm_schema.c @@ -147,6 +147,20 @@ int dm_schema_init(struct UTUN_INSTANCE* inst) { if (rc == SQLITE_OK) rc = sqlite3_step(st); final = sqlite3_finalize(st); if (rc != SQLITE_DONE || final != SQLITE_OK) { schema_error(db, "retain DEVICE outbox recipients"); goto fail; } + if (schema_exec(db, "CREATE TABLE IF NOT EXISTS dm_pending_messages(recipient INTEGER NOT NULL,sender INTEGER NOT NULL," + "seq INTEGER NOT NULL,msg BLOB NOT NULL,retry_at INTEGER NOT NULL DEFAULT 0,PRIMARY KEY(recipient,sender,seq));" + "CREATE TABLE IF NOT EXISTS dm_mail_receipts(recipient INTEGER NOT NULL,sender INTEGER NOT NULL," + "seq INTEGER NOT NULL,receipt BLOB NOT NULL,PRIMARY KEY(recipient,sender,seq))")) goto fail; + const char* mailbox_tables[] = {"dm_pending_messages", "dm_mail_receipts"}; + for (unsigned i = 0; i < 2; i++) { + found = schema_has_column(db, mailbox_tables[i], "group_id"); + if (found < 0) goto fail; + if (found) { + char sql[128]; snprintf(sql, sizeof(sql), "ALTER TABLE %s DROP COLUMN group_id", mailbox_tables[i]); + if (schema_exec(db, sql)) goto fail; + DEBUG_INFO(DEBUG_CATEGORY_DM, "DM mailbox schema: retained %s, removed route binding", mailbox_tables[i]); + } + } if (schema_exec(db, "RELEASE dm_schema")) goto fail; DEBUG_INFO(DEBUG_CATEGORY_DM, "DM schema: ready; author sequence, frozen parties and per-device delivery"); return 0;