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

Authorize attachment key exchange and cancel durable revoked work

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 21:00:59 +01:00
1 parent 1b34f5ad82
commit f3cff571dc
3 files changed
+137 -9
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/chat_attachment.c
+126 -8
View File
@@ -8,6 +8,7 @@
#include "../utun_instance.h"
#include "../transport_layer/etcp.h"
#include "../routing_layer/etcp_router.h"
#include "../routing_layer/topo_group.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../media_delivery/media_index.h"
#include "../media_async/media_async.h"
@@ -148,6 +149,88 @@ static void ak_file_work(void* arg) {
static void ak_rollback(struct UTUN_INSTANCE* inst, const char* operation, const char* channel);
/* Отменить работу пары одним commit, сохранив полученные ключи и завершённые решения. */
static int ak_cancel_peer(struct UTUN_INSTANCE* inst, const char* channel, uint64_t peer) {
sqlite3* db = CC(inst)->db; sqlite3_stmt* st = NULL;
const char* stage = "begin revoke";
int rc = sqlite3_exec(db, "BEGIN IMMEDIATE /* attachment revoke */", NULL, NULL, NULL);
if (rc != SQLITE_OK) goto fail;
const char* queries[] = {
"UPDATE attachment_keys SET wanted=0 WHERE channel=?1 AND author=?2 AND key IS NULL AND wanted<>0",
"UPDATE attachment_key_requests SET state=5 WHERE channel=?1 AND peer=?2"
" AND ((incoming=1 AND state IN(0,2)) OR (incoming=0 AND state IN(0,1)))",
"DELETE FROM attachment_key_outbox WHERE channel=?1 AND peer=?2",
"SELECT 1 FROM attachment_keys WHERE channel=?1 AND author=?2 AND key IS NULL AND wanted<>0"
" UNION ALL SELECT 1 FROM attachment_key_requests WHERE channel=?1 AND peer=?2"
" AND ((incoming=1 AND state IN(0,2)) OR (incoming=0 AND state IN(0,1)))"
" UNION ALL SELECT 1 FROM attachment_key_outbox WHERE channel=?1 AND peer=?2 LIMIT 1"
};
int changed[3] = {0};
for (unsigned i = 0; i < 4; i++) {
stage = queries[i]; rc = sqlite3_prepare_v2(db, queries[i], -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
rc = sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC);
if (rc != SQLITE_OK) goto fail;
rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)peer);
if (rc != SQLITE_OK) goto fail;
rc = sqlite3_step(st);
if (rc != SQLITE_DONE) goto fail; /* Последний SELECT обязан быть пуст: silent IGNORE также откатывает отмену. */
if (i < 3) changed[i] = sqlite3_changes(db);
rc = sqlite3_finalize(st); st = NULL;
if (rc != SQLITE_OK) goto fail;
}
stage = "commit revoke"; rc = sqlite3_exec(db, "COMMIT /* attachment revoke */", NULL, NULL, NULL);
if (rc != SQLITE_OK) goto fail;
if (changed[0] || changed[1] || changed[2]) {
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: revoked channel=%s peer=%016llx wanted=%d requests=%d outbox=%d",
channel, (unsigned long long)peer, changed[0], changed[1], changed[2]);
ak_changed(inst, channel);
}
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: revoke channel=%s peer=%016llx stage=%s rc=%d db=%s",
channel, (unsigned long long)peer, stage, rc, sqlite3_errmsg(db));
int final_rc = sqlite3_finalize(st);
if (final_rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: revoke cleanup finalize rc=%d", final_rc);
ak_rollback(inst, "revoke", channel); return -1;
}
/* Снимок пар закрывается до изменений БД; повторный проход безопасен после SQL-отказа или перезапуска. */
int chat_attachment_revalidate_members(struct UTUN_INSTANCE* inst, uint64_t group) {
if (!inst || !CC(inst) || !CC(inst)->attachment_initialized || CC(inst)->attachment_closing) return 0;
sqlite3* db = CC(inst)->db;
if (!sqlite3_get_autocommit(db)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: revoke postponed inside external transaction group=%016llx",
(unsigned long long)group); return -1;
}
char filter[64] = "1", sql[768];
if (group) snprintf(filter, sizeof(filter), "channel='%llu'", (unsigned long long)group);
snprintf(sql, sizeof(sql), "SELECT channel,peer FROM ("
"SELECT channel,peer FROM attachment_key_requests WHERE (incoming=1 AND state IN(0,2)) OR (incoming=0 AND state IN(0,1))"
" UNION SELECT channel,peer FROM attachment_key_outbox"
" UNION SELECT channel,author AS peer FROM attachment_keys WHERE wanted<>0 AND key IS NULL) WHERE %s", filter);
char** pairs = NULL; int rows = 0, columns = 0;
int rc = sqlite3_get_table(db, sql, &pairs, &rows, &columns, NULL);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: revoke scan failed group=%016llx rc=%d db=%s",
(unsigned long long)group, rc, sqlite3_errmsg(db)); sqlite3_free_table(pairs); return -1;
}
int result = 0;
for (int i = 0; i < rows; i++) {
const char* channel = pairs[(i + 1) * columns];
const char* identity = pairs[(i + 1) * columns + 1];
if (!ak_channel(channel) || !identity) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: invalid revoke identity row=%d", i); result = -1; break;
}
uint64_t peer = (uint64_t)strtoll(identity, NULL, 10);
if (!topo_group_chat_allowed(inst, strtoull(channel, NULL, 10), peer) && ak_cancel_peer(inst, channel, peer)) {
result = -1; break;
}
if (CC(inst)->attachment_closing) { result = -1; break; }
}
sqlite3_free_table(pairs); return result;
}
/* 1 — строка существует, 0 — канал/вложение удалены, -1 — SQL-отказ, который нельзя считать удалением. */
static int ak_present(struct ak_file* job) {
sqlite3* db = CC(job->inst)->db; sqlite3_stmt* st = NULL;
@@ -238,6 +321,9 @@ static void ak_finish_file(struct ak_file* job) {
struct UTUN_INSTANCE* inst = job->inst;
int valid = CC(inst)->attachment_closing ? 0 : ak_store_file_result(job);
if (valid < 0) return;
if (valid && job->encrypt && !topo_group_chat_allowed(inst, strtoull(job->channel, NULL, 10), inst->node_id)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: publication revoked channel=%s", job->channel); valid = 0;
}
int success = !job->completion_error && !job->error && valid;
if (success && job->encrypt) {
char name[38], hex[33]; ak_hex(job->id, hex); snprintf(name, sizeof(name), "%s.uenc", hex);
@@ -355,6 +441,9 @@ int chat_attachment_send(struct UTUN_INSTANCE* inst, const char* channel, const
!source || !source[0] || strlen(source) >= 1024 || !reply || encrypted_file_validate_type(type) || attachment_validate(info)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: invalid send arguments/state"); return -1;
}
if (!topo_group_chat_allowed(inst, strtoull(channel, NULL, 10), inst->node_id)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: preparation denied channel=%s", channel); return -1;
}
if (CC(inst)->attachment_file_count >= MEDIA_ASYNC_MAX_TASKS) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: file task limit reached count=%u", CC(inst)->attachment_file_count); return -1;
}
@@ -559,6 +648,9 @@ fail:
/* Неподписанные и незашифрованные ответы сервис не принимает. */
static int ak_send(struct UTUN_INSTANCE* inst, const char* channel, uint64_t peer, const uint8_t* body, size_t length) {
uint64_t group = strtoull(channel, NULL, 10);
if (!topo_group_chat_allowed(inst, group, peer)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: send denied channel=%s peer=%016llx", channel, (unsigned long long)peer); return -1;
}
if (!etcp_router_has_route(inst, group, peer)) return 1;
struct ETCP_ROUTER_CONN* conn = etcp_router_conn_get(inst, group, peer, ETCP_RT_ID_ATTACHMENT_KEYS);
if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: router channel allocation failed"); return -1; }
@@ -605,6 +697,10 @@ fail:
/* Разрешение проверяет полное подписанное членство и immutable историю. */
static int ak_authorized(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32], uint64_t peer) {
uint8_t key[32];
if (!topo_group_chat_allowed(inst, strtoull(channel, NULL, 10), peer)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: author access denied channel=%s peer=%016llx",
channel, (unsigned long long)peer); return 0;
}
if (member_sync_verified_ed25519(inst, channel, peer, key)) return 0;
char sql[256]; snprintf(sql, sizeof(sql), "SELECT 1 FROM msg_%s m JOIN attachment_keys a ON a.message_id=m.message_id"
" AND a.channel=? WHERE m.message_id=? AND m.node_id=? AND a.author=? AND length(a.key)=32", channel);
@@ -672,15 +768,16 @@ static int ak_accept_request(struct UTUN_INSTANCE* inst, const char* channel, ui
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)) {
if (incoming != 1 || (state != 0 && state != 2 && state != 3 && state != 4 && state != CHAT_ATTACHMENT_REQUEST_CANCELLED)) {
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) {
if (!exists || state == CHAT_ATTACHMENT_REQUEST_CANCELLED) {
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);
" VALUES(?,?,?,1,0,unixepoch()) ON CONFLICT(channel,message_id,peer) DO UPDATE SET state=0,created_at=unixepoch()"
" WHERE incoming=1 AND state=5", -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;
@@ -698,6 +795,7 @@ static int ak_accept_request(struct UTUN_INSTANCE* inst, const char* channel, ui
changed, channel, (unsigned long long)peer);
stage = "insert row count"; rc = SQLITE_CONSTRAINT; goto fail;
}
state = 0;
}
int command = state == 0 ? 0 : state == 3 ? AK_DENY : AK_GRANT;
stage = "outbox";
@@ -843,11 +941,17 @@ static void ak_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
uint8_t* d = entry->dgram;
uint64_t peer = 0, group = 0, target = 0;
char channel[32];
if (entry->len == ROUTER_SVC_HDR_SIZE) {
memcpy(&group, d + ROUTER_SVC_GROUP_OFF, 8);
queue_dgram_free(entry); queue_entry_free(entry);
chat_attachment_revalidate_members(inst, group); return;
}
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 || 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);
if (!topo_group_chat_allowed(inst, group, peer)) goto invalid;
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 ||
@@ -1143,6 +1247,10 @@ static void ak_pump(struct UTUN_INSTANCE* inst) {
/* Создать исходящий запрос вместе с очередью; перечитать условия под собственной блокировкой записи. */
static int ak_request_file(struct ak_file* job) {
struct UTUN_INSTANCE* inst = job->inst;
if (!topo_group_chat_allowed(inst, strtoull(job->channel, NULL, 10), job->author)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: request denied channel=%s author=%016llx",
job->channel, (unsigned long long)job->author); return -1;
}
sqlite3* db = CC(inst)->db; sqlite3_stmt* st = NULL;
if (!sqlite3_get_autocommit(db)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: outgoing request rejected inside external transaction channel=%s", job->channel);
@@ -1154,7 +1262,7 @@ static int ak_request_file(struct ak_file* job) {
stage = "prepare eligibility";
rc = sqlite3_prepare_v2(db, "SELECT 1 FROM attachment_keys a WHERE channel=? AND message_id=? AND author=?"
" AND downloaded=1 AND wanted=1 AND ready=0 AND decrypting=0 AND error=0 AND key IS NULL"
" AND NOT EXISTS(SELECT 1 FROM attachment_key_requests r WHERE r.channel=a.channel AND r.message_id=a.message_id)", -1, &st, NULL);
" AND NOT EXISTS(SELECT 1 FROM attachment_key_requests r WHERE r.channel=a.channel AND r.message_id=a.message_id AND r.state<>5)", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
stage = "bind channel"; rc = sqlite3_bind_text(st, 1, job->channel, -1, SQLITE_STATIC);
if (rc != SQLITE_OK) goto fail;
@@ -1168,7 +1276,8 @@ static int ak_request_file(struct ak_file* job) {
if (rc != SQLITE_OK) goto fail;
stage = "prepare request";
rc = sqlite3_prepare_v2(db, "INSERT INTO attachment_key_requests(channel,message_id,peer,incoming,state,created_at)"
" VALUES(?,?,?,0,0,unixepoch())", -1, &st, NULL);
" VALUES(?,?,?,0,0,unixepoch()) ON CONFLICT(channel,message_id,peer) DO UPDATE SET state=0,created_at=unixepoch()"
" WHERE incoming=0 AND state=5", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
stage = "bind channel"; rc = sqlite3_bind_text(st, 1, job->channel, -1, SQLITE_STATIC);
if (rc != SQLITE_OK) goto fail;
@@ -1253,6 +1362,7 @@ static void ak_tick(void* arg) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: pending files/outbox postponed inside external transaction");
goto schedule;
}
if (chat_attachment_revalidate_members(inst, 0)) goto schedule;
for (struct ak_file* job = CC(inst)->attachment_files; job;) {
struct ak_file* next = job->next;
if (job->completed) ak_finish_file(job);
@@ -1263,7 +1373,8 @@ static void ak_tick(void* arg) {
if (CC(inst)->attachment_file_count >= MEDIA_ASYNC_MAX_TASKS) break;
sqlite3_stmt* st = ak_prepare(inst, "SELECT channel,message_id,crypto_id,author,key,cipher_hash,cipher_path,plain_path FROM attachment_keys a"
" WHERE downloaded=1 AND wanted=1 AND ready=0 AND decrypting=0 AND error=0 AND"
" (length(key)=32 OR NOT EXISTS(SELECT 1 FROM attachment_key_requests r WHERE r.channel=a.channel AND r.message_id=a.message_id)) LIMIT 1");
" (length(key)=32 OR NOT EXISTS(SELECT 1 FROM attachment_key_requests r"
" WHERE r.channel=a.channel AND r.message_id=a.message_id AND r.state<>5)) LIMIT 1");
if (!st) break;
int rc = sqlite3_step(st);
if (rc != SQLITE_ROW) {
@@ -1434,9 +1545,16 @@ int chat_attachment_init(struct UTUN_INSTANCE* inst) {
CC(inst)->attachment_closing = 0;
CC(inst)->attachment_initialized = 1;
ak_recover_preparations(inst);
if (etcp_router_bind(inst, ETCP_RT_ID_ATTACHMENT_KEYS, ak_receive)) return -1;
if (etcp_router_bind(inst, ETCP_RT_ID_ATTACHMENT_KEYS, ak_receive) ||
etcp_router_require_crypto(inst, ETCP_RT_ID_ATTACHMENT_KEYS, ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED) ||
etcp_router_set_authorize(inst, ETCP_RT_ID_ATTACHMENT_KEYS, topo_group_chat_allowed)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: router policy initialization failed");
chat_attachment_quiesce(inst); return -1;
}
CC(inst)->attachment_timer = uasync_set_timeout(inst->ua, 10000, inst, ak_tick, "attachment keys");
if (!CC(inst)->attachment_timer) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: initial timer allocation failed"); return -1; }
if (!CC(inst)->attachment_timer) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: initial timer allocation failed"); chat_attachment_quiesce(inst); return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: initialized persistent requests and separate ciphertext/plaintext storage"); return 0;
}
src/chat/chat_attachment.h
+9 -1
View File
@@ -8,7 +8,9 @@
* повтор с тем же результатом допустим, конфликтующий ключ/решение отклоняется. Отказ SQL сохраняет прежнее состояние.
* Приём REQUEST владеет собственной транзакцией проверки идентичности/прав и сохранения запроса/повтора ответа.
* Внешняя транзакция отклоняется без изменений; ACK/событие — после commit. Дубликат сохраняет решение и дату:
* state=0 ждёт автора, 2/4 повторяют GRANT, 3 повторяет DENY; чужое направление и другие состояния отклоняются.
* state=0 ждёт автора, 2/4 повторяют GRANT, 3 повторяет DENY; 5 — отменён при отзыве, новый REQUEST снова требует решения.
* Отзыв self/peer прекращает outbox и ожидающие запросы; полученные ключи и локальная расшифровка сохраняются.
* После readmission исходящий запрос запускается новым want; отменённый запрос не принимает поздний результат.
* Исходящий REQUEST сохраняется собственной транзакцией request+outbox после повторной проверки состояния.
* Уведомление/отправка следуют после commit; внешняя транзакция откладывает весь проход файлов и очереди.
* Перед сетевой попыткой отдельно фиксируется retry_at; отказ SQL не отправляет команду и сохраняет очередь/дату.
@@ -26,6 +28,8 @@ extern "C" {
#endif
struct UTUN_INSTANCE;
#define CHAT_ATTACHMENT_REQUEST_CANCELLED 5 /* Сохранённый запрос отменён; ключ/решение по завершённому запросу не затрагивается. */
// ── Жизненный цикл модуля ──
// Подключить таблицы, router-сервис и обработку сохранённых задач; 0/-1. Внешняя транзакция запрещена.
@@ -37,6 +41,10 @@ void chat_attachment_quiesce(struct UTUN_INSTANCE* inst);
// После workers убрать незавершённые подготовки; открытая транзакция откладывает recovery, сохраняя файлы.
void chat_attachment_destroy(struct UTUN_INSTANCE* inst);
/* После commit допуска: атомарно отменить запрещённую сетевую работу, сохранить локальные ключи/историю.
* group=0 проверяет все сохранённые пары (tick/restart); иначе только указанную группу. 0/-1; SQL-отказ допускает повтор. */
int chat_attachment_revalidate_members(struct UTUN_INSTANCE* inst, uint64_t group);
// ── Отправка и учёт вложений ──
/* Асинхронно отправить подготовленный файл; channel — десятичный ID, reply — 32 байта ID ответа либо нули.
src/routing_layer/topo_group.c
+2
View File
@@ -36,6 +36,7 @@
#include "../chat/group_avatar.h"
#include "../chat/chat_join.h"
#include "../chat/chat_sync.h"
#include "../chat/chat_attachment.h"
#include "../member_files/member_files.h"
#define TOPO_GROUP_UTUN 0x8000000000000000ULL
@@ -451,6 +452,7 @@ void topo_group_revalidate_members(struct TOPO_GROUP* group) {
member_files_revalidate_members(group->instance, group->group_id);
group_avatar_revalidate_members(group->instance, group->group_id);
chat_join_revalidate_members(group->instance, group->group_id);
chat_attachment_revalidate_members(group->instance, group->group_id);
chat_sync_revalidate_join(group->instance, group->group_id);
if (group->stopping) return;
/* Удаление вызывает callbacks и меняет список; каждый раз ищем следующего заново. */