jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Complete grouped NAT control and cancel tasks with group lifetime

Browse Source
This commit is contained in:
evgeny committed 2026-10-12 18:08:47 +02:00
1 parent a0a4493ef2
commit 8990574c1d
3 files changed
+245 -191
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/routing_layer/nat_detection.c
+236 -188
View File
@@ -55,7 +55,7 @@ struct nat_socket {
struct sockaddr_storage baseline, helper_seen, alternate_seen;
struct sockaddr_storage measured_address;
struct address_profile measured_profile;
uint64_t select_due;
uint64_t select_due, available_fingerprint;
uint8_t fallback_allowed;
uint8_t observer_key[32], helper_key[32];
uint64_t group_id, round, observer, helper, token, deadline, due, completed_tb, preferred_observer;
@@ -821,84 +821,10 @@ static int helper_send(struct nat_probe *probe) {
probe->message.sent++; probe->active = 1; return 0;
}
static void handle_find(struct NAT_DETECTION *detector, struct ETCP_CONN *from, struct nat_wire *message) {
struct nat_binding *binding = binding_find(detector, from->peer_node_id, read64(message->generation), read64(message->round));
struct sockaddr_storage target;
if (!binding || read64(message->owner) != from->peer_node_id ||
route_ping_addr_decode(message->target, &target) || !route_ping_addr_equal(&binding->target, &target, 1) ||
memcmp(binding->owner_key, message->owner_key, 32)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper selection rejected: no matching own observation peer=%016llx",
(unsigned long long)from->peer_node_id); return;
}
message->command = NAT_OFFER; message->reason = NAT_REASON_NO_HELPER; memset(message->helper, 0, 8);
uint64_t selected = UINT64_MAX;
struct ETCP_LINK *selected_link = NULL;
for (struct ll_entry *entry = detector->instance->connections->head; entry; entry = entry->next) {
struct ETCP_CONN *candidate = ((struct conn_queue_entry *)entry->data)->conn;
if (!candidate || candidate == from || !live_connection(detector, candidate->peer_node_id) ||
candidate->peer_node_id >= selected) continue;
int excluded = 0;
for (unsigned i = 0; i < message->excluded_count; i++)
if (read64(message->excluded[i]) == candidate->peer_node_id) excluded = 1;
if (excluded) continue;
for (struct ETCP_LINK *link = candidate->links; link; link = link->next) {
if (link->is_tcp || !link->link_status || !link->initialized ||
link->remote_addr.ss_family != target.ss_family || !public_address(detector, &link->remote_addr) ||
(link->remote_type != CFG_SERVER_TYPE_PUBLIC && !detector->allow_local)) continue;
selected = candidate->peer_node_id; selected_link = link; break;
}
}
if (selected_link) {
write64(message->helper, selected); message->helper_socket = selected_link->remote_socket_id;
memcpy(message->helper_key, selected_link->etcp->crypto_ctx.peer_public_key, 32);
route_ping_addr_encode(message->primary, &selected_link->remote_addr);
message->reason = NAT_REASON_NONE;
}
if (send_control(detector, from->peer_node_id, message))
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper offer delivery failed");
}
static void handle_offer(struct nat_socket *state, struct nat_wire *message) {
uint64_t helper = read64(message->helper);
if (!helper) {
if (state->fallback_helper) {
state->helper = state->fallback_helper; state->helper_socket = state->fallback_socket;
memcpy(state->helper_key, state->fallback_key, 32); state->helper_address = state->fallback_address;
state->filter_clean = 0;
if (owner_run(state, NAT_STEP_PRE)) owner_finish(state, NAT_REASON_CONTROL_LOST);
} else owner_finish(state, NAT_REASON_NO_HELPER);
return;
}
struct sockaddr_storage address;
if (route_ping_addr_decode(message->primary, &address) || sc_derive_node_id_from_pubkey(message->helper_key) != helper) {
owner_finish(state, NAT_REASON_INVALID_OBSERVATION); return;
}
int same_ip = route_ping_addr_equal(&state->observer_address, &address, 0);
int old_contact = contacted(state, &address);
if (same_ip || old_contact) {
if (!same_ip && !state->fallback_helper) {
state->fallback_helper = helper; state->fallback_socket = message->helper_socket;
state->fallback_address = address; memcpy(state->fallback_key, message->helper_key, 32);
}
if (state->excluded_count < NAT_MAX_HELPERS) {
state->excluded[state->excluded_count++] = helper; state->helper_reselections++;
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT request another helper socket=%s round=%016llx rejected=%016llx reason=%s ip=%s",
state->socket->name, (unsigned long long)state->round, (unsigned long long)helper,
same_ip ? "same_external_ip" : "previous_udp_contact", sockaddr_storage_to_str(&address).str);
if (owner_find(state)) owner_finish(state, NAT_REASON_CONTROL_LOST);
} else owner_finish(state, NAT_REASON_NO_HELPER);
return;
}
state->helper = helper; state->helper_socket = message->helper_socket;
state->helper_address = address; memcpy(state->helper_key, message->helper_key, 32);
state->filter_clean = 1;
if (owner_run(state, NAT_STEP_PRE)) owner_finish(state, NAT_REASON_CONTROL_LOST);
}
static int tcp_target_advertised(struct NAT_DETECTION *detector, const struct nat_wire *message,
static int tcp_target_advertised(struct NAT_DETECTION *detector, uint64_t group_id, const struct nat_wire *message,
const struct sockaddr_storage *target) {
const struct TOPO_NODE *owner = topo_node_registry_find(detector->instance->topo_groups, read64(message->owner));
if (!owner || memcmp(owner->public_key, message->owner_key, 32)) return 0;
const struct TOPO_NODE *owner = group_node(detector, group_id, read64(message->owner));
if (!owner) return 0;
uint8_t address[ROUTE_ADDR_WIRE_SIZE];
if (route_ping_addr_encode(address, target)) return 0;
if (target->ss_family == AF_INET) {
@@ -913,57 +839,88 @@ static int tcp_target_advertised(struct NAT_DETECTION *detector, const struct na
return 0;
}
static void handle_run(struct NAT_DETECTION *detector, struct ETCP_CONN *from, struct nat_wire *message) {
uint64_t owner = read64(message->owner), helper = read64(message->helper);
/* Broker проверяет реальное наблюдение owner, затем делегирует выбранному владельцем помощнику через ту же группу. */
static void handle_run(struct NAT_DETECTION *detector, uint64_t group_id, uint64_t sender, struct nat_wire *message) {
uint64_t owner = read64(message->owner), observer = read64(message->observer), helper = read64(message->helper);
const struct TOPO_NODE *owner_node = group_node(detector, group_id, owner);
const struct TOPO_NODE *observer_node = group_node(detector, group_id, observer);
struct sockaddr_storage target;
if (route_ping_addr_decode(message->target, &target) || sc_derive_node_id_from_pubkey(message->owner_key) != owner) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT run has invalid owner/target"); return;
uint8_t reason = NAT_REASON_INVALID_OBSERVATION;
if (!owner_node || !observer_node || owner == observer || owner == helper ||
route_ping_addr_decode(message->target, &target) || !public_address(detector, &target)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT run has invalid participants/target group=%016llx owner=%016llx observer=%016llx helper=%016llx",
(unsigned long long)group_id, (unsigned long long)owner, (unsigned long long)observer,
(unsigned long long)helper); goto reject;
}
if (owner == from->peer_node_id && helper != detector->instance->node_id) {
struct nat_binding *binding = binding_find(detector, owner, read64(message->generation), read64(message->round));
if (!binding || !route_ping_addr_equal(&binding->target, &target, 1) ||
memcmp(binding->owner_key, message->owner_key, 32) || !live_connection(detector, helper)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT run forwarding rejected: stale observation/helper"); return;
if (sender == owner && observer == detector->instance->node_id) {
if (message->step != NAT_STEP_TCP) {
struct nat_binding *binding = binding_find(detector, group_id, owner, read64(message->generation), read64(message->round));
if (!binding || binding->owner_socket != message->owner_socket || !route_ping_addr_equal(&binding->target, &target, 1) ||
memcmp(binding->owner_key, owner_node->public_key, 32) || helper == observer) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT delegation rejected: stale own observation group=%016llx owner=%016llx",
(unsigned long long)group_id, (unsigned long long)owner); goto reject;
}
binding->helper = helper; binding->token = read64(message->token); binding->expires = get_time_tb() + NAT_ROUND_TB;
}
if (send_control(detector, helper, message)) DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT run forwarding failed");
return;
if (helper != detector->instance->node_id) {
if (!control_route(detector, group_id, helper) || send_control(detector, group_id, helper, message)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT delegation route failed group=%016llx helper=%016llx",
(unsigned long long)group_id, (unsigned long long)helper);
reason = NAT_REASON_CONTROL_LOST; goto reject;
}
return;
}
} else if (sender != observer || helper != detector->instance->node_id) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT run role mismatch group=%016llx sender=%016llx observer=%016llx helper=%016llx",
(unsigned long long)group_id, (unsigned long long)sender, (unsigned long long)observer,
(unsigned long long)helper); goto reject;
}
if (helper != detector->instance->node_id || (message->step == NAT_STEP_TCP && !tcp_target_advertised(detector, message, &target))) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper run rejected helper=%016llx step=%u", (unsigned long long)helper, message->step);
return;
if (message->step == NAT_STEP_TCP && (helper != observer || !tcp_target_advertised(detector, group_id, message, &target))) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT TCP target is not the signed owner listener group=%016llx", (unsigned long long)group_id);
goto reject;
}
unsigned count = 0;
for (struct nat_probe *probe = detector->probes; probe; probe = probe->next) {
count++;
if (read64(probe->message.owner) == owner && read64(probe->message.round) == read64(message->round) &&
if (probe->group_id == group_id && read64(probe->message.owner) == owner &&
read64(probe->message.generation) == read64(message->generation) && read64(probe->message.round) == read64(message->round) &&
read64(probe->message.token) == read64(message->token)) {
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT duplicate helper run ignored"); return;
}
}
if (count >= NAT_MAX_PROBES) { DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper probe limit reached"); return; }
if (count >= NAT_MAX_PROBES) { DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper probe limit reached"); reason = NAT_REASON_SEND_ERROR; goto reject; }
struct ETCP_SOCKET *source = NULL;
if (message->step != NAT_STEP_TCP) {
struct sockaddr_storage primary;
struct sockaddr_storage observer_address;
source = find_socket(detector, message->helper_socket, 0);
if (!source || source->is_tcp || source->probe_only || source->socks_udp ||
route_ping_addr_decode(message->primary, &primary) || !route_ping_addr_equal(&primary, &source->local_addr, 1)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper source is not the requested native UDP endpoint"); return;
source->type == CFG_SERVER_TYPE_PRIVATE || source->type == CFG_SERVER_TYPE_LOCAL ||
source->local_addr.ss_family != target.ss_family ||
route_ping_addr_decode(message->observer_address, &observer_address) ||
observer_address.ss_family != target.ss_family || !public_address(detector, &observer_address)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper source is not a native socket group=%016llx socket=%u",
(unsigned long long)group_id, message->helper_socket);
reason = source && source->socks_udp ? NAT_REASON_PROXY : NAT_REASON_NO_HELPER; goto reject;
}
struct ETCP_SOCKET *alternate = auxiliary_socket(detector, source);
if (!alternate || route_ping_addr_encode(message->secondary, &alternate->local_addr)) return;
}
struct nat_probe *probe = u_calloc(1, sizeof(*probe));
if (!probe) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT helper probe allocation failed"); return; }
probe->detector = detector; probe->message = *message; probe->reply = from->peer_node_id;
if (!probe) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT helper probe allocation failed"); reason = NAT_REASON_SEND_ERROR; goto reject; }
probe->detector = detector; probe->message = *message; probe->group_id = group_id; probe->reply = sender;
memcpy(probe->owner_key, owner_node->public_key, 32); memcpy(probe->observer_key, observer_node->public_key, 32);
if (source) { probe->source_socket = source->sock_id; probe->source_generation = source->generation; }
probe->expires = get_time_tb() + NAT_ROUND_TB;
probe->next = detector->probes; detector->probes = probe;
if (helper_send(probe)) helper_finish(probe, NAT_REASON_SEND_ERROR);
return;
reject:
message->command = NAT_RESULT; message->reason = reason;
if (send_control(detector, group_id, sender, message)) DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT rejection delivery failed");
}
static void handle_result(struct nat_socket *state, const struct nat_wire *message) {
uint8_t step = state->phase == NAT_PRE ? NAT_STEP_PRE : state->phase == NAT_POST ? NAT_STEP_POST :
state->phase == NAT_TCP ? NAT_STEP_TCP : 0;
state->phase == NAT_TCP ? NAT_STEP_TCP : state->phase == NAT_PREPARE_C ? NAT_STEP_PREPARE :
state->phase == NAT_PREPARE_ALT ? NAT_STEP_ALT : 0;
if (!step || message->step != step || read64(message->token) != state->token || read64(message->helper) != state->helper) {
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT stale helper result ignored socket=%s phase=%u", state->socket->name, state->phase); return;
}
@@ -975,12 +932,33 @@ static void handle_result(struct nat_socket *state, const struct nat_wire *messa
else owner_finish(state, NAT_REASON_TIMEOUT);
return;
}
if (step == NAT_STEP_PRE) {
if (step == NAT_STEP_PREPARE) {
struct sockaddr_storage primary;
if (route_ping_addr_decode(message->primary, &primary) || primary.ss_family != state->local.ss_family ||
!public_address(state->detector, &primary)) { owner_finish(state, NAT_REASON_INVALID_OBSERVATION); return; }
int same_ip = route_ping_addr_equal(&state->observer_address, &primary, 0), old_contact = contacted(state, &primary);
if (same_ip || (old_contact && !state->using_fallback)) {
if (!same_ip && !state->fallback.group) state->fallback = (struct nat_pair){state->group_id, state->observer, state->helper};
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT try another pair socket=%s group=%016llx helper=%016llx endpoint=%s reason=%s",
state->socket->name, (unsigned long long)state->group_id, (unsigned long long)state->helper,
sockaddr_storage_to_str(&primary).str, same_ip ? "same_external_ip" : "previous_udp_contact");
owner_finish(state, same_ip ? NAT_REASON_NO_HELPER : NAT_REASON_CONTACT_HISTORY); return;
}
state->helper_address = primary; state->filter_clean = !old_contact;
if (owner_run(state, NAT_STEP_PRE)) owner_finish(state, NAT_REASON_CONTROL_LOST);
return;
}
if (step == NAT_STEP_ALT) {
if (route_ping_addr_decode(message->secondary, &state->alternate_address) ||
!route_ping_addr_equal(&state->alternate_address, &state->helper_address, 0) ||
route_ping_addr_equal(&state->alternate_address, &state->helper_address, 1)) {
owner_finish(state, NAT_REASON_INVALID_OBSERVATION); return;
}
state->phase = NAT_OBSERVE_ALT;
} else if (step == NAT_STEP_PRE) {
if (!state->pre_received && route_ping_addr_decode(message->primary, &state->helper_address)) {
owner_finish(state, NAT_REASON_INVALID_OBSERVATION); return;
}
state->pre_healthy = message->sent == NAT_PROBE_COUNT;
state->phase = NAT_OBSERVE_C;
} else state->phase = NAT_REPEAT_B;
@@ -988,69 +966,74 @@ static void handle_result(struct nat_socket *state, const struct nat_wire *messa
if (owner_observe(state)) owner_finish(state, NAT_REASON_SEND_ERROR);
}
/* Потеря router-сессии отменяет только роли этого узла в данной группе, не общий транспорт. */
static void cancel_route(struct NAT_DETECTION *detector, uint64_t group_id, uint64_t peer) {
for (struct nat_socket *state = detector->sockets; state; state = state->next)
if (state->phase && state->group_id == group_id && (state->observer == peer || state->helper == peer))
owner_finish(state, NAT_REASON_CONTROL_LOST);
struct nat_probe *probe = detector->probes;
while (probe) {
struct nat_probe *next = probe->next;
if (probe->group_id == group_id && (probe->reply == peer || read64(probe->message.owner) == peer)) helper_remove(probe);
probe = next;
}
for (struct nat_binding *binding = detector->bindings; binding; binding = binding->next)
if (binding->group_id == group_id && (binding->owner == peer || binding->helper == peer)) binding->expires = 0;
}
/* SOURCE и group берём из проверенного router-заголовка, а не из ETCP next hop. */
static void receive_control(struct ETCP_CONN *from, struct ll_entry *entry) {
struct NAT_DETECTION *detector = from && from->instance ? from->instance->nat_det : NULL;
if (!detector || !entry || entry->len != sizeof(struct nat_wire)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT control invalid size=%zu", entry ? entry->len : 0);
if (!detector || !entry || !entry->dgram || entry->len < ROUTER_SVC_HDR_SIZE) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT routed control invalid size=%zu", entry ? entry->len : 0);
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return;
}
struct nat_wire message; memcpy(&message, entry->dgram, sizeof(message));
queue_dgram_free(entry); queue_entry_free(entry);
if (message.cmd != ETCP_ID_NAT_DETECTION || message.version != NAT_WIRE_VERSION ||
message.excluded_count > NAT_MAX_HELPERS || !read64(message.generation) || !read64(message.round) ||
message.command < NAT_FIND || message.command > NAT_RESULT || message.reason > NAT_REASON_INVALID_OBSERVATION ||
message.step > NAT_STEP_TCP) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT control malformed command=%u version=%u", message.command, message.version); return;
uint64_t group_id, sender;
memcpy(&group_id, entry->dgram + ROUTER_SVC_GROUP_OFF, 8); memcpy(&sender, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
if (entry->len == ROUTER_SVC_HDR_SIZE) {
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT routed control closed group=%016llx peer=%016llx",
(unsigned long long)group_id, (unsigned long long)sender);
queue_dgram_free(entry); queue_entry_free(entry); cancel_route(detector, group_id, sender); return;
}
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT control RX command=%u step=%u from=%016llx owner=%016llx round=%016llx",
message.command, message.step, (unsigned long long)from->peer_node_id,
(unsigned long long)read64(message.owner), (unsigned long long)read64(message.round));
if (message.command == NAT_FIND) { handle_find(detector, from, &message); return; }
if (message.command == NAT_RUN) { handle_run(detector, from, &message); return; }
if (entry->len != ROUTER_SVC_HDR_SIZE + sizeof(struct nat_wire) ||
entry->dgram[ROUTER_SVC_FLAGS_OFF] != (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED) ||
!group_node(detector, group_id, sender)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT routed control rejected group=%016llx sender=%016llx size=%zu flags=%u",
(unsigned long long)group_id, (unsigned long long)sender, entry->len, entry->dgram[ROUTER_SVC_FLAGS_OFF]);
queue_dgram_free(entry); queue_entry_free(entry); return;
}
struct nat_wire message; memcpy(&message, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, sizeof(message));
queue_dgram_free(entry); queue_entry_free(entry);
if (message.cmd != ETCP_RT_ID_NAT_DETECTION || message.version != NAT_WIRE_VERSION ||
!read64(message.owner) || !read64(message.observer) || !read64(message.helper) ||
!read64(message.generation) || !read64(message.round) || !read64(message.token) ||
message.command < NAT_RUN || message.command > NAT_RESULT || message.reason > NAT_REASON_DEPENDENT_MAPPING ||
message.step < NAT_STEP_PREPARE || message.step > NAT_STEP_TCP) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT control malformed command=%u version=%u step=%u", message.command, message.version, message.step);
return;
}
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT control RX group=%016llx command=%u step=%u sender=%016llx via=%016llx owner=%016llx round=%016llx",
(unsigned long long)group_id, message.command, message.step, (unsigned long long)sender,
(unsigned long long)from->peer_node_id, (unsigned long long)read64(message.owner), (unsigned long long)read64(message.round));
if (message.command == NAT_RUN) { handle_run(detector, group_id, sender, &message); return; }
uint64_t owner = read64(message.owner);
if (message.command == NAT_RESULT && owner != detector->instance->node_id) {
struct nat_binding *binding = binding_find(detector, owner, read64(message.generation), read64(message.round));
if (!binding || read64(message.helper) != from->peer_node_id) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper result has no matching delegation"); return;
if (owner != detector->instance->node_id) {
struct nat_binding *binding = binding_find(detector, group_id, owner, read64(message.generation), read64(message.round));
if (!binding || binding->helper != sender || read64(message.helper) != sender || binding->token != read64(message.token) ||
read64(message.observer) != detector->instance->node_id) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT result has no matching delegation group=%016llx owner=%016llx sender=%016llx",
(unsigned long long)group_id, (unsigned long long)owner, (unsigned long long)sender); return;
}
if (send_control(detector, owner, &message)) DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper result forwarding failed");
if (send_control(detector, group_id, owner, &message)) DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT result forwarding failed");
return;
}
struct ETCP_SOCKET *socket = find_socket(detector, message.owner_socket, read64(message.generation));
struct nat_socket *state = socket_state(detector, socket, 0);
if (!state || owner != detector->instance->node_id || state->round != read64(message.round) ||
state->observer != from->peer_node_id || state->phase == NAT_IDLE) {
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT stale/foreign owner response ignored"); return;
if (!state || state->group_id != group_id || state->round != read64(message.round) ||
state->observer != sender || read64(message.observer) != sender || state->phase == NAT_IDLE) {
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT stale/foreign owner response ignored group=%016llx", (unsigned long long)group_id); return;
}
if (message.command == NAT_OFFER) {
if (state->phase == NAT_FIND_C) handle_offer(state, &message);
else DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT stale helper offer ignored");
} else if (message.command == NAT_RESULT) handle_result(state, &message);
}
static int select_observer(struct nat_socket *state) {
struct NAT_DETECTION *detector = state->detector;
uint64_t selected = UINT64_MAX;
struct ETCP_LINK *selected_link = NULL;
for (struct ll_entry *entry = detector->instance->connections->head; entry; entry = entry->next) {
struct ETCP_CONN *connection = ((struct conn_queue_entry *)entry->data)->conn;
if (!connection || !live_connection(detector, connection->peer_node_id) || connection->peer_node_id >= selected) continue;
if (state->preferred_observer && state->preferred_observer != connection->peer_node_id) continue;
for (struct ETCP_LINK *link = connection->links; link; link = link->next) {
if (!link->link_status || !link->initialized) continue;
if (state->socket->is_tcp) {
if (!public_address(detector, &link->remote_addr)) continue;
} else {
if (link->is_tcp || link->remote_addr.ss_family != state->local.ss_family ||
!public_address(detector, &link->remote_addr) ||
(link->remote_type != CFG_SERVER_TYPE_PUBLIC && !detector->allow_local)) continue;
}
selected = connection->peer_node_id; selected_link = link; break;
}
}
if (!selected_link) return -1;
state->observer = selected; state->observer_address = selected_link->remote_addr;
memcpy(state->observer_key, selected_link->etcp->crypto_ctx.peer_public_key, 32); return 0;
handle_result(state, &message);
}
static void owner_start(struct nat_socket *state) {
@@ -1067,27 +1050,22 @@ static void owner_start(struct nat_socket *state) {
DEBUG_INFO(DEBUG_CATEGORY_NAT, "TCP inbound profile remains UNKNOWN: no public listener candidate socket=%s local=%s",
state->socket->name, sockaddr_storage_to_str(&state->local).str); return;
}
if (select_observer(state)) {
if (state->reason != NAT_REASON_NO_OBSERVER)
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT recheck has no live observer socket=%s retained=%s/%s source=%s", state->socket->name,
address_mapping_name(state->socket->address_profile.mapping),
address_inbound_name(state->socket->address_profile.inbound),
address_profile_source_name(state->socket->address_profile.source));
state->reason = NAT_REASON_NO_OBSERVER; state->fallback_allowed = 1;
select_profile(state);
state->due = get_time_tb() + (state->socket->address_profile.source ? NAT_REFRESH_TB : NAT_RETRY_TB); return;
if (!state->cycling) {
state->cycling = 1; state->cursor = (struct nat_pair){0}; state->fallback = (struct nat_pair){0}; state->using_fallback = 0;
}
if (random_id(&state->round)) { state->reason = NAT_REASON_SEND_ERROR; state->due = get_time_tb() + NAT_RETRY_TB; return; }
if (select_pair(state)) { owner_finish(state, state->reason); return; }
if (random_id(&state->round)) { owner_finish(state, NAT_REASON_SEND_ERROR); return; }
history_load(state);
state->rounds++; state->excluded_count = 0; state->fallback_helper = 0; state->helper = 0;
state->rounds++;
memset(&state->helper_seen, 0, sizeof(state->helper_seen)); memset(&state->alternate_seen, 0, sizeof(state->alternate_seen));
state->retries = 0; state->pre_received = 0; state->post_received = 0; state->filter_clean = 0; state->pre_healthy = 0;
state->deadline = get_time_tb() + NAT_ROUND_TB; state->reason = NAT_REASON_NONE;
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT round start socket=%s id=%u generation=%016llx round=%016llx observer=%016llx protocol=%s",
state->socket->name, state->socket->sock_id, (unsigned long long)state->socket->generation,
(unsigned long long)state->round, (unsigned long long)state->observer, state->socket->is_tcp ? "TCP" : "UDP");
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT round start socket=%s group=%016llx id=%u generation=%016llx round=%016llx observer=%016llx helper=%016llx protocol=%s",
state->socket->name, (unsigned long long)state->group_id, state->socket->sock_id,
(unsigned long long)state->socket->generation, (unsigned long long)state->round,
(unsigned long long)state->observer, (unsigned long long)state->helper, state->socket->is_tcp ? "TCP" : "UDP");
if (state->socket->is_tcp) {
state->baseline = state->local;
state->helper = state->observer;
if (owner_run(state, NAT_STEP_TCP)) owner_finish(state, NAT_REASON_CONTROL_LOST);
} else {
state->phase = NAT_OBSERVE_B;
@@ -1129,28 +1107,92 @@ static void detector_tick(void *arg) {
if (!detector->timer) DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT detector timer scheduling failed");
}
void nat_detection_cancel_for_conn(struct NAT_DETECTION *detector, struct ETCP_CONN *connection) {
if (!detector || !connection || detector->stopping) return;
for (struct nat_socket *state = detector->sockets; state; state = state->next)
if (state->phase && (state->observer == connection->peer_node_id || state->helper == connection->peer_node_id))
owner_finish(state, NAT_REASON_CONTROL_LOST);
/* Отпечаток доступных endpoint не содержит timestamp/имя/профиль: обычный повтор NODEINFO не запускает пробы. */
static uint64_t candidate_fingerprint(struct nat_socket *state) {
struct NAT_DETECTION *detector = state->detector;
uint64_t total = 0;
for (struct ll_entry *e = detector->instance->topo_groups->group_list->head; e; e = e->next) {
struct TOPO_GROUP *group = (struct TOPO_GROUP *)e;
if (group->stopping || !group->local_node) continue;
for (struct ll_entry *n = group->nodes->head; n; n = n->next) {
uint64_t id = ((struct TOPO_GROUP_NODE *)n)->node_id;
if (id == detector->instance->node_id || !control_route(detector, group->group_id, id)) continue;
const struct TOPO_NODE *node = group_node(detector, group->group_id, id);
struct sockaddr_storage endpoint = {0}; uint8_t socket_id = 0;
uint8_t bytes[16 + ROUTE_ADDR_WIRE_SIZE + 1] = {0};
write64(bytes, group->group_id); write64(bytes + 8, id);
if (!state->socket->is_tcp) {
if (!node_udp_endpoint(detector, node, state->local.ss_family, 1, &endpoint, &socket_id)) continue;
if (route_ping_addr_encode(bytes + 16, &endpoint)) continue;
bytes[16 + ROUTE_ADDR_WIRE_SIZE] = socket_id;
}
uint64_t hash = UINT64_C(14695981039346656037);
for (unsigned i = 0; i < sizeof(bytes); i++) { hash ^= bytes[i]; hash *= UINT64_C(1099511628211); }
total += hash;
}
}
return total;
}
/* Изменения нескольких групп объединяются; известный полный ETCP-профиль не перепроверяется из-за нового соседа. */
static void topology_wake(void *arg) {
struct NAT_DETECTION *detector = arg; detector->topology_wake = NULL;
for (struct nat_socket *state = detector->sockets; state; state = state->next) {
uint64_t fingerprint = candidate_fingerprint(state);
int changed = fingerprint != state->available_fingerprint; state->available_fingerprint = fingerprint;
if (state->phase) {
if (!control_route(detector, state->group_id, state->observer) || !control_route(detector, state->group_id, state->helper))
owner_finish(state, NAT_REASON_CONTROL_LOST);
continue;
}
if (!changed || !fingerprint || !detector->automatic || state->cycling) continue;
if (state->measured_profile.source == ADDRESS_SOURCE_ETCP && state->measured_profile.mapping && state->measured_profile.inbound)
continue;
state->cursor = (struct nat_pair){0}; state->fallback = (struct nat_pair){0}; state->using_fallback = 0;
state->due = get_time_tb() + 2500;
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT available candidates changed socket=%s; early measurement scheduled", state->socket->name);
}
}
void nat_detection_topology_changed(struct NAT_DETECTION *detector) {
if (!detector || detector->stopping || detector->topology_wake) return;
detector->topology_wake = uasync_call_soon(detector->instance->ua, detector, topology_wake);
if (!detector->topology_wake) DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT topology wake scheduling failed");
}
void nat_detection_cancel_group(struct NAT_DETECTION *detector, uint64_t group_id) {
if (!detector || detector->stopping) return;
for (struct nat_socket *state = detector->sockets; state; state = state->next) {
if (state->fallback.group == group_id) state->fallback = (struct nat_pair){0};
if (state->group_id != group_id || (!state->phase && !state->cycling)) continue;
owner_finish(state, NAT_REASON_CANCELLED); state->forced = 0; state->group_id = 0; state->due = get_time_tb() + 2500;
}
struct nat_probe *probe = detector->probes;
while (probe) {
struct nat_probe *next = probe->next;
if (probe->reply == connection->peer_node_id) helper_remove(probe);
if (probe->group_id == group_id) helper_remove(probe);
probe = next;
}
struct nat_binding **cursor = &detector->bindings;
while (*cursor) {
struct nat_binding *binding = *cursor;
if (binding->group_id != group_id) { cursor = &binding->next; continue; }
*cursor = binding->next; u_free(binding);
}
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT group tasks cancelled group=%016llx", (unsigned long long)group_id);
nat_detection_topology_changed(detector);
}
void nat_detection_cancel_for_conn(struct NAT_DETECTION *detector, struct ETCP_CONN *connection) {
if (!detector || !connection || detector->stopping) return;
nat_detection_topology_changed(detector);
}
static void connection_status(struct ETCP_CONN *connection, int status, void *arg) {
struct NAT_DETECTION *detector = arg;
if (status == ETCP_CONN_STATUS_DOWN || status == ETCP_CONN_STATUS_DELETE || status == ETCP_CONN_STATUS_REINIT)
nat_detection_cancel_for_conn(detector, connection);
else if (status == ETCP_CONN_STATUS_UP) {
for (struct nat_socket *state = detector->sockets; state; state = state->next)
if (!state->phase && (state->reason == NAT_REASON_NO_OBSERVER || state->reason == NAT_REASON_NO_HELPER))
state->due = get_time_tb() + 2500;
}
else if (status == ETCP_CONN_STATUS_UP) nat_detection_topology_changed(detector);
}
void nat_detection_link_ready(struct NAT_DETECTION *detector, struct ETCP_LINK *link) {
@@ -1179,7 +1221,8 @@ void nat_detection_socket_removed(struct NAT_DETECTION *detector, struct ETCP_SO
struct nat_probe *probe = detector->probes;
while (probe) {
struct nat_probe *next = probe->next;
if (probe->source_socket == socket->sock_id && probe->source_generation == socket->generation) helper_remove(probe);
if ((probe->source_socket == socket->sock_id && probe->source_generation == socket->generation) ||
(probe->io_socket == socket->sock_id && probe->io_generation == socket->generation)) helper_remove(probe);
probe = next;
}
struct nat_socket **cursor = &detector->sockets;
@@ -1205,20 +1248,21 @@ static void socket_changed(struct ETCP_SOCKET *socket, int event, void *arg) {
}
struct NAT_DETECTION *nat_detection_create(struct UTUN_INSTANCE *instance) {
if (!instance || instance->api_bindings.callbacks[ETCP_ID_NAT_DETECTION]) {
if (!instance || !instance->topo_groups || !instance->router_conns ||
instance->router_bindings.callbacks[ETCP_RT_ID_NAT_DETECTION]) {
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT detector invalid instance/binding collision"); return NULL;
}
struct NAT_DETECTION *detector = u_calloc(1, sizeof(*detector));
if (!detector) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT detector allocation failed"); return NULL; }
detector->instance = instance; detector->automatic = 1;
if (etcp_bind(instance, ETCP_ID_NAT_DETECTION, receive_control)) { u_free(detector); return NULL; }
if (etcp_router_bind(instance, ETCP_RT_ID_NAT_DETECTION, receive_control)) { u_free(detector); return NULL; }
etcp_add_conn_status_cbk(instance, connection_status, detector);
etcp_add_socket_cbk(instance, socket_changed, detector, ETCP_SOCKET_EVENT_ADDR_CHANGED);
detector->timer = uasync_set_timeout(instance->ua, NAT_TICK_TB, detector, detector_tick, "nat_detection");
if (!detector->timer) {
etcp_remove_conn_status_cbk(instance, connection_status, detector);
etcp_remove_socket_cbk(instance, socket_changed, detector);
etcp_unbind(instance, ETCP_ID_NAT_DETECTION); u_free(detector);
etcp_router_unbind(instance, ETCP_RT_ID_NAT_DETECTION); u_free(detector);
DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT detector initial timer failed"); return NULL;
}
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT endpoint detector started: UDP mapping/inbound measured separately, TCP listener probes");
@@ -1229,9 +1273,10 @@ void nat_detection_destroy(struct NAT_DETECTION *detector) {
if (!detector) return;
detector->stopping = 1;
if (detector->timer) uasync_cancel_timeout(detector->instance->ua, detector->timer);
if (detector->topology_wake) uasync_call_soon_cancel(detector->instance->ua, detector->topology_wake);
etcp_remove_conn_status_cbk(detector->instance, connection_status, detector);
etcp_remove_socket_cbk(detector->instance, socket_changed, detector);
etcp_unbind(detector->instance, ETCP_ID_NAT_DETECTION);
etcp_router_unbind(detector->instance, ETCP_RT_ID_NAT_DETECTION);
while (detector->probes) helper_remove(detector->probes);
while (detector->sockets) {
struct nat_socket *state = detector->sockets; detector->sockets = state->next;
@@ -1252,6 +1297,7 @@ void nat_detection_request_check_all(struct NAT_DETECTION *detector) {
struct nat_socket *state = socket_state(detector, socket, 1);
if (!state) continue;
if (state->phase) owner_finish(state, NAT_REASON_CANCELLED);
state->cycling = 0; state->cursor = (struct nat_pair){0}; state->fallback = (struct nat_pair){0}; state->using_fallback = 0;
state->due = get_time_tb(); state->forced = 1; state->preferred_observer = 0;
}
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT endpoint recheck requested");
@@ -1259,10 +1305,11 @@ void nat_detection_request_check_all(struct NAT_DETECTION *detector) {
int nat_detection_request_check_socket(struct NAT_DETECTION *detector, struct ETCP_SOCKET *socket, uint64_t observer) {
struct nat_socket *state = socket_state(detector, socket, 1);
if (!state || !observer || !live_connection(detector, observer)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT endpoint recheck has no live observer"); return -1;
if (!state || !observer) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT endpoint recheck has invalid endpoint/observer"); return -1;
}
if (state->phase) owner_finish(state, NAT_REASON_CANCELLED);
state->cycling = 0; state->cursor = (struct nat_pair){0}; state->fallback = (struct nat_pair){0}; state->using_fallback = 0;
state->preferred_observer = observer; state->due = get_time_tb(); state->forced = 1;
DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT endpoint recheck socket=%s observer=%016llx", socket->name, (unsigned long long)observer);
return 0;
@@ -1281,7 +1328,8 @@ void nat_detection_set_automatic(struct NAT_DETECTION *detector, int automatic)
int nat_detection_get_status(struct NAT_DETECTION *detector, struct ETCP_SOCKET *socket, struct nat_check_status *status) {
struct nat_socket *state = socket_state(detector, socket, 0);
if (!state || !status) return -1;
*status = (struct nat_check_status){.generation = socket->generation, .round = state->round, .observer = state->observer,
*status = (struct nat_check_status){.generation = socket->generation, .round = state->round, .group_id = state->group_id,
.observer = state->observer,
.helper = state->helper, .rounds = state->rounds, .helper_reselections = state->helper_reselections,
.phase = state->phase, .reason = state->reason, .history_complete = state->history_complete,
.measured_profile = state->measured_profile, .effective_profile = socket->address_profile, .fallback_allowed = state->fallback_allowed};
src/routing_layer/topo_group.c
+3
View File
@@ -101,6 +101,7 @@ int topo_group_peer_ready(const struct TOPO_GROUP* group, uint64_t peer_id) {
static void topo_group_notify_ready(struct TOPO_GROUP_CONN_ITEM* peer, int ready) {
if (peer->ready_notified == ready) return;
peer->ready_notified = ready;
nat_detection_topology_changed(peer->group->instance->nat_det);
DEBUG_INFO(DEBUG_CATEGORY_BGP, "BGP peer %s: group=%016llx peer=%016llx local_epoch=%016llx remote_epoch=%016llx "
"links_up=%u transport_down=%u accepted=%u table_sent=%u table_received=%u send_failed=%u",
ready ? "READY (group exchange complete)" : "NOT READY (group session unavailable)",
@@ -625,6 +626,7 @@ static struct TOPO_GROUP* topo_group_create(struct UTUN_INSTANCE* instance, uint
static void topo_group_destroy(struct TOPO_GROUP* group) {
if (!group) return;
group->stopping = 1;
nat_detection_cancel_group(group->instance->nat_det, group->group_id);
stream_cancel_group(group->instance, group->group_id);
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 0 enter grp=%p id=0x%016llx rec_list=%p connect=%p",
group, (unsigned long long)group->group_id, group->recovery, group->connect);
@@ -1959,6 +1961,7 @@ void utun_remove_nodeinfo_cbk(struct UTUN_INSTANCE* instance, nodeinfo_cbk_fn fn
/* Вызывает все глобальные nodeinfo-колбэки для узла. */
void topo_fire_nodeinfo_cbk(struct UTUN_INSTANCE* instance, struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* node) {
if (!instance || !group || !node) return;
nat_detection_topology_changed(instance->nat_det);
struct nodeinfo_cbk_entry* c = instance->nodeinfo_cbks;
while (c) { c->fn(group, node, c->arg); c = c->next; }
}
src/utun_instance.c
+6 -3
View File
@@ -211,9 +211,6 @@ static int instance_init_common(struct UTUN_INSTANCE* instance, struct UASYNC* u
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "Socket initialization failed: result=%d", socket_result);
return -1;
}
instance->nat_det = nat_detection_create(instance);
if (!instance->nat_det) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "NAT detection creation failed"); return -1; }
if (config->global.db_path[0]) mkdir_recursive(config->global.db_path);
if (g_topo_group_enabled) {
instance->topo_groups = topo_groups_init(instance);
@@ -252,6 +249,12 @@ static int instance_init_common(struct UTUN_INSTANCE* instance, struct UASYNC* u
return -1;
}
/* NAT управляет пробами через router; binding создаётся после его инициализации. */
if (instance->topo_groups) {
instance->nat_det = nat_detection_create(instance);
if (!instance->nat_det) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT detection creation failed"); return -1; }
}
// Bind CONN_MGR handler AFTER etcp_router_init (init затирает router_bindings)
if (g_topo_group_enabled && instance->topo_groups)
etcp_router_bind(instance, ETCP_RT_ID_CONN_MGR, conn_mgr_router_recv_handler);