Authorize referenced stream handles in cross-group overload controls
This commit is contained in:
1 parent
3fee6e6f82
commit
9b4ef4d83b
2 files changed
+44
-1
No files matched your search
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in new issue
Block a user