diff --git a/build.zig.zon b/build.zig.zon index a5896897a..0423095cc 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -37,8 +37,8 @@ .hash = "rocksdb-9.7.4-z_CUTnfJAAB2izSeJNWGGdy-Q0aB3wVMa0AohWJDVVBy", }, .zig_snappy = .{ - .url = "git+https://github.com/blockblaz/zig-snappy#5058ae298101e409af9d35cf7eeddc2650477164", - .hash = "zig_snappy-0.0.3-bDFzXmBjAAAd1yTK5LQvRJ-srdnJbhkHZajl0KcXVJDx", + .url = "git+https://github.com/blockblaz/zig-snappy?ref=v0.0.5#050f529bc5fa11242140312ecf550c186f05af1e", + .hash = "zig_snappy-0.0.5-bDFzXlh4AADPRkDi2swDMAf8shH39Fc1g5oYRAhFQfTt", }, .snappyframesz = .{ .url = "git+https://github.com/blockblaz/snappyframesz#df262c69ca24f072dd71b26c70972649458b6659", diff --git a/pkgs/network/src/ethlibp2p.zig b/pkgs/network/src/ethlibp2p.zig index d4036b8b1..4cec770e2 100644 --- a/pkgs/network/src/ethlibp2p.zig +++ b/pkgs/network/src/ethlibp2p.zig @@ -39,14 +39,48 @@ const MAX_RPC_MESSAGE_SIZE: usize = 4 * 1024 * 1024; /// /// Set to 50 MB to accommodate current devnet block sizes with room to grow. /// Revisit once the leanSpec formalises a MAX_GOSSIP_BLOCK_SIZE constant. +/// +/// TODO(#855 review #14): 50 MB × N peers is a real memory-pressure surface. +/// Track in a follow-up issue once the spec lands and we can lower this. const MAX_GOSSIP_BLOCK_SIZE: usize = 50 * 1024 * 1024; const MAX_VARINT_BYTES: usize = uvarint.bufferSize(usize); const FrameDecodeError = error{ EmptyFrame, + MalformedVarint, PayloadTooLarge, Incomplete, -} || uvarint.VarintParseError; +}; + +/// Failure modes returned by the snappy block-format header validators. +/// Each variant maps to a distinct ops/attacker shape; callers should keep +/// them distinct in logs and (eventually) metrics. +const SnappyHeaderValidationError = error{ + /// Empty buffer — nothing to decode. + EmptyMessage, + /// Leading varint is corrupt (truncated, oversized, or u64-overflow). + InvalidVarint, + /// Varint decoded cleanly but declares a payload larger than the limit + /// allowed for this protocol/topic. Strict `>` to match the upstream + /// `snappyz.decodeWithMax` contract (`if (block.blockLen > max_size)`). + /// Pinning that comparison here so a future upstream change to `>=` + /// flips the boundary and is caught loudly via this comment plus tests, + /// rather than silently disagreeing across a 1-byte gap. + DeclaredPayloadTooLarge, + /// Header parsed cleanly and declared a non-zero payload, but the + /// buffer contains only the header bytes (no body). Distinct from + /// `InvalidVarint` because the header itself is well-formed; this is a + /// truncated message, not a malformed one. + HeaderWithoutBody, +}; + +/// Successful decode of a snappy block-format header: the declared +/// uncompressed length and the number of bytes occupied by the varint +/// header itself. +const SnappyHeader = struct { + value: usize, + length: usize, +}; const LeanSupportedProtocol = interface.LeanSupportedProtocol; @@ -64,10 +98,33 @@ 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; +/// Validate a snappy block-format header against an arbitrary size limit. +/// Used by both the gossip path (`validateGossipSnappyHeader`) and the RPC +/// frame parsers (`validateRpcSnappyHeader`); each caller passes its own +/// per-protocol/per-topic limit. +/// +/// On success, returns the decoded length and the header byte count. On +/// failure, returns one of the `SnappyHeaderValidationError` variants so +/// callers can attribute different attacker shapes (corrupt varint vs. +/// oversized claim vs. missing body) in logs and metrics. +/// +/// Header-only validation: this is *not* a full body integrity check. A +/// well-formed header followed by a body shorter than `decoded.value` +/// (but at least one byte) is accepted here — the actual decoder is +/// authoritative for body checks. We only reject the degenerate case +/// where the buffer is exactly the header and nothing else, because that +/// can never compress to a non-zero declared size. +fn validateSnappyHeader( + message_bytes: []const u8, + max_size: usize, +) SnappyHeaderValidationError!SnappyHeader { + if (message_bytes.len == 0) return error.EmptyMessage; + const decoded = decodeVarint(message_bytes) catch return error.InvalidVarint; + if (decoded.value > max_size) return error.DeclaredPayloadTooLarge; + // A valid snappy block must have at least the header byte(s) and may have + // zero compressed bytes only when the declared uncompressed size is zero. + if (decoded.value > 0 and decoded.length == message_bytes.len) { + return error.HeaderWithoutBody; } return .{ .value = decoded.value, @@ -75,6 +132,47 @@ fn validateGossipSnappyHeader(message_bytes: []const u8) (uvarint.VarintParseErr }; } +/// RPC frame snappy-header validator. Used by `parseRequestFrame` and +/// `parseResponseFrame` to bound declared sizes before snappy-frame decode. +/// (Renamed from `validateGossipSnappyHeader` in PR #855: the original +/// name was inverted — it was always RPC, never gossip.) +fn validateRpcSnappyHeader(message_bytes: []const u8) FrameDecodeError!SnappyHeader { + return validateSnappyHeader(message_bytes, MAX_RPC_MESSAGE_SIZE) catch |e| switch (e) { + error.EmptyMessage => return error.EmptyFrame, + error.InvalidVarint => return error.MalformedVarint, + error.DeclaredPayloadTooLarge => return error.PayloadTooLarge, + // Header-only is not a fatal RPC frame condition: the body bytes + // may simply not have arrived yet on this read. Treat as Incomplete. + error.HeaderWithoutBody => return error.Incomplete, + }; +} + +/// Gossip block-format snappy-header validator. Called from +/// `handleMsgFromRustBridge` before invoking `snappyz.decodeWithMax` so +/// malformed varint headers and oversized declared sizes are rejected +/// before any heap allocation. Per-topic `max_size` lets the caller +/// pass `MAX_GOSSIP_BLOCK_SIZE` for blocks vs. `MAX_RPC_MESSAGE_SIZE` +/// for attestations/aggregations. +/// +/// This guard rejects malformed varint headers and oversized declared +/// sizes; it does not (and cannot) verify body integrity — that's the +/// decoder's job. +/// +/// Two-layer defense exit criteria (PR #855 review #6): keep this guard +/// permanently. It serves three purposes the upstream zig-snappy library +/// can't: (a) rejects oversized declared sizes pre-allocation using zeam's +/// per-topic limits, (b) gives callers a typed error so we can attribute +/// attacker shapes in logs/metrics, (c) acts as a safety net if a future +/// upstream version regresses on malformed-input handling. The varint +/// decode is the only piece that overlaps with the upstream decoder; that +/// overlap is documented in `handleMsgFromRustBridge`'s call site. +fn validateGossipSnappyHeader( + message_bytes: []const u8, + max_size: usize, +) SnappyHeaderValidationError!SnappyHeader { + return validateSnappyHeader(message_bytes, max_size); +} + /// 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) { @@ -114,7 +212,7 @@ fn parseRequestFrame(bytes: []const u8) FrameDecodeError!struct { return error.EmptyFrame; } - const decoded = try validateGossipSnappyHeader(bytes); + const decoded = try validateRpcSnappyHeader(bytes); return .{ .declared_len = decoded.value, @@ -134,7 +232,7 @@ fn parseResponseFrame(bytes: []const u8) FrameDecodeError!struct { return error.Incomplete; } - const decoded = try validateGossipSnappyHeader(bytes[1..]); + const decoded = try validateRpcSnappyHeader(bytes[1..]); return .{ .code = bytes[0], @@ -252,6 +350,23 @@ fn serverStreamIsFinished(ptr: *anyopaque) bool { return ctx.finished; } +/// 1-of-N sample counter for malformed-message debug dumps. Without this gate, +/// a peer spamming garbage gossip (e.g. sustained 1k msg/s) would fill the +/// disk with one debug file per message (PR #855 review #5). We only persist +/// `1` of every `MALFORMED_DUMP_SAMPLE_RATE` rejections; the rest are logged +/// inline. The counter is process-local and racy across threads, which is +/// fine — the goal is *not* exact 1:1024 sampling, just bounded disk pressure. +const MALFORMED_DUMP_SAMPLE_RATE: usize = 1024; +var malformed_dump_counter: std.atomic.Value(usize) = std.atomic.Value(usize).init(0); + +/// Returns true iff the caller should persist this malformed message to disk. +/// Always persists the very first malformed message of a process so a single +/// reproducible failure during testing isn't lost behind the sampler. +fn shouldPersistMalformedDump() bool { + const n = malformed_dump_counter.fetchAdd(1, .monotonic); + return n == 0 or (n % MALFORMED_DUMP_SAMPLE_RATE) == 0; +} + /// Writes failed deserialization bytes to disk for debugging purposes. /// Logs the outcome (success or failure) itself; returns true on success. /// @@ -318,6 +433,43 @@ fn deserializeGossipMessage( return message_data; } +/// Log + sample-dump a gossip rejection from the snappy-header guard. Each +/// `SnappyHeaderValidationError` variant maps to a distinct attacker shape: +/// corrupt varint = malformed bytes, declared-too-large = oversized claim, +/// header-without-body = truncated stream, empty = degenerate. Keeping these +/// separate in the log line preserves attribution; collapsing into a single +/// "malformed snappy header" line (as the original PR did) loses the signal. +fn rejectMalformedGossip( + zigHandler: *EthLibp2p, + err: SnappyHeaderValidationError, + topic_slice: []const u8, + sender_peer_id_slice: []const u8, + message_bytes: []const u8, +) void { + const reason: []const u8 = switch (err) { + error.EmptyMessage => "empty gossip payload", + error.InvalidVarint => "corrupt snappy varint header", + error.DeclaredPayloadTooLarge => "declared snappy payload exceeds per-topic limit", + error.HeaderWithoutBody => "snappy header parsed but body bytes are missing", + }; + const dump_label: []const u8 = switch (err) { + error.EmptyMessage => "snappy_empty", + error.InvalidVarint => "snappy_varint", + error.DeclaredPayloadTooLarge => "snappy_oversized", + error.HeaderWithoutBody => "snappy_truncated", + }; + const node_name = zigHandler.node_registry.getNodeNameFromPeerId(sender_peer_id_slice); + zigHandler.logger.err( + "Rejecting gossip message: {s} (topic={s}, len={d}, peer={s}{f})", + .{ reason, topic_slice, message_bytes.len, sender_peer_id_slice, node_name }, + ); + if (shouldPersistMalformedDump()) { + if (!writeFailedBytes(message_bytes, dump_label, zigHandler.allocator, null, zigHandler.logger)) { + zigHandler.logger.err("Failed to persist malformed gossip dump ({s})", .{dump_label}); + } + } +} + export fn handleMsgFromRustBridge(zigHandler: *EthLibp2p, topic_str: [*:0]const u8, message_ptr: [*]const u8, message_len: usize, sender_peer_id: [*:0]const u8) void { const topic = interface.LeanNetworkTopic.decode(zigHandler.allocator, topic_str) catch |err| { zigHandler.logger.err("Ignoring Invalid topic_id={s} sent in handleMsgFromRustBridge: {any}", .{ std.mem.span(topic_str), err }); @@ -325,20 +477,52 @@ export fn handleMsgFromRustBridge(zigHandler: *EthLibp2p, topic_str: [*:0]const }; const message_bytes: []const u8 = message_ptr[0..message_len]; + const sender_peer_id_slice = std.mem.span(sender_peer_id); + const topic_slice = std.mem.span(topic_str); // Block gossip messages carry XMSS/post-quantum aggregated signatures and can be // substantially larger than the 4 MB RPC limit (devnet4 saw ~9.37 MB — issue #723). // Use the larger MAX_GOSSIP_BLOCK_SIZE for block topics; keep the tighter limit for // small messages (attestations, aggregations) to bound memory use. + // + // TODO(#855 review #9): attestations/aggregations rarely approach + // MAX_RPC_MESSAGE_SIZE (4 MB). Tighter per-kind ceilings would let us + // reject earlier and reduce attacker amplification. Track separately. const decode_limit: usize = switch (topic.gossip_topic.kind) { .block => MAX_GOSSIP_BLOCK_SIZE, else => MAX_RPC_MESSAGE_SIZE, }; + // Defense-in-depth gate before the third-party decoder. Rejects malformed + // varint headers and oversized declared sizes so the gossip thread can't + // panic on adversarial input regardless of upstream decoder state. Returns + // typed errors so we can attribute attacker shapes (corrupt varint vs. + // oversized claim vs. truncated body) in logs and — eventually — metrics. + // + // Note (PR #855 review #8): this decodes the leading varint, and so does + // `snappyz.decodeWithMax` further down. The duplication is intentional and + // worth O(10ns) per gossip message; both decoders MUST agree on the same + // size-limit comparison (strict `>`, see `SnappyHeaderValidationError`). + // If the upstream contract ever changes (e.g. to `>=`), the boundary tests + // pinned in the test block below will go red. + _ = validateGossipSnappyHeader(message_bytes, decode_limit) catch |e| { + rejectMalformedGossip(zigHandler, e, topic_slice, sender_peer_id_slice, message_bytes); + // TODO(#855 review #4): apply a libp2p gossipsub score penalty here + // so a peer spamming malformed gossip is ejected by the protocol + // instead of getting unlimited free retries. Out of scope for the + // panic fix; tracked separately. + return; + }; + const uncompressed_message = snappyz.decodeWithMax(zigHandler.allocator, message_bytes, decode_limit) 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)) { - zigHandler.logger.err("Snappyz decode failed - could not create debug file", .{}); + zigHandler.logger.err( + "Error in snappyz decoding the message for topic={s} from peer={s}: {any}", + .{ topic_slice, sender_peer_id_slice, e }, + ); + if (shouldPersistMalformedDump()) { + if (!writeFailedBytes(message_bytes, "snappyz_decode", zigHandler.allocator, null, zigHandler.logger)) { + zigHandler.logger.err("Snappyz decode failed - could not create debug file", .{}); + } } return; }; @@ -383,7 +567,6 @@ export fn handleMsgFromRustBridge(zigHandler: *EthLibp2p, topic_str: [*:0]const }; defer message.deinit(); - const sender_peer_id_slice = std.mem.span(sender_peer_id); const node_name = zigHandler.node_registry.getNodeNameFromPeerId(sender_peer_id_slice); switch (message) { .block => |signed_block| { @@ -438,7 +621,7 @@ export fn handleMsgFromRustBridge(zigHandler: *EthLibp2p, topic_str: [*:0]const "network-{d}:: gossip payload json topic={s} from peer={s}{f}: {f}", .{ zigHandler.params.networkId, - std.mem.span(topic_str), + topic_slice, sender_peer_id_slice, node_name, zeam_utils.LazyJson(interface.GossipMessage).init(zigHandler.allocator, &message), @@ -1548,8 +1731,123 @@ pub const EthLibp2p = struct { } }; -test "validateGossipSnappyHeader rejects oversized declared size" { +test "validateRpcSnappyHeader 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)); + try std.testing.expectError(error.PayloadTooLarge, validateRpcSnappyHeader(encoded)); +} + +test "validateGossipSnappyHeader returns typed errors for each rejection class" { + // Regression for Hive `gossip: ignores malformed ssz` (test 390 on + // hive.leanroadmap.org / suite 1778305924-...). The simulator publishes + // 1024 bytes of 0xef on a valid block topic; an unguarded decoder hit an + // `integer overflow` panic in the third-party snappy uvarint and crashed + // the network thread, cascading into ~26 follow-up failures as the second + // node became unreachable. + // + // Each sub-case asserts both the rejection AND its specific error variant + // so log/metric attribution stays distinct (PR #855 review #2). + + // 1024 bytes of 0xef: the original Hive panic payload. Rejected as + // InvalidVarint (every byte is a continuation byte; no terminator). + const garbage = [_]u8{0xef} ** 1024; + try std.testing.expectError( + error.InvalidVarint, + validateGossipSnappyHeader(&garbage, MAX_GOSSIP_BLOCK_SIZE), + ); + + // 11 continuation bytes then a terminator: varint > u64. + var long_varint: [12]u8 = undefined; + @memset(long_varint[0..11], 0xff); + long_varint[11] = 0x01; + try std.testing.expectError( + error.InvalidVarint, + validateGossipSnappyHeader(&long_varint, MAX_GOSSIP_BLOCK_SIZE), + ); + + // Empty payload. + const empty = [_]u8{}; + try std.testing.expectError( + error.EmptyMessage, + validateGossipSnappyHeader(&empty, MAX_GOSSIP_BLOCK_SIZE), + ); + + // Declared size exceeds the per-topic limit (oversized claim). + var oversize_buf: [MAX_VARINT_BYTES + 1]u8 = undefined; + const oversize_header = uvarint.encode(usize, MAX_GOSSIP_BLOCK_SIZE + 1, oversize_buf[0..MAX_VARINT_BYTES]); + oversize_buf[oversize_header.len] = 0x00; // payload byte so it isn't header-only + try std.testing.expectError( + error.DeclaredPayloadTooLarge, + validateGossipSnappyHeader(oversize_buf[0 .. oversize_header.len + 1], MAX_GOSSIP_BLOCK_SIZE), + ); + + // Header-only buffer for a non-zero declared size: header is valid, body + // is missing. Distinct error so callers can attribute truncated streams + // separately from corrupt headers (PR #855 review #7). + var header_only_buf: [MAX_VARINT_BYTES]u8 = undefined; + const header_only = uvarint.encode(usize, 32, &header_only_buf); + try std.testing.expectError( + error.HeaderWithoutBody, + validateGossipSnappyHeader(header_only, MAX_GOSSIP_BLOCK_SIZE), + ); + + // Well-formed header followed by at least one payload byte: accepted. + // The validator does *not* check that the body length matches the + // declared uncompressed size — that's the decoder's job. See doc comment + // on `validateSnappyHeader` (PR #855 review #12). + var ok_buf: [MAX_VARINT_BYTES + 1]u8 = undefined; + const ok_header = uvarint.encode(usize, 32, ok_buf[0..MAX_VARINT_BYTES]); + ok_buf[ok_header.len] = 0x00; + const ok = try validateGossipSnappyHeader(ok_buf[0 .. ok_header.len + 1], MAX_GOSSIP_BLOCK_SIZE); + try std.testing.expectEqual(@as(usize, 32), ok.value); + + // Zero-length declared payload with no body is accepted (snappy can + // legitimately describe an empty uncompressed block as just the varint 0). + const zero_header = [_]u8{0x00}; + const zero_ok = try validateGossipSnappyHeader(&zero_header, MAX_GOSSIP_BLOCK_SIZE); + try std.testing.expectEqual(@as(usize, 0), zero_ok.value); + + // Body shorter than declared but well-formed header: validator accepts + // by design. The decoder is authoritative for body integrity. Pinning + // current behaviour so a future change to also enforce body length here + // is a deliberate breaking change rather than a silent drift. + var short_body_buf: [MAX_VARINT_BYTES + 4]u8 = undefined; + const short_body_header = uvarint.encode(usize, 1024, short_body_buf[0..MAX_VARINT_BYTES]); + @memset(short_body_buf[short_body_header.len .. short_body_header.len + 4], 0x00); + const short_body_ok = try validateGossipSnappyHeader( + short_body_buf[0 .. short_body_header.len + 4], + MAX_GOSSIP_BLOCK_SIZE, + ); + try std.testing.expectEqual(@as(usize, 1024), short_body_ok.value); + + // Boundary: declared == max_size is accepted (strict `>`); declared == + // max_size + 1 is rejected. Pinned to match upstream + // `snappyz.decodeWithMax`'s `if (block.blockLen > max_size)` contract + // (PR #855 review #13). If upstream ever flips to `>=` this test will + // fail loudly instead of the validator silently disagreeing across a + // 1-byte gap. + var boundary_buf: [MAX_VARINT_BYTES + 1]u8 = undefined; + const at_limit_header = uvarint.encode(usize, MAX_GOSSIP_BLOCK_SIZE, boundary_buf[0..MAX_VARINT_BYTES]); + boundary_buf[at_limit_header.len] = 0x00; + const at_limit_ok = try validateGossipSnappyHeader( + boundary_buf[0 .. at_limit_header.len + 1], + MAX_GOSSIP_BLOCK_SIZE, + ); + try std.testing.expectEqual(@as(usize, MAX_GOSSIP_BLOCK_SIZE), at_limit_ok.value); +} + +test "snappyz.decodeWithMax regression canary: 1024 bytes of 0xef must not panic" { + // REGRESSION CANARY (PR #855 review #11): if this test ever panics the + // whole test binary instead of returning `error.Corrupt`, the upstream + // `zig-snappy` dependency has been downgraded below v0.0.5 and the + // uvarint integer-overflow fix is gone. Restore the pin in `build.zig.zon` + // before doing anything else — the gossip thread will crash on the next + // malformed payload. + // + // Belt-and-suspenders: even if the dep is rolled back, the + // `validateGossipSnappyHeader` test above still ensures the gossip + // handler short-circuits before reaching the decoder. + const garbage = [_]u8{0xef} ** 1024; + const result = snappyz.decodeWithMax(std.testing.allocator, &garbage, MAX_GOSSIP_BLOCK_SIZE); + try std.testing.expectError(error.Corrupt, result); }