diff --git a/.changeset/eighty-waves-run.md b/.changeset/eighty-waves-run.md new file mode 100644 index 000000000..fb2edad19 --- /dev/null +++ b/.changeset/eighty-waves-run.md @@ -0,0 +1,6 @@ +--- +"github.com/livekit/protocol": patch +"@livekit/protocol": patch +--- + +Move data track packet serialization from livekit package diff --git a/datatrack/datatracktest/testutils.go b/datatrack/datatracktest/testutils.go new file mode 100644 index 000000000..ea29b5fa5 --- /dev/null +++ b/datatrack/datatracktest/testutils.go @@ -0,0 +1,77 @@ +// 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 datatracktest provides helpers for generating data track packets in tests. +package datatracktest + +import ( + "math/rand" + "time" + + "github.com/livekit/protocol/datatrack" +) + +func GenerateRawDataPackets(handle uint16, seqNum uint16, frameNum uint16, numFrames int, frameSize int, frameDuration time.Duration) [][]byte { + if seqNum == 0 { + seqNum = uint16(rand.Intn(256) + 1) + } + if frameNum == 0 { + frameNum = uint16(rand.Intn(256) + 1) + } + timestamp := uint32(rand.Intn(1024)) + + packetsPerFrame := (frameSize + 255) / 256 // using 256 bytes of payload per packet + if packetsPerFrame == 0 { + return nil + } + numPackets := packetsPerFrame * numFrames + rawPackets := make([][]byte, 0, numPackets) + for range numFrames { + remainingSize := frameSize + for packetIdx := range packetsPerFrame { + payloadSize := min(remainingSize, 256) + payload := make([]byte, payloadSize) + for i := range len(payload) { + payload[i] = byte(255 - i) + } + packet := &datatrack.Packet{ + Header: datatrack.Header{ + Version: 0, + IsStartOfFrame: packetIdx == 0, + IsFinalOfFrame: packetIdx == packetsPerFrame-1, + Handle: handle, + SequenceNumber: seqNum, + FrameNumber: frameNum, + Timestamp: timestamp, + }, + Payload: payload, + } + if extParticipantSid, err := datatrack.NewExtensionParticipantSid("test_participant"); err == nil { + if ext, err := extParticipantSid.Marshal(); err == nil { + packet.AddExtension(ext) + } + } + rawPacket, err := packet.Marshal() + if err == nil { + rawPackets = append(rawPackets, rawPacket) + } + seqNum++ + remainingSize -= payloadSize + } + frameNum++ + timestamp += uint32(90000 * frameDuration.Seconds()) + } + + return rawPackets +} diff --git a/datatrack/extension_participant_sid.go b/datatrack/extension_participant_sid.go new file mode 100644 index 000000000..acac8384e --- /dev/null +++ b/datatrack/extension_participant_sid.go @@ -0,0 +1,59 @@ +// 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 datatrack + +import ( + "errors" + + "github.com/livekit/protocol/livekit" +) + +type ExtensionParticipantSid struct { + participantID livekit.ParticipantID +} + +func NewExtensionParticipantSid(participantID livekit.ParticipantID) (*ExtensionParticipantSid, error) { + if len(participantID) >= 256 { + return nil, errors.New("participantID too long") + } + + return &ExtensionParticipantSid{participantID}, nil +} + +func (e *ExtensionParticipantSid) ParticipantID() livekit.ParticipantID { + return e.participantID +} + +func (e *ExtensionParticipantSid) Marshal() (Extension, error) { + data := make([]byte, len(e.participantID)) + copy(data, e.participantID) + return Extension{ + id: uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID), + data: data, + }, nil +} + +func (e *ExtensionParticipantSid) Unmarshal(ext Extension) error { + if ext.id != uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID) { + return errors.New("invalid extension ID") + } + + if len(ext.data) == 0 { + return errors.New("empty extension data") + } + + e.participantID = livekit.ParticipantID(ext.data) + return nil +} diff --git a/datatrack/extension_participant_sid_test.go b/datatrack/extension_participant_sid_test.go new file mode 100644 index 000000000..b2a58d04c --- /dev/null +++ b/datatrack/extension_participant_sid_test.go @@ -0,0 +1,46 @@ +// 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 datatrack + +import ( + "testing" + + "github.com/livekit/protocol/livekit" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestExtensionParticipantSid(t *testing.T) { + longTestParticipantID := livekit.ParticipantID(make([]byte, 256)) + _, err := NewExtensionParticipantSid(longTestParticipantID) + require.Error(t, err) + + testParticipantID := livekit.ParticipantID("test") + extParticipantSid, err := NewExtensionParticipantSid(testParticipantID) + require.NoError(t, err) + + expectedExt := Extension{ + id: uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID), + data: []byte{'t', 'e', 's', 't'}, + } + ext, err := extParticipantSid.Marshal() + require.NoError(t, err) + require.Equal(t, expectedExt, ext) + + var unmarshaled ExtensionParticipantSid + err = unmarshaled.Unmarshal(ext) + require.NoError(t, err) + assert.Equal(t, testParticipantID, unmarshaled.ParticipantID()) +} diff --git a/datatrack/packet.go b/datatrack/packet.go new file mode 100644 index 000000000..9fbea968f --- /dev/null +++ b/datatrack/packet.go @@ -0,0 +1,322 @@ +// 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 datatrack + +import ( + "encoding/binary" + "errors" + "fmt" +) + +var ( + ErrHeaderSizeInsufficient = errors.New("data track packet header size insufficient") + ErrBufferSizeInsufficient = errors.New("data track packet buffer size insufficient") + ErrExtensionSizeInsufficient = errors.New("data track packet extension size insufficient") + ErrExtensionNotFound = errors.New("data track packet extension not found") + ErrExtensionSizeTooBig = errors.New("extension size is too big") +) + +const ( + headerLength = 12 + + versionShift = 5 + versionMask = (1 << 3) - 1 + + startOfFrameShift = 4 + startOfFrameMask = (1 << 1) - 1 + + finalOfFrameShift = 3 + finalOfFrameMask = (1 << 1) - 1 + + extensionsShift = 2 + extensionsMask = (1 << 1) - 1 + + handleOffset = 2 + handleLength = 2 + + seqNumOffset = 4 + seqNumLength = 2 + + frameNumOffset = 6 + frameNumLength = 2 + + timestampOffset = 8 + timestampLength = 4 + + extensionsSizeOffset = headerLength + extensionsSizeLength = 2 + + extensionIDLength = 1 + extensionSizeLength = 1 +) + +type Extension struct { + id uint8 + data []byte +} + +func NewExtension(id uint8, data []byte) Extension { + return Extension{id: id, data: data} +} + +func (e Extension) ID() uint8 { + return e.id +} + +func (e Extension) Data() []byte { + return e.data +} + +type Header struct { + Version uint8 + IsStartOfFrame bool + IsFinalOfFrame bool + HasExtensions bool + Handle uint16 + SequenceNumber uint16 + FrameNumber uint16 + Timestamp uint32 + ExtensionsSize uint16 + Extensions []Extension +} + +/* + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + ┆* 0 1 2 3 + ┆* 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + ┆* |V |S|F|X| reserved | handle | + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + ┆* | sequence number | frame number | + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + ┆* | timestamp | + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + |* Extensions Size if X=1 | Extensions... | + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Each extension + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + ┆* 0 1 2 3 + ┆* 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + ┆* | Extension ID | Extension size| Extension payload | + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + End of all extensions + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + |* padded to 4 byte boundary if aggregate of `Extensions Size` | + |* field and all extensions do not end on a 4 byte boundary | + ┆* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +*/ + +func (h *Header) Unmarshal(buf []byte) (int, error) { + if len(buf) < headerLength { + return 0, fmt.Errorf("%w: %d < %d", ErrHeaderSizeInsufficient, len(buf), headerLength) + } + + hdrSize := headerLength + h.Version = buf[0] >> versionShift & versionMask + h.IsStartOfFrame = (buf[0] >> startOfFrameShift & startOfFrameMask) > 0 + h.IsFinalOfFrame = (buf[0] >> finalOfFrameShift & finalOfFrameMask) > 0 + h.HasExtensions = (buf[0] >> extensionsShift & extensionsMask) > 0 + + h.Handle = binary.BigEndian.Uint16(buf[handleOffset : handleOffset+handleLength]) + h.SequenceNumber = binary.BigEndian.Uint16(buf[seqNumOffset : seqNumOffset+seqNumLength]) + h.FrameNumber = binary.BigEndian.Uint16(buf[frameNumOffset : frameNumOffset+frameNumLength]) + h.Timestamp = binary.BigEndian.Uint32(buf[timestampOffset : timestampOffset+timestampLength]) + + if h.HasExtensions { + if len(buf) < extensionsSizeOffset+extensionsSizeLength { + return 0, fmt.Errorf("%w: %d < %d", ErrHeaderSizeInsufficient, len(buf), extensionsSizeOffset+extensionsSizeLength) + } + extensionsSize := (int(binary.BigEndian.Uint16(buf[extensionsSizeOffset:extensionsSizeOffset+extensionsSizeLength]))+1)*4 - extensionsSizeLength + hdrSize += extensionsSizeLength + + extensionHeaderSize := extensionIDLength + extensionSizeLength + remainingSize := extensionsSize + idx := extensionsSizeOffset + extensionsSizeLength + for remainingSize != 0 { + // read extension header + if len(buf[idx:]) < extensionIDLength || remainingSize < extensionIDLength { + return 0, fmt.Errorf("%w: %d/%d < %d", ErrExtensionSizeInsufficient, remainingSize, len(buf[idx:]), extensionIDLength) + } + id := buf[idx] + if id == 0 { + // end of extensions, padding has started + if len(buf[idx:]) < remainingSize { + return 0, fmt.Errorf("%w: %d/%d < %d", ErrExtensionSizeInsufficient, remainingSize, len(buf[idx:]), remainingSize) + } + hdrSize += remainingSize + break + } + + if len(buf[idx+1:]) < extensionSizeLength || remainingSize < extensionSizeLength { + return 0, fmt.Errorf("%w: %d/%d < %d", ErrExtensionSizeInsufficient, remainingSize, len(buf[idx:]), extensionSizeLength) + } + size := int(buf[idx+1]) + + remainingSize -= extensionHeaderSize + idx += extensionHeaderSize + hdrSize += extensionHeaderSize + + // read extension data + if len(buf[idx:]) < size || remainingSize < size { + return 0, fmt.Errorf("%w: %d/%d < %d", ErrExtensionSizeInsufficient, remainingSize, len(buf[idx:]), size) + } + h.Extensions = append(h.Extensions, Extension{id: id, data: buf[idx : idx+size]}) + + remainingSize -= size + idx += size + hdrSize += size + } + h.ExtensionsSize = uint16(extensionsSize - remainingSize) + } + + return hdrSize, nil +} + +func (h *Header) MarshalSize() int { + extensionsSize := 0 + if h.HasExtensions { + extensionsSize += extensionsSizeLength + for _, ext := range h.Extensions { + extensionsSize += len(ext.data) + extensionIDLength + extensionSizeLength + } + } + + return headerLength + (extensionsSize+3)/4*4 +} + +func (h *Header) MarshalTo(buf []byte) (int, error) { + if len(buf) < headerLength { + return 0, fmt.Errorf("%w: %d < %d", ErrHeaderSizeInsufficient, len(buf), headerLength) + } + + hdrSize := headerLength + buf[0] = h.Version << versionShift + if h.IsStartOfFrame { + buf[0] |= (1 << startOfFrameShift) + } + if h.IsFinalOfFrame { + buf[0] |= (1 << finalOfFrameShift) + } + if h.HasExtensions { + buf[0] |= (1 << extensionsShift) + } + + binary.BigEndian.PutUint16(buf[handleOffset:handleOffset+handleLength], h.Handle) + binary.BigEndian.PutUint16(buf[seqNumOffset:seqNumOffset+seqNumLength], h.SequenceNumber) + binary.BigEndian.PutUint16(buf[frameNumOffset:frameNumOffset+frameNumLength], h.FrameNumber) + binary.BigEndian.PutUint32(buf[timestampOffset:timestampOffset+timestampLength], h.Timestamp) + + if h.HasExtensions { + extensionsSize := (extensionsSizeLength + h.ExtensionsSize + 3) / 4 * 4 + binary.BigEndian.PutUint16(buf[extensionsSizeOffset:extensionsSizeOffset+extensionsSizeLength], (extensionsSize/4)-1) + hdrSize += extensionsSizeLength + + addedSize := 0 + idx := extensionsSizeOffset + extensionsSizeLength + for _, ext := range h.Extensions { + buf[idx] = ext.id + if len(ext.data) > 255 { + return 0, fmt.Errorf("%w: %d > 255", ErrExtensionSizeTooBig, len(ext.data)) + } + buf[idx+extensionIDLength] = byte(len(ext.data)) + copy(buf[idx+extensionIDLength+extensionSizeLength:], ext.data) + + extSize := len(ext.data) + extensionIDLength + extensionSizeLength + idx += extSize + hdrSize += extSize + addedSize += extSize + } + + paddingSize := extensionsSize - extensionsSizeLength - uint16(addedSize) + for i := range paddingSize { + buf[idx+int(i)] = 0 + } + idx += int(paddingSize) + hdrSize += int(paddingSize) + } + + return hdrSize, nil +} + +func (h *Header) AddExtension(ext Extension) { + for i, existingExt := range h.Extensions { + if existingExt.id == ext.id { + h.ExtensionsSize -= uint16(len(existingExt.data) + extensionIDLength + extensionSizeLength) + h.Extensions[i].data = ext.data + h.ExtensionsSize += uint16(len(h.Extensions[i].data) + extensionIDLength + extensionSizeLength) + return + } + } + + h.Extensions = append(h.Extensions, ext) + h.ExtensionsSize += uint16(len(ext.data) + extensionIDLength + extensionSizeLength) + h.HasExtensions = true +} + +func (h *Header) GetExtension(id uint8) (Extension, error) { + for _, ext := range h.Extensions { + if ext.id == id { + return ext, nil + } + } + return Extension{}, fmt.Errorf("%w, id: %d", ErrExtensionNotFound, id) +} + +// ---------------------------------------------------- + +type Packet struct { + Header + Payload []byte +} + +func (p *Packet) Unmarshal(buf []byte) error { + hdrSize, err := p.Header.Unmarshal(buf) + if err != nil { + return err + } + if hdrSize > len(buf) { + return fmt.Errorf("%w: %d < %d", ErrBufferSizeInsufficient, len(buf), hdrSize) + } + + p.Payload = buf[hdrSize:] + return nil +} + +func (p *Packet) Marshal() ([]byte, error) { + buf := make([]byte, p.Header.MarshalSize()+len(p.Payload)) + if err := p.MarshalTo(buf); err != nil { + return nil, err + } + + return buf, nil +} + +func (p *Packet) MarshalTo(buf []byte) error { + size := p.Header.MarshalSize() + len(p.Payload) + if len(buf) < size { + return fmt.Errorf("%w: %d < %d", ErrBufferSizeInsufficient, len(buf), size) + } + + hdrSize, err := p.Header.MarshalTo(buf) + if err != nil { + return err + } + + copy(buf[hdrSize:], p.Payload) + return nil +} diff --git a/datatrack/packet_test.go b/datatrack/packet_test.go new file mode 100644 index 000000000..0eba4cfe8 --- /dev/null +++ b/datatrack/packet_test.go @@ -0,0 +1,294 @@ +// 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 datatrack + +import ( + "testing" + + "github.com/livekit/protocol/livekit" + "github.com/stretchr/testify/require" +) + +func TestPacket(t *testing.T) { + t.Run("without extension", func(t *testing.T) { + payload := make([]byte, 6) + for i := range len(payload) { + payload[i] = byte(255 - i) + } + packet := &Packet{ + Header: Header{ + Version: 0, + IsStartOfFrame: true, + IsFinalOfFrame: true, + Handle: 3333, + SequenceNumber: 6666, + FrameNumber: 9999, + Timestamp: 0xdeadbeef, + }, + Payload: payload, + } + rawPacket, err := packet.Marshal() + require.NoError(t, err) + + expectedRawPacket := []byte{ + 0x18, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f, + 0xde, 0xad, 0xbe, 0xef, 0xff, 0xfe, 0xfd, 0xfc, + 0xfb, 0xfa, + } + require.Equal(t, expectedRawPacket, rawPacket) + + var unmarshaled Packet + err = unmarshaled.Unmarshal(rawPacket) + require.NoError(t, err) + require.Equal(t, packet, &unmarshaled) + }) + + t.Run("with extension", func(t *testing.T) { + payload := make([]byte, 4) + for i := range len(payload) { + payload[i] = byte(255 - i) + } + packet := &Packet{ + Header: Header{ + Version: 0, + IsStartOfFrame: true, + IsFinalOfFrame: false, + Handle: 3333, + SequenceNumber: 6666, + FrameNumber: 9999, + Timestamp: 0xdeadbeef, + }, + Payload: payload, + } + if extParticipantSid, err := NewExtensionParticipantSid("test_participant"); err == nil { + if ext, err := extParticipantSid.Marshal(); err == nil { + packet.AddExtension(ext) + } + } + rawPacket, err := packet.Marshal() + require.NoError(t, err) + + expectedRawPacket := []byte{ + 0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f, + 0xde, 0xad, 0xbe, 0xef, 0x00, 0x04, 0x01, 0x10, + 0x74, 0x65, 0x73, 0x74, 0x5f, 0x70, 0x61, 0x72, + 0x74, 0x69, 0x63, 0x69, 0x70, 0x61, 0x6e, 0x74, + 0xff, 0xfe, 0xfd, 0xfc, + } + require.Equal(t, expectedRawPacket, rawPacket) + + var unmarshaled Packet + err = unmarshaled.Unmarshal(rawPacket) + require.NoError(t, err) + require.Equal(t, packet, &unmarshaled) + + ext, err := unmarshaled.GetExtension(uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID)) + require.NoError(t, err) + + var extParticipantSid ExtensionParticipantSid + require.NoError(t, extParticipantSid.Unmarshal(ext)) + require.Equal(t, livekit.ParticipantID("test_participant"), extParticipantSid.ParticipantID()) + }) + + t.Run("with extension padding", func(t *testing.T) { + payload := make([]byte, 4) + for i := range len(payload) { + payload[i] = byte(255 - i) + } + packet := &Packet{ + Header: Header{ + Version: 0, + IsStartOfFrame: true, + IsFinalOfFrame: false, + Handle: 3333, + SequenceNumber: 6666, + FrameNumber: 9999, + Timestamp: 0xdeadbeef, + }, + Payload: payload, + } + if extParticipantSid, err := NewExtensionParticipantSid("participant"); err == nil { + if ext, err := extParticipantSid.Marshal(); err == nil { + packet.AddExtension(ext) + } + } + rawPacket, err := packet.Marshal() + require.NoError(t, err) + + expectedRawPacket := []byte{ + 0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f, + 0xde, 0xad, 0xbe, 0xef, 0x00, 0x03, 0x01, 0x0b, + 0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70, + 0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc, + } + require.Equal(t, expectedRawPacket, rawPacket) + + var unmarshaled Packet + err = unmarshaled.Unmarshal(rawPacket) + require.NoError(t, err) + require.Equal(t, packet, &unmarshaled) + + ext, err := unmarshaled.GetExtension(uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID)) + require.NoError(t, err) + + var extParticipantSid ExtensionParticipantSid + require.NoError(t, extParticipantSid.Unmarshal(ext)) + require.Equal(t, livekit.ParticipantID("participant"), extParticipantSid.ParticipantID()) + }) + + t.Run("replace extension", func(t *testing.T) { + payload := make([]byte, 4) + for i := range len(payload) { + payload[i] = byte(255 - i) + } + packet := &Packet{ + Header: Header{ + Version: 0, + IsStartOfFrame: true, + IsFinalOfFrame: false, + Handle: 3333, + SequenceNumber: 6666, + FrameNumber: 9999, + Timestamp: 0xdeadbeef, + }, + Payload: payload, + } + if extParticipantSid, err := NewExtensionParticipantSid("participant"); err == nil { + if ext, err := extParticipantSid.Marshal(); err == nil { + packet.AddExtension(ext) + } + } + rawPacket, err := packet.Marshal() + require.NoError(t, err) + + expectedRawPacket := []byte{ + 0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f, + 0xde, 0xad, 0xbe, 0xef, 0x00, 0x03, 0x01, 0x0b, + 0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70, + 0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc, + } + require.Equal(t, expectedRawPacket, rawPacket) + + // replace existing extension ID and ensure that marshalled packet is updated + if extParticipantSid, err := NewExtensionParticipantSid("test_participant"); err == nil { + if ext, err := extParticipantSid.Marshal(); err == nil { + packet.AddExtension(ext) + } + } + rawPacket, err = packet.Marshal() + require.NoError(t, err) + + expectedRawPacket = []byte{ + 0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f, + 0xde, 0xad, 0xbe, 0xef, 0x00, 0x04, 0x01, 0x10, + 0x74, 0x65, 0x73, 0x74, 0x5f, 0x70, 0x61, 0x72, + 0x74, 0x69, 0x63, 0x69, 0x70, 0x61, 0x6e, 0x74, + 0xff, 0xfe, 0xfd, 0xfc, + } + require.Equal(t, expectedRawPacket, rawPacket) + + var unmarshaled Packet + err = unmarshaled.Unmarshal(rawPacket) + require.NoError(t, err) + require.Equal(t, packet, &unmarshaled) + + ext, err := unmarshaled.GetExtension(uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID)) + require.NoError(t, err) + + var extParticipantSid ExtensionParticipantSid + require.NoError(t, extParticipantSid.Unmarshal(ext)) + require.Equal(t, livekit.ParticipantID("test_participant"), extParticipantSid.ParticipantID()) + }) + + t.Run("bad packet", func(t *testing.T) { + var unmarshaled Packet + // extensions size too small + badPacket := []byte{ + 0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f, + 0xde, 0xad, 0xbe, 0xef, 0x00, 0x02, 0x01, 0x0b, + 0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70, + 0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc, + } + err := unmarshaled.Unmarshal(badPacket) + require.Error(t, err) + + // get an invalid extension id + badPacket = []byte{ + 0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f, + 0xde, 0xad, 0xbe, 0xef, 0x00, 0x03, 0x02, 0x0b, + 0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70, + 0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc, + } + err = unmarshaled.Unmarshal(badPacket) + require.NoError(t, err) + _, err = unmarshaled.GetExtension(uint8(livekit.DataTrackExtensionID_DTEI_PARTICIPANT_SID)) + require.Error(t, err) + + // extension payload size bigger than payload + badPacket = []byte{ + 0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f, + 0xde, 0xad, 0xbe, 0xef, 0x00, 0x03, 0x01, 0x0d, + 0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70, + 0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc, + } + err = unmarshaled.Unmarshal(badPacket) + require.Error(t, err) + + // extension payload size smaller than payload + badPacket = []byte{ + 0x14, 0x00, 0x0d, 0x05, 0x1a, 0x0a, 0x27, 0x0f, + 0xde, 0xad, 0xbe, 0xef, 0x00, 0x03, 0x01, 0x07, + 0x70, 0x61, 0x72, 0x74, 0x69, 0x63, 0x69, 0x70, + 0x61, 0x6e, 0x74, 0x00, 0xff, 0xfe, 0xfd, 0xfc, + } + err = unmarshaled.Unmarshal(badPacket) + require.Error(t, err) + }) + + t.Run("oversized extension padding does not panic", func(t *testing.T) { + var unmarshaled Packet + // HasExtensions set, extensionsSize describes more bytes than present, + // terminated by a 0x00 padding id -> hdrSize would exceed len(buf) + badPacket := []byte{ + 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + } + err := unmarshaled.Unmarshal(badPacket) + require.Error(t, err) + }) + + t.Run("extensions size wraparound does not panic", func(t *testing.T) { + var unmarshaled Packet + // 0xFFFF extensions-size field wraps (raw+1)*4 uint16 arithmetic to a huge + // remainingSize; the 0x00 padding id must not push hdrSize past len(buf) + badPacket := []byte{ + 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0x00, + } + err := unmarshaled.Unmarshal(badPacket) + require.Error(t, err) + }) + + t.Run("truncated extensions size field does not panic", func(t *testing.T) { + var unmarshaled Packet + // HasExtensions set but buffer too short to hold the extensionsSize field + badPacket := []byte{ + 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, + } + err := unmarshaled.Unmarshal(badPacket) + require.Error(t, err) + }) +}