utun2/tests/test_group_connect.c

#include <assert.h>
#include <stdio.h>
#include <string.h>
#include <openssl/sha.h>
#include "utun_instance.h"
#include "routing_layer/topo_group.h"
#include "routing_layer/topo_group_connect.h"
#include "routing_layer/topo_node_sqlite.h"
#include "transport_layer/etcp.h"
#include "transport_layer/etcp_api.h"
#include "transport_layer/node_conn_direct.h"
#include "../lib/mem.h"
#include "../lib/memory_pool.h"
#include "../lib/debug_config.h"
#ifdef UTUN_HAVE_STANDBY
#include "standby.h"
#endif

#define PEERS 8

struct fixture {
    struct UASYNC* ua;
    struct UTUN_INSTANCE* inst;
    struct TOPO_GROUP* group;
    uint64_t ids[PEERS];
    struct ETCP_CONN* conns[PEERS];
    struct NODE_CONN_DIRECT* owners[PEERS];
};

static struct TOPO_GROUP_CONN_ITEM* peer(struct fixture* f, int index) {
    for (struct ll_entry* e = f->group->senders_list->head; e; e = e->next) {
        struct TOPO_GROUP_CONN_ITEM* p = (struct TOPO_GROUP_CONN_ITEM*)e->data;
        if (p->node_id == f->ids[index]) return p;
    }
    return NULL;
}

static void poll_events(struct fixture* f) { for (int i = 0; i < 8; i++) uasync_poll(f->ua, 0); }

/* Типы/роли реальные, транспортные события задаём явно без удалённых процессов. */
static void create(struct fixture* f, int device, int supernode) {
    memset(f, 0, sizeof(*f));
    f->ua = uasync_create(); assert(f->ua);
#ifdef UTUN_HAVE_STANDBY
    standby_init(f->ua);
#endif
    char config[512];
    snprintf(config, sizeof(config), "[global]\nclient_type=%d\n"
        "my_private_key=704f2e012c8fa8768130cb0f988a997dccb628372bc5ceccacc78dcbfec5916f\n"
        "my_public_key=b3193173def895bd0fcea6f86af077c7d77216f10395275f627ac18242ec0f01\n"
        "[server: udp]\naddr=127.0.0.1:0\ntype=private\n"
        "[chatserver]\ngroup_autoconnect=0\n", device);
    f->inst = utun_instance_create_from_str(f->ua, config);
    assert(f->inst && utun_instance_init(f->inst) == 0);
    sqlite3* db = f->inst->topo_sqlite_db;
    uint8_t key[32] = {1}, sig[64] = {1};
    assert(topo_node_sqlite_channel_put(db, "42", "connect", f->inst->node_id, key, NULL, key, NULL, sig) == 0);
    assert(topo_node_sqlite_member_block_put(db, "42", f->inst->node_id, sig, 1, sig, 1,
        f->inst->my_keys.public_key, f->inst->my_ed25519_pubkey, "{}", NULL, 0, sig) == 0);
    if (supernode) assert(topo_node_sqlite_member_owner_put(db, "42", f->inst->node_id, "{\"supernode\":\"yes\"}", sig, 0) == 0);
    f->group = topo_groups_create_group(f->inst->topo_groups, 42, TOPO_GROUP_TYPE_CHAT, "42"); assert(f->group);
    for (int i = 0; i < PEERS; i++) {
        struct SC_MYKEYS keys; assert(sc_generate_keypair(&keys) == SC_OK);
        struct TOPO_NODE* node = u_calloc(1, sizeof(*node)); assert(node);
        memcpy(node->public_key, keys.public_key, 32);
        assert(sc_derive_ed25519_pubkey(keys.private_key, node->ed25519_public_key) == SC_OK);
        f->ids[i] = node->node_id = sc_derive_node_id_from_pubkey(keys.public_key);
        node->timestamp = 1;
        node->client_type = i < 5 ? CLIENT_TYPE_DESKTOP : i < 7 ? CLIENT_TYPE_MOBILE : CLIENT_TYPE_SERVER;
        node->v4_addrs = memory_pool_alloc(f->inst->topo_groups->v4_addr_pool); assert(node->v4_addrs);
        *node->v4_addrs = (struct TOPO_ADDR4){ .addr = {127,0,0,1}, .port = (uint16_t)(21000 + i),
                                            .protocol = TOPO_PROTO_UDP, .type = TOPO_ADDR_INTERFACE };
        uint8_t message[TOPO_SIG_MSG_MAX_SIZE];
        int size = topo_node_build_sig_msg(node, message, sizeof(message)); assert(size > 0);
        uint8_t ed_private[SHA512_DIGEST_LENGTH];
        SHA512(keys.private_key, SC_PRIVKEY_SIZE, ed_private);
        assert(sc_ed25519_sign(ed_private, message, (size_t)size, node->x25519_self_sig) == SC_OK);
        assert(topo_node_verify(node) == 0);
        assert(topo_node_registry_store(f->inst->topo_groups, node));
        assert(topo_node_sqlite_node_update_verified(db, node->node_id, "peer", node->public_key,
                                                   node->ed25519_public_key, 1, 1) == 0);
        assert(topo_node_sqlite_snapshot_put(db, node) == 0);
        assert(topo_node_sqlite_addrs_put(db, node->node_id, node) == 0);
        assert(topo_node_sqlite_member_block_put(db, "42", node->node_id, sig, 1, sig, 1,
                                               node->public_key, node->ed25519_public_key, "{}", NULL, 0, sig) == 0);
        if (i < 3) assert(topo_node_sqlite_member_owner_put(db, "42", node->node_id, "{\"supernode\":\"yes\"}", sig, 0) == 0);
        assert(topo_node_sqlite_nodeinfo_updated(db, node->node_id) == 0);
        topo_node_sqlite_update_rtt(db, node->node_id, (uint16_t)(i + 1));
        /* История нижнего класса не должна обходить суперузлы. */
        if (i >= 3) assert(topo_node_sqlite_set_connected(db, "42", node->node_id, 1) == 0);
        struct ETCP_CONN* conn = f->conns[i] = etcp_connection_create(f->inst, "connect_fixture"); assert(conn);
        conn->peer_node_id = node->node_id;
        assert(sc_init_ctx(&conn->crypto_ctx, &f->inst->my_keys) == SC_OK);
        assert(sc_set_peer_public_key(&conn->crypto_ctx, node->public_key, SC_PEER_PUBKEY_BIN) == SC_OK);
        queue_set_callback(conn->send_input_q, NULL, NULL);
        etcp_conn_ready(conn);
        /* Другой владелец позволяет проверить отмену без закрытия общего транспорта. */
        assert(node_conn_direct_open(f->inst, node->node_id, NULL, NULL, &f->owners[i], NULL) != NCD_ERR);
    }
}

static void destroy(struct fixture* f) {
#ifdef UTUN_HAVE_STANDBY
    standby_deinit();
#endif
    topo_groups_remove_group(f->inst->topo_groups, 42);
    for (int i = 0; i < PEERS; i++) {
        node_conn_direct_close(f->owners[i]);
        topo_node_registry_unref(f->inst->topo_groups, f->ids[i]);
    }
    f->inst->running = 0; utun_instance_destroy(f->inst);
    uasync_poll(f->ua, 0); uasync_destroy(f->ua, 0);
}

/* UP без JOIN_ACCEPT/TABLE_COMPLETE обязан завершить автоматическую попытку. */
static void up(struct fixture* f, int index) {
    f->conns[index]->links_up = 1;
    etcp_cbk_fire(f->conns[index], ETCP_CBK_EVENT_UP); poll_events(f);
    assert(peer(f, index) && peer(f, index)->retained && !peer(f, index)->requests);
    assert(!topo_group_peer_ready(f->group, f->ids[index]));
}

static void down(struct fixture* f, int index) {
    f->conns[index]->links_up = 0;
    etcp_cbk_fire(f->conns[index], ETCP_CBK_EVENT_DOWN); poll_events(f);
}

/* Потеря транспорта до первого UP должна разрешить следующий класс. */
static void fail(struct fixture* f, int index) {
    assert(peer(f, index) && peer(f, index)->requests);
    topo_group_remove_conn(f->group, f->conns[index], TOPO_REMOVE_TRANSPORT_DOWN); poll_events(f);
    assert(!peer(f, index));
    assert(!f->conns[index]->close_requested);
}

static void start(struct fixture* f) {
    assert(topo_group_connect_init(f->group) == 0); poll_events(f);
    assert(peer(f, 0) && peer(f, 1) && !peer(f, 2));
    for (int i = 3; i < PEERS; i++) assert(!peer(f, i));
}

static void two_supernodes(void) {
    struct fixture f; create(&f, CLIENT_TYPE_DESKTOP, 0); start(&f);
    up(&f, 0);
    assert(topo_group_connect_active_count(f.group) == 1 && peer(&f, 1));
    for (int i = 3; i < PEERS; i++) assert(!peer(&f, i));
    up(&f, 1);
    assert(topo_group_connect_active_count(f.group) == 2 && !peer(&f, 2));
    topo_group_connect_init(f.group); poll_events(&f);
    assert(peer(&f, 0)->retained && peer(&f, 1)->retained);
    down(&f, 0);
    assert(peer(&f, 2) && !peer(&f, 0)); /* сначала ещё не проверенный суперузел */
    for (int i = 3; i < PEERS; i++) assert(!peer(&f, i));
    destroy(&f);
}

static void retry_done(void* arg) { *(int*)arg = 1; }

/* Повтор должен сработать от таймера даже при достигнутой резервной цели. */
static void wait_retry(struct fixture* f) {
    int done = 0;
    assert(uasync_set_timeout(f->ua, 12000, &done, retry_done, "connect_test_retry"));
    while (!done) uasync_poll(f->ua, 100);
    poll_events(f);
}

static void fallback(int device) {
    struct fixture f; create(&f, device, 0); start(&f);
    fail(&f, 0); assert(peer(&f, 2) && !peer(&f, 3));
    fail(&f, 1); assert(!peer(&f, 3));
    fail(&f, 2);
    assert(peer(&f, 3) && peer(&f, 4) && !peer(&f, 5));
    up(&f, 3); fail(&f, 4);
    assert(!peer(&f, 5)); /* одного UP десктопа достаточно, чтобы запретить Android */
    down(&f, 3);
    assert(peer(&f, 5) && peer(&f, 6));
    up(&f, 5); up(&f, 6);
    assert(topo_group_connect_active_count(f.group) == 2);
    wait_retry(&f);
    assert(peer(&f, 0) || peer(&f, 1) || peer(&f, 2));
    topo_group_connect_restart(f.group); poll_events(&f);
    assert(peer(&f, 0)); up(&f, 0);
    assert(peer(&f, 5) && peer(&f, 6)); /* сохраняем даже без READY */
    assert(!peer(&f, 3) && !peer(&f, 4));
    assert(!f.conns[3]->close_requested && !f.conns[4]->close_requested);
    down(&f, 5);
    assert(!peer(&f, 5) && peer(&f, 6) && !peer(&f, 7));
    assert(!f.group->recovery);
    destroy(&f);
}

static void all_supernodes(void) {
    struct fixture f; create(&f, CLIENT_TYPE_SERVER, 1); start(&f);
    up(&f, 0); assert(peer(&f, 2));
    up(&f, 1); up(&f, 2);
    assert(topo_group_connect_active_count(f.group) == 3);
    for (int i = 3; i < PEERS; i++) assert(!peer(&f, i));
    down(&f, 0); assert(peer(&f, 0));
    for (int i = 3; i < PEERS; i++) assert(!peer(&f, i));
    destroy(&f);
}

/* Тип берётся из сохранённого snapshot; приватный адрес и пустые адреса не снимают роль. */
static void persisted_classification(void) {
    struct fixture f; create(&f, CLIENT_TYPE_DESKTOP, 0);
    struct topo_connect_peer* list = NULL; size_t count = 0;
    assert(topo_node_sqlite_get_connect_peers(f.inst->topo_sqlite_db, "42", &list, &count) == 0);
    assert(count == PEERS + 1);
    for (size_t i = 0; i < count; i++) {
        for (int n = 0; n < PEERS; n++) if (list[i].node_id == f.ids[n]) {
            assert(list[i].supernode == (n < 3));
            assert(list[i].client_type == (n < 5 ? CLIENT_TYPE_DESKTOP : n < 7 ? CLIENT_TYPE_MOBILE : CLIENT_TYPE_SERVER));
        }
    }
    u_free(list);
    struct TOPO_NODE* ni = topo_node_sqlite_snapshot_load(f.inst->topo_sqlite_db, f.inst->topo_groups, f.ids[0]); assert(ni);
    memory_pool_free(f.inst->topo_groups->v4_addr_pool, ni->v4_addrs); ni->v4_addrs = NULL;
    assert(topo_node_sqlite_addrs_put(f.inst->topo_sqlite_db, ni->node_id, ni) == 0);
    assert(topo_node_sqlite_nodeinfo_updated(f.inst->topo_sqlite_db, ni->node_id) == 0);
    assert(topo_node_sqlite_get_node_type(f.inst->topo_sqlite_db, "42", ni->node_id) == 4);
    uint64_t* ids = NULL; int number = 0;
    assert(topo_node_sqlite_get_supernode_peers(f.inst->topo_sqlite_db, "42", &ids, &number) == 0 && number == 3);
    u_free(ids); topo_node_destroy(f.inst->topo_groups, ni);
    destroy(&f);
}

#ifdef UTUN_HAVE_STANDBY
/* DOWN в SLEEP меняет состояние сразу, а новые попытки ждут активной фазы. */
static void standby_down(void) {
    struct fixture f; create(&f, CLIENT_TYPE_MOBILE, 0); start(&f);
    up(&f, 0); up(&f, 1);
    standby_set_intervals_ms(100, 200, 0);
    assert(standby_set_enabled(f.inst, 1) == 0);
    while (!standby_get_sleep_tb()) uasync_poll(f.ua, 100);
    down(&f, 0);
    assert(topo_group_connect_active_count(f.group) == 1 && !peer(&f, 0) && !peer(&f, 2));
    assert(standby_set_enabled(f.inst, 0) == 0); poll_events(&f);
    assert(peer(&f, 2));
    destroy(&f);

    create(&f, CLIENT_TYPE_MOBILE, 0);
    standby_set_intervals_ms(100, 200, 0);
    assert(standby_set_enabled(f.inst, 1) == 0);
    while (!standby_get_sleep_tb()) uasync_poll(f.ua, 100);
    assert(topo_group_connect_init(f.group) == 0); poll_events(&f);
    assert(!peer(&f, 0) && !peer(&f, 1));
    /* Остаток SLEEP может обнулиться раньше доставки фазового callback. */
    while (!standby_get_active_remaining_tb()) uasync_poll(f.ua, 100);
    poll_events(&f); assert(peer(&f, 0) && peer(&f, 1));
    assert(standby_set_enabled(f.inst, 0) == 0);
    destroy(&f);
}
#endif

int main(void) {
    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
    debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_INFO);
    utun_instance_set_tun_init_enabled(0);
    persisted_classification();
    two_supernodes();
    fallback(CLIENT_TYPE_DESKTOP);
    fallback(CLIENT_TYPE_MOBILE);
    all_supernodes();
#ifdef UTUN_HAVE_STANDBY
    standby_down();
#endif
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "group connect tests passed: UP, two/all supernodes, desktop/mobile fallback, retries, ownership");
    return 0;
}