From cee5b26c1307c555240a0753af4c40ef8858d0ff Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 20:06:29 +0200 Subject: [PATCH] Sign stream batches before RS and authenticate reconstructed data --- src/stream/stream_block.c | 164 ++++++++++++++++++++++++------------- src/stream/stream_buffer.c | 117 ++++++++++++++++---------- 2 files changed, 181 insertions(+), 100 deletions(-) diff --git a/src/stream/stream_block.c b/src/stream/stream_block.c index 2b3f9ce8..5ff0aa7f 100644 --- a/src/stream/stream_block.c +++ b/src/stream/stream_block.c @@ -19,7 +19,8 @@ struct stream_encoder { /* Проверить ключ трансляции и параметры до расчёта размеров/выделения памяти. */ int stream_descriptor_validate(const struct stream_descriptor *d) { if (!d || !d->group_id || !d->source_node_id || !d->source_epoch || !d->n || d->r >= d->n - || d->block_size <= RS_HEADER_SIZE || d->block_size > RS_MAX_BLOCK_SIZE) { + || d->block_size <= RS_HEADER_SIZE || d->block_size > RS_MAX_BLOCK_SIZE + || (size_t)(d->n - d->r) * (d->block_size - RS_HEADER_SIZE) <= STREAM_BATCH_OVERHEAD) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "descriptor invalid: present=%d group=%llx source=%llx epoch=%llx n=%u r=%u size=%u", d != NULL, d ? (unsigned long long)d->group_id : 0, d ? (unsigned long long)d->source_node_id : 0, d ? (unsigned long long)d->source_epoch : 0, d ? d->n : 0, d ? d->r : 0, d ? d->block_size : 0); @@ -28,12 +29,13 @@ int stream_descriptor_validate(const struct stream_descriptor *d) { return STREAM_OK; } -size_t stream_block_size(const struct stream_descriptor *d) { +size_t stream_block_capacity(const struct stream_descriptor *d) { return stream_descriptor_validate(d) == STREAM_OK ? d->block_size + STREAM_BLOCK_OVERHEAD : 0; } size_t stream_input_size(const struct stream_descriptor *d) { - return stream_descriptor_validate(d) == STREAM_OK ? (size_t)(d->n - d->r) * (d->block_size - RS_HEADER_SIZE) : 0; + return stream_descriptor_validate(d) == STREAM_OK + ? (size_t)(d->n - d->r) * (d->block_size - RS_HEADER_SIZE) - STREAM_BATCH_OVERHEAD : 0; } /* Создать рабочую память RS и получить публичный ключ именно переданного Ed25519 seed. */ @@ -64,8 +66,8 @@ int stream_encoder_create(struct stream_encoder **out, const struct stream_descr stream_encoder_destroy(encoder); return STREAM_ERR_FEC; } - encoder->input = u_malloc((uint32_t)stream_input_size(d)); - encoder->wire = u_malloc((uint32_t)(d->n * stream_block_size(d))); + encoder->input = u_malloc((uint32_t)rs_input_capacity(encoder->rs)); + encoder->wire = u_malloc((uint32_t)(d->n * stream_block_capacity(d))); if (!encoder->input || !encoder->wire) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder working memory allocation failed n=%u size=%u", d->n, d->block_size); stream_encoder_destroy(encoder); @@ -74,7 +76,7 @@ int stream_encoder_create(struct stream_encoder **out, const struct stream_descr *out = encoder; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "encoder created group=%llx source=%llx epoch=%llx FEC=%u+%u block=%zu input=%zu", (unsigned long long)d->group_id, (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, - d->n - d->r, d->r, stream_block_size(d), stream_input_size(d)); + d->n - d->r, d->r, stream_block_capacity(d), stream_input_size(d)); return STREAM_OK; } @@ -84,7 +86,7 @@ void stream_encoder_destroy(struct stream_encoder *encoder) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "encoder destroy source=%llx epoch=%llx", (unsigned long long)encoder->descriptor.source_node_id, (unsigned long long)encoder->descriptor.source_epoch); OPENSSL_cleanse(encoder->private_key, sizeof(encoder->private_key)); - if (encoder->input) OPENSSL_cleanse(encoder->input, stream_input_size(&encoder->descriptor)); + if (encoder->input) OPENSSL_cleanse(encoder->input, rs_input_capacity(encoder->rs)); u_free(encoder->input); u_free(encoder->wire); rs_free(encoder->rs); @@ -96,85 +98,137 @@ const uint8_t *stream_encoder_public_key(const struct stream_encoder *encoder) { return encoder->public_key; } -/* Сформировать канонический заголовок, RS-кодограммы и подпись каждого блока. */ +/* Подписать метаданные и содержимое один раз, затем защитить всё сообщение кодированием RS. */ 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) { - if (!encoder || !input || !blocks || !length || seq == UINT64_MAX || (flags & ~STREAM_BLOCK_TIME_RESET) + const uint8_t *input, size_t length, const uint8_t **blocks, size_t *block_length) { + if (!encoder || !input || !blocks || !block_length || !length || seq == UINT64_MAX || (flags & ~STREAM_BLOCK_TIME_RESET) || length > stream_input_size(&encoder->descriptor)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode invalid arguments seq=%llu len=%zu flags=%u", (unsigned long long)seq, length, flags); return STREAM_ERR_ARGUMENT; } const struct stream_descriptor *d = &encoder->descriptor; - size_t wire_size = stream_block_size(d); + size_t signed_size = STREAM_BATCH_HEADER + length, batch_size = signed_size + SC_SIGN_SIZE; + size_t rs_size = RS_HEADER_SIZE + (batch_size + d->n - d->r - 1) / (d->n - d->r); + size_t wire_size = STREAM_BLOCK_OVERHEAD + rs_size, stride = stream_block_capacity(d); uint8_t *rs_blocks[RS_MAX_BLOCKS]; - memcpy(encoder->input, input, length); - memset(encoder->input + length, 0, stream_input_size(d) - length); + uint8_t *batch = encoder->input; + memcpy(batch, STREAM_BATCH_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE); + stream_write_be(batch + STREAM_BLOCK_GROUP_OFF, d->group_id, 8); + stream_write_be(batch + STREAM_BLOCK_SOURCE_OFF, d->source_node_id, 8); + stream_write_be(batch + STREAM_BLOCK_EPOCH_OFF, d->source_epoch, 8); + batch[STREAM_BLOCK_N_OFF] = d->n; batch[STREAM_BLOCK_R_OFF] = d->r; + stream_write_be(batch + STREAM_BLOCK_SIZE_OFF, rs_size, 2); + stream_write_be(batch + STREAM_BLOCK_TIMESTAMP_OFF, timestamp_ms, 2); + stream_write_be(batch + STREAM_BLOCK_LENGTH_OFF, length, 4); + batch[STREAM_BLOCK_FLAGS_OFF] = flags; + stream_write_be(batch + STREAM_BATCH_SEQ_OFF, seq, 8); + memcpy(batch + STREAM_BATCH_HEADER, input, length); + if (sc_ed25519_sign(encoder->private_key, batch, signed_size, batch + signed_size) != SC_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode batch signature failed source=%llx seq=%llu length=%zu", + (unsigned long long)d->source_node_id, (unsigned long long)seq, length); + return STREAM_ERR_CRYPTO; + } for (unsigned i = 0; i < d->n; i++) { - uint8_t *wire = encoder->wire + i * wire_size; + uint8_t *wire = encoder->wire + i * stride; + memcpy(wire, batch, STREAM_BLOCK_RS_OFF); memcpy(wire, STREAM_BLOCK_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE); - stream_write_be(wire + STREAM_BLOCK_GROUP_OFF, d->group_id, 8); - stream_write_be(wire + STREAM_BLOCK_SOURCE_OFF, d->source_node_id, 8); - stream_write_be(wire + STREAM_BLOCK_EPOCH_OFF, d->source_epoch, 8); - wire[STREAM_BLOCK_N_OFF] = d->n; - wire[STREAM_BLOCK_R_OFF] = d->r; - stream_write_be(wire + STREAM_BLOCK_SIZE_OFF, d->block_size, 2); - stream_write_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, timestamp_ms, 2); - stream_write_be(wire + STREAM_BLOCK_LENGTH_OFF, length, 4); - wire[STREAM_BLOCK_FLAGS_OFF] = flags; rs_blocks[i] = wire + STREAM_BLOCK_RS_OFF; } - if (rs_encode(encoder->rs, seq, encoder->input, rs_blocks) != RS_OK) { + size_t encoded_size = 0; + if (rs_encode(encoder->rs, seq, batch, batch_size, rs_blocks, &encoded_size) != RS_OK || encoded_size != rs_size) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode RS failed seq=%llu", (unsigned long long)seq); return STREAM_ERR_FEC; } - for (unsigned i = 0; i < d->n; i++) { - uint8_t *wire = encoder->wire + i * wire_size; - if (sc_ed25519_sign(encoder->private_key, wire, wire_size - SC_SIGN_SIZE, wire + wire_size - SC_SIGN_SIZE) != SC_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode signature failed seq=%llu sub=%u", (unsigned long long)seq, i); - return STREAM_ERR_CRYPTO; - } - } - for (unsigned i = 0; i < d->n; i++) blocks[i] = encoder->wire + i * wire_size; - DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "encoded seq=%llu timestamp=%u len=%zu n=%u flags=%u", - (unsigned long long)seq, timestamp_ms, length, d->n, flags); + for (unsigned i = 0; i < d->n; i++) blocks[i] = encoder->wire + i * stride; + *block_length = wire_size; + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "encoded signed batch seq=%llu timestamp=%u len=%zu n=%u flags=%u part=%zu max=%zu padding=%zu", + (unsigned long long)seq, timestamp_ms, length, d->n, flags, wire_size, stride, + (size_t)(d->n - d->r) * (rs_size - RS_HEADER_SIZE) - batch_size); return STREAM_OK; } -/* Проверить размеры, область подписи, дескриптор, метаданные и подпись до публикации view. */ -int stream_block_verify(const struct stream_descriptor *d, const uint8_t public_key[32], const uint8_t *wire, - size_t length, struct stream_block_view *view) { - if (!public_key || !wire || !view || stream_descriptor_validate(d) != STREAM_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "verify invalid arguments length=%zu", length); +/* Только структурная проверка до хранения/пересылки; подлинность определяет восстановленная пачка. */ +int stream_block_parse(const struct stream_descriptor *d, const uint8_t *wire, size_t length, struct stream_block_view *view) { + if (!wire || !view || stream_descriptor_validate(d) != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "part parse invalid arguments length=%zu", length); return STREAM_ERR_ARGUMENT; } - if (length != stream_block_size(d) || memcmp(wire, STREAM_BLOCK_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE) + if (length <= STREAM_BLOCK_OVERHEAD + RS_HEADER_SIZE || length > stream_block_capacity(d) + || memcmp(wire, STREAM_BLOCK_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE) || stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8) != d->group_id || stream_read_be(wire + STREAM_BLOCK_SOURCE_OFF, 8) != d->source_node_id || stream_read_be(wire + STREAM_BLOCK_EPOCH_OFF, 8) != d->source_epoch || wire[STREAM_BLOCK_N_OFF] != d->n || wire[STREAM_BLOCK_R_OFF] != d->r - || stream_read_be(wire + STREAM_BLOCK_SIZE_OFF, 2) != d->block_size) { - DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify format/descriptor mismatch source=%llx epoch=%llx length=%zu expected=%zu", - (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, length, stream_block_size(d)); + || stream_read_be(wire + STREAM_BLOCK_SIZE_OFF, 2) != length - STREAM_BLOCK_OVERHEAD) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "part format/descriptor mismatch source=%llx epoch=%llx length=%zu max=%zu", + (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, length, stream_block_capacity(d)); return STREAM_ERR_FORMAT; } struct stream_block_view parsed = { - .seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8), - .data_length = (uint32_t)stream_read_be(wire + STREAM_BLOCK_LENGTH_OFF, 4), - .timestamp_ms = (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2), - .subblock = wire[STREAM_BLOCK_SUB_OFF], .flags = wire[STREAM_BLOCK_FLAGS_OFF], .rs = wire + STREAM_BLOCK_RS_OFF + .batch = {.seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8), + .data_length = (uint32_t)stream_read_be(wire + STREAM_BLOCK_LENGTH_OFF, 4), + .timestamp_ms = (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2), + .block_size = (uint16_t)(length - STREAM_BLOCK_OVERHEAD), .flags = wire[STREAM_BLOCK_FLAGS_OFF]}, + .subblock = wire[STREAM_BLOCK_SUB_OFF], .rs = wire + STREAM_BLOCK_RS_OFF }; - if (parsed.seq == UINT64_MAX || parsed.subblock >= d->n || !parsed.data_length - || parsed.data_length > stream_input_size(d) || (parsed.flags & ~STREAM_BLOCK_TIME_RESET)) { - DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify invalid metadata seq=%llu sub=%u length=%u flags=%u", - (unsigned long long)parsed.seq, parsed.subblock, parsed.data_length, parsed.flags); + const struct stream_batch_info *info = &parsed.batch; + if (info->seq == UINT64_MAX || parsed.subblock >= d->n || !info->data_length || info->data_length > stream_input_size(d) + || (info->flags & ~STREAM_BLOCK_TIME_RESET) + || info->block_size != RS_HEADER_SIZE + (STREAM_BATCH_OVERHEAD + info->data_length + d->n - d->r - 1) / (d->n - d->r)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "part invalid metadata seq=%llu sub=%u length=%u flags=%u size=%u", + (unsigned long long)info->seq, parsed.subblock, info->data_length, info->flags, info->block_size); + return STREAM_ERR_FORMAT; + } + *view = parsed; + return STREAM_OK; +} + +/* Подпись проверяется после RS; внешние подсказки обязаны совпадать с подписанными полями пачки. */ +int stream_batch_verify(const struct stream_descriptor *d, const uint8_t public_key[32], + const struct stream_batch_info *expected, const uint8_t *decoded, size_t decoded_length) { + if (!public_key || !expected || !decoded || stream_descriptor_validate(d) != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "batch verify invalid arguments length=%zu", decoded_length); + return STREAM_ERR_ARGUMENT; + } + if (expected->seq == UINT64_MAX || !expected->data_length || expected->data_length > stream_input_size(d) + || (expected->flags & ~STREAM_BLOCK_TIME_RESET) || expected->block_size > d->block_size + || expected->block_size != RS_HEADER_SIZE + (STREAM_BATCH_OVERHEAD + expected->data_length + d->n - d->r - 1) / (d->n - d->r) + || decoded_length != (size_t)(d->n - d->r) * (expected->block_size - RS_HEADER_SIZE) + || decoded_length < STREAM_BATCH_OVERHEAD + expected->data_length) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "batch invalid size source=%llx seq=%llu decoded=%zu data=%u part=%u", + (unsigned long long)d->source_node_id, (unsigned long long)expected->seq, + decoded_length, expected->data_length, expected->block_size); return STREAM_ERR_FORMAT; } - if (sc_ed25519_verify(public_key, wire, length - SC_SIGN_SIZE, wire + length - SC_SIGN_SIZE) != SC_OK) { - DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify source signature failed source=%llx epoch=%llx seq=%llu sub=%u", + size_t signed_size = STREAM_BATCH_HEADER + expected->data_length; + if (sc_ed25519_verify(public_key, decoded, signed_size, decoded + signed_size) != SC_OK) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "batch source signature rejected source=%llx epoch=%llx seq=%llu data=%u part=%u", (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, - (unsigned long long)parsed.seq, parsed.subblock); + (unsigned long long)expected->seq, expected->data_length, expected->block_size); return STREAM_ERR_SIGNATURE; } - *view = parsed; + if (memcmp(decoded, STREAM_BATCH_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE) + || stream_read_be(decoded + STREAM_BLOCK_GROUP_OFF, 8) != d->group_id + || stream_read_be(decoded + STREAM_BLOCK_SOURCE_OFF, 8) != d->source_node_id + || stream_read_be(decoded + STREAM_BLOCK_EPOCH_OFF, 8) != d->source_epoch + || decoded[STREAM_BLOCK_N_OFF] != d->n || decoded[STREAM_BLOCK_R_OFF] != d->r + || stream_read_be(decoded + STREAM_BLOCK_SIZE_OFF, 2) != expected->block_size + || stream_read_be(decoded + STREAM_BLOCK_TIMESTAMP_OFF, 2) != expected->timestamp_ms + || stream_read_be(decoded + STREAM_BLOCK_LENGTH_OFF, 4) != expected->data_length + || decoded[STREAM_BLOCK_FLAGS_OFF] != expected->flags + || stream_read_be(decoded + STREAM_BATCH_SEQ_OFF, 8) != expected->seq) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "batch signed metadata mismatch source=%llx epoch=%llx seq=%llu", + (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, (unsigned long long)expected->seq); + return STREAM_ERR_FORMAT; + } + for (size_t i = signed_size + SC_SIGN_SIZE; i < decoded_length; i++) { + if (decoded[i]) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "batch nonzero padding source=%llx seq=%llu offset=%zu", + (unsigned long long)d->source_node_id, (unsigned long long)expected->seq, i); + return STREAM_ERR_FORMAT; + } + } + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "batch source signature verified source=%llx seq=%llu length=%u part=%u", + (unsigned long long)d->source_node_id, (unsigned long long)expected->seq, expected->data_length, expected->block_size); return STREAM_OK; } diff --git a/src/stream/stream_buffer.c b/src/stream/stream_buffer.c index dcb1c9ab..80d4d53d 100644 --- a/src/stream/stream_buffer.c +++ b/src/stream/stream_buffer.c @@ -23,8 +23,10 @@ struct stream_cached_batch { uint32_t first_slot; uint32_t data_length; uint16_t timestamp_ms; + uint16_t block_size; uint16_t count; uint8_t flags; + int rejected; }; struct stream_buffer { @@ -114,7 +116,7 @@ static void buffer_evict_oldest(struct stream_buffer *b) { uint32_t before = b->slots[slot].sub_previous, after = b->slots[slot].sub_next; if (before == STREAM_SLOT_NONE) b->sub_head[sub] = after; else b->slots[before].sub_next = after; if (after == STREAM_SLOT_NONE) b->sub_tail[sub] = before; else b->slots[after].sub_previous = before; - if (b->observer) b->observer(b->observer_arg, b->data + slot * b->block_bytes, b->block_bytes, 0); + if (b->observer) b->observer(b->observer_arg, b->data + slot * b->block_bytes, STREAM_BLOCK_OVERHEAD + batch->block_size, 0); u_free(b->slots[slot].path); b->slots[slot].path = NULL; if (b->latest_valid[sub] && b->latest[sub] == batch->seq) b->latest_valid[sub] = 0; b->slots[slot].next = b->free_slot; @@ -123,7 +125,7 @@ static void buffer_evict_oldest(struct stream_buffer *b) { } b->stats.expired_before = batch->seq + 1; b->stats.blocks -= batch->count; - b->stats.stored_bytes -= batch->count * b->block_bytes; + b->stats.stored_bytes -= batch->count * (STREAM_BLOCK_OVERHEAD + batch->block_size); b->stats.evicted_batches++; DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer evicted seq=%llu blocks=%u retained=%u bytes=%zu low=%llu", (unsigned long long)batch->seq, batch->count, b->stats.blocks, b->stats.stored_bytes, @@ -137,7 +139,7 @@ int stream_buffer_create(struct stream_buffer **out, const struct stream_descrip DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer create invalid arguments out=%p key=%p", (void *)out, (const void *)public_key); return STREAM_ERR_ARGUMENT; } - size_t block_bytes = stream_block_size(d); + size_t block_bytes = stream_block_capacity(d); if (!bytes) bytes = STREAM_BUFFER_BYTES; if (bytes > STREAM_BUFFER_BYTES || bytes < d->n * block_bytes) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer budget invalid bytes=%zu batch=%zu max=%u", bytes, d->n * block_bytes, STREAM_BUFFER_BYTES); @@ -156,7 +158,7 @@ int stream_buffer_create(struct stream_buffer **out, const struct stream_descrip b->slots = u_calloc(b->capacity, sizeof(*b->slots)); b->heap = u_calloc(b->capacity, sizeof(*b->heap)); b->hash = u_calloc(b->hash_size, sizeof(*b->hash)); - b->decode_output = u_malloc((uint32_t)stream_input_size(d)); + b->decode_output = u_malloc((uint32_t)(stream_input_size(d) + STREAM_BATCH_OVERHEAD)); if (!b->data || !b->slots || !b->heap || !b->hash || !b->decode_output) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer arrays allocation failed capacity=%u bytes=%zu", b->capacity, b->stats.capacity_bytes); stream_buffer_destroy(b); @@ -194,7 +196,7 @@ void stream_buffer_destroy(struct stream_buffer *b) { u_free(b); } -/* Вставить только подписанный блок; OOM/конфликт проверяются до вытеснения ранее сохранённых данных. */ +/* Сохранить часть после проверки формата; подпись источника проверяется при восстановлении пачки. */ int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t length) { struct stream_path path = {.count = 1}; if (b) path.nodes[0] = b->descriptor.source_node_id; @@ -207,42 +209,40 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t if (!path || !path->count || path->count > STREAM_PATH_MAX || stream_path_validate(path, b->descriptor.source_node_id, path->nodes[path->count - 1], 0) != STREAM_OK) return STREAM_ERR_FORMAT; - struct stream_cached_batch *batch = NULL; - if (length == b->block_bytes) { - batch = buffer_batch(b, stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8)); - uint8_t sub = wire[STREAM_BLOCK_SUB_OFF]; - if (batch && sub < b->descriptor.n) { - uint32_t slot = buffer_slot(b, batch, sub); - if (slot != STREAM_SLOT_NONE && !memcmp(b->data + slot * b->block_bytes, wire, length)) { - b->stats.duplicates++; - DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer duplicate seq=%llu sub=%u", (unsigned long long)batch->seq, sub); - return STREAM_DUPLICATE; - } - } - } struct stream_block_view view; - int result = stream_block_verify(&b->descriptor, b->public_key, wire, length, &view); + int result = stream_block_parse(&b->descriptor, wire, length, &view); if (result != STREAM_OK) return result; - if (view.seq < b->stats.expired_before) { + const struct stream_batch_info *info = &view.batch; + struct stream_cached_batch *batch = buffer_batch(b, info->seq); + if (batch && batch->block_size == info->block_size) { + uint32_t slot = buffer_slot(b, batch, view.subblock); + if (slot != STREAM_SLOT_NONE && !memcmp(b->data + slot * b->block_bytes, wire, length)) { + b->stats.duplicates++; + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer duplicate seq=%llu sub=%u", (unsigned long long)batch->seq, view.subblock); + return STREAM_DUPLICATE; + } + } + if (info->seq < b->stats.expired_before) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer expired incoming seq=%llu sub=%u low=%llu full=%d", - (unsigned long long)view.seq, view.subblock, (unsigned long long)b->stats.expired_before, b->stats.blocks == b->capacity); + (unsigned long long)info->seq, view.subblock, (unsigned long long)b->stats.expired_before, b->stats.blocks == b->capacity); return STREAM_EXPIRED; } - if (batch && (batch->timestamp_ms != view.timestamp_ms || batch->data_length != view.data_length || batch->flags != view.flags + if (batch && (batch->timestamp_ms != info->timestamp_ms || batch->data_length != info->data_length || batch->flags != info->flags + || batch->block_size != info->block_size || buffer_slot(b, batch, view.subblock) != STREAM_SLOT_NONE)) { - DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer source conflict seq=%llu sub=%u timestamp=%u/%u length=%u/%u flags=%u/%u", - (unsigned long long)view.seq, view.subblock, batch->timestamp_ms, view.timestamp_ms, - batch->data_length, view.data_length, batch->flags, view.flags); + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer part conflict seq=%llu sub=%u timestamp=%u/%u length=%u/%u flags=%u/%u size=%u/%u", + (unsigned long long)info->seq, view.subblock, batch->timestamp_ms, info->timestamp_ms, + batch->data_length, info->data_length, batch->flags, info->flags, batch->block_size, info->block_size); return STREAM_ERR_CONFLICT; } - if (b->stats.blocks == b->capacity && view.seq <= b->heap[0]->seq) { + if (b->stats.blocks == b->capacity && info->seq <= b->heap[0]->seq) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer full: incoming seq=%llu sub=%u cannot replace newer data oldest=%llu", - (unsigned long long)view.seq, view.subblock, (unsigned long long)b->heap[0]->seq); + (unsigned long long)info->seq, view.subblock, (unsigned long long)b->heap[0]->seq); return STREAM_EXPIRED; } struct stream_cached_path *stored_path = path->count > 1 ? u_malloc(sizeof(*stored_path) + path->count * 8u) : NULL; if (path->count > 1 && !stored_path) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer path allocation failed seq=%llu", (unsigned long long)view.seq); + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer path allocation failed seq=%llu", (unsigned long long)info->seq); return STREAM_ERR_MEMORY; } if (stored_path) { stored_path->count = path->count; memcpy(stored_path->nodes, path->nodes, path->count * 8u); } @@ -250,14 +250,15 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t if (new_batch) { batch = u_calloc(1, sizeof(*batch)); if (!batch) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer batch allocation failed seq=%llu", (unsigned long long)view.seq); + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer batch allocation failed seq=%llu", (unsigned long long)info->seq); u_free(stored_path); return STREAM_ERR_MEMORY; } - batch->seq = view.seq; + batch->seq = info->seq; batch->first_slot = STREAM_SLOT_NONE; - batch->timestamp_ms = view.timestamp_ms; - batch->data_length = view.data_length; - batch->flags = view.flags; + batch->timestamp_ms = info->timestamp_ms; + batch->data_length = info->data_length; + batch->block_size = info->block_size; + batch->flags = info->flags; } if (b->stats.blocks == b->capacity) buffer_evict_oldest(b); if (new_batch) buffer_insert_batch(b, batch); @@ -265,7 +266,7 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t b->free_slot = b->slots[slot].next; b->slots[slot].subblock = view.subblock; b->slots[slot].path = stored_path; - b->slots[slot].seq = view.seq; b->slots[slot].received_ms = get_time_tb() / 10; + b->slots[slot].seq = info->seq; b->slots[slot].received_ms = get_time_tb() / 10; b->slots[slot].sub_next = STREAM_SLOT_NONE; b->slots[slot].sub_previous = b->sub_tail[view.subblock]; if (b->sub_tail[view.subblock] == STREAM_SLOT_NONE) b->sub_head[view.subblock] = slot; else b->slots[b->sub_tail[view.subblock]].sub_next = slot; @@ -276,13 +277,13 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t memcpy(b->data + slot * b->block_bytes, wire, length); b->stats.blocks++; b->stats.stored_bytes += length; - if (view.seq > b->stats.highest_seq) b->stats.highest_seq = view.seq; - if (!b->latest_valid[view.subblock] || view.seq > b->latest[view.subblock]) { - b->latest[view.subblock] = view.seq; + if (info->seq > b->stats.highest_seq) b->stats.highest_seq = info->seq; + if (!b->latest_valid[view.subblock] || info->seq > b->latest[view.subblock]) { + b->latest[view.subblock] = info->seq; b->latest_valid[view.subblock] = 1; } DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer stored seq=%llu sub=%u present=%u/%u bytes=%zu", - (unsigned long long)view.seq, view.subblock, batch->count, b->descriptor.n, b->stats.stored_bytes); + (unsigned long long)info->seq, view.subblock, batch->count, b->descriptor.n, b->stats.stored_bytes); if (b->observer) b->observer(b->observer_arg, b->data + slot * b->block_bytes, length, 1); return STREAM_OK; } @@ -293,18 +294,20 @@ void stream_buffer_set_observer(struct stream_buffer *b, stream_buffer_event_fn } /* Список субпотока упорядочен по монотонному времени вставки, включая поступившие вне порядка индексы. */ -int stream_buffer_pending(struct stream_buffer *b, uint64_t seq, uint8_t sub, uint64_t *count, uint64_t *oldest) { - if (!b || sub >= b->descriptor.n || (!count && !oldest)) { +int stream_buffer_pending(struct stream_buffer *b, uint64_t seq, uint8_t sub, uint64_t *count, uint64_t *bytes, uint64_t *oldest) { + if (!b || sub >= b->descriptor.n || (!count && !bytes && !oldest)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer pending invalid arguments sub=%u", sub); return STREAM_ERR_ARGUMENT; } - uint64_t found = 0, first_ms = 0; + uint64_t found = 0, total_bytes = 0, first_ms = 0; for (uint32_t slot = b->sub_head[sub]; slot != STREAM_SLOT_NONE; slot = b->slots[slot].sub_next) { if (b->slots[slot].seq < seq) continue; if (!found) first_ms = b->slots[slot].received_ms; found++; - if (!count) break; + total_bytes += STREAM_BLOCK_OVERHEAD + stream_read_be(b->data + slot * b->block_bytes + STREAM_BLOCK_SIZE_OFF, 2); + if (!count && !bytes) break; } if (count) *count = found; + if (bytes) *bytes = total_bytes; if (oldest) *oldest = first_ms; return found ? STREAM_OK : STREAM_WAIT; } @@ -363,7 +366,7 @@ int stream_buffer_read(struct stream_buffer *b, uint64_t seq, uint8_t subblock, return STREAM_GAP; } *wire = b->data + slot * b->block_bytes; - *length = b->block_bytes; + *length = STREAM_BLOCK_OVERHEAD + batch->block_size; *actual_seq = seq; return STREAM_OK; } @@ -407,6 +410,7 @@ int stream_buffer_decode(struct stream_buffer *b, uint64_t seq, uint8_t *output, if (seq < b->stats.expired_before) return STREAM_EXPIRED; struct stream_cached_batch *batch = buffer_batch(b, seq); if (!batch) return !b->stats.blocks || seq > b->stats.highest_seq ? STREAM_WAIT : STREAM_GAP; + if (batch->rejected) return STREAM_REJECTED; if (batch->count < b->descriptor.n - b->descriptor.r) return STREAM_WAIT; if (capacity < batch->data_length) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode output too small seq=%llu capacity=%zu needed=%u", @@ -416,11 +420,34 @@ int stream_buffer_decode(struct stream_buffer *b, uint64_t seq, uint8_t *output, const uint8_t *blocks[RS_MAX_BLOCKS] = {0}; for (uint32_t slot = batch->first_slot; slot != STREAM_SLOT_NONE; slot = b->slots[slot].next) blocks[b->slots[slot].subblock] = b->data + slot * b->block_bytes + STREAM_BLOCK_RS_OFF; - if (rs_decode(b->rs, blocks, b->decode_output) != RS_OK) { + size_t decoded_size = (size_t)(b->descriptor.n - b->descriptor.r) * (batch->block_size - RS_HEADER_SIZE); + if (rs_decode(b->rs, blocks, batch->block_size, b->decode_output, rs_input_capacity(b->rs)) != RS_OK) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode RS failed seq=%llu present=%u", (unsigned long long)seq, batch->count); return STREAM_ERR_FEC; } - memcpy(output, b->decode_output, batch->data_length); + struct stream_batch_info expected = {.seq = batch->seq, .data_length = batch->data_length, + .timestamp_ms = batch->timestamp_ms, .block_size = batch->block_size, .flags = batch->flags}; + int result = stream_batch_verify(&b->descriptor, b->public_key, &expected, b->decode_output, decoded_size); + if (result != STREAM_OK) { + if (result != STREAM_ERR_SIGNATURE && result != STREAM_ERR_FORMAT) return result; + batch->rejected = 1; b->stats.rejected_batches++; + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer rejected batch group=%llx source=%llx epoch=%llx seq=%llu present=%u required=%u result=%d", + (unsigned long long)b->descriptor.group_id, (unsigned long long)b->descriptor.source_node_id, + (unsigned long long)b->descriptor.source_epoch, (unsigned long long)seq, + batch->count, b->descriptor.n - b->descriptor.r, result); + unsigned used = 0; + for (unsigned sub = 0; sub < b->descriptor.n && used < (unsigned)(b->descriptor.n - b->descriptor.r); sub++) { + if (!blocks[sub]) continue; + uint32_t slot = buffer_slot(b, batch, sub); + const struct stream_cached_path *path = b->slots[slot].path; + uint64_t supplier = path && path->count > 1 ? path->nodes[path->count - 2] : b->descriptor.source_node_id; + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer rejected batch contribution seq=%llu sub=%u supplier=%llx part=%u", + (unsigned long long)seq, sub, (unsigned long long)supplier, batch->block_size); + used++; + } + return STREAM_REJECTED; + } + memcpy(output, b->decode_output + STREAM_BATCH_HEADER, batch->data_length); *length = batch->data_length; DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer decoded seq=%llu present=%u length=%u", (unsigned long long)seq, batch->count, batch->data_length); return STREAM_OK;