From f3cff571dc66d0ec7a2e442f618b98abd21b47ad Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 21:00:59 +0100 Subject: [PATCH] Authorize attachment key exchange and cancel durable revoked work --- src/chat/chat_attachment.c | 134 +++++++++++++++++++++++++++++++-- src/chat/chat_attachment.h | 10 ++- src/routing_layer/topo_group.c | 2 + 3 files changed, 137 insertions(+), 9 deletions(-) diff --git a/src/chat/chat_attachment.c b/src/chat/chat_attachment.c index fb0c53dd6..b3f133329 100644 --- a/src/chat/chat_attachment.c +++ b/src/chat/chat_attachment.c @@ -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; } diff --git a/src/chat/chat_attachment.h b/src/chat/chat_attachment.h index b487482a1..14313adea 100644 --- a/src/chat/chat_attachment.h +++ b/src/chat/chat_attachment.h @@ -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 ответа либо нули. diff --git a/src/routing_layer/topo_group.c b/src/routing_layer/topo_group.c index f0e1e30fd..c2b5ca27a 100644 --- a/src/routing_layer/topo_group.c +++ b/src/routing_layer/topo_group.c @@ -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 и меняет список; каждый раз ищем следующего заново. */