From 1891a9ea689f71e6156662e6bf40941198df14bc Mon Sep 17 00:00:00 2001 From: Julien Neuhart Date: Mon, 7 Sep 2026 17:52:05 +0200 Subject: [PATCH] fix(chromium): bound CONNECT tunnels so a silent upstream cannot pin them forever --- pkg/modules/chromium/pinning_proxy.go | 119 +++++++++++++++-- pkg/modules/chromium/pinning_proxy_test.go | 146 +++++++++++++++++++++ 2 files changed, 257 insertions(+), 8 deletions(-) diff --git a/pkg/modules/chromium/pinning_proxy.go b/pkg/modules/chromium/pinning_proxy.go index d018e5bc..ebc1e0b8 100644 --- a/pkg/modules/chromium/pinning_proxy.go +++ b/pkg/modules/chromium/pinning_proxy.go @@ -11,6 +11,7 @@ import ( "net/netip" "net/url" "sync" + "sync/atomic" "time" "github.com/dlclark/regexp2" @@ -57,6 +58,12 @@ type pinningProxy struct { server *http.Server wg sync.WaitGroup + // closing is closed by Stop to force in-flight CONNECT tunnels shut. + // [http.Server.Shutdown] cannot do it: net/http untracks a connection once + // a handler hijacks it, so a tunnel would otherwise outlive the proxy that + // created it. Recreated on every Start. + closing chan struct{} + logger *slog.Logger started bool mu sync.Mutex @@ -110,6 +117,7 @@ func (p *pinningProxy) Start(logger *slog.Logger) error { } p.listener = l + p.closing = make(chan struct{}) p.logger = logger.With(slog.String("logger", "pinning-proxy")) p.server = &http.Server{ Handler: http.HandlerFunc(p.serveHTTP), @@ -140,9 +148,18 @@ func (p *pinningProxy) Stop(logger *slog.Logger) error { return nil } srv := p.server + closing := p.closing + p.closing = nil p.started = false p.mu.Unlock() + // Force in-flight tunnels shut before draining the server. Shutdown does + // not reach them, so a tunnel whose upstream never answers would otherwise + // survive the proxy, and with it every Chromium restart. + if closing != nil { + close(closing) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -264,24 +281,110 @@ func (p *pinningProxy) handleConnect(w http.ResponseWriter, req *http.Request) { return } - // Splice bytes in both directions until either side closes. - var splice sync.WaitGroup - splice.Add(2) + p.mu.Lock() + closing := p.closing + p.mu.Unlock() + + spliceTunnel(client, upstream, closing, spliceIdleTimeout) +} + +// spliceIdleTimeout bounds a CONNECT tunnel in which no byte has moved in +// either direction. +// +// Nothing else bounds one. The hijacked connections carry no deadline: the +// server clears the header read deadline once the request line is in, and +// net.Dialer.Timeout only covers the connect. net/http also untracks a +// connection once it is hijacked, so neither Server.Shutdown nor a Chromium +// restart reaps it. Left alone, an upstream that accepts the tunnel and then +// answers nothing holds two goroutines and two sockets until the process dies. +// +// Sized well above any legitimate pause between a request and its response, so +// a slow origin is never cut off. A transfer that keeps making progress +// refreshes the deadline and runs for as long as it needs. +const spliceIdleTimeout = 2 * time.Minute + +// spliceTunnel copies bytes between the two ends of a CONNECT tunnel until +// both directions finish, the tunnel sits idle for idleTimeout, or closing is +// closed because the proxy is shutting down. Callers pass +// [spliceIdleTimeout]; only tests shorten it. +// +// Each direction half-closes its destination once its source reaches EOF, so a +// peer that waits for the request to end before answering still sees the EOF. +// Idleness is tracked across both directions rather than per direction: the +// client sends nothing for the length of a download, and half-closing its write +// side then would tell the origin the client had gone away. +func spliceTunnel(client, upstream net.Conn, closing <-chan struct{}, idleTimeout time.Duration) { + var lastActivity atomic.Int64 + lastActivity.Store(time.Now().UnixNano()) + + var wg sync.WaitGroup + wg.Add(2) go func() { - defer splice.Done() - _, _ = io.Copy(upstream, client) + defer wg.Done() + copyTracking(upstream, client, &lastActivity) if cw, ok := upstream.(interface{ CloseWrite() error }); ok { _ = cw.CloseWrite() } }() go func() { - defer splice.Done() - _, _ = io.Copy(client, upstream) + defer wg.Done() + copyTracking(client, upstream, &lastActivity) if cw, ok := client.(interface{ CloseWrite() error }); ok { _ = cw.CloseWrite() } }() - splice.Wait() + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + ticker := time.NewTicker(idleTimeout / 4) + defer ticker.Stop() + + for { + select { + case <-done: + return + case <-closing: + case <-ticker.C: + if time.Since(time.Unix(0, lastActivity.Load())) < idleTimeout { + continue + } + } + + // Closing both ends unblocks whichever copy is still reading. The + // caller's own deferred Close calls then become no-ops. + _ = client.Close() + _ = upstream.Close() + <-done + + return + } +} + +// 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) { + buf := make([]byte, 32*1024) + + for { + n, readErr := src.Read(buf) + if n > 0 { + lastActivity.Store(time.Now().UnixNano()) + + _, writeErr := dst.Write(buf[:n]) + if writeErr != nil { + return + } + + lastActivity.Store(time.Now().UnixNano()) + } + if readErr != nil { + return + } + } } // handleForward handles plain HTTP requests sent to the proxy as absolute diff --git a/pkg/modules/chromium/pinning_proxy_test.go b/pkg/modules/chromium/pinning_proxy_test.go index 337945d0..71ce32ce 100644 --- a/pkg/modules/chromium/pinning_proxy_test.go +++ b/pkg/modules/chromium/pinning_proxy_test.go @@ -898,3 +898,149 @@ func TestPinningProxy_StopIdempotent(t *testing.T) { t.Fatalf("second Stop on stopped proxy: %v", err) } } + +// tcpPair returns the two ends of a connected loopback TCP connection. Both +// ends are closed when the test finishes. +func tcpPair(t *testing.T) (net.Conn, net.Conn) { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer func() { _ = listener.Close() }() + + type accepted struct { + conn net.Conn + err error + } + + acceptChan := make(chan accepted, 1) + go func() { + conn, acceptErr := listener.Accept() + acceptChan <- accepted{conn: conn, err: acceptErr} + }() + + dialed, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + + res := <-acceptChan + if res.err != nil { + t.Fatalf("accept: %v", res.err) + } + + t.Cleanup(func() { + _ = dialed.Close() + _ = res.conn.Close() + }) + + return dialed, res.conn +} + +// TestSpliceTunnel_IdleTunnelIsClosed covers the leak where an upstream that +// accepted a CONNECT tunnel and then never spoke pinned both splice goroutines +// and both sockets for the lifetime of the process. A hijacked connection +// carries no deadline and net/http stops tracking it, so the idle bound in +// spliceTunnel is the only thing that ends such a tunnel. +func TestSpliceTunnel_IdleTunnelIsClosed(t *testing.T) { + // The peers are kept open by the pair's cleanup: the tunnel is silent, not + // finished. + client, _ := tcpPair(t) + upstream, _ := tcpPair(t) + + done := make(chan struct{}) + go func() { + spliceTunnel(client, upstream, nil, 100*time.Millisecond) + close(done) + }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("spliceTunnel did not return on an idle tunnel") + } +} + +// TestSpliceTunnel_ClosingShutsTunnelDown verifies that stopping the proxy +// reaps in-flight tunnels. http.Server.Shutdown cannot: it stops tracking a +// connection once a handler hijacks it, so without this signal a tunnel would +// outlive the proxy and every Chromium restart after it. +func TestSpliceTunnel_ClosingShutsTunnelDown(t *testing.T) { + client, _ := tcpPair(t) + upstream, _ := tcpPair(t) + + closing := make(chan struct{}) + done := make(chan struct{}) + + go func() { + // An idle timeout far beyond the test: only closing can end this. + spliceTunnel(client, upstream, closing, time.Hour) + close(done) + }() + + close(closing) + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("spliceTunnel did not return when the proxy shut down") + } +} + +// TestSpliceTunnel_ActiveTransferOutlivesIdleTimeout guards the idle bound +// against cutting a healthy transfer. Idleness is tracked across both +// directions, so a download that keeps making progress must survive well past +// the timeout even though the client sends nothing throughout. +func TestSpliceTunnel_ActiveTransferOutlivesIdleTimeout(t *testing.T) { + const ( + idleTimeout = 100 * time.Millisecond + chunks = 10 + interval = 30 * time.Millisecond + ) + + client, clientPeer := tcpPair(t) + upstream, upstreamPeer := tcpPair(t) + + done := make(chan struct{}) + go func() { + spliceTunnel(client, upstream, nil, idleTimeout) + close(done) + }() + + // Trickle a response for well over the idle timeout, then finish. + go func() { + for range chunks { + _, _ = upstreamPeer.Write([]byte("x")) + time.Sleep(interval) + } + _ = upstreamPeer.Close() + }() + + received := 0 + buf := make([]byte, chunks) + + for received < chunks { + err := clientPeer.SetReadDeadline(time.Now().Add(10 * time.Second)) + if err != nil { + t.Fatalf("set read deadline: %v", err) + } + + n, readErr := clientPeer.Read(buf) + received += n + if readErr != nil { + break + } + } + + if received != chunks { + t.Fatalf("received %d bytes, want %d: the tunnel was cut while still transferring", received, chunks) + } + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("spliceTunnel did not return after the upstream closed") + } +}