node,ingress_limit: operational hardening — rate limits and opening TTL (closes #22)

src/ingress_limit.zig

@@ -0,0 +1,158 @@
+//! 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");
+
+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);
+}

src/node.zig

@@ -8,10 +8,14 @@
 //! 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");
 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 +31,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 +55,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 +93,7 @@ pub const Node = struct {
         peer_pubkey_compressed: [33]u8,
         remote: RemoteUdp,
         message_nonce: [12]u8,
+        created_ms: u64,
     };
 
     const OutboundPing = struct {
@@ -84,6 +107,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 +130,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 +152,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 +210,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 +292,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 +324,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 +362,7 @@ pub const Node = struct {
             .peer_pubkey_compressed = peer_pubkey_compressed,
             .remote = remote,
             .message_nonce = nonce,
+            .created_ms = now_ms,
         });
         return pkt;
     }
@@ -338,6 +379,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 +821,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 +1198,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 +1294,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 +1335,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);