dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 00:44:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

base62 encode user provided name/identity. (#1220)

Browse Source
This commit is contained in:
Raja Subramanian
2022-12-10 13:02:31 +05:30
committed by GitHub
parent 7a1273151f
commit b2df257ac6
7 changed files with 205 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+2 -2
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/routing/localrouter.go
+15 -3
View File
@@ -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
pkg/routing/redis.go
+16 -12
View File
@@ -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() {
pkg/routing/redisrouter.go
+71 -23
View File
@@ -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
}
pkg/routing/utils.go
+50 -6
View File
@@ -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
}
pkg/routing/utils_test.go
+49
View File
@@ -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)
}