Limit rtp payload size (#364)

integration_test.go

@@ -24,12 +24,14 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v3"
 	"github.com/pion/webrtc/v3/pkg/media"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
 
 	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/server-sdk-go/pkg/interceptor"
 )
 
 // The integration test of the SDK. can't run this test standalone, should be run with `mage test`
@@ -313,3 +315,36 @@ func TestSubscribeMutedTrack(t *testing.T) {
 	pub.Disconnect()
 	sub.Disconnect()
 }
+
+func TestLimitPayloadSize(t *testing.T) {
+	pub, err := createAgent(t.Name(), nil, "publisher")
+	require.NoError(t, err)
+
+	videoTrackName := "video_of_pub1"
+
+	localTrack, err := webrtc.NewTrackLocalStaticRTP(webrtc.RTPCodecCapability{
+		MimeType:  webrtc.MimeTypeVP8,
+		ClockRate: 90000,
+	}, videoTrackName, videoTrackName)
+	require.NoError(t, err)
+
+	_, err = pub.LocalParticipant.PublishTrack(localTrack, &TrackPublicationOptions{Name: videoTrackName})
+	require.NoError(t, err)
+
+	// wait for track to be published
+	time.Sleep(500 * time.Millisecond)
+
+	rtpPkt := rtp.Packet{
+		Header: rtp.Header{
+			SequenceNumber: 1,
+			Timestamp:      1,
+		},
+		Payload: make([]byte, interceptor.MaxPayloadSize),
+	}
+	require.NoError(t, localTrack.WriteRTP(&rtpPkt))
+	rtpPkt.SequenceNumber++
+	rtpPkt.Payload = make([]byte, interceptor.MaxPayloadSize+1)
+	require.ErrorIs(t, localTrack.WriteRTP(&rtpPkt), interceptor.ErrPayloadSizeTooLarge)
+
+	pub.Disconnect()
+}

pkg/interceptor/limitsizeinteceptor.go

@@ -0,0 +1,38 @@
+package interceptor
+
+import (
+	"fmt"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/rtp"
+)
+
+const (
+	MaxPayloadSize = 1200
+)
+
+var ErrPayloadSizeTooLarge = fmt.Errorf("packetization payload size should not greater than %d bytes", MaxPayloadSize)
+
+type LimitSizeInterceptorFactory struct {
+}
+
+func NewLimitSizeInterceptorFactory() *LimitSizeInterceptorFactory {
+	return &LimitSizeInterceptorFactory{}
+}
+
+func (l *LimitSizeInterceptorFactory) NewInterceptor(id string) (interceptor.Interceptor, error) {
+	return &LimitSizeInterceptor{}, nil
+}
+
+type LimitSizeInterceptor struct {
+	interceptor.NoOp
+}
+
+func (l *LimitSizeInterceptor) BindLocalStream(stream *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
+	return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
+		if len(payload) > MaxPayloadSize {
+			return 0, ErrPayloadSizeTooLarge
+		}
+		return writer.Write(header, payload, attributes)
+	})
+}

transport.go

@@ -117,6 +117,8 @@ func NewPCTransport(params PCTransportParams) (*PCTransport, error) {
 		return nil, err
 	}
 
+	i.Add(sdkinterceptor.NewLimitSizeInterceptorFactory())
+
 	se := webrtc.SettingEngine{}
 	se.SetSRTPProtectionProfiles(dtls.SRTP_AEAD_AES_128_GCM, dtls.SRTP_AES128_CM_HMAC_SHA1_80)
 	se.SetDTLSRetransmissionInterval(dtlsRetransmissionInterval)