diff --git a/src/routing_layer/conn_mgr_core.c b/src/routing_layer/conn_mgr_core.c index 1886d2f0b..37fca430b 100644 --- a/src/routing_layer/conn_mgr_core.c +++ b/src/routing_layer/conn_mgr_core.c @@ -48,14 +48,18 @@ static int cm_peer_allowed(struct CONN_MGR* mgr, uint64_t node) { return 0; } -/* Сначала запретить UP/повторы, затем уведомить владельцев; callbacks вправе закрыть handles и соседние entries. */ +/* Сначала отсоединить владельцев и снять entry, затем callbacks: повторный open не присоединится к отозванной попытке. */ static void cm_revoke(struct CONN_MGR_ENTRY* entry) { if (!entry || entry->state == CONN_MGR_STATE_DISCONNECTED) return; DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: revoke group=%016llx peer=%016llx handles=%d", (unsigned long long)entry->mgr->group->group_id, (unsigned long long)entry->node_id, entry->handles != NULL); - entry->state = CONN_MGR_STATE_DISCONNECTED; - cm_deliver_event(entry, CONN_EVENT_TIMEOUT); + struct CONN_MGR_HANDLE* handle = entry->handles; cm_entry_cleanup(entry); + while (handle) { + struct CONN_MGR_HANDLE* next = handle->next; + if (!handle->closed && handle->cb) handle->cb(handle, handle->node_id, handle->group_id, CONN_EVENT_TIMEOUT, handle->cb_arg); + handle = next; + } } static struct CONN_MGR_HANDLE* cm_handle_new(struct CONN_MGR_ENTRY* entry, uint64_t group_id, diff --git a/tests/test_chat_group_isolation.c b/tests/test_chat_group_isolation.c index 4cf39c250..77fafecec 100644 --- a/tests/test_chat_group_isolation.c +++ b/tests/test_chat_group_isolation.c @@ -28,6 +28,8 @@ static struct { struct UTUN_INSTANCE* nodes[2]; struct TOPO_GROUP* groups[2][2]; struct ETCP_CONN* transport; + struct CONN_MGR_HANDLE* handles[2]; + unsigned up[2], denied[2]; etcp_recv_fn receivers[256]; unsigned broadcast[2], radio[2], observed, passed, failed; uint8_t expected[1024]; @@ -40,6 +42,17 @@ static void check(const char* name, int ok) { else { fixture.failed++; DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "FAIL %s", name); } } +/* TIMEOUT при отзыве должен прийти уже отсоединённому handle; другой канал продолжает получать UP. */ +static void connection_event(struct CONN_MGR_HANDLE* handle, uint64_t node, uint64_t group, enum conn_mgr_event event, void* arg) { + (void)node; (void)group; + unsigned index = (unsigned)(uintptr_t)arg; + if (event == CONN_EVENT_UP) fixture.up[index]++; + if (event == CONN_EVENT_TIMEOUT) { + fixture.denied[index]++; + check("revocation callback observes detached CM handle", !handle->entry); + } +} + /* Копия наблюдаемого пакета сохраняется до отправки; настоящий dispatcher не заменяется заглушкой. */ static void observe(struct ETCP_CONN* conn, struct ll_entry* entry) { uint8_t service = entry->dgram[0]; @@ -107,11 +120,9 @@ static int send_radio(unsigned group, uint16_t stream) { } /* Подписанный запрос с настоящими эпохами уже открытого router-канала. */ -static int send_direct(unsigned group, uint32_t request, struct ETCP_ROUTER_CONN* receiver) { +static int send_direct(uint32_t request, const struct SVC_ROUTE_HDR* saved) { struct UTUN_INSTANCE* source = fixture.nodes[1]; - struct SVC_ROUTE_HDR header = {.cmd = ETCP_ID_SVC_ROUTE, .group_id = 111 + group, - .src_node_id = source->node_id, .dst_node_id = fixture.nodes[0]->node_id, .svc_id = ETCP_RT_ID_CONN_MGR, - .reset_id = receiver->peer_reset_id, .peer_reset_id = receiver->reset_id, .seq = receiver->rx_seq}; + struct SVC_ROUTE_HDR header = *saved; struct CM_DIRECT_REQ message = {ETCP_RT_ID_CONN_MGR, CM_SUBCMD_DIRECT_REQ, request}; uint8_t plain[sizeof(header) + sizeof(message)]; memcpy(plain, &header, sizeof(header)); memcpy(plain + sizeof(header), &message, sizeof(message)); @@ -205,6 +216,13 @@ static int run(void) { struct ETCP_ROUTER_CONN* route = etcp_router_conn_get(server, 112, peer, ETCP_RT_ID_CONN_MGR); REQUIRE(route && route->peer_reset_id); check("admitted DIRECT_REQ changes only B manager", !cm_find_entry(fixture.groups[0][0]->conn_mgr, peer)); + for (unsigned group = 0; group < 2; group++) + REQUIRE(!conn_mgr_open(server, 111 + group, peer, connection_event, (void*)(uintptr_t)group, &fixture.handles[group])); + deadline = get_time_tb() + 30000; + while ((!fixture.up[0] || !fixture.up[1]) && get_time_tb() < deadline) uasync_poll(fixture.ua, 1); + REQUIRE(fixture.up[0] == 1 && fixture.up[1] == 1); + struct SVC_ROUTE_HDR saved = {.cmd = ETCP_ID_SVC_ROUTE, .group_id = 112, .src_node_id = peer, .dst_node_id = server->node_id, + .svc_id = ETCP_RT_ID_CONN_MGR, .reset_id = route->peer_reset_id, .peer_reset_id = route->reset_id, .seq = route->rx_seq}; char sql[160]; snprintf(sql, sizeof(sql), "UPDATE peers_112 SET admitted=0 WHERE node_id=%lld", (long long)peer); REQUIRE(sqlite3_exec(server->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK); REQUIRE(topo_group_chat_allowed(server, 111, peer) && !topo_group_chat_allowed(server, 112, peer)); @@ -214,13 +232,36 @@ static int run(void) { check("revoked peer cannot deliver BROADCAST in B", fixture.broadcast[1] == 1); REQUIRE(!send_radio(1, 2)); check("revoked peer cannot deliver RADIO in B", fixture.radio[1] == 1); - REQUIRE(!send_direct(1, 2, route)); + REQUIRE(!send_direct(2, &saved)); state = cm_find_entry(manager, peer); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "isolation CM side effect group=112 request=%u", state ? state->peer_reverse_request_id : 0); check("revoked signed DIRECT_REQ cannot change B manager", !state || state->peer_reverse_request_id == 1); + struct CONN_MGR_HANDLE* denied = NULL; + int opened = conn_mgr_open(server, 112, peer, connection_event, (void*)(uintptr_t)1, &denied); + check("local CM open cannot bypass revoked CHAT admission", opened == -1 && !denied); + if (denied) conn_mgr_close(denied); + topo_group_revalidate_members(fixture.groups[0][1]); + check("proactive revoke cancels B owner and retains A owner", fixture.denied[1] == 1 && !fixture.handles[1]->entry + && !fixture.denied[0] && fixture.handles[0]->entry && conn_mgr_get_conn(fixture.handles[0])); REQUIRE(!send_broadcast(0, 3) && !send_radio(0, 3)); check("neighbouring A and the same ETCP still work", fixture.broadcast[0] == 1 && fixture.radio[0] == 1 && fixture.transport == instance_find_conn(fixture.nodes[1], server->node_id) && fixture.transport->links_up); + /* Повторить отклонённые UUID/stream после нового допуска: запрещённый пакет не должен загрязнить dedup. */ + snprintf(sql, sizeof(sql), "UPDATE peers_112 SET admitted=1 WHERE node_id=%lld", (long long)peer); + REQUIRE(sqlite3_exec(server->topo_sqlite_db, sql, 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))); + deadline = get_time_tb() + 50000; + while ((!topo_group_peer_ready(fixture.groups[0][1], peer) + || !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], peer) && topo_group_peer_ready(fixture.groups[1][1], server->node_id)); + REQUIRE(!send_broadcast(1, 2) && !send_radio(1, 2)); + check("denied packets did not poison BROADCAST/RADIO dedup", fixture.broadcast[1] == 2 && fixture.radio[1] == 2); + 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); + uint8_t data = 1; + check("revoked self cannot start BROADCAST or RADIO", broadcast_send(fixture.groups[0][1], &data, 1) == -1 + && radio_talk_begin(server, 112) == -1); return 0; } @@ -233,6 +274,7 @@ int main(void) { debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_INFO); } if (setup() || run()) fixture.failed++; + for (unsigned i = 0; i < 2; i++) if (fixture.handles[i]) conn_mgr_close(fixture.handles[i]); for (unsigned i = 0; i < 2; i++) if (fixture.nodes[i]) { member_sync_destroy(fixture.nodes[i]); utun_instance_destroy(fixture.nodes[i]); } if (fixture.ua) uasync_destroy(fixture.ua, 0); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "CHAT isolation: passed=%u failed=%u", fixture.passed, fixture.failed);