diff --git a/cmd/sing-box/cmd_tools_networkquality.go b/cmd/sing-box/cmd_tools_networkquality.go new file mode 100644 index 000000000..5f63571de --- /dev/null +++ b/cmd/sing-box/cmd_tools_networkquality.go @@ -0,0 +1,121 @@ +package main + +import ( + "fmt" + "os" + "strings" + "time" + + "github.com/sagernet/sing-box/common/networkquality" + "github.com/sagernet/sing-box/log" + + "github.com/spf13/cobra" +) + +var ( + commandNetworkQualityFlagConfigURL string + commandNetworkQualityFlagSerial bool + commandNetworkQualityFlagMaxRuntime int + commandNetworkQualityFlagHTTP3 bool +) + +var commandNetworkQuality = &cobra.Command{ + Use: "networkquality", + Short: "Run a network quality test", + Run: func(cmd *cobra.Command, args []string) { + err := runNetworkQuality() + if err != nil { + log.Fatal(err) + } + }, +} + +func init() { + commandNetworkQuality.Flags().StringVar( + &commandNetworkQualityFlagConfigURL, + "config-url", "", + "Network quality test config URL (default: Apple mensura)", + ) + commandNetworkQuality.Flags().BoolVar( + &commandNetworkQualityFlagSerial, + "serial", false, + "Run download and upload tests sequentially instead of in parallel", + ) + commandNetworkQuality.Flags().IntVar( + &commandNetworkQualityFlagMaxRuntime, + "max-runtime", int(networkquality.DefaultMaxRuntime/time.Second), + "Network quality maximum runtime in seconds", + ) + commandNetworkQuality.Flags().BoolVar( + &commandNetworkQualityFlagHTTP3, + "http3", false, + "Use HTTP/3 (QUIC) for measurement traffic", + ) + commandTools.AddCommand(commandNetworkQuality) +} + +func runNetworkQuality() error { + instance, err := createPreStartedClient() + if err != nil { + return err + } + defer instance.Close() + + dialer, err := createDialer(instance, commandToolsFlagOutbound) + if err != nil { + return err + } + + httpClient := networkquality.NewHTTPClient(dialer) + defer httpClient.CloseIdleConnections() + + measurementClientFactory, err := networkquality.NewOptionalHTTP3Factory(dialer, commandNetworkQualityFlagHTTP3) + if err != nil { + return err + } + + fmt.Fprintln(os.Stderr, "==== NETWORK QUALITY TEST ====") + + result, err := networkquality.Run(networkquality.Options{ + ConfigURL: commandNetworkQualityFlagConfigURL, + HTTPClient: httpClient, + NewMeasurementClient: measurementClientFactory, + Serial: commandNetworkQualityFlagSerial, + MaxRuntime: time.Duration(commandNetworkQualityFlagMaxRuntime) * time.Second, + Context: globalCtx, + OnProgress: func(p networkquality.Progress) { + if !commandNetworkQualityFlagSerial && p.Phase != networkquality.PhaseIdle { + fmt.Fprintf(os.Stderr, "\rDownload: %s RPM: %d Upload: %s RPM: %d", + networkquality.FormatBitrate(p.DownloadCapacity), p.DownloadRPM, + networkquality.FormatBitrate(p.UploadCapacity), p.UploadRPM) + return + } + switch networkquality.Phase(p.Phase) { + case networkquality.PhaseIdle: + if p.IdleLatencyMs > 0 { + fmt.Fprintf(os.Stderr, "\rIdle Latency: %d ms", p.IdleLatencyMs) + } else { + fmt.Fprint(os.Stderr, "\rMeasuring idle latency...") + } + case networkquality.PhaseDownload: + fmt.Fprintf(os.Stderr, "\rDownload: %s RPM: %d", + networkquality.FormatBitrate(p.DownloadCapacity), p.DownloadRPM) + case networkquality.PhaseUpload: + fmt.Fprintf(os.Stderr, "\rUpload: %s RPM: %d", + networkquality.FormatBitrate(p.UploadCapacity), p.UploadRPM) + } + }, + }) + if err != nil { + return err + } + + fmt.Fprintln(os.Stderr) + fmt.Fprintln(os.Stderr, strings.Repeat("-", 40)) + fmt.Fprintf(os.Stderr, "Idle Latency: %d ms\n", result.IdleLatencyMs) + fmt.Fprintf(os.Stderr, "Download Capacity: %-20s Accuracy: %s\n", networkquality.FormatBitrate(result.DownloadCapacity), result.DownloadCapacityAccuracy) + fmt.Fprintf(os.Stderr, "Upload Capacity: %-20s Accuracy: %s\n", networkquality.FormatBitrate(result.UploadCapacity), result.UploadCapacityAccuracy) + fmt.Fprintf(os.Stderr, "Download Responsiveness: %-20s Accuracy: %s\n", fmt.Sprintf("%d RPM", result.DownloadRPM), result.DownloadRPMAccuracy) + fmt.Fprintf(os.Stderr, "Upload Responsiveness: %-20s Accuracy: %s\n", fmt.Sprintf("%d RPM", result.UploadRPM), result.UploadRPMAccuracy) + return nil +} diff --git a/cmd/sing-box/cmd_tools_stun.go b/cmd/sing-box/cmd_tools_stun.go new file mode 100644 index 000000000..f13086caa --- /dev/null +++ b/cmd/sing-box/cmd_tools_stun.go @@ -0,0 +1,79 @@ +package main + +import ( + "fmt" + "os" + + "github.com/sagernet/sing-box/common/stun" + "github.com/sagernet/sing-box/log" + + "github.com/spf13/cobra" +) + +var commandSTUNFlagServer string + +var commandSTUN = &cobra.Command{ + Use: "stun", + Short: "Run a STUN test", + Args: cobra.NoArgs, + Run: func(cmd *cobra.Command, args []string) { + err := runSTUN() + if err != nil { + log.Fatal(err) + } + }, +} + +func init() { + commandSTUN.Flags().StringVarP(&commandSTUNFlagServer, "server", "s", stun.DefaultServer, "STUN server address") + commandTools.AddCommand(commandSTUN) +} + +func runSTUN() error { + instance, err := createPreStartedClient() + if err != nil { + return err + } + defer instance.Close() + + dialer, err := createDialer(instance, commandToolsFlagOutbound) + if err != nil { + return err + } + + fmt.Fprintln(os.Stderr, "==== STUN TEST ====") + + result, err := stun.Run(stun.Options{ + Server: commandSTUNFlagServer, + Dialer: dialer, + Context: globalCtx, + OnProgress: func(p stun.Progress) { + switch p.Phase { + case stun.PhaseBinding: + if p.ExternalAddr != "" { + fmt.Fprintf(os.Stderr, "\rExternal Address: %s (%d ms)", p.ExternalAddr, p.LatencyMs) + } else { + fmt.Fprint(os.Stderr, "\rSending binding request...") + } + case stun.PhaseNATMapping: + fmt.Fprint(os.Stderr, "\rDetecting NAT mapping behavior...") + case stun.PhaseNATFiltering: + fmt.Fprint(os.Stderr, "\rDetecting NAT filtering behavior...") + } + }, + }) + if err != nil { + return err + } + + fmt.Fprintln(os.Stderr) + fmt.Fprintf(os.Stderr, "External Address: %s\n", result.ExternalAddr) + fmt.Fprintf(os.Stderr, "Latency: %d ms\n", result.LatencyMs) + if result.NATTypeSupported { + fmt.Fprintf(os.Stderr, "NAT Mapping: %s\n", result.NATMapping) + fmt.Fprintf(os.Stderr, "NAT Filtering: %s\n", result.NATFiltering) + } else { + fmt.Fprintln(os.Stderr, "NAT Type Detection: not supported by server") + } + return nil +} diff --git a/common/networkquality/http.go b/common/networkquality/http.go new file mode 100644 index 000000000..f9ff2a4a5 --- /dev/null +++ b/common/networkquality/http.go @@ -0,0 +1,142 @@ +package networkquality + +import ( + "context" + "fmt" + "net" + "net/http" + "strings" + + C "github.com/sagernet/sing-box/constant" + sBufio "github.com/sagernet/sing/common/bufio" + E "github.com/sagernet/sing/common/exceptions" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" +) + +func FormatBitrate(bps int64) string { + switch { + case bps >= 1_000_000_000: + return fmt.Sprintf("%.1f Gbps", float64(bps)/1_000_000_000) + case bps >= 1_000_000: + return fmt.Sprintf("%.1f Mbps", float64(bps)/1_000_000) + case bps >= 1_000: + return fmt.Sprintf("%.1f Kbps", float64(bps)/1_000) + default: + return fmt.Sprintf("%d bps", bps) + } +} + +func NewHTTPClient(dialer N.Dialer) *http.Client { + transport := &http.Transport{ + ForceAttemptHTTP2: true, + TLSHandshakeTimeout: C.TCPTimeout, + } + if dialer != nil { + transport.DialContext = func(ctx context.Context, network string, addr string) (net.Conn, error) { + return dialer.DialContext(ctx, network, M.ParseSocksaddr(addr)) + } + } + return &http.Client{Transport: transport} +} + +func baseTransportFromClient(client *http.Client) (*http.Transport, error) { + if client == nil { + return nil, E.New("http client is nil") + } + if client.Transport == nil { + return http.DefaultTransport.(*http.Transport).Clone(), nil + } + transport, ok := client.Transport.(*http.Transport) + if !ok { + return nil, E.New("http client transport must be *http.Transport") + } + return transport.Clone(), nil +} + +func newMeasurementClient( + baseClient *http.Client, + connectEndpoint string, + singleConnection bool, + disableKeepAlives bool, + readCounters []N.CountFunc, + writeCounters []N.CountFunc, +) (*http.Client, error) { + transport, err := baseTransportFromClient(baseClient) + if err != nil { + return nil, err + } + transport.DisableCompression = true + transport.DisableKeepAlives = disableKeepAlives + if singleConnection { + transport.MaxConnsPerHost = 1 + transport.MaxIdleConnsPerHost = 1 + transport.MaxIdleConns = 1 + } + + baseDialContext := transport.DialContext + if baseDialContext == nil { + dialer := &net.Dialer{} + baseDialContext = dialer.DialContext + } + transport.DialContext = func(ctx context.Context, network string, addr string) (net.Conn, error) { + dialAddr := addr + if connectEndpoint != "" { + dialAddr = rewriteDialAddress(addr, connectEndpoint) + } + conn, dialErr := baseDialContext(ctx, network, dialAddr) + if dialErr != nil { + return nil, dialErr + } + if len(readCounters) > 0 || len(writeCounters) > 0 { + return sBufio.NewCounterConn(conn, readCounters, writeCounters), nil + } + return conn, nil + } + + return &http.Client{ + Transport: transport, + CheckRedirect: baseClient.CheckRedirect, + Jar: baseClient.Jar, + Timeout: baseClient.Timeout, + }, nil +} + +type MeasurementClientFactory func( + connectEndpoint string, + singleConnection bool, + disableKeepAlives bool, + readCounters []N.CountFunc, + writeCounters []N.CountFunc, +) (*http.Client, error) + +func defaultMeasurementClientFactory(baseClient *http.Client) MeasurementClientFactory { + return func(connectEndpoint string, singleConnection, disableKeepAlives bool, readCounters, writeCounters []N.CountFunc) (*http.Client, error) { + return newMeasurementClient(baseClient, connectEndpoint, singleConnection, disableKeepAlives, readCounters, writeCounters) + } +} + +func NewOptionalHTTP3Factory(dialer N.Dialer, useHTTP3 bool) (MeasurementClientFactory, error) { + if !useHTTP3 { + return nil, nil + } + return NewHTTP3MeasurementClientFactory(dialer) +} + +func rewriteDialAddress(addr string, connectEndpoint string) string { + connectEndpoint = strings.TrimSpace(connectEndpoint) + host, port, err := net.SplitHostPort(addr) + if err != nil { + return addr + } + endpointHost, endpointPort, err := net.SplitHostPort(connectEndpoint) + if err == nil { + host = endpointHost + if endpointPort != "" { + port = endpointPort + } + } else if connectEndpoint != "" { + host = connectEndpoint + } + return net.JoinHostPort(host, port) +} diff --git a/common/networkquality/http3.go b/common/networkquality/http3.go new file mode 100644 index 000000000..5e28d9fd6 --- /dev/null +++ b/common/networkquality/http3.go @@ -0,0 +1,55 @@ +//go:build with_quic + +package networkquality + +import ( + "context" + "crypto/tls" + "net" + "net/http" + + "github.com/sagernet/quic-go" + "github.com/sagernet/quic-go/http3" + sBufio "github.com/sagernet/sing/common/bufio" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" +) + +func NewHTTP3MeasurementClientFactory(dialer N.Dialer) (MeasurementClientFactory, error) { + // singleConnection and disableKeepAlives are not applied: + // HTTP/3 multiplexes streams over a single QUIC connection by default. + return func(connectEndpoint string, _, _ bool, readCounters, writeCounters []N.CountFunc) (*http.Client, error) { + transport := &http3.Transport{ + Dial: func(ctx context.Context, addr string, tlsCfg *tls.Config, cfg *quic.Config) (*quic.Conn, error) { + dialAddr := addr + if connectEndpoint != "" { + dialAddr = rewriteDialAddress(addr, connectEndpoint) + } + destination := M.ParseSocksaddr(dialAddr) + var udpConn net.Conn + var dialErr error + if dialer != nil { + udpConn, dialErr = dialer.DialContext(ctx, N.NetworkUDP, destination) + } else { + var netDialer net.Dialer + udpConn, dialErr = netDialer.DialContext(ctx, N.NetworkUDP, destination.String()) + } + if dialErr != nil { + return nil, dialErr + } + var wrappedConn net.Conn = udpConn + if len(readCounters) > 0 || len(writeCounters) > 0 { + wrappedConn = sBufio.NewCounterConn(udpConn, readCounters, writeCounters) + } + packetConn := sBufio.NewUnbindPacketConn(wrappedConn) + quicConn, dialErr := quic.DialEarly(ctx, packetConn, udpConn.RemoteAddr(), tlsCfg, cfg) + if dialErr != nil { + udpConn.Close() + return nil, dialErr + } + return quicConn, nil + }, + } + return &http.Client{Transport: transport}, nil + }, nil +} diff --git a/common/networkquality/http3_stub.go b/common/networkquality/http3_stub.go new file mode 100644 index 000000000..632837e68 --- /dev/null +++ b/common/networkquality/http3_stub.go @@ -0,0 +1,12 @@ +//go:build !with_quic + +package networkquality + +import ( + C "github.com/sagernet/sing-box/constant" + N "github.com/sagernet/sing/common/network" +) + +func NewHTTP3MeasurementClientFactory(dialer N.Dialer) (MeasurementClientFactory, error) { + return nil, C.ErrQUICNotIncluded +} diff --git a/common/networkquality/networkquality.go b/common/networkquality/networkquality.go new file mode 100644 index 000000000..a4c73472c --- /dev/null +++ b/common/networkquality/networkquality.go @@ -0,0 +1,1413 @@ +package networkquality + +import ( + "context" + "crypto/tls" + "encoding/json" + "io" + "math" + "math/rand" + "net/http" + "net/http/httptrace" + "net/url" + "sort" + "strings" + "sync" + "sync/atomic" + "time" + + sBufio "github.com/sagernet/sing/common/bufio" + E "github.com/sagernet/sing/common/exceptions" + N "github.com/sagernet/sing/common/network" +) + +const DefaultConfigURL = "https://mensura.cdn-apple.com/api/v1/gm/config" + +type Config struct { + Version int `json:"version"` + TestEndpoint string `json:"test_endpoint"` + URLs URLs `json:"urls"` +} + +type URLs struct { + SmallHTTPSDownloadURL string `json:"small_https_download_url"` + LargeHTTPSDownloadURL string `json:"large_https_download_url"` + HTTPSUploadURL string `json:"https_upload_url"` + SmallDownloadURL string `json:"small_download_url"` + LargeDownloadURL string `json:"large_download_url"` + UploadURL string `json:"upload_url"` +} + +func (u *URLs) smallDownloadURL() string { + if u.SmallHTTPSDownloadURL != "" { + return u.SmallHTTPSDownloadURL + } + return u.SmallDownloadURL +} + +func (u *URLs) largeDownloadURL() string { + if u.LargeHTTPSDownloadURL != "" { + return u.LargeHTTPSDownloadURL + } + return u.LargeDownloadURL +} + +func (u *URLs) uploadURL() string { + if u.HTTPSUploadURL != "" { + return u.HTTPSUploadURL + } + return u.UploadURL +} + +type Accuracy int32 + +const ( + AccuracyLow Accuracy = 0 + AccuracyMedium Accuracy = 1 + AccuracyHigh Accuracy = 2 +) + +func (a Accuracy) String() string { + switch a { + case AccuracyHigh: + return "High" + case AccuracyMedium: + return "Medium" + default: + return "Low" + } +} + +type Result struct { + DownloadCapacity int64 + UploadCapacity int64 + DownloadRPM int32 + UploadRPM int32 + IdleLatencyMs int32 + DownloadCapacityAccuracy Accuracy + UploadCapacityAccuracy Accuracy + DownloadRPMAccuracy Accuracy + UploadRPMAccuracy Accuracy +} + +type Progress struct { + Phase Phase + DownloadCapacity int64 + UploadCapacity int64 + DownloadRPM int32 + UploadRPM int32 + IdleLatencyMs int32 + ElapsedMs int64 + DownloadCapacityAccuracy Accuracy + UploadCapacityAccuracy Accuracy + DownloadRPMAccuracy Accuracy + UploadRPMAccuracy Accuracy +} + +type Phase int32 + +const ( + PhaseIdle Phase = 0 + PhaseDownload Phase = 1 + PhaseUpload Phase = 2 + PhaseDone Phase = 3 +) + +type Options struct { + ConfigURL string + HTTPClient *http.Client + NewMeasurementClient MeasurementClientFactory + Serial bool + MaxRuntime time.Duration + OnProgress func(Progress) + Context context.Context +} + +const DefaultMaxRuntime = 20 * time.Second + +type measurementSettings struct { + idleProbeCount int + testTimeout time.Duration + stabilityInterval time.Duration + sampleInterval time.Duration + progressInterval time.Duration + maxProbesPerSecond int + initialConnections int + maxConnections int + movingAvgDistance int + trimPercent int + stdDevTolerancePct float64 + maxProbeCapacityPct float64 +} + +var settings = measurementSettings{ + idleProbeCount: 5, + testTimeout: DefaultMaxRuntime, + stabilityInterval: time.Second, + sampleInterval: 250 * time.Millisecond, + progressInterval: 500 * time.Millisecond, + maxProbesPerSecond: 100, + initialConnections: 1, + maxConnections: 16, + movingAvgDistance: 4, + trimPercent: 5, + stdDevTolerancePct: 5, + maxProbeCapacityPct: 0.05, +} + +type resolvedConfig struct { + smallURL *url.URL + largeURL *url.URL + uploadURL *url.URL + connectEndpoint string +} + +type directionPlan struct { + dataURL *url.URL + probeURL *url.URL + connectEndpoint string + isUpload bool +} + +type probeTrace struct { + reused bool + connectStart time.Time + connectDone time.Time + tlsStart time.Time + tlsDone time.Time + tlsVersion uint16 + gotConn time.Time + wroteRequest time.Time + firstResponseByte time.Time +} + +type probeMeasurement struct { + total time.Duration + tcp time.Duration + tls time.Duration + httpFirst time.Duration + httpLoaded time.Duration + bytes int64 + reused bool +} + +type probeRound struct { + interval int + tcp time.Duration + tls time.Duration + httpFirst time.Duration + httpLoaded time.Duration +} + +func (p probeRound) responsivenessLatency() float64 { + var foreignSamples []float64 + if p.tcp > 0 { + foreignSamples = append(foreignSamples, durationMillis(p.tcp)) + } + if p.tls > 0 { + foreignSamples = append(foreignSamples, durationMillis(p.tls)) + } + if p.httpFirst > 0 { + foreignSamples = append(foreignSamples, durationMillis(p.httpFirst)) + } + if len(foreignSamples) == 0 || p.httpLoaded <= 0 { + return 0 + } + return (meanFloat64s(foreignSamples) + durationMillis(p.httpLoaded)) / 2 +} + +const maxConsecutiveErrors = 3 + +type loadConnection struct { + client *http.Client + dataURL *url.URL + isUpload bool + active atomic.Bool + ready atomic.Bool +} + +func (c *loadConnection) run(ctx context.Context, onError func(error)) { + defer c.client.CloseIdleConnections() + markActive := func() { + c.ready.Store(true) + c.active.Store(true) + } + var consecutiveErrors int + for { + select { + case <-ctx.Done(): + return + default: + } + var err error + if c.isUpload { + err = runUploadRequest(ctx, c.client, c.dataURL.String(), markActive) + } else { + err = runDownloadRequest(ctx, c.client, c.dataURL.String(), markActive) + } + c.active.Store(false) + if err != nil { + if ctx.Err() != nil { + return + } + consecutiveErrors++ + if consecutiveErrors > maxConsecutiveErrors { + onError(err) + return + } + c.client.CloseIdleConnections() + continue + } + consecutiveErrors = 0 + } +} + +type intervalThroughput struct { + interval int + bps float64 +} + +type intervalWindow struct { + lower int + upper int +} + +type stabilityTracker struct { + window int + stdDevTolerancePct float64 + instantaneous []float64 + movingAverages []float64 +} + +func (s *stabilityTracker) add(value float64) bool { + if value <= 0 || math.IsNaN(value) || math.IsInf(value, 0) { + return false + } + s.instantaneous = append(s.instantaneous, value) + if len(s.instantaneous) > s.window { + s.instantaneous = s.instantaneous[len(s.instantaneous)-s.window:] + } + s.movingAverages = append(s.movingAverages, meanFloat64s(s.instantaneous)) + if len(s.movingAverages) > s.window { + s.movingAverages = s.movingAverages[len(s.movingAverages)-s.window:] + } + return s.stable() +} + +func (s *stabilityTracker) ready() bool { + return len(s.movingAverages) >= s.window +} + +func (s *stabilityTracker) accuracy() Accuracy { + if s.stable() { + return AccuracyHigh + } + if s.ready() { + return AccuracyMedium + } + return AccuracyLow +} + +func (s *stabilityTracker) stable() bool { + if len(s.movingAverages) < s.window { + return false + } + currentAverage := s.movingAverages[len(s.movingAverages)-1] + if currentAverage <= 0 { + return false + } + return stdDevFloat64s(s.movingAverages) <= currentAverage*(s.stdDevTolerancePct/100) +} + +type directionMeasurement struct { + capacity int64 + rpm int32 + capacityAccuracy Accuracy + rpmAccuracy Accuracy +} + +type directionRunner struct { + factory MeasurementClientFactory + plan directionPlan + probeBytes int64 + + errCh chan error + errOnce sync.Once + wg sync.WaitGroup + + totalBytes atomic.Int64 + currentCapacity atomic.Int64 + currentRPM atomic.Int32 + currentInterval atomic.Int64 + + connMu sync.Mutex + connections []*loadConnection + + probeMu sync.Mutex + probeRounds []probeRound + intervalProbeValues []float64 + responsivenessWindow *intervalWindow + throughputs []intervalThroughput + throughputWindow *intervalWindow +} + +func newDirectionRunner(factory MeasurementClientFactory, plan directionPlan, probeBytes int64) *directionRunner { + return &directionRunner{ + factory: factory, + plan: plan, + probeBytes: probeBytes, + errCh: make(chan error, 1), + } +} + +func (r *directionRunner) fail(err error) { + if err == nil { + return + } + r.errOnce.Do(func() { + select { + case r.errCh <- err: + default: + } + }) +} + +func (r *directionRunner) onConnectionFailed(err error) { + r.connMu.Lock() + activeCount := 0 + for _, conn := range r.connections { + if conn.active.Load() { + activeCount++ + } + } + r.connMu.Unlock() + if activeCount == 0 { + r.fail(err) + } +} + +func (r *directionRunner) addConnection(ctx context.Context) error { + counter := N.CountFunc(func(n int64) { r.totalBytes.Add(n) }) + var readCounters, writeCounters []N.CountFunc + if r.plan.isUpload { + writeCounters = []N.CountFunc{counter} + } else { + readCounters = []N.CountFunc{counter} + } + client, err := r.factory(r.plan.connectEndpoint, true, false, readCounters, writeCounters) + if err != nil { + return err + } + conn := &loadConnection{ + client: client, + dataURL: r.plan.dataURL, + isUpload: r.plan.isUpload, + } + r.connMu.Lock() + r.connections = append(r.connections, conn) + r.connMu.Unlock() + r.wg.Add(1) + go func() { + defer r.wg.Done() + conn.run(ctx, r.onConnectionFailed) + }() + return nil +} + +func (r *directionRunner) connectionCount() int { + r.connMu.Lock() + defer r.connMu.Unlock() + return len(r.connections) +} + +func (r *directionRunner) pickReadyConnection() *loadConnection { + r.connMu.Lock() + defer r.connMu.Unlock() + var ready []*loadConnection + for _, conn := range r.connections { + if conn.ready.Load() && conn.active.Load() { + ready = append(ready, conn) + } + } + if len(ready) == 0 { + return nil + } + return ready[rand.Intn(len(ready))] +} + +func (r *directionRunner) startProber(ctx context.Context) { + r.wg.Add(1) + go func() { + defer r.wg.Done() + ticker := time.NewTicker(r.probeInterval()) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + conn := r.pickReadyConnection() + if conn == nil { + continue + } + go func(selfClient *http.Client) { + foreignClient, err := r.factory(r.plan.connectEndpoint, true, true, nil, nil) + if err != nil { + return + } + round, err := collectProbeRound(ctx, foreignClient, selfClient, r.plan.probeURL.String()) + foreignClient.CloseIdleConnections() + if err != nil { + return + } + r.recordProbeRound(probeRound{ + interval: int(r.currentInterval.Load()), + tcp: round.tcp, + tls: round.tls, + httpFirst: round.httpFirst, + httpLoaded: round.httpLoaded, + }) + }(conn.client) + ticker.Reset(r.probeInterval()) + } + }() +} + +func (r *directionRunner) probeInterval() time.Duration { + interval := time.Second / time.Duration(settings.maxProbesPerSecond) + capacity := r.currentCapacity.Load() + if capacity <= 0 || r.probeBytes <= 0 || settings.maxProbeCapacityPct <= 0 { + return interval + } + bitsPerRound := float64(r.probeBytes*2) * 8 + minSeconds := bitsPerRound / (float64(capacity) * settings.maxProbeCapacityPct) + if minSeconds <= 0 { + return interval + } + capacityInterval := time.Duration(minSeconds * float64(time.Second)) + if capacityInterval > interval { + interval = capacityInterval + } + return interval +} + +func (r *directionRunner) recordProbeRound(round probeRound) { + r.probeMu.Lock() + r.probeRounds = append(r.probeRounds, round) + if latency := round.responsivenessLatency(); latency > 0 { + r.intervalProbeValues = append(r.intervalProbeValues, latency) + } + r.currentRPM.Store(calculateRPM(r.probeRounds)) + r.probeMu.Unlock() +} + +func (r *directionRunner) swapIntervalProbeValues() []float64 { + r.probeMu.Lock() + defer r.probeMu.Unlock() + values := append([]float64(nil), r.intervalProbeValues...) + r.intervalProbeValues = nil + return values +} + +func (r *directionRunner) setResponsivenessWindow(currentInterval int) { + lower := currentInterval - settings.movingAvgDistance + 1 + if lower < 0 { + lower = 0 + } + r.probeMu.Lock() + r.responsivenessWindow = &intervalWindow{lower: lower, upper: currentInterval} + r.probeMu.Unlock() +} + +func (r *directionRunner) recordThroughput(interval int, bps float64) { + r.probeMu.Lock() + r.throughputs = append(r.throughputs, intervalThroughput{interval: interval, bps: bps}) + r.probeMu.Unlock() +} + +func (r *directionRunner) setThroughputWindow(currentInterval int) { + lower := currentInterval - settings.movingAvgDistance + 1 + if lower < 0 { + lower = 0 + } + r.probeMu.Lock() + r.throughputWindow = &intervalWindow{lower: lower, upper: currentInterval} + r.probeMu.Unlock() +} + +func (r *directionRunner) finalRPM() int32 { + r.probeMu.Lock() + defer r.probeMu.Unlock() + if r.responsivenessWindow == nil { + return calculateRPM(r.probeRounds) + } + var rounds []probeRound + for _, round := range r.probeRounds { + if round.interval >= r.responsivenessWindow.lower && round.interval <= r.responsivenessWindow.upper { + rounds = append(rounds, round) + } + } + if len(rounds) == 0 { + rounds = r.probeRounds + } + return calculateRPM(rounds) +} + +func (r *directionRunner) finalCapacity(totalDuration time.Duration) int64 { + r.probeMu.Lock() + defer r.probeMu.Unlock() + var samples []float64 + if r.throughputWindow != nil { + for _, sample := range r.throughputs { + if sample.interval >= r.throughputWindow.lower && sample.interval <= r.throughputWindow.upper { + samples = append(samples, sample.bps) + } + } + } + if len(samples) == 0 { + for _, sample := range r.throughputs { + samples = append(samples, sample.bps) + } + } + if len(samples) > 0 { + return int64(math.Round(upperTrimmedMean(samples, settings.trimPercent))) + } + if totalDuration > 0 { + return int64(float64(r.totalBytes.Load()) * 8 / totalDuration.Seconds()) + } + return 0 +} + +func (r *directionRunner) wait() { + r.wg.Wait() +} + +func Run(options Options) (*Result, error) { + ctx := options.Context + if ctx == nil { + ctx = context.Background() + } + if options.HTTPClient == nil { + return nil, E.New("http client is required") + } + maxRuntime, err := normalizeMaxRuntime(options.MaxRuntime) + if err != nil { + return nil, err + } + configURL := resolveConfigURL(options.ConfigURL) + config, err := fetchConfig(ctx, options.HTTPClient, configURL) + if err != nil { + return nil, E.Cause(err, "fetch config") + } + resolved, err := validateConfig(config) + if err != nil { + return nil, E.Cause(err, "validate config") + } + + start := time.Now() + report := func(progress Progress) { + if options.OnProgress == nil { + return + } + progress.ElapsedMs = time.Since(start).Milliseconds() + options.OnProgress(progress) + } + + factory := options.NewMeasurementClient + if factory == nil { + factory = defaultMeasurementClientFactory(options.HTTPClient) + } + + report(Progress{Phase: PhaseIdle}) + idleLatency, probeBytes, err := measureIdleLatency(ctx, factory, resolved) + if err != nil { + return nil, E.Cause(err, "measure idle latency") + } + report(Progress{Phase: PhaseIdle, IdleLatencyMs: idleLatency}) + + start = time.Now() + + var download, upload *directionMeasurement + if options.Serial { + download, upload, err = measureSerial( + ctx, + factory, + resolved, + idleLatency, + probeBytes, + maxRuntime, + report, + ) + } else { + download, upload, err = measureParallel( + ctx, + factory, + resolved, + idleLatency, + probeBytes, + maxRuntime, + report, + ) + } + if err != nil { + return nil, err + } + + result := &Result{ + DownloadCapacity: download.capacity, + UploadCapacity: upload.capacity, + DownloadRPM: download.rpm, + UploadRPM: upload.rpm, + IdleLatencyMs: idleLatency, + DownloadCapacityAccuracy: download.capacityAccuracy, + UploadCapacityAccuracy: upload.capacityAccuracy, + DownloadRPMAccuracy: download.rpmAccuracy, + UploadRPMAccuracy: upload.rpmAccuracy, + } + report(Progress{ + Phase: PhaseDone, + DownloadCapacity: result.DownloadCapacity, + UploadCapacity: result.UploadCapacity, + DownloadRPM: result.DownloadRPM, + UploadRPM: result.UploadRPM, + IdleLatencyMs: result.IdleLatencyMs, + DownloadCapacityAccuracy: result.DownloadCapacityAccuracy, + UploadCapacityAccuracy: result.UploadCapacityAccuracy, + DownloadRPMAccuracy: result.DownloadRPMAccuracy, + UploadRPMAccuracy: result.UploadRPMAccuracy, + }) + return result, nil +} + +func normalizeMaxRuntime(maxRuntime time.Duration) (time.Duration, error) { + if maxRuntime == 0 { + return settings.testTimeout, nil + } + if maxRuntime < 0 { + return 0, E.New("max runtime must be positive") + } + return maxRuntime, nil +} + +func measureSerial( + ctx context.Context, + factory MeasurementClientFactory, + resolved *resolvedConfig, + idleLatency int32, + probeBytes int64, + maxRuntime time.Duration, + report func(Progress), +) (*directionMeasurement, *directionMeasurement, error) { + downloadRuntime, uploadRuntime := splitRuntimeBudget(maxRuntime, 2) + report(Progress{Phase: PhaseDownload, IdleLatencyMs: idleLatency}) + download, err := measureDirection(ctx, factory, directionPlan{ + dataURL: resolved.largeURL, + probeURL: resolved.smallURL, + connectEndpoint: resolved.connectEndpoint, + }, probeBytes, downloadRuntime, func(capacity int64, rpm int32) { + report(Progress{ + Phase: PhaseDownload, + DownloadCapacity: capacity, + DownloadRPM: rpm, + IdleLatencyMs: idleLatency, + }) + }) + if err != nil { + return nil, nil, E.Cause(err, "measure download") + } + + report(Progress{ + Phase: PhaseUpload, + DownloadCapacity: download.capacity, + DownloadRPM: download.rpm, + IdleLatencyMs: idleLatency, + }) + upload, err := measureDirection(ctx, factory, directionPlan{ + dataURL: resolved.uploadURL, + probeURL: resolved.smallURL, + connectEndpoint: resolved.connectEndpoint, + isUpload: true, + }, probeBytes, uploadRuntime, func(capacity int64, rpm int32) { + report(Progress{ + Phase: PhaseUpload, + DownloadCapacity: download.capacity, + UploadCapacity: capacity, + DownloadRPM: download.rpm, + UploadRPM: rpm, + IdleLatencyMs: idleLatency, + }) + }) + if err != nil { + return nil, nil, E.Cause(err, "measure upload") + } + return download, upload, nil +} + +func measureParallel( + ctx context.Context, + factory MeasurementClientFactory, + resolved *resolvedConfig, + idleLatency int32, + probeBytes int64, + maxRuntime time.Duration, + report func(Progress), +) (*directionMeasurement, *directionMeasurement, error) { + type parallelResult struct { + measurement *directionMeasurement + err error + } + type progressState struct { + sync.Mutex + downloadCapacity int64 + uploadCapacity int64 + downloadRPM int32 + uploadRPM int32 + } + + parallelCtx, cancel := context.WithCancel(ctx) + defer cancel() + + report(Progress{Phase: PhaseDownload, IdleLatencyMs: idleLatency}) + report(Progress{Phase: PhaseUpload, IdleLatencyMs: idleLatency}) + + var state progressState + sendProgress := func(phase Phase, capacity int64, rpm int32) { + state.Lock() + if phase == PhaseDownload { + state.downloadCapacity = capacity + state.downloadRPM = rpm + } else { + state.uploadCapacity = capacity + state.uploadRPM = rpm + } + snapshot := Progress{ + Phase: phase, + DownloadCapacity: state.downloadCapacity, + UploadCapacity: state.uploadCapacity, + DownloadRPM: state.downloadRPM, + UploadRPM: state.uploadRPM, + IdleLatencyMs: idleLatency, + } + state.Unlock() + report(snapshot) + } + + var wg sync.WaitGroup + downloadCh := make(chan parallelResult, 1) + uploadCh := make(chan parallelResult, 1) + wg.Add(2) + go func() { + defer wg.Done() + measurement, err := measureDirection(parallelCtx, factory, directionPlan{ + dataURL: resolved.largeURL, + probeURL: resolved.smallURL, + connectEndpoint: resolved.connectEndpoint, + }, probeBytes, maxRuntime, func(capacity int64, rpm int32) { + sendProgress(PhaseDownload, capacity, rpm) + }) + if err != nil { + cancel() + downloadCh <- parallelResult{err: E.Cause(err, "measure download")} + return + } + downloadCh <- parallelResult{measurement: measurement} + }() + go func() { + defer wg.Done() + measurement, err := measureDirection(parallelCtx, factory, directionPlan{ + dataURL: resolved.uploadURL, + probeURL: resolved.smallURL, + connectEndpoint: resolved.connectEndpoint, + isUpload: true, + }, probeBytes, maxRuntime, func(capacity int64, rpm int32) { + sendProgress(PhaseUpload, capacity, rpm) + }) + if err != nil { + cancel() + uploadCh <- parallelResult{err: E.Cause(err, "measure upload")} + return + } + uploadCh <- parallelResult{measurement: measurement} + }() + + download := <-downloadCh + upload := <-uploadCh + wg.Wait() + if download.err != nil { + return nil, nil, download.err + } + if upload.err != nil { + return nil, nil, upload.err + } + return download.measurement, upload.measurement, nil +} + +func resolveConfigURL(rawURL string) string { + rawURL = strings.TrimSpace(rawURL) + if rawURL == "" { + return DefaultConfigURL + } + if !strings.Contains(rawURL, "://") && !strings.Contains(rawURL, "/") { + return "https://" + rawURL + "/.well-known/nq" + } + parsedURL, err := url.Parse(rawURL) + if err != nil { + return rawURL + } + if parsedURL.Scheme != "" && parsedURL.Host != "" && (parsedURL.Path == "" || parsedURL.Path == "/") { + parsedURL.Path = "/.well-known/nq" + return parsedURL.String() + } + return rawURL +} + +func fetchConfig(ctx context.Context, client *http.Client, configURL string) (*Config, error) { + req, err := newRequest(ctx, http.MethodGet, configURL, nil) + if err != nil { + return nil, err + } + resp, err := client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if err = validateResponse(resp); err != nil { + return nil, err + } + var config Config + if err = json.NewDecoder(resp.Body).Decode(&config); err != nil { + return nil, E.Cause(err, "decode config") + } + return &config, nil +} + +func validateConfig(config *Config) (*resolvedConfig, error) { + if config == nil { + return nil, E.New("config is nil") + } + if config.Version != 1 { + return nil, E.New("unsupported config version: ", config.Version) + } + parseURL := func(name string, rawURL string) (*url.URL, error) { + if rawURL == "" { + return nil, E.New("config missing required URL: ", name) + } + parsedURL, err := url.Parse(rawURL) + if err != nil { + return nil, E.Cause(err, "parse "+name) + } + if parsedURL.Scheme != "http" && parsedURL.Scheme != "https" { + return nil, E.New("unsupported URL scheme in ", name, ": ", parsedURL.Scheme) + } + if parsedURL.Host == "" { + return nil, E.New("config missing host in ", name) + } + return parsedURL, nil + } + + smallURL, err := parseURL("small_download_url", config.URLs.smallDownloadURL()) + if err != nil { + return nil, err + } + largeURL, err := parseURL("large_download_url", config.URLs.largeDownloadURL()) + if err != nil { + return nil, err + } + uploadURL, err := parseURL("upload_url", config.URLs.uploadURL()) + if err != nil { + return nil, err + } + + if smallURL.Host != largeURL.Host || smallURL.Host != uploadURL.Host { + return nil, E.New("config URLs must use the same host") + } + + return &resolvedConfig{ + smallURL: smallURL, + largeURL: largeURL, + uploadURL: uploadURL, + connectEndpoint: strings.TrimSpace(config.TestEndpoint), + }, nil +} + +func measureIdleLatency(ctx context.Context, factory MeasurementClientFactory, config *resolvedConfig) (int32, int64, error) { + var latencies []int64 + var maxProbeBytes int64 + for i := 0; i < settings.idleProbeCount; i++ { + select { + case <-ctx.Done(): + return 0, 0, ctx.Err() + default: + } + client, err := factory(config.connectEndpoint, true, true, nil, nil) + if err != nil { + return 0, 0, err + } + measurement, err := runProbe(ctx, client, config.smallURL.String(), false) + client.CloseIdleConnections() + if err != nil { + return 0, 0, err + } + latencies = append(latencies, measurement.total.Milliseconds()) + if measurement.bytes > maxProbeBytes { + maxProbeBytes = measurement.bytes + } + } + sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] }) + return int32(latencies[len(latencies)/2]), maxProbeBytes, nil +} + +func measureDirection( + ctx context.Context, + factory MeasurementClientFactory, + plan directionPlan, + probeBytes int64, + maxRuntime time.Duration, + onProgress func(capacity int64, rpm int32), +) (*directionMeasurement, error) { + phaseCtx, phaseCancel := context.WithTimeout(ctx, maxRuntime) + defer phaseCancel() + + runner := newDirectionRunner(factory, plan, probeBytes) + defer runner.wait() + + for i := 0; i < settings.initialConnections; i++ { + err := runner.addConnection(phaseCtx) + if err != nil { + return nil, err + } + } + + runner.startProber(phaseCtx) + + throughputTracker := stabilityTracker{ + window: settings.movingAvgDistance, + stdDevTolerancePct: settings.stdDevTolerancePct, + } + responsivenessTracker := stabilityTracker{ + window: settings.movingAvgDistance, + stdDevTolerancePct: settings.stdDevTolerancePct, + } + + start := time.Now() + sampleTicker := time.NewTicker(settings.sampleInterval) + defer sampleTicker.Stop() + intervalTicker := time.NewTicker(settings.stabilityInterval) + defer intervalTicker.Stop() + progressTicker := time.NewTicker(settings.progressInterval) + defer progressTicker.Stop() + + prevSampleBytes := int64(0) + prevSampleTime := start + prevIntervalBytes := int64(0) + prevIntervalTime := start + var ewmaCapacity float64 + var goodputSaturated bool + var intervalIndex int + + for { + select { + case err := <-runner.errCh: + return nil, err + case now := <-sampleTicker.C: + currentBytes := runner.totalBytes.Load() + elapsed := now.Sub(prevSampleTime).Seconds() + if elapsed > 0 { + instantaneousBps := float64(currentBytes-prevSampleBytes) * 8 / elapsed + if ewmaCapacity == 0 { + ewmaCapacity = instantaneousBps + } else { + ewmaCapacity = 0.3*instantaneousBps + 0.7*ewmaCapacity + } + runner.currentCapacity.Store(int64(ewmaCapacity)) + } + prevSampleBytes = currentBytes + prevSampleTime = now + case <-intervalTicker.C: + now := time.Now() + currentBytes := runner.totalBytes.Load() + elapsed := now.Sub(prevIntervalTime).Seconds() + if elapsed > 0 { + intervalBps := float64(currentBytes-prevIntervalBytes) * 8 / elapsed + runner.recordThroughput(intervalIndex, intervalBps) + throughputStable := throughputTracker.add(intervalBps) + if throughputStable && runner.throughputWindow == nil { + runner.setThroughputWindow(intervalIndex) + } + if !goodputSaturated && (throughputStable || (runner.connectionCount() >= settings.maxConnections && throughputTracker.ready())) { + goodputSaturated = true + } + if runner.connectionCount() < settings.maxConnections { + err := runner.addConnection(phaseCtx) + if err != nil { + return nil, err + } + } + } + if goodputSaturated { + if values := runner.swapIntervalProbeValues(); len(values) > 0 { + if responsivenessTracker.add(upperTrimmedMean(values, settings.trimPercent)) && runner.responsivenessWindow == nil { + runner.setResponsivenessWindow(intervalIndex) + phaseCancel() + } + } + } + prevIntervalBytes = currentBytes + prevIntervalTime = now + intervalIndex++ + runner.currentInterval.Store(int64(intervalIndex)) + case <-progressTicker.C: + if onProgress != nil { + onProgress(int64(ewmaCapacity), runner.currentRPM.Load()) + } + case <-phaseCtx.Done(): + if ctx.Err() != nil { + return nil, ctx.Err() + } + totalDuration := time.Since(start) + return &directionMeasurement{ + capacity: runner.finalCapacity(totalDuration), + rpm: runner.finalRPM(), + capacityAccuracy: throughputTracker.accuracy(), + rpmAccuracy: responsivenessTracker.accuracy(), + }, nil + } + } +} + +func splitRuntimeBudget(total time.Duration, directions int) (time.Duration, time.Duration) { + if directions <= 1 { + return total, total + } + first := total / time.Duration(directions) + second := total - first + return first, second +} + +func collectProbeRound(ctx context.Context, foreignClient *http.Client, selfClient *http.Client, rawURL string) (probeMeasurement, error) { + var foreignResult probeMeasurement + var selfResult probeMeasurement + var foreignErr error + var selfErr error + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + foreignResult, foreignErr = runProbe(ctx, foreignClient, rawURL, false) + }() + go func() { + defer wg.Done() + selfResult, selfErr = runProbe(ctx, selfClient, rawURL, true) + }() + wg.Wait() + + if foreignErr != nil { + return probeMeasurement{}, E.Cause(foreignErr, "foreign probe") + } + if selfErr != nil { + return probeMeasurement{}, E.Cause(selfErr, "self probe") + } + return probeMeasurement{ + tcp: foreignResult.tcp, + tls: foreignResult.tls, + httpFirst: foreignResult.httpFirst, + httpLoaded: selfResult.httpLoaded, + }, nil +} + +func runProbe(ctx context.Context, client *http.Client, rawURL string, expectReuse bool) (probeMeasurement, error) { + var trace probeTrace + start := time.Now() + req, err := newRequest(httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{ + ConnectStart: func(string, string) { + if trace.connectStart.IsZero() { + trace.connectStart = time.Now() + } + }, + ConnectDone: func(string, string, error) { + if trace.connectDone.IsZero() { + trace.connectDone = time.Now() + } + }, + TLSHandshakeStart: func() { + if trace.tlsStart.IsZero() { + trace.tlsStart = time.Now() + } + }, + TLSHandshakeDone: func(state tls.ConnectionState, _ error) { + if trace.tlsDone.IsZero() { + trace.tlsDone = time.Now() + trace.tlsVersion = state.Version + } + }, + GotConn: func(info httptrace.GotConnInfo) { + trace.reused = info.Reused + trace.gotConn = time.Now() + }, + WroteRequest: func(httptrace.WroteRequestInfo) { + trace.wroteRequest = time.Now() + }, + GotFirstResponseByte: func() { + trace.firstResponseByte = time.Now() + }, + }), http.MethodGet, rawURL, nil) + if err != nil { + return probeMeasurement{}, err + } + if !expectReuse { + req.Close = true + } + resp, err := client.Do(req) + if err != nil { + return probeMeasurement{}, err + } + defer resp.Body.Close() + if err = validateResponse(resp); err != nil { + return probeMeasurement{}, err + } + n, err := io.Copy(io.Discard, resp.Body) + end := time.Now() + if err != nil { + return probeMeasurement{}, err + } + if expectReuse && !trace.reused { + return probeMeasurement{}, E.New("self probe did not reuse an existing connection") + } + + httpStart := trace.wroteRequest + if httpStart.IsZero() { + switch { + case !trace.tlsDone.IsZero(): + httpStart = trace.tlsDone + case !trace.connectDone.IsZero(): + httpStart = trace.connectDone + case !trace.gotConn.IsZero(): + httpStart = trace.gotConn + default: + httpStart = start + } + } + + measurement := probeMeasurement{ + total: end.Sub(start), + bytes: n, + reused: trace.reused, + } + if !trace.connectStart.IsZero() && !trace.connectDone.IsZero() && trace.connectDone.After(trace.connectStart) { + measurement.tcp = trace.connectDone.Sub(trace.connectStart) + } + if !trace.tlsStart.IsZero() && !trace.tlsDone.IsZero() && trace.tlsDone.After(trace.tlsStart) { + measurement.tls = trace.tlsDone.Sub(trace.tlsStart) + if roundTrips := tlsHandshakeRoundTrips(trace.tlsVersion); roundTrips > 1 { + measurement.tls /= time.Duration(roundTrips) + } + } + if !trace.firstResponseByte.IsZero() && trace.firstResponseByte.After(httpStart) { + measurement.httpFirst = trace.firstResponseByte.Sub(httpStart) + } + if end.After(httpStart) { + measurement.httpLoaded = end.Sub(httpStart) + } + return measurement, nil +} + +func runDownloadRequest(ctx context.Context, client *http.Client, rawURL string, onActive func()) error { + req, err := newRequest(ctx, http.MethodGet, rawURL, nil) + if err != nil { + return err + } + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + err = validateResponse(resp) + if err != nil { + return err + } + if onActive != nil { + onActive() + } + _, err = sBufio.Copy(io.Discard, resp.Body) + if ctx.Err() != nil { + return nil + } + return err +} + +func runUploadRequest(ctx context.Context, client *http.Client, rawURL string, onActive func()) error { + body := &uploadBody{ + ctx: ctx, + onActive: onActive, + } + req, err := newRequest(ctx, http.MethodPost, rawURL, body) + if err != nil { + return err + } + req.ContentLength = -1 + req.Header.Set("Content-Type", "application/octet-stream") + resp, err := client.Do(req) + if err != nil { + if ctx.Err() != nil { + return nil + } + return err + } + defer resp.Body.Close() + err = validateResponse(resp) + if err != nil { + return err + } + _, _ = io.Copy(io.Discard, resp.Body) + <-ctx.Done() + return nil +} + +func newRequest(ctx context.Context, method string, rawURL string, body io.Reader) (*http.Request, error) { + req, err := http.NewRequestWithContext(ctx, method, rawURL, body) + if err != nil { + return nil, err + } + req.Header.Set("Accept-Encoding", "identity") + return req, nil +} + +func validateResponse(resp *http.Response) error { + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return E.New("unexpected status: ", resp.Status) + } + if encoding := resp.Header.Get("Content-Encoding"); encoding != "" { + return E.New("unexpected content encoding: ", encoding) + } + return nil +} + +func calculateRPM(rounds []probeRound) int32 { + if len(rounds) == 0 { + return 0 + } + var tcpSamples []float64 + var tlsSamples []float64 + var httpFirstSamples []float64 + var httpLoadedSamples []float64 + for _, round := range rounds { + if round.tcp > 0 { + tcpSamples = append(tcpSamples, durationMillis(round.tcp)) + } + if round.tls > 0 { + tlsSamples = append(tlsSamples, durationMillis(round.tls)) + } + if round.httpFirst > 0 { + httpFirstSamples = append(httpFirstSamples, durationMillis(round.httpFirst)) + } + if round.httpLoaded > 0 { + httpLoadedSamples = append(httpLoadedSamples, durationMillis(round.httpLoaded)) + } + } + httpLoaded := upperTrimmedMean(httpLoadedSamples, settings.trimPercent) + if httpLoaded <= 0 { + return 0 + } + var foreignComponents []float64 + if tcp := upperTrimmedMean(tcpSamples, settings.trimPercent); tcp > 0 { + foreignComponents = append(foreignComponents, tcp) + } + if tls := upperTrimmedMean(tlsSamples, settings.trimPercent); tls > 0 { + foreignComponents = append(foreignComponents, tls) + } + if httpFirst := upperTrimmedMean(httpFirstSamples, settings.trimPercent); httpFirst > 0 { + foreignComponents = append(foreignComponents, httpFirst) + } + if len(foreignComponents) == 0 { + return 0 + } + foreignLatency := meanFloat64s(foreignComponents) + foreignRPM := 60000.0 / foreignLatency + loadedRPM := 60000.0 / httpLoaded + return int32(math.Round((foreignRPM + loadedRPM) / 2)) +} + +func tlsHandshakeRoundTrips(version uint16) int { + switch version { + case tls.VersionTLS12, tls.VersionTLS11, tls.VersionTLS10: + return 2 + default: + return 1 + } +} + +func durationMillis(value time.Duration) float64 { + return float64(value) / float64(time.Millisecond) +} + +func upperTrimmedMean(values []float64, trimPercent int) float64 { + trimmed := upperTrimFloat64s(values, trimPercent) + if len(trimmed) == 0 { + return 0 + } + return meanFloat64s(trimmed) +} + +func upperTrimFloat64s(values []float64, trimPercent int) []float64 { + if len(values) == 0 { + return nil + } + trimmed := append([]float64(nil), values...) + sort.Float64s(trimmed) + if trimPercent <= 0 { + return trimmed + } + trimCount := int(math.Floor(float64(len(trimmed)) * float64(trimPercent) / 100)) + if trimCount <= 0 || trimCount >= len(trimmed) { + return trimmed + } + return trimmed[:len(trimmed)-trimCount] +} + +func meanFloat64s(values []float64) float64 { + if len(values) == 0 { + return 0 + } + var total float64 + for _, value := range values { + total += value + } + return total / float64(len(values)) +} + +func stdDevFloat64s(values []float64) float64 { + if len(values) == 0 { + return 0 + } + mean := meanFloat64s(values) + var total float64 + for _, value := range values { + delta := value - mean + total += delta * delta + } + return math.Sqrt(total / float64(len(values))) +} + +type uploadBody struct { + ctx context.Context + activated atomic.Bool + onActive func() +} + +func (u *uploadBody) Read(p []byte) (int, error) { + if err := u.ctx.Err(); err != nil { + return 0, err + } + clear(p) + n := len(p) + if n > 0 && u.onActive != nil && u.activated.CompareAndSwap(false, true) { + u.onActive() + } + return n, nil +} + +func (u *uploadBody) Close() error { + return nil +} diff --git a/common/stun/stun.go b/common/stun/stun.go new file mode 100644 index 000000000..b4c2313f0 --- /dev/null +++ b/common/stun/stun.go @@ -0,0 +1,607 @@ +package stun + +import ( + "context" + "crypto/rand" + "encoding/binary" + "fmt" + "net" + "net/netip" + "time" + + E "github.com/sagernet/sing/common/exceptions" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" +) + +const ( + DefaultServer = "stun.voipgate.com:3478" + + magicCookie = 0x2112A442 + headerSize = 20 + + bindingRequest = 0x0001 + bindingSuccessResponse = 0x0101 + bindingErrorResponse = 0x0111 + + attrMappedAddress = 0x0001 + attrChangeRequest = 0x0003 + attrErrorCode = 0x0009 + attrXORMappedAddress = 0x0020 + attrOtherAddress = 0x802c + + familyIPv4 = 0x01 + familyIPv6 = 0x02 + + changeIP = 0x04 + changePort = 0x02 + + defaultRTO = 500 * time.Millisecond + minRTO = 250 * time.Millisecond + maxRetransmit = 2 +) + +type Phase int32 + +const ( + PhaseBinding Phase = iota + PhaseNATMapping + PhaseNATFiltering + PhaseDone +) + +type NATMapping int32 + +const ( + NATMappingUnknown NATMapping = iota + _ // reserved + NATMappingEndpointIndependent + NATMappingAddressDependent + NATMappingAddressAndPortDependent +) + +func (m NATMapping) String() string { + switch m { + case NATMappingEndpointIndependent: + return "Endpoint Independent" + case NATMappingAddressDependent: + return "Address Dependent" + case NATMappingAddressAndPortDependent: + return "Address and Port Dependent" + default: + return "Unknown" + } +} + +type NATFiltering int32 + +const ( + NATFilteringUnknown NATFiltering = iota + NATFilteringEndpointIndependent + NATFilteringAddressDependent + NATFilteringAddressAndPortDependent +) + +func (f NATFiltering) String() string { + switch f { + case NATFilteringEndpointIndependent: + return "Endpoint Independent" + case NATFilteringAddressDependent: + return "Address Dependent" + case NATFilteringAddressAndPortDependent: + return "Address and Port Dependent" + default: + return "Unknown" + } +} + +type TransactionID [12]byte + +type Options struct { + Server string + Dialer N.Dialer + Context context.Context + OnProgress func(Progress) +} + +type Progress struct { + Phase Phase + ExternalAddr string + LatencyMs int32 + NATMapping NATMapping + NATFiltering NATFiltering +} + +type Result struct { + ExternalAddr string + LatencyMs int32 + NATMapping NATMapping + NATFiltering NATFiltering + NATTypeSupported bool +} + +type parsedResponse struct { + xorMappedAddr netip.AddrPort + mappedAddr netip.AddrPort + otherAddr netip.AddrPort +} + +func (r *parsedResponse) externalAddr() (netip.AddrPort, bool) { + if r.xorMappedAddr.IsValid() { + return r.xorMappedAddr, true + } + if r.mappedAddr.IsValid() { + return r.mappedAddr, true + } + return netip.AddrPort{}, false +} + +type stunAttribute struct { + typ uint16 + value []byte +} + +func newTransactionID() TransactionID { + var id TransactionID + _, _ = rand.Read(id[:]) + return id +} + +func buildBindingRequest(txID TransactionID, attrs ...stunAttribute) []byte { + attrLen := 0 + for _, attr := range attrs { + attrLen += 4 + len(attr.value) + paddingLen(len(attr.value)) + } + + buf := make([]byte, headerSize+attrLen) + binary.BigEndian.PutUint16(buf[0:2], bindingRequest) + binary.BigEndian.PutUint16(buf[2:4], uint16(attrLen)) + binary.BigEndian.PutUint32(buf[4:8], magicCookie) + copy(buf[8:20], txID[:]) + + offset := headerSize + for _, attr := range attrs { + binary.BigEndian.PutUint16(buf[offset:offset+2], attr.typ) + binary.BigEndian.PutUint16(buf[offset+2:offset+4], uint16(len(attr.value))) + copy(buf[offset+4:offset+4+len(attr.value)], attr.value) + offset += 4 + len(attr.value) + paddingLen(len(attr.value)) + } + + return buf +} + +func changeRequestAttr(flags byte) stunAttribute { + return stunAttribute{ + typ: attrChangeRequest, + value: []byte{0, 0, 0, flags}, + } +} + +func parseResponse(data []byte, expectedTxID TransactionID) (*parsedResponse, error) { + if len(data) < headerSize { + return nil, E.New("response too short") + } + + msgType := binary.BigEndian.Uint16(data[0:2]) + if msgType&0xC000 != 0 { + return nil, E.New("invalid STUN message: top 2 bits not zero") + } + + cookie := binary.BigEndian.Uint32(data[4:8]) + if cookie != magicCookie { + return nil, E.New("invalid magic cookie") + } + + var txID TransactionID + copy(txID[:], data[8:20]) + if txID != expectedTxID { + return nil, E.New("transaction ID mismatch") + } + + msgLen := int(binary.BigEndian.Uint16(data[2:4])) + if msgLen > len(data)-headerSize { + return nil, E.New("message length exceeds data") + } + + attrData := data[headerSize : headerSize+msgLen] + + if msgType == bindingErrorResponse { + return nil, parseErrorResponse(attrData) + } + if msgType != bindingSuccessResponse { + return nil, E.New("unexpected message type: ", fmt.Sprintf("0x%04x", msgType)) + } + + resp := &parsedResponse{} + offset := 0 + for offset+4 <= len(attrData) { + attrType := binary.BigEndian.Uint16(attrData[offset : offset+2]) + attrLen := int(binary.BigEndian.Uint16(attrData[offset+2 : offset+4])) + if offset+4+attrLen > len(attrData) { + break + } + attrValue := attrData[offset+4 : offset+4+attrLen] + + switch attrType { + case attrXORMappedAddress: + addr, err := parseXORMappedAddress(attrValue, txID) + if err == nil { + resp.xorMappedAddr = addr + } + case attrMappedAddress: + addr, err := parseMappedAddress(attrValue) + if err == nil { + resp.mappedAddr = addr + } + case attrOtherAddress: + addr, err := parseMappedAddress(attrValue) + if err == nil { + resp.otherAddr = addr + } + } + + offset += 4 + attrLen + paddingLen(attrLen) + } + + return resp, nil +} + +func parseErrorResponse(data []byte) error { + offset := 0 + for offset+4 <= len(data) { + attrType := binary.BigEndian.Uint16(data[offset : offset+2]) + attrLen := int(binary.BigEndian.Uint16(data[offset+2 : offset+4])) + if offset+4+attrLen > len(data) { + break + } + if attrType == attrErrorCode && attrLen >= 4 { + attrValue := data[offset+4 : offset+4+attrLen] + class := int(attrValue[2] & 0x07) + number := int(attrValue[3]) + code := class*100 + number + if attrLen > 4 { + return E.New("STUN error ", code, ": ", string(attrValue[4:])) + } + return E.New("STUN error ", code) + } + offset += 4 + attrLen + paddingLen(attrLen) + } + return E.New("STUN error response") +} + +func parseXORMappedAddress(data []byte, txID TransactionID) (netip.AddrPort, error) { + if len(data) < 4 { + return netip.AddrPort{}, E.New("XOR-MAPPED-ADDRESS too short") + } + + family := data[1] + xPort := binary.BigEndian.Uint16(data[2:4]) + port := xPort ^ uint16(magicCookie>>16) + + switch family { + case familyIPv4: + if len(data) < 8 { + return netip.AddrPort{}, E.New("XOR-MAPPED-ADDRESS IPv4 too short") + } + var ip [4]byte + binary.BigEndian.PutUint32(ip[:], binary.BigEndian.Uint32(data[4:8])^magicCookie) + return netip.AddrPortFrom(netip.AddrFrom4(ip), port), nil + case familyIPv6: + if len(data) < 20 { + return netip.AddrPort{}, E.New("XOR-MAPPED-ADDRESS IPv6 too short") + } + var ip [16]byte + var xorKey [16]byte + binary.BigEndian.PutUint32(xorKey[0:4], magicCookie) + copy(xorKey[4:16], txID[:]) + for i := range 16 { + ip[i] = data[4+i] ^ xorKey[i] + } + return netip.AddrPortFrom(netip.AddrFrom16(ip), port), nil + default: + return netip.AddrPort{}, E.New("unknown address family: ", family) + } +} + +func parseMappedAddress(data []byte) (netip.AddrPort, error) { + if len(data) < 4 { + return netip.AddrPort{}, E.New("MAPPED-ADDRESS too short") + } + + family := data[1] + port := binary.BigEndian.Uint16(data[2:4]) + + switch family { + case familyIPv4: + if len(data) < 8 { + return netip.AddrPort{}, E.New("MAPPED-ADDRESS IPv4 too short") + } + return netip.AddrPortFrom( + netip.AddrFrom4([4]byte{data[4], data[5], data[6], data[7]}), port, + ), nil + case familyIPv6: + if len(data) < 20 { + return netip.AddrPort{}, E.New("MAPPED-ADDRESS IPv6 too short") + } + var ip [16]byte + copy(ip[:], data[4:20]) + return netip.AddrPortFrom(netip.AddrFrom16(ip), port), nil + default: + return netip.AddrPort{}, E.New("unknown address family: ", family) + } +} + +func roundTrip(conn net.PacketConn, addr net.Addr, txID TransactionID, attrs []stunAttribute, rto time.Duration) (*parsedResponse, time.Duration, error) { + request := buildBindingRequest(txID, attrs...) + currentRTO := rto + retransmitCount := 0 + + sendTime := time.Now() + _, err := conn.WriteTo(request, addr) + if err != nil { + return nil, 0, E.Cause(err, "send STUN request") + } + + buf := make([]byte, 1024) + for { + err = conn.SetReadDeadline(sendTime.Add(currentRTO)) + if err != nil { + return nil, 0, E.Cause(err, "set read deadline") + } + + n, _, readErr := conn.ReadFrom(buf) + if readErr != nil { + if E.IsTimeout(readErr) && retransmitCount < maxRetransmit { + retransmitCount++ + currentRTO *= 2 + sendTime = time.Now() + _, err = conn.WriteTo(request, addr) + if err != nil { + return nil, 0, E.Cause(err, "retransmit STUN request") + } + continue + } + return nil, 0, E.Cause(readErr, "read STUN response") + } + + if n < headerSize || buf[0]&0xC0 != 0 || + binary.BigEndian.Uint32(buf[4:8]) != magicCookie { + continue + } + var receivedTxID TransactionID + copy(receivedTxID[:], buf[8:20]) + if receivedTxID != txID { + continue + } + + latency := time.Since(sendTime) + + resp, parseErr := parseResponse(buf[:n], txID) + if parseErr != nil { + return nil, 0, parseErr + } + + return resp, latency, nil + } +} + +func Run(options Options) (*Result, error) { + ctx := options.Context + if ctx == nil { + ctx = context.Background() + } + + server := options.Server + if server == "" { + server = DefaultServer + } + serverSocksaddr := M.ParseSocksaddr(server) + if serverSocksaddr.Port == 0 { + serverSocksaddr.Port = 3478 + } + + reportProgress := options.OnProgress + if reportProgress == nil { + reportProgress = func(Progress) {} + } + + var ( + packetConn net.PacketConn + serverAddr net.Addr + err error + ) + + if options.Dialer != nil { + packetConn, err = options.Dialer.ListenPacket(ctx, serverSocksaddr) + if err != nil { + return nil, E.Cause(err, "create UDP socket") + } + serverAddr = serverSocksaddr + } else { + serverUDPAddr, resolveErr := net.ResolveUDPAddr("udp", serverSocksaddr.String()) + if resolveErr != nil { + return nil, E.Cause(resolveErr, "resolve STUN server") + } + packetConn, err = net.ListenPacket("udp", "") + if err != nil { + return nil, E.Cause(err, "create UDP socket") + } + serverAddr = serverUDPAddr + } + defer func() { + _ = packetConn.Close() + }() + + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + } + + rto := defaultRTO + + // Phase 1: Binding + reportProgress(Progress{Phase: PhaseBinding}) + + txID := newTransactionID() + resp, latency, err := roundTrip(packetConn, serverAddr, txID, nil, rto) + if err != nil { + return nil, E.Cause(err, "binding request") + } + + rto = max(minRTO, 3*latency) + + externalAddr, ok := resp.externalAddr() + if !ok { + return nil, E.New("no mapped address in response") + } + + result := &Result{ + ExternalAddr: externalAddr.String(), + LatencyMs: int32(latency.Milliseconds()), + } + + reportProgress(Progress{ + Phase: PhaseBinding, + ExternalAddr: result.ExternalAddr, + LatencyMs: result.LatencyMs, + }) + + otherAddr := resp.otherAddr + if !otherAddr.IsValid() { + result.NATTypeSupported = false + reportProgress(Progress{ + Phase: PhaseDone, + ExternalAddr: result.ExternalAddr, + LatencyMs: result.LatencyMs, + }) + return result, nil + } + result.NATTypeSupported = true + + select { + case <-ctx.Done(): + return result, nil + default: + } + + // Phase 2: NAT Mapping Detection (RFC 5780 Section 4.3) + reportProgress(Progress{ + Phase: PhaseNATMapping, + ExternalAddr: result.ExternalAddr, + LatencyMs: result.LatencyMs, + }) + + result.NATMapping = detectNATMapping( + packetConn, serverSocksaddr.Port, externalAddr, otherAddr, rto, + ) + + reportProgress(Progress{ + Phase: PhaseNATMapping, + ExternalAddr: result.ExternalAddr, + LatencyMs: result.LatencyMs, + NATMapping: result.NATMapping, + }) + + select { + case <-ctx.Done(): + return result, nil + default: + } + + // Phase 3: NAT Filtering Detection (RFC 5780 Section 4.4) + reportProgress(Progress{ + Phase: PhaseNATFiltering, + ExternalAddr: result.ExternalAddr, + LatencyMs: result.LatencyMs, + NATMapping: result.NATMapping, + }) + + result.NATFiltering = detectNATFiltering(packetConn, serverAddr, rto) + + reportProgress(Progress{ + Phase: PhaseDone, + ExternalAddr: result.ExternalAddr, + LatencyMs: result.LatencyMs, + NATMapping: result.NATMapping, + NATFiltering: result.NATFiltering, + }) + + return result, nil +} + +func detectNATMapping( + conn net.PacketConn, + serverPort uint16, + externalAddr netip.AddrPort, + otherAddr netip.AddrPort, + rto time.Duration, +) NATMapping { + // Mapping Test II: Send to other_ip:server_port + testIIAddr := net.UDPAddrFromAddrPort( + netip.AddrPortFrom(otherAddr.Addr(), serverPort), + ) + txID2 := newTransactionID() + resp2, _, err := roundTrip(conn, testIIAddr, txID2, nil, rto) + if err != nil { + return NATMappingUnknown + } + + externalAddr2, ok := resp2.externalAddr() + if !ok { + return NATMappingUnknown + } + + if externalAddr == externalAddr2 { + return NATMappingEndpointIndependent + } + + // Mapping Test III: Send to other_ip:other_port + testIIIAddr := net.UDPAddrFromAddrPort(otherAddr) + txID3 := newTransactionID() + resp3, _, err := roundTrip(conn, testIIIAddr, txID3, nil, rto) + if err != nil { + return NATMappingUnknown + } + + externalAddr3, ok := resp3.externalAddr() + if !ok { + return NATMappingUnknown + } + + if externalAddr2 == externalAddr3 { + return NATMappingAddressDependent + } + return NATMappingAddressAndPortDependent +} + +func detectNATFiltering( + conn net.PacketConn, + serverAddr net.Addr, + rto time.Duration, +) NATFiltering { + // Filtering Test II: Request response from different IP and port + txID := newTransactionID() + _, _, err := roundTrip(conn, serverAddr, txID, + []stunAttribute{changeRequestAttr(changeIP | changePort)}, rto) + if err == nil { + return NATFilteringEndpointIndependent + } + + // Filtering Test III: Request response from different port only + txID = newTransactionID() + _, _, err = roundTrip(conn, serverAddr, txID, + []stunAttribute{changeRequestAttr(changePort)}, rto) + if err == nil { + return NATFilteringAddressDependent + } + + return NATFilteringAddressAndPortDependent +} + +func paddingLen(n int) int { + if n%4 == 0 { + return 0 + } + return 4 - n%4 +} diff --git a/daemon/started_service.go b/daemon/started_service.go index 136cdf753..db1696623 100644 --- a/daemon/started_service.go +++ b/daemon/started_service.go @@ -8,6 +8,9 @@ import ( "time" "github.com/sagernet/sing-box/adapter" + "github.com/sagernet/sing-box/common/dialer" + "github.com/sagernet/sing-box/common/networkquality" + "github.com/sagernet/sing-box/common/stun" "github.com/sagernet/sing-box/common/urltest" "github.com/sagernet/sing-box/experimental/clashapi" "github.com/sagernet/sing-box/experimental/clashapi/trafficontrol" @@ -688,7 +691,7 @@ func (s *StartedService) SetSystemProxyEnabled(ctx context.Context, request *Set if err != nil { return nil, err } - return nil, err + return &emptypb.Empty{}, nil } func (s *StartedService) TriggerDebugCrash(ctx context.Context, request *DebugCrashRequest) (*emptypb.Empty, error) { @@ -1077,6 +1080,210 @@ func (s *StartedService) GetStartedAt(ctx context.Context, empty *emptypb.Empty) return &StartedAt{StartedAt: s.startedAt.UnixMilli()}, nil } +func (s *StartedService) ListOutbounds(ctx context.Context, _ *emptypb.Empty) (*OutboundList, error) { + s.serviceAccess.RLock() + if s.serviceStatus.Status != ServiceStatus_STARTED { + s.serviceAccess.RUnlock() + return nil, os.ErrInvalid + } + boxService := s.instance + s.serviceAccess.RUnlock() + historyStorage := boxService.urlTestHistoryStorage + outbounds := boxService.instance.Outbound().Outbounds() + var list OutboundList + for _, ob := range outbounds { + item := &GroupItem{ + Tag: ob.Tag(), + Type: ob.Type(), + } + if history := historyStorage.LoadURLTestHistory(adapter.OutboundTag(ob)); history != nil { + item.UrlTestTime = history.Time.Unix() + item.UrlTestDelay = int32(history.Delay) + } + list.Outbounds = append(list.Outbounds, item) + } + return &list, nil +} + +func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.ServerStreamingServer[OutboundList]) error { + err := s.waitForStarted(server.Context()) + if err != nil { + return err + } + subscription, done, err := s.urlTestObserver.Subscribe() + if err != nil { + return err + } + defer s.urlTestObserver.UnSubscribe(subscription) + for { + s.serviceAccess.RLock() + if s.serviceStatus.Status != ServiceStatus_STARTED { + s.serviceAccess.RUnlock() + return os.ErrInvalid + } + boxService := s.instance + s.serviceAccess.RUnlock() + historyStorage := boxService.urlTestHistoryStorage + outbounds := boxService.instance.Outbound().Outbounds() + var list OutboundList + for _, ob := range outbounds { + item := &GroupItem{ + Tag: ob.Tag(), + Type: ob.Type(), + } + if history := historyStorage.LoadURLTestHistory(adapter.OutboundTag(ob)); history != nil { + item.UrlTestTime = history.Time.Unix() + item.UrlTestDelay = int32(history.Delay) + } + list.Outbounds = append(list.Outbounds, item) + } + err = server.Send(&list) + if err != nil { + return err + } + select { + case <-subscription: + case <-s.ctx.Done(): + return s.ctx.Err() + case <-server.Context().Done(): + return server.Context().Err() + case <-done: + return nil + } + } +} + +func resolveOutbound(instance *Instance, tag string) (adapter.Outbound, error) { + if tag == "" { + return instance.instance.Outbound().Default(), nil + } + outbound, loaded := instance.instance.Outbound().Outbound(tag) + if !loaded { + return nil, E.New("outbound not found: ", tag) + } + return outbound, nil +} + +func (s *StartedService) StartNetworkQualityTest( + request *NetworkQualityTestRequest, + server grpc.ServerStreamingServer[NetworkQualityTestProgress], +) error { + err := s.waitForStarted(server.Context()) + if err != nil { + return err + } + s.serviceAccess.RLock() + boxService := s.instance + s.serviceAccess.RUnlock() + + outbound, err := resolveOutbound(boxService, request.OutboundTag) + if err != nil { + return err + } + + resolvedDialer := dialer.NewResolveDialer(boxService.ctx, outbound, true, "", adapter.DNSQueryOptions{}, 0) + httpClient := networkquality.NewHTTPClient(resolvedDialer) + defer httpClient.CloseIdleConnections() + + measurementClientFactory, err := networkquality.NewOptionalHTTP3Factory(resolvedDialer, request.Http3) + if err != nil { + return err + } + + result, nqErr := networkquality.Run(networkquality.Options{ + ConfigURL: request.ConfigURL, + HTTPClient: httpClient, + NewMeasurementClient: measurementClientFactory, + Serial: request.Serial, + MaxRuntime: time.Duration(request.MaxRuntimeSeconds) * time.Second, + Context: server.Context(), + OnProgress: func(p networkquality.Progress) { + _ = server.Send(&NetworkQualityTestProgress{ + Phase: int32(p.Phase), + DownloadCapacity: p.DownloadCapacity, + UploadCapacity: p.UploadCapacity, + DownloadRPM: p.DownloadRPM, + UploadRPM: p.UploadRPM, + IdleLatencyMs: p.IdleLatencyMs, + ElapsedMs: p.ElapsedMs, + DownloadCapacityAccuracy: int32(p.DownloadCapacityAccuracy), + UploadCapacityAccuracy: int32(p.UploadCapacityAccuracy), + DownloadRPMAccuracy: int32(p.DownloadRPMAccuracy), + UploadRPMAccuracy: int32(p.UploadRPMAccuracy), + }) + }, + }) + if nqErr != nil { + return server.Send(&NetworkQualityTestProgress{ + IsFinal: true, + Error: nqErr.Error(), + }) + } + return server.Send(&NetworkQualityTestProgress{ + Phase: int32(networkquality.PhaseDone), + DownloadCapacity: result.DownloadCapacity, + UploadCapacity: result.UploadCapacity, + DownloadRPM: result.DownloadRPM, + UploadRPM: result.UploadRPM, + IdleLatencyMs: result.IdleLatencyMs, + IsFinal: true, + DownloadCapacityAccuracy: int32(result.DownloadCapacityAccuracy), + UploadCapacityAccuracy: int32(result.UploadCapacityAccuracy), + DownloadRPMAccuracy: int32(result.DownloadRPMAccuracy), + UploadRPMAccuracy: int32(result.UploadRPMAccuracy), + }) +} + +func (s *StartedService) StartSTUNTest( + request *STUNTestRequest, + server grpc.ServerStreamingServer[STUNTestProgress], +) error { + err := s.waitForStarted(server.Context()) + if err != nil { + return err + } + s.serviceAccess.RLock() + boxService := s.instance + s.serviceAccess.RUnlock() + + outbound, err := resolveOutbound(boxService, request.OutboundTag) + if err != nil { + return err + } + + resolvedDialer := dialer.NewResolveDialer(boxService.ctx, outbound, true, "", adapter.DNSQueryOptions{}, 0) + + result, stunErr := stun.Run(stun.Options{ + Server: request.Server, + Dialer: resolvedDialer, + Context: server.Context(), + OnProgress: func(p stun.Progress) { + _ = server.Send(&STUNTestProgress{ + Phase: int32(p.Phase), + ExternalAddr: p.ExternalAddr, + LatencyMs: p.LatencyMs, + NatMapping: int32(p.NATMapping), + NatFiltering: int32(p.NATFiltering), + }) + }, + }) + if stunErr != nil { + return server.Send(&STUNTestProgress{ + IsFinal: true, + Error: stunErr.Error(), + }) + } + return server.Send(&STUNTestProgress{ + Phase: int32(stun.PhaseDone), + ExternalAddr: result.ExternalAddr, + LatencyMs: result.LatencyMs, + NatMapping: int32(result.NATMapping), + NatFiltering: int32(result.NATFiltering), + IsFinal: true, + NatTypeSupported: result.NATTypeSupported, + }) +} + func (s *StartedService) mustEmbedUnimplementedStartedServiceServer() { } diff --git a/daemon/started_service.pb.go b/daemon/started_service.pb.go index 271f80a11..c48ea4fe7 100644 --- a/daemon/started_service.pb.go +++ b/daemon/started_service.pb.go @@ -1836,6 +1836,418 @@ func (x *StartedAt) GetStartedAt() int64 { return 0 } +type OutboundList struct { + state protoimpl.MessageState `protogen:"open.v1"` + Outbounds []*GroupItem `protobuf:"bytes,1,rep,name=outbounds,proto3" json:"outbounds,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *OutboundList) Reset() { + *x = OutboundList{} + mi := &file_daemon_started_service_proto_msgTypes[26] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *OutboundList) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*OutboundList) ProtoMessage() {} + +func (x *OutboundList) ProtoReflect() protoreflect.Message { + mi := &file_daemon_started_service_proto_msgTypes[26] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use OutboundList.ProtoReflect.Descriptor instead. +func (*OutboundList) Descriptor() ([]byte, []int) { + return file_daemon_started_service_proto_rawDescGZIP(), []int{26} +} + +func (x *OutboundList) GetOutbounds() []*GroupItem { + if x != nil { + return x.Outbounds + } + return nil +} + +type NetworkQualityTestRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + ConfigURL string `protobuf:"bytes,1,opt,name=configURL,proto3" json:"configURL,omitempty"` + OutboundTag string `protobuf:"bytes,2,opt,name=outboundTag,proto3" json:"outboundTag,omitempty"` + Serial bool `protobuf:"varint,3,opt,name=serial,proto3" json:"serial,omitempty"` + MaxRuntimeSeconds int32 `protobuf:"varint,4,opt,name=maxRuntimeSeconds,proto3" json:"maxRuntimeSeconds,omitempty"` + Http3 bool `protobuf:"varint,5,opt,name=http3,proto3" json:"http3,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *NetworkQualityTestRequest) Reset() { + *x = NetworkQualityTestRequest{} + mi := &file_daemon_started_service_proto_msgTypes[27] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *NetworkQualityTestRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*NetworkQualityTestRequest) ProtoMessage() {} + +func (x *NetworkQualityTestRequest) ProtoReflect() protoreflect.Message { + mi := &file_daemon_started_service_proto_msgTypes[27] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use NetworkQualityTestRequest.ProtoReflect.Descriptor instead. +func (*NetworkQualityTestRequest) Descriptor() ([]byte, []int) { + return file_daemon_started_service_proto_rawDescGZIP(), []int{27} +} + +func (x *NetworkQualityTestRequest) GetConfigURL() string { + if x != nil { + return x.ConfigURL + } + return "" +} + +func (x *NetworkQualityTestRequest) GetOutboundTag() string { + if x != nil { + return x.OutboundTag + } + return "" +} + +func (x *NetworkQualityTestRequest) GetSerial() bool { + if x != nil { + return x.Serial + } + return false +} + +func (x *NetworkQualityTestRequest) GetMaxRuntimeSeconds() int32 { + if x != nil { + return x.MaxRuntimeSeconds + } + return 0 +} + +func (x *NetworkQualityTestRequest) GetHttp3() bool { + if x != nil { + return x.Http3 + } + return false +} + +type NetworkQualityTestProgress struct { + state protoimpl.MessageState `protogen:"open.v1"` + Phase int32 `protobuf:"varint,1,opt,name=phase,proto3" json:"phase,omitempty"` + DownloadCapacity int64 `protobuf:"varint,2,opt,name=downloadCapacity,proto3" json:"downloadCapacity,omitempty"` + UploadCapacity int64 `protobuf:"varint,3,opt,name=uploadCapacity,proto3" json:"uploadCapacity,omitempty"` + DownloadRPM int32 `protobuf:"varint,4,opt,name=downloadRPM,proto3" json:"downloadRPM,omitempty"` + UploadRPM int32 `protobuf:"varint,5,opt,name=uploadRPM,proto3" json:"uploadRPM,omitempty"` + IdleLatencyMs int32 `protobuf:"varint,6,opt,name=idleLatencyMs,proto3" json:"idleLatencyMs,omitempty"` + ElapsedMs int64 `protobuf:"varint,7,opt,name=elapsedMs,proto3" json:"elapsedMs,omitempty"` + IsFinal bool `protobuf:"varint,8,opt,name=isFinal,proto3" json:"isFinal,omitempty"` + Error string `protobuf:"bytes,9,opt,name=error,proto3" json:"error,omitempty"` + DownloadCapacityAccuracy int32 `protobuf:"varint,10,opt,name=downloadCapacityAccuracy,proto3" json:"downloadCapacityAccuracy,omitempty"` + UploadCapacityAccuracy int32 `protobuf:"varint,11,opt,name=uploadCapacityAccuracy,proto3" json:"uploadCapacityAccuracy,omitempty"` + DownloadRPMAccuracy int32 `protobuf:"varint,12,opt,name=downloadRPMAccuracy,proto3" json:"downloadRPMAccuracy,omitempty"` + UploadRPMAccuracy int32 `protobuf:"varint,13,opt,name=uploadRPMAccuracy,proto3" json:"uploadRPMAccuracy,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *NetworkQualityTestProgress) Reset() { + *x = NetworkQualityTestProgress{} + mi := &file_daemon_started_service_proto_msgTypes[28] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *NetworkQualityTestProgress) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*NetworkQualityTestProgress) ProtoMessage() {} + +func (x *NetworkQualityTestProgress) ProtoReflect() protoreflect.Message { + mi := &file_daemon_started_service_proto_msgTypes[28] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use NetworkQualityTestProgress.ProtoReflect.Descriptor instead. +func (*NetworkQualityTestProgress) Descriptor() ([]byte, []int) { + return file_daemon_started_service_proto_rawDescGZIP(), []int{28} +} + +func (x *NetworkQualityTestProgress) GetPhase() int32 { + if x != nil { + return x.Phase + } + return 0 +} + +func (x *NetworkQualityTestProgress) GetDownloadCapacity() int64 { + if x != nil { + return x.DownloadCapacity + } + return 0 +} + +func (x *NetworkQualityTestProgress) GetUploadCapacity() int64 { + if x != nil { + return x.UploadCapacity + } + return 0 +} + +func (x *NetworkQualityTestProgress) GetDownloadRPM() int32 { + if x != nil { + return x.DownloadRPM + } + return 0 +} + +func (x *NetworkQualityTestProgress) GetUploadRPM() int32 { + if x != nil { + return x.UploadRPM + } + return 0 +} + +func (x *NetworkQualityTestProgress) GetIdleLatencyMs() int32 { + if x != nil { + return x.IdleLatencyMs + } + return 0 +} + +func (x *NetworkQualityTestProgress) GetElapsedMs() int64 { + if x != nil { + return x.ElapsedMs + } + return 0 +} + +func (x *NetworkQualityTestProgress) GetIsFinal() bool { + if x != nil { + return x.IsFinal + } + return false +} + +func (x *NetworkQualityTestProgress) GetError() string { + if x != nil { + return x.Error + } + return "" +} + +func (x *NetworkQualityTestProgress) GetDownloadCapacityAccuracy() int32 { + if x != nil { + return x.DownloadCapacityAccuracy + } + return 0 +} + +func (x *NetworkQualityTestProgress) GetUploadCapacityAccuracy() int32 { + if x != nil { + return x.UploadCapacityAccuracy + } + return 0 +} + +func (x *NetworkQualityTestProgress) GetDownloadRPMAccuracy() int32 { + if x != nil { + return x.DownloadRPMAccuracy + } + return 0 +} + +func (x *NetworkQualityTestProgress) GetUploadRPMAccuracy() int32 { + if x != nil { + return x.UploadRPMAccuracy + } + return 0 +} + +type STUNTestRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Server string `protobuf:"bytes,1,opt,name=server,proto3" json:"server,omitempty"` + OutboundTag string `protobuf:"bytes,2,opt,name=outboundTag,proto3" json:"outboundTag,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *STUNTestRequest) Reset() { + *x = STUNTestRequest{} + mi := &file_daemon_started_service_proto_msgTypes[29] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *STUNTestRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*STUNTestRequest) ProtoMessage() {} + +func (x *STUNTestRequest) ProtoReflect() protoreflect.Message { + mi := &file_daemon_started_service_proto_msgTypes[29] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use STUNTestRequest.ProtoReflect.Descriptor instead. +func (*STUNTestRequest) Descriptor() ([]byte, []int) { + return file_daemon_started_service_proto_rawDescGZIP(), []int{29} +} + +func (x *STUNTestRequest) GetServer() string { + if x != nil { + return x.Server + } + return "" +} + +func (x *STUNTestRequest) GetOutboundTag() string { + if x != nil { + return x.OutboundTag + } + return "" +} + +type STUNTestProgress struct { + state protoimpl.MessageState `protogen:"open.v1"` + Phase int32 `protobuf:"varint,1,opt,name=phase,proto3" json:"phase,omitempty"` + ExternalAddr string `protobuf:"bytes,2,opt,name=externalAddr,proto3" json:"externalAddr,omitempty"` + LatencyMs int32 `protobuf:"varint,3,opt,name=latencyMs,proto3" json:"latencyMs,omitempty"` + NatMapping int32 `protobuf:"varint,4,opt,name=natMapping,proto3" json:"natMapping,omitempty"` + NatFiltering int32 `protobuf:"varint,5,opt,name=natFiltering,proto3" json:"natFiltering,omitempty"` + IsFinal bool `protobuf:"varint,6,opt,name=isFinal,proto3" json:"isFinal,omitempty"` + Error string `protobuf:"bytes,7,opt,name=error,proto3" json:"error,omitempty"` + NatTypeSupported bool `protobuf:"varint,8,opt,name=natTypeSupported,proto3" json:"natTypeSupported,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *STUNTestProgress) Reset() { + *x = STUNTestProgress{} + mi := &file_daemon_started_service_proto_msgTypes[30] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *STUNTestProgress) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*STUNTestProgress) ProtoMessage() {} + +func (x *STUNTestProgress) ProtoReflect() protoreflect.Message { + mi := &file_daemon_started_service_proto_msgTypes[30] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use STUNTestProgress.ProtoReflect.Descriptor instead. +func (*STUNTestProgress) Descriptor() ([]byte, []int) { + return file_daemon_started_service_proto_rawDescGZIP(), []int{30} +} + +func (x *STUNTestProgress) GetPhase() int32 { + if x != nil { + return x.Phase + } + return 0 +} + +func (x *STUNTestProgress) GetExternalAddr() string { + if x != nil { + return x.ExternalAddr + } + return "" +} + +func (x *STUNTestProgress) GetLatencyMs() int32 { + if x != nil { + return x.LatencyMs + } + return 0 +} + +func (x *STUNTestProgress) GetNatMapping() int32 { + if x != nil { + return x.NatMapping + } + return 0 +} + +func (x *STUNTestProgress) GetNatFiltering() int32 { + if x != nil { + return x.NatFiltering + } + return 0 +} + +func (x *STUNTestProgress) GetIsFinal() bool { + if x != nil { + return x.IsFinal + } + return false +} + +func (x *STUNTestProgress) GetError() string { + if x != nil { + return x.Error + } + return "" +} + +func (x *STUNTestProgress) GetNatTypeSupported() bool { + if x != nil { + return x.NatTypeSupported + } + return false +} + type Log_Message struct { state protoimpl.MessageState `protogen:"open.v1"` Level LogLevel `protobuf:"varint,1,opt,name=level,proto3,enum=daemon.LogLevel" json:"level,omitempty"` @@ -1846,7 +2258,7 @@ type Log_Message struct { func (x *Log_Message) Reset() { *x = Log_Message{} - mi := &file_daemon_started_service_proto_msgTypes[26] + mi := &file_daemon_started_service_proto_msgTypes[31] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1858,7 +2270,7 @@ func (x *Log_Message) String() string { func (*Log_Message) ProtoMessage() {} func (x *Log_Message) ProtoReflect() protoreflect.Message { - mi := &file_daemon_started_service_proto_msgTypes[26] + mi := &file_daemon_started_service_proto_msgTypes[31] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2023,7 +2435,44 @@ const file_daemon_started_service_proto_rawDesc = "" + "\x11deprecatedVersion\x18\x05 \x01(\tR\x11deprecatedVersion\x12*\n" + "\x10scheduledVersion\x18\x06 \x01(\tR\x10scheduledVersion\")\n" + "\tStartedAt\x12\x1c\n" + - "\tstartedAt\x18\x01 \x01(\x03R\tstartedAt*U\n" + + "\tstartedAt\x18\x01 \x01(\x03R\tstartedAt\"?\n" + + "\fOutboundList\x12/\n" + + "\toutbounds\x18\x01 \x03(\v2\x11.daemon.GroupItemR\toutbounds\"\xb7\x01\n" + + "\x19NetworkQualityTestRequest\x12\x1c\n" + + "\tconfigURL\x18\x01 \x01(\tR\tconfigURL\x12 \n" + + "\voutboundTag\x18\x02 \x01(\tR\voutboundTag\x12\x16\n" + + "\x06serial\x18\x03 \x01(\bR\x06serial\x12,\n" + + "\x11maxRuntimeSeconds\x18\x04 \x01(\x05R\x11maxRuntimeSeconds\x12\x14\n" + + "\x05http3\x18\x05 \x01(\bR\x05http3\"\x8e\x04\n" + + "\x1aNetworkQualityTestProgress\x12\x14\n" + + "\x05phase\x18\x01 \x01(\x05R\x05phase\x12*\n" + + "\x10downloadCapacity\x18\x02 \x01(\x03R\x10downloadCapacity\x12&\n" + + "\x0euploadCapacity\x18\x03 \x01(\x03R\x0euploadCapacity\x12 \n" + + "\vdownloadRPM\x18\x04 \x01(\x05R\vdownloadRPM\x12\x1c\n" + + "\tuploadRPM\x18\x05 \x01(\x05R\tuploadRPM\x12$\n" + + "\ridleLatencyMs\x18\x06 \x01(\x05R\ridleLatencyMs\x12\x1c\n" + + "\telapsedMs\x18\a \x01(\x03R\telapsedMs\x12\x18\n" + + "\aisFinal\x18\b \x01(\bR\aisFinal\x12\x14\n" + + "\x05error\x18\t \x01(\tR\x05error\x12:\n" + + "\x18downloadCapacityAccuracy\x18\n" + + " \x01(\x05R\x18downloadCapacityAccuracy\x126\n" + + "\x16uploadCapacityAccuracy\x18\v \x01(\x05R\x16uploadCapacityAccuracy\x120\n" + + "\x13downloadRPMAccuracy\x18\f \x01(\x05R\x13downloadRPMAccuracy\x12,\n" + + "\x11uploadRPMAccuracy\x18\r \x01(\x05R\x11uploadRPMAccuracy\"K\n" + + "\x0fSTUNTestRequest\x12\x16\n" + + "\x06server\x18\x01 \x01(\tR\x06server\x12 \n" + + "\voutboundTag\x18\x02 \x01(\tR\voutboundTag\"\x8a\x02\n" + + "\x10STUNTestProgress\x12\x14\n" + + "\x05phase\x18\x01 \x01(\x05R\x05phase\x12\"\n" + + "\fexternalAddr\x18\x02 \x01(\tR\fexternalAddr\x12\x1c\n" + + "\tlatencyMs\x18\x03 \x01(\x05R\tlatencyMs\x12\x1e\n" + + "\n" + + "natMapping\x18\x04 \x01(\x05R\n" + + "natMapping\x12\"\n" + + "\fnatFiltering\x18\x05 \x01(\x05R\fnatFiltering\x12\x18\n" + + "\aisFinal\x18\x06 \x01(\bR\aisFinal\x12\x14\n" + + "\x05error\x18\a \x01(\tR\x05error\x12*\n" + + "\x10natTypeSupported\x18\b \x01(\bR\x10natTypeSupported*U\n" + "\bLogLevel\x12\t\n" + "\x05PANIC\x10\x00\x12\t\n" + "\x05FATAL\x10\x01\x12\t\n" + @@ -2035,7 +2484,7 @@ const file_daemon_started_service_proto_rawDesc = "" + "\x13ConnectionEventType\x12\x18\n" + "\x14CONNECTION_EVENT_NEW\x10\x00\x12\x1b\n" + "\x17CONNECTION_EVENT_UPDATE\x10\x01\x12\x1b\n" + - "\x17CONNECTION_EVENT_CLOSED\x10\x022\xf5\f\n" + + "\x17CONNECTION_EVENT_CLOSED\x10\x022\xac\x0f\n" + "\x0eStartedService\x12=\n" + "\vStopService\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\x12?\n" + "\rReloadService\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\x12K\n" + @@ -2059,7 +2508,11 @@ const file_daemon_started_service_proto_rawDesc = "" + "\x0fCloseConnection\x12\x1e.daemon.CloseConnectionRequest\x1a\x16.google.protobuf.Empty\"\x00\x12G\n" + "\x13CloseAllConnections\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\"\x00\x12M\n" + "\x15GetDeprecatedWarnings\x12\x16.google.protobuf.Empty\x1a\x1a.daemon.DeprecatedWarnings\"\x00\x12;\n" + - "\fGetStartedAt\x12\x16.google.protobuf.Empty\x1a\x11.daemon.StartedAt\"\x00B%Z#github.com/sagernet/sing-box/daemonb\x06proto3" + "\fGetStartedAt\x12\x16.google.protobuf.Empty\x1a\x11.daemon.StartedAt\"\x00\x12?\n" + + "\rListOutbounds\x12\x16.google.protobuf.Empty\x1a\x14.daemon.OutboundList\"\x00\x12F\n" + + "\x12SubscribeOutbounds\x12\x16.google.protobuf.Empty\x1a\x14.daemon.OutboundList\"\x000\x01\x12d\n" + + "\x17StartNetworkQualityTest\x12!.daemon.NetworkQualityTestRequest\x1a\".daemon.NetworkQualityTestProgress\"\x000\x01\x12F\n" + + "\rStartSTUNTest\x12\x17.daemon.STUNTestRequest\x1a\x18.daemon.STUNTestProgress\"\x000\x01B%Z#github.com/sagernet/sing-box/daemonb\x06proto3" var ( file_daemon_started_service_proto_rawDescOnce sync.Once @@ -2075,7 +2528,7 @@ func file_daemon_started_service_proto_rawDescGZIP() []byte { var ( file_daemon_started_service_proto_enumTypes = make([]protoimpl.EnumInfo, 4) - file_daemon_started_service_proto_msgTypes = make([]protoimpl.MessageInfo, 27) + file_daemon_started_service_proto_msgTypes = make([]protoimpl.MessageInfo, 32) file_daemon_started_service_proto_goTypes = []any{ (LogLevel)(0), // 0: daemon.LogLevel (ConnectionEventType)(0), // 1: daemon.ConnectionEventType @@ -2107,14 +2560,19 @@ var ( (*DeprecatedWarnings)(nil), // 27: daemon.DeprecatedWarnings (*DeprecatedWarning)(nil), // 28: daemon.DeprecatedWarning (*StartedAt)(nil), // 29: daemon.StartedAt - (*Log_Message)(nil), // 30: daemon.Log.Message - (*emptypb.Empty)(nil), // 31: google.protobuf.Empty + (*OutboundList)(nil), // 30: daemon.OutboundList + (*NetworkQualityTestRequest)(nil), // 31: daemon.NetworkQualityTestRequest + (*NetworkQualityTestProgress)(nil), // 32: daemon.NetworkQualityTestProgress + (*STUNTestRequest)(nil), // 33: daemon.STUNTestRequest + (*STUNTestProgress)(nil), // 34: daemon.STUNTestProgress + (*Log_Message)(nil), // 35: daemon.Log.Message + (*emptypb.Empty)(nil), // 36: google.protobuf.Empty } ) var file_daemon_started_service_proto_depIdxs = []int32{ 2, // 0: daemon.ServiceStatus.status:type_name -> daemon.ServiceStatus.Type - 30, // 1: daemon.Log.messages:type_name -> daemon.Log.Message + 35, // 1: daemon.Log.messages:type_name -> daemon.Log.Message 0, // 2: daemon.DefaultLogLevel.level:type_name -> daemon.LogLevel 11, // 3: daemon.Groups.group:type_name -> daemon.Group 12, // 4: daemon.Group.items:type_name -> daemon.GroupItem @@ -2124,58 +2582,67 @@ var file_daemon_started_service_proto_depIdxs = []int32{ 22, // 8: daemon.ConnectionEvents.events:type_name -> daemon.ConnectionEvent 25, // 9: daemon.Connection.processInfo:type_name -> daemon.ProcessInfo 28, // 10: daemon.DeprecatedWarnings.warnings:type_name -> daemon.DeprecatedWarning - 0, // 11: daemon.Log.Message.level:type_name -> daemon.LogLevel - 31, // 12: daemon.StartedService.StopService:input_type -> google.protobuf.Empty - 31, // 13: daemon.StartedService.ReloadService:input_type -> google.protobuf.Empty - 31, // 14: daemon.StartedService.SubscribeServiceStatus:input_type -> google.protobuf.Empty - 31, // 15: daemon.StartedService.SubscribeLog:input_type -> google.protobuf.Empty - 31, // 16: daemon.StartedService.GetDefaultLogLevel:input_type -> google.protobuf.Empty - 31, // 17: daemon.StartedService.ClearLogs:input_type -> google.protobuf.Empty - 6, // 18: daemon.StartedService.SubscribeStatus:input_type -> daemon.SubscribeStatusRequest - 31, // 19: daemon.StartedService.SubscribeGroups:input_type -> google.protobuf.Empty - 31, // 20: daemon.StartedService.GetClashModeStatus:input_type -> google.protobuf.Empty - 31, // 21: daemon.StartedService.SubscribeClashMode:input_type -> google.protobuf.Empty - 16, // 22: daemon.StartedService.SetClashMode:input_type -> daemon.ClashMode - 13, // 23: daemon.StartedService.URLTest:input_type -> daemon.URLTestRequest - 14, // 24: daemon.StartedService.SelectOutbound:input_type -> daemon.SelectOutboundRequest - 15, // 25: daemon.StartedService.SetGroupExpand:input_type -> daemon.SetGroupExpandRequest - 31, // 26: daemon.StartedService.GetSystemProxyStatus:input_type -> google.protobuf.Empty - 19, // 27: daemon.StartedService.SetSystemProxyEnabled:input_type -> daemon.SetSystemProxyEnabledRequest - 20, // 28: daemon.StartedService.TriggerDebugCrash:input_type -> daemon.DebugCrashRequest - 31, // 29: daemon.StartedService.TriggerOOMReport:input_type -> google.protobuf.Empty - 21, // 30: daemon.StartedService.SubscribeConnections:input_type -> daemon.SubscribeConnectionsRequest - 26, // 31: daemon.StartedService.CloseConnection:input_type -> daemon.CloseConnectionRequest - 31, // 32: daemon.StartedService.CloseAllConnections:input_type -> google.protobuf.Empty - 31, // 33: daemon.StartedService.GetDeprecatedWarnings:input_type -> google.protobuf.Empty - 31, // 34: daemon.StartedService.GetStartedAt:input_type -> google.protobuf.Empty - 31, // 35: daemon.StartedService.StopService:output_type -> google.protobuf.Empty - 31, // 36: daemon.StartedService.ReloadService:output_type -> google.protobuf.Empty - 4, // 37: daemon.StartedService.SubscribeServiceStatus:output_type -> daemon.ServiceStatus - 7, // 38: daemon.StartedService.SubscribeLog:output_type -> daemon.Log - 8, // 39: daemon.StartedService.GetDefaultLogLevel:output_type -> daemon.DefaultLogLevel - 31, // 40: daemon.StartedService.ClearLogs:output_type -> google.protobuf.Empty - 9, // 41: daemon.StartedService.SubscribeStatus:output_type -> daemon.Status - 10, // 42: daemon.StartedService.SubscribeGroups:output_type -> daemon.Groups - 17, // 43: daemon.StartedService.GetClashModeStatus:output_type -> daemon.ClashModeStatus - 16, // 44: daemon.StartedService.SubscribeClashMode:output_type -> daemon.ClashMode - 31, // 45: daemon.StartedService.SetClashMode:output_type -> google.protobuf.Empty - 31, // 46: daemon.StartedService.URLTest:output_type -> google.protobuf.Empty - 31, // 47: daemon.StartedService.SelectOutbound:output_type -> google.protobuf.Empty - 31, // 48: daemon.StartedService.SetGroupExpand:output_type -> google.protobuf.Empty - 18, // 49: daemon.StartedService.GetSystemProxyStatus:output_type -> daemon.SystemProxyStatus - 31, // 50: daemon.StartedService.SetSystemProxyEnabled:output_type -> google.protobuf.Empty - 31, // 51: daemon.StartedService.TriggerDebugCrash:output_type -> google.protobuf.Empty - 31, // 52: daemon.StartedService.TriggerOOMReport:output_type -> google.protobuf.Empty - 23, // 53: daemon.StartedService.SubscribeConnections:output_type -> daemon.ConnectionEvents - 31, // 54: daemon.StartedService.CloseConnection:output_type -> google.protobuf.Empty - 31, // 55: daemon.StartedService.CloseAllConnections:output_type -> google.protobuf.Empty - 27, // 56: daemon.StartedService.GetDeprecatedWarnings:output_type -> daemon.DeprecatedWarnings - 29, // 57: daemon.StartedService.GetStartedAt:output_type -> daemon.StartedAt - 35, // [35:58] is the sub-list for method output_type - 12, // [12:35] is the sub-list for method input_type - 12, // [12:12] is the sub-list for extension type_name - 12, // [12:12] is the sub-list for extension extendee - 0, // [0:12] is the sub-list for field type_name + 12, // 11: daemon.OutboundList.outbounds:type_name -> daemon.GroupItem + 0, // 12: daemon.Log.Message.level:type_name -> daemon.LogLevel + 36, // 13: daemon.StartedService.StopService:input_type -> google.protobuf.Empty + 36, // 14: daemon.StartedService.ReloadService:input_type -> google.protobuf.Empty + 36, // 15: daemon.StartedService.SubscribeServiceStatus:input_type -> google.protobuf.Empty + 36, // 16: daemon.StartedService.SubscribeLog:input_type -> google.protobuf.Empty + 36, // 17: daemon.StartedService.GetDefaultLogLevel:input_type -> google.protobuf.Empty + 36, // 18: daemon.StartedService.ClearLogs:input_type -> google.protobuf.Empty + 6, // 19: daemon.StartedService.SubscribeStatus:input_type -> daemon.SubscribeStatusRequest + 36, // 20: daemon.StartedService.SubscribeGroups:input_type -> google.protobuf.Empty + 36, // 21: daemon.StartedService.GetClashModeStatus:input_type -> google.protobuf.Empty + 36, // 22: daemon.StartedService.SubscribeClashMode:input_type -> google.protobuf.Empty + 16, // 23: daemon.StartedService.SetClashMode:input_type -> daemon.ClashMode + 13, // 24: daemon.StartedService.URLTest:input_type -> daemon.URLTestRequest + 14, // 25: daemon.StartedService.SelectOutbound:input_type -> daemon.SelectOutboundRequest + 15, // 26: daemon.StartedService.SetGroupExpand:input_type -> daemon.SetGroupExpandRequest + 36, // 27: daemon.StartedService.GetSystemProxyStatus:input_type -> google.protobuf.Empty + 19, // 28: daemon.StartedService.SetSystemProxyEnabled:input_type -> daemon.SetSystemProxyEnabledRequest + 20, // 29: daemon.StartedService.TriggerDebugCrash:input_type -> daemon.DebugCrashRequest + 36, // 30: daemon.StartedService.TriggerOOMReport:input_type -> google.protobuf.Empty + 21, // 31: daemon.StartedService.SubscribeConnections:input_type -> daemon.SubscribeConnectionsRequest + 26, // 32: daemon.StartedService.CloseConnection:input_type -> daemon.CloseConnectionRequest + 36, // 33: daemon.StartedService.CloseAllConnections:input_type -> google.protobuf.Empty + 36, // 34: daemon.StartedService.GetDeprecatedWarnings:input_type -> google.protobuf.Empty + 36, // 35: daemon.StartedService.GetStartedAt:input_type -> google.protobuf.Empty + 36, // 36: daemon.StartedService.ListOutbounds:input_type -> google.protobuf.Empty + 36, // 37: daemon.StartedService.SubscribeOutbounds:input_type -> google.protobuf.Empty + 31, // 38: daemon.StartedService.StartNetworkQualityTest:input_type -> daemon.NetworkQualityTestRequest + 33, // 39: daemon.StartedService.StartSTUNTest:input_type -> daemon.STUNTestRequest + 36, // 40: daemon.StartedService.StopService:output_type -> google.protobuf.Empty + 36, // 41: daemon.StartedService.ReloadService:output_type -> google.protobuf.Empty + 4, // 42: daemon.StartedService.SubscribeServiceStatus:output_type -> daemon.ServiceStatus + 7, // 43: daemon.StartedService.SubscribeLog:output_type -> daemon.Log + 8, // 44: daemon.StartedService.GetDefaultLogLevel:output_type -> daemon.DefaultLogLevel + 36, // 45: daemon.StartedService.ClearLogs:output_type -> google.protobuf.Empty + 9, // 46: daemon.StartedService.SubscribeStatus:output_type -> daemon.Status + 10, // 47: daemon.StartedService.SubscribeGroups:output_type -> daemon.Groups + 17, // 48: daemon.StartedService.GetClashModeStatus:output_type -> daemon.ClashModeStatus + 16, // 49: daemon.StartedService.SubscribeClashMode:output_type -> daemon.ClashMode + 36, // 50: daemon.StartedService.SetClashMode:output_type -> google.protobuf.Empty + 36, // 51: daemon.StartedService.URLTest:output_type -> google.protobuf.Empty + 36, // 52: daemon.StartedService.SelectOutbound:output_type -> google.protobuf.Empty + 36, // 53: daemon.StartedService.SetGroupExpand:output_type -> google.protobuf.Empty + 18, // 54: daemon.StartedService.GetSystemProxyStatus:output_type -> daemon.SystemProxyStatus + 36, // 55: daemon.StartedService.SetSystemProxyEnabled:output_type -> google.protobuf.Empty + 36, // 56: daemon.StartedService.TriggerDebugCrash:output_type -> google.protobuf.Empty + 36, // 57: daemon.StartedService.TriggerOOMReport:output_type -> google.protobuf.Empty + 23, // 58: daemon.StartedService.SubscribeConnections:output_type -> daemon.ConnectionEvents + 36, // 59: daemon.StartedService.CloseConnection:output_type -> google.protobuf.Empty + 36, // 60: daemon.StartedService.CloseAllConnections:output_type -> google.protobuf.Empty + 27, // 61: daemon.StartedService.GetDeprecatedWarnings:output_type -> daemon.DeprecatedWarnings + 29, // 62: daemon.StartedService.GetStartedAt:output_type -> daemon.StartedAt + 30, // 63: daemon.StartedService.ListOutbounds:output_type -> daemon.OutboundList + 30, // 64: daemon.StartedService.SubscribeOutbounds:output_type -> daemon.OutboundList + 32, // 65: daemon.StartedService.StartNetworkQualityTest:output_type -> daemon.NetworkQualityTestProgress + 34, // 66: daemon.StartedService.StartSTUNTest:output_type -> daemon.STUNTestProgress + 40, // [40:67] is the sub-list for method output_type + 13, // [13:40] is the sub-list for method input_type + 13, // [13:13] is the sub-list for extension type_name + 13, // [13:13] is the sub-list for extension extendee + 0, // [0:13] is the sub-list for field type_name } func init() { file_daemon_started_service_proto_init() } @@ -2189,7 +2656,7 @@ func file_daemon_started_service_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_daemon_started_service_proto_rawDesc), len(file_daemon_started_service_proto_rawDesc)), NumEnums: 4, - NumMessages: 27, + NumMessages: 32, NumExtensions: 0, NumServices: 1, }, diff --git a/daemon/started_service.proto b/daemon/started_service.proto index 27f8667fb..6d30da2f9 100644 --- a/daemon/started_service.proto +++ b/daemon/started_service.proto @@ -34,6 +34,11 @@ service StartedService { rpc CloseAllConnections(google.protobuf.Empty) returns(google.protobuf.Empty) {} rpc GetDeprecatedWarnings(google.protobuf.Empty) returns(DeprecatedWarnings) {} rpc GetStartedAt(google.protobuf.Empty) returns(StartedAt) {} + + rpc ListOutbounds(google.protobuf.Empty) returns (OutboundList) {} + rpc SubscribeOutbounds(google.protobuf.Empty) returns (stream OutboundList) {} + rpc StartNetworkQualityTest(NetworkQualityTestRequest) returns (stream NetworkQualityTestProgress) {} + rpc StartSTUNTest(STUNTestRequest) returns (stream STUNTestProgress) {} } message ServiceStatus { @@ -229,3 +234,47 @@ message DeprecatedWarning { message StartedAt { int64 startedAt = 1; } + +message OutboundList { + repeated GroupItem outbounds = 1; +} + +message NetworkQualityTestRequest { + string configURL = 1; + string outboundTag = 2; + bool serial = 3; + int32 maxRuntimeSeconds = 4; + bool http3 = 5; +} + +message NetworkQualityTestProgress { + int32 phase = 1; + int64 downloadCapacity = 2; + int64 uploadCapacity = 3; + int32 downloadRPM = 4; + int32 uploadRPM = 5; + int32 idleLatencyMs = 6; + int64 elapsedMs = 7; + bool isFinal = 8; + string error = 9; + int32 downloadCapacityAccuracy = 10; + int32 uploadCapacityAccuracy = 11; + int32 downloadRPMAccuracy = 12; + int32 uploadRPMAccuracy = 13; +} + +message STUNTestRequest { + string server = 1; + string outboundTag = 2; +} + +message STUNTestProgress { + int32 phase = 1; + string externalAddr = 2; + int32 latencyMs = 3; + int32 natMapping = 4; + int32 natFiltering = 5; + bool isFinal = 6; + string error = 7; + bool natTypeSupported = 8; +} diff --git a/daemon/started_service_grpc.pb.go b/daemon/started_service_grpc.pb.go index bdf81e4a6..fec7afa0b 100644 --- a/daemon/started_service_grpc.pb.go +++ b/daemon/started_service_grpc.pb.go @@ -15,29 +15,33 @@ import ( const _ = grpc.SupportPackageIsVersion9 const ( - StartedService_StopService_FullMethodName = "/daemon.StartedService/StopService" - StartedService_ReloadService_FullMethodName = "/daemon.StartedService/ReloadService" - StartedService_SubscribeServiceStatus_FullMethodName = "/daemon.StartedService/SubscribeServiceStatus" - StartedService_SubscribeLog_FullMethodName = "/daemon.StartedService/SubscribeLog" - StartedService_GetDefaultLogLevel_FullMethodName = "/daemon.StartedService/GetDefaultLogLevel" - StartedService_ClearLogs_FullMethodName = "/daemon.StartedService/ClearLogs" - StartedService_SubscribeStatus_FullMethodName = "/daemon.StartedService/SubscribeStatus" - StartedService_SubscribeGroups_FullMethodName = "/daemon.StartedService/SubscribeGroups" - StartedService_GetClashModeStatus_FullMethodName = "/daemon.StartedService/GetClashModeStatus" - StartedService_SubscribeClashMode_FullMethodName = "/daemon.StartedService/SubscribeClashMode" - StartedService_SetClashMode_FullMethodName = "/daemon.StartedService/SetClashMode" - StartedService_URLTest_FullMethodName = "/daemon.StartedService/URLTest" - StartedService_SelectOutbound_FullMethodName = "/daemon.StartedService/SelectOutbound" - StartedService_SetGroupExpand_FullMethodName = "/daemon.StartedService/SetGroupExpand" - StartedService_GetSystemProxyStatus_FullMethodName = "/daemon.StartedService/GetSystemProxyStatus" - StartedService_SetSystemProxyEnabled_FullMethodName = "/daemon.StartedService/SetSystemProxyEnabled" - StartedService_TriggerDebugCrash_FullMethodName = "/daemon.StartedService/TriggerDebugCrash" - StartedService_TriggerOOMReport_FullMethodName = "/daemon.StartedService/TriggerOOMReport" - StartedService_SubscribeConnections_FullMethodName = "/daemon.StartedService/SubscribeConnections" - StartedService_CloseConnection_FullMethodName = "/daemon.StartedService/CloseConnection" - StartedService_CloseAllConnections_FullMethodName = "/daemon.StartedService/CloseAllConnections" - StartedService_GetDeprecatedWarnings_FullMethodName = "/daemon.StartedService/GetDeprecatedWarnings" - StartedService_GetStartedAt_FullMethodName = "/daemon.StartedService/GetStartedAt" + StartedService_StopService_FullMethodName = "/daemon.StartedService/StopService" + StartedService_ReloadService_FullMethodName = "/daemon.StartedService/ReloadService" + StartedService_SubscribeServiceStatus_FullMethodName = "/daemon.StartedService/SubscribeServiceStatus" + StartedService_SubscribeLog_FullMethodName = "/daemon.StartedService/SubscribeLog" + StartedService_GetDefaultLogLevel_FullMethodName = "/daemon.StartedService/GetDefaultLogLevel" + StartedService_ClearLogs_FullMethodName = "/daemon.StartedService/ClearLogs" + StartedService_SubscribeStatus_FullMethodName = "/daemon.StartedService/SubscribeStatus" + StartedService_SubscribeGroups_FullMethodName = "/daemon.StartedService/SubscribeGroups" + StartedService_GetClashModeStatus_FullMethodName = "/daemon.StartedService/GetClashModeStatus" + StartedService_SubscribeClashMode_FullMethodName = "/daemon.StartedService/SubscribeClashMode" + StartedService_SetClashMode_FullMethodName = "/daemon.StartedService/SetClashMode" + StartedService_URLTest_FullMethodName = "/daemon.StartedService/URLTest" + StartedService_SelectOutbound_FullMethodName = "/daemon.StartedService/SelectOutbound" + StartedService_SetGroupExpand_FullMethodName = "/daemon.StartedService/SetGroupExpand" + StartedService_GetSystemProxyStatus_FullMethodName = "/daemon.StartedService/GetSystemProxyStatus" + StartedService_SetSystemProxyEnabled_FullMethodName = "/daemon.StartedService/SetSystemProxyEnabled" + StartedService_TriggerDebugCrash_FullMethodName = "/daemon.StartedService/TriggerDebugCrash" + StartedService_TriggerOOMReport_FullMethodName = "/daemon.StartedService/TriggerOOMReport" + StartedService_SubscribeConnections_FullMethodName = "/daemon.StartedService/SubscribeConnections" + StartedService_CloseConnection_FullMethodName = "/daemon.StartedService/CloseConnection" + StartedService_CloseAllConnections_FullMethodName = "/daemon.StartedService/CloseAllConnections" + StartedService_GetDeprecatedWarnings_FullMethodName = "/daemon.StartedService/GetDeprecatedWarnings" + StartedService_GetStartedAt_FullMethodName = "/daemon.StartedService/GetStartedAt" + StartedService_ListOutbounds_FullMethodName = "/daemon.StartedService/ListOutbounds" + StartedService_SubscribeOutbounds_FullMethodName = "/daemon.StartedService/SubscribeOutbounds" + StartedService_StartNetworkQualityTest_FullMethodName = "/daemon.StartedService/StartNetworkQualityTest" + StartedService_StartSTUNTest_FullMethodName = "/daemon.StartedService/StartSTUNTest" ) // StartedServiceClient is the client API for StartedService service. @@ -67,6 +71,10 @@ type StartedServiceClient interface { CloseAllConnections(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*emptypb.Empty, error) GetDeprecatedWarnings(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*DeprecatedWarnings, error) GetStartedAt(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*StartedAt, error) + ListOutbounds(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*OutboundList, error) + SubscribeOutbounds(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (grpc.ServerStreamingClient[OutboundList], error) + StartNetworkQualityTest(ctx context.Context, in *NetworkQualityTestRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[NetworkQualityTestProgress], error) + StartSTUNTest(ctx context.Context, in *STUNTestRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[STUNTestProgress], error) } type startedServiceClient struct { @@ -361,6 +369,73 @@ func (c *startedServiceClient) GetStartedAt(ctx context.Context, in *emptypb.Emp return out, nil } +func (c *startedServiceClient) ListOutbounds(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*OutboundList, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(OutboundList) + err := c.cc.Invoke(ctx, StartedService_ListOutbounds_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *startedServiceClient) SubscribeOutbounds(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (grpc.ServerStreamingClient[OutboundList], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &StartedService_ServiceDesc.Streams[6], StartedService_SubscribeOutbounds_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[emptypb.Empty, OutboundList]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StartedService_SubscribeOutboundsClient = grpc.ServerStreamingClient[OutboundList] + +func (c *startedServiceClient) StartNetworkQualityTest(ctx context.Context, in *NetworkQualityTestRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[NetworkQualityTestProgress], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &StartedService_ServiceDesc.Streams[7], StartedService_StartNetworkQualityTest_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[NetworkQualityTestRequest, NetworkQualityTestProgress]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StartedService_StartNetworkQualityTestClient = grpc.ServerStreamingClient[NetworkQualityTestProgress] + +func (c *startedServiceClient) StartSTUNTest(ctx context.Context, in *STUNTestRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[STUNTestProgress], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &StartedService_ServiceDesc.Streams[8], StartedService_StartSTUNTest_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[STUNTestRequest, STUNTestProgress]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StartedService_StartSTUNTestClient = grpc.ServerStreamingClient[STUNTestProgress] + // StartedServiceServer is the server API for StartedService service. // All implementations must embed UnimplementedStartedServiceServer // for forward compatibility. @@ -388,6 +463,10 @@ type StartedServiceServer interface { CloseAllConnections(context.Context, *emptypb.Empty) (*emptypb.Empty, error) GetDeprecatedWarnings(context.Context, *emptypb.Empty) (*DeprecatedWarnings, error) GetStartedAt(context.Context, *emptypb.Empty) (*StartedAt, error) + ListOutbounds(context.Context, *emptypb.Empty) (*OutboundList, error) + SubscribeOutbounds(*emptypb.Empty, grpc.ServerStreamingServer[OutboundList]) error + StartNetworkQualityTest(*NetworkQualityTestRequest, grpc.ServerStreamingServer[NetworkQualityTestProgress]) error + StartSTUNTest(*STUNTestRequest, grpc.ServerStreamingServer[STUNTestProgress]) error mustEmbedUnimplementedStartedServiceServer() } @@ -489,6 +568,22 @@ func (UnimplementedStartedServiceServer) GetDeprecatedWarnings(context.Context, func (UnimplementedStartedServiceServer) GetStartedAt(context.Context, *emptypb.Empty) (*StartedAt, error) { return nil, status.Error(codes.Unimplemented, "method GetStartedAt not implemented") } + +func (UnimplementedStartedServiceServer) ListOutbounds(context.Context, *emptypb.Empty) (*OutboundList, error) { + return nil, status.Error(codes.Unimplemented, "method ListOutbounds not implemented") +} + +func (UnimplementedStartedServiceServer) SubscribeOutbounds(*emptypb.Empty, grpc.ServerStreamingServer[OutboundList]) error { + return status.Error(codes.Unimplemented, "method SubscribeOutbounds not implemented") +} + +func (UnimplementedStartedServiceServer) StartNetworkQualityTest(*NetworkQualityTestRequest, grpc.ServerStreamingServer[NetworkQualityTestProgress]) error { + return status.Error(codes.Unimplemented, "method StartNetworkQualityTest not implemented") +} + +func (UnimplementedStartedServiceServer) StartSTUNTest(*STUNTestRequest, grpc.ServerStreamingServer[STUNTestProgress]) error { + return status.Error(codes.Unimplemented, "method StartSTUNTest not implemented") +} func (UnimplementedStartedServiceServer) mustEmbedUnimplementedStartedServiceServer() {} func (UnimplementedStartedServiceServer) testEmbeddedByValue() {} @@ -882,6 +977,57 @@ func _StartedService_GetStartedAt_Handler(srv interface{}, ctx context.Context, return interceptor(ctx, in, info, handler) } +func _StartedService_ListOutbounds_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(emptypb.Empty) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(StartedServiceServer).ListOutbounds(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: StartedService_ListOutbounds_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(StartedServiceServer).ListOutbounds(ctx, req.(*emptypb.Empty)) + } + return interceptor(ctx, in, info, handler) +} + +func _StartedService_SubscribeOutbounds_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(emptypb.Empty) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(StartedServiceServer).SubscribeOutbounds(m, &grpc.GenericServerStream[emptypb.Empty, OutboundList]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StartedService_SubscribeOutboundsServer = grpc.ServerStreamingServer[OutboundList] + +func _StartedService_StartNetworkQualityTest_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(NetworkQualityTestRequest) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(StartedServiceServer).StartNetworkQualityTest(m, &grpc.GenericServerStream[NetworkQualityTestRequest, NetworkQualityTestProgress]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StartedService_StartNetworkQualityTestServer = grpc.ServerStreamingServer[NetworkQualityTestProgress] + +func _StartedService_StartSTUNTest_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(STUNTestRequest) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(StartedServiceServer).StartSTUNTest(m, &grpc.GenericServerStream[STUNTestRequest, STUNTestProgress]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type StartedService_StartSTUNTestServer = grpc.ServerStreamingServer[STUNTestProgress] + // StartedService_ServiceDesc is the grpc.ServiceDesc for StartedService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -957,6 +1103,10 @@ var StartedService_ServiceDesc = grpc.ServiceDesc{ MethodName: "GetStartedAt", Handler: _StartedService_GetStartedAt_Handler, }, + { + MethodName: "ListOutbounds", + Handler: _StartedService_ListOutbounds_Handler, + }, }, Streams: []grpc.StreamDesc{ { @@ -989,6 +1139,21 @@ var StartedService_ServiceDesc = grpc.ServiceDesc{ Handler: _StartedService_SubscribeConnections_Handler, ServerStreams: true, }, + { + StreamName: "SubscribeOutbounds", + Handler: _StartedService_SubscribeOutbounds_Handler, + ServerStreams: true, + }, + { + StreamName: "StartNetworkQualityTest", + Handler: _StartedService_StartNetworkQualityTest_Handler, + ServerStreams: true, + }, + { + StreamName: "StartSTUNTest", + Handler: _StartedService_StartSTUNTest_Handler, + ServerStreams: true, + }, }, Metadata: "daemon/started_service.proto", } diff --git a/experimental/libbox/command.go b/experimental/libbox/command.go index e3af6a196..8a43bc954 100644 --- a/experimental/libbox/command.go +++ b/experimental/libbox/command.go @@ -6,4 +6,5 @@ const ( CommandGroup CommandClashMode CommandConnections + CommandOutbounds ) diff --git a/experimental/libbox/command_client.go b/experimental/libbox/command_client.go index 159c1710f..d72c95d04 100644 --- a/experimental/libbox/command_client.go +++ b/experimental/libbox/command_client.go @@ -47,6 +47,7 @@ type CommandClientHandler interface { WriteLogs(messageList LogIterator) WriteStatus(message *StatusMessage) WriteGroups(message OutboundGroupIterator) + WriteOutbounds(message OutboundGroupItemIterator) InitializeClashMode(modeList StringIterator, currentMode string) UpdateClashMode(newMode string) WriteConnectionEvents(events *ConnectionEvents) @@ -243,6 +244,8 @@ func (c *CommandClient) dispatchCommands() error { go c.handleClashModeStream() case CommandConnections: go c.handleConnectionsStream() + case CommandOutbounds: + go c.handleOutboundsStream() default: return E.New("unknown command: ", command) } @@ -456,6 +459,25 @@ func (c *CommandClient) handleConnectionsStream() { } } +func (c *CommandClient) handleOutboundsStream() { + client, ctx := c.getStreamContext() + + stream, err := client.SubscribeOutbounds(ctx, &emptypb.Empty{}) + if err != nil { + c.handler.Disconnected(err.Error()) + return + } + + for { + list, err := stream.Recv() + if err != nil { + c.handler.Disconnected(err.Error()) + return + } + c.handler.WriteOutbounds(outboundGroupItemListFromGRPC(list)) + } +} + func (c *CommandClient) SelectOutbound(groupTag string, outboundTag string) error { _, err := callWithResult(c, func(client daemon.StartedServiceClient) (*emptypb.Empty, error) { return client.SelectOutbound(context.Background(), &daemon.SelectOutboundRequest{ @@ -603,3 +625,98 @@ func (c *CommandClient) SetGroupExpand(groupTag string, isExpand bool) error { }) return err } + +func (c *CommandClient) ListOutbounds() (OutboundGroupItemIterator, error) { + return callWithResult(c, func(client daemon.StartedServiceClient) (OutboundGroupItemIterator, error) { + list, err := client.ListOutbounds(context.Background(), &emptypb.Empty{}) + if err != nil { + return nil, err + } + return outboundGroupItemListFromGRPC(list), nil + }) +} + +func (c *CommandClient) StartNetworkQualityTest(configURL string, outboundTag string, serial bool, maxRuntimeSeconds int32, http3 bool, handler NetworkQualityTestHandler) error { + client, err := c.getClientForCall() + if err != nil { + return err + } + if c.standalone { + defer c.closeConnection() + } + stream, err := client.StartNetworkQualityTest(context.Background(), &daemon.NetworkQualityTestRequest{ + ConfigURL: configURL, + OutboundTag: outboundTag, + Serial: serial, + MaxRuntimeSeconds: maxRuntimeSeconds, + Http3: http3, + }) + if err != nil { + return err + } + for { + event, recvErr := stream.Recv() + if recvErr != nil { + handler.OnError(recvErr.Error()) + return recvErr + } + if event.IsFinal { + if event.Error != "" { + handler.OnError(event.Error) + } else { + handler.OnResult(&NetworkQualityResult{ + DownloadCapacity: event.DownloadCapacity, + UploadCapacity: event.UploadCapacity, + DownloadRPM: event.DownloadRPM, + UploadRPM: event.UploadRPM, + IdleLatencyMs: event.IdleLatencyMs, + DownloadCapacityAccuracy: event.DownloadCapacityAccuracy, + UploadCapacityAccuracy: event.UploadCapacityAccuracy, + DownloadRPMAccuracy: event.DownloadRPMAccuracy, + UploadRPMAccuracy: event.UploadRPMAccuracy, + }) + } + return nil + } + handler.OnProgress(networkQualityProgressFromGRPC(event)) + } +} + +func (c *CommandClient) StartSTUNTest(server string, outboundTag string, handler STUNTestHandler) error { + client, err := c.getClientForCall() + if err != nil { + return err + } + if c.standalone { + defer c.closeConnection() + } + stream, err := client.StartSTUNTest(context.Background(), &daemon.STUNTestRequest{ + Server: server, + OutboundTag: outboundTag, + }) + if err != nil { + return err + } + for { + event, recvErr := stream.Recv() + if recvErr != nil { + handler.OnError(recvErr.Error()) + return recvErr + } + if event.IsFinal { + if event.Error != "" { + handler.OnError(event.Error) + } else { + handler.OnResult(&STUNTestResult{ + ExternalAddr: event.ExternalAddr, + LatencyMs: event.LatencyMs, + NATMapping: event.NatMapping, + NATFiltering: event.NatFiltering, + NATTypeSupported: event.NatTypeSupported, + }) + } + return nil + } + handler.OnProgress(stunTestProgressFromGRPC(event)) + } +} diff --git a/experimental/libbox/command_types.go b/experimental/libbox/command_types.go index b811aaf46..89e5343bf 100644 --- a/experimental/libbox/command_types.go +++ b/experimental/libbox/command_types.go @@ -339,6 +339,22 @@ func outboundGroupIteratorFromGRPC(groups *daemon.Groups) OutboundGroupIterator return newIterator(libboxGroups) } +func outboundGroupItemListFromGRPC(list *daemon.OutboundList) OutboundGroupItemIterator { + if list == nil || len(list.Outbounds) == 0 { + return newIterator([]*OutboundGroupItem{}) + } + var items []*OutboundGroupItem + for _, ob := range list.Outbounds { + items = append(items, &OutboundGroupItem{ + Tag: ob.Tag, + Type: ob.Type, + URLTestTime: ob.UrlTestTime, + URLTestDelay: ob.UrlTestDelay, + }) + } + return newIterator(items) +} + func connectionFromGRPC(conn *daemon.Connection) Connection { var processInfo *ProcessInfo if conn.ProcessInfo != nil { diff --git a/experimental/libbox/command_types_nq.go b/experimental/libbox/command_types_nq.go new file mode 100644 index 000000000..fc8957e2e --- /dev/null +++ b/experimental/libbox/command_types_nq.go @@ -0,0 +1,51 @@ +package libbox + +import "github.com/sagernet/sing-box/daemon" + +type NetworkQualityProgress struct { + Phase int32 + DownloadCapacity int64 + UploadCapacity int64 + DownloadRPM int32 + UploadRPM int32 + IdleLatencyMs int32 + ElapsedMs int64 + DownloadCapacityAccuracy int32 + UploadCapacityAccuracy int32 + DownloadRPMAccuracy int32 + UploadRPMAccuracy int32 +} + +type NetworkQualityResult struct { + DownloadCapacity int64 + UploadCapacity int64 + DownloadRPM int32 + UploadRPM int32 + IdleLatencyMs int32 + DownloadCapacityAccuracy int32 + UploadCapacityAccuracy int32 + DownloadRPMAccuracy int32 + UploadRPMAccuracy int32 +} + +type NetworkQualityTestHandler interface { + OnProgress(progress *NetworkQualityProgress) + OnResult(result *NetworkQualityResult) + OnError(message string) +} + +func networkQualityProgressFromGRPC(event *daemon.NetworkQualityTestProgress) *NetworkQualityProgress { + return &NetworkQualityProgress{ + Phase: event.Phase, + DownloadCapacity: event.DownloadCapacity, + UploadCapacity: event.UploadCapacity, + DownloadRPM: event.DownloadRPM, + UploadRPM: event.UploadRPM, + IdleLatencyMs: event.IdleLatencyMs, + ElapsedMs: event.ElapsedMs, + DownloadCapacityAccuracy: event.DownloadCapacityAccuracy, + UploadCapacityAccuracy: event.UploadCapacityAccuracy, + DownloadRPMAccuracy: event.DownloadRPMAccuracy, + UploadRPMAccuracy: event.UploadRPMAccuracy, + } +} diff --git a/experimental/libbox/command_types_stun.go b/experimental/libbox/command_types_stun.go new file mode 100644 index 000000000..22846c327 --- /dev/null +++ b/experimental/libbox/command_types_stun.go @@ -0,0 +1,35 @@ +package libbox + +import "github.com/sagernet/sing-box/daemon" + +type STUNTestProgress struct { + Phase int32 + ExternalAddr string + LatencyMs int32 + NATMapping int32 + NATFiltering int32 +} + +type STUNTestResult struct { + ExternalAddr string + LatencyMs int32 + NATMapping int32 + NATFiltering int32 + NATTypeSupported bool +} + +type STUNTestHandler interface { + OnProgress(progress *STUNTestProgress) + OnResult(result *STUNTestResult) + OnError(message string) +} + +func stunTestProgressFromGRPC(event *daemon.STUNTestProgress) *STUNTestProgress { + return &STUNTestProgress{ + Phase: event.Phase, + ExternalAddr: event.ExternalAddr, + LatencyMs: event.LatencyMs, + NATMapping: event.NatMapping, + NATFiltering: event.NatFiltering, + } +} diff --git a/experimental/libbox/networkquality.go b/experimental/libbox/networkquality.go new file mode 100644 index 000000000..fcbe6f3a6 --- /dev/null +++ b/experimental/libbox/networkquality.go @@ -0,0 +1,74 @@ +package libbox + +import ( + "context" + "time" + + "github.com/sagernet/sing-box/common/networkquality" +) + +type NetworkQualityTest struct { + ctx context.Context + cancel context.CancelFunc +} + +func NewNetworkQualityTest() *NetworkQualityTest { + ctx, cancel := context.WithCancel(context.Background()) + return &NetworkQualityTest{ctx: ctx, cancel: cancel} +} + +func (t *NetworkQualityTest) Start(configURL string, serial bool, maxRuntimeSeconds int32, http3 bool, handler NetworkQualityTestHandler) { + go func() { + httpClient := networkquality.NewHTTPClient(nil) + defer httpClient.CloseIdleConnections() + + measurementClientFactory, err := networkquality.NewOptionalHTTP3Factory(nil, http3) + if err != nil { + handler.OnError(err.Error()) + return + } + + result, err := networkquality.Run(networkquality.Options{ + ConfigURL: configURL, + HTTPClient: httpClient, + NewMeasurementClient: measurementClientFactory, + Serial: serial, + MaxRuntime: time.Duration(maxRuntimeSeconds) * time.Second, + Context: t.ctx, + OnProgress: func(p networkquality.Progress) { + handler.OnProgress(&NetworkQualityProgress{ + Phase: int32(p.Phase), + DownloadCapacity: p.DownloadCapacity, + UploadCapacity: p.UploadCapacity, + DownloadRPM: p.DownloadRPM, + UploadRPM: p.UploadRPM, + IdleLatencyMs: p.IdleLatencyMs, + ElapsedMs: p.ElapsedMs, + DownloadCapacityAccuracy: int32(p.DownloadCapacityAccuracy), + UploadCapacityAccuracy: int32(p.UploadCapacityAccuracy), + DownloadRPMAccuracy: int32(p.DownloadRPMAccuracy), + UploadRPMAccuracy: int32(p.UploadRPMAccuracy), + }) + }, + }) + if err != nil { + handler.OnError(err.Error()) + return + } + handler.OnResult(&NetworkQualityResult{ + DownloadCapacity: result.DownloadCapacity, + UploadCapacity: result.UploadCapacity, + DownloadRPM: result.DownloadRPM, + UploadRPM: result.UploadRPM, + IdleLatencyMs: result.IdleLatencyMs, + DownloadCapacityAccuracy: int32(result.DownloadCapacityAccuracy), + UploadCapacityAccuracy: int32(result.UploadCapacityAccuracy), + DownloadRPMAccuracy: int32(result.DownloadRPMAccuracy), + UploadRPMAccuracy: int32(result.UploadRPMAccuracy), + }) + }() +} + +func (t *NetworkQualityTest) Cancel() { + t.cancel() +} diff --git a/experimental/libbox/setup.go b/experimental/libbox/setup.go index 01a454044..ac706e9db 100644 --- a/experimental/libbox/setup.go +++ b/experimental/libbox/setup.go @@ -9,6 +9,8 @@ import ( "strings" "time" + "github.com/sagernet/sing-box/common/networkquality" + "github.com/sagernet/sing-box/common/stun" C "github.com/sagernet/sing-box/constant" "github.com/sagernet/sing-box/experimental/locale" "github.com/sagernet/sing-box/log" @@ -129,6 +131,56 @@ func FormatDuration(duration int64) string { return log.FormatDuration(time.Duration(duration) * time.Millisecond) } +func FormatBitrate(bps int64) string { + return networkquality.FormatBitrate(bps) +} + +const NetworkQualityDefaultConfigURL = networkquality.DefaultConfigURL + +const NetworkQualityDefaultMaxRuntimeSeconds = int32(networkquality.DefaultMaxRuntime / time.Second) + +const ( + NetworkQualityAccuracyLow = int32(networkquality.AccuracyLow) + NetworkQualityAccuracyMedium = int32(networkquality.AccuracyMedium) + NetworkQualityAccuracyHigh = int32(networkquality.AccuracyHigh) +) + +const ( + NetworkQualityPhaseIdle = int32(networkquality.PhaseIdle) + NetworkQualityPhaseDownload = int32(networkquality.PhaseDownload) + NetworkQualityPhaseUpload = int32(networkquality.PhaseUpload) + NetworkQualityPhaseDone = int32(networkquality.PhaseDone) +) + +const STUNDefaultServer = stun.DefaultServer + +const ( + STUNPhaseBinding = int32(stun.PhaseBinding) + STUNPhaseNATMapping = int32(stun.PhaseNATMapping) + STUNPhaseNATFiltering = int32(stun.PhaseNATFiltering) + STUNPhaseDone = int32(stun.PhaseDone) +) + +const ( + NATMappingEndpointIndependent = int32(stun.NATMappingEndpointIndependent) + NATMappingAddressDependent = int32(stun.NATMappingAddressDependent) + NATMappingAddressAndPortDependent = int32(stun.NATMappingAddressAndPortDependent) +) + +const ( + NATFilteringEndpointIndependent = int32(stun.NATFilteringEndpointIndependent) + NATFilteringAddressDependent = int32(stun.NATFilteringAddressDependent) + NATFilteringAddressAndPortDependent = int32(stun.NATFilteringAddressAndPortDependent) +) + +func FormatNATMapping(value int32) string { + return stun.NATMapping(value).String() +} + +func FormatNATFiltering(value int32) string { + return stun.NATFiltering(value).String() +} + func ProxyDisplayType(proxyType string) string { return C.ProxyDisplayName(proxyType) } diff --git a/experimental/libbox/stun.go b/experimental/libbox/stun.go new file mode 100644 index 000000000..3f38815d7 --- /dev/null +++ b/experimental/libbox/stun.go @@ -0,0 +1,50 @@ +package libbox + +import ( + "context" + + "github.com/sagernet/sing-box/common/stun" +) + +type STUNTest struct { + ctx context.Context + cancel context.CancelFunc +} + +func NewSTUNTest() *STUNTest { + ctx, cancel := context.WithCancel(context.Background()) + return &STUNTest{ctx: ctx, cancel: cancel} +} + +func (t *STUNTest) Start(server string, handler STUNTestHandler) { + go func() { + result, err := stun.Run(stun.Options{ + Server: server, + Context: t.ctx, + OnProgress: func(p stun.Progress) { + handler.OnProgress(&STUNTestProgress{ + Phase: int32(p.Phase), + ExternalAddr: p.ExternalAddr, + LatencyMs: p.LatencyMs, + NATMapping: int32(p.NATMapping), + NATFiltering: int32(p.NATFiltering), + }) + }, + }) + if err != nil { + handler.OnError(err.Error()) + return + } + handler.OnResult(&STUNTestResult{ + ExternalAddr: result.ExternalAddr, + LatencyMs: result.LatencyMs, + NATMapping: int32(result.NATMapping), + NATFiltering: int32(result.NATFiltering), + NATTypeSupported: result.NATTypeSupported, + }) + }() +} + +func (t *STUNTest) Cancel() { + t.cancel() +}