From 7c8470f44a95122f28f85bb9a04019b168841b91 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 20:03:08 +0200 Subject: [PATCH] Define variable stream parts with a signed RS batch envelope --- src/stream/stream_block.h | 36 ++++++++++------ src/stream/stream_proto.h | 13 ++++-- tests/test_reed_solomon.c | 90 +++++++++++++++++++++++++++++---------- 3 files changed, 101 insertions(+), 38 deletions(-) diff --git a/src/stream/stream_block.h b/src/stream/stream_block.h index 4822b60e..f7a8b3ef 100644 --- a/src/stream/stream_block.h +++ b/src/stream/stream_block.h @@ -1,7 +1,8 @@ -/* Формирование и проверка подписанных FEC-блоков без сокетов и внешних callbacks. +/* Одна Ed25519-подпись пачки до RS, части переменного размера без отдельных подписей. * Контексты одного uasync-потока; descriptor и ключ источника уже проверены членством группы. * encode возвращает заимствованные блоки до следующего encode/destroy; ошибочные результаты не публикуют. - * verify не меняет выходной view при отказе. Ключ непосредственного ретранслятора здесь не используется. */ + * parse проверяет только формат части; подпись источника проверяется после RS в batch_verify. + * Заимствованные части действуют до следующего encode/destroy; ошибочный результат не публикуют. */ #ifndef STREAM_BLOCK_H #define STREAM_BLOCK_H @@ -13,17 +14,23 @@ extern "C" { struct stream_encoder; -struct stream_block_view { +struct stream_batch_info { uint64_t seq; uint32_t data_length; uint16_t timestamp_ms; - uint8_t subblock; + uint16_t block_size; /* Фактический размер RS-кодограммы одной пачки. */ uint8_t flags; - const uint8_t *rs; /* Заимствованный диапазон descriptor.block_size байт. */ +}; + +struct stream_block_view { + struct stream_batch_info batch; + uint8_t subblock; + const uint8_t *rs; /* Заимствованный диапазон batch.block_size байт. */ }; int stream_descriptor_validate(const struct stream_descriptor *descriptor); -size_t stream_block_size(const struct stream_descriptor *descriptor); +size_t stream_block_capacity(const struct stream_descriptor *descriptor); +/* Максимальная длина содержимого без подписанных метаданных/подписи. */ size_t stream_input_size(const struct stream_descriptor *descriptor); /* out заранее NULL; ключ копируется и очищается при destroy. */ @@ -34,13 +41,18 @@ void stream_encoder_destroy(struct stream_encoder *encoder); const uint8_t *stream_encoder_public_key(const struct stream_encoder *encoder); /* blocks — массив descriptor.n указателей; input не перекрывается с рабочими буферами encoder. - * 0 < length <= stream_input_size; остаток пачки дополняется нулями. seq != UINT64_MAX. - * При ошибке blocks не изменяется; собственные рабочие буферы могут измениться. */ + * 0 < length <= stream_input_size; подпись входит в RS, дополнение не превышает n-r-1 байт. + * block_length получает общий фактический размер частей. seq != UINT64_MAX. + * При ошибке blocks/block_length не изменяются; собственная рабочая память может измениться. */ int stream_encoder_encode(struct stream_encoder *encoder, uint64_t seq, uint16_t timestamp_ms, uint8_t flags, - const uint8_t *input, size_t length, const uint8_t **blocks); - -int stream_block_verify(const struct stream_descriptor *descriptor, const uint8_t public_key[SC_PUBKEY_SIZE], - const uint8_t *wire, size_t length, struct stream_block_view *view); + const uint8_t *input, size_t length, const uint8_t **blocks, size_t *block_length); + +/* Формат, дескриптор и канонический размер; криптографической проверки части нет. view не меняется при отказе. */ +int stream_block_parse(const struct stream_descriptor *descriptor, const uint8_t *wire, size_t length, struct stream_block_view *view); +/* После RS сверить подписанные поля с ожидаемой пачкой, нулевой хвост и одну подпись источника. + * decoded_length — полный восстановленный размер с padding; содержимое начинается с STREAM_BATCH_HEADER. */ +int stream_batch_verify(const struct stream_descriptor *descriptor, const uint8_t public_key[SC_PUBKEY_SIZE], + const struct stream_batch_info *expected, const uint8_t *decoded, size_t decoded_length); #ifdef __cplusplus } diff --git a/src/stream/stream_proto.h b/src/stream/stream_proto.h index f75570a6..6751e2a9 100644 --- a/src/stream/stream_proto.h +++ b/src/stream/stream_proto.h @@ -1,4 +1,4 @@ -/* Неизменяемый блок источника: все числа big-endian, подпись охватывает тело до signature. +/* Части RS имеют непроверенный внешний заголовок; подписанная пачка находится внутри RS. * Данные ETCP-выдачи (handle/сессия/путь) добавляются снаружи и в подпись не входят. * seq/subblock берутся из заголовка RS; второго независимого номера блока нет. */ #ifndef STREAM_PROTO_H @@ -10,7 +10,7 @@ #include "../../lib/reed_solomon.h" #include "../transport_layer/secure_channel.h" -#define STREAM_BLOCK_DOMAIN "stream_block" +#define STREAM_BLOCK_DOMAIN "stream_part1" #define STREAM_BLOCK_DOMAIN_SIZE 12 #define STREAM_BLOCK_GROUP_OFF 12 #define STREAM_BLOCK_SOURCE_OFF 20 @@ -24,7 +24,11 @@ #define STREAM_BLOCK_RS_OFF 47 #define STREAM_BLOCK_SEQ_OFF STREAM_BLOCK_RS_OFF #define STREAM_BLOCK_SUB_OFF (STREAM_BLOCK_RS_OFF + 8) -#define STREAM_BLOCK_OVERHEAD (STREAM_BLOCK_RS_OFF + SC_SIGN_SIZE) +#define STREAM_BLOCK_OVERHEAD STREAM_BLOCK_RS_OFF +#define STREAM_BATCH_DOMAIN "stream_batch" +#define STREAM_BATCH_SEQ_OFF STREAM_BLOCK_RS_OFF +#define STREAM_BATCH_HEADER (STREAM_BATCH_SEQ_OFF + 8) +#define STREAM_BATCH_OVERHEAD (STREAM_BATCH_HEADER + SC_SIGN_SIZE) #define STREAM_BLOCK_DEFAULT_SIZE 1300 #define STREAM_RS_DEFAULT_SIZE (STREAM_BLOCK_DEFAULT_SIZE - STREAM_BLOCK_OVERHEAD) #define STREAM_BLOCK_MAX_SIZE (RS_MAX_BLOCK_SIZE + STREAM_BLOCK_OVERHEAD) @@ -38,7 +42,7 @@ struct stream_descriptor { uint64_t group_id; uint64_t source_node_id; uint64_t source_epoch; - uint16_t block_size; /* Полный размер RS-кодограммы, без stream-метаданных/подписи. */ + uint16_t block_size; /* Максимальный размер RS-кодограммы; фактический размер задан в каждой пачке. */ uint8_t n; uint8_t r; }; @@ -49,6 +53,7 @@ enum stream_result { STREAM_WAIT = 2, STREAM_GAP = 3, STREAM_EXPIRED = 4, + STREAM_REJECTED = 5, /* Пачка восстановлена, но подпись/подписанные метаданные не прошли проверку. */ STREAM_ERR_ARGUMENT = -1, STREAM_ERR_MEMORY = -2, STREAM_ERR_FORMAT = -3, diff --git a/tests/test_reed_solomon.c b/tests/test_reed_solomon.c index 8e4e9dd9..5ac388d2 100644 --- a/tests/test_reed_solomon.c +++ b/tests/test_reed_solomon.c @@ -68,7 +68,7 @@ static void fixture_init(struct fixture *f, unsigned n, unsigned r, size_t block f->k = n - r; f->block_size = block_size; CHECK(rs_init(&f->ctx, n, r, block_size) == RS_OK); - f->input_size = rs_input_size(f->ctx); + f->input_size = rs_input_capacity(f->ctx); CHECK(f->input_size == f->k * (block_size - RS_HEADER_SIZE)); f->storage_size = n * (block_size + 2 * GUARD); f->input_storage = u_malloc((uint32_t)(f->input_size + 2 * GUARD)); @@ -93,7 +93,8 @@ static void fixture_init(struct fixture *f, unsigned n, unsigned r, size_t block f->received[i] = f->blocks[i]; } size_t allocations = u_get_allocated_count(); - CHECK(rs_encode(f->ctx, seq, f->input, f->blocks) == RS_OK); + size_t encoded_size = 0; + CHECK(rs_encode(f->ctx, seq, f->input, f->input_size, f->blocks, &encoded_size) == RS_OK && encoded_size == block_size); CHECK(u_get_allocated_count() == allocations); memcpy(f->saved, f->storage, f->storage_size); for (unsigned i = 0; i < n; i++) { @@ -118,7 +119,7 @@ static void fixture_free(struct fixture *f) { static void check_decode(struct fixture *f) { size_t allocations = u_get_allocated_count(); memset(f->output, MARKER, f->input_size); - CHECK(rs_decode(f->ctx, f->received, f->output) == RS_OK); + CHECK(rs_decode(f->ctx, f->received, f->block_size, f->output, f->input_size) == RS_OK); CHECK(u_get_allocated_count() == allocations); CHECK(memcmp(f->input, f->output, f->input_size) == 0); CHECK(memcmp(f->storage, f->saved, f->storage_size) == 0); @@ -191,7 +192,8 @@ static void test_known_vector(void) { fixture_init(&f, 4, 2, 10, 7); f.input[0] = 1; f.input[1] = 2; - CHECK(rs_encode(f.ctx, 7, f.input, f.blocks) == RS_OK); + size_t encoded_size = 0; + CHECK(rs_encode(f.ctx, 7, f.input, f.input_size, f.blocks, &encoded_size) == RS_OK && encoded_size == f.block_size); CHECK(f.blocks[2][9] == 4 && f.blocks[3][9] == 8); memcpy(f.saved, f.storage, f.storage_size); f.received[0] = NULL; @@ -246,40 +248,51 @@ static void test_errors(void) { for (unsigned i = 0; i < sizeof(sizes) / sizeof(sizes[0]); i++) { CHECK(rs_init(&ctx, 16, 2, sizes[i]) == RS_ERR_INVALID_ARG && !ctx); } - CHECK(rs_input_size(NULL) == 0); + CHECK(rs_input_capacity(NULL) == 0); rs_free(NULL); struct fixture f; fixture_init(&f, 16, 2, 23, 100); - CHECK(rs_encode(NULL, 100, f.input, f.blocks) == RS_ERR_INVALID_ARG); - CHECK(rs_encode(f.ctx, 100, NULL, f.blocks) == RS_ERR_INVALID_ARG); - CHECK(rs_encode(f.ctx, 100, f.input, NULL) == RS_ERR_INVALID_ARG); + size_t encoded_size = 123; + CHECK(rs_encode(NULL, 100, f.input, f.input_size, f.blocks, &encoded_size) == RS_ERR_INVALID_ARG); + CHECK(rs_encode(f.ctx, 100, NULL, f.input_size, f.blocks, &encoded_size) == RS_ERR_INVALID_ARG); + CHECK(rs_encode(f.ctx, 100, f.input, f.input_size, NULL, &encoded_size) == RS_ERR_INVALID_ARG); + CHECK(rs_encode(f.ctx, 100, f.input, 0, f.blocks, &encoded_size) == RS_ERR_INVALID_ARG); + CHECK(rs_encode(f.ctx, 100, f.input, f.input_size + 1, f.blocks, &encoded_size) == RS_ERR_INVALID_ARG); + CHECK(rs_encode(f.ctx, 100, f.input, SIZE_MAX, f.blocks, &encoded_size) == RS_ERR_INVALID_ARG); + CHECK(rs_encode(f.ctx, 100, f.input, f.input_size, f.blocks, NULL) == RS_ERR_INVALID_ARG); + CHECK(encoded_size == 123); f.blocks[15] = NULL; - CHECK(rs_encode(f.ctx, 101, f.input, f.blocks) == RS_ERR_INVALID_ARG); + CHECK(rs_encode(f.ctx, 101, f.input, f.input_size, f.blocks, &encoded_size) == RS_ERR_INVALID_ARG && encoded_size == 123); CHECK(memcmp(f.storage, f.saved, f.storage_size) == 0); f.blocks[15] = (uint8_t *)f.received[15]; - CHECK(rs_decode(NULL, f.received, f.output) == RS_ERR_INVALID_ARG); - CHECK(rs_decode(f.ctx, NULL, f.output) == RS_ERR_INVALID_ARG); - CHECK(rs_decode(f.ctx, f.received, NULL) == RS_ERR_INVALID_ARG); + CHECK(rs_decode(NULL, f.received, f.block_size, f.output, f.input_size) == RS_ERR_INVALID_ARG); + CHECK(rs_decode(f.ctx, NULL, f.block_size, f.output, f.input_size) == RS_ERR_INVALID_ARG); + CHECK(rs_decode(f.ctx, f.received, f.block_size, NULL, f.input_size) == RS_ERR_INVALID_ARG); + CHECK(rs_decode(f.ctx, f.received, RS_HEADER_SIZE, f.output, f.input_size) == RS_ERR_INVALID_ARG); + CHECK(rs_decode(f.ctx, f.received, f.block_size + 1, f.output, f.input_size) == RS_ERR_INVALID_ARG); + CHECK(rs_decode(f.ctx, f.received, SIZE_MAX, f.output, f.input_size) == RS_ERR_INVALID_ARG); + CHECK(rs_decode(f.ctx, f.received, f.block_size, f.output, f.input_size - 1) == RS_ERR_INVALID_ARG); + check_unchanged_output(&f); f.received[0] = f.received[1] = f.received[2] = NULL; - CHECK(rs_decode(f.ctx, f.received, f.output) == RS_ERR_INSUFFICIENT); + CHECK(rs_decode(f.ctx, f.received, f.block_size, f.output, f.input_size) == RS_ERR_INSUFFICIENT); check_unchanged_output(&f); for (unsigned i = 0; i < 16; i++) f.received[i] = NULL; - CHECK(rs_decode(f.ctx, f.received, f.output) == RS_ERR_INSUFFICIENT); + CHECK(rs_decode(f.ctx, f.received, f.block_size, f.output, f.input_size) == RS_ERR_INSUFFICIENT); check_unchanged_output(&f); for (unsigned i = 0; i < 16; i++) f.received[i] = f.blocks[i]; f.blocks[15][0] ^= 1; - CHECK(rs_decode(f.ctx, f.received, f.output) == RS_ERR_HEADER); + CHECK(rs_decode(f.ctx, f.received, f.block_size, f.output, f.input_size) == RS_ERR_HEADER); check_unchanged_output(&f); f.blocks[15][0] ^= 1; f.blocks[15][8] = 0; - CHECK(rs_decode(f.ctx, f.received, f.output) == RS_ERR_HEADER); + CHECK(rs_decode(f.ctx, f.received, f.block_size, f.output, f.input_size) == RS_ERR_HEADER); check_unchanged_output(&f); f.blocks[15][8] = 255; - CHECK(rs_decode(f.ctx, f.received, f.output) == RS_ERR_HEADER); + CHECK(rs_decode(f.ctx, f.received, f.block_size, f.output, f.input_size) == RS_ERR_HEADER); f.blocks[15][8] = 15; f.received[0] = f.blocks[1]; f.received[1] = f.blocks[0]; - CHECK(rs_decode(f.ctx, f.received, f.output) == RS_ERR_HEADER); + CHECK(rs_decode(f.ctx, f.received, f.block_size, f.output, f.input_size) == RS_ERR_HEADER); check_unchanged_output(&f); f.received[0] = f.blocks[0]; f.received[1] = f.blocks[1]; @@ -288,7 +301,38 @@ static void test_errors(void) { puts("[PASS] reed_solomon: invalid parameters, missing buffers, insufficient blocks, mixed seq and invalid indices"); } -/* Приложение само разбивает поток, дополняет хвост и собирает пачки с учётом seq. */ +/* Один контекст принимает разные длины; вход точно ограничен, выход не затрагивает хвосты буферов. */ +static void test_variable_lengths(void) { + struct fixture f; + fixture_init(&f, 8, 2, 4096, 0); + const size_t lengths[] = {1, 5, 6, 7, 17, 97, 1000, 7080, 24521, 24522, 1}; + for (unsigned step = 0; step < sizeof(lengths) / sizeof(lengths[0]); step++) { + size_t length = lengths[step], size = 0; + uint8_t *input = u_malloc((uint32_t)length); + CHECK(input); + memcpy(input, f.input, length); + memset(f.storage, MARKER, f.storage_size); + memset(f.output, MARKER, f.input_size); + size_t allocations = u_get_allocated_count(); + CHECK(rs_encode(f.ctx, step, input, length, f.blocks, &size) == RS_OK); + CHECK(size == RS_HEADER_SIZE + (length + f.k - 1) / f.k); + for (unsigned i = 0; i < f.n; i++) { + for (size_t b = size; b < f.block_size; b++) CHECK(f.blocks[i][b] == MARKER); + f.received[i] = i == 0 || i == 3 ? NULL : f.blocks[i]; + } + size_t padded = f.k * (size - RS_HEADER_SIZE); + CHECK(rs_decode(f.ctx, f.received, size, f.output, padded) == RS_OK); + CHECK(!memcmp(input, f.output, length)); + for (size_t i = length; i < padded; i++) CHECK(f.output[i] == 0); + for (size_t i = padded; i < f.input_size; i++) CHECK(f.output[i] == MARKER); + CHECK(u_get_allocated_count() == allocations); + u_free(input); + } + fixture_free(&f); + puts("[PASS] reed_solomon: changing batch lengths, exact input bounds, minimal zero padding, bounded writes, no per-call allocations"); +} + +/* Приложение разбивает поток, кодек уменьшает хвостовую пачку, сборка сохраняет исходную длину. */ static void test_stream(void) { struct fixture f; fixture_init(&f, 16, 2, 101, 0); @@ -301,9 +345,9 @@ static void test_stream(void) { for (uint64_t seq = 0; offset < length; seq++) { size_t valid = length - offset; if (valid > f.input_size) valid = f.input_size; - memset(f.input, 0, f.input_size); memcpy(f.input, source + offset, valid); - CHECK(rs_encode(f.ctx, seq, f.input, f.blocks) == RS_OK); + size_t encoded_size = 0; + CHECK(rs_encode(f.ctx, seq, f.input, valid, f.blocks, &encoded_size) == RS_OK); memcpy(f.saved, f.storage, f.storage_size); for (unsigned i = 0; i < f.n; i++) f.received[i] = NULL; for (unsigned i = f.n; i > 0; i--) { @@ -311,7 +355,8 @@ static void test_stream(void) { if (index == seq % f.n || index == (seq + 1) % f.n) continue; f.received[index] = f.blocks[i - 1]; } - check_decode(&f); + CHECK(rs_decode(f.ctx, f.received, encoded_size, f.output, f.input_size) == RS_OK); + CHECK(!memcmp(f.input, f.output, valid)); memcpy(result + offset, f.output, valid); offset += valid; } @@ -332,6 +377,7 @@ int main(void) { test_all_sizes(); test_configurations(); test_errors(); + test_variable_lengths(); debug_set_category_level(DEBUG_CATEGORY_GENERAL, DEBUG_LEVEL_TRACE); test_stream(); CHECK(u_get_allocated_count() == baseline);