From 760398b96cacea70a1f4956196ccfad57db6530e Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 01:02:55 +0100 Subject: [PATCH 001/164] Commit incoming attachment requests before acknowledging them --- src/chat/chat_attachment.c | 139 ++++++++++++++++++++++++++++--------- src/chat/chat_attachment.h | 7 +- 2 files changed, 112 insertions(+), 34 deletions(-) diff --git a/src/chat/chat_attachment.c b/src/chat/chat_attachment.c index b3b3d856c..b62058980 100644 --- a/src/chat/chat_attachment.c +++ b/src/chat/chat_attachment.c @@ -341,19 +341,35 @@ done: /* Кодограммы привязаны к immutable сообщению, UUID и полному hash контейнера. */ static int ak_body(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32], uint8_t cmd, uint8_t wire[AK_WIRE]) { - sqlite3_stmt* st = ak_prepare(inst, "SELECT crypto_id,cipher_hash,key FROM attachment_keys WHERE channel=? AND message_id=?"); - if (!st) return -1; - sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); sqlite3_bind_blob(st, 2, message, 32, SQLITE_STATIC); - int rc = sqlite3_step(st), ok = rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 16 && sqlite3_column_bytes(st, 1) == 32 && - (cmd != AK_GRANT || sqlite3_column_bytes(st, 2) == 32); - if (ok) { - memcpy(wire, "AK01", 4); wire[4] = cmd; memcpy(wire + 5, message, 32); - memcpy(wire + 37, sqlite3_column_blob(st, 0), 16); memcpy(wire + 53, sqlite3_column_blob(st, 1), 32); - if (cmd == AK_GRANT) memcpy(wire + AK_BODY, sqlite3_column_blob(st, 2), 32); + sqlite3* db = CC(inst)->db; sqlite3_stmt* st = NULL; + const char* stage = "prepare"; + int rc = sqlite3_prepare_v2(db, "SELECT crypto_id,cipher_hash,key FROM attachment_keys WHERE channel=? AND message_id=?", -1, &st, NULL); + if (rc != SQLITE_OK) goto fail; + stage = "bind channel"; rc = sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "bind message"; rc = sqlite3_bind_blob(st, 2, message, 32, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "step"; rc = sqlite3_step(st); + if (rc != SQLITE_ROW) goto fail; + if (sqlite3_column_bytes(st, 0) != 16 || sqlite3_column_bytes(st, 1) != 32 || + (cmd == AK_GRANT && sqlite3_column_bytes(st, 2) != 32)) { + stage = "identity/key size"; rc = SQLITE_MISMATCH; goto fail; } - sqlite3_finalize(st); - if (!ok) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: wire identity/key unavailable channel=%s command=%u rc=%d", channel, cmd, rc); - return ok ? (cmd == AK_GRANT ? AK_WIRE : AK_BODY) : -1; + const void* id = sqlite3_column_blob(st, 0); const void* hash = sqlite3_column_blob(st, 1); + const void* key = cmd == AK_GRANT ? sqlite3_column_blob(st, 2) : NULL; + if (!id || !hash || (cmd == AK_GRANT && !key)) { stage = "identity/key allocation"; rc = SQLITE_NOMEM; goto fail; } + memcpy(wire, "AK01", 4); wire[4] = cmd; memcpy(wire + 5, message, 32); + memcpy(wire + 37, id, 16); memcpy(wire + 53, hash, 32); + if (cmd == AK_GRANT) memcpy(wire + AK_BODY, key, 32); + stage = "finalize"; rc = sqlite3_finalize(st); st = NULL; + if (rc != SQLITE_OK) goto fail; + return cmd == AK_GRANT ? AK_WIRE : AK_BODY; +fail: + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: wire identity/key channel=%s command=%u stage=%s rc=%d (%s) db=%s", + channel, cmd, stage, rc, sqlite3_errstr(rc), sqlite3_errmsg(db)); + int final_rc = sqlite3_finalize(st); + if (final_rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: wire cleanup finalize rc=%d", final_rc); + return -1; } /* Неподписанные и незашифрованные ответы сервис не принимает. */ @@ -440,6 +456,82 @@ static void ak_ack(struct UTUN_INSTANCE* inst, const char* channel, uint64_t pee if (rc < 0) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: acknowledgment deferred channel=%s command=%u", channel, cmd); } +/* Новый запрос и повтор прежнего решения фиксируются до ACK; внешняя транзакция остаётся вызывающему. */ +static int ak_accept_request(struct UTUN_INSTANCE* inst, const char* channel, uint64_t peer, const uint8_t* body) { + sqlite3* db = CC(inst)->db; + if (!sqlite3_get_autocommit(db)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: request rejected inside external transaction channel=%s peer=%llu", + channel, (unsigned long long)peer); return -1; + } + sqlite3_stmt* st = NULL; uint8_t expected[AK_WIRE]; + const char* stage = "begin"; + int rc = sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + stage = "identity"; + if (ak_body(inst, channel, body + 5, AK_REQUEST, expected) < 0 || memcmp(body + 5, expected + 5, AK_BODY - 5)) { + rc = SQLITE_MISMATCH; goto fail; + } + stage = "authorization"; + if (!ak_authorized(inst, channel, body + 5, peer)) { rc = SQLITE_AUTH; goto fail; } + stage = "read prepare"; + rc = sqlite3_prepare_v2(db, "SELECT incoming,state FROM attachment_key_requests WHERE channel=? AND message_id=? AND peer=?", -1, &st, NULL); + if (rc != SQLITE_OK) goto fail; + stage = "read bind channel"; rc = sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "read bind message"; rc = sqlite3_bind_blob(st, 2, body + 5, 32, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "read bind peer"; rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)peer); + if (rc != SQLITE_OK) goto fail; + stage = "read step"; rc = sqlite3_step(st); + if (rc != SQLITE_ROW && rc != SQLITE_DONE) goto fail; + int exists = rc == SQLITE_ROW, incoming = exists ? sqlite3_column_int(st, 0) : 1; + int state = exists ? sqlite3_column_int(st, 1) : 0; + stage = "read finalize"; rc = sqlite3_finalize(st); st = NULL; + if (rc != SQLITE_OK) goto fail; + if (incoming != 1 || (state != 0 && state != 2 && state != 3 && state != 4)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: request conflict channel=%s peer=%llu incoming=%d state=%d", + channel, (unsigned long long)peer, incoming, state); + stage = "request state"; rc = SQLITE_CONSTRAINT; goto fail; + } + if (!exists) { + stage = "insert prepare"; + rc = sqlite3_prepare_v2(db, "INSERT INTO attachment_key_requests(channel,message_id,peer,incoming,state,created_at)" + " VALUES(?,?,?,1,0,unixepoch())", -1, &st, NULL); + if (rc != SQLITE_OK) goto fail; + stage = "insert bind channel"; rc = sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "insert bind message"; rc = sqlite3_bind_blob(st, 2, body + 5, 32, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "insert bind peer"; rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)peer); + if (rc != SQLITE_OK) goto fail; + stage = "insert step"; rc = sqlite3_step(st); + if (rc != SQLITE_DONE) goto fail; + int changed = sqlite3_changes(db); + stage = "insert finalize"; rc = sqlite3_finalize(st); st = NULL; + if (rc != SQLITE_OK) goto fail; + if (changed != 1) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: request insert rows=%d expected=1 channel=%s peer=%llu", + changed, channel, (unsigned long long)peer); + stage = "insert row count"; rc = SQLITE_CONSTRAINT; goto fail; + } + } + int command = state == 0 ? 0 : state == 3 ? AK_DENY : AK_GRANT; + stage = "outbox"; + if (command && ak_queue(inst, channel, body + 5, peer, command)) { rc = SQLITE_ERROR; goto fail; } + stage = "commit"; rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: request committed channel=%s peer=%llu message=%02x%02x%02x%02x duplicate=%d state=%d queued=%d", + channel, (unsigned long long)peer, body[5], body[6], body[7], body[8], exists, state, command); + return 0; +fail: + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: request channel=%s peer=%llu stage=%s rc=%d (%s) db=%s", + channel, (unsigned long long)peer, stage, rc, sqlite3_errstr(rc), sqlite3_errmsg(db)); + int final_rc = sqlite3_finalize(st); + if (final_rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: request cleanup finalize rc=%d", final_rc); + ak_rollback(inst, "request", channel); + return -1; +} + /* Ответ меняет ключ, запрос и очередь одним commit; сетевое подтверждение остаётся вызывающему. */ static int ak_apply_result(struct UTUN_INSTANCE* inst, const char* channel, uint64_t peer, const uint8_t* body, int* next_out) { sqlite3* db = CC(inst)->db; @@ -569,33 +661,18 @@ static void ak_receive(struct ETCP_CONN* conn, struct ll_entry* entry) { char channel[32]; if (!CC(inst) || CC(inst)->attachment_closing || entry->len < ROUTER_SVC_PAYLOAD_OFF + AK_BODY) goto invalid; memcpy(&peer, d + ROUTER_SVC_SRC_OFF, 8); memcpy(&group, d + ROUTER_SVC_GROUP_OFF, 8); memcpy(&target, d + ROUTER_SVC_DST_OFF, 8); - if (target != inst->node_id || peer == inst->node_id || + if (target != inst->node_id || !peer || peer == inst->node_id || (d[ROUTER_SVC_FLAGS_OFF] & (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) != (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) goto invalid; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group); uint8_t* body = d + ROUTER_SVC_PAYLOAD_OFF; size_t bytes = entry->len - ROUTER_SVC_PAYLOAD_OFF; uint8_t cmd = body[4], expected[AK_WIRE]; if (!ak_channel(channel) || memcmp(body, "AK01", 4) || cmd < AK_REQUEST || cmd > AK_RESULT_ACK || - bytes != (cmd == AK_GRANT ? AK_WIRE : AK_BODY) || ak_body(inst, channel, body + 5, AK_REQUEST, expected) < 0 || - memcmp(body + 5, expected + 5, AK_BODY - 5)) goto invalid; + bytes != (cmd == AK_GRANT ? AK_WIRE : AK_BODY)) goto invalid; if (cmd == AK_REQUEST) { - if (!ak_authorized(inst, channel, body + 5, peer) || ak_exec(inst, "SAVEPOINT attachment_request")) goto invalid; - sqlite3_stmt* st = ak_prepare(inst, "INSERT INTO attachment_key_requests(channel,message_id,peer,incoming,state,created_at)" - " VALUES(?,?,?,1,0,unixepoch()) ON CONFLICT(channel,message_id,peer) DO NOTHING"); - if (st) { sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); sqlite3_bind_blob(st, 2, body + 5, 32, SQLITE_STATIC); - sqlite3_bind_int64(st, 3, (sqlite3_int64)peer); } - if (!st || ak_step(inst, st)) goto rollback_request; - st = ak_prepare(inst, "SELECT state FROM attachment_key_requests WHERE channel=? AND message_id=? AND peer=? AND incoming=1"); - if (!st) goto rollback_request; - sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); sqlite3_bind_blob(st, 2, body + 5, 32, SQLITE_STATIC); - sqlite3_bind_int64(st, 3, (sqlite3_int64)peer); - int rc = sqlite3_step(st), state = rc == SQLITE_ROW ? sqlite3_column_int(st, 0) : -1; sqlite3_finalize(st); - if (state < 0 || (state && ak_queue(inst, channel, body + 5, peer, state == 3 ? AK_DENY : AK_GRANT)) || - ak_exec(inst, "RELEASE attachment_request")) goto rollback_request; - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: request saved channel=%s peer=%llu state=%d", channel, (unsigned long long)peer, state); + if (ak_accept_request(inst, channel, peer, body)) goto done; ak_ack(inst, channel, peer, body, AK_REQUEST_ACK); ak_changed(inst, channel); goto done; -rollback_request: - ak_exec(inst, "ROLLBACK TO attachment_request"); ak_exec(inst, "RELEASE attachment_request"); goto done; } + if (ak_body(inst, channel, body + 5, AK_REQUEST, expected) < 0 || memcmp(body + 5, expected + 5, AK_BODY - 5)) goto invalid; int next; if (ak_apply_result(inst, channel, peer, body, &next)) goto done; DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: control committed channel=%s peer=%llu command=%u state=%d", channel, diff --git a/src/chat/chat_attachment.h b/src/chat/chat_attachment.h index e018ba225..6d462c2bc 100644 --- a/src/chat/chat_attachment.h +++ b/src/chat/chat_attachment.h @@ -6,9 +6,10 @@ * Ответы REQUEST_ACK/GRANT/DENY/RESULT_ACK атомарно меняют ключ, запрос и outbox собственной транзакцией. * Внешняя транзакция для ответов запрещена. GRANT/DENY подтверждаются только после проверенного commit; * повтор с тем же результатом допустим, конфликтующий ключ/решение отклоняется. Отказ SQL сохраняет прежнее состояние. - * Незавершённый rollback отключает модуль и запрашивает остановку ядра владельцем вне callback. - * Известное ограничение приёма REQUEST: вложенный SAVEPOINT допускает ACK/событие до commit внешней транзакции; - * её rollback теряет уже подтверждённый запрос. См. doc/chat_audit_findings.txt, «Аудит приёма REQUEST». */ + * Приём REQUEST владеет собственной транзакцией проверки идентичности/прав и сохранения запроса/повтора ответа. + * Внешняя транзакция отклоняется без изменений; ACK/событие — после commit. Дубликат сохраняет решение и дату: + * state=0 ждёт автора, 2/4 повторяют GRANT, 3 повторяет DENY; чужое направление и другие состояния отклоняются. + * Незавершённый rollback отключает модуль и запрашивает остановку ядра владельцем вне callback. */ #ifndef CHAT_ATTACHMENT_H #define CHAT_ATTACHMENT_H #include "../media_async/attachment.h" From 98a6d3cce088da3c8fb608a4168979470b4298ab Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 01:05:49 +0100 Subject: [PATCH 002/164] Test committed REQUEST acknowledgments, duplicates and SQL failures --- tests/Makefile.am | 7 + tests/test_attachment_requests.c | 384 +++++++++++++++++++++++++++++++ 2 files changed, 391 insertions(+) create mode 100644 tests/test_attachment_requests.c diff --git a/tests/Makefile.am b/tests/Makefile.am index 2a35afb48..791c41735 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -117,6 +117,7 @@ check_PROGRAMS += \ test_chat_attachment \ test_attachment_results \ test_attachment_decisions \ + test_attachment_requests \ test_db_sync_integrity \ test_reality_fallback \ test_etcp_padding \ @@ -467,6 +468,12 @@ test_encrypted_attachment_LDFLAGS += -Wl,--wrap=uasync_set_timeout -Wl,--wrap=me -Wl,--wrap=sqlite3_bind_blob -Wl,--wrap=etcp_router_conn_send endif if OS_LINUX +test_attachment_requests_SOURCES = test_attachment_requests.c +test_attachment_requests_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) +test_attachment_requests_LDFLAGS = -Wl,--wrap=sqlite3_exec -Wl,--wrap=sqlite3_prepare_v2 -Wl,--wrap=sqlite3_bind_blob \ + -Wl,--wrap=sqlite3_bind_text -Wl,--wrap=sqlite3_bind_int64 -Wl,--wrap=sqlite3_bind_int -Wl,--wrap=sqlite3_step \ + -Wl,--wrap=sqlite3_finalize -Wl,--wrap=sqlite3_changes -Wl,--wrap=etcp_router_has_route -Wl,--wrap=etcp_router_conn_get \ + -Wl,--wrap=etcp_router_send_q_has_room -Wl,--wrap=etcp_router_conn_send test_attachment_decisions_SOURCES = test_attachment_decisions.c test_attachment_decisions_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) test_attachment_decisions_LDFLAGS = -Wl,--wrap=sqlite3_exec -Wl,--wrap=sqlite3_prepare_v2 -Wl,--wrap=sqlite3_bind_blob \ diff --git a/tests/test_attachment_requests.c b/tests/test_attachment_requests.c new file mode 100644 index 000000000..a8233813a --- /dev/null +++ b/tests/test_attachment_requests.c @@ -0,0 +1,384 @@ +/* Настоящие core/SQLite/подписи; вход после router, независимый reader в ACK/событии. */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include "../src/utun_instance.h" +#include "../src/chat/chat_core.h" +#include "../src/chat/chat_attachment.h" +#include "../src/chat/chat_event.h" +#include "../src/chat/member_sync.h" +#include "../src/routing_layer/topo_node_sqlite.h" +#include "../src/routing_layer/etcp_router.h" +#include "../src/transport_layer/etcp.h" +#include "../lib/platform_compat.h" +#include "../lib/debug_config.h" +#include "../lib/mem.h" + +enum operation { NO_OPERATION, IDENTITY, MEMBER, AUTHOR, READ, INSERT, OUTBOX }; +enum failure { NO_FAILURE, BEGIN, PREPARE, BIND, STEP, FINALIZE, CHANGES, COMMIT }; +static enum failure fault; +static enum operation fault_operation, last_operation; +static int fault_index, fail_rollback; +static unsigned fault_hits, cases; + +struct snapshot { + uint8_t key[32]; + int exists, incoming, state, command, error; + int64_t created, retry; +}; +struct fixture { + struct UASYNC* ua; + struct UTUN_INSTANCE* inst; + struct ETCP_ROUTER_CONN route; /* Только наблюдение ACK; сеть/crypto проверяет encrypted_attachment. */ + sqlite3* reader; + char directory[256], channel[32]; + uint8_t message[32], id[16], hash[32], key[32]; + uint64_t peer; + struct snapshot expected; + int64_t started; + int allow_result, route_up, room, send_rc, corrupt, flags; + unsigned acks, events; +}; +static struct fixture* active; + +static enum operation operation(const char* query) { + if (!query) return NO_OPERATION; + static const char* prefixes[] = {NULL, "SELECT crypto_id,cipher_hash,key FROM attachment_keys WHERE", + "SELECT x25519_pubkey,ed25519_pubkey,signed_by,signature,join_sig,join_ts", "SELECT 1 FROM msg_", + "SELECT incoming,state FROM attachment_key_requests WHERE", "INSERT INTO attachment_key_requests(", + "INSERT INTO attachment_key_outbox("}; + for (enum operation op = IDENTITY; op <= OUTBOX; op++) + if (!strncmp(query, prefixes[op], strlen(prefixes[op]))) return op; + return NO_OPERATION; +} + +static int take(enum failure point, enum operation op, int index) { + if (fault != point || fault_operation != op || (point == BIND && fault_index != index)) return 0; + fault = NO_FAILURE; fault_hits++; + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "request test: injected point=%d operation=%d bind=%d", point, op, index); return 1; +} + +extern int __real_sqlite3_exec(sqlite3*, const char*, int (*)(void*, int, char**, char**), void*, char**); +int __wrap_sqlite3_exec(sqlite3* db, const char* query, int (*cb)(void*, int, char**, char**), void* arg, char** error) { + if (active && db == active->inst->topo_sqlite_db) { + if (!strcmp(query, "BEGIN IMMEDIATE") && take(BEGIN, NO_OPERATION, 0)) return SQLITE_BUSY; + if (!strcmp(query, "COMMIT") && take(COMMIT, NO_OPERATION, 0)) return SQLITE_BUSY; + if (!strcmp(query, "ROLLBACK") && fail_rollback) { + fail_rollback = 0; DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "request test: injected rollback failure"); return SQLITE_IOERR; + } + } + return __real_sqlite3_exec(db, query, cb, arg, error); +} + +extern int __real_sqlite3_prepare_v2(sqlite3*, const char*, int, sqlite3_stmt**, const char**); +int __wrap_sqlite3_prepare_v2(sqlite3* db, const char* query, int length, sqlite3_stmt** st, const char** tail) { + enum operation op = operation(query); + if (op && take(PREPARE, op, 0)) { *st = NULL; return SQLITE_NOMEM; } + return __real_sqlite3_prepare_v2(db, query, length, st, tail); +} + +extern int __real_sqlite3_bind_blob(sqlite3_stmt*, int, const void*, int, void (*)(void*)); +int __wrap_sqlite3_bind_blob(sqlite3_stmt* st, int index, const void* value, int length, void (*destroy)(void*)) { + if (take(BIND, operation(sqlite3_sql(st)), index)) return SQLITE_NOMEM; + return __real_sqlite3_bind_blob(st, index, value, length, destroy); +} + +extern int __real_sqlite3_bind_text(sqlite3_stmt*, int, const char*, int, void (*)(void*)); +int __wrap_sqlite3_bind_text(sqlite3_stmt* st, int index, const char* value, int length, void (*destroy)(void*)) { + if (take(BIND, operation(sqlite3_sql(st)), index)) return SQLITE_NOMEM; + return __real_sqlite3_bind_text(st, index, value, length, destroy); +} + +extern int __real_sqlite3_bind_int64(sqlite3_stmt*, int, sqlite3_int64); +int __wrap_sqlite3_bind_int64(sqlite3_stmt* st, int index, sqlite3_int64 value) { + if (take(BIND, operation(sqlite3_sql(st)), index)) return SQLITE_NOMEM; + return __real_sqlite3_bind_int64(st, index, value); +} + +extern int __real_sqlite3_bind_int(sqlite3_stmt*, int, int); +int __wrap_sqlite3_bind_int(sqlite3_stmt* st, int index, int value) { + if (take(BIND, operation(sqlite3_sql(st)), index)) return SQLITE_NOMEM; + return __real_sqlite3_bind_int(st, index, value); +} + +extern int __real_sqlite3_step(sqlite3_stmt*); +int __wrap_sqlite3_step(sqlite3_stmt* st) { + last_operation = operation(sqlite3_sql(st)); + if (last_operation && take(STEP, last_operation, 0)) return SQLITE_IOERR; + return __real_sqlite3_step(st); +} + +extern int __real_sqlite3_finalize(sqlite3_stmt*); +int __wrap_sqlite3_finalize(sqlite3_stmt* st) { + enum operation op = st ? operation(sqlite3_sql(st)) : NO_OPERATION; + int failed = op && take(FINALIZE, op, 0); + int rc = __real_sqlite3_finalize(st); return failed ? SQLITE_IOERR : rc; +} + +extern int __real_sqlite3_changes(sqlite3*); +int __wrap_sqlite3_changes(sqlite3* db) { + if (last_operation && take(CHANGES, last_operation, 0)) return 0; + return __real_sqlite3_changes(db); +} + +static void sql(struct fixture* f, const char* query) { + int rc = sqlite3_exec(f->inst->topo_sqlite_db, query, NULL, NULL, NULL); + if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "request test SQL rc=%d: %s", rc, sqlite3_errmsg(f->inst->topo_sqlite_db)); + assert(rc == SQLITE_OK); +} + +static struct snapshot snapshot(struct fixture* f) { + sqlite3_stmt* st = NULL; struct snapshot result = {0}; + assert(sqlite3_prepare_v2(f->reader, "SELECT a.key,a.error,r.message_id IS NOT NULL,coalesce(r.incoming,-1),coalesce(r.state,-1)," + "coalesce(r.created_at,0),coalesce(o.command,0),coalesce(o.retry_at,0) FROM attachment_keys a" + " LEFT JOIN attachment_key_requests r ON r.channel=a.channel AND r.message_id=a.message_id AND r.peer=?3" + " LEFT JOIN attachment_key_outbox o ON o.channel=a.channel AND o.message_id=a.message_id AND o.peer=?3" + " WHERE a.channel=?1 AND a.message_id=?2", -1, &st, NULL) == SQLITE_OK); + assert(sqlite3_bind_text(st, 1, f->channel, -1, SQLITE_STATIC) == SQLITE_OK); + assert(sqlite3_bind_blob(st, 2, f->message, 32, SQLITE_STATIC) == SQLITE_OK); + assert(sqlite3_bind_int64(st, 3, (sqlite3_int64)f->peer) == SQLITE_OK && sqlite3_step(st) == SQLITE_ROW); + assert(sqlite3_column_bytes(st, 0) == 32); memcpy(result.key, sqlite3_column_blob(st, 0), 32); + result.error = sqlite3_column_int(st, 1); result.exists = sqlite3_column_int(st, 2); + result.incoming = sqlite3_column_int(st, 3); result.state = sqlite3_column_int(st, 4); + result.created = sqlite3_column_int64(st, 5); result.command = sqlite3_column_int(st, 6); result.retry = sqlite3_column_int64(st, 7); + assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK); + return result; +} + +static void equal(struct fixture* f) { + struct snapshot current = snapshot(f); + if (f->expected.created < 0) { + assert(current.created >= f->started && current.created <= (int64_t)time(NULL)); f->expected.created = current.created; + } + assert(!memcmp(current.key, f->expected.key, 32) && current.error == f->expected.error && current.exists == f->expected.exists); + assert(current.incoming == f->expected.incoming && current.state == f->expected.state && current.created == f->expected.created); + assert(current.command == f->expected.command && current.retry == f->expected.retry); + sqlite3_stmt* st = NULL; + assert(sqlite3_prepare_v2(f->reader, "SELECT key,error FROM attachment_keys WHERE channel='900'", -1, &st, NULL) == SQLITE_OK); + assert(sqlite3_step(st) == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32 && sqlite3_column_int(st, 1) == 7); + uint8_t key[32] = {99}; assert(!memcmp(sqlite3_column_blob(st, 0), key, 32)); + assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK); +} + +extern int __real_etcp_router_has_route(struct UTUN_INSTANCE*, uint64_t, uint64_t); +int __wrap_etcp_router_has_route(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer) { + if (active && inst == active->inst && group == active->route.group_id && peer == active->peer) return active->route_up; + return __real_etcp_router_has_route(inst, group, peer); +} + +extern struct ETCP_ROUTER_CONN* __real_etcp_router_conn_get(struct UTUN_INSTANCE*, uint64_t, uint64_t, uint8_t); +struct ETCP_ROUTER_CONN* __wrap_etcp_router_conn_get(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, uint8_t service) { + if (active && inst == active->inst && group == active->route.group_id && peer == active->peer && service == ETCP_RT_ID_ATTACHMENT_KEYS) + return &active->route; + return __real_etcp_router_conn_get(inst, group, peer, service); +} + +extern int __real_etcp_router_send_q_has_room(struct UTUN_INSTANCE*, uint64_t, uint64_t, uint8_t); +int __wrap_etcp_router_send_q_has_room(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, uint8_t service) { + if (active && inst == active->inst && group == active->route.group_id && peer == active->peer && service == ETCP_RT_ID_ATTACHMENT_KEYS) + return active->room; + return __real_etcp_router_send_q_has_room(inst, group, peer, service); +} + +extern int __real_etcp_router_conn_send(struct ETCP_ROUTER_CONN*, const uint8_t*, size_t, int); +int __wrap_etcp_router_conn_send(struct ETCP_ROUTER_CONN* conn, const uint8_t* data, size_t length, int flags) { + if (!active || conn != &active->route) return __real_etcp_router_conn_send(conn, data, length, flags); + assert(active->allow_result && length == 85 && !memcmp(data, "AK01", 4) && data[4] == 2); + assert(flags == (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED) && sqlite3_get_autocommit(active->inst->topo_sqlite_db)); + assert(!memcmp(data + 5, active->message, 32) && !memcmp(data + 37, active->id, 16) && !memcmp(data + 53, active->hash, 32)); + equal(active); active->acks++; return active->send_rc; +} + +static void event(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int bytes) { + if (type != CHAT_EVT_ATTACHMENT_KEYS_CHANGED) return; + struct fixture* f = active; + assert(f && f->inst == inst && f->allow_result && sqlite3_get_autocommit(inst->topo_sqlite_db)); + assert(bytes == (int)strlen(f->channel) + 1 && data[0] == strlen(f->channel) && !memcmp(data + 1, f->channel, strlen(f->channel))); + equal(f); f->events++; +} + +static void receive(struct fixture* f, int success) { + f->expected = snapshot(f); f->allow_result = success; f->acks = f->events = 0; f->started = (int64_t)time(NULL); + if (success) { + if (!f->expected.exists) { f->expected.exists = 1; f->expected.incoming = 1; f->expected.state = 0; f->expected.created = -1; } + if (f->expected.state) { f->expected.command = f->expected.state == 3 ? 4 : 3; f->expected.retry = 0; } + } + struct ll_entry* entry = ll_alloc_lldgram(ROUTER_SVC_PAYLOAD_OFF + 85); assert(entry); + entry->len = ROUTER_SVC_PAYLOAD_OFF + 85; memset(entry->dgram, 0, entry->len); + entry->dgram[0] = ETCP_RT_ID_ATTACHMENT_KEYS; entry->dgram[ROUTER_SVC_FLAGS_OFF] = f->flags; + memcpy(entry->dgram + ROUTER_SVC_SRC_OFF, &f->peer, 8); memcpy(entry->dgram + ROUTER_SVC_DST_OFF, &f->inst->node_id, 8); + memcpy(entry->dgram + ROUTER_SVC_GROUP_OFF, &f->route.group_id, 8); + uint8_t* body = entry->dgram + ROUTER_SVC_PAYLOAD_OFF; memcpy(body, "AK01", 4); body[4] = 1; + memcpy(body + 5, f->message, 32); memcpy(body + 37, f->id, 16); memcpy(body + 53, f->hash, 32); + if (f->corrupt) body[53] ^= 1; + struct ETCP_CONN conn = {.instance = f->inst}; + assert(f->inst->router_bindings.callbacks[ETCP_RT_ID_ATTACHMENT_KEYS]); + f->inst->router_bindings.callbacks[ETCP_RT_ID_ATTACHMENT_KEYS](&conn, entry); + equal(f); assert(f->events == (unsigned)success && f->acks == (unsigned)(success && f->route_up && f->room)); +} + +static void reset(struct fixture* f, int state, int queued) { + char query[768]; + snprintf(query, sizeof(query), "DELETE FROM attachment_key_requests WHERE channel='%s';DELETE FROM attachment_key_outbox WHERE channel='%s'", + f->channel, f->channel); sql(f, query); + if (state >= 0) { + snprintf(query, sizeof(query), "INSERT INTO attachment_key_requests(channel,message_id,peer,incoming,state,created_at)" + " SELECT channel,message_id,%lld,1,%d,123 FROM attachment_keys WHERE channel='%s'", + (long long)(sqlite3_int64)f->peer, state, f->channel); sql(f, query); + } + if (queued) { + snprintf(query, sizeof(query), "INSERT INTO attachment_key_outbox(channel,message_id,peer,command,retry_at)" + " SELECT channel,message_id,peer,%d,17 FROM attachment_key_requests WHERE channel='%s'", state == 3 ? 4 : 3, f->channel); + sql(f, query); + } + f->route_up = f->room = 1; f->send_rc = f->corrupt = 0; f->flags = ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED; +} + +static void setup(struct fixture* f) { + memset(f, 0, sizeof(*f)); strcpy(f->directory, "/tmp/utun-attachment-request-XXXXXX"); assert(mkdtemp(f->directory)); + char config[1024], path[512]; + snprintf(config, sizeof(config), "[global]\ndb_path=%s\n[ntp]\nenabled=no\n" + "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n", f->directory); + f->ua = uasync_create(); assert(f->ua); f->inst = utun_instance_create_from_str(f->ua, config); + assert(f->inst && !utun_core_start(f->inst) && !chat_service_start(f->inst)); + chat_core_create_channel_auto(f->inst, "request contract"); + sqlite3_stmt* st = NULL; + assert(sqlite3_prepare_v2(f->inst->topo_sqlite_db, "SELECT channel_id FROM channels WHERE name='request contract'", -1, &st, NULL) == SQLITE_OK); + assert(sqlite3_step(st) == SQLITE_ROW); snprintf(f->channel, sizeof(f->channel), "%s", sqlite3_column_text(st, 0)); + assert(sqlite3_finalize(st) == SQLITE_OK); + uint8_t x[32], ed[32], private_key[32], peer_ed[32], seed[64], join[64], certificate[64], body[112]; + struct SC_MYKEYS peer; assert(sc_generate_keypair(&peer) == SC_OK); + f->peer = sc_derive_node_id_from_pubkey(peer.public_key); assert(f->peer && f->peer != f->inst->node_id); + assert(sc_derive_ed25519_pubkey(peer.private_key, peer_ed) == SC_OK); SHA512(peer.private_key, 32, seed); + assert(topo_node_sqlite_channel_get(f->inst->topo_sqlite_db, f->channel, NULL, 0, NULL, x, ed, NULL) == 0); + assert(topo_node_sqlite_channel_get_priv(f->inst->topo_sqlite_db, f->channel, private_key) == 0); + assert(member_sync_build_join_msg(x, ed, f->peer, peer.public_key, 1, body, sizeof(body)) == 112); + assert(sc_ed25519_sign(seed, body, sizeof(body), join) == SC_OK && member_sync_sign_pubkey(private_key, peer.public_key, certificate) == 0); + assert(member_sync_put(f->inst, f->channel, f->peer, peer.public_key, peer_ed, join, 1, + NULL, 0, "{}", NULL, NULL, 0, 0, certificate) >= 0); + assert(member_sync_verified_ed25519(f->inst, f->channel, f->peer, peer_ed) == 0); + struct chat_msg_submit request = {.inst = f->inst, .data = (uint8_t*)"request identity", .data_len = 16}; + strcpy(request.channel_id, f->channel); strcpy(request.content_type, "text"); + assert(chat_core_submit_message(f->inst, &request) == 0); memcpy(f->message, request.message_id, 32); + f->key[0] = 55; f->id[0] = 2; f->hash[0] = 3; + assert(sqlite3_prepare_v2(f->inst->topo_sqlite_db, "INSERT INTO attachment_keys(channel,crypto_id,message_id,author,key,cipher_hash,cipher_path,plain_path)" + " VALUES(?,?,?,?,?,?,'','')", -1, &st, NULL) == SQLITE_OK); + assert(sqlite3_bind_text(st, 1, f->channel, -1, SQLITE_STATIC) == SQLITE_OK); + assert(sqlite3_bind_blob(st, 2, f->id, 16, SQLITE_STATIC) == SQLITE_OK && sqlite3_bind_blob(st, 3, f->message, 32, SQLITE_STATIC) == SQLITE_OK); + assert(sqlite3_bind_int64(st, 4, (sqlite3_int64)f->inst->node_id) == SQLITE_OK); + assert(sqlite3_bind_blob(st, 5, f->key, 32, SQLITE_STATIC) == SQLITE_OK && sqlite3_bind_blob(st, 6, f->hash, 32, SQLITE_STATIC) == SQLITE_OK); + assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK); + sql(f, "INSERT INTO attachment_keys(channel,crypto_id,author,key,cipher_path,plain_path,error)" + " VALUES('900',zeroblob(16),999,x'6300000000000000000000000000000000000000000000000000000000000000','','',7)"); + snprintf(path, sizeof(path), "%s/chats.db", f->directory); + assert(sqlite3_open_v2(path, &f->reader, SQLITE_OPEN_READONLY, NULL) == SQLITE_OK); + f->route.inst = f->inst; f->route.group_id = strtoull(f->channel, NULL, 10); f->route.remote_node_id = f->peer; + f->route.svc_id = ETCP_RT_ID_ATTACHMENT_KEYS; active = f; chat_event_set_handler(f->inst, event); reset(f, -1, 0); +} + +static void cleanup(struct fixture* f) { + chat_event_set_handler(f->inst, NULL); active = NULL; utun_instance_destroy(f->inst); + assert(sqlite3_close(f->reader) == SQLITE_OK); uasync_destroy(f->ua, 0); +} + +static void failed_then_retry(struct fixture* f, int state, int queued, enum failure point, enum operation op, int index) { + reset(f, state, queued); fault = point; fault_operation = op; fault_index = index; unsigned previous = fault_hits; + receive(f, 0); assert(!fault && fault_hits == previous + 1 && sqlite3_get_autocommit(f->inst->topo_sqlite_db)); + receive(f, 1); receive(f, 1); cases++; +} + +static void test_faults(void) { + struct fixture f; setup(&f); const int states[] = {-1, 0, 2, 3, 4}; + for (unsigned i = 0; i < sizeof(states) / sizeof(states[0]); i++) for (int queued = 0; queued <= (states[i] >= 2); queued++) { + int state = states[i]; failed_then_retry(&f, state, queued, BEGIN, NO_OPERATION, 0); + for (enum operation op = IDENTITY; op <= OUTBOX; op++) { + if ((op == INSERT && state >= 0) || (op == OUTBOX && state < 2)) continue; + failed_then_retry(&f, state, queued, PREPARE, op, 0); + int bindings = op == IDENTITY ? 2 : op == MEMBER ? 1 : op == AUTHOR || op == OUTBOX ? 4 : 3; + for (int index = 1; index <= bindings; index++) failed_then_retry(&f, state, queued, BIND, op, index); + failed_then_retry(&f, state, queued, STEP, op, 0); failed_then_retry(&f, state, queued, FINALIZE, op, 0); + if (op == INSERT || op == OUTBOX) failed_then_retry(&f, state, queued, CHANGES, op, 0); + } + failed_then_retry(&f, state, queued, COMMIT, NO_OPERATION, 0); + } + cleanup(&f); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment requests: %u SQL faults/retries, new and duplicate 0/2/3/4, exact committed snapshots", cases); +} + +static int veto(void* arg) { (*(unsigned*)arg)++; return 1; } + +static void test_controls(void) { + struct fixture f; setup(&f); const int states[] = {-1, 0, 2, 3, 4}; + for (unsigned i = 0; i < sizeof(states) / sizeof(states[0]); i++) { + reset(&f, states[i], 0); sql(&f, "BEGIN;UPDATE attachment_keys SET error=8 WHERE channel='900'"); + receive(&f, 0); assert(!sqlite3_get_autocommit(f.inst->topo_sqlite_db)); + sqlite3_stmt* st = NULL; + assert(sqlite3_prepare_v2(f.inst->topo_sqlite_db, "SELECT error FROM attachment_keys WHERE channel='900'", -1, &st, NULL) == SQLITE_OK); + assert(sqlite3_step(st) == SQLITE_ROW && sqlite3_column_int(st, 0) == 8 && sqlite3_finalize(st) == SQLITE_OK); + sql(&f, "ROLLBACK"); receive(&f, 1); + } + const char* triggers[] = { + "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_requests BEGIN SELECT RAISE(ABORT,'request'); END", + "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_requests BEGIN SELECT RAISE(IGNORE); END", + "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_outbox BEGIN SELECT RAISE(ABORT,'queue'); END", + "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_outbox BEGIN SELECT RAISE(IGNORE); END", + "CREATE TEMP TRIGGER reject_request BEFORE UPDATE ON attachment_key_outbox BEGIN SELECT RAISE(ABORT,'upsert'); END", + "CREATE TEMP TRIGGER reject_request BEFORE UPDATE ON attachment_key_outbox BEGIN SELECT RAISE(IGNORE); END" + }; + for (unsigned i = 0; i < sizeof(triggers) / sizeof(triggers[0]); i++) { + reset(&f, i < 2 ? -1 : 2, i >= 4); sql(&f, triggers[i]); receive(&f, 0); + sql(&f, "DROP TRIGGER reject_request"); receive(&f, 1); + } + for (int state = -1; state <= 2; state += 3) { + reset(&f, state, 0); unsigned commits = 0; sqlite3_commit_hook(f.inst->topo_sqlite_db, veto, &commits); + receive(&f, 0); assert(commits == 1); sqlite3_commit_hook(f.inst->topo_sqlite_db, NULL, NULL); receive(&f, 1); + } + for (int state = 0; state <= 5; state++) if (state == 1 || state == 5) { reset(&f, state, 1); receive(&f, 0); } + reset(&f, 0, 0); char query[256]; + snprintf(query, sizeof(query), "UPDATE attachment_key_requests SET incoming=0 WHERE channel='%s'", f.channel); + sql(&f, query); receive(&f, 0); + reset(&f, -1, 0); f.corrupt = 1; receive(&f, 0); f.corrupt = 0; + f.flags = ROUTER_FLAG_SIGNED; receive(&f, 0); f.flags |= ROUTER_FLAG_ENCRYPTED; + snprintf(query, sizeof(query), "UPDATE peers_%s SET deleted=1 WHERE node_id=%lld", f.channel, (long long)(sqlite3_int64)f.peer); + sql(&f, query); receive(&f, 0); + snprintf(query, sizeof(query), "UPDATE peers_%s SET deleted=0 WHERE node_id=%lld", f.channel, (long long)(sqlite3_int64)f.peer); + sql(&f, query); receive(&f, 1); cleanup(&f); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment requests: foreign transaction preserved, ABORT/IGNORE, commit veto, direction/state/identity/member checks"); +} + +static void test_ack_retry(void) { + struct fixture f; setup(&f); + for (int unavailable = 0; unavailable < 3; unavailable++) { + reset(&f, -1, 0); + if (unavailable == 0) f.route_up = 0; + else if (unavailable == 1) f.room = 0; + else f.send_rc = -1; + receive(&f, 1); struct snapshot saved = snapshot(&f); + f.route_up = f.room = 1; f.send_rc = 0; receive(&f, 1); + assert(saved.created == f.expected.created && f.expected.command == 0); + } + cleanup(&f); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment requests: missing route, backpressure and send failure retain request; duplicate retries ACK without new consent"); +} + +static void test_rollback_failure(void) { + struct fixture f; setup(&f); fault = COMMIT; fault_operation = NO_OPERATION; fail_rollback = 1; receive(&f, 0); + assert(!fault && !fail_rollback && !f.inst->running && !sqlite3_get_autocommit(f.inst->topo_sqlite_db)); + assert(!f.inst->router_bindings.callbacks[ETCP_RT_ID_ATTACHMENT_KEYS]); + chat_event_set_handler(f.inst, NULL); active = NULL; utun_instance_destroy(f.inst); + equal(&f); assert(sqlite3_close(f.reader) == SQLITE_OK); uasync_destroy(f.ua, 0); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment requests: failed rollback sends no ACK/event, quiesces/stops core, owner close rolls back request"); +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_INFO); + utun_instance_set_tun_init_enabled(0); size_t baseline = u_get_allocated_count(); + test_faults(); test_controls(); test_ack_retry(); test_rollback_failure(); + assert(!fault && !fail_rollback && u_get_allocated_count() == baseline); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment requests: reader before ACK/events, stable duplicate date, neighboring key and allocations preserved"); + return 0; +} From 2fe024fbbba655059576dabacbe6393d1115dfe9 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 00:54:11 +0100 Subject: [PATCH 003/164] docs(dm): clarify core and history ownership contracts --- src/dm/dm_core.h | 90 +++++++++++++++++++++++++++++---------------- src/dm/dm_history.h | 14 +++++-- 2 files changed, 69 insertions(+), 35 deletions(-) diff --git a/src/dm/dm_core.h b/src/dm/dm_core.h index 3bd92a6e0..00f72fcf6 100644 --- a/src/dm/dm_core.h +++ b/src/dm/dm_core.h @@ -9,7 +9,11 @@ * при недоступности пира — через dm_mailbox (суперузлы). * - содержимое шифруется E2E (AES-256-CCM), запись подписывается Ed25519. * - * Два независимых направленных потока (я→пир и пир→я) с монотонным seq. + * Два независимых направленных потока (я→пир и пир→я) с монотонным исходящим seq. + * Все API и трамплины выполняются в uasync-потоке с живым instance/SQLite/router. + * Входные строки/тела заимствуются на время вызова; функции не забирают их владение. + * Запросы трамплинов — один блок u_malloc/u_calloc с NUL-строками; обработчик освобождает блок. + * Instance должен пережить запрос и его отложенное завершение; остановка сериализуется через uasync. */ #ifndef DM_CORE_H @@ -29,7 +33,9 @@ extern "C" { /* ── Каноническое тело DM-сообщения (общий wire-формат dm_core и dm_mailbox) ── * [conv_id:8][seq:8][ts:8][author:8][ct_len:1][ct:ct_len][data_len:2][data_enc:data_len][media?:56][sig:64] * media присутствует только для ct="file", подписан вместе с сообщением. - * data_enc — AES-256-CCM (шифротекст + tag), sig — Ed25519 автора по всем полям до sig. */ + * data_enc — AES-256-CCM (шифротекст + tag), sig — Ed25519 автора по всем полям до sig. + * Числа native-endian; seq=1..INT64_MAX, ts — миллисекунды NTP-времени instance. + * ETCP_RT_ID_DM=0x34: [MSG=1][тело] либо [ACK=2][квитанция:136]; router-пакет подписан. */ #define DM_MSG_FIXED_HDR 33 /* conv_id+seq+ts+author(32) + ct_len(1) — до ct */ #define DM_MSG_SIG_SIZE 64 #define DM_DATA_MAX 1024 @@ -54,22 +60,29 @@ static inline int dm_msg_len(const uint8_t* msg, size_t avail, size_t* out_len) return 1; } -/* Общие операции протокола DM; все вызовы в uasync-потоке. - * group_id — только область маршрутизации, идентичность беседы от неё не зависит. - * route_group выбирает общую группу с живым путём, возвращает 1/0. - * verify_message проверяет каноническое тело и подпись автора для recipient. - * accept_message возвращает 0 только после commit (включая точный повтор). - * Квитанция: ["DMACK001":8][conv:8][seq:8][author:8][recipient:8][SHA256(msg):32][sig:64]. - * verify_receipt проверяет подпись получателя и при msg!=NULL точное соответствие телу. - * accept_receipt удаляет только соответствующее исходящее pending-сообщение. */ +/* Выбрать общую CHAT-группу с путём router к peer, сначала preferred (0=без предпочтения). + * 1 и *group_id при успехе; 0 при отсутствии пути/ошибке, выход не меняется. + * group_id обязателен; наличие пути не гарантирует успешной последующей отправки. */ int dm_route_group(struct UTUN_INSTANCE* inst, uint64_t peer, uint64_t preferred, uint64_t* group_id); +/* Проверить точную длину msg/len, conv/author/recipient/seq, descriptor и подпись автора из nodes. + * 0=корректно, -1=отказ. Нужна SQLite, владелец dm_core не требуется. + * Не расшифровывает тело и не проверяет членство/роль router-отправителя: это обязанность вызывающего. */ int dm_verify_message(struct UTUN_INSTANCE* inst, uint64_t recipient, const uint8_t* msg, size_t len); +/* Проверить и расшифровать входящее тело для локального узла, атомарно сохранить историю/беседу/media. + * 0 и receipt после commit либо точного повтора; -1 при отказе. Для file нужен живой dm_media. + * Точный удалённый повтор подтверждается по отпечатку без восстановления истории; конфликт отклоняется. + * receipt — обязательные 136 байт; при -1 буфер не использовать. Ошибка подписи квитанции возможна после commit. */ int dm_accept_message(struct UTUN_INSTANCE* inst, const uint8_t* msg, size_t len, uint8_t receipt[DM_RECEIPT_SIZE]); +/* receipt — ровно 136 доступных байт: ["DMACK001":8][conv:8][seq:8][author:8][recipient:8][SHA256(msg):32][sig:64]. + * Проверяет идентичность и подпись B по первым 72 байтам; при msg!=NULL также точное тело msg/len. + * 0=корректно, -1=отказ. Не проверяет наличие outbox и ничего не записывает. */ int dm_verify_receipt(struct UTUN_INSTANCE* inst, const uint8_t receipt[DM_RECEIPT_SIZE], const uint8_t* msg, size_t len); +/* Для локального автора проверить receipt и удалить только точный outbox; история/file receipt сохраняются. + * 0=удалено либо outbox уже отсутствует, -1=ошибка/конфликт. Требуется запущенный dm_core. */ int dm_accept_receipt(struct UTUN_INSTANCE* inst, const uint8_t receipt[DM_RECEIPT_SIZE]); /* Жизненный цикл в uasync: init после chat_core_init с SQLite/router/ua ядра; 0=успех, -1=ошибка. - * Повторный init запущенного владельца возвращает 0. Для send/accept нужен также dm_media_init. + * Повторный init запущенного владельца возвращает 0. Для file-send/accept нужен также dm_media_init. * destroy отменяет invite/таймер и снимает bind; БД/outbox сохраняются. Вызывать до chat_core_destroy. */ int dm_core_init(struct UTUN_INSTANCE* inst); void dm_core_destroy(struct UTUN_INSTANCE* inst); @@ -99,9 +112,16 @@ int dm_start(struct UTUN_INSTANCE* inst, uint64_t peer_node_id, int dm_send(struct UTUN_INSTANCE* inst, const char* conv_id, const char* content_type, const uint8_t* data, uint32_t data_len); -/* Локальное удаление; generation защищает повторно открытую беседу от старого запроса. */ -struct dm_delete_req { struct UTUN_INSTANCE* inst; char conv_id[64]; uint64_t generation, request_id; }; +/* Локальное удаление по dm_history_delete; generation защищает повторно открытую беседу. + * CHAT_EVT_DM_DELETE_RESULT: [request_id:8][conv:u64][generation:8][success:u8], всего 25 байт native-endian. */ +struct dm_delete_req { + struct UTUN_INSTANCE* inst; /* Заимствованное ядро до завершения запроса. */ + char conv_id[64]; /* Десятичный ID беседы с NUL. */ + uint64_t generation; /* Ненулевое ожидаемое поколение. */ + uint64_t request_id; /* Корреляция результата с запросом UI. */ +}; void dm_delete_trampoline(void* arg); +/* После commit удаления отменить ожидающий invite этой беседы; квитанции/другие беседы не меняет. */ void dm_cancel_pending_history(struct UTUN_INSTANCE* inst, const char* conv); /* До 200 видимых бесед по last_ts DESC: [{conv_id,peer_node_id,peer_name,last_ts,out_seq,in_seq,generation}]. @@ -113,7 +133,9 @@ int dm_list_conversations_json(struct UTUN_INSTANCE* inst, char* buf, size_t buf * count=1..200 задаёт предел; count<=0 или count>200 означает 200. * offset>=0 отсчитывается от новых сообщений; offset=0 выбирает последние count. * Выбранная страница возвращается по возрастанию ts, seq, dir. - * buf — выходной буфер с NUL; 0 и *out_len байт без NUL при успехе, -1 при ошибке. */ + * buf — выходной буфер с NUL; 0 и *out_len байт без NUL при успехе, -1 при ошибке. + * pending относится к квитанции сообщения. media_status: delivered/ready/downloading/expired/pending. + * hidden здесь не фильтруется; интерфейс выбирает видимую беседу по списку/поколению. */ 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); @@ -123,20 +145,20 @@ int dm_list_messages_json(struct UTUN_INSTANCE* inst, const char* conv_id, int c * пира в таблице nodes (как headless hc_handle_dm_start), после чего вызывает * dm_start(). Результат — событие CHAT_EVT_DM_CONV_UPDATED. */ struct dm_start_req { - struct UTUN_INSTANCE* inst; - uint64_t peer_node_id; - char source_ch_id[64]; - char peer_name[128]; + struct UTUN_INSTANCE* inst; /* Заимствованное ядро. */ + uint64_t peer_node_id; /* Другой участник общей группы; ключи читаются из nodes. */ + char source_ch_id[64]; /* Десятичный ID общей группы либо пустая строка. */ + char peer_name[128]; /* UTF-8 имя с NUL; dm_start ограничит до 63 байт. */ }; void dm_start_trampoline(void* arg); /* Отправить текстовое сообщение в беседу (content_type="text"). */ struct dm_send_req { - struct UTUN_INSTANCE* inst; - char conv_id[64]; + struct UTUN_INSTANCE* inst; /* Заимствованное ядро. */ + char conv_id[64]; /* Десятичный ID беседы с NUL. */ uint64_t generation; /* 0: захватить при входе в uasync, до любой фоновой работы. */ - uint32_t data_len; - uint8_t* data; /* data_len байт (выделяется в хвосте структуры) */ + uint32_t data_len; /* Число байт текста, до DM_DATA_MAX. */ + uint8_t* data; /* Указывает на data_len байт в том же блоке после структуры. */ }; void dm_send_trampoline(void* arg); @@ -151,12 +173,15 @@ enum dm_invite_result { DM_INVITE_SEND_ERROR = -3, DM_INVITE_CANCELLED = -4, DM_INVITE_BUSY = -5 }; struct dm_invite_req { - struct UTUN_INSTANCE* inst; - uint64_t request_id, peer_node_id, channel_id, generation; + struct UTUN_INSTANCE* inst; /* Заимствованное ядро до итогового callback. */ + uint64_t request_id; /* Ненулевой ID; cancel использует тот же ID. */ + uint64_t peer_node_id; /* Фиксированный получатель, не локальный узел. */ + uint64_t channel_id; /* Выбранная активная CHAT-группа отправителя. */ + uint64_t generation; /* Заполняет uasync до регистрации; проверяет callback. */ char source_ch_id[64]; /* общая группа для новой PM; из шапки PM может быть пустой */ - char peer_name[128]; - int cancel; - struct chat_invite_build* pending; /* только uasync */ + char peer_name[128]; /* UTF-8 имя для новой беседы с NUL. */ + int cancel; /* 0=создать, 1=отменить ожидающий request_id. */ + struct chat_invite_build* pending; /* Внутренний handle; запрос создаётся обнулённым. */ }; void dm_invite_trampoline(void* arg); @@ -164,15 +189,18 @@ void dm_invite_trampoline(void* arg); * [conv_id_len:1][conv_id:var][generation:8][JSON:utf8]. Все чтения в uasync. * Ненулевое generation проверяется до чтения; 0 захватывает текущее поколение. GUI отбрасывает старый снимок. */ struct dm_messages_req { - struct UTUN_INSTANCE* inst; - char conv_id[64]; + struct UTUN_INSTANCE* inst; /* Заимствованное ядро. */ + char conv_id[64]; /* Десятичный ID с NUL, также для запроса маршрута звонка. */ uint64_t generation; /* Поколение выбранного PM; для запроса маршрута не используется. */ }; void dm_messages_trampoline(void* arg); -/* Живой BGP-маршрут PM; group_id=0 в снимке означает недоступность. Все вызовы в uasync. */ +/* 1 при живом BGP-маршруте общей группы, 0 при недоступности/ошибке. peer/group_id обязательны. + * При корректных аргументах выходы сначала обнуляются; peer может быть известен и без маршрута. + * Трамплин выдаёт CHAT_EVT_DM_CALL_ROUTE: [conv_len:1][conv][peer:8][group:8]; group=0 без маршрута. */ int dm_call_route(struct UTUN_INSTANCE* inst, const char* conv_id, uint64_t* peer, uint64_t* group_id); void dm_call_route_trampoline(void* arg); /* аргумент: dm_messages_req */ -struct dm_conversations_req { struct UTUN_INSTANCE* inst; }; +/* Снимок бесед публикуется как CHAT_EVT_DM_CONVERSATIONS с JSON без NUL. */ +struct dm_conversations_req { struct UTUN_INSTANCE* inst; /* Заимствованное ядро. */ }; void dm_conversations_trampoline(void* arg); #ifdef __cplusplus diff --git a/src/dm/dm_history.h b/src/dm/dm_history.h index 22eb42121..9c906de58 100644 --- a/src/dm/dm_history.h +++ b/src/dm/dm_history.h @@ -1,7 +1,9 @@ /* Локальная история DM: поколения, скрытые беседы и точные отпечатки удалённых сообщений. * Все операции выполняются в uasync с SQLite ядра. init вызывается после создания DM-таблиц. * Удаление скрывает беседу и сохраняет её ключи/счётчики в dm_conversations; show открывает её снова. - * Поколение передавать в отложенные операции и проверять current перед применением результата. */ + * Поколение передавать в отложенные операции и проверять current перед применением результата. + * Беседой/seq владеет dm_core, файлами/загрузками — dm_media; этот модуль связывает их удаление. + * Аргументы заимствованы на время вызова; самостоятельные транзакции выполняются без внешней транзакции. */ #ifndef DM_HISTORY_H #define DM_HISTORY_H #include @@ -10,7 +12,8 @@ extern "C" { #endif struct UTUN_INSTANCE; -/* Создать служебные таблицы, сбросить busy файлов после restart. 0=успех, -1=ошибка. +/* Создать dm_history, dm_deleted_messages и dm_local_files; сбросить busy после restart. + * 0=успех, -1=ошибка. Вызывает dm_core_init; повторять при живых workers нельзя: они ещё держат пути. * inst и его topo_sqlite_db обязательны; частично созданные таблицы при ошибке остаются. */ int dm_history_init(struct UTUN_INSTANCE* inst); /* Поколение существующей беседы: 1 до первого удаления; 0 при отсутствии/ошибке. @@ -28,8 +31,11 @@ int dm_history_show(struct UTUN_INSTANCE* inst, const char* conv); int dm_history_deleted(struct UTUN_INSTANCE* inst, const uint8_t* body, size_t len, int dir, uint8_t* media_receipt); /* Удалить видимую беседу текущего generation