Test multicast revocation in shared queues and routed reports
This commit is contained in:
1 parent
0c5c5c03ca
commit
64d9657aa9
2 files changed
+117
-2
No files matched your search
@@ -30,6 +30,8 @@
|
||||
#include "test_utils.h"
|
||||
/* Наблюдаем очереди детектора напрямую: отказ должен предшествовать созданию binding/задания. */
|
||||
#include "../src/routing_layer/nat_detection.c"
|
||||
/* Управляем границей отложенной выдачи, не заменяя обработчики MULTICAST. */
|
||||
#include "../src/routing_layer/multicast.c"
|
||||
|
||||
#define REQUIRE(expr) do { if (!(expr)) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "isolation setup: %s", #expr); return -1; } } while (0)
|
||||
|
||||
@@ -215,7 +217,7 @@ static int setup(void) {
|
||||
if (ready == 4) break;
|
||||
REQUIRE(get_time_tb() < deadline); uasync_poll(fixture.ua, 1);
|
||||
}
|
||||
const uint8_t services[] = {ETCP_ID_BROADCAST, ETCP_ID_RADIO, ETCP_ID_SVC_ROUTE};
|
||||
const uint8_t services[] = {ETCP_ID_BROADCAST, ETCP_ID_RADIO, ETCP_ID_MULTICAST, ETCP_ID_SVC_ROUTE};
|
||||
for (unsigned i = 0; i < sizeof(services); i++) {
|
||||
uint8_t service = services[i]; fixture.receivers[service] = fixture.nodes[0]->api_bindings.callbacks[service];
|
||||
REQUIRE(fixture.receivers[service]); fixture.nodes[0]->api_bindings.callbacks[service] = observe;
|
||||
@@ -581,6 +583,118 @@ static int nat_isolation(void) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void multicast_received(struct multicast_subscription* sub, const uint8_t* data, size_t size, void* arg) {
|
||||
(void)sub; (void)data; (void)size;
|
||||
unsigned* count = arg; (*count)++;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "isolation multicast delivery count=%u", *count);
|
||||
}
|
||||
|
||||
static void multicast_lost(struct UTUN_INSTANCE* inst, const struct multicast_loss* loss, void* arg) {
|
||||
(void)inst;
|
||||
unsigned* count = arg; (*count)++;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "isolation multicast loss group=%llu count=%u", (unsigned long long)loss->group_id, *count);
|
||||
}
|
||||
|
||||
/* Wire идёт через настоящий ETCP; descriptor принадлежит отправителю, group выбирает атакующий. */
|
||||
static int inject_multicast(uint64_t group, uint16_t descriptor) {
|
||||
uint8_t wire[MULTICAST_HEADER_SIZE + 8 + 1 + 9] = {ETCP_ID_MULTICAST};
|
||||
mc_write(wire + 1, group, 8); mc_write(wire + 9, descriptor, 2); mc_write(wire + 11, 1, 2);
|
||||
mc_write(wire + 13, fixture.nodes[1]->node_id, 8); mc_write(wire + MULTICAST_HEADER_SIZE, fixture.nodes[0]->node_id, 8);
|
||||
wire[MULTICAST_HEADER_SIZE + 8] = 0x5a;
|
||||
mc_write(wire + sizeof(wire) - 9, fixture.nodes[1]->node_id, 8); wire[sizeof(wire) - 1] = 1;
|
||||
return inject(wire, sizeof(wire));
|
||||
}
|
||||
|
||||
/* Две группы делят FIFO и callback-очередь. Отзыв B должен сохранить работу A в обеих очередях. */
|
||||
static int multicast_isolation(void) {
|
||||
struct UTUN_INSTANCE* server = fixture.nodes[0]; struct UTUN_INSTANCE* client = fixture.nodes[1];
|
||||
REQUIRE(sqlite3_exec(server->topo_sqlite_db, "UPDATE peers_112 SET admitted=1", NULL, NULL, NULL) == SQLITE_OK);
|
||||
for (unsigned i = 0; i < 2; i++)
|
||||
REQUIRE(!topo_group_new_conn(fixture.groups[i][1], instance_find_conn(fixture.nodes[i], fixture.nodes[1 - i]->node_id)));
|
||||
uint64_t deadline = get_time_tb() + 50000;
|
||||
while ((!topo_group_peer_ready(fixture.groups[0][1], client->node_id) ||
|
||||
!topo_group_peer_ready(fixture.groups[1][1], server->node_id)) && get_time_tb() < deadline) uasync_poll(fixture.ua, 1);
|
||||
REQUIRE(topo_group_peer_ready(fixture.groups[0][1], client->node_id) && topo_group_peer_ready(fixture.groups[1][1], server->node_id));
|
||||
int descriptors[2][2]; unsigned received[2][2] = {{0}}, losses[2][2] = {{0}};
|
||||
struct multicast_subscription* subscriptions[2][2];
|
||||
for (unsigned i = 0; i < 2; i++) for (unsigned group = 0; group < 2; group++) {
|
||||
descriptors[i][group] = multicast_open(fixture.groups[i][group], multicast_lost, &losses[i][group]);
|
||||
REQUIRE(descriptors[i][group] > 0);
|
||||
}
|
||||
for (unsigned i = 0; i < 2; i++) for (unsigned group = 0; group < 2; group++) {
|
||||
subscriptions[i][group] = multicast_subscribe(fixture.groups[i][group], fixture.nodes[1 - i]->node_id,
|
||||
descriptors[1 - i][group], multicast_received, &received[i][group]);
|
||||
REQUIRE(subscriptions[i][group]);
|
||||
}
|
||||
REQUIRE(!inject_multicast(112, descriptors[1][1]));
|
||||
check("admitted MULTICAST delivers B", received[0][1] == 1);
|
||||
uint8_t report[MULTICAST_REPORT_HEADER_SIZE + 8] = {0};
|
||||
mc_write(report, descriptors[0][1], 2); mc_write(report + 2, client->node_id, 8); mc_write(report + 10, server->node_id, 8);
|
||||
mc_write(report + 18, 1, 4); mc_write(report + 22, 1, 2); mc_write(report + MULTICAST_REPORT_HEADER_SIZE, server->node_id, 8);
|
||||
struct ll_entry* entry = ll_alloc_lldgram(1 + sizeof(report)); REQUIRE(entry);
|
||||
entry->dgram[0] = ETCP_RT_ID_MULTICAST_REPORT; memcpy(entry->dgram + 1, report, sizeof(report)); entry->len = 1 + sizeof(report);
|
||||
REQUIRE(!etcp_route_send(client, 112, server->node_id, entry, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT));
|
||||
deadline = get_time_tb() + 30000;
|
||||
while (!losses[0][1] && get_time_tb() < deadline) uasync_poll(fixture.ua, 1);
|
||||
REQUIRE(losses[0][1] == 1);
|
||||
struct ETCP_ROUTER_CONN* route = etcp_router_conn_get(server, 112, client->node_id, ETCP_RT_ID_MULTICAST_REPORT);
|
||||
REQUIRE(route && route->peer_reset_id);
|
||||
struct SVC_ROUTE_HDR saved = {.cmd = ETCP_ID_SVC_ROUTE, .group_id = 112, .src_node_id = client->node_id,
|
||||
.dst_node_id = server->node_id, .svc_id = ETCP_RT_ID_MULTICAST_REPORT, .reset_id = route->peer_reset_id,
|
||||
.peer_reset_id = route->reset_id, .seq = route->rx_seq};
|
||||
check("admitted signed MULTICAST report reaches B descriptor", losses[0][1] == 1 && !losses[0][0]);
|
||||
|
||||
struct multicast_ctx* ctx = server->multicast;
|
||||
queue_set_callback(ctx->deliveries, NULL, NULL);
|
||||
REQUIRE(!inject_multicast(112, descriptors[1][1]) && !inject_multicast(111, descriptors[1][0]));
|
||||
REQUIRE(ctx->deliveries->count == 2 && received[0][1] == 1 && !received[0][0]);
|
||||
struct mc_out* output = mc_out_get(ctx, instance_find_conn(server, client->node_id)); REQUIRE(output);
|
||||
queue_set_callback(output->q, NULL, NULL);
|
||||
for (unsigned group = 0; group < 2; group++)
|
||||
REQUIRE(!multicast_send(server, descriptors[0][group], &client->node_id, 1, "queued", 6));
|
||||
REQUIRE(output->q->count == 2);
|
||||
char sql[160]; snprintf(sql, sizeof(sql), "UPDATE peers_112 SET admitted=0 WHERE node_id=%lld", (long long)client->node_id);
|
||||
REQUIRE(sqlite3_exec(server->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK);
|
||||
topo_group_revalidate_members(fixture.groups[0][1]);
|
||||
check("MULTICAST revoke detaches only B subscription and pending delivery", !subscriptions[0][1]->ctx &&
|
||||
subscriptions[0][0]->ctx == ctx && ctx->deliveries->count == 1 &&
|
||||
mc_read(((struct mc_delivery*)ctx->deliveries->head)->subscription->key, 8) == 111);
|
||||
check("MULTICAST revoke filters B out of shared FIFO", output->q->count == 1 && mc_read(output->q->head->dgram + 1, 8) == 111);
|
||||
queue_set_callback(ctx->deliveries, mc_deliver, ctx); queue_resume_callback(ctx->deliveries);
|
||||
queue_set_callback(output->q, mc_out_drain, output); queue_resume_callback(output->q);
|
||||
deadline = get_time_tb() + 30000;
|
||||
while ((!received[0][0] || !received[1][0]) && get_time_tb() < deadline) uasync_poll(fixture.ua, 1);
|
||||
check("MULTICAST A drains both queues on shared ETCP", received[0][0] == 1 && received[1][0] == 1 &&
|
||||
received[0][1] == 1 && !received[1][1] && fixture.transport->links_up);
|
||||
REQUIRE(!inject_multicast(112, descriptors[1][1]));
|
||||
REQUIRE(!send_routed(&saved, report, sizeof(report), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT));
|
||||
check("revoked raw MULTICAST and signed report cause no callback", received[0][1] == 1 && losses[0][1] == 1 &&
|
||||
!ctx->deliveries->count && !ctx->reports);
|
||||
check("MULTICAST cannot subscribe to revoked source", !multicast_subscribe(fixture.groups[0][1], client->node_id,
|
||||
descriptors[1][1], multicast_received, &received[0][1]));
|
||||
|
||||
/* Self отозван после постановки loopback: проверка есть и в callback без proactive cleanup. */
|
||||
struct multicast_subscription* local = multicast_subscribe(fixture.groups[0][1], server->node_id,
|
||||
descriptors[0][1], multicast_received, &received[0][1]); REQUIRE(local);
|
||||
REQUIRE(!multicast_send(server, descriptors[0][1], &server->node_id, 1, "local", 5));
|
||||
REQUIRE(ctx->deliveries->count == 1);
|
||||
snprintf(sql, sizeof(sql), "UPDATE peers_112 SET admitted=0 WHERE node_id=%lld", (long long)server->node_id);
|
||||
REQUIRE(sqlite3_exec(server->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK);
|
||||
uasync_drain_immediate(fixture.ua);
|
||||
check("MULTICAST rechecks self before deferred callback", !ctx->deliveries->count && received[0][1] == 1);
|
||||
check("revoked self cannot open or send MULTICAST", multicast_open(fixture.groups[0][1], multicast_lost, &losses[0][1]) < 0 &&
|
||||
multicast_send(server, descriptors[0][1], &server->node_id, 1, "denied", 6) < 0);
|
||||
topo_group_revalidate_members(fixture.groups[0][1]);
|
||||
check("MULTICAST self revoke closes B descriptor and local subscription", !local->ctx &&
|
||||
!mc_descriptor_find(ctx, descriptors[0][1]) && mc_descriptor_find(ctx, descriptors[0][0]));
|
||||
multicast_unsubscribe(local);
|
||||
for (unsigned i = 0; i < 2; i++) for (unsigned group = 0; group < 2; group++) {
|
||||
multicast_unsubscribe(subscriptions[i][group]);
|
||||
if (i || !group) multicast_close(fixture.nodes[i], descriptors[i][group]);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
int main(void) {
|
||||
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
|
||||
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
|
||||
@@ -590,7 +704,7 @@ int main(void) {
|
||||
debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_INFO);
|
||||
debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_DEBUG);
|
||||
}
|
||||
if (setup() || run() || invitation_isolation() || join_isolation() || nat_isolation()) fixture.failed++;
|
||||
if (setup() || run() || invitation_isolation() || join_isolation() || nat_isolation() || multicast_isolation()) fixture.failed++;
|
||||
else {
|
||||
for (unsigned i = 0; i < 2; i++) if (member_files_init(fixture.nodes[i])) fixture.failed++;
|
||||
if (!fixture.failed) {
|
||||
|
||||
@@ -143,6 +143,7 @@ static struct fixture* setup(void) {
|
||||
for (int j = 0; j < GROUPS; j++) {
|
||||
struct TOPO_GROUP* g = (struct TOPO_GROUP*)queue_entry_new(sizeof(*g) - sizeof(struct ll_entry)); CHECK(g);
|
||||
g->instance = inst; g->group_id = j ? 0x55 : TOPO_GROUP_UTUN;
|
||||
g->group_type = TOPO_GROUP_TYPE_UTUN; /* Транспортный стенд без CHAT-таблиц; допуск проверяет test_chat_group_isolation. */
|
||||
g->nodes = queue_new(f->ua, 256, 0, 8, "mc_test_nodes"); CHECK(g->nodes);
|
||||
g->senders_list = queue_new(f->ua, 0, 0, 0, "mc_test_peers"); CHECK(g->senders_list);
|
||||
CHECK(queue_data_put_with_index(inst->topo_groups->group_list, &g->ll) == 0);
|
||||
|
||||
Reference in new issue
Block a user