Skip to content

Support SRTP #232

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Apr 10, 2025
Merged

Support SRTP #232

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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=
Expand Down
6 changes: 3 additions & 3 deletions pkg/media/dtmf/dtmf.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down
41 changes: 30 additions & 11 deletions pkg/media/rtp/conn.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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
}
Expand All @@ -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)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we want to alert on this? (sure sending invalid packet, potentially as an attempt to break our system)? If not, should this be a Warn or Info?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd prefer to not pass broken packet. Go crashes are fine and easy to recover from/fix, but crashes from C code are not. We could specifically fuzz the system to (hopefully) detect those.

As for the error/warn, even though this case won't break the system, it will still cause audio glitches in a specific call. I'd like to know when it happens, especially when a fix is as easy as increasing the buffer size. So I'd like to keep it as an error here.

}
continue // ignore partial messages
}

p = rtp.Packet{}
if err := p.Unmarshal(buf[:n]); err != nil {
Expand All @@ -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) {
Expand Down
8 changes: 4 additions & 4 deletions pkg/media/rtp/jitter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
}
Expand Down
7 changes: 4 additions & 3 deletions pkg/media/rtp/listen.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
})
}
Expand All @@ -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
}
Expand Down
18 changes: 9 additions & 9 deletions pkg/media/rtp/mux.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
51 changes: 26 additions & 25 deletions pkg/media/rtp/rtp.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ package rtp
import (
"fmt"
"math/rand/v2"
"slices"
"sync"

"github.com/pion/interceptor"
Expand All @@ -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 {
Expand All @@ -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
}
Expand All @@ -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
Expand All @@ -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
}

Expand Down Expand Up @@ -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))
}
Loading
Loading