mirror of
https://github.com/XTLS/Xray-core.git
synced 2026-10-25 00:51:37 +00:00
XDRIVE transport: Support h3 with "alpn" ["h3"]
This commit is contained in:
parent
39fb9f4e53
commit
3fdaf41a4c
5 changed files with 604 additions and 11 deletions
|
|
@ -90,6 +90,7 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
|
|||
config.ProtocolName = protocol
|
||||
}
|
||||
|
||||
var xdriveH3 bool
|
||||
switch strings.ToLower(c.Security) {
|
||||
case "", "none":
|
||||
case "tls":
|
||||
|
|
@ -104,6 +105,7 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
|
|||
tm := serial.ToTypedMessage(ts)
|
||||
config.SecuritySettings = append(config.SecuritySettings, tm)
|
||||
config.SecurityType = tm.Type
|
||||
xdriveH3 = config.ProtocolName == "xdrive" && tlsSettings.ALPN != nil && slices.Equal(*tlsSettings.ALPN, StringList{"h3"})
|
||||
if config.ProtocolName == "xdrive" && tlsSettings.ALPN != nil && slices.Equal(*tlsSettings.ALPN, StringList{"http/1.1"}) && tlsSettings.ECHConfigList == "" {
|
||||
// a fingerprint puts its own ALPN into the hello, except Go TLS and the
|
||||
// randomized ones, which take it from the config or send none
|
||||
|
|
@ -219,6 +221,7 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
|
|||
})
|
||||
}
|
||||
if c.XDRIVESettings != nil {
|
||||
c.XDRIVESettings.h3 = xdriveH3
|
||||
xs, err := c.XDRIVESettings.Build()
|
||||
if err != nil {
|
||||
return nil, errors.New("Failed to build XDRIVE config.").Base(err)
|
||||
|
|
@ -260,5 +263,20 @@ func (c *StreamConfig) Build() (*internet.StreamConfig, error) {
|
|||
}
|
||||
}
|
||||
|
||||
if xdriveH3 && c.FinalMask != nil {
|
||||
if c.XDRIVESettings != nil && c.XDRIVESettings.Service == "google-drive" {
|
||||
// noise only sends datagrams of its own ahead of the QUIC ones, the
|
||||
// other masks change those or need a peer that undoes them
|
||||
for _, mask := range c.FinalMask.Udp {
|
||||
if strings.ToLower(mask.Type) != "noise" {
|
||||
return nil, errors.New(`XDRIVE: over h3 to Google Drive the only "udp" mask that works is "noise", not "`, mask.Type, `"`)
|
||||
}
|
||||
}
|
||||
}
|
||||
if config.QuicParams != nil && (config.QuicParams.Congestion == "brutal" || config.QuicParams.Congestion == "force-brutal") {
|
||||
return nil, errors.New(`XDRIVE: brutal is for Hysteria servers, over h3 "congestion" is "bbr" or "reno", not "`, config.QuicParams.Congestion, `"`)
|
||||
}
|
||||
}
|
||||
|
||||
return config, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -998,6 +998,9 @@ type XDriveConfig struct {
|
|||
HoleTimeoutMs uint32 `json:"holeTimeoutMs"`
|
||||
Template json.RawMessage `json:"template"`
|
||||
Xmux XmuxConfig `json:"xmux"`
|
||||
|
||||
// set by StreamConfig when "alpn" is ["h3"]
|
||||
h3 bool
|
||||
}
|
||||
|
||||
// Build implements Buildable.
|
||||
|
|
@ -1025,9 +1028,9 @@ func (c *XDriveConfig) Build() (proto.Message, error) {
|
|||
if c.Xmux == (XmuxConfig{}) {
|
||||
c.Xmux.MaxConnections.From = 1
|
||||
c.Xmux.MaxConnections.To = 1
|
||||
if c.Service == "google-drive" {
|
||||
if c.Service == "google-drive" && !c.h3 {
|
||||
// Google gives an h2 connection 1 MiB of window, which a few sessions on
|
||||
// a long rtt would share
|
||||
// a long rtt would share, an h3 connection has no such limit
|
||||
c.Xmux.MaxConnections.From = 4
|
||||
c.Xmux.MaxConnections.To = 4
|
||||
}
|
||||
|
|
|
|||
|
|
@ -425,6 +425,81 @@ func TestXDriveHTTP1ALPNWarning(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// over h3 Google Drive takes only the noise mask, and no service takes brutal
|
||||
func TestXDriveH3Checks(t *testing.T) {
|
||||
const (
|
||||
drive = `"method": "xdrive", "xdriveSettings": {"remoteFolder": "f", "service": "google-drive", "secrets": ["a", "b", "c"]}`
|
||||
local = `"method": "xdrive", "xdriveSettings": {"remoteFolder": "/tmp/xdrive", "service": "local"}`
|
||||
noise = `{"type": "Noise", "settings": {"noise": [{"rand": "50-100"}]}}`
|
||||
salamander = `{"type": "salamander", "settings": {"password": "p"}}`
|
||||
forceBrutal = `{"quicParams": {"congestion": "force-brutal", "brutalUp": "10 mbps"}}`
|
||||
)
|
||||
for _, c := range []struct {
|
||||
alpn, transport, finalmask string
|
||||
ok bool
|
||||
}{
|
||||
{`["h3"]`, drive, `{"udp": [` + noise + `]}`, true},
|
||||
{`["h3"]`, drive, `{"udp": [` + salamander + `]}`, false},
|
||||
{`["h3"]`, drive, `{"udp": [` + noise + `, ` + salamander + `]}`, false},
|
||||
{`["h3"]`, local, `{"udp": [` + salamander + `]}`, true},
|
||||
{`["h3"]`, `"method": "xdrive"`, `{"udp": [` + salamander + `]}`, true},
|
||||
{`["h2"]`, drive, `{"udp": [` + salamander + `]}`, true},
|
||||
{`null`, drive, `{"udp": [` + salamander + `]}`, true},
|
||||
{`["h3"]`, drive, `null`, true},
|
||||
{`["h3"]`, drive, `{"quicParams": {"congestion": "bbr"}}`, true},
|
||||
{`["h3"]`, drive, `{"quicParams": {"congestion": "brutal"}}`, false},
|
||||
{`["h3"]`, local, forceBrutal, false},
|
||||
{`["h3", "h2"]`, drive, `{"quicParams": {"congestion": "brutal"}}`, true},
|
||||
{`["h3"]`, `"method": "xhttp"`, forceBrutal, true},
|
||||
} {
|
||||
config := new(StreamConfig)
|
||||
if err := json.Unmarshal([]byte(`{
|
||||
`+c.transport+`,
|
||||
"security": "tls",
|
||||
"tlsSettings": {"alpn": `+c.alpn+`},
|
||||
"finalmask": `+c.finalmask+`
|
||||
}`), config); err != nil {
|
||||
t.Fatalf("Unmarshal: %v", err)
|
||||
}
|
||||
if _, err := config.Build(); (err == nil) != c.ok {
|
||||
t.Errorf("alpn %s, %s, finalmask %s: %v", c.alpn, c.transport, c.finalmask, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// one h3 connection is not held back by a window like h2, so h3 takes one by default
|
||||
func TestXDriveH3Connections(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
alpn string
|
||||
want int32
|
||||
}{
|
||||
{`["h3"]`, 1},
|
||||
{`["h2"]`, 4},
|
||||
{`null`, 4},
|
||||
} {
|
||||
config := new(StreamConfig)
|
||||
if err := json.Unmarshal([]byte(`{
|
||||
"method": "xdrive",
|
||||
"xdriveSettings": {"remoteFolder": "f", "service": "google-drive", "secrets": ["a", "b", "c"]},
|
||||
"security": "tls",
|
||||
"tlsSettings": {"alpn": `+c.alpn+`}
|
||||
}`), config); err != nil {
|
||||
t.Fatalf("Unmarshal: %v", err)
|
||||
}
|
||||
built, err := config.Build()
|
||||
if err != nil {
|
||||
t.Fatalf("alpn %s: %v", c.alpn, err)
|
||||
}
|
||||
xs, err := built.TransportSettings[0].Settings.GetInstance()
|
||||
if err != nil {
|
||||
t.Fatalf("alpn %s: %v", c.alpn, err)
|
||||
}
|
||||
if got := xs.(*xdrive.Config).Xmux.MaxConnections; got.From != c.want || got.To != c.want {
|
||||
t.Errorf("alpn %s: maxConnections is %d-%d, want %d", c.alpn, got.From, got.To, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestXDriveTemplateNeedsTemplate(t *testing.T) {
|
||||
config := new(XDriveConfig)
|
||||
if err := json.Unmarshal([]byte(`{"remoteFolder": "f", "service": "template"}`), config); err != nil {
|
||||
|
|
|
|||
|
|
@ -2,14 +2,23 @@ package xdrive
|
|||
|
||||
import (
|
||||
"context"
|
||||
gotls "crypto/tls"
|
||||
"io"
|
||||
"net/http"
|
||||
"reflect"
|
||||
"runtime"
|
||||
"slices"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/apernet/quic-go"
|
||||
"github.com/apernet/quic-go/http3"
|
||||
"github.com/xtls/xray-core/common/errors"
|
||||
"github.com/xtls/xray-core/common/net"
|
||||
"github.com/xtls/xray-core/common/net/cnc"
|
||||
"github.com/xtls/xray-core/transport/internet"
|
||||
"github.com/xtls/xray-core/transport/internet/hysteria/congestion"
|
||||
"github.com/xtls/xray-core/transport/internet/hysteria/congestion/bbr"
|
||||
"github.com/xtls/xray-core/transport/internet/splithttp"
|
||||
"github.com/xtls/xray-core/transport/internet/tls"
|
||||
)
|
||||
|
|
@ -113,6 +122,8 @@ func newServiceClient(connCtx context.Context, h1Hosts *sync.Map, streamSettings
|
|||
sockopt *internet.SocketConfig
|
||||
fronting *net.Destination
|
||||
keepAlive = net.ChromeH2KeepAlivePeriod
|
||||
// h3 keeps alive only when asked to, see below
|
||||
h3KeepAlive time.Duration
|
||||
)
|
||||
if streamSettings != nil {
|
||||
tlsConfig = tls.ConfigFromStreamSettings(streamSettings)
|
||||
|
|
@ -122,6 +133,7 @@ func newServiceClient(connCtx context.Context, h1Hosts *sync.Map, streamSettings
|
|||
switch period := config.Xmux.HKeepAlivePeriod; {
|
||||
case period > 0:
|
||||
keepAlive = time.Duration(period) * time.Second
|
||||
h3KeepAlive = keepAlive
|
||||
case period < 0:
|
||||
keepAlive = 0
|
||||
}
|
||||
|
|
@ -132,8 +144,10 @@ func newServiceClient(connCtx context.Context, h1Hosts *sync.Map, streamSettings
|
|||
}
|
||||
fingerprint := tls.GetFingerprint(tlsConfig.Fingerprint)
|
||||
|
||||
dial := func(ctx context.Context, addr string) (net.Conn, net.Destination, error) {
|
||||
host, err := net.ParseDestination("tcp:" + addr)
|
||||
// fronting takes only the address and port, the network stays that of the
|
||||
// dial, which is UDP for h3 though the destination is built as TCP
|
||||
dial := func(ctx context.Context, network, addr string) (net.Conn, net.Destination, error) {
|
||||
host, err := net.ParseDestination(network + ":" + addr)
|
||||
if err != nil {
|
||||
return nil, host, errors.New("bad address: ", addr).Base(err)
|
||||
}
|
||||
|
|
@ -147,14 +161,22 @@ func newServiceClient(connCtx context.Context, h1Hosts *sync.Map, streamSettings
|
|||
}
|
||||
|
||||
var conn net.Conn
|
||||
if streamSettings.FinalMask != nil {
|
||||
conn, err = streamSettings.FinalMask.DialTCP(ctx, target)
|
||||
} else {
|
||||
switch {
|
||||
case streamSettings.FinalMask == nil:
|
||||
conn, err = internet.DialSystem(ctx, target, sockopt)
|
||||
case target.Network == net.Network_UDP:
|
||||
conn, err = streamSettings.FinalMask.DialUDP(ctx, target)
|
||||
default:
|
||||
conn, err = streamSettings.FinalMask.DialTCP(ctx, target)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, host, errors.New("failed to dial to dest").Base(err)
|
||||
}
|
||||
// h3 closes its QUIC connection instead, as the server never hears of a UDP
|
||||
// socket that closes
|
||||
if network == "udp" {
|
||||
return conn, host, nil
|
||||
}
|
||||
// CloseIdleConnections leaves an h2 connection open while a stream just
|
||||
// canceled has not left it yet
|
||||
stop := context.AfterFunc(connCtx, func() { conn.Close() })
|
||||
|
|
@ -162,13 +184,13 @@ func newServiceClient(connCtx context.Context, h1Hosts *sync.Map, streamSettings
|
|||
}
|
||||
|
||||
dialPlain := func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||
conn, _, err := dial(ctx, addr)
|
||||
conn, _, err := dial(ctx, "tcp", addr)
|
||||
return conn, err
|
||||
}
|
||||
|
||||
// http.Transport picks h2 or h1 by the ALPN in the ConnectionState of the conn
|
||||
dialTLS := func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||
conn, host, err := dial(ctx, addr)
|
||||
conn, host, err := dial(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -203,6 +225,85 @@ func newServiceClient(connCtx context.Context, h1Hosts *sync.Map, streamSettings
|
|||
}
|
||||
}
|
||||
|
||||
var https http.RoundTripper = &secureTransport{h2: secure(1), h1: secure(maxConns), h1Hosts: h1Hosts}
|
||||
// no server offers h3 by ALPN over TCP, so it is used only when asked for alone
|
||||
if slices.Equal(tlsConfig.NextProtocol, []string{"h3"}) {
|
||||
quicParams := streamSettings.QuicParams
|
||||
if quicParams == nil {
|
||||
quicParams = &internet.QuicParams{BbrProfile: string(bbr.ProfileStandard)}
|
||||
}
|
||||
// the defaults are those of quic-go, with no keepalive unless asked for, as
|
||||
// in DoH: http3 never closes a connection that is idle or that it dropped
|
||||
// after an error, so only the idle timeout or the end of connCtx ends it
|
||||
quicConfig := &quic.Config{
|
||||
InitialStreamReceiveWindow: quicParams.InitStreamReceiveWindow,
|
||||
MaxStreamReceiveWindow: quicParams.MaxStreamReceiveWindow,
|
||||
InitialConnectionReceiveWindow: quicParams.InitConnReceiveWindow,
|
||||
MaxConnectionReceiveWindow: quicParams.MaxConnReceiveWindow,
|
||||
MaxIdleTimeout: time.Duration(quicParams.MaxIdleTimeout) * time.Second,
|
||||
KeepAlivePeriod: time.Duration(quicParams.KeepAlivePeriod) * time.Second,
|
||||
MaxIncomingStreams: quicParams.MaxIncomingStreams,
|
||||
DisablePathMTUDiscovery: quicParams.DisablePathMtuDiscovery || (runtime.GOOS != "linux" && runtime.GOOS != "windows" && runtime.GOOS != "darwin"),
|
||||
ChromeParrot: !quicParams.DisableChromeParrot,
|
||||
}
|
||||
if quicConfig.KeepAlivePeriod == 0 {
|
||||
quicConfig.KeepAlivePeriod = h3KeepAlive
|
||||
}
|
||||
|
||||
https = &http3.Transport{
|
||||
QUICConfig: quicConfig,
|
||||
Dial: func(ctx context.Context, addr string, _ *gotls.Config, cfg *quic.Config) (*quic.Conn, error) {
|
||||
switch quicParams.Congestion {
|
||||
case "", "bbr", "reno":
|
||||
default:
|
||||
// brutal is for Hysteria servers, not services, and a panic here, in
|
||||
// a goroutine of http3, would end the process
|
||||
return nil, errors.New("unsupported congestion control over h3: ", quicParams.Congestion)
|
||||
}
|
||||
|
||||
conn, host, err := dial(ctx, "udp", addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var pktConn net.PacketConn
|
||||
var udpAddr net.Addr
|
||||
switch c := conn.(type) {
|
||||
case *net.PacketConnWrapper:
|
||||
pktConn = c.PacketConn
|
||||
udpAddr = c.RemoteAddr()
|
||||
case *cnc.Connection:
|
||||
pktConn = &internet.FakePacketConn{Conn: c}
|
||||
udpAddr = &net.UDPAddr{IP: []byte{0, 0, 0, 0}}
|
||||
default:
|
||||
conn.Close()
|
||||
return nil, errors.New("failed to dial to dest: unexpected ", reflect.TypeOf(c))
|
||||
}
|
||||
|
||||
// built for each dial with the host as SNI, as for h2: http3 sets the
|
||||
// SNI on a copy, which a pinned CA and the ECH lookup do not see
|
||||
gotlsConfig := tlsConfig.GetTLSConfig(tls.WithDestination(host))
|
||||
// it is for servers, and ChromeParrot refuses a config that has it
|
||||
gotlsConfig.GetCertificate = nil
|
||||
tr := &quic.Transport{Conn: pktConn, DisableGSO: quicParams.DisableGSO}
|
||||
if !quicParams.DisableChromeParrot {
|
||||
tr.ConnectionIDGenerator = quic.ZeroLengthConnectionIDGenerator{}
|
||||
}
|
||||
qconn, err := tr.Dial(ctx, udpAddr, gotlsConfig, cfg)
|
||||
if err != nil {
|
||||
tr.Close()
|
||||
pktConn.Close()
|
||||
return nil, err
|
||||
}
|
||||
stop := context.AfterFunc(connCtx, func() { qconn.CloseWithError(quic.ApplicationErrorCode(http3.ErrCodeNoError), "") })
|
||||
context.AfterFunc(qconn.Context(), func() { stop(); tr.Close(); pktConn.Close() })
|
||||
if quicParams.Congestion != "reno" {
|
||||
congestion.UseBBR(qconn, bbr.Profile(quicParams.BbrProfile))
|
||||
}
|
||||
return qconn, nil
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
return &http.Client{
|
||||
Transport: &serviceTransport{
|
||||
plain: &http.Transport{
|
||||
|
|
@ -212,7 +313,7 @@ func newServiceClient(connCtx context.Context, h1Hosts *sync.Map, streamSettings
|
|||
MaxIdleConnsPerHost: maxConns,
|
||||
MaxConnsPerHost: maxConns,
|
||||
},
|
||||
secure: &secureTransport{h2: secure(1), h1: secure(maxConns), h1Hosts: h1Hosts},
|
||||
secure: https,
|
||||
stall: stall,
|
||||
},
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import (
|
|||
"bytes"
|
||||
"context"
|
||||
gotls "crypto/tls"
|
||||
"crypto/x509"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"io"
|
||||
|
|
@ -11,13 +12,20 @@ import (
|
|||
"net/http"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/apernet/quic-go"
|
||||
"github.com/apernet/quic-go/http3"
|
||||
xnet "github.com/xtls/xray-core/common/net"
|
||||
"github.com/xtls/xray-core/common/net/cnc"
|
||||
"github.com/xtls/xray-core/common/protocol/tls/cert"
|
||||
"github.com/xtls/xray-core/transport/internet"
|
||||
"github.com/xtls/xray-core/transport/internet/finalmask"
|
||||
"github.com/xtls/xray-core/transport/internet/finalmask/noise"
|
||||
"github.com/xtls/xray-core/transport/internet/splithttp"
|
||||
"github.com/xtls/xray-core/transport/internet/tls"
|
||||
)
|
||||
|
|
@ -552,7 +560,8 @@ type roundTrip func(*http.Request) (*http.Response, error)
|
|||
|
||||
func (f roundTrip) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
|
||||
|
||||
// "hKeepAlivePeriod" of XMUX sets when an idle h2 connection pings, -1 never
|
||||
// "hKeepAlivePeriod" of XMUX sets when an idle h2 connection pings, -1 never,
|
||||
// and h3 keeps alive only when it is set
|
||||
func TestServiceClientKeepAlive(t *testing.T) {
|
||||
for period, want := range map[int64]time.Duration{0: xnet.ChromeH2KeepAlivePeriod, 20: 20 * time.Second, -1: 0} {
|
||||
settings := &internet.MemoryStreamConfig{ProtocolSettings: &Config{Xmux: &splithttp.XmuxConfig{HKeepAlivePeriod: period}}}
|
||||
|
|
@ -562,4 +571,391 @@ func TestServiceClientKeepAlive(t *testing.T) {
|
|||
t.Errorf("hKeepAlivePeriod %d pings after %v, want %v", period, got, want)
|
||||
}
|
||||
}
|
||||
for period, want := range map[int64]time.Duration{0: 0, 20: 20 * time.Second, -1: 0} {
|
||||
settings := &internet.MemoryStreamConfig{
|
||||
ProtocolSettings: &Config{Xmux: &splithttp.XmuxConfig{HKeepAlivePeriod: period}},
|
||||
SecuritySettings: &tls.Config{NextProtocol: []string{"h3"}},
|
||||
}
|
||||
client := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8)
|
||||
if got := client.Transport.(*serviceTransport).secure.(*http3.Transport).QUICConfig.KeepAlivePeriod; got != want {
|
||||
t.Errorf("hKeepAlivePeriod %d keeps h3 alive every %v, want %v", period, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// h3 is used when "alpn" asks for it alone, and only then
|
||||
func TestServiceClientH3Selection(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
alpn []string
|
||||
h3 bool
|
||||
}{
|
||||
{nil, false},
|
||||
{[]string{"h2"}, false},
|
||||
{[]string{"http/1.1"}, false},
|
||||
{[]string{"h3", "h2"}, false},
|
||||
{[]string{"h2", "h3"}, false},
|
||||
{[]string{"h3"}, true},
|
||||
} {
|
||||
client := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{SecuritySettings: &tls.Config{NextProtocol: c.alpn}}, 5*time.Second, 8)
|
||||
if _, h3 := client.Transport.(*serviceTransport).secure.(*http3.Transport); h3 != c.h3 {
|
||||
t.Errorf("alpn %v: h3 %v, want %v", c.alpn, h3, c.h3)
|
||||
}
|
||||
}
|
||||
if _, h3 := newServiceClient(context.Background(), new(sync.Map), &internet.MemoryStreamConfig{}, 5*time.Second, 8).Transport.(*serviceTransport).secure.(*http3.Transport); h3 {
|
||||
t.Error("h3 without tlsSettings")
|
||||
}
|
||||
}
|
||||
|
||||
// h3Server is an http3 server that keeps the SNI of each connection, and counts
|
||||
// its connections and those closed
|
||||
type h3Server struct {
|
||||
port xnet.Port
|
||||
hash, caHash [32]byte
|
||||
mu sync.Mutex
|
||||
snis []string
|
||||
conns atomic.Int32
|
||||
closed atomic.Int32
|
||||
}
|
||||
|
||||
func newH3Server(t *testing.T, handler http.Handler) *h3Server {
|
||||
t.Helper()
|
||||
|
||||
// ECDSA, as ChromeParrot offers no Ed25519, from a CA of its own
|
||||
ca, caHash := cert.MustGenerate(nil, cert.Authority(true), cert.KeyUsage(x509.KeyUsageCertSign))
|
||||
certificate, hash := cert.MustGenerate(ca, cert.DNSNames("www.googleapis.com", "oauth2.googleapis.com"))
|
||||
key, err := x509.ParsePKCS8PrivateKey(certificate.PrivateKey)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
conn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s := &h3Server{port: xnet.Port(conn.LocalAddr().(*net.UDPAddr).Port), hash: hash, caHash: caHash}
|
||||
server := &http3.Server{
|
||||
Handler: handler,
|
||||
TLSConfig: http3.ConfigureTLSConfig(&gotls.Config{
|
||||
Certificates: []gotls.Certificate{{Certificate: [][]byte{certificate.Certificate, ca.Certificate}, PrivateKey: key}},
|
||||
GetConfigForClient: func(hello *gotls.ClientHelloInfo) (*gotls.Config, error) {
|
||||
s.mu.Lock()
|
||||
s.snis = append(s.snis, hello.ServerName)
|
||||
s.mu.Unlock()
|
||||
return nil, nil
|
||||
},
|
||||
}),
|
||||
ConnContext: func(ctx context.Context, c *quic.Conn) context.Context {
|
||||
s.conns.Add(1)
|
||||
context.AfterFunc(c.Context(), func() { s.closed.Add(1) })
|
||||
return ctx
|
||||
},
|
||||
}
|
||||
go server.Serve(conn)
|
||||
t.Cleanup(func() {
|
||||
server.Close()
|
||||
conn.Close()
|
||||
})
|
||||
return s
|
||||
}
|
||||
|
||||
// settings sends every host to the server over h3, through a destination built
|
||||
// as TCP the way conf builds it, and pins the certificate of the server in config
|
||||
func (s *h3Server) settings(config *tls.Config) *internet.MemoryStreamConfig {
|
||||
config.NextProtocol = []string{"h3"}
|
||||
config.PinnedPeerCertSha256 = [][]byte{s.hash[:]}
|
||||
return &internet.MemoryStreamConfig{
|
||||
ProtocolName: protocolName,
|
||||
Destination: &xnet.Destination{Address: xnet.LocalHostIP, Port: s.port, Network: xnet.Network_TCP},
|
||||
SecuritySettings: config,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *h3Server) seenSNIs() []string {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return slices.Sorted(slices.Values(s.snis))
|
||||
}
|
||||
|
||||
// one client serves two hosts, and each gets its own SNI unless serverName
|
||||
// says otherwise, which is also the name a pinned CA checks the certificate for
|
||||
func TestServiceClientH3SNI(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
serverName string
|
||||
pinCA bool
|
||||
want []string
|
||||
}{
|
||||
{"", false, []string{"oauth2.googleapis.com", "www.googleapis.com"}},
|
||||
{"www.google.com", false, []string{"www.google.com", "www.google.com"}},
|
||||
{"", true, []string{"oauth2.googleapis.com", "www.googleapis.com"}},
|
||||
} {
|
||||
server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
|
||||
settings := server.settings(&tls.Config{ServerName: c.serverName})
|
||||
if c.pinCA {
|
||||
settings.SecuritySettings.(*tls.Config).PinnedPeerCertSha256 = [][]byte{server.caHash[:]}
|
||||
}
|
||||
client := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8)
|
||||
for _, url := range []string{"https://www.googleapis.com/drive/v3/files", "https://oauth2.googleapis.com/token"} {
|
||||
resp, err := client.Get(url)
|
||||
if err != nil {
|
||||
t.Fatalf("serverName %q, pinned CA %v, %s: %v", c.serverName, c.pinCA, url, err)
|
||||
}
|
||||
resp.Body.Close()
|
||||
if resp.Proto != "HTTP/3.0" {
|
||||
t.Errorf("serverName %q, %s: went over %s", c.serverName, url, resp.Proto)
|
||||
}
|
||||
}
|
||||
if snis := server.seenSNIs(); !slices.Equal(snis, c.want) {
|
||||
t.Errorf("serverName %q: SNIs %v, want %v", c.serverName, snis, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// a request that stands still ends over h3 too, and says so
|
||||
func TestServiceClientStallOverH3(t *testing.T) {
|
||||
const stall = 200 * time.Millisecond
|
||||
server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path == "/stuck" {
|
||||
w.Write([]byte("x"))
|
||||
w.(http.Flusher).Flush()
|
||||
}
|
||||
time.Sleep(5 * stall)
|
||||
}))
|
||||
client := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{}), stall, 8)
|
||||
|
||||
for _, path := range []string{"/stuck", "/silent"} {
|
||||
start := time.Now()
|
||||
resp, err := client.Get("https://www.googleapis.com" + path)
|
||||
if err == nil {
|
||||
if resp.Proto != "HTTP/3.0" {
|
||||
t.Fatalf("%s went over %s", path, resp.Proto)
|
||||
}
|
||||
_, err = io.ReadAll(resp.Body)
|
||||
resp.Body.Close()
|
||||
}
|
||||
if !errors.Is(err, errStalled) {
|
||||
t.Errorf("%s gave %v, want %v", path, err, errStalled)
|
||||
}
|
||||
if took := time.Since(start); took > 4*stall {
|
||||
t.Errorf("%s took %v to give up", path, took)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// brutal and force-brutal, which are for Hysteria servers, give an error and
|
||||
// no panic in a goroutine of http3, where XHTTP and DoH panic on brutal
|
||||
func TestServiceClientH3Congestion(t *testing.T) {
|
||||
server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
|
||||
for _, cc := range []string{"", "bbr", "reno", "brutal", "force-brutal"} {
|
||||
settings := server.settings(&tls.Config{})
|
||||
settings.QuicParams = &internet.QuicParams{Congestion: cc, BrutalUp: 1 << 20}
|
||||
resp, err := newServiceClient(context.Background(), new(sync.Map), settings, 5*time.Second, 8).Get("https://www.googleapis.com/")
|
||||
if err == nil {
|
||||
resp.Body.Close()
|
||||
}
|
||||
if brutal := strings.HasSuffix(cc, "brutal"); brutal != (err != nil) || brutal && !strings.Contains(err.Error(), cc) {
|
||||
t.Errorf("congestion %q gave %v", cc, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// recordConn keeps the datagrams sent through it
|
||||
type recordConn struct {
|
||||
net.PacketConn
|
||||
mu sync.Mutex
|
||||
sent [][]byte
|
||||
closed atomic.Bool
|
||||
}
|
||||
|
||||
func (c *recordConn) WriteTo(p []byte, addr net.Addr) (int, error) {
|
||||
c.mu.Lock()
|
||||
c.sent = append(c.sent, bytes.Clone(p))
|
||||
c.mu.Unlock()
|
||||
return c.PacketConn.WriteTo(p, addr)
|
||||
}
|
||||
|
||||
func (c *recordConn) Close() error {
|
||||
c.closed.Store(true)
|
||||
return c.PacketConn.Close()
|
||||
}
|
||||
|
||||
// recorder dials UDP through recordConns and keeps them
|
||||
type recorder struct {
|
||||
mu sync.Mutex
|
||||
conns []*recordConn
|
||||
}
|
||||
|
||||
func (r *recorder) dialUDP(ctx context.Context, dest xnet.Destination) (net.PacketConn, net.Addr, error) {
|
||||
conn, err := internet.DialSystem(ctx, dest, nil)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
rec := &recordConn{PacketConn: conn.(*xnet.PacketConnWrapper).PacketConn}
|
||||
r.mu.Lock()
|
||||
r.conns = append(r.conns, rec)
|
||||
r.mu.Unlock()
|
||||
return rec, conn.RemoteAddr(), nil
|
||||
}
|
||||
|
||||
// a connection ends with its socket once it is idle for its idle timeout, as
|
||||
// keepalive is off unless asked for, and so does one whose handshake failed,
|
||||
// and neither leaves anything on the context that closes the connections
|
||||
func TestServiceClientH3Closes(t *testing.T) {
|
||||
server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
for _, pin := range [][]byte{server.hash[:], make([]byte, 32)} {
|
||||
rec := new(recorder)
|
||||
settings := server.settings(&tls.Config{})
|
||||
settings.SecuritySettings.(*tls.Config).PinnedPeerCertSha256 = [][]byte{pin}
|
||||
// ChromeParrot would keep the idle timeout at 30 s
|
||||
settings.QuicParams = &internet.QuicParams{MaxIdleTimeout: 1, DisableChromeParrot: true}
|
||||
settings.FinalMask = finalmask.NewFinalMask(nil, nil, nil, nil, rec.dialUDP, nil)
|
||||
connCtx := &afterFuncs{Context: ctx}
|
||||
resp, err := newServiceClient(connCtx, new(sync.Map), settings, 5*time.Second, 8).Get("https://www.googleapis.com/")
|
||||
if err == nil {
|
||||
resp.Body.Close()
|
||||
}
|
||||
if bytes.Equal(pin, server.hash[:]) != (err == nil) {
|
||||
t.Fatalf("pin %x: %v", pin[:4], err)
|
||||
}
|
||||
|
||||
closed := func() bool {
|
||||
rec.mu.Lock()
|
||||
defer rec.mu.Unlock()
|
||||
return len(rec.conns) == 1 && rec.conns[0].closed.Load()
|
||||
}
|
||||
for deadline := time.Now().Add(5 * time.Second); !closed() && time.Now().Before(deadline); {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
if !closed() {
|
||||
t.Errorf("pin %x: the socket was left open", pin[:4])
|
||||
}
|
||||
waitFor(t, "the closed connection to leave the context", func() bool { return connCtx.live.Load() == 0 })
|
||||
}
|
||||
}
|
||||
|
||||
// cncDialer gives UDP as a stream conn, the way dialerProxy does
|
||||
type cncDialer struct{}
|
||||
|
||||
func (cncDialer) Dial(ctx context.Context, src xnet.Address, dest xnet.Destination, sockopt *internet.SocketConfig) (net.Conn, error) {
|
||||
conn, err := net.Dial("udp", dest.NetAddr())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return cnc.NewConnection(cnc.ConnectionInput(conn), cnc.ConnectionOutput(conn), cnc.ConnectionOnClose(conn)), nil
|
||||
}
|
||||
|
||||
func (cncDialer) DestIpAddress() net.IP { return nil }
|
||||
|
||||
// h3 goes over the stream conn of a dialerProxy too
|
||||
func TestServiceClientH3DialerProxy(t *testing.T) {
|
||||
internet.UseAlternativeSystemDialer(cncDialer{})
|
||||
defer internet.UseAlternativeSystemDialer(nil)
|
||||
|
||||
server := newH3Server(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
|
||||
resp, err := newServiceClient(context.Background(), new(sync.Map), server.settings(&tls.Config{}), 5*time.Second, 8).Get("https://www.googleapis.com/")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
resp.Body.Close()
|
||||
if resp.Proto != "HTTP/3.0" {
|
||||
t.Errorf("went over %s", resp.Proto)
|
||||
}
|
||||
}
|
||||
|
||||
// xdrive moves data over h3 to a fake Drive, with the noise of a UDP mask sent
|
||||
// ahead of the Initial of each connection
|
||||
func TestDriveOverH3(t *testing.T) {
|
||||
drive := newFakeDrive(t)
|
||||
server := newH3Server(t, drive.server.Config.Handler)
|
||||
driveTokenURL = "https://oauth2.googleapis.com/token"
|
||||
driveFilesURL = "https://www.googleapis.com/files"
|
||||
driveUploadURL = "https://www.googleapis.com/upload"
|
||||
|
||||
rec := new(recorder)
|
||||
settings := server.settings(&tls.Config{})
|
||||
settings.ProtocolSettings = driveSettings().ProtocolSettings
|
||||
settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{MaxConnections: &splithttp.RangeConfig{From: 4, To: 4}}
|
||||
masks := []finalmask.UDPMask{&noise.Config{Items: []*noise.Item{{Packet: []byte("noise")}}}}
|
||||
settings.FinalMask = finalmask.NewFinalMask(nil, masks, nil, nil, rec.dialUDP, nil)
|
||||
|
||||
client, srv, cleanup := pairWith(t, settings)
|
||||
defer cleanup()
|
||||
|
||||
if _, err := client.Write([]byte("ping")); err != nil {
|
||||
t.Fatalf("client write: %v", err)
|
||||
}
|
||||
expectRead(t, srv, "ping")
|
||||
if _, err := srv.Write([]byte("pong")); err != nil {
|
||||
t.Fatalf("server write: %v", err)
|
||||
}
|
||||
expectRead(t, client, "pong")
|
||||
|
||||
payload := make([]byte, 300000)
|
||||
for i := range payload {
|
||||
payload[i] = byte(i % 251)
|
||||
}
|
||||
go client.Write(payload)
|
||||
if err := srv.SetReadDeadline(time.Now().Add(testPatience)); err != nil {
|
||||
t.Fatalf("SetReadDeadline: %v", err)
|
||||
}
|
||||
got := make([]byte, len(payload))
|
||||
if _, err := io.ReadFull(srv, got); err != nil {
|
||||
t.Fatalf("ReadFull: %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, payload) {
|
||||
t.Fatal("payload mismatch")
|
||||
}
|
||||
|
||||
rec.mu.Lock()
|
||||
defer rec.mu.Unlock()
|
||||
// one for the listener to www.googleapis.com, one for the token to
|
||||
// oauth2.googleapis.com and one of XMUX for each end of the session
|
||||
if len(rec.conns) != 4 {
|
||||
t.Errorf("%d connections went through the mask, want 4", len(rec.conns))
|
||||
}
|
||||
for i, conn := range rec.conns {
|
||||
conn.mu.Lock()
|
||||
sent := conn.sent
|
||||
conn.mu.Unlock()
|
||||
if len(sent) < 2 {
|
||||
t.Errorf("connection %d sent %d datagrams", i, len(sent))
|
||||
continue
|
||||
}
|
||||
// the noise, then a QUIC v1 Initial of 1250 bytes with no source
|
||||
// connection ID, as ChromeParrot sends it
|
||||
first, initial := sent[0], sent[1]
|
||||
if string(first) != "noise" || len(initial) != 1250 || initial[0]&0xf0 != 0xc0 || !bytes.Equal(initial[1:5], []byte{0, 0, 0, 1}) || initial[6+int(initial[5])] != 0 {
|
||||
t.Errorf("connection %d began with %q and %d bytes %x", i, first, len(initial), initial[:min(len(initial), 24)])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// over h3 a connection of XMUX is a QUIC connection, which closes once XMUX
|
||||
// drops it, so that the server hears of it
|
||||
func TestDriveXmuxOverH3(t *testing.T) {
|
||||
drive := newFakeDrive(t)
|
||||
server := newH3Server(t, drive.server.Config.Handler)
|
||||
driveTokenURL = "https://www.googleapis.com/token"
|
||||
driveFilesURL = "https://www.googleapis.com/files"
|
||||
driveUploadURL = "https://www.googleapis.com/upload"
|
||||
settings := server.settings(&tls.Config{})
|
||||
settings.ProtocolSettings = driveSettings().ProtocolSettings
|
||||
settings.ProtocolSettings.(*Config).Xmux = &splithttp.XmuxConfig{HMaxRequestTimes: &splithttp.RangeConfig{From: 3, To: 3}}
|
||||
storage, err := newDriveStorage(settings, settings.ProtocolSettings.(*Config))
|
||||
if err != nil {
|
||||
t.Fatalf("newDriveStorage: %v", err)
|
||||
}
|
||||
session := sessionStorage(storage)
|
||||
defer session.Close()
|
||||
|
||||
for _, name := range []string{"a", "b", "c", "d"} {
|
||||
if err := session.Put(context.Background(), "streams/abc/c2s/"+name, []byte("x")); err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
}
|
||||
// one for the token, the first of XMUX for 3 requests and one more
|
||||
if n := server.conns.Load(); n != 3 {
|
||||
t.Errorf("the token and 4 requests with 3 to a connection took %d connections", n)
|
||||
}
|
||||
waitFor(t, "the first connection of XMUX to close", func() bool { return server.closed.Load() == 1 })
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue