diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 47113a07a..4531f0dcd 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -358,9 +358,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra bitrates = int(ti.Layers[layer].GetBitrate()) } - if t.IsSimulcast() { - t.MediaTrackReceiver.SetLayerSsrc(mime, track.RID(), uint32(track.SSRC())) - } + t.MediaTrackReceiver.SetLayerSsrc(mime, track.RID(), uint32(track.SSRC())) buff.Bind(receiver.GetParameters(), track.Codec().RTPCodecCapability, bitrates) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 82e5920c1..f8dc63e06 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -316,19 +316,25 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { return } - if rtpPacket.Version != 2 || (b.payloadType != 0 && rtpPacket.PayloadType != b.payloadType) { + if err = utils.ValidateRTPPacket(&rtpPacket, b.payloadType, b.mediaSSRC); err != nil { b.logger.Warnw( - "invalid RTP packet", nil, + "validating RTP packet failed", err, "version", rtpPacket.Version, - "sn", rtpPacket.SequenceNumber, - "timestamp", rtpPacket.Timestamp, - "payloadSize", len(rtpPacket.Payload), + "padding", rtpPacket.Padding, + "marker", rtpPacket.Marker, + "expectedPayloadType", b.payloadType, "payloadType", rtpPacket.PayloadType, + "sequenceNumber", rtpPacket.SequenceNumber, + "timestamp", rtpPacket.Timestamp, + "expectedSSRC", b.mediaSSRC, "ssrc", rtpPacket.SSRC, + "numExtensions", len(rtpPacket.Extensions), + "payloadSize", len(rtpPacket.Payload), "rtpStats", b.rtpStats, "snRangeMap", b.snRangeMap, ) - // TODO-REMOVE-AFTER-DEBUG + b.Unlock() + return } now := time.Now() diff --git a/pkg/sfu/buffer/buffer_test.go b/pkg/sfu/buffer/buffer_test.go index e97ad1db4..e5ce7fdc4 100644 --- a/pkg/sfu/buffer/buffer_test.go +++ b/pkg/sfu/buffer/buffer_test.go @@ -44,7 +44,7 @@ var opusCodec = webrtc.RTPCodecParameters{ MimeType: "audio/opus", ClockRate: 48000, }, - PayloadType: 96, + PayloadType: 111, } func TestNack(t *testing.T) { @@ -81,7 +81,13 @@ func TestNack(t *testing.T) { time.Sleep(500 * time.Millisecond) // even a long wait should not exceed max retries } pkt := rtp.Packet{ - Header: rtp.Header{SequenceNumber: uint16(i), Timestamp: uint32(i)}, + Header: rtp.Header{ + Version: 2, + PayloadType: 96, + SequenceNumber: uint16(i), + Timestamp: uint32(i), + SSRC: 123, + }, Payload: []byte{0xff, 0xff, 0xff, 0xfd, 0xb4, 0x9f, 0x94, 0x1}, } b, err := pkt.Marshal() @@ -140,7 +146,13 @@ func TestNack(t *testing.T) { time.Sleep(500 * time.Millisecond) // even a long wait should not exceed max retries } pkt := rtp.Packet{ - Header: rtp.Header{SequenceNumber: uint16(i + 65533), Timestamp: uint32(i)}, + Header: rtp.Header{ + Version: 2, + PayloadType: 96, + SequenceNumber: uint16(i + 65533), + Timestamp: uint32(i), + SSRC: 123, + }, Payload: []byte{0xff, 0xff, 0xff, 0xfd, 0xb4, 0x9f, 0x94, 0x1}, } b, err := pkt.Marshal() @@ -166,23 +178,35 @@ func TestNewBuffer(t *testing.T) { var TestPackets = []*rtp.Packet{ { Header: rtp.Header{ + Version: 2, + PayloadType: 96, SequenceNumber: 65533, + SSRC: 123, }, }, { Header: rtp.Header{ + Version: 2, + PayloadType: 96, SequenceNumber: 65534, + SSRC: 123, }, Payload: []byte{1}, }, { Header: rtp.Header{ + Version: 2, + PayloadType: 96, SequenceNumber: 2, + SSRC: 123, }, }, { Header: rtp.Header{ + Version: 2, + PayloadType: 96, SequenceNumber: 65535, + SSRC: 123, }, }, } @@ -232,7 +256,13 @@ func TestFractionLostReport(t *testing.T) { }, opusCodec.RTPCodecCapability, 0) for i := 0; i < 15; i++ { pkt := rtp.Packet{ - Header: rtp.Header{SequenceNumber: uint16(i), Timestamp: uint32(i)}, + Header: rtp.Header{ + Version: 2, + PayloadType: 111, + SequenceNumber: uint16(i), + Timestamp: uint32(i), + SSRC: 123, + }, Payload: []byte{0xff, 0xff, 0xff, 0xfd, 0xb4, 0x9f, 0x94, 0x1}, } b, err := pkt.Marshal() @@ -264,7 +294,13 @@ func TestFractionLostReport(t *testing.T) { }, opusCodec.RTPCodecCapability, 0) for i := 0; i < 15; i++ { pkt := rtp.Packet{ - Header: rtp.Header{SequenceNumber: uint16(i), Timestamp: uint32(i)}, + Header: rtp.Header{ + Version: 2, + PayloadType: 111, + SequenceNumber: uint16(i), + Timestamp: uint32(i), + SSRC: 123, + }, Payload: []byte{0xff, 0xff, 0xff, 0xfd, 0xb4, 0x9f, 0x94, 0x1}, } b, err := pkt.Marshal() diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index 890b70257..d106dd9e5 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -264,7 +264,7 @@ func (s *StreamTrackerManager) RemoveAllTrackers() { s.trackers[layer] = nil } s.availableLayers = make([]int32, 0) - s.maxExpectedLayerFromTrackInfo() + s.maxExpectedLayerFromTrackInfoLocked() s.paused = false ddTracker := s.ddTracker s.ddTracker = nil @@ -530,6 +530,13 @@ func (s *StreamTrackerManager) removeAvailableLayer(layer int32) { } func (s *StreamTrackerManager) maxExpectedLayerFromTrackInfo() { + s.lock.Lock() + defer s.lock.Unlock() + + s.maxExpectedLayerFromTrackInfoLocked() +} + +func (s *StreamTrackerManager) maxExpectedLayerFromTrackInfoLocked() { s.maxExpectedLayer = buffer.InvalidLayerSpatial ti := s.trackInfo.Load() if ti != nil { diff --git a/pkg/sfu/utils/helpers.go b/pkg/sfu/utils/helpers.go index 476050416..f3f12161e 100644 --- a/pkg/sfu/utils/helpers.go +++ b/pkg/sfu/utils/helpers.go @@ -15,9 +15,11 @@ package utils import ( + "errors" "strings" "github.com/pion/interceptor" + "github.com/pion/rtp" "github.com/pion/webrtc/v3" ) @@ -51,3 +53,20 @@ func GetHeaderExtensionID(extensions []interceptor.RTPHeaderExtension, extension } return 0 } + +// ValidateRTPPacket checks for a valid RTP packet and returns an error if fields are incorrect +func ValidateRTPPacket(pkt *rtp.Packet, expectedPayloadType uint8, expectedSSRC uint32) error { + if pkt.Version != 2 { + return errors.New("invalid RTP version") + } + + if expectedPayloadType != 0 && pkt.PayloadType != expectedPayloadType { + return errors.New("invalid RTP payload type") + } + + if expectedSSRC != 0 && pkt.SSRC != expectedSSRC { + return errors.New("invalid RTP SSRC") + } + + return nil +}