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

Commit incoming attachment requests before acknowledging them

Browse Source
This commit is contained in:
evgeny committed 2026-10-27 01:02:55 +01:00
1 parent 15ec41f2fd
commit 760398b96c
2 files changed
+112 -34
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/chat_attachment.c
+108 -31
View File
@@ -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,
src/chat/chat_attachment.h
+4 -3
View File
@@ -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"