diff --git a/pkg/service/turn.go b/pkg/service/turn.go index 6e8da2e0e..99c0e41ad 100644 --- a/pkg/service/turn.go +++ b/pkg/service/turn.go @@ -19,6 +19,7 @@ import ( "crypto/tls" "fmt" "net" + "net/netip" "strconv" "strings" "time" @@ -109,6 +110,17 @@ func NewTurnServer(conf *config.Config, authHandler turn.AuthHandler, standalone logValues = append(logValues, "turn.relay_range_end", turnConf.RelayPortRangeEnd) logValues = append(logValues, "turn.per_user_relay_allocation_limit", turnConf.PerUserRelayAllocationLimit) + // embedded TURN shares the host with the SFU, limit relays to this host to its media ports + var localIPs map[netip.Addr]struct{} + var mediaPorts []portRange + if !standalone { + localIPs, err = getTURNLocalIPs(&conf.RTC) + if err != nil { + return nil, errors.Wrap(err, "could not get local IPs for TURN relay policy") + } + mediaPorts = getTURNMediaPorts(&conf.RTC) + } + for _, addr := range turnConf.BindAddresses { var nodeIP string if net.ParseIP(addr).To4() != nil { @@ -127,18 +139,10 @@ func NewTurnServer(conf *config.Config, authHandler turn.AuthHandler, standalone MaxPort: turnConf.RelayPortRangeEnd, MaxRetries: allocateRetries, } - if standalone { - relayAddrGen = telemetry.NewRelayAddressGenerator(relayAddrGen) - } permissionHandler := func(_clientAddr net.Addr, peerIP net.IP) bool { // restricted peer IP is denied by default, unless allowed by the allow list, - if peerIP.IsLoopback() || - peerIP.IsLinkLocalUnicast() || - peerIP.IsLinkLocalMulticast() || - peerIP.IsMulticast() || - peerIP.IsPrivate() || - peerIP.IsUnspecified() { + if isRestrictedPeerIP(peerIP) { allowed := false for _, ipnet := range allowRestrictedPeerCIDRs { if ipnet.Contains(peerIP) { @@ -162,6 +166,13 @@ func NewTurnServer(conf *config.Config, authHandler turn.AuthHandler, standalone return true } + relayAddrGen = newRejectTCPRelayAddressGenerator(relayAddrGen) + if standalone { + relayAddrGen = telemetry.NewRelayAddressGenerator(relayAddrGen) + } else { + relayAddrGen = newTURNRelayPolicy(relayAddrGen, localIPs, mediaPorts) + } + if turnConf.TLSPort > 0 { listener, err := newTURNTCPListener(turnConf, net.JoinHostPort(addr, strconv.Itoa(turnConf.TLSPort))) if err != nil { diff --git a/pkg/service/turnrelay.go b/pkg/service/turnrelay.go new file mode 100644 index 000000000..618e66db0 --- /dev/null +++ b/pkg/service/turnrelay.go @@ -0,0 +1,189 @@ +// 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" + "net/netip" + + "github.com/pion/turn/v5" + "github.com/pkg/errors" + + "github.com/livekit/livekit-server/pkg/config" +) + +var ErrTURNTCPRelayDisabled = errors.New("TURN TCP relay allocations are disabled") + +// restricted ranges not covered by the net.IP predicates +var extraRestrictedPeerPrefixes = []netip.Prefix{ + netip.MustParsePrefix("0.0.0.0/8"), // "this" network + netip.MustParsePrefix("100.64.0.0/10"), // carrier-grade NAT + netip.MustParsePrefix("198.18.0.0/15"), // benchmarking + netip.MustParsePrefix("240.0.0.0/4"), // reserved, includes broadcast + netip.MustParsePrefix("64:ff9b::/96"), // NAT64, can map to restricted IPv4 + netip.MustParsePrefix("64:ff9b:1::/48"), // local-use NAT64 +} + +func isRestrictedPeerIP(ip net.IP) bool { + if ip.IsLoopback() || + ip.IsLinkLocalUnicast() || + ip.IsLinkLocalMulticast() || + ip.IsMulticast() || + ip.IsPrivate() || + ip.IsUnspecified() { + return true + } + + addr, ok := netip.AddrFromSlice(ip) + if !ok { + return true + } + addr = addr.Unmap() + for _, prefix := range extraRestrictedPeerPrefixes { + if prefix.Contains(addr) { + return true + } + } + return false +} + +type portRange struct { + start uint16 + end uint16 +} + +// rejectTCPRelayAddressGenerator refuses RFC 6062 TCP relays, which would let +// a client open TCP connections to any permitted address. +type rejectTCPRelayAddressGenerator struct { + turn.RelayAddressGenerator +} + +func newRejectTCPRelayAddressGenerator(g turn.RelayAddressGenerator) *rejectTCPRelayAddressGenerator { + return &rejectTCPRelayAddressGenerator{RelayAddressGenerator: g} +} + +func (g *rejectTCPRelayAddressGenerator) AllocateListener(turn.AllocateListenerConfig) (net.Listener, net.Addr, error) { + return nil, nil, ErrTURNTCPRelayDisabled +} + +func (g *rejectTCPRelayAddressGenerator) AllocateConn(turn.AllocateConnConfig) (net.Conn, error) { + return nil, ErrTURNTCPRelayDisabled +} + +// turnRelayPolicy enforces what TURN permissions cannot, since they are per IP +// only: relays to this host's own addresses may reach only the RTC media +// ports, because such packets are delivered locally and bypass any firewall +// in front of the host. Used for embedded TURN, which shares the host with +// the SFU. +type turnRelayPolicy struct { + turn.RelayAddressGenerator + + localIPs map[netip.Addr]struct{} + mediaPorts []portRange +} + +func newTURNRelayPolicy(g turn.RelayAddressGenerator, localIPs map[netip.Addr]struct{}, mediaPorts []portRange) *turnRelayPolicy { + return &turnRelayPolicy{ + RelayAddressGenerator: g, + localIPs: localIPs, + mediaPorts: mediaPorts, + } +} + +func (g *turnRelayPolicy) allows(addr net.Addr) bool { + udpAddr, ok := addr.(*net.UDPAddr) + if !ok { + return false + } + ap := udpAddr.AddrPort() + if _, ok := g.localIPs[ap.Addr().Unmap().WithZone("")]; !ok { + return true + } + for _, r := range g.mediaPorts { + if ap.Port() >= r.start && ap.Port() <= r.end { + return true + } + } + return false +} + +func (g *turnRelayPolicy) AllocatePacketConn(c turn.AllocateListenerConfig) (net.PacketConn, net.Addr, error) { + conn, addr, err := g.RelayAddressGenerator.AllocatePacketConn(c) + if err != nil { + return conn, addr, err + } + return &turnRelayPacketConn{PacketConn: conn, policy: g}, addr, nil +} + +// turnRelayPacketConn silently drops relayed packets to peers the policy does +// not allow. +type turnRelayPacketConn struct { + net.PacketConn + policy *turnRelayPolicy +} + +func (c *turnRelayPacketConn) WriteTo(p []byte, addr net.Addr) (int, error) { + if !c.policy.allows(addr) { + return len(p), nil + } + return c.PacketConn.WriteTo(p, addr) +} + +// getTURNLocalIPs returns the addresses that are delivered to this host: its +// interface addresses and the node IPs, which may be 1:1 NAT addresses. +func getTURNLocalIPs(rtcConf *config.RTCConfig) (map[netip.Addr]struct{}, error) { + ips := make(map[netip.Addr]struct{}) + + ifAddrs, err := net.InterfaceAddrs() + if err != nil { + return nil, err + } + for _, ifAddr := range ifAddrs { + ipNet, ok := ifAddr.(*net.IPNet) + if !ok { + continue + } + if addr, ok := netip.AddrFromSlice(ipNet.IP); ok { + ips[addr.Unmap()] = struct{}{} + } + } + + for _, ip := range rtcConf.NodeIP.ToStringSlice() { + addr, err := netip.ParseAddr(ip) + if err != nil { + return nil, err + } + ips[addr.Unmap()] = struct{}{} + } + return ips, nil +} + +func getTURNMediaPorts(rtcConf *config.RTCConfig) []portRange { + if rtcConf.ForceTCP { + return nil + } + + var ports []portRange + if rtcConf.ICEPortRangeStart != 0 && rtcConf.ICEPortRangeEnd != 0 { + ports = append(ports, portRange{start: uint16(rtcConf.ICEPortRangeStart), end: uint16(rtcConf.ICEPortRangeEnd)}) + } else if rtcConf.UDPPort.Valid() { + end := max(rtcConf.UDPPort.End, rtcConf.UDPPort.Start) + ports = append(ports, portRange{start: uint16(rtcConf.UDPPort.Start), end: uint16(end)}) + if rtcConf.UseStunPortAsICE { + ports = append(ports, portRange{start: 3478, end: 3478}) + } + } + return ports +} diff --git a/pkg/service/turnrelay_test.go b/pkg/service/turnrelay_test.go new file mode 100644 index 000000000..ca28890af --- /dev/null +++ b/pkg/service/turnrelay_test.go @@ -0,0 +1,311 @@ +// 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" + "net/netip" + "testing" + "time" + + "github.com/pion/turn/v5" + "github.com/stretchr/testify/require" + + "github.com/livekit/mediatransportutil/pkg/rtcconfig" + + "github.com/livekit/livekit-server/pkg/config" +) + +func TestGetTURNMediaPorts(t *testing.T) { + ports := getTURNMediaPorts(&config.RTCConfig{RTCConfig: rtcconfig.RTCConfig{ + UDPPort: rtcconfig.PortRange{Start: 7882, End: 7885}, + UseStunPortAsICE: true, + }}) + require.Equal(t, []portRange{{7882, 7885}, {3478, 3478}}, ports) + + ports = getTURNMediaPorts(&config.RTCConfig{RTCConfig: rtcconfig.RTCConfig{ + UDPPort: rtcconfig.PortRange{Start: 7882}, + ICEPortRangeStart: 50000, + ICEPortRangeEnd: 60000, + }}) + require.Equal(t, []portRange{{50000, 60000}}, ports) + + require.Empty(t, getTURNMediaPorts(&config.RTCConfig{RTCConfig: rtcconfig.RTCConfig{ + UDPPort: rtcconfig.PortRange{Start: 7882}, + ForceTCP: true, + }})) +} + +func TestGetTURNLocalIPs(t *testing.T) { + ips, err := getTURNLocalIPs(&config.RTCConfig{RTCConfig: rtcconfig.RTCConfig{ + NodeIP: rtcconfig.NodeIP{V4: "203.0.113.1", V6: "2001:db8::1"}, + }}) + require.NoError(t, err) + require.Contains(t, ips, netip.MustParseAddr("127.0.0.1")) + require.Contains(t, ips, netip.MustParseAddr("203.0.113.1")) + require.Contains(t, ips, netip.MustParseAddr("2001:db8::1")) +} + +func TestTURNRelayPolicy_Allows(t *testing.T) { + policy := newTURNRelayPolicy(nil, + map[netip.Addr]struct{}{netip.MustParseAddr("203.0.113.1"): {}}, + []portRange{{7882, 7882}, {50000, 60000}}, + ) + + for _, tc := range []struct { + addr net.Addr + allowed bool + }{ + {&net.UDPAddr{IP: net.ParseIP("203.0.113.1"), Port: 7882}, true}, + {&net.UDPAddr{IP: net.ParseIP("203.0.113.1").To16(), Port: 55000}, true}, + {&net.UDPAddr{IP: net.ParseIP("203.0.113.1"), Port: 22}, false}, + {&net.UDPAddr{IP: net.ParseIP("203.0.113.1"), Port: 443}, false}, + {&net.UDPAddr{IP: net.ParseIP("203.0.113.1"), Port: 60001}, false}, + {&net.UDPAddr{IP: net.ParseIP("198.51.100.7"), Port: 7882}, true}, + {&net.UDPAddr{IP: net.ParseIP("198.51.100.7"), Port: 443}, true}, + {&net.TCPAddr{IP: net.ParseIP("198.51.100.7"), Port: 443}, false}, + } { + require.Equal(t, tc.allowed, policy.allows(tc.addr), "%s", tc.addr) + } +} + +type fakeRelayAddressGenerator struct { + conn net.PacketConn +} + +func (g *fakeRelayAddressGenerator) Validate() error { return nil } + +func (g *fakeRelayAddressGenerator) AllocatePacketConn(turn.AllocateListenerConfig) (net.PacketConn, net.Addr, error) { + return g.conn, g.conn.LocalAddr(), nil +} + +func (g *fakeRelayAddressGenerator) AllocateListener(turn.AllocateListenerConfig) (net.Listener, net.Addr, error) { + panic("must not be called") +} + +func (g *fakeRelayAddressGenerator) AllocateConn(turn.AllocateConnConfig) (net.Conn, error) { + panic("must not be called") +} + +func listenUDP(t *testing.T) net.PacketConn { + t.Helper() + conn, err := net.ListenPacket("udp4", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +func TestTURNRelayPolicy_PacketConn(t *testing.T) { + relay := listenUDP(t) + mediaPeer := listenUDP(t) + otherPeer := listenUDP(t) + + mediaPort := uint16(mediaPeer.LocalAddr().(*net.UDPAddr).Port) + policy := newTURNRelayPolicy( + &fakeRelayAddressGenerator{conn: relay}, + map[netip.Addr]struct{}{netip.MustParseAddr("127.0.0.1"): {}}, + []portRange{{mediaPort, mediaPort}}, + ) + + conn, _, err := policy.AllocatePacketConn(turn.AllocateListenerConfig{UserID: "PA_x"}) + require.NoError(t, err) + + buf := make([]byte, 64) + + // writes to a non-media port on this host are dropped + n, err := conn.WriteTo([]byte("denied"), otherPeer.LocalAddr()) + require.NoError(t, err) + require.Equal(t, len("denied"), n) + require.NoError(t, otherPeer.SetReadDeadline(time.Now().Add(100*time.Millisecond))) + _, _, err = otherPeer.ReadFrom(buf) + require.Error(t, err) + + _, err = conn.WriteTo([]byte("allowed"), mediaPeer.LocalAddr()) + require.NoError(t, err) + require.NoError(t, mediaPeer.SetReadDeadline(time.Now().Add(2*time.Second))) + n, _, err = mediaPeer.ReadFrom(buf) + require.NoError(t, err) + require.Equal(t, "allowed", string(buf[:n])) +} + +func TestRejectTCPRelayAddressGenerator(t *testing.T) { + relay := listenUDP(t) + g := newRejectTCPRelayAddressGenerator(&fakeRelayAddressGenerator{conn: relay}) + + _, _, err := g.AllocateListener(turn.AllocateListenerConfig{}) + require.ErrorIs(t, err, ErrTURNTCPRelayDisabled) + _, err = g.AllocateConn(turn.AllocateConnConfig{}) + require.ErrorIs(t, err, ErrTURNTCPRelayDisabled) + + conn, _, err := g.AllocatePacketConn(turn.AllocateListenerConfig{}) + require.NoError(t, err) + require.Same(t, relay, conn) +} + +// allocateTestTurnRelay starts a TURN server on loopback, with the RTC media +// port set to mediaPort, and returns a relay allocated through it. +func allocateTestTurnRelay(t *testing.T, standalone bool, mediaPort int) net.PacketConn { + t.Helper() + + turnPort := listenUDP(t) + turnAddr := turnPort.LocalAddr().String() + require.NoError(t, turnPort.Close()) + + conf := &config.Config{ + RTC: config.RTCConfig{RTCConfig: rtcconfig.RTCConfig{ + NodeIP: rtcconfig.NodeIP{V4: "127.0.0.1"}, + UDPPort: rtcconfig.PortRange{Start: mediaPort}, + }}, + TURN: config.TURNConfig{ + Enabled: true, + UDPPort: turnPort.LocalAddr().(*net.UDPAddr).Port, + BindAddresses: []string{"127.0.0.1"}, + RelayPortRangeStart: 40000, + RelayPortRangeEnd: 40100, + AllowRestrictedPeerCIDRs: []string{"127.0.0.0/8"}, + }, + } + h := newTestTurnAuthHandler() + server, err := NewTurnServer(conf, h.HandleAuth, standalone) + require.NoError(t, err) + t.Cleanup(func() { _ = server.Close() }) + + username, expiry := h.CreateUsername(turnTestAPIKey, "PA_relay", 300) + password, err := h.CreatePassword(turnTestAPIKey, "PA_relay", expiry) + require.NoError(t, err) + + client, err := turn.NewClient(&turn.ClientConfig{ + STUNServerAddr: turnAddr, + TURNServerAddr: turnAddr, + Username: username, + Password: password, + Realm: LivekitRealm, + Conn: listenUDP(t), + }) + require.NoError(t, err) + t.Cleanup(client.Close) + require.NoError(t, client.Listen()) + + relayConn, err := client.Allocate() + require.NoError(t, err) + t.Cleanup(func() { _ = relayConn.Close() }) + return relayConn +} + +func TestNewTurnServer_RelayPolicy(t *testing.T) { + mediaPeer := listenUDP(t) + otherPeer := listenUDP(t) + relayConn := allocateTestTurnRelay(t, false, mediaPeer.LocalAddr().(*net.UDPAddr).Port) + + buf := make([]byte, 64) + + _, err := relayConn.WriteTo([]byte("denied"), otherPeer.LocalAddr()) + require.NoError(t, err) + _, err = relayConn.WriteTo([]byte("allowed"), mediaPeer.LocalAddr()) + require.NoError(t, err) + + require.NoError(t, mediaPeer.SetReadDeadline(time.Now().Add(2*time.Second))) + n, relayAddr, err := mediaPeer.ReadFrom(buf) + require.NoError(t, err) + require.Equal(t, "allowed", string(buf[:n])) + + require.NoError(t, otherPeer.SetReadDeadline(time.Now().Add(200*time.Millisecond))) + _, _, err = otherPeer.ReadFrom(buf) + require.Error(t, err, "relay to a non-media port on this host must be dropped") + + // the media peer can reply through the relay + _, err = mediaPeer.WriteTo([]byte("reply"), relayAddr) + require.NoError(t, err) + require.NoError(t, relayConn.SetReadDeadline(time.Now().Add(2*time.Second))) + n, from, err := relayConn.ReadFrom(buf) + require.NoError(t, err) + require.Equal(t, "reply", string(buf[:n])) + require.Equal(t, mediaPeer.LocalAddr().String(), from.String()) +} + +func TestNewTurnServer_StandaloneRelayPolicy(t *testing.T) { + mediaPeer := listenUDP(t) + otherPeer := listenUDP(t) + relayConn := allocateTestTurnRelay(t, true, mediaPeer.LocalAddr().(*net.UDPAddr).Port) + + // standalone TURN has no SFU on the host, so ports on this host are not limited + _, err := relayConn.WriteTo([]byte("allowed"), otherPeer.LocalAddr()) + require.NoError(t, err) + + buf := make([]byte, 64) + require.NoError(t, otherPeer.SetReadDeadline(time.Now().Add(2*time.Second))) + n, _, err := otherPeer.ReadFrom(buf) + require.NoError(t, err) + require.Equal(t, "allowed", string(buf[:n])) +} + +func TestNewTurnServer_RejectsTCPRelay(t *testing.T) { + for _, standalone := range []bool{false, true} { + t.Run(fmt.Sprintf("standalone=%v", standalone), func(t *testing.T) { + ln, err := net.Listen("tcp4", "127.0.0.1:0") + require.NoError(t, err) + turnAddr := ln.Addr().String() + require.NoError(t, ln.Close()) + + conf := &config.Config{ + RTC: config.RTCConfig{RTCConfig: rtcconfig.RTCConfig{ + NodeIP: rtcconfig.NodeIP{V4: "127.0.0.1"}, + UDPPort: rtcconfig.PortRange{Start: 7882}, + }}, + TURN: config.TURNConfig{ + Enabled: true, + Domain: "turn.example.com", + TLSPort: ln.Addr().(*net.TCPAddr).Port, + ExternalTLS: true, + BindAddresses: []string{"127.0.0.1"}, + RelayPortRangeStart: 40000, + RelayPortRangeEnd: 40100, + AllowRestrictedPeerCIDRs: []string{"127.0.0.0/8"}, + }, + } + h := newTestTurnAuthHandler() + server, err := NewTurnServer(conf, h.HandleAuth, standalone) + require.NoError(t, err) + t.Cleanup(func() { _ = server.Close() }) + + username, expiry := h.CreateUsername(turnTestAPIKey, "PA_tcp", 300) + password, err := h.CreatePassword(turnTestAPIKey, "PA_tcp", expiry) + require.NoError(t, err) + + tcpConn, err := net.Dial("tcp4", turnAddr) + require.NoError(t, err) + client, err := turn.NewClient(&turn.ClientConfig{ + STUNServerAddr: turnAddr, + TURNServerAddr: turnAddr, + Username: username, + Password: password, + Realm: LivekitRealm, + Conn: turn.NewSTUNConn(tcpConn), + }) + require.NoError(t, err) + t.Cleanup(client.Close) + require.NoError(t, client.Listen()) + + _, err = client.AllocateTCP() + require.Error(t, err, "TCP relay allocation must be refused") + + // a UDP relay over the same TCP connection still works + relayConn, err := client.Allocate() + require.NoError(t, err) + _ = relayConn.Close() + }) + } +}