From 672e73180a405b0f3fd320d889742855eee37993 Mon Sep 17 00:00:00 2001 From: ch4r10t33r Date: Sun, 10 May 2026 11:44:42 +0100 Subject: [PATCH 1/2] node,ingress_limit: inbound rate limits and outbound opening TTL (issue 22) - New ingress_limit: sliding-window global and per-peer caps, LRU-capped peer table. - Node.handleReceive rate limit before alloc/copy/decode; ingressRateKey for UDP endpoints. - Config.ingress and outbound_opening_ttl_ms; prune stale opening handshakes on receive. - allocOpeningPingHandshake takes now_ms for created_ms (API change). - Tests for limiter, per-peer cap, and outbound expiry. Closes #22. --- src/ingress_limit.zig | 157 ++++++++++++++++++++++++++++++++++++++++++ src/node.zig | 108 ++++++++++++++++++++++++++++- src/root.zig | 2 + 3 files changed, 264 insertions(+), 3 deletions(-) create mode 100644 src/ingress_limit.zig diff --git a/src/ingress_limit.zig b/src/ingress_limit.zig new file mode 100644 index 0000000..db7dfab --- /dev/null +++ b/src/ingress_limit.zig @@ -0,0 +1,157 @@ +//! Sliding-window counters for inbound datagrams (per peer key and global). +//! Peer keys are opaque **u64**s (e.g. IPv4 host+port); see **Node** for encoding. +//! Applied before crypto decode so cheap bogus traffic still consumes quota. + +const std = @import("std"); + +pub const RateLimited = error{RateLimited}; + +pub const Config = struct { + /// Max datagrams per peer key per window. Null disables. + per_peer_max_packets: ?u32 = null, + per_peer_window_ms: u64 = 1000, + /// Max datagrams from all peers combined per window. Null disables. + global_max_packets: ?u32 = null, + global_window_ms: u64 = 1000, + /// Max distinct peer keys tracked; LRU by **last_seen_ms** when full. + peer_table_cap: usize = 4096, +}; + +const PeerRec = struct { + count: u32, + window_start: u64, + last_seen_ms: u64, +}; + +pub const IngressLimiter = struct { + cfg: Config, + global_window_start: u64 = 0, + global_count: u32 = 0, + peers: std.AutoHashMapUnmanaged(u64, PeerRec) = .{}, + + pub fn init(cfg: Config) IngressLimiter { + return .{ .cfg = cfg }; + } + + pub fn deinit(self: *IngressLimiter, allocator: std.mem.Allocator) void { + self.peers.deinit(allocator); + self.* = undefined; + } + + fn resetGlobalWindow(self: *IngressLimiter, now_ms: u64) void { + self.global_count = 0; + self.global_window_start = now_ms; + } + + fn touchGlobal(self: *IngressLimiter, now_ms: u64) RateLimited!void { + const max = self.cfg.global_max_packets orelse return; + const win = self.cfg.global_window_ms; + if (win == 0) return error.RateLimited; + + if (self.global_count == 0) { + self.global_window_start = now_ms; + } else if (now_ms -| self.global_window_start >= win) { + self.resetGlobalWindow(now_ms); + } + + if (self.global_count >= max) return error.RateLimited; + self.global_count += 1; + } + + fn evictOldestPeer(self: *IngressLimiter) void { + if (self.peers.count() == 0) return; + var worst_k: u64 = 0; + var worst_t: u64 = std.math.maxInt(u64); + var it = self.peers.iterator(); + while (it.next()) |e| { + if (e.value_ptr.last_seen_ms < worst_t) { + worst_t = e.value_ptr.last_seen_ms; + worst_k = e.key_ptr.*; + } + } + _ = self.peers.remove(worst_k); + } + + fn touchPeer(self: *IngressLimiter, allocator: std.mem.Allocator, key: u64, now_ms: u64) (RateLimited || std.mem.Allocator.Error)!void { + const max = self.cfg.per_peer_max_packets orelse return; + const win = self.cfg.per_peer_window_ms; + if (win == 0) return error.RateLimited; + + const is_new = !self.peers.contains(key); + if (is_new and self.peers.count() >= self.cfg.peer_table_cap) { + self.evictOldestPeer(); + } + + const gop = try self.peers.getOrPut(allocator, key); + if (!gop.found_existing) { + gop.value_ptr.* = .{ + .count = 0, + .window_start = now_ms, + .last_seen_ms = now_ms, + }; + } + const rec = gop.value_ptr; + + rec.last_seen_ms = now_ms; + + if (rec.count == 0) { + rec.window_start = now_ms; + } else if (now_ms -| rec.window_start >= win) { + rec.count = 0; + rec.window_start = now_ms; + } + + if (rec.count >= max) return error.RateLimited; + rec.count += 1; + } + + /// Count this inbound datagram. Fails with **error.RateLimited** if a cap is exceeded. + pub fn recordInbound( + self: *IngressLimiter, + allocator: std.mem.Allocator, + peer_key: u64, + now_ms: u64, + ) (RateLimited || std.mem.Allocator.Error)!void { + const any_peer = self.cfg.per_peer_max_packets != null; + const any_global = self.cfg.global_max_packets != null; + if (!any_peer and !any_global) return; + + if (any_global) try self.touchGlobal(now_ms); + if (any_peer) try self.touchPeer(allocator, peer_key, now_ms); + } +}; + +test "ingress global window" { + const alloc = std.testing.allocator; + var lim = IngressLimiter.init(.{ + .global_max_packets = 2, + .global_window_ms = 1000, + .per_peer_max_packets = null, + }); + defer lim.deinit(alloc); + + try lim.recordInbound(alloc, 1, 0); + try lim.recordInbound(alloc, 2, 0); + try std.testing.expectError(error.RateLimited, lim.recordInbound(alloc, 3, 0)); + try lim.recordInbound(alloc, 1, 1000); +} + +test "ingress per-peer window" { + const alloc = std.testing.allocator; + var lim = IngressLimiter.init(.{ + .per_peer_max_packets = 2, + .per_peer_window_ms = 500, + .global_max_packets = null, + .peer_table_cap = 8, + }); + defer lim.deinit(alloc); + + const ka: u64 = 0x0100007f_3030; // arbitrary distinct keys + const kb: u64 = 0x0200007f_3030; + try lim.recordInbound(alloc, ka, 0); + try lim.recordInbound(alloc, ka, 0); + try std.testing.expectError(error.RateLimited, lim.recordInbound(alloc, ka, 0)); + try lim.recordInbound(alloc, kb, 0); + try lim.recordInbound(alloc, kb, 0); + try lim.recordInbound(alloc, ka, 500); +} diff --git a/src/node.zig b/src/node.zig index e620555..ade171f 100644 --- a/src/node.zig +++ b/src/node.zig @@ -12,6 +12,7 @@ const std = @import("std"); const builtin = @import("builtin"); const enr = @import("enr.zig"); +const ingress_limit = @import("ingress_limit.zig"); const handshake = @import("handshake.zig"); const identity_v4 = @import("identity_v4.zig"); const message = @import("message.zig"); @@ -27,6 +28,20 @@ pub const RemoteUdp = struct { port: u16, }; +/// Opaque key for **ingress_limit** (IPv4 address big-endian in high bits, port in low 16 bits; IPv6 uses Wyhash). +pub fn ingressRateKey(remote: RemoteUdp) u64 { + switch (remote.ip) { + .v4 => |b| { + const ipu = std.mem.readInt(u32, &b, .big); + return (@as(u64, ipu) << 16) | @as(u64, remote.port); + }, + .v6 => |b| { + const h = std.hash.Wyhash.hash(0, &b); + return h ^ (@as(u64, remote.port) << 48); + }, + } +} + pub const Config = struct { secret_key: [32]u8, enr_seq: u64 = 0, @@ -37,6 +52,10 @@ pub const Config = struct { pending_challenge_ttl_ms: ?u64 = null, /// Upper bound on pending WHOAREYOU entries; oldest is evicted when full. pending_challenge_cap: usize = 256, + /// Inbound datagram rate limits (applied before decode). Defaults disable limits. + ingress: ingress_limit.Config = .{}, + /// Drop **allocOpeningPingHandshake** state when no WHOAREYOU arrives within this time. Null disables. + outbound_opening_ttl_ms: ?u64 = null, }; /// Max number of logarithmic distances in one FINDNODE (discv5 clients typically use small lists). @@ -71,6 +90,7 @@ pub const Node = struct { peer_pubkey_compressed: [33]u8, remote: RemoteUdp, message_nonce: [12]u8, + created_ms: u64, }; const OutboundPing = struct { @@ -84,6 +104,8 @@ pub const Node = struct { enr_seq: u64, pending_challenge_ttl_ms: ?u64, pending_challenge_cap: usize, + outbound_opening_ttl_ms: ?u64, + ingress: ingress_limit.IngressLimiter, sessions: session.SessionTable, routing: routing.RoutingTable, /// Raw ENR RLP bytes keyed by node id (e.g. from verified inbound handshakes). Used for NODES replies. @@ -105,6 +127,8 @@ pub const Node = struct { .enr_seq = cfg.enr_seq, .pending_challenge_ttl_ms = cfg.pending_challenge_ttl_ms, .pending_challenge_cap = @max(1, cfg.pending_challenge_cap), + .outbound_opening_ttl_ms = cfg.outbound_opening_ttl_ms, + .ingress = ingress_limit.IngressLimiter.init(cfg.ingress), .sessions = sessions, .routing = routing.RoutingTable.init(nid), .peer_enrs = .empty, @@ -125,6 +149,7 @@ pub const Node = struct { for (self.outbound_pings.items) |e| self.allocator.free(e.req_id); self.outbound_pings.deinit(self.allocator); self.outbound.deinit(self.allocator); + self.ingress.deinit(self.allocator); self.sessions.deinit(); } @@ -182,6 +207,16 @@ pub const Node = struct { return .{ .node_id = peer_id, .ip = ip, .port = port }; } + fn pruneStaleOutbounds(self: *Node, now_ms: u64) void { + const ttl = self.outbound_opening_ttl_ms orelse return; + var i: usize = 0; + while (i < self.outbound.items.len) { + if (now_ms -| self.outbound.items[i].created_ms > ttl) { + _ = self.outbound.swapRemove(i); + } else i += 1; + } + } + fn pruneExpiredPending(self: *Node, now_ms: u64) void { const ttl = self.pending_challenge_ttl_ms orelse return; var i: usize = 0; @@ -254,6 +289,7 @@ pub const Node = struct { std.mem.Allocator.Error || std.crypto.errors.IdentityElementError || std.crypto.errors.NonCanonicalError || + ingress_limit.RateLimited || error{ MissingHandshakePending, EmptyHandshakeRecord, EnrNodeIdMismatch, BadHandshakeSignatureLength, FindnodeResponseTooLarge }; fn clearOutboundForPeer(self: *Node, peer_id: NodeId) void { @@ -285,6 +321,7 @@ pub const Node = struct { remote: RemoteUdp, req_id: []const u8, ping_enr_seq: u64, + now_ms: u64, ) OpeningHandshakeError![]u8 { const alloc = self.allocator; const derived = try identity_v4.nodeIdV4FromCompressedSec1(peer_pubkey_compressed); @@ -322,6 +359,7 @@ pub const Node = struct { .peer_pubkey_compressed = peer_pubkey_compressed, .remote = remote, .message_nonce = nonce, + .created_ms = now_ms, }); return pkt; } @@ -338,6 +376,12 @@ pub const Node = struct { const alloc = self.allocator; self.pruneExpiredPending(now_ms); + self.pruneStaleOutbounds(now_ms); + + self.ingress.recordInbound(alloc, ingressRateKey(remote), now_ms) catch |err| switch (err) { + error.RateLimited => return error.RateLimited, + else => |e| return e, + }; const copy = try alloc.dupe(u8, datagram); defer alloc.free(copy); @@ -774,6 +818,64 @@ fn buildMinimalEnrRlp(allocator: std.mem.Allocator, secret_key: [32]u8, compress return try enr_mod.encodeV4RecordSigned(allocator, secret_key, seq, pairs.items); } +test "handleReceive ingress per-peer cap" { + const alloc = std.testing.allocator; + + var sk: [32]u8 = @splat(0); + sk[31] = 3; + var node_b = try Node.init(alloc, .{ + .secret_key = sk, + .ingress = .{ + .per_peer_max_packets = 2, + .per_peer_window_ms = 60_000, + .global_max_packets = null, + }, + }); + defer node_b.deinit(); + + const remote: RemoteUdp = .{ .ip = .{ .v4 = .{ 203, 0, 113, 7 } }, .port = 7777 }; + var junk: [packet.min_packet_size]u8 = @splat(0xaa); + + var responses: std.ArrayList([]u8) = .empty; + defer { + for (responses.items) |s| alloc.free(s); + responses.deinit(alloc); + } + + try std.testing.expectError(error.InvalidProtocol, node_b.handleReceive(remote, &junk, &responses, 0)); + try std.testing.expectError(error.InvalidProtocol, node_b.handleReceive(remote, &junk, &responses, 1)); + try std.testing.expectError(error.RateLimited, node_b.handleReceive(remote, &junk, &responses, 2)); +} + +test "outbound opening handshake state expires" { + const alloc = std.testing.allocator; + + var sk_a: [32]u8 = @splat(0); + sk_a[31] = 41; + var node_a = try Node.init(alloc, .{ .secret_key = sk_a, .outbound_opening_ttl_ms = 50 }); + defer node_a.deinit(); + + var sk_b: [32]u8 = @splat(0); + sk_b[31] = 43; + const id_b = try identity_v4.nodeIdV4FromSecretKey(sk_b); + const pk_b = try identity_v4.compressedPubkeyFromSecretKey(sk_b); + const remote_b: RemoteUdp = .{ .ip = .{ .v4 = .{ 10, 0, 0, 2 } }, .port = 40000 }; + + const pkt = try node_a.allocOpeningPingHandshake(id_b, pk_b, remote_b, &.{0xab}, 1, 0); + defer alloc.free(pkt); + try std.testing.expectEqual(@as(usize, 1), node_a.outbound.items.len); + + var junk: [packet.min_packet_size]u8 = @splat(0xbb); + var responses: std.ArrayList([]u8) = .empty; + defer { + for (responses.items) |s| alloc.free(s); + responses.deinit(alloc); + } + const r: RemoteUdp = .{ .ip = .{ .v4 = .{ 9, 9, 9, 9 } }, .port = 1 }; + try std.testing.expectError(error.InvalidProtocol, node_a.handleReceive(r, &junk, &responses, 100)); + try std.testing.expectEqual(@as(usize, 0), node_a.outbound.items.len); +} + test "unknown session ordinary yields WHOAREYOU with echoed nonce" { const alloc = std.testing.allocator; @@ -1093,7 +1195,7 @@ test "initiator opening ping completes handshake after WHOAREYOU" { const remote_a: RemoteUdp = .{ .ip = .{ .v4 = .{ 10, 0, 0, 11 } }, .port = 51111 }; const remote_b: RemoteUdp = .{ .ip = .{ .v4 = .{ 10, 0, 0, 22 } }, .port = 52222 }; - const ordinary = try node_a.allocOpeningPingHandshake(id_b, pk_b, remote_b, &.{ 0xca, 0xfe }, 9); + const ordinary = try node_a.allocOpeningPingHandshake(id_b, pk_b, remote_b, &.{ 0xca, 0xfe }, 9, 0); defer alloc.free(ordinary); var from_b: std.ArrayList([]u8) = .empty; @@ -1189,7 +1291,7 @@ test "initiator omits handshake record when WHOAREYOU enr-seq matches cached seq const remote_a: RemoteUdp = .{ .ip = .{ .v4 = .{ 10, 0, 0, 11 } }, .port = 51111 }; const remote_b: RemoteUdp = .{ .ip = .{ .v4 = .{ 10, 0, 0, 22 } }, .port = 52222 }; - const ordinary1 = try node_a.allocOpeningPingHandshake(id_b, pk_b, remote_b, &.{0x10}, 9); + const ordinary1 = try node_a.allocOpeningPingHandshake(id_b, pk_b, remote_b, &.{0x10}, 9, 0); defer alloc.free(ordinary1); var from_b: std.ArrayList([]u8) = .empty; @@ -1230,7 +1332,7 @@ test "initiator omits handshake record when WHOAREYOU enr-seq matches cached seq _ = node_a.sessions.remove(ep_on_a); _ = node_b.sessions.remove(ep_on_b); - const ordinary2 = try node_a.allocOpeningPingHandshake(id_b, pk_b, remote_b, &.{0x20}, 1); + const ordinary2 = try node_a.allocOpeningPingHandshake(id_b, pk_b, remote_b, &.{0x20}, 1, 0); defer alloc.free(ordinary2); for (from_b.items) |s| alloc.free(s); diff --git a/src/root.zig b/src/root.zig index 25b8480..099b6b8 100644 --- a/src/root.zig +++ b/src/root.zig @@ -14,6 +14,7 @@ pub const packet = @import("packet.zig"); pub const message = @import("message.zig"); pub const message_crypto = @import("message_crypto.zig"); pub const routing = @import("routing.zig"); +pub const ingress_limit = @import("ingress_limit.zig"); pub const identity_v4 = @import("identity_v4.zig"); pub const session = @import("session.zig"); pub const topic = @import("topic.zig"); @@ -32,6 +33,7 @@ test { _ = message; _ = message_crypto; _ = routing; + _ = ingress_limit; _ = identity_v4; _ = session; _ = topic; From 2f06e1a68e6f9843f8bc0262747ec44d0feeac0e Mon Sep 17 00:00:00 2001 From: ch4r10t33r Date: Sun, 10 May 2026 11:47:53 +0100 Subject: [PATCH 2/2] udp_runtime: pumpOnceEx egress limiter (issue 22) - PumpOpts.egress_limiter and pumpOnceEx; pumpOnce delegates with empty opts. - PumpResult enum for stable return type across pumpOnce/pumpOnceEx. - ingress_limit module doc: same limiter for outbound counting. - Node module doc: cross-reference pumpOnceEx egress. - Test: per-peer egress cap blocks second reply to same UDP source. --- src/ingress_limit.zig | 7 +-- src/node.zig | 3 ++ src/udp_runtime.zig | 110 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 115 insertions(+), 5 deletions(-) diff --git a/src/ingress_limit.zig b/src/ingress_limit.zig index db7dfab..52efa0f 100644 --- a/src/ingress_limit.zig +++ b/src/ingress_limit.zig @@ -1,6 +1,7 @@ -//! Sliding-window counters for inbound datagrams (per peer key and global). -//! Peer keys are opaque **u64**s (e.g. IPv4 host+port); see **Node** for encoding. -//! Applied before crypto decode so cheap bogus traffic still consumes quota. +//! Sliding-window packet counters (per peer key and global). +//! Peer keys are opaque **u64**s (e.g. IPv4 host+port); see **Node.ingressRateKey**. +//! For **inbound**, call **recordInbound** before allocate/decode so bogus traffic still consumes quota. +//! The same **IngressLimiter** type can bound **outbound** sends (e.g. **udp_runtime.pumpOnceEx**). const std = @import("std"); diff --git a/src/node.zig b/src/node.zig index ade171f..2f101b4 100644 --- a/src/node.zig +++ b/src/node.zig @@ -8,6 +8,9 @@ //! marks the peer verified; for strict PING tracking call **registerOutboundPing** before sending an encrypted //! PING, then an inbound **PONG** with the same **req_id** sets **ping_replied**. Use **RoutingTable.pickStaleBucketForRefresh** //! / **markBucketRefreshed** for periodic bucket refresh. +//! +//! Operational limits: **Config.ingress** and **udp_runtime.pumpOnceEx** with **PumpOpts.egress_limiter** (same +//! **ingress_limit.IngressLimiter** type for send-side windows). const std = @import("std"); const builtin = @import("builtin"); diff --git a/src/udp_runtime.zig b/src/udp_runtime.zig index 67b74b5..ada5cb3 100644 --- a/src/udp_runtime.zig +++ b/src/udp_runtime.zig @@ -6,6 +6,7 @@ const std = @import("std"); const builtin = @import("builtin"); const node_mod = @import("node.zig"); +const ingress_limit = @import("ingress_limit.zig"); const identity_v4 = @import("identity_v4.zig"); const message = @import("message.zig"); const message_crypto = @import("message_crypto.zig"); @@ -118,7 +119,15 @@ pub fn sendDatagram(sock: UdpSocket, remote: RemoteUdp, payload: []const u8) Sen } } -pub const PumpError = RecvError || SendError || Node.ReceiveError || std.mem.Allocator.Error; +pub const PumpError = RecvError || SendError || Node.ReceiveError || ingress_limit.RateLimited || std.mem.Allocator.Error; + +pub const PumpResult = enum { idle, progressed }; + +/// Optional limits for **pumpOnceEx** (send-side uses the same sliding-window type as inbound). +pub const PumpOpts = struct { + /// If non-null, each outbound datagram to the source address counts before **sendDatagram**; excess returns **error.RateLimited** (unsent replies are freed in defer). + egress_limiter: ?*ingress_limit.IngressLimiter = null, +}; /// Receives at most one datagram, runs `Node.handleReceive`, and sends each reply to the source address. /// `responses` must be empty on entry; allocated replies are freed before returning. @@ -131,7 +140,21 @@ pub fn pumpOnce( responses: *std.ArrayList([]u8), recv_flags: u32, now_ms: u64, -) PumpError!enum { idle, progressed } { +) PumpError!PumpResult { + return pumpOnceEx(allocator, sock, node_ptr, recv_buf, responses, recv_flags, now_ms, .{}); +} + +/// Like **pumpOnce** with optional **egress_limiter** for per-pump send caps. +pub fn pumpOnceEx( + allocator: std.mem.Allocator, + sock: UdpSocket, + node_ptr: *Node, + recv_buf: []u8, + responses: *std.ArrayList([]u8), + recv_flags: u32, + now_ms: u64, + opts: PumpOpts, +) PumpError!PumpResult { std.debug.assert(responses.items.len == 0); const got = try recvDatagram(sock, recv_buf, recv_flags) orelse return .idle; @@ -143,7 +166,11 @@ pub fn pumpOnce( responses.clearRetainingCapacity(); } + const peer_key = node_mod.ingressRateKey(got.remote); for (responses.items) |pkt| { + if (opts.egress_limiter) |lim| { + try lim.recordInbound(allocator, peer_key, now_ms); + } try sendDatagram(sock, got.remote, pkt); } return .progressed; @@ -232,3 +259,82 @@ test "UDP pump sends WHOAREYOU to peer socket" { try std.testing.expect(parsed.header.flag == .whoareyou); try std.testing.expectEqualSlices(u8, &nonce, &parsed.header.nonce); } + +test "pumpOnceEx egress limiter blocks second reply burst to same peer" { + const alloc = std.testing.allocator; + + var egress = ingress_limit.IngressLimiter.init(.{ + .per_peer_max_packets = 1, + .per_peer_window_ms = 60_000, + .global_max_packets = null, + }); + defer egress.deinit(alloc); + + var server = try UdpSocket.initIpv4Udp(); + defer server.close(); + try server.bindIpv4Any(0); + const server_port = try server.localPort(); + + var client = try UdpSocket.initIpv4Udp(); + defer client.close(); + try client.bindIpv4Any(0); + + var sk_b: [32]u8 = @splat(0); + sk_b[31] = 3; + var node_b = try Node.init(alloc, .{ .secret_key = sk_b, .enr_seq = 9 }); + defer node_b.deinit(); + + var sk_a: [32]u8 = @splat(0); + sk_a[31] = 5; + const id_a = try identity_v4.nodeIdV4FromSecretKey(sk_a); + + var iv: [16]u8 = undefined; + @memset(&iv, 0x2a); + var nonce: [12]u8 = undefined; + for (&nonce, 0..) |*b, i| b.* = @truncate(i + 1); + + const ping_pt = try message.encodePingPlaintext(alloc, &.{0x07}, 4); + defer alloc.free(ping_pt); + + const key = [_]u8{0x55} ** 16; + var prefix: [packet.static_prefix_size + packet.message_auth_size]u8 = undefined; + @memcpy(prefix[0..16], &iv); + var static_plain: [packet.static_header_size]u8 = undefined; + packet.writePlaintextStaticHeader(&static_plain, .message, nonce, packet.message_auth_size); + @memcpy(prefix[16..][0..packet.static_header_size], &static_plain); + @memcpy(prefix[packet.static_prefix_size..][0..packet.message_auth_size], &id_a); + + const ct = try message_crypto.encryptMessage(alloc, key, nonce, ping_pt, &prefix); + defer alloc.free(ct); + + const ordinary = try packet.encodeOrdinaryMessagePacket(alloc, node_b.node_id, iv, nonce, id_a, ct); + defer alloc.free(ordinary); + + const dst: RemoteUdp = .{ .ip = .{ .v4 = .{ 127, 0, 0, 1 } }, .port = server_port }; + try sendDatagram(client, dst, ordinary); + + var recv_buf: [packet.max_packet_size]u8 = undefined; + var responses: std.ArrayList([]u8) = .empty; + defer responses.deinit(alloc); + + const st1 = try pumpOnceEx(alloc, server, &node_b, &recv_buf, &responses, 0, 0, .{ .egress_limiter = &egress }); + try std.testing.expectEqual(@as(@TypeOf(st1), .progressed), st1); + + var reply_buf: [packet.max_packet_size]u8 = undefined; + _ = (try recvDatagram(client, &reply_buf, 0)) orelse unreachable; + + var nonce2: [12]u8 = undefined; + for (&nonce2, 0..) |*b, i| b.* = @truncate(i + 3); + const ping_pt2 = try message.encodePingPlaintext(alloc, &.{0x08}, 4); + defer alloc.free(ping_pt2); + packet.writePlaintextStaticHeader(&static_plain, .message, nonce2, packet.message_auth_size); + @memcpy(prefix[16..][0..packet.static_header_size], &static_plain); + @memcpy(prefix[packet.static_prefix_size..][0..packet.message_auth_size], &id_a); + const ct2 = try message_crypto.encryptMessage(alloc, key, nonce2, ping_pt2, &prefix); + defer alloc.free(ct2); + const ordinary2 = try packet.encodeOrdinaryMessagePacket(alloc, node_b.node_id, iv, nonce2, id_a, ct2); + defer alloc.free(ordinary2); + try sendDatagram(client, dst, ordinary2); + + try std.testing.expectError(error.RateLimited, pumpOnceEx(alloc, server, &node_b, &recv_buf, &responses, 0, 0, .{ .egress_limiter = &egress })); +}