From 1487a96f67ae6747b549f1072e943da5c8dcf63f Mon Sep 17 00:00:00 2001 From: k2safe Date: Tue, 22 Sep 2026 22:38:37 +0800 Subject: [PATCH] feat: harden experimental HTTP2 request limits --- CHANGELOG.md | 1 + transport/http2/server.go | 21 +++- transport/http2/server_test.go | 181 +++++++++++++++++++++++++++++++++ transport/native/HTTP2.md | 9 +- 4 files changed, 206 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c6def9..9e7cf31 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## Unreleased - 新增独立 `transport/http2` 实验 Runner：支持 h2c/TLS h2、连接上限、全局请求并发闸门、GOAWAY 排空、RST_STREAM 单流取消和生命周期统计；不改变 native HTTP/1.1 默认路径。 +- HTTP/2 实验 Runner 增加默认 8 MiB 请求体上限、128 路同连接多路复用和 2 MiB 慢读取流控验收；超限请求在进入业务 handler 前返回 413。 - native TLS 明确限制 ALPN 为 `http/1.1`：缺省自动宣告该协议，误配置 `h2` 等未实现协议时在启动阶段报错。 - native `Server.Stats()` 新增连接和请求过载拒绝计数；新增可信代理配置，访问日志只在 socket 对端属于配置网段时解析 `X-Forwarded-For`。 - 新增 transport-neutral `metrics` 包，支持请求数、错误数、in-flight、累计耗时和状态分布，并可直接注册 `/metrics` 接口。 diff --git a/transport/http2/server.go b/transport/http2/server.go index db0de49..6df4e63 100644 --- a/transport/http2/server.go +++ b/transport/http2/server.go @@ -23,6 +23,7 @@ import ( const ( defaultShutdownTimeout = 10 * time.Second defaultTLSHandshake = 10 * time.Second + defaultMaxBodyBytes = 8 << 20 http2ALPNProtocol = "h2" ) @@ -34,6 +35,7 @@ type Server struct { MaxConnections int MaxConcurrentRequests int MaxConcurrentStreams uint32 + MaxBodyBytes int64 IdleTimeout time.Duration TLSHandshakeTimeout time.Duration ShutdownTimeout time.Duration @@ -71,7 +73,11 @@ type Stats struct { // New creates an HTTP/2 transport with the supplied handler. The handler may // also be supplied by App.RunWith through the Runner interface. func New(handler http.Handler) *Server { - return &Server{handler: handler, conns: make(map[net.Conn]struct{})} + return &Server{ + handler: handler, + conns: make(map[net.Conn]struct{}), + MaxBodyBytes: defaultMaxBodyBytes, + } } // Stats returns a lock-free snapshot suitable for a metrics handler. @@ -203,7 +209,18 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error { } s.mu.Unlock() - gate := &requestGate{server: s, next: s.handler, slots: s.requestSlots} + requestHandler := http.Handler(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if s.MaxBodyBytes > 0 { + if request.ContentLength > s.MaxBodyBytes { + _ = request.Body.Close() + http.Error(writer, "request body too large\n", http.StatusRequestEntityTooLarge) + return + } + request.Body = http.MaxBytesReader(writer, request.Body, s.MaxBodyBytes) + } + s.handler.ServeHTTP(writer, request) + })) + gate := &requestGate{server: s, next: requestHandler, slots: s.requestSlots} protocolServer := &xhttp2.Server{ MaxConcurrentStreams: s.MaxConcurrentStreams, IdleTimeout: s.IdleTimeout, diff --git a/transport/http2/server_test.go b/transport/http2/server_test.go index c335b2c..4f070da 100644 --- a/transport/http2/server_test.go +++ b/transport/http2/server_test.go @@ -1,6 +1,7 @@ package http2 import ( + "bytes" "context" "crypto/rand" "crypto/rsa" @@ -8,10 +9,13 @@ import ( "crypto/x509" "crypto/x509/pkix" "errors" + "fmt" + "io" "math/big" "net" "net/http" "sync" + "sync/atomic" "testing" "time" @@ -122,6 +126,183 @@ func TestServerEnforcesGlobalRequestLimitAndRecovers(t *testing.T) { } } +func TestServerRejectsBodyAboveLimitAndRecovers(t *testing.T) { + server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + body, err := io.ReadAll(request.Body) + if err != nil { + http.Error(writer, err.Error(), http.StatusRequestEntityTooLarge) + return + } + _, _ = writer.Write(body) + })) + server.MaxBodyBytes = 4 + address, stop := startServer(t, server) + defer stop() + transport := &xhttp2.Transport{ + AllowHTTP: true, + DialTLSContext: func(ctx context.Context, network, address string, _ *tls.Config) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, address) + }, + } + client := &http.Client{Transport: transport, Timeout: 3 * time.Second} + defer transport.CloseIdleConnections() + + request, err := http.NewRequest(http.MethodPost, "http://"+address+"/too-large", bytes.NewReader([]byte("12345"))) + if err != nil { + t.Fatal(err) + } + response, err := client.Do(request) + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != http.StatusRequestEntityTooLarge { + t.Fatalf("expected 413 for oversized body, got %d", response.StatusCode) + } + + request, err = http.NewRequest(http.MethodPost, "http://"+address+"/ok", bytes.NewReader([]byte("1234"))) + if err != nil { + t.Fatal(err) + } + response, err = client.Do(request) + if err != nil { + t.Fatal(err) + } + body, readErr := io.ReadAll(response.Body) + response.Body.Close() + if readErr != nil || response.StatusCode != http.StatusOK || string(body) != "1234" { + t.Fatalf("body limit did not recover: status=%d body=%q err=%v", response.StatusCode, body, readErr) + } +} + +func TestServerMultiplexes128Streams(t *testing.T) { + started := make(chan struct{}, 128) + release := make(chan struct{}) + var active, maximum atomic.Int32 + server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + current := active.Add(1) + for { + seen := maximum.Load() + if current <= seen || maximum.CompareAndSwap(seen, current) { + break + } + } + started <- struct{}{} + <-release + active.Add(-1) + _, _ = io.WriteString(writer, request.URL.Path) + })) + server.MaxConcurrentStreams = 128 + address, stop := startServer(t, server) + defer stop() + transport := &xhttp2.Transport{ + AllowHTTP: true, + DialTLSContext: func(ctx context.Context, network, address string, _ *tls.Config) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, address) + }, + } + client := &http.Client{Transport: transport, Timeout: 5 * time.Second} + defer transport.CloseIdleConnections() + + type multiplexedResponse struct { + response *http.Response + expected string + } + responses := make(chan multiplexedResponse, 128) + errorsCh := make(chan error, 128) + for i := 0; i < 128; i++ { + go func(i int) { + response, err := client.Get(fmt.Sprintf("http://%s/stream-%d", address, i)) + if err != nil { + errorsCh <- err + return + } + responses <- multiplexedResponse{response: response, expected: fmt.Sprintf("/stream-%d", i)} + }(i) + } + for i := 0; i < 128; i++ { + select { + case <-started: + case err := <-errorsCh: + close(release) + t.Fatal(err) + case <-time.After(5 * time.Second): + close(release) + t.Fatal("HTTP/2 streams did not reach the handler") + } + } + if got := maximum.Load(); got < 128 { + close(release) + t.Fatalf("expected 128 concurrent streams, got %d", got) + } + if stats := server.Stats(); stats.ActiveConnections != 1 { + close(release) + t.Fatalf("expected one multiplexed connection, got %+v", stats) + } + close(release) + for i := 0; i < 128; i++ { + select { + case result := <-responses: + body, err := io.ReadAll(result.response.Body) + result.response.Body.Close() + if err != nil || result.response.StatusCode != http.StatusOK || string(body) != result.expected { + t.Fatalf("unexpected multiplexed response: status=%d body=%q expected=%q err=%v", result.response.StatusCode, body, result.expected, err) + } + case err := <-errorsCh: + t.Fatal(err) + case <-time.After(5 * time.Second): + t.Fatal("HTTP/2 streams did not finish") + } + } +} + +func TestServerStreamsLargeBodyWithFlowControl(t *testing.T) { + server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + var total int64 + buffer := make([]byte, 16<<10) + for { + n, err := request.Body.Read(buffer) + total += int64(n) + if n > 0 { + time.Sleep(time.Millisecond) + } + if err == io.EOF { + break + } + if err != nil { + http.Error(writer, err.Error(), http.StatusBadRequest) + return + } + } + _, _ = fmt.Fprintf(writer, "%d", total) + })) + server.MaxBodyBytes = 4 << 20 + address, stop := startServer(t, server) + defer stop() + transport := &xhttp2.Transport{ + AllowHTTP: true, + DialTLSContext: func(ctx context.Context, network, address string, _ *tls.Config) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, address) + }, + } + client := &http.Client{Transport: transport, Timeout: 10 * time.Second} + defer transport.CloseIdleConnections() + payload := bytes.Repeat([]byte("x"), 2<<20) + request, err := http.NewRequest(http.MethodPost, "http://"+address+"/large", bytes.NewReader(payload)) + if err != nil { + t.Fatal(err) + } + response, err := client.Do(request) + if err != nil { + t.Fatal(err) + } + body, readErr := io.ReadAll(response.Body) + response.Body.Close() + if readErr != nil || response.StatusCode != http.StatusOK || string(body) != "2097152" { + t.Fatalf("large body flow-control result: status=%d body=%q err=%v", response.StatusCode, body, readErr) + } +} + func TestServerTLSUsesH2ALPN(t *testing.T) { certificate, roots := testCertificate(t) server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { diff --git a/transport/native/HTTP2.md b/transport/native/HTTP2.md index 8f8c14b..713c838 100644 --- a/transport/native/HTTP2.md +++ b/transport/native/HTTP2.md @@ -2,7 +2,7 @@ 当前 `transport/native` 是经过 Linux 长连接和反向代理验收的 HTTP/1.1 实现。HTTP/2 暂不直接打开，原因是它把一个 TCP 连接变成多个独立请求流，连接级的 Keep-Alive、超时和并发限制不能直接复用现有 HTTP/1.1 状态机。 -当前已提供独立的 `transport/http2` 实验 Runner。它已经验证 TLS/h2、h2c、本地连接和请求闸门、GOAWAY 排空以及 RST_STREAM 取消；这些结果用于验证生命周期设计，不代表 native 已经启用 HTTP/2。 +当前已提供独立的 `transport/http2` 实验 Runner。它已经验证 TLS/h2、h2c、本地连接和请求闸门、GOAWAY 排空以及 RST_STREAM 取消；这些结果用于验证生命周期设计，不代表 native 已经启用 HTTP/2。实验 Runner 还覆盖了默认 8 MiB 请求体上限、128 路同连接 stream 多路复用，以及 2 MiB 慢读取 body 在窗口耗尽后的恢复。 ## 目标边界 @@ -17,7 +17,8 @@ 2. TLS 协商得到 `h2` 后进入 HTTP/2 connection state；其他 ALPN 直接关闭，不能把 HTTP/2 preface 交给 HTTP/1.1 解析器。 3. 每个 stream 创建独立的 `http.Request` 和取消 context。`RST_STREAM` 只取消当前请求，TCP 断开才取消该连接上的所有请求。 4. `MaxConcurrentRequests` 是所有连接共享的全局请求闸门；`MaxConcurrentStreams` 限制单连接并发流；两者都必须在收到完整请求体前生效。 -5. 服务关闭时停止 accept，发送 GOAWAY 拒绝新流，等待活动流排空；超过 `ShutdownTimeout` 后关闭连接并取消剩余请求。实验 Runner 已覆盖这条顺序。 +5. `MaxBodyBytes` 默认限制请求体为 8 MiB；已声明超限的请求在进入业务 handler 前返回 413，未声明长度的 body 通过受限 reader 读取，避免业务层无意间把整个请求体加载到内存。 +6. 服务关闭时停止 accept，发送 GOAWAY 拒绝新流，等待活动流排空；超过 `ShutdownTimeout` 后关闭连接并取消剩余请求。实验 Runner 已覆盖这条顺序。 ## 必须可观测的状态 @@ -26,8 +27,8 @@ HTTP/2 版本需要在 native `Stats` 之外至少记录：活动 stream、被 ## 进入实现前的验收矩阵 - TLS ALPN：只协商 `h2`，HTTP/1.1 客户端仍由现有 native 路径处理；不支持的 ALPN 在握手前后都不会误入错误解析器。 -- 多路复用：一个连接并发至少 128 个 stream，响应和 requestId 不串线，单个慢 stream 不阻塞其他 stream。 -- 流控：慢读取、慢写、窗口耗尽和大 body 都能恢复；内存不会随未消费 body 无限增长。 +- 多路复用：一个连接并发至少 128 个 stream，响应和 requestId 不串线，单个慢 stream 不阻塞其他 stream（实验 Runner 已覆盖 128 路同连接 stream）。 +- 流控：慢读取、慢写、窗口耗尽和大 body 都能恢复；内存不会随未消费 body 无限增长（实验 Runner 已覆盖 2 MiB 慢读取 body 和请求体上限）。 - 取消：RST_STREAM、客户端断开、服务关闭分别只影响应受影响的请求，并能在 metrics 中看到回收；当前实验 Runner 已覆盖 RST_STREAM 与服务关闭。 - 背压：连接上限、全局 stream 上限、单连接 stream 上限和 header/body 上限均能快速拒绝并自动恢复。 - 代理联调：Nginx/Cloudflare 的 TLS、Keep-Alive、超时、GOAWAY 和优雅发布行为与 HTTP/1.1 回源结果一致。