From 18fe51ad25ecf8a200966232e0851f3cf3025577 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 00:35:53 +0200 Subject: [PATCH] Define direct stream messages, paths and signed completion --- src/stream/stream_wire.c | 82 ++++++++++++++++++++++++++++++++++ src/stream/stream_wire.h | 43 ++++++++++++++++++ src/transport_layer/etcp_api.h | 6 ++- 3 files changed, 129 insertions(+), 2 deletions(-) create mode 100644 src/stream/stream_wire.c create mode 100644 src/stream/stream_wire.h diff --git a/src/stream/stream_wire.c b/src/stream/stream_wire.c new file mode 100644 index 00000000..ae9d264d --- /dev/null +++ b/src/stream/stream_wire.c @@ -0,0 +1,82 @@ +#include + +#include "../../lib/debug_config.h" +#include "../transport_layer/etcp_api.h" +#include "stream_wire.h" + +/* Заголовок одинаков у raw и router payload; cmd raw сохраняется внутри routed-кодограммы. */ +int stream_wire_parse(const uint8_t *wire, size_t length, struct stream_message *m) { + if (!wire || !m || length < STREAM_WIRE_HEADER || wire[0] != ETCP_ID_STREAM || wire[1] != STREAM_WIRE_VERSION + || wire[2] < STREAM_CMD_INFO_REQ || wire[2] > STREAM_CMD_REVERSE_REQ) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream wire header invalid bytes=%zu", length); return STREAM_ERR_FORMAT; + } + struct stream_message result = {.command = wire[2], .subblock = wire[3], .payload = wire + STREAM_WIRE_HEADER, + .length = length - STREAM_WIRE_HEADER}; + result.descriptor.group_id = stream_read_be(wire + 4, 8); + result.descriptor.source_node_id = stream_read_be(wire + 12, 8); + result.descriptor.source_epoch = stream_read_be(wire + 20, 8); + result.sender_epoch = stream_read_be(wire + 28, 8); result.receiver_epoch = stream_read_be(wire + 36, 8); + result.handle = stream_read_be(wire + 44, 8); + if (!result.descriptor.group_id || !result.descriptor.source_node_id || !result.descriptor.source_epoch || !result.sender_epoch) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream wire zero key/epoch command=%u", result.command); return STREAM_ERR_FORMAT; + } + *m = result; return STREAM_OK; +} + +void stream_wire_header(uint8_t *wire, const struct stream_message *m) { + wire[0] = ETCP_ID_STREAM; wire[1] = STREAM_WIRE_VERSION; wire[2] = m->command; wire[3] = m->subblock; + stream_write_be(wire + 4, m->descriptor.group_id, 8); stream_write_be(wire + 12, m->descriptor.source_node_id, 8); + stream_write_be(wire + 20, m->descriptor.source_epoch, 8); stream_write_be(wire + 28, m->sender_epoch, 8); + stream_write_be(wire + 36, m->receiver_epoch, 8); stream_write_be(wire + 44, m->handle, 8); +} + +/* История блока проверяется отдельно от BGP-пути; получатель не должен уже встречаться в ней. */ +int stream_path_validate(const struct stream_path *path, uint64_t source, uint64_t sender, uint64_t forbidden) { + if (!path || !path->count || path->count > STREAM_PATH_MAX || path->nodes[0] != source + || path->nodes[path->count - 1] != sender) goto invalid; + for (unsigned i = 0; i < path->count; i++) { + if (!path->nodes[i] || (forbidden && path->nodes[i] == forbidden)) goto invalid; + for (unsigned j = 0; j < i; j++) if (path->nodes[i] == path->nodes[j]) goto invalid; + } + return STREAM_OK; +invalid: + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream path invalid/loop source=%llx sender=%llx receiver=%llx count=%u", + (unsigned long long)source, (unsigned long long)sender, (unsigned long long)forbidden, path ? path->count : 0); + return STREAM_ERR_FORMAT; +} + +int stream_path_parse(const uint8_t *wire, size_t length, struct stream_path *path, size_t *used) { + if (!wire || !path || !used || !length || !wire[0] || wire[0] > STREAM_PATH_MAX || length < 1u + wire[0] * 8u) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream path size invalid bytes=%zu", length); return STREAM_ERR_FORMAT; + } + struct stream_path result = {.count = wire[0]}; + for (unsigned i = 0; i < result.count; i++) result.nodes[i] = stream_read_be(wire + 1 + i * 8, 8); + *path = result; *used = 1 + result.count * 8; return STREAM_OK; +} + +size_t stream_path_write(uint8_t *wire, const struct stream_path *path) { + wire[0] = path->count; + for (unsigned i = 0; i < path->count; i++) stream_write_be(wire + 1 + i * 8, path->nodes[i], 8); + return 1 + path->count * 8; +} + +/* END имеет собственный domain и подписывает ключ запуска + последний индекс; UINT64_MAX означает пустой поток. */ +int stream_end_sign(const struct stream_descriptor *d, uint64_t last, const uint8_t key[32], uint8_t wire[STREAM_END_SIZE]) { + if (!d || !key || !wire) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream END sign invalid arguments"); return STREAM_ERR_ARGUMENT; } + memcpy(wire, "stream_end", 10); stream_write_be(wire + 10, d->group_id, 8); + stream_write_be(wire + 18, d->source_node_id, 8); stream_write_be(wire + 26, d->source_epoch, 8); + stream_write_be(wire + 34, last, 8); + if (sc_ed25519_sign(key, wire, 42, wire + 42) != SC_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream END signature failed"); return STREAM_ERR_CRYPTO; + } + return STREAM_OK; +} + +int stream_end_verify(const struct stream_descriptor *d, const uint8_t key[32], const uint8_t *wire, size_t length, uint64_t *last) { + if (!d || !key || !wire || !last || length != STREAM_END_SIZE || memcmp(wire, "stream_end", 10) + || stream_read_be(wire + 10, 8) != d->group_id || stream_read_be(wire + 18, 8) != d->source_node_id + || stream_read_be(wire + 26, 8) != d->source_epoch || sc_ed25519_verify(key, wire, 42, wire + 42) != SC_OK) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream END invalid signature/key bytes=%zu", length); return STREAM_ERR_SIGNATURE; + } + *last = stream_read_be(wire + 34, 8); return STREAM_OK; +} diff --git a/src/stream/stream_wire.h b/src/stream/stream_wire.h new file mode 100644 index 00000000..f4dad5ea --- /dev/null +++ b/src/stream/stream_wire.h @@ -0,0 +1,43 @@ +/* Прямой stream-протокол: канонический big-endian заголовок, пути и подписанный END. + * IDs ETCP определены только в etcp_api.h. Указатели parser заимствованы до освобождения кодограммы. */ +#ifndef STREAM_WIRE_H +#define STREAM_WIRE_H + +#include "stream_proto.h" + +#define STREAM_WIRE_VERSION 1 +#define STREAM_WIRE_HEADER 52 +#define STREAM_PATH_MAX 32 +#define STREAM_END_SIZE 106 +#define STREAM_INFO_SIZE 21 + +enum stream_command { + STREAM_CMD_INFO_REQ = 1, STREAM_CMD_INFO_RESP, STREAM_CMD_SUBSCRIBE, STREAM_CMD_ACCEPT, + STREAM_CMD_REJECT, STREAM_CMD_RENEW, STREAM_CMD_UNSUBSCRIBE, STREAM_CMD_CLOSED, + STREAM_CMD_DATA, STREAM_CMD_GAP, STREAM_CMD_END, STREAM_CMD_REVERSE_REQ +}; +enum stream_reject { + STREAM_REASON_STOPPED = 1, STREAM_REASON_POSITION, STREAM_REASON_EXPIRED, STREAM_REASON_STALE, + STREAM_REASON_CONFLICT, STREAM_REASON_LIMIT, STREAM_REASON_LOOP, STREAM_REASON_TRANSPORT, + STREAM_REASON_LEASE, STREAM_REASON_MEMBER, STREAM_REASON_PROTOCOL +}; +struct stream_path { uint64_t nodes[STREAM_PATH_MAX]; uint8_t count; }; +struct stream_message { + struct stream_descriptor descriptor; + uint64_t sender_epoch, receiver_epoch, handle; + uint8_t command, subblock; + const uint8_t *payload; + size_t length; +}; + +/* Проверяет размер, версию, ненулевые ключи/epoch и известный command; payload разбирает диспетчер. */ +int stream_wire_parse(const uint8_t *wire, size_t length, struct stream_message *message); +void stream_wire_header(uint8_t *wire, const struct stream_message *message); +/* Путь без нулей/повторов, начинается источником и заканчивается отправителем; forbidden не входит в него. */ +int stream_path_validate(const struct stream_path *path, uint64_t source, uint64_t sender, uint64_t forbidden); +int stream_path_parse(const uint8_t *wire, size_t length, struct stream_path *path, size_t *used); +size_t stream_path_write(uint8_t *wire, const struct stream_path *path); +int stream_end_sign(const struct stream_descriptor *descriptor, uint64_t last, const uint8_t key[32], uint8_t wire[STREAM_END_SIZE]); +int stream_end_verify(const struct stream_descriptor *descriptor, const uint8_t key[32], const uint8_t *wire, size_t length, + uint64_t *last); +#endif diff --git a/src/transport_layer/etcp_api.h b/src/transport_layer/etcp_api.h index 0aa2cf7d..67555a44 100644 --- a/src/transport_layer/etcp_api.h +++ b/src/transport_layer/etcp_api.h @@ -38,12 +38,12 @@ extern "C" { // (raw-пространство) — помечены «(raw)»; их значения занимают raw-реестр. // // raw: 0x00 DATA, 0x01 TOPO_ENTRY, 0x02 NAT_DETECTION, 0x03 SVC_ROUTE, -// 0x04 NTP_TIME, 0x05 BROADCAST, 0x06 RADIO, 0x12 NCD_CONTROL, +// 0x04 NTP_TIME, 0x05 BROADCAST, 0x06 RADIO, 0x07 STREAM, 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 +// 0x35 DM_MAILBOX, 0x36 CALL, 0x37 FILE_TRANSFER, 0x38 DM_MEDIA, 0x39 STREAM // // !!! Новый ID: впиши его в нужный реестр выше и убедись, что значение // не занято в том же пространстве. @@ -57,6 +57,8 @@ extern "C" { #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_RT_ID_STREAM 0x39 // stream REVERSE_REQ/CLOSED через группу // === Router-сервисы (etcp_router_bind/etcp_route_send) === #define ETCP_RT_ID_DATA 0x00 // routing.c — маршрутизация данных