fix: validate inbound payloads

Author: anshalshukla

.github/workflows/ci.yml

@@ -128,7 +128,7 @@ jobs:
 
   build-all-provers:
     name: build-all-provers
-    runs-on: ubuntu-latest
+    runs-on: ${{ matrix.os }}
     needs: build  # Only run if the build job succeeds
     strategy:
       matrix:
@@ -182,7 +182,7 @@ jobs:
 
   test:
     name: test
-    runs-on: ubuntu-latest
+    runs-on: ${{ matrix.os }}
     strategy:
       matrix:
         os: [ubuntu-latest, macos-latest]

pkgs/cli/test/integration.zig

@@ -523,9 +523,10 @@ test "SSE events integration test - wait for justification and finalization" {
     // Node3 starts after first finalization and syncs via parent block requests (blocks_by_root).
     // We verify sync by waiting for finalization to advance beyond the first finalized slot,
     // which proves the chain continued progressing after node3 joined.
-    const timeout_ms: u64 = 240000; // 240 seconds timeout
+    const timeout_ms: u64 = 240000; // 240 seconds to reach first justification/finalization
+    const post_finalization_timeout_ms: u64 = 120000; // extra time for node3 sync/finalization advancement
     const start_ns = std.time.nanoTimestamp();
-    const deadline_ns = start_ns + timeout_ms * std.time.ns_per_ms;
+    var deadline_ns = start_ns + timeout_ms * std.time.ns_per_ms;
     var got_justification = false;
     var got_finalization = false;
     var got_node3_sync = false;
@@ -555,8 +556,13 @@ test "SSE events integration test - wait for justification and finalization" {
                         got_finalization = true;
                         first_finalized_slot = slot;
                         head_count_at_finalization = sse_client.getEventCount("new_head");
+                        const sync_deadline_ns = std.time.nanoTimestamp() + post_finalization_timeout_ms * std.time.ns_per_ms;
+                        if (sync_deadline_ns > deadline_ns) {
+                            deadline_ns = sync_deadline_ns;
+                        }
                         std.debug.print("INFO: First finalization at slot {} — node 3 will start syncing via parent block requests\n", .{slot});
                         std.debug.print("INFO: Head events at finalization: {}\n", .{head_count_at_finalization});
+                        std.debug.print("INFO: Extended deadline by {}ms for node3 sync verification\n", .{post_finalization_timeout_ms});
                     } else if (got_finalization and slot > first_finalized_slot and !got_node3_sync) {
                         // Finalization advanced beyond the first finalized slot.
                         // This means the chain continued progressing after node3 joined.

pkgs/network/src/ethlibp2p.zig

@@ -26,6 +26,24 @@ const ServerStreamError = error{
 const MAX_RPC_MESSAGE_SIZE: usize = 4 * 1024 * 1024;
 const MAX_VARINT_BYTES: usize = uvarint.bufferSize(usize);
 
+// SSZ size constants derived directly from type definitions for payload validation.
+const CHECKPOINT_SSZ_SIZE = zeam_utils.fixedSszSize(types.Checkpoint);
+const ATTESTATION_DATA_SSZ_SIZE = zeam_utils.fixedSszSize(types.AttestationData);
+const SIGNED_ATTESTATION_SSZ_SIZE = zeam_utils.fixedSszSize(types.SignedAttestation);
+
+comptime {
+    if (ATTESTATION_DATA_SSZ_SIZE != 8 + 3 * CHECKPOINT_SSZ_SIZE) {
+        @compileError("AttestationData SSZ layout changed; revisit payload size assumptions");
+    }
+    if (SIGNED_ATTESTATION_SSZ_SIZE != 8 + ATTESTATION_DATA_SSZ_SIZE + types.SIGSIZE) {
+        @compileError("SignedAttestation SSZ layout changed; revisit payload size assumptions");
+    }
+}
+
+// SignedBlockWithAttestation is variable-size with 2 variable fields (message, signature).
+// SSZ struct encoding: at least 2 offsets (4 bytes each) = 8 bytes minimum.
+const MIN_SIGNED_BLOCK_WITH_ATTESTATION_SSZ_SIZE = 8;
+
 const FrameDecodeError = error{
     EmptyFrame,
     PayloadTooLarge,
@@ -48,6 +66,53 @@ fn decodeVarint(bytes: []const u8) uvarint.VarintParseError!struct { value: usiz
     };
 }
 
+fn validateGossipSnappyHeader(message_bytes: []const u8) (uvarint.VarintParseError || error{PayloadTooLarge})!struct { value: usize, length: usize } {
+    const decoded = try decodeVarint(message_bytes);
+    if (decoded.value > MAX_RPC_MESSAGE_SIZE) {
+        return error.PayloadTooLarge;
+    }
+    return .{
+        .value = decoded.value,
+        .length = decoded.length,
+    };
+}
+
+fn validateSignedBlockWithAttestation(bytes: []const u8) !void {
+    if (bytes.len < MIN_SIGNED_BLOCK_WITH_ATTESTATION_SSZ_SIZE) {
+        return error.InvalidEncoding;
+    }
+
+    const message_offset: usize = @intCast(std.mem.readInt(u32, bytes[0..4], .little));
+    const signature_offset: usize = @intCast(std.mem.readInt(u32, bytes[4..8], .little));
+
+    if (message_offset != MIN_SIGNED_BLOCK_WITH_ATTESTATION_SSZ_SIZE) {
+        return error.InvalidEncoding;
+    }
+    if (signature_offset < message_offset) {
+        return error.InvalidEncoding;
+    }
+    if (signature_offset > bytes.len) {
+        return error.InvalidEncoding;
+    }
+}
+
+fn initSignedBlockWithAttestation(allocator: Allocator) !types.SignedBlockWithAttestation {
+    var block: types.BeamBlock = undefined;
+    try block.setToDefault(allocator);
+    errdefer block.deinit();
+
+    var signatures = try types.createBlockSignatures(allocator, 0);
+    errdefer signatures.deinit();
+
+    return .{
+        .message = .{
+            .block = block,
+            .proposer_attestation = undefined,
+        },
+        .signature = signatures,
+    };
+}
+
 /// Build a request frame with varint-encoded uncompressed size followed by snappy-framed payload.
 fn buildRequestFrame(allocator: Allocator, uncompressed_size: usize, snappy_payload: []const u8) ![]u8 {
     if (uncompressed_size > MAX_RPC_MESSAGE_SIZE) {
@@ -87,11 +152,7 @@ fn parseRequestFrame(bytes: []const u8) FrameDecodeError!struct {
         return error.EmptyFrame;
     }
 
-    const decoded = try decodeVarint(bytes);
-
-    if (decoded.value > MAX_RPC_MESSAGE_SIZE) {
-        return error.PayloadTooLarge;
-    }
+    const decoded = try validateGossipSnappyHeader(bytes);
 
     return .{
         .declared_len = decoded.value,
@@ -111,11 +172,7 @@ fn parseResponseFrame(bytes: []const u8) FrameDecodeError!struct {
         return error.Incomplete;
     }
 
-    const decoded = try decodeVarint(bytes[1..]);
-
-    if (decoded.value > MAX_RPC_MESSAGE_SIZE) {
-        return error.PayloadTooLarge;
-    }
+    const decoded = try validateGossipSnappyHeader(bytes[1..]);
 
     return .{
         .code = bytes[0],
@@ -283,7 +340,6 @@ export fn handleMsgFromRustBridge(zigHandler: *EthLibp2p, topic_str: [*:0]const
     };
 
     const message_bytes: []const u8 = message_ptr[0..message_len];
-
     const uncompressed_message = snappyz.decode(zigHandler.allocator, message_bytes) catch |e| {
         zigHandler.logger.err("Error in snappyz decoding the message for topic={s}: {any}", .{ std.mem.span(topic_str), e });
         if (writeFailedBytes(message_bytes, "snappyz_decode", zigHandler.allocator, null, zigHandler.logger)) |filename| {
@@ -294,9 +350,32 @@ export fn handleMsgFromRustBridge(zigHandler: *EthLibp2p, topic_str: [*:0]const
         return;
     };
     defer zigHandler.allocator.free(uncompressed_message);
+
+    if (uncompressed_message.len > MAX_RPC_MESSAGE_SIZE) {
+        zigHandler.logger.err(
+            "Gossip message decompressed size {d} exceeds limit {d} for topic={s}",
+            .{ uncompressed_message.len, MAX_RPC_MESSAGE_SIZE, std.mem.span(topic_str) },
+        );
+        return;
+    }
+
     const message: interface.GossipMessage = switch (topic.gossip_topic) {
         .block => blockmessage: {
-            var message_data: types.SignedBlockWithAttestation = undefined;
+            validateSignedBlockWithAttestation(uncompressed_message) catch {
+                const message_offset = if (uncompressed_message.len >= 4) std.mem.readInt(u32, uncompressed_message[0..4], .little) else 0;
+                const signature_offset = if (uncompressed_message.len >= 8) std.mem.readInt(u32, uncompressed_message[4..8], .little) else 0;
+                zigHandler.logger.err(
+                    "Invalid gossip block top-level SSZ offsets: len={d} message_offset={d} signature_offset={d}",
+                    .{ uncompressed_message.len, message_offset, signature_offset },
+                );
+                return;
+            };
+            var message_data = initSignedBlockWithAttestation(zigHandler.allocator) catch |e| {
+                zigHandler.logger.err("Error initializing signed block payload before deserialization: {any}", .{e});
+                return;
+            };
+            var decode_succeeded = false;
+            defer if (!decode_succeeded) message_data.deinit();
             ssz.deserialize(types.SignedBlockWithAttestation, uncompressed_message, &message_data, zigHandler.allocator) catch |e| {
                 zigHandler.logger.err("Error in deserializing the signed block message: {any}", .{e});
                 if (writeFailedBytes(uncompressed_message, "block", zigHandler.allocator, null, zigHandler.logger)) |filename| {
@@ -306,10 +385,18 @@ export fn handleMsgFromRustBridge(zigHandler: *EthLibp2p, topic_str: [*:0]const
                 }
                 return;
             };
+            decode_succeeded = true;
 
             break :blockmessage .{ .block = message_data };
         },
         .attestation => attestationmessage: {
+            if (uncompressed_message.len != SIGNED_ATTESTATION_SSZ_SIZE) {
+                zigHandler.logger.err(
+                    "Gossip attestation message size mismatch: got {d} bytes, expected {d}",
+                    .{ uncompressed_message.len, SIGNED_ATTESTATION_SSZ_SIZE },
+                );
+                return;
+            }
             var message_data: types.SignedAttestation = undefined;
             ssz.deserialize(types.SignedAttestation, uncompressed_message, &message_data, zigHandler.allocator) catch |e| {
                 zigHandler.logger.err("Error in deserializing the signed attestation message: {any}", .{e});
@@ -1222,3 +1309,38 @@ pub const EthLibp2p = struct {
         return result;
     }
 };
+
+test "SIGNED_ATTESTATION_SSZ_SIZE matches actual serialized size" {
+    const attestation = types.SignedAttestation{
+        .validator_id = 0,
+        .message = .{
+            .slot = 0,
+            .head = .{ .root = [_]u8{0} ** 32, .slot = 0 },
+            .target = .{ .root = [_]u8{0} ** 32, .slot = 0 },
+            .source = .{ .root = [_]u8{0} ** 32, .slot = 0 },
+        },
+        .signature = [_]u8{0} ** types.SIGSIZE,
+    };
+    var serialized: std.ArrayList(u8) = .empty;
+    defer serialized.deinit(std.testing.allocator);
+    try ssz.serialize(types.SignedAttestation, attestation, &serialized, std.testing.allocator);
+    try std.testing.expectEqual(SIGNED_ATTESTATION_SSZ_SIZE, serialized.items.len);
+}
+
+test "validateGossipSnappyHeader rejects oversized declared size" {
+    var scratch: [MAX_VARINT_BYTES]u8 = undefined;
+    const encoded = uvarint.encode(usize, MAX_RPC_MESSAGE_SIZE + 1, &scratch);
+    try std.testing.expectError(error.PayloadTooLarge, validateGossipSnappyHeader(encoded));
+}
+
+test "validateSignedBlockWithAttestationTopLevelOffsets rejects invalid offsets" {
+    var bad_first: [8]u8 = undefined;
+    std.mem.writeInt(u32, bad_first[0..4], 4, .little);
+    std.mem.writeInt(u32, bad_first[4..8], 8, .little);
+    try std.testing.expectError(error.InvalidEncoding, validateSignedBlockWithAttestation(&bad_first));
+
+    var bad_second: [8]u8 = undefined;
+    std.mem.writeInt(u32, bad_second[0..4], 8, .little);
+    std.mem.writeInt(u32, bad_second[4..8], 4, .little);
+    try std.testing.expectError(error.InvalidEncoding, validateSignedBlockWithAttestation(&bad_second));
+}