Add recovery handlers for SDP parsing.

pkg/sip/inbound.go

@@ -341,7 +341,7 @@ func updateRemoteFromSDP(media *MediaPort, log logger.Logger, codecs *msdk.Codec
 	if len(body) == 0 || media == nil {
 		return
 	}
-	desc, err := sdp.ParseWith(codecs, body)
+	desc, err := parseWith(media.mon, codecs, body)
 	if err != nil {
 		log.Warnw("failed to parse re-INVITE SDP, RTP destination not updated", err)
 		return

pkg/sip/media_port.go

@@ -16,11 +16,13 @@ package sip
 
 import (
 	"errors"
+	"fmt"
 	"io"
 	"math"
 	"net"
 	"net/netip"
 	"os"
+	"runtime/debug"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -453,12 +455,12 @@ type MediaPort struct {
 	dtmfOutRTP   *rtp.Stream
 	dtmfOutAudio msdk.PCM16Writer
 
-	audioOutRTP       *rtp.Stream
-	audioOut          *msdk.SwitchWriter // LK PCM -> SIP RTP
-	audioIn           *msdk.SwitchWriter // SIP RTP -> LK PCM
-	audioInHandler    rtp.Handler        // for debug only
-	dtmfIn            atomic.Pointer[func(ev dtmf.Event)]
-	lastDTMFEvent atomic.Uint64 // composite (timestamp, event code) of last DTMF packet seen
+	audioOutRTP    *rtp.Stream
+	audioOut       *msdk.SwitchWriter // LK PCM -> SIP RTP
+	audioIn        *msdk.SwitchWriter // SIP RTP -> LK PCM
+	audioInHandler rtp.Handler        // for debug only
+	dtmfIn         atomic.Pointer[func(ev dtmf.Event)]
+	lastDTMFEvent  atomic.Uint64 // composite (timestamp, event code) of last DTMF packet seen
 }
 
 func (p *MediaPort) DisableOut() {
@@ -709,6 +711,35 @@ func (p *MediaPort) GetAudioWriter() msdk.PCM16Writer {
 	return p.audioOut
 }
 
+func parseSDPWithRecovery[T any](mon *stats.CallMonitor, fn func(*msdk.CodecSet, []byte) (*T, error), codecs *msdk.CodecSet, data []byte) (res *T, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			panicErr, ok := r.(error)
+			if !ok {
+				panicErr = fmt.Errorf("%v", r)
+			}
+			logger.GetLogger().Errorw("panic while parsing SDP", panicErr, "stacktrace", string(debug.Stack()))
+			if mon != nil {
+				mon.SDPParsePanic()
+			}
+			res, err = nil, errors.New("invalid SDP")
+		}
+	}()
+	return fn(codecs, data)
+}
+
+func parseWith(mon *stats.CallMonitor, codecs *msdk.CodecSet, data []byte) (*sdp.Description, error) {
+	return parseSDPWithRecovery(mon, sdp.ParseWith, codecs, data)
+}
+
+func parseOfferWith(mon *stats.CallMonitor, codecs *msdk.CodecSet, data []byte) (*sdp.Offer, error) {
+	return parseSDPWithRecovery(mon, sdp.ParseOfferWith, codecs, data)
+}
+
+func parseAnswerWith(mon *stats.CallMonitor, codecs *msdk.CodecSet, data []byte) (*sdp.Answer, error) {
+	return parseSDPWithRecovery(mon, sdp.ParseAnswerWith, codecs, data)
+}
+
 // NewOffer generates an SDP offer for the media.
 func (p *MediaPort) NewOffer(codecs *msdk.CodecSet, encrypted sdp.Encryption) (*sdp.Offer, error) {
 	return sdp.NewOfferWith(codecs, p.externalIP, p.Port(), encrypted)
@@ -717,7 +748,7 @@ func (p *MediaPort) NewOffer(codecs *msdk.CodecSet, encrypted sdp.Encryption) (*
 // SetAnswer decodes and applies SDP answer for offer from NewOffer.
 // SetConfig must be called with the decoded configuration.
 func (p *MediaPort) SetAnswer(offer *sdp.Offer, answerData []byte, codecs *msdk.CodecSet, enc sdp.Encryption) (*MediaConf, []byte, error) {
-	answer, err := sdp.ParseAnswerWith(codecs, answerData)
+	answer, err := parseAnswerWith(p.mon, codecs, answerData)
 	if err != nil {
 		return nil, nil, SDPError{Err: err}
 	}
@@ -734,7 +765,7 @@ func (p *MediaPort) SetAnswer(offer *sdp.Offer, answerData []byte, codecs *msdk.
 
 // SetOffer decodes the offer from another party and returns encoded answer. To accept the offer, call SetConfig.
 func (p *MediaPort) SetOffer(offerData []byte, codecs *msdk.CodecSet, enc sdp.Encryption) (*sdp.Answer, *MediaConf, error) {
-	offer, err := sdp.ParseOfferWith(codecs, offerData)
+	offer, err := parseOfferWith(p.mon, codecs, offerData)
 	if err != nil {
 		return nil, nil, SDPError{Err: err}
 	}

pkg/stats/monitor.go

@@ -76,6 +76,7 @@ type Monitor struct {
 	cpuLoad                  prometheus.Gauge
 	sdpSize                  *prometheus.HistogramVec
 	sdpParsed                *prometheus.CounterVec
+	sdpParsePanics           *prometheus.CounterVec
 	codecOffered             *prometheus.CounterVec
 	nodeAvailable            prometheus.GaugeFunc
 	transfersTotal           *prometheus.CounterVec
@@ -252,6 +253,14 @@ func (m *Monitor) Start(conf *config.Config) error {
 		ConstLabels: prometheus.Labels{"node_id": conf.NodeID},
 	}, []string{"dir", "provider"}))
 
+	m.sdpParsePanics = mustRegister(m, prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace:   "livekit",
+		Subsystem:   "sip",
+		Name:        "sdp_parse_panics_total",
+		Help:        "Total number of SDP parses that resulted in a recovered panic",
+		ConstLabels: prometheus.Labels{"node_id": conf.NodeID},
+	}, []string{"dir", "provider"}))
+
 	m.codecOffered = mustRegister(m, prometheus.NewCounterVec(prometheus.CounterOpts{
 		Namespace:   "livekit",
 		Subsystem:   "sip",
@@ -546,6 +555,15 @@ func (c *CallMonitor) SDPSize(sz int, isOffer bool) {
 	c.m.sdpSize.WithLabelValues(typ).Observe(float64(sz))
 }
 
+// SDPParsePanic increments a counter denoting the number of times a panic has
+// occurred during SDP parsing.
+func (c *CallMonitor) SDPParsePanic() {
+	c.m.sdpParsePanics.With(prometheus.Labels{
+		"dir":      c.dir,
+		"provider": c.providerLabel(),
+	}).Inc()
+}
+
 func (m *Monitor) TransferStarted(dir CallDir) {
 	m.transfersTotal.WithLabelValues(dir.String()).Inc()
 	m.transfersActive.WithLabelValues(dir.String()).Inc()