From 2beb1aba7753985dd6367741d029db9d17a37dde Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Thu, 3 Sep 2026 16:15:36 -0700
Subject: [PATCH 1/6] Resolve data track cryptors per session

---
 internal/datatrack/local.go       | 16 ++++++++++++----
 internal/datatrack/local_test.go  |  2 +-
 internal/datatrack/remote.go      | 16 +++++++++++++---
 internal/datatrack/remote_test.go |  2 +-
 4 files changed, 27 insertions(+), 9 deletions(-)

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..18a4be7b 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) {
@@ -555,7 +563,9 @@ func (t *RemoteTrack) activate(handle trackHandle) (previous trackHandle, replac
 
 	var decryptor Decryptor
 	if t.info.UsesE2EE {
-		decryptor = t.manager.params.Decryptor
+		if decryptor = t.manager.decryptor(); decryptor == nil {
+			t.manager.params.Logger.Warnw("subscribing to an end-to-end encrypted data track without encryption enabled, frames are delivered as received", nil, "sid", t.info.SID)
+		}
 	}
 	t.packets = make(chan *dtp.Packet, packetBufferCount)
 	t.streamList = new(atomic.Pointer[[]*Stream])
diff --git a/internal/datatrack/remote_test.go b/internal/datatrack/remote_test.go
index db4d1a74..c64cb62a 100644
--- a/internal/datatrack/remote_test.go
+++ b/internal/datatrack/remote_test.go
@@ -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)
 

From fe57e969158e38ee346224bf7467393338d46282 Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Thu, 3 Sep 2026 16:16:24 -0700
Subject: [PATCH 2/6] Add data track cryptor over DataCryptor

---
 datatrackcryptor.go      | 72 +++++++++++++++++++++++++++++++++++++++
 datatrackcryptor_test.go | 73 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 145 insertions(+)
 create mode 100644 datatrackcryptor.go
 create mode 100644 datatrackcryptor_test.go

diff --git a/datatrackcryptor.go b/datatrackcryptor.go
new file mode 100644
index 00000000..14ef8d4e
--- /dev/null
+++ b/datatrackcryptor.go
@@ -0,0 +1,72 @@
+// 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"
+
+	"github.com/livekit/server-sdk-go/v2/datatrack"
+	"github.com/livekit/server-sdk-go/v2/e2ee"
+	"github.com/livekit/server-sdk-go/v2/e2ee/types"
+)
+
+// 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, datatrack.E2EEExtension, error) {
+	encrypted, err := c.cryptor.EncryptPayload(payload)
+	if err != nil {
+		return nil, datatrack.E2EEExtension{}, err
+	}
+	if encrypted.KeyIndex > math.MaxUint8 {
+		return nil, datatrack.E2EEExtension{}, types.ErrKeyIndexOutOfRange
+	}
+
+	ext := datatrack.E2EEExtension{KeyIndex: uint8(encrypted.KeyIndex)}
+	if len(encrypted.IV) != len(ext.IV) {
+		return nil, datatrack.E2EEExtension{}, types.ErrIncorrectIVLength
+	}
+	copy(ext.IV[:], encrypted.IV)
+	return encrypted.Ciphertext, ext, nil
+}
+
+func (c dataTrackCryptor) Decrypt(payload []byte, ext datatrack.E2EEExtension) ([]byte, error) {
+	return c.cryptor.DecryptPayload(e2ee.EncryptedPayload{
+		Ciphertext: payload,
+		KeyIndex:   uint32(ext.KeyIndex),
+		IV:         ext.IV[:],
+	})
+}
+
+// dataTrackEncryptor returns the encryptor for the current session, or nil when data encryption
+// is not enabled.
+func (r *Room) dataTrackEncryptor() datatrack.Encryptor {
+	if r.engine.dataCryptor == nil {
+		return nil
+	}
+	return dataTrackCryptor{cryptor: r.engine.dataCryptor}
+}
+
+// dataTrackDecryptor returns the decryptor for the current session, or nil when data encryption
+// is not enabled.
+func (r *Room) dataTrackDecryptor() datatrack.Decryptor {
+	if r.engine.dataCryptor == nil {
+		return nil
+	}
+	return dataTrackCryptor{cryptor: r.engine.dataCryptor}
+}
diff --git a/datatrackcryptor_test.go b/datatrackcryptor_test.go
new file mode 100644
index 00000000..03059ffe
--- /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)
+}

From d86c61685032315253c03b7230e81fe1736ddd9e Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Thu, 3 Sep 2026 16:16:32 -0700
Subject: [PATCH 3/6] Enable end-to-end encryption for data tracks

---
 datatrackcryptor.go               | 38 ++++++++++++++++---------------
 internal/datatrack/remote_test.go |  2 +-
 localparticipant.go               |  2 +-
 room.go                           |  2 +-
 4 files changed, 23 insertions(+), 21 deletions(-)

diff --git a/datatrackcryptor.go b/datatrackcryptor.go
index 14ef8d4e..2a97325d 100644
--- a/datatrackcryptor.go
+++ b/datatrackcryptor.go
@@ -17,9 +17,10 @@ package lksdk
 import (
 	"math"
 
-	"github.com/livekit/server-sdk-go/v2/datatrack"
+	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
@@ -28,45 +29,46 @@ type dataTrackCryptor struct {
 	cryptor *e2ee.DataCryptor
 }
 
-func (c dataTrackCryptor) Encrypt(payload []byte) ([]byte, datatrack.E2EEExtension, error) {
+func (c dataTrackCryptor) Encrypt(payload []byte) ([]byte, dtp.ExtensionE2EE, error) {
 	encrypted, err := c.cryptor.EncryptPayload(payload)
 	if err != nil {
-		return nil, datatrack.E2EEExtension{}, err
+		return nil, dtp.ExtensionE2EE{}, err
 	}
 	if encrypted.KeyIndex > math.MaxUint8 {
-		return nil, datatrack.E2EEExtension{}, types.ErrKeyIndexOutOfRange
+		return nil, dtp.ExtensionE2EE{}, types.ErrKeyIndexOutOfRange
 	}
 
-	ext := datatrack.E2EEExtension{KeyIndex: uint8(encrypted.KeyIndex)}
-	if len(encrypted.IV) != len(ext.IV) {
-		return nil, datatrack.E2EEExtension{}, types.ErrIncorrectIVLength
+	var iv [dtp.ExtensionE2EEIVLength]byte
+	if len(encrypted.IV) != len(iv) {
+		return nil, dtp.ExtensionE2EE{}, types.ErrIncorrectIVLength
 	}
-	copy(ext.IV[:], encrypted.IV)
-	return encrypted.Ciphertext, ext, nil
+	copy(iv[:], encrypted.IV)
+	return encrypted.Ciphertext, *dtp.NewExtensionE2EE(uint8(encrypted.KeyIndex), iv), nil
 }
 
-func (c dataTrackCryptor) Decrypt(payload []byte, ext datatrack.E2EEExtension) ([]byte, error) {
+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:         ext.IV[:],
+		KeyIndex:   uint32(ext.KeyIndex()),
+		IV:         iv[:],
 	})
 }
 
 // dataTrackEncryptor returns the encryptor for the current session, or nil when data encryption
 // is not enabled.
-func (r *Room) dataTrackEncryptor() datatrack.Encryptor {
-	if r.engine.dataCryptor == nil {
+func (e *RTCEngine) dataTrackEncryptor() datatrack.Encryptor {
+	if e.dataCryptor == nil {
 		return nil
 	}
-	return dataTrackCryptor{cryptor: r.engine.dataCryptor}
+	return dataTrackCryptor{cryptor: e.dataCryptor}
 }
 
 // dataTrackDecryptor returns the decryptor for the current session, or nil when data encryption
 // is not enabled.
-func (r *Room) dataTrackDecryptor() datatrack.Decryptor {
-	if r.engine.dataCryptor == nil {
+func (e *RTCEngine) dataTrackDecryptor() datatrack.Decryptor {
+	if e.dataCryptor == nil {
 		return nil
 	}
-	return dataTrackCryptor{cryptor: r.engine.dataCryptor}
+	return dataTrackCryptor{cryptor: e.dataCryptor}
 }
diff --git a/internal/datatrack/remote_test.go b/internal/datatrack/remote_test.go
index c64cb62a..3ed2bdb3 100644
--- a/internal/datatrack/remote_test.go
+++ b/internal/datatrack/remote_test.go
@@ -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)
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
 }
 

From da218671711f74255874307848e38267ef210a8a Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Thu, 3 Sep 2026 16:24:17 -0700
Subject: [PATCH 4/6] Refuse to subscribe to encrypted data tracks without
 encryption

---
 internal/datatrack/errors.go      | 19 ++++++++++---------
 internal/datatrack/remote.go      | 21 ++++++++++++++-------
 internal/datatrack/remote_test.go | 25 ++++++++++++++++++++++++-
 3 files changed, 48 insertions(+), 17 deletions(-)

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/remote.go b/internal/datatrack/remote.go
index 18a4be7b..33064f9c 100644
--- a/internal/datatrack/remote.go
+++ b/internal/datatrack/remote.go
@@ -436,6 +436,10 @@ func (t *RemoteTrack) Subscribe(ctx context.Context, opts ...SubscribeOption) (*
 		t.mu.Unlock()
 		return nil, ErrUnpublished
 	}
+	if t.info.UsesE2EE && m.decryptor() == nil {
+		m.mu.Unlock()
+		return nil, ErrEncryptionDisabled
+	}
 	if t.subscription == subscriptionActive {
 		stream := t.addStreamLocked(options.BufferSize)
 		t.mu.Unlock()
@@ -563,14 +567,12 @@ func (t *RemoteTrack) activate(handle trackHandle) (previous trackHandle, replac
 
 	var decryptor Decryptor
 	if t.info.UsesE2EE {
-		if decryptor = t.manager.decryptor(); decryptor == nil {
-			t.manager.params.Logger.Warnw("subscribing to an end-to-end encrypted data track without encryption enabled, frames are delivered as received", nil, "sid", t.info.SID)
-		}
+		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 {
@@ -754,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.
@@ -775,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 3ed2bdb3..de778e9c 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)
@@ -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: &E2EEExtension{}}.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)
+}

From aafe79564b0b8f931f3c7a98039a679dca1fd8aa Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Tue, 22 Sep 2026 16:39:02 -0700
Subject: [PATCH 5/6] Fix build error

---
 internal/datatrack/remote.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/internal/datatrack/remote.go b/internal/datatrack/remote.go
index 33064f9c..158caaa4 100644
--- a/internal/datatrack/remote.go
+++ b/internal/datatrack/remote.go
@@ -436,8 +436,8 @@ func (t *RemoteTrack) Subscribe(ctx context.Context, opts ...SubscribeOption) (*
 		t.mu.Unlock()
 		return nil, ErrUnpublished
 	}
-	if t.info.UsesE2EE && m.decryptor() == nil {
-		m.mu.Unlock()
+	if t.info.UsesE2EE && t.manager.decryptor() == nil {
+		t.mu.Unlock()
 		return nil, ErrEncryptionDisabled
 	}
 	if t.subscription == subscriptionActive {

From 393c500861d24a06eeff9377ee45de34d2252338 Mon Sep 17 00:00:00 2001
From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com>
Date: Mon, 19 Oct 2026 12:40:42 -0700
Subject: [PATCH 6/6] Use upstream E2EE extension type in data track cryptor

---
 datatrackcryptor.go               | 1 +
 datatrackcryptor_test.go          | 2 +-
 internal/datatrack/remote_test.go | 2 +-
 3 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/datatrackcryptor.go b/datatrackcryptor.go
index 2a97325d..4f2771b7 100644
--- a/datatrackcryptor.go
+++ b/datatrackcryptor.go
@@ -18,6 +18,7 @@ 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"
diff --git a/datatrackcryptor_test.go b/datatrackcryptor_test.go
index 03059ffe..b7f9ad32 100644
--- a/datatrackcryptor_test.go
+++ b/datatrackcryptor_test.go
@@ -44,7 +44,7 @@ func TestDataTrackCryptorRoundTrip(t *testing.T) {
 
 	ciphertext, ext, err := c.Encrypt([]byte("hello encrypted world"))
 	require.NoError(t, err)
-	require.Equal(t, uint8(3), ext.KeyIndex)
+	require.Equal(t, uint8(3), ext.KeyIndex())
 	require.NotEqual(t, []byte("hello encrypted world"), ciphertext)
 
 	plaintext, err := c.Decrypt(ciphertext, ext)
diff --git a/internal/datatrack/remote_test.go b/internal/datatrack/remote_test.go
index de778e9c..ca95bafa 100644
--- a/internal/datatrack/remote_test.go
+++ b/internal/datatrack/remote_test.go
@@ -658,7 +658,7 @@ func TestRemotePipeline_DropsEncryptedFrameWithoutDecryptor(t *testing.T) {
 
 	header := testHeader()
 	FrameMarkerSingle.apply(&header)
-	Extensions{E2EE: &E2EEExtension{}}.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)