From f00a5bd306f15eec232212654d92b9e7bc80ab00 Mon Sep 17 00:00:00 2001
From: evgeny <evgeny@adaptto.ru>
Date: Fri, 23 Oct 2026 13:45:23 +0200
Subject: [PATCH] Apply INIT reachability before reply and publish guarded RX
 transitions

---
 src/transport_layer/etcp_api.c         |  17 ++-
 src/transport_layer/etcp_connections.c | 181 +++++++++++++------------
 tests/test_etcp_init_liveness.c        |   4 +-
 3 files changed, 110 insertions(+), 92 deletions(-)

diff --git a/src/transport_layer/etcp_api.c b/src/transport_layer/etcp_api.c
index 267f027d..72cb0531 100644
--- a/src/transport_layer/etcp_api.c
+++ b/src/transport_layer/etcp_api.c
@@ -93,13 +93,22 @@ void etcp_remove_link_status_cbk(struct UTUN_INSTANCE* inst, etcp_link_status_cb
 
 void etcp_fire_link_status_cbk(struct ETCP_LINK* link, int old_state, int old_status) {
     if (!link || !link->etcp || !link->etcp->instance) return;
-    link->etcp->callbacks_running++;
-    struct etcp_link_status_cbk_entry* cbe = link->etcp->instance->link_status_cbks;
+    struct ETCP_CONN* conn = link->etcp;
+    int was_attached = 0;
+    for (struct ETCP_LINK* current = conn->links; current; current = current->next) if (current == link) { was_attached = 1; break; }
+    conn->callbacks_running++;
+    struct etcp_link_status_cbk_entry* cbe = conn->instance->link_status_cbks;
     while (cbe) {
-        cbe->fn(link->etcp, link, old_state, old_status, cbe->arg);
+        cbe->fn(conn, link, old_state, old_status, cbe->arg);
+        if (was_attached) {
+            int attached = 0;
+            for (struct ETCP_LINK* current = conn->links; current; current = current->next)
+                if (current == link) { attached = 1; break; }
+            if (!attached) break;
+        }
         cbe = cbe->next;
     }
-    link->etcp->callbacks_running--;
+    conn->callbacks_running--;
 }
 
 static void etcp_status_cbk_add_chain(struct etcp_status_cbk_entry** head, etcp_conn_status_fn fn, void* arg) {
diff --git a/src/transport_layer/etcp_connections.c b/src/transport_layer/etcp_connections.c
index 8c0900b9..7bb06828 100644
--- a/src/transport_layer/etcp_connections.c
+++ b/src/transport_layer/etcp_connections.c
@@ -1976,9 +1976,54 @@ static int handle_pong(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt, const
     return 0;
 }
 
-// Сервер: собрать и отправить INIT_RESPONSE (mtu, link_id, NAT-адрес клиента, ed25519),
-// провести NAT-детекцию, поднять линк и запустить keepalive.
+/* Принятый RX доказывает прямую достижимость; обратную сообщает только входной flag_up.
+ * Вызывается до переиспользования pkt для ответа, без callbacks; возвращает прежний статус. */
+static int etcp_link_record_rx(struct ETCP_LINK* link, const struct ETCP_DGRAM* pkt) {
+    int previous = link->link_status;
+    link->recv_keepalive = 1;
+    link->remote_keepalive = pkt->flag_up;
+    link->link_status = link->is_tcp ? 1 : link->remote_keepalive;
+    link->last_recv_local_time = get_time_tb();
+    link->last_recv_timestamp = pkt->timestamp;
+    link->last_recv_updated = 1;
+    link->total_decrypted += pkt->data_len;
+    return previous;
+}
+
+/* Проверка заимствованного link после callback: conn удерживается вызывающим, сам link мог быть удалён. */
+static int etcp_link_rx_alive(struct ETCP_CONN* conn, struct ETCP_LINK* link) {
+    if (conn->close_requested) return 0;
+    for (struct ETCP_LINK* current = conn->links; current; current = current->next) if (current == link) return 1;
+    return 0;
+}
+
+/* Публикация готового состояния: callbacks могут запросить close или удалить линк.
+ * Вызывающий удерживает conn через callbacks_running; 0 запрещает дальнейшее использование link. */
+static int etcp_link_publish_rx(struct ETCP_LINK* link, int old_state, int old_status) {
+    struct ETCP_CONN* conn = link->etcp;
+    if (conn->close_requested) return 0;
+    if (old_state != link->link_state || old_status != link->link_status)
+        etcp_fire_link_status_cbk(link, old_state, old_status);
+    if (!etcp_link_rx_alive(conn, link)) return 0;
+    if (old_status != link->link_status)
+        DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] RX link=%u status=%d->%u recv=%u remote=%u state=%u init=%u",
+            conn->log_name, link->local_link_id, old_status, link->link_status, link->recv_keepalive,
+            link->remote_keepalive, link->link_state, link->initialized);
+    if (old_status && !link->link_status) etcp_on_link_down(conn, link);
+    else if (link->initialized && link->link_status) loadbalancer_link_ready(link);
+    return etcp_link_rx_alive(conn, link);
+}
+
+// Сервер: принять RX INIT, отправить ответ и запустить keepalive даже при неподтверждённом обратном пути.
 static void send_init_response(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, struct ETCP_CONN* conn, const struct ETCP_INIT_REQUEST_PKT* req, const struct sockaddr_storage* addr, uint8_t send_reset, uint32_t req_src_ip, uint16_t req_src_port, size_t pkt_len) {
+    if (conn->close_requested) { memory_pool_free(e_sock->instance->pkt_pool, pkt); return; }
+    conn->callbacks_running++;
+    int old_state = link->link_state, old_status = etcp_link_record_rx(link, pkt);
+    uint8_t client_socket_id = req->socket_id;
+    link->is_server = 1;
+    link->initialized = 1;
+    link->link_state = LINK_STATE_CONNECTED;
+    if (link->init_timer) { uasync_cancel_timeout(conn->instance->ua, link->init_timer); link->init_timer = NULL; }
     struct ETCP_INIT_RESPONSE_PKT* resp = (struct ETCP_INIT_RESPONSE_PKT*)pkt->data;
 
     // Set response code: 0x03 (with reset) or 0x05 (without reset)
@@ -2010,15 +2055,12 @@ static void send_init_response(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pk
     }
     DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "INIT client address claim=%s:%u observed_peer=%s bytes=%zu server_socket=%s client_sid=%u",
                 ip_to_str(&req_src_ip, AF_INET).str, req_src_port, sockaddr_storage_to_str(addr).str, pkt_len,
-                e_sock->name, req->socket_id);
+                e_sock->name, client_socket_id);
     memcpy(resp->ed25519_pubkey, e_sock->instance->my_ed25519_pubkey, SC_PUBKEY_SIZE);
     resp->device_type = e_sock->instance->client_type;
     wire_write16(resp->keepalive,e_sock->instance->keepalive_interval);
     pkt->noencrypt_len=0;
     pkt->link=link;
-    link->recv_keepalive = 1;
-    link->last_recv_local_time = get_time_tb();
-    link->last_recv_timestamp = pkt->timestamp;
 
     int xoffset=sizeof(struct ETCP_INIT_RESPONSE_PKT);
     // padding
@@ -2039,32 +2081,22 @@ static void send_init_response(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pk
     for (int i=0; i<to_add; i++) pkt->data[xoffset++]=rand();// fill pad
     // padding end
     pkt->data_len=xoffset;
-    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] INIT_RESPONSE sent (link=%d, mtu=%d, dst=%s)",
+    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] INIT_RESPONSE sending (link=%d, mtu=%d, dst=%s)",
                 link->etcp->log_name, link->local_link_id, link->mtu_local,
                 sockaddr_storage_to_str(&link->remote_addr).str);
-    etcp_encrypt_send(pkt);
+    if (etcp_encrypt_send(pkt) < 0)
+        DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] INIT_RESPONSE send failed link=%u", conn->log_name, link->local_link_id);
 
     memory_pool_free(e_sock->instance->pkt_pool, pkt);
-    if (!link->is_server) {
-        DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] link %u accepted incoming INIT: client -> server",
-                   link->etcp->log_name, link->local_link_id);
-        link->is_server = 1;
+    if (!conn->initialized) etcp_conn_ready(conn);
+    if (etcp_link_rx_alive(conn, link)) {
+        DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] INIT accepted link=%u status=%u recv=%u remote=%u", conn->log_name,
+            link->local_link_id, link->link_status, link->recv_keepalive, link->remote_keepalive);
+        etcp_keepalive_link_ready(link);
+        if (etcp_link_publish_rx(link, old_state, old_status) && conn->instance->nat_det)
+            nat_detection_link_ready(conn->instance->nat_det, link);
     }
-    link->initialized = 1;
-    { int old_state = link->link_state; link->link_state = LINK_STATE_CONNECTED; etcp_fire_link_status_cbk(link, old_state, link->link_status); }
-    if (link->init_timer) {
-        uasync_cancel_timeout(link->etcp->instance->ua, link->init_timer);
-        link->init_timer = NULL;
-    }
-    if (link->etcp->initialized == 0) {
-      etcp_conn_ready(link->etcp);
-      DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] Connection established (socket=%s, link=%d, addr=%s)", link->etcp->log_name, e_sock->name, link->local_link_id, sockaddr_storage_to_str(&link->remote_addr).str);
-      }
-    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] Link %d UP (server, mtu=%d, addr=%s)", link->etcp->log_name, link->local_link_id, link->mtu_local, sockaddr_storage_to_str(&link->remote_addr).str);
-    etcp_keepalive_link_ready(link);
-    loadbalancer_link_ready(link);
-    // Restart NAT check after link is up (e.g. after address change or reinit)
-    if (link->etcp->instance->nat_det) nat_detection_link_ready(link->etcp->instance->nat_det, link);
+    conn->callbacks_running--;
 }
 
 // Клиент: обработка INIT_RESPONSE — MTU, link_id, NAT-адрес, ed25519, keepalive, подъём линка.
@@ -2081,6 +2113,8 @@ static int handle_init_response_client(struct ETCP_SOCKET* e_sock, struct ETCP_D
     if (server_node_id != sc_derive_node_id_from_pubkey(link->etcp->crypto_ctx.peer_public_key)) {
         DEBUG_WARN(DEBUG_CATEGORY_ETCP, "INIT_RESPONSE node ID does not match authenticated peer key"); return 46;
     }
+    struct ETCP_CONN* conn = link->etcp;
+    int old_state = link->link_state, old_status = etcp_link_record_rx(link, pkt);
     uint64_t resp_reset_id = wire_read64(resp->reset_id);
     DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "INIT_RESPONSE rid=%016llx", (unsigned long long)resp_reset_id);
     etcp_conn_apply_peer_reset_id(link->etcp, resp_reset_id);
@@ -2124,26 +2158,20 @@ static int handle_init_response_client(struct ETCP_SOCKET* e_sock, struct ETCP_D
     etcp_update_log_name(link->etcp);
 
     link->initialized = 1;// получен init response (client)
-    { int old_state = link->link_state; link->link_state = LINK_STATE_CONNECTED; etcp_fire_link_status_cbk(link, old_state, link->link_status); }
+    link->link_state = LINK_STATE_CONNECTED;
     if (link->init_timer) {
         uasync_cancel_timeout(link->etcp->instance->ua, link->init_timer);
         link->init_timer = NULL;
     }
 
-    if (link->etcp->initialized == 0) {
-      etcp_conn_ready(link->etcp);
-      }
-
-    loadbalancer_link_ready(link);
-    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] Link %d UP (client, mtu=%d, addr=%s)", link->etcp->log_name, link->local_link_id, link->mtu, sockaddr_storage_to_str(&link->remote_addr).str);
-
-    etcp_keepalive_link_ready(link);
-
-    loadbalancer_link_ready(link);
-    
-    DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "etcp client: Link initialized successfully! Server node_id=%016llx, mtu=%d, local_link_id=%d, remote_link_id=%d", (unsigned long long)server_node_id, link->mtu, link->local_link_id, link->remote_link_id);
-    
     memory_pool_free(e_sock->instance->pkt_pool, pkt);
+    if (!conn->initialized) etcp_conn_ready(conn);
+    if (etcp_link_rx_alive(conn, link)) {
+        DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] INIT_RESPONSE accepted link=%u status=%u recv=%u remote=%u mtu=%u",
+            conn->log_name, link->local_link_id, link->link_status, link->recv_keepalive, link->remote_keepalive, link->mtu);
+        etcp_keepalive_link_ready(link);
+        etcp_link_publish_rx(link, old_state, old_status);
+    }
     return 0;
 }
 
@@ -2598,17 +2626,22 @@ ec_fr:
 int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
                            struct ETCP_LINK* link, size_t pkt_len) {
     DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "Decrypt ok - normal pkt");
-    if (pkt_len<3) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "decrypted packet too small, size=%zu", pkt_len); return 46; }
+    struct ETCP_CONN* conn = link->etcp;
+    uint32_t reset = conn->reset_count;
+    int result = 0;
+    conn->callbacks_running++;
+    if (pkt_len<3) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "decrypted packet too small, size=%zu", pkt_len); result = 46; goto done; }
     pkt->data_len=pkt_len-3;
     pkt->noencrypt_len=0;
     pkt->link=link;
-    if (!pkt->data_len) { DEBUG_WARN(DEBUG_CATEGORY_ETCP,"empty decrypted ETCP packet"); return 46; }
+    if (!pkt->data_len) { DEBUG_WARN(DEBUG_CATEGORY_ETCP,"empty decrypted ETCP packet"); result = 46; goto done; }
+    if (conn->close_requested) { memory_pool_free(conn->instance->pkt_pool, pkt); goto done; }
     if (etcp_session_receive(link->etcp,pkt->data,pkt->data_len)) {
-        memory_pool_free(link->etcp->instance->pkt_pool,pkt); return 0;
+        memory_pool_free(conn->instance->pkt_pool,pkt); goto done;
     }
     if (pkt->data[0] == ETCP_SESSION_DATA || pkt->data[0] <= 1) {
         if (!etcp_session_accept_data(link->etcp,pkt->data,pkt->data_len)) {
-            memory_pool_free(link->etcp->instance->pkt_pool,pkt); return 0;
+            memory_pool_free(conn->instance->pkt_pool,pkt); goto done;
         }
         pkt->data_len -= ETCP_SESSION_HEADER_SIZE;
         memmove(pkt->data,pkt->data+ETCP_SESSION_HEADER_SIZE,pkt->data_len);
@@ -2620,39 +2653,17 @@ int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
                     link->recv_keepalive, (void*)e_sock);
     }
 
-    link->remote_keepalive = pkt->flag_up;
-    if (link->recv_keepalive != 1) {
-        link->recv_keepalive = 1;
-        DEBUG_DEBUG(DEBUG_CATEGORY_KEEPALIVE, "[%s] Link %d recv_keepalive restored to 1 (was 0) link_status=%d remote_ka=%d state=%d init=%d links_up=%d",
-                   link->etcp->log_name, link->local_link_id, link->link_status, link->remote_keepalive, link->link_state, link->initialized, link->etcp->links_up);
+    uint8_t pkt_code = pkt->data[0];
+    if (pkt_code == ETCP_INIT_RESPONSE || pkt_code == ETCP_INIT_RESPONSE_NOINIT) {
+        result = handle_init_response_client(e_sock, pkt, link, pkt_code, pkt_len);
+        goto done;
     }
-
-    int was_up = link->link_status;
-    link->link_status = link->is_tcp ? link->recv_keepalive : (link->remote_keepalive && link->recv_keepalive);
-    if (link->link_status != was_up)
-        etcp_fire_link_status_cbk(link, link->link_state, was_up);
-    if (link->link_status && !was_up && link->initialized) {
-        DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] Link %d status popped UP: recv_ka=%d remote_ka=%d state=%d init=%d", link->etcp->log_name, link->local_link_id, link->recv_keepalive, link->remote_keepalive, link->link_state, link->initialized);
-        loadbalancer_link_ready(link);
-    }
-
-    link->last_recv_local_time=get_time_tb();
-    link->last_recv_timestamp=pkt->timestamp;
-    link->last_recv_updated=1;
-    link->total_decrypted += pkt->data_len;
-    if (was_up && !link->link_status) {
-        struct ETCP_CONN* conn = link->etcp;
-        DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] peer reports link DOWN id=%u remote_ka=%u", conn->log_name,
-                   link->local_link_id, link->remote_keepalive);
-        etcp_on_link_down(conn, link);
-        int alive = 0;
-        for (struct ETCP_LINK* current = conn->links; current; current = current->next) if (current == link) { alive = 1; break; }
-        if (conn->close_requested || !alive) { memory_pool_free(conn->instance->pkt_pool, pkt); return 0; }
+    int old_status = etcp_link_record_rx(link, pkt);
+    if (!etcp_link_publish_rx(link, link->link_state, old_status) || conn->reset_count != reset) {
+        memory_pool_free(conn->instance->pkt_pool, pkt); goto done;
     }
     if (pkt->data[0] != ETCP_KEEPALIVE) etcp_keepalive_watch_rx(link);
 
-    uint8_t pkt_code = pkt->data[0];
-
     if (pkt_code == ETCP_TCP_ADDRESS) {
         struct UTUN_INSTANCE *inst = link->etcp->instance;
         static const uint8_t zero[12];
@@ -2661,11 +2672,11 @@ int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
             (pkt->data[1] == 4 && memcmp(pkt->data + 6, zero, sizeof(zero)))) {
             DEBUG_WARN(DEBUG_CATEGORY_NAT, "TCP IP report invalid transport/format peer=%s size=%u",
                        link->etcp->log_name, pkt->data_len);
-            memory_pool_free(inst->pkt_pool, pkt); return 0;
+            memory_pool_free(inst->pkt_pool, pkt); goto done;
         }
         if (!link->is_server && inst->config && inst->config->global.socks_enabled) {
             DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "TCP IP report ignored through SOCKS peer=%s", link->etcp->log_name);
-            memory_pool_free(inst->pkt_pool, pkt); return 0;
+            memory_pool_free(inst->pkt_pool, pkt); goto done;
         }
         struct sockaddr_storage observed = {0};
         if (pkt->data[1] == 4) {
@@ -2677,7 +2688,7 @@ int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
         const struct sockaddr_storage *local = stcp_link_get_local_addr(link->tcp_link);
         if (local && inst->nat_det && link->conn) nat_detection_tcp_observed(inst->nat_det, link, local, &observed);
         else if (!local) DEBUG_WARN(DEBUG_CATEGORY_NAT, "TCP IP report cannot read local socket peer=%s", link->etcp->log_name);
-        memory_pool_free(inst->pkt_pool, pkt); return 0;
+        memory_pool_free(inst->pkt_pool, pkt); goto done;
     }
 
 #ifdef UTUN_HAVE_STANDBY
@@ -2692,18 +2703,14 @@ int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
 #endif
 
     if (etcp_keepalive_on_recv(pkt)) {
-        return 0;
-    }
-
-    if (pkt_code == ETCP_INIT_RESPONSE || pkt_code == ETCP_INIT_RESPONSE_NOINIT) {
-        int ret = handle_init_response_client(e_sock, pkt, link, pkt_code, pkt_len);
-        if (ret) return ret;
-        return 0;
+        goto done;
     }
     if (link->link_state == LINK_STATE_TRY_RECONNECT) {// 0 - just init, 1 - handshake, 2 - try reconnect, 3 - connected
         etcp_keepalive_link_ready(link);
         int old_state = link->link_state; link->link_state = LINK_STATE_CONNECTED;
-        etcp_fire_link_status_cbk(link, old_state, link->link_status);
+        if (!etcp_link_publish_rx(link, old_state, link->link_status) || conn->reset_count != reset) {
+            memory_pool_free(conn->instance->pkt_pool, pkt); goto done;
+        }
     }
 
     if (link->link_state == LINK_STATE_CONNECTED) {
@@ -2715,7 +2722,9 @@ int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
     } else {
         memory_pool_free(link->etcp->instance->pkt_pool, pkt);
     }
-    return 0;
+done:
+    conn->callbacks_running--;
+    return result;
 }
 
 /* Создаёт один серверный сокет из CFG_SERVER: UDP (etcp_socket_add) или TCP
diff --git a/tests/test_etcp_init_liveness.c b/tests/test_etcp_init_liveness.c
index 7b40d30a..bcb979e8 100644
--- a/tests/test_etcp_init_liveness.c
+++ b/tests/test_etcp_init_liveness.c
@@ -166,8 +166,8 @@ int main(void) {
     if (getenv("UTUN_TEST_DEBUG")) debug_set_category_level(DEBUG_CATEGORY_ETCP, DEBUG_LEVEL_DEBUG);
     test_clock_init(10); CHECK(socket_platform_init() == 0);
     size_t baseline = u_get_allocated_count();
-    for (unsigned swapped = 0; swapped < 2; swapped++) for (unsigned blocked = 0; blocked < 2; blocked++) {
-        one_way(swapped, blocked); CHECK(u_get_allocated_count() == baseline);
+    for (unsigned swapped = 0; swapped < 2; swapped++) for (unsigned order = 0; order < 2; order++) {
+        one_way(swapped, 1 - order); CHECK(u_get_allocated_count() == baseline);
     }
     for (unsigned action = 1; action <= 2; action++) { close_in_callback(action); CHECK(u_get_allocated_count() == baseline); }
     DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS encrypted INIT liveness: cold start, one-way loss, repeated INIT, recovery and callback close");