fix(tuic): bound what unauthenticated connections cost

Anyone reaching a TUIC port could start QUIC handshakes without ever reading a reply, each holding state and goroutines, and an unauthenticated connection got a handler goroutine per stream it opened. Both grew until the panel ran out of memory. The listener now counts connections from their first packet until they authenticate or close: past 64 they must answer a stateless Retry, and past 1024 in total or 32 per verified source (IPv4 address or IPv6 /64) they are refused. Before authenticating, a connection may start at most 16 handlers per stream kind; further streams are reset.
2026-11-02 19:25:05 +03:00 · 2026-10-31 15:03:21 +01:00
parent e51a32b045
commit 6d90dcfe0c
3 changed files with 240 additions and 5 deletions
@@ -0,0 +1,112 @@
+package tuic
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/netip"
+	"sync"
+)
+
+// A connection is pending from its first Initial packet until it authenticates or closes;
+// each one holds handshake state and goroutines, so an unauthenticated flood is bounded here.
+const (
+	// pendingRetryThreshold is where new connections must first answer a stateless Retry,
+	// so a sender that never reads replies, or forges its source, creates no state.
+	pendingRetryThreshold = 64
+	maxPendingConns       = 1024
+	// maxPendingPerSource counts verified sources only (an IPv4 address or IPv6 /64).
+	maxPendingPerSource = 32
+	// maxPreAuthStreams bounds the stream handlers one connection may start per stream
+	// kind before it authenticates.
+	maxPreAuthStreams = 16
+)
+
+var errTooManyPendingConns = errors.New("tuic: too many connections awaiting authentication")
+
+type pendingConns struct {
+	mu       sync.Mutex
+	total    int
+	bySource map[netip.Prefix]int
+}
+
+type pendingSlotKey struct{}
+
+// pendingSlot is one connection's place in the count, released exactly once.
+type pendingSlot struct {
+	once    sync.Once
+	release func()
+}
+
+func (p *pendingSlot) done() {
+	if p != nil {
+		p.once.Do(p.release)
+	}
+}
+
+func (p *pendingConns) needsRetry(net.Addr) bool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.total >= pendingRetryThreshold
+}
+
+// admit takes a slot for a new connection, or refuses it at the ceilings; the slot frees
+// when ctx (the connection's own) ends or the connection authenticates.
+func (p *pendingConns) admit(ctx context.Context, info connInfo) (context.Context, error) {
+	source, counted := pendingSource(info)
+	p.mu.Lock()
+	if p.total >= maxPendingConns || (counted && p.bySource[source] >= maxPendingPerSource) {
+		p.mu.Unlock()
+		return ctx, errTooManyPendingConns
+	}
+	p.total++
+	if counted {
+		if p.bySource == nil {
+			p.bySource = make(map[netip.Prefix]int)
+		}
+		p.bySource[source]++
+	}
+	p.mu.Unlock()
+	slot := &pendingSlot{release: func() {
+		p.mu.Lock()
+		defer p.mu.Unlock()
+		p.total--
+		if counted {
+			if p.bySource[source]--; p.bySource[source] <= 0 {
+				delete(p.bySource, source)
+			}
+		}
+	}}
+	context.AfterFunc(ctx, slot.done)
+	return context.WithValue(ctx, pendingSlotKey{}, slot), nil
+}
+
+// connInfo is what admit needs of quic.ClientInfo.
+type connInfo struct {
+	remote          net.Addr
+	addressVerified bool
+}
+
+func pendingSource(info connInfo) (netip.Prefix, bool) {
+	udp, ok := info.remote.(*net.UDPAddr)
+	if !ok || !info.addressVerified {
+		return netip.Prefix{}, false
+	}
+	addr, ok := netip.AddrFromSlice(udp.IP)
+	if !ok {
+		return netip.Prefix{}, false
+	}
+	addr = addr.Unmap()
+	bits := 32
+	if addr.Is6() {
+		bits = 64
+	}
+	prefix, err := addr.Prefix(bits)
+	return prefix, err == nil
+}
+
+func releasePendingSlot(ctx context.Context) {
+	if slot, ok := ctx.Value(pendingSlotKey{}).(*pendingSlot); ok {
+		slot.done()
+	}
+}
@@ -0,0 +1,82 @@
+package tuic
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"net"
+	"runtime"
+	"testing"
+	"time"
+
+	clientquic "github.com/quic-go/quic-go"
+)
+
+// Connections awaiting authentication are counted from their first packet: past the
+// threshold they must prove their address, and past the ceilings they are refused.
+func TestPendingConnsRetryAndCeilings(t *testing.T) {
+	var p pendingConns
+	verified := func(ip string) connInfo {
+		return connInfo{remote: &net.UDPAddr{IP: net.ParseIP(ip), Port: 1}, addressVerified: true}
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	for i := range pendingRetryThreshold {
+		if p.needsRetry(nil) {
+			t.Fatalf("Retry demanded with only %d pending", i)
+		}
+		if _, err := p.admit(ctx, connInfo{remote: &net.UDPAddr{IP: net.IPv4(203, 0, 113, byte(i)), Port: 1}}); err != nil {
+			t.Fatalf("admit %d: %v", i, err)
+		}
+	}
+	if !p.needsRetry(nil) {
+		t.Fatal("no Retry demanded at the threshold")
+	}
+	for range maxPendingPerSource {
+		if _, err := p.admit(ctx, verified("2001:db8::1")); err != nil {
+			t.Fatalf("admit from one source: %v", err)
+		}
+	}
+	if _, err := p.admit(ctx, verified("2001:db8::2")); !errors.Is(err, errTooManyPendingConns) {
+		t.Fatalf("one /64 past its share = %v, want refused", err)
+	}
+	cancel()
+	deadline := time.Now().Add(time.Second)
+	for {
+		p.mu.Lock()
+		total := p.total
+		p.mu.Unlock()
+		if total == 0 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("%d slots still held after their connections ended", total)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}
+
+// A peer that never authenticates must not get a handler goroutine per stream it opens.
+func TestUnauthenticatedStreamsShareASmallHandlerBudget(t *testing.T) {
+	server, _, _, _ := startLifecycleTestServer(t, "127.0.0.1:1", "preauth@x")
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	peer, err := clientquic.DialAddr(ctx, server.packetConn.LocalAddr().String(),
+		&tls.Config{InsecureSkipVerify: true, NextProtos: []string{"h3"}}, &clientquic.Config{})
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	defer func() { _ = peer.CloseWithError(0, "") }()
+	time.Sleep(100 * time.Millisecond)
+	before := runtime.NumGoroutine()
+	for range 90 {
+		stream, err := peer.OpenStreamSync(ctx)
+		if err != nil {
+			break
+		}
+		_, _ = stream.Write([]byte{ProtocolVersion, CmdConnect})
+	}
+	time.Sleep(300 * time.Millisecond)
+	if grown := runtime.NumGoroutine() - before; grown > 2*maxPreAuthStreams {
+		t.Fatalf("90 unauthenticated streams started %d goroutines", grown)
+	}
+}
@@ -38,10 +38,12 @@ type Server struct {
 	users *UserRegistry
 	relay *SocksRelay

-	tlsConfig    *tls.Config
-	quicConfig   *quic.Config
-	quicListener *quic.Listener
-	packetConn   net.PacketConn
+	tlsConfig     *tls.Config
+	quicConfig    *quic.Config
+	quicListener  *quic.Listener
+	quicTransport *quic.Transport
+	pending       pendingConns
+	packetConn    net.PacketConn

 	lastOnline  sync.Map // email string -> time.Time
 	logThrottle sync.Map // event name -> *atomic.Int64 timestamp
@@ -146,12 +148,21 @@ func (s *Server) Start() error {
 	}
 	s.packetConn = pConn

-	ln, err := quic.Listen(pConn, s.tlsConfig, s.quicConfig)
+	transport := &quic.Transport{
+		Conn:                pConn,
+		VerifySourceAddress: s.pending.needsRetry,
+		ConnContext: func(ctx context.Context, info *quic.ClientInfo) (context.Context, error) {
+			return s.pending.admit(ctx, connInfo{remote: info.RemoteAddr, addressVerified: info.AddrVerified})
+		},
+	}
+	ln, err := transport.Listen(s.tlsConfig, s.quicConfig)
 	if err != nil {
+		_ = transport.Close()
 		_ = pConn.Close()
 		return fmt.Errorf("tuic: quic listen on %s: %w", s.listenAddr, err)
 	}
 	s.quicListener = ln
+	s.quicTransport = transport
 	s.running.Store(true)
 	s.logf(tuicLogInfo, "listener started on %s", s.listenAddr)

@@ -391,6 +402,7 @@ func (s *Server) handleConn(conn *quic.Conn) {
 			authUser.Store(user)
 			s.markActive(user.Email)
 			authTimer.Stop()
+			releasePendingSlot(conn.Context())
 			s.logf(tuicLogInfo, "client authenticated")
 			authOnce.Do(func() { close(authSignal) })
 			return true
@@ -430,6 +442,19 @@ func (s *Server) handleConn(conn *quic.Conn) {
 	defer cleanup()

 	var innerWg sync.WaitGroup
+	// Before authentication each stream kind may start only a few handlers; a peer that never
+	// authenticates would otherwise get a goroutine per stream it opens.
+	var preAuthUni, preAuthBi atomic.Int32
+	preAuthSlot := func(count *atomic.Int32) (release func(), ok bool) {
+		if authState.Load() == 1 {
+			return func() {}, true
+		}
+		if count.Add(1) > maxPreAuthStreams {
+			count.Add(-1)
+			return nil, false
+		}
+		return func() { count.Add(-1) }, true
+	}
 	// Loop 1: Unidirectional streams
 	innerWg.Add(1)
 	go func() {
@@ -439,9 +464,15 @@ func (s *Server) handleConn(conn *quic.Conn) {
 			if err != nil {
 				return
 			}
+			release, ok := preAuthSlot(&preAuthUni)
+			if !ok {
+				uniStream.CancelRead(0x100)
+				continue
+			}
 			innerWg.Add(1)
 			go func(stream *quic.ReceiveStream) {
 				defer innerWg.Done()
+				defer release()
 				s.handleUniStream(sessCtx, conn, stream, authenticate, waitForAuth, udpAssociations, &relayWg)
 			}(uniStream)
 		}
@@ -456,9 +487,16 @@ func (s *Server) handleConn(conn *quic.Conn) {
 			if err != nil {
 				return
 			}
+			release, ok := preAuthSlot(&preAuthBi)
+			if !ok {
+				biStream.CancelRead(0x100)
+				biStream.CancelWrite(0x100)
+				continue
+			}
 			innerWg.Add(1)
 			go func(stream *quic.Stream) {
 				defer innerWg.Done()
+				defer release()
 				s.handleBiStream(sessCtx, conn, stream, authenticate, waitForAuth)
 			}(biStream)
 		}
@@ -1032,6 +1070,9 @@ func (s *Server) Close() error {
 	if s.quicListener != nil {
 		err = s.quicListener.Close()
 	}
+	if s.quicTransport != nil {
+		_ = s.quicTransport.Close()
+	}
 	if s.packetConn != nil {
 		_ = s.packetConn.Close()
 	}