From c77a4d0e4932a53f3eeafafdfef877b56fd660ad Mon Sep 17 00:00:00 2001 From: Cluvex <125141320+CluvexStudio@users.noreply.github.com> Date: Mon, 12 Oct 2026 23:10:04 +0330 Subject: [PATCH] MASQUE client: Support XMUX (`"maxConnections": 1` by default); Use Chrome's User-Agent by default (#7121) https://github.com/XTLS/Xray-core/pull/6810#issuecomment-5842441136 https://github.com/XTLS/Xray-core/pull/6878#issuecomment-6095112915 https://github.com/XTLS/Xray-core/pull/7121#issuecomment-6117636043 --------- Co-authored-by: SVLAVR --- app/proxyman/inbound/inbound.go | 2 +- infra/conf/masque_test.go | 79 ++++++++-- infra/conf/transport_method.go | 18 +++ infra/conf/xray.go | 6 +- proxy/masque/client.go | 137 +++++++++++++----- proxy/masque/client_test.go | 54 +++++++ transport/internet/masque/config.go | 9 ++ transport/internet/masque/config.pb.go | 32 ++-- transport/internet/masque/config.proto | 3 + transport/internet/masque/dialer.go | 51 ++++--- transport/internet/masque/dialer_test.go | 25 ++++ transport/internet/masque/http2.go | 29 ++-- .../internet/masque/http2_server_test.go | 2 +- transport/internet/masque/http2_test.go | 64 +++++++- transport/internet/splithttp/mux.go | 16 +- transport/internet/splithttp/mux_test.go | 20 +++ 16 files changed, 456 insertions(+), 91 deletions(-) create mode 100644 proxy/masque/client_test.go diff --git a/app/proxyman/inbound/inbound.go b/app/proxyman/inbound/inbound.go index 411de8598..c0ced47aa 100644 --- a/app/proxyman/inbound/inbound.go +++ b/app/proxyman/inbound/inbound.go @@ -174,7 +174,7 @@ func NewHandler(ctx context.Context, config *core.InboundHandlerConfig) (inbound Mark: streamSettings.SocketSettings.Mark, }) } - if streamSettings != nil && streamSettings.ProtocolName == "splithttp" { + if streamSettings != nil && (streamSettings.ProtocolName == "splithttp" || streamSettings.ProtocolName == "masque") { ctx = session.ContextWithAllowedNetwork(ctx, net.Network_UDP) } diff --git a/infra/conf/masque_test.go b/infra/conf/masque_test.go index 559164df1..803f397dd 100644 --- a/infra/conf/masque_test.go +++ b/infra/conf/masque_test.go @@ -16,8 +16,61 @@ import ( . "github.com/xtls/xray-core/infra/conf" masqueproxy "github.com/xtls/xray-core/proxy/masque" "github.com/xtls/xray-core/transport/internet/masque" + "github.com/xtls/xray-core/transport/internet/splithttp" ) +func defaultMasqueXmux() *splithttp.XmuxConfig { + return &splithttp.XmuxConfig{ + MaxConcurrency: &splithttp.RangeConfig{}, + MaxConnections: &splithttp.RangeConfig{From: 1, To: 1}, + CMaxReuseTimes: &splithttp.RangeConfig{}, + HMaxRequestTimes: &splithttp.RangeConfig{}, + HMaxReusableSecs: &splithttp.RangeConfig{From: 1800, To: 3000}, + } +} + +func TestMasqueXmuxConfig(t *testing.T) { + creator := func() Buildable { + return new(MasqueConfig) + } + + runMultiTestCase(t, []TestCase{ + { + Input: `{"xmux": {"maxConnections": "2-4", "hMaxReusableSecs": 60, "hKeepAlivePeriod": -1}}`, + Parser: loadJSON(creator), + Output: &masque.Config{ + Path: "/.well-known/masque/ip/*/*/", + Xmux: &splithttp.XmuxConfig{ + MaxConcurrency: &splithttp.RangeConfig{}, + MaxConnections: &splithttp.RangeConfig{From: 2, To: 4}, + CMaxReuseTimes: &splithttp.RangeConfig{}, + HMaxRequestTimes: &splithttp.RangeConfig{}, + HMaxReusableSecs: &splithttp.RangeConfig{From: 60, To: 60}, + HKeepAlivePeriod: -1, + }, + }, + }, + { + Input: `{"xmux": {"maxConcurrency": 8}}`, + Parser: loadJSON(creator), + Output: &masque.Config{ + Path: "/.well-known/masque/ip/*/*/", + Xmux: &splithttp.XmuxConfig{ + MaxConcurrency: &splithttp.RangeConfig{From: 8, To: 8}, + MaxConnections: &splithttp.RangeConfig{}, + CMaxReuseTimes: &splithttp.RangeConfig{}, + HMaxRequestTimes: &splithttp.RangeConfig{}, + HMaxReusableSecs: &splithttp.RangeConfig{}, + }, + }, + }, + }) + + if _, err := loadJSON(creator)(`{"xmux": {"maxConnections": 2, "maxConcurrency": 8}}`); err == nil { + t.Error("expected an error for maxConnections with maxConcurrency") + } +} + func TestMasqueConfig(t *testing.T) { creator := func() Buildable { return new(MasqueConfig) @@ -27,7 +80,7 @@ func TestMasqueConfig(t *testing.T) { { Input: `{}`, Parser: loadJSON(creator), - Output: &masque.Config{Path: "/.well-known/masque/ip/*/*/"}, + Output: &masque.Config{Path: "/.well-known/masque/ip/*/*/", Xmux: defaultMasqueXmux()}, }, { Input: `{ @@ -40,12 +93,13 @@ func TestMasqueConfig(t *testing.T) { Host: "example.com:8443", Path: "/.well-known/masque/ip/*/*/", Headers: map[string]string{"Authorization": "Basic dTpw"}, + Xmux: defaultMasqueXmux(), }, }, { Input: `{"path": "/masque/ip{?target,ipproto}"}`, Parser: loadJSON(creator), - Output: &masque.Config{Path: "/masque/ip?target=*&ipproto=*"}, + Output: &masque.Config{Path: "/masque/ip?target=*&ipproto=*", Xmux: defaultMasqueXmux()}, }, { Input: `{"user": "u", "pass": "p:q", "headers": {"X-Token": "a"}}`, @@ -53,6 +107,7 @@ func TestMasqueConfig(t *testing.T) { Output: &masque.Config{ Path: "/.well-known/masque/ip/*/*/", Headers: map[string]string{"Authorization": "Basic dTpwOnE=", "X-Token": "a"}, + Xmux: defaultMasqueXmux(), }, }, }) @@ -122,6 +177,7 @@ func TestMasqueWarpConfig(t *testing.T) { Host: "cloudflareaccess.com", Path: "/", Warp: &masque.Warp{PrivateKey: pkcs8, PublicKey: publicKey, Address: address}, + Xmux: defaultMasqueXmux(), }, }, }) @@ -134,6 +190,7 @@ func TestMasqueWarpConfig(t *testing.T) { Host: "example.com", Path: "/warp", Warp: &masque.Warp{PrivateKey: pkcs8, PublicKey: publicKey, Address: address}, + Xmux: defaultMasqueXmux(), }, }, }) @@ -164,6 +221,7 @@ func TestMasqueWarpConfig(t *testing.T) { Host: "cloudflareaccess.com", Path: "/", Warp: &masque.Warp{PrivateKey: pkcs8, PublicKey: publicKey, Address: []string{"172.16.0.2/32"}}, + Xmux: defaultMasqueXmux(), }, }, }) @@ -199,18 +257,17 @@ func TestMasqueOutboundConfig(t *testing.T) { return err } - if err := build(`{ - "protocol": "masque", - "settings": {"address": "example.com", "port": 443}, - "streamSettings": {"network": "masque", "security": "tls"}, - "mux": {"enabled": false, "concurrency": -1} - }`); err != nil { - t.Error(err) + for _, input := range []string{ + `{"protocol": "masque", "settings": {"address": "example.com", "port": 443}, "streamSettings": {"network": "masque", "security": "tls"}, "mux": {"enabled": false, "concurrency": -1}}`, + `{"protocol": "masque", "settings": {"address": "example.com", "port": 443}, "streamSettings": {"network": "masque", "security": "tls"}, "mux": {"enabled": true}}`, + `{"protocol": "masque", "settings": {"address": "example.com", "port": 443}, "streamSettings": {"network": "masque", "security": "tls"}, "mux": {"enabled": true, "concurrency": -1, "xudpConcurrency": 8}}`, + } { + if err := build(input); err != nil { + t.Errorf("%s: %v", input, err) + } } for _, input := range []string{ `{"protocol": "masque", "settings": {"address": "example.com"}, "streamSettings": {"network": "masque", "security": "tls"}}`, - `{"protocol": "masque", "settings": {"address": "example.com", "port": 443}, "streamSettings": {"network": "masque", "security": "tls"}, "mux": {"enabled": true}}`, - `{"protocol": "masque", "settings": {"address": "example.com", "port": 443}, "streamSettings": {"network": "masque", "security": "tls"}, "mux": {"enabled": true, "concurrency": -1}}`, `{"protocol": "freedom", "streamSettings": {"network": "masque", "security": "tls"}}`, } { if err := build(input); err == nil { diff --git a/infra/conf/transport_method.go b/infra/conf/transport_method.go index 2d088ce00..d36d6e809 100644 --- a/infra/conf/transport_method.go +++ b/infra/conf/transport_method.go @@ -808,6 +808,7 @@ type MasqueConfig struct { Pass string `json:"pass"` Headers map[string]string `json:"headers"` Warp *MasqueWarpConfig `json:"warp"` + Xmux XmuxConfig `json:"xmux"` } func (c *MasqueConfig) Build() (proto.Message, error) { @@ -878,11 +879,28 @@ func (c *MasqueConfig) Build() (proto.Message, error) { } headers["Authorization"] = "Basic " + base64.StdEncoding.EncodeToString([]byte(c.User+":"+c.Pass)) } + if c.Xmux.MaxConnections.To > 0 && c.Xmux.MaxConcurrency.To > 0 { + return nil, errors.New("maxConnections cannot be specified together with maxConcurrency") + } + if c.Xmux == (XmuxConfig{}) { + c.Xmux.MaxConnections.From = 1 + c.Xmux.MaxConnections.To = 1 + c.Xmux.HMaxReusableSecs.From = 1800 + c.Xmux.HMaxReusableSecs.To = 3000 + } return &masque.Config{ Host: host, Path: path, Headers: headers, Warp: warp, + Xmux: &splithttp.XmuxConfig{ + MaxConcurrency: newRangeConfig(c.Xmux.MaxConcurrency), + MaxConnections: newRangeConfig(c.Xmux.MaxConnections), + CMaxReuseTimes: newRangeConfig(c.Xmux.CMaxReuseTimes), + HMaxRequestTimes: newRangeConfig(c.Xmux.HMaxRequestTimes), + HMaxReusableSecs: newRangeConfig(c.Xmux.HMaxReusableSecs), + HKeepAlivePeriod: c.Xmux.HKeepAlivePeriod, + }, }, nil } diff --git a/infra/conf/xray.go b/infra/conf/xray.go index 1ce1bd06d..d6617024b 100644 --- a/infra/conf/xray.go +++ b/infra/conf/xray.go @@ -344,11 +344,7 @@ func (c *OutboundDetourConfig) Build() (*core.OutboundHandlerConfig, error) { return nil, err } - if _, ok := ts.(*masque.ClientConfig); ok { - if ms := senderSettings.MultiplexSettings; ms != nil && ms.Enabled { - return nil, errors.New(`masque outbound does not support "mux"`) - } - } else if senderSettings.StreamSettings != nil && senderSettings.StreamSettings.ProtocolName == "masque" { + if _, ok := ts.(*masque.ClientConfig); !ok && senderSettings.StreamSettings != nil && senderSettings.StreamSettings.ProtocolName == "masque" { return nil, errors.New("the masque transport can only be used by the masque outbound") } diff --git a/proxy/masque/client.go b/proxy/masque/client.go index 161542f6b..fc6999c3f 100644 --- a/proxy/masque/client.go +++ b/proxy/masque/client.go @@ -7,7 +7,6 @@ import ( "net/netip" "slices" "sync" - "sync/atomic" "time" "golang.zx2c4.com/wireguard/tun" @@ -26,6 +25,7 @@ import ( "github.com/xtls/xray-core/transport" "github.com/xtls/xray-core/transport/internet" "github.com/xtls/xray-core/transport/internet/masque" + "github.com/xtls/xray-core/transport/internet/splithttp" "github.com/xtls/xray-core/transport/internet/stat" "github.com/xtls/xray-core/transport/internet/tls" ) @@ -43,11 +43,15 @@ type Client struct { ctx context.Context cancel context.CancelFunc - tunnel atomic.Pointer[tunnel] - mu sync.Mutex + xmux *splithttp.XmuxManager + dialCtx context.Context + dialer internet.Dialer lastErr error lastErrAt time.Time + + tunnelsMu sync.Mutex + tunnels map[*tunnel]struct{} } func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) { @@ -55,7 +59,8 @@ func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) { p := v.GetFeature(policy.ManagerType()).(policy.Manager) streamSettings := session.StreamSettingsFromContext(ctx).(*internet.MemoryStreamConfig) - if _, ok := streamSettings.ProtocolSettings.(*masque.Config); !ok { + transportConfig, ok := streamSettings.ProtocolSettings.(*masque.Config) + if !ok { return nil, errors.New("not masque transport") } if tls.ConfigFromStreamSettings(streamSettings) == nil { @@ -86,7 +91,13 @@ func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) { server: server, policyManager: p, remoteDNS: remoteDNS, + tunnels: make(map[*tunnel]struct{}), } + xmuxConfig := transportConfig.Xmux + if xmuxConfig == nil { + xmuxConfig = &splithttp.XmuxConfig{} + } + c.xmux = splithttp.NewXmuxManager(*xmuxConfig, c.newXmuxConn) c.ctx, c.cancel = context.WithCancel(context.Background()) return c, nil } @@ -100,10 +111,11 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter ob.Name = "masque" ob.CanSpliceCopy = 3 - t, err := c.getTunnel(ctx, dialer) + xmuxClient, t, err := c.getTunnel(ctx, dialer) if err != nil { return errors.New("failed to establish CONNECT-IP tunnel").Base(err) } + defer xmuxClient.DoneRunning() var newCtx context.Context var newCancel context.CancelFunc @@ -180,40 +192,63 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter return nil } -func (c *Client) getTunnel(ctx context.Context, dialer internet.Dialer) (*tunnel, error) { +func (c *Client) getTunnel(ctx context.Context, dialer internet.Dialer) (*splithttp.XmuxClient, *tunnel, error) { c.mu.Lock() defer c.mu.Unlock() if c.ctx.Err() != nil { - return nil, errors.New("closed") - } - if t := c.tunnel.Load(); t != nil { - select { - case <-t.done: - default: - return t, nil - } + return nil, nil, errors.New("closed") } if err := ctx.Err(); err != nil { - return nil, err + return nil, nil, err } + var xmuxClient *splithttp.XmuxClient if c.lastErr != nil && time.Since(c.lastErrAt) < retryInterval { - return nil, c.lastErr + if xmuxClient = c.xmux.GetExistingXmuxClient(ctx); xmuxClient == nil { + return nil, nil, c.lastErr + } + } else { + c.dialCtx, c.dialer = ctx, dialer + xmuxClient = c.xmux.GetXmuxClient(ctx) + c.dialCtx, c.dialer = nil, nil + if err := xmuxClient.XmuxConn.(*tunnel).err; err != nil { + if xmuxClient = c.xmux.GetExistingXmuxClient(ctx); xmuxClient == nil { + return nil, nil, err + } + } } + t := xmuxClient.XmuxConn.(*tunnel) + if c.ctx.Err() != nil { + return nil, nil, errors.New("closed") + } + xmuxClient.AddRunning() + return xmuxClient, t, nil +} - t, err := c.establish(ctx, dialer) +func (c *Client) newXmuxConn() splithttp.XmuxConn { + t, err := c.establish(c.dialCtx, c.dialer) if err != nil { c.lastErr, c.lastErrAt = err, time.Now() - return nil, err + return failedTunnel(err) } c.lastErr = nil - c.tunnel.Store(t) - if c.ctx.Err() != nil { - if c.tunnel.CompareAndSwap(t, nil) { - t.close() - } - return nil, errors.New("closed") + c.tunnelsMu.Lock() + if c.tunnels == nil { + c.tunnelsMu.Unlock() + t.close() + return failedTunnel(errors.New("closed")) } - return t, nil + c.tunnels[t] = struct{}{} + c.tunnelsMu.Unlock() + if t.IsClosed() { + c.removeTunnel(t) + } + return t +} + +func (c *Client) removeTunnel(t *tunnel) { + c.tunnelsMu.Lock() + delete(c.tunnels, t) + c.tunnelsMu.Unlock() } func (c *Client) establish(ctx context.Context, dialer internet.Dialer) (*tunnel, error) { @@ -229,7 +264,7 @@ func (c *Client) establish(ctx context.Context, dialer internet.Dialer) (*tunnel conn.Close() return nil, errors.New("not a CONNECT-IP connection") } - t, err := newTunnel(conn, mconn.LocalAddrs(), c.remoteDNS) + t, err := newTunnel(conn, mconn.LocalAddrs(), c.remoteDNS, c.removeTunnel) if err != nil { conn.Close() return nil, err @@ -240,7 +275,11 @@ func (c *Client) establish(ctx context.Context, dialer internet.Dialer) (*tunnel func (c *Client) Close() error { c.cancel() - if t := c.tunnel.Swap(nil); t != nil { + c.tunnelsMu.Lock() + tunnels := c.tunnels + c.tunnels = nil + c.tunnelsMu.Unlock() + for t := range tunnels { t.close() } return nil @@ -252,9 +291,31 @@ type tunnel struct { tnet *wireguard.Net done chan struct{} closeOnce sync.Once + onClose func(*tunnel) + err error } -func newTunnel(conn stat.Connection, local []netip.Addr, remoteDNS []netip.Addr) (*tunnel, error) { +func failedTunnel(err error) *tunnel { + t := &tunnel{err: err, done: make(chan struct{})} + t.close() + return t +} + +func (t *tunnel) IsClosed() bool { + select { + case <-t.done: + return true + default: + return false + } +} + +func (t *tunnel) Close() error { + t.close() + return nil +} + +func newTunnel(conn stat.Connection, local []netip.Addr, remoteDNS []netip.Addr, onClose func(*tunnel)) (*tunnel, error) { var dns []netip.Addr for _, addr := range remoteDNS { if slices.ContainsFunc(local, func(l netip.Addr) bool { return l.Is4() == addr.Is4() }) { @@ -271,10 +332,11 @@ func newTunnel(conn stat.Connection, local []netip.Addr, remoteDNS []netip.Addr) return nil, err } t := &tunnel{ - conn: conn, - dev: dev, - tnet: tnet, - done: make(chan struct{}), + conn: conn, + dev: dev, + tnet: tnet, + done: make(chan struct{}), + onClose: onClose, } go t.readFromTunnel() go t.writeToTunnel() @@ -316,8 +378,15 @@ func (t *tunnel) writeToTunnel() { func (t *tunnel) close() { t.closeOnce.Do(func() { close(t.done) - t.conn.Close() - t.dev.Close() + if t.conn != nil { + t.conn.Close() + } + if t.dev != nil { + t.dev.Close() + } + if t.onClose != nil { + t.onClose(t) + } }) } diff --git a/proxy/masque/client_test.go b/proxy/masque/client_test.go new file mode 100644 index 000000000..303c8b442 --- /dev/null +++ b/proxy/masque/client_test.go @@ -0,0 +1,54 @@ +package masque + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/xtls/xray-core/transport/internet/splithttp" +) + +func newTestClient(newConn func(c *Client) splithttp.XmuxConn) *Client { + c := &Client{ctx: context.Background()} + c.xmux = splithttp.NewXmuxManager(splithttp.XmuxConfig{ + MaxConnections: &splithttp.RangeConfig{From: 3, To: 3}, + }, func() splithttp.XmuxConn { + return newConn(c) + }) + return c +} + +func TestGetTunnelFallsBack(t *testing.T) { + refused := errors.New("refused") + healthy := &tunnel{done: make(chan struct{})} + dials := 0 + c := newTestClient(func(c *Client) splithttp.XmuxConn { + if dials++; dials == 1 { + return healthy + } + c.lastErr, c.lastErrAt = refused, time.Now() + return failedTunnel(refused) + }) + for i := 0; i < 5; i++ { + if _, got, err := c.getTunnel(context.Background(), nil); err != nil || got != healthy { + t.Fatalf("flow %d: got %p, %v instead of the healthy tunnel", i, got, err) + } + } + if dials != 2 { + t.Errorf("dialed %d times, a failed tunnel should not be retried within %v", dials, retryInterval) + } +} + +func TestGetTunnelFailsWithoutTunnels(t *testing.T) { + refused := errors.New("refused") + c := newTestClient(func(c *Client) splithttp.XmuxConn { + c.lastErr, c.lastErrAt = refused, time.Now() + return failedTunnel(refused) + }) + for i := 0; i < 2; i++ { + if _, _, err := c.getTunnel(context.Background(), nil); err != refused { + t.Fatalf("flow %d: got %v instead of %v", i, err, refused) + } + } +} diff --git a/transport/internet/masque/config.go b/transport/internet/masque/config.go index f3584cfc0..7ef28467f 100644 --- a/transport/internet/masque/config.go +++ b/transport/internet/masque/config.go @@ -1,6 +1,8 @@ package masque import ( + "time" + "github.com/xtls/xray-core/common" "github.com/xtls/xray-core/transport/internet" ) @@ -9,6 +11,13 @@ const protocolName = "masque" const DefaultPath = "/.well-known/masque/ip/*/*/" +func (c *Config) keepAlivePeriod() time.Duration { + if c.Xmux == nil { + return 0 + } + return time.Duration(c.Xmux.HKeepAlivePeriod) * time.Second +} + func init() { common.Must(internet.RegisterProtocolConfigCreator(protocolName, func() interface{} { return &Config{ diff --git a/transport/internet/masque/config.pb.go b/transport/internet/masque/config.pb.go index 903fd25c7..72ef8da4f 100644 --- a/transport/internet/masque/config.pb.go +++ b/transport/internet/masque/config.pb.go @@ -7,6 +7,7 @@ package masque import ( + splithttp "github.com/xtls/xray-core/transport/internet/splithttp" protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" reflect "reflect" @@ -27,6 +28,7 @@ type Config struct { Path string `protobuf:"bytes,2,opt,name=path,proto3" json:"path,omitempty"` Headers map[string]string `protobuf:"bytes,3,rep,name=headers,proto3" json:"headers,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` Warp *Warp `protobuf:"bytes,4,opt,name=warp,proto3" json:"warp,omitempty"` + Xmux *splithttp.XmuxConfig `protobuf:"bytes,5,opt,name=xmux,proto3" json:"xmux,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -89,6 +91,13 @@ func (x *Config) GetWarp() *Warp { return nil } +func (x *Config) GetXmux() *splithttp.XmuxConfig { + if x != nil { + return x.Xmux + } + return nil +} + type Warp struct { state protoimpl.MessageState `protogen:"open.v1"` PrivateKey []byte `protobuf:"bytes,1,opt,name=private_key,json=privateKey,proto3" json:"private_key,omitempty"` @@ -153,12 +162,13 @@ var File_transport_internet_masque_config_proto protoreflect.FileDescriptor const file_transport_internet_masque_config_proto_rawDesc = "" + "\n" + - "&transport/internet/masque/config.proto\x12\x1exray.transport.internet.masque\"\xf5\x01\n" + + "&transport/internet/masque/config.proto\x12\x1exray.transport.internet.masque\x1a)transport/internet/splithttp/config.proto\"\xb8\x02\n" + "\x06Config\x12\x12\n" + "\x04host\x18\x01 \x01(\tR\x04host\x12\x12\n" + "\x04path\x18\x02 \x01(\tR\x04path\x12M\n" + "\aheaders\x18\x03 \x03(\v23.xray.transport.internet.masque.Config.HeadersEntryR\aheaders\x128\n" + - "\x04warp\x18\x04 \x01(\v2$.xray.transport.internet.masque.WarpR\x04warp\x1a:\n" + + "\x04warp\x18\x04 \x01(\v2$.xray.transport.internet.masque.WarpR\x04warp\x12A\n" + + "\x04xmux\x18\x05 \x01(\v2-.xray.transport.internet.splithttp.XmuxConfigR\x04xmux\x1a:\n" + "\fHeadersEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"`\n" + @@ -184,18 +194,20 @@ func file_transport_internet_masque_config_proto_rawDescGZIP() []byte { var file_transport_internet_masque_config_proto_msgTypes = make([]protoimpl.MessageInfo, 3) var file_transport_internet_masque_config_proto_goTypes = []any{ - (*Config)(nil), // 0: xray.transport.internet.masque.Config - (*Warp)(nil), // 1: xray.transport.internet.masque.Warp - nil, // 2: xray.transport.internet.masque.Config.HeadersEntry + (*Config)(nil), // 0: xray.transport.internet.masque.Config + (*Warp)(nil), // 1: xray.transport.internet.masque.Warp + nil, // 2: xray.transport.internet.masque.Config.HeadersEntry + (*splithttp.XmuxConfig)(nil), // 3: xray.transport.internet.splithttp.XmuxConfig } var file_transport_internet_masque_config_proto_depIdxs = []int32{ 2, // 0: xray.transport.internet.masque.Config.headers:type_name -> xray.transport.internet.masque.Config.HeadersEntry 1, // 1: xray.transport.internet.masque.Config.warp:type_name -> xray.transport.internet.masque.Warp - 2, // [2:2] is the sub-list for method output_type - 2, // [2:2] is the sub-list for method input_type - 2, // [2:2] is the sub-list for extension type_name - 2, // [2:2] is the sub-list for extension extendee - 0, // [0:2] is the sub-list for field type_name + 3, // 2: xray.transport.internet.masque.Config.xmux:type_name -> xray.transport.internet.splithttp.XmuxConfig + 3, // [3:3] is the sub-list for method output_type + 3, // [3:3] is the sub-list for method input_type + 3, // [3:3] is the sub-list for extension type_name + 3, // [3:3] is the sub-list for extension extendee + 0, // [0:3] is the sub-list for field type_name } func init() { file_transport_internet_masque_config_proto_init() } diff --git a/transport/internet/masque/config.proto b/transport/internet/masque/config.proto index 394c0895c..4bb482e3e 100644 --- a/transport/internet/masque/config.proto +++ b/transport/internet/masque/config.proto @@ -6,11 +6,14 @@ option go_package = "github.com/xtls/xray-core/transport/internet/masque"; option java_package = "com.xray.transport.internet.masque"; option java_multiple_files = true; +import "transport/internet/splithttp/config.proto"; + message Config { string host = 1; string path = 2; map headers = 3; Warp warp = 4; + xray.transport.internet.splithttp.XmuxConfig xmux = 5; } message Warp { diff --git a/transport/internet/masque/dialer.go b/transport/internet/masque/dialer.go index cbacdaced..3a576caa3 100644 --- a/transport/internet/masque/dialer.go +++ b/transport/internet/masque/dialer.go @@ -72,7 +72,11 @@ func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.Me quicConfig.MaxIdleTimeout = 30 * time.Second } if quicParams.KeepAlivePeriod == 0 { - quicConfig.KeepAlivePeriod = net.QuicgoH3KeepAlivePeriod + if keepAlivePeriod := config.keepAlivePeriod(); keepAlivePeriod == 0 { + quicConfig.KeepAlivePeriod = net.QuicgoH3KeepAlivePeriod + } else if keepAlivePeriod > 0 { + quicConfig.KeepAlivePeriod = keepAlivePeriod + } } var pktConn net.PacketConn @@ -179,7 +183,11 @@ func dialHTTP2(ctx context.Context, dest net.Destination, streamSettings *intern return nil, errors.New("the server negotiated ", strconv.Quote(protocol), " instead of h2") } - cc, err := newHTTP2ClientConn(conn) + keepAlivePeriod := config.keepAlivePeriod() + if keepAlivePeriod == 0 { + keepAlivePeriod = net.ChromeH2KeepAlivePeriod + } + cc, err := newHTTP2ClientConn(conn, keepAlivePeriod) if err != nil { conn.Close() return nil, err @@ -212,22 +220,7 @@ func establish(ctx context.Context, client tunnelClient, hconn httpConn, abort f for k, v := range config.Headers { header.Set(k, v) } - switch header.Get("User-Agent") { - case "": - header["User-Agent"] = nil - case "chrome": - header.Set("User-Agent", utils.ChromeUA) - case "firefox": - header.Set("User-Agent", utils.FirefoxUA) - case "safari": - header.Set("User-Agent", utils.SafariUA) - case "edge": - header.Set("User-Agent", utils.MSEdgeUA) - case "curl": - header.Set("User-Agent", utils.CurlUA) - case "golang": - header.Del("User-Agent") - } + setUserAgent(header) ipConn, _, err := client.Dial(req) if err != nil { @@ -327,6 +320,28 @@ func authority(config *Config, serverName string, port net.Port) string { return net.JoinHostPort(host, port.String()) } +func setUserAgent(header http.Header) { + if _, ok := header["User-Agent"]; !ok { + header.Set("User-Agent", "chrome") + } + switch header.Get("User-Agent") { + case "": + header["User-Agent"] = nil + case "chrome": + header.Set("User-Agent", utils.ChromeUA) + case "firefox": + header.Set("User-Agent", utils.FirefoxUA) + case "safari": + header.Set("User-Agent", utils.SafariUA) + case "edge": + header.Set("User-Agent", utils.MSEdgeUA) + case "curl": + header.Set("User-Agent", utils.CurlUA) + case "golang": + header.Del("User-Agent") + } +} + func init() { common.Must(internet.RegisterTransportDialer(protocolName, Dial)) } diff --git a/transport/internet/masque/dialer_test.go b/transport/internet/masque/dialer_test.go index ddeef00b2..ae81ce21b 100644 --- a/transport/internet/masque/dialer_test.go +++ b/transport/internet/masque/dialer_test.go @@ -1,11 +1,13 @@ package masque import ( + "net/http" "net/netip" "slices" "testing" "github.com/xtls/xray-core/common/net" + "github.com/xtls/xray-core/common/utils" "github.com/xtls/xray-core/transport/internet/masque/connectip" "github.com/xtls/xray-core/transport/internet/tls" ) @@ -77,3 +79,26 @@ func TestLocalAddrs(t *testing.T) { } } } + +func TestSetUserAgent(t *testing.T) { + for _, tc := range []struct { + name string + header http.Header + want []string + }{ + {"default", http.Header{}, []string{utils.ChromeUA}}, + {"empty", http.Header{"User-Agent": {""}}, nil}, + {"preset", http.Header{"User-Agent": {"firefox"}}, []string{utils.FirefoxUA}}, + {"custom", http.Header{"User-Agent": {"custom/1.0"}}, []string{"custom/1.0"}}, + } { + setUserAgent(tc.header) + if got := tc.header["User-Agent"]; !slices.Equal(got, tc.want) { + t.Errorf("%s: got %q, want %q", tc.name, got, tc.want) + } + } + header := http.Header{"User-Agent": {"golang"}} + setUserAgent(header) + if _, ok := header["User-Agent"]; ok { + t.Errorf("golang: the header should be removed, got %q", header["User-Agent"]) + } +} diff --git a/transport/internet/masque/http2.go b/transport/internet/masque/http2.go index a7cb4e4d9..57d7fef43 100644 --- a/transport/internet/masque/http2.go +++ b/transport/internet/masque/http2.go @@ -32,6 +32,7 @@ const ( http2WindowUpdateSize = 1 << 20 http2KeepAlivePeriod = 10 * time.Second http2IdleTimeout = 30 * time.Second + http2PingTimeout = 15 * time.Second http2DefaultUserAgent = "Go-http-client/2.0" ) @@ -79,7 +80,7 @@ type http2ClientConn struct { recvUnacked int64 } -func newHTTP2ClientConn(conn net.Conn) (*http2ClientConn, error) { +func newHTTP2ClientConn(conn net.Conn, keepAlivePeriod time.Duration) (*http2ClientConn, error) { c := &http2ClientConn{ conn: conn, bw: bufio.NewWriter(conn), @@ -119,7 +120,9 @@ func newHTTP2ClientConn(conn net.Conn) (*http2ClientConn, error) { return nil, err } go c.readLoop() - go c.keepAlive() + if keepAlivePeriod > 0 { + go c.keepAlive(keepAlivePeriod) + } return c, nil } @@ -363,21 +366,29 @@ func (c *http2ClientConn) abortStream(err error, reset bool) { } } -func (c *http2ClientConn) keepAlive() { - ticker := time.NewTicker(http2KeepAlivePeriod) +func (c *http2ClientConn) keepAlive(period time.Duration) { + ticker := time.NewTicker(max(min(period, http2PingTimeout)/3, time.Second)) defer ticker.Stop() + var pingAt time.Time for { select { case <-c.done: return case <-ticker.C: } - idle := time.Since(time.Unix(0, c.lastFrame.Load())) - if idle >= http2IdleTimeout { - c.fail(errHTTP2IdleTimeout) - return + lastFrame := time.Unix(0, c.lastFrame.Load()) + if !pingAt.IsZero() { + if lastFrame.Before(pingAt) { + if time.Since(pingAt) >= http2PingTimeout { + c.fail(errHTTP2IdleTimeout) + return + } + continue + } + pingAt = time.Time{} } - if idle >= http2KeepAlivePeriod { + if time.Since(lastFrame) >= period { + pingAt = time.Now() go c.write(func(fr *http2.Framer) error { return fr.WritePing(false, [8]byte{}) }) diff --git a/transport/internet/masque/http2_server_test.go b/transport/internet/masque/http2_server_test.go index 510e37438..cde9d6e4a 100644 --- a/transport/internet/masque/http2_server_test.go +++ b/transport/internet/masque/http2_server_test.go @@ -127,7 +127,7 @@ func TestHTTP2ServerRoundTrip(t *testing.T) { _, err := io.Copy(w, r.Body) assert.NoError(t, err) }) - cc, err := newHTTP2ClientConn(serveHTTP2Pipe(t, handler)) + cc, err := newHTTP2ClientConn(serveHTTP2Pipe(t, handler), http2KeepAlivePeriod) require.NoError(t, err) defer cc.Close() diff --git a/transport/internet/masque/http2_test.go b/transport/internet/masque/http2_test.go index 5df664ab4..7d1a97513 100644 --- a/transport/internet/masque/http2_test.go +++ b/transport/internet/masque/http2_test.go @@ -1,13 +1,16 @@ package masque import ( + "bufio" "bytes" "context" "io" "net" "net/http" "strings" + "sync" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/require" @@ -52,7 +55,7 @@ func newHTTP2Peer(t *testing.T, settings ...http2.Setting) (*http2ClientConn, *h var cc *http2ClientConn go func() { var err error - cc, err = newHTTP2ClientConn(client) + cc, err = newHTTP2ClientConn(client, http2KeepAlivePeriod) ccErr <- err }() preface := make([]byte, len(http2.ClientPreface)) @@ -423,3 +426,62 @@ func TestHTTP2ClientAnswersPings(t *testing.T) { require.True(t, f.(*http2.PingFrame).IsAck()) require.Equal(t, data, f.(*http2.PingFrame).Data) } + +func TestHTTP2ClientKeepAlive(t *testing.T) { + for _, answer := range []bool{false, true} { + synctest.Test(t, func(t *testing.T) { + client, server := net.Pipe() + defer server.Close() + var mu sync.Mutex + var pings []time.Duration + start := time.Now() + go func() { + br := bufio.NewReader(server) + if _, err := io.ReadFull(br, make([]byte, len(http2.ClientPreface))); err != nil { + return + } + fr := http2.NewFramer(server, br) + time.Sleep(2 * time.Second) + if fr.WriteSettings() != nil { + return + } + for { + f, err := fr.ReadFrame() + if err != nil { + return + } + if p, ok := f.(*http2.PingFrame); ok && !p.IsAck() { + mu.Lock() + pings = append(pings, time.Since(start)) + mu.Unlock() + if answer && fr.WritePing(true, p.Data) != nil { + return + } + } + } + }() + cc, err := newHTTP2ClientConn(client, 60*time.Second) + require.NoError(t, err) + if answer { + time.Sleep(200 * time.Second) + select { + case <-cc.done: + t.Fatal("the connection closed although every PING was answered") + default: + } + cc.conn.Close() + <-cc.done + mu.Lock() + defer mu.Unlock() + require.GreaterOrEqual(t, len(pings), 2) + return + } + <-cc.done + closed := time.Since(start) + mu.Lock() + defer mu.Unlock() + require.Len(t, pings, 1, "the connection must send a PING before closing") + require.GreaterOrEqual(t, closed-pings[0], http2PingTimeout) + }) + } +} diff --git a/transport/internet/splithttp/mux.go b/transport/internet/splithttp/mux.go index 4897ace45..1945224e6 100644 --- a/transport/internet/splithttp/mux.go +++ b/transport/internet/splithttp/mux.go @@ -79,6 +79,14 @@ func (m *XmuxManager) newXmuxClient() *XmuxClient { } func (m *XmuxManager) GetXmuxClient(ctx context.Context) *XmuxClient { // when locking + return m.getXmuxClient(ctx, true) +} + +func (m *XmuxManager) GetExistingXmuxClient(ctx context.Context) *XmuxClient { + return m.getXmuxClient(ctx, false) +} + +func (m *XmuxManager) getXmuxClient(ctx context.Context, create bool) *XmuxClient { for i := 0; i < len(m.xmuxClients); { xmuxClient := m.xmuxClients[i] if xmuxClient.XmuxConn.IsClosed() || @@ -99,11 +107,14 @@ func (m *XmuxManager) GetXmuxClient(ctx context.Context) *XmuxClient { // when l } if len(m.xmuxClients) == 0 { + if !create { + return nil + } errors.LogDebug(ctx, "XMUX: creating xmuxClient because xmuxClients is empty") return m.newXmuxClient() } - if m.connections > 0 && len(m.xmuxClients) < int(m.connections) { + if create && m.connections > 0 && len(m.xmuxClients) < int(m.connections) { errors.LogDebug(ctx, "XMUX: creating xmuxClient because maxConnections was not hit, xmuxClients = ", len(m.xmuxClients)) return m.newXmuxClient() } @@ -120,6 +131,9 @@ func (m *XmuxManager) GetXmuxClient(ctx context.Context) *XmuxClient { // when l } if len(xmuxClients) == 0 { + if !create { + return nil + } errors.LogDebug(ctx, "XMUX: creating xmuxClient because maxConcurrency was hit, xmuxClients = ", len(m.xmuxClients)) return m.newXmuxClient() } diff --git a/transport/internet/splithttp/mux_test.go b/transport/internet/splithttp/mux_test.go index 2f5b3520e..6ae22c7f0 100644 --- a/transport/internet/splithttp/mux_test.go +++ b/transport/internet/splithttp/mux_test.go @@ -90,3 +90,23 @@ func TestDefault(t *testing.T) { t.Error("did not get 1 distinct clients, got ", len(xmuxClients)) } } + +func TestGetExistingXmuxClient(t *testing.T) { + xmuxConfig := XmuxConfig{ + MaxConnections: &RangeConfig{From: 2, To: 2}, + } + + xmuxManager := NewXmuxManager(xmuxConfig, func() XmuxConn { + return &fakeRoundTripper{} + }) + + if xmuxManager.GetExistingXmuxClient(context.Background()) != nil { + t.Error("got a client without creating one") + } + xmuxClient := xmuxManager.GetXmuxClient(context.Background()) + for i := 0; i < 4; i++ { + if xmuxManager.GetExistingXmuxClient(context.Background()) != xmuxClient { + t.Error("did not reuse the existing client") + } + } +}