eburet-oss/sing-tun-lx
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Reject flows on selector range exhaustion

Browse source
This commit is contained in:
世界 2026-07-08 17:14:49 +08:00
parent 15b67423c3
commit 2de01a4c29
No known key found for this signature in database
GPG key ID: CD109927C34A63C4
1 changed files with 28 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

36
flow_dispatch.go
View file

@ -113,6 +113,7 @@ type ForwardDispatcher struct {
activeNATs []*portNAT activeNATs []*portNAT
writebackBatch [][]byte writebackBatch [][]byte
returnPath forwardReturn returnPath forwardReturn
exhaustedLogAt int64
segmentBuffers [][]byte segmentBuffers [][]byte
segmentSizes []int segmentSizes []int
@ -247,14 +248,23 @@ func (d *ForwardDispatcher) judgeAndInstall(key flowKey, packet *forwardPacket,
switch verdict.Action { switch verdict.Action {
case ActionFlow: case ActionFlow:
if verdict.Port != nil { if verdict.Port != nil {
flow, created := d.createFlow(packet, verdict) flow, result := d.createFlow(packet, verdict)
if created { if result == createFlowOK {
entry := &flowEntry{action: ActionFlow, flow: flow, idle: d.flowIdle(flow)} entry := &flowEntry{action: ActionFlow, flow: flow, idle: d.flowIdle(flow)}
entry.deadline = now + int64(entry.idle) entry.deadline = now + int64(entry.idle)
d.insertEntry(key, entry, now) d.insertEntry(key, entry, now)
d.forwardToPort(flow, packet, raw) d.forwardToPort(flow, packet, raw)
return true return true
} }
if result == createFlowExhausted {
if now-d.exhaustedLogAt >= int64(exhaustedLogInterval) {
d.exhaustedLogAt = now
d.logger.Warn("port selector range exhausted, rejecting flow to ", packet.destination)
}
d.installSimple(key, ActionReject, packet.protocol, now)
d.stageReject(packet)
return true
}
} }
d.installSimple(key, ActionAccept, packet.protocol, now) d.installSimple(key, ActionAccept, packet.protocol, now)
return false return false
@ -302,7 +312,17 @@ func (d *ForwardDispatcher) flowIdle(flow *forwardFlow) time.Duration {
return d.idleTimeout(flow.protocol, established) return d.idleTimeout(flow.protocol, established)
} }
func (d *ForwardDispatcher) createFlow(packet *forwardPacket, verdict FlowVerdict) (*forwardFlow, bool) { type createFlowResult uint8
const (
createFlowOK createFlowResult = iota
createFlowUnsupported
createFlowExhausted
)
const exhaustedLogInterval = 5 * time.Second
func (d *ForwardDispatcher) createFlow(packet *forwardPacket, verdict FlowVerdict) (*forwardFlow, createFlowResult) {
var portAddress netip.Addr var portAddress netip.Addr
inet4Address, inet6Address := verdict.Port.PortAddresses() inet4Address, inet6Address := verdict.Port.PortAddresses()
if packet.ipVersion == 6 { if packet.ipVersion == 6 {
@ -311,11 +331,11 @@ func (d *ForwardDispatcher) createFlow(packet *forwardPacket, verdict FlowVerdic
portAddress = inet4Address portAddress = inet4Address
} }
if !portAddress.IsValid() { if !portAddress.IsValid() {
return nil, false return nil, createFlowUnsupported
} }
effectiveMTU := verdict.Port.PortMTU() effectiveMTU := verdict.Port.PortMTU()
if packet.ipVersion == 6 && effectiveMTU != 0 && effectiveMTU < header.IPv6MinimumMTU { if packet.ipVersion == 6 && effectiveMTU != 0 && effectiveMTU < header.IPv6MinimumMTU {
return nil, false return nil, createFlowUnsupported
} }
isICMP := isICMPProtocol(packet.protocol) isICMP := isICMPProtocol(packet.protocol)
clientDestinationAddress := packet.destination.Addr() clientDestinationAddress := packet.destination.Addr()
@ -330,11 +350,11 @@ func (d *ForwardDispatcher) createFlow(packet *forwardPacket, verdict FlowVerdic
} }
nat := d.natFor(verdict.Port) nat := d.natFor(verdict.Port)
if nat == nil { if nat == nil {
return nil, false return nil, createFlowUnsupported
} }
selector, reverseKey, allocated := nat.allocateSelector(packet.protocol, portAddress, serverAddress, serverPort, packet.source.Port()) selector, reverseKey, allocated := nat.allocateSelector(packet.protocol, portAddress, serverAddress, serverPort, packet.source.Port())
if !allocated { if !allocated {
return nil, false return nil, createFlowExhausted
} }
var udpTimeout time.Duration var udpTimeout time.Duration
if packet.protocol == uint8(header.UDPProtocolNumber) { if packet.protocol == uint8(header.UDPProtocolNumber) {
@ -385,7 +405,7 @@ func (d *ForwardDispatcher) createFlow(packet *forwardPacket, verdict FlowVerdic
} }
} }
nat.insert(reverseKey, flow) nat.insert(reverseKey, flow)
return flow, true return flow, createFlowOK
} }
func (d *ForwardDispatcher) natFor(port Port) *portNAT { func (d *ForwardDispatcher) natFor(port Port) *portNAT {