jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Detach revoked CM owners before callbacks and test group recovery

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 19:58:33 +01:00
1 parent f30ef75384
commit 6965e11fe1
2 files changed
+54 -8
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/routing_layer/conn_mgr_core.c
+7 -3
View File
@@ -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,
tests/test_chat_group_isolation.c
+47 -5
View File
@@ -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);