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

1203 lines
67 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/* dm_media — E2E файлы бесед и отдельное ограниченное хранилище M. */
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/stat.h>
#include <openssl/sha.h>
#include <openssl/rand.h>
#include "../../lib/mem.h"
#include "../../lib/platform_compat.h"
#include "../../lib/debug_config.h"
#include "../../lib/json_flat.h"
#include "dm_media_priv.h"
#include "dm_history.h"
#include "dm_crypto.h"
#include "dm_mailbox.h"
#include "dm_mailbox_media.h"
#include "../utun_instance.h"
#include "../ntp_time.h"
#include "../chat/chat_event.h"
#include "../chat/chat_setting.h"
#include "../routing_layer/etcp_router.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../transport_layer/secure_channel.h"
#include "../transport_layer/etcp.h"
#include "../media_delivery/file_transfer.h"
#include "../media_async/encrypted_file.h"
enum dm_file_work { DM_FILE_PREPARE, DM_FILE_RECEIVE, DM_FILE_CUSTODY, DM_FILE_DELETE };
struct dm_file_job;
struct dm_media_state {
struct UTUN_INSTANCE* inst;
char root[768];
char pending[768];
struct dm_file_job* jobs;
void* timer;
int closing;
int cleanup_active;
};
struct dm_file_job {
struct dm_file_job* next;
struct dm_media_state* media;
enum dm_file_work work;
struct file_transfer_op* transfer;
uint8_t id[16], key[32], hash[32];
uint64_t author, size;
char conv[64];
struct attachment_info info;
int remove_source;
char input[1024], cipher[1024], plain[1024], temp[1024], output[1024];
int error;
uint64_t generation;
unsigned held;
};
sqlite3_stmt* dm_media_prepare(struct UTUN_INSTANCE* inst, const char* sql) {
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: SQL prepare failed: %s statement=%s", sqlite3_errmsg(inst->topo_sqlite_db), sql);
return st;
}
/* Шаг записи всегда завершает statement; SQL-ошибки нельзя скрывать ACK. */
int dm_media_step(struct UTUN_INSTANCE* inst, sqlite3_stmt* st) {
if (!st) return -1;
int rc = sqlite3_step(st);
if (rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: SQL step rc=%d error=%s statement=%s",
rc, sqlite3_errmsg(inst->topo_sqlite_db), sqlite3_sql(st));
sqlite3_finalize(st);
return rc == SQLITE_DONE ? 0 : -1;
}
int dm_media_exec(struct UTUN_INSTANCE* inst, const char* sql) {
int rc = sqlite3_exec(inst->topo_sqlite_db, sql, NULL, NULL, NULL);
if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: SQL rc=%d error=%s statement=%s",
rc, sqlite3_errmsg(inst->topo_sqlite_db), sql);
return rc == SQLITE_OK ? 0 : -1;
}
const uint8_t* dm_message_media(const uint8_t* msg, size_t len) {
size_t actual;
if (!dm_msg_len(msg, len, &actual) || actual != len || msg[32] != 4 || memcmp(msg + 33, "file", 4)) return NULL;
return msg + len - DM_MSG_SIG_SIZE - DM_MEDIA_DESC_SIZE;
}
uint64_t dm_media_cipher_size(uint64_t size) {
return size + (size ? (size + DM_MEDIA_PLAIN_SIZE - 1) / DM_MEDIA_PLAIN_SIZE : 1) * DM_TAG_SIZE;
}
/* Пути зависят только от UUID, имя пользователя не попадает в файловый путь. */
static int dm_file_path(struct dm_media_state* media, int custody, const uint8_t id[16], const char* suffix,
char* path, size_t cap) {
char hex[33];
for (int i = 0; i < 16; i++) snprintf(hex + 2 * i, 3, "%02x", id[i]);
int n = snprintf(path, cap, "%s/%s%s", custody ? media->pending : media->root, hex, suffix);
if (n < 0 || (size_t)n >= cap) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: file path too long"); return -1; }
return 0;
}
/* fflush не гарантирует сохранение: fsync/_commit нужен до публикации и ACK. */
static int dm_file_sync(FILE* file) {
if (fflush(file) != 0) goto fail;
#ifdef _WIN32
if (_commit(_fileno(file)) != 0) goto fail;
#else
if (fsync(fileno(file)) != 0) goto fail;
#endif
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: file sync failed: %s", strerror(errno));
return -1;
}
static int dm_file_remove(const char* path) {
if (remove(path) != 0 && errno != ENOENT) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: remove failed path=%s: %s", path, strerror(errno));
return -1;
}
return 0;
}
/* Манифест хранит только пути приложения. Поколение делает очередь удаления durable. */
int dm_media_file_hold(struct UTUN_INSTANCE* inst, const char* conv, uint64_t generation, const char* path) {
if (!path || !path[0] || !dm_history_current(inst, conv, generation)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM files: hold rejected conv=%s path=%s", conv ? conv : "", path ? path : ""); return -1;
}
sqlite3_stmt* st = dm_media_prepare(inst,
"INSERT INTO dm_local_files(path,conv_id,generation,busy) VALUES(?,?,?,1) ON CONFLICT(path) DO UPDATE SET busy=busy+1"
" WHERE conv_id=excluded.conv_id AND generation=excluded.generation AND busy>=0");
if (!st) return -1;
sqlite3_bind_text(st, 1, path, -1, SQLITE_STATIC); sqlite3_bind_text(st, 2, conv, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 3, (sqlite3_int64)generation);
if (dm_media_step(inst, st)) return -1;
if (sqlite3_changes(inst->topo_sqlite_db) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: path belongs to another task/generation path=%s", path); return -1;
}
DEBUG_TRACE(DEBUG_CATEGORY_DM, "DM files: hold conv=%s generation=%llu path=%s", conv, (unsigned long long)generation, path);
return 0;
}
void dm_media_file_release(struct UTUN_INSTANCE* inst, const char* path) {
sqlite3_stmt* st = dm_media_prepare(inst, "UPDATE dm_local_files SET busy=busy-1 WHERE path=? AND busy>0");
if (!st) return;
sqlite3_bind_text(st, 1, path, -1, SQLITE_STATIC);
if (!dm_media_step(inst, st) && sqlite3_changes(inst->topo_sqlite_db) != 1)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: release without owner path=%s", path);
}
struct dm_cleanup_job { struct dm_media_state* media; char path[1024]; int error; };
static void dm_cleanup_pump(struct dm_media_state* media);
static void dm_cleanup_work(void* arg) {
struct dm_cleanup_job* job = arg;
job->error = dm_file_remove(job->path);
if (!job->error) job->error = file_io_sync_directory(job->path);
}
static void dm_cleanup_done(void* arg, int error) {
struct dm_cleanup_job* job = arg;
struct dm_media_state* media = job->media;
media->cleanup_active = 0;
int success = !error && !job->error;
sqlite3_stmt* st = dm_media_prepare(media->inst, success ? "DELETE FROM dm_local_files WHERE path=? AND busy=-1" :
"UPDATE dm_local_files SET busy=0,retry_at=?2 WHERE path=?1 AND busy=-1");
if (st && !success) sqlite3_bind_int64(st, 2, (sqlite3_int64)(ntp_time_get_seconds(media->inst) + 30));
if (st) { sqlite3_bind_text(st, 1, job->path, -1, SQLITE_STATIC); if (dm_media_step(media->inst, st)) success = 0; }
else success = 0;
if (success) DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM files: removed deleted history path=%s", job->path);
else DEBUG_WARN(DEBUG_CATEGORY_DM, "DM files: cleanup retained for retry path=%s callback=%d worker=%d", job->path, error, job->error);
u_free(job);
if (success && !media->closing) dm_cleanup_pump(media);
}
static void dm_cleanup_pump(struct dm_media_state* media) {
if (media->closing || media->cleanup_active) return;
sqlite3_stmt* st = dm_media_prepare(media->inst,
"SELECT f.path FROM dm_local_files f JOIN dm_history h USING(conv_id)"
" WHERE f.generation<h.generation AND f.busy<=0 AND f.retry_at<=? ORDER BY f.retry_at,f.path LIMIT 1");
if (!st) return;
sqlite3_bind_int64(st, 1, (sqlite3_int64)ntp_time_get_seconds(media->inst));
int rc = sqlite3_step(st);
if (rc != SQLITE_ROW) {
if (rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: cleanup scan failed rc=%d", rc);
sqlite3_finalize(st); return;
}
struct dm_cleanup_job* job = u_calloc(1, sizeof(*job));
if (!job) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: cleanup allocation failed"); sqlite3_finalize(st); return; }
const char* path = (const char*)sqlite3_column_text(st, 0);
if (!path || strlen(path) >= sizeof(job->path)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: invalid cleanup path"); sqlite3_finalize(st); u_free(job); return;
}
job->media = media; snprintf(job->path, sizeof(job->path), "%s", path); sqlite3_finalize(st);
st = dm_media_prepare(media->inst, "UPDATE dm_local_files SET busy=-1 WHERE path=? AND busy<=0");
if (!st) { u_free(job); return; }
sqlite3_bind_text(st, 1, job->path, -1, SQLITE_STATIC);
if (dm_media_step(media->inst, st)) { u_free(job); return; }
media->cleanup_active = 1;
media_async_submit(media->inst->media_async, media->inst->ua, dm_cleanup_work, job, dm_cleanup_done, job);
}
int dm_media_forget_prepare(struct UTUN_INSTANCE* inst, const char* conv, uint64_t generation) {
struct dm_media_state* media = inst->dm_media;
sqlite3_stmt* rows = dm_media_prepare(inst, "SELECT id FROM dm_files WHERE conv_id=?");
if (!rows) return -1;
sqlite3_bind_text(rows, 1, conv, -1, SQLITE_STATIC);
int rc, result = 0;
while ((rc = sqlite3_step(rows)) == SQLITE_ROW) {
if (sqlite3_column_bytes(rows, 0) != 16) { result = -1; break; }
const char* suffixes[] = {".file", ".cipher", ".encrypt", ".download", ".decrypt"};
for (unsigned i = 0; i < sizeof(suffixes) / sizeof(suffixes[0]); i++) {
char path[1024];
if (dm_file_path(media, 0, sqlite3_column_blob(rows, 0), suffixes[i], path, sizeof(path))) { result = -1; break; }
sqlite3_stmt* st = dm_media_prepare(inst,
"INSERT INTO dm_local_files(path,conv_id,generation) VALUES(?,?,?) ON CONFLICT(path) DO NOTHING");
if (!st) { result = -1; break; }
sqlite3_bind_text(st, 1, path, -1, SQLITE_STATIC); sqlite3_bind_text(st, 2, conv, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 3, (sqlite3_int64)generation);
if (dm_media_step(inst, st)) { result = -1; break; }
}
if (result) break;
}
sqlite3_finalize(rows);
if (result || rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: delete manifest failed conv=%s rc=%d", conv, rc); return -1; }
sqlite3_stmt* st = dm_media_prepare(inst, "DELETE FROM dm_files WHERE conv_id=?");
if (!st) return -1;
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
return dm_media_step(inst, st);
}
void dm_media_forget_committed(struct UTUN_INSTANCE* inst, const char* conv) {
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT media_id FROM dm_deleted_messages WHERE conv_id=? AND media_id IS NOT NULL");
if (st) {
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
int rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
if (sqlite3_column_bytes(st, 0) == 16) file_transfer_cancel_object(inst->file_transfer, sqlite3_column_blob(st, 0));
else DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: damaged deleted media ID conv=%s", conv);
}
if (rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: cancellation scan failed conv=%s rc=%d", conv, rc);
sqlite3_finalize(st);
}
dm_cleanup_pump(inst->dm_media);
}
/* Подписанная роль и живые мемберы в одной области; тип NAT не даёт полномочий. */
int dm_media_scope(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t node,
const char* role, uint64_t author, uint64_t recipient) {
if (!group) return 0;
char sql[256];
snprintf(sql, sizeof(sql), "SELECT node_id,adm_tags FROM \"peers_%llu\" WHERE deleted=0 AND node_id IN (?,?,?)",
(unsigned long long)group);
sqlite3_stmt* st = dm_media_prepare(inst, sql);
if (!st) return 0;
sqlite3_bind_int64(st, 1, (sqlite3_int64)node);
sqlite3_bind_int64(st, 2, (sqlite3_int64)author);
sqlite3_bind_int64(st, 3, (sqlite3_int64)recipient);
int has_a = 0, has_b = 0, allowed = 0, rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
uint64_t id = (uint64_t)sqlite3_column_int64(st, 0);
if (id == author) has_a = 1;
if (id == recipient) has_b = 1;
if (id == node) {
const char* tags = (const char*)sqlite3_column_text(st, 1);
char value[16];
allowed = !role || (tags && json_flat_get(tags, role, value, sizeof(value)) == 0 && !strcmp(value, "yes"));
}
}
if (rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: role scan failed rc=%d", rc);
sqlite3_finalize(st);
return rc == SQLITE_DONE && allowed && has_a && has_b;
}
void dm_media_notify(struct UTUN_INSTANCE* inst, uint64_t conv) {
char id[64];
int n = snprintf(id, sizeof(id), "%llu", (unsigned long long)conv);
uint8_t evt[1 + 64 + 8];
evt[0] = (uint8_t)n;
memcpy(evt + 1, id, (size_t)n);
memcpy(evt + 1 + n, &inst->node_id, 8);
chat_event_post(inst, CHAT_EVT_DM_MSG_RECEIVED, evt, n + 9);
}
/* Квитанция подписывает UUID и хеш всего сообщения с descriptor. */
int dm_media_verify_receipt(struct UTUN_INSTANCE* inst, const uint8_t* r, const uint8_t* msg, size_t len) {
uint64_t conv, author, recipient, seq;
if (!inst || !r || memcmp(r, "DMFILE01", 8)) goto invalid;
memcpy(&conv, r + 8, 8); memcpy(&seq, r + 16, 8);
memcpy(&author, r + 24, 8); memcpy(&recipient, r + 32, 8);
if (!author || !recipient || author == recipient || !seq || seq > INT64_MAX ||
conv != dm_derive_conv_id(author, recipient)) goto invalid;
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT ed25519_pubkey FROM nodes WHERE node_id=?");
if (!st) return -1;
sqlite3_bind_int64(st, 1, (sqlite3_int64)recipient);
uint8_t key[32];
int found = sqlite3_step(st) == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32;
if (found) memcpy(key, sqlite3_column_blob(st, 0), sizeof(key));
sqlite3_finalize(st);
if (!found || sc_ed25519_verify(key, r, 88, r + 88) != SC_OK) goto invalid;
if (msg) {
const uint8_t* desc = dm_message_media(msg, len);
uint8_t hash[32];
if (!desc || memcmp(r + 8, msg, 16) || memcmp(r + 24, msg + 24, 8) || memcmp(r + 40, desc, 16)) goto invalid;
SHA256(msg, len, hash);
if (memcmp(hash, r + 56, 32)) goto invalid;
}
return 0;
invalid:
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: rejected media receipt body_bytes=%zu", len);
return -1;
}
/* Транзакция dm_core связывает историю, очередь и состояние файла. */
int dm_media_track(struct UTUN_INSTANCE* inst, int dir, const uint8_t* msg, size_t len) {
const uint8_t* desc = dm_message_media(msg, len);
if (!desc) return 0;
struct dm_media_state* media = inst->dm_media;
if (!media || media->closing) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: file received while owner stopped"); return -1; }
uint64_t conv, seq, author, size, recipient, group;
memcpy(&conv, msg, 8); memcpy(&seq, msg + 8, 8); memcpy(&author, msg + 24, 8); memcpy(&size, desc + 16, 8);
char id[64], path[1024];
snprintf(id, sizeof(id), "%llu", (unsigned long long)conv);
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT peer_node_id,group_id FROM dm_conversations WHERE conv_id=?");
if (!st) return -1;
sqlite3_bind_text(st, 1, id, -1, SQLITE_STATIC);
int rc = sqlite3_step(st);
if (rc != SQLITE_ROW) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: conversation absent conv=%s rc=%d", id, rc); sqlite3_finalize(st); return -1; }
recipient = dir ? (uint64_t)sqlite3_column_int64(st, 0) : inst->node_id;
group = (uint64_t)sqlite3_column_int64(st, 1);
sqlite3_finalize(st);
if (dm_file_path(media, 0, desc, ".file", path, sizeof(path))) return -1;
if (dir) {
struct stat sb;
if (stat(path, &sb) || (uint64_t)sb.st_size != size) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: outgoing file not prepared path=%s size=%llu", path, (unsigned long long)size);
return -1;
}
}
st = dm_media_prepare(inst, "INSERT INTO dm_files(id,conv_id,dir,seq,author,recipient,plain_size,cipher_hash,body,group_id,"
"local_path,ready,holder) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO NOTHING");
if (!st) return -1;
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC);
sqlite3_bind_text(st, 2, id, -1, SQLITE_STATIC); sqlite3_bind_int(st, 3, dir);
sqlite3_bind_int64(st, 4, (sqlite3_int64)seq); sqlite3_bind_int64(st, 5, (sqlite3_int64)author);
sqlite3_bind_int64(st, 6, (sqlite3_int64)recipient); sqlite3_bind_int64(st, 7, (sqlite3_int64)size);
sqlite3_bind_blob(st, 8, desc + 24, 32, SQLITE_STATIC); sqlite3_bind_blob(st, 9, msg, (int)len, SQLITE_STATIC);
sqlite3_bind_int64(st, 10, (sqlite3_int64)group); sqlite3_bind_text(st, 11, path, -1, SQLITE_STATIC);
sqlite3_bind_int(st, 12, dir ? 1 : 0); sqlite3_bind_int64(st, 13, (sqlite3_int64)author);
if (dm_media_step(inst, st)) return -1;
st = dm_media_prepare(inst, "SELECT body FROM dm_files WHERE id=?");
if (!st) return -1;
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC);
rc = sqlite3_step(st);
int exact = rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == (int)len && !memcmp(sqlite3_column_blob(st, 0), msg, len);
sqlite3_finalize(st);
if (!exact) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: conflicting UUID/body conv=%s seq=%llu", id, (unsigned long long)seq);
return exact ? 0 : -1;
}
/* Квитанция endpoint удаляет только media pending, история/собственный файл остаются. */
int dm_media_accept_receipt(struct UTUN_INSTANCE* inst, const uint8_t* r) {
if (dm_media_verify_receipt(inst, r, NULL, 0)) return -1;
uint64_t author;
memcpy(&author, r + 24, 8);
if (author != inst->node_id) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: receipt for another author"); return -1; }
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body,receipt FROM dm_files WHERE id=? AND dir=1");
if (!st) return -1;
sqlite3_bind_blob(st, 1, r + 40, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
if (rc == SQLITE_DONE) {
sqlite3_finalize(st);
st = dm_media_prepare(inst,
"SELECT body_hash FROM dm_deleted_messages WHERE conv_id=? AND dir=1 AND seq=? AND media_id=?");
if (!st) return -1;
uint64_t conv, seq; char id[64]; memcpy(&conv, r + 8, 8); memcpy(&seq, r + 16, 8);
snprintf(id, sizeof(id), "%llu", (unsigned long long)conv);
sqlite3_bind_text(st, 1, id, -1, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
sqlite3_bind_blob(st, 3, r + 40, 16, SQLITE_STATIC);
rc = sqlite3_step(st);
int exact = rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32 && !memcmp(sqlite3_column_blob(st, 0), r + 56, 32);
sqlite3_finalize(st);
if (exact) { DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM files: late receipt for deleted file conv=%s", id); return 0; }
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM files: receipt for unknown/conflicting deleted file conv=%s", id); return -1;
}
if (rc != SQLITE_ROW || dm_media_verify_receipt(inst, r, sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0))) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: receipt does not match outgoing artifact rc=%d", rc);
sqlite3_finalize(st);
return -1;
}
int already = sqlite3_column_bytes(st, 1) == DM_MEDIA_RECEIPT_SIZE;
sqlite3_finalize(st);
st = dm_media_prepare(inst, "UPDATE dm_files SET receipt=? WHERE id=? AND dir=1");
if (!st) return -1;
sqlite3_bind_blob(st, 1, r, DM_MEDIA_RECEIPT_SIZE, SQLITE_STATIC); sqlite3_bind_blob(st, 2, r + 40, 16, SQLITE_STATIC);
if (dm_media_step(inst, st)) return -1;
if (!already) {
uint64_t conv;
memcpy(&conv, r + 8, 8);
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: recipient confirmed file conv=%llu", (unsigned long long)conv);
dm_media_notify(inst, conv);
}
return 0;
}
static void dm_media_control(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source,
uint8_t cmd, const uint8_t* body, size_t len);
/* Local S/M тоже используют тот же проверяемый обработчик, без искусственного CM к себе. */
int dm_media_send(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer,
uint8_t cmd, const uint8_t* body, size_t len) {
if (len > DM_MEDIA_BODY_MAX + 24 || !inst->dm_media || inst->dm_media->closing) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: invalid/stopped control send bytes=%zu", len);
return -1;
}
if (peer == inst->node_id) { dm_media_control(inst, group, peer, cmd, body, len); return 0; }
struct ETCP_ROUTER_CONN* rc = etcp_router_conn_get(inst, group, peer, ETCP_RT_ID_DM_MEDIA);
if (!rc) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: router channel allocation failed"); return -1; }
if (!etcp_router_send_q_has_room(inst, group, peer, ETCP_RT_ID_DM_MEDIA)) return -1;
uint8_t packet[1 + DM_MEDIA_BODY_MAX + 24];
packet[0] = cmd;
memcpy(packet + 1, body, len);
int error = etcp_router_conn_send(rc, packet, len + 1, ROUTER_FLAG_SIGNED);
if (error) DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: control send failed peer=%llu cmd=%u rc=%d",
(unsigned long long)peer, cmd, error);
return error;
}
static struct dm_file_job* dm_job_find(struct dm_media_state* media, const uint8_t id[16], int custody) {
for (struct dm_file_job* job = media->jobs; job; job = job->next)
if ((job->work == DM_FILE_CUSTODY || job->work == DM_FILE_DELETE) == custody && !memcmp(job->id, id, 16)) return job;
return NULL;
}
static int dm_job_paths(struct dm_file_job* job, const char** paths) {
if (job->work != DM_FILE_PREPARE && job->work != DM_FILE_RECEIVE) return 0;
paths[0] = job->cipher; paths[1] = job->temp;
paths[2] = job->work == DM_FILE_PREPARE ? job->plain : job->output;
if (job->work == DM_FILE_PREPARE && job->remove_source) { paths[3] = job->input; return 4; }
return 3;
}
static int dm_job_hold(struct dm_file_job* job) {
const char* paths[4]; int count = dm_job_paths(job, paths);
for (int i = 0; i < count; i++) {
if (dm_media_file_hold(job->media->inst, job->conv, job->generation, paths[i])) return -1;
job->held++;
}
return 0;
}
static void dm_job_free(struct dm_file_job* job) {
struct dm_file_job** p = &job->media->jobs;
while (*p && *p != job) p = &(*p)->next;
if (*p) *p = job->next;
const char* paths[4]; dm_job_paths(job, paths);
for (unsigned i = 0; i < job->held; i++) dm_media_file_release(job->media->inst, paths[i]);
OPENSSL_cleanse(job->key, sizeof(job->key));
u_free(job);
}
/* Копирование своего source фиксирует один снимок до шифрования. */
static int dm_file_copy(const char* source, const char* dest) {
FILE* in = fopen(source, "rb");
FILE* out = in ? fopen(dest, "wb") : NULL;
int error = -1;
if (!in || !out) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: copy open failed %s -> %s: %s", source, dest, strerror(errno)); goto done; }
uint8_t buf[32768];
uint64_t size = 0;
size_t n;
while ((n = fread(buf, 1, sizeof(buf), in))) {
size += n;
if (size > DM_MEDIA_MAX_SIZE || fwrite(buf, 1, n, out) != n) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: copy size/write failed bytes=%llu", (unsigned long long)size);
goto done;
}
}
if (ferror(in)) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: source read failed: %s", strerror(errno)); goto done; }
error = dm_file_sync(out);
done:
if (in && fclose(in)) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: source close failed"); error = -1; }
if (out && fclose(out)) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: copy close failed"); error = -1; }
return error;
}
/* Worker не обращается к БД/сети/контексту instance. */
static void dm_file_work(void* arg) {
struct dm_file_job* job = arg;
job->error = -1;
if (job->work == DM_FILE_DELETE) {
if (!dm_file_remove(job->cipher) && !dm_file_remove(job->temp)) job->error = file_io_sync_directory(job->cipher);
return;
}
if (job->work == DM_FILE_PREPARE) {
int copied = dm_file_copy(job->input, job->plain);
if (job->remove_source && dm_file_remove(job->input)) return;
if (copied) return;
}
FILE* in = fopen(job->work == DM_FILE_PREPARE ? job->plain : job->cipher, "rb");
FILE* out = job->work == DM_FILE_CUSTODY ? NULL : fopen(job->temp, "wb");
if (!in || (job->work != DM_FILE_CUSTODY && !out)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: worker open failed: %s", strerror(errno));
goto done;
}
if (job->work == DM_FILE_PREPARE) {
job->error = media_encrypt_stream(job->key, job->author, job->id, in, out, &job->size, job->hash);
} else if (job->work == DM_FILE_RECEIVE) {
job->error = media_decrypt_stream(job->key, job->author, job->id, job->size, job->hash, in, out);
} else {
SHA256_CTX hash;
SHA256_Init(&hash);
uint8_t buf[32768], digest[32];
uint64_t size = 0;
size_t n;
while ((n = fread(buf, 1, sizeof(buf), in))) { size += n; SHA256_Update(&hash, buf, n); }
SHA256_Final(digest, &hash);
if (ferror(in) || size != dm_media_cipher_size(job->size) || memcmp(digest, job->hash, 32)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: custody hash/size/read failed bytes=%llu expected=%llu",
(unsigned long long)size, (unsigned long long)dm_media_cipher_size(job->size));
} else job->error = 0;
if (!job->error) {
FILE* durable = fopen(job->cipher, "rb+");
if (!durable) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: custody sync open failed"); job->error = -1; }
else { job->error = dm_file_sync(durable); if (fclose(durable)) job->error = -1; }
}
}
if (!job->error && out) job->error = dm_file_sync(out);
done:
if (in && fclose(in)) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: worker input close failed"); job->error = -1; }
if (out && fclose(out)) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: worker output close failed"); job->error = -1; }
if (!job->error) {
if (job->work == DM_FILE_PREPARE) {
job->error = file_io_publish(job->temp, job->cipher);
if (!job->error) job->error = file_io_sync_directory(job->plain);
} else job->error = file_io_publish(job->work == DM_FILE_CUSTODY ? job->cipher : job->temp, job->output);
}
}
static void dm_custody_finalize(struct dm_media_state* media, const uint8_t id[16]);
/* Только uasync публикует проверенный файл и изменяет durable состояние. */
static void dm_file_done(void* arg, int error) {
struct dm_file_job* job = arg;
struct dm_media_state* media = job->media;
struct UTUN_INSTANCE* inst = media->inst;
if (job->work == DM_FILE_DELETE) {
if (!error && !job->error && !media->closing) dm_custody_finalize(media, job->id);
else DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: custody deletion worker failed/cancelled error=%d worker=%d", error, job->error);
dm_job_free(job);
return;
}
int custody = job->work == DM_FILE_CUSTODY;
if (error || job->error || media->closing) goto fail;
if (!custody && !dm_history_current(inst, job->conv, job->generation)) {
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM files: discard deleted preparation conv=%s generation=%llu work=%d", job->conv,
(unsigned long long)job->generation, job->work); goto fail;
}
if (job->work == DM_FILE_PREPARE) {
uint8_t data[DM_MEDIA_DESC_SIZE + ATTACHMENT_WIRE_MAX];
memcpy(data, job->id, 16); memcpy(data + 16, &job->size, 8); memcpy(data + 24, job->hash, 32);
int n = attachment_encode(&job->info, data + DM_MEDIA_DESC_SIZE, ATTACHMENT_WIRE_MAX);
if (n < 0) goto fail;
if (dm_send(inst, job->conv, "file", data, (uint32_t)(DM_MEDIA_DESC_SIZE + n))) goto fail;
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: prepared conv=%s bytes=%llu name=%s", job->conv,
(unsigned long long)job->size, job->info.name);
chat_event_attachment(inst, 1, job->conv, job->id, 1);
dm_job_free(job);
return;
}
const char* sql = custody ? "SELECT body,supernode,group_id,ready,expires FROM dm_custody WHERE id=?" :
"SELECT body,author,group_id,ready,0 FROM dm_files WHERE id=? AND dir=0";
sqlite3_stmt* st = dm_media_prepare(inst, sql);
if (!st) goto fail;
sqlite3_bind_blob(st, 1, job->id, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
uint8_t body[DM_MEDIA_BODY_MAX];
size_t len = rc == SQLITE_ROW ? (size_t)sqlite3_column_bytes(st, 0) : 0;
uint64_t peer = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 1) : 0;
uint64_t group = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 2) : 0;
int ready = rc == SQLITE_ROW ? sqlite3_column_int(st, 3) : -1;
uint64_t expires = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 4) : 0;
if (len <= sizeof(body) && len) memcpy(body, sqlite3_column_blob(st, 0), len);
sqlite3_finalize(st);
if (ready < 0 || !len || len > sizeof(body) || (custody && expires <= ntp_time_get_seconds(inst))) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: completed task no longer admissible custody=%d ready=%d expiry=%llu",
custody, ready, (unsigned long long)expires);
goto fail;
}
uint8_t receipt[DM_MEDIA_RECEIPT_SIZE];
if (!custody) {
memcpy(receipt, "DMFILE01", 8); memcpy(receipt + 8, body, 16);
memcpy(receipt + 24, body + 24, 8); memcpy(receipt + 32, &inst->node_id, 8);
memcpy(receipt + 40, job->id, 16); SHA256(body, len, receipt + 56);
if (sc_ed25519_sign(inst->my_ed25519_privkey, receipt, 88, receipt + 88) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: media receipt signing failed");
goto fail;
}
}
st = dm_media_prepare(inst, custody ? "UPDATE dm_custody SET ready=1 WHERE id=? AND ready=0" :
"UPDATE dm_files SET ready=1,downloading=0,receipt=? WHERE id=? AND ready<=0");
if (!st) goto fail;
int idx = 1;
if (!custody) sqlite3_bind_blob(st, idx++, receipt, sizeof(receipt), SQLITE_STATIC);
sqlite3_bind_blob(st, idx, job->id, 16, SQLITE_STATIC);
if (dm_media_step(inst, st) || sqlite3_changes(inst->topo_sqlite_db) != 1) goto fail;
if (custody) {
uint8_t stored[24]; memcpy(stored, job->id, 16); memcpy(stored + 16, &expires, 8);
dm_media_send(inst, group, peer, DM_MEDIA_STORED, stored, sizeof(stored));
} else {
dm_file_remove(job->cipher);
dm_media_send(inst, group, peer, DM_MEDIA_RECEIPT, receipt, sizeof(receipt));
dm_mailbox_media_receipt(inst, receipt);
uint64_t conv; memcpy(&conv, body, 8); dm_media_notify(inst, conv);
}
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: file committed custody=%d peer=%llu bytes=%llu expires=%llu", custody,
(unsigned long long)peer, (unsigned long long)job->size, (unsigned long long)expires);
dm_job_free(job);
return;
fail:
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: file task failed/cancelled work=%d error=%d worker=%d closing=%d",
job->work, error, job->error, media->closing);
if (job->work == DM_FILE_PREPARE) {
chat_event_attachment(inst, 1, job->conv, job->id, 2);
dm_file_remove(job->plain); dm_file_remove(job->cipher);
if (job->remove_source) dm_file_remove(job->input);
}
else if (!custody) {
sqlite3_stmt* st = dm_media_prepare(inst, "UPDATE dm_files SET downloading=0 WHERE id=? AND dir=0");
if (st) { sqlite3_bind_blob(st, 1, job->id, 16, SQLITE_STATIC); dm_media_step(inst, st); }
}
dm_file_remove(job->temp);
if (job->output[0]) dm_file_remove(job->output);
if (job->work != DM_FILE_PREPARE) dm_file_remove(job->cipher);
dm_job_free(job);
}
static void dm_file_received(void* arg, int error) {
struct dm_file_job* job = arg;
job->transfer = NULL;
if (error || job->media->closing) { dm_file_done(job, error ? error : -2); return; }
media_async_submit(job->media->inst->media_async, job->media->inst->ua, dm_file_work, job, dm_file_done, job);
}
/* Получать файл может endpoint или M; обе операции владеют одним CM handle в transfer. */
static void dm_file_pull(struct dm_media_state* media, int custody, uint64_t group, uint64_t peer, const uint8_t* desc,
uint64_t author) {
if (dm_job_find(media, desc, custody)) return;
struct dm_file_job* job = u_calloc(1, sizeof(*job));
if (!job) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: download allocation failed"); return; }
job->media = media; job->work = custody ? DM_FILE_CUSTODY : DM_FILE_RECEIVE; job->author = author;
memcpy(job->id, desc, 16); memcpy(&job->size, desc + 16, 8); memcpy(job->hash, desc + 24, 32);
if (dm_file_path(media, custody, job->id, ".download", job->cipher, sizeof(job->cipher)) ||
dm_file_path(media, custody, job->id, ".decrypt", job->temp, sizeof(job->temp)) ||
dm_file_path(media, custody, job->id, custody ? ".cipher" : ".file", job->output, sizeof(job->output))) { u_free(job); return; }
if (!custody) {
snprintf(job->conv, sizeof(job->conv), "%llu", (unsigned long long)dm_derive_conv_id(media->inst->node_id, author));
job->generation = dm_history_generation(media->inst, job->conv);
if (dm_job_hold(job)) { dm_job_free(job); return; }
sqlite3_stmt* st = dm_media_prepare(media->inst, "SELECT x25519_pubkey FROM nodes WHERE node_id=?");
if (!st) { dm_job_free(job); return; }
sqlite3_bind_int64(st, 1, (sqlite3_int64)author);
int rc = sqlite3_step(st);
uint8_t pub[32];
int found = rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == sizeof(pub);
if (found) memcpy(pub, sqlite3_column_blob(st, 0), sizeof(pub));
sqlite3_finalize(st);
if (!found || dm_derive_content_key(media->inst->my_keys.private_key, pub, job->key)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: target key unavailable author=%llu", (unsigned long long)author);
dm_job_free(job);
return;
}
}
job->next = media->jobs; media->jobs = job;
job->transfer = file_transfer_pull(media->inst->file_transfer, group, peer, job->id, dm_media_cipher_size(job->size),
job->cipher, dm_file_received, job);
if (!job->transfer) { dm_file_done(job, -1); return; }
if (!custody) {
sqlite3_stmt* st = dm_media_prepare(media->inst, "UPDATE dm_files SET downloading=1 WHERE id=? AND dir=0");
if (st) { sqlite3_bind_blob(st, 1, job->id, 16, SQLITE_STATIC); dm_media_step(media->inst, st); }
}
}
/* Общий авторизатор сервера transfer: A отдаёт B/storage, M — только B по custody. */
static int dm_file_lookup(void* arg, uint64_t group, uint64_t peer, const uint8_t id[16], uint64_t size,
char* path, size_t cap) {
struct dm_media_state* media = arg;
struct UTUN_INSTANCE* inst = media->inst;
if (media->closing) return -1;
for (int custody = 0; custody <= 1; custody++) {
sqlite3_stmt* st = dm_media_prepare(inst, custody ?
"SELECT author,recipient,plain_size,ready,expires FROM dm_custody WHERE id=?" :
"SELECT author,recipient,plain_size,ready,0 FROM dm_files WHERE id=? AND dir=1");
if (!st) return -1;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
uint64_t author = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 0) : 0;
uint64_t recipient = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 1) : 0;
uint64_t plain = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 2) : 0;
int ready = rc == SQLITE_ROW ? sqlite3_column_int(st, 3) : 0;
uint64_t expiry = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 4) : 0;
sqlite3_finalize(st);
if (!author || ready != 1 || size != dm_media_cipher_size(plain) ||
(custody && expiry <= ntp_time_get_seconds(inst))) continue;
if ((peer == recipient && dm_media_scope(inst, group, peer, NULL, author, recipient)) ||
(!custody && dm_media_scope(inst, group, peer, "storage", author, recipient)))
return dm_file_path(media, custody, id, ".cipher", path, cap);
}
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: file serve denied peer=%llu bytes=%llu", (unsigned long long)peer, (unsigned long long)size);
return -1;
}
/* Путь исходника копируется; caller может освободить строку после возврата. */
int dm_send_file(struct UTUN_INSTANCE* inst, const char* conv, const char* path, const struct attachment_info* info,
int remove_source) {
struct dm_media_state* media = inst ? inst->dm_media : NULL;
if (!media || media->closing || !inst->media_async || !conv || !path || attachment_validate(info) || strlen(path) >= 1024) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: invalid file send arguments/state");
return -1;
}
unsigned preparing = 0;
for (struct dm_file_job* job = media->jobs; job; job = job->next) if (job->work == DM_FILE_PREPARE) preparing++;
if (preparing >= 4) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: preparation limit reached count=%u", preparing); return -1; }
struct dm_file_job* job = u_calloc(1, sizeof(*job));
if (!job) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: preparation allocation failed"); return -1; }
job->media = media; job->work = DM_FILE_PREPARE; job->author = inst->node_id;
job->generation = dm_history_generation(inst, conv);
if (!dm_history_current(inst, conv, job->generation)) { u_free(job); return -1; }
snprintf(job->conv, sizeof(job->conv), "%s", conv); job->info = *info; job->remove_source = remove_source;
snprintf(job->input, sizeof(job->input), "%s", path);
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT peer_x25519 FROM dm_conversations WHERE conv_id=?");
if (!st) { u_free(job); return -1; }
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
uint8_t pub[32];
int rc = sqlite3_step(st), found = rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == sizeof(pub);
if (found) memcpy(pub, sqlite3_column_blob(st, 0), sizeof(pub));
sqlite3_finalize(st);
if (!found || dm_derive_content_key(inst->my_keys.private_key, pub, job->key) || RAND_bytes(job->id, sizeof(job->id)) != 1 ||
dm_file_path(media, 0, job->id, ".file", job->plain, sizeof(job->plain)) ||
dm_file_path(media, 0, job->id, ".cipher", job->cipher, sizeof(job->cipher)) ||
dm_file_path(media, 0, job->id, ".encrypt", job->temp, sizeof(job->temp))) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: preparation keys/paths failed conv=%s", conv);
OPENSSL_cleanse(job->key, sizeof(job->key)); u_free(job); return -1;
}
if (dm_job_hold(job)) { dm_job_free(job); return -1; }
job->next = media->jobs; media->jobs = job;
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: preparing conv=%s kind=%d name=%s", conv, info->kind, info->name);
chat_event_attachment(inst, 1, conv, job->id, 0);
media_async_submit(inst->media_async, inst->ua, dm_file_work, job, dm_file_done, job);
return 0;
}
/* Удаление непринятого временного файла тоже выполняется вне uasync. */
static void dm_source_cleanup(void* arg) {
struct dm_file_req* req = arg;
dm_file_remove(req->path);
}
static void dm_source_cleaned(void* arg, int error) {
if (error) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: source cleanup cancelled error=%d", error);
dm_source_cleanup(arg);
}
u_free(arg);
}
void dm_send_file_trampoline(void* arg) {
struct dm_file_req* req = arg;
if (!req) return;
uint64_t generation = req->generation ? req->generation : dm_history_generation(req->inst, req->conv);
int current = dm_history_current(req->inst, req->conv, generation);
if (!current || dm_send_file(req->inst, req->conv, req->path, &req->info, req->remove_source)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM files: send rejected conv=%s generation=%llu", req->conv, (unsigned long long)generation);
const uint8_t rejected[16] = {0};
if (current) chat_event_attachment(req->inst, 1, req->conv, rejected, 2);
if (req->remove_source) {
media_async_submit(req->inst ? req->inst->media_async : NULL, req->inst ? req->inst->ua : NULL,
dm_source_cleanup, req, dm_source_cleaned, req);
return;
}
}
u_free(req);
}
/* STORE сначала резервирует место и фиксирует expiry; повтор не продлевает TTL. */
static void dm_custody_store(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source, const uint8_t* p, size_t len) {
if (len < 8) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: short STORE"); return; }
uint64_t recipient, author, size;
memcpy(&recipient, p, 8); p += 8; len -= 8;
const uint8_t* desc = dm_message_media(p, len);
if (!desc || dm_verify_message(inst, recipient, p, len)) return;
memcpy(&author, p + 24, 8); memcpy(&size, desc + 16, 8);
if (!dm_media_scope(inst, group, source, "supernode", author, recipient) ||
!dm_media_scope(inst, group, inst->node_id, "storage", author, recipient)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: unauthorized STORE source=%llu group=%llu",
(unsigned long long)source, (unsigned long long)group);
return;
}
uint8_t hash[32]; SHA256(p, len, hash);
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body_hash,receipt FROM dm_custody_done WHERE id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
if (rc == SQLITE_ROW) {
int exact = sqlite3_column_bytes(st, 0) == 32 && !memcmp(sqlite3_column_blob(st, 0), hash, 32);
uint8_t receipt[DM_MEDIA_RECEIPT_SIZE];
int delivered = sqlite3_column_bytes(st, 1) == sizeof(receipt);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 1), sizeof(receipt));
sqlite3_finalize(st);
if (!exact) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: STORE conflicts with retired UUID"); return; }
dm_media_send(inst, group, source, delivered ? DM_MEDIA_RECEIPT : DM_MEDIA_EXPIRED,
delivered ? receipt : desc, delivered ? sizeof(receipt) : 16);
return;
}
sqlite3_finalize(st);
if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: tombstone lookup failed rc=%d", rc); return; }
st = dm_media_prepare(inst, "SELECT body,ready,expires FROM dm_custody WHERE id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC);
rc = sqlite3_step(st);
if (rc == SQLITE_ROW) {
int exact = sqlite3_column_bytes(st, 0) == (int)len && !memcmp(sqlite3_column_blob(st, 0), p, len);
int ready = sqlite3_column_int(st, 1);
uint64_t expiry = (uint64_t)sqlite3_column_int64(st, 2);
sqlite3_finalize(st);
if (!exact) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: conflicting custody body"); return; }
if (ready == 1 && expiry > ntp_time_get_seconds(inst)) {
uint8_t stored[24]; memcpy(stored, desc, 16); memcpy(stored + 16, &expiry, 8);
dm_media_send(inst, group, source, DM_MEDIA_STORED, stored, sizeof(stored));
}
return;
}
sqlite3_finalize(st);
if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: custody lookup failed rc=%d", rc); return; }
st = dm_media_prepare(inst, "SELECT COALESCE(sum(cipher_size),0) FROM dm_custody");
if (!st) return;
rc = sqlite3_step(st);
uint64_t used = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 0) : UINT64_MAX;
sqlite3_finalize(st);
uint64_t quota = (uint64_t)chat_setting_get_int(inst, "dm_media_storage_mb", 1024) * 1024 * 1024;
uint64_t cipher_size = dm_media_cipher_size(size);
if (used > quota || cipher_size > quota - used) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: custody quota refused used=%llu requested=%llu quota=%llu",
(unsigned long long)used, (unsigned long long)cipher_size, (unsigned long long)quota);
uint8_t reject[17]; memcpy(reject, desc, 16); reject[16] = 1;
dm_media_send(inst, group, source, DM_MEDIA_REJECT, reject, sizeof(reject));
return;
}
uint64_t expires = ntp_time_get_seconds(inst) + (uint64_t)chat_setting_get_int(inst, "dm_media_ttl_sec", 604800);
st = dm_media_prepare(inst, "INSERT INTO dm_custody(id,author,recipient,plain_size,cipher_size,body,body_hash,"
"group_id,supernode,expires) VALUES(?,?,?,?,?,?,?,?,?,?)");
if (!st) return;
sqlite3_bind_blob(st, 1, desc, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)author);
sqlite3_bind_int64(st, 3, (sqlite3_int64)recipient); sqlite3_bind_int64(st, 4, (sqlite3_int64)size);
sqlite3_bind_int64(st, 5, (sqlite3_int64)cipher_size); sqlite3_bind_blob(st, 6, p, (int)len, SQLITE_STATIC);
sqlite3_bind_blob(st, 7, hash, 32, SQLITE_STATIC); sqlite3_bind_int64(st, 8, (sqlite3_int64)group);
sqlite3_bind_int64(st, 9, (sqlite3_int64)source); sqlite3_bind_int64(st, 10, (sqlite3_int64)expires);
if (!dm_media_step(inst, st))
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: custody reserved author=%llu recipient=%llu bytes=%llu expires=%llu",
(unsigned long long)author, (unsigned long long)recipient, (unsigned long long)cipher_size, (unsigned long long)expires);
}
/* Удаление подтверждается только после durable удаления файла и копии из БД. */
static void dm_custody_finalize(struct dm_media_state* media, const uint8_t id[16]) {
struct UTUN_INSTANCE* inst = media->inst;
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body_hash,receipt,supernode,group_id FROM dm_custody WHERE id=? AND ready=-1");
if (!st) return;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
uint8_t hash[32], receipt[DM_MEDIA_RECEIPT_SIZE];
if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != sizeof(hash)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: delete state missing/corrupt rc=%d", rc);
sqlite3_finalize(st); return;
}
memcpy(hash, sqlite3_column_blob(st, 0), sizeof(hash));
int delivered = sqlite3_column_bytes(st, 1) == sizeof(receipt);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 1), sizeof(receipt));
uint64_t super = (uint64_t)sqlite3_column_int64(st, 2), group = (uint64_t)sqlite3_column_int64(st, 3);
sqlite3_finalize(st);
if (dm_media_exec(inst, "BEGIN IMMEDIATE")) return;
st = dm_media_prepare(inst, "INSERT INTO dm_custody_done(id,body_hash,receipt) VALUES(?,?,?)"
" ON CONFLICT(id) DO UPDATE SET receipt=COALESCE(excluded.receipt,receipt)");
if (!st) goto rollback;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC); sqlite3_bind_blob(st, 2, hash, 32, SQLITE_STATIC);
if (delivered) sqlite3_bind_blob(st, 3, receipt, sizeof(receipt), SQLITE_STATIC); else sqlite3_bind_null(st, 3);
if (dm_media_step(inst, st)) goto rollback;
st = dm_media_prepare(inst, "DELETE FROM dm_custody WHERE id=? AND ready=-1");
if (!st) goto rollback;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
if (dm_media_step(inst, st) || dm_media_exec(inst, "COMMIT")) goto rollback;
dm_media_send(inst, group, super, delivered ? DM_MEDIA_DELETE_ACK : DM_MEDIA_EXPIRED, id, 16);
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: custody copy removed reason=%s super=%llu",
delivered ? "delivered" : "expired", (unsigned long long)super);
return;
rollback:
dm_media_exec(inst, "ROLLBACK");
}
/* Удаление больших файлов и fsync каталога выполняются worker, а не uasync. */
static void dm_custody_remove(struct dm_media_state* media, const uint8_t id[16]) {
if (dm_job_find(media, id, 1)) return;
struct dm_file_job* job = u_calloc(1, sizeof(*job));
if (!job) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: delete worker allocation failed"); return; }
job->media = media; job->work = DM_FILE_DELETE;
memcpy(job->id, id, 16);
if (dm_file_path(media, 1, id, ".cipher", job->cipher, sizeof(job->cipher)) ||
dm_file_path(media, 1, id, ".download", job->temp, sizeof(job->temp))) { u_free(job); return; }
job->next = media->jobs; media->jobs = job;
media_async_submit(media->inst->media_async, media->inst->ua, dm_file_work, job, dm_file_done, job);
}
/* Подпись B позволяет любому авторизованному S завершить удаление точного артефакта. */
static void dm_custody_delete(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source, const uint8_t* r) {
if (dm_media_verify_receipt(inst, r, NULL, 0)) return;
uint64_t author, recipient;
memcpy(&author, r + 24, 8); memcpy(&recipient, r + 32, 8);
/* Для существующей копии точная подпись B разрешает удаление даже после отзыва роли M/S. */
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT body FROM dm_custody WHERE id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, r + 40, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
if (rc == SQLITE_ROW) {
int valid = !dm_media_verify_receipt(inst, r, sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0));
sqlite3_finalize(st);
if (!valid) return;
st = dm_media_prepare(inst, "UPDATE dm_custody SET ready=-1,receipt=?,retry_at=0 WHERE id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, r, DM_MEDIA_RECEIPT_SIZE, SQLITE_STATIC); sqlite3_bind_blob(st, 2, r + 40, 16, SQLITE_STATIC);
if (dm_media_step(inst, st)) return;
file_transfer_cancel_object(inst->file_transfer, r + 40);
dm_custody_remove(inst->dm_media, r + 40);
return;
}
sqlite3_finalize(st);
if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: delete lookup failed rc=%d", rc); return; }
st = dm_media_prepare(inst, "SELECT body_hash FROM dm_custody_done WHERE id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, r + 40, 16, SQLITE_STATIC);
rc = sqlite3_step(st);
int exact = rc == SQLITE_DONE || (rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32 &&
!memcmp(sqlite3_column_blob(st, 0), r + 56, 32));
sqlite3_finalize(st);
if (!exact) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: delete conflicts with retired artifact"); return; }
if (rc == SQLITE_DONE && (!dm_media_scope(inst, group, source, "supernode", author, recipient) ||
!dm_media_scope(inst, group, inst->node_id, "storage", author, recipient))) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: unknown DELETE without custody grant source=%llu", (unsigned long long)source); return;
}
st = dm_media_prepare(inst, "INSERT INTO dm_custody_done(id,body_hash,receipt) VALUES(?,?,?)"
" ON CONFLICT(id) DO UPDATE SET receipt=excluded.receipt");
if (!st) return;
sqlite3_bind_blob(st, 1, r + 40, 16, SQLITE_STATIC); sqlite3_bind_blob(st, 2, r + 56, 32, SQLITE_STATIC);
sqlite3_bind_blob(st, 3, r, DM_MEDIA_RECEIPT_SIZE, SQLITE_STATIC);
if (!dm_media_step(inst, st)) dm_media_send(inst, group, source, DM_MEDIA_DELETE_ACK, r + 40, 16);
}
/* OFFER не меняет идентичность беседы. Чужой holder допустим только от S. */
static void dm_media_offer(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source, const uint8_t* p, size_t len) {
if (len <= 24) { DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: short OFFER"); return; }
uint64_t recipient, holder, expiry, author;
memcpy(&recipient, p, 8); memcpy(&holder, p + 8, 8); memcpy(&expiry, p + 16, 8);
p += 24; len -= 24;
const uint8_t* desc = dm_message_media(p, len);
if (recipient != inst->node_id || !desc || dm_verify_message(inst, recipient, p, len)) return;
memcpy(&author, p + 24, 8);
if ((source == author && (holder != author || expiry)) ||
(source != author && !dm_media_scope(inst, group, source, "supernode", author, recipient)) ||
(holder != author && !dm_media_scope(inst, group, holder, "storage", author, recipient))) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: unauthorized OFFER source=%llu holder=%llu",
(unsigned long long)source, (unsigned long long)holder); return;
}
uint8_t receipt[DM_RECEIPT_SIZE];
if (dm_accept_message(inst, p, len, receipt)) return;
dm_mailbox_receipt(inst, receipt);
uint8_t file_receipt[DM_MEDIA_RECEIPT_SIZE];
int deleted = dm_history_deleted(inst, p, len, 0, file_receipt);
if (deleted) {
if (deleted > 0 && !memcmp(file_receipt, "DMFILE01", 8)) {
dm_media_send(inst, group, author, DM_MEDIA_RECEIPT, file_receipt, sizeof(file_receipt));
dm_mailbox_media_receipt(inst, file_receipt);
}
return;
}
sqlite3_stmt* st = dm_media_prepare(inst, "UPDATE dm_files SET holder=?,expires=?,group_id=?,retry_at=0 WHERE id=? AND ready<=0");
if (!st) return;
sqlite3_bind_int64(st, 1, (sqlite3_int64)holder); sqlite3_bind_int64(st, 2, (sqlite3_int64)expiry);
sqlite3_bind_int64(st, 3, (sqlite3_int64)group); sqlite3_bind_blob(st, 4, desc, 16, SQLITE_STATIC);
dm_media_step(inst, st);
}
/* Строгие границы control, независимые подписи DM и квитанций. */
static void dm_media_control(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source,
uint8_t cmd, const uint8_t* p, size_t len) {
if (!inst->dm_media || inst->dm_media->closing) return;
if (cmd == DM_MEDIA_OFFER) dm_media_offer(inst, group, source, p, len);
else if (cmd == DM_MEDIA_STORE) dm_custody_store(inst, group, source, p, len);
else if (cmd == DM_MEDIA_DELETE && len == DM_MEDIA_RECEIPT_SIZE) dm_custody_delete(inst, group, source, p);
else if (cmd == DM_MEDIA_RECEIPT && len == DM_MEDIA_RECEIPT_SIZE) {
uint64_t author; memcpy(&author, p + 24, 8);
if (author == inst->node_id) {
dm_media_accept_receipt(inst, p);
} else dm_mailbox_media_recv(dm_mailbox_get_media(inst), group, source, cmd, p, len);
} else if ((cmd == DM_MEDIA_STORED && len == 24) || (cmd == DM_MEDIA_DELETE_ACK && len == 16) ||
(cmd == DM_MEDIA_EXPIRED && len == 16) || (cmd == DM_MEDIA_REJECT && len == 17)) {
dm_mailbox_media_recv(dm_mailbox_get_media(inst), group, source, cmd, p, len);
if (cmd == DM_MEDIA_EXPIRED) {
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT author,recipient FROM dm_files WHERE id=? AND dir=0 AND ready<=0");
if (!st) return;
sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
uint64_t author = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 0) : 0;
uint64_t recipient = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 1) : 0;
sqlite3_finalize(st);
if (author && dm_media_scope(inst, group, source, "supernode", author, recipient)) {
st = dm_media_prepare(inst, "UPDATE dm_files SET ready=-1,expires=0 WHERE id=? AND ready<=0");
if (st) { sqlite3_bind_blob(st, 1, p, 16, SQLITE_STATIC); dm_media_step(inst, st); }
dm_media_notify(inst, dm_derive_conv_id(author, recipient));
}
}
} else DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: invalid control cmd=%u bytes=%zu", cmd, len);
}
static void dm_media_recv(struct ETCP_CONN* conn, struct ll_entry* e) {
if (!conn || !e || !e->dgram || e->len <= ROUTER_SVC_PAYLOAD_OFF) goto done;
uint64_t group, source;
memcpy(&group, e->dgram + ROUTER_SVC_GROUP_OFF, 8); memcpy(&source, e->dgram + ROUTER_SVC_SRC_OFF, 8);
if (!(e->dgram[ROUTER_SVC_FLAGS_OFF] & ROUTER_FLAG_SIGNED)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: unsigned control source=%llu", (unsigned long long)source); goto done;
}
const uint8_t* p = e->dgram + ROUTER_SVC_PAYLOAD_OFF;
dm_media_control(conn->instance, group, source, p[0], p + 1, e->len - ROUTER_SVC_PAYLOAD_OFF - 1);
done:
if (e) { queue_dgram_free(e); queue_entry_free(e); }
}
/* Текст и файл имеют независимые durable retries; наличие text ACK не останавливает медиа. */
static void dm_files_tick(struct dm_media_state* media, uint64_t now) {
struct UTUN_INSTANCE* inst = media->inst;
for (int i = 0; i < 8; i++) {
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT id,dir,author,recipient,group_id,body,ready,holder,expires,receipt"
" FROM dm_files WHERE retry_at<=? ORDER BY retry_at,seq LIMIT 1");
if (!st) return;
sqlite3_bind_int64(st, 1, (sqlite3_int64)now);
int rc = sqlite3_step(st);
if (rc == SQLITE_DONE) { sqlite3_finalize(st); break; }
if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != 16) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: local files scan failed rc=%d", rc); sqlite3_finalize(st); return;
}
uint8_t id[16], body[DM_MEDIA_BODY_MAX], receipt[DM_MEDIA_RECEIPT_SIZE];
memcpy(id, sqlite3_column_blob(st, 0), 16);
int dir = sqlite3_column_int(st, 1), ready = sqlite3_column_int(st, 6);
uint64_t author = (uint64_t)sqlite3_column_int64(st, 2), recipient = (uint64_t)sqlite3_column_int64(st, 3);
uint64_t group = (uint64_t)sqlite3_column_int64(st, 4), holder = (uint64_t)sqlite3_column_int64(st, 7);
uint64_t expiry = (uint64_t)sqlite3_column_int64(st, 8);
size_t len = (size_t)sqlite3_column_bytes(st, 5);
int delivered = sqlite3_column_bytes(st, 9) == sizeof(receipt);
if (len <= sizeof(body)) memcpy(body, sqlite3_column_blob(st, 5), len);
if (delivered) memcpy(receipt, sqlite3_column_blob(st, 9), sizeof(receipt));
sqlite3_finalize(st);
st = dm_media_prepare(inst, "UPDATE dm_files SET retry_at=? WHERE id=?");
if (!st) return;
sqlite3_bind_int64(st, 1, (sqlite3_int64)(now + (dir && delivered ? 3600 : 5)));
sqlite3_bind_blob(st, 2, id, 16, SQLITE_STATIC);
if (dm_media_step(inst, st)) return;
const uint8_t* desc = len <= sizeof(body) ? dm_message_media(body, len) : NULL;
if (!desc) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: corrupt saved file descriptor"); continue; }
if (dir && !delivered) {
uint64_t gid = group;
if (dm_route_group(inst, recipient, group, &gid)) {
uint8_t offer[24 + DM_MEDIA_BODY_MAX];
memcpy(offer, &recipient, 8); memcpy(offer + 8, &author, 8); memset(offer + 16, 0, 8);
memcpy(offer + 24, body, len);
dm_media_send(inst, gid, recipient, DM_MEDIA_OFFER, offer, len + 24);
} else dm_mailbox_put(inst, recipient, author, body, len);
} else if (!dir && ready == 1 && delivered) {
uint64_t gid = group;
if (dm_route_group(inst, author, group, &gid)) dm_media_send(inst, gid, author, DM_MEDIA_RECEIPT, receipt, sizeof(receipt));
dm_mailbox_media_receipt(inst, receipt);
} else if (!dir && ready <= 0 && !dm_job_find(media, id, 0)) {
uint64_t peer = author, gid = group;
if (!dm_route_group(inst, author, group, &gid)) {
peer = holder;
if (holder == author || (expiry && expiry <= now) || !dm_route_group(inst, holder, group, &gid)) continue;
}
dm_file_pull(media, 0, gid, peer, desc, author);
}
}
}
/* Периодическая уборка TTL не вытесняет подтверждённые файлы раньше их expiry. */
static void dm_custody_tick(struct dm_media_state* media, uint64_t now) {
struct UTUN_INSTANCE* inst = media->inst;
for (int i = 0; i < 8; i++) {
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT id,body,author,supernode,group_id,expires,ready FROM dm_custody"
" WHERE retry_at<=? ORDER BY retry_at LIMIT 1");
if (!st) return;
sqlite3_bind_int64(st, 1, (sqlite3_int64)now);
int rc = sqlite3_step(st);
if (rc == SQLITE_DONE) { sqlite3_finalize(st); break; }
if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != 16) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: custody scan failed rc=%d", rc); sqlite3_finalize(st); return;
}
uint8_t id[16], body[DM_MEDIA_BODY_MAX];
memcpy(id, sqlite3_column_blob(st, 0), 16);
size_t len = (size_t)sqlite3_column_bytes(st, 1);
if (len <= sizeof(body)) memcpy(body, sqlite3_column_blob(st, 1), len);
uint64_t author = (uint64_t)sqlite3_column_int64(st, 2), super = (uint64_t)sqlite3_column_int64(st, 3);
uint64_t group = (uint64_t)sqlite3_column_int64(st, 4), expires = (uint64_t)sqlite3_column_int64(st, 5);
int ready = sqlite3_column_int(st, 6);
sqlite3_finalize(st);
st = dm_media_prepare(inst, "UPDATE dm_custody SET retry_at=?,ready=CASE WHEN expires<=? THEN -1 ELSE ready END WHERE id=?");
if (!st) return;
sqlite3_bind_int64(st, 1, (sqlite3_int64)(now + 5)); sqlite3_bind_int64(st, 2, (sqlite3_int64)now);
sqlite3_bind_blob(st, 3, id, 16, SQLITE_STATIC);
if (dm_media_step(inst, st)) return;
if (expires <= now || ready < 0) {
file_transfer_cancel_object(inst->file_transfer, id);
dm_custody_remove(media, id);
} else if (!ready) {
const uint8_t* desc = len <= sizeof(body) ? dm_message_media(body, len) : NULL;
uint64_t gid = group;
if (desc && dm_route_group(inst, author, group, &gid)) dm_file_pull(media, 1, gid, author, desc, author);
} else {
uint8_t stored[24]; memcpy(stored, id, 16); memcpy(stored + 16, &expires, 8);
dm_media_send(inst, group, super, DM_MEDIA_STORED, stored, sizeof(stored));
}
}
}
static void dm_media_tick(void* arg) {
struct dm_media_state* media = arg;
media->timer = NULL;
uint64_t now = ntp_time_get_seconds(media->inst);
dm_files_tick(media, now);
dm_custody_tick(media, now);
dm_cleanup_pump(media);
media->timer = uasync_set_timeout(media->inst->ua, 10000, media, dm_media_tick, "dm_media");
if (!media->timer) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: retry timer allocation failed");
}
int dm_media_init(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->dm || !inst->media_async || inst->dm_media) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: invalid initialization state"); return -1;
}
struct dm_media_state* media = u_calloc(1, sizeof(*media));
if (!media) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: owner allocation failed"); return -1; }
media->inst = inst;
const char* base = inst->config->global.db_path;
if (!base[0]) base = ".";
int n = snprintf(media->root, sizeof(media->root), "%s/dm_media", base);
int m = snprintf(media->pending, sizeof(media->pending), "%s/dm_pending", base);
if (n < 0 || (size_t)n >= sizeof(media->root) || m < 0 || (size_t)m >= sizeof(media->pending) ||
(utun_mkdir(media->root, 0700) && errno != EEXIST) || (utun_mkdir(media->pending, 0700) && errno != EEXIST)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: cannot create file roots: %s", strerror(errno)); goto fail;
}
if (dm_media_exec(inst, "CREATE TABLE IF NOT EXISTS dm_files(id BLOB PRIMARY KEY,conv_id TEXT NOT NULL,dir INTEGER NOT NULL,"
"seq INTEGER NOT NULL,author INTEGER NOT NULL,recipient INTEGER NOT NULL,plain_size INTEGER NOT NULL,"
"cipher_hash BLOB NOT NULL,body BLOB NOT NULL,group_id INTEGER NOT NULL,local_path TEXT NOT NULL,"
"ready INTEGER NOT NULL DEFAULT 0,holder INTEGER NOT NULL,expires INTEGER NOT NULL DEFAULT 0,"
"receipt BLOB,downloading INTEGER NOT NULL DEFAULT 0,retry_at INTEGER NOT NULL DEFAULT 0,"
"UNIQUE(conv_id,dir,seq))") ||
dm_media_exec(inst, "CREATE TABLE IF NOT EXISTS dm_custody(id BLOB PRIMARY KEY,author INTEGER NOT NULL,recipient INTEGER NOT NULL,"
"plain_size INTEGER NOT NULL,cipher_size INTEGER NOT NULL,body BLOB NOT NULL,body_hash BLOB NOT NULL,"
"group_id INTEGER NOT NULL,supernode INTEGER NOT NULL,expires INTEGER NOT NULL,"
"ready INTEGER NOT NULL DEFAULT 0,receipt BLOB,retry_at INTEGER NOT NULL DEFAULT 0)") ||
dm_media_exec(inst, "CREATE TABLE IF NOT EXISTS dm_custody_done(id BLOB PRIMARY KEY,body_hash BLOB NOT NULL,receipt BLOB)") ||
dm_media_exec(inst, "UPDATE dm_files SET downloading=0,retry_at=0") || dm_media_exec(inst, "UPDATE dm_custody SET retry_at=0")) goto fail;
inst->dm_media = media;
if (!file_transfer_create(inst, dm_file_lookup, media) || etcp_router_bind(inst, ETCP_RT_ID_DM_MEDIA, dm_media_recv)) goto fail;
media->timer = uasync_set_timeout(inst->ua, 10000, media, dm_media_tick, "dm_media");
if (!media->timer) goto fail;
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: initialized root=%s custody=%s ttl=%ds quota=%dMiB", media->root, media->pending,
chat_setting_get_int(inst, "dm_media_ttl_sec", 604800), chat_setting_get_int(inst, "dm_media_storage_mb", 1024));
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: initialization failed");
if (inst->dm_media) dm_media_destroy(inst); else u_free(media);
return -1;
}
/* Снять producers до ожидания workers; callbacks отмены ещё имеют живого owner. */
void dm_media_quiesce(struct UTUN_INSTANCE* inst) {
struct dm_media_state* media = inst ? inst->dm_media : NULL;
if (!media || media->closing) return;
media->closing = 1;
if (media->timer) uasync_cancel_timeout(inst->ua, media->timer);
media->timer = NULL;
etcp_router_unbind(inst, ETCP_RT_ID_DM_MEDIA);
file_transfer_destroy(inst->file_transfer);
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: quiesced, waiting for workers");
}
void dm_media_destroy(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->dm_media) return;
dm_media_quiesce(inst);
struct dm_media_state* media = inst->dm_media;
if (media->jobs || media->cleanup_active) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: destroy before media_async workers drained");
return;
}
inst->dm_media = NULL;
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: destroyed, durable files/jobs retained");
u_free(media);
}
Reference in new issue View Git Blame Copy Permalink