jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Test batch authentication after losses and recovery after rejection

master
evgeny 6 days ago
parent
f15c97d11a
commit
e96b813fcb
2 changed files with 200 additions and 61 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 222
      tests/test_stream_core.c
  2. 39
      tests/test_stream_service.c

222
tests/test_stream_core.c
Unescape View File

@ -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);

39
tests/test_stream_service.c
Unescape View File

@ -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));

Write Preview
Loading…
Cancel
Save