diff --git a/internal/protocols/webrtc/from_stream_test.go b/internal/protocols/webrtc/from_stream_test.go index 48f5f23f..57a68d76 100644 --- a/internal/protocols/webrtc/from_stream_test.go +++ b/internal/protocols/webrtc/from_stream_test.go @@ -136,10 +136,10 @@ func TestFromStreamResampleOpus(t *testing.T) { require.NoError(t, err) defer pc2.Close() - offer, err := pc1.CreatePartialOffer() + offer, err := pc1.CreatePartialOffer(false) require.NoError(t, err) - answer, err := pc2.CreateFullAnswer(offer) + answer, err := pc2.CreateFullAnswer(offer, false) require.NoError(t, err) err = pc1.SetAnswer(answer) @@ -264,10 +264,10 @@ func TestFromStreamResampleOpusAbsoluteTimestamp(t *testing.T) { require.NoError(t, err) t.Cleanup(pcPublisher.Close) - offer, err := pcReader.CreatePartialOffer() + offer, err := pcReader.CreatePartialOffer(false) require.NoError(t, err) - answer, err := pcPublisher.CreateFullAnswer(offer) + answer, err := pcPublisher.CreateFullAnswer(offer, false) require.NoError(t, err) err = pcReader.SetAnswer(answer) diff --git a/internal/protocols/webrtc/peer_connection.go b/internal/protocols/webrtc/peer_connection.go index c64510f0..4df76f9d 100644 --- a/internal/protocols/webrtc/peer_connection.go +++ b/internal/protocols/webrtc/peer_connection.go @@ -174,12 +174,16 @@ type PeerConnection struct { newLocalCandidate chan *webrtc.ICECandidateInit incomingTrack chan trackRecvPair - connected chan struct{} - failed chan struct{} - closed chan struct{} - gatheringDone chan struct{} - done chan struct{} - chStartReading chan struct{} + + stateMutex sync.Mutex + state webrtc.PeerConnectionState + stateChanged chan struct{} + + gatheringMutex sync.Mutex + gatheringDone chan struct{} + + done chan struct{} + chStartReading chan struct{} } // Start starts the peer connection. @@ -322,9 +326,7 @@ func (co *PeerConnection) Start() error { co.ctx, co.ctxCancel = context.WithCancel(context.Background()) co.newLocalCandidate = make(chan *webrtc.ICECandidateInit) - co.connected = make(chan struct{}) - co.failed = make(chan struct{}) - co.closed = make(chan struct{}) + co.stateChanged = make(chan struct{}) co.gatheringDone = make(chan struct{}) co.incomingTrack = make(chan trackRecvPair) co.done = make(chan struct{}) @@ -371,64 +373,44 @@ func (co *PeerConnection) Start() error { }) } - var stateChangeMutex sync.Mutex - co.wr.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { - stateChangeMutex.Lock() - defer stateChangeMutex.Unlock() + co.stateMutex.Lock() + defer co.stateMutex.Unlock() - select { - case <-co.closed: + if co.state == webrtc.PeerConnectionStateFailed || co.state == webrtc.PeerConnectionStateClosed { return - default: } + co.state = state + close(co.stateChanged) + co.stateChanged = make(chan struct{}) + co.Log.Log(logger.Debug, "peer connection state: "+state.String()) - switch state { - case webrtc.PeerConnectionStateConnected: - // PeerConnectionStateConnected can arrive twice, since state can - // switch from "disconnected" to "connected". - // contrarily, we're interested into emitting "connected" once. - select { - case <-co.connected: - return - default: - } - + if state == webrtc.PeerConnectionStateConnected { co.Log.Log(logger.Info, "peer connection established, local candidate: %v, remote candidate: %v", co.LocalCandidate(), co.RemoteCandidate()) - - close(co.connected) - - case webrtc.PeerConnectionStateFailed: - close(co.failed) - - case webrtc.PeerConnectionStateClosed: - // "closed" can arrive before "failed" and without - // the Close() method being called at all. - // It happens when the other peer sends a termination - // message like a DTLS CloseNotify. - select { - case <-co.failed: - default: - close(co.failed) - } - - close(co.closed) } }) co.wr.OnICECandidate(func(i *webrtc.ICECandidate) { + co.gatheringMutex.Lock() + defer co.gatheringMutex.Unlock() + if i != nil { v := i.ToJSON() + select { case co.newLocalCandidate <- &v: - case <-co.connected: + case <-co.Connected(): case <-co.ctx.Done(): } } else { - close(co.gatheringDone) + select { + case <-co.gatheringDone: + default: + close(co.gatheringDone) + } } }) @@ -460,7 +442,7 @@ func (co *PeerConnection) run() { // we have to wait for OnConnectionStateChange to return anyway, // since it is executed in an uncontrolled goroutine. // https://github.com/pion/webrtc/blob/v4.2.8/peerconnection.go#L529 - <-co.closed + <-co.failedNoContext() }() for { @@ -609,8 +591,28 @@ func (co *PeerConnection) filterLocalDescription(desc *webrtc.SessionDescription } // CreatePartialOffer creates a partial offer. -func (co *PeerConnection) CreatePartialOffer() (*webrtc.SessionDescription, error) { - tmp, err := co.wr.CreateOffer(nil) +func (co *PeerConnection) CreatePartialOffer(restart bool) (*webrtc.SessionDescription, error) { + var options *webrtc.OfferOptions + if restart { + co.gatheringMutex.Lock() + + select { + case <-co.gatheringDone: + default: + co.gatheringMutex.Unlock() + return nil, fmt.Errorf("tried an ICE restart before candidate gathering is complete") + } + + co.gatheringDone = make(chan struct{}) + + co.gatheringMutex.Unlock() + + options = &webrtc.OfferOptions{ + ICERestart: true, + } + } + + tmp, err := co.wr.CreateOffer(options) if err != nil { return nil, err } @@ -662,13 +664,36 @@ func (co *PeerConnection) SetAnswer(answer *webrtc.SessionDescription) error { return co.wr.SetRemoteDescription(*answer) } +// RemoteDescription returns the current remote description. +func (co *PeerConnection) RemoteDescription() *webrtc.SessionDescription { + return co.wr.RemoteDescription() +} + // AddRemoteCandidate adds a remote candidate. func (co *PeerConnection) AddRemoteCandidate(candidate *webrtc.ICECandidateInit) error { return co.wr.AddICECandidate(*candidate) } -// CreateFullAnswer creates a full answer. -func (co *PeerConnection) CreateFullAnswer(offer *webrtc.SessionDescription) (*webrtc.SessionDescription, error) { +// CreateFullAnswer accepts an offer and creates a full answer. +func (co *PeerConnection) CreateFullAnswer( + offer *webrtc.SessionDescription, + restarted bool, +) (*webrtc.SessionDescription, error) { + if restarted { + co.gatheringMutex.Lock() + + select { + case <-co.gatheringDone: + default: + co.gatheringMutex.Unlock() + return nil, fmt.Errorf("tried an ICE restart before candidate gathering is complete") + } + + co.gatheringDone = make(chan struct{}) + + co.gatheringMutex.Unlock() + } + err := co.wr.SetRemoteDescription(*offer) if err != nil { return nil, err @@ -726,7 +751,7 @@ outer: case <-t.C: return fmt.Errorf("deadline exceeded while waiting connection") - case <-co.connected: + case <-co.Connected(): break outer case <-co.ctx.Done(): @@ -781,12 +806,83 @@ func (co *PeerConnection) GatherIncomingTracks(timeout time.Duration) error { // Connected returns when connected. func (co *PeerConnection) Connected() <-chan struct{} { - return co.connected + ch := make(chan struct{}) + + go func() { + for { + co.stateMutex.Lock() + state := co.state + stateChanged := co.stateChanged + co.stateMutex.Unlock() + + if state == webrtc.PeerConnectionStateConnected { + close(ch) + return + } + + select { + case <-stateChanged: + case <-co.ctx.Done(): + // exit without closing ch + return + } + } + }() + + return ch } -// Failed returns when failed. +// Failed returns when failed or closed. func (co *PeerConnection) Failed() <-chan struct{} { - return co.failed + ch := make(chan struct{}) + + go func() { + defer close(ch) + for { + co.stateMutex.Lock() + state := co.state + stateChanged := co.stateChanged + co.stateMutex.Unlock() + + // "closed" can arrive before "failed" and without + // the Close() method being called at all. + // It happens when the other peer sends a termination + // message like a DTLS CloseNotify. + if state == webrtc.PeerConnectionStateFailed || state == webrtc.PeerConnectionStateClosed { + return + } + + select { + case <-stateChanged: + case <-co.ctx.Done(): + return + } + } + }() + + return ch +} + +func (co *PeerConnection) failedNoContext() <-chan struct{} { + ch := make(chan struct{}) + + go func() { + for { + co.stateMutex.Lock() + state := co.state + stateChanged := co.stateChanged + co.stateMutex.Unlock() + + if state == webrtc.PeerConnectionStateFailed || state == webrtc.PeerConnectionStateClosed { + close(ch) + return + } + + <-stateChanged + } + }() + + return ch } // NewLocalCandidate returns when there's a new local candidate. diff --git a/internal/protocols/webrtc/peer_connection_test.go b/internal/protocols/webrtc/peer_connection_test.go index f8266e88..006d4e22 100644 --- a/internal/protocols/webrtc/peer_connection_test.go +++ b/internal/protocols/webrtc/peer_connection_test.go @@ -66,7 +66,7 @@ func TestPeerConnectionCloseImmediately2(t *testing.T) { require.NoError(t, err) defer pc.Close() - _, err = pc.CreatePartialOffer() + _, err = pc.CreatePartialOffer(false) require.NoError(t, err) // wait for ICE candidates to be generated @@ -143,7 +143,7 @@ func TestPeerConnectionCandidates(t *testing.T) { require.NoError(t, err) defer pc.Close() - answer, err := pc.CreateFullAnswer(&offer) + answer, err := pc.CreateFullAnswer(&offer, false) require.NoError(t, err) n := len(regexp.MustCompile("(?m)^a=candidate:.+? udp .+? typ host").FindAllString(answer.SDP, -1)) @@ -255,10 +255,10 @@ func TestPeerConnectionConnectivity(t *testing.T) { require.NoError(t, err) defer serverPC.Close() - offer, err := clientPC.CreatePartialOffer() + offer, err := clientPC.CreatePartialOffer(false) require.NoError(t, err) - answer, err := serverPC.CreateFullAnswer(offer) + answer, err := serverPC.CreateFullAnswer(offer, false) require.NoError(t, err) require.Equal(t, 2, strings.Count(answer.SDP, "a=candidate:")) @@ -331,7 +331,7 @@ func TestPeerConnectionRead(t *testing.T) { err = pub.SetLocalDescription(offer) require.NoError(t, err) - answer, err := reader.CreateFullAnswer(&offer) + answer, err := reader.CreateFullAnswer(&offer, false) require.NoError(t, err) err = pub.SetRemoteDescription(*answer) @@ -500,7 +500,7 @@ func TestPeerConnectionReadSimulcast(t *testing.T) { err = pub.SetLocalDescription(offer) require.NoError(t, err) - answer, err := reader.CreateFullAnswer(&offer) + answer, err := reader.CreateFullAnswer(&offer, false) require.NoError(t, err) err = pub.SetRemoteDescription(*answer) @@ -614,7 +614,7 @@ func TestPeerConnectionStripIncomingTWCC(t *testing.T) { err = pub.SetLocalDescription(offer) require.NoError(t, err) - answer, err := reader.CreateFullAnswer(&offer) + answer, err := reader.CreateFullAnswer(&offer, false) require.NoError(t, err) err = pub.SetRemoteDescription(*answer) @@ -706,10 +706,10 @@ func TestPeerConnectionPublishRead(t *testing.T) { require.NoError(t, err) defer pc2.Close() - offer, err := pc1.CreatePartialOffer() + offer, err := pc1.CreatePartialOffer(false) require.NoError(t, err) - answer, err := pc2.CreateFullAnswer(offer) + answer, err := pc2.CreateFullAnswer(offer, false) require.NoError(t, err) err = pc1.SetAnswer(answer) @@ -796,10 +796,10 @@ func TestPeerConnectionFallbackCodecs(t *testing.T) { require.NoError(t, err) defer pc2.Close() - offer, err := pc1.CreatePartialOffer() + offer, err := pc1.CreatePartialOffer(false) require.NoError(t, err) - answer, err := pc2.CreateFullAnswer(offer) + answer, err := pc2.CreateFullAnswer(offer, false) require.NoError(t, err) var s sdp.SessionDescription @@ -871,7 +871,7 @@ func TestPeerConnectionPublishDataChannel(t *testing.T) { require.NoError(t, err) defer pc2.Close() - answer, err := pc2.CreateFullAnswer(&offer) + answer, err := pc2.CreateFullAnswer(&offer, false) require.NoError(t, err) err = pc1.SetRemoteDescription(*answer) diff --git a/internal/protocols/webrtc/to_stream_test.go b/internal/protocols/webrtc/to_stream_test.go index 5655f042..baf2a75c 100644 --- a/internal/protocols/webrtc/to_stream_test.go +++ b/internal/protocols/webrtc/to_stream_test.go @@ -357,10 +357,10 @@ func TestToStream(t *testing.T) { require.NoError(t, err) defer pc2.Close() - offer, err := pc1.CreatePartialOffer() + offer, err := pc1.CreatePartialOffer(false) require.NoError(t, err) - answer, err := pc2.CreateFullAnswer(offer) + answer, err := pc2.CreateFullAnswer(offer, false) require.NoError(t, err) err = pc1.SetAnswer(answer) diff --git a/internal/protocols/whip/client.go b/internal/protocols/whip/client.go index 1b1a1f44..3fa3b022 100644 --- a/internal/protocols/whip/client.go +++ b/internal/protocols/whip/client.go @@ -8,6 +8,7 @@ import ( "io" "net/http" "net/url" + "strconv" "strings" "time" @@ -23,6 +24,55 @@ const ( maxInboundSDPSize = 128 * 1024 ) +func whipAnswer(body []byte) *pwebrtc.SessionDescription { + return &pwebrtc.SessionDescription{ + Type: pwebrtc.SDPTypeAnswer, + SDP: string(body), + } +} + +func offerAndCandidateToSDPFragment( + offer *pwebrtc.SessionDescription, + candidate *pwebrtc.ICECandidateInit, +) (*SDPFragment, error) { + f := &SDPFragment{} + + var desc sdp.SessionDescription + err := desc.Unmarshal([]byte(offer.SDP)) + if err != nil { + return nil, err + } + + if candidate.SDPMLineIndex == nil { + return nil, fmt.Errorf("sdpMLineIndex is null") + } + + if len(desc.MediaDescriptions) < int(*candidate.SDPMLineIndex)+1 { + return nil, fmt.Errorf("sdpMLineIndex is out of range") + } + + media := desc.MediaDescriptions[*candidate.SDPMLineIndex] + + iceUFrag, _ := media.Attribute("ice-ufrag") + icePwd, _ := media.Attribute("ice-pwd") + + if iceUFrag == "" || icePwd == "" { + return nil, fmt.Errorf("ice-ufrag or ice-pwd are missing in the media of the candidate") + } + + f.Medias = append(f.Medias, &sdp.MediaDescription{ + MediaName: media.MediaName, + Attributes: []sdp.Attribute{ + {Key: "mid", Value: strconv.FormatUint(uint64(*candidate.SDPMLineIndex), 10)}, + {Key: "ice-ufrag", Value: iceUFrag}, + {Key: "ice-pwd", Value: icePwd}, + {Key: "candidate", Value: candidate.Candidate}, + }, + }) + + return f, nil +} + // Client is a WHIP client. type Client struct { URL *url.URL @@ -99,7 +149,7 @@ func (c *Client) initializeInner(ctx context.Context) error { var offer *pwebrtc.SessionDescription if c.useTrickleICE { var err error - offer, err = c.pc.CreatePartialOffer() + offer, err = c.pc.CreatePartialOffer(false) if err != nil { return err } @@ -303,13 +353,8 @@ func (c *Client) postOffer( return nil, err } - answer := &pwebrtc.SessionDescription{ - Type: pwebrtc.SDPTypeAnswer, - SDP: string(sdp), - } - return &whipPostOfferResponse{ - Answer: answer, + Answer: whipAnswer(sdp), Location: Location, ETag: etag, }, nil @@ -321,12 +366,17 @@ func (c *Client) patchCandidate( etag string, candidate *pwebrtc.ICECandidateInit, ) error { - frag, err := ICEFragmentMarshal(offer.SDP, []*pwebrtc.ICECandidateInit{candidate}) + frag, err := offerAndCandidateToSDPFragment(offer, candidate) if err != nil { return err } - req, err := http.NewRequestWithContext(ctx, http.MethodPatch, c.URL.String(), bytes.NewReader(frag)) + enc, err := frag.Marshal() + if err != nil { + return err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPatch, c.URL.String(), bytes.NewReader(enc)) if err != nil { return err } diff --git a/internal/protocols/whip/client_test.go b/internal/protocols/whip/client_test.go index 3f3a6b12..9fcb7007 100644 --- a/internal/protocols/whip/client_test.go +++ b/internal/protocols/whip/client_test.go @@ -103,7 +103,7 @@ func TestClientRead(t *testing.T) { require.NotContains(t, offer.SDP, "a=candidate:") - answer, err2 := pc.CreateFullAnswer(offer) + answer, err2 := pc.CreateFullAnswer(offer, false) require.NoError(t, err2) w.Header().Set("Content-Type", "application/sdp") @@ -273,7 +273,7 @@ func TestClientPublish(t *testing.T) { require.NotContains(t, offer.SDP, "a=candidate:") - answer, err2 := pc.CreateFullAnswer(offer) + answer, err2 := pc.CreateFullAnswer(offer, false) require.NoError(t, err2) w.Header().Set("Content-Type", "application/sdp") @@ -461,7 +461,7 @@ func TestClientBearerToken(t *testing.T) { require.NoError(t, err2) offer := whipOffer(body) - answer, err2 := pc.CreateFullAnswer(offer) + answer, err2 := pc.CreateFullAnswer(offer, false) require.NoError(t, err2) w.Header().Set("Content-Type", "application/sdp") @@ -537,7 +537,7 @@ func TestClientNoTrickleICE(t *testing.T) { require.Contains(t, offer.SDP, "a=candidate:") - answer, err2 := pc.CreateFullAnswer(offer) + answer, err2 := pc.CreateFullAnswer(offer, false) require.NoError(t, err2) w.Header().Set("Content-Type", "application/sdp") diff --git a/internal/protocols/whip/ice_fragment.go b/internal/protocols/whip/ice_fragment.go deleted file mode 100644 index 358661eb..00000000 --- a/internal/protocols/whip/ice_fragment.go +++ /dev/null @@ -1,87 +0,0 @@ -package whip - -import ( - "fmt" - "strconv" - - "github.com/pion/sdp/v3" - "github.com/pion/webrtc/v4" -) - -// ICEFragmentUnmarshal decodes an ICE fragment. -func ICEFragmentUnmarshal(buf []byte) ([]*webrtc.ICECandidateInit, error) { - buf = append([]byte("v=0\r\no=- 0 0 IN IP4 0.0.0.0\r\ns=-\r\nt=0 0\r\n"), buf...) - - var sdp sdp.SessionDescription - err := sdp.Unmarshal(buf) - if err != nil { - return nil, err - } - - var ret []*webrtc.ICECandidateInit - - for _, media := range sdp.MediaDescriptions { - mid, ok := media.Attribute("mid") - if !ok { - return nil, fmt.Errorf("mid attribute is missing") - } - - var tmp uint64 - tmp, err = strconv.ParseUint(mid, 10, 16) - if err != nil { - return nil, fmt.Errorf("invalid mid attribute") - } - midNum := uint16(tmp) - - for _, attr := range media.Attributes { - if attr.Key == "candidate" { - ret = append(ret, &webrtc.ICECandidateInit{ - Candidate: attr.Value, - SDPMid: &mid, - SDPMLineIndex: &midNum, - }) - } - } - } - - return ret, nil -} - -// ICEFragmentMarshal encodes an ICE fragment. -func ICEFragmentMarshal(offer string, candidates []*webrtc.ICECandidateInit) ([]byte, error) { - var sdp sdp.SessionDescription - err := sdp.Unmarshal([]byte(offer)) - if err != nil || len(sdp.MediaDescriptions) == 0 { - return nil, err - } - - firstMedia := sdp.MediaDescriptions[0] - iceUfrag, _ := firstMedia.Attribute("ice-ufrag") - icePwd, _ := firstMedia.Attribute("ice-pwd") - - candidatesByMedia := make(map[uint16][]*webrtc.ICECandidateInit) - for _, candidate := range candidates { - if candidate.SDPMLineIndex == nil { - return nil, fmt.Errorf("sdpMLineIndex is null") - } - mid := *candidate.SDPMLineIndex - candidatesByMedia[mid] = append(candidatesByMedia[mid], candidate) - } - - frag := "a=ice-ufrag:" + iceUfrag + "\r\n" + - "a=ice-pwd:" + icePwd + "\r\n" - - for mid, media := range sdp.MediaDescriptions { - cbm, ok := candidatesByMedia[uint16(mid)] - if ok { - frag += "m=" + media.MediaName.String() + "\r\n" + - "a=mid:" + strconv.FormatUint(uint64(mid), 10) + "\r\n" - - for _, candidate := range cbm { - frag += "a=candidate:" + candidate.Candidate + "\r\n" - } - } - } - - return []byte(frag), nil -} diff --git a/internal/protocols/whip/ice_fragment_test.go b/internal/protocols/whip/ice_fragment_test.go deleted file mode 100644 index 3cde6704..00000000 --- a/internal/protocols/whip/ice_fragment_test.go +++ /dev/null @@ -1,206 +0,0 @@ -package whip - -import ( - "testing" - - "github.com/pion/webrtc/v4" - "github.com/stretchr/testify/require" -) - -func ptrOf[T any](v T) *T { - p := new(T) - *p = v - return p -} - -var iceFragmentCases = []struct { - name string - offer string - candidates []*webrtc.ICECandidateInit - enc string -}{ - { - "a", - "v=0\n" + - "o=- 8429658789122714282 1690995382 IN IP4 0.0.0.0\n" + - "s=-\n" + - "t=0 0\n" + - "a=fingerprint:sha-256 EA:05:9D:04:8F:56:41:92:3E:D5:2B:55:03:" + - "1B:5A:2C:3D:D8:B3:FB:1B:D9:F7:1F:DA:77:0E:B9:E0:3D:B6:FF\n" + - "a=extmap-allow-mixed\n" + - "a=group:BUNDLE 0\n" + - "m=video 9 UDP/TLS/RTP/SAVPF 96 97 98 99 100 101 102 121 127 120 125 107 108 109 123 118 45 46 116\n" + - "c=IN IP4 0.0.0.0\n" + - "a=setup:actpass\n" + - "a=mid:0\n" + - "a=ice-ufrag:tUQMzoQAVLzlvBys\n" + - "a=ice-pwd:pimyGfJcjjRwvUjnmGOODSjtIxyDljQj\n" + - "a=rtcp-mux\n" + - "a=rtcp-rsize\n" + - "a=rtpmap:96 VP8/90000\n" + - "a=rtcp-fb:96 goog-remb \n" + - "a=rtcp-fb:96 ccm fir\n" + - "a=rtcp-fb:96 nack \n" + - "a=rtcp-fb:96 nack pli\n" + - "a=rtcp-fb:96 nack \n" + - "a=rtcp-fb:96 nack pli\n" + - "a=rtcp-fb:96 transport-cc \n" + - "a=rtpmap:97 rtx/90000\n" + - "a=fmtp:97 apt=96\n" + - "a=rtcp-fb:97 nack \n" + - "a=rtcp-fb:97 nack pli\n" + - "a=rtcp-fb:97 transport-cc \n" + - "a=rtpmap:98 VP9/90000\n" + - "a=fmtp:98 profile-id=0\n" + - "a=rtcp-fb:98 goog-remb \n" + - "a=rtcp-fb:98 ccm fir\n" + - "a=rtcp-fb:98 nack \n" + - "a=rtcp-fb:98 nack pli\n" + - "a=rtcp-fb:98 nack \n" + - "a=rtcp-fb:98 nack pli\n" + - "a=rtcp-fb:98 transport-cc \n" + - "a=rtpmap:99 rtx/90000\n" + - "a=fmtp:99 apt=98\n" + - "a=rtcp-fb:99 nack \n" + - "a=rtcp-fb:99 nack pli\n" + - "a=rtcp-fb:99 transport-cc \n" + - "a=rtpmap:100 VP9/90000\n" + - "a=fmtp:100 profile-id=1\n" + - "a=rtcp-fb:100 goog-remb \n" + - "a=rtcp-fb:100 ccm fir\n" + - "a=rtcp-fb:100 nack \n" + - "a=rtcp-fb:100 nack pli\n" + - "a=rtcp-fb:100 nack \n" + - "a=rtcp-fb:100 nack pli\n" + - "a=rtcp-fb:100 transport-cc \n" + - "a=rtpmap:101 rtx/90000\n" + - "a=fmtp:101 apt=100\n" + - "a=rtcp-fb:101 nack \n" + - "a=rtcp-fb:101 nack pli\n" + - "a=rtcp-fb:101 transport-cc \n" + - "a=rtpmap:102 H264/90000\n" + - "a=fmtp:102 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42001f\n" + - "a=rtcp-fb:102 goog-remb \n" + - "a=rtcp-fb:102 ccm fir\n" + - "a=rtcp-fb:102 nack \n" + - "a=rtcp-fb:102 nack pli\n" + - "a=rtcp-fb:102 nack \n" + - "a=rtcp-fb:102 nack pli\n" + - "a=rtcp-fb:102 transport-cc \n" + - "a=rtpmap:121 rtx/90000\n" + - "a=fmtp:121 apt=102\n" + - "a=rtcp-fb:121 nack \n" + - "a=rtcp-fb:121 nack pli\n" + - "a=rtcp-fb:121 transport-cc \n" + - "a=rtpmap:127 H264/90000\n" + - "a=fmtp:127 level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42001f\n" + - "a=rtcp-fb:127 goog-remb \n" + - "a=rtcp-fb:127 ccm fir\n" + - "a=rtcp-fb:127 nack \n" + - "a=rtcp-fb:127 nack pli\n" + - "a=rtcp-fb:127 nack \n" + - "a=rtcp-fb:127 nack pli\n" + - "a=rtcp-fb:127 transport-cc \n" + - "a=rtpmap:120 rtx/90000\n" + - "a=fmtp:120 apt=127\n" + - "a=rtcp-fb:120 nack \n" + - "a=rtcp-fb:120 nack pli\n" + - "a=rtcp-fb:120 transport-cc \n" + - "a=rtpmap:125 H264/90000\n" + - "a=fmtp:125 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f\n" + - "a=rtcp-fb:125 goog-remb \n" + - "a=rtcp-fb:125 ccm fir\n" + - "a=rtcp-fb:125 nack \n" + - "a=rtcp-fb:125 nack pli\n" + - "a=rtcp-fb:125 nack \n" + - "a=rtcp-fb:125 nack pli\n" + - "a=rtcp-fb:125 transport-cc \n" + - "a=rtpmap:107 rtx/90000\n" + - "a=fmtp:107 apt=125\n" + - "a=rtcp-fb:107 nack \n" + - "a=rtcp-fb:107 nack pli\n" + - "a=rtcp-fb:107 transport-cc \n" + - "a=rtpmap:108 H264/90000\n" + - "a=fmtp:108 level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42e01f\n" + - "a=rtcp-fb:108 goog-remb \n" + - "a=rtcp-fb:108 ccm fir\n" + - "a=rtcp-fb:108 nack \n" + - "a=rtcp-fb:108 nack pli\n" + - "a=rtcp-fb:108 nack \n" + - "a=rtcp-fb:108 nack pli\n" + - "a=rtcp-fb:108 transport-cc \n" + - "a=rtpmap:109 rtx/90000\n" + - "a=fmtp:109 apt=108\n" + - "a=rtcp-fb:109 nack \n" + - "a=rtcp-fb:109 nack pli\n" + - "a=rtcp-fb:109 transport-cc \n" + - "a=rtpmap:123 H264/90000\n" + - "a=fmtp:123 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640032\n" + - "a=rtcp-fb:123 goog-remb \n" + - "a=rtcp-fb:123 ccm fir\n" + - "a=rtcp-fb:123 nack \n" + - "a=rtcp-fb:123 nack pli\n" + - "a=rtcp-fb:123 nack \n" + - "a=rtcp-fb:123 nack pli\n" + - "a=rtcp-fb:123 transport-cc \n" + - "a=rtpmap:118 rtx/90000\n" + - "a=fmtp:118 apt=123\n" + - "a=rtcp-fb:118 nack \n" + - "a=rtcp-fb:118 nack pli\n" + - "a=rtcp-fb:118 transport-cc \n" + - "a=rtpmap:45 AV1/90000\n" + - "a=rtcp-fb:45 goog-remb \n" + - "a=rtcp-fb:45 ccm fir\n" + - "a=rtcp-fb:45 nack \n" + - "a=rtcp-fb:45 nack pli\n" + - "a=rtcp-fb:45 nack \n" + - "a=rtcp-fb:45 nack pli\n" + - "a=rtcp-fb:45 transport-cc \n" + - "a=rtpmap:46 rtx/90000\n" + - "a=fmtp:46 apt=45\n" + - "a=rtcp-fb:46 nack \n" + - "a=rtcp-fb:46 nack pli\n" + - "a=rtcp-fb:46 transport-cc \n" + - "a=rtpmap:116 ulpfec/90000\n" + - "a=rtcp-fb:116 nack \n" + - "a=rtcp-fb:116 nack pli\n" + - "a=rtcp-fb:116 transport-cc \n" + - "a=extmap:1 http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01\n" + - "a=ssrc:3421396091 cname:BmFVQDtOlcBwXZCl\n" + - "a=ssrc:3421396091 msid:BmFVQDtOlcBwXZCl CLgunVCazXXKLyEx\n" + - "a=ssrc:3421396091 mslabel:BmFVQDtOlcBwXZCl\n" + - "a=ssrc:3421396091 label:CLgunVCazXXKLyEx\n" + - "a=msid:BmFVQDtOlcBwXZCl CLgunVCazXXKLyEx\n" + - "a=sendrecv\n", - []*webrtc.ICECandidateInit{{ - Candidate: "3628911098 1 udp 2130706431 192.168.3.218 49462 typ host", - SDPMid: ptrOf("0"), - SDPMLineIndex: ptrOf(uint16(0)), - }}, - "a=ice-ufrag:tUQMzoQAVLzlvBys\r\n" + - "a=ice-pwd:pimyGfJcjjRwvUjnmGOODSjtIxyDljQj\r\n" + - "m=video 9 UDP/TLS/RTP/SAVPF 96 97 98 99 100 101 102 121 127 120 125 107 108 109 123 118 45 46 116\r\n" + - "a=mid:0\r\n" + - "a=candidate:3628911098 1 udp 2130706431 192.168.3.218 49462 typ host\r\n", - }, -} - -func TestICEFragmentUnmarshal(t *testing.T) { - for _, ca := range iceFragmentCases { - t.Run(ca.name, func(t *testing.T) { - candidates, err := ICEFragmentUnmarshal([]byte(ca.enc)) - require.NoError(t, err) - require.Equal(t, ca.candidates, candidates) - }) - } -} - -func TestICEFragmentMarshal(t *testing.T) { - for _, ca := range iceFragmentCases { - t.Run(ca.name, func(t *testing.T) { - byts, err := ICEFragmentMarshal(ca.offer, ca.candidates) - require.NoError(t, err) - require.Equal(t, ca.enc, string(byts)) - }) - } -} diff --git a/internal/protocols/whip/sdp_fragment.go b/internal/protocols/whip/sdp_fragment.go new file mode 100644 index 00000000..ae9685a2 --- /dev/null +++ b/internal/protocols/whip/sdp_fragment.go @@ -0,0 +1,56 @@ +package whip + +import ( + "strings" + + "github.com/pion/sdp/v3" +) + +// SDPFragment is a SDP fragment. +// It's basically a SDP without most mandatory fields. +type SDPFragment struct { + Attributes []sdp.Attribute + Medias []*sdp.MediaDescription +} + +// Unmarshal decodes a SDP fragment. +func (f *SDPFragment) Unmarshal(buf []byte) error { + buf = append([]byte("v=0\r\no=- 0 0 IN IP4 0.0.0.0\r\ns=-\r\nt=0 0\r\n"), buf...) + + var sdp sdp.SessionDescription + err := sdp.Unmarshal(buf) + if err != nil { + return err + } + + f.Attributes = sdp.Attributes + f.Medias = sdp.MediaDescriptions + + return nil +} + +// Marshal encodes a SDP fragment. +func (f SDPFragment) Marshal() ([]byte, error) { + var b strings.Builder + + for _, a := range f.Attributes { + if a.Value != "" { + b.WriteString("a=" + a.Key + ":" + a.Value + "\r\n") + } else { + b.WriteString("a=" + a.Key + "\r\n") + } + } + + for _, m := range f.Medias { + b.WriteString("m=" + m.MediaName.String() + "\r\n") + for _, a := range m.Attributes { + if a.Value != "" { + b.WriteString("a=" + a.Key + ":" + a.Value + "\r\n") + } else { + b.WriteString("a=" + a.Key + "\r\n") + } + } + } + + return []byte(b.String()), nil +} diff --git a/internal/protocols/whip/sdp_fragment_test.go b/internal/protocols/whip/sdp_fragment_test.go new file mode 100644 index 00000000..c7692efa --- /dev/null +++ b/internal/protocols/whip/sdp_fragment_test.go @@ -0,0 +1,193 @@ +package whip + +import ( + "testing" + + "github.com/pion/sdp/v3" + "github.com/stretchr/testify/require" +) + +var sdpFragmentCases = []struct { + name string + enc string + dec *SDPFragment +}{ + { + "session-wide credentials", + "a=ice-ufrag:tUQMzoQAVLzlvBys\r\n" + + "a=ice-pwd:pimyGfJcjjRwvUjnmGOODSjtIxyDljQj\r\n" + + "m=video 9 UDP/TLS/RTP/SAVPF 96 97 98 99 100 101 102 121 127 120 125 107 108 109 123 118 45 46 116\r\n" + + "a=mid:0\r\n" + + "a=candidate:3628911098 1 udp 2130706431 192.168.3.218 49462 typ host\r\n", + &SDPFragment{ + Attributes: []sdp.Attribute{ + {Key: "ice-ufrag", Value: "tUQMzoQAVLzlvBys"}, + {Key: "ice-pwd", Value: "pimyGfJcjjRwvUjnmGOODSjtIxyDljQj"}, + }, + Medias: []*sdp.MediaDescription{{ + MediaName: sdp.MediaName{ + Media: "video", + Port: sdp.RangedPort{Value: 9}, + Protos: []string{"UDP", "TLS", "RTP", "SAVPF"}, + Formats: []string{ + "96", "97", "98", "99", "100", "101", "102", + "121", "127", "120", "125", "107", "108", "109", "123", "118", "45", "46", "116", + }, + }, + Attributes: []sdp.Attribute{ + {Key: "mid", Value: "0"}, + {Key: "candidate", Value: "3628911098 1 udp 2130706431 192.168.3.218 49462 typ host"}, + }, + }}, + }, + }, + { + "rfc9725 trickle ice patch request", + "a=group:BUNDLE 0 1\r\n" + + "m=audio 9 UDP/TLS/RTP/SAVPF 111\r\n" + + "a=mid:0\r\n" + + "a=ice-ufrag:EsAw\r\n" + + "a=ice-pwd:P2uYro0UCOQ4zxjKXaWCBui1\r\n" + + "a=candidate:1387637174 1 udp 2122260223 192.0.2.1 61764 typ host" + + " generation 0 ufrag EsAw network-id 1\r\n" + + "a=candidate:3471623853 1 udp 2122194687 198.51.100.2 61765 typ host" + + " generation 0 ufrag EsAw network-id 2\r\n" + + "a=candidate:473322822 1 tcp 1518280447 192.0.2.1 9 typ host tcptype active" + + " generation 0 ufrag EsAw network-id 1\r\n" + + "a=candidate:2154773085 1 tcp 1518214911 198.51.100.2 9 typ host tcptype active" + + " generation 0 ufrag EsAw network-id 2\r\n" + + "a=end-of-candidates\r\n", + &SDPFragment{ + Attributes: []sdp.Attribute{ + {Key: "group", Value: "BUNDLE 0 1"}, + }, + Medias: []*sdp.MediaDescription{ + { + MediaName: sdp.MediaName{ + Media: "audio", + Port: sdp.RangedPort{Value: 9}, + Protos: []string{"UDP", "TLS", "RTP", "SAVPF"}, + Formats: []string{"111"}, + }, + Attributes: []sdp.Attribute{ + {Key: "mid", Value: "0"}, + {Key: "ice-ufrag", Value: "EsAw"}, + {Key: "ice-pwd", Value: "P2uYro0UCOQ4zxjKXaWCBui1"}, + {Key: "candidate", Value: "1387637174 1 udp 2122260223 192.0.2.1 61764 typ host" + + " generation 0 ufrag EsAw network-id 1"}, + {Key: "candidate", Value: "3471623853 1 udp 2122194687 198.51.100.2 61765 typ host" + + " generation 0 ufrag EsAw network-id 2"}, + {Key: "candidate", Value: "473322822 1 tcp 1518280447 192.0.2.1 9 typ host tcptype active" + + " generation 0 ufrag EsAw network-id 1"}, + {Key: "candidate", Value: "2154773085 1 tcp 1518214911 198.51.100.2 9 typ host tcptype active" + + " generation 0 ufrag EsAw network-id 2"}, + {Key: "end-of-candidates", Value: ""}, + }, + }, + }, + }, + }, + { + "rfc9725 ice restart patch request", + "a=ice-options:trickle ice2\r\n" + + "a=group:BUNDLE 0 1\r\n" + + "m=audio 9 UDP/TLS/RTP/SAVPF 111\r\n" + + "a=mid:0\r\n" + + "a=ice-ufrag:ysXw\r\n" + + "a=ice-pwd:vw5LmwG4y/e6dPP/zAP9Gp5k\r\n" + + "a=candidate:1387637174 1 udp 2122260223 192.0.2.1 61764 typ host" + + " generation 0 ufrag EsAw network-id 1\r\n" + + "a=candidate:3471623853 1 udp 2122194687 198.51.100.2 61765 typ host" + + " generation 0 ufrag EsAw network-id 2\r\n" + + "a=candidate:473322822 1 tcp 1518280447 192.0.2.1 9 typ host tcptype active" + + " generation 0 ufrag EsAw network-id 1\r\n" + + "a=candidate:2154773085 1 tcp 1518214911 198.51.100.2 9 typ host tcptype active" + + " generation 0 ufrag EsAw network-id 2\r\n", + &SDPFragment{ + Attributes: []sdp.Attribute{ + {Key: "ice-options", Value: "trickle ice2"}, + {Key: "group", Value: "BUNDLE 0 1"}, + }, + Medias: []*sdp.MediaDescription{ + { + MediaName: sdp.MediaName{ + Media: "audio", + Port: sdp.RangedPort{Value: 9}, + Protos: []string{"UDP", "TLS", "RTP", "SAVPF"}, + Formats: []string{"111"}, + }, + Attributes: []sdp.Attribute{ + {Key: "mid", Value: "0"}, + {Key: "ice-ufrag", Value: "ysXw"}, + {Key: "ice-pwd", Value: "vw5LmwG4y/e6dPP/zAP9Gp5k"}, + {Key: "candidate", Value: "1387637174 1 udp 2122260223 192.0.2.1 61764 typ host" + + " generation 0 ufrag EsAw network-id 1"}, + {Key: "candidate", Value: "3471623853 1 udp 2122194687 198.51.100.2 61765 typ host" + + " generation 0 ufrag EsAw network-id 2"}, + {Key: "candidate", Value: "473322822 1 tcp 1518280447 192.0.2.1 9 typ host tcptype active" + + " generation 0 ufrag EsAw network-id 1"}, + {Key: "candidate", Value: "2154773085 1 tcp 1518214911 198.51.100.2 9 typ host tcptype active" + + " generation 0 ufrag EsAw network-id 2"}, + }, + }, + }, + }, + }, + { + "rfc9725 ice restart patch response", + "a=ice-lite\r\n" + + "a=ice-options:trickle ice2\r\n" + + "a=group:BUNDLE 0 1\r\n" + + "m=audio 9 UDP/TLS/RTP/SAVPF 111\r\n" + + "a=mid:0\r\n" + + "a=ice-ufrag:289b31b754eaa438\r\n" + + "a=ice-pwd:0b66f472495ef0ccac7bda653ab6be49ea13114472a5d10a\r\n" + + "a=candidate:1 1 udp 2130706431 198.51.100.1 39132 typ host\r\n" + + "a=end-of-candidates\r\n", + &SDPFragment{ + Attributes: []sdp.Attribute{ + {Key: "ice-lite", Value: ""}, + {Key: "ice-options", Value: "trickle ice2"}, + {Key: "group", Value: "BUNDLE 0 1"}, + }, + Medias: []*sdp.MediaDescription{ + { + MediaName: sdp.MediaName{ + Media: "audio", + Port: sdp.RangedPort{Value: 9}, + Protos: []string{"UDP", "TLS", "RTP", "SAVPF"}, + Formats: []string{"111"}, + }, + Attributes: []sdp.Attribute{ + {Key: "mid", Value: "0"}, + {Key: "ice-ufrag", Value: "289b31b754eaa438"}, + {Key: "ice-pwd", Value: "0b66f472495ef0ccac7bda653ab6be49ea13114472a5d10a"}, + {Key: "candidate", Value: "1 1 udp 2130706431 198.51.100.1 39132 typ host"}, + {Key: "end-of-candidates", Value: ""}, + }, + }, + }, + }, + }, +} + +func TestSDPFragmentUnmarshal(t *testing.T) { + for _, ca := range sdpFragmentCases { + t.Run(ca.name, func(t *testing.T) { + frag := &SDPFragment{} + err := frag.Unmarshal([]byte(ca.enc)) + require.NoError(t, err) + require.Equal(t, ca.dec, frag) + }) + } +} + +func TestSDPFragmentMarshal(t *testing.T) { + for _, ca := range sdpFragmentCases { + t.Run(ca.name, func(t *testing.T) { + byts, err := ca.dec.Marshal() + require.NoError(t, err) + require.Equal(t, ca.enc, string(byts)) + }) + } +} diff --git a/internal/servers/webrtc/http_server.go b/internal/servers/webrtc/http_server.go index 5a2dbaaf..45f394af 100644 --- a/internal/servers/webrtc/http_server.go +++ b/internal/servers/webrtc/http_server.go @@ -275,16 +275,17 @@ func (s *httpServer) onWHIPPatch(ctx *gin.Context, pathName string, rawSecret st return } - candidates, err := whip.ICEFragmentUnmarshal(byts) + var frag whip.SDPFragment + err = frag.Unmarshal(byts) if err != nil { s.writeErrorNoLog(ctx, http.StatusBadRequest, err) return } res := s.parent.addSessionCandidates(webRTCAddSessionCandidatesReq{ - pathName: pathName, - secret: secret, - candidates: candidates, + pathName: pathName, + secret: secret, + fragment: &frag, }) if res.err != nil { if errors.Is(res.err, ErrSessionNotFound) { @@ -295,6 +296,28 @@ func (s *httpServer) onWHIPPatch(ctx *gin.Context, pathName string, rawSecret st return } + if res.answer != nil { + var enc []byte + enc, err = res.answer.Marshal() + if err != nil { + s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) + return + } + + var ufrag string + ufrag, _, err = sdpFragmentToCredentials(res.answer) + if err != nil { + s.writeErrorNoLog(ctx, http.StatusInternalServerError, err) + return + } + + ctx.Header("Content-Type", "application/trickle-ice-sdpfrag") + ctx.Header("ETag", `"`+ufrag+`"`) + ctx.Writer.WriteHeader(http.StatusOK) + ctx.Writer.Write(enc) //nolint:errcheck + return + } + ctx.AbortWithStatusJSON(http.StatusNoContent, &defs.APIOK{ Status: defs.APIOKStatusOK, }) diff --git a/internal/servers/webrtc/server.go b/internal/servers/webrtc/server.go index f657788e..24dd0b1d 100644 --- a/internal/servers/webrtc/server.go +++ b/internal/servers/webrtc/server.go @@ -28,6 +28,7 @@ import ( "github.com/bluenviron/mediamtx/internal/externalcmd" "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/webrtc" + "github.com/bluenviron/mediamtx/internal/protocols/whip" "github.com/bluenviron/mediamtx/internal/restrictnetwork" ) @@ -149,15 +150,16 @@ type webRTCNewSessionReq struct { } type webRTCAddSessionCandidatesRes struct { - sx *session - err error + sx *session + answer *whip.SDPFragment + err error } type webRTCAddSessionCandidatesReq struct { - pathName string - secret uuid.UUID - candidates []*pwebrtc.ICECandidateInit - res chan webRTCAddSessionCandidatesRes + pathName string + secret uuid.UUID + fragment *whip.SDPFragment + res chan webRTCAddSessionCandidatesRes } type webRTCDeleteSessionRes struct { diff --git a/internal/servers/webrtc/server_test.go b/internal/servers/webrtc/server_test.go index 36eb8bdb..f279dd40 100644 --- a/internal/servers/webrtc/server_test.go +++ b/internal/servers/webrtc/server_test.go @@ -27,14 +27,16 @@ import ( "github.com/bluenviron/mediamtx/internal/unit" "github.com/google/uuid" "github.com/pion/rtp" + "github.com/pion/sdp/v3" pwebrtc "github.com/pion/webrtc/v4" "github.com/stretchr/testify/require" ) -func ptrOf[T any](v T) *T { - p := new(T) - *p = v - return p +func whipAnswer(body []byte) *pwebrtc.SessionDescription { + return &pwebrtc.SessionDescription{ + Type: pwebrtc.SDPTypeAnswer, + SDP: string(body), + } } func checkClose(t *testing.T, closeFunc func() error) { @@ -328,10 +330,15 @@ func TestServerOptionsICEServer(t *testing.T) { func TestServerPublish(t *testing.T) { var strm *stream.Stream + defer func() { + strm.Close() + }() + var reader *stream.Reader defer func() { strm.RemoveReader(reader) }() + dataReceived := make(chan struct{}) pathManager := &test.PathManager{ @@ -847,24 +854,39 @@ func TestServerPatchNotFound(t *testing.T) { defer tr.CloseIdleConnections() hc := &http.Client{Transport: tr} - pc, err := pwebrtc.NewPeerConnection(pwebrtc.Configuration{}) - require.NoError(t, err) - defer pc.GracefulClose() //nolint:errcheck + var frag whip.SDPFragment + frag.Medias = append(frag.Medias, &sdp.MediaDescription{ + MediaName: sdp.MediaName{ + Media: "video", + Port: sdp.RangedPort{Value: 9}, + Protos: []string{"UDP", "TLS", "RTP", "SAVPF"}, + Formats: []string{"96"}, + }, + Attributes: []sdp.Attribute{ + { + Key: "mid", + Value: "0", + }, + { + Key: "ice-ufrag", + Value: "dummy", + }, + { + Key: "ice-pwd", + Value: "dummydummydummydummydummydummy12", + }, + { + Key: "candidate", + Value: "candidate:1 1 UDP 2130706431 192.168.0.1 12345 typ host", + }, + }, + }) - _, err = pc.AddTransceiverFromKind(pwebrtc.RTPCodecTypeVideo) - require.NoError(t, err) - - offer, err := pc.CreateOffer(nil) - require.NoError(t, err) - - frag, err := whip.ICEFragmentMarshal(offer.SDP, []*pwebrtc.ICECandidateInit{{ - Candidate: "mycandidate", - SDPMLineIndex: ptrOf(uint16(0)), - }}) + enc, err := frag.Marshal() require.NoError(t, err) req, err := http.NewRequest(http.MethodPatch, - "http://localhost:8886/nonexisting/whep/"+uuid.UUID{}.String(), bytes.NewReader(frag)) + "http://localhost:8886/nonexisting/whep/"+uuid.UUID{}.String(), bytes.NewReader(enc)) require.NoError(t, err) req.Header.Set("Content-Type", "application/trickle-ice-sdpfrag") @@ -876,6 +898,203 @@ func TestServerPatchNotFound(t *testing.T) { require.Equal(t, http.StatusNotFound, res.StatusCode) } +func TestServerICERestart(t *testing.T) { + var strm *stream.Stream + defer func() { + strm.Close() + }() + + var reader *stream.Reader + defer func() { + strm.RemoveReader(reader) + }() + + dataReceived := make(chan struct{}, 10) + + pathManager := &test.PathManager{ + FindPathConfImpl: func(_ defs.PathFindPathConfReq) (*defs.PathFindPathConfRes, error) { + return &defs.PathFindPathConfRes{Conf: &conf.Path{}, User: ""}, nil + }, + AddPublisherImpl: func(req defs.PathAddPublisherReq) (*defs.PathAddPublisherRes, error) { + strm = &stream.Stream{ + Desc: req.Desc, + WriteQueueSize: 512, + RTPMaxPayloadSize: 1450, + Parent: test.NilLogger, + } + err := strm.Initialize() + require.NoError(t, err) + + subStream := &stream.SubStream{ + Stream: strm, + UseRTPPackets: true, + } + err = subStream.Initialize() + require.NoError(t, err) + + reader = &stream.Reader{Parent: test.NilLogger} + n := 0 + + reader.OnData( + strm.Desc.Medias[0], + strm.Desc.Medias[0].Formats[0], + func(u *unit.Unit) error { + switch n { + case 0: + require.Equal(t, unit.PayloadH264{ + {1}, + }, u.Payload) + + case 1: + require.Equal(t, unit.PayloadH264{ + {2}, + }, u.Payload) + } + n++ + dataReceived <- struct{}{} + return nil + }) + + strm.AddReader(reader) + + return &defs.PathAddPublisherRes{Path: &dummyPath{}, SubStream: subStream}, nil + }, + } + + s := &Server{ + Address: "127.0.0.1:8886", + AllowOrigins: []string{"*"}, + TrustedProxies: conf.IPNetworks{}, + ReadTimeout: conf.Duration(10 * time.Second), + WriteTimeout: conf.Duration(10 * time.Second), + LocalUDPAddress: "127.0.0.1:8887", + LocalTCPAddress: "127.0.0.1:8887", + IPsFromInterfaces: true, + IPsFromInterfacesList: []string{}, + AdditionalHosts: []string{}, + ICEServers: []conf.WebRTCICEServer{}, + HandshakeTimeout: conf.Duration(10 * time.Second), + TrackGatherTimeout: conf.Duration(2 * time.Second), + STUNGatherTimeout: conf.Duration(5 * time.Second), + PathManager: pathManager, + Parent: test.NilLogger, + } + err := s.Initialize() + require.NoError(t, err) + defer s.Close() + + tr := &http.Transport{} + defer tr.CloseIdleConnections() + hc := &http.Client{Transport: tr} + + su, err := url.Parse("http://localhost:8886/teststream/whip") + require.NoError(t, err) + + track := &webrtc.OutgoingTrack{ + Caps: pwebrtc.RTPCodecCapability{ + MimeType: pwebrtc.MimeTypeH264, + ClockRate: 90000, + SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", + }, + } + + wc := &whip.Client{ + HTTPClient: hc, + URL: su, + Publish: true, + OutgoingTracks: []*webrtc.OutgoingTrack{track}, + Log: test.NilLogger, + } + + err = wc.Initialize(context.Background()) + require.NoError(t, err) + defer checkClose(t, wc.Close) + + err = track.WriteRTP(&rtp.Packet{ + Header: rtp.Header{ + Version: 2, + Marker: true, + PayloadType: 96, + SequenceNumber: 1, + Timestamp: 1, + SSRC: 1, + }, + Payload: []byte{1}, + }) + require.NoError(t, err) + + <-dataReceived + + pc := wc.PeerConnection() + + offer, err := pc.CreatePartialOffer(true) + require.NoError(t, err) + + f := &whip.SDPFragment{} + + var desc sdp.SessionDescription + err = desc.Unmarshal([]byte(offer.SDP)) + require.NoError(t, err) + + media := desc.MediaDescriptions[0] + ufrag, _ := media.Attribute("ice-ufrag") + pwd, _ := media.Attribute("ice-pwd") + + f.Medias = append(f.Medias, &sdp.MediaDescription{ + MediaName: media.MediaName, + Attributes: []sdp.Attribute{ + {Key: "mid", Value: "0"}, + {Key: "ice-ufrag", Value: ufrag}, + {Key: "ice-pwd", Value: pwd}, + }, + }) + + enc, err := f.Marshal() + require.NoError(t, err) + + req, err := http.NewRequest(http.MethodPatch, wc.URL.String(), bytes.NewReader(enc)) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/trickle-ice-sdpfrag") + + res, err := hc.Do(req) + require.NoError(t, err) + defer res.Body.Close() + require.Equal(t, http.StatusOK, res.StatusCode) + require.Equal(t, "application/trickle-ice-sdpfrag", res.Header.Get("Content-Type")) + + body, err := io.ReadAll(res.Body) + require.NoError(t, err) + + var resFrag whip.SDPFragment + err = resFrag.Unmarshal(body) + require.NoError(t, err) + + resUfrag, _ := resFrag.Medias[0].Attribute("ice-ufrag") + resPwd, _ := resFrag.Medias[0].Attribute("ice-pwd") + + patchedSDP := replaceICECredentials([]byte(pc.RemoteDescription().SDP), resUfrag, resPwd) + err = pc.SetAnswer(whipAnswer(patchedSDP)) + require.NoError(t, err) + + err = pc.WaitUntilConnected(2 * time.Second) + require.NoError(t, err) + + err = track.WriteRTP(&rtp.Packet{ + Header: rtp.Header{ + Version: 2, + Marker: true, + PayloadType: 96, + SequenceNumber: 2, + Timestamp: uint32(2), + SSRC: 1, + }, + Payload: []byte{2}, + }) + require.NoError(t, err) + + <-dataReceived +} + func TestServerDeleteNotFound(t *testing.T) { s := initializeTestServer(t) defer s.Close() diff --git a/internal/servers/webrtc/session.go b/internal/servers/webrtc/session.go index 925d0395..e87c9552 100644 --- a/internal/servers/webrtc/session.go +++ b/internal/servers/webrtc/session.go @@ -7,6 +7,8 @@ import ( "fmt" "net" "net/http" + "strconv" + "strings" "sync" "time" @@ -24,6 +26,7 @@ import ( "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/httpp" "github.com/bluenviron/mediamtx/internal/protocols/webrtc" + "github.com/bluenviron/mediamtx/internal/protocols/whip" "github.com/bluenviron/mediamtx/internal/stream" ) @@ -34,6 +37,187 @@ func whipOffer(body []byte) *pwebrtc.SessionDescription { } } +func parseOfferUfrag(offer []byte) string { + var desc sdp.SessionDescription + if err := desc.Unmarshal(offer); err != nil { + return "" + } + + // per-media credentials (priority matches sdpFragmentToCredentials) + for _, media := range desc.MediaDescriptions { + if ufrag, ok := media.Attribute("ice-ufrag"); ok && ufrag != "" { + return ufrag + } + } + + // session-level credentials + for _, attr := range desc.Attributes { + if attr.Key == "ice-ufrag" && attr.Value != "" { + return attr.Value + } + } + return "" +} + +func replaceICECredentials(offerSDP []byte, ufrag, pwd string) []byte { + s := string(offerSDP) + sep := "\r\n" + if !strings.Contains(s, "\r\n") { + sep = "\n" + } + lines := strings.Split(s, sep) + for i, line := range lines { + if strings.HasPrefix(line, "a=ice-ufrag:") { + lines[i] = "a=ice-ufrag:" + ufrag + } else if strings.HasPrefix(line, "a=ice-pwd:") { + lines[i] = "a=ice-pwd:" + pwd + } + } + return []byte(strings.Join(lines, sep)) +} + +func sdpFragmentToCredentials(frag *whip.SDPFragment) (string, string, error) { + // media credentials + for _, media := range frag.Medias { + ufrag, _ := media.Attribute("ice-ufrag") + pwd, _ := media.Attribute("ice-pwd") + if ufrag != "" && pwd != "" { + return ufrag, pwd, nil + } + } + + // session-wide credentials + var ufrag, pwd string + for _, attr := range frag.Attributes { + switch attr.Key { + case "ice-ufrag": + ufrag = attr.Value + case "ice-pwd": + pwd = attr.Value + } + } + if ufrag != "" && pwd != "" { + return ufrag, pwd, nil + } + + return "", "", fmt.Errorf("ICE credentials not found") +} + +func sdpFragmentToCandidates(frag *whip.SDPFragment) ([]*pwebrtc.ICECandidateInit, error) { + var candidates []*pwebrtc.ICECandidateInit + + for _, media := range frag.Medias { + mid, ok := media.Attribute("mid") + if !ok { + return nil, fmt.Errorf("mid attribute is missing") + } + + tmp, err := strconv.ParseUint(mid, 10, 16) + if err != nil { + return nil, fmt.Errorf("invalid mid attribute") + } + midNum := uint16(tmp) + + for _, attr := range media.Attributes { + if attr.Key == "candidate" { + candidates = append(candidates, &pwebrtc.ICECandidateInit{ + Candidate: attr.Value, + SDPMid: &mid, + SDPMLineIndex: &midNum, + }) + } + } + } + + return candidates, nil +} + +func mediaHasCredentialsOrCandidates(media *sdp.MediaDescription) bool { + hasUfrag := false + hasPwd := false + + for _, attr := range media.Attributes { + if attr.Value != "" { + switch attr.Key { + case "ice-ufrag": + hasUfrag = true + + case "ice-pwd": + hasPwd = true + + case "candidate": + return true + } + } + } + + return (hasUfrag && hasPwd) +} + +func fullAnswerToSDPFragment(answerSDP string) (*whip.SDPFragment, error) { + var psdp sdp.SessionDescription + err := psdp.Unmarshal([]byte(answerSDP)) + if err != nil { + return nil, err + } + + frag := &whip.SDPFragment{ + Attributes: []sdp.Attribute{ + {Key: "ice-options", Value: "trickle ice2"}, + }, + } + + filled := false + + for _, attr := range psdp.Attributes { + switch attr.Key { + case "ice-ufrag", "ice-pwd": + frag.Attributes = append(frag.Attributes, sdp.Attribute{Key: attr.Key, Value: attr.Value}) + filled = true + } + } + + for _, media := range psdp.MediaDescriptions { + if mediaHasCredentialsOrCandidates(media) { + filled = true + + mid, ok := media.Attribute("mid") + if !ok { + return nil, fmt.Errorf("mid attribute is missing") + } + + mediaFrag := &sdp.MediaDescription{ + MediaName: media.MediaName, + Attributes: []sdp.Attribute{ + {Key: "mid", Value: mid}, + }, + } + + ufrag, _ := media.Attribute("ice-ufrag") + pwd, _ := media.Attribute("ice-pwd") + if ufrag != "" && pwd != "" { + mediaFrag.Attributes = append(mediaFrag.Attributes, sdp.Attribute{Key: "ice-ufrag", Value: ufrag}) + mediaFrag.Attributes = append(mediaFrag.Attributes, sdp.Attribute{Key: "ice-pwd", Value: pwd}) + } + + for _, attr := range media.Attributes { + if attr.Key == "candidate" { + mediaFrag.Attributes = append(mediaFrag.Attributes, attr) + } + } + mediaFrag.Attributes = append(mediaFrag.Attributes, sdp.Attribute{Key: "end-of-candidates"}) + + frag.Medias = append(frag.Medias, mediaFrag) + } + } + + if !filled { + return nil, fmt.Errorf("no credentials or candidates found in the answer") + } + + return frag, nil +} + type sessionParent interface { closeSession(sx *session) generateICEServers(clientConfig bool) ([]pwebrtc.ICEServer, error) @@ -214,7 +398,7 @@ func (s *session) runPublish() (int, error) { return http.StatusNotAcceptable, err } - answer, err := pc.CreateFullAnswer(offer) + answer, err := pc.CreateFullAnswer(offer, false) if err != nil { return http.StatusBadRequest, err } @@ -352,7 +536,7 @@ func (s *session) runRead() (int, error) { offer := whipOffer(s.req.offer) - answer, err := pc.CreateFullAnswer(offer) + answer, err := pc.CreateFullAnswer(offer, false) if err != nil { return http.StatusBadRequest, err } @@ -410,16 +594,57 @@ func (s *session) writeAnswer(answer *pwebrtc.SessionDescription) { } func (s *session) readRemoteCandidates(pc *webrtc.PeerConnection) { + remoteUfrag := parseOfferUfrag(s.req.offer) + for { select { case req := <-s.chAddCandidates: - for _, candidate := range req.candidates { - err := pc.AddRemoteCandidate(candidate) + // do not check for errors since credentials are optional + ufrag, pwd, _ := sdpFragmentToCredentials(req.fragment) + + candidates, err := sdpFragmentToCandidates(req.fragment) + if err != nil { + req.res <- webRTCAddSessionCandidatesRes{err: err} + continue + } + + // ICE restart: client sent new credentials + var answer *pwebrtc.SessionDescription + if ufrag != "" && ufrag != remoteUfrag { + sdp := replaceICECredentials(s.req.offer, ufrag, pwd) + + answer, err = pc.CreateFullAnswer(whipOffer(sdp), true) if err != nil { req.res <- webRTCAddSessionCandidatesRes{err: err} + continue } } - req.res <- webRTCAddSessionCandidatesRes{} + + var addErr error + for _, candidate := range candidates { + addErr = pc.AddRemoteCandidate(candidate) + if addErr != nil { + break + } + } + if addErr != nil { + req.res <- webRTCAddSessionCandidatesRes{err: addErr} + continue + } + + if ufrag != "" && ufrag != remoteUfrag { + var frag *whip.SDPFragment + frag, err = fullAnswerToSDPFragment(answer.SDP) + if err != nil { + req.res <- webRTCAddSessionCandidatesRes{err: err} + continue + } + + remoteUfrag = ufrag + req.res <- webRTCAddSessionCandidatesRes{answer: frag} + } else { + req.res <- webRTCAddSessionCandidatesRes{} + } case <-s.ctx.Done(): return diff --git a/internal/staticsources/webrtc/source_test.go b/internal/staticsources/webrtc/source_test.go index baf329da..b9d6f6c6 100644 --- a/internal/staticsources/webrtc/source_test.go +++ b/internal/staticsources/webrtc/source_test.go @@ -68,7 +68,7 @@ func TestSource(t *testing.T) { require.NoError(t, err2) offer := whipOffer(body) - answer, err2 := pc.CreateFullAnswer(offer) + answer, err2 := pc.CreateFullAnswer(offer, false) require.NoError(t, err2) w.Header().Set("Content-Type", "application/sdp")