Define variable stream parts with a signed RS batch envelope

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.
+ * 0 < length <= stream_input_size; подпись входит в RS, дополнение не превышает n-r-1 байт.
- * При ошибке blocks не изменяется; собственные рабочие буферы могут измениться. */
+ * 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);
+                          const uint8_t *input, size_t length, const uint8_t **blocks, size_t *block_length);
 
-int stream_block_verify(const struct stream_descriptor *descriptor, const uint8_t public_key[SC_PUBKEY_SIZE],
+/* Формат, дескриптор и канонический размер; криптографической проверки части нет. view не меняется при отказе. */
-                        const uint8_t *wire, size_t length, struct stream_block_view *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
 }

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,

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);
+    size_t encoded_size = 123;
-    CHECK(rs_encode(f.ctx, 100, NULL, f.blocks) == RS_ERR_INVALID_ARG);
+    CHECK(rs_encode(NULL, 100, f.input, f.input_size, f.blocks, &encoded_size) == RS_ERR_INVALID_ARG);
-    CHECK(rs_encode(f.ctx, 100, f.input, NULL) == 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(NULL, f.received, f.block_size, f.output, f.input_size) == RS_ERR_INVALID_ARG);
-    CHECK(rs_decode(f.ctx, NULL, f.output) == 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, NULL) == 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);