Refactor lifecycle to scoped cleanup

This commit is contained in:
世界
2026-10-09 12:36:03 +08:00
parent a364ff4794
commit a691a44023
115 changed files with 1595 additions and 2030 deletions
+8 -6
View File
@@ -79,7 +79,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -88,12 +88,14 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
scope.Add(h.tlsConfig.Close)
}
return h.listener.Start()
}
func (h *Inbound) Close() error {
return common.Close(h.listener, h.tlsConfig)
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+2 -7
View File
@@ -36,7 +36,6 @@ type Outbound struct {
tlsConfig tls.Config
clientOptions anytls.ClientConfig
clientMetadata string
client *anytls.Client
sessionClient *session.Client
uotClient *uot.Client
logger log.ContextLogger
@@ -89,7 +88,7 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL
return outbound, nil
}
func (h *Outbound) Start(stage adapter.StartStage) error {
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
@@ -97,7 +96,7 @@ func (h *Outbound) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
h.client = client
scope.Add(client.Close)
h.sessionClient = sessionClientOf(client)
h.uotClient = &uot.Client{
Dialer: anytlsDialer(h.createProxy),
@@ -160,7 +159,3 @@ func (h *Outbound) ListenPacket(ctx context.Context, destination M.Socksaddr) (n
h.logger.InfoContext(ctx, "outbound UoT packet connection to ", destination)
return h.uotClient.ListenPacket(ctx, destination)
}
func (h *Outbound) Close() error {
return common.Close(common.PtrOrNil(h.client))
}
+12 -18
View File
@@ -41,16 +41,13 @@ type backendBase struct {
returnAccess sync.Mutex
returnPaths []tun.Return
egressAccess sync.Mutex
forwardingRestore []sysctlState
unregister func()
egressAccess sync.Mutex
session adapter.BridgeSession
currentEgress string
closeOnce sync.Once
closed chan struct{}
readDone chan struct{}
closed <-chan struct{}
readGroup sync.WaitGroup
}
func (b *backendBase) init(ctx context.Context, logger logger.ContextLogger, networkManager adapter.NetworkManager, tag string, options option.BridgeOutboundOptions) error {
@@ -100,12 +97,14 @@ func (b *backendBase) DetachReturn(returnPath tun.Return) error {
return nil
}
func (b *backendBase) registerMonitors(syncFunc func()) {
var unregisterFuncs []func()
func (b *backendBase) registerMonitors(scope *adapter.Scope, syncFunc func()) {
networkMonitor := b.networkManager.NetworkMonitor()
if networkMonitor != nil {
networkElement := networkMonitor.RegisterCallback(syncFunc)
unregisterFuncs = append(unregisterFuncs, func() { networkMonitor.UnregisterCallback(networkElement) })
scope.Add(func() error {
networkMonitor.UnregisterCallback(networkElement)
return nil
})
} else if b.boundInterface != "" {
b.logger.Debug("network monitor unavailable, pinned egress will not track interface changes")
}
@@ -113,14 +112,10 @@ func (b *backendBase) registerMonitors(syncFunc func()) {
interfaceMonitor := b.networkManager.InterfaceMonitor()
if interfaceMonitor != nil {
interfaceElement := interfaceMonitor.RegisterCallback(func(_ *control.Interface, _ int) { syncFunc() })
unregisterFuncs = append(unregisterFuncs, func() { interfaceMonitor.UnregisterCallback(interfaceElement) })
}
}
if len(unregisterFuncs) > 0 {
b.unregister = func() {
for _, unregisterFunc := range unregisterFuncs {
unregisterFunc()
}
scope.Add(func() error {
interfaceMonitor.UnregisterCallback(interfaceElement)
return nil
})
}
}
}
@@ -166,7 +161,6 @@ func (b *backendBase) resolveEgress() string {
}
func (b *backendBase) readLoop() {
defer close(b.readDone)
buffer := make([]byte, tun.PacketOffset+bridgeTunMTU)
for {
n, err := b.tunInterface.Read(buffer)
+45 -58
View File
@@ -72,24 +72,30 @@ func newBackend(ctx context.Context, logger logger.ContextLogger, networkManager
return instance, nil
}
func (b *backendDarwin) Start(stage adapter.StartStage) error {
if stage != adapter.StartStateStart {
return nil
}
err := b.start()
if err != nil {
b.Close()
return err
func (b *backendDarwin) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateInitialize:
scope.Add(func() error {
releaseBridgeIndex(b.index)
return nil
})
case adapter.StartStateStart:
b.closed = scope.Context().Done()
if b.platform != nil {
return b.startPlatform(scope)
}
return b.start(scope)
}
return nil
}
func (b *backendDarwin) start() error {
if b.platform != nil {
return b.startPlatform()
}
func (b *backendDarwin) start(scope *adapter.Scope) error {
b.tunName = tun.CalculateInterfaceName(b.bridgeName)
b.anchorName = "com.apple/sing-box-" + b.tunName
scope.Add(func() error {
b.readGroup.Wait()
return nil
})
tunInterface, err := tun.New(tun.Options{
Name: b.tunName,
MTU: bridgeTunMTUDarwin,
@@ -102,12 +108,17 @@ func (b *backendDarwin) start() error {
if err != nil {
return E.Cause(err, "create bridge tun")
}
scope.Add(tunInterface.Close)
b.tunInterface = tunInterface
err = tunInterface.Start()
if err != nil {
return E.Cause(err, "start bridge tun")
}
b.forwardingRestore = enableDarwinForwarding(b.logger, b.inet4Port.IsValid(), b.inet6Port.IsValid())
forwardingRestore := enableDarwinForwarding(b.logger, b.inet4Port.IsValid(), b.inet6Port.IsValid())
scope.Add(func() error {
restoreDarwinForwarding(forwardingRestore)
return nil
})
err = assignBridgePortAddress(b.tunName, b.inet4Local, b.inet4Port)
if err != nil {
return E.Cause(err, "add bridge route")
@@ -121,23 +132,31 @@ func (b *backendDarwin) start() error {
if err != nil {
return E.Cause(err, "enable pf")
}
scope.Add(func() error {
_ = b.pfDevice.StopReference(b.pfToken)
return b.pfDevice.Close()
})
dropRules := bridgeDropRules(b.tunName, b.inet4Port, b.inet6Port)
err = b.pfDevice.LoadAnchor(b.anchorName, dropRules)
if err != nil {
return E.Cause(err, "initialize bridge pf rules")
}
scope.Add(func() error {
b.egressAccess.Lock()
_ = b.pfDevice.LoadAnchor(b.anchorName, nil)
b.egressAccess.Unlock()
return nil
})
b.currentRules = dropRules
b.batchTUN = tunInterface.(tun.DarwinTUN)
b.closed = make(chan struct{})
b.readDone = make(chan struct{})
b.registerMonitors(b.syncEgress)
b.registerMonitors(scope, b.syncEgress)
b.syncEgress()
go b.batchReadLoop()
b.readGroup.Go(b.batchReadLoop)
b.logger.Info("bridge started at ", b.tunName, " (masquerade, egress ", b.egressLabel(), ")")
return nil
}
func (b *backendDarwin) startPlatform() error {
func (b *backendDarwin) startPlatform(scope *adapter.Scope) error {
session, err := b.platform.CreateBridge(adapter.BridgeOptions{
BridgeName: b.bridgeName,
MTU: bridgeTunMTUDarwin,
@@ -148,11 +167,16 @@ func (b *backendDarwin) startPlatform() error {
if err != nil {
return E.Cause(err, "create bridge")
}
scope.Add(session.Close)
b.session = session
b.tunName = session.Name()
if !session.Inet6Active() {
b.inet6Port = netip.Addr{}
}
scope.Add(func() error {
b.readGroup.Wait()
return nil
})
tunInterface, err := tun.New(tun.Options{
Name: b.tunName,
MTU: bridgeTunMTUDarwin,
@@ -164,17 +188,16 @@ func (b *backendDarwin) startPlatform() error {
if err != nil {
return E.Cause(err, "create bridge tun")
}
scope.Add(tunInterface.Close)
b.tunInterface = tunInterface
err = tunInterface.Start()
if err != nil {
return E.Cause(err, "start bridge tun")
}
b.batchTUN = tunInterface.(tun.DarwinTUN)
b.closed = make(chan struct{})
b.readDone = make(chan struct{})
b.registerMonitors(b.syncSessionEgress)
b.registerMonitors(scope, b.syncSessionEgress)
b.syncSessionEgress()
go b.batchReadLoop()
b.readGroup.Go(b.batchReadLoop)
b.logger.Info("bridge started at ", b.tunName, " (platform, egress ", b.egressLabel(), ")")
return nil
}
@@ -186,41 +209,6 @@ func (b *backendDarwin) egressLabel() string {
return "auto"
}
func (b *backendDarwin) Close() error {
b.closeOnce.Do(func() {
if b.closed != nil {
close(b.closed)
}
if b.unregister != nil {
b.unregister()
}
if b.pfDevice != nil && b.anchorName != "" {
b.egressAccess.Lock()
_ = b.pfDevice.LoadAnchor(b.anchorName, nil)
b.egressAccess.Unlock()
}
restoreDarwinForwarding(b.forwardingRestore)
b.forwardingRestore = nil
if b.pfDevice != nil {
if b.pfToken != 0 {
_ = b.pfDevice.StopReference(b.pfToken)
}
_ = b.pfDevice.Close()
}
if b.tunInterface != nil {
b.tunInterface.Close()
}
if b.readDone != nil {
<-b.readDone
}
if b.session != nil {
_ = b.session.Close()
}
releaseBridgeIndex(b.index)
})
return nil
}
func (b *backendDarwin) PortMTU() uint32 {
return bridgeTunMTUDarwin
}
@@ -247,7 +235,6 @@ func (b *backendDarwin) WritePackets(packets [][]byte) error {
}
func (b *backendDarwin) batchReadLoop() {
defer close(b.readDone)
headroom := -1
var buffers [][]byte
var batch [][]byte
+68 -61
View File
@@ -63,24 +63,30 @@ func newBackend(ctx context.Context, logger logger.ContextLogger, networkManager
return instance, nil
}
func (b *backendLinux) Start(stage adapter.StartStage) error {
if stage != adapter.StartStateStart {
return nil
}
err := b.start()
if err != nil {
b.Close()
return err
func (b *backendLinux) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateInitialize:
scope.Add(func() error {
releaseBridgeIndex(b.index)
return nil
})
case adapter.StartStateStart:
b.closed = scope.Context().Done()
if b.platform != nil {
return b.startPlatform(scope)
}
return b.start(scope)
}
return nil
}
func (b *backendLinux) start() error {
if b.platform != nil {
return b.startPlatform()
}
func (b *backendLinux) start(scope *adapter.Scope) error {
b.tunName = tun.CalculateInterfaceName(b.bridgeName)
b.nftTableName = "sing-box-" + b.tunName
scope.Add(func() error {
b.readGroup.Wait()
return nil
})
tunInterface, err := tun.New(tun.Options{
Name: b.tunName,
MTU: bridgeTunMTU,
@@ -92,6 +98,7 @@ func (b *backendLinux) start() error {
if err != nil {
return E.Cause(err, "create bridge tun")
}
scope.Add(tunInterface.Close)
b.tunInterface = tunInterface
err = tunInterface.Start()
if err != nil {
@@ -112,13 +119,38 @@ func (b *backendLinux) start() error {
if err != nil {
return E.Cause(err, "set up bridge netfilter")
}
scope.Add(func() error {
b.egressAccess.Lock()
cleanupBridgeNetfilter(b.nftTableName)
b.egressAccess.Unlock()
return nil
})
if !inet6Active {
b.inet6Port = netip.Addr{}
}
b.forwardingRestore = enableBridgeForwarding(b.logger, b.tunName, b.inet4Port.IsValid(), b.inet6Port.IsValid())
forwardingRestore := enableBridgeForwarding(b.logger, b.tunName, b.inet4Port.IsValid(), b.inet6Port.IsValid())
scope.Add(func() error {
restoreBridgeForwarding(forwardingRestore)
return nil
})
if b.routeTable != 0 {
scope.Add(func() error {
b.egressAccess.Lock()
flushBridgeRouteTable(b.routeTable)
b.egressAccess.Unlock()
return nil
})
}
if b.boundInterface != "" {
b.syncEgress()
}
scope.Add(func() error {
b.egressAccess.Lock()
removeBridgeFamily(b.tunName, b.ruleIndex, b.routeTable, unix.AF_INET, b.inet4Port)
removeBridgeFamily(b.tunName, b.ruleIndex, b.routeTable, unix.AF_INET6, b.inet6Port)
b.egressAccess.Unlock()
return nil
})
err = setupBridgeFamily(b.tunName, b.ruleIndex, b.routeTable, unix.AF_INET, b.inet4Port)
if err != nil {
return E.Cause(err, "set up bridge routing")
@@ -129,12 +161,10 @@ func (b *backendLinux) start() error {
removeBridgeFamily(b.tunName, b.ruleIndex, b.routeTable, unix.AF_INET6, b.inet6Port)
b.inet6Port = netip.Addr{}
}
b.closed = make(chan struct{})
b.readDone = make(chan struct{})
if b.batchTUN != nil {
go b.batchReadLoop()
b.readGroup.Go(b.batchReadLoop)
} else {
go b.readLoop()
b.readGroup.Go(b.readLoop)
}
egress := "auto"
if b.boundInterface != "" {
@@ -142,7 +172,10 @@ func (b *backendLinux) start() error {
monitor := b.networkManager.NetworkMonitor()
if monitor != nil {
element := monitor.RegisterCallback(func() { b.syncEgress() })
b.unregister = func() { monitor.UnregisterCallback(element) }
scope.Add(func() error {
monitor.UnregisterCallback(element)
return nil
})
} else {
b.logger.Debug("network monitor unavailable, pinned egress will not track interface changes")
}
@@ -151,7 +184,10 @@ func (b *backendLinux) start() error {
monitor := b.networkManager.InterfaceMonitor()
if monitor != nil {
element := monitor.RegisterCallback(func(_ *control.Interface, _ int) { b.updateClamp() })
b.unregister = func() { monitor.UnregisterCallback(element) }
scope.Add(func() error {
monitor.UnregisterCallback(element)
return nil
})
}
b.updateClamp()
}
@@ -163,7 +199,7 @@ func (b *backendLinux) start() error {
return nil
}
func (b *backendLinux) startPlatform() error {
func (b *backendLinux) startPlatform(scope *adapter.Scope) error {
session, err := b.platform.CreateBridge(adapter.BridgeOptions{
BridgeName: b.bridgeName,
MTU: bridgeTunMTU,
@@ -175,11 +211,16 @@ func (b *backendLinux) startPlatform() error {
if err != nil {
return E.Cause(err, "create bridge")
}
scope.Add(session.Close)
b.session = session
b.tunName = session.Name()
if !session.Inet6Active() {
b.inet6Port = netip.Addr{}
}
scope.Add(func() error {
b.readGroup.Wait()
return nil
})
tunInterface, err := tun.New(tun.Options{
Name: b.tunName,
MTU: bridgeTunMTU,
@@ -190,6 +231,7 @@ func (b *backendLinux) startPlatform() error {
if err != nil {
return E.Cause(err, "create bridge tun")
}
scope.Add(tunInterface.Close)
b.tunInterface = tunInterface
err = tunInterface.Start()
if err != nil {
@@ -204,17 +246,18 @@ func (b *backendLinux) startPlatform() error {
b.writeBuffers[i] = make([]byte, b.writeHeadroom+maxPacketLength)
}
}
b.closed = make(chan struct{})
b.readDone = make(chan struct{})
if b.batchTUN != nil {
go b.batchReadLoop()
b.readGroup.Go(b.batchReadLoop)
} else {
go b.readLoop()
b.readGroup.Go(b.readLoop)
}
monitor := b.networkManager.InterfaceMonitor()
if monitor != nil {
element := monitor.RegisterCallback(func(_ *control.Interface, _ int) { b.syncSessionEgress() })
b.unregister = func() { monitor.UnregisterCallback(element) }
scope.Add(func() error {
monitor.UnregisterCallback(element)
return nil
})
}
b.syncSessionEgress()
egress := "auto"
@@ -225,41 +268,6 @@ func (b *backendLinux) startPlatform() error {
return nil
}
func (b *backendLinux) Close() error {
b.closeOnce.Do(func() {
if b.closed != nil {
close(b.closed)
}
if b.unregister != nil {
b.unregister()
}
if b.tunInterface != nil {
b.tunInterface.Close()
}
if b.readDone != nil {
<-b.readDone
}
if b.session != nil {
_ = b.session.Close()
} else {
b.egressAccess.Lock()
if b.tunName != "" {
cleanupBridgeNetfilter(b.nftTableName)
removeBridgeFamily(b.tunName, b.ruleIndex, b.routeTable, unix.AF_INET, b.inet4Port)
removeBridgeFamily(b.tunName, b.ruleIndex, b.routeTable, unix.AF_INET6, b.inet6Port)
}
if b.routeTable != 0 {
flushBridgeRouteTable(b.routeTable)
}
b.egressAccess.Unlock()
restoreBridgeForwarding(b.forwardingRestore)
b.forwardingRestore = nil
}
releaseBridgeIndex(b.index)
})
return nil
}
func (b *backendLinux) PortMTU() uint32 {
return 0
}
@@ -308,7 +316,6 @@ func (b *backendLinux) WritePackets(packets [][]byte) error {
// BatchRead completes any kernel-deferred checksums while splitting GRO frames
// (virtio NEEDS_CSUM), so unlike readLoop no checksum fix is needed here.
func (b *backendLinux) batchReadLoop() {
defer close(b.readDone)
batchSize := b.batchTUN.BatchSize()
sizes := make([]int, batchSize)
batch := make([][]byte, 0, batchSize)
+23 -32
View File
@@ -140,20 +140,21 @@ func newBackend(ctx context.Context, logger logger.ContextLogger, networkManager
return instance, nil
}
func (b *backendWindows) Start(stage adapter.StartStage) error {
if stage != adapter.StartStateStart {
return nil
}
err := b.start()
if err != nil {
b.Close()
return err
func (b *backendWindows) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateInitialize:
scope.Add(func() error {
releaseBridgeIndex(b.index)
return nil
})
case adapter.StartStateStart:
return b.start(scope)
}
return nil
}
func (b *backendWindows) start() error {
b.closed = make(chan struct{})
func (b *backendWindows) start(scope *adapter.Scope) error {
b.closed = scope.Context().Done()
state := b.currentEgressState()
b.egress.Store(state)
@@ -162,11 +163,16 @@ func (b *backendWindows) start() error {
if err != nil {
return err
}
scope.Add(func() error {
b.reservation.Close()
return nil
})
injectHandle, err := windivert.Open(nil, windivert.LayerNetwork, windivert.PriorityHighest, windivert.FlagSendOnly)
if err != nil {
return E.Cause(err, "bridge: open injection handle (Administrator required)")
}
scope.Add(injectHandle.Close)
b.injectHandle = injectHandle
b.sendBuffer = make([]byte, 0, bridgeBatchBufferSize)
b.sendAddrs = make([]windivert.Address, 0, windivert.BatchMax)
@@ -177,8 +183,14 @@ func (b *backendWindows) start() error {
if err != nil {
return err
}
scope.Add(func() error {
b.egressAccess.Lock()
b.closeDivertersLocked()
b.egressAccess.Unlock()
return nil
})
b.registerMonitors(b.syncEgress)
b.registerMonitors(scope, b.syncEgress)
b.syncEgress()
state = b.egress.Load()
if !state.inet4.IsValid() && !state.inet6.IsValid() {
@@ -799,27 +811,6 @@ func sortSegments(segments []localSegment) {
})
}
func (b *backendWindows) Close() error {
b.closeOnce.Do(func() {
if b.closed != nil {
close(b.closed)
}
if b.unregister != nil {
b.unregister()
}
b.egressAccess.Lock()
b.closeDivertersLocked()
b.egressAccess.Unlock()
if b.injectHandle != nil {
b.injectHandle.Close()
}
b.reservation.Close()
b.reservation = nil
releaseBridgeIndex(b.index)
})
return nil
}
type transportInfo struct {
protocol tcpip.TransportProtocolNumber
transport []byte
+2 -6
View File
@@ -57,12 +57,8 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL
}, nil
}
func (o *Outbound) Start(stage adapter.StartStage) error {
return o.backend.Start(stage)
}
func (o *Outbound) Close() error {
return o.backend.Close()
func (o *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
return o.backend.Start(stage, scope)
}
func (o *Outbound) PreferredDomain(metadata *adapter.InboundContext, domain string) bool {
+7 -6
View File
@@ -92,15 +92,16 @@ type Inbound struct {
references []string
}
func (i *Inbound) Start(stage adapter.StartStage) error {
func (i *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
return i.service.Start()
}
func (i *Inbound) Close() error {
return i.service.Close()
err := i.service.Start()
if err != nil {
return err
}
scope.Add(i.service.Close)
return nil
}
type routerDialer struct {
+7 -6
View File
@@ -70,21 +70,22 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (i *Inbound) Start(stage adapter.StartStage) error {
func (i *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
return i.listener.Start()
err := i.listener.Start()
if err != nil {
return err
}
scope.Add(i.listener.Close)
return nil
}
func (i *Inbound) InterfaceUpdated(ctx context.Context) {
i.udpNat.Purge()
}
func (i *Inbound) Close() error {
return i.listener.Close()
}
func (i *Inbound) NewPacket(buffer *buf.Buffer, source M.Socksaddr) {
i.udpNat.NewPacket([][]byte{buffer.Bytes()}, source, i.listener.UDPAddr(), nil)
}
+5 -8
View File
@@ -88,8 +88,12 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL
return outbound, nil
}
func (h *Outbound) Start(stage adapter.StartStage) error {
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateInitialize:
if h.icmpPort != nil {
scope.Add(h.icmpPort.Close)
}
case adapter.StartStatePostStart, adapter.StartStateStarted:
if len(h.myAddresses.Load()) == 0 {
h.fetchMyAddresses()
@@ -205,13 +209,6 @@ func (h *Outbound) WritePackets(packets [][]byte) error {
return h.icmpPort.WritePackets(packets)
}
func (h *Outbound) Close() error {
if h.icmpPort != nil {
return h.icmpPort.Close()
}
return nil
}
func (h *Outbound) DialParallel(ctx context.Context, network string, destination M.Socksaddr, destinationAddresses []netip.Addr) (net.Conn, error) {
if h.isMyLoopbackAddress(destinationAddresses...) {
return nil, E.New("loopback connection to TUN range")
+4 -1
View File
@@ -73,7 +73,10 @@ func (s *Selector) Network() []string {
return selected.Network()
}
func (s *Selector) Start() error {
func (s *Selector) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
for i, tag := range s.tags {
detour, loaded := s.outbound.Outbound(tag)
if !loaded {
+2 -7
View File
@@ -71,7 +71,7 @@ func NewURLTest(ctx context.Context, router adapter.Router, logger log.ContextLo
return outbound, nil
}
func (s *URLTest) Start(stage adapter.StartStage) error {
func (s *URLTest) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateStart:
outbounds := make([]adapter.Outbound, 0, len(s.tags))
@@ -89,16 +89,11 @@ func (s *URLTest) Start(stage adapter.StartStage) error {
s.group = group
case adapter.StartStateStarted:
s.group.PostStart()
scope.Add(s.group.Close)
}
return nil
}
func (s *URLTest) Close() error {
return common.Close(
common.PtrOrNil(s.group),
)
}
func (s *URLTest) Now() string {
if s.group.selectedOutboundTCP != nil {
return s.group.selectedOutboundTCP.Tag()
+8 -9
View File
@@ -66,7 +66,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -75,15 +75,14 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return E.Cause(err, "create TLS config")
}
scope.Add(h.tlsConfig.Close)
}
return h.listener.Start()
}
func (h *Inbound) Close() error {
return common.Close(
h.listener,
h.tlsConfig,
)
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+9 -10
View File
@@ -147,7 +147,7 @@ func (h *Inbound) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn,
h.router.RoutePacketConnectionEx(ctx, conn, metadata, onClose)
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -156,18 +156,17 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
scope.Add(h.tlsConfig.Close)
}
packetConn, err := h.listener.ListenUDP()
if err != nil {
return err
}
return h.service.Start(packetConn)
}
func (h *Inbound) Close() error {
return common.Close(
h.listener,
h.tlsConfig,
common.PtrOrNil(h.service),
)
scope.Add(h.listener.Close)
err = h.service.Start(packetConn)
if err != nil {
return err
}
scope.Add(h.service.Close)
return nil
}
+8 -2
View File
@@ -131,6 +131,12 @@ func (h *Outbound) CloseIdleConnections() {
h.client.CloseIdleConnections()
}
func (h *Outbound) Close() error {
return h.client.CloseWithError(os.ErrClosed)
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
scope.Add(func() error {
return h.client.CloseWithError(os.ErrClosed)
})
return nil
}
+9 -10
View File
@@ -268,7 +268,7 @@ func (h *Inbound) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn,
h.router.RoutePacketConnectionEx(ctx, conn, metadata, onClose)
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -277,22 +277,21 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
scope.Add(h.tlsConfig.Close)
}
packetConn, err := h.listener.ListenUDP()
if err != nil {
return err
}
return h.service.Start(packetConn)
scope.Add(h.listener.Close)
err = h.service.Start(packetConn)
if err != nil {
return err
}
scope.Add(h.service.Close)
return nil
}
func (h *Inbound) InterfaceUpdated(ctx context.Context) {
h.service.Reset()
}
func (h *Inbound) Close() error {
return common.Close(
h.listener,
h.tlsConfig,
common.PtrOrNil(h.service),
)
}
+8 -2
View File
@@ -223,6 +223,12 @@ func (h *Outbound) CloseIdleConnections() {
h.client.CloseIdleConnections()
}
func (h *Outbound) Close() error {
return h.client.CloseWithError(os.ErrClosed)
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
scope.Add(func() error {
return h.client.CloseWithError(os.ErrClosed)
})
return nil
}
+15 -11
View File
@@ -122,7 +122,14 @@ func NewRealmService(ctx context.Context, logger log.ContextLogger, tag string,
return s, nil
}
func (s *RealmService) Start(stage adapter.StartStage) error {
func (s *RealmService) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage == adapter.StartStateInitialize {
scope.Add(func() error {
s.cancel()
return nil
})
return nil
}
if stage != adapter.StartStateStart {
return nil
}
@@ -131,11 +138,17 @@ func (s *RealmService) Start(stage adapter.StartStage) error {
if err != nil {
return E.Cause(err, "create TLS config")
}
scope.Add(s.tlsConfig.Close)
}
tcpListener, err := s.listener.ListenTCP()
if err != nil {
return err
}
scope.Add(s.listener.Close)
scope.Add(func() error {
s.server.closeAll()
return nil
})
if s.tlsConfig != nil {
if !common.Contains(s.tlsConfig.NextProtos(), http2.NextProtoTLS) {
s.tlsConfig.SetNextProtos(append([]string{"h2"}, s.tlsConfig.NextProtos()...))
@@ -148,15 +161,6 @@ func (s *RealmService) Start(stage adapter.StartStage) error {
s.logger.Error("serve error: ", err)
}
}()
scope.Add(s.httpServer.Close)
return nil
}
func (s *RealmService) Close() error {
s.cancel()
err := common.Close(common.PtrOrNil(s.httpServer))
s.server.closeAll()
return E.Errors(err, common.Close(
common.PtrOrNil(s.listener),
s.tlsConfig,
))
}
+8 -9
View File
@@ -79,7 +79,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -88,15 +88,14 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return E.Cause(err, "create TLS config")
}
scope.Add(h.tlsConfig.Close)
}
return h.listener.Start()
}
func (h *Inbound) Close() error {
return common.Close(
h.listener,
h.tlsConfig,
)
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+5 -12
View File
@@ -52,7 +52,6 @@ type Inbound struct {
authenticator *auth.Authenticator
tlsConfig tls.ServerConfig
httpServer *http.Server
h3Server io.Closer
}
func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.NaiveInboundOptions) (adapter.Inbound, error) {
@@ -89,7 +88,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (n *Inbound) Start(stage adapter.StartStage) error {
func (n *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -98,7 +97,9 @@ func (n *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return E.Cause(err, "create TLS config")
}
scope.Add(n.tlsConfig.Close)
}
scope.Add(n.listener.Close)
if common.Contains(n.network, N.NetworkTCP) {
tcpListener, err := n.listener.ListenTCP()
if err != nil {
@@ -126,12 +127,13 @@ func (n *Inbound) Start(stage adapter.StartStage) error {
n.logger.Error("http server serve error: ", sErr)
}
}()
scope.Add(n.httpServer.Close)
}
if common.Contains(n.network, N.NetworkUDP) {
http3Server, err := ConfigureHTTP3ListenerFunc(n.ctx, n.logger, n.listener, n, n.tlsConfig, n.options)
if err == nil {
n.h3Server = http3Server
scope.Add(http3Server.Close)
} else if len(n.network) > 1 {
n.logger.Warn(E.Cause(err, "naive http3 disabled"))
} else {
@@ -142,15 +144,6 @@ func (n *Inbound) Start(stage adapter.StartStage) error {
return nil
}
func (n *Inbound) Close() error {
return common.Close(
n.listener,
common.PtrOrNil(n.httpServer),
n.h3Server,
n.tlsConfig,
)
}
func (n *Inbound) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
ctx := log.ContextWithNewID(request.Context())
if request.Method != "CONNECT" {
+2 -5
View File
@@ -221,7 +221,7 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL
}, nil
}
func (h *Outbound) Start(stage adapter.StartStage) error {
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -229,6 +229,7 @@ func (h *Outbound) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
scope.Add(h.client.Close)
h.logger.Info("NaiveProxy started, version: ", h.client.Engine().Version())
return nil
}
@@ -260,10 +261,6 @@ func (h *Outbound) InterfaceUpdated(ctx context.Context) {
h.client.CloseAllConnections()
}
func (h *Outbound) Close() error {
return h.client.Close()
}
func (h *Outbound) Client() *cronet.NaiveClient {
return h.client
}
+64 -65
View File
@@ -43,25 +43,22 @@ var (
type Endpoint struct {
endpointBase
loopContext context.Context
cancelLoop context.CancelFunc
dnsRouter adapter.DNSRouter
client *openconnect.Client
device openconnecttransport.Device
server string
flavor string
stateAccess sync.Mutex
state atomic.Pointer[clientState]
dnsTransportAccess sync.Mutex
dnsTransport *DNSTransport
deviceStarted bool
readLoopDone chan struct{}
statusAccess sync.Mutex
statusUpdated chan struct{}
terminalError string
authFormLoopDone chan struct{}
activeTransportLoopDone chan struct{}
hotpCounter atomic.Uint64
loopContext context.Context
cancelLoop context.CancelFunc
dnsRouter adapter.DNSRouter
client *openconnect.Client
device openconnecttransport.Device
server string
flavor string
stateAccess sync.Mutex
state atomic.Pointer[clientState]
dnsTransportAccess sync.Mutex
dnsTransport *DNSTransport
deviceStarted bool
statusAccess sync.Mutex
statusUpdated chan struct{}
terminalError string
hotpCounter atomic.Uint64
}
type clientState struct {
@@ -418,34 +415,60 @@ func (e *Endpoint) updateState(update func(state *clientState)) {
e.state.Store(&newState)
}
func (e *Endpoint) Start(stage adapter.StartStage) error {
if stage != adapter.StartStatePostStart {
return nil
func (e *Endpoint) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateInitialize:
scope.Add(func() error {
e.notifyStatusUpdated()
return nil
})
scope.Add(e.device.Close)
scope.Add(func() error {
e.cancelLoop()
return nil
})
case adapter.StartStatePostStart:
var loopGroup sync.WaitGroup
scope.Add(func() error {
loopGroup.Wait()
return nil
})
err := e.client.Start()
if err != nil {
return err
}
scope.Add(e.client.Close)
e.stateAccess.Lock()
e.updateState(func(state *clientState) {
state.started = true
})
e.stateAccess.Unlock()
scope.Add(func() error {
e.stateAccess.Lock()
e.updateState(func(state *clientState) {
state.started = false
})
e.stateAccess.Unlock()
return nil
})
loopGroup.Go(func() {
e.readLoop(scope.Context())
})
loopGroup.Go(func() {
e.watchAuthForms(scope.Context())
})
loopGroup.Go(func() {
e.watchActiveTransport(scope.Context())
})
}
err := e.client.Start()
if err != nil {
return err
}
e.stateAccess.Lock()
e.updateState(func(state *clientState) {
state.started = true
})
e.readLoopDone = make(chan struct{})
e.authFormLoopDone = make(chan struct{})
e.activeTransportLoopDone = make(chan struct{})
e.stateAccess.Unlock()
go e.readLoop()
go e.watchAuthForms()
go e.watchActiveTransport()
return nil
}
func (e *Endpoint) readLoop() {
defer close(e.readLoopDone)
func (e *Endpoint) readLoop(ctx context.Context) {
for {
packetBuffers, err := e.client.ReadDataPackets(e.loopContext)
packetBuffers, err := e.client.ReadDataPackets(ctx)
if err != nil {
if E.IsClosedOrCanceled(err) || e.loopContext.Err() != nil {
if E.IsClosedOrCanceled(err) || ctx.Err() != nil {
return
}
e.logger.Error(E.Cause(err, "client terminated"))
@@ -463,30 +486,6 @@ func (e *Endpoint) readLoop() {
}
}
func (e *Endpoint) Close() error {
e.stateAccess.Lock()
e.updateState(func(state *clientState) {
state.started = false
})
readLoopDone := e.readLoopDone
authFormLoopDone := e.authFormLoopDone
activeTransportLoopDone := e.activeTransportLoopDone
e.stateAccess.Unlock()
e.cancelLoop()
err := E.Errors(e.client.Close(), e.device.Close())
if readLoopDone != nil {
<-readLoopDone
}
if authFormLoopDone != nil {
<-authFormLoopDone
}
if activeTransportLoopDone != nil {
<-activeTransportLoopDone
}
e.notifyStatusUpdated()
return err
}
func (e *Endpoint) InterfaceUpdated(ctx context.Context) {
e.client.RestartSession()
}
+40 -42
View File
@@ -30,18 +30,19 @@ func RegisterDNSTransport(registry *dns.TransportRegistry) {
type DNSTransport struct {
dns.TransportAdapter
ctx context.Context
logger logger.ContextLogger
endpointTag string
acceptDefaultResolvers bool
acceptSearchDomain bool
endpointManager adapter.EndpointManager
endpoint *Endpoint
dialer N.Dialer
access sync.RWMutex
closed bool
routes []openConnectDNSRoute
searchDomains []string
defaultResolvers []adapter.DNSTransport
resolverScope *adapter.Scope
}
type openConnectDNSRoute struct {
@@ -55,6 +56,7 @@ func NewDNSTransport(ctx context.Context, logger log.ContextLogger, tag string,
}
return &DNSTransport{
TransportAdapter: dns.NewTransportAdapter(C.DNSTypeOpenConnect, tag, nil),
ctx: ctx,
logger: logger,
endpointTag: options.Endpoint,
acceptDefaultResolvers: options.AcceptDefaultResolvers,
@@ -63,7 +65,7 @@ func NewDNSTransport(ctx context.Context, logger log.ContextLogger, tag string,
}, nil
}
func (t *DNSTransport) Start(stage adapter.StartStage) error {
func (t *DNSTransport) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
@@ -75,13 +77,34 @@ func (t *DNSTransport) Start(stage adapter.StartStage) error {
if !isOpenConnect {
return E.New("endpoint is not OpenConnect: ", t.endpointTag)
}
scope.Add(func() error {
t.access.Lock()
resolverScope := t.resolverScope
t.closed = true
t.routes = nil
t.searchDomains = nil
t.defaultResolvers = nil
t.resolverScope = nil
t.access.Unlock()
if resolverScope != nil {
return resolverScope.Close()
}
return nil
})
openConnectEndpoint.dnsTransportAccess.Lock()
if openConnectEndpoint.dnsTransport != nil && openConnectEndpoint.dnsTransport.Tag() != t.Tag() {
openConnectEndpoint.dnsTransportAccess.Unlock()
return E.New("only one DNS server is allowed for an endpoint")
}
openConnectEndpoint.dnsTransport = t
t.endpoint = openConnectEndpoint
scope.Add(func() error {
openConnectEndpoint.dnsTransportAccess.Lock()
if openConnectEndpoint.dnsTransport == t {
openConnectEndpoint.dnsTransport = nil
}
openConnectEndpoint.dnsTransportAccess.Unlock()
return nil
})
t.dialer = openConnectEndpoint
state := openConnectEndpoint.state.Load()
if state.started && state.tunnelConfigured && openConnectEndpoint.client.Ready() {
@@ -176,32 +199,29 @@ func (t *DNSTransport) updateConfiguration(configuration openconnecttransport.Co
defaultResolvers = nil
}
resolverScope := adapter.NewScope(t.ctx, t.logger)
for _, resolver := range resolverByAddress {
err := resolver.Start(adapter.StartStateStart, resolverScope)
if err != nil {
resolverScope.Close()
t.logger.Error(E.Cause(err, "start DNS resolver"))
return
}
}
t.access.Lock()
if t.closed {
t.access.Unlock()
for _, resolver := range resolverByAddress {
_ = resolver.Close()
}
resolverScope.Close()
return
}
oldResolvers := t.collectResolversLocked()
oldResolverScope := t.resolverScope
t.routes = routes
t.searchDomains = searchDomains
t.defaultResolvers = defaultResolvers
activeResolvers := t.collectResolversLocked()
t.resolverScope = resolverScope
t.access.Unlock()
for _, resolver := range oldResolvers {
_ = resolver.Close()
}
activeResolverSet := make(map[adapter.DNSTransport]bool, len(activeResolvers))
for _, resolver := range activeResolvers {
activeResolverSet[resolver] = true
}
for _, resolver := range resolverByAddress {
if !activeResolverSet[resolver] {
_ = resolver.Close()
}
if oldResolverScope != nil {
oldResolverScope.Close()
}
if len(resolverByAddress) > 0 {
t.logger.Info("updated ", len(routes), " DNS routes and ", len(resolverByAddress), " resolvers")
@@ -219,28 +239,6 @@ func (t *DNSTransport) Reset() {
}
}
func (t *DNSTransport) Close() error {
if t.endpoint != nil {
t.endpoint.dnsTransportAccess.Lock()
if t.endpoint.dnsTransport == t {
t.endpoint.dnsTransport = nil
}
t.endpoint.dnsTransportAccess.Unlock()
}
t.access.Lock()
resolvers := t.collectResolversLocked()
t.closed = true
t.routes = nil
t.searchDomains = nil
t.defaultResolvers = nil
t.access.Unlock()
var closeErr error
for _, resolver := range resolvers {
closeErr = E.Errors(closeErr, resolver.Close())
}
return closeErr
}
func (t *DNSTransport) PreferredDomain(domain string) bool {
canonicalDomain := mDNS.Fqdn(canonicalOpenConnectDomain(domain))
t.access.RLock()
+5 -6
View File
@@ -1,6 +1,7 @@
package openconnect
import (
"context"
"net/http"
"slices"
@@ -132,8 +133,7 @@ func (e *Endpoint) setTerminalError(err error) {
e.statusAccess.Unlock()
}
func (e *Endpoint) watchAuthForms() {
defer close(e.authFormLoopDone)
func (e *Endpoint) watchAuthForms(ctx context.Context) {
var loggedAuthChallengeID string
for {
authChallengeUpdated := e.client.AuthChallengeUpdated()
@@ -148,15 +148,14 @@ func (e *Endpoint) watchAuthForms() {
}
e.notifyStatusUpdated()
select {
case <-e.loopContext.Done():
case <-ctx.Done():
return
case <-authChallengeUpdated:
}
}
}
func (e *Endpoint) watchActiveTransport() {
defer close(e.activeTransportLoopDone)
func (e *Endpoint) watchActiveTransport(ctx context.Context) {
for {
transportUpdated := e.client.ActiveTransportUpdated()
transport := e.client.ActiveTransport()
@@ -167,7 +166,7 @@ func (e *Endpoint) watchActiveTransport() {
e.stateAccess.Unlock()
e.notifyStatusUpdated()
select {
case <-e.loopContext.Done():
case <-ctx.Done():
return
case <-transportUpdated:
}
+60 -57
View File
@@ -42,23 +42,21 @@ var (
type ClientEndpoint struct {
endpointBase
ctx context.Context
loopContext context.Context
cancelLoop context.CancelFunc
dnsRouter adapter.DNSRouter
outboundDialer N.Dialer
queryOptions adapter.DNSQueryOptions
client *ovpn.Client
device ovpntransport.Device
stateAccess sync.Mutex
state atomic.Pointer[clientState]
dnsTransport *DNSTransport
deviceStarted bool
readLoopDone chan struct{}
statusAccess sync.Mutex
statusUpdated chan struct{}
terminalError string
challengeLoopDone chan struct{}
ctx context.Context
loopContext context.Context
cancelLoop context.CancelFunc
dnsRouter adapter.DNSRouter
outboundDialer N.Dialer
queryOptions adapter.DNSQueryOptions
client *ovpn.Client
device ovpntransport.Device
stateAccess sync.Mutex
state atomic.Pointer[clientState]
dnsTransport *DNSTransport
deviceStarted bool
statusAccess sync.Mutex
statusUpdated chan struct{}
terminalError string
}
type clientState struct {
@@ -596,32 +594,57 @@ func (c *ClientEndpoint) uninstallDNSTransport(dnsTransport *DNSTransport) {
c.stateAccess.Unlock()
}
func (c *ClientEndpoint) Start(stage adapter.StartStage) error {
if stage != adapter.StartStatePostStart {
return nil
func (c *ClientEndpoint) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateInitialize:
scope.Add(func() error {
c.notifyStatusUpdated()
return nil
})
scope.Add(c.device.Close)
scope.Add(func() error {
c.cancelLoop()
return nil
})
case adapter.StartStatePostStart:
var loopGroup sync.WaitGroup
scope.Add(func() error {
loopGroup.Wait()
return nil
})
err := c.client.Start()
if err != nil {
return err
}
scope.Add(c.client.Close)
c.stateAccess.Lock()
c.updateState(func(state *clientState) {
state.started = true
})
c.stateAccess.Unlock()
scope.Add(func() error {
c.stateAccess.Lock()
c.updateState(func(state *clientState) {
state.started = false
})
c.stateAccess.Unlock()
return nil
})
loopGroup.Go(func() {
c.readLoop(scope.Context())
})
loopGroup.Go(func() {
c.watchChallenges(scope.Context())
})
}
err := c.client.Start()
if err != nil {
return err
}
c.stateAccess.Lock()
c.updateState(func(state *clientState) {
state.started = true
})
c.readLoopDone = make(chan struct{})
c.challengeLoopDone = make(chan struct{})
c.stateAccess.Unlock()
go c.readLoop()
go c.watchChallenges()
return nil
}
func (c *ClientEndpoint) readLoop() {
defer close(c.readLoopDone)
func (c *ClientEndpoint) readLoop(ctx context.Context) {
for {
packetBuffers, err := c.client.ReadDataPackets(c.loopContext)
packetBuffers, err := c.client.ReadDataPackets(ctx)
if err != nil {
if E.IsClosedOrCanceled(err) || c.loopContext.Err() != nil {
if E.IsClosedOrCanceled(err) || ctx.Err() != nil {
return
}
c.logger.Error(E.Cause(err, "client terminated"))
@@ -639,26 +662,6 @@ func (c *ClientEndpoint) readLoop() {
}
}
func (c *ClientEndpoint) Close() error {
c.stateAccess.Lock()
c.updateState(func(state *clientState) {
state.started = false
})
readLoopDone := c.readLoopDone
challengeLoopDone := c.challengeLoopDone
c.stateAccess.Unlock()
c.cancelLoop()
err := E.Errors(c.client.Close(), c.device.Close())
if readLoopDone != nil {
<-readLoopDone
}
if challengeLoopDone != nil {
<-challengeLoopDone
}
c.notifyStatusUpdated()
return err
}
func (c *ClientEndpoint) InterfaceUpdated(ctx context.Context) {
c.client.RestartSession()
}
+50 -47
View File
@@ -43,7 +43,6 @@ type DNSTransport struct {
acceptDefaultResolvers bool
acceptSearchDomain bool
endpointManager adapter.EndpointManager
endpoint *ClientEndpoint
dialer N.Dialer
updateAccess sync.Mutex
access sync.RWMutex
@@ -51,6 +50,7 @@ type DNSTransport struct {
routes map[string][]adapter.DNSTransport
searchDomains []string
defaultResolvers []adapter.DNSTransport
resolverScope *adapter.Scope
}
func NewDNSTransport(ctx context.Context, logger log.ContextLogger, tag string, options option.OpenVPNDNSServerOptions) (adapter.DNSTransport, error) {
@@ -68,7 +68,7 @@ func NewDNSTransport(ctx context.Context, logger log.ContextLogger, tag string,
}, nil
}
func (t *DNSTransport) Start(stage adapter.StartStage) error {
func (t *DNSTransport) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
@@ -80,14 +80,31 @@ func (t *DNSTransport) Start(stage adapter.StartStage) error {
if !isOpenVPN {
return E.New("endpoint is not an OpenVPN client: ", t.endpointTag)
}
t.endpoint = endpoint
t.dialer = endpoint
scope.Add(func() error {
t.updateAccess.Lock()
t.access.Lock()
resolverScope := t.resolverScope
t.closed = true
t.routes = nil
t.searchDomains = nil
t.defaultResolvers = nil
t.resolverScope = nil
t.access.Unlock()
t.updateAccess.Unlock()
if resolverScope != nil {
return resolverScope.Close()
}
return nil
})
err := endpoint.installDNSTransport(t)
if err != nil {
t.endpoint = nil
t.dialer = nil
return err
}
scope.Add(func() error {
endpoint.uninstallDNSTransport(t)
return nil
})
return nil
}
@@ -110,7 +127,7 @@ func (t *DNSTransport) updateResolvers(configuration ovpntransport.Configuration
routes := make(map[string][]adapter.DNSTransport)
searchDomains := normalizeOpenVPNDomains(configuration.SearchDomains)
var defaultResolvers []adapter.DNSTransport
var newResolvers []adapter.DNSTransport
resolverScope := adapter.NewScope(t.ctx, t.logger)
servers := slices.Clone(configuration.DNSServers)
slices.SortFunc(servers, func(left ovpntransport.DNSServer, right ovpntransport.DNSServer) int {
return left.Priority - right.Priority
@@ -119,18 +136,21 @@ func (t *DNSTransport) updateResolvers(configuration ovpntransport.Configuration
if len(servers) > 0 {
server := servers[0]
if server.DNSSEC == "yes" {
return t.failResolverUpdate(newResolvers, E.New("DNSSEC validation is required but is not supported"))
return t.failResolverUpdate(resolverScope, E.New("DNSSEC validation is required but is not supported"))
}
for _, address := range server.Addresses {
resolver, err := t.createResolver(server, address)
if err != nil {
return t.failResolverUpdate(newResolvers, err)
return t.failResolverUpdate(resolverScope, err)
}
err = resolver.Start(adapter.StartStateStart, resolverScope)
if err != nil {
return t.failResolverUpdate(resolverScope, err)
}
selectedResolvers = append(selectedResolvers, resolver)
newResolvers = append(newResolvers, resolver)
}
if len(selectedResolvers) == 0 {
return t.failResolverUpdate(newResolvers, E.New("DNS server ", server.Priority, " has no addresses"))
return t.failResolverUpdate(resolverScope, E.New("DNS server ", server.Priority, " has no addresses"))
}
if len(server.ResolveDomains) == 0 {
defaultResolvers = slices.Clone(selectedResolvers)
@@ -145,12 +165,15 @@ func (t *DNSTransport) updateResolvers(configuration ovpntransport.Configuration
} else {
for _, address := range configuration.DNS {
resolver := dnsTransport.NewUDPRaw(t.logger, t.TransportAdapter, t.dialer, M.SocksaddrFrom(address, 53))
err := resolver.Start(adapter.StartStateStart, resolverScope)
if err != nil {
return t.failResolverUpdate(resolverScope, err)
}
selectedResolvers = append(selectedResolvers, resolver)
newResolvers = append(newResolvers, resolver)
}
if len(configuration.DNSRoutes) > 0 {
if len(selectedResolvers) == 0 {
return t.failResolverUpdate(newResolvers, E.New("DOMAIN-ROUTE requires traditional pushed DNS servers"))
return t.failResolverUpdate(resolverScope, E.New("DOMAIN-ROUTE requires traditional pushed DNS servers"))
}
for _, domain := range configuration.DNSRoutes {
normalizedDomain := normalizeOpenVPNDomain(domain)
@@ -163,32 +186,40 @@ func (t *DNSTransport) updateResolvers(configuration ovpntransport.Configuration
}
}
if len(searchDomains) > 0 && len(selectedResolvers) == 0 {
return t.failResolverUpdate(newResolvers, E.New("search domains require pushed DNS servers"))
return t.failResolverUpdate(resolverScope, E.New("search domains require pushed DNS servers"))
}
for _, searchDomain := range searchDomains {
routes[searchDomain] = slices.Clone(selectedResolvers)
}
t.access.Lock()
oldResolvers := t.collectResolversLocked()
oldResolverScope := t.resolverScope
t.routes = routes
t.searchDomains = searchDomains
t.defaultResolvers = defaultResolvers
t.resolverScope = resolverScope
t.access.Unlock()
closeErr := closeDNSTransports(oldResolvers)
var closeErr error
if oldResolverScope != nil {
closeErr = oldResolverScope.Close()
}
t.logger.Info("updated ", len(routes), " DNS routes, ", len(searchDomains), " search domains and ", len(defaultResolvers), " default resolvers")
return closeErr
}
func (t *DNSTransport) failResolverUpdate(newResolvers []adapter.DNSTransport, updateErr error) error {
newCloseErr := closeDNSTransports(newResolvers)
func (t *DNSTransport) failResolverUpdate(resolverScope *adapter.Scope, updateErr error) error {
newCloseErr := resolverScope.Close()
t.access.Lock()
oldResolvers := t.collectResolversLocked()
oldResolverScope := t.resolverScope
t.routes = nil
t.searchDomains = nil
t.defaultResolvers = nil
t.resolverScope = nil
t.access.Unlock()
oldCloseErr := closeDNSTransports(oldResolvers)
var oldCloseErr error
if oldResolverScope != nil {
oldCloseErr = oldResolverScope.Close()
}
return E.Errors(updateErr, newCloseErr, oldCloseErr)
}
@@ -255,24 +286,6 @@ func (t *DNSTransport) Reset() {
}
}
func (t *DNSTransport) Close() error {
if t.endpoint != nil {
t.endpoint.uninstallDNSTransport(t)
}
t.updateAccess.Lock()
t.access.Lock()
resolvers := t.collectResolversLocked()
t.closed = true
t.routes = nil
t.searchDomains = nil
t.defaultResolvers = nil
t.access.Unlock()
t.endpoint = nil
t.dialer = nil
t.updateAccess.Unlock()
return closeDNSTransports(resolvers)
}
func (t *DNSTransport) Raw() bool {
return true
}
@@ -384,16 +397,6 @@ func (t *DNSTransport) collectResolversLocked() []adapter.DNSTransport {
return common.Uniq(resolvers)
}
func closeDNSTransports(resolvers []adapter.DNSTransport) error {
var err error
for _, resolver := range common.Uniq(resolvers) {
err = E.Append(err, resolver.Close(), func(closeErr error) error {
return E.Cause(closeErr, "close DNS resolver")
})
}
return err
}
func normalizeOpenVPNDomain(domain string) string {
normalized := strings.TrimSpace(strings.ToLower(domain))
if normalized == "." {
+27 -37
View File
@@ -6,6 +6,7 @@ import (
"net/netip"
"slices"
"strconv"
"sync"
"sync/atomic"
"time"
@@ -35,7 +36,6 @@ var (
type ServerEndpoint struct {
endpointBase
ctx context.Context
loopContext context.Context
cancelLoop context.CancelFunc
options option.OpenVPNServerEndpointOptions
serverOptions ovpn.ServerOptions
@@ -45,7 +45,6 @@ type ServerEndpoint struct {
device ovpntransport.Device
localAddresses []netip.Prefix
started atomic.Bool
readLoopDone chan struct{}
}
type udpEgressPacketConn struct {
@@ -80,7 +79,6 @@ func NewServerEndpoint(ctx context.Context, router adapter.Router, logger log.Co
logger: logger,
},
ctx: ctx,
loopContext: loopContext,
cancelLoop: cancelLoop,
options: options,
dnsRouter: service.FromContext[adapter.DNSRouter](ctx),
@@ -161,7 +159,15 @@ func validateServerTopology(topology string) error {
}
}
func (s *ServerEndpoint) Start(stage adapter.StartStage) error {
func (s *ServerEndpoint) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage == adapter.StartStateInitialize {
scope.Add(s.device.Close)
scope.Add(func() error {
s.cancelLoop()
return nil
})
return nil
}
if stage != adapter.StartStateStart {
return nil
}
@@ -225,6 +231,7 @@ func (s *ServerEndpoint) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
scope.Add(s.listener.Close)
serverOptions := s.serverOptions
if streamListener != nil {
serverOptions.Transport.ListenAddress = streamListener.Addr().String()
@@ -238,26 +245,31 @@ func (s *ServerEndpoint) Start(stage adapter.StartStage) error {
if packetConn != nil {
_ = packetConn.Close()
}
s.listener.Close()
return err
}
s.server = server
var loopGroup sync.WaitGroup
scope.Add(func() error {
loopGroup.Wait()
return nil
})
scope.Add(server.Close)
err = s.device.Start()
if err != nil {
s.listener.Close()
server.Close()
return err
}
err = server.Start()
if err != nil {
s.device.Close()
s.listener.Close()
server.Close()
return err
}
s.started.Store(true)
s.readLoopDone = make(chan struct{})
go s.readLoop()
scope.Add(func() error {
s.started.Store(false)
return nil
})
loopGroup.Go(func() {
s.readLoop(scope.Context())
})
return nil
}
@@ -598,12 +610,11 @@ func applyServerPushOptions(serverOptions *ovpn.ServerOptions, options option.Op
return nil
}
func (s *ServerEndpoint) readLoop() {
defer close(s.readLoopDone)
func (s *ServerEndpoint) readLoop(ctx context.Context) {
for {
serverPacketBuffers, err := s.server.ReadDataPackets(s.loopContext)
serverPacketBuffers, err := s.server.ReadDataPackets(ctx)
if err != nil {
if E.IsClosedOrCanceled(err) || s.loopContext.Err() != nil {
if E.IsClosedOrCanceled(err) || ctx.Err() != nil {
return
}
s.logger.Error(E.Cause(err, "server terminated"))
@@ -622,27 +633,6 @@ func (s *ServerEndpoint) readLoop() {
}
}
func (s *ServerEndpoint) Close() error {
s.started.Store(false)
s.cancelLoop()
var serverErr error
if s.server != nil {
serverErr = s.server.Close()
}
if s.readLoopDone != nil {
<-s.readLoopDone
}
var deviceErr error
if s.device != nil {
deviceErr = s.device.Close()
}
var listenerErr error
if s.listener != nil {
listenerErr = s.listener.Close()
}
return E.Errors(serverErr, deviceErr, listenerErr)
}
func (s *ServerEndpoint) PreMatchFlow(network string, destination netip.Addr) adapter.PreMatchAction {
return adapter.PreMatchFlow
}
+3 -3
View File
@@ -1,6 +1,7 @@
package openvpn
import (
"context"
"slices"
"github.com/sagernet/sing-box/adapter"
@@ -81,8 +82,7 @@ func (c *ClientEndpoint) setTerminalError(err error) {
c.statusAccess.Unlock()
}
func (c *ClientEndpoint) watchChallenges() {
defer close(c.challengeLoopDone)
func (c *ClientEndpoint) watchChallenges(ctx context.Context) {
var loggedChallengeID string
for {
challengeUpdated := c.client.ChallengeUpdated()
@@ -93,7 +93,7 @@ func (c *ClientEndpoint) watchChallenges() {
}
c.notifyStatusUpdated()
select {
case <-c.loopContext.Done():
case <-ctx.Done():
return
case <-challengeUpdated:
}
+7 -6
View File
@@ -42,15 +42,16 @@ func NewRedirect(ctx context.Context, router adapter.Router, logger log.ContextL
return redirect, nil
}
func (h *Redirect) Start(stage adapter.StartStage) error {
func (h *Redirect) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
return h.listener.Start()
}
func (h *Redirect) Close() error {
return h.listener.Close()
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
func (h *Redirect) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+5 -8
View File
@@ -70,7 +70,7 @@ func NewTProxy(ctx context.Context, router adapter.Router, logger log.ContextLog
return tproxy, nil
}
func (t *TProxy) Start(stage adapter.StartStage) error {
func (t *TProxy) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -78,22 +78,19 @@ func (t *TProxy) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
scope.Add(t.udpNat.Close)
err = t.listener.Start()
if err != nil {
_ = t.udpNat.Close()
return err
}
return err
scope.Add(t.listener.Close)
return nil
}
func (t *TProxy) InterfaceUpdated(ctx context.Context) {
t.udpNat.Purge()
}
func (t *TProxy) Close() error {
_ = t.udpNat.Close()
return t.listener.Close()
}
func (t *TProxy) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
metadata.Inbound = t.Tag()
metadata.InboundType = t.Type()
+7 -6
View File
@@ -95,15 +95,16 @@ func newInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, err
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
return h.listener.Start()
}
func (h *Inbound) Close() error {
return h.listener.Close()
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
//nolint:staticcheck
+7 -6
View File
@@ -109,15 +109,16 @@ func newMultiInbound(ctx context.Context, router adapter.Router, logger log.Cont
return inbound, err
}
func (h *MultiInbound) Start(stage adapter.StartStage) error {
func (h *MultiInbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
return h.listener.Start()
}
func (h *MultiInbound) Close() error {
return h.listener.Close()
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
func (h *MultiInbound) SetTracker(tracker adapter.SSMTracker) {
+7 -6
View File
@@ -86,15 +86,16 @@ func newRelayInbound(ctx context.Context, router adapter.Router, logger log.Cont
return inbound, err
}
func (h *RelayInbound) Start(stage adapter.StartStage) error {
func (h *RelayInbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
return h.listener.Start()
}
func (h *RelayInbound) Close() error {
return h.listener.Close()
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
//nolint:staticcheck
+8 -2
View File
@@ -152,8 +152,14 @@ func (h *Outbound) CloseIdleConnections() {
}
}
func (h *Outbound) Close() error {
return common.Close(common.PtrOrNil(h.multiplexDialer))
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
if h.multiplexDialer != nil {
scope.Add(h.multiplexDialer.Close)
}
return nil
}
var _ N.Dialer = (*shadowsocksDialer)(nil)
+7 -6
View File
@@ -104,15 +104,16 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
return h.listener.Start()
}
func (h *Inbound) Close() error {
return h.listener.Close()
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+7 -6
View File
@@ -119,15 +119,16 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
return h.listener.Start()
}
func (h *Inbound) Close() error {
return h.listener.Close()
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+6 -2
View File
@@ -163,6 +163,10 @@ func (h *Outbound) CloseIdleConnections() {
h.client.CloseIdleConnections()
}
func (h *Outbound) Close() error {
return h.client.Close()
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
scope.Add(h.client.Close)
return nil
}
+7 -6
View File
@@ -60,15 +60,16 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
return h.listener.Start()
}
func (h *Inbound) Close() error {
return h.listener.Close()
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+8 -2
View File
@@ -260,8 +260,14 @@ func (s *Outbound) releaseStream(client *ssh.Client, keepSession bool) {
common.Close(clientConn)
}
func (s *Outbound) Close() error {
return common.Close(s.clientConn)
func (s *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
scope.Add(func() error {
return common.Close(s.clientConn)
})
return nil
}
func (s *Outbound) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) {
+1 -5
View File
@@ -63,7 +63,7 @@ func NewCertificateProvider(ctx context.Context, _ log.ContextLogger, tag string
}, nil
}
func (p *CertificateProvider) Start(stage adapter.StartStage) error {
func (p *CertificateProvider) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -82,10 +82,6 @@ func (p *CertificateProvider) Start(stage adapter.StartStage) error {
return nil
}
func (p *CertificateProvider) Close() error {
return nil
}
func (p *CertificateProvider) GetCertificate(clientHello *tls.ClientHelloInfo) (*tls.Certificate, error) {
localClient := p.localClient
if localClient == nil {
+49 -32
View File
@@ -59,6 +59,7 @@ type DNSTransport struct {
magicHosts nDNSResolver.MagicDNSHosts
searchDomains []string
defaultResolvers []adapter.DNSTransport
resolverScope *adapter.Scope
}
func NewDNSTransport(ctx context.Context, logger log.ContextLogger, tag string, options option.TailscaleDNSServerOptions) (adapter.DNSTransport, error) {
@@ -77,7 +78,7 @@ func NewDNSTransport(ctx context.Context, logger log.ContextLogger, tag string,
}, nil
}
func (t *DNSTransport) Start(stage adapter.StartStage) error {
func (t *DNSTransport) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
@@ -94,6 +95,21 @@ func (t *DNSTransport) Start(stage adapter.StartStage) error {
}
ep.onReconfigHook = t.onReconfig
t.endpoint = ep
scope.Add(func() error {
t.access.Lock()
resolverScope := t.resolverScope
t.routePrefixes = nil
t.routes = nil
t.hosts = nil
t.magicHosts = nil
t.defaultResolvers = nil
t.resolverScope = nil
t.access.Unlock()
if resolverScope != nil {
return resolverScope.Close()
}
return nil
})
return nil
}
@@ -119,13 +135,14 @@ func (t *DNSTransport) updateDNSServers(routeConfig *router.Config, dnsConfig *n
directDialer := common.Must1(dialer.NewDefault(t.ctx, option.DialerOptions{}))
return &DNSDialer{transport: t, fallbackDialer: directDialer}
})
resolverScope := adapter.NewScope(t.ctx, t.logger)
routes := make(map[string][]adapter.DNSTransport)
for domain, resolvers := range dnsConfig.Routes {
var myResolvers []adapter.DNSTransport
for _, resolver := range resolvers {
myResolver, err := t.createResolver(directDialerOnce, resolver)
myResolver, err := t.createResolver(resolverScope, directDialerOnce, resolver)
if err != nil {
return err
return E.Errors(err, resolverScope.Close())
}
myResolvers = append(myResolvers, myResolver)
}
@@ -140,25 +157,26 @@ func (t *DNSTransport) updateDNSServers(routeConfig *router.Config, dnsConfig *n
})
var defaultResolvers []adapter.DNSTransport
for _, resolver := range dnsConfig.DefaultResolvers {
myResolver, err := t.createResolver(directDialerOnce, resolver)
myResolver, err := t.createResolver(resolverScope, directDialerOnce, resolver)
if err != nil {
return err
return E.Errors(err, resolverScope.Close())
}
defaultResolvers = append(defaultResolvers, myResolver)
}
t.access.Lock()
oldResolvers := t.collectResolversLocked()
oldResolverScope := t.resolverScope
t.routePrefixes = routePrefixes
t.routes = routes
t.hosts = hosts
t.magicHosts = t.endpoint.server.ExportLocalBackend().ExportMagicDNSHosts()
t.searchDomains = searchDomains
t.defaultResolvers = defaultResolvers
t.resolverScope = resolverScope
t.access.Unlock()
for _, transport := range oldResolvers {
transport.Close()
if oldResolverScope != nil {
oldResolverScope.Close()
}
if len(defaultResolvers) > 0 {
@@ -170,7 +188,7 @@ func (t *DNSTransport) updateDNSServers(routeConfig *router.Config, dnsConfig *n
return nil
}
func (t *DNSTransport) createResolver(directDialer func() N.Dialer, resolver *dnstype.Resolver) (adapter.DNSTransport, error) {
func (t *DNSTransport) createResolver(resolverScope *adapter.Scope, directDialer func() N.Dialer, resolver *dnstype.Resolver) (adapter.DNSTransport, error) {
serverURL, parseURLErr := url.Parse(resolver.Addr)
isHTTPScheme := parseURLErr == nil && (serverURL.Scheme == "http" || serverURL.Scheme == "https")
var myDialer N.Dialer
@@ -181,6 +199,10 @@ func (t *DNSTransport) createResolver(directDialer func() N.Dialer, resolver *dn
}
if len(resolver.BootstrapResolution) > 0 {
bootstrapTransport := transport.NewUDPRaw(t.logger, t.TransportAdapter, myDialer, M.SocksaddrFrom(resolver.BootstrapResolution[0], 53))
err := bootstrapTransport.Start(adapter.StartStateStart, resolverScope)
if err != nil {
return nil, err
}
myDialer = dialer.NewResolveDialer(t.ctx, myDialer, false, "", adapter.DNSQueryOptions{Transport: bootstrapTransport}, 0)
} else {
myDialer = dialer.NewResolveDialer(t.ctx, myDialer, false, "", t.endpoint.queryOptions, 0)
@@ -202,12 +224,22 @@ func (t *DNSTransport) createResolver(directDialer func() N.Dialer, resolver *dn
if err != nil {
return nil, E.Cause(err, "create TLS config for resolver ", resolver.Addr)
}
return transport.NewHTTPSRaw(t.TransportAdapter, t.logger, myDialer, serverURL, http.Header{}, serverAddr, tlsConfig), nil
httpsTransport := transport.NewHTTPSRaw(t.TransportAdapter, t.logger, myDialer, serverURL, http.Header{}, serverAddr, tlsConfig)
err = httpsTransport.Start(adapter.StartStateStart, resolverScope)
if err != nil {
return nil, err
}
return httpsTransport, nil
case "http":
if serverAddr.Port == 0 {
serverAddr.Port = 80
}
return transport.NewHTTPSRaw(t.TransportAdapter, t.logger, myDialer, serverURL, http.Header{}, serverAddr, nil), nil
httpTransport := transport.NewHTTPSRaw(t.TransportAdapter, t.logger, myDialer, serverURL, http.Header{}, serverAddr, nil)
err := httpTransport.Start(adapter.StartStateStart, resolverScope)
if err != nil {
return nil, err
}
return httpTransport, nil
}
}
serverAddr := M.ParseSocksaddr(resolver.Addr)
@@ -220,7 +252,12 @@ func (t *DNSTransport) createResolver(directDialer func() N.Dialer, resolver *dn
if serverAddr.Port == 0 {
serverAddr.Port = 53
}
return transport.NewUDPRaw(t.logger, t.TransportAdapter, myDialer, serverAddr), nil
udpTransport := transport.NewUDPRaw(t.logger, t.TransportAdapter, myDialer, serverAddr)
err := udpTransport.Start(adapter.StartStateStart, resolverScope)
if err != nil {
return nil, err
}
return udpTransport, nil
}
func buildRoutePrefixes(routeConfig *router.Config) []netip.Prefix {
@@ -244,26 +281,6 @@ func buildRoutePrefixes(routeConfig *router.Config) []netip.Prefix {
return ipSet.Prefixes()
}
func (t *DNSTransport) Close() error {
t.access.Lock()
transports := t.collectResolversLocked()
t.routePrefixes = nil
t.routes = nil
t.hosts = nil
t.magicHosts = nil
t.defaultResolvers = nil
t.access.Unlock()
var err error
for _, transport := range transports {
name := "resolver/" + transport.Type() + "[" + transport.Tag() + "]"
err = E.Append(err, transport.Close(), func(err error) error {
return E.Cause(err, "close ", name)
})
}
return err
}
func (t *DNSTransport) Raw() bool {
return true
}
+59 -68
View File
@@ -48,7 +48,6 @@ import (
tailscaleroot "github.com/sagernet/tailscale"
_ "github.com/sagernet/tailscale/feature/relayserver"
"github.com/sagernet/tailscale/ipn"
"github.com/sagernet/tailscale/ipn/ipnlocal"
tsDNS "github.com/sagernet/tailscale/net/dns"
"github.com/sagernet/tailscale/net/netmon"
"github.com/sagernet/tailscale/net/netns"
@@ -93,7 +92,6 @@ type Endpoint struct {
detour string
server *tsnet.Server
stack *stack.Stack
icmpForwarder *tun.ICMPForwarder
returnAccess sync.Mutex
returnPath tun.Return
wgEngine wgengine.ExportedUserspaceEngine
@@ -122,17 +120,13 @@ type Endpoint struct {
sshServerInstance *tailssh.Server
sshServerOptions *option.TailscaleSSHServerOptions
taildrop *taildropManager
localBackend *ipnlocal.LocalBackend
systemInterface bool
systemInterfaceName string
systemInterfaceMTU uint32
keyAuth bool
serverStarted bool
started atomic.Bool
systemTun tun.Tun
systemDialer *dialer.DefaultDialer
fallbackTCPCloser func()
}
func NewEndpoint(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.TailscaleEndpointOptions) (adapter.Endpoint, error) {
@@ -261,7 +255,7 @@ func (t *Endpoint) References() []string {
return []string{t.detour}
}
func (t *Endpoint) Start(stage adapter.StartStage) error {
func (t *Endpoint) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateInitialize:
mkdirErr := filemanager.MkdirAll(t.ctx, t.server.Dir, 0o700)
@@ -276,14 +270,14 @@ func (t *Endpoint) Start(stage adapter.StartStage) error {
}
t.server.PeerDNSQueryHandler = (*peerDNSQueryHandler)(t)
case adapter.StartStateStart:
return t.start()
return t.start(scope)
case adapter.StartStatePostStart:
return t.postStart()
return t.postStart(scope)
}
return nil
}
func (t *Endpoint) start() error {
func (t *Endpoint) start(scope *adapter.Scope) error {
if t.platformInterface != nil && t.platformInterface.UsePlatformNetworkInterfaces() {
err := t.network.UpdateInterfaces()
if err != nil {
@@ -308,6 +302,10 @@ func (t *Endpoint) start() error {
}
}), nil
})
scope.Add(func() error {
netmon.RegisterInterfaceGetter(nil)
return nil
})
}
if t.systemInterface {
mtu := t.systemInterfaceMTU
@@ -333,14 +331,13 @@ func (t *Endpoint) start() error {
if err != nil {
return err
}
scope.Add(systemTun.Close)
err = systemTun.Start()
if err != nil {
_ = systemTun.Close()
return err
}
wgTunDevice, err := newTunDeviceAdapter(systemTun, int(mtu), t.logger)
if err != nil {
_ = systemTun.Close()
return err
}
systemDialer, err := dialer.NewDefault(t.ctx, option.DialerOptions{
@@ -349,15 +346,17 @@ func (t *Endpoint) start() error {
},
})
if err != nil {
_ = systemTun.Close()
return err
}
t.systemTun = systemTun
t.systemDialer = systemDialer
t.server.Tun = wgTunDevice
}
if t.network.AutoRedirectOutputMark() != 0 {
netns.SetControlFunc(t.network.AutoRedirectOutputMarkFunc())
scope.Add(func() error {
netns.SetControlFunc(nil)
return nil
})
} else if t.platformInterface != nil && t.platformInterface.UsePlatformNetworkInterfaces() {
if t.platformInterface.UsePlatformAutoDetectInterfaceControl() {
netns.SetControlFunc(func(network, address string, conn syscall.RawConn) error {
@@ -365,6 +364,10 @@ func (t *Endpoint) start() error {
return t.platformInterface.AutoDetectInterfaceControl(int(fileDescriptor))
})
})
scope.Add(func() error {
netns.SetControlFunc(nil)
return nil
})
} else {
// NEPacketTunnelProvider sockets are excluded from tunnel routes by
// NECP; the empty override only suppresses tailscale's own
@@ -372,12 +375,21 @@ func (t *Endpoint) start() error {
netns.SetControlFunc(func(string, string, syscall.RawConn) error {
return nil
})
scope.Add(func() error {
netns.SetControlFunc(nil)
return nil
})
}
} else {
bindFunc := t.network.AutoDetectInterfaceFunc()
if bindFunc != nil {
netns.SetControlFunc(bindFunc)
netns.SetListenPacketFunc(t.listenPacket)
scope.Add(func() error {
netns.SetControlFunc(nil)
netns.SetListenPacketFunc(nil)
return nil
})
}
}
return nil
@@ -408,29 +420,35 @@ func (t *Endpoint) listenPacket(ctx context.Context, network string, address str
return tun.NewUDPEgressConn(udpConn, egressPool), nil
}
func (t *Endpoint) postStart() error {
func (t *Endpoint) postStart(scope *adapter.Scope) error {
err := t.server.Start()
if err != nil {
if t.systemTun != nil {
_ = t.systemTun.Close()
}
return err
}
t.serverStarted = true
if t.fallbackTCPCloser == nil {
t.fallbackTCPCloser = t.server.RegisterFallbackTCPHandler(func(src, dst netip.AddrPort) (handler func(net.Conn), intercept bool) {
return func(conn net.Conn) {
ctx := log.ContextWithNewID(t.ctx)
source := M.SocksaddrFrom(src.Addr(), src.Port())
destination := M.SocksaddrFrom(dst.Addr(), dst.Port())
t.NewConnectionEx(ctx, conn, source, destination, nil)
}, true
})
}
scope.Add(t.server.Close)
fallbackTCPCloser := t.server.RegisterFallbackTCPHandler(func(src, dst netip.AddrPort) (handler func(net.Conn), intercept bool) {
return func(conn net.Conn) {
ctx := log.ContextWithNewID(t.ctx)
source := M.SocksaddrFrom(src.Addr(), src.Port())
destination := M.SocksaddrFrom(dst.Addr(), dst.Port())
t.NewConnectionEx(ctx, conn, source, destination, nil)
}, true
})
scope.Add(func() error {
fallbackTCPCloser()
return nil
})
localBackend := t.server.ExportLocalBackend()
t.localBackend = localBackend
scope.Add(func() error {
t.taildrop.close()
return nil
})
if !version.IsAppleTV() {
registerTaildropEndpoint(localBackend, t)
scope.Add(func() error {
unregisterTaildropEndpoint(localBackend)
return nil
})
go t.taildrop.start()
}
wgEngine := localBackend.ExportEngine().(wgengine.ExportedUserspaceEngine)
@@ -447,10 +465,10 @@ func (t *Endpoint) postStart() error {
return gonet.TranslateNetstackError(gErr)
}
icmpForwarder := tun.NewICMPForwarder(ipStack, t, t.logger)
scope.Add(icmpForwarder.Close)
ipStack.SetTransportProtocolHandler(icmp.ProtocolNumber4, icmpForwarder.HandlePacket)
ipStack.SetTransportProtocolHandler(icmp.ProtocolNumber6, icmpForwarder.HandlePacket)
t.stack = ipStack
t.icmpForwarder = icmpForwarder
netstack := t.server.ExportNetstack()
if netstack != nil {
previousTCP := netstack.GetTCPHandlerForFlow
@@ -510,15 +528,20 @@ func (t *Endpoint) postStart() error {
if err != nil {
return E.Cause(err, "start SSH server")
}
scope.Add(sshServer.Close)
t.sshReconfigHook = sshServer.OnReconfig
t.sshServerInstance = sshServer
}
go t.watchState()
go t.watchState(scope.Context())
t.started.Store(true)
scope.Add(func() error {
t.started.Store(false)
return nil
})
return nil
}
func (t *Endpoint) watchState() {
func (t *Endpoint) watchState(ctx context.Context) {
localBackend := t.server.ExportLocalBackend()
var reportedAuthURL string
exitNodePending := t.exitNode != ""
@@ -533,7 +556,7 @@ func (t *Endpoint) watchState() {
}
for {
var busError string
localBackend.WatchNotifications(t.ctx, ipn.NotifyInitialState|ipn.NotifyPeerPatches, nil, func(roNotify *ipn.Notify) (keepGoing bool) {
localBackend.WatchNotifications(ctx, ipn.NotifyInitialState|ipn.NotifyPeerPatches, nil, func(roNotify *ipn.Notify) (keepGoing bool) {
if roNotify.ErrMessage != nil {
busError = *roNotify.ErrMessage
return false
@@ -578,7 +601,7 @@ func (t *Endpoint) watchState() {
}
return true
})
if t.ctx.Err() != nil {
if ctx.Err() != nil {
return
}
if busError != "" {
@@ -587,7 +610,7 @@ func (t *Endpoint) watchState() {
t.logger.Warn("state watcher stopped unexpectedly, restarting")
}
select {
case <-t.ctx.Done():
case <-ctx.Done():
return
case <-time.After(time.Second):
}
@@ -719,38 +742,6 @@ func (t *Endpoint) Logout(ctx context.Context) error {
return nil
}
func (t *Endpoint) Close() error {
var err error
t.started.Store(false)
if t.localBackend != nil {
unregisterTaildropEndpoint(t.localBackend)
t.localBackend = nil
}
t.taildrop.close()
if t.icmpForwarder != nil {
t.icmpForwarder.Close()
t.icmpForwarder = nil
}
common.Close(common.PtrOrNil(t.sshServerInstance))
t.sshServerInstance = nil
if t.serverStarted {
err = common.Close(common.PtrOrNil(t.server))
t.serverStarted = false
}
netmon.RegisterInterfaceGetter(nil)
netns.SetControlFunc(nil)
netns.SetListenPacketFunc(nil)
if t.fallbackTCPCloser != nil {
t.fallbackTCPCloser()
t.fallbackTCPCloser = nil
}
if t.systemTun != nil {
t.systemTun.Close()
t.systemTun = nil
}
return err
}
func (t *Endpoint) InterfaceUpdated(ctx context.Context) {
if !t.started.Load() {
return
+10 -22
View File
@@ -40,7 +40,6 @@ type Outbound struct {
startConf *tor.StartConf
options map[string]string
events chan control.Event
instance *tor.Tor
socksClient *socks.Client
}
@@ -96,7 +95,7 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL
}, nil
}
func (t *Outbound) Start(stage adapter.StartStage) error {
func (t *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateInitialize:
if t.startConf.DataDir == "" {
@@ -127,11 +126,8 @@ func (t *Outbound) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
err = t.start()
if err != nil {
t.Close()
return err
}
scope.Add(proxy.Close)
return t.start(scope)
}
return nil
}
@@ -144,13 +140,17 @@ var torLogEvents = []control.EventCode{
control.EventCodeLogWarn,
}
func (t *Outbound) start() error {
func (t *Outbound) start(scope *adapter.Scope) error {
t.events = make(chan control.Event, 8)
scope.Add(func() error {
close(t.events)
return nil
})
torInstance, err := tor.Start(t.ctx, t.startConf)
if err != nil {
return E.New(strings.ToLower(err.Error()))
}
t.instance = torInstance
t.events = make(chan control.Event, 8)
scope.Add(torInstance.Close)
err = torInstance.Control.AddEventListener(t.events, torLogEvents...)
if err != nil {
return err
@@ -225,18 +225,6 @@ func (t *Outbound) recvLoop() {
}
}
func (t *Outbound) Close() error {
err := common.Close(
common.PtrOrNil(t.proxy),
common.PtrOrNil(t.instance),
)
if t.events != nil {
close(t.events)
t.events = nil
}
return err
}
func (t *Outbound) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) {
t.logger.InfoContext(ctx, "outbound connection to ", destination)
return t.socksClient.DialContext(ctx, network, destination)
+10 -10
View File
@@ -120,7 +120,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -129,10 +129,18 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return E.Cause(err, "create TLS config")
}
scope.Add(h.tlsConfig.Close)
}
if h.transport == nil {
return h.listener.Start()
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
scope.Add(h.transport.Close)
scope.Add(h.listener.Close)
if common.Contains(h.transport.Network(), N.NetworkTCP) {
tcpListener, err := h.listener.ListenTCP()
if err != nil {
@@ -160,14 +168,6 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
return nil
}
func (h *Inbound) Close() error {
return common.Close(
h.listener,
h.tlsConfig,
h.transport,
)
}
func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
if h.tlsConfig != nil && h.transport == nil {
tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig)
+11 -2
View File
@@ -150,8 +150,17 @@ func (h *Outbound) CloseIdleConnections() {
}
}
func (h *Outbound) Close() error {
return common.Close(common.PtrOrNil(h.multiplexDialer), h.transport)
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
if h.transport != nil {
scope.Add(h.transport.Close)
}
if h.multiplexDialer != nil {
scope.Add(h.multiplexDialer.Close)
}
return nil
}
type trojanDialer Outbound
+9 -10
View File
@@ -153,7 +153,7 @@ func (h *Inbound) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn,
h.router.RoutePacketConnectionEx(ctx, conn, metadata, onClose)
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -162,18 +162,17 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
scope.Add(h.tlsConfig.Close)
}
packetConn, err := h.listener.ListenUDP()
if err != nil {
return err
}
return h.server.Start(packetConn)
}
func (h *Inbound) Close() error {
return common.Close(
h.listener,
h.tlsConfig,
common.PtrOrNil(h.server),
)
scope.Add(h.listener.Close)
err = h.server.Start(packetConn)
if err != nil {
return err
}
scope.Add(h.server.Close)
return nil
}
+8 -2
View File
@@ -161,6 +161,12 @@ func (h *Outbound) CloseIdleConnections() {
h.client.CloseIdleConnections()
}
func (h *Outbound) Close() error {
return h.client.CloseWithError(os.ErrClosed)
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
scope.Add(func() error {
return h.client.CloseWithError(os.ErrClosed)
})
return nil
}
+38 -38
View File
@@ -25,7 +25,6 @@ import (
M "github.com/sagernet/sing/common/metadata"
N "github.com/sagernet/sing/common/network"
"github.com/sagernet/sing/common/ranges"
"github.com/sagernet/sing/common/x/list"
"github.com/sagernet/sing/service"
"go4.org/netipx"
@@ -36,29 +35,27 @@ func RegisterInbound(registry *inbound.Registry) {
}
type Inbound struct {
tag string
ctx context.Context
router adapter.Router
networkManager adapter.NetworkManager
logger log.ContextLogger
tunOptions tun.Options
udpTimeout time.Duration
udpMapping tun.NATMapping
udpFiltering tun.NATFiltering
udpNATMax uint32
dnsHijackAddress []netip.Addr
stack string
tunIf tun.Tun
tunStack tun.Stack
platformInterface adapter.PlatformInterface
platformOptions option.TunPlatformOptions
autoRedirect tun.AutoRedirect
routeRuleSet []adapter.RuleSet
routeRuleSetCallback []*list.Element[adapter.RuleSetUpdateCallback]
routeExcludeRuleSet []adapter.RuleSet
routeExcludeRuleSetCallback []*list.Element[adapter.RuleSetUpdateCallback]
routeAddressSet []*netipx.IPSet
routeExcludeAddressSet []*netipx.IPSet
tag string
ctx context.Context
router adapter.Router
networkManager adapter.NetworkManager
logger log.ContextLogger
tunOptions tun.Options
udpTimeout time.Duration
udpMapping tun.NATMapping
udpFiltering tun.NATFiltering
udpNATMax uint32
dnsHijackAddress []netip.Addr
stack string
tunIf tun.Tun
tunStack tun.Stack
platformInterface adapter.PlatformInterface
platformOptions option.TunPlatformOptions
autoRedirect tun.AutoRedirect
routeRuleSet []adapter.RuleSet
routeExcludeRuleSet []adapter.RuleSet
routeAddressSet []*netipx.IPSet
routeExcludeAddressSet []*netipx.IPSet
}
func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.TunInboundOptions) (adapter.Inbound, error) {
@@ -328,9 +325,12 @@ func (t *Inbound) Tag() string {
return t.tag
}
func (t *Inbound) Start(stage adapter.StartStage) error {
func (t *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateInitialize:
if t.autoRedirect != nil {
scope.Add(t.autoRedirect.Close)
}
if t.tunOptions.DNSModeOrDefault() != tun.DNSModeDisabled && len(t.tunOptions.DNSAddress) == 0 {
inet4DNSAddress, _ := t.tunOptions.Inet4DNSAddress()
inet6DNSAddress, _ := t.tunOptions.Inet6DNSAddress()
@@ -383,7 +383,11 @@ func (t *Inbound) Start(stage adapter.StartStage) error {
routeRuleSet.IncRef()
t.routeAddressSet = append(t.routeAddressSet, ipSets...)
if t.autoRedirect != nil {
t.routeRuleSetCallback = append(t.routeRuleSetCallback, routeRuleSet.RegisterCallback(t.updateRouteAddressSet))
callback := routeRuleSet.RegisterCallback(t.updateRouteAddressSet)
scope.Add(func() error {
routeRuleSet.UnregisterCallback(callback)
return nil
})
}
}
for _, routeExcludeRuleSet := range t.routeExcludeRuleSet {
@@ -394,7 +398,11 @@ func (t *Inbound) Start(stage adapter.StartStage) error {
routeExcludeRuleSet.IncRef()
t.routeExcludeAddressSet = append(t.routeExcludeAddressSet, ipSets...)
if t.autoRedirect != nil {
t.routeExcludeRuleSetCallback = append(t.routeExcludeRuleSetCallback, routeExcludeRuleSet.RegisterCallback(t.updateRouteAddressSet))
callback := routeExcludeRuleSet.RegisterCallback(t.updateRouteAddressSet)
scope.Add(func() error {
routeExcludeRuleSet.UnregisterCallback(callback)
return nil
})
}
}
}
@@ -435,14 +443,13 @@ func (t *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return E.Cause(err, "configure tun interface")
}
scope.Add(tunInterface.Close)
t.logger.Trace("creating stack")
t.tunIf = tunInterface
if t.platformInterface != nil {
err = t.platformInterface.ProcessPlatformOptions(t.platformOptions)
if err != nil {
closeError := t.tunIf.Close()
t.tunIf = nil
return E.Errors(E.Cause(err, "process platform options"), closeError)
return E.Cause(err, "process platform options")
}
}
var includeAllNetworks bool
@@ -467,6 +474,7 @@ func (t *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
scope.Add(tunStack.Close)
t.tunStack = tunStack
t.logger.Info("started at ", t.tunOptions.Name)
case adapter.StartStatePostStart:
@@ -512,14 +520,6 @@ func (t *Inbound) InterfaceUpdated(ctx context.Context) {
}
}
func (t *Inbound) Close() error {
return common.Close(
t.tunStack,
t.tunIf,
t.autoRedirect,
)
}
func (t *Inbound) JudgeFlow(network uint8, source netip.AddrPort, destination netip.AddrPort, firstPacket []byte) tun.FlowVerdict {
if slices.Contains(t.dnsHijackAddress, destination.Addr()) {
if network == uint8(header.UDPProtocolNumber) {
+10 -11
View File
@@ -102,7 +102,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -111,10 +111,18 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
scope.Add(h.tlsConfig.Close)
}
if h.transport == nil {
return h.listener.Start()
err := h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
scope.Add(h.transport.Close)
scope.Add(h.listener.Close)
if common.Contains(h.transport.Network(), N.NetworkTCP) {
tcpListener, err := h.listener.ListenTCP()
if err != nil {
@@ -142,15 +150,6 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
return nil
}
func (h *Inbound) Close() error {
return common.Close(
h.service,
h.listener,
h.tlsConfig,
h.transport,
)
}
func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
if h.tlsConfig != nil && h.transport == nil {
tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig)
+11 -2
View File
@@ -170,8 +170,17 @@ func (h *Outbound) CloseIdleConnections() {
}
}
func (h *Outbound) Close() error {
return common.Close(common.PtrOrNil(h.multiplexDialer), h.transport)
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
if h.transport != nil {
scope.Add(h.transport.Close)
}
if h.multiplexDialer != nil {
scope.Add(h.multiplexDialer.Close)
}
return nil
}
type vlessDialer Outbound
+11 -11
View File
@@ -100,7 +100,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
return inbound, nil
}
func (h *Inbound) Start(stage adapter.StartStage) error {
func (h *Inbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateStart {
return nil
}
@@ -108,15 +108,24 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
scope.Add(h.service.Close)
if h.tlsConfig != nil {
err = h.tlsConfig.Start()
if err != nil {
return err
}
scope.Add(h.tlsConfig.Close)
}
if h.transport == nil {
return h.listener.Start()
err = h.listener.Start()
if err != nil {
return err
}
scope.Add(h.listener.Close)
return nil
}
scope.Add(h.transport.Close)
scope.Add(h.listener.Close)
if common.Contains(h.transport.Network(), N.NetworkTCP) {
tcpListener, err := h.listener.ListenTCP()
if err != nil {
@@ -144,15 +153,6 @@ func (h *Inbound) Start(stage adapter.StartStage) error {
return nil
}
func (h *Inbound) Close() error {
return common.Close(
h.service,
h.listener,
h.tlsConfig,
h.transport,
)
}
func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
if h.tlsConfig != nil && h.transport == nil {
tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig)
+11 -2
View File
@@ -148,8 +148,17 @@ func (h *Outbound) CloseIdleConnections() {
}
}
func (h *Outbound) Close() error {
return common.Close(common.PtrOrNil(h.multiplexDialer), h.transport)
func (h *Outbound) Start(stage adapter.StartStage, scope *adapter.Scope) error {
if stage != adapter.StartStateInitialize {
return nil
}
if h.transport != nil {
scope.Add(h.transport.Close)
}
if h.multiplexDialer != nil {
scope.Add(h.multiplexDialer.Close)
}
return nil
}
func (h *Outbound) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) {
+9 -8
View File
@@ -134,8 +134,10 @@ func NewEndpoint(ctx context.Context, router adapter.Router, logger log.ContextL
return ep, nil
}
func (w *Endpoint) Start(stage adapter.StartStage) error {
func (w *Endpoint) Start(stage adapter.StartStage, scope *adapter.Scope) error {
switch stage {
case adapter.StartStateInitialize:
scope.Add(w.endpoint.Close)
case adapter.StartStateStart:
return w.endpoint.Start(false)
case adapter.StartStatePostStart:
@@ -144,17 +146,16 @@ func (w *Endpoint) Start(stage adapter.StartStage) error {
return err
}
w.started.Store(true)
scope.Add(func() error {
w.bindAccess.Lock()
w.started.Store(false)
w.bindAccess.Unlock()
return nil
})
}
return nil
}
func (w *Endpoint) Close() error {
w.bindAccess.Lock()
w.started.Store(false)
w.bindAccess.Unlock()
return w.endpoint.Close()
}
func (w *Endpoint) InterfaceUpdated(ctx context.Context) {
if !w.started.Load() {
return