Separate file custody requests from per-device message mailboxes
This commit is contained in:
1 parent
093bd92bad
commit
1dcd2f1ce6
1 file changed
+39
-10
+39
-10
@@ -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;
|
||||
}
|
||||
Reference in new issue
Block a user