fix: proxy dos caps — connections, hpack, http2 streams

src/network/proxy/hpack.zig

@@ -31,6 +31,15 @@ pub const IntegerDecode = struct {
 };
 
 const dynamic_table_default_max_size = 4096;
+
+// DoS bounds for decoding attacker-controlled header blocks. HPACK integers
+// and string lengths are otherwise unbounded; a malicious peer could drive a
+// near-usize length (overflowing the `start + value` arithmetic in
+// decodeString → panic) or list enough headers to exhaust memory.
+const max_hpack_integer: usize = 1 << 24; // 16 MiB — far above any real header
+const max_headers_per_block: usize = 256; // generous; real requests are <~100
+const max_decoded_header_bytes: usize = 1 << 20; // 1 MiB total per block
+const max_dynamic_table_size: usize = 1 << 20; // cap a peer's table-size update
 const static_table = [_]StaticHeaderField{
     .{ .name = ":authority", .value = "" },
     .{ .name = ":method", .value = "GET" },
@@ -117,6 +126,9 @@ pub fn decodeInteger(buf: []const u8, prefix_bits: u3) DecodeError!IntegerDecode
         const payload = byte & 0x7f;
         if (shift >= @bitSizeOf(usize)) return error.IntegerOverflow;
         value += (@as(usize, payload) << @intCast(shift));
+        // bound the running value so a crafted continuation can't drive it
+        // near usize-max (which would overflow downstream length arithmetic).
+        if (value > max_hpack_integer) return error.IntegerOverflow;
         consumed += 1;
         if ((byte & 0x80) == 0) break;
         shift += 7;
@@ -153,12 +165,19 @@ pub fn decodeHeaderBlock(alloc: std.mem.Allocator, block: []const u8) Error!std.
     var dynamic_table: DynamicTable = .{};
     defer dynamic_table.deinit(alloc);
 
+    // running total of decoded name+value bytes — bounds total memory even
+    // when individual fields are small but numerous.
+    var decoded_bytes: usize = 0;
+
     var pos: usize = 0;
     while (pos < block.len) {
+        if (headers.items.len >= max_headers_per_block) return error.IntegerOverflow;
         const byte = block[pos];
         if ((byte & 0x80) != 0) {
             const index_info = try decodeInteger(block[pos..], 7);
             const field = lookupHeader(index_info.value, &dynamic_table) orelse return error.InvalidIndex;
+            decoded_bytes += field.name.len + field.value.len;
+            if (decoded_bytes > max_decoded_header_bytes) return error.IntegerOverflow;
             try headers.append(alloc, .{
                 .name = try alloc.dupe(u8, field.name),
                 .value = try alloc.dupe(u8, field.value),
@@ -196,6 +215,9 @@ pub fn decodeHeaderBlock(alloc: std.mem.Allocator, block: []const u8) Error!std.
         };
         errdefer alloc.free(value);
 
+        decoded_bytes += name.len + value.len;
+        if (decoded_bytes > max_decoded_header_bytes) return error.IntegerOverflow;
+
         try headers.append(alloc, .{ .name = name, .value = value });
         if (incremental_indexing) {
             try dynamic_table.add(alloc, name, value);
@@ -243,7 +265,8 @@ const DynamicTable = struct {
     }
 
     fn updateMaxSize(self: *DynamicTable, alloc: std.mem.Allocator, new_max_size: usize) void {
-        self.max_size = new_max_size;
+        // clamp a peer's table-size update so it can't pin large memory.
+        self.max_size = @min(new_max_size, max_dynamic_table_size);
         self.evictToLimit(alloc);
     }
 
@@ -499,3 +522,33 @@ test "decodeHeaderBlock reuses incremental dynamic entries" {
     try std.testing.expectEqualStrings("x-test", headers.items[1].name);
     try std.testing.expectEqualStrings("ok", headers.items[1].value);
 }
+
+test "decodeInteger rejects a value above the cap" {
+    // 5-bit prefix all-ones, then continuation bytes that accumulate past
+    // max_hpack_integer. 0xff prefix + 0xff*N continuation + terminator.
+    var buf: [16]u8 = undefined;
+    buf[0] = 0x1f; // prefix = 31 (5-bit all ones), enter continuation
+    var i: usize = 1;
+    while (i < buf.len - 1) : (i += 1) buf[i] = 0xff;
+    buf[buf.len - 1] = 0x7f; // terminator with high bits set
+    try std.testing.expectError(error.IntegerOverflow, decodeInteger(&buf, 5));
+}
+
+test "decodeHeaderBlock rejects too many headers" {
+    const alloc = std.testing.allocator;
+    // 0xbe is an indexed header (dynamic index resolving to a static-ish
+    // entry); simpler: repeat an indexed static field (0x82 = :method GET).
+    var block: std.ArrayList(u8) = .empty;
+    defer block.deinit(alloc);
+    var i: usize = 0;
+    while (i < max_headers_per_block + 1) : (i += 1) try block.append(alloc, 0x82);
+
+    try std.testing.expectError(error.IntegerOverflow, decodeHeaderBlock(alloc, block.items));
+}
+
+test "updateMaxSize clamps a peer's oversized table-size update" {
+    var table: DynamicTable = .{};
+    defer table.deinit(std.testing.allocator);
+    table.updateMaxSize(std.testing.allocator, 1 << 30); // 1 GiB request
+    try std.testing.expectEqual(max_dynamic_table_size, table.max_size);
+}

src/network/proxy/http2_connection_router.zig

@@ -15,6 +15,12 @@ const upstream_mod = @import("upstream.zig");
 const ip = @import("../ip.zig");
 const hpack = @import("hpack.zig");
 
+// DoS bounds for a single HTTP/2 connection. without these a client can open
+// unlimited concurrent streams or stall mid-frame while streaming bytes,
+// growing the per-connection buffers until OOM.
+const max_concurrent_streams: usize = 128;
+const max_downstream_buf_bytes: usize = 1 << 20; // 1 MiB of un-parsed frames
+
 pub fn proxyConnection(
     alloc: std.mem.Allocator,
     routes: []const router.Route,
@@ -122,6 +128,10 @@ const ConnectionRouter = struct {
                 const bytes_read = posix.read(self.client_fd, &buf) catch return error.ReceiveFailed;
                 if (bytes_read == 0) break;
                 try self.downstream_buf.appendSlice(self.allocator, buf[0..bytes_read]);
+                // bound un-parsed downstream bytes: a client that streams data
+                // without ever completing a frame would otherwise grow this
+                // without limit.
+                if (self.downstream_buf.items.len > max_downstream_buf_bytes) return error.ReceiveFailed;
                 self.last_activity_ms = nowMs();
             } else if (poll_fds.items[0].revents & (posix.POLL.ERR | posix.POLL.HUP) != 0) {
                 break;
@@ -327,6 +337,16 @@ const ConnectionRouter = struct {
             return;
         }
 
+        // cap concurrent streams: refuse a new stream past the limit rather
+        // than grow `streams` (and dial upstreams) without bound. handled
+        // before any upstream dial so no fd is leaked.
+        if (self.streams.items.len >= max_concurrent_streams) {
+            try self.sendLocalStreamResponse(parsed.request.stream_id, .service_unavailable, "{\"error\":\"too many concurrent streams\"}");
+            proxy_runtime.recordResponse(.service_unavailable);
+            try self.consumeDownstreamBytes(parsed.consumed);
+            return;
+        }
+
         const route = self.matchRouteForParsedRequest(parsed) orelse {
             try self.sendLocalStreamResponse(parsed.request.stream_id, .not_found, "{\"error\":\"route not found\"}");
             proxy_runtime.recordResponse(.not_found);

src/network/proxy/listener_runtime.zig

@@ -13,6 +13,14 @@ pub const StateChangeHook = *const fn () void;
 pub const drain_timeout_ms: u64 = 5000;
 const drain_poll_interval_ms: u64 = 10;
 
+/// upper bound on concurrent L7 proxy connections. each connection is a
+/// detached thread; without this cap a connection flood spawns threads
+/// until the OS thread/fd limit is hit. mirrors the API server's
+/// `max_connections` guard (src/api/server/connection_runtime.zig). at the
+/// cap, new connections are accepted and immediately closed so the kernel
+/// accept queue keeps draining rather than backing up.
+pub const max_connections: u32 = 1024;
+
 pub const Snapshot = struct {
     enabled: bool,
     running: bool,
@@ -301,10 +309,12 @@ fn acceptLoop(alloc: std.mem.Allocator) void {
             break;
         }
 
-        mutex.lockUncancelable(std.Options.debug_io);
-        accepted_connections_total += 1;
-        active_connections += 1;
-        mutex.unlock(std.Options.debug_io);
+        // shed load at the cap: close immediately rather than spawn an
+        // unbounded number of worker threads under a connection flood.
+        if (!admitConnection()) {
+            linux_platform.posix.close(client_fd);
+            continue;
+        }
 
         const thread = std.Thread.spawn(.{}, connectionWorker, .{ alloc, client_fd }) catch {
             mutex.lockUncancelable(std.Options.debug_io);
@@ -318,6 +328,19 @@ fn acceptLoop(alloc: std.mem.Allocator) void {
     }
 }
 
+/// count an accepted connection and decide whether to admit it. always bumps
+/// `accepted_connections_total`; admits (and bumps `active_connections`) only
+/// when below `max_connections`. returns false at the cap so the caller sheds
+/// the connection. mutex-guarded so it's safe under the accept loop.
+fn admitConnection() bool {
+    mutex.lockUncancelable(std.Options.debug_io);
+    defer mutex.unlock(std.Options.debug_io);
+    accepted_connections_total += 1;
+    if (active_connections >= max_connections) return false;
+    active_connections += 1;
+    return true;
+}
+
 fn connectionWorker(alloc: std.mem.Allocator, client_fd: posix.fd_t) void {
     defer {
         mutex.lockUncancelable(std.Options.debug_io);
@@ -432,3 +455,27 @@ test "listener drain observes active connection count" {
 
     try std.testing.expect(waitForConnectionsToDrain(0));
 }
+
+test "admitConnection sheds load at the connection cap" {
+    resetForTest();
+    defer resetForTest();
+
+    // seed active_connections to one below the cap: the next admit succeeds,
+    // the one after is shed.
+    mutex.lockUncancelable(std.Options.debug_io);
+    active_connections = max_connections - 1;
+    mutex.unlock(std.Options.debug_io);
+
+    try std.testing.expect(admitConnection()); // reaches the cap
+    try std.testing.expect(!admitConnection()); // over the cap → shed
+    try std.testing.expect(!admitConnection());
+
+    // active_connections never exceeds the cap; both rejected attempts still
+    // counted toward accepted_connections_total.
+    mutex.lockUncancelable(std.Options.debug_io);
+    const active = active_connections;
+    const accepted = accepted_connections_total;
+    mutex.unlock(std.Options.debug_io);
+    try std.testing.expectEqual(max_connections, active);
+    try std.testing.expectEqual(@as(u64, 3), accepted);
+}