mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-11-02 03:05:02 +03:00
fix(amneziawg): cap port-forward relays and UDP sessions
A forwarded port accepts unauthenticated traffic, and every TCP connection and every new UDP source got its own relay with a goroutine and a buffer, with no limit, so anyone reaching the port could exhaust the panel's memory. Live TCP relays and UDP sessions are now capped at 2048 each across all forwards; past that a connection is closed and a datagram from a new source is dropped, while open flows keep going.
This commit is contained in:
@@ -270,6 +270,28 @@ func openPortForwardListener(gstack *stack.Stack, inboundID int, key portForward
|
||||
|
||||
const portForwardDialTimeout = 10 * time.Second
|
||||
|
||||
// portForwardMaxFlows caps live TCP relays, and separately UDP sessions, over every
|
||||
// listener: senders are unauthenticated, and each flow pins a goroutine and a buffer.
|
||||
const portForwardMaxFlows = 2048
|
||||
|
||||
// Process-wide on purpose: a per-listener cap still multiplies by the forwarded ports
|
||||
// and by the number of forwarding peers.
|
||||
var (
|
||||
portForwardTCPSlots = make(chan struct{}, portForwardMaxFlows)
|
||||
portForwardUDPSlots = make(chan struct{}, portForwardMaxFlows)
|
||||
)
|
||||
|
||||
// acquirePortForwardSlot never blocks the accept or read loop; false means the flow is
|
||||
// refused. The owner gives the slot back by receiving from slots.
|
||||
func acquirePortForwardSlot(slots chan struct{}) bool {
|
||||
select {
|
||||
case slots <- struct{}{}:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// acceptRetryDelay paces an accept loop's retries, so an error that lasts
|
||||
// (EMFILE, until descriptors free up) neither spins it nor floods the log.
|
||||
const acceptRetryDelay = 100 * time.Millisecond
|
||||
@@ -349,7 +371,14 @@ func (l *tcpForwardListener) acceptLoop(gstack *stack.Stack, inboundID int, key
|
||||
}
|
||||
continue
|
||||
}
|
||||
go l.relayTCPForward(gstack, conn, inboundID, key, target)
|
||||
if !acquirePortForwardSlot(portForwardTCPSlots) {
|
||||
conn.Close()
|
||||
continue
|
||||
}
|
||||
go func() {
|
||||
defer func() { <-portForwardTCPSlots }()
|
||||
l.relayTCPForward(gstack, conn, inboundID, key, target)
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -477,3 +477,116 @@ primed:
|
||||
t.Errorf("read on the relayed connection after its forward was withdrawn = %v, want io.EOF", err)
|
||||
}
|
||||
}
|
||||
|
||||
// leavePortForwardSlots fills slots until only free remain, so a test hits the
|
||||
// ceiling with a handful of flows; returns how many it holds (cleanup releases).
|
||||
func leavePortForwardSlots(t *testing.T, slots chan struct{}, free int) int {
|
||||
t.Helper()
|
||||
held := 0
|
||||
for acquirePortForwardSlot(slots) {
|
||||
held++
|
||||
}
|
||||
for range free {
|
||||
<-slots
|
||||
held--
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
for range held {
|
||||
<-slots
|
||||
}
|
||||
})
|
||||
return held
|
||||
}
|
||||
|
||||
func waitPortForwardSlots(t *testing.T, slots chan struct{}, want int) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for len(slots) > want {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("slots still taken = %d, want at most %d", len(slots), want)
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPortForwardUDPSessionsAreCapped(t *testing.T) {
|
||||
const port = 58913
|
||||
const free, sources = 3, 10
|
||||
held := leavePortForwardSlots(t, portForwardUDPSlots, free)
|
||||
|
||||
gs := newTestStack(t, "10.211.2.1")
|
||||
l := listenPortForwardUDP(gs, 503, portForwardKey{email: "a@x", port: port, proto: udpForward}, func(string) (netip.Addr, bool) {
|
||||
return netip.MustParseAddr("10.211.2.2"), true
|
||||
})
|
||||
if l == nil {
|
||||
t.Fatal("listenPortForwardUDP returned nil")
|
||||
}
|
||||
defer l.Close()
|
||||
|
||||
for range sources {
|
||||
conn, err := net.Dial("udp", fmt.Sprintf("127.0.0.1:%d", port))
|
||||
if err != nil {
|
||||
t.Fatalf("dial udp: %v", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
if _, err := conn.Write([]byte("x")); err != nil {
|
||||
t.Fatalf("write udp: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Long enough for readLoop to have drained every datagram sent above.
|
||||
most := 0
|
||||
for settle := time.Now().Add(500 * time.Millisecond); time.Now().Before(settle); time.Sleep(10 * time.Millisecond) {
|
||||
l.mu.Lock()
|
||||
most = max(most, len(l.sessions))
|
||||
l.mu.Unlock()
|
||||
}
|
||||
if most < free || most > portForwardMaxFlows-held {
|
||||
t.Fatalf("sessions opened by %d sources = %d, want the %d free slots", sources, most, free)
|
||||
}
|
||||
|
||||
// A slot that outlived its session would shrink the ceiling for good.
|
||||
l.Close()
|
||||
waitPortForwardSlots(t, portForwardUDPSlots, held)
|
||||
}
|
||||
|
||||
func TestPortForwardTCPConnectionsAreCapped(t *testing.T) {
|
||||
const port = 58914
|
||||
const free, clients = 3, 10
|
||||
held := leavePortForwardSlots(t, portForwardTCPSlots, free)
|
||||
|
||||
// Parks every relayed connection at its first step, holding its slot.
|
||||
release := make(chan struct{})
|
||||
gs := newTestStack(t, "10.211.3.1")
|
||||
l := listenPortForwardTCP(gs, 504, portForwardKey{email: "a@x", port: port, proto: tcpForward}, func(string) (netip.Addr, bool) {
|
||||
<-release
|
||||
return netip.Addr{}, false
|
||||
})
|
||||
if l == nil {
|
||||
t.Fatal("listenPortForwardTCP returned nil")
|
||||
}
|
||||
defer l.Close()
|
||||
|
||||
relayed := 0
|
||||
for range clients {
|
||||
conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("dial tcp: %v", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
// A refused connection is closed at once; a relayed one stays silent.
|
||||
_ = conn.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
|
||||
if _, err := conn.Read(make([]byte, 1)); err != nil {
|
||||
var nerr net.Error
|
||||
if errors.As(err, &nerr) && nerr.Timeout() {
|
||||
relayed++
|
||||
}
|
||||
}
|
||||
}
|
||||
if relayed < free || relayed > portForwardMaxFlows-held {
|
||||
t.Fatalf("connections relayed out of %d = %d, want the %d free slots", clients, relayed, free)
|
||||
}
|
||||
|
||||
close(release)
|
||||
waitPortForwardSlots(t, portForwardTCPSlots, held)
|
||||
}
|
||||
|
||||
@@ -78,9 +78,14 @@ func (l *udpForwardListener) readLoop(gstack *stack.Stack, inboundID int, key po
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
// Dropped silently: a flood of fresh sources is when a per-datagram log line hurts.
|
||||
if !acquirePortForwardSlot(portForwardUDPSlots) {
|
||||
continue
|
||||
}
|
||||
raddr := tunnelFullAddress(addr, key.port)
|
||||
conn, err := gonet.DialUDP(gstack, nil, &raddr, tunnelNetwork(addr))
|
||||
if err != nil {
|
||||
<-portForwardUDPSlots
|
||||
logger.Warningf("amneziawgnet: port-forward: inbound %d peer %q: dial tunnel %s:%d: %v", inboundID, key.email, addr, key.port, err)
|
||||
continue
|
||||
}
|
||||
@@ -109,6 +114,7 @@ func (l *udpForwardListener) pump(src netip.AddrPort, sess *udpForwardSession) {
|
||||
delete(l.sessions, src)
|
||||
l.mu.Unlock()
|
||||
sess.conn.Close()
|
||||
<-portForwardUDPSlots
|
||||
}()
|
||||
buf := make([]byte, 65536)
|
||||
for {
|
||||
|
||||
Reference in New Issue
Block a user