dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-31 20:19:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

add room/participant name limit (#2704)

Browse Source 
* add room/participant name limit

* defaults

* simplify

* omitempty

* handle 0 config

* fix race

* unlock

* tidy
This commit is contained in:
Paul Wells
2024-05-06 17:25:18 -07:00
committed by GitHub
parent 65f06da1ca
commit 9a5db132eb
7 changed files with 63 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/config/config.go
+15 -11
View File
@@ -229,15 +229,17 @@ type VideoConfig struct {
type RoomConfig struct {
// enable rooms to be automatically created
AutoCreate bool `yaml:"auto_create,omitempty"`
EnabledCodecs []CodecSpec `yaml:"enabled_codecs,omitempty"`
MaxParticipants uint32 `yaml:"max_participants,omitempty"`
EmptyTimeout uint32 `yaml:"empty_timeout,omitempty"`
DepartureTimeout uint32 `yaml:"departure_timeout,omitempty"`
EnableRemoteUnmute bool `yaml:"enable_remote_unmute,omitempty"`
MaxMetadataSize uint32 `yaml:"max_metadata_size,omitempty"`
PlayoutDelay PlayoutDelayConfig `yaml:"playout_delay,omitempty"`
SyncStreams bool `yaml:"sync_streams,omitempty"`
AutoCreate bool `yaml:"auto_create,omitempty"`
EnabledCodecs []CodecSpec `yaml:"enabled_codecs,omitempty"`
MaxParticipants uint32 `yaml:"max_participants,omitempty"`
EmptyTimeout uint32 `yaml:"empty_timeout,omitempty"`
DepartureTimeout uint32 `yaml:"departure_timeout,omitempty"`
EnableRemoteUnmute bool `yaml:"enable_remote_unmute,omitempty"`
MaxMetadataSize uint32 `yaml:"max_metadata_size,omitempty"`
PlayoutDelay PlayoutDelayConfig `yaml:"playout_delay,omitempty"`
SyncStreams bool `yaml:"sync_streams,omitempty"`
MaxRoomNameLength int `yaml:"max_room_name_length,omitempty"`
MaxParticipantIdentityLength int `yaml:"max_participant_identity_length,omitempty"`
}
type CodecSpec struct {
@@ -484,8 +486,10 @@ var DefaultConfig = Config{
{Mime: webrtc.MimeTypeVP9},
{Mime: webrtc.MimeTypeAV1},
},
EmptyTimeout: 5 * 60,
DepartureTimeout: 20,
EmptyTimeout: 5 * 60,
DepartureTimeout: 20,
MaxRoomNameLength: 256,
MaxParticipantIdentityLength: 256,
},
Logging: LoggingConfig{
PionLevel: "error",
pkg/service/errors.go
+21 -19
View File
@@ -19,23 +19,25 @@ import (
)
var (
ErrEgressNotFound = psrpc.NewErrorf(psrpc.NotFound, "egress does not exist")
ErrEgressNotConnected = psrpc.NewErrorf(psrpc.Internal, "egress not connected (redis required)")
ErrIdentityEmpty = psrpc.NewErrorf(psrpc.InvalidArgument, "identity cannot be empty")
ErrIngressNotConnected = psrpc.NewErrorf(psrpc.Internal, "ingress not connected (redis required)")
ErrIngressNotFound = psrpc.NewErrorf(psrpc.NotFound, "ingress does not exist")
ErrIngressNonReusable = psrpc.NewErrorf(psrpc.InvalidArgument, "ingress is not reusable and cannot be modified")
ErrMetadataExceedsLimits = psrpc.NewErrorf(psrpc.InvalidArgument, "metadata size exceeds limits")
ErrOperationFailed = psrpc.NewErrorf(psrpc.Internal, "operation cannot be completed")
ErrParticipantNotFound = psrpc.NewErrorf(psrpc.NotFound, "participant does not exist")
ErrRoomNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested room does not exist")
ErrRoomLockFailed = psrpc.NewErrorf(psrpc.Internal, "could not lock room")
ErrRoomUnlockFailed = psrpc.NewErrorf(psrpc.Internal, "could not unlock room, lock token does not match")
ErrRemoteUnmuteNoteEnabled = psrpc.NewErrorf(psrpc.FailedPrecondition, "remote unmute not enabled")
ErrTrackNotFound = psrpc.NewErrorf(psrpc.NotFound, "track is not found")
ErrWebHookMissingAPIKey = psrpc.NewErrorf(psrpc.InvalidArgument, "api_key is required to use webhooks")
ErrSIPNotConnected = psrpc.NewErrorf(psrpc.Internal, "sip not connected (redis required)")
ErrSIPTrunkNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip trunk does not exist")
ErrSIPDispatchRuleNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip dispatch rule does not exist")
ErrSIPParticipantNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip participant does not exist")
ErrEgressNotFound = psrpc.NewErrorf(psrpc.NotFound, "egress does not exist")
ErrEgressNotConnected = psrpc.NewErrorf(psrpc.Internal, "egress not connected (redis required)")
ErrIdentityEmpty = psrpc.NewErrorf(psrpc.InvalidArgument, "identity cannot be empty")
ErrIngressNotConnected = psrpc.NewErrorf(psrpc.Internal, "ingress not connected (redis required)")
ErrIngressNotFound = psrpc.NewErrorf(psrpc.NotFound, "ingress does not exist")
ErrIngressNonReusable = psrpc.NewErrorf(psrpc.InvalidArgument, "ingress is not reusable and cannot be modified")
ErrMetadataExceedsLimits = psrpc.NewErrorf(psrpc.InvalidArgument, "metadata size exceeds limits")
ErrRoomNameExceedsLimits = psrpc.NewErrorf(psrpc.InvalidArgument, "room name length exceeds limits")
ErrParticipantIdentityExceedsLimits = psrpc.NewErrorf(psrpc.InvalidArgument, "participant identity length exceeds limits")
ErrOperationFailed = psrpc.NewErrorf(psrpc.Internal, "operation cannot be completed")
ErrParticipantNotFound = psrpc.NewErrorf(psrpc.NotFound, "participant does not exist")
ErrRoomNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested room does not exist")
ErrRoomLockFailed = psrpc.NewErrorf(psrpc.Internal, "could not lock room")
ErrRoomUnlockFailed = psrpc.NewErrorf(psrpc.Internal, "could not unlock room, lock token does not match")
ErrRemoteUnmuteNoteEnabled = psrpc.NewErrorf(psrpc.FailedPrecondition, "remote unmute not enabled")
ErrTrackNotFound = psrpc.NewErrorf(psrpc.NotFound, "track is not found")
ErrWebHookMissingAPIKey = psrpc.NewErrorf(psrpc.InvalidArgument, "api_key is required to use webhooks")
ErrSIPNotConnected = psrpc.NewErrorf(psrpc.Internal, "sip not connected (redis required)")
ErrSIPTrunkNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip trunk does not exist")
ErrSIPDispatchRuleNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip dispatch rule does not exist")
ErrSIPParticipantNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip participant does not exist")
)
pkg/service/roomservice.go
+5
View File
@@ -16,6 +16,7 @@ package service
import (
"context"
"fmt"
"strconv"
"github.com/avast/retry-go/v4"
@@ -83,6 +84,10 @@ func (s *RoomService) CreateRoom(ctx context.Context, req *livekit.CreateRoomReq
return nil, ErrEgressNotConnected
}
if limit := s.roomConf.MaxRoomNameLength; limit > 0 && len(req.Name) > limit {
return nil, fmt.Errorf("%w: max length %d", ErrRoomNameExceedsLimits, limit)
}
rm, created, err := s.roomAllocator.CreateRoom(ctx, req)
if err != nil {
err = errors.Wrap(err, "could not create room")
pkg/service/rtcservice.go
+6
View File
@@ -120,6 +120,9 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic
if claims.Identity == "" {
return "", pi, http.StatusBadRequest, ErrIdentityEmpty
}
if limit := s.config.Room.MaxParticipantIdentityLength; limit > 0 && len(claims.Identity) > limit {
return "", pi, http.StatusBadRequest, fmt.Errorf("%w: max length %d", ErrParticipantIdentityExceedsLimits, limit)
}
roomName := livekit.RoomName(r.FormValue("room"))
reconnectParam := r.FormValue("reconnect")
@@ -133,6 +136,9 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic
if onlyName != "" {
roomName = onlyName
}
if limit := s.config.Room.MaxRoomNameLength; limit > 0 && len(roomName) > limit {
return "", pi, http.StatusBadRequest, fmt.Errorf("%w: max length %d", ErrRoomNameExceedsLimits, limit)
}
// this is new connection for existing participant - with publish only permissions
if publishParam != "" {
pkg/service/wsprotocol.go
+4 -2
View File
@@ -162,8 +162,10 @@ func (c *WSSignalConnection) WriteServerMessage(msg *livekit.ServerMessage) (int
}
func (c *WSSignalConnection) pingWorker() {
for {
<-time.After(pingFrequency)
ticker := time.NewTicker(pingFrequency)
defer ticker.Stop()
for range ticker.C {
err := c.conn.WriteControl(websocket.PingMessage, []byte(""), time.Now().Add(pingTimeout))
if err != nil {
return
pkg/sfu/streamallocator/streamallocator.go
+3 -4
View File
@@ -698,13 +698,12 @@ func (s *StreamAllocator) handleSignalProbeClusterDone(event Event) {
func (s *StreamAllocator) handleSignalResume(event Event) {
s.videoTracksMu.Lock()
track := s.videoTracks[event.TrackID]
updated := track != nil && track.SetStreamState(StreamStateActive)
s.videoTracksMu.Unlock()
if track != nil {
if updated {
update := NewStreamStateUpdate()
if track.SetStreamState(StreamStateActive) {
update.HandleStreamingChange(track, StreamStateActive)
}
update.HandleStreamingChange(track, StreamStateActive)
s.maybeSendUpdate(update)
}
}
pkg/telemetry/telemetryservice.go
+9 -16
View File
@@ -129,8 +129,9 @@ func (t *telemetryService) FlushStats() {
t.workersMu.RUnlock()
now := time.Now()
var prev, reapHead, reapTail *StatsWorker
var prev, reap *StatsWorker
for worker != nil {
next := worker.next
if closed := worker.Flush(now); closed {
if prev == nil {
// this worker was at the head of the list
@@ -148,27 +149,19 @@ func (t *telemetryService) FlushStats() {
prev.next = worker.next
}
if reapHead == nil {
reapHead = worker
reapTail = worker
} else {
reapTail.next = worker
reapTail = worker
}
worker.next = reap
reap = worker
} else {
prev = worker
}
worker = worker.next
worker = next
}
if reapHead != nil {
if reap != nil {
t.workersMu.Lock()
for {
delete(t.workers, reapHead.participantID)
if reapHead == reapTail {
break
}
reapHead = reapHead.next
for reap != nil {
delete(t.workers, reap.participantID)
reap = reap.next
}
t.workersMu.Unlock()
}