Refactor

The relay session context was derived from the inbound connection of the
source address that created the session. After a client changes its source
address (NAT rebinding), the UDP worker of the old address is cleaned up after
~2 minutes of inactivity and cancels that context, which tears down the
dispatched outbound flow; the next packet then creates a brand-new session and
outbound flow (new local port on the server, reset replay window).

Sessions are routed by client session ID (SIP022 3.2.4), so detach the session
context from the originating connection's cancellation and let the session's
own inactivity timer govern its lifetime.

app/observatory/burst/healthping_result.go

@@ -109,7 +109,7 @@ func (h *HealthPingRTTS) getStatistics() *HealthPingStats {
 		stats.Min = 0
 		return stats
 	}
-	stats.Average = time.Duration(int(sum) / cnt)
+	stats.Average = sum / time.Duration(cnt)
 	var std float64
 	if cnt < 2 {
 		// no enough data for standard deviation, we assume it's half of the average rtt

core/core.go

@@ -20,7 +20,7 @@ import (
 var (
 	Version_x byte = 26
 	Version_y byte = 10
-	Version_z byte = 10
+	Version_z byte = 17
 )
 
 var (

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.")

infra/conf/transport_method.go

@@ -1004,6 +1004,10 @@ func (c *XDriveConfig) Build() (proto.Message, error) {
 	switch c.Service {
 	case "local":
 	case "Google Drive":
+		errors.PrintDeprecatedFeatureWarning(`"service": "Google Drive"`, `"service": "google-drive"`)
+		c.Service = "google-drive"
+		fallthrough
+	case "google-drive":
 		if len(c.Secrets) != 3 {
 			return nil, errors.New("Google Drive needs 3 secrets in order of ClientID, ClientSecret, RefreshToken")
 		}
@@ -1012,7 +1016,7 @@ func (c *XDriveConfig) Build() (proto.Message, error) {
 			return nil, errors.New(`service "template" needs a "template" object`)
 		}
 	default:
-		return nil, errors.New("unsupported service")
+		return nil, errors.New(`unknown "service": "`, c.Service, `", use "google-drive", "template" or "local"`)
 	}
 	config := &xdrive.Config{
 		RemoteFolder:      c.RemoteFolder,

infra/conf/transport_test.go

@@ -5,9 +5,11 @@ 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"
+	"github.com/xtls/xray-core/transport/internet/xdrive"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -321,8 +323,25 @@ func TestXDriveRejectsUnknownService(t *testing.T) {
 	if err := json.Unmarshal([]byte(`{"remoteFolder": "/tmp/xdrive", "service": "Dropbox"}`), config); err != nil {
 		t.Fatalf("Unmarshal: %v", err)
 	}
-	if _, err := config.Build(); err == nil {
-		t.Fatal("Build accepted an unsupported service")
+	if _, err := config.Build(); err == nil || !strings.Contains(err.Error(), `use "google-drive", "template" or "local"`) {
+		t.Fatalf("expected an unknown service error, got %v", err)
+	}
+}
+
+func TestXDriveGoogleDriveName(t *testing.T) {
+	for _, service := range []string{"google-drive", "Google Drive"} {
+		config := &XDriveConfig{RemoteFolder: "folder-id", Service: service, Secrets: []string{"a", "b", "c"}}
+		built, err := config.Build()
+		if err != nil {
+			t.Fatalf("%s: %v", service, err)
+		}
+		if got := built.(*xdrive.Config).Service; got != "google-drive" {
+			t.Errorf("%s: service is %q, want google-drive", service, got)
+		}
+		config = &XDriveConfig{RemoteFolder: "folder-id", Service: service, Secrets: []string{"a", "b"}}
+		if _, err := config.Build(); err == nil {
+			t.Errorf("%s: accepted two secrets", service)
+		}
 	}
 }
 
@@ -355,6 +374,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 {
@@ -364,3 +433,83 @@ func TestXDriveTemplateNeedsTemplate(t *testing.T) {
 		t.Fatal("Build accepted a template service without a template")
 	}
 }
+
+const (
+	xdriveStream = `"streamSettings": {"network": "xdrive", "xdriveSettings": {"remoteFolder": "f", "service": "local"}}`
+	xdriveID     = `"id": "27848739-7e62-4138-9fd3-098a63964b6b"`
+	xdriveServer = `"address": "127.0.0.1", "port": 443, ` + xdriveID
+	xdriveEnc    = `"encryption": "mlkem768x25519plus.native.0rtt.XKbaq1BuzJKKKPxZbwgRpn-gYQbnhrYketgezASCL1I"`
+	xdriveDec    = `"decryption": "mlkem768x25519plus.native.600s.UKfUEMDdjlEnycTMBP0a08acky8vRRKSrM1MRnaYk3s"`
+)
+
+func TestXDriveOutboundConfig(t *testing.T) {
+	build := func(s string) error {
+		c := new(OutboundDetourConfig)
+		if err := json.Unmarshal([]byte(s), c); err != nil {
+			return err
+		}
+		_, err := c.Build()
+		return err
+	}
+
+	for _, input := range []string{
+		`{"protocol": "vless", "settings": {` + xdriveServer + `, ` + xdriveEnc + `}, ` + xdriveStream + `, "mux": {"enabled": true}}`,
+		`{"protocol": "vless", "settings": {` + xdriveServer + `, ` + xdriveEnc + `}, ` + xdriveStream + `, "mux": {"enabled": true, "concurrency": 0, "xudpConcurrency": 8, "xudpProxyUDP443": "allow"}}`,
+		`{"protocol": "vless", "settings": {` + xdriveServer + `, ` + xdriveEnc + `, "reverse": {"tag": "r"}}, ` + xdriveStream + `}`,
+		`{"protocol": "vless", "settings": {"vnext": [{"address": "127.0.0.1", "port": 443, "users": [{` + xdriveID + `, ` + xdriveEnc + `}]}]}, ` + xdriveStream + `, "mux": {"enabled": true}}`,
+	} {
+		if err := build(input); err != nil {
+			t.Errorf("%s: %v", input, err)
+		}
+	}
+	for _, c := range []struct{ input, err string }{
+		{`{"protocol": "vless", "settings": {` + xdriveServer + `, "encryption": "none"}, ` + xdriveStream + `, "mux": {"enabled": true}}`, `"encryption"`},
+		{`{"protocol": "vless", "settings": {"address": "example.com", "port": 443, ` + xdriveID + `, "encryption": "none"}, ` + xdriveStream + `, "mux": {"enabled": true}}`, `"encryption"`},
+		{`{"protocol": "vless", "settings": {` + xdriveServer + `, ` + xdriveEnc + `}, ` + xdriveStream + `}`, `"mux"`},
+		{`{"protocol": "vless", "settings": {` + xdriveServer + `, ` + xdriveEnc + `}, ` + xdriveStream + `, "mux": {"enabled": false}}`, `"mux"`},
+		{`{"protocol": "vless", "settings": {` + xdriveServer + `, ` + xdriveEnc + `}, ` + xdriveStream + `, "mux": {"enabled": true, "concurrency": -1}}`, `"mux"`},
+		{`{"protocol": "vless", "settings": {` + xdriveServer + `, ` + xdriveEnc + `}, ` + xdriveStream + `, "mux": {"enabled": true, "xudpConcurrency": -1}}`, `"mux"`},
+		{`{"protocol": "vless", "settings": {` + xdriveServer + `, ` + xdriveEnc + `}, ` + xdriveStream + `, "mux": {"enabled": true, "xudpProxyUDP443": "skip"}}`, `"mux"`},
+		{`{"protocol": "vless", "settings": {` + xdriveServer + `, ` + xdriveEnc + `, "flow": "xtls-rprx-vision"}, ` + xdriveStream + `, "mux": {"enabled": true, "concurrency": -1}}`, `"flow"`},
+		{`{"protocol": "vless", "settings": {"vnext": [{"address": "127.0.0.1", "port": 443, "users": [{` + xdriveID + `, ` + xdriveEnc + `, "flow": "xtls-rprx-vision"}]}]}, ` + xdriveStream + `, "mux": {"enabled": true, "concurrency": -1}}`, `"flow"`},
+		{`{"protocol": "vless", "settings": {` + xdriveServer + `, ` + xdriveEnc + `, "testpre": 1}, ` + xdriveStream + `, "mux": {"enabled": true}}`, `"testpre"`},
+		{`{"protocol": "vmess", "settings": {` + xdriveServer + `}, ` + xdriveStream + `, "mux": {"enabled": true}}`, "VLESS"},
+		{`{"protocol": "trojan", "settings": {"address": "example.com", "port": 443, "password": "p"}, ` + xdriveStream + `, "mux": {"enabled": true}}`, "VLESS"},
+		{`{"protocol": "freedom", ` + xdriveStream + `}`, "VLESS"},
+	} {
+		if err := build(c.input); err == nil || !strings.Contains(err.Error(), "xdrive transport") || !strings.Contains(err.Error(), c.err) {
+			t.Errorf("expected an xdrive error about %s for %s, got %v", c.err, c.input, err)
+		}
+	}
+}
+
+func TestXDriveInboundConfig(t *testing.T) {
+	build := func(s string) error {
+		c := new(InboundDetourConfig)
+		if err := json.Unmarshal([]byte(s), c); err != nil {
+			return err
+		}
+		_, err := c.Build()
+		return err
+	}
+
+	for _, input := range []string{
+		`{"protocol": "vless", "port": 443, "settings": {"clients": [{` + xdriveID + `}], ` + xdriveDec + `}, ` + xdriveStream + `}`,
+		`{"protocol": "vless", "listen": "@xdrive", "settings": {"clients": [{` + xdriveID + `}], ` + xdriveDec + `}, ` + xdriveStream + `}`,
+	} {
+		if err := build(input); err != nil {
+			t.Errorf("%s: %v", input, err)
+		}
+	}
+	for _, c := range []struct{ input, err string }{
+		{`{"protocol": "vless", "port": 443, "settings": {"clients": [{` + xdriveID + `}], "decryption": "none"}, ` + xdriveStream + `}`, `"decryption"`},
+		{`{"protocol": "vless", "port": 443, "settings": {"clients": [{` + xdriveID + `}], ` + xdriveDec + `, "flow": "xtls-rprx-vision"}, ` + xdriveStream + `}`, `"flow"`},
+		{`{"protocol": "vless", "port": 443, "settings": {"clients": [{` + xdriveID + `, "flow": "xtls-rprx-vision"}], ` + xdriveDec + `}, ` + xdriveStream + `}`, `"flow"`},
+		{`{"protocol": "vmess", "port": 443, "settings": {"clients": [{` + xdriveID + `}]}, ` + xdriveStream + `}`, "VLESS"},
+		{`{"protocol": "vless", "port": "10800-10803", "settings": {"clients": [{` + xdriveID + `}], ` + xdriveDec + `}, ` + xdriveStream + `}`, "one port"},
+	} {
+		if err := build(c.input); err == nil || !strings.Contains(err.Error(), "xdrive transport") || !strings.Contains(err.Error(), c.err) {
+			t.Errorf("expected an xdrive error about %s for %s, got %v", c.err, c.input, err)
+		}
+	}
+}

infra/conf/xray.go

@@ -17,7 +17,11 @@ import (
 	core "github.com/xtls/xray-core/core"
 	"github.com/xtls/xray-core/proxy/freedom"
 	"github.com/xtls/xray-core/proxy/masque"
+	"github.com/xtls/xray-core/proxy/vless"
+	"github.com/xtls/xray-core/proxy/vless/inbound"
+	"github.com/xtls/xray-core/proxy/vless/outbound"
 	"github.com/xtls/xray-core/transport/internet"
+	"google.golang.org/protobuf/proto"
 )
 
 var (
@@ -209,6 +213,9 @@ func (c *InboundDetourConfig) Build() (*core.InboundHandlerConfig, error) {
 	if _, ok := ts.(*masque.ServerConfig); !ok && receiverSettings.StreamSettings != nil && receiverSettings.StreamSettings.ProtocolName == "masque" {
 		return nil, errors.New("the masque transport can only be used by the masque inbound")
 	}
+	if err := validateXDrive(ts, receiverSettings.StreamSettings, nil, receiverSettings.PortList); err != nil {
+		return nil, err
+	}
 
 	return &core.InboundHandlerConfig{
 		Tag:              c.Tag,
@@ -262,6 +269,54 @@ func validateOutboundTransportSecurity(rawConfig interface{}, senderSettings *pr
 	return nil
 }
 
+// validateXDrive allows only VLESS with encryption on xdrive, and on the client Mux.Cool for every connection
+func validateXDrive(ts proto.Message, ss *internet.StreamConfig, ms *proxyman.MultiplexingConfig, pl *net.PortList) error {
+	if ss == nil || ss.ProtocolName != "xdrive" {
+		return nil
+	}
+	switch c := ts.(type) {
+	case *outbound.Config:
+		a, err := c.Vnext.User.Account.GetInstance()
+		if err != nil {
+			return err
+		}
+		account := a.(*vless.Account)
+		if account.Flow != "" {
+			return errors.New(`the xdrive transport doesn't support VLESS "flow"`)
+		}
+		if account.Encryption == "none" {
+			return errors.New(`the xdrive transport needs VLESS "encryption", run "xray vlessenc" to get one`)
+		}
+		// reverse opens its own mux
+		if account.Reverse == nil && (ms == nil || !ms.Enabled || ms.Concurrency < 0 || ms.XudpConcurrency < 0 || ms.XudpProxyUDP443 == "skip") {
+			return errors.New(`the xdrive transport needs "mux" enabled, with no negative "concurrency" or "xudpConcurrency" and no "xudpProxyUDP443":"skip"`)
+		}
+		if account.Testpre != 0 {
+			return errors.New(`the xdrive transport doesn't support VLESS "testpre"`)
+		}
+	case *inbound.Config:
+		if c.Decryption == "none" {
+			return errors.New(`the xdrive transport needs VLESS "decryption", run "xray vlessenc" to get one`)
+		}
+		for _, user := range c.Users {
+			a, err := user.Account.GetInstance()
+			if err != nil {
+				return err
+			}
+			if a.(*vless.Account).Flow != "" {
+				return errors.New(`the xdrive transport doesn't support VLESS "flow"`)
+			}
+		}
+		// every port gets its own listener on the same folder, and each of them accepts the same session
+		if pl != nil && len(pl.Ports()) != 1 {
+			return errors.New("the xdrive transport can only listen on one port")
+		}
+	default:
+		return errors.New("the xdrive transport can only be used by VLESS")
+	}
+	return nil
+}
+
 // Build implements Buildable.
 func (c *OutboundDetourConfig) Build() (*core.OutboundHandlerConfig, error) {
 	if c.ProxySettings != nil {
@@ -340,6 +395,9 @@ func (c *OutboundDetourConfig) Build() (*core.OutboundHandlerConfig, error) {
 	if err != nil {
 		return nil, errors.New("failed to build outbound handler for protocol ", c.Protocol).Base(err)
 	}
+	if err := validateXDrive(ts, senderSettings.StreamSettings, senderSettings.MultiplexSettings, nil); err != nil {
+		return nil, err
+	}
 	if err := validateOutboundTransportSecurity(rawConfig, senderSettings); err != nil {
 		return nil, err
 	}

proxy/proxy.go

@@ -180,14 +180,14 @@ type VisionReader struct {
 	isUplink     bool
 	conn         net.Conn
 	input        *bytes.Reader
-	rawInput     *bytes.Buffer
+	rawInput     **bytes.Buffer
 	ob           *session.Outbound
 
 	// internal
 	directReadCounter stats.Counter
 }
 
-func NewVisionReader(reader buf.Reader, trafficState *TrafficState, isUplink bool, ctx context.Context, conn net.Conn, input *bytes.Reader, rawInput *bytes.Buffer, ob *session.Outbound) *VisionReader {
+func NewVisionReader(reader buf.Reader, trafficState *TrafficState, isUplink bool, ctx context.Context, conn net.Conn, input *bytes.Reader, rawInput **bytes.Buffer, ob *session.Outbound) *VisionReader {
 	return &VisionReader{
 		Reader:       reader,
 		trafficState: trafficState,
@@ -261,12 +261,15 @@ func (w *VisionReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
 		if inputBuffer, err := buf.ReadFrom(w.input); err == nil && !inputBuffer.IsEmpty() {
 			buffer, _ = buf.MergeMulti(buffer, inputBuffer)
 		}
-		if rawInputBuffer, err := buf.ReadFrom(w.rawInput); err == nil && !rawInputBuffer.IsEmpty() {
-			buffer, _ = buf.MergeMulti(buffer, rawInputBuffer)
-		}
 		*w.input = bytes.Reader{} // release memory
 		w.input = nil
-		*w.rawInput = bytes.Buffer{} // release memory
+
+		if raw := *w.rawInput; raw != nil {
+			if rawInputBuffer, err := buf.ReadFrom(raw); err == nil && !rawInputBuffer.IsEmpty() {
+				buffer, _ = buf.MergeMulti(buffer, rawInputBuffer)
+			}
+			// *raw = bytes.Buffer{} // may not be suitable for Go 1.27.3+
+		}
 		w.rawInput = nil
 
 		if inbound := session.InboundFromContext(w.ctx); inbound != nil && inbound.Conn != nil {

proxy/shadowsocks_2022/inbound.go

@@ -53,21 +53,23 @@ func NewServer(ctx context.Context, config *ServerConfig) (*Inbound, error) {
 		return nil, err
 	}
 
-	udpCodec, err := NewUDPServerCodec(method, psk, 500*time.Second)
+	user := &protocol.MemoryUser{
+		Email: config.Email,
+		Level: uint32(config.Level),
+	}
+
+	udpCodec, err := NewUDPServerCodec(method, psk, user, 500*time.Second)
 	if err != nil {
 		return nil, err
 	}
 
 	v := core.MustFromContext(ctx)
 	return &Inbound{
-		networks:   networks,
-		method:     method,
-		psk:        psk,
-		saltFilter: antireplay.NewMapFilter[[32]byte](60),
-		user: &protocol.MemoryUser{
-			Email: config.Email,
-			Level: uint32(config.Level),
-		},
+		networks:      networks,
+		method:        method,
+		psk:           psk,
+		saltFilter:    antireplay.NewMapFilter[[32]byte](60),
+		user:          user,
 		udpCodec:      udpCodec,
 		policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager),
 	}, nil
@@ -155,20 +157,12 @@ func (i *Inbound) processUDP(ctx context.Context, conn stat.Connection, dispatch
 		}
 
 		for _, b := range mb {
-			decoded, err := i.udpCodec.DecodePacket(b.Bytes())
+			sessionItem, decoded, err := i.udpCodec.DecodePacket(b.Bytes())
 			b.Release()
 			if err != nil || decoded.HeaderType != HeaderTypeClient {
 				continue
 			}
 
-			sessionItem := i.udpCodec.GetSession(decoded.SessionID)
-			if sessionItem.User == nil {
-				sessionItem.Lock()
-				if sessionItem.User == nil {
-					sessionItem.User = i.user
-				}
-				sessionItem.Unlock()
-			}
 			link := sessionItem.link.Load()
 			if link != nil {
 				sessionItem.UpdateConn(conn)

proxy/shadowsocks_2022/inbound_multi.go

@@ -2,8 +2,6 @@ package shadowsocks_2022
 
 import (
 	"context"
-	"crypto/cipher"
-	"encoding/binary"
 	"strconv"
 	"strings"
 	"sync"
@@ -41,8 +39,7 @@ type MultiUserInbound struct {
 	usersByEmail    *utils.TypedSyncMap[string, *protocol.MemoryUser]
 	userCount       atomic.Int64
 	saltFilter      *antireplay.ReplayFilter[[32]byte]
-	udpSessions     *UDPSessionManager
-	udpMasterCipher cipher.Block
+	udpCodec        *UDPMultiServerCodec
 	policyManager   policy.Manager
 }
 
@@ -68,22 +65,21 @@ func NewMultiServer(ctx context.Context, config *MultiUserServerConfig) (*MultiU
 		return nil, err
 	}
 
-	masterBlock, err := method.NewBlock(masterPSK)
+	udpCodec, err := NewUDPMultiServerCodec(method, masterPSK, 500*time.Second)
 	if err != nil {
 		return nil, err
 	}
 
 	v := core.MustFromContext(ctx)
 	i := &MultiUserInbound{
-		networks:        networks,
-		method:          method,
-		masterPSK:       masterPSK,
-		usersByHash:     utils.NewTypedSyncMap[[AESBlockSize]byte, *protocol.MemoryUser](),
-		usersByEmail:    utils.NewTypedSyncMap[string, *protocol.MemoryUser](),
-		saltFilter:      antireplay.NewMapFilter[[32]byte](60),
-		udpSessions:     NewUDPSessionManager(500 * time.Second),
-		udpMasterCipher: masterBlock,
-		policyManager:   v.GetFeature(policy.ManagerType()).(policy.Manager),
+		networks:      networks,
+		method:        method,
+		masterPSK:     masterPSK,
+		usersByHash:   utils.NewTypedSyncMap[[AESBlockSize]byte, *protocol.MemoryUser](),
+		usersByEmail:  utils.NewTypedSyncMap[string, *protocol.MemoryUser](),
+		saltFilter:    antireplay.NewMapFilter[[32]byte](60),
+		udpCodec:      udpCodec,
+		policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager),
 	}
 
 	for idx, user := range config.Users {
@@ -281,68 +277,26 @@ func (i *MultiUserInbound) processUDP(ctx context.Context, conn stat.Connection,
 		}
 
 		for _, b := range mb {
-			// In multi-user UDP:
-			// Packet header is 16 bytes: Encrypted(SessionID + PacketID)
-			// Followed by 16 bytes EIH
-			packetBytes := b.Bytes()
-			if len(packetBytes) < 32+1+8+2 {
-				b.Release()
-				continue
-			}
-
-			var rawHeader [16]byte
-			i.udpMasterCipher.Decrypt(rawHeader[:], packetBytes[:16])
-
-			sessionID := binary.BigEndian.Uint64(rawHeader[:8])
-			packetID := binary.BigEndian.Uint64(rawHeader[8:16])
-
-			sessionItem := i.udpSessions.GetOrCreate(sessionID)
-
-			if !sessionItem.CheckPacketID(packetID) {
-				b.Release()
-				continue
-			}
-
-			var userPSK []byte
-			var currentUser *protocol.MemoryUser
-			sessionItem.Lock()
-			currentUser = sessionItem.User
-			userPSK = sessionItem.UserPSK
-			sessionItem.Unlock()
-
-			if currentUser == nil {
-				// Decrypt EIH
-				decryptedHash := DecryptUDPEIH(i.udpMasterCipher, rawHeader[:], packetBytes[16:32])
-
-				user, ok := i.usersByHash.Load(decryptedHash)
+			sessionItem, decoded, user, err := i.udpCodec.DecodePacket(b.Bytes(), func(hash [AESBlockSize]byte) (*protocol.MemoryUser, []byte, bool) {
+				u, ok := i.usersByHash.Load(hash)
 				if !ok {
-					b.Release()
-					continue
+					return nil, nil, false
 				}
-				currentUser = user
-				userPSK = user.Account.(*MemoryAccount).Key
-			}
-
-			decoded, err := sessionItem.DecryptAESPayload(i.method, userPSK, sessionID, packetID, rawHeader[:], packetBytes[32:])
+				return u, u.Account.(*MemoryAccount).Key, true
+			})
 			b.Release()
-			if err != nil {
+			if err != nil || decoded.HeaderType != HeaderTypeClient {
 				continue
 			}
 
-			sessionItem.Lock()
-			if sessionItem.User == nil {
-				sessionItem.User = currentUser
-				sessionItem.UserPSK = userPSK
-			}
-			sessionItem.Unlock()
-
 			link := sessionItem.link.Load()
 			if link != nil {
 				sessionItem.UpdateConn(conn)
 			} else {
+				userPSK := user.Account.(*MemoryAccount).Key
 				var err error
 				link, err = sessionItem.EnsureLink(ctx, conn, decoded.Destination, dispatcher, i.policyManager, func(replyDest net.Destination, payload []byte) ([]byte, error) {
-					return i.encodeServerUDPPacket(sessionID, userPSK, replyDest, payload)
+					return i.udpCodec.EncodeServerPacket(decoded.SessionID, userPSK, replyDest, payload)
 				})
 				if err != nil {
 					continue
@@ -356,7 +310,3 @@ func (i *MultiUserInbound) processUDP(ctx context.Context, conn stat.Connection,
 		}
 	}
 }
-
-func (i *MultiUserInbound) encodeServerUDPPacket(clientSessionID uint64, userPSK []byte, dest net.Destination, payload []byte) ([]byte, error) {
-	return i.udpSessions.EncodeServerPacket(i.method, userPSK, clientSessionID, dest, payload)
-}

proxy/shadowsocks_2022/inbound_relay.go

@@ -243,16 +243,14 @@ func (i *RelayInbound) processUDP(ctx context.Context, conn stat.Connection, dis
 			dest.Network = net.Network_UDP
 
 			sessionItem := i.udpSessions.GetOrCreate(sessionID)
+			sessionItem.Lock()
 			if sessionItem.User == nil {
-				sessionItem.Lock()
-				if sessionItem.User == nil {
-					sessionItem.User = &protocol.MemoryUser{
-						Email: targetDest.email,
-						Level: targetDest.level,
-					}
+				sessionItem.User = &protocol.MemoryUser{
+					Email: targetDest.email,
+					Level: targetDest.level,
 				}
-				sessionItem.Unlock()
 			}
+			sessionItem.Unlock()
 			link := sessionItem.link.Load()
 			if link != nil {
 				sessionItem.UpdateConn(conn)

proxy/shadowsocks_2022/outbound.go

@@ -30,7 +30,7 @@ type Outbound struct {
 	method        *CipherMethod
 	pskList       [][]byte
 	finalPSK      []byte
-	udpCodec      *UDPPacketCodec
+	udpCodec      *UDPClientCodec
 	policyManager policy.Manager
 }
 
@@ -50,9 +50,9 @@ func NewClient(ctx context.Context, config *ClientConfig) (*Outbound, error) {
 	}
 
 	finalPSK := pskList[len(pskList)-1]
-	udpCodec, err := NewUDPPacketCodec(method, pskList)
+	udpCodec, err := NewUDPClientCodec(method, pskList)
 	if err != nil {
-		return nil, errors.New("failed to create udp packet codec").Base(err)
+		return nil, errors.New("failed to create udp client codec").Base(err)
 	}
 
 	v := core.MustFromContext(ctx)

proxy/shadowsocks_2022/packet.go

@@ -7,33 +7,78 @@ import (
 	"io"
 	"math"
 	mrand "math/rand/v2"
+	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/xtls/xray-core/common/buf"
 	"github.com/xtls/xray-core/common/errors"
 	"github.com/xtls/xray-core/common/net"
+	"github.com/xtls/xray-core/common/protocol"
 )
 
-type UDPCodec struct {
+type UDPClientCodec struct {
 	method       *CipherMethod
 	pskList      [][]byte
 	psk          []byte
 	blockCipher  cipher.Block
 	blockCiphers []cipher.Block
 	chachaCipher cipher.AEAD
+}
+
+func NewUDPClientCodec(method *CipherMethod, pskList [][]byte) (*UDPClientCodec, error) {
+	if len(pskList) == 0 {
+		return nil, errors.New("empty psk list")
+	}
+	if method.IsChaCha && len(pskList) > 1 {
+		return nil, errors.New("multi-key is not supported for chacha20-poly1305")
+	}
+	finalPSK := pskList[len(pskList)-1]
+	c := &UDPClientCodec{
+		method:  method,
+		pskList: pskList,
+		psk:     finalPSK,
+	}
+	var err error
+	if method.IsChaCha {
+		c.chachaCipher, err = method.NewUDPCipher(finalPSK)
+	} else {
+		c.blockCipher, err = method.NewBlock(finalPSK)
+	}
+	if err != nil {
+		return nil, err
+	}
+	if len(pskList) > 1 {
+		c.blockCiphers = make([]cipher.Block, len(pskList))
+		for i, psk := range pskList {
+			c.blockCiphers[i], err = method.NewBlock(psk)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+	return c, nil
+}
+
+func NewUDPPacketCodec(method *CipherMethod, pskList [][]byte) (*UDPClientCodec, error) {
+	return NewUDPClientCodec(method, pskList)
+}
+
+type UDPServerCodec struct {
+	method       *CipherMethod
+	psk          []byte
+	user         *protocol.MemoryUser
+	blockCipher  cipher.Block
+	chachaCipher cipher.AEAD
 	sessions     *UDPSessionManager
 }
 
-type (
-	UDPPacketCodec = UDPCodec
-	UDPServerCodec = UDPCodec
-)
-
-func newUDPCodec(method *CipherMethod, psk []byte) (*UDPCodec, error) {
-	c := &UDPCodec{
-		method: method,
-		psk:    psk,
+func NewUDPServerCodec(method *CipherMethod, psk []byte, user *protocol.MemoryUser, sessionTimeout time.Duration) (*UDPServerCodec, error) {
+	c := &UDPServerCodec{
+		method:   method,
+		psk:      psk,
+		user:     user,
+		sessions: NewUDPSessionManager(sessionTimeout),
 	}
 	var err error
 	if method.IsChaCha {
@@ -47,46 +92,129 @@ func newUDPCodec(method *CipherMethod, psk []byte) (*UDPCodec, error) {
 	return c, nil
 }
 
-func NewUDPPacketCodec(method *CipherMethod, pskList [][]byte) (*UDPCodec, error) {
-	if method.IsChaCha && len(pskList) > 1 {
-		return nil, errors.New("multi-key is not supported for chacha20-poly1305")
-	}
-	finalPSK := pskList[len(pskList)-1]
-	c, err := newUDPCodec(method, finalPSK)
-	if err != nil {
-		return nil, err
-	}
-	c.pskList = pskList
-	if len(pskList) > 1 {
-		c.blockCiphers = make([]cipher.Block, len(pskList))
-		for i, psk := range pskList {
-			c.blockCiphers[i], err = method.NewBlock(psk)
-			if err != nil {
-				return nil, err
-			}
-		}
-	}
-	return c, nil
-}
-
-func NewUDPServerCodec(method *CipherMethod, psk []byte, sessionTimeout time.Duration) (*UDPCodec, error) {
-	c, err := newUDPCodec(method, psk)
-	if err != nil {
-		return nil, err
-	}
-	c.sessions = NewUDPSessionManager(sessionTimeout)
-	return c, nil
-}
-
-func (c *UDPCodec) Sessions() *UDPSessionManager {
+func (c *UDPServerCodec) Sessions() *UDPSessionManager {
 	return c.sessions
 }
 
-func (c *UDPCodec) GetSession(sessionID uint64) *ServerUDPSession {
+func (c *UDPServerCodec) GetSession(sessionID uint64) *ServerUDPSession {
 	if c.sessions == nil {
 		return nil
 	}
-	return c.sessions.GetOrCreate(sessionID)
+	return c.sessions.Get(sessionID)
+}
+
+type UDPMultiServerCodec struct {
+	method      *CipherMethod
+	masterPSK   []byte
+	masterBlock cipher.Block
+	sessions    *UDPSessionManager
+}
+
+func NewUDPMultiServerCodec(method *CipherMethod, masterPSK []byte, sessionTimeout time.Duration) (*UDPMultiServerCodec, error) {
+	if method.IsChaCha {
+		return nil, errors.New("multi-user is not supported for chacha20-poly1305")
+	}
+	masterBlock, err := method.NewBlock(masterPSK)
+	if err != nil {
+		return nil, err
+	}
+	return &UDPMultiServerCodec{
+		method:      method,
+		masterPSK:   masterPSK,
+		masterBlock: masterBlock,
+		sessions:    NewUDPSessionManager(sessionTimeout),
+	}, nil
+}
+
+func (c *UDPMultiServerCodec) Sessions() *UDPSessionManager {
+	return c.sessions
+}
+
+func (c *UDPMultiServerCodec) GetSession(sessionID uint64) *ServerUDPSession {
+	if c.sessions == nil {
+		return nil
+	}
+	return c.sessions.Get(sessionID)
+}
+
+type UserLookupFunc func(hash [AESBlockSize]byte) (*protocol.MemoryUser, []byte, bool)
+
+func (c *UDPMultiServerCodec) DecodePacket(data []byte, userLookup UserLookupFunc) (*ServerUDPSession, DecodedUDPPacket, *protocol.MemoryUser, error) {
+	if len(data) < 2*AESBlockSize+AEADTagSize+RequestHeaderFixedChunkLength {
+		return nil, DecodedUDPPacket{}, nil, ErrPacketTooShort
+	}
+
+	var rawHeader [16]byte
+	c.masterBlock.Decrypt(rawHeader[:], data[:16])
+
+	sessionID := binary.BigEndian.Uint64(rawHeader[:8])
+	packetID := binary.BigEndian.Uint64(rawHeader[8:16])
+
+	sessionItem := c.sessions.Get(sessionID)
+	var currentUser *protocol.MemoryUser
+	var userPSK []byte
+	var bodyAead cipher.AEAD
+
+	if sessionItem != nil {
+		sessionItem.Lock()
+		currentUser = sessionItem.User
+		userPSK = sessionItem.UserPSK
+		bodyAead = sessionItem.clientBodyCipher
+		sessionItem.Unlock()
+
+		if currentUser != nil {
+			if !sessionItem.CheckPacketID(packetID) {
+				return nil, DecodedUDPPacket{}, nil, ErrPacketIdNotUnique
+			}
+		}
+	}
+
+	if currentUser == nil {
+		decryptedHash := DecryptUDPEIH(c.masterBlock, rawHeader[:], data[16:32])
+		user, psk, ok := userLookup(decryptedHash)
+		if !ok {
+			return nil, DecodedUDPPacket{}, nil, errors.New("user not found")
+		}
+		currentUser = user
+		userPSK = psk
+	}
+
+	if bodyAead == nil {
+		bodyKey := DeriveSessionSubKey(userPSK, rawHeader[:8], c.method.KeySaltLength)
+		var err error
+		bodyAead, err = c.method.NewAEAD(bodyKey)
+		if err != nil {
+			return nil, DecodedUDPPacket{}, nil, err
+		}
+	}
+
+	bodyNonce := rawHeader[4:16]
+	bodyCipher := data[32:]
+	bodyPlain, err := bodyAead.Open(nil, bodyNonce, bodyCipher, nil)
+	if err != nil {
+		return nil, DecodedUDPPacket{}, nil, errors.New("failed to decrypt aes udp body").Base(err)
+	}
+
+	decoded, err := parsePlainUDPPacket(sessionID, packetID, bodyPlain)
+	if err != nil {
+		return nil, DecodedUDPPacket{}, nil, err
+	}
+	if decoded.HeaderType != HeaderTypeClient {
+		return nil, DecodedUDPPacket{}, nil, ErrBadHeaderType
+	}
+
+	if sessionItem == nil {
+		sessionItem = c.sessions.GetOrCreate(sessionID)
+	}
+	if !sessionItem.Commit(packetID, bodyAead, currentUser, userPSK) {
+		return nil, DecodedUDPPacket{}, nil, ErrPacketIdNotUnique
+	}
+
+	return sessionItem, decoded, currentUser, nil
+}
+
+func (c *UDPMultiServerCodec) EncodeServerPacket(clientSessionID uint64, userPSK []byte, dest net.Destination, payload []byte) ([]byte, error) {
+	return c.sessions.EncodeServerPacket(c.method, userPSK, clientSessionID, dest, payload)
 }
 
 type DecodedUDPPacket struct {
@@ -153,7 +281,7 @@ func parsePlainUDPPacket(sessionID, packetID uint64, bodyPlain []byte) (DecodedU
 		return DecodedUDPPacket{}, ErrBadHeaderType
 	}
 	epoch := binary.BigEndian.Uint64(bodyPlain[1:9])
-	diff := int(math.Abs(float64(time.Now().Unix() - int64(epoch))))
+	diff := math.Abs(float64(time.Now().Unix() - int64(epoch)))
 	if diff > 30 {
 		return DecodedUDPPacket{}, ErrBadTimestamp
 	}
@@ -193,44 +321,42 @@ func parsePlainUDPPacket(sessionID, packetID uint64, bodyPlain []byte) (DecodedU
 	}, nil
 }
 
-func (c *UDPCodec) DecodePacket(data []byte) (DecodedUDPPacket, error) {
+func (c *UDPServerCodec) DecodePacket(data []byte) (*ServerUDPSession, DecodedUDPPacket, error) {
 	if len(data) < PacketMinimalHeaderSize {
-		return DecodedUDPPacket{}, ErrPacketTooShort
+		return nil, DecodedUDPPacket{}, ErrPacketTooShort
 	}
 
 	if c.method.IsChaCha {
 		if len(data) < PacketNonceSize+AEADTagSize {
-			return DecodedUDPPacket{}, ErrPacketTooShort
+			return nil, DecodedUDPPacket{}, ErrPacketTooShort
 		}
 		nonce := data[:PacketNonceSize]
 		ciphertext := data[PacketNonceSize:]
 		plain, err := c.chachaCipher.Open(nil, nonce, ciphertext, nil)
 		if err != nil {
-			return DecodedUDPPacket{}, errors.New("failed to decrypt chacha udp packet").Base(err)
+			return nil, DecodedUDPPacket{}, errors.New("failed to decrypt chacha udp packet").Base(err)
 		}
 		if len(plain) < 16+1+8+2 {
-			return DecodedUDPPacket{}, ErrPacketTooShort
+			return nil, DecodedUDPPacket{}, ErrPacketTooShort
 		}
 
 		sessionID := binary.BigEndian.Uint64(plain[:8])
 		packetID := binary.BigEndian.Uint64(plain[8:16])
 
-		sessionItem := c.sessions.GetOrCreate(sessionID)
-		if !sessionItem.CheckPacketID(packetID) {
-			return DecodedUDPPacket{}, ErrPacketIdNotUnique
-		}
-
 		decoded, err := parsePlainUDPPacket(sessionID, packetID, plain[16:])
 		if err != nil {
-			return DecodedUDPPacket{}, err
+			return nil, DecodedUDPPacket{}, err
 		}
 
 		if decoded.HeaderType != HeaderTypeClient {
-			return DecodedUDPPacket{}, ErrBadHeaderType
+			return nil, DecodedUDPPacket{}, ErrBadHeaderType
 		}
 
-		sessionItem.AddPacketID(packetID)
-		return decoded, nil
+		sessionItem := c.sessions.GetOrCreate(sessionID)
+		if !sessionItem.Commit(packetID, nil, c.user, c.psk) {
+			return nil, DecodedUDPPacket{}, ErrPacketIdNotUnique
+		}
+		return sessionItem, decoded, nil
 	}
 
 	// AES mode
@@ -239,49 +365,48 @@ func (c *UDPCodec) DecodePacket(data []byte) (DecodedUDPPacket, error) {
 	sessionID := binary.BigEndian.Uint64(rawHeader[:8])
 	packetID := binary.BigEndian.Uint64(rawHeader[8:16])
 
-	sessionItem := c.sessions.GetOrCreate(sessionID)
-	if !sessionItem.CheckPacketID(packetID) {
-		return DecodedUDPPacket{}, ErrPacketIdNotUnique
+	sessionItem := c.sessions.Get(sessionID)
+	var bodyAead cipher.AEAD
+	if sessionItem != nil {
+		if !sessionItem.CheckPacketID(packetID) {
+			return nil, DecodedUDPPacket{}, ErrPacketIdNotUnique
+		}
+		sessionItem.Lock()
+		bodyAead = sessionItem.clientBodyCipher
+		sessionItem.Unlock()
 	}
 
-	return sessionItem.DecryptAESPayload(c.method, c.psk, sessionID, packetID, rawHeader[:], data[16:])
-}
-
-func (s *ServerUDPSession) DecryptAESPayload(method *CipherMethod, psk []byte, sessionID, packetID uint64, rawHeader, bodyCipher []byte) (DecodedUDPPacket, error) {
-	bodyAead := s.clientBodyCipher
-	isNewCipher := false
 	if bodyAead == nil {
-		bodyKey := DeriveSessionSubKey(psk, rawHeader[:8], method.KeySaltLength)
+		bodyKey := DeriveSessionSubKey(c.psk, rawHeader[:8], c.method.KeySaltLength)
 		var err error
-		bodyAead, err = method.NewAEAD(bodyKey)
+		bodyAead, err = c.method.NewAEAD(bodyKey)
 		if err != nil {
-			return DecodedUDPPacket{}, err
+			return nil, DecodedUDPPacket{}, err
 		}
-		isNewCipher = true
 	}
 
 	bodyNonce := rawHeader[4:16]
+	bodyCipher := data[16:]
 	bodyPlain, err := bodyAead.Open(nil, bodyNonce, bodyCipher, nil)
 	if err != nil {
-		return DecodedUDPPacket{}, errors.New("failed to decrypt aes udp body").Base(err)
+		return nil, DecodedUDPPacket{}, errors.New("failed to decrypt aes udp body").Base(err)
 	}
 
 	decoded, err := parsePlainUDPPacket(sessionID, packetID, bodyPlain)
 	if err != nil {
-		return DecodedUDPPacket{}, err
+		return nil, DecodedUDPPacket{}, err
 	}
-
 	if decoded.HeaderType != HeaderTypeClient {
-		return DecodedUDPPacket{}, ErrBadHeaderType
+		return nil, DecodedUDPPacket{}, ErrBadHeaderType
 	}
 
-	s.AddPacketID(packetID)
-
-	if isNewCipher {
-		s.clientBodyCipher = bodyAead
+	if sessionItem == nil {
+		sessionItem = c.sessions.GetOrCreate(sessionID)
 	}
-
-	return decoded, nil
+	if !sessionItem.Commit(packetID, bodyAead, c.user, c.psk) {
+		return nil, DecodedUDPPacket{}, ErrPacketIdNotUnique
+	}
+	return sessionItem, decoded, nil
 }
 
 func (s *ServerUDPSession) EnsureServerState(method *CipherMethod, psk []byte) error {
@@ -303,6 +428,9 @@ func (s *ServerUDPSession) EnsureServerState(method *CipherMethod, psk []byte) e
 	if method.IsChaCha {
 		var err error
 		s.serverChaCha, err = method.NewUDPCipher(psk)
+		if err != nil {
+			s.ServerSessionID = 0
+		}
 		return err
 	}
 
@@ -387,11 +515,12 @@ func (s *ServerUDPSession) EncodeServerPacket(method *CipherMethod, clientSessio
 	return res, nil
 }
 
-func (c *UDPCodec) EncodeServerPacket(clientSessionID uint64, dest net.Destination, payload []byte) ([]byte, error) {
+func (c *UDPServerCodec) EncodeServerPacket(clientSessionID uint64, dest net.Destination, payload []byte) ([]byte, error) {
 	return c.sessions.EncodeServerPacket(c.method, c.psk, clientSessionID, dest, payload)
 }
 
 type serverSessionState struct {
+	sync.Mutex
 	sessionID uint64
 	window    *SlidingWindow
 	cipher    cipher.AEAD
@@ -399,21 +528,20 @@ type serverSessionState struct {
 }
 
 func (st *serverSessionState) check(packetID uint64) bool {
-	if st.window == nil {
-		st.window = new(SlidingWindow)
-	}
+	st.Lock()
+	defer st.Unlock()
 	return st.window.Check(packetID)
 }
 
-func (st *serverSessionState) add(packetID uint64) {
-	if st.window == nil {
-		st.window = new(SlidingWindow)
-	}
-	st.window.Add(packetID)
+func (st *serverSessionState) commit(packetID uint64) bool {
+	st.Lock()
+	defer st.Unlock()
+	return st.window.CheckAndAdd(packetID)
 }
 
 type ClientUDPSession struct {
-	codec            *UDPCodec
+	mu               sync.Mutex
+	codec            *UDPClientCodec
 	clientSessionID  uint64
 	nextPacketID     atomic.Uint64
 	clientBodyCipher cipher.AEAD
@@ -421,7 +549,7 @@ type ClientUDPSession struct {
 	old              atomic.Pointer[serverSessionState]
 }
 
-func (c *UDPCodec) NewClientSession() (*ClientUDPSession, error) {
+func (c *UDPClientCodec) NewClientSession() (*ClientUDPSession, error) {
 	var sessID [8]byte
 	if _, err := io.ReadFull(rand.Reader, sessID[:]); err != nil {
 		return nil, err
@@ -446,7 +574,7 @@ func (c *UDPCodec) NewClientSession() (*ClientUDPSession, error) {
 	}, nil
 }
 
-func (s *ClientUDPSession) getServerSession(sessionID uint64, now int64) (*serverSessionState, error) {
+func (s *ClientUDPSession) getOrTransitionServerSession(sessionID uint64, cipher cipher.AEAD, now int64) (*serverSessionState, error) {
 	cur := s.current.Load()
 	if cur != nil && cur.sessionID == sessionID {
 		return cur, nil
@@ -455,39 +583,42 @@ func (s *ClientUDPSession) getServerSession(sessionID uint64, now int64) (*serve
 	old := s.old.Load()
 	if old != nil && old.sessionID == sessionID {
 		if now-old.lastSeen.Load() > 60 {
-			s.old.CompareAndSwap(old, nil)
 			return nil, errors.New("old server session expired")
 		}
 		return old, nil
 	}
 
-	// New server session:
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	cur = s.current.Load()
+	if cur != nil && cur.sessionID == sessionID {
+		return cur, nil
+	}
+
+	old = s.old.Load()
+	if old != nil && old.sessionID == sessionID {
+		if now-old.lastSeen.Load() > 60 {
+			return nil, errors.New("old server session expired")
+		}
+		return old, nil
+	}
+
 	// Spec §3.2.4: reject newer server sessions when the last packet received from the old session is less than 1 minute old.
 	if old != nil && now-old.lastSeen.Load() < 60 {
 		return nil, errors.New("newer server session rejected: old session is less than 1 minute old")
 	}
 
-	var bodyAead cipher.AEAD
-	if !s.codec.method.IsChaCha {
-		var sessBytes [8]byte
-		binary.BigEndian.PutUint64(sessBytes[:], sessionID)
-		bodyKey := DeriveSessionSubKey(s.codec.psk, sessBytes[:], s.codec.method.KeySaltLength)
-		var err error
-		bodyAead, err = s.codec.method.NewAEAD(bodyKey)
-		if err != nil {
-			return nil, err
-		}
-	}
-
 	newState := &serverSessionState{
 		sessionID: sessionID,
-		cipher:    bodyAead,
+		window:    new(SlidingWindow),
+		cipher:    cipher,
 	}
 	newState.lastSeen.Store(now)
 
 	if cur == nil {
-		s.current.CompareAndSwap(nil, newState)
-		return s.current.Load(), nil
+		s.current.Store(newState)
+		return newState, nil
 	}
 
 	s.old.Store(cur)
@@ -637,15 +768,6 @@ func (s *ClientUDPSession) DecodePacket(data []byte) (DecodedUDPPacket, error) {
 		sessionID := binary.BigEndian.Uint64(plain[:8])
 		packetID := binary.BigEndian.Uint64(plain[8:16])
 
-		now := time.Now().Unix()
-		st, err := s.getServerSession(sessionID, now)
-		if err != nil {
-			return DecodedUDPPacket{}, err
-		}
-		if !st.check(packetID) {
-			return DecodedUDPPacket{}, ErrPacketIdNotUnique
-		}
-
 		decoded, err := parsePlainUDPPacket(sessionID, packetID, plain[16:])
 		if err != nil {
 			return DecodedUDPPacket{}, err
@@ -658,7 +780,16 @@ func (s *ClientUDPSession) DecodePacket(data []byte) (DecodedUDPPacket, error) {
 			return DecodedUDPPacket{}, errors.New("client session ID mismatch")
 		}
 
-		st.add(packetID)
+		now := time.Now().Unix()
+		st, err := s.getOrTransitionServerSession(sessionID, nil, now)
+		if err != nil {
+			return DecodedUDPPacket{}, err
+		}
+
+		// Atomically commit after all validations succeed
+		if !st.commit(packetID) {
+			return DecodedUDPPacket{}, ErrPacketIdNotUnique
+		}
 		st.lastSeen.Store(now)
 
 		return decoded, nil
@@ -671,18 +802,79 @@ func (s *ClientUDPSession) DecodePacket(data []byte) (DecodedUDPPacket, error) {
 	packetID := binary.BigEndian.Uint64(rawHeader[8:16])
 
 	now := time.Now().Unix()
-	st, err := s.getServerSession(sessionID, now)
+
+	// 1. Fast path for known active session
+	cur := s.current.Load()
+	var st *serverSessionState
+	var bodyAead cipher.AEAD
+
+	if cur != nil && cur.sessionID == sessionID {
+		st = cur
+		bodyAead = cur.cipher
+	} else {
+		old := s.old.Load()
+		if old != nil && old.sessionID == sessionID {
+			if now-old.lastSeen.Load() > 60 {
+				return DecodedUDPPacket{}, errors.New("old server session expired")
+			}
+			st = old
+			bodyAead = old.cipher
+		}
+	}
+
+	if st != nil {
+		if !st.check(packetID) {
+			return DecodedUDPPacket{}, ErrPacketIdNotUnique
+		}
+		bodyNonce := rawHeader[4:16]
+		bodyCipher := data[16:]
+		bodyPlain, err := bodyAead.Open(nil, bodyNonce, bodyCipher, nil)
+		if err != nil {
+			return DecodedUDPPacket{}, errors.New("failed to decrypt aes udp body").Base(err)
+		}
+
+		decoded, err := parsePlainUDPPacket(sessionID, packetID, bodyPlain)
+		if err != nil {
+			return DecodedUDPPacket{}, err
+		}
+
+		if decoded.HeaderType != HeaderTypeServer {
+			return DecodedUDPPacket{}, ErrBadHeaderType
+		}
+		if decoded.ClientSessionID != s.clientSessionID {
+			return DecodedUDPPacket{}, errors.New("client session ID mismatch")
+		}
+
+		if !st.commit(packetID) {
+			return DecodedUDPPacket{}, ErrPacketIdNotUnique
+		}
+		st.lastSeen.Store(now)
+
+		return decoded, nil
+	}
+
+	// 2. Unknown session: Trial path without mutating s.current or s.old yet!
+	// Re-check cur in case another worker concurrently transitioned to this sessionID
+	cur = s.current.Load()
+	old := s.old.Load()
+	if (cur == nil || cur.sessionID != sessionID) && (old == nil || old.sessionID != sessionID) {
+		// Spec §3.2.4: reject newer server sessions when the last packet received from the old session is less than 1 minute old.
+		if old != nil && now-old.lastSeen.Load() < 60 {
+			return DecodedUDPPacket{}, errors.New("newer server session rejected: old session is less than 1 minute old")
+		}
+	}
+
+	var sessBytes [8]byte
+	binary.BigEndian.PutUint64(sessBytes[:], sessionID)
+	bodyKey := DeriveSessionSubKey(s.codec.psk, sessBytes[:], s.codec.method.KeySaltLength)
+	trialAead, err := s.codec.method.NewAEAD(bodyKey)
 	if err != nil {
 		return DecodedUDPPacket{}, err
 	}
-	if !st.check(packetID) {
-		return DecodedUDPPacket{}, ErrPacketIdNotUnique
-	}
-	bodyAead := st.cipher
 
 	bodyNonce := rawHeader[4:16]
 	bodyCipher := data[16:]
-	bodyPlain, err := bodyAead.Open(nil, bodyNonce, bodyCipher, nil)
+	bodyPlain, err := trialAead.Open(nil, bodyNonce, bodyCipher, nil)
 	if err != nil {
 		return DecodedUDPPacket{}, errors.New("failed to decrypt aes udp body").Base(err)
 	}
@@ -699,7 +891,15 @@ func (s *ClientUDPSession) DecodePacket(data []byte) (DecodedUDPPacket, error) {
 		return DecodedUDPPacket{}, errors.New("client session ID mismatch")
 	}
 
-	st.add(packetID)
+	// Payload authenticated! Safely commit and transition session under lock
+	st, err = s.getOrTransitionServerSession(sessionID, trialAead, now)
+	if err != nil {
+		return DecodedUDPPacket{}, err
+	}
+
+	if !st.commit(packetID) {
+		return DecodedUDPPacket{}, ErrPacketIdNotUnique
+	}
 	st.lastSeen.Store(now)
 
 	return decoded, nil

proxy/shadowsocks_2022/relay_test.go

@@ -191,7 +191,7 @@ func TestRelayUDPSessionStabilityAndDispatch(t *testing.T) {
 	// Verify downstream destination can decode both packets
 	method, err := GetCipherMethod(MethodAES128GCM)
 	common.Must(err)
-	destCodec, err := NewUDPServerCodec(method, destKey, 300*time.Second)
+	destCodec, err := NewUDPServerCodec(method, destKey, nil, 300*time.Second)
 	common.Must(err)
 
 	mu.Lock()
@@ -202,7 +202,8 @@ func TestRelayUDPSessionStabilityAndDispatch(t *testing.T) {
 		t.Fatalf("expected 2 received packets at destination, got %d", len(pkts))
 	}
 
-	dec1, err := destCodec.DecodePacket(pkts[0])
+	dec1, dec2 := DecodedUDPPacket{}, DecodedUDPPacket{}
+	_, dec1, err = destCodec.DecodePacket(pkts[0])
 	if err != nil {
 		t.Fatalf("dest failed to decode packet 1: %v", err)
 	}
@@ -210,7 +211,7 @@ func TestRelayUDPSessionStabilityAndDispatch(t *testing.T) {
 		t.Fatalf("dec1 mismatch: sess=%x, pktID=%d, payload=%s", dec1.SessionID, dec1.PacketID, string(dec1.Payload))
 	}
 
-	dec2, err := destCodec.DecodePacket(pkts[1])
+	_, dec2, err = destCodec.DecodePacket(pkts[1])
 	if err != nil {
 		t.Fatalf("dest failed to decode packet 2: %v", err)
 	}

proxy/shadowsocks_2022/replay.go

@@ -49,12 +49,12 @@ func (f *SlidingWindow) Add(counter uint64) {
 
 	if counter > f.last {
 		lastBlockIndex := f.last >> swBlockBitLog
-		diff := int(blockIndex - lastBlockIndex)
+		diff := blockIndex - lastBlockIndex
 		if diff > swRingBlocks {
 			diff = swRingBlocks
 		}
 
-		for i := 0; i < diff; i++ {
+		for i := 0; i < int(diff); i++ {
 			lastBlockIndex = (lastBlockIndex + 1) & swBlockMask
 			f.ring[lastBlockIndex] = 0
 		}
@@ -100,19 +100,27 @@ type ServerUDPSession struct {
 func (s *ServerUDPSession) CheckPacketID(packetID uint64) bool {
 	s.Lock()
 	defer s.Unlock()
-	if s.Window == nil {
-		s.Window = new(SlidingWindow)
-	}
 	return s.Window.Check(packetID)
 }
 
-func (s *ServerUDPSession) AddPacketID(packetID uint64) {
+func (s *ServerUDPSession) Commit(packetID uint64, bodyAead cipher.AEAD, user *protocol.MemoryUser, userPSK []byte) bool {
 	s.Lock()
 	defer s.Unlock()
-	if s.Window == nil {
-		s.Window = new(SlidingWindow)
+	if s.User != nil && user != nil && s.User != user {
+		return false
 	}
-	s.Window.Add(packetID)
+	if !s.Window.CheckAndAdd(packetID) {
+		return false
+	}
+	if s.clientBodyCipher == nil && bodyAead != nil {
+		s.clientBodyCipher = bodyAead
+	}
+	if s.User == nil && user != nil {
+		s.User = user
+		s.UserPSK = userPSK
+	}
+	s.LastActive.Store(time.Now().Unix())
+	return true
 }
 
 type UDPSessionManager struct {
@@ -128,6 +136,13 @@ func NewUDPSessionManager(timeout time.Duration) *UDPSessionManager {
 	}
 }
 
+func (m *UDPSessionManager) Get(sessionID uint64) *ServerUDPSession {
+	if s, ok := m.sessions.Load(sessionID); ok {
+		return s
+	}
+	return nil
+}
+
 func (m *UDPSessionManager) GetOrCreate(sessionID uint64) *ServerUDPSession {
 	now := time.Now().Unix()
 	if s, ok := m.sessions.Load(sessionID); ok {
@@ -137,6 +152,7 @@ func (m *UDPSessionManager) GetOrCreate(sessionID uint64) *ServerUDPSession {
 
 	s := &ServerUDPSession{
 		SessionID: sessionID,
+		Window:    new(SlidingWindow),
 		manager:   m,
 	}
 	s.LastActive.Store(now)

proxy/shadowsocks_2022/shadowsocks_2022.go

@@ -63,10 +63,16 @@ func (s *ServerUDPSession) EnsureLink(
 
 	s.currentConn.Store(conn)
 
-	sessCtx, cancel := context.WithCancel(ctx)
+	// UDP Session might travel across multiple connections
+	// don't bind its lifetime to a parent context
+	sessCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
 	inbound := session.InboundFromContext(sessCtx)
-	if inbound != nil && s.User != nil {
-		inbound.User = s.User
+	if inbound != nil {
+		newInbound := *inbound
+		if s.User != nil {
+			newInbound.User = s.User
+		}
+		sessCtx = session.ContextWithInbound(sessCtx, &newInbound)
 	}
 	var email string
 	var level uint32

proxy/shadowsocks_2022/shadowsocks_2022_test.go

@@ -265,7 +265,7 @@ func TestUDPCodec(t *testing.T) {
 
 			clientCodec, err := NewUDPPacketCodec(method, [][]byte{psk})
 			common.Must(err)
-			serverCodec, err := NewUDPServerCodec(method, psk, time.Minute)
+			serverCodec, err := NewUDPServerCodec(method, psk, nil, time.Minute)
 			common.Must(err)
 
 			session, err := clientCodec.NewClientSession()
@@ -277,7 +277,7 @@ func TestUDPCodec(t *testing.T) {
 			rawCopy := make([]byte, pktBuf.Len())
 			copy(rawCopy, pktBuf.Bytes())
 
-			decoded, err := serverCodec.DecodePacket(pktBuf.Bytes())
+			_, decoded, err := serverCodec.DecodePacket(pktBuf.Bytes())
 			common.Must(err)
 
 			if decoded.HeaderType != HeaderTypeClient {
@@ -291,7 +291,7 @@ func TestUDPCodec(t *testing.T) {
 			}
 
 			// Replay same packet wire bytes should fail with ErrPacketIdNotUnique
-			_, err = serverCodec.DecodePacket(rawCopy)
+			_, _, err = serverCodec.DecodePacket(rawCopy)
 			if err != ErrPacketIdNotUnique {
 				t.Fatalf("expected ErrPacketIdNotUnique on replay, got: %v", err)
 			}
@@ -427,7 +427,7 @@ func TestClientUDPSessionMultiDestination(t *testing.T) {
 
 			clientCodec, err := NewUDPPacketCodec(method, [][]byte{rawKey})
 			common.Must(err)
-			serverCodec, err := NewUDPServerCodec(method, rawKey, time.Minute)
+			serverCodec, err := NewUDPServerCodec(method, rawKey, nil, time.Minute)
 			common.Must(err)
 
 			session, err := clientCodec.NewClientSession()
@@ -448,9 +448,9 @@ func TestClientUDPSessionMultiDestination(t *testing.T) {
 			defer pkt2.Release()
 
 			// Server decodes both
-			dec1, err := serverCodec.DecodePacket(pkt1.Bytes())
+			_, dec1, err := serverCodec.DecodePacket(pkt1.Bytes())
 			common.Must(err)
-			dec2, err := serverCodec.DecodePacket(pkt2.Bytes())
+			_, dec2, err := serverCodec.DecodePacket(pkt2.Bytes())
 			common.Must(err)
 
 			if dec1.SessionID != session.ClientSessionID() || dec2.SessionID != session.ClientSessionID() {

proxy/shadowsocks_2022/stream.go

@@ -254,7 +254,7 @@ func ReadClientRequestHeaderWithFixed(reader *StreamReader, fixedChunk []byte) (
 	}
 
 	epoch := binary.BigEndian.Uint64(plainFixed[1:9])
-	diff := int(math.Abs(float64(time.Now().Unix() - int64(epoch))))
+	diff := math.Abs(float64(time.Now().Unix() - int64(epoch)))
 	if diff > 30 {
 		return nil, ErrBadTimestamp
 	}
@@ -431,7 +431,7 @@ func ReadTCPResponse(r io.Reader, method *CipherMethod, psk []byte, clientSalt [
 	}
 
 	serverEpoch := binary.BigEndian.Uint64(decryptedFixed[1:9])
-	diff := int(math.Abs(float64(time.Now().Unix() - int64(serverEpoch))))
+	diff := math.Abs(float64(time.Now().Unix() - int64(serverEpoch)))
 	if diff > 30 {
 		return nil, ErrBadTimestamp
 	}

proxy/tun/handler.go

@@ -144,7 +144,8 @@ func (t *Handler) Start() error {
 				errors.LogInfo(context.Background(), "[tun] falied to set interface > iface == nil")
 				return errors.New("iface not found")
 			}
-			return c.Control(func(fd uintptr) {
+			var controlErr, sockErr error
+			controlErr = c.Control(func(fd uintptr) {
 				addrPort, _ := netip.ParseAddrPort(address)
 				// skip loopback
 				if addrPort.Addr().IsLoopback() || strings.HasPrefix(strings.ToLower(address), "localhost:") {
@@ -152,9 +153,13 @@ func (t *Handler) Start() error {
 				}
 				err := setinterface(network, address, fd, iface)
 				if err != nil {
-					errors.LogInfoInner(context.Background(), err, "[tun] falied to set interface")
+					sockErr = errors.New("[tun] falied to set interface").Base(err)
 				}
 			})
+			if sockErr != nil {
+				return sockErr
+			}
+			return controlErr
 		})
 	}
 

proxy/vless/inbound/inbound.go

@@ -6,11 +6,9 @@ import (
 	gotls "crypto/tls"
 	"encoding/base64"
 	"io"
-	"reflect"
 	"strconv"
 	"strings"
 	"time"
-	"unsafe"
 
 	"github.com/xtls/xray-core/app/dispatcher"
 	"github.com/xtls/xray-core/app/reverse"
@@ -548,7 +546,7 @@ func (h *Handler) Process(ctx context.Context, network net.Network, connection s
 	}
 
 	var input *bytes.Reader
-	var rawInput *bytes.Buffer
+	var rawInput **bytes.Buffer
 	switch requestAddons.Flow {
 	case vless.XRV:
 		if account.Flow == requestAddons.Flow {
@@ -560,30 +558,23 @@ func (h *Handler) Process(ctx context.Context, network net.Network, connection s
 				inbound.CanSpliceCopy = 3
 				fallthrough // we will break Mux connections that contain TCP requests
 			case protocol.RequestCommandTCP:
-				var t reflect.Type
-				var p uintptr
+				var target net.Conn
 				if commonConn, ok := connection.(*encryption.CommonConn); ok {
 					if _, ok := commonConn.Conn.(*encryption.XorConn); ok || !proxy.IsRAWTransportWithoutSecurity(iConn) {
 						inbound.CanSpliceCopy = 3 // full-random xorConn / non-RAW transport / another securityConn should not be penetrated
 					}
-					t = reflect.TypeOf(commonConn).Elem()
-					p = uintptr(unsafe.Pointer(commonConn))
+					target = commonConn
 				} else if tlsConn, ok := iConn.(*tls.Conn); ok {
 					if tlsConn.ConnectionState().Version != gotls.VersionTLS13 {
 						return errors.New(`failed to use `+requestAddons.Flow+`, found outer tls version `, tlsConn.ConnectionState().Version)
 					}
-					t = reflect.TypeOf(tlsConn.Conn).Elem()
-					p = uintptr(unsafe.Pointer(tlsConn.Conn))
+					target = tlsConn.Conn
 				} else if realityConn, ok := iConn.(*reality.Conn); ok {
-					t = reflect.TypeOf(realityConn.Conn).Elem()
-					p = uintptr(unsafe.Pointer(realityConn.Conn))
+					target = realityConn.Conn
 				} else {
 					return errors.New("XTLS only supports TLS and REALITY directly for now.")
 				}
-				i, _ := t.FieldByName("input")
-				r, _ := t.FieldByName("rawInput")
-				input = (*bytes.Reader)(unsafe.Pointer(p + i.Offset))
-				rawInput = (*bytes.Buffer)(unsafe.Pointer(p + r.Offset))
+				input, rawInput = vless.AccessTLSBuffers(target)
 			}
 		} else {
 			return errors.New("account " + account.ID.String() + " is not able to use the flow " + requestAddons.Flow)

proxy/vless/outbound/outbound.go

@@ -5,11 +5,9 @@ import (
 	"context"
 	gotls "crypto/tls"
 	"encoding/base64"
-	"reflect"
 	"strings"
 	"sync"
 	"time"
-	"unsafe"
 
 	utls "github.com/refraction-networking/utls"
 	proxymanConfig "github.com/xtls/xray-core/app/proxyman"
@@ -244,7 +242,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
 	}
 
 	var input *bytes.Reader
-	var rawInput *bytes.Buffer
+	var rawInput **bytes.Buffer
 	allowUDP443 := false
 	switch requestAddons.Flow {
 	case vless.XRV + "-udp443":
@@ -261,30 +259,22 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
 		case protocol.RequestCommandMux:
 			fallthrough // let server break Mux connections that contain TCP requests
 		case protocol.RequestCommandTCP, protocol.RequestCommandRvs:
-			var t reflect.Type
-			var p uintptr
+			var target net.Conn
 			if commonConn, ok := conn.(*encryption.CommonConn); ok {
 				if _, ok := commonConn.Conn.(*encryption.XorConn); ok || !proxy.IsRAWTransportWithoutSecurity(iConn) {
 					ob.CanSpliceCopy = 3 // full-random xorConn / non-RAW transport / another securityConn should not be penetrated
 				}
-				t = reflect.TypeOf(commonConn).Elem()
-				p = uintptr(unsafe.Pointer(commonConn))
+				target = commonConn
 			} else if tlsConn, ok := iConn.(*tls.Conn); ok {
-				t = reflect.TypeOf(tlsConn.Conn).Elem()
-				p = uintptr(unsafe.Pointer(tlsConn.Conn))
+				target = tlsConn.Conn
 			} else if utlsConn, ok := iConn.(*tls.UConn); ok {
-				t = reflect.TypeOf(utlsConn.Conn).Elem()
-				p = uintptr(unsafe.Pointer(utlsConn.Conn))
+				target = utlsConn.Conn
 			} else if realityConn, ok := iConn.(*reality.UConn); ok {
-				t = reflect.TypeOf(realityConn.Conn).Elem()
-				p = uintptr(unsafe.Pointer(realityConn.Conn))
+				target = realityConn.Conn
 			} else {
 				return errors.New("XTLS only supports TLS and REALITY directly for now.")
 			}
-			i, _ := t.FieldByName("input")
-			r, _ := t.FieldByName("rawInput")
-			input = (*bytes.Reader)(unsafe.Pointer(p + i.Offset))
-			rawInput = (*bytes.Buffer)(unsafe.Pointer(p + r.Offset))
+			input, rawInput = vless.AccessTLSBuffers(target)
 		default:
 			panic("unknown VLESS request command")
 		}

proxy/vless/vless.go

@@ -5,7 +5,28 @@
 // clients with 'socks' for proxying.
 package vless
 
+import (
+	"bytes"
+	"net"
+	"reflect"
+
+	"github.com/xtls/xray-core/common/utils"
+)
+
 const (
 	None = "none"
 	XRV  = "xtls-rprx-vision"
 )
+
+// Old TLS lib holds bytes.Buffer, but new TLS lib holds *bytes.Buffer
+// Since Xray has 3 TLS libraries (std, utls, REALITY server)
+// we detect the type of rawInput and return the appropriate pointer
+func AccessTLSBuffers(conn net.Conn) (*bytes.Reader, **bytes.Buffer) {
+	input := utils.AccessField[bytes.Reader](conn, "input")
+	field := reflect.ValueOf(conn).Elem().FieldByName("rawInput")
+	if field.Kind() == reflect.Pointer {
+		return input, utils.AccessField[*bytes.Buffer](conn, "rawInput")
+	}
+	raw := utils.AccessField[bytes.Buffer](conn, "rawInput")
+	return input, &raw
+}

proxy/wireguard/bind.go

@@ -16,8 +16,9 @@ import (
 )
 
 type bind struct {
+	ctx         context.Context
 	resolveFunc func(host string) (net.IP, error)
-	listenFunc  func() (net.PacketConn, error)
+	listenFunc  func(ctx context.Context) (net.PacketConn, error)
 	downFunc    func() error
 	reserved    []byte
 
@@ -26,6 +27,12 @@ type bind struct {
 	mu      sync.Mutex
 }
 
+func (b *bind) SetCtx(ctx context.Context) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.ctx = ctx
+}
+
 func (b *bind) Open(port uint16) (fns []conn.ReceiveFunc, actualPort uint16, err error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -34,7 +41,7 @@ func (b *bind) Open(port uint16) (fns []conn.ReceiveFunc, actualPort uint16, err
 		return nil, 0, conn.ErrBindAlreadyOpen
 	}
 
-	c, err := b.listenFunc()
+	c, err := b.listenFunc(b.ctx)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -52,12 +59,9 @@ func (b *bind) Open(port uint16) (fns []conn.ReceiveFunc, actualPort uint16, err
 						case <-ch:
 						default:
 							errors.LogErrorInner(context.Background(), err, "unexpected closed")
-							b.mu.Lock()
-							downFunc := b.downFunc
-							b.mu.Unlock()
-							if downFunc != nil {
+							if b.downFunc != nil {
 								go func() {
-									common.Must(downFunc())
+									common.Must(b.downFunc())
 								}()
 							}
 						}
@@ -79,13 +83,6 @@ func (b *bind) Open(port uint16) (fns []conn.ReceiveFunc, actualPort uint16, err
 	}, uint16(c.LocalAddr().(*net.UDPAddr).Port), nil
 }
 
-// setDownFunc sets downFunc after the device is created, since the device may already be using the bind.
-func (b *bind) setDownFunc(f func() error) {
-	b.mu.Lock()
-	defer b.mu.Unlock()
-	b.downFunc = f
-}
-
 func (b *bind) Close() error {
 	b.mu.Lock()
 	defer b.mu.Unlock()

proxy/wireguard/client.go

@@ -7,6 +7,7 @@ import (
 	"reflect"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"golang.zx2c4.com/wireguard/tun"
@@ -31,18 +32,14 @@ import (
 )
 
 type Handler struct {
-	conf          *DeviceConfig
-	policyManager policy.Manager
-	dns           dns.Client
+	p policy.Manager
 
-	streamSettings  *internet.MemoryStreamConfig
-	uplinkCounter   stats.Counter
-	downlinkCounter stats.Counter
-
-	tun  tun.Device
-	tnet *Net
-	dev  *device.Device
-	mu   sync.Mutex
+	canup *atomic.Bool
+	tnet  *Net
+	bind  *bind
+	dev   *device.Device
+	ipc   string
+	mu    sync.Mutex
 }
 
 func NewClient(ctx context.Context, conf *DeviceConfig) (*Handler, error) {
@@ -124,17 +121,86 @@ func NewClient(ctx context.Context, conf *DeviceConfig) (*Handler, error) {
 		return nil, err
 	}
 
+	blockup := newblockup(tun)
+	bind := &bind{}
+	logger := &device.Logger{
+		Verbosef: func(format string, args ...any) {
+			log.Record(&log.GeneralMessage{
+				Severity: log.Severity_Debug,
+				Content:  fmt.Sprintf(format, args...),
+			})
+		},
+		Errorf: func(format string, args ...any) {
+			log.Record(&log.GeneralMessage{
+				Severity: log.Severity_Error,
+				Content:  fmt.Sprintf(format, args...),
+			})
+		},
+	}
+	dev := device.NewDevice(blockup, bind, logger)
+	bind.ctx = context.Background()
+	bind.resolveFunc = MakeResolveLocal(d, streamSettings)
+	bind.listenFunc = func(ctx context.Context) (net.PacketConn, error) {
+		dest, err := net.ParseDestination("udp:" + conf.Peers[0].Endpoint)
+		if err != nil {
+			return nil, err
+		}
+		var pktConn net.PacketConn
+		if streamSettings.FinalMask != nil {
+			conn, err := streamSettings.FinalMask.DialUDP(ctx, dest)
+			if err != nil {
+				return nil, errors.New("failed to dial to dest").Base(err)
+			}
+			pktConn = conn.(*net.PacketConnWrapper).PacketConn
+		} else {
+			conn, err := internet.DialSystem(ctx, dest, streamSettings.SocketSettings)
+			if err != nil {
+				return nil, errors.New("failed to dial to dest").Base(err)
+			}
+			switch c := conn.(type) {
+			case *net.PacketConnWrapper:
+				pktConn = c.PacketConn
+			case *cnc.Connection:
+				pktConn = &internet.FakePacketConn{Conn: c}
+			default:
+				panic(reflect.TypeOf(c))
+			}
+		}
+		if uplinkCounter != nil || downlinkCounter != nil {
+			pktConn = &PacketCounterConnection{
+				PacketConn:   pktConn,
+				ReadCounter:  downlinkCounter,
+				WriteCounter: uplinkCounter,
+			}
+		}
+		return pktConn, nil
+	}
+	bind.downFunc = dev.Down
+	bind.reserved = conf.Reserved
+	var cfg strings.Builder
+	cfg.WriteString("private_key=" + conf.SecretKey + "\n")
+	for _, peer := range conf.Peers {
+		cfg.WriteString("public_key=" + peer.PublicKey + "\n")
+		if peer.PreSharedKey != "" {
+			cfg.WriteString("preshared_key=" + peer.PreSharedKey + "\n")
+		}
+		cfg.WriteString("endpoint=" + peer.Endpoint + "\n")
+		for _, ip := range peer.AllowedIps {
+			cfg.WriteString("allowed_ip=" + ip + "\n")
+		}
+		if peer.KeepAlive != "" {
+			cfg.WriteString("persistent_keepalive_interval=" + peer.KeepAlive + "\n")
+		}
+	}
+
 	return &Handler{
-		conf:          conf,
-		policyManager: p,
-		dns:           d,
+		p: p,
 
-		streamSettings:  streamSettings,
-		uplinkCounter:   uplinkCounter,
-		downlinkCounter: downlinkCounter,
-
-		tun:  tun,
-		tnet: tnet,
+		canup: blockup.canup,
+		tnet:  tnet,
+		bind:  bind,
+		dev:   dev,
+		ipc:   cfg.String(),
 	}, nil
 }
 
@@ -149,7 +215,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
 	ob.CanSpliceCopy = 3
 	dialer.SetOutboundGateway(ctx, ob)
 
-	if err := h.init(ctx); err != nil {
+	if err := h.up(ctx); err != nil {
 		return err
 	}
 
@@ -159,7 +225,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
 		newCtx, newCancel = context.WithCancel(context.Background())
 	}
 
-	sessionPolicy := h.policyManager.ForLevel(0)
+	sessionPolicy := h.p.ForLevel(0)
 	ctx, cancel := context.WithCancel(ctx)
 	timer := signal.CancelAfterInactivity(ctx, func() {
 		cancel()
@@ -234,150 +300,72 @@ func (h *Handler) Close() (err error) {
 	if h.dev != nil {
 		h.dev.Close()
 		h.dev = nil
-		h.tun = nil
-	} else if h.tun != nil {
-		h.tun.Close()
-		h.tun = nil
 	}
 	return nil
 }
 
-func (h *Handler) init(ctx context.Context) error {
+func (h *Handler) up(ctx context.Context) error {
 	h.mu.Lock()
 	defer h.mu.Unlock()
-	if h.tun == nil {
+	if h.dev == nil {
 		return errors.New("closed")
 	}
-	if h.dev != nil {
-		return h.dev.Up()
+	h.bind.SetCtx(ctx)
+	if !h.canup.Load() {
+		h.canup.Store(true)
 	}
-	resolveFunc := h.resolveLocal
-	listenFunc := func() (net.PacketConn, error) {
-		dest, err := net.ParseDestination("udp:" + h.conf.Peers[0].Endpoint)
+	if len(h.ipc) > 0 {
+		if err := h.dev.IpcSet(h.ipc); err != nil {
+			return err
+		}
+		h.ipc = ""
+	}
+	return h.dev.Up()
+}
+
+func MakeResolveLocal(d dns.Client, streamSettings *internet.MemoryStreamConfig) func(host string) (net.IP, error) {
+	return func(host string) (net.IP, error) {
+		if ip := net.ParseIP(host); ip != nil {
+			return ip, nil
+		}
+		ips, _, err := d.LookupIP(host, dns.IPOption{IPv4Enable: true, IPv6Enable: true})
 		if err != nil {
 			return nil, err
 		}
-		var pktConn net.PacketConn
-		if h.streamSettings.FinalMask != nil {
-			conn, err := h.streamSettings.FinalMask.DialUDP(ctx, dest)
-			if err != nil {
-				return nil, errors.New("failed to dial to dest").Base(err)
+		got := ips
+		if streamSettings.SocketSettings != nil {
+			var got4, got6 []net.IP
+			for _, ip := range ips {
+				if ip.To4() != nil {
+					got4 = append(got4, ip)
+				} else {
+					got6 = append(got6, ip)
+				}
 			}
-			pktConn = conn.(*net.PacketConnWrapper).PacketConn
-		} else {
-			conn, err := internet.DialSystem(ctx, dest, h.streamSettings.SocketSettings)
-			if err != nil {
-				return nil, errors.New("failed to dial to dest").Base(err)
-			}
-			switch c := conn.(type) {
-			case *net.PacketConnWrapper:
-				pktConn = c.PacketConn
-			case *cnc.Connection:
-				pktConn = &internet.FakePacketConn{Conn: c}
-			default:
-				panic(reflect.TypeOf(c))
-			}
-		}
-		if h.uplinkCounter != nil || h.downlinkCounter != nil {
-			pktConn = &PacketCounterConnection{
-				PacketConn:   pktConn,
-				ReadCounter:  h.downlinkCounter,
-				WriteCounter: h.uplinkCounter,
-			}
-		}
-		return pktConn, nil
-	}
-	// device.NewDevice may use the bind right away (Up -> BindUpdate -> Open),
-	// so everything it reads must be set before creating the device.
-	bind := &bind{
-		resolveFunc: resolveFunc,
-		listenFunc:  listenFunc,
-		reserved:    h.conf.Reserved,
-	}
-	logger := &device.Logger{
-		Verbosef: func(format string, args ...any) {
-			log.Record(&log.GeneralMessage{
-				Severity: log.Severity_Debug,
-				Content:  fmt.Sprintf(format, args...),
-			})
-		},
-		Errorf: func(format string, args ...any) {
-			log.Record(&log.GeneralMessage{
-				Severity: log.Severity_Error,
-				Content:  fmt.Sprintf(format, args...),
-			})
-		},
-	}
-	dev := device.NewDevice(h.tun, bind, logger)
-	bind.setDownFunc(dev.Down)
-	var cfg strings.Builder
-	cfg.WriteString("private_key=" + h.conf.SecretKey + "\n")
-	for _, peer := range h.conf.Peers {
-		cfg.WriteString("public_key=" + peer.PublicKey + "\n")
-		if peer.PreSharedKey != "" {
-			cfg.WriteString("preshared_key=" + peer.PreSharedKey + "\n")
-		}
-		cfg.WriteString("endpoint=" + peer.Endpoint + "\n")
-		for _, ip := range peer.AllowedIps {
-			cfg.WriteString("allowed_ip=" + ip + "\n")
-		}
-		if peer.KeepAlive != "" {
-			cfg.WriteString("persistent_keepalive_interval=" + peer.KeepAlive + "\n")
-		}
-	}
-	err := dev.IpcSet(cfg.String())
-	if err != nil {
-		return err
-	}
-	err = dev.Up()
-	if err != nil {
-		return err
-	}
-	h.dev = dev
-	return nil
-}
-
-func (h *Handler) resolveLocal(host string) (net.IP, error) {
-	if ip := net.ParseIP(host); ip != nil {
-		return ip, nil
-	}
-	ips, _, err := h.dns.LookupIP(host, dns.IPOption{IPv4Enable: true, IPv6Enable: true})
-	if err != nil {
-		return nil, err
-	}
-	got := ips
-	if h.streamSettings.SocketSettings != nil {
-		var got4, got6 []net.IP
-		for _, ip := range ips {
-			if ip.To4() != nil {
-				got4 = append(got4, ip)
-			} else {
-				got6 = append(got6, ip)
-			}
-		}
-		switch h.streamSettings.SocketSettings.DomainStrategy {
-		case internet.DomainStrategy_AS_IS, internet.DomainStrategy_USE_IP, internet.DomainStrategy_FORCE_IP:
-			got = ips
-		case internet.DomainStrategy_USE_IP4, internet.DomainStrategy_FORCE_IP4:
-			got = got4
-		case internet.DomainStrategy_USE_IP6, internet.DomainStrategy_FORCE_IP6:
-			got = got6
-		case internet.DomainStrategy_USE_IP46, internet.DomainStrategy_FORCE_IP46:
-			got = got4
-			if len(got) == 0 {
-				got = got6
-			}
-		case internet.DomainStrategy_USE_IP64, internet.DomainStrategy_FORCE_IP64:
-			got = got6
-			if len(got) == 0 {
+			switch streamSettings.SocketSettings.DomainStrategy {
+			case internet.DomainStrategy_AS_IS, internet.DomainStrategy_USE_IP, internet.DomainStrategy_FORCE_IP:
+				got = ips
+			case internet.DomainStrategy_USE_IP4, internet.DomainStrategy_FORCE_IP4:
 				got = got4
+			case internet.DomainStrategy_USE_IP6, internet.DomainStrategy_FORCE_IP6:
+				got = got6
+			case internet.DomainStrategy_USE_IP46, internet.DomainStrategy_FORCE_IP46:
+				got = got4
+				if len(got) == 0 {
+					got = got6
+				}
+			case internet.DomainStrategy_USE_IP64, internet.DomainStrategy_FORCE_IP64:
+				got = got6
+				if len(got) == 0 {
+					got = got4
+				}
+			}
+			if len(got) == 0 {
+				return nil, dns.ErrEmptyResponse
 			}
 		}
-		if len(got) == 0 {
-			return nil, dns.ErrEmptyResponse
-		}
+		return got[dice.Roll(len(got))], nil
 	}
-	return got[dice.Roll(len(got))], nil
 }
 
 type UDPConnClient struct {
@@ -510,3 +498,31 @@ func (c *cache) Cache(host string, saddr []string, ttl uint32) {
 		deadline: time.Now().Add(time.Second * time.Duration(ttl)),
 	}
 }
+
+type blockup struct {
+	tun.Device
+	ch    chan tun.Event
+	canup *atomic.Bool
+}
+
+func newblockup(t tun.Device) *blockup {
+	ch := make(chan tun.Event, cap(t.Events()))
+	canup := &atomic.Bool{}
+	go func() {
+		for e := range t.Events() {
+			if e&tun.EventUp != 0 && !canup.Load() {
+				continue
+			}
+			select {
+			case ch <- e:
+			default:
+			}
+		}
+		close(ch)
+	}()
+	return &blockup{Device: t, ch: ch, canup: canup}
+}
+
+func (b *blockup) Events() <-chan tun.Event {
+	return b.ch
+}

proxy/wireguard/server.go

@@ -255,7 +255,7 @@ func (s *Server) Start() error {
 	if s.src.Address.Family().IsDomain() {
 		return errors.New("address is domain")
 	}
-	listenFunc := func() (net.PacketConn, error) {
+	listenFunc := func(context.Context) (net.PacketConn, error) {
 		var pktConn net.PacketConn
 		var err error
 		if s.streamSettings.FinalMask != nil {

transport/internet/hysteria/conn.go

@@ -22,6 +22,12 @@ type interConn struct {
 
 	client bool
 	user   *protocol.MemoryUser
+	lock   chan struct{}
+	mu     sync.Mutex
+}
+
+func (c *interConn) unlock() {
+	<-c.lock
 }
 
 func (c *interConn) User() *protocol.MemoryUser {
@@ -33,6 +39,9 @@ func (c *interConn) Read(b []byte) (int, error) {
 }
 
 func (c *interConn) Write(b []byte) (int, error) {
+	c.lock <- struct{}{}
+	defer c.unlock()
+
 	if c.client {
 		c.client = false
 		if _, err := c.stream.Write(append(quicvarint.Append(nil, FrameTypeTCPRequest), b...)); err != nil {
@@ -40,11 +49,20 @@ func (c *interConn) Write(b []byte) (int, error) {
 		}
 		return len(b), nil
 	}
-
 	return c.stream.Write(b)
 }
 
 func (c *interConn) Close() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	select {
+	case c.lock <- struct{}{}:
+		defer c.unlock()
+	default:
+		c.stream.CancelRead(0)
+		c.stream.CancelWrite(0)
+		return nil
+	}
 	c.stream.CancelRead(0)
 	return c.stream.Close()
 }

transport/internet/hysteria/dialer.go

@@ -257,6 +257,7 @@ func (c *client) tcp(ctx context.Context) (stat.Connection, error) {
 		remote: c.conn.RemoteAddr(),
 
 		client: true,
+		lock:   make(chan struct{}, 1),
 	}, nil
 }
 

transport/internet/hysteria/hub.go

@@ -134,6 +134,7 @@ func (h *httpHandler) StreamDispatcher(ft http3.FrameType, stream *quic.Stream,
 			remote: h.conn.RemoteAddr(),
 
 			user: h.user,
+			lock: make(chan struct{}, 1),
 		})
 		return true, nil
 	default:

transport/internet/sockopt.go

@@ -28,3 +28,22 @@ func (v *SocketConfig) ParseTFOValue() int {
 	}
 	return tfo
 }
+
+// Dial should fail on bind errors, return this to indicate such errors.
+type BindError struct {
+	err error
+}
+
+func (e *BindError) Unwrap() error {
+	return e.err
+}
+
+func (e *BindError) Error() string {
+	return "failed to bind to interface: " + e.err.Error()
+}
+
+func NewBindError(err error) *BindError {
+	return &BindError{
+		err: err,
+	}
+}

transport/internet/sockopt_darwin.go

@@ -136,15 +136,15 @@ func applyOutboundSocketOptions(network string, address string, fd uintptr, conf
 	if config.Interface != "" {
 		iface, err := net.InterfaceByName(config.Interface)
 		if err != nil {
-			return errors.New("failed to get interface ", config.Interface).Base(err)
+			return NewBindError(errors.New("failed to get interface ", config.Interface).Base(err))
 		}
 		if network == "tcp6" || network == "udp6" {
 			if err := unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_BOUND_IF, iface.Index); err != nil {
-				return errors.New("failed to set IPV6_BOUND_IF").Base(err)
+				return NewBindError(errors.New("failed to set IPV6_BOUND_IF").Base(err))
 			}
 		} else {
 			if err := unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_BOUND_IF, iface.Index); err != nil {
-				return errors.New("failed to set IP_BOUND_IF").Base(err)
+				return NewBindError(errors.New("failed to set IP_BOUND_IF").Base(err))
 			}
 		}
 	}

transport/internet/sockopt_linux.go

@@ -22,7 +22,7 @@ func applyOutboundSocketOptions(network string, address string, fd uintptr, conf
 
 	if config.Interface != "" {
 		if err := syscall.BindToDevice(int(fd), config.Interface); err != nil {
-			return errors.New("failed to set Interface").Base(err)
+			return NewBindError(errors.New("failed to set Interface").Base(err))
 		}
 	}
 

transport/internet/sockopt_windows.go

@@ -35,7 +35,7 @@ func applyOutboundSocketOptions(network string, address string, fd uintptr, conf
 	if config.Interface != "" {
 		inf, err := net.InterfaceByName(config.Interface)
 		if err != nil {
-			return errors.New("failed to find the interface").Base(err)
+			return NewBindError(errors.New("failed to find the interface").Base(err))
 		}
 		// easy way to check if the address is ipv4
 		isV4 := strings.Contains(address, ".")
@@ -47,20 +47,20 @@ func applyOutboundSocketOptions(network string, address string, fd uintptr, conf
 			binary.BigEndian.PutUint32(bytes[:], uint32(inf.Index))
 			idx := *(*uint32)(unsafe.Pointer(&bytes[0]))
 			if err := syscall.SetsockoptInt(syscall.Handle(fd), syscall.IPPROTO_IP, IP_UNICAST_IF, int(idx)); err != nil {
-				return errors.New("failed to set IP_UNICAST_IF").Base(err)
+				return NewBindError(errors.New("failed to set IP_UNICAST_IF").Base(err))
 			}
 			if ip := net.ParseIP(host); ip != nil && ip.IsMulticast() && isUDPSocket(network) {
 				if err := syscall.SetsockoptInt(syscall.Handle(fd), syscall.IPPROTO_IP, syscall.IP_MULTICAST_IF, int(idx)); err != nil {
-					return errors.New("failed to set IP_MULTICAST_IF").Base(err)
+					return NewBindError(errors.New("failed to set IP_MULTICAST_IF").Base(err))
 				}
 			}
 		} else {
 			if err := syscall.SetsockoptInt(syscall.Handle(fd), syscall.IPPROTO_IPV6, IPV6_UNICAST_IF, inf.Index); err != nil {
-				return errors.New("failed to set IPV6_UNICAST_IF").Base(err)
+				return NewBindError(errors.New("failed to set IPV6_UNICAST_IF").Base(err))
 			}
 			if ip := net.ParseIP(host); ip != nil && ip.IsMulticast() && isUDPSocket(network) {
 				if err := syscall.SetsockoptInt(syscall.Handle(fd), syscall.IPPROTO_IPV6, syscall.IPV6_MULTICAST_IF, inf.Index); err != nil {
-					return errors.New("failed to set IPV6_MULTICAST_IF").Base(err)
+					return NewBindError(errors.New("failed to set IPV6_MULTICAST_IF").Base(err))
 				}
 			}
 		}

transport/internet/system_dialer.go

@@ -68,19 +68,27 @@ func (d *DefaultSystemDialer) Dial(ctx context.Context, src net.Address, dest ne
 			}
 		}
 		var lc net.ListenConfig
-		lc.Control = func(network, address string, c syscall.RawConn) error {
-			for _, ctl := range Controllers {
-				if err := ctl(network, address, c); err != nil {
-					errors.LogInfoInner(ctx, err, "failed to apply external controller")
-				}
-			}
-			return c.Control(func(fd uintptr) {
-				if sockopt != nil {
-					if err := applyOutboundSocketOptions(network, destAddr.String(), fd, sockopt); err != nil {
-						errors.LogInfo(ctx, err, "failed to apply socket options")
+		if sockopt != nil || len(Controllers) > 0 {
+			lc.Control = func(network, address string, c syscall.RawConn) error {
+				for _, ctl := range Controllers {
+					if err := ctl(network, address, c); err != nil {
+						return errors.New("failed to apply external controller").Base(err)
 					}
 				}
-			})
+				if sockopt != nil {
+					var sockErr, controlErr error
+					controlErr = c.Control(func(fd uintptr) {
+						if sockErr = applyOutboundSocketOptions(network, destAddr.String(), fd, sockopt); sockErr != nil {
+							errors.LogInfoInner(ctx, sockErr, "failed to apply socket options")
+						}
+					})
+					if _, ok := sockErr.(*BindError); ok {
+						return sockErr
+					}
+					return controlErr
+				}
+				return nil
+			}
 		}
 		packetConn, err := lc.ListenPacket(ctx, srcAddr.Network(), srcAddr.String())
 		if err != nil {
@@ -128,16 +136,22 @@ func (d *DefaultSystemDialer) Dial(ctx context.Context, src net.Address, dest ne
 		dialer.Control = func(network, address string, c syscall.RawConn) error {
 			for _, ctl := range Controllers {
 				if err := ctl(network, address, c); err != nil {
-					errors.LogInfoInner(ctx, err, "failed to apply external controller")
+					return errors.New("failed to apply external controller").Base(err)
 				}
 			}
-			return c.Control(func(fd uintptr) {
-				if sockopt != nil {
-					if err := applyOutboundSocketOptions(network, address, fd, sockopt); err != nil {
-						errors.LogInfoInner(ctx, err, "failed to apply socket options")
+			if sockopt != nil {
+				var sockErr, controlErr error
+				controlErr = c.Control(func(fd uintptr) {
+					if sockErr = applyOutboundSocketOptions(network, address, fd, sockopt); sockErr != nil {
+						errors.LogInfoInner(ctx, sockErr, "failed to apply socket options")
 					}
+				})
+				if _, ok := sockErr.(*BindError); ok {
+					return sockErr
 				}
-			})
+				return controlErr
+			}
+			return nil
 		}
 	}
 

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
 }

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) }

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 (
@@ -153,7 +154,7 @@ func jitter(backoff time.Duration) time.Duration {
 	if half <= 0 {
 		return backoff
 	}
-	return half + time.Duration(dice.Roll(int(half)))
+	return half + time.Duration(dice.RollInt63n(int64(half)))
 }
 
 func rateLimited(payload []byte) bool {

transport/internet/xdrive/drive_live_test.go

@@ -42,7 +42,7 @@ func liveDriveConfig(t *testing.T) *Config {
 
 	config := &Config{
 		RemoteFolder:      secrets.Folder,
-		Service:           "Google Drive",
+		Service:           "google-drive",
 		Secrets:           []string{secrets.ClientID, secrets.ClientSecret, secrets.RefreshToken},
 		SegmentBytes:      256 * 1024,
 		FlushIntervalMs:   100,

transport/internet/xdrive/drive_test.go

@@ -317,7 +317,7 @@ func driveSettings() *internet.MemoryStreamConfig {
 		ProtocolName: protocolName,
 		ProtocolSettings: &Config{
 			RemoteFolder:      "folder-id",
-			Service:           "Google Drive",
+			Service:           "google-drive",
 			Secrets:           []string{"client", "secret", "refresh"},
 			FlushIntervalMs:   5,
 			PollIntervalMs:    5,
@@ -564,6 +564,16 @@ func TestDriveSharedClient(t *testing.T) {
 	}
 }
 
+func TestDriveOldServiceName(t *testing.T) {
+	newFakeDrive(t)
+
+	settings := driveSettings()
+	settings.ProtocolSettings.(*Config).Service = "Google Drive"
+	if _, err := newStorage(settings); err != nil {
+		t.Fatalf("newStorage: %v", err)
+	}
+}
+
 func TestDriveInlineListing(t *testing.T) {
 	drive := newFakeDrive(t)
 	storage := newDriveBackend(t)

transport/internet/xdrive/storage.go

@@ -35,7 +35,7 @@ func newStorage(streamSettings *internet.MemoryStreamConfig) (Storage, error) {
 	switch config.Service {
 	case "local":
 		return newLocalStorage(config.RemoteFolder)
-	case "Google Drive":
+	case "google-drive", "Google Drive":
 		return sharedStorage(streamSettings, config, func() (Storage, error) {
 			return newDriveStorage(streamSettings, config)
 		})