From dee34e0300e62ec204fe53ab7fd714f32a9541ad Mon Sep 17 00:00:00 2001 From: Shtorm <108103062+shtorm-7@users.noreply.github.com> Date: Mon, 19 Oct 2026 17:05:38 +0300 Subject: [PATCH] masque: Add L4 proxy mode and handshake timeout --- common/cloudflare/constant.go | 7 +- examples/masque/client.json | 6 +- option/masque.go | 2 + protocol/masque/outbound.go | 52 ++++++--- transport/masque/l4proxy.go | 209 ++++++++++++++++++++++++++++++++++ transport/masque/masque.go | 70 +++++++----- transport/masque/options.go | 1 + transport/masque/tunnel.go | 10 +- 8 files changed, 312 insertions(+), 45 deletions(-) create mode 100644 transport/masque/l4proxy.go diff --git a/common/cloudflare/constant.go b/common/cloudflare/constant.go index e5108b07..d4c9006f 100644 --- a/common/cloudflare/constant.go +++ b/common/cloudflare/constant.go @@ -1,9 +1,10 @@ package cloudflare const ( - ApiUrl = "https://api.cloudflareclient.com" - ApiVersion = "v0a4471" - ConnectSNI = "consumer-masque.cloudflareclient.com" + ApiUrl = "https://api.cloudflareclient.com" + ApiVersion = "v0a4471" + ConnectSNI = "consumer-masque.cloudflareclient.com" + L4ConnectSNI = "consumer-masque-proxy.cloudflareclient.com" // unused for now ZeroTierSNI = "zt-masque.cloudflareclient.com" ConnectURI = "https://cloudflareaccess.com" diff --git a/examples/masque/client.json b/examples/masque/client.json index 40cff115..2e10f676 100644 --- a/examples/masque/client.json +++ b/examples/masque/client.json @@ -28,6 +28,8 @@ "system": false, "name": "masque0", "use_http2": false, + // TCP-only CONNECT proxy over HTTP/3 instead of a CONNECT-IP tunnel + "use_l4_proxy": false, "use_ipv6": false, "profile": { "detour": "direct", @@ -43,10 +45,12 @@ "udp_keepalive_period": "30s", "udp_initial_packet_size": 0, "reconnect_delay": "5s", + // Timeout for establishing a connection to the endpoint; 0 = no timeout + "handshake_timeout": "0s", "congestion_controller": "bbr", "cwnd": 0, "tls": { // TLS fields for HTTP2 - "server_name": "", // SNI; empty = default "consumer-masque.cloudflareclient.com" + "server_name": "", // SNI; empty = default "consumer-masque.cloudflareclient.com" ("consumer-masque-proxy.cloudflareclient.com" with use_l4_proxy) "insecure": false, "cipher_suites": [], "curve_preferences": [], diff --git a/option/masque.go b/option/masque.go index 94c7048e..bb50f957 100644 --- a/option/masque.go +++ b/option/masque.go @@ -12,6 +12,7 @@ type MASQUEOutboundOptions struct { Name string `json:"name,omitempty"` AllowedIPs badoption.Listable[netip.Prefix] `json:"allowed_ips,omitempty"` UseHTTP2 bool `json:"use_http2,omitempty"` + UseL4Proxy bool `json:"use_l4_proxy,omitempty"` UseIPv6 bool `json:"use_ipv6,omitempty"` Profile CloudflareProfile `json:"profile,omitempty"` Address string `json:"address,omitempty"` @@ -20,6 +21,7 @@ type MASQUEOutboundOptions struct { UDPKeepalivePeriod badoption.Duration `json:"udp_keepalive_period,omitempty"` UDPInitialPacketSize uint16 `json:"udp_initial_packet_size,omitempty"` ReconnectDelay badoption.Duration `json:"reconnect_delay,omitempty"` + HandshakeTimeout badoption.Duration `json:"handshake_timeout,omitempty"` CongestionController string `json:"congestion_controller,omitempty"` CWND int `json:"cwnd,omitempty"` MASQUEOutboundTLSOptionsContainer diff --git a/protocol/masque/outbound.go b/protocol/masque/outbound.go index 415198a8..3eed12f0 100644 --- a/protocol/masque/outbound.go +++ b/protocol/masque/outbound.go @@ -33,21 +33,30 @@ func RegisterOutbound(registry *outbound.Registry) { outbound.Register[option.MASQUEOutboundOptions](registry, C.TypeMASQUE, NewOutbound) } +type tunnelClient interface { + N.Dialer + Close() error +} + type Outbound struct { outbound.Adapter ctx context.Context dnsRouter adapter.DNSRouter logger logger.ContextLogger options option.MASQUEOutboundOptions - tunnel *masque.Tunnel + client tunnelClient startHandler func() await chan struct{} } func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.MASQUEOutboundOptions) (adapter.Outbound, error) { + networks := []string{N.NetworkTCP, N.NetworkUDP} + if options.UseL4Proxy { + networks = []string{N.NetworkTCP} + } outbound := &Outbound{ - Adapter: outbound.NewAdapterWithDialerOptions(C.TypeMASQUE, tag, []string{N.NetworkTCP, N.NetworkUDP}, options.DialerOptions), + Adapter: outbound.NewAdapterWithDialerOptions(C.TypeMASQUE, tag, networks, options.DialerOptions), ctx: ctx, dnsRouter: service.FromContext[adapter.DNSRouter](ctx), logger: logger, @@ -103,6 +112,9 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL return } serverName := cloudflare.ConnectSNI + if options.UseL4Proxy { + serverName = cloudflare.L4ConnectSNI + } if options.TLS != nil && options.TLS.ServerName != "" { serverName = options.TLS.ServerName } @@ -111,6 +123,7 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL logger.ErrorContext(ctx, E.New("failed to prepare TLS config: ", err)) return } + useHTTP2 := options.UseHTTP2 && !options.UseL4Proxy var endpoint net.Addr if options.Address != "" { endpointPort := options.Port @@ -122,13 +135,13 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL logger.ErrorContext(ctx, E.New("invalid endpoint address: ", options.Address)) return } - if options.UseHTTP2 { + if useHTTP2 { endpoint = &net.TCPAddr{IP: endpointIP, Port: int(endpointPort)} } else { endpoint = &net.UDPAddr{IP: endpointIP, Port: int(endpointPort)} } } else { - endpoint, err = appConfig.SelectEndpointFromConfig(options.UseHTTP2, options.UseIPv6, 443) + endpoint, err = appConfig.SelectEndpointFromConfig(useHTTP2, options.UseIPv6, 443) if err != nil { logger.ErrorContext(ctx, E.New("failed to select endpoint: ", err)) return @@ -165,6 +178,18 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL logger.ErrorContext(ctx, err) return } + if options.UseL4Proxy { + outbound.client = masque.NewL4Client(logger, masque.L4ClientOptions{ + Dialer: outboundDialer, + TLSConfig: tlsConfig, + Endpoint: endpoint, + UDPKeepalivePeriod: udpKeepalivePeriod, + UDPInitialPacketSize: options.UDPInitialPacketSize, + HandshakeTimeout: time.Duration(options.HandshakeTimeout), + CongestionControl: congestionControl, + }) + return + } tunnel, err := masque.NewTunnel( ctx, logger, @@ -191,6 +216,7 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL UDPKeepalivePeriod: udpKeepalivePeriod, UDPInitialPacketSize: options.UDPInitialPacketSize, ReconnectDelay: options.ReconnectDelay.Build(), + HandshakeTimeout: time.Duration(options.HandshakeTimeout), CongestionControl: congestionControl, }, ) @@ -198,12 +224,12 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL logger.ErrorContext(ctx, err) return } - outbound.tunnel = tunnel - if err = outbound.tunnel.Start(false); err != nil { + outbound.client = tunnel + if err = tunnel.Start(false); err != nil { logger.ErrorContext(ctx, err) return } - if err = outbound.tunnel.Start(true); err != nil { + if err = tunnel.Start(true); err != nil { logger.ErrorContext(ctx, err) return } @@ -219,7 +245,7 @@ func (w *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error { if err := w.isTunnelInitialized(w.ctx); err != nil { return err } - return w.tunnel.Close() + return w.client.Close() }) go w.startHandler() return nil @@ -240,11 +266,11 @@ func (w *Outbound) DialContext(ctx context.Context, network string, destination if err != nil { return nil, err } - return N.DialSerial(ctx, w.tunnel, network, destination, destinationAddresses) + return N.DialSerial(ctx, w.client, network, destination, destinationAddresses) } else if !destination.Addr.IsValid() { return nil, E.New("invalid destination: ", destination) } - return w.tunnel.DialContext(ctx, network, destination) + return w.client.DialContext(ctx, network, destination) } func (w *Outbound) ListenPacketWithDestination(ctx context.Context, destination M.Socksaddr) (net.PacketConn, netip.Addr, error) { @@ -257,9 +283,9 @@ func (w *Outbound) ListenPacketWithDestination(ctx context.Context, destination if err != nil { return nil, netip.Addr{}, err } - return N.ListenSerial(ctx, w.tunnel, destination, destinationAddresses) + return N.ListenSerial(ctx, w.client, destination, destinationAddresses) } - packetConn, err := w.tunnel.ListenPacket(ctx, destination) + packetConn, err := w.client.ListenPacket(ctx, destination) if err != nil { return nil, netip.Addr{}, err } @@ -286,7 +312,7 @@ func (w *Outbound) isTunnelInitialized(ctx context.Context) error { case <-ctx.Done(): return ctx.Err() } - if w.tunnel == nil { + if w.client == nil { return E.New("tunnel not initialized") } return nil diff --git a/transport/masque/l4proxy.go b/transport/masque/l4proxy.go new file mode 100644 index 00000000..169f9265 --- /dev/null +++ b/transport/masque/l4proxy.go @@ -0,0 +1,209 @@ +package masque + +import ( + "context" + "net" + "net/http" + "time" + + "github.com/sagernet/quic-go" + "github.com/sagernet/quic-go/congestion" + "github.com/sagernet/quic-go/http3" + E "github.com/sagernet/sing/common/exceptions" + "github.com/sagernet/sing/common/logger" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" + aTLS "github.com/sagernet/sing/common/tls" + "golang.org/x/sync/semaphore" +) + +type L4ClientOptions struct { + Dialer N.Dialer + TLSConfig aTLS.Config + Endpoint net.Addr + UDPKeepalivePeriod time.Duration + UDPInitialPacketSize uint16 + HandshakeTimeout time.Duration + CongestionControl func(conn *quic.Conn) congestion.CongestionControl +} + +type L4Client struct { + logger logger.ContextLogger + options L4ClientOptions + access *semaphore.Weighted + connection *l4Connection + closed bool +} + +type l4Connection struct { + packetConn net.PacketConn + quicConn *quic.Conn + clientConn *http3.ClientConn +} + +func NewL4Client(logger logger.ContextLogger, options L4ClientOptions) *L4Client { + return &L4Client{ + logger: logger, + options: options, + access: semaphore.NewWeighted(1), + } +} + +func (c *L4Client) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) { + if N.NetworkName(network) != N.NetworkTCP { + return nil, E.New("MASQUE L4 proxy mode does not support ", network) + } + connection, err := c.openConnection(ctx) + if err != nil { + return nil, err + } + stream, err := connection.clientConn.OpenRequestStream(ctx) + if err != nil { + c.closeConnection(connection) + return nil, err + } + address := destination.String() + request, err := http.NewRequestWithContext(ctx, http.MethodConnect, "https://"+address, nil) + if err != nil { + stream.CancelRead(quic.StreamErrorCode(http3.ErrCodeRequestCanceled)) + stream.CancelWrite(quic.StreamErrorCode(http3.ErrCodeRequestCanceled)) + return nil, err + } + request.Host = address + stop := context.AfterFunc(ctx, func() { + stream.CancelRead(quic.StreamErrorCode(http3.ErrCodeRequestCanceled)) + stream.CancelWrite(quic.StreamErrorCode(http3.ErrCodeRequestCanceled)) + }) + var response *http.Response + err = stream.SendRequestHeader(request) + if err == nil { + response, err = stream.ReadResponse() + } + if err == nil && !stop() { + err = ctx.Err() + } + if err == nil && (response.StatusCode < 200 || response.StatusCode > 299) { + err = E.New("CONNECT rejected with status ", response.StatusCode) + } + if err != nil { + stop() + stream.CancelRead(quic.StreamErrorCode(http3.ErrCodeRequestCanceled)) + stream.CancelWrite(quic.StreamErrorCode(http3.ErrCodeRequestCanceled)) + return nil, err + } + return &l4StreamConn{ + stream: stream, + quicConn: connection.quicConn, + }, nil +} + +func (c *L4Client) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) { + return nil, E.New("MASQUE L4 proxy mode does not support UDP") +} + +func (c *L4Client) Close() error { + _ = c.access.Acquire(context.Background(), 1) + connection := c.connection + c.connection = nil + c.closed = true + c.access.Release(1) + if connection != nil { + connection.Close() + } + return nil +} + +func (c *L4Client) openConnection(ctx context.Context) (*l4Connection, error) { + err := c.access.Acquire(ctx, 1) + if err != nil { + return nil, err + } + defer c.access.Release(1) + if c.closed { + return nil, net.ErrClosed + } + if c.connection != nil { + if c.connection.quicConn.Context().Err() == nil { + return c.connection, nil + } + c.connection.Close() + c.connection = nil + } + dialCtx := ctx + if c.options.HandshakeTimeout > 0 { + var cancel context.CancelFunc + dialCtx, cancel = context.WithTimeout(ctx, c.options.HandshakeTimeout) + defer cancel() + } + c.logger.InfoContext(ctx, "establishing MASQUE L4 proxy connection to ", c.options.Endpoint) + packetConn, quicConn, err := dialQUIC( + dialCtx, + c.options.Dialer, + c.options.TLSConfig, + DefaultQuicConfig(c.options.UDPKeepalivePeriod, c.options.UDPInitialPacketSize), + c.options.Endpoint, + c.options.CongestionControl, + ) + if err != nil { + return nil, E.Cause(err, "connect MASQUE L4 proxy") + } + transport := &http3.Transport{} + c.connection = &l4Connection{ + packetConn: packetConn, + quicConn: quicConn, + clientConn: transport.NewClientConn(quicConn), + } + return c.connection, nil +} + +func (c *L4Client) closeConnection(connection *l4Connection) { + _ = c.access.Acquire(context.Background(), 1) + if c.connection == connection { + c.connection = nil + } + c.access.Release(1) + connection.Close() +} + +func (c *l4Connection) Close() { + _ = c.clientConn.CloseWithError(0, "client closed") + _ = c.packetConn.Close() +} + +type l4StreamConn struct { + stream *http3.RequestStream + quicConn *quic.Conn +} + +func (c *l4StreamConn) Read(b []byte) (int, error) { + return c.stream.Read(b) +} + +func (c *l4StreamConn) Write(b []byte) (int, error) { + return c.stream.Write(b) +} + +func (c *l4StreamConn) Close() error { + c.stream.CancelRead(quic.StreamErrorCode(http3.ErrCodeNoError)) + return c.stream.Close() +} + +func (c *l4StreamConn) LocalAddr() net.Addr { + return c.quicConn.LocalAddr() +} + +func (c *l4StreamConn) RemoteAddr() net.Addr { + return c.quicConn.RemoteAddr() +} + +func (c *l4StreamConn) SetDeadline(t time.Time) error { + return c.stream.SetDeadline(t) +} + +func (c *l4StreamConn) SetReadDeadline(t time.Time) error { + return c.stream.SetReadDeadline(t) +} + +func (c *l4StreamConn) SetWriteDeadline(t time.Time) error { + return c.stream.SetWriteDeadline(t) +} diff --git a/transport/masque/masque.go b/transport/masque/masque.go index fe419bb9..79a09956 100644 --- a/transport/masque/masque.go +++ b/transport/masque/masque.go @@ -74,36 +74,13 @@ func ConnectTunnel(ctx context.Context, dialer N.Dialer, tlsConfig aTLS.Config, return ConnectTunnelH2(ctx, dialer, tlsConfig, h2Endpoint, connectUri) } - quicEndpoint, ok := endpoint.(*net.UDPAddr) - if !ok || quicEndpoint == nil { - return nil, nil, nil, errors.New("missing HTTP/3 UDP endpoint") - } - endpointAddrPort := quicEndpoint.AddrPort() - udpConn, err := dialer.ListenPacket( - ctx, - M.SocksaddrFromNetIP( - netip.AddrPortFrom( - endpointAddrPort.Addr().Unmap(), - endpointAddrPort.Port(), - ), - ), - ) + udpConn, conn, err := dialQUIC(ctx, dialer, tlsConfig, quicConfig, endpoint, congestionControl) if err != nil { return nil, nil, nil, err } - conn, err := qtls.Dial( - ctx, - bufio.NewBindPacketConn(udpConn, quicEndpoint), - NewTLSConfig(tlsConfig, 20), - quicConfig, - ) - if err != nil { - _ = udpConn.Close() - return nil, nil, nil, err - } - if congestionControl != nil { - conn.SetCongestionControl(congestionControl(conn)) - } + stop := context.AfterFunc(ctx, func() { + _ = conn.CloseWithError(0, "handshake canceled") + }) tr := &http3.Transport{ EnableDatagrams: true, AdditionalSettings: map[uint64]uint64{ @@ -119,6 +96,7 @@ func ConnectTunnel(ctx context.Context, dialer N.Dialer, tlsConfig aTLS.Config, } ipConn, rsp, err := connectip.Dial(ctx, hconn, template, "cf-connect-ip", additionalHeaders, true) if err != nil { + stop() _ = tr.Close() _ = conn.CloseWithError(0, "connect-ip dial failed") _ = udpConn.Close() @@ -144,7 +122,11 @@ func ConnectTunnel(ctx context.Context, dialer N.Dialer, tlsConfig aTLS.Config, }), }, }) + if err == nil && !stop() { + err = ctx.Err() + } if err != nil { + stop() _ = ipConn.Close() _ = tr.Close() _ = udpConn.Close() @@ -158,3 +140,37 @@ func ConnectTunnel(ctx context.Context, dialer N.Dialer, tlsConfig aTLS.Config, }) return closer, newQuicIpConn(ipConn), rsp, nil } + +func dialQUIC(ctx context.Context, dialer N.Dialer, tlsConfig aTLS.Config, quicConfig *quic.Config, endpoint net.Addr, congestionControl func(conn *quic.Conn) congestion.CongestionControl) (net.PacketConn, *quic.Conn, error) { + quicEndpoint, ok := endpoint.(*net.UDPAddr) + if !ok || quicEndpoint == nil { + return nil, nil, errors.New("missing HTTP/3 UDP endpoint") + } + endpointAddrPort := quicEndpoint.AddrPort() + udpConn, err := dialer.ListenPacket( + ctx, + M.SocksaddrFromNetIP( + netip.AddrPortFrom( + endpointAddrPort.Addr().Unmap(), + endpointAddrPort.Port(), + ), + ), + ) + if err != nil { + return nil, nil, err + } + conn, err := qtls.Dial( + ctx, + bufio.NewBindPacketConn(udpConn, quicEndpoint), + NewTLSConfig(tlsConfig, 20), + quicConfig, + ) + if err != nil { + _ = udpConn.Close() + return nil, nil, err + } + if congestionControl != nil { + conn.SetCongestionControl(congestionControl(conn)) + } + return udpConn, conn, nil +} diff --git a/transport/masque/options.go b/transport/masque/options.go index be65a1cd..b98a8e81 100644 --- a/transport/masque/options.go +++ b/transport/masque/options.go @@ -25,5 +25,6 @@ type TunnelOptions struct { UDPKeepalivePeriod time.Duration UDPInitialPacketSize uint16 ReconnectDelay time.Duration + HandshakeTimeout time.Duration CongestionControl func(conn *quic.Conn) congestion.CongestionControl } diff --git a/transport/masque/tunnel.go b/transport/masque/tunnel.go index a8bc8cef..933c7dfe 100644 --- a/transport/masque/tunnel.go +++ b/transport/masque/tunnel.go @@ -179,8 +179,15 @@ func (e *Tunnel) getIpConn() (IpConn, error) { defer timer.Stop() for { e.logger.NoticeContext(e.ctx, fmt.Errorf("Establishing MASQUE connection to %s", e.options.Endpoint)) + var connectCtx context.Context + var cancel context.CancelFunc + if e.options.HandshakeTimeout > 0 { + connectCtx, cancel = context.WithTimeout(e.ctx, e.options.HandshakeTimeout) + } else { + connectCtx, cancel = context.WithCancel(e.ctx) + } closer, ipConn, rsp, err := ConnectTunnel( - e.ctx, + connectCtx, e.options.Dialer, e.options.TLSConfig, DefaultQuicConfig(e.options.UDPKeepalivePeriod, e.options.UDPInitialPacketSize), @@ -189,6 +196,7 @@ func (e *Tunnel) getIpConn() (IpConn, error) { e.options.UseHTTP2, e.options.CongestionControl, ) + cancel() if err != nil { e.logger.ErrorContext(e.ctx, fmt.Errorf("Failed to connect tunnel: %v", err)) timer.Reset(e.options.ReconnectDelay)