Add sing-box API service

This commit is contained in:
世界 2026-06-11 21:11:10 +08:00
parent 1ee3596db9
commit 3a5d654463
No known key found for this signature in database
GPG key ID: CD109927C34A63C4
36 changed files with 2075 additions and 849 deletions

View file

@ -264,8 +264,8 @@ lib_apple_new:
$(SING_FFI) generate --config $(LIBBOX_FFI_CONFIG) --platform-type apple
lib_install:
go install -v github.com/sagernet/gomobile/cmd/gomobile@v0.1.12
go install -v github.com/sagernet/gomobile/cmd/gobind@v0.1.12
go install -v github.com/sagernet/gomobile/cmd/gomobile@v0.1.13
go install -v github.com/sagernet/gomobile/cmd/gobind@v0.1.13
docs:
venv/bin/mkdocs serve

View file

@ -13,9 +13,9 @@ import (
type ClashServer interface {
LifecycleService
ConnectionTracker
Mode() string
ModeList() []string
SetMode(mode string)
SetModeUpdateHook(hook *observable.Subscriber[struct{}])
HistoryStorage() URLTestHistoryStorage
}

18
box.go
View file

@ -19,6 +19,8 @@ import (
"github.com/sagernet/sing-box/common/httpclient"
"github.com/sagernet/sing-box/common/taskmonitor"
"github.com/sagernet/sing-box/common/tls"
"github.com/sagernet/sing-box/common/trafficcontrol"
"github.com/sagernet/sing-box/common/urltest"
C "github.com/sagernet/sing-box/constant"
"github.com/sagernet/sing-box/dns"
"github.com/sagernet/sing-box/experimental"
@ -154,6 +156,12 @@ func New(options Options) (*Box, error) {
if experimentalOptions.V2RayAPI != nil && experimentalOptions.V2RayAPI.Listen != "" {
needV2RayAPI = true
}
needAPIService := common.Any(options.Services, func(it option.Service) bool {
return it.Type == C.TypeAPI
})
if needAPIService && service.PtrFromContext[urltest.HistoryStorage](ctx) == nil {
ctx = service.ContextWithPtr(ctx, urltest.NewHistoryStorage())
}
platformInterface := service.FromContext[adapter.PlatformInterface](ctx)
var defaultLogWriter io.Writer
if platformInterface != nil {
@ -162,7 +170,7 @@ func New(options Options) (*Box, error) {
logFactory, err := log.New(log.Options{
Context: ctx,
Options: common.PtrValueOrDefault(options.Log),
Observable: needClashAPI,
Observable: needClashAPI || needAPIService,
DefaultWriter: defaultLogWriter,
BaseTime: createdAt,
PlatformWriter: options.PlatformLogWriter,
@ -170,6 +178,7 @@ func New(options Options) (*Box, error) {
if err != nil {
return nil, E.Cause(err, "create log factory")
}
service.MustRegister[log.Factory](ctx, logFactory)
var internalServices []adapter.LifecycleService
routeOptions := common.PtrValueOrDefault(options.Route)
@ -221,6 +230,12 @@ func New(options Options) (*Box, error) {
if err != nil {
return nil, E.Cause(err, "initialize router")
}
if needClashAPI || needAPIService {
trafficManager := trafficcontrol.NewManager(outboundManager)
service.MustRegisterPtr(ctx, trafficManager)
router.AppendTracker(trafficManager)
internalServices = append(internalServices, trafficManager)
}
ntpOptions := common.PtrValueOrDefault(options.NTP)
var timeService *tls.TimeServiceWrapper
if ntpOptions.Enabled {
@ -398,7 +413,6 @@ func New(options Options) (*Box, error) {
if err != nil {
return nil, E.Cause(err, "create clash-server")
}
router.AppendTracker(clashServer)
service.MustRegister[adapter.ClashServer](ctx, clashServer)
internalServices = append(internalServices, clashServer)
}

View file

@ -0,0 +1,166 @@
package trafficcontrol
import (
"sync"
"sync/atomic"
"time"
"github.com/sagernet/sing-box/adapter"
"github.com/sagernet/sing-box/common/compatible"
"github.com/sagernet/sing/common/cleanup"
"github.com/sagernet/sing/common/observable"
"github.com/sagernet/sing/common/x/list"
"github.com/gofrs/uuid/v5"
)
type ConnectionEventType int
const (
ConnectionEventNew ConnectionEventType = iota
ConnectionEventClosed
)
type ConnectionEvent struct {
Type ConnectionEventType
ID uuid.UUID
Metadata *TrackerMetadata
ClosedAt time.Time
}
const closedConnectionsLimit = 1000
var (
_ adapter.ConnectionTracker = (*Manager)(nil)
_ adapter.LifecycleService = (*Manager)(nil)
)
type Manager struct {
outbound adapter.OutboundManager
uploadTotal atomic.Int64
downloadTotal atomic.Int64
connections compatible.Map[uuid.UUID, Tracker]
closedConnectionsAccess sync.Mutex
closedConnections list.List[TrackerMetadata]
eventSubscriber *observable.Subscriber[ConnectionEvent]
eventObserver *observable.Observer[ConnectionEvent]
cleaner *cleanup.Cleaner
}
func NewManager(outbound adapter.OutboundManager) *Manager {
manager := &Manager{
outbound: outbound,
eventSubscriber: observable.NewSubscriber[ConnectionEvent](256),
}
manager.eventObserver = observable.NewObserver(manager.eventSubscriber, 64)
manager.cleaner = cleanup.Add(manager.Clear)
return manager
}
func (m *Manager) Name() string {
return "traffic manager"
}
func (m *Manager) Start(stage adapter.StartStage) error {
return nil
}
func (m *Manager) Close() error {
m.cleaner.Close()
return m.eventObserver.Close()
}
func (m *Manager) SubscribeEvents() (observable.Subscription[ConnectionEvent], <-chan struct{}, error) {
return m.eventObserver.Subscribe()
}
func (m *Manager) UnSubscribeEvents(subscription observable.Subscription[ConnectionEvent]) {
m.eventObserver.UnSubscribe(subscription)
}
func (m *Manager) join(tracker Tracker) {
metadata := tracker.Metadata()
m.connections.Store(metadata.ID, tracker)
m.eventSubscriber.Emit(ConnectionEvent{
Type: ConnectionEventNew,
ID: metadata.ID,
Metadata: metadata,
})
}
func (m *Manager) leave(tracker Tracker) {
metadata := tracker.Metadata()
_, loaded := m.connections.LoadAndDelete(metadata.ID)
if !loaded {
return
}
closedAt := time.Now()
metadata.ClosedAt = closedAt
metadataCopy := *metadata
m.closedConnectionsAccess.Lock()
if m.closedConnections.Len() >= closedConnectionsLimit {
m.closedConnections.PopFront()
}
m.closedConnections.PushBack(metadataCopy)
m.closedConnectionsAccess.Unlock()
m.eventSubscriber.Emit(ConnectionEvent{
Type: ConnectionEventClosed,
ID: metadata.ID,
Metadata: &metadataCopy,
ClosedAt: closedAt,
})
}
func (m *Manager) Total() (uplinkTotal int64, downlinkTotal int64) {
return m.uploadTotal.Load(), m.downloadTotal.Load()
}
func (m *Manager) ConnectionsLen() int {
return m.connections.Len()
}
func (m *Manager) Connections() []*TrackerMetadata {
var connections []*TrackerMetadata
m.connections.Range(func(_ uuid.UUID, tracker Tracker) bool {
connections = append(connections, tracker.Metadata())
return true
})
return connections
}
func (m *Manager) ClosedConnections() []*TrackerMetadata {
m.closedConnectionsAccess.Lock()
values := m.closedConnections.Array()
m.closedConnectionsAccess.Unlock()
if len(values) == 0 {
return nil
}
connections := make([]*TrackerMetadata, len(values))
for i := range values {
connections[i] = &values[i]
}
return connections
}
func (m *Manager) Connection(id uuid.UUID) Tracker {
connection, loaded := m.connections.Load(id)
if !loaded {
return nil
}
return connection
}
func (m *Manager) CloseAllConnections() {
m.connections.Range(func(_ uuid.UUID, tracker Tracker) bool {
tracker.Close()
return true
})
}
func (m *Manager) Clear() {
m.closedConnectionsAccess.Lock()
defer m.closedConnectionsAccess.Unlock()
m.closedConnections.Init()
}

View file

@ -0,0 +1,163 @@
package trafficcontrol
import (
"context"
"net"
"sync/atomic"
"time"
"github.com/sagernet/sing-box/adapter"
"github.com/sagernet/sing/common"
"github.com/sagernet/sing/common/bufio"
N "github.com/sagernet/sing/common/network"
"github.com/gofrs/uuid/v5"
)
type TrackerMetadata struct {
ID uuid.UUID
Metadata adapter.InboundContext
CreatedAt time.Time
ClosedAt time.Time
Upload *atomic.Int64
Download *atomic.Int64
Chain []string
Rule adapter.Rule
Outbound string
OutboundType string
}
type Tracker interface {
Metadata() *TrackerMetadata
Close() error
}
func (m *Manager) RoutedConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, matchedRule adapter.Rule, matchOutbound adapter.Outbound) net.Conn {
upload := new(atomic.Int64)
download := new(atomic.Int64)
tracker := &connTracker{
ExtendedConn: bufio.NewCounterConn(conn, []N.CountFunc{func(n int64) {
upload.Add(n)
m.uploadTotal.Add(n)
}}, []N.CountFunc{func(n int64) {
download.Add(n)
m.downloadTotal.Add(n)
}}),
metadata: m.newTrackerMetadata(metadata, matchedRule, matchOutbound, upload, download),
manager: m,
}
m.join(tracker)
return tracker
}
func (m *Manager) RoutedPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, matchedRule adapter.Rule, matchOutbound adapter.Outbound) N.PacketConn {
upload := new(atomic.Int64)
download := new(atomic.Int64)
tracker := &packetConnTracker{
PacketConn: bufio.NewCounterPacketConn(conn, []N.CountFunc{func(n int64) {
upload.Add(n)
m.uploadTotal.Add(n)
}}, []N.CountFunc{func(n int64) {
download.Add(n)
m.downloadTotal.Add(n)
}}),
metadata: m.newTrackerMetadata(metadata, matchedRule, matchOutbound, upload, download),
manager: m,
}
m.join(tracker)
return tracker
}
func (m *Manager) newTrackerMetadata(metadata adapter.InboundContext, matchedRule adapter.Rule, matchOutbound adapter.Outbound, upload *atomic.Int64, download *atomic.Int64) TrackerMetadata {
id, _ := uuid.NewV4()
var (
chain []string
next string
outbound string
outboundType string
)
if matchOutbound != nil {
next = matchOutbound.Tag()
} else {
next = m.outbound.Default().Tag()
}
for {
detour, loaded := m.outbound.Outbound(next)
if !loaded {
break
}
chain = append(chain, next)
outbound = detour.Tag()
outboundType = detour.Type()
outboundGroup, isGroup := detour.(adapter.OutboundGroup)
if !isGroup {
break
}
next = outboundGroup.Now()
}
return TrackerMetadata{
ID: id,
Metadata: metadata,
CreatedAt: time.Now(),
Upload: upload,
Download: download,
Chain: common.Reverse(chain),
Rule: matchedRule,
Outbound: outbound,
OutboundType: outboundType,
}
}
type connTracker struct {
N.ExtendedConn
metadata TrackerMetadata
manager *Manager
}
func (t *connTracker) Metadata() *TrackerMetadata {
return &t.metadata
}
func (t *connTracker) Close() error {
t.manager.leave(t)
return t.ExtendedConn.Close()
}
func (t *connTracker) Upstream() any {
return t.ExtendedConn
}
func (t *connTracker) ReaderReplaceable() bool {
return true
}
func (t *connTracker) WriterReplaceable() bool {
return true
}
type packetConnTracker struct {
N.PacketConn
metadata TrackerMetadata
manager *Manager
}
func (t *packetConnTracker) Metadata() *TrackerMetadata {
return &t.metadata
}
func (t *packetConnTracker) Close() error {
t.manager.leave(t)
return t.PacketConn.Close()
}
func (t *packetConnTracker) Upstream() any {
return t.PacketConn
}
func (t *packetConnTracker) ReaderReplaceable() bool {
return true
}
func (t *packetConnTracker) WriterReplaceable() bool {
return true
}

View file

@ -29,6 +29,7 @@ const (
TypeDERP = "derp"
TypeResolved = "resolved"
TypeSSMAPI = "ssm-api"
TypeAPI = "api"
TypeCCM = "ccm"
TypeOCM = "ocm"
TypeOOMKiller = "oom-killer"

View file

@ -0,0 +1,28 @@
package daemon
import (
"context"
"time"
"github.com/sagernet/sing-box/log"
)
const defaultAttachedLogMaxLines = 3000
// StartOrReloadService and CloseService must not be called on an attached service.
func NewAttachedService(ctx context.Context) *StartedService {
instance := attachInstance(ctx)
s := NewStartedService(ServiceOptions{
Context: ctx,
LogMaxLines: defaultAttachedLogMaxLines,
})
s.instance = instance
s.serviceStatus = &ServiceStatus{Status: ServiceStatus_STARTED}
s.startedAt = time.Now()
instance.urlTestHistoryStorage.SetHook(s.urlTestSubscriber)
if instance.clashServer != nil {
instance.clashServer.SetModeUpdateHook(s.clashModeSubscriber)
}
instance.logFactory.(log.ObservableFactory).AttachPlatformWriter(s)
return s
}

85
daemon/client.go Normal file
View file

@ -0,0 +1,85 @@
package daemon
import (
"context"
"crypto/tls"
"net"
"net/url"
E "github.com/sagernet/sing/common/exceptions"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/metadata"
)
type RemoteClientOptions struct {
ServerURL string
Secret string
}
func (o RemoteClientOptions) ServerTarget() (string, credentials.TransportCredentials, error) {
if o.ServerURL == "" {
return "", nil, E.New("missing server URL")
}
serverURL, err := url.Parse(o.ServerURL)
if err != nil {
return "", nil, E.Cause(err, "invalid server URL: ", o.ServerURL)
}
var enableTLS bool
switch serverURL.Scheme {
case "http":
case "https":
enableTLS = true
default:
return "", nil, E.New("invalid server URL scheme: ", serverURL.Scheme, ", expected http or https")
}
host := serverURL.Hostname()
if host == "" {
return "", nil, E.New("missing host in server URL: ", o.ServerURL)
}
port := serverURL.Port()
if port == "" {
if enableTLS {
port = "443"
} else {
port = "80"
}
}
transportCredentials := insecure.NewCredentials()
if enableTLS {
transportCredentials = credentials.NewTLS(&tls.Config{ServerName: host})
}
return net.JoinHostPort(host, port), transportCredentials, nil
}
func NewRemoteClient(options RemoteClientOptions) (*grpc.ClientConn, error) {
target, transportCredentials, err := options.ServerTarget()
if err != nil {
return nil, err
}
return grpc.NewClient(target,
grpc.WithTransportCredentials(transportCredentials),
grpc.WithUnaryInterceptor(NewClientAuthUnaryInterceptor(options.Secret)),
grpc.WithStreamInterceptor(NewClientAuthStreamInterceptor(options.Secret)),
)
}
func NewClientAuthUnaryInterceptor(secret string) grpc.UnaryClientInterceptor {
return func(ctx context.Context, method string, request, reply any, clientConn *grpc.ClientConn, invoker grpc.UnaryInvoker, options ...grpc.CallOption) error {
if secret != "" {
ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+secret)
}
return invoker(ctx, method, request, reply, clientConn, options...)
}
}
func NewClientAuthStreamInterceptor(secret string) grpc.StreamClientInterceptor {
return func(ctx context.Context, desc *grpc.StreamDesc, clientConn *grpc.ClientConn, method string, streamer grpc.Streamer, options ...grpc.CallOption) (grpc.ClientStream, error) {
if secret != "" {
ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+secret)
}
return streamer(ctx, desc, clientConn, method, options...)
}
}

42
daemon/errors.go Normal file
View file

@ -0,0 +1,42 @@
package daemon
import (
"context"
"errors"
"os"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
func UnaryErrorInterceptor(ctx context.Context, request any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
response, err := handler(ctx, request)
if err != nil {
return nil, mapStatusError(err)
}
return response, nil
}
func StreamErrorInterceptor(server any, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
err := handler(server, stream)
if err != nil {
return mapStatusError(err)
}
return nil
}
func mapStatusError(err error) error {
if _, loaded := status.FromError(err); loaded {
return err
}
switch {
case errors.Is(err, os.ErrInvalid):
return status.Error(codes.FailedPrecondition, "service not started")
case errors.Is(err, os.ErrClosed):
return status.Error(codes.Unavailable, "service is closing")
case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
return status.FromContextError(err).Err()
}
return err
}

View file

@ -6,10 +6,10 @@ import (
"github.com/sagernet/sing-box"
"github.com/sagernet/sing-box/adapter"
"github.com/sagernet/sing-box/common/trafficcontrol"
"github.com/sagernet/sing-box/common/urltest"
C "github.com/sagernet/sing-box/constant"
"github.com/sagernet/sing-box/experimental/deprecated"
"github.com/sagernet/sing-box/include"
"github.com/sagernet/sing-box/log"
"github.com/sagernet/sing-box/option"
"github.com/sagernet/sing/common"
@ -25,9 +25,13 @@ type Instance struct {
instance *box.Box
connectionManager adapter.ConnectionManager
clashServer adapter.ClashServer
trafficManager *trafficcontrol.Manager
cacheFile adapter.CacheFile
pauseManager pause.Manager
urlTestHistoryStorage *urltest.HistoryStorage
urlTestHistoryStorage adapter.URLTestHistoryStorage
outboundManager adapter.OutboundManager
endpointManager adapter.EndpointManager
logFactory log.Factory
}
func (s *StartedService) CheckConfig(configContent string) error {
@ -71,7 +75,7 @@ type OverrideOptions struct {
func (s *StartedService) newInstance(profileContent string, overrideOptions *OverrideOptions) (*Instance, error) {
ctx := service.ExtendContext(s.ctx)
service.MustRegister[deprecated.Manager](ctx, new(deprecatedManager))
ctx, cancel := context.WithCancel(include.Context(ctx))
ctx, cancel := context.WithCancel(ctx)
options, err := parseConfig(ctx, profileContent)
if err != nil {
cancel()
@ -120,12 +124,31 @@ func (s *StartedService) newInstance(profileContent string, overrideOptions *Ove
i.instance = boxInstance
i.connectionManager = service.FromContext[adapter.ConnectionManager](ctx)
i.clashServer = service.FromContext[adapter.ClashServer](ctx)
i.trafficManager = service.PtrFromContext[trafficcontrol.Manager](ctx)
i.pauseManager = service.FromContext[pause.Manager](ctx)
i.cacheFile = service.FromContext[adapter.CacheFile](ctx)
i.outboundManager = service.FromContext[adapter.OutboundManager](ctx)
i.endpointManager = service.FromContext[adapter.EndpointManager](ctx)
i.logFactory = boxInstance.LogFactory()
log.SetStdLogger(boxInstance.LogFactory().Logger())
return i, nil
}
func attachInstance(ctx context.Context) *Instance {
return &Instance{
ctx: ctx,
connectionManager: service.FromContext[adapter.ConnectionManager](ctx),
clashServer: service.FromContext[adapter.ClashServer](ctx),
trafficManager: service.PtrFromContext[trafficcontrol.Manager](ctx),
pauseManager: service.FromContext[pause.Manager](ctx),
cacheFile: service.FromContext[adapter.CacheFile](ctx),
urlTestHistoryStorage: service.PtrFromContext[urltest.HistoryStorage](ctx),
outboundManager: service.FromContext[adapter.OutboundManager](ctx),
endpointManager: service.FromContext[adapter.EndpointManager](ctx),
logFactory: service.FromContext[log.Factory](ctx),
}
}
func (i *Instance) Start() error {
return i.instance.Start()
}

66
daemon/server.go Normal file
View file

@ -0,0 +1,66 @@
package daemon
import (
"context"
"strings"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/health"
"google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/reflection"
"google.golang.org/grpc/status"
)
func NewServer(startedService *StartedService, secret string) *grpc.Server {
server := grpc.NewServer(
grpc.ChainUnaryInterceptor(newUnaryAuthInterceptor(secret), UnaryErrorInterceptor),
grpc.ChainStreamInterceptor(newStreamAuthInterceptor(secret), StreamErrorInterceptor),
)
healthServer := health.NewServer()
RegisterStartedServiceServer(server, startedService)
healthServer.SetServingStatus(StartedService_ServiceDesc.ServiceName, grpc_health_v1.HealthCheckResponse_SERVING)
grpc_health_v1.RegisterHealthServer(server, healthServer)
reflection.Register(server)
return server
}
func newUnaryAuthInterceptor(secret string) grpc.UnaryServerInterceptor {
return func(ctx context.Context, request any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
err := authenticate(ctx, secret)
if err != nil {
return nil, err
}
return handler(ctx, request)
}
}
func newStreamAuthInterceptor(secret string) grpc.StreamServerInterceptor {
return func(server any, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
err := authenticate(stream.Context(), secret)
if err != nil {
return err
}
return handler(server, stream)
}
}
func authenticate(ctx context.Context, secret string) error {
if secret == "" {
return nil
}
md, loaded := metadata.FromIncomingContext(ctx)
if !loaded {
return status.Error(codes.Unauthenticated, "missing metadata")
}
values := md.Get("authorization")
if len(values) == 0 {
return status.Error(codes.Unauthenticated, "missing authorization")
}
token, isBearer := strings.CutPrefix(values[0], "Bearer ")
if !isBearer || token != secret {
return status.Error(codes.Unauthenticated, "invalid authorization")
}
return nil
}

View file

@ -11,17 +11,15 @@ import (
"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/trafficcontrol"
"github.com/sagernet/sing-box/common/urltest"
C "github.com/sagernet/sing-box/constant"
"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-box/service/oomkiller"
"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"
@ -34,6 +32,8 @@ import (
"google.golang.org/protobuf/types/known/emptypb"
)
const APIVersion = 1
var _ StartedServiceServer = (*StartedService)(nil)
type StartedService struct {
@ -65,9 +65,6 @@ type StartedService struct {
urlTestHistoryStorage *urltest.HistoryStorage
clashModeSubscriber *observable.Subscriber[struct{}]
clashModeObserver *observable.Observer[struct{}]
connectionEventSubscriber *observable.Subscriber[trafficontrol.ConnectionEvent]
connectionEventObserver *observable.Observer[trafficontrol.ConnectionEvent]
}
type ServiceOptions struct {
@ -101,22 +98,27 @@ func NewStartedService(options ServiceOptions) *StartedService {
// 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),
connectionEventSubscriber: observable.NewSubscriber[trafficontrol.ConnectionEvent](256),
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)
s.connectionEventObserver = observable.NewObserver(s.connectionEventSubscriber, 64)
return s
}
func (s *StartedService) GetVersion(ctx context.Context, empty *emptypb.Empty) (*Version, error) {
return &Version{
Version: C.Version,
ApiVersion: APIVersion,
}, nil
}
func (s *StartedService) resetLogs() {
s.logAccess.Lock()
s.logLines = list.List[*log.Entry]{}
@ -163,12 +165,12 @@ func (s *StartedService) waitForStarted(ctx context.Context) error {
return ctx.Err()
case <-s.ctx.Done():
return s.ctx.Err()
case status := <-subscription:
switch status.Status {
case statusUpdate := <-subscription:
switch statusUpdate.Status {
case ServiceStatus_STARTED:
return nil
case ServiceStatus_FATAL:
return E.New(status.ErrorMessage)
return status.Error(codes.FailedPrecondition, statusUpdate.ErrorMessage)
case ServiceStatus_IDLE, ServiceStatus_STOPPING:
return os.ErrInvalid
}
@ -203,7 +205,6 @@ func (s *StartedService) StartOrReloadService(profileContent string, options *Ov
instance.urlTestHistoryStorage.SetHook(s.urlTestSubscriber)
if instance.clashServer != nil {
instance.clashServer.SetModeUpdateHook(s.clashModeSubscriber)
instance.clashServer.(*clashapi.Server).TrafficManager().SetEventHook(s.connectionEventSubscriber)
}
s.serviceAccess.Unlock()
err = instance.Start()
@ -227,7 +228,6 @@ func (s *StartedService) Close() {
s.logSubscriber.Close()
s.urlTestSubscriber.Close()
s.clashModeSubscriber.Close()
s.connectionEventSubscriber.Close()
}
func (s *StartedService) CloseService() error {
@ -363,7 +363,7 @@ func (s *StartedService) GetDefaultLogLevel(ctx context.Context, empty *emptypb.
s.serviceAccess.RUnlock()
return nil, os.ErrInvalid
}
logLevel := s.instance.instance.LogFactory().Level()
logLevel := s.instance.logFactory.Level()
s.serviceAccess.RUnlock()
return &DefaultLogLevel{Level: LogLevel(logLevel)}, nil
}
@ -415,13 +415,10 @@ func (s *StartedService) readStatus() *Status {
if nowService != nil && nowService.connectionManager != nil {
status.ConnectionsOut = int32(nowService.connectionManager.Count())
}
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())
}
if nowService != nil && nowService.trafficManager != nil {
status.TrafficAvailable = true
status.UplinkTotal, status.DownlinkTotal = nowService.trafficManager.Total()
status.ConnectionsIn = int32(nowService.trafficManager.ConnectionsLen())
}
return &status
}
@ -463,7 +460,7 @@ func (s *StartedService) SubscribeGroups(empty *emptypb.Empty, server grpc.Serve
func (s *StartedService) readGroups() *Groups {
historyStorage := s.instance.urlTestHistoryStorage
boxService := s.instance
outbounds := boxService.instance.Outbound().Outbounds()
outbounds := boxService.outboundManager.Outbounds()
var iGroups []adapter.OutboundGroup
for _, it := range outbounds {
if group, isGroup := it.(adapter.OutboundGroup); isGroup {
@ -484,7 +481,7 @@ func (s *StartedService) readGroups() *Groups {
}
for _, itemTag := range iGroup.All() {
itemOutbound, isLoaded := boxService.instance.Outbound().Outbound(itemTag)
itemOutbound, isLoaded := boxService.outboundManager.Outbound(itemTag)
if !isLoaded {
continue
}
@ -515,7 +512,7 @@ func (s *StartedService) GetClashModeStatus(ctx context.Context, empty *emptypb.
clashServer := s.instance.clashServer
s.serviceAccess.RUnlock()
if clashServer == nil {
return nil, os.ErrInvalid
return nil, status.Error(codes.Unimplemented, "clash mode not available")
}
return &ClashModeStatus{
ModeList: clashServer.ModeList(),
@ -539,7 +536,12 @@ func (s *StartedService) SubscribeClashMode(empty *emptypb.Empty, server grpc.Se
s.serviceAccess.RUnlock()
return os.ErrInvalid
}
message := &ClashMode{Mode: s.instance.clashServer.Mode()}
clashServer := s.instance.clashServer
if clashServer == nil {
s.serviceAccess.RUnlock()
return status.Error(codes.Unimplemented, "clash mode not available")
}
message := &ClashMode{Mode: clashServer.Mode()}
s.serviceAccess.RUnlock()
err = server.Send(message)
if err != nil {
@ -565,7 +567,10 @@ func (s *StartedService) SetClashMode(ctx context.Context, request *ClashMode) (
}
clashServer := s.instance.clashServer
s.serviceAccess.RUnlock()
clashServer.(*clashapi.Server).SetMode(request.Mode)
if clashServer == nil {
return nil, status.Error(codes.Unimplemented, "clash mode not available")
}
clashServer.SetMode(request.Mode)
return &emptypb.Empty{}, nil
}
@ -578,13 +583,13 @@ func (s *StartedService) URLTest(ctx context.Context, request *URLTestRequest) (
boxService := s.instance
s.serviceAccess.RUnlock()
groupTag := request.OutboundTag
abstractOutboundGroup, isLoaded := boxService.instance.Outbound().Outbound(groupTag)
abstractOutboundGroup, isLoaded := boxService.outboundManager.Outbound(groupTag)
if !isLoaded {
return nil, E.New("outbound group not found: ", groupTag)
return nil, status.Error(codes.NotFound, "outbound group not found: "+groupTag)
}
outboundGroup, isOutboundGroup := abstractOutboundGroup.(adapter.OutboundGroup)
if !isOutboundGroup {
return nil, E.New("outbound is not a group: ", groupTag)
return nil, status.Error(codes.InvalidArgument, "outbound is not a group: "+groupTag)
}
urlTest, isURLTest := abstractOutboundGroup.(*group.URLTest)
if isURLTest {
@ -593,7 +598,7 @@ func (s *StartedService) URLTest(ctx context.Context, request *URLTestRequest) (
historyStorage := boxService.urlTestHistoryStorage
outbounds := common.Filter(common.Map(outboundGroup.All(), func(it string) adapter.Outbound {
itOutbound, _ := boxService.instance.Outbound().Outbound(it)
itOutbound, _ := boxService.outboundManager.Outbound(it)
return itOutbound
}), func(it adapter.Outbound) bool {
if it == nil {
@ -631,18 +636,18 @@ func (s *StartedService) SelectOutbound(ctx context.Context, request *SelectOutb
s.serviceAccess.RUnlock()
return nil, os.ErrInvalid
}
boxService := s.instance.instance
boxService := s.instance
s.serviceAccess.RUnlock()
outboundGroup, isLoaded := boxService.Outbound().Outbound(request.GroupTag)
outboundGroup, isLoaded := boxService.outboundManager.Outbound(request.GroupTag)
if !isLoaded {
return nil, E.New("selector not found: ", request.GroupTag)
return nil, status.Error(codes.NotFound, "selector not found: "+request.GroupTag)
}
selector, isSelector := outboundGroup.(*group.Selector)
if !isSelector {
return nil, E.New("outbound is not a selector: ", request.GroupTag)
return nil, status.Error(codes.InvalidArgument, "outbound is not a selector: "+request.GroupTag)
}
if !selector.SelectOutbound(request.OutboundTag) {
return nil, E.New("outbound not found in selector: ", request.OutboundTag)
return nil, status.Error(codes.NotFound, "outbound not found in selector: "+request.OutboundTag)
}
s.urlTestObserver.Emit(struct{}{})
return &emptypb.Empty{}, nil
@ -688,17 +693,16 @@ func (s *StartedService) SubscribeConnections(request *SubscribeConnectionsReque
boxService := s.instance
s.serviceAccess.RUnlock()
if boxService.clashServer == nil {
return E.New("clash server not available")
trafficManager := boxService.trafficManager
if trafficManager == nil {
return status.Error(codes.Unimplemented, "connection tracking not available")
}
trafficManager := boxService.clashServer.(*clashapi.Server).TrafficManager()
subscription, done, err := s.connectionEventObserver.Subscribe()
subscription, done, err := trafficManager.SubscribeEvents()
if err != nil {
return err
}
defer s.connectionEventObserver.UnSubscribe(subscription)
defer trafficManager.UnSubscribeEvents(subscription)
connectionSnapshots := make(map[uuid.UUID]connectionSnapshot)
initialEvents := s.buildInitialConnectionState(trafficManager, connectionSnapshots)
@ -768,7 +772,7 @@ type connectionSnapshot struct {
hadTraffic bool
}
func (s *StartedService) buildInitialConnectionState(manager *trafficontrol.Manager, snapshots map[uuid.UUID]connectionSnapshot) []*ConnectionEvent {
func (s *StartedService) buildInitialConnectionState(manager *trafficcontrol.Manager, snapshots map[uuid.UUID]connectionSnapshot) []*ConnectionEvent {
var events []*ConnectionEvent
for _, metadata := range manager.Connections() {
@ -796,9 +800,9 @@ func (s *StartedService) buildInitialConnectionState(manager *trafficontrol.Mana
return events
}
func (s *StartedService) applyConnectionEvent(event trafficontrol.ConnectionEvent, snapshots map[uuid.UUID]connectionSnapshot) *ConnectionEvent {
func (s *StartedService) applyConnectionEvent(event trafficcontrol.ConnectionEvent, snapshots map[uuid.UUID]connectionSnapshot) *ConnectionEvent {
switch event.Type {
case trafficontrol.ConnectionEventNew:
case trafficcontrol.ConnectionEventNew:
if _, exists := snapshots[event.ID]; exists {
return nil
}
@ -811,7 +815,7 @@ func (s *StartedService) applyConnectionEvent(event trafficontrol.ConnectionEven
Id: event.ID.String(),
Connection: buildConnectionProto(event.Metadata),
}
case trafficontrol.ConnectionEventClosed:
case trafficcontrol.ConnectionEventClosed:
delete(snapshots, event.ID)
protoEvent := &ConnectionEvent{
Type: ConnectionEventType_CONNECTION_EVENT_CLOSED,
@ -836,9 +840,9 @@ func (s *StartedService) applyConnectionEvent(event trafficontrol.ConnectionEven
}
}
func (s *StartedService) buildTrafficUpdates(manager *trafficontrol.Manager, snapshots map[uuid.UUID]connectionSnapshot) []*ConnectionEvent {
func (s *StartedService) buildTrafficUpdates(manager *trafficcontrol.Manager, snapshots map[uuid.UUID]connectionSnapshot) []*ConnectionEvent {
activeConnections := manager.Connections()
activeIndex := make(map[uuid.UUID]*trafficontrol.TrackerMetadata, len(activeConnections))
activeIndex := make(map[uuid.UUID]*trafficcontrol.TrackerMetadata, len(activeConnections))
var events []*ConnectionEvent
for _, metadata := range activeConnections {
@ -902,13 +906,13 @@ func (s *StartedService) buildTrafficUpdates(manager *trafficontrol.Manager, sna
}
}
var closedIndex map[uuid.UUID]*trafficontrol.TrackerMetadata
var closedIndex map[uuid.UUID]*trafficcontrol.TrackerMetadata
for id := range snapshots {
if _, exists := activeIndex[id]; exists {
continue
}
if closedIndex == nil {
closedIndex = make(map[uuid.UUID]*trafficontrol.TrackerMetadata)
closedIndex = make(map[uuid.UUID]*trafficcontrol.TrackerMetadata)
for _, metadata := range manager.ClosedConnections() {
closedIndex[metadata.ID] = metadata
}
@ -934,7 +938,7 @@ func (s *StartedService) buildTrafficUpdates(manager *trafficontrol.Manager, sna
return events
}
func buildConnectionProto(metadata *trafficontrol.TrackerMetadata) *Connection {
func buildConnectionProto(metadata *trafficcontrol.TrackerMetadata) *Connection {
var rule string
if metadata.Rule != nil {
rule = metadata.Rule.String()
@ -984,7 +988,10 @@ func (s *StartedService) CloseConnection(ctx context.Context, request *CloseConn
}
boxService := s.instance
s.serviceAccess.RUnlock()
targetConn := boxService.clashServer.(*clashapi.Server).TrafficManager().Connection(uuid.FromStringOrNil(request.Id))
if boxService.trafficManager == nil {
return nil, status.Error(codes.Unimplemented, "connection tracking not available")
}
targetConn := boxService.trafficManager.Connection(uuid.FromStringOrNil(request.Id))
if targetConn != nil {
targetConn.Close()
}
@ -1009,7 +1016,11 @@ func (s *StartedService) GetDeprecatedWarnings(ctx context.Context, empty *empty
}
boxService := s.instance
s.serviceAccess.RUnlock()
notes := service.FromContext[deprecated.Manager](boxService.ctx).(*deprecatedManager).Get()
manager, isCollecting := service.FromContext[deprecated.Manager](boxService.ctx).(*deprecatedManager)
if !isCollecting {
return &DeprecatedWarnings{}, nil
}
notes := manager.Get()
return &DeprecatedWarnings{
Warnings: common.Map(notes, func(it deprecated.Note) *DeprecatedWarning {
return &DeprecatedWarning{
@ -1050,7 +1061,7 @@ func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.Server
s.serviceAccess.RUnlock()
historyStorage := boxService.urlTestHistoryStorage
var list OutboundList
for _, ob := range boxService.instance.Outbound().Outbounds() {
for _, ob := range boxService.outboundManager.Outbounds() {
item := &GroupItem{
Tag: ob.Tag(),
Type: ob.Type(),
@ -1061,7 +1072,7 @@ func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.Server
}
list.Outbounds = append(list.Outbounds, item)
}
for _, ep := range boxService.instance.Endpoint().Endpoints() {
for _, ep := range boxService.endpointManager.Endpoints() {
item := &GroupItem{
Tag: ep.Tag(),
Type: ep.Type(),
@ -1090,11 +1101,11 @@ func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.Server
func resolveOutbound(instance *Instance, tag string) (adapter.Outbound, error) {
if tag == "" {
return instance.instance.Outbound().Default(), nil
return instance.outboundManager.Default(), nil
}
outbound, loaded := instance.instance.Outbound().Outbound(tag)
outbound, loaded := instance.outboundManager.Outbound(tag)
if !loaded {
return nil, E.New("outbound not found: ", tag)
return nil, status.Error(codes.NotFound, "outbound not found: "+tag)
}
return outbound, nil
}
@ -1103,10 +1114,10 @@ func resolveTailscaleEndpoint(instance *Instance, tag string) (adapter.Endpoint,
endpointManager := service.FromContext[adapter.EndpointManager](instance.ctx)
endpoint, loaded := endpointManager.Get(tag)
if !loaded {
return nil, E.New("endpoint not found: ", tag)
return nil, status.Error(codes.NotFound, "endpoint not found: "+tag)
}
if endpoint.Type() != C.TypeTailscale {
return nil, E.New("endpoint is not Tailscale: ", tag)
return nil, status.Error(codes.InvalidArgument, "endpoint is not Tailscale: "+tag)
}
return endpoint, nil
}

File diff suppressed because it is too large Load diff

View file

@ -6,6 +6,7 @@ option go_package = "github.com/sagernet/sing-box/daemon";
import "google/protobuf/empty.proto";
service StartedService {
rpc GetVersion(google.protobuf.Empty) returns(Version) {}
rpc SubscribeServiceStatus(google.protobuf.Empty) returns(stream ServiceStatus) {}
rpc SubscribeLog(google.protobuf.Empty) returns(stream Log) {}
rpc GetDefaultLogLevel(google.protobuf.Empty) returns(DefaultLogLevel) {}
@ -39,6 +40,11 @@ service StartedService {
rpc StartTailscaleSSHSession(stream TailscaleSSHClientMessage) returns (stream TailscaleSSHServerMessage) {}
}
message Version {
string version = 1;
int32 apiVersion = 2;
}
message ServiceStatus {
enum Type {
IDLE = 0;

View file

@ -15,6 +15,7 @@ import (
const _ = grpc.SupportPackageIsVersion9
const (
StartedService_GetVersion_FullMethodName = "/daemon.StartedService/GetVersion"
StartedService_SubscribeServiceStatus_FullMethodName = "/daemon.StartedService/SubscribeServiceStatus"
StartedService_SubscribeLog_FullMethodName = "/daemon.StartedService/SubscribeLog"
StartedService_GetDefaultLogLevel_FullMethodName = "/daemon.StartedService/GetDefaultLogLevel"
@ -47,6 +48,7 @@ const (
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
type StartedServiceClient interface {
GetVersion(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*Version, error)
SubscribeServiceStatus(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ServiceStatus], error)
SubscribeLog(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Log], error)
GetDefaultLogLevel(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*DefaultLogLevel, error)
@ -83,6 +85,16 @@ func NewStartedServiceClient(cc grpc.ClientConnInterface) StartedServiceClient {
return &startedServiceClient{cc}
}
func (c *startedServiceClient) GetVersion(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*Version, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(Version)
err := c.cc.Invoke(ctx, StartedService_GetVersion_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *startedServiceClient) SubscribeServiceStatus(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ServiceStatus], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &StartedService_ServiceDesc.Streams[0], StartedService_SubscribeServiceStatus_FullMethodName, cOpts...)
@ -449,6 +461,7 @@ type StartedService_StartTailscaleSSHSessionClient = grpc.BidiStreamingClient[Ta
// All implementations must embed UnimplementedStartedServiceServer
// for forward compatibility.
type StartedServiceServer interface {
GetVersion(context.Context, *emptypb.Empty) (*Version, error)
SubscribeServiceStatus(*emptypb.Empty, grpc.ServerStreamingServer[ServiceStatus]) error
SubscribeLog(*emptypb.Empty, grpc.ServerStreamingServer[Log]) error
GetDefaultLogLevel(context.Context, *emptypb.Empty) (*DefaultLogLevel, error)
@ -485,6 +498,10 @@ type StartedServiceServer interface {
// pointer dereference when methods are called.
type UnimplementedStartedServiceServer struct{}
func (UnimplementedStartedServiceServer) GetVersion(context.Context, *emptypb.Empty) (*Version, error) {
return nil, status.Error(codes.Unimplemented, "method GetVersion not implemented")
}
func (UnimplementedStartedServiceServer) SubscribeServiceStatus(*emptypb.Empty, grpc.ServerStreamingServer[ServiceStatus]) error {
return status.Error(codes.Unimplemented, "method SubscribeServiceStatus not implemented")
}
@ -609,6 +626,24 @@ func RegisterStartedServiceServer(s grpc.ServiceRegistrar, srv StartedServiceSer
s.RegisterService(&StartedService_ServiceDesc, srv)
}
func _StartedService_GetVersion_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).GetVersion(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: StartedService_GetVersion_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(StartedServiceServer).GetVersion(ctx, req.(*emptypb.Empty))
}
return interceptor(ctx, in, info, handler)
}
func _StartedService_SubscribeServiceStatus_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(emptypb.Empty)
if err := stream.RecvMsg(m); err != nil {
@ -996,6 +1031,10 @@ var StartedService_ServiceDesc = grpc.ServiceDesc{
ServiceName: "daemon.StartedService",
HandlerType: (*StartedServiceServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "GetVersion",
Handler: _StartedService_GetVersion_Handler,
},
{
MethodName: "GetDefaultLogLevel",
Handler: _StartedService_GetDefaultLogLevel_Handler,

View file

@ -121,7 +121,7 @@ func (s *StartedService) StartTailscaleSSHSession(
}
sshClient := ssh.NewClient(sshConn, chans, reqs)
if start.ForwardAgent {
if start.ForwardAgent && s.handler != nil {
agentChannels := sshClient.HandleChannelOpen("auth-agent@openssh.com")
if agentChannels != nil {
go func() {
@ -176,7 +176,7 @@ func (s *StartedService) StartTailscaleSSHSession(
}))
}
if start.ForwardAgent {
if start.ForwardAgent && s.handler != nil {
err = agent.RequestAgentForwarding(sshSession)
if err != nil {
common.Close(sshSession, sshClient)

View file

@ -0,0 +1,57 @@
---
icon: material/new-box
---
!!! question "Since sing-box 1.14.0"
# sing-box API
The sing-box API service is a gRPC server for observing and controlling the running sing-box instance.
The server also accepts [gRPC-Web](https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-WEB.md) requests,
including the WebSocket transport of [@improbable-eng/grpc-web](https://github.com/improbable-eng/grpc-web)
for bidirectional streaming methods.
### Structure
```json
{
"type": "api",
... // Listen Fields
"secret": "",
"access_control_allow_origin": [],
"access_control_allow_private_network": false,
"tls": {}
}
```
### Listen Fields
See [Listen Fields](/configuration/shared/listen/) for details.
### Fields
#### secret
Secret for the API.
Clients authenticate with the standard `authorization: Bearer <secret>` gRPC metadata header.
If empty, authentication is disabled.
#### access_control_allow_origin
CORS allowed origins, `*` will be used if empty.
#### access_control_allow_private_network
Allow access from private network.
#### tls
TLS configuration, see [TLS](/configuration/shared/tls/#inbound).
Connection tracking and Clash mode methods require [Clash API](/configuration/experimental/clash-api/)
to be configured, otherwise they fail with `UNIMPLEMENTED`.

View file

@ -0,0 +1,56 @@
---
icon: material/new-box
---
!!! question "自 sing-box 1.14.0 起"
# sing-box API
sing-box API 服务是用于观察与控制正在运行的 sing-box 实例的 gRPC 服务器。
服务器同时接受 [gRPC-Web](https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-WEB.md) 请求,
包括用于双向流方法的 [@improbable-eng/grpc-web](https://github.com/improbable-eng/grpc-web) WebSocket 传输。
### 结构
```json
{
"type": "api",
... // 监听字段
"secret": "",
"access_control_allow_origin": [],
"access_control_allow_private_network": false,
"tls": {}
}
```
### 监听字段
参阅 [监听字段](/zh/configuration/shared/listen/)。
### 字段
#### secret
API 密钥。
客户端通过标准的 `authorization: Bearer <secret>` gRPC metadata 头认证。
留空则禁用认证。
#### access_control_allow_origin
允许的 CORS 来源,默认使用 `*`
#### access_control_allow_private_network
允许从私有网络访问。
#### tls
TLS 配置,参阅 [TLS](/zh/configuration/shared/tls/#inbound)。
连接跟踪与 Clash 模式方法需要配置 [Clash API](/zh/configuration/experimental/clash-api/),
否则将以 `UNIMPLEMENTED` 失败。

View file

@ -23,6 +23,7 @@ icon: material/new-box
| Type | Format |
|-------------------|---------------------------------------|
| `api` | [sing-box API](./api) |
| `ccm` | [CCM](./ccm) |
| `derp` | [DERP](./derp) |
| `hysteria-realm` | [Hysteria Realm](./hysteria-realm) |

View file

@ -23,6 +23,7 @@ icon: material/new-box
| 类型 | 格式 |
|-------------------|---------------------------------------|
| `api` | [sing-box API](./api) |
| `ccm` | [CCM](./ccm) |
| `derp` | [DERP](./derp) |
| `hysteria-realm` | [Hysteria Realm](./hysteria-realm) |

View file

@ -5,10 +5,10 @@ import (
"context"
"net"
"net/http"
"runtime"
"runtime/debug"
"time"
"github.com/sagernet/sing-box/experimental/clashapi/trafficontrol"
"github.com/sagernet/sing/common/json"
"github.com/sagernet/ws"
"github.com/sagernet/ws/wsutil"
@ -28,7 +28,7 @@ func (s *Server) setupMetaAPI(r chi.Router) {
})
r.Mount("/", middleware.Profiler())
}
r.Get("/memory", memory(s.ctx, s.trafficManager))
r.Get("/memory", memory(s.ctx))
r.Mount("/group", groupRouter(s))
r.Mount("/upgrade", upgradeRouter(s))
}
@ -38,7 +38,13 @@ type Memory struct {
OSLimit uint64 `json:"oslimit"` // maybe we need it in the future
}
func memory(ctx context.Context, trafficManager *trafficontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
func inuseMemory() uint64 {
var memStats runtime.MemStats
runtime.ReadMemStats(&memStats)
return memStats.StackInuse + memStats.HeapInuse + memStats.HeapIdle - memStats.HeapReleased
}
func memory(ctx context.Context) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
var conn net.Conn
if r.Header.Get("Upgrade") == "websocket" {
@ -68,7 +74,7 @@ func memory(ctx context.Context, trafficManager *trafficontrol.Manager) func(w h
}
buf.Reset()
inuse := trafficManager.Snapshot().Memory
inuse := inuseMemory()
// make chat.js begin with zero
// this is shit var,but we need output 0 for first time

View file

@ -8,7 +8,10 @@ import (
"time"
"github.com/sagernet/sing-box/adapter"
"github.com/sagernet/sing-box/experimental/clashapi/trafficontrol"
"github.com/sagernet/sing-box/common/trafficcontrol"
C "github.com/sagernet/sing-box/constant"
"github.com/sagernet/sing/common"
F "github.com/sagernet/sing/common/format"
"github.com/sagernet/sing/common/json"
"github.com/sagernet/ws"
"github.com/sagernet/ws/wsutil"
@ -18,7 +21,7 @@ import (
"github.com/gofrs/uuid/v5"
)
func connectionRouter(ctx context.Context, network adapter.NetworkManager, trafficManager *trafficontrol.Manager) http.Handler {
func connectionRouter(ctx context.Context, network adapter.NetworkManager, trafficManager *trafficcontrol.Manager) http.Handler {
r := chi.NewRouter()
r.Get("/", getConnections(ctx, trafficManager))
r.Delete("/", closeAllConnections(network, trafficManager))
@ -26,11 +29,85 @@ func connectionRouter(ctx context.Context, network adapter.NetworkManager, traff
return r
}
func getConnections(ctx context.Context, trafficManager *trafficontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
func connectionsSnapshot(trafficManager *trafficcontrol.Manager) render.M {
uplinkTotal, downlinkTotal := trafficManager.Total()
connections := common.Filter(trafficManager.Connections(), func(metadata *trafficcontrol.TrackerMetadata) bool {
return metadata.OutboundType != C.TypeDNS
})
return render.M{
"downloadTotal": downlinkTotal,
"uploadTotal": uplinkTotal,
"connections": common.Map(connections, func(metadata *trafficcontrol.TrackerMetadata) connectionObject {
return connectionObject(*metadata)
}),
"memory": inuseMemory(),
}
}
type connectionObject trafficcontrol.TrackerMetadata
func (c connectionObject) MarshalJSON() ([]byte, error) {
var inbound string
if c.Metadata.Inbound != "" {
inbound = c.Metadata.InboundType + "/" + c.Metadata.Inbound
} else {
inbound = c.Metadata.InboundType
}
var domain string
if c.Metadata.Domain != "" {
domain = c.Metadata.Domain
} else {
domain = c.Metadata.Destination.Fqdn
}
var processPath string
if c.Metadata.ProcessInfo != nil {
if c.Metadata.ProcessInfo.ProcessPath != "" {
processPath = c.Metadata.ProcessInfo.ProcessPath
} else if len(c.Metadata.ProcessInfo.AndroidPackageNames) > 0 {
processPath = c.Metadata.ProcessInfo.AndroidPackageNames[0]
}
if processPath == "" {
if c.Metadata.ProcessInfo.UserId != -1 {
processPath = F.ToString(c.Metadata.ProcessInfo.UserId)
}
} else if c.Metadata.ProcessInfo.UserName != "" {
processPath = F.ToString(processPath, " (", c.Metadata.ProcessInfo.UserName, ")")
} else if c.Metadata.ProcessInfo.UserId != -1 {
processPath = F.ToString(processPath, " (", c.Metadata.ProcessInfo.UserId, ")")
}
}
var rule string
if c.Rule != nil {
rule = F.ToString(c.Rule, " => ", c.Rule.Action())
} else {
rule = "final"
}
return json.Marshal(map[string]any{
"id": c.ID,
"metadata": map[string]any{
"network": c.Metadata.Network,
"type": inbound,
"sourceIP": c.Metadata.Source.Addr,
"destinationIP": c.Metadata.Destination.Addr,
"sourcePort": F.ToString(c.Metadata.Source.Port),
"destinationPort": F.ToString(c.Metadata.Destination.Port),
"host": domain,
"dnsMode": "normal",
"processPath": processPath,
},
"upload": c.Upload.Load(),
"download": c.Download.Load(),
"start": c.CreatedAt,
"chains": c.Chain,
"rule": rule,
"rulePayload": "",
})
}
func getConnections(ctx context.Context, trafficManager *trafficcontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Upgrade") != "websocket" {
snapshot := trafficManager.Snapshot()
render.JSON(w, r, snapshot)
render.JSON(w, r, connectionsSnapshot(trafficManager))
return
}
@ -56,9 +133,9 @@ func getConnections(ctx context.Context, trafficManager *trafficontrol.Manager)
buf := &bytes.Buffer{}
sendSnapshot := func() error {
buf.Reset()
snapshot := trafficManager.Snapshot()
if err := json.NewEncoder(buf).Encode(snapshot); err != nil {
return err
encodeErr := json.NewEncoder(buf).Encode(connectionsSnapshot(trafficManager))
if encodeErr != nil {
return encodeErr
}
return wsutil.WriteServerText(conn, buf.Bytes())
}
@ -82,26 +159,20 @@ func getConnections(ctx context.Context, trafficManager *trafficontrol.Manager)
}
}
func closeConnection(trafficManager *trafficontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
func closeConnection(trafficManager *trafficcontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
id := uuid.FromStringOrNil(chi.URLParam(r, "id"))
snapshot := trafficManager.Snapshot()
for _, c := range snapshot.Connections {
if id == c.Metadata().ID {
c.Close()
break
}
targetConnection := trafficManager.Connection(id)
if targetConnection != nil {
targetConnection.Close()
}
render.NoContent(w, r)
}
}
func closeAllConnections(network adapter.NetworkManager, trafficManager *trafficontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
func closeAllConnections(network adapter.NetworkManager, trafficManager *trafficcontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
snapshot := trafficManager.Snapshot()
for _, c := range snapshot.Connections {
c.Close()
}
trafficManager.CloseAllConnections()
network.ResetNetwork()
render.NoContent(w, r)
}

View file

@ -14,17 +14,15 @@ import (
"github.com/sagernet/cors"
"github.com/sagernet/sing-box/adapter"
"github.com/sagernet/sing-box/common/trafficcontrol"
"github.com/sagernet/sing-box/common/urltest"
C "github.com/sagernet/sing-box/constant"
"github.com/sagernet/sing-box/experimental"
"github.com/sagernet/sing-box/experimental/clashapi/trafficontrol"
"github.com/sagernet/sing-box/log"
"github.com/sagernet/sing-box/option"
"github.com/sagernet/sing/common"
"github.com/sagernet/sing/common/cleanup"
E "github.com/sagernet/sing/common/exceptions"
"github.com/sagernet/sing/common/json"
N "github.com/sagernet/sing/common/network"
"github.com/sagernet/sing/common/observable"
"github.com/sagernet/sing/service"
"github.com/sagernet/sing/service/filemanager"
@ -50,10 +48,9 @@ type Server struct {
endpoint adapter.EndpointManager
logger log.Logger
httpServer *http.Server
trafficManager *trafficontrol.Manager
trafficManager *trafficcontrol.Manager
urlTestHistory adapter.URLTestHistoryStorage
logDebug bool
cleaner *cleanup.Cleaner
mode string
modeList []string
@ -66,7 +63,10 @@ type Server struct {
}
func NewServer(ctx context.Context, logFactory log.ObservableFactory, options option.ClashAPIOptions) (adapter.ClashServer, error) {
trafficManager := trafficontrol.NewManager()
trafficManager := service.PtrFromContext[trafficcontrol.Manager](ctx)
if trafficManager == nil {
return nil, E.New("missing traffic manager")
}
chiRouter := chi.NewRouter()
s := &Server{
ctx: ctx,
@ -86,7 +86,6 @@ func NewServer(ctx context.Context, logFactory log.ObservableFactory, options op
externalController: options.ExternalController != "",
externalUIDownloadURL: options.ExternalUIDownloadURL,
externalUIDownloadDetour: options.ExternalUIDownloadDetour,
cleaner: cleanup.Add(trafficManager.Clear),
}
s.urlTestHistory = service.FromContext[adapter.URLTestHistoryStorage](ctx)
if s.urlTestHistory == nil {
@ -196,9 +195,7 @@ func (s *Server) Start(stage adapter.StartStage) error {
func (s *Server) Close() error {
return common.Close(
common.PtrOrNil(s.httpServer),
s.trafficManager,
s.urlTestHistory,
common.PtrOrNil(s.cleaner),
)
}
@ -245,18 +242,6 @@ func (s *Server) HistoryStorage() adapter.URLTestHistoryStorage {
return s.urlTestHistory
}
func (s *Server) TrafficManager() *trafficontrol.Manager {
return s.trafficManager
}
func (s *Server) RoutedConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, matchedRule adapter.Rule, matchOutbound adapter.Outbound) net.Conn {
return trafficontrol.NewTCPTracker(conn, s.trafficManager, metadata, s.outbound, matchedRule, matchOutbound)
}
func (s *Server) RoutedPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, matchedRule adapter.Rule, matchOutbound adapter.Outbound) N.PacketConn {
return trafficontrol.NewUDPTracker(conn, s.trafficManager, metadata, s.outbound, matchedRule, matchOutbound)
}
func authentication(serverSecret string) func(next http.Handler) http.Handler {
return func(next http.Handler) http.Handler {
fn := func(w http.ResponseWriter, r *http.Request) {
@ -309,7 +294,7 @@ type Traffic struct {
Down int64 `json:"down"`
}
func traffic(ctx context.Context, trafficManager *trafficontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
func traffic(ctx context.Context, trafficManager *trafficcontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
var conn net.Conn
if r.Header.Get("Upgrade") == "websocket" {

View file

@ -1,182 +0,0 @@
package trafficontrol
import (
"runtime"
"sync"
"sync/atomic"
"time"
"github.com/sagernet/sing-box/common/compatible"
C "github.com/sagernet/sing-box/constant"
"github.com/sagernet/sing/common"
"github.com/sagernet/sing/common/json"
"github.com/sagernet/sing/common/observable"
"github.com/sagernet/sing/common/x/list"
"github.com/gofrs/uuid/v5"
)
type ConnectionEventType int
const (
ConnectionEventNew ConnectionEventType = iota
ConnectionEventUpdate
ConnectionEventClosed
)
type ConnectionEvent struct {
Type ConnectionEventType
ID uuid.UUID
Metadata *TrackerMetadata
UplinkDelta int64
DownlinkDelta int64
ClosedAt time.Time
}
const closedConnectionsLimit = 1000
type Manager struct {
uploadTotal atomic.Int64
downloadTotal atomic.Int64
connections compatible.Map[uuid.UUID, Tracker]
closedConnectionsAccess sync.Mutex
closedConnections list.List[TrackerMetadata]
memory uint64
eventSubscriber *observable.Subscriber[ConnectionEvent]
}
func NewManager() *Manager {
return &Manager{}
}
func (m *Manager) SetEventHook(subscriber *observable.Subscriber[ConnectionEvent]) {
m.eventSubscriber = subscriber
}
func (m *Manager) Join(c Tracker) {
metadata := c.Metadata()
m.connections.Store(metadata.ID, c)
if m.eventSubscriber != nil {
m.eventSubscriber.Emit(ConnectionEvent{
Type: ConnectionEventNew,
ID: metadata.ID,
Metadata: metadata,
})
}
}
func (m *Manager) Leave(c Tracker) {
metadata := c.Metadata()
_, loaded := m.connections.LoadAndDelete(metadata.ID)
if loaded {
closedAt := time.Now()
metadata.ClosedAt = closedAt
metadataCopy := *metadata
m.closedConnectionsAccess.Lock()
if m.closedConnections.Len() >= closedConnectionsLimit {
m.closedConnections.PopFront()
}
m.closedConnections.PushBack(metadataCopy)
m.closedConnectionsAccess.Unlock()
if m.eventSubscriber != nil {
m.eventSubscriber.Emit(ConnectionEvent{
Type: ConnectionEventClosed,
ID: metadata.ID,
Metadata: &metadataCopy,
ClosedAt: closedAt,
})
}
}
}
func (m *Manager) PushUploaded(size int64) {
m.uploadTotal.Add(size)
}
func (m *Manager) PushDownloaded(size int64) {
m.downloadTotal.Add(size)
}
func (m *Manager) Total() (up int64, down int64) {
return m.uploadTotal.Load(), m.downloadTotal.Load()
}
func (m *Manager) ConnectionsLen() int {
return m.connections.Len()
}
func (m *Manager) Connections() []*TrackerMetadata {
var connections []*TrackerMetadata
m.connections.Range(func(_ uuid.UUID, value Tracker) bool {
connections = append(connections, value.Metadata())
return true
})
return connections
}
func (m *Manager) ClosedConnections() []*TrackerMetadata {
m.closedConnectionsAccess.Lock()
values := m.closedConnections.Array()
m.closedConnectionsAccess.Unlock()
if len(values) == 0 {
return nil
}
connections := make([]*TrackerMetadata, len(values))
for i := range values {
connections[i] = &values[i]
}
return connections
}
func (m *Manager) Connection(id uuid.UUID) Tracker {
connection, loaded := m.connections.Load(id)
if !loaded {
return nil
}
return connection
}
func (m *Manager) Snapshot() *Snapshot {
var connections []Tracker
m.connections.Range(func(_ uuid.UUID, value Tracker) bool {
if value.Metadata().OutboundType != C.TypeDNS {
connections = append(connections, value)
}
return true
})
var memStats runtime.MemStats
runtime.ReadMemStats(&memStats)
m.memory = memStats.StackInuse + memStats.HeapInuse + memStats.HeapIdle - memStats.HeapReleased
return &Snapshot{
Upload: m.uploadTotal.Load(),
Download: m.downloadTotal.Load(),
Connections: connections,
Memory: m.memory,
}
}
func (m *Manager) Clear() {
m.closedConnectionsAccess.Lock()
defer m.closedConnectionsAccess.Unlock()
m.closedConnections.Init()
}
type Snapshot struct {
Download int64
Upload int64
Connections []Tracker
Memory uint64
}
func (s *Snapshot) MarshalJSON() ([]byte, error) {
return json.Marshal(map[string]any{
"downloadTotal": s.Download,
"uploadTotal": s.Upload,
"connections": common.Map(s.Connections, func(t Tracker) *TrackerMetadata { return t.Metadata() }),
"memory": s.Memory,
})
}

View file

@ -1,254 +0,0 @@
package trafficontrol
import (
"net"
"sync/atomic"
"time"
"github.com/sagernet/sing-box/adapter"
"github.com/sagernet/sing/common"
"github.com/sagernet/sing/common/bufio"
F "github.com/sagernet/sing/common/format"
"github.com/sagernet/sing/common/json"
N "github.com/sagernet/sing/common/network"
"github.com/gofrs/uuid/v5"
)
type TrackerMetadata struct {
ID uuid.UUID
Metadata adapter.InboundContext
CreatedAt time.Time
ClosedAt time.Time
Upload *atomic.Int64
Download *atomic.Int64
Chain []string
Rule adapter.Rule
Outbound string
OutboundType string
}
func (t TrackerMetadata) MarshalJSON() ([]byte, error) {
var inbound string
if t.Metadata.Inbound != "" {
inbound = t.Metadata.InboundType + "/" + t.Metadata.Inbound
} else {
inbound = t.Metadata.InboundType
}
var domain string
if t.Metadata.Domain != "" {
domain = t.Metadata.Domain
} else {
domain = t.Metadata.Destination.Fqdn
}
var processPath string
if t.Metadata.ProcessInfo != nil {
if t.Metadata.ProcessInfo.ProcessPath != "" {
processPath = t.Metadata.ProcessInfo.ProcessPath
} else if len(t.Metadata.ProcessInfo.AndroidPackageNames) > 0 {
processPath = t.Metadata.ProcessInfo.AndroidPackageNames[0]
}
if processPath == "" {
if t.Metadata.ProcessInfo.UserId != -1 {
processPath = F.ToString(t.Metadata.ProcessInfo.UserId)
}
} else if t.Metadata.ProcessInfo.UserName != "" {
processPath = F.ToString(processPath, " (", t.Metadata.ProcessInfo.UserName, ")")
} else if t.Metadata.ProcessInfo.UserId != -1 {
processPath = F.ToString(processPath, " (", t.Metadata.ProcessInfo.UserId, ")")
}
}
var rule string
if t.Rule != nil {
rule = F.ToString(t.Rule, " => ", t.Rule.Action())
} else {
rule = "final"
}
return json.Marshal(map[string]any{
"id": t.ID,
"metadata": map[string]any{
"network": t.Metadata.Network,
"type": inbound,
"sourceIP": t.Metadata.Source.Addr,
"destinationIP": t.Metadata.Destination.Addr,
"sourcePort": F.ToString(t.Metadata.Source.Port),
"destinationPort": F.ToString(t.Metadata.Destination.Port),
"host": domain,
"dnsMode": "normal",
"processPath": processPath,
},
"upload": t.Upload.Load(),
"download": t.Download.Load(),
"start": t.CreatedAt,
"chains": t.Chain,
"rule": rule,
"rulePayload": "",
})
}
type Tracker interface {
Metadata() *TrackerMetadata
Close() error
}
type TCPConn struct {
N.ExtendedConn
metadata TrackerMetadata
manager *Manager
}
func (tt *TCPConn) Metadata() *TrackerMetadata {
return &tt.metadata
}
func (tt *TCPConn) Close() error {
tt.manager.Leave(tt)
return tt.ExtendedConn.Close()
}
func (tt *TCPConn) Upstream() any {
return tt.ExtendedConn
}
func (tt *TCPConn) ReaderReplaceable() bool {
return true
}
func (tt *TCPConn) WriterReplaceable() bool {
return true
}
func NewTCPTracker(conn net.Conn, manager *Manager, metadata adapter.InboundContext, outboundManager adapter.OutboundManager, matchRule adapter.Rule, matchOutbound adapter.Outbound) *TCPConn {
id, _ := uuid.NewV4()
var (
chain []string
next string
outbound string
outboundType string
)
if matchOutbound != nil {
next = matchOutbound.Tag()
} else {
next = outboundManager.Default().Tag()
}
for {
detour, loaded := outboundManager.Outbound(next)
if !loaded {
break
}
chain = append(chain, next)
outbound = detour.Tag()
outboundType = detour.Type()
group, isGroup := detour.(adapter.OutboundGroup)
if !isGroup {
break
}
next = group.Now()
}
upload := new(atomic.Int64)
download := new(atomic.Int64)
tracker := &TCPConn{
ExtendedConn: bufio.NewCounterConn(conn, []N.CountFunc{func(n int64) {
upload.Add(n)
manager.PushUploaded(n)
}}, []N.CountFunc{func(n int64) {
download.Add(n)
manager.PushDownloaded(n)
}}),
metadata: TrackerMetadata{
ID: id,
Metadata: metadata,
CreatedAt: time.Now(),
Upload: upload,
Download: download,
Chain: common.Reverse(chain),
Rule: matchRule,
Outbound: outbound,
OutboundType: outboundType,
},
manager: manager,
}
manager.Join(tracker)
return tracker
}
type UDPConn struct {
N.PacketConn `json:"-"`
metadata TrackerMetadata
manager *Manager
}
func (ut *UDPConn) Metadata() *TrackerMetadata {
return &ut.metadata
}
func (ut *UDPConn) Close() error {
ut.manager.Leave(ut)
return ut.PacketConn.Close()
}
func (ut *UDPConn) Upstream() any {
return ut.PacketConn
}
func (ut *UDPConn) ReaderReplaceable() bool {
return true
}
func (ut *UDPConn) WriterReplaceable() bool {
return true
}
func NewUDPTracker(conn N.PacketConn, manager *Manager, metadata adapter.InboundContext, outboundManager adapter.OutboundManager, matchRule adapter.Rule, matchOutbound adapter.Outbound) *UDPConn {
id, _ := uuid.NewV4()
var (
chain []string
next string
outbound string
outboundType string
)
if matchOutbound != nil {
next = matchOutbound.Tag()
} else {
next = outboundManager.Default().Tag()
}
for {
detour, loaded := outboundManager.Outbound(next)
if !loaded {
break
}
chain = append(chain, next)
outbound = detour.Tag()
outboundType = detour.Type()
group, isGroup := detour.(adapter.OutboundGroup)
if !isGroup {
break
}
next = group.Now()
}
upload := new(atomic.Int64)
download := new(atomic.Int64)
trackerConn := &UDPConn{
PacketConn: bufio.NewCounterPacketConn(conn, []N.CountFunc{func(n int64) {
upload.Add(n)
manager.PushUploaded(n)
}}, []N.CountFunc{func(n int64) {
download.Add(n)
manager.PushDownloaded(n)
}}),
metadata: TrackerMetadata{
ID: id,
Metadata: metadata,
CreatedAt: time.Now(),
Upload: upload,
Download: download,
Chain: common.Reverse(chain),
Rule: matchRule,
Outbound: outbound,
OutboundType: outboundType,
},
manager: manager,
}
manager.Join(trackerConn)
return trackerConn
}

View file

@ -28,6 +28,7 @@ type CommandClient struct {
grpcClient daemon.StartedServiceClient
grpcManagedClient daemon.ManagedServiceClient
options CommandClientOptions
remote *remoteConnection
ctx context.Context
cancel context.CancelFunc
clientMutex sync.RWMutex
@ -147,23 +148,41 @@ func networkConnectionFromFileDescriptor(fileDescriptor int32) (net.Conn, error)
return networkConnection, nil
}
func (c *CommandClient) dialWithRetry(target string, contextDialer func(context.Context, string) (net.Conn, error), retryDial bool) (*grpc.ClientConn, daemon.StartedServiceClient, error) {
func localDialOptions(contextDialer func(context.Context, string) (net.Conn, error)) []grpc.DialOption {
options := []grpc.DialOption{
grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithUnaryInterceptor(unaryClientAuthInterceptor),
grpc.WithStreamInterceptor(streamClientAuthInterceptor),
}
if contextDialer != nil {
options = append(options, grpc.WithContextDialer(contextDialer))
}
return options
}
// establishConnection dials the command server the client is bound to: the
// local command server (over socket/XPC) or a remote API service.
func (c *CommandClient) establishConnection() (*grpc.ClientConn, daemon.StartedServiceClient, error) {
if c.remote != nil {
return c.dialRemote()
}
target, contextDialer := dialTarget()
return c.dialWithRetry(target, localDialOptions(contextDialer), true)
}
// dialWithRetry connects to the local command server. The retry loop exists to
// wait out the server starting up: WaitForReady keeps the probe redialing and
// the loop reissues it with a growing delay, so a freshly launched extension is
// picked up without surfacing a transient "unavailable" to the UI.
func (c *CommandClient) dialWithRetry(target string, dialOptions []grpc.DialOption, retryDial bool) (*grpc.ClientConn, daemon.StartedServiceClient, error) {
var connection *grpc.ClientConn
var client daemon.StartedServiceClient
var lastError error
for attempt := range commandClientDialAttempts {
if connection == nil {
options := []grpc.DialOption{
grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithUnaryInterceptor(unaryClientAuthInterceptor),
grpc.WithStreamInterceptor(streamClientAuthInterceptor),
}
if contextDialer != nil {
options = append(options, grpc.WithContextDialer(contextDialer))
}
var err error
connection, err = grpc.NewClient(target, options...)
connection, err = grpc.NewClient(target, dialOptions...)
if err != nil {
lastError = err
if !retryDial {
@ -174,8 +193,7 @@ func (c *CommandClient) dialWithRetry(target string, contextDialer func(context.
}
client = daemon.NewStartedServiceClient(connection)
}
waitDuration := commandClientDialDelay(attempt)
ctx, cancel := context.WithTimeout(context.Background(), waitDuration)
ctx, cancel := context.WithTimeout(context.Background(), commandClientDialDelay(attempt))
_, err := client.GetStartedAt(ctx, &emptypb.Empty{}, grpc.WaitForReady(true))
cancel()
if err == nil {
@ -190,12 +208,27 @@ func (c *CommandClient) dialWithRetry(target string, contextDialer func(context.
return nil, nil, E.Cause(lastError, "probe command server")
}
func (c *CommandClient) dialRemote() (*grpc.ClientConn, daemon.StartedServiceClient, error) {
connection, err := grpc.NewClient(c.remote.target, c.remote.dialOptions...)
if err != nil {
return nil, nil, E.Cause(err, "create remote command client")
}
client := daemon.NewStartedServiceClient(connection)
ctx, cancel := context.WithTimeout(context.Background(), commandClientRemoteProbeTimeout)
defer cancel()
_, err = client.GetStartedAt(ctx, &emptypb.Empty{})
if err != nil {
connection.Close()
return nil, nil, E.Cause(err, "connect to remote server")
}
return connection, client, nil
}
func (c *CommandClient) Connect() error {
c.clientMutex.Lock()
common.Close(common.PtrOrNil(c.grpcConn))
target, contextDialer := dialTarget()
connection, client, err := c.dialWithRetry(target, contextDialer, true)
connection, client, err := c.establishConnection()
if err != nil {
c.clientMutex.Unlock()
return err
@ -219,9 +252,9 @@ func (c *CommandClient) ConnectWithFD(fd int32) error {
c.clientMutex.Unlock()
return err
}
connection, client, err := c.dialWithRetry("passthrough:///xpc", func(ctx context.Context, _ string) (net.Conn, error) {
connection, client, err := c.dialWithRetry("passthrough:///xpc", localDialOptions(func(ctx context.Context, _ string) (net.Conn, error) {
return networkConnection, nil
}, false)
}), false)
if err != nil {
networkConnection.Close()
c.clientMutex.Unlock()
@ -283,8 +316,7 @@ func (c *CommandClient) getClientForCall() (daemon.StartedServiceClient, context
return c.grpcClient, c.ctx, nil
}
target, contextDialer := dialTarget()
connection, client, err := c.dialWithRetry(target, contextDialer, true)
connection, client, err := c.establishConnection()
if err != nil {
return nil, nil, E.Cause(err, "get command client")
}

View file

@ -0,0 +1,101 @@
package libbox
import (
"context"
"crypto/tls"
"net"
"net/url"
"strings"
"time"
E "github.com/sagernet/sing/common/exceptions"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/metadata"
)
type RemoteConnectionOptions struct {
URL string
Secret string
}
const commandClientRemoteProbeTimeout = 10 * time.Second
type remoteConnection struct {
target string
dialOptions []grpc.DialOption
}
func newRemoteConnection(options *RemoteConnectionOptions) (*remoteConnection, error) {
if options == nil {
return nil, E.New("missing remote connection options")
}
urlString := options.URL
if !strings.Contains(urlString, "://") {
urlString = "http://" + urlString
}
serverURL, err := url.Parse(urlString)
if err != nil {
return nil, E.Cause(err, "parse server URL")
}
host := serverURL.Hostname()
if host == "" {
return nil, E.New("missing host in server URL: ", options.URL)
}
var (
transportCredentials credentials.TransportCredentials
defaultPort string
)
switch serverURL.Scheme {
case "http":
transportCredentials = insecure.NewCredentials()
defaultPort = "80"
case "https":
transportCredentials = credentials.NewTLS(&tls.Config{ServerName: host})
defaultPort = "443"
default:
return nil, E.New("unsupported server URL scheme: ", serverURL.Scheme, ", expected http or https")
}
port := serverURL.Port()
if port == "" {
port = defaultPort
}
dialOptions := []grpc.DialOption{
grpc.WithTransportCredentials(transportCredentials),
}
if options.Secret != "" {
authorization := "Bearer " + options.Secret
dialOptions = append(dialOptions,
grpc.WithUnaryInterceptor(func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
return invoker(metadata.AppendToOutgoingContext(ctx, "authorization", authorization), method, req, reply, cc, opts...)
}),
grpc.WithStreamInterceptor(func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
return streamer(metadata.AppendToOutgoingContext(ctx, "authorization", authorization), desc, cc, method, opts...)
}),
)
}
return &remoteConnection{
target: net.JoinHostPort(host, port),
dialOptions: dialOptions,
}, nil
}
func NewRemoteCommandClient(handler CommandClientHandler, options *CommandClientOptions, remoteOptions *RemoteConnectionOptions) (*CommandClient, error) {
remote, err := newRemoteConnection(remoteOptions)
if err != nil {
return nil, err
}
client := NewCommandClient(handler, options)
client.remote = remote
return client, nil
}
func NewStandaloneRemoteCommandClient(remoteOptions *RemoteConnectionOptions) (*CommandClient, error) {
remote, err := newRemoteConnection(remoteOptions)
if err != nil {
return nil, err
}
return &CommandClient{remote: remote}, nil
}

View file

@ -36,6 +36,7 @@ import (
"github.com/sagernet/sing-box/protocol/tun"
"github.com/sagernet/sing-box/protocol/vless"
"github.com/sagernet/sing-box/protocol/vmess"
"github.com/sagernet/sing-box/service/api"
originca "github.com/sagernet/sing-box/service/origin_ca"
"github.com/sagernet/sing-box/service/resolved"
"github.com/sagernet/sing-box/service/ssmapi"
@ -133,6 +134,7 @@ func DNSTransportRegistry() *dns.TransportRegistry {
func ServiceRegistry() *service.Registry {
registry := service.NewRegistry()
api.RegisterService(registry)
resolved.RegisterService(registry)
ssmapi.RegisterService(registry)

View file

@ -22,6 +22,7 @@ type Factory interface {
type ObservableFactory interface {
Factory
observable.Observable[Entry]
AttachPlatformWriter(writer PlatformWriter)
}
type Entry struct {

View file

@ -80,6 +80,9 @@ func (f *nopFactory) FatalContext(ctx context.Context, args ...any) {
func (f *nopFactory) PanicContext(ctx context.Context, args ...any) {
}
func (f *nopFactory) AttachPlatformWriter(writer PlatformWriter) {
}
func (f *nopFactory) Subscribe() (subscription observable.Subscription[Entry], done <-chan struct{}, err error) {
return nil, nil, os.ErrInvalid
}

View file

@ -4,6 +4,7 @@ import (
"context"
"io"
"os"
"sync/atomic"
"time"
"github.com/sagernet/sing/common"
@ -12,7 +13,7 @@ import (
"github.com/sagernet/sing/service/filemanager"
)
var _ Factory = (*defaultFactory)(nil)
var _ ObservableFactory = (*defaultFactory)(nil)
type defaultFactory struct {
ctx context.Context
@ -21,7 +22,7 @@ type defaultFactory struct {
writer io.Writer
file *os.File
filePath string
platformWriter PlatformWriter
platformWriters atomic.Pointer[[]PlatformWriter]
needObservable bool
level Level
subscriber *observable.Subscriber[Entry]
@ -45,11 +46,13 @@ func NewDefaultFactory(
},
writer: writer,
filePath: filePath,
platformWriter: platformWriter,
needObservable: needObservable,
level: LevelTrace,
subscriber: observable.NewSubscriber[Entry](128),
}
if platformWriter != nil {
factory.platformWriters.Store(&[]PlatformWriter{platformWriter})
}
/*if platformWriter != nil {
factory.platformFormatter.DisableColors = platformWriter.DisableColors()
}*/
@ -78,6 +81,19 @@ func (f *defaultFactory) Close() error {
)
}
func (f *defaultFactory) AttachPlatformWriter(writer PlatformWriter) {
writers := append(f.loadPlatformWriters(), writer)
f.platformWriters.Store(&writers)
}
func (f *defaultFactory) loadPlatformWriters() []PlatformWriter {
writers := f.platformWriters.Load()
if writers == nil {
return nil
}
return *writers
}
func (f *defaultFactory) Level() Level {
return f.level
}
@ -111,7 +127,8 @@ type observableLogger struct {
func (l *observableLogger) Log(ctx context.Context, level Level, args []any) {
level = OverrideLevelFromContext(level, ctx)
if level > l.level && l.platformWriter == nil && !l.needObservable {
platformWriters := l.loadPlatformWriters()
if level > l.level && len(platformWriters) == 0 && !l.needObservable {
return
}
nowTime := time.Now()
@ -137,8 +154,11 @@ func (l *observableLogger) Log(ctx context.Context, level Level, args []any) {
os.Exit(1)
}
}
if l.platformWriter != nil {
l.platformWriter.WriteMessage(level, l.platformFormatter.Format(ctx, level, l.tag, F.ToString(args...), nowTime))
if len(platformWriters) > 0 {
message := l.platformFormatter.Format(ctx, level, l.tag, F.ToString(args...), nowTime)
for _, platformWriter := range platformWriters {
platformWriter.WriteMessage(level, message)
}
}
}

View file

@ -187,6 +187,7 @@ nav:
- URLTest: configuration/outbound/urltest.md
- Service:
- configuration/service/index.md
- sing-box API: configuration/service/api.md
- DERP: configuration/service/derp.md
- Resolved: configuration/service/resolved.md
- SSM API: configuration/service/ssm-api.md

11
option/api.go Normal file
View file

@ -0,0 +1,11 @@
package option
import "github.com/sagernet/sing/common/json/badoption"
type APIServiceOptions struct {
ListenOptions
Secret string `json:"secret,omitempty"`
AccessControlAllowOrigin badoption.Listable[string] `json:"access_control_allow_origin,omitempty"`
AccessControlAllowPrivateNetwork bool `json:"access_control_allow_private_network,omitempty"`
InboundTLSOptionsContainer
}

124
service/api/server.go Normal file
View file

@ -0,0 +1,124 @@
package api
import (
"context"
"net"
"net/http"
"github.com/sagernet/sing-box/adapter"
boxService "github.com/sagernet/sing-box/adapter/service"
"github.com/sagernet/sing-box/common/listener"
"github.com/sagernet/sing-box/common/tls"
C "github.com/sagernet/sing-box/constant"
"github.com/sagernet/sing-box/daemon"
"github.com/sagernet/sing-box/log"
"github.com/sagernet/sing-box/option"
"github.com/sagernet/sing/common"
E "github.com/sagernet/sing/common/exceptions"
N "github.com/sagernet/sing/common/network"
aTLS "github.com/sagernet/sing/common/tls"
"golang.org/x/net/http2"
"golang.org/x/net/http2/h2c"
"google.golang.org/grpc"
)
func RegisterService(registry *boxService.Registry) {
boxService.Register[option.APIServiceOptions](registry, C.TypeAPI, NewService)
}
type Service struct {
boxService.Adapter
ctx context.Context
cancel context.CancelFunc
logger log.ContextLogger
options option.APIServiceOptions
listener *listener.Listener
tlsConfig tls.ServerConfig
startedService *daemon.StartedService
grpcServer *grpc.Server
httpServer *http.Server
}
func NewService(ctx context.Context, logger log.ContextLogger, tag string, options option.APIServiceOptions) (adapter.Service, error) {
ctx, cancel := context.WithCancel(ctx)
s := &Service{
Adapter: boxService.NewAdapter(C.TypeAPI, tag),
ctx: ctx,
cancel: cancel,
logger: logger,
options: options,
listener: listener.New(listener.Options{
Context: ctx,
Logger: logger,
Network: []string{N.NetworkTCP},
Listen: options.ListenOptions,
}),
}
if options.TLS != nil {
tlsConfig, err := tls.NewServer(ctx, logger, common.PtrValueOrDefault(options.TLS))
if err != nil {
cancel()
return nil, err
}
s.tlsConfig = tlsConfig
}
return s, nil
}
func (s *Service) Start(stage adapter.StartStage) error {
if stage != adapter.StartStateStarted {
return nil
}
s.startedService = daemon.NewAttachedService(s.ctx)
s.grpcServer = daemon.NewServer(s.startedService, s.options.Secret)
s.httpServer = &http.Server{
Handler: h2c.NewHandler(newHTTPHandler(s.logger, s.grpcServer, s.options), new(http2.Server)),
BaseContext: func(net.Listener) context.Context {
return s.ctx
},
}
if s.tlsConfig != nil {
err := s.tlsConfig.Start()
if err != nil {
return E.Cause(err, "create TLS config")
}
if !common.Contains(s.tlsConfig.NextProtos(), http2.NextProtoTLS) {
s.tlsConfig.SetNextProtos(append([]string{http2.NextProtoTLS}, s.tlsConfig.NextProtos()...))
}
if !common.Contains(s.tlsConfig.NextProtos(), "http/1.1") {
s.tlsConfig.SetNextProtos(append(s.tlsConfig.NextProtos(), "http/1.1"))
}
}
tcpListener, err := s.listener.ListenTCP()
if err != nil {
return err
}
if s.tlsConfig != nil {
tcpListener = aTLS.NewListener(tcpListener, s.tlsConfig)
}
go func() {
serveErr := s.httpServer.Serve(tcpListener)
if serveErr != nil && s.ctx.Err() == nil {
s.logger.Error("serve error: ", serveErr)
}
}()
return nil
}
func (s *Service) Close() error {
s.cancel()
if s.httpServer != nil {
s.httpServer.Close()
}
if s.grpcServer != nil {
s.grpcServer.Stop()
}
if s.startedService != nil {
s.startedService.Close()
}
return common.Close(
common.PtrOrNil(s.listener),
s.tlsConfig,
)
}

235
service/api/web_bridge.go Normal file
View file

@ -0,0 +1,235 @@
package api
import (
"bytes"
"encoding/base64"
"encoding/binary"
"io"
"net/http"
"strings"
"github.com/sagernet/cors"
"github.com/sagernet/sing-box/log"
"github.com/sagernet/sing-box/option"
"golang.org/x/net/http2"
"google.golang.org/grpc"
)
const (
contentTypeGRPC = "application/grpc"
contentTypeGRPCWeb = "application/grpc-web"
contentTypeGRPCWebText = "application/grpc-web-text"
)
// newHTTPHandler additionally accepts gRPC-Web requests
// (https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-WEB.md) and gRPC-Web
// streams over WebSocket, wire compatible with the improbable-eng/grpc-web
// client transports.
func newHTTPHandler(logger log.ContextLogger, grpcServer *grpc.Server, options option.APIServiceOptions) http.Handler {
allowedOrigins := options.AccessControlAllowOrigin
if len(allowedOrigins) == 0 {
allowedOrigins = []string{"*"}
}
corsHandler := cors.New(cors.Options{
AllowedOrigins: allowedOrigins,
AllowedMethods: []string{http.MethodPost, http.MethodOptions},
AllowedHeaders: []string{"Content-Type", "Authorization", "X-Grpc-Web", "X-User-Agent", "Grpc-Timeout"},
ExposedHeaders: []string{"Grpc-Status", "Grpc-Message", "Grpc-Status-Details-Bin"},
AllowPrivateNetwork: options.AccessControlAllowPrivateNetwork,
MaxAge: 300,
})
return corsHandler.Handler(&webBridge{
logger: logger,
grpcServer: grpcServer,
})
}
type webBridge struct {
logger log.ContextLogger
grpcServer *grpc.Server
}
func (b *webBridge) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
contentType := request.Header.Get("Content-Type")
switch {
case isWebSocketGRPCRequest(request):
b.serveWebSocket(writer, request)
case request.Method == http.MethodPost && strings.HasPrefix(contentType, contentTypeGRPCWeb):
b.serveWeb(writer, request)
case request.ProtoMajor == 2 && strings.HasPrefix(contentType, contentTypeGRPC):
b.grpcServer.ServeHTTP(writer, request)
default:
http.NotFound(writer, request)
}
}
func (b *webBridge) serveWeb(writer http.ResponseWriter, request *http.Request) {
isTextFormat := strings.HasPrefix(request.Header.Get("Content-Type"), contentTypeGRPCWebText)
webContentType := contentTypeGRPCWeb
grpcRequest := request.Clone(request.Context())
if isTextFormat {
webContentType = contentTypeGRPCWebText
grpcRequest.Body = &bodyReadCloser{
Reader: base64.NewDecoder(base64.StdEncoding, request.Body),
Closer: request.Body,
}
}
// The gRPC server handler transport only accepts requests it sees as
// native gRPC over HTTP/2.
grpcRequest.ProtoMajor = 2
grpcRequest.ProtoMinor = 0
grpcRequest.Header.Set("Content-Type", strings.Replace(request.Header.Get("Content-Type"), webContentType, contentTypeGRPC, 1))
grpcRequest.Header.Del("Content-Length")
response := newWebResponseWriter(writer, isTextFormat)
b.grpcServer.ServeHTTP(response, grpcRequest)
response.finish()
}
type bodyReadCloser struct {
io.Reader
io.Closer
}
// webResponseWriter translates a native gRPC response into a gRPC-Web
// response: headers set after the first write, including the gRPC status the
// handler transport sets via http2.TrailerPrefix keys, become a trailer
// frame at the end of the body instead of HTTP trailers.
type webResponseWriter struct {
writer http.ResponseWriter
rawWriter http.ResponseWriter
header http.Header
contentType string
wroteHeaders bool
wroteBody bool
}
func newWebResponseWriter(writer http.ResponseWriter, isTextFormat bool) *webResponseWriter {
response := &webResponseWriter{
writer: writer,
rawWriter: writer,
header: make(http.Header),
contentType: contentTypeGRPCWeb,
}
if isTextFormat {
response.writer = newBase64ResponseWriter(writer)
response.contentType = contentTypeGRPCWebText
}
return response
}
func (w *webResponseWriter) Header() http.Header {
return w.header
}
func (w *webResponseWriter) Write(content []byte) (int, error) {
if !w.wroteHeaders {
w.prepareHeaders()
w.wroteHeaders = true
}
w.wroteBody = true
return w.writer.Write(content)
}
func (w *webResponseWriter) WriteHeader(statusCode int) {
if !w.wroteHeaders {
w.prepareHeaders()
w.wroteHeaders = true
}
w.writer.WriteHeader(statusCode)
}
func (w *webResponseWriter) Flush() {
// Flushing before anything was written would commit a 200 response
// even for requests that end up as trailers-only responses.
if w.wroteHeaders || w.wroteBody {
flushWriter(w.writer)
}
}
func (w *webResponseWriter) prepareHeaders() {
rawHeader := w.rawWriter.Header()
for key, values := range w.header {
canonicalKey := http.CanonicalHeaderKey(strings.TrimPrefix(key, http2.TrailerPrefix))
if canonicalKey == "Trailer" {
continue
}
if canonicalKey == "Content-Type" {
newValues := make([]string, 0, len(values))
for _, value := range values {
newValues = append(newValues, strings.Replace(value, contentTypeGRPC, w.contentType, 1))
}
values = newValues
}
rawHeader[canonicalKey] = values
}
}
func (w *webResponseWriter) finish() {
if w.wroteHeaders || w.wroteBody {
w.writeTrailerFrame()
} else {
w.WriteHeader(http.StatusOK)
flushWriter(w.writer)
}
}
func (w *webResponseWriter) writeTrailerFrame() {
flushedKeys := make(map[string]bool)
for key := range w.rawWriter.Header() {
flushedKeys[strings.ToLower(key)] = true
}
trailerHeader := make(http.Header)
for key, values := range w.header {
lowerKey := strings.ToLower(strings.TrimPrefix(key, http2.TrailerPrefix))
if lowerKey == "trailer" || flushedKeys[lowerKey] {
continue
}
trailerHeader[lowerKey] = values
}
var trailerBuffer bytes.Buffer
trailerHeader.Write(&trailerBuffer)
w.writer.Write(webMetadataFrameHeader(trailerBuffer.Len()))
w.writer.Write(trailerBuffer.Bytes())
flushWriter(w.writer)
}
func webMetadataFrameHeader(payloadLength int) []byte {
return binary.BigEndian.AppendUint32([]byte{1 << 7}, uint32(payloadLength))
}
func flushWriter(writer http.ResponseWriter) {
flusher, isFlusher := writer.(http.Flusher)
if isFlusher {
flusher.Flush()
}
}
type base64ResponseWriter struct {
wrapped http.ResponseWriter
encoder io.WriteCloser
}
func newBase64ResponseWriter(wrapped http.ResponseWriter) http.ResponseWriter {
writer := &base64ResponseWriter{wrapped: wrapped}
writer.encoder = base64.NewEncoder(base64.StdEncoding, wrapped)
return writer
}
func (w *base64ResponseWriter) Header() http.Header {
return w.wrapped.Header()
}
func (w *base64ResponseWriter) Write(content []byte) (int, error) {
return w.encoder.Write(content)
}
func (w *base64ResponseWriter) WriteHeader(statusCode int) {
w.wrapped.WriteHeader(statusCode)
}
func (w *base64ResponseWriter) Flush() {
w.encoder.Close()
w.encoder = base64.NewEncoder(base64.StdEncoding, w.wrapped)
flushWriter(w.wrapped)
}

View file

@ -0,0 +1,249 @@
package api
import (
"bufio"
"bytes"
"context"
"io"
"net/http"
"net/textproto"
"strings"
"time"
E "github.com/sagernet/sing/common/exceptions"
"github.com/coder/websocket"
"golang.org/x/net/http/httpguts"
"golang.org/x/net/http2"
)
const (
webSocketSubprotocol = "grpc-websockets"
webSocketReadLimit = 1 << 22
webSocketPingInterval = 30 * time.Second
)
func isWebSocketGRPCRequest(request *http.Request) bool {
return httpguts.HeaderValuesContainsToken(request.Header.Values("Upgrade"), "websocket") &&
httpguts.HeaderValuesContainsToken(request.Header.Values("Sec-Websocket-Protocol"), webSocketSubprotocol)
}
// serveWebSocket carries a single gRPC stream over a WebSocket connection:
// the first client message contains the request metadata, each subsequent
// binary message is prefixed with 0 for body data or is a single 1 byte for
// the half-close signal, and the server sends gRPC-Web frames back.
func (b *webBridge) serveWebSocket(writer http.ResponseWriter, request *http.Request) {
conn, err := websocket.Accept(writer, request, &websocket.AcceptOptions{
Subprotocols: []string{webSocketSubprotocol},
InsecureSkipVerify: true,
})
if err != nil {
b.logger.Error("upgrade websocket request: ", err)
return
}
conn.SetReadLimit(webSocketReadLimit)
ctx, cancel := context.WithCancel(request.Context())
defer cancel()
messageType, firstMessage, err := conn.Read(ctx)
if err != nil {
conn.CloseNow()
return
}
if messageType != websocket.MessageBinary {
conn.CloseNow()
return
}
header, err := parseWebSocketHeader(firstMessage)
if err != nil {
b.logger.Error("parse websocket request metadata: ", err)
conn.CloseNow()
return
}
contentType := header.Get("Content-Type")
if contentType == "" {
header.Set("Content-Type", contentTypeGRPC)
} else {
header.Set("Content-Type", strings.Replace(contentType, contentTypeGRPCWeb, contentTypeGRPC, 1))
}
header.Del("Content-Length")
response := newWebSocketResponseWriter(ctx, conn)
grpcRequest := request.WithContext(ctx)
grpcRequest.Method = http.MethodPost
grpcRequest.ProtoMajor = 2
grpcRequest.ProtoMinor = 0
grpcRequest.Header = header
grpcRequest.Body = &webSocketBodyReader{
ctx: ctx,
cancel: cancel,
conn: conn,
response: response,
}
go keepWebSocketAlive(ctx, conn)
b.grpcServer.ServeHTTP(response, grpcRequest)
response.writeTrailerFrame()
conn.Close(websocket.StatusNormalClosure, "")
}
func parseWebSocketHeader(content []byte) (http.Header, error) {
reader := textproto.NewReader(bufio.NewReader(io.MultiReader(bytes.NewReader(content), strings.NewReader("\r\n"))))
mimeHeader, err := reader.ReadMIMEHeader()
if err != nil {
return nil, err
}
return http.Header(mimeHeader), nil
}
func keepWebSocketAlive(ctx context.Context, conn *websocket.Conn) {
ticker := time.NewTicker(webSocketPingInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
err := conn.Ping(ctx)
if err != nil {
return
}
}
}
}
type webSocketResponseWriter struct {
ctx context.Context
conn *websocket.Conn
header http.Header
flushedHeader http.Header
wroteHeaders bool
wroteTrailers bool
}
func newWebSocketResponseWriter(ctx context.Context, conn *websocket.Conn) *webSocketResponseWriter {
return &webSocketResponseWriter{
ctx: ctx,
conn: conn,
header: make(http.Header),
flushedHeader: make(http.Header),
}
}
func (w *webSocketResponseWriter) Header() http.Header {
return w.header
}
func (w *webSocketResponseWriter) Write(content []byte) (int, error) {
if !w.wroteHeaders {
w.WriteHeader(http.StatusOK)
}
err := w.conn.Write(w.ctx, websocket.MessageBinary, content)
if err != nil {
return 0, err
}
return len(content), nil
}
func (w *webSocketResponseWriter) WriteHeader(statusCode int) {
if w.wroteHeaders {
return
}
w.wroteHeaders = true
headerFrame := make(http.Header)
for key, values := range w.header {
canonicalKey := http.CanonicalHeaderKey(key)
if canonicalKey == "Trailer" {
continue
}
w.flushedHeader[canonicalKey] = values
headerFrame[canonicalKey] = values
}
w.writeHeaderFrame(headerFrame)
}
func (w *webSocketResponseWriter) Flush() {
}
func (w *webSocketResponseWriter) writeHeaderFrame(header http.Header) {
var headerBuffer bytes.Buffer
header.Write(&headerBuffer)
frame := make([]byte, 0, 5+headerBuffer.Len())
frame = append(frame, webMetadataFrameHeader(headerBuffer.Len())...)
frame = append(frame, headerBuffer.Bytes()...)
w.conn.Write(w.ctx, websocket.MessageBinary, frame)
}
func (w *webSocketResponseWriter) writeTrailerFrame() {
if w.wroteTrailers {
return
}
w.wroteTrailers = true
trailerHeader := make(http.Header)
for key, values := range w.header {
lowerKey := strings.ToLower(strings.TrimPrefix(key, http2.TrailerPrefix))
if lowerKey == "trailer" {
continue
}
_, flushed := w.flushedHeader[http.CanonicalHeaderKey(lowerKey)]
if flushed {
continue
}
trailerHeader[lowerKey] = values
}
w.writeHeaderFrame(trailerHeader)
}
type webSocketBodyReader struct {
ctx context.Context
cancel context.CancelFunc
conn *websocket.Conn
response *webSocketResponseWriter
remaining []byte
}
func (r *webSocketBodyReader) Read(buffer []byte) (int, error) {
if len(r.remaining) > 0 {
n := copy(buffer, r.remaining)
r.remaining = r.remaining[n:]
return n, nil
}
for {
messageType, payload, err := r.conn.Read(r.ctx)
if err != nil {
r.cancel()
return 0, io.EOF
}
if messageType != websocket.MessageBinary {
return 0, E.New("unexpected non-binary websocket message")
}
if len(payload) == 0 {
continue
}
if payload[0] == 1 {
go r.waitForClose()
return 0, io.EOF
}
content := payload[1:]
if len(content) == 0 {
continue
}
n := copy(buffer, content)
r.remaining = content[n:]
return n, nil
}
}
func (r *webSocketBodyReader) waitForClose() {
for {
_, _, err := r.conn.Read(r.ctx)
if err != nil {
r.cancel()
return
}
}
}
// Close is called by the gRPC handler transport after the stream status has
// been written; the trailer frame must be sent before the connection closes.
func (r *webSocketBodyReader) Close() error {
r.response.writeTrailerFrame()
return r.conn.Close(websocket.StatusNormalClosure, "")
}