diff --git a/src/dm/dm_mailbox.c b/src/dm/dm_mailbox.c index 5c0154501..f1542b609 100644 --- a/src/dm/dm_mailbox.c +++ b/src/dm/dm_mailbox.c @@ -24,7 +24,8 @@ #define MB_PUT 1 /* [recipient:8][каноническое тело] */ #define MB_DELIVER 2 /* [каноническое тело] */ #define MB_RECEIPT 3 /* [подписанная квитанция получателя] */ -#define MB_STORED 4 /* [conv:8][seq:8] — durable PUT подтверждён суперузлом */ +#define MB_STORED 4 /* [conv:8][seq:8][recipient:8] — durable PUT конкретному устройству */ +#define MB_MEDIA_REQUEST 5 /* [recipient:8][file-message], отдельный заказ хранения */ struct dm_mb_state { struct UTUN_INSTANCE* inst; @@ -99,7 +100,7 @@ static int mb_exec(struct dm_mb_state* mb, const char* sql) { } /* Принять неизменяемое тело; подтверждать только durable запись или точный повтор. */ -static void mb_put_received(struct dm_mb_state* mb, uint64_t gid, uint64_t source, const uint8_t* p, size_t len) { +static void mb_put_received(struct dm_mb_state* mb, uint64_t gid, uint64_t source, const uint8_t* p, size_t len, int media_only) { if (len < 8) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: short PUT bytes=%zu", MB_ID, len); return; } uint64_t recipient, author, seq; memcpy(&recipient, p, 8); @@ -115,8 +116,9 @@ static void mb_put_received(struct dm_mb_state* mb, uint64_t gid, uint64_t sourc } sqlite3_stmt* st = NULL; if (mb_exec(mb, "BEGIN IMMEDIATE") != 0) return; - if (dm_mailbox_media_register(mb->media, gid, recipient, p, len) != 0) { - mb_exec(mb, "ROLLBACK"); + if (media_only) { + if (!dm_message_media(p, len) || dm_mailbox_media_register(mb->media, gid, recipient, p, len) || mb_exec(mb, "COMMIT")) + mb_exec(mb, "ROLLBACK"); return; } int rc = sqlite3_prepare_v2(mb->db, @@ -165,7 +167,8 @@ static void mb_put_received(struct dm_mb_state* mb, uint64_t gid, uint64_t sourc sqlite3_finalize(st); st = NULL; if (mb_exec(mb, "COMMIT") != 0) goto fail; - mb_send(mb, gid, author, MB_STORED, p, 16); + uint8_t stored[24]; memcpy(stored, p, 16); memcpy(stored + 16, &recipient, 8); + mb_send(mb, gid, author, MB_STORED, stored, sizeof(stored)); DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: stored sender=%llu recipient=%llu seq=%llu group=%llu", MB_ID, (unsigned long long)author, (unsigned long long)recipient, (unsigned long long)seq, (unsigned long long)gid); return; @@ -256,7 +259,7 @@ static void mb_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unsigned control source=%llu", MB_ID, (unsigned long long)source); goto done; } - if (p[0] == MB_PUT) mb_put_received(mb, gid, source, p + 1, len); + if (p[0] == MB_PUT || p[0] == MB_MEDIA_REQUEST) mb_put_received(mb, gid, source, p + 1, len, p[0] == MB_MEDIA_REQUEST); else if (p[0] == MB_DELIVER) { if (len < DM_MSG_FIXED_HDR) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: short delivery", MB_ID); goto done; } uint64_t author; @@ -269,10 +272,22 @@ static void mb_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { if (dm_accept_message(mb->inst, p + 1, len, receipt) == 0) mb_send(mb, gid, source, MB_RECEIPT, receipt, sizeof(receipt)); } else if (p[0] == MB_RECEIPT && len == DM_RECEIPT_SIZE) mb_receipt_received(mb, gid, source, p + 1); - else if (p[0] == MB_STORED && len == 16) { - uint64_t conv, seq; + else if (p[0] == MB_STORED && len == 24) { + uint64_t conv, seq, recipient; memcpy(&conv, p + 1, 8); memcpy(&seq, p + 9, 8); + memcpy(&recipient, p + 17, 8); + if (!mb_scope(mb, gid, source, mb->inst->node_id, recipient)) { + DEBUG_WARN(DEBUG_CATEGORY_DM, "DM mailbox: unauthorized STORED super=%llu", (unsigned long long)source); goto done; + } + char id[64]; snprintf(id, sizeof(id), "%llu", (unsigned long long)conv); + sqlite3_stmt* st = dm_media_prepare(mb->inst, + "UPDATE dm_delivery SET supernode=? WHERE conv_id=? AND author=? AND seq=? AND recipient=? AND receipt IS NULL"); + if (!st) goto done; + sqlite3_bind_int64(st, 1, (sqlite3_int64)source); sqlite3_bind_text(st, 2, id, -1, SQLITE_STATIC); + sqlite3_bind_int64(st, 3, (sqlite3_int64)mb->inst->node_id); sqlite3_bind_int64(st, 4, (sqlite3_int64)seq); + sqlite3_bind_int64(st, 5, (sqlite3_int64)recipient); + if (dm_media_step(mb->inst, st)) goto done; DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: custody confirmed super=%llu conv=%llu seq=%llu (await recipient)", MB_ID, (unsigned long long)source, (unsigned long long)conv, (unsigned long long)seq); } else DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: invalid command=%u bytes=%zu", MB_ID, p[0], len); @@ -416,13 +431,13 @@ static int mb_to_supernodes(struct dm_mb_state* mb, uint64_t recipient, uint64_t if (!path.next_hop_node_id) continue; if (!mb_send(mb, ids[i], node, cmd, body, len)) { sent++; - if (cmd == MB_PUT) break; + if (cmd == MB_PUT || cmd == MB_MEDIA_REQUEST) break; } } if (rc != SQLITE_ROW && rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: supernode scan failed rc=%d", MB_ID, rc); sqlite3_finalize(st); - if (sent && cmd == MB_PUT) break; + if (sent && (cmd == MB_PUT || cmd == MB_MEDIA_REQUEST)) break; } u_free(ids); return sent; @@ -453,3 +468,17 @@ void dm_mailbox_receipt(struct UTUN_INSTANCE* inst, const uint8_t* receipt) { memcpy(&recipient, receipt + 32, 8); mb_to_supernodes(mb, recipient, author, MB_RECEIPT, receipt, DM_RECEIPT_SIZE); } + +int dm_mailbox_store_file(struct UTUN_INSTANCE* inst, uint64_t recipient, const uint8_t* msg, size_t len) { + struct dm_mb_state* mb = inst ? inst->dm_mailbox : NULL; + uint64_t author = 0; + if (len >= DM_MSG_FIXED_HDR) memcpy(&author, msg + 24, 8); + if (!mb || author != inst->node_id || !dm_message_media(msg, len) || dm_verify_message(inst, recipient, msg, len)) { + DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM mailbox: invalid local file storage request"); return -1; + } + uint8_t body[8 + DM_MEDIA_BODY_MAX]; memcpy(body, &recipient, 8); memcpy(body + 8, msg, len); + int sent = mb_to_supernodes(mb, recipient, author, MB_MEDIA_REQUEST, body, len + 8); + if (sent <= 0) DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM file storage: waiting for authorized reachable S recipient=%llu", + (unsigned long long)recipient); + return sent > 0 ? 0 : -1; +}