#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_RELAY_CANCEL) { 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; } /* Неизвестное измерение явно отличается от нулевого lag/rate; числа не сужаются до старых uint16 RTT. */ void stream_metric_write(uint8_t *wire, const struct stream_metric_snapshot *s, int valid) { wire[0] = valid ? 1 | (s->source_rate_valid ? 2 : 0) | (s->receive_rate_valid ? 4 : 0) | (s->stopped ? 8 : 0) : 0; stream_write_be(wire + 1, s->lag_ms, 8); stream_write_be(wire + 9, s->source_rate_bytes_sec, 8); stream_write_be(wire + 17, s->receive_rate_bytes_sec, 8); stream_write_be(wire + 25, s->blocks, 8); stream_write_be(wire + 33, s->late_blocks, 8); stream_write_be(wire + 41, s->stops, 8); } int stream_metric_read(const uint8_t *wire, struct stream_metric_snapshot *s, int *valid) { if (wire[0] & ~15u || (!(wire[0] & 1) && wire[0])) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream metric flags invalid=%u", wire[0]); return STREAM_ERR_FORMAT; } *s = (struct stream_metric_snapshot){.lag_ms = stream_read_be(wire + 1, 8), .source_rate_bytes_sec = stream_read_be(wire + 9, 8), .receive_rate_bytes_sec = stream_read_be(wire + 17, 8), .blocks = stream_read_be(wire + 25, 8), .late_blocks = stream_read_be(wire + 33, 8), .stops = stream_read_be(wire + 41, 8), .source_rate_valid = !!(wire[0] & 2), .receive_rate_valid = !!(wire[0] & 4), .stopped = !!(wire[0] & 8)}; *valid = !!(wire[0] & 1); return STREAM_OK; } /* 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; }