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

Encrypt P2P call media and validate service IDs

Browse Source 
Require signed and encrypted call audio/video using static identity keys. Fix attachment service ID collisions and media download handle lifetime; isolate LAN tests and audit the shared service registry.
This commit is contained in:
evgeny committed 2026-10-21 15:06:27 +02:00
1 parent 8e46cf9254
commit abf342534f
54 files changed
+516 -241
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

doc/chat_call_arch.md
+8 -1
View File
@@ -74,7 +74,14 @@ CALL_MEDIA 0x10 {call_id:8, track_id:1, seq:2, ts_ms:4, capture_pts_us:8, capt
- `call_id` — случайный uint64, генерит вызывающий; идентифицирует сессию на обеих сторонах.
- `group_id` — CHAT-группа или DM-группа (общая маршрутизация).
- Сигналинг: `etcp_route_send(..., ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT)`.
- Медиа: без подписи (уже внутри шифрованного канала etcp_router).
- Сигналинг, аудио и видео: обязательные `ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT` между конечными узлами.
- AES-128-CCM шифрует тело, Ed25519 подписывает заголовок роутера и шифртекст. Посредники не имеют ключа для расшифровки.
- Ключ получается из постоянных X25519-идентичностей собеседников; forward secrecy отсутствует.
- Маска защиты задаётся через `etcp_router_require_crypto`: открытые/частично защищённые данные отбрасываются
на конечном узле до изменения состояния приёма, очередей и `receive_ready`; отправка без требуемых флагов возвращает ошибку.
Пустые служебные пакеты роутера не являются медиаданными. Перехода на открытое медиа при ошибках нет.
- Каждый пакет получает случайный nonce(13) от OpenSSL CSPRNG; ошибка RNG прекращает шифрование.
Надбавка к сервисным данным: nonce(13) + CRC(4) + CCM tag(16) + Ed25519(64) = 97 байт.
- `CALL_MEDIA` — общий контейнер; `track_id` оставляет место видео (`type=VIDEO`).
- Аудио: Opus 48kHz mono, 20ms (960 сэмплов), ~24–32 kbps. ~50 кадров/с.
- Заголовок медиа — 36 байт с subcmd. `ts_ms` — время отправки для оценки сетевого джиттера;
doc/encrypted_attachments.md
+1 -1
View File
@@ -76,7 +76,7 @@ Nonce метаданных использует `part=UINT64_MAX`; файл —
## Адресный протокол
Router-сервис `ETCP_RT_ID_ATTACHMENT_KEYS=0x40`. Каждый пакет обязан иметь
Router-сервис `ETCP_RT_ID_ATTACHMENT_KEYS=0x42`. Каждый пакет обязан иметь
`ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED`; промежуточные узлы ключа не видят.
Протокол не создаёт личную беседу и не публикует служебные сообщения в канал.
doc/tasks.md
+18 -3
View File
@@ -176,13 +176,16 @@
Логи: /tmp/video-warmup-qt-tests.log, /tmp/video-warmup-clock-verified-test.log,
/tmp/video-warmup-full-check.log; подробности: doc/call_video.txt.
[ ] **Разобрать нестабильный flood-тест test_local_transport_packets** — 20.10.2026:
[+] **Разобрать нестабильный flood-тест test_local_transport_packets** — 20.10.2026:
Общий check.sh при подготовке codec cache: line=198 failed: invitations - before == 8.
Отдельный повтор PASS: миллион hostile packets, bounded flood/events/memory, discovery recovery.
Код local transport не менялся. Логи: /tmp/video-warmup-full-check.log,
/tmp/video-warmup-local-packets-retry.log; причина и связь с нагрузкой пока не установлены.
Проверка multicast: line=209, invitations == before + 9; отдельный повтор PASS.
Логи: /tmp/multicast-full-check.log, /tmp/multicast-local-transport-retry.log.
Исправлено 21.10: сетевые LAN-тесты используют общий discovery-порт и мешали друг другу.
Runner запускает оба после общего пула по одному; проверка барьера добавлена в test_run_tests.py.
Полный check.sh 165/165 + proxy/burst/load PASS: /tmp/test-fixes-full-check.log.
[+] **Android: fullscreen вдоль длинной стороны и повторное использование codec probes** — 20.10.2026:
Общее вписывание со стримами поворачивает только изображение большого видео, интерфейс остаётся
@@ -224,14 +227,17 @@
Все 24 реальных KEY декодированы desktop и вторым Android. Подробности: doc/call_video.txt.
Сборки, native video, Android unit и Qt video/camera/clock прошли. Живой звонок после обновления обеих сторон ещё не проверен.
[ ] **Разобрать нестабильный test_local_transport под общей нагрузкой** — 19.10.2026:
[+] **Разобрать нестабильный test_local_transport под общей нагрузкой** — 19.10.2026:
В check.sh для HEVC: 162/164, кроме прежнего encrypted_attachment упал peers():95 failed: found;
перед отказом endpoint peer=2 удалён по TTL (age_ms=30331). Отдельный повтор прошёл.
Логи: /tmp/hevc-high-core-check.log, /tmp/hevc-high-local-transport-retry.log; код local_transport не менялся.
Повторилось при исправлении Android Surface ACK: age_ms=30335, тот же peers():95;
отдельный повтор PASS. Логи /tmp/video-surface-freeze-full-check.log, /tmp/video-surface-freeze-local-retry.log.
21.10 воспроизведено при совместном запуске с test_local_transport_packets: посторонний UDP flood,
endpoint peer=2 удаляется по TTL, отправка request=103 не проходит. Оба LAN-теста теперь выполняются отдельно от пула.
Логи /tmp/local-packets-debug/test_local_transport.log, /tmp/test-fixes-full-check.log (165/165 PASS).
[ ] **Разобрать ожидание запросов ключей после перезапуска автора в test_encrypted_attachment** — 19.10.2026:
[+] **Разобрать ожидание запросов ключей после перезапуска автора в test_encrypted_attachment** — 19.10.2026:
При проверке настроек входящих звонков общий check.sh: 160/162; test_encrypted_attachment остановился в phase=5,
heartbeats=12001. Отдельный повтор также остановился в phase=5: после старта автора не получены два pending incoming-запроса.
Причина и связь с изменениями не установлены. Логи: /tmp/phone-busy-full-check.log,
@@ -241,6 +247,11 @@
артефакты /tmp/utun_encrypted_attachment_IW0W28. Код вложений в этой работе не менялся.
Повторилось при проверке Full HD звонков: check.sh 163/164, тот же phase=5/heartbeats=12001;
/tmp/fullhd-core-check.log, артефакты /tmp/utun_encrypted_attachment_MIwfM3.
Исправлено 21.10: ATTACHMENT_KEYS и RADIO_CALL имели ID=0x40; radio_call перезаписывал обработчик вложений.
Для ATTACHMENT_KEYS выделен 0x42. Следующий этап выявил поздний CM UP с освобождённым media_download:
TIMEOUT терял живой handle. Handle закрывается при отказе, сохраняется при QUERY и разделяется блоками держателя.
Добавлен детерминированный тест отмены позднего UP; вложение проходит все 10 фаз, полный check.sh 165/165 PASS.
Логи /tmp/attachment-core-stack.log, /tmp/media-handles-focused/, /tmp/test-fixes-full-check.log.
[+] **Full HD в видеозвонках и реальные разрешения desktop-камеры** — 19.10.2026:
Desktop использует общий со стримами отбор аппаратных режимов до 1920x1080 при 30/29.97fps.
@@ -259,6 +270,10 @@
В том же общем прогоне storage st1/st2 не собрали файл, n2 исчерпал попытки блока 0 (15/19 внутренних проверок).
Отдельный повтор прошёл 19/19. Причина не установлена; сравнить маршруты и готовность storage под общей нагрузкой.
Логи: tests/logs/test_media_delivery_chat.log, /tmp/phone-busy-media-recheck.log.
Повторилось 21.10 при аудите ID сервисов: 164/165, те же 15/19 внутренних проверок.
Отдельный запуск и 16 параллельных повторов прошли. Причина остаётся открытой;
тест теперь сохраняет подробный media/router лог и БД при отказе в /tmp/utun_mdc_*/.
Логи /tmp/service-ids-full-check.log, /tmp/service-ids-media-chat-debug.log, /tmp/service-ids-media-chat-stress/.
[ ] **Разобрать нестабильное ожидание ACK в test_local_transport** — 19.10.2026:
Общий check.sh: 161/162, chat_adapter упал на строке 173 при ожидании LOCAL_DELIVERY_FORBIDDEN для request=508.
lib/ll_queue.h
+1 -1
View File
@@ -378,7 +378,7 @@ struct ll_entry* queue_entry_new(size_t data_size);
/**
* @brief Создаёт элемент из пула памяти.
* @param pool пул
* @return элемент или NULL, если пул пуст
* @return элемент или NULL при ошибке выделения; пустой кеш пула пополняется автоматически
*/
struct ll_entry* queue_entry_new_from_pool(struct memory_pool* pool);
src/call/call.c
+10 -7
View File
@@ -4,7 +4,7 @@
// медиа (CALL_MEDIA) → HANGUP/ENDED. Звонок из CHAT-группы или DM — модуль
// оперирует (group_id, peer_node_id), маршрутизация через etcp_router.
//
// Сигналинг идёт с ROUTE_CRYPTO_SIGN|ROUTE_CRYPTO_ENCRYPT, медиа — без подписи.
// Сигналинг, аудио и видео требуют ROUTE_CRYPTO_SIGN|ROUTE_CRYPTO_ENCRYPT между конечными узлами.
// Реестр сессий: ll_queue с hash-индексом по call_id (8 байт).
//
// Thread-модель: всё в uasync-потоке (single-threaded). Освобождение сессии
@@ -815,9 +815,9 @@ static void call_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
const uint8_t* body = d + ROUTER_SVC_PAYLOAD_OFF;
size_t blen = entry->len - ROUTER_SVC_PAYLOAD_OFF;
if (subcmd != CALL_SUBCMD_MEDIA &&
(d[ROUTER_SVC_FLAGS_OFF] & (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) != (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "unauthenticated call signaling command=%u", subcmd);
if ((d[ROUTER_SVC_FLAGS_OFF] & (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) != (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "unprotected call packet command=%u peer=%016llx flags=%02x",
subcmd, (unsigned long long)from, d[ROUTER_SVC_FLAGS_OFF]);
queue_dgram_free(entry); queue_entry_free(entry); return;
}
@@ -916,12 +916,15 @@ int call_init(struct UTUN_INSTANCE* inst) {
}
ctx->initialized = 1;
if (etcp_router_bind(inst, ETCP_RT_ID_CALL_AUDIO, call_etcp_recv) || call_video_transport_bind(inst)) {
if (etcp_router_require_crypto(inst, ETCP_RT_ID_CALL, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT)
|| etcp_router_bind(inst, ETCP_RT_ID_CALL_AUDIO, call_etcp_recv)
|| etcp_router_require_crypto(inst, ETCP_RT_ID_CALL_AUDIO, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT)
|| call_video_transport_bind(inst)) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call media channel bind failed");
call_destroy(inst); return -1;
}
call_publish_phone_status(inst);
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: initialized (svc=0x%02x)", CALL_ID, ETCP_RT_ID_CALL);
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: initialized (svc=0x%02x audio/video E2E=AES-128-CCM+Ed25519 required)", CALL_ID, ETCP_RT_ID_CALL);
return 0;
}
@@ -1012,7 +1015,7 @@ static void call_tx_flush(struct call_session* s) {
uint32_t wait_ms = (uint32_t)(get_time_tb() / 10) - packet.ts_ms;
++s->tx_wait_frames; s->tx_wait_total_ms += wait_ms;
if (wait_ms > s->tx_wait_max_ms) s->tx_wait_max_ms = wait_ms;
if (etcp_route_send(ctx->inst, s->group_id, s->peer_node_id, e, 1, 0) == 0) {
if (etcp_route_send(ctx->inst, s->group_id, s->peer_node_id, e, 1, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0) {
s->c_tx_frames++;
if (packet.capture_pts_us != INT64_MIN) call_video_audio_sent(s, packet.capture_pts_us + 20000);
} else DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: media enqueue failed id=%016llx", CALL_ID, (unsigned long long)s->call_id);
src/call/call.h
+7 -4
View File
@@ -1,11 +1,14 @@
// call.h — P2P аудио-звонок (сигналинг + медиа), per-instance
// call — управление P2P аудио/видеозвонком одного узла.
// Начать вызов через start, ответить через accept/decline, завершить через hangup.
// Состояния и результаты приходят событиями CHAT_EVT_CALL_* из chat_event.h.
// Аудио подключается через call_audio.h, видеотрек — через call_video_transport.h или call_video_port.h.
//
// Модуль оперирует (group_id, peer_node_id): звонок можно инициировать из
// CHAT-группы или из DM (group_id — группа маршрутизации, как в dm_core).
// Сигналинг идёт через etcp_router (ROUTE_CRYPTO_SIGN|ENCRYPT), соединение
// Сигналинг, аудио и видео идут через etcp_router с обязательными ROUTE_CRYPTO_SIGN|ENCRYPT; соединение
// оптимизируется через параллельные пробы conn_mgr (DIRECT/REVERSE/INDIRECT).
//
// Все функции — в uasync-потоке. GUI/headless вызывают через trampoline
// Все функции, кроме явно отмеченных, — в uasync-потоке. GUI/headless вызывают через trampoline
// (gui_bridge_post_uasync_fn / uasync_post).
#ifndef CALL_H
#define CALL_H
@@ -62,7 +65,7 @@ int call_send_media(struct UTUN_INSTANCE* inst, uint64_t call_id, const uint8_t*
const struct call_capture_time* timing);
/* Коллбэк приёма медиа: вызывается в uasync-потоке на каждый Opus-кадр пира.
* Используется headless-аудио-сокетом (и в фазе 3 — декодером GUI). */
* Используется headless-аудиосокетом; opus заимствован до возврата callback. */
typedef void (*call_media_recv_fn)(struct UTUN_INSTANCE* inst, uint64_t call_id,
const uint8_t* opus, int len, void* arg);
void call_set_media_recv_cb(struct UTUN_INSTANCE* inst, call_media_recv_fn fn, void* arg);
src/call/call_proto.h
+2 -2
View File
@@ -1,8 +1,8 @@
// call_proto.h — wire-формат и константы P2P аудио-звонка (svc_id ETCP_RT_ID_CALL 0x36)
//
// Приём/отправка через etcp_router. На проводе первый байт payload = subcmd.
// Сигналинг идёт с ROUTE_CRYPTO_SIGN|ROUTE_CRYPTO_ENCRYPT, медиа — без подписи
// (уже внутри шифрованного канала etcp_router).
// Сигналинг и медиа требуют ROUTE_CRYPTO_SIGN|ROUTE_CRYPTO_ENCRYPT: шифрование тела
// и подпись заголовка/шифртекста между конечными узлами, независимо от маршрута.
#ifndef CALL_PROTO_H
#define CALL_PROTO_H
src/call/call_video_transport.c
+14 -1
View File
@@ -262,7 +262,9 @@ static void flush(struct call_video_transport *v) {
f->data = head->data + f->offset;
size_t length;
if (call_video_fragment_encode(f, wire, sizeof(wire), &length)
|| etcp_router_conn_send(conn, wire, length, 0)) { discard_tx(v, "router_send"); return; }
|| etcp_router_conn_send(conn, wire, length, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT)) {
discard_tx(v, "router_send"); return;
}
f->offset += f->length;
if (f->offset == f->size) {
if (f->pts_us % 1000000 < 34000)
@@ -483,6 +485,13 @@ static void receive(struct ETCP_CONN *conn, struct ll_entry *entry) {
}
const uint8_t *data = entry->dgram + ROUTER_SVC_PAYLOAD_OFF;
size_t length = entry->len - ROUTER_SVC_PAYLOAD_OFF;
if (!length) goto done; // Локальное уведомление роутера о закрытии, не медиапакет.
if ((entry->dgram[ROUTER_SVC_FLAGS_OFF] & (ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED))
!= (ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED)) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "unprotected call video packet svc=%u flags=%02x bytes=%zu",
entry->dgram[0], entry->dgram[ROUTER_SVC_FLAGS_OFF], length);
goto done;
}
uint64_t group, peer, id = 0;
memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, 8); memcpy(&peer, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
int control = entry->dgram[0] == ETCP_RT_ID_CALL_VIDEO_CONTROL;
@@ -558,6 +567,10 @@ int call_video_transport_bind(struct UTUN_INSTANCE *inst) {
if (etcp_router_bind(inst, ETCP_RT_ID_CALL_VIDEO, receive)) {
etcp_router_unbind(inst, ETCP_RT_ID_CALL_VIDEO_CONTROL); return -1;
}
if (etcp_router_require_crypto(inst, ETCP_RT_ID_CALL_VIDEO_CONTROL, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT)
|| etcp_router_require_crypto(inst, ETCP_RT_ID_CALL_VIDEO, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT)) {
etcp_router_unbind(inst, ETCP_RT_ID_CALL_VIDEO); etcp_router_unbind(inst, ETCP_RT_ID_CALL_VIDEO_CONTROL); return -1;
}
etcp_router_set_receive_ready(inst, ETCP_RT_ID_CALL_VIDEO, ready); return 0;
}
void call_video_transport_unbind(struct UTUN_INSTANCE *inst) {
src/chat/chat_event.h
+7 -2
View File
@@ -1,8 +1,10 @@
/*
* chat_event.h — система нотификаций о событиях чата
* chat_event — доставка событий ядра чата интерфейсу или headless-клиенту.
*
* Производители (chat_core, chat_sync, ...) вызывают chat_event_post().
* Потребители (chatgui, headless, будущий HTTP API) регистрируют один обработчик.
* На экземпляр регистрируется один обработчик через chat_event_set_handler().
* post вызывает его сразу, в том же uasync-потоке; очереди событий здесь нет.
* data заимствован только до возврата обработчика. Для передачи в другой поток нужна копия.
*
* Без обработчика — данные логируются через DEBUG_DEBUG + log_dump.
* Все вызовы — из uasync-потока.
@@ -92,8 +94,11 @@ extern "C" {
typedef void (*chat_event_handler_fn)(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len);
/* Заменить обработчик; NULL отключает доставку. Предыдущий обработчик не вызывается. */
void chat_event_set_handler(struct UTUN_INSTANCE* inst, chat_event_handler_fn handler);
/* Синхронно передать type и len байт data; память остаётся у отправителя. Формат зависит от CHAT_EVT_*. */
void chat_event_post(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len);
/* Сообщить этап подготовки вложения: state=0 начало, 1 готово, 2 ошибка; target — ID канала или DM. */
void chat_event_attachment(struct UTUN_INSTANCE* inst, int is_dm, const char* target, const uint8_t id[16], int state);
/* request_id=0 отключает подтверждение; вызывается один раз после сохранения либо ошибки подготовки. */
void chat_event_submitted(struct UTUN_INSTANCE* inst, uint64_t request_id, int success);
src/chat/chat_join.h
+1 -1
View File
@@ -63,7 +63,7 @@
* - Recovery восстанавливает сессии существующих участников и не использует
* invite-ключ как замену локальной таблицы участников.
*
* J <-> C: ETCP_RT_ID_CHAT_SYNC, заголовок [service:1][group_id:8][type:1], payload
* J <-> C: ETCP_ID_CHAT_SYNC, заголовок [service:1][group_id:8][type:1], payload
* описан в chat_sync.h. A <-> C: ETCP_RT_ID_JOIN через etcp_router с подписью
* и шифрованием; subcommands и payload описаны ниже.
*
src/chat/chat_sync.c
+3 -3
View File
@@ -188,7 +188,7 @@ static int cs_send(struct chat_sync* cs, const char* ch_id, uint64_t dst,
uint8_t* buf = u_malloc(total);
if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: cs_send alloc(%zu) failed", CS_ID, total); return -1; }
uint8_t* p = buf;
*p++ = ETCP_RT_ID_CHAT_SYNC;
*p++ = ETCP_ID_CHAT_SYNC;
cs_write_group_id(p, ch_id); p += 8;
memcpy(p, payload, len);
struct ll_entry* entry = queue_entry_new(0);
@@ -771,7 +771,7 @@ int chat_sync_init(struct UTUN_INSTANCE* inst) {
member_sync_destroy(inst); inst->chat_sync = NULL; u_free(cs); return -1;
}
etcp_bind(inst, ETCP_RT_ID_CHAT_SYNC, chat_sync_recv_cb);
etcp_bind(inst, ETCP_ID_CHAT_SYNC, chat_sync_recv_cb);
etcp_add_conn_status_cbk(inst, cs_on_conn_status, cs);
cs_refresh_channels(cs);
@@ -812,7 +812,7 @@ void chat_sync_destroy(struct UTUN_INSTANCE* inst) {
chat_join_destroy(inst);
cs->initialized = 0; inst->chat_sync = NULL;
etcp_unbind(inst, ETCP_RT_ID_CHAT_SYNC);
etcp_unbind(inst, ETCP_ID_CHAT_SYNC);
etcp_remove_conn_status_cbk(inst, cs_on_conn_status, cs);
if (cs->refresh_timer) { uasync_cancel_timeout(inst->ua, cs->refresh_timer); cs->refresh_timer = NULL; }
#ifdef UTUN_HAVE_STANDBY
src/chat/chat_sync.h
+1 -1
View File
@@ -20,7 +20,7 @@ extern "C" {
struct UTUN_INSTANCE;
struct UASYNC;
/* etcp service IDs (direct P2P): ETCP_RT_ID_CHAT_SYNC / ETCP_RT_ID_MEMBER_SYNC
/* etcp service IDs (direct P2P): ETCP_ID_CHAT_SYNC / ETCP_ID_MEMBER_SYNC
* (etcp_bind, см. etcp_api.h) */
/* Message types */
src/chat/db_sync.c
+4 -4
View File
@@ -769,14 +769,14 @@ static int db_record_insert_cascade(struct DB_SYNC_INSTANCE* si,
// Send
// ============================================================
// Отправляет сообщение узлу через ETCP с заголовком: [ETCP_RT_ID_DB_SYNC:1][group_id_be:8][payload]. Возвращает код etcp_send
// Отправляет сообщение узлу через ETCP с заголовком: [ETCP_ID_DB_SYNC:1][group_id_be:8][payload]. Возвращает код etcp_send
static int db_sync_send_gid(struct DB_SYNC* db, uint64_t node_id, uint64_t group_id, const uint8_t* payload, size_t plen)
{
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "queue_entry_new"); return -1; }
uint8_t* buf = u_malloc(plen + 9);
if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "db_sync_send_gid: u_malloc(%zu) failed for dst=%016llx", plen + 9, (unsigned long long)node_id); queue_entry_free(entry); return -1; }
buf[0] = ETCP_RT_ID_DB_SYNC;
buf[0] = ETCP_ID_DB_SYNC;
uint64_t gb = htobe64(group_id);
memcpy(buf + 1, &gb, 8);
memcpy(buf + 9, payload, plen);
@@ -1750,7 +1750,7 @@ int db_sync_enable(struct UTUN_INSTANCE* inst) {
}
db->enabled = 1;
etcp_bind(inst, ETCP_RT_ID_DB_SYNC, db_sync_recv_cb);
etcp_bind(inst, ETCP_ID_DB_SYNC, db_sync_recv_cb);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "db_sync: initialized (enabled=%d)", db->enabled);
return 0;
@@ -1764,7 +1764,7 @@ void db_sync_destroy(struct UTUN_INSTANCE* inst)
struct DB_SYNC* db = inst->db_sync;
inst->db_sync = NULL;
etcp_unbind(inst, ETCP_RT_ID_DB_SYNC);
etcp_unbind(inst, ETCP_ID_DB_SYNC);
if (db->peer_check_timer) { uasync_cancel_timeout(inst->ua, db->peer_check_timer); db->peer_check_timer = NULL; }
#ifdef UTUN_HAVE_STANDBY
src/chat/db_sync.h
+1 -1
View File
@@ -18,7 +18,7 @@ extern "C" {
struct UTUN_INSTANCE;
struct DB_SYNC_INSTANCE;
// etcp service ID: ETCP_RT_ID_DB_SYNC (etcp_bind, см. etcp_api.h)
// etcp service ID: ETCP_ID_DB_SYNC (etcp_bind, см. etcp_api.h)
// Message types
#define DB_MSG_INIT_SYNC 0x01
src/chat/db_sync_arch.md
+2 -2
View File
@@ -99,8 +99,8 @@ chain_hash[0] = SHA256(0x00..00 [32] || id_0 || ts_0 || author_0 || sig_0)
### 1.4 Интеграция с ETCP
- **Service ID:** `ETCP_RT_ID_DB_SYNC = 0x20`
- **Биндинг:** `etcp_bind(inst, 0x20, db_sync_recv_cb)` — приём сообщений
- **Service ID:** `ETCP_ID_DB_SYNC = 0x20`
- **Биндинг:** `etcp_bind(inst, ETCP_ID_DB_SYNC, db_sync_recv_cb)` — приём сообщений
- **Connection callbacks:** `etcp_add_conn_status_cbk(inst, db_sync_on_conn_status, NULL)` — отслеживание UP/DOWN соединений
- **Отправка:** через `etcp_send()` в прямое P2P-соединение
- **Маршрутизация:** каждый пакет начинается с `[svc:1][hash_be:8]`, hash идентифицирует инстанс на приёмной стороне
src/chat/member_sync.c
+1 -1
View File
@@ -1117,7 +1117,7 @@ void member_sync_subscribe_group(struct UTUN_INSTANCE* inst, struct TOPO_GROUP*
int member_sync_init(struct UTUN_INSTANCE* inst) {
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: init — inst is NULL", MS_ID); return -1; }
int rc = merkle_sync_init(inst, ETCP_RT_ID_MEMBER_SYNC, &g_member_ops, inst);
int rc = merkle_sync_init(inst, ETCP_ID_MEMBER_SYNC, &g_member_ops, inst);
if (rc != 0) return rc;
/* очистка битых записей во всех каналах перед построением дерева */
member_sync_verify_and_purge_all(inst);
src/chat/member_sync.h
+1 -1
View File
@@ -88,7 +88,7 @@ void member_sync_remove_apply_cbk(struct UTUN_INSTANCE* inst, member_apply_cbk_f
/*
* Инициализировать модуль: создаёт merkle_sync с коллбэками для мемберов
* (ETCP_RT_ID_MEMBER_SYNC), проверяет локальные записи, восстанавливает деревья
* (ETCP_ID_MEMBER_SYNC), проверяет локальные записи, восстанавливает деревья
* и подписывается на события CHAT-групп. 0/-1.
*/
int member_sync_init(struct UTUN_INSTANCE* inst);
src/media_delivery/media_download.c
+20 -11
View File
@@ -50,6 +50,10 @@ static int md_dl_block_index(struct media_download* dl, const uint8_t* block_id)
return -1;
}
static int md_dl_block_inflight(const struct media_download_block* block) {
return block->state == MD_BLK_CONN || block->state == MD_BLK_REQ || block->state == MD_BLK_RECV;
}
static void md_dl_reset_block_file(struct media_download* dl, int bi) {
char tmp[2048];
snprintf(tmp, sizeof(tmp), "%s.chunk_%d", dl->dest_path, bi);
@@ -60,7 +64,10 @@ static void md_dl_reset_block_file(struct media_download* dl, int bi) {
static struct media_download_peer* md_dl_get_peer(struct media_download* dl, uint64_t node_id) {
for (int pi = 0; pi < dl->num_peers; pi++)
if (dl->peers[pi].node_id == node_id) return &dl->peers[pi];
if (dl->num_peers >= MD_MAX_PEERS) return NULL;
if (dl->num_peers >= MD_MAX_PEERS) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: peer limit reached node=%016llx peers=%d", MDL_ID,
(unsigned long long)node_id, dl->num_peers); return NULL;
}
struct media_download_peer* p = &dl->peers[dl->num_peers++];
memset(p, 0, sizeof(*p));
p->node_id = node_id;
@@ -183,11 +190,11 @@ static int md_dl_request_block(struct media_download* dl, int bi) {
dl->inflight++;
struct UTUN_INSTANCE* inst = dl->inst;
if (instance_find_conn(inst, holder)) {
if ((p->cm_handle && p->connected) || instance_find_conn(inst, holder)) {
p->connected = 1;
b->state = MD_BLK_REQ;
md_dl_send_block_req(inst, holder, dl, bi);
} else {
} else if (!p->cm_handle) {
struct TOPO_GROUP* grp = inst->topo_groups ? topo_groups_find(inst->topo_groups, dl->group_id) : NULL;
if (grp && grp->conn_mgr) {
int rc = conn_mgr_open(inst, grp->group_id, holder, md_dl_conn_cb, dl, &p->cm_handle);
@@ -204,7 +211,8 @@ static int md_dl_request_block(struct media_download* dl, int bi) {
b->state = MD_BLK_REQ;
md_dl_send_block_req(inst, holder, dl, bi);
}
}
} else DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: waiting for existing CM request node=%016llx block=%d handle=%p",
MDL_ID, (unsigned long long)holder, bi, (void*)p->cm_handle);
return 0;
}
@@ -359,9 +367,11 @@ void md_dl_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64_t group_i
}
}
} else {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: conn FAIL node=0x%016llx event=%d", MDL_ID, (unsigned long long)node_id, event);
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: conn FAIL node=0x%016llx event=%d closing handle=%p",
MDL_ID, (unsigned long long)node_id, event, (void*)h);
struct media_download_peer* p = md_dl_get_peer(dl, node_id);
if (p) { p->connected = 0; p->cm_handle = NULL; }
conn_mgr_close(h); // DOWN/TIMEOUT не освобождают handle; отменяем в том числе поздний UP.
uint64_t now = get_time_tb();
int failed = 0;
for (int bi = 0; bi < dl->num_blocks; bi++) {
@@ -479,9 +489,9 @@ void media_download_handle_query_resp(struct UTUN_INSTANCE* inst,
dl->last_activity_tb = get_time_tb();
/* сбросить holders у незавершённых блоков (re-QUERY пересобирает список) */
/* QUERY меняет только ожидающие блоки: у активных сохраняем текущего держателя и его индекс. */
for (int bi = 0; bi < dl->num_blocks; bi++) {
if (dl->blocks[bi].state != MD_BLK_DONE) {
if (dl->blocks[bi].state != MD_BLK_DONE && !md_dl_block_inflight(&dl->blocks[bi])) {
dl->blocks[bi].num_holders = 0;
dl->blocks[bi].holder_idx = -1;
}
@@ -492,7 +502,7 @@ void media_download_handle_query_resp(struct UTUN_INSTANCE* inst,
int bi = md_dl_block_index(dl, entries[i].block_id);
if (bi < 0) continue;
struct media_download_block* b = &dl->blocks[bi];
if (b->state == MD_BLK_DONE) continue;
if (b->state == MD_BLK_DONE || md_dl_block_inflight(b)) continue;
int dup = 0;
for (int h = 0; h < b->num_holders; h++) if (b->holders[h] == entries[i].node_id) { dup = 1; break; }
if (!dup && b->num_holders < MD_MAX_HOLDERS_PER_BLOCK)
@@ -502,13 +512,12 @@ void media_download_handle_query_resp(struct UTUN_INSTANCE* inst,
/* fallback: блок без держателя — автор */
for (int bi = 0; bi < dl->num_blocks; bi++) {
struct media_download_block* b = &dl->blocks[bi];
if (b->state == MD_BLK_DONE) continue;
if (b->state == MD_BLK_DONE || md_dl_block_inflight(b)) continue;
if (b->num_holders == 0 && dl->author_node_id && dl->author_node_id != inst->node_id)
b->holders[b->num_holders++] = dl->author_node_id;
}
/* пересобрать peers (уникальные node_id всех держателей) */
dl->num_peers = 0;
/* Реестр peers живёт до finish: повторный QUERY не должен терять ожидающие CM handles. */
for (int bi = 0; bi < dl->num_blocks; bi++) {
struct media_download_block* b = &dl->blocks[bi];
for (int h = 0; h < b->num_holders; h++)
src/media_delivery/media_download.h
+1 -1
View File
@@ -51,7 +51,7 @@ struct media_download_block {
struct media_download_peer {
uint64_t node_id;
uint8_t connected;
struct CONN_MGR_HANDLE* cm_handle;
struct CONN_MGR_HANDLE* cm_handle; /* Один запрос на держателя; закрывается после DOWN/TIMEOUT или завершения загрузки. */
};
/* ── внутреннее состояние загрузки (доступно для тестов) ── */
src/routing_layer/etcp_router.c
+28 -4
View File
@@ -338,7 +338,7 @@ static int router_build_packet(struct ETCP_ROUTER_CONN* rconn, struct ll_entry*
uint8_t* base = u_calloc(1, len);
if (!base) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router build: allocation failed"); return -1; }
struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)base;
hdr->cmd = ETCP_RT_ID_SVC_ROUTE;
hdr->cmd = ETCP_ID_SVC_ROUTE;
hdr->group_id = rconn->group_id;
hdr->dst_node_id = rconn->remote_node_id;
hdr->src_node_id = rconn->inst->node_id;
@@ -364,7 +364,7 @@ static void router_control(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* con
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router control: allocation failed"); return; }
struct SVC_ROUTE_HDR* h = (struct SVC_ROUTE_HDR*)e->dgram;
memset(h, 0, sizeof(*h));
h->cmd = ETCP_RT_ID_SVC_ROUTE; h->group_id = rconn->group_id;
h->cmd = ETCP_ID_SVC_ROUTE; h->group_id = rconn->group_id;
h->src_node_id = rconn->inst->node_id; h->dst_node_id = rconn->remote_node_id; h->svc_id = rconn->svc_id;
h->flags = flags; h->reset_id = rconn->reset_id; h->peer_reset_id = peer_id; h->challenge = challenge;
e->len = SVC_ROUTE_HDR_SIZE;
@@ -805,6 +805,12 @@ static void router_retrans_timer_cb(void* arg) {
// Собрать+закодировать пакет в начале отправки и положить в send_q.
static int router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t len, int force, int mode) {
uint8_t required = rconn->inst->router_bindings.required_crypto[rconn->svc_id];
if ((mode & required) != required) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router send: insufficient crypto svc=%u peer=%016llx flags=%02x required=%02x",
rconn->svc_id, (unsigned long long)rconn->remote_node_id, mode, required);
return -1;
}
size_t overhead = SVC_ROUTE_HDR_SIZE + ROUTER_PATH_MAX_SIZE + ((mode & ROUTE_CRYPTO_SIGN) ? SC_SIGN_SIZE : 0)
+ ((mode & ROUTE_CRYPTO_ENCRYPT) ? SC_NONCE_SIZE + SC_CRC32_SIZE + SC_TAG_SIZE : 0);
if (rconn->closed || !payload || !len || len > PKTNORM_MAX_DGRAM_SIZE - overhead || (mode & ~(ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT))) {
@@ -941,7 +947,7 @@ static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) {
}
struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)u_malloc(SVC_ROUTE_HDR_SIZE);
if (!hdr) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_ack: u_malloc failed"); return; }
hdr->cmd = ETCP_RT_ID_SVC_ROUTE;
hdr->cmd = ETCP_ID_SVC_ROUTE;
hdr->group_id = rconn->group_id;
hdr->dst_node_id = rconn->remote_node_id;
hdr->src_node_id = inst->node_id;
@@ -1307,7 +1313,7 @@ static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETC
}
// ====================================================================
// etcp_router_recv_cb — обработчик ETCP_RT_ID_SVC_ROUTE
// etcp_router_recv_cb — обработчик ETCP_ID_SVC_ROUTE
// ====================================================================
// Входная точка приёма SVC_ROUTE-кодограмм: различает транзит, RST/CLOSE/ACK (пустой payload)
@@ -1329,6 +1335,12 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
router_forward_transit(inst, entry, (struct SVC_ROUTE_HDR*)entry->dgram); return;
}
entry->len = body;
uint8_t required = inst->router_bindings.required_crypto[h->svc_id];
if (body > SVC_ROUTE_HDR_SIZE && (h->flags & required) != required) {
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router receive: insufficient crypto svc=%u peer=%016llx flags=%02x required=%02x",
h->svc_id, (unsigned long long)h->src_node_id, h->flags, required);
free_entry(entry); return;
}
struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, h->group_id, h->src_node_id, h->svc_id);
if (h->flags & (ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED)) {
uint8_t* decoded = NULL; size_t len = 0; uint8_t flags = 0;
@@ -1429,6 +1441,7 @@ int etcp_router_bind(struct UTUN_INSTANCE* inst, uint8_t svc_id, etcp_recv_fn ca
if (!callback) {
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "unbind svc_id=%u", svc_id);
inst->router_bindings.callbacks[svc_id] = NULL;
inst->router_bindings.required_crypto[svc_id] = 0;
return 0;
}
if (inst->router_bindings.callbacks[svc_id])
@@ -1447,10 +1460,21 @@ int etcp_router_unbind(struct UTUN_INSTANCE* inst, uint8_t svc_id) {
}
inst->router_bindings.callbacks[svc_id] = NULL;
inst->router_bindings.receive_ready[svc_id] = NULL;
inst->router_bindings.required_crypto[svc_id] = 0;
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "svc_id=%u", svc_id);
return 0;
}
int etcp_router_require_crypto(struct UTUN_INSTANCE* inst, uint8_t svc_id, uint8_t flags) {
if (!inst || !inst->router_bindings.callbacks[svc_id] || (flags & ~(ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED))) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router crypto policy invalid inst=%p svc=%u flags=%02x", (void*)inst, svc_id, flags);
return -1;
}
inst->router_bindings.required_crypto[svc_id] = flags;
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router crypto policy svc=%u required=%02x", svc_id, flags);
return 0;
}
void etcp_router_set_receive_ready(struct UTUN_INSTANCE* inst, uint8_t svc_id,
int (*ready)(struct UTUN_INSTANCE*, uint64_t, uint64_t, const uint8_t*, size_t)) {
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router receive readiness: NULL instance"); return; }
src/routing_layer/etcp_router.h
+7 -1
View File
@@ -42,7 +42,7 @@ extern "C" {
#pragma pack(push, 1)
struct SVC_ROUTE_HDR {
uint8_t cmd; // ETCP_ID_SVC_ROUTE (0x03) / ETCP_RT_ID_SVC_ROUTE
uint8_t cmd; // ETCP_ID_SVC_ROUTE (0x03)
uint64_t group_id; // идентификатор группы
uint64_t dst_node_id;
uint64_t src_node_id;
@@ -234,6 +234,7 @@ struct ROUTER_INFLIGHT {
struct ETCP_ROUTER_BINDINGS {
etcp_recv_fn callbacks[SVC_ROUTE_MAX_BINDINGS];
int (*receive_ready[SVC_ROUTE_MAX_BINDINGS])(struct UTUN_INSTANCE*, uint64_t, uint64_t, const uint8_t*, size_t);
uint8_t required_crypto[SVC_ROUTE_MAX_BINDINGS]; // Обязательные ENCRYPTED/SIGNED для сервисных данных; 0 — без требования.
};
// ====================================================================
@@ -251,6 +252,11 @@ int etcp_router_bind(struct UTUN_INSTANCE* inst, uint8_t svc_id, etcp_recv_fn c
// Снять обработчик; само по себе не закрывает логические каналы сервиса.
int etcp_router_unbind(struct UTUN_INSTANCE* inst, uint8_t svc_id);
// Задать обязательные ROUTER_FLAG_ENCRYPTED/SIGNED для зарегистрированного сервиса, в обе стороны.
// Проверка входящих данных — до состояния/очередей/receive_ready, только на конечном узле; управляющие пакеты без payload исключены.
// Работает в uasync; unbind очищает требование. Возвращает 0 при успехе, -1 при неверном instance/сервисе/маске.
int etcp_router_require_crypto(struct UTUN_INSTANCE* inst, uint8_t svc_id, uint8_t flags);
// Проверяется до доставки/ACK. 0 оставляет пакет в recv_q, сохраняя порядок; CLOSE не блокируется.
// ready получает заимствованный проверенный payload и работает в uasync без I/O. NULL снимает ограничение.
void etcp_router_set_receive_ready(struct UTUN_INSTANCE* inst, uint8_t svc_id,
src/routing_layer/topo_group_invite.c
+5 -5
View File
@@ -103,7 +103,7 @@ void topo_group_invite_recv_handler(struct ETCP_CONN* conn, struct ll_entry* ent
return;
}
uint8_t* d = entry->dgram;
if (d[0] != ETCP_RT_ID_GROUP_INVITE) { queue_dgram_free(entry); queue_entry_free(entry); return; }
if (d[0] != ETCP_ID_GROUP_INVITE) { queue_dgram_free(entry); queue_entry_free(entry); return; }
uint8_t sub = d[1];
if (sub == TGI_SUBCMD_INFO_REQ) tgi_handle_info_req(conn, d, entry->len);
else if (sub == TGI_SUBCMD_INFO_RESP) tgi_handle_info_resp(conn, d, entry->len);
@@ -135,7 +135,7 @@ static void tgi_handle_info_req(struct ETCP_CONN* conn, const uint8_t* data, siz
size_t rs = TGI_INVITE_RESP_HDR_SIZE + nl;
uint8_t* rb = u_calloc(1, rs); if (!rb) return;
struct TGI_INVITE_RESP* resp = (struct TGI_INVITE_RESP*)rb;
resp->cmd = ETCP_RT_ID_GROUP_INVITE; resp->subcmd = TGI_SUBCMD_INFO_RESP;
resp->cmd = ETCP_ID_GROUP_INVITE; resp->subcmd = TGI_SUBCMD_INFO_RESP;
resp->group_id = req->group_id; resp->is_member = 0;
memcpy(resp->ed25519_pubkey, inst->my_ed25519_pubkey, SC_PUBKEY_SIZE);
resp->node_name_len = (uint8_t)nl; if (nl) memcpy(resp->node_name, nm, nl);
@@ -204,7 +204,7 @@ static void tgi_handle_info_req(struct ETCP_CONN* conn, const uint8_t* data, siz
memset(rb, 0, rs);
{ struct TGI_INVITE_RESP* resp = (struct TGI_INVITE_RESP*)rb;
resp->cmd = ETCP_RT_ID_GROUP_INVITE; resp->subcmd = TGI_SUBCMD_INFO_RESP;
resp->cmd = ETCP_ID_GROUP_INVITE; resp->subcmd = TGI_SUBCMD_INFO_RESP;
resp->group_id = req->group_id; resp->is_member = 1;
memcpy(resp->ed25519_pubkey, inst->my_ed25519_pubkey, SC_PUBKEY_SIZE);
resp->node_name_len = (uint8_t)nl; if (nl) memcpy(resp->node_name, nm, nl); }
@@ -372,7 +372,7 @@ static void tgi_ncd_cb(struct NODE_CONN_DIRECT* h, enum ncd_event ev, void* arg)
p->state = TGI_WAIT_INFO;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "invite: connected to 0x%016llx — sending membership check", (unsigned long long)p->node_id);
struct TGI_INVITE_REQ req;
req.cmd = ETCP_RT_ID_GROUP_INVITE; req.subcmd = TGI_SUBCMD_INFO_REQ; req.group_id = p->group_id;
req.cmd = ETCP_ID_GROUP_INVITE; req.subcmd = TGI_SUBCMD_INFO_REQ; req.group_id = p->group_id;
struct ll_entry* qe = queue_entry_new(0);
if (!qe) { tgi_fail(p); return; }
qe->dgram = u_malloc(sizeof(req));
@@ -431,7 +431,7 @@ void topo_group_invite_cancel_all(struct UTUN_INSTANCE* inst) {
void topo_group_invite_init(struct UTUN_INSTANCE* inst) {
if (!inst) return;
etcp_bind(inst, ETCP_RT_ID_GROUP_INVITE, topo_group_invite_recv_handler);
etcp_bind(inst, ETCP_ID_GROUP_INVITE, topo_group_invite_recv_handler);
}
/* ═══════════ входная точка джойнера ═══════════ */
src/routing_layer/topo_group_invite.h
+2 -2
View File
@@ -19,7 +19,7 @@ struct ETCP_CONN;
struct ll_entry;
struct NODE_CONN_DIRECT;
/* service id протокола invite: ETCP_RT_ID_GROUP_INVITE (etcp_bind, см. etcp_api.h) */
/* service id протокола invite: ETCP_ID_GROUP_INVITE (etcp_bind, см. etcp_api.h) */
#define TGI_SUBCMD_INFO_REQ 0x01
#define TGI_SUBCMD_INFO_RESP 0x02
@@ -30,7 +30,7 @@ struct NODE_CONN_DIRECT;
typedef void (*tgi_cb_t)(uint64_t node_id, uint64_t group_id, int event, void* arg);
/** Идемпотентная инициализация: биндит обработчик ETCP_RT_ID_GROUP_INVITE. */
/** Идемпотентная инициализация: биндит обработчик ETCP_ID_GROUP_INVITE. */
void topo_group_invite_init(struct UTUN_INSTANCE* inst);
/**
src/transport_layer/etcp_api.c
+2 -2
View File
@@ -195,9 +195,9 @@ void etcp_int_recv(struct ll_queue* queue, void* arg) {
if (inst->api_bindings.callbacks[id]) {
DEBUG_TRACE(DEBUG_CATEGORY_TRAFFIC, "dispatch id=0x%02x conn=%s len=%zu", id, conn->log_name, e->len);
inst->api_bindings.callbacks[id](conn, e);
} else if (inst->api_bindings.callbacks[0]) {
} else if (inst->api_bindings.callbacks[ETCP_ID_DATA]) {
DEBUG_TRACE(DEBUG_CATEGORY_TRAFFIC, "dispatch id=0x%02x (fallback) conn=%s len=%zu", id, conn->log_name, e->len);
inst->api_bindings.callbacks[0](conn, e);
inst->api_bindings.callbacks[ETCP_ID_DATA](conn, e);
} else { queue_dgram_free(e); queue_entry_free(e); }
queue_resume_callback(queue);
}
src/transport_layer/etcp_api.h
+19 -37
View File
@@ -28,72 +28,54 @@ extern "C" {
// • ETCP_ID_* — raw ETCP: etcp_bind() / etcp_send()
// • ETCP_RT_ID_* — router-сервисы: etcp_router_bind() / etcp_route_send()
//
// Одинаковое значение в РАЗНЫХ пространствах — НЕ коллизия (напр. 0x00 =
// ETCP_ID_DATA и ETCP_RT_ID_DATA). Коллизия — дубль значения В ПРЕДЕЛАХ одного
// пространства: последний bind затирает предыдущий обработчик и молча ломает
// сервис (реальный случай: ETCP_RT_ID_CHAT 0x11 затирал ETCP_RT_ID_CONN_MGR
// 0x11 → переставал работать conn_mgr).
//
// Часть констант ETCP_RT_ID_* исторически биндится через etcp_bind()
// (raw-пространство) — помечены «(raw)»; их значения занимают raw-реестр.
//
// raw: 0x00 DATA, 0x01 TOPO_ENTRY, 0x02 NAT_DETECTION, 0x03 SVC_ROUTE,
// 0x04 NTP_TIME, 0x05 BROADCAST, 0x06 RADIO, 0x07 STREAM, 0x08 MULTICAST, 0x12 NCD_CONTROL,
// 0x13 GROUP_INVITE, 0x20 DB_SYNC, 0x30 CHAT_SYNC, 0x31 MEMBER_SYNC
// router: 0x00 DATA, 0x02 NAT, 0x04 TCP_PROXY_SERVER, 0x05 UDP_PROXY,
// 0x06 ICMP_PROXY, 0x07 MEDIA_DELIVERY, 0x08 TCP_PROXY_CLIENT,
// 0x11 CONN_MGR, 0x32 ADMIN_KEY, 0x33 JOIN, 0x34 DM,
// 0x35 DM_MAILBOX, 0x36 CALL, 0x37 FILE_TRANSFER, 0x38 DM_MEDIA, 0x39 MEMBER_FILES,
// 0x3A GROUP_AVATAR, 0x3B NAT_DETECTION, 0x3C STREAM,
// 0x3D CALL_VIDEO_CONTROL, 0x3E CALL_VIDEO, 0x3F CALL_AUDIO, 0x40 RADIO_CALL, 0x41 MULTICAST_REPORT
//
// !!! Новый ID: впиши его в нужный реестр выше и убедись, что значение
// не занято в том же пространстве.
// Значения 0..255 уникальны внутри своей таблицы. Между таблицами совпадения
// допустимы: например, ETCP_ID_DATA и ETCP_RT_ID_DATA имеют значение 0x00.
// Префикс соответствует таблице без псевдонимов и исключений.
// tests/test_service_ids.py проверяет объявления, уникальность и использование
// при make check. Подкоманды отдельных протоколов объявляются в их заголовках.
// ═══════════════════════════════════════════════════════════════════════════
// === Raw ETCP packet IDs (etcp_bind/etcp_send) ===
#define ETCP_ID_DATA 0x00 // пакет для передачи адресату
#define ETCP_ID_DATA 0x00 // обработчик прямых данных и fallback для неизвестного cmd
#define ETCP_ID_TOPO_ENTRY 0x01 // BGP-топология (topo_group)
#define ETCP_ID_NAT_DETECTION 0x02 // NAT-детекция (STUN-like ping)
#define ETCP_ID_SVC_ROUTE 0x03 // транспорт роутера (etcp_router)
#define ETCP_ID_NTP_TIME 0x04 // NTP time sync P2P
#define ETCP_ID_BROADCAST 0x05 // broadcast-сообщения между пирами
#define ETCP_ID_RADIO 0x06 // рация (PTT мультикаст голоса) между пирами
#define ETCP_ID_STREAM 0x07 // прямой stream INFO/выдачи/DATA
#define ETCP_ID_MULTICAST 0x08 // адресный мультикаст с разделением по маршрутам группы
#define ETCP_RT_ID_MULTICAST_REPORT 0x41 // отчёты о переполнении multicast FIFO
#define ETCP_RT_ID_STREAM 0x3C // stream REVERSE_REQ/CLOSED и помощь суперузлов через группу
#define ETCP_ID_NCD_CONTROL 0x12 // node_conn_direct CLOSE/KEEP_ALIVE
#define ETCP_ID_GROUP_INVITE 0x13 // topo_group_invite
#define ETCP_ID_DB_SYNC 0x20 // db_sync
#define ETCP_ID_CHAT_SYNC 0x30 // chat_sync
#define ETCP_ID_MEMBER_SYNC 0x31 // member_sync/merkle_sync
// === Router-сервисы (etcp_router_bind/etcp_route_send) ===
#define ETCP_RT_ID_DATA 0x00 // routing.c — маршрутизация данных
#define ETCP_RT_ID_NAT 0x02 // NAT-трафик между узлами
#define ETCP_RT_ID_SVC_ROUTE 0x03 // транспорт роутера (raw, = ETCP_ID_SVC_ROUTE)
#define ETCP_RT_ID_TCP_PROXY_SERVER 0x04 // TCP proxy server (CONNECT/DATA от клиентов)
#define ETCP_RT_ID_UDP_PROXY 0x05 // UDP datagram прокси (client ↔ exit)
#define ETCP_RT_ID_ICMP_PROXY 0x06 // ICMP echo прокси (ping через exit)
#define ETCP_RT_ID_MEDIA_DELIVERY 0x07 // распространение медиа (аудио/видео)
#define ETCP_RT_ID_TCP_PROXY_CLIENT 0x08 // TCP proxy client (DATA/CLOSE/ERROR/FIN)
#define ETCP_RT_ID_CONN_MGR 0x11 // Connection Manager — management connections
#define ETCP_RT_ID_NCD_CONTROL 0x12 // node_conn_direct CLOSE/KEEP_ALIVE (raw)
#define ETCP_RT_ID_GROUP_INVITE 0x13 // topo_group_invite (raw)
#define ETCP_RT_ID_DB_SYNC 0x20 // db_sync (raw)
#define ETCP_RT_ID_CHAT_SYNC 0x30 // chat_sync (raw)
#define ETCP_RT_ID_MEMBER_SYNC 0x31 // member_sync/merkle_sync (raw)
#define ETCP_RT_ID_ADMIN_KEY 0x32 // chat_admin — канальный приватный ключ
#define ETCP_RT_ID_JOIN 0x33 // chat_join — join-протокол
#define ETCP_RT_ID_DM 0x34 // dm_core — прямой p2p чат
#define ETCP_RT_ID_DM_MAILBOX 0x35 // dm_mailbox — offline-хранение
#define ETCP_RT_ID_CALL 0x36 // call — P2P аудио-звонок
#define ETCP_RT_ID_CALL_VIDEO_CONTROL 0x3D // возможности и восстановление видеотрека
#define ETCP_RT_ID_CALL_VIDEO 0x3E // видео, отдельный ACK/debt/backpressure
#define ETCP_RT_ID_CALL_AUDIO 0x3F // аудио, независимо от сигналинга/видео
#define ETCP_RT_ID_RADIO_CALL 0x40 // адресное приглашение участника в рацию CHAT-канала
#define ETCP_RT_ID_FILE_TRANSFER 0x37 // file_transfer — передача неизменяемых файлов
#define ETCP_RT_ID_DM_MEDIA 0x38 // dm_media — custody, location и подтверждения медиа
#define ETCP_RT_ID_MEMBER_FILES 0x39 // member_files — каталоги и файлы участников группы
#define ETCP_RT_ID_GROUP_AVATAR 0x3A // подписанная аватарка группы, блоки до 4 КиБ
#define ETCP_RT_ID_NAT_DETECTION 0x3B // управление NAT-пробами через маршруты выбранной группы
#define ETCP_RT_ID_ATTACHMENT_KEYS 0x40 // запросы и ручная E2E-выдача ключей групповых вложений
#define ETCP_RT_ID_STREAM 0x3C // stream REVERSE_REQ/CLOSED и помощь суперузлов через группу
#define ETCP_RT_ID_CALL_VIDEO_CONTROL 0x3D // возможности и восстановление видеотрека
#define ETCP_RT_ID_CALL_VIDEO 0x3E // видео, отдельный ACK/debt/backpressure
#define ETCP_RT_ID_CALL_AUDIO 0x3F // аудио, независимо от сигналинга/видео
#define ETCP_RT_ID_RADIO_CALL 0x40 // адресное приглашение участника в рацию CHAT-канала
#define ETCP_RT_ID_MULTICAST_REPORT 0x41 // отчёты о переполнении multicast FIFO
#define ETCP_RT_ID_ATTACHMENT_KEYS 0x42 // запросы и ручная E2E-выдача ключей групповых вложений
// Connection status events (instance-level callback).
// NEW начинает жизнь объекта, UP/DOWN отражают доступность, DELETE завершает её.
@@ -323,7 +305,7 @@ int etcp_send(struct ETCP_CONN* conn, struct ll_entry* entry);
* @param callback Коллбэк для обработки пакетов с этим ID
* @return 0 при успехе, -1 при ошибке
*
* @note ID=0 зарезервирован для обычных пакетов данных
* @note ETCP_ID_DATA также получает пакеты без отдельного обработчика
* @note Можно зарегистрировать только один коллбэк на каждый ID в пределах instance
*/
int etcp_bind(struct UTUN_INSTANCE* inst, uint8_t id, etcp_recv_fn callback);
src/transport_layer/etcp_api_doc.md
+7 -10
View File
@@ -54,17 +54,14 @@ etcp_set_new_conn_cbk(instance, on_new_conn, my_data);
**Важно:** коллбэки `set_*` заменяют всю цепочку одним обработчиком; `add_*`/`remove_*` добавляют/убирают обработчик в связном списке из нескольких коллбэков.
### Сервисные ID (для etcp_router)
### Сервисные ID
Используются `ETCP_RT_ID_*`, отличаются от сырых `ETCP_ID_*` (пакет оборачивается в транспортный ID `ETCP_ID_SVC_ROUTE=0x03`):
- `ETCP_RT_ID_DATA` (0x00) — маршрутизация данных
- `ETCP_RT_ID_NAT` (0x02) — NAT-трафик между узлами
- `ETCP_RT_ID_SVC_ROUTE` (0x03) — транспорт роутера
- `ETCP_RT_ID_TCP_PROXY` (0x04) — TCP-прокси
- `ETCP_RT_ID_UDP_PROXY` (0x05) — UDP-прокси
- `ETCP_RT_ID_ICMP_PROXY` (0x06) — ICMP-прокси
- `ETCP_RT_ID_CONN_MGR` (0x11) — управление соединениями
- `ETCP_ID_NTP_TIME` (0x04) — синхронизация времени (raw ETCP, P2P)
Полный реестр с назначением каждого ID находится в [etcp_api.h](etcp_api.h).
`ETCP_ID_*` используются для прямых кодограмм через `etcp_bind` / `etcp_send`;
`ETCP_RT_ID_*` — для сервисов через `etcp_router_bind` / `etcp_route_send`.
Роутер оборачивает сервисные данные в прямую кодограмму `ETCP_ID_SVC_ROUTE`.
Значения уникальны внутри каждой таблицы; совпадения между двумя таблицами допустимы.
Проверка реестра и регистраций выполняется автоматически при `make check`.
### Фоновые соединения (etcp_connect)
src/transport_layer/etcp_connections.c
+8 -24
View File
@@ -283,15 +283,13 @@ static void etcp_link_send_init(struct ETCP_LINK* link, uint8_t reset, uint8_t c
uint8_t obfuscated_pubkey[SC_PUBKEY_SIZE];
sc_obfuscate_pubkey(salt, link->etcp->crypto_ctx.peer_public_key,
link->etcp->instance->my_keys.public_key, obfuscated_pubkey);
DEBUG_DEBUG(DEBUG_CATEGORY_CRYPTO, "INIT_SEND: salt=%02x%02x%02x%02x%02x%02x%02x%02x obf_pub=%02x%02x%02x%02x peer_pub=%02x%02x%02x%02x my_pub=%02x%02x%02x%02x seskey=%02x%02x%02x%02x",
DEBUG_DEBUG(DEBUG_CATEGORY_CRYPTO, "INIT_SEND: salt=%02x%02x%02x%02x%02x%02x%02x%02x obf_pub=%02x%02x%02x%02x peer_pub=%02x%02x%02x%02x my_pub=%02x%02x%02x%02x",
salt[0], salt[1], salt[2], salt[3], salt[4], salt[5], salt[6], salt[7],
obfuscated_pubkey[0], obfuscated_pubkey[1], obfuscated_pubkey[2], obfuscated_pubkey[3],
link->etcp->crypto_ctx.peer_public_key[0], link->etcp->crypto_ctx.peer_public_key[1],
link->etcp->crypto_ctx.peer_public_key[2], link->etcp->crypto_ctx.peer_public_key[3],
link->etcp->instance->my_keys.public_key[0], link->etcp->instance->my_keys.public_key[1],
link->etcp->instance->my_keys.public_key[2], link->etcp->instance->my_keys.public_key[3],
link->etcp->crypto_ctx.session_key[0], link->etcp->crypto_ctx.session_key[1],
link->etcp->crypto_ctx.session_key[2], link->etcp->crypto_ctx.session_key[3]);
link->etcp->instance->my_keys.public_key[2], link->etcp->instance->my_keys.public_key[3]);
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wstringop-overflow"
memcpy(dgram->data + offset, obfuscated_pubkey, SC_PUBKEY_SIZE);
@@ -1464,7 +1462,6 @@ int etcp_encrypt_send(struct ETCP_DGRAM* dgram) {
dgram->timestamp=get_current_timestamp();
dgram->link->total_encrypted += dgram->data_len;
if (debug_should_output(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_CRYPTO)) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_CRYPTO, "Before encryption", dgram->data, dgram->data_len);
uint8_t plain[PACKET_DATA_SIZE + ETCP_SESSION_HEADER_SIZE + 3];
size_t plain_len = etcp_encode_plain(dgram, plain);
sc_encrypt(sc, plain, plain_len, enc_buf, &enc_buf_len);
@@ -2199,9 +2196,7 @@ static void etcp_process_packet(struct ETCP_SOCKET* e_sock, uint8_t* data,
// }
if (link!=NULL && link->etcp!=NULL && link->etcp->crypto_ctx.session_ready) {
DEBUG_TRACE(DEBUG_CATEGORY_CRYPTO, "NORM_DECRYPT_TRY: seskey=%02x%02x%02x%02x recv_len=%zd rx=%llu",
link->etcp->crypto_ctx.session_key[0], link->etcp->crypto_ctx.session_key[1],
link->etcp->crypto_ctx.session_key[2], link->etcp->crypto_ctx.session_key[3],
DEBUG_TRACE(DEBUG_CATEGORY_CRYPTO, "NORM_DECRYPT_TRY: recv_len=%zd rx=%llu",
recv_len, (unsigned long long)link->etcp->crypto_ctx.rx_counter);
sc_status_t dec_rc = sc_decrypt(&link->etcp->crypto_ctx, data, recv_len, (uint8_t*)&pkt->timestamp, &pkt_len);
if (!dec_rc) {
@@ -2209,12 +2204,10 @@ static void etcp_process_packet(struct ETCP_SOCKET* e_sock, uint8_t* data,
if (ec) { errorcode = ec; goto ec_fr; }
return;
}
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "etcp: DECRYPT FAIL on existing link. rc=%d link=%p log=%s from=%s sess=%d link_state=%d keepalive=%d enc_errs=%zu my_pub=%016llx peer_pub=%016llx seskey=%02x%02x%02x%02x",
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "etcp: DECRYPT FAIL on existing link. rc=%d link=%p log=%s from=%s sess=%d link_state=%d keepalive=%d enc_errs=%zu my_pub=%016llx peer_pub=%016llx",
dec_rc, link, link->etcp->log_name, sockaddr_storage_to_str(&addr).str, link->etcp->crypto_ctx.session_ready,
link->link_state, link->recv_keepalive, link->encrypt_errors,
(unsigned long long)wire_read64(link->etcp->crypto_ctx.pk->public_key), (unsigned long long)wire_read64(link->etcp->crypto_ctx.peer_public_key),
link->etcp->crypto_ctx.session_key[0], link->etcp->crypto_ctx.session_key[1],
link->etcp->crypto_ctx.session_key[2], link->etcp->crypto_ctx.session_key[3]);
(unsigned long long)wire_read64(link->etcp->crypto_ctx.pk->public_key), (unsigned long long)wire_read64(link->etcp->crypto_ctx.peer_public_key));
} else {
DEBUG_DEBUG(DEBUG_CATEGORY_CRYPTO, "SKIP normal decrypt: link=%p session_ready=%d — trying init decrypt",
link, link && link->etcp ? link->etcp->crypto_ctx.session_ready : -1);
@@ -2244,8 +2237,7 @@ static void etcp_process_packet(struct ETCP_SOCKET* e_sock, uint8_t* data,
goto ec_fr;
}
DEBUG_DEBUG(DEBUG_CATEGORY_CRYPTO, "X25519 decrypt OK from %s", sockaddr_storage_to_str(&addr).str);
DEBUG_DEBUG(DEBUG_CATEGORY_CRYPTO, "INIT_DECRYPT_TRY: seskey=%02x%02x%02x%02x recv_len=%zd strip40=%zd salt=%02x%02x%02x%02x%02x%02x%02x%02x enc_pub=%02x%02x%02x%02x",
sc.session_key[0], sc.session_key[1], sc.session_key[2], sc.session_key[3],
DEBUG_DEBUG(DEBUG_CATEGORY_CRYPTO, "INIT_DECRYPT_TRY: recv_len=%zd strip40=%zd salt=%02x%02x%02x%02x%02x%02x%02x%02x enc_pub=%02x%02x%02x%02x",
recv_len, recv_len - SC_PUBKEY_ENC_SIZE,
salt[0], salt[1], salt[2], salt[3], salt[4], salt[5], salt[6], salt[7],
encrypted_pubkey[0], encrypted_pubkey[1], encrypted_pubkey[2], encrypted_pubkey[3]);
@@ -2386,16 +2378,8 @@ static void etcp_process_packet(struct ETCP_SOCKET* e_sock, uint8_t* data,
conn=etcp_connection_create(e_sock->instance,"");
if (!conn) { errorcode=55; DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "failed to create connection"); goto ec_fr; }
memcpy(&conn->crypto_ctx, &sc, sizeof(sc));
DEBUG_DEBUG(DEBUG_CATEGORY_CRYPTO, "CRYPTO_CTX_INIT: log=%s seskey=%02x%02x%02x%02x peer_pub=%02x%02x%02x%02x my_priv=%02x%02x%02x%02x",
conn->log_name,
conn->crypto_ctx.session_key[0], conn->crypto_ctx.session_key[1],
conn->crypto_ctx.session_key[2], conn->crypto_ctx.session_key[3],
conn->crypto_ctx.peer_public_key[0], conn->crypto_ctx.peer_public_key[1],
conn->crypto_ctx.peer_public_key[2], conn->crypto_ctx.peer_public_key[3],
conn->crypto_ctx.pk ? conn->crypto_ctx.pk->public_key[0] : 0,
conn->crypto_ctx.pk ? conn->crypto_ctx.pk->public_key[1] : 0,
conn->crypto_ctx.pk ? conn->crypto_ctx.pk->public_key[2] : 0,
conn->crypto_ctx.pk ? conn->crypto_ctx.pk->public_key[3] : 0);
DEBUG_DEBUG(DEBUG_CATEGORY_CRYPTO, "CRYPTO_CTX_INIT: peer=%016llx session_ready=%d",
(unsigned long long)peer_id, conn->crypto_ctx.session_ready);
conn->peer_node_id = peer_id;
etcp_update_log_name(conn);
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "New connection received on socket %s: log_name=%s peer_id=%lu peer:%s", e_sock->name, conn->log_name, (unsigned long)peer_id, sockaddr_storage_to_str(&addr).str);
src/transport_layer/node_conn_direct.c
+4 -4
View File
@@ -685,7 +685,7 @@ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e)
if (entry->handle_count > 0) {
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] refusing CLOSE, still have %d handles for 0x%016llx",
entry->handle_count, (unsigned long long)sender_id);
{ struct ncd_control_msg resp = { ETCP_RT_ID_NCD_CONTROL, NCD_SUBCMD_KEEP_ALIVE, conn->instance->node_id };
{ struct ncd_control_msg resp = { ETCP_ID_NCD_CONTROL, NCD_SUBCMD_KEEP_ALIVE, conn->instance->node_id };
struct ll_entry* qe = ll_alloc_lldgram(sizeof(resp));
if (!qe) DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE allocation failed");
else {
@@ -726,11 +726,11 @@ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e)
int node_conn_direct_init(struct UTUN_INSTANCE* inst) {
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] init without instance"); return -1; }
if (inst->ncd_control_bound) return 0;
if (etcp_bind(inst, ETCP_RT_ID_NCD_CONTROL, ncd_recv_control_handler) < 0) {
if (etcp_bind(inst, ETCP_ID_NCD_CONTROL, ncd_recv_control_handler) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot bind CLOSE/KEEP_ALIVE handler"); return -1;
}
inst->ncd_control_bound = 1;
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] bound ETCP_RT_ID_NCD_CONTROL handler");
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] bound ETCP_ID_NCD_CONTROL handler");
return 0;
}
@@ -1159,7 +1159,7 @@ static void ncd_handle_close(struct NODE_CONN_DIRECT* h, int force) {
}
if (entry->connect_timer) { uasync_cancel_timeout(entry->ua, entry->connect_timer); entry->connect_timer = NULL; }
conn->fin_wait = 1; conn->fin_wait_clear_cb = ncd_fin_wait_cancelled; conn->fin_wait_clear_arg = entry;
struct ncd_control_msg msg = { ETCP_RT_ID_NCD_CONTROL, NCD_SUBCMD_CLOSE, entry->inst->node_id };
struct ncd_control_msg msg = { ETCP_ID_NCD_CONTROL, NCD_SUBCMD_CLOSE, entry->inst->node_id };
struct ll_entry* e = ll_alloc_lldgram(sizeof(msg));
if (!e) DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] CLOSE allocation failed");
else {
src/transport_layer/node_conn_direct.h
+2 -2
View File
@@ -114,14 +114,14 @@ uint64_t node_conn_direct_node_id(const struct NODE_CONN_DIRECT* h);
/* Заимствованный транспорт, в том числе до UP; NULL после CLOSED. Не закрывать напрямую. */
struct ETCP_CONN* node_conn_direct_get_conn(struct NODE_CONN_DIRECT* h);
/* ─── Протокол CLOSE / KEEP_ALIVE (ETCP_RT_ID_NCD_CONTROL = 0x12) ─── */
/* ─── Протокол CLOSE / KEEP_ALIVE (ETCP_ID_NCD_CONTROL = 0x12) ─── */
#define NCD_SUBCMD_CLOSE 0x01
#define NCD_SUBCMD_KEEP_ALIVE 0x02
#pragma pack(push, 1)
struct ncd_control_msg {
uint8_t cmd; /* ETCP_RT_ID_NCD_CONTROL */
uint8_t cmd; /* ETCP_ID_NCD_CONTROL */
uint8_t subcmd; /* NCD_SUBCMD_CLOSE / KEEP_ALIVE */
uint64_t node_id; /* node_id отправителя */
};
src/transport_layer/secure_channel.c
+10 -48
View File
@@ -21,18 +21,9 @@
#include <openssl/evp.h>
#include <openssl/sha.h>
#include <openssl/err.h>
#include <openssl/rand.h>
#include "../lib/platform_compat.h"
static uint8_t sc_urandom_seed[8] = {0};
static int sc_urandom_initialized = 0;
static void sc_init_random_seed(void)
{
if (random_bytes(sc_urandom_seed, 8) == 0) {
sc_urandom_initialized = 1;
}
}
// Конвертация hex строки в бинарный формат
int sc_hex_to_binary(const char *hex_str, uint8_t *binary, size_t binary_len) {
if (!hex_str || !binary || strlen(hex_str) != binary_len * 2) return -1;
@@ -225,12 +216,7 @@ sc_status_t sc_set_peer_public_key(sc_context_t *ctx, const uint8_t *peer_public
}
sc_derive_session_key(shared_secret, ctx->session_key);
DEBUG_DEBUG(DEBUG_CATEGORY_CRYPTO, "ECDH: priv=%02x%02x%02x%02x peer_pub=%02x%02x%02x%02x shared=%02x%02x%02x%02x%02x%02x%02x%02x seskey=%02x%02x%02x%02x",
ctx->pk->private_key[0], ctx->pk->private_key[1], ctx->pk->private_key[2], ctx->pk->private_key[3],
peer_public_key[0], peer_public_key[1], peer_public_key[2], peer_public_key[3],
shared_secret[0], shared_secret[1], shared_secret[2], shared_secret[3],
shared_secret[4], shared_secret[5], shared_secret[6], shared_secret[7],
ctx->session_key[0], ctx->session_key[1], ctx->session_key[2], ctx->session_key[3]);
OPENSSL_cleanse(shared_secret, sizeof(shared_secret));
memcpy(ctx->peer_public_key, peer_public_key, SC_PUBKEY_SIZE);
ctx->peer_key_set = 1;
ctx->session_ready = 1;
@@ -242,29 +228,8 @@ sc_status_t sc_set_peer_public_key(sc_context_t *ctx, const uint8_t *peer_public
return SC_OK;
}
static void sc_build_nonce(uint64_t counter, uint8_t *nonce_out) {
SHA256_CTX sha_ctx;
uint8_t hash[32];
struct timeval tv;
uint8_t data[24];
if (!sc_urandom_initialized) sc_init_random_seed();
utun_gettimeofday(&tv, NULL);
memcpy(data, sc_urandom_seed, 8);
data[8] = (counter >> 0) & 0xFF; data[9] = (counter >> 8) & 0xFF;
data[10] = (counter >> 16) & 0xFF; data[11] = (counter >> 24) & 0xFF;
data[12] = (counter >> 32) & 0xFF; data[13] = (counter >> 40) & 0xFF;
data[14] = (counter >> 48) & 0xFF; data[15] = (counter >> 56) & 0xFF;
data[16] = (tv.tv_sec >> 0) & 0xFF; data[17] = (tv.tv_sec >> 8) & 0xFF;
data[18] = (tv.tv_sec >> 16) & 0xFF; data[19] = (tv.tv_sec >> 24) & 0xFF;
data[20] = (tv.tv_usec >> 0) & 0xFF; data[21] = (tv.tv_usec >> 8) & 0xFF;
data[22] = (tv.tv_usec >> 16) & 0xFF; data[23] = (tv.tv_usec >> 24) & 0xFF;
SHA256_Init(&sha_ctx);
SHA256_Update(&sha_ctx, data, 24);
SHA256_Final(hash, &sha_ctx);
memcpy(nonce_out, hash, SC_NONCE_SIZE);
}
sc_status_t sc_encrypt(sc_context_t *ctx, const uint8_t *plaintext, size_t plaintext_len, uint8_t *ciphertext, size_t *ciphertext_len) {
if (ciphertext_len) *ciphertext_len = 0;
sc_status_t status = validate_encrypt_inputs(ctx, plaintext, ciphertext, ciphertext_len, plaintext_len);
if (status != SC_OK) return status;
@@ -272,12 +237,12 @@ sc_status_t sc_encrypt(sc_context_t *ctx, const uint8_t *plaintext, size_t plain
append_crc32(plaintext, plaintext_len, plaintext_with_crc);
size_t total_plaintext_len = plaintext_len + SC_CRC32_SIZE;
uint8_t nonce[SC_NONCE_SIZE];
sc_build_nonce(ctx->tx_counter, nonce);
DEBUG_TRACE(DEBUG_CATEGORY_CRYPTO, "ENCRYPT: seskey=%02x%02x%02x%02x tx=%llu nonce=%02x%02x%02x%02x data[0..3]=%02x%02x%02x%02x",
ctx->session_key[0], ctx->session_key[1], ctx->session_key[2], ctx->session_key[3],
(unsigned long long)ctx->tx_counter,
nonce[0], nonce[1], nonce[2], nonce[3],
plaintext_with_crc[0], plaintext_with_crc[1], plaintext_with_crc[2], plaintext_with_crc[3]);
// Случайный nonce не зависит от часов, сбросов контекста и параллельных экземпляров ядра.
if (RAND_bytes(nonce, sizeof(nonce)) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "CCM nonce generation failed tx=%llu", (unsigned long long)ctx->tx_counter);
return SC_ERR_CRYPTO;
}
DEBUG_TRACE(DEBUG_CATEGORY_CRYPTO, "ENCRYPT: tx=%llu bytes=%zu", (unsigned long long)ctx->tx_counter, plaintext_len);
EVP_CIPHER_CTX *ectx = EVP_CIPHER_CTX_new();
if (!ectx) return SC_ERR_CRYPTO;
if (EVP_EncryptInit_ex(ectx, EVP_aes_128_ccm(), NULL, NULL, NULL) != 1
@@ -311,10 +276,7 @@ sc_status_t sc_decrypt(sc_context_t *ctx, const uint8_t *ciphertext, size_t ciph
uint8_t nonce[SC_NONCE_SIZE];
memcpy(nonce, ciphertext, SC_NONCE_SIZE);
DEBUG_TRACE(DEBUG_CATEGORY_CRYPTO, "DECRYPT: seskey=%02x%02x%02x%02x nonce=%02x%02x%02x%02x ct_len=%zu",
ctx->session_key[0], ctx->session_key[1], ctx->session_key[2], ctx->session_key[3],
nonce[0], nonce[1], nonce[2], nonce[3],
ciphertext_len);
DEBUG_TRACE(DEBUG_CATEGORY_CRYPTO, "DECRYPT: rx=%llu bytes=%zu", (unsigned long long)ctx->rx_counter, ciphertext_len);
const uint8_t *encrypted_data = ciphertext + SC_NONCE_SIZE;
size_t encrypted_len = ciphertext_len - SC_NONCE_SIZE;
size_t total_plaintext_len = encrypted_len - SC_TAG_SIZE;
src/transport_layer/secure_channel.h
+1
View File
@@ -2,6 +2,7 @@
* Инициализировать локальные X25519-ключи → init_ctx → set_peer_public_key → encrypt/decrypt.
* ctx заимствует SC_MYKEYS до конца использования; состояние счётчиков и stream-объекты имеют одного владельца.
* Пакеты: AES-128-CCM над plaintext+CRC32; выход = nonce(13)+ciphertext+tag(16).
* Каждый nonce создаёт OpenSSL CSPRNG; при ошибке RNG encrypt возвращает ошибку с нулевой длиной выхода.
* Отдельные API дают обфускацию ключа, потоковый AES-CTR и Ed25519; CTR сам по себе не аутентифицирует данные.
* Возврат SC_OK / отрицательный SC_ERR_*; caller владеет входными/выходными буферами. */
#ifndef SECURE_CHANNEL_H
src/transport_layer/stcp_link.c
+3 -3
View File
@@ -81,11 +81,11 @@ static void link_rx_cb(struct ll_queue *q, void *arg) {
queue_dgram_free(e); queue_entry_free(e);
} else {
struct UTUN_INSTANCE *inst = link->inst;
uint8_t id = (e->dgram && e->len > 0) ? e->dgram[0] : 0;
uint8_t id = (e->dgram && e->len > 0) ? e->dgram[0] : ETCP_ID_DATA;
if (inst && inst->api_bindings.callbacks[id])
inst->api_bindings.callbacks[id](link->etcp_conn ? link->etcp_conn : NULL, e);
else if (inst && inst->api_bindings.callbacks[0])
inst->api_bindings.callbacks[0](link->etcp_conn ? link->etcp_conn : NULL, e);
else if (inst && inst->api_bindings.callbacks[ETCP_ID_DATA])
inst->api_bindings.callbacks[ETCP_ID_DATA](link->etcp_conn ? link->etcp_conn : NULL, e);
else { queue_dgram_free(e); queue_entry_free(e); }
}
queue_resume_callback(q);
src/utun_instance.c
+2 -2
View File
@@ -450,7 +450,7 @@ static void instance_destroy(struct UTUN_INSTANCE* instance, int owns_config) {
/* Phase F: BGP */
if (instance->topo_groups) {
etcp_router_unbind(instance, ETCP_RT_ID_CONN_MGR);
etcp_unbind(instance, ETCP_RT_ID_GROUP_INVITE);
etcp_unbind(instance, ETCP_ID_GROUP_INVITE);
topo_group_invite_cancel_all(instance);
topo_groups_destroy(instance);
}
@@ -873,7 +873,7 @@ void chat_service_stop(struct UTUN_INSTANCE* instance) {
member_files_destroy(instance);
media_delivery_destroy(instance);
call_destroy(instance); dm_media_destroy(instance); dm_core_destroy(instance); dm_mailbox_destroy(instance);
etcp_unbind(instance, ETCP_RT_ID_GROUP_INVITE);
etcp_unbind(instance, ETCP_ID_GROUP_INVITE);
topo_group_invite_cancel_all(instance);
chat_sync_destroy(instance);
struct ll_entry* entry = instance->topo_groups ? instance->topo_groups->group_list->head : NULL;
src/utun_instance.h
+1 -1
View File
@@ -289,7 +289,7 @@ struct UTUN_INSTANCE {
// Per-instance NCD state (replaces global statics in node_conn_direct.c)
void* ncd_registry; // struct ncd_entry* linked list
uint8_t ncd_control_bound; // 1 = etcp_bind(ETCP_RT_ID_NCD_CONTROL) сделан
uint8_t ncd_control_bound; // 1 = etcp_bind(ETCP_ID_NCD_CONTROL) сделан
};
// Создать ядро по файлу конфигурации; NULL при ошибке. Сервисы ещё не запущены.
tests/Makefile.am
+7 -2
View File
@@ -1,7 +1,7 @@
# Tests Makefile.am for utun - all tests with new ll_queue library
EXTRA_DIST = fixtures/reality_openssl_hello.hex data/silero_speech.pcm data/silero_speech.txt test_net.h nat_fixture.h \
run_tests.py test_run_tests.py test_build_script.py local_transport_ipv6.sh stream_netns/topology.py stream_netns/network.py \
run_tests.py test_run_tests.py test_service_ids.py test_build_script.py local_transport_ipv6.sh stream_netns/topology.py stream_netns/network.py \
stream_netns/probe.py stream_netns/run.py stream_netns/scenario.py stream_netns/README.md stream_netns/TESTED.md \
stream_netns/test_isolation.py stream_netns/nat_scenario.py stream_netns/stun_server.py stream_netns/stun_scenario.py
@@ -315,6 +315,7 @@ test_etcp_bbr_LDADD = $(LIBUTUN) $(COMMON_LIBS)
test_etcp_crypto_SOURCES = test_etcp_crypto.c
test_etcp_crypto_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
test_etcp_crypto_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
test_etcp_crypto_LDFLAGS = -Wl,--wrap=RAND_bytes
test_stream_cipher_SOURCES = test_stream_cipher.c
test_stream_cipher_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
@@ -853,7 +854,7 @@ clean-local:
TEST_LOG_DIR = $(top_builddir)/tests/logs
# По умолчанию jobs = доступные CPU; TEST_JOBS=1 сохраняет последовательный запуск.
check-local: check-runner $(check_PROGRAMS)
check-local: check-runner check-service-ids $(check_PROGRAMS)
@python3 "$(srcdir)/run_tests.py" --log-dir "$(TEST_LOG_DIR)" $(check_PROGRAMS)
.PHONY: check-runner
@@ -861,6 +862,10 @@ check-runner:
@python3 "$(srcdir)/test_run_tests.py"
@echo "[INFO] test runner regression checks passed"
.PHONY: check-service-ids
check-service-ids:
@python3 "$(srcdir)/test_service_ids.py"
.PHONY: check-build-script
check-build-script:
@if command -v python3 >/dev/null 2>&1; then \
tests/bbr_integration/test_bbr_integration.c
+2 -2
View File
@@ -191,7 +191,7 @@ static void send_burst(struct test_ctx* ctx) {
while (sent < BURST_MAX) {
struct ll_entry* e = ll_alloc_lldgram(1 + PAYLOAD_SIZE);
if (!e) break;
e->dgram[0] = ETCP_RT_ID_DATA;
e->dgram[0] = ETCP_ID_DATA;
e->len = 1 + PAYLOAD_SIZE;
if (etcp_send(conn, e) == 0) { ctx->bytes_sent += PAYLOAD_SIZE; sent++; }
else { queue_entry_free(e); break; }
@@ -354,7 +354,7 @@ int main(void) {
printf("Init sender...\n");
if (utun_instance_init(ctx.sender) < 0) { printf("ERROR: sender init failed\n"); return 1; }
etcp_bind(ctx.receiver, ETCP_RT_ID_DATA, on_recv);
etcp_bind(ctx.receiver, ETCP_ID_DATA, on_recv);
int receiver_port = ss_get_port(&test_etcp_bound_socket(ctx.receiver, "srv")->local_addr);
int sender_port = ss_get_port(&test_etcp_bound_socket(ctx.sender, "local")->local_addr);
tests/run_tests.py
+9 -6
View File
@@ -11,11 +11,14 @@ import sys
import time
# Проверки точных порогов CPU/wall-time выполняем после общего пула.
EXCLUSIVE = frozenset({
"test_media_async", "test_uasync_load",
})
EXCLUSIVE |= {name + ".exe" for name in EXCLUSIVE}
# CPU-замеры и проверки общего LAN discovery-порта выполняем после общего пула по одному.
EXCLUSIVE = {
"test_media_async": "check CPU/wall-time limits",
"test_uasync_load": "check CPU/wall-time limits",
"test_local_transport": "isolate the LAN discovery port",
"test_local_transport_packets": "isolate the LAN discovery port",
}
EXCLUSIVE.update({name + ".exe": reason for name, reason in list(EXCLUSIVE.items())})
def available_cpus():
@@ -122,7 +125,7 @@ def run_tests(args, requested_signals):
continue
command = [str(Path(name).resolve())]
if name in EXCLUSIVE:
print(f"[INFO] {name}: running alone to check CPU/wall-time limits", flush=True)
print(f"[INFO] {name}: running alone to {EXCLUSIVE[name]}", flush=True)
if name == "test_auto_socket_dynamic" and use_sudo:
command = ["sudo", "-n", "--"] + command
print(f"[INFO] {name}: running as root via sudo -n", flush=True)
tests/test_bgp_paths.c
+37 -8
View File
@@ -25,6 +25,7 @@ static unsigned graph, delivered;
static uint32_t random_state = 1;
static int router_test, blocked_from = -1, blocked_to = -1;
static unsigned received, transit_packets;
static unsigned relay_crypto_checks, media_received;
static unsigned next_random(void) {
random_state = random_state * 1664525U + 1013904223U;
@@ -57,12 +58,21 @@ static int step(void) {
if ((graph & edge(a, b)) && e->len && e->dgram[0] == ETCP_ID_TOPO_ENTRY) {
delivered++;
nodes[b]->api_bindings.callbacks[ETCP_ID_TOPO_ENTRY](conns[b][a], e);
} else if ((graph & edge(a, b)) && e->len && e->dgram[0] == ETCP_RT_ID_SVC_ROUTE) {
} else if ((graph & edge(a, b)) && e->len && e->dgram[0] == ETCP_ID_SVC_ROUTE) {
unsigned count = e->dgram[e->len - 1];
assert(count && count <= ROUTER_MAX_VISITED);
uint64_t last; memcpy(&last, e->dgram + e->len - 9, 8); assert(last == nodes[a]->node_id);
if (count > 1) transit_packets++;
nodes[b]->api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE](conns[b][a], e);
const struct SVC_ROUTE_HDR* header = (const struct SVC_ROUTE_HDR*)e->dgram;
size_t body = e->len - count * 8 - 1;
if ((header->svc_id == ETCP_RT_ID_CALL_AUDIO || header->svc_id == ETCP_RT_ID_CALL_VIDEO)
&& body > SVC_ROUTE_HDR_SIZE && a != 0 && a != 3) {
assert(header->flags == (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED));
uint8_t* decoded = NULL, flags = 0; size_t decoded_size = 0;
assert(route_crypto_decode(nodes[a], e->dgram, body, &decoded, &decoded_size, &flags) == -1 && !decoded);
relay_crypto_checks++;
}
nodes[b]->api_bindings.callbacks[ETCP_ID_SVC_ROUTE](conns[b][a], e);
} else release(e);
queue_resume_callback(q);
}
@@ -174,14 +184,15 @@ static void metric_jitter(void) {
}
static void service_receive(struct ETCP_CONN* conn, struct ll_entry* e) {
assert(conn->instance == nodes[3]);
if (e->len > ROUTER_SVC_HDR_SIZE) {
assert(conn->instance == nodes[3]);
uint64_t source, group;
memcpy(&source, e->dgram + ROUTER_SVC_SRC_OFF, 8); memcpy(&group, e->dgram + ROUTER_SVC_GROUP_OFF, 8);
assert(source == nodes[0]->node_id && group == GROUP);
assert(e->dgram[ROUTER_SVC_FLAGS_OFF] == (ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED));
assert(e->len == ROUTER_SVC_HDR_SIZE + 4 && !memcmp(e->dgram + ROUTER_SVC_HDR_SIZE, "test", 4));
received++;
if (e->dgram[0] == ETCP_RT_ID_CALL_AUDIO || e->dgram[0] == ETCP_RT_ID_CALL_VIDEO) media_received++;
else received++;
}
release(e);
}
@@ -189,7 +200,7 @@ static void service_receive(struct ETCP_CONN* conn, struct ll_entry* e) {
static struct ll_entry* route_packet(uint8_t flags, const uint64_t* ids, unsigned count) {
size_t body = SVC_ROUTE_HDR_SIZE + (flags ? 0 : 4);
struct ll_entry* e = ll_alloc_lldgram(body + count * 8 + 1); assert(e);
struct SVC_ROUTE_HDR h = { .cmd = ETCP_RT_ID_SVC_ROUTE, .group_id = GROUP,
struct SVC_ROUTE_HDR h = { .cmd = ETCP_ID_SVC_ROUTE, .group_id = GROUP,
.src_node_id = ids[0], .dst_node_id = nodes[3]->node_id, .svc_id = 0x41, .flags = flags, .reset_id = 1 };
memcpy(e->dgram, &h, sizeof(h));
if (!flags) memcpy(e->dgram + sizeof(h), "loop", 4);
@@ -199,7 +210,7 @@ static struct ll_entry* route_packet(uint8_t flags, const uint64_t* ids, unsigne
}
static void inject_router(struct ll_entry* e) {
nodes[1]->api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE](conns[1][0], e);
nodes[1]->api_bindings.callbacks[ETCP_ID_SVC_ROUTE](conns[1][0], e);
}
static void blocked_ready(struct ll_queue* q, void* arg) {
@@ -211,14 +222,31 @@ static void router_paths(void) {
router_test = 1;
unsigned chain = edge(0, 1) | edge(1, 2) | edge(2, 3);
change(chain); settle(); verify();
assert(etcp_router_bind(nodes[3], 0x42, service_receive) == 0);
struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(nodes[0], GROUP, nodes[3]->node_id, 0x42); assert(c);
assert(etcp_router_bind(nodes[3], 0xF0, service_receive) == 0);
struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(nodes[0], GROUP, nodes[3]->node_id, 0xF0); assert(c);
assert(etcp_router_conn_send(c, (const uint8_t*)"test", 4, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0);
uint64_t deadline = get_time_tb() + 20000;
while ((!received || c->tx_acked != c->tx_seq) && get_time_tb() < deadline) { step(); uasync_poll(ua, 1); }
assert(received == 1 && c->tx_acked == c->tx_seq && transit_packets > 0);
settle();
const uint8_t media[] = {ETCP_RT_ID_CALL_AUDIO, ETCP_RT_ID_CALL_VIDEO};
for (unsigned i = 0; i < sizeof(media); i++) {
assert(etcp_router_bind(nodes[0], media[i], service_receive) == 0);
assert(etcp_router_bind(nodes[3], media[i], service_receive) == 0);
for (int side = 0; side < N; side += 3)
assert(etcp_router_require_crypto(nodes[side], media[i], ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0);
struct ETCP_ROUTER_CONN* channel = etcp_router_conn_get(nodes[0], GROUP, nodes[3]->node_id, media[i]); assert(channel);
assert(etcp_router_conn_send(channel, (const uint8_t*)"test", 4, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0);
deadline = get_time_tb() + 20000;
while ((media_received != i + 1 || channel->tx_acked != channel->tx_seq) && get_time_tb() < deadline) { step(); uasync_poll(ua, 1); }
assert(media_received == i + 1 && channel->tx_acked == channel->tx_seq);
settle();
}
assert(relay_crypto_checks >= 4);
DEBUG_INFO(DEBUG_CATEGORY_CALL, "audio/video E2E: two transit nodes cannot decrypt, destination receives exact payload checks=%u",
relay_crypto_checks);
/* Actual transient two-hop loop B -> C -> B. Each hop must append once,
* for DATA, ACK and every control flag, before B rejects its second visit. */
struct TOPO_GROUP_NODE* target = topo_node_find_by_id(groups[2], nodes[3]->node_id);
@@ -303,6 +331,7 @@ static void router_paths(void) {
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_DEBUG);
debug_set_category_level(DEBUG_CATEGORY_CALL, DEBUG_LEVEL_INFO);
utun_instance_set_tun_init_enabled(0);
create();
unsigned triangle = edge(0, 1) | edge(1, 2) | edge(0, 2) | edge(2, 3);
tests/test_call.c
+1
View File
@@ -1058,6 +1058,7 @@ int main(int argc, char** argv) {
g_signal_recv[i] = t.inst[i]->router_bindings.callbacks[ETCP_RT_ID_CALL];
etcp_router_unbind(t.inst[i], ETCP_RT_ID_CALL);
etcp_router_bind(t.inst[i], ETCP_RT_ID_CALL, signal_recv_loss);
if (etcp_router_require_crypto(t.inst[i], ETCP_RT_ID_CALL, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT)) return 1;
}
chat_event_set_handler(t.inst[IDX_A], call_evt_handler);
chat_event_set_handler(t.inst[IDX_B], call_evt_handler);
tests/test_call_video_flush.c
+1 -1
View File
@@ -32,7 +32,7 @@ struct ETCP_ROUTER_CONN *test_router_conn_get(struct UTUN_INSTANCE *inst, uint64
(void)inst; (void)group; (void)peer; assert(svc == ETCP_RT_ID_CALL_VIDEO); return &router;
}
int test_router_conn_send(struct ETCP_ROUTER_CONN *conn, const uint8_t *data, size_t len, int mode) {
assert(conn == &router && !mode);
assert(conn == &router && mode == (ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT));
struct call_video_frame f;
assert(!call_video_fragment_parse(data, len, &f));
assert(f.call_id == 7 && f.frame_id == 1 && f.size == FRAME_BYTES && f.offset == received);
tests/test_dm_invite.c
+3 -3
View File
@@ -46,7 +46,7 @@ static void drop_history(struct ETCP_CONN* conn, struct ll_entry* entry) {
dropped_requests++;
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite test: dropped history request=%d group=%016llx", dropped_requests, gid);
if (dropped_requests == 2)
for (int i = 0; i < 2; i++) etcp_bind(nodes[i], ETCP_RT_ID_DB_SYNC, history_receivers[i]);
for (int i = 0; i < 2; i++) etcp_bind(nodes[i], ETCP_ID_DB_SYNC, history_receivers[i]);
}
queue_dgram_free(entry); queue_entry_free(entry);
}
@@ -242,8 +242,8 @@ int main(void) {
if (getenv("UTUN_TEST_HISTORY_TIMEOUT")) {
dropped_group = target;
for (int i = 0; i < 2; i++) {
history_receivers[i] = nodes[i]->api_bindings.callbacks[ETCP_RT_ID_DB_SYNC];
etcp_bind(nodes[i], ETCP_RT_ID_DB_SYNC, drop_history);
history_receivers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_DB_SYNC];
etcp_bind(nodes[i], ETCP_ID_DB_SYNC, drop_history);
}
}
CHECK(!join_link(ua, link), "B joins target group using received PM link");
tests/test_encrypted_attachment.c
+2 -1
View File
@@ -292,8 +292,9 @@ int main(void) {
if (getenv("UTUN_TEST_DEBUG")) {
debug_set_category_level_by_name("dm", "debug"); debug_set_category_level_by_name("media", "debug");
debug_set_category_level_by_name("chat_sync", "debug"); debug_set_category_level_by_name("member_sync", "info");
debug_set_category_level_by_name("bgp", "info"); debug_set_category_level_by_name("etcp_route", "info");
debug_set_category_level_by_name("bgp", "info"); debug_set_category_level_by_name("etcp_route", "debug");
debug_set_category_level_by_name("connection", "debug"); debug_set_category_level_by_name("etcp", "info");
debug_set_category_level_by_name("general", "debug");
}
utun_instance_set_tun_init_enabled(0);
snprintf(t.dir, sizeof(t.dir), "/tmp/utun_encrypted_attachment_XXXXXX");
tests/test_etcp_congestion.c
+2 -2
View File
@@ -194,7 +194,7 @@ static void send_burst(struct test_ctx* ctx) {
while (sent < 64) {
struct ll_entry* e = ll_alloc_lldgram(sizeof(uint8_t) + PAYLOAD_SIZE);
if (!e) break;
e->dgram[0] = ETCP_RT_ID_DATA;
e->dgram[0] = ETCP_ID_DATA;
e->len = 1 + PAYLOAD_SIZE;
if (etcp_send(conn, e) == 0) { ctx->bytes_sent += e->len - 1; sent++; }
else { queue_entry_free(e); break; }
@@ -331,7 +331,7 @@ int main(void) {
printf("Init sender...\n");
if (utun_instance_init(ctx.sender) < 0) { printf("Sender init failed\n"); return 1; }
etcp_bind(ctx.receiver, ETCP_RT_ID_DATA, on_recv);
etcp_bind(ctx.receiver, ETCP_ID_DATA, on_recv);
printf("Creating dummynet...\n");
ctx.dn[0] = dummynet_create(ctx.ua, "127.0.0.1", DN1_PORT);
tests/test_etcp_crypto.c
+27
View File
@@ -9,6 +9,12 @@
#define TEST_DATA "Hello, ETCP encrypted world!"
#define TEST_DATA_LEN 28
static int fail_nonce;
extern int __real_RAND_bytes(unsigned char* data, int size);
int __wrap_RAND_bytes(unsigned char* data, int size) {
return fail_nonce ? 0 : __real_RAND_bytes(data, size);
}
int main() {
debug_config_init();
debug_set_level(DEBUG_LEVEL_TRACE);
@@ -75,6 +81,27 @@ int main() {
}
DEBUG_INFO(DEBUG_CATEGORY_CRYPTO, "Client->Server encryption/decryption: PASSED");
uint8_t first_nonce[SC_NONCE_SIZE]; memcpy(first_nonce, ciphertext, sizeof(first_nonce));
client_ctx.tx_counter = 0; // Пересозданный контекст/вытеснение кэша не должны повторять nonce.
if (sc_encrypt(&client_ctx, plaintext, TEST_DATA_LEN, ciphertext, &ciphertext_len) != SC_OK
|| !memcmp(first_nonce, ciphertext, sizeof(first_nonce))) {
DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "Nonce repeated after counter reset"); return 1;
}
uint64_t counter = client_ctx.tx_counter;
fail_nonce = 1; ciphertext_len = 777;
sc_status_t result = sc_encrypt(&client_ctx, plaintext, TEST_DATA_LEN, ciphertext, &ciphertext_len);
fail_nonce = 0;
if (result != SC_ERR_CRYPTO || ciphertext_len || client_ctx.tx_counter != counter) {
DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "RNG failure produced output or advanced counter"); return 1;
}
if (sc_encrypt(&client_ctx, plaintext, TEST_DATA_LEN, ciphertext, &ciphertext_len) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "Encryption did not recover after RNG failure"); return 1;
}
ciphertext[ciphertext_len - 1] ^= 1;
if (sc_decrypt(&server_ctx, ciphertext, ciphertext_len, decrypted, &decrypted_len) != SC_ERR_AUTH_FAILED) {
DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "Corrupted CCM tag accepted"); return 1;
}
DEBUG_INFO(DEBUG_CATEGORY_CRYPTO, "Nonce after reset, RNG failure without plaintext fallback, corrupted tag: PASSED");
DEBUG_INFO(DEBUG_CATEGORY_CRYPTO, "ETCP crypto test PASSED");
return 0;
}
tests/test_etcp_reinit_inflight.c
+2 -2
View File
@@ -215,7 +215,7 @@ static void monitor(void* arg) {
ctx->receiver = create_instance(ctx->ua, 0x2222222222222222ULL, s_priv, s_pub);
add_server(ctx->receiver, "srv1", SRV_PORT);
utun_instance_init(ctx->receiver);
etcp_bind(ctx->receiver, ETCP_RT_ID_DATA, on_recv);
etcp_bind(ctx->receiver, ETCP_ID_DATA, on_recv);
}
ctx->phase = 4;
break;
@@ -302,7 +302,7 @@ int main(void) {
if (utun_instance_init(ctx.receiver) < 0) { printf("receiver init failed\n"); return 1; }
if (utun_instance_init(ctx.sender) < 0) { printf("sender init failed\n"); return 1; }
etcp_bind(ctx.receiver, ETCP_RT_ID_DATA, on_recv);
etcp_bind(ctx.receiver, ETCP_ID_DATA, on_recv);
printf("Creating dummynet...\n");
ctx.dn = dummynet_create(ctx.ua, "127.0.0.1", DN_PORT);
tests/test_etcp_router_faults.c
+1 -1
View File
@@ -65,7 +65,7 @@ static void receive(struct ETCP_CONN* conn, struct ll_entry* e) {
if (closed_in_handler) { closed_in_handler = 0; etcp_router_conn_close(rc(side, svc)); }
}
static void inject(int side, struct ll_entry* e) {
ep[side].inst.api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE](&ep[side].conn, e);
ep[side].inst.api_bindings.callbacks[ETCP_ID_SVC_ROUTE](&ep[side].conn, e);
}
static void wire(int from, struct ll_queue* q) {
if ((unsigned)q->count > max_wire) max_wire = q->count;
tests/test_etcp_router_unit.c
+55 -12
View File
@@ -70,7 +70,7 @@ static void inject_wire(struct ETCP_CONN* conn, struct ll_entry* e) {
memcpy(wire, e->dgram, len); memcpy(wire + len, &src, 8); wire[len + 8] = 1;
queue_dgram_free(e); e->dgram = wire; e->len = len + 9;
conn->peer_node_id = src;
conn->instance->api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE](conn, e);
conn->instance->api_bindings.callbacks[ETCP_ID_SVC_ROUTE](conn, e);
}
// Настоящий HELLO/challenge/CONFIRM через публичный wire callback; состояние не подменяем.
@@ -79,7 +79,7 @@ static void establish(etcp_recv_fn cb, struct UTUN_INSTANCE* inst, uint64_t src,
if (c->peer_reset_id == epoch) return;
c->challenge_sent_tb = 0; // тест не ждёт throttle между двумя намеренными рестартами
struct SVC_ROUTE_HDR h = {0};
h.cmd = ETCP_RT_ID_SVC_ROUTE; h.group_id = TOPO_GROUP_UTUN;
h.cmd = ETCP_ID_SVC_ROUTE; h.group_id = TOPO_GROUP_UTUN;
h.src_node_id = src; h.dst_node_id = inst->node_id; h.svc_id = svc;
h.reset_id = epoch; h.flags = ROUTER_FLAG_START;
struct ll_entry* e = ll_alloc_lldgram(sizeof(h));
@@ -93,7 +93,7 @@ static void inject(etcp_recv_fn recv_cb, struct UTUN_INSTANCE* inst,
const uint8_t* pl, size_t pl_len, int is_ack, uint8_t flags) {
struct SVC_ROUTE_HDR hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.cmd = ETCP_RT_ID_SVC_ROUTE;
hdr.cmd = ETCP_ID_SVC_ROUTE;
hdr.group_id = TOPO_GROUP_UTUN;
hdr.dst_node_id = inst->node_id;
hdr.src_node_id = src;
@@ -123,7 +123,7 @@ static void inject_epoch(etcp_recv_fn recv_cb, struct UTUN_INSTANCE* inst,
const uint8_t* pl, size_t pl_len, int is_ack, uint8_t flags, uint64_t reset_id) {
struct SVC_ROUTE_HDR hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.cmd = ETCP_RT_ID_SVC_ROUTE;
hdr.cmd = ETCP_ID_SVC_ROUTE;
hdr.group_id = TOPO_GROUP_UTUN;
hdr.dst_node_id = inst->node_id;
hdr.src_node_id = src;
@@ -268,7 +268,7 @@ static void inject_signed(etcp_recv_fn recv_cb, struct UTUN_INSTANCE* inst,
struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)dgram;
memset(hdr, 0, SVC_ROUTE_HDR_SIZE);
hdr->cmd = ETCP_RT_ID_SVC_ROUTE;
hdr->cmd = ETCP_ID_SVC_ROUTE;
hdr->group_id = TOPO_GROUP_UTUN;
hdr->dst_node_id = inst->node_id;
hdr->src_node_id = src;
@@ -309,11 +309,11 @@ static int test_init_destroy(void) {
if (etcp_router_init(&inst) != 0) FAIL("init");
if (inst.router_conns == NULL) FAIL("router_conns NULL after init");
if (inst.api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE] == NULL) FAIL("recv_cb not bound");
if (inst.api_bindings.callbacks[ETCP_ID_SVC_ROUTE] == NULL) FAIL("recv_cb not bound");
etcp_router_destroy(&inst);
if (inst.router_conns != NULL) FAIL("router_conns not NULL after destroy");
if (inst.api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE] != NULL) FAIL("recv_cb still bound");
if (inst.api_bindings.callbacks[ETCP_ID_SVC_ROUTE] != NULL) FAIL("recv_cb still bound");
uasync_destroy(ua, 0);
PASS();
@@ -1035,7 +1035,7 @@ static int test_sign_too_short(void) {
rx_reset(0xF5);
struct SVC_ROUTE_HDR hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.cmd = ETCP_RT_ID_SVC_ROUTE;
hdr.cmd = ETCP_ID_SVC_ROUTE;
hdr.group_id = TOPO_GROUP_UTUN;
hdr.dst_node_id = inst.node_id;
hdr.src_node_id = SIGNER_NODE;
@@ -1132,7 +1132,7 @@ static void inject_encrypted(struct UTUN_INSTANCE* inst, etcp_recv_fn recv_cb,
}
struct SVC_ROUTE_HDR hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.cmd = ETCP_RT_ID_SVC_ROUTE;
hdr.cmd = ETCP_ID_SVC_ROUTE;
hdr.group_id = TOPO_GROUP_UTUN;
hdr.dst_node_id = inst->node_id;
hdr.src_node_id = src;
@@ -1209,7 +1209,7 @@ static int test_e2e_decrypt_tampered(void) {
struct SVC_ROUTE_HDR hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.cmd = ETCP_RT_ID_SVC_ROUTE; hdr.group_id = TOPO_GROUP_UTUN;
hdr.cmd = ETCP_ID_SVC_ROUTE; hdr.group_id = TOPO_GROUP_UTUN;
hdr.dst_node_id = inst.node_id; hdr.src_node_id = E2E_PEER_NODE;
hdr.seq = 0; hdr.svc_id = TEST_SVC_ID; hdr.flags = ROUTER_FLAG_ENCRYPTED;
establish(recv_cb, &inst, E2E_PEER_NODE, TEST_SVC_ID, E2E_PEER_NODE);
@@ -1267,10 +1267,11 @@ static int test_e2e_cache_reuse(void) {
}
static int test_e2e_combined_signed(void) {
TEST("E2E ENCRYPTED+SIGNED — verify+decrypt+deliver");
TEST("E2E required ENCRYPTED+SIGNED — header/ciphertext tampering, stripped flag, replay and delivery");
struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb;
SETUP();
setup_bgp_for_sign_test(&inst, 1);
TESTASSERT(etcp_router_require_crypto(&inst, TEST_SVC_ID, ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED) == 0);
memcpy(&inst.my_keys, &g_sign_keys, sizeof(g_sign_keys));
struct SC_MYKEYS peer_keys;
@@ -1299,7 +1300,7 @@ static int test_e2e_combined_signed(void) {
uint8_t* dgram = u_malloc(total);
struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)dgram;
memset(hdr, 0, SVC_ROUTE_HDR_SIZE);
hdr->cmd = ETCP_RT_ID_SVC_ROUTE; hdr->group_id = TOPO_GROUP_UTUN;
hdr->cmd = ETCP_ID_SVC_ROUTE; hdr->group_id = TOPO_GROUP_UTUN;
hdr->dst_node_id = inst.node_id; hdr->src_node_id = SIGNER_NODE;
hdr->seq = 0; hdr->svc_id = TEST_SVC_ID;
hdr->flags = ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED;
@@ -1310,12 +1311,25 @@ static int test_e2e_combined_signed(void) {
if (sc_ed25519_sign(g_sign_ed25519_privkey, dgram, hdr_pl, dgram + hdr_pl) != SC_OK) { u_free(dgram); u_free(enc_buf); FAIL("sign"); }
const size_t changed[] = {offsetof(struct SVC_ROUTE_HDR, flags), offsetof(struct SVC_ROUTE_HDR, seq), SVC_ROUTE_HDR_SIZE + SC_NONCE_SIZE};
for (unsigned i = 0; i < sizeof(changed) / sizeof(changed[0]); i++) {
struct ll_entry* bad = ll_alloc_lldgram(total); TESTASSERT(bad);
memcpy(bad->dgram, dgram, total); bad->len = total;
bad->dgram[changed[i]] ^= i == 0 ? ROUTER_FLAG_ENCRYPTED : 1;
recv_cb(&fake_conn, bad);
TESTASSERT(!rx.delivered && !c->rx_seq && !c->incoming_q->count);
}
struct ll_entry* replay = ll_alloc_lldgram(total); TESTASSERT(replay);
memcpy(replay->dgram, dgram, total); replay->len = total;
struct ll_entry* e = queue_entry_new(0);
if (!e) { u_free(dgram); u_free(enc_buf); FAIL("entry"); }
e->dgram = dgram; e->len = total; u_free(enc_buf);
recv_cb(&fake_conn, e);
if (rx.delivered != 1 || rx.errors != 0) FAIL("combined signed+encrypted");
recv_cb(&fake_conn, replay);
TESTASSERT(rx.delivered == 1 && !rx.errors && c->rx_seq == 1);
teardown_bgp_for_sign_test(&inst);
TEARDOWN();
@@ -1324,8 +1338,10 @@ static int test_e2e_combined_signed(void) {
}
static int receive_capacity;
static unsigned ready_checks;
static int receive_ready(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* payload, size_t bytes) {
(void)inst; (void)group; (void)peer;
ready_checks++;
if (bytes != 1 || payload[0] != rx.marker) rx.errors++;
return rx.delivered < receive_capacity;
}
@@ -1364,6 +1380,32 @@ static int test_receive_backpressure(void) {
TEARDOWN(); PASS(); return 0;
}
static int test_required_crypto(void) {
TEST("Required crypto: plaintext/partial flags rejected before receive state and ready callback");
struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb;
SETUP();
TESTASSERT(etcp_router_require_crypto(NULL, TEST_SVC_ID, ROUTER_FLAG_ENCRYPTED) == -1);
TESTASSERT(etcp_router_require_crypto(&inst, TEST_SVC_ID2, ROUTER_FLAG_ENCRYPTED) == -1);
TESTASSERT(etcp_router_require_crypto(&inst, TEST_SVC_ID, ROUTER_FLAG_START) == -1);
TESTASSERT(etcp_router_require_crypto(&inst, TEST_SVC_ID, ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED) == 0);
etcp_router_set_receive_ready(&inst, TEST_SVC_ID, receive_ready);
rx_reset(0xA7); ready_checks = 0;
const uint8_t payload = 0xA7;
const uint8_t modes[] = {0, ROUTER_FLAG_ENCRYPTED, ROUTER_FLAG_SIGNED, ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED};
for (unsigned i = 0; i < sizeof(modes); i++) {
inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, &payload, 1, 0, modes[i]);
}
struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID);
TESTASSERT(!rx.delivered && !ready_checks && !c->rx_seq && !c->incoming_q->count);
for (unsigned i = 0; i < 3; i++) TESTASSERT(etcp_router_conn_send(c, &payload, 1, modes[i]) == -1);
TESTASSERT(!c->tx_seq && !c->send_q->count);
TESTASSERT(etcp_router_unbind(&inst, TEST_SVC_ID) == 0 && !inst.router_bindings.required_crypto[TEST_SVC_ID]);
TESTASSERT(etcp_router_bind(&inst, TEST_SVC_ID, test_handler) == 0);
TESTASSERT(etcp_router_require_crypto(&inst, TEST_SVC_ID, ROUTER_FLAG_ENCRYPTED) == 0);
TESTASSERT(etcp_router_bind(&inst, TEST_SVC_ID, NULL) == 0 && !inst.router_bindings.required_crypto[TEST_SVC_ID]);
TEARDOWN(); PASS(); return 0;
}
// ======================== Main ========================
int main(void) {
// Run all tests
@@ -1374,6 +1416,7 @@ int main(void) {
test_conn_close();
test_in_order();
test_receive_backpressure();
test_required_crypto();
test_out_of_order();
test_duplicate_same();
test_duplicate_past();
tests/test_etcp_server_lifecycle.c
+2 -2
View File
@@ -84,10 +84,10 @@ static struct ETCP_CONN* ready_conn(int tcp) {
}
static void receive_control(struct ETCP_CONN* conn, uint8_t subcmd) {
struct ncd_control_msg msg = { ETCP_RT_ID_NCD_CONTROL, subcmd, conn->peer_node_id };
struct ncd_control_msg msg = { ETCP_ID_NCD_CONTROL, subcmd, conn->peer_node_id };
struct ll_entry* entry = ll_alloc_lldgram(sizeof(msg)); CHECK(entry);
memcpy(entry->dgram, &msg, sizeof(msg)); entry->len = sizeof(msg);
inst.api_bindings.callbacks[ETCP_RT_ID_NCD_CONTROL](conn, entry);
inst.api_bindings.callbacks[ETCP_ID_NCD_CONTROL](conn, entry);
uasync_drain_immediate(inst.ua);
}
tests/test_local_transport_packets.c
+3
View File
@@ -204,6 +204,9 @@ int main(void) {
packet(LOCAL_ANNOUNCE, 2, 0, i & 1 ? "Peer A" : "Peer B", 6);
processed += inject(discovery);
}
if (processed != 256 || changes != changes_before || log_lines != logs_before)
fprintf(stderr, "local flood: processed=%d/256 changes=%d logs=%d real_udp=%d\n",
processed, changes - changes_before, log_lines - logs_before, wire_packets);
REQUIRE(processed == 256 && changes == changes_before && log_lines == logs_before);
next_window(ua); REQUIRE(changes == changes_before + 1 && log_lines - logs_before <= 10);
packet(LOCAL_MESSAGE, 2, 999, link, strlen(link)); REQUIRE(inject(unicast)); REQUIRE(invitations == before + 9);
tests/test_media_delivery_chat.c
+9 -1
View File
@@ -524,6 +524,9 @@ static void setup_db_sync(void) {
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
debug_set_category_level_by_name("media", "debug");
debug_set_category_level_by_name("etcp_route", "debug");
debug_set_category_level_by_name("general", "info");
if (getenv("UTUN_TEST_DEBUG")) {
debug_set_category_level_by_name("media", "info");
debug_set_category_level_by_name("general", "info");
@@ -535,6 +538,9 @@ int main(void) {
srand((unsigned)time(NULL));
printf("=== test_media_delivery_chat ===\n");
t_mkdtemp(g_tdir);
char log_path[512]; snprintf(log_path, sizeof(log_path), "%s/test.log", g_tdir);
debug_enable_file_output(log_path, 1); debug_enable_console(0);
printf("[INFO] media chat diagnostics: %s\n", log_path); fflush(stdout);
for (int i = 0; i < MEDIA_FILE_SIZE; i++) g_file_data[i] = (uint8_t)((i * 31 + 17) & 0xFF);
@@ -718,6 +724,8 @@ done:
g_result = 1;
for (int i = 0; i < N_NODES; i++) { if (g_inst[i]) { g_inst[i]->running = 0; utun_instance_destroy(g_inst[i]); g_inst[i] = NULL; } }
if (g_ua) { uasync_destroy(g_ua, 0); g_ua = NULL; }
t_rmdir(g_tdir);
debug_disable_file_output();
if (!G_FAILED) t_rmdir(g_tdir);
else printf("[INFO] media chat failure artifacts preserved: %s\n", g_tdir);
return G_FAILED > 0 ? 1 : 0;
}
tests/test_media_delivery_download.c
+31
View File
@@ -23,6 +23,7 @@
#include "../transport_layer/secure_channel.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/conn_mgr.h"
#include "../routing_layer/conn_mgr_priv.h"
#include <sqlite3.h>
#include <openssl/evp.h>
@@ -477,6 +478,35 @@ static void test_dedup(void) {
}
}
static void late_up(void* arg) { (*(int*)arg)++; }
/* Повторный QUERY и несколько блоков сохраняют один CM запрос; TIMEOUT отменяет поздний UP. */
static void test_cm_handle_lifetime(void) {
TEST("CM handle survives QUERY; TIMEOUT cancels late UP");
sqlite3* db = NULL; sqlite3_open(":memory:", &db);
struct UTUN_INSTANCE* inst = make_minimal_instance(db);
struct media_download* dl = make_test_dl(db, inst, 2);
struct CONN_MGR_HANDLE* h = u_calloc(1, sizeof(*h));
if (!dl || !h) { fprintf(stderr, "CM fixture allocation failed\n"); exit(1); }
int late = 0;
h->ua = g_ua; h->deliver_up_token = uasync_call_soon(g_ua, &late, late_up);
dl->num_peers = 1; dl->peers[0].node_id = 0x1111; dl->peers[0].cm_handle = h;
uint8_t wire[MEDIA_QUERY_RESP_HDR_SIZE + 2 * sizeof(struct media_pkt_query_resp_entry)] = {MEDIA_SUBCMD_QUERY_RESP};
uint16_t count = 2; memcpy(wire + 1, &count, 2);
struct media_pkt_query_resp_entry* entries = (struct media_pkt_query_resp_entry*)(wire + MEDIA_QUERY_RESP_HDR_SIZE);
for (int bi = 0; bi < 2; bi++) { entries[bi].node_id = 0x1111; memcpy(entries[bi].block_id, dl->blocks[bi].block_id, 16); }
media_download_handle_query_resp(inst, wire, sizeof(wire));
media_download_handle_query_resp(inst, wire, sizeof(wire));
int retained = dl->num_peers == 1 && dl->peers[0].cm_handle == h && dl->blocks[0].state == MD_BLK_CONN
&& dl->blocks[1].state == MD_BLK_CONN && dl->blocks[0].holder_idx == 0 && dl->blocks[1].holder_idx == 0;
md_dl_conn_cb(h, 0x1111, dl->group_id, CONN_EVENT_TIMEOUT, dl);
int closed = h->closed && !h->deliver_up_token && !dl->peers[0].cm_handle;
if (!h->closed) conn_mgr_close(h);
free_test_dl(inst, dl); sqlite3_close(db); u_free(inst);
uasync_poll(g_ua, 0);
if (retained && closed && !late) OK(); else FAIL("retained=%d closed=%d late=%d", retained, closed, late);
}
int main(void) {
test_setup();
printf("=== test_media_delivery_download ===\n");
@@ -493,6 +523,7 @@ int main(void) {
test_query_resp_empty();
test_query_resp_assign();
test_dedup();
test_cm_handle_lifetime();
printf("\n%d/%d passed, %d failed\n", g_passed, g_total, g_failed);
test_cleanup();
tests/test_run_tests.py
+11
View File
@@ -109,6 +109,17 @@ time.sleep(0.05)
self.assertEqual({name for name, _ in events[:2]}, set(names))
self.assertEqual([state for _, state in events], ["start", "start", "end", "end"])
def test_local_discovery_runs_alone_after_pool(self):
names = ["test_local_transport", "test_local_transport_packets"]
for name in ["ordinary", *names]:
self.fixture(name, "time.sleep(0.05)\n")
result = self.run_runner([names[0], "ordinary", names[1]])
self.assertEqual(result.returncode, 0, result.stdout)
events = [json.loads(line) for line in (self.root / "events").read_text().splitlines()]
self.assertEqual(events, [[name, state] for name in ["ordinary", *names] for state in ["start", "end"]])
for name in names:
self.assertIn(f"{name}: running alone to isolate the LAN discovery port", result.stdout)
def test_exit_codes_and_separate_logs(self):
self.fixture("ok")
self.fixture("bad", "print('failure detail', flush=True)\nsys.exit(42)\n")
tests/test_service_ids.py
+107
View File
@@ -0,0 +1,107 @@
#!/usr/bin/env python3
"""Проверка общего реестра ETCP: значения, единственное объявление и таблица регистрации."""
import os
from pathlib import Path
import re
import unittest
ROOT = Path(__file__).resolve().parents[1]
HEADER = Path("src/transport_layer/etcp_api.h")
ID = r"ETCP_(?:RT_)?ID_[A-Z0-9_]+"
DEFINE = re.compile(r"^\s*#\s*define\s+(" + ID + r")\s+([^\n]+)", re.MULTILINE)
COMMENTS = re.compile(r'"(?:\\.|[^"\\])*"|\'(?:\\.|[^\'\\])*\'|/\*.*?\*/|//[^\n]*', re.DOTALL)
BIND = re.compile(r"\b(etcp_bind|etcp_unbind|merkle_sync_init|etcp_router_bind|etcp_router_unbind|"
r"etcp_router_require_crypto|etcp_router_set_receive_ready)\s*\([^,]+,\s*(" + ID + r"|0x[\da-fA-F]+|\d+)\b")
CALLBACK = re.compile(r"\b(api_bindings|router_bindings)\.callbacks\s*\[\s*(" + ID + r"|0x[\da-fA-F]+|\d+)\s*\]")
def code(text):
return COMMENTS.sub(lambda m: m[0] if m[0][0] in "\"'" else "\n" * m[0].count("\n"), text)
def registry(text):
ids, values, errors = {}, {}, []
for match in DEFINE.finditer(code(text)):
name, literal = match.groups()
if name in ids:
errors.append(f"{name}: repeated declaration")
try:
value = int(literal.strip(), 0)
except ValueError:
errors.append(f"{name}: expected a numeric literal, got {literal!r}")
continue
if not 0 <= value <= 255:
errors.append(f"{name}: ID {value} does not fit in one byte")
space = "router" if name.startswith("ETCP_RT_ID_") else "raw"
previous = values.get((space, value))
if previous:
errors.append(f"{space} ID {value:#04x}: {previous} conflicts with {name}")
ids[name] = value
values[space, value] = name
return ids, errors
def check_source(text, ids):
text = code(text)
errors = [f"{name}: declare only in {HEADER}" for name, _ in DEFINE.findall(text)]
errors += [f"{name}: missing from {HEADER}" for name in sorted(set(re.findall(ID, text)) - ids.keys())]
uses = [(m[1], m[2], m[1].startswith("etcp_router_")) for m in BIND.finditer(text)]
uses += [(m[1], m[2], m[1] == "router_bindings") for m in CALLBACK.finditer(text)]
for operation, name, routed in uses:
prefix = "ETCP_RT_ID_" if routed else "ETCP_ID_"
if not name.startswith(prefix):
errors.append(f"{operation}: {name} must use a declared {prefix} identifier")
return errors
def sources():
excluded = {"tdesktop-dev", "third_party", "zxing-cpp", "openssl", "libopus", "speexdsp", "SoundTouch"}
for tree in ("src", "lib", "tools"):
for directory, children, files in os.walk(ROOT / tree):
children[:] = sorted(name for name in children if not name.startswith((".", "build")) and name not in excluded)
for name in sorted(files):
path = Path(directory) / name
if path.suffix in {".c", ".h", ".cpp", ".hpp", ".cc", ".cxx"} and path != ROOT / HEADER:
yield path
class ServiceIdsTests(unittest.TestCase):
def test_separate_tables(self):
ids, errors = registry("#define ETCP_ID_A 1\n#define ETCP_RT_ID_B 1\n")
self.assertEqual(errors, [])
self.assertEqual(len(ids), 2)
def test_conflicting_values_and_names(self):
for text in ("#define ETCP_RT_ID_A 0x40\n#define ETCP_RT_ID_B 0x40\n",
"#define ETCP_ID_A 1\n#define ETCP_ID_A 2\n"):
self.assertTrue(registry(text)[1])
def test_range_and_alias(self):
for literal in ("-1", "256", "ETCP_ID_OTHER"):
self.assertTrue(registry(f"#define ETCP_ID_A {literal}\n")[1])
def test_wrong_binding_and_missing_declaration(self):
ids = {"ETCP_ID_A": 1, "ETCP_RT_ID_B": 2}
for text in ("etcp_bind(inst, ETCP_RT_ID_B, cb);", "etcp_router_bind(inst, ETCP_ID_A, cb);",
"inst->api_bindings.callbacks[ETCP_RT_ID_B](conn, packet);", "etcp_bind(inst, 7, cb);",
"inst->router_bindings.callbacks[0](conn, packet);",
"packet->cmd = ETCP_ID_MISSING;", "#define ETCP_ID_A 1\n"):
self.assertTrue(check_source(text, ids), text)
self.assertFalse(check_source("etcp_bind(inst, ETCP_ID_A, cb); etcp_router_bind(inst, ETCP_RT_ID_B, cb);", ids))
def test_repository(self):
ids, errors = registry((ROOT / HEADER).read_text())
self.assertTrue(ids, f"empty registry in {HEADER}")
checked = 0
for path in sources():
errors += [f"{path.relative_to(ROOT)}: {error}" for error in check_source(path.read_text(), ids)]
checked += 1
self.assertEqual(errors, [], "\n" + "\n".join(errors))
raw = sum(not name.startswith("ETCP_RT_ID_") for name in ids)
print(f"[INFO] service ID audit passed: raw={raw} router={len(ids) - raw} source_files={checked}", flush=True)
if __name__ == "__main__":
unittest.main()
tools/chatgui-android/libutun_lite/instance_lite.c
-5
View File
@@ -13,7 +13,6 @@
#include "config_parser.h"
#include "config_updater.h"
#include "secure_channel.h"
#include "etcp_router.h"
#include "topo_node_sqlite.h"
#include "chat/chat_core.h"
#include "chat/chat_sync.h"
@@ -319,9 +318,6 @@ static void* instance_thread(void* arg) {
chat_core_sync_my_addresses(g_inst);
fire_local_sockets_event();
/* Bind chat sync via etcp_router */
etcp_router_bind(g_inst, ETCP_RT_ID_CHAT_SYNC, NULL);
utun_add_nodeinfo_cbk(g_inst, nodeinfo_event_cb, NULL);
etcp_add_link_status_cbk(g_inst, on_link_status_changed, NULL);
@@ -399,7 +395,6 @@ static void* instance_thread(void* arg) {
call_video_port_attach(g_inst);
fire_local_sockets_event();
if (g_inst->config->global.name[0]) chat_core_update_my_name(g_inst, g_inst->config->global.name);
etcp_router_bind(g_inst, ETCP_RT_ID_CHAT_SYNC, NULL);
utun_add_nodeinfo_cbk(g_inst, nodeinfo_event_cb, NULL);
etcp_add_link_status_cbk(g_inst, on_link_status_changed, NULL);
chat_event_post(g_inst, CHAT_EVT_KEYS_GENERATED, (const uint8_t*)config->global.my_public_key_hex, 64);