fix(chromium): bound CONNECT tunnels so a silent upstream cannot pin them forever

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

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")
+	}
+}