From 9b4ef4d83b7bb349295cb4e05d6c9da0a6a8eac9 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 4 Nov 2026 01:55:03 +0100 Subject: [PATCH] Authorize referenced stream handles in cross-group overload controls --- src/stream/stream_network.c | 32 +++++++++++++++++++++++++++++++- tests/test_group_api_isolation.c | 13 +++++++++++++ 2 files changed, 44 insertions(+), 1 deletion(-) diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index a982dd99e..91bdc9ecc 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -218,6 +218,30 @@ static void peer_cancel_wait(struct stream_peer *p) { memset(&p->waiter, 0, sizeof(p->waiter)); p->wait_q = NULL; } +/* OVERLOAD общий для транспорта: каждый существующий handle сохраняет допуск своей группы. + * Неизвестные/уже закрытые handles не дают доступа к данным; протокольную проверку выполняет pressure_receive. */ +static int pressure_members_allowed(struct stream_peer *p, const struct stream_message *m, int receiving) { + if (m->command != STREAM_CMD_OVERLOAD && m->command != STREAM_CMD_OVERLOAD_SELECT) return 1; + int overload = m->command == STREAM_CMD_OVERLOAD; + size_t header = overload ? STREAM_OVERLOAD_HEADER : 9, stride = overload ? 16 : 8; + if (m->length < header || m->length != header + m->payload[header - 1] * stride) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream pressure invalid handle list peer=%llx command=%u bytes=%zu", + (unsigned long long)p->node_id, m->command, m->length); return 0; + } + for (unsigned i = 0; i < m->payload[header - 1]; i++) { + uint64_t handle = stream_read_be(m->payload + header + i * stride, 8); + for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) { + if (f->info.handle != handle || f->outgoing != (overload == receiving)) continue; + if (!topo_group_chat_allowed(p->network->inst, f->descriptor.group_id, p->node_id)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream pressure handle denied group=%llx peer=%llx handle=%llu receiving=%d", + (unsigned long long)f->descriptor.group_id, (unsigned long long)p->node_id, + (unsigned long long)handle, receiving); return 0; + } + } + } + return 1; +} + /* Управление мало и ограничено; FIFO фиксирует порядок выделенных handles. В ней никогда нет DATA. */ static int control_put(struct stream_peer *p, struct stream_message m, const uint8_t *payload, size_t length) { if (length + STREAM_WIRE_HEADER > STREAM_CONTROL_MAX || p->control->count >= STREAM_CONTROL_LIMIT) { @@ -247,6 +271,10 @@ static int control_put(struct stream_peer *p, struct stream_message m, const uin static int route_control(struct stream_network *net, uint64_t peer, struct stream_message m, const uint8_t *payload, size_t length, struct stream_relay_request *request) { struct UTUN_INSTANCE *inst = net->inst; + if (!topo_group_chat_allowed(inst, m.descriptor.group_id, peer)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed send denied group=%llx peer=%llx command=%u", + (unsigned long long)m.descriptor.group_id, (unsigned long long)peer, m.command); return STREAM_ERR_NOT_MEMBER; + } int route = etcp_router_has_route(inst, m.descriptor.group_id, peer); if (!route) { DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for group route peer=%llx group=%llx command=%u", @@ -800,7 +828,8 @@ static void peer_pump(void *arg) { while ((e = queue_data_get(p->control)) != NULL) { struct stream_message m; if (stream_wire_parse(e->dgram, e->len, &m) == STREAM_OK - && topo_group_chat_allowed(p->network->inst, m.descriptor.group_id, p->node_id)) break; + && topo_group_chat_allowed(p->network->inst, m.descriptor.group_id, p->node_id) + && pressure_members_allowed(p, &m, 0)) break; DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream queued control discarded after revoke peer=%llx bytes=%u", (unsigned long long)p->node_id, e->len); queue_dgram_free(e); queue_entry_free(e); @@ -1970,6 +1999,7 @@ static void pressure_select(struct stream_peer *p, uint64_t now) { /* Проверяем весь список до действия: повторы handles и несуществующие элементы не создают частичных закрытий. */ static void pressure_receive(struct stream_peer *p, const struct stream_message *m) { + if (!pressure_members_allowed(p, m, 1)) return; uint64_t now = get_time_tb() / 10; if (m->command == STREAM_CMD_OVERLOAD) { if (m->length < STREAM_OVERLOAD_HEADER) goto malformed; diff --git a/tests/test_group_api_isolation.c b/tests/test_group_api_isolation.c index 2a57c5cea..8c386b091 100644 --- a/tests/test_group_api_isolation.c +++ b/tests/test_group_api_isolation.c @@ -300,6 +300,19 @@ static int api_streams(void) { } REQUIRE(!send_routed(&saved, relay_wire, sizeof(relay_wire), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT)); check("STREAM late signed relay cannot allocate after revoke", net->relay_request_count == 1); + /* Сохранённый запрос разгрузки до отзыва; ACK из A не должен закрыть handle группы B. */ + if (net->timer) uasync_cancel_timeout(fixture.ua, net->timer); + net->timer = NULL; + p->pressure = (struct stream_pressure_request){.id = 900, .required_rate = 1, .count = 1, + .required_count = 1, .active = 1, .deadline_ms = get_time_tb() / 10 + 5000, .cuts = {{.handle = 1, .rate = 1}}}; + struct stream_message pressure = {.descriptor = info[0].descriptor, .command = STREAM_CMD_OVERLOAD_SELECT, + .sender_epoch = p->other_epoch, .receiver_epoch = p->own_epoch}; + uint8_t selection[17] = {0}; stream_write_be(selection, 900, 8); selection[8] = 1; stream_write_be(selection + 9, 1, 8); + REQUIRE(!api_stream_packet(&pressure, selection, sizeof(selection))); f = flow_find(p, &m, 0); + check("STREAM neighbouring pressure cannot act on revoked group handle", f && f->info.state == STREAM_RECEIVE_ACTIVE + && !p->pressure.released_count); + p->pressure.active = 0; + net->timer = uasync_set_timeout(fixture.ua, STREAM_TICK_TB, net, network_tick, "stream_network"); REQUIRE(net->timer); REQUIRE(!api_admit(112, client->node_id, 1)); struct stream_message queued = {.descriptor = info[1].descriptor, .command = STREAM_CMD_REJECT, .handle = 99}; uint8_t reason = STREAM_REASON_STOPPED;