mtfyun/sing-box
1
0 Fork 0
mirror of https://github.com/SagerNet/sing-box.git synced 2026-11-10 13:14:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix per-destination backpressure to endpoints

Browse Source
This commit is contained in:
世界 committed 2026-11-09 15:17:20 +00:00
1 parent c0d0364af1
commit 1a23343ed1
25 files changed
+564 -354
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

protocol/bridge/backend_darwin.go
+15 -23
View File
@@ -232,11 +232,16 @@ func (b *backendDarwin) WritePackets(packets [][]byte) error {
}
func (b *backendDarwin) batchReadLoop() {
headroom := -1
var buffers [][]byte
var batch [][]byte
var (
batch [][]byte
returnPaths []tun.Return
)
for {
packets, err := b.batchTUN.BatchRead()
headroom := 0
if len(returnPaths) > 0 {
headroom = returnPaths[0].ReturnHeadroom()
}
packets, err := b.batchTUN.BatchRead(headroom, 0)
if err != nil {
select {
case <-b.closed:
@@ -253,34 +258,20 @@ func (b *backendDarwin) batchReadLoop() {
continue
}
b.returnAccess.Lock()
returnPaths := b.returnPaths
returnPaths = b.returnPaths
b.returnAccess.Unlock()
if len(returnPaths) == 0 {
buf.ReleaseMulti(packets)
continue
}
pathHeadroom := returnPaths[0].ReturnHeadroom()
if pathHeadroom != headroom {
headroom = pathHeadroom
buffers = buffers[:0]
}
for len(buffers) < len(packets) {
buffers = append(buffers, make([]byte, headroom+bridgeTunMTU))
}
batch = batch[:0]
for _, packet := range packets {
payload := packet.Bytes()
if len(payload) == 0 {
if packet.IsEmpty() {
continue
}
fixReturnChecksum(payload)
buffer := buffers[len(batch)][:headroom+len(payload)]
copy(buffer[headroom:], payload)
batch = append(batch, buffer)
}
buf.ReleaseMulti(packets)
if len(batch) == 0 {
continue
fixReturnChecksum(packet.Bytes())
packet.ExtendHeader(headroom)
batch = append(batch, packet.Bytes())
}
unconsumed := batch
currentHeadroom := headroom
@@ -302,6 +293,7 @@ func (b *backendDarwin) batchReadLoop() {
}
unconsumed = returnPath.ReturnPackets(unconsumed)
}
buf.ReleaseMulti(packets)
}
}
protocol/masque/client.go
+5 -1
View File
@@ -205,6 +205,10 @@ func (c *ClientEndpoint) WriteInboundBuffers(packetBuffers []*buf.Buffer) error
return err
}
func (c *ClientEndpoint) FrontHeadroom() int {
return c.device.FrontHeadroom()
}
func (c *ClientEndpoint) InterfaceUpdated(ctx context.Context) {
c.client.RestartSession()
}
@@ -272,7 +276,7 @@ func (c *ClientEndpoint) WritePackets(packets [][]byte) error {
return c.client.WritePacketBuffers(common.Map(packets, func(packet []byte) *buf.Buffer {
packetBuffer := buf.NewSize(masque.PacketHeadroom + len(packet))
packetBuffer.Resize(masque.PacketHeadroom, 0)
packetBuffer.Write(packet)
common.Must1(packetBuffer.Write(packet))
return packetBuffer
}), true)
}
protocol/masque/endpoint.go
+14 -14
View File
@@ -37,20 +37,20 @@ func newDeviceOptions(ctx context.Context, logger log.ContextLogger, handler tun
udpTimeout = C.UDPTimeout
}
return &device.Options{
Context: ctx,
Logger: logger,
System: options.System,
Handler: handler,
UDPTimeout: udpTimeout,
ICMPTimeout: C.ICMPTimeout,
UDPMapping: tun.NATMapping(options.UDPMapping),
UDPFiltering: tun.NATFiltering(options.UDPFiltering),
UDPNATMax: options.UDPNATMax,
InterfaceFinder: service.FromContext[adapter.NetworkManager](ctx).InterfaceFinder(),
Name: options.Name,
NamePrefix: "masque",
MTU: options.MTU,
PacketHeadroom: masque.PacketHeadroom,
Context: ctx,
Logger: logger,
System: options.System,
Handler: handler,
UDPTimeout: udpTimeout,
ICMPTimeout: C.ICMPTimeout,
UDPMapping: tun.NATMapping(options.UDPMapping),
UDPFiltering: tun.NATFiltering(options.UDPFiltering),
UDPNATMax: options.UDPNATMax,
InterfaceFinder: service.FromContext[adapter.NetworkManager](ctx).InterfaceFinder(),
Name: options.Name,
NamePrefix: "masque",
MTU: options.MTU,
PacketFrontHeadroom: masque.PacketHeadroom,
Configuration: device.Configuration{
MTU: options.MTU,
Address: address,
protocol/masque/server.go
+15 -1
View File
@@ -135,6 +135,7 @@ func NewServerEndpoint(ctx context.Context, router adapter.Router, logger log.Co
ConnectionHandler: (*serverConnectionHandler)(serverEndpoint),
})
serverEndpoint.deviceOptions = newDeviceOptions(ctx, logger, serverEndpoint, options.MASQUEEndpointOptions, time.Duration(options.UDPTimeout), options.Address)
serverEndpoint.deviceOptions.Route = server.RouteOutbound
return serverEndpoint, nil
}
@@ -209,6 +210,14 @@ func (s *ServerEndpoint) WriteInboundBuffers(packetBuffers []*buf.Buffer) error
return err
}
func (s *ServerEndpoint) FrontHeadroom() int {
return s.device.FrontHeadroom()
}
func (s *ServerEndpoint) NewOutboundQueue(handler func(packetBuffers []*buf.Buffer)) *tun.OutboundQueue {
return s.device.NewOutboundQueue(handler)
}
func (s *ServerEndpoint) PreMatchFlow(network string, destination netip.Addr) adapter.PreMatchAction {
return adapter.PreMatchFlow
}
@@ -241,7 +250,12 @@ func (s *ServerEndpoint) WritePackets(packets [][]byte) error {
if !s.started.Load() {
return E.New("endpoint is not ready yet")
}
return s.server.WritePacketBuffers(common.Map(packets, buf.As), true)
return s.server.WritePacketBuffers(common.Map(packets, func(packet []byte) *buf.Buffer {
packetBuffer := buf.NewSize(masque.PacketHeadroom + len(packet))
packetBuffer.Resize(masque.PacketHeadroom, 0)
common.Must1(packetBuffer.Write(packet))
return packetBuffer
}), true)
}
func (s *ServerEndpoint) writePacketBuffers(packetBuffers []*buf.Buffer) error {
protocol/openconnect/client.go
+15 -14
View File
@@ -161,20 +161,21 @@ func NewEndpoint(ctx context.Context, router adapter.Router, logger log.ContextL
}
networkManager := service.FromContext[adapter.NetworkManager](ctx)
openConnectEndpoint.deviceOptions = &device.Options{
Context: ctx,
Logger: logger,
System: options.System,
Handler: openConnectEndpoint,
UDPTimeout: udpTimeout,
ICMPTimeout: C.ICMPTimeout,
UDPMapping: tun.NATMapping(options.UDPMapping),
UDPFiltering: tun.NATFiltering(options.UDPFiltering),
UDPNATMax: options.UDPNATMax,
InterfaceFinder: networkManager.InterfaceFinder(),
Name: options.Name,
NamePrefix: "oc",
MTU: openconnecttransport.DefaultMTU,
PacketHeadroom: openconnecttransport.PacketHeadroom,
Context: ctx,
Logger: logger,
System: options.System,
Handler: openConnectEndpoint,
UDPTimeout: udpTimeout,
ICMPTimeout: C.ICMPTimeout,
UDPMapping: tun.NATMapping(options.UDPMapping),
UDPFiltering: tun.NATFiltering(options.UDPFiltering),
UDPNATMax: options.UDPNATMax,
InterfaceFinder: networkManager.InterfaceFinder(),
Name: options.Name,
NamePrefix: "oc",
MTU: openconnecttransport.DefaultMTU,
PacketFrontHeadroom: openconnect.PacketHeadroom,
PacketRearHeadroom: openconnect.PacketRearHeadroom,
Configuration: device.Configuration{
MTU: openconnecttransport.DefaultMTU,
},
protocol/openvpn/client.go
+19 -15
View File
@@ -127,26 +127,31 @@ func NewClientEndpoint(ctx context.Context, router adapter.Router, logger log.Co
if deviceMTU == 0 {
deviceMTU = ovpntransport.DefaultMTU
}
packetFrontHeadroom, packetRearHeadroom := clientOptions.DataPacketHeadroom()
clientEndpoint.deviceOptions = &device.Options{
Context: ctx,
Logger: logger,
System: options.System,
Handler: clientEndpoint,
UDPTimeout: udpTimeout,
ICMPTimeout: C.ICMPTimeout,
UDPMapping: tun.NATMapping(options.UDPMapping),
UDPFiltering: tun.NATFiltering(options.UDPFiltering),
UDPNATMax: options.UDPNATMax,
InterfaceFinder: service.FromContext[adapter.NetworkManager](ctx).InterfaceFinder(),
Name: options.Name,
NamePrefix: "ovpn",
MTU: deviceMTU,
PacketHeadroom: ovpntransport.PacketHeadroom,
Context: ctx,
Logger: logger,
System: options.System,
Handler: clientEndpoint,
UDPTimeout: udpTimeout,
ICMPTimeout: C.ICMPTimeout,
UDPMapping: tun.NATMapping(options.UDPMapping),
UDPFiltering: tun.NATFiltering(options.UDPFiltering),
UDPNATMax: options.UDPNATMax,
InterfaceFinder: service.FromContext[adapter.NetworkManager](ctx).InterfaceFinder(),
Name: options.Name,
NamePrefix: "ovpn",
MTU: deviceMTU,
PacketFrontHeadroom: packetFrontHeadroom,
PacketRearHeadroom: packetRearHeadroom,
Configuration: device.Configuration{
MTU: deviceMTU,
Address: clientOptions.Tunnel.LocalAddress,
},
}
clientOptions.IncomingPacketHeadroom = func() int {
return clientEndpoint.device.FrontHeadroom()
}
client, err := ovpn.NewClient(clientOptions)
if err != nil {
return nil, err
@@ -402,7 +407,6 @@ func buildClientDataChannelOptions(options option.OpenVPNClientEndpointOptions)
AllowCompression: options.AllowCompression,
ReplayWindow: options.ReplayWindow,
ReplayWindowTime: time.Duration(options.ReplayWindowTime),
PacketHeadroom: ovpntransport.PacketHeadroom,
}
}
protocol/openvpn/server.go
+40 -30
View File
@@ -103,21 +103,24 @@ func NewServerEndpoint(ctx context.Context, router adapter.Router, logger log.Co
if options.UDPTimeout != 0 {
udpTimeout = time.Duration(options.UDPTimeout)
}
packetFrontHeadroom, packetRearHeadroom := serverOptions.DataPacketHeadroom()
serverEndpoint.deviceOptions = &device.Options{
Context: ctx,
Logger: logger,
System: options.System,
Handler: serverEndpoint,
UDPTimeout: udpTimeout,
ICMPTimeout: C.ICMPTimeout,
UDPMapping: tun.NATMapping(options.UDPMapping),
UDPFiltering: tun.NATFiltering(options.UDPFiltering),
UDPNATMax: options.UDPNATMax,
InterfaceFinder: service.FromContext[adapter.NetworkManager](ctx).InterfaceFinder(),
Name: options.Name,
NamePrefix: "ovpn",
MTU: options.MTU,
PacketHeadroom: ovpntransport.PacketHeadroom,
Context: ctx,
Logger: logger,
System: options.System,
Handler: serverEndpoint,
UDPTimeout: udpTimeout,
ICMPTimeout: C.ICMPTimeout,
UDPMapping: tun.NATMapping(options.UDPMapping),
UDPFiltering: tun.NATFiltering(options.UDPFiltering),
UDPNATMax: options.UDPNATMax,
InterfaceFinder: service.FromContext[adapter.NetworkManager](ctx).InterfaceFinder(),
Name: options.Name,
NamePrefix: "ovpn",
MTU: options.MTU,
PacketFrontHeadroom: packetFrontHeadroom,
PacketRearHeadroom: packetRearHeadroom,
Route: serverEndpoint.routeOutbound,
Configuration: device.Configuration{
MTU: options.MTU,
Address: options.Address,
@@ -245,6 +248,10 @@ func (s *ServerEndpoint) Start(stage adapter.StartStage, scope *adapter.Scope) e
}
serverOptions.Transport.Listener = streamListener
serverOptions.Transport.PacketConn = packetConn
serverOptions.NewOutboundQueue = func(write func(buffers []*buf.Buffer)) ovpn.OutboundQueue {
return s.device.NewOutboundQueue(write)
}
serverOptions.IncomingPacketHeadroom = s.device.FrontHeadroom
server, err := ovpn.NewServer(serverOptions)
if err != nil {
if packetConn != nil {
@@ -340,7 +347,6 @@ func buildServerOptions(options option.OpenVPNServerEndpointOptions) (ovpn.Serve
Auth: options.Auth,
ReplayWindow: options.ReplayWindow,
ReplayWindowTime: time.Duration(options.ReplayWindowTime),
PacketHeadroom: ovpntransport.PacketHeadroom,
},
TLS: tlsOptions,
Timing: ovpn.ServerTimingOptions{
@@ -441,7 +447,6 @@ func buildStaticKeyServerOptions(options option.OpenVPNServerEndpointOptions, pr
Auth: options.Auth,
ReplayWindow: options.ReplayWindow,
ReplayWindowTime: time.Duration(options.ReplayWindowTime),
PacketHeadroom: ovpntransport.PacketHeadroom,
},
Timing: ovpn.ServerTimingOptions{
PingInterval: time.Duration(options.PingInterval),
@@ -670,17 +675,18 @@ func (s *ServerEndpoint) WritePackets(packets [][]byte) error {
if !s.started.Load() {
return E.New("endpoint is not ready yet")
}
packetBuffers := make([]*buf.Buffer, len(packets))
for i, packet := range packets {
packetBuffers[i] = buf.As(packet)
}
routeMisses, err := s.server.WriteDataPacketBuffersByDestination(packetBuffers)
routeMisses, err := s.server.WriteDataPacketsByDestination(packets)
if len(routeMisses) > 0 {
s.writeRouteMisses(routeMisses)
}
return err
}
func (s *ServerEndpoint) routeOutbound(packet []byte) *tun.OutboundQueue {
outboundQueue, _ := s.server.RouteOutbound(packet).(*tun.OutboundQueue)
return outboundQueue
}
func (s *ServerEndpoint) writePacketBuffersByDestination(packetBuffers []*buf.Buffer) error {
routeMisses, err := s.server.WriteDataPacketBuffersByDestination(packetBuffers)
if len(routeMisses) > 0 {
@@ -690,21 +696,25 @@ func (s *ServerEndpoint) writePacketBuffersByDestination(packetBuffers []*buf.Bu
}
func (s *ServerEndpoint) writeRouteMisses(routeMisses []*ovpn.RouteMissError) {
returnPath, headroom := s.device.ReturnPath()
if returnPath == nil {
return
}
headroom := s.device.FrontHeadroom()
inet4Address, inet6Address := s.PortAddresses()
replies := make([][]byte, 0, len(routeMisses))
replies := make([]*buf.Buffer, 0, len(routeMisses))
for _, routeMiss := range routeMisses {
sourceAddress := packetSourceAddress(routeMiss.Packet, inet4Address, inet6Address)
reply, built := tun.BuildICMPError(routeMiss.Packet, tun.ICMPErrorNoRoute, sourceAddress, 0, headroom)
if built {
replies = append(replies, reply)
if !built {
continue
}
replyBuffer := buf.As(reply)
replyBuffer.Advance(headroom)
replies = append(replies, replyBuffer)
}
if len(replies) > 0 {
returnPath.ReturnPackets(replies)
if len(replies) == 0 {
return
}
err := s.device.WriteInboundBuffers(replies)
if err != nil {
s.logger.Debug(E.Cause(err, "write ICMP error"))
}
}
test/go.mod
+9 -9
View File
@@ -7,18 +7,19 @@ require github.com/sagernet/sing-box v0.0.0
replace github.com/sagernet/sing-box => ../
require (
github.com/coder/websocket v1.8.14
github.com/docker/docker v27.3.1+incompatible
github.com/docker/go-connections v0.5.0
github.com/gofrs/uuid/v5 v5.5.1
github.com/opencontainers/image-spec v1.1.0
github.com/sagernet/quic-go v0.61.0-sing-box-mod.7
github.com/sagernet/sing v0.9.6-0.20260920093640-814d533a21ef
github.com/sagernet/sing-openvpn v0.0.0-20260903200517-e060dda5b1f1
github.com/sagernet/sing-quic v0.7.1-0.20260904135313-497364e8ee3e
github.com/sagernet/sing v0.9.6-0.20260922013359-4ca3bebe0b8e
github.com/sagernet/sing-openvpn v0.0.0-20260925112415-fe3a4fdc2e64
github.com/sagernet/sing-quic v0.7.1-0.20260927144857-8601a428f4db
github.com/sagernet/sing-shadowsocks v0.2.8
github.com/sagernet/sing-shadowsocks2 v0.2.1
github.com/sagernet/sing-tun v0.9.6-0.20260919141142-a39eab51450b
github.com/sagernet/tailscale v1.102.1-sing-box-1.14-mod.5.0.20260916060041-a8fbeb4b0838
github.com/sagernet/sing-tun v0.9.6-0.20260925112405-97d11460f2ea
github.com/sagernet/tailscale v1.102.1-sing-box-1.14-mod.5.0.20260925112514-35e61219dedd
github.com/spyzhov/ajson v0.9.4
github.com/stretchr/testify v1.12.0
go.uber.org/goleak v1.3.0
@@ -42,7 +43,6 @@ require (
github.com/caddyserver/certmagic v0.25.3-0.20260421143802-60d9d8b415d6 // indirect
github.com/caddyserver/zerossl v0.1.5 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/coder/websocket v1.8.14 // indirect
github.com/containerd/log v0.1.0 // indirect
github.com/coreos/go-iptables v0.7.1-0.20240112124308-65c67c9f46e6 // indirect
github.com/coreos/go-oidc/v3 v3.17.0 // indirect
@@ -155,16 +155,16 @@ require (
github.com/sagernet/gvisor v0.0.0-20250822052253-5558536cf237 // indirect
github.com/sagernet/netlink v0.0.0-20260814022025-64455d367bbf // indirect
github.com/sagernet/nftables v0.3.0-mod.4 // indirect
github.com/sagernet/sing-anytls v0.0.0-20260904135308-cec2d74334be // indirect
github.com/sagernet/sing-anytls v0.0.0-20260924021732-7ca72921ac6a // indirect
github.com/sagernet/sing-cloudflared v0.1.3-0.20260706062323-d9787e794aa3 // indirect
github.com/sagernet/sing-mux v0.3.9-0.20260927144857-0257b84c582a // indirect
github.com/sagernet/sing-openconnect v0.0.0-20260903200519-8b89c968949d // indirect
github.com/sagernet/sing-openconnect v0.0.0-20260925112412-098ce1337fbe // indirect
github.com/sagernet/sing-shadowtls v0.2.1 // indirect
github.com/sagernet/sing-snell v0.0.0-20260904135315-bc5a12ac736f // indirect
github.com/sagernet/sing-usbip v0.0.0-20260817040617-28bd42667eca // indirect
github.com/sagernet/sing-vmess v0.2.8 // indirect
github.com/sagernet/smux v1.5.50-sing-box-mod.1 // indirect
github.com/sagernet/wireguard-go v0.0.8-0.20260917055437-bc6039037038 // indirect
github.com/sagernet/wireguard-go v0.0.8-0.20260925112423-da3fb928cdc1 // indirect
github.com/sagernet/ws v0.0.0-20231204124109-acfe8907c854 // indirect
github.com/smallstep/pkcs7 v0.1.1 // indirect
github.com/tailscale/certstore v0.1.1-0.20260409135935-3638fb84b77d // indirect
test/go.sum
+14 -14
View File
@@ -331,20 +331,20 @@ github.com/sagernet/nftables v0.3.0-mod.4 h1:vnOtcDYeSXv2e5RoRuGH0lrpttQFJ8iC4IC
github.com/sagernet/nftables v0.3.0-mod.4/go.mod h1:8kslHG4VvYNihcco+i6uxIX7qbT8A56T0y5q7U44ZaQ=
github.com/sagernet/quic-go v0.61.0-sing-box-mod.7 h1:sW2O+DoNF+my1PGlSZ3zG2N+fdxLR3DC0l4sXCM0lBE=
github.com/sagernet/quic-go v0.61.0-sing-box-mod.7/go.mod h1:hmLC8GJPp+BrpBgrgJnacvI4fZGIhasuy1ROJJ8GR4E=
github.com/sagernet/sing v0.9.6-0.20260920093640-814d533a21ef h1:xutL8qCHi73T2KBXc3AAF97qUFZJRAJAe43yCkqenR0=
github.com/sagernet/sing v0.9.6-0.20260920093640-814d533a21ef/go.mod h1:K3Owt3xPhHugvlnlPPxZJ/exXdaJfEPOTNorGk4AXjo=
github.com/sagernet/sing-anytls v0.0.0-20260904135308-cec2d74334be h1:0uMXu8kcJU0XLc2vfjGH6q6khmECCc5WYpD9dqlu1Xg=
github.com/sagernet/sing-anytls v0.0.0-20260904135308-cec2d74334be/go.mod h1:VRKLnj0ObEsN228h+zuzf91p+6sjtYE8SBaLPVsBy5o=
github.com/sagernet/sing v0.9.6-0.20260922013359-4ca3bebe0b8e h1:QWET7c7thcrm0svrxVmEIaDnYbmfHbE570gvO89RNN4=
github.com/sagernet/sing v0.9.6-0.20260922013359-4ca3bebe0b8e/go.mod h1:K3Owt3xPhHugvlnlPPxZJ/exXdaJfEPOTNorGk4AXjo=
github.com/sagernet/sing-anytls v0.0.0-20260924021732-7ca72921ac6a h1:pBs90r7Ju2qsT+5PwKkZgmxwHVD5AnGlEXVehDXVplE=
github.com/sagernet/sing-anytls v0.0.0-20260924021732-7ca72921ac6a/go.mod h1:VRKLnj0ObEsN228h+zuzf91p+6sjtYE8SBaLPVsBy5o=
github.com/sagernet/sing-cloudflared v0.1.3-0.20260706062323-d9787e794aa3 h1:3y6++yIa8XlDhxPkpR4p+7RUHVY2KTP9CPIGnWmOlO8=
github.com/sagernet/sing-cloudflared v0.1.3-0.20260706062323-d9787e794aa3/go.mod h1:XEqEDYRCAYLaoPjZ1ifVWJg5iWAJHL2gOAXe/PM28Cg=
github.com/sagernet/sing-mux v0.3.9-0.20260927144857-0257b84c582a h1:VrlU4QE8Bh4kEeLgvvWLBtyTiuXqc6/5vf3B+xQCTRE=
github.com/sagernet/sing-mux v0.3.9-0.20260927144857-0257b84c582a/go.mod h1:jricpwCOndDsTEM1TM3piJvhL/2Mm7OhLkm+JadJJ1Q=
github.com/sagernet/sing-openconnect v0.0.0-20260903200519-8b89c968949d h1:Jt/Tb0boIr/xRr4sWr2vC/rrVlYNoumZITLEQpqZ8UE=
github.com/sagernet/sing-openconnect v0.0.0-20260903200519-8b89c968949d/go.mod h1:4AKZLVcvY3r54UaK2Gbnm7aN8pOwdLz+y4EP0QFZ5Eg=
github.com/sagernet/sing-openvpn v0.0.0-20260903200517-e060dda5b1f1 h1:Zn7+pNY5mX1e4DO8oKhTHz5sxqUKNdCWKbfZUtM26Qg=
github.com/sagernet/sing-openvpn v0.0.0-20260903200517-e060dda5b1f1/go.mod h1:PWX7WygD8jpwfqfaGNySXpJYTn0SOwjBI1BKHHC2+Bw=
github.com/sagernet/sing-quic v0.7.1-0.20260904135313-497364e8ee3e h1:2r0FRN0DBSfggWYtdH/69uW91kqEOByBWw/5hiCq6qU=
github.com/sagernet/sing-quic v0.7.1-0.20260904135313-497364e8ee3e/go.mod h1:rJFeVxbEej2iuyeCJ3drKW+p5rOgtLIW3XhZ6ty7Fig=
github.com/sagernet/sing-openconnect v0.0.0-20260925112412-098ce1337fbe h1:Bt33IHF3ucryCV71KQVZq4tsjBUouy+ol8GByZawAm0=
github.com/sagernet/sing-openconnect v0.0.0-20260925112412-098ce1337fbe/go.mod h1:4AKZLVcvY3r54UaK2Gbnm7aN8pOwdLz+y4EP0QFZ5Eg=
github.com/sagernet/sing-openvpn v0.0.0-20260925112415-fe3a4fdc2e64 h1:bhCnU5nB/Ds8l9ChZRnpbs92EYn+B6PaAWZI2D+qrac=
github.com/sagernet/sing-openvpn v0.0.0-20260925112415-fe3a4fdc2e64/go.mod h1:PWX7WygD8jpwfqfaGNySXpJYTn0SOwjBI1BKHHC2+Bw=
github.com/sagernet/sing-quic v0.7.1-0.20260927144857-8601a428f4db h1:BXT2PPAWBakTs+UF6q+2rrkEFu3FZ31xwS9s8k96gK4=
github.com/sagernet/sing-quic v0.7.1-0.20260927144857-8601a428f4db/go.mod h1:rJFeVxbEej2iuyeCJ3drKW+p5rOgtLIW3XhZ6ty7Fig=
github.com/sagernet/sing-shadowsocks v0.2.8 h1:PURj5PRoAkqeHh2ZW205RWzN9E9RtKCVCzByXruQWfE=
github.com/sagernet/sing-shadowsocks v0.2.8/go.mod h1:lo7TWEMDcN5/h5B8S0ew+r78ZODn6SwVaFhvB6H+PTI=
github.com/sagernet/sing-shadowsocks2 v0.2.1 h1:dWV9OXCeFPuYGHb6IRqlSptVnSzOelnqqs2gQ2/Qioo=
@@ -359,10 +359,10 @@ github.com/sagernet/sing-vmess v0.2.8 h1:xd5nnDOMlC76RgrLksS4jlk3eMt3c3CvQY3NsjW
github.com/sagernet/sing-vmess v0.2.8/go.mod h1:P11scgTxMxVVQ8dlM27yNm3Cro40mD0+gHbnqrNGDuY=
github.com/sagernet/smux v1.5.50-sing-box-mod.1 h1:XkJcivBC9V4wBjiGXIXZ229aZCU1hzcbp6kSkkyQ478=
github.com/sagernet/smux v1.5.50-sing-box-mod.1/go.mod h1:NjhsCEWedJm7eFLyhuBgIEzwfhRmytrUoiLluxs5Sk8=
github.com/sagernet/tailscale v1.102.1-sing-box-1.14-mod.5.0.20260916060041-a8fbeb4b0838 h1:YlcCeRWjFIp6dqxvfdH0i6ku17Lv3YzVuHY5qOA7Nv8=
github.com/sagernet/tailscale v1.102.1-sing-box-1.14-mod.5.0.20260916060041-a8fbeb4b0838/go.mod h1:xGiEZax5sZ6gHJcwQxp0GmJVtDHGUXPXTgr6Ksm+yIc=
github.com/sagernet/wireguard-go v0.0.8-0.20260917055437-bc6039037038 h1:1APVQQ0gdlzLXg1fcBVGx+k5xgCiQwB3rvsIKpUT+9A=
github.com/sagernet/wireguard-go v0.0.8-0.20260917055437-bc6039037038/go.mod h1:er10sELpmzLXq7S7Pbc1Zsbyapcr+/gxNAHKTo6fzVA=
github.com/sagernet/tailscale v1.102.1-sing-box-1.14-mod.5.0.20260925112514-35e61219dedd h1:PzEFeBqDaSUdoWE77NptloLTd1sNT1JAiFThgGq5rlo=
github.com/sagernet/tailscale v1.102.1-sing-box-1.14-mod.5.0.20260925112514-35e61219dedd/go.mod h1:Z8Eql5l0q6mQ+etunojfaZxNJVhEkpbKGS/nnx9ZQpE=
github.com/sagernet/wireguard-go v0.0.8-0.20260925112423-da3fb928cdc1 h1:Dsp0Pi6oZk8tNF9jJo2IInDopLx6GHRUBlR6sm7osYs=
github.com/sagernet/wireguard-go v0.0.8-0.20260925112423-da3fb928cdc1/go.mod h1:er10sELpmzLXq7S7Pbc1Zsbyapcr+/gxNAHKTo6fzVA=
github.com/sagernet/ws v0.0.0-20231204124109-acfe8907c854 h1:6uUiZcDRnZSAegryaUGwPC/Fj13JSHwiTftrXhMmYOc=
github.com/sagernet/ws v0.0.0-20231204124109-acfe8907c854/go.mod h1:LtfoSK3+NG57tvnVEHgcuBW9ujgE8enPSgzgwStwCAA=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
transport/device/device.go
+38 -28
View File
@@ -7,6 +7,7 @@ import (
"time"
"github.com/sagernet/sing-tun"
"github.com/sagernet/sing/common"
"github.com/sagernet/sing/common/buf"
"github.com/sagernet/sing/common/control"
E "github.com/sagernet/sing/common/exceptions"
@@ -26,27 +27,30 @@ type Device interface {
PortMTU() uint32
AttachReturn(returnPath tun.Return) error
DetachReturn(returnPath tun.Return) error
ReturnPath() (tun.Return, int)
FrontHeadroom() int
NewOutboundQueue(handler func(packetBuffers []*buf.Buffer)) *tun.OutboundQueue
Close() error
}
type Options struct {
Context context.Context
Logger logger.ContextLogger
System bool
Handler tun.Handler
UDPTimeout time.Duration
ICMPTimeout time.Duration
UDPMapping tun.NATMapping
UDPFiltering tun.NATFiltering
UDPNATMax uint32
InterfaceFinder control.InterfaceFinder
MemoryPressure func() tun.MemoryPressure
Name string
NamePrefix string
MTU uint32
PacketHeadroom int
Configuration Configuration
Context context.Context
Logger logger.ContextLogger
System bool
Handler tun.Handler
UDPTimeout time.Duration
ICMPTimeout time.Duration
UDPMapping tun.NATMapping
UDPFiltering tun.NATFiltering
UDPNATMax uint32
InterfaceFinder control.InterfaceFinder
MemoryPressure func() tun.MemoryPressure
Name string
NamePrefix string
MTU uint32
PacketFrontHeadroom int
PacketRearHeadroom int
Route func(packet []byte) *tun.OutboundQueue
Configuration Configuration
}
type Configuration struct {
@@ -80,9 +84,8 @@ func newStack(options Options, memoryTun *tun.MemoryTun) (*tun.Go, error) {
}
type baseDevice struct {
packetHeadroom int
packetWriter PacketWriter
returnState atomic.Pointer[returnPathState]
packetWriter PacketWriter
returnState atomic.Pointer[returnPathState]
}
func (d *baseDevice) SetPacketWriter(writer PacketWriter) {
@@ -106,7 +109,18 @@ func (d *baseDevice) processInboundBuffers(packetBuffers []*buf.Buffer, writeBuf
return writeBuffers(packetBuffers)
}
packets := make([][]byte, len(packetBuffers))
var temporaryBuffers []*buf.Buffer
defer func() {
buf.ReleaseMulti(temporaryBuffers)
}()
for i, packetBuffer := range packetBuffers {
if packetBuffer.Start() < state.headroom {
temporaryBuffer := buf.NewSize(state.headroom + packetBuffer.Len())
temporaryBuffer.Resize(state.headroom, 0)
common.Must1(temporaryBuffer.Write(packetBuffer.Bytes()))
temporaryBuffers = append(temporaryBuffers, temporaryBuffer)
packetBuffer = temporaryBuffer
}
packetBuffer.ExtendHeader(state.headroom)
packets[i] = packetBuffer.Bytes()
}
@@ -124,13 +138,9 @@ func (d *baseDevice) processInboundBuffers(packetBuffers []*buf.Buffer, writeBuf
}
func (d *baseDevice) AttachReturn(returnPath tun.Return) error {
headroom := returnPath.ReturnHeadroom()
if headroom > d.packetHeadroom {
return E.New("return path headroom ", headroom, " exceeds available ", d.packetHeadroom)
}
newState := &returnPathState{
returnPath: returnPath,
headroom: headroom,
headroom: returnPath.ReturnHeadroom(),
}
for {
currentState := d.returnState.Load()
@@ -154,12 +164,12 @@ func (d *baseDevice) DetachReturn(returnPath tun.Return) error {
return nil
}
func (d *baseDevice) ReturnPath() (tun.Return, int) {
func (d *baseDevice) FrontHeadroom() int {
state := d.returnState.Load()
if state == nil {
return nil, 0
return 0
}
return state.returnPath, state.headroom
return state.headroom
}
type returnPathState struct {
transport/device/stack.go
+15 -5
View File
@@ -29,19 +29,25 @@ type stackDevice struct {
func newStackDevice(options Options) (*stackDevice, error) {
device := &stackDevice{
baseDevice: baseDevice{packetHeadroom: options.PacketHeadroom},
options: options,
options: options,
}
device.inet4Address, device.inet6Address = firstAddresses(options.Configuration.Address)
device.memoryTun = tun.NewMemoryTun(tun.MemoryTunOptions{
MTU: int(options.MTU),
Headroom: options.PacketHeadroom,
RearSpace: systemDevicePacketRearSpace,
Outbound: device.writeOutbound,
Headroom: options.PacketFrontHeadroom,
RearSpace: options.PacketRearHeadroom,
Outbound: func(packetBuffers []*buf.Buffer) {
err := device.writeOutbound(packetBuffers)
if err != nil {
options.Logger.Debug(E.Cause(err, "write packet batch"))
}
},
Route: options.Route,
})
var err error
device.stack, err = newStack(options, device.memoryTun)
if err != nil {
device.memoryTun.Close()
return nil, err
}
return device, nil
@@ -139,6 +145,10 @@ func (d *stackDevice) PortMTU() uint32 {
return d.options.MTU
}
func (d *stackDevice) NewOutboundQueue(handler func(packetBuffers []*buf.Buffer)) *tun.OutboundQueue {
return d.memoryTun.NewOutboundQueue(handler)
}
func (d *stackDevice) Close() error {
var err error
d.closeOnce.Do(func() {
transport/device/system.go
+32 -25
View File
@@ -13,6 +13,7 @@ import (
"github.com/sagernet/sing-box/option"
"github.com/sagernet/sing-tun"
"github.com/sagernet/sing-tun/gtcpip/header"
"github.com/sagernet/sing/common"
"github.com/sagernet/sing/common/buf"
E "github.com/sagernet/sing/common/exceptions"
M "github.com/sagernet/sing/common/metadata"
@@ -20,12 +21,7 @@ import (
"github.com/sagernet/sing/service"
)
var _ Device = (*systemDevice)(nil)
const (
systemDeviceReadBufferSize = 65535 + tun.PacketOffset
systemDevicePacketRearSpace = 64
)
const systemDeviceReadBufferSize = 65535 + tun.PacketOffset
type systemDevice struct {
baseDevice
@@ -52,7 +48,6 @@ func newSystemDevice(options Options) (*systemDevice, error) {
}
inet4Address, inet6Address := firstAddresses(options.Configuration.Address)
return &systemDevice{
baseDevice: baseDevice{packetHeadroom: options.PacketHeadroom},
options: options,
dialer: interfaceDialer,
inet4Address: inet4Address,
@@ -126,12 +121,11 @@ func (d *systemDevice) readLoop(tunInterface tun.Tun, mtu int) {
d.readLoopDarwin(darwinTUN)
return
}
packetBuffer := buf.NewSize(d.packetHeadroom + systemDeviceReadBufferSize + systemDevicePacketRearSpace)
defer packetBuffer.Release()
frontHeadroom := d.options.PacketFrontHeadroom
rearHeadroom := d.options.PacketRearHeadroom
readBuffer := make([]byte, systemDeviceReadBufferSize)
for {
packetBuffer.Reset()
packetBuffer.Resize(d.packetHeadroom, 0)
readN, err := tunInterface.Read(packetBuffer.FreeBytes()[:systemDeviceReadBufferSize])
readN, err := tunInterface.Read(readBuffer)
if err != nil {
if E.IsClosed(err) {
return
@@ -142,14 +136,14 @@ func (d *systemDevice) readLoop(tunInterface tun.Tun, mtu int) {
if readN <= tun.PacketOffset {
continue
}
packetBuffer.Truncate(readN)
packetBuffer.Advance(tun.PacketOffset)
if d.blockIPv6Enabled() && header.IPVersion(packetBuffer.Bytes()) == header.IPv6Version {
packet := readBuffer[tun.PacketOffset:readN]
if d.blockIPv6Enabled() && header.IPVersion(packet) == header.IPv6Version {
continue
}
packetBuffer.IncRef()
packetBuffer := buf.NewSize(frontHeadroom + len(packet) + rearHeadroom)
packetBuffer.Resize(frontHeadroom, 0)
common.Must1(packetBuffer.Write(packet))
err = d.writeOutbound([]*buf.Buffer{packetBuffer})
packetBuffer.DecRef()
if err != nil {
d.options.Logger.Debug(E.Cause(err, "write packet"))
}
@@ -157,39 +151,39 @@ func (d *systemDevice) readLoop(tunInterface tun.Tun, mtu int) {
}
func (d *systemDevice) readLoopLinux(tunInterface tun.LinuxTUN, batchSize int, mtu int) {
frontHeadroom := d.options.PacketFrontHeadroom
packetSize := frontHeadroom + mtu + d.options.PacketRearHeadroom
packetBuffers := make([]*buf.Buffer, batchSize)
readBuffers := make([][]byte, batchSize)
packetSizes := make([]int, batchSize)
outboundBuffers := make([]*buf.Buffer, 0, batchSize)
for i := range packetBuffers {
packetBuffers[i] = buf.NewSize(d.packetHeadroom + mtu + systemDevicePacketRearSpace)
packetBuffers[i] = buf.NewSize(packetSize)
}
defer buf.ReleaseMulti(packetBuffers)
for {
for i, packetBuffer := range packetBuffers {
packetBuffer.Reset()
packetBuffer.Resize(d.packetHeadroom, 0)
packetBuffer.Resize(frontHeadroom, 0)
readBuffers[i] = packetBuffer.FreeBytes()[:mtu]
}
packetCount, readErr := tunInterface.BatchRead(readBuffers, 0, packetSizes)
outboundBuffers = outboundBuffers[:0]
blockIPv6 := d.blockIPv6Enabled()
for i := range packetCount {
packetBuffers[i].Truncate(packetSizes[i])
if blockIPv6 && header.IPVersion(packetBuffers[i].Bytes()) == header.IPv6Version {
continue
}
packetBuffers[i].IncRef()
outboundBuffers = append(outboundBuffers, packetBuffers[i])
packetBuffers[i] = buf.NewSize(packetSize)
}
if len(outboundBuffers) > 0 {
writeErr := d.writeOutbound(outboundBuffers)
for _, packetBuffer := range outboundBuffers {
packetBuffer.DecRef()
}
if writeErr != nil {
d.options.Logger.Debug(E.Cause(writeErr, "write packet batch"))
}
clear(outboundBuffers)
outboundBuffers = outboundBuffers[:0]
}
if readErr != nil {
if E.IsClosed(readErr) {
@@ -202,8 +196,10 @@ func (d *systemDevice) readLoopLinux(tunInterface tun.LinuxTUN, batchSize int, m
}
func (d *systemDevice) readLoopDarwin(tunInterface tun.DarwinTUN) {
frontHeadroom := d.options.PacketFrontHeadroom
rearHeadroom := d.options.PacketRearHeadroom
for {
packetBuffers, readErr := tunInterface.BatchRead()
packetBuffers, readErr := tunInterface.BatchRead(frontHeadroom, rearHeadroom)
outboundBuffers := packetBuffers[:0]
blockIPv6 := d.blockIPv6Enabled()
for _, packetBuffer := range packetBuffers {
@@ -265,6 +261,17 @@ func (d *systemDevice) blockIPv6Enabled() bool {
return d.options.Configuration.BlockIPv6
}
func (d *systemDevice) FrontHeadroom() int {
d.stateAccess.RLock()
tunInterface := d.device
d.stateAccess.RUnlock()
linuxTUN, isLinuxTUN := tunInterface.(tun.LinuxTUN)
if !isLinuxTUN {
return d.baseDevice.FrontHeadroom()
}
return max(d.baseDevice.FrontHeadroom(), linuxTUN.FrontHeadroom())
}
func (d *systemDevice) WriteInboundBuffers(packetBuffers []*buf.Buffer) error {
return d.processInboundBuffers(packetBuffers, d.writeBuffers)
}
transport/device/system_stack.go
+5
View File
@@ -3,6 +3,7 @@ package device
import (
"net/netip"
"github.com/sagernet/sing-tun"
"github.com/sagernet/sing-tun/gtcpip/header"
"github.com/sagernet/sing/common/buf"
E "github.com/sagernet/sing/common/exceptions"
@@ -111,6 +112,10 @@ func packetDestination(packet []byte) netip.Addr {
}
}
func (d *systemStackDevice) NewOutboundQueue(handler func(packetBuffers []*buf.Buffer)) *tun.OutboundQueue {
return d.stackDevice.NewOutboundQueue(handler)
}
func (d *systemStackDevice) Close() error {
return E.Errors(d.stackDevice.Close(), d.systemDevice.Close())
}
transport/http/capsule.go
+31
View File
@@ -12,6 +12,7 @@ import (
"sync"
"time"
"github.com/sagernet/sing/common"
"github.com/sagernet/sing/common/buf"
E "github.com/sagernet/sing/common/exceptions"
F "github.com/sagernet/sing/common/format"
@@ -162,6 +163,36 @@ func WriteDatagramCapsule(writer io.Writer, datagram *buf.Buffer) error {
return err
}
func WriteDatagramCapsules(writer io.Writer, datagrams []*buf.Buffer) error {
if len(datagrams) == 1 {
return WriteDatagramCapsule(writer, datagrams[0])
}
defer buf.ReleaseMulti(datagrams)
var totalLength int
for _, datagram := range datagrams {
totalLength += 1 + VarintLen(uint64(datagram.Len())) + datagram.Len()
}
capsules := buf.NewSize(min(totalLength, buf.MaxPooledBufferSize))
defer capsules.Release()
for _, datagram := range datagrams {
length := uint64(datagram.Len())
headerLength := 1 + VarintLen(length)
if headerLength+datagram.Len() > capsules.FreeLen() {
_, err := writer.Write(capsules.Bytes())
if err != nil {
return err
}
capsules.Reset()
}
header := capsules.Extend(headerLength)
header[0] = CapsuleTypeDatagram
PutVarint(header[1:], length)
common.Must1(capsules.Write(datagram.Bytes()))
}
_, err := writer.Write(capsules.Bytes())
return err
}
type capsuleConn struct {
reader *std_bufio.Reader
writer io.Writer
transport/masque/client.go
+12 -11
View File
@@ -29,6 +29,7 @@ type Configuration struct {
type ClientHandler interface {
UpdateConfiguration(configuration Configuration) error
WriteInboundBuffers(packetBuffers []*buf.Buffer) error
FrontHeadroom() int
}
type ClientOptions struct {
@@ -178,7 +179,7 @@ func (c *Client) connect() (bool, error) {
}
return false, nil
}
current.session = newSession(c.ctx, stream, current, false)
current.session = newSession(c.ctx, stream, current, c.handler.FrontHeadroom)
c.current = current
c.access.Unlock()
err = current.writeCapsule(newAddressCapsule(capsuleTypeAddressRequest, []AssignedAddress{
@@ -299,7 +300,8 @@ func (c *Client) WritePacketBuffers(packetBuffers []*buf.Buffer, forwarded bool)
}
inet4Address, inet6Address := firstAddresses(configuration.Address)
var replies []*buf.Buffer
for i, packetBuffer := range packetBuffers {
routedBuffers := packetBuffers[:0]
for _, packetBuffer := range packetBuffers {
_, destination, protocol, valid := packetAddresses(packetBuffer.Bytes())
if !valid {
packetBuffer.Release()
@@ -312,19 +314,18 @@ func (c *Client) WritePacketBuffers(packetBuffers []*buf.Buffer, forwarded bool)
errorType = tun.ICMPErrorHopLimitExceeded
}
if !routed {
reply, built := buildICMPError(packetBuffer.Bytes(), errorType, inet4Address, inet6Address, 0, PacketHeadroom)
reply, built := buildICMPError(packetBuffer.Bytes(), errorType, inet4Address, inet6Address, 0, c.handler.FrontHeadroom())
if built {
replies = append(replies, reply)
}
packetBuffer.Release()
continue
}
err := current.writePacket(packetBuffer)
if err != nil {
current.cancel(err)
buf.ReleaseMulti(packetBuffers[i+1:])
break
}
routedBuffers = append(routedBuffers, packetBuffer)
}
err := current.writePackets(routedBuffers)
if err != nil {
current.cancel(err)
}
if len(replies) > 0 {
return c.handler.WriteInboundBuffers(replies)
@@ -421,7 +422,7 @@ func (s *clientSession) handlePacket(buffer *buf.Buffer) {
reply, built := buildICMPError(buffer.Bytes(), tun.ICMPErrorNoRoute, inet4Address, inet6Address, 0, transportHTTP.CapsuleHeadroom)
buffer.Release()
if built {
_ = s.writePacket(reply)
_ = s.writePackets([]*buf.Buffer{reply})
}
return
}
@@ -435,7 +436,7 @@ func (s *clientSession) handlePacketTooBig(buffer *buf.Buffer, mtu int) {
s.access.Lock()
inet4Address, inet6Address := firstAddresses(s.configuration.Address)
s.access.Unlock()
reply, built := buildICMPError(buffer.Bytes(), tun.ICMPErrorPacketTooBig, inet4Address, inet6Address, mtu, PacketHeadroom)
reply, built := buildICMPError(buffer.Bytes(), tun.ICMPErrorPacketTooBig, inet4Address, inet6Address, mtu, s.client.handler.FrontHeadroom())
buffer.Release()
if !built {
return
transport/masque/server.go
+62 -18
View File
@@ -22,6 +22,8 @@ import (
type ServerHandler interface {
WriteInboundBuffers(packetBuffers []*buf.Buffer) error
FrontHeadroom() int
NewOutboundQueue(handler func(packetBuffers []*buf.Buffer)) *tun.OutboundQueue
}
type ServerOptions struct {
@@ -51,6 +53,7 @@ type Server struct {
type serverSession struct {
*session
queue *tun.OutboundQueue
server *Server
ctx context.Context
user string
@@ -190,7 +193,13 @@ func (s *Server) NewTunnelRequest(ctx context.Context, request transportHTTP.Tun
s.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", request.Source()))
return
}
current.session = newSession(s.ctx, stream, current, true)
current.session = newSession(s.ctx, stream, current, s.receivedPacketHeadroom)
current.queue = s.handler.NewOutboundQueue(func(packetBuffers []*buf.Buffer) {
err := current.writePackets(packetBuffers)
if err != nil {
current.cancel(err)
}
})
s.access.Lock()
for _, address := range current.addresses {
s.addresses[address] = current
@@ -228,6 +237,9 @@ func (s *Server) releaseSession(current *serverSession) {
return it == current
})
s.access.Unlock()
if current.queue != nil {
current.queue.Close()
}
for _, address := range current.addresses {
for _, pool := range s.pools {
if pool.prefix.Contains(address) {
@@ -255,6 +267,26 @@ func (s *Server) lookup(destination netip.Addr, protocol uint8) *serverSession {
return nil
}
func (s *Server) route(source netip.Addr, destination netip.Addr, protocol uint8) *serverSession {
target := s.lookup(destination, protocol)
if target == nil || !target.accepts(source, protocol) {
return nil
}
return target
}
func (s *Server) RouteOutbound(packet []byte) *tun.OutboundQueue {
source, destination, protocol, valid := packetAddresses(packet)
if !valid {
return nil
}
target := s.route(source, destination, protocol)
if target == nil {
return nil
}
return target.queue
}
func (s *Server) Contains(address netip.Addr) bool {
s.access.RLock()
defer s.access.RUnlock()
@@ -265,7 +297,11 @@ func (s *Server) Contains(address netip.Addr) bool {
}
func (s *Server) WritePacketBuffers(packetBuffers []*buf.Buffer, forwarded bool) error {
var replies []*buf.Buffer
var (
replies []*buf.Buffer
currentTarget *serverSession
)
currentBatch := make([]*buf.Buffer, 0, len(packetBuffers))
for _, packetBuffer := range packetBuffers {
source, destination, protocol, valid := packetAddresses(packetBuffer.Bytes())
if !valid {
@@ -273,24 +309,28 @@ func (s *Server) WritePacketBuffers(packetBuffers []*buf.Buffer, forwarded bool)
continue
}
errorType := tun.ICMPErrorNoRoute
target := s.lookup(destination, protocol)
routed := target != nil && target.accepts(source, protocol)
if routed && forwarded && !decrementHopLimit(packetBuffer.Bytes()) {
routed = false
target := s.route(source, destination, protocol)
if target != nil && forwarded && !decrementHopLimit(packetBuffer.Bytes()) {
target = nil
errorType = tun.ICMPErrorHopLimitExceeded
}
if routed {
buffer := buf.NewSize(transportHTTP.CapsuleHeadroom + packetBuffer.Len())
buffer.Resize(transportHTTP.CapsuleHeadroom, 0)
buffer.Write(packetBuffer.Bytes())
target.queuePacket(buffer)
} else {
reply, built := buildICMPError(packetBuffer.Bytes(), errorType, s.inet4Address, s.inet6Address, 0, PacketHeadroom)
if target == nil {
reply, built := buildICMPError(packetBuffer.Bytes(), errorType, s.inet4Address, s.inet6Address, 0, s.handler.FrontHeadroom())
if built {
replies = append(replies, reply)
}
packetBuffer.Release()
continue
}
packetBuffer.Release()
if len(currentBatch) > 0 && target != currentTarget {
currentTarget.queue.WriteBuffers(currentBatch)
currentBatch = currentBatch[:0]
}
currentTarget = target
currentBatch = append(currentBatch, packetBuffer)
}
if len(currentBatch) > 0 {
currentTarget.queue.WriteBuffers(currentBatch)
}
if len(replies) > 0 {
return s.handler.WriteInboundBuffers(replies)
@@ -298,6 +338,10 @@ func (s *Server) WritePacketBuffers(packetBuffers []*buf.Buffer, forwarded bool)
return nil
}
func (s *Server) receivedPacketHeadroom() int {
return max(PacketHeadroom, s.handler.FrontHeadroom())
}
func (s *Server) Close() error {
s.access.RLock()
sessions := make([]*serverSession, 0, len(s.addresses))
@@ -403,18 +447,18 @@ func (s *serverSession) handlePacket(buffer *buf.Buffer) {
errorType = tun.ICMPErrorHopLimitExceeded
break
}
target.queuePacket(buffer)
target.queue.WriteBuffers([]*buf.Buffer{buffer})
return
}
reply, built := buildICMPError(buffer.Bytes(), errorType, s.server.inet4Address, s.server.inet6Address, 0, transportHTTP.CapsuleHeadroom)
buffer.Release()
if built {
s.queuePacket(reply)
s.queue.WriteBuffers([]*buf.Buffer{reply})
}
}
func (s *serverSession) handlePacketTooBig(buffer *buf.Buffer, mtu int) {
reply, built := buildICMPError(buffer.Bytes(), tun.ICMPErrorPacketTooBig, s.server.inet4Address, s.server.inet6Address, mtu, PacketHeadroom)
reply, built := buildICMPError(buffer.Bytes(), tun.ICMPErrorPacketTooBig, s.server.inet4Address, s.server.inet6Address, mtu, s.server.receivedPacketHeadroom())
buffer.Release()
if !built {
return
@@ -422,7 +466,7 @@ func (s *serverSession) handlePacketTooBig(buffer *buf.Buffer, mtu int) {
_, destination, protocol, _ := packetAddresses(reply.Bytes())
target := s.server.lookup(destination, protocol)
if target != nil {
target.queuePacket(reply)
target.queue.WriteBuffers([]*buf.Buffer{reply})
return
}
err := s.server.handler.WriteInboundBuffers([]*buf.Buffer{reply})
transport/masque/session.go
+53 -76
View File
@@ -8,6 +8,7 @@ import (
"sync"
transportHTTP "github.com/sagernet/sing-box/transport/http"
"github.com/sagernet/sing/common"
"github.com/sagernet/sing/common/buf"
E "github.com/sagernet/sing/common/exceptions"
)
@@ -15,11 +16,10 @@ import (
const (
upgradeToken = "connect-ip"
DefaultMTU = 1280
PacketHeadroom = 64
PacketHeadroom = transportHTTP.CapsuleHeadroom
QUICPacketOverhead = 51
minimumLinkMTU = 1280
maxPacketSize = 65535
sendQueueSize = 256
)
type sessionHandler interface {
@@ -31,29 +31,27 @@ type sessionHandler interface {
}
type session struct {
ctx context.Context
cancel context.CancelCauseFunc
stream io.ReadWriteCloser
datagrams transportHTTP.DatagramStream
reader *std_bufio.Reader
handler sessionHandler
sendQueue chan *buf.Buffer
writeAccess sync.Mutex
ctx context.Context
cancel context.CancelCauseFunc
stream io.ReadWriteCloser
datagrams transportHTTP.DatagramStream
reader *std_bufio.Reader
handler sessionHandler
packetHeadroom func() int
writeAccess sync.Mutex
}
func newSession(ctx context.Context, stream io.ReadWriteCloser, handler sessionHandler, queued bool) *session {
func newSession(ctx context.Context, stream io.ReadWriteCloser, handler sessionHandler, packetHeadroom func() int) *session {
sessionCtx, cancel := context.WithCancelCause(ctx)
datagrams, _ := stream.(transportHTTP.DatagramStream)
current := &session{
ctx: sessionCtx,
cancel: cancel,
stream: stream,
datagrams: datagrams,
reader: std_bufio.NewReader(stream),
handler: handler,
}
if queued {
current.sendQueue = make(chan *buf.Buffer, sendQueueSize)
ctx: sessionCtx,
cancel: cancel,
stream: stream,
datagrams: datagrams,
reader: std_bufio.NewReader(stream),
handler: handler,
packetHeadroom: packetHeadroom,
}
return current
}
@@ -67,9 +65,6 @@ func (s *session) run() error {
if s.datagrams != nil {
loops.Go(s.loopDatagram)
}
if s.sendQueue != nil {
loops.Go(s.loopSend)
}
err := s.loopCapsule()
s.cancel(err)
s.stream.Close()
@@ -88,34 +83,14 @@ func (s *session) loopDatagram() {
if !valid || contextID != 0 || len(datagram) == contextLength {
continue
}
buffer := buf.NewSize(PacketHeadroom + len(datagram) - contextLength)
buffer.Resize(PacketHeadroom, 0)
buffer.Write(datagram[contextLength:])
headroom := s.packetHeadroom()
buffer := buf.NewSize(headroom + len(datagram) - contextLength)
buffer.Resize(headroom, 0)
common.Must1(buffer.Write(datagram[contextLength:]))
s.handler.handlePacket(buffer)
}
}
func (s *session) loopSend() {
for {
select {
case buffer := <-s.sendQueue:
err := s.writePacket(buffer)
if err != nil {
s.cancel(err)
}
case <-s.ctx.Done():
for {
select {
case buffer := <-s.sendQueue:
buffer.Release()
default:
return
}
}
}
}
}
func (s *session) loopCapsule() error {
for {
capsuleType, _, err := transportHTTP.ReadVarint(s.reader)
@@ -156,8 +131,9 @@ func (s *session) readDatagramCapsule(length int) error {
_, err = s.reader.Discard(payloadLength)
return err
}
buffer := buf.NewSize(PacketHeadroom + payloadLength)
buffer.Resize(PacketHeadroom, 0)
headroom := s.packetHeadroom()
buffer := buf.NewSize(headroom + payloadLength)
buffer.Resize(headroom, 0)
_, err = buffer.ReadFullFrom(s.reader, payloadLength)
if err != nil {
buffer.Release()
@@ -211,39 +187,40 @@ func (s *session) writeCapsule(capsule *buf.Buffer) error {
return err
}
func (s *session) queuePacket(buffer *buf.Buffer) {
select {
case s.sendQueue <- buffer:
default:
buffer.Release()
}
}
func (s *session) writePacket(buffer *buf.Buffer) error {
datagram := transportHTTP.PrependContextID(buffer)
if s.datagrams != nil {
err := s.datagrams.SendDatagram(datagram.Bytes())
var tooLarge *transportHTTP.DatagramTooLargeError
switch {
case err == nil:
datagram.Release()
return nil
case errors.As(err, &tooLarge):
mtu := tooLarge.MaxPayloadSize - 1
if mtu < minimumLinkMTU {
func (s *session) writePackets(buffers []*buf.Buffer) error {
capsules := buffers[:0]
for i, buffer := range buffers {
datagram := transportHTTP.PrependContextID(buffer)
if s.datagrams != nil {
err := s.datagrams.SendDatagram(datagram.Bytes())
var tooLarge *transportHTTP.DatagramTooLargeError
switch {
case err == nil:
datagram.Release()
return E.New("QUIC connection is unable to carry ", minimumLinkMTU, " bytes packets")
continue
case errors.As(err, &tooLarge):
mtu := tooLarge.MaxPayloadSize - 1
if mtu >= minimumLinkMTU {
datagram.Advance(1)
s.handler.handlePacketTooBig(datagram, mtu)
continue
}
err = E.New("QUIC connection is unable to carry ", minimumLinkMTU, " bytes packets")
case errors.Is(err, transportHTTP.ErrDatagramUnsupported):
capsules = append(capsules, datagram)
continue
}
datagram.Advance(1)
s.handler.handlePacketTooBig(datagram, mtu)
return nil
case errors.Is(err, transportHTTP.ErrDatagramUnsupported):
default:
datagram.Release()
buf.ReleaseMulti(capsules)
buf.ReleaseMulti(buffers[i+1:])
return err
}
capsules = append(capsules, datagram)
}
if len(capsules) == 0 {
return nil
}
s.writeAccess.Lock()
defer s.writeAccess.Unlock()
return transportHTTP.WriteDatagramCapsule(s.stream, datagram)
return transportHTTP.WriteDatagramCapsules(s.stream, capsules)
}
transport/openconnect/configuration.go
+1 -6
View File
@@ -3,14 +3,9 @@ package openconnect
import (
"net/netip"
"time"
"github.com/sagernet/sing-openconnect"
)
const (
DefaultMTU = 1500
PacketHeadroom = openconnect.PacketHeadroom
)
const DefaultMTU = 1500
type Configuration struct {
MTU uint32
transport/openvpn/configuration.go
+1 -4
View File
@@ -2,10 +2,7 @@ package openvpn
import "net/netip"
const (
DefaultMTU = 1500
PacketHeadroom = 4096
)
const DefaultMTU = 1500
type Configuration struct {
MTU uint32
transport/wireguard/device.go
+1 -1
View File
@@ -18,7 +18,7 @@ type Device interface {
wgTun.Device
N.Dialer
Start() error
SetDevice(device *device.Device)
SetDevice(device *device.Device, peers []*device.Peer)
Inet4Address() netip.Addr
Inet6Address() netip.Addr
}
transport/wireguard/device_peer_router.go
+122
View File
@@ -0,0 +1,122 @@
package wireguard
import (
"net/netip"
"sync/atomic"
"github.com/sagernet/sing-tun"
"github.com/sagernet/sing-tun/gtcpip/header"
"github.com/sagernet/sing/common/buf"
"github.com/sagernet/wireguard-go/device"
)
type peerRouter struct {
memoryTun *tun.MemoryTun
inet4Address netip.Addr
inet6Address netip.Addr
routes atomic.Pointer[peerRoutes]
}
type peerRoutes struct {
allowedIPs *device.AllowedIPs
queues map[*device.Peer]*peerQueue
}
type peerQueue struct {
peer *device.Peer
queue *tun.OutboundQueue
packets [][]byte
}
func newPeerRouter(options DeviceOptions) *peerRouter {
router := &peerRouter{}
router.inet4Address, router.inet6Address = deviceAddresses(options.Address)
router.memoryTun = tun.NewMemoryTun(tun.MemoryTunOptions{
MTU: int(options.MTU),
Outbound: router.writeUnrouted,
Route: router.route,
})
return router
}
func (r *peerRouter) setPeers(allowedIPs *device.AllowedIPs, peers []*device.Peer) {
routes := &peerRoutes{
allowedIPs: allowedIPs,
queues: make(map[*device.Peer]*peerQueue, len(peers)),
}
for _, peer := range peers {
current := &peerQueue{peer: peer}
current.queue = r.memoryTun.NewOutboundQueue(current.write)
routes.queues[peer] = current
}
previous := r.routes.Swap(routes)
if previous != nil {
previous.close()
}
}
func (r *peerRouter) route(packet []byte) *tun.OutboundQueue {
routes := r.routes.Load()
if routes == nil {
return nil
}
var source, destination netip.Addr
switch header.IPVersion(packet) {
case header.IPv4Version:
ipv4Header := header.IPv4(packet)
source, destination = ipv4Header.SourceAddr(), ipv4Header.DestinationAddr()
case header.IPv6Version:
ipv6Header := header.IPv6(packet)
source, destination = ipv6Header.SourceAddr(), ipv6Header.DestinationAddr()
default:
return nil
}
current := routes.queues[routes.allowedIPs.LookupFromPacket(source, destination, packet)]
if current == nil {
return nil
}
return current.queue
}
func (r *peerRouter) writeUnrouted(packetBuffers []*buf.Buffer) {
defer buf.ReleaseMulti(packetBuffers)
replies := make([][]byte, 0, len(packetBuffers))
for _, packetBuffer := range packetBuffers {
packet := packetBuffer.Bytes()
source := r.inet4Address
if header.IPVersion(packet) == header.IPv6Version {
source = r.inet6Address
}
reply, built := tun.BuildICMPError(packet, tun.ICMPErrorNoRoute, source, 0, 0)
if built {
replies = append(replies, reply)
}
}
if len(replies) > 0 {
r.memoryTun.WritePackets(replies)
}
}
func (r *peerRouter) close() error {
routes := r.routes.Swap(nil)
if routes != nil {
routes.close()
}
return r.memoryTun.Close()
}
func (r *peerRoutes) close() {
for _, current := range r.queues {
current.queue.Close()
}
}
func (q *peerQueue) write(packetBuffers []*buf.Buffer) {
for _, packetBuffer := range packetBuffers {
q.packets = append(q.packets, packetBuffer.Bytes())
}
q.peer.WritePackets(q.packets)
clear(q.packets)
q.packets = q.packets[:0]
buf.ReleaseMulti(packetBuffers)
}
transport/wireguard/device_stack.go
+15 -10
View File
@@ -12,7 +12,6 @@ import (
E "github.com/sagernet/sing/common/exceptions"
M "github.com/sagernet/sing/common/metadata"
N "github.com/sagernet/sing/common/network"
"github.com/sagernet/wireguard-go/conn"
"github.com/sagernet/wireguard-go/device"
wgTun "github.com/sagernet/wireguard-go/tun"
)
@@ -21,26 +20,29 @@ var _ Device = (*stackDevice)(nil)
type stackDevice struct {
stack *tun.Go
memoryTun *tun.MemoryTun
router *peerRouter
mtu uint32
events chan wgTun.Event
closed chan struct{}
closeOnce sync.Once
inet4Address netip.Addr
inet6Address netip.Addr
}
func newStackDevice(options DeviceOptions) (*stackDevice, error) {
memoryTun := tun.NewMemoryTun(tun.MemoryTunOptions{MTU: int(options.MTU)})
stack, err := newStack(options, memoryTun)
router := newPeerRouter(options)
stack, err := newStack(options, router.memoryTun)
if err != nil {
router.close()
return nil, err
}
inet4Address, inet6Address := deviceAddresses(options.Address)
return &stackDevice{
stack: stack,
memoryTun: memoryTun,
router: router,
mtu: options.MTU,
events: make(chan wgTun.Event, 1),
closed: make(chan struct{}),
inet4Address: inet4Address,
inet6Address: inet6Address,
}, nil
@@ -105,7 +107,8 @@ func (w *stackDevice) Inet6Address() netip.Addr {
return w.inet6Address
}
func (w *stackDevice) SetDevice(device *device.Device) {
func (w *stackDevice) SetDevice(device *device.Device, peers []*device.Peer) {
w.router.setPeers(device.AllowedIPs(), peers)
}
func (w *stackDevice) Start() error {
@@ -122,7 +125,8 @@ func (w *stackDevice) File() *os.File {
}
func (w *stackDevice) Read(bufs [][]byte, sizes []int, offset int) (int, error) {
return w.memoryTun.ReadPackets(bufs, sizes, offset)
<-w.closed
return 0, os.ErrClosed
}
func (w *stackDevice) Write(bufs [][]byte, offset int) (int, error) {
@@ -130,7 +134,7 @@ func (w *stackDevice) Write(bufs [][]byte, offset int) (int, error) {
for _, packet := range bufs {
packets = append(packets, packet[offset:])
}
return w.memoryTun.WritePackets(packets)
return w.router.memoryTun.WritePackets(packets)
}
func (w *stackDevice) Flush() error {
@@ -153,11 +157,12 @@ func (w *stackDevice) Close() error {
var err error
w.closeOnce.Do(func() {
close(w.events)
err = E.Errors(w.stack.Close(), w.memoryTun.Close())
close(w.closed)
err = E.Errors(w.stack.Close(), w.router.close())
})
return err
}
func (w *stackDevice) BatchSize() int {
return conn.IdealBatchSize
return 1
}
transport/wireguard/device_system.go
+1 -1
View File
@@ -61,7 +61,7 @@ func (w *systemDevice) Inet6Address() netip.Addr {
return w.inet6Address
}
func (w *systemDevice) SetDevice(device *device.Device) {
func (w *systemDevice) SetDevice(device *device.Device, peers []*device.Peer) {
}
func (w *systemDevice) Start() error {
transport/wireguard/device_system_stack.go
+13 -35
View File
@@ -3,11 +3,9 @@ package wireguard
import (
"net/netip"
"sync"
"sync/atomic"
"github.com/sagernet/sing-tun"
"github.com/sagernet/sing-tun/gtcpip/header"
"github.com/sagernet/sing/common/buf"
E "github.com/sagernet/sing/common/exceptions"
"github.com/sagernet/wireguard-go/device"
)
@@ -17,8 +15,7 @@ var _ Device = (*systemStackDevice)(nil)
type systemStackDevice struct {
*systemDevice
stack *tun.Go
memoryTun *tun.MemoryTun
device atomic.Pointer[device.Device]
router *peerRouter
closeOnce sync.Once
}
@@ -27,17 +24,21 @@ func newSystemStackDevice(options DeviceOptions) (*systemStackDevice, error) {
if err != nil {
return nil, err
}
stackDevice := &systemStackDevice{systemDevice: system}
stackDevice.memoryTun = tun.NewMemoryTun(tun.MemoryTunOptions{MTU: int(options.MTU), Outbound: stackDevice.inputPackets})
stackDevice.stack, err = newStack(options, stackDevice.memoryTun)
router := newPeerRouter(options)
stack, err := newStack(options, router.memoryTun)
if err != nil {
router.close()
return nil, err
}
return stackDevice, nil
return &systemStackDevice{
systemDevice: system,
stack: stack,
router: router,
}, nil
}
func (w *systemStackDevice) SetDevice(device *device.Device) {
w.device.Store(device)
func (w *systemStackDevice) SetDevice(device *device.Device, peers []*device.Peer) {
w.router.setPeers(device.AllowedIPs(), peers)
}
func (w *systemStackDevice) Start() error {
@@ -63,7 +64,7 @@ func (w *systemStackDevice) Write(bufs [][]byte, offset int) (int, error) {
}
}
if len(stackPackets) > 0 {
_, err := w.memoryTun.WritePackets(stackPackets)
_, err := w.router.memoryTun.WritePackets(stackPackets)
if err != nil {
return 0, err
}
@@ -92,33 +93,10 @@ func (w *systemStackDevice) isLocalDestination(packet []byte) bool {
return false
}
func (w *systemStackDevice) inputPackets(packets []*buf.Buffer) error {
defer buf.ReleaseMulti(packets)
wgDevice := w.device.Load()
if wgDevice == nil {
return nil
}
references := make([]*device.InputPacketRef, 0, len(packets))
for _, packet := range packets {
var destination []byte
switch header.IPVersion(packet.Bytes()) {
case header.IPv4Version:
destination = header.IPv4(packet.Bytes()).DestinationAddressSlice()
case header.IPv6Version:
destination = header.IPv6(packet.Bytes()).DestinationAddressSlice()
default:
continue
}
references = append(references, &device.InputPacketRef{Destination: destination, PacketSlices: [][]byte{packet.Bytes()}})
}
wgDevice.InputPackets(references)
return nil
}
func (w *systemStackDevice) Close() error {
var err error
w.closeOnce.Do(func() {
err = E.Errors(w.stack.Close(), w.memoryTun.Close(), w.systemDevice.Close())
err = E.Errors(w.stack.Close(), w.router.close(), w.systemDevice.Close())
})
return err
}
transport/wireguard/endpoint.go
+16 -13
View File
@@ -73,7 +73,10 @@ func NewEndpoint(options EndpointOptions) (*Endpoint, error) {
if err != nil {
return nil, E.Cause(err, "decode public key for peer ", peerIndex)
}
peer.publicKeyHex = hex.EncodeToString(publicKeyBytes)
if len(publicKeyBytes) != device.NoisePublicKeySize {
return nil, E.New("invalid public key for peer ", peerIndex, ", required ", device.NoisePublicKeySize, " bytes, got ", len(publicKeyBytes))
}
peer.publicKey = device.NoisePublicKey(publicKeyBytes)
if rawPeer.PreSharedKey != "" {
preSharedKeyBytes, err := base64.StdEncoding.DecodeString(rawPeer.PreSharedKey)
if err != nil {
@@ -215,19 +218,11 @@ func (e *Endpoint) Start(postStart bool) error {
},
}
wgDevice := device.NewDevice(e.options.Context, e.returnDevice, bind, logger, e.options.Workers)
e.tunDevice.SetDevice(wgDevice)
domainPeers := make(map[device.NoisePublicKey]*peerConfig)
for peerIndex, peer := range e.peers {
if !peer.destination.IsDomain() {
continue
if peer.destination.IsDomain() {
domainPeers[peer.publicKey] = &e.peers[peerIndex]
}
var publicKey device.NoisePublicKey
err = publicKey.FromHex(peer.publicKeyHex)
if err != nil {
wgDevice.Close()
return E.Cause(err, "decode public key for peer ", peerIndex)
}
domainPeers[publicKey] = &e.peers[peerIndex]
}
if len(domainPeers) > 0 {
wgDevice.SetEndpointResolverFunc(func(publicKey device.NoisePublicKey) ([]conn.Endpoint, error) {
@@ -264,6 +259,14 @@ func (e *Endpoint) Start(postStart bool) error {
wgDevice.Close()
return E.Cause(err, "setup wireguard: \n", ipcConf.String())
}
wgPeers := make([]*device.Peer, 0, len(e.peers))
for _, peer := range e.peers {
wgPeer, loaded := wgDevice.LookupActivePeer(peer.publicKey)
if loaded {
wgPeers = append(wgPeers, wgPeer)
}
}
e.tunDevice.SetDevice(wgDevice, wgPeers)
e.device.Store(wgDevice)
e.pause = service.FromContext[pause.Manager](e.options.Context)
if e.pause != nil {
@@ -392,7 +395,7 @@ func (e *Endpoint) updateNetworkPause() {
type peerConfig struct {
destination M.Socksaddr
endpoint netip.AddrPort
publicKeyHex string
publicKey device.NoisePublicKey
preSharedKeyHex string
allowedIPs []netip.Prefix
keepalive uint16
@@ -401,7 +404,7 @@ type peerConfig struct {
func (c peerConfig) GenerateIpcLines() string {
var ipcLines strings.Builder
ipcLines.WriteString("\npublic_key=" + c.publicKeyHex)
ipcLines.WriteString("\npublic_key=" + hex.EncodeToString(c.publicKey[:]))
if c.endpoint.IsValid() {
ipcLines.WriteString("\nendpoint=" + c.endpoint.String())
}