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

Verify relay handoff expiry, replay protection and complete teardown

stream-multiview
evgeny 1 week ago
parent
035a2eb6d0
commit
e73fab0a73
3 changed files with 37 additions and 21 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. 2
      src/stream/stream.c
  2. 7
      src/stream/stream_network.c
  3. 49
      tests/test_stream_service.c

2
src/stream/stream.c
Unescape View File

@ -549,7 +549,7 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript
if (created) session_close(s, STREAM_CLOSE_FAILED, 0, 0);
}
u_free(h); s->holds--;
if (!s->holds && !s->delivery) session_free(s);
if (s->closing && !s->holds && !s->delivery) session_free(s);
return result;
}
s->holds--;

7
src/stream/stream_network.c
Unescape View File

@ -748,8 +748,6 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
struct stream_message m; struct stream_local local;
if (stream_wire_parse(wire, length, &m) != STREAM_OK || !sender || sender == net->inst->node_id) return;
if (m.command >= STREAM_CMD_RELAY_REQUEST) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay dispatcher self=%llx sender=%llx command=%u routed=%d bytes=%zu",
(unsigned long long)net->inst->node_id, (unsigned long long)sender, m.command, routed, length);
if (routed) relay_dispatch(net, sender, &m);
else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay command requires signed routed control peer=%llx command=%u",
(unsigned long long)sender, m.command);
@ -910,11 +908,6 @@ static void receive_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
/* Router сообщает реального отправителя и признаки проверенной подписи, не next hop. */
static void receive_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
struct stream_network *net = conn ? stream_network_get(conn->instance) : NULL;
if (entry && entry->dgram && entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
&& entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] >= STREAM_CMD_RELAY_REQUEST)
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay routed entry self=%llx net=%p bytes=%u flags=%u cmd=%u",
(unsigned long long)(conn && conn->instance ? conn->instance->node_id : 0), (void *)net, entry->len,
entry->dgram[ROUTER_SVC_FLAGS_OFF], entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2]);
if (net && entry && entry->dgram && entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
&& (entry->dgram[ROUTER_SVC_FLAGS_OFF] & ROUTER_FLAG_SIGNED)) {
uint64_t sender, group; memcpy(&sender, entry->dgram + ROUTER_SVC_SRC_OFF, 8);

49
tests/test_stream_service.c
Unescape View File

@ -1018,10 +1018,6 @@ static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
/* Сохраняем реальный подписанный routed-запрос для проверки позднего повтора после освобождения буфера. */
static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay test routed callback self=%llx bytes=%u payload_command=%u flags=%u",
(unsigned long long)(conn && conn->instance ? conn->instance->node_id : 0), entry->len,
entry->len > ROUTER_SVC_PAYLOAD_OFF + 2 ? entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] : 0,
entry->len > ROUTER_SVC_FLAGS_OFF ? entry->dgram[ROUTER_SVC_FLAGS_OFF] : 0);
if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
&& entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_REQUEST) {
memcpy(relay_request_wire, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, STREAM_WIRE_HEADER); relay_requests_seen++;
@ -1037,12 +1033,18 @@ static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_eve
if (event->type == STREAM_EVENT_END) relay_ends[index]++;
}
/* Повторяем управление через штатный подписанный router, сохраняя nonce/эпохи исходной попытки. */
static void relay_test_control(const struct stream_message *message) {
struct ll_entry *entry = ll_alloc_lldgram(1 + STREAM_WIRE_HEADER); CHECK(entry);
entry->dgram[0] = ETCP_RT_ID_STREAM; stream_wire_header(entry->dgram + 1, message);
entry->len = 1 + STREAM_WIRE_HEADER;
CHECK(etcp_route_send(nodes[2], GROUP2, nodes[1]->node_id, entry, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0);
}
/* Все проверки помощи используют подписанное управление и настоящий raw ETCP, не прямой ingest. */
static void test_supernode_relay(void) {
int route_level = g_debug_config.category_levels[DEBUG_CATEGORY_ETCPROUTE];
debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_INFO);
int diagnostic_level = g_debug_config.category_levels[DEBUG_CATEGORY_DEBUG];
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_DEBUG);
struct TOPO_PEER_REQUEST *connections[3] = {0};
uint64_t deadline = get_time_tb() + 150000;
for (unsigned i = 0; i < 3; i++) {
@ -1059,8 +1061,6 @@ static void test_supernode_relay(void) {
stream_set_event_callback(nodes[i], relay_test_event, NULL);
}
relay_route_handler = nodes[1]->router_bindings.callbacks[ETCP_RT_ID_STREAM];
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay test original routed handler=%p network=%p", (void *)relay_route_handler,
(void *)stream_network_get(nodes[1]));
CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_test_routed) == 0);
struct stream_parameters parameters = {.n = 2, .r = 1, .block_size = STREAM_RS_DEFAULT_SIZE};
struct STREAM_HANDLE *publisher = NULL, *viewers[3] = {0};
@ -1099,12 +1099,18 @@ static void test_supernode_relay(void) {
stream_close(viewers[1]); viewers[1] = NULL;
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
/* Поздний повтор завершённого запроса не подписывает суперузел заново. */
struct ll_entry *entry = ll_alloc_lldgram(1 + STREAM_WIRE_HEADER); CHECK(entry);
entry->dgram[0] = ETCP_RT_ID_STREAM; memcpy(entry->dgram + 1, relay_request_wire, STREAM_WIRE_HEADER);
entry->len = 1 + STREAM_WIRE_HEADER;
CHECK(etcp_route_send(nodes[2], GROUP2, nodes[1]->node_id, entry, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0);
struct stream_message replay; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &replay) == STREAM_OK);
CHECK(replay.receiver_epoch); relay_test_control(&replay);
test_clock_run_for(ua, 50000);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
/* Даже после удаления tombstone старое подтверждение поколения не может снова выделить owner. */
test_clock_run_for(ua, 700000); relay_test_control(&replay); test_clock_run_for(ua, 20000);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
replay.handle ^= UINT64_C(0x123456789abcdef0); replay.sender_epoch ^= UINT64_C(0x9876543210fedcba);
replay.receiver_epoch = 0; replay.command = STREAM_CMD_RELAY_CANCEL; relay_test_control(&replay);
replay.command = STREAM_CMD_RELAY_REQUEST; relay_test_control(&replay);
test_clock_run_for(ua, 20000);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
/* Два relay-only экземпляра не выбирают друг друга, даже если имеют буферы и живые BGP-пути. */
struct STREAM_HANDLE *helpers[2] = {0};
relay_role(0, 1, 100);
@ -1119,6 +1125,22 @@ static void test_supernode_relay(void) {
}
relay_role(0, 0, 101);
stream_close(publisher); publisher = NULL; test_clock_run_for(ua, 10000);
/* READY без SUBSCRIBE истекает; публичный reader сохраняется, внутренний приём останавливается. */
CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
for (unsigned i = 0; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK);
CHECK(stream_subscribe_manual_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
stream_relay_demand(nodes[2], &origin.descriptor, 1, get_time_tb() / 10);
deadline = get_time_tb() + 100000;
while (!stream_relay_confirmed(nodes[2], &origin.descriptor, nodes[1]->node_id) && get_time_tb() < deadline)
test_clock_poll(ua, 100);
CHECK(stream_relay_confirmed(nodes[2], &origin.descriptor, nodes[1]->node_id));
test_clock_run_for(ua, 100000);
CHECK(!stream_relay_confirmed(nodes[2], &origin.descriptor, nodes[1]->node_id));
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT && stream_handle_active(viewers[2]));
stream_close(viewers[2]); viewers[2] = NULL; stream_close(publisher); publisher = NULL;
test_clock_run_for(ua, 10000);
/* Пустой END проходит помощь до READY/DATA; буфер освобождается, хотя источник handle ещё жив. */
CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
@ -1141,7 +1163,6 @@ static void test_supernode_relay(void) {
}
for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]);
debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, route_level);
debug_set_category_level(DEBUG_CATEGORY_DEBUG, diagnostic_level);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] supernode assistance: direct success, fallback, shared viewers, idle release, replay, relay-only exclusion, empty END");
}
@ -1202,6 +1223,8 @@ int main(int argc, char **argv) {
nodes[i] = utun_instance_create_from_str(ua, config); CHECK(nodes[i] != NULL);
snprintf(nodes[i]->name, sizeof(nodes[i]->name), "node%u", i);
CHECK(utun_core_start(nodes[i]) == 0 && chat_service_start(nodes[i]) == 0 && nodes[i]->stream);
CHECK(nodes[i]->router_bindings.callbacks[ETCP_RT_ID_STREAM]
&& nodes[i]->router_bindings.callbacks[ETCP_RT_ID_STREAM] != nodes[i]->router_bindings.callbacks[ETCP_RT_ID_MEMBER_FILES]);
for (unsigned g = 0; g < 2; g++) {
char id[32]; uint64_t group = g ? GROUP2 : GROUP1; snprintf(id, sizeof(id), "%llu", (unsigned long long)group);
uint8_t signature[64] = {0};

Write Preview
Loading…
Cancel
Save