fix(ingress): support WebSocket over HTTP/2 (replace gorilla/websocket)

components/ingress/go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/alibaba/OpenSandbox/sandbox-k8s v0.0.0
 	github.com/alibaba/opensandbox/internal v0.0.0
 	github.com/alicebob/miniredis/v2 v2.37.0
-	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674
+	github.com/coder/websocket v1.8.15
 	github.com/redis/go-redis/v9 v9.18.0
 	github.com/stretchr/testify v1.11.1
 	k8s.io/apimachinery v0.34.3

components/ingress/go.sum

@@ -8,6 +8,8 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA=
+github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
@@ -46,8 +48,6 @@ github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db h1:097atOisP2aRj7vFgY
 github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5THxAzdVpqr6/geYxZytqFMBCOtn/ujyeo=
-github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674/go.mod h1:r4w70xmWCQKmi1ONH4KIaBptdivuRPyosB9RmPlGEwA=
 github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c=
 github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=

components/ingress/main.go

@@ -92,7 +92,15 @@ func main() {
 	http.Handle("/", reverseProxy)
 	http.HandleFunc("/status.ok", proxy.Healthz)
 
-	if err := http.ListenAndServe(fmt.Sprintf(":%v", flag.Port), nil); err != nil {
+	protos := new(http.Protocols)
+	protos.SetHTTP1(true)
+	protos.SetUnencryptedHTTP2(true)
+	srv := &http.Server{
+		Addr:      fmt.Sprintf(":%v", flag.Port),
+		Protocols: protos,
+	}
+
+	if err := srv.ListenAndServe(); err != nil {
 		log.Panicf("Error starting http server: %v", err)
 	}
 

components/ingress/pkg/proxy/proxy.go

@@ -128,16 +128,31 @@ func (p *Proxy) serve(w http.ResponseWriter, r *http.Request) {
 }
 
 func (p *Proxy) isWebSocketRequest(r *http.Request) bool {
-	if r.Method != http.MethodGet {
-		return false
+	// HTTP/2 Extended CONNECT (RFC 8441)
+	if r.Method == http.MethodConnect && r.ProtoMajor >= 2 &&
+		strings.EqualFold(r.Header.Get(":protocol"), "websocket") {
+		return true
 	}
-	if r.Header.Get("Upgrade") != "websocket" {
+	// HTTP/1.1 Upgrade
+	if r.Method != http.MethodGet {
 		return false
 	}
-	if r.Header.Get("Connection") != "Upgrade" {
-		return false
+	return headerContainsToken(r.Header, "Connection", "Upgrade") &&
+		headerContainsToken(r.Header, "Upgrade", "websocket")
+}
+
+// headerContainsToken checks whether a comma-separated HTTP header contains
+// the given token (case-insensitive). This handles L7 proxies that send
+// multi-value headers like "Connection: keep-alive, Upgrade".
+func headerContainsToken(h http.Header, key, token string) bool {
+	for _, v := range h[http.CanonicalHeaderKey(key)] {
+		for _, s := range strings.Split(v, ",") {
+			if strings.EqualFold(strings.TrimSpace(s), token) {
+				return true
+			}
+		}
 	}
-	return true
+	return false
 }
 
 func (p *Proxy) resolveRealHost(host *sandboxHost) (string, error, int) {

components/ingress/pkg/proxy/proxy_test.go

@@ -54,6 +54,31 @@ func TestIsWebSocketRequest(t *testing.T) {
 	req.Header.Set("Upgrade", "websocket")
 	req.Header.Set("Connection", "Upgrade")
 	assert.False(t, proxy.isWebSocketRequest(req))
+
+	// Connection header with multiple tokens (L7 proxy scenario)
+	req = httptest.NewRequest(http.MethodGet, "/ws", nil)
+	req.Header.Set("Upgrade", "websocket")
+	req.Header.Set("Connection", "keep-alive, Upgrade")
+	assert.True(t, proxy.isWebSocketRequest(req))
+
+	// Case-insensitive headers
+	req = httptest.NewRequest(http.MethodGet, "/ws", nil)
+	req.Header.Set("Upgrade", "WebSocket")
+	req.Header.Set("Connection", "upgrade")
+	assert.True(t, proxy.isWebSocketRequest(req))
+
+	// HTTP/2 Extended CONNECT (RFC 8441)
+	req = httptest.NewRequest(http.MethodConnect, "/ws", nil)
+	req.ProtoMajor = 2
+	req.ProtoMinor = 0
+	req.Header.Set(":protocol", "websocket")
+	assert.True(t, proxy.isWebSocketRequest(req))
+
+	// HTTP/2 CONNECT without :protocol is not WebSocket
+	req = httptest.NewRequest(http.MethodConnect, "/ws", nil)
+	req.ProtoMajor = 2
+	req.ProtoMinor = 0
+	assert.False(t, proxy.isWebSocketRequest(req))
 }
 
 func TestParseHostRoute(t *testing.T) {

components/ingress/pkg/proxy/websocket.go

@@ -15,31 +15,16 @@
 package proxy
 
 import (
-	"fmt"
+	"context"
+	"errors"
 	"io"
 	"net"
 	"net/http"
 	"net/url"
 	"strings"
 
 	slogger "github.com/alibaba/opensandbox/internal/logger"
-	"github.com/gorilla/websocket"
-)
-
-var (
-	// defaultWebSocketDialer is a dialer with all fields set to the default zero values.
-	defaultWebSocketDialer = websocket.DefaultDialer
-
-	// defaultUpgrader specifies the parameters for upgrading an HTTP
-	// connection to a WebSocket connection.
-	defaultUpgrader = &websocket.Upgrader{
-		ReadBufferSize:  1024,
-		WriteBufferSize: 1024,
-		// Allow any Origin: ingress sits behind trusted gateways where Host/Origin
-		// often diverge (e.g. browser UI vs internal target). gorilla's default
-		// same-origin check rejects those upgrades.
-		CheckOrigin: func(_ *http.Request) bool { return true },
-	}
+	"github.com/coder/websocket"
 )
 
 // WebSocketProxy is an HTTP Handler that takes an incoming WebSocket
@@ -54,14 +39,6 @@ type WebSocketProxy struct {
 	// the incoming WebSocket connection. Request is the initial incoming and
 	// unmodified request.
 	backend func(*http.Request) *url.URL
-
-	//  dialer contains options for connecting to the backend WebSocket server.
-	//  If nil, DefaultDialer is used.
-	dialer *websocket.Dialer
-
-	// upgrader specifies the parameters for upgrading a incoming HTTP
-	// connection to a WebSocket connection. If nil, DefaultUpgrader is used.
-	upgrader *websocket.Upgrader
 }
 
 // ProxyHandler returns a new http.Handler interface that reverse proxies the
@@ -95,11 +72,6 @@ func (w *WebSocketProxy) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	dialer := w.dialer
-	if w.dialer == nil {
-		dialer = defaultWebSocketDialer
-	}
-
 	// Forward all incoming headers to the backend except hop-by-hop headers
 	// (RFC 7230 §6.1) and WebSocket handshake headers managed by the dialer.
 	// Per RFC 7230, also strip any header named by Connection tokens.
@@ -157,88 +129,131 @@ func (w *WebSocketProxy) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 		w.director(r, requestHeader)
 	}
 
-	// Connect to the backend URL, also pass the headers we get from the requst
-	// together with the Forwarded headers we prepared above.
-	connBackend, resp, err := dialer.Dial(backendURL.String(), requestHeader)
+	// HTTP/2 Extended CONNECT (RFC 8441) — raw bidirectional tunnel.
+	if r.ProtoMajor >= 2 && r.Method == http.MethodConnect {
+		w.serveH2Tunnel(rw, r, backendURL, requestHeader)
+		return
+	}
+
+	w.serveH1(rw, r, backendURL, requestHeader)
+}
+
+// serveH1 handles the traditional HTTP/1.1 WebSocket upgrade path.
+func (w *WebSocketProxy) serveH1(rw http.ResponseWriter, r *http.Request, backendURL *url.URL, requestHeader http.Header) {
+	ctx := r.Context()
+
+	// Dial the backend first so we can relay errors before upgrading the client.
+	connBackend, resp, err := websocket.Dial(ctx, backendURL.String(), &websocket.DialOptions{
+		HTTPHeader: requestHeader,
+	})
 	if err != nil {
 		Logger.With(slogger.Field{Key: "error", Value: err}).Errorf("WebSocketProxy: couldn't dial to remote backend")
 		if resp != nil {
-			// If the WebSocket handshake fails, ErrBadHandshake is returned
-			// along with a non-nil *http.Response so that callers can handle
-			// redirects, authentication, etcetera.
-			if err := copyResponse(rw, resp); err != nil {
-				Logger.With(slogger.Field{Key: "error", Value: err}).Errorf("WebSocketProxy: couldn't write response after failed remote backend handshake")
+			if copyErr := copyResponse(rw, resp); copyErr != nil {
+				Logger.With(slogger.Field{Key: "error", Value: copyErr}).Errorf("WebSocketProxy: couldn't write response after failed remote backend handshake")
 			}
 		} else {
 			http.Error(rw, http.StatusText(http.StatusServiceUnavailable), http.StatusServiceUnavailable)
 		}
 		return
 	}
-	defer connBackend.Close()
-
-	upgrader := w.upgrader
-	if w.upgrader == nil {
-		upgrader = defaultUpgrader
-	}
+	defer connBackend.CloseNow()
 
-	// Only pass those headers to the upgrader.
-	upgradeHeader := http.Header{}
-	if hdr := resp.Header.Get(SecWebSocketProtocol); hdr != "" {
-		upgradeHeader.Set(SecWebSocketProtocol, hdr)
-	}
-	if hdr := resp.Header.Get(SetCookie); hdr != "" {
-		upgradeHeader.Set(SetCookie, hdr)
-	}
-
-	// Now upgrade the existing incoming request to a WebSocket connection.
-	// Also pass the header that we gathered from the Dial handshake.
-	connPub, err := upgrader.Upgrade(rw, r, upgradeHeader)
+	// Accept the client-side WebSocket upgrade.
+	connPub, err := websocket.Accept(rw, r, &websocket.AcceptOptions{
+		InsecureSkipVerify: true,
+		Subprotocols:       subprotocolsFromResponse(resp),
+	})
 	if err != nil {
 		Logger.With(slogger.Field{Key: "error", Value: err}).Errorf("WebSocketProxy: couldn't upgrade websocket connection")
 		return
 	}
-	defer connPub.Close()
+	defer connPub.CloseNow()
 
+	// Bidirectional relay.
 	errClient := make(chan error, 1)
 	errBackend := make(chan error, 1)
-	replicateWebsocketConn := func(dst, src *websocket.Conn, errc chan error) {
-		for {
-			msgType, msg, err := src.ReadMessage()
-			if err != nil {
-				m := websocket.FormatCloseMessage(websocket.CloseNormalClosure, fmt.Sprintf("%v", err))
-				if e, ok := err.(*websocket.CloseError); ok { //nolint:errorlint
-					if e.Code != websocket.CloseNoStatusReceived {
-						m = websocket.FormatCloseMessage(e.Code, e.Text)
-					}
-				}
-				errc <- err
-				_ = dst.WriteMessage(websocket.CloseMessage, m)
-				break
-			}
-			err = dst.WriteMessage(msgType, msg)
-			if err != nil {
-				errc <- err
-				break
-			}
-		}
-	}
 
-	go replicateWebsocketConn(connPub, connBackend, errClient)
-	go replicateWebsocketConn(connBackend, connPub, errBackend)
+	go replicateConn(ctx, connPub, connBackend, errClient)
+	go replicateConn(ctx, connBackend, connPub, errBackend)
 
 	var message string
 	select {
 	case err = <-errClient:
 		message = "WebSocketProxy: Error when copying from backend to client: %v"
 	case err = <-errBackend:
 		message = "WebSocketProxy: Error when copying from client to backend: %v"
-
 	}
-	if e, ok := err.(*websocket.CloseError); !ok || e.Code == websocket.CloseAbnormalClosure { //nolint:errorlint
+
+	var closeErr websocket.CloseError
+	if !errors.As(err, &closeErr) || closeErr.Code == websocket.StatusAbnormalClosure {
 		Logger.With(slogger.Field{Key: "error", Value: err}).Errorf(message, err)
 	}
 }
 
+// serveH2Tunnel handles HTTP/2 Extended CONNECT (RFC 8441) by creating
+// a raw bidirectional tunnel between the h2 stream and a backend h1 WebSocket.
+func (w *WebSocketProxy) serveH2Tunnel(rw http.ResponseWriter, r *http.Request, backendURL *url.URL, requestHeader http.Header) {
+	ctx := r.Context()
+
+	connBackend, _, err := websocket.Dial(ctx, backendURL.String(), &websocket.DialOptions{
+		HTTPHeader: requestHeader,
+	})
+	if err != nil {
+		Logger.With(slogger.Field{Key: "error", Value: err}).Errorf("WebSocketProxy: couldn't dial to remote backend (h2 tunnel)")
+		http.Error(rw, http.StatusText(http.StatusServiceUnavailable), http.StatusServiceUnavailable)
+		return
+	}
+	backendNetConn := websocket.NetConn(ctx, connBackend, websocket.MessageBinary)
+	defer backendNetConn.Close()
+
+	rc := http.NewResponseController(rw)
+	if err := rc.EnableFullDuplex(); err != nil {
+		Logger.With(slogger.Field{Key: "error", Value: err}).Errorf("WebSocketProxy: EnableFullDuplex failed")
+		http.Error(rw, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
+		return
+	}
+	rw.WriteHeader(http.StatusOK)
+	if err := rc.Flush(); err != nil {
+		Logger.With(slogger.Field{Key: "error", Value: err}).Errorf("WebSocketProxy: flush failed")
+		return
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		io.Copy(backendNetConn, r.Body)
+	}()
+	io.Copy(rw, backendNetConn)
+	<-done
+}
+
+func replicateConn(ctx context.Context, dst, src *websocket.Conn, errc chan error) {
+	for {
+		msgType, msg, err := src.Read(ctx)
+		if err != nil {
+			errc <- err
+			dst.Close(websocket.StatusNormalClosure, "")
+			break
+		}
+		err = dst.Write(ctx, msgType, msg)
+		if err != nil {
+			errc <- err
+			break
+		}
+	}
+}
+
+func subprotocolsFromResponse(resp *http.Response) []string {
+	if resp == nil {
+		return nil
+	}
+	if proto := resp.Header.Get(SecWebSocketProtocol); proto != "" {
+		return []string{proto}
+	}
+	return nil
+}
+
 func copyResponse(rw http.ResponseWriter, resp *http.Response) error {
 	copyHeader(rw.Header(), resp.Header)
 	rw.WriteHeader(resp.StatusCode)