From e96b813fcb93fcf49e8653555991efe6f766e427 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 20:13:45 +0200 Subject: [PATCH] Test batch authentication after losses and recovery after rejection --- tests/test_stream_core.c | 222 ++++++++++++++++++++++++++++-------- tests/test_stream_service.c | 39 +++++-- 2 files changed, 200 insertions(+), 61 deletions(-) diff --git a/tests/test_stream_core.c b/tests/test_stream_core.c index c9d59db2..ab13f5fc 100644 --- a/tests/test_stream_core.c +++ b/tests/test_stream_core.c @@ -23,7 +23,8 @@ static const struct stream_descriptor descriptor = { static uint8_t private_key[32]; static unsigned cache_added, cache_removed; static void cache_observer(void *arg, const uint8_t *wire, size_t length, int added) { - (void)arg; CHECK(length == 1300 && stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8) == descriptor.group_id); + (void)arg; CHECK(length > STREAM_BLOCK_OVERHEAD + RS_HEADER_SIZE && length <= STREAM_BLOCK_DEFAULT_SIZE + && stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8) == descriptor.group_id); if (added) cache_added++; else cache_removed++; } @@ -70,54 +71,85 @@ static void pattern(uint8_t *data, size_t length) { } } -/* Подмена любого класса неизменяемых полей отвергается, view и входной блок сохраняются. */ +/* Подпись находится внутри RS; внешние поля проверяются структурно, подмена обнаруживается после восстановления. */ static void test_signed_blocks(void) { struct stream_encoder *encoder = NULL; CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); - CHECK(stream_block_size(&descriptor) == 1300 && stream_input_size(&descriptor) == 7080); - uint8_t input[7080]; pattern(input, sizeof(input)); + CHECK(stream_block_capacity(&descriptor) == 1300 && stream_input_size(&descriptor) == STREAM_DEFAULT_INPUT_CAPACITY); + uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input)); const uint8_t *blocks[RS_MAX_BLOCKS]; uint64_t seq = UINT64_C(0x1122334455667788); - CHECK(stream_encoder_encode(encoder, seq, 65530, STREAM_BLOCK_TIME_RESET, input, sizeof(input), blocks) == STREAM_OK); + size_t size = 0; + CHECK(stream_encoder_encode(encoder, seq, 65530, STREAM_BLOCK_TIME_RESET, input, sizeof(input), blocks, &size) == STREAM_OK); + CHECK(size == 1300); const uint8_t *public_key = stream_encoder_public_key(encoder); + struct stream_block_view view; + const uint8_t *received[RS_MAX_BLOCKS] = {0}; for (unsigned i = 0; i < descriptor.n; i++) { - struct stream_block_view view; - CHECK(stream_block_verify(&descriptor, public_key, blocks[i], 1300, &view) == STREAM_OK); - CHECK(view.seq == seq && view.subblock == i && view.timestamp_ms == 65530 && view.data_length == sizeof(input)); - CHECK(view.flags == STREAM_BLOCK_TIME_RESET && view.rs == blocks[i] + STREAM_BLOCK_RS_OFF); - CHECK(!memcmp(blocks[i], "stream_block", 12)); + CHECK(stream_block_parse(&descriptor, blocks[i], size, &view) == STREAM_OK); + CHECK(view.batch.seq == seq && view.subblock == i && view.batch.timestamp_ms == 65530 && view.batch.data_length == sizeof(input)); + CHECK(view.batch.flags == STREAM_BLOCK_TIME_RESET && view.rs == blocks[i] + STREAM_BLOCK_RS_OFF); + CHECK(view.batch.block_size == descriptor.block_size); + CHECK(!memcmp(blocks[i], STREAM_BLOCK_DOMAIN, 12)); CHECK(!memcmp(blocks[i] + STREAM_BLOCK_GROUP_OFF, "\x01\x23\x45\x67\x89\xab\xcd\xef", 8)); CHECK(!memcmp(blocks[i] + STREAM_BLOCK_SEQ_OFF, "\x11\x22\x33\x44\x55\x66\x77\x88", 8)); - CHECK(sc_ed25519_verify(public_key, blocks[i], 1300 - 64, blocks[i] + 1300 - 64) == SC_OK); + received[i] = view.rs; } - const unsigned changed[] = {0, 12, 20, 28, 36, 37, 38, 40, 45, 46, 47, 55, 56, 1235, 1299}; + struct rs_codec *rs = NULL; + CHECK(rs_init(&rs, descriptor.n, descriptor.r, descriptor.block_size) == RS_OK); + uint8_t decoded[STREAM_DEFAULT_INPUT_CAPACITY + STREAM_BATCH_OVERHEAD]; + CHECK(rs_decode(rs, received, descriptor.block_size, decoded, sizeof(decoded)) == RS_OK); + CHECK(!memcmp(decoded, STREAM_BATCH_DOMAIN, 12)); + CHECK(!memcmp(decoded + STREAM_BATCH_HEADER, input, sizeof(input))); + CHECK(stream_batch_verify(&descriptor, public_key, &view.batch, decoded, sizeof(decoded)) == STREAM_OK); + CHECK(sc_ed25519_verify(public_key, decoded, STREAM_BATCH_HEADER + sizeof(input), + decoded + STREAM_BATCH_HEADER + sizeof(input)) == SC_OK); + const unsigned changed[] = {0, 12, 20, 28, 36, 37, 38, 44}; uint8_t wire[STREAM_BLOCK_MAX_SIZE]; for (unsigned i = 0; i < sizeof(changed) / sizeof(changed[0]); i++) { - memcpy(wire, blocks[0], 1300); wire[changed[i]] ^= 1; - struct stream_block_view view; memset(&view, 0xa5, sizeof(view)); - uint8_t before[sizeof(view)]; memcpy(before, &view, sizeof(view)); - CHECK(stream_block_verify(&descriptor, public_key, wire, 1300, &view) < 0); - CHECK(!memcmp(before, &view, sizeof(view))); + memcpy(wire, blocks[0], size); wire[changed[i]] ^= 1; + struct stream_block_view parsed; memset(&parsed, 0xa5, sizeof(parsed)); + uint8_t before[sizeof(parsed)]; memcpy(before, &parsed, sizeof(parsed)); + CHECK(stream_block_parse(&descriptor, wire, size, &parsed) < 0); + CHECK(!memcmp(before, &parsed, sizeof(parsed))); } - struct stream_block_view view; - CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 0, &view) == STREAM_ERR_FORMAT); - CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 1299, &view) == STREAM_ERR_FORMAT); - CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 1301, &view) == STREAM_ERR_FORMAT); + CHECK(stream_block_parse(&descriptor, blocks[0], 0, &view) == STREAM_ERR_FORMAT); + CHECK(stream_block_parse(&descriptor, blocks[0], size - 1, &view) == STREAM_ERR_FORMAT); + CHECK(stream_block_parse(&descriptor, blocks[0], size + 1, &view) == STREAM_ERR_FORMAT); uint8_t wrong_key[32]; memcpy(wrong_key, public_key, 32); wrong_key[0] ^= 1; - CHECK(stream_block_verify(&descriptor, wrong_key, blocks[0], 1300, &view) == STREAM_ERR_SIGNATURE); - CHECK(stream_encoder_encode(encoder, UINT64_MAX, 0, 0, input, 1, blocks) == STREAM_ERR_ARGUMENT); - CHECK(stream_encoder_encode(encoder, 1, 0, 0, input, 0, blocks) == STREAM_ERR_ARGUMENT); - CHECK(stream_encoder_encode(encoder, 1, 0, 2, input, 1, blocks) == STREAM_ERR_ARGUMENT); + CHECK(stream_batch_verify(&descriptor, wrong_key, &view.batch, decoded, sizeof(decoded)) == STREAM_ERR_SIGNATURE); + const unsigned signed_fields[] = {0, 12, 20, 28, 36, 37, 38, 40, 45, 46, 47}; + uint8_t modified[sizeof(decoded)]; + for (unsigned i = 0; i < sizeof(signed_fields) / sizeof(signed_fields[0]); i++) { + memcpy(modified, decoded, sizeof(decoded)); modified[signed_fields[i]] ^= 1; + CHECK(stream_batch_verify(&descriptor, public_key, &view.batch, modified, sizeof(modified)) == STREAM_ERR_SIGNATURE); + CHECK(sc_ed25519_sign(private_key, modified, STREAM_BATCH_HEADER + sizeof(input), + modified + STREAM_BATCH_HEADER + sizeof(input)) == SC_OK); + CHECK(stream_batch_verify(&descriptor, public_key, &view.batch, modified, sizeof(modified)) == STREAM_ERR_FORMAT); + } + memcpy(wire, blocks[0], size); wire[STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE + STREAM_BATCH_HEADER] ^= 1; + CHECK(stream_block_parse(&descriptor, wire, size, &view) == STREAM_OK); + received[0] = wire + STREAM_BLOCK_RS_OFF; + CHECK(rs_decode(rs, received, descriptor.block_size, decoded, sizeof(decoded)) == RS_OK); + CHECK(stream_batch_verify(&descriptor, public_key, &view.batch, decoded, sizeof(decoded)) == STREAM_ERR_SIGNATURE); + rs_free(rs); + size = 999; + CHECK(stream_encoder_encode(encoder, UINT64_MAX, 0, 0, input, 1, blocks, &size) == STREAM_ERR_ARGUMENT); + CHECK(stream_encoder_encode(encoder, 1, 0, 0, input, 0, blocks, &size) == STREAM_ERR_ARGUMENT); + CHECK(stream_encoder_encode(encoder, 1, 0, 2, input, 1, blocks, &size) == STREAM_ERR_ARGUMENT); + CHECK(stream_encoder_encode(encoder, 1, 0, 0, input, sizeof(input) + 1, blocks, &size) == STREAM_ERR_ARGUMENT && size == 999); struct stream_descriptor invalid = descriptor; invalid.n = 0; CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT); invalid = descriptor; invalid.r = invalid.n; CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT); invalid = descriptor; invalid.block_size = 9; CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT); + invalid = descriptor; invalid.n = 1; invalid.r = 0; invalid.block_size = RS_HEADER_SIZE + STREAM_BATCH_OVERHEAD; + CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT); invalid = descriptor; invalid.source_epoch = 0; CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT); stream_encoder_destroy(encoder); - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] signed blocks: canonical bytes, all field classes, wrong key and malformed sizes"); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] signed RS batch: canonical bytes, payload/signature recovery, signed metadata and size validation"); } /* Wire сохраняет все 64 бита и не меняет выход при усечении; путь не допускает повторного посещения. */ @@ -148,19 +180,20 @@ static void test_wire(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] stream wire: 64-bit fields, truncation, version, path repeats/receiver loop and size"); } -/* Любая пара потерь при 6+2 восстанавливается из подписанных блоков, отсутствующие блоки не регенерируются. */ +/* Любая пара потерь восстанавливает содержимое и общую подпись, отсутствующие части не регенерируются. */ static void test_fec_and_padding(void) { struct stream_encoder *encoder = NULL; CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); - uint8_t input[7080], output[7080]; pattern(input, sizeof(input)); + uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY], output[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input)); const uint8_t *blocks[RS_MAX_BLOCKS]; - CHECK(stream_encoder_encode(encoder, 0, 15, 0, input, sizeof(input), blocks) == STREAM_OK); + size_t size = 0; + CHECK(stream_encoder_encode(encoder, 0, 15, 0, input, sizeof(input), blocks, &size) == STREAM_OK && size == 1300); for (unsigned first = 0; first < 8; first++) { for (unsigned second = first + 1; second < 8; second++) { struct stream_buffer *buffer = NULL; CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 8 * 1300) == STREAM_OK); for (unsigned i = 0; i < 8; i++) - if (i != first && i != second) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK); + if (i != first && i != second) CHECK(stream_buffer_put(buffer, blocks[i], size) == STREAM_OK); size_t length = 0; memset(output, 0xcc, sizeof(output)); size_t allocations = u_get_allocated_count(); @@ -176,24 +209,24 @@ static void test_fec_and_padding(void) { } struct stream_buffer *buffer = NULL; CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 8 * 1300) == STREAM_OK); - for (unsigned i = 0; i < 5; i++) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK); + for (unsigned i = 0; i < 5; i++) CHECK(stream_buffer_put(buffer, blocks[i], size) == STREAM_OK); memset(output, 0xcc, sizeof(output)); size_t length = 123; CHECK(stream_buffer_decode(buffer, 0, output, sizeof(output), &length) == STREAM_WAIT); CHECK(length == 123); for (unsigned i = 0; i < sizeof(output); i++) CHECK(output[i] == 0xcc); - CHECK(stream_buffer_put(buffer, blocks[5], 1300) == STREAM_OK); + CHECK(stream_buffer_put(buffer, blocks[5], size) == STREAM_OK); CHECK(stream_buffer_decode(buffer, 0, output, 1, &length) == STREAM_ERR_ARGUMENT); stream_buffer_destroy(buffer); stream_encoder_destroy(encoder); - const struct { uint8_t n, r; uint16_t size; } variants[] = {{1, 0, 1189}, {3, 1, 20}, {16, 5, 4096}, {255, 254, 10}}; + const struct { uint8_t n, r; uint16_t size; } variants[] = {{1, 0, 1189}, {3, 1, 70}, {16, 5, 4096}, {255, 254, 130}}; for (unsigned v = 0; v < sizeof(variants) / sizeof(variants[0]); v++) { struct stream_descriptor d = descriptor; d.n = variants[v].n; d.r = variants[v].r; d.block_size = variants[v].size; encoder = NULL; buffer = NULL; CHECK(stream_encoder_create(&encoder, &d, private_key) == STREAM_OK); - size_t size = stream_block_size(&d), wanted = stream_input_size(&d) > 17 ? 17 : stream_input_size(&d); - CHECK(stream_encoder_encode(encoder, 37, 0, 0, input, wanted, blocks) == STREAM_OK); - CHECK(stream_buffer_create(&buffer, &d, stream_encoder_public_key(encoder), d.n * size) == STREAM_OK); + size_t size = 0, wanted = stream_input_size(&d) > 17 ? 17 : stream_input_size(&d); + CHECK(stream_encoder_encode(encoder, 37, 0, 0, input, wanted, blocks, &size) == STREAM_OK); + CHECK(stream_buffer_create(&buffer, &d, stream_encoder_public_key(encoder), d.n * stream_block_capacity(&d)) == STREAM_OK); for (unsigned i = d.r; i < d.n; i++) CHECK(stream_buffer_put(buffer, blocks[i], size) == STREAM_OK); uint8_t partial[17]; length = 0; CHECK(stream_buffer_decode(buffer, 37, partial, wanted, &length) == STREAM_OK); @@ -203,12 +236,102 @@ static void test_fec_and_padding(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] FEC: all 28 double losses, insufficient set, partial data, n=1..255 and r=0"); } -/* Вставлять подписанную пачку в тестовое кольцо до повторного использования буферов encoder. */ +/* Размеры чередуются в одном кольце; каждая пачка восстанавливается без двух исходных частей. */ +static void test_variable_batches(void) { + struct stream_encoder *encoder = NULL; struct stream_buffer *buffer = NULL; + CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); + CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 64 * 1300) == STREAM_OK); + uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY], output[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input)); + const size_t lengths[] = {1, 7, sizeof(input), 17, 1000, 6, 2}; + size_t sizes[sizeof(lengths) / sizeof(lengths[0])], stored = 0; + const uint8_t *blocks[RS_MAX_BLOCKS]; + for (unsigned seq = 0; seq < sizeof(lengths) / sizeof(lengths[0]); seq++) { + CHECK(stream_encoder_encode(encoder, seq, (uint16_t)seq, 0, input, lengths[seq], blocks, &sizes[seq]) == STREAM_OK); + CHECK(sizes[seq] == STREAM_BLOCK_OVERHEAD + RS_HEADER_SIZE + (STREAM_BATCH_OVERHEAD + lengths[seq] + 5) / 6); + for (unsigned sub = 8; sub-- > 2;) CHECK(stream_buffer_put(buffer, blocks[sub], sizes[seq]) == STREAM_OK); + stored += 6 * sizes[seq]; + } + for (unsigned seq = 0; seq < sizeof(lengths) / sizeof(lengths[0]); seq++) { + size_t length = 999; memset(output, 0xcc, sizeof(output)); + CHECK(stream_buffer_decode(buffer, seq, output, sizeof(output), &length) == STREAM_OK); + CHECK(length == lengths[seq] && !memcmp(input, output, length)); + for (size_t i = length; i < sizeof(output); i++) CHECK(output[i] == 0xcc); + const uint8_t *wire; uint64_t actual; + CHECK(stream_buffer_read(buffer, seq, 2, &wire, &length, &actual) == STREAM_OK && length == sizes[seq] && actual == seq); + } + uint64_t count = 0, bytes = 0; + CHECK(stream_buffer_pending(buffer, 0, 2, &count, &bytes, NULL) == STREAM_OK); + CHECK(count == sizeof(lengths) / sizeof(lengths[0]) && bytes == stored / 6); + struct stream_buffer_stats stats; + CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK && stats.stored_bytes == stored && !stats.rejected_batches); + stream_buffer_destroy(buffer); stream_encoder_destroy(encoder); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] variable batches: tiny/full/odd lengths, two source losses, exact output, actual cache and queue bytes"); +} + +/* Повреждение содержимого/подписи и подмена внешних подсказок дают отказ после RS; следующая пачка проходит. */ +static void test_rejected_batches(void) { + struct stream_encoder *encoder = NULL; + CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); + uint8_t input[1000], output[1000], parts[8][1300]; pattern(input, sizeof(input)); + const uint8_t *blocks[RS_MAX_BLOCKS]; + for (unsigned variant = 0; variant < 5; variant++) { + struct stream_buffer *buffer = NULL; + uint8_t key[32]; memcpy(key, stream_encoder_public_key(encoder), sizeof(key)); + if (variant == 4) key[0] ^= 1; + CHECK(stream_buffer_create(&buffer, &descriptor, key, 16 * 1300) == STREAM_OK); + size_t size = 0; + CHECK(stream_encoder_encode(encoder, 10, 42, 0, input, sizeof(input), blocks, &size) == STREAM_OK); + size_t payload = size - STREAM_BLOCK_OVERHEAD - RS_HEADER_SIZE; + for (unsigned sub = 0; sub < 8; sub++) memcpy(parts[sub], blocks[sub], size); + if (variant < 2) { + size_t offset = STREAM_BATCH_HEADER + (variant == 1 ? sizeof(input) : 0); + parts[offset / payload][STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE + offset % payload] ^= 1; + } else if (variant == 2) { + for (unsigned sub = 0; sub < 8; sub++) parts[sub][STREAM_BLOCK_TIMESTAMP_OFF] ^= 1; + } else if (variant == 3) { + for (unsigned sub = 0; sub < 8; sub++) stream_write_be(parts[sub] + STREAM_BLOCK_SEQ_OFF, 10 + 1, 8); + } + uint64_t seq = variant == 3 ? 11 : 10; + for (unsigned sub = 0; sub < 6; sub++) CHECK(stream_buffer_put(buffer, parts[sub], size) == STREAM_OK); + size_t length = 999; memset(output, 0xcc, sizeof(output)); + CHECK(stream_buffer_decode(buffer, seq, output, sizeof(output), &length) == STREAM_REJECTED && length == 999); + CHECK(stream_buffer_decode(buffer, seq, output, sizeof(output), &length) == STREAM_REJECTED); + for (unsigned i = 0; i < sizeof(output); i++) CHECK(output[i] == 0xcc); + struct stream_buffer_stats stats; + CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK && stats.rejected_batches == 1); + if (variant != 4) { + CHECK(stream_encoder_encode(encoder, seq + 1, 43, 0, input, 17, blocks, &size) == STREAM_OK); + for (unsigned sub = 0; sub < 6; sub++) CHECK(stream_buffer_put(buffer, blocks[sub], size) == STREAM_OK); + CHECK(stream_buffer_decode(buffer, seq + 1, output, sizeof(output), &length) == STREAM_OK); + CHECK(length == 17 && !memcmp(input, output, length)); + } + stream_buffer_destroy(buffer); + } + /* Изменение padding не меняет подписанное тело, но нарушает канонический формат пачки. */ + struct rs_codec *rs = NULL; size_t size = 0; + CHECK(rs_init(&rs, descriptor.n, descriptor.r, descriptor.block_size) == RS_OK); + CHECK(stream_encoder_encode(encoder, 20, 42, 0, input, 17, blocks, &size) == STREAM_OK); + const uint8_t *received[RS_MAX_BLOCKS] = {0}; + for (unsigned sub = 0; sub < 8; sub++) received[sub] = blocks[sub] + STREAM_BLOCK_RS_OFF; + uint8_t decoded[STREAM_DEFAULT_INPUT_CAPACITY + STREAM_BATCH_OVERHEAD]; + size_t rs_size = size - STREAM_BLOCK_OVERHEAD, decoded_size = 6 * (rs_size - RS_HEADER_SIZE); + CHECK(rs_decode(rs, received, rs_size, decoded, sizeof(decoded)) == RS_OK); + struct stream_block_view view; CHECK(stream_block_parse(&descriptor, blocks[0], size, &view) == STREAM_OK); + CHECK(stream_batch_verify(&descriptor, stream_encoder_public_key(encoder), &view.batch, decoded, decoded_size) == STREAM_OK); + CHECK(decoded_size > STREAM_BATCH_OVERHEAD + 17); + decoded[decoded_size - 1] = 1; + CHECK(stream_batch_verify(&descriptor, stream_encoder_public_key(encoder), &view.batch, decoded, decoded_size) == STREAM_ERR_FORMAT); + rs_free(rs); stream_encoder_destroy(encoder); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] rejected batches: payload/signature, signed time/index, wrong key, padding, unchanged output and recovery"); +} + +/* Сохранять части подписанной пачки до повторного использования буферов encoder. */ static void put_batch(struct stream_encoder *encoder, struct stream_buffer *buffer, uint64_t seq, unsigned mask) { uint8_t input[37]; pattern(input, sizeof(input)); const uint8_t *blocks[RS_MAX_BLOCKS]; - CHECK(stream_encoder_encode(encoder, seq, (uint16_t)seq, 0, input, sizeof(input), blocks) == STREAM_OK); - for (unsigned i = 0; i < 8; i++) if (mask & (1u << i)) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK); + size_t size = 0; + CHECK(stream_encoder_encode(encoder, seq, (uint16_t)seq, 0, input, sizeof(input), blocks, &size) == STREAM_OK); + for (unsigned i = 0; i < 8; i++) if (mask & (1u << i)) CHECK(stream_buffer_put(buffer, blocks[i], size) == STREAM_OK); } /* Общее кольцо хранит реальные блоки, принимает порядок 10/12/11 и вытесняет целую минимальную пачку. */ @@ -223,14 +346,13 @@ static void test_ring_and_conflicts(void) { CHECK(stream_buffer_read(buffer, 10, 0, &wire, &length, &actual) == STREAM_OK); CHECK(stream_buffer_put(buffer, wire, length) == STREAM_DUPLICATE); uint8_t bad[1300]; memcpy(bad, wire, length); bad[STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE] ^= 1; - CHECK(stream_buffer_put(buffer, bad, length) == STREAM_ERR_SIGNATURE); - CHECK(sc_ed25519_sign(private_key, bad, sizeof(bad) - 64, bad + sizeof(bad) - 64) == SC_OK); CHECK(stream_buffer_put(buffer, bad, length) == STREAM_ERR_CONFLICT); put_batch(encoder, buffer, 11, 1); CHECK(cache_added == 17 && cache_removed == 8); uint64_t pending = 0, oldest = 0; - CHECK(stream_buffer_pending(buffer, 11, 0, &pending, &oldest) == STREAM_OK && pending == 2 && oldest); - CHECK(stream_buffer_pending(buffer, 12, 0, &pending, &oldest) == STREAM_OK && pending == 1); + uint64_t bytes = 0; + CHECK(stream_buffer_pending(buffer, 11, 0, &pending, &bytes, &oldest) == STREAM_OK && pending == 2 && bytes == 2 * length && oldest); + CHECK(stream_buffer_pending(buffer, 12, 0, &pending, NULL, &oldest) == STREAM_OK && pending == 1); CHECK(stream_buffer_read(buffer, 10, 0, &wire, &length, &actual) == STREAM_EXPIRED); CHECK(stream_buffer_read(buffer, 11, 1, &wire, &length, &actual) == STREAM_GAP); CHECK(stream_buffer_read(buffer, 13, 0, &wire, &length, &actual) == STREAM_WAIT); @@ -238,10 +360,10 @@ static void test_ring_and_conflicts(void) { uint64_t first = 0, last = 0; CHECK(stream_buffer_range(buffer, 0, &first, &last) == STREAM_OK && first == 11 && last == 12); struct stream_buffer_stats stats; CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK); - CHECK(stats.blocks == 9 && stats.batches == 2 && stats.stored_bytes == 9 * 1300 && stats.expired_before == 11); + CHECK(stats.blocks == 9 && stats.batches == 2 && stats.stored_bytes == 9 * length && stats.expired_before == 11); const uint8_t *blocks[RS_MAX_BLOCKS]; uint8_t input[37]; pattern(input, sizeof(input)); - CHECK(stream_encoder_encode(encoder, 10, 10, 0, input, sizeof(input), blocks) == STREAM_OK); - CHECK(stream_buffer_put(buffer, blocks[0], 1300) == STREAM_EXPIRED); + CHECK(stream_encoder_encode(encoder, 10, 10, 0, input, sizeof(input), blocks, &length) == STREAM_OK); + CHECK(stream_buffer_put(buffer, blocks[0], length) == STREAM_EXPIRED); put_batch(encoder, buffer, 13, 255); CHECK(stream_buffer_read(buffer, 11, 0, &wire, &length, &actual) == STREAM_EXPIRED); put_batch(encoder, buffer, UINT64_MAX - 1, 1); @@ -254,7 +376,7 @@ static void test_ring_and_conflicts(void) { CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 16 * 1300) == STREAM_OK); for (unsigned seq = 0; seq < 16; seq++) put_batch(encoder, buffer, seq, 1); CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK && stats.blocks == 16 && stats.batches == 16); - CHECK(stats.stored_bytes == stats.capacity_bytes); /* Ни один из семи пустых субпотоков не резервирует место. */ + CHECK(stats.stored_bytes == 16 * length && stats.stored_bytes < stats.capacity_bytes); /* Ёмкость слотов и длины данных различаются. */ put_batch(encoder, buffer, 16, 1); CHECK(stream_buffer_read(buffer, 0, 0, &wire, &length, &actual) == STREAM_EXPIRED); CHECK(stream_buffer_range(buffer, 0, &first, &last) == STREAM_OK && first == 1 && last == 16); @@ -325,7 +447,7 @@ static void source_ready(struct stream_source *source, uint64_t seq, void *arg) CHECK(stats.blocks == events->count * 8 && stats.highest_seq == seq); const uint8_t *wire = NULL; size_t length = 0; uint64_t actual = UINT64_MAX; CHECK(stream_buffer_read(stream_source_buffer(source), seq, 0, &wire, &length, &actual) == STREAM_OK); - CHECK(actual == seq && length == STREAM_BLOCK_DEFAULT_SIZE); + CHECK(actual == seq && length <= STREAM_BLOCK_DEFAULT_SIZE); events->timestamp_ms = (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2); events->flags = wire[STREAM_BLOCK_FLAGS_OFF]; for (unsigned sub = 1; sub < 8; sub++) { @@ -341,7 +463,7 @@ static void test_source_timer(void) { struct UASYNC *ua = uasync_create(); CHECK(ua != NULL); struct stream_source *source = NULL; struct source_events events = {0}; CHECK(stream_source_create(&source, ua, &descriptor, private_key, source_ready, &events) == STREAM_OK); - uint8_t input[7080], output[7080]; pattern(input, sizeof(input)); + uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY], output[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input)); CHECK(stream_source_input_capacity(source) == sizeof(input)); size_t accepted = 999; CHECK(stream_source_write(source, NULL, 0, &accepted) == STREAM_OK && accepted == 0); @@ -452,6 +574,8 @@ int main(void) { test_probe_schedule(); CHECK(u_get_allocated_count() == allocations); test_supplier_score(); CHECK(u_get_allocated_count() == allocations); test_fec_and_padding(); CHECK(u_get_allocated_count() == allocations); + test_variable_batches(); CHECK(u_get_allocated_count() == allocations); + test_rejected_batches(); CHECK(u_get_allocated_count() == allocations); test_ring_and_conflicts(); CHECK(u_get_allocated_count() == allocations); test_relative_metrics(); CHECK(u_get_allocated_count() == allocations); test_source_timer(); CHECK(u_get_allocated_count() == allocations); diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index ef1e7a1a..3de6db01 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -137,27 +137,41 @@ static void on_restart(struct UTUN_INSTANCE *inst, const struct stream_event *ev static void test_restart_bound(struct STREAM_HANDLE *subscriber, const struct stream_descriptor *descriptor) { struct restart_events events = {.next = 50}; struct stream_encoder *encoder = NULL; const uint8_t *blocks[RS_MAX_BLOCKS]; + size_t size = 0; CHECK(stream_encoder_create(&encoder, descriptor, nodes[0]->my_ed25519_privkey) == STREAM_OK); stream_set_event_callback(nodes[3], on_restart, &events); - CHECK(stream_encoder_encode(encoder, 50, 1, 0, expected, sizeof(expected), blocks) == STREAM_OK); - for (unsigned sub = 0; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], 1300) == STREAM_OK); + CHECK(stream_encoder_encode(encoder, 50, 1, 0, expected, sizeof(expected), blocks, &size) == STREAM_OK); + for (unsigned sub = 0; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], size) == STREAM_OK); run_until(&events.data, 1); CHECK(stream_receive_restart(subscriber) == STREAM_OK); - CHECK(stream_encoder_encode(encoder, 60, 2, 0, expected, sizeof(expected), blocks) == STREAM_OK); - CHECK(stream_ingest_block(nodes[3], blocks[0], 1300) == STREAM_OK); + CHECK(stream_encoder_encode(encoder, 60, 2, 0, expected, sizeof(expected), blocks, &size) == STREAM_OK); + CHECK(stream_ingest_block(nodes[3], blocks[0], size) == STREAM_OK); CHECK(stream_receive_restart(subscriber) == STREAM_OK); - CHECK(stream_ingest_block(nodes[3], blocks[0], 1300) == STREAM_DUPLICATE); - for (unsigned sub = 1; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], 1300) == STREAM_OK); + CHECK(stream_ingest_block(nodes[3], blocks[0], size) == STREAM_DUPLICATE); + for (unsigned sub = 1; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], size) == STREAM_OK); run_until(&events.data, 2); CHECK(events.gaps == 1 && events.next == 61); CHECK(stream_receive_restart(subscriber) == STREAM_OK); CHECK(stream_receive_restart(subscriber) == STREAM_OK); - CHECK(stream_ingest_block(nodes[3], blocks[0], 1300) == STREAM_DUPLICATE); + CHECK(stream_ingest_block(nodes[3], blocks[0], size) == STREAM_DUPLICATE); test_clock_run_for(ua, 1000); struct stream_local local; CHECK(stream_local_get(nodes[3], descriptor, &local) == STREAM_OK); CHECK(local.anchored && local.start == 61 && local.next == 61 && events.data == 2 && events.gaps == 1); + /* Неверная подпись пачки даёт ровно один GAP и сохраняет owner/приём следующего индекса. */ + CHECK(stream_encoder_encode(encoder, 61, 3, 0, expected, sizeof(expected), blocks, &size) == STREAM_OK); + uint8_t bad[STREAM_BLOCK_MAX_SIZE]; memcpy(bad, blocks[0], size); bad[STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE] ^= 1; + CHECK(stream_ingest_block(nodes[3], bad, size) == STREAM_OK); + for (unsigned sub = 1; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], size) == STREAM_OK); + run_until(&events.gaps, 2); + CHECK(events.data == 2 && events.next == 62 && stream_handle_active(subscriber)); + CHECK(stream_encoder_encode(encoder, 62, 4, 0, expected, sizeof(expected), blocks, &size) == STREAM_OK); + for (unsigned sub = 0; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], size) == STREAM_OK); + run_until(&events.data, 3); + CHECK(events.gaps == 2 && events.next == 63 && stream_handle_active(subscriber)); + struct stream_info info; + CHECK(stream_get_info(subscriber, &info) == STREAM_OK && info.buffer.rejected_batches == 1); stream_set_event_callback(nodes[3], NULL, NULL); stream_encoder_destroy(encoder); - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] repeated restart: deferred GAP retained; stale bootstrap never replays DATA"); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] restart and rejected batch: exact GAP, unchanged owner, following authenticated DATA"); } static void pressure_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { @@ -1482,13 +1496,14 @@ int main(int argc, char **argv) { struct stream_encoder *encoder = NULL; CHECK(stream_encoder_create(&encoder, &info.descriptor, nodes[0]->my_ed25519_privkey) == STREAM_OK); const uint8_t *blocks[RS_MAX_BLOCKS]; - CHECK(stream_encoder_encode(encoder, 100, 65530, 0, expected, sizeof(expected), blocks) == STREAM_OK); + size_t size = 0; + CHECK(stream_encoder_encode(encoder, 100, 65530, 0, expected, sizeof(expected), blocks, &size) == STREAM_OK); close_readers_in_callback = 1; - for (unsigned i = 0; i < 8; i++) if (i != 1 && i != 5) CHECK(stream_ingest_block(nodes[1], blocks[i], 1300) == STREAM_OK); + for (unsigned i = 0; i < 8; i++) if (i != 1 && i != 5) CHECK(stream_ingest_block(nodes[1], blocks[i], size) == STREAM_OK); run_until(&data_events[1], 1); run_until(&closed_events[1], 1); CHECK(!reader1 && !reader2 && stream_handle_active(other_group)); - uint8_t invalid[1300]; memcpy(invalid, blocks[0], sizeof(invalid)); invalid[1299] ^= 1; - CHECK(stream_ingest_block(nodes[1], invalid, sizeof(invalid)) == STREAM_WAIT); /* Нет подписки — нет crypto. */ + uint8_t invalid[STREAM_BLOCK_MAX_SIZE]; memcpy(invalid, blocks[0], size); invalid[size - 1] ^= 1; + CHECK(stream_ingest_block(nodes[1], invalid, size) == STREAM_WAIT); /* Нет подписки — нет decode/crypto. */ stream_encoder_destroy(encoder); size_t accepted = 0; CHECK(stream_publish_write(publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));