Browse Source

Confirm new TCP sessions and bound reciprocal challenge retries

Negotiate new TCP sessions even when the peer epoch is known, and let the retry timer send reciprocal challenges to avoid feedback loops when confirmations are lost.

Cover peer restarts, simultaneous connects, REALITY, confirmation loss, second links, stale confirmations and deadlines. Remove forced client resets and count only validated test payloads in the UDP reconnect test.

Validation: 149 tests passed; proxy, burst and load integration tests passed; Android assembleDebug succeeded.
master
evgeny 7 days ago
parent
commit
90d3206071
  1. 3
      src/transport_layer/etcp_connections.c
  2. 6
      src/transport_layer/etcp_session.c
  3. 6
      tests/Makefile.am
  4. 32
      tests/test_etcp_reconnect.c
  5. 113
      tests/test_etcp_session.c
  6. 379
      tests/test_etcp_stcp.c

3
src/transport_layer/etcp_connections.c

@ -1287,7 +1287,8 @@ void etcp_link_enter_ready_tcp(struct ETCP_LINK *link) {
if (!link->mtu_remote) link->mtu_remote = link->mtu;
etcp->tcp_link_count++;
if (link->tcp_link) { const uint8_t* ed = stcp_link_get_peer_ed25519_pubkey(link->tcp_link); if (ed) memcpy(etcp->peer_ed25519_pubkey, ed, SC_PUBKEY_SIZE); }
if (!etcp->peer_reset_id) etcp_session_start(etcp); // сервер создаёт ETCP_CONN после STCP handshake
// Известная эпоха пира не подтверждает новый поток; следующий линк не перезапускает согласование.
if (!etcp->initialized && !etcp->session_required) etcp_session_start(etcp);
if (etcp->initialized == 0) etcp_conn_ready(etcp);
if (etcp->state == 2) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "enter_ready_tcp: conn DELETED after conn_ready, link=%p", (void*)link); return; }
etcp_keepalive_link_ready(link);

6
src/transport_layer/etcp_session.c

@ -93,10 +93,8 @@ int etcp_session_receive(struct ETCP_CONN* c, const uint8_t* data, size_t len) {
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP,"[%s] session: stale target or empty cookie",c->log_name);
} else if (data[0] == ETCP_SESSION_CHALLENGE) {
session_send(c,ETCP_SESSION_CONFIRM,peer,cookie);
if (c->session_required || peer != c->peer_reset_id) {
etcp_session_observe(c,peer);
if (c->session_challenge) session_send(c,ETCP_SESSION_CHALLENGE,peer,c->session_challenge);
}
// Встречный challenge отправляет таймер: немедленный ответ зацикливает обмен при потере CONFIRM.
if (c->session_required || peer != c->peer_reset_id) etcp_session_observe(c,peer);
} else if (peer == c->session_candidate && cookie == c->session_challenge) {
etcp_session_cancel(c);
if (peer != c->peer_reset_id) etcp_conn_reinit_id(c,"confirmed peer epoch",c->reset_id);

6
tests/Makefile.am

@ -326,9 +326,13 @@ test_transport_SOURCES = test_stcp_link.c
test_transport_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
test_transport_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
test_etcp_transport_SOURCES = test_etcp_stcp.c
test_etcp_transport_SOURCES = test_etcp_stcp.c $(test_clock_sources)
test_etcp_transport_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
test_etcp_transport_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
test_etcp_transport_LDFLAGS = $(test_clock_ldflags)
if OS_LINUX
test_etcp_transport_LDFLAGS += -Wl,--wrap=etcp_encrypt_send
endif
test_stcp_SOURCES = test_stcp.c
test_stcp_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib

32
tests/test_etcp_reconnect.c

@ -53,6 +53,7 @@ static int restart_action_done = 0;
static int _dump_interval = 50;
static int _dump_timer = 0;
static uint8_t packet_buffer[PACKET_SIZE];
static uint8_t received_sequences[8][TOTAL_PACKETS];
static int write_configs(void) {
FILE* f = fopen(server_config_path, "w");
@ -115,7 +116,12 @@ static int is_connection_established(struct UTUN_INSTANCE* inst) {
struct ll_entry* entry = inst->connections->head;
while (entry) {
struct conn_queue_entry* ce = (struct conn_queue_entry*)entry->data;
if (ce->conn->initialized) return 1;
struct ETCP_CONN* conn = ce->conn;
struct UTUN_INSTANCE* peer = inst == client_instance ? server_instance : client_instance;
struct ETCP_CONN* remote = peer ? instance_find_conn(peer, inst->node_id) : NULL;
if (remote && conn->initialized && remote->initialized && !conn->session_required && !remote->session_required &&
!conn->reinit_pending && !remote->reinit_pending && conn->peer_reset_id == remote->reset_id &&
remote->peer_reset_id == conn->reset_id) return 1;
entry = entry->next;
}
return 0;
@ -131,7 +137,18 @@ static void drain_received(int count_flag) {
queue_set_callback(conn->output_queue, NULL, NULL);
struct ETCP_FRAGMENT* pkt;
while ((pkt = (struct ETCP_FRAGMENT*)queue_data_get(conn->output_queue)) != NULL) {
if (count_flag) packets_received++;
// output_queue содержит также служебные кодограммы; считаем только наши маркированные пакеты.
if (count_flag && pkt->ll.len >= 2 && pkt->ll.dgram[0] == 'T' && pkt->ll.dgram[1] == 'R') {
int valid = pkt->ll.len == PACKET_SIZE;
unsigned seq = valid ? pkt->ll.dgram[4] | ((unsigned)pkt->ll.dgram[5] << 8) : TOTAL_PACKETS;
if (valid) valid = pkt->ll.dgram[2] == phase && seq < TOTAL_PACKETS && !received_sequences[phase][seq];
for (int i = 6; valid && i < PACKET_SIZE; i++) valid = pkt->ll.dgram[i] == (uint8_t)(seq + i + phase);
if (valid) { received_sequences[phase][seq] = 1; packets_received++; }
else {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "reconnect: invalid/duplicate payload phase=%d seq=%u len=%u", phase, seq, pkt->ll.len);
test_completed = 2;
}
}
if (pkt->ll.dgram) memory_pool_free(conn->instance->data_pool, pkt->ll.dgram);
queue_entry_free((struct ll_entry*)pkt);
}
@ -146,8 +163,9 @@ static void send_packets(void) {
if (!conn || !conn->input_queue) return;
while (packets_sent < TOTAL_PACKETS) {
if (queue_entry_count(conn->input_queue) >= MAX_QUEUE_SIZE) break;
packet_buffer[0] = (uint8_t)(packets_sent & 0xFF);
for (int i = 1; i < PACKET_SIZE; i++) packet_buffer[i] = (uint8_t)((packets_sent + i) % 256);
packet_buffer[0] = 'T'; packet_buffer[1] = 'R'; packet_buffer[2] = phase; packet_buffer[3] = 0;
packet_buffer[4] = packets_sent; packet_buffer[5] = packets_sent >> 8;
for (int i = 6; i < PACKET_SIZE; i++) packet_buffer[i] = (uint8_t)(packets_sent + i + phase);
if (etcp_int_send(conn, packet_buffer, PACKET_SIZE) != 0) break;
packets_sent++;
}
@ -207,12 +225,10 @@ static void monitor(void* arg) {
return;
}
printf("Server recreated (node_id=%llx)\n", (unsigned long long)server_instance->node_id);
// Force client to reinit — otherwise conn_ok stays true on old initialized=1
{ struct ll_entry* e = client_instance->connections->head;
while (e) { struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data; etcp_conn_reinit(ce->conn, "test force reinit"); e = e->next; } }
restart_action_done = 1;
}
drain_received(0);
conn_ok = is_connection_established(client_instance);
if (conn_ok) {
printf("=== Phase 3: Reconnected after server restart ===\n");
packets_sent = 0;
@ -228,6 +244,7 @@ static void monitor(void* arg) {
last_reinit_count = cur_reinit;
packets_sent = 0;
packets_received = 0;
memset(received_sequences[phase], 0, sizeof(received_sequences[phase]));
drain_received(0);
}
if (is_connection_established(client_instance))
@ -291,6 +308,7 @@ static void monitor(void* arg) {
last_reinit_count = cur_reinit;
packets_sent = 0;
packets_received = 0;
memset(received_sequences[phase], 0, sizeof(received_sequences[phase]));
drain_received(0);
}
if (is_connection_established(client_instance))

113
tests/test_etcp_session.c

@ -2,6 +2,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <limits.h>
#include "etcp.h"
#include "etcp_api.h"
#include "etcp_connections.h"
@ -171,6 +172,7 @@ static void teardown(void) {
for (int i=0;i<2;i++) {
for (int j=0;j<2;j++) {
if (ep[i].links[j].rx_timeout_timer) uasync_cancel_timeout(ua, ep[i].links[j].rx_timeout_timer);
if (ep[i].links[j].keepalive_timer) uasync_cancel_timeout(ua, ep[i].links[j].keepalive_timer);
}
ep[i].c->links=NULL; // fixture owns synthetic authenticated links
etcp_connection_close(ep[i].c);
@ -179,6 +181,107 @@ static void teardown(void) {
for (int i=0;i<2;i++) { queue_free(ep[i].inst.connections); memory_pool_destroy(ep[i].inst.pkt_pool); }
uasync_poll(ua,0); CHECK(ua->timer_alloc_count == ua->timer_free_count); uasync_destroy(ua,0);
}
// Пересоздаём настоящий ETCP_CONN с теми же ключами; линки/сокеты принадлежат fixture.
static void fresh_endpoint(int i) {
for (int j = 0; j < 2; j++) {
struct ETCP_LINK* link = &ep[i].links[j];
if (link->keepalive_timer) { uasync_cancel_timeout(ua, link->keepalive_timer); link->keepalive_timer = NULL; }
if (link->rx_timeout_timer) { uasync_cancel_timeout(ua, link->rx_timeout_timer); link->rx_timeout_timer = NULL; }
}
ep[i].c->links = NULL;
etcp_connection_close(ep[i].c); uasync_poll(ua, 0);
ep[i].c = etcp_connection_create(&ep[i].inst, "fresh_session"); CHECK(ep[i].c);
ep[i].c->peer_node_id = 2 - i;
CHECK(sc_init_ctx(&ep[i].c->crypto_ctx, &ep[i].inst.my_keys) == SC_OK);
CHECK(sc_set_peer_public_key(&ep[i].c->crypto_ctx, ep[1-i].inst.my_keys.public_key, SC_PEER_PUBKEY_BIN) == SC_OK);
struct ll_entry* entry = ep[i].c->conn_queue_entry;
queue_remove_data(ep[i].c->conn_queue, entry);
((struct conn_queue_entry*)entry->data)->peer_node_id = ep[i].c->peer_node_id;
CHECK(queue_data_put_with_index(ep[i].c->conn_queue, entry) == 0);
ep[i].c->links = &ep[i].links[0];
for (int j = 0; j < 2; j++) ep[i].links[j].etcp = ep[i].c;
etcp_conn_add_cbk(ep[i].c, event, (void*)(intptr_t)i, ETCP_CBK_EVENT_REINIT | ETCP_CBK_EVENT_DOWN);
reinit[i] = down[i] = 0;
}
static void transport_ready(int i, int j) {
struct ETCP_LINK* link = &ep[i].links[j];
link->keepalive_interval = link->ka_period_ms = 1000; link->keepalive_timeout = 10000;
etcp_link_enter_ready_tcp(link);
}
static void run_until(uint64_t end) {
while (get_time_tb() < end) pump();
}
static void initial_session_case(int scenario) {
setup(scenario == 4 || scenario == 5);
uint64_t local = ep[0].c->reset_id;
ep[0].c->next_tx_id = 73; ep[0].c->last_delivered_id = 42; ep[0].c->got_initial_pkt = 1;
if (scenario == 5) {
transport_ready(0, 1);
CHECK(!ep[0].c->session_required && !ep[0].c->session_timer && !reinit[0]);
CHECK(ep[0].c->reset_id == local && ep[0].c->next_tx_id == 73 && ep[0].c->last_delivered_id == 42);
} else {
if (scenario >= 1 && scenario <= 3) fresh_endpoint(0);
fresh_endpoint(1);
etcp_conn_apply_peer_reset_id(ep[1].c, ep[0].c->reset_id);
if (scenario >= 1 && scenario <= 3) {
if (scenario != 1) etcp_conn_apply_peer_reset_id(ep[0].c, ep[1].c->reset_id);
transport_ready(0, 0);
}
transport_ready(1, 0);
CHECK(ep[1].c->session_required && !ep[1].c->initialized);
if (scenario == 3 || scenario == 4) for (int i = 0; i < 2; i++) drop[i][2] = INT_MAX;
if (scenario == 3) {
unsigned before = tail;
run_until(get_time_tb() + 6000);
CHECK(ep[0].c->session_required && ep[1].c->session_required && drops);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Dropped CONFIRM: frames=%u dropped=%u interval=600ms", tail - before, drops);
CHECK(tail - before < 100); // Без таймера встречные CHALLENGE порождают тысячи ответов.
} else if (scenario == 4) {
for (unsigned k = 0; k < 100 && !ep[1].c->session_challenge; k++) pump();
CHECK(ep[1].c->session_required && ep[1].c->session_challenge);
uint64_t cookie = ep[1].c->session_challenge, started = ep[1].c->session_started;
void* timer = ep[1].c->session_timer;
transport_ready(1, 1);
CHECK(ep[1].c->session_challenge == cookie && ep[1].c->session_started == started && ep[1].c->session_timer == timer);
}
memset(drop, 0, sizeof(drop)); converge();
if (scenario == 0 || scenario == 4) {
CHECK(ep[0].c->reset_id == local && ep[0].c->next_tx_id == 1 && ep[0].c->last_delivered_id == 0);
CHECK(reinit[0] == 1 && reinit[1] == 0);
}
}
teardown();
printf("PASS initial session scenario %d (restart, cold start, simultaneous, lost CONFIRM, pending/ready second link)\n", scenario);
}
static void restart_during_confirmation(void) {
setup(0);
fresh_endpoint(1); etcp_conn_apply_peer_reset_id(ep[1].c, ep[0].c->reset_id); transport_ready(1, 0);
for (int i = 0; i < 2; i++) drop[i][2] = INT_MAX;
for (unsigned k = 0; k < 100 && !ep[0].c->session_challenge; k++) pump();
CHECK(ep[0].c->session_challenge);
uint64_t old_peer = ep[0].c->peer_reset_id, candidate = ep[1].c->reset_id;
// Сохраняем правильный ответ первой попытки; после второго рестарта он должен стать stale.
struct ETCP_DGRAM* packet = memory_pool_alloc(ep[1].inst.pkt_pool); CHECK(packet);
packet->link = &ep[1].links[0]; packet->data_len = ETCP_SESSION_CONTROL_SIZE; packet->noencrypt_len = 0;
etcp_session_encode(packet->data, ETCP_SESSION_CONFIRM, candidate, ep[0].c->reset_id);
uint64_t cookie = htobe64(ep[0].c->session_challenge); memcpy(packet->data + 17, &cookie, sizeof(cookie));
unsigned last = tail;
CHECK(etcp_encrypt_send(packet) > 0); memory_pool_free(ep[1].inst.pkt_pool, packet);
CHECK(tail == last + 1); struct frame stale = frames[last % 4096];
fresh_endpoint(1); etcp_conn_apply_peer_reset_id(ep[1].c, ep[0].c->reset_id); transport_ready(1, 0);
CHECK(ep[1].c->reset_id != candidate);
for (unsigned k = 0; k < 100 && ep[0].c->session_candidate != ep[1].c->reset_id; k++) pump();
CHECK(ep[0].c->session_candidate == ep[1].c->reset_id);
deliver(&stale, 0); CHECK(ep[0].c->peer_reset_id == old_peer && !reinit[0]);
memset(drop, 0, sizeof(drop)); converge(); CHECK(reinit[0] == 1);
teardown();
puts("PASS peer restarts during confirmation; old valid CONFIRM cannot adopt the previous epoch");
}
static struct frame stream_packet(int data) {
struct ETCP_DGRAM* p=memory_pool_alloc(ep[0].inst.pkt_pool); CHECK(p);
p->link=&ep[0].links[0]; p->noencrypt_len=0; p->data_len=8; memset(p->data,0,8); p->data[0]=data ? 0 : 1; if (data) p->data[1]=99;
@ -263,6 +366,10 @@ int main(void) {
timestamp_boundaries(); CHECK(u_get_allocated_count() == baseline);
receive_window_case(0); CHECK(u_get_allocated_count() == baseline);
receive_window_case(1); CHECK(u_get_allocated_count() == baseline);
for (int scenario = 0; scenario < 6; scenario++) {
initial_session_case(scenario); CHECK(u_get_allocated_count() == baseline);
}
restart_during_confirmation(); CHECK(u_get_allocated_count() == baseline);
for (int scenario=0;scenario<9;scenario++) {
setup(scenario == 5);
saved[2]=stream_packet(0); saved[3]=stream_packet(1);
@ -272,10 +379,12 @@ int main(void) {
if (scenario == 3) etcp_conn_fatal_reinit(ep[1].c,"simultaneous reset");
if (scenario == 4) { pump(); etcp_conn_fatal_reinit(ep[0].c,"second reset before confirmation"); }
if (scenario == 6) {
blackhole=1; ep[0].c->session_started=get_time_tb()-100000;
blackhole=1;
uint64_t started = ep[0].c->session_started;
CHECK(etcp_conn_ref_take(ep[0].c)==0);
ep[0].c->links=NULL; // timeout closes a real connection, links belong to fixture
while (ep[0].c->state!=2) pump();
while (ep[0].c->state!=2 && get_time_tb() < started + 110000) pump();
CHECK(ep[0].c->state == 2 && get_time_tb() - started >= 100000);
CHECK(!ep[0].c->session_timer && ep[0].c->delete_complete);
etcp_conn_ref_free(ep[0].c); ep[0].c=NULL;
ep[1].c->links=NULL; etcp_connection_close(ep[1].c); uasync_poll(ua,0);

379
tests/test_etcp_stcp.c

@ -1,162 +1,267 @@
// test_etcp_stcp.c — integration: etcp_send + etcp_bind over STCP link
#include "stcp_link.h"
#include "etcp_api.h"
// Реальные STCP/REALITY-линки: встречное подключение и рестарт пира без reset выжившего узла.
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "etcp.h"
#include "etcp_api.h"
#include "etcp_connections.h"
#include "secure_channel.h"
#include "etcp_session.h"
#include "node_conn_direct.h"
#include "topo_group.h"
#include "topo_node.h"
#include "node_conn_direct.h"
#include "../src/utun_instance.h"
#include "../lib/u_async.h"
#include "../lib/ll_queue.h"
#include "utun_instance.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <time.h>
#include "test_clock.h"
static int test_failed = 0;
static struct SC_MYKEYS s_keys, c_keys;
static struct UTUN_INSTANCE *srv_inst, *cli_inst;
static struct ETCP_CONN *srv_conn, *cli_conn;
static int g_srv_recv_count = 0;
static uint8_t g_srv_recv_data[256];
#define CHECK(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "FAIL %s:%d: %s", __func__, __LINE__, #x); exit(1); } } while (0)
#define TEST_CMD 253
#define MESSAGE_COUNT 16
#define FULL_MASK ((1u << MESSAGE_COUNT) - 1)
#define TASSERT(cond) do { \
if (!(cond)) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, " FAIL: %s", #cond); test_failed = 1; return test_failed; } \
} while(0)
struct endpoint {
struct UTUN_INSTANCE* inst;
struct NODE_CONN_DIRECT* owner;
struct ETCP_CONN* conn;
uint64_t id;
uint16_t port;
unsigned received, reinit, deleted;
};
struct message { uint8_t cmd, from, phase, seq, body[64]; };
static struct endpoint endpoints[2];
static struct UASYNC* ua;
static unsigned phase;
static int failed;
static uint64_t stream_packets, session_packets;
static const char* public_keys[2] = {
"b3193173def895bd0fcea6f86af077c7d77216f10395275f627ac18242ec0f01",
"ce8871f07fa056c636d297115f231b08c29cdf94e0d440fce83a07c34416d36a"
};
static const char* private_keys[2] = {
"704f2e012c8fa8768130cb0f988a997dccb628372bc5ceccacc78dcbfec5916f",
"38240cb82199e504686507f11f6eaa4f740fde6f0c425c495e49a523019a5d68"
};
static void etcp_recv_cb(struct ETCP_CONN *conn, struct ll_entry *entry) {
#ifdef __linux__
// Считаем реальные отправки потока/control: тест ловит шторм независимо от скорости CPU.
int __real_etcp_encrypt_send(struct ETCP_DGRAM* packet);
int __wrap_etcp_encrypt_send(struct ETCP_DGRAM* packet) {
if (packet && packet->data_len) {
if (packet->data[0] <= ETCP_SECTION_ACK) stream_packets++;
if (packet->data[0] >= ETCP_SESSION_HELLO && packet->data[0] <= ETCP_SESSION_CONFIRM) session_packets++;
}
return __real_etcp_encrypt_send(packet);
}
#endif
static void fill_message(struct message* message, int from, unsigned seq) {
message->cmd = TEST_CMD; message->from = from; message->phase = phase; message->seq = seq;
for (unsigned i = 0; i < sizeof(message->body); i++) message->body[i] = (phase * 17 + seq * 3 + i) ^ from;
}
static void receive_message(struct ETCP_CONN* conn, struct ll_entry* entry) {
int to = conn->instance == endpoints[0].inst ? 0 : 1;
struct message expected;
int valid = entry->len == sizeof(expected) && entry->dgram;
const struct message* received = valid ? (const struct message*)entry->dgram : NULL;
if (valid) valid = received->phase == phase && received->from == 1 - to && received->seq < MESSAGE_COUNT;
if (valid) {
fill_message(&expected, 1 - to, received->seq);
valid = !memcmp(received, &expected, sizeof(expected)) && !(endpoints[to].received & (1u << received->seq));
}
if (valid) endpoints[to].received |= 1u << received->seq;
else { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "Invalid/duplicate message: to=%d phase=%u bytes=%u", to, phase, entry->len); failed = 1; }
queue_dgram_free(entry); queue_entry_free(entry);
}
static void connection_event(struct ETCP_CONN* conn, int event, void* arg) {
(void)conn;
g_srv_recv_count++;
if (entry->dgram && entry->len) {
size_t n = entry->len < 256 ? entry->len : 255;
memcpy(g_srv_recv_data, entry->dgram, n);
struct endpoint* endpoint = arg;
if (event == ETCP_CBK_EVENT_REINIT) endpoint->reinit++;
if (event == ETCP_CBK_EVENT_DELETE) endpoint->deleted++;
}
static void connect_done(void* arg, struct ETCP_CONN* conn, int type) {
struct endpoint* endpoint = arg;
if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "Connection failed: local=%016llx", (unsigned long long)endpoint->id); failed = 1; }
else if (type & ETCP_CONNECT_EARLY) endpoint->conn = conn;
}
static void start_endpoint(int i, int reality) {
struct endpoint* endpoint = &endpoints[i];
struct SC_MYKEYS keys;
CHECK(sc_init_local_keys(&keys, public_keys[i], private_keys[i]) == SC_OK);
uint64_t id = sc_derive_node_id_from_pubkey(keys.public_key);
char config[2048], reality_config[512] = "";
if (reality) snprintf(reality_config, sizeof(reality_config),
"[reality]\nenabled=1\nprivate_key=%s\nshort_ids=0102030405060708\nserver_name=localhost.test\n"
"dest=127.0.0.1:1\nfingerprint=openssl\n", private_keys[i]);
snprintf(config, sizeof(config),
"[global]\nmy_node_id=0x%016llx\nmy_node_name=transport-%d\nmy_public_key=%s\nmy_private_key=%s\n"
"[server:stcp]\naddr=127.0.0.1:%u\ntype=public\ntransport=tcp\nreality=%d\n%s",
(unsigned long long)id, i, public_keys[i], private_keys[i], endpoint->port, reality, reality_config);
endpoint->inst = utun_instance_create_from_str(ua, config); CHECK(endpoint->inst);
CHECK(utun_instance_init(endpoint->inst) >= 0);
CHECK(etcp_bind(endpoint->inst, TEST_CMD, receive_message) == 0);
endpoint->id = endpoint->inst->node_id;
struct ETCP_SOCKET* socket = endpoint->inst->etcp_sockets;
while (socket && !socket->is_tcp) socket = socket->next;
CHECK(socket && socket->local_addr.ss_family == AF_INET);
endpoint->port = ntohs(((struct sockaddr_in*)&socket->local_addr)->sin_port); CHECK(endpoint->port);
endpoint->conn = NULL; endpoint->owner = NULL; endpoint->received = endpoint->reinit = endpoint->deleted = 0;
}
static void stop_endpoint(int i) {
struct endpoint* endpoint = &endpoints[i];
if (endpoint->owner) { node_conn_direct_close(endpoint->owner); endpoint->owner = NULL; }
endpoint->inst->running = 0;
utun_instance_destroy(endpoint->inst); endpoint->inst = NULL; endpoint->conn = NULL;
}
static struct TOPO_NODE* import_peer(int to) {
struct UTUN_INSTANCE* from = endpoints[1-to].inst;
struct TOPO_GROUP_NODE* local = topo_groups_get_default(from->topo_groups)->local_node;
struct TOPO_NODE* info = topo_node_registry_find(from->topo_groups, from->node_id); CHECK(info);
uint8_t wire[4096]; int len = topo_node_serialize(info, local, 0, 0, wire, sizeof(wire), 0); CHECK(len > 0);
struct TOPO_NODE* imported = NULL;
CHECK(topo_node_deserialize(topo_groups_get_default(endpoints[to].inst->topo_groups), wire, len,
&imported, NULL, NULL, NULL, NULL) == 0);
CHECK(topo_node_registry_store(endpoints[to].inst->topo_groups, imported));
return imported;
}
static int pair_ready(void) {
for (int i = 0; i < 2; i++) {
endpoints[i].conn = instance_find_conn(endpoints[i].inst, endpoints[1-i].id);
struct ETCP_CONN* conn = endpoints[i].conn;
if (!conn || !conn->initialized || !conn->links_up || conn->session_required || conn->reinit_pending) return 0;
}
queue_dgram_free(entry);
queue_entry_free(entry);
return endpoints[0].conn->peer_reset_id == endpoints[1].conn->reset_id &&
endpoints[1].conn->peer_reset_id == endpoints[0].conn->reset_id;
}
static void connect_cb(void *arg, struct ETCP_CONN *conn, int type) {
(void)arg;
if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "etcp_connect failed"); test_failed = 1; return; }
if (type & ETCP_CONNECT_EARLY) {
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "etcp_connect: EARLY ready");
cli_conn = conn;
static void wait_pair(void) {
uint64_t end = get_time_tb() + 50000;
while (!failed && get_time_tb() < end) {
test_clock_poll(ua, 100);
if (pair_ready()) return;
}
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "Session recovery timed out: phase=%u stream=%llu control=%llu", phase,
(unsigned long long)stream_packets, (unsigned long long)session_packets);
CHECK(0);
}
static char *build_server_config(int port) {
static char buf[512];
snprintf(buf, sizeof(buf),
"[global]\n"
"my_node_id=0xAAAAAAAAAAAAAAAA\n"
"my_private_key=38240cb82199e504686507f11f6eaa4f740fde6f0c425c495e49a523019a5d68\n"
"my_public_key=ce8871f07fa056c636d297115f231b08c29cdf94e0d440fce83a07c34416d36a\n"
"\n"
"[server: stcp]\n"
"addr=127.0.0.1:%d\n"
"type=public\n"
"transport=tcp\n",
port);
return buf;
static void check_unique(int i) {
unsigned count = 0;
for (struct ll_entry* e = endpoints[i].inst->connections->head; e; e = e->next)
if (((struct conn_queue_entry*)e->data)->peer_node_id == endpoints[1-i].id) count++;
CHECK(count == 1);
}
static char *build_client_config(int port) {
static char buf[512];
snprintf(buf, sizeof(buf),
"[global]\n"
"my_node_id=0xBBBBBBBBBBBBBBBB\n"
"my_private_key=704f2e012c8fa8768130cb0f988a997dccb628372bc5ceccacc78dcbfec5916f\n"
"my_public_key=b3193173def895bd0fcea6f86af077c7d77216f10395275f627ac18242ec0f01\n"
"\n"
"[server: stcp]\n"
"addr=127.0.0.1:%d\n"
"type=public\n"
"transport=tcp\n",
port);
return buf;
static void hold_connection(int i) {
struct endpoint* endpoint = &endpoints[i];
struct TOPO_NODE* peer = topo_node_registry_find(endpoint->inst->topo_groups, endpoints[1-i].id); CHECK(peer);
CHECK(node_conn_direct_open_node(endpoint->inst, peer->node_id, NULL, NULL, &endpoint->owner, peer, NULL) == NCD_REUSED);
CHECK(node_conn_direct_get_conn(endpoint->owner) == endpoint->conn);
etcp_conn_add_cbk(endpoint->conn, connection_event, endpoint, ETCP_CBK_EVENT_REINIT | ETCP_CBK_EVENT_DELETE);
}
int main(void) {
debug_config_init();
debug_set_level(DEBUG_LEVEL_INFO);
debug_set_categories(DEBUG_CATEGORY_GENERAL | DEBUG_CATEGORY_SOCKET | DEBUG_CATEGORY_CRYPTO);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "=== etcp_send/bind over STCP link ===");
srand((unsigned)time(NULL));
struct UASYNC *ua = uasync_create(); TASSERT(ua);
utun_instance_set_tun_init_enabled(0);
srv_inst = utun_instance_create_from_str(ua, build_server_config(0));
TASSERT(srv_inst);
TASSERT(utun_instance_init(srv_inst) >= 0);
cli_inst = utun_instance_create_from_str(ua, build_client_config(0));
TASSERT(cli_inst);
TASSERT(utun_instance_init(cli_inst) >= 0);
TASSERT(etcp_bind(srv_inst, 0, etcp_recv_cb) >= 0);
struct TOPO_GROUP_NODE *srv_node = topo_groups_get_default(srv_inst->topo_groups) ? topo_groups_get_default(srv_inst->topo_groups)->local_node : NULL;
TASSERT(srv_node);
{ struct TOPO_NODE* srv_ni = topo_node_registry_find(srv_inst->topo_groups, srv_node->node_id); TASSERT(srv_ni);
uint8_t wire[4096];
int len = topo_node_serialize(srv_ni, srv_node, 0, 0, wire, sizeof(wire), 0);
TASSERT(len > 0);
struct TOPO_NODE* cli_ni = NULL;
struct TOPO_GROUP* cli_group = topo_groups_get_default(cli_inst->topo_groups);
TASSERT(topo_node_deserialize(cli_group, wire, (size_t)len, &cli_ni, NULL, NULL, NULL, NULL) == 0);
TASSERT(topo_node_registry_store(cli_inst->topo_groups, cli_ni)); }
cli_inst->etcp_connect_timeout_tb = 100000;
TASSERT(etcp_connect(cli_inst, srv_node, connect_cb, NULL, ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE) == 0);
int ticks = 0;
while (!cli_conn && ticks < 5000) { uasync_poll(ua, 10); ticks++; }
TASSERT(cli_conn);
int initial_links = 0;
for (struct ETCP_LINK* link = cli_conn->links; link; link = link->next) initial_links++;
struct NODE_CONN_DIRECT *owner1 = NULL, *owner2 = NULL;
struct TOPO_NODE* peer = topo_node_registry_find(cli_inst->topo_groups, srv_node->node_id);
TASSERT(node_conn_direct_open_node(cli_inst, peer->node_id, NULL, NULL, &owner1, peer, NULL) == NCD_REUSED);
TASSERT(node_conn_direct_open_node(cli_inst, peer->node_id, NULL, NULL, &owner2, peer, NULL) == NCD_REUSED);
TASSERT(node_conn_direct_update_node(cli_inst, peer->node_id) == 0);
int repeated_links = 0;
for (struct ETCP_LINK* link = cli_conn->links; link; link = link->next) repeated_links++;
TASSERT(repeated_links == initial_links);
node_conn_direct_close(owner1);
TASSERT(node_conn_direct_get_conn(owner2) == cli_conn);
{ struct ll_entry* e = srv_inst->connections->head;
while (e) { struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
struct ETCP_CONN *c = ce->conn;
struct ETCP_LINK *l = c->links; while (l) { if (l->is_tcp) { srv_conn = c; break; } l = l->next; }
if (srv_conn) break;
e = e->next; }
static void exchange_messages(void) {
for (int i = 0; i < 2; i++) endpoints[i].received = 0;
for (int i = 0; i < 2; i++) for (unsigned seq = 0; seq < MESSAGE_COUNT; seq++) {
struct ll_entry* entry = ll_alloc_lldgram(sizeof(struct message)); CHECK(entry);
fill_message((struct message*)entry->dgram, i, seq); entry->len = sizeof(struct message);
CHECK(etcp_send(endpoints[i].conn, entry) == 0);
}
uint64_t end = get_time_tb() + 30000;
while (!failed && get_time_tb() < end) {
test_clock_poll(ua, 100);
if (endpoints[0].received != FULL_MASK || endpoints[1].received != FULL_MASK) continue;
int drained = 1;
for (int i = 0; i < 2; i++) {
struct ETCP_CONN* conn = endpoints[i].conn;
if (conn->input_wait_ack->count || conn->input_send_q->count || conn->input_queue->count || conn->send_input_q->count) drained = 0;
}
if (drained) break;
}
TASSERT(srv_conn);
const char *msg = "hello via etcp_send!";
struct ll_entry *e = queue_entry_new(0);
e->dgram = u_malloc(strlen(msg) + 1);
strcpy((char *)e->dgram, msg);
e->len = (uint16_t)strlen(msg);
TASSERT(etcp_send(cli_conn, e) == 0);
ticks = 0;
while (g_srv_recv_count < 1 && ticks < 2000) { uasync_poll(ua, 10); ticks++; }
TASSERT(g_srv_recv_count == 1);
TASSERT(memcmp(g_srv_recv_data, msg, strlen(msg)) == 0);
node_conn_direct_close(owner2);
srv_inst->running = 0; cli_inst->running = 0;
utun_instance_destroy(srv_inst); srv_inst = NULL;
utun_instance_destroy(cli_inst); cli_inst = NULL;
CHECK(!failed && endpoints[0].received == FULL_MASK && endpoints[1].received == FULL_MASK);
for (int i = 0; i < 2; i++) CHECK(!endpoints[i].conn->input_wait_ack->count && !endpoints[i].conn->input_send_q->count);
}
static void run_case(int reality, int simultaneous, int restarted) {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Transport case: reality=%d simultaneous=%d restart=%d", reality, simultaneous, restarted);
memset(endpoints, 0, sizeof(endpoints)); phase = 0; failed = 0; stream_packets = session_packets = 0;
ua = uasync_create(); CHECK(ua);
start_endpoint(0, reality); start_endpoint(1, reality);
struct TOPO_NODE* peers[2] = {import_peer(0), import_peer(1)};
for (int i = 0; i <= simultaneous; i++) {
struct TOPO_GROUP_NODE node = {0}; node.node_id = peers[i]->node_id;
CHECK(etcp_connect(endpoints[i].inst, &node, connect_done, &endpoints[i], ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE) == 0);
}
wait_pair();
for (int i = 0; i < 2; i++) { check_unique(i); hold_connection(i); }
// Повторные владельцы и обновление адресов не создают дополнительные линки.
unsigned links = 0; for (struct ETCP_LINK* l = endpoints[0].conn->links; l; l = l->next) links++;
struct NODE_CONN_DIRECT* extra = NULL;
CHECK(node_conn_direct_open_node(endpoints[0].inst, peers[0]->node_id, NULL, NULL, &extra, peers[0], NULL) == NCD_REUSED);
CHECK(node_conn_direct_update_node(endpoints[0].inst, peers[0]->node_id) == 0);
unsigned updated = 0; for (struct ETCP_LINK* l = endpoints[0].conn->links; l; l = l->next) updated++;
CHECK(updated == links); node_conn_direct_close(extra);
exchange_messages();
if (restarted < 2) {
int survivor = 1 - restarted;
struct ETCP_CONN* retained = endpoints[survivor].conn;
CHECK(etcp_conn_ref_take(retained) == 0);
uint64_t local_epoch = retained->reset_id, peer_epoch = retained->peer_reset_id;
unsigned before = endpoints[survivor].reinit;
uint64_t packets_before = stream_packets;
stop_endpoint(restarted); start_endpoint(restarted, reality);
phase++; wait_pair();
CHECK(!endpoints[survivor].deleted && endpoints[survivor].conn == retained);
CHECK(retained->reset_id == local_epoch && retained->peer_reset_id != peer_epoch);
CHECK(endpoints[survivor].reinit == before + 1);
CHECK(node_conn_direct_get_conn(endpoints[survivor].owner) == retained);
#ifdef __linux__
CHECK(stream_packets - packets_before < 256);
#endif
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Recovery complete: stream=%llu control=%llu REINIT=1 survivor_epoch=%016llx",
(unsigned long long)(stream_packets - packets_before), (unsigned long long)session_packets,
(unsigned long long)local_epoch);
etcp_conn_ref_free(retained);
check_unique(0); check_unique(1); exchange_messages();
} else {
uint64_t old_epochs[2] = {endpoints[0].conn->reset_id, endpoints[1].conn->reset_id};
stop_endpoint(0); stop_endpoint(1); start_endpoint(0, reality); start_endpoint(1, reality);
peers[0] = import_peer(0); peers[1] = import_peer(1);
for (int i = 0; i < 2; i++) {
struct TOPO_GROUP_NODE node = {0}; node.node_id = peers[i]->node_id;
CHECK(etcp_connect(endpoints[i].inst, &node, connect_done, &endpoints[i], ETCP_CONNECT_EARLY) == 0);
}
phase++; wait_pair();
for (int i = 0; i < 2; i++) { CHECK(endpoints[i].conn->reset_id != old_epochs[i]); check_unique(i); }
exchange_messages();
}
stop_endpoint(0); stop_endpoint(1);
uasync_poll(ua, 0);
CHECK(ua->timer_alloc_count == ua->timer_free_count && ua->socket_alloc_count == ua->socket_free_count);
uasync_destroy(ua, 0);
printf("PASS transport reality=%d simultaneous=%d restart=%d: epochs, bidirectional delivery and resource balance\n",
reality, simultaneous, restarted);
}
if (test_failed) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "=== FAILED ==="); return 1; }
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "=== PASSED ===");
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
if (getenv("UTUN_TEST_DEBUG")) {
debug_set_category_level(DEBUG_CATEGORY_ETCP, DEBUG_LEVEL_DEBUG);
debug_set_category_level(DEBUG_CATEGORY_GENERAL, DEBUG_LEVEL_INFO);
}
test_clock_init(100); utun_instance_set_tun_init_enabled(0);
for (int reality = 0; reality < 2; reality++) {
for (int simultaneous = 0; simultaneous < 2; simultaneous++)
for (int restarted = 0; restarted < 2; restarted++) run_case(reality, simultaneous, restarted);
run_case(reality, 1, 2);
}
test_clock_finish();
return 0;
}

Loading…
Cancel
Save