Support SRTP (#232)

Initial support for SRTP.

Author: dennwc

go.mod

@@ -15,12 +15,13 @@ require (
 	github.com/livekit/protocol v1.36.2-0.20250408143132-c193b8d080da
 	github.com/livekit/psrpc v0.6.1-0.20250205181828-a0beed2e4126
 	github.com/livekit/server-sdk-go/v2 v2.5.0
-	github.com/livekit/sipgo v0.13.2-0.20250130142851-36ed3228d934
+	github.com/livekit/sipgo v0.13.2-0.20250410120437-ca5b8ca7b53d
 	github.com/mjibson/go-dsp v0.0.0-20180508042940-11479a337f12
 	github.com/ory/dockertest/v3 v3.11.0
 	github.com/pion/interceptor v0.1.37
 	github.com/pion/rtp v1.8.11
 	github.com/pion/sdp/v3 v3.0.10
+	github.com/pion/srtp/v3 v3.0.4
 	github.com/pion/webrtc/v4 v4.0.9
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.20.5
@@ -98,7 +99,6 @@ require (
 	github.com/pion/randutil v0.1.0 // indirect
 	github.com/pion/rtcp v1.2.15 // indirect
 	github.com/pion/sctp v1.8.35 // 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

@@ -133,8 +133,8 @@ github.com/livekit/psrpc v0.6.1-0.20250205181828-a0beed2e4126 h1:fzuYpAQbCid7ySP
 github.com/livekit/psrpc v0.6.1-0.20250205181828-a0beed2e4126/go.mod h1:X5WtEZ7OnEs72Fi5/J+i0on3964F1aynQpCalcgMqRo=
 github.com/livekit/server-sdk-go/v2 v2.5.0 h1:HCKm3f6PvefGp8emNC2mi9+9IXzBYrynuGbtUdp5u+w=
 github.com/livekit/server-sdk-go/v2 v2.5.0/go.mod h1:98/Sa+Wgb27ABwu0WYxLaMZaRfGljrrtoZDQ2xA4oVg=
-github.com/livekit/sipgo v0.13.2-0.20250130142851-36ed3228d934 h1:BKDNIg729VUlRCqQ0dNXFFxuEvxBBIPcqbRADPfkz54=
-github.com/livekit/sipgo v0.13.2-0.20250130142851-36ed3228d934/go.mod h1:nbNi0IsYn4tyY2ab7Rafvifty07miHYvgedPMKWbaI4=
+github.com/livekit/sipgo v0.13.2-0.20250410120437-ca5b8ca7b53d h1:x3JSKtsQpWt/fynro+s7MusrOqIcaGdCnuSXKGEajXc=
+github.com/livekit/sipgo v0.13.2-0.20250410120437-ca5b8ca7b53d/go.mod h1:nbNi0IsYn4tyY2ab7Rafvifty07miHYvgedPMKWbaI4=
 github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o=
 github.com/mackerelio/go-osstat v0.2.5/go.mod h1:atxwWF+POUZcdtR1wnsUcQxTytoHG4uhl2AKKzrOajY=
 github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg=

pkg/media/dtmf/dtmf.go

@@ -161,11 +161,11 @@ func Decode(data []byte) (Event, error) {
 	}, nil
 }
 
-func DecodeRTP(p *rtp.Packet) (Event, bool) {
-	if !p.Marker {
+func DecodeRTP(h *rtp.Header, payload []byte) (Event, bool) {
+	if !h.Marker {
 		return Event{}, false
 	}
-	ev, err := Decode(p.Payload)
+	ev, err := Decode(payload)
 	if err != nil {
 		return Event{}, false
 	}

pkg/media/rtp/conn.go

@@ -16,17 +16,21 @@ package rtp
 
 import (
 	"net"
+	"net/netip"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"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()
@@ -40,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,
@@ -67,7 +75,9 @@ type UDPConn interface {
 	Close() error
 }
 
+// Deprecated: use MediaPort instead
 type Conn struct {
+	log            logger.Logger
 	wmu            sync.Mutex
 	conn           UDPConn
 	closed         core.Fuse
@@ -131,9 +141,11 @@ func (c *Conn) Listen(portMin, portMax int, listenAddr string) error {
 	if listenAddr == "" {
 		listenAddr = "0.0.0.0"
 	}
-
-	var err error
-	c.conn, err = ListenUDPPortRange(portMin, portMax, net.ParseIP(listenAddr))
+	ip, err := netip.ParseAddr(listenAddr)
+	if err != nil {
+		return err
+	}
+	c.conn, err = ListenUDPPortRange(portMin, portMax, ip)
 	if err != nil {
 		return err
 	}
@@ -150,13 +162,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 {
@@ -167,24 +187,23 @@ func (c *Conn) readLoop() {
 			close(c.received)
 		}
 		if h := c.onRTP.Load(); h != nil {
-			_ = (*h).HandleRTP(&p)
+			_ = (*h).HandleRTP(&p.Header, p.Payload)
 		}
 	}
 }
 
-func (c *Conn) WriteRTP(p *rtp.Packet) error {
+func (c *Conn) WriteRTP(h *rtp.Header, payload []byte) (int, error) {
 	addr := c.dest.Load()
 	if addr == nil {
-		return nil
+		return 0, nil
 	}
-	data, err := p.Marshal()
+	data, err := (&rtp.Packet{Header: *h, Payload: payload}).Marshal()
 	if err != nil {
-		return err
+		return 0, err
 	}
 	c.wmu.Lock()
 	defer c.wmu.Unlock()
-	_, err = c.conn.WriteToUDP(data, addr)
-	return err
+	return c.conn.WriteToUDP(data, addr)
 }
 
 func (c *Conn) ReadRTP() (*rtp.Packet, *net.UDPAddr, error) {

pkg/media/rtp/jitter.go

@@ -38,11 +38,11 @@ type jitterHandler struct {
 	buf *jitter.Buffer
 }
 
-func (h *jitterHandler) HandleRTP(p *rtp.Packet) error {
-	h.buf.Push(p.Clone())
+func (r *jitterHandler) HandleRTP(h *rtp.Header, payload []byte) error {
+	r.buf.Push(&rtp.Packet{Header: *h, Payload: payload})
 	var last error
-	for _, p := range h.buf.Pop(false) {
-		if err := h.h.HandleRTP(p); err != nil {
+	for _, p := range r.buf.Pop(false) {
+		if err := r.h.HandleRTP(&p.Header, p.Payload); err != nil {
 			last = err
 		}
 	}

pkg/media/rtp/listen.go

@@ -18,14 +18,15 @@ import (
 	"errors"
 	"math/rand"
 	"net"
+	"net/netip"
 )
 
 var ListenErr = errors.New("failed to listen on udp port")
 
-func ListenUDPPortRange(portMin, portMax int, IP net.IP) (*net.UDPConn, error) {
+func ListenUDPPortRange(portMin, portMax int, ip netip.Addr) (*net.UDPConn, error) {
 	if portMin == 0 && portMax == 0 {
 		return net.ListenUDP("udp", &net.UDPAddr{
-			IP:   IP,
+			IP:   ip.AsSlice(),
 			Port: 0,
 		})
 	}
@@ -48,7 +49,7 @@ func ListenUDPPortRange(portMin, portMax int, IP net.IP) (*net.UDPConn, error) {
 	portCurrent := portStart
 
 	for {
-		c, e := net.ListenUDP("udp", &net.UDPAddr{IP: IP, Port: portCurrent})
+		c, e := net.ListenUDP("udp", &net.UDPAddr{IP: ip.AsSlice(), Port: portCurrent})
 		if e == nil {
 			return c, nil
 		}

pkg/media/rtp/mux.go

@@ -35,25 +35,25 @@ type Mux struct {
 
 // HandleRTP selects a Handler based on payload type.
 // Types can be registered with Register. If no handler is set, a default one will be used.
-func (m *Mux) HandleRTP(p *rtp.Packet) error {
+func (m *Mux) HandleRTP(h *rtp.Header, payload []byte) error {
 	if m == nil {
 		return nil
 	}
-	var h Handler
+	var r Handler
 	m.mu.RLock()
-	if p.PayloadType < byte(len(m.static)) {
-		h = m.static[p.PayloadType]
+	if h.PayloadType < byte(len(m.static)) {
+		r = m.static[h.PayloadType]
 	} else {
-		h = m.dynamic[p.PayloadType]
+		r = m.dynamic[h.PayloadType]
 	}
-	if h == nil {
-		h = m.def
+	if r == nil {
+		r = m.def
 	}
 	m.mu.RUnlock()
-	if h == nil {
+	if r == nil {
 		return nil
 	}
-	return h.HandleRTP(p)
+	return r.HandleRTP(h, payload)
 }
 
 // SetDefault sets a default RTP handler.

pkg/media/rtp/rtp.go

@@ -17,6 +17,7 @@ package rtp
 import (
 	"fmt"
 	"math/rand/v2"
+	"slices"
 	"sync"
 
 	"github.com/pion/interceptor"
@@ -31,21 +32,21 @@ type BytesFrame interface {
 }
 
 type Writer interface {
-	WriteRTP(p *rtp.Packet) error
+	WriteRTP(h *rtp.Header, payload []byte) (int, error)
 }
 
 type Reader interface {
 	ReadRTP() (*rtp.Packet, interceptor.Attributes, error)
 }
 
 type Handler interface {
-	HandleRTP(p *rtp.Packet) error
+	HandleRTP(h *rtp.Header, payload []byte) error
 }
 
-type HandlerFunc func(p *rtp.Packet) error
+type HandlerFunc func(h *rtp.Header, payload []byte) error
 
-func (fnc HandlerFunc) HandleRTP(p *rtp.Packet) error {
-	return fnc(p)
+func (fnc HandlerFunc) HandleRTP(h *rtp.Header, payload []byte) error {
+	return fnc(h, payload)
 }
 
 func HandleLoop(r Reader, h Handler) error {
@@ -54,7 +55,7 @@ func HandleLoop(r Reader, h Handler) error {
 		if err != nil {
 			return err
 		}
-		err = h.HandleRTP(p)
+		err = h.HandleRTP(&p.Header, p.Payload)
 		if err != nil {
 			return err
 		}
@@ -64,26 +65,27 @@ func HandleLoop(r Reader, h Handler) error {
 // Buffer is a Writer that clones and appends RTP packets into a slice.
 type Buffer []*Packet
 
-func (b *Buffer) WriteRTP(p *Packet) error {
-	p2 := p.Clone()
-	*b = append(*b, p2)
-	return nil
+func (b *Buffer) WriteRTP(h *rtp.Header, payload []byte) (int, error) {
+	*b = append(*b, &rtp.Packet{
+		Header:  *h,
+		Payload: slices.Clone(payload),
+	})
+	return len(payload), nil
 }
 
 // NewSeqWriter creates an RTP writer that automatically increments the sequence number.
 func NewSeqWriter(w Writer) *SeqWriter {
 	s := &SeqWriter{w: w}
-	s.p = rtp.Packet{
-		Header: rtp.Header{
-			Version:        2,
-			SSRC:           rand.Uint32(),
-			SequenceNumber: 0,
-		},
+	s.h = rtp.Header{
+		Version:        2,
+		SSRC:           rand.Uint32(),
+		SequenceNumber: 0,
 	}
 	return s
 }
 
 type Packet = rtp.Packet
+type Header = rtp.Header
 
 type Event struct {
 	Type      byte
@@ -95,20 +97,19 @@ type Event struct {
 type SeqWriter struct {
 	mu sync.Mutex
 	w  Writer
-	p  Packet
+	h  Header
 }
 
 func (s *SeqWriter) WriteEvent(ev *Event) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	s.p.PayloadType = ev.Type
-	s.p.Payload = ev.Payload
-	s.p.Marker = ev.Marker
-	s.p.Timestamp = ev.Timestamp
-	if err := s.w.WriteRTP(&s.p); err != nil {
+	s.h.PayloadType = ev.Type
+	s.h.Marker = ev.Marker
+	s.h.Timestamp = ev.Timestamp
+	if _, err := s.w.WriteRTP(&s.h, ev.Payload); err != nil {
 		return err
 	}
-	s.p.Header.SequenceNumber++
+	s.h.SequenceNumber++
 	return nil
 }
 
@@ -211,6 +212,6 @@ func (s *MediaStreamIn[T]) String() string {
 	return fmt.Sprintf("RTP(%d) -> %s", s.Writer.SampleRate(), s.Writer)
 }
 
-func (s *MediaStreamIn[T]) HandleRTP(p *rtp.Packet) error {
-	return s.Writer.WriteSample(T(p.Payload))
+func (s *MediaStreamIn[T]) HandleRTP(_ *rtp.Header, payload []byte) error {
+	return s.Writer.WriteSample(T(payload))
 }