Fix control server buffer bounds and client resource ownership

src/control_server.c

@@ -377,7 +377,6 @@ static void accept_callback(socket_t fd, void* arg) {
 #endif
         return;
     }
-    DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Accept...");
     
     /* Check max clients */
     if (server->client_count >= server->max_clients) {
@@ -429,7 +428,6 @@ static void accept_callback(socket_t fd, void* arg) {
     }
     client->subscribed_nodes = 0;
     client->write_registered = 0;
-    client->idle_timer = uasync_set_timeout(server->ua, 300000, client, control_client_idle_cb, "ctrl_idle");
 
     /* Register with uasync */
     client->socket_id = uasync_add_socket_t(server->ua, client_fd,
@@ -446,6 +444,7 @@ static void accept_callback(socket_t fd, void* arg) {
             fflush(server->log_file);
         }
         DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to register client socket with uasync");
+        queue_free(client->output_queue);
         u_free(client);
 #ifdef _WIN32
         closesocket(client_fd);
@@ -460,6 +459,13 @@ static void accept_callback(socket_t fd, void* arg) {
     server->clients = client;
     server->client_count++;
 
+    client->idle_timer = uasync_set_timeout(server->ua, 300000, client, control_client_idle_cb, "ctrl_idle");
+    if (!client->idle_timer) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to arm control client idle timer fd=%d", (int)client_fd);
+        close_client(server, client);
+        return;
+    }
+
     const char* addr_str;
     if (client_addr.ss_family == AF_INET) {
         struct sockaddr_in* sin = (struct sockaddr_in*)&client_addr;
@@ -587,7 +593,7 @@ static void client_write_callback(socket_t fd, void* arg) {
         struct ll_entry* entry = q->head;
 
 #ifdef _WIN32
-        int sent = send(fd, (const char*)entry->dgram, entry->len, 0);
+        int sent = send(fd, (const char*)entry->dgram + client->output_offset, entry->len - client->output_offset, 0);
         if (sent == SOCKET_ERROR) {
             int err = WSAGetLastError();
             if (err != WSAEWOULDBLOCK) {
@@ -598,7 +604,7 @@ static void client_write_callback(socket_t fd, void* arg) {
             return;
         }
 #else
-        ssize_t sent = send(fd, entry->dgram, entry->len, 0);
+        ssize_t sent = send(fd, entry->dgram + client->output_offset, entry->len - client->output_offset, 0);
         if (sent < 0) {
             if (errno != EAGAIN && errno != EWOULDBLOCK) {
                 DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Client send error: %s, closing", strerror(errno));
@@ -608,10 +614,21 @@ static void client_write_callback(socket_t fd, void* arg) {
             return;
         }
 #endif
-        if ((size_t)sent < entry->len) { entry->dgram += sent; entry->len -= (uint16_t)sent; q->total_bytes -= (size_t)sent; return; }
+        if (sent == 0) {
+            DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control client send made no progress fd=%d, closing", (int)fd);
+            close_client(server, client);
+            return;
+        }
+        client->output_offset += (uint16_t)sent;
+        if (client->output_offset < entry->len) {
+            DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Control partial send fd=%d offset=%u size=%u queued_bytes=%zu",
+                        (int)fd, client->output_offset, entry->len, queue_total_bytes(q));
+            return;
+        }
 
         struct ll_entry* e = queue_data_get(q);
         queue_dgram_free(e); queue_entry_free(e);
+        client->output_offset = 0;
     }
 
     uasync_set_socket_write(server->ua, client->socket_id, 0);
@@ -631,14 +648,22 @@ static void client_except_callback(socket_t fd, void* arg) {
     (void)fd;
 }
 
+/* Принимает владение буфером, включая ветки ошибки постановки в очередь. */
 static int control_client_send(struct control_server* server, struct control_client* client, uint8_t* dgram, uint16_t len) {
     (void)server;
-    if (!client || !client->output_queue || !dgram || len == 0) return -1;
+    if (!client || !client->output_queue || !dgram || len == 0 || len > ETCPMON_MAX_MSG_SIZE) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid control output message client=%p size=%u", (void*)client, len);
+        u_free(dgram);
+        return -1;
+    }
     struct ll_entry* entry = queue_entry_new(0);
-    if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate output queue entry"); return -1; }
+    if (!entry) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate output queue entry size=%u", len);
+        u_free(dgram);
+        return -1;
+    }
     entry->dgram = dgram;
     entry->len = len;
-    entry->int_len = len;
     queue_data_put(client->output_queue, entry);
     if (!client->write_registered) {
         uasync_set_socket_write(client->server->ua, client->socket_id, 1);
@@ -657,6 +682,10 @@ static void control_client_idle_cb(void* arg) {
 
 static void close_client(struct control_server* server, struct control_client* client) {
     if (!server || !client) return;
+
+    DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Closing control client fd=%d queued_packets=%d queued_bytes=%zu output_offset=%u timer=%d",
+                (int)client->fd, queue_entry_count(client->output_queue), queue_total_bytes(client->output_queue),
+                client->output_offset, client->idle_timer != NULL);
     
     if (client->idle_timer) { uasync_cancel_timeout(server->ua, client->idle_timer); client->idle_timer = NULL; }
 
@@ -680,7 +709,15 @@ static void close_client(struct control_server* server, struct control_client* c
     }
 #endif
 
-    if (client->output_queue) { queue_free(client->output_queue); client->output_queue = NULL; }
+    if (client->output_queue) {
+        struct ll_entry* entry;
+        while ((entry = queue_data_get(client->output_queue))) {
+            queue_dgram_free(entry);
+            queue_entry_free(entry);
+        }
+        queue_free(client->output_queue);
+        client->output_queue = NULL;
+    }
 
     /* Remove from list */
     struct control_client** curr = &server->clients;
@@ -822,25 +859,29 @@ static void handle_client_data(struct control_server* server, struct control_cli
                 send_debug_config(server, client, req_seq);
                 break;
 
-            case ETCPMON_CMD_SET_DEBUG_CONFIG:
-                if (payload_size >= 2) {
-                    uint8_t global_lvl = payload[0];
-                    uint8_t cat_count = payload[1];
-                    if (cat_count > ETCPMON_MAX_DEBUG_CATEGORIES) cat_count = ETCPMON_MAX_DEBUG_CATEGORIES;
-                    if (global_lvl <= DEBUG_LEVEL_TRACE) debug_set_level((debug_level_t)global_lvl);
-                    const uint8_t* levels = payload + 2;
-                    for (uint8_t i = 0; i < cat_count && i + 1 < DEBUG_CATEGORY_COUNT; i++) {
-                        if (levels[i] <= DEBUG_LEVEL_TRACE)
-                            debug_set_category_level((debug_category_t)(i + 1), (debug_level_t)levels[i]);
-                    }
-                    DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Debug config applied: global=%d categories=%d", global_lvl, cat_count);
-                    if (server->log_file) {
-                        fprintf(server->log_file, "%llu: [LOG] Debug config set: global=%d categories=%d\n",
-                               (unsigned long long)get_timestamp_ms(), global_lvl, cat_count);
-                        fflush(server->log_file);
-                    }
+            case ETCPMON_CMD_SET_DEBUG_CONFIG: {
+                if (payload_size < 2 || payload[1] > ETCPMON_MAX_DEBUG_CATEGORIES || payload_size != 2u + payload[1]) {
+                    DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Bad SET_DEBUG_CONFIG payload size=%u categories=%d",
+                                payload_size, payload_size >= 2 ? payload[1] : -1);
+                    close_client(server, client);
+                    return;
+                }
+                uint8_t global_lvl = payload[0];
+                uint8_t cat_count = payload[1];
+                if (global_lvl <= DEBUG_LEVEL_TRACE) debug_set_level((debug_level_t)global_lvl);
+                const uint8_t* levels = payload + 2;
+                for (uint8_t i = 0; i < cat_count && i + 1 < DEBUG_CATEGORY_COUNT; i++) {
+                    if (levels[i] <= DEBUG_LEVEL_TRACE)
+                        debug_set_category_level((debug_category_t)(i + 1), (debug_level_t)levels[i]);
+                }
+                DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Debug config applied: global=%d categories=%d", global_lvl, cat_count);
+                if (server->log_file) {
+                    fprintf(server->log_file, "%llu: [LOG] Debug config set: global=%d categories=%d\n",
+                           (unsigned long long)get_timestamp_ms(), global_lvl, cat_count);
+                    fflush(server->log_file);
                 }
                 break;
+            }
 
             case ETCPMON_CMD_SUBSCRIBE_NODES:
                 client->subscribed_nodes = 1;
@@ -871,6 +912,11 @@ static void handle_client_data(struct control_server* server, struct control_cli
         /* Reset idle timer on each valid command */
         if (client->idle_timer) { uasync_cancel_timeout(server->ua, client->idle_timer); }
         client->idle_timer = uasync_set_timeout(server->ua, 300000, client, control_client_idle_cb, "ctrl_idle");
+        if (!client->idle_timer) {
+            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to reset control client idle timer fd=%d", (int)client->fd);
+            close_client(server, client);
+            return;
+        }
 
         /* Remove processed message from buffer */
         uint16_t msg_size = hdr->size;
@@ -957,12 +1003,18 @@ static void send_socket_list(struct control_server* server, struct control_clien
     uint8_t count = 0;
     struct ETCP_SOCKET* sock = instance->etcp_sockets;
     while (sock && count < ETCPMON_MAX_CONNECTIONS) {
-        if (!sock->probe_only) count++;
+        if (!sock->probe_only) {
+            if (ETCPMON_SOCKET_LIST_SIZE(count + 1u) > ETCPMON_MAX_MSG_SIZE) {
+                DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Socket list exceeds protocol limit: count>%u next_size=%zu max_size=%u",
+                           count, ETCPMON_SOCKET_LIST_SIZE(count + 1u), ETCPMON_MAX_MSG_SIZE);
+                send_error(client, ETCPMON_ERR_SERVER_BUSY, "Socket list exceeds protocol limit", seq_id);
+                return;
+            }
+            count++;
+        }
         sock = sock->next;
     }
 
-    if (count > ETCPMON_MAX_CONNECTIONS) count = ETCPMON_MAX_CONNECTIONS;
-
     uint16_t rsp_size = ETCPMON_SOCKET_LIST_SIZE(count);
     uint8_t* buffer = (uint8_t*)u_malloc(rsp_size);
     if (!buffer) {
@@ -1241,6 +1293,7 @@ static void send_metrics(struct control_server* server, struct control_client* c
     struct etcpmon_link_metrics* link_info = (struct etcpmon_link_metrics*)(buffer + sizeof(*hdr) + sizeof(*rsp));
     link = conn->links;
     for (uint8_t i = 0; i < links_count && link; i++) {
+        memset(&link_info[i], 0, sizeof(link_info[i]));
         link_info[i].local_link_id = link->local_link_id;
         link_info[i].status = link->link_status;
         link_info[i].encrypt_errors = (uint32_t)link->encrypt_errors;
@@ -1278,8 +1331,6 @@ static void send_metrics(struct control_server* server, struct control_client* c
             uint32_t max_bw = link->bbr->bw_hi[0] > link->bbr->bw_hi[1] ? link->bbr->bw_hi[0] : link->bbr->bw_hi[1];
             link_info[i].bbr_bw_hi = max_bw ? (uint32_t)((uint64_t)max_bw * link->mtu * 1000000ULL / 16777216ULL) : 0;
             link_info[i].bbr_bw_lo = link->bbr->bw_lo != ~0U ? (uint32_t)((uint64_t)link->bbr->bw_lo * link->mtu * 1000000ULL / 16777216ULL) : 0;
-        } else {
-            memset(&link_info[i].bbr_mode, 0, 11 * 4 + 2 + 2); // zero out all BBR fields
         }
         link = link->next;
     }

src/control_server.h

@@ -1,7 +1,8 @@
 /*
  * control_server.h - Control Socket Server for ETCP Monitoring
  * 
- * Provides TCP control interface for ETCP connection monitoring
+ * TCP-мониторинг ETCP: init → обработка команд в uasync-потоке → shutdown.
+ * Метрики отправляются по запросу; доступ — по IPv4 control_allow, idle timeout — 30 секунд.
  */
 
 #ifndef CONTROL_SERVER_H
@@ -40,6 +41,7 @@ struct control_client {
     uint64_t selected_peer_id;          /* 0 = none selected */
 
     struct ll_queue* output_queue;      /* Output queue for async sends */
+    uint16_t output_offset;             /* Отправленные байты головы очереди; dgram/len не меняются */
     void* idle_timer;                  /* Client idle timeout */
     uint8_t subscribed_nodes;           /* 1 = subscribed to node updates */
     uint8_t write_registered;           /* 1 = write events registered in uasync */
@@ -97,10 +99,8 @@ int control_server_init(struct control_server* server,
  */
 void control_server_shutdown(struct control_server* server);
 
-/* Process periodic updates
- * 
- * Should be called periodically (e.g., from main loop) to send
- * metrics updates to clients that have selected connections
+/* Разобрать накопленные команды всех клиентов в uasync-потоке.
+ * Чтение сокета также запускает разбор; метрики отправляются только по CMD_GET_METRICS.
  * 
  * Parameters:
  *   server - Server state