From 223587e140390a8d0ffb6161ba8026c25f6ccd9c Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 11 Aug 2026 23:12:07 +0530 Subject: [PATCH] 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) * 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) * 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) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- pkg/config/config.go | 19 +++- pkg/service/turn.go | 9 ++ pkg/service/turnquota.go | 187 +++++++++++++++++++++++++++++++ pkg/service/turnquota_test.go | 203 ++++++++++++++++++++++++++++++++++ 4 files changed, 415 insertions(+), 3 deletions(-) create mode 100644 pkg/service/turnquota.go create mode 100644 pkg/service/turnquota_test.go diff --git a/pkg/config/config.go b/pkg/config/config.go index 0eb3ec747..2eb4445a5 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -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", diff --git a/pkg/service/turn.go b/pkg/service/turn.go index c476aeac1..0b3834c91 100644 --- a/pkg/service/turn.go +++ b/pkg/service/turn.go @@ -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 diff --git a/pkg/service/turnquota.go b/pkg/service/turnquota.go new file mode 100644 index 000000000..a260b77a2 --- /dev/null +++ b/pkg/service/turnquota.go @@ -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, + } +} diff --git a/pkg/service/turnquota_test.go b/pkg/service/turnquota_test.go new file mode 100644 index 000000000..8ab527e42 --- /dev/null +++ b/pkg/service/turnquota_test.go @@ -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)]) + } +}