From b2df257ac663cd98a8ca4d999189468b315bc3eb Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 10 Dec 2022 13:02:31 +0530 Subject: [PATCH] base62 encode user provided name/identity. (#1220) --- go.mod | 4 +- go.sum | 4 +- pkg/routing/localrouter.go | 18 ++++++-- pkg/routing/redis.go | 28 +++++++----- pkg/routing/redisrouter.go | 94 ++++++++++++++++++++++++++++---------- pkg/routing/utils.go | 56 ++++++++++++++++++++--- pkg/routing/utils_test.go | 49 ++++++++++++++++++++ 7 files changed, 205 insertions(+), 48 deletions(-) create mode 100644 pkg/routing/utils_test.go diff --git a/go.mod b/go.mod index 9ac164b86..73b47d571 100644 --- a/go.mod +++ b/go.mod @@ -15,9 +15,10 @@ require ( github.com/gorilla/websocket v1.5.0 github.com/hashicorp/go-version v1.6.0 github.com/hashicorp/golang-lru v0.6.0 + github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20221002073820-d9198083cfdc github.com/livekit/mediatransportutil v0.0.0-20221007030528-7440725c362b - github.com/livekit/protocol v1.3.0 + github.com/livekit/protocol v1.3.1-0.20221210053957-e43a6056a641 github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 github.com/mackerelio/go-osstat v0.2.3 github.com/magefile/mage v1.14.0 @@ -66,7 +67,6 @@ require ( github.com/google/subcommands v1.2.0 // indirect github.com/google/uuid v1.3.0 // indirect github.com/josharian/native v1.0.0 // indirect - github.com/jxskiss/base62 v1.1.0 // indirect github.com/lithammer/shortuuid/v3 v3.0.7 // indirect github.com/mattn/go-runewidth v0.0.9 // indirect github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect diff --git a/go.sum b/go.sum index 9380bfcca..ea43e5386 100644 --- a/go.sum +++ b/go.sum @@ -245,8 +245,8 @@ github.com/livekit/mageutil v0.0.0-20221002073820-d9198083cfdc h1:e3GIA9AL6h4a38 github.com/livekit/mageutil v0.0.0-20221002073820-d9198083cfdc/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20221007030528-7440725c362b h1:RBNV8TckETSkIkKxcD12d8nZKVkB9GSY/sQlMoaruP4= github.com/livekit/mediatransportutil v0.0.0-20221007030528-7440725c362b/go.mod h1:1Dlx20JPoIKGP45eo+yuj0HjeE25zmyeX/EWHiPCjFw= -github.com/livekit/protocol v1.3.0 h1:9rzcKrLxZQIsL5Us2ZARmAkTw1dZJMF3WJoS6MCZUIw= -github.com/livekit/protocol v1.3.0/go.mod h1:87MLIMbKaZs0SbDebMw1b7n/gD5utwEDq3zIRYcHJHA= +github.com/livekit/protocol v1.3.1-0.20221210053957-e43a6056a641 h1:CXGhAjIOaq/yzQz7OVEqN2CZcEZ8J3Bq7GbMufe35CA= +github.com/livekit/protocol v1.3.1-0.20221210053957-e43a6056a641/go.mod h1:87MLIMbKaZs0SbDebMw1b7n/gD5utwEDq3zIRYcHJHA= github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 h1:vOaY2qvfLihDyeZtnGGN1Law9wRrw8BMGCr1TygTvMw= github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995/go.mod h1:116ych8UaEs9vfIE8n6iZCZ30iagUFTls0vRmC+Ix5U= github.com/mackerelio/go-osstat v0.2.3 h1:jAMXD5erlDE39kdX2CU7YwCGRcxIO33u/p8+Fhe5dJw= diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index 375e6d3a6..97eb640a8 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -130,16 +130,20 @@ func (r *LocalRouter) StartParticipantSignal(ctx context.Context, roomName livek } func (r *LocalRouter) WriteParticipantRTC(_ context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity, msg *livekit.RTCNodeMessage) error { + r.lock.Lock() if r.rtcMessageChan.IsClosed() { // create a new one r.rtcMessageChan = NewMessageChannel(localRTCChannelSize) } - msg.ParticipantKey = string(participantKey(roomName, identity)) + r.lock.Unlock() + msg.ParticipantKey = string(participantKeyLegacy(roomName, identity)) + msg.ParticipantKeyB62 = string(participantKey(roomName, identity)) return r.writeRTCMessage(r.rtcMessageChan, msg) } func (r *LocalRouter) WriteRoomRTC(ctx context.Context, roomName livekit.RoomName, msg *livekit.RTCNodeMessage) error { - msg.ParticipantKey = string(participantKey(roomName, "")) + msg.ParticipantKey = string(participantKeyLegacy(roomName, "")) + msg.ParticipantKeyB62 = string(participantKey(roomName, "")) return r.WriteNodeRTC(ctx, r.currentNode.Id, msg) } @@ -246,7 +250,15 @@ func (r *LocalRouter) rtcMessageWorker() { // consume messages from for msg := range msgChan { if rtcMsg, ok := msg.(*livekit.RTCNodeMessage); ok { - room, identity, err := parseParticipantKey(livekit.ParticipantKey(rtcMsg.ParticipantKey)) + var room livekit.RoomName + var identity livekit.ParticipantIdentity + var err error + if rtcMsg.ParticipantKeyB62 != "" { + room, identity, err = parseParticipantKey(livekit.ParticipantKey(rtcMsg.ParticipantKeyB62)) + } + if err != nil { + room, identity, err = parseParticipantKeyLegacy(livekit.ParticipantKey(rtcMsg.ParticipantKey)) + } if err != nil { logger.Errorw("could not process RTC message", err) continue diff --git a/pkg/routing/redis.go b/pkg/routing/redis.go index afa0163fa..b31548d96 100644 --- a/pkg/routing/redis.go +++ b/pkg/routing/redis.go @@ -38,9 +38,10 @@ func signalNodeChannel(nodeID livekit.NodeID) string { return "signal_channel:" + string(nodeID) } -func publishRTCMessage(rc redis.UniversalClient, nodeID livekit.NodeID, participantKey livekit.ParticipantKey, msg proto.Message) error { +func publishRTCMessage(rc redis.UniversalClient, nodeID livekit.NodeID, participantKey livekit.ParticipantKey, participantKeyB62 livekit.ParticipantKey, msg proto.Message) error { rm := &livekit.RTCNodeMessage{ - ParticipantKey: string(participantKey), + ParticipantKey: string(participantKey), + ParticipantKeyB62: string(participantKeyB62), } switch o := msg.(type) { case *livekit.StartSession: @@ -54,6 +55,7 @@ func publishRTCMessage(rc redis.UniversalClient, nodeID livekit.NodeID, particip case *livekit.RTCNodeMessage: rm = o rm.ParticipantKey = string(participantKey) + rm.ParticipantKeyB62 = string(participantKeyB62) default: return ErrInvalidRouterMessage } @@ -94,18 +96,20 @@ func publishSignalMessage(rc redis.UniversalClient, nodeID livekit.NodeID, conne } type RTCNodeSink struct { - rc redis.UniversalClient - nodeID livekit.NodeID - participantKey livekit.ParticipantKey - isClosed atomic.Bool - onClose func() + rc redis.UniversalClient + nodeID livekit.NodeID + participantKey livekit.ParticipantKey + participantKeyB62 livekit.ParticipantKey + isClosed atomic.Bool + onClose func() } -func NewRTCNodeSink(rc redis.UniversalClient, nodeID livekit.NodeID, participantKey livekit.ParticipantKey) *RTCNodeSink { +func NewRTCNodeSink(rc redis.UniversalClient, nodeID livekit.NodeID, participantKey livekit.ParticipantKey, participantKeyB62 livekit.ParticipantKey) *RTCNodeSink { return &RTCNodeSink{ - rc: rc, - nodeID: nodeID, - participantKey: participantKey, + rc: rc, + nodeID: nodeID, + participantKey: participantKey, + participantKeyB62: participantKeyB62, } } @@ -113,7 +117,7 @@ func (s *RTCNodeSink) WriteMessage(msg proto.Message) error { if s.isClosed.Load() { return ErrChannelClosed } - return publishRTCMessage(s.rc, s.nodeID, s.participantKey, msg) + return publishRTCMessage(s.rc, s.nodeID, s.participantKey, s.participantKeyB62, msg) } func (s *RTCNodeSink) Close() { diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index 48206ed84..e4c0e2dae 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -148,7 +148,8 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek // create a new connection id connectionID = livekit.ConnectionID(utils.NewGuid("CO_")) - pKey := participantKey(roomName, pi.Identity) + pKey := participantKeyLegacy(roomName, pi.Identity) + pKeyB62 := participantKey(roomName, pi.Identity) // map signal & rtc nodes if err = r.setParticipantSignalNode(connectionID, r.currentNode.Id); err != nil { @@ -159,7 +160,7 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek // set up response channel before sending StartSession and be ready to receive responses. resChan := r.getOrCreateMessageChannel(r.responseChannels, string(connectionID)) - sink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNode.Id), pKey) + sink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNode.Id), pKey, pKeyB62) // serialize claims ss, err := pi.ToStartSession(roomName, connectionID) @@ -177,14 +178,16 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek } func (r *RedisRouter) WriteParticipantRTC(_ context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity, msg *livekit.RTCNodeMessage) error { - pkey := participantKey(roomName, identity) - rtcNode, err := r.getParticipantRTCNode(pkey) + pkey := participantKeyLegacy(roomName, identity) + pkeyB62 := participantKey(roomName, identity) + rtcNode, err := r.getParticipantRTCNode(pkey, pkeyB62) if err != nil { return err } - rtcSink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNode), pkey) - msg.ParticipantKey = string(participantKey(roomName, identity)) + rtcSink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNode), pkey, pkeyB62) + msg.ParticipantKey = string(participantKeyLegacy(roomName, identity)) + msg.ParticipantKeyB62 = string(participantKey(roomName, identity)) return r.writeRTCMessage(rtcSink, msg) } @@ -193,16 +196,17 @@ func (r *RedisRouter) WriteRoomRTC(ctx context.Context, roomName livekit.RoomNam if err != nil { return err } - msg.ParticipantKey = string(participantKey(roomName, "")) + msg.ParticipantKey = string(participantKeyLegacy(roomName, "")) + msg.ParticipantKeyB62 = string(participantKey(roomName, "")) return r.WriteNodeRTC(ctx, node.Id, msg) } func (r *RedisRouter) WriteNodeRTC(_ context.Context, rtcNodeID string, msg *livekit.RTCNodeMessage) error { - rtcSink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNodeID), livekit.ParticipantKey(msg.ParticipantKey)) + rtcSink := NewRTCNodeSink(r.rc, livekit.NodeID(rtcNodeID), livekit.ParticipantKey(msg.ParticipantKey), livekit.ParticipantKey(msg.ParticipantKeyB62)) return r.writeRTCMessage(rtcSink, msg) } -func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantKey livekit.ParticipantKey) error { +func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantKey livekit.ParticipantKey, participantKeyB62 livekit.ParticipantKey) error { // find the node where the room is hosted at rtcNode, err := r.GetNodeForRoom(r.ctx, livekit.RoomName(ss.RoomName)) if err != nil { @@ -217,7 +221,7 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK return err } - if err := r.setParticipantRTCNode(participantKey, rtcNode.Id); err != nil { + if err := r.setParticipantRTCNode(participantKey, participantKeyB62, rtcNode.Id); err != nil { return err } @@ -236,7 +240,16 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK // the previous rtc worker thread is still consuming off of it. // we'll want to sever the connection and switch to the new one r.lock.RLock() - requestChan, ok := r.requestChannels[string(participantKey)] + var requestChan *MessageChannel + var ok bool + var pkey livekit.ParticipantKey + if participantKeyB62 != "" { + requestChan, ok = r.requestChannels[string(participantKeyB62)] + pkey = participantKeyB62 + } else { + requestChan, ok = r.requestChannels[string(participantKey)] + pkey = participantKey + } r.lock.RUnlock() if ok { requestChan.Close() @@ -247,7 +260,7 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK return err } - reqChan := r.getOrCreateMessageChannel(r.requestChannels, string(participantKey)) + reqChan := r.getOrCreateMessageChannel(r.requestChannels, string(pkey)) resSink := NewSignalNodeSink(r.rc, livekit.NodeID(signalNode), livekit.ConnectionID(ss.ConnectionId)) go func() { err := r.onNewParticipant( @@ -307,10 +320,19 @@ func (r *RedisRouter) Stop() { r.cancel() } -func (r *RedisRouter) setParticipantRTCNode(participantKey livekit.ParticipantKey, nodeID string) error { - err := r.rc.Set(r.ctx, participantRTCKey(participantKey), nodeID, participantMappingTTL).Err() - if err != nil { - err = errors.Wrap(err, "could not set rtc node") +func (r *RedisRouter) setParticipantRTCNode(participantKey livekit.ParticipantKey, participantKeyB62 livekit.ParticipantKey, nodeID string) error { + var err error + if participantKey != "" { + err1 := r.rc.Set(r.ctx, participantRTCKey(participantKey), nodeID, participantMappingTTL).Err() + if err1 != nil { + err = errors.Wrap(err, "could not set rtc node") + } + } + if participantKeyB62 != "" { + err2 := r.rc.Set(r.ctx, participantRTCKey(participantKeyB62), nodeID, participantMappingTTL).Err() + if err2 != nil { + err = errors.Wrap(err, "could not set rtc node") + } } return err } @@ -322,10 +344,22 @@ func (r *RedisRouter) setParticipantSignalNode(connectionID livekit.ConnectionID return nil } -func (r *RedisRouter) getParticipantRTCNode(participantKey livekit.ParticipantKey) (string, error) { - val, err := r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result() - if err == redis.Nil { - err = ErrNodeNotFound +func (r *RedisRouter) getParticipantRTCNode(participantKey livekit.ParticipantKey, participantKeyB62 livekit.ParticipantKey) (string, error) { + var val string + var err error + if participantKeyB62 != "" { + val, err = r.rc.Get(r.ctx, participantRTCKey(participantKeyB62)).Result() + if err == redis.Nil { + val, err = r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result() + if err == redis.Nil { + err = ErrNodeNotFound + } + } + } else { + val, err = r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result() + if err == redis.Nil { + err = ErrNodeNotFound + } } return val, err } @@ -449,17 +483,23 @@ func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error { func (r *RedisRouter) handleRTCMessage(rm *livekit.RTCNodeMessage) error { pKey := livekit.ParticipantKey(rm.ParticipantKey) + pKeyB62 := livekit.ParticipantKey(rm.ParticipantKeyB62) switch rmb := rm.Message.(type) { case *livekit.RTCNodeMessage_StartSession: // RTC session should start on this node - if err := r.startParticipantRTC(rmb.StartSession, pKey); err != nil { + if err := r.startParticipantRTC(rmb.StartSession, pKey, pKeyB62); err != nil { return errors.Wrap(err, "could not start participant") } case *livekit.RTCNodeMessage_Request: r.lock.RLock() - requestChan := r.requestChannels[string(pKey)] + var requestChan *MessageChannel + if pKeyB62 != "" { + requestChan = r.requestChannels[string(pKeyB62)] + } else { + requestChan = r.requestChannels[string(pKey)] + } r.lock.RUnlock() if requestChan == nil { return ErrChannelClosed @@ -498,7 +538,15 @@ func (r *RedisRouter) handleRTCMessage(rm *livekit.RTCNodeMessage) error { default: // route it to handler if r.onRTCMessage != nil { - roomName, identity, err := parseParticipantKey(pKey) + var roomName livekit.RoomName + var identity livekit.ParticipantIdentity + var err error + if pKeyB62 != "" { + roomName, identity, err = parseParticipantKey(pKeyB62) + } + if err != nil { + roomName, identity, err = parseParticipantKeyLegacy(pKey) + } if err != nil { return err } diff --git a/pkg/routing/utils.go b/pkg/routing/utils.go index 7e0ac9388..8db4f649c 100644 --- a/pkg/routing/utils.go +++ b/pkg/routing/utils.go @@ -1,22 +1,66 @@ package routing import ( - "errors" + "fmt" "strings" + "github.com/jxskiss/base62" + "github.com/livekit/protocol/livekit" ) -func participantKey(roomName livekit.RoomName, identity livekit.ParticipantIdentity) livekit.ParticipantKey { +func participantKeyLegacy(roomName livekit.RoomName, identity livekit.ParticipantIdentity) livekit.ParticipantKey { return livekit.ParticipantKey(string(roomName) + "|" + string(identity)) } -func parseParticipantKey(pkey livekit.ParticipantKey) (roomName livekit.RoomName, identity livekit.ParticipantIdentity, err error) { +func parseParticipantKeyLegacy(pkey livekit.ParticipantKey) (roomName livekit.RoomName, identity livekit.ParticipantIdentity, err error) { parts := strings.Split(string(pkey), "|") - if len(parts) != 2 { - err = errors.New("invalid participant key") + if len(parts) == 2 { + roomName = livekit.RoomName(parts[0]) + identity = livekit.ParticipantIdentity(parts[1]) return } - return livekit.RoomName(parts[0]), livekit.ParticipantIdentity(parts[1]), nil + err = fmt.Errorf("invalid participant key: %s", pkey) + return +} + +func participantKey(roomName livekit.RoomName, identity livekit.ParticipantIdentity) livekit.ParticipantKey { + return livekit.ParticipantKey(encode(string(roomName), string(identity))) +} + +func parseParticipantKey(pkey livekit.ParticipantKey) (roomName livekit.RoomName, identity livekit.ParticipantIdentity, err error) { + parts, err := decode(string(pkey)) + if err != nil { + return + } + if len(parts) == 2 { + roomName = livekit.RoomName(parts[0]) + identity = livekit.ParticipantIdentity(parts[1]) + return + } + + err = fmt.Errorf("invalid participant key: %s", pkey) + return +} + +func encode(str ...string) string { + encoded := make([]string, 0, len(str)) + for _, s := range str { + encoded = append(encoded, base62.EncodeToString([]byte(s))) + } + return strings.Join(encoded, "|") +} + +func decode(encoded string) ([]string, error) { + split := strings.Split(encoded, "|") + decoded := make([]string, 0, len(split)) + for _, s := range split { + part, err := base62.DecodeString(s) + if err != nil { + return nil, err + } + decoded = append(decoded, string(part)) + } + return decoded, nil } diff --git a/pkg/routing/utils_test.go b/pkg/routing/utils_test.go new file mode 100644 index 000000000..127d3fd13 --- /dev/null +++ b/pkg/routing/utils_test.go @@ -0,0 +1,49 @@ +package routing + +import ( + "testing" + + "github.com/livekit/protocol/livekit" + "github.com/stretchr/testify/require" +) + +func TestUtils_ParticipantKey(t *testing.T) { + // encode/decode empty + encoded := participantKey("", "") + roomName, identity, err := parseParticipantKey(encoded) + require.NoError(t, err) + require.Equal(t, livekit.RoomName(""), roomName) + require.Equal(t, livekit.ParticipantIdentity(""), identity) + + // decode invalid + _, _, err = parseParticipantKey("abcd") + require.Error(t, err) + + // encode/decode without delimiter + encoded = participantKey("room1", "identity1") + roomName, identity, err = parseParticipantKey(encoded) + require.NoError(t, err) + require.Equal(t, livekit.RoomName("room1"), roomName) + require.Equal(t, livekit.ParticipantIdentity("identity1"), identity) + + // encode/decode with delimiter in roomName + encoded = participantKey("room1|alter_room1", "identity1") + roomName, identity, err = parseParticipantKey(encoded) + require.NoError(t, err) + require.Equal(t, livekit.RoomName("room1|alter_room1"), roomName) + require.Equal(t, livekit.ParticipantIdentity("identity1"), identity) + + // encode/decode with delimiter in identity + encoded = participantKey("room1", "identity1|alter-identity1") + roomName, identity, err = parseParticipantKey(encoded) + require.NoError(t, err) + require.Equal(t, livekit.RoomName("room1"), roomName) + require.Equal(t, livekit.ParticipantIdentity("identity1|alter-identity1"), identity) + + // encode/decode with delimiter in both and multiple delimiters in both + encoded = participantKey("room1|alter_room1|again_room1", "identity1|alter-identity1|again-identity1") + roomName, identity, err = parseParticipantKey(encoded) + require.NoError(t, err) + require.Equal(t, livekit.RoomName("room1|alter_room1|again_room1"), roomName) + require.Equal(t, livekit.ParticipantIdentity("identity1|alter-identity1|again-identity1"), identity) +}