dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-27 22:34:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implements source-specific permissions and client-driven metadata updates (#1590)

Browse Source 
Closes #1565
This commit is contained in:
David Zhao
2023-04-07 23:47:49 -07:00
committed by GitHub
parent e32eaa451f
commit e03f75d6a1
9 changed files with 147 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/rtc/helper_test.go
+1 -1
View File
@@ -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{
pkg/rtc/participant.go
+31 -38
View File
@@ -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()
pkg/rtc/participant_internal_test.go
+26
View File
@@ -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) {
pkg/rtc/signalhandler.go
+10
View File
@@ -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
}
pkg/rtc/types/interfaces.go
+1 -1
View File
@@ -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
pkg/rtc/types/typesfakes/fake_local_participant.go
+74 -65
View File
@@ -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()
pkg/service/rtcservice.go
+1 -1
View File
@@ -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
}