Data tracks: end-to-end encryption (E2EE) support

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}
+}

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)
+}

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

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,

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

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)

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)
+}

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
 }
 

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
 }