diff --git a/infra/conf/transport_internet.go b/infra/conf/transport_internet.go index e016d7948bfb..efee0363ad43 100644 --- a/infra/conf/transport_internet.go +++ b/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 } diff --git a/infra/conf/transport_method.go b/infra/conf/transport_method.go index a7d79dcd76c0..f10a2cf4082c 100644 --- a/infra/conf/transport_method.go +++ b/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 } diff --git a/infra/conf/transport_test.go b/infra/conf/transport_test.go index 957186cc9b70..dfb66c42693e 100644 --- a/infra/conf/transport_test.go +++ b/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") + } +} diff --git a/transport/internet/xdrive/client.go b/transport/internet/xdrive/client.go index 00a3f16905d1..d1ccf3faa515 100644 --- a/transport/internet/xdrive/client.go +++ b/transport/internet/xdrive/client.go @@ -2,14 +2,25 @@ package xdrive import ( "context" + gotls "crypto/tls" "io" "net/http" + "reflect" + "runtime" + "slices" "sync" "time" + "github.com/apernet/quic-go" + "github.com/apernet/quic-go/http3" "github.com/xtls/xray-core/common/errors" "github.com/xtls/xray-core/common/net" + "github.com/xtls/xray-core/common/net/cnc" + "github.com/xtls/xray-core/common/utils" "github.com/xtls/xray-core/transport/internet" + "github.com/xtls/xray-core/transport/internet/hysteria/congestion" + "github.com/xtls/xray-core/transport/internet/hysteria/congestion/bbr" + "github.com/xtls/xray-core/transport/internet/splithttp" "github.com/xtls/xray-core/transport/internet/tls" ) @@ -39,6 +50,10 @@ func (t *serviceTransport) RoundTrip(r *http.Request) (*http.Response, error) { cancel(nil) } r = r.WithContext(ctx) + // the headers of chrome as with XHTTP and DoH, net/http and http3 would send + // their own user agent after a hello of chrome + r.Header = r.Header.Clone() + utils.TryDefaultHeadersWith(r.Header, "fetch") if r.Body != nil && r.Body != http.NoBody { r.Body = &progressBody{ReadCloser: r.Body, alive: alive} } @@ -104,25 +119,40 @@ func (b *progressBody) Close() error { } // newServiceClient returns a client whose requests end after stall without -// progress, see serviceTransport.RoundTrip -func newServiceClient(streamSettings *internet.MemoryStreamConfig, stall time.Duration, maxConns int) *http.Client { +// progress, see serviceTransport.RoundTrip, and whose connections close when +// connCtx is done. Clients that share h1Hosts share which hosts answer over h1 +func newServiceClient(connCtx context.Context, h1Hosts *sync.Map, streamSettings *internet.MemoryStreamConfig, stall time.Duration, maxConns int) *http.Client { var ( tlsConfig *tls.Config sockopt *internet.SocketConfig fronting *net.Destination + keepAlive = net.ChromeH2KeepAlivePeriod + // h3 keeps alive only when asked to, see below + h3KeepAlive time.Duration ) if streamSettings != nil { tlsConfig = tls.ConfigFromStreamSettings(streamSettings) sockopt = streamSettings.SocketSettings fronting = streamSettings.Destination + if config, ok := streamSettings.ProtocolSettings.(*Config); ok && config.Xmux != nil { + switch period := config.Xmux.HKeepAlivePeriod; { + case period > 0: + keepAlive = time.Duration(period) * time.Second + h3KeepAlive = keepAlive + case period < 0: + keepAlive = 0 + } + } } if tlsConfig == nil { tlsConfig = &tls.Config{} } fingerprint := tls.GetFingerprint(tlsConfig.Fingerprint) - dial := func(ctx context.Context, addr string) (net.Conn, net.Destination, error) { - host, err := net.ParseDestination("tcp:" + addr) + // fronting takes only the address and port, the network stays that of the + // dial, which is UDP for h3 though the destination is built as TCP + dial := func(ctx context.Context, network, addr string) (net.Conn, net.Destination, error) { + host, err := net.ParseDestination(network + ":" + addr) if err != nil { return nil, host, errors.New("bad address: ", addr).Base(err) } @@ -136,25 +166,36 @@ func newServiceClient(streamSettings *internet.MemoryStreamConfig, stall time.Du } var conn net.Conn - if streamSettings.FinalMask != nil { - conn, err = streamSettings.FinalMask.DialTCP(ctx, target) - } else { + switch { + case streamSettings.FinalMask == nil: conn, err = internet.DialSystem(ctx, target, sockopt) + case target.Network == net.Network_UDP: + conn, err = streamSettings.FinalMask.DialUDP(ctx, target) + default: + conn, err = streamSettings.FinalMask.DialTCP(ctx, target) } if err != nil { return nil, host, errors.New("failed to dial to dest").Base(err) } - return conn, host, nil + // h3 closes its QUIC connection instead, as the server never hears of a UDP + // socket that closes + if network == "udp" { + return conn, host, nil + } + // CloseIdleConnections leaves an h2 connection open while a stream just + // canceled has not left it yet + stop := context.AfterFunc(connCtx, func() { conn.Close() }) + return &ctxConn{Conn: conn, stop: stop}, host, nil } dialPlain := func(ctx context.Context, network, addr string) (net.Conn, error) { - conn, _, err := dial(ctx, addr) + conn, _, err := dial(ctx, "tcp", addr) return conn, err } // http.Transport picks h2 or h1 by the ALPN in the ConnectionState of the conn dialTLS := func(ctx context.Context, network, addr string) (net.Conn, error) { - conn, host, err := dial(ctx, addr) + conn, host, err := dial(ctx, "tcp", addr) if err != nil { return nil, err } @@ -185,7 +226,86 @@ func newServiceClient(streamSettings *internet.MemoryStreamConfig, stall time.Du MaxIdleConns: maxConns, MaxIdleConnsPerHost: maxConns, MaxConnsPerHost: connsPerHost, - HTTP2: &http.HTTP2Config{SendPingTimeout: net.ChromeH2KeepAlivePeriod}, + HTTP2: &http.HTTP2Config{SendPingTimeout: keepAlive}, + } + } + + var https http.RoundTripper = &secureTransport{h2: secure(1), h1: secure(maxConns), h1Hosts: h1Hosts} + // no server offers h3 by ALPN over TCP, so it is used only when asked for alone + if slices.Equal(tlsConfig.NextProtocol, []string{"h3"}) { + quicParams := streamSettings.QuicParams + if quicParams == nil { + quicParams = &internet.QuicParams{BbrProfile: string(bbr.ProfileStandard)} + } + // the defaults are those of quic-go, with no keepalive unless asked for, as + // in DoH: http3 never closes a connection that is idle or that it dropped + // after an error, so only the idle timeout or the end of connCtx ends it + quicConfig := &quic.Config{ + InitialStreamReceiveWindow: quicParams.InitStreamReceiveWindow, + MaxStreamReceiveWindow: quicParams.MaxStreamReceiveWindow, + InitialConnectionReceiveWindow: quicParams.InitConnReceiveWindow, + MaxConnectionReceiveWindow: quicParams.MaxConnReceiveWindow, + MaxIdleTimeout: time.Duration(quicParams.MaxIdleTimeout) * time.Second, + KeepAlivePeriod: time.Duration(quicParams.KeepAlivePeriod) * time.Second, + MaxIncomingStreams: quicParams.MaxIncomingStreams, + DisablePathMTUDiscovery: quicParams.DisablePathMtuDiscovery || (runtime.GOOS != "linux" && runtime.GOOS != "windows" && runtime.GOOS != "darwin"), + ChromeParrot: !quicParams.DisableChromeParrot, + } + if quicConfig.KeepAlivePeriod == 0 { + quicConfig.KeepAlivePeriod = h3KeepAlive + } + + https = &http3.Transport{ + QUICConfig: quicConfig, + Dial: func(ctx context.Context, addr string, _ *gotls.Config, cfg *quic.Config) (*quic.Conn, error) { + switch quicParams.Congestion { + case "", "bbr", "reno": + default: + // brutal is for Hysteria servers, not services, and a panic here, in + // a goroutine of http3, would end the process + return nil, errors.New("unsupported congestion control over h3: ", quicParams.Congestion) + } + + conn, host, err := dial(ctx, "udp", addr) + if err != nil { + return nil, err + } + var pktConn net.PacketConn + var udpAddr net.Addr + switch c := conn.(type) { + case *net.PacketConnWrapper: + pktConn = c.PacketConn + udpAddr = c.RemoteAddr() + case *cnc.Connection: + pktConn = &internet.FakePacketConn{Conn: c} + udpAddr = &net.UDPAddr{IP: []byte{0, 0, 0, 0}} + default: + conn.Close() + return nil, errors.New("failed to dial to dest: unexpected ", reflect.TypeOf(c)) + } + + // built for each dial with the host as SNI, as for h2: http3 sets the + // SNI on a copy, which a pinned CA and the ECH lookup do not see + gotlsConfig := tlsConfig.GetTLSConfig(tls.WithDestination(host)) + // it is for servers, and ChromeParrot refuses a config that has it + gotlsConfig.GetCertificate = nil + tr := &quic.Transport{Conn: pktConn, DisableGSO: quicParams.DisableGSO} + if !quicParams.DisableChromeParrot { + tr.ConnectionIDGenerator = quic.ZeroLengthConnectionIDGenerator{} + } + qconn, err := tr.Dial(ctx, udpAddr, gotlsConfig, cfg) + if err != nil { + tr.Close() + pktConn.Close() + return nil, err + } + stop := context.AfterFunc(connCtx, func() { qconn.CloseWithError(quic.ApplicationErrorCode(http3.ErrCodeNoError), "") }) + context.AfterFunc(qconn.Context(), func() { stop(); tr.Close(); pktConn.Close() }) + if quicParams.Congestion != "reno" { + congestion.UseBBR(qconn, bbr.Profile(quicParams.BbrProfile)) + } + return qconn, nil + }, } } @@ -198,18 +318,30 @@ func newServiceClient(streamSettings *internet.MemoryStreamConfig, stall time.Du MaxIdleConnsPerHost: maxConns, MaxConnsPerHost: maxConns, }, - secure: &secureTransport{h2: secure(1), h1: secure(maxConns)}, + secure: https, stall: stall, }, } } +// ctxConn is a connection that connCtx no longer closes once it is closed, +// so that one that ends on its own leaves nothing behind in connCtx +type ctxConn struct { + net.Conn + stop func() bool +} + +func (c *ctxConn) Close() error { + c.stop() + return c.Conn.Close() +} + // secureTransport keeps a host to one connection, all that h2 needs, as without // a limit parallel requests would each dial their own. A host that answers over // h1 moves to a transport with more, since h1 carries one request at a time type secureTransport struct { h2, h1 http.RoundTripper - h1Hosts sync.Map + h1Hosts *sync.Map } func (t *secureTransport) RoundTrip(r *http.Request) (*http.Response, error) { @@ -222,3 +354,129 @@ func (t *secureTransport) RoundTrip(r *http.Request) (*http.Response, error) { } return resp, err } + +var errSessionClosed = errors.New("the session is closed") + +// connPool hands the sessions of a storage the connections of XMUX, each a +// client of its own made by newClient +type connPool struct { + mu sync.Mutex + xmux *splithttp.XmuxManager +} + +func newConnPool(config *splithttp.XmuxConfig, newClient func(connCtx context.Context) *http.Client) *connPool { + var xmuxConfig splithttp.XmuxConfig + if config != nil { + xmuxConfig = *config + } + return &connPool{xmux: splithttp.NewXmuxManager(xmuxConfig, func() splithttp.XmuxConn { + ctx, cancel := context.WithCancel(context.Background()) + return &poolConn{next: newClient(ctx).Transport, cancel: cancel} + })} +} + +// poolConn is a connection of XMUX, which closes once XMUX drops it and the +// requests on it end +type poolConn struct { + next http.RoundTripper + cancel context.CancelFunc + + mu sync.Mutex + inflight int + closing bool +} + +func (c *poolConn) IsClosed() bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.closing +} + +func (c *poolConn) Close() error { + c.mu.Lock() + defer c.mu.Unlock() + c.closing = true + if c.inflight == 0 { + c.cancel() + } + return nil +} + +func (c *poolConn) done() { + c.mu.Lock() + defer c.mu.Unlock() + c.inflight-- + if c.closing && c.inflight == 0 { + c.cancel() + } +} + +// roundTrip sends a request counted in inflight by the caller +func (c *poolConn) roundTrip(r *http.Request) (*http.Response, error) { + resp, err := c.next.RoundTrip(r) + if err != nil { + c.done() + return nil, err + } + resp.Body = &doneBody{ReadCloser: resp.Body, done: sync.OnceFunc(c.done)} + return resp, nil +} + +type doneBody struct { + io.ReadCloser + done func() +} + +func (b *doneBody) Close() error { + err := b.ReadCloser.Close() + b.done() + return err +} + +// sessionTransport sends the requests of one session over a connection of +// XMUX, and over another once that one has no requests or time left +type sessionTransport struct { + pool *connPool + + mu sync.Mutex + client *splithttp.XmuxClient + closed bool +} + +func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) { + t.mu.Lock() + if t.closed { + t.mu.Unlock() + return nil, errSessionClosed + } + client := t.client + if client == nil || client.LeftRequests.Add(-1) < 0 || !client.UnreusableAt.IsZero() && time.Now().After(client.UnreusableAt) { + t.pool.mu.Lock() + t.client = t.pool.xmux.GetXmuxClient(r.Context()) + t.pool.mu.Unlock() + t.client.AddRunning() + t.client.LeftRequests.Add(-1) + if client != nil { + client.DoneRunning() + } + client = t.client + } + // counted under mu, or another request of the session that moves on could + // close the connection before this one is sent + conn := client.XmuxConn.(*poolConn) + conn.mu.Lock() + conn.inflight++ + conn.mu.Unlock() + t.mu.Unlock() + return conn.roundTrip(r) +} + +func (t *sessionTransport) Close() { + t.mu.Lock() + defer t.mu.Unlock() + t.closed = true + if t.client != nil { + t.client.DoneRunning() + t.client = nil + } +} diff --git a/transport/internet/xdrive/client_test.go b/transport/internet/xdrive/client_test.go index 62187de0148d..58b3f25112d1 100644 --- a/transport/internet/xdrive/client_test.go +++ b/transport/internet/xdrive/client_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" gotls "crypto/tls" + "crypto/x509" "encoding/base64" "errors" "io" @@ -11,13 +12,22 @@ import ( "net/http" "net/http/httptest" "slices" + "strings" "sync" "sync/atomic" "testing" "time" + "github.com/apernet/quic-go" + "github.com/apernet/quic-go/http3" xnet "github.com/xtls/xray-core/common/net" + "github.com/xtls/xray-core/common/net/cnc" + "github.com/xtls/xray-core/common/protocol/tls/cert" + "github.com/xtls/xray-core/common/utils" "github.com/xtls/xray-core/transport/internet" + "github.com/xtls/xray-core/transport/internet/finalmask" + "github.com/xtls/xray-core/transport/internet/finalmask/noise" + "github.com/xtls/xray-core/transport/internet/splithttp" "github.com/xtls/xray-core/transport/internet/tls" ) @@ -75,7 +85,7 @@ func sniForSettings(t *testing.T, serverName string) string { driveFilesURL = "https://www.googleapis.com/drive/v3/files" defer func() { driveFilesURL = prev }() - client := newServiceClient(settings, 5*time.Second, 8) + client := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8) req, err := http.NewRequest(http.MethodGet, driveFilesURL, nil) if err != nil { t.Fatalf("request: %v", err) @@ -145,7 +155,7 @@ func TestServiceClientEndsOnlyStalledRequests(t *testing.T) { } })) defer server.Close() - client := newServiceClient(&internet.MemoryStreamConfig{}, stall, 8) + client := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{}, stall, 8) fetch := func(path string, body io.Reader) (int, error) { req, err := http.NewRequest(http.MethodPost, server.URL+path, body) @@ -217,7 +227,7 @@ func TestServiceClientStallOverTLS(t *testing.T) { } time.Sleep(5 * stall) }) - client := newServiceClient(server.settings(&tls.Config{}), stall, 8) + client := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{}), stall, 8) for _, path := range []string{"/stuck", "/silent"} { start := time.Now() @@ -240,12 +250,13 @@ func TestServiceClientStallOverTLS(t *testing.T) { } // tlsServer is a TLS server that keeps the last hello it got and counts its -// connections +// connections, and those closed type tlsServer struct { *httptest.Server - mu sync.Mutex - hello *gotls.ClientHelloInfo - conns atomic.Int32 + mu sync.Mutex + hello *gotls.ClientHelloInfo + conns atomic.Int32 + closed atomic.Int32 } func newTLSServer(t *testing.T, config *gotls.Config, handler http.HandlerFunc) *tlsServer { @@ -260,8 +271,11 @@ func newTLSServer(t *testing.T, config *gotls.Config, handler http.HandlerFunc) } s.TLS = config s.Config.ConnState = func(_ net.Conn, state http.ConnState) { - if state == http.StateNew { + switch state { + case http.StateNew: s.conns.Add(1) + case http.StateClosed: + s.closed.Add(1) } } s.StartTLS() @@ -298,7 +312,7 @@ func TestServiceClientSpeaksWhatServerPicks(t *testing.T) { for _, protos := range [][]string{chromeALPN, {"http/1.1"}} { server := newTLSServer(t, &gotls.Config{NextProtos: protos}, func(w http.ResponseWriter, r *http.Request) {}) for _, config := range []*tls.Config{{}, {NextProtocol: []string{"h2"}}, {NextProtocol: []string{"http/1.1"}}, {Fingerprint: "unsafe"}} { - client := newServiceClient(server.settings(config), 5*time.Second, 8) + client := newServiceClient(context.Background(), new(sync.Map), server.settings(config), 5*time.Second, 8) resp, err := client.Get(server.URL) if err != nil { t.Fatalf("server %v, alpn %v, fingerprint %q: %v", protos, config.NextProtocol, config.Fingerprint, err) @@ -318,7 +332,7 @@ func TestServiceClientSpeaksWhatServerPicks(t *testing.T) { // offers certificate compression func TestServiceClientDefaultsToChrome(t *testing.T) { server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) {}) - client := newServiceClient(&internet.MemoryStreamConfig{}, 5*time.Second, 8) + client := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{}, 5*time.Second, 8) // the test certificate is not trusted, the hello is all that is needed if resp, err := client.Get(server.URL); err == nil { resp.Body.Close() @@ -332,6 +346,33 @@ func TestServiceClientDefaultsToChrome(t *testing.T) { } } +// a request says it is chrome, as the hello does, unless it says what it is +func TestServiceClientUserAgent(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.WriteString(w, r.UserAgent()) + })) + defer server.Close() + client := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{}, 5*time.Second, 8) + for ua, want := range map[string]string{"": utils.ChromeUA, "firefox": utils.FirefoxUA, "custom/1.0": "custom/1.0"} { + req, _ := http.NewRequest(http.MethodGet, server.URL, nil) + if ua != "" { + req.Header.Set("User-Agent", ua) + } + resp, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + got, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if string(got) != want { + t.Errorf("with %q the server got %q, not %q", ua, got, want) + } + if req.Header.Get("User-Agent") != ua || len(req.Header) > min(len(ua), 1) { + t.Errorf("with %q the request was changed to %v", ua, req.Header) + } + } +} + // parallel requests share one h2 connection instead of dialing one each, the // first ones as the later ones, and next to a host that answers over h1 func TestServiceClientOneConnection(t *testing.T) { @@ -339,7 +380,7 @@ func TestServiceClientOneConnection(t *testing.T) { time.Sleep(50 * time.Millisecond) }) h1 := newTLSServer(t, &gotls.Config{NextProtos: []string{"http/1.1"}}, func(w http.ResponseWriter, r *http.Request) {}) - client := newServiceClient(server.settings(&tls.Config{}), 5*time.Second, driveMaxInflight) + client := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{}), 5*time.Second, driveMaxInflight) get := func(url string) { resp, err := client.Get(url) if err != nil { @@ -395,7 +436,7 @@ func TestServiceClientHTTP1(t *testing.T) { server := newTLSServer(t, &gotls.Config{NextProtos: c.protos, EncryptedClientHelloKeys: echKeys}, func(w http.ResponseWriter, r *http.Request) { time.Sleep(50 * time.Millisecond) }) - client := newServiceClient(server.settings(c.config), 5*time.Second, 8) + client := newServiceClient(context.Background(), new(sync.Map), server.settings(c.config), 5*time.Second, 8) get := func() { resp, err := client.Get(server.URL) if err != nil { @@ -426,7 +467,7 @@ func TestServiceClientHTTP1(t *testing.T) { // without the keys the server gets the outer hello, which offers the ALPN of // Chrome outer := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) {}) - if resp, err := newServiceClient(outer.settings(ech), 5*time.Second, 8).Get(outer.URL); err == nil { + if resp, err := newServiceClient(context.Background(), new(sync.Map), outer.settings(ech), 5*time.Second, 8).Get(outer.URL); err == nil { resp.Body.Close() } if hello := outer.lastHello(); hello == nil { @@ -462,7 +503,7 @@ func TestServiceClientHungHandshake(t *testing.T) { server.Listener = &muteFirst{Listener: server.Listener} server.StartTLS() t.Cleanup(server.Close) - client := newServiceClient(server.settings(&tls.Config{Fingerprint: fingerprint}), stall, 8) + client := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{Fingerprint: fingerprint}), stall, 8) if _, err := client.Get(server.URL); err == nil { t.Fatalf("fingerprint %q: a request got through a handshake that hung", fingerprint) @@ -488,7 +529,7 @@ func TestServiceClientClosesFailedHandshake(t *testing.T) { for _, fingerprint := range []string{"", "unsafe"} { settings := &internet.MemoryStreamConfig{ProtocolName: protocolName, SecuritySettings: &tls.Config{Fingerprint: fingerprint}} - go newServiceClient(settings, 5*time.Second, 8).Get("https://" + ln.Addr().String()) + go newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8).Get("https://" + ln.Addr().String()) conn, err := ln.Accept() if err != nil { t.Fatal(err) @@ -503,6 +544,446 @@ func TestServiceClientClosesFailedHandshake(t *testing.T) { } } +// afterFuncs counts the functions that context.AfterFunc registers on it and +// that are not stopped yet +type afterFuncs struct { + context.Context + live atomic.Int32 +} + +// Value hides the cancelCtx below, which context.AfterFunc would use instead +func (c *afterFuncs) Value(any) any { return nil } + +func (c *afterFuncs) AfterFunc(func()) func() bool { + c.live.Add(1) + return func() bool { + c.live.Add(-1) + return true + } +} + +// a connection that ends on its own, here after Connection: close, leaves +// nothing on the context that closes the connections of the client +func TestServiceClientForgetsClosedConnections(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Connection", "close") + })) + defer server.Close() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + connCtx := &afterFuncs{Context: ctx} + client := newServiceClient(connCtx, new(sync.Map), &internet.MemoryStreamConfig{}, 5*time.Second, 8) + + for range 4 { + resp, err := client.Get(server.URL) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + } + waitFor(t, "the closed connections to leave the context", func() bool { return connCtx.live.Load() == 0 }) +} + type roundTrip func(*http.Request) (*http.Response, error) func (f roundTrip) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +// "hKeepAlivePeriod" of XMUX sets when an idle h2 connection pings, -1 never, +// and h3 keeps alive only when it is set +func TestServiceClientKeepAlive(t *testing.T) { + for period, want := range map[int64]time.Duration{0: xnet.ChromeH2KeepAlivePeriod, 20: 20 * time.Second, -1: 0} { + settings := &internet.MemoryStreamConfig{ProtocolSettings: &Config{Xmux: &splithttp.XmuxConfig{HKeepAlivePeriod: period}}} + client := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8) + h2 := client.Transport.(*serviceTransport).secure.(*secureTransport).h2.(*http.Transport) + if got := h2.HTTP2.SendPingTimeout; got != want { + t.Errorf("hKeepAlivePeriod %d pings after %v, want %v", period, got, want) + } + } + for period, want := range map[int64]time.Duration{0: 0, 20: 20 * time.Second, -1: 0} { + settings := &internet.MemoryStreamConfig{ + ProtocolSettings: &Config{Xmux: &splithttp.XmuxConfig{HKeepAlivePeriod: period}}, + SecuritySettings: &tls.Config{NextProtocol: []string{"h3"}}, + } + client := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8) + if got := client.Transport.(*serviceTransport).secure.(*http3.Transport).QUICConfig.KeepAlivePeriod; got != want { + t.Errorf("hKeepAlivePeriod %d keeps h3 alive every %v, want %v", period, got, want) + } + } +} + +// h3 is used when "alpn" asks for it alone, and only then +func TestServiceClientH3Selection(t *testing.T) { + for _, c := range []struct { + alpn []string + h3 bool + }{ + {nil, false}, + {[]string{"h2"}, false}, + {[]string{"http/1.1"}, false}, + {[]string{"h3", "h2"}, false}, + {[]string{"h2", "h3"}, false}, + {[]string{"h3"}, true}, + } { + client := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{SecuritySettings: &tls.Config{NextProtocol: c.alpn}}, 5*time.Second, 8) + if _, h3 := client.Transport.(*serviceTransport).secure.(*http3.Transport); h3 != c.h3 { + t.Errorf("alpn %v: h3 %v, want %v", c.alpn, h3, c.h3) + } + } + if _, h3 := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{}, 5*time.Second, 8).Transport.(*serviceTransport).secure.(*http3.Transport); h3 { + t.Error("h3 without tlsSettings") + } +} + +// h3Server is an http3 server that keeps the SNI of each connection, and counts +// its connections and those closed +type h3Server struct { + port xnet.Port + hash, caHash [32]byte + mu sync.Mutex + snis []string + conns atomic.Int32 + closed atomic.Int32 +} + +func newH3Server(t *testing.T, handler http.Handler) *h3Server { + t.Helper() + + // ECDSA, as ChromeParrot offers no Ed25519, from a CA of its own + ca, caHash := cert.MustGenerate(nil, cert.Authority(true), cert.KeyUsage(x509.KeyUsageCertSign)) + certificate, hash := cert.MustGenerate(ca, cert.DNSNames("www.googleapis.com", "oauth2.googleapis.com")) + key, err := x509.ParsePKCS8PrivateKey(certificate.PrivateKey) + if err != nil { + t.Fatal(err) + } + conn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + s := &h3Server{port: xnet.Port(conn.LocalAddr().(*net.UDPAddr).Port), hash: hash, caHash: caHash} + server := &http3.Server{ + Handler: handler, + TLSConfig: http3.ConfigureTLSConfig(&gotls.Config{ + Certificates: []gotls.Certificate{{Certificate: [][]byte{certificate.Certificate, ca.Certificate}, PrivateKey: key}}, + GetConfigForClient: func(hello *gotls.ClientHelloInfo) (*gotls.Config, error) { + s.mu.Lock() + s.snis = append(s.snis, hello.ServerName) + s.mu.Unlock() + return nil, nil + }, + }), + ConnContext: func(ctx context.Context, c *quic.Conn) context.Context { + s.conns.Add(1) + context.AfterFunc(c.Context(), func() { s.closed.Add(1) }) + return ctx + }, + } + go server.Serve(conn) + t.Cleanup(func() { + server.Close() + conn.Close() + }) + return s +} + +// settings sends every host to the server over h3, through a destination built +// as TCP the way conf builds it, and pins the certificate of the server in config +func (s *h3Server) settings(config *tls.Config) *internet.MemoryStreamConfig { + config.NextProtocol = []string{"h3"} + config.PinnedPeerCertSha256 = [][]byte{s.hash[:]} + return &internet.MemoryStreamConfig{ + ProtocolName: protocolName, + Destination: &xnet.Destination{Address: xnet.LocalHostIP, Port: s.port, Network: xnet.Network_TCP}, + SecuritySettings: config, + } +} + +func (s *h3Server) seenSNIs() []string { + s.mu.Lock() + defer s.mu.Unlock() + return slices.Sorted(slices.Values(s.snis)) +} + +// one client serves two hosts, and each gets its own SNI unless serverName +// says otherwise, which is also the name a pinned CA checks the certificate for +func TestServiceClientH3SNI(t *testing.T) { + for _, c := range []struct { + serverName string + pinCA bool + want []string + }{ + {"", false, []string{"oauth2.googleapis.com", "www.googleapis.com"}}, + {"www.google.com", false, []string{"www.google.com", "www.google.com"}}, + {"", true, []string{"oauth2.googleapis.com", "www.googleapis.com"}}, + } { + server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + settings := server.settings(&tls.Config{ServerName: c.serverName}) + if c.pinCA { + settings.SecuritySettings.(*tls.Config).PinnedPeerCertSha256 = [][]byte{server.caHash[:]} + } + client := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8) + for _, url := range []string{"https://www.googleapis.com/drive/v3/files", "https://oauth2.googleapis.com/token"} { + resp, err := client.Get(url) + if err != nil { + t.Fatalf("serverName %q, pinned CA %v, %s: %v", c.serverName, c.pinCA, url, err) + } + resp.Body.Close() + if resp.Proto != "HTTP/3.0" { + t.Errorf("serverName %q, %s: went over %s", c.serverName, url, resp.Proto) + } + } + if snis := server.seenSNIs(); !slices.Equal(snis, c.want) { + t.Errorf("serverName %q: SNIs %v, want %v", c.serverName, snis, c.want) + } + } +} + +// a request that stands still ends over h3 too, and says so +func TestServiceClientStallOverH3(t *testing.T) { + const stall = 200 * time.Millisecond + server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/stuck" { + w.Write([]byte("x")) + w.(http.Flusher).Flush() + } + time.Sleep(5 * stall) + })) + client := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{}), stall, 8) + + for _, path := range []string{"/stuck", "/silent"} { + start := time.Now() + resp, err := client.Get("https://www.googleapis.com" + path) + if err == nil { + if resp.Proto != "HTTP/3.0" { + t.Fatalf("%s went over %s", path, resp.Proto) + } + _, err = io.ReadAll(resp.Body) + resp.Body.Close() + } + if !errors.Is(err, errStalled) { + t.Errorf("%s gave %v, want %v", path, err, errStalled) + } + if took := time.Since(start); took > 4*stall { + t.Errorf("%s took %v to give up", path, took) + } + } +} + +// brutal and force-brutal, which are for Hysteria servers, give an error and +// no panic in a goroutine of http3, where XHTTP and DoH panic on brutal +func TestServiceClientH3Congestion(t *testing.T) { + server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + for _, cc := range []string{"", "bbr", "reno", "brutal", "force-brutal"} { + settings := server.settings(&tls.Config{}) + settings.QuicParams = &internet.QuicParams{Congestion: cc, BrutalUp: 1 << 20} + resp, err := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8).Get("https://www.googleapis.com/") + if err == nil { + resp.Body.Close() + } + if brutal := strings.HasSuffix(cc, "brutal"); brutal != (err != nil) || brutal && !strings.Contains(err.Error(), cc) { + t.Errorf("congestion %q gave %v", cc, err) + } + } +} + +// recordConn keeps the datagrams sent through it +type recordConn struct { + net.PacketConn + mu sync.Mutex + sent [][]byte + closed atomic.Bool +} + +func (c *recordConn) WriteTo(p []byte, addr net.Addr) (int, error) { + c.mu.Lock() + c.sent = append(c.sent, bytes.Clone(p)) + c.mu.Unlock() + return c.PacketConn.WriteTo(p, addr) +} + +func (c *recordConn) Close() error { + c.closed.Store(true) + return c.PacketConn.Close() +} + +// recorder dials UDP through recordConns and keeps them +type recorder struct { + mu sync.Mutex + conns []*recordConn +} + +func (r *recorder) dialUDP(ctx context.Context, dest xnet.Destination) (net.PacketConn, net.Addr, error) { + conn, err := internet.DialSystem(ctx, dest, nil) + if err != nil { + return nil, nil, err + } + rec := &recordConn{PacketConn: conn.(*xnet.PacketConnWrapper).PacketConn} + r.mu.Lock() + r.conns = append(r.conns, rec) + r.mu.Unlock() + return rec, conn.RemoteAddr(), nil +} + +// a connection ends with its socket once it is idle for its idle timeout, as +// keepalive is off unless asked for, and so does one whose handshake failed, +// and neither leaves anything on the context that closes the connections +func TestServiceClientH3Closes(t *testing.T) { + server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + for _, pin := range [][]byte{server.hash[:], make([]byte, 32)} { + rec := new(recorder) + settings := server.settings(&tls.Config{}) + settings.SecuritySettings.(*tls.Config).PinnedPeerCertSha256 = [][]byte{pin} + // ChromeParrot would keep the idle timeout at 30 s + settings.QuicParams = &internet.QuicParams{MaxIdleTimeout: 1, DisableChromeParrot: true} + settings.FinalMask = finalmask.NewFinalMask(nil, nil, nil, nil, rec.dialUDP, nil) + connCtx := &afterFuncs{Context: ctx} + resp, err := newServiceClient(connCtx, new(sync.Map), settings, 5*time.Second, 8).Get("https://www.googleapis.com/") + if err == nil { + resp.Body.Close() + } + if bytes.Equal(pin, server.hash[:]) != (err == nil) { + t.Fatalf("pin %x: %v", pin[:4], err) + } + + closed := func() bool { + rec.mu.Lock() + defer rec.mu.Unlock() + return len(rec.conns) == 1 && rec.conns[0].closed.Load() + } + for deadline := time.Now().Add(5 * time.Second); !closed() && time.Now().Before(deadline); { + time.Sleep(10 * time.Millisecond) + } + if !closed() { + t.Errorf("pin %x: the socket was left open", pin[:4]) + } + waitFor(t, "the closed connection to leave the context", func() bool { return connCtx.live.Load() == 0 }) + } +} + +// cncDialer gives UDP as a stream conn, the way dialerProxy does +type cncDialer struct{} + +func (cncDialer) Dial(ctx context.Context, src xnet.Address, dest xnet.Destination, sockopt *internet.SocketConfig) (net.Conn, error) { + conn, err := net.Dial("udp", dest.NetAddr()) + if err != nil { + return nil, err + } + return cnc.NewConnection(cnc.ConnectionInput(conn), cnc.ConnectionOutput(conn), cnc.ConnectionOnClose(conn)), nil +} + +func (cncDialer) DestIpAddress() net.IP { return nil } + +// h3 goes over the stream conn of a dialerProxy too +func TestServiceClientH3DialerProxy(t *testing.T) { + internet.UseAlternativeSystemDialer(cncDialer{}) + defer internet.UseAlternativeSystemDialer(nil) + + server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + resp, err := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{}), 5*time.Second, 8).Get("https://www.googleapis.com/") + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.Proto != "HTTP/3.0" { + t.Errorf("went over %s", resp.Proto) + } +} + +// xdrive moves data over h3 to a fake Drive, with the noise of a UDP mask sent +// ahead of the Initial of each connection +func TestDriveOverH3(t *testing.T) { + drive := newFakeDrive(t) + server := newH3Server(t, drive.server.Config.Handler) + driveTokenURL = "https://oauth2.googleapis.com/token" + driveFilesURL = "https://www.googleapis.com/files" + driveUploadURL = "https://www.googleapis.com/upload" + + rec := new(recorder) + settings := server.settings(&tls.Config{}) + settings.ProtocolSettings = driveSettings().ProtocolSettings + settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{MaxConnections: &splithttp.RangeConfig{From: 4, To: 4}} + masks := []finalmask.UDPMask{&noise.Config{Items: []*noise.Item{{Packet: []byte("noise")}}}} + settings.FinalMask = finalmask.NewFinalMask(nil, masks, nil, nil, rec.dialUDP, nil) + + client, srv, cleanup := pairWith(t, settings) + defer cleanup() + + if _, err := client.Write([]byte("ping")); err != nil { + t.Fatalf("client write: %v", err) + } + expectRead(t, srv, "ping") + if _, err := srv.Write([]byte("pong")); err != nil { + t.Fatalf("server write: %v", err) + } + expectRead(t, client, "pong") + + payload := make([]byte, 300000) + for i := range payload { + payload[i] = byte(i % 251) + } + go client.Write(payload) + if err := srv.SetReadDeadline(time.Now().Add(testPatience)); err != nil { + t.Fatalf("SetReadDeadline: %v", err) + } + got := make([]byte, len(payload)) + if _, err := io.ReadFull(srv, got); err != nil { + t.Fatalf("ReadFull: %v", err) + } + if !bytes.Equal(got, payload) { + t.Fatal("payload mismatch") + } + + rec.mu.Lock() + defer rec.mu.Unlock() + // one for the listener to www.googleapis.com, one for the token to + // oauth2.googleapis.com and one of XMUX for each end of the session + if len(rec.conns) != 4 { + t.Errorf("%d connections went through the mask, want 4", len(rec.conns)) + } + for i, conn := range rec.conns { + conn.mu.Lock() + sent := conn.sent + conn.mu.Unlock() + if len(sent) < 2 { + t.Errorf("connection %d sent %d datagrams", i, len(sent)) + continue + } + // the noise, then a QUIC v1 Initial of 1250 bytes with no source + // connection ID, as ChromeParrot sends it + first, initial := sent[0], sent[1] + if string(first) != "noise" || len(initial) != 1250 || initial[0]&0xf0 != 0xc0 || !bytes.Equal(initial[1:5], []byte{0, 0, 0, 1}) || initial[6+int(initial[5])] != 0 { + t.Errorf("connection %d began with %q and %d bytes %x", i, first, len(initial), initial[:min(len(initial), 24)]) + } + } +} + +// over h3 a connection of XMUX is a QUIC connection, which closes once XMUX +// drops it, so that the server hears of it +func TestDriveXmuxOverH3(t *testing.T) { + drive := newFakeDrive(t) + server := newH3Server(t, drive.server.Config.Handler) + driveTokenURL = "https://www.googleapis.com/token" + driveFilesURL = "https://www.googleapis.com/files" + driveUploadURL = "https://www.googleapis.com/upload" + settings := server.settings(&tls.Config{}) + settings.ProtocolSettings = driveSettings().ProtocolSettings + settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{HMaxRequestTimes: &splithttp.RangeConfig{From: 3, To: 3}} + storage, err := newDriveStorage(settings, settings.ProtocolSettings.(*Config)) + if err != nil { + t.Fatalf("newDriveStorage: %v", err) + } + session := sessionStorage(storage) + defer session.Close() + + for _, name := range []string{"a", "b", "c", "d"} { + if err := session.Put(context.Background(), "streams/abc/c2s/"+name, []byte("x")); err != nil { + t.Fatalf("Put: %v", err) + } + } + // one for the token, the first of XMUX for 3 requests and one more + if n := server.conns.Load(); n != 3 { + t.Errorf("the token and 4 requests with 3 to a connection took %d connections", n) + } + waitFor(t, "the first connection of XMUX to close", func() bool { return server.closed.Load() == 1 }) +} diff --git a/transport/internet/xdrive/config.pb.go b/transport/internet/xdrive/config.pb.go index fa733163a884..a350f0149c0d 100644 --- a/transport/internet/xdrive/config.pb.go +++ b/transport/internet/xdrive/config.pb.go @@ -7,6 +7,7 @@ package xdrive 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" @@ -35,6 +36,7 @@ type Config struct { EagerWindowMs uint32 `protobuf:"varint,10,opt,name=eager_window_ms,json=eagerWindowMs,proto3" json:"eager_window_ms,omitempty"` HoleTimeoutMs uint32 `protobuf:"varint,11,opt,name=hole_timeout_ms,json=holeTimeoutMs,proto3" json:"hole_timeout_ms,omitempty"` Template string `protobuf:"bytes,12,opt,name=template,proto3" json:"template,omitempty"` + Xmux *splithttp.XmuxConfig `protobuf:"bytes,13,opt,name=xmux,proto3" json:"xmux,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -153,11 +155,18 @@ func (x *Config) GetTemplate() string { return "" } +func (x *Config) GetXmux() *splithttp.XmuxConfig { + if x != nil { + return x.Xmux + } + return nil +} + var File_transport_internet_xdrive_config_proto protoreflect.FileDescriptor const file_transport_internet_xdrive_config_proto_rawDesc = "" + "\n" + - "&transport/internet/xdrive/config.proto\x12\x1exray.transport.internet.xdrive\"\xcb\x03\n" + + "&transport/internet/xdrive/config.proto\x12\x1exray.transport.internet.xdrive\x1a)transport/internet/splithttp/config.proto\"\x8e\x04\n" + "\x06Config\x12#\n" + "\rremote_folder\x18\x01 \x01(\tR\fremoteFolder\x12\x18\n" + "\aservice\x18\x02 \x01(\tR\aservice\x12\x18\n" + @@ -171,7 +180,8 @@ const file_transport_internet_xdrive_config_proto_rawDesc = "" + "\x0feager_window_ms\x18\n" + " \x01(\rR\reagerWindowMs\x12&\n" + "\x0fhole_timeout_ms\x18\v \x01(\rR\rholeTimeoutMs\x12\x1a\n" + - "\btemplate\x18\f \x01(\tR\btemplateB5Z3github.com/xtls/xray-core/transport/internet/xdriveb\x06proto3" + "\btemplate\x18\f \x01(\tR\btemplate\x12A\n" + + "\x04xmux\x18\r \x01(\v2-.xray.transport.internet.splithttp.XmuxConfigR\x04xmuxB5Z3github.com/xtls/xray-core/transport/internet/xdriveb\x06proto3" var ( file_transport_internet_xdrive_config_proto_rawDescOnce sync.Once @@ -187,14 +197,16 @@ func file_transport_internet_xdrive_config_proto_rawDescGZIP() []byte { var file_transport_internet_xdrive_config_proto_msgTypes = make([]protoimpl.MessageInfo, 1) var file_transport_internet_xdrive_config_proto_goTypes = []any{ - (*Config)(nil), // 0: xray.transport.internet.xdrive.Config + (*Config)(nil), // 0: xray.transport.internet.xdrive.Config + (*splithttp.XmuxConfig)(nil), // 1: xray.transport.internet.splithttp.XmuxConfig } var file_transport_internet_xdrive_config_proto_depIdxs = []int32{ - 0, // [0:0] is the sub-list for method output_type - 0, // [0:0] is the sub-list for method input_type - 0, // [0:0] is the sub-list for extension type_name - 0, // [0:0] is the sub-list for extension extendee - 0, // [0:0] is the sub-list for field type_name + 1, // 0: xray.transport.internet.xdrive.Config.xmux:type_name -> xray.transport.internet.splithttp.XmuxConfig + 1, // [1:1] is the sub-list for method output_type + 1, // [1:1] is the sub-list for method input_type + 1, // [1:1] is the sub-list for extension type_name + 1, // [1:1] is the sub-list for extension extendee + 0, // [0:1] is the sub-list for field type_name } func init() { file_transport_internet_xdrive_config_proto_init() } diff --git a/transport/internet/xdrive/config.proto b/transport/internet/xdrive/config.proto index 224a1f66f8fe..f10687a9ea1b 100644 --- a/transport/internet/xdrive/config.proto +++ b/transport/internet/xdrive/config.proto @@ -3,6 +3,8 @@ syntax = "proto3"; package xray.transport.internet.xdrive; option go_package = "github.com/xtls/xray-core/transport/internet/xdrive"; +import "transport/internet/splithttp/config.proto"; + message Config { string remote_folder = 1; string service = 2; @@ -16,4 +18,5 @@ message Config { uint32 eager_window_ms = 10; uint32 hole_timeout_ms = 11; string template = 12; + xray.transport.internet.splithttp.XmuxConfig xmux = 13; } diff --git a/transport/internet/xdrive/drive.go b/transport/internet/xdrive/drive.go index 7a2403386610..472efc91b95c 100644 --- a/transport/internet/xdrive/drive.go +++ b/transport/internet/xdrive/drive.go @@ -39,11 +39,21 @@ var ( ) type driveStorage struct { + *driveShared + // a session has a client and inflight of its own, see session + client *http.Client + inflight chan struct{} + release func() +} + +// driveShared is what the sessions of a storage have in common +type driveShared struct { + conns *connPool folder string clientID string clientSecret string refreshToken string - client *http.Client + tokenClient *http.Client tokenURL string filesURL string uploadURL string @@ -53,8 +63,6 @@ type driveStorage struct { token string tokenExpiry time.Time - inflight chan struct{} - idMu sync.Mutex ids map[string]string } @@ -72,21 +80,41 @@ func newDriveStorage(streamSettings *internet.MemoryStreamConfig, config *Config } } + h1Hosts := new(sync.Map) + client := newServiceClient(context.Background(), h1Hosts, streamSettings, driveTimeout, driveMaxInflight) return &driveStorage{ - folder: config.RemoteFolder, - clientID: config.Secrets[0], - clientSecret: config.Secrets[1], - refreshToken: config.Secrets[2], - client: newServiceClient(streamSettings, driveTimeout, driveMaxInflight), - tokenURL: driveTokenURL, - filesURL: driveFilesURL, - uploadURL: driveUploadURL, - backoff: driveInitialBackoff, - inflight: make(chan struct{}, driveMaxInflight), - ids: make(map[string]string), + driveShared: &driveShared{ + conns: newConnPool(config.Xmux, func(connCtx context.Context) *http.Client { + return newServiceClient(connCtx, h1Hosts, streamSettings, driveTimeout, driveMaxInflight) + }), + folder: config.RemoteFolder, + clientID: config.Secrets[0], + clientSecret: config.Secrets[1], + refreshToken: config.Secrets[2], + tokenClient: client, + tokenURL: driveTokenURL, + filesURL: driveFilesURL, + uploadURL: driveUploadURL, + backoff: driveInitialBackoff, + ids: make(map[string]string), + }, + client: client, + inflight: make(chan struct{}, driveMaxInflight), }, nil } +// session returns a storage for one session, over the connections of XMUX +// until Close, with the token and ids of s +func (s *driveStorage) session() Storage { + transport := &sessionTransport{pool: s.conns} + return &driveStorage{ + driveShared: s.driveShared, + client: &http.Client{Transport: transport}, + inflight: make(chan struct{}, driveMaxInflight), + release: transport.Close, + } +} + func flatten(name string) string { return strings.ReplaceAll(name, "/", flatSeparator) } @@ -115,7 +143,7 @@ func (s *driveStorage) accessToken(ctx context.Context) (string, error) { } req.Header.Set("Content-Type", "application/x-www-form-urlencoded") - resp, err := s.client.Do(req) + resp, err := s.tokenClient.Do(req) if err != nil { return "", errors.New("failed to refresh the access token").Base(err) } @@ -572,5 +600,8 @@ func (s *driveStorage) List(ctx context.Context, prefix string) ([]Entry, error) } func (s *driveStorage) Close() error { + if s.release != nil { + s.release() + } return nil } diff --git a/transport/internet/xdrive/drive_test.go b/transport/internet/xdrive/drive_test.go index b43235093e0c..7a5a5b888c05 100644 --- a/transport/internet/xdrive/drive_test.go +++ b/transport/internet/xdrive/drive_test.go @@ -562,6 +562,12 @@ func TestDriveSharedClient(t *testing.T) { if first != second { t.Fatal("same settings did not share one storage") } + // a session has an inflight of its own, as large + session := sessionStorage(first).(*driveStorage) + defer session.Close() + if session.inflight == first.(*driveStorage).inflight || cap(session.inflight) != driveMaxInflight { + t.Fatalf("a session shares inflight or has cap %d", cap(session.inflight)) + } } func TestDriveOldServiceName(t *testing.T) { diff --git a/transport/internet/xdrive/storage.go b/transport/internet/xdrive/storage.go index c83a8b5f110a..dd49a13f9116 100644 --- a/transport/internet/xdrive/storage.go +++ b/transport/internet/xdrive/storage.go @@ -48,6 +48,19 @@ func newStorage(streamSettings *internet.MemoryStreamConfig) (Storage, error) { } } +// sessionStorage returns the storage of one session, which takes connections +// of XMUX until Close, with the token and cache of storage. Local storage has +// no connections +func sessionStorage(storage Storage) Storage { + switch s := storage.(type) { + case *driveStorage: + return s.session() + case *templateStorage: + return s.session() + } + return storage +} + var ( sharedMu sync.Mutex shared = make(map[string]Storage) diff --git a/transport/internet/xdrive/template.go b/transport/internet/xdrive/template.go index 9b7be908732c..4d6560f4d5a9 100644 --- a/transport/internet/xdrive/template.go +++ b/transport/internet/xdrive/template.go @@ -54,12 +54,20 @@ type storageTemplate struct { } type templateStorage struct { - tmpl *storageTemplate - client *http.Client - folder string - secrets []string - + *templateShared + // a session has a client and inflight of its own, see session + client *http.Client inflight chan struct{} + release func() +} + +// templateShared is what the sessions of a storage have in common +type templateShared struct { + conns *connPool + tmpl *storageTemplate + tokenClient *http.Client + folder string + secrets []string tokenMu sync.Mutex token string @@ -94,15 +102,35 @@ func newTemplateStorage(streamSettings *internet.MemoryStreamConfig, config *Con conc = maxTemplateConcurrency } + h1Hosts := new(sync.Map) + client := newServiceClient(context.Background(), h1Hosts, streamSettings, driveTimeout, conc) return &templateStorage{ - tmpl: tmpl, - client: newServiceClient(streamSettings, driveTimeout, conc), - folder: config.RemoteFolder, - secrets: config.Secrets, + templateShared: &templateShared{ + conns: newConnPool(config.Xmux, func(connCtx context.Context) *http.Client { + return newServiceClient(connCtx, h1Hosts, streamSettings, driveTimeout, conc) + }), + tmpl: tmpl, + tokenClient: client, + folder: config.RemoteFolder, + secrets: config.Secrets, + }, + client: client, inflight: make(chan struct{}, conc), }, nil } +// session returns a storage for one session, over the connections of XMUX +// until Close, with the token of s +func (s *templateStorage) session() Storage { + transport := &sessionTransport{pool: s.conns} + return &templateStorage{ + templateShared: s.templateShared, + client: &http.Client{Transport: transport}, + inflight: make(chan struct{}, cap(s.inflight)), + release: transport.Close, + } +} + const maxTemplateConcurrency = 256 func (s *templateStorage) baseVars() map[string]string { @@ -216,7 +244,7 @@ func (s *templateStorage) accessToken(ctx context.Context) (string, error) { } req.Header.Set("Content-Type", "application/x-www-form-urlencoded") - resp, err := s.client.Do(req) + resp, err := s.tokenClient.Do(req) if err != nil { return "", errors.New("failed to fetch the token").Base(err) } @@ -441,5 +469,8 @@ func (s *templateStorage) List(ctx context.Context, prefix string) ([]Entry, err } func (s *templateStorage) Close() error { + if s.release != nil { + s.release() + } return nil } diff --git a/transport/internet/xdrive/template_test.go b/transport/internet/xdrive/template_test.go index 446cc953d4f1..bc6316c5baeb 100644 --- a/transport/internet/xdrive/template_test.go +++ b/transport/internet/xdrive/template_test.go @@ -113,7 +113,7 @@ func templateSettings(store *fakeStore, auth map[string]interface{}, secrets []s "delete": map[string]interface{}{"method": "DELETE", "url": base + "/folder/{name}"}, "list": map[string]interface{}{ "method": "PROPFIND", - "url": base + "/folder/", + "url": base + "/folder/?prefix={prefix}", "namesRegex": `/folder/([^<]+)`, }, "retry": map[string]interface{}{"status": []int{429, 500, 502, 503}}, @@ -297,6 +297,12 @@ func TestTemplateConcurrency(t *testing.T) { if cap(storage.inflight) != 4 { t.Fatalf("inflight cap is %d, want 4 from the template", cap(storage.inflight)) } + // and so has a session, of its own + session := storage.session().(*templateStorage) + defer session.Close() + if session.inflight == storage.inflight || cap(session.inflight) != 4 { + t.Fatalf("a session shares inflight or has cap %d, want 4", cap(session.inflight)) + } } func TestTemplateConcurrencyDefault(t *testing.T) { diff --git a/transport/internet/xdrive/xdrive.go b/transport/internet/xdrive/xdrive.go index 3c078b25208b..e79e738d32e2 100644 --- a/transport/internet/xdrive/xdrive.go +++ b/transport/internet/xdrive/xdrive.go @@ -87,6 +87,7 @@ func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.Me if err != nil { return nil, err } + storage = sessionStorage(storage) session, err := newSessionID() if err != nil { @@ -228,8 +229,10 @@ func (l *Listener) claim(session string) bool { } func (l *Listener) newSessionConn(session string) *Conn { - return newConn(l.ctx, l.storage, + storage := sessionStorage(l.storage) + return newConn(l.ctx, storage, downlinkPrefix(session), uplinkPrefix(session), l.params, func() { + storage.Close() l.mu.Lock() delete(l.active, session) l.mu.Unlock() diff --git a/transport/internet/xdrive/xdrive_test.go b/transport/internet/xdrive/xdrive_test.go index b91247f76fa7..aed8e78738b3 100644 --- a/transport/internet/xdrive/xdrive_test.go +++ b/transport/internet/xdrive/xdrive_test.go @@ -4,16 +4,21 @@ import ( "bytes" "context" "crypto/rand" + gotls "crypto/tls" "io" + "net/http" "os" "strings" + "sync" "sync/atomic" "testing" "time" "github.com/xtls/xray-core/common/net" "github.com/xtls/xray-core/transport/internet" + "github.com/xtls/xray-core/transport/internet/splithttp" "github.com/xtls/xray-core/transport/internet/stat" + "github.com/xtls/xray-core/transport/internet/tls" ) const testPatience = 30 * time.Second @@ -581,3 +586,388 @@ func TestInlinePayload(t *testing.T) { t.Fatalf("called Get %d times for an inline payload", got) } } + +// runSessions opens n sessions at once, both ends in this process, passes data +// both ways over each and closes them +func runSessions(t *testing.T, streamSettings *internet.MemoryStreamConfig, n int) { + t.Helper() + + accepted := make(chan stat.Connection, 1) + listener, err := Serve(context.Background(), net.LocalHostIP, net.Port(0), streamSettings, func(conn stat.Connection) { + accepted <- conn + }) + if err != nil { + t.Fatalf("Serve: %v", err) + } + defer listener.Close() + + var ends []stat.Connection + defer func() { + for _, end := range ends { + end.Close() + } + }() + for range n { + client, err := Dial(context.Background(), net.Destination{}, streamSettings) + if err != nil { + t.Fatalf("Dial: %v", err) + } + ends = append(ends, client) + select { + case peer := <-accepted: + ends = append(ends, peer) + case <-time.After(testPatience): + t.Fatal("listener did not accept the session") + } + } + for i := 0; i < len(ends); i += 2 { + if _, err := ends[i].Write([]byte("ping")); err != nil { + t.Fatalf("client write: %v", err) + } + expectRead(t, ends[i+1], "ping") + if _, err := ends[i+1].Write([]byte("pong")); err != nil { + t.Fatalf("server write: %v", err) + } + expectRead(t, ends[i], "pong") + } +} + +// sessionEnd is the session and end of it that a request to the fake store +// comes from, or "" for the listener and the token +func sessionEnd(r *http.Request) string { + name := strings.TrimPrefix(r.URL.Path, "/folder/") + if r.Method == "PROPFIND" { + name = r.URL.Query().Get("prefix") + } + write := r.Method == http.MethodPut + parts := strings.Split(name, flatSeparator) + switch { + case len(parts) > 2 && parts[0] == streamsDir: + // each end writes its own direction and reads the other + if (parts[2] == uplinkDir) == write { + return parts[1] + " client" + } + return parts[1] + " server" + case len(parts) == 2 && parts[0] == sessionsDir && write: + _, session, _ := strings.Cut(parts[1], "-") + return session + " client" + } + return "" +} + +// with "maxConcurrency" 1 every end of a session has a connection of its own, +// with "maxConnections" the ends share that many, and the sessions after them +// take the same ones again. The listener and the token use none of them, and +// the sessions get the token once +func TestXmuxConnections(t *testing.T) { + const n = 3 + for _, c := range []struct { + name string + xmux *splithttp.XmuxConfig + conns int + shared bool + }{ + {"maxConcurrency", &splithttp.XmuxConfig{MaxConcurrency: &splithttp.RangeConfig{From: 1, To: 1}}, 2 * n, false}, + {"maxConnections", &splithttp.XmuxConfig{MaxConnections: &splithttp.RangeConfig{From: 2, To: 2}}, 2, true}, + } { + t.Run(c.name, func(t *testing.T) { + resetSharedStorage() + t.Cleanup(resetSharedStorage) + + store := &fakeStore{objects: make(map[string][]byte), needAuth: "Bearer tok-fake"} + var ( + mu sync.Mutex + conns = make(map[string]map[string]bool) + ) + server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) { + end := sessionEnd(r) + mu.Lock() + if conns[end] == nil { + conns[end] = make(map[string]bool) + } + conns[end][r.RemoteAddr] = true + mu.Unlock() + store.handle(w, r) + }) + store.server = server.Server + settings := templateSettings(store, map[string]interface{}{ + "type": "oauth2", + "tokenUrl": server.URL + "/token", + "form": map[string]interface{}{"grant_type": "refresh_token", "refresh_token": "{secret0}"}, + "header": map[string]interface{}{"Authorization": "Bearer {token}"}, + }, []string{"refresh"}) + settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings + settings.ProtocolSettings.(*Config).Xmux = c.xmux + + runSessions(t, settings, n) + mu.Lock() + first := conns + conns = make(map[string]map[string]bool) + mu.Unlock() + dialed := server.conns.Load() + runSessions(t, settings, n) + if got := server.conns.Load(); got != dialed { + t.Errorf("the next %d sessions dialed %d connections", n, got-dialed) + } + + owner := make(map[string]string) + for end, addrs := range first { + if end == "" { + continue + } + if len(addrs) != 1 { + t.Errorf("%s used %d connections", end, len(addrs)) + } + for addr := range addrs { + if owner[addr] != "" && !c.shared { + t.Errorf("%s and %s shared a connection", end, owner[addr]) + } + owner[addr] = end + } + } + if len(owner) != c.conns { + t.Errorf("%d sessions had %d connections, want %d", 2*n, len(owner), c.conns) + } + for addr := range first[""] { + if owner[addr] != "" { + t.Errorf("the listener or the token went over the connection of %s", owner[addr]) + } + } + store.mu.Lock() + defer store.mu.Unlock() + if store.tokens != 1 { + t.Errorf("the sessions got the token %d times", store.tokens) + } + }) + } +} + +func TestDriveXmuxConnections(t *testing.T) { + drive := newFakeDrive(t) + server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, drive.server.Config.Handler.ServeHTTP) + driveTokenURL = server.URL + "/token" + driveFilesURL = server.URL + "/files" + driveUploadURL = server.URL + "/upload" + settings := driveSettings() + settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings + settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{MaxConnections: &splithttp.RangeConfig{From: 3, To: 3}} + + runSessions(t, settings, 3) + if got := server.conns.Load(); got != 4 { + t.Errorf("6 ends of sessions with 3 connections and the listener took %d connections", got) + } + + drive.mu.Lock() + defer drive.mu.Unlock() + if drive.tokens != 1 { + t.Errorf("6 ends of sessions got the token %d times", drive.tokens) + } +} + +// a session knows the hosts that answered the storage over h1, the polling of +// the listener for one, so it sends requests in a row over one connection, not +// first over one for h2 that then stays idle, and parallel ones over more +func TestSessionKnowsHTTP1Hosts(t *testing.T) { + check := func(name string, server *tlsServer, storage Storage) { + put := func(s Storage, names ...string) { + for _, n := range names { + if err := s.Put(context.Background(), "streams/abc/c2s/"+n, []byte("x")); err != nil { + t.Errorf("%s: Put: %v", name, err) + } + } + } + put(storage, "a", "b") + before := server.conns.Load() + session := sessionStorage(storage) + defer session.Close() + put(session, "c", "d") + if n := server.conns.Load() - before; n != 1 { + t.Errorf("%s: 2 requests in a row of a session over h1 took %d connections", name, n) + } + + before = server.conns.Load() + var wg sync.WaitGroup + for _, n := range []string{"e", "f", "g", "h"} { + wg.Go(func() { put(session, n) }) + } + wg.Wait() + if server.conns.Load() == before { + t.Errorf("%s: 4 parallel requests of a session over h1 took one connection", name) + } + } + slow := func(next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + time.Sleep(20 * time.Millisecond) + next(w, r) + } + } + + store := &fakeStore{objects: make(map[string][]byte)} + server := newTLSServer(t, &gotls.Config{NextProtos: []string{"http/1.1"}}, slow(store.handle)) + store.server = server.Server + settings := templateSettings(store, map[string]interface{}{"type": "none"}, nil) + settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings + template, err := newTemplateStorage(settings, settings.ProtocolSettings.(*Config)) + if err != nil { + t.Fatalf("newTemplateStorage: %v", err) + } + check("template", server, template) + + drive := newFakeDrive(t) + server = newTLSServer(t, &gotls.Config{NextProtos: []string{"http/1.1"}}, slow(drive.server.Config.Handler.ServeHTTP)) + driveTokenURL = server.URL + "/token" + driveFilesURL = server.URL + "/files" + settings = driveSettings() + settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings + google, err := newDriveStorage(settings, settings.ProtocolSettings.(*Config)) + if err != nil { + t.Fatalf("newDriveStorage: %v", err) + } + check("drive", server, google) +} + +// with "hMaxRequestTimes" a session moves on to a new connection after that +// many requests, and the old one closes once the requests on it end, not before +func TestXmuxRequestLimit(t *testing.T) { + store := &fakeStore{objects: make(map[string][]byte)} + arrived := make(chan struct{}, 1) + release := make(chan struct{}) + server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + arrived <- struct{}{} + <-release + } + store.handle(w, r) + }) + store.server = server.Server + settings := templateSettings(store, map[string]interface{}{"type": "none"}, nil) + settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings + settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{HMaxRequestTimes: &splithttp.RangeConfig{From: 3, To: 3}} + storage, err := newTemplateStorage(settings, settings.ProtocolSettings.(*Config)) + if err != nil { + t.Fatalf("newTemplateStorage: %v", err) + } + session := sessionStorage(storage) + defer session.Close() + put := func(names ...string) { + for _, name := range names { + if err := session.Put(context.Background(), "streams/abc/c2s/"+name, []byte("x")); err != nil { + t.Fatalf("Put: %v", err) + } + } + } + + put("a") + got := make(chan error, 1) + go func() { + _, err := session.Get(context.Background(), "streams/abc/c2s/a") + got <- err + }() + <-arrived + put("b", "c") + if n := server.conns.Load(); n != 2 { + t.Errorf("4 requests with 3 to a connection took %d connections", n) + } + if server.closed.Load() != 0 { + t.Error("the first connection closed under a request") + } + close(release) + if err := <-got; err != nil { + t.Errorf("Get on the first connection: %v", err) + } + waitFor(t, "the first connection to close", func() bool { return server.closed.Load() == 1 }) + + put("d", "e", "f") + if n := server.conns.Load(); n != 3 { + t.Errorf("7 requests with 3 to a connection took %d connections", n) + } + waitFor(t, "the second connection to close", func() bool { return server.closed.Load() == 2 }) +} + +// a session that fails to announce itself lets go of its connection, which the +// next one takes with "maxConcurrency" 1 +func TestFailedDialReleasesConnection(t *testing.T) { + resetSharedStorage() + t.Cleanup(resetSharedStorage) + + store := &fakeStore{objects: make(map[string][]byte)} + var refuse atomic.Bool + refuse.Store(true) + server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) { + if refuse.Load() { + w.WriteHeader(http.StatusUnauthorized) + return + } + store.handle(w, r) + }) + store.server = server.Server + settings := templateSettings(store, map[string]interface{}{"type": "none"}, nil) + settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings + settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{MaxConcurrency: &splithttp.RangeConfig{From: 1, To: 1}} + + if _, err := Dial(context.Background(), net.Destination{}, settings); err == nil { + t.Fatal("Dial announced a session that the service refused") + } + refuse.Store(false) + conn, err := Dial(context.Background(), net.Destination{}, settings) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer conn.Close() + if n := server.conns.Load(); n != 1 { + t.Errorf("a failed Dial and the next took %d connections", n) + } +} + +// a session gets the token over a connection of the storage, not over one of +// XMUX +func TestSessionTokenConnection(t *testing.T) { + check := func(name string, server *tlsServer, storage Storage) { + session := sessionStorage(storage) + defer session.Close() + if err := session.Put(context.Background(), "sessions/abc", nil); err != nil { + t.Fatalf("%s: Put: %v", name, err) + } + if n := server.conns.Load(); n != 2 { + t.Errorf("%s: the token and a Put of a session took %d connections, want 2", name, n) + } + } + + store := &fakeStore{objects: make(map[string][]byte), needAuth: "Bearer tok-fake"} + server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, store.handle) + store.server = server.Server + settings := templateSettings(store, map[string]interface{}{ + "type": "oauth2", + "tokenUrl": server.URL + "/token", + "header": map[string]interface{}{"Authorization": "Bearer {token}"}, + }, nil) + settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings + template, err := newTemplateStorage(settings, settings.ProtocolSettings.(*Config)) + if err != nil { + t.Fatalf("newTemplateStorage: %v", err) + } + check("template", server, template) + + drive := newFakeDrive(t) + server = newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, drive.server.Config.Handler.ServeHTTP) + driveTokenURL = server.URL + "/token" + driveFilesURL = server.URL + "/files" + settings = driveSettings() + settings.SecuritySettings = server.settings(&tls.Config{}).SecuritySettings + google, err := newDriveStorage(settings, settings.ProtocolSettings.(*Config)) + if err != nil { + t.Fatalf("newDriveStorage: %v", err) + } + check("drive", server, google) +} + +// local storage has no connections to keep apart +func TestLocalSessionStorage(t *testing.T) { + storage, err := newLocalStorage(t.TempDir()) + if err != nil { + t.Fatalf("newLocalStorage: %v", err) + } + if sessionStorage(storage) != Storage(storage) { + t.Fatal("a session of local storage got a storage of its own") + } +}