From f988e9677f6ed3d6ca95bcd1c4bdb42620551c05 Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Sat, 31 Oct 2026 15:18:43 +0100 Subject: [PATCH] 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. --- internal/amneziawgnet/portfwd.go | 31 ++++++- internal/amneziawgnet/portfwd_test.go | 113 ++++++++++++++++++++++++++ internal/amneziawgnet/portfwd_udp.go | 6 ++ 3 files changed, 149 insertions(+), 1 deletion(-) diff --git a/internal/amneziawgnet/portfwd.go b/internal/amneziawgnet/portfwd.go index 0a7aa5a4d..9aa03dfd4 100644 --- a/internal/amneziawgnet/portfwd.go +++ b/internal/amneziawgnet/portfwd.go @@ -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) + }() } } diff --git a/internal/amneziawgnet/portfwd_test.go b/internal/amneziawgnet/portfwd_test.go index 43eeef3e1..9a55c7f29 100644 --- a/internal/amneziawgnet/portfwd_test.go +++ b/internal/amneziawgnet/portfwd_test.go @@ -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) +} diff --git a/internal/amneziawgnet/portfwd_udp.go b/internal/amneziawgnet/portfwd_udp.go index 1c315bcb5..0e8d4a11b 100644 --- a/internal/amneziawgnet/portfwd_udp.go +++ b/internal/amneziawgnet/portfwd_udp.go @@ -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 {