Bound stream path allocations and validate wire rejection cases

src/stream/stream_buffer.c

@@ -6,10 +6,12 @@
 
 #define STREAM_SLOT_NONE UINT32_MAX
 
+struct stream_cached_path { uint8_t count; uint64_t nodes[]; };
+
 struct stream_cached_slot {
     uint32_t next;
     uint8_t subblock;
-    struct stream_path *path; /* NULL — исходный путь {source}, без отдельной аллокации. */
+    struct stream_cached_path *path; /* NULL — исходный путь {source}, без отдельной аллокации. */
 };
 
 struct stream_cached_batch {
@@ -227,12 +229,12 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t
                    (unsigned long long)view.seq, view.subblock, (unsigned long long)b->heap[0]->seq);
         return STREAM_EXPIRED;
     }
-    struct stream_path *stored_path = path->count > 1 ? u_malloc(sizeof(*stored_path)) : NULL;
+    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);
         return STREAM_ERR_MEMORY;
     }
-    if (stored_path) *stored_path = *path;
+    if (stored_path) { stored_path->count = path->count; memcpy(stored_path->nodes, path->nodes, path->count * 8u); }
     int new_batch = batch == NULL;
     if (new_batch) {
         batch = u_calloc(1, sizeof(*batch));
@@ -276,7 +278,11 @@ int stream_buffer_path(struct stream_buffer *b, uint64_t seq, uint8_t subblock,
     struct stream_cached_batch *batch = buffer_batch(b, seq);
     uint32_t slot = batch ? buffer_slot(b, batch, subblock) : STREAM_SLOT_NONE;
     if (slot == STREAM_SLOT_NONE) return seq < b->stats.expired_before ? STREAM_EXPIRED : STREAM_GAP;
-    *path = b->slots[slot].path ? *b->slots[slot].path : (struct stream_path){.nodes = {b->descriptor.source_node_id}, .count = 1};
+    *path = (struct stream_path){.nodes = {b->descriptor.source_node_id}, .count = 1};
+    if (b->slots[slot].path) {
+        path->count = b->slots[slot].path->count;
+        memcpy(path->nodes, b->slots[slot].path->nodes, path->count * 8u);
+    }
     return STREAM_OK;
 }
 

src/stream/stream_network.c

@@ -143,6 +143,10 @@ static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) {
         else if (!f->outgoing) route_control(p, m, &payload, 1);
     }
     f->info.state = STREAM_RECEIVE_CLOSED; f->info.reason = reason;
+    if (reason != STREAM_REASON_STOPPED)
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream issuance failure peer=%llx handle=%llu sub=%u reason=%d next=%llu",
+                   (unsigned long long)f->info.peer, (unsigned long long)f->info.handle, f->subblock,
+                   reason, (unsigned long long)f->info.next);
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream issuance closed peer=%llx group=%llx source=%llx handle=%llu sub=%u out=%d reason=%d blocks=%llu",
                (unsigned long long)f->info.peer, (unsigned long long)f->descriptor.group_id,
                (unsigned long long)f->descriptor.source_node_id, (unsigned long long)f->info.handle, f->subblock,
@@ -492,7 +496,11 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
     }
     struct stream_peer *p = peer_find(net, sender);
     if (routed && m.command == STREAM_CMD_REVERSE_REQ) {
-        if (m.length || m.handle || m.receiver_epoch || !authorized(net, sender, &m, &local)) return;
+        if (m.length || m.handle || m.receiver_epoch) {
+            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream REVERSE_REQ malformed peer=%llx bytes=%zu handle=%llu receiver_epoch=%llx",
+                       (unsigned long long)sender, m.length, (unsigned long long)m.handle, (unsigned long long)m.receiver_epoch); return;
+        }
+        if (!authorized(net, sender, &m, &local)) return;
         if (!p) p = peer_open(net, sender);
         if (p) { p->idle = get_time_tb(); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream REVERSE_REQ peer=%llx attempt=%llx NCD=%p",
             (unsigned long long)sender, (unsigned long long)m.sender_epoch, (void *)p->ncd); }
@@ -502,7 +510,11 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
         DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed DATA/control forbidden command=%u", m.command); return;
     }
     if (m.command == STREAM_CMD_INFO_REQ && !routed) {
-        if (m.handle || m.length || m.subblock >= local.descriptor.n || !authorized(net, sender, &m, &local)) return;
+        if (m.handle || m.length || m.subblock >= local.descriptor.n) {
+            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO_REQ malformed peer=%llx bytes=%zu handle=%llu sub=%u n=%u",
+                       (unsigned long long)sender, m.length, (unsigned long long)m.handle, m.subblock, local.descriptor.n); return;
+        }
+        if (!authorized(net, sender, &m, &local)) return;
         if (!p) p = peer_open(net, sender);
         if (!p || (m.receiver_epoch && m.receiver_epoch != p->own_epoch)) return;
         if (p->other_epoch && p->other_epoch != m.sender_epoch) {

tests/test_stream_core.c

@@ -78,6 +78,34 @@ static void test_signed_blocks(void) {
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] signed blocks: canonical bytes, all field classes, wrong key and malformed sizes");
 }
 
+/* Wire сохраняет все 64 бита и не меняет выход при усечении; путь не допускает повторного посещения. */
+static void test_wire(void) {
+    uint8_t wire[STREAM_WIRE_HEADER + 8];
+    struct stream_message input = {.descriptor = descriptor, .sender_epoch = UINT64_MAX, .receiver_epoch = UINT64_MAX - 1,
+        .handle = UINT64_MAX - 2, .command = STREAM_CMD_SUBSCRIBE, .subblock = 7};
+    stream_wire_header(wire, &input); stream_write_be(wire + STREAM_WIRE_HEADER, UINT64_MAX - 3, 8);
+    struct stream_message parsed;
+    CHECK(stream_wire_parse(wire, sizeof(wire), &parsed) == STREAM_OK);
+    CHECK(parsed.sender_epoch == input.sender_epoch && parsed.receiver_epoch == input.receiver_epoch && parsed.handle == input.handle);
+    CHECK(parsed.descriptor.group_id == descriptor.group_id && parsed.descriptor.source_epoch == descriptor.source_epoch);
+    CHECK(parsed.subblock == 7 && parsed.length == 8 && stream_read_be(parsed.payload, 8) == UINT64_MAX - 3);
+    for (size_t i = 0; i < STREAM_WIRE_HEADER; i++) {
+        memset(&parsed, 0xa5, sizeof(parsed)); uint8_t before[sizeof(parsed)]; memcpy(before, &parsed, sizeof(parsed));
+        CHECK(stream_wire_parse(wire, i, &parsed) == STREAM_ERR_FORMAT && !memcmp(before, &parsed, sizeof(parsed)));
+    }
+    wire[1]++; CHECK(stream_wire_parse(wire, sizeof(wire), &parsed) == STREAM_ERR_FORMAT); wire[1]--;
+    wire[2] = 255; CHECK(stream_wire_parse(wire, sizeof(wire), &parsed) == STREAM_ERR_FORMAT);
+    struct stream_path path = {.nodes = {descriptor.source_node_id, 1, 2}, .count = 3}, read;
+    uint8_t encoded[1 + STREAM_PATH_MAX * 8]; size_t used = 0, length = stream_path_write(encoded, &path);
+    CHECK(stream_path_parse(encoded, length, &read, &used) == STREAM_OK && used == length && read.count == 3 && read.nodes[2] == 2);
+    CHECK(stream_path_validate(&read, descriptor.source_node_id, 2, 3) == STREAM_OK);
+    CHECK(stream_path_validate(&read, descriptor.source_node_id, 2, 1) == STREAM_ERR_FORMAT);
+    read.nodes[2] = 1; CHECK(stream_path_validate(&read, descriptor.source_node_id, 1, 0) == STREAM_ERR_FORMAT);
+    for (size_t i = 0; i < length; i++) CHECK(stream_path_parse(encoded, i, &read, &used) == STREAM_ERR_FORMAT);
+    encoded[0] = STREAM_PATH_MAX + 1; CHECK(stream_path_parse(encoded, sizeof(encoded), &read, &used) == STREAM_ERR_FORMAT);
+    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;
@@ -317,6 +345,7 @@ int main(void) {
     for (unsigned i = 0; i < sizeof(private_key); i++) private_key[i] = (uint8_t)(i * 37 + 11);
     size_t allocations = u_get_allocated_count();
     test_signed_blocks(); CHECK(u_get_allocated_count() == allocations);
+    test_wire(); CHECK(u_get_allocated_count() == allocations);
     test_fec_and_padding(); 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);