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

nat: align probe timing and test receiver IP rejection and endpoint failover

stream-multiview
evgeny 1 week ago
parent
38447da5e8
commit
4fab849917
2 changed files with 115 additions and 51 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 105
      src/routing_layer/nat_detection.c
  2. 61
      tests/test_nat_detection_faults.c

105
src/routing_layer/nat_detection.c
Unescape View File

@ -248,34 +248,34 @@ static int select_pair(struct nat_socket *state) {
const struct TOPO_NODE *bn = group_node(detector, group->group_id, observer);
uint8_t previous_b[ROUTE_ADDR_WIRE_SIZE] = {0};
struct sockaddr_storage ba = {0}; uint8_t bs = 0;
for (int more_b = state->socket->is_tcp || node_udp_endpoint(detector, bn, state->local.ss_family, 0, previous_b, &ba, &bs);
more_b; more_b = !state->socket->is_tcp && node_udp_endpoint(detector, bn, state->local.ss_family, 0, previous_b, &ba, &bs)) {
if (!state->socket->is_tcp && route_ping_addr_encode(previous_b, &ba)) break;
observers++;
for (struct ll_entry *c = state->socket->is_tcp ? b : group->nodes->head; c; c = state->socket->is_tcp ? NULL : c->next) {
uint64_t helper = ((struct TOPO_GROUP_NODE *)c)->node_id;
if (helper == detector->instance->node_id || (!state->socket->is_tcp && helper == observer) ||
!control_route(detector, group->group_id, helper)) continue;
const struct TOPO_NODE *cn = group_node(detector, group->group_id, helper);
struct sockaddr_storage ca = {0}; uint8_t cs = 0, previous_c[ROUTE_ADDR_WIRE_SIZE] = {0};
for (int more_c = state->socket->is_tcp || node_udp_endpoint(detector, cn, state->local.ss_family, 1, previous_c, &ca, &cs);
more_c; more_c = !state->socket->is_tcp && node_udp_endpoint(detector, cn, state->local.ss_family, 1, previous_c, &ca, &cs)) {
if (!state->socket->is_tcp && route_ping_addr_encode(previous_c, &ca)) break;
struct nat_pair candidate = {.group = group->group_id, .observer = observer, .helper = helper};
memcpy(candidate.observer_endpoint, previous_b, sizeof(previous_b));
memcpy(candidate.helper_endpoint, previous_c, sizeof(previous_c));
if (state->using_fallback) {
if (pair_after(candidate, state->fallback) || pair_after(state->fallback, candidate)) continue;
} else if (!pair_after(candidate, state->cursor)) continue;
if (!state->socket->is_tcp && state->prefer_fresh && !state->using_fallback && contacted(state, &ca)) continue;
if (best.group && !pair_after(best, candidate)) continue;
best = candidate; observer_address = ba; helper_address = ca; helper_socket = cs;
memcpy(observer_key, bn->public_key, 32); memcpy(helper_key, cn->public_key, 32);
if (state->socket->is_tcp) break;
int more_b = state->socket->is_tcp || node_udp_endpoint(detector, bn, state->local.ss_family, 0, previous_b, &ba, &bs);
for (; more_b; more_b = !state->socket->is_tcp &&
node_udp_endpoint(detector, bn, state->local.ss_family, 0, previous_b, &ba, &bs)) {
if (!state->socket->is_tcp && route_ping_addr_encode(previous_b, &ba)) break;
observers++;
for (struct ll_entry *c = state->socket->is_tcp ? b : group->nodes->head; c; c = state->socket->is_tcp ? NULL : c->next) {
uint64_t helper = ((struct TOPO_GROUP_NODE *)c)->node_id;
if (helper == detector->instance->node_id || (!state->socket->is_tcp && helper == observer) ||
!control_route(detector, group->group_id, helper)) continue;
const struct TOPO_NODE *cn = group_node(detector, group->group_id, helper);
struct sockaddr_storage ca = {0}; uint8_t cs = 0, previous_c[ROUTE_ADDR_WIRE_SIZE] = {0};
int more_c = state->socket->is_tcp || node_udp_endpoint(detector, cn, state->local.ss_family, 1, previous_c, &ca, &cs);
for (; more_c; more_c = !state->socket->is_tcp &&
node_udp_endpoint(detector, cn, state->local.ss_family, 1, previous_c, &ca, &cs)) {
if (!state->socket->is_tcp && route_ping_addr_encode(previous_c, &ca)) break;
struct nat_pair candidate = {.group = group->group_id, .observer = observer, .helper = helper};
memcpy(candidate.observer_endpoint, previous_b, sizeof(previous_b));
memcpy(candidate.helper_endpoint, previous_c, sizeof(previous_c));
if (state->using_fallback) {
if (pair_after(candidate, state->fallback) || pair_after(state->fallback, candidate)) continue;
} else if (!pair_after(candidate, state->cursor)) continue;
if (!state->socket->is_tcp && state->prefer_fresh && !state->using_fallback && contacted(state, &ca)) continue;
if (best.group && !pair_after(best, candidate)) continue;
best = candidate; observer_address = ba; helper_address = ca; helper_socket = cs;
memcpy(observer_key, bn->public_key, 32); memcpy(helper_key, cn->public_key, 32);
}
}
}
if (state->socket->is_tcp) break;
}
}
}
if (!best.group) {
@ -460,6 +460,16 @@ static int send_control(struct NAT_DETECTION *detector, uint64_t group_id, uint6
return 0;
}
/* Один бюджет ожидания для UDP-пробы, серий помощника и паузы доставки последнего пакета. */
static int probe_timeout_ms(struct nat_socket *state) {
int timeout = 1000;
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;
return timeout;
}
static int make_message(struct nat_socket *state, struct nat_wire *message, uint8_t command) {
memset(message, 0, sizeof(*message)); message->command = command;
message->owner_socket = state->socket->sock_id; message->helper_socket = state->helper_socket;
@ -469,12 +479,7 @@ static int make_message(struct nat_socket *state, struct nat_wire *message, uint
if (route_ping_addr_encode(message->target, state->socket->is_tcp ? &state->local : &state->baseline)) return -1;
if (state->helper_address.ss_family && route_ping_addr_encode(message->primary, &state->helper_address)) return -1;
if (state->observer_address.ss_family && route_ping_addr_encode(message->observer_address, &state->observer_address)) return -1;
int timeout = 1000;
struct ETCP_CONN *observer = topo_group_find_conn_for_node(
topo_groups_find(state->detector->instance->topo_groups, state->group_id), state->observer);
for (struct ETCP_LINK *link = observer ? observer->links : NULL; link; link = link->next)
if (link->rtt_last / 10 * 4 > timeout) timeout = link->rtt_last / 10 * 4;
if (timeout > 3000) timeout = 3000;
int timeout = probe_timeout_ms(state);
message->timeout_ms[0] = timeout >> 8; message->timeout_ms[1] = timeout;
return 0;
}
@ -644,11 +649,7 @@ static int owner_observe(struct nat_socket *state) {
const struct sockaddr_storage *target = &state->observer_address;
const uint8_t *key = state->observer_key;
if (state->phase == NAT_OBSERVE_C || state->phase == NAT_REPEAT_C) { target = &state->helper_address; key = state->helper_key; }
int timeout = 1000;
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;
int timeout = probe_timeout_ms(state);
if (state->phase == NAT_OBSERVE_C && !state->retries) {
state->contact_cutoff = state->contact_count;
if (contacted(state, &state->helper_address)) state->filter_clean = 0;
@ -732,7 +733,8 @@ static void owner_ping_cb(int success, uint16_t rtt, void *arg, uint64_t nonce,
ADDRESS_MAP_UNCHANGED : ADDRESS_MAP_EIM) : ADDRESS_MAP_DEPENDENT;
uint8_t inbound = state->filter_clean && state->pre_received ? ADDRESS_IN_OPEN :
state->filter_clean && state->pre_healthy && state->post_received && same &&
state->heartbeat_received >= state->pre_started ? ADDRESS_IN_CONTACT : ADDRESS_IN_UNKNOWN;
state->heartbeat_received >= state->pre_started &&
get_time_tb() - state->heartbeat_received <= (uint64_t)probe_timeout_ms(state) * 20 ? ADDRESS_IN_CONTACT : ADDRESS_IN_UNKNOWN;
owner_apply(state, mapping, inbound); return;
} else {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT observation in unexpected phase=%u", state->phase);
@ -888,7 +890,7 @@ static int helper_send(struct nat_probe *probe) {
struct sockaddr_storage target;
if (route_ping_addr_decode(probe->message.target, &target)) return -1;
int timeout = ((unsigned)probe->message.timeout_ms[0] << 8) | probe->message.timeout_ms[1];
if (timeout < 500 || timeout > 10000) {
if (timeout < 500 || timeout > NAT_ROUND_TB / 10) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT helper has invalid timeout_ms=%d", timeout); return -1;
}
int result;
@ -1060,9 +1062,10 @@ static void handle_result(struct nat_socket *state, const struct nat_wire *messa
return;
}
state->pre_healthy = step == NAT_STEP_PRE ? message->sent == NAT_PROBE_COUNT &&
state->heartbeat_received >= state->pre_started : state->pre_healthy;
state->heartbeat_received >= state->pre_started &&
get_time_tb() - state->heartbeat_received <= (uint64_t)probe_timeout_ms(state) * 20 : state->pre_healthy;
state->phase = step == NAT_STEP_PRE ? NAT_PRE_WAIT : NAT_POST_WAIT;
state->due = get_time_tb() + 10000; /* Отчёт router может обогнать последнюю UDP-пробу. */
state->due = get_time_tb() + (uint64_t)probe_timeout_ms(state) * 10; /* Router может обогнать последнюю UDP-пробу. */
DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT ingress series complete socket=%s round=%016llx step=%u sent=%u pre=%u post=%u clean=%u anchor=%u",
state->socket->name, (unsigned long long)state->round, step, message->sent,
state->pre_received, state->post_received, state->filter_clean, state->pre_healthy);
@ -1250,7 +1253,7 @@ static void detector_tick(void *arg) {
}
if (!state->heartbeat_active && state->baseline.ss_family && now >= state->heartbeat_due) {
if (route_ping_observe(state->socket, state->observer_key, &state->observer_address, state->group_id, state->round,
1000, owner_heartbeat_cb, state)) {
probe_timeout_ms(state), owner_heartbeat_cb, state)) {
DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT anchor probe send failed socket=%s round=%016llx",
state->socket->name, (unsigned long long)state->round);
state->heartbeat_due = now + 10000;
@ -1297,16 +1300,22 @@ static uint64_t candidate_fingerprint(struct nat_socket *state, int ready) {
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, NULL, &endpoint, &socket_id)) continue;
if (route_ping_addr_encode(bytes + 16, &endpoint)) continue;
bytes[16 + ROUTE_ADDR_WIRE_SIZE] = socket_id;
struct sockaddr_storage observer_address = {0}; uint8_t observer_socket = 0;
observers += node_udp_endpoint(detector, node, state->local.ss_family, 0, NULL, &observer_address, &observer_socket);
}
candidates++;
uint64_t hash = UINT64_C(14695981039346656037);
for (unsigned i = 0; i < sizeof(bytes); i++) { hash ^= bytes[i]; hash *= UINT64_C(1099511628211); }
group_total += hash;
uint8_t previous[ROUTE_ADDR_WIRE_SIZE] = {0};
int more = state->socket->is_tcp || node_udp_endpoint(detector, node, state->local.ss_family, 1, previous, &endpoint, &socket_id);
for (; more; more = !state->socket->is_tcp &&
node_udp_endpoint(detector, node, state->local.ss_family, 1, previous, &endpoint, &socket_id)) {
if (!state->socket->is_tcp) {
if (route_ping_addr_encode(previous, &endpoint)) break;
memcpy(bytes + 16, previous, sizeof(previous)); 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); }
group_total += hash;
}
}
if (ready || (candidates >= 2 && observers)) total += group_total;
if (!ready) DEBUG_TRACE(DEBUG_CATEGORY_NAT, "NAT cold group candidates node=%016llx socket=%s group=%016llx candidates=%u observers=%u",

61
tests/test_nat_detection_faults.c
Unescape View File

@ -4,7 +4,7 @@
#include "test_clock.h"
enum fault_mode { FAULT_NONE, FAULT_NO_PONG, FAULT_FILTERED, FAULT_LATE_PRE, FAULT_CONTACT,
FAULT_DEPENDENT, FAULT_UNSTABLE, FAULT_NO_REPORT };
FAULT_DEPENDENT, FAULT_UNSTABLE, FAULT_NO_REPORT, FAULT_INGRESS_BLOCKED, FAULT_NO_ANCHOR };
static struct {
struct nat_fixture *fixture;
@ -57,7 +57,7 @@ ssize_t __wrap_socket_sendto(socket_t fd, const void *packet, size_t length, con
struct sockaddr_storage destination = {0}; assert(target_length <= sizeof(destination));
memcpy(&destination, target, target_length);
int to_owner = route_ping_addr_equal(&destination, &fault.socket->local_addr, 1);
if (fault.mode == FAULT_NO_PONG && to_owner && plain[3] == ETCP_PONG) {
if ((fault.mode == FAULT_NO_PONG || fault.mode == FAULT_NO_ANCHOR) && to_owner && plain[3] == ETCP_PONG) {
fault.dropped++; return length;
}
if (source == fault.socket && plain[3] == ETCP_PING && (plain[4] & ETCP_PING_FLAG_OBSERVE) &&
@ -79,7 +79,8 @@ ssize_t __wrap_socket_sendto(socket_t fd, const void *packet, size_t length, con
&source->local_addr, 500, ignored_ping, NULL, NULL, 0, 0));
DEBUG_INFO(DEBUG_CATEGORY_NAT, "fault fixture: real outgoing PING contaminates PRE source IP");
}
if ((fault.mode == FAULT_FILTERED || fault.mode == FAULT_LATE_PRE) && !fault.contacted) {
if (fault.mode == FAULT_INGRESS_BLOCKED ||
((fault.mode == FAULT_FILTERED || fault.mode == FAULT_LATE_PRE || fault.mode == FAULT_NO_ANCHOR) && !fault.contacted)) {
if (fault.mode == FAULT_LATE_PRE && !fault.delayed_length && !fault.injected) {
memcpy(fault.delayed, packet, length); fault.delayed_length = length; fault.delayed_fd = fd;
fault.delayed_target = destination; fault.delayed_target_length = target_length;
@ -138,6 +139,57 @@ static void run_case(enum fault_mode mode, uint8_t mapping, uint8_t inbound) {
DEBUG_INFO(DEBUG_CATEGORY_NAT, "PASS NAT fault mode=%d", mode);
}
/* Разные ключи не означают разные IP: неподходящего помощника отклоняет получатель. */
static void run_alias(const char *alias_ip) {
struct nat_fixture fixture; nat_fixture_init(&fixture);
struct SC_MYKEYS first, second; assert(sc_generate_keypair(&first) == SC_OK && sc_generate_keypair(&second) == SC_OK);
if (sc_derive_node_id_from_pubkey(first.public_key) > sc_derive_node_id_from_pubkey(second.public_key)) {
struct SC_MYKEYS swap = first; first = second; second = swap;
}
struct UTUN_INSTANCE *owner = nat_fixture_node(&fixture, "127.0.0.2", NULL, NULL);
struct UTUN_INSTANCE *observer = nat_fixture_node(&fixture, "127.0.0.3", NULL, NULL);
struct UTUN_INSTANCE *alias = nat_fixture_node(&fixture, alias_ip, &first, NULL);
struct UTUN_INSTANCE *helper = nat_fixture_node(&fixture, "127.0.0.4", &second, NULL);
nat_detection_set_automatic(observer->nat_det, 0); nat_detection_set_automatic(alias->nat_det, 0);
nat_detection_set_automatic(helper->nat_det, 0);
nat_fixture_connect(&fixture, observer, "udp", alias); nat_fixture_connect(&fixture, observer, "udp", helper);
nat_fixture_connect(&fixture, owner, "udp", observer); nat_detection_set_automatic(owner->nat_det, 0);
struct ETCP_SOCKET *socket = test_etcp_bound_socket(owner, "udp");
assert(!nat_detection_request_check_socket(owner->nat_det, socket, observer->node_id));
struct nat_check_status status = {0}; uint64_t end = get_time_tb() + 400000;
do { test_clock_poll(fixture.ua, 1000); assert(!nat_detection_get_status(owner->nat_det, socket, &status)); }
while (get_time_tb() < end && (!status.rounds || status.phase || status.cycling));
assert(socket->address_profile.mapping == ADDRESS_MAP_UNCHANGED && socket->address_profile.inbound == ADDRESS_IN_OPEN);
assert(status.helper == helper->node_id && status.helper_reselections);
nat_fixture_cleanup(&fixture);
DEBUG_INFO(DEBUG_CATEGORY_NAT, "PASS NAT receiver rejects helper source=%s", alias_ip);
}
/* У наблюдателя первый UDP listener недоступен, но управление по TCP и второй IP работают. */
static void run_endpoint_retry(void) {
struct nat_fixture fixture; nat_fixture_init(&fixture);
struct UTUN_INSTANCE *owner = nat_fixture_node(&fixture, "127.0.0.2", NULL,
"[server:tcp]\naddr=127.0.0.2:0\ntransport=tcp\ntype=public\n");
struct UTUN_INSTANCE *observer = nat_fixture_node(&fixture, "127.0.0.3", NULL,
"[server:other]\naddr=127.0.0.8:0\ntype=public\n[server:tcp]\naddr=127.0.0.3:0\ntransport=tcp\ntype=public\n");
struct UTUN_INSTANCE *helper = nat_fixture_node(&fixture, "127.0.0.4", NULL,
"[server:tcp]\naddr=127.0.0.4:0\ntransport=tcp\ntype=public\n");
nat_detection_set_automatic(observer->nat_det, 0); nat_detection_set_automatic(helper->nat_det, 0);
struct ETCP_SOCKET *bad = test_etcp_bound_socket(observer, "udp");
assert(!uasync_remove_socket_t(fixture.ua, bad->fd)); bad->socket_id = NULL;
nat_fixture_connect(&fixture, observer, "tcp", helper); nat_fixture_connect(&fixture, owner, "tcp", observer);
nat_detection_set_automatic(owner->nat_det, 0);
struct ETCP_SOCKET *socket = test_etcp_bound_socket(owner, "udp");
assert(!nat_detection_request_check_socket(owner->nat_det, socket, observer->node_id));
struct nat_check_status status = {0}; uint64_t end = get_time_tb() + 400000;
do { test_clock_poll(fixture.ua, 1000); assert(!nat_detection_get_status(owner->nat_det, socket, &status)); }
while (get_time_tb() < end && (!status.rounds || status.phase || status.cycling));
assert(socket->address_profile.mapping == ADDRESS_MAP_UNCHANGED && socket->address_profile.inbound == ADDRESS_IN_OPEN);
assert(status.helper_reselections && status.observer == observer->node_id);
nat_fixture_cleanup(&fixture);
DEBUG_INFO(DEBUG_CATEGORY_NAT, "PASS NAT retries another observer IP after timeout");
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_NAT, DEBUG_LEVEL_DEBUG);
test_clock_init(1000);
@ -148,5 +200,8 @@ int main(void) {
run_case(FAULT_DEPENDENT, ADDRESS_MAP_DEPENDENT, ADDRESS_IN_OPEN);
run_case(FAULT_UNSTABLE, ADDRESS_MAP_UNKNOWN, ADDRESS_IN_UNKNOWN);
run_case(FAULT_NO_REPORT, ADDRESS_MAP_UNKNOWN, ADDRESS_IN_UNKNOWN);
run_case(FAULT_INGRESS_BLOCKED, ADDRESS_MAP_UNCHANGED, ADDRESS_IN_UNKNOWN);
run_case(FAULT_NO_ANCHOR, ADDRESS_MAP_UNCHANGED, ADDRESS_IN_UNKNOWN);
run_alias("127.0.0.3"); run_alias("127.0.0.2"); run_endpoint_retry();
test_clock_finish(); return 0;
}

Write Preview
Loading…
Cancel
Save