diff --git a/src/media_delivery/media_delivery.c b/src/media_delivery/media_delivery.c index d8f932d2d..c6889132c 100644 --- a/src/media_delivery/media_delivery.c +++ b/src/media_delivery/media_delivery.c @@ -5,6 +5,7 @@ #include "../../lib/file_io.h" #include "../utun_instance.h" #include "../routing_layer/etcp_router.h" +#include "../routing_layer/route_crypto.h" #include "../routing_layer/topo_group.h" #include "../routing_layer/topo_node.h" #include "../routing_layer/topo_node_sqlite.h" @@ -19,6 +20,7 @@ #include "../lib/u_async.h" #include "../lib/sqlite3.h" #include +#include #include #include #include "../lib/platform_compat.h" @@ -32,8 +34,6 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry); static void md_on_conn_status(struct ETCP_CONN* conn, int status, void* arg); static void md_on_props_changed(uint64_t node_id, const char* adm_tags, const char* channel_id, void* arg); static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_id, uint64_t group_id); -static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, - const uint8_t* data, size_t len, int force); static void md_repl_pump(void* arg); static void md_repl_watchdog(void* arg); static void md_repl_notify(struct media_delivery_ctx* md, uint64_t group_id); @@ -550,9 +550,15 @@ static void md_repl_notify(struct media_delivery_ctx* md, uint64_t group_id) { /* ── send via etcp_router helper ── */ /* Репликация использует force=0; остальные команды сохраняют существующую политику очереди. */ -static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, +int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, const uint8_t* data, size_t len, int force) { - if (!inst || !data || len == 0) return -1; + if (!inst || !data || !len || len >= UINT16_MAX) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: invalid send bytes=%zu", MD_ID, len); return -1; + } + if (!topo_group_chat_allowed(inst, group_id, dst_node_id)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: send denied group=%llx peer=%llx cmd=%u", MD_ID, + (unsigned long long)group_id, (unsigned long long)dst_node_id, data[0]); return -1; + } struct TOPO_GROUP* grp = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL; struct ETCP_CONN* conn = grp ? topo_group_find_conn_for_node(grp, dst_node_id) : NULL; if (!conn) @@ -564,7 +570,7 @@ static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_n if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: send payload allocation failed", MD_ID); queue_entry_free(e); return -1; } e->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY; memcpy(e->dgram + 1, data, len); e->len = (uint16_t)(len + 1); - int rc = etcp_route_send(inst, group_id, dst_node_id, e, force, 0); + int rc = etcp_route_send(inst, group_id, dst_node_id, e, force, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT); if (rc != 0) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: md_send FAIL rc=%d dst=0x%016llx group=0x%016llx", MD_ID, rc, (unsigned long long)dst_node_id, (unsigned long long)group_id); @@ -886,6 +892,11 @@ static void stream_pump(void* arg) { struct media_block_source* source = sc->source; sc->pump = NULL; if (!md->initialized) { stream_free(sc); return; } + if (!topo_group_chat_allowed(md->inst, sc->group_id, sc->dst_node_id)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: TX revoked group=%llx peer=%llx", MD_ID, + (unsigned long long)sc->group_id, (unsigned long long)sc->dst_node_id); + stream_free(sc); return; + } etcp_router_cancel_send_ready(md->inst, sc->group_id, sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY, &sc->waiter); for (int budget = 0; budget < 4; budget++) { if (!source->failed && sc->offset == source->available && !source->verified) return; @@ -1225,8 +1236,17 @@ static void md_reconcile_timer(void* arg) { /* member_sync сообщает и изменение подписанных свойств, и BGP-события; REMOVE маршрута не удаляет участника. */ static void md_on_props_changed(uint64_t node_id, const char* adm_tags, const char* channel_id, void* arg) { - (void)node_id; (void)adm_tags; (void)channel_id; + (void)adm_tags; struct media_delivery_ctx* md = arg; + char* end = NULL; + uint64_t group = channel_id ? strtoull(channel_id, &end, 10) : 0; + if (!group || !end || *end) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: invalid membership channel", MD_ID); return; } + if (!sqlite3_get_autocommit(md->db)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: access reconciliation deferred during SQL transaction", MD_ID); return; + } + etcp_router_revalidate_group(md->inst, group); + if (node_id == md->self_node_id && !topo_group_chat_allowed(md->inst, group, node_id)) + media_delivery_cancel_channel(md->inst, group); media_delivery_refresh_roles(md->inst); } @@ -1263,10 +1283,19 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { const uint8_t* data = entry->dgram; size_t len = entry->len; /* Единый формат router_deliver: [svc_id][src_node_id(8)][dst_node_id(8)][subcmd][...] */ - if (len < ROUTER_SVC_PAYLOAD_OFF + 1) { queue_dgram_free(entry); queue_entry_free(entry); return; } + if (len < ROUTER_SVC_PAYLOAD_OFF) { queue_dgram_free(entry); queue_entry_free(entry); return; } uint64_t from_node, group_id; memcpy(&from_node, data + ROUTER_SVC_SRC_OFF, 8); memcpy(&group_id, data + ROUTER_SVC_GROUP_OFF, 8); + if (len == ROUTER_SVC_PAYLOAD_OFF) { + for (struct stream_ctx* sc = md->streams; sc;) { + struct stream_ctx* next = sc->next; + if (sc->group_id == group_id && sc->dst_node_id == from_node) stream_free(sc); + sc = next; + } + media_download_peer_closed(inst, group_id, from_node); + queue_dgram_free(entry); queue_entry_free(entry); return; + } uint8_t subcmd = data[ROUTER_SVC_PAYLOAD_OFF]; data += ROUTER_SVC_PAYLOAD_OFF; len -= ROUTER_SVC_PAYLOAD_OFF; @@ -1422,6 +1451,10 @@ int media_delivery_init(struct UTUN_INSTANCE* inst) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: failed to bind ETCP_RT_ID_MEDIA_DELIVERY", MD_ID); return -1; } + if (etcp_router_require_crypto(inst, ETCP_RT_ID_MEDIA_DELIVERY, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) || + etcp_router_set_authorize(inst, ETCP_RT_ID_MEDIA_DELIVERY, topo_group_chat_allowed)) { + etcp_router_unbind(inst, ETCP_RT_ID_MEDIA_DELIVERY); return -1; + } if (member_sync_add_props_cbk(inst, md_on_props_changed, md)) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: cannot subscribe to member roles", MD_ID); @@ -1568,15 +1601,6 @@ void media_delivery_send_have_block(struct UTUN_INSTANCE* inst, uint64_t group_i } if (md_sql_finish(inst->md.db, st, rc, "read own announcement timestamp")) return; - uint8_t smsg[64]; size_t soff = 0; - memcpy(smsg + soff, hb.block_id, 16); soff += 16; - memcpy(smsg + soff, &hb.chunk, 4); soff += 4; - memcpy(smsg + soff, &hb.timestamp, 8); soff += 8; - uint64_t self = inst->node_id; memcpy(smsg + soff, &self, 8); soff += 8; - if (sc_ed25519_sign(inst->my_ed25519_privkey, smsg, soff, hb.node_sign) != SC_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: Ed25519 sign failed for HAVE_BLOCK", MD_ID); - return; - } md_send(inst, group_id, dst_node_id, (const uint8_t*)&hb, sizeof(hb), 1); } diff --git a/src/media_delivery/media_delivery.h b/src/media_delivery/media_delivery.h index 571e61d5e..51a27c65c 100644 --- a/src/media_delivery/media_delivery.h +++ b/src/media_delivery/media_delivery.h @@ -20,6 +20,10 @@ extern "C" { struct UTUN_INSTANCE; +/* Общая отправка медиа из delivery/download: копирует payload, проверяет допуск, подписывает и шифрует. + * force разрешает превышение порога send_q; 0 — принято, -1 — ошибка. Только uasync. */ +int md_send(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* data, size_t len, int force); + // ── константы ── #define MEDIA_REPL_TIMEOUT_TB 20000 // начальный таймаут 2s (0.1ms) @@ -147,7 +151,7 @@ void media_delivery_announce_media(struct UTUN_INSTANCE* inst, uint64_t group_id const uint8_t* media_id, const uint8_t* block_ids, int num_blocks); -/* Отправить один HAVE_BLOCK суперузлу (подпись Ed25519 + md_send). Общий helper +/* Отправить один HAVE_BLOCK суперузлу через подписанный/зашифрованный router. Общий helper * для announce_media / announce_local_blocks / media_download. */ void media_delivery_send_have_block(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, const uint8_t* media_id, const uint8_t* block_id, uint32_t chunk); diff --git a/src/media_delivery/media_delivery_proto.h b/src/media_delivery/media_delivery_proto.h index 6c621cf43..a6f758cfa 100644 --- a/src/media_delivery/media_delivery_proto.h +++ b/src/media_delivery/media_delivery_proto.h @@ -120,7 +120,7 @@ struct media_pkt_have_block { uint8_t block_id[16]; uint32_t chunk; int64_t timestamp; // unix epoch - uint8_t node_sign[64]; // Ed25519 подпись + // Отправитель, группа и все поля защищены обязательной подписью router. }; #define MEDIA_HAVE_BLOCK_SIZE (sizeof(struct media_pkt_have_block)) diff --git a/src/media_delivery/media_download.c b/src/media_delivery/media_download.c index a1eb79e48..7a314a7b4 100644 --- a/src/media_delivery/media_download.c +++ b/src/media_delivery/media_download.c @@ -110,28 +110,6 @@ static int md_mkdir_parent(const char* filepath) { /* ── send helpers ── */ -static int md_dl_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst, const uint8_t* data, size_t len) { - if (!inst || !data || !len || len >= UINT16_MAX) { - DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: invalid send bytes=%zu", MDL_ID, len); return -1; - } - { struct TOPO_GROUP* grp = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL; - struct ETCP_CONN* gc = grp ? topo_group_find_conn_for_node(grp, dst) : NULL; - if (!gc) - DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: no conn to 0x%016llx subcmd=0x%02x — packet queued, no route", - MDL_ID, (unsigned long long)dst, (unsigned)data[0]); - } - struct ll_entry* e = queue_entry_new(0); - if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new failed for send to 0x%016llx", MDL_ID, (unsigned long long)dst); return -1; } - e->dgram = u_malloc(len + 1); - if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: u_malloc(%zu) failed for send to 0x%016llx", MDL_ID, len + 1, (unsigned long long)dst); queue_entry_free(e); return -1; } - e->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY; - memcpy(e->dgram + 1, data, len); e->len = (uint16_t)(len + 1); - int rc = etcp_route_send(inst, group_id, dst, e, 1, 0); - if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: send failed rc=%d node=%016llx group=%016llx", - MDL_ID, rc, (unsigned long long)dst, (unsigned long long)group_id); - return rc; -} - static int md_dl_send_block_req(struct UTUN_INSTANCE* inst, uint64_t dst, struct media_download* dl, int bi) { struct media_pkt_block_req req; memset(&req, 0, sizeof(req)); @@ -144,7 +122,7 @@ static int md_dl_send_block_req(struct UTUN_INSTANCE* inst, uint64_t dst, struct DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_REQ to 0x%016llx blk=%d/%d block_id=%02x%02x%02x%02x...", MDL_ID, (unsigned long long)dst, bi, dl->num_blocks, dl->blocks[bi].block_id[0], dl->blocks[bi].block_id[1], dl->blocks[bi].block_id[2], dl->blocks[bi].block_id[3]); - return md_dl_send(inst, dl->group_id, dst, (const uint8_t*)&req, sizeof(req)); + return md_send(inst, dl->group_id, dst, (const uint8_t*)&req, sizeof(req), 1); } /* ── forward decl ── */ @@ -178,19 +156,9 @@ static void md_dl_send_block_processing(struct UTUN_INSTANCE* inst, struct media hb.chunk = (uint32_t)bi; hb.timestamp = (int64_t)time(NULL); - uint8_t smsg[64]; size_t soff = 0; - memcpy(smsg + soff, dl->blocks[bi].block_id, 16); soff += 16; - memcpy(smsg + soff, &hb.chunk, 4); soff += 4; - memcpy(smsg + soff, &hb.timestamp, 8); soff += 8; - uint64_t self = inst->node_id; memcpy(smsg + soff, &self, 8); soff += 8; - if (sc_ed25519_sign(inst->my_ed25519_privkey, smsg, soff, hb.node_sign) != SC_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: Ed25519 sign failed for BLOCK_PROCESSING", MDL_ID); - return; - } - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_PROCESSING to super 0x%016llx block=%d", MDL_ID, (unsigned long long)super, bi); - md_dl_send(inst, dl->group_id, super, (const uint8_t*)&hb, sizeof(hb)); + md_send(inst, dl->group_id, super, (const uint8_t*)&hb, sizeof(hb), 1); } /* ── запрос блока у текущего держателя ── */ @@ -201,7 +169,7 @@ static void md_dl_end_attempt(struct media_download* dl, int bi) { struct media_pkt_cancel can = {.subcmd = MEDIA_SUBCMD_CANCEL, .chunk = (uint32_t)bi}; memcpy(can.media_id, dl->media_id, 16); memcpy(can.block_id, b->block_id, 16); memcpy(can.transfer_id, b->transfer_id, 16); - md_dl_send(dl->inst, dl->group_id, b->holders[b->holder_idx], (const uint8_t*)&can, sizeof(can)); + md_send(dl->inst, dl->group_id, b->holders[b->holder_idx], (const uint8_t*)&can, sizeof(can), 1); } if (b->source) { if (!b->source->verified) md_block_abort(b->source); @@ -217,6 +185,10 @@ static int md_dl_request_block(struct media_download* dl, int bi) { b->state = MD_BLK_WAIT; b->wait_reason = MD_WAIT_SEARCH; return 0; } uint64_t holder = b->holders[b->holder_idx]; + if (!topo_group_chat_allowed(dl->inst, dl->group_id, holder)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: block holder denied group=%llx holder=%llx", MDL_ID, + (unsigned long long)dl->group_id, (unsigned long long)holder); return -1; + } struct media_download_peer* p = md_dl_get_peer(dl, holder); if (!p || p->pending_event) { b->state = MD_BLK_WAIT; b->wait_reason = MD_WAIT_SLOT; return 0; } if (md_dl_reset_block_file(dl, bi)) { dl->err = -1; return -1; } @@ -392,6 +364,10 @@ void md_dl_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64_t group_i static void md_dl_pump_cb(void* arg) { struct media_download* dl = arg; dl->pump_timer = NULL; + if (!topo_group_chat_allowed(dl->inst, dl->group_id, dl->inst->node_id)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: download membership revoked group=%llx", MDL_ID, (unsigned long long)dl->group_id); + md_dl_finish(dl, -1); return; + } int failed = dl->err != 0; for (int pi = 0; pi < dl->num_peers; pi++) { struct media_download_peer* p = &dl->peers[pi]; @@ -493,7 +469,7 @@ static int md_dl_query_pump(struct media_download* dl, uint64_t now_tb) { DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: QUERY peer=%016llx source=%d blocks=%d id=%02x%02x deadline=%llu", MDL_ID, (unsigned long long)dl->query_node, source, dl->query_count, dl->query_id[0], dl->query_id[1], (unsigned long long)dl->query_deadline_tb); - if (!md_dl_send(dl->inst, dl->group_id, dl->query_node, packet, MEDIA_QUERY_HDR_SIZE + (size_t)q->num_blocks * 16)) + if (!md_send(dl->inst, dl->group_id, dl->query_node, packet, MEDIA_QUERY_HDR_SIZE + (size_t)q->num_blocks * 16, 1)) return 0; dl->query_deadline_tb = 0; // Отказ отправки: следующий источник той же пачки. } @@ -889,6 +865,26 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, return 0; } +void media_download_peer_closed(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer) { + for (struct ll_entry* e = inst->md.downloads ? inst->md.downloads->head : NULL; e; e = e->next) { + struct media_download* dl = (struct media_download*)e->data; + if (dl->group_id != group) continue; + int affected = dl->query_deadline_tb && dl->query_node == peer; + if (affected) dl->query_deadline_tb = 0; + for (int i = 0; i < dl->num_peers; i++) { + struct media_download_peer* p = &dl->peers[i]; + if (p->node_id != peer) continue; + if (p->cm_handle) { conn_mgr_close(p->cm_handle); p->cm_handle = NULL; } + p->connected = 0; p->pending_event = CONN_EVENT_DOWN + 1; affected = 1; + } + if (!affected) continue; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: router closed group=%llx peer=%llx; defer download failover", MDL_ID, + (unsigned long long)group, (unsigned long long)peer); + if (!dl->pump_timer) dl->pump_timer = uasync_call_soon(inst->ua, dl, md_dl_pump_cb); + if (!dl->pump_timer) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: cannot schedule closed peer", MDL_ID); dl->err = -1; } + } +} + int media_download_cancel(struct UTUN_INSTANCE* inst, const uint8_t* media_id, const uint8_t* block_id, uint32_t chunk) { (void)block_id; (void)chunk; diff --git a/src/media_delivery/media_download.h b/src/media_delivery/media_download.h index 0cb8635fd..321d0ffa7 100644 --- a/src/media_delivery/media_download.h +++ b/src/media_delivery/media_download.h @@ -136,6 +136,9 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, int media_download_cancel(struct UTUN_INSTANCE* inst, const uint8_t* media_id, const uint8_t* block_id, uint32_t chunk); +/* Router закрыл group/peer: отменить его CM, отложить failover активных попыток; callbacks загрузки здесь не вызываются. */ +void media_download_peer_closed(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer); + /* ── handlers for incoming data, called from md_etcp_recv dispatch ── */ /* Полный ответ пачки. group_id/from_node — из router; несовпадающие/истёкшие попытки игнорируются. * Повреждённый пакет не меняет состояние; ошибка поиска продолжает обход источников.