Fix tests on warp-4.2.18

dtlstransport.go

@@ -326,6 +326,64 @@
 	})
 }
 
+// dtlsPacketInjector feeds embedded DTLS packets into a dtls.Conn from its own
+// goroutine, in arrival order.
+//
+// Conn.InjectInboundPacket sends on an unbuffered channel that only
+// Conn.nextPacket reads, so it blocks until the DTLS stack asks for its next
+// packet. The SPED callback runs on the ICE agent's task loop, the one
+// goroutine handling all inbound STUN, and blocking that stalls the agent so
+// the handshake never completes. Queue instead and let this goroutine block.
+type dtlsPacketInjector struct {
+	packets   chan pendingDTLSPacket
+	done      chan struct{}
+	closeOnce sync.Once
+}
+
+// dtlsInjectorQueueLen is deliberately generous: a DTLS handshake flight is a
+// handful of packets, so reaching this bound means the DTLS stack has stopped
+// consuming altogether and dropping is the right answer.
+const dtlsInjectorQueueLen = 128
+
+func newDTLSPacketInjector(conn *dtls.Conn) *dtlsPacketInjector {
+	injector := &dtlsPacketInjector{
+		packets: make(chan pendingDTLSPacket, dtlsInjectorQueueLen),
+		done:    make(chan struct{}),
+	}
+
+	go func() {
+		for {
+			select {
+			case <-injector.done:
+				return
+			case packet := <-injector.packets:
+				conn.InjectInboundPacket(packet.packet, packet.rAddr)
+			}
+		}
+	}()
+
+	return injector
+}
+
+// enqueue is the SPED callback handed to ICETransport. It never blocks.
+func (i *dtlsPacketInjector) enqueue(packet []byte, rAddr net.Addr) {
+	buffered := make([]byte, len(packet))
+	copy(buffered, packet)
+
+	select {
+	case i.packets <- pendingDTLSPacket{packet: buffered, rAddr: rAddr}:
+	case <-i.done:
+	default:
+		// Queue full: drop rather than block the caller. DTLS retransmits its
+		// flights, so a drop costs a retransmission. Blocking would cost the
+		// whole ICE agent.
+	}
+}
+
+func (i *dtlsPacketInjector) stop() {
+	i.closeOnce.Do(func() { close(i.done) })
+}
+
 func (t *DTLSTransport) start(remoteParameters DTLSParameters, handshake func(*dtls.Conn) error) error {
 	role, certificate, err := t.prepareStart(remoteParameters)
 	if err != nil {
@@ -347,9 +405,9 @@
 
 	// Configure ICE for SPED after we created the DTLS transport.
 	if t.api.settingEngine.enableSped {
-		t.iceTransport.SetDtlsCallback(func(packet []byte, rAddr net.Addr) {
-			dtlsConn.InjectInboundPacket(packet, rAddr)
-		})
+		injector := newDTLSPacketInjector(dtlsConn)
+		defer injector.stop()
+		t.iceTransport.SetDtlsCallback(injector.enqueue)
 	}
 
 	// This awaits the DTLS handshake.
@@ -402,7 +460,7 @@
 		dtls.WithCertificates(certificate),
 		dtls.WithSRTPProtectionProfiles(t.srtpProtectionProfiles()...),
 		dtls.WithExtendedMasterSecret(t.api.settingEngine.dtls.extendedMasterSecret),
 		// TODO: this should be the default, DTLS runs over ICE which *hopefully* checks the source.
 		dtls.WithInsecureSkipVerify(true),
 		dtls.WithLoggerFactory(t.api.settingEngine.LoggerFactory),
 		dtls.WithVerifyPeerCertificate(t.verifyPeerCertificateFunc()),
@@ -415,7 +473,7 @@
 		)
 	}
 
 	// TODO: should this initially be set to one day for SPED?
 	if t.api.settingEngine.dtls.retransmissionInterval > 0 {
 		sharedOpts = append(
 			sharedOpts,

go.mod

@@ -36,6 +36,6 @@ require (
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
-replace github.com/pion/dtls/v3 => github.com/livekit/dtls/v3 v3.1.4-warp.1
+replace github.com/pion/dtls/v3 => github.com/livekit/dtls/v3 v3.1.5-warp.1
 
-replace github.com/pion/ice/v4 => github.com/livekit/ice/v4 v4.2.7-warp.1
+replace github.com/pion/ice/v4 => github.com/livekit/ice/v4 v4.4.0-warp.1

go.sum

@@ -25,10 +25,10 @@ github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/livekit/dtls/v3 v3.1.4-warp.1 h1:oS+wSZDhgIXDLX8cmSeFhfdX28XZxwxLGNOmTP8B+yk=
-github.com/livekit/dtls/v3 v3.1.4-warp.1/go.mod h1:cr/qotLISUw/9C1m83ZPNZtj9WnXkYLpfCptPqbkInc=
-github.com/livekit/ice/v4 v4.2.7-warp.1 h1:6ieMfxp6ADkM10SR0Fyu2LYe+TJQADxY//1BJAphn4o=
-github.com/livekit/ice/v4 v4.2.7-warp.1/go.mod h1:9SNPaq0c7El/ki8leJzyCkK10zsskprR3zTNbO3monY=
+github.com/livekit/dtls/v3 v3.1.5-warp.1 h1:5jo2qQreDnUrgw6Al1F+9xVyXHvWQ2IaQkjF53FHEyU=
+github.com/livekit/dtls/v3 v3.1.5-warp.1/go.mod h1:gz1K4jg6c+fq86oQMH4pilpCEOEPwmEr2jY+VcF/mkU=
+github.com/livekit/ice/v4 v4.4.0-warp.1 h1:P9Y1f02VVx6kkRJxY3jzoS7HnblFbYkgMwiu6iOqCOU=
+github.com/livekit/ice/v4 v4.4.0-warp.1/go.mod h1:obAyD+J+Hzs7QA7Y8YXHp5uIn6gb7z87pKedXZkrcFU=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=

icetransport.go

@@ -19,6 +19,12 @@ import (
 	"github.com/pion/webrtc/v4/internal/util"
 )
 
+// pendingDTLSPacket is an embedded DTLS packet held until DTLSTransport is ready.
+type pendingDTLSPacket struct {
+	packet []byte
+	rAddr  net.Addr
+}
+
 // ICETransport allows an application access to information about the ICE
 // transport over which packets are sent and received.
 type ICETransport struct {
@@ -42,6 +48,10 @@ type ICETransport struct {
 
 	dtlsCallback func(packet []byte, rAddr net.Addr)
 
+	// Inbound DTLS packets the agent delivered before DTLSTransport configured
+	// its callback (SPED only)
+	pendingDTLSPackets []pendingDTLSPacket
+
 	log logging.LeveledLogger
 }
 
@@ -123,7 +133,16 @@ func (t *ICETransport) StartContext(
 	if agent == nil {
 		return fmt.Errorf("%w: unable to start ICETransport", errICEAgentNotExist)
 	}
-	agent.SetDtlsCallback(t.dtlsCallback)
+
+	// The agent only piggybacks DTLS once a callback is set, and DTLSTransport
+	// cannot set the real one until the mux exists (created below). If ICE
+	// connects in that gap the handshake falls back to plain DTLS and loses the
+	// round trip SPED saves. Install a shim now and buffer until it arrives.
+	if t.gatherer.api.settingEngine.enableSped && t.dtlsCallback == nil {
+		agent.SetDtlsCallback(t.forwardDTLSPacket)
+	} else {
+		agent.SetDtlsCallback(t.dtlsCallback)
+	}
 
 	if err := agent.OnConnectionStateChange(func(iceState ice.ConnectionState) {
 		state := newICETransportStateFromICE(iceState)
@@ -179,6 +198,10 @@ func (t *ICETransport) StartContext(
 	if err != nil {
 		t.lock.Lock()
 
+		// The derived context is unused past this point
+		ctxCancel()
+		t.ctxCancel = nil
+
 		return err
 	}
 
@@ -222,12 +245,53 @@ func (t *ICETransport) StartContext(
 
 func (t *ICETransport) SetDtlsCallback(cb func(packet []byte, rAddr net.Addr)) {
 	t.lock.Lock()
-	defer t.lock.Unlock()
-	if agent := t.gatherer.getAgent(); agent != nil {
-		agent.SetDtlsCallback(cb)
-	} else {
-		t.dtlsCallback = cb
+	agent := t.gatherer.getAgent()
+
+	if agent == nil || !t.gatherer.api.settingEngine.enableSped {
+		if agent != nil {
+			agent.SetDtlsCallback(cb)
+		} else {
+			t.dtlsCallback = cb
+		}
+		t.lock.Unlock()
+
+		return
 	}
+
+	// SPED: StartContext put a shim on the agent that forwards to this field, so
+	// record the callback and hand it whatever arrived before it existed.
+	t.dtlsCallback = cb
+	pending := t.pendingDTLSPackets
+	t.pendingDTLSPackets = nil
+	t.lock.Unlock()
+
+	if cb == nil {
+		// Teardown: take the shim off the agent as well.
+		agent.SetDtlsCallback(nil)
+
+		return
+	}
+
+	for _, p := range pending {
+		cb(p.packet, p.rAddr)
+	}
+}
+
+// forwardDTLSPacket hands an embedded DTLS packet to DTLSTransport, buffering it
+// while the DTLS transport is still being built.
+func (t *ICETransport) forwardDTLSPacket(packet []byte, rAddr net.Addr) {
+	t.lock.Lock()
+	if cb := t.dtlsCallback; cb != nil {
+		t.lock.Unlock()
+		cb(packet, rAddr)
+
+		return
+	}
+
+	buffered := make([]byte, len(packet))
+	copy(buffered, packet)
+	t.pendingDTLSPackets = append(t.pendingDTLSPackets, pendingDTLSPacket{packet: buffered, rAddr: rAddr})
+	t.lock.Unlock()
 }
 
 // restart is not exposed currently because ORTC has users create a whole new ICETransport

peerconnection.go

@@ -2959,6 +2959,7 @@ func (pc *PeerConnection) generateUnmatchedSDP(
 		nil,
 		pc.api.settingEngine.getSCTPMaxMessageSize(),
 		false,
+		pc.api.settingEngine.enableSped,
 	)
 }
 
@@ -3147,6 +3148,7 @@ func (pc *PeerConnection) generateMatchedSDP(
 		bundleGroup,
 		pc.api.settingEngine.getSCTPMaxMessageSize(),
 		ignoreRidPauseForRecv,
+		pc.api.settingEngine.enableSped,
 	)
 }
 

sdp.go

@@ -710,6 +710,7 @@ func populateSDP(
 	matchBundleGroup *string,
 	sctpMaxMessageSize uint32,
 	ignoreRidPauseForRecv bool,
+	enableSped bool,
 ) (*sdp.SessionDescription, error) {
 	var err error
 	mediaDtlsFingerprints := []DTLSFingerprint{}
@@ -791,6 +792,14 @@ func populateSDP(
 		descr = descr.WithValueAttribute(sdp.AttrKeyICELite, "")
 	}
 
+	if enableSped {
+		// Advertise SPED so a libwebrtc peer enables DTLS-in-STUN.
+		// Without it, it answers with plain DTLS and the fold silently does not happen.
+		// "trickle" (RFC 8840 S4.1.1) is kept in the list because a peer that sees
+		// ice-options without it treats us as a non-trickle agent and de-paces ICE.
+		descr = descr.WithValueAttribute(sdp.AttrKeyICEOptions, "trickle goog-sped-v1")
+	}
+
 	if isExtmapAllowMixed {
 		descr = descr.WithPropertyAttribute(sdp.AttrKeyExtMapAllowMixed)
 	}

sdp_test.go

@@ -761,6 +761,7 @@ func TestMediaDescriptionFingerprints(t *testing.T) {
 				nil,
 				0,
 				false,
+				false,
 			)
 			assert.NoError(t, err)
 
@@ -816,6 +817,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			nil,
 			se.getSCTPMaxMessageSize(),
 			se.ignoreRidPauseForRecv,
+			false,
 		)
 		assert.Nil(t, err)
 
@@ -878,6 +880,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			nil,
 			se.getSCTPMaxMessageSize(),
 			se.ignoreRidPauseForRecv,
+			false,
 		)
 		assert.Nil(t, err)
 
@@ -938,6 +941,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			nil,
 			se.getSCTPMaxMessageSize(),
 			false,
+			false,
 		)
 		assert.Nil(t, err)
 
@@ -979,6 +983,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			nil,
 			se.getSCTPMaxMessageSize(),
 			false,
+			false,
 		)
 		assert.Nil(t, err)
 
@@ -1038,6 +1043,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			nil,
 			se.getSCTPMaxMessageSize(),
 			false,
+			false,
 		)
 		assert.NoError(t, err)
 
@@ -1072,6 +1078,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			nil,
 			se.getSCTPMaxMessageSize(),
 			false,
+			false,
 		)
 		assert.Nil(t, err)
 
@@ -1103,6 +1110,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			nil,
 			se.getSCTPMaxMessageSize(),
 			false,
+			false,
 		)
 		assert.Nil(t, err)
 
@@ -1148,6 +1156,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			nil,
 			se.getSCTPMaxMessageSize(),
 			false,
+			false,
 		)
 		assert.Nil(t, err)
 
@@ -1189,6 +1198,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			&matchedBundle,
 			se.getSCTPMaxMessageSize(),
 			false,
+			false,
 		)
 		assert.Nil(t, err)
 
@@ -1232,6 +1242,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			&matchedBundle,
 			se.getSCTPMaxMessageSize(),
 			false,
+			false,
 		)
 		assert.Nil(t, err)
 
@@ -1266,6 +1277,7 @@ func TestPopulateSDP(t *testing.T) { //nolint:gocyclo,cyclop,maintidx
 			nil,
 			se.getSCTPMaxMessageSize(),
 			false,
+			false,
 		)
 		assert.Nil(t, err)