[pull] master from livekit:master

pkg/service/rtcservice.go

@@ -16,6 +16,8 @@ package service
 
 import (
 	"context"
+	"encoding/base64"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"math/rand"
@@ -30,16 +32,17 @@ 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"
 	"github.com/livekit/livekit-server/pkg/rtc"
 	"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,15 +94,28 @@ 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
 	}
 	_, _ = 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

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)

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 (

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) {