jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
d840688e96b8bbf8c473a48f7f329e343143dba8
utun2/src/dm/dm_core.c
T

1512 lines
77 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/*
* dm_core.c — прямой p2p чат между двумя пользователями (DM)
*
* Отдельная подсистема (не канал): беседы хранятся в dm_conversations,
* сообщения в dm_messages, синхронизация идёт напрямую через etcp_router
* (сервис 0x34), при недоступности пира — через dm_mailbox (0x35).
*
* Два независимых направленных потока (я→пир, пир→я) с монотонным seq.
* Каноническое тело сообщения — см. dm_core.h (общее с dm_mailbox).
*/
#include "dm_core.h"
#include "dm_crypto.h"
#include "dm_history.h"
#include "dm_schema.h"
#include "dm_mailbox.h"
#include "../chat/chat_core.h"
#include "../chat/chat_event.h"
#include "../chat/invite_build.h"
#include "../chat/chat_join.h"
#include "../utun_instance.h"
#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_api.h"
#include "../transport_layer/etcp.h"
#include "../transport_layer/secure_channel.h"
#include "../../lib/mem.h"
#include "../../lib/ll_queue.h"
#include "../../lib/debug_config.h"
#include "../../lib/platform_compat.h"
#include "../../lib/strbuf.h"
#include <string.h>
#include <stdlib.h>
#include <sqlite3.h>
#include <openssl/sha.h>
#include <openssl/crypto.h>
#define DM_ID "dm_core"
/* Подкоманды протокола DM (первый байт payload после ROUTER_SVC-заголовка). */
#define DM_SUBCMD_MSG 0x01 /* [msg] — сообщение */
#define DM_SUBCMD_ACK 0x02 /* Подписанная квитанция получателя. */
/* Per-instance состояние модуля. */
struct dm_state {
struct UTUN_INSTANCE* inst;
sqlite3* db;
int initialized;
void* timer;
struct dm_invite_req* invite;
};
/* ── вспомогательные ── */
static struct dm_state* dm_of(struct UTUN_INSTANCE* inst) {
return inst ? inst->dm : NULL;
}
/* Выполнить SQL без результата. Возвращает 0/ошибку. */
static int dm_exec(struct dm_state* dm, const char* sql) {
char* err = NULL;
if (sqlite3_exec(dm->db, sql, NULL, NULL, &err) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: sql failed: %s (%s)", DM_ID, sql, err ? err : "?");
if (err) sqlite3_free(err);
return -1;
}
return 0;
}
/* Прочитать оба pubkey узла (x25519 + ed25519) из таблицы nodes. 0=ок, <0=нет. */
static int dm_node_pubkeys(struct dm_state* dm, uint64_t node_id, uint8_t x25519[32], uint8_t ed25519[32]) {
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"SELECT x25519_pubkey, ed25519_pubkey FROM nodes WHERE node_id=?",
-1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: node keys query failed: %s", DM_ID, sqlite3_errmsg(dm->db));
return -1;
}
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
int rc = -1;
if (sqlite3_step(st) == SQLITE_ROW) {
const uint8_t* x = (const uint8_t*)sqlite3_column_blob(st, 0);
const uint8_t* e = (const uint8_t*)sqlite3_column_blob(st, 1);
if (x && e && sqlite3_column_bytes(st, 0) == 32 && sqlite3_column_bytes(st, 1) == 32) {
memcpy(x25519, x, 32);
memcpy(ed25519, e, 32);
rc = 0;
}
}
sqlite3_finalize(st);
if (rc) DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: missing node keys peer=%llu", DM_ID, (unsigned long long)node_id);
return rc;
}
/* Общий CHAT-канал с пиром. Возвращает 1 и кладёт group_id (== channel_id) в *out_group_id. */
static int dm_shared_group_id(struct dm_state* dm, uint64_t peer_id, uint64_t* out_group_id) {
if (!dm || !out_group_id) return 0;
uint64_t* ch_ids = NULL;
int ch_count = 0;
if (topo_node_sqlite_get_member_channels(dm->db, dm->inst->node_id, &ch_ids, &ch_count) != 0)
return 0;
int shared = 0;
for (int i = 0; i < ch_count; i++) {
char ch[64]; snprintf(ch, sizeof(ch), "%llu", (unsigned long long)ch_ids[i]);
if (topo_node_sqlite_member_in_channel(dm->db, ch, peer_id)) {
*out_group_id = ch_ids[i];
shared = 1;
break;
}
}
if (ch_ids) u_free(ch_ids);
return shared;
}
/* Выбрать общую область с действительно живым маршрутом, включая BGP-транзит. */
static int dm_find_route_group(struct UTUN_INSTANCE* inst, uint64_t peer, uint64_t preferred,
uint64_t* group_id, int bgp_only) {
if (!inst || !inst->topo_sqlite_db || !group_id) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: invalid route lookup", DM_ID);
return 0;
}
uint64_t* ids = NULL;
int count = 0;
if (topo_node_sqlite_get_member_channels(inst->topo_sqlite_db, inst->node_id, &ids, &count) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: route membership lookup failed", DM_ID);
return 0;
}
int found = 0;
for (int i = -1; i < count; i++) {
uint64_t gid = i < 0 ? preferred : ids[i];
if (!gid) continue;
char ch[64];
snprintf(ch, sizeof(ch), "%llu", (unsigned long long)gid);
if (!topo_node_sqlite_member_in_channel(inst->topo_sqlite_db, ch, inst->node_id) ||
!topo_node_sqlite_member_in_channel(inst->topo_sqlite_db, ch, peer)) continue;
if (bgp_only) {
struct TOPO_GROUP* g = topo_groups_find(inst->topo_groups, gid);
if (!topo_group_find_conn_for_node(g, peer)) continue;
} else {
struct ETCP_ROUTER_PATH path = etcp_router_get_path(inst, gid, peer, ETCP_RT_ID_DM);
if (!path.next_hop_node_id) continue;
}
*group_id = gid;
found = 1;
break;
}
u_free(ids);
return found;
}
int dm_route_group(struct UTUN_INSTANCE* inst, uint64_t peer, uint64_t preferred, uint64_t* group_id) {
return dm_find_route_group(inst, peer, preferred, group_id, 0);
}
/* Router владеет entry после вызова, в том числе при ошибке. */
static int dm_route_send(struct dm_state* dm, uint64_t group_id, uint64_t dst, uint8_t subcmd,
const uint8_t* body, size_t body_len) {
if (!etcp_router_conn_get(dm->inst, group_id, dst, ETCP_RT_ID_DM)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: router channel allocation failed peer=%llu", DM_ID, (unsigned long long)dst);
return -1;
}
if (!etcp_router_send_q_has_room(dm->inst, group_id, dst, ETCP_RT_ID_DM)) {
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: send deferred by backpressure peer=%llu", DM_ID, (unsigned long long)dst);
return -1;
}
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: packet allocation failed", DM_ID); return -1; }
entry->dgram = u_malloc(2 + body_len);
if (!entry->dgram) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: payload allocation failed", DM_ID);
queue_entry_free(entry);
return -1;
}
entry->dgram[0] = ETCP_RT_ID_DM;
entry->dgram[1] = subcmd;
if (body_len) memcpy(entry->dgram + 2, body, body_len);
entry->len = (uint16_t)(2 + body_len);
int rc = etcp_route_send(dm->inst, group_id, dst, entry, 1, ROUTER_FLAG_SIGNED);
if (rc) DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: route send failed peer=%llu cmd=%u rc=%d",
DM_ID, (unsigned long long)dst, subcmd, rc);
return rc;
}
/* ── работа с беседой ── */
/* Полный локальный снимок беседы с выведенным content_key. */
struct dm_conv {
char conv_id[64];
uint64_t peer_node_id;
uint8_t peer_x25519[32];
uint8_t peer_ed25519[32];
char peer_name[DM_PEER_NAME_MAX + 1];
uint64_t group_id;
uint8_t content_key[DM_CONTENT_KEY_SIZE];
uint64_t last_out_seq;
uint64_t last_in_seq;
uint64_t last_ts;
enum dm_mode mode;
struct dm_party parties[2]; /* Локальная сторона, собеседник; неизменны до удаления беседы. */
uint8_t stored_key[DM_KEY_WRAP_SIZE]; /* Content key, защищённый ключом конкретного устройства. */
int stored_key_valid, closed;
};
enum dm_conv_result { DM_CONV_ERROR = -1, DM_CONV_FOUND = 0, DM_CONV_NOT_FOUND = 1 };
/* Привязка локальной упаковки к полной беседе защищает от перестановки строк/ключей. */
static size_t dm_local_key_info(const struct dm_conv* c, uint8_t info[192]) {
const char domain[] = "utun-dm-local-history-v1"; size_t pos = sizeof(domain);
uint64_t conv = strtoull(c->conv_id, NULL, 10);
memcpy(info, domain, pos);
for (unsigned i = 0; i < 8; i++) info[pos++] = (uint8_t)(conv >> (8 * i));
if (dm_party_encode(c->parties, info + pos) || dm_party_encode(c->parties + 1, info + pos + DM_PARTY_WIRE_SIZE)) return 0;
return pos + 2 * DM_PARTY_WIRE_SIZE;
}
/* Архив читаем, но новые операции требуют текущей принадлежности закреплённой стороне. */
static int dm_conv_active(struct dm_state* dm, const struct dm_conv* c) {
if (!c->closed && (c->mode == DM_MODE_DEVICE || dm_party_member(dm->inst, c->parties, dm->inst->node_id) == 1)) return 1;
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM conversation is archived/inactive conv=%s mode=%d", c->conv_id, c->mode);
return 0;
}
/* Копировать целый UTF-8 префикс; NUL внутри SQL-строки и битый суффикс не попадают в имя/JSON. */
static void dm_copy_peer_name(char out[DM_PEER_NAME_MAX + 1], const char* name, size_t len, const char* conv) {
size_t size = 0;
while (size < len && size < DM_PEER_NAME_MAX) {
unsigned c = (unsigned char)name[size];
if (c && c < 0x80) { size++; continue; }
unsigned count, scalar, minimum;
if (c >= 0xc2 && c <= 0xdf) { count = 1; scalar = c & 31; minimum = 0x80; }
else if (c >= 0xe0 && c <= 0xef) { count = 2; scalar = c & 15; minimum = 0x800; }
else if (c >= 0xf0 && c <= 0xf4) { count = 3; scalar = c & 7; minimum = 0x10000; }
else goto invalid;
if (count > len - size - 1) goto invalid;
for (unsigned i = 1; i <= count; i++) {
c = (unsigned char)name[size + i];
if ((c & 0xc0) != 0x80) goto invalid;
scalar = (scalar << 6) | (c & 63);
}
if (scalar < minimum || scalar > 0x10ffff || (scalar >= 0xd800 && scalar <= 0xdfff)) goto invalid;
if (count + 1 > DM_PEER_NAME_MAX - size) break;
size += count + 1;
}
if (size < len) DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: peer name shortened conv=%s bytes=%zu->%zu", DM_ID, conv, len, size);
goto done;
invalid:
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: invalid peer name encoding conv=%s offset=%zu bytes=%zu; suffix omitted",
DM_ID, conv, size, len);
done:
if (size) memcpy(out, name, size);
out[size] = '\0';
}
/* FOUND выдаёт полную беседу; NOT_FOUND — только SQLITE_DONE; ERROR не меняет *c и БД. */
static enum dm_conv_result dm_conv_load(struct dm_state* dm, const char* conv_id, struct dm_conv* c) {
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(dm->db,
"SELECT peer_node_id, peer_x25519, peer_ed25519, COALESCE(peer_name,''),"
" COALESCE(group_id,0), last_out_seq, last_in_seq, last_ts,mode,local_party,peer_party,local_key,closed"
" FROM dm_conversations WHERE conv_id=?",
-1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
rc = sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
if (rc != SQLITE_OK) goto fail;
rc = sqlite3_step(st);
if (rc == SQLITE_DONE) {
rc = sqlite3_finalize(st);
st = NULL;
if (rc != SQLITE_OK) goto fail;
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: conversation not found conv=%s", DM_ID, conv_id);
return DM_CONV_NOT_FOUND;
}
if (rc != SQLITE_ROW) goto fail;
struct dm_conv loaded = {0};
snprintf(loaded.conv_id, sizeof(loaded.conv_id), "%s", conv_id);
loaded.peer_node_id = (uint64_t)sqlite3_column_int64(st, 0);
const uint8_t* x = (const uint8_t*)sqlite3_column_blob(st, 1);
const uint8_t* e = (const uint8_t*)sqlite3_column_blob(st, 2);
if (!x || !e || sqlite3_column_bytes(st, 1) != 32 || sqlite3_column_bytes(st, 2) != 32) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: damaged conversation keys conv=%s", DM_ID, conv_id);
sqlite3_finalize(st);
return DM_CONV_ERROR;
}
memcpy(loaded.peer_x25519, x, 32);
memcpy(loaded.peer_ed25519, e, 32);
const char* name = (const char*)sqlite3_column_text(st, 3);
dm_copy_peer_name(loaded.peer_name, name, (size_t)sqlite3_column_bytes(st, 3), conv_id);
loaded.group_id = (uint64_t)sqlite3_column_int64(st, 4);
loaded.last_out_seq = (uint64_t)sqlite3_column_int64(st, 5);
loaded.last_in_seq = (uint64_t)sqlite3_column_int64(st, 6);
loaded.last_ts = (uint64_t)sqlite3_column_int64(st, 7);
loaded.mode = sqlite3_column_int(st, 8); loaded.closed = sqlite3_column_int(st, 12);
if (loaded.mode == DM_MODE_GROUPS) {
if (sqlite3_column_bytes(st, 9) != DM_PARTY_WIRE_SIZE || sqlite3_column_bytes(st, 10) != DM_PARTY_WIRE_SIZE ||
sqlite3_column_bytes(st, 11) != DM_KEY_WRAP_SIZE || dm_party_decode(sqlite3_column_blob(st, 9), loaded.parties) ||
dm_party_decode(sqlite3_column_blob(st, 10), loaded.parties + 1) ||
dm_identity_conv_id(loaded.mode, loaded.parties) != strtoull(conv_id, NULL, 10)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM conversation: invalid pinned identity conv=%s", conv_id);
sqlite3_finalize(st); return DM_CONV_ERROR;
}
memcpy(loaded.stored_key, sqlite3_column_blob(st, 11), DM_KEY_WRAP_SIZE); loaded.stored_key_valid = 1;
} else if (loaded.mode == DM_MODE_DEVICE) {
loaded.parties[0].node = dm->inst->node_id; loaded.parties[1].node = loaded.peer_node_id;
memcpy(loaded.parties[0].x, dm->inst->my_keys.public_key, 32); memcpy(loaded.parties[0].ed, dm->inst->my_ed25519_pubkey, 32);
memcpy(loaded.parties[1].x, loaded.peer_x25519, 32); memcpy(loaded.parties[1].ed, loaded.peer_ed25519, 32);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM conversation: unknown mode=%d conv=%s", loaded.mode, conv_id);
sqlite3_finalize(st); return DM_CONV_ERROR;
}
/* Если область ещё не выбрана, используем известную общую группу. */
if (loaded.group_id == 0) {
uint64_t gid = 0;
if (dm_shared_group_id(dm, loaded.peer_node_id, &gid)) loaded.group_id = gid;
}
rc = sqlite3_finalize(st);
st = NULL;
if (rc != SQLITE_OK) goto fail;
int key_error;
if (loaded.mode == DM_MODE_GROUPS) {
uint8_t info[192]; size_t info_size = dm_local_key_info(&loaded, info);
key_error = !info_size || dm_key_open(dm->inst->my_keys.private_key, info, info_size, loaded.stored_key, loaded.content_key);
} else key_error = dm_derive_content_key(dm->inst->my_keys.private_key, loaded.peer_x25519, loaded.content_key);
if (key_error) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation key derivation failed conv=%s", DM_ID, conv_id);
return DM_CONV_ERROR;
}
*c = loaded;
DEBUG_TRACE(DEBUG_CATEGORY_DM, "%s: conversation loaded conv=%s out=%llu in=%llu", DM_ID, conv_id,
(unsigned long long)c->last_out_seq, (unsigned long long)c->last_in_seq);
return DM_CONV_FOUND;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation load failed conv=%s rc=%d sql=%s", DM_ID, conv_id, rc, sqlite3_errmsg(dm->db));
sqlite3_finalize(st);
return DM_CONV_ERROR;
}
/* Вставить/обновить беседу в БД. Возвращает 0/ошибку. */
static int dm_conv_save(struct dm_state* dm, const struct dm_conv* c) {
sqlite3_stmt* st = NULL;
uint8_t parties[2][DM_PARTY_WIRE_SIZE], stored[DM_KEY_WRAP_SIZE];
if (c->mode == DM_MODE_GROUPS) {
if (dm_party_encode(c->parties, parties[0]) || dm_party_encode(c->parties + 1, parties[1])) return -1;
if (c->stored_key_valid) memcpy(stored, c->stored_key, sizeof(stored));
else {
uint8_t info[192]; size_t size = dm_local_key_info(c, info);
if (!size || dm_key_seal(dm->inst->my_keys.public_key, info, size, c->content_key, stored)) return -1;
}
}
if (sqlite3_prepare_v2(dm->db,
"INSERT INTO dm_conversations(conv_id,peer_node_id,peer_x25519,peer_ed25519,"
" peer_name,group_id,last_out_seq,last_in_seq,created_at,last_ts,mode,local_party,peer_party,local_key,closed)"
" VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
" ON CONFLICT(conv_id) DO UPDATE SET"
" peer_x25519=excluded.peer_x25519, peer_ed25519=excluded.peer_ed25519,"
" peer_name=excluded.peer_name, group_id=excluded.group_id,"
" last_out_seq=excluded.last_out_seq, last_in_seq=excluded.last_in_seq, last_ts=excluded.last_ts,"
" mode=excluded.mode,local_party=excluded.local_party,peer_party=excluded.peer_party,local_key=excluded.local_key,closed=excluded.closed",
-1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation save prepare failed: %s", DM_ID, sqlite3_errmsg(dm->db));
return -1;
}
sqlite3_bind_text(st, 1, c->conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)c->peer_node_id);
sqlite3_bind_blob(st, 3, c->peer_x25519, 32, SQLITE_STATIC);
sqlite3_bind_blob(st, 4, c->peer_ed25519, 32, SQLITE_STATIC);
sqlite3_bind_text(st, 5, c->peer_name, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 6, (sqlite3_int64)c->group_id);
sqlite3_bind_int64(st, 7, (sqlite3_int64)c->last_out_seq);
sqlite3_bind_int64(st, 8, (sqlite3_int64)c->last_in_seq);
sqlite3_bind_int64(st, 9, (sqlite3_int64)ntp_time_get_seconds(dm->inst));
sqlite3_bind_int64(st, 10, (sqlite3_int64)c->last_ts);
sqlite3_bind_int(st, 11, c->mode);
if (c->mode == DM_MODE_GROUPS) {
sqlite3_bind_blob(st, 12, parties[0], DM_PARTY_WIRE_SIZE, SQLITE_STATIC);
sqlite3_bind_blob(st, 13, parties[1], DM_PARTY_WIRE_SIZE, SQLITE_STATIC);
sqlite3_bind_blob(st, 14, stored, DM_KEY_WRAP_SIZE, SQLITE_STATIC);
}
sqlite3_bind_int(st, 15, c->closed);
int rc = sqlite3_step(st) == SQLITE_DONE ? 0 : -1;
if (rc) DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation save failed conv=%s: %s", DM_ID, c->conv_id, sqlite3_errmsg(dm->db));
sqlite3_finalize(st);
return rc;
}
/* ── сборка/разбор канонического тела ── */
/* Собрать каноническое тело сообщения (зашифровать + подписать).
* out — буфер (u_malloc), *out_len — длина. dir=1 для исходящего. */
static int dm_build_body(struct dm_state* dm, const struct dm_conv* c, uint64_t seq, uint64_t ts,
const char* ct, const uint8_t* data, uint32_t data_len,
uint8_t** out, size_t* out_len) {
if (!c || !ct || (!data && data_len) || data_len > DM_DATA_MAX || strlen(ct) > 255) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: invalid message size=%u", DM_ID, data_len);
return -1;
}
uint8_t ct_len = (uint8_t)strnlen(ct, 255);
const uint8_t* media = NULL;
if (!strcmp(ct, "file")) {
if (data_len <= DM_MEDIA_DESC_SIZE) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: file message lacks descriptor/name", DM_ID);
return -1;
}
media = data;
data += DM_MEDIA_DESC_SIZE;
data_len -= DM_MEDIA_DESC_SIZE;
struct attachment_info info;
if (attachment_decode(data, data_len, &info)) return -1;
}
uint64_t conv_num = (uint64_t)strtoull(c->conv_id, NULL, 10);
uint8_t nonce[DM_NONCE_SIZE];
dm_build_nonce(conv_num, dm->inst->node_id, seq, nonce);
/* шифротекст = data_len + tag(16) */
uint8_t* enc = u_malloc(data_len + DM_TAG_SIZE);
if (!enc) return -1;
size_t enc_len = 0;
if (media_ccm_encrypt(c->content_key, nonce, data, data_len, enc, &enc_len) != 0) {
u_free(enc);
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: encrypt failed conv=%s", DM_ID, c->conv_id);
return -1;
}
uint64_t author = dm->inst->node_id;
uint8_t context[DM_CONTEXT_MAX]; size_t context_size = 0;
if (c->mode == DM_MODE_GROUPS && dm_context_encode(dm->inst, c->parties, context, sizeof(context), &context_size)) {
u_free(enc); return -1;
}
size_t body_len = DM_MSG_FIXED_HDR + ct_len + 2 + enc_len + (media ? DM_MEDIA_DESC_SIZE : 0) + DM_MSG_SIG_SIZE;
if (context_size) body_len += context_size + 1;
uint8_t* body = u_malloc(body_len);
if (!body) { u_free(enc); return -1; }
size_t off = 0;
memcpy(body + off, &conv_num, 8); off += 8;
memcpy(body + off, &seq, 8); off += 8;
memcpy(body + off, &ts, 8); off += 8;
memcpy(body + off, &author, 8); off += 8;
if (context_size) { body[off++] = 0; memcpy(body + off, context, context_size); off += context_size; }
body[off++] = ct_len;
memcpy(body + off, ct, ct_len); off += ct_len;
uint16_t el = (uint16_t)enc_len;
memcpy(body + off, &el, 2); off += 2;
memcpy(body + off, enc, enc_len); off += enc_len;
if (media) { memcpy(body + off, media, DM_MEDIA_DESC_SIZE); off += DM_MEDIA_DESC_SIZE; }
/* подпись по всем полям до sig */
if (sc_ed25519_sign(dm->inst->my_ed25519_privkey, body, off, body + off) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: message signing failed", DM_ID);
u_free(enc);
u_free(body);
return -1;
}
u_free(enc);
*out = body;
*out_len = body_len;
return 0;
}
/* Проверить тело перед любой записью: подпись связывает conv, автора и адресата. */
int dm_verify_message(struct UTUN_INSTANCE* inst, uint64_t recipient, const uint8_t* msg, size_t len) {
struct dm_state view = { .inst = inst, .db = inst ? inst->topo_sqlite_db : NULL };
struct dm_message_view message;
if (!view.db || dm_message_parse(msg, len, &message) || message.size != len) goto invalid;
uint64_t conv = message.conv, seq = message.seq, author = message.author;
if (!recipient || !author || author == recipient || !seq || seq > INT64_MAX) goto invalid;
uint8_t x[32], ed[32];
if (message.mode == DM_MODE_GROUPS) {
if (dm_message_verify_targets(&message) || dm_message_target(&message, recipient, NULL, NULL) < 0 ||
dm_message_target(&message, author, x, ed) != 0) goto invalid;
} else if (conv != dm_derive_conv_id(author, recipient) || dm_node_pubkeys(&view, author, x, ed)) goto invalid;
const uint8_t* media = message.media;
if (media) {
uint64_t size;
memcpy(&size, media + 16, 8);
if (size > DM_MEDIA_MAX_SIZE) goto invalid;
}
if (sc_ed25519_verify(ed, msg, len - DM_MSG_SIG_SIZE, message.signature) != SC_OK) goto invalid;
return 0;
invalid:
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: rejected message recipient=%llu bytes=%zu", DM_ID,
(unsigned long long)recipient, len);
return -1;
}
/* Сохранить тело; точный повтор допустим, другое тело с тем же seq запрещено. Транзакция у вызывающего. */
static int dm_store_message(struct dm_state* dm, const char* conv, int dir, const uint8_t* msg, size_t len) {
struct dm_message_view view;
if (dm_message_parse(msg, len, &view) || view.size != len) return -1;
uint64_t seq = view.seq, ts = view.timestamp, author = view.author;
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(dm->db,
"INSERT INTO dm_messages(conv_id,dir,seq,ts,author,ct,data,sig,body) VALUES(?,?,?,?,?,?,?,?,?)"
" ON CONFLICT(conv_id,author,seq) DO NOTHING", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
sqlite3_bind_int(st, 2, dir);
sqlite3_bind_int64(st, 3, (sqlite3_int64)seq);
sqlite3_bind_int64(st, 4, (sqlite3_int64)ts);
sqlite3_bind_int64(st, 5, (sqlite3_int64)author);
sqlite3_bind_text(st, 6, (const char*)view.content_type, view.content_type_size, SQLITE_STATIC);
sqlite3_bind_blob(st, 7, view.ciphertext, view.ciphertext_size, SQLITE_STATIC);
sqlite3_bind_blob(st, 8, view.signature, DM_MSG_SIG_SIZE, SQLITE_STATIC);
sqlite3_bind_blob(st, 9, msg, (int)len, SQLITE_STATIC);
rc = sqlite3_step(st);
if (rc != SQLITE_DONE) goto fail;
int inserted = sqlite3_changes(dm->db);
sqlite3_finalize(st);
st = NULL;
if (inserted) return 1;
rc = sqlite3_prepare_v2(dm->db, "SELECT sig,body FROM dm_messages WHERE conv_id=? AND author=? AND seq=?", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)author);
sqlite3_bind_int64(st, 3, (sqlite3_int64)seq);
rc = sqlite3_step(st);
if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != DM_MSG_SIG_SIZE) goto fail;
if (memcmp(sqlite3_column_blob(st, 0), view.signature, DM_MSG_SIG_SIZE) ||
(sqlite3_column_type(st, 1) != SQLITE_NULL &&
(sqlite3_column_bytes(st, 1) != (int)len || memcmp(sqlite3_column_blob(st, 1), msg, len)))) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: conflicting body for saved message conv=%s dir=%d seq=%llu", DM_ID,
conv, dir, (unsigned long long)seq);
sqlite3_finalize(st);
return -1;
}
sqlite3_finalize(st);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: message write/conflict conv=%s dir=%d seq=%llu rc=%d sql=%s",
DM_ID, conv, dir, (unsigned long long)seq, rc, sqlite3_errmsg(dm->db));
sqlite3_finalize(st);
return -1;
}
/* Принять сообщение и создать подписанную квитанцию только после успешного commit. */
int dm_accept_message(struct UTUN_INSTANCE* inst, const uint8_t* msg, size_t len, uint8_t receipt[DM_RECEIPT_SIZE]) {
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized || !receipt) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: receive without initialized owner", DM_ID);
return -1;
}
if (dm_verify_message(inst, inst->node_id, msg, len) != 0) return -1;
struct dm_message_view message;
if (dm_message_parse(msg, len, &message)) return -1;
int side = message.mode == DM_MODE_GROUPS ? dm_message_target(&message, inst->node_id, NULL, NULL) : 1;
int dir = side == 0;
int deleted = dm_history_deleted(inst, msg, len, dir, NULL);
if (deleted < 0) return -1;
uint64_t conv, seq, author;
memcpy(&conv, msg, 8);
memcpy(&seq, msg + 8, 8);
memcpy(&author, msg + 24, 8);
char conv_id[64];
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv);
if (deleted) goto receipt;
struct dm_conv c;
enum dm_conv_result loaded = dm_conv_load(dm, conv_id, &c);
if (loaded == DM_CONV_ERROR) return -1;
int created = loaded == DM_CONV_NOT_FOUND;
if (created) {
memset(&c, 0, sizeof(c));
snprintf(c.conv_id, sizeof(c.conv_id), "%s", conv_id);
c.mode = message.mode;
c.peer_node_id = author;
if (!dm_shared_group_id(dm, author, &c.group_id)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unknown correspondent peer=%llu", DM_ID, (unsigned long long)author);
return -1;
}
if (c.mode == DM_MODE_GROUPS) {
c.parties[0] = message.parties[side]; c.parties[1] = message.parties[1 - side];
if (!side) {
c.peer_node_id = 0;
for (unsigned i = 0; i < message.target_count; i++) {
struct dm_target target;
if (dm_message_target_at(&message, i, &target)) return -1;
if (target.side == 1) { c.peer_node_id = target.node; break; }
}
}
memcpy(c.peer_x25519, c.parties[1].x, 32); memcpy(c.peer_ed25519, c.parties[1].ed, 32);
uint8_t private_key[32];
if (dm_party_private(inst, c.parties, private_key)) return -1;
int error = dm_derive_content_key(private_key, c.parties[1].x, c.content_key);
OPENSSL_cleanse(private_key, sizeof(private_key));
if (error) return -1;
} else if (dm_node_pubkeys(dm, author, c.peer_x25519, c.peer_ed25519) != 0 ||
dm_derive_content_key(inst->my_keys.private_key, c.peer_x25519, c.content_key) != 0) return -1;
char name[128] = {0};
chat_core_get_node_name(inst, c.peer_node_id, name, sizeof(name));
dm_copy_peer_name(c.peer_name, name, strlen(name), conv_id);
}
if (c.mode != message.mode || (c.mode == DM_MODE_DEVICE && c.peer_node_id != author) ||
(c.mode == DM_MODE_GROUPS && (!dm_party_equal(c.parties, message.parties + side) ||
!dm_party_equal(c.parties + 1, message.parties + 1 - side)))) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation author mismatch conv=%s", DM_ID, conv_id);
return -1;
}
if (!dm_conv_active(dm, &c)) return -1;
uint8_t nonce[DM_NONCE_SIZE], plain[DM_DATA_MAX + 1];
size_t plain_len = 0;
dm_build_nonce(conv, author, seq, nonce);
if (media_ccm_decrypt(c.content_key, nonce, message.ciphertext, message.ciphertext_size, plain, &plain_len) != 0) return -1;
if (dm_message_media(msg, len)) {
struct attachment_info info;
if (attachment_decode(plain, plain_len, &info)) return -1;
}
if (dm_exec(dm, "BEGIN IMMEDIATE") != 0) return -1;
int inserted = dm_store_message(dm, conv_id, dir, msg, len);
if (!dir && seq > c.last_in_seq) c.last_in_seq = seq; /* Только статистика; не курсор dedup. */
uint64_t ts;
memcpy(&ts, msg + 16, 8);
if (ts > c.last_ts) c.last_ts = ts;
if (inserted < 0 || dm_conv_save(dm, &c) != 0 || dm_media_track(inst, dir, msg, len) != 0 ||
(inserted && dm_history_show(inst, conv_id)) || dm_exec(dm, "COMMIT") != 0) {
dm_exec(dm, "ROLLBACK");
return -1;
}
receipt:
memcpy(receipt, message.mode == DM_MODE_GROUPS ? "DMACKG01" : "DMACK001", 8);
memcpy(receipt + 8, msg, 16);
memcpy(receipt + 24, &author, 8);
memcpy(receipt + 32, &inst->node_id, 8);
SHA256(msg, len, receipt + 40);
if (sc_ed25519_sign(inst->my_ed25519_privkey, receipt, 72, receipt + 72) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: receipt signing failed conv=%s seq=%llu", DM_ID, conv_id, (unsigned long long)seq);
return -1;
}
if (deleted) return 0;
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: committed receive conv=%s seq=%llu new=%d plain=%zu", DM_ID,
conv_id, (unsigned long long)seq, inserted, plain_len);
if (inserted) {
size_t cl = strlen(conv_id);
uint8_t evt[1 + 64 + 8];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
memcpy(evt + 1 + cl, &author, 8);
if (created) {
uint64_t generation = dm_history_generation(inst, conv_id);
memcpy(evt + 1 + cl, &generation, 8);
chat_event_post(inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl + 8);
memcpy(evt + 1 + cl, &author, 8);
}
chat_event_post(inst, CHAT_EVT_DM_MSG_RECEIVED, evt, 1 + (int)cl + 8);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: received conv=%s seq=%llu author=%llu", DM_ID, conv_id,
(unsigned long long)seq, (unsigned long long)author);
}
return 0;
}
/* Квитанцию может пересылать суперузел, но подписывает её только получатель. */
int dm_verify_receipt(struct UTUN_INSTANCE* inst, const uint8_t receipt[DM_RECEIPT_SIZE], const uint8_t* msg, size_t len) {
struct dm_state view = { .inst = inst, .db = inst ? inst->topo_sqlite_db : NULL };
uint64_t conv, seq, author, recipient;
if (!view.db || !receipt) goto invalid;
int grouped = !memcmp(receipt, "DMACKG01", 8);
if (!grouped && memcmp(receipt, "DMACK001", 8)) goto invalid;
memcpy(&conv, receipt + 8, 8);
memcpy(&seq, receipt + 16, 8);
memcpy(&author, receipt + 24, 8);
memcpy(&recipient, receipt + 32, 8);
if (!author || !recipient || author == recipient || !seq || seq > INT64_MAX || !conv || conv > INT64_MAX) goto invalid;
if (!grouped && conv != dm_derive_conv_id(author, recipient)) goto invalid;
uint8_t x[32], ed[32];
if (dm_node_pubkeys(&view, recipient, x, ed) != 0 ||
sc_ed25519_verify(ed, receipt, 72, receipt + 72) != SC_OK) goto invalid;
if (msg) {
uint8_t digest[32];
struct dm_message_view message;
if (dm_message_parse(msg, len, &message) || message.size != len || memcmp(receipt + 8, msg, 16) ||
memcmp(receipt + 24, msg + 24, 8)) goto invalid;
if (grouped != (message.mode == DM_MODE_GROUPS)) goto invalid;
if (message.mode == DM_MODE_GROUPS) {
uint8_t target_ed[32];
if (dm_message_target(&message, recipient, NULL, target_ed) < 0 || memcmp(target_ed, ed, 32)) goto invalid;
} else if (conv != dm_derive_conv_id(author, recipient)) goto invalid;
SHA256(msg, len, digest);
if (memcmp(digest, receipt + 40, 32)) goto invalid;
}
return 0;
invalid:
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: invalid delivery receipt bytes=%zu", DM_ID, len);
return -1;
}
/* Удалить исходящую задачу по точной подписанной квитанции; историю беседы оставить. */
int dm_accept_receipt(struct UTUN_INSTANCE* inst, const uint8_t receipt[DM_RECEIPT_SIZE]) {
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized || dm_verify_receipt(inst, receipt, NULL, 0) != 0) return -1;
uint64_t author, conv, seq, recipient;
memcpy(&author, receipt + 24, 8);
memcpy(&conv, receipt + 8, 8);
memcpy(&seq, receipt + 16, 8);
memcpy(&recipient, receipt + 32, 8);
if (author != inst->node_id) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: receipt for another sender=%llu", DM_ID, (unsigned long long)author);
return -1;
}
char conv_id[64];
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv);
sqlite3_stmt* st = NULL;
int transaction = 0;
int rc = sqlite3_prepare_v2(dm->db, "SELECT body FROM dm_outbox WHERE conv_id=? AND seq=?", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
rc = sqlite3_step(st);
if (rc == SQLITE_DONE) { sqlite3_finalize(st); return 0; }
if (rc != SQLITE_ROW) goto fail;
if (dm_verify_receipt(inst, receipt, sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0)) != 0) {
sqlite3_finalize(st);
return -1;
}
sqlite3_finalize(st);
st = NULL;
if (dm_exec(dm, "BEGIN IMMEDIATE")) return -1;
transaction = 1;
rc = sqlite3_prepare_v2(dm->db, "UPDATE dm_delivery SET receipt=? WHERE conv_id=? AND author=? AND seq=? AND recipient=?", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_blob(st, 1, receipt, DM_RECEIPT_SIZE, SQLITE_STATIC);
sqlite3_bind_text(st, 2, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 3, (sqlite3_int64)author); sqlite3_bind_int64(st, 4, (sqlite3_int64)seq);
sqlite3_bind_int64(st, 5, (sqlite3_int64)recipient);
rc = sqlite3_step(st);
if (rc != SQLITE_DONE || sqlite3_changes(dm->db) != 1) goto fail;
sqlite3_finalize(st); st = NULL;
rc = sqlite3_prepare_v2(dm->db, "DELETE FROM dm_outbox WHERE conv_id=? AND seq=? AND NOT EXISTS"
" (SELECT 1 FROM dm_delivery d WHERE d.conv_id=dm_outbox.conv_id AND d.seq=dm_outbox.seq AND d.author=? AND d.receipt IS NULL)",
-1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
sqlite3_bind_int64(st, 3, (sqlite3_int64)author);
rc = sqlite3_step(st);
if (rc != SQLITE_DONE) goto fail;
int complete = sqlite3_changes(dm->db);
sqlite3_finalize(st); st = NULL;
if (dm_exec(dm, "COMMIT")) goto fail;
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: delivery confirmed conv=%s seq=%llu recipient=%llu complete=%d", DM_ID, conv_id,
(unsigned long long)seq, (unsigned long long)recipient, complete);
dm_media_notify(inst, conv);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: receipt database error rc=%d sql=%s", DM_ID, rc, sqlite3_errmsg(dm->db));
sqlite3_finalize(st);
if (transaction) dm_exec(dm, "ROLLBACK");
return -1;
}
/* Ограниченная порция durable outbox; retry_at обеспечивает продвижение всей очереди. */
static void dm_pump(struct dm_state* dm) {
int64_t now = (int64_t)ntp_time_get_seconds(dm->inst);
for (int i = 0; i < 32; i++) {
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(dm->db,
"SELECT o.conv_id,o.seq,o.body,d.recipient,c.group_id,d.direct_until FROM dm_outbox o"
" JOIN dm_conversations c USING(conv_id) JOIN dm_delivery d ON d.conv_id=o.conv_id AND d.seq=o.seq"
" WHERE d.retry_at<=? AND d.receipt IS NULL AND c.closed=0 ORDER BY d.retry_at,o.seq,d.recipient LIMIT 1", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_int64(st, 1, now);
rc = sqlite3_step(st);
if (rc == SQLITE_DONE) { sqlite3_finalize(st); break; }
if (rc != SQLITE_ROW) goto fail;
char conv[64];
snprintf(conv, sizeof(conv), "%s", (const char*)sqlite3_column_text(st, 0));
uint64_t seq = (uint64_t)sqlite3_column_int64(st, 1);
size_t len = (size_t)sqlite3_column_bytes(st, 2);
uint8_t* body = u_malloc(len);
if (!body) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: outbox allocation failed bytes=%zu", DM_ID, len);
sqlite3_finalize(st);
break;
}
memcpy(body, sqlite3_column_blob(st, 2), len);
uint64_t peer = (uint64_t)sqlite3_column_int64(st, 3), gid = (uint64_t)sqlite3_column_int64(st, 4);
int64_t deadline = sqlite3_column_int64(st, 5);
if (!deadline) deadline = now + 5;
sqlite3_finalize(st);
st = NULL;
rc = sqlite3_prepare_v2(dm->db, "UPDATE dm_delivery SET retry_at=?,direct_until=?,"
" first_attempt=CASE WHEN first_attempt=0 THEN ? ELSE first_attempt END WHERE conv_id=? AND seq=? AND recipient=?",
-1, &st, NULL);
if (rc == SQLITE_OK) {
sqlite3_bind_int64(st, 1, now + 5);
sqlite3_bind_int64(st, 2, deadline); sqlite3_bind_int64(st, 3, now);
sqlite3_bind_text(st, 4, conv, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 5, (sqlite3_int64)seq); sqlite3_bind_int64(st, 6, (sqlite3_int64)peer);
rc = sqlite3_step(st);
}
sqlite3_finalize(st);
st = NULL;
if (rc != SQLITE_DONE) { u_free(body); goto fail; }
int direct = dm_route_group(dm->inst, peer, gid, &gid);
if (direct) {
if (!dm_route_send(dm, gid, peer, DM_SUBCMD_MSG, body, len))
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: outbox direct conv=%s seq=%llu group=%llu", DM_ID,
conv, (unsigned long long)seq, (unsigned long long)gid);
}
if (!direct || now >= deadline) {
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM delivery: mailbox fallback conv=%s seq=%llu recipient=%llu reason=%s", conv,
(unsigned long long)seq, (unsigned long long)peer, direct ? "direct ACK deadline" : "no route");
dm_mailbox_put(dm->inst, peer, dm->inst->node_id, body, len);
}
u_free(body);
continue;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: outbox scan failed rc=%d sql=%s", DM_ID, rc, sqlite3_errmsg(dm->db));
sqlite3_finalize(st);
break;
}
}
/* Durable-задачи переживают stop; таймер является только локальным триггером. */
static void dm_tick(void* arg) {
struct dm_state* dm = arg;
dm->timer = NULL;
dm_pump(dm);
dm->timer = uasync_set_timeout(dm->inst->ua, 10000, dm, dm_tick, "dm_outbox");
if (!dm->timer) DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: retry timer allocation failed", DM_ID);
}
/* Live-доставка использует тот же commit/receipt, что и доставка с суперузла. */
static void dm_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!entry || !conn || !entry->dgram || entry->len <= ROUTER_SVC_PAYLOAD_OFF) {
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
return; /* Уведомление router о закрытии канала. */
}
struct dm_state* dm = dm_of(conn->instance);
if (!dm || !dm->initialized) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: receive while stopped", DM_ID);
goto done;
}
const uint8_t* p = entry->dgram + ROUTER_SVC_PAYLOAD_OFF;
size_t len = entry->len - ROUTER_SVC_PAYLOAD_OFF - 1;
uint64_t source, gid;
memcpy(&source, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
memcpy(&gid, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
if (!(entry->dgram[ROUTER_SVC_FLAGS_OFF] & ROUTER_FLAG_SIGNED)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unsigned route packet source=%llu", DM_ID, (unsigned long long)source);
goto done;
}
if (p[0] == DM_SUBCMD_MSG) {
uint8_t receipt[DM_RECEIPT_SIZE];
if (dm_accept_message(dm->inst, p + 1, len, receipt) == 0) {
dm_route_send(dm, gid, source, DM_SUBCMD_ACK, receipt, sizeof(receipt));
dm_mailbox_receipt(dm->inst, receipt);
}
} else if (p[0] == DM_SUBCMD_ACK && len == DM_RECEIPT_SIZE) {
if (dm_accept_receipt(dm->inst, p + 1) == 0) dm_mailbox_receipt(dm->inst, p + 1);
} else {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: invalid command=%u bytes=%zu", DM_ID, p[0], len);
}
done:
queue_dgram_free(entry);
queue_entry_free(entry);
}
/* ── публичный API ── */
/* Открыть существующую беседу либо создать при подтверждённом отсутствии записи. */
int dm_start(struct UTUN_INSTANCE* inst, uint64_t peer_node_id,
const uint8_t peer_x25519[32], const uint8_t peer_ed25519[32],
const char* peer_name, const char* source_ch_id, enum dm_mode mode, uint64_t* result_conv) {
if (!inst || !peer_node_id || !peer_x25519 || !peer_ed25519) return -1;
if (peer_node_id == inst->node_id) return -1;
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized) return -1;
uint64_t conv_num = dm_derive_conv_id(inst->node_id, peer_node_id);
struct dm_party parties[2] = {{0}};
if (mode == DM_MODE_GROUPS) {
if (dm_party_resolve(inst, inst->node_id, mode, parties) || dm_party_resolve(inst, peer_node_id, mode, parties + 1)) return -1;
conv_num = dm_identity_conv_id(mode, parties);
if (!conv_num) return -1;
} else if (mode != DM_MODE_DEVICE) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM start: invalid mode=%d", mode); return -1;
}
char conv_id[64];
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv_num);
struct dm_conv c;
enum dm_conv_result loaded = dm_conv_load(dm, conv_id, &c);
if (loaded == DM_CONV_ERROR) return -1;
if (loaded == DM_CONV_FOUND) {
if (c.mode != mode || (mode == DM_MODE_DEVICE && c.peer_node_id != peer_node_id) ||
(mode == DM_MODE_GROUPS && (!dm_party_equal(c.parties, parties) || !dm_party_equal(c.parties + 1, parties + 1)))) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM start: conversation identity collision conv=%s", conv_id); return -1;
}
/* Беседа уже есть, но могла быть авто-создана с group_id=0 — дописываем из source_ch_id,
* чтобы маршрутизация шла через группу (а не только прямое соединение). */
if (c.group_id == 0 && source_ch_id && source_ch_id[0]) {
uint64_t gid = strtoull(source_ch_id, NULL, 10);
if (gid != 0) {
c.group_id = gid;
if (dm_conv_save(dm, &c) != 0) return -1;
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: conversation exists conv=%s — updated group=%llu",
DM_ID, conv_id, (unsigned long long)gid);
}
} else {
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: conversation already exists conv=%s", DM_ID, conv_id);
}
} else {
memset(&c, 0, sizeof(c));
snprintf(c.conv_id, sizeof(c.conv_id), "%s", conv_id);
c.mode = mode;
c.peer_node_id = peer_node_id;
memcpy(c.peer_x25519, peer_x25519, 32);
memcpy(c.peer_ed25519, peer_ed25519, 32);
dm_copy_peer_name(c.peer_name, peer_name, peer_name ? strlen(peer_name) : 0, conv_id);
c.group_id = source_ch_id ? strtoull(source_ch_id, NULL, 10) : 0;
if (mode == DM_MODE_GROUPS) {
memcpy(c.parties, parties, sizeof(parties));
memcpy(c.peer_x25519, parties[1].x, 32); memcpy(c.peer_ed25519, parties[1].ed, 32);
uint8_t private_key[32];
if (dm_party_private(inst, parties, private_key)) return -1;
int error = dm_derive_content_key(private_key, parties[1].x, c.content_key);
OPENSSL_cleanse(private_key, sizeof(private_key));
if (error) return -1;
} else if (dm_derive_content_key(inst->my_keys.private_key, c.peer_x25519, c.content_key) != 0) return -1;
if (dm_conv_save(dm, &c) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conv_save failed conv=%s", DM_ID, conv_id);
return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: started conv=%s peer=0x%016llx name=%s group=%llu",
DM_ID, conv_id, (unsigned long long)peer_node_id, c.peer_name,
(unsigned long long)c.group_id);
}
if (dm_history_show(inst, conv_id)) return -1;
/* событие GUI (всегда, чтобы GUI мог открыть/выделить беседу) */
{
size_t cl = strlen(conv_id);
uint8_t evt[1 + 64 + 8];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
uint64_t generation = dm_history_generation(inst, conv_id);
memcpy(evt + 1 + cl, &generation, 8);
chat_event_post(inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl + 8);
}
if (result_conv) *result_conv = conv_num;
return 0;
}
/* Снимок доставки сохраняется только источником, в транзакции отправки. */
static int dm_queue_targets(struct dm_state* dm, const struct dm_conv* c, const uint8_t* body, size_t length) {
struct dm_message_view view;
if (dm_message_parse(body, length, &view) || view.size != length || view.author != dm->inst->node_id) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM queue: invalid source/body conv=%s", c->conv_id); return -1;
}
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(dm->db, "INSERT INTO dm_delivery(conv_id,author,seq,recipient) VALUES(?,?,?,?)", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
unsigned count = view.mode == DM_MODE_GROUPS ? view.target_count : 1, queued = 0;
for (unsigned i = 0; i < count; i++) {
struct dm_target target = {.node = c->peer_node_id, .side = 1};
if (view.mode == DM_MODE_GROUPS && dm_message_target_at(&view, i, &target)) goto fail;
if (target.node == dm->inst->node_id) continue;
rc = sqlite3_reset(st);
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, c->conv_id, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)view.author);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)view.seq);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)target.node);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc != SQLITE_DONE) goto fail;
queued++;
}
rc = sqlite3_finalize(st); st = NULL;
if (rc != SQLITE_OK || !queued) goto fail;
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM queue: frozen delivery list conv=%s seq=%llu recipients=%u", c->conv_id,
(unsigned long long)view.seq, queued);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM queue: recipient insert failed conv=%s rc=%d sql=%s", c->conv_id, rc, sqlite3_errmsg(dm->db));
sqlite3_finalize(st); return -1;
}
/* Атомарно сохранить seq, историю и неизменяемое тело исходящей задачи. */
int dm_send(struct UTUN_INSTANCE* inst, const char* conv_id, const char* content_type,
const uint8_t* data, uint32_t data_len) {
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized || !conv_id || !content_type || (!data && data_len)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: send invalid arguments/state", DM_ID);
return -1;
}
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) != DM_CONV_FOUND || c.last_out_seq >= INT64_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: send conversation missing/seq exhausted conv=%s", DM_ID, conv_id);
return -1;
}
if (!dm_conv_active(dm, &c)) return -1;
if (!dm_history_current(inst, conv_id, dm_history_generation(inst, conv_id))) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM send: hidden conversation conv=%s", conv_id); return -1;
}
uint64_t seq;
if (dm_sequence_next(inst, &seq)) return -1;
uint64_t ts = (uint64_t)(ntp_time_get_us(inst) / 1000);
uint8_t* body = NULL;
size_t len = 0;
if (dm_build_body(dm, &c, seq, ts, content_type, data, data_len, &body, &len) != 0) return -1;
sqlite3_stmt* st = NULL;
if (dm_exec(dm, "BEGIN IMMEDIATE") != 0) { u_free(body); return -1; }
c.last_out_seq = seq;
if (ts > c.last_ts) c.last_ts = ts;
if (dm_sequence_take(inst, seq) || dm_store_message(dm, conv_id, 1, body, len) != 1 || dm_conv_save(dm, &c) != 0 ||
dm_media_track(inst, 1, body, len) != 0 || dm_queue_targets(dm, &c, body, len)) goto rollback;
int rc = sqlite3_prepare_v2(dm->db, "INSERT INTO dm_outbox(conv_id,seq,body) VALUES(?,?,?)", -1, &st, NULL);
if (rc == SQLITE_OK) {
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
sqlite3_bind_blob(st, 3, body, (int)len, SQLITE_STATIC);
rc = sqlite3_step(st);
}
sqlite3_finalize(st);
st = NULL;
if (rc != SQLITE_DONE) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: outbox insert failed conv=%s seq=%llu rc=%d sql=%s", DM_ID,
conv_id, (unsigned long long)seq, rc, sqlite3_errmsg(dm->db));
goto rollback;
}
if (dm_exec(dm, "COMMIT") != 0) goto rollback;
u_free(body);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: queued conv=%s seq=%llu peer=%llu bytes=%u", DM_ID, conv_id,
(unsigned long long)seq, (unsigned long long)c.peer_node_id, data_len);
dm_pump(dm);
size_t cl = strlen(conv_id);
uint8_t evt[1 + 64 + 8];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
memcpy(evt + 1 + cl, &inst->node_id, 8);
chat_event_post(inst, CHAT_EVT_DM_MSG_RECEIVED, evt, 1 + (int)cl + 8);
return 0;
rollback:
dm_exec(dm, "ROLLBACK");
u_free(body);
return -1;
}
/* Таблицы принадлежат DM, соединение с SQLite — общему ядру. */
int dm_core_init(struct UTUN_INSTANCE* inst) {
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: init without instance", DM_ID); return -1; }
if (inst->dm && inst->dm->initialized) return 0;
struct dm_state* dm = u_calloc(1, sizeof(*dm));
if (!dm) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: state allocation failed", DM_ID); return -1; }
inst->dm = dm;
dm->inst = inst;
dm->db = inst->topo_sqlite_db;
if (!dm->db || dm_schema_init(inst) != 0 || dm_history_init(inst)) goto fail;
if (etcp_router_bind(inst, ETCP_RT_ID_DM, dm_recv_cb) != 0) goto fail;
dm->initialized = 1;
dm->timer = uasync_set_timeout(inst->ua, 10000, dm, dm_tick, "dm_outbox");
if (!dm->timer) {
etcp_router_unbind(inst, ETCP_RT_ID_DM);
goto fail;
}
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: initialized persistent outbox", DM_ID);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: initialization failed db=%p", DM_ID, (void*)dm->db);
u_free(dm);
inst->dm = NULL;
return -1;
}
/* Снять триггеры перед освобождением контекста; pending остаётся в БД. */
void dm_core_destroy(struct UTUN_INSTANCE* inst) {
struct dm_state* dm = dm_of(inst);
if (!dm) return;
if (dm->invite) chat_invite_build_cancel(dm->invite->pending);
if (dm->timer) uasync_cancel_timeout(inst->ua, dm->timer);
dm->timer = NULL;
dm->initialized = 0;
etcp_router_unbind(inst, ETCP_RT_ID_DM);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: destroyed, durable outbox preserved", DM_ID);
u_free(dm);
inst->dm = NULL;
}
/* ── JSON для GUI/headless ── */
/* Экранирование строки для вставки в JSON-строку: " \ \n \r \t, прочие <0x20 — drop. */
static void json_escape(const char* src, char* dst, size_t dst_sz) {
size_t i = 0;
if (!src) src = "";
for (; *src && i + 2 < dst_sz; src++) {
unsigned char c = (unsigned char)*src;
switch (c) {
case '"': dst[i++] = '\\'; dst[i++] = '"'; break;
case '\\': dst[i++] = '\\'; dst[i++] = '\\'; break;
case '\n': dst[i++] = '\\'; dst[i++] = 'n'; break;
case '\r': dst[i++] = '\\'; dst[i++] = 'r'; break;
case '\t': dst[i++] = '\\'; dst[i++] = 't'; break;
default:
if (c < 0x20) break;
dst[i++] = (char)c;
}
}
dst[i] = '\0';
}
/* Список бесед в JSON. */
int dm_list_conversations_json(struct UTUN_INSTANCE* inst, char* buf, size_t buf_size, size_t* out_len) {
struct dm_state* dm = dm_of(inst);
if (!buf || !out_len || !dm || !dm->db) return -1;
*out_len = 0;
struct strbuf json = strbuf_new();
int failed = strbuf_addf(&json, "[") < 0;
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"SELECT c.conv_id,peer_node_id,COALESCE(peer_name,''),last_ts,last_out_seq,last_in_seq,COALESCE(h.generation,1)"
" FROM dm_conversations c LEFT JOIN dm_history h USING(conv_id) WHERE COALESCE(h.hidden,0)=0"
" ORDER BY last_ts DESC LIMIT 200",
-1, &st, NULL) == SQLITE_OK) {
int first = 1;
while (sqlite3_step(st) == SQLITE_ROW) {
const char* conv_id = (const char*)sqlite3_column_text(st, 0);
uint64_t peer = (uint64_t)sqlite3_column_int64(st, 1);
const char* name = (const char*)sqlite3_column_text(st, 2);
size_t name_len = (size_t)sqlite3_column_bytes(st, 2);
uint64_t last_ts = (uint64_t)sqlite3_column_int64(st, 3);
uint64_t out_seq = (uint64_t)sqlite3_column_int64(st, 4);
uint64_t in_seq = (uint64_t)sqlite3_column_int64(st, 5);
char resolved[128];
if (!name || !name_len) {
resolved[0] = '\0';
chat_core_get_node_name(inst, peer, resolved, sizeof(resolved));
name = resolved;
name_len = strlen(resolved);
}
char safe_name[DM_PEER_NAME_MAX + 1];
dm_copy_peer_name(safe_name, name, name_len, conv_id);
char esc_name[512];
json_escape(safe_name, esc_name, sizeof(esc_name));
if (strbuf_addf(&json,
"%s{\"conv_id\":\"%s\",\"peer_node_id\":%llu,\"peer_name\":\"%s\","
"\"last_ts\":%llu,\"out_seq\":%llu,\"in_seq\":%llu,\"generation\":%llu}",
first ? "" : ",", conv_id, (unsigned long long)peer,
esc_name, (unsigned long long)last_ts,
(unsigned long long)out_seq, (unsigned long long)in_seq,
(unsigned long long)sqlite3_column_int64(st, 6)) < 0) { failed = 1; break; }
first = 0;
}
if (sqlite3_errcode(dm->db) != SQLITE_DONE && sqlite3_errcode(dm->db) != SQLITE_OK) failed = 1;
} else failed = 1;
if (st) sqlite3_finalize(st);
if (strbuf_addf(&json, "]") < 0) failed = 1;
if (failed || json.len >= buf_size) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversations snapshot failed bytes=%zu capacity=%zu sql=%s", DM_ID,
json.len, buf_size, sqlite3_errmsg(dm->db));
strbuf_free(&json); return -1;
}
memcpy(buf, strbuf_str(&json), json.len + 1);
*out_len = json.len;
strbuf_free(&json);
return 0;
}
/* Сообщения беседы (расшифрованные) в JSON. */
int dm_list_messages_json(struct UTUN_INSTANCE* inst, const char* conv_id, int count, int offset,
char* buf, size_t buf_size, size_t* out_len) {
struct dm_state* dm = dm_of(inst);
if (!conv_id || !buf || !out_len || !dm || !dm->db || offset < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: invalid messages snapshot request", DM_ID);
return -1;
}
*out_len = 0;
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) != DM_CONV_FOUND) return -1;
/* Ограниченный снимок не блокирует uasync на длинной истории. */
if (count <= 0 || count > 200) count = 200;
const char* sql = "SELECT * FROM (SELECT m.dir,m.seq,m.ts,m.author,m.ct,m.data,o.seq IS NOT NULL AS pending,"
"COALESCE(f.ready,0) AS ready,COALESCE(f.downloading,0) AS downloading,f.receipt,"
"COALESCE(f.local_path,'') AS path,COALESCE(f.plain_size,0) AS size"
" FROM dm_messages m LEFT JOIN dm_outbox o ON m.dir=1 AND o.conv_id=m.conv_id AND o.seq=m.seq"
" LEFT JOIN dm_files f ON f.conv_id=m.conv_id AND f.dir=m.dir AND f.seq=m.seq"
" WHERE m.conv_id=? ORDER BY m.ts DESC,m.seq DESC,m.dir DESC LIMIT ? OFFSET ?) ORDER BY ts,seq,dir";
struct strbuf sb = strbuf_new();
uint64_t conv_num = (uint64_t)strtoull(conv_id, NULL, 10);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db, sql, -1, &st, NULL) != SQLITE_OK || strbuf_addf(&sb, "[") < 0) goto fail;
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int(st, 2, count); sqlite3_bind_int(st, 3, offset);
int first = 1, rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
int dir = sqlite3_column_int(st, 0);
uint64_t seq = (uint64_t)sqlite3_column_int64(st, 1), ts = (uint64_t)sqlite3_column_int64(st, 2);
uint64_t author = (uint64_t)sqlite3_column_int64(st, 3);
const char* ct = (const char*)sqlite3_column_text(st, 4);
const uint8_t* enc = sqlite3_column_blob(st, 5);
int enc_len = sqlite3_column_bytes(st, 5);
char text[DM_DATA_MAX + 1];
size_t plen = 0;
uint8_t nonce[DM_NONCE_SIZE];
dm_build_nonce(conv_num, author, seq, nonce);
if (enc_len < DM_TAG_SIZE || enc_len > DM_DATA_MAX + DM_TAG_SIZE ||
media_ccm_decrypt(c.content_key, nonce, enc, (size_t)enc_len, (uint8_t*)text, &plen)) goto fail;
text[plen] = '\0';
const char* group_name = "";
int invite = ct && !strcmp(ct, "invite");
if (invite) {
char* separator = memchr(text, '\0', plen);
if (!separator || (size_t)(separator - text) + 8 > plen ||
strncmp(separator + 1, "utun://", 7) || memchr(separator + 1, '\0', plen - (separator + 1 - text))) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: malformed payload conv=%s seq=%llu bytes=%zu", conv_id,
(unsigned long long)seq, plen);
goto fail;
}
group_name = text;
}
struct attachment_info attachment = {0};
int file = ct && !strcmp(ct, "file");
if (file) {
if (attachment_decode((const uint8_t*)text, plen, &attachment)) goto fail;
snprintf(text, sizeof(text), "%s", attachment.name);
}
int ready = sqlite3_column_int(st, 7), downloading = sqlite3_column_int(st, 8);
int delivered = sqlite3_column_bytes(st, 9) == DM_MEDIA_RECEIPT_SIZE;
const char* path = ready == 1 ? (const char*)sqlite3_column_text(st, 10) : "";
const char* status = delivered && dir ? "delivered" : ready == 1 ? "ready" : downloading ? "downloading" :
ready < 0 ? "expired" : "pending";
char esc_ct[512], esc_data[DM_DATA_MAX * 2 + 8], esc_group[DM_DATA_MAX * 2 + 8], esc_path[2048];
json_escape(group_name, esc_group, sizeof(esc_group));
const char* display_data = invite ? text + strlen(group_name) + 1 : text;
json_escape(ct ? ct : "", esc_ct, sizeof(esc_ct)); json_escape(display_data, esc_data, sizeof(esc_data));
json_escape(path, esc_path, sizeof(esc_path));
if (strbuf_addf(&sb, "%s{\"dir\":%d,\"seq\":%llu,\"ts\":%llu,\"author\":\"%llu\",\"ct\":\"%s\",\"data\":\"%s\","
"\"pending\":%s,\"media_status\":\"%s\",\"media_path\":\"%s\",\"media_size\":%llu,"
"\"tags\":{\"group_name\":\"%s\"},"
"\"content_type\":\"%s\",\"duration_ms\":%u,\"width\":%u,\"height\":%u,\"waveform\":\"%s\"}", first ? "" : ",", dir,
(unsigned long long)seq, (unsigned long long)ts, (unsigned long long)author, esc_ct, esc_data,
sqlite3_column_int(st, 6) ? "true" : "false", status, esc_path,
(unsigned long long)sqlite3_column_int64(st, 11), esc_group,
file ? attachment_content_type(attachment.kind) : "text/plain",
attachment.duration_ms, attachment.width, attachment.height, attachment.waveform) < 0) goto fail;
first = 0;
}
if (rc != SQLITE_DONE || strbuf_addf(&sb, "]") < 0 || sb.len + 1 > buf_size) goto fail;
memcpy(buf, strbuf_str(&sb), sb.len + 1); *out_len = sb.len;
sqlite3_finalize(st); strbuf_free(&sb);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: message snapshot failed conv=%s buffer=%zu required=%zu sql=%s", DM_ID,
conv_id, buf_size, sb.len + 1, sqlite3_errmsg(dm->db));
sqlite3_finalize(st); strbuf_free(&sb);
return -1;
}
/* ── трамплины для gui_bridge (GUI → uasync) ── */
/* Начать беседу с узлом из общей группы: находим pubkey пира в nodes, зовём dm_start. */
void dm_start_trampoline(void* arg) {
struct dm_start_req* req = (struct dm_start_req*)arg;
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
enum dm_mode mode = req->mode;
uint64_t peer = req->peer_node_id;
char src[64]; strncpy(src, req->source_ch_id, sizeof(src) - 1); src[sizeof(src) - 1] = '\0';
char name[128]; strncpy(name, req->peer_name, sizeof(name) - 1); name[sizeof(name) - 1] = '\0';
u_free(arg);
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized || !inst || !peer) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_start_trampoline — not initialized", DM_ID);
return;
}
if (peer == inst->node_id) return;
uint8_t x25519[32], ed25519[32];
if (dm_node_pubkeys(dm, peer, x25519, ed25519) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_start_trampoline — no pubkeys for peer=0x%016llx",
DM_ID, (unsigned long long)peer);
return;
}
if (dm_start(inst, peer, x25519, ed25519, name[0] ? name : NULL, src[0] ? src : NULL, mode, NULL) != 0)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_start_trampoline — dm_start failed peer=0x%016llx",
DM_ID, (unsigned long long)peer);
}
/* Отправить текстовое сообщение в беседу. */
void dm_send_trampoline(void* arg) {
struct dm_send_req* req = (struct dm_send_req*)arg;
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
char conv[64]; strncpy(conv, req->conv_id, sizeof(conv) - 1); conv[sizeof(conv) - 1] = '\0';
uint64_t generation = req->generation ? req->generation : dm_history_generation(inst, conv);
uint64_t request_id = req->request_id;
int rc = dm_history_current(inst, conv, generation) ? dm_send(inst, conv, "text", req->data, req->data_len) : -1;
chat_event_submitted(inst, request_id, rc == 0);
u_free(arg);
if (rc != 0)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_send_trampoline — dm_send failed conv=%s",
DM_ID, conv);
}
/* Завершить операцию и вернуть результат с неизменяемым адресатом. */
static void dm_invite_done(struct dm_invite_req* req, int result) {
struct dm_state* dm = dm_of(req->inst);
if (dm && dm->invite == req) dm->invite = NULL;
uint64_t conv = dm_derive_conv_id(req->inst->node_id, req->peer_node_id);
int32_t status = result;
uint8_t event[36];
memcpy(event, &req->request_id, 8); memcpy(event + 8, &status, 4);
memcpy(event + 12, &req->peer_node_id, 8); memcpy(event + 20, &conv, 8);
char id[64]; snprintf(id, sizeof(id), "%llu", (unsigned long long)conv);
uint64_t generation = dm_history_generation(req->inst, id); memcpy(event + 28, &generation, 8);
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: finished request=%llu peer=%llu ch=%llu result=%d",
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id,
(unsigned long long)req->channel_id, result);
chat_event_post(req->inst, CHAT_EVT_DM_INVITE_RESULT, event, sizeof(event));
u_free(req);
}
/* ACK регистрации разрешает отправку ссылки; ошибка/отмена не создаёт сообщение. */
static void dm_invite_ready(void* arg, int result, const char* link) {
struct dm_invite_req* req = arg;
req->pending = NULL;
if (result == CHAT_JOIN_CANCELLED) { dm_invite_done(req, DM_INVITE_CANCELLED); return; }
if (result != CHAT_JOIN_OK || !link || !link[0]) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: registration failed request=%llu ch=%llu result=%d",
(unsigned long long)req->request_id, (unsigned long long)req->channel_id, result);
dm_invite_done(req, DM_INVITE_BUILD_ERROR); return;
}
char conv[64];
snprintf(conv, sizeof(conv), "%llu", (unsigned long long)dm_derive_conv_id(req->inst->node_id, req->peer_node_id));
if (!dm_history_current(req->inst, conv, req->generation)) { dm_invite_done(req, DM_INVITE_CANCELLED); return; }
size_t bytes = strlen(link);
sqlite3_stmt* st = NULL;
char channel[64], payload[DM_DATA_MAX];
snprintf(channel, sizeof(channel), "%llu", (unsigned long long)req->channel_id);
struct dm_state* dm = dm_of(req->inst);
int rc = sqlite3_prepare_v2(dm->db, "SELECT name FROM channels WHERE channel_id=?", -1, &st, NULL);
if (rc == SQLITE_OK) { sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); rc = sqlite3_step(st); }
size_t name_bytes = rc == SQLITE_ROW ? (size_t)sqlite3_column_bytes(st, 0) : 0;
if (rc != SQLITE_ROW || name_bytes + 1 + bytes > sizeof(payload)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: group metadata failed ch=%s rc=%d name_bytes=%zu link_bytes=%zu sql=%s",
channel, rc, name_bytes, bytes, sqlite3_errmsg(dm->db));
sqlite3_finalize(st); dm_invite_done(req, DM_INVITE_SEND_ERROR); return;
}
memcpy(payload, sqlite3_column_text(st, 0), name_bytes); payload[name_bytes] = '\0';
memcpy(payload + name_bytes + 1, link, bytes);
sqlite3_finalize(st);
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: registered request=%llu peer=%llu ch=%llu name_bytes=%zu link_bytes=%zu ttl=%ds",
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id,
(unsigned long long)req->channel_id, name_bytes, bytes, JOIN_KEY_TTL_SECONDS);
rc = dm_send(req->inst, conv, "invite", (const uint8_t*)payload, (uint32_t)(name_bytes + 1 + bytes));
if (rc) DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: send failed request=%llu conv=%s rc=%d",
(unsigned long long)req->request_id, conv, rc);
dm_invite_done(req, rc ? DM_INVITE_SEND_ERROR : DM_INVITE_OK);
}
/* Создать/найти PM, затем зарегистрировать свежий invite; все данные принадлежат запросу. */
void dm_invite_trampoline(void* arg) {
struct dm_invite_req* req = arg;
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: missing request"); return; }
struct dm_state* dm = dm_of(req->inst);
if (req->cancel) {
if (dm && dm->invite && dm->invite->request_id == req->request_id) {
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: cancelling request=%llu", (unsigned long long)req->request_id);
chat_invite_build_cancel(dm->invite->pending);
} else DEBUG_DEBUG(DEBUG_CATEGORY_DM, "invite PM: cancel already completed request=%llu",
(unsigned long long)req->request_id);
u_free(req); return;
}
if (!req->inst) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: missing instance"); u_free(req); return; }
if (!dm || !dm->initialized || !req->request_id || !req->peer_node_id || req->peer_node_id == req->inst->node_id) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: invalid state/recipient request=%llu peer=%llu",
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id);
dm_invite_done(req, DM_INVITE_START_ERROR); return;
}
if (dm->invite) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "invite PM: busy request=%llu pending=%llu",
(unsigned long long)req->request_id, (unsigned long long)dm->invite->request_id);
dm_invite_done(req, DM_INVITE_BUSY); return;
}
char channel[64], conv[64];
snprintf(channel, sizeof(channel), "%llu", (unsigned long long)req->channel_id);
struct TOPO_GROUP* group = topo_groups_find(req->inst->topo_groups, req->channel_id);
if (!group || group->group_type != TOPO_GROUP_TYPE_CHAT ||
!topo_node_sqlite_member_in_channel(dm->db, channel, req->inst->node_id)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: no active membership ch=%s", channel);
dm_invite_done(req, DM_INVITE_BUILD_ERROR); return;
}
snprintf(conv, sizeof(conv), "%llu", (unsigned long long)dm_derive_conv_id(req->inst->node_id, req->peer_node_id));
struct dm_conv c;
enum dm_conv_result loaded = dm_conv_load(dm, conv, &c);
if (loaded == DM_CONV_ERROR) { dm_invite_done(req, DM_INVITE_START_ERROR); return; }
if (loaded == DM_CONV_NOT_FOUND) {
uint8_t x25519[32], ed25519[32];
if (dm_node_pubkeys(dm, req->peer_node_id, x25519, ed25519) != 0 ||
dm_start(req->inst, req->peer_node_id, x25519, ed25519, req->peer_name, req->source_ch_id, DM_MODE_DEVICE, NULL) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: conversation failed peer=%llu", (unsigned long long)req->peer_node_id);
dm_invite_done(req, DM_INVITE_START_ERROR); return;
}
}
if (dm_history_show(req->inst, conv)) { dm_invite_done(req, DM_INVITE_START_ERROR); return; }
req->generation = dm_history_generation(req->inst, conv);
if (!dm_history_current(req->inst, conv, req->generation)) { dm_invite_done(req, DM_INVITE_CANCELLED); return; }
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: building request=%llu peer=%llu source=%s ch=%s conv=%s",
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id, req->source_ch_id, channel, conv);
dm->invite = req;
req->pending = chat_invite_build_link(req->inst, req->channel_id, 0, NULL, dm_invite_ready, req);
if (!req->pending) dm_invite_ready(req, CHAT_JOIN_ERROR, NULL);
}
void dm_cancel_pending_history(struct UTUN_INSTANCE* inst, const char* conv) {
struct dm_state* dm = dm_of(inst);
if (dm && dm->invite && dm_derive_conv_id(inst->node_id, dm->invite->peer_node_id) == strtoull(conv, NULL, 10))
chat_invite_build_cancel(dm->invite->pending);
}
void dm_delete_trampoline(void* arg) {
struct dm_delete_req* req = arg;
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM delete: missing request"); return; }
int success = dm_history_delete(req->inst, req->conv_id, req->generation) == 0;
uint64_t conv = strtoull(req->conv_id, NULL, 10);
uint8_t event[25];
memcpy(event, &req->request_id, 8); memcpy(event + 8, &conv, 8); memcpy(event + 16, &req->generation, 8); event[24] = success;
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM delete result: request=%llu conv=%s generation=%llu success=%d",
(unsigned long long)req->request_id, req->conv_id, (unsigned long long)req->generation, success);
chat_event_post(req->inst, CHAT_EVT_DM_DELETE_RESULT, event, sizeof(event));
u_free(req);
}
/* Единая BGP-проверка для кнопки PM и запуска звонка; только uasync-поток. */
int dm_call_route(struct UTUN_INSTANCE* inst, const char* conv_id, uint64_t* peer, uint64_t* group_id) {
struct dm_state* dm = dm_of(inst);
if (!peer || !group_id || !conv_id || !dm || !dm->db) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "PM route: invalid request or stopped service"); return 0;
}
*peer = 0; *group_id = 0;
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db, "SELECT peer_node_id,group_id FROM dm_conversations WHERE conv_id=?",
-1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "PM route: query failed: %s", sqlite3_errmsg(dm->db)); return 0;
}
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
int rc = sqlite3_step(st);
uint64_t preferred = 0;
if (rc == SQLITE_ROW) {
*peer = (uint64_t)sqlite3_column_int64(st, 0);
preferred = (uint64_t)sqlite3_column_int64(st, 1);
} else if (rc == SQLITE_DONE) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "PM route: conversation missing conv=%s", conv_id);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "PM route: read failed conv=%s: %s", conv_id, sqlite3_errmsg(dm->db));
}
sqlite3_finalize(st);
int available = *peer && *peer != inst->node_id && dm_find_route_group(inst, *peer, preferred, group_id, 1);
DEBUG_TRACE(DEBUG_CATEGORY_CALL, "PM route: conv=%s peer=%llu preferred=%llu group=%llu available=%d",
conv_id, (unsigned long long)*peer, (unsigned long long)preferred, (unsigned long long)*group_id, available);
return available;
}
void dm_call_route_trampoline(void* arg) {
struct dm_messages_req* req = arg;
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "PM route: empty request"); return; }
uint64_t peer = 0, group = 0;
dm_call_route(req->inst, req->conv_id, &peer, &group);
size_t len = strlen(req->conv_id);
uint8_t evt[1 + 63 + 16];
evt[0] = (uint8_t)len;
memcpy(evt + 1, req->conv_id, len);
memcpy(evt + 1 + len, &peer, 8);
memcpy(evt + 9 + len, &group, 8);
chat_event_post(req->inst, CHAT_EVT_DM_CALL_ROUTE, evt, (int)(17 + len));
u_free(req);
}
/* Ограниченный JSON-снимок доставляется GUI асинхронно, без доступа UI к instance/SQLite. */
void dm_messages_trampoline(void* arg) {
struct dm_messages_req* req = arg;
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: missing history request", DM_ID); return; }
struct UTUN_INSTANCE* inst = req->inst;
char conv[64];
snprintf(conv, sizeof(conv), "%s", req->conv_id);
uint64_t generation = req->generation;
u_free(req);
if (!dm_of(inst)) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: snapshot after stop", DM_ID); return; }
if (!generation) generation = dm_history_generation(inst, conv);
if (!dm_history_current(inst, conv, generation)) {
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: skipped stale history conv=%s generation=%llu", DM_ID, conv,
(unsigned long long)generation); return;
}
size_t cl = strlen(conv), cap = 1024 * 1024, len = 0;
uint8_t* evt = u_malloc(9 + cl + cap);
if (!evt) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: snapshot allocation failed", DM_ID); return; }
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv, cl);
memcpy(evt + 1 + cl, &generation, 8);
if (!dm_list_messages_json(inst, conv, 200, 0, (char*)evt + 9 + cl, cap, &len)) {
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: history snapshot conv=%s generation=%llu bytes=%zu", DM_ID, conv,
(unsigned long long)generation, len);
chat_event_post(inst, CHAT_EVT_DM_MESSAGES, evt, (int)(9 + cl + len));
}
u_free(evt);
}
void dm_conversations_trampoline(void* arg) {
struct dm_conversations_req* req = arg;
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
u_free(req);
if (!dm_of(inst)) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: conversations snapshot after stop", DM_ID); return; }
size_t cap = 256 * 1024, len = 0;
char* json = u_malloc(cap);
if (!json) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversations allocation failed", DM_ID); return; }
if (!dm_list_conversations_json(inst, json, cap, &len))
chat_event_post(inst, CHAT_EVT_DM_CONVERSATIONS, (const uint8_t*)json, (int)len);
u_free(json);
}
Reference in new issue View Git Blame Copy Permalink