utun2/tests/test_auto_socket_dynamic.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdarg.h>
#include "../lib/platform_compat.h"
#include "test_utils.h"
#ifndef _WIN32
#include <unistd.h>
#include <net/if.h>
#endif

#include "etcp.h"
#include "etcp_connections.h"
#include "etcp_api.h"
#include "../src/config_parser.h"
#include "../src/config_updater.h"
#include "../src/utun_instance.h"
#include "topo_group.h"
#include "topo_node.h"
#include "secure_channel.h"
#include "../src/transport_layer/node_conn_direct.h"
#include "../lib/u_async.h"
#include "../lib/ll_queue.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"

#define TIMEOUT_TB       300000
#define POLL_MS          5
#define STEP_TB          3000     /* 300ms между фазами */
#define TRAF_SEND_TB     50       /* 5ms  — отправка */
#define TRAF_MON_TB      1000     /* 100ms — мониторинг */
#define ETCP_RT_ID_TEST  0xF0

typedef void (*timeout_cb)(void*);

static struct test_ctx {
    struct UTUN_INSTANCE *server, *client;
    struct UASYNC* ua;
    int round;                    /* 0..7 */
    int step;
    int keep_tcp;
    int ip_changes_on_last;
    char cur_iface[IFNAMSIZ];
    char prev_iface[IFNAMSIZ];
    struct ETCP_CONN* srv_conn;
    uint64_t srv_node_id;
    uint8_t srv_pubkey[SC_PUBKEY_SIZE];
    int result;                   /* 0=running, 1=fail, 2=pass */

    /* traffic */
    uint32_t send_seq, send_count, pong_count, total_recv;
} ctx;

static void* timeout_handle;
static char tdir[] = "/tmp/utun_as_XXXXXX";
static char scf[256], ccf[256];

/* ── rounds ── */
static const struct {
    const char* ifname, *ip1, *ip2;
    int keep_tcp;
} rounds[] = {
    {"dummy_cli1", "10.90.0.2/24",  "10.90.1.2/24",  1},
    {"dummy_cli2", "10.90.0.3/24",  "10.90.1.3/24",  1},
    {"dummy_cli3", "10.90.0.4/24",  "10.90.1.4/24",  0},
    {"dummy_cli4", "10.90.0.5/24",  "10.90.1.5/24",  0},
    {"dummy_cli5", "10.90.0.6/24",  "10.90.1.6/24",  1},
    {"dummy_cli6", "10.90.0.7/24",  "10.90.1.7/24",  1},
    {"dummy_cli7", "10.90.0.8/24",  "10.90.1.8/24",  0},
    {"dummy_cli8", "10.90.0.9/24",  "10.90.1.9/24",  0},
};
#define N_ROUNDS (int)(sizeof(rounds)/sizeof(rounds[0]))

/* ── helpers ── */
static int wf(const char* p, const char* f, ...) {
    va_list ap; FILE* fp = fopen(p, "w"); if (!fp) return -1;
    va_start(ap, f); vfprintf(fp, f, ap); va_end(ap); fclose(fp); return 0;
}
static char* gv(const char* p, const char* k) {
    struct utun_config* c = parse_config(p); if (!c) return NULL;
    char* r = strcmp(k, "pub") == 0 ? u_strdup(c->global.my_public_key_hex)
                                    : u_strdup(c->global.my_private_key_hex);
    free_config(c); return r;
}
static void fail(const char* msg) {
    fprintf(stderr, "FAIL r=%d s=%d: %s\n", ctx.round, ctx.step, msg); fflush(stderr);
    ctx.result = 1;
}
static void to_cb(void* arg) { (void)arg; fprintf(stderr, "TIMEOUT\n"); ctx.result = 1; }

static int count_links_to_srv(void) {
    int n = 0; struct ll_entry* e = ctx.client->connections->head;
    while (e) { struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
        if (ce->conn->peer_node_id == ctx.srv_node_id) {
            struct ETCP_LINK* l = ce->conn->links;
            while (l) { if (l->initialized && l->link_status) n++; l = l->next; }
        } e = e->next; }
    return n;
}
static int all_links_are_type(void) {
    struct ll_entry* e = ctx.client->connections->head;
    while (e) { struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
        if (ce->conn->peer_node_id == ctx.srv_node_id) {
            struct ETCP_LINK* l = ce->conn->links;
            while (l) { if (l->initialized && l->link_status && l->is_tcp != ctx.keep_tcp) return 0; l = l->next; }
        } e = e->next; }
    return 1;
}

/* ── filter sockets ── */
static void keep_only_socket_type(int keep_tcp) {
    struct ETCP_SOCKET *s = ctx.client->etcp_sockets, *rm[16];
    int n = 0;
    while (s) { if (s->local_addr.ss_family == AF_INET && s->is_tcp != keep_tcp && n < 16) rm[n++] = s; s = s->next; }
    for (int i = 0; i < n; i++) {
        ncd_remove_socket_links(ctx.client, rm[i]);
        etcp_socket_remove(rm[i]);
    }
}

/* ── server node (2 addrs: UDP + TCP) ── */
static struct TOPO_GROUP_NODE* mk_srv_node(void) {
    struct TOPO_GROUP_NODE* nq = u_calloc(1, sizeof(struct TOPO_GROUP_NODE));
    struct TOPO_NODE* ni = u_calloc(1, sizeof(struct TOPO_NODE));
    if (!nq || !ni) { u_free(nq); u_free(ni); return NULL; }
    ni->group_ref_count = 1; ni->node_id = ctx.srv_node_id; ni->group_id = TOPO_GROUP_UTUN;
    memcpy(ni->public_key, ctx.srv_pubkey, SC_PUBKEY_SIZE);
    nq->ll.size = sizeof(struct TOPO_GROUP_NODE) - sizeof(struct ll_entry);
    nq->node_id = ctx.srv_node_id;

    struct TOPO_SOCKMETA4* sm = memory_pool_alloc(ctx.client->topo_groups->v4_sock_meta_pool);
    if (!sm) { u_free(nq); u_free(ni); return NULL; }
    sm->id = 0; sm->config_type = CFG_SERVER_TYPE_PUBLIC; sm->nat_type = NAT_TYPE_UNKNOWN;
    sm->next = ni->v4_sock_meta; ni->v4_sock_meta = sm;

    struct TOPO_ADDR4* a_udp = memory_pool_alloc(ctx.client->topo_groups->v4_addr_pool);
    struct TOPO_ADDR4* a_tcp = memory_pool_alloc(ctx.client->topo_groups->v4_addr_pool);
    if (!a_udp || !a_tcp) { u_free(nq); u_free(ni); return NULL; }
    a_udp->addr[0] = 10; a_udp->addr[1] = 90; a_udp->addr[2] = 0; a_udp->addr[3] = 1;
    a_udp->port = 9001; a_udp->protocol = TOPO_PROTO_UDP;
    a_udp->type = TOPO_ADDR_INTERFACE; a_udp->socket_id = 0;
    a_tcp->addr[0] = 10; a_tcp->addr[1] = 90; a_tcp->addr[2] = 0; a_tcp->addr[3] = 1;
    a_tcp->port = 9001; a_tcp->protocol = TOPO_PROTO_TCP;
    a_tcp->type = TOPO_ADDR_INTERFACE; a_tcp->socket_id = 0;
    a_tcp->next = a_udp; ni->v4_addrs = a_tcp;

    topo_node_registry_store(ctx.client->topo_groups, ni);
    return nq;
}

/* ── traffic ── */
static void srv_traffic_handler(struct ETCP_CONN* conn, struct ll_entry* entry) {
    if (!entry || entry->len < 5) { if (entry) queue_entry_free(entry); return; }
    struct ll_entry* reply = queue_entry_new(0);
    if (!reply) { queue_entry_free(entry); return; }
    reply->dgram = u_malloc(entry->len);
    if (reply->dgram) { memcpy(reply->dgram, entry->dgram, entry->len); reply->len = entry->len; }
    queue_entry_free(entry);
    if (reply->dgram) etcp_send(conn, reply);
    else queue_entry_free(reply);
}
static void cli_traffic_handler(struct ETCP_CONN* conn, struct ll_entry* entry) {
    (void)conn;
    if (!entry || entry->len < 5) { if (entry) queue_entry_free(entry); return; }
    ctx.pong_count++; ctx.total_recv++;
    queue_entry_free(entry);
}
static void traffic_send_timer(void* arg) {
    (void)arg;
    if (ctx.result) return;
    if (ctx.srv_conn && ctx.srv_conn->state != 2) {
        uint8_t buf[5]; buf[0] = ETCP_RT_ID_TEST;
        ctx.send_seq++; memcpy(buf + 1, &ctx.send_seq, 4);
        struct ll_entry* e = queue_entry_new(0);
        if (e) { e->dgram = u_malloc(5); if (e->dgram) { memcpy(e->dgram, buf, 5); e->len = 5;
            etcp_send(ctx.srv_conn, e); ctx.send_count++; } else queue_entry_free(e); }
    }
    if (!ctx.result) timeout_handle = uasync_set_timeout(ctx.ua, TRAF_SEND_TB, NULL, traffic_send_timer, "traf_snd");
}
static void traffic_monitor_timer(void* arg) {
    (void)arg;
    if (ctx.result) return;
    uint32_t d = ctx.pong_count; ctx.pong_count = 0;
    fprintf(stderr, "  [traf] r=%d tx=%u rx=%u+d=%u rate=%u/s\n",
            ctx.round, ctx.send_count, ctx.total_recv, d, d * 10); fflush(stderr);
    if (!ctx.result) timeout_handle = uasync_set_timeout(ctx.ua, TRAF_MON_TB, NULL, traffic_monitor_timer, "traf_mon");
}

/* ── etcp_connect callback ── */
static void connect_cb(void* arg, struct ETCP_CONN* conn, int type) {
    (void)arg;
    if (type == ETCP_CONNECT_EARLY || type == ETCP_CONNECT_LATE)
    { if (conn) { ctx.srv_conn = conn; ctx.step = 9; } }
}

static void start_traffic(void) {
    timeout_handle = uasync_set_timeout(ctx.ua, TRAF_SEND_TB, NULL, traffic_send_timer, "traf_snd");
    timeout_handle = uasync_set_timeout(ctx.ua, TRAF_MON_TB,  NULL, traffic_monitor_timer, "traf_mon");
}

/* ── ip addr add/del via system ── */
static int ip_addr_add(const char* ifname, const char* cidr) {
    char cmd[256]; snprintf(cmd, sizeof(cmd), "ip addr add %s dev %s 2>/dev/null", cidr, ifname);
    return system(cmd);
}
static int ip_addr_del(const char* ifname, const char* cidr) {
    char cmd[256]; snprintf(cmd, sizeof(cmd), "ip addr del %s dev %s 2>/dev/null", cidr, ifname);
    return system(cmd);
}
static int ip_link_add(const char* ifname) {
    char cmd[256]; snprintf(cmd, sizeof(cmd), "ip link add %s type dummy 2>/dev/null", ifname);
    return system(cmd);
}
static int ip_link_del(const char* ifname) {
    char cmd[256]; snprintf(cmd, sizeof(cmd), "ip link del %s 2>/dev/null", ifname);
    return system(cmd);
}
static int ip_link_up(const char* ifname) {
    char cmd[256]; snprintf(cmd, sizeof(cmd), "ip link set %s up 2>/dev/null", ifname);
    return system(cmd);
}

/* ═══════════════════════════════════════════════════════════
 * Phases
 * ═══════════════════════════════════════════════════════════ */
static void phase_check_ip(void* arg);
static void phase_change_ip(void* arg);
static void phase_check_del(void* arg);
static void phase_del_prev(void* arg);
static void phase_check_add(void* arg);
static void phase_add(void* arg);
static void phase_del_last(void* arg);
static void phase_done(void* arg);

static void phase_done(void* arg) {
    (void)arg;
    if (ctx.result) return;
    fprintf(stderr, "=== ALL PASSED ===\n"); fflush(stderr);
    ctx.result = 2;
}

static void phase_del_last(void* arg) {
    (void)arg;
    if (ctx.result) return;
    ctx.step = 11;
    ip_link_del(rounds[N_ROUNDS-1].ifname);
    uasync_set_timeout(ctx.ua, STEP_TB, NULL, (timeout_cb)phase_del_last, "chk_del_last");
    return; /* перепланируем один раз для проверки */
}

/* ══ check wrappers ══ */
static void do_check(void) {
    int l = count_links_to_srv();
    if (l < 1) { fail("no links"); return; }
    if (!all_links_are_type()) { fail("wrong link type"); return; }
}

static void phase_check_del_last(void* arg) {
    (void)arg;
    ctx.step = 12;
    int l = count_links_to_srv();
    if (l > 0) { fail("links survived last del"); return; }
    /* повторная проверка через 300ms — после второго захода считаем ОК */
    static int cnt = 0;
    if (++cnt < 2) { uasync_set_timeout(ctx.ua, STEP_TB, NULL, (timeout_cb)phase_check_del_last, "chk_del_last"); return; }
    fprintf(stderr, "  r=%d s=%d: no links after del_last (OK)\n", ctx.round, ctx.step); fflush(stderr);
    uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_done, "done");
}

static void phase_check_ip(void* arg) {
    (void)arg; if (ctx.result) return;
    ctx.step = 6;
    do_check(); if (ctx.result) return;

    if (ctx.round == N_ROUNDS - 1 && ctx.ip_changes_on_last < 2) {
        ctx.ip_changes_on_last++;
        fprintf(stderr, "  r=%d s=%d: IP change #%d verified — another\n", ctx.round, ctx.step, ctx.ip_changes_on_last);
        fflush(stderr);
        uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_change_ip, "chg_ip2");
    } else if (ctx.round == N_ROUNDS - 1 && ctx.ip_changes_on_last >= 2) {
        fprintf(stderr, "  r=%d s=%d: both IP changes on last round — deleting\n", ctx.round, ctx.step);
        fflush(stderr);
        uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_del_last, "del_last");
    } else {
        ctx.round++;
        uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_add, "next_add");
    }
}

static void phase_change_ip(void* arg) {
    (void)arg; if (ctx.result) return;
    ctx.step = 5;
    const char* old_ip = (ctx.ip_changes_on_last >= 2) ? rounds[ctx.round].ip2 : rounds[ctx.round].ip1;
    const char* new_ip = (ctx.ip_changes_on_last >= 2) ? rounds[ctx.round].ip1 : rounds[ctx.round].ip2;
    ip_addr_del(rounds[ctx.round].ifname, old_ip);
    ip_addr_add(rounds[ctx.round].ifname, new_ip);
    keep_only_socket_type(ctx.keep_tcp);
    uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_check_ip, "chk_ip");
}

static void phase_check_del(void* arg) {
    (void)arg; if (ctx.result) return;
    ctx.step = 4;
    do_check(); if (ctx.result) return;
    fprintf(stderr, "  r=%d s=%d: del_prev OK links=%d\n", ctx.round, ctx.step, count_links_to_srv());
    fflush(stderr);
    uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_change_ip, "chg_ip");
}

static void phase_del_prev(void* arg) {
    (void)arg; if (ctx.result) return;
    ctx.step = 3;
    ip_link_del(ctx.prev_iface);
    uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_check_del, "chk_del");
}

static void phase_check_add(void* arg) {
    (void)arg; if (ctx.result) return;
    ctx.step = 2;
    /* wait for etcp_connect on round 0 */
    if (ctx.round == 0 && ctx.step == 2 && !ctx.srv_conn) {
        uasync_set_timeout(ctx.ua, STEP_TB / 3, NULL, phase_check_add, "chk_add");
        return;
    }
    if (ctx.round == 0 && !ctx.srv_conn) { fail("etcp_connect didn't fire"); return; }
    do_check(); if (ctx.result) return;
    fprintf(stderr, "  r=%d s=%d: add OK links=%d type=%s\n",
            ctx.round, ctx.step, count_links_to_srv(), ctx.keep_tcp ? "TCP" : "UDP");
    fflush(stderr);

    if (ctx.round == 0) start_traffic();

    if (ctx.prev_iface[0]) {
        uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_del_prev, "del_prev");
    } else {
        uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_change_ip, "chg_ip");
    }
}

static void phase_add(void* arg) {
    (void)arg; if (ctx.result) return;
    ctx.step = 1;
    ctx.keep_tcp = rounds[ctx.round].keep_tcp;
    strncpy(ctx.cur_iface, rounds[ctx.round].ifname, IFNAMSIZ - 1);

    ip_link_add(ctx.cur_iface);
    ip_addr_add(ctx.cur_iface, rounds[ctx.round].ip1);
    ip_link_up(ctx.cur_iface);
    keep_only_socket_type(ctx.keep_tcp);

    if (ctx.round == 0) {
        struct TOPO_GROUP_NODE* sn = mk_srv_node();
        if (!sn) { fail("mk_srv_node"); return; }
        queue_data_put_with_index(topo_groups_get_default(ctx.client->topo_groups)->nodes, &sn->ll);
        etcp_connect(ctx.client, sn, connect_cb, NULL, ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE);
    }
    uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_check_add, "chk_add");
}

/* ═══════════════════════════════════════════════════════════
 * setup / cleanup / main
 * ═══════════════════════════════════════════════════════════ */
static void cleanup_ifaces(void) {
    (void)system("ip link del dummy_srv 2>/dev/null");
    for (int i = 0; i < N_ROUNDS; i++) ip_link_del(rounds[i].ifname);
}

static void setup(void) {
    if (geteuid() != 0) { fprintf(stderr, "SKIP: test requires root\n"); exit(77); }
    atexit(cleanup_ifaces);
    cleanup_ifaces();
    test_mkdtemp(tdir);
    snprintf(scf, sizeof(scf), "%s/s.conf", tdir);
    snprintf(ccf, sizeof(ccf), "%s/c.conf", tdir);

    /* create db directories */
    { char dbd[256]; snprintf(dbd, sizeof(dbd), "%s/db_srv", tdir); utun_mkdir(dbd, 0755);
      snprintf(dbd, sizeof(dbd), "%s/db_cli", tdir); utun_mkdir(dbd, 0755); }

    /* server: fixed [server] on dummy_srv */
    ip_link_add("dummy_srv");
    ip_addr_add("dummy_srv", "10.90.0.1/24");
    ip_link_up("dummy_srv");

    /* client: first interface */
    ip_link_add(rounds[0].ifname);
    ip_addr_add(rounds[0].ifname, rounds[0].ip1);
    ip_link_up(rounds[0].ifname);

    /* write initial configs */
    wf(scf, "[global]\ntun_ip=10.99.0.1/24\ntun_ifname=tun_srv\ntun_test_mode=1\n"
            "auto_sockets=no\ndb_path=%s/db_srv\n"
            "[server: fixed]\naddr=10.90.0.1:9001\ntype=public\n[allowed_keys]\nallow_all=1\n", tdir);
    wf(ccf, "[global]\ntun_ip=10.99.0.2/24\ntun_ifname=tun_cli\ntun_test_mode=1\n"
            "auto_sockets=yes\ndb_path=%s/db_cli\n[allowed_keys]\nallow_all=1\n", tdir);
    config_ensure_keys_and_node_id(scf);
    config_ensure_keys_and_node_id(ccf);

    { struct utun_config* cs = parse_config(scf); ctx.srv_node_id = cs->global.my_node_id; free_config(cs); }
    { struct utun_config* cc = parse_config(ccf); (void)cc->global.my_node_id; free_config(cc); }

    char *spub = gv(scf, "pub"), *spriv = gv(scf, "priv");
    char *cpub = gv(ccf, "pub"), *cpriv = gv(ccf, "priv");

    wf(scf, "[global]\nmy_private_key=%s\nmy_public_key=%s\ntun_ip=10.99.0.1/24\ntun_ifname=tun_srv\n"
            "tun_test_mode=1\nauto_sockets=no\ndb_path=%s/db_srv\n"
            "[server: fixed]\naddr=10.90.0.1:9001\ntype=public\n[allowed_keys]\nallow_all=1\n",
            spriv, spub, tdir);
    wf(ccf, "[global]\nmy_private_key=%s\nmy_public_key=%s\ntun_ip=10.99.0.2/24\ntun_ifname=tun_cli\n"
            "tun_test_mode=1\nauto_sockets=yes\ndb_path=%s/db_cli\n[allowed_keys]\nallow_all=1\n",
            cpriv, cpub, tdir);

    /* verify config */
    { struct utun_config* ck = parse_config(ccf);
      fprintf(stderr, "  [setup] client: auto_sockets=%d node_id=0x%016llx\n",
              ck ? ck->global.auto_sockets : -1,
              (unsigned long long)(ck ? ck->global.my_node_id : 0));
      free_config(ck);
      ck = parse_config(scf);
      fprintf(stderr, "  [setup] server: auto_sockets=%d node_id=0x%016llx\n",
              ck ? ck->global.auto_sockets : -1,
              (unsigned long long)(ck ? ck->global.my_node_id : 0));
      fflush(stderr);
      free_config(ck); }

    struct utun_config* cs2 = parse_config(scf);
    if (cs2 && cs2->global.my_public_key_hex)
        sc_hex_to_binary(cs2->global.my_public_key_hex, ctx.srv_pubkey, SC_PUBKEY_SIZE);
    free_config(cs2);

    u_free(spub); u_free(spriv); u_free(cpub); u_free(cpriv);
}

static void cleanup(void) { test_unlink(scf); test_unlink(ccf); test_rmdir(tdir); }

int main(void) {
    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
    utun_instance_set_tun_init_enabled(0);
    setup();

    ctx.ua = uasync_create();
    ctx.server = utun_instance_create(ctx.ua, scf);
    ctx.client = utun_instance_create(ctx.ua, ccf);
    if (!ctx.server || !ctx.client) goto done;

    utun_instance_init(ctx.server);
    utun_instance_init(ctx.client);

    etcp_bind(ctx.server, ETCP_RT_ID_TEST, srv_traffic_handler);
    etcp_bind(ctx.client, ETCP_RT_ID_TEST, cli_traffic_handler);

    /* store init state */
    ctx.prev_iface[0] = '\0';
    strncpy(ctx.prev_iface, rounds[0].ifname, IFNAMSIZ - 1);
    ctx.prev_iface[0] = '\0'; /* round 0 has no prev */

    uasync_set_timeout(ctx.ua, STEP_TB, NULL, phase_add, "init");

    timeout_handle = uasync_set_timeout(ctx.ua, TIMEOUT_TB, NULL, to_cb, "to");

    { uint64_t start = get_time_tb();
      while (!ctx.result && (int)(get_time_tb() - start) < TIMEOUT_TB + 50000)
          uasync_poll(ctx.ua, POLL_MS); }

    fprintf(stderr, "final result=%d\n", ctx.result); fflush(stderr);

done:
    if (timeout_handle) uasync_cancel_timeout(ctx.ua, timeout_handle);
    if (ctx.server) { ctx.server->running = 0; utun_instance_destroy(ctx.server); }
    if (ctx.client) { ctx.client->running = 0; utun_instance_destroy(ctx.client); }
    if (ctx.ua) { uasync_destroy(ctx.ua, 0); ctx.ua = NULL; }
    cleanup();
    return (ctx.result == 2) ? 0 : 1;
}