automatic configuration of TURN, with per-room credentials. version 0.6.0

cmd/server/main.go

@@ -267,8 +267,8 @@ func printPorts(c *cli.Context) error {
 
 	if conf.TURN.Enabled {
 		udpPorts = append(udpPorts, fmt.Sprintf("%d-%d", conf.TURN.PortRangeStart, conf.TURN.PortRangeEnd))
-		udpPorts = append(udpPorts, strconv.Itoa(conf.TURN.ListenPort))
-		tcpPorts = append(tcpPorts, strconv.Itoa(conf.TURN.ListenPort))
+		udpPorts = append(udpPorts, strconv.Itoa(conf.TURN.TCPPort))
+		tcpPorts = append(tcpPorts, strconv.Itoa(conf.TURN.TCPPort))
 	}
 
 	fmt.Println("TCP Ports")

deploy/aws-ecs/config.tf

@@ -5,6 +5,13 @@ locals {
       port_range_start = var.udp_port_start
       port_range_end = var.udp_port_end
     }
+    turn = {
+      enabled = var.turn_enabled
+      tcp_port = var.turn_tcp_port
+      udp_port = var.turn_udp_port
+      port_range_start = var.turn_port_start
+      port_range_end = var.turn_port_end
+    }
     development = true
     keys = var.api_keys
     redis = {
@@ -12,15 +19,22 @@ locals {
     }
   }
 
-  port_mapping = concat([{
+  // mapping contains only the main listening ports
+  // other UDP ports don't have to be mapped, due to
+  port_mapping = [
+    {
       containerPort = var.http_port
       protocol = "tcp"
-    }], [
-    for p in range(var.udp_port_start, var.udp_port_end): {
-      containerPort = p
+    },
+    {
+      containerPort = var.turn_tcp_port
+      protocol = "tcp"
+    },
+    {
+      containerPort = var.turn_udp_port
       protocol = "udp"
-    }
-  ])
+    },
+  ]
 
   task_config = [{
     name = "livekit"

deploy/aws-ecs/ecs_task.tf

@@ -23,13 +23,17 @@ resource "aws_ecs_service" "livekit" {
     field = "instanceId"
   }
 
-  // load balancer for TCP port
+  // load balancer for HTTP port
   load_balancer {
-    target_group_arn = aws_lb_target_group.main.arn
+    target_group_arn = aws_lb_target_group.http.arn
     container_name = "livekit"
     container_port = var.http_port
   }
 
+  depends_on = [
+    aws_lb_listener.http
+  ]
+
   // lifecycle {
   //   ignore_changes = [desired_count]
   // }

deploy/aws-ecs/example.tfvars

@@ -38,5 +38,16 @@ api_keys = {
   "key" = "secret"
 }
 
-udp_port_start = 9000
-udp_port_end = 9100
+# UDP port range for WebRTC, uncomment to override
+// udp_port_start = 9000
+// udp_port_end = 11000
+
+# Use embedded TURN server, defaults true
+// turn_enabled = true
+// turn_tcp_port = 3478
+// turn_udp_port = 3479
+
+# UDP port range for embedded TURN server
+// turn_port_start = 11001
+// turn_port_end = 13000
+

deploy/aws-ecs/load_balancer.tf

@@ -1,23 +1,21 @@
 
 
 // configure target group
-resource "aws_lb_target_group" "main" {
-  name = "livekit-${var.name}"
+resource "aws_lb_target_group" "http" {
+  name = "livekit-${var.name}-http"
   port = 80
   protocol = "HTTP"
   vpc_id = data.aws_vpc.main.id
 }
 
 resource "aws_lb" "main" {
-  name = "livekit-${var.name}"
+  name = "livekit-${var.name}-http"
   internal           = false
   load_balancer_type = "application"
   security_groups    = [aws_security_group.lb.id]
   subnets = var.subnet_ids
 }
 
-// TODO: HTTPS
-
 resource "aws_lb_listener" "http" {
   load_balancer_arn = aws_lb.main.arn
   port              = "80"
@@ -26,6 +24,6 @@ resource "aws_lb_listener" "http" {
 
   default_action {
     type             = "forward"
-    target_group_arn = aws_lb_target_group.main.arn
+    target_group_arn = aws_lb_target_group.http.arn
   }
 }

deploy/aws-ecs/main.tf

@@ -77,7 +77,7 @@ variable "udp_port_start" {
 
 variable "udp_port_end" {
   type = number
-  default = 9100
+  default = 11000
 }
 
 variable "api_keys" {
@@ -89,6 +89,31 @@ variable "redis_address" {
   default = ""
 }
 
+variable "turn_enabled" {
+  type = bool
+  default = true
+}
+
+variable "turn_tcp_port" {
+  type = number
+  default = 3478
+}
+
+variable "turn_udp_port" {
+  type = number
+  default = 3479
+}
+
+variable "turn_port_start" {
+  type = number
+  default = 12000
+}
+
+variable "turn_port_end" {
+  type = number
+  default = 14000
+}
+
 output "livekit_lb" {
   value = aws_lb.main.dns_name
 }

deploy/aws-ecs/networking.tf

@@ -15,6 +15,32 @@ resource "aws_security_group" "main" {
     cidr_blocks = ["0.0.0.0/0"]
   }
 
+  ingress {
+    description = "UDP port for TURN"
+    from_port   = var.turn_port_start
+    to_port     = var.turn_port_end
+    protocol    = "udp"
+    cidr_blocks = ["0.0.0.0/0"]
+  }
+
+  // for TURN server
+  ingress {
+    description = "TURN TCP"
+    from_port   = var.turn_tcp_port
+    to_port     = var.turn_tcp_port
+    protocol    = "tcp"
+    cidr_blocks = ["0.0.0.0/0"]
+  }
+
+  // for TURN server
+  ingress {
+    description = "TURN UDP"
+    from_port   = var.turn_udp_port
+    to_port     = var.turn_udp_port
+    protocol    = "udp"
+    cidr_blocks = ["0.0.0.0/0"]
+  }
+
   ingress {
     description = "internal traffic"
     from_port   = 0

pkg/config/config.go

@@ -52,7 +52,8 @@ type RedisConfig struct {
 
 type TURNConfig struct {
 	Enabled        bool   `yaml:"enabled"`
-	ListenPort     int    `yaml:"listen_port"`
+	TCPPort        int    `yaml:"tcp_port"`
+	UDPPort        int    `yaml:"udp_port"`
 	PortRangeStart uint16 `yaml:"port_range_start"`
 	PortRangeEnd   uint16 `yaml:"port_range_end"`
 }
@@ -79,9 +80,10 @@ func NewConfig(confString string) (*Config, error) {
 		Redis: RedisConfig{},
 		TURN: TURNConfig{
 			Enabled:        false,
-			ListenPort:     3478,
+			TCPPort:        3478,
+			UDPPort:        3478,
 			PortRangeStart: 12000,
-			PortRangeEnd:   16000,
+			PortRangeEnd:   14000,
 		},
 		Keys: map[string]string{},
 	}

pkg/rtc/participant.go

@@ -375,7 +375,7 @@ func (p *ParticipantImpl) RemoveSubscriber(participantId string) {
 }
 
 // signal connection methods
-func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.Room, otherParticipants []types.Participant) error {
+func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.Room, otherParticipants []types.Participant, iceServers []*livekit.ICEServer) error {
 	// send Join response
 	return p.writeMessage(&livekit.SignalResponse{
 		Message: &livekit.SignalResponse_Join{
@@ -384,6 +384,7 @@ func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.Room, otherParticip
 				Participant:       p.ToProto(),
 				OtherParticipants: ToProtoParticipants(otherParticipants),
 				ServerVersion:     version.Version,
+				IceServers:        iceServers,
 			},
 		},
 	})

pkg/rtc/room.go

@@ -18,8 +18,9 @@ const (
 
 type Room struct {
 	livekit.Room
-	config WebRTCConfig
-	lock   sync.RWMutex
+	config     WebRTCConfig
+	iceServers []*livekit.ICEServer
+	lock       sync.RWMutex
 	// map of identity -> Participant
 	participants map[string]types.Participant
 	// time the first participant joined the room
@@ -36,10 +37,11 @@ type Room struct {
 	onClose              func()
 }
 
-func NewRoom(room *livekit.Room, config WebRTCConfig, audioUpdateInterval uint32) *Room {
+func NewRoom(room *livekit.Room, config WebRTCConfig, iceServers []*livekit.ICEServer, audioUpdateInterval uint32) *Room {
 	r := &Room{
 		Room:                *room,
 		config:              config,
+		iceServers:          iceServers,
 		audioUpdateInterval: audioUpdateInterval,
 		lock:                sync.RWMutex{},
 		participants:        make(map[string]types.Participant),
@@ -179,7 +181,7 @@ func (r *Room) Join(participant types.Participant) error {
 		r.onParticipantChanged(participant)
 	}
 
-	return participant.SendJoinResponse(&r.Room, otherParticipants)
+	return participant.SendJoinResponse(&r.Room, otherParticipants, r.iceServers)
 }
 
 func (r *Room) RemoveParticipant(identity string) {

pkg/rtc/room_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/livekit/livekit-server/pkg/logger"
 	"github.com/livekit/livekit-server/pkg/rtc"
@@ -62,10 +63,11 @@ func TestRoomJoin(t *testing.T) {
 		rm.Join(pNew)
 
 		// expect new participant to get a JoinReply
-		info, participants := pNew.SendJoinResponseArgsForCall(0)
+		info, participants, iceServers := pNew.SendJoinResponseArgsForCall(0)
 		assert.Equal(t, info.Sid, rm.Sid)
 		assert.Len(t, participants, numParticipants)
 		assert.Len(t, rm.GetParticipants(), numParticipants+1)
+		require.NotEmpty(t, iceServers)
 	})
 
 	t.Run("subscribe to existing channels upon join", func(t *testing.T) {
@@ -265,6 +267,13 @@ func newRoomWithParticipants(t *testing.T, num int) *rtc.Room {
 	rm := rtc.NewRoom(
 		&livekit.Room{Name: "room"},
 		rtc.WebRTCConfig{},
+		[]*livekit.ICEServer{
+			{
+				Urls: []string{
+					"stun:stun.l.google.com:19302",
+				},
+			},
+		},
 		audioUpdateInterval,
 	)
 	for i := 0; i < num; i++ {

pkg/rtc/types/interfaces.go

@@ -41,7 +41,7 @@ type Participant interface {
 	AddICECandidate(candidate webrtc.ICECandidateInit, target livekit.SignalTarget) error
 	AddSubscriber(op Participant) error
 	RemoveSubscriber(peerId string)
-	SendJoinResponse(info *livekit.Room, otherParticipants []Participant) error
+	SendJoinResponse(info *livekit.Room, otherParticipants []Participant, iceServers []*livekit.ICEServer) error
 	SendParticipantUpdate(participants []*livekit.ParticipantInfo) error
 	SendActiveSpeakers(speakers []*livekit.SpeakerInfo) error
 	SetTrackMuted(trackId string, muted bool)

pkg/rtc/types/typesfakes/fake_participant.go

@@ -206,11 +206,12 @@ type FakeParticipant struct {
 	sendActiveSpeakersReturnsOnCall map[int]struct {
 		result1 error
 	}
-	SendJoinResponseStub        func(*livekit.Room, []types.Participant) error
+	SendJoinResponseStub        func(*livekit.Room, []types.Participant, []*livekit.ICEServer) error
 	sendJoinResponseMutex       sync.RWMutex
 	sendJoinResponseArgsForCall []struct {
 		arg1 *livekit.Room
 		arg2 []types.Participant
+		arg3 []*livekit.ICEServer
 	}
 	sendJoinResponseReturns struct {
 		result1 error
@@ -1353,24 +1354,30 @@ func (fake *FakeParticipant) SendActiveSpeakersReturnsOnCall(i int, result1 erro
 	}{result1}
 }
 
-func (fake *FakeParticipant) SendJoinResponse(arg1 *livekit.Room, arg2 []types.Participant) error {
+func (fake *FakeParticipant) SendJoinResponse(arg1 *livekit.Room, arg2 []types.Participant, arg3 []*livekit.ICEServer) error {
 	var arg2Copy []types.Participant
 	if arg2 != nil {
 		arg2Copy = make([]types.Participant, len(arg2))
 		copy(arg2Copy, arg2)
 	}
+	var arg3Copy []*livekit.ICEServer
+	if arg3 != nil {
+		arg3Copy = make([]*livekit.ICEServer, len(arg3))
+		copy(arg3Copy, arg3)
+	}
 	fake.sendJoinResponseMutex.Lock()
 	ret, specificReturn := fake.sendJoinResponseReturnsOnCall[len(fake.sendJoinResponseArgsForCall)]
 	fake.sendJoinResponseArgsForCall = append(fake.sendJoinResponseArgsForCall, struct {
 		arg1 *livekit.Room
 		arg2 []types.Participant
-	}{arg1, arg2Copy})
+		arg3 []*livekit.ICEServer
+	}{arg1, arg2Copy, arg3Copy})
 	stub := fake.SendJoinResponseStub
 	fakeReturns := fake.sendJoinResponseReturns
-	fake.recordInvocation("SendJoinResponse", []interface{}{arg1, arg2Copy})
+	fake.recordInvocation("SendJoinResponse", []interface{}{arg1, arg2Copy, arg3Copy})
 	fake.sendJoinResponseMutex.Unlock()
 	if stub != nil {
-		return stub(arg1, arg2)
+		return stub(arg1, arg2, arg3)
 	}
 	if specificReturn {
 		return ret.result1
@@ -1384,17 +1391,17 @@ func (fake *FakeParticipant) SendJoinResponseCallCount() int {
 	return len(fake.sendJoinResponseArgsForCall)
 }
 
-func (fake *FakeParticipant) SendJoinResponseCalls(stub func(*livekit.Room, []types.Participant) error) {
+func (fake *FakeParticipant) SendJoinResponseCalls(stub func(*livekit.Room, []types.Participant, []*livekit.ICEServer) error) {
 	fake.sendJoinResponseMutex.Lock()
 	defer fake.sendJoinResponseMutex.Unlock()
 	fake.SendJoinResponseStub = stub
 }
 
-func (fake *FakeParticipant) SendJoinResponseArgsForCall(i int) (*livekit.Room, []types.Participant) {
+func (fake *FakeParticipant) SendJoinResponseArgsForCall(i int) (*livekit.Room, []types.Participant, []*livekit.ICEServer) {
 	fake.sendJoinResponseMutex.RLock()
 	defer fake.sendJoinResponseMutex.RUnlock()
 	argsForCall := fake.sendJoinResponseArgsForCall[i]
-	return argsForCall.arg1, argsForCall.arg2
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
 }
 
 func (fake *FakeParticipant) SendJoinResponseReturns(result1 error) {

pkg/service/roommanager.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"encoding/json"
+	"fmt"
 	"sync"
 	"time"
 
@@ -28,13 +29,13 @@ type RoomManager struct {
 	selector    routing.NodeSelector
 	router      routing.Router
 	currentNode routing.LocalNode
-	config      *rtc.WebRTCConfig
-	audioConfig config.AudioConfig
+	rtcConfig   *rtc.WebRTCConfig
+	config      *config.Config
 	rooms       map[string]*rtc.Room
 }
 
-func NewRoomManager(rp RoomStore, router routing.Router, currentNode routing.LocalNode, selector routing.NodeSelector, config *config.Config) (*RoomManager, error) {
-	rtcConf, err := rtc.NewWebRTCConfig(&config.RTC, currentNode.Ip)
+func NewRoomManager(rp RoomStore, router routing.Router, currentNode routing.LocalNode, selector routing.NodeSelector, conf *config.Config) (*RoomManager, error) {
+	rtcConf, err := rtc.NewWebRTCConfig(&conf.RTC, currentNode.Ip)
 	if err != nil {
 		return nil, err
 	}
@@ -42,8 +43,8 @@ func NewRoomManager(rp RoomStore, router routing.Router, currentNode routing.Loc
 	return &RoomManager{
 		lock:        sync.RWMutex{},
 		roomStore:   rp,
-		config:      rtcConf,
-		audioConfig: config.Audio,
+		rtcConfig:   rtcConf,
+		config:      conf,
 		router:      router,
 		selector:    selector,
 		currentNode: currentNode,
@@ -209,7 +210,7 @@ func (r *RoomManager) StartSession(roomName, identity, metadata string, reconnec
 		"num_participants", len(room.GetParticipants()),
 	)
 
-	participant, err = rtc.NewParticipant(identity, r.config, responseSink, r.audioConfig)
+	participant, err = rtc.NewParticipant(identity, r.rtcConfig, responseSink, r.config.Audio)
 	if err != nil {
 		logger.Errorw("could not create participant", "error", err)
 		return
@@ -246,7 +247,8 @@ func (r *RoomManager) getOrCreateRoom(roomName string) (*rtc.Room, error) {
 		return nil, err
 	}
 
-	room = rtc.NewRoom(ri, *r.config, r.audioConfig.UpdateInterval)
+	// construct ice servers
+	room = rtc.NewRoom(ri, *r.rtcConfig, r.iceServersForRoom(ri), r.config.Audio.UpdateInterval)
 	room.OnClose(func() {
 		if err := r.DeleteRoom(roomName); err != nil {
 			logger.Errorw("could not delete room", "error", err)
@@ -373,3 +375,31 @@ func (r *RoomManager) handleRTCMessage(roomName, identity string, msg *livekit.R
 		participant.SetTrackMuted(rm.MuteTrack.TrackSid, rm.MuteTrack.Muted)
 	}
 }
+
+func (r *RoomManager) iceServersForRoom(ri *livekit.Room) []*livekit.ICEServer {
+	var iceServers []*livekit.ICEServer
+
+	if len(r.rtcConfig.Configuration.ICEServers) > 0 {
+		iceServers = append(iceServers, &livekit.ICEServer{
+			Urls: r.rtcConfig.Configuration.ICEServers[0].URLs,
+		})
+	}
+	if r.config.TURN.Enabled {
+		if r.config.TURN.TCPPort > 0 {
+			iceServers = append(iceServers, &livekit.ICEServer{
+				Urls:       []string{fmt.Sprintf("turn:%s:%d?transport=tcp", r.currentNode.Ip, r.config.TURN.TCPPort)},
+				Username:   ri.Name,
+				Credential: ri.TurnPassword,
+			})
+		}
+
+		if r.config.TURN.UDPPort > 0 {
+			iceServers = append(iceServers, &livekit.ICEServer{
+				Urls:       []string{fmt.Sprintf("turn:%s:%d?transport=udp", r.currentNode.Ip, r.config.TURN.UDPPort)},
+				Username:   ri.Name,
+				Credential: ri.TurnPassword,
+			})
+		}
+	}
+	return iceServers
+}

pkg/service/turn.go

@@ -28,42 +28,47 @@ func NewTurnServer(conf *config.Config, roomStore RoomStore, node routing.LocalN
 		AuthHandler: newTurnAuthHandler(roomStore),
 	}
 
-	tcpListener, err := net.Listen("tcp4", "0.0.0.0:"+strconv.Itoa(turnConf.ListenPort))
-	if err != nil {
-		return nil, errors.Wrap(err, "could not listen on TURN TCP port")
-	}
-	serverConfig.ListenerConfigs = []turn.ListenerConfig{
-		{
-			Listener: tcpListener,
-			RelayAddressGenerator: &turn.RelayAddressGeneratorPortRange{
-				RelayAddress: net.ParseIP(node.Ip),
-				Address:      "0.0.0.0",
-				MinPort:      turnConf.PortRangeStart,
-				MaxPort:      turnConf.PortRangeEnd,
-				MaxRetries:   allocateRetries,
+	if turnConf.TCPPort > 0 {
+		tcpListener, err := net.Listen("tcp4", "0.0.0.0:"+strconv.Itoa(turnConf.TCPPort))
+		if err != nil {
+			return nil, errors.Wrap(err, "could not listen on TURN TCP port")
+		}
+		serverConfig.ListenerConfigs = []turn.ListenerConfig{
+			{
+				Listener: tcpListener,
+				RelayAddressGenerator: &turn.RelayAddressGeneratorPortRange{
+					RelayAddress: net.ParseIP(node.Ip),
+					Address:      "0.0.0.0",
+					MinPort:      turnConf.PortRangeStart,
+					MaxPort:      turnConf.PortRangeEnd,
+					MaxRetries:   allocateRetries,
+				},
 			},
-		},
+		}
 	}
 
-	udpListener, err := net.ListenPacket("udp4", "0.0.0.0:"+strconv.Itoa(turnConf.ListenPort))
-	if err != nil {
-		return nil, errors.Wrap(err, "could not listen on TURN UDP port")
-	}
-	serverConfig.PacketConnConfigs = []turn.PacketConnConfig{
-		{
-			PacketConn: udpListener,
-			RelayAddressGenerator: &turn.RelayAddressGeneratorPortRange{
-				RelayAddress: net.ParseIP(node.Ip), // Claim that we are listening on IP passed by user (This should be your Public IP)
-				Address:      "0.0.0.0",            // But actually be listening on every interface
-				MinPort:      turnConf.PortRangeStart,
-				MaxPort:      turnConf.PortRangeEnd,
-				MaxRetries:   allocateRetries,
+	if turnConf.UDPPort > 0 {
+		udpListener, err := net.ListenPacket("udp4", "0.0.0.0:"+strconv.Itoa(turnConf.UDPPort))
+		if err != nil {
+			return nil, errors.Wrap(err, "could not listen on TURN UDP port")
+		}
+		serverConfig.PacketConnConfigs = []turn.PacketConnConfig{
+			{
+				PacketConn: udpListener,
+				RelayAddressGenerator: &turn.RelayAddressGeneratorPortRange{
+					RelayAddress: net.ParseIP(node.Ip), // Claim that we are listening on IP passed by user (This should be your Public IP)
+					Address:      "0.0.0.0",            // But actually be listening on every interface
+					MinPort:      turnConf.PortRangeStart,
+					MaxPort:      turnConf.PortRangeEnd,
+					MaxRetries:   allocateRetries,
+				},
 			},
-		},
+		}
 	}
 
 	logger.Infow("Starting TURN server",
-		"port", turnConf.ListenPort,
+		"TCP port", turnConf.TCPPort,
+		"UDP port", turnConf.UDPPort,
 		"portRange", fmt.Sprintf("%d-%d", turnConf.PortRangeStart, turnConf.PortRangeEnd))
 	return turn.NewServer(serverConfig)
 }

proto/rtc.proto

@@ -72,6 +72,7 @@ message JoinResponse {
   ParticipantInfo participant = 2;
   repeated ParticipantInfo other_participants = 3;
   string server_version = 4;
+  repeated ICEServer ice_servers = 5;
 }
 
 message TrackPublishedResponse {
@@ -117,4 +118,10 @@ message UpdateTrackSettings {
   repeated string track_sids = 1;
   bool mute = 3;
   VideoQuality quality = 4;
-}
+}
+
+message ICEServer {
+  repeated string urls = 1;
+  string username = 2;
+  string credential = 3;
+}

test/turn_test.go

@@ -47,15 +47,15 @@ func TestTurnServer(t *testing.T) {
 	require.NoError(t, roomStore.CreateRoom(rm))
 
 	turnConf := &turn.ClientConfig{
-		STUNServerAddr: fmt.Sprintf("localhost:%d", conf.TURN.ListenPort),
-		TURNServerAddr: fmt.Sprintf("%s:%d", currentNode.Ip, conf.TURN.ListenPort),
+		STUNServerAddr: fmt.Sprintf("localhost:%d", conf.TURN.UDPPort),
+		TURNServerAddr: fmt.Sprintf("%s:%d", currentNode.Ip, conf.TURN.UDPPort),
 		Username:       rm.Name,
 		Password:       rm.TurnPassword,
 		Realm:          "livekit",
 	}
 
 	t.Run("TURN works over TCP", func(t *testing.T) {
-		conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", conf.TURN.ListenPort))
+		conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", conf.TURN.TCPPort))
 		require.NoError(t, err)
 
 		tc := *turnConf

version/version.go

@@ -1,3 +1,3 @@
 package version
 
-const Version = "0.5.7"
+const Version = "0.6.0"