From c5f91232cb6a4c634dbff9dfb7c12a476d93ae83 Mon Sep 17 00:00:00 2001 From: David Manouchehri Date: Sun, 28 Dec 2025 23:18:41 +0000 Subject: [PATCH] Add TCP CONNECT support to MASQUE (RFC 9114) Extend MASQUE implementation to support TCP tunneling via standard HTTP/3 CONNECT method, in addition to existing UDP support via CONNECT-UDP: - Add StreamConn type for TCP data transfer over HTTP/3 stream body - Update handler to dispatch based on :protocol pseudo-header: - "connect-udp" for UDP (RFC 9298) - Empty/"HTTP/3.0" for TCP (RFC 9114) - Add handleConnectTCP() using bidirectional stream relay - Update connector to support both TCP and UDP networks - Add connectTCP() for standard CONNECT requests TCP uses :authority for target address and stream body for data. UDP uses path template and HTTP/3 datagrams for data. --- connector/masque/connector.go | 112 ++++++++++++++++++-- handler/masque/handler.go | 193 ++++++++++++++++++++++++++++++---- internal/util/masque/conn.go | 102 ++++++++++++++++++ 3 files changed, 382 insertions(+), 25 deletions(-) diff --git a/connector/masque/connector.go b/connector/masque/connector.go index 9a2b8971..8040bced 100644 --- a/connector/masque/connector.go +++ b/connector/masque/connector.go @@ -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) diff --git a/handler/masque/handler.go b/handler/masque/handler.go index 4330e415..d6e798d4 100644 --- a/handler/masque/handler.go +++ b/handler/masque/handler.go @@ -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 diff --git a/internal/util/masque/conn.go b/internal/util/masque/conn.go index 3f31c6f6..3ea66e20 100644 --- a/internal/util/masque/conn.go +++ b/internal/util/masque/conn.go @@ -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)