mirror of
https://github.com/SagerNet/sing-box.git
synced 2026-09-29 16:14:30 +00:00
Fix outbound group resolution and interruption for nested groups
This commit is contained in:
parent
cabfdc7134
commit
a063e9b10a
13 changed files with 206 additions and 178 deletions
|
|
@ -151,8 +151,9 @@ func (s *SavedBinary) UnmarshalBinary(data []byte) error {
|
|||
|
||||
type OutboundGroup interface {
|
||||
Outbound
|
||||
Now() string
|
||||
All() []string
|
||||
Selected(network string) Outbound
|
||||
AttachConnection(closer io.Closer) (detach func())
|
||||
}
|
||||
|
||||
type URLTestGroup interface {
|
||||
|
|
|
|||
|
|
@ -58,6 +58,7 @@ type InboundContext struct {
|
|||
|
||||
RouteRule string
|
||||
RouteOutbound string
|
||||
OutboundChain []Outbound
|
||||
|
||||
// sniffer
|
||||
|
||||
|
|
|
|||
2
box.go
2
box.go
|
|
@ -246,7 +246,7 @@ func New(options Options) (*Box, error) {
|
|||
return nil, E.Cause(err, "initialize router")
|
||||
}
|
||||
if needClashAPI || needAPIService || options.PlatformLogWriter != nil {
|
||||
trafficManager := trafficcontrol.NewManager(outboundManager)
|
||||
trafficManager := trafficcontrol.NewManager()
|
||||
service.MustRegisterPtr(ctx, trafficManager)
|
||||
router.AppendTracker(trafficManager)
|
||||
internalServices = append(internalServices, trafficManager)
|
||||
|
|
|
|||
|
|
@ -519,6 +519,8 @@ func (d *DefaultDialer) dialAttribution(ctx context.Context, destination M.Socks
|
|||
}
|
||||
attribution.Rule = metadata.RouteRule
|
||||
attribution.Outbound = metadata.Outbound
|
||||
attribution.Chain = common.Map(metadata.OutboundChain, adapter.Outbound.Tag)
|
||||
slices.Reverse(attribution.Chain)
|
||||
if d.outboundManager != nil {
|
||||
if metadata.Outbound != "" {
|
||||
outbound, loaded := d.outboundManager.Outbound(metadata.Outbound)
|
||||
|
|
@ -526,9 +528,6 @@ func (d *DefaultDialer) dialAttribution(ctx context.Context, destination M.Socks
|
|||
attribution.OutboundType = outbound.Type()
|
||||
}
|
||||
}
|
||||
if metadata.RouteOutbound != "" {
|
||||
attribution.Chain = d.outboundChain(metadata.RouteOutbound)
|
||||
}
|
||||
}
|
||||
if metadata.Destination.IsValid() {
|
||||
attribution.Destination = metadata.Destination.String()
|
||||
|
|
@ -540,22 +539,3 @@ func (d *DefaultDialer) dialAttribution(ctx context.Context, destination M.Socks
|
|||
}
|
||||
return attribution
|
||||
}
|
||||
|
||||
func (d *DefaultDialer) outboundChain(head string) []string {
|
||||
var chain []string
|
||||
next := head
|
||||
for {
|
||||
detour, loaded := d.outboundManager.Outbound(next)
|
||||
if !loaded {
|
||||
break
|
||||
}
|
||||
chain = append(chain, next)
|
||||
outboundGroup, isGroup := detour.(adapter.OutboundGroup)
|
||||
if !isGroup {
|
||||
break
|
||||
}
|
||||
next = outboundGroup.Now()
|
||||
}
|
||||
slices.Reverse(chain)
|
||||
return chain
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,6 +22,17 @@ func NewGroup() *Group {
|
|||
return &Group{}
|
||||
}
|
||||
|
||||
func (g *Group) Add(closer io.Closer, isExternal bool) (remove func()) {
|
||||
g.access.Lock()
|
||||
defer g.access.Unlock()
|
||||
element := g.connections.PushBack(&groupConnItem{closer, isExternal})
|
||||
return func() {
|
||||
g.access.Lock()
|
||||
defer g.access.Unlock()
|
||||
g.connections.Remove(element)
|
||||
}
|
||||
}
|
||||
|
||||
func (g *Group) NewConn(conn net.Conn, isExternal bool) net.Conn {
|
||||
g.access.Lock()
|
||||
defer g.access.Unlock()
|
||||
|
|
|
|||
|
|
@ -35,8 +35,6 @@ var (
|
|||
)
|
||||
|
||||
type Manager struct {
|
||||
outbound adapter.OutboundManager
|
||||
|
||||
connections compatible.Map[uuid.UUID, Tracker]
|
||||
closedConnectionsAccess sync.Mutex
|
||||
closedConnections list.List[TrackerMetadata]
|
||||
|
|
@ -48,9 +46,8 @@ type Manager struct {
|
|||
cleaner *cleanup.Cleaner
|
||||
}
|
||||
|
||||
func NewManager(outbound adapter.OutboundManager) *Manager {
|
||||
func NewManager() *Manager {
|
||||
return &Manager{
|
||||
outbound: outbound,
|
||||
eventSubscriber: observable.NewSubscriber[ConnectionEvent](256),
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ package trafficcontrol
|
|||
import (
|
||||
"context"
|
||||
"net"
|
||||
"slices"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
|
|
@ -66,41 +67,19 @@ func (m *Manager) RoutedFlow(ctx context.Context, metadata adapter.InboundContex
|
|||
|
||||
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()
|
||||
}
|
||||
chain := common.Map(metadata.OutboundChain, adapter.Outbound.Tag)
|
||||
slices.Reverse(chain)
|
||||
outbound := metadata.OutboundChain[len(metadata.OutboundChain)-1]
|
||||
return TrackerMetadata{
|
||||
ID: id,
|
||||
Metadata: metadata,
|
||||
CreatedAt: time.Now(),
|
||||
Upload: upload,
|
||||
Download: download,
|
||||
Chain: common.Reverse(chain),
|
||||
Chain: chain,
|
||||
Rule: matchedRule,
|
||||
Outbound: outbound,
|
||||
OutboundType: outboundType,
|
||||
Outbound: outbound.Tag(),
|
||||
OutboundType: outbound.Type(),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ import (
|
|||
"github.com/sagernet/sing-box/service/oomkiller"
|
||||
"github.com/sagernet/sing/common"
|
||||
"github.com/sagernet/sing/common/memory"
|
||||
N "github.com/sagernet/sing/common/network"
|
||||
"github.com/sagernet/sing/common/observable"
|
||||
"github.com/sagernet/sing/service"
|
||||
|
||||
|
|
@ -601,7 +602,9 @@ func (s *StartedService) readGroups() *Groups {
|
|||
g.Tag = iGroup.Tag()
|
||||
g.Type = iGroup.Type()
|
||||
_, g.Selectable = iGroup.(*group.Selector)
|
||||
g.Selected = iGroup.Now()
|
||||
if selected := iGroup.Selected(N.NetworkTCP); selected != nil {
|
||||
g.Selected = selected.Tag()
|
||||
}
|
||||
if boxService.cacheFile != nil {
|
||||
if isExpand, loaded := boxService.cacheFile.LoadGroupExpand(g.Tag); loaded {
|
||||
g.IsExpand = isExpand
|
||||
|
|
@ -617,7 +620,7 @@ func (s *StartedService) readGroups() *Groups {
|
|||
var item GroupItem
|
||||
item.Tag = itemTag
|
||||
item.Type = itemOutbound.Type()
|
||||
if history := historyStorage.LoadURLTestHistory(group.RealTag(boxService.outboundManager, itemOutbound)); history != nil {
|
||||
if history := historyStorage.LoadURLTestHistory(group.RealTag(itemOutbound, N.NetworkTCP)); history != nil {
|
||||
item.UrlTestTime = history.Time.Unix()
|
||||
item.UrlTestDelay = int32(history.Delay)
|
||||
}
|
||||
|
|
@ -1179,7 +1182,7 @@ func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.Server
|
|||
Tag: ob.Tag(),
|
||||
Type: ob.Type(),
|
||||
}
|
||||
if history := historyStorage.LoadURLTestHistory(group.RealTag(boxService.outboundManager, ob)); history != nil {
|
||||
if history := historyStorage.LoadURLTestHistory(group.RealTag(ob, N.NetworkTCP)); history != nil {
|
||||
item.UrlTestTime = history.Time.Unix()
|
||||
item.UrlTestDelay = int32(history.Delay)
|
||||
}
|
||||
|
|
@ -1190,7 +1193,7 @@ func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.Server
|
|||
Tag: ep.Tag(),
|
||||
Type: ep.Type(),
|
||||
}
|
||||
if history := historyStorage.LoadURLTestHistory(group.RealTag(boxService.outboundManager, ep)); history != nil {
|
||||
if history := historyStorage.LoadURLTestHistory(group.RealTag(ep, N.NetworkTCP)); history != nil {
|
||||
item.UrlTestTime = history.Time.Unix()
|
||||
item.UrlTestDelay = int32(history.Delay)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -70,14 +70,18 @@ func proxyInfo(server *Server, detour adapter.Outbound) *badjson.JSONObject {
|
|||
info.Put("type", clashType)
|
||||
info.Put("name", detour.Tag())
|
||||
info.Put("udp", common.Contains(detour.Network(), N.NetworkUDP))
|
||||
delayHistory := server.urlTestHistory.LoadURLTestHistory(group.RealTag(server.outbound, detour))
|
||||
delayHistory := server.urlTestHistory.LoadURLTestHistory(group.RealTag(detour, N.NetworkTCP))
|
||||
if delayHistory != nil {
|
||||
info.Put("history", []*adapter.URLTestHistory{delayHistory})
|
||||
} else {
|
||||
info.Put("history", []*adapter.URLTestHistory{})
|
||||
}
|
||||
if group, isGroup := detour.(adapter.OutboundGroup); isGroup {
|
||||
info.Put("now", group.Now())
|
||||
var now string
|
||||
if selected := group.Selected(N.NetworkTCP); selected != nil {
|
||||
now = selected.Tag()
|
||||
}
|
||||
info.Put("now", now)
|
||||
info.Put("all", group.All())
|
||||
}
|
||||
return &info
|
||||
|
|
@ -193,7 +197,7 @@ func groupContains(outboundManager adapter.OutboundManager, outboundGroup adapte
|
|||
if !loaded {
|
||||
continue
|
||||
}
|
||||
if group.RealTag(outboundManager, member) == tag {
|
||||
if group.RealTag(member, N.NetworkTCP) == tag {
|
||||
return true
|
||||
}
|
||||
memberGroup, isGroup := member.(adapter.OutboundGroup)
|
||||
|
|
@ -228,7 +232,7 @@ func getProxyDelay(server *Server) func(w http.ResponseWriter, r *http.Request)
|
|||
|
||||
delay, err := urltest.URLTest(ctx, url, proxy)
|
||||
defer func() {
|
||||
realTag := group.RealTag(server.outbound, proxy)
|
||||
realTag := group.RealTag(proxy, N.NetworkTCP)
|
||||
if err != nil {
|
||||
server.urlTestHistory.DeleteURLTestHistory(realTag)
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ package group
|
|||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net"
|
||||
|
||||
"github.com/sagernet/sing-box/adapter"
|
||||
|
|
@ -24,17 +25,14 @@ func RegisterSelector(registry *outbound.Registry) {
|
|||
}
|
||||
|
||||
var (
|
||||
_ adapter.OutboundGroup = (*Selector)(nil)
|
||||
_ adapter.Referrer = (*Selector)(nil)
|
||||
_ adapter.ConnectionHandler = (*Selector)(nil)
|
||||
_ adapter.PacketConnectionHandler = (*Selector)(nil)
|
||||
_ adapter.OutboundGroup = (*Selector)(nil)
|
||||
_ adapter.Referrer = (*Selector)(nil)
|
||||
)
|
||||
|
||||
type Selector struct {
|
||||
outbound.Adapter
|
||||
ctx context.Context
|
||||
outbound adapter.OutboundManager
|
||||
connection adapter.ConnectionManager
|
||||
logger logger.ContextLogger
|
||||
tags []string
|
||||
defaultTag string
|
||||
|
|
@ -50,7 +48,6 @@ func NewSelector(ctx context.Context, router adapter.Router, logger log.ContextL
|
|||
Adapter: outbound.NewAdapter(C.TypeSelector, tag, nil, options.Outbounds),
|
||||
ctx: ctx,
|
||||
outbound: service.FromContext[adapter.OutboundManager](ctx),
|
||||
connection: service.FromContext[adapter.ConnectionManager](ctx),
|
||||
logger: logger,
|
||||
tags: options.Outbounds,
|
||||
defaultTag: options.Default,
|
||||
|
|
@ -109,20 +106,24 @@ func (s *Selector) Start() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *Selector) Now() string {
|
||||
selected := s.selected.Load()
|
||||
if selected == nil {
|
||||
return s.tags[0]
|
||||
}
|
||||
return selected.Tag()
|
||||
}
|
||||
|
||||
func (s *Selector) All() []string {
|
||||
return s.tags
|
||||
}
|
||||
|
||||
func (s *Selector) Selected(network string) adapter.Outbound {
|
||||
return s.selected.Load()
|
||||
}
|
||||
|
||||
func (s *Selector) AttachConnection(closer io.Closer) func() {
|
||||
return s.interruptGroup.Add(closer, true)
|
||||
}
|
||||
|
||||
func (s *Selector) References() []string {
|
||||
return []string{s.Now()}
|
||||
selected := s.selected.Load()
|
||||
if selected == nil {
|
||||
return s.tags[:1]
|
||||
}
|
||||
return []string{selected.Tag()}
|
||||
}
|
||||
|
||||
func (s *Selector) SelectOutbound(tag string) bool {
|
||||
|
|
@ -165,38 +166,15 @@ func (s *Selector) ListenPacket(ctx context.Context, destination M.Socksaddr) (n
|
|||
return s.interruptGroup.NewPacketConn(conn, interrupt.IsExternalConnectionFromContext(ctx)), nil
|
||||
}
|
||||
|
||||
func (s *Selector) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
|
||||
ctx = interrupt.ContextWithIsExternalConnection(ctx)
|
||||
selected := s.selected.Load()
|
||||
if outboundHandler, isHandler := selected.(adapter.ConnectionHandler); isHandler {
|
||||
outboundHandler.NewConnection(ctx, conn, metadata, onClose)
|
||||
} else {
|
||||
s.connection.NewConnection(ctx, s, conn, metadata, onClose)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Selector) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
|
||||
ctx = interrupt.ContextWithIsExternalConnection(ctx)
|
||||
selected := s.selected.Load()
|
||||
if outboundHandler, isHandler := selected.(adapter.PacketConnectionHandler); isHandler {
|
||||
outboundHandler.NewPacketConnection(ctx, conn, metadata, onClose)
|
||||
} else {
|
||||
s.connection.NewPacketConnection(ctx, s, conn, metadata, onClose)
|
||||
}
|
||||
}
|
||||
|
||||
func RealTag(outboundManager adapter.OutboundManager, detour adapter.Outbound) string {
|
||||
tag := detour.Tag()
|
||||
func RealTag(detour adapter.Outbound, network string) string {
|
||||
for {
|
||||
group, isGroup := detour.(adapter.OutboundGroup)
|
||||
if !isGroup {
|
||||
return tag
|
||||
return detour.Tag()
|
||||
}
|
||||
tag = group.Now()
|
||||
var loaded bool
|
||||
detour, loaded = outboundManager.Outbound(tag)
|
||||
if !loaded {
|
||||
return tag
|
||||
detour = group.Selected(network)
|
||||
if detour == nil {
|
||||
return ""
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ package group
|
|||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"maps"
|
||||
"net"
|
||||
"sync"
|
||||
|
|
@ -39,7 +40,6 @@ type URLTest struct {
|
|||
outbound.Adapter
|
||||
ctx context.Context
|
||||
outbound adapter.OutboundManager
|
||||
connection adapter.ConnectionManager
|
||||
logger log.ContextLogger
|
||||
tags []string
|
||||
link string
|
||||
|
|
@ -56,7 +56,6 @@ func NewURLTest(ctx context.Context, router adapter.Router, logger log.ContextLo
|
|||
Adapter: outbound.NewAdapter(C.TypeURLTest, tag, []string{N.NetworkTCP, N.NetworkUDP}, options.Outbounds),
|
||||
ctx: ctx,
|
||||
outbound: service.FromContext[adapter.OutboundManager](ctx),
|
||||
connection: service.FromContext[adapter.ConnectionManager](ctx),
|
||||
logger: logger,
|
||||
tags: options.Outbounds,
|
||||
link: options.URL,
|
||||
|
|
@ -99,19 +98,28 @@ func (s *URLTest) Close() error {
|
|||
)
|
||||
}
|
||||
|
||||
func (s *URLTest) Now() string {
|
||||
if s.group.selectedOutboundTCP != nil {
|
||||
return s.group.selectedOutboundTCP.Tag()
|
||||
} else if s.group.selectedOutboundUDP != nil {
|
||||
return s.group.selectedOutboundUDP.Tag()
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (s *URLTest) All() []string {
|
||||
return s.tags
|
||||
}
|
||||
|
||||
func (s *URLTest) Selected(network string) adapter.Outbound {
|
||||
var outbound adapter.Outbound
|
||||
if network == N.NetworkUDP {
|
||||
outbound = s.group.selectedOutboundUDP
|
||||
} else {
|
||||
outbound = s.group.selectedOutboundTCP
|
||||
}
|
||||
if outbound == nil {
|
||||
outbound, _ = s.group.Select(network)
|
||||
}
|
||||
return outbound
|
||||
}
|
||||
|
||||
func (s *URLTest) AttachConnection(closer io.Closer) func() {
|
||||
s.group.Touch()
|
||||
return s.group.interruptGroup.Add(closer, true)
|
||||
}
|
||||
|
||||
func (s *URLTest) References() []string {
|
||||
group := s.group
|
||||
if group == nil {
|
||||
|
|
@ -201,16 +209,6 @@ func (s *URLTest) ListenPacket(ctx context.Context, destination M.Socksaddr) (ne
|
|||
return nil, err
|
||||
}
|
||||
|
||||
func (s *URLTest) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
|
||||
ctx = interrupt.ContextWithIsExternalConnection(ctx)
|
||||
s.connection.NewConnection(ctx, s, conn, metadata, onClose)
|
||||
}
|
||||
|
||||
func (s *URLTest) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
|
||||
ctx = interrupt.ContextWithIsExternalConnection(ctx)
|
||||
s.connection.NewPacketConnection(ctx, s, conn, metadata, onClose)
|
||||
}
|
||||
|
||||
type URLTestGroup struct {
|
||||
ctx context.Context
|
||||
outbound adapter.OutboundManager
|
||||
|
|
@ -314,14 +312,14 @@ func (g *URLTestGroup) Select(network string) (adapter.Outbound, bool) {
|
|||
switch network {
|
||||
case N.NetworkTCP:
|
||||
if g.selectedOutboundTCP != nil {
|
||||
if history := g.history.LoadURLTestHistory(RealTag(g.outbound, g.selectedOutboundTCP)); history != nil {
|
||||
if history := g.history.LoadURLTestHistory(RealTag(g.selectedOutboundTCP, N.NetworkTCP)); history != nil {
|
||||
minOutbound = g.selectedOutboundTCP
|
||||
minDelay = history.Delay
|
||||
}
|
||||
}
|
||||
case N.NetworkUDP:
|
||||
if g.selectedOutboundUDP != nil {
|
||||
if history := g.history.LoadURLTestHistory(RealTag(g.outbound, g.selectedOutboundUDP)); history != nil {
|
||||
if history := g.history.LoadURLTestHistory(RealTag(g.selectedOutboundUDP, N.NetworkUDP)); history != nil {
|
||||
minOutbound = g.selectedOutboundUDP
|
||||
minDelay = history.Delay
|
||||
}
|
||||
|
|
@ -331,7 +329,7 @@ func (g *URLTestGroup) Select(network string) (adapter.Outbound, bool) {
|
|||
if !common.Contains(detour.Network(), network) {
|
||||
continue
|
||||
}
|
||||
history := g.history.LoadURLTestHistory(RealTag(g.outbound, detour))
|
||||
history := g.history.LoadURLTestHistory(RealTag(detour, network))
|
||||
if history == nil {
|
||||
continue
|
||||
}
|
||||
|
|
@ -427,7 +425,7 @@ func URLTestOutbounds(ctx context.Context, outboundManager adapter.OutboundManag
|
|||
testBatch.test(outbounds, link, interval, force)
|
||||
b.Wait()
|
||||
for _, outboundGroup := range testBatch.groups {
|
||||
groupHistory := history.LoadURLTestHistory(RealTag(outboundManager, outboundGroup))
|
||||
groupHistory := history.LoadURLTestHistory(RealTag(outboundGroup, N.NetworkTCP))
|
||||
if groupHistory != nil {
|
||||
testBatch.result[outboundGroup.Tag()] = groupHistory.Delay
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,12 +8,14 @@ import (
|
|||
"github.com/sagernet/sing-box/adapter"
|
||||
"github.com/sagernet/sing-box/log"
|
||||
"github.com/sagernet/sing-tun"
|
||||
"github.com/sagernet/sing/common"
|
||||
"github.com/sagernet/sing/common/byteformats"
|
||||
N "github.com/sagernet/sing/common/network"
|
||||
)
|
||||
|
||||
var (
|
||||
_ tun.FlowTracker = (*flowLogger)(nil)
|
||||
_ tun.FlowTracker = (*flowInterrupter)(nil)
|
||||
_ tun.FlowTracker = multiFlowTracker(nil)
|
||||
)
|
||||
|
||||
|
|
@ -99,3 +101,50 @@ func (t multiFlowTracker) CloseFlow(reason tun.FlowCloseReason) {
|
|||
tracker.CloseFlow(reason)
|
||||
}
|
||||
}
|
||||
|
||||
type flowInterrupter struct {
|
||||
groups []adapter.OutboundGroup
|
||||
removers []func()
|
||||
}
|
||||
|
||||
func newFlowInterrupter(chain []adapter.Outbound) *flowInterrupter {
|
||||
groups := common.FilterIsInstance(chain, func(it adapter.Outbound) (adapter.OutboundGroup, bool) {
|
||||
group, isGroup := it.(adapter.OutboundGroup)
|
||||
return group, isGroup
|
||||
})
|
||||
if len(groups) == 0 {
|
||||
return nil
|
||||
}
|
||||
return &flowInterrupter{groups: groups}
|
||||
}
|
||||
|
||||
type flowCloser struct {
|
||||
tun.FlowHandle
|
||||
}
|
||||
|
||||
func (c flowCloser) Close() error {
|
||||
c.CloseFlow()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *flowInterrupter) AttachFlow(handle tun.FlowHandle) {
|
||||
closer := flowCloser{handle}
|
||||
for _, group := range t.groups {
|
||||
t.removers = append(t.removers, group.AttachConnection(closer))
|
||||
}
|
||||
}
|
||||
|
||||
func (t *flowInterrupter) CountForward(n int) {
|
||||
}
|
||||
|
||||
func (t *flowInterrupter) CountReverse(n int) {
|
||||
}
|
||||
|
||||
func (t *flowInterrupter) FlowEstablished() {
|
||||
}
|
||||
|
||||
func (t *flowInterrupter) CloseFlow(reason tun.FlowCloseReason) {
|
||||
for _, remove := range t.removers {
|
||||
remove()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
119
route/route.go
119
route/route.go
|
|
@ -3,6 +3,7 @@ package route
|
|||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"net/netip"
|
||||
"slices"
|
||||
|
|
@ -10,6 +11,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/sagernet/sing-box/adapter"
|
||||
"github.com/sagernet/sing-box/common/interrupt"
|
||||
"github.com/sagernet/sing-box/common/sniff"
|
||||
C "github.com/sagernet/sing-box/constant"
|
||||
"github.com/sagernet/sing-box/log"
|
||||
|
|
@ -118,10 +120,6 @@ func (r *Router) routeConnection(ctx context.Context, conn net.Conn, metadata ad
|
|||
buf.ReleaseMulti(buffers)
|
||||
return E.New("outbound not found: ", action.Outbound)
|
||||
}
|
||||
if !common.Contains(selectedOutbound.Network(), N.NetworkTCP) {
|
||||
buf.ReleaseMulti(buffers)
|
||||
return E.New("TCP is not supported by outbound: ", selectedOutbound.Tag())
|
||||
}
|
||||
case *R.RuleActionBypass:
|
||||
if action.Outbound == "" {
|
||||
break
|
||||
|
|
@ -132,10 +130,6 @@ func (r *Router) routeConnection(ctx context.Context, conn net.Conn, metadata ad
|
|||
buf.ReleaseMulti(buffers)
|
||||
return E.New("outbound not found: ", action.Outbound)
|
||||
}
|
||||
if !common.Contains(selectedOutbound.Network(), N.NetworkTCP) {
|
||||
buf.ReleaseMulti(buffers)
|
||||
return E.New("TCP is not supported by outbound: ", selectedOutbound.Tag())
|
||||
}
|
||||
case *R.RuleActionReject:
|
||||
buf.ReleaseMulti(buffers)
|
||||
if action.Method == C.RuleActionRejectMethodReply {
|
||||
|
|
@ -151,14 +145,13 @@ func (r *Router) routeConnection(ctx context.Context, conn net.Conn, metadata ad
|
|||
}
|
||||
}
|
||||
if selectedRule == nil {
|
||||
defaultOutbound := r.outbound.Default()
|
||||
if !common.Contains(defaultOutbound.Network(), N.NetworkTCP) {
|
||||
buf.ReleaseMulti(buffers)
|
||||
return E.New("TCP is not supported by default outbound: ", defaultOutbound.Tag())
|
||||
}
|
||||
selectedOutbound = defaultOutbound
|
||||
selectedOutbound = r.outbound.Default()
|
||||
}
|
||||
chain, err := resolveOutbound(selectedOutbound, N.NetworkTCP)
|
||||
if err != nil {
|
||||
buf.ReleaseMulti(buffers)
|
||||
return err
|
||||
}
|
||||
|
||||
for _, buffer := range buffers {
|
||||
conn = bufio.NewCachedConn(conn, buffer)
|
||||
}
|
||||
|
|
@ -166,17 +159,59 @@ func (r *Router) routeConnection(ctx context.Context, conn net.Conn, metadata ad
|
|||
metadata.RouteRule = selectedRule.String()
|
||||
}
|
||||
metadata.RouteOutbound = selectedOutbound.Tag()
|
||||
metadata.OutboundChain = chain
|
||||
for _, tracker := range r.trackers {
|
||||
conn = tracker.RoutedConnection(ctx, conn, metadata, selectedRule, selectedOutbound)
|
||||
}
|
||||
if outboundHandler, isHandler := selectedOutbound.(adapter.ConnectionHandler); isHandler {
|
||||
ctx = interrupt.ContextWithIsExternalConnection(ctx)
|
||||
onClose = registerInterrupt(chain, conn, onClose)
|
||||
outbound := chain[len(chain)-1]
|
||||
if outboundHandler, isHandler := outbound.(adapter.ConnectionHandler); isHandler {
|
||||
outboundHandler.NewConnection(ctx, conn, metadata, onClose)
|
||||
} else {
|
||||
r.connection.NewConnection(ctx, selectedOutbound, conn, metadata, onClose)
|
||||
r.connection.NewConnection(ctx, outbound, conn, metadata, onClose)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func resolveOutbound(outbound adapter.Outbound, network string) ([]adapter.Outbound, error) {
|
||||
chain := []adapter.Outbound{outbound}
|
||||
for {
|
||||
group, isGroup := outbound.(adapter.OutboundGroup)
|
||||
if !isGroup {
|
||||
break
|
||||
}
|
||||
outbound = group.Selected(network)
|
||||
if outbound == nil {
|
||||
return nil, E.New(strings.ToUpper(network), " is not supported by outbound: ", group.Tag())
|
||||
}
|
||||
chain = append(chain, outbound)
|
||||
}
|
||||
if !common.Contains(outbound.Network(), network) {
|
||||
return nil, E.New(strings.ToUpper(network), " is not supported by outbound: ", outbound.Tag())
|
||||
}
|
||||
return chain, nil
|
||||
}
|
||||
|
||||
func registerInterrupt(chain []adapter.Outbound, closer io.Closer, onClose N.CloseHandlerFunc) N.CloseHandlerFunc {
|
||||
var removers []func()
|
||||
for _, outbound := range chain {
|
||||
group, isGroup := outbound.(adapter.OutboundGroup)
|
||||
if !isGroup {
|
||||
continue
|
||||
}
|
||||
removers = append(removers, group.AttachConnection(closer))
|
||||
}
|
||||
if len(removers) == 0 {
|
||||
return onClose
|
||||
}
|
||||
return N.AppendClose(onClose, func(it error) {
|
||||
for _, remove := range removers {
|
||||
remove()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Router) RoutePacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
|
||||
done := make(chan any)
|
||||
err := r.routePacketConnection(ctx, conn, metadata, N.OnceClose(func(it error) {
|
||||
|
|
@ -254,10 +289,6 @@ func (r *Router) routePacketConnection(ctx context.Context, conn N.PacketConn, m
|
|||
N.ReleaseMultiPacketBuffer(packetBuffers)
|
||||
return E.New("outbound not found: ", action.Outbound)
|
||||
}
|
||||
if !common.Contains(selectedOutbound.Network(), N.NetworkUDP) {
|
||||
N.ReleaseMultiPacketBuffer(packetBuffers)
|
||||
return E.New("UDP is not supported by outbound: ", selectedOutbound.Tag())
|
||||
}
|
||||
case *R.RuleActionBypass:
|
||||
if action.Outbound == "" {
|
||||
break
|
||||
|
|
@ -268,10 +299,6 @@ func (r *Router) routePacketConnection(ctx context.Context, conn N.PacketConn, m
|
|||
N.ReleaseMultiPacketBuffer(packetBuffers)
|
||||
return E.New("outbound not found: ", action.Outbound)
|
||||
}
|
||||
if !common.Contains(selectedOutbound.Network(), N.NetworkUDP) {
|
||||
N.ReleaseMultiPacketBuffer(packetBuffers)
|
||||
return E.New("UDP is not supported by outbound: ", selectedOutbound.Tag())
|
||||
}
|
||||
case *R.RuleActionReject:
|
||||
N.ReleaseMultiPacketBuffer(packetBuffers)
|
||||
if action.Method == C.RuleActionRejectMethodReply {
|
||||
|
|
@ -283,12 +310,12 @@ func (r *Router) routePacketConnection(ctx context.Context, conn N.PacketConn, m
|
|||
}
|
||||
}
|
||||
if selectedRule == nil || selectReturn {
|
||||
defaultOutbound := r.outbound.Default()
|
||||
if !common.Contains(defaultOutbound.Network(), N.NetworkUDP) {
|
||||
N.ReleaseMultiPacketBuffer(packetBuffers)
|
||||
return E.New("UDP is not supported by outbound: ", defaultOutbound.Tag())
|
||||
}
|
||||
selectedOutbound = defaultOutbound
|
||||
selectedOutbound = r.outbound.Default()
|
||||
}
|
||||
chain, err := resolveOutbound(selectedOutbound, N.NetworkUDP)
|
||||
if err != nil {
|
||||
N.ReleaseMultiPacketBuffer(packetBuffers)
|
||||
return err
|
||||
}
|
||||
for _, buffer := range slices.Backward(packetBuffers) {
|
||||
conn = bufio.NewCachedPacketConn(conn, buffer.Buffer, buffer.Destination)
|
||||
|
|
@ -298,16 +325,20 @@ func (r *Router) routePacketConnection(ctx context.Context, conn N.PacketConn, m
|
|||
metadata.RouteRule = selectedRule.String()
|
||||
}
|
||||
metadata.RouteOutbound = selectedOutbound.Tag()
|
||||
metadata.OutboundChain = chain
|
||||
for _, tracker := range r.trackers {
|
||||
conn = tracker.RoutedPacketConnection(ctx, conn, metadata, selectedRule, selectedOutbound)
|
||||
}
|
||||
if metadata.FakeIP {
|
||||
conn = newFakeIPNATPacketConn(bufio.NewNetPacketConn(conn), metadata.OriginDestination, metadata.Destination)
|
||||
}
|
||||
if outboundHandler, isHandler := selectedOutbound.(adapter.PacketConnectionHandler); isHandler {
|
||||
ctx = interrupt.ContextWithIsExternalConnection(ctx)
|
||||
onClose = registerInterrupt(chain, conn, onClose)
|
||||
outbound := chain[len(chain)-1]
|
||||
if outboundHandler, isHandler := outbound.(adapter.PacketConnectionHandler); isHandler {
|
||||
outboundHandler.NewPacketConnection(ctx, conn, metadata, onClose)
|
||||
} else {
|
||||
r.connection.NewPacketConnection(ctx, selectedOutbound, conn, metadata, onClose)
|
||||
r.connection.NewPacketConnection(ctx, outbound, conn, metadata, onClose)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
@ -459,20 +490,11 @@ func (r *Router) preMatchFlow(ctx context.Context, metadata *adapter.InboundCont
|
|||
return continueResult
|
||||
}
|
||||
}
|
||||
for range 8 {
|
||||
group, isGroup := outbound.(adapter.OutboundGroup)
|
||||
if !isGroup {
|
||||
break
|
||||
}
|
||||
selectedOutbound, selectedLoaded := r.outbound.Outbound(group.Now())
|
||||
if !selectedLoaded {
|
||||
return continueResult
|
||||
}
|
||||
outbound = selectedOutbound
|
||||
}
|
||||
if !common.Contains(outbound.Network(), metadata.Network) {
|
||||
chain, err := resolveOutbound(outbound, metadata.Network)
|
||||
if err != nil {
|
||||
return continueResult
|
||||
}
|
||||
outbound = chain[len(chain)-1]
|
||||
flowOutbound, isFlowOutbound := outbound.(adapter.FlowOutbound)
|
||||
if !isFlowOutbound {
|
||||
return continueResult
|
||||
|
|
@ -522,11 +544,16 @@ func (r *Router) preMatchFlow(ctx context.Context, metadata *adapter.InboundCont
|
|||
} else if metadata.Destination != packetDestination {
|
||||
result.Destination = metadata.Destination.AddrPort()
|
||||
}
|
||||
metadata.OutboundChain = chain
|
||||
metadataCopy := *metadata
|
||||
result.NewTracker = func() tun.FlowTracker {
|
||||
r.logger.InfoContext(ctx, "pre-match: forward ", metadataCopy.Network, " connection from ", metadataCopy.Source.AddrString(), " to ", metadataCopy.Destination.AddrString(), " via outbound/", outbound.Type(), "[", outbound.Tag(), "]")
|
||||
flowTrackers := make([]tun.FlowTracker, 0, len(r.trackers)+1)
|
||||
flowTrackers := make([]tun.FlowTracker, 0, len(r.trackers)+2)
|
||||
flowTrackers = append(flowTrackers, newFlowLogger(ctx, r.logger, metadataCopy, outbound))
|
||||
flowInterrupter := newFlowInterrupter(chain)
|
||||
if flowInterrupter != nil {
|
||||
flowTrackers = append(flowTrackers, flowInterrupter)
|
||||
}
|
||||
for _, tracker := range r.trackers {
|
||||
flowTracker := tracker.RoutedFlow(ctx, metadataCopy, matchedRule, outbound)
|
||||
if flowTracker != nil {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue