From 9a5db132eb9586b5df785f35eda6c562a87cbfeb Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Mon, 6 May 2024 17:25:18 -0700 Subject: [PATCH] add room/participant name limit (#2704) * add room/participant name limit * defaults * simplify * omitempty * handle 0 config * fix race * unlock * tidy --- pkg/config/config.go | 26 ++++++++------ pkg/service/errors.go | 40 ++++++++++++---------- pkg/service/roomservice.go | 5 +++ pkg/service/rtcservice.go | 6 ++++ pkg/service/wsprotocol.go | 6 ++-- pkg/sfu/streamallocator/streamallocator.go | 7 ++-- pkg/telemetry/telemetryservice.go | 25 +++++--------- 7 files changed, 63 insertions(+), 52 deletions(-) diff --git a/pkg/config/config.go b/pkg/config/config.go index 78cd0f1e1..67006eb33 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -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", diff --git a/pkg/service/errors.go b/pkg/service/errors.go index 9a77f5fde..26f24389d 100644 --- a/pkg/service/errors.go +++ b/pkg/service/errors.go @@ -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") ) diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 319e06cb5..54ab2c1c9 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -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") diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index efe314c5a..c20db4302 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -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 != "" { diff --git a/pkg/service/wsprotocol.go b/pkg/service/wsprotocol.go index 5ac7c744f..d94ce3f90 100644 --- a/pkg/service/wsprotocol.go +++ b/pkg/service/wsprotocol.go @@ -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 diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index e68d51717..87092b434 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -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) } } diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index e20e2a543..8af2fb6c4 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -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() }