Detect MTU overflow.

go.mod

@@ -21,7 +21,8 @@ require (
 	github.com/pion/interceptor v0.1.37
 	github.com/pion/rtp v1.8.9
 	github.com/pion/sdp/v3 v3.0.9
-	github.com/pion/webrtc/v4 v4.0.4
+	github.com/pion/srtp/v3 v3.0.4
+	github.com/pion/webrtc/v4 v4.0.5
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.20.5
 	github.com/sirupsen/logrus v1.9.3
@@ -96,7 +97,6 @@ require (
 	github.com/pion/randutil v0.1.0 // indirect
 	github.com/pion/rtcp v1.2.14 // indirect
 	github.com/pion/sctp v1.8.34 // indirect
-	github.com/pion/srtp/v3 v3.0.4 // indirect
 	github.com/pion/stun/v3 v3.0.0 // indirect
 	github.com/pion/transport/v3 v3.0.7 // indirect
 	github.com/pion/turn/v4 v4.0.0 // indirect

go.sum

@@ -192,8 +192,8 @@ github.com/pion/transport/v3 v3.0.7 h1:iRbMH05BzSNwhILHoBoAPxoB9xQgOaJk+591KC9P1
 github.com/pion/transport/v3 v3.0.7/go.mod h1:YleKiTZ4vqNxVwh77Z0zytYi7rXHl7j6uPLGhhz9rwo=
 github.com/pion/turn/v4 v4.0.0 h1:qxplo3Rxa9Yg1xXDxxH8xaqcyGUtbHYw4QSCvmFWvhM=
 github.com/pion/turn/v4 v4.0.0/go.mod h1:MuPDkm15nYSklKpN8vWJ9W2M0PlyQZqYt1McGuxG7mA=
-github.com/pion/webrtc/v4 v4.0.4 h1:X+gkoBLKDsR6FliKKQ/VXGBjnMR3yOPcyXEPt3z7Ep0=
-github.com/pion/webrtc/v4 v4.0.4/go.mod h1:LvP8Np5b/sM0uyJIcUPvJcCvhtjHxJwzh2H2PYzE6cQ=
+github.com/pion/webrtc/v4 v4.0.5 h1:8cVPojcv3cQTwVga2vF1rzCNvkiEimnYdCCG7yF317I=
+github.com/pion/webrtc/v4 v4.0.5/go.mod h1:LvP8Np5b/sM0uyJIcUPvJcCvhtjHxJwzh2H2PYzE6cQ=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

pkg/media/rtp/conn.go

@@ -23,11 +23,14 @@ import (
 
 	"github.com/frostbyte73/core"
 	"github.com/pion/rtp"
+
+	"github.com/livekit/protocol/logger"
 )
 
 var _ Writer = (*Conn)(nil)
 
 type ConnConfig struct {
+	Log                 logger.Logger
 	MediaTimeoutInitial time.Duration
 	MediaTimeout        time.Duration
 	TimeoutCallback     func()
@@ -41,14 +44,18 @@ func NewConnWith(conn UDPConn, conf *ConnConfig) *Conn {
 	if conf == nil {
 		conf = &ConnConfig{}
 	}
+	if conf.Log == nil {
+		conf.Log = logger.GetLogger()
+	}
 	if conf.MediaTimeoutInitial <= 0 {
 		conf.MediaTimeoutInitial = 30 * time.Second
 	}
 	if conf.MediaTimeout <= 0 {
 		conf.MediaTimeout = 15 * time.Second
 	}
 	c := &Conn{
-		readBuf:        make([]byte, 1500), // MTU
+		log:            conf.Log,
+		readBuf:        make([]byte, MTUSize+1), // larger buffer to detect overflow
 		received:       make(chan struct{}),
 		conn:           conn,
 		timeout:        conf.MediaTimeout,
@@ -69,6 +76,7 @@ type UDPConn interface {
 }
 
 type Conn struct {
+	log            logger.Logger
 	wmu            sync.Mutex
 	conn           UDPConn
 	closed         core.Fuse
@@ -153,13 +161,21 @@ func (c *Conn) ListenAndServe(portMin, portMax int, listenAddr string) error {
 
 func (c *Conn) readLoop() {
 	conn, buf := c.conn, c.readBuf
+	overflow := false
 	var p rtp.Packet
 	for {
 		n, srcAddr, err := conn.ReadFromUDP(buf)
 		if err != nil {
 			return
 		}
 		c.dest.Store(srcAddr)
+		if n > MTUSize {
+			if !overflow {
+				overflow = true
+				c.log.Errorw("RTP packet is larger than MTU limit", nil)
+			}
+			continue // ignore partial messages
+		}
 
 		p = rtp.Packet{}
 		if err := p.Unmarshal(buf[:n]); err != nil {

pkg/media/rtp/session.go

@@ -22,9 +22,14 @@ import (
 
 	"github.com/frostbyte73/core"
 	"github.com/pion/rtp"
+
+	"github.com/livekit/protocol/logger"
 )
 
-const enableZeroCopy = true
+const (
+	enableZeroCopy = true
+	MTUSize        = 1500
+)
 
 type Session interface {
 	OpenWriteStream() (WriteStream, error)
@@ -42,21 +47,24 @@ type ReadStream interface {
 	ReadRTP(h *rtp.Header, payload []byte) (int, error)
 }
 
-func NewSession(conn net.Conn) Session {
+func NewSession(log logger.Logger, conn net.Conn) Session {
 	return &session{
+		log:    log,
 		conn:   conn,
 		w:      &writeStream{conn: conn},
 		bySSRC: make(map[uint32]*readStream),
+		rbuf:   make([]byte, MTUSize+1), // larger buffer to detect overflow
 	}
 }
 
 type session struct {
+	log    logger.Logger
 	conn   net.Conn
 	closed core.Fuse
 	w      *writeStream
 
 	rmu    sync.Mutex
-	rbuf   [1500]byte
+	rbuf   []byte
 	bySSRC map[uint32]*readStream
 }
 
@@ -67,11 +75,19 @@ func (s *session) OpenWriteStream() (WriteStream, error) {
 func (s *session) AcceptStream() (ReadStream, uint32, error) {
 	s.rmu.Lock()
 	defer s.rmu.Unlock()
+	overflow := false
 	for {
 		n, err := s.conn.Read(s.rbuf[:])
 		if err != nil {
 			return nil, 0, err
 		}
+		if n > MTUSize {
+			overflow = true
+			if !overflow {
+				s.log.Errorw("RTP packet is larger than MTU limit", nil)
+			}
+			continue // ignore partial messages
+		}
 		buf := s.rbuf[:n]
 		var p rtp.Packet
 		err = p.Unmarshal(buf)

pkg/media/srtp/srtp.go

@@ -20,8 +20,9 @@ import (
 	"net"
 
 	prtp "github.com/pion/rtp"
-	"github.com/pion/srtp/v2"
+	"github.com/pion/srtp/v3"
 
+	"github.com/livekit/protocol/logger"
 	"github.com/livekit/sip/pkg/media/rtp"
 )
 
@@ -96,16 +97,17 @@ type Profile struct {
 type Config = srtp.Config
 type SessionKeys = srtp.SessionKeys
 
-func NewSession(conn net.Conn, conf *Config) (rtp.Session, error) {
+func NewSession(log logger.Logger, conn net.Conn, conf *Config) (rtp.Session, error) {
 	s, err := srtp.NewSessionSRTP(conn, conf)
 	if err != nil {
 		return nil, err
 	}
-	return &session{s: s}, nil
+	return &session{log: log, s: s}, nil
 }
 
 type session struct {
-	s *srtp.SessionSRTP
+	log logger.Logger
+	s   *srtp.SessionSRTP
 }
 
 func (s *session) OpenWriteStream() (rtp.WriteStream, error) {

pkg/sip/media_port.go

@@ -251,9 +251,9 @@ func (p *MediaPort) SetConfig(c *MediaConf) error {
 		err  error
 	)
 	if c.Crypto != nil {
-		sess, err = srtp.NewSession(p.port, c.Crypto)
+		sess, err = srtp.NewSession(p.log, p.port, c.Crypto)
 	} else {
-		sess = rtp.NewSession(p.port)
+		sess = rtp.NewSession(p.log, p.port)
 	}
 	if err != nil {
 		return err
@@ -295,7 +295,8 @@ func (p *MediaPort) rtpLoop(sess rtp.Session) {
 }
 
 func (p *MediaPort) rtpReadLoop(r rtp.ReadStream) {
-	buf := make([]byte, 1500)
+	buf := make([]byte, rtp.MTUSize+1)
+	overflow := false
 	var h rtp.Header
 	for {
 		h = rtp.Header{}
@@ -306,6 +307,13 @@ func (p *MediaPort) rtpReadLoop(r rtp.ReadStream) {
 			p.log.Errorw("read RTP failed", err)
 			return
 		}
+		if n > rtp.MTUSize {
+			overflow = true
+			if !overflow {
+				p.log.Errorw("RTP packet is larger than MTU limit", nil)
+			}
+			continue // ignore partial messages
+		}
 
 		ptr := p.hnd.Load()
 		if ptr == nil {