From 6be77de7f64ceedaea82ad24b1ab6df403dd4b9e Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 19:07:51 +0100 Subject: [PATCH] Scope media registrations and active roles to their CHAT group --- src/media_delivery/media_delivery.c | 50 +++++++++++++++++++---------- src/media_delivery/media_delivery.h | 2 +- src/media_delivery/media_download.c | 4 +++ 3 files changed, 38 insertions(+), 18 deletions(-) diff --git a/src/media_delivery/media_delivery.c b/src/media_delivery/media_delivery.c index c6889132c..a390985a0 100644 --- a/src/media_delivery/media_delivery.c +++ b/src/media_delivery/media_delivery.c @@ -580,30 +580,30 @@ int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, /* ── subcommand handlers ── */ -static void md_handle_serve_reg(struct media_delivery_ctx* md, uint64_t from_node, +static void md_handle_serve_reg(struct media_delivery_ctx* md, uint64_t group_id, uint64_t from_node, const uint8_t* data, size_t len) { - if (len < sizeof(struct media_pkt_serve_reg)) return; - struct media_pkt_serve_reg* pkt = (struct media_pkt_serve_reg*)data; + const struct media_pkt_serve_reg* pkt = (const void*)data; + if (len != sizeof(*pkt) || pkt->group_id != group_id) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: SERVE_REG invalid size/group bytes=%zu group=%llx", MD_ID, len, + (unsigned long long)group_id); return; + } if (!md->served_nodes) { - md->served_nodes = queue_new(md->inst->ua, 64, offsetof(struct media_served_node, node_id), 8, "md_served"); + md->served_nodes = queue_new(md->inst->ua, 64, offsetof(struct media_served_node, node_id), 16, "md_served"); if (!md->served_nodes) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_new(served_nodes) failed", MD_ID); return; } } - struct ll_entry* e = queue_find_data_by_index(md->served_nodes, &from_node); + uint64_t key[2] = {from_node, group_id}; + struct ll_entry* e = queue_find_data_by_index(md->served_nodes, key); if (!e) { struct ll_entry* qe = queue_entry_new(sizeof(struct media_served_node)); if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new(served_node) failed", MD_ID); return; } struct media_served_node* sn = (struct media_served_node*)qe->data; memset(sn, 0, sizeof(*sn)); - sn->node_id = from_node; sn->group_id = pkt->group_id; + sn->node_id = from_node; sn->group_id = group_id; sn->joined_at = (int64_t)time(NULL); - memcpy(qe->data, &from_node, 8); queue_data_put_with_index(md->served_nodes, qe); } - uint8_t ack[9] = { MEDIA_SUBCMD_SERVE_ACK, 0, 0 }; - memcpy(ack + 1, &pkt->group_id, 8); - (void)from_node; DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: SERVE_REG node=0x%016llx group=%016llx", MD_ID, (unsigned long long)from_node, (unsigned long long)pkt->group_id); } @@ -1121,7 +1121,8 @@ static int md_group_role(struct media_delivery_ctx* md, uint64_t group_id, uint6 struct TOPO_GROUP* g = md->inst->topo_groups ? topo_groups_find(md->inst->topo_groups, group_id) : NULL; if (!g || g->group_type != TOPO_GROUP_TYPE_CHAT || g->stopping || !g->channel_id[0]) return 0; char sql[256]; - sqlite3_snprintf(sizeof(sql), sql, "SELECT adm_tags FROM \"peers_%w\" WHERE node_id=? AND admitted=1", g->channel_id); + sqlite3_snprintf(sizeof(sql), sql, + "SELECT adm_tags FROM \"peers_%w\" WHERE node_id=? AND source=0 AND deleted=0 AND admitted=1", g->channel_id); sqlite3_stmt* st = NULL; int rc = sqlite3_prepare_v2(md->db, sql, -1, &st, NULL); if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id); @@ -1212,7 +1213,8 @@ void media_delivery_refresh_roles(struct UTUN_INSTANCE* inst) { any = 1; md_seed_local(md, group->group_id); char sql[256]; - sqlite3_snprintf(sizeof(sql), sql, "SELECT node_id,adm_tags FROM \"peers_%w\" WHERE node_id!=? AND admitted=1", group->channel_id); + sqlite3_snprintf(sizeof(sql), sql, + "SELECT node_id,adm_tags FROM \"peers_%w\" WHERE node_id!=? AND source=0 AND deleted=0 AND admitted=1", group->channel_id); sqlite3_stmt* st = NULL; int rc = sqlite3_prepare_v2(md->db, sql, -1, &st, NULL); if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)md->self_node_id); @@ -1258,9 +1260,12 @@ static void md_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) { switch (status) { case ETCP_CONN_STATUS_DOWN: case ETCP_CONN_STATUS_DELETE: - if (md->served_nodes) { - struct ll_entry* e = queue_find_data_by_index(md->served_nodes, &node_id); - if (e) { queue_remove_data(md->served_nodes, e); queue_entry_free(e); } + for (struct ll_entry* e = md->served_nodes ? md->served_nodes->head : NULL; e;) { + struct ll_entry* next = e->next; + if (((struct media_served_node*)e->data)->node_id == node_id) { + queue_remove_data(md->served_nodes, e); queue_entry_free(e); + } + e = next; } break; default: break; @@ -1288,6 +1293,11 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { memcpy(&from_node, data + ROUTER_SVC_SRC_OFF, 8); memcpy(&group_id, data + ROUTER_SVC_GROUP_OFF, 8); if (len == ROUTER_SVC_PAYLOAD_OFF) { + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: channel closed group=%llx peer=%llx", MD_ID, + (unsigned long long)group_id, (unsigned long long)from_node); + uint64_t key[2] = {from_node, group_id}; + struct ll_entry* served = md->served_nodes ? queue_find_data_by_index(md->served_nodes, key) : NULL; + if (served) { queue_remove_data(md->served_nodes, served); queue_entry_free(served); } 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); @@ -1312,13 +1322,19 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { } switch (subcmd) { case MEDIA_SUBCMD_SERVE_REG: - md_handle_serve_reg(md, from_node, data, len); + md_handle_serve_reg(md, group_id, from_node, data, len); break; case MEDIA_SUBCMD_SERVE_LEAVE: + if (len != sizeof(struct media_pkt_serve_leave) + || ((const struct media_pkt_serve_leave*)data)->group_id != group_id) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: SERVE_LEAVE invalid size/group bytes=%zu group=%llx", MD_ID, len, + (unsigned long long)group_id); break; + } DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: SERVE_LEAVE from 0x%016llx", MD_ID, (unsigned long long)from_node); if (md->served_nodes) { - struct ll_entry* e = queue_find_data_by_index(md->served_nodes, &from_node); + uint64_t key[2] = {from_node, group_id}; + struct ll_entry* e = queue_find_data_by_index(md->served_nodes, key); if (e) { queue_remove_data(md->served_nodes, e); queue_entry_free(e); } } break; diff --git a/src/media_delivery/media_delivery.h b/src/media_delivery/media_delivery.h index 51a27c65c..b4e62f278 100644 --- a/src/media_delivery/media_delivery.h +++ b/src/media_delivery/media_delivery.h @@ -114,7 +114,7 @@ struct media_delivery_ctx { int query_timeout_tb; // Дедлайн одной попытки QUERY, независимо от входящих чанков; 0 — 2с. int dl_max_attempts; // попыток на блок (по умолч. MEDIA_DL_MAX_ATTEMPTS) sqlite3* db; // = inst->topo_sqlite_db (кешируем для удобства) - struct ll_queue* served_nodes; // media_served_node — только суперузел + struct ll_queue* served_nodes; // media_served_node, индекс (node_id, group_id); регистрации независимы по группам struct ll_queue* super_peers; // media_super_peer — только суперузел struct ll_queue* downloads; // media_download — активные загрузки struct ll_queue* block_sources; // media_block_source, заимствованный индекс по block_id. diff --git a/src/media_delivery/media_download.c b/src/media_delivery/media_download.c index 7a314a7b4..9beb6c56c 100644 --- a/src/media_delivery/media_download.c +++ b/src/media_delivery/media_download.c @@ -779,6 +779,10 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, } struct media_delivery_ctx* md = &inst->md; if (!md->initialized) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: start before media_delivery init", MDL_ID); return -1; } + if (!topo_group_chat_allowed(inst, group_id, inst->node_id)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: start denied outside active CHAT membership group=%llx", MDL_ID, + (unsigned long long)group_id); return -1; + } char location[MEDIA_PATH_MAX], normalized[MEDIA_PATH_MAX]; uint8_t author_pubkey[32];