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

Test file revocation before and after disk execution starts

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 20:15:23 +01:00
1 parent 0edb47fb4f
commit abbf1cc60c
2 files changed
+156 -12
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/Makefile.am
+3
View File
@@ -883,6 +883,9 @@ test_broadcast_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
test_chat_group_isolation_SOURCES = test_chat_group_isolation.c test_membership.h test_net.h
test_chat_group_isolation_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
test_chat_group_isolation_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
if OS_LINUX
test_chat_group_isolation_LDFLAGS = -Wl,--wrap=mf_storage_execute
endif
test_conn_mgr_SOURCES = test_conn_mgr.c
test_conn_mgr_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
tests/test_chat_group_isolation.c
+153 -12
View File
@@ -4,11 +4,14 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include "utun_instance.h"
#include "config_updater.h"
#include "broadcast.h"
#include "chat/member_sync.h"
#include "radio/radio.h"
#include "member_files/private.h"
#include "member_files/disk.h"
#include "routing_layer/topo_group.h"
#include "routing_layer/topo_node_sqlite.h"
#include "routing_layer/conn_mgr_priv.h"
@@ -20,6 +23,7 @@
#include "../lib/mem.h"
#include "test_net.h"
#include "test_membership.h"
#include "test_utils.h"
#define REQUIRE(expr) do { if (!(expr)) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "isolation setup: %s", #expr); return -1; } } while (0)
@@ -34,7 +38,15 @@ static struct {
unsigned broadcast[2], radio[2], observed, passed, failed;
uint8_t expected[1024];
size_t expected_size;
} fixture;
char directory[512];
struct mf_pending* pending[2];
unsigned replies[2];
int status[2];
pthread_mutex_t disk_mutex;
pthread_cond_t disk_condition;
int disk_started, disk_released;
uint64_t block_request;
} fixture = { .disk_mutex = PTHREAD_MUTEX_INITIALIZER, .disk_condition = PTHREAD_COND_INITIALIZER };
/* Лог содержит и проверяемый эффект, и результат; отрицательный тест не обрывает соседние проверки. */
static void check(const char* name, int ok) {
@@ -119,15 +131,14 @@ static int send_radio(unsigned group, uint16_t stream) {
return inject(wire, sizeof(wire));
}
/* Подписанный запрос с настоящими эпохами уже открытого router-канала. */
static int send_direct(uint32_t request, const struct SVC_ROUTE_HDR* saved) {
/* Пакет со своими настоящими ключами и сохранёнными эпохами; локальный router атакующего обходится. */
static int send_routed(const struct SVC_ROUTE_HDR* saved, const void* payload, size_t bytes, uint8_t mode) {
struct UTUN_INSTANCE* source = fixture.nodes[1];
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));
uint8_t plain[768]; REQUIRE(sizeof(header) + bytes <= sizeof(plain));
memcpy(plain, &header, sizeof(header)); memcpy(plain + sizeof(header), payload, bytes);
uint8_t* encoded = NULL; size_t size = 0;
REQUIRE(!route_crypto_encode(source, header.dst_node_id, plain, sizeof(plain), ROUTE_CRYPTO_SIGN, &encoded, &size));
REQUIRE(!route_crypto_encode(source, header.dst_node_id, plain, sizeof(header) + bytes, mode, &encoded, &size));
uint8_t wire[1024];
if (size + 9 > sizeof(wire)) { u_free(encoded); DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "isolation encoded packet too large"); return -1; }
memcpy(wire, encoded, size); u_free(encoded);
@@ -137,6 +148,8 @@ static int send_direct(uint32_t request, const struct SVC_ROUTE_HDR* saved) {
/* Подписанный состав обоих каналов одинаков; отзыв позже меняется только у получателя. */
static int setup(void) {
snprintf(fixture.directory, sizeof(fixture.directory), "/tmp/utun_group_isolation_XXXXXX");
REQUIRE(!test_mkdtemp(fixture.directory));
struct SC_MYKEYS keys[2]; char public_keys[2][65], private_keys[2][65];
for (unsigned i = 0; i < 2; i++) {
REQUIRE(sc_generate_keypair(&keys[i]) == SC_OK);
@@ -147,9 +160,9 @@ static int setup(void) {
for (unsigned i = 0; i < 2; i++) {
char config[2048], client[256] = "";
if (i) snprintf(client, sizeof(client), "[client: peer]\npeer_public_key=%s\nlink=udp:127.0.0.1:0\n", public_keys[0]);
snprintf(config, sizeof(config), "[global]\nauto_sockets=no\nmy_public_key=%s\nmy_private_key=%s\n"
snprintf(config, sizeof(config), "[global]\nauto_sockets=no\nmy_public_key=%s\nmy_private_key=%s\ndb_path=%s/node%u\n"
"[server: udp]\naddr=127.0.0.1:0\ntype=public\n[allowed_keys]\nallow_all=1\n"
"[chatserver]\ngroup_autoconnect=0\n%s", public_keys[i], private_keys[i], client);
"[chatserver]\ngroup_autoconnect=0\n%s", public_keys[i], private_keys[i], fixture.directory, i, client);
fixture.nodes[i] = utun_instance_create_from_str(fixture.ua, config); REQUIRE(fixture.nodes[i]);
}
test_etcp_set_peer(fixture.nodes[1], "peer", fixture.nodes[0], "udp");
@@ -164,8 +177,12 @@ static int setup(void) {
owner->my_keys.public_key, NULL, owner->my_ed25519_pubkey, NULL, signature));
for (unsigned peer = 0; peer < 2; peer++) {
struct UTUN_INSTANCE* node = fixture.nodes[peer];
const char* tags = peer ? "{\"ver\":\"1\"}" : "{\"storage\":\"yes\",\"ver\":\"1\"}";
uint8_t message[256], tag_signature[64];
int length = member_sync_build_owner_msg(node->node_id, tags, message, sizeof(message));
REQUIRE(length > 0 && sc_ed25519_sign(owner->my_ed25519_privkey, message, length, tag_signature) == SC_OK);
REQUIRE(test_member_seed(inst, channel, node->my_keys.public_key, node->my_ed25519_pubkey,
node->my_ed25519_privkey, owner->my_ed25519_privkey, "{}", NULL, NULL) >= 0);
node->my_ed25519_privkey, owner->my_ed25519_privkey, "{}", tags, tag_signature) >= 0);
}
fixture.groups[i][group] = topo_groups_create_group(inst->topo_groups, 111 + group, TOPO_GROUP_TYPE_CHAT, channel);
REQUIRE(fixture.groups[i][group]); fixture.groups[i][group]->radio_active = 1;
@@ -232,7 +249,8 @@ 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(2, &saved));
struct CM_DIRECT_REQ forbidden = {ETCP_RT_ID_CONN_MGR, CM_SUBCMD_DIRECT_REQ, 2};
REQUIRE(!send_routed(&saved, &forbidden, sizeof(forbidden), ROUTE_CRYPTO_SIGN));
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);
@@ -265,6 +283,111 @@ static int run(void) {
return 0;
}
/* В отличие от задержки по времени барьер доказывает фазу work перед отзывом. */
static void disk_block(void* arg) {
(void)arg;
pthread_mutex_lock(&fixture.disk_mutex);
fixture.disk_started = 1; pthread_cond_signal(&fixture.disk_condition);
while (!fixture.disk_released) pthread_cond_wait(&fixture.disk_condition, &fixture.disk_mutex);
pthread_mutex_unlock(&fixture.disk_mutex);
}
static void disk_done(void* arg, int error) { (void)arg; check("disk barrier completed", !error); }
#ifdef __linux__
void __real_mf_storage_execute(struct mf_storage*, uint64_t, uint64_t, const struct mf_message*, struct mf_message*);
void __wrap_mf_storage_execute(struct mf_storage* storage, uint64_t group, uint64_t peer,
const struct mf_message* request, struct mf_message* response) {
pthread_mutex_lock(&fixture.disk_mutex);
if (fixture.block_request && request->request_id == fixture.block_request) {
fixture.disk_started = 1; pthread_cond_signal(&fixture.disk_condition);
while (!fixture.disk_released) pthread_cond_wait(&fixture.disk_condition, &fixture.disk_mutex);
}
pthread_mutex_unlock(&fixture.disk_mutex);
__real_mf_storage_execute(storage, group, peer, request, response);
}
#endif
static void file_response(void* arg, const struct mf_message* response) {
unsigned index = (unsigned)(uintptr_t)arg;
fixture.pending[index] = NULL; fixture.replies[index]++; fixture.status[index] = response->status;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "isolation file response group=%u status=%s bytes=%u", 111 + index,
mf_status_text(response->status), response->data_len);
}
static int file_isolation(unsigned running) {
struct UTUN_INSTANCE* server = fixture.nodes[0];
struct UTUN_INSTANCE* client = fixture.nodes[1];
struct member_files* files = server->member_files;
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() + 30000;
while ((!topo_group_peer_ready(fixture.groups[0][1], client->node_id) ||
!topo_group_peer_ready(fixture.groups[1][1], server->node_id) || files->worker) && get_time_tb() < deadline) uasync_poll(fixture.ua, 1);
REQUIRE(topo_group_peer_ready(fixture.groups[0][1], client->node_id) && !files->worker);
pthread_mutex_lock(&fixture.disk_mutex);
fixture.disk_started = fixture.disk_released = 0; fixture.block_request = running ? 6000 + running : 0;
pthread_mutex_unlock(&fixture.disk_mutex);
if (!running) {
mf_worker_submit(files->disk, disk_block, NULL, disk_done, NULL);
pthread_mutex_lock(&fixture.disk_mutex);
while (!fixture.disk_started) pthread_cond_wait(&fixture.disk_condition, &fixture.disk_mutex);
pthread_mutex_unlock(&fixture.disk_mutex);
}
struct mf_message request = { .command = MF_MKDIR, .request_id = 6000 + running, .owner = client->node_id };
snprintf(request.path, sizeof(request.path), "revoke_%u", running);
unsigned before = fixture.replies[1];
fixture.pending[1] = member_files_request(client, 112, server->node_id, &request, file_response, (void*)(uintptr_t)1);
REQUIRE(fixture.pending[1]);
deadline = get_time_tb() + 30000;
for (;;) {
pthread_mutex_lock(&fixture.disk_mutex); int started = fixture.disk_started; pthread_mutex_unlock(&fixture.disk_mutex);
if (files->worker && started) break;
REQUIRE(get_time_tb() < deadline); uasync_poll(fixture.ua, 1);
}
struct ETCP_ROUTER_CONN* route = etcp_router_conn_get(server, 112, client->node_id, ETCP_RT_ID_MEMBER_FILES); REQUIRE(route);
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_MEMBER_FILES, .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)client->node_id);
REQUIRE(sqlite3_exec(server->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK);
topo_group_revalidate_members(fixture.groups[0][1]);
if (!running) {
deadline = get_time_tb() + 30000;
while (files->worker && get_time_tb() < deadline) uasync_poll(fixture.ua, 1);
check("queued file mutation cancelled before disk release", !files->worker && !files->job_count);
}
struct mf_message attack = request; attack.kind = MF_REQUEST; attack.request_id += 100; snprintf(attack.path, sizeof(attack.path), "forbidden_%u", running);
uint8_t wire[MF_WIRE_MAX]; int bytes = mf_encode(&attack, wire, sizeof(wire)); REQUIRE(bytes > 0);
unsigned jobs = files->job_count;
REQUIRE(!send_routed(&saved, wire, (size_t)bytes, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT));
check("revoked signed file request cannot create a job", files->job_count == jobs);
request.request_id += 200;
unsigned good_before = fixture.replies[0];
fixture.pending[0] = member_files_request(client, 111, server->node_id, &request, file_response, (void*)(uintptr_t)0);
REQUIRE(fixture.pending[0]);
pthread_mutex_lock(&fixture.disk_mutex); fixture.disk_released = 1; pthread_cond_signal(&fixture.disk_condition);
pthread_mutex_unlock(&fixture.disk_mutex);
deadline = get_time_tb() + 30000;
while ((fixture.replies[0] == good_before || files->worker) && get_time_tb() < deadline) uasync_poll(fixture.ua, 1);
check("neighbouring file group works on the same ETCP", fixture.replies[0] == good_before + 1 && fixture.status[0] == MF_OK &&
fixture.transport == instance_find_conn(client, server->node_id) && fixture.transport->links_up);
check("revoked file command produces no response", fixture.replies[1] == before);
char root_path[768], path[128]; snprintf(root_path, sizeof(root_path), "%s/node0/member_files", fixture.directory);
struct mf_disk_root root; struct mf_disk_info info;
REQUIRE(mf_disk_root_open(&root, root_path) == MF_OK);
snprintf(path, sizeof(path), "112/%llu/revoke_%u", (unsigned long long)client->node_id, running);
int result = mf_disk_stat(&root, path, &info);
check(running ? "started file command finishes after revoke" : "queued file command has no filesystem effect",
result == (running ? MF_OK : MF_NOT_FOUND));
snprintf(path, sizeof(path), "112/%llu/forbidden_%u", (unsigned long long)client->node_id, running);
check("forbidden file packet has no filesystem effect", mf_disk_stat(&root, path, &info) == MF_NOT_FOUND);
mf_disk_root_close(&root);
if (fixture.pending[1]) member_files_cancel(fixture.pending[1]);
return 0;
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
@@ -274,9 +397,27 @@ int main(void) {
debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_INFO);
}
if (setup() || run()) fixture.failed++;
else {
for (unsigned i = 0; i < 2; i++) if (member_files_init(fixture.nodes[i])) fixture.failed++;
if (!fixture.failed) {
uint64_t deadline = get_time_tb() + 30000;
while (get_time_tb() < deadline && (fixture.nodes[0]->member_files->storage_status == MF_BUSY ||
fixture.nodes[1]->member_files->storage_status == MF_BUSY)) uasync_poll(fixture.ua, 1);
if (fixture.nodes[0]->member_files->storage_status != MF_OK || fixture.nodes[1]->member_files->storage_status != MF_OK ||
file_isolation(0)) fixture.failed++;
#ifdef __linux__
if (!fixture.failed && file_isolation(1)) fixture.failed++;
#endif
}
}
pthread_mutex_lock(&fixture.disk_mutex); fixture.disk_released = 1; pthread_cond_signal(&fixture.disk_condition);
pthread_mutex_unlock(&fixture.disk_mutex);
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]); }
for (unsigned i = 0; i < 2; i++) if (fixture.nodes[i]) {
member_files_destroy(fixture.nodes[i]); member_sync_destroy(fixture.nodes[i]); utun_instance_destroy(fixture.nodes[i]);
}
if (fixture.ua) uasync_destroy(fixture.ua, 0);
pthread_cond_destroy(&fixture.disk_condition); pthread_mutex_destroy(&fixture.disk_mutex);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "CHAT isolation: passed=%u failed=%u", fixture.passed, fixture.failed);
return fixture.failed ? 1 : 0;
}