Cover control server regressions and correct API documentation

src/control_server_doc.md

@@ -10,9 +10,15 @@ TCP-сервер для мониторинга ETCP-соединений. При
 4. При остановке вызвать `control_server_shutdown()`.
 
 **Нюансы:**
-- Доступ по IP контролируется через `control_allow` в конфиге (белый список); без правил доступ запрещён.
-- Клиенты отключаются по idle-таймауту (300 секунд без команд).
-- Лог пишется в `control_server.log` (полные дампы RX/TX).
+- Доступ разрешён только IPv4-клиентам по `control_allow` в конфиге; без правил доступ запрещён. IPv6-клиенты отклоняются.
+- Клиенты отключаются по idle-таймауту (30 секунд без команд). Каждая обработанная команда перезапускает таймер.
+- Лог пишется в `control_server.log`: дампы содержат первые 256 байт каждого чтения RX и некоторых ответов TX.
+- `control_server_process_updates()` разбирает команды; метрики отправляются только по `CMD_GET_METRICS`.
+- Очередь отправки владеет исходными буферами до полной отправки или отключения; частичная отправка хранит отдельное смещение.
+- Если список сокетов превышает предел сообщения (32 768 байт; сейчас помещается 230 записей),
+  возвращается `RSP_ERROR` с кодом `SERVER_BUSY`. Список не усекается; сокеты `probe_only` не учитываются.
+- `SET_DEBUG_CONFIG` требует ровно `2 + category_count` байт, не более 64 категорий.
+  Несоответствие длины закрывает соединение без применения настроек.
 
 ## 3. API
 
@@ -42,7 +48,7 @@ TCP-сервер для мониторинга ETCP-соединений. При
 | `CMD_SELECT_CONN` | — | Выбрать соединение по `peer_node_id` для метрик. |
 | `CMD_GET_METRICS` | `RSP_METRICS` | Полные метрики: ETCP, TUN, router, все каналы с BBR. |
 | `CMD_LIST_SOCKETS` | `RSP_SOCKET_LIST` | Список локальных сокетов с NAT-типом и NAT-адресом. |
-| `CMD_ACTION` | `RSP_ACTION_RESULT` | `"nat"` — запрос NAT-проверки; `"nodes"` — дамп BGP-узлов текстом. |
+| `CMD_ACTION` | Для `"nodes"` — `RSP_ACTION_RESULT`; для `"nat"` — без ответа | `"nat"` — запрос NAT-проверки; `"nodes"` — дамп BGP-узлов текстом. |
 | `CMD_GET_DEBUG_CONFIG` | `RSP_DEBUG_CONFIG` | Текущие уровни отладки (глобальный + по категориям). |
 | `CMD_SET_DEBUG_CONFIG` | — | Установить уровни отладки. |
 | `CMD_SUBSCRIBE_NODES` | `RSP_NODE_INFO` (список) + `RSP_NODE_INFO_END` | Подписка на изменения узлов + текущий список. |

tests/Makefile.am

@@ -309,6 +309,11 @@ test_config_updater_LDFLAGS = -Wl,--wrap=fopen -Wl,--wrap=fdopen -Wl,--wrap=frea
     -Wl,--wrap=fwrite -Wl,--wrap=fflush -Wl,--wrap=fclose -Wl,--wrap=fsync -Wl,--wrap=fchmod -Wl,--wrap=rename -Wl,--wrap=open \
     -Wl,--wrap=u_malloc_impl -Wl,--wrap=u_calloc_impl -Wl,--wrap=u_realloc_impl -Wl,--wrap=u_strdup_impl
 
+check_PROGRAMS += test_control_server
+test_control_server_SOURCES = test_control_server.c
+test_control_server_LDADD = $(LIBUTUN) $(COMMON_LIBS)
+test_control_server_LDFLAGS = -Wl,--wrap=accept -Wl,--wrap=send -Wl,--wrap=uasync_add_socket_t \
+    -Wl,--wrap=uasync_set_timeout -Wl,--wrap=u_malloc_impl
 async_regression_sources = async_regression.c async_regression.h
 async_regression_ldflags = -Wl,--wrap=u_malloc_impl -Wl,--wrap=u_calloc_impl -Wl,--wrap=u_realloc_impl \
     -Wl,--wrap=gettimeofday -Wl,--wrap=epoll_wait -Wl,--wrap=epoll_ctl -Wl,--wrap=eventfd -Wl,--wrap=epoll_create1

tests/test_control_server.c

@@ -0,0 +1,248 @@
+/* Регрессии control_server: настоящие очереди/таймеры, управляемые короткие send
+ * и ошибки регистрации. Включение реализации даёт доступ к socket callbacks. */
+#include "../src/control_server.c"
+#include <assert.h>
+
+static socket_t pending_accept = -1, capture_fd = -1;
+static int fail_registration, fail_timer, fail_entry, send_mode;
+static size_t send_chunk, wire_size;
+static uint8_t wire[ETCPMON_MAX_MSG_SIZE];
+
+socket_t __real_accept(socket_t, struct sockaddr*, socklen_t*);
+socket_t __wrap_accept(socket_t fd, struct sockaddr* addr, socklen_t* len) {
+    if (pending_accept < 0) return __real_accept(fd, addr, len);
+    struct sockaddr_in ip = {.sin_family = AF_INET};
+    ip.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    assert(*len >= sizeof(ip)); memcpy(addr, &ip, sizeof(ip)); *len = sizeof(ip);
+    socket_t result = pending_accept; pending_accept = -1;
+    return result;
+}
+
+void* __real_uasync_add_socket_t(struct UASYNC*, socket_t, socket_t_callback_t, socket_t_callback_t,
+                                socket_t_callback_t, const char*, void*);
+void* __wrap_uasync_add_socket_t(struct UASYNC* ua, socket_t fd, socket_t_callback_t read, socket_t_callback_t write,
+                                socket_t_callback_t error, const char* name, void* arg) {
+    if (fail_registration && !strcmp(name, "control_client")) { fail_registration = 0; return NULL; }
+    return __real_uasync_add_socket_t(ua, fd, read, write, error, name, arg);
+}
+
+void* __real_uasync_set_timeout(struct UASYNC*, int, void*, timeout_callback_t, const char*);
+void* __wrap_uasync_set_timeout(struct UASYNC* ua, int tb, void* arg, timeout_callback_t callback, const char* name) {
+    if (fail_timer && !strcmp(name, "ctrl_idle")) { fail_timer = 0; return NULL; }
+    return __real_uasync_set_timeout(ua, tb, arg, callback, name);
+}
+
+void* __real_u_malloc_impl(uint32_t, const char*);
+void* __wrap_u_malloc_impl(uint32_t size, const char* location) {
+    if (fail_entry && strstr(location, "ll_queue.c:")) { fail_entry = 0; return NULL; }
+    return __real_u_malloc_impl(size, location);
+}
+
+ssize_t __real_send(socket_t, const void*, size_t, int);
+ssize_t __wrap_send(socket_t fd, const void* data, size_t len, int flags) {
+    if (fd != capture_fd) return __real_send(fd, data, len, flags);
+    if (send_mode == 1) { errno = EAGAIN; return -1; }
+    if (send_mode == 2) { errno = EPIPE; return -1; }
+    if (send_mode == 3) return 0;
+    size_t take = len < send_chunk ? len : send_chunk;
+    assert(take > 0 && wire_size + take <= sizeof(wire));
+    memcpy(wire + wire_size, data, take); wire_size += take;
+    return (ssize_t)take;
+}
+
+struct fixture {
+    struct UTUN_INSTANCE inst;
+    struct control_server server;
+    socket_t peer;
+    size_t allocations;
+};
+
+static void fixture_init(struct fixture* f) {
+    memset(f, 0, sizeof(*f));
+    f->allocations = u_get_allocated_count();
+    f->inst.ua = uasync_create(); assert(f->inst.ua);
+    f->inst.config = u_calloc(1, sizeof(*f->inst.config)); assert(f->inst.config);
+    f->inst.config->global.control_allow_count = 1;
+    f->inst.config->global.control_allows[0].network = INADDR_LOOPBACK;
+    f->inst.config->global.control_allows[0].netmask = UINT32_MAX;
+    f->server.ua = f->inst.ua; f->server.instance = &f->inst;
+    f->server.max_clients = 8; f->server.listen_fd = -1; f->peer = -1;
+}
+
+static struct control_client* fixture_accept(struct fixture* f) {
+    socket_t pair[2]; assert(socketpair(AF_UNIX, SOCK_STREAM, 0, pair) == 0);
+    assert(f->peer == -1); f->peer = pair[1]; pending_accept = pair[0];
+    accept_callback(-1, &f->server);
+    return f->server.clients;
+}
+
+static void fixture_destroy(struct fixture* f) {
+    capture_fd = -1; send_mode = 0;
+    control_server_shutdown(&f->server);
+    if (f->peer >= 0) close(f->peer);
+    assert(timeout_heap_get_size(f->inst.ua->timeout_heap) == 0);
+    assert(f->inst.ua->timer_alloc_count == f->inst.ua->timer_free_count);
+    uasync_destroy(f->inst.ua, 0); u_free(f->inst.config);
+    assert(u_get_allocated_count() == f->allocations);
+}
+
+static void enqueue(struct fixture* f, struct control_client* client, const char* text) {
+    uint16_t len = (uint16_t)strlen(text);
+    uint8_t* buffer = u_malloc(len); assert(buffer); memcpy(buffer, text, len);
+    assert(control_client_send(&f->server, client, buffer, len) == 0);
+}
+
+/* Проверяем wire-байты и учёт памяти после нескольких partial send и EAGAIN. */
+static void test_partial_send(void) {
+    struct fixture f; fixture_init(&f);
+    struct control_client* client = fixture_accept(&f); assert(client);
+    capture_fd = client->fd; send_chunk = 3; wire_size = 0;
+    enqueue(&f, client, "0123456789"); enqueue(&f, client, "abcdefgh");
+    uint8_t* base = client->output_queue->head->dgram;
+    client_write_callback(client->fd, client);
+    assert(client->output_offset == 3 && client->output_queue->head->dgram == base);
+    assert(queue_total_bytes(client->output_queue) == 18);
+    send_mode = 1; client_write_callback(client->fd, client);
+    assert(client->output_offset == 3 && wire_size == 3);
+    send_mode = 0;
+    while (client->output_queue->head) client_write_callback(client->fd, client);
+    assert(wire_size == 18 && !memcmp(wire, "0123456789abcdefgh", 18));
+    assert(!client->output_offset && !client->write_registered && !queue_total_bytes(client->output_queue));
+    fixture_destroy(&f);
+    puts("PASS partial send: exact wire bytes, EAGAIN, stable buffer and queue accounting");
+}
+
+static void test_cleanup(void) {
+    for (int mode = 0; mode <= 3; mode++) {
+        struct fixture f; fixture_init(&f);
+        struct control_client* client = fixture_accept(&f); assert(client);
+        enqueue(&f, client, "first message"); enqueue(&f, client, "second message");
+        capture_fd = client->fd; send_chunk = 2; wire_size = 0;
+        client_write_callback(client->fd, client); assert(client->output_offset == 2);
+        if (mode >= 2) { send_mode = mode; client_write_callback(client->fd, client); }
+        else if (mode == 1) {
+            uasync_cancel_timeout(f.inst.ua, client->idle_timer);
+            control_client_idle_cb(client);
+        } else close_client(&f.server, client);
+        assert(!f.server.clients && !f.server.client_count);
+        fixture_destroy(&f);
+    }
+    for (int fault = 0; fault < 2; fault++) {
+        struct fixture f; fixture_init(&f);
+        fail_registration = fault == 0; fail_timer = fault == 1;
+        assert(!fixture_accept(&f));
+        assert(!f.server.client_count && !fail_registration && !fail_timer);
+        fixture_destroy(&f);
+    }
+    struct fixture f; fixture_init(&f);
+    struct control_client* client = fixture_accept(&f); assert(client);
+    uint8_t* buffer = u_malloc(8); assert(buffer);
+    size_t before = u_get_allocated_count(); fail_entry = 1;
+    assert(control_client_send(&f.server, client, buffer, 8) == -1 && !fail_entry);
+    assert(u_get_allocated_count() == before - 1 && !client->output_queue->head);
+    fixture_destroy(&f);
+    puts("PASS cleanup: pending/partial sends, idle, send failure, registration/timer failure, allocation failure; no leaks");
+}
+
+/* Байты после объявленного SET_DEBUG_CONFIG не должны менять настройки. */
+static void test_debug_payload(void) {
+    for (int variant = 0; variant < 6; variant++) {
+        struct fixture f; fixture_init(&f);
+        struct control_client* client = fixture_accept(&f); assert(client);
+        debug_config_t saved = g_debug_config;
+        struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)client->recv_buffer;
+        uint8_t* payload = client->recv_buffer + sizeof(*hdr);
+        memset(payload, DEBUG_LEVEL_TRACE, 100);
+        payload[0] = DEBUG_LEVEL_ERROR; payload[1] = 3;
+        uint16_t len = variant == 0 ? 1 : variant == 1 ? 2 : variant == 2 ? 4 : variant == 3 ? 6 : 5;
+        if (variant == 4) { payload[1] = ETCPMON_MAX_DEBUG_CATEGORIES + 1; len = 2 + payload[1]; }
+        etcpmon_build_header(hdr, len, ETCPMON_CMD_SET_DEBUG_CONFIG, 17);
+        client->recv_len = hdr->size;
+        if (variant == 5) {
+            client->recv_len--; handle_client_data(&f.server, client);
+            assert(!memcmp(&saved, &g_debug_config, sizeof(saved)) && f.server.clients == client);
+            client->recv_len++; handle_client_data(&f.server, client);
+            assert(g_debug_config.level == DEBUG_LEVEL_ERROR);
+            for (int i = 1; i <= 3; i++) assert(g_debug_config.category_levels[i] == DEBUG_LEVEL_TRACE);
+        } else {
+            handle_client_data(&f.server, client);
+            assert(!f.server.clients && !memcmp(&saved, &g_debug_config, sizeof(saved)));
+        }
+        g_debug_config = saved;
+        fixture_destroy(&f);
+    }
+    puts("PASS SET_DEBUG_CONFIG: short, truncated, extra and excessive payloads rejected; fragmented valid command applied");
+}
+
+static void test_metrics(void) {
+    struct fixture f; fixture_init(&f);
+    struct control_client* client = fixture_accept(&f); assert(client);
+    struct ll_queue empty = {0};
+    struct ETCP_CONN conn = {.peer_node_id = 42, .input_queue = &empty, .input_send_q = &empty,
+        .input_wait_ack = &empty, .ack_q = &empty, .recv_q = &empty, .output_queue = &empty};
+    struct ETCP_LINK links[2] = {{.local_link_id = 7}, {.local_link_id = 8}};
+    links[0].next = &links[1]; conn.links = links;
+    f.inst.connections = queue_new(f.inst.ua, 16, 0, 8, "test connections"); assert(f.inst.connections);
+    struct ll_entry* entry = queue_entry_new(sizeof(struct conn_queue_entry)); assert(entry);
+    struct conn_queue_entry* ce = (struct conn_queue_entry*)entry->data;
+    ce->peer_node_id = conn.peer_node_id; ce->conn = &conn;
+    assert(queue_data_put_with_index(f.inst.connections, entry) == 0);
+    client->selected_peer_id = conn.peer_node_id;
+    for (int count = 1; count <= 2; count++) {
+        links[0].next = count == 1 ? NULL : &links[1];
+        send_metrics(&f.server, client, 23);
+        struct ll_entry* response = queue_data_get(client->output_queue); assert(response);
+        u_check(response->dgram, "metrics bounds", LOCATION);
+        struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)response->dgram;
+        assert(hdr->type == ETCPMON_RSP_METRICS && hdr->seq_id == 23 && hdr->size == ETCPMON_METRICS_SIZE(count));
+        struct etcpmon_link_metrics* metrics = (void*)(response->dgram + sizeof(*hdr) + sizeof(struct etcpmon_rsp_metrics));
+        for (int i = 0; i < count; i++) {
+            assert(metrics[i].local_link_id == 7 + i);
+            const uint8_t* start = &metrics[i].bbr_mode;
+            for (size_t j = 0; j < sizeof(*metrics) - offsetof(struct etcpmon_link_metrics, bbr_mode); j++) assert(start[j] == 0);
+        }
+        queue_dgram_free(response); queue_entry_free(response);
+    }
+    queue_entry_free(queue_data_get(f.inst.connections)); queue_free(f.inst.connections);
+    fixture_destroy(&f);
+    puts("PASS metrics: one/two links without BBR, zeroed BBR fields and intact allocation guards");
+}
+
+static void test_socket_limit(void) {
+    struct fixture f; fixture_init(&f);
+    struct control_client* client = fixture_accept(&f); assert(client);
+    size_t limit = (ETCPMON_MAX_MSG_SIZE - ETCPMON_SOCKET_LIST_SIZE(0)) / sizeof(struct etcpmon_socket_info);
+    struct ETCP_SOCKET* sockets = u_calloc(251, sizeof(*sockets)); assert(sockets);
+    for (size_t i = 0; i < 251; i++) {
+        snprintf(sockets[i].name, sizeof(sockets[i].name), "socket-%zu", i);
+        sockets[i].interface_addr.ss_family = AF_INET;
+        if (i < 250) sockets[i].next = &sockets[i + 1];
+    }
+    f.inst.etcp_sockets = sockets;
+    const size_t counts[] = {0, limit, limit + 1, 250};
+    for (size_t n = 0; n < sizeof(counts) / sizeof(counts[0]); n++) {
+        size_t count = counts[n];
+        for (size_t i = 0; i < 251; i++) sockets[i].probe_only = i >= count;
+        send_socket_list(&f.server, client, 31);
+        struct ll_entry* response = queue_data_get(client->output_queue); assert(response);
+        struct etcpmon_msg_header* hdr = (void*)response->dgram;
+        assert(etcpmon_validate_header(hdr) == 0 && hdr->size == response->len && hdr->seq_id == 31);
+        if (count <= limit) {
+            assert(hdr->type == ETCPMON_RSP_SOCKET_LIST && hdr->size == ETCPMON_SOCKET_LIST_SIZE(count));
+            assert(response->dgram[sizeof(*hdr)] == count);
+        } else {
+            assert(hdr->type == ETCPMON_RSP_ERROR && response->dgram[sizeof(*hdr)] == ETCPMON_ERR_SERVER_BUSY);
+        }
+        queue_dgram_free(response); queue_entry_free(response);
+    }
+    u_free(sockets); f.inst.etcp_sockets = NULL;
+    fixture_destroy(&f);
+    puts("PASS socket list: zero, maximum fitting, first excessive and 250 sockets; probe-only sockets excluded");
+}
+
+int main(void) {
+    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
+    debug_set_category_level(DEBUG_CATEGORY_CONTROL, DEBUG_LEVEL_DEBUG);
+    test_partial_send(); test_cleanup(); test_debug_payload(); test_metrics(); test_socket_limit();
+    return 0;
+}