merge: PR #76 — MASQUE CONNECT-UDP and CONNECT-TCP support (RFC 9298/9114)

Extend MASQUE implementation to support TCP tunneling via standard HTTP/3 CONNECT method, in addition to UDP via CONNECT-UDP.
2026-05-21 22:50:23 +08:00
parent 8435716166 c5f91232cb
commit 66b53580af
3 changed files with 382 additions and 25 deletions
@@ -26,9 +26,10 @@ func init() {
 }

 var (
-	ErrInvalidConnection = errors.New("masque: invalid connection type, expected MasqueConn")
-	ErrConnectFailed     = errors.New("masque: CONNECT-UDP failed")
-	ErrCapsuleRequired   = errors.New("masque: server did not confirm capsule-protocol")
+	ErrInvalidConnection  = errors.New("masque: invalid connection type, expected MasqueConn")
+	ErrConnectFailed      = errors.New("masque: CONNECT failed")
+	ErrCapsuleRequired    = errors.New("masque: server did not confirm capsule-protocol")
+	ErrUnsupportedNetwork = errors.New("masque: unsupported network type")
 )

 type masqueConnector struct {
@@ -60,12 +61,109 @@ func (c *masqueConnector) Connect(ctx context.Context, conn net.Conn, network, a
 		"sid":     string(ctxvalue.SidFromContext(ctx)),
 	})

-	// Only support UDP
-	if !strings.HasPrefix(network, "udp") {
-		return nil, fmt.Errorf("masque connector only supports UDP, got %s", network)
+	// Dispatch based on network type
+	if strings.HasPrefix(network, "tcp") {
+		return c.connectTCP(ctx, conn, address, log)
+	}
+	if strings.HasPrefix(network, "udp") {
+		return c.connectUDP(ctx, conn, address, log)
 	}

-	log.Debugf("connect-udp %s/%s", address, network)
+	log.Errorf("masque: unsupported network: %s", network)
+	return nil, fmt.Errorf("%w: %s", ErrUnsupportedNetwork, network)
+}
+
+func (c *masqueConnector) connectTCP(ctx context.Context, conn net.Conn, address string, log logger.Logger) (net.Conn, error) {
+	log.Debugf("connect-tcp %s", address)
+
+	// Get the MasqueConn from the dialer
+	masqueConn, ok := conn.(*masque_dialer.MasqueConn)
+	if !ok {
+		log.Error(ErrInvalidConnection)
+		return nil, ErrInvalidConnection
+	}
+
+	// Get pre-opened stream from dialer
+	reqStream := masqueConn.GetRequestStream()
+	proxyHost := masqueConn.GetHost()
+
+	// Apply connect timeout to the actual stream
+	if c.md.connectTimeout > 0 {
+		reqStream.SetDeadline(time.Now().Add(c.md.connectTimeout))
+		defer reqStream.SetDeadline(time.Time{})
+	}
+
+	// Ensure stream is closed on any error
+	success := false
+	defer func() {
+		if !success {
+			reqStream.Close()
+		}
+	}()
+
+	// Create standard HTTP/3 CONNECT request (RFC 9114)
+	// No :protocol pseudo-header for standard CONNECT
+	req := &http.Request{
+		Method: http.MethodConnect,
+		URL:    &url.URL{},
+		Host:   address, // Target address goes in :authority
+		Header: http.Header{},
+		Proto:  "HTTP/3.0",
+	}
+
+	// Add proxy authentication if configured
+	if c.options.Auth != nil {
+		u := c.options.Auth.Username()
+		p, _ := c.options.Auth.Password()
+		auth := u + ":" + p
+		encoded := base64.StdEncoding.EncodeToString([]byte(auth))
+		req.Header.Set("Proxy-Authorization", "Basic "+encoded)
+	}
+
+	if log.IsLevelEnabled(logger.TraceLevel) {
+		log.Tracef("CONNECT request: %s Host=%s ProxyHost=%s", req.Method, address, proxyHost)
+	}
+	log.Debugf("sending CONNECT request for %s", address)
+
+	// Send request headers
+	if err := reqStream.SendRequestHeader(req); err != nil {
+		log.Error("masque: failed to send request:", err)
+		return nil, err
+	}
+
+	// Read response
+	resp, err := reqStream.ReadResponse()
+	if err != nil {
+		log.Error("masque: failed to read response:", err)
+		return nil, err
+	}
+
+	if log.IsLevelEnabled(logger.TraceLevel) {
+		log.Tracef("CONNECT response: %d %s", resp.StatusCode, resp.Status)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		log.Errorf("masque: proxy returned status %d", resp.StatusCode)
+		return nil, fmt.Errorf("%w: status %d", ErrConnectFailed, resp.StatusCode)
+	}
+
+	log.Debugf("CONNECT established to %s", address)
+
+	// Resolve target address
+	raddr, err := net.ResolveTCPAddr("tcp", address)
+	if err != nil {
+		return nil, err
+	}
+
+	// Create stream connection for TCP data transfer
+	streamConn := masque_util.NewStreamConnFromRequestStream(reqStream, conn.LocalAddr(), raddr)
+
+	success = true // Prevent defer from closing stream - streamConn now owns it
+	return streamConn, nil
+}
+
+func (c *masqueConnector) connectUDP(ctx context.Context, conn net.Conn, address string, log logger.Logger) (net.Conn, error) {
+	log.Debugf("connect-udp %s", address)

 	// Get the MasqueConn from the dialer
 	masqueConn, ok := conn.(*masque_dialer.MasqueConn)
@@ -1,6 +1,7 @@
 package masque

 import (
+	"bytes"
 	"context"
 	"encoding/base64"
 	"errors"
@@ -23,6 +24,7 @@ import (
 	xbypass "github.com/go-gost/x/bypass"
 	xctx "github.com/go-gost/x/ctx"
 	ictx "github.com/go-gost/x/internal/ctx"
+	xnet "github.com/go-gost/x/internal/net"
 	"github.com/go-gost/x/internal/net/udp"
 	masque_util "github.com/go-gost/x/internal/util/masque"
 	stats_util "github.com/go-gost/x/internal/util/stats"
@@ -169,7 +171,29 @@ func (h *masqueHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
 		return ErrBadRequest
 	}

-	return h.handleConnectUDP(ctx, w, r, conn.LocalAddr(), ro, log)
+	// Validate CONNECT method
+	if r.Method != http.MethodConnect {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		log.Error(ErrMethodNotAllowed)
+		return ErrMethodNotAllowed
+	}
+
+	// Dispatch based on :protocol pseudo-header
+	// In quic-go, the :protocol pseudo-header is stored in r.Proto
+	switch r.Proto {
+	case "connect-udp":
+		// Extended CONNECT for UDP (RFC 9298)
+		return h.handleConnectUDP(ctx, w, r, conn.LocalAddr(), ro, log)
+	case "HTTP/3.0", "":
+		// Standard CONNECT for TCP (RFC 9114)
+		// r.Proto is "HTTP/3.0" for standard HTTP/3 CONNECT (no :protocol header)
+		ro.Network = "tcp"
+		return h.handleConnectTCP(ctx, w, r, conn.LocalAddr(), ro, log)
+	default:
+		w.WriteHeader(http.StatusBadRequest)
+		log.Errorf("masque: unsupported protocol: %s", r.Proto)
+		return ErrBadRequest
+	}
 }

 func (h *masqueHandler) Close() error {
@@ -180,22 +204,7 @@ func (h *masqueHandler) Close() error {
 }

 func (h *masqueHandler) handleConnectUDP(ctx context.Context, w http.ResponseWriter, r *http.Request, laddr net.Addr, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
-	// Validate Extended CONNECT request
-	if r.Method != http.MethodConnect {
-		w.WriteHeader(http.StatusMethodNotAllowed)
-		log.Error(ErrMethodNotAllowed)
-		return ErrMethodNotAllowed
-	}
-
-	// Check for :protocol pseudo-header (Extended CONNECT)
-	// In quic-go, the :protocol pseudo-header is stored in r.Proto
-	if r.Proto != "connect-udp" {
-		w.WriteHeader(http.StatusBadRequest)
-		log.Error("masque: expected :protocol=connect-udp, got: ", r.Proto)
-		return ErrBadRequest
-	}
-
-	// Validate capsule-protocol header
+	// Validate capsule-protocol header (required for CONNECT-UDP per RFC 9298)
 	if r.Header.Get("Capsule-Protocol") != "?1" {
 		w.WriteHeader(http.StatusBadRequest)
 		log.Error(ErrCapsuleRequired)
@@ -305,15 +314,20 @@ func (h *masqueHandler) handleConnectUDP(ctx context.Context, w http.ResponseWri
 		ctx = xctx.ContextWithHash(ctx, &xctx.Hash{Source: targetAddr})
 	}

+	var buf bytes.Buffer
+
 	if h.options.Router != nil {
 		// Use router to dial through chain (for forwarding through upstream proxy)
-		c, err := h.options.Router.Dial(ctx, "udp", targetAddr)
+		c, err := h.options.Router.Dial(ictx.ContextWithBuffer(ctx, &buf), "udp", targetAddr)
+		ro.Route = buf.String()
 		if err != nil {
 			log.Error("masque: failed to dial through router: ", err)
 			return err
 		}
 		defer c.Close()

+		ro.SrcAddr = c.LocalAddr().String()
+
 		// The connection from router should be a PacketConn (e.g., from masque connector)
 		if pc, ok := c.(net.PacketConn); ok {
 			targetPC = pc
@@ -332,6 +346,8 @@ func (h *masqueHandler) handleConnectUDP(ctx context.Context, w http.ResponseWri
 		}
 		defer directConn.Close()

+		ro.SrcAddr = directConn.LocalAddr().String()
+
 		// Wrap with fixed target address
 		targetPC = &fixedTargetPacketConn{
 			PacketConn: directConn,
@@ -352,6 +368,147 @@ func (h *masqueHandler) handleConnectUDP(ctx context.Context, w http.ResponseWri
 	return relay.Run(ctx)
 }

+func (h *masqueHandler) handleConnectTCP(ctx context.Context, w http.ResponseWriter, r *http.Request, laddr net.Addr, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
+	// Extract user for logging
+	if u, _, _ := h.basicProxyAuth(r.Header.Get("Proxy-Authorization")); u != "" {
+		log = log.WithFields(map[string]any{"user": u})
+		ro.ClientID = u
+	}
+
+	// Authenticate
+	clientID, ok := h.authenticate(ctx, w, r, log)
+	if !ok {
+		return errors.New("authentication failed")
+	}
+
+	if clientID != "" {
+		log = log.WithFields(map[string]any{"clientID": clientID})
+		ro.ClientID = clientID
+	}
+	ctx = xctx.ContextWithClientID(ctx, xctx.ClientID(clientID))
+
+	// Target address comes from :authority (r.Host) for standard CONNECT
+	targetAddr := r.Host
+	if targetAddr == "" {
+		w.WriteHeader(http.StatusBadRequest)
+		log.Error("masque: missing target address in :authority")
+		return ErrBadRequest
+	}
+
+	// Ensure port is present
+	if _, port, _ := net.SplitHostPort(targetAddr); port == "" {
+		targetAddr = net.JoinHostPort(strings.Trim(targetAddr, "[]"), "443")
+	}
+
+	ro.Host = targetAddr
+	log = log.WithFields(map[string]any{
+		"target": targetAddr,
+	})
+	log.Debug("connect-tcp request to ", targetAddr)
+
+	// Check bypass
+	if h.options.Bypass != nil &&
+		h.options.Bypass.Contains(ctx, "tcp", targetAddr, bypass.WithService(h.options.Service)) {
+		w.WriteHeader(http.StatusForbidden)
+		log.Debug("bypass: ", targetAddr)
+		return xbypass.ErrBypass
+	}
+
+	// Get HTTP/3 stream
+	streamer, ok := w.(http3.HTTPStreamer)
+	if !ok {
+		w.WriteHeader(http.StatusInternalServerError)
+		log.Error("masque: failed to get HTTP/3 streamer")
+		return errors.New("masque: HTTP/3 streamer not available")
+	}
+
+	// Dial target connection
+	switch h.md.hash {
+	case "host":
+		ctx = xctx.ContextWithHash(ctx, &xctx.Hash{Source: targetAddr})
+	}
+
+	var cc net.Conn
+	var err error
+	var buf bytes.Buffer
+
+	if h.options.Router != nil {
+		cc, err = h.options.Router.Dial(ictx.ContextWithBuffer(ctx, &buf), "tcp", targetAddr)
+	} else {
+		cc, err = net.Dial("tcp", targetAddr)
+	}
+	ro.Route = buf.String()
+
+	if err != nil {
+		w.WriteHeader(http.StatusServiceUnavailable)
+		log.Error("masque: failed to dial target: ", err)
+		return err
+	}
+	defer cc.Close()
+
+	ro.SrcAddr = cc.LocalAddr().String()
+	ro.DstAddr = cc.RemoteAddr().String()
+
+	// Send 200 OK response
+	w.WriteHeader(http.StatusOK)
+
+	// Flush the response headers
+	if flusher, ok := w.(http.Flusher); ok {
+		flusher.Flush()
+	}
+
+	// Get the underlying HTTP/3 stream
+	stream := streamer.HTTPStream()
+
+	// Resolve target address for StreamConn
+	raddr, err := net.ResolveTCPAddr("tcp", targetAddr)
+	if err != nil {
+		log.Error("masque: failed to resolve target address: ", err)
+		return err
+	}
+
+	// Create stream connection wrapping the HTTP/3 stream
+	streamConn := masque_util.NewStreamConn(stream, laddr, raddr)
+	defer streamConn.Close()
+
+	// Wrap with traffic limiter
+	var clientConn net.Conn = streamConn
+	clientConn = traffic_wrapper.WrapConn(
+		clientConn,
+		h.limiter,
+		clientID,
+		limiter.ServiceOption(h.options.Service),
+		limiter.ScopeOption(limiter.ScopeClient),
+		limiter.NetworkOption("tcp"),
+		limiter.AddrOption(targetAddr),
+		limiter.ClientOption(clientID),
+		limiter.SrcOption(ro.RemoteAddr),
+	)
+
+	// Track per-client connection stats
+	if h.options.Observer != nil {
+		pstats := h.stats.Stats(clientID)
+		pstats.Add(stats.KindTotalConns, 1)
+		pstats.Add(stats.KindCurrentConns, 1)
+		defer pstats.Add(stats.KindCurrentConns, -1)
+		clientConn = stats_wrapper.WrapConn(clientConn, pstats)
+	}
+
+	// Wrap target with metrics
+	cc = metrics.WrapConn(h.options.Service, cc)
+
+	log.Infof("%s <-> %s", ro.RemoteAddr, targetAddr)
+
+	// Bidirectional relay
+	xnet.Pipe(ctx, clientConn, cc)
+
+	log.WithFields(map[string]any{
+		"duration": time.Since(ro.Time),
+	}).Infof("%s >-< %s", ro.RemoteAddr, targetAddr)
+
+	return nil
+}
+
 // fixedTargetPacketConn wraps a PacketConn and always reads/writes to a fixed target address
 type fixedTargetPacketConn struct {
 	net.PacketConn
@@ -174,3 +174,105 @@ var (
 	_ net.Conn       = (*DatagramConn)(nil)
 	_ net.PacketConn = (*DatagramConn)(nil)
 )
+
+// StreamReadWriter is an interface for reading and writing to HTTP/3 streams.
+// Both http3.Stream and http3.RequestStream implement this interface.
+type StreamReadWriter interface {
+	io.Reader
+	io.Writer
+	io.Closer
+	SetDeadline(t time.Time) error
+	SetReadDeadline(t time.Time) error
+	SetWriteDeadline(t time.Time) error
+}
+
+// StreamConn wraps an HTTP/3 stream as net.Conn for TCP tunneling (RFC 9114).
+// Unlike DatagramConn which uses HTTP/3 datagrams, StreamConn reads/writes
+// directly to the HTTP/3 stream body for reliable, ordered byte streams.
+type StreamConn struct {
+	stream     StreamReadWriter
+	localAddr  net.Addr
+	remoteAddr net.Addr
+	closed     chan struct{}
+	closeOnce  sync.Once
+}
+
+// NewStreamConn creates a new StreamConn wrapping an HTTP/3 stream (server-side).
+// This is used for TCP CONNECT tunneling where data flows through the stream body.
+func NewStreamConn(stream *http3.Stream, laddr, raddr net.Addr) *StreamConn {
+	return &StreamConn{
+		stream:     stream,
+		localAddr:  laddr,
+		remoteAddr: raddr,
+		closed:     make(chan struct{}),
+	}
+}
+
+// NewStreamConnFromRequestStream creates a new StreamConn wrapping an HTTP/3 request stream (client-side).
+// This is used for TCP CONNECT tunneling where data flows through the stream body.
+func NewStreamConnFromRequestStream(stream *http3.RequestStream, laddr, raddr net.Addr) *StreamConn {
+	return &StreamConn{
+		stream:     stream,
+		localAddr:  laddr,
+		remoteAddr: raddr,
+		closed:     make(chan struct{}),
+	}
+}
+
+// Read reads data from the HTTP/3 stream body.
+func (c *StreamConn) Read(b []byte) (n int, err error) {
+	select {
+	case <-c.closed:
+		return 0, net.ErrClosed
+	default:
+	}
+	return c.stream.Read(b)
+}
+
+// Write writes data to the HTTP/3 stream body.
+func (c *StreamConn) Write(b []byte) (n int, err error) {
+	select {
+	case <-c.closed:
+		return 0, net.ErrClosed
+	default:
+	}
+	return c.stream.Write(b)
+}
+
+// Close closes the stream connection.
+func (c *StreamConn) Close() error {
+	var err error
+	c.closeOnce.Do(func() {
+		close(c.closed)
+		err = c.stream.Close()
+	})
+	return err
+}
+
+// LocalAddr returns the local network address.
+func (c *StreamConn) LocalAddr() net.Addr {
+	return c.localAddr
+}
+
+// RemoteAddr returns the remote network address.
+func (c *StreamConn) RemoteAddr() net.Addr {
+	return c.remoteAddr
+}
+
+// SetDeadline sets the read and write deadlines.
+func (c *StreamConn) SetDeadline(t time.Time) error {
+	return c.stream.SetDeadline(t)
+}
+
+// SetReadDeadline sets the read deadline.
+func (c *StreamConn) SetReadDeadline(t time.Time) error {
+	return c.stream.SetReadDeadline(t)
+}
+
+// SetWriteDeadline sets the write deadline.
+func (c *StreamConn) SetWriteDeadline(t time.Time) error {
+	return c.stream.SetWriteDeadline(t)
+}
+
+// Ensure StreamConn implements net.Conn
+var _ net.Conn = (*StreamConn)(nil)