diff --git a/datatrackcryptor.go b/datatrackcryptor.go new file mode 100644 index 00000000..4f2771b7 --- /dev/null +++ b/datatrackcryptor.go @@ -0,0 +1,75 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package lksdk + +import ( + "math" + + dtp "github.com/livekit/protocol/datatrack" + + "github.com/livekit/server-sdk-go/v2/e2ee" + "github.com/livekit/server-sdk-go/v2/e2ee/types" + "github.com/livekit/server-sdk-go/v2/internal/datatrack" +) + +// dataTrackCryptor seals and opens data track frames with the room's DataCryptor, carrying the key +// index and IV in the packet header's E2EE extension. +type dataTrackCryptor struct { + cryptor *e2ee.DataCryptor +} + +func (c dataTrackCryptor) Encrypt(payload []byte) ([]byte, dtp.ExtensionE2EE, error) { + encrypted, err := c.cryptor.EncryptPayload(payload) + if err != nil { + return nil, dtp.ExtensionE2EE{}, err + } + if encrypted.KeyIndex > math.MaxUint8 { + return nil, dtp.ExtensionE2EE{}, types.ErrKeyIndexOutOfRange + } + + var iv [dtp.ExtensionE2EEIVLength]byte + if len(encrypted.IV) != len(iv) { + return nil, dtp.ExtensionE2EE{}, types.ErrIncorrectIVLength + } + copy(iv[:], encrypted.IV) + return encrypted.Ciphertext, *dtp.NewExtensionE2EE(uint8(encrypted.KeyIndex), iv), nil +} + +func (c dataTrackCryptor) Decrypt(payload []byte, ext dtp.ExtensionE2EE) ([]byte, error) { + iv := ext.IV() + return c.cryptor.DecryptPayload(e2ee.EncryptedPayload{ + Ciphertext: payload, + KeyIndex: uint32(ext.KeyIndex()), + IV: iv[:], + }) +} + +// dataTrackEncryptor returns the encryptor for the current session, or nil when data encryption +// is not enabled. +func (e *RTCEngine) dataTrackEncryptor() datatrack.Encryptor { + if e.dataCryptor == nil { + return nil + } + return dataTrackCryptor{cryptor: e.dataCryptor} +} + +// dataTrackDecryptor returns the decryptor for the current session, or nil when data encryption +// is not enabled. +func (e *RTCEngine) dataTrackDecryptor() datatrack.Decryptor { + if e.dataCryptor == nil { + return nil + } + return dataTrackCryptor{cryptor: e.dataCryptor} +} diff --git a/datatrackcryptor_test.go b/datatrackcryptor_test.go new file mode 100644 index 00000000..b7f9ad32 --- /dev/null +++ b/datatrackcryptor_test.go @@ -0,0 +1,73 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package lksdk + +import ( + "bytes" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/livekit/server-sdk-go/v2/e2ee" + "github.com/livekit/server-sdk-go/v2/e2ee/types" +) + +// fixedKeyProvider serves one key at any index, standing in for a provider without an index bound. +type fixedKeyProvider struct { + index uint32 +} + +func (p fixedKeyProvider) GetKey(uint32) ([]byte, error) { + return bytes.Repeat([]byte{0x11}, types.KeySizeBytes), nil +} + +func (p fixedKeyProvider) CurrentKeyIndex() uint32 { + return p.index +} + +func TestDataTrackCryptorRoundTrip(t *testing.T) { + kp := e2ee.NewExternalKeyProvider() + require.NoError(t, kp.SetRawKey(bytes.Repeat([]byte{0x11}, types.KeySizeBytes), 3)) + c := dataTrackCryptor{cryptor: e2ee.NewDataCryptor(kp)} + + ciphertext, ext, err := c.Encrypt([]byte("hello encrypted world")) + require.NoError(t, err) + require.Equal(t, uint8(3), ext.KeyIndex()) + require.NotEqual(t, []byte("hello encrypted world"), ciphertext) + + plaintext, err := c.Decrypt(ciphertext, ext) + require.NoError(t, err) + require.Equal(t, []byte("hello encrypted world"), plaintext) +} + +func TestDataTrackCryptorRejectsOutOfRangeKeyIndex(t *testing.T) { + c := dataTrackCryptor{cryptor: e2ee.NewDataCryptor(fixedKeyProvider{index: 256})} + + _, _, err := c.Encrypt([]byte("hello")) + require.ErrorIs(t, err, types.ErrKeyIndexOutOfRange) +} + +func TestDataTrackCryptorWrongKeyFails(t *testing.T) { + encryptor := dataTrackCryptor{cryptor: e2ee.NewDataCryptor(fixedKeyProvider{})} + kp := e2ee.NewExternalKeyProvider() + require.NoError(t, kp.SetRawKey(bytes.Repeat([]byte{0x22}, types.KeySizeBytes), 0)) + decryptor := dataTrackCryptor{cryptor: e2ee.NewDataCryptor(kp)} + + ciphertext, ext, err := encryptor.Encrypt([]byte("secret")) + require.NoError(t, err) + + _, err = decryptor.Decrypt(ciphertext, ext) + require.Error(t, err) +} diff --git a/internal/datatrack/errors.go b/internal/datatrack/errors.go index 4c98fc4d..ffc19795 100644 --- a/internal/datatrack/errors.go +++ b/internal/datatrack/errors.go @@ -17,13 +17,14 @@ package datatrack import "errors" var ( - ErrNotAllowed = errors.New("data track publishing unauthorized") - ErrDuplicateName = errors.New("track name already taken") - ErrInvalidName = errors.New("track name invalid") - ErrLimitReached = errors.New("data track publication limit reached") - ErrPublishTimeout = errors.New("timed out publishing data track") - ErrSubscribeTimeout = errors.New("timed out subscribing to data track") - ErrDisconnected = errors.New("room disconnected") - ErrUnpublished = errors.New("track unpublished") - ErrQueueFull = errors.New("queue full") + ErrNotAllowed = errors.New("data track publishing unauthorized") + ErrDuplicateName = errors.New("track name already taken") + ErrInvalidName = errors.New("track name invalid") + ErrLimitReached = errors.New("data track publication limit reached") + ErrPublishTimeout = errors.New("timed out publishing data track") + ErrSubscribeTimeout = errors.New("timed out subscribing to data track") + ErrDisconnected = errors.New("room disconnected") + ErrUnpublished = errors.New("track unpublished") + ErrQueueFull = errors.New("queue full") + ErrEncryptionDisabled = errors.New("track is end-to-end encrypted but data encryption is not enabled") ) diff --git a/internal/datatrack/local.go b/internal/datatrack/local.go index 3437c151..fb337e94 100644 --- a/internal/datatrack/local.go +++ b/internal/datatrack/local.go @@ -71,8 +71,9 @@ func WithFrameEncoding(encoding FrameEncoding) PublishOption { type LocalManagerParams struct { Transport LocalTransport - // Encryptor enables end-to-end encryption for every published track; nil disables it. - Encryptor Encryptor + // Encryptor returns the encryptor for the current session; nil, or a nil result, disables + // end-to-end encryption for tracks published from then on. + Encryptor func() Encryptor Logger logger.Logger } @@ -115,6 +116,13 @@ func (m *LocalManager) logger() logger.Logger { return *m.log.Load() } +func (m *LocalManager) encryptor() Encryptor { + if m.params.Encryptor == nil { + return nil + } + return m.params.Encryptor() +} + // Publish requests a publication and waits for the SFU's answer. Ending ctx abandons the request; // should the SFU accept it afterwards, the track is unpublished right away. func (m *LocalManager) Publish(ctx context.Context, options PublishOptions) (*LocalTrack, error) { @@ -145,7 +153,7 @@ func (m *LocalManager) Publish(ctx context.Context, options PublishOptions) (*Lo request := publishRequest{ handle: handle, name: options.Name, - usesE2EE: m.params.Encryptor != nil, + usesE2EE: m.encryptor() != nil, schema: options.Schema, frameEncoding: options.FrameEncoding, } @@ -354,7 +362,7 @@ type LocalTrack struct { func newLocalTrack(manager *LocalManager, info Info) *LocalTrack { var encryptor Encryptor if info.UsesE2EE { - encryptor = manager.params.Encryptor + encryptor = manager.encryptor() } return &LocalTrack{ manager: manager, diff --git a/internal/datatrack/local_test.go b/internal/datatrack/local_test.go index 34702cfe..dfff3f76 100644 --- a/internal/datatrack/local_test.go +++ b/internal/datatrack/local_test.go @@ -229,7 +229,7 @@ func TestLocalManager_PublishCancelledWhileTransportBlocked(t *testing.T) { func TestLocalManager_PublishWithE2EE(t *testing.T) { transport := newFakeLocalTransport() - m := NewLocalManager(LocalManagerParams{Transport: transport, Encryptor: prefixingEncryptor{}}) + m := NewLocalManager(LocalManagerParams{Transport: transport, Encryptor: func() Encryptor { return prefixingEncryptor{} }}) result := publishAsync(context.Background(), m, "secure") diff --git a/internal/datatrack/remote.go b/internal/datatrack/remote.go index 08fc638e..158caaa4 100644 --- a/internal/datatrack/remote.go +++ b/internal/datatrack/remote.go @@ -75,8 +75,9 @@ type PipelineOptions struct { type RemoteManagerParams struct { Transport RemoteTransport - // Decryptor opens frames of tracks that use end-to-end encryption; nil disables decryption. - Decryptor Decryptor + // Decryptor returns the decryptor for the current session; nil, or a nil result, leaves frames of + // end-to-end encrypted tracks as received. + Decryptor func() Decryptor Logger logger.Logger } @@ -116,6 +117,13 @@ func (m *RemoteManager) logger() logger.Logger { return *m.log.Load() } +func (m *RemoteManager) decryptor() Decryptor { + if m.params.Decryptor == nil { + return nil + } + return m.params.Decryptor() +} + // HandleParticipantUpdate applies the data tracks listed for each participant in a // ParticipantUpdate; tracks a participant no longer lists are unpublished. func (m *RemoteManager) HandleParticipantUpdate(participants []*livekit.ParticipantInfo, localIdentity string) { @@ -428,6 +436,10 @@ func (t *RemoteTrack) Subscribe(ctx context.Context, opts ...SubscribeOption) (* t.mu.Unlock() return nil, ErrUnpublished } + if t.info.UsesE2EE && t.manager.decryptor() == nil { + t.mu.Unlock() + return nil, ErrEncryptionDisabled + } if t.subscription == subscriptionActive { stream := t.addStreamLocked(options.BufferSize) t.mu.Unlock() @@ -555,12 +567,12 @@ func (t *RemoteTrack) activate(handle trackHandle) (previous trackHandle, replac var decryptor Decryptor if t.info.UsesE2EE { - decryptor = t.manager.params.Decryptor + decryptor = t.manager.decryptor() } t.packets = make(chan *dtp.Packet, packetBufferCount) t.streamList = new(atomic.Pointer[[]*Stream]) t.streamList.Store(&[]*Stream{}) - go t.runPipeline(t.packets, t.streamList, newRemotePipeline(decryptor, t.manager.logger())) + go t.runPipeline(t.packets, t.streamList, newRemotePipeline(t.info.UsesE2EE, decryptor, t.manager.logger())) t.subHandle = handle t.subscription = subscriptionActive for _, waiter := range t.waiters { @@ -744,13 +756,14 @@ func (s *Stream) close() { // remotePipeline turns a subscription's packets back into frames. It is owned by one goroutine. type remotePipeline struct { + encrypted bool decryptor Decryptor log logger.Logger depacketizer *depacketizer } -func newRemotePipeline(decryptor Decryptor, log logger.Logger) *remotePipeline { - return &remotePipeline{decryptor: decryptor, log: log, depacketizer: newDepacketizer()} +func newRemotePipeline(encrypted bool, decryptor Decryptor, log logger.Logger) *remotePipeline { + return &remotePipeline{encrypted: encrypted, decryptor: decryptor, log: log, depacketizer: newDepacketizer()} } // processPacket reports whether the packet completed a frame. @@ -765,9 +778,13 @@ func (p *remotePipeline) processPacket(packet *dtp.Packet, maxPartialFrames int) } frame := Frame{Payload: result.frame.payload, UserTimestamp: result.frame.extensions.UserTimestamp} - if p.decryptor == nil { + if !p.encrypted { return frame, true } + if p.decryptor == nil { + p.log.Errorw("dropping encrypted data track frame, data encryption is not enabled", nil) + return Frame{}, false + } e2ee := result.frame.extensions.E2EE if e2ee == nil { p.log.Errorw("dropping data track frame without E2EE extension", nil) diff --git a/internal/datatrack/remote_test.go b/internal/datatrack/remote_test.go index db4d1a74..ca95bafa 100644 --- a/internal/datatrack/remote_test.go +++ b/internal/datatrack/remote_test.go @@ -167,7 +167,7 @@ func pushInterleavedTwoFramePair(t *testing.T, m *RemoteManager, handle trackHan func TestRemotePipeline_ProcessPacket(t *testing.T) { const payloadLen = 1024 - pipeline := newRemotePipeline(nil, logger.GetLogger()) + pipeline := newRemotePipeline(false, nil, logger.GetLogger()) header := testHeader() FrameMarkerSingle.apply(&header) @@ -338,7 +338,7 @@ func TestRemoteManager_SubscribeReceivesFrame(t *testing.T) { func TestRemoteManager_SubscribeWithE2EE(t *testing.T) { transport := newFakeRemoteTransport() - m := NewRemoteManager(RemoteManagerParams{Transport: transport, Decryptor: prefixStrippingDecryptor{}}) + m := NewRemoteManager(RemoteManagerParams{Transport: transport, Decryptor: func() Decryptor { return prefixStrippingDecryptor{} }}) trackSID, subHandle := livekit.TrackID("DTR_1234"), trackHandle(0x1234) @@ -554,7 +554,7 @@ func (d blockingDecryptor) Decrypt(payload []byte, _ dtp.ExtensionE2EE) ([]byte, func TestRemoteManager_ResubscribeDoesNotReceivePreviousSubscriptionFrames(t *testing.T) { decryptor := blockingDecryptor{entered: make(chan struct{}, 1), release: make(chan struct{})} transport := newFakeRemoteTransport() - m := NewRemoteManager(RemoteManagerParams{Transport: transport, Decryptor: decryptor}) + m := NewRemoteManager(RemoteManagerParams{Transport: transport, Decryptor: func() Decryptor { return decryptor }}) trackSID := livekit.TrackID("DTR_1234") oldHandle, newHandle := trackHandle(0x1234), trackHandle(0x1235) @@ -652,3 +652,26 @@ func TestStream_DropsOldestWhenFull(t *testing.T) { stream.close() expectClosed(t, stream) } + +func TestRemotePipeline_DropsEncryptedFrameWithoutDecryptor(t *testing.T) { + pipeline := newRemotePipeline(true, nil, logger.GetLogger()) + + header := testHeader() + FrameMarkerSingle.apply(&header) + Extensions{E2EE: &dtp.ExtensionE2EE{}}.apply(&header) + packet := &dtp.Packet{Header: header, Payload: []byte{1, 2, 3, 4, 5}} + + _, ok := pipeline.processPacket(packet, defaultMaxPartialFrames) + require.False(t, ok) +} + +func TestRemoteManager_SubscribeEncryptedTrackWithoutDecryptorFails(t *testing.T) { + transport := newFakeRemoteTransport() + m := NewRemoteManager(RemoteManagerParams{Transport: transport}) + + track := publishTrack(t, m, transport, "id", Info{SID: "DTR_1234", pubHandle: 1, Name: "secure", UsesE2EE: true}) + + _, err := track.Subscribe(context.Background()) + require.ErrorIs(t, err, ErrEncryptionDisabled) + expectNoEvent(t, transport.subscriptionUpdates) +} diff --git a/localparticipant.go b/localparticipant.go index 8f4a37e9..f55f61cb 100644 --- a/localparticipant.go +++ b/localparticipant.go @@ -56,7 +56,7 @@ func newLocalParticipant(engine *RTCEngine, roomcallback *RoomCallback, serverIn rpcPendingAcks: &sync.Map{}, rpcPendingResponses: &sync.Map{}, } - p.dataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: engine}, Logger: p.log}) + p.dataTracks = datatrack.NewLocalManager(datatrack.LocalManagerParams{Transport: localDataTrackTransport{engine: engine}, Encryptor: engine.dataTrackEncryptor, Logger: p.log}) return p } diff --git a/room.go b/room.go index 7a3633db..34a1b47b 100644 --- a/room.go +++ b/room.go @@ -368,7 +368,7 @@ func NewRoom(callback *RoomCallback) *Room { r.engine = NewRTCEngine(r.useSinglePeerConnection, r, r.getLocalParticipantSID, r.regionURLProvider) r.LocalParticipant = newLocalParticipant(r.engine, r.callback, r.serverInfo, r.log) - r.remoteDataTracks = datatrack.NewRemoteManager(datatrack.RemoteManagerParams{Transport: &remoteDataTrackTransport{room: r}, Logger: r.log}) + r.remoteDataTracks = datatrack.NewRemoteManager(datatrack.RemoteManagerParams{Transport: &remoteDataTrackTransport{room: r}, Decryptor: r.engine.dataTrackDecryptor, Logger: r.log}) return r }