433 lines
27 KiB
C
433 lines
27 KiB
C
/* Две CHAT-группы на одном настоящем ETCP: побочные эффекты после локального отзыва допуска.
|
|
* Пакеты атакующего идут через etcp_send, минуя только его исходящую авторизацию. */
|
|
#include <stdint.h>
|
|
#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"
|
|
#include "routing_layer/etcp_router.h"
|
|
#include "routing_layer/route_crypto.h"
|
|
#include "transport_layer/etcp.h"
|
|
#include "transport_layer/etcp_api.h"
|
|
#include "../lib/debug_config.h"
|
|
#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)
|
|
|
|
static struct {
|
|
struct UASYNC* ua;
|
|
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];
|
|
size_t expected_size;
|
|
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) {
|
|
if (ok) { fixture.passed++; DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS %s", name); }
|
|
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];
|
|
if (entry->len == fixture.expected_size && !memcmp(entry->dgram, fixture.expected, entry->len)) {
|
|
fixture.observed++;
|
|
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "isolation wire delivered service=%u peer=%016llx bytes=%u", service,
|
|
(unsigned long long)conn->peer_node_id, entry->len);
|
|
}
|
|
fixture.receivers[service](conn, entry);
|
|
}
|
|
|
|
/* Считаем реальную выдачу данных пользовательскому callback группы. */
|
|
static void broadcast_received(const uint8_t* uuid, const uint8_t* data, uint16_t len, void* arg) {
|
|
(void)uuid; (void)data; (void)len;
|
|
unsigned group = (unsigned)(uintptr_t)arg;
|
|
fixture.broadcast[group]++;
|
|
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "isolation broadcast callback group=%u count=%u", 111 + group, fixture.broadcast[group]);
|
|
}
|
|
|
|
static void radio_received(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t source, uint16_t stream,
|
|
uint16_t seq, uint8_t fin, const uint8_t* opus, int len, void* arg) {
|
|
(void)inst; (void)stream; (void)seq; (void)fin; (void)opus; (void)len; (void)arg;
|
|
if (group < 111 || group > 112) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "unexpected radio group=%llu", (unsigned long long)group); return; }
|
|
fixture.radio[group - 111]++;
|
|
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "isolation radio callback group=%llu source=%016llx count=%u",
|
|
(unsigned long long)group, (unsigned long long)source, fixture.radio[group - 111]);
|
|
}
|
|
|
|
/* Дождаться не только wire, но и отложенной обработки router/service очередей. */
|
|
static int inject(const void* wire, size_t size) {
|
|
REQUIRE(size <= sizeof(fixture.expected));
|
|
struct ll_entry* entry = ll_alloc_lldgram(size);
|
|
REQUIRE(entry);
|
|
memcpy(entry->dgram, wire, size); entry->len = size;
|
|
memcpy(fixture.expected, wire, size); fixture.expected_size = size;
|
|
unsigned before = fixture.observed;
|
|
if (etcp_send(fixture.transport, entry)) {
|
|
queue_dgram_free(entry); queue_entry_free(entry);
|
|
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "isolation injection failed bytes=%zu", size); return -1;
|
|
}
|
|
uint64_t deadline = get_time_tb() + 30000;
|
|
while (fixture.observed == before && get_time_tb() < deadline) uasync_poll(fixture.ua, 1);
|
|
REQUIRE(fixture.observed == before + 1);
|
|
for (unsigned i = 0; i < 20; i++) uasync_poll(fixture.ua, 1);
|
|
return 0;
|
|
}
|
|
|
|
static int send_broadcast(unsigned group, unsigned id) {
|
|
uint8_t wire[28] = {ETCP_ID_BROADCAST}; uint64_t gid = 111 + group;
|
|
memcpy(wire + 1, &gid, 8); memcpy(wire + 9, &id, sizeof(id));
|
|
wire[25] = 1; wire[27] = 0x5a;
|
|
return inject(wire, sizeof(wire));
|
|
}
|
|
|
|
static int send_radio(unsigned group, uint16_t stream) {
|
|
uint8_t wire[1 + RADIO_HDR_SIZE + 8 + 3] = {ETCP_ID_RADIO};
|
|
uint8_t* payload = wire + 1; uint64_t gid = 111 + group, source = fixture.nodes[1]->node_id;
|
|
uint16_t size = 1;
|
|
memcpy(payload + RADIO_OFF_GROUP, &gid, 8); memcpy(payload + RADIO_OFF_SRC, &source, 8);
|
|
memcpy(payload + RADIO_OFF_STREAM, &stream, 2);
|
|
payload[RADIO_OFF_HOPS] = payload[RADIO_OFF_COUNT] = 1;
|
|
memcpy(payload + RADIO_OFF_HOPLIST, &source, 8);
|
|
memcpy(payload + RADIO_HDR_SIZE + 8, &size, 2); payload[RADIO_HDR_SIZE + 10] = 0x5a;
|
|
return inject(wire, sizeof(wire));
|
|
}
|
|
|
|
/* Пакет со своими настоящими ключами и сохранёнными эпохами; локальный 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;
|
|
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(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);
|
|
memcpy(wire + size, &source->node_id, 8); wire[size + 8] = 1;
|
|
return inject(wire, size + 9);
|
|
}
|
|
|
|
/* Подписанный состав обоих каналов одинаков; отзыв позже меняется только у получателя. */
|
|
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);
|
|
bytes_to_hex(keys[i].public_key, 32, public_keys[i], 65);
|
|
bytes_to_hex(keys[i].private_key, 32, private_keys[i], 65);
|
|
}
|
|
fixture.ua = uasync_create(); REQUIRE(fixture.ua);
|
|
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\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], 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");
|
|
for (unsigned i = 0; i < 2; i++) {
|
|
struct UTUN_INSTANCE* inst = fixture.nodes[i];
|
|
REQUIRE(!utun_core_start(inst) && !member_sync_init(inst));
|
|
for (unsigned group = 0; group < 2; group++) {
|
|
char channel[32]; snprintf(channel, sizeof(channel), "%u", 111 + group);
|
|
struct UTUN_INSTANCE* owner = fixture.nodes[0]; uint8_t signature[64];
|
|
REQUIRE(sc_ed25519_sign(owner->my_ed25519_privkey, (const uint8_t*)channel, strlen(channel), signature) == SC_OK);
|
|
REQUIRE(!topo_node_sqlite_channel_put(inst->topo_sqlite_db, channel, "isolation", owner->node_id,
|
|
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, "{}", 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;
|
|
}
|
|
}
|
|
uint64_t deadline = get_time_tb() + 50000;
|
|
for (;;) {
|
|
fixture.transport = instance_find_conn(fixture.nodes[1], fixture.nodes[0]->node_id);
|
|
if (fixture.transport && fixture.transport->links_up) break;
|
|
REQUIRE(get_time_tb() < deadline); uasync_poll(fixture.ua, 1);
|
|
}
|
|
for (unsigned i = 0; i < 2; i++) for (unsigned group = 0; group < 2; group++) {
|
|
struct ETCP_CONN* conn = instance_find_conn(fixture.nodes[i], fixture.nodes[1 - i]->node_id); REQUIRE(conn);
|
|
REQUIRE(!topo_group_new_conn(fixture.groups[i][group], conn));
|
|
}
|
|
deadline = get_time_tb() + 50000;
|
|
for (;;) {
|
|
unsigned ready = 0;
|
|
for (unsigned i = 0; i < 2; i++) for (unsigned group = 0; group < 2; group++)
|
|
ready += topo_group_peer_ready(fixture.groups[i][group], fixture.nodes[1 - i]->node_id);
|
|
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};
|
|
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;
|
|
}
|
|
for (unsigned group = 0; group < 2; group++) broadcast_add_cbk(fixture.groups[0][group], broadcast_received, (void*)(uintptr_t)group);
|
|
radio_set_frame_cb(fixture.nodes[0], radio_received, NULL);
|
|
return 0;
|
|
}
|
|
|
|
/* Отзыв без proactive cleanup проверяет сам вход: старый транспорт и сессии намеренно остаются живы. */
|
|
static int run(void) {
|
|
struct UTUN_INSTANCE* server = fixture.nodes[0]; uint64_t peer = fixture.nodes[1]->node_id;
|
|
REQUIRE(!send_broadcast(1, 1) && !send_radio(1, 1));
|
|
check("admitted peer receives BROADCAST and RADIO in B", fixture.broadcast[1] == 1 && fixture.radio[1] == 1);
|
|
struct CM_DIRECT_REQ warmup = {ETCP_RT_ID_CONN_MGR, CM_SUBCMD_DIRECT_REQ, 1};
|
|
struct ll_entry* entry = ll_alloc_lldgram(1 + sizeof(warmup)); REQUIRE(entry);
|
|
entry->dgram[0] = ETCP_RT_ID_CONN_MGR; memcpy(entry->dgram + 1, &warmup, sizeof(warmup)); entry->len = 1 + sizeof(warmup);
|
|
REQUIRE(!etcp_route_send(fixture.nodes[1], 112, server->node_id, entry, 1, ROUTE_CRYPTO_SIGN));
|
|
struct CONN_MGR* manager = fixture.groups[0][1]->conn_mgr; struct CONN_MGR_ENTRY* state = NULL;
|
|
uint64_t deadline = get_time_tb() + 30000;
|
|
do { uasync_poll(fixture.ua, 1); state = cm_find_entry(manager, peer); }
|
|
while ((!state || state->peer_reverse_request_id != 1) && get_time_tb() < deadline);
|
|
REQUIRE(state && state->peer_reverse_request_id == 1);
|
|
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));
|
|
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "isolation revoked group=112 peer=%016llx shared_transport=%p", (unsigned long long)peer,
|
|
(void*)fixture.transport);
|
|
REQUIRE(!send_broadcast(1, 2));
|
|
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);
|
|
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);
|
|
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;
|
|
}
|
|
|
|
/* В отличие от задержки по времени барьер доказывает фазу 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) {
|
|
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "isolation running disk command blocked group=%llu id=%llu",
|
|
(unsigned long long)group, (unsigned long long)request->request_id);
|
|
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]);
|
|
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "isolation waiting for disk command running=%u id=%llu", running, (unsigned long long)request.request_id);
|
|
/* После readmission router меняет эпоху и очищает старый packet; MF повторяет запрос через5с плюс tick1с. */
|
|
deadline = get_time_tb() + 70000;
|
|
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);
|
|
if (getenv("UTUN_TEST_DEBUG")) {
|
|
debug_set_category_level(DEBUG_CATEGORY_GENERAL, DEBUG_LEVEL_DEBUG);
|
|
debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_DEBUG);
|
|
debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_INFO);
|
|
debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_DEBUG);
|
|
}
|
|
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);
|
|
/* На сервере в этом сценарии только проверяемая команда: sweep не занимает f->worker вместо неё. */
|
|
struct member_files* files = fixture.nodes[0]->member_files;
|
|
if (files->tick) uasync_cancel_timeout(fixture.ua, files->tick);
|
|
files->tick = NULL;
|
|
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_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;
|
|
}
|