jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Restrict media delivery to admitted channel members

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 18:54:54 +01:00
1 parent 5f2eb2ef70
commit 7255e76ef7
5 files changed
+81 -54
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/media_delivery/media_delivery.c
+40 -16
View File
@@ -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 <string.h>
#include <stdlib.h>
#include <time.h>
#include <errno.h>
#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);
}
src/media_delivery/media_delivery.h
+5 -1
View File
@@ -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);
src/media_delivery/media_delivery_proto.h
+1 -1
View File
@@ -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))
src/media_delivery/media_download.c
+32 -36
View File
@@ -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;
src/media_delivery/media_download.h
+3
View File
@@ -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; несовпадающие/истёкшие попытки игнорируются.
* Повреждённый пакет не меняет состояние; ошибка поиска продолжает обход источников.