From cddaa0fa57e5182b7678c6cc133be27a02ee4337 Mon Sep 17 00:00:00 2001 From: Julien Neuhart Date: Mon, 7 Sep 2026 18:07:20 +0200 Subject: [PATCH] fix(chromium): ceiling the CONNECT tunnels in flight and bound tunnel writes --- pkg/modules/chromium/pinning_proxy.go | 64 +++++++++- pkg/modules/chromium/pinning_proxy_test.go | 131 +++++++++++++++++++++ 2 files changed, 192 insertions(+), 3 deletions(-) diff --git a/pkg/modules/chromium/pinning_proxy.go b/pkg/modules/chromium/pinning_proxy.go index ebc1e0b8..0f4a3cd1 100644 --- a/pkg/modules/chromium/pinning_proxy.go +++ b/pkg/modules/chromium/pinning_proxy.go @@ -64,6 +64,12 @@ type pinningProxy struct { // created it. Recreated on every Start. closing chan struct{} + // maxTunnels ceilings the CONNECT handlers in flight. Tests may lower it. + maxTunnels int64 + + // tunnels counts the CONNECT handlers in flight. + tunnels atomic.Int64 + logger *slog.Logger started bool mu sync.Mutex @@ -89,6 +95,7 @@ func newPinningProxy(allowList, denyList []*regexp2.Regexp, denyPrivateIPs, deny dialer := &net.Dialer{Timeout: 10 * time.Second} return dialer.DialContext(ctx, network, addr) }, + maxTunnels: maxConcurrentTunnels, } if enableEnvironmentProxy { @@ -198,6 +205,25 @@ func (p *pinningProxy) serveHTTP(w http.ResponseWriter, req *http.Request) { // Chromium then negotiates TLS end-to-end with the original hostname in // SNI. func (p *pinningProxy) handleConnect(w http.ResponseWriter, req *http.Request) { + // A ceiling, not a tuning knob: it bounds what a tunnel that refuses to end + // can accumulate, whatever keeps it alive. [spliceIdleTimeout] ends a silent + // tunnel, but a peer trickling a byte just under it stays "active" forever, + // and a compromised renderer can hold the client side open to match. + // + // Chromium caps itself well below this. Its socket pool manager allows 128 + // sockets per proxy chain for normal traffic plus 128 for WebSocket + // traffic, and every request Gotenberg's Chromium makes traverses this one + // proxy chain, so an honest browser cannot exceed 256 tunnels here. At + // double that, a real page never meets the ceiling and a hostile one stops + // at it. + if !p.acquireTunnel() { + p.logger.WarnContext(req.Context(), fmt.Sprintf("CONNECT to '%s' refused: %d tunnels already in flight", req.Host, p.maxTunnels)) + http.Error(w, "too many tunnels", http.StatusServiceUnavailable) + + return + } + defer p.releaseTunnel() + _, port, err := net.SplitHostPort(req.Host) if err != nil { http.Error(w, "bad CONNECT target", http.StatusBadRequest) @@ -288,6 +314,31 @@ func (p *pinningProxy) handleConnect(w http.ResponseWriter, req *http.Request) { spliceTunnel(client, upstream, closing, spliceIdleTimeout) } +// maxConcurrentTunnels is the default for [pinningProxy.maxTunnels]. See +// [pinningProxy.handleConnect] for how the value is derived. +const maxConcurrentTunnels = 512 + +// acquireTunnel reserves a slot for one CONNECT handler, reporting false when +// the proxy is already at [pinningProxy.maxTunnels]. The compare-and-swap loop +// keeps the check and the increment atomic, so concurrent handlers cannot +// overshoot the ceiling between them. +func (p *pinningProxy) acquireTunnel() bool { + for { + current := p.tunnels.Load() + if current >= p.maxTunnels { + return false + } + if p.tunnels.CompareAndSwap(current, current+1) { + return true + } + } +} + +// releaseTunnel returns a slot taken by [pinningProxy.acquireTunnel]. +func (p *pinningProxy) releaseTunnel() { + p.tunnels.Add(-1) +} + // spliceIdleTimeout bounds a CONNECT tunnel in which no byte has moved in // either direction. // @@ -321,14 +372,14 @@ func spliceTunnel(client, upstream net.Conn, closing <-chan struct{}, idleTimeou wg.Add(2) go func() { defer wg.Done() - copyTracking(upstream, client, &lastActivity) + copyTracking(upstream, client, &lastActivity, idleTimeout) if cw, ok := upstream.(interface{ CloseWrite() error }); ok { _ = cw.CloseWrite() } }() go func() { defer wg.Done() - copyTracking(client, upstream, &lastActivity) + copyTracking(client, upstream, &lastActivity, idleTimeout) if cw, ok := client.(interface{ CloseWrite() error }); ok { _ = cw.CloseWrite() } @@ -366,7 +417,7 @@ func spliceTunnel(client, upstream net.Conn, closing <-chan struct{}, idleTimeou // copyTracking copies src into dst, recording the time of every chunk that // moves so [spliceTunnel] can tell a busy tunnel from an idle one. -func copyTracking(dst, src net.Conn, lastActivity *atomic.Int64) { +func copyTracking(dst, src net.Conn, lastActivity *atomic.Int64, writeTimeout time.Duration) { buf := make([]byte, 32*1024) for { @@ -374,6 +425,13 @@ func copyTracking(dst, src net.Conn, lastActivity *atomic.Int64) { if n > 0 { lastActivity.Store(time.Now().UnixNano()) + // Bound the write. A destination that has gone away accepts the + // first chunk into its send buffer and only fails on the next one, + // so without a deadline this direction keeps a dead tunnel alive + // for one more chunk. A destination that stops reading altogether + // would block here forever. + _ = dst.SetWriteDeadline(time.Now().Add(writeTimeout)) + _, writeErr := dst.Write(buf[:n]) if writeErr != nil { return diff --git a/pkg/modules/chromium/pinning_proxy_test.go b/pkg/modules/chromium/pinning_proxy_test.go index 71ce32ce..7ebb17f7 100644 --- a/pkg/modules/chromium/pinning_proxy_test.go +++ b/pkg/modules/chromium/pinning_proxy_test.go @@ -1044,3 +1044,134 @@ func TestSpliceTunnel_ActiveTransferOutlivesIdleTimeout(t *testing.T) { t.Fatal("spliceTunnel did not return after the upstream closed") } } + +// TestPinningProxy_CONNECT_TunnelCeiling verifies the ceiling that bounds what +// tunnels refusing to end can accumulate. The idle bound cannot cover a peer +// that trickles just under it, so the count is what stops the growth. +func TestPinningProxy_CONNECT_TunnelCeiling(t *testing.T) { + // An upstream that accepts and then says nothing: the tunnel stays open. + upstreamAddr, stop := newRawTCPServer(t, func(c net.Conn) { + <-make(chan struct{}) + }) + t.Cleanup(stop) + + p := newPinningProxy(nil, nil, false, false, false) + p.maxTunnels = 1 + p.decide = func(_ context.Context, _ string, _, _ []*regexp2.Regexp, _ time.Time) (gotenberg.OutboundDecision, error) { + return gotenberg.OutboundDecision{Pinned: []netip.Addr{netip.MustParseAddr("127.0.0.1")}}, nil + } + p.dialPinned = func(_ context.Context, network string, _ []netip.Addr, _ string) (net.Conn, error) { + return net.Dial(network, upstreamAddr) + } + proxyURL := newProxyForTest(t, p) + proxyAddr := strings.TrimPrefix(proxyURL, "http://") + + connect := func(t *testing.T) *bufio.Reader { + t.Helper() + + conn, err := net.Dial("tcp", proxyAddr) + if err != nil { + t.Fatalf("dial proxy: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + + err = conn.SetDeadline(time.Now().Add(10 * time.Second)) + if err != nil { + t.Fatalf("set deadline: %v", err) + } + + _, err = fmt.Fprintf(conn, "CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n") + if err != nil { + t.Fatalf("write CONNECT: %v", err) + } + + return bufio.NewReader(conn) + } + + first := connect(t) + statusLine, err := first.ReadString('\n') + if err != nil { + t.Fatalf("read first status: %v", err) + } + if !strings.Contains(statusLine, " 200 ") { + t.Fatalf("first CONNECT status = %q, want 200", statusLine) + } + + // The first tunnel now holds the only slot. + second := connect(t) + resp, err := http.ReadResponse(second, nil) + if err != nil { + t.Fatalf("read second response: %v", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("second CONNECT status = %d, want %d", resp.StatusCode, http.StatusServiceUnavailable) + } + + if got := p.tunnels.Load(); got != 1 { + t.Errorf("tunnels in flight = %d, want 1: a refused CONNECT must not consume a slot", got) + } +} + +// TestPinningProxy_TunnelSlotIsReleased verifies a completed tunnel gives its +// slot back, so the ceiling bounds concurrency rather than lifetime totals. +func TestPinningProxy_TunnelSlotIsReleased(t *testing.T) { + upstreamAddr, stop := newRawTCPServer(t, func(c net.Conn) { + defer c.Close() + _, _ = c.Write([]byte("HI")) + }) + t.Cleanup(stop) + + p := newPinningProxy(nil, nil, false, false, false) + p.maxTunnels = 1 + p.decide = func(_ context.Context, _ string, _, _ []*regexp2.Regexp, _ time.Time) (gotenberg.OutboundDecision, error) { + return gotenberg.OutboundDecision{Pinned: []netip.Addr{netip.MustParseAddr("127.0.0.1")}}, nil + } + p.dialPinned = func(_ context.Context, network string, _ []netip.Addr, _ string) (net.Conn, error) { + return net.Dial(network, upstreamAddr) + } + proxyURL := newProxyForTest(t, p) + proxyAddr := strings.TrimPrefix(proxyURL, "http://") + + for attempt := range 3 { + conn, err := net.Dial("tcp", proxyAddr) + if err != nil { + t.Fatalf("attempt %d dial proxy: %v", attempt, err) + } + + err = conn.SetDeadline(time.Now().Add(10 * time.Second)) + if err != nil { + t.Fatalf("attempt %d set deadline: %v", attempt, err) + } + + _, err = fmt.Fprintf(conn, "CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n") + if err != nil { + t.Fatalf("attempt %d write CONNECT: %v", attempt, err) + } + + br := bufio.NewReader(conn) + statusLine, err := br.ReadString('\n') + if err != nil { + t.Fatalf("attempt %d read status: %v", attempt, err) + } + if !strings.Contains(statusLine, " 200 ") { + t.Fatalf("attempt %d CONNECT status = %q, want 200: the slot was not released", attempt, statusLine) + } + + // Drain until the upstream's close ends the tunnel, then release it. + _, _ = io.ReadAll(br) + _ = conn.Close() + + // The handler returns just after the splice ends. + for range 100 { + if p.tunnels.Load() == 0 { + break + } + time.Sleep(20 * time.Millisecond) + } + if got := p.tunnels.Load(); got != 0 { + t.Fatalf("attempt %d: tunnels in flight = %d, want 0", attempt, got) + } + } +}