Add per-participant concurrent TURN allocation quota (#4744)

* Add per-participant concurrent TURN allocation quota The embedded TURN server authenticated each Allocate request but placed no cap on how many relay allocations a single participant credential could hold. One participant could reuse its credential across many client 5-tuples and open one relay socket/port per request, exhausting the shared relay-port range for everyone else. Add a configurable per-participant limit (turn.per_user_relay_allocation_limit, default 4) wired to Pion's QuotaHandler, keyed by the participant ID from HandleAuth. Slots are reserved before allocation and released when the allocation ends, under a single lock, so concurrent Allocate bursts cannot race past the limit; reservations are keyed by source address so retransmits are idempotent. Over-quota requests receive 486 (Allocation Quota Reached). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Reclaim unconfirmed TURN allocation reservations Allow reserved a quota slot before Pion built the relay, but the slot was only released on the allocation-deleted event. An Allocate that passed the quota check and then failed to create a relay (e.g. relay-port range exhausted) emits no event, so the reservation leaked: after enough failures a participant could lock itself out with 486, and the tracking map grew without bound. Reservations now start pending and are confirmed on allocation-created; each pending reservation carries a reclaim timer that frees the slot after a TTL, so a failed attempt cannot hold a slot forever while concurrent-burst safety is preserved. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Make TURN reservation reclaim identity-aware The reclaim timer captured only userID+key. Because Timer.Stop cannot cancel a callback that has already fired and is waiting on the lock, a stale timer could delete a replacement reservation created for the same userID+key after the original was released, leaving a live allocation untracked and letting the participant exceed its cap. reclaimPending now captures the slot pointer and only removes the entry when the map still holds that exact slot, so a stale timer is a no-op. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-08-22 18:40:34 +00:00 · 2026-08-11 23:12:07 +05:30
parent 7167f91493
commit 223587e140
4 changed files with 415 additions and 3 deletions
@@ -59,6 +59,12 @@ const (
 	// DefaultExternalTURNTTLSeconds is the default TTL applied to external TURN
 	// (static-auth-secret) credentials when the configured TTL is left at 0.
 	DefaultExternalTURNTTLSeconds = 14400
+
+	// DefaultTURNPerUserRelayAllocationLimit bounds how many concurrent relay
+	// allocations a single participant credential may hold on the embedded TURN
+	// server. It prevents one authenticated participant from exhausting the
+	// shared relay-port range for everyone else. A value <= 0 disables the quota.
+	DefaultTURNPerUserRelayAllocationLimit = 4
 )

 var (
@@ -268,6 +274,12 @@ type TURNConfig struct {
 	RelayPortRangeEnd   uint16   `yaml:"relay_range_end,omitempty"`
 	ExternalTLS         bool     `yaml:"external_tls,omitempty"`
 	BindAddresses       []string `yaml:"bind_addresses,omitempty"`
+	// PerUserRelayAllocationLimit caps the number of concurrent relay allocations
+	// a single participant credential may hold, keyed by the participant ID. This
+	// stops one authenticated participant from consuming the shared relay-port
+	// range and denying TURN relays to others. Defaults to
+	// DefaultTURNPerUserRelayAllocationLimit; a value <= 0 disables the quota.
+	PerUserRelayAllocationLimit int `yaml:"per_user_relay_allocation_limit,omitempty"`
 	// TTL of the TURN credentials in seconds - defaults to 300. Values <= 0 fall back to the
 	// 300s (5m) default and large values are capped at TURNMaxTTLSeconds.
 	TTLSeconds int `yaml:"ttl_seconds,omitempty"`
@@ -536,9 +548,10 @@ var DefaultConfig = Config{
 		PionLevel: "error",
 	},
 	TURN: TURNConfig{
-		Enabled:       false,
-		BindAddresses: []string{"0.0.0.0"},
-		TTLSeconds:    DefaultTURNTTLSeconds,
+		Enabled:                     false,
+		BindAddresses:               []string{"0.0.0.0"},
+		TTLSeconds:                  DefaultTURNTTLSeconds,
+		PerUserRelayAllocationLimit: DefaultTURNPerUserRelayAllocationLimit,
 	},
 	NodeSelector: NodeSelectorConfig{
 		Kind:         "any",
@@ -95,9 +95,18 @@ func NewTurnServer(conf *config.Config, authHandler turn.AuthHandler, standalone
 		LoggerFactory: pionlogger.NewLoggerFactory(logger.GetLogger()),
 	}

+	// cap concurrent relay allocations per participant so one credential cannot
+	// exhaust the shared relay-port range (a value <= 0 disables the quota)
+	if turnConf.PerUserRelayAllocationLimit > 0 {
+		quota := newTURNAllocationQuota(turnConf.PerUserRelayAllocationLimit)
+		serverConfig.QuotaHandler = quota.Allow
+		serverConfig.EventHandler = quota.eventHandler()
+	}
+
 	var logValues []any
 	logValues = append(logValues, "turn.relay_range_start", turnConf.RelayPortRangeStart)
 	logValues = append(logValues, "turn.relay_range_end", turnConf.RelayPortRangeEnd)
+	logValues = append(logValues, "turn.per_user_relay_allocation_limit", turnConf.PerUserRelayAllocationLimit)

 	for _, addr := range turnConf.BindAddresses {
 		var nodeIP string
@@ -0,0 +1,187 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"net"
+	"sync"
+	"time"
+
+	"github.com/pion/turn/v5"
+
+	"github.com/livekit/protocol/logger"
+)
+
+// defaultTURNReservationTTL bounds how long a reserved-but-unconfirmed
+// allocation slot may live. Pion calls the QuotaHandler and then synchronously
+// creates the allocation, so a healthy reservation is confirmed within
+// microseconds; this generous window only matters if allocation creation fails
+// after the quota check (see turnAllocationQuota).
+const defaultTURNReservationTTL = 30 * time.Second
+
+// turnAllocationQuota enforces a per-user cap on concurrent relay allocations on
+// the embedded TURN server. Without it, a single authenticated participant can
+// reuse its credential across many client 5-tuples and open one relay socket/port
+// per request, exhausting the shared relay-port range for every other participant.
+//
+// The quota is keyed by the user ID returned from TURNAuthHandler.HandleAuth,
+// which is the stable participant ID embedded in the signed TURN username. A
+// participant cannot forge a different ID without the API secret, so this bounds
+// the total relay footprint of any one participant.
+//
+// Slots are reserved in Allow (which Pion calls before creating an allocation)
+// and released in OnDeleted, all under a single lock, so a burst of concurrent
+// Allocate requests cannot race past the limit. Reservations are keyed by the
+// client source address so that Allocate retransmissions from the same 5-tuple
+// are idempotent rather than double-counted.
+//
+// A reservation starts out pending and is confirmed in OnCreated. Pion emits no
+// event when an Allocate passes the quota check but then fails to create a relay
+// (e.g. the relay-port range is exhausted), so each pending reservation carries a
+// reclaim timer that frees the slot after reservationTTL. This keeps a failed
+// attempt from occupying a slot forever, which would otherwise let a participant
+// lock itself out and grow the tracking map without bound.
+type turnAllocationQuota struct {
+	limit          int
+	reservationTTL time.Duration
+
+	mu    sync.Mutex
+	users map[string]map[string]*turnAllocationSlot // userID -> source-address key -> slot
+}
+
+type turnAllocationSlot struct {
+	confirmed bool
+	timer     *time.Timer
+}
+
+func newTURNAllocationQuota(limit int) *turnAllocationQuota {
+	return &turnAllocationQuota{
+		limit:          limit,
+		reservationTTL: defaultTURNReservationTTL,
+		users:          make(map[string]map[string]*turnAllocationSlot),
+	}
+}
+
+func srcAddrKey(srcAddr net.Addr) string {
+	if srcAddr == nil {
+		return ""
+	}
+	// include the network so an ip:port reused across UDP and TCP relays is not
+	// collapsed into a single slot
+	return srcAddr.Network() + "|" + srcAddr.String()
+}
+
+// Allow implements turn.QuotaHandler. It returns true if the allocation may
+// proceed, reserving a slot for the user, and false (Pion replies 486 Allocation
+// Quota Reached) once the user is at its limit.
+func (q *turnAllocationQuota) Allow(userID, _ string, srcAddr net.Addr) bool {
+	key := srcAddrKey(srcAddr)
+
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	slots := q.users[userID]
+	if _, ok := slots[key]; ok {
+		// already reserved for this 5-tuple (retransmit / create race) - idempotent
+		return true
+	}
+	if len(slots) >= q.limit {
+		logger.Infow("TURN allocation quota reached",
+			"participantID", userID,
+			"limit", q.limit,
+		)
+		return false
+	}
+	if slots == nil {
+		slots = make(map[string]*turnAllocationSlot)
+		q.users[userID] = slots
+	}
+	// reserve a pending slot; reclaim it if the allocation is never created so a
+	// failed attempt (Pion emits no created/deleted event) cannot hold it forever
+	slot := &turnAllocationSlot{}
+	slot.timer = time.AfterFunc(q.reservationTTL, func() {
+		q.reclaimPending(userID, key, slot)
+	})
+	slots[key] = slot
+	return true
+}
+
+// OnCreated confirms a reservation once the relay allocation actually exists,
+// cancelling its reclaim timer so a live allocation is never evicted early.
+func (q *turnAllocationQuota) OnCreated(srcAddr, _ net.Addr, _, userID, _ string, _ net.Addr, _ int) {
+	key := srcAddrKey(srcAddr)
+
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	slot := q.users[userID][key]
+	if slot == nil {
+		return
+	}
+	slot.confirmed = true
+	if slot.timer != nil {
+		slot.timer.Stop()
+		slot.timer = nil
+	}
+}
+
+// OnDeleted releases the slot reserved in Allow so the user regains capacity once
+// an allocation ends.
+func (q *turnAllocationQuota) OnDeleted(srcAddr, _ net.Addr, _, userID, _ string) {
+	key := srcAddrKey(srcAddr)
+
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	q.removeLocked(userID, key)
+}
+
+// reclaimPending drops a reservation that was never confirmed, freeing a slot
+// left behind by an Allocate that passed the quota but failed to create a relay.
+// It is identity-aware: a timer that fires after its slot was already released
+// must not evict a replacement reservation created for the same userID+key.
+func (q *turnAllocationQuota) reclaimPending(userID, key string, slot *turnAllocationSlot) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	if q.users[userID][key] != slot || slot.confirmed {
+		return
+	}
+	q.removeLocked(userID, key)
+}
+
+func (q *turnAllocationQuota) removeLocked(userID, key string) {
+	slots := q.users[userID]
+	slot := slots[key]
+	if slot == nil {
+		return
+	}
+	if slot.timer != nil {
+		slot.timer.Stop()
+	}
+	delete(slots, key)
+	if len(slots) == 0 {
+		delete(q.users, userID)
+	}
+}
+
+// eventHandler returns the Pion EventHandler wired to confirm and release quota
+// slots as allocations are created and torn down.
+func (q *turnAllocationQuota) eventHandler() turn.EventHandler {
+	return turn.EventHandler{
+		OnAllocationCreated: q.OnCreated,
+		OnAllocationDeleted: q.OnDeleted,
+	}
+}
@@ -0,0 +1,203 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"fmt"
+	"net"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func udpAddr(t *testing.T, port int) net.Addr {
+	t.Helper()
+	return &net.UDPAddr{IP: net.IPv4(10, 0, 0, 1), Port: port}
+}
+
+func TestTURNAllocationQuota_LimitsPerUser(t *testing.T) {
+	q := newTURNAllocationQuota(4)
+
+	// four distinct source 5-tuples for the same participant are allowed
+	for i := 0; i < 4; i++ {
+		require.True(t, q.Allow("participantA", LivekitRealm, udpAddr(t, 5000+i)), "allocation %d", i)
+	}
+	// the fifth is rejected (Pion returns 486 Allocation Quota Reached)
+	require.False(t, q.Allow("participantA", LivekitRealm, udpAddr(t, 6000)))
+}
+
+func TestTURNAllocationQuota_IsolatedPerUser(t *testing.T) {
+	q := newTURNAllocationQuota(2)
+
+	require.True(t, q.Allow("A", LivekitRealm, udpAddr(t, 5000)))
+	require.True(t, q.Allow("A", LivekitRealm, udpAddr(t, 5001)))
+	require.False(t, q.Allow("A", LivekitRealm, udpAddr(t, 5002)))
+
+	// a different participant has its own budget
+	require.True(t, q.Allow("B", LivekitRealm, udpAddr(t, 5000)))
+	require.True(t, q.Allow("B", LivekitRealm, udpAddr(t, 5001)))
+	require.False(t, q.Allow("B", LivekitRealm, udpAddr(t, 5002)))
+}
+
+func TestTURNAllocationQuota_ReleaseFreesSlot(t *testing.T) {
+	q := newTURNAllocationQuota(1)
+
+	addr := udpAddr(t, 5000)
+	require.True(t, q.Allow("A", LivekitRealm, addr))
+	require.False(t, q.Allow("A", LivekitRealm, udpAddr(t, 5001)))
+
+	// once the allocation is deleted the slot is reclaimed
+	q.OnDeleted(addr, nil, "udp", "A", LivekitRealm)
+	require.True(t, q.Allow("A", LivekitRealm, udpAddr(t, 5001)))
+}
+
+func TestTURNAllocationQuota_RetransmitIsIdempotent(t *testing.T) {
+	q := newTURNAllocationQuota(1)
+
+	addr := udpAddr(t, 5000)
+	// same 5-tuple retried multiple times must consume only one slot
+	require.True(t, q.Allow("A", LivekitRealm, addr))
+	require.True(t, q.Allow("A", LivekitRealm, addr))
+	require.True(t, q.Allow("A", LivekitRealm, addr))
+
+	// a different 5-tuple is still over quota
+	require.False(t, q.Allow("A", LivekitRealm, udpAddr(t, 5001)))
+}
+
+func TestTURNAllocationQuota_ForgetsUserWhenEmpty(t *testing.T) {
+	q := newTURNAllocationQuota(1)
+
+	addr := udpAddr(t, 5000)
+	require.True(t, q.Allow("A", LivekitRealm, addr))
+	q.OnDeleted(addr, nil, "udp", "A", LivekitRealm)
+
+	q.mu.Lock()
+	_, tracked := q.users["A"]
+	q.mu.Unlock()
+	require.False(t, tracked, "user with no active allocations should not be retained")
+}
+
+func TestTURNAllocationQuota_UnconfirmedReservationReclaimed(t *testing.T) {
+	// an Allocate that passes the quota but never produces an allocation (e.g.
+	// relay ports exhausted) must not hold its slot forever
+	q := newTURNAllocationQuota(1)
+	q.reservationTTL = 10 * time.Millisecond
+
+	require.True(t, q.Allow("A", LivekitRealm, udpAddr(t, 5000)))
+	require.False(t, q.Allow("A", LivekitRealm, udpAddr(t, 5001)), "at quota while reservation pending")
+
+	require.Eventually(t, func() bool {
+		return q.Allow("A", LivekitRealm, udpAddr(t, 5002))
+	}, time.Second, 5*time.Millisecond, "slot should be reclaimed after TTL")
+
+	// the reclaimed user should not linger once fully drained
+	q.OnDeleted(udpAddr(t, 5002), nil, "udp", "A", LivekitRealm)
+	q.mu.Lock()
+	_, tracked := q.users["A"]
+	q.mu.Unlock()
+	require.False(t, tracked)
+}
+
+func TestTURNAllocationQuota_ConfirmedReservationSurvivesTTL(t *testing.T) {
+	// a confirmed (live) allocation must never be reclaimed by the pending timer
+	q := newTURNAllocationQuota(1)
+	q.reservationTTL = 10 * time.Millisecond
+
+	addr := udpAddr(t, 5000)
+	require.True(t, q.Allow("A", LivekitRealm, addr))
+	q.OnCreated(addr, nil, "udp", "A", LivekitRealm, nil, 0)
+
+	time.Sleep(30 * time.Millisecond) // past the TTL
+
+	require.False(t, q.Allow("A", LivekitRealm, udpAddr(t, 5001)), "confirmed allocation must still hold its slot")
+
+	q.OnDeleted(addr, nil, "udp", "A", LivekitRealm)
+	require.True(t, q.Allow("A", LivekitRealm, udpAddr(t, 5001)))
+}
+
+func TestTURNAllocationQuota_StaleReclaimDoesNotEvictReplacement(t *testing.T) {
+	// a reclaim timer that fires after its slot was released must not delete a
+	// fresh reservation created for the same participant + client address
+	q := newTURNAllocationQuota(1)
+	q.reservationTTL = time.Hour // keep real timers from firing during the test
+
+	addr := udpAddr(t, 5000)
+	key := srcAddrKey(addr)
+
+	require.True(t, q.Allow("A", LivekitRealm, addr))
+	q.mu.Lock()
+	stale := q.users["A"][key]
+	q.mu.Unlock()
+
+	// the original allocation is torn down, then the same 5-tuple is reused and
+	// its replacement allocation is confirmed
+	q.OnDeleted(addr, nil, "udp", "A", LivekitRealm)
+	require.True(t, q.Allow("A", LivekitRealm, addr))
+	q.OnCreated(addr, nil, "udp", "A", LivekitRealm, nil, 0)
+
+	// the original slot's timer fires late: it must be a no-op
+	q.reclaimPending("A", key, stale)
+
+	q.mu.Lock()
+	_, live := q.users["A"][key]
+	q.mu.Unlock()
+	require.True(t, live, "replacement slot must survive a stale reclaim")
+	require.False(t, q.Allow("A", LivekitRealm, udpAddr(t, 5001)), "replacement must still count against quota")
+}
+
+// TestTURNAllocationQuota_ConcurrentAllocatesRespectLimit is the core security
+// property: a burst of concurrent Allocate requests reusing one credential must
+// not be able to race past the limit.
+func TestTURNAllocationQuota_ConcurrentAllocatesRespectLimit(t *testing.T) {
+	const limit = 4
+	q := newTURNAllocationQuota(limit)
+
+	var granted atomic.Int32
+	var wg sync.WaitGroup
+	for i := 0; i < 200; i++ {
+		wg.Add(1)
+		go func(port int) {
+			defer wg.Done()
+			if q.Allow("attacker", LivekitRealm, udpAddr(t, port)) {
+				granted.Add(1)
+			}
+		}(7000 + i)
+	}
+	wg.Wait()
+
+	require.Equal(t, int32(limit), granted.Load(), "concurrent allocations must not exceed the quota")
+}
+
+func TestTURNAllocationQuota_SrcAddrKeyDistinguishesNetwork(t *testing.T) {
+	udp := &net.UDPAddr{IP: net.IPv4(10, 0, 0, 1), Port: 5000}
+	tcp := &net.TCPAddr{IP: net.IPv4(10, 0, 0, 1), Port: 5000}
+	require.NotEqual(t, srcAddrKey(udp), srcAddrKey(tcp))
+	require.Equal(t, "", srcAddrKey(nil))
+}
+
+func BenchmarkTURNAllocationQuota_Allow(b *testing.B) {
+	q := newTURNAllocationQuota(1000000)
+	addrs := make([]net.Addr, 0, 64)
+	for i := 0; i < 64; i++ {
+		addrs = append(addrs, &net.UDPAddr{IP: net.IPv4(10, 0, 0, 1), Port: 5000 + i})
+	}
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		q.Allow(fmt.Sprintf("u%d", i%128), LivekitRealm, addrs[i%len(addrs)])
+	}
+}