From e03f75d6a1bc3ae4096cb2a8c94dd25962ba37fb Mon Sep 17 00:00:00 2001 From: David Zhao Date: Fri, 7 Apr 2023 23:47:49 -0700 Subject: [PATCH] Implements source-specific permissions and client-driven metadata updates (#1590) Closes #1565 --- go.mod | 2 +- go.sum | 4 +- pkg/rtc/helper_test.go | 2 +- pkg/rtc/participant.go | 69 ++++----- pkg/rtc/participant_internal_test.go | 26 ++++ pkg/rtc/signalhandler.go | 10 ++ pkg/rtc/types/interfaces.go | 2 +- .../typesfakes/fake_local_participant.go | 139 ++++++++++-------- pkg/service/rtcservice.go | 2 +- 9 files changed, 147 insertions(+), 109 deletions(-) diff --git a/go.mod b/go.mod index 55bdf34be..7c052ce66 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26 - github.com/livekit/protocol v1.5.2-0.20230405195605-927c9ea2b4c6 + github.com/livekit/protocol v1.5.2 github.com/livekit/psrpc v0.2.11-0.20230405191830-d76f71512630 github.com/mackerelio/go-osstat v0.2.4 github.com/magefile/mage v1.14.0 diff --git a/go.sum b/go.sum index c963460f4..59640e93b 100644 --- a/go.sum +++ b/go.sum @@ -235,8 +235,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26 h1:QlQFyMwCDgjyySsrgmrMcVbEBA6KZcyTzvK+z346tUA= github.com/livekit/mediatransportutil v0.0.0-20230326055817-ed569ca13d26/go.mod h1:eDA41kiySZoG+wy4Etsjb3w0jjLx69i/vAmSjG4bteA= -github.com/livekit/protocol v1.5.2-0.20230405195605-927c9ea2b4c6 h1:rvkmoc5s+VJTpShWkY+QWFQ5XhLDMFFyZIZrrr7PgJE= -github.com/livekit/protocol v1.5.2-0.20230405195605-927c9ea2b4c6/go.mod h1:UFgAWejoO4eshaaDe2jynTdQWwSktNO+8Wx19V7bs+o= +github.com/livekit/protocol v1.5.2 h1:mbbkJNxbStvb9sDtB7CFX7NnTObYKFumNU7wWm4UOfY= +github.com/livekit/protocol v1.5.2/go.mod h1:UFgAWejoO4eshaaDe2jynTdQWwSktNO+8Wx19V7bs+o= github.com/livekit/psrpc v0.2.11-0.20230405191830-d76f71512630 h1:Rm5KLZgQxWnTidY+H8MsAV6sk1iiFxeXqPFgSLkMing= github.com/livekit/psrpc v0.2.11-0.20230405191830-d76f71512630/go.mod h1:K0j8f1PgLShR7Lx80KbmwFkDH2BvOnycXGV0OSRURKc= github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs= diff --git a/pkg/rtc/helper_test.go b/pkg/rtc/helper_test.go index dff552233..dc97f13a0 100644 --- a/pkg/rtc/helper_test.go +++ b/pkg/rtc/helper_test.go @@ -21,7 +21,7 @@ func newMockParticipant(identity livekit.ParticipantIdentity, protocol types.Pro p.StateReturns(livekit.ParticipantInfo_JOINED) p.ProtocolVersionReturns(protocol) p.CanSubscribeReturns(true) - p.CanPublishReturns(!hidden) + p.CanPublishSourceReturns(!hidden) p.CanPublishDataReturns(!hidden) p.HiddenReturns(hidden) p.ToProtoReturns(&livekit.ParticipantInfo{ diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 3e140499c..aa3d4443d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -339,22 +339,13 @@ func (p *ParticipantImpl) SetPermission(permission *livekit.ParticipantPermissio } p.lock.Lock() video := p.grants.Video - hasChanged := video.GetCanSubscribe() != permission.CanSubscribe || - video.GetCanPublish() != permission.CanPublish || - video.GetCanPublishData() != permission.CanPublishData || - video.Hidden != permission.Hidden || - video.Recorder != permission.Recorder - if !hasChanged { + if video.MatchesPermission(permission) { p.lock.Unlock() return false } - video.SetCanSubscribe(permission.CanSubscribe) - video.SetCanPublish(permission.CanPublish) - video.SetCanPublishData(permission.CanPublishData) - video.Hidden = permission.Hidden - video.Recorder = permission.Recorder + video.UpdateFromPermission(permission) canPublish := video.GetCanPublish() canSubscribe := video.GetCanSubscribe() @@ -362,9 +353,9 @@ func (p *ParticipantImpl) SetPermission(permission *livekit.ParticipantPermissio onClaimsChanged := p.onClaimsChanged p.lock.Unlock() - // publish permission has been revoked then remove all published tracks - if !canPublish { - for _, track := range p.GetPublishedTracks() { + // publish permission has been revoked then remove offending tracks + for _, track := range p.GetPublishedTracks() { + if !video.GetCanPublishSource(track.Source()) { p.RemovePublishedTrack(track, false, false) if p.ProtocolVersion().SupportsUnpublish() { p.sendTrackUnpublished(track.ID()) @@ -374,6 +365,7 @@ func (p *ParticipantImpl) SetPermission(permission *livekit.ParticipantPermissio } } } + if canSubscribe { // reconcile everything p.SubscriptionManager.queueReconcile("") @@ -519,10 +511,6 @@ func (p *ParticipantImpl) onPublisherAnswer(answer webrtc.SessionDescription) er return err } - // received an offer from the client, if publishing is allowed, mark this - // participant as a publisher - p.setIsPublisher(p.CanPublish()) - if p.MigrateState() == types.MigrateStateSync { go p.handleMigrateMutedTrack() } @@ -582,7 +570,7 @@ func (p *ParticipantImpl) AddTrack(req *livekit.AddTrackRequest) { p.lock.Lock() defer p.lock.Unlock() - if !p.grants.Video.GetCanPublish() { + if !p.grants.Video.GetCanPublishSource(req.Source) { p.params.Logger.Warnw("no permission to publish track", nil) return } @@ -872,11 +860,10 @@ func (p *ParticipantImpl) IsPublisher() bool { return p.isPublisher.Load() } -func (p *ParticipantImpl) CanPublish() bool { +func (p *ParticipantImpl) CanPublishSource(source livekit.TrackSource) bool { p.lock.RLock() defer p.lock.RUnlock() - - return p.grants.Video.GetCanPublish() + return p.grants.Video.GetCanPublishSource(source) } func (p *ParticipantImpl) CanSubscribe() bool { @@ -1136,22 +1123,8 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w return } - if !p.CanPublish() { - p.params.Logger.Warnw("no permission to publish mediaTrack", nil) - return - } - publishedTrack, isNewTrack := p.mediaTrackReceived(track, rtpReceiver) - - if publishedTrack != nil { - p.params.Logger.Infow("mediaTrack published", - "kind", track.Kind().String(), - "trackID", publishedTrack.ID(), - "rid", track.RID(), - "SSRC", track.SSRC(), - "mime", track.Codec().MimeType, - ) - } else { + if publishedTrack == nil { p.params.Logger.Warnw("webrtc Track published but can't find MediaTrack", nil, "kind", track.Kind().String(), "webrtcTrackID", track.ID(), @@ -1159,9 +1132,29 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w "SSRC", track.SSRC(), "mime", track.Codec().MimeType, ) + return } - if !isNewTrack && publishedTrack != nil && !publishedTrack.HasPendingCodec() && p.IsReady() { + if !p.CanPublishSource(publishedTrack.Source()) { + p.params.Logger.Warnw("no permission to publish mediaTrack", nil, + "source", publishedTrack.Source(), + ) + return + } + + if !p.IsPublisher() { + p.setIsPublisher(true) + } + + p.params.Logger.Infow("mediaTrack published", + "kind", track.Kind().String(), + "trackID", publishedTrack.ID(), + "rid", track.RID(), + "SSRC", track.SSRC(), + "mime", track.Codec().MimeType, + ) + + if !isNewTrack && !publishedTrack.HasPendingCodec() && p.IsReady() { p.lock.RLock() onTrackUpdated := p.onTrackUpdated p.lock.RUnlock() diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 4d294c646..3c585c3b0 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -176,6 +176,32 @@ func TestTrackPublishing(t *testing.T) { // check SID is the same require.Equal(t, p.pendingTracks["cid"].trackInfos[0].Sid, p.pendingTracks["cid"].trackInfos[1].Sid) }) + + t.Run("should not allow adding disallowed sources", func(t *testing.T) { + p := newParticipantForTest("test") + p.SetPermission(&livekit.ParticipantPermission{ + CanPublish: true, + CanPublishSources: []livekit.TrackSource{ + livekit.TrackSource_CAMERA, + }, + }) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + p.AddTrack(&livekit.AddTrackRequest{ + Cid: "cid", + Name: "webcam", + Source: livekit.TrackSource_CAMERA, + Type: livekit.TrackType_VIDEO, + }) + require.Equal(t, 1, sink.WriteMessageCallCount()) + + p.AddTrack(&livekit.AddTrackRequest{ + Cid: "cid2", + Name: "rejected source", + Type: livekit.TrackType_AUDIO, + Source: livekit.TrackSource_MICROPHONE, + }) + require.Equal(t, 1, sink.WriteMessageCallCount()) + }) } func TestOutOfOrderUpdates(t *testing.T) { diff --git a/pkg/rtc/signalhandler.go b/pkg/rtc/signalhandler.go index dc49ade12..e8749aa64 100644 --- a/pkg/rtc/signalhandler.go +++ b/pkg/rtc/signalhandler.go @@ -73,6 +73,16 @@ func HandleParticipantSignal(room types.Room, participant types.LocalParticipant if msg.PingReq.Rtt > 0 { participant.UpdateSignalingRTT(uint32(msg.PingReq.Rtt)) } + + case *livekit.SignalRequest_UpdateMetadata: + if participant.ClaimGrants().Video.GetCanUpdateOwnMetadata() { + if msg.UpdateMetadata.Metadata != "" { + participant.SetMetadata(msg.UpdateMetadata.Metadata) + } + if msg.UpdateMetadata.Name != "" { + participant.SetName(msg.UpdateMetadata.Name) + } + } } return nil } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 6bfc95edf..a83c6206f 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -251,7 +251,7 @@ type LocalParticipant interface { // permissions ClaimGrants() *auth.ClaimGrants SetPermission(permission *livekit.ParticipantPermission) bool - CanPublish() bool + CanPublishSource(source livekit.TrackSource) bool CanSubscribe() bool CanPublishData() bool diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 67ce745dd..7ae08e75c 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -67,16 +67,6 @@ type FakeLocalParticipant struct { arg2 *webrtc.RTPTransceiver arg3 sfu.DownTrackState } - CanPublishStub func() bool - canPublishMutex sync.RWMutex - canPublishArgsForCall []struct { - } - canPublishReturns struct { - result1 bool - } - canPublishReturnsOnCall map[int]struct { - result1 bool - } CanPublishDataStub func() bool canPublishDataMutex sync.RWMutex canPublishDataArgsForCall []struct { @@ -87,6 +77,17 @@ type FakeLocalParticipant struct { canPublishDataReturnsOnCall map[int]struct { result1 bool } + CanPublishSourceStub func(livekit.TrackSource) bool + canPublishSourceMutex sync.RWMutex + canPublishSourceArgsForCall []struct { + arg1 livekit.TrackSource + } + canPublishSourceReturns struct { + result1 bool + } + canPublishSourceReturnsOnCall map[int]struct { + result1 bool + } CanSubscribeStub func() bool canSubscribeMutex sync.RWMutex canSubscribeArgsForCall []struct { @@ -1060,59 +1061,6 @@ func (fake *FakeLocalParticipant) CacheDownTrackArgsForCall(i int) (livekit.Trac return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } -func (fake *FakeLocalParticipant) CanPublish() bool { - fake.canPublishMutex.Lock() - ret, specificReturn := fake.canPublishReturnsOnCall[len(fake.canPublishArgsForCall)] - fake.canPublishArgsForCall = append(fake.canPublishArgsForCall, struct { - }{}) - stub := fake.CanPublishStub - fakeReturns := fake.canPublishReturns - fake.recordInvocation("CanPublish", []interface{}{}) - fake.canPublishMutex.Unlock() - if stub != nil { - return stub() - } - if specificReturn { - return ret.result1 - } - return fakeReturns.result1 -} - -func (fake *FakeLocalParticipant) CanPublishCallCount() int { - fake.canPublishMutex.RLock() - defer fake.canPublishMutex.RUnlock() - return len(fake.canPublishArgsForCall) -} - -func (fake *FakeLocalParticipant) CanPublishCalls(stub func() bool) { - fake.canPublishMutex.Lock() - defer fake.canPublishMutex.Unlock() - fake.CanPublishStub = stub -} - -func (fake *FakeLocalParticipant) CanPublishReturns(result1 bool) { - fake.canPublishMutex.Lock() - defer fake.canPublishMutex.Unlock() - fake.CanPublishStub = nil - fake.canPublishReturns = struct { - result1 bool - }{result1} -} - -func (fake *FakeLocalParticipant) CanPublishReturnsOnCall(i int, result1 bool) { - fake.canPublishMutex.Lock() - defer fake.canPublishMutex.Unlock() - fake.CanPublishStub = nil - if fake.canPublishReturnsOnCall == nil { - fake.canPublishReturnsOnCall = make(map[int]struct { - result1 bool - }) - } - fake.canPublishReturnsOnCall[i] = struct { - result1 bool - }{result1} -} - func (fake *FakeLocalParticipant) CanPublishData() bool { fake.canPublishDataMutex.Lock() ret, specificReturn := fake.canPublishDataReturnsOnCall[len(fake.canPublishDataArgsForCall)] @@ -1166,6 +1114,67 @@ func (fake *FakeLocalParticipant) CanPublishDataReturnsOnCall(i int, result1 boo }{result1} } +func (fake *FakeLocalParticipant) CanPublishSource(arg1 livekit.TrackSource) bool { + fake.canPublishSourceMutex.Lock() + ret, specificReturn := fake.canPublishSourceReturnsOnCall[len(fake.canPublishSourceArgsForCall)] + fake.canPublishSourceArgsForCall = append(fake.canPublishSourceArgsForCall, struct { + arg1 livekit.TrackSource + }{arg1}) + stub := fake.CanPublishSourceStub + fakeReturns := fake.canPublishSourceReturns + fake.recordInvocation("CanPublishSource", []interface{}{arg1}) + fake.canPublishSourceMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) CanPublishSourceCallCount() int { + fake.canPublishSourceMutex.RLock() + defer fake.canPublishSourceMutex.RUnlock() + return len(fake.canPublishSourceArgsForCall) +} + +func (fake *FakeLocalParticipant) CanPublishSourceCalls(stub func(livekit.TrackSource) bool) { + fake.canPublishSourceMutex.Lock() + defer fake.canPublishSourceMutex.Unlock() + fake.CanPublishSourceStub = stub +} + +func (fake *FakeLocalParticipant) CanPublishSourceArgsForCall(i int) livekit.TrackSource { + fake.canPublishSourceMutex.RLock() + defer fake.canPublishSourceMutex.RUnlock() + argsForCall := fake.canPublishSourceArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) CanPublishSourceReturns(result1 bool) { + fake.canPublishSourceMutex.Lock() + defer fake.canPublishSourceMutex.Unlock() + fake.CanPublishSourceStub = nil + fake.canPublishSourceReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) CanPublishSourceReturnsOnCall(i int, result1 bool) { + fake.canPublishSourceMutex.Lock() + defer fake.canPublishSourceMutex.Unlock() + fake.CanPublishSourceStub = nil + if fake.canPublishSourceReturnsOnCall == nil { + fake.canPublishSourceReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.canPublishSourceReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) CanSubscribe() bool { fake.canSubscribeMutex.Lock() ret, specificReturn := fake.canSubscribeReturnsOnCall[len(fake.canSubscribeArgsForCall)] @@ -5221,10 +5230,10 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.addTransceiverFromTrackToSubscriberMutex.RUnlock() fake.cacheDownTrackMutex.RLock() defer fake.cacheDownTrackMutex.RUnlock() - fake.canPublishMutex.RLock() - defer fake.canPublishMutex.RUnlock() fake.canPublishDataMutex.RLock() defer fake.canPublishDataMutex.RUnlock() + fake.canPublishSourceMutex.RLock() + defer fake.canPublishSourceMutex.RUnlock() fake.canSubscribeMutex.RLock() defer fake.canSubscribeMutex.RUnlock() fake.claimGrantsMutex.RLock() diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 22592a4c5..52d633423 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -110,7 +110,7 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic // this is new connection for existing participant - with publish only permissions if publishParam != "" { - // Make sure grant has CanPublish set, + // Make sure grant has GetCanPublish set, if !claims.Video.GetCanPublish() { return "", routing.ParticipantInit{}, http.StatusUnauthorized, rtc.ErrPermissionDenied }