From b496c83afb47f1c263e03249ffc2a40b0b15ecea Mon Sep 17 00:00:00 2001 From: Kevin Caffrey Date: Fri, 7 Aug 2026 10:47:37 -0400 Subject: [PATCH 1/3] Add allocation-free API for routing RTCP In order to route RTCP packets, a caller must currently perform a full unmarshal to split a compound packet into individual feedbacks and then call DestinationSSRC on each parsed feedback. Many feedback messages, however, allocate in their Unmarshal methods. For example, TWCC allocates once per recv delta, or in other words, once for every packet sent to the remote. To avoid these allocations that scale per packet, a new UnmarshalRaw method is introduced to split a payload into a slice of unparsed RawPackets. RawPacket also gains a ParseDestinationSSRC method that does a wire parse similar to Unmarshal, but only pulls out SSRCs (matching the behavior of DestinationSSRC on the full unmarshalled feedback). Unlike Unmarshal, UnmarshalRaw (and the append variant) only validates the framing of a compound packet, in line with the recommendations in RFC 3550 section 6.1 and appendix A.2. The intended usage for this API is in the srtp package, where we currently unmarshal and remarshal each feedback. This will be replaced by an UnmarshalRaw + ParseDestinationSSRC. There will be a minor behavior difference in srtp using this new API on account of only validating framing rather than full packet contents, which is that previously one bad packet in a compound could prevent the entire batch from being routed. Now, as long as the framing is valid according to RFC 3550 (amended by RFC 5506 for reduced size packets), the packets will be forwarded as-is with no modification (as long as they can be parsed sufficiently to get the destination SSRC). I don't think this behavior change will be breaking. A test ensures that UnmarshalRaw + ParseDestinationSSRC exactly matches the behavior of Unmarshal + DestinationSSRC for every feedback type successfully parsed by Unmarshal, as well as a second test ensuring that the test covers every packet type that Unmarshal can handle. A fuzz test adds some extra safety on top. Allocations are pinned to zero with a third test. --- application_defined.go | 11 ++ extended_report.go | 46 +++++ full_intra_request.go | 15 ++ goodbye.go | 17 ++ raw_packet.go | 138 ++++++++++++++- raw_packet_test.go | 240 ++++++++++++++++++++++++++ receiver_estimated_maximum_bitrate.go | 27 +++ receiver_report.go | 24 +++ rfc8888.go | 32 ++++ sender_report.go | 29 ++++ source_description.go | 42 +++++ 11 files changed, 620 insertions(+), 1 deletion(-) diff --git a/application_defined.go b/application_defined.go index 86ae1f6..f55bcd7 100644 --- a/application_defined.go +++ b/application_defined.go @@ -110,6 +110,17 @@ func (a *ApplicationDefined) Unmarshal(rawPacket []byte) error { return nil } +func appendApplicationDefinedSSRCs(dst []uint32, header Header, pkt []byte) ([]uint32, error) { + if header.Type != TypeApplicationDefined { + return dst, errWrongType + } + if len(pkt) < 12 { + return dst, errPacketTooShort + } + + return append(dst, binary.BigEndian.Uint32(pkt[4:8])), nil +} + // MarshalSize returns the size of the packet once marshaled. func (a *ApplicationDefined) MarshalSize() int { dataLength := len(a.Data) diff --git a/extended_report.go b/extended_report.go index 9bf50e7..8cbad50 100644 --- a/extended_report.go +++ b/extended_report.go @@ -4,6 +4,7 @@ package rtcp import ( + "encoding/binary" "fmt" ) @@ -661,6 +662,51 @@ func (x *ExtendedReport) Unmarshal(b []byte) error { return nil } +const ( + xrHeaderLength = 4 + dlrrReportLength = 12 +) + +func appendXRSSRCs(dst []uint32, header Header, pkt []byte) ([]uint32, error) { + if header.Type != TypeExtendedReport { + return dst, errWrongType + } + if len(pkt) < headerLength+ssrcLength { + return dst, errPacketTooShort + } + // The sender's SSRC, then one SSRC per report block that refers to a + // source (the receiver reference time block does not); see + // ExtendedReport for the packet layout. + dst = append(dst, binary.BigEndian.Uint32(pkt[headerLength:])) + for offset := headerLength + ssrcLength; offset < len(pkt); { + if offset+xrHeaderLength > len(pkt) { + return dst, errPacketTooShort + } + // Each block begins with an XRHeader. + blockLen := (int(binary.BigEndian.Uint16(pkt[offset+2:])) + 1) * 4 + // A block whose declared length overruns the packet is truncated at + // the packet end, mirroring packetBuffer.split. + if offset+blockLen > len(pkt) { + blockLen = len(pkt) - offset + } + switch pkt[offset] { + case LossRLEReportBlockType, DuplicateRLEReportBlockType, PacketReceiptTimesReportBlockType, + StatisticsSummaryReportBlockType, VoIPMetricsReportBlockType: + if blockLen < xrHeaderLength+ssrcLength { + return dst, errPacketTooShort + } + dst = append(dst, binary.BigEndian.Uint32(pkt[offset+xrHeaderLength:])) + case DLRRReportBlockType: + for sub := offset + xrHeaderLength; sub+dlrrReportLength <= offset+blockLen; sub += dlrrReportLength { + dst = append(dst, binary.BigEndian.Uint32(pkt[sub:])) + } + } + offset += blockLen + } + + return dst, nil +} + // DestinationSSRC returns an array of SSRC values that this packet refers to. func (x *ExtendedReport) DestinationSSRC() []uint32 { ssrc := make([]uint32, 0, len(x.Reports)+1) diff --git a/full_intra_request.go b/full_intra_request.go index 2dd6655..b8cbfa4 100644 --- a/full_intra_request.go +++ b/full_intra_request.go @@ -86,6 +86,21 @@ func (p *FullIntraRequest) Unmarshal(rawPacket []byte) error { return nil } +func appendFIRSSRCs(dst []uint32, header Header, pkt []byte) ([]uint32, error) { + if header.Type != TypePayloadSpecificFeedback || header.Count != FormatFIR { + return dst, errWrongType + } + if 4*int(header.Length)-firOffset <= 0 || (4*int(header.Length))%8 != 0 { + return dst, errBadLength + } + for i := headerLength + firOffset; i < (headerLength + 4*int(header.Length)); i += 8 { + entry := pkt[i : i+8] + dst = append(dst, binary.BigEndian.Uint32(entry)) + } + + return dst, nil +} + // Header returns the Header associated with this packet. func (p *FullIntraRequest) Header() Header { return Header{ diff --git a/goodbye.go b/goodbye.go index f3a3324..5680e2b 100644 --- a/goodbye.go +++ b/goodbye.go @@ -121,6 +121,23 @@ func (g *Goodbye) Unmarshal(rawPacket []byte) error { return nil } +func appendGoodbyeSSRCs(dst []uint32, header Header, pkt []byte) ([]uint32, error) { + if header.Type != TypeGoodbye { + return dst, errWrongType + } + reasonOffset := int(headerLength + header.Count*ssrcLength) + if reasonOffset > len(pkt) { + return dst, errPacketTooShort + } + for i := 0; i < int(header.Count); i++ { + offset := headerLength + i*ssrcLength + + dst = append(dst, binary.BigEndian.Uint32(pkt[offset:])) + } + + return dst, nil +} + // Header returns the Header associated with this packet. func (g *Goodbye) Header() Header { return Header{ diff --git a/raw_packet.go b/raw_packet.go index 35828d4..0656ea0 100644 --- a/raw_packet.go +++ b/raw_packet.go @@ -3,7 +3,10 @@ package rtcp -import "fmt" +import ( + "encoding/binary" + "fmt" +) // RawPacket represents an unparsed RTCP packet. It's returned by Unmarshal when // a packet with an unknown type is encountered. @@ -51,3 +54,136 @@ func (r RawPacket) String() string { func (r RawPacket) MarshalSize() int { return len(r) } + +// UnmarshalRaw takes an entire udp datagram (which may consist of multiple +// RTCP packets) and returns the raw packets it contains, without unmarshaling +// their contents. +// +// The returned packets are subslices of data, not copies. +func UnmarshalRaw(data []byte) ([]RawPacket, error) { + return AppendRawPackets(nil, data) +} + +// AppendRawPackets appends the raw packets of the datagram data, as produced +// by UnmarshalRaw, to dst and returns the extended slice. On error dst is +// returned unchanged. +func AppendRawPackets(dst []RawPacket, data []byte) ([]RawPacket, error) { + orig := dst + found := 0 + for rest := data; len(rest) > 0; { + var header Header + if err := header.Unmarshal(rest); err != nil { + return orig, err + } + size := (int(header.Length) + 1) * 4 + if size > len(rest) { + return orig, errPacketTooShort + } + dst = append(dst, RawPacket(rest[:size:size])) + rest = rest[size:] + found++ + } + if found == 0 { + return orig, errInvalidHeader + } + + return dst, nil +} + +// ParseDestinationSSRC parses the destination SSRCs of the packet out of its +// raw bytes, appends them to dst, and returns the extended slice. It reports +// the SSRCs that Unmarshal followed by Packet.DestinationSSRC would for every +// packet type known to this package; unknown packet types have none. r must +// be a single RTCP packet, as produced by UnmarshalRaw. +// +// Only the structure needed to locate the SSRCs is validated; a packet +// accepted by this method may still fail a full Unmarshal. +// +// ParseDestinationSSRC will only allocate in order to resize dst as needed; +// the parsing itself is allocation-free. +// +//nolint:cyclop +func (r RawPacket) ParseDestinationSSRC(dst []uint32) ([]uint32, error) { + pkt := []byte(r) + var header Header + if err := header.Unmarshal(pkt); err != nil { + return dst, err + } + if size := (int(header.Length) + 1) * 4; size != len(pkt) { + return dst, errBadLength + } + + switch header.Type { + case TypeSenderReport: + return appendSenderReportSSRCs(dst, header, pkt) + case TypeReceiverReport: + return appendReceiverReportSSRCs(dst, header, pkt) + case TypeSourceDescription: + return appendSourceDescriptionSSRCs(dst, header, pkt) + case TypeGoodbye: + return appendGoodbyeSSRCs(dst, header, pkt) + case TypeApplicationDefined: + return appendApplicationDefinedSSRCs(dst, header, pkt) + case TypeTransportSpecificFeedback: + switch header.Count { + case FormatRRR: + var rrr RapidResynchronizationRequest + if err := rrr.Unmarshal(pkt); err != nil { + return dst, err + } + + return append(dst, rrr.MediaSSRC), nil + case FormatTLN, FormatTCC: + return appendMediaSSRC(dst, header, pkt) + case FormatCCFB: + return appendCCFBSSRCs(dst, header, pkt) + } + case TypePayloadSpecificFeedback: + switch header.Count { + case FormatPLI: + var pli PictureLossIndication + if err := pli.Unmarshal(pkt); err != nil { + return dst, err + } + + return append(dst, pli.MediaSSRC), nil + case FormatSLI: + return appendMediaSSRC(dst, header, pkt) + case FormatFIR: + return appendFIRSSRCs(dst, header, pkt) + case FormatREMB: + return appendREMBSSRCs(dst, header, pkt) + } + case TypeExtendedReport: + return appendXRSSRCs(dst, header, pkt) + } + + return dst, nil +} + +// appendMediaSSRC handles the feedback packets whose only destination is the +// media SSRC in the fixed part of the packet and whose full Unmarshal +// allocates (SLI, NACK, TWCC). They share the common packet format for +// feedback messages: +// +// https://tools.ietf.org/html/rfc4585#section-6.1 +// +// 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=2|P| FMT | PT | length | +// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +// | SSRC of packet sender | +// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +// | SSRC of media source | +// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +// : Feedback Control Information (FCI) : +// : : +// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +func appendMediaSSRC(dst []uint32, _ Header, pkt []byte) ([]uint32, error) { + if len(pkt) < (headerLength + (ssrcLength * 2)) { + return dst, errPacketTooShort + } + + return append(dst, binary.BigEndian.Uint32(pkt[headerLength+ssrcLength:])), nil +} diff --git a/raw_packet_test.go b/raw_packet_test.go index 11f4b2e..aaa5f0a 100644 --- a/raw_packet_test.go +++ b/raw_packet_test.go @@ -4,6 +4,7 @@ package rtcp import ( + "reflect" "testing" "github.com/stretchr/testify/assert" @@ -59,3 +60,242 @@ func TestRawPacketRoundTrip(t *testing.T) { assert.Equalf(t, test.Packet, decoded, "Unmarshal %q", test.Name) } } + +func destinationSSRCCorpus(tb testing.TB) [][]byte { + tb.Helper() + + packets := [][]Packet{ + {&SenderReport{SSRC: 0xAAAA0001, Reports: []ReceptionReport{{SSRC: 0xBBBB0001}, {SSRC: 0xBBBB0002}}}}, + {&ReceiverReport{SSRC: 0xAAAA0002, Reports: []ReceptionReport{{SSRC: 0xBBBB0001}, {SSRC: 0xBBBB0003}}}}, + {&SourceDescription{Chunks: []SourceDescriptionChunk{ + {Source: 0xCCCC0001, Items: []SourceDescriptionItem{{Type: SDESCNAME, Text: "alice@example"}}}, + {Source: 0xCCCC0002, Items: []SourceDescriptionItem{{Type: SDESCNAME, Text: "bob"}}}, + }}}, + {&Goodbye{Sources: []uint32{0xDDDD0001, 0xDDDD0002}, Reason: "shutdown"}}, + {&ApplicationDefined{SSRC: 0xEEEE0001, Name: "TEST", Data: []byte{1, 2, 3, 4}}}, + {&PictureLossIndication{SenderSSRC: 0xAAAA0001, MediaSSRC: 0xBBBB0001}}, + {&SliceLossIndication{ + SenderSSRC: 0xAAAA0001, + MediaSSRC: 0xBBBB0002, + SLI: []SLIEntry{{First: 1, Number: 2, Picture: 3}}, + }}, + {&FullIntraRequest{ + SenderSSRC: 0xAAAA0001, + MediaSSRC: 0xBBBB0001, + FIR: []FIREntry{{SSRC: 0xBBBB0001, SequenceNumber: 1}, {SSRC: 0xBBBB0002, SequenceNumber: 2}}, + }}, + {&ReceiverEstimatedMaximumBitrate{SenderSSRC: 0xAAAA0001, Bitrate: 1e6, SSRCs: []uint32{0xBBBB0001, 0xBBBB0002}}}, + {&ReceiverEstimatedMaximumBitrate{SenderSSRC: 0xAAAA0001, Bitrate: 1e6}}, + {&TransportLayerNack{ + SenderSSRC: 0xAAAA0001, + MediaSSRC: 0xBBBB0001, + Nacks: []NackPair{{PacketID: 42, LostPackets: 2}}, + }}, + {&RapidResynchronizationRequest{SenderSSRC: 0xAAAA0001, MediaSSRC: 0xBBBB0001}}, + {&CCFeedbackReport{SenderSSRC: 0xAAAA0001, ReportBlocks: []CCFeedbackReportBlock{ + {MediaSSRC: 0xBBBB0001, BeginSequence: 1, MetricBlocks: []CCFeedbackMetricBlock{ + {Received: true, ArrivalTimeOffset: 12}, {Received: true, ArrivalTimeOffset: 34}, {Received: false}, + }}, + {MediaSSRC: 0xBBBB0002, BeginSequence: 7, MetricBlocks: []CCFeedbackMetricBlock{ + {Received: true, ArrivalTimeOffset: 56}, + }}, + }, ReportTimestamp: 0x01020304}}, + {&ExtendedReport{SenderSSRC: 0xAAAA0001, Reports: []ReportBlock{ + // An even number of chunks keeps the block 32-bit aligned. + &LossRLEReportBlock{SSRC: 0xBBBB0001, Chunks: []Chunk{0x4006, 0x0006}}, + &DLRRReportBlock{Reports: []DLRRReport{ + {SSRC: 0xBBBB0002, LastRR: 1, DLRR: 2}, {SSRC: 0xBBBB0003, LastRR: 3, DLRR: 4}, + }}, + &ReceiverReferenceTimeReportBlock{NTPTimestamp: 0x0102030405060708}, + &DuplicateRLEReportBlock{SSRC: 0xBBBB0004, Chunks: []Chunk{0x4006, 0x0006}}, + &PacketReceiptTimesReportBlock{SSRC: 0xBBBB0005}, + &StatisticsSummaryReportBlock{SSRC: 0xBBBB0006}, + &VoIPMetricsReportBlock{SSRC: 0xBBBB0007}, + }}}, + } + + var datagrams [][]byte + for _, pkts := range packets { + raw, err := Marshal(pkts) + assert.NoErrorf(tb, err, "marshal corpus packets %T", pkts[0]) + datagrams = append(datagrams, raw) + } + + // RR + TWCC, the typical receiver->sender feedback compound. The TWCC + // packet carries one run-length chunk of two received small deltas. + twcc := []byte{ + 0x8f, 0xcd, 0x00, 0x05, // V=2, FMT=15, PT=205, length=5 + 0x11, 0x11, 0x11, 0x11, // sender SSRC + 0x22, 0x22, 0x22, 0x22, // media SSRC + 0x03, 0xe8, 0x00, 0x02, // base sequence 1000, packet status count 2 + 0x01, 0x23, 0x45, 0x03, // reference time, fb pkt count + 0x20, 0x02, 0x04, 0x08, // run-length chunk, recv deltas 1ms and 2ms + } + rrAndTWCC := append(append([]byte{}, datagrams[1]...), twcc...) + datagrams = append(datagrams, rrAndTWCC) + + // An unknown packet type (195), parsed as a RawPacket with no + // destinations, alone and inside a compound. + unknown := []byte{0x80, 195, 0x00, 0x01, 0xDE, 0xAD, 0xBE, 0xEF} + datagrams = append(datagrams, unknown) + + // An unknown feedback format (2) within a known packet type (205), also + // parsed as a RawPacket. + datagrams = append(datagrams, []byte{0x82, 0xCD, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00}) + datagrams = append(datagrams, append(append([]byte{}, datagrams[0]...), unknown...)) + + datagrams = append(datagrams, realPacket()) + + // A datagram Unmarshal rejects would be silently skipped by the tests + // comparing against it. + for i, datagram := range datagrams { + _, err := Unmarshal(datagram) + assert.NoErrorf(tb, err, "corpus datagram %d does not unmarshal", i) + } + + return datagrams +} + +// ParseDestinationSSRC must report the same destination SSRCs per packet as +// a full unmarshal does. +func TestParseDestinationSSRCMatchesUnmarshal(t *testing.T) { + for di, datagram := range destinationSSRCCorpus(t) { + raws, err := UnmarshalRaw(datagram) + assert.NoErrorf(t, err, "datagram %d", di) + + pkts, err := Unmarshal(datagram) + assert.NoErrorf(t, err, "datagram %d", di) + if !assert.Len(t, raws, len(pkts)) { + continue + } + + for i, pkt := range pkts { + got, err := raws[i].ParseDestinationSSRC(nil) + assert.NoErrorf(t, err, "datagram %d packet %d (%T)", di, i, pkt) + assert.ElementsMatchf(t, pkt.DestinationSSRC(), got, "destination SSRCs of datagram %d packet %d (%T)", di, i, pkt) + } + } +} + +// Every packet type the Unmarshal factory can produce, RawPacket included, +// must appear in the corpus, so TestParseDestinationSSRCMatchesUnmarshal +// cannot silently lose coverage when a new packet type or format is added. +func TestRawPacketCorpusCoverage(t *testing.T) { + corpusTypes := map[reflect.Type]bool{} + for _, datagram := range destinationSSRCCorpus(t) { + pkts, err := Unmarshal(datagram) + assert.NoError(t, err) + for _, pkt := range pkts { + corpusTypes[reflect.TypeOf(pkt)] = true + } + } + + for packetType := 0; packetType <= 0xFF; packetType++ { + formats := 1 + if PacketType(packetType) == TypeTransportSpecificFeedback || PacketType(packetType) == TypePayloadSpecificFeedback { + formats = 32 + } + for format := range formats { + // The factory picks the concrete type before unmarshaling, so a + // bare header is enough to learn the mapping. + //nolint:gosec // G115, both values fit in a byte + pkt, _, _ := unmarshal([]byte{0x80 | byte(format), byte(packetType), 0x00, 0x00}) + typ := reflect.TypeOf(pkt) + assert.Truef(t, corpusTypes[typ], "PT=%d FMT=%d maps to %v, not covered by the corpus", packetType, format, typ) + } + } +} + +func FuzzParseDestinationSSRC(f *testing.F) { + for _, datagram := range destinationSSRCCorpus(f) { + f.Add(datagram) + } + f.Fuzz(func(t *testing.T, data []byte) { + raws, rawErr := UnmarshalRaw(data) + + pkts, err := Unmarshal(data) + if err != nil { + // ParseDestinationSSRC must not panic or read out of bounds no + // matter how malformed the packet. + for _, raw := range raws { + _, _ = raw.ParseDestinationSSRC(nil) + } + + return + } + + // If Unmarshal parses the packet without error, UnmarshalRaw must + // as well, returning the same number of packets. + if !assert.NoError(t, rawErr) || !assert.Len(t, raws, len(pkts)) { + return + } + + // Each packet must produce the same destination SSRCs as what + // Unmarshal would. + for i, pkt := range pkts { + ssrcs, err := raws[i].ParseDestinationSSRC(nil) + assert.NoErrorf(t, err, "packet %d (%T)", i, pkt) + assert.ElementsMatchf(t, pkt.DestinationSSRC(), ssrcs, "packet %d (%T)", i, pkt) + } + }) +} + +func TestUnmarshalRawInvalid(t *testing.T) { + for _, test := range []struct { + Name string + Data []byte + }{ + {Name: "empty", Data: []byte{}}, + {Name: "truncated header", Data: []byte{0x80, 0xC8}}, + {Name: "bad version", Data: []byte{0x40, 0xC8, 0x00, 0x00}}, + {Name: "length overruns datagram", Data: []byte{0x80, 0xC8, 0x00, 0x02, 0x00, 0x00, 0x00, 0x00}}, + {Name: "trailing partial packet", Data: []byte{0x80, 0xC8, 0x00, 0x00, 0x80}}, + } { + t.Run(test.Name, func(t *testing.T) { + _, err := UnmarshalRaw(test.Data) + assert.Error(t, err) + }) + } +} + +// Returned packets are views into the datagram; appending to one must grow a +// copy, not overwrite the packets that follow it in the shared backing. +func TestAppendRawPacketsCapacity(t *testing.T) { + datagram := []byte{ + 0x80, 0xc3, 0x00, 0x00, // unknown packet type 195, length=0 + 0x80, 0xc4, 0x00, 0x00, // unknown packet type 196, length=0 + } + raws, err := UnmarshalRaw(datagram) + assert.NoError(t, err) + assert.Len(t, raws, 2) + + second := append([]byte{}, raws[1]...) + raws[0] = append(raws[0], 0xFF) + assert.Equal(t, second, []byte(raws[1])) +} + +// The append APIs exist to be allocation-free with reused buffers; pin that +// to zero for every packet type in the corpus. +func TestRawPacketAllocations(t *testing.T) { + var raws []RawPacket + var ssrcs []uint32 + for i, datagram := range destinationSSRCCorpus(t) { + // Reach steady-state buffer capacity and verify the datagram parses + // before measuring. + var err error + raws, err = AppendRawPackets(raws[:0], datagram) + assert.NoErrorf(t, err, "corpus datagram %d", i) + for _, raw := range raws { + ssrcs, err = raw.ParseDestinationSSRC(ssrcs[:0]) + assert.NoErrorf(t, err, "corpus datagram %d", i) + } + + allocs := testing.AllocsPerRun(100, func() { + raws, _ = AppendRawPackets(raws[:0], datagram) + for _, raw := range raws { + ssrcs, _ = raw.ParseDestinationSSRC(ssrcs[:0]) + } + }) + assert.Zerof(t, allocs, "corpus datagram %d allocates", i) + } +} diff --git a/receiver_estimated_maximum_bitrate.go b/receiver_estimated_maximum_bitrate.go index 683db1c..9aa92e9 100644 --- a/receiver_estimated_maximum_bitrate.go +++ b/receiver_estimated_maximum_bitrate.go @@ -256,6 +256,33 @@ func (p *ReceiverEstimatedMaximumBitrate) Unmarshal(buf []byte) (err error) { return nil } +func appendREMBSSRCs(dst []uint32, header Header, pkt []byte) ([]uint32, error) { + if header.Type != TypePayloadSpecificFeedback || header.Count != FormatREMB { + return dst, errWrongType + } + if len(pkt) < 20 { + return dst, errPacketTooShort + } + + // The unique identifier is the discriminator: FMT 15 is generic + // application layer feedback, of which REMB is only one kind. + if !bytes.Equal(pkt[12:16], []byte{'R', 'E', 'M', 'B'}) { + return dst, errMissingREMBidentifier + } + + // The byte after the 'REMB' unique identifier is the number of SSRC + // entries at the end. + num := int(pkt[16]) + if len(pkt) != 20+4*num { + return dst, errSSRCNumAndLengthMismatch + } + for n := 20; n < len(pkt); n += 4 { + dst = append(dst, binary.BigEndian.Uint32(pkt[n:n+4])) + } + + return dst, nil +} + // Header returns the Header associated with this packet. func (p *ReceiverEstimatedMaximumBitrate) Header() Header { return Header{ diff --git a/receiver_report.go b/receiver_report.go index 7e08a2a..c20cdb3 100644 --- a/receiver_report.go +++ b/receiver_report.go @@ -159,6 +159,30 @@ func (r *ReceiverReport) Unmarshal(rawPacket []byte) error { return nil } +func appendReceiverReportSSRCs(dst []uint32, header Header, pkt []byte) ([]uint32, error) { + if header.Type != TypeReceiverReport { + return dst, errWrongType + } + if len(pkt) < (headerLength + ssrcLength) { + return dst, errPacketTooShort + } + + reports := 0 + for i := rrReportOffset; i < len(pkt) && reports < int(header.Count); i += receptionReportLength { + var rr ReceptionReport + if err := rr.Unmarshal(pkt[i:]); err != nil { + return dst, err + } + dst = append(dst, rr.SSRC) + reports++ + } + if reports != int(header.Count) { + return dst, errInvalidHeader + } + + return dst, nil +} + // MarshalSize returns the size of the packet once marshaled. func (r *ReceiverReport) MarshalSize() int { repsLength := 0 diff --git a/rfc8888.go b/rfc8888.go index 2719551..69959a6 100644 --- a/rfc8888.go +++ b/rfc8888.go @@ -211,6 +211,38 @@ func (b *CCFeedbackReport) Unmarshal(rawPacket []byte) error { return nil } +func appendCCFBSSRCs(dst []uint32, header Header, pkt []byte) ([]uint32, error) { + if header.Type != TypeTransportSpecificFeedback || header.Count != FormatCCFB { + return dst, errWrongType + } + if len(pkt) < headerLength+ssrcLength+reportTimestampLength { + return dst, errPacketTooShort + } + + // Each block's contents are bounded by the end of the packet rather than + // the start of the timestamp, as in CCFeedbackReport.Unmarshal. + reportTimestampOffset := len(pkt) - reportTimestampLength + offset := reportBlockOffset + for offset < reportTimestampOffset { + block := pkt[offset:] + if len(block) < reportsOffset { + return dst, errReportBlockLength + } + dst = append(dst, binary.BigEndian.Uint32(block[:beginSequenceOffset])) + + numReports := int(binary.BigEndian.Uint16(block[numReportsOffset:])) + if len(block) < reportsOffset+numReports*2 { + return dst, errIncorrectNumReports + } + if numReports%2 != 0 { + numReports++ + } + offset += reportsOffset + 2*numReports + } + + return dst, nil +} + const ( ssrcOffset = 0 beginSequenceOffset = 4 diff --git a/sender_report.go b/sender_report.go index 9e5ed57..46baf75 100644 --- a/sender_report.go +++ b/sender_report.go @@ -219,6 +219,35 @@ func (r *SenderReport) Unmarshal(rawPacket []byte) error { return nil } +func appendSenderReportSSRCs(dst []uint32, header Header, pkt []byte) ([]uint32, error) { + if header.Type != TypeSenderReport { + return dst, errWrongType + } + // Reception report SSRCs, then the sender's own SSRC. + if len(pkt) < (headerLength + srHeaderLength) { + return dst, errPacketTooShort + } + + packetBody := pkt[headerLength:] + offset := srReportOffset + for i := 0; i < int(header.Count); i++ { + rrEnd := offset + receptionReportLength + if rrEnd > len(packetBody) { + return dst, errPacketTooShort + } + rrBody := packetBody[offset : offset+receptionReportLength] + offset = rrEnd + + var rr ReceptionReport + if err := rr.Unmarshal(rrBody); err != nil { + return dst, err + } + dst = append(dst, rr.SSRC) + } + + return append(dst, binary.BigEndian.Uint32(packetBody[srSSRCOffset:])), nil +} + // DestinationSSRC returns an array of SSRC values that this packet refers to. func (r *SenderReport) DestinationSSRC() []uint32 { out := make([]uint32, len(r.Reports)+1) diff --git a/source_description.go b/source_description.go index 951bf8f..844b7e0 100644 --- a/source_description.go +++ b/source_description.go @@ -173,6 +173,48 @@ func (s *SourceDescription) Unmarshal(rawPacket []byte) error { return nil } +func appendSourceDescriptionSSRCs(dst []uint32, header Header, pkt []byte) ([]uint32, error) { + if header.Type != TypeSourceDescription { + return dst, errWrongType + } + numChunks := 0 + for i := headerLength; i < len(pkt); numChunks++ { + chunk := pkt[i:] + if len(chunk) < (sdesSourceLen + sdesTypeLen) { + return dst, errPacketTooShort + } + dst = append(dst, binary.BigEndian.Uint32(chunk)) + + chunkLen := -1 + for itemOffset := sdesSourceLen; itemOffset < len(chunk); { + item := chunk[itemOffset:] + if pktType := SDESType(item[sdesTypeOffset]); pktType == SDESEnd { + chunkLen = itemOffset + sdesTypeLen + chunkLen += getPadding(chunkLen) + + break + } + if len(item) < (sdesTypeLen + sdesOctetCountLen) { + return dst, errPacketTooShort + } + octetCount := int(item[sdesOctetCountOffset]) + if sdesTextOffset+octetCount > len(item) { + return dst, errPacketTooShort + } + itemOffset += sdesTypeLen + sdesOctetCountLen + octetCount + } + if chunkLen < 0 { + return dst, errPacketTooShort + } + i += chunkLen + } + if numChunks != int(header.Count) { + return dst, errInvalidHeader + } + + return dst, nil +} + // MarshalSize returns the size of the packet once marshaled. func (s *SourceDescription) MarshalSize() int { chunksLength := 0 From 71f9a0d86b99014098905df85a1229a534ee88c9 Mon Sep 17 00:00:00 2001 From: Kevin Caffrey Date: Fri, 7 Aug 2026 12:55:23 -0400 Subject: [PATCH 2/3] Add tests for parse failures Make sure various malformed packets return an error from ParseDestinationSSRC --- raw_packet_test.go | 55 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/raw_packet_test.go b/raw_packet_test.go index aaa5f0a..c20648d 100644 --- a/raw_packet_test.go +++ b/raw_packet_test.go @@ -274,6 +274,61 @@ func TestAppendRawPacketsCapacity(t *testing.T) { assert.Equal(t, second, []byte(raws[1])) } +// One packet for every error return reachable through the method; the corpus +// covers only valid packets. +func TestParseDestinationSSRCInvalid(t *testing.T) { + for _, test := range []struct { + Name string + Data []byte + }{ + {Name: "bad version", Data: []byte{0x00, 0xc8, 0x00, 0x00}}, + {Name: "length field mismatch", Data: []byte{0x80, 0xc8, 0x00, 0x01}}, + {Name: "sender report too short", Data: []byte{0x80, 0xc8, 0x00, 0x01, 0, 0, 0, 0}}, + {Name: "sender report truncated report", Data: append([]byte{0x81, 0xc8, 0x00, 0x06}, make([]byte, 24)...)}, + {Name: "receiver report too short", Data: []byte{0x80, 0xc9, 0x00, 0x00}}, + {Name: "receiver report truncated report", Data: append([]byte{0x81, 0xc9, 0x00, 0x02}, make([]byte, 8)...)}, + {Name: "receiver report count mismatch", Data: []byte{0x81, 0xc9, 0x00, 0x01, 0, 0, 0, 0}}, + {Name: "source description short chunk", Data: []byte{0x81, 0xca, 0x00, 0x01, 0, 0, 0, 0}}, + {Name: "source description item overrun", Data: []byte{0x81, 0xca, 0x00, 0x02, 0, 0, 0, 0, 0x01, 0xc8, 0x00, 0x00}}, + { + Name: "source description missing terminator", + Data: []byte{0x81, 0xca, 0x00, 0x02, 0, 0, 0, 0, 0x01, 0x02, 0x00, 0x00}, + }, + { + Name: "source description short item", + Data: []byte{0x81, 0xca, 0x00, 0x02, 0, 0, 0, 0, 0x01, 0x01, 0x00, 0x01}, + }, + {Name: "source description count mismatch", Data: []byte{0x82, 0xca, 0x00, 0x02, 0, 0, 0, 0, 0, 0, 0, 0}}, + {Name: "goodbye sources overrun", Data: []byte{0x82, 0xcb, 0x00, 0x01, 0, 0, 0, 0}}, + {Name: "application defined too short", Data: []byte{0x80, 0xcc, 0x00, 0x01, 0, 0, 0, 0}}, + {Name: "nack too short", Data: []byte{0x81, 0xcd, 0x00, 0x01, 0, 0, 0, 0}}, + {Name: "rrr too short", Data: []byte{0x85, 0xcd, 0x00, 0x01, 0, 0, 0, 0}}, + {Name: "pli too short", Data: []byte{0x81, 0xce, 0x00, 0x01, 0, 0, 0, 0}}, + {Name: "fir without entries", Data: append([]byte{0x84, 0xce, 0x00, 0x02}, make([]byte, 8)...)}, + {Name: "remb too short", Data: append([]byte{0x8f, 0xce, 0x00, 0x03}, make([]byte, 12)...)}, + { + Name: "remb wrong identifier", + Data: append(append([]byte{0x8f, 0xce, 0x00, 0x04}, make([]byte, 8)...), 'O', 'P', 'U', 'S', 0, 0, 0, 0), + }, + { + Name: "remb ssrc count mismatch", + Data: append(append([]byte{0x8f, 0xce, 0x00, 0x04}, make([]byte, 8)...), 'R', 'E', 'M', 'B', 0x05, 0, 0, 0), + }, + {Name: "ccfb too short", Data: []byte{0x8b, 0xcd, 0x00, 0x01, 0, 0, 0, 0}}, + {Name: "ccfb metric overrun", Data: append(append([]byte{0x8b, 0xcd, 0x00, 0x03}, make([]byte, 10)...), 0xff, 0xff)}, + {Name: "xr too short", Data: []byte{0x80, 0xcf, 0x00, 0x00}}, + { + Name: "xr short rle block", + Data: append(append([]byte{0x80, 0xcf, 0x00, 0x02}, make([]byte, 4)...), 0x01, 0x00, 0x00, 0x00), + }, + } { + t.Run(test.Name, func(t *testing.T) { + _, err := RawPacket(test.Data).ParseDestinationSSRC(nil) + assert.Error(t, err) + }) + } +} + // The append APIs exist to be allocation-free with reused buffers; pin that // to zero for every packet type in the corpus. func TestRawPacketAllocations(t *testing.T) { From 0abdf9b27e28ff4f9e38de6d537bcd813207b50e Mon Sep 17 00:00:00 2001 From: Kevin Caffrey Date: Fri, 7 Aug 2026 13:06:30 -0400 Subject: [PATCH 3/3] Add defensive guard in source description packets --- source_description.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/source_description.go b/source_description.go index 844b7e0..a626c05 100644 --- a/source_description.go +++ b/source_description.go @@ -173,6 +173,7 @@ func (s *SourceDescription) Unmarshal(rawPacket []byte) error { return nil } +//nolint:cyclop func appendSourceDescriptionSSRCs(dst []uint32, header Header, pkt []byte) ([]uint32, error) { if header.Type != TypeSourceDescription { return dst, errWrongType @@ -203,7 +204,7 @@ func appendSourceDescriptionSSRCs(dst []uint32, header Header, pkt []byte) ([]ui } itemOffset += sdesTypeLen + sdesOctetCountLen + octetCount } - if chunkLen < 0 { + if chunkLen < 0 || chunkLen > len(chunk) { return dst, errPacketTooShort } i += chunkLen