From 2d9bf12581b5f680b7bf92307c147a18bae590fd Mon Sep 17 00:00:00 2001 From: RaphaelFakhri <153192858+RaphaelFakhri@users.noreply.github.com> Date: Tue, 29 Sep 2026 20:47:42 +0300 Subject: [PATCH] rtc: measure data blob keys by their content, not their text form (#4923) --- pkg/config/config.go | 22 ++++++-- pkg/rtc/participant_data_blob_handler.go | 3 +- pkg/rtc/participant_data_blob_handler_test.go | 55 +++++++++++++++++++ 3 files changed, 74 insertions(+), 6 deletions(-) diff --git a/pkg/config/config.go b/pkg/config/config.go index 3ca89107e..5dc2c91fd 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -399,8 +399,20 @@ func (l LimitConfig) CheckAttributesSize(attributes map[string]string) bool { return uint32(total) <= l.MaxAttributesSize } -func (l LimitConfig) CheckDataBlobKeyLength(key string) bool { - return l.MaxDataBlobKeyLength == 0 || len(key) <= l.MaxDataBlobKeyLength +// DataBlobKeyLength returns the length of the identifying content of a data blob key. +// The text form of the key message is not used, as it adds field names and quotes. +func DataBlobKeyLength(key *livekit.DataBlobKey) int { + switch k := key.GetKey().(type) { + case *livekit.DataBlobKey_Generic: + return len(k.Generic) + case *livekit.DataBlobKey_SchemaId: + return len(k.SchemaId.GetName()) + len(k.SchemaId.GetEncoding().GetCustom()) + } + return 0 +} + +func (l LimitConfig) CheckDataBlobKeyLength(key *livekit.DataBlobKey) bool { + return l.MaxDataBlobKeyLength == 0 || DataBlobKeyLength(key) <= l.MaxDataBlobKeyLength } func (l LimitConfig) CheckDataTrackCustomEncodingLength(identifier string) bool { @@ -430,7 +442,7 @@ func (l LimitConfig) CheckDataBlobsSize(dataBlobs []*livekit.DataBlob) bool { total := 0 for _, dataBlob := range dataBlobs { - total += len(dataBlob.GetKey().String()) + len(dataBlob.Contents) + total += DataBlobKeyLength(dataBlob.GetKey()) + len(dataBlob.Contents) } return uint32(total) <= l.MaxDataBlobSize } @@ -442,9 +454,9 @@ func (l LimitConfig) CanAddDataBlob(dataBlobs []*livekit.DataBlob, toAdd *liveki total := 0 for _, dataBlob := range dataBlobs { - total += len(dataBlob.Key.String()) + len(dataBlob.Contents) + total += DataBlobKeyLength(dataBlob.GetKey()) + len(dataBlob.Contents) } - return uint32(total+len(toAdd.GetKey().String())+len(toAdd.Contents)) <= l.MaxDataBlobSize + return uint32(total+DataBlobKeyLength(toAdd.GetKey())+len(toAdd.Contents)) <= l.MaxDataBlobSize } // --------------------------------- diff --git a/pkg/rtc/participant_data_blob_handler.go b/pkg/rtc/participant_data_blob_handler.go index 8c19ee765..b9d131ade 100644 --- a/pkg/rtc/participant_data_blob_handler.go +++ b/pkg/rtc/participant_data_blob_handler.go @@ -15,6 +15,7 @@ package rtc import ( + "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" @@ -31,7 +32,7 @@ func (p *ParticipantImpl) HandleStoreDataBlobRequest(req *livekit.StoreDataBlobR return } - if req.Blob == nil || req.Blob.Key == nil || len(req.Blob.Key.String()) == 0 || !p.params.LimitConfig.CheckDataBlobKeyLength(req.Blob.Key.String()) { + if req.Blob == nil || req.Blob.Key == nil || config.DataBlobKeyLength(req.Blob.Key) == 0 || !p.params.LimitConfig.CheckDataBlobKeyLength(req.Blob.Key) { p.pubLogger.Warnw("data blob is invalid", nil, "req", logger.Proto(req)) p.sendRequestResponse(&livekit.RequestResponse{ RequestId: req.RequestId, diff --git a/pkg/rtc/participant_data_blob_handler_test.go b/pkg/rtc/participant_data_blob_handler_test.go index 411a6a203..409cc1ddf 100644 --- a/pkg/rtc/participant_data_blob_handler_test.go +++ b/pkg/rtc/participant_data_blob_handler_test.go @@ -124,6 +124,61 @@ func TestHandleStoreDataBlobRequest(t *testing.T) { require.Equal(t, livekit.RequestResponse_INVALID_REQUEST, rr.Reason) }) + t.Run("returns INVALID_REQUEST when generic key is empty", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + p.HandleStoreDataBlobRequest(&livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Key: genericKey(""), + Contents: []byte("def"), + }, + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + rr := lastRequestResponse(t, sink, 0) + require.Equal(t, livekit.RequestResponse_INVALID_REQUEST, rr.Reason) + require.Empty(t, p.dataBlob.GetAll()) + }) + + t.Run("accepts a key exactly at the length limit", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 5, 0) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + key := genericKey("abcde") + p.HandleStoreDataBlobRequest(&livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Key: key, + Contents: []byte("def"), + }, + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + msg := sink.WriteMessageArgsForCall(0).(*livekit.SignalResponse) + _, ok := msg.Message.(*livekit.SignalResponse_StoreDataBlobResponse) + require.True(t, ok, "expected SignalResponse_StoreDataBlobResponse, got %T", msg.Message) + require.NotNil(t, p.dataBlob.Get(key)) + }) + + t.Run("counts only key and contents bytes against the size limit", func(t *testing.T) { + p := newParticipantWithDataBlob(t, true, 0, 9) + sink := p.params.Sink.(*routingfakes.FakeMessageSink) + + key := genericKey("abc") + p.HandleStoreDataBlobRequest(&livekit.StoreDataBlobRequest{ + Blob: &livekit.DataBlob{ + Key: key, + Contents: []byte("defghi"), + }, + }) + + require.Equal(t, 1, sink.WriteMessageCallCount()) + msg := sink.WriteMessageArgsForCall(0).(*livekit.SignalResponse) + _, ok := msg.Message.(*livekit.SignalResponse_StoreDataBlobResponse) + require.True(t, ok, "expected SignalResponse_StoreDataBlobResponse, got %T", msg.Message) + require.NotNil(t, p.dataBlob.Get(key)) + }) + t.Run("returns INVALID_REQUEST when contents is empty", func(t *testing.T) { p := newParticipantWithDataBlob(t, true, 0, 0) sink := p.params.Sink.(*routingfakes.FakeMessageSink)