mirror of
https://github.com/SagerNet/sing-box.git
synced 2026-04-13 20:28:32 +10:00
856 lines
24 KiB
Go
856 lines
24 KiB
Go
package daemon
|
|
|
|
import (
|
|
"context"
|
|
"os"
|
|
"runtime"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/sagernet/sing-box/adapter"
|
|
"github.com/sagernet/sing-box/common/conntrack"
|
|
"github.com/sagernet/sing-box/common/urltest"
|
|
"github.com/sagernet/sing-box/experimental/clashapi"
|
|
"github.com/sagernet/sing-box/experimental/clashapi/trafficontrol"
|
|
"github.com/sagernet/sing-box/experimental/deprecated"
|
|
"github.com/sagernet/sing-box/log"
|
|
"github.com/sagernet/sing-box/protocol/group"
|
|
"github.com/sagernet/sing/common"
|
|
"github.com/sagernet/sing/common/batch"
|
|
E "github.com/sagernet/sing/common/exceptions"
|
|
"github.com/sagernet/sing/common/memory"
|
|
"github.com/sagernet/sing/common/observable"
|
|
"github.com/sagernet/sing/common/x/list"
|
|
"github.com/sagernet/sing/service"
|
|
|
|
"github.com/gofrs/uuid/v5"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/protobuf/types/known/emptypb"
|
|
)
|
|
|
|
var _ StartedServiceServer = (*StartedService)(nil)
|
|
|
|
type StartedService struct {
|
|
ctx context.Context
|
|
// platform adapter.PlatformInterface
|
|
handler PlatformHandler
|
|
debug bool
|
|
logMaxLines int
|
|
// workingDirectory string
|
|
// tempDirectory string
|
|
// userID int
|
|
// groupID int
|
|
// systemProxyEnabled bool
|
|
serviceAccess sync.RWMutex
|
|
serviceStatus *ServiceStatus
|
|
serviceStatusSubscriber *observable.Subscriber[*ServiceStatus]
|
|
serviceStatusObserver *observable.Observer[*ServiceStatus]
|
|
logAccess sync.RWMutex
|
|
logLines list.List[*log.Entry]
|
|
logSubscriber *observable.Subscriber[*log.Entry]
|
|
logObserver *observable.Observer[*log.Entry]
|
|
instance *Instance
|
|
startedAt time.Time
|
|
urlTestSubscriber *observable.Subscriber[struct{}]
|
|
urlTestObserver *observable.Observer[struct{}]
|
|
urlTestHistoryStorage *urltest.HistoryStorage
|
|
clashModeSubscriber *observable.Subscriber[struct{}]
|
|
clashModeObserver *observable.Observer[struct{}]
|
|
}
|
|
|
|
type ServiceOptions struct {
|
|
Context context.Context
|
|
// Platform adapter.PlatformInterface
|
|
Handler PlatformHandler
|
|
Debug bool
|
|
LogMaxLines int
|
|
// WorkingDirectory string
|
|
// TempDirectory string
|
|
// UserID int
|
|
// GroupID int
|
|
// SystemProxyEnabled bool
|
|
}
|
|
|
|
func NewStartedService(options ServiceOptions) *StartedService {
|
|
s := &StartedService{
|
|
ctx: options.Context,
|
|
// platform: options.Platform,
|
|
handler: options.Handler,
|
|
debug: options.Debug,
|
|
logMaxLines: options.LogMaxLines,
|
|
// workingDirectory: options.WorkingDirectory,
|
|
// tempDirectory: options.TempDirectory,
|
|
// userID: options.UserID,
|
|
// groupID: options.GroupID,
|
|
// systemProxyEnabled: options.SystemProxyEnabled,
|
|
serviceStatus: &ServiceStatus{Status: ServiceStatus_IDLE},
|
|
serviceStatusSubscriber: observable.NewSubscriber[*ServiceStatus](4),
|
|
logSubscriber: observable.NewSubscriber[*log.Entry](128),
|
|
urlTestSubscriber: observable.NewSubscriber[struct{}](1),
|
|
urlTestHistoryStorage: urltest.NewHistoryStorage(),
|
|
clashModeSubscriber: observable.NewSubscriber[struct{}](1),
|
|
}
|
|
s.serviceStatusObserver = observable.NewObserver(s.serviceStatusSubscriber, 2)
|
|
s.logObserver = observable.NewObserver(s.logSubscriber, 64)
|
|
s.urlTestObserver = observable.NewObserver(s.urlTestSubscriber, 1)
|
|
s.clashModeObserver = observable.NewObserver(s.clashModeSubscriber, 1)
|
|
return s
|
|
}
|
|
|
|
func (s *StartedService) resetLogs() {
|
|
s.logAccess.Lock()
|
|
s.logLines = list.List[*log.Entry]{}
|
|
s.logAccess.Unlock()
|
|
s.logSubscriber.Emit(nil)
|
|
}
|
|
|
|
func (s *StartedService) updateStatus(newStatus ServiceStatus_Type) {
|
|
statusObject := &ServiceStatus{Status: newStatus}
|
|
s.serviceStatusSubscriber.Emit(statusObject)
|
|
s.serviceStatus = statusObject
|
|
}
|
|
|
|
func (s *StartedService) updateStatusError(err error) error {
|
|
statusObject := &ServiceStatus{Status: ServiceStatus_FATAL, ErrorMessage: err.Error()}
|
|
s.serviceStatusSubscriber.Emit(statusObject)
|
|
s.serviceStatus = statusObject
|
|
s.serviceAccess.Unlock()
|
|
return err
|
|
}
|
|
|
|
func (s *StartedService) waitForStarted(ctx context.Context) error {
|
|
s.serviceAccess.RLock()
|
|
currentStatus := s.serviceStatus.Status
|
|
s.serviceAccess.RUnlock()
|
|
|
|
switch currentStatus {
|
|
case ServiceStatus_STARTED:
|
|
return nil
|
|
case ServiceStatus_STARTING:
|
|
default:
|
|
return os.ErrInvalid
|
|
}
|
|
|
|
subscription, done, err := s.serviceStatusObserver.Subscribe()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer s.serviceStatusObserver.UnSubscribe(subscription)
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-s.ctx.Done():
|
|
return s.ctx.Err()
|
|
case status := <-subscription:
|
|
switch status.Status {
|
|
case ServiceStatus_STARTED:
|
|
return nil
|
|
case ServiceStatus_FATAL:
|
|
return E.New(status.ErrorMessage)
|
|
case ServiceStatus_IDLE, ServiceStatus_STOPPING:
|
|
return os.ErrInvalid
|
|
}
|
|
case <-done:
|
|
return os.ErrClosed
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *StartedService) StartOrReloadService(profileContent string, options *OverrideOptions) error {
|
|
s.serviceAccess.Lock()
|
|
switch s.serviceStatus.Status {
|
|
case ServiceStatus_IDLE, ServiceStatus_STARTED, ServiceStatus_STARTING:
|
|
default:
|
|
s.serviceAccess.Unlock()
|
|
return os.ErrInvalid
|
|
}
|
|
oldInstance := s.instance
|
|
if oldInstance != nil {
|
|
s.updateStatus(ServiceStatus_STOPPING)
|
|
s.serviceAccess.Unlock()
|
|
_ = oldInstance.Close()
|
|
s.serviceAccess.Lock()
|
|
}
|
|
s.updateStatus(ServiceStatus_STARTING)
|
|
s.resetLogs()
|
|
instance, err := s.newInstance(profileContent, options)
|
|
if err != nil {
|
|
return s.updateStatusError(err)
|
|
}
|
|
s.instance = instance
|
|
instance.urlTestHistoryStorage.SetHook(s.urlTestSubscriber)
|
|
if instance.clashServer != nil {
|
|
instance.clashServer.SetModeUpdateHook(s.clashModeSubscriber)
|
|
}
|
|
s.serviceAccess.Unlock()
|
|
err = instance.Start()
|
|
s.serviceAccess.Lock()
|
|
if s.serviceStatus.Status != ServiceStatus_STARTING {
|
|
s.serviceAccess.Unlock()
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return s.updateStatusError(err)
|
|
}
|
|
s.startedAt = time.Now()
|
|
s.updateStatus(ServiceStatus_STARTED)
|
|
s.serviceAccess.Unlock()
|
|
runtime.GC()
|
|
return nil
|
|
}
|
|
|
|
func (s *StartedService) CloseService() error {
|
|
s.serviceAccess.Lock()
|
|
switch s.serviceStatus.Status {
|
|
case ServiceStatus_STARTING, ServiceStatus_STARTED:
|
|
default:
|
|
s.serviceAccess.Unlock()
|
|
return os.ErrInvalid
|
|
}
|
|
s.updateStatus(ServiceStatus_STOPPING)
|
|
if s.instance != nil {
|
|
err := s.instance.Close()
|
|
if err != nil {
|
|
return s.updateStatusError(err)
|
|
}
|
|
}
|
|
s.instance = nil
|
|
s.startedAt = time.Time{}
|
|
s.updateStatus(ServiceStatus_IDLE)
|
|
s.serviceAccess.Unlock()
|
|
runtime.GC()
|
|
return nil
|
|
}
|
|
|
|
func (s *StartedService) SetError(err error) {
|
|
s.serviceAccess.Lock()
|
|
s.updateStatusError(err)
|
|
s.WriteMessage(log.LevelError, err.Error())
|
|
}
|
|
|
|
func (s *StartedService) StopService(ctx context.Context, empty *emptypb.Empty) (*emptypb.Empty, error) {
|
|
err := s.handler.ServiceStop()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &emptypb.Empty{}, nil
|
|
}
|
|
|
|
func (s *StartedService) ReloadService(ctx context.Context, empty *emptypb.Empty) (*emptypb.Empty, error) {
|
|
err := s.handler.ServiceReload()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &emptypb.Empty{}, nil
|
|
}
|
|
|
|
func (s *StartedService) SubscribeServiceStatus(empty *emptypb.Empty, server grpc.ServerStreamingServer[ServiceStatus]) error {
|
|
subscription, done, err := s.serviceStatusObserver.Subscribe()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer s.serviceStatusObserver.UnSubscribe(subscription)
|
|
err = server.Send(s.serviceStatus)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for {
|
|
select {
|
|
case <-s.ctx.Done():
|
|
return s.ctx.Err()
|
|
case <-server.Context().Done():
|
|
return server.Context().Err()
|
|
case newStatus := <-subscription:
|
|
err = server.Send(newStatus)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
case <-done:
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *StartedService) SubscribeLog(empty *emptypb.Empty, server grpc.ServerStreamingServer[Log]) error {
|
|
var savedLines []*log.Entry
|
|
s.logAccess.Lock()
|
|
savedLines = make([]*log.Entry, 0, s.logLines.Len())
|
|
for element := s.logLines.Front(); element != nil; element = element.Next() {
|
|
savedLines = append(savedLines, element.Value)
|
|
}
|
|
subscription, done, err := s.logObserver.Subscribe()
|
|
s.logAccess.Unlock()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer s.logObserver.UnSubscribe(subscription)
|
|
err = server.Send(&Log{
|
|
Messages: common.Map(savedLines, func(it *log.Entry) *Log_Message {
|
|
return &Log_Message{
|
|
Level: LogLevel(it.Level),
|
|
Message: it.Message,
|
|
}
|
|
}),
|
|
Reset_: true,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for {
|
|
select {
|
|
case <-s.ctx.Done():
|
|
return s.ctx.Err()
|
|
case <-server.Context().Done():
|
|
return server.Context().Err()
|
|
case message := <-subscription:
|
|
var rawMessage Log
|
|
if message == nil {
|
|
rawMessage.Reset_ = true
|
|
} else {
|
|
rawMessage.Messages = append(rawMessage.Messages, &Log_Message{
|
|
Level: LogLevel(message.Level),
|
|
Message: message.Message,
|
|
})
|
|
}
|
|
fetch:
|
|
for {
|
|
select {
|
|
case message = <-subscription:
|
|
if message == nil {
|
|
rawMessage.Messages = nil
|
|
rawMessage.Reset_ = true
|
|
} else {
|
|
rawMessage.Messages = append(rawMessage.Messages, &Log_Message{
|
|
Level: LogLevel(message.Level),
|
|
Message: message.Message,
|
|
})
|
|
}
|
|
default:
|
|
break fetch
|
|
}
|
|
}
|
|
err = server.Send(&rawMessage)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
case <-done:
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *StartedService) GetDefaultLogLevel(ctx context.Context, empty *emptypb.Empty) (*DefaultLogLevel, error) {
|
|
s.serviceAccess.RLock()
|
|
switch s.serviceStatus.Status {
|
|
case ServiceStatus_STARTING, ServiceStatus_STARTED:
|
|
default:
|
|
s.serviceAccess.RUnlock()
|
|
return nil, os.ErrInvalid
|
|
}
|
|
logLevel := s.instance.instance.LogFactory().Level()
|
|
s.serviceAccess.RUnlock()
|
|
return &DefaultLogLevel{Level: LogLevel(logLevel)}, nil
|
|
}
|
|
|
|
func (s *StartedService) ClearLogs(ctx context.Context, empty *emptypb.Empty) (*emptypb.Empty, error) {
|
|
s.resetLogs()
|
|
return &emptypb.Empty{}, nil
|
|
}
|
|
|
|
func (s *StartedService) SubscribeStatus(request *SubscribeStatusRequest, server grpc.ServerStreamingServer[Status]) error {
|
|
interval := time.Duration(request.Interval)
|
|
if interval <= 0 {
|
|
interval = time.Second // Default to 1 second
|
|
}
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
status := s.readStatus()
|
|
uploadTotal := status.UplinkTotal
|
|
downloadTotal := status.DownlinkTotal
|
|
for {
|
|
err := server.Send(status)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
select {
|
|
case <-s.ctx.Done():
|
|
return s.ctx.Err()
|
|
case <-server.Context().Done():
|
|
return server.Context().Err()
|
|
case <-ticker.C:
|
|
}
|
|
status = s.readStatus()
|
|
upload := status.UplinkTotal - uploadTotal
|
|
download := status.DownlinkTotal - downloadTotal
|
|
uploadTotal = status.UplinkTotal
|
|
downloadTotal = status.DownlinkTotal
|
|
status.Uplink = upload
|
|
status.Downlink = download
|
|
}
|
|
}
|
|
|
|
func (s *StartedService) readStatus() *Status {
|
|
var status Status
|
|
status.Memory = memory.Inuse()
|
|
status.Goroutines = int32(runtime.NumGoroutine())
|
|
status.ConnectionsOut = int32(conntrack.Count())
|
|
s.serviceAccess.RLock()
|
|
nowService := s.instance
|
|
s.serviceAccess.RUnlock()
|
|
if nowService != nil {
|
|
if clashServer := nowService.clashServer; clashServer != nil {
|
|
status.TrafficAvailable = true
|
|
trafficManager := clashServer.(*clashapi.Server).TrafficManager()
|
|
status.UplinkTotal, status.DownlinkTotal = trafficManager.Total()
|
|
status.ConnectionsIn = int32(trafficManager.ConnectionsLen())
|
|
}
|
|
}
|
|
return &status
|
|
}
|
|
|
|
func (s *StartedService) SubscribeGroups(empty *emptypb.Empty, server grpc.ServerStreamingServer[Groups]) 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
|
|
}
|
|
groups := s.readGroups()
|
|
s.serviceAccess.RUnlock()
|
|
err = server.Send(groups)
|
|
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 (s *StartedService) readGroups() *Groups {
|
|
historyStorage := s.instance.urlTestHistoryStorage
|
|
boxService := s.instance
|
|
outbounds := boxService.instance.Outbound().Outbounds()
|
|
var iGroups []adapter.OutboundGroup
|
|
for _, it := range outbounds {
|
|
if group, isGroup := it.(adapter.OutboundGroup); isGroup {
|
|
iGroups = append(iGroups, group)
|
|
}
|
|
}
|
|
var gs Groups
|
|
for _, iGroup := range iGroups {
|
|
var g Group
|
|
g.Tag = iGroup.Tag()
|
|
g.Type = iGroup.Type()
|
|
_, g.Selectable = iGroup.(*group.Selector)
|
|
g.Selected = iGroup.Now()
|
|
if boxService.cacheFile != nil {
|
|
if isExpand, loaded := boxService.cacheFile.LoadGroupExpand(g.Tag); loaded {
|
|
g.IsExpand = isExpand
|
|
}
|
|
}
|
|
|
|
for _, itemTag := range iGroup.All() {
|
|
itemOutbound, isLoaded := boxService.instance.Outbound().Outbound(itemTag)
|
|
if !isLoaded {
|
|
continue
|
|
}
|
|
|
|
var item GroupItem
|
|
item.Tag = itemTag
|
|
item.Type = itemOutbound.Type()
|
|
if history := historyStorage.LoadURLTestHistory(adapter.OutboundTag(itemOutbound)); history != nil {
|
|
item.UrlTestTime = history.Time.Unix()
|
|
item.UrlTestDelay = int32(history.Delay)
|
|
}
|
|
g.Items = append(g.Items, &item)
|
|
}
|
|
if len(g.Items) < 2 {
|
|
continue
|
|
}
|
|
gs.Group = append(gs.Group, &g)
|
|
}
|
|
return &gs
|
|
}
|
|
|
|
func (s *StartedService) GetClashModeStatus(ctx context.Context, empty *emptypb.Empty) (*ClashModeStatus, error) {
|
|
s.serviceAccess.RLock()
|
|
if s.serviceStatus.Status != ServiceStatus_STARTED {
|
|
s.serviceAccess.RUnlock()
|
|
return nil, os.ErrInvalid
|
|
}
|
|
clashServer := s.instance.clashServer
|
|
s.serviceAccess.RUnlock()
|
|
if clashServer == nil {
|
|
return nil, os.ErrInvalid
|
|
}
|
|
return &ClashModeStatus{
|
|
ModeList: clashServer.ModeList(),
|
|
CurrentMode: clashServer.Mode(),
|
|
}, nil
|
|
}
|
|
|
|
func (s *StartedService) SubscribeClashMode(empty *emptypb.Empty, server grpc.ServerStreamingServer[ClashMode]) error {
|
|
err := s.waitForStarted(server.Context())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
subscription, done, err := s.clashModeObserver.Subscribe()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer s.clashModeObserver.UnSubscribe(subscription)
|
|
for {
|
|
s.serviceAccess.RLock()
|
|
if s.serviceStatus.Status != ServiceStatus_STARTED {
|
|
s.serviceAccess.RUnlock()
|
|
return os.ErrInvalid
|
|
}
|
|
message := &ClashMode{Mode: s.instance.clashServer.Mode()}
|
|
s.serviceAccess.RUnlock()
|
|
err = server.Send(message)
|
|
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 (s *StartedService) SetClashMode(ctx context.Context, request *ClashMode) (*emptypb.Empty, error) {
|
|
s.serviceAccess.RLock()
|
|
if s.serviceStatus.Status != ServiceStatus_STARTED {
|
|
s.serviceAccess.RUnlock()
|
|
return nil, os.ErrInvalid
|
|
}
|
|
clashServer := s.instance.clashServer
|
|
s.serviceAccess.RUnlock()
|
|
clashServer.(*clashapi.Server).SetMode(request.Mode)
|
|
return &emptypb.Empty{}, nil
|
|
}
|
|
|
|
func (s *StartedService) URLTest(ctx context.Context, request *URLTestRequest) (*emptypb.Empty, error) {
|
|
s.serviceAccess.RLock()
|
|
if s.serviceStatus.Status != ServiceStatus_STARTED {
|
|
s.serviceAccess.RUnlock()
|
|
return nil, os.ErrInvalid
|
|
}
|
|
boxService := s.instance
|
|
s.serviceAccess.RUnlock()
|
|
groupTag := request.OutboundTag
|
|
abstractOutboundGroup, isLoaded := boxService.instance.Outbound().Outbound(groupTag)
|
|
if !isLoaded {
|
|
return nil, E.New("outbound group not found: ", groupTag)
|
|
}
|
|
outboundGroup, isOutboundGroup := abstractOutboundGroup.(adapter.OutboundGroup)
|
|
if !isOutboundGroup {
|
|
return nil, E.New("outbound is not a group: ", groupTag)
|
|
}
|
|
urlTest, isURLTest := abstractOutboundGroup.(*group.URLTest)
|
|
if isURLTest {
|
|
go urlTest.CheckOutbounds()
|
|
} else {
|
|
historyStorage := boxService.urlTestHistoryStorage
|
|
|
|
outbounds := common.Filter(common.Map(outboundGroup.All(), func(it string) adapter.Outbound {
|
|
itOutbound, _ := boxService.instance.Outbound().Outbound(it)
|
|
return itOutbound
|
|
}), func(it adapter.Outbound) bool {
|
|
if it == nil {
|
|
return false
|
|
}
|
|
_, isGroup := it.(adapter.OutboundGroup)
|
|
if isGroup {
|
|
return false
|
|
}
|
|
return true
|
|
})
|
|
b, _ := batch.New(boxService.ctx, batch.WithConcurrencyNum[any](10))
|
|
for _, detour := range outbounds {
|
|
outboundToTest := detour
|
|
outboundTag := outboundToTest.Tag()
|
|
b.Go(outboundTag, func() (any, error) {
|
|
t, err := urltest.URLTest(boxService.ctx, "", outboundToTest)
|
|
if err != nil {
|
|
historyStorage.DeleteURLTestHistory(outboundTag)
|
|
} else {
|
|
historyStorage.StoreURLTestHistory(outboundTag, &adapter.URLTestHistory{
|
|
Time: time.Now(),
|
|
Delay: t,
|
|
})
|
|
}
|
|
return nil, nil
|
|
})
|
|
}
|
|
}
|
|
return &emptypb.Empty{}, nil
|
|
}
|
|
|
|
func (s *StartedService) SelectOutbound(ctx context.Context, request *SelectOutboundRequest) (*emptypb.Empty, error) {
|
|
s.serviceAccess.RLock()
|
|
switch s.serviceStatus.Status {
|
|
case ServiceStatus_STARTING, ServiceStatus_STARTED:
|
|
default:
|
|
s.serviceAccess.RUnlock()
|
|
return nil, os.ErrInvalid
|
|
}
|
|
boxService := s.instance.instance
|
|
s.serviceAccess.RUnlock()
|
|
outboundGroup, isLoaded := boxService.Outbound().Outbound(request.GroupTag)
|
|
if !isLoaded {
|
|
return nil, E.New("selector not found: ", request.GroupTag)
|
|
}
|
|
selector, isSelector := outboundGroup.(*group.Selector)
|
|
if !isSelector {
|
|
return nil, E.New("outbound is not a selector: ", request.GroupTag)
|
|
}
|
|
if !selector.SelectOutbound(request.OutboundTag) {
|
|
return nil, E.New("outbound not found in selector: ", request.OutboundTag)
|
|
}
|
|
s.urlTestObserver.Emit(struct{}{})
|
|
return &emptypb.Empty{}, nil
|
|
}
|
|
|
|
func (s *StartedService) SetGroupExpand(ctx context.Context, request *SetGroupExpandRequest) (*emptypb.Empty, error) {
|
|
s.serviceAccess.RLock()
|
|
switch s.serviceStatus.Status {
|
|
case ServiceStatus_STARTING, ServiceStatus_STARTED:
|
|
default:
|
|
s.serviceAccess.RUnlock()
|
|
return nil, os.ErrInvalid
|
|
}
|
|
boxService := s.instance
|
|
s.serviceAccess.RUnlock()
|
|
if boxService.cacheFile != nil {
|
|
err := boxService.cacheFile.StoreGroupExpand(request.GroupTag, request.IsExpand)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return &emptypb.Empty{}, nil
|
|
}
|
|
|
|
func (s *StartedService) GetSystemProxyStatus(ctx context.Context, empty *emptypb.Empty) (*SystemProxyStatus, error) {
|
|
return s.handler.SystemProxyStatus()
|
|
}
|
|
|
|
func (s *StartedService) SetSystemProxyEnabled(ctx context.Context, request *SetSystemProxyEnabledRequest) (*emptypb.Empty, error) {
|
|
err := s.handler.SetSystemProxyEnabled(request.Enabled)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
func (s *StartedService) SubscribeConnections(request *SubscribeConnectionsRequest, server grpc.ServerStreamingServer[Connections]) error {
|
|
err := s.waitForStarted(server.Context())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.serviceAccess.RLock()
|
|
boxService := s.instance
|
|
s.serviceAccess.RUnlock()
|
|
ticker := time.NewTicker(time.Duration(request.Interval))
|
|
defer ticker.Stop()
|
|
trafficManager := boxService.clashServer.(*clashapi.Server).TrafficManager()
|
|
var (
|
|
connections = make(map[uuid.UUID]*Connection)
|
|
outConnections []*Connection
|
|
)
|
|
for {
|
|
outConnections = outConnections[:0]
|
|
for _, connection := range trafficManager.Connections() {
|
|
outConnections = append(outConnections, newConnection(connections, connection, false))
|
|
}
|
|
for _, connection := range trafficManager.ClosedConnections() {
|
|
outConnections = append(outConnections, newConnection(connections, connection, true))
|
|
}
|
|
err := server.Send(&Connections{Connections: outConnections})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
select {
|
|
case <-s.ctx.Done():
|
|
return s.ctx.Err()
|
|
case <-server.Context().Done():
|
|
return server.Context().Err()
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func newConnection(connections map[uuid.UUID]*Connection, metadata trafficontrol.TrackerMetadata, isClosed bool) *Connection {
|
|
if oldConnection, loaded := connections[metadata.ID]; loaded {
|
|
if isClosed {
|
|
if oldConnection.ClosedAt == 0 {
|
|
oldConnection.Uplink = 0
|
|
oldConnection.Downlink = 0
|
|
oldConnection.ClosedAt = metadata.ClosedAt.UnixMilli()
|
|
}
|
|
return oldConnection
|
|
}
|
|
lastUplink := oldConnection.UplinkTotal
|
|
lastDownlink := oldConnection.DownlinkTotal
|
|
uplinkTotal := metadata.Upload.Load()
|
|
downlinkTotal := metadata.Download.Load()
|
|
oldConnection.Uplink = uplinkTotal - lastUplink
|
|
oldConnection.Downlink = downlinkTotal - lastDownlink
|
|
oldConnection.UplinkTotal = uplinkTotal
|
|
oldConnection.DownlinkTotal = downlinkTotal
|
|
return oldConnection
|
|
}
|
|
var rule string
|
|
if metadata.Rule != nil {
|
|
rule = metadata.Rule.String()
|
|
}
|
|
uplinkTotal := metadata.Upload.Load()
|
|
downlinkTotal := metadata.Download.Load()
|
|
uplink := uplinkTotal
|
|
downlink := downlinkTotal
|
|
var closedAt int64
|
|
if !metadata.ClosedAt.IsZero() {
|
|
closedAt = metadata.ClosedAt.UnixMilli()
|
|
uplink = 0
|
|
downlink = 0
|
|
}
|
|
var processInfo *ProcessInfo
|
|
if metadata.Metadata.ProcessInfo != nil {
|
|
processInfo = &ProcessInfo{
|
|
ProcessId: metadata.Metadata.ProcessInfo.ProcessID,
|
|
UserId: metadata.Metadata.ProcessInfo.UserId,
|
|
UserName: metadata.Metadata.ProcessInfo.UserName,
|
|
ProcessPath: metadata.Metadata.ProcessInfo.ProcessPath,
|
|
PackageName: metadata.Metadata.ProcessInfo.AndroidPackageName,
|
|
}
|
|
}
|
|
connection := &Connection{
|
|
Id: metadata.ID.String(),
|
|
Inbound: metadata.Metadata.Inbound,
|
|
InboundType: metadata.Metadata.InboundType,
|
|
IpVersion: int32(metadata.Metadata.IPVersion),
|
|
Network: metadata.Metadata.Network,
|
|
Source: metadata.Metadata.Source.String(),
|
|
Destination: metadata.Metadata.Destination.String(),
|
|
Domain: metadata.Metadata.Domain,
|
|
Protocol: metadata.Metadata.Protocol,
|
|
User: metadata.Metadata.User,
|
|
FromOutbound: metadata.Metadata.Outbound,
|
|
CreatedAt: metadata.CreatedAt.UnixMilli(),
|
|
ClosedAt: closedAt,
|
|
Uplink: uplink,
|
|
Downlink: downlink,
|
|
UplinkTotal: uplinkTotal,
|
|
DownlinkTotal: downlinkTotal,
|
|
Rule: rule,
|
|
Outbound: metadata.Outbound,
|
|
OutboundType: metadata.OutboundType,
|
|
ChainList: metadata.Chain,
|
|
ProcessInfo: processInfo,
|
|
}
|
|
connections[metadata.ID] = connection
|
|
return connection
|
|
}
|
|
|
|
func (s *StartedService) CloseConnection(ctx context.Context, request *CloseConnectionRequest) (*emptypb.Empty, error) {
|
|
s.serviceAccess.RLock()
|
|
switch s.serviceStatus.Status {
|
|
case ServiceStatus_STARTING, ServiceStatus_STARTED:
|
|
default:
|
|
s.serviceAccess.RUnlock()
|
|
return nil, os.ErrInvalid
|
|
}
|
|
boxService := s.instance
|
|
s.serviceAccess.RUnlock()
|
|
targetConn := boxService.clashServer.(*clashapi.Server).TrafficManager().Connection(uuid.FromStringOrNil(request.Id))
|
|
if targetConn != nil {
|
|
targetConn.Close()
|
|
}
|
|
return &emptypb.Empty{}, nil
|
|
}
|
|
|
|
func (s *StartedService) CloseAllConnections(ctx context.Context, empty *emptypb.Empty) (*emptypb.Empty, error) {
|
|
conntrack.Close()
|
|
return &emptypb.Empty{}, nil
|
|
}
|
|
|
|
func (s *StartedService) GetDeprecatedWarnings(ctx context.Context, empty *emptypb.Empty) (*DeprecatedWarnings, error) {
|
|
s.serviceAccess.RLock()
|
|
if s.serviceStatus.Status != ServiceStatus_STARTED {
|
|
s.serviceAccess.RUnlock()
|
|
return nil, os.ErrInvalid
|
|
}
|
|
boxService := s.instance
|
|
s.serviceAccess.RUnlock()
|
|
notes := service.FromContext[deprecated.Manager](boxService.ctx).(*deprecatedManager).Get()
|
|
return &DeprecatedWarnings{
|
|
Warnings: common.Map(notes, func(it deprecated.Note) *DeprecatedWarning {
|
|
return &DeprecatedWarning{
|
|
Message: it.Message(),
|
|
Impending: it.Impending(),
|
|
MigrationLink: it.MigrationLink,
|
|
}
|
|
}),
|
|
}, nil
|
|
}
|
|
|
|
func (s *StartedService) GetStartedAt(ctx context.Context, empty *emptypb.Empty) (*StartedAt, error) {
|
|
s.serviceAccess.RLock()
|
|
defer s.serviceAccess.RUnlock()
|
|
return &StartedAt{StartedAt: s.startedAt.UnixMilli()}, nil
|
|
}
|
|
|
|
func (s *StartedService) SubscribeHelperEvents(empty *emptypb.Empty, server grpc.ServerStreamingServer[HelperRequest]) error {
|
|
return os.ErrInvalid
|
|
}
|
|
|
|
func (s *StartedService) SendHelperResponse(ctx context.Context, response *HelperResponse) (*emptypb.Empty, error) {
|
|
return nil, os.ErrInvalid
|
|
}
|
|
|
|
func (s *StartedService) mustEmbedUnimplementedStartedServiceServer() {
|
|
}
|
|
|
|
func (s *StartedService) WriteMessage(level log.Level, message string) {
|
|
item := &log.Entry{Level: level, Message: message}
|
|
s.logAccess.Lock()
|
|
s.logLines.PushBack(item)
|
|
if s.logLines.Len() > s.logMaxLines {
|
|
s.logLines.Remove(s.logLines.Front())
|
|
}
|
|
s.logAccess.Unlock()
|
|
s.logSubscriber.Emit(item)
|
|
if s.debug {
|
|
s.handler.WriteDebugMessage(message)
|
|
}
|
|
}
|
|
|
|
func (s *StartedService) Instance() *Instance {
|
|
s.serviceAccess.RLock()
|
|
defer s.serviceAccess.RUnlock()
|
|
return s.instance
|
|
}
|