diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index f1c23d230..a58200963 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -16,6 +16,8 @@ package service import ( "context" + "encoding/base64" + "encoding/json" "errors" "fmt" "math/rand" @@ -30,6 +32,10 @@ import ( "go.uber.org/atomic" "golang.org/x/exp/maps" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/livekit/psrpc" + "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/routing/selector" @@ -37,9 +43,6 @@ import ( "github.com/livekit/livekit-server/pkg/telemetry" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/livekit-server/pkg/utils" - "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/logger" - "github.com/livekit/psrpc" ) type RTCService struct { @@ -91,7 +94,8 @@ func (s *RTCService) SetupRoutes(mux *http.ServeMux) { } func (s *RTCService) validate(w http.ResponseWriter, r *http.Request) { - _, _, code, err := s.validateInternal(r) + log := utils.GetLogger(r.Context()) + _, _, code, err := s.validateInternal(log, r, true) if err != nil { HandleError(w, r, code, err) return @@ -99,7 +103,19 @@ func (s *RTCService) validate(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("success")) } -func (s *RTCService) validateInternal(r *http.Request) (livekit.RoomName, routing.ParticipantInit, int, error) { +func decodeAttributes(str string) (map[string]string, error) { + data, err := base64.URLEncoding.DecodeString(str) + if err != nil { + return nil, err + } + var attrs map[string]string + if err := json.Unmarshal(data, &attrs); err != nil { + return nil, err + } + return attrs, nil +} + +func (s *RTCService) validateInternal(log logger.Logger, r *http.Request, strict bool) (livekit.RoomName, routing.ParticipantInit, int, error) { claims := GetGrants(r.Context()) var pi routing.ParticipantInit @@ -129,6 +145,7 @@ func (s *RTCService) validateInternal(r *http.Request) (livekit.RoomName, routin participantID := r.FormValue("sid") subscriberAllowPauseParam := r.FormValue("subscriber_allow_pause") disableICELite := r.FormValue("disable_ice_lite") + attributesStr := r.FormValue("attributes") if onlyName != "" { roomName = onlyName @@ -174,6 +191,31 @@ func (s *RTCService) validateInternal(r *http.Request) (livekit.RoomName, routin } SetRoomConfiguration(createRequest, claims.GetRoomConfiguration()) + // Add extra attributes to the participant + if attributesStr != "" { + // Make sure grant has GetCanUpdateOwnMetadata set + if !claims.Video.GetCanUpdateOwnMetadata() { + return "", routing.ParticipantInit{}, http.StatusUnauthorized, rtc.ErrPermissionDenied + } + attrs, err := decodeAttributes(attributesStr) + if err != nil { + if strict { + return "", pi, http.StatusBadRequest, errors.New("cannot decode attributes") + } + log.Debugw("failed to decode attributes", "error", err) + // attrs will be empty here, so just proceed + } + if len(attrs) != 0 && claims.Attributes == nil { + claims.Attributes = make(map[string]string, len(attrs)) + } + for k, v := range attrs { + if v == "" { + continue // do not allow deleting existing attributes + } + claims.Attributes[k] = v + } + } + pi = routing.ParticipantInit{ Reconnect: boolValue(reconnectParam), ReconnectReason: livekit.ReconnectReason(reconnectReason), @@ -257,7 +299,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { loggerResolved = false } - roomName, pi, code, err = s.validateInternal(r) + roomName, pi, code, err = s.validateInternal(pLogger, r, false) if err != nil { HandleError(w, r, code, err) return diff --git a/test/client/client.go b/test/client/client.go index f5bae8b46..68467c4b6 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -16,6 +16,8 @@ package client import ( "context" + "encoding/base64" + "encoding/json" "errors" "fmt" "io" @@ -113,6 +115,7 @@ var ( type Options struct { AutoSubscribe bool Publish string + Attributes map[string]string ClientInfo *livekit.ClientInfo DisabledCodecs []webrtc.RTPCodecCapability TokenCustomizer func(token *auth.AccessToken, grants *auth.VideoGrant) @@ -135,6 +138,13 @@ func NewWebSocketConn(host, token string, opts *Options) (*websocket.Conn, error if opts.Publish != "" { connectUrl += encodeQueryParam("publish", opts.Publish) } + if len(opts.Attributes) != 0 { + data, err := json.Marshal(opts.Attributes) + if err != nil { + return nil, err + } + connectUrl += encodeQueryParam("attributes", base64.URLEncoding.EncodeToString(data)) + } if opts.ClientInfo != nil { if opts.ClientInfo.DeviceModel != "" { connectUrl += encodeQueryParam("device_model", opts.ClientInfo.DeviceModel) diff --git a/test/integration_helpers.go b/test/integration_helpers.go index 66bb351e5..abb6bc149 100644 --- a/test/integration_helpers.go +++ b/test/integration_helpers.go @@ -25,17 +25,18 @@ import ( "github.com/redis/go-redis/v9" "github.com/twitchtv/twirp" + "github.com/livekit/mediatransportutil/pkg/rtcconfig" + "github.com/livekit/protocol/auth" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/utils/guid" + "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/service" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/livekit-server/pkg/testutils" testclient "github.com/livekit/livekit-server/test/client" - "github.com/livekit/mediatransportutil/pkg/rtcconfig" - "github.com/livekit/protocol/auth" - "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/utils/guid" ) const ( diff --git a/test/singlenode_test.go b/test/singlenode_test.go index 91668123b..761f8398b 100644 --- a/test/singlenode_test.go +++ b/test/singlenode_test.go @@ -20,6 +20,7 @@ import ( "errors" "fmt" "net/http" + "reflect" "strings" "testing" "time" @@ -542,6 +543,57 @@ func TestSingleNodeUpdateSubscriptionPermissions(t *testing.T) { }) } +func TestSingleNodeAttributes(t *testing.T) { + if testing.Short() { + t.SkipNow() + return + } + _, finish := setupSingleNodeTest("TestSingleNodeAttributes") + defer finish() + + pub := createRTCClient("pub", defaultServerPort, &testclient.Options{ + Attributes: map[string]string{ + "b": "2", + "c": "3", + }, + TokenCustomizer: func(token *auth.AccessToken, grants *auth.VideoGrant) { + T := true + grants.CanUpdateOwnMetadata = &T + token.SetAttributes(map[string]string{ + "a": "0", + "b": "1", + }) + }, + }) + grant := &auth.VideoGrant{RoomJoin: true, Room: testRoom} + grant.SetCanSubscribe(false) + at := auth.NewAccessToken(testApiKey, testApiSecret). + SetVideoGrant(grant). + SetIdentity("sub") + token, err := at.ToJWT() + require.NoError(t, err) + sub := createRTCClientWithToken(token, defaultServerPort, nil) + + waitUntilConnected(t, pub, sub) + + // wait sub receives initial attributes + testutils.WithTimeout(t, func() string { + pubRemote := sub.GetRemoteParticipant(pub.ID()) + if pubRemote == nil { + return "could not find remote publisher" + } + attrs := pubRemote.Attributes + if !reflect.DeepEqual(attrs, map[string]string{ + "a": "0", + "b": "2", + "c": "3", + }) { + return fmt.Sprintf("did not receive expected attributes: %v", attrs) + } + return "" + }) +} + // TestDeviceCodecOverride checks that codecs that are incompatible with a device is not // negotiated by the server func TestDeviceCodecOverride(t *testing.T) {