Reject stale NCD notifications and validate complete control frames

src/transport_layer/node_conn_direct.c

@@ -43,6 +43,7 @@ struct ncd_entry {
     struct UASYNC*              ua;
     struct UTUN_INSTANCE*      inst;
     unsigned                   dispatch_depth;
+    uint64_t                   event_generation;
     uint8_t                    closed;
     uint8_t                    node_ref;
     uint64_t                   node_timestamp;
@@ -132,7 +133,14 @@ static struct TOPO_NODE* ncd_lookup_node(struct UTUN_INSTANCE* inst, uint64_t no
     }
     if (inst->topo_sqlite_db && inst->topo_groups) {
         struct TOPO_NODE* ni = topo_node_sqlite_node_load(inst->topo_sqlite_db, inst->topo_groups, node_id);
-        if (ni) return topo_node_registry_store(inst->topo_groups, ni);
+        if (ni) {
+            struct TOPO_NODE* stored = topo_node_registry_store(inst->topo_groups, ni);
+            if (!stored) {
+                DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot register loaded node=%016llx", (unsigned long long)node_id);
+                topo_node_destroy(inst->topo_groups, ni);
+            }
+            return stored;
+        }
     }
     return NULL;
 }
@@ -469,8 +477,13 @@ static void ncd_event_dispatch(struct ncd_entry* entry, enum ncd_event event) {
     DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] event=%d node=%016llx handles=%d depth=%u", event,
                (unsigned long long)entry->node_id, entry->handle_count, entry->dispatch_depth);
     entry->dispatch_depth++;
+    uint64_t generation = ++entry->event_generation;
     // Закрытые элементы не освобождаются до выхода из внешней рассылки. Новые добавляются в голову.
     for (struct NODE_CONN_DIRECT* h = entry->handles; h; h = h->next) {
+        if (entry->event_generation != generation) {
+            DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] stop superseded event=%d node=%016llx", event, (unsigned long long)entry->node_id);
+            break;
+        }
         if (event != NCD_EVENT_UP) ncd_cancel_deliver_up(h, entry->ua);
         if (!h->closed && h->cb && (!entry->closed || event == NCD_EVENT_CLOSED)) h->cb(h, event, h->cb_arg);
     }
@@ -661,15 +674,17 @@ static void ncd_fin_wait_timeout_cb(void* arg) {
  */
 
 static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e) {
-    if (!conn || !e || e->len < NCD_CONTROL_MSG_SIZE) {
-        DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] malformed control conn=%p size=%u", (void*)conn, e ? e->len : 0);
+    if (!conn || !e || !e->dgram || e->len != NCD_CONTROL_MSG_SIZE || e->dgram[0] != ETCP_ID_NCD_CONTROL) {
+        DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] malformed control conn=%p size=%zu", (void*)conn, e ? (size_t)e->len : 0);
         if (e) { queue_dgram_free(e); queue_entry_free(e); }
         return;
     }
-    struct ncd_control_msg* msg = (struct ncd_control_msg*)e->dgram;
+    struct ncd_control_msg message;
+    memcpy(&message, e->dgram, sizeof(message));
+    const struct ncd_control_msg* msg = &message;
     uint64_t sender_id = msg->node_id;
     struct ncd_entry* entry = ncd_registry_find(conn->instance, sender_id);
-    if (sender_id != conn->peer_node_id || (entry && entry->conn != conn)) {
+    if (!sender_id || sender_id != conn->peer_node_id || (entry && entry->conn != conn)) {
         DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] control from wrong transport node=%016llx", (unsigned long long)sender_id);
         queue_dgram_free(e); queue_entry_free(e); return;
     }
@@ -1345,11 +1360,11 @@ int ncd_remove_socket_links(struct UTUN_INSTANCE* inst, struct ETCP_SOCKET* sock
         if (etcp_conn_ref_take(e->conn) == 0) conns[held++] = e->conn;
     // DOWN callback может удалить текущий и следующий entry. Снимок удерживает только транспорт.
     for (size_t i = 0; i < held; i++) {
-        struct ETCP_LINK* link = conns[i]->links;
-        while (link) {
-            struct ETCP_LINK* next = link->next;
-            if (link->conn == sock) { etcp_link_close(link); removed++; }
-            link = next;
+        for (;;) {
+            struct ETCP_LINK* link = conns[i]->links;
+            while (link && link->conn != sock) link = link->next;
+            if (!link) break;
+            etcp_link_close(link); removed++;
         }
         etcp_conn_ref_free(conns[i]);
     }

tests/test_ncd_faults.c

@@ -147,11 +147,19 @@ static void control_length(void) {
     CHECK(node_conn_direct_open(inst, 42, NULL, NULL, &handle, NULL) == NCD_REUSED);
     node_conn_direct_close(handle); CHECK(conn->fin_wait);
     struct ncd_control_msg message = { ETCP_ID_NCD_CONTROL, NCD_SUBCMD_KEEP_ALIVE, 42 };
-    struct ll_entry* packet = ll_alloc_lldgram(sizeof(message) + 1); CHECK(packet);
-    memcpy(packet->dgram, &message, sizeof(message)); packet->dgram[sizeof(message)] = 0;
-    packet->len = sizeof(message) + 1;
-    inst->api_bindings.callbacks[ETCP_ID_NCD_CONTROL](conn, packet);
-    CHECK(conn->fin_wait);
+    for (size_t len = 0; len <= sizeof(message) + 1; len++) {
+        if (len == sizeof(message)) continue;
+        struct ll_entry* packet = ll_alloc_lldgram(sizeof(message) + 1); CHECK(packet);
+        memcpy(packet->dgram, &message, sizeof(message)); packet->dgram[sizeof(message)] = 0;
+        packet->len = len;
+        inst->api_bindings.callbacks[ETCP_ID_NCD_CONTROL](conn, packet);
+        CHECK(conn->fin_wait);
+    }
+    struct ll_entry* empty = queue_entry_new(0); CHECK(empty); empty->len = sizeof(message);
+    inst->api_bindings.callbacks[ETCP_ID_NCD_CONTROL](conn, empty); CHECK(conn->fin_wait);
+    struct ll_entry* wrong = ll_alloc_lldgram(sizeof(message)); CHECK(wrong);
+    memcpy(wrong->dgram, &message, sizeof(message)); wrong->dgram[0] = 0; wrong->len = sizeof(message);
+    inst->api_bindings.callbacks[ETCP_ID_NCD_CONTROL](conn, wrong); CHECK(conn->fin_wait);
     test_clock_run_for(ua, 51000); CHECK(!inst->ncd_registry); teardown();
 }