Measure helper endpoints and protect controlled UDP ingress trials

src/routing_layer/nat_detection.c

@@ -530,16 +530,23 @@ static void select_profile(struct nat_socket *state) {
 }
 
 static void owner_finish(struct nat_socket *state, uint8_t reason) {
+    int next_pair = reason && state->phase && state->cycling && !state->using_fallback &&
+        reason != NAT_REASON_CANCELLED && reason != NAT_REASON_ADDRESS_CHANGED;
     etcp_cancel_pings(state->detector->instance, owner_ping_cb, state);
     state->active_ping = 0; state->phase = NAT_IDLE; state->reason = reason;
     if (reason == NAT_REASON_NO_OBSERVER || reason == NAT_REASON_NO_HELPER) state->fallback_allowed = 1;
-    state->preferred_observer = 0;
+    if (next_pair) {
-    state->due = get_time_tb() + (reason == NAT_REASON_ADDRESS_CHANGED ? 2500 :
+        state->forced = 1; state->due = get_time_tb() + 1000; state->helper_reselections++;
-        state->socket->address_profile.source || !reason ? NAT_REFRESH_TB : NAT_RETRY_TB);
+    } else {
+        state->cycling = 0; state->using_fallback = 0; state->preferred_observer = 0;
+        state->cursor = (struct nat_pair){0}; state->fallback = (struct nat_pair){0};
+        state->due = get_time_tb() + (reason == NAT_REASON_ADDRESS_CHANGED ? 2500 :
+            state->socket->address_profile.source || !reason ? NAT_REFRESH_TB : NAT_RETRY_TB);
+    }
     if (reason) {
-        DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT round incomplete socket=%s generation=%016llx round=%016llx reason=%s",
+        DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT round incomplete socket=%s group=%016llx generation=%016llx round=%016llx reason=%s next_pair=%d",
-                   state->socket->name, (unsigned long long)state->socket->generation,
+                   state->socket->name, (unsigned long long)state->group_id, (unsigned long long)state->socket->generation,
-                   (unsigned long long)state->round, nat_check_reason_name(reason));
+                   (unsigned long long)state->round, nat_check_reason_name(reason), next_pair);
     }
     select_profile(state);
 }
@@ -568,17 +575,14 @@ static void owner_apply(struct nat_socket *state, uint8_t mapping, uint8_t inbou
     owner_finish(state, NAT_REASON_NONE); state->reason = partial_reason;
 }
 
-static int owner_find(struct nat_socket *state) {
-    struct nat_wire message; make_message(state, &message, NAT_FIND);
-    state->phase = NAT_FIND_C;
-    return send_control(state->detector, state->observer, &message);
-}
-
 static int owner_run(struct nat_socket *state, uint8_t step) {
     if (random_id(&state->token)) return -1;
-    struct nat_wire message; make_message(state, &message, NAT_RUN); message.step = step;
+    struct nat_wire message;
-    state->phase = step == NAT_STEP_TCP ? NAT_TCP : step == NAT_STEP_PRE ? NAT_PRE : NAT_POST;
+    if (make_message(state, &message, NAT_RUN)) return -1;
-    return send_control(state->detector, state->observer, &message);
+    message.step = step;
+    state->phase = step == NAT_STEP_TCP ? NAT_TCP : step == NAT_STEP_PREPARE ? NAT_PREPARE_C :
+                   step == NAT_STEP_PRE ? NAT_PRE : step == NAT_STEP_ALT ? NAT_PREPARE_ALT : NAT_POST;
+    return send_control(state->detector, state->group_id, state->observer, &message);
 }
 
 static int owner_observe(struct nat_socket *state) {
@@ -589,11 +593,12 @@ static int owner_observe(struct nat_socket *state) {
         key = state->helper_key;
     }
     int timeout = 1000;
-    struct ETCP_CONN *connection = live_connection(state->detector, state->observer);
+    struct ETCP_CONN *connection = topo_group_find_conn_for_node(
+        topo_groups_find(state->detector->instance->topo_groups, state->group_id), state->observer);
     for (struct ETCP_LINK *link = connection ? connection->links : NULL; link; link = link->next)
         if (link->rtt_last / 10 * 4 > timeout) timeout = link->rtt_last / 10 * 4;
     if (timeout > 3000) timeout = 3000;
-    if (route_ping_observe(state->socket, key, target, state->round, timeout, owner_ping_cb, state)) return -1;
+    if (route_ping_observe(state->socket, key, target, state->group_id, state->round, timeout, owner_ping_cb, state)) return -1;
     state->active_ping = 1; return 0;
 }
 
@@ -603,7 +608,7 @@ static void owner_ping_cb(int success, uint16_t rtt, void *arg, uint64_t nonce,
     state->active_ping = 0;
     if (state->phase == NAT_IDLE) return;
     struct sockaddr_storage observed;
-    if (!success || route_ping_observation(data, length, state->socket, state->round, &observed)) {
+    if (!success || route_ping_observation(data, length, state->socket, state->group_id, state->round, &observed)) {
         if (++state->retries < NAT_PROBE_COUNT) { state->due = get_time_tb() + 1000; return; }
         owner_finish(state, success ? NAT_REASON_INVALID_OBSERVATION : NAT_REASON_TIMEOUT); return;
     }
@@ -615,15 +620,21 @@ static void owner_ping_cb(int success, uint16_t rtt, void *arg, uint64_t nonce,
         if ((state->measured_profile.mapping == ADDRESS_MAP_UNCHANGED && !route_ping_addr_equal(&state->local, &observed, 1)) ||
             (state->socket->nat_addr.ss_family && !route_ping_addr_equal(&state->socket->nat_addr, &observed, 1)))
             clear_profile(state, NAT_REASON_ADDRESS_CHANGED);
-        if (owner_find(state)) owner_finish(state, NAT_REASON_CONTROL_LOST);
+        if (owner_run(state, NAT_STEP_PREPARE)) owner_finish(state, NAT_REASON_CONTROL_LOST);
         return;
     }
     if (state->phase == NAT_OBSERVE_C) {
-        state->helper_seen = observed; state->phase = NAT_OBSERVE_ALT;
+        state->helper_seen = observed;
+        if (route_ping_addr_equal(&state->baseline, &observed, 1)) {
+            if (owner_run(state, NAT_STEP_ALT)) owner_finish(state, NAT_REASON_CONTROL_LOST);
+            return;
+        }
+        state->phase = NAT_REPEAT_B;
     } else if (state->phase == NAT_OBSERVE_ALT) {
         state->alternate_seen = observed;
         if (state->filter_clean && state->pre_healthy && !state->pre_received &&
-            route_ping_addr_equal(&state->baseline, &state->helper_seen, 1)) {
+            route_ping_addr_equal(&state->baseline, &state->helper_seen, 1) &&
+            route_ping_addr_equal(&state->baseline, &state->alternate_seen, 1)) {
             if (owner_run(state, NAT_STEP_POST)) owner_finish(state, NAT_REASON_CONTROL_LOST);
             return;
         }
@@ -653,16 +664,21 @@ void nat_detection_receive_ingress(struct NAT_DETECTION *detector, struct ETCP_S
     if (!state || !tag || length != NAT_QUERY_TAG_SIZE) {
         DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT ingress ignored without pending endpoint/valid tag length=%zu", length); return;
     }
-    uint8_t expected[ROUTE_OBSERVE_TAG_SIZE]; route_ping_make_tag(expected, socket, state->round);
+    uint8_t expected[ROUTE_OBSERVE_TAG_SIZE]; route_ping_make_tag(expected, socket, state->group_id, state->round);
     uint8_t step = state->phase == NAT_PRE ? NAT_STEP_PRE : state->phase == NAT_POST ? NAT_STEP_POST : 0;
-    if (!step || memcmp(tag, expected, sizeof(expected)) || tag[25] != step || read64(tag + 26) != state->token ||
+    if (!step || memcmp(tag, expected, sizeof(expected)) || tag[ROUTE_OBSERVE_TAG_SIZE] != step ||
+        read64(tag + ROUTE_OBSERVE_TAG_SIZE + 1) != state->token ||
         sc_derive_node_id_from_pubkey(peer_key) != state->helper || memcmp(peer_key, state->helper_key, 32) ||
-        !route_ping_addr_equal(source, &state->helper_address, 1) ||
+        source->ss_family != state->local.ss_family || !public_address(detector, source) ||
+        ((step == NAT_STEP_POST || state->pre_received) && !route_ping_addr_equal(source, &state->helper_address, 1)) ||
         route_ping_addr_equal(source, &state->observer_address, 0)) {
         DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT ingress correlation/source mismatch socket=%s phase=%u round=%016llx source=%s",
                    socket->name, state->phase, (unsigned long long)state->round, sockaddr_storage_to_str(source).str); return;
     }
-    if (step == NAT_STEP_PRE) state->pre_received = 1;
+    if (step == NAT_STEP_PRE) {
+        if (!state->pre_received) { state->filter_clean = !contacted(state, source); state->helper_address = *source; }
+        state->pre_received = 1;
+    }
     else state->post_received = 1;
     DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT direct ingress socket=%s round=%016llx step=%u source=%s clean=%u",
                 socket->name, (unsigned long long)state->round, step, sockaddr_storage_to_str(source).str, state->filter_clean);
@@ -676,6 +692,8 @@ static struct ETCP_SOCKET *auxiliary_socket(struct NAT_DETECTION *detector, stru
     struct CFG_SERVER server; memset(&server, 0, sizeof(server));
     server.ip = source->local_addr; server.netif_index = source->netif_index;
     server.type = CFG_SERVER_TYPE_PRIVATE; server.mtu = source->mtu; server.fib = -1;
+    for (const struct CFG_SERVER *cfg = detector->instance->config->servers; cfg; cfg = cfg->next)
+        if (!strcmp(cfg->name, source->name)) { server.so_mark = cfg->so_mark; server.fib = cfg->fib; break; }
     snprintf(server.name, sizeof(server.name), "nat-observer-%u", source->sock_id);
     if (server.ip.ss_family == AF_INET) ((struct sockaddr_in *)&server.ip)->sin_port = 0;
     else ((struct sockaddr_in6 *)&server.ip)->sin6_port = 0;
@@ -704,20 +722,64 @@ static void helper_finish(struct nat_probe *probe, uint8_t reason) {
     DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT helper done owner=%016llx round=%016llx step=%u sent=%u received=%u errors=%u reason=%s",
                 (unsigned long long)read64(probe->message.owner), (unsigned long long)read64(probe->message.round),
                 probe->message.step, probe->message.sent, probe->message.received, probe->message.errors, nat_check_reason_name(reason));
-    if (send_control(probe->detector, probe->reply, &probe->message))
+    if (send_control(probe->detector, probe->group_id, probe->reply, &probe->message))
         DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper result delivery failed");
     helper_remove(probe);
 }
 
+/* Входящая проба несёт идентификаторы owner; OBSERVE с помощника — идентификаторы его собственного socket. */
+static void ingress_tag(const struct nat_wire *message, uint64_t group_id, uint8_t tag[NAT_QUERY_TAG_SIZE]) {
+    memcpy(tag, "NAT3UDP!", 8); tag[8] = message->owner_socket;
+    memcpy(tag + 9, message->generation, 8); memcpy(tag + 17, message->round, 8); write64(tag + 25, group_id);
+    tag[ROUTE_OBSERVE_TAG_SIZE] = message->step;
+    memcpy(tag + ROUTE_OBSERVE_TAG_SIZE + 1, message->token, 8);
+}
+
 static void helper_ping_cb(int success, uint16_t rtt, void *arg, uint64_t nonce, const uint8_t *data, size_t length) {
-    (void)rtt; (void)nonce; (void)data; (void)length;
+    (void)rtt; (void)nonce;
     struct nat_probe *probe = arg;
     if (probe->cancelled) return;
     probe->active = 0; probe->tcp = NULL;
-    if (success) probe->message.received++;
+    uint8_t step = probe->message.step;
-    if (probe->message.step == NAT_STEP_TCP || probe->message.sent >= NAT_PROBE_COUNT)
+    if (step == NAT_STEP_PREPARE || step == NAT_STEP_ALT) {
-        helper_finish(probe, NAT_REASON_NONE);
+        struct ETCP_SOCKET *socket = find_socket(probe->detector, probe->io_socket, probe->io_generation);
-    else probe->due = get_time_tb() + 1000;
+        struct sockaddr_storage mapped;
+        if (success && socket && !route_ping_observation(data, length, socket, probe->group_id,
+                                                        read64(probe->message.round), &mapped)) {
+            uint8_t *endpoint = step == NAT_STEP_PREPARE ? probe->message.primary : probe->message.secondary;
+            if (route_ping_addr_encode(endpoint, &mapped)) { helper_finish(probe, NAT_REASON_INVALID_OBSERVATION); return; }
+            probe->message.received++;
+            DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT helper endpoint measured group=%016llx owner=%016llx step=%u socket=%s mapped=%s fallback=%u",
+                        (unsigned long long)probe->group_id, (unsigned long long)read64(probe->message.owner), step,
+                        socket->name, sockaddr_storage_to_str(&mapped).str, probe->alternate_fallback);
+            helper_finish(probe, NAT_REASON_NONE); return;
+        }
+        if (success) { helper_finish(probe, NAT_REASON_INVALID_OBSERVATION); return; }
+        if (probe->message.sent >= NAT_PROBE_COUNT) {
+            if (step == NAT_STEP_ALT && !probe->alternate_fallback) {
+                probe->alternate_fallback = 1; probe->message.sent = 0;
+                DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT alternate endpoint blocked by owner; observe through broker group=%016llx owner=%016llx",
+                            (unsigned long long)probe->group_id, (unsigned long long)read64(probe->message.owner));
+            } else { helper_finish(probe, NAT_REASON_TIMEOUT); return; }
+        }
+    } else if (step == NAT_STEP_TCP) {
+        if (success) probe->message.received++;
+        helper_finish(probe, success ? NAT_REASON_NONE : NAT_REASON_TIMEOUT); return;
+    } else {
+        if (success) {
+            uint8_t expected[NAT_QUERY_TAG_SIZE]; ingress_tag(&probe->message, probe->group_id, expected);
+            struct sockaddr_storage mapped;
+            if (length != ROUTE_ADDR_WIRE_SIZE + sizeof(expected) || memcmp(data + ROUTE_ADDR_WIRE_SIZE, expected, sizeof(expected)) ||
+                route_ping_addr_decode(data, &mapped) || route_ping_addr_encode(probe->message.primary, &mapped)) {
+                DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT ingress PONG has invalid endpoint/tag group=%016llx length=%zu",
+                           (unsigned long long)probe->group_id, length);
+                helper_finish(probe, NAT_REASON_INVALID_OBSERVATION); return;
+            }
+            probe->message.received++;
+        }
+        if (probe->message.sent >= NAT_PROBE_COUNT) { helper_finish(probe, NAT_REASON_NONE); return; }
+    }
+    probe->due = get_time_tb() + 1000;
 }
 
 static int helper_send(struct nat_probe *probe) {
@@ -732,16 +794,27 @@ static int helper_send(struct nat_probe *probe) {
         const struct TOPO_NODE *owner = topo_node_registry_find(detector->instance->topo_groups, read64(probe->message.owner));
         const struct TOPO_REALITY_SOCK *parameters = topo_node_find_reality_sock(owner, probe->message.owner_socket);
         if (parameters) { topo_node_reality_config(&reality, parameters); camouflage = &reality; }
-        result = etcp_send_tcp_ping(detector->instance, probe->message.owner_key, &target, timeout,
+        result = etcp_send_tcp_ping(detector->instance, probe->owner_key, &target, timeout,
                                     helper_ping_cb, probe, &probe->tcp, camouflage);
+    } else if (probe->message.step == NAT_STEP_PREPARE || probe->message.step == NAT_STEP_ALT) {
+        struct ETCP_SOCKET *source = find_socket(detector, probe->source_socket, probe->source_generation);
+        if (!source) { DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper source generation disappeared"); return -1; }
+        struct ETCP_SOCKET *socket = probe->message.step == NAT_STEP_ALT ? auxiliary_socket(detector, source) : source;
+        if (!socket) return -1;
+        const uint8_t *peer_key = probe->owner_key;
+        if (probe->message.step == NAT_STEP_PREPARE || probe->alternate_fallback) {
+            if (route_ping_addr_decode(probe->message.observer_address, &target)) return -1;
+            peer_key = probe->observer_key;
+        }
+        probe->io_socket = socket->sock_id; probe->io_generation = socket->generation;
+        result = route_ping_observe(socket, peer_key, &target, probe->group_id, read64(probe->message.round),
+                                    timeout, helper_ping_cb, probe);
     } else {
         struct ETCP_SOCKET *socket = find_socket(detector, probe->source_socket, probe->source_generation);
         if (!socket || socket->is_tcp || socket->socks_udp) return -1;
         uint8_t tag[NAT_QUERY_TAG_SIZE];
-        memcpy(tag, "NAT2UDP!", 8); tag[8] = probe->message.owner_socket;
+        ingress_tag(&probe->message, probe->group_id, tag);
-        memcpy(tag + 9, probe->message.generation, 8); memcpy(tag + 17, probe->message.round, 8);
+        result = etcp_send_ping_to_socket(detector->instance, socket, probe->owner_key, &target, timeout,
-        tag[25] = probe->message.step; memcpy(tag + 26, probe->message.token, 8);
-        result = etcp_send_ping_to_socket(detector->instance, socket, probe->message.owner_key, &target, timeout,
                                           helper_ping_cb, probe, tag, sizeof(tag), ETCP_PING_FLAG_ADDRESS_CHECK);
     }
     if (result) { probe->message.errors++; return -1; }

src/transport_layer/auto_socket.c

@@ -1443,6 +1443,7 @@ static int auto_socket_reconcile_config(struct AUTO_SOCKET* as) {
     struct ETCP_SOCKET* sock = inst->etcp_sockets;
     while (sock) {
         struct ETCP_SOCKET* next = sock->next;
+        if (sock->probe_only) { sock = next; continue; }
         const struct CFG_SERVER* srv = NULL;
         for (const struct CFG_SERVER* s = inst->config->servers; s; s = s->next)
             if (strcmp(s->name, sock->name) == 0) { srv = s; break; }

src/transport_layer/etcp_connections.c

@@ -1822,10 +1822,10 @@ static int handle_ping(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt, const
         memcpy(p, &nid, 8); p += 8;
         uint64_t n = htobe64(nonce);
         memcpy(p, &n, 8); p += 8;
-        uint16_t response_size = ulen + ((flags & ETCP_PING_FLAG_OBSERVE) ? ROUTE_ADDR_WIRE_SIZE : 0);
+        uint16_t response_size = ulen + ((flags & (ETCP_PING_FLAG_OBSERVE | ETCP_PING_FLAG_ADDRESS_CHECK)) ? ROUTE_ADDR_WIRE_SIZE : 0);
         uint16_t resp_ulen_be = htobe16(response_size);
         memcpy(p, &resp_ulen_be, 2); p += 2;
-        if (flags & ETCP_PING_FLAG_OBSERVE) {
+        if (flags & (ETCP_PING_FLAG_OBSERVE | ETCP_PING_FLAG_ADDRESS_CHECK)) {
             if (route_ping_addr_encode(p, addr)) { u_free(resp); return 7; }
             p += ROUTE_ADDR_WIRE_SIZE;
         }