diff --git a/src/network/proxy/hpack.zig b/src/network/proxy/hpack.zig index 0aeefd0..e5e5157 100644 --- a/src/network/proxy/hpack.zig +++ b/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); +} diff --git a/src/network/proxy/http2_connection_router.zig b/src/network/proxy/http2_connection_router.zig index f1b2afd..87c836f 100644 --- a/src/network/proxy/http2_connection_router.zig +++ b/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); diff --git a/src/network/proxy/listener_runtime.zig b/src/network/proxy/listener_runtime.zig index 0ed3578..006f2c7 100644 --- a/src/network/proxy/listener_runtime.zig +++ b/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); +}