XDRIVE transport: Support XMUX ("maxConnections": 4 on Google Drive by default) and h3

infra/conf/transport_internet.go

@@ -90,6 +90,7 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
 		config.ProtocolName = protocol
 	}
 
+	var xdriveH3 bool
 	switch strings.ToLower(c.Security) {
 	case "", "none":
 	case "tls":
@@ -104,6 +105,7 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
 		tm := serial.ToTypedMessage(ts)
 		config.SecuritySettings = append(config.SecuritySettings, tm)
 		config.SecurityType = tm.Type
+		xdriveH3 = config.ProtocolName == "xdrive" && tlsSettings.ALPN != nil && slices.Equal(*tlsSettings.ALPN, StringList{"h3"})
 		if config.ProtocolName == "xdrive" && tlsSettings.ALPN != nil && slices.Equal(*tlsSettings.ALPN, StringList{"http/1.1"}) && tlsSettings.ECHConfigList == "" {
 			// a fingerprint puts its own ALPN into the hello, except Go TLS and the
 			// randomized ones, which take it from the config or send none
@@ -219,6 +221,7 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
 		})
 	}
 	if c.XDRIVESettings != nil {
+		c.XDRIVESettings.h3 = xdriveH3
 		xs, err := c.XDRIVESettings.Build()
 		if err != nil {
 			return nil, errors.New("Failed to build XDRIVE config.").Base(err)
@@ -260,5 +263,20 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
 		}
 	}
 
+	if xdriveH3 && c.FinalMask != nil {
+		if c.XDRIVESettings != nil && c.XDRIVESettings.Service == "google-drive" {
+			// noise only sends datagrams of its own ahead of the QUIC ones, the
+			// other masks change those or need a peer that undoes them
+			for _, mask := range c.FinalMask.Udp {
+				if strings.ToLower(mask.Type) != "noise" {
+					return nil, errors.New(`XDRIVE: over h3 to Google Drive the only "udp" mask that works is "noise", not "`, mask.Type, `"`)
+				}
+			}
+		}
+		if config.QuicParams != nil && (config.QuicParams.Congestion == "brutal" || config.QuicParams.Congestion == "force-brutal") {
+			return nil, errors.New(`XDRIVE: brutal is for Hysteria servers, over h3 "congestion" is "bbr" or "reno", not "`, config.QuicParams.Congestion, `"`)
+		}
+	}
+
 	return config, nil
 }

infra/conf/transport_method.go

@@ -997,6 +997,10 @@ type XDriveConfig struct {
 	EagerWindowMs     uint32          `json:"eagerWindowMs"`
 	HoleTimeoutMs     uint32          `json:"holeTimeoutMs"`
 	Template          json.RawMessage `json:"template"`
+	Xmux              XmuxConfig      `json:"xmux"`
+
+	// set by StreamConfig when "alpn" is ["h3"]
+	h3 bool
 }
 
 // Build implements Buildable.
@@ -1018,6 +1022,21 @@ func (c *XDriveConfig) Build() (proto.Message, error) {
 	default:
 		return nil, errors.New(`unknown "service": "`, c.Service, `", use "google-drive", "template" or "local"`)
 	}
+	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
+		if c.Service == "google-drive" && !c.h3 {
+			// Google gives an h2 connection 1 MiB of window, which a few sessions on
+			// a long rtt would share, an h3 connection has no such limit
+			c.Xmux.MaxConnections.From = 4
+			c.Xmux.MaxConnections.To = 4
+		}
+		c.Xmux.HMaxReusableSecs.From = 1800
+		c.Xmux.HMaxReusableSecs.To = 3000
+	}
 	config := &xdrive.Config{
 		RemoteFolder:      c.RemoteFolder,
 		Service:           c.Service,
@@ -1031,6 +1050,14 @@ func (c *XDriveConfig) Build() (proto.Message, error) {
 		EagerWindowMs:     c.EagerWindowMs,
 		HoleTimeoutMs:     c.HoleTimeoutMs,
 		Template:          string(c.Template),
+		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,
+		},
 	}
 	return config, nil
 }

infra/conf/transport_test.go

@@ -9,6 +9,7 @@ import (
 	. "github.com/xtls/xray-core/infra/conf"
 	"github.com/xtls/xray-core/transport/internet"
 	finalmaskcustom "github.com/xtls/xray-core/transport/internet/finalmask/header/custom"
+	"github.com/xtls/xray-core/transport/internet/splithttp"
 	"github.com/xtls/xray-core/transport/internet/xdrive"
 	"google.golang.org/protobuf/proto"
 )
@@ -424,6 +425,81 @@ func TestXDriveHTTP1ALPNWarning(t *testing.T) {
 	}
 }
 
+// over h3 Google Drive takes only the noise mask, and no service takes brutal
+func TestXDriveH3Checks(t *testing.T) {
+	const (
+		drive       = `"method": "xdrive", "xdriveSettings": {"remoteFolder": "f", "service": "google-drive", "secrets": ["a", "b", "c"]}`
+		local       = `"method": "xdrive", "xdriveSettings": {"remoteFolder": "/tmp/xdrive", "service": "local"}`
+		noise       = `{"type": "Noise", "settings": {"noise": [{"rand": "50-100"}]}}`
+		salamander  = `{"type": "salamander", "settings": {"password": "p"}}`
+		forceBrutal = `{"quicParams": {"congestion": "force-brutal", "brutalUp": "10 mbps"}}`
+	)
+	for _, c := range []struct {
+		alpn, transport, finalmask string
+		ok                         bool
+	}{
+		{`["h3"]`, drive, `{"udp": [` + noise + `]}`, true},
+		{`["h3"]`, drive, `{"udp": [` + salamander + `]}`, false},
+		{`["h3"]`, drive, `{"udp": [` + noise + `, ` + salamander + `]}`, false},
+		{`["h3"]`, local, `{"udp": [` + salamander + `]}`, true},
+		{`["h3"]`, `"method": "xdrive"`, `{"udp": [` + salamander + `]}`, true},
+		{`["h2"]`, drive, `{"udp": [` + salamander + `]}`, true},
+		{`null`, drive, `{"udp": [` + salamander + `]}`, true},
+		{`["h3"]`, drive, `null`, true},
+		{`["h3"]`, drive, `{"quicParams": {"congestion": "bbr"}}`, true},
+		{`["h3"]`, drive, `{"quicParams": {"congestion": "brutal"}}`, false},
+		{`["h3"]`, local, forceBrutal, false},
+		{`["h3", "h2"]`, drive, `{"quicParams": {"congestion": "brutal"}}`, true},
+		{`["h3"]`, `"method": "xhttp"`, forceBrutal, true},
+	} {
+		config := new(StreamConfig)
+		if err := json.Unmarshal([]byte(`{
+			`+c.transport+`,
+			"security": "tls",
+			"tlsSettings": {"alpn": `+c.alpn+`},
+			"finalmask": `+c.finalmask+`
+		}`), config); err != nil {
+			t.Fatalf("Unmarshal: %v", err)
+		}
+		if _, err := config.Build(); (err == nil) != c.ok {
+			t.Errorf("alpn %s, %s, finalmask %s: %v", c.alpn, c.transport, c.finalmask, err)
+		}
+	}
+}
+
+// one h3 connection is not held back by a window like h2, so h3 takes one by default
+func TestXDriveH3Connections(t *testing.T) {
+	for _, c := range []struct {
+		alpn string
+		want int32
+	}{
+		{`["h3"]`, 1},
+		{`["h2"]`, 4},
+		{`null`, 4},
+	} {
+		config := new(StreamConfig)
+		if err := json.Unmarshal([]byte(`{
+			"method": "xdrive",
+			"xdriveSettings": {"remoteFolder": "f", "service": "google-drive", "secrets": ["a", "b", "c"]},
+			"security": "tls",
+			"tlsSettings": {"alpn": `+c.alpn+`}
+		}`), config); err != nil {
+			t.Fatalf("Unmarshal: %v", err)
+		}
+		built, err := config.Build()
+		if err != nil {
+			t.Fatalf("alpn %s: %v", c.alpn, err)
+		}
+		xs, err := built.TransportSettings[0].Settings.GetInstance()
+		if err != nil {
+			t.Fatalf("alpn %s: %v", c.alpn, err)
+		}
+		if got := xs.(*xdrive.Config).Xmux.MaxConnections; got.From != c.want || got.To != c.want {
+			t.Errorf("alpn %s: maxConnections is %d-%d, want %d", c.alpn, got.From, got.To, c.want)
+		}
+	}
+}
+
 func TestXDriveTemplateNeedsTemplate(t *testing.T) {
 	config := new(XDriveConfig)
 	if err := json.Unmarshal([]byte(`{"remoteFolder": "f", "service": "template"}`), config); err != nil {
@@ -513,3 +589,50 @@ func TestXDriveInboundConfig(t *testing.T) {
 		}
 	}
 }
+
+func TestXDriveXmux(t *testing.T) {
+	defaults := func(connections int32) *splithttp.XmuxConfig {
+		return &splithttp.XmuxConfig{
+			MaxConcurrency:   &splithttp.RangeConfig{},
+			MaxConnections:   &splithttp.RangeConfig{From: connections, To: connections},
+			CMaxReuseTimes:   &splithttp.RangeConfig{},
+			HMaxRequestTimes: &splithttp.RangeConfig{},
+			HMaxReusableSecs: &splithttp.RangeConfig{From: 1800, To: 3000},
+		}
+	}
+	for _, c := range []struct {
+		input string
+		want  *splithttp.XmuxConfig
+	}{
+		{`{"service": "google-drive", "secrets": ["a", "b", "c"]}`, defaults(4)},
+		{`{"service": "template", "template": {}}`, defaults(1)},
+		{`{"service": "google-drive", "secrets": ["a", "b", "c"], "xmux": {"maxConcurrency": "1-2", "hMaxRequestTimes": 100, "hKeepAlivePeriod": -1}}`, &splithttp.XmuxConfig{
+			MaxConcurrency:   &splithttp.RangeConfig{From: 1, To: 2},
+			MaxConnections:   &splithttp.RangeConfig{},
+			CMaxReuseTimes:   &splithttp.RangeConfig{},
+			HMaxRequestTimes: &splithttp.RangeConfig{From: 100, To: 100},
+			HMaxReusableSecs: &splithttp.RangeConfig{},
+			HKeepAlivePeriod: -1,
+		}},
+	} {
+		config := new(XDriveConfig)
+		if err := json.Unmarshal([]byte(c.input), config); err != nil {
+			t.Fatalf("%s: Unmarshal: %v", c.input, err)
+		}
+		built, err := config.Build()
+		if err != nil {
+			t.Fatalf("%s: Build: %v", c.input, err)
+		}
+		if got := built.(*xdrive.Config).Xmux; !proto.Equal(got, c.want) {
+			t.Errorf("%s: xmux is %v, want %v", c.input, got, c.want)
+		}
+	}
+
+	config := new(XDriveConfig)
+	if err := json.Unmarshal([]byte(`{"service": "template", "template": {}, "xmux": {"maxConnections": 2, "maxConcurrency": 8}}`), config); err != nil {
+		t.Fatalf("Unmarshal: %v", err)
+	}
+	if _, err := config.Build(); err == nil {
+		t.Error("Build accepted maxConnections with maxConcurrency")
+	}
+}