diff --git a/infra/conf/transport_internet.go b/infra/conf/transport_internet.go index ee1e5c3df..e016d7948 100644 --- a/infra/conf/transport_internet.go +++ b/infra/conf/transport_internet.go @@ -1,11 +1,15 @@ package conf import ( + "context" + "slices" "strings" + utls "github.com/refraction-networking/utls" "github.com/xtls/xray-core/common/errors" "github.com/xtls/xray-core/common/serial" "github.com/xtls/xray-core/transport/internet" + "github.com/xtls/xray-core/transport/internet/tls" ) type TransportProtocol string @@ -100,6 +104,14 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) { tm := serial.ToTypedMessage(ts) config.SecuritySettings = append(config.SecuritySettings, tm) config.SecurityType = tm.Type + 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 + if fp := tls.GetFingerprint(strings.ToLower(tlsSettings.Fingerprint)); fp != nil && fp.Client != utls.HelloGolang.Client && + !strings.HasPrefix(fp.Client, utls.HelloRandomized.Client) { + errors.LogWarning(context.Background(), `XDRIVE: "alpn" ["http/1.1"] reaches the wire only inside ECH, without "echConfigList" the hello carries the ALPN of the fingerprint and the server may pick h2`) + } + } case "reality": if config.ProtocolName != "tcp" && config.ProtocolName != "splithttp" && config.ProtocolName != "grpc" { return nil, errors.New("REALITY only supports RAW, XHTTP and gRPC for now.") diff --git a/infra/conf/transport_test.go b/infra/conf/transport_test.go index a6288ac36..47d75970c 100644 --- a/infra/conf/transport_test.go +++ b/infra/conf/transport_test.go @@ -5,6 +5,7 @@ import ( "strings" "testing" + "github.com/xtls/xray-core/common/log" . "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" @@ -355,6 +356,56 @@ func TestXDriveTemplateStreamConfig(t *testing.T) { } } +type logLines []string + +func (l *logLines) Handle(msg log.Message) { + *l = append(*l, msg.String()) +} + +// "alpn" ["http/1.1"] is warned about where the hello does not carry it, which +// is outside ECH with a fingerprint that has an ALPN of its own +func TestXDriveHTTP1ALPNWarning(t *testing.T) { + var lines logLines + log.RegisterHandler(&lines) + + for _, c := range []struct { + method, tlsSettings string + warn bool + }{ + {"xdrive", `{}`, false}, + {"xdrive", `{"alpn": ["http/1.1"]}`, true}, + {"xdrive", `{"alpn": ["http/1.1"], "fingerprint": "Firefox"}`, true}, + {"raw", `{"alpn": ["http/1.1"]}`, false}, + {"xdrive", `{"alpn": ["h2", "http/1.1"]}`, false}, + {"xdrive", `{"alpn": ["http/1.1"], "echConfigList": "udp://1.1.1.1"}`, false}, + {"xdrive", `{"alpn": ["http/1.1"], "fingerprint": "unsafe"}`, false}, + {"xdrive", `{"alpn": ["http/1.1"], "fingerprint": "hellogolang"}`, false}, + {"xdrive", `{"alpn": ["http/1.1"], "fingerprint": "randomized"}`, false}, + {"xdrive", `{"alpn": ["http/1.1"], "fingerprint": "hellorandomizednoalpn"}`, false}, + } { + lines = nil + config := new(StreamConfig) + if err := json.Unmarshal([]byte(`{ + "method": "`+c.method+`", + "security": "tls", + "tlsSettings": `+c.tlsSettings+`, + "xdriveSettings": {"remoteFolder": "/tmp/xdrive", "service": "local"} + }`), config); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + if _, err := config.Build(); err != nil { + t.Fatalf("%s %s: %v", c.method, c.tlsSettings, err) + } + warned := false + for _, line := range lines { + warned = warned || strings.HasPrefix(line, "[Warning]") && strings.Contains(line, `"alpn"`) + } + if warned != c.warn { + t.Errorf("%s %s: warned %v, want %v", c.method, c.tlsSettings, warned, c.warn) + } + } +} + func TestXDriveTemplateNeedsTemplate(t *testing.T) { config := new(XDriveConfig) if err := json.Unmarshal([]byte(`{"remoteFolder": "f", "service": "template"}`), config); err != nil { diff --git a/transport/internet/xdrive/client.go b/transport/internet/xdrive/client.go index 9cbd14607..00a3f1690 100644 --- a/transport/internet/xdrive/client.go +++ b/transport/internet/xdrive/client.go @@ -2,44 +2,124 @@ package xdrive import ( "context" - gotls "crypto/tls" + "io" "net/http" + "sync" "time" "github.com/xtls/xray-core/common/errors" "github.com/xtls/xray-core/common/net" "github.com/xtls/xray-core/transport/internet" - "github.com/xtls/xray-core/transport/internet/reality" "github.com/xtls/xray-core/transport/internet/tls" - "golang.org/x/net/http2" ) +var errStalled = errors.New("the request made no progress") + type serviceTransport struct { plain http.RoundTripper secure http.RoundTripper + stall time.Duration } +// RoundTrip ends a request that stands still for t.stall, with no answer that +// long after its body went out or no byte of either body, but not one that just +// runs long: a large object on a slow link takes minutes and is fine as long as +// its bytes move func (t *serviceTransport) RoundTrip(r *http.Request) (*http.Response, error) { + next := t.plain if r.URL.Scheme == "https" { - return t.secure.RoundTrip(r) + next = t.secure } - return t.plain.RoundTrip(r) + + ctx, cancel := context.WithCancelCause(r.Context()) + timer := time.AfterFunc(t.stall, func() { cancel(errStalled) }) + alive := func() { timer.Reset(t.stall) } + end := func() { + timer.Stop() + cancel(nil) + } + r = r.WithContext(ctx) + if r.Body != nil && r.Body != http.NoBody { + r.Body = &progressBody{ReadCloser: r.Body, alive: alive} + } + // a body sent again by the transport itself, after a GOAWAY or on a reused + // connection that failed, moves just the same + if getBody := r.GetBody; getBody != nil { + r.GetBody = func() (io.ReadCloser, error) { + body, err := getBody() + if err != nil || body == http.NoBody { + return body, err + } + return &progressBody{ReadCloser: body, alive: alive}, nil + } + } + + resp, err := next.RoundTrip(r) + if err != nil { + err = stalled(ctx, err) + end() + return nil, err + } + alive() + resp.Body = &progressBody{ReadCloser: resp.Body, alive: alive, end: end, ctx: ctx} + return resp, nil } -func newServiceClient(streamSettings *internet.MemoryStreamConfig, timeout time.Duration, maxConns int) *http.Client { +// stalled returns errStalled for an error that the stall gave, which http2 reports +// as a plain cancel +func stalled(ctx context.Context, err error) error { + if context.Cause(ctx) == errStalled { + return errStalled + } + return err +} + +type progressBody struct { + io.ReadCloser + alive func() + end func() + // ctx is that of the request, for the body of its response + ctx context.Context +} + +func (b *progressBody) Read(p []byte) (int, error) { + // http2 reads a request body up to 512 KiB at a time and sends it all before + // the next read, which on a slow link could look like no progress + n, err := b.ReadCloser.Read(p[:min(len(p), 16<<10)]) + if n > 0 { + b.alive() + } + if err != nil && err != io.EOF && b.ctx != nil { + err = stalled(b.ctx, err) + } + return n, err +} + +func (b *progressBody) Close() error { + err := b.ReadCloser.Close() + if b.end != nil { + b.end() + } + return err +} + +// 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 { var ( - tlsConfig *tls.Config - realityConfig *reality.Config - sockopt *internet.SocketConfig - fronting *net.Destination + tlsConfig *tls.Config + sockopt *internet.SocketConfig + fronting *net.Destination ) if streamSettings != nil { tlsConfig = tls.ConfigFromStreamSettings(streamSettings) - realityConfig = reality.ConfigFromStreamSettings(streamSettings) sockopt = streamSettings.SocketSettings fronting = streamSettings.Destination } - overHTTP2 := allowsHTTP2(tlsConfig, realityConfig) + 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) @@ -72,58 +152,40 @@ func newServiceClient(streamSettings *internet.MemoryStreamConfig, timeout time. return conn, err } - dialTLS := func(ctx context.Context, addr string) (net.Conn, error) { + // 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) if err != nil { return nil, err } - if realityConfig != nil { - return reality.UClient(conn, realityConfig, ctx, host) + gotlsConfig := tlsConfig.GetTLSConfig(tls.WithDestination(host)) + var tlsConn net.Conn + if fingerprint == nil { + tlsConn = tls.Client(conn, gotlsConfig) + } else { + tlsConn = tls.UClient(conn, gotlsConfig, fingerprint) } - - gotlsConfig := &gotls.Config{ServerName: host.Address.String()} - if tlsConfig != nil { - gotlsConfig = tlsConfig.GetTLSConfig(tls.WithDestination(host)) + // the dial outlives the request that started it, and a handshake that hangs + // would keep the one connection of the host to itself + ctx, cancel := context.WithTimeout(ctx, stall) + defer cancel() + if err := tlsConn.(tls.Interface).HandshakeContext(ctx); err != nil { + conn.Close() + return nil, err } - if len(gotlsConfig.NextProtos) != 1 { - if overHTTP2 { - gotlsConfig.NextProtos = []string{"h2"} - } else { - gotlsConfig.NextProtos = []string{"http/1.1"} - } - } - - if tlsConfig != nil { - if fingerprint := tls.GetFingerprint(tlsConfig.Fingerprint); fingerprint != nil { - uconn := tls.UClient(conn, gotlsConfig, fingerprint) - if err := uconn.(*tls.UConn).HandshakeContext(ctx); err != nil { - conn.Close() - return nil, err - } - return uconn, nil - } - } - return tls.Client(conn, gotlsConfig), nil + return tlsConn, nil } - var secure http.RoundTripper - if overHTTP2 { - secure = &http2.Transport{ - DialTLSContext: func(ctx context.Context, network, addr string, cfg *gotls.Config) (net.Conn, error) { - return dialTLS(ctx, addr) - }, - IdleConnTimeout: net.ConnIdleTimeout, - } - } else { - secure = &http.Transport{ - DialTLSContext: func(ctx context.Context, network, addr string) (net.Conn, error) { - return dialTLS(ctx, addr) - }, + secure := func(connsPerHost int) *http.Transport { + return &http.Transport{ + DialTLSContext: dialTLS, + ForceAttemptHTTP2: true, IdleConnTimeout: net.ConnIdleTimeout, MaxIdleConns: maxConns, MaxIdleConnsPerHost: maxConns, - MaxConnsPerHost: maxConns, + MaxConnsPerHost: connsPerHost, + HTTP2: &http.HTTP2Config{SendPingTimeout: net.ChromeH2KeepAlivePeriod}, } } @@ -136,18 +198,27 @@ func newServiceClient(streamSettings *internet.MemoryStreamConfig, timeout time. MaxIdleConnsPerHost: maxConns, MaxConnsPerHost: maxConns, }, - secure: secure, + secure: &secureTransport{h2: secure(1), h1: secure(maxConns)}, + stall: stall, }, - Timeout: timeout, } } -func allowsHTTP2(tlsConfig *tls.Config, realityConfig *reality.Config) bool { - if realityConfig != nil { - return true - } - if tlsConfig == nil { - return false - } - return len(tlsConfig.NextProtocol) == 1 && tlsConfig.NextProtocol[0] == "h2" +// 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 +} + +func (t *secureTransport) RoundTrip(r *http.Request) (*http.Response, error) { + if _, ok := t.h1Hosts.Load(r.URL.Host); ok { + return t.h1.RoundTrip(r) + } + resp, err := t.h2.RoundTrip(r) + if err == nil && resp.ProtoMajor == 1 { + t.h1Hosts.Store(r.URL.Host, true) + } + return resp, err } diff --git a/transport/internet/xdrive/client_test.go b/transport/internet/xdrive/client_test.go index 44e059a1b..62187de01 100644 --- a/transport/internet/xdrive/client_test.go +++ b/transport/internet/xdrive/client_test.go @@ -1,11 +1,18 @@ package xdrive import ( + "bytes" "context" gotls "crypto/tls" + "encoding/base64" + "errors" + "io" "net" "net/http" + "net/http/httptest" + "slices" "sync" + "sync/atomic" "testing" "time" @@ -99,3 +106,403 @@ func TestServiceSNIOverride(t *testing.T) { t.Fatalf("explicit serverName gave SNI %q, want www.google.com", got) } } + +// slowReader hands out one byte every gap, the way a body crosses a slow link +type slowReader struct { + left int + gap time.Duration +} + +func (r *slowReader) Read(p []byte) (int, error) { + if r.left == 0 { + return 0, io.EOF + } + time.Sleep(r.gap) + r.left-- + p[0] = 'x' + return 1, nil +} + +// a request that keeps moving outlives the stall limit many times over, one +// that stands still does not get much past it +func TestServiceClientEndsOnlyStalledRequests(t *testing.T) { + const stall = 200 * time.Millisecond + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + switch r.URL.Path { + case "/slow": + for i := 0; i < 20; i++ { + w.Write([]byte("x")) + w.(http.Flusher).Flush() + time.Sleep(stall / 4) + } + case "/stuck": + w.Write([]byte("x")) + w.(http.Flusher).Flush() + time.Sleep(5 * stall) + case "/silent": + time.Sleep(5 * stall) + } + })) + defer server.Close() + client := newServiceClient(&internet.MemoryStreamConfig{}, stall, 8) + + fetch := func(path string, body io.Reader) (int, error) { + req, err := http.NewRequest(http.MethodPost, server.URL+path, body) + if err != nil { + t.Fatal(err) + } + resp, err := client.Do(req) + if err != nil { + return 0, err + } + defer resp.Body.Close() + got, err := io.ReadAll(resp.Body) + return len(got), err + } + + if _, err := fetch("/", &slowReader{left: 20, gap: stall / 4}); err != nil { + t.Errorf("a slow upload failed: %v", err) + } + if n, err := fetch("/slow", nil); err != nil || n != 20 { + t.Errorf("a slow download gave %d bytes, %v", n, err) + } + for _, path := range []string{"/stuck", "/silent"} { + start := time.Now() + _, err := fetch(path, nil) + 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) + } + } + + // a body that the transport sends again by itself keeps the request going too + replay := &serviceTransport{stall: stall, plain: roundTrip(func(r *http.Request) (*http.Response, error) { + body, err := r.GetBody() + if err != nil { + return nil, err + } + for b := make([]byte, 1); ; time.Sleep(stall / 4) { + if _, err := body.Read(b); err != nil { + break + } + } + if err := context.Cause(r.Context()); err != nil { + return nil, err + } + return &http.Response{StatusCode: http.StatusOK, Body: http.NoBody}, nil + })} + req, err := http.NewRequest(http.MethodPost, "http://drive/", bytes.NewReader(make([]byte, 20))) + if err != nil { + t.Fatal(err) + } + if resp, err := replay.RoundTrip(req); err != nil { + t.Errorf("a body sent again failed: %v", err) + } else { + resp.Body.Close() + } +} + +// over TLS a request that stands still ends the same way, and says so, on h2 +// as on h1 +func TestServiceClientStallOverTLS(t *testing.T) { + const stall = 200 * time.Millisecond + for _, protos := range [][]string{chromeALPN, {"http/1.1"}} { + server := newTLSServer(t, &gotls.Config{NextProtos: protos}, 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(server.settings(&tls.Config{}), stall, 8) + + for _, path := range []string{"/stuck", "/silent"} { + start := time.Now() + resp, err := client.Get(server.URL + path) + if err == nil { + if resp.Proto != picked(protos) { + t.Fatalf("%s went over %s", path, resp.Proto) + } + _, err = io.ReadAll(resp.Body) + resp.Body.Close() + } + if !errors.Is(err, errStalled) { + t.Errorf("server %v: %s gave %v, want %v", protos, path, err, errStalled) + } + if took := time.Since(start); took > 4*stall { + t.Errorf("server %v: %s took %v to give up", protos, path, took) + } + } + } +} + +// tlsServer is a TLS server that keeps the last hello it got and counts its +// connections +type tlsServer struct { + *httptest.Server + mu sync.Mutex + hello *gotls.ClientHelloInfo + conns atomic.Int32 +} + +func newTLSServer(t *testing.T, config *gotls.Config, handler http.HandlerFunc) *tlsServer { + t.Helper() + + s := &tlsServer{Server: httptest.NewUnstartedServer(handler)} + config.GetConfigForClient = func(hello *gotls.ClientHelloInfo) (*gotls.Config, error) { + s.mu.Lock() + s.hello = hello + s.mu.Unlock() + return nil, nil + } + s.TLS = config + s.Config.ConnState = func(_ net.Conn, state http.ConnState) { + if state == http.StateNew { + s.conns.Add(1) + } + } + s.StartTLS() + t.Cleanup(s.Close) + return s +} + +func (s *tlsServer) lastHello() *gotls.ClientHelloInfo { + s.mu.Lock() + defer s.mu.Unlock() + return s.hello +} + +// settings pins the certificate of the server in config +func (s *tlsServer) settings(config *tls.Config) *internet.MemoryStreamConfig { + config.PinnedPeerCertSha256 = [][]byte{tls.GenerateCertHash(s.Certificate())} + return &internet.MemoryStreamConfig{ProtocolName: protocolName, SecuritySettings: config} +} + +// picked is the protocol a server with ALPN protos answers a Chrome hello with +func picked(protos []string) string { + if protos[0] == "h2" { + return "HTTP/2.0" + } + return "HTTP/1.1" +} + +var chromeALPN = []string{"h2", "http/1.1"} + +// the hello offers the ALPN of the fingerprint whatever "alpn" says, or with Go +// TLS that of the config, h2 and http/1.1 by default, and the client speaks what +// the server picks from it +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) + resp, err := client.Get(server.URL) + if err != nil { + t.Fatalf("server %v, alpn %v, fingerprint %q: %v", protos, config.NextProtocol, config.Fingerprint, err) + } + resp.Body.Close() + if resp.Proto != picked(protos) { + t.Errorf("server %v, alpn %v, fingerprint %q: went over %s", protos, config.NextProtocol, config.Fingerprint, resp.Proto) + } + if offered := server.lastHello().SupportedProtos; !slices.Equal(offered, chromeALPN) { + t.Errorf("server %v, alpn %v, fingerprint %q: the hello offered %v, want %v", protos, config.NextProtocol, config.Fingerprint, offered, chromeALPN) + } + } + } +} + +// with no tlsSettings the hello is that of Chrome too, which unlike Go TLS +// 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) + // 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() + } + hello := server.lastHello() + if hello == nil { + t.Fatal("no hello") + } + if !slices.Equal(hello.SupportedProtos, chromeALPN) || !slices.Contains(hello.Extensions, 27) { + t.Errorf("the hello offered %v with extensions %v, not like Chrome", hello.SupportedProtos, hello.Extensions) + } +} + +// 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) { + server := newTLSServer(t, &gotls.Config{NextProtos: chromeALPN}, func(w http.ResponseWriter, r *http.Request) { + 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) + get := func(url string) { + resp, err := client.Get(url) + if err != nil { + t.Error(err) + return + } + resp.Body.Close() + } + + for i := range 2 { + if i == 1 { + get(h1.URL) + } + var wg sync.WaitGroup + for range driveMaxInflight { + wg.Go(func() { get(server.URL) }) + } + wg.Wait() + if n := server.conns.Load(); n != 1 { + t.Errorf("round %d: %d parallel requests took %d connections", i, driveMaxInflight, n) + } + } +} + +// in the format of xray tls ech, for public name public.example +const ( + testECHConfigList = "AEH+DQA9AAAgACBziE85FLxJySxZAPRHs8CDCbuz1h76I9pLhctC/kgScAAEAAEAAQAOcHVibGljLmV4YW1wbGUAAA==" + testECHServerKeys = "ACD2AjC5yuBQ3430geNDLdIRndloCry/DBb9ZeYp0TmeowBB/g0APQAAIAAgc4hPORS8ScksWQD0R7PAgwm7s9Ye+iPaS4XLQv5IEnAABAABAAEADnB1YmxpYy5leGFtcGxlAAA=" +) + +// a server that speaks only h1, or gets ALPN http/1.1 inside ECH or from Go TLS, +// answers over h1, and after that answer parallel requests get more than one +// connection +func TestServiceClientHTTP1(t *testing.T) { + keys, err := base64.StdEncoding.DecodeString(testECHServerKeys) + if err != nil { + t.Fatal(err) + } + echKeys, err := tls.ConvertToGoECHKeys(keys) + if err != nil { + t.Fatal(err) + } + ech := &tls.Config{ServerName: "example.com", NextProtocol: []string{"http/1.1"}, EchConfigList: testECHConfigList} + + for _, c := range []struct { + protos, seen []string + config *tls.Config + }{ + {[]string{"http/1.1"}, chromeALPN, &tls.Config{}}, + {chromeALPN, []string{"http/1.1"}, ech}, + {chromeALPN, []string{"http/1.1"}, &tls.Config{NextProtocol: []string{"http/1.1"}, Fingerprint: "unsafe"}}, + } { + 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) + get := func() { + resp, err := client.Get(server.URL) + if err != nil { + t.Error(err) + return + } + resp.Body.Close() + if resp.Proto != "HTTP/1.1" || resp.TLS.ECHAccepted != (c.config.EchConfigList != "") { + t.Errorf("server %v, ALPN %v: went over %s, ECH accepted %v", c.protos, c.seen, resp.Proto, resp.TLS.ECHAccepted) + } + } + + get() + first := server.conns.Load() + var wg sync.WaitGroup + for range 4 { + wg.Go(get) + } + wg.Wait() + if offered := server.lastHello().SupportedProtos; !slices.Equal(offered, c.seen) { + t.Errorf("server %v: saw ALPN %v, want %v", c.protos, offered, c.seen) + } + if n := server.conns.Load() - first; n < 2 { + t.Errorf("server %v, ALPN %v: 4 parallel requests over h1 took %d connections", c.protos, c.seen, n) + } + } + + // 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 { + resp.Body.Close() + } + if hello := outer.lastHello(); hello == nil { + t.Error("no outer hello") + } else if !slices.Equal(hello.SupportedProtos, chromeALPN) { + t.Errorf("the outer hello offered %v, want %v", hello.SupportedProtos, chromeALPN) + } +} + +// muteFirst leaves the first connection it accepts unanswered +type muteFirst struct { + net.Listener + muted net.Conn +} + +func (l *muteFirst) Accept() (net.Conn, error) { + if l.muted == nil { + conn, err := l.Listener.Accept() + if err != nil { + return nil, err + } + l.muted = conn + } + return l.Listener.Accept() +} + +// a handshake that hangs ends after stall, and does not keep the host from +// getting another connection +func TestServiceClientHungHandshake(t *testing.T) { + const stall = 200 * time.Millisecond + for _, fingerprint := range []string{"", "unsafe"} { + server := &tlsServer{Server: httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))} + server.Listener = &muteFirst{Listener: server.Listener} + server.StartTLS() + t.Cleanup(server.Close) + client := newServiceClient(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) + } + for i := range 2 { + resp, err := client.Get(server.URL) + if err != nil { + t.Fatalf("fingerprint %q: request %d after a handshake that hung: %v", fingerprint, i, err) + } + resp.Body.Close() + } + } +} + +// a connection whose handshake failed is closed, not left to the garbage +// collector +func TestServiceClientClosesFailedHandshake(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + + 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()) + conn, err := ln.Accept() + if err != nil { + t.Fatal(err) + } + // not TLS, so the handshake fails at once + conn.Write([]byte("HTTP/1.1 400 Bad Request\r\n\r\n")) + conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + if _, err := io.Copy(io.Discard, conn); err != nil { + t.Errorf("fingerprint %q: the client left the connection open: %v", fingerprint, err) + } + conn.Close() + } +} + +type roundTrip func(*http.Request) (*http.Response, error) + +func (f roundTrip) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } diff --git a/transport/internet/xdrive/drive.go b/transport/internet/xdrive/drive.go index d5da0cfb4..7a2403386 100644 --- a/transport/internet/xdrive/drive.go +++ b/transport/internet/xdrive/drive.go @@ -26,7 +26,8 @@ const ( driveMaxAttempts = 8 driveMaxInflight = 32 driveInlineLimit = 12000 - driveTimeout = 60 * time.Second + // driveTimeout ends a request that makes no progress for that long + driveTimeout = 60 * time.Second ) var (